m2pool_payment/internal/blockchain/eth/eth.go

package eth

import (
	"context"
	"fmt"
	"log"
	"m2pool-payment/internal/db"
	message "m2pool-payment/internal/msg"
	"m2pool-payment/internal/utils"
	"math/big"
	"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"
	"github.com/ethereum/go-ethereum/ethclient"
)

const erc20ABI = `
[
  {
    "constant": true,
    "inputs": [{"name": "_owner", "type": "address"}],
    "name": "balanceOf",
    "outputs": [{"name": "balance", "type": "uint256"}],
    "type": "function"
  },
  {
    "constant": false,
    "inputs": [
      {"name": "_to", "type": "address"},
      {"name": "_value", "type": "uint256"}
    ],
    "name": "transfer",
    "outputs": [{"name": "", "type": "bool"}],
    "type": "function"
  },
  {
    "anonymous": false,
    "inputs": [
      {"indexed": true, "name": "from", "type": "address"},
      {"indexed": true, "name": "to",   "type": "address"},
      {"indexed": false,"name": "value","type": "uint256"}
    ],
    "name": "Transfer",
    "type": "event"
  }
]
`

// USDT 合约地址（主网）
const USDTAddress = "0xdAC17F958D2ee523a2206206994597C13D831ec7"

// 状态码常量
const (
	STATUS_FAILED        = 0 // 失败
	STATUS_SUCCESS       = 1 // 成功
	STATUS_PENDING       = 2 // 待确认
	STATUS_VERIFY_FAILED = 3 // 验证失败
)

type ETHNode struct {
	decodeKey     string   // 私钥解密密钥，仅针对普通钱包
	NetID         *big.Int // 网络ID，主网为1，其他ID可通过rpc.NetworkID方法获取
	Config        message.ETHConfig
	ConfirmHeight uint64
	WsClient      *ethclient.Client
	RpcClient     *ethclient.Client
	Db            db.MySQLPool
	// {"address": message.TopupMsg_req{}, ...}，仅针对充值
	TopupMsg map[string]*message.TopupMsg_req
	// {"fromAddress": message.WithdrawMsg_req{}, ...}，仅针对提现
	WithdrawMsg map[string]*message.WithdrawMsg_req
	// {"fromAddress": message.PayMsg_req{}, ...}, 仅针对支付
	PayMsg map[string]*message.PayMsg_req
	// {"tx_hash": message.Transaction}
	UnConfirmTxs map[string]*message.Transaction
	LogsChan     chan *types.Header
	USDT         *USDT
	RealData     *RealData
	Ctx          context.Context
	Cancel       context.CancelFunc
	mu           sync.Mutex
}

type USDT struct {
	Address            common.Address // USDT合约地址
	ListeningAddresses map[string]any // 监听的USDT转账消息
	ABI                abi.ABI        // USDT ABI
	TransferSig        common.Hash    // USDT函数签名
	LogsChan           chan types.Log
}

type RealData struct {
	mu        sync.Mutex
	Heihgt    uint64
	GasLimit  uint64
	GasTipCap *big.Int
	GasFeeCap *big.Int
	GasPrice  *big.Int // 老版本转账使用的gas
}

