diff --git a/README.en.md b/README.en.md index bd0dad1..e541b5b 100644 --- a/README.en.md +++ b/README.en.md @@ -45,6 +45,7 @@ Deploy it privately, integrate with the HTTP API, and start receiving crypto pay | **Solana** | USDT, USDC | | **BEP20** (BSC) | USDT, USDC, BNB | | **Polygon** | USDT, USDC | +| **Aptos** | USDC, USDT | | **More** | Expanding continuously | > Actual support depends on the [latest release](https://github.com/GMWalletApp/epusdt/releases) and the [official docs](https://epusdt.com). @@ -68,7 +69,7 @@ No matter what kind of platform you operate, Epusdt can fit into existing paymen ## Core Features -- **Multi-chain and multi-token** — TRC20, ERC20, BEP20, Polygon, and more +- **Multi-chain and multi-token** — TRC20, ERC20, BEP20, Polygon, Aptos, and more - **Self-hosted** — full control of funds and wallet infrastructure - **Low-dependency runtime** — a single binary can run the service; low-traffic setups do not require MySQL + Redis - **Cross-platform** — supports x86 / ARM on Windows / Linux / Mac diff --git a/README.md b/README.md index ef6e818..8845987 100644 --- a/README.md +++ b/README.md @@ -45,6 +45,7 @@ | **Solana** | USDT、USDC | | **BEP20** (BSC) | USDT、USDC、BNB | | **Polygon** | USDT、USDC | +| **Aptos** | USDC、USDT | | **更多** | 持续扩展中… | > 具体支持的链与代币以 [最新版本](https://github.com/GMWalletApp/epusdt/releases) 及 [官方文档](https://epusdt.com) 为准。 @@ -74,7 +75,7 @@ Epusdt 已完成第三方安全审计。 ## 核心特性 -- **多链多币种** — 支持 TRC20、ERC20、BEP20、Polygon 等主流网络 +- **多链多币种** — 支持 TRC20、ERC20、BEP20、Polygon、Aptos 等主流网络 - **私有化部署** — 资金完全自主掌控 - **零依赖运行** — 单个二进制即可启动，低并发场景无需 MySQL + Redis - **跨平台** — 支持 x86 / ARM 架构的 Windows / Linux / Mac diff --git a/src/controller/admin/chain_controller.go b/src/controller/admin/chain_controller.go index a225a30..a86cc36 100644 --- a/src/controller/admin/chain_controller.go +++ b/src/controller/admin/chain_controller.go @@ -40,7 +40,7 @@ func (c *BaseAdminController) ListChains(ctx echo.Context) error { // @Security AdminJWT // @Accept json // @Produce json -// @Param network path string true "Network name (e.g. ton, tron, ethereum, solana)" +// @Param network path string true "Network name (e.g. ton, tron, ethereum, solana, aptos)" // @Param request body admin.UpdateChainRequest true "Chain settings" // @Success 200 {object} response.ApiResponse // @Failure 400 {object} response.ApiResponse diff --git a/src/controller/admin/chain_token_controller.go b/src/controller/admin/chain_token_controller.go index b6cd00b..d31dad4 100644 --- a/src/controller/admin/chain_token_controller.go +++ b/src/controller/admin/chain_token_controller.go @@ -12,8 +12,10 @@ import ( // CreateChainTokenRequest is the payload for creating a chain token. type CreateChainTokenRequest struct { - Network string `json:"network" validate:"required" example:"tron"` - Symbol string `json:"symbol" validate:"required|maxLen:32" example:"USDT"` + Network string `json:"network" validate:"required" example:"tron"` + Symbol string `json:"symbol" validate:"required|maxLen:32" example:"USDT"` + // Contract address or chain-native asset identifier. For Aptos this is + // the Move asset id (coin type, metadata address, or object type). ContractAddress string `json:"contract_address" example:"TR7NHqjeKQxGTCi8q8ZY4pL8otSzgjLj6t"` Decimals int `json:"decimals" example:"6"` Enabled *bool `json:"enabled" example:"true"` @@ -22,6 +24,8 @@ type CreateChainTokenRequest struct { // UpdateChainTokenRequest is the payload for updating a chain token. type UpdateChainTokenRequest struct { + // Contract address or chain-native asset identifier. For Aptos this is + // the Move asset id (coin type, metadata address, or object type). ContractAddress *string `json:"contract_address" example:"TR7NHqjeKQxGTCi8q8ZY4pL8otSzgjLj6t"` Decimals *int `json:"decimals" example:"6"` Enabled *bool `json:"enabled" example:"true"` @@ -33,6 +37,17 @@ type ChangeChainTokenStatusRequest struct { Enabled bool `json:"enabled" example:"true"` } +func validateChainTokenPaymentConfig(row *mdb.ChainToken) error { + if row == nil { + return nil + } + contractAddress := strings.TrimSpace(row.ContractAddress) + if row.Enabled && row.Network == mdb.NetworkAptos && contractAddress == "" { + return constant.ParamsMarshalErr + } + return nil +} + // ListChainTokens returns rows, optionally filtered by network. // @Summary List chain tokens // @Description Returns chain tokens, optionally filtered by network @@ -54,7 +69,7 @@ func (c *BaseAdminController) ListChainTokens(ctx echo.Context) error { // CreateChainToken inserts a new row. Default decimals=6 when 0 passed. // @Summary Create chain token -// @Description Create a new chain token entry +// @Description Create a new chain token entry. For Aptos, contract_address stores the Move asset id rather than an EVM contract address. // @Tags Admin Chain Tokens // @Security AdminJWT // @Accept json @@ -87,6 +102,9 @@ func (c *BaseAdminController) CreateChainToken(ctx echo.Context) error { Enabled: enabled, MinAmount: req.MinAmount, } + if err := validateChainTokenPaymentConfig(row); err != nil { + return c.FailJson(ctx, err) + } if err := data.CreateChainToken(row); err != nil { return c.FailJson(ctx, err) } @@ -95,7 +113,7 @@ func (c *BaseAdminController) CreateChainToken(ctx echo.Context) error { // UpdateChainToken patches mutable columns. // @Summary Update chain token -// @Description Patch chain token fields +// @Description Patch chain token fields. For Aptos, contract_address stores the Move asset id rather than an EVM contract address. // @Tags Admin Chain Tokens // @Security AdminJWT // @Accept json @@ -115,18 +133,35 @@ func (c *BaseAdminController) UpdateChainToken(ctx echo.Context) error { return c.FailJson(ctx, constant.ParamsMarshalErr) } fields := map[string]interface{}{} + existing, err := data.GetChainTokenByID(id) + if err != nil { + return c.FailJson(ctx, err) + } + if existing == nil { + existing = &mdb.ChainToken{} + } + next := *existing if req.ContractAddress != nil { - fields["contract_address"] = *req.ContractAddress + next.ContractAddress = strings.TrimSpace(*req.ContractAddress) + fields["contract_address"] = next.ContractAddress } if req.Decimals != nil { + next.Decimals = *req.Decimals fields["decimals"] = *req.Decimals } if req.Enabled != nil { + next.Enabled = *req.Enabled fields["enabled"] = *req.Enabled } if req.MinAmount != nil { + next.MinAmount = *req.MinAmount fields["min_amount"] = *req.MinAmount } + if next.ID > 0 { + if err = validateChainTokenPaymentConfig(&next); err != nil { + return c.FailJson(ctx, err) + } + } if err := data.UpdateChainTokenFields(id, fields); err != nil { return c.FailJson(ctx, err) } @@ -154,6 +189,17 @@ func (c *BaseAdminController) ChangeChainTokenStatus(ctx echo.Context) error { if err := ctx.Bind(req); err != nil { return c.FailJson(ctx, constant.ParamsMarshalErr) } + row, err := data.GetChainTokenByID(id) + if err != nil { + return c.FailJson(ctx, err) + } + if row != nil && row.ID > 0 { + next := *row + next.Enabled = req.Enabled + if err = validateChainTokenPaymentConfig(&next); err != nil { + return c.FailJson(ctx, err) + } + } if err := data.UpdateChainTokenFields(id, map[string]interface{}{"enabled": req.Enabled}); err != nil { return c.FailJson(ctx, err) } diff --git a/src/controller/admin/chain_token_controller_test.go b/src/controller/admin/chain_token_controller_test.go new file mode 100644 index 0000000..45c74f7 --- /dev/null +++ b/src/controller/admin/chain_token_controller_test.go @@ -0,0 +1,25 @@ +package admin + +import ( + "testing" + + "github.com/GMWalletApp/epusdt/model/mdb" + "github.com/GMWalletApp/epusdt/util/constant" +) + +func TestValidateChainTokenPaymentConfigRejectsEnabledAptosTokenWithoutAssetID(t *testing.T) { + for _, row := range []mdb.ChainToken{ + {Network: mdb.NetworkAptos, Symbol: "USDT", Enabled: true}, + } { + if err := validateChainTokenPaymentConfig(&row); err != constant.ParamsMarshalErr { + t.Fatalf("validateChainTokenPaymentConfig(%+v) = %v, want %v", row, err, constant.ParamsMarshalErr) + } + } +} + +func TestValidateChainTokenPaymentConfigAllowsDisabledAptosTokenWithoutAssetID(t *testing.T) { + row := mdb.ChainToken{Network: mdb.NetworkAptos, Symbol: "USDT", Enabled: false} + if err := validateChainTokenPaymentConfig(&row); err != nil { + t.Fatalf("validateChainTokenPaymentConfig(%+v) = %v, want nil", row, err) + } +} diff --git a/src/controller/admin/order_controller.go b/src/controller/admin/order_controller.go index 243e744..c045ceb 100644 --- a/src/controller/admin/order_controller.go +++ b/src/controller/admin/order_controller.go @@ -141,6 +141,8 @@ func (c *BaseAdminController) CloseOrder(ctx echo.Context) error { // @Summary Mark order paid // @Description Manually mark a waiting order as paid (operator补单) // @Description TON accepts canonical ton:::, lt:hash, or a unique recent hash-only reference for the order receive address. +// @Description Aptos accepts a transaction hash. +// @Description Aptos automatic scanning polls fullnode ledger-version transaction ranges with a runtime cursor. // @Tags Admin Orders // @Security AdminJWT // @Accept json diff --git a/src/controller/comm/pay_controller.go b/src/controller/comm/pay_controller.go index 45bef1b..b6703e1 100644 --- a/src/controller/comm/pay_controller.go +++ b/src/controller/comm/pay_controller.go @@ -81,6 +81,8 @@ func (c *BaseCommController) CheckStatus(ctx echo.Context) (err error) { // @Summary Submit transaction hash // @Description Submit an on-chain transaction hash from the cashier. The hash is verified against the order and, when valid, the same manual payment processing path used by admin mark-paid is executed. OkPay/provider orders are not supported. // @Description TON accepts canonical ton:::, lt:hash, or a unique recent hash-only reference for the order receive address. +// @Description Aptos accepts a transaction hash. +// @Description Aptos automatic scanning polls fullnode ledger-version transaction ranges with a runtime cursor. // @Tags Payment // @Accept json // @Produce json diff --git a/src/controller/comm/supported_asset_controller.go b/src/controller/comm/supported_asset_controller.go index c8aa38e..e447e86 100644 --- a/src/controller/comm/supported_asset_controller.go +++ b/src/controller/comm/supported_asset_controller.go @@ -10,6 +10,7 @@ import ( "github.com/GMWalletApp/epusdt/model/data" "github.com/GMWalletApp/epusdt/model/mdb" "github.com/GMWalletApp/epusdt/model/response" + "github.com/GMWalletApp/epusdt/model/service" "github.com/GMWalletApp/epusdt/util/constant" "github.com/labstack/echo/v4" ) @@ -44,6 +45,9 @@ func buildSupportedAssets() ([]response.NetworkTokenSupport, error) { } symbols := make([]string, 0, len(tokens)) for _, t := range tokens { + if !service.ChainTokenReadyForPayment(t) { + continue + } sym := strings.ToUpper(strings.TrimSpace(t.Symbol)) if sym == "" { continue diff --git a/src/controller/comm/supported_asset_controller_test.go b/src/controller/comm/supported_asset_controller_test.go new file mode 100644 index 0000000..ad6a017 --- /dev/null +++ b/src/controller/comm/supported_asset_controller_test.go @@ -0,0 +1,51 @@ +package comm + +import ( + "testing" + + "github.com/GMWalletApp/epusdt/internal/testutil" + "github.com/GMWalletApp/epusdt/model/dao" + "github.com/GMWalletApp/epusdt/model/data" + "github.com/GMWalletApp/epusdt/model/mdb" +) + +func TestBuildSupportedAssetsIncludesConfiguredAptosTokens(t *testing.T) { + cleanup := testutil.SetupTestDatabases(t) + defer cleanup() + + if err := dao.Mdb.Create(&mdb.Chain{Network: mdb.NetworkAptos, Enabled: true, DisplayName: "Aptos"}).Error; err != nil { + t.Fatalf("create aptos chain: %v", err) + } + if _, err := data.AddWalletAddressWithNetwork(mdb.NetworkAptos, "0x1"); err != nil { + t.Fatalf("add aptos wallet: %v", err) + } + for _, row := range []mdb.ChainToken{ + {Network: mdb.NetworkAptos, Symbol: "USDT", ContractAddress: "0x357b0b74bc833e95a115ad22604854d6b0fca151cecd94111770e5d6ffc9dc2b", Decimals: 6, Enabled: true}, + {Network: mdb.NetworkAptos, Symbol: "MOVEUSD", ContractAddress: "0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", Decimals: 6, Enabled: true}, + {Network: mdb.NetworkAptos, Symbol: "USDC", ContractAddress: "", Decimals: 6, Enabled: true}, + } { + if err := dao.Mdb.Create(&row).Error; err != nil { + t.Fatalf("create token %s: %v", row.Symbol, err) + } + } + + supports, err := buildSupportedAssets() + if err != nil { + t.Fatalf("buildSupportedAssets: %v", err) + } + if len(supports) != 1 { + t.Fatalf("supported assets len = %d, want 1: %#v", len(supports), supports) + } + if supports[0].Network != mdb.NetworkAptos { + t.Fatalf("network = %q, want %q", supports[0].Network, mdb.NetworkAptos) + } + want := []string{"MOVEUSD", "USDT"} + if len(supports[0].Tokens) != len(want) { + t.Fatalf("tokens = %#v, want %#v", supports[0].Tokens, want) + } + for i := range want { + if supports[0].Tokens[i] != want[i] { + t.Fatalf("tokens = %#v, want %#v", supports[0].Tokens, want) + } + } +} diff --git a/src/model/dao/mdb_table_init.go b/src/model/dao/mdb_table_init.go index 7357168..1512bf9 100644 --- a/src/model/dao/mdb_table_init.go +++ b/src/model/dao/mdb_table_init.go @@ -65,6 +65,7 @@ func seedChains() { {Network: mdb.NetworkPolygon, DisplayName: "Polygon", Enabled: true, MinConfirmations: 3, ScanIntervalSec: 5}, {Network: mdb.NetworkPlasma, DisplayName: "Plasma", Enabled: true, MinConfirmations: 1, ScanIntervalSec: 5}, {Network: mdb.NetworkTon, DisplayName: "TON", Enabled: true, MinConfirmations: 1, ScanIntervalSec: 5}, + {Network: mdb.NetworkAptos, DisplayName: "Aptos", Enabled: true, MinConfirmations: 1, ScanIntervalSec: 5}, } if err := Mdb.Clauses(clause.OnConflict{DoNothing: true}).Create(&defaults).Error; err != nil { color.Red.Printf("[store_db] seed chains err=%s\n", err) @@ -102,6 +103,8 @@ func seedChainTokens() { // TON {Network: mdb.NetworkTon, Symbol: "TON", ContractAddress: "", Decimals: 9, Enabled: true}, {Network: mdb.NetworkTon, Symbol: "USDT", ContractAddress: "0:b113a994b5024a16719f69139328eb759596c38a25f59028b146fecdc3621dfe", Decimals: 6, Enabled: true}, + {Network: mdb.NetworkAptos, Symbol: "USDC", ContractAddress: "0xbae207659db88bea0cbead6da0ed00aac12edcdda169e591cd41c94180b46f3b", Decimals: 6, Enabled: true}, + {Network: mdb.NetworkAptos, Symbol: "USDT", ContractAddress: "0x357b0b74bc833e95a115ad22604854d6b0fca151cecd94111770e5d6ffc9dc2b", Decimals: 6, Enabled: true}, } if err := Mdb.Clauses(clause.OnConflict{DoNothing: true}).Create(&defaults).Error; err != nil { color.Red.Printf("[store_db] seed chain_tokens err=%s\n", err) @@ -138,6 +141,7 @@ func defaultRpcNodes() []mdb.RpcNode { {Network: mdb.NetworkPolygon, Url: "wss://polygon-bor-rpc.publicnode.com", Type: mdb.RpcNodeTypeWs, Weight: 1, Enabled: true, Purpose: mdb.RpcNodePurposeGeneral, Status: mdb.RpcNodeStatusUnknown}, {Network: mdb.NetworkPlasma, Url: "wss://rpc.plasma.to", Type: mdb.RpcNodeTypeWs, Weight: 1, Enabled: true, Purpose: mdb.RpcNodePurposeGeneral, Status: mdb.RpcNodeStatusUnknown}, {Network: mdb.NetworkTon, Url: "https://ton-blockchain.github.io/global.config.json", Type: mdb.RpcNodeTypeLite, Weight: 1, Enabled: true, Purpose: mdb.RpcNodePurposeGeneral, Status: mdb.RpcNodeStatusUnknown}, + {Network: mdb.NetworkAptos, Url: "https://aptos-rest.publicnode.com/", Type: mdb.RpcNodeTypeHttp, Weight: 1, Enabled: true, Purpose: mdb.RpcNodePurposeGeneral, Status: mdb.RpcNodeStatusUnknown}, {Network: mdb.NetworkEthereum, Url: "https://rpc.epusdt.com/ethereum", Type: mdb.RpcNodeTypeHttp, Weight: 1, Enabled: true, Purpose: mdb.RpcNodePurposeManualVerify, Status: mdb.RpcNodeStatusUnknown}, {Network: mdb.NetworkBsc, Url: "https://rpc.epusdt.com/binance", Type: mdb.RpcNodeTypeHttp, Weight: 1, Enabled: true, Purpose: mdb.RpcNodePurposeManualVerify, Status: mdb.RpcNodeStatusUnknown}, {Network: mdb.NetworkPolygon, Url: "https://rpc.epusdt.com/polygon", Type: mdb.RpcNodeTypeHttp, Weight: 1, Enabled: true, Purpose: mdb.RpcNodePurposeManualVerify, Status: mdb.RpcNodeStatusUnknown}, diff --git a/src/model/dao/mdb_table_init_test.go b/src/model/dao/mdb_table_init_test.go index 662ce6d..7e1be57 100644 --- a/src/model/dao/mdb_table_init_test.go +++ b/src/model/dao/mdb_table_init_test.go @@ -2,6 +2,7 @@ package dao import ( "path/filepath" + "strings" "testing" "github.com/GMWalletApp/epusdt/model/mdb" @@ -74,6 +75,80 @@ func TestDefaultRpcNodesIncludesTonLiteGeneralNode(t *testing.T) { } } +func TestDefaultRpcNodesIncludesAptosPublicNode(t *testing.T) { + var got *mdb.RpcNode + nodes := defaultRpcNodes() + for i := range nodes { + node := nodes[i] + if node.Network == mdb.NetworkAptos { + got = &node + break + } + } + + if got == nil { + t.Fatal("missing Aptos seed rpc node") + } + if got.Url != "https://aptos-rest.publicnode.com/" { + t.Fatalf("Aptos seed url = %q", got.Url) + } + if got.Type != mdb.RpcNodeTypeHttp { + t.Fatalf("Aptos seed type = %q, want %q", got.Type, mdb.RpcNodeTypeHttp) + } + if got.Purpose != mdb.RpcNodePurposeGeneral { + t.Fatalf("Aptos seed purpose = %q, want %q", got.Purpose, mdb.RpcNodePurposeGeneral) + } + if !got.Enabled { + t.Fatal("Aptos seed enabled = false, want true") + } + if got.Status != mdb.RpcNodeStatusUnknown { + t.Fatalf("Aptos seed status = %q, want %q", got.Status, mdb.RpcNodeStatusUnknown) + } +} + +func TestSeedChainsIncludesAptos(t *testing.T) { + db := setupSeedTableTestDB(t, &mdb.Chain{}) + Mdb = db + + seedChains() + + var row mdb.Chain + if err := Mdb.Where("network = ?", mdb.NetworkAptos).Take(&row).Error; err != nil { + t.Fatalf("load Aptos chain seed: %v", err) + } + if !row.Enabled { + t.Fatal("Aptos chain enabled = false, want true") + } +} + +func TestSeedChainTokensIncludesAptosAssets(t *testing.T) { + db := setupSeedTableTestDB(t, &mdb.ChainToken{}) + Mdb = db + + seedChainTokens() + + wantEnabled := map[string]bool{ + mdb.NetworkAptos + "/USDC": true, + mdb.NetworkAptos + "/USDT": true, + } + wantContract := map[string]string{ + mdb.NetworkAptos + "/USDT": "0x357b0b74bc833e95a115ad22604854d6b0fca151cecd94111770e5d6ffc9dc2b", + } + for key, enabled := range wantEnabled { + parts := strings.Split(key, "/") + var row mdb.ChainToken + if err := Mdb.Where("network = ? AND symbol = ?", parts[0], parts[1]).Take(&row).Error; err != nil { + t.Fatalf("load token seed %s: %v", key, err) + } + if row.Enabled != enabled { + t.Fatalf("%s enabled = %v, want %v", key, row.Enabled, enabled) + } + if want, ok := wantContract[key]; ok && row.ContractAddress != want { + t.Fatalf("%s contract_address = %q, want %q", key, row.ContractAddress, want) + } + } +} + func TestSeedDefaultSettingsIncludesSystemLogLevel(t *testing.T) { db := setupSeedSettingsTestDB(t) Mdb = db @@ -194,3 +269,19 @@ func setupSeedSettingsTestDB(t *testing.T) *gorm.DB { } return db } + +func setupSeedTableTestDB(t *testing.T, models ...interface{}) *gorm.DB { + t.Helper() + oldDB := Mdb + t.Cleanup(func() { + Mdb = oldDB + }) + db, err := gorm.Open(sqlite.Open(filepath.Join(t.TempDir(), "seed-table.db")), &gorm.Config{}) + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + if err := db.AutoMigrate(models...); err != nil { + t.Fatalf("migrate seed table: %v", err) + } + return db +} diff --git a/src/model/data/order_data.go b/src/model/data/order_data.go index f358c65..e876060 100644 --- a/src/model/data/order_data.go +++ b/src/model/data/order_data.go @@ -49,6 +49,11 @@ func normalizeLockAddress(network, address string) string { return raw } } + if network == mdb.NetworkAptos { + if normalized, err := addressutil.NormalizeMoveAddress(address); err == nil { + return normalized + } + } return address } diff --git a/src/model/data/order_data_test.go b/src/model/data/order_data_test.go index e658cc7..9f48536 100644 --- a/src/model/data/order_data_test.go +++ b/src/model/data/order_data_test.go @@ -184,3 +184,19 @@ func TestTonTransactionLockAddressUsesRawKey(t *testing.T) { t.Fatalf("ton raw lock lookup = %q, want trade-ton", gotTradeID) } } + +func TestAptosTransactionLockAddressUsesCanonicalKey(t *testing.T) { + cleanup := testutil.SetupTestDatabases(t) + defer cleanup() + + if err := LockTransaction(mdb.NetworkAptos, "0xA", "USDT", "trade-aptos", 1.23, time.Hour); err != nil { + t.Fatalf("lock aptos transaction: %v", err) + } + gotTradeID, err := GetTradeIdByWalletAddressAndAmountAndToken(mdb.NetworkAptos, "a", "USDT", 1.23) + if err != nil { + t.Fatalf("lookup aptos lock: %v", err) + } + if gotTradeID != "trade-aptos" { + t.Fatalf("aptos lock lookup = %q, want trade-aptos", gotTradeID) + } +} diff --git a/src/model/data/wallet_address_data.go b/src/model/data/wallet_address_data.go index 3ea892a..14902c1 100644 --- a/src/model/data/wallet_address_data.go +++ b/src/model/data/wallet_address_data.go @@ -41,6 +41,9 @@ func normalizeWalletAddressByNetworkE(network, address string) (string, error) { if network == mdb.NetworkTon { return addressutil.NormalizeTonAddress(address) } + if network == mdb.NetworkAptos { + return addressutil.NormalizeMoveAddress(address) + } return normalizeWalletAddressByNetwork(network, address), nil } diff --git a/src/model/data/wallet_address_data_test.go b/src/model/data/wallet_address_data_test.go index d0c3c4c..65fa5dc 100644 --- a/src/model/data/wallet_address_data_test.go +++ b/src/model/data/wallet_address_data_test.go @@ -103,3 +103,20 @@ func TestAddWalletAddressWithNetworkNormalizesTonAddressVariants(t *testing.T) { t.Fatalf("add equivalent ton wallet error = %v, want already exists", err) } } + +func TestAddWalletAddressWithNetworkNormalizesMoveAddressVariants(t *testing.T) { + cleanup := testutil.SetupTestDatabases(t) + defer cleanup() + + want := "0x000000000000000000000000000000000000000000000000000000000000000a" + row, err := AddWalletAddressWithNetwork(mdb.NetworkAptos, " A ") + if err != nil { + t.Fatalf("add aptos wallet: %v", err) + } + if row.Address != want { + t.Fatalf("stored Aptos address = %q, want %q", row.Address, want) + } + if _, err = AddWalletAddressWithNetwork(mdb.NetworkAptos, "0x0A"); err != constant.WalletAddressAlreadyExists { + t.Fatalf("add equivalent aptos wallet error = %v, want already exists", err) + } +} diff --git a/src/model/mdb/chain_token.go b/src/model/mdb/chain_token.go index 5698563..e8181b2 100644 --- a/src/model/mdb/chain_token.go +++ b/src/model/mdb/chain_token.go @@ -5,10 +5,11 @@ const ( ChainTokenStatusDisable = 2 ) -// ChainToken describes a token to watch on a given chain. Replaces the -// hardcoded USDT/USDC contract addresses in task/listen_eth.go. Scanners -// load this table on startup (and refresh periodically) to learn which -// contracts to subscribe to. +// ChainToken describes a token to watch on a given chain. For EVM/TRON rows +// ContractAddress is the token contract; for Solana it is the mint; for +// TON it is the jetton master; for Aptos it is the Move asset identifier +// (coin type, metadata address or object type). Native rows use an empty +// ContractAddress. type ChainToken struct { Network string `gorm:"column:network;size:32;uniqueIndex:chain_tokens_network_symbol_uindex,priority:1" json:"network" example:"tron"` Symbol string `gorm:"column:symbol;size:32;uniqueIndex:chain_tokens_network_symbol_uindex,priority:2" json:"symbol" example:"USDT"` diff --git a/src/model/mdb/wallet_address_mdb.go b/src/model/mdb/wallet_address_mdb.go index 6f8e9a5..423c4ec 100644 --- a/src/model/mdb/wallet_address_mdb.go +++ b/src/model/mdb/wallet_address_mdb.go @@ -13,6 +13,7 @@ const ( NetworkPolygon = "polygon" NetworkPlasma = "plasma" NetworkTon = "ton" + NetworkAptos = "aptos" ) const ( diff --git a/src/model/request/order_request.go b/src/model/request/order_request.go index c4d61b3..0ecfc96 100644 --- a/src/model/request/order_request.go +++ b/src/model/request/order_request.go @@ -7,7 +7,7 @@ type CreateTransactionRequest struct { OrderId string `json:"order_id" form:"order_id" validate:"required|maxLen:32" example:"ORD20260416001"` Currency string `json:"currency" form:"currency" validate:"required" example:"cny"` // 法币 如：cny Token string `json:"token" form:"token" example:"usdt"` // 币种 如：usdt、ton；可与 network 同时缺省创建占位订单 - Network string `json:"network" form:"network" example:"tron"` // 网络 如：ton、tron；可与 token 同时缺省创建占位订单 + Network string `json:"network" form:"network" example:"tron"` // 网络 如：ton、tron、aptos；可与 token 同时缺省创建占位订单 Amount float64 `json:"amount" form:"amount" validate:"required|isFloat|gt:0.01" example:"100.00"` NotifyUrl string `json:"notify_url" form:"notify_url" validate:"required" example:"https://example.com/notify"` Signature string `json:"signature" form:"signature" validate:"required" example:"a1b2c3d4e5f6..."` @@ -58,7 +58,7 @@ func (r ManualPaymentRequest) Translates() map[string]string { type SwitchNetworkRequest struct { TradeId string `json:"trade_id" validate:"required" example:"3nQ9pL2xV7sK1mR8cT4yB_aZ"` Token string `json:"token" validate:"required" example:"USDT"` - Network string `json:"network" validate:"required" example:"okpay,tron,solana,ethereum"` + Network string `json:"network" validate:"required" example:"okpay,tron,solana,ethereum,aptos"` } func (r SwitchNetworkRequest) Translates() map[string]string { diff --git a/src/model/service/aptos_task.go b/src/model/service/aptos_task.go new file mode 100644 index 0000000..1db459d --- /dev/null +++ b/src/model/service/aptos_task.go @@ -0,0 +1,526 @@ +package service + +import ( + "fmt" + "io" + "math/big" + "net/http" + "strconv" + "strings" + + "github.com/GMWalletApp/epusdt/model/data" + "github.com/GMWalletApp/epusdt/model/mdb" + addressutil "github.com/GMWalletApp/epusdt/util/address" + "github.com/GMWalletApp/epusdt/util/log" + "github.com/tidwall/gjson" +) + +func AptosGetLedgerVersion() (int64, error) { + body, err := aptosGet("/v1") + if err != nil { + return 0, err + } + latest := gjson.GetBytes(body, "ledger_version").Int() + log.Sugar.Debugf("[APTOS] latest ledger version=%d", latest) + return latest, nil +} + +func AptosGetTransactionByHash(txID string) ([]byte, error) { + txID = normalizeAptosTxID(txID) + log.Sugar.Infof("[APTOS] fetch transaction by hash tx=%s", txID) + return aptosGet(fmt.Sprintf("/v1/transactions/by_hash/%s", txID)) +} + +func AptosGetTransactions(start int64, limit int64) ([]byte, error) { + if start < 0 { + start = 0 + } + if limit <= 0 { + limit = 1 + } + log.Sugar.Debugf("[APTOS] fetch transactions start=%d limit=%d", start, limit) + return aptosGet(fmt.Sprintf("/v1/transactions?start=%d&limit=%d", start, limit)) +} + +func ParseAptosTransfers(body []byte, address string, tokens []mdb.ChainToken) ([]MoveObservedTransfer, error) { + receive, err := addressutil.NormalizeMoveAddress(address) + if err != nil { + return nil, err + } + return ParseAptosTransfersForWallets(body, map[string]struct{}{receive: {}}, tokens) +} + +func ParseAptosTransfersForWallets(body []byte, wallets map[string]struct{}, tokens []mdb.ChainToken) ([]MoveObservedTransfer, error) { + root := gjson.ParseBytes(body) + if !root.IsArray() { + return nil, fmt.Errorf("unexpected Aptos transactions response") + } + var out []MoveObservedTransfer + for _, tx := range root.Array() { + if tx.Get("type").String() != "" && tx.Get("type").String() != "user_transaction" { + continue + } + if !tx.Get("success").Bool() { + continue + } + txID := strings.TrimSpace(tx.Get("hash").String()) + if txID == "" { + continue + } + blockTimeMs := parseAptosTimestampMs(tx.Get("timestamp")) + version := tx.Get("version").Int() + txctx := buildAptosTransferContext(tx) + events := buildAptosFungibleEvents(tx, txctx, tokens) + deposits := make([]aptosFungibleEvent, 0) + withdrawals := make(map[string]int) + for _, event := range events { + if event.token == nil { + continue + } + switch event.action { + case "withdraw": + withdrawals[aptosTransferMatchKey(event.token.Symbol, event.rawAmount)]++ + case "deposit": + deposits = append(deposits, event) + } + } + usedDeposits := make([]bool, len(deposits)) + for i, event := range deposits { + matchKey := aptosTransferMatchKey(event.token.Symbol, event.rawAmount) + if withdrawals[matchKey] <= 0 { + continue + } + if appendAptosObservedTransfer(&out, txID, version, blockTimeMs, event, wallets) { + withdrawals[matchKey]-- + usedDeposits[i] = true + } else { + logAptosUnwatchedDeposit(txID, version, event, "direct") + } + } + for i := 0; i < len(deposits); i++ { + if usedDeposits[i] { + continue + } + for j := i + 1; j < len(deposits); j++ { + if usedDeposits[j] || !aptosDepositPairCanMatch(deposits[i], deposits[j]) { + continue + } + sum := new(big.Int).Add(deposits[i].rawAmount, deposits[j].rawAmount) + matchKey := aptosTransferMatchKey(deposits[i].token.Symbol, sum) + if withdrawals[matchKey] <= 0 { + continue + } + appendedFirst := appendAptosObservedTransfer(&out, txID, version, blockTimeMs, deposits[i], wallets) + appendedSecond := appendAptosObservedTransfer(&out, txID, version, blockTimeMs, deposits[j], wallets) + if !appendedFirst && !appendedSecond { + logAptosUnwatchedDeposit(txID, version, deposits[i], "split") + logAptosUnwatchedDeposit(txID, version, deposits[j], "split") + continue + } + withdrawals[matchKey]-- + usedDeposits[i] = true + usedDeposits[j] = true + break + } + } + } + return out, nil +} + +func appendAptosObservedTransfer(out *[]MoveObservedTransfer, txID string, version int64, blockTimeMs int64, event aptosFungibleEvent, wallets map[string]struct{}) bool { + recipient := event.owner + if !moveWalletWatched(recipient, wallets) { + return false + } + *out = append(*out, MoveObservedTransfer{ + Network: mdb.NetworkAptos, + ReceiveAddress: recipient, + Token: NormalizeAptosPaymentSymbol(event.token.Symbol), + RawAmount: event.rawAmount, + Decimals: event.token.Decimals, + MinAmount: event.token.MinAmount, + Amount: rawToFloat(event.rawAmount, event.token.Decimals), + TxID: normalizeAptosTxID(txID), + BlockTimeMs: blockTimeMs, + Version: version, + TransferKey: moveAptosTransferKey(version, event.eventIndex, recipient, event.token.Symbol, event.rawAmount), + }) + return true +} + +func logAptosUnwatchedDeposit(txID string, version int64, event aptosFungibleEvent, matchMode string) { + log.Sugar.Debugf( + "[APTOS] matched deposit skipped owner_not_watched mode=%s version=%d tx=%s token=%s amount=%.8f raw=%s owner=%s", + matchMode, + version, + normalizeAptosTxID(txID), + NormalizeAptosPaymentSymbol(event.token.Symbol), + rawToFloat(event.rawAmount, event.token.Decimals), + event.rawAmount.String(), + event.owner, + ) +} + +func aptosDepositPairCanMatch(first, second aptosFungibleEvent) bool { + if first.token == nil || second.token == nil || first.rawAmount == nil || second.rawAmount == nil { + return false + } + if NormalizeAptosPaymentSymbol(first.token.Symbol) != NormalizeAptosPaymentSymbol(second.token.Symbol) { + return false + } + return first.token.ContractAddress == second.token.ContractAddress +} + +type aptosFungibleEvent struct { + eventIndex int + action string + owner string + rawAmount *big.Int + token *mdb.ChainToken +} + +func buildAptosFungibleEvents(tx gjson.Result, ctx aptosTransferContext, tokens []mdb.ChainToken) []aptosFungibleEvent { + out := make([]aptosFungibleEvent, 0) + for eventIndex, event := range tx.Get("events").Array() { + action := aptosFungibleEventAction(event) + if action == "" { + continue + } + rawAmount, ok := parseSignedBigInt(aptosEventRawAmount(event, ctx)) + if !ok || rawAmount.Sign() <= 0 { + continue + } + token := resolveAptosPaymentToken(aptosEventAssetID(event, ctx), tokens) + if token == nil { + continue + } + out = append(out, aptosFungibleEvent{ + eventIndex: eventIndex, + action: action, + owner: aptosEventRecipient(event, ctx), + rawAmount: rawAmount, + token: token, + }) + } + return out +} + +func aptosFungibleEventAction(event gjson.Result) string { + typ := strings.ToLower(strings.TrimSpace(event.Get("type").String())) + switch typ { + case "0x1::fungible_asset::deposit": + return "deposit" + case "0x1::fungible_asset::withdraw": + return "withdraw" + default: + return "" + } +} + +func aptosTransferMatchKey(symbol string, rawAmount *big.Int) string { + raw := "" + if rawAmount != nil { + raw = rawAmount.String() + } + return NormalizeAptosPaymentSymbol(symbol) + ":" + raw +} + +func resolveAptosPaymentToken(assetID string, tokens []mdb.ChainToken) *mdb.ChainToken { + return resolveMoveToken(mdb.NetworkAptos, assetID, tokens) +} + +func NormalizeAptosPaymentSymbol(symbol string) string { + return strings.ToUpper(strings.TrimSpace(symbol)) +} + +type aptosTransferContext struct { + storeOwners map[string]string + storeMetadata map[string]string +} + +func buildAptosTransferContext(tx gjson.Result) aptosTransferContext { + ctx := aptosTransferContext{ + storeOwners: make(map[string]string), + storeMetadata: make(map[string]string), + } + for _, change := range tx.Get("changes").Array() { + if change.Get("type").String() != "write_resource" { + continue + } + store, err := addressutil.NormalizeMoveAddress(change.Get("address").String()) + if err != nil { + continue + } + changeType := strings.TrimSpace(change.Get("data.type").String()) + switch changeType { + case "0x1::fungible_asset::FungibleStore": + if metadata := firstNonEmpty( + change.Get("data.data.metadata.inner").String(), + change.Get("data.data.metadata").String(), + ); metadata != "" { + ctx.storeMetadata[store] = normalizeAptosObservedAssetID(metadata) + } + case "0x1::object::ObjectCore": + if owner := firstMoveAddress(change.Get("data.data.owner").String()); owner != "" { + ctx.storeOwners[store] = owner + } + } + } + return ctx +} + +func aptosEventRecipient(event gjson.Result, ctx aptosTransferContext) string { + if store := firstMoveAddress(event.Get("data.store").String()); store != "" { + if owner := ctx.storeOwners[store]; owner != "" { + return owner + } + } + return "" +} + +func aptosEventRawAmount(event gjson.Result, ctx aptosTransferContext) string { + return event.Get("data.amount").String() +} + +func aptosEventAssetID(event gjson.Result, ctx aptosTransferContext) string { + if store := firstMoveAddress(event.Get("data.store").String()); store != "" { + if metadata := ctx.storeMetadata[store]; metadata != "" { + return metadata + } + } + return "" +} + +func normalizeAptosObservedAssetID(assetID string) string { + return addressutil.NormalizeMoveAssetID(assetID) +} + +func parseAptosTimestampMs(value gjson.Result) int64 { + raw := strings.TrimSpace(value.String()) + if raw == "" { + return 0 + } + n, err := strconv.ParseInt(raw, 10, 64) + if err != nil { + return 0 + } + return n / 1000 +} + +func normalizeAptosTxID(txID string) string { + txID = strings.ToLower(strings.TrimSpace(txID)) + if strings.HasPrefix(txID, "0x") { + return txID + } + return "0x" + txID +} + +func ValidateManualAptosPayment(order *mdb.Orders, txID string) (string, error) { + txID = normalizeAptosTxID(txID) + nodes, err := data.ListManualPaymentRpcCandidates(mdb.NetworkAptos, mdb.RpcNodeTypeHttp) + if err != nil { + return "", err + } + if len(nodes) == 0 { + return "", fmt.Errorf("no enabled %s %s RPC node configured in rpc_nodes", mdb.NetworkAptos, mdb.RpcNodeTypeHttp) + } + + var verifyErrors []string + for _, node := range nodes { + if strings.TrimSpace(node.Url) == "" { + continue + } + if _, err = validateManualAptosPaymentWithNode(order, txID, node); err != nil { + verifyErrors = append(verifyErrors, fmt.Sprintf("%s: %v", data.RpcNodeLogLabel(node), err)) + continue + } + return txID, nil + } + if len(verifyErrors) > 0 { + return "", fmt.Errorf("manual Aptos verification failed: %s", strings.Join(verifyErrors, "; ")) + } + return "", fmt.Errorf("no enabled %s %s RPC node configured in rpc_nodes", mdb.NetworkAptos, mdb.RpcNodeTypeHttp) +} + +func validateManualAptosPaymentWithNode(order *mdb.Orders, txID string, node mdb.RpcNode) (string, error) { + txID = normalizeAptosTxID(txID) + body, err := aptosGetWithNode(node, fmt.Sprintf("/v1/transactions/by_hash/%s", txID)) + if err != nil { + return "", fmt.Errorf("fetch aptos transaction: %w", err) + } + if err = validateManualAptosTransactionWithConfirm(order, txID, body, func(version int64) error { + return ensureAptosTransferConfirmedWithNode(version, node) + }); err != nil { + return "", err + } + return txID, nil +} + +func validateManualAptosTransaction(order *mdb.Orders, txID string, body []byte) error { + return validateManualAptosTransactionWithConfirm(order, txID, body, EnsureAptosTransferConfirmed) +} + +func validateManualAptosTransactionWithConfirm(order *mdb.Orders, txID string, body []byte, confirm func(int64) error) error { + tokens, err := data.ListEnabledChainTokensByNetwork(mdb.NetworkAptos) + if err != nil { + return err + } + transfers, err := ParseAptosTransfers([]byte("["+string(body)+"]"), order.ReceiveAddress, tokens) + if err != nil { + return err + } + amountMismatch := false + for _, transfer := range transfers { + if !strings.EqualFold(transfer.TxID, txID) { + continue + } + if err = confirm(transfer.Version); err != nil { + return err + } + if err = EnsureMoveTransferMatchesOrder(order, transfer); err == nil { + return nil + } + if strings.Contains(err.Error(), "amount mismatch") { + amountMismatch = true + } else { + return err + } + } + if amountMismatch { + return fmt.Errorf("transaction amount mismatch") + } + return fmt.Errorf("matching aptos transfer to order address not found") +} + +func aptosGet(path string) (body []byte, err error) { + network := mdb.NetworkAptos + defer recordMoveRPCResult(network, &err) + + tried := make([]uint64, 0, 3) + var lastErr error + for attempts := 0; attempts < 3; attempts++ { + node, resolveErr := resolveAptosRpcNode(tried...) + if resolveErr != nil { + if lastErr != nil { + return nil, lastErr + } + return nil, resolveErr + } + if len(tried) > 0 { + log.Sugar.Warnf("[APTOS] trying alternate RPC node path=%s node=%s", path, data.RpcNodeLogLabel(*node)) + } + + body, err = aptosGetWithNode(*node, path) + if err == nil { + data.RecordRpcNodeSuccess(node.ID) + return body, nil + } + + lastErr = err + failures, cooling := data.RecordRpcNodeFailure(node.ID) + nodeLabel := data.RpcNodeLogLabel(*node) + if cooling { + log.Sugar.Warnf("[APTOS] RPC node reached fail threshold path=%s node=%s failures=%d/%d", path, nodeLabel, failures, data.RpcFailoverThreshold) + } else { + log.Sugar.Warnf("[APTOS] RPC node failed path=%s node=%s failures=%d/%d", path, nodeLabel, failures, data.RpcFailoverThreshold) + } + tried = append(tried, node.ID) + } + return nil, lastErr +} + +func resolveAptosRpcNode(excludeIDs ...uint64) (*mdb.RpcNode, error) { + node, err := data.SelectGeneralRpcNode(mdb.NetworkAptos, mdb.RpcNodeTypeHttp, excludeIDs...) + if err != nil { + return nil, err + } + if node == nil || node.ID == 0 { + return nil, fmt.Errorf("no enabled %s %s RPC node configured in rpc_nodes", mdb.NetworkAptos, mdb.RpcNodeTypeHttp) + } + rpcURL := strings.TrimSpace(node.Url) + if rpcURL == "" { + return nil, fmt.Errorf("rpc_nodes id=%d has empty url", node.ID) + } + node.Url = rpcURL + return node, nil +} + +func aptosGetWithNode(node mdb.RpcNode, path string) (body []byte, err error) { + endpoint, err := aptosJoinURL(node.Url, path) + if err != nil { + return nil, err + } + req, err := http.NewRequest(http.MethodGet, endpoint, nil) + if err != nil { + return nil, err + } + req.Header.Set("Accept", "application/json") + resp, err := (&http.Client{Timeout: manualVerifyRequestTimeout}).Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + body, _ = io.ReadAll(resp.Body) + if resp.StatusCode >= http.StatusBadRequest { + return nil, fmt.Errorf("aptos rpc HTTP %d: %s", resp.StatusCode, string(body)) + } + return body, nil +} + +func aptosJoinURL(base, path string) (string, error) { + base = strings.TrimSpace(base) + if base == "" { + return "", fmt.Errorf("empty aptos rpc url") + } + if strings.TrimSpace(path) == "" { + return strings.TrimRight(base, "/"), nil + } + path = strings.TrimSpace(path) + trimmedBase := strings.TrimRight(base, "/") + lowerBase := strings.ToLower(trimmedBase) + if strings.HasSuffix(lowerBase, "/v1") && strings.HasPrefix(path, "/v1") { + path = strings.TrimPrefix(path, "/v1") + if path == "" { + path = "/" + } + } + return trimmedBase + path, nil +} + +func EnsureAptosTransferConfirmed(version int64) error { + if version <= 0 { + return fmt.Errorf("aptos transaction version missing") + } + latest, err := AptosGetLedgerVersion() + if err != nil { + return fmt.Errorf("fetch aptos latest ledger version: %w", err) + } + return ensureAptosConfirmations(version, latest) +} + +func ensureAptosTransferConfirmedWithNode(version int64, node mdb.RpcNode) error { + if version <= 0 { + return fmt.Errorf("aptos transaction version missing") + } + body, err := aptosGetWithNode(node, "/v1") + if err != nil { + return fmt.Errorf("fetch aptos latest ledger version: %w", err) + } + latest := gjson.GetBytes(body, "ledger_version").Int() + return ensureAptosConfirmations(version, latest) +} + +func ensureAptosConfirmations(version, latest int64) error { + chain, err := data.GetChainByNetwork(mdb.NetworkAptos) + if err != nil { + return err + } + minConfirmations := int64(1) + if chain != nil && chain.MinConfirmations > 0 { + minConfirmations = int64(chain.MinConfirmations) + } + if latest-version+1 < minConfirmations { + return fmt.Errorf("aptos transaction confirmations insufficient") + } + return nil +} diff --git a/src/model/service/manual_payment_verify.go b/src/model/service/manual_payment_verify.go index 33e57d2..3fe82e6 100644 --- a/src/model/service/manual_payment_verify.go +++ b/src/model/service/manual_payment_verify.go @@ -89,6 +89,8 @@ func validateManualOrderPaymentDefault(order *mdb.Orders, blockTransactionID str canonicalTxID, err = validateManualSolanaPayment(order, txID) case mdb.NetworkTon: canonicalTxID, err = validateManualTonPayment(order, txID) + case mdb.NetworkAptos: + canonicalTxID, err = ValidateManualAptosPayment(order, txID) case mdb.NetworkEthereum, mdb.NetworkBsc, mdb.NetworkPolygon, mdb.NetworkPlasma: canonicalTxID, err = validateManualEvmPayment(order, txID) default: @@ -123,7 +125,7 @@ func ensureManualBlockTransactionUnused(order *mdb.Orders, canonicalTxID string) func manualBlockTransactionIDIsHex(network string) bool { switch strings.ToLower(strings.TrimSpace(network)) { - case mdb.NetworkTron, mdb.NetworkEthereum, mdb.NetworkBsc, mdb.NetworkPolygon, mdb.NetworkPlasma: + case mdb.NetworkTron, mdb.NetworkEthereum, mdb.NetworkBsc, mdb.NetworkPolygon, mdb.NetworkPlasma, mdb.NetworkAptos: return true default: return false @@ -149,7 +151,7 @@ func equivalentManualBlockTransactionIDs(network, canonicalTxID string) []string add(canonicalTxID) switch network { - case mdb.NetworkEthereum, mdb.NetworkBsc, mdb.NetworkPolygon, mdb.NetworkPlasma: + case mdb.NetworkEthereum, mdb.NetworkBsc, mdb.NetworkPolygon, mdb.NetworkPlasma, mdb.NetworkAptos: body := strings.TrimPrefix(strings.TrimPrefix(canonicalTxID, "0x"), "0X") body = strings.ToLower(body) add("0x" + body) diff --git a/src/model/service/manual_payment_verify_test.go b/src/model/service/manual_payment_verify_test.go index e49c9fb..69b7702 100644 --- a/src/model/service/manual_payment_verify_test.go +++ b/src/model/service/manual_payment_verify_test.go @@ -113,6 +113,66 @@ func TestManualVerifyNormalizeTronTxIDAcceptsOptional0x(t *testing.T) { } } +func TestValidateManualAptosPayment(t *testing.T) { + cleanup := testutil.SetupTestDatabases(t) + defer cleanup() + + if err := dao.Mdb.Create(&mdb.Chain{Network: mdb.NetworkAptos, Enabled: true, MinConfirmations: 1}).Error; err != nil { + t.Fatalf("create aptos chain: %v", err) + } + usdc := "0xbae207659db88bea0cbead6da0ed00aac12edcdda169e591cd41c94180b46f3b" + if err := dao.Mdb.Create(&mdb.ChainToken{ + Network: mdb.NetworkAptos, + Symbol: "USDC", + ContractAddress: usdc, + Decimals: 6, + Enabled: true, + }).Error; err != nil { + t.Fatalf("create aptos usdc token: %v", err) + } + + receive, err := addressutil.NormalizeMoveAddress("0xa") + if err != nil { + t.Fatalf("normalize aptos receive: %v", err) + } + store, err := addressutil.NormalizeMoveAddress("0x11") + if err != nil { + t.Fatalf("normalize aptos store: %v", err) + } + senderStore, err := addressutil.NormalizeMoveAddress("0x12") + if err != nil { + t.Fatalf("normalize aptos sender store: %v", err) + } + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + switch { + case strings.HasPrefix(r.URL.Path, "/v1/transactions/by_hash/"): + _, _ = w.Write([]byte(`{"type":"user_transaction","success":true,"hash":"0xabc","version":"101","timestamp":"1700000000123456","payload":{"function":"0x1::primary_fungible_store::transfer","arguments":[{"inner":"` + usdc + `"},"` + receive + `","4200000"],"type":"entry_function_payload"},"events":[{"type":"0x1::fungible_asset::Withdraw","data":{"amount":"4200000","store":"` + senderStore + `"}},{"type":"0x1::fungible_asset::Deposit","data":{"amount":"4200000","store":"` + store + `"}}],"changes":[{"address":"` + senderStore + `","data":{"type":"0x1::fungible_asset::FungibleStore","data":{"metadata":{"inner":"` + usdc + `"}}},"type":"write_resource"},{"address":"` + store + `","data":{"type":"0x1::fungible_asset::FungibleStore","data":{"metadata":{"inner":"` + usdc + `"}}},"type":"write_resource"},{"address":"` + store + `","data":{"type":"0x1::object::ObjectCore","data":{"owner":"` + receive + `"}},"type":"write_resource"}]}`)) + case r.URL.Path == "/v1": + _, _ = w.Write([]byte(`{"ledger_version":"101"}`)) + default: + http.NotFound(w, r) + } + })) + defer server.Close() + seedTestAptosRPCNode(t, server.URL, mdb.RpcNodePurposeGeneral, mdb.RpcNodeStatusOk, 100) + + order := &mdb.Orders{ + BaseModel: mdb.BaseModel{ID: 1, CreatedAt: *carbon.NewTime(carbon.CreateFromTimestampMilli(1699999999000))}, + Network: mdb.NetworkAptos, + Token: "USDC", + ActualAmount: 4.2, + ReceiveAddress: receive, + } + got, err := ValidateManualAptosPayment(order, "ABC") + if err != nil { + t.Fatalf("ValidateManualAptosPayment(): %v", err) + } + if got != "0xabc" { + t.Fatalf("canonical tx id = %q, want %q", got, "0xabc") + } +} + func TestParseManualTonTxRefAcceptsCanonicalLTAndHashOnly(t *testing.T) { receiveRaw := "0:ba295e33b3c4c9b5265aa4ead1166a92931ce9abea120a8c5e91044a1257f89c" hash := strings.Repeat("a", 64) diff --git a/src/model/service/move_rpc_routing_test.go b/src/model/service/move_rpc_routing_test.go new file mode 100644 index 0000000..56912af --- /dev/null +++ b/src/model/service/move_rpc_routing_test.go @@ -0,0 +1,203 @@ +package service + +import ( + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/GMWalletApp/epusdt/internal/testutil" + "github.com/GMWalletApp/epusdt/model/dao" + "github.com/GMWalletApp/epusdt/model/data" + "github.com/GMWalletApp/epusdt/model/mdb" + addressutil "github.com/GMWalletApp/epusdt/util/address" + "github.com/dromara/carbon/v2" +) + +func seedTestAptosRPCNode(t *testing.T, url, purpose, status string, weight int) mdb.RpcNode { + t.Helper() + node := mdb.RpcNode{ + Network: mdb.NetworkAptos, + Url: url, + Type: mdb.RpcNodeTypeHttp, + Weight: weight, + Enabled: true, + Purpose: purpose, + Status: status, + } + if node.Purpose == "" { + node.Purpose = mdb.RpcNodePurposeGeneral + } + if node.Status == "" { + node.Status = mdb.RpcNodeStatusOk + } + if err := dao.Mdb.Create(&node).Error; err != nil { + t.Fatalf("seed aptos rpc_nodes: %v", err) + } + return node +} + +func TestAptosGetLedgerVersionUsesRpcNodes(t *testing.T) { + cleanup := testutil.SetupTestDatabases(t) + defer cleanup() + data.ResetRpcFailoverForTest() + t.Cleanup(data.ResetRpcFailoverForTest) + + var disabledCalls int + disabled := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + disabledCalls++ + _, _ = w.Write([]byte(`{"ledger_version":"999"}`)) + })) + defer disabled.Close() + disabledNode := mdb.RpcNode{ + Network: mdb.NetworkAptos, + Url: disabled.URL, + Type: mdb.RpcNodeTypeHttp, + Weight: 1000, + Enabled: false, + Purpose: mdb.RpcNodePurposeGeneral, + Status: mdb.RpcNodeStatusOk, + } + if err := dao.Mdb.Create(&disabledNode).Error; err != nil { + t.Fatalf("seed disabled aptos rpc_nodes: %v", err) + } + if err := dao.Mdb.Model(&mdb.RpcNode{}).Where("id = ?", disabledNode.ID).Update("enabled", false).Error; err != nil { + t.Fatalf("disable aptos rpc_nodes: %v", err) + } + + var calls int + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls++ + if r.URL.Path != "/v1" { + http.NotFound(w, r) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"ledger_version":"123"}`)) + })) + defer server.Close() + seedTestAptosRPCNode(t, server.URL, mdb.RpcNodePurposeGeneral, mdb.RpcNodeStatusOk, 100) + + got, err := AptosGetLedgerVersion() + if err != nil { + t.Fatalf("AptosGetLedgerVersion(): %v", err) + } + if got != 123 { + t.Fatalf("ledger version = %d, want 123", got) + } + if calls != 1 { + t.Fatalf("rpc calls = %d, want 1", calls) + } + if disabledCalls != 0 { + t.Fatalf("disabled rpc calls = %d, want 0", disabledCalls) + } +} + +func TestAptosGetTransactionsFallsBackToAlternateRpcNode(t *testing.T) { + cleanup := testutil.SetupTestDatabases(t) + defer cleanup() + data.ResetRpcFailoverForTest() + t.Cleanup(data.ResetRpcFailoverForTest) + + var primaryCalls int + primary := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + primaryCalls++ + http.Error(w, "temporary failure", http.StatusBadGateway) + })) + defer primary.Close() + var backupCalls int + backup := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + backupCalls++ + if r.URL.Path != "/v1/transactions" || r.URL.Query().Get("start") != "10" || r.URL.Query().Get("limit") != "2" { + http.NotFound(w, r) + return + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`[]`)) + })) + defer backup.Close() + seedTestAptosRPCNode(t, primary.URL, mdb.RpcNodePurposeGeneral, mdb.RpcNodeStatusOk, 100) + seedTestAptosRPCNode(t, backup.URL, mdb.RpcNodePurposeGeneral, mdb.RpcNodeStatusOk, 1) + + body, err := AptosGetTransactions(10, 2) + if err != nil { + t.Fatalf("AptosGetTransactions(): %v", err) + } + if string(body) != "[]" { + t.Fatalf("transactions body = %s, want []", string(body)) + } + if primaryCalls != 1 || backupCalls != 1 { + t.Fatalf("rpc calls primary=%d backup=%d, want 1/1", primaryCalls, backupCalls) + } +} + +func TestValidateManualAptosPaymentUsesManualRpcFallback(t *testing.T) { + cleanup := testutil.SetupTestDatabases(t) + defer cleanup() + + if err := dao.Mdb.Create(&mdb.Chain{Network: mdb.NetworkAptos, Enabled: true, MinConfirmations: 1}).Error; err != nil { + t.Fatalf("create aptos chain: %v", err) + } + const usdc = "0xbae207659db88bea0cbead6da0ed00aac12edcdda169e591cd41c94180b46f3b" + upsertTestChainToken(t, mdb.ChainToken{ + Network: mdb.NetworkAptos, + Symbol: "USDC", + ContractAddress: usdc, + Decimals: 6, + Enabled: true, + }) + + receive, err := addressutil.NormalizeMoveAddress("0xa") + if err != nil { + t.Fatalf("normalize aptos receive: %v", err) + } + store, err := addressutil.NormalizeMoveAddress("0x31") + if err != nil { + t.Fatalf("normalize aptos store: %v", err) + } + senderStore, err := addressutil.NormalizeMoveAddress("0x32") + if err != nil { + t.Fatalf("normalize aptos sender store: %v", err) + } + + var generalCalls int + general := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + generalCalls++ + http.Error(w, "general unavailable", http.StatusBadGateway) + })) + defer general.Close() + var manualCalls int + manual := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + manualCalls++ + w.Header().Set("Content-Type", "application/json") + switch { + case strings.HasPrefix(r.URL.Path, "/v1/transactions/by_hash/"): + _, _ = w.Write([]byte(`{"type":"user_transaction","success":true,"hash":"0xabc","version":"101","timestamp":"1700000000123456","payload":{"function":"0x1::primary_fungible_store::transfer","arguments":[{"inner":"` + usdc + `"},"` + receive + `","4200000"],"type":"entry_function_payload"},"events":[{"type":"0x1::fungible_asset::Withdraw","data":{"amount":"4200000","store":"` + senderStore + `"}},{"type":"0x1::fungible_asset::Deposit","data":{"amount":"4200000","store":"` + store + `"}}],"changes":[{"address":"` + senderStore + `","data":{"type":"0x1::fungible_asset::FungibleStore","data":{"metadata":{"inner":"` + usdc + `"}}},"type":"write_resource"},{"address":"` + store + `","data":{"type":"0x1::fungible_asset::FungibleStore","data":{"metadata":{"inner":"` + usdc + `"}}},"type":"write_resource"},{"address":"` + store + `","data":{"type":"0x1::object::ObjectCore","data":{"owner":"` + receive + `"}},"type":"write_resource"}]}`)) + case r.URL.Path == "/v1": + _, _ = w.Write([]byte(`{"ledger_version":"101"}`)) + default: + http.NotFound(w, r) + } + })) + defer manual.Close() + seedTestAptosRPCNode(t, general.URL, mdb.RpcNodePurposeGeneral, mdb.RpcNodeStatusOk, 100) + seedTestAptosRPCNode(t, manual.URL, mdb.RpcNodePurposeManualVerify, mdb.RpcNodeStatusOk, 1) + + order := &mdb.Orders{ + BaseModel: mdb.BaseModel{ID: 1, CreatedAt: *carbon.NewTime(carbon.CreateFromTimestampMilli(1699999999000))}, + Network: mdb.NetworkAptos, + Token: "USDC", + ActualAmount: 4.2, + ReceiveAddress: receive, + } + got, err := ValidateManualAptosPayment(order, "ABC") + if err != nil { + t.Fatalf("ValidateManualAptosPayment(): %v", err) + } + if got != "0xabc" { + t.Fatalf("canonical tx id = %q, want %q", got, "0xabc") + } + if generalCalls != 1 || manualCalls != 2 { + t.Fatalf("rpc calls general=%d manual=%d, want 1/2", generalCalls, manualCalls) + } +} diff --git a/src/model/service/move_task.go b/src/model/service/move_task.go new file mode 100644 index 0000000..dcb1650 --- /dev/null +++ b/src/model/service/move_task.go @@ -0,0 +1,232 @@ +package service + +import ( + "errors" + "fmt" + "math/big" + "strings" + "sync" + "time" + + "github.com/GMWalletApp/epusdt/model/data" + "github.com/GMWalletApp/epusdt/model/mdb" + "github.com/GMWalletApp/epusdt/model/request" + addressutil "github.com/GMWalletApp/epusdt/util/address" + "github.com/GMWalletApp/epusdt/util/constant" + "github.com/GMWalletApp/epusdt/util/log" +) + +var gProcessedMoveTx sync.Map // transfer-level key -> unix timestamp + +var processMoveOrder = OrderProcessing + +type MoveObservedTransfer struct { + Network string + ReceiveAddress string + Token string + RawAmount *big.Int + Decimals int + MinAmount float64 + Amount float64 + TxID string + BlockTimeMs int64 + Version int64 + TransferKey string +} + +func ProcessMoveObservedTransferResult(transfer MoveObservedTransfer) error { + net := moveLogLabel(transfer.Network) + cacheKey := MoveTransferCacheKey(transfer) + now := time.Now().Unix() + if _, loaded := gProcessedMoveTx.LoadOrStore(cacheKey, now); loaded { + return nil + } + clearCache := func() { + gProcessedMoveTx.Delete(cacheKey) + } + if transfer.RawAmount == nil || transfer.RawAmount.Sign() <= 0 || transfer.Amount <= 0 { + return nil + } + if transfer.MinAmount > 0 && transfer.Amount < transfer.MinAmount { + log.Sugar.Debugf("[%s-%s][%s] skip below min amount tx=%s amount=%.2f min=%.2f", net, transfer.Token, transfer.ReceiveAddress, transfer.TxID, transfer.Amount, transfer.MinAmount) + return nil + } + + tradeID, err := data.GetTradeIdByWalletAddressAndAmountAndToken(transfer.Network, transfer.ReceiveAddress, transfer.Token, transfer.Amount) + if err != nil { + clearCache() + log.Sugar.Errorf("[%s-%s][%s] lock lookup failed tx=%s: %v", net, transfer.Token, transfer.ReceiveAddress, transfer.TxID, err) + return err + } + if tradeID == "" { + raw := "" + if transfer.RawAmount != nil { + raw = transfer.RawAmount.String() + } + log.Sugar.Infof("[%s-%s][%s] skip unmatched tx=%s amount=%.8f raw=%s decimals=%d", net, transfer.Token, transfer.ReceiveAddress, transfer.TxID, transfer.Amount, raw, transfer.Decimals) + return nil + } + order, err := data.GetOrderInfoByTradeId(tradeID) + if err != nil { + clearCache() + log.Sugar.Errorf("[%s-%s][%s] load order trade_id=%s tx=%s: %v", net, transfer.Token, transfer.ReceiveAddress, tradeID, transfer.TxID, err) + return err + } + if err := EnsureMoveTransferMatchesOrder(order, transfer); err != nil { + log.Sugar.Warnf("[%s-%s][%s] skip tx=%s trade_id=%s: %v", net, transfer.Token, transfer.ReceiveAddress, transfer.TxID, tradeID, err) + return nil + } + req := &request.OrderProcessingRequest{ + ReceiveAddress: transfer.ReceiveAddress, + Token: transfer.Token, + Network: transfer.Network, + TradeId: tradeID, + Amount: transfer.Amount, + BlockTransactionId: transfer.TxID, + } + if err = processMoveOrder(req); err != nil { + if errors.Is(err, constant.OrderBlockAlreadyProcess) || errors.Is(err, constant.OrderStatusConflict) { + log.Sugar.Infof("[%s-%s][%s] already resolved trade_id=%s tx=%s reason=%v", net, transfer.Token, transfer.ReceiveAddress, tradeID, transfer.TxID, err) + return nil + } + clearCache() + log.Sugar.Errorf("[%s-%s][%s] OrderProcessing failed trade_id=%s tx=%s: %v", net, transfer.Token, transfer.ReceiveAddress, tradeID, transfer.TxID, err) + return err + } + sendPaymentNotification(order) + log.Sugar.Infof("[%s-%s][%s] payment processed trade_id=%s tx=%s", net, transfer.Token, transfer.ReceiveAddress, tradeID, transfer.TxID) + return nil +} + +func MoveTransferCacheKey(transfer MoveObservedTransfer) string { + if transfer.TransferKey != "" { + return strings.ToLower(strings.TrimSpace(transfer.TransferKey)) + } + receive := strings.ToLower(strings.TrimSpace(transfer.ReceiveAddress)) + token := strings.ToUpper(strings.TrimSpace(transfer.Token)) + raw := "" + if transfer.RawAmount != nil { + raw = transfer.RawAmount.String() + } + switch strings.ToLower(strings.TrimSpace(transfer.Network)) { + case mdb.NetworkAptos: + return fmt.Sprintf("aptos:%d:%s:%s:%s", transfer.Version, receive, token, raw) + default: + return strings.ToLower(strings.TrimSpace(transfer.Network)) + ":" + strings.TrimSpace(transfer.TxID) + } +} + +func EnsureMoveTransferMatchesOrder(order *mdb.Orders, transfer MoveObservedTransfer) error { + if order == nil || order.ID == 0 { + return fmt.Errorf("order not found") + } + if strings.ToLower(strings.TrimSpace(order.Network)) != strings.ToLower(strings.TrimSpace(transfer.Network)) { + return fmt.Errorf("network mismatch") + } + orderAddr, err := addressutil.NormalizeMoveAddress(order.ReceiveAddress) + if err != nil { + return fmt.Errorf("invalid order receive address: %w", err) + } + transferAddr, err := addressutil.NormalizeMoveAddress(transfer.ReceiveAddress) + if err != nil { + return fmt.Errorf("invalid transfer receive address: %w", err) + } + if orderAddr != transferAddr { + return fmt.Errorf("transaction recipient mismatch") + } + if !strings.EqualFold(order.Token, transfer.Token) { + return fmt.Errorf("token mismatch") + } + if transfer.BlockTimeMs <= 0 { + return fmt.Errorf("transaction block time missing") + } + if transfer.BlockTimeMs < order.CreatedAt.TimestampMilli() { + return fmt.Errorf("transaction predates the order") + } + if !amountMatchesRaw(order.ActualAmount, transfer.RawAmount, transfer.Decimals) { + return fmt.Errorf("transaction amount mismatch") + } + return nil +} + +func moveWalletWatched(address string, wallets map[string]struct{}) bool { + if address == "" || len(wallets) == 0 { + return false + } + _, ok := wallets[address] + return ok +} + +func moveAptosTransferKey(version int64, eventIndex int, receive, token string, raw *big.Int) string { + rawText := "" + if raw != nil { + rawText = raw.String() + } + return fmt.Sprintf("aptos:%d:%d:%s:%s:%s", version, eventIndex, strings.ToLower(strings.TrimSpace(receive)), strings.ToUpper(strings.TrimSpace(token)), rawText) +} + +func resolveMoveToken(network, assetID string, tokens []mdb.ChainToken) *mdb.ChainToken { + assetID = addressutil.NormalizeMoveAssetID(assetID) + for i := range tokens { + contract := strings.TrimSpace(tokens[i].ContractAddress) + if contract == "" { + continue + } + if addressutil.MoveAssetMatches(assetID, contract) { + return &tokens[i] + } + } + return nil +} + +func firstMoveAddress(values ...string) string { + for _, v := range values { + if normalized, err := addressutil.NormalizeMoveAddress(v); err == nil { + return normalized + } + } + return "" +} + +func firstNonEmpty(values ...string) string { + for _, v := range values { + if strings.TrimSpace(v) != "" { + return v + } + } + return "" +} + +func parseSignedBigInt(raw string) (*big.Int, bool) { + raw = strings.TrimSpace(raw) + if raw == "" { + return nil, false + } + out, ok := new(big.Int).SetString(raw, 10) + return out, ok +} + +func rawToFloat(raw *big.Int, decimals int) float64 { + if raw == nil { + return 0 + } + value, _ := new(big.Rat).SetFrac(new(big.Int).Set(raw), new(big.Int).Exp(big.NewInt(10), big.NewInt(int64(decimals)), nil)).Float64() + return value +} + +func recordMoveRPCResult(network string, errp *error) { + if errp != nil && *errp != nil { + data.RecordRpcFailure(network) + return + } + data.RecordRpcSuccess(network) +} + +func moveLogLabel(network string) string { + switch strings.ToLower(strings.TrimSpace(network)) { + case mdb.NetworkAptos: + return "APTOS" + default: + return strings.ToUpper(network) + } +} diff --git a/src/model/service/move_task_test.go b/src/model/service/move_task_test.go new file mode 100644 index 0000000..129392e --- /dev/null +++ b/src/model/service/move_task_test.go @@ -0,0 +1,284 @@ +package service + +import ( + "math/big" + "strings" + "testing" + "time" + + "github.com/GMWalletApp/epusdt/model/mdb" + addressutil "github.com/GMWalletApp/epusdt/util/address" + "github.com/dromara/carbon/v2" +) + +func TestParseAptosTransfersUSDTAndUSDCFungibleEvents(t *testing.T) { + receive, _ := addressutil.NormalizeMoveAddress("0xa") + receive2, _ := addressutil.NormalizeMoveAddress("0xb") + usdc := "0xbae207659db88bea0cbead6da0ed00aac12edcdda169e591cd41c94180b46f3b" + usdt := "0x357b0b74bc833e95a115ad22604854d6b0fca151cecd94111770e5d6ffc9dc2b" + usdcStore, _ := addressutil.NormalizeMoveAddress("0x12") + usdcSenderStore, _ := addressutil.NormalizeMoveAddress("0x13") + usdtStore, _ := addressutil.NormalizeMoveAddress("0x14") + usdtSenderStore, _ := addressutil.NormalizeMoveAddress("0x15") + noOwnerStore, _ := addressutil.NormalizeMoveAddress("0x16") + body := []byte(`[ + { + "type":"user_transaction", + "success":true, + "hash":"abc", + "version":"101", + "timestamp":"1700000000123456", + "events":[ + {"type":"0x1::fungible_asset::Withdraw","data":{"amount":"2500000","store":"` + usdcSenderStore + `"}}, + {"type":"0x1::fungible_asset::Deposit","data":{"amount":"2500000","store":"` + usdcStore + `"}} + ], + "changes":[ + {"address":"` + usdcSenderStore + `","data":{"type":"0x1::fungible_asset::FungibleStore","data":{"metadata":{"inner":"` + usdc + `"}}},"type":"write_resource"}, + {"address":"` + usdcStore + `","data":{"type":"0x1::fungible_asset::FungibleStore","data":{"metadata":{"inner":"` + usdc + `"}}},"type":"write_resource"}, + {"address":"` + usdcStore + `","data":{"type":"0x1::object::ObjectCore","data":{"owner":"0xa"}},"type":"write_resource"} + ] + }, + { + "type":"user_transaction", + "success":true, + "hash":"def", + "version":"102", + "timestamp":"1700000001123456", + "payload":{"function":"0x1::primary_fungible_store::transfer","arguments":[{"inner":"` + usdt + `"},"` + receive2 + `","3100000"],"type":"entry_function_payload"}, + "events":[ + {"type":"0x1::fungible_asset::Withdraw","data":{"amount":"3100000","store":"` + usdtSenderStore + `"}}, + {"type":"0x1::fungible_asset::Deposit","data":{"amount":"3100000","store":"` + usdtStore + `"}} + ], + "changes":[ + {"address":"` + usdtSenderStore + `","data":{"type":"0x1::fungible_asset::FungibleStore","data":{"metadata":{"inner":"` + usdt + `"}}},"type":"write_resource"}, + {"address":"` + usdtStore + `","data":{"type":"0x1::fungible_asset::FungibleStore","data":{"metadata":{"inner":"` + usdt + `"}}},"type":"write_resource"}, + {"address":"` + usdtStore + `","data":{"type":"0x1::object::ObjectCore","data":{"owner":"` + receive2 + `"}},"type":"write_resource"} + ] + }, + { + "type":"user_transaction", + "success":true, + "hash":"skip-withdraw-only", + "version":"103", + "timestamp":"1700000002123456", + "events":[ + {"type":"0x1::fungible_asset::Withdraw","data":{"amount":"4100000","metadata_address":"` + usdc + `"}} + ] + }, + { + "type":"user_transaction", + "success":true, + "hash":"skip-no-owner", + "version":"104", + "timestamp":"1700000003123456", + "events":[ + {"type":"0x1::fungible_asset::Withdraw","data":{"amount":"5100000","store":"` + usdcSenderStore + `"}}, + {"type":"0x1::fungible_asset::Deposit","data":{"amount":"5100000","store":"` + noOwnerStore + `"}} + ], + "changes":[ + {"address":"` + noOwnerStore + `","data":{"type":"0x1::fungible_asset::FungibleStore","data":{"metadata":{"inner":"` + usdc + `"}}},"type":"write_resource"} + ] + }, + { + "type":"user_transaction", + "success":false, + "hash":"skip-failed", + "events":[ + {"type":"0x1::fungible_asset::Deposit","data":{"to":"` + receive + `","amount":"1","metadata_address":"` + usdc + `"}} + ] + }, + { + "type":"user_transaction", + "success":true, + "hash":"skip-non-store-owner-fallback", + "version":"105", + "timestamp":"1700000004123456", + "events":[ + {"type":"0x1::fungible_asset::Withdraw","data":{"amount":"6100000","metadata_address":"` + usdc + `"}}, + {"type":"0x1::fungible_asset::Deposit","data":{"to":"` + receive + `","amount":"6100000","metadata_address":"` + usdc + `"}} + ] + } + ]`) + tokens := []mdb.ChainToken{ + {Network: mdb.NetworkAptos, Symbol: "USDC", ContractAddress: usdc, Decimals: 6, Enabled: true, MinAmount: 1}, + {Network: mdb.NetworkAptos, Symbol: "USDT", ContractAddress: usdt, Decimals: 6, Enabled: true, MinAmount: 3}, + } + + got, err := ParseAptosTransfers(body, receive, tokens) + if err != nil { + t.Fatalf("ParseAptosTransfers: %v", err) + } + if len(got) != 1 { + t.Fatalf("transfers len = %d, want 1: %#v", len(got), got) + } + if got[0].Token != "USDC" || got[0].Amount != 2.5 || got[0].TxID != "0xabc" || got[0].ReceiveAddress != receive { + t.Fatalf("usdc transfer = %#v", got[0]) + } + if got[0].TransferKey == "" { + t.Fatalf("transfer key should be non-empty: %#v", got[0]) + } + if got[0].MinAmount != 1 { + t.Fatalf("usdc min amount = %.2f, want 1", got[0].MinAmount) + } + + gotSecondWallet, err := ParseAptosTransfers(body, receive2, tokens) + if err != nil { + t.Fatalf("ParseAptosTransfers second wallet: %v", err) + } + if len(gotSecondWallet) != 1 || gotSecondWallet[0].Token != "USDT" || gotSecondWallet[0].Amount != 3.1 || gotSecondWallet[0].TxID != "0xdef" { + t.Fatalf("usdt transfer = %#v", gotSecondWallet) + } +} + +func TestParseAptosTransfersConfiguredCustomFungibleEvent(t *testing.T) { + receive, _ := addressutil.NormalizeMoveAddress("0xa") + custom := "0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + store, _ := addressutil.NormalizeMoveAddress("0x21") + senderStore, _ := addressutil.NormalizeMoveAddress("0x22") + body := []byte(`[ + { + "type":"user_transaction", + "success":true, + "hash":"custom", + "version":"201", + "timestamp":"1700000000123456", + "events":[ + {"type":"0x1::fungible_asset::Withdraw","data":{"amount":"7500000","store":"` + senderStore + `"}}, + {"type":"0x1::fungible_asset::Deposit","data":{"amount":"7500000","store":"` + store + `"}} + ], + "changes":[ + {"address":"` + senderStore + `","data":{"type":"0x1::fungible_asset::FungibleStore","data":{"metadata":{"inner":"` + custom + `"}}},"type":"write_resource"}, + {"address":"` + store + `","data":{"type":"0x1::fungible_asset::FungibleStore","data":{"metadata":{"inner":"` + custom + `"}}},"type":"write_resource"}, + {"address":"` + store + `","data":{"type":"0x1::object::ObjectCore","data":{"owner":"` + receive + `"}},"type":"write_resource"} + ] + } + ]`) + tokens := []mdb.ChainToken{ + {Network: mdb.NetworkAptos, Symbol: "MOVEUSD", ContractAddress: custom, Decimals: 6, Enabled: true}, + } + + got, err := ParseAptosTransfers(body, receive, tokens) + if err != nil { + t.Fatalf("ParseAptosTransfers: %v", err) + } + if len(got) != 1 { + t.Fatalf("transfers len = %d, want 1: %#v", len(got), got) + } + if got[0].Token != "MOVEUSD" || got[0].Amount != 7.5 || got[0].TxID != "0xcustom" { + t.Fatalf("custom transfer = %#v", got[0]) + } +} + +func TestParseAptosTransfersSplitDepositsMatchSingleWithdraw(t *testing.T) { + receive, _ := addressutil.NormalizeMoveAddress("0xa") + receive2, _ := addressutil.NormalizeMoveAddress("0xb") + usdt := "0x357b0b74bc833e95a115ad22604854d6b0fca151cecd94111770e5d6ffc9dc2b" + withdrawStore, _ := addressutil.NormalizeMoveAddress("0x21") + depositStore, _ := addressutil.NormalizeMoveAddress("0x22") + depositStore2, _ := addressutil.NormalizeMoveAddress("0x23") + body := []byte(`[ + { + "type":"user_transaction", + "success":true, + "hash":"split", + "version":"201", + "timestamp":"1700000000123456", + "events":[ + {"type":"0x1::fungible_asset::Withdraw","data":{"amount":"3000000","store":"` + withdrawStore + `"}}, + {"type":"0x1::fungible_asset::Deposit","data":{"amount":"1000000","store":"` + depositStore + `"}}, + {"type":"0x1::fungible_asset::Deposit","data":{"amount":"2000000","store":"` + depositStore2 + `"}} + ], + "changes":[ + {"address":"` + withdrawStore + `","data":{"type":"0x1::fungible_asset::FungibleStore","data":{"metadata":{"inner":"` + usdt + `"}}},"type":"write_resource"}, + {"address":"` + depositStore + `","data":{"type":"0x1::fungible_asset::FungibleStore","data":{"metadata":{"inner":"` + usdt + `"}}},"type":"write_resource"}, + {"address":"` + depositStore + `","data":{"type":"0x1::object::ObjectCore","data":{"owner":"` + receive + `"}},"type":"write_resource"}, + {"address":"` + depositStore2 + `","data":{"type":"0x1::fungible_asset::FungibleStore","data":{"metadata":{"inner":"` + usdt + `"}}},"type":"write_resource"}, + {"address":"` + depositStore2 + `","data":{"type":"0x1::object::ObjectCore","data":{"owner":"` + receive2 + `"}},"type":"write_resource"} + ] + } + ]`) + tokens := []mdb.ChainToken{ + {Network: mdb.NetworkAptos, Symbol: "USDT", ContractAddress: usdt, Decimals: 6, Enabled: true}, + } + wallets := map[string]struct{}{receive: {}, receive2: {}} + + got, err := ParseAptosTransfersForWallets(body, wallets, tokens) + if err != nil { + t.Fatalf("ParseAptosTransfersForWallets: %v", err) + } + if len(got) != 2 { + t.Fatalf("transfers len = %d, want 2: %#v", len(got), got) + } + if got[0].Token != "USDT" || got[0].Amount != 1 || got[0].ReceiveAddress != receive || got[0].TxID != "0xsplit" { + t.Fatalf("first split transfer = %#v", got[0]) + } + if got[1].Token != "USDT" || got[1].Amount != 2 || got[1].ReceiveAddress != receive2 || got[1].TxID != "0xsplit" { + t.Fatalf("second split transfer = %#v", got[1]) + } +} + +func TestParseAptosTransfersUsesStoreOwnerWhenGuidAccountIsZero(t *testing.T) { + receive, _ := addressutil.NormalizeMoveAddress("0xa") + usdt := "0x357b0b74bc833e95a115ad22604854d6b0fca151cecd94111770e5d6ffc9dc2b" + store, _ := addressutil.NormalizeMoveAddress("0x31") + senderStore, _ := addressutil.NormalizeMoveAddress("0x32") + body := []byte(`[ + { + "type":"user_transaction", + "success":true, + "hash":"decimal", + "version":"301", + "timestamp":"1700000000123456", + "events":[ + {"guid":{"creation_number":"0","account_address":"0x0"},"sequence_number":"0","type":"0x1::fungible_asset::Withdraw","data":{"amount":"10000","store":"` + senderStore + `"}}, + {"guid":{"creation_number":"0","account_address":"0x0"},"sequence_number":"0","type":"0x1::fungible_asset::Deposit","data":{"amount":"10000","store":"` + store + `"}} + ], + "changes":[ + {"address":"` + senderStore + `","data":{"type":"0x1::fungible_asset::FungibleStore","data":{"metadata":{"inner":"` + usdt + `"}}},"type":"write_resource"}, + {"address":"` + store + `","data":{"type":"0x1::fungible_asset::FungibleStore","data":{"metadata":{"inner":"` + usdt + `"}}},"type":"write_resource"}, + {"address":"` + store + `","data":{"type":"0x1::object::ObjectCore","data":{"owner":"` + receive + `"}},"type":"write_resource"} + ] + } + ]`) + tokens := []mdb.ChainToken{ + {Network: mdb.NetworkAptos, Symbol: "USDT", ContractAddress: usdt, Decimals: 6, Enabled: true}, + } + + got, err := ParseAptosTransfers(body, receive, tokens) + if err != nil { + t.Fatalf("ParseAptosTransfers: %v", err) + } + if len(got) != 1 { + t.Fatalf("transfers len = %d, want 1: %#v", len(got), got) + } + if got[0].ReceiveAddress != receive || got[0].Decimals != 6 || got[0].Amount != 0.01 || got[0].RawAmount.String() != "10000" { + t.Fatalf("transfer amount = %#v", got[0]) + } +} + +func TestEnsureMoveTransferMatchesOrder(t *testing.T) { + receive, _ := addressutil.NormalizeMoveAddress("0xc") + order := &mdb.Orders{ + BaseModel: mdb.BaseModel{ID: 1, CreatedAt: *carbon.NewTime(carbon.CreateFromTimestampMilli(time.Now().Add(-time.Minute).UnixMilli()))}, + Network: mdb.NetworkAptos, + Token: "USDC", + ActualAmount: 3.5, + ReceiveAddress: strings.ToUpper(receive), + } + transfer := MoveObservedTransfer{ + Network: mdb.NetworkAptos, + ReceiveAddress: receive, + Token: "USDC", + RawAmount: big.NewInt(3_500_000), + Decimals: 6, + BlockTimeMs: time.Now().UnixMilli(), + } + if err := EnsureMoveTransferMatchesOrder(order, transfer); err != nil { + t.Fatalf("EnsureMoveTransferMatchesOrder: %v", err) + } + + transfer.RawAmount = big.NewInt(3_400_000) + if err := EnsureMoveTransferMatchesOrder(order, transfer); err == nil || !strings.Contains(err.Error(), "amount") { + t.Fatalf("amount mismatch err = %v, want amount error", err) + } +} diff --git a/src/model/service/order_service.go b/src/model/service/order_service.go index c53979d..1101386 100644 --- a/src/model/service/order_service.go +++ b/src/model/service/order_service.go @@ -54,6 +54,11 @@ func normalizeOrderAddressByNetwork(network, address string) string { return normalized } return address + case mdb.NetworkAptos: + if normalized, err := addressutil.NormalizeMoveAddress(address); err == nil { + return normalized + } + return address default: return address } @@ -67,6 +72,9 @@ func ensureEnabledOrderAsset(network, token string) (*mdb.ChainToken, error) { if tokenRow == nil || tokenRow.ID == 0 { return nil, constant.SupportedAssetNotFound } + if !ChainTokenReadyForPayment(*tokenRow) { + return nil, constant.SupportedAssetNotFound + } return tokenRow, nil } diff --git a/src/model/service/order_service_test.go b/src/model/service/order_service_test.go index bf81d02..ab54e1a 100644 --- a/src/model/service/order_service_test.go +++ b/src/model/service/order_service_test.go @@ -469,6 +469,56 @@ func TestCreateTransactionRejectsDisabledTonTokenBeforeRate(t *testing.T) { } } +func TestCreateTransactionRejectsEnabledMisconfiguredAptosUSDT(t *testing.T) { + cleanup := testutil.SetupTestDatabases(t) + defer cleanup() + + if err := dao.Mdb.Create(&mdb.Chain{Network: mdb.NetworkAptos, Enabled: true}).Error; err != nil { + t.Fatalf("create aptos chain: %v", err) + } + if err := dao.Mdb.Create(&mdb.ChainToken{ + Network: mdb.NetworkAptos, + Symbol: "USDT", + ContractAddress: "", + Decimals: 6, + Enabled: true, + }).Error; err != nil { + t.Fatalf("create aptos usdt token: %v", err) + } + + req := newCreateTransactionRequest("order_aptos_misconfigured_usdt_1", 10) + req.Network = mdb.NetworkAptos + req.Token = "USDT" + + _, err := CreateTransaction(req, nil) + if err != constant.SupportedAssetNotFound { + t.Fatalf("create misconfigured Aptos USDT error = %v, want %v", err, constant.SupportedAssetNotFound) + } +} + +func TestEnsureEnabledOrderAssetAllowsConfiguredAptosCustomToken(t *testing.T) { + cleanup := testutil.SetupTestDatabases(t) + defer cleanup() + + if err := dao.Mdb.Create(&mdb.ChainToken{ + Network: mdb.NetworkAptos, + Symbol: "MOVEUSD", + ContractAddress: "0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + Decimals: 6, + Enabled: true, + }).Error; err != nil { + t.Fatalf("create aptos custom token: %v", err) + } + + token, err := ensureEnabledOrderAsset(mdb.NetworkAptos, "MOVEUSD") + if err != nil { + t.Fatalf("ensureEnabledOrderAsset(): %v", err) + } + if token.Symbol != "MOVEUSD" { + t.Fatalf("token symbol = %q, want MOVEUSD", token.Symbol) + } +} + func TestCreateTransactionSupportedTonTokenWithoutRateReturnsRateError(t *testing.T) { cleanup := testutil.SetupTestDatabases(t) defer cleanup() diff --git a/src/model/service/payment_asset.go b/src/model/service/payment_asset.go new file mode 100644 index 0000000..d2facc7 --- /dev/null +++ b/src/model/service/payment_asset.go @@ -0,0 +1,32 @@ +package service + +import ( + "strings" + + "github.com/GMWalletApp/epusdt/model/mdb" +) + +func ChainTokenReadyForPayment(token mdb.ChainToken) bool { + if strings.TrimSpace(token.Symbol) == "" { + return false + } + if chainTokenRequiresAssetID(token) && strings.TrimSpace(token.ContractAddress) == "" { + return false + } + return true +} + +func chainTokenRequiresAssetID(token mdb.ChainToken) bool { + network := strings.ToLower(strings.TrimSpace(token.Network)) + symbol := strings.ToUpper(strings.TrimSpace(token.Symbol)) + switch network { + case mdb.NetworkTron: + return symbol != "TRX" + case mdb.NetworkSolana: + return symbol != "SOL" + case mdb.NetworkTon: + return symbol != TonNativeSymbol + default: + return true + } +} diff --git a/src/task/listen.go b/src/task/listen.go index 173b334..452813c 100644 --- a/src/task/listen.go +++ b/src/task/listen.go @@ -15,6 +15,7 @@ func Start() { go StartPlasmaWebSocketListener() go StartTronBlockScannerListener() go StartTonBlockScannerListener() + go StartAptosLedgerScannerListener() c := cron.New() diff --git a/src/task/listen_aptos_scanner.go b/src/task/listen_aptos_scanner.go new file mode 100644 index 0000000..f2b2070 --- /dev/null +++ b/src/task/listen_aptos_scanner.go @@ -0,0 +1,255 @@ +package task + +import ( + "context" + "fmt" + "sync" + "time" + + "github.com/GMWalletApp/epusdt/model/data" + "github.com/GMWalletApp/epusdt/model/mdb" + "github.com/GMWalletApp/epusdt/model/service" + "github.com/GMWalletApp/epusdt/util/log" +) + +const ( + aptosLedgerChunkSize int64 = 100 + aptosLedgerWorkerCount = 3 + aptosLedgerQueueLimit = 100 +) + +type aptosRuntimeCursor struct { + initialized bool + lastSeenVersion int64 +} + +type aptosLedgerRange struct { + start int64 + limit int64 + state moveWatchState +} + +type aptosChainProvider interface { + LatestLedgerVersion() (int64, error) + Transactions(start int64, limit int64) ([]byte, error) +} + +type serviceAptosProvider struct{} + +func (serviceAptosProvider) LatestLedgerVersion() (int64, error) { + return service.AptosGetLedgerVersion() +} + +func (serviceAptosProvider) Transactions(start int64, limit int64) ([]byte, error) { + return service.AptosGetTransactions(start, limit) +} + +var ( + parseAptosTransfersForWallets = service.ParseAptosTransfersForWallets + processAptosObservedTransferFunc = service.ProcessMoveObservedTransferResult +) + +func StartAptosLedgerScannerListener() { + provider := serviceAptosProvider{} + cursor := &aptosRuntimeCursor{} + log.Sugar.Infof("[APTOS] ledger scanner starting mode=full_ledger chunk_size=%d workers=%d queue_limit=%d", + aptosLedgerChunkSize, + aptosLedgerWorkerCount, + aptosLedgerQueueLimit, + ) + for { + if err := runAptosLedgerScanner(context.Background(), provider, cursor); err != nil { + log.Sugar.Warnf("[APTOS] scanner stopped: %v", err) + sleepOrDone(context.Background(), moveScannerRetryDelay) + continue + } + } +} + +func runAptosLedgerScanner(ctx context.Context, provider aptosChainProvider, cursor *aptosRuntimeCursor) error { + if cursor == nil { + cursor = &aptosRuntimeCursor{} + } + watchSignature := "" + for { + chain, interval := moveChainConfig(mdb.NetworkAptos) + if chain == nil || !chain.Enabled { + log.Sugar.Debug("[APTOS] chain disabled or not configured, idling") + if !sleepOrDone(ctx, interval) { + return nil + } + continue + } + state, err := loadMoveWatchState(mdb.NetworkAptos) + if err != nil { + return err + } + if len(state.wallets) == 0 || len(state.tokens) == 0 { + log.Sugar.Debug("[APTOS] no enabled wallets or tokens, idling") + if !sleepOrDone(ctx, interval) { + return nil + } + continue + } + if sig := moveWatchStateSignature(state); sig != watchSignature { + watchSignature = sig + log.Sugar.Infof("[APTOS] ledger scanner watching wallets=%d tokens=%d", len(state.wallets), len(state.tokens)) + } + + latest, err := provider.LatestLedgerVersion() + if err != nil { + return fmt.Errorf("get latest ledger version: %w", err) + } + data.RecordRpcBlockHeight(mdb.NetworkAptos, latest) + confirmed := moveConfirmedCursor(latest, chain.MinConfirmations) + if !cursor.initialized { + cursor.initialized = true + cursor.lastSeenVersion = confirmed - aptosLedgerChunkSize + if cursor.lastSeenVersion < -1 { + cursor.lastSeenVersion = -1 + } + log.Sugar.Infof( + "[APTOS] initialized ledger cursor at version=%d confirmed_head=%d chunk_size=%d mode=full_ledger_watch", + cursor.lastSeenVersion, + confirmed, + aptosLedgerChunkSize, + ) + } + + catchup, err := processAptosLedgerRound(ctx, provider, state, cursor, confirmed) + if err != nil { + return err + } + if !sleepOrDone(ctx, chooseAptosIdleDelay(catchup, interval)) { + return nil + } + } +} + +func processAptosLedgerRound(ctx context.Context, provider aptosChainProvider, state moveWatchState, cursor *aptosRuntimeCursor, confirmedVersion int64) (bool, error) { + ranges := buildAptosLedgerRanges(cursor.lastSeenVersion, confirmedVersion, state) + if len(ranges) == 0 { + return false, nil + } + lastRange := ranges[len(ranges)-1] + log.Sugar.Infof( + "[APTOS] queued ledger ranges count=%d from=%d to=%d cursor=%d confirmed_head=%d workers=%d", + len(ranges), + ranges[0].start, + lastRange.start+lastRange.limit-1, + cursor.lastSeenVersion, + confirmedVersion, + aptosLedgerWorkerCount, + ) + + type rangeResult struct { + item aptosLedgerRange + err error + } + jobs := make(chan aptosLedgerRange, len(ranges)) + results := make(chan rangeResult, len(ranges)) + workers := aptosLedgerWorkerCount + if len(ranges) < workers { + workers = len(ranges) + } + + var wg sync.WaitGroup + for i := 0; i < workers; i++ { + wg.Add(1) + go func() { + defer wg.Done() + for item := range jobs { + results <- rangeResult{item: item, err: processAptosLedgerRange(ctx, provider, item, confirmedVersion)} + } + }() + } + for _, item := range ranges { + jobs <- item + } + close(jobs) + wg.Wait() + close(results) + + errByStart := make(map[int64]error, len(ranges)) + for result := range results { + if result.err != nil { + errByStart[result.item.start] = result.err + } + } + + for _, item := range ranges { + if err := errByStart[item.start]; err != nil { + return cursor.lastSeenVersion < confirmedVersion, err + } + cursor.lastSeenVersion = item.start + item.limit - 1 + } + return cursor.lastSeenVersion < confirmedVersion, nil +} + +func buildAptosLedgerRanges(lastSeenVersion int64, confirmedVersion int64, state moveWatchState) []aptosLedgerRange { + if confirmedVersion <= lastSeenVersion { + return nil + } + ranges := make([]aptosLedgerRange, 0, aptosLedgerQueueLimit) + next := lastSeenVersion + 1 + if next < 0 { + next = 0 + } + for next <= confirmedVersion && len(ranges) < aptosLedgerQueueLimit { + limit := confirmedVersion - next + 1 + if limit > aptosLedgerChunkSize { + limit = aptosLedgerChunkSize + } + ranges = append(ranges, aptosLedgerRange{start: next, limit: limit, state: state}) + next += limit + } + return ranges +} + +func processAptosLedgerRange(ctx context.Context, provider aptosChainProvider, item aptosLedgerRange, confirmedVersion int64) error { + if err := ctx.Err(); err != nil { + return err + } + body, err := provider.Transactions(item.start, item.limit) + if err != nil { + return fmt.Errorf("get transactions start=%d limit=%d: %w", item.start, item.limit, err) + } + transfers, err := parseAptosTransfersForWallets(body, item.state.wallets, item.state.tokens) + if err != nil { + return fmt.Errorf("parse transactions start=%d limit=%d: %w", item.start, item.limit, err) + } + for _, transfer := range transfers { + log.Sugar.Infof( + "[APTOS] observed transfer version=%d tx=%s token=%s amount=%.8f raw=%s decimals=%d receive=%s key=%s", + transfer.Version, + transfer.TxID, + transfer.Token, + transfer.Amount, + transfer.RawAmount.String(), + transfer.Decimals, + transfer.ReceiveAddress, + transfer.TransferKey, + ) + if err = processAptosObservedTransferFunc(transfer); err != nil { + return err + } + } + if len(transfers) > 0 { + log.Sugar.Infof( + "[APTOS] ledger range processed start=%d limit=%d to=%d transfers=%d confirmed_head=%d", + item.start, + item.limit, + item.start+item.limit-1, + len(transfers), + confirmedVersion, + ) + } + return nil +} + +func chooseAptosIdleDelay(catchup bool, interval time.Duration) time.Duration { + if catchup { + return moveCatchupYield + } + return interval +} diff --git a/src/task/listen_aptos_scanner_test.go b/src/task/listen_aptos_scanner_test.go new file mode 100644 index 0000000..c50a8bc --- /dev/null +++ b/src/task/listen_aptos_scanner_test.go @@ -0,0 +1,440 @@ +package task + +import ( + "context" + "errors" + "fmt" + "sort" + "sync" + "testing" + "time" + + "github.com/GMWalletApp/epusdt/internal/testutil" + "github.com/GMWalletApp/epusdt/model/dao" + "github.com/GMWalletApp/epusdt/model/data" + "github.com/GMWalletApp/epusdt/model/mdb" + "github.com/GMWalletApp/epusdt/model/service" + addressutil "github.com/GMWalletApp/epusdt/util/address" +) + +type fakeAptosProvider struct { + mu sync.Mutex + + latest int64 + + bodiesByStart map[int64][]byte + errByStart map[int64]error + + calls []aptosRangeCall +} + +type aptosRangeCall struct { + start int64 + limit int64 +} + +func (f *fakeAptosProvider) LatestLedgerVersion() (int64, error) { + return f.latest, nil +} + +func (f *fakeAptosProvider) Transactions(start int64, limit int64) ([]byte, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.calls = append(f.calls, aptosRangeCall{start: start, limit: limit}) + if err := f.errByStart[start]; err != nil { + return nil, err + } + if body := f.bodiesByStart[start]; len(body) > 0 { + return append([]byte(nil), body...), nil + } + return []byte("[]"), nil +} + +func (f *fakeAptosProvider) rangeCalls() []aptosRangeCall { + f.mu.Lock() + defer f.mu.Unlock() + calls := append([]aptosRangeCall(nil), f.calls...) + sort.Slice(calls, func(i, j int) bool { + if calls[i].start == calls[j].start { + return calls[i].limit < calls[j].limit + } + return calls[i].start < calls[j].start + }) + return calls +} + +func resetAptosScannerHooks(t *testing.T) { + t.Helper() + oldParse := parseAptosTransfersForWallets + oldProcess := processAptosObservedTransferFunc + t.Cleanup(func() { + parseAptosTransfersForWallets = oldParse + processAptosObservedTransferFunc = oldProcess + }) +} + +func seedAptosScannerChain(t *testing.T, wallets ...string) []string { + t.Helper() + if err := dao.Mdb.Create(&mdb.Chain{Network: mdb.NetworkAptos, Enabled: true, MinConfirmations: 1, ScanIntervalSec: 1}).Error; err != nil { + t.Fatalf("seed Aptos chain: %v", err) + } + if err := dao.Mdb.Create(&[]mdb.ChainToken{ + {Network: mdb.NetworkAptos, Symbol: "USDC", ContractAddress: "0xbae207659db88bea0cbead6da0ed00aac12edcdda169e591cd41c94180b46f3b", Decimals: 6, Enabled: true}, + {Network: mdb.NetworkAptos, Symbol: "USDT", ContractAddress: "0x357b0b74bc833e95a115ad22604854d6b0fca151cecd94111770e5d6ffc9dc2b", Decimals: 6, Enabled: true}, + {Network: mdb.NetworkAptos, Symbol: "MOVEUSD", ContractAddress: "0xaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", Decimals: 6, Enabled: true}, + }).Error; err != nil { + t.Fatalf("seed Aptos tokens: %v", err) + } + if len(wallets) == 0 { + wallets = []string{"0xa"} + } + out := make([]string, 0, len(wallets)) + for _, wallet := range wallets { + receive, err := addressutil.NormalizeMoveAddress(wallet) + if err != nil { + t.Fatalf("normalize Aptos address %q: %v", wallet, err) + } + if _, err = data.AddWalletAddressWithNetwork(mdb.NetworkAptos, receive); err != nil { + t.Fatalf("seed Aptos wallet %s: %v", receive, err) + } + out = append(out, receive) + } + return out +} + +func TestProcessAptosLedgerRangePassesConfiguredTokens(t *testing.T) { + cleanup := testutil.SetupTestDatabases(t) + defer cleanup() + resetAptosScannerHooks(t) + seedAptosScannerChain(t, "0xa") + + state, err := loadMoveWatchState(mdb.NetworkAptos) + if err != nil { + t.Fatalf("load state: %v", err) + } + var gotSymbols []string + parseAptosTransfersForWallets = func(_ []byte, _ map[string]struct{}, tokens []mdb.ChainToken) ([]service.MoveObservedTransfer, error) { + for _, token := range tokens { + gotSymbols = append(gotSymbols, token.Symbol) + } + sort.Strings(gotSymbols) + return nil, nil + } + provider := &fakeAptosProvider{bodiesByStart: map[int64][]byte{100: []byte("[]")}} + + err = processAptosLedgerRange(context.Background(), provider, aptosLedgerRange{start: 100, limit: 1, state: state}, 100) + if err != nil { + t.Fatalf("processAptosLedgerRange(): %v", err) + } + want := []string{"MOVEUSD", "USDC", "USDT"} + if len(gotSymbols) != len(want) { + t.Fatalf("tokens = %#v, want %#v", gotSymbols, want) + } + for i := range want { + if gotSymbols[i] != want[i] { + t.Fatalf("tokens = %#v, want %#v", gotSymbols, want) + } + } +} + +func TestRunAptosLedgerScannerIdlesWhenChainDisabled(t *testing.T) { + cleanup := testutil.SetupTestDatabases(t) + defer cleanup() + resetAptosScannerHooks(t) + if err := dao.Mdb.Create(&mdb.Chain{Network: mdb.NetworkAptos, Enabled: false, MinConfirmations: 1, ScanIntervalSec: 1}).Error; err != nil { + t.Fatalf("seed disabled Aptos chain: %v", err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + + provider := &fakeAptosProvider{latest: 999} + cursor := &aptosRuntimeCursor{} + if err := runAptosLedgerScanner(ctx, provider, cursor); err != nil { + t.Fatalf("runAptosLedgerScanner(): %v", err) + } + if cursor.initialized { + t.Fatalf("cursor initialized while chain disabled: %#v", cursor) + } + if calls := provider.rangeCalls(); len(calls) != 0 { + t.Fatalf("range calls = %#v, want none", calls) + } +} + +func TestRunAptosLedgerScannerInitializesAtTrailingConfirmedChunk(t *testing.T) { + cleanup := testutil.SetupTestDatabases(t) + defer cleanup() + resetAptosScannerHooks(t) + seedAptosScannerChain(t, "0xa") + + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + + provider := &fakeAptosProvider{latest: 999} + cursor := &aptosRuntimeCursor{} + if err := runAptosLedgerScanner(ctx, provider, cursor); err != nil { + t.Fatalf("runAptosLedgerScanner(): %v", err) + } + + if !cursor.initialized || cursor.lastSeenVersion != 999 { + t.Fatalf("cursor = %#v, want initialized and advanced to confirmed head 999", cursor) + } + calls := provider.rangeCalls() + if len(calls) != 1 || calls[0].start != 900 || calls[0].limit != 100 { + t.Fatalf("range calls = %#v, want start=900 limit=100", calls) + } +} + +func TestProcessAptosLedgerRoundDoesNotAdvanceUnconfirmedVersion(t *testing.T) { + cleanup := testutil.SetupTestDatabases(t) + defer cleanup() + resetAptosScannerHooks(t) + seedAptosScannerChain(t, "0xa") + + provider := &fakeAptosProvider{latest: 100} + cursor := &aptosRuntimeCursor{initialized: true, lastSeenVersion: 99} + state, err := loadMoveWatchState(mdb.NetworkAptos) + if err != nil { + t.Fatalf("load state: %v", err) + } + + catchup, err := processAptosLedgerRound(context.Background(), provider, state, cursor, 99) + if err != nil { + t.Fatalf("processAptosLedgerRound(): %v", err) + } + if catchup { + t.Fatal("catchup = true, want false") + } + if cursor.lastSeenVersion != 99 { + t.Fatalf("lastSeenVersion = %d, want 99", cursor.lastSeenVersion) + } + if calls := provider.rangeCalls(); len(calls) != 0 { + t.Fatalf("range calls = %#v, want none", calls) + } +} + +func TestProcessAptosLedgerRoundAdvancesAfterSuccessfulRange(t *testing.T) { + cleanup := testutil.SetupTestDatabases(t) + defer cleanup() + resetAptosScannerHooks(t) + seedAptosScannerChain(t, "0xa") + + state, err := loadMoveWatchState(mdb.NetworkAptos) + if err != nil { + t.Fatalf("load state: %v", err) + } + provider := &fakeAptosProvider{bodiesByStart: map[int64][]byte{100: []byte("[]")}} + cursor := &aptosRuntimeCursor{initialized: true, lastSeenVersion: 99} + + catchup, err := processAptosLedgerRound(context.Background(), provider, state, cursor, 250) + if err != nil { + t.Fatalf("processAptosLedgerRound(): %v", err) + } + if catchup { + t.Fatal("catchup = true, want false") + } + if cursor.lastSeenVersion != 250 { + t.Fatalf("lastSeenVersion = %d, want 250", cursor.lastSeenVersion) + } + calls := provider.rangeCalls() + if len(calls) != 2 || calls[0].start != 100 || calls[0].limit != 100 || calls[1].start != 200 || calls[1].limit != 51 { + t.Fatalf("range calls = %#v, want [100/100 200/51]", calls) + } +} + +func TestProcessAptosLedgerRoundDoesNotAdvanceOnRangeFailure(t *testing.T) { + cleanup := testutil.SetupTestDatabases(t) + defer cleanup() + resetAptosScannerHooks(t) + seedAptosScannerChain(t, "0xa") + + state, err := loadMoveWatchState(mdb.NetworkAptos) + if err != nil { + t.Fatalf("load state: %v", err) + } + provider := &fakeAptosProvider{errByStart: map[int64]error{100: errors.New("temporary rpc failure")}} + cursor := &aptosRuntimeCursor{initialized: true, lastSeenVersion: 99} + + if _, err = processAptosLedgerRound(context.Background(), provider, state, cursor, 150); err == nil { + t.Fatal("processAptosLedgerRound() error = nil, want error") + } + if cursor.lastSeenVersion != 99 { + t.Fatalf("lastSeenVersion = %d, want 99", cursor.lastSeenVersion) + } +} + +func TestProcessAptosLedgerRoundProcessesMultipleChunks(t *testing.T) { + cleanup := testutil.SetupTestDatabases(t) + defer cleanup() + resetAptosScannerHooks(t) + seedAptosScannerChain(t, "0xa") + + state, err := loadMoveWatchState(mdb.NetworkAptos) + if err != nil { + t.Fatalf("load state: %v", err) + } + provider := &fakeAptosProvider{ + bodiesByStart: map[int64][]byte{ + 100: []byte("[]"), + 200: []byte("[]"), + }, + } + cursor := &aptosRuntimeCursor{initialized: true, lastSeenVersion: 99} + + catchup, err := processAptosLedgerRound(context.Background(), provider, state, cursor, 250) + if err != nil { + t.Fatalf("processAptosLedgerRound(): %v", err) + } + if catchup { + t.Fatal("catchup = true, want false") + } + if cursor.lastSeenVersion != 250 { + t.Fatalf("lastSeenVersion = %d, want 250", cursor.lastSeenVersion) + } + calls := provider.rangeCalls() + if len(calls) != 2 || calls[0].start != 100 || calls[0].limit != 100 || calls[1].start != 200 || calls[1].limit != 51 { + t.Fatalf("range calls = %#v, want [100/100 200/51]", calls) + } +} + +func TestBuildAptosLedgerRangesStopsAtQueueLimit(t *testing.T) { + state := moveWatchState{} + ranges := buildAptosLedgerRanges(0, aptosLedgerChunkSize*aptosLedgerQueueLimit+500, state) + if len(ranges) != aptosLedgerQueueLimit { + t.Fatalf("ranges len = %d, want %d", len(ranges), aptosLedgerQueueLimit) + } + first := ranges[0] + last := ranges[len(ranges)-1] + if first.start != 1 || first.limit != aptosLedgerChunkSize { + t.Fatalf("first range = %#v, want start=1 limit=%d", first, aptosLedgerChunkSize) + } + if last.start != 1+aptosLedgerChunkSize*int64(aptosLedgerQueueLimit-1) || last.limit != aptosLedgerChunkSize { + t.Fatalf("last range = %#v", last) + } +} + +func TestProcessAptosLedgerRoundMarksMatchingUSDTOrderPaid(t *testing.T) { + cleanup := testutil.SetupTestDatabases(t) + defer cleanup() + resetAptosScannerHooks(t) + wallets := seedAptosScannerChain(t, "0xa") + receive := wallets[0] + amount := 3.1 + tradeID := "aptos_trade_1" + usdt := "0x357b0b74bc833e95a115ad22604854d6b0fca151cecd94111770e5d6ffc9dc2b" + body := aptosFungibleTransferBody(t, "0xabc", 101, receive, usdt, "3100000") + + order := &mdb.Orders{ + TradeId: tradeID, + OrderId: "aptos_order_1", + Amount: amount, + ActualAmount: amount, + ReceiveAddress: receive, + Token: "USDT", + Network: mdb.NetworkAptos, + Status: mdb.StatusWaitPay, + CallBackConfirm: mdb.CallBackConfirmOk, + PaymentType: mdb.PaymentTypeGmpay, + PayProvider: mdb.PaymentProviderOnChain, + } + if err := dao.Mdb.Create(order).Error; err != nil { + t.Fatalf("create order: %v", err) + } + if err := data.LockTransaction(mdb.NetworkAptos, receive, "USDT", tradeID, amount, time.Hour); err != nil { + t.Fatalf("lock transaction: %v", err) + } + + state, err := loadMoveWatchState(mdb.NetworkAptos) + if err != nil { + t.Fatalf("load state: %v", err) + } + provider := &fakeAptosProvider{bodiesByStart: map[int64][]byte{101: body}} + cursor := &aptosRuntimeCursor{initialized: true, lastSeenVersion: 100} + + if _, err = processAptosLedgerRound(context.Background(), provider, state, cursor, 101); err != nil { + t.Fatalf("processAptosLedgerRound(): %v", err) + } + paid, err := data.GetOrderInfoByTradeId(tradeID) + if err != nil { + t.Fatalf("reload order: %v", err) + } + if paid.Status != mdb.StatusPaySuccess || paid.CallBackConfirm != mdb.CallBackConfirmNo || paid.BlockTransactionId != "0xabc" { + t.Fatalf("paid order = %#v", paid) + } + lockTradeID, err := data.GetTradeIdByWalletAddressAndAmountAndToken(mdb.NetworkAptos, receive, "USDT", amount) + if err != nil { + t.Fatalf("lookup lock: %v", err) + } + if lockTradeID != "" { + t.Fatalf("transaction lock still exists: %s", lockTradeID) + } +} + +func TestProcessAptosLedgerRoundDoesNotReprocessDuplicateTransferKey(t *testing.T) { + cleanup := testutil.SetupTestDatabases(t) + defer cleanup() + resetAptosScannerHooks(t) + wallets := seedAptosScannerChain(t, "0xa") + receive := wallets[0] + usdc := "0xbae207659db88bea0cbead6da0ed00aac12edcdda169e591cd41c94180b46f3b" + + state, err := loadMoveWatchState(mdb.NetworkAptos) + if err != nil { + t.Fatalf("load state: %v", err) + } + provider := &fakeAptosProvider{ + bodiesByStart: map[int64][]byte{ + 101: aptosFungibleTransferBody(t, "0xabc", 101, receive, usdc, "1200000"), + 102: aptosFungibleTransferBody(t, "0xabc", 101, receive, usdc, "1200000"), + }, + } + var processed int + processAptosObservedTransferFunc = func(transfer service.MoveObservedTransfer) error { + processed++ + return service.ProcessMoveObservedTransferResult(transfer) + } + cursor := &aptosRuntimeCursor{initialized: true, lastSeenVersion: 100} + + if _, err = processAptosLedgerRound(context.Background(), provider, state, cursor, 101); err != nil { + t.Fatalf("first processAptosLedgerRound(): %v", err) + } + if _, err = processAptosLedgerRound(context.Background(), provider, state, cursor, 102); err != nil { + t.Fatalf("second processAptosLedgerRound(): %v", err) + } + if processed != 2 { + t.Fatalf("scanner processed callback count = %d, want 2", processed) + } + if cursor.lastSeenVersion != 102 { + t.Fatalf("lastSeenVersion = %d, want 102", cursor.lastSeenVersion) + } +} + +func aptosFungibleTransferBody(t *testing.T, hash string, version int64, receive string, metadata string, rawAmount string) []byte { + t.Helper() + store, err := addressutil.NormalizeMoveAddress(fmt.Sprintf("0x%x", version+1000)) + if err != nil { + t.Fatalf("normalize store: %v", err) + } + senderStore, err := addressutil.NormalizeMoveAddress(fmt.Sprintf("0x%x", version+2000)) + if err != nil { + t.Fatalf("normalize sender store: %v", err) + } + return []byte(`[ + { + "type":"user_transaction", + "success":true, + "hash":"` + hash + `", + "version":"` + fmt.Sprintf("%d", version) + `", + "timestamp":"` + fmt.Sprintf("%d", time.Now().UnixMicro()) + `", + "events":[ + {"type":"0x1::fungible_asset::Withdraw","data":{"amount":"` + rawAmount + `","store":"` + senderStore + `"}}, + {"type":"0x1::fungible_asset::Deposit","data":{"amount":"` + rawAmount + `","store":"` + store + `"}} + ], + "changes":[ + {"address":"` + senderStore + `","data":{"type":"0x1::fungible_asset::FungibleStore","data":{"metadata":{"inner":"` + metadata + `"}}},"type":"write_resource"}, + {"address":"` + store + `","data":{"type":"0x1::fungible_asset::FungibleStore","data":{"metadata":{"inner":"` + metadata + `"}}},"type":"write_resource"}, + {"address":"` + store + `","data":{"type":"0x1::object::ObjectCore","data":{"owner":"` + receive + `"}},"type":"write_resource"} + ] + } + ]`) +} diff --git a/src/task/listen_move_common.go b/src/task/listen_move_common.go new file mode 100644 index 0000000..b9e228a --- /dev/null +++ b/src/task/listen_move_common.go @@ -0,0 +1,77 @@ +package task + +import ( + "fmt" + "strings" + "time" + + "github.com/GMWalletApp/epusdt/model/data" + "github.com/GMWalletApp/epusdt/model/mdb" + addressutil "github.com/GMWalletApp/epusdt/util/address" + "github.com/GMWalletApp/epusdt/util/log" +) + +const ( + moveScannerRetryDelay = 3 * time.Second + moveCatchupYield = 250 * time.Millisecond +) + +type moveWatchState struct { + wallets map[string]struct{} + tokens []mdb.ChainToken +} + +func moveChainConfig(network string) (*mdb.Chain, time.Duration) { + row, err := data.GetChainByNetwork(network) + if err != nil || row == nil || row.ID == 0 { + if err != nil { + log.Sugar.Warnf("[%s] load chain config failed: %v", strings.ToUpper(network), err) + } + return nil, 5 * time.Second + } + interval := time.Duration(row.ScanIntervalSec) * time.Second + if interval <= 0 { + interval = 5 * time.Second + } + return row, interval +} + +func loadMoveWatchState(network string) (moveWatchState, error) { + rows, err := data.GetAvailableWalletAddressByNetwork(network) + if err != nil { + return moveWatchState{}, err + } + state := moveWatchState{wallets: make(map[string]struct{})} + for _, row := range rows { + address, err := addressutil.NormalizeMoveAddress(row.Address) + if err != nil { + log.Sugar.Warnf("[%s] skip invalid wallet address=%s err=%v", strings.ToUpper(network), row.Address, err) + continue + } + state.wallets[address] = struct{}{} + } + tokens, err := data.ListEnabledChainTokensByNetwork(network) + if err != nil { + return moveWatchState{}, err + } + state.tokens = tokens + return state, nil +} + +func moveWatchStateSignature(state moveWatchState) string { + return fmt.Sprintf("wallets=%d/tokens=%d", len(state.wallets), len(state.tokens)) +} + +func moveConfirmedCursor(latest int64, minConfirmations int) int64 { + if latest <= 0 { + return 0 + } + if minConfirmations <= 1 { + return latest + } + confirmed := latest - int64(minConfirmations) + 1 + if confirmed < 0 { + return 0 + } + return confirmed +} diff --git a/src/util/address/move.go b/src/util/address/move.go new file mode 100644 index 0000000..ea25978 --- /dev/null +++ b/src/util/address/move.go @@ -0,0 +1,67 @@ +package addressutil + +import ( + "encoding/hex" + "fmt" + "strings" +) + +// NormalizeMoveAddress returns the canonical 0x-prefixed, 64-hex-character +// account address used by Aptos. +func NormalizeMoveAddress(input string) (string, error) { + raw := strings.ToLower(strings.TrimSpace(input)) + raw = strings.TrimPrefix(raw, "0x") + if raw == "" { + return "", fmt.Errorf("move address is empty") + } + if len(raw) > 64 { + return "", fmt.Errorf("move address is too long") + } + if _, err := hex.DecodeString(padEvenHex(raw)); err != nil { + return "", fmt.Errorf("invalid move address") + } + return "0x" + strings.Repeat("0", 64-len(raw)) + raw, nil +} + +func padEvenHex(raw string) string { + if len(raw)%2 == 0 { + return raw + } + return "0" + raw +} + +// NormalizeMoveAssetID canonicalizes Move asset identifiers used in +// chain_tokens.contract_address for Aptos. A plain address is normalized to +// 64 hex chars; structured types keep module/name suffixes after normalizing +// the leading address. +func NormalizeMoveAssetID(input string) string { + raw := strings.ToLower(strings.TrimSpace(input)) + if raw == "" { + return "" + } + parts := strings.Split(raw, "::") + if len(parts) == 1 { + if addr, err := NormalizeMoveAddress(raw); err == nil { + return addr + } + return raw + } + if addr, err := NormalizeMoveAddress(parts[0]); err == nil { + parts[0] = addr + } + return strings.Join(parts, "::") +} + +func MoveAssetMatches(observed, configured string) bool { + observed = NormalizeMoveAssetID(observed) + configured = NormalizeMoveAssetID(configured) + if configured == "" { + return observed == "" + } + if observed == configured { + return true + } + // Aptos FA event payloads may expose only the metadata address while other + // payloads expose a type string containing the same address. + return strings.Contains(observed, configured) +} diff --git a/src/util/address/move_test.go b/src/util/address/move_test.go new file mode 100644 index 0000000..e09e6b4 --- /dev/null +++ b/src/util/address/move_test.go @@ -0,0 +1,42 @@ +package addressutil + +import ( + "strings" + "testing" +) + +func TestNormalizeMoveAddress(t *testing.T) { + got, err := NormalizeMoveAddress(" A ") + if err != nil { + t.Fatalf("NormalizeMoveAddress: %v", err) + } + want := "0x000000000000000000000000000000000000000000000000000000000000000a" + if got != want { + t.Fatalf("address = %q, want %q", got, want) + } + + got, err = NormalizeMoveAddress("0X" + "ABCDEF") + if err != nil { + t.Fatalf("NormalizeMoveAddress uppercase: %v", err) + } + want = "0x0000000000000000000000000000000000000000000000000000000000abcdef" + if got != want { + t.Fatalf("uppercase address = %q, want %q", got, want) + } +} + +func TestNormalizeMoveAddressRejectsInvalid(t *testing.T) { + for _, input := range []string{"", "0xzz", "0x" + strings.Repeat("1", 65)} { + if _, err := NormalizeMoveAddress(input); err == nil { + t.Fatalf("NormalizeMoveAddress(%q) succeeded, want error", input) + } + } +} + +func TestNormalizeMoveAssetIDNormalizesLeadingAddress(t *testing.T) { + got := NormalizeMoveAssetID("0X1::FUNGIBLE_ASSET::METADATA") + want := "0x0000000000000000000000000000000000000000000000000000000000000001::fungible_asset::metadata" + if got != want { + t.Fatalf("asset id = %q, want %q", got, want) + } +}