diff --git a/src/model/service/task_service.go b/src/model/service/task_service.go index 85498bc..52ebc1b 100644 --- a/src/model/service/task_service.go +++ b/src/model/service/task_service.go @@ -1,32 +1,23 @@ 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 { @@ -42,296 +33,128 @@ func resolveTronNode() (string, string, error) { return rpcURL, node.ApiKey, nil } -func Trc20CallBack(address string, wg *sync.WaitGroup) { - defer wg.Done() +func ResolveTronNode() (string, string, error) { + return resolveTronNode() +} + +func TryProcessTronTRC20Transfer(toAddr string, rawValue *big.Int, txHash string, blockTsMs int64) { defer func() { if err := recover(); err != nil { - log.Sugar.Error(err) + log.Sugar.Errorf("[TRC20][%s] TryProcessTronTRC20Transfer panic: %v", toAddr, err) } }() - var innerWg sync.WaitGroup - innerWg.Add(2) - go checkTrxTransfers(address, &innerWg) - go checkTrc20Transfers(address, &innerWg) - innerWg.Wait() + 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) } -func checkTrxTransfers(address string, wg *sync.WaitGroup) { - defer wg.Done() +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) + } + }() - // 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") + 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.Errorf("[TRX][%s] load chain_tokens err=%v", address, err) + log.Sugar.Warnf("[TRX][%s] lock lookup: %v", addr, err) return } - if trxCfg == nil || trxCfg.ID == 0 { - log.Sugar.Debugf("[TRX][%s] native TRX disabled in chain_tokens, skipping", address) + if tradeID == "" { + log.Sugar.Debugf("[TRX][%s] skip unmatched tx hash=%s amount=%.2f", addr, txHash, amount) 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() + order, err := data.GetOrderInfoByTradeId(tradeID) if err != nil { - log.Sugar.Errorf("[TRX][%s] resolve rpc_nodes err=%v", address, err) + 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 } - 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) + req := &request.OrderProcessingRequest{ + ReceiveAddress: addr, + Token: "TRX", + Network: mdb.NetworkTron, + TradeId: tradeID, + Amount: amount, + BlockTransactionId: txHash, + } + err = OrderProcessing(req) 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()) + 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 + } + log.Sugar.Errorf("[TRX][%s] OrderProcessing trade_id=%s hash=%s: %v", addr, tradeID, txHash, err) 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 - } - if transfer.Get("ret.0.contractRet").String() != "SUCCESS" { - continue - } - - 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 - } - - 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) - } + sendPaymentNotification(order) + log.Sugar.Infof("[TRX][%s] payment processed trade_id=%s hash=%s", addr, tradeID, txHash) } func evmChainLogLabel(chainNetwork string) string { @@ -349,9 +172,6 @@ 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 { @@ -359,38 +179,46 @@ func TryProcessEvmERC20Transfer(chainNetwork string, contract common.Address, to } }() - 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()) + 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 } - 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 - } + 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("[%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(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 - } + amount := math.MustParsePrecFloat64(decimalQuant.Div(decimal.NewFromInt(1_000_000)).InexactFloat64(), 2) if amount <= 0 { log.Sugar.Warnf("[%s-%s][%s] skip non-positive amount %.2f", net, tokenSym, walletAddr, amount) return @@ -421,11 +249,6 @@ 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 e99b99f..36b4a05 100644 --- a/src/task/listen.go +++ b/src/task/listen.go @@ -13,26 +13,11 @@ 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 new file mode 100644 index 0000000..dea5509 --- /dev/null +++ b/src/task/listen_trc20_block.go @@ -0,0 +1,378 @@ +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 deleted file mode 100644 index eb8b832..0000000 --- a/src/task/listen_trc20_job.go +++ /dev/null @@ -1,43 +0,0 @@ -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") -}