denet-node-server/routes/sdk.js

var router = require('koa-router')();
var moralis = require('../model/moralis_sdk.js')
var utils = require('../model/utils.js');
var { reids_token_config, account_config } = require('../config/config.js');
const logger = require('../model/logger.js');
router.prefix('/sdk');
const redis = require("../model/db/redis_db")  //导入 db.js
const withdraw_db = require("../model/db/withdraw_db")  //导入 db.js
const report = require("../model/report")  //导入 db.js
const BigNumber = require('bignumber.js')
/** 
 * 获取代币价格
 * @param {*} ctx 
 */
async function getAllTotkenPrice(ctx) {
    console.log('getTotkenPrice in:')
    var ret = await moralis.getAllTotkenPrice()
    console.log('getTotkenPrice result:', ret)
    if (ret)
        ctx.body = utils.toJson(0, ret, null);
    else ctx.body = utils.toJson(-1, null, "redis read error.");
}

/**
 * 获取交易记录
 * @param {*} ctx 
 */
async function getTransfers(ctx) {
    const obj = ctx.request.body;
    console.log("getTransfers body", obj);
    if (!obj.chain)//默认 bsc 币安链
        obj.chain = 'bsc_mainnet'
    var temp_obj = { ...obj }
    var index = 0
    //提交归集任务
    if (temp_obj.address) {
        logger.log('pushCollectConisObj>>>', temp_obj.address)
        redis.redis_push(reids_token_config.COLLECT_CONIS_QUEUE_KEY, JSON.stringify(temp_obj))
    }
    // for (let index = 0; index < 30; index++) {
    await moralis.getTokenTransfers(obj).then((result) => {
        logger.log('getTransfers response', 'index=' + index, result)
        ctx.body = result;
        if (result) {
            //提交归集任务 native 能获取到 gas 、token 无法获取到 gas 费
            try {
                if (temp_obj.address && moralis.isTransferSucceed(result)) {
                    var log_obj = { ...obj }
                    log_obj.results = result
                    log_obj.type = report.REPORT_TYPE.transfer_record
                    //埋点日志上报-入金检查
                    report.logReport(log_obj)

                    var json_obj = JSON.parse(result);
                    //缓存当前交易的 gas 费用
                    var tr = moralis.getTransferRecordGasFree('native', json_obj, temp_obj.address)
                    logger.log('getTransferRecordGasFree:', tr, temp_obj.address)
                    if (tr && tr.totalGasFree > 0) {
                        logger.log('getTransferRecordGasFree redis_set LAST_TOTAL_BNB_FREE:', tr.totalGasFree.toString())
                        logger.log('getTransferRecordGasFree redis_set LAST_TOTAL_TOKEN_FREE:', (parseInt(tr.totalGasFree) * parseInt(account_config.TOKEN_GAS_LIMIT)).toString())
                        redis.redis_set(reids_token_config.LAST_TOTAL_BNB_FREE, tr.totalGasFree.toString());
                        redis.redis_set(reids_token_config.LAST_TOTAL_TOKEN_FREE, (parseInt(tr.gas_price) * parseInt(account_config.TOKEN_GAS_LIMIT)).toString());
                    }


                }
            } catch (error) {
                console.error('pushCollectConisObj error=', error)
            }
        }
    })
    // }
}



async function getAllTokenWithdrawInfoLists(ctx) {
    if (ctx.request == null || ctx.request.body == null) {
        ctx.body = utils.toJson(-1, null, "request error. ");
        return
    }
    ctx.body = await moralis.getAllTokenWithdrawInfoLists(ctx);
}

