tron ws

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(
 		"🎉 <b>收款成功通知</b>\n\n"+

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

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
 		}

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
 		}

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

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
 		}

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
 		}

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