game2006web3/server/web3dbspider/services/dbevent_process.js

const app = require('j7/app');
const utils = require('j7/utils');
const bcutils = require('j7/bcutils');
const log = require('j7/log');
const BaseService = require('./baseservice');
const metaFactory = require('../metadata/factory');

class DbEventProcess extends BaseService {

  static #tableMaxIdxHash = {};

  static async staticInit() {
    const {err, conn} = await app.getDbConn('BcEventDb0');
    const tables = metaFactory.getAllTables();
    await utils.serial
    (
      Object.values(tables),
      async (tblObj) => {
        {
          const {err, maxIdx} = await conn.getMaxIdx(tblObj['table_name']);
          if (err) {
            throw 'DbEventProcess error:' + err;
          }
          DbEventProcess.#tableMaxIdxHash[tblObj['table_name']] = maxIdx;
        }
        const updateMaxIdxFunc = async () => {
          while (true) {
            const {err, maxIdx} = await conn.getMaxIdx(tblObj['table_name']);
            if (!err) {
              DbEventProcess.#tableMaxIdxHash[tblObj['table_name']] = maxIdx;
              await utils.sleep(500 + utils.randRange(500, 1500));
            } else {
              await utils.sleep(5000 + utils.randRange(500, 1500));
            }
          }
        };
        updateMaxIdxFunc();
      }
    );
  }

  async init(net, event) {
    const {err, conn} = await app.getDbConn('BcEventDb0');
    this.conn = conn;
    this.net = net;
    this.event = event;
    this.lastIdx = BigInt(0);
    this.eventConf = this.event['eventConf'];
    this.progInfo = this.event['progressInfo'];
    this.contractAddress = this.net.getContractAddressByName(this.getContractName());
    this.lastIdx = await this.getLastIdx();
    this.progInfo['proclastIdx'] = this.lastIdx.toString();
    await this.start();
  }

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

  async pullEvent() {
    const logHead = this.getInstanceName() + ' pullDbEvent: ';
    const tableName = this.getTableName();
    try {
      const startIdx = this.getStartIdx();
      const endIdx = this.getEndIdx();
      ++this.progInfo['procPullCount'];
      if (startIdx >= endIdx) {
        return;
      }
      const {err, rows} = await this.conn.execQuery(
        'SELECT * FROM ${tableName} WHERE `idx` > ? AND `idx` <= ? ' +
          'AND net_id = ? AND event_name = ? AND contract_address = ? ' +
          'LIMIT ' + LIMIT_COUNT,
        [
          startIdx.toString(),
          endIdx.toString(),
          this.getNetId(),
          this.getEventName(),
          this.getContractAddress()
        ]);
      if (err) {
        throw err;
      }
      if (rows.length > 0) {
        await utils.serial(
          rows,
          async (row) => {
            await this.saveToDb(row);
          }
        );
        this.progInfo['procEventCount'] += rows.length;
      } else {
        this.lastIdx = endIdx;
      }
      this.progInfo['procLastIdx'] = this.lastIdx.toString();
      await this.saveLastIdx(this.lastIdx);
    } catch (err) {
        log.error(logHead + err);
        await utils.sleep(5000 + utils.randRange(1000, 3000));
    }
  }

  getStartIdx() {
    return this.lastIdx;
  }

  getEndIdx() {
    return this.getMaxIdx();
  }

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

  getEventName() {
    return this.eventConf['event_name'];
  }

  getContractAddress() {
    return this.contractAddress;
  }

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

  getInstanceName() {
    const instName = this.getNetId() + '.' + this.getContractName() + '.' + this.getEventName();
    return instName;
  }

  getTableName() {
    return this.eventConf['table_name'];
  }

  async getLastIdx() {
    const logHead = this.getInstanceName() + ' getLastIdx: ';
    while (true) {
      const {err, row} = await this.conn.ormSelectOne(
        't_dbprocess_last_idx',
        [
          ['net_id', this.getNetId()],
          ['contract_address', this.getContractAddress()],
          ['event_name', this.getEventName()],
        ]
      );
      if (err) {
        log.error(logHead + err);
        await utils.sleep(5000 + utils.randRange(500, 1500));
        continue;
      }
      if (row) {
        return BigInt(row['last_idx']);
      }
    }
    return BigInt(0);
  }

  async saveLastIdx(lastIdx) {
    const logHead = this.getInstanceName() + ' saveLastIdx: ';
    while (true) {
      const {err} = await this.conn.upsert(
        't_dbprocess_last_idx',
        [
          ['net_id', this.getNetId()],
          ['contract_address', this.getContractAddress()],
          ['event_name', this.getEventName()],
        ],
        [
          ['last_idx', lastIdx.toString()],
          ['modifytime', utils.getUtcTime()],
        ],
        [
          ['net_id', this.getNetId()],
          ['contract_address', this.getContractAddress()],
          ['event_name', this.getEventName()],
          ['contract_name', this.getContractName()],
          ['last_idx', lastIdx],
          ['createtime', utils.getUtcTime()],
          ['modifytime', utils.getUtcTime()],
        ]
      );
      if (!err) {
        break;
      }
      log.error(logHead + err);
      await utils.sleep(5000 + utils.randRange(500, 1500));
    }
  }

  getMaxIdx() {
    if (utils.hasKey(DbEventProcess.#tableMaxIdxHash, this.getTableName())) {
      return DbEventProcess.#tableMaxIdxHash[this.getTableName()];
    } else {
      return BigInt(0);
    }
  }

}

module.exports = DbEventProcess;