func NewETHNode(cfg message.ETHConfig, decodeKey string) (*ETHNode, error) {
	// 连入ETH节点的ws
	ws_client, err := ethclient.Dial(cfg.WsURL)
	if err != nil {
		return nil, fmt.Errorf("failed to connect to Ethereum node: %w", err)
	}
	// 连入ETH节点的rpc
	rpc_client, err := ethclient.Dial(cfg.RpcURL)
	if err != nil {
		return nil, fmt.Errorf("failed to connect to Ethereum node rpc: %w", err)
	}
	// 创建可取消的 context
	ctx, cancel := context.WithCancel(context.Background())
	// 获得net_id
	netId, err := rpc_client.NetworkID(ctx)
	if err != nil {
		cancel()
		return nil, fmt.Errorf("failed to connect to get node net_id: %w", err)
	}
	// 构造USDT合约相关
	usdt := &USDT{}
	usdt.Address = common.HexToAddress("0xdAC17F958D2ee523a2206206994597C13D831ec7")        // 解析合约地址
	usdt.ABI = func() abi.ABI { a, _ := abi.JSON(strings.NewReader(erc20ABI)); 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)

	// 初始化数据库
	dbConn, err := db.NewMySQLPool(cfg.DbConfig)
	if err != nil {
		cancel()
		return nil, fmt.Errorf("mysql connect error: %w", err)
	}
	// 初始化结构
	topup := make(map[string]*message.TopupMsg_req)
	withdraw := make(map[string]*message.WithdrawMsg_req)
	pay := make(map[string]*message.PayMsg_req)
	unConfirmTxs := make(map[string]*message.Transaction)
	logsChan := make(chan *types.Header, 1000)
	// 启动时初始化实时数据

	realData := &RealData{}
	return &ETHNode{
		decodeKey:     decodeKey,
		NetID:         netId,
		Config:        cfg,
		ConfirmHeight: cfg.ConfirmHeight,
		WsClient:      ws_client,
		RpcClient:     rpc_client,
		Db:            *dbConn,
		TopupMsg:      topup,
		WithdrawMsg:   withdraw,
		PayMsg:        pay,
		UnConfirmTxs:  unConfirmTxs,
		LogsChan:      logsChan,
		USDT:          usdt,
		RealData:      realData,
		Ctx:           ctx,
		Cancel:        cancel,
	}, nil
}

// ============================ 抽象接口 ============================
func (e *ETHNode) AddAddress(msg any) error {
	switch v := msg.(type) {
	case message.TopupMsg_req:
		e.mu.Lock()
		addr := strings.ToLower(v.Address)
		e.TopupMsg[addr] = &v
		e.mu.Unlock()
	case message.WithdrawMsg_req:
		e.mu.Lock()
		addr := strings.ToLower(v.ToAddress)
		e.WithdrawMsg[addr] = &v
		e.mu.Unlock()
	case message.PayMsg_req:
		e.mu.Lock()
		addr := strings.ToLower(v.FromAddress)
		e.PayMsg[addr] = &v
		e.mu.Unlock()
	default:
		return fmt.Errorf("unsupported message type: %T", msg)
	}
	return nil
}

func (e *ETHNode) RemoveAddress(msg any) error {
	switch v := msg.(type) {
	case message.RemoveListenMsg_req:
		e.mu.Lock()
		addr := strings.ToLower(v.Address)
		delete(e.TopupMsg, addr)
		e.mu.Unlock()
	case message.WithdrawMsg_req:
		e.mu.Lock()
		addr := strings.ToLower(v.ToAddress)
		delete(e.WithdrawMsg, addr)
		e.mu.Unlock()
	case message.PayMsg_req:
		e.mu.Lock()
		addr := strings.ToLower(v.FromAddress)
		delete(e.PayMsg, addr)
		e.mu.Unlock()
	default:
		return fmt.Errorf("unsupported message type: %T", msg)
	}
	return nil
}

func (e *ETHNode) Listen(ch chan any) {
	log.Println("✅ 开始监听 ETH 和 USDT 转账事件...")
	go func() {
		err := e.listenETHTransactions(ch)
		if err != nil {
			log.Fatalf("Listen ETH Transactions Error: %v", err)
		}
	}()

	go func() {
		err := e.listenUSDTTransactions(ch)
		if err != nil {
			log.Fatalf("Listen USDT Transactions Error: %v", err)
		}
	}()
}

