feat: 🎸 Add process to batch block migrations

db/migrations/3_polyx_transactions_entity.sql

@@ -24,6 +24,7 @@ create table if not exists polyx_transactions
   event_id public_enum_8f5a39c8ee,
   memo text,
   extrinsic_id text,
+  event_idx numeric not null,
   "datetime" timestamp without time zone not null,
   created_block_id text not null,
   updated_block_id text not null,

db/migrations/4_add_migration_processed_block.sql

@@ -0,0 +1,3 @@
+alter table migrations add column if not exists processed_block integer;
+update migrations set processed_block = 0 where processed_block is null;
+alter table migrations alter column processed_block set not null;

db/schemaMigrations.ts

@@ -43,8 +43,8 @@ const getOldMigrationQueries = (): string[] => {
 const migrationInsert = (
   migrationNumber: number,
   id?: string
-) => `INSERT INTO "public"."migrations" ("id", "number", "version", "executed", "created_at", "updated_at")
-VALUES ('${id || migrationNumber}', ${migrationNumber}, '${latestVersion}', false, now(), now())
+) => `INSERT INTO "public"."migrations" ("id", "number", "version", "executed", "processed_block", "created_at", "updated_at")
+VALUES ('${id || migrationNumber}', ${migrationNumber}, '${latestVersion}', false, 0, now(), now())
 ON CONFLICT(id) DO UPDATE SET "updated_at" = now();`;
 
 export const schemaMigrations = async (connection?: Connection): Promise<void> => {

schema.graphql

@@ -2082,6 +2082,7 @@ type Migration @entity {
   number: Int!
   version: String!
   executed: Boolean! @index
+  processedBlock: Int!
 }
 
 """

src/mappings/consts.ts

@@ -39,3 +39,11 @@ export const systematicIssuers = {
     accountId: 'pm/rewrd',
   },
 };
+
+/**
+ * Maximum permissible blocks that should be processed for migrating older events into newer entities
+ *
+ * Generally, it takes about a minute to process 100000 blocks. So if this constant is set to a very high number,
+ * the `--timeout` option for the subquery container needs to be increased accordingly (based on number of blocks to be processed)
+ */
+export const MAX_PERMISSIBLE_BLOCKS = 1000000;

src/mappings/migrations/migrationHandlers.ts

@@ -1,47 +1,72 @@
 import { Event, Migration } from '../../types';
+import { MAX_PERMISSIBLE_BLOCKS } from '../consts';
 import { mapPolyxTransaction } from '../entities/mapPolyxTransaction';
 
-let dataMigrationStarted = false;
+let dataMigrationCompleted = false;
 
 /**
- * Adds data for older blocks using the `mapPolyxTransaction` handler
- * This will only be executed when introducing the polyxTransaction entity (i.e. with migration handler 3)
+ * Adds data for older blocks using the `mapPolyxTransaction` handler.
+ * This will only be executed migration version 3 is not yet changed to executed state.
  */
-const handlePolyxMigration = async (blockId: number, ss58Format?: number) => {
-  for (let blockNumber = 0; blockNumber <= blockId; blockNumber++) {
-    logger.info(`Processing block - ${blockNumber} for mapping polyx transactions`);
-    const events = await Event.getByBlockId(blockNumber.toString());
+const handlePolyxMigration = async (
+  migratedBlock: number,
+  indexedBlock: number,
+  ss58Format?: number
+) => {
+  let currentBlock = migratedBlock;
+  while (currentBlock <= indexedBlock && currentBlock <= migratedBlock + MAX_PERMISSIBLE_BLOCKS) {
+    logger.debug(`Processing block - ${currentBlock} for mapping polyx transactions`);
+    const events = await Event.getByBlockId(currentBlock.toString());
     if (events) {
       for (const event of events) {
         await mapPolyxTransaction(event, ss58Format);
       }
     }
+    currentBlock++;
   }
+  logger.info(`Processed block - ${currentBlock} for mapping polyx transactions`);
+  return currentBlock;
 };
 
 /**
- * This method processes the already indexed blocks and their events once on startup.
+ * This method processes the already indexed blocks and their events once, on startup.
  * This helps running data migration queries using the SQ entities.
  *
- * Once a new block is received, the method is triggered.
+ * Once a new block is received, the method is triggered to process older blocks (max upto `MAX_PERMISSIBLE_BLOCKS`).
  * For each block, it fetches the events and process accordingly as per data migration requirements
  *
- * @note For adding in schema changes (like new events in chain upgrade, adding a type to existing enum), we still need to use the sql migration queries as those are supposed to be executed before the SQ startup.
+ * @note For adding in schema changes (like new events in chain upgrade, adding a type to existing enum),
+ * we still need to use the sql migration queries as those are supposed to be executed before the SQ startup.
+ * Also, processing say millions of blocks in a single transaction commit may take upto hours to process,
+ * thus a MAX_PERMISSIBLE_BLOCKS limit is set which makes sure that the block processor doesn't timeout.
  */
 export default async (blockId: number, ss58Format?: number): Promise<void> => {
-  if (dataMigrationStarted) {
+  if (dataMigrationCompleted) {
     return;
   }
-  dataMigrationStarted = true;
 
   const migrations = await Migration.getByExecuted(false);
+
   logger.info(`Executing migration handlers for ${migrations.length} new migration(s)`);
 
   for (const migration of migrations) {
-    if (migration.number === 3) {
-      await handlePolyxMigration(blockId, ss58Format);
+    const { processedBlock, number } = migration;
+    let lastProcessedBlock = 0;
+
+    if (number === 3) {
+      lastProcessedBlock = await handlePolyxMigration(processedBlock, blockId, ss58Format);
     } else {
       logger.info(`No mapping handlers are associated for migration - ${migration.id}`);
+      migration.executed = true;
+    }
+
+    migration.processedBlock = lastProcessedBlock;
+
+    if (processedBlock >= blockId) {
+      migration.executed = true;
+      dataMigrationCompleted = true;
     }
+
+    await migration.save();
   }
 };