feat: create from, to account id column; create job update it (#916)

ci/config.json.ci

@@ -527,5 +527,10 @@
   },
   "crawlEvmAccountBalanceNonce": {
     "concurrency": 100
+  },
+  "updateAccountIdInEVMTransaction": {
+    "key": "updateAccountIdInEVMTransaction",
+    "blocksPerCall": 100,
+    "millisecondCrawl": 1000
   }
 }

config.json

@@ -526,5 +526,10 @@
   },
   "crawlEvmAccountBalanceNonce": {
     "concurrency": 100
+  },
+  "updateAccountIdInEVMTransaction": {
+    "key": "updateAccountIdInEVMTransaction",
+    "blocksPerCall": 100,
+    "millisecondCrawl": 1000
   }
 }

migrations/evm/20241003034314_create_from_to_account_id_in_evm_tx.ts

@@ -0,0 +1,14 @@
+import { Knex } from 'knex';
+
+export async function up(knex: Knex): Promise<void> {
+  await knex.schema.alterTable('evm_transaction', (table) => {
+    table.integer('from_account_id').index();
+    table.integer('to_account_id').index();
+  });
+}
+
+export async function down(knex: Knex): Promise<void> {
+  await knex.schema.alterTable('evm_transaction', (table) => {
+    table.dropColumns('from_account_id', 'to_account_id');
+  });
+}

src/common/utils/db_connection.ts

@@ -29,7 +29,12 @@ export async function batchUpdate(
           }
           return '{}';
         }
-
+        if (record[field]?.type === 'number') {
+          if (record[field].value !== undefined) {
+            return record[field].value;
+          }
+          return 'NULL';
+        }
         if (record[field] !== undefined) {
           return `'${record[field]}'`;
         }

src/services/evm/constant.ts

@@ -187,6 +187,10 @@ export const SERVICE = {
         key: 'CreateConstraintInEVMBlockPartition',
         path: 'v1.CreateConstraintInEVMBlockPartition',
       },
+      UpdateAccountIdInEVMTransaction: {
+        key: 'UpdateAccountIdInEVMTransaction',
+        path: 'v1.UpdateAccountIdInEVMTransaction',
+      },
     },
     CrawlEvmBlock: {
       key: 'CrawlEvmBlock',
@@ -297,6 +301,7 @@ export const BULL_JOB_NAME = {
   CRAWL_EVM_ACCOUNT_PUBKEY: 'crawl:evm-account-pubkey',
   UPDATE_EVM_ASSETS: 'update:evm-assets',
   CRAWL_EVM_ACCOUNT_BALANCE_NONCE: 'crawl:evm-account-balance-nonce',
+  UPDATE_ACCOUNT_ID_IN_EVM_TX: 'update:account-id-in-evm-tx',
 };
 
 export const MSG_TYPE = {

src/services/evm/job/update_account_id_in_evm_transaction.service.ts

@@ -0,0 +1,117 @@
+import { Service } from '@ourparentcenter/moleculer-decorators-extended';
+import _ from 'lodash';
+import { bytesToHex } from 'viem';
+import { Account, BlockCheckpoint, EVMTransaction } from '../../../models';
+import { BULL_JOB_NAME, SERVICE } from '../constant';
+import BullableService, { QueueHandler } from '../../../base/bullable.service';
+import config from '../../../../config.json' assert { type: 'json' };
+import knex, { batchUpdate } from '../../../common/utils/db_connection';
+
+@Service({
+  name: SERVICE.V1.JobService.UpdateAccountIdInEVMTransaction.key,
+  version: 1,
+})
+export default class UpdateAccountIdInEVMTransaction extends BullableService {
+  @QueueHandler({
+    queueName: BULL_JOB_NAME.UPDATE_ACCOUNT_ID_IN_EVM_TX,
+    jobName: BULL_JOB_NAME.UPDATE_ACCOUNT_ID_IN_EVM_TX,
+  })
+  async updateAccountIdEvmTx() {
+    const [startBlock, endBlock, blockCheckpoint] =
+      await BlockCheckpoint.getCheckpoint(
+        BULL_JOB_NAME.UPDATE_ACCOUNT_ID_IN_EVM_TX,
+        [BULL_JOB_NAME.CRAWL_EVM_ACCOUNT],
+        config.crawlEvmAccount.key
+      );
+    this.logger.info(
+      `Update account id in evm_transaction from block ${startBlock} to ${endBlock}`
+    );
+
+    const [fromTx, toTx] = await Promise.all([
+      EVMTransaction.query()
+        .select('id')
+        .findOne('height', '>', startBlock)
+        .orderBy('height', 'asc')
+        .orderBy('index', 'asc')
+        .limit(1),
+      EVMTransaction.query()
+        .select('id')
+        .findOne('height', '<=', endBlock)
+        .orderBy('height', 'desc')
+        .orderBy('index', 'desc')
+        .limit(1),
+    ]);
+
+    if (!fromTx || !toTx) {
+      const error = 'Cannot found fromTx or toTx';
+      this.logger.error(error);
+      throw Error(error);
+    }
+
+    const txs = await EVMTransaction.query()
+      .select('id', 'from', 'to')
+      .where('id', '>=', fromTx.id)
+      .andWhere('id', '<=', toTx.id);
+    const listAddress = txs.flatMap((tx) => [
+      tx.from ? bytesToHex(tx.from) : undefined,
+      tx.to ? bytesToHex(tx.to) : undefined,
+    ]);
+    const listAddressUnique = _.uniq(listAddress.filter((e) => e));
+    const listAddressDB = _.keyBy(
+      await Account.query()
+        .select('id', 'evm_address')
+        // eslint-disable-next-line @typescript-eslint/ban-ts-comment
+        // @ts-ignore
+        .whereIn('evm_address', listAddressUnique),
+      'evm_address'
+    );
+
+    const listUpdateAccountInTx: any = [];
+    txs.forEach((tx) => {
+      listUpdateAccountInTx.push({
+        id: tx.id,
+        from_account_id: {
+          type: 'number',
+          value: tx.from ? listAddressDB[bytesToHex(tx.from)]?.id : undefined,
+        },
+        to_account_id: {
+          type: 'number',
+          value: tx.to ? listAddressDB[bytesToHex(tx.to)]?.id : undefined,
+        },
+      });
+    });
+    this.logger.debug(listUpdateAccountInTx);
+    await knex.transaction(async (trx) => {
+      await batchUpdate(trx, 'evm_transaction', listUpdateAccountInTx, [
+        'from_account_id',
+        'to_account_id',
+      ]);
+      if (blockCheckpoint) {
+        blockCheckpoint.height = endBlock;
+        await BlockCheckpoint.query()
+          .insert(blockCheckpoint)
+          .onConflict('job_name')
+          .merge()
+          .transacting(trx);
+      }
+    });
+  }
+
+  async _start(): Promise<void> {
+    this.createJob(
+      BULL_JOB_NAME.UPDATE_ACCOUNT_ID_IN_EVM_TX,
+      BULL_JOB_NAME.UPDATE_ACCOUNT_ID_IN_EVM_TX,
+      {},
+      {
+        removeOnComplete: true,
+        removeOnFail: {
+          count: 3,
+        },
+        repeat: {
+          every: config.updateAccountIdInEVMTransaction.millisecondCrawl,
+        },
+      }
+    );
+    return super._start();
+  }
+}