m2pool_payment/internal/blockchain/eth/eth_prv.go

package eth

import (
	"context"
	"database/sql"
	"fmt"
	"log"
	"m2pool-payment/internal/constant"
	message "m2pool-payment/internal/msg"
	"m2pool-payment/internal/utils"
	"math/big"
	"os"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/ethereum/go-ethereum"
	"github.com/ethereum/go-ethereum/accounts/abi"
	"github.com/ethereum/go-ethereum/common"
	"github.com/ethereum/go-ethereum/core/types"
	"github.com/ethereum/go-ethereum/crypto"
)

func init_USDT() *USDT {
	// 构造USDT合约相关
	usdt := &USDT{}
	usdt.Address = common.HexToAddress(constant.ETH_ERC20_USDT_CONTRACT_ADDRESS)                          // 解析合约地址
	usdt.ABI = func() abi.ABI { a, _ := abi.JSON(strings.NewReader(constant.ETH_ERC20_ABI)); return a }() // 解析合约ABI
	usdt.TransferSig = crypto.Keccak256Hash([]byte("Transfer(address,address,uint256)"))                  // 解析合约transfer函数签名
	usdt.LogsChan = make(chan types.Log, 1000)                                                            // 初始化合约日志通道
	usdt.ListeningAddresses = make(map[string]any)
	return usdt
}

func (e *ETHNode) initTables() error {
	sql_script_path := "../public/eth.sql"
	sql_byte, err := os.ReadFile(sql_script_path)
	if err != nil {
		return fmt.Errorf("open msg-sql file error: %v", err)
	}

	err = e.SqliteDB.CreateTable(string(sql_byte))
	if err != nil {
		return fmt.Errorf("exec eth-sql error: %v", err)
	}

	return nil
}

func (e *ETHNode) updateNetInfo(height uint64) {
	// 创建 WaitGroup
	var wg sync.WaitGroup

	// 用于保存每个方法的结果
	var gasLimit uint64
	var suggestGasPrice *big.Int
	var maxFeePerGas, maxPriorityFeePerGas *big.Int
	var gasLimitErr, suggestGasPriceErr, eip1559GasFeesErr error

	// 启动协程并等待所有结果
	wg.Add(3)

	// 获取 Gas Limit
	go func() {
		defer wg.Done()
		gasLimit, gasLimitErr = e.getGasLimit()
		if gasLimitErr != nil {
			log.Printf("Failed to get gas limit: %v", gasLimitErr)
		} else {
			// log.Printf("Gas Limit: %d", gasLimit)
		}
	}()

	// 获取建议 Gas Price
	go func() {
		defer wg.Done()
		suggestGasPrice, suggestGasPriceErr = e.getSuggestGasPrice()
		if suggestGasPriceErr != nil {
			log.Printf("Failed to get suggested gas price: %v", suggestGasPriceErr)
		} else {
			// log.Printf("Suggested Gas Price: %v Gwei", new(big.Int).Div(suggestGasPrice, big.NewInt(1000000000)))
		}
	}()

	// 获取 EIP-1559 Gas Fees
	go func() {
		defer wg.Done()
		maxFeePerGas, maxPriorityFeePerGas, eip1559GasFeesErr = e.getEIP1559GasFees()
		if eip1559GasFeesErr != nil {
			log.Printf("Failed to get EIP-1559 gas fees: %v", eip1559GasFeesErr)
		} else {
			// log.Printf("EIP-1559 Gas Fees: MaxFeePerGas: %v Gwei, MaxPriorityFeePerGas: %v Gwei",
			// 	new(big.Int).Div(maxFeePerGas, big.NewInt(1000000000)),
			// 	new(big.Int).Div(maxPriorityFeePerGas, big.NewInt(1000000000)))
		}
	}()

	// 等待所有协程完成
	wg.Wait()

	// 检查是否有任何错误
	if gasLimitErr != nil || suggestGasPriceErr != nil || eip1559GasFeesErr != nil {
		log.Println("One or more methods failed. Not updating RealData.")
		return
	}

	// 更新 RealData
	e.NetInfo.mu.Lock()
	defer e.NetInfo.mu.Unlock()
	e.NetInfo = &NetInfo{
		Height:    height,
		GasLimit:  gasLimit,
		GasTipCap: maxPriorityFeePerGas,
		GasFeeCap: maxFeePerGas,
		GasPrice:  suggestGasPrice,
	}
	// log.Println("✅ RealData updated successfully.")
}

func (e *ETHNode) getGasLimit() (uint64, error) {
	// 对于ERC20转账，使用固定的gas limit
	// 通常ERC20 transfer需要约65,000-100,000 gas
	// 这里设置为80,000，足够覆盖大部分情况
	return 80000, nil
}

func (e *ETHNode) getSuggestGasPrice() (*big.Int, error) {
	ctx := context.Background()
	gasPrice, err := e.RpcClient.SuggestGasPrice(ctx)
	if err != nil {
		return nil, fmt.Errorf("get suggest-gasprice error:%v", err)
	}

	// 设置gas price上限，避免在网络拥堵时费用过高
	// 这里设置为20 Gwei (2 * 10^9 wei)
	maxGasPrice := new(big.Int).SetUint64(2000000000) // 2 Gwei

	if gasPrice.Cmp(maxGasPrice) > 0 {
		log.Printf("⚠️ 建议gas price过高 (%v wei)，使用上限 2 Gwei", new(big.Int).Div(gasPrice, big.NewInt(1e18)))
		return maxGasPrice, nil
	}

	// log.Printf("✅ 使用建议gas price: %v wei", new(big.Int).Div(gasPrice, big.NewInt(1e18)))
	return gasPrice, nil
}

// getEIP1559GasFees 获取EIP-1559的gas费用参数
func (e *ETHNode) getEIP1559GasFees() (*big.Int, *big.Int, error) {
	ctx := context.Background()

	// 获取基础费用
	latestBlock, err := e.RpcClient.BlockByNumber(ctx, nil)
	if err != nil {
		return nil, nil, fmt.Errorf("failed to get latest block: %w", err)
	}

	baseFee := latestBlock.BaseFee()
	if baseFee == nil {
		return nil, nil, fmt.Errorf("base fee not available")
	}

	// 设置优先级费用（tip），这里设置为2 Gwei
	maxPriorityFeePerGas := new(big.Int).SetUint64(2000000) // 0.2 Gwei

	// 计算最大费用 = 基础费用 + 优先级费用
	maxFeePerGas := new(big.Int).Add(baseFee, maxPriorityFeePerGas)

	// 设置最大费用上限为2 Gwei
	maxFeeLimit := new(big.Int).SetUint64(2000000000) // 2 Gwei
	if maxFeePerGas.Cmp(maxFeeLimit) > 0 {
		log.Printf("⚠️ 计算的最大费用过高 (%v wei)，使用上限 2 Gwei", new(big.Int).Div(maxFeePerGas, big.NewInt(1e18)))
		maxFeePerGas = maxFeeLimit
	}

	// log.Printf("✅ EIP-1559 Gas费用: BaseFee=%v wei, MaxPriorityFee=%v wei, MaxFee=%v wei",
	// new(big.Int).Div(baseFee, big.NewInt(1e18)),
	// new(big.Int).Div(maxPriorityFeePerGas, big.NewInt(1e18)),
	// new(big.Int).Div(maxFeePerGas, big.NewInt(1e18)))

	return maxFeePerGas, maxPriorityFeePerGas, nil
}

func (e *ETHNode) getHeight() (uint64, error) {
	height, err := e.RpcClient.BlockNumber(e.Ctx)
	if err != nil {
		return 0, fmt.Errorf("[inital net info]get block height error: %v", err)
	}
	return height, nil
}

// Helper function to convert []string to []interface{}
func toInterfaceSlice(addresses []string) []interface{} {
	result := make([]interface{}, len(addresses))
	for i, v := range addresses {
		result[i] = v
	}
	return result
}