// 转账
func (e *ETHNode) Transfer(msg any) error {
	switch v := msg.(type) {
	case message.WithdrawMsg_req:
		// 1. 校验余额
		verifyResult, err := e.verifyBalance(v.Symbol, v.FromAddress, v.Amount)
		if err != nil || !verifyResult {
			log.Printf("address (%s) balance verification failed: %v", v.FromAddress, err)
			return err
		}

		// 2. 构建未签名交易
		unSignTx, err := e.contractTx(v.Symbol, v.FromAddress, v.ToAddress, v.Amount)
		if err != nil {
			log.Printf("failed to create contract transaction: %v", err)
			return err
		}

		// 3. 签名交易
		signedTx, err := e.signTx(unSignTx, v.FromAddress)
		if err != nil {
			log.Printf("failed to sign transaction: %v", err)
			return err
		}
		txHash := signedTx.Hash().Hex()
		// 4. 发送交易并存入交易池
		if err := e.sendTransaction(signedTx, txHash, v.Symbol, v.FromAddress, v.ToAddress, v.Amount); err != nil {
			log.Printf("failed to send transaction: %v", err)
		}
		// 5. 将tx_hash添加到消息中
		if wm, ok := e.WithdrawMsg[v.FromAddress]; ok {
			wm.TxHash = txHash
			e.WithdrawMsg[v.FromAddress] = wm
		} else {
			// 如果不存在原始消息，创建一条新的记录并写回 map（根据实际结构字段调整）
			v.TxHash = txHash
			e.WithdrawMsg[v.FromAddress] = &v
		}
	case message.PayMsg_req:
		// 1. 校验余额
		verifyResult, err := e.verifyBalance(v.Symbol, v.FromAddress, v.TotalAmount)
		if err != nil || !verifyResult {
			log.Printf("address (%s) balance verification failed: %v", v.FromAddress, err)
			return err
		}

		// 2. 预先获取起始 nonce 避免并发冲突
		startNonce, err := e.getTransactionNonce(v.FromAddress)
		if err != nil {
			log.Printf("failed to get start nonce: %v", err)
			return err
		}

		// 3. 按顺序发送多笔交易（注意：顺序发送可以避免 nonce 冲突，但可能较慢）
		i := 0
		for toAddr, tx := range v.Trasnactions {
			// 构建未签名交易（使用递增的 nonce）
			unSignTx, err := e.buildContractTxWithNonce(v.Symbol, v.FromAddress, tx.ToAddress, tx.Amount, startNonce+uint64(i))
			if err != nil {
				log.Printf("failed to create contract transaction: %v", err)
				i++
				continue
			}

			// 签名交易
			signedTx, err := e.signTx(unSignTx, v.FromAddress)
			if err != nil {
				log.Printf("failed to sign transaction: %v", err)
				i++
				continue
			}
			txHash := signedTx.Hash().Hex()

			// 发送交易并存入交易池
			if err := e.sendTransaction(signedTx, txHash, v.Symbol, v.FromAddress, tx.ToAddress, tx.Amount); err != nil {
				log.Printf("failed to send transaction: %v", err)
				i++
				continue
			}

			// 将tx_hash添加到消息中
			e.mu.Lock()
			tx.TxHash = txHash
			e.PayMsg[v.FromAddress].Trasnactions[toAddr] = tx
			e.mu.Unlock()
			i++
		}
	}
	return nil
}

func (e *ETHNode) Stop() {
	if e.Cancel != nil {
		e.Cancel()
	}
	log.Println("🛑 停止监听...")
}

// ============================ rpc节点方法 ============================
func (e *ETHNode) getETHBalance(address string) (*big.Int, error) {
	account := common.HexToAddress(address)
	ctx := context.Background()
	balance, err := e.RpcClient.BalanceAt(ctx, account, nil) // nil表示最新高度
	if err != nil {
		return nil, fmt.Errorf("failed to get eth balance:%w", err)
	}
	// fBalance := new(big.Float).SetInt(balance)
	// ethValue := new(big.Float).Quo(fBalance, big.NewFloat(1e18)) // 转 ETH

	// value, _ := ethValue.Float64() // 转 float64
	return balance, nil
}

func (e *ETHNode) getUSDTBalance(address string) (*big.Int, error) {
	// 统一转换为小写（common.HexToAddress会自动处理，但为了一致性显式转换）
	address = strings.ToLower(address)
	contractAddress := e.USDT.Address
	accountAddress := common.HexToAddress(address)
	data, err := e.USDT.ABI.Pack("balanceOf", accountAddress)
	if err != nil {
		return nil, fmt.Errorf("failed to pack balanceOf data: %w", err)
	}
	msg := ethereum.CallMsg{
		To:   &contractAddress,
		Data: data,
	}
	// 使用 CallContract 方法查询合约余额
	res, err := e.RpcClient.CallContract(e.Ctx, msg, nil)
	if err != nil {
		return nil, fmt.Errorf("failed to get contract balance: %w", err)
	}
	// 解析返回的字节为 *big.Int
	outputs, err := e.USDT.ABI.Unpack("balanceOf", res)
	if err != nil || len(outputs) == 0 {
		return nil, fmt.Errorf("failed to unpack balanceOf result: %w", err)
	}
	balance, ok := outputs[0].(*big.Int)
	if !ok {
		return nil, fmt.Errorf("unexpected type for balanceOf result")
	}
	// bal := utils.BigIntUSDTToFloat64(balance)
	return balance, nil
}

