Add IpldStatus table and IPLD jobs in job-runner in codegen (#100)

packages/codegen/src/data/entities/IpldStatus.yaml

@@ -1,10 +1,18 @@
-className: HookStatus
+className: IpldStatus
 indexOn: []
 columns:
   - name: id
     tsType: number
     columnType: PrimaryGeneratedColumn
-  - name: latestProcessedBlockNumber
+  - name: latestHooksBlockNumber
+    pgType: integer
+    tsType: number
+    columnType: Column
+  - name: latestCheckpointBlockNumber
+    pgType: integer
+    tsType: number
+    columnType: Column
+  - name: latestIPFSBlockNumber
     pgType: integer
     tsType: number
     columnType: Column

packages/codegen/src/entity.ts

@@ -179,7 +179,7 @@ export class Entity {
     this._addContractEntity();
     this._addBlockProgressEntity();
     this._addIPLDBlockEntity();
-    this._addHookStatusEntity();
+    this._addIpldStatusEntity();
 
     const template = Handlebars.compile(this._templateString);
     this._entities.forEach(entityObj => {
@@ -279,8 +279,8 @@ export class Entity {
     this._entities.push(entity);
   }
 
-  _addHookStatusEntity (): void {
-    const entity = yaml.load(fs.readFileSync(path.resolve(__dirname, TABLES_DIR, 'HookStatus.yaml'), 'utf8'));
+  _addIpldStatusEntity (): void {
+    const entity = yaml.load(fs.readFileSync(path.resolve(__dirname, TABLES_DIR, 'IpldStatus.yaml'), 'utf8'));
     this._entities.push(entity);
   }
 

packages/codegen/src/templates/database-template.handlebars

@@ -11,7 +11,7 @@ import { IPLDDatabase as BaseDatabase, IPLDDatabaseInterface, QueryOptions, Stat
 import { Contract } from './entity/Contract';
 import { Event } from './entity/Event';
 import { SyncStatus } from './entity/SyncStatus';
-import { HookStatus } from './entity/HookStatus';
+import { IpldStatus } from './entity/IpldStatus';
 import { BlockProgress } from './entity/BlockProgress';
 import { IPLDBlock } from './entity/IPLDBlock';
 {{#each queries as | query |}}
@@ -120,16 +120,28 @@ export class Database implements IPLDDatabaseInterface {
     await this._baseDatabase.removeIPLDBlocks(repo, blockNumber, kind);
   }
 
-  async getHookStatus (): Promise<HookStatus | undefined> {
-    const repo = this._conn.getRepository(HookStatus);
+  async getIPLDStatus (): Promise<IpldStatus | undefined> {
+    const repo = this._conn.getRepository(IpldStatus);
 
-    return this._baseDatabase.getHookStatus(repo);
+    return this._baseDatabase.getIPLDStatus(repo);
   }
 
-  async updateHookStatusProcessedBlock (queryRunner: QueryRunner, blockNumber: number, force?: boolean): Promise<HookStatus> {
-    const repo = queryRunner.manager.getRepository(HookStatus);
+  async updateIPLDStatusHooksBlock (queryRunner: QueryRunner, blockNumber: number, force?: boolean): Promise<IpldStatus> {
+    const repo = queryRunner.manager.getRepository(IpldStatus);
 
-    return this._baseDatabase.updateHookStatusProcessedBlock(repo, blockNumber, force);
+    return this._baseDatabase.updateIPLDStatusHooksBlock(repo, blockNumber, force);
+  }
+
+  async updateIPLDStatusCheckpointBlock (queryRunner: QueryRunner, blockNumber: number, force?: boolean): Promise<IpldStatus> {
+    const repo = queryRunner.manager.getRepository(IpldStatus);
+
+    return this._baseDatabase.updateIPLDStatusCheckpointBlock(repo, blockNumber, force);
+  }
+
+  async updateIPLDStatusIPFSBlock (queryRunner: QueryRunner, blockNumber: number, force?: boolean): Promise<IpldStatus> {
+    const repo = queryRunner.manager.getRepository(IpldStatus);
+
+    return this._baseDatabase.updateIPLDStatusIPFSBlock(repo, blockNumber, force);
   }
 
   async getContracts (): Promise<Contract[]> {

packages/codegen/src/templates/events-template.handlebars

@@ -13,12 +13,8 @@ import {
   EventWatcherInterface,
   QUEUE_BLOCK_PROCESSING,
   QUEUE_EVENT_PROCESSING,
-  QUEUE_BLOCK_CHECKPOINT,
-  QUEUE_HOOKS,
-  QUEUE_IPFS,
   UNKNOWN_EVENT_NAME,
-  UpstreamConfig,
-  JOB_KIND_PRUNE
+  UpstreamConfig
 } from '@vulcanize/util';
 
 import { Indexer } from './indexer';
@@ -60,8 +56,6 @@ export class EventWatcher implements EventWatcherInterface {
 
     await this.initBlockProcessingOnCompleteHandler();
     await this.initEventProcessingOnCompleteHandler();
-    await this.initHooksOnCompleteHandler();
-    await this.initBlockCheckpointOnCompleteHandler();
     this._baseEventWatcher.startBlockProcessing();
   }
 
@@ -71,7 +65,7 @@ export class EventWatcher implements EventWatcherInterface {
 
   async initBlockProcessingOnCompleteHandler (): Promise<void> {
     this._jobQueue.onComplete(QUEUE_BLOCK_PROCESSING, async (job) => {
-      const { id, data: { failed, request: { data: { kind } } } } = job;
+      const { id, data: { failed } } = job;
 
       if (failed) {
         log(`Job ${id} for queue ${QUEUE_BLOCK_PROCESSING} failed`);
@@ -79,11 +73,6 @@ export class EventWatcher implements EventWatcherInterface {
       }
 
       await this._baseEventWatcher.blockProcessingCompleteHandler(job);
-
-      // If it's a pruning job: Create a hooks job.
-      if (kind === JOB_KIND_PRUNE) {
-        await this.createHooksJob();
-      }
     });
   }
 
@@ -117,27 +106,6 @@ export class EventWatcher implements EventWatcherInterface {
     });
   }
 
-  async initHooksOnCompleteHandler (): Promise<void> {
-    this._jobQueue.onComplete(QUEUE_HOOKS, async (job) => {
-      const { data: { request: { data: { blockNumber, blockHash } } } } = job;
-
-      await this._indexer.updateHookStatusProcessedBlock(blockNumber);
-
-      // Create a checkpoint job after completion of a hook job.
-      await this.createCheckpointJob(blockHash, blockNumber);
-    });
-  }
-
-  async initBlockCheckpointOnCompleteHandler (): Promise<void> {
-    this._jobQueue.onComplete(QUEUE_BLOCK_CHECKPOINT, async (job) => {
-      const { data: { request: { data: { blockHash } } } } = job;
-
-      if (this._indexer.isIPFSConfigured()) {
-        await this.createIPFSPutJob(blockHash);
-      }
-    });
-  }
-
   async publishEventToSubscribers (dbEvent: Event, timeElapsedInSeconds: number): Promise<void> {
     if (dbEvent && dbEvent.eventName !== UNKNOWN_EVENT_NAME) {
       const resultEvent = this._indexer.getResultEvent(dbEvent);
@@ -150,38 +118,4 @@ export class EventWatcher implements EventWatcherInterface {
       });
     }
   }
-
-  async createHooksJob (): Promise<void> {
-    // Get the latest canonical block
-    const latestCanonicalBlock = await this._indexer.getLatestCanonicalBlock();
-
-    // Create a hooks job for parent block of latestCanonicalBlock because pruning for first block is skipped as it is assumed to be a canonical block.
-    await this._jobQueue.pushJob(
-      QUEUE_HOOKS,
-      {
-        blockHash: latestCanonicalBlock.parentHash,
-        blockNumber: latestCanonicalBlock.blockNumber - 1
-      }
-    );
-  }
-
-  async createCheckpointJob (blockHash: string, blockNumber: number): Promise<void> {
-    await this._jobQueue.pushJob(
-      QUEUE_BLOCK_CHECKPOINT,
-      {
-        blockHash,
-        blockNumber
-      }
-    );
-  }
-
-  async createIPFSPutJob (blockHash: string): Promise<void> {
-    const ipldBlocks = await this._indexer.getIPLDBlocksByHash(blockHash);
-
-    for (const ipldBlock of ipldBlocks) {
-      const data = this._indexer.getIPLDData(ipldBlock);
-
-      await this._jobQueue.pushJob(QUEUE_IPFS, { data });
-    }
-  }
 }

packages/codegen/src/templates/indexer-template.handlebars

@@ -38,7 +38,7 @@ import { createInitialState, handleEvent, createStateDiff, createStateCheckpoint
 import { Contract } from './entity/Contract';
 import { Event } from './entity/Event';
 import { SyncStatus } from './entity/SyncStatus';
-import { HookStatus } from './entity/HookStatus';
+import { IpldStatus } from './entity/IpldStatus';
 import { BlockProgress } from './entity/BlockProgress';
 import { IPLDBlock } from './entity/IPLDBlock';
 
@@ -309,9 +309,7 @@ export class Indexer implements IndexerInterface {
     return createStateCheckpoint(this, contractAddress, blockHash);
   }
 
-  async processCanonicalBlock (job: any): Promise<void> {
-    const { data: { blockHash } } = job;
-
+  async processCanonicalBlock (blockHash: string): Promise<void> {
     // Finalize staged diff blocks if any.
     await this._baseIndexer.finalizeDiffStaged(blockHash);
 
@@ -319,12 +317,11 @@ export class Indexer implements IndexerInterface {
     await createStateDiff(this, blockHash);
   }
 
-  async processCheckpoint (job: any): Promise<void> {
+  async processCheckpoint (blockHash: string): Promise<void> {
     // Return if checkpointInterval is <= 0.
     const checkpointInterval = this._serverConfig.checkpointInterval;
     if (checkpointInterval <= 0) return;
 
-    const { data: { blockHash } } = job;
     await this._baseIndexer.processCheckpoint(this, blockHash, checkpointInterval);
   }
 
@@ -489,16 +486,50 @@ export class Indexer implements IndexerInterface {
   }
 
   {{/each}}
-  async getHookStatus (): Promise<HookStatus | undefined> {
-    return this._db.getHookStatus();
+  async getIPLDStatus (): Promise<IpldStatus | undefined> {
+    return this._db.getIPLDStatus();
   }
 
-  async updateHookStatusProcessedBlock (blockNumber: number, force?: boolean): Promise<HookStatus> {
+  async updateIPLDStatusHooksBlock (blockNumber: number, force?: boolean): Promise<IpldStatus> {
     const dbTx = await this._db.createTransactionRunner();
     let res;
 
     try {
-      res = await this._db.updateHookStatusProcessedBlock(dbTx, blockNumber, force);
+      res = await this._db.updateIPLDStatusHooksBlock(dbTx, blockNumber, force);
+      await dbTx.commitTransaction();
+    } catch (error) {
+      await dbTx.rollbackTransaction();
+      throw error;
+    } finally {
+      await dbTx.release();
+    }
+
+    return res;
+  }
+
+  async updateIPLDStatusCheckpointBlock (blockNumber: number, force?: boolean): Promise<IpldStatus> {
+    const dbTx = await this._db.createTransactionRunner();
+    let res;
+
+    try {
+      res = await this._db.updateIPLDStatusCheckpointBlock(dbTx, blockNumber, force);
+      await dbTx.commitTransaction();
+    } catch (error) {
+      await dbTx.rollbackTransaction();
+      throw error;
+    } finally {
+      await dbTx.release();
+    }
+
+    return res;
+  }
+
+  async updateIPLDStatusIPFSBlock (blockNumber: number, force?: boolean): Promise<IpldStatus> {
+    const dbTx = await this._db.createTransactionRunner();
+    let res;
+
+    try {
+      res = await this._db.updateIPLDStatusIPFSBlock(dbTx, blockNumber, force);
       await dbTx.commitTransaction();
     } catch (error) {
       await dbTx.rollbackTransaction();

packages/codegen/src/templates/job-runner-template.handlebars

@@ -19,6 +19,7 @@ import {
   QUEUE_BLOCK_CHECKPOINT,
   QUEUE_HOOKS,
   QUEUE_IPFS,
+  JOB_KIND_PRUNE,
   JobQueueConfig,
   DEFAULT_CONFIG_PATH,
   initClients
@@ -53,9 +54,16 @@ export class JobRunner {
 
   async subscribeBlockProcessingQueue (): Promise<void> {
     await this._jobQueue.subscribe(QUEUE_BLOCK_PROCESSING, async (job) => {
-      // TODO Call pre-block hook here (Directly or indirectly (Like done through indexer.processEvent for events)).
-
       await this._baseJobRunner.processBlock(job);
+
+      const { data: { kind } } = job;
+
+      // If it's a pruning job: Create a hooks job.
+      if (kind === JOB_KIND_PRUNE) {
+        await this.createHooksJob();
+      }
+
+      await this._jobQueue.markComplete(job);
     });
   }
 
@@ -67,26 +75,38 @@ export class JobRunner {
 
   async subscribeHooksQueue (): Promise<void> {
     await this._jobQueue.subscribe(QUEUE_HOOKS, async (job) => {
-      const { data: { blockNumber } } = job;
+      const { data: { blockHash, blockNumber } } = job;
 
-      const hookStatus = await this._indexer.getHookStatus();
+      // Get the current IPLD Status.
+      const ipldStatus = await this._indexer.getIPLDStatus();
+
+      if (ipldStatus) {
+        if (ipldStatus.latestHooksBlockNumber < (blockNumber - 1)) {
+          // Create hooks job for parent block.
+          const [parentBlock] = await this._indexer.getBlocksAtHeight(blockNumber - 1, false);
+          await this.createHooksJob(parentBlock.blockHash, parentBlock.blockNumber);
 
-      if (hookStatus) {
-        if (hookStatus.latestProcessedBlockNumber < (blockNumber - 1)) {
           const message = `Hooks for blockNumber ${blockNumber - 1} not processed yet, aborting`;
           log(message);
 
           throw new Error(message);
         }
 
-        if (hookStatus.latestProcessedBlockNumber > (blockNumber - 1)) {
+        if (ipldStatus.latestHooksBlockNumber > (blockNumber - 1)) {
           log(`Hooks for blockNumber ${blockNumber} already processed`);
 
           return;
         }
       }
 
-      await this._indexer.processCanonicalBlock(job);
+      // Process the hooks for the given block number.
+      await this._indexer.processCanonicalBlock(blockHash);
+
+      // Update the IPLD status.
+      await this._indexer.updateIPLDStatusHooksBlock(blockNumber);
+
+      // Create a checkpoint job after completion of a hook job.
+      await this.createCheckpointJob(blockHash, blockNumber);
 
       await this._jobQueue.markComplete(job);
     });
@@ -94,7 +114,41 @@ export class JobRunner {
 
   async subscribeBlockCheckpointQueue (): Promise<void> {
     await this._jobQueue.subscribe(QUEUE_BLOCK_CHECKPOINT, async (job) => {
-      await this._indexer.processCheckpoint(job);
+      const { data: { blockHash, blockNumber } } = job;
+
+      // Get the current IPLD Status.
+      const ipldStatus = await this._indexer.getIPLDStatus();
+      assert(ipldStatus);
+
+      if (ipldStatus.latestCheckpointBlockNumber >= 0) {
+        if (ipldStatus.latestCheckpointBlockNumber < (blockNumber - 1)) {
+          // Create a checkpoint job for parent block.
+          const [parentBlock] = await this._indexer.getBlocksAtHeight(blockNumber - 1, false);
+          await this.createCheckpointJob(parentBlock.blockHash, parentBlock.blockNumber);
+
+          const message = `Checkpoints for blockNumber ${blockNumber - 1} not processed yet, aborting`;
+          log(message);
+
+          throw new Error(message);
+        }
+
+        if (ipldStatus.latestCheckpointBlockNumber > (blockNumber - 1)) {
+          log(`Checkpoints for blockNumber ${blockNumber} already processed`);
+
+          return;
+        }
+      }
+
+      // Process checkpoints for the given block.
+      await this._indexer.processCheckpoint(blockHash);
+
+      // Update the IPLD status.
+      await this._indexer.updateIPLDStatusCheckpointBlock(blockNumber);
+
+      // Create an IPFS job after completion of a checkpoint job.
+      if (this._indexer.isIPFSConfigured()) {
+        await this.createIPFSPutJob(blockHash, blockNumber);
+      }
 
       await this._jobQueue.markComplete(job);
     });
@@ -102,13 +156,84 @@ export class JobRunner {
 
   async subscribeIPFSQueue (): Promise<void> {
     await this._jobQueue.subscribe(QUEUE_IPFS, async (job) => {
-      const { data: { data } } = job;
+      const { data: { blockHash, blockNumber } } = job;
 
-      await this._indexer.pushToIPFS(data);
+      const ipldStatus = await this._indexer.getIPLDStatus();
+      assert(ipldStatus);
+
+      if (ipldStatus.latestIPFSBlockNumber >= 0) {
+        if (ipldStatus.latestIPFSBlockNumber < (blockNumber - 1)) {
+          // Create a IPFS job for parent block.
+          const [parentBlock] = await this._indexer.getBlocksAtHeight(blockNumber - 1, false);
+          await this.createIPFSPutJob(parentBlock.blockHash, parentBlock.blockNumber);
+
+          const message = `IPFS for blockNumber ${blockNumber - 1} not processed yet, aborting`;
+          log(message);
+
+          throw new Error(message);
+        }
+
+        if (ipldStatus.latestIPFSBlockNumber > (blockNumber - 1)) {
+          log(`IPFS for blockNumber ${blockNumber} already processed`);
+
+          return;
+        }
+      }
+
+      // Get IPLDBlocks for the given blocHash.
+      const ipldBlocks = await this._indexer.getIPLDBlocksByHash(blockHash);
+
+      // Push all the IPLDBlocks to IPFS.
+      for (const ipldBlock of ipldBlocks) {
+        const data = this._indexer.getIPLDData(ipldBlock);
+        await this._indexer.pushToIPFS(data);
+      }
+
+      // Update the IPLD status.
+      await this._indexer.updateIPLDStatusIPFSBlock(blockNumber);
 
       await this._jobQueue.markComplete(job);
     });
   }
+
+  async createHooksJob (blockHash?: string, blockNumber?: number): Promise<void> {
+    if (!blockNumber || !blockHash) {
+      // Get the latest canonical block
+      const latestCanonicalBlock = await this._indexer.getLatestCanonicalBlock();
+
+      // Create a hooks job for parent block of latestCanonicalBlock because pruning for first block is skipped as it is assumed to be a canonical block.
+      blockHash = latestCanonicalBlock.parentHash;
+      blockNumber = latestCanonicalBlock.blockNumber - 1;
+    }
+
+    await this._jobQueue.pushJob(
+      QUEUE_HOOKS,
+      {
+        blockHash,
+        blockNumber
+      }
+    );
+  }
+
+  async createCheckpointJob (blockHash: string, blockNumber: number): Promise<void> {
+    await this._jobQueue.pushJob(
+      QUEUE_BLOCK_CHECKPOINT,
+      {
+        blockHash,
+        blockNumber
+      }
+    );
+  }
+
+  async createIPFSPutJob (blockHash: string, blockNumber: number): Promise<void> {
+    await this._jobQueue.pushJob(
+      QUEUE_IPFS,
+      {
+        blockHash,
+        blockNumber
+      }
+    );
+  }
 }
 
 export const main = async (): Promise<any> => {
@@ -164,8 +289,3 @@ main().then(() => {
 process.on('uncaughtException', err => {
   log('uncaughtException', err);
 });
-
-process.on('SIGINT', () => {
-  log(`Exiting process ${process.pid} with code 0`);
-  process.exit(0);
-});

packages/codegen/src/templates/reset-state-template.handlebars

@@ -62,8 +62,8 @@ export const handler = async (argv: any): Promise<void> => {
   const syncStatus = await indexer.getSyncStatus();
   assert(syncStatus, 'Missing syncStatus');
 
-  const hooksStatus = await indexer.getHookStatus();
-  assert(hooksStatus, 'Missing hooksStatus');
+  const ipldStatus = await indexer.getIPLDStatus();
+  assert(ipldStatus, 'Missing ipldStatus');
 
   const blockProgresses = await indexer.getBlocksAtHeight(argv.blockNumber, false);
   assert(blockProgresses.length, `No blocks at specified block number ${argv.blockNumber}`);
@@ -93,8 +93,16 @@ export const handler = async (argv: any): Promise<void> => {
       await indexer.updateSyncStatusCanonicalBlock(blockProgress.blockHash, blockProgress.blockNumber, true);
     }
 
-    if (hooksStatus.latestProcessedBlockNumber > blockProgress.blockNumber) {
-      await indexer.updateHookStatusProcessedBlock(blockProgress.blockNumber, true);
+    if (ipldStatus.latestHooksBlockNumber > blockProgress.blockNumber) {
+      await indexer.updateIPLDStatusHooksBlock(blockProgress.blockNumber, true);
+    }
+
+    if (ipldStatus.latestCheckpointBlockNumber > blockProgress.blockNumber) {
+      await indexer.updateIPLDStatusCheckpointBlock(blockProgress.blockNumber, true);
+    }
+
+    if (ipldStatus.latestIPFSBlockNumber > blockProgress.blockNumber) {
+      await indexer.updateIPLDStatusIPFSBlock(blockProgress.blockNumber, true);
     }
 
     await indexer.updateSyncStatusChainHead(blockProgress.blockHash, blockProgress.blockNumber, true);

packages/eden-watcher/src/gql/subscriptions/onEvent.gql

@@ -20,11 +20,14 @@ subscription onEvent{
                 from
                 to
                 value
+                tokenId
             }
             ... on ApprovalEvent {
                 owner
                 spender
                 value
+                approved
+                tokenId
             }
             ... on AuthorizationUsedEvent {
                 authorizer