func (e *ETHNode) getAddressesPks(addresses []string) (map[string]string, error) {
	// If no addresses are provided, return an error
	if len(addresses) == 0 {
		return nil, fmt.Errorf("no addresses provided")
	}

	// Build a placeholder string for the IN clause
	placeholders := make([]string, len(addresses))
	for i := range addresses {
		placeholders[i] = "?"
	}
	inClause := strings.Join(placeholders, ", ")

	// Construct the SQL query
	str := fmt.Sprintf("SELECT address, private_key FROM eth_balance WHERE address IN (%s);", inClause)

	// Execute the query with the addresses
	rows, err := e.MysqlDB.Query(str, toInterfaceSlice(addresses)...)
	if err != nil {
		return nil, err
	}
	defer rows.Close() // Ensure the rows are closed after processing

	// Prepare a map to store the results
	result := make(map[string]string)

	// Iterate over the rows and populate the map
	var storedAddress, privateKey string
	for rows.Next() {
		err := rows.Scan(&storedAddress, &privateKey)
		if err != nil {
			return nil, err
		}

		// Add the address and private key to the map
		result[storedAddress] = privateKey
	}

	// Check if there were any errors during row iteration
	if err := rows.Err(); err != nil {
		return nil, err
	}

	return result, nil
}

func (e *ETHNode) getAddressPk(address string) (string, error) {
	str := "SELECT address, private_key FROM eth_balance WHERE address = ? LIMIT 1;"
	row := e.MysqlDB.QueryRow(str, address)
	// Variables to store the result
	var storedAddress, privateKey string

	// Scan the result into the variables
	err := row.Scan(&storedAddress, &privateKey)
	if err != nil {
		if err == sql.ErrNoRows {
			// Return an error if no rows are found
			return "", fmt.Errorf("no address found for %s", address)
		}
		// Return any other error
		return "", fmt.Errorf("error scanning row: %v", err)
	}

	// Return the private key
	return privateKey, nil
}

// 检查钱包余额是否足够
// symbol=ETH时，target_amount_eth传入实际gas费用+转账金额，target_amount_usdt传入0即可
// symbol=USDT时，target_amount_eth传入实际gas费用，target_amount_usdt传入转账金额+冻结金额
func (e *ETHNode) checkBalance(symbol, address string, target_amount_eth, target_amount_usdt *big.Int) (bool, error) {
	e.Wallets[address].mu.Lock()
	defer e.Wallets[address].mu.Unlock()
	if wallet, ok := e.Wallets[address]; ok {
		log.Printf("需要的USDT：%d，钱包的USDT：%d", target_amount_usdt, e.Wallets[address].usdt_balance.balance)
		switch symbol {
		case "ETH":
			// ETH余额不足支付转账金额+GAS费用
			if wallet.eth_balance.balance.Cmp(target_amount_eth) < 0 {
				return false, nil
			} else {
				return true, nil
			}
		case "USDT":
			// gas费不足或USDT余额不足
			if wallet.eth_balance.balance.Cmp(target_amount_eth) < 0 || wallet.usdt_balance.balance.Cmp(target_amount_usdt) < 0 {
				return false, nil
			} else {
				return true, nil
			}
		default:
			return false, fmt.Errorf("Symbol(%s) Error", symbol)
		}
	} else {
		return false, fmt.Errorf("Balance(%s) not found", address)
	}
}

func (e *ETHNode) loadWallets() error {
	// 使用 JOIN 查询 ETH_wallets, ETH_balances 和 USDT_balances，筛选 status 为 1 的钱包
	query := `
		SELECT 
			ew.address, ew.queue_id, ew.timestamp, ew.sign, ew.status,
			eb.used_gas AS eth_used_gas, eb.balance AS eth_balance, 
			eb.success_tx_hash AS eth_successed_tx_hash, eb.failed_tx_hash AS eth_failed_tx_hash,
			ub.freeze_num AS usdt_freeze_num, ub.balance AS usdt_balance, 
			ub.success_tx_hash AS usdt_successed_tx_hash, ub.failed_tx_hash AS usdt_failed_tx_hash
		FROM ETH_wallets ew
		LEFT JOIN ETH_balances eb ON ew.address = eb.address
		LEFT JOIN USDT_balances ub ON ew.address = ub.address
		WHERE ew.status = ?`

	// 执行查询
	rows, err := e.MysqlDB.Query(query, 1)
	if err != nil {
		return fmt.Errorf("failed to get wallets: %v", err)
	}
	defer rows.Close()

	// 解析字段为 big.Int 类型
	parseBigInt := func(val sql.NullString) *big.Int {
		if val.Valid && strings.TrimSpace(val.String) != "" {
			if bi, ok := new(big.Int).SetString(strings.TrimSpace(val.String), 10); ok {
				return bi
			}
		}
		log.Printf("Invalid or empty string for big.Int: %s", val.String) // 添加调试日志
		return big.NewInt(0)                                              // 如果无法转换，返回 0
	}

	// 解析字符串为交易哈希列表
	parseHashList := func(val sql.NullString) []string {
		if !val.Valid || val.String == "" {
			return []string{}
		}
		parts := strings.Split(val.String, ",")
		res := make([]string, 0, len(parts))
		for _, p := range parts {
			p = strings.TrimSpace(p)
			if p != "" {
				res = append(res, p)
			}
		}
		return res
	}

	// 用来存储钱包信息
	wallets := make(map[string]*Wallets)
	addresses := []string{}

	// 逐行处理查询结果
	for rows.Next() {
		var (
			addrStr, queueID, sign  sql.NullString
			timestamp               sql.NullInt64
			status                  sql.NullInt64
			ethUsedGas, ethBalance  sql.NullString
			ethSuccess, ethFailed   sql.NullString
			usdtFreeze, usdtBalance sql.NullString
			usdtSuccess, usdtFailed sql.NullString
		)

		// 扫描查询结果到变量
		if err := rows.Scan(
			&addrStr, &queueID, &timestamp, &sign, &status,
			&ethUsedGas, &ethBalance,
			&ethSuccess, &ethFailed,
			&usdtFreeze, &usdtBalance,
			&usdtSuccess, &usdtFailed,
		); err != nil {
			return fmt.Errorf("failed to scan row: %v", err)
		}

		// 如果地址字段为空，则跳过该行
		if !addrStr.Valid {
			continue
		}

		addr := strings.ToLower(addrStr.String)

		// 创建钱包对象并填充数据
		wallet := &Wallets{
			address: addr,
			queueId: queueID.String,
			sign:    sign.String,
			timestamp: func() uint64 {
				if timestamp.Valid {
					return uint64(timestamp.Int64)
				}
				return 0
			}(),
			status: func() int {
				if status.Valid {
					return int(status.Int64)
				}
				return 0
			}(),
		}

		// 解析 ETH 余额和信息
		ethBalanceStr := strings.TrimSpace(ethBalance.String)
		log.Printf("Parsed ETH balance: %s for address: %s", ethBalanceStr, addrStr.String) // 调试日志
		ethBalanceValue, err := strconv.ParseFloat(ethBalanceStr, 64)
		if err != nil {
			log.Printf("Failed to parse ETH balance: %v, address: %s", err, addrStr.String)
			ethBalanceValue = 0 // 给一个默认值
		}

		wallet.eth_balance = &eth_balance{
			symbol:   "ETH",
			used_gas: parseBigInt(ethUsedGas),
			// balance:           big.NewInt(int64(ethBalanceValue)), // 使用 big.Int 存储数值
			balance:           utils.Float64ToBigInt("ETH", ethBalanceValue),
			successed_tx_hash: parseHashList(ethSuccess),
			failed_tx_hash:    parseHashList(ethFailed),
		}

		// 解析 USDT 余额和信息
		usdtBalanceStr := strings.TrimSpace(usdtBalance.String)
		log.Printf("Parsed USDT balance: %s for address: %s", usdtBalanceStr, addrStr.String) // 调试日志
		usdtBalanceValue, err := strconv.ParseFloat(usdtBalanceStr, 64)
		if err != nil {
			log.Printf("Failed to parse USDT balance: %v, address: %s", err, addrStr.String)
			usdtBalanceValue = 0 // 给一个默认值
		}

		wallet.usdt_balance = &usdt_balance{
			symbol:            "USDT",
			freeze_num:        parseBigInt(usdtFreeze),
			balance:           utils.Float64ToBigInt("USDT", usdtBalanceValue), // 使用 big.Int 存储数值
			successed_tx_hash: parseHashList(usdtSuccess),
			failed_tx_hash:    parseHashList(usdtFailed),
		}

		// 添加钱包地址到列表并将钱包信息存储到 map 中
		addresses = append(addresses, addr)
		wallets[addr] = wallet
	}

	// 检查是否有查询错误
	if err := rows.Err(); err != nil {
		return fmt.Errorf("error occurred while iterating rows: %v", err)
	}

	// 打印已加载的地址
	log.Printf("ETH钱包加载成功：%v", addresses)

	// 如果有地址，获取钱包的私钥
	if len(addresses) > 0 {
		pks, err := e.getAddressesPks(addresses)
		if err != nil {
			return fmt.Errorf("initial balance private key error: %v", err)
		}
		e.mu.Lock()
		e.Wallets = wallets
		// 将私钥绑定到钱包对象中
		for address, pk := range pks {
			e.Wallets[address].pk = pk
		}
		e.mu.Unlock()
	}

	// 打印每个钱包的余额
	for addr, balance := range e.Wallets {
		log.Printf("钱包：%s, ETH余额: %s, USDT余额: %s", addr, balance.eth_balance.balance.String(), balance.usdt_balance.balance.String())
	}

	return nil
}

