From 5e68bf8309f3f40e1b1d081c3144b87159584dda Mon Sep 17 00:00:00 2001 From: Ginta <775650117@qq.com> Date: Sun, 12 Apr 2026 19:12:53 +0800 Subject: [PATCH] test: bsc pol plasma --- src/model/mdb/wallet_address_mdb.go | 3 + src/model/service/task_service.go | 92 ++++++++++++++++------ src/task/listen.go | 3 + src/task/listen_bsc.go | 112 +++++++++++++++++++++++++++ src/task/listen_eth.go | 93 +++++++---------------- src/task/listen_evm_ws.go | 81 ++++++++++++++++++++ src/task/listen_plasma.go | 108 ++++++++++++++++++++++++++ src/task/listen_polygon.go | 114 ++++++++++++++++++++++++++++ 8 files changed, 518 insertions(+), 88 deletions(-) create mode 100644 src/task/listen_bsc.go create mode 100644 src/task/listen_evm_ws.go create mode 100644 src/task/listen_plasma.go create mode 100644 src/task/listen_polygon.go diff --git a/src/model/mdb/wallet_address_mdb.go b/src/model/mdb/wallet_address_mdb.go index 7cf2fdb..310b58d 100644 --- a/src/model/mdb/wallet_address_mdb.go +++ b/src/model/mdb/wallet_address_mdb.go @@ -9,6 +9,9 @@ const ( NetworkTron = "tron" NetworkSolana = "solana" NetworkEthereum = "eth" + NetworkBsc = "bsc" + NetworkPolygon = "polygon" + NetworkPlasma = "plasma" ) type WalletAddress struct { diff --git a/src/model/service/task_service.go b/src/model/service/task_service.go index a666513..131da67 100644 --- a/src/model/service/task_service.go +++ b/src/model/service/task_service.go @@ -258,66 +258,104 @@ func checkTrc20Transfers(address string, wg *sync.WaitGroup) { } } -func TryProcessEthereumERC20Transfer(contract common.Address, toAddr common.Address, rawValue *big.Int, txHash string, blockTsMs int64) { +func evmChainLogLabel(chainNetwork string) string { + switch chainNetwork { + case mdb.NetworkEthereum: + return "ETH" + case mdb.NetworkBsc: + return "BSC" + case mdb.NetworkPolygon: + return "POLYGON" + case mdb.NetworkPlasma: + return "PLASMA" + default: + return "EVM" + } +} + +// TryProcessEvmERC20Transfer 处理各 EVM 链上 USDT/USDC（及 Polygon USDC.e）的 Transfer 入账（合约与 network 需一致）。 +func TryProcessEvmERC20Transfer(chainNetwork string, contract common.Address, toAddr common.Address, rawValue *big.Int, txHash string, blockTsMs int64) { defer func() { if err := recover(); err != nil { - log.Sugar.Errorf("[ETH-WS] TryProcessEthereumERC20Transfer panic: %v", err) + log.Sugar.Errorf("[%s-WS] TryProcessEvmERC20Transfer panic: %v", evmChainLogLabel(chainNetwork), err) } }() - usdt := common.HexToAddress("0xdAC17F958D2ee523a2206206994597C13D831ec7") - usdc := common.HexToAddress("0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48") - var tokenSym string - switch contract { - case usdt: - tokenSym = "USDT" - case usdc: - tokenSym = "USDC" + var usdt, usdc common.Address + var polygonUsdcE common.Address + switch chainNetwork { + case mdb.NetworkEthereum: + usdt = common.HexToAddress("0xdAC17F958D2ee523a2206206994597C13D831ec7") + usdc = common.HexToAddress("0xA0b86991c6218b36c1d19d4a2e9eb0ce3606eb48") + case mdb.NetworkBsc: + usdt = common.HexToAddress("0x55d398326f99059fF775485246999027B3197955") + usdc = common.HexToAddress("0x8AC76a51cc950d9822D68b83fE1Ad97B32Cd580d") + case mdb.NetworkPolygon: + usdt = common.HexToAddress("0xc2132D05D31c914a87C6611C10748AEb04B58e8F") + usdc = common.HexToAddress("0x3c499c542cEF5E3811e1192ce70d8cC03d5c3359") + polygonUsdcE = common.HexToAddress("0x2791Bca1f2de4661ED88A30C99A7a9449Aa84174") + case mdb.NetworkPlasma: + // USDT0（官方），6 decimals；链上暂无与 ETH 同级的 Circle USDC 部署，仅匹配 USDT 订单 + usdt = common.HexToAddress("0xB8CE59FC3717ada4C02eaDF9682A9e934F625ebb") default: - log.Sugar.Warnf("[ETH-WS] skip unsupported contract %s", contract.Hex()) return } + var tokenSym string + switch { + case contract == usdt: + tokenSym = "USDT" + case contract == usdc || (polygonUsdcE != (common.Address{}) && contract == polygonUsdcE): + tokenSym = "USDC" + default: + net := evmChainLogLabel(chainNetwork) + log.Sugar.Warnf("[%s-WS] skip unsupported contract %s", net, contract.Hex()) + return + } + + net := evmChainLogLabel(chainNetwork) walletAddr := strings.ToLower(toAddr.Hex()) if rawValue == nil || rawValue.Sign() <= 0 { - log.Sugar.Infof("[ETH-%s][%s] skip non-positive or nil amount", tokenSym, walletAddr) + log.Sugar.Infof("[%s-%s][%s] skip non-positive or nil amount", net, tokenSym, walletAddr) return } decimalQuant := decimal.NewFromBigInt(rawValue, 0) amount := math.MustParsePrecFloat64(decimalQuant.Div(decimal.NewFromInt(1_000_000)).InexactFloat64(), 2) if amount <= 0 { - log.Sugar.Warnf("[ETH-%s][%s] skip non-positive amount %.2f", tokenSym, walletAddr, amount) + log.Sugar.Warnf("[%s-%s][%s] skip non-positive amount %.2f", net, tokenSym, walletAddr, amount) return } - tradeID, err := data.GetTradeIdByWalletAddressAndAmountAndToken(mdb.NetworkEthereum, walletAddr, tokenSym, amount) + log.Sugar.Debugf("[%s-%s][%s] processing transfer hash=%s amount=%.2f", net, tokenSym, walletAddr, txHash, amount) + + tradeID, err := data.GetTradeIdByWalletAddressAndAmountAndToken(chainNetwork, walletAddr, tokenSym, amount) if err != nil { - log.Sugar.Warnf("[ETH-%s][%s] lock lookup: %v", tokenSym, walletAddr, err) + log.Sugar.Warnf("[%s-%s][%s] lock lookup: %v", net, tokenSym, walletAddr, err) return } if tradeID == "" { - log.Sugar.Warnf("[ETH-%s][%s] skip unmatched tx hash=%s amount=%.2f", tokenSym, walletAddr, txHash, amount) + log.Sugar.Warnf("[%s-%s][%s] skip unmatched tx hash=%s amount=%.2f", net, tokenSym, walletAddr, txHash, amount) return } order, err := data.GetOrderInfoByTradeId(tradeID) if err != nil { - log.Sugar.Warnf("[ETH-%s][%s] load order: %v", tokenSym, walletAddr, err) + log.Sugar.Warnf("[%s-%s][%s] load order: %v", net, tokenSym, walletAddr, err) return } - if strings.ToLower(strings.TrimSpace(order.Network)) != mdb.NetworkEthereum { - log.Sugar.Warnf("[ETH-%s][%s] skip trade_id=%s network=%q", tokenSym, walletAddr, tradeID, order.Network) + if strings.ToLower(strings.TrimSpace(order.Network)) != chainNetwork { + log.Sugar.Warnf("[%s-%s][%s] skip trade_id=%s network=%q", net, tokenSym, walletAddr, tradeID, order.Network) return } if strings.ToUpper(strings.TrimSpace(order.Token)) != tokenSym { - log.Sugar.Warnf("[ETH-%s][%s] skip trade_id=%s token mismatch order=%s", tokenSym, walletAddr, tradeID, order.Token) + log.Sugar.Warnf("[%s-%s][%s] skip trade_id=%s token mismatch order=%s", net, tokenSym, walletAddr, tradeID, order.Token) return } req := &request.OrderProcessingRequest{ ReceiveAddress: walletAddr, Token: tokenSym, - Network: mdb.NetworkEthereum, + Network: chainNetwork, TradeId: tradeID, Amount: amount, BlockTransactionId: txHash, @@ -325,15 +363,15 @@ func TryProcessEthereumERC20Transfer(contract common.Address, toAddr common.Addr err = OrderProcessing(req) if err != nil { if errors.Is(err, constant.OrderBlockAlreadyProcess) || errors.Is(err, constant.OrderStatusConflict) { - log.Sugar.Infof("[ETH-%s][%s] skip resolved trade_id=%s hash=%s err=%v", tokenSym, walletAddr, tradeID, txHash, err) + log.Sugar.Infof("[%s-%s][%s] skip resolved trade_id=%s hash=%s err=%v", net, tokenSym, walletAddr, tradeID, txHash, err) return } - log.Sugar.Errorf("[ETH-%s][%s] OrderProcessing: %v", tokenSym, walletAddr, err) + log.Sugar.Errorf("[%s-%s][%s] OrderProcessing: %v", net, tokenSym, walletAddr, err) return } sendPaymentNotification(order) - log.Sugar.Infof("[ETH-%s][%s] payment processed trade_id=%s hash=%s", tokenSym, walletAddr, tradeID, txHash) + log.Sugar.Infof("[%s-%s][%s] payment processed trade_id=%s hash=%s", net, tokenSym, walletAddr, tradeID, txHash) } func sendPaymentNotification(order *mdb.Orders) { @@ -372,6 +410,12 @@ func networkDisplay(n string) string { return "Solana" case mdb.NetworkEthereum: return "Ethereum" + case mdb.NetworkBsc: + return "BSC" + case mdb.NetworkPolygon: + return "Polygon" + case mdb.NetworkPlasma: + return "Plasma" default: if n == "" { return "Tron" diff --git a/src/task/listen.go b/src/task/listen.go index 72121d5..a05177c 100644 --- a/src/task/listen.go +++ b/src/task/listen.go @@ -8,6 +8,9 @@ import ( func Start() { log.Sugar.Info("[task] Starting task scheduler...") go StartEthereumWebSocketListener() + go StartBscWebSocketListener() + go StartPolygonWebSocketListener() + go StartPlasmaWebSocketListener() c := cron.New() // trc20钱包监听 diff --git a/src/task/listen_bsc.go b/src/task/listen_bsc.go new file mode 100644 index 0000000..6cae2f7 --- /dev/null +++ b/src/task/listen_bsc.go @@ -0,0 +1,112 @@ +package task + +import ( + "context" + "math/big" + "strings" + "sync/atomic" + "time" + + "github.com/assimon/luuu/model/data" + "github.com/assimon/luuu/model/mdb" + "github.com/assimon/luuu/model/service" + "github.com/assimon/luuu/util/log" + + "github.com/ethereum/go-ethereum" + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/ethclient" +) + +var ( + // USDT / USDC 合约地址（BSC 主网） + bscUsdtContract = common.HexToAddress("0x55d398326f99059fF775485246999027B3197955") + bscUsdcContract = common.HexToAddress("0x8AC76a51cc950d9822D68b83fE1Ad97B32Cd580d") +) + +type bscRecipientSnapshot struct { + addrs map[string]struct{} +} + +var bscWatchedRecipients atomic.Pointer[bscRecipientSnapshot] + +func StartBscWebSocketListener() { + wallets, err := data.GetAvailableWalletAddressByNetwork(mdb.NetworkBsc) + if err != nil { + log.Sugar.Errorf("[BSC-WS] Failed to get wallet addresses: %v", err) + return + } + storeBscRecipientsFromWallets(wallets) + go func() { + ticker := time.NewTicker(5 * time.Second) + defer ticker.Stop() + for range ticker.C { + w, err := data.GetAvailableWalletAddressByNetwork(mdb.NetworkBsc) + if err != nil { + log.Sugar.Warnf("[BSC-WS] refresh wallet addresses: %v", err) + continue + } + storeBscRecipientsFromWallets(w) + } + }() + wsURL := "wss://bsc.publicnode.com" + query := ethereum.FilterQuery{ + Addresses: []common.Address{ + bscUsdtContract, + bscUsdcContract, + }, + Topics: [][]common.Hash{}, + } + + runEvmWsLogListener("[BSC-WS]", wsURL, query, func(client *ethclient.Client, vLog types.Log) { + if len(vLog.Topics) < 3 { + return + } + + event := vLog.Topics[0].String() + if event != transferEventHash.String() { + return + } + + amount := new(big.Int).SetBytes(vLog.Data) + + toAddr := common.HexToAddress(vLog.Topics[2].Hex()) + + if !isWatchedBscRecipient(toAddr) { + return + } + + var blockTsMs int64 + header, err := client.HeaderByNumber(context.Background(), big.NewInt(int64(vLog.BlockNumber))) + if err != nil { + log.Sugar.Warnf("[BSC-WS] HeaderByNumber block=%d: %v, using local time", vLog.BlockNumber, err) + blockTsMs = time.Now().UnixMilli() + } else { + blockTsMs = int64(header.Time) * 1000 + } + + service.TryProcessEvmERC20Transfer(mdb.NetworkBsc, vLog.Address, toAddr, amount, vLog.TxHash.Hex(), blockTsMs) + }) +} + +func storeBscRecipientsFromWallets(wallets []mdb.WalletAddress) int { + m := make(map[string]struct{}) + for _, w := range wallets { + a := strings.TrimSpace(w.Address) + if !common.IsHexAddress(a) { + continue + } + m[strings.ToLower(common.HexToAddress(a).Hex())] = struct{}{} + } + bscWatchedRecipients.Store(&bscRecipientSnapshot{addrs: m}) + return len(m) +} + +func isWatchedBscRecipient(to common.Address) bool { + snap := bscWatchedRecipients.Load() + if snap == nil || len(snap.addrs) == 0 { + return false + } + _, ok := snap.addrs[strings.ToLower(to.Hex())] + return ok +} diff --git a/src/task/listen_eth.go b/src/task/listen_eth.go index 127a6b8..d04bada 100644 --- a/src/task/listen_eth.go +++ b/src/task/listen_eth.go @@ -2,7 +2,6 @@ package task import ( "context" - "fmt" "math/big" "strings" "sync/atomic" @@ -38,7 +37,7 @@ var ethWatchedRecipients atomic.Pointer[ethRecipientSnapshot] func StartEthereumWebSocketListener() { wallets, err := data.GetAvailableWalletAddressByNetwork(mdb.NetworkEthereum) if err != nil { - log.Sugar.Fatalf("Failed to get wallet addresses: %v", err) + log.Sugar.Errorf("Failed to get wallet addresses: %v", err) return } StoreEthRecipientsFromWallets(wallets) @@ -55,16 +54,6 @@ func StartEthereumWebSocketListener() { } }() wsURL := "wss://ethereum.publicnode.com" - - client, err := ethclient.Dial(wsURL) - if err != nil { - log.Sugar.Fatal("连接失败:", err) - } - - // 创建日志通道 - logsCh := make(chan types.Log) - - // 订阅条件 query := ethereum.FilterQuery{ Addresses: []common.Address{ usdtContract, @@ -73,59 +62,35 @@ func StartEthereumWebSocketListener() { Topics: [][]common.Hash{}, } - // 订阅日志（核心） - sub, err := client.SubscribeFilterLogs(context.Background(), query, logsCh) - if err != nil { - log.Sugar.Fatal("订阅失败:", err) - } - - fmt.Println("🚀 开始监听 USDT / USDC 收款...") - - for { - select { - case err := <-sub.Err(): - log.Sugar.Fatal("订阅错误:", err) - - case vLog := <-logsCh: - - // topics: - // [0] Transfer event - // [1] from - // [2] to - if len(vLog.Topics) < 3 { - continue - } - - event := vLog.Topics[0].String() - if event != transferEventHash.String() { - continue - } - - // value from data - amount := new(big.Int).SetBytes(vLog.Data) - - - if event != transferEventHash.String() { - continue - } - toAddr := common.HexToAddress(vLog.Topics[2].Hex()) - - if !isWatchedEthRecipient(toAddr) { - continue - } - - var blockTsMs int64 - header, err := client.HeaderByNumber(context.Background(), big.NewInt(int64(vLog.BlockNumber))) - if err != nil { - log.Sugar.Warnf("[ETH-WS] HeaderByNumber block=%d: %v, using local time", vLog.BlockNumber, err) - blockTsMs = time.Now().UnixMilli() - } else { - blockTsMs = int64(header.Time) * 1000 - } - - service.TryProcessEthereumERC20Transfer(vLog.Address, toAddr, amount, vLog.TxHash.Hex(), blockTsMs) + runEvmWsLogListener("[ETH-WS]", wsURL, query, func(client *ethclient.Client, vLog types.Log) { + if len(vLog.Topics) < 3 { + return } - } + + event := vLog.Topics[0].String() + if event != transferEventHash.String() { + return + } + + amount := new(big.Int).SetBytes(vLog.Data) + + toAddr := common.HexToAddress(vLog.Topics[2].Hex()) + + if !isWatchedEthRecipient(toAddr) { + return + } + + var blockTsMs int64 + header, err := client.HeaderByNumber(context.Background(), big.NewInt(int64(vLog.BlockNumber))) + if err != nil { + log.Sugar.Warnf("[ETH-WS] HeaderByNumber block=%d: %v, using local time", vLog.BlockNumber, err) + blockTsMs = time.Now().UnixMilli() + } else { + blockTsMs = int64(header.Time) * 1000 + } + + service.TryProcessEvmERC20Transfer(mdb.NetworkEthereum, vLog.Address, toAddr, amount, vLog.TxHash.Hex(), blockTsMs) + }) } func StoreEthRecipientsFromWallets(wallets []mdb.WalletAddress) int { diff --git a/src/task/listen_evm_ws.go b/src/task/listen_evm_ws.go new file mode 100644 index 0000000..087b1f7 --- /dev/null +++ b/src/task/listen_evm_ws.go @@ -0,0 +1,81 @@ +package task + +import ( + "context" + "time" + + "github.com/assimon/luuu/util/log" + + "github.com/ethereum/go-ethereum" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/ethclient" +) + +func runEvmWsLogListener(logPrefix, wsURL string, query ethereum.FilterQuery, handleLog func(*ethclient.Client, types.Log)) { + const ( + minBackoff = 2 * time.Second + maxBackoff = 60 * time.Second + rejoinWait = 3 * time.Second + ) + failWait := minBackoff + + for { + client, err := ethclient.Dial(wsURL) + if err != nil { + log.Sugar.Warnf("%s dial: %v, retry in %s", logPrefix, err, failWait) + time.Sleep(failWait) + failWait = nextBackoff(failWait, maxBackoff) + continue + } + + logsCh := make(chan types.Log) + sub, err := client.SubscribeFilterLogs(context.Background(), query, logsCh) + if err != nil { + client.Close() + log.Sugar.Warnf("%s subscribe: %v, retry in %s", logPrefix, err, failWait) + time.Sleep(failWait) + failWait = nextBackoff(failWait, maxBackoff) + continue + } + failWait = minBackoff + + log.Sugar.Infof("%s connected, subscribed to USDT/USDC Transfer logs", logPrefix) + + recvLoop(client, sub, logsCh, logPrefix, handleLog) + + time.Sleep(rejoinWait) + } +} + +func recvLoop(client *ethclient.Client, sub ethereum.Subscription, logsCh <-chan types.Log, logPrefix string, handleLog func(*ethclient.Client, types.Log)) { + defer func() { + sub.Unsubscribe() + client.Close() + }() + + for { + select { + case err := <-sub.Err(): + if err != nil { + log.Sugar.Warnf("%s subscription error: %v, reconnecting", logPrefix, err) + } else { + log.Sugar.Warnf("%s subscription closed, reconnecting", logPrefix) + } + return + case vLog, ok := <-logsCh: + if !ok { + log.Sugar.Warnf("%s log channel closed, reconnecting", logPrefix) + return + } + handleLog(client, vLog) + } + } +} + +func nextBackoff(cur, max time.Duration) time.Duration { + n := cur * 2 + if n > max { + return max + } + return n +} diff --git a/src/task/listen_plasma.go b/src/task/listen_plasma.go new file mode 100644 index 0000000..9b44703 --- /dev/null +++ b/src/task/listen_plasma.go @@ -0,0 +1,108 @@ +package task + +import ( + "context" + "math/big" + "strings" + "sync/atomic" + "time" + + "github.com/assimon/luuu/model/data" + "github.com/assimon/luuu/model/mdb" + "github.com/assimon/luuu/model/service" + "github.com/assimon/luuu/util/log" + + "github.com/ethereum/go-ethereum" + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/ethclient" +) + +var ( + plasmaUsdt0Contract = common.HexToAddress("0xB8CE59FC3717ada4C02eaDF9682A9e934F625ebb") +) + +type plasmaRecipientSnapshot struct { + addrs map[string]struct{} +} + +var plasmaWatchedRecipients atomic.Pointer[plasmaRecipientSnapshot] + +func StartPlasmaWebSocketListener() { + wallets, err := data.GetAvailableWalletAddressByNetwork(mdb.NetworkPlasma) + if err != nil { + log.Sugar.Errorf("[PLASMA-WS] Failed to get wallet addresses: %v", err) + return + } + storePlasmaRecipientsFromWallets(wallets) + go func() { + ticker := time.NewTicker(5 * time.Second) + defer ticker.Stop() + for range ticker.C { + w, err := data.GetAvailableWalletAddressByNetwork(mdb.NetworkPlasma) + if err != nil { + log.Sugar.Warnf("[PLASMA-WS] refresh wallet addresses: %v", err) + continue + } + storePlasmaRecipientsFromWallets(w) + } + }() + // 文档提供 https://rpc.plasma.to；常见实现同主机 wss + wsURL := "wss://rpc.plasma.to" + query := ethereum.FilterQuery{ + Addresses: []common.Address{plasmaUsdt0Contract}, + Topics: [][]common.Hash{}, + } + + runEvmWsLogListener("[PLASMA-WS]", wsURL, query, func(client *ethclient.Client, vLog types.Log) { + if len(vLog.Topics) < 3 { + return + } + + event := vLog.Topics[0].String() + if event != transferEventHash.String() { + return + } + + amount := new(big.Int).SetBytes(vLog.Data) + + toAddr := common.HexToAddress(vLog.Topics[2].Hex()) + + if !isWatchedPlasmaRecipient(toAddr) { + return + } + + var blockTsMs int64 + header, err := client.HeaderByNumber(context.Background(), big.NewInt(int64(vLog.BlockNumber))) + if err != nil { + log.Sugar.Warnf("[PLASMA-WS] HeaderByNumber block=%d: %v, using local time", vLog.BlockNumber, err) + blockTsMs = time.Now().UnixMilli() + } else { + blockTsMs = int64(header.Time) * 1000 + } + + service.TryProcessEvmERC20Transfer(mdb.NetworkPlasma, vLog.Address, toAddr, amount, vLog.TxHash.Hex(), blockTsMs) + }) +} + +func storePlasmaRecipientsFromWallets(wallets []mdb.WalletAddress) int { + m := make(map[string]struct{}) + for _, w := range wallets { + a := strings.TrimSpace(w.Address) + if !common.IsHexAddress(a) { + continue + } + m[strings.ToLower(common.HexToAddress(a).Hex())] = struct{}{} + } + plasmaWatchedRecipients.Store(&plasmaRecipientSnapshot{addrs: m}) + return len(m) +} + +func isWatchedPlasmaRecipient(to common.Address) bool { + snap := plasmaWatchedRecipients.Load() + if snap == nil || len(snap.addrs) == 0 { + return false + } + _, ok := snap.addrs[strings.ToLower(to.Hex())] + return ok +} diff --git a/src/task/listen_polygon.go b/src/task/listen_polygon.go new file mode 100644 index 0000000..9734e4d --- /dev/null +++ b/src/task/listen_polygon.go @@ -0,0 +1,114 @@ +package task + +import ( + "context" + "math/big" + "strings" + "sync/atomic" + "time" + + "github.com/assimon/luuu/model/data" + "github.com/assimon/luuu/model/mdb" + "github.com/assimon/luuu/model/service" + "github.com/assimon/luuu/util/log" + + "github.com/ethereum/go-ethereum" + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/ethclient" +) + +var ( + // Polygon PoS 主网 USDT / USDC（原生）/ USDC.e（桥接） + polygonUsdtContract = common.HexToAddress("0xc2132D05D31c914a87C6611C10748AEb04B58e8F") + polygonUsdcContract = common.HexToAddress("0x3c499c542cEF5E3811e1192ce70d8cC03d5c3359") + polygonUsdcEContract = common.HexToAddress("0x2791Bca1f2de4661ED88A30C99A7a9449Aa84174") +) + +type polygonRecipientSnapshot struct { + addrs map[string]struct{} +} + +var polygonWatchedRecipients atomic.Pointer[polygonRecipientSnapshot] + +func StartPolygonWebSocketListener() { + wallets, err := data.GetAvailableWalletAddressByNetwork(mdb.NetworkPolygon) + if err != nil { + log.Sugar.Errorf("[POLYGON-WS] Failed to get wallet addresses: %v", err) + return + } + storePolygonRecipientsFromWallets(wallets) + go func() { + ticker := time.NewTicker(5 * time.Second) + defer ticker.Stop() + for range ticker.C { + w, err := data.GetAvailableWalletAddressByNetwork(mdb.NetworkPolygon) + if err != nil { + log.Sugar.Warnf("[POLYGON-WS] refresh wallet addresses: %v", err) + continue + } + storePolygonRecipientsFromWallets(w) + } + }() + wsURL := "wss://polygon-bor-ws.publicnode.com" + query := ethereum.FilterQuery{ + Addresses: []common.Address{ + polygonUsdtContract, + polygonUsdcContract, + polygonUsdcEContract, + }, + Topics: [][]common.Hash{}, + } + + runEvmWsLogListener("[POLYGON-WS]", wsURL, query, func(client *ethclient.Client, vLog types.Log) { + if len(vLog.Topics) < 3 { + return + } + + event := vLog.Topics[0].String() + if event != transferEventHash.String() { + return + } + + amount := new(big.Int).SetBytes(vLog.Data) + + toAddr := common.HexToAddress(vLog.Topics[2].Hex()) + + if !isWatchedPolygonRecipient(toAddr) { + return + } + + var blockTsMs int64 + header, err := client.HeaderByNumber(context.Background(), big.NewInt(int64(vLog.BlockNumber))) + if err != nil { + log.Sugar.Warnf("[POLYGON-WS] HeaderByNumber block=%d: %v, using local time", vLog.BlockNumber, err) + blockTsMs = time.Now().UnixMilli() + } else { + blockTsMs = int64(header.Time) * 1000 + } + + service.TryProcessEvmERC20Transfer(mdb.NetworkPolygon, vLog.Address, toAddr, amount, vLog.TxHash.Hex(), blockTsMs) + }) +} + +func storePolygonRecipientsFromWallets(wallets []mdb.WalletAddress) int { + m := make(map[string]struct{}) + for _, w := range wallets { + a := strings.TrimSpace(w.Address) + if !common.IsHexAddress(a) { + continue + } + m[strings.ToLower(common.HexToAddress(a).Hex())] = struct{}{} + } + polygonWatchedRecipients.Store(&polygonRecipientSnapshot{addrs: m}) + return len(m) +} + +func isWatchedPolygonRecipient(to common.Address) bool { + snap := polygonWatchedRecipients.Load() + if snap == nil || len(snap.addrs) == 0 { + return false + } + _, ok := snap.addrs[strings.ToLower(to.Hex())] + return ok +}