Feat/reindex batch erc721 staging

network.json

@@ -31,7 +31,9 @@
     "LCD": ["https://lcd.serenity.aura.network"],
     "databaseName": "horoscope_dev_serenitytestnet001",
     "redisDBNumber": 3,
-    "moleculerNamespace": "namespace-serenity"
+    "moleculerNamespace": "namespace-serenity",
+    "EVMJSONRPC": ["https://jsonrpc.serenity.aura.network/"],
+    "EVMchainId": 1236
   },
   {
     "chainId": "auradev_1235-3",

src/models/erc721_contract.ts

@@ -230,5 +230,24 @@ export class Erc721Contract extends BaseModel {
       stateMutability: 'view',
       type: 'function',
     },
+    {
+      inputs: [
+        {
+          internalType: 'uint256',
+          name: '_index',
+          type: 'uint256',
+        },
+      ],
+      name: 'tokenByIndex',
+      outputs: [
+        {
+          internalType: 'uint256',
+          name: 'tokenId',
+          type: 'uint256',
+        },
+      ],
+      stateMutability: 'view',
+      type: 'function',
+    },
   ];
 }

src/services/api-gateways/api_gateway.service.ts

@@ -44,6 +44,7 @@ import config from '../../../config.json' assert { type: 'json' };
           'v1.job.update-delegator-validator',
           'v1.job.signature-mapping',
           'v1.job.insert-verify-by-codehash',