func (e *ETHNode) loadUnConfirmedTxs() error {
	// 查询语句
	query_str := "SELECT queue_id, tx_type, chain, symbol, from_addr, to_addr, tx_hash, height, amount, status FROM ETH_unconfirmed_tx WHERE status = ?"
	// 执行查询
	rows, err := e.MysqlDB.Query(query_str, 2)
	if err != nil {
		return fmt.Errorf("failed to query unconfirmed transactions: %v", err)
	}
	defer rows.Close() // 确保查询结束后关闭 rows

	// 遍历查询结果
	for rows.Next() {
		var queueId, chain, symbol, fromAddr, toAddr, txHash, amount string
		var txType, height, status int
		// 读取当前行数据
		if err := rows.Scan(&queueId, &txType, &chain, &symbol, &fromAddr, &toAddr, &txHash, &height, &amount, &status); err != nil {
			return fmt.Errorf("failed to scan row: %v", err)
		}

		// 将 amount 转换为 big.Int
		bigAmount := new(big.Int)
		bigAmountParsed, success := bigAmount.SetString(amount, 10)
		if !success {
			return fmt.Errorf("failed to convert amount '%s' to big.Int", amount)
		}

		// 锁定并填充数据
		e.UnConfirmedTxs.mu.Lock()
		e.UnConfirmedTxs.Transactions[queueId] = message.Transaction{
			QueueId: queueId,
			TxType:  txType,
			Chain:   chain,
			Symbol:  symbol,
			From:    fromAddr,
			To:      toAddr,
			TxHash:  txHash,
			Height:  uint64(height),
			Amount:  bigAmountParsed, // 使用转换后的 big.Int
			Status:  status,
		}
		e.UnConfirmedTxs.mu.Unlock()
	}

	// 检查是否存在查询错误
	if err := rows.Err(); err != nil {
		return fmt.Errorf("error during row iteration: %v", err)
	}

	return nil
}

// 获取交易nonce
func (e *ETHNode) getTransactionNonce(address string) (uint64, error) {
	nonce, err := e.RpcClient.PendingNonceAt(e.Ctx, common.HexToAddress(address))
	if err != nil {
		return 0, fmt.Errorf("failed to get nonce: %w", err)
	}
	return nonce, nil
}

// 构建交易
func (e *ETHNode) contractTx(symbol, from, to string, amount float64) (*types.Transaction, error) {
	nonce, err := e.getTransactionNonce(from)
	if err != nil {
		return nil, fmt.Errorf("failed to get nonce: %v", err)
	}
	return e.buildContractTxWithNonce(symbol, to, amount, nonce)
}

// 构建交易（指定 nonce）
func (e *ETHNode) buildContractTxWithNonce(symbol, to string, amount float64, nonce uint64) (*types.Transaction, error) {
	e.mu.Lock()
	maxFeePerGas, maxPriorityFeePerGas, gasLimit := e.NetInfo.GasFeeCap, e.NetInfo.GasTipCap, e.NetInfo.GasLimit
	if maxFeePerGas == nil || maxPriorityFeePerGas == nil || gasLimit == 0 {
		e.mu.Unlock()
		return nil, fmt.Errorf("chain data not initialized!")
	}
	netID := e.NetID
	e.mu.Unlock()

	addr := common.HexToAddress(to)
	eip1559Tx := &types.DynamicFeeTx{
		ChainID:   netID,
		Nonce:     nonce,
		GasTipCap: maxPriorityFeePerGas,
		GasFeeCap: maxFeePerGas,
		Gas:       gasLimit,
		To:        &addr,
		Data:      []byte{},
	}

	switch symbol {
	case "ETH":
		eip1559Tx.Value = utils.Float64ToBigIntETH(amount)
	case "USDT":
		eip1559Tx.Value = big.NewInt(0)
		eip1559Tx.To = &e.USDT.Address
		data, err := e.USDT.ABI.Pack("transfer", common.HexToAddress(to), utils.Float64ToBigIntUSDT(amount))
		if err != nil {
			return nil, fmt.Errorf("failed to pack transfer data: %v", err)
		}
		eip1559Tx.Data = data
	default:
		return nil, fmt.Errorf("ETH NetWork error symbol: %s", symbol)
	}

	return types.NewTx(eip1559Tx), nil
}

// 签名交易
func (e *ETHNode) signTx(tx *types.Transaction, from string) (*types.Transaction, error) {
	// originalKey := e.decodePrivatekey(from)
	// if originalKey == "" {
	// 	return nil, fmt.Errorf("failed to query private key for address: %s", from)
	// }
	e.mu.Lock()
	originalKey := e.Wallets[from].pk
	e.mu.Unlock()
	privateKey, err := crypto.HexToECDSA(originalKey)
	if err != nil {
		return nil, fmt.Errorf("failed to parse private key: %v", err)
	}

	signer := types.LatestSignerForChainID(e.NetID)
	return types.SignTx(tx, signer, privateKey)
}

// 发送交易
func (e *ETHNode) sendTransaction(tx *types.Transaction) (string, error) {
	// 发送交易
	err := e.RpcClient.SendTransaction(e.Ctx, tx)
	if err != nil {
		return "", fmt.Errorf("failed to send transaction: %v", err)
	}
	txHash := tx.Hash().Hex()
	return txHash, nil
}

// 验证交易
func (e *ETHNode) checkTransaction(tx_hash string) (bool, *big.Int, error) {
	receipt, err := e.RpcClient.TransactionReceipt(e.Ctx, common.HexToHash(tx_hash))
	if err != nil {
		return false, nil, fmt.Errorf("check tx(%s) error: %v", tx_hash, err)
	}
	if receipt.Status == types.ReceiptStatusSuccessful {
		effectiveGasPrice := receipt.EffectiveGasPrice
		actual_gas := big.NewInt(int64(receipt.GasUsed))
		actualGasCost := new(big.Int).Mul(actual_gas, effectiveGasPrice)
		return true, actualGasCost, nil
	} else {
		return false, nil, nil
	}
}

// 完整的执行交易
func (e *ETHNode) handleTx(symbol, from, to string, amount float64) error {
	unsignedTx, err := e.contractTx(symbol, from, to, amount)
	if err != nil {
		return err
	}
	signedTx, err := e.signTx(unsignedTx, from)
	if err != nil {
		return err
	}
	tx_hash, err := e.sendTransaction(signedTx)
	if err != nil {
		return err
	}
	var big_amount *big.Int
	switch symbol {
	case "ETH":
		big_amount = utils.Float64ToBigIntETH(amount)
	case "USDT":
		big_amount = utils.Float64ToBigIntUSDT(amount)
	default:
		return fmt.Errorf("symbol(%s) error at func handleTx", symbol)
	}
	e.UnConfirmedTxs.mu.Lock()
	e.UnConfirmedTxs.Transactions[tx_hash] = message.Transaction{
		Chain:  "ETH",
		Symbol: symbol,
		From:   from,
		To:     to,
		TxHash: tx_hash,
		Amount: big_amount,
		Status: constant.STATUS_UNTRANSFER,
	}
	e.UnConfirmedTxs.mu.Unlock()
	return nil
}

