From 7d0217b618d7e441fed4c526ff31e9920a587ebc Mon Sep 17 00:00:00 2001 From: Ginta <775650117@qq.com> Date: Wed, 15 Apr 2026 21:21:21 +0800 Subject: [PATCH] tron ws --- src/model/service/task_service.go | 74 +++++++++++++++++ src/task/listen.go | 10 +-- src/task/listen_bsc.go | 2 +- src/task/listen_eth.go | 2 +- src/task/listen_evm_ws.go | 2 +- src/task/listen_plasma.go | 2 +- src/task/listen_polygon.go | 2 +- src/task/listen_trc20_job.go | 129 ++++++++++++++++++++++++------ 8 files changed, 184 insertions(+), 39 deletions(-) diff --git a/src/model/service/task_service.go b/src/model/service/task_service.go index 131da67..26592c0 100644 --- a/src/model/service/task_service.go +++ b/src/model/service/task_service.go @@ -374,6 +374,80 @@ func TryProcessEvmERC20Transfer(chainNetwork string, contract common.Address, to log.Sugar.Infof("[%s-%s][%s] payment processed trade_id=%s hash=%s", net, tokenSym, walletAddr, tradeID, txHash) } +// TryProcessTronTRC20Transfer 处理 Tron 链上 TRC20-USDT 的 Transfer 入账。 +func TryProcessTronTRC20Transfer(toAddr string, rawValue *big.Int, txHash string, blockTsMs int64) { + defer func() { + if err := recover(); err != nil { + log.Sugar.Errorf("[TRON-WS] TryProcessTronTRC20Transfer panic: %v", err) + } + }() + + walletAddr := strings.TrimSpace(toAddr) + if walletAddr == "" { + return + } + if rawValue == nil || rawValue.Sign() <= 0 { + log.Sugar.Infof("[TRON-USDT][%s] skip non-positive or nil amount", 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("[TRON-USDT][%s] skip non-positive amount %.2f", walletAddr, amount) + return + } + + tradeID, err := data.GetTradeIdByWalletAddressAndAmountAndToken(mdb.NetworkTron, walletAddr, "USDT", amount) + if err != nil { + log.Sugar.Warnf("[TRON-USDT][%s] lock lookup: %v", walletAddr, err) + return + } + if tradeID == "" { + log.Sugar.Debugf("[TRON-USDT][%s] skip unmatched tx hash=%s amount=%.2f", walletAddr, txHash, amount) + return + } + + order, err := data.GetOrderInfoByTradeId(tradeID) + if err != nil { + log.Sugar.Warnf("[TRON-USDT][%s] load order: %v", walletAddr, err) + return + } + if blockTsMs > 0 && blockTsMs < order.CreatedAt.TimestampMilli() { + log.Sugar.Warnf("[TRON-USDT][%s] skip tx %s because block time %d is before order create time %d", walletAddr, txHash, blockTsMs, order.CreatedAt.TimestampMilli()) + return + } + if strings.ToLower(strings.TrimSpace(order.Network)) != mdb.NetworkTron { + log.Sugar.Warnf("[TRON-USDT][%s] skip trade_id=%s network=%q", walletAddr, tradeID, order.Network) + return + } + if strings.ToUpper(strings.TrimSpace(order.Token)) != "USDT" { + log.Sugar.Warnf("[TRON-USDT][%s] skip trade_id=%s token mismatch order=%s", walletAddr, tradeID, order.Token) + return + } + + req := &request.OrderProcessingRequest{ + ReceiveAddress: walletAddr, + Token: "USDT", + Network: mdb.NetworkTron, + TradeId: tradeID, + Amount: amount, + BlockTransactionId: txHash, + } + err = OrderProcessing(req) + if err != nil { + if errors.Is(err, constant.OrderBlockAlreadyProcess) || errors.Is(err, constant.OrderStatusConflict) { + log.Sugar.Infof("[TRON-USDT][%s] skip resolved trade_id=%s hash=%s err=%v", walletAddr, tradeID, txHash, err) + return + } + log.Sugar.Errorf("[TRON-USDT][%s] OrderProcessing: %v", walletAddr, err) + return + } + + sendPaymentNotification(order) + log.Sugar.Infof("[TRON-USDT][%s] payment processed trade_id=%s hash=%s", walletAddr, tradeID, txHash) +} + func sendPaymentNotification(order *mdb.Orders) { msg := fmt.Sprintf( "🎉 收款成功通知\n\n"+ diff --git a/src/task/listen.go b/src/task/listen.go index a05177c..57e849e 100644 --- a/src/task/listen.go +++ b/src/task/listen.go @@ -11,17 +11,11 @@ func Start() { go StartBscWebSocketListener() go StartPolygonWebSocketListener() go StartPlasmaWebSocketListener() + go StartTronTrc20WebSocketListener() c := cron.New() - // trc20钱包监听 - _, err := c.AddJob("@every 5s", ListenTrc20Job{}) - if err != nil { - log.Sugar.Errorf("[task] Failed to add ListenTrc20Job: %v", err) - return - } - log.Sugar.Info("[task] ListenTrc20Job scheduled successfully (@every 5s)") // solana钱包监听 - _, err = c.AddJob("@every 5s", ListenSolJob{}) + _, err := c.AddJob("@every 5s", ListenSolJob{}) if err != nil { log.Sugar.Errorf("[task] Failed to add ListenSolJob: %v", err) return diff --git a/src/task/listen_bsc.go b/src/task/listen_bsc.go index 4990c7e..124e672 100644 --- a/src/task/listen_bsc.go +++ b/src/task/listen_bsc.go @@ -58,7 +58,7 @@ func StartBscWebSocketListener() { Topics: [][]common.Hash{}, } - runEvmWsLogListener("[BSC-WS]", wsURL, query, func(client *ethclient.Client, vLog types.Log) { + runWsLogListener("[BSC-WS]", wsURL, query, func(client *ethclient.Client, vLog types.Log) { if len(vLog.Topics) < 3 { return } diff --git a/src/task/listen_eth.go b/src/task/listen_eth.go index d04bada..b94bf8f 100644 --- a/src/task/listen_eth.go +++ b/src/task/listen_eth.go @@ -62,7 +62,7 @@ func StartEthereumWebSocketListener() { Topics: [][]common.Hash{}, } - runEvmWsLogListener("[ETH-WS]", wsURL, query, func(client *ethclient.Client, vLog types.Log) { + runWsLogListener("[ETH-WS]", wsURL, query, func(client *ethclient.Client, vLog types.Log) { if len(vLog.Topics) < 3 { return } diff --git a/src/task/listen_evm_ws.go b/src/task/listen_evm_ws.go index 087b1f7..d832ac5 100644 --- a/src/task/listen_evm_ws.go +++ b/src/task/listen_evm_ws.go @@ -11,7 +11,7 @@ import ( "github.com/ethereum/go-ethereum/ethclient" ) -func runEvmWsLogListener(logPrefix, wsURL string, query ethereum.FilterQuery, handleLog func(*ethclient.Client, types.Log)) { +func runWsLogListener(logPrefix, wsURL string, query ethereum.FilterQuery, handleLog func(*ethclient.Client, types.Log)) { const ( minBackoff = 2 * time.Second maxBackoff = 60 * time.Second diff --git a/src/task/listen_plasma.go b/src/task/listen_plasma.go index 9b44703..8cbe113 100644 --- a/src/task/listen_plasma.go +++ b/src/task/listen_plasma.go @@ -54,7 +54,7 @@ func StartPlasmaWebSocketListener() { Topics: [][]common.Hash{}, } - runEvmWsLogListener("[PLASMA-WS]", wsURL, query, func(client *ethclient.Client, vLog types.Log) { + runWsLogListener("[PLASMA-WS]", wsURL, query, func(client *ethclient.Client, vLog types.Log) { if len(vLog.Topics) < 3 { return } diff --git a/src/task/listen_polygon.go b/src/task/listen_polygon.go index 90aba84..9bd5653 100644 --- a/src/task/listen_polygon.go +++ b/src/task/listen_polygon.go @@ -60,7 +60,7 @@ func StartPolygonWebSocketListener() { Topics: [][]common.Hash{}, } - runEvmWsLogListener("[POLYGON-WS]", wsURL, query, func(client *ethclient.Client, vLog types.Log) { + runWsLogListener("[POLYGON-WS]", wsURL, query, func(client *ethclient.Client, vLog types.Log) { if len(vLog.Topics) < 3 { return } diff --git a/src/task/listen_trc20_job.go b/src/task/listen_trc20_job.go index 676d808..c36eaba 100644 --- a/src/task/listen_trc20_job.go +++ b/src/task/listen_trc20_job.go @@ -1,39 +1,116 @@ package task import ( - "sync" + "context" + "math/big" + "strings" + "sync/atomic" + "time" + tron "github.com/assimon/luuu/crypto" "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" ) -type ListenTrc20Job struct{} +const tronUsdtContractBase58 = "TR7NHqjeKQxGTCi8q8ZY4pL8otSzgjLj6t" -var gListenTrc20JobLock sync.Mutex +// TronGrid 的 JSON-RPC 使用 20-byte hex 地址（去掉 Tron 主网 0x41 前缀）。 +var tronUsdtTrc20Contract = common.HexToAddress("0xa614f803b6fd780986a42c78ec9c7f77e6ded13c") -func (r ListenTrc20Job) Run() { - gListenTrc20JobLock.Lock() - defer gListenTrc20JobLock.Unlock() - log.Sugar.Debug("[ListenTrc20Job] Job triggered") - walletAddress, err := data.GetAvailableWalletAddressByNetwork(mdb.NetworkTron) - if err != nil { - log.Sugar.Errorf("[ListenTrc20Job] Failed to get wallet addresses: %v", err) - return - } - if len(walletAddress) <= 0 { - log.Sugar.Debug("[ListenTrc20Job] No available wallet addresses") - return - } - log.Sugar.Infof("[ListenTrc20Job] Found %d wallet addresses to monitor", len(walletAddress)) - var wg sync.WaitGroup - for _, address := range walletAddress { - log.Sugar.Infof("[ListenTrc20Job] Listening to address: %s", address.Address) - - wg.Add(1) - go service.Trc20CallBack(address.Address, &wg) - } - wg.Wait() - log.Sugar.Debug("[ListenTrc20Job] Job completed") +type tronRecipientSnapshot struct { + addrs map[string]struct{} +} + +var tronWatchedRecipients atomic.Pointer[tronRecipientSnapshot] + +func StartTronTrc20WebSocketListener() { + log.Sugar.Infof("[TRON-WS] listening contract base58=%s hex=%s", tronUsdtContractBase58, tronUsdtTrc20Contract.Hex()) + + wallets, err := data.GetAvailableWalletAddressByNetwork(mdb.NetworkTron) + if err != nil { + log.Sugar.Errorf("[TRON-WS] Failed to get wallet addresses: %v", err) + return + } + storeTronRecipientsFromWallets(wallets) + go func() { + ticker := time.NewTicker(5 * time.Second) + defer ticker.Stop() + for range ticker.C { + w, err := data.GetAvailableWalletAddressByNetwork(mdb.NetworkTron) + if err != nil { + log.Sugar.Warnf("[TRON-WS] refresh wallet addresses: %v", err) + continue + } + storeTronRecipientsFromWallets(w) + } + }() + + wsURL := "wss://api.trongrid.io/jsonrpc" + query := ethereum.FilterQuery{ + Addresses: []common.Address{tronUsdtTrc20Contract}, + Topics: [][]common.Hash{}, + } + runWsLogListener("[TRON-WS]", wsURL, query, func(client *ethclient.Client, vLog types.Log) { + if len(vLog.Topics) < 3 { + return + } + if vLog.Topics[0].String() != transferEventHash.String() { + return + } + + toAddr := tronHexToBase58(common.HexToAddress(vLog.Topics[2].Hex())) + //if !isWatchedTronRecipient(toAddr) { + // return + //} + + amount := new(big.Int).SetBytes(vLog.Data) + if amount.Sign() <= 0 { + return + } + + log.Sugar.Infof("[TRON-WS] Detected TRC20 transfer to %s amount=%s tx=%s", toAddr, amount.String(), vLog.TxHash.Hex()) + var blockTsMs int64 + header, err := client.HeaderByNumber(context.Background(), big.NewInt(int64(vLog.BlockNumber))) + if err != nil { + log.Sugar.Warnf("[TRON-WS] HeaderByNumber block=%d: %v, using local time", vLog.BlockNumber, err) + blockTsMs = time.Now().UnixMilli() + } else { + blockTsMs = int64(header.Time) * 1000 + } + + service.TryProcessTronTRC20Transfer(toAddr, amount, vLog.TxHash.Hex(), blockTsMs) + }) +} + +func storeTronRecipientsFromWallets(wallets []mdb.WalletAddress) int { + m := make(map[string]struct{}) + for _, w := range wallets { + addr := strings.TrimSpace(w.Address) + if addr == "" { + continue + } + m[addr] = struct{}{} + } + tronWatchedRecipients.Store(&tronRecipientSnapshot{addrs: m}) + return len(m) +} + +func isWatchedTronRecipient(to string) bool { + snap := tronWatchedRecipients.Load() + if snap == nil || len(snap.addrs) == 0 { + return false + } + _, ok := snap.addrs[strings.TrimSpace(to)] + return ok +} + +func tronHexToBase58(addr common.Address) string { + raw := append([]byte{tron.PrefixMainnet}, addr.Bytes()...) + return tron.EncodeCheck(raw) }