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')
const czz = require('../model/czz')
/** 
 * 获取代币价格
 * @param {*} ctx 
 */
async function getAllTotkenPrice(ctx) {
    console.log('getTotkenPrice in:')
    var ret = await moralis.getAllTotkenPrice(ctx.request.body)
    console.log('getTotkenPrice result:', ret)
    if (ret)
        ctx.body = utils.toJson(0, ret, null);
    else ctx.body = utils.toJson(-1, null, "redis read error.");
}

async function getAllTokenPrice(ctx) {
    var ret = await moralis.getAllTotkenPrice(ctx.request.body)
    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

    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_PRICE:', tr)
                        // redis.redis_set(reids_token_config.LAST_BNB_PRICE, tr.gas_price.toString());
                        // redis.redis_set(reids_token_config.LAST_TOKEN_PRICE, tr.gas_price.toString());
                        redis.writeAppendRedis(reids_token_config.LAST_BNB_PRICE, temp_obj.chain, '', tr.gas_price.toString());
                        redis.writeAppendRedis(reids_token_config.LAST_TOKEN_PRICE, temp_obj.chain, '', tr.gas_price.toString());
                    }
                    if (json_obj.data.total > 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))
                        }
                    }
                    if (json_obj.data.total > 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))
                        }
                    }

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

/**
 * 获取交易记录
 * @param {*} ctx 
 */