func (e *ETHNode) listenETHTransactions() error {
	headers := make(chan *types.Header, 10)

	// 负责重连
	for {
		// 订阅新区块头
		sub, err := e.WsClient.SubscribeNewHead(e.Ctx, headers)
		if err != nil {
			fmt.Println("❌ 订阅ETH交易失败, 5秒后重试:", err)
			time.Sleep(5 * time.Second)
			continue
		}
		fmt.Println("✅ ETH交易订阅成功")

		// 处理新区块
		for {
			select {
			case err := <-sub.Err():
				fmt.Println("⚠️ ETH交易订阅异常，准备重连:", err)
				sub.Unsubscribe()
				time.Sleep(3 * time.Second)
				goto reconnect

			case header := <-headers:
				// 每当有新区块，检查待确认交易
				currentHeight := header.Number.Uint64()
				e.updateNetInfo(currentHeight)
				go e.handleETHEvent(header)
				go e.confirm()
			case <-e.Ctx.Done():
				fmt.Println("🛑 收到停止信号，退出ETH交易监听")
				sub.Unsubscribe()
				return e.Ctx.Err()
			}
		}
	reconnect:
	}
}

func (e *ETHNode) handleHistoryETHEvent(height uint64) {
	block, err := e.RpcClient.BlockByNumber(e.Ctx, big.NewInt(int64(height)))
	if err != nil {
		log.Println(err)
		return
	}

	for _, tx := range block.Transactions() {
		txHash := tx.Hash().Hex()
		// 只处理ETH转账（Value > 0）
		if tx.Value().Sign() <= 0 {
			continue
		}
		// 使用 types.Sender 获取发送方地址
		signer := types.LatestSignerForChainID(e.NetID)
		from, err := types.Sender(signer, tx)
		if err != nil {
			log.Println("获取发送方地址失败:", err)
			continue
		}

		toAddr := ""
		if tx.To() != nil {
			toAddr = strings.ToLower(tx.To().Hex())
		}
		fromAddr := strings.ToLower(from.Hex())
		// 获取交易金额
		amount := utils.BigIntETHToFloat64(tx.Value())

		if _, ok := e.Wallets[toAddr]; ok {
			msg, err := e.MessageServer.FindTopupMsgWithToaddress("ETH", toAddr)
			if err != nil {
				log.Printf("❌ 未查找到ETH充值消息中有address(%s)信息", toAddr)
				continue
			}
			log.Printf("当前交易txHash:%s, fromAddr: %s, toAddr: %s, amount: %fETH", txHash, fromAddr, toAddr, amount)
			log.Println(msg)
		}
	}
}

func (e *ETHNode) handleETHEvent(header *types.Header) {
	height := header.Number.Uint64()
	log.Printf("当前区块高度：%d", height)
	// 获取区块中的所有交易
	block, err := e.RpcClient.BlockByHash(e.Ctx, header.Hash())
	if err != nil {
		log.Printf("无法获取区块信息: %v", err)
		return
	}
	// 遍历区块中的每笔交易
	for _, tx := range block.Transactions() {
		txHash := tx.Hash().Hex()
		// 只处理ETH转账（Value > 0）
		if tx.Value().Sign() <= 0 {
			continue
		}
		// 使用 types.Sender 获取发送方地址
		signer := types.LatestSignerForChainID(e.NetID)
		from, err := types.Sender(signer, tx)
		if err != nil {
			log.Println("获取发送方地址失败:", err)
			continue
		}

		toAddr := ""
		if tx.To() != nil {
			toAddr = strings.ToLower(tx.To().Hex())
		}
		fromAddr := strings.ToLower(from.Hex())
		// 获取交易金额
		amount := utils.BigIntETHToFloat64(tx.Value())
		// toAddr和监听钱包一致，表示(充值)
		if _, ok := e.Wallets[toAddr]; ok {
			log.Printf("监听地址: %s，交易信息：%v", toAddr, tx)
			msg, err := e.MessageServer.FindTopupMsgWithToaddress("ETH", toAddr)
			if err != nil {
				log.Printf("❌ 未查找到ETH充值消息中有address(%s)信息", toAddr)
				continue
			}
			resp := message.TopupMsg_resp{
				QueueId:     msg.QueueId,
				Chain:       "ETH",
				Symbol:      "ETH",
				FromAddress: fromAddr,
				Address:     toAddr,
				TxHash:      txHash,
				Amount:      amount,
				BlockHeight: height,
				Status:      constant.STATUS_PENDING,
			}
			go e.asyncSendMsgToListen(resp, 3, 5*time.Second)

			e.UnConfirmedTxs.mu.Lock()
			e.UnConfirmedTxs.Transactions[txHash] = message.Transaction{
				QueueId: msg.QueueId,
				TxType:  0,
				Chain:   "ETH",
				Symbol:  "ETH",
				From:    fromAddr,
				To:      toAddr,
				TxHash:  txHash,
				Height:  height,
				Amount:  tx.Value(),
				Status:  constant.STATUS_PENDING,
			}
			e.UnConfirmedTxs.mu.Unlock()
			go func() {
				str := "INSERT INTO ETH_unconfirmed_tx(queue_id, tx_type, chain, symbol, from_addr, to_addr, tx_hash, height, amount) VALUES (?,?,?,?,?,?,?,?,?)"
				params := []any{msg.QueueId, 0, "ETH", "ETH", fromAddr, toAddr, txHash, height, amount}
				_, err := e.MysqlDB.Insert(str, [][]any{params})
				if err != nil {
					log.Fatalf("INSERT ETH_unconfirmed_tx table error: %v", err)
					return
				}
			}()
			continue
		}

		// fromAddr和监听钱包一致，表示(提现/支付)
		if _, ok := e.Wallets[fromAddr]; ok {
			var msg any
			var ee error
			msg, ee = e.MessageServer.FindPayMsgWithToaddress("ETH", fromAddr)
			if ee != nil {
				msg, ee = e.MessageServer.FindWithdrawMsgWithToaddress("ETH", fromAddr)
				if ee != nil {
					log.Printf("❌ 未查找到ETH提现和支付消息中有address(%s)信息", fromAddr)
					continue
				}
			}
			switch v := msg.(type) {
			case message.WithdrawMsg_req:
				if v.FromAddress == fromAddr && v.ToAddress == toAddr && v.Amount == amount {
					resp := message.WithdrawMsg_resp{
						QueueId:     v.QueueId,
						Chain:       "ETH",
						Symbol:      "ETH", // 使用消息中的Symbol（应该是ETH）
						FromAddress: fromAddr,
						ToAddress:   toAddr,
						TxHash:      txHash,
						Amount:      amount,
						Fee:         v.Fee,
						BlockHeight: height,
						Status:      constant.STATUS_PENDING,
					}
					go e.asyncSendMsgToListen(resp, 3, 5*time.Second)
					e.UnConfirmedTxs.mu.Lock()
					e.UnConfirmedTxs.Transactions[txHash] = message.Transaction{
						QueueId: v.QueueId,
						TxType:  1, // 提现类型
						Chain:   "ETH",
						Symbol:  "ETH", // 使用消息中的Symbol（应该是ETH）
						From:    fromAddr,
						To:      toAddr,
						TxHash:  txHash,
						Height:  height,
						Amount:  tx.Value(),
						Status:  constant.STATUS_PENDING,
					}
					e.UnConfirmedTxs.mu.Unlock()
					go func() {
						str := "INSERT INTO ETH_unconfirmed_tx(queue_id, tx_type, chain, symbol, from_addr, to_addr, tx_hash, height, amount) VALUES (?,?,?,?,?,?,?,?,?)"
						params := []any{v.QueueId, 1, "ETH", "ETH", fromAddr, toAddr, txHash, height, amount}
						_, err := e.MysqlDB.Insert(str, [][]any{params})
						if err != nil {
							log.Fatalf("INSERT ETH_unconfirmed_tx table error: %v", err)
							return
						}
					}()
					continue
				}

			case message.PayMsg_req:
				if v.FromAddress == fromAddr && v.ToAddress == toAddr && v.Amount == amount {
					resp := message.PayMsg_resp{
						QueueId:     v.QueueId,
						Chain:       "ETH",
						Symbol:      "ETH", // 使用消息中的Symbol（应该是ETH）
						FromAddress: fromAddr,
						ToAddress:   toAddr,
						TxHash:      txHash,
						Amount:      amount,
						Fee:         v.Fee,
						BlockHeight: height,
						Status:      constant.STATUS_PENDING,
					}
					go e.asyncSendMsgToListen(resp, 3, 5*time.Second)
					e.UnConfirmedTxs.mu.Lock()
					e.UnConfirmedTxs.Transactions[txHash] = message.Transaction{
						QueueId: v.QueueId,
						TxType:  2, // 提现类型
						Chain:   "ETH",
						Symbol:  "ETH", // 使用消息中的Symbol（应该是ETH）
						From:    fromAddr,
						To:      toAddr,
						TxHash:  txHash,
						Height:  height,
						Amount:  tx.Value(),
						Status:  constant.STATUS_PENDING,
					}
					e.UnConfirmedTxs.mu.Unlock()
					go func() {
						str := "INSERT INTO ETH_unconfirmed_tx(queue_id, tx_type, chain, symbol, from_addr, to_addr, tx_hash, height, amount) VALUES (?,?,?,?,?,?,?,?,?)"
						params := []any{v.QueueId, 2, "ETH", "ETH", fromAddr, toAddr, txHash, height, amount}
						_, err := e.MysqlDB.Insert(str, [][]any{params})
						if err != nil {
							log.Fatalf("INSERT ETH_unconfirmed_tx table error: %v", err)
							return
						}
					}()
					continue
				}

			default:

			}
		}
	}
}