async function collect_conis_task() {
    logger.log("collect_conis_task start")
    while (true) {
        var start_time = utils.getTimestamp()
        var exec_obj = await redis.redis_pop(reids_token_config.COLLECT_CONIS_QUEUE_KEY)
        if (!exec_obj) {
            await utils.sleep(30000)
            logger.log("没有归集任务")
            continue
        }
        try {
            exec_obj = JSON.parse(exec_obj)
            logger.log('collect_conis_task exec item>>>>', exec_obj);

            if (exec_obj.address == await redis.readRedis(reids_token_config.LAST_COLLECT_PUBLIC_KEY) && utils.getTimestamp() - await redis.readRedis(reids_token_config.LAST_COLLECT_TIME) < 60 * 2 * 1000) {
                logger.log('间隔不足 1 分钟', exec_obj);
                continue
            }

            redis.redis_set(reids_token_config.LAST_COLLECT_TIME, utils.getTimestamp())
            redis.redis_set(reids_token_config.LAST_COLLECT_PUBLIC_KEY, exec_obj.address)
            //开始收集用户地址里面的币到归集地址
            var ret = await moralis.collectCoins(exec_obj)
            logger.log('collect_conis_task ret =', exec_obj, ret)
            try {
                var ret_obj = JSON.parse(ret)
                if (ret_obj.code == 0) {
                    logger.log('触发归集 delay collect_conis_task ret =', exec_obj, ret)
                    //间隔 10s 归集，避免提交任务过多
                    await utils.sleep(10000)
                }
            } catch (error) { }
        } catch (error) {
            logger.error('collect_conis_task error', error.toString());
            redis.redis_set(reids_token_config.LAST_COLLECT_TIME, 0)
            redis.redis_set(reids_token_config.LAST_COLLECT_PUBLIC_KEY, 0)
        }
        logger.log("collect_conis_task cost-time", utils.getTimestamp() - start_time, exec_obj)
    }
}


async function withdraw_task() {
    logger.log("withdraw_task start")
    let last_time = 0
    let last_hash = ''

    while (true) {
        var exec_obj = await redis.redis_pop(reids_token_config.WITHDRAW_QUEUE_KEY)
        if (!exec_obj) {
            await utils.sleep(60000)
            logger.log("没有出金任务")
            continue
        }
        try {
            exec_obj = JSON.parse(exec_obj)
        } catch (error) {
            logger.error('item parse error', error);
            break
        }

        var temp_obj = { ...exec_obj }
        if (utils.getTimestamp() - last_time < 60000) {
            //有可能上一个区块还未更新,这里做一个尝试限制
            //Error: Failed to make "eth_sendRawTransaction" request with networkConnector: "already known"
            //通过 交易 hash 获取块。last_hash
            if (last_hash) {
                var options = {
                    transaction_hash: last_hash,
                    chain: temp_obj.chain,
                    endTime: '2099-01-01'
                }
                var tryCount = 10;
                do {
                    try {
                        //通过获取上一个交易记录来进行确认
                        var transaction = await moralis.getTokenTransfers(options);
                        logger.log('withdraw_task exectransaction', transaction, options, tryCount);
                        transaction = JSON.parse(transaction)
                        if (transaction.code == 0) {
                            if (transaction.data.result.length <= 0) {
                                logger.log('等待10s');
                                await utils.sleep(10000)
                            } else {
                                logger.log('等待5s');
                                await utils.sleep(5000)
                                break
                            }
                        } else {
                            break
                        }
                        tryCount -= 1
                    } catch (error) {
                        logger.error('withdraw_task exectransaction', error.toString());
                    }
                } while (tryCount >= 0);
            }
        }

        //如果失败重试一次
        var tryCount = 1;
        for (let index = 0; index < 1 + tryCount; index++) {
            try {
                var result = await withdraw_({ ...temp_obj })
                last_time = utils.getTimestamp()
                logger.log('withdraw_task withdraw_ =', result, last_time)
                if (result && moralis.isTransferSucceed(result)) {
                    var obj = JSON.parse(result)
                    var nonce = obj.data.nonce
                    var curGasPrice = BigNumber(obj.data.gasPrice.hex).toNumber()
                    var curGasLimit = BigNumber(obj.data.gasLimit.hex).toNumber()
                    var value = BigNumber(obj.data.value.hex).toNumber()
                    var hash = obj.data.hash
                    last_hash = hash
                    var update_obj = {}
                    update_obj.withdraw_status = 2
                    update_obj.withdraw_hash = hash
                    update_obj.nonce = nonce
                    update_obj.gas_price = curGasPrice.toString()
                    update_obj.gas_limit = curGasLimit.toString()
                    update_obj.value = value.toString()
                    update_obj.errorMsg = ''
                    await withdraw_db.update_withdraw_task(exec_obj.withdraw_id, update_obj)
                    break
                } else {
                    logger.error('withdraw_task withdraw_ error=', result, JSON.stringify(temp_obj))
                    if (index < 1 + tryCount && result.includes('eth_sendRawTransaction')) {
                        logger.error('try withdraw_:', JSON.stringify(temp_obj), index)
                        await utils.sleep(3000)
                        continue
                    }
                    var update_obj = {}
                    update_obj.withdraw_status = 3
                    if (typeof result === 'string') {
                        try {
                            result = JSON.parse(result)
                            update_obj.errorMsg = result.errMsg
                        } catch (error) {
                            logger.error('withdraw_task=', result)
                        }
                    }
                    await withdraw_db.update_withdraw_task(exec_obj.withdraw_id, update_obj)
                    break
                }
            } catch (error) {
                var update_obj = {}
                update_obj.withdraw_status = 3
                update_obj.errorMsg = error.toString()
                await withdraw_db.update_withdraw_task(exec_obj.withdraw_id, update_obj)
                logger.error('withdraw_task error=', error.toString())
                break
            }
        }
    }
    logger.log("withdraw_task end")
}

