钱包业务层 - 5. 实现交易所内部交易（归集、转冷、转热）业务

归集、热转冷、冷转热业务实现

归集业务、热转冷业务、冷转热业务在我们交易所中，可以将其归为一大类。因为这类的交易， 只需要交易所掌控的地址之间进行交互集合，无须与外部地址进行交易（充值、提现需要和外部地址进行交互） 所以，我们称这类业务为 Internal 内部交易。 下面是这三种交易的区别：

归集：from 地址为用户地址，to 地址为热钱包地址（归集地址）

热转冷： from 地址为热钱包地址，to 地址为冷钱包地址

冷转热：from 地址为冷钱包地址，to 地址为热钱包地址

完整项目 github 地址（如果对您有用，请给个小 star ⭐️）：

1. exchange-wallet-service：钱包业务层服务： <https://github.com/Shawn-Shaw-x/exchange-wallet-service>
2. signature-machine：离线签名机：  <https://github.com/Shawn-Shaw-x/signature-machine>
3. chains-union-rpc：多链统一 rpc：  <https://github.com/Shawn-Shaw-x/chains-union-rpc>

交易所内归集交易的步骤

在交易所内，为了保证资金的安全，降低资金被盗的风险（热钱包地址安全级别更高），以及降低对账、提现等业务的难度。 通常来讲，会做一个资金的归集过程，也就是说：交易所会采取一系列的策略去将大量的用户地址的资金归集到一个归集地址上面去。 一般来说，归集的触发会有一个 “用户最小充值资金” 的概念，如果说用户充值金额很小， 交易所有可能考虑到手续费的磨损、归集频繁程度，可能不会对小额充值进行归集。

归集业务有几种实现方式：

1. 批量归集 对于某些链，原生支持批量归集的操作。例如 UTXO 模型的链（如比特币），支持多个 UTXO 的输入，单个归集地址的输出。 这样就能原生实现链上的批量归集业务了。

2. 单笔归集 但是对于某些链，比如说非合约地址作为用户地址的以太坊，并不是原生支持批量转账的（Pectra 升级之前，升级之后可以批量转账）， 在这种情况下，交易所对这种链的归集只能对进行每个用户地址进行单笔交易的归集。

• 提问：归集业务中，如果某个用户地址没有主币，那手续费谁来付？ 一般来说，有两种情况：

  1. 一种情况是这个链具有原生支持 gas 代付的能力（例如 solana），那么，这个手续费只需要交易所在进行归集操作时，指定一个代付账号地址即可。

  2. 另一种情况是这个链不原生支持 gas 代付的能力（例如 Pectra 升级之前的以太坊），那么， 这个手续费必须先由交易所的某个地址下发到需要归集的用户地址再进行归集的操作。

• 提问：某些链不需要归集，是怎么实现的？

某些链，可以在交易的时候携带 Tag（memo），这样的链无需进行归集的操作。因为：

1. 用户充值时候，将交易所的用户 id 填写到这个 Tag（memo）字段中。

2. 充值的资金直接打到交易所的热钱包地址中，无需给用户分配用户地址。

3. 交易所扫链发现这笔交易，将充值的资金分配给这个交易所的用户 id 即可。

内部交易的整体流程图

详细解释见下面交易所交易的实现

内部交易的实现

内部交易的实现实际上和我们的提现业务非常类似，都是由项目方发起， 且都是需要启动定时任务去扫描数据库中的已签名交易，发送到区块链网络中。 下面我来介绍下详细的步骤：

1. 项目方调用钱包业务层，生成未签名交易，获得 transactionId 和 32 字节的 messageHash

2. 项目方使用 messageHash 调用自己部署的签名机，签名这笔交易。

3. 项目方使用 transactionId 和 signature 去钱包层构建已签名交易。（钱包层会保存到数据库中）

4. 钱包层启动定时任务，扫描数据库中的内部交易（归集、转冷、转热），发送到区块链网络中，交易状态为已广播。

5. 钱包层的扫链同步器、交易发现器发现这笔内部交易，更新交易的状态为完成。

/*
启动内部交易处理任务
处理归集、热转冷、冷转热
交易的发送到链上，更新库、余额
*/
func (in *Internal) Start() error {
    log.Info("starting internal worker.......")
    in.tasks.Go(func() error {
        for {
            select {
            case <-in.ticker.C:
                log.Info("starting internal worker...")
                businessList, err := in.db.Business.QueryBusinessList()
                if err != nil {
                    log.Error("failed to query business list", "err", err)
                    continue
                }
                for _, business := range businessList {
                    /*分项目方处理*/
                    unSendTransactionList, err := in.db.Internals.UnSendInternalsList(business.BusinessUid)
                    if err != nil {
                        log.Error("failed to query unsend internals list", "err", err)
                        continue
                    }
                    if unSendTransactionList == nil || len(unSendTransactionList) <= 0 {
                        log.Error("failed to query unsend internals list", "err", err)
                        continue
                    }

                    var balanceList []*database.Balances

                    for _, unSendTransaction := range unSendTransactionList {
                        /*分单笔交易发送*/
                        txHash, err := in.rpcClient.SendTx(unSendTransaction.TxSignHex)
                        if err != nil {
                            log.Error("failed to send internal transaction", "err", err)
                            continue
                        } else {
                            /*发送成功, 处理from 地址余额*/
                            balanceItem := &database.Balances{
                                TokenAddress: unSendTransaction.TokenAddress,
                                Address:      unSendTransaction.FromAddress,
                                LockBalance:  unSendTransaction.Amount,
                            }
                            /*todo 缺少 to 地址的余额处理？*/

                            balanceList = append(balanceList, balanceItem)

                            unSendTransaction.TxHash = common.HexToHash(txHash)
                            unSendTransaction.Status = constant.TxStatusBroadcasted
                        }
                    }
                    retryStrategy := &retry.ExponentialStrategy{Min: 1000, Max: 20_000, MaxJitter: 250}
                    if _, err := retry.Do[interface{}](in.resourceCtx, 10, retryStrategy, func() (interface{}, error) {
                        if err := in.db.Gorm.Transaction(func(tx *gorm.DB) error {
                            /*处理内部交易余额*/
                            if len(balanceList) > 0 {
                                log.Info("Update address balance", "totalTx", len(balanceList))
                                if err := in.db.Balances.UpdateBalanceListByTwoAddress(business.BusinessUid, balanceList); err != nil {
                                    log.Error("Update address balance fail", "err", err)
                                    return err
                                }

                            }
                            /*保存内部交易状态*/
                            if len(unSendTransactionList) > 0 {
                                err = in.db.Internals.UpdateInternalListById(business.BusinessUid, unSendTransactionList)
                                if err != nil {
                                    log.Error("update internals status fail", "err", err)
                                    return err
                                }
                            }
                            return nil
                        }); err != nil {
                            log.Error("unable to persist batch", "err", err)
                            return nil, err
                        }
                        return nil, nil
                    }); err != nil {
                        return err
                    }
                }

            case <-in.resourceCtx.Done():
                log.Info("worker is shutting down")
                return nil
            }
        }
    })
    return nil
}

归集测试

1. 构建未签名交易

2. 签名机签名

3. 构建已签名交易

4. 归集前余额

5. 启动同步器、发现器、内部交易定时任务后查看余额变化

热转冷测试

1. 交易构建和签名过程和之前的测试一样，这里省略...

2. 热转冷前的余额

3. 热转冷后的余额

冷转热测试

1. 交易构建和签名过程和之前的测试一样，这里省略...

2. 冷转热之前的余额

3. 冷转热之后的余额