func (e *ETHNode) listenUSDTTransactions() error {
	// 过滤掉非USDT数据
	query := ethereum.FilterQuery{
		Addresses: []common.Address{e.USDT.Address},
	}
	// 负责重连
	for {
		// 订阅日志
		sub, err := e.WsClient.SubscribeFilterLogs(e.Ctx, query, e.USDT.LogsChan)
		if err != nil {
			fmt.Println("❌ USDT交易订阅失败, 5秒后重试:", err)
			time.Sleep(5 * time.Second)
			continue
		}
		fmt.Println("✅ USDT交易订阅成功")
		// 处理事件
		for {

			select {
			case err := <-sub.Err():
				fmt.Println("⚠️ USDT交易订阅异常，准备重连:", err)
				sub.Unsubscribe() // 清理旧订阅
				time.Sleep(3 * time.Second)
				goto reconnect // 跳出内层循环，回到外层重新订阅

			case vLog := <-e.USDT.LogsChan:
				go e.handleUSDTEvent(vLog) // 事件解析 + 分类，传递链消息的通道是vLog而非ch，且一次只传递一笔交易
			case <-e.Ctx.Done():
				fmt.Println("🛑 收到停止信号，退出USDT交易监听")
				sub.Unsubscribe()
				return e.Ctx.Err()
			}
		}
	reconnect:
	}
}

func (e *ETHNode) handleUSDTEvent(vLog types.Log) {
	from := common.HexToAddress(vLog.Topics[1].Hex())
	to := common.HexToAddress(vLog.Topics[2].Hex())
	height := vLog.BlockNumber
	fromAddr := strings.ToLower(from.Hex())
	toAddr := strings.ToLower(to.Hex())
	var transferEvent struct{ Value *big.Int }
	if err := e.USDT.ABI.UnpackIntoInterface(&transferEvent, "Transfer", vLog.Data); err != nil {
		fmt.Println("ABI 解析错误:", err)
		return
	}
	txHash := vLog.TxHash.Hex()
	value_float := utils.BigIntUSDTToFloat64(transferEvent.Value)
	// toAddr和监听钱包一致，表示(充值)
	if _, ok := e.Wallets[toAddr]; ok {
		msg, err := e.MessageServer.FindTopupMsgWithToaddress("ETH", toAddr)
		if err != nil {
			log.Printf("❌ 未查找到ETH充值消息中有address(%s)信息", toAddr)
			return
		}
		resp := message.TopupMsg_resp{
			QueueId:     msg.QueueId,
			Chain:       "ETH",
			Symbol:      "USDT",
			FromAddress: fromAddr,
			Address:     toAddr,
			TxHash:      txHash,
			Amount:      value_float,
			BlockHeight: height,
			Status:      constant.STATUS_PENDING,
		}
		go e.asyncSendMsgToListen(resp, 3, 5*time.Second)

		e.UnConfirmedTxs.mu.Lock()
		e.UnConfirmedTxs.Transactions[txHash] = message.Transaction{
			QueueId: msg.QueueId,
			TxType:  0,
			Chain:   "ETH",
			Symbol:  "USDT",
			From:    fromAddr,
			To:      toAddr,
			TxHash:  txHash,
			Height:  height,
			Amount:  transferEvent.Value,
			Status:  constant.STATUS_PENDING,
		}
		e.UnConfirmedTxs.mu.Unlock()
		go func() {
			str := "INSERT INTO ETH_unconfirmed_tx(queue_id, tx_type, chain, symbol, from_addr, to_addr, tx_hash, height, amount) VALUES (?,?,?,?,?,?,?,?,?)"
			params := []any{msg.QueueId, 0, "ETH", "USDT", fromAddr, toAddr, txHash, height, value_float}
			_, err := e.MysqlDB.Insert(str, [][]any{params})
			if err != nil {
				log.Fatalf("INSERT ETH_unconfirmed_tx table error: %v", err)
				return
			}
		}()
		return
	}

	// fromAddr和监听钱包一致，表示(提现/支付)
	if _, ok := e.Wallets[fromAddr]; ok {
		var msg any
		var ee error
		msg, ee = e.MessageServer.FindPayMsgWithToaddress("ETH", fromAddr)
		if ee != nil {
			msg, ee = e.MessageServer.FindWithdrawMsgWithToaddress("ETH", fromAddr)
			if ee != nil {
				log.Printf("❌ 未查找到ETH提现和支付消息中有address(%s)信息", fromAddr)
				return
			}
		}
		switch v := msg.(type) {
		case message.WithdrawMsg_req:
			if v.FromAddress == fromAddr && v.ToAddress == toAddr && v.Amount == value_float {
				resp := message.WithdrawMsg_resp{
					QueueId:     v.QueueId,
					Chain:       "ETH",
					Symbol:      "USDT",
					FromAddress: fromAddr,
					ToAddress:   toAddr,
					TxHash:      txHash,
					Amount:      value_float,
					Fee:         v.Fee,
					BlockHeight: height,
					Status:      constant.STATUS_PENDING,
				}
				go e.asyncSendMsgToListen(resp, 3, 5*time.Second)
				e.UnConfirmedTxs.mu.Lock()
				e.UnConfirmedTxs.Transactions[txHash] = message.Transaction{
					QueueId: v.QueueId,
					TxType:  1, // 提现类型
					Chain:   "ETH",
					Symbol:  "USDT", // 使用消息中的Symbol（应该是USDT）
					From:    fromAddr,
					To:      toAddr,
					TxHash:  txHash,
					Height:  height,
					Amount:  transferEvent.Value,
					Status:  constant.STATUS_PENDING,
				}
				e.UnConfirmedTxs.mu.Unlock()
				go func() {
					str := "INSERT INTO ETH_unconfirmed_tx(queue_id, tx_type, chain, symbol, from_addr, to_addr, tx_hash, height, amount) VALUES (?,?,?,?,?,?,?,?,?)"
					params := []any{v.QueueId, 1, "ETH", "USDT", fromAddr, toAddr, txHash, height, value_float}
					_, err := e.MysqlDB.Insert(str, [][]any{params})
					if err != nil {
						log.Fatalf("INSERT ETH_unconfirmed_tx table error: %v", err)
						return
					}
				}()
				return
			}

		case message.PayMsg_req:
			if v.FromAddress == fromAddr && v.ToAddress == toAddr && v.Amount == value_float {
				resp := message.PayMsg_resp{
					QueueId:     v.QueueId,
					Chain:       "ETH",
					Symbol:      "USDT",
					FromAddress: fromAddr,
					ToAddress:   toAddr,
					TxHash:      txHash,
					Amount:      value_float,
					Fee:         v.Fee,
					BlockHeight: height,
					Status:      constant.STATUS_PENDING,
				}
				go e.asyncSendMsgToListen(resp, 3, 5*time.Second)
				e.UnConfirmedTxs.mu.Lock()
				e.UnConfirmedTxs.Transactions[txHash] = message.Transaction{
					QueueId: v.QueueId,
					TxType:  1, // 提现类型
					Chain:   "ETH",
					Symbol:  "USDT", // 使用消息中的Symbol（应该是USDT）
					From:    fromAddr,
					To:      toAddr,
					TxHash:  txHash,
					Height:  height,
					Amount:  transferEvent.Value,
					Status:  constant.STATUS_PENDING,
				}
				e.UnConfirmedTxs.mu.Unlock()
				go func() {
					str := "INSERT INTO ETH_unconfirmed_tx(queue_id, tx_type, chain, symbol, from_addr, to_addr, tx_hash, height, amount) VALUES (?,?,?,?,?,?,?,?,?)"
					params := []any{v.QueueId, 2, "ETH", "USDT", fromAddr, toAddr, txHash, height, value_float}
					_, err := e.MysqlDB.Insert(str, [][]any{params})
					if err != nil {
						log.Fatalf("INSERT ETH_unconfirmed_tx table error: %v", err)
						return
					}
				}()
				return
			}
		default:

		}
	}
}

