Add a CLI in eden-watcher to fill state for a given range (#176)

* Add a CLI to fill state for a given range

* Refactor code

* Add a CLI to reset IPLD state

* Replace ORDER BY clause in the query to get latest IPLD block

* Optimize delete query in CLI to reset IPLD state

* Add an option to decouple subgraph state creation from mapping code

* Use a raw SQL query to delete IPLD blocks in a block range

* Accomodate changes in codegen

packages/codegen/src/client.ts

@@ -26,7 +26,6 @@ export class Client {
 
   /**
    * Stores the query to be passed to the template.
-   * @param mode Code generation mode.
    * @param name Name of the query.
    * @param params Parameters to the query.
    * @param returnType Return type for the query.

packages/codegen/src/fill.ts

@@ -7,15 +7,24 @@ import path from 'path';
 import Handlebars from 'handlebars';
 import { Writable } from 'stream';
 
-const TEMPLATE_FILE = './templates/fill-template.handlebars';
+const FILL_TEMPLATE_FILE = './templates/fill-template.handlebars';
+const FILL_STATE_TEMPLATE_FILE = './templates/fill-state-template.handlebars';
 
 /**
  * Writes the fill file generated from a template to a stream.
- * @param outStream A writable output stream to write the fill file to.
+ * @param fillOutStream A writable output stream to write the fill file to.
+ * @param fillStateOutStream A writable output stream to write the fill state file to.
  */
-export function exportFill (outStream: Writable, subgraphPath: string): void {
-  const templateString = fs.readFileSync(path.resolve(__dirname, TEMPLATE_FILE)).toString();
+export function exportFill (fillOutStream: Writable, fillStateOutStream: Writable | undefined, subgraphPath: string): void {
+  const templateString = fs.readFileSync(path.resolve(__dirname, FILL_TEMPLATE_FILE)).toString();
   const template = Handlebars.compile(templateString);
   const fill = template({ subgraphPath });
-  outStream.write(fill);
+  fillOutStream.write(fill);
+
+  if (fillStateOutStream) {
+    const templateString = fs.readFileSync(path.resolve(__dirname, FILL_STATE_TEMPLATE_FILE)).toString();
+    const template = Handlebars.compile(templateString);
+    const fillState = template({});
+    fillStateOutStream.write(fillState);
+  }
 }

packages/codegen/src/generate-code.ts

@@ -246,10 +246,18 @@ function generateWatcher (visitor: Visitor, contracts: any[], config: any) {
     : process.stdout;
   exportHooks(outStream);
 
-  outStream = outputDir
+  const fillOutStream = outputDir
     ? fs.createWriteStream(path.join(outputDir, 'src/fill.ts'))
     : process.stdout;
-  exportFill(outStream, config.subgraphPath);
+
+  let fillStateOutStream;
+  if (config.subgraphPath) {
+    fillStateOutStream = outputDir
+      ? fs.createWriteStream(path.join(outputDir, 'src/fill-state.ts'))
+      : process.stdout;
+  }
+
+  exportFill(fillOutStream, fillStateOutStream, config.subgraphPath);
 
   outStream = outputDir
     ? fs.createWriteStream(path.join(outputDir, 'src/types.ts'))
@@ -273,19 +281,25 @@ function generateWatcher (visitor: Visitor, contracts: any[], config: any) {
     : process.stdout;
   visitor.exportClient(outStream, schemaContent, path.join(outputDir, 'src/gql'));
 
-  let resetOutStream, resetJQOutStream, resetStateOutStream;
+  let resetOutStream, resetJQOutStream, resetStateOutStream, resetIPLDStateOutStream;
 
   if (outputDir) {
     resetOutStream = fs.createWriteStream(path.join(outputDir, 'src/cli/reset.ts'));
     resetJQOutStream = fs.createWriteStream(path.join(outputDir, 'src/cli/reset-cmds/job-queue.ts'));
     resetStateOutStream = fs.createWriteStream(path.join(outputDir, 'src/cli/reset-cmds/state.ts'));
+    if (config.subgraphPath) {
+      resetIPLDStateOutStream = fs.createWriteStream(path.join(outputDir, 'src/cli/reset-cmds/ipld-state.ts'));
+    }
   } else {
     resetOutStream = process.stdout;
     resetJQOutStream = process.stdout;
     resetStateOutStream = process.stdout;
+    if (config.subgraphPath) {
+      resetIPLDStateOutStream = process.stdout;
+    }
   }
 
-  visitor.exportReset(resetOutStream, resetJQOutStream, resetStateOutStream, config.subgraphPath);
+  visitor.exportReset(resetOutStream, resetJQOutStream, resetStateOutStream, resetIPLDStateOutStream, config.subgraphPath);
 
   outStream = outputDir
     ? fs.createWriteStream(path.join(outputDir, 'src/cli/export-state.ts'))

packages/codegen/src/indexer.ts

@@ -35,7 +35,7 @@ export class Indexer {
    * @param name Name of the query.
    * @param params Parameters to the query.
    * @param returnType Return type for the query.
-   * @param stateVariableTypeName Type of the state variable in case of state variable query.
+   * @param stateVariableType Type of the state variable in case of state variable query.
    */
   addQuery (contract: string, mode: string, name: string, params: Array<Param>, returnType: string, stateVariableType?: string): void {
     // Check if the query is already added.

packages/codegen/src/reset.ts

@@ -10,25 +10,26 @@ import { Writable } from 'stream';
 const RESET_TEMPLATE_FILE = './templates/reset-template.handlebars';
 const RESET_JQ_TEMPLATE_FILE = './templates/reset-job-queue-template.handlebars';
 const RESET_STATE_TEMPLATE_FILE = './templates/reset-state-template.handlebars';
+const RESET_IPLD_STATE_TEMPLATE_FILE = './templates/reset-ipld-state-template.handlebars';
 
 export class Reset {
   _queries: Array<any>;
   _resetTemplateString: string;
   _resetJQTemplateString: string;
   _resetStateTemplateString: string;
+  _resetIPLDStateTemplateString: string;
 
   constructor () {
     this._queries = [];
     this._resetTemplateString = fs.readFileSync(path.resolve(__dirname, RESET_TEMPLATE_FILE)).toString();
     this._resetJQTemplateString = fs.readFileSync(path.resolve(__dirname, RESET_JQ_TEMPLATE_FILE)).toString();
     this._resetStateTemplateString = fs.readFileSync(path.resolve(__dirname, RESET_STATE_TEMPLATE_FILE)).toString();
+    this._resetIPLDStateTemplateString = fs.readFileSync(path.resolve(__dirname, RESET_IPLD_STATE_TEMPLATE_FILE)).toString();
   }
 
   /**
    * Stores the query to be passed to the template.
    * @param name Name of the query.
-   * @param params Parameters to the query.
-   * @param returnType Return type for the query.
    */
   addQuery (name: string): void {
     // Check if the query is already added.
@@ -74,8 +75,9 @@ export class Reset {
    * @param resetOutStream A writable output stream to write the reset file to.
    * @param resetJQOutStream A writable output stream to write the reset job-queue file to.
    * @param resetStateOutStream A writable output stream to write the reset state file to.
+   * @param resetIPLDStateOutStream A writable output stream to write the reset IPLD state file to.
    */
-  exportReset (resetOutStream: Writable, resetJQOutStream: Writable, resetStateOutStream: Writable, subgraphPath: string): void {
+  exportReset (resetOutStream: Writable, resetJQOutStream: Writable, resetStateOutStream: Writable, resetIPLDStateOutStream: Writable | undefined, subgraphPath: string): void {
     const resetTemplate = Handlebars.compile(this._resetTemplateString);
     const resetString = resetTemplate({});
     resetOutStream.write(resetString);
@@ -91,5 +93,11 @@ export class Reset {
     };
     const resetState = resetStateTemplate(obj);
     resetStateOutStream.write(resetState);
+
+    if (resetIPLDStateOutStream) {
+      const resetIPLDStateTemplate = Handlebars.compile(this._resetIPLDStateTemplateString);
+      const resetIPLDStateString = resetIPLDStateTemplate({});
+      resetIPLDStateOutStream.write(resetIPLDStateString);
+    }
   }
 }

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

@@ -14,6 +14,12 @@
 
   {{#if (subgraphPath)}}
   subgraphPath = "{{subgraphPath}}"
+
+  # Disable creation of state from subgraph entity updates
+  # CAUTION: Disable only if subgraph state is not desired or can be filled subsequently
+  disableSubgraphState = false
+
+  # Interval to restart wasm instance periodically
   wasmRestartBlocksInterval = 20
 
   {{/if}}

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

@@ -120,6 +120,12 @@ export class Database implements IPLDDatabaseInterface {
     await this._baseDatabase.removeIPLDBlocks(repo, blockNumber, kind);
   }
 
+  async removeIPLDBlocksInRange (dbTx: QueryRunner, startBlock: number, endBlock: number): Promise<void> {
+    const repo = dbTx.manager.getRepository(IPLDBlock);
+
+    await this._baseDatabase.removeIPLDBlocksInRange(repo, startBlock, endBlock);
+  }
+
   async getIPLDStatus (): Promise<IpldStatus | undefined> {
     const repo = this._conn.getRepository(IpldStatus);
 

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

@@ -0,0 +1,96 @@
+//
+// Copyright 2022 Vulcanize, Inc.
+//
+
+import 'reflect-metadata';
+import debug from 'debug';
+import { Between } from 'typeorm';
+
+import { Database as GraphDatabase, prepareEntityState } from '@vulcanize/graph-node';
+
+import { Indexer } from './indexer';
+
+const log = debug('vulcanize:fill-state');
+
+export const fillState = async (
+  indexer: Indexer,
+  graphDb: GraphDatabase,
+  dataSources: any[],
+  argv: {
+    startBlock: number,
+    endBlock: number
+  }
+): Promise<void> => {
+  const { startBlock, endBlock } = argv;
+  if (startBlock > endBlock) {
+    log('endBlock should be greater than or equal to startBlock');
+    process.exit(1);
+  }
+
+  // NOTE: Assuming all blocks in the given range are in the pruned region
+  log(`Filling state for subgraph entities in range: [${startBlock}, ${endBlock}]`);
+
+  // Check that there are no existing diffs in this range
+  const existingStates = await indexer.getIPLDBlocks({ block: { blockNumber: Between(startBlock, endBlock) } });
+  if (existingStates.length > 0) {
+    log('found existing state(s) in the given range');
+    process.exit(1);
+  }
+
+  // Map: contractAddress -> entities updated
+  const contractEntitiesMap: Map<string, string[]> = new Map();
+
+  // Populate contractEntitiesMap using data sources from subgraph
+  // NOTE: Assuming each entity type is only mapped to a single contract
+  //       This is true for eden subgraph; may not be the case for other subgraphs
+  dataSources.forEach((dataSource: any) => {
+    const { source: { address: contractAddress }, mapping: { entities } } = dataSource;
+    contractEntitiesMap.set(contractAddress, entities as string[]);
+  });
+
+  // Fill state for blocks in the given range
+  for (let blockNumber = startBlock; blockNumber <= endBlock; blockNumber++) {
+    // Get the canonical block hash at current height
+    const blocks = await indexer.getBlocksAtHeight(blockNumber, false);
+
+    if (blocks.length === 0) {
+      log(`block not found at height ${blockNumber}`);
+      process.exit(1);
+    } else if (blocks.length > 1) {
+      log(`found more than one non-pruned block at height ${blockNumber}`);
+      process.exit(1);
+    }
+
+    const blockHash = blocks[0].blockHash;
+
+    // Fill state for each contract in contractEntitiesMap
+    const contractStatePromises = Array.from(contractEntitiesMap.entries())
+      .map(async ([contractAddress, entities]): Promise<void> => {
+        // Get all the updated entities at this block
+        const updatedEntitiesListPromises = entities.map(async (entity): Promise<any[]> => {
+          return graphDb.getEntitiesForBlock(blockHash, entity);
+        });
+        const updatedEntitiesList = await Promise.all(updatedEntitiesListPromises);
+
+        // Populate state with all the updated entities of each entity type
+        updatedEntitiesList.forEach((updatedEntities, index) => {
+          const entityName = entities[index];
+
+          updatedEntities.forEach((updatedEntity) => {
+            // Prepare diff data for the entity update
+            const diffData = prepareEntityState(updatedEntity, entityName, indexer.getRelationsMap());
+
+            // Update the in-memory subgraph state
+            indexer.updateSubgraphState(contractAddress, diffData);
+          });
+        });
+      });
+
+    await Promise.all(contractStatePromises);
+
+    // Persist subgraph state to the DB
+    await indexer.dumpSubgraphState(blockHash, true);
+  }
+
+  log(`Filled state for subgraph entities in range: [${startBlock}, ${endBlock}]`);
+};

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

@@ -20,6 +20,9 @@ import { GraphWatcher, Database as GraphDatabase } from '@vulcanize/graph-node';
 import { Database } from './database';
 import { Indexer } from './indexer';
 import { EventWatcher } from './events';
+{{#if (subgraphPath)}}
+import { fillState } from './fill-state';
+{{/if}}
 
 const log = debug('vulcanize:server');
 
@@ -41,6 +44,13 @@ export const main = async (): Promise<any> => {
       demandOption: true,
       describe: 'Block number to start processing at'
     },
+    {{#if (subgraphPath)}}
+    state: {
+      type: 'boolean',
+      default: false,
+      describe: 'Fill state for subgraph entities'
+    },
+    {{/if}}
     endBlock: {
       type: 'number',
       demandOption: true,
@@ -86,6 +96,11 @@ export const main = async (): Promise<any> => {
 
   graphWatcher.setIndexer(indexer);
   await graphWatcher.init();
+
+  if (argv.state) {
+    await fillState(indexer, graphDb, graphWatcher.dataSources, argv);
+    return;
+  }
   {{/if}}
 
   // Note: In-memory pubsub works fine for now, as each watcher is a single process anyway.

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

@@ -386,6 +386,10 @@ export class Indexer implements IPLDIndexerInterface {
     return this._baseIndexer.getIPLDBlockByCid(cid);
   }
 
+  async getIPLDBlocks (where: FindConditions<IPLDBlock>): Promise<IPLDBlock[]> {
+    return this._db.getIPLDBlocks(where);
+  }
+
   getIPLDData (ipldBlock: IPLDBlock): any {
     return this._baseIndexer.getIPLDData(ipldBlock);
   }
@@ -469,7 +473,7 @@ export class Indexer implements IPLDIndexerInterface {
     await this._graphWatcher.handleBlock(blockHash);
 
     // Persist subgraph state to the DB.
-    await this._dumpSubgraphState(blockHash);
+    await this.dumpSubgraphState(blockHash);
   }
 
   {{/if}}
@@ -673,6 +677,23 @@ export class Indexer implements IPLDIndexerInterface {
     this._subgraphStateMap.set(contractAddress, updatedData);
   }
 
+  async dumpSubgraphState (blockHash: string, isStateFinalized = false): Promise<void> {
+    // Create a diff for each contract in the subgraph state map.
+    const createDiffPromises = Array.from(this._subgraphStateMap.entries())
+      .map(([contractAddress, data]): Promise<void> => {
+        if (isStateFinalized) {
+          return this.createDiff(contractAddress, blockHash, data);
+        }
+
+        return this.createDiffStaged(contractAddress, blockHash, data);
+      });
+
+    await Promise.all(createDiffPromises);
+
+    // Reset the subgraph state map.
+    this._subgraphStateMap.clear();
+  }
+
   _populateEntityTypesMap (): void {
   {{#each subgraphEntities as | subgraphEntity |}}
     this._entityTypesMap.set('{{subgraphEntity.className}}', {
@@ -710,19 +731,6 @@ export class Indexer implements IPLDIndexerInterface {
     {{/if}}
   {{/each}}
   }
-
-  async _dumpSubgraphState (blockHash: string): Promise<void> {
-    // Create a diff for each contract in the subgraph state map.
-    const createDiffPromises = Array.from(this._subgraphStateMap.entries())
-      .map(([contractAddress, data]): Promise<void> => {
-        return this.createDiffStaged(contractAddress, blockHash, data);
-      });
-
-    await Promise.all(createDiffPromises);
-
-    // Reset the subgraph state map.
-    this._subgraphStateMap.clear();
-  }
   {{/if}}
 
   async _fetchAndSaveEvents ({ cid: blockCid, blockHash }: DeepPartial<BlockProgress>): Promise<BlockProgress> {

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

@@ -15,6 +15,9 @@
     "job-runner:dev": "DEBUG=vulcanize:* ts-node src/job-runner.ts",
     "watch:contract": "DEBUG=vulcanize:* ts-node src/cli/watch-contract.ts",
     "fill": "DEBUG=vulcanize:* ts-node src/fill.ts",
+    {{#if (subgraphPath)}}
+    "fill:state": "DEBUG=vulcanize:* ts-node src/fill.ts --state",
+    {{/if}}
     "reset": "DEBUG=vulcanize:* ts-node src/cli/reset.ts",
     "checkpoint": "DEBUG=vulcanize:* ts-node src/cli/checkpoint.ts",
     "export-state": "DEBUG=vulcanize:* ts-node src/cli/export-state.ts",

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

@@ -0,0 +1,55 @@
+//
+// Copyright 2022 Vulcanize, Inc.
+//
+
+import debug from 'debug';
+
+import { getConfig } from '@vulcanize/util';
+
+import { Database } from '../../database';
+
+const log = debug('vulcanize:reset-ipld-state');
+
+export const command = 'ipld-state';
+
+export const desc = 'Reset IPLD state in the given range';
+
+export const builder = {
+  startBlock: {
+    type: 'number'
+  },
+  endBlock: {
+    type: 'number'
+  }
+};
+
+export const handler = async (argv: any): Promise<void> => {
+  const { startBlock, endBlock } = argv;
+  if (startBlock > endBlock) {
+    log('endBlock should be greater than or equal to startBlock');
+    process.exit(1);
+  }
+
+  const config = await getConfig(argv.configFile);
+
+  // Initialize database
+  const db = new Database(config.database);
+  await db.init();
+
+  // Create a DB transaction
+  const dbTx = await db.createTransactionRunner();
+
+  try {
+    // Delete all IPLDBlock entries in the given range
+    await db.removeIPLDBlocksInRange(dbTx, startBlock, endBlock);
+
+    dbTx.commitTransaction();
+  } catch (error) {
+    await dbTx.rollbackTransaction();
+    throw error;
+  } finally {
+    await dbTx.release();
+  }
+
+  log(`Reset ipld-state successfully for range [${startBlock}, ${endBlock}]`);
+};

packages/codegen/src/visitor.ts

@@ -215,8 +215,8 @@ export class Visitor {
    * @param resetJQOutStream A writable output stream to write the reset job-queue file to.
    * @param resetStateOutStream A writable output stream to write the reset state file to.
    */
-  exportReset (resetOutStream: Writable, resetJQOutStream: Writable, resetStateOutStream: Writable, subgraphPath: string): void {
-    this._reset.exportReset(resetOutStream, resetJQOutStream, resetStateOutStream, subgraphPath);
+  exportReset (resetOutStream: Writable, resetJQOutStream: Writable, resetStateOutStream: Writable, resetIPLDStateOutStream: Writable | undefined, subgraphPath: string): void {
+    this._reset.exportReset(resetOutStream, resetJQOutStream, resetStateOutStream, resetIPLDStateOutStream, subgraphPath);
   }
 
   /**