func (e *ETHNode) getTransactionNonce(address string) (uint64, error) {
	nonce, err := e.RpcClient.PendingNonceAt(e.Ctx, common.HexToAddress(address))
	if err != nil {
		log.Fatalf("failed to get nonce: %v", err)
		return 0, err
	}
	return nonce, nil
}

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 (20 * 10^9 wei)
	maxGasPrice := new(big.Int).SetUint64(20000000000) // 20 Gwei

	if gasPrice.Cmp(maxGasPrice) > 0 {
		log.Printf("⚠️ 建议gas price过高 (%v wei)，使用上限 20 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
}

func (e *ETHNode) checkTransaction(tx_hash string) (bool, error) {
	receipt, err := e.RpcClient.TransactionReceipt(e.Ctx, common.HexToHash(tx_hash))
	if err != nil {
		return false, fmt.Errorf("check tx(%s) error: %v", tx_hash, err)
	}
	if receipt.Status == types.ReceiptStatusSuccessful {
		return true, nil
	} else {
		return false, 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(2000000000) // 2 Gwei

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

	// 设置最大费用上限为30 Gwei
	maxFeeLimit := new(big.Int).SetUint64(30000000000) // 30 Gwei
	if maxFeePerGas.Cmp(maxFeeLimit) > 0 {
		log.Printf("⚠️ 计算的最大费用过高 (%v wei)，使用上限 30 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
}

// ============================ 业务逻辑 ============================
// 监听ETH转账
func (e *ETHNode) listenETHTransactions(ch chan any) error {
	fmt.Println("🔍 开始ETH交易...")

	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()
				go e.updateRealData(currentHeight)
				go e.handleETHEvent(header, ch)
				go e.confirm(ch)
			case <-e.Ctx.Done():
				fmt.Println("🛑 收到停止信号，退出ETH交易监听")
				sub.Unsubscribe()
				return e.Ctx.Err()
			}
		}
	reconnect:
	}
}

// 监听USDT转账
func (e *ETHNode) listenUSDTTransactions(ch chan any) error {
	fmt.Println("🔍 开始USDT交易...")
	// 过滤掉非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, ch) // 事件解析 + 分类，传递链消息的通道是vLog而非ch，且一次只传递一笔交易
			case <-e.Ctx.Done():
				fmt.Println("🛑 收到停止信号，退出USDT交易监听")
				sub.Unsubscribe()
				return e.Ctx.Err()
			}
		}
	reconnect:
	}
}

func (e *ETHNode) updateRealData(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.RealData.mu.Lock()
	defer e.RealData.mu.Unlock()
	e.RealData = &RealData{
		Heihgt:    height,
		GasLimit:  gasLimit,
		GasTipCap: maxPriorityFeePerGas,
		GasFeeCap: maxFeePerGas,
		GasPrice:  suggestGasPrice,
	}
	// log.Println("✅ RealData updated successfully.")
}