var tableMap = map[string]string{
	"ETH":  "ETH_balances",
	"USDT": "USDT_balances",
}

func (e *ETHNode) confirm() {

	e.mu.Lock()
	unconfirmedTxs := e.UnConfirmedTxs
	now_height := e.NetInfo.Height
	e.mu.Unlock()
	var responses []any
	e.UnConfirmedTxs.mu.Lock()
	for txHash, tx := range unconfirmedTxs.Transactions {

		// 高度成熟：当前高度 >= 交易高度 + 确认高度
		if now_height >= tx.Height+e.ConfirmHeight {
			check_result, actual_gas, err := e.checkTransaction(txHash)
			if err != nil {
				log.Printf("Transaction(%s) check error: %v, 跳过此交易", txHash, err)
				continue
			}
			var status int
			var msg any
			var ee error
			// 交易检查成功，通过queue_id找到对应的消息，并根据消息发送回
			// 统一处理交易结果
			if check_result {
				msg, ee = e.MessageServer.FindMsgWithQueueID("ETH", tx.QueueId, tx.TxType)
				if ee != nil {
					log.Printf("Query Message error: %v, queue_id: %s, tx_type: %d", ee, tx.QueueId, tx.TxType)
					status = constant.STATUS_FAILED

					// 如果找不到消息，仍然需要处理交易失败的情况
					msg = nil
				} else {
					status = constant.STATUS_SUCCESS
				}
			} else {
				status = constant.STATUS_FAILED
				// 交易失败时也需要查找消息以发送响应
				msg, ee = e.MessageServer.FindMsgWithQueueID("ETH", tx.QueueId, tx.TxType)
				if ee != nil {
					log.Printf("Query Message error for failed tx: %v, queue_id: %s, tx_type: %d", ee, tx.QueueId, tx.TxType)
					msg = nil
				}
			}

			var float_amount = utils.BigIntToFloat64(tx.Symbol, tx.Amount)

			// 如果找不到消息，跳过处理
			if msg == nil {
				delete(e.UnConfirmedTxs.Transactions, txHash)
				continue
			}

			switch v := msg.(type) {
			case message.TopupMsg_req:
				if status == constant.STATUS_SUCCESS {
					// 确认充值消息，通道返回resp，并修改钱包、数据库
					switch tx.Symbol {
					case "ETH":
						e.Wallets[tx.To].eth_balance.balance = new(big.Int).Add(e.Wallets[tx.To].eth_balance.balance, tx.Amount)
					case "USDT":
						e.Wallets[tx.To].usdt_balance.balance = new(big.Int).Add(e.Wallets[tx.To].usdt_balance.balance, tx.Amount)
					default:
						log.Printf("error symbol(%s): %v, 跳过此交易", tx.Symbol, tx.Symbol)
						delete(e.UnConfirmedTxs.Transactions, txHash)
						continue
					}
				}
				response := message.TopupMsg_resp{
					QueueId:     tx.QueueId,
					Address:     tx.To,
					FromAddress: tx.From,
					Status:      status,
					Chain:       tx.Chain,
					Symbol:      tx.Symbol,
					Amount:      float_amount,
					TxHash:      txHash,
					BlockHeight: tx.Height,
				}
				// 修改钱包表
				go func() {
					var str string
					var params []any
					if status == constant.STATUS_SUCCESS {
						str = "UPDATE " + tableMap[tx.Symbol] + " SET success_tx_hash = CONCAT(success_tx_hash, ?),  balance = balance + ? WHERE address = ?"
						params = []any{txHash + ",", float_amount, v.Address}
					} else {
						str = "UPDATE " + tableMap[tx.Symbol] + " SET failed_tx_hash = CONCAT(failed_tx_hash, ?) WHERE address = ?"
						params = []any{txHash + ",", v.Address}
					}

					_, err := e.MysqlDB.Update(str, params)
					if err != nil {
						// 更详细的错误日志，包括 QueueId 和 Status
						log.Printf("Failed to update success_tx_hash/failed_tx_hash for queue_id %s: %v", v.QueueId, err)
					}
				}()
				// 修改待确认交易表
				go func() {
					str := "UPDATE ETH_unconfirmed_tx SET status = ? WHERE tx_hash = ?"
					params := []any{status, txHash}
					_, err := e.MysqlDB.Update(str, params)
					if err != nil {
						log.Printf("Failed to update ETH_unconfirmed_tx: %v", err)
					}
				}()
				responses = append(responses, response) // 将消息提交至responses
			case message.WithdrawMsg_req:
				if status == constant.STATUS_SUCCESS {
					// 将gas费添加到钱包中
					e.Wallets[tx.From].eth_balance.used_gas = new(big.Int).Add(e.Wallets[tx.From].eth_balance.used_gas, actual_gas)

					switch tx.Symbol {
					case "ETH":
						e.Wallets[tx.From].eth_balance.balance = new(big.Int).Sub(e.Wallets[tx.From].eth_balance.balance, tx.Amount)
					case "USDT":
						e.Wallets[tx.From].usdt_balance.balance = new(big.Int).Sub(e.Wallets[tx.From].usdt_balance.balance, tx.Amount)
						// 提现USDT时会有一定数量的USDT作为gas费冻结
						e.Wallets[tx.From].usdt_balance.freeze_num = new(big.Int).Add(e.Wallets[tx.From].usdt_balance.freeze_num, utils.Float64ToBigIntUSDT(v.Fee))
					default:
						log.Printf("error symbol(%s): %v, 跳过此交易", tx.Symbol, tx.Symbol)
						delete(e.UnConfirmedTxs.Transactions, txHash)
						continue
					}
				}
				response := message.WithdrawMsg_resp{
					QueueId:     tx.QueueId,
					Chain:       tx.Chain,
					Symbol:      tx.Symbol,
					Status:      status,
					Amount:      float_amount,
					Fee:         v.Fee,
					TxHash:      txHash,
					FromAddress: tx.From,
					ToAddress:   tx.To,
					BlockHeight: tx.Height,
				}
				// 修改钱包表
				go func() {
					var str, str1 string
					var params, params1 []any
					txHashWithComma := txHash + ","
					if status == constant.STATUS_SUCCESS {
						if tableMap[tx.Symbol] == "ETH_balances" {
							str = "UPDATE ETH_balances SET success_tx_hash = CONCAT(success_tx_hash, ?),  balance = balance - ?, used_gas = used_gas + ? WHERE address = ?"
							params = []any{txHashWithComma, float_amount, utils.BigIntETHToFloat64(actual_gas), v.FromAddress}
							str1 = ""
							params1 = []any{}
						} else {
							str = "UPDATE USDT_balances SET success_tx_hash = CONCAT(success_tx_hash, ?),  balance = balance - ?, freeze_num = freeze_num + ? WHERE address = ?"
							params = []any{txHashWithComma, float_amount, v.Fee, v.FromAddress}
							str1 = "UPDATE ETH_balances SET used_gas = used_gas + ? WHERE address = ?"
							params1 = []any{utils.BigIntETHToFloat64(actual_gas), v.FromAddress}
						}
					} else {
						str = "UPDATE " + tableMap[tx.Symbol] + " SET failed_tx_hash = CONCAT(failed_tx_hash, ?) WHERE address = ?"
						params = []any{txHashWithComma, v.FromAddress}
						str1 = ""
						params1 = []any{}
					}

					err := e.MysqlDB.ExecuteTransactions([]string{str, str1}, [][]any{params, params1})
					if err != nil {
						// 更详细的错误日志，包括 QueueId 和 Status
						log.Printf("Failed to update success_tx_hash/failed_tx_hash for queue_id %s: %v", v.QueueId, err)
					}
				}()
				// 修改待确认交易表
				go func() {
					str := "UPDATE ETH_unconfirmed_tx SET status = ? WHERE tx_hash = ?"
					params := []any{status, txHash}
					_, err := e.MysqlDB.Update(str, params)
					if err != nil {
						log.Printf("Failed to update ETH_unconfirmed_tx: %v", err)
					}
				}()
				responses = append(responses, response) // 将消息提交至responses
			case message.PayMsg_req:
				e.Wallets[tx.From].eth_balance.used_gas = new(big.Int).Add(e.Wallets[tx.From].eth_balance.used_gas, actual_gas)

				switch tx.Symbol {
				case "ETH":
					e.Wallets[tx.From].eth_balance.balance = new(big.Int).Sub(e.Wallets[tx.From].eth_balance.balance, tx.Amount)
				case "USDT":
					e.Wallets[tx.From].usdt_balance.balance = new(big.Int).Sub(e.Wallets[tx.From].usdt_balance.balance, tx.Amount)
					// 支付USDT时需要冻结部分USDT作为gas费(暂不执行)
					// e.Wallets[tx.From].usdt_balance.freeze_num = new(big.Int).Add(e.Wallets[tx.From].usdt_balance.freeze_num, tx.Amount)
				default:
					log.Printf("error symbol(%s): %v, 跳过此交易", tx.Symbol, tx.Symbol)
					delete(e.UnConfirmedTxs.Transactions, txHash)
					continue
				}
				response := message.PayMsg_resp{
					QueueId:     tx.QueueId,
					Chain:       tx.Chain,
					Symbol:      tx.Symbol,
					Status:      status,
					Amount:      float_amount,
					Fee:         v.Fee,
					TxHash:      txHash,
					FromAddress: tx.From,
					ToAddress:   tx.To,
					BlockHeight: tx.Height,
				}
				// 修改钱包表
				go func() {
					var str, str1 string
					var params, params1 []any
					txHashWithComma := txHash + ","
					if status == constant.STATUS_SUCCESS {
						if tableMap[tx.Symbol] == "ETH_balances" {
							str = "UPDATE ETH_balances SET success_tx_hash = CONCAT(success_tx_hash, ?),  balance = balance - ?, used_gas = used_gas + ? WHERE address = ?"
							params = []any{txHashWithComma, float_amount, utils.BigIntETHToFloat64(actual_gas), v.FromAddress}
							str1 = ""
							params1 = []any{}
						} else {
							str = "UPDATE USDT_balances SET success_tx_hash = CONCAT(success_tx_hash, ?),  balance = balance - ?, freeze_num = freeze_num + ? WHERE address = ?"
							params = []any{txHashWithComma, float_amount, v.Fee, v.FromAddress}
							str1 = "UPDATE ETH_balances SET used_gas = used_gas + ? WHERE address = ?"
							params1 = []any{utils.BigIntETHToFloat64(actual_gas), v.FromAddress}
						}
					} else {
						str = "UPDATE " + tableMap[tx.Symbol] + " SET failed_tx_hash = CONCAT(failed_tx_hash, ?) WHERE address = ?"
						params = []any{txHashWithComma, v.FromAddress}
						str1 = ""
						params1 = []any{}
					}

					err := e.MysqlDB.ExecuteTransactions([]string{str, str1}, [][]any{params, params1})
					if err != nil {
						// 更详细的错误日志，包括 QueueId 和 Status
						log.Printf("Failed to update success_tx_hash/failed_tx_hash for queue_id %s: %v", v.QueueId, err)
					}
				}()
				// 修改待确认交易表
				go func() {
					str := "UPDATE ETH_unconfirmed_tx SET status = ? WHERE tx_hash = ?"
					params := []any{status, txHash}
					_, err := e.MysqlDB.Update(str, params)
					if err != nil {
						log.Printf("Failed to update ETH_unconfirmed_tx: %v", err)
					}
				}()
				responses = append(responses, response) // 将消息提交至responses
			default:
				log.Printf("未知的消息类型: %v, 跳过此交易", v)
				delete(e.UnConfirmedTxs.Transactions, txHash)
				continue
			}
			delete(e.UnConfirmedTxs.Transactions, txHash)
		}
	}
	e.UnConfirmedTxs.mu.Unlock()
	if len(responses) > 0 {
		for _, v := range responses {
			go e.asyncSendMsgToListen(v, 3, 5*time.Second)
		}
	}
}

