game2006web3/server/web3bcspider/services/erc721_refresher.js

const app = require('j7/app');
const utils = require('j7/utils');
const bcutils = require('j7/bcutils');
const log = require('j7/log');
const constant = require('common/constant');
const dbpool = require('common/dbpool');
const BaseService = require('./baseservice');

const LIMIT_COUNT = 100;

class Erc721Refresher extends BaseService {

  async init(bc, net, refresher) {
    this.bc = bc;
    this.net = net;
    this.refresher = refresher;
    this.lastIdx = BigInt(0);
    this.conf = this.refresher['conf'];
    this.progInfo = this.refresher['progressInfo'];
    await this.start();
  }

  async start() {
    while (true) {
      await this.pullEvent();
      await utils.sleep(2000 + utils.randRange(500, 1500));
    }
  }

  async pullEvent() {
    const logHead = this.genLogHead('pullEvent ');
    try {
      const {err, rows} = await dbpool.execBcEventConn(
        app,
        'execQuery',
        ',SELECT * FROM t_erc721_refresh WHERE idx > ' + this.lastIdx.toString() + ' ' +
          'AND net_id = ? AND contract_address = ? AND status = 0 ' +
          'LIMIT ' + LIMIT_COUNT,
        [
          this.getNetId(),
          this.getContractAddress()
        ]);
      if (err) {
        throw new Error(err);
      }
      if (rows.length > 0) {
        this.progInfo['pendingCount'] = rows.length;
        await utils.serial(
          rows,
          async (row) => {
            const nftExists = await this.tokenIsExists(row);
            if (nftExists) {
              await this.refresh(row);
              --this.progInfo['pendingCount'];
            } else {
              ++this.progInfo['skipCount'];
            }
          }
        );
      } else {
        this.lastIdx = BigInt(0);
      }
      ++this.progInfo['refreshedCount'];
      this.progInfo['pendingCount'] = 0;
      this.progInfo['skipCount'] = 0;
    } catch (err) {
      utils.safeDumpErrStack(err);
      log.error(logHead + err);
      await utils.sleep(5000 + utils.randRange(1000, 3000));
    }
  }

  getNetId() {
    return this.net['net_id'];
  }

  getContractAddress() {
    return this.bc.getContractAddressByName(this.getContractName());
  }

  getContractName() {
    return this.conf['contract_name'];
  }

  genLogHead(msg) {
    const head = this.getNetId() + '.' + this.getContractName() + ' erc721 refresher ' + msg;
    return head;
  }

  async refresh(row) {
    while (true) {
      await this.bc.mustBeActive();
      const oldBlockNumber = this.bc.getCurrBlockNumber();
      let oldOwner = await this.getOwner(row['token_id']);
      while (oldBlockNumber + 8 > this.bc.getCurrBlockNumber()) {
        await utils.sleep(1000 + utils.randRange(0, 500));
      }

      await this.bc.mustBeActive();
      const newBlockNumber = this.bc.getCurrBlockNumber();
      let newOwner = await this.getOwner(row['token_id']);

      if (oldOwner == newOwner) {
        const ok = await this.updateConfirmed(newOwner, oldBlockNumber, row);
        if (ok) {
          await this.update
          (row,
           [
             ['status', 1]
           ]);
          break;
        }
      }
      await utils.sleep(1000 + utils.randRange(500, 1500));
    }
  }

  async getOwner(tokenId) {
    while (true) {
      await this.bc.lockQuery();
      try {
        let owner = await this.bc.ownerOf721(this.getContractName(), tokenId);
        return owner;
      } catch (err) {
        const reason = utils.getVal(err, 'reason');
        if (reason == 'ERC721: owner query for nonexistent token') {
          return '';
        }
        if (err == 'Error: Returned error: VM Exception while processing transaction: revert ERC721: owner query for nonexistent token' ||
           err == 'Error: Returned error: execution reverted: ERC721: owner query for nonexistent token') {
          return '';
        }
        log.error(err);
      } finally {
        await this.bc.unlockQuery();
      }
      await utils.sleep(5000 + utils.randRange(1500, 2500));
    }
  }


  async updateConfirmed(newOwner, blockNumber, row) {
    const logHead = this.genLogHead('updateConfirmed ');
    const tokenId = row['token_id'];
    const {err} = await dbpool.execBcNftConn(
      app,
      'update',
      't_nft',
      [
        ['net_id', row['net_id']],
        ['token_id', row['token_id']],
        ['contract_address', row['contract_address']],
      ],
      [
        ['owner_address', bcutils.toNormalAddress(newOwner)],
        ['!confirm_count', () => {
          return 'confirm_count + 1';
        }],
        ['confirm_block_number', blockNumber]
      ]);
    if (err) {
      utils.safeDumpErrStack(err);
      log.error(logHead + tokenId + ' err:' + err);
      return false;
    }
    return true;
  }

  async tokenIsExists(rawRow) {
    const logHead = this.genLogHead('tokenIsExists ');
    const tokenId = rawRow['token_id'];
    const {err, row} = await dbpool.execBcNftConn(
      app,
      'ormSelectOne',
      't_nft',
      [
        ['net_id', rawRow['net_id']],
        ['token_id', rawRow['token_id']],
        ['contract_address', rawRow['contract_address']],
      ]);
    if (err) {
      utils.safeDumpErrStack(err);
      log.error(logHead + tokenId + ' err:' + err);
      return false;
    }
    if (row) {
      return true;
    }
    return false;
  }

  async update(row, fields) {
    const {err} = await this.dbpool.execBcEventConn(
      app,
      'update',
      't_erc721_refresh',
      [
        ['idx', row['idx']],
        ['net_id', row['net_id']],
        ['token_id', row['token_id']],
        ['contract_address', row['contract_address']],
        ['refresh_count', row['refresh_count']],
      ],
      fields,
    );
    const lastIdx = BigInt(row['idx']);
    if (lastIdx > this.lastIdx) {
      this.lastIdx = lastIdx;
    }
  }

}

module.exports = Erc721Refresher;