test: bsc pol plasma

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 {

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")
+	var usdt, usdc common.Address
-	usdc := common.HexToAddress("0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48")
+	var polygonUsdcE common.Address
-	var tokenSym string
+	switch chainNetwork {
-	switch contract {
+	case mdb.NetworkEthereum:
-	case usdt:
+		usdt = common.HexToAddress("0xdAC17F958D2ee523a2206206994597C13D831ec7")
-		tokenSym = "USDT"
+		usdc = common.HexToAddress("0xA0b86991c6218b36c1d19d4a2e9eb0ce3606eb48")
-	case usdc:
+	case mdb.NetworkBsc:
-		tokenSym = "USDC"
+		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 {
+	if strings.ToLower(strings.TrimSpace(order.Network)) != chainNetwork {
-		log.Sugar.Warnf("[ETH-%s][%s] skip trade_id=%s network=%q", tokenSym, walletAddr, tradeID, order.Network)
+		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"

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钱包监听

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
+}

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,45 +62,22 @@ func StartEthereumWebSocketListener() {
 		Topics: [][]common.Hash{},
 	}
 
-	// 订阅日志（核心）
+	runEvmWsLogListener("[ETH-WS]", wsURL, query, func(client *ethclient.Client, vLog types.Log) {
-	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
+			return
 		}
 
 		event := vLog.Topics[0].String()
 		if event != transferEventHash.String() {
-				continue
+			return
 		}
 
-			// 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
+			return
 		}
 
 		var blockTsMs int64
@@ -123,9 +89,8 @@ func StartEthereumWebSocketListener() {
 			blockTsMs = int64(header.Time) * 1000
 		}
 
-			service.TryProcessEthereumERC20Transfer(vLog.Address, toAddr, amount, vLog.TxHash.Hex(), blockTsMs)
+		service.TryProcessEvmERC20Transfer(mdb.NetworkEthereum, vLog.Address, toAddr, amount, vLog.TxHash.Hex(), blockTsMs)
-		}
+	})
-	}
 }
 
 func StoreEthRecipientsFromWallets(wallets []mdb.WalletAddress) int {

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
+}

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
+}

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
+}