/**
 * 队列版本
 * @param {*} ctx 
 * @returns 
 */
async function withdrawV3(ctx) {
    logger.log('withdrawV3')
    if (ctx.request == null || ctx.request.body == null) {
        ctx.body = utils.toJson(-1, null, "request error. ");
        return
    }
    const obj = ctx.request.body;

    // for (let index = 0; index < 10; index++) {
    var log_obj = { ...obj }
    logger.log('withdrawV3', log_obj)
    var obj_ = decrypt_withdraw_content(log_obj.content)
    obj_.withdraw_id = obj_.withdrawId;
    // obj_.withdraw_id = utils.getTimestamp().toString();
    // var obj_ = log_obj
    if (obj_.withdraw_id) {
        var isExist = await withdraw_db.withdraw_id_exist(obj_.withdraw_id)
        if (isExist) {
            logger.error('withdraw_id_exist', obj_.withdraw_id + ' is already in the queue.')
            ctx.body = utils.toJson(-2, null, obj_.withdraw_id + ' is already in the queue.')
            return
        }
        redis.redis_push(reids_token_config.WITHDRAW_QUEUE_KEY, JSON.stringify(obj_))
        var info = await moralis.queryCompanyInfoFromId(0);
        obj_.user_address = info.user_address
        await withdraw_db.create_withdraw_task(obj_)
        // withdraw_task()
        ctx.body = utils.toJson(0, obj_.withdraw_id, null)
    } else {
        return utils.toJson(-2, null, ' withdraw_id not empty.')
    }
    // }
}

async function withdrawV3Test(ctx) {
    logger.log('withdrawV3Test')
    if (ctx.request == null || ctx.request.body == null) {
        ctx.body = utils.toJson(-1, null, "request error. ");
        return
    }
    const obj = ctx.request.body;

    // for (let index = 0; index < 10; index++) {
    var log_obj = { ...obj }
    logger.log('withdrawV3', log_obj)
    var obj_ = decrypt_withdraw_content(log_obj.content)
    obj_.withdraw_id = obj_.withdrawId;
    obj_.withdraw_id = utils.getTimestamp().toString();
    // var obj_ = log_obj
    if (obj_.withdraw_id) {
        var isExist = await withdraw_db.withdraw_id_exist(obj_.withdraw_id)
        if (isExist) {
            logger.error('withdraw_id_exist', obj_.withdraw_id + ' is already in the queue.')
            ctx.body = utils.toJson(-2, null, obj_.withdraw_id + ' is already in the queue.')
            return
        }
        redis.redis_push(reids_token_config.WITHDRAW_QUEUE_KEY, JSON.stringify(obj_))
        var info = await moralis.queryCompanyInfoFromId(0);
        obj_.user_address = info.user_address
        await withdraw_db.create_withdraw_task(obj_)
        ctx.body = utils.toJson(0, obj_.withdraw_id, null)
    } else {
        return utils.toJson(-2, null, ' withdraw_id not empty.')
    }
    // }
}


function decrypt_withdraw_content(content) {
    // const encryptText = utils.encrypt(log_obj);
    const encryptText = content;
    logger.log("加密", encryptText);
    let decryptObj = utils.decrypt(encryptText);
    try {
        logger.log("解密 before", decryptObj);
        decryptObj = JSON.parse(decryptObj);
        console.log("解密 json parse", decryptObj);

    } catch (error) {
        logger.error("json error:", error);
        decryptObj = null;
    }
    return decryptObj;
}

/**
 * 
 * @param {鉴权版本} ctx 
 */