+          'v1.erc721-admin.*',
         ],
       },
       {

src/services/api-gateways/erc721_admin.service.ts

@@ -0,0 +1,49 @@
+import { Post, Service } from '@ourparentcenter/moleculer-decorators-extended';
+import { Context, ServiceBroker } from 'moleculer';
+import networks from '../../../network.json' assert { type: 'json' };
+import BaseService from '../../base/base.service';
+
+@Service({
+  name: 'erc721-admin',
+  version: 1,
+})
+export default class Erc721AdminService extends BaseService {
+  public constructor(public broker: ServiceBroker) {
+    super(broker);
+  }
+
+  @Post('/erc721-reindexing', {
+    name: 'erc721Reindexing',
+    params: {
+      chainid: {
+        type: 'string',
+        optional: false,
+        enum: networks.map((network) => network.chainId),
+      },
+      addresses: {
+        type: 'array',
+        optional: false,
+        items: 'string',
+      },
+    },
+  })
+  async erc721Reindexing(
+    ctx: Context<
+      {
+        chainid: string;
+        addresses: string[];
+      },
+      Record<string, unknown>
+    >
+  ) {
+    const selectedChain = networks.find(
+      (network) => network.chainId === ctx.params.chainid
+    );
+    return this.broker.call(
+      `v1.Erc721.reindexing@${selectedChain?.moleculerNamespace}`,
+      {
+        addresses: ctx.params.addresses,
+      }
+    );
+  }
+}

src/services/evm/constant.ts

@@ -121,6 +121,10 @@ export const SERVICE = {
         key: 'insertNewErc721Contracts',
         path: 'v1.Erc721.insertNewErc721Contracts',
       },
+      reindexing: {
+        key: 'reindexing',
+        path: 'v1.Erc721.reindexing',
+      },
     },
     SignatureMappingEVM: {
       key: 'SignatureMappingEVM',

src/services/evm/erc721.service.ts

@@ -3,7 +3,7 @@ import {
   Service,
 } from '@ourparentcenter/moleculer-decorators-extended';
 import { Knex } from 'knex';
-import _, { Dictionary } from 'lodash';
+import _ from 'lodash';
 import { Context, ServiceBroker } from 'moleculer';
 import { PublicClient, getContract } from 'viem';
 import config from '../../../config.json' assert { type: 'json' };
@@ -12,22 +12,17 @@ import { Config } from '../../common';
 import knex from '../../common/utils/db_connection';
 import { getViemClient } from '../../common/utils/etherjs_client';
 import {
-  Block,
   BlockCheckpoint,
   EVMSmartContract,
   Erc721Activity,
   Erc721Stats,
   Erc721Token,
-  EvmEvent,
 } from '../../models';
 import { Erc721Contract } from '../../models/erc721_contract';
 import { BULL_JOB_NAME, SERVICE } from './constant';
-import {
-  ERC721_ACTION,
-  ERC721_EVENT_TOPIC0,
-  Erc721Handler,
-} from './erc721_handler';
+import { Erc721Handler } from './erc721_handler';
 import * as Erc721MediaHandler from './erc721_media_handler';
+import { Erc721Reindexer } from './erc721_reindex';
 
 const { NODE_ENV } = Config;
 @Service({
@@ -90,55 +85,14 @@ export default class Erc721Service extends BullableService {
           ],
           config.erc721.key
         );
-      const erc721Events = await EvmEvent.query()
-        .transacting(trx)
-        .joinRelated('[evm_smart_contract,evm_transaction]')
-        .innerJoin(
-          'erc721_contract',
-          'evm_event.address',
-          'erc721_contract.address'
-        )
-        .where('evm_event.block_height', '>', startBlock)
-        .andWhere('evm_event.block_height', '<=', endBlock)
-        .orderBy('evm_event.id', 'asc')
-        .select(
-          'evm_event.*',
-          'evm_transaction.from as sender',
-          'evm_smart_contract.id as evm_smart_contract_id',
-          'evm_transaction.id as evm_tx_id',
-          'erc721_contract.track as track'
+      const erc721Activities: Erc721Activity[] =
+        await Erc721Handler.getErc721Activities(
+          startBlock,
+          endBlock,
+          trx,
+          this.logger
         );
-      await this.handleMissingErc721Contract(erc721Events, trx);
-      const erc721Activities: Erc721Activity[] = [];
-      erc721Events
-        .filter((e) => e.track)
-        .forEach((e) => {
-          if (e.topic0 === ERC721_EVENT_TOPIC0.TRANSFER) {
-            const activity = Erc721Handler.buildTransferActivity(
-              e,
-              this.logger
-            );
-            if (activity) {
-              erc721Activities.push(activity);
-            }
-          } else if (e.topic0 === ERC721_EVENT_TOPIC0.APPROVAL) {
-            const activity = Erc721Handler.buildApprovalActivity(
-              e,
-              this.logger
-            );
-            if (activity) {
-              erc721Activities.push(activity);
-            }
-          } else if (e.topic0 === ERC721_EVENT_TOPIC0.APPROVAL_FOR_ALL) {
-            const activity = Erc721Handler.buildApprovalForAllActivity(
-              e,
-              this.logger
-            );
-            if (activity) {
-              erc721Activities.push(activity);
-            }
-          }
-        });
+      await this.handleMissingErc721Contract(erc721Activities, trx);
       if (erc721Activities.length > 0) {
         const erc721Tokens = _.keyBy(
           await Erc721Token.query()
@@ -155,7 +109,7 @@ export default class Erc721Service extends BullableService {
         );
         const erc721Handler = new Erc721Handler(erc721Tokens, erc721Activities);
         erc721Handler.process();
-        await this.updateErc721(
+        await Erc721Handler.updateErc721(
           erc721Activities,
           Object.values(erc721Handler.erc721Tokens),
           trx
@@ -263,7 +217,7 @@ export default class Erc721Service extends BullableService {
     jobName: BULL_JOB_NAME.REFRESH_ERC721_STATS,
   })
   async jobHandlerRefresh(): Promise<void> {
-    const erc721Stats = await this.calErc721Stats();
+    const erc721Stats = await Erc721Handler.calErc721Stats();
     // Upsert erc721 stats
     await Erc721Stats.query()
       .insert(
@@ -280,6 +234,17 @@ export default class Erc721Service extends BullableService {
       .returning('id');
   }
 
+  @QueueHandler({
+    queueName: BULL_JOB_NAME.REINDEX_ERC721,
+    jobName: BULL_JOB_NAME.REINDEX_ERC721,
+  })
+  async reindexErc721(_payload: { address: `0x${string}` }): Promise<void> {
+    const { address } = _payload;
+    const erc721Reindexer = new Erc721Reindexer(this.viemClient, this.logger);
+    await erc721Reindexer.reindex(address);
+    this.logger.info(`Reindex erc721 contract ${address} done.`);
+  }
+
   @Action({
     name: SERVICE.V1.Erc721.insertNewErc721Contracts.key,
     params: {
@@ -315,49 +280,40 @@ export default class Erc721Service extends BullableService {
     }
   }
 
-  async updateErc721(
-    erc721Activities: Erc721Activity[],
-    erc721Tokens: Erc721Token[],
-    trx: Knex.Transaction
+  @Action({
+    name: SERVICE.V1.Erc721.reindexing.key,
+    params: {
+      addresses: {
+        type: 'array',
+        items: 'string',
+        optional: false,
+      },
+    },
+  })
+  public async reindexing(
+    ctx: Context<{
+      addresses: `0x${string}`[];
+    }>
   ) {
-    let updatedTokens: Dictionary<Erc721Token> = {};
-    if (erc721Tokens.length > 0) {
-      updatedTokens = _.keyBy(
-        await Erc721Token.query()
-          .insert(
-            erc721Tokens.map((token) =>
-              Erc721Token.fromJson({
-                token_id: token.token_id,
-                owner: token.owner,
-                erc721_contract_address: token.erc721_contract_address,
-                last_updated_height: token.last_updated_height,
-              })
-            )
+    let { addresses } = ctx.params;
+    const erc721Reindexer = new Erc721Reindexer(this.viemClient, this.logger);
+    addresses = await erc721Reindexer.filterReindex(addresses);
+    if (addresses.length > 0) {
+      await Promise.all(
+        addresses.map((address) =>
+          this.createJob(
+            BULL_JOB_NAME.REINDEX_ERC721,
+            BULL_JOB_NAME.REINDEX_ERC721,
+            {
+              address,
+            },
+            {
+              jobId: address,
+              removeOnComplete: true,
+            }
           )
-          .onConflict(['token_id', 'erc721_contract_address'])
-          .merge()
-          .transacting(trx),
-        (o) => `${o.erc721_contract_address}_${o.token_id}`
-      );
-    }
-    if (erc721Activities.length > 0) {
-      erc721Activities.forEach((activity) => {
-        const token =
-          updatedTokens[
-            `${activity.erc721_contract_address}_${activity.token_id}`
-          ];
-        if (token) {
-          // eslint-disable-next-line no-param-reassign
-          activity.erc721_token_id = token.id;
-        }
-      });
-      await knex
-        .batchInsert(
-          'erc721_activity',
-          erc721Activities.map((e) => _.omit(e, 'token_id')),
-          config.erc721.chunkSizeInsert
         )
-        .transacting(trx);
+      );
     }
   }
 
@@ -403,32 +359,15 @@ export default class Erc721Service extends BullableService {
     }));
   }
 
-  async calErc721Stats(): Promise<Erc721Contract[]> {
-    // Get once block height 24h ago.
-    const blockSince24hAgo = await Block.query()
-      .select('height')
-      .where('time', '<=', knex.raw("now() - '24 hours'::interval"))
-      .orderBy('height', 'desc')
-      .limit(1);
-
-    // Calculate total activity and transfer_24h of erc721
-    return Erc721Contract.query()
-      .count('erc721_activity.id AS total_activity')
-      .select(
-        knex.raw(
-          `SUM( CASE WHEN erc721_activity.height >= ? AND erc721_activity.action = '${ERC721_ACTION.TRANSFER}' THEN 1 ELSE 0 END ) AS transfer_24h`,
-          blockSince24hAgo[0]?.height
-        )
-      )
-      .select('erc721_contract.id as erc721_contract_id')
-      .where('erc721_contract.track', '=', true)
-      .joinRelated('erc721_activity')
-      .groupBy('erc721_contract.id');
-  }
-
-  async handleMissingErc721Contract(events: EvmEvent[], trx: Knex.Transaction) {
+  async handleMissingErc721Contract(
+    erc721Activities: Erc721Activity[],
+    trx: Knex.Transaction
+  ) {
     try {
-      const eventsUniqByAddress = _.keyBy(events, (e) => e.address);
+      const eventsUniqByAddress = _.keyBy(
+        erc721Activities,
+        (e) => e.erc721_contract_address
+      );
       const addresses = Object.keys(eventsUniqByAddress);
       const erc721ContractsByAddress = _.keyBy(
         await Erc721Contract.query()