From 74c3a3a08cf4a3114118bbe0b397630e22c2da61 Mon Sep 17 00:00:00 2001 From: Ginta <775650117@qq.com> Date: Mon, 27 Apr 2026 00:04:05 +0800 Subject: [PATCH] Revert "new trc (#54)" This reverts commit 0a19854602ed9748c9282ed82e49f0ef75929f4d. --- src/model/service/task_service.go | 463 +++++++++++++++++++++--------- src/task/listen.go | 19 +- src/task/listen_trc20_block.go | 378 ------------------------ src/task/listen_trc20_job.go | 43 +++ 4 files changed, 380 insertions(+), 523 deletions(-) delete mode 100644 src/task/listen_trc20_block.go create mode 100644 src/task/listen_trc20_job.go diff --git a/src/model/service/task_service.go b/src/model/service/task_service.go index 6ff16be..8e6c0d2 100644 --- a/src/model/service/task_service.go +++ b/src/model/service/task_service.go @@ -1,23 +1,32 @@ package service import ( + "encoding/hex" "errors" "fmt" "math/big" + "net/http" "strings" + "sync" + tron "github.com/GMWalletApp/epusdt/crypto" "github.com/GMWalletApp/epusdt/model/data" "github.com/GMWalletApp/epusdt/model/mdb" "github.com/GMWalletApp/epusdt/model/request" "github.com/GMWalletApp/epusdt/notify" "github.com/GMWalletApp/epusdt/util/constant" + "github.com/GMWalletApp/epusdt/util/http_client" "github.com/GMWalletApp/epusdt/util/log" "github.com/GMWalletApp/epusdt/util/math" "github.com/dromara/carbon/v2" "github.com/ethereum/go-ethereum/common" + "github.com/gookit/goutil/stdutil" "github.com/shopspring/decimal" + "github.com/tidwall/gjson" ) +// resolveTronNode returns (baseURL, apiKey) for the TRON HTTP RPC node. +// It reads the first healthy (or any enabled) row from the rpc_nodes table. func resolveTronNode() (string, string, error) { node, err := data.SelectRpcNode(mdb.NetworkTron, mdb.RpcNodeTypeHttp) if err != nil { @@ -33,128 +42,296 @@ func resolveTronNode() (string, string, error) { return rpcURL, node.ApiKey, nil } -func ResolveTronNode() (string, string, error) { - return resolveTronNode() -} - -func TryProcessTronTRC20Transfer(toAddr string, rawValue *big.Int, txHash string, blockTsMs int64) { +func Trc20CallBack(address string, wg *sync.WaitGroup) { + defer wg.Done() defer func() { if err := recover(); err != nil { - log.Sugar.Errorf("[TRC20][%s] TryProcessTronTRC20Transfer panic: %v", toAddr, err) + log.Sugar.Error(err) } }() - addr := strings.TrimSpace(toAddr) - if addr == "" || rawValue == nil || rawValue.Sign() <= 0 { - return - } - - decimalQuant := decimal.NewFromBigInt(rawValue, 0) - amount := math.MustParsePrecFloat64(decimalQuant.Div(decimal.NewFromInt(1_000_000)).InexactFloat64(), 2) - if amount <= 0 { - return - } - - tradeID, err := data.GetTradeIdByWalletAddressAndAmountAndToken(mdb.NetworkTron, addr, "USDT", amount) - if err != nil { - log.Sugar.Warnf("[TRC20][%s] lock lookup: %v", addr, err) - return - } - if tradeID == "" { - log.Sugar.Debugf("[TRC20][%s] skip unmatched tx hash=%s amount=%.2f", addr, txHash, amount) - return - } - - order, err := data.GetOrderInfoByTradeId(tradeID) - if err != nil { - log.Sugar.Warnf("[TRC20][%s] load order: %v", addr, err) - return - } - if blockTsMs > 0 && blockTsMs < order.CreatedAt.TimestampMilli() { - log.Sugar.Warnf("[TRC20][%s] skip tx %s because block time %d is before order create time %d", addr, txHash, blockTsMs, order.CreatedAt.TimestampMilli()) - return - } - - req := &request.OrderProcessingRequest{ - ReceiveAddress: addr, - 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("[TRC20][%s] skip resolved transfer trade_id=%s hash=%s err=%v", addr, tradeID, txHash, err) - return - } - log.Sugar.Errorf("[TRC20][%s] OrderProcessing trade_id=%s hash=%s: %v", addr, tradeID, txHash, err) - return - } - - sendPaymentNotification(order) - log.Sugar.Infof("[TRC20][%s] payment processed trade_id=%s hash=%s", addr, tradeID, txHash) + var innerWg sync.WaitGroup + innerWg.Add(2) + go checkTrxTransfers(address, &innerWg) + go checkTrc20Transfers(address, &innerWg) + innerWg.Wait() } -func TryProcessTronTRXTransfer(toAddr string, rawSun int64, txHash string, blockTsMs int64) { - defer func() { - if err := recover(); err != nil { - log.Sugar.Errorf("[TRX][%s] TryProcessTronTRXTransfer panic: %v", toAddr, err) +func checkTrxTransfers(address string, wg *sync.WaitGroup) { + defer wg.Done() + + // Native TRX is gated by a chain_tokens row with empty + // contract_address and symbol=TRX. Admin can disable this row to + // stop scanning native transfers without touching the chain toggle. + trxCfg, err := data.GetEnabledChainTokenBySymbol(mdb.NetworkTron, "TRX") + if err != nil { + log.Sugar.Errorf("[TRX][%s] load chain_tokens err=%v", address, err) + return + } + if trxCfg == nil || trxCfg.ID == 0 { + log.Sugar.Debugf("[TRX][%s] native TRX disabled in chain_tokens, skipping", address) + return + } + trxDecimals := trxCfg.Decimals + if trxDecimals <= 0 { + trxDecimals = 6 + } + + client := http_client.GetHttpClient() + startTime := carbon.Now().AddHours(-24).TimestampMilli() + endTime := carbon.Now().TimestampMilli() + tronBaseURL, tronAPIKey, err := resolveTronNode() + if err != nil { + log.Sugar.Errorf("[TRX][%s] resolve rpc_nodes err=%v", address, err) + return + } + url := fmt.Sprintf("%s/v1/accounts/%s/transactions", tronBaseURL, address) + + resp, err := client.R().SetQueryParams(map[string]string{ + "order_by": "block_timestamp,desc", + "limit": "100", + "only_to": "true", + "min_timestamp": stdutil.ToString(startTime), + "max_timestamp": stdutil.ToString(endTime), + }).SetHeader("TRON-PRO-API-KEY", tronAPIKey).Get(url) + if err != nil { + log.Sugar.Errorf("[TRX][%s] HTTP request failed: %v", address, err) + return + } + if resp.StatusCode() != http.StatusOK { + log.Sugar.Errorf("[TRX][%s] API returned status %d", address, resp.StatusCode()) + return + } + + success := gjson.GetBytes(resp.Body(), "success").Bool() + if !success { + log.Sugar.Errorf("[TRX][%s] API response indicates failure", address) + return + } + + transfers := gjson.GetBytes(resp.Body(), "data").Array() + if len(transfers) == 0 { + log.Sugar.Debugf("[TRX][%s] no transfer records found", address) + return + } + log.Sugar.Debugf("[TRX][%s] fetched %d transfer records", address, len(transfers)) + + for i, transfer := range transfers { + if transfer.Get("raw_data.contract.0.type").String() != "TransferContract" { + continue } - }() - - addr := strings.TrimSpace(toAddr) - if addr == "" || rawSun <= 0 { - return - } - - decimalQuant := decimal.NewFromInt(rawSun) - amount := math.MustParsePrecFloat64(decimalQuant.Div(decimal.NewFromInt(1_000_000)).InexactFloat64(), 2) - if amount <= 0 { - return - } - - tradeID, err := data.GetTradeIdByWalletAddressAndAmountAndToken(mdb.NetworkTron, addr, "TRX", amount) - if err != nil { - log.Sugar.Warnf("[TRX][%s] lock lookup: %v", addr, err) - return - } - if tradeID == "" { - log.Sugar.Debugf("[TRX][%s] skip unmatched tx hash=%s amount=%.2f", addr, txHash, amount) - return - } - - order, err := data.GetOrderInfoByTradeId(tradeID) - if err != nil { - log.Sugar.Warnf("[TRX][%s] load order: %v", addr, err) - return - } - if blockTsMs > 0 && blockTsMs < order.CreatedAt.TimestampMilli() { - log.Sugar.Warnf("[TRX][%s] skip tx %s because block time %d is before order create time %d", addr, txHash, blockTsMs, order.CreatedAt.TimestampMilli()) - return - } - - req := &request.OrderProcessingRequest{ - ReceiveAddress: addr, - Token: "TRX", - 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("[TRX][%s] skip resolved transfer trade_id=%s hash=%s err=%v", addr, tradeID, txHash, err) - return + if transfer.Get("ret.0.contractRet").String() != "SUCCESS" { + continue } - log.Sugar.Errorf("[TRX][%s] OrderProcessing trade_id=%s hash=%s: %v", addr, tradeID, txHash, err) + + toAddressHex := transfer.Get("raw_data.contract.0.parameter.value.to_address").String() + toBytes, err := hex.DecodeString(toAddressHex) + if err != nil { + log.Sugar.Errorf("[TRX][%s] decode address failed on tx #%d: %v", address, i, err) + continue + } + if tron.EncodeCheck(toBytes) != address { + continue + } + + rawAmount := transfer.Get("raw_data.contract.0.parameter.value.amount").String() + decimalQuant, err := decimal.NewFromString(rawAmount) + if err != nil { + log.Sugar.Errorf("[TRX][%s] parse amount failed on tx #%d: %v", address, i, err) + continue + } + amount := math.MustParsePrecFloat64(decimalQuant.Div(decimal.New(1, int32(trxDecimals))).InexactFloat64(), 2) + if trxCfg.MinAmount > 0 && amount < trxCfg.MinAmount { + continue + } + if amount <= 0 { + continue + } + + txID := transfer.Get("txID").String() + tradeID, err := data.GetTradeIdByWalletAddressAndAmountAndToken(mdb.NetworkTron, address, strings.ToUpper(strings.TrimSpace(trxCfg.Symbol)), amount) + if err != nil { + log.Sugar.Errorf("[TRX][%s] lookup trade_id failed hash=%s err=%v", address, txID, err) + continue + } + if tradeID == "" { + log.Sugar.Debugf("[TRX][%s] skip unmatched tx hash=%s amount=%.2f", address, txID, amount) + continue + } + log.Sugar.Infof("[TRX][%s] matched trade_id=%s hash=%s amount=%.2f", address, tradeID, txID, amount) + + order, err := data.GetOrderInfoByTradeId(tradeID) + if err != nil { + log.Sugar.Errorf("[TRX][%s] get order failed trade_id=%s err=%v", address, tradeID, err) + continue + } + blockTimestamp := transfer.Get("block_timestamp").Int() + createTime := order.CreatedAt.TimestampMilli() + if blockTimestamp < createTime { + log.Sugar.Warnf("[TRX][%s] skip tx %s because block time %d is before order create time %d", address, txID, blockTimestamp, createTime) + continue + } + + req := &request.OrderProcessingRequest{ + ReceiveAddress: address, + Token: strings.ToUpper(strings.TrimSpace(trxCfg.Symbol)), + Network: mdb.NetworkTron, + TradeId: tradeID, + Amount: amount, + BlockTransactionId: txID, + } + err = OrderProcessing(req) + if err != nil { + if errors.Is(err, constant.OrderBlockAlreadyProcess) || errors.Is(err, constant.OrderStatusConflict) { + log.Sugar.Infof("[TRX][%s] skip resolved transfer trade_id=%s hash=%s err=%v", address, tradeID, txID, err) + continue + } + log.Sugar.Errorf("[TRX][%s] order processing failed trade_id=%s hash=%s err=%v", address, tradeID, txID, err) + continue + } + + sendPaymentNotification(order) + log.Sugar.Infof("[TRX][%s] payment processed trade_id=%s hash=%s", address, tradeID, txID) + } +} + +func checkTrc20Transfers(address string, wg *sync.WaitGroup) { + defer wg.Done() + + // Build contract -> token map for the TRON network. If nothing is + // configured, skip — preserves the previous behavior of only watching + // admin-approved tokens. + tokens, err := data.ListEnabledChainTokensByNetwork(mdb.NetworkTron) + if err != nil { + log.Sugar.Errorf("[TRC20][%s] load chain_tokens err=%v", address, err) + return + } + if len(tokens) == 0 { + log.Sugar.Debugf("[TRC20][%s] no enabled chain_tokens, skipping", address) + return + } + contractTokens := make(map[string]*mdb.ChainToken, len(tokens)) + for i := range tokens { + c := strings.TrimSpace(tokens[i].ContractAddress) + if c == "" { + continue + } + contractTokens[c] = &tokens[i] + } + + client := http_client.GetHttpClient() + startTime := carbon.Now().AddHours(-24).TimestampMilli() + endTime := carbon.Now().TimestampMilli() + tronBaseURL, tronAPIKey, err := resolveTronNode() + if err != nil { + log.Sugar.Errorf("[TRC20][%s] resolve rpc_nodes err=%v", address, err) + return + } + url := fmt.Sprintf("%s/v1/accounts/%s/transactions/trc20", tronBaseURL, address) + + resp, err := client.R().SetQueryParams(map[string]string{ + "order_by": "block_timestamp,desc", + "limit": "100", + "only_to": "true", + "min_timestamp": stdutil.ToString(startTime), + "max_timestamp": stdutil.ToString(endTime), + }).SetHeader("TRON-PRO-API-KEY", tronAPIKey).Get(url) + if err != nil { + log.Sugar.Errorf("[TRC20][%s] HTTP request failed: %v", address, err) + return + } + if resp.StatusCode() != http.StatusOK { + log.Sugar.Errorf("[TRC20][%s] API returned status %d", address, resp.StatusCode()) return } - sendPaymentNotification(order) - log.Sugar.Infof("[TRX][%s] payment processed trade_id=%s hash=%s", addr, tradeID, txHash) + success := gjson.GetBytes(resp.Body(), "success").Bool() + if !success { + log.Sugar.Errorf("[TRC20][%s] API response indicates failure", address) + return + } + + transfers := gjson.GetBytes(resp.Body(), "data").Array() + if len(transfers) == 0 { + log.Sugar.Debugf("[TRC20][%s] no transfer records found", address) + return + } + log.Sugar.Debugf("[TRC20][%s] fetched %d transfer records", address, len(transfers)) + + for i, transfer := range transfers { + contractAddr := transfer.Get("token_info.address").String() + cfg, ok := contractTokens[contractAddr] + if !ok { + continue + } + if transfer.Get("to").String() != address { + continue + } + tokenSym := strings.ToUpper(strings.TrimSpace(cfg.Symbol)) + + valueStr := transfer.Get("value").String() + decimalQuant, err := decimal.NewFromString(valueStr) + if err != nil { + log.Sugar.Errorf("[TRC20][%s] parse value failed on tx #%d: %v", address, i, err) + continue + } + tokenDecimals := transfer.Get("token_info.decimals").Int() + if tokenDecimals <= 0 { + tokenDecimals = int64(cfg.Decimals) + } + amount := math.MustParsePrecFloat64(decimalQuant.Div(decimal.New(1, int32(tokenDecimals))).InexactFloat64(), 2) + if cfg.MinAmount > 0 && amount < cfg.MinAmount { + continue + } + if amount <= 0 { + continue + } + + txID := transfer.Get("transaction_id").String() + tradeID, err := data.GetTradeIdByWalletAddressAndAmountAndToken(mdb.NetworkTron, address, tokenSym, amount) + if err != nil { + log.Sugar.Errorf("[TRC20][%s] lookup trade_id failed hash=%s err=%v", address, txID, err) + continue + } + if tradeID == "" { + log.Sugar.Debugf("[TRC20][%s] skip unmatched %s tx hash=%s amount=%.2f", address, tokenSym, txID, amount) + continue + } + log.Sugar.Infof("[TRC20][%s] matched %s trade_id=%s hash=%s amount=%.2f", address, tokenSym, tradeID, txID, amount) + + order, err := data.GetOrderInfoByTradeId(tradeID) + if err != nil { + log.Sugar.Errorf("[TRC20][%s] get order failed trade_id=%s err=%v", address, tradeID, err) + continue + } + blockTimestamp := transfer.Get("block_timestamp").Int() + createTime := order.CreatedAt.TimestampMilli() + if blockTimestamp < createTime { + log.Sugar.Warnf("[TRC20][%s] skip tx %s because block time %d is before order create time %d", address, txID, blockTimestamp, createTime) + continue + } + + req := &request.OrderProcessingRequest{ + ReceiveAddress: address, + Token: tokenSym, + Network: mdb.NetworkTron, + TradeId: tradeID, + Amount: amount, + BlockTransactionId: txID, + } + err = OrderProcessing(req) + if err != nil { + if errors.Is(err, constant.OrderBlockAlreadyProcess) || errors.Is(err, constant.OrderStatusConflict) { + log.Sugar.Infof("[TRC20][%s] skip resolved transfer trade_id=%s hash=%s err=%v", address, tradeID, txID, err) + continue + } + log.Sugar.Errorf("[TRC20][%s] order processing failed trade_id=%s hash=%s err=%v", address, tradeID, txID, err) + continue + } + + sendPaymentNotification(order) + log.Sugar.Infof("[TRC20][%s] payment processed trade_id=%s hash=%s", address, tradeID, txID) + } } func evmChainLogLabel(chainNetwork string) string { @@ -172,6 +349,9 @@ func evmChainLogLabel(chainNetwork string) string { } } +// TryProcessEvmERC20Transfer 处理各 EVM 链上代币的 Transfer 入账。 +// 代币识别、符号和 decimals 全部从 chain_tokens 表动态查询 — +// 管理后台新增 token 即可立即生效，无需代码改动。 func TryProcessEvmERC20Transfer(chainNetwork string, contract common.Address, toAddr common.Address, rawValue *big.Int, txHash string, blockTsMs int64) { defer func() { if err := recover(); err != nil { @@ -179,46 +359,38 @@ func TryProcessEvmERC20Transfer(chainNetwork string, contract common.Address, to } }() - 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: - 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) + token, err := data.GetEnabledChainTokenByContract(chainNetwork, contract.Hex()) + if err != nil { + log.Sugar.Warnf("[%s-WS] chain_tokens lookup err=%v contract=%s", net, err, contract.Hex()) + return + } + if token == nil || token.ID == 0 { + log.Sugar.Debugf("[%s-WS] skip unconfigured contract %s", net, contract.Hex()) + return + } + tokenSym := strings.ToUpper(strings.TrimSpace(token.Symbol)) + if tokenSym == "" { + log.Sugar.Warnf("[%s-WS] chain_token id=%d has empty symbol", net, token.ID) + return + } + decimals := token.Decimals + if decimals <= 0 { + decimals = 6 + } + walletAddr := strings.ToLower(toAddr.Hex()) if rawValue == nil || rawValue.Sign() <= 0 { log.Sugar.Infof("[%s-%s][%s] skip non-positive or nil amount", net, tokenSym, walletAddr) return } + divisor := decimal.New(1, int32(decimals)) decimalQuant := decimal.NewFromBigInt(rawValue, 0) - amount := math.MustParsePrecFloat64(decimalQuant.Div(decimal.NewFromInt(1_000_000)).InexactFloat64(), 2) + amount := math.MustParsePrecFloat64(decimalQuant.Div(divisor).InexactFloat64(), 2) + if token.MinAmount > 0 && amount < token.MinAmount { + log.Sugar.Debugf("[%s-%s][%s] skip amount %.2f below min_amount %.2f", net, tokenSym, walletAddr, amount, token.MinAmount) + return + } if amount <= 0 { log.Sugar.Warnf("[%s-%s][%s] skip non-positive amount %.2f", net, tokenSym, walletAddr, amount) return @@ -249,6 +421,11 @@ func TryProcessEvmERC20Transfer(chainNetwork string, contract common.Address, to log.Sugar.Warnf("[%s-%s][%s] skip trade_id=%s token mismatch order=%s", net, tokenSym, walletAddr, tradeID, order.Token) return } + if blockTsMs > 0 && blockTsMs < order.CreatedAt.TimestampMilli() { + log.Sugar.Warnf("[%s-%s][%s] skip tx %s because block_time_ms=%d is before order created_ms=%d", + net, tokenSym, walletAddr, txHash, blockTsMs, order.CreatedAt.TimestampMilli()) + return + } req := &request.OrderProcessingRequest{ ReceiveAddress: walletAddr, diff --git a/src/task/listen.go b/src/task/listen.go index 36b4a05..e99b99f 100644 --- a/src/task/listen.go +++ b/src/task/listen.go @@ -13,11 +13,26 @@ func Start() { go StartBscWebSocketListener() go StartPolygonWebSocketListener() go StartPlasmaWebSocketListener() - go StartTronBlockScannerListener() c := cron.New() + // TRC20 polling + _, 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 polling + _, err = c.AddJob("@every 5s", ListenSolJob{}) + if err != nil { + log.Sugar.Errorf("[task] Failed to add ListenSolJob: %v", err) + return + } + + log.Sugar.Info("[task] ListenSolJob scheduled successfully (@every 5s)") // RPC node health checks - _, err := c.AddJob("@every 30s", RpcHealthJob{}) + _, err = c.AddJob("@every 30s", RpcHealthJob{}) if err != nil { log.Sugar.Errorf("[task] Failed to add RpcHealthJob: %v", err) return diff --git a/src/task/listen_trc20_block.go b/src/task/listen_trc20_block.go deleted file mode 100644 index dea5509..0000000 --- a/src/task/listen_trc20_block.go +++ /dev/null @@ -1,378 +0,0 @@ -package task - -import ( - "bytes" - "encoding/hex" - "encoding/json" - "fmt" - "io" - "math/big" - "net/http" - "strings" - "time" - - tron "github.com/GMWalletApp/epusdt/crypto" - "github.com/GMWalletApp/epusdt/model/service" - "github.com/GMWalletApp/epusdt/util/log" -) - -const ( - // USDT 合约地址 (TRC20 主网) - USDTContractHex = "41a614f803b6fd780986a42c78ec9c7f77e6ded13c" // Base58: TR7NHqjeKQxGTCi8q8ZY4pL8otSzgjLj6t - // transfer(address,uint256) 方法签名前4字节 - TransferMethodID = "a9059cbb" - - PollInterval = 3 * time.Second - RequestTimeout = 10 * time.Second -) - -func HexToTronAddress(hexAddr string) (string, error) { - hexAddr = strings.TrimPrefix(hexAddr, "0x") - hexAddr = strings.TrimPrefix(hexAddr, "0X") - - raw, err := hex.DecodeString(hexAddr) - if err != nil { - return "", err - } - - // 确保有 0x41 前缀（TRON 主网地址前缀） - if len(raw) == 20 { - raw = append([]byte{0x41}, raw...) - } - if len(raw) != 21 { - return "", fmt.Errorf("地址长度非法: %d bytes", len(raw)) - } - - return tron.EncodeCheck(raw), nil -} - -type BlockHeader struct { - RawData struct { - Number int64 `json:"number"` - Timestamp int64 `json:"timestamp"` - WitnessAddress string `json:"witness_address"` - ParentHash string `json:"parentHash"` - Version int `json:"version"` - } `json:"raw_data"` -} - -type TriggerSmartContractValue struct { - OwnerAddress string `json:"owner_address"` - ContractAddress string `json:"contract_address"` - Data string `json:"data"` - CallValue int64 `json:"call_value"` -} - -type ContractParam struct { - TypeURL string `json:"type_url"` - Value json.RawMessage `json:"value"` -} - -type Transaction struct { - TxID string `json:"txID"` - RawData struct { - Contract []struct { - Type string `json:"type"` - Parameter ContractParam `json:"parameter"` - } `json:"contract"` - Timestamp int64 `json:"timestamp"` - FeeLimit int64 `json:"fee_limit"` - } `json:"raw_data"` - Ret []struct { - ContractRet string `json:"contractRet"` - } `json:"ret"` -} - -type Block struct { - BlockID string `json:"blockID"` - BlockHeader BlockHeader `json:"block_header"` - Transactions []Transaction `json:"transactions"` -} - -type USDTTransfer struct { - TxID string - From string - To string - Raw *big.Int // 原始数值（6 decimals） - Status string -} - -type TRXTransfer struct { - TxID string - From string - To string - RawSun int64 // 单位: SUN - Status string -} - -type TransferContractValue struct { - OwnerAddress string `json:"owner_address"` - ToAddress string `json:"to_address"` - Amount int64 `json:"amount"` // 单位: SUN -} - -var httpClient = &http.Client{Timeout: RequestTimeout} - -func doPost(url string, apiKey string, body interface{}) ([]byte, error) { - data, _ := json.Marshal(body) - req, err := http.NewRequest("POST", url, bytes.NewReader(data)) - if err != nil { - return nil, err - } - req.Header.Set("Content-Type", "application/json") - req.Header.Set("Accept", "application/json") - if apiKey = strings.TrimSpace(apiKey); apiKey != "" { - req.Header.Set("TRON-PRO-API-KEY", apiKey) - } - - resp, err := httpClient.Do(req) - if err != nil { - return nil, err - } - defer resp.Body.Close() - - b, _ := io.ReadAll(resp.Body) - if resp.StatusCode != http.StatusOK { - return nil, fmt.Errorf("HTTP %d: %s", resp.StatusCode, string(b)) - } - return b, nil -} - -func GetNowBlock(baseURL string, apiKey string) (*Block, error) { - b, err := doPost(baseURL+"/wallet/getnowblock", apiKey, map[string]interface{}{}) - if err != nil { - return nil, err - } - var block Block - return &block, json.Unmarshal(b, &block) -} - -func GetBlockByNum(baseURL string, apiKey string, num int64) (*Block, error) { - b, err := doPost(baseURL+"/wallet/getblockbynum", apiKey, map[string]interface{}{"num": num}) - if err != nil { - return nil, err - } - var block Block - return &block, json.Unmarshal(b, &block) -} - -func parseUSDTTransfer(tx Transaction) *USDTTransfer { - if len(tx.RawData.Contract) == 0 { - return nil - } - c := tx.RawData.Contract[0] - if c.Type != "TriggerSmartContract" { - return nil - } - - var val TriggerSmartContractValue - if err := json.Unmarshal(c.Parameter.Value, &val); err != nil { - return nil - } - - // 检查是否是 USDT 合约 - contractHex := strings.ToLower(strings.TrimPrefix(val.ContractAddress, "0x")) - if contractHex != USDTContractHex { - return nil - } - - // 解析 data 字段 - // 格式: [4字节方法ID][32字节 to 地址][32字节 amount] - data := strings.TrimPrefix(strings.ToLower(val.Data), "0x") - if len(data) < 8+64+64 { - return nil - } - if data[:8] != TransferMethodID { - return nil - } - - // to 地址：后 40 个十六进制字符（20 字节） - toHex := data[8+24 : 8+64] // 跳过前12字节填充，取后20字节 - toAddr, err := HexToTronAddress(toHex) - if err != nil { - return nil - } - - // amount：后 32 字节大整数 - amountHex := data[8+64 : 8+64+64] - amountBytes, err := hex.DecodeString(amountHex) - if err != nil { - return nil - } - amountBig := new(big.Int).SetBytes(amountBytes) - - // from 地址 - fromAddr, err := HexToTronAddress(val.OwnerAddress) - if err != nil { - fromAddr = val.OwnerAddress - } - - // 交易状态 - status := "SUCCESS" - if len(tx.Ret) > 0 && tx.Ret[0].ContractRet != "" && tx.Ret[0].ContractRet != "SUCCESS" { - status = tx.Ret[0].ContractRet - } - - return &USDTTransfer{ - TxID: tx.TxID, - From: fromAddr, - To: toAddr, - Raw: amountBig, - Status: status, - } -} - -func parseTRXTransfer(tx Transaction) *TRXTransfer { - if len(tx.RawData.Contract) == 0 { - return nil - } - c := tx.RawData.Contract[0] - if c.Type != "TransferContract" { - return nil - } - - var val TransferContractValue - if err := json.Unmarshal(c.Parameter.Value, &val); err != nil { - return nil - } - if val.Amount <= 0 { - return nil - } - - fromAddr, err := HexToTronAddress(val.OwnerAddress) - if err != nil { - fromAddr = val.OwnerAddress - } - toAddr, err := HexToTronAddress(val.ToAddress) - if err != nil { - toAddr = val.ToAddress - } - - status := "SUCCESS" - if len(tx.Ret) > 0 && tx.Ret[0].ContractRet != "" && tx.Ret[0].ContractRet != "SUCCESS" { - status = tx.Ret[0].ContractRet - } - - return &TRXTransfer{ - TxID: tx.TxID, - From: fromAddr, - To: toAddr, - RawSun: val.Amount, - Status: status, - } -} - -func processBlock(block *Block) { - blockTsMs := block.BlockHeader.RawData.Timestamp - for _, tx := range block.Transactions { - if t := parseUSDTTransfer(tx); t != nil { - if t.Status != "SUCCESS" { - continue - } - service.TryProcessTronTRC20Transfer(t.To, t.Raw, t.TxID, blockTsMs) - continue - } - if t := parseTRXTransfer(tx); t != nil { - if t.Status != "SUCCESS" { - continue - } - service.TryProcessTronTRXTransfer(t.To, t.RawSun, t.TxID, blockTsMs) - } - } -} - -type Scanner struct { - baseURL string - apiKey string - lastBlock int64 - // 统计 - totalBlocks int64 - totalUSDTTxs int64 - totalTRXTxs int64 -} - -func NewScanner() *Scanner { - return &Scanner{} -} - -func (s *Scanner) Init() error { - baseURL, apiKey, err := service.ResolveTronNode() - if err != nil { - return fmt.Errorf("resolve tron node: %w", err) - } - s.baseURL = baseURL - s.apiKey = apiKey - - log.Sugar.Infof("[TRON-BLOCK] node=%s", s.baseURL) - block, err := GetNowBlock(s.baseURL, s.apiKey) - if err != nil { - return fmt.Errorf("获取初始块失败: %w", err) - } - s.lastBlock = block.BlockHeader.RawData.Number - log.Sugar.Infof("[TRON-BLOCK] start block=%d", s.lastBlock) - return nil -} - -func (s *Scanner) Run() { - log.Sugar.Info("[TRON-BLOCK] start scanning (USDT TRC20 + TRX)") - ticker := time.NewTicker(PollInterval) - defer ticker.Stop() - - statTicker := time.NewTicker(60 * time.Second) - defer statTicker.Stop() - - for { - select { - case <-statTicker.C: - log.Sugar.Infof("[TRON-BLOCK] stats blocks=%d usdt=%d trx=%d", s.totalBlocks, s.totalUSDTTxs, s.totalTRXTxs) - case <-ticker.C: - s.poll() - } - } -} - -func (s *Scanner) poll() { - latest, err := GetNowBlock(s.baseURL, s.apiKey) - if err != nil { - log.Sugar.Warnf("[TRON-BLOCK] get latest block: %v", err) - return - } - latestNum := latest.BlockHeader.RawData.Number - if latestNum <= s.lastBlock { - return - } - - for num := s.lastBlock + 1; num <= latestNum; num++ { - var block *Block - if num == latestNum { - block = latest - } else { - block, err = GetBlockByNum(s.baseURL, s.apiKey, num) - if err != nil { - log.Sugar.Warnf("[TRON-BLOCK] get block %d: %v", num, err) - continue - } - time.Sleep(200 * time.Millisecond) - } - processBlock(block) - s.lastBlock = num - s.totalBlocks++ - - for _, tx := range block.Transactions { - if parseUSDTTransfer(tx) != nil { - s.totalUSDTTxs++ - } else if parseTRXTransfer(tx) != nil { - s.totalTRXTxs++ - } - } - } -} - -func StartTronBlockScannerListener() { - scanner := NewScanner() - if err := scanner.Init(); err != nil { - log.Sugar.Errorf("[TRON-BLOCK] init: %v", err) - return - } - scanner.Run() -} diff --git a/src/task/listen_trc20_job.go b/src/task/listen_trc20_job.go new file mode 100644 index 0000000..eb8b832 --- /dev/null +++ b/src/task/listen_trc20_job.go @@ -0,0 +1,43 @@ +package task + +import ( + "sync" + + "github.com/GMWalletApp/epusdt/model/data" + "github.com/GMWalletApp/epusdt/model/mdb" + "github.com/GMWalletApp/epusdt/model/service" + "github.com/GMWalletApp/epusdt/util/log" +) + +type ListenTrc20Job struct{} + +var gListenTrc20JobLock sync.Mutex + +func (r ListenTrc20Job) Run() { + gListenTrc20JobLock.Lock() + defer gListenTrc20JobLock.Unlock() + log.Sugar.Debug("[ListenTrc20Job] Job triggered") + if !data.IsChainEnabled(mdb.NetworkTron) { + log.Sugar.Debug("[ListenTrc20Job] chain disabled, skipping") + return + } + 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") +}