async function withdrawV2(ctx) {
    if (ctx.request == null || ctx.request.body == null) {
        ctx.body = utils.toJson(-1, null, "request error. ");
        return
    }
    const obj = ctx.request.body;
    var log_obj = { ...obj }

    // const encryptText = utils.encrypt(log_obj);
    const encryptText = log_obj.content;
    logger.log("加密", encryptText);

    let decryptObj = utils.decrypt(encryptText);
    try {
        logger.log("解密 before", decryptObj);
        decryptObj = JSON.parse(decryptObj);
        // console.log("解密 json parse", decryptObj);
        await withdraw_(decryptObj).then(result => {
            ctx.body = result;
        })
    } catch (error) {
        logger.error("json error:", error);
        ctx.body = utils.toJson(-1, null, error.toString());
    }
}

async function withdraw_(obj) {
    console.log("withdraw_", obj);
    var log_obj = { ...obj }
    var info = await moralis.queryCompanyInfoFromId(0);
    log_obj.company_address_total_balance_before = await moralis.queryCollectBalance(info.user_address, obj.chain)
    log_obj.company_public_key = info.user_address
    logger.log('withdraw log', log_obj);
    return new Promise((resolve) => {
        moralis.withdraw(obj).then((result) => {
            if (moralis.isTransferSucceed(result)) {
                //提币日志上报
                log_obj.results = result
                log_obj.type = report.REPORT_TYPE.withdraw
                //缓存当前交易的 gas 费用
                if (result && log_obj.contractAddress) {
                    var tr = moralis.getTransferGasFree('token', result)
                    log_obj.withdrawTotalGasFee = tr.totalGasFree.toString()
                } else {
                    var tr = moralis.getTransferGasFree('native', result)
                    log_obj.withdrawTotalGasFee = tr.totalGasFree.toString()
                }
                // log_obj.receiver_address_total_balance_after = await queryCollectBalance(info.user_address, utils.getChainName(obj.chain))
                //日志上报
                report.logReport(log_obj)
            }
            resolve(result)
        });
    })
}

//出金
async function withdraw(ctx) {
    if (ctx.request == null || ctx.request.body == null) {
        ctx.body = utils.toJson(-1, null, "request error. ");
        return
    }
    const obj = ctx.request.body;
    await withdraw_(obj).then(result => {
        ctx.body = result;
    })
}

/**
 * 查询出金状态
 * @param {*} ctx 
 */
async function getWithdrawStatus(ctx) {
    if (ctx.request == null || ctx.request.body == null) {
        ctx.body = utils.toJson(-1, null, "request error. ");
        return
    }
    const obj = ctx.request.body;
    var info = await withdraw_db.queryWithdrawInfoFromWithdrawId(obj.withdrawId)
    logger.log('getWithdrawStatus info', JSON.stringify(info))
    if (info) {
        if (info.withdraw_status != 3) {
            ctx.body = utils.toJson(0, {
                withdrawId: info.withdraw_id,
                withdrawStatus: info.withdraw_status,
                withdrawHash: info.withdraw_hash,
                chainId: info.chain_id,
                transferTimestamp: info.update_time,
            }, null)
        } else {
            ctx.body = utils.toJson(0, {
                withdrawId: info.withdraw_id,
                withdrawStatus: info.withdraw_status,
                withdrawHash: info.withdraw_hash,
                chainId: info.chain_id,
                transferTimestamp: info.update_time,
                errorMsg: info.errorMsg
            }, null)
        }
    } else {
        ctx.body = utils.toJson(-1, null, obj.withdraw_id + ' id does not exist.')
    }
}

//获取交易记录
router.post('/getTransfers', getTransfers)
// 获取所有代币价格
router.post('/getAllTotkenPrice', getAllTotkenPrice)

// router.post('/transfer', transfer)
//提现
router.post('/withdraw', withdraw);
//提现鉴权-body 加密
router.post('/withdrawV2', withdrawV2);
//队列的形式
router.post('/withdrawV3', withdrawV3);

if (process.env.NODE_ENV == 'dev')
    router.post('/withdrawV3Test', withdrawV3Test);

//查询出金服务
router.post('/getWithdrawStatus', getWithdrawStatus);
//获取所有地址的所要消耗的最低提取费
router.post('/getAllTokenWithdrawInfoLists', getAllTokenWithdrawInfoLists)

// 定时任务 提币+归集
withdraw_task();
collect_conis_task();


module.exports = router