Implement IPLD refactoring in codegen (#85)

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

@@ -35,9 +35,13 @@ columns:
     tsType: string
     columnType: Column
   - name: kind
-    pgType: varchar
-    tsType: string
+    tsType: StateKind
     columnType: Column
+    columnOptions:
+      - option: type
+        value: "'enum'"
+      - option: enum
+        value: StateKind
   - name: data
     pgType: bytea
     tsType: Buffer
@@ -50,6 +54,9 @@ imports:
       - Index
       - ManyToOne
     from: typeorm
+  - toImport:
+      - StateKind
+    from: '@vulcanize/util'
   - toImport:
       - BlockProgress
     from: ./BlockProgress

packages/codegen/src/generate-code.ts

@@ -32,7 +32,6 @@ import { exportFill } from './fill';
 import { exportCheckpoint } from './checkpoint';
 import { exportState } from './export-state';
 import { importState } from './import-state';
-import { exportIPFS } from './ipfs';
 import { exportInspectCID } from './inspect-cid';
 
 const main = async (): Promise<void> => {
@@ -305,11 +304,6 @@ function generateWatcher (contractStrings: string[], visitor: Visitor, argv: any
     : process.stdout;
   importState(outStream);
 
-  outStream = outputDir
-    ? fs.createWriteStream(path.join(outputDir, 'src/ipfs.ts'))
-    : process.stdout;
-  exportIPFS(outStream);
-
   outStream = outputDir
     ? fs.createWriteStream(path.join(outputDir, 'src/cli/inspect-cid.ts'))
     : process.stdout;

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

@@ -3,10 +3,10 @@
 //
 
 import assert from 'assert';
-import { Connection, ConnectionOptions, DeepPartial, FindConditions, QueryRunner, FindManyOptions, MoreThan } from 'typeorm';
+import { Connection, ConnectionOptions, DeepPartial, FindConditions, QueryRunner, FindManyOptions } from 'typeorm';
 import path from 'path';
 
-import { Database as BaseDatabase, QueryOptions, Where, MAX_REORG_DEPTH, DatabaseInterface } from '@vulcanize/util';
+import { IPLDDatabase as BaseDatabase, IPLDDatabaseInterface, QueryOptions, StateKind, Where } from '@vulcanize/util';
 
 import { Contract } from './entity/Contract';
 import { Event } from './entity/Event';
@@ -17,9 +17,11 @@ import { IPLDBlock } from './entity/IPLDBlock';
 
 {{#each queries as | query |}}
 import { {{query.entityName}} } from './entity/{{query.entityName}}';
-{{/each}}
+{{#unless @last}}
 
-export class Database implements DatabaseInterface {
+{{/unless}}
+{{/each}}
+export class Database implements IPLDDatabaseInterface {
   _config: ConnectionOptions;
   _conn!: Connection;
   _baseDatabase: BaseDatabase;
@@ -78,168 +80,63 @@ export class Database implements DatabaseInterface {
   }
 
   {{/each}}
+  getNewIPLDBlock (): IPLDBlock {
+    return new IPLDBlock();
+  }
+
   async getIPLDBlocks (where: FindConditions<IPLDBlock>): Promise<IPLDBlock[]> {
     const repo = this._conn.getRepository(IPLDBlock);
-    return repo.find({ where, relations: ['block'] });
+
+    return this._baseDatabase.getIPLDBlocks(repo, where);
   }
 
-  async getLatestIPLDBlock (contractAddress: string, kind: string | null, blockNumber?: number): Promise<IPLDBlock | undefined> {
+  async getLatestIPLDBlock (contractAddress: string, kind: StateKind | null, blockNumber?: number): Promise<IPLDBlock | undefined> {
     const repo = this._conn.getRepository(IPLDBlock);
 
-    let queryBuilder = repo.createQueryBuilder('ipld_block')
-      .leftJoinAndSelect('ipld_block.block', 'block')
-      .where('block.is_pruned = false')
-      .andWhere('ipld_block.contract_address = :contractAddress', { contractAddress })
-      .orderBy('block.block_number', 'DESC');
-
-    // Filter out blocks after the provided block number.
-    if (blockNumber) {
-      queryBuilder.andWhere('block.block_number <= :blockNumber', { blockNumber });
-    }
-
-    // Filter using kind if specified else order by id to give preference to checkpoint.
-    queryBuilder = kind
-      ? queryBuilder.andWhere('ipld_block.kind = :kind', { kind })
-      : queryBuilder.andWhere('ipld_block.kind != :kind', { kind: 'diff_staged' })
-        .addOrderBy('ipld_block.id', 'DESC');
-
-    return queryBuilder.getOne();
-  }
-
-  async getPrevIPLDBlock (queryRunner: QueryRunner, blockHash: string, contractAddress: string, kind?: string): Promise<IPLDBlock | undefined> {
-    const heirerchicalQuery = `
-      WITH RECURSIVE cte_query AS
-      (
-        SELECT
-          b.block_hash,
-          b.block_number,
-          b.parent_hash,
-          1 as depth,
-          i.id,
-          i.kind
-        FROM
-          block_progress b
-          LEFT JOIN
-            ipld_block i ON i.block_id = b.id
-            AND i.contract_address = $2
-        WHERE
-          b.block_hash = $1
-        UNION ALL
-          SELECT
-            b.block_hash,
-            b.block_number,
-            b.parent_hash,
-            c.depth + 1,
-            i.id,
-            i.kind
-          FROM
-            block_progress b
-            LEFT JOIN
-              ipld_block i
-              ON i.block_id = b.id
-              AND i.contract_address = $2
-            INNER JOIN
-              cte_query c ON c.parent_hash = b.block_hash
-            WHERE
-              c.depth < $3
-      )
-      SELECT
-        block_number, id, kind
-      FROM
-        cte_query
-      ORDER BY block_number DESC, id DESC
-    `;
-
-    // Fetching block and id for previous IPLDBlock in frothy region.
-    const queryResult = await queryRunner.query(heirerchicalQuery, [blockHash, contractAddress, MAX_REORG_DEPTH]);
-    const latestRequiredResult = kind
-      ? queryResult.find((obj: any) => obj.kind === kind)
-      : queryResult.find((obj: any) => obj.id);
-
-    let result: IPLDBlock | undefined;
-    if (latestRequiredResult) {
-      result = await queryRunner.manager.findOne(IPLDBlock, { id: latestRequiredResult.id }, { relations: ['block'] });
-    } else {
-      // If IPLDBlock not found in frothy region get latest IPLDBlock in the pruned region.
-      // Filter out IPLDBlocks from pruned blocks.
-      const canonicalBlockNumber = queryResult.pop().block_number + 1;
-
-      let queryBuilder = queryRunner.manager.createQueryBuilder(IPLDBlock, 'ipld_block')
-        .leftJoinAndSelect('ipld_block.block', 'block')
-        .where('block.is_pruned = false')
-        .andWhere('ipld_block.contract_address = :contractAddress', { contractAddress })
-        .andWhere('block.block_number <= :canonicalBlockNumber', { canonicalBlockNumber })
-        .orderBy('block.block_number', 'DESC');
-
-      // Filter using kind if specified else order by id to give preference to checkpoint.
-      queryBuilder = kind
-        ? queryBuilder.andWhere('ipld_block.kind = :kind', { kind })
-        : queryBuilder.addOrderBy('ipld_block.id', 'DESC');
-
-      result = await queryBuilder.getOne();
-    }
-
-    return result;
-  }
-
-  // Fetch all diff IPLDBlocks after the specified checkpoint.
-  async getDiffIPLDBlocksByCheckpoint (contractAddress: string, checkpointBlockNumber: number): Promise<IPLDBlock[]> {
+    return this._baseDatabase.getLatestIPLDBlock(repo, contractAddress, kind, blockNumber);
+  }
+
+  async getPrevIPLDBlock (blockHash: string, contractAddress: string, kind?: string): Promise<IPLDBlock | undefined> {
     const repo = this._conn.getRepository(IPLDBlock);
 
-    return repo.find({
-      relations: ['block'],
-      where: {
-        contractAddress,
-        kind: 'diff',
-        block: {
-          isPruned: false,
-          blockNumber: MoreThan(checkpointBlockNumber)
-        }
-      },
-      order: {
-        block: 'ASC'
-      }
-    });
-  }
-
-  async saveOrUpdateIPLDBlock (ipldBlock: IPLDBlock): Promise<IPLDBlock> {
+    return this._baseDatabase.getPrevIPLDBlock(repo, blockHash, contractAddress, kind);
+  }
+
+  // Fetch all diff IPLDBlocks after the specified block number.
+  async getDiffIPLDBlocksByBlocknumber (contractAddress: string, blockNumber: number): Promise<IPLDBlock[]> {
     const repo = this._conn.getRepository(IPLDBlock);
-    return repo.save(ipldBlock);
+
+    return this._baseDatabase.getDiffIPLDBlocksByBlocknumber(repo, contractAddress, blockNumber);
   }
 
-  async getHookStatus (queryRunner: QueryRunner): Promise<HookStatus | undefined> {
-    const repo = queryRunner.manager.getRepository(HookStatus);
+  async saveOrUpdateIPLDBlock (dbTx: QueryRunner, ipldBlock: IPLDBlock): Promise<IPLDBlock> {
+    const repo = dbTx.manager.getRepository(IPLDBlock);
 
-    return repo.findOne();
+    return this._baseDatabase.saveOrUpdateIPLDBlock(repo, ipldBlock);
   }
 
-  async updateHookStatusProcessedBlock (queryRunner: QueryRunner, blockNumber: number, force?: boolean): Promise<HookStatus> {
-    const repo = queryRunner.manager.getRepository(HookStatus);
-    let entity = await repo.findOne();
+  async removeIPLDBlocks (dbTx: QueryRunner, blockNumber: number, kind: string): Promise<void> {
+    const repo = dbTx.manager.getRepository(IPLDBlock);
 
-    if (!entity) {
-      entity = repo.create({
-        latestProcessedBlockNumber: blockNumber
-      });
-    }
+    await this._baseDatabase.removeIPLDBlocks(repo, blockNumber, kind);
+  }
 
-    if (force || blockNumber > entity.latestProcessedBlockNumber) {
-      entity.latestProcessedBlockNumber = blockNumber;
-    }
+  async getHookStatus (): Promise<HookStatus | undefined> {
+    const repo = this._conn.getRepository(HookStatus);
 
-    return repo.save(entity);
+    return this._baseDatabase.getHookStatus(repo);
   }
 
-  async getContracts (): Promise<Contract[]> {
-    const repo = this._conn.getRepository(Contract);
+  async updateHookStatusProcessedBlock (queryRunner: QueryRunner, blockNumber: number, force?: boolean): Promise<HookStatus> {
+    const repo = queryRunner.manager.getRepository(HookStatus);
 
-    return this._baseDatabase.getContracts(repo);
+    return this._baseDatabase.updateHookStatusProcessedBlock(repo, blockNumber, force);
   }
 
-  async getContract (address: string): Promise<Contract | undefined> {
+  async getContracts (): Promise<Contract[]> {
     const repo = this._conn.getRepository(Contract);
 
-    return this._baseDatabase.getContract(repo, address);
+    return this._baseDatabase.getContracts(repo);
   }
 
   async createTransactionRunner (): Promise<QueryRunner> {
@@ -276,7 +173,7 @@ export class Database implements DatabaseInterface {
     return this._baseDatabase.saveEvents(blockRepo, eventRepo, block, events);
   }
 
-  async saveContract (address: string, kind: string, checkpoint: boolean, startingBlock: number): Promise<Contract> {
+  async saveContract (queryRunner: QueryRunner, address: string, kind: string, checkpoint: boolean, startingBlock: number): Promise<Contract> {
     const repo = queryRunner.manager.getRepository(Contract);
 
     return this._baseDatabase.saveContract(repo, address, kind, checkpoint, startingBlock);

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

@@ -20,12 +20,14 @@ export class {{className}} {{~#if implements}} implements {{implements}} {{~/if}
   {{#each columns as | column |}}
   {{#if (compare column.columnType 'ManyToOne')}}
   @{{column.columnType}}({{column.lhs}} => {{column.rhs}}
+  {{~#if column.columnOptions}}, {{/if}}
   {{~else}}
   @{{column.columnType}}(
   {{~#if column.pgType~}} '{{column.pgType}}'
+  {{~#if column.columnOptions}}, {{/if}}
   {{~/if}}
   {{~/if}}
-  {{~#if column.columnOptions}}, {
+  {{~#if column.columnOptions}}{
   {{~#each column.columnOptions}} {{this.option}}: {{{this.value}}}
   {{~#unless @last}},{{/unless}}
   {{~/each}} }

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

@@ -80,7 +80,10 @@ export class EventWatcher implements EventWatcherInterface {
 
       await this._baseEventWatcher.blockProcessingCompleteHandler(job);
 
-      await this.createHooksJob(kind);
+      // If it's a pruning job: Create a hooks job.
+      if (kind === JOB_KIND_PRUNE) {
+        await this.createHooksJob();
+      }
     });
   }
 
@@ -148,20 +151,18 @@ export class EventWatcher implements EventWatcherInterface {
     }
   }
 
-  async createHooksJob (kind: string): Promise<void> {
-    // If it's a pruning job: Create a hook job for the latest canonical block.
-    if (kind === JOB_KIND_PRUNE) {
-      const latestCanonicalBlock = await this._indexer.getLatestCanonicalBlock();
-      assert(latestCanonicalBlock);
-
-      await this._jobQueue.pushJob(
-        QUEUE_HOOKS,
-        {
-          blockHash: latestCanonicalBlock.blockHash,
-          blockNumber: latestCanonicalBlock.blockNumber
-        }
-      );
-    }
+  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> {

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

@@ -61,7 +61,7 @@ export const main = async (): Promise<any> => {
   const graphDb = new GraphDatabase(config.database, path.resolve(__dirname, 'entity/*'));
   await graphDb.init();
 
-  const graphWatcher = new GraphWatcher(graphDb, postgraphileClient, ethProvider, config.server.subgraphPath);
+  const graphWatcher = new GraphWatcher(graphDb, postgraphileClient, ethProvider, config.server);
 
   // Note: In-memory pubsub works fine for now, as each watcher is a single process anyway.
   // Later: https://www.apollographql.com/docs/apollo-server/data/subscriptions/#production-pubsub-libraries