func (e *ETHNode) handleListen_Topup_req(msg message.TopupMsg_req) {
	pk, err := e.getAddressPk(msg.Address)
	if err != nil {
		log.Printf("Query balance(%s-%s) private_key error: %v", msg.Chain, msg.Address, err)
		go e.asyncSendMsgToListen(message.TopupMsg_resp{
			QueueId: msg.QueueId,
			Chain:   msg.Chain,
			Symbol:  msg.Symbol,
			Address: msg.Address,
			Status:  msg.Status,
		}, 3, 5*time.Second)
		return
	}
	// 添加到钱包
	e.mu.Lock()
	e.Wallets[msg.Address] = &Wallets{
		address: msg.Address,
		queueId: msg.QueueId,
		pk:      pk,
		eth_balance: &eth_balance{
			symbol:            "ETH",
			used_gas:          big.NewInt(0),
			balance:           big.NewInt(0),
			successed_tx_hash: []string{},
			failed_tx_hash:    []string{},
		},
		usdt_balance: &usdt_balance{
			symbol:            "USDT",
			freeze_num:        big.NewInt(0),
			balance:           big.NewInt(0),
			successed_tx_hash: []string{},
			failed_tx_hash:    []string{},
		},
		timestamp: msg.Timestamp,
		sign:      msg.Sign,
		status:    constant.STATUS_SUCCESS,
	}
	e.mu.Unlock()
	// 记录到钱包数据库
	go func() {
		str1 := "INSERT INTO ETH_wallets (address, queue_id, timestamp, sign, status) VALUES (?,?,?,?,?)"
		params1 := []any{msg.Address, msg.QueueId, msg.Timestamp, msg.Sign, constant.STATUS_SUCCESS}
		str2 := "INSERT INTO ETH_balances (address) VALUES (?)"
		params2 := []any{msg.Address}
		str3 := "INSERT INTO USDT_balances (address) VALUES (?)"
		params3 := []any{msg.Address}
		err = e.MysqlDB.ExecuteTransactions([]string{str1, str2, str3}, [][]any{params1, params2, params3})
		if err != nil {
			log.Printf("Received ListenServer Topup_req msg: insert sqlite3 db error: %v", err)
		}
		go e.asyncSendMsgToListen(message.UpdateReqState{
			QueueId: msg.QueueId,
			MsgType: 0,
			Status:  constant.STATUS_SUCCESS,
		}, 3, 5*time.Second)
	}()

}