func (e *ETHNode) handleETHEvent(header *types.Header, ch chan any) {
	height := header.Number.Uint64()

	// 获取区块中的所有交易
	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())

		// 处理充值
		for k, v := range e.TopupMsg {
			if k == toAddr {
				// 锁定并更新未确认的交易
				e.mu.Lock()
				e.UnConfirmTxs[txHash] = &message.Transaction{
					From:   fromAddr,
					To:     toAddr,
					Height: height,
					TxHash: txHash,
					Symbol: v.Symbol,
					Value:  amount,
					Status: STATUS_PENDING,
				}
				e.TopupMsg[k].Status = STATUS_PENDING
				e.mu.Unlock()

				// 创建待确认充值消息
				topup_unconfirm_msg_resp := message.TopupMsg_resp{
					QueueId:     v.QueueId,
					Address:     v.Address,
					Status:      STATUS_PENDING,
					Chain:       v.Chain,
					Amount:      amount,
					TxHash:      txHash,
					BlockHeight: height,
				}

				// 异步发送消息到通道
				go func(msg message.TopupMsg_resp) {
					select {
					case ch <- msg:
						log.Printf("✅ 待确认充值消息已发送")
					default:
						log.Printf("⚠️ 通道阻塞，待确认消息发送失败")
					}
				}(topup_unconfirm_msg_resp)
			}
		}

		// 处理提现
		for k, v := range e.WithdrawMsg {
			if strings.EqualFold(v.FromAddress, fromAddr) && strings.EqualFold(v.ToAddress, toAddr) && v.Amount == amount {
				e.mu.Lock()
				e.UnConfirmTxs[txHash] = &message.Transaction{
					From:   fromAddr,
					To:     toAddr,
					Height: height,
					TxHash: txHash,
					Symbol: v.Symbol,
					Value:  amount,
					Status: STATUS_PENDING,
				}
				e.WithdrawMsg[k].Status = STATUS_PENDING
				e.mu.Unlock()
			}
		}

		// 处理支付
		for k, v := range e.PayMsg {
			if strings.EqualFold(v.FromAddress, fromAddr) {
				for i, pay := range v.Trasnactions {
					if strings.EqualFold(pay.ToAddress, toAddr) && pay.Amount == amount {
						e.mu.Lock()
						e.UnConfirmTxs[txHash] = &message.Transaction{
							From:   fromAddr,
							To:     toAddr,
							Height: height,
							TxHash: txHash,
							Symbol: v.Symbol,
							Value:  amount,
							Status: STATUS_PENDING,
						}
						e.PayMsg[k].Trasnactions[i].Status = STATUS_PENDING
						e.mu.Unlock()
					}
				}
			}
		}
	}
}

func (e *ETHNode) handleUSDTEvent(vLog types.Log, ch chan any) {
	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
	}
	tx_hash := vLog.TxHash.Hex()
	value_float := utils.BigIntUSDTToFloat64(transferEvent.Value)
	var status int = 2
	// 分别验证3组消息
	// 充值
	for k, v := range e.TopupMsg {
		if k == toAddr {

			e.mu.Lock()
			e.UnConfirmTxs[tx_hash] = &message.Transaction{
				From:   fromAddr,
				To:     toAddr,
				Height: height,
				TxHash: tx_hash,
				Symbol: v.Symbol,
				Value:  value_float,
				Status: status,
			}
			e.mu.Unlock()
			topup_unconfirm_msg_resp := message.TopupMsg_resp{
				QueueId:     v.QueueId,
				Address:     v.Address,
				Status:      status,
				Symbol:      v.Symbol,
				Chain:       v.Chain,
				Amount:      value_float,
				TxHash:      tx_hash,
				BlockHeight: height,
			}
			// 异步发送
			go func(msg message.TopupMsg_resp) {
				select {
				case ch <- msg:
					log.Printf("✅ 待确认充值消息已发送")
				default:
					log.Printf("⚠️ 通道阻塞，待确认消息发送失败")
				}
			}(topup_unconfirm_msg_resp)
		}
	}
	// 提现
	for k, v := range e.WithdrawMsg {
		if k == fromAddr && v.ToAddress == toAddr && v.Amount == value_float {
			e.mu.Lock()
			e.UnConfirmTxs[tx_hash] = &message.Transaction{
				From:   fromAddr,
				To:     toAddr,
				Height: height,
				TxHash: tx_hash,
				Symbol: v.Symbol,
				Value:  value_float,
				Status: status,
			}
			e.WithdrawMsg[k].Status = STATUS_PENDING
			e.mu.Unlock()
		}
	}
	// 充值
	for k, v := range e.PayMsg {
		if k == fromAddr {
			for i, pay := range v.Trasnactions {
				if pay.ToAddress == toAddr && pay.Amount == value_float {
					e.mu.Lock()
					e.UnConfirmTxs[tx_hash] = &message.Transaction{
						From:   fromAddr,
						To:     toAddr,
						Height: height,
						TxHash: tx_hash,
						Symbol: v.Symbol,
						Value:  value_float,
						Status: status,
					}
					e.PayMsg[k].Trasnactions[i].Status = STATUS_PENDING
					e.mu.Unlock()
				}

			}
		}
	}
}

