8 Commits

Author SHA1 Message Date
Ginta 06933451c5 actions 2026-04-28 16:28:01 +08:00
line-6000 d17d212c24 change tron scan logic 2026-04-28 15:28:00 +08:00
Ginta 51a2accd9f 🐛 fix link 2026-04-27 15:57:26 +08:00
Ginta 96ea7481bc Revert "Revert "new trc (#54)""
This reverts commit 74c3a3a08c.
2026-04-27 15:52:50 +08:00
Ginta de9c45a944 static 2026-04-27 15:52:12 +08:00
Ginta 9e885d8b2f order notify 2026-04-27 15:43:35 +08:00
Ginta 74c3a3a08c Revert "new trc (#54)"
This reverts commit 0a19854602.
2026-04-27 00:04:05 +08:00
Ginta 0a19854602 new trc (#54)
* new trct
2026-04-26 23:32:43 +08:00
8 changed files with 771 additions and 361 deletions
+1 -1
View File
@@ -33,7 +33,7 @@ jobs:
images: ${{ env.DOCKERHUB_NAMESPACE }}/${{ env.IMAGE_NAME }}
tags: |
type=raw,value=alpine
type=raw,value=latest,enable={{is_default_branch}}
type=raw,value=latest
type=ref,event=branch,suffix=-alpine
type=ref,event=tag
type=sha,prefix=sha-,suffix=-alpine
+91
View File
@@ -3,6 +3,7 @@ package config
import (
"errors"
"fmt"
"io"
"log"
"net/url"
"os"
@@ -62,6 +63,9 @@ func Init() {
LogLevel = normalizeLogLevel(viper.GetString("log_level"))
StaticPath = normalizeStaticURLPath(viper.GetString("static_path"))
StaticFilePath = filepath.Join(configRootPath, strings.TrimPrefix(StaticPath, "/"))
if err = ensureConfiguredStaticFiles(); err != nil {
panic(err)
}
RuntimePath = resolvePathFromBase(configRootPath, viper.GetString("runtime_root_path"), filepath.Join(configRootPath, "runtime"))
LogSavePath = resolvePathFromBase(RuntimePath, viper.GetString("log_save_path"), filepath.Join(RuntimePath, "logs"))
mustMkdir(RuntimePath)
@@ -82,6 +86,93 @@ func mustMkdir(path string) {
}
}
func ensureConfiguredStaticFiles() error {
if strings.TrimSpace(StaticFilePath) == "" {
return nil
}
exePath, err := os.Executable()
if err != nil {
return err
}
exePath, err = filepath.EvalSymlinks(exePath)
if err != nil {
return err
}
srcDir := filepath.Join(filepath.Dir(exePath), "static")
srcInfo, err := os.Stat(srcDir)
if err != nil {
if errors.Is(err, os.ErrNotExist) {
return nil
}
return err
}
if !srcInfo.IsDir() {
return nil
}
srcAbs, err := filepath.Abs(srcDir)
if err != nil {
return err
}
dstAbs, err := filepath.Abs(StaticFilePath)
if err != nil {
return err
}
if srcAbs == dstAbs {
return nil
}
return copyMissingStaticFiles(srcAbs, dstAbs)
}
func copyMissingStaticFiles(srcDir, dstDir string) error {
return filepath.WalkDir(srcDir, func(path string, d os.DirEntry, err error) error {
if err != nil {
return err
}
rel, err := filepath.Rel(srcDir, path)
if err != nil {
return err
}
dstPath := filepath.Join(dstDir, rel)
if d.IsDir() {
return os.MkdirAll(dstPath, 0o755)
}
if _, err = os.Stat(dstPath); err == nil {
return nil
} else if !errors.Is(err, os.ErrNotExist) {
return err
}
if err = os.MkdirAll(filepath.Dir(dstPath), 0o755); err != nil {
return err
}
return copyFile(path, dstPath)
})
}
func copyFile(srcPath, dstPath string) error {
in, err := os.Open(srcPath)
if err != nil {
return err
}
defer in.Close()
out, err := os.OpenFile(dstPath, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o644)
if err != nil {
if errors.Is(err, os.ErrExist) {
return nil
}
return err
}
if _, err = io.Copy(out, in); err != nil {
_ = out.Close()
return err
}
return out.Close()
}
func normalizeLogLevel(level string) string {
switch strings.ToLower(strings.TrimSpace(level)) {
case "debug", "info", "warn", "error":
+54
View File
@@ -177,6 +177,60 @@ func TestResolveConfigFilePathPrefersExplicitOverEnv(t *testing.T) {
}
}
func TestCopyMissingStaticFilesCopiesAssetsWithoutOverwriting(t *testing.T) {
root := t.TempDir()
src := filepath.Join(root, "app-static")
dst := filepath.Join(root, "data-static")
if err := os.MkdirAll(filepath.Join(src, "images"), 0o755); err != nil {
t.Fatalf("mkdir source: %v", err)
}
if err := os.WriteFile(filepath.Join(src, "index.html"), []byte("source-index"), 0o644); err != nil {
t.Fatalf("write source index: %v", err)
}
if err := os.WriteFile(filepath.Join(src, "payment.js"), []byte("source-payment"), 0o644); err != nil {
t.Fatalf("write source payment: %v", err)
}
if err := os.WriteFile(filepath.Join(src, "images", "logo.png"), []byte("source-logo"), 0o644); err != nil {
t.Fatalf("write source logo: %v", err)
}
if err := os.MkdirAll(dst, 0o755); err != nil {
t.Fatalf("mkdir destination: %v", err)
}
if err := os.WriteFile(filepath.Join(dst, "index.html"), []byte("custom-index"), 0o644); err != nil {
t.Fatalf("write existing destination index: %v", err)
}
if err := copyMissingStaticFiles(src, dst); err != nil {
t.Fatalf("copy missing static files: %v", err)
}
index, err := os.ReadFile(filepath.Join(dst, "index.html"))
if err != nil {
t.Fatalf("read destination index: %v", err)
}
if string(index) != "custom-index" {
t.Fatalf("existing index was overwritten: %q", string(index))
}
payment, err := os.ReadFile(filepath.Join(dst, "payment.js"))
if err != nil {
t.Fatalf("read copied payment: %v", err)
}
if string(payment) != "source-payment" {
t.Fatalf("payment.js = %q, want source-payment", string(payment))
}
logo, err := os.ReadFile(filepath.Join(dst, "images", "logo.png"))
if err != nil {
t.Fatalf("read copied nested asset: %v", err)
}
if string(logo) != "source-logo" {
t.Fatalf("logo.png = %q, want source-logo", string(logo))
}
}
func TestGetUsdtRatePrefersPositiveAdminOverride(t *testing.T) {
viper.Reset()
t.Cleanup(viper.Reset)
+164 -309
View File
@@ -1,32 +1,22 @@
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 +32,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 +171,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 +178,67 @@ 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)
chainTokens, err := data.ListChainTokens(chainNetwork)
if err != nil {
log.Sugar.Warnf("[%s-%s] load chain tokens: %v", net, tokenSym, err)
return
}
var tokenConfig *mdb.ChainToken
for _, t := range chainTokens {
if strings.EqualFold(t.Symbol, tokenSym) {
tokenConfig = &t
break
}
}
if tokenConfig == nil || !tokenConfig.Enabled {
log.Sugar.Warnf("[%s-%s] token not enabled or configured in chain_tokens", net, tokenSym)
return
}
pow := decimal.New(1, int32(tokenConfig.Decimals))
log.Sugar.Warnf("tokenConfig.Decimals %d pow %s", tokenConfig.Decimals, pow.String())
decimalQuant := decimal.NewFromBigInt(rawValue, 0)
amount := math.MustParsePrecFloat64(decimalQuant.Div(pow).InexactFloat64(), 2)
if amount <= 0 {
log.Sugar.Warnf("[%s-%s][%s] skip non-positive amount %.2f", net, tokenSym, walletAddr, amount)
return
@@ -421,11 +269,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,
@@ -450,6 +293,18 @@ func TryProcessEvmERC20Transfer(chainNetwork string, contract common.Address, to
}
func sendPaymentNotification(order *mdb.Orders) {
if order == nil {
return
}
if strings.TrimSpace(order.TradeId) != "" {
latest, err := data.GetOrderInfoByTradeId(order.TradeId)
if err != nil {
log.Sugar.Warnf("[notify] reload order failed trade_id=%s err=%v", order.TradeId, err)
} else if latest != nil && latest.TradeId != "" {
order = latest
}
}
msg := fmt.Sprintf(
"🎉 <b>收款成功通知</b>\n\n"+
"💰 <b>金额信息</b>\n"+
@@ -472,7 +327,7 @@ func sendPaymentNotification(order *mdb.Orders) {
networkDisplay(order.Network),
order.ReceiveAddress,
order.CreatedAt.ToDateTimeString(),
carbon.Now().ToDateTimeString(),
order.UpdatedAt.ToDateTimeString(),
)
notify.Dispatch(mdb.NotifyEventPaySuccess, msg)
}
+79
View File
@@ -0,0 +1,79 @@
package service
import (
"strings"
"testing"
"time"
"github.com/GMWalletApp/epusdt/internal/testutil"
"github.com/GMWalletApp/epusdt/model/dao"
"github.com/GMWalletApp/epusdt/model/mdb"
"github.com/GMWalletApp/epusdt/notify"
)
func TestSendPaymentNotificationUsesLatestOrderUpdatedAt(t *testing.T) {
cleanup := testutil.SetupTestDatabases(t)
defer cleanup()
const channelType = "test-pay-success-time"
got := make(chan string, 1)
notify.RegisterSender(channelType, func(config, text string) error {
got <- text
return nil
})
if err := dao.Mdb.Create(&mdb.NotificationChannel{
Type: channelType,
Name: "test",
Config: "{}",
Events: `{"pay_success":true}`,
Enabled: true,
}).Error; err != nil {
t.Fatalf("seed notification channel: %v", err)
}
order := &mdb.Orders{
TradeId: "T202604270001",
OrderId: "ORD202604270001",
Amount: 100,
Currency: "cny",
ActualAmount: 14.28,
Token: "USDT",
Network: mdb.NetworkTron,
ReceiveAddress: "TTestAddress",
Status: mdb.StatusWaitPay,
}
if err := dao.Mdb.Create(order).Error; err != nil {
t.Fatalf("seed order: %v", err)
}
const createdAt = "2026-04-27 09:00:00"
const staleUpdatedAt = "2026-04-27 09:01:00"
const paidAt = "2026-04-27 10:20:30"
if err := dao.Mdb.Exec("UPDATE orders SET created_at = ?, updated_at = ? WHERE trade_id = ?", createdAt, staleUpdatedAt, order.TradeId).Error; err != nil {
t.Fatalf("set initial timestamps: %v", err)
}
var staleOrderModel mdb.Orders
if err := dao.Mdb.Where("trade_id = ?", order.TradeId).Take(&staleOrderModel).Error; err != nil {
t.Fatalf("load stale order: %v", err)
}
if err := dao.Mdb.Exec("UPDATE orders SET status = ?, updated_at = ? WHERE trade_id = ?", mdb.StatusPaySuccess, paidAt, order.TradeId).Error; err != nil {
t.Fatalf("set paid timestamp: %v", err)
}
sendPaymentNotification(&staleOrderModel)
select {
case text := <-got:
if !strings.Contains(text, "支付时间:"+paidAt) {
t.Fatalf("notification payment time = %q, want %s", text, paidAt)
}
if strings.Contains(text, "支付时间:"+staleUpdatedAt) {
t.Fatalf("notification used stale payment time: %q", text)
}
case <-time.After(2 * time.Second):
t.Fatal("timed out waiting for notification")
}
}
+4 -8
View File
@@ -13,24 +13,19 @@ 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{})
_, 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{})
if err != nil {
@@ -38,6 +33,7 @@ func Start() {
return
}
log.Sugar.Info("[task] RpcHealthJob scheduled successfully (@every 30s)")
c.Start()
log.Sugar.Info("[task] Task scheduler started")
}
+378
View File
@@ -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()
}
-43
View File
@@ -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")
}