func (e *ETHNode) handleListen_Withdraw_req(msg message.WithdrawMsg_req) {
	// 先获得当前最高gas费用
	e.NetInfo.mu.Lock()
	maxGas := e.NetInfo.GasFeeCap
	e.NetInfo.mu.Unlock()
	var target_amount_eth, target_amount_usdt *big.Int
	// 将 msg.Amount 和 msg.Fee 转换为 big.Int，只调用一次
	amountBigInt := utils.Float64ToBigInt(msg.Symbol, msg.Amount)
	feeBigInt := utils.Float64ToBigInt(msg.Symbol, msg.Fee)
	switch msg.Symbol {
	case "ETH":
		// 计算目标金额
		target_amount_eth = new(big.Int).Add(maxGas, amountBigInt) // maxGas + msg.Amount
		target_amount_eth.Add(target_amount_eth, feeBigInt)        // (maxGas + msg.Amount) + msg.Fee
		target_amount_usdt = big.NewInt(0)
	case "USDT":
		target_amount_eth = maxGas
		target_amount_usdt = new(big.Int).Add(amountBigInt, feeBigInt)
	default:
		return
	}
	// 构建相应通用数据，Status根据后续情况变化
	result_msg := message.WithdrawMsg_resp{
		QueueId:     msg.QueueId,
		Chain:       msg.Chain,
		Symbol:      msg.Symbol,
		Amount:      msg.Amount,
		Fee:         msg.Fee,
		FromAddress: msg.FromAddress,
		ToAddress:   msg.ToAddress,
	}

	check_result, err := e.checkBalance(msg.Symbol, msg.FromAddress, target_amount_eth, target_amount_usdt)
	// 余额校验错误，绕过转账，返回错误响应
	if err != nil {
		log.Printf("check balance error: %v", err)
		result_msg.Status = constant.STATUS_ERROR
		go e.asyncSendMsgToListen(result_msg, 3, 5*time.Second)
		return
	}
	// 校验成功
	if check_result {
		// 开始转账
		err := e.Transfer(msg.FromAddress, msg.ToAddress, msg.Symbol, msg.Amount, msg.Fee)
		// 转账失败，返回转账失败结果
		if err != nil {
			log.Printf("withdraw - transfer error: %v", err)
			result_msg.Status = constant.STATUS_ERROR
			// 提现转账错误
			go e.asyncSendMsgToListen(result_msg, 3, 5*time.Second)
			return
		}
		// 转账成功，等待chain server listen监听到该笔交易
		go e.asyncSendMsgToListen(message.UpdateReqState{
			QueueId: msg.QueueId,
			MsgType: 1,
			Status:  constant.STATUS_SUCCESS,
		}, 3, 5*time.Second)
		log.Printf("withdraw - transfer success: QueueId(%s)", msg.QueueId)
	} else { // 校验失败
		// 提现转账账户余额不足
		result_msg.Status = constant.STATUS_BALANCE_NOT_ENOUGH
		go e.asyncSendMsgToListen(result_msg, 3, 5*time.Second)
		return
	}
}

func (e *ETHNode) handleListen_Pay_req(msg message.PayMsg_req) {
	// 先获得当前最高gas费用
	e.NetInfo.mu.Lock()
	maxGas := e.NetInfo.GasFeeCap
	e.NetInfo.mu.Unlock()
	var target_amount_eth, target_amount_usdt *big.Int
	// 将 msg.Amount 和 msg.Fee 转换为 big.Int，只调用一次
	amountBigInt := utils.Float64ToBigInt(msg.Symbol, msg.Amount)
	feeBigInt := utils.Float64ToBigInt(msg.Symbol, msg.Fee)
	switch msg.Symbol {
	case "ETH":
		// 计算目标金额
		target_amount_eth = new(big.Int).Add(maxGas, amountBigInt) // maxGas + msg.Amount
		target_amount_eth.Add(target_amount_eth, feeBigInt)        // (maxGas + msg.Amount) + msg.Fee
		target_amount_usdt = big.NewInt(0)
	case "USDT":
		target_amount_eth = maxGas
		target_amount_usdt = new(big.Int).Add(amountBigInt, feeBigInt)
	default:
		return
	}
	// 构建相应通用数据，Status根据后续情况变化
	result_msg := message.PayMsg_resp{
		QueueId:     msg.QueueId,
		Chain:       msg.Chain,
		Symbol:      msg.Symbol,
		Amount:      msg.Amount,
		Fee:         msg.Fee,
		FromAddress: msg.FromAddress,
		ToAddress:   msg.ToAddress,
	}
	check_result, err := e.checkBalance(msg.Symbol, msg.FromAddress, target_amount_eth, target_amount_usdt)
	// 余额校验错误，绕过转账，返回错误响应
	if err != nil {
		log.Printf("check balance error: %v", err)
		result_msg.Status = constant.STATUS_ERROR
		go e.asyncSendMsgToListen(result_msg, 3, 5*time.Second)
		return
	}
	// 校验成功
	if check_result {
		// 开始转账
		err := e.Transfer(msg.FromAddress, msg.ToAddress, msg.Symbol, msg.Amount, msg.Fee)
		// 转账失败，返回转账失败结果
		if err != nil {
			log.Printf("withdraw - transfer error: %v", err)
			result_msg.Status = constant.STATUS_ERROR
			// 提现转账错误
			go e.asyncSendMsgToListen(result_msg, 3, 5*time.Second)
			return
		}
		// 转账成功，等待chain server listen监听到该笔交易
		go e.asyncSendMsgToListen(message.UpdateReqState{
			QueueId: msg.QueueId,
			MsgType: 1,
			Status:  constant.STATUS_SUCCESS,
		}, 3, 5*time.Second)
		log.Printf("withdraw - transfer success: QueueId(%s)", msg.QueueId)
	} else { // 校验失败
		// 提现转账账户余额不足
		result_msg.Status = constant.STATUS_BALANCE_NOT_ENOUGH
		go e.asyncSendMsgToListen(result_msg, 3, 5*time.Second)
		return
	}
}

func (e *ETHNode) handleListen_Remove_req(msg message.RemoveListenMsg_req) {
	e.mu.Lock()
	delete(e.Wallets, msg.Address)
	e.mu.Unlock()
	result_msg := message.RemoveListenMsg_resp{
		QueueId: msg.QueueId,
		MsgType: msg.MsgType,
		Chain:   msg.Chain,
		Symbol:  msg.Symbol,
		Status:  constant.STATUS_SUCCESS,
	}
	str := "UPDATE ETH_wallets SET status = ? WHERE address = ?"
	params := []any{0, msg.Address}
	_, err := e.MysqlDB.Update(str, params)
	if err != nil {
		log.Printf("Remove address(%s) error: %v", msg.Address, err)
		// result_msg.Status = constant.STATUS_FAILED
	}
	go e.asyncSendMsgToListen(result_msg, 3, 5*time.Second)
}

func (e *ETHNode) asyncSendMsgToListen(msg any, retries int, timeout time.Duration) {
	for retries > 0 {
		select {
		case e.MessageServer.ChFromChainServer["ETH"] <- msg: // 如果通道没有满，就发送消息
			// log.Printf("Sent message to rmq_ch_out: %v", msg)
			return
		case <-time.After(timeout): // 超时控制
			log.Printf("Timeout sending message ETH-ChainServer to ListenServer: %v", msg)
			retries--
			if retries == 0 {
				log.Printf("Max retries reached, giving up on sending message: %v", msg)
				return
			}
			// 在超时后进行重试
			log.Printf("Retrying sending message: %v, retries left: %d", msg, retries)
		}
	}
}