feat: add Aptos USDT/USDC payment support

This commit is contained in:
line-6000
2026-06-20 11:08:19 +08:00
parent 79ca1730cd
commit de96a5d3e7
32 changed files with 2331 additions and 16 deletions
+1 -1
View File
@@ -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
+51 -5
View File
@@ -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)
}
@@ -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)
}
}
+2
View File
@@ -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:<receive_raw>:<lt>:<hash>, 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
+2
View File
@@ -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:<receive_raw>:<lt>:<hash>, 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
@@ -44,6 +44,13 @@ func buildSupportedAssets() ([]response.NetworkTokenSupport, error) {
}
symbols := make([]string, 0, len(tokens))
for _, t := range tokens {
contractAddress := strings.TrimSpace(t.ContractAddress)
if network == mdb.NetworkAptos && !aptosPublicPaymentToken(t.Symbol) {
continue
}
if network == mdb.NetworkAptos && contractAddress == "" {
continue
}
sym := strings.ToUpper(strings.TrimSpace(t.Symbol))
if sym == "" {
continue
@@ -68,6 +75,15 @@ func buildSupportedAssets() ([]response.NetworkTokenSupport, error) {
return supports, nil
}
func aptosPublicPaymentToken(symbol string) bool {
switch strings.ToUpper(strings.TrimSpace(symbol)) {
case "USDT", "USDC":
return true
default:
return false
}
}
// GetPublicConfig returns payment/site config consumed by cashier/frontends.
// The public route returns brand/site display config plus non-sensitive
// epay/okpay knobs, while the authenticated admin route also includes OkPay
@@ -0,0 +1,44 @@
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 TestBuildSupportedAssetsSkipsMisconfiguredAptosTokens(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: "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)
}
if len(supports[0].Tokens) != 1 || supports[0].Tokens[0] != "USDT" {
t.Fatalf("tokens = %#v, want [USDT]", supports[0].Tokens)
}
}
+4
View File
@@ -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},
+91
View File
@@ -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
}
+5
View File
@@ -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
}
+16
View File
@@ -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)
}
}
+3
View File
@@ -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
}
@@ -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)
}
}
+5 -4
View File
@@ -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"`
+1
View File
@@ -13,6 +13,7 @@ const (
NetworkPolygon = "polygon"
NetworkPlasma = "plasma"
NetworkTon = "ton"
NetworkAptos = "aptos"
)
const (
+2 -2
View File
@@ -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 {
+456
View File
@@ -0,0 +1,456 @@
package service
import (
"fmt"
"io"
"math/big"
"net/http"
"strconv"
"strings"
"sync"
"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"
)
const defaultAptosFullnodeURL = "https://aptos-rest.publicnode.com/"
var aptosFullnodeURL = defaultAptosFullnodeURL
var aptosFixedNodeLogOnce sync.Once
func AptosFixedFullnodeURL() string {
return strings.TrimSpace(aptosFullnodeURL)
}
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 rpc=%s", txID, aptosFullnodeURL)
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 rpc=%s", start, limit, aptosFullnodeURL)
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 {
token := resolveMoveToken(mdb.NetworkAptos, assetID, tokens)
if token == nil {
return nil
}
switch NormalizeAptosPaymentSymbol(token.Symbol) {
case "USDT", "USDC":
return token
default:
return nil
}
}
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)
body, err := AptosGetTransactionByHash(txID)
if err != nil {
return "", fmt.Errorf("fetch aptos transaction: %w", err)
}
if err = validateManualAptosTransaction(order, txID, body); err != nil {
return "", err
}
return txID, nil
}
func validateManualAptosTransaction(order *mdb.Orders, txID string, body []byte) 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 = EnsureAptosTransferConfirmed(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)
logAptosFixedNode()
endpoint, err := aptosJoinURL(aptosFullnodeURL, 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 logAptosFixedNode() {
aptosFixedNodeLogOnce.Do(func() {
log.Sugar.Infof("[APTOS] using fixed public fullnode url=%s", AptosFixedFullnodeURL())
})
}
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 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
}
+4 -2
View File
@@ -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)
@@ -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()
overrideAptosFullnodeURLForTest(t, server.URL)
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)
+137
View File
@@ -0,0 +1,137 @@
package service
import (
"net/http"
"net/http/httptest"
"strings"
"sync"
"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 overrideAptosFullnodeURLForTest(t *testing.T, url string) {
t.Helper()
oldURL := aptosFullnodeURL
aptosFullnodeURL = url
aptosFixedNodeLogOnce = sync.Once{}
t.Cleanup(func() {
aptosFullnodeURL = oldURL
aptosFixedNodeLogOnce = sync.Once{}
})
}
func TestAptosGetLedgerVersionUsesFixedPublicNode(t *testing.T) {
cleanup := testutil.SetupTestDatabases(t)
defer cleanup()
data.ResetRpcFailoverForTest()
t.Cleanup(data.ResetRpcFailoverForTest)
var fixedCalls int
fixed := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
fixedCalls++
if r.URL.Path != "/v1" {
http.NotFound(w, r)
return
}
w.Header().Set("Content-Type", "application/json")
_, _ = w.Write([]byte(`{"ledger_version":"123"}`))
}))
defer fixed.Close()
overrideAptosFullnodeURLForTest(t, fixed.URL)
rows := []mdb.RpcNode{
{Network: mdb.NetworkAptos, Url: "https://old-aptos-rpc.example.com", Type: mdb.RpcNodeTypeHttp, Weight: 100, Enabled: true, Purpose: mdb.RpcNodePurposeBoth, Status: mdb.RpcNodeStatusOk},
}
if err := dao.Mdb.Create(&rows).Error; err != nil {
t.Fatalf("seed aptos rpc_nodes: %v", err)
}
got, err := AptosGetLedgerVersion()
if err != nil {
t.Fatalf("AptosGetLedgerVersion(): %v", err)
}
if got != 123 {
t.Fatalf("ledger version = %d, want 123", got)
}
if fixedCalls != 1 {
t.Fatalf("fixed node calls = %d, want 1", fixedCalls)
}
}
func TestValidateManualAptosPaymentUsesFixedPublicNode(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 fixedCalls int
fixed := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
fixedCalls++
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 fixed.Close()
overrideAptosFullnodeURLForTest(t, fixed.URL)
rows := []mdb.RpcNode{
{Network: mdb.NetworkAptos, Url: "https://old-both.example.com", Type: mdb.RpcNodeTypeHttp, Weight: 100, Enabled: true, Purpose: mdb.RpcNodePurposeBoth, Status: mdb.RpcNodeStatusOk},
{Network: mdb.NetworkAptos, Url: "https://old-manual.example.com", Type: mdb.RpcNodeTypeHttp, Weight: 1, Enabled: true, Purpose: mdb.RpcNodePurposeManualVerify, Status: mdb.RpcNodeStatusOk},
}
if err := dao.Mdb.Create(&rows).Error; err != nil {
t.Fatalf("seed aptos rpc_nodes: %v", err)
}
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 fixedCalls != 2 {
t.Fatalf("fixed node calls = %d, want 2", fixedCalls)
}
}
+232
View File
@@ -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)
}
}
+245
View File
@@ -0,0 +1,245 @@
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 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)
}
}
+16
View File
@@ -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,17 @@ func ensureEnabledOrderAsset(network, token string) (*mdb.ChainToken, error) {
if tokenRow == nil || tokenRow.ID == 0 {
return nil, constant.SupportedAssetNotFound
}
if tokenRow.Network == mdb.NetworkAptos {
switch NormalizeAptosPaymentSymbol(tokenRow.Symbol) {
case "USDT", "USDC":
default:
return nil, constant.SupportedAssetNotFound
}
}
contractAddress := strings.TrimSpace(tokenRow.ContractAddress)
if tokenRow.Network == mdb.NetworkAptos && contractAddress == "" {
return nil, constant.SupportedAssetNotFound
}
return tokenRow, nil
}
+27
View File
@@ -469,6 +469,33 @@ 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 TestCreateTransactionSupportedTonTokenWithoutRateReturnsRateError(t *testing.T) {
cleanup := testutil.SetupTestDatabases(t)
defer cleanup()
+1
View File
@@ -15,6 +15,7 @@ func Start() {
go StartPlasmaWebSocketListener()
go StartTronBlockScannerListener()
go StartTonBlockScannerListener()
go StartAptosLedgerScannerListener()
c := cron.New()
+268
View File
@@ -0,0 +1,268 @@
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 rpc=%s chunk_size=%d workers=%d queue_limit=%d",
service.AptosFixedFullnodeURL(),
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
}
state.tokens = filterAptosPaymentTokens(state.tokens)
if len(state.wallets) == 0 || len(state.tokens) == 0 {
log.Sugar.Debug("[APTOS] no enabled wallets or USDT/USDC 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 filterAptosPaymentTokens(tokens []mdb.ChainToken) []mdb.ChainToken {
out := make([]mdb.ChainToken, 0, len(tokens))
for _, token := range tokens {
switch service.NormalizeAptosPaymentSymbol(token.Symbol) {
case "USDT", "USDC":
out = append(out, token)
}
}
return out
}
func chooseAptosIdleDelay(catchup bool, interval time.Duration) time.Duration {
if catchup {
return moveCatchupYield
}
return interval
}
+410
View File
@@ -0,0 +1,410 @@
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},
}).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 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)
}
state.tokens = filterAptosPaymentTokens(state.tokens)
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)
}
state.tokens = filterAptosPaymentTokens(state.tokens)
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)
}
state.tokens = filterAptosPaymentTokens(state.tokens)
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)
}
state.tokens = filterAptosPaymentTokens(state.tokens)
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)
}
state.tokens = filterAptosPaymentTokens(state.tokens)
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)
}
state.tokens = filterAptosPaymentTokens(state.tokens)
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"}
]
}
]`)
}
+77
View File
@@ -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
}
+67
View File
@@ -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)
}
+42
View File
@@ -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)
}
}