func (e *ETHNode) decodePrivatekey(address string) string {
	// 统一转换为小写
	address = strings.ToLower(address)
	// 查询加密后的私钥
	querySql := "SELECT `private_key` FROM eth_balance WHERE address = ? LIMIT 1;"
	log.Println("查询私钥的钱包地址：", address)
	var encryptedKey string
	err := e.Db.QueryRow(querySql, address).Scan(&encryptedKey)
	if err != nil {
		log.Println("❌ 查询私钥失败:", err)
		return ""
	}
	// 使用key解密
	privateKey := encryptedKey // 实际使用时替换成具体的解密代码
	return privateKey
}

// 验证余额
func (e *ETHNode) verifyBalance(symbol, from string, amount float64) (bool, error) {
	var amount_b *big.Int
	e.mu.Lock()
	maxGas := e.RealData.GasFeeCap
	if maxGas == nil {
		return false, fmt.Errorf("chain data not initialized, maxGas = nil")
	}
	e.mu.Unlock()

	switch symbol {
	case "ETH":
		amount_b = utils.Float64ToBigIntETH(amount)
		totalAmount := new(big.Int).Add(amount_b, maxGas)
		ethBalance, err := e.getETHBalance(from)
		if err != nil {
			return false, fmt.Errorf("get (%s) eth-balance error: %v", from, err)
		}
		if ethBalance.Cmp(totalAmount) == -1 {
			return false, nil // 余额不足
		}
		return true, nil
	case "USDT":
		amount_b = utils.Float64ToBigIntUSDT(amount)
		usdtBalance, err := e.getUSDTBalance(from)
		if err != nil {
			return false, fmt.Errorf("get (%s) usdt-balance error: %v", from, err)
		}
		if usdtBalance.Cmp(amount_b) == -1 {
			return false, nil // 余额不足
		}
		return true, nil
	default:
		return false, fmt.Errorf("ETH NetWork error symbol: %s", symbol)
	}
}

// 构建交易
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, from, to, amount, nonce)
}

// 构建交易（指定 nonce）
func (e *ETHNode) buildContractTxWithNonce(symbol, from, to string, amount float64, nonce uint64) (*types.Transaction, error) {
	e.mu.Lock()
	maxFeePerGas, maxPriorityFeePerGas, gasLimit := e.RealData.GasFeeCap, e.RealData.GasTipCap, e.RealData.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)
		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)
	}

	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, txHash, symbol, from, to string, amount float64) error {
	// 发送交易
	// err := e.RpcClient.SendTransaction(e.Ctx, tx)
	// if err != nil {
	// 	return fmt.Errorf("failed to send transaction: %v", err)
	// }

	// 获取当前区块高度
	e.mu.Lock()
	height := e.RealData.Heihgt
	e.mu.Unlock()

	// 将交易存入待交易池
	e.mu.Lock()
	e.UnConfirmTxs[txHash] = &message.Transaction{
		Symbol: symbol,
		From:   from,
		To:     to,
		Height: height,
		TxHash: txHash,
		Value:  amount,
		Status: STATUS_PENDING,
	}
	e.mu.Unlock()
	return nil
}