async function getTransfersV2(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

    await moralis.getTokenTransfersV2(obj).then((result) => {
        logger.log('getTokenTransfersV2 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)
                        // redis.redis_set(reids_token_config.LAST_BNB_PRICE, tr.gas_price.toString());
                        // redis.redis_set(reids_token_config.LAST_TOKEN_PRICE, tr.gas_price.toString().toString());
                        redis.writeAppendRedis(reids_token_config.LAST_BNB_PRICE, temp_obj.chain, '', tr.gas_price.toString());
                        redis.writeAppendRedis(reids_token_config.LAST_TOKEN_PRICE, temp_obj.chain, '', tr.gas_price.toString());
                    }

                    if (json_obj.data.total > 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))
                        }
                    }
                }
            } 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 check_czz_withdraw_task() {
    while (true) {
        var exec_obj;
        try {
            exec_obj = await redis.redis_pop(reids_token_config.CHECK_CZZ_WITHDRAW_STATUS_QUEUE)
            logger.log("check_czz_withdraw_task redis_pop",exec_obj)
            if (!exec_obj) {
                logger.log("没有 czz hash check")
                await utils.sleep(2 * 60 * 1000)
                continue
            }
            exec_obj = JSON.parse(exec_obj)
            if (utils.getTimestamp() - exec_obj.create_time > exec_obj.lifecycle) {
                logger.error('已过期 check_czz_withdraw_task ：', JSON.stringify(exec_obj))
                var update_obj = {}
                update_obj.withdraw_status = 3
                update_obj.withdraw_hash = ''
                update_obj.nonce = -1
                update_obj.gas_price = ''
                update_obj.gas_limit = ''
                update_obj.value = '0'
                update_obj.errorMsg = 'czz timeout'
                await withdraw_db.update_withdraw_task(exec_obj.withdraw_id, update_obj)
                continue
            }

            var obj = await czz.check_withdraw_status(exec_obj)
            if (obj.code == 0) {
                var nonce = obj.data.nonce
                var curGasPrice = obj.data.gasPrice.toString()
                var curGasLimit = obj.data.gasLimit.toString()
                var hash = obj.data.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 = utils.scientificNotationToString(obj.data.value).toString()
                update_obj.errorMsg = ''
                await withdraw_db.update_withdraw_task(exec_obj.withdraw_id, update_obj)
            } else {
                redis.redis_push(reids_token_config.CHECK_CZZ_WITHDRAW_STATUS_QUEUE, JSON.stringify(exec_obj))
            }
            await utils.sleep(2 * 60 * 1000)
        } catch (error) {
            logger.error('check_czz_withdraw_task error', error.toString(), JSON.stringify(exec_obj))
        }
    }
}

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);
            //开始收集用户地址里面的币到归集地址
            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());
        }
        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 = ''
    let last_chain = ''

    while (true) {
        var exec_obj = await redis.redis_pop(reids_token_config.WITHDRAW_QUEUE_KEY)
        if (!exec_obj) {
            await utils.sleep(10000)
            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 && last_chain) {
                var options = {
                    transaction_hash: last_hash,
                    chain: last_chain,
                    endTime: '2099-01-01'
                }
                var tryCount = 3;
                do {
                    try {
                        //通过获取上一个交易记录来进行确认
                        var transaction = await moralis.getTokenTransfersV2(options);
                        logger.log('withdraw_task exectransaction', transaction, options, tryCount);
                        if (typeof transaction === 'string')
                            transaction = JSON.parse(transaction)
                        if (transaction.code == 0) {
                            if (transaction.data.results.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 err', error.toString());
                    }
                    if (tryCount < 0) {
                        logger.error('withdraw_task getTokenTransfersV2 警告交易未更新:', JSON.stringify(options));
                    }
                } 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.getTransferCode(result) == 0) {
                    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 = 0
                    if (obj.data.chainId == 97) {
                        value = BigNumber(obj.data.value.hex).toNumber()
                    } else {
                        value = obj.data.value.number
                    }
                    var hash = obj.data.hash
                    last_hash = hash
                    last_chain = temp_obj.chain
                    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 = utils.scientificNotationToString(value).toString()
                    update_obj.errorMsg = ''
                    await withdraw_db.update_withdraw_task(exec_obj.withdraw_id, update_obj)
                    break
                } else if (result && moralis.getTransferCode(result) == 1) {
                    moralis.pushChainDetailTOQueue(1, exec_obj, result);
                    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
        }

        var info = await moralis.queryCompanyInfoFromId(0);
        obj_.user_address = info.user_address
        await withdraw_db.create_withdraw_task(obj_)
        redis.redis_push(reids_token_config.WITHDRAW_QUEUE_KEY, JSON.stringify(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 = utils.getCurrentDateFormat('YYYY-MM-DD-HH:mm:ss:SSS').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.getTransferCode(result) == 0) {
                //提币日志上报
                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()
                }
                //日志上报
                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.')
    }
}


async function timer_collect_conis_bsc_task() {
    var index = 0
    var delay = 60 * 1000 * 60
    while (1) {
        var temp_obj = {
            "chain": "bsc_testnet",
            "address": "0x3B525c35DdC323B08241493f148340D89e3A73a7"
        }
        redis.redis_push(reids_token_config.COLLECT_CONIS_QUEUE_KEY, JSON.stringify(temp_obj))
        await utils.sleep(delay)
        index += 1
    }
}

async function timer_collect_conis_czz_task() {
    var index = 0
    var delay = 60 * 1000 * 60
    while (1) {
        var temp_obj = {
            "chain": "czz",
            "address": "0x39ACD9CC975D792D8160215Dc84fa00E4934F076",
        }
        redis.redis_push(reids_token_config.COLLECT_CONIS_QUEUE_KEY, JSON.stringify(temp_obj))
        await utils.sleep(delay)
        index += 1
    }
}

async function timer_transfer_bsc_task() {
    var index = 0
    var delay = 60 * 1000 * 60
    while (1) {
        var obj_ = {
            "type": "erc20",
            "contractAddress": "0xFF94950Ee8A79c52cC4B0Aa5178C8cEa48A3F3A6",
            "amount": "123000000000000000000",
            "chain": "bsc_testnet",
            "receiver": "0x3B525c35DdC323B08241493f148340D89e3A73a7",
            "withdrawId": index.toString()
        }
        obj_.withdraw_id = utils.getCurrentDateFormat('YYYY-MM-DD-HH:mm:ss:SSS').toString()
        var info = await moralis.queryCompanyInfoFromId(0);
        obj_.user_address = info.user_address
        await withdraw_db.create_withdraw_task(obj_)

        redis.redis_push(reids_token_config.WITHDRAW_QUEUE_KEY, JSON.stringify(obj_))
        await utils.sleep(delay)
        index += 1
    }
}

async function timer_transfer_czz_task() {
    var index = 0
    var delay = 60 * 1000 * 60
    while (1) {
        var obj_ = {
            "type": "erc20",
            "contractAddress": "0xfb16179d5e84b0e3e7524ed61a9cf7b98d039b20",
            "amount": "100000000000000000000",
            "chain": "czz",
            "receiver": "0x39ACD9CC975D792D8160215Dc84fa00E4934F076"
        }
        obj_.withdraw_id = utils.getCurrentDateFormat('YYYY-MM-DD-HH:mm:ss:SSS').toString()
        var info = await moralis.queryCompanyInfoFromId(0);
        obj_.user_address = info.user_address
        await withdraw_db.create_withdraw_task(obj_)
        redis.redis_push(reids_token_config.WITHDRAW_QUEUE_KEY, JSON.stringify(obj_))
        await utils.sleep(delay)
        index += 1
    }
}

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

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

if (process.env.NODE_ENV == 'dev' || process.env.NODE_ENV == 'test') {
    router.post('/withdrawV3Test', withdrawV3Test);
}
//查询出金服务
router.post('/getWithdrawStatus', getWithdrawStatus);
//获取所有地址的所要消耗的最低提取费
router.post('/getAllTokenWithdrawInfoLists', getAllTokenWithdrawInfoLists)

// 定时任务 提币+归集
withdraw_task();
collect_conis_task();
//czz 504 检查
check_czz_withdraw_task();

if (process.env.NODE_ENV == 'dev' || process.env.NODE_ENV == 'test') {
    timer_transfer_bsc_task()
    timer_transfer_czz_task()
    timer_collect_conis_bsc_task()
    timer_collect_conis_czz_task()
}

module.exports = router