Feat/reindex cw20 ( staging ) (#319)

* feat: reindex cw20 service

* test: test cw20 reindexing

* fix: idiot

* refactor: code

* feat: api admin

ci/config.json.ci

@@ -108,7 +108,10 @@
   "cw20": {
     "blocksPerCall": 100,
     "millisecondRepeatJob": 2000,
-    "key": "cw20"
+    "key": "cw20",
+    "reindexHistory": {
+      "limitRecordGet": 500
+    }
   },
   "dashboardStatistics": {
     "millisecondCrawl": 10000,

config.json

@@ -103,7 +103,10 @@
   "cw20": {
     "blocksPerCall": 100,
     "millisecondRepeatJob": 2000,
-    "key": "cw20"
+    "key": "cw20",
+    "reindexHistory": {
+      "limitRecordGet": 500
+    }
   },
   "crawlContractEvent": {
     "key": "crawlContractEvent",

src/common/constant.ts

@@ -71,6 +71,8 @@ export const BULL_JOB_NAME = {
   REINDEX_CW721_HISTORY: 'reindex:cw721-history',
   HANDLE_MIGRATE_CONTRACT: 'handle:migrate-contract',
   JOB_REDECODE_TX: 'job:redecode-tx',
+  REINDEX_CW20_CONTRACT: 'reindex:cw20-contract',
+  REINDEX_CW20_HISTORY: 'reindex:cw20-history',
   CRAWL_IBC_TAO: 'crawl:ibc-tao',
   CRAWL_GENESIS_IBC_TAO: 'crawl:genesis-ibc-tao',
 };
@@ -250,6 +252,14 @@ export const SERVICE = {
       key: 'DailyStatsJobsService',
       name: 'v1.DailyStatsJobsService',
     },
+    Cw20ReindexingService: {
+      key: 'Cw20ReindexingService',
+      name: 'v1.Cw20ReindexingService',
+      Reindexing: {
+        key: 'reindexing',
+        path: 'v1.Cw20ReindexingService.reindexing',
+      },
+    },
     CrawlIBCTaoService: {
       key: 'CrawlIBCTaoService',
       name: 'v1.CrawlIBCTaoService',

src/models/cw20_activity.ts

@@ -7,6 +7,8 @@ import { SmartContractEvent } from './smart_contract_event';
 import { Event } from './event';
 
 export class Cw20Event extends BaseModel {
+  static softDelete = false;
+
   [relation: string]: any;
 
   id!: number;

src/models/cw20_contract.ts

@@ -26,6 +26,8 @@ export interface IContractInfo {
   name?: string;
 }
 export class Cw20Contract extends BaseModel {
+  static softDelete = false;
+
   [relation: string]: any;
 
   id!: number;

src/models/cw20_holder.ts

@@ -5,6 +5,8 @@ import { Cw20Contract } from './cw20_contract';
 import { SmartContract } from './smart_contract';
 
 export class CW20Holder extends BaseModel {
+  static softDelete = false;
+
   [relation: string]: any;
 
   id?: number;

src/models/cw20_total_holder_stats.ts

@@ -3,6 +3,8 @@ import BaseModel from './base';
 import { Cw20Contract } from './cw20_contract';
 
 export class CW20TotalHolderStats extends BaseModel {
+  static softDelete = false;
+
   [relation: string]: any;
 
   date!: Date;

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

@@ -33,6 +33,12 @@ import { bullBoardMixin } from '../../mixins/bullBoard/bullBoard.mixin';
         mappingPolicy: 'restrict', // allow action called with exact method
         whitelist: ['v2.statistics.syncPrevDateStatsByChainId'],
       },
+      {
+        path: '/admin',
+        autoAliases: true, // allow generate rest info (GET/PUT/POST...) in the services
+        mappingPolicy: 'restrict', // allow action called with exact method
+        whitelist: ['v1.cw20-admin.*'],
+      },
     ],
     // empty cors object will have moleculer to generate handler for preflight request and CORS header which allow all origin
     cors: {},

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

@@ -0,0 +1,45 @@
+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: 'cw20-admin',
+  version: 1,
+})
+export default class Cw20AdminService extends BaseService {
+  public constructor(public broker: ServiceBroker) {
+    super(broker);
+  }
+
+  @Post('/cw20-reindexing', {
+    name: 'cw20Reindexing',
+    params: {
+      chainid: {
+        type: 'string',
+        optional: false,
+        enum: networks.map((network) => network.chainId),
+      },
+      contractAddress: {
+        type: 'string',
+        optional: false,
+      },
+    },
+  })
+  async cw20ReindexingByChainId(
+    ctx: Context<
+      { chainid: string; contractAddress: string },
+      Record<string, unknown>
+    >
+  ) {
+    const selectedChain = networks.find(
+      (network) => network.chainId === ctx.params.chainid
+    );
+    return this.broker.call(
+      `v1.Cw20ReindexingService.reindexing@${selectedChain?.moleculerNamespace}`,
+      {
+        contractAddress: ctx.params.contractAddress,
+      }
+    );
+  }
+}

src/services/cw20/cw20.service.ts

@@ -1,4 +1,5 @@
 import { Service } from '@ourparentcenter/moleculer-decorators-extended';
+import { Queue } from 'bullmq';
 import { Knex } from 'knex';
 import _ from 'lodash';
 import { ServiceBroker } from 'moleculer';
@@ -34,6 +35,13 @@ export const CW20_ACTION = {
   BURN_FROM: 'burn_from',
   SEND_FROM: 'send_from',
 };
+export interface ICw20ReindexingHistoryParams {
+  smartContractId: number;
+  startBlock: number;
+  endBlock: number;
+  prevId: number;
+  contractAddress: string;
+}
 @Service({
   name: SERVICE.V1.Cw20.key,
   version: 1,
@@ -205,41 +213,38 @@ export default class Cw20Service extends BullableService {
     }
   }
 
-  async getCw20ContractEvents(startBlock: number, endBlock: number) {
+  async getCw20ContractEvents(
+    startBlock: number,
+    endBlock: number,
+    smartContractId?: number,
+    page?: { prevId: number; limit: number }
+  ) {
     return SmartContractEvent.query()
-      .alias('smart_contract_event')
-      .withGraphJoined(
-        '[message(selectMessage), tx(selectTransaction), attributes(selectAttribute), smart_contract(selectSmartContract).code(selectCode)]'
-      )
-      .modifiers({
-        selectCode(builder) {
-          builder.select('type');
-        },
-        selectTransaction(builder) {
-          builder.select('hash', 'height');
-        },
-        selectMessage(builder) {
-          builder.select('sender', 'content');
-        },
-        selectAttribute(builder) {
-          builder.select('key', 'value');
-        },
-        selectSmartContract(builder) {
-          builder.select('address', 'id');
-        },
-      })
+      .withGraphFetched('attributes(selectAttribute)')
+      .joinRelated('[message, tx, smart_contract.code]')
       .where('smart_contract:code.type', 'CW20')
       .where('tx.height', '>', startBlock)
       .andWhere('tx.height', '<=', endBlock)
+      .modify((builder) => {
+        if (smartContractId) {
+          builder.andWhere('smart_contract.id', smartContractId);
+        }
+        if (page) {
+          builder
+            .andWhere('smart_contract_event.id', '>', page.prevId)
+            .orderBy('smart_contract_event.id', 'asc')
+            .limit(page.limit);
+        }
+      })
       .select(
         'message.sender as sender',
         'smart_contract.address as contract_address',
         'smart_contract_event.action',
-        'smart_contract_event.event_id as event_id',
-        'smart_contract_event.index',
+        'smart_contract_event.event_id',
         'smart_contract.id as smart_contract_id',
-        'tx.height as height',
-        'smart_contract_event.id as smart_contract_event_id'
+        'smart_contract_event.id as smart_contract_event_id',
+        'tx.hash',
+        'tx.height'
       )
       .orderBy('smart_contract_event.id', 'asc');
   }
@@ -304,4 +309,45 @@ export default class Cw20Service extends BullableService {
     }
     return super._start();
   }
+
+  @QueueHandler({
+    queueName: BULL_JOB_NAME.REINDEX_CW20_HISTORY,
+    jobName: BULL_JOB_NAME.REINDEX_CW20_HISTORY,
+  })
+  public async reindexHistory(_payload: ICw20ReindexingHistoryParams) {
+    const { smartContractId, startBlock, endBlock, prevId, contractAddress } =
+      _payload;
+    // insert data from event_attribute_backup to event_attribute
+    const { limitRecordGet } = config.cw20.reindexHistory;
+    const events = await this.getCw20ContractEvents(
+      startBlock,
+      endBlock,
+      smartContractId,
+      { limit: limitRecordGet, prevId }
+    );
+    if (events.length > 0) {
+      await knex.transaction(async (trx) => {
+        await this.handleCw20Histories(events, trx);
+      });
+      await this.createJob(
+        BULL_JOB_NAME.REINDEX_CW20_HISTORY,
+        BULL_JOB_NAME.REINDEX_CW20_HISTORY,
+        {
+          smartContractId,
+          startBlock,
+          endBlock,
+          prevId: events[events.length - 1].smart_contract_event_id,
+          contractAddress,
+        } satisfies ICw20ReindexingHistoryParams,
+        {
+          removeOnComplete: true,
+        }
+      );
+    } else {
+      const queue: Queue = this.getQueueManager().getQueue(
+        BULL_JOB_NAME.REINDEX_CW20_CONTRACT
+      );
+      (await queue.getJob(contractAddress))?.remove();
+    }
+  }
 }