// 确认信息并返回
func (e *ETHNode) confirm(ch chan any) {
	e.mu.Lock()
	height := e.RealData.Heihgt
	e.mu.Unlock()
	var needSendMsg []any
	var toDeleteTxs []string // 用于记录需要从 UnConfirmTxs 中删除的交易

	for k, v := range e.UnConfirmTxs {
		if v.Height+e.ConfirmHeight >= height {
			tx_result, err := e.checkTransaction(k)
			if err != nil {
				log.Printf("❌ check tx(%s) error: %v", k, err)
				continue
			}

			e.mu.Lock()
			if tx_result {
				// 交易成功
				// 判断是否在充值消息中
				for toAddr, topup_msg := range e.TopupMsg {
					if strings.EqualFold(toAddr, v.To) {
						msg := message.TopupMsg_resp{
							QueueId:     topup_msg.QueueId,
							Address:     v.To,
							Status:      STATUS_SUCCESS,
							Chain:       topup_msg.Chain,
							Symbol:      topup_msg.Symbol,
							Amount:      v.Value,
							TxHash:      v.TxHash,
							BlockHeight: height,
						}
						needSendMsg = append(needSendMsg, msg)
						break // 充值成功后不删除，可能还有后续充值
					}
				}

				// 判断是否在提现消息中
				for key, withdraw_msg := range e.WithdrawMsg {
					if v.TxHash == withdraw_msg.TxHash {
						msg := message.WithdrawMsg_resp{
							QueueId:     withdraw_msg.QueueId,
							Chain:       withdraw_msg.Chain,
							Symbol:      withdraw_msg.Symbol,
							Status:      STATUS_SUCCESS,
							Amount:      v.Value,
							TxHash:      v.TxHash,
							FromAddress: v.From,
							ToAddress:   v.To,
							BlockHeight: v.Height,
						}
						needSendMsg = append(needSendMsg, msg)
						// 删除提现消息
						e.RemoveAddress(withdraw_msg)
						delete(e.WithdrawMsg, key)
						break
					}
				}

				// 判断是否在支付消息中
				for _, pay_msg := range e.PayMsg {
					for payKey, pay := range pay_msg.Trasnactions {
						if v.TxHash == pay.TxHash {
							pay_msg.Trasnactions[payKey].Status = STATUS_SUCCESS
						}
					}
				}
			} else {
				// 交易失败
				// 判断是否在充值消息中
				for toAddr, topup_msg := range e.TopupMsg {
					if strings.EqualFold(toAddr, v.To) {
						msg := message.TopupMsg_resp{
							QueueId:     topup_msg.QueueId,
							Address:     v.To,
							Status:      STATUS_FAILED,
							Chain:       topup_msg.Chain,
							Symbol:      topup_msg.Symbol,
							Amount:      v.Value,
							TxHash:      v.TxHash,
							BlockHeight: height,
						}
						needSendMsg = append(needSendMsg, msg)
						// 充值失败，删除该监听地址
						delete(e.TopupMsg, toAddr)
						break
					}
				}

				// 判断是否在提现消息中
				for key, withdraw_msg := range e.WithdrawMsg {
					if v.TxHash == withdraw_msg.TxHash {
						msg := message.WithdrawMsg_resp{
							QueueId:     withdraw_msg.QueueId,
							Chain:       withdraw_msg.Chain,
							Symbol:      withdraw_msg.Symbol,
							Status:      STATUS_FAILED,
							Amount:      v.Value,
							TxHash:      v.TxHash,
							FromAddress: v.From,
							ToAddress:   v.To,
							BlockHeight: v.Height,
						}
						needSendMsg = append(needSendMsg, msg)
						// 删除提现消息
						e.RemoveAddress(withdraw_msg)
						delete(e.WithdrawMsg, key)
						break
					}
				}

				// 判断是否在支付消息中
				for _, pay_msg := range e.PayMsg {
					for payKey, pay := range pay_msg.Trasnactions {
						if v.TxHash == pay.TxHash {
							pay_msg.Trasnactions[payKey].Status = STATUS_FAILED
						}
					}
				}
			}
			e.mu.Unlock()

			// 标记为删除
			toDeleteTxs = append(toDeleteTxs, k)
		}
	}

	// 删除已确认的交易
	if len(toDeleteTxs) > 0 {
		e.mu.Lock()
		for _, txHash := range toDeleteTxs {
			delete(e.UnConfirmTxs, txHash)
		}
		e.mu.Unlock()
	}

	// 检查支付消息是否需要发送完整响应
	e.mu.Lock()
	var payMsgsToDelete []string
	for key, pay_msg := range e.PayMsg {
		allConfirmed := true
		for _, tx := range pay_msg.Trasnactions {
			if tx.Status == STATUS_PENDING {
				allConfirmed = false
				break
			}
		}
		if allConfirmed {
			// 所有交易都已确认，发送完整响应
			needSendMsg = append(needSendMsg, pay_msg)
			payMsgsToDelete = append(payMsgsToDelete, key)
		}
	}
	// 删除已全部确认的支付消息
	for _, key := range payMsgsToDelete {
		e.RemoveAddress(e.PayMsg[key])
		delete(e.PayMsg, key)
	}
	e.mu.Unlock()

	// 异步发送所有响应消息
	if len(needSendMsg) != 0 {
		for _, data := range needSendMsg {
			go func(msg any) {
				select {
				case ch <- msg:
					log.Printf("✅ confirm message sent: %+v", msg)
				default:
					log.Printf("⚠️ 通道阻塞，待确认消息发送失败")
				}
			}(data)
		}
	}
}