Update graph-test-watcher (#45)

2026-09-09 17:24:07 +00:00 · 2021-12-28 16:08:05 +05:30
parent 16bb955213
commit 73dcb882a6
40 changed files with 2051 additions and 302 deletions
@@ -8,6 +8,12 @@
  yarn
  ```

+* Run the IPFS (go-ipfs version 0.9.0) daemon:
+
+  ```bash
+  ipfs daemon
+  ```
+
 * Create a postgres12 database for the watcher:

  ```bash
@@ -35,9 +41,13 @@
  graph-test-watcher-job-queue=# exit
  ```

-* Update the [config](./environments/local.toml) with database connection settings.
+* In the [config file](./environments/local.toml):

-* Update the `upstream` config in the [config file](./environments/local.toml) and provide the `ipld-eth-server` GQL API and the `indexer-db` postgraphile endpoints.
+  * Update the database connection settings.
+
+  * Update the `upstream` config and provide the `ipld-eth-server` GQL API and the `indexer-db` postgraphile endpoints.
+
+  * Update the `server` config with state checkpoint settings and provide the IPFS API address.

 ## Customize

@@ -45,7 +55,17 @@

  * Edit the custom hook function `handleEvent` (triggered on an event) in [hooks.ts](./src/hooks.ts) to perform corresponding indexing using the `Indexer` object.

-  * Refer to [hooks.example.ts](./src/hooks.example.ts) for an example hook function for events in an ERC20 contract.
+  * While using the indexer storage methods for indexing, pass `diff` as true if default state is desired to be generated using the state variables being indexed.
+
+* Generating state:
+
+  * Edit the custom hook function `createInitialCheckpoint` (triggered on watch-contract, checkpoint: `true`) in [hooks.ts](./src/hooks.ts) to save an initial checkpoint `IPLDBlock` using the `Indexer` object.
+
+  * Edit the custom hook function `createStateDiff` (triggered on a block) in [hooks.ts](./src/hooks.ts) to save the state in a `diff` `IPLDBlock` using the `Indexer` object. The default state (if exists) is updated.
+
+  * Edit the custom hook function `createStateCheckpoint` (triggered just before default and CLI checkpoint) in [hooks.ts](./src/hooks.ts) to save the state in a `checkpoint` `IPLDBlock` using the `Indexer` object.
+
+* The existing example hooks in [hooks.ts](./src/hooks.ts) are for an `ERC20` contract.

 ## Run

@@ -68,11 +88,104 @@ GQL console: http://localhost:3008/graphql
  * To watch a contract:

    ```bash
-    yarn watch:contract --address <contract-address> --kind Example --starting-block [block-number]
+    yarn watch:contract --address <contract-address> --kind <contract-kind> --checkpoint <true | false> --starting-block [block-number]
+    ```
+
+    * `address`: Address or identifier of the contract to be watched.
+    * `kind`: Kind of the contract.
+    * `checkpoint`: Turn checkpointing on (`true` | `false`).
+    * `starting-block`: Starting block for the contract (default: `1`).
+
+    Examples:
+    
+    Watch a contract with its address and checkpointing on:
+
+    ```bash
+    yarn watch:contract --address 0x1F78641644feB8b64642e833cE4AFE93DD6e7833 --kind ERC20 --checkpoint true
+    ```
+
+    Watch a contract with its identifier and checkpointing on:
+
+    ```bash
+    yarn watch:contract --address MyProtocol --kind protocol --checkpoint true
    ```

  * To fill a block range:

    ```bash
-    yarn fill --startBlock <from-block> --endBlock <to-block>
+    yarn fill --start-block <from-block> --end-block <to-block>
    ```
+
+    * `start-block`: Block number to start filling from.
+    * `end-block`: Block number till which to fill. 
+
+  * To create a checkpoint for a contract:
+
+    ```bash
+    yarn checkpoint --address <contract-address> --block-hash [block-hash]
+    ```
+
+    * `address`: Address or identifier of the contract for which to create a checkpoint.
+    * `block-hash`: Hash of a block (in the pruned region) at which to create the checkpoint (default: latest canonical block hash).
+
+  * To reset the watcher to a previous block number:
+
+    * Reset state:
+
+      ```bash
+      yarn reset state --block-number <previous-block-number>
+      ```
+
+    * Reset job-queue:
+
+      ```bash
+      yarn reset job-queue --block-number <previous-block-number>
+      ```
+    
+    * `block-number`: Block number to which to reset the watcher.
+
+  * To export and import the watcher state:
+
+    * In source watcher, export watcher state:
+
+      ```bash
+      yarn export-state --export-file [export-file-path]
+      ```
+
+      * `export-file`: Path of JSON file to which to export the watcher data.
+
+    * In target watcher, run job-runner:
+
+      ```bash
+      yarn job-runner
+      ```
+    
+    * Import watcher state:
+
+      ```bash
+      yarn import-state --import-file <import-file-path>
+      ```
+
+      * `import-file`: Path of JSON file from which to import the watcher data.
+    
+    * Run fill:
+
+      ```bash
+      yarn fill --start-block <snapshot-block> --end-block <to-block>
+      ```
+
+      * `snapshot-block`: Block number at which the watcher state was exported.
+
+    * Run server:
+
+      ```bash
+      yarn server
+      ```
+
+  * To inspect a CID:
+
+    ```bash
+    yarn inspect-cid --cid <cid>
+    ```
+
+    * `cid`: CID to be inspected.
@@ -2,6 +2,16 @@
  host = "127.0.0.1"
  port = 3008
  kind = "active"
+
+  # Checkpointing state.
+  checkpointing = true
+
+  # Checkpoint interval in number of blocks.
+  checkpointInterval = 2000
+
+  # IPFS API address (can be taken from the output on running the IPFS daemon).
+  ipfsApiAddr = "/ip4/127.0.0.1/tcp/5001"
+  
  subgraphPath = "../graph-node/test/subgraph/example1/build"

 [database]
@@ -10,7 +10,12 @@
    "server": "DEBUG=vulcanize:* ts-node src/server.ts",
    "job-runner": "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"
+    "fill": "DEBUG=vulcanize:* ts-node src/fill.ts",
+    "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",
+    "import-state": "DEBUG=vulcanize:* ts-node src/cli/import-state.ts",
+    "inspect-cid": "DEBUG=vulcanize:* ts-node src/cli/inspect-cid.ts"
  },
  "repository": {
    "type": "git",
@@ -23,9 +28,10 @@
  },
  "homepage": "https://github.com/vulcanize/watcher-ts#readme",
  "dependencies": {
+    "@apollo/client": "^3.3.19",
    "@ethersproject/providers": "5.3.0",
+    "@ipld/dag-cbor": "^6.0.12",
    "@vulcanize/cache": "^0.1.0",
-    "@vulcanize/graph-node": "^0.1.0",
    "@vulcanize/ipld-eth-client": "^0.1.0",
    "@vulcanize/solidity-mapper": "^0.1.0",
    "@vulcanize/util": "^0.1.0",
@@ -36,7 +42,10 @@
    "express": "^4.17.1",
    "graphql": "^15.5.0",
    "graphql-import-node": "^0.0.4",
+    "ipfs-http-client": "^53.0.1",
    "json-bigint": "^1.0.0",
+    "lodash": "^4.17.21",
+    "multiformats": "^9.4.8",
    "reflect-metadata": "^0.1.13",
    "typeorm": "^0.2.32",
    "yargs": "^17.0.1"
@@ -0,0 +1,69 @@
+//
+// Copyright 2021 Vulcanize, Inc.
+//
+
+import path from 'path';
+import yargs from 'yargs';
+import 'reflect-metadata';
+import debug from 'debug';
+
+import { Config, DEFAULT_CONFIG_PATH, getConfig, initClients } from '@vulcanize/util';
+import { GraphWatcher, Database as GraphDatabase } from '@vulcanize/graph-node';
+
+import { Database } from '../database';
+import { Indexer } from '../indexer';
+
+const log = debug('vulcanize:checkpoint');
+
+const main = async (): Promise<void> => {
+  const argv = await yargs.parserConfiguration({
+    'parse-numbers': false
+  }).options({
+    configFile: {
+      alias: 'f',
+      type: 'string',
+      require: true,
+      demandOption: true,
+      describe: 'Configuration file path (toml)',
+      default: DEFAULT_CONFIG_PATH
+    },
+    address: {
+      type: 'string',
+      require: true,
+      demandOption: true,
+      describe: 'Contract address to create the checkpoint for.'
+    },
+    blockHash: {
+      type: 'string',
+      describe: 'Blockhash at which to create the checkpoint.'
+    }
+  }).argv;
+
+  const config: Config = await getConfig(argv.configFile);
+  const { ethClient, postgraphileClient, ethProvider } = await initClients(config);
+
+  const db = new Database(config.database);
+  await db.init();
+
+  const graphDb = new GraphDatabase(config.database, path.resolve(__dirname, 'entity/*'));
+  await graphDb.init();
+
+  const graphWatcher = new GraphWatcher(graphDb, postgraphileClient, config.server.subgraphPath);
+
+  const indexer = new Indexer(config.server, db, ethClient, postgraphileClient, ethProvider, graphWatcher);
+
+  graphWatcher.setIndexer(indexer);
+  await graphWatcher.init();
+
+  const blockHash = await indexer.processCLICheckpoint(argv.address, argv.blockHash);
+
+  log(`Created a checkpoint for contract ${argv.address} at block-hash ${blockHash}`);
+
+  await db.close();
+};
+
+main().catch(err => {
+  log(err);
+}).finally(() => {
+  process.exit(0);
+});
@@ -0,0 +1,119 @@
+//
+// Copyright 2021 Vulcanize, Inc.
+//
+
+import assert from 'assert';
+import yargs from 'yargs';
+import 'reflect-metadata';
+import debug from 'debug';
+import fs from 'fs';
+import path from 'path';
+
+import { Config, DEFAULT_CONFIG_PATH, getConfig, initClients } from '@vulcanize/util';
+import { GraphWatcher, Database as GraphDatabase } from '@vulcanize/graph-node';
+import * as codec from '@ipld/dag-cbor';
+
+import { Database } from '../database';
+import { Indexer } from '../indexer';
+
+const log = debug('vulcanize:export-state');
+
+const main = async (): Promise<void> => {
+  const argv = await yargs.parserConfiguration({
+    'parse-numbers': false
+  }).options({
+    configFile: {
+      alias: 'f',
+      type: 'string',
+      require: true,
+      demandOption: true,
+      describe: 'Configuration file path (toml)',
+      default: DEFAULT_CONFIG_PATH
+    },
+    exportFile: {
+      alias: 'o',
+      type: 'string',
+      describe: 'Export file path'
+    }
+  }).argv;
+
+  const config: Config = await getConfig(argv.configFile);
+  const { ethClient, postgraphileClient, ethProvider } = await initClients(config);
+
+  const db = new Database(config.database);
+  await db.init();
+
+  const graphDb = new GraphDatabase(config.database, path.resolve(__dirname, 'entity/*'));
+  await graphDb.init();
+
+  const graphWatcher = new GraphWatcher(graphDb, postgraphileClient, config.server.subgraphPath);
+
+  const indexer = new Indexer(config.server, db, ethClient, postgraphileClient, ethProvider, graphWatcher);
+
+  graphWatcher.setIndexer(indexer);
+  await graphWatcher.init();
+
+  const exportData: any = {
+    snapshotBlock: {},
+    contracts: [],
+    ipldCheckpoints: []
+  };
+
+  const contracts = await db.getContracts({});
+
+  // Get latest canonical block.
+  const block = await indexer.getLatestCanonicalBlock();
+  assert(block);
+
+  // Export snapshot block.
+  exportData.snapshotBlock = {
+    blockNumber: block.blockNumber,
+    blockHash: block.blockHash
+  };
+
+  // Export contracts and checkpoints.
+  for (const contract of contracts) {
+    exportData.contracts.push({
+      address: contract.address,
+      kind: contract.kind,
+      checkpoint: contract.checkpoint,
+      startingBlock: block.blockNumber
+    });
+
+    // Create and export checkpoint if checkpointing is on for the contract.
+    if (contract.checkpoint) {
+      await indexer.createCheckpoint(contract.address, block.blockHash);
+
+      const ipldBlock = await indexer.getLatestIPLDBlock(contract.address, 'checkpoint', block.blockNumber);
+      assert(ipldBlock);
+
+      const data = codec.decode(Buffer.from(ipldBlock.data)) as any;
+
+      exportData.ipldCheckpoints.push({
+        contractAddress: ipldBlock.contractAddress,
+        cid: ipldBlock.cid,
+        kind: ipldBlock.kind,
+        data
+      });
+    }
+  }
+
+  if (argv.exportFile) {
+    const encodedExportData = codec.encode(exportData);
+
+    const filePath = path.resolve(argv.exportFile);
+    const fileDir = path.dirname(filePath);
+
+    if (!fs.existsSync(fileDir)) fs.mkdirSync(fileDir, { recursive: true });
+
+    fs.writeFileSync(filePath, encodedExportData);
+  } else {
+    log(exportData);
+  }
+};
+
+main().catch(err => {
+  log(err);
+}).finally(() => {
+  process.exit(0);
+});
@@ -0,0 +1,118 @@
+//
+// Copyright 2021 Vulcanize, Inc.
+//
+
+import assert from 'assert';
+import 'reflect-metadata';
+import yargs from 'yargs';
+import { hideBin } from 'yargs/helpers';
+import debug from 'debug';
+import { PubSub } from 'apollo-server-express';
+import fs from 'fs';
+import path from 'path';
+
+import { getConfig, fillBlocks, JobQueue, DEFAULT_CONFIG_PATH, Config, initClients } from '@vulcanize/util';
+import { GraphWatcher, Database as GraphDatabase } from '@vulcanize/graph-node';
+import * as codec from '@ipld/dag-cbor';
+
+import { Database } from '../database';
+import { Indexer } from '../indexer';
+import { EventWatcher } from '../events';
+import { IPLDBlock } from '../entity/IPLDBlock';
+
+const log = debug('vulcanize:import-state');
+
+export const main = async (): Promise<any> => {
+  const argv = await yargs(hideBin(process.argv)).parserConfiguration({
+    'parse-numbers': false
+  }).options({
+    configFile: {
+      alias: 'f',
+      type: 'string',
+      demandOption: true,
+      describe: 'configuration file path (toml)',
+      default: DEFAULT_CONFIG_PATH
+    },
+    importFile: {
+      alias: 'i',
+      type: 'string',
+      demandOption: true,
+      describe: 'Import file path (JSON)'
+    }
+  }).argv;
+
+  const config: Config = await getConfig(argv.configFile);
+  const { ethClient, postgraphileClient, ethProvider } = await initClients(config);
+
+  const db = new Database(config.database);
+  await db.init();
+
+  const graphDb = new GraphDatabase(config.database, path.resolve(__dirname, 'entity/*'));
+  await graphDb.init();
+
+  const graphWatcher = new GraphWatcher(graphDb, postgraphileClient, config.server.subgraphPath);
+
+  // 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
+  const pubsub = new PubSub();
+  const indexer = new Indexer(config.server, db, ethClient, postgraphileClient, ethProvider, graphWatcher);
+
+  graphWatcher.setIndexer(indexer);
+  await graphWatcher.init();
+
+  const jobQueueConfig = config.jobQueue;
+  assert(jobQueueConfig, 'Missing job queue config');
+
+  const { dbConnectionString, maxCompletionLagInSecs } = jobQueueConfig;
+  assert(dbConnectionString, 'Missing job queue db connection string');
+
+  const jobQueue = new JobQueue({ dbConnectionString, maxCompletionLag: maxCompletionLagInSecs });
+  await jobQueue.start();
+
+  const eventWatcher = new EventWatcher(config.upstream, ethClient, postgraphileClient, indexer, pubsub, jobQueue);
+
+  // Import data.
+  const importFilePath = path.resolve(argv.importFile);
+  const encodedImportData = fs.readFileSync(importFilePath);
+  const importData = codec.decode(Buffer.from(encodedImportData)) as any;
+
+  // Fill the snapshot block.
+  await fillBlocks(
+    jobQueue,
+    indexer,
+    postgraphileClient,
+    eventWatcher,
+    config.upstream.ethServer.blockDelayInMilliSecs,
+    {
+      startBlock: importData.snapshotBlock.blockNumber,
+      endBlock: importData.snapshotBlock.blockNumber
+    }
+  );
+
+  // Fill the Contracts.
+  for (const contract of importData.contracts) {
+    await db.saveContract(contract.address, contract.kind, contract.checkpoint, contract.startingBlock);
+  }
+
+  // Get the snapshot block.
+  const block = await indexer.getBlockProgress(importData.snapshotBlock.blockHash);
+  assert(block);
+
+  // Fill the IPLDBlocks.
+  for (const checkpoint of importData.ipldCheckpoints) {
+    let ipldBlock = new IPLDBlock();
+
+    ipldBlock = Object.assign(ipldBlock, checkpoint);
+    ipldBlock.block = block;
+
+    ipldBlock.data = Buffer.from(codec.encode(ipldBlock.data));
+
+    await db.saveOrUpdateIPLDBlock(ipldBlock);
+  }
+};
+
+main().catch(err => {
+  log(err);
+}).finally(() => {
+  process.exit(0);
+});
@@ -0,0 +1,68 @@
+//
+// Copyright 2021 Vulcanize, Inc.
+//
+
+import path from 'path';
+import assert from 'assert';
+import yargs from 'yargs';
+import 'reflect-metadata';
+import debug from 'debug';
+import util from 'util';
+
+import { Config, DEFAULT_CONFIG_PATH, getConfig, initClients } from '@vulcanize/util';
+import { GraphWatcher, Database as GraphDatabase } from '@vulcanize/graph-node';
+
+import { Database } from '../database';
+import { Indexer } from '../indexer';
+
+const log = debug('vulcanize:inspect-cid');
+
+const main = async (): Promise<void> => {
+  const argv = await yargs.parserConfiguration({
+    'parse-numbers': false
+  }).options({
+    configFile: {
+      alias: 'f',
+      type: 'string',
+      require: true,
+      demandOption: true,
+      describe: 'Configuration file path (toml)',
+      default: DEFAULT_CONFIG_PATH
+    },
+    cid: {
+      alias: 'c',
+      type: 'string',
+      demandOption: true,
+      describe: 'CID to be inspected'
+    }
+  }).argv;
+
+  const config: Config = await getConfig(argv.configFile);
+  const { ethClient, postgraphileClient, ethProvider } = await initClients(config);
+
+  const db = new Database(config.database);
+  await db.init();
+
+  const graphDb = new GraphDatabase(config.database, path.resolve(__dirname, 'entity/*'));
+  await graphDb.init();
+
+  const graphWatcher = new GraphWatcher(graphDb, postgraphileClient, config.server.subgraphPath);
+
+  const indexer = new Indexer(config.server, db, ethClient, postgraphileClient, ethProvider, graphWatcher);
+
+  graphWatcher.setIndexer(indexer);
+  await graphWatcher.init();
+
+  const ipldBlock = await indexer.getIPLDBlockByCid(argv.cid);
+  assert(ipldBlock, 'IPLDBlock for the provided CID doesn\'t exist.');
+
+  const ipldData = await indexer.getIPLDData(ipldBlock);
+
+  log(util.inspect(ipldData, false, null));
+};
+
+main().catch(err => {
+  log(err);
+}).finally(() => {
+  process.exit(0);
+});
@@ -0,0 +1,22 @@
+//
+// Copyright 2021 Vulcanize, Inc.
+//
+
+import debug from 'debug';
+
+import { getConfig, resetJobs } from '@vulcanize/util';
+
+const log = debug('vulcanize:reset-job-queue');
+
+export const command = 'job-queue';
+
+export const desc = 'Reset job queue';
+
+export const builder = {};
+
+export const handler = async (argv: any): Promise<void> => {
+  const config = await getConfig(argv.configFile);
+  await resetJobs(config);
+
+  log('Job queue reset successfully');
+};
@@ -0,0 +1,94 @@
+//
+// Copyright 2021 Vulcanize, Inc.
+//
+
+import path from 'path';
+import debug from 'debug';
+import { MoreThan } from 'typeorm';
+import assert from 'assert';
+
+import { getConfig, initClients, resetJobs } from '@vulcanize/util';
+import { GraphWatcher, Database as GraphDatabase } from '@vulcanize/graph-node';
+
+import { Database } from '../../database';
+import { Indexer } from '../../indexer';
+import { BlockProgress } from '../../entity/BlockProgress';
+
+import { GetMethod } from '../../entity/GetMethod';
+import { _Test } from '../../entity/_Test';
+
+const log = debug('vulcanize:reset-state');
+
+export const command = 'state';
+
+export const desc = 'Reset state to block number';
+
+export const builder = {
+  blockNumber: {
+    type: 'number'
+  }
+};
+
+export const handler = async (argv: any): Promise<void> => {
+  const config = await getConfig(argv.configFile);
+  await resetJobs(config);
+  const { ethClient, postgraphileClient, ethProvider } = await initClients(config);
+
+  // Initialize database.
+  const db = new Database(config.database);
+  await db.init();
+
+  const graphDb = new GraphDatabase(config.database, path.resolve(__dirname, 'entity/*'));
+  await graphDb.init();
+
+  const graphWatcher = new GraphWatcher(graphDb, postgraphileClient, config.server.subgraphPath);
+
+  const indexer = new Indexer(config.server, db, ethClient, postgraphileClient, ethProvider, graphWatcher);
+
+  graphWatcher.setIndexer(indexer);
+  await graphWatcher.init();
+
+  const syncStatus = await indexer.getSyncStatus();
+  assert(syncStatus, 'Missing syncStatus');
+
+  const hooksStatus = await indexer.getHookStatus();
+  assert(hooksStatus, 'Missing hooksStatus');
+
+  const blockProgresses = await indexer.getBlocksAtHeight(argv.blockNumber, false);
+  assert(blockProgresses.length, `No blocks at specified block number ${argv.blockNumber}`);
+  assert(!blockProgresses.some(block => !block.isComplete), `Incomplete block at block number ${argv.blockNumber} with unprocessed events`);
+  const [blockProgress] = blockProgresses;
+
+  const dbTx = await db.createTransactionRunner();
+
+  try {
+    const entities = [BlockProgress, GetMethod, _Test];
+
+    const removeEntitiesPromise = entities.map(async entityClass => {
+      return db.removeEntities<any>(dbTx, entityClass, { blockNumber: MoreThan(argv.blockNumber) });
+    });
+
+    await Promise.all(removeEntitiesPromise);
+
+    if (syncStatus.latestIndexedBlockNumber > blockProgress.blockNumber) {
+      await indexer.updateSyncStatusIndexedBlock(blockProgress.blockHash, blockProgress.blockNumber, true);
+    }
+
+    if (syncStatus.latestCanonicalBlockNumber > blockProgress.blockNumber) {
+      await indexer.updateSyncStatusCanonicalBlock(blockProgress.blockHash, blockProgress.blockNumber, true);
+    }
+
+    if (hooksStatus.latestProcessedBlockNumber > blockProgress.blockNumber) {
+      await indexer.updateHookStatusProcessedBlock(blockProgress.blockNumber, true);
+    }
+
+    dbTx.commitTransaction();
+  } catch (error) {
+    await dbTx.rollbackTransaction();
+    throw error;
+  } finally {
+    await dbTx.release();
+  }
+
+  log('Reset state successfully');
+};
@@ -0,0 +1,24 @@
+//
+// Copyright 2021 Vulcanize, Inc.
+//
+
+import 'reflect-metadata';
+import debug from 'debug';
+
+import { getResetYargs } from '@vulcanize/util';
+
+const log = debug('vulcanize:reset');
+
+const main = async () => {
+  return getResetYargs()
+    .commandDir('reset-cmds', { extensions: ['ts', 'js'], exclude: /([a-zA-Z0-9\s_\\.\-:])+(.d.ts)$/ })
+    .demandCommand(1)
+    .help()
+    .argv;
+};
+
+main().then(() => {
+  process.exit();
+}).catch(err => {
+  log(err);
+});
@@ -2,15 +2,20 @@
 // Copyright 2021 Vulcanize, Inc.
 //

-import assert from 'assert';
+import path from 'path';
 import yargs from 'yargs';
 import 'reflect-metadata';
+import debug from 'debug';

-import { Config, DEFAULT_CONFIG_PATH, getConfig } from '@vulcanize/util';
+import { Config, DEFAULT_CONFIG_PATH, getConfig, initClients } from '@vulcanize/util';
+import { GraphWatcher, Database as GraphDatabase } from '@vulcanize/graph-node';

 import { Database } from '../database';
+import { Indexer } from '../indexer';

-(async () => {
+const log = debug('vulcanize:watch-contract');
+
+const main = async (): Promise<void> => {
  const argv = await yargs.parserConfiguration({
    'parse-numbers': false
  }).options({
@@ -19,7 +24,7 @@ import { Database } from '../database';
      type: 'string',
      require: true,
      demandOption: true,
-      describe: 'configuration file path (toml)',
+      describe: 'Configuration file path (toml)',
      default: DEFAULT_CONFIG_PATH
    },
    address: {
@@ -34,22 +39,41 @@ import { Database } from '../database';
      demandOption: true,
      describe: 'Kind of contract'
    },
+    checkpoint: {
+      type: 'boolean',
+      require: true,
+      demandOption: true,
+      describe: 'Turn checkpointing on'
+    },
    startingBlock: {
      type: 'number',
-      default: 1,
      describe: 'Starting block'
    }
  }).argv;

  const config: Config = await getConfig(argv.configFile);
-  const { database: dbConfig } = config;
+  const { ethClient, postgraphileClient, ethProvider } = await initClients(config);

-  assert(dbConfig);
-
-  const db = new Database(dbConfig);
+  const db = new Database(config.database);
  await db.init();

-  await db.saveContract(argv.address, argv.kind, argv.startingBlock);
+  const graphDb = new GraphDatabase(config.database, path.resolve(__dirname, 'entity/*'));
+  await graphDb.init();
+
+  const graphWatcher = new GraphWatcher(graphDb, postgraphileClient, config.server.subgraphPath);
+
+  const indexer = new Indexer(config.server, db, ethClient, postgraphileClient, ethProvider, graphWatcher);
+
+  graphWatcher.setIndexer(indexer);
+  await graphWatcher.init();
+
+  await indexer.watchContract(argv.address, argv.kind, argv.checkpoint, argv.startingBlock);

  await db.close();
-})();
+};
+
+main().catch(err => {
+  log(err);
+}).finally(() => {
+  process.exit(0);
+});
@@ -3,7 +3,6 @@
 //

 import { gql } from '@apollo/client/core';
-
 import { GraphQLClient, GraphQLConfig } from '@vulcanize/ipld-eth-client';

 import { queries, mutations, subscriptions } from './gql';
@@ -18,7 +17,6 @@ export class Client {
    this._client = new GraphQLClient(config);
  }

-  // eslint-disable-next-line camelcase
  async getGetMethod (blockHash: string, contractAddress: string): Promise<any> {
    const { getMethod } = await this._client.query(
      gql(queries.getMethod),
@@ -28,8 +26,7 @@ export class Client {
    return getMethod;
  }

-  // eslint-disable-next-line camelcase
-  async get_test (blockHash: string, contractAddress: string): Promise<any> {
+  async _getTest (blockHash: string, contractAddress: string): Promise<any> {
    const { _test } = await this._client.query(
      gql(queries._test),
      { blockHash, contractAddress }
@@ -3,20 +3,22 @@
 //

 import assert from 'assert';
-import { Connection, ConnectionOptions, DeepPartial, FindConditions, QueryRunner, FindManyOptions } from 'typeorm';
+import { Connection, ConnectionOptions, DeepPartial, FindConditions, QueryRunner, FindManyOptions, MoreThan } from 'typeorm';
 import path from 'path';

-import { Database as BaseDatabase } from '@vulcanize/util';
+import { Database as BaseDatabase, MAX_REORG_DEPTH, DatabaseInterface } from '@vulcanize/util';

 import { Contract } from './entity/Contract';
 import { Event } from './entity/Event';
 import { SyncStatus } from './entity/SyncStatus';
+import { HookStatus } from './entity/HookStatus';
 import { BlockProgress } from './entity/BlockProgress';
+import { IPLDBlock } from './entity/IPLDBlock';

 import { GetMethod } from './entity/GetMethod';
 import { _Test } from './entity/_Test';

-export class Database {
+export class Database implements DatabaseInterface {
  _config: ConnectionOptions;
  _conn!: Connection;
  _baseDatabase: BaseDatabase;
@@ -43,7 +45,6 @@ export class Database {
    return this._baseDatabase.close();
  }

-  // eslint-disable-next-line camelcase
  async getGetMethod ({ blockHash, contractAddress }: { blockHash: string, contractAddress: string }): Promise<GetMethod | undefined> {
    return this._conn.getRepository(GetMethod)
      .findOne({
@@ -52,8 +53,7 @@ export class Database {
      });
  }

-  // eslint-disable-next-line camelcase
-  async get_test ({ blockHash, contractAddress }: { blockHash: string, contractAddress: string }): Promise<_Test | undefined> {
+  async _getTest ({ blockHash, contractAddress }: { blockHash: string, contractAddress: string }): Promise<_Test | undefined> {
    return this._conn.getRepository(_Test)
      .findOne({
        blockHash,
@@ -61,20 +61,175 @@ export class Database {
      });
  }

-  // eslint-disable-next-line camelcase
-  async saveGetMethod ({ blockHash, contractAddress, value, proof }: DeepPartial<GetMethod>): Promise<GetMethod> {
+  async saveGetMethod ({ blockHash, blockNumber, contractAddress, value, proof }: DeepPartial<GetMethod>): Promise<GetMethod> {
    const repo = this._conn.getRepository(GetMethod);
-    const entity = repo.create({ blockHash, contractAddress, value, proof });
+    const entity = repo.create({ blockHash, blockNumber, contractAddress, value, proof });
    return repo.save(entity);
  }

-  // eslint-disable-next-line camelcase
-  async save_test ({ blockHash, contractAddress, value, proof }: DeepPartial<_Test>): Promise<_Test> {
+  async _saveTest ({ blockHash, blockNumber, contractAddress, value, proof }: DeepPartial<_Test>): Promise<_Test> {
    const repo = this._conn.getRepository(_Test);
-    const entity = repo.create({ blockHash, contractAddress, value, proof });
+    const entity = repo.create({ blockHash, blockNumber, contractAddress, value, proof });
    return repo.save(entity);
  }

+  async getIPLDBlocks (where: FindConditions<IPLDBlock>): Promise<IPLDBlock[]> {
+    const repo = this._conn.getRepository(IPLDBlock);
+    return repo.find({ where, relations: ['block'] });
+  }
+
+  async getLatestIPLDBlock (contractAddress: string, kind: string | 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[]> {
+    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> {
+    const repo = this._conn.getRepository(IPLDBlock);
+    return repo.save(ipldBlock);
+  }
+
+  async getHookStatus (queryRunner: QueryRunner): Promise<HookStatus | undefined> {
+    const repo = queryRunner.manager.getRepository(HookStatus);
+
+    return repo.findOne();
+  }
+
+  async updateHookStatusProcessedBlock (queryRunner: QueryRunner, blockNumber: number, force?: boolean): Promise<HookStatus> {
+    const repo = queryRunner.manager.getRepository(HookStatus);
+    let entity = await repo.findOne();
+
+    if (!entity) {
+      entity = repo.create({
+        latestProcessedBlockNumber: blockNumber
+      });
+    }
+
+    if (force || blockNumber > entity.latestProcessedBlockNumber) {
+      entity.latestProcessedBlockNumber = blockNumber;
+    }
+
+    return repo.save(entity);
+  }
+
+  async getContracts (where: FindConditions<Contract>): Promise<Contract[]> {
+    const repo = this._conn.getRepository(Contract);
+    return repo.find({ where });
+  }
+
  async getContract (address: string): Promise<Contract | undefined> {
    const repo = this._conn.getRepository(Contract);

@@ -115,24 +270,24 @@ export class Database {
    return this._baseDatabase.saveEvents(blockRepo, eventRepo, block, events);
  }

-  async saveContract (address: string, kind: string, startingBlock: number): Promise<void> {
+  async saveContract (address: string, kind: string, checkpoint: boolean, startingBlock: number): Promise<void> {
    await this._conn.transaction(async (tx) => {
      const repo = tx.getRepository(Contract);

-      return this._baseDatabase.saveContract(repo, address, startingBlock, kind);
+      return this._baseDatabase.saveContract(repo, address, kind, checkpoint, startingBlock);
    });
  }

-  async updateSyncStatusIndexedBlock (queryRunner: QueryRunner, blockHash: string, blockNumber: number): Promise<SyncStatus> {
+  async updateSyncStatusIndexedBlock (queryRunner: QueryRunner, blockHash: string, blockNumber: number, force = false): Promise<SyncStatus> {
    const repo = queryRunner.manager.getRepository(SyncStatus);

-    return this._baseDatabase.updateSyncStatusIndexedBlock(repo, blockHash, blockNumber);
+    return this._baseDatabase.updateSyncStatusIndexedBlock(repo, blockHash, blockNumber, force);
  }

-  async updateSyncStatusCanonicalBlock (queryRunner: QueryRunner, blockHash: string, blockNumber: number): Promise<SyncStatus> {
+  async updateSyncStatusCanonicalBlock (queryRunner: QueryRunner, blockHash: string, blockNumber: number, force = false): Promise<SyncStatus> {
    const repo = queryRunner.manager.getRepository(SyncStatus);

-    return this._baseDatabase.updateSyncStatusCanonicalBlock(repo, blockHash, blockNumber);
+    return this._baseDatabase.updateSyncStatusCanonicalBlock(repo, blockHash, blockNumber, force);
  }

  async updateSyncStatusChainHead (queryRunner: QueryRunner, blockHash: string, blockNumber: number): Promise<SyncStatus> {
@@ -13,6 +13,9 @@ export class BlockProgress implements BlockProgressInterface {
  @PrimaryGeneratedColumn()
  id!: number;

+  @Column('varchar')
+  cid!: string;
+
  @Column('varchar', { length: 66 })
  blockHash!: string;

@@ -13,9 +13,12 @@ export class Contract {
  @Column('varchar', { length: 42 })
  address!: string;

-  @Column('varchar', { length: 8 })
+  @Column('varchar')
  kind!: string;

+  @Column('boolean')
+  checkpoint!: boolean;
+
  @Column('integer')
  startingBlock!: number;
 }
@@ -12,7 +12,7 @@ export class Event {
  @PrimaryGeneratedColumn()
  id!: number;

-  @ManyToOne(() => BlockProgress)
+  @ManyToOne(() => BlockProgress, { onDelete: 'CASCADE' })
  block!: BlockProgress;

  @Column('varchar', { length: 66 })
@@ -3,25 +3,25 @@
 //

 import { Entity, PrimaryColumn, Column } from 'typeorm';
+import { bigintTransformer } from '@vulcanize/util';

@Entity()
 export class ExampleEntity {
  @PrimaryColumn('varchar')
  id!: string;

-  // https://typeorm.io/#/entities/primary-columns
  @PrimaryColumn('varchar', { length: 66 })
-  blockHash!: string
+  blockHash!: string;

  @Column('integer')
  blockNumber!: number;

-  @Column('bigint')
-  count!: bigint
+  @Column('bigint', { transformer: bigintTransformer })
+  count!: bigint;

  @Column('varchar')
-  param1!: string
+  param1!: string;

  @Column('integer')
-  param2!: number
+  param2!: number;
 }
@@ -13,6 +13,9 @@ export class GetMethod {
  @Column('varchar', { length: 66 })
  blockHash!: string;

+  @Column('integer')
+  blockNumber!: number;
+
  @Column('varchar', { length: 42 })
  contractAddress!: string;

@@ -0,0 +1,14 @@
+//
+// Copyright 2021 Vulcanize, Inc.
+//
+
+import { Entity, PrimaryGeneratedColumn, Column } from 'typeorm';
+
+@Entity()
+export class HookStatus {
+  @PrimaryGeneratedColumn()
+  id!: number;
+
+  @Column('integer')
+  latestProcessedBlockNumber!: number;
+}
@@ -0,0 +1,30 @@
+//
+// Copyright 2021 Vulcanize, Inc.
+//
+
+import { Entity, PrimaryGeneratedColumn, Column, Index, ManyToOne } from 'typeorm';
+import { BlockProgress } from './BlockProgress';
+
+@Entity()
+@Index(['cid'], { unique: true })
+@Index(['block', 'contractAddress'])
+@Index(['block', 'contractAddress', 'kind'], { unique: true })
+export class IPLDBlock {
+  @PrimaryGeneratedColumn()
+  id!: number;
+
+  @ManyToOne(() => BlockProgress, { onDelete: 'CASCADE' })
+  block!: BlockProgress;
+
+  @Column('varchar', { length: 42 })
+  contractAddress!: string;
+
+  @Column('varchar')
+  cid!: string;
+
+  @Column('varchar')
+  kind!: string;
+
+  @Column('bytea')
+  data!: Buffer;
+}
@@ -14,10 +14,13 @@ export class _Test {
  @Column('varchar', { length: 66 })
  blockHash!: string;

+  @Column('integer')
+  blockNumber!: number;
+
  @Column('varchar', { length: 42 })
  contractAddress!: string;

-  @Column('numeric', { transformer: bigintTransformer })
+  @Column('bigint', { transformer: bigintTransformer })
  value!: bigint;

  @Column('text', { nullable: true })
@@ -10,10 +10,15 @@ import { EthClient } from '@vulcanize/ipld-eth-client';
 import {
  JobQueue,
  EventWatcher as BaseEventWatcher,
+  EventWatcherInterface,
  QUEUE_BLOCK_PROCESSING,
  QUEUE_EVENT_PROCESSING,
+  QUEUE_BLOCK_CHECKPOINT,
+  QUEUE_HOOKS,
+  QUEUE_IPFS,
  UNKNOWN_EVENT_NAME,
-  UpstreamConfig
+  UpstreamConfig,
+  JOB_KIND_PRUNE
 } from '@vulcanize/util';

 import { Indexer } from './indexer';
@@ -23,7 +28,7 @@ const EVENT = 'event';

 const log = debug('vulcanize:events');

-export class EventWatcher {
+export class EventWatcher implements EventWatcherInterface {
  _ethClient: EthClient
  _indexer: Indexer
  _subscription: ZenObservable.Subscription | undefined
@@ -55,6 +60,8 @@ export class EventWatcher {

    await this.initBlockProcessingOnCompleteHandler();
    await this.initEventProcessingOnCompleteHandler();
+    await this.initHooksOnCompleteHandler();
+    await this.initBlockCheckpointOnCompleteHandler();
    this._baseEventWatcher.startBlockProcessing();
  }

@@ -64,7 +71,7 @@ export class EventWatcher {

  async initBlockProcessingOnCompleteHandler (): Promise<void> {
    this._jobQueue.onComplete(QUEUE_BLOCK_PROCESSING, async (job) => {
-      const { id, data: { failed } } = job;
+      const { id, data: { failed, request: { data: { kind } } } } = job;

      if (failed) {
        log(`Job ${id} for queue ${QUEUE_BLOCK_PROCESSING} failed`);
@@ -72,6 +79,8 @@ export class EventWatcher {
      }

      await this._baseEventWatcher.blockProcessingCompleteHandler(job);
+
+      await this.createHooksJob(kind);
    });
  }

@@ -99,6 +108,27 @@ export class EventWatcher {
    });
  }

+  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);
@@ -111,4 +141,40 @@ export class EventWatcher {
      });
    }
  }
+
+  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 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 });
+    }
+  }
 }
@@ -2,17 +2,15 @@
 // Copyright 2021 Vulcanize, Inc.
 //

+import path from 'path';
 import assert from 'assert';
 import 'reflect-metadata';
 import yargs from 'yargs';
 import { hideBin } from 'yargs/helpers';
 import debug from 'debug';
 import { PubSub } from 'apollo-server-express';
-import path from 'path';

-import { getCache } from '@vulcanize/cache';
-import { EthClient } from '@vulcanize/ipld-eth-client';
-import { getConfig, fillBlocks, JobQueue, DEFAULT_CONFIG_PATH, getCustomProvider } from '@vulcanize/util';
+import { Config, getConfig, fillBlocks, JobQueue, DEFAULT_CONFIG_PATH, initClients } from '@vulcanize/util';
 import { GraphWatcher, Database as GraphDatabase } from '@vulcanize/graph-node';

 import { Database } from './database';
@@ -44,46 +42,27 @@ export const main = async (): Promise<any> => {
    }
  }).argv;

-  const config = await getConfig(argv.configFile);
+  const config: Config = await getConfig(argv.configFile);
+  const { ethClient, postgraphileClient, ethProvider } = await initClients(config);

-  assert(config.server, 'Missing server config');
-
-  const { upstream, database: dbConfig, jobQueue: jobQueueConfig, server: { subgraphPath } } = config;
-
-  assert(dbConfig, 'Missing database config');
-
-  const db = new Database(dbConfig);
+  const db = new Database(config.database);
  await db.init();

-  assert(upstream, 'Missing upstream config');
-  const { ethServer: { gqlApiEndpoint, gqlPostgraphileEndpoint, rpcProviderEndpoint, blockDelayInMilliSecs }, cache: cacheConfig } = upstream;
-  assert(gqlPostgraphileEndpoint, 'Missing upstream ethServer.gqlPostgraphileEndpoint');
-
-  const cache = await getCache(cacheConfig);
-
-  const ethClient = new EthClient({
-    gqlEndpoint: gqlApiEndpoint,
-    gqlSubscriptionEndpoint: gqlPostgraphileEndpoint,
-    cache
-  });
-
-  const postgraphileClient = new EthClient({
-    gqlEndpoint: gqlPostgraphileEndpoint,
-    cache
-  });
-
-  const ethProvider = getCustomProvider(rpcProviderEndpoint);
-
-  const graphDb = new GraphDatabase(dbConfig, path.resolve(__dirname, 'entity/*'));
+  const graphDb = new GraphDatabase(config.database, path.resolve(__dirname, 'entity/*'));
  await graphDb.init();

-  const graphWatcher = new GraphWatcher(graphDb, postgraphileClient, subgraphPath);
-  await graphWatcher.init();
+  const graphWatcher = new GraphWatcher(graphDb, postgraphileClient, config.server.subgraphPath);

  // 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
  const pubsub = new PubSub();
-  const indexer = new Indexer(db, ethClient, postgraphileClient, ethProvider, graphWatcher);
+  const indexer = new Indexer(config.server, db, ethClient, postgraphileClient, ethProvider, graphWatcher);
+
+  graphWatcher.setIndexer(indexer);
+  await graphWatcher.init();
+
+  const jobQueueConfig = config.jobQueue;
+  assert(jobQueueConfig, 'Missing job queue config');

  const { dbConnectionString, maxCompletionLagInSecs } = jobQueueConfig;
  assert(dbConnectionString, 'Missing job queue db connection string');
@@ -91,11 +70,9 @@ export const main = async (): Promise<any> => {
  const jobQueue = new JobQueue({ dbConnectionString, maxCompletionLag: maxCompletionLagInSecs });
  await jobQueue.start();

-  const eventWatcher = new EventWatcher(upstream, ethClient, postgraphileClient, indexer, pubsub, jobQueue);
+  const eventWatcher = new EventWatcher(config.upstream, ethClient, postgraphileClient, indexer, pubsub, jobQueue);

-  assert(jobQueueConfig, 'Missing job queue config');
-
-  await fillBlocks(jobQueue, indexer, postgraphileClient, eventWatcher, blockDelayInMilliSecs, argv);
+  await fillBlocks(jobQueue, indexer, postgraphileClient, eventWatcher, config.upstream.ethServer.blockDelayInMilliSecs, argv);
 };

 main().catch(err => {
@@ -1,3 +1,3 @@
-mutation watchContract($contractAddress: String!, $startingBlock: Int){
-    watchContract(contractAddress: $contractAddress, startingBlock: $startingBlock)
+mutation watchContract($address: String!, $kind: String!, $checkpoint: Boolean!, $startingBlock: Int){
+    watchContract(address: $address, kind: $kind, checkpoint: $checkpoint, startingBlock: $startingBlock)
 }
@@ -1,6 +1,7 @@
 query events($blockHash: String!, $contractAddress: String!, $name: String){
    events(blockHash: $blockHash, contractAddress: $contractAddress, name: $name){
        block{
+            cid
            hash
            number
            timestamp
@@ -1,6 +1,7 @@
 query eventsInRange($fromBlockNumber: Int!, $toBlockNumber: Int!){
    eventsInRange(fromBlockNumber: $fromBlockNumber, toBlockNumber: $toBlockNumber){
        block{
+            cid
            hash
            number
            timestamp
@@ -0,0 +1,8 @@
+query exampleEntity($id: String!, $blockHash: String!){
+    exampleEntity(id: $id, blockHash: $blockHash){
+        id
+        count
+        param1
+        param2
+    }
+}
@@ -0,0 +1,15 @@
+query getState($blockHash: String!, $contractAddress: String!, $kind: String){
+    getState(blockHash: $blockHash, contractAddress: $contractAddress, kind: $kind){
+        block{
+            cid
+            hash
+            number
+            timestamp
+            parentHash
+        }
+        contractAddress
+        cid
+        kind
+        data
+    }
+}
@@ -0,0 +1,15 @@
+query getStateByCID($cid: String!){
+    getStateByCID(cid: $cid){
+        block{
+            cid
+            hash
+            number
+            timestamp
+            parentHash
+        }
+        contractAddress
+        cid
+        kind
+        data
+    }
+}
@@ -5,3 +5,6 @@ export const events = fs.readFileSync(path.join(__dirname, 'events.gql'), 'utf8'
 export const eventsInRange = fs.readFileSync(path.join(__dirname, 'eventsInRange.gql'), 'utf8');
 export const getMethod = fs.readFileSync(path.join(__dirname, 'getMethod.gql'), 'utf8');
 export const _test = fs.readFileSync(path.join(__dirname, '_test.gql'), 'utf8');
+export const exampleEntity = fs.readFileSync(path.join(__dirname, 'exampleEntity.gql'), 'utf8');
+export const getStateByCID = fs.readFileSync(path.join(__dirname, 'getStateByCID.gql'), 'utf8');
+export const getState = fs.readFileSync(path.join(__dirname, 'getState.gql'), 'utf8');
@@ -1,6 +1,7 @@
 subscription onEvent{
    onEvent{
        block{
+            cid
            hash
            number
            timestamp
@@ -1,51 +0,0 @@
-//
-// Copyright 2021 Vulcanize, Inc.
-//
-
-import assert from 'assert';
-
-import { Indexer, ResultEvent } from './indexer';
-
-/**
- * Event hook function.
- * @param indexer Indexer instance that contains methods to fetch and update the contract values in the database.
- * @param eventData ResultEvent object containing necessary information.
- */
-export async function handleEvent (indexer: Indexer, eventData: ResultEvent): Promise<void> {
-  assert(indexer);
-  assert(eventData);
-
-  // The following code is for ERC20 contract implementation.
-
-  // Perform indexing based on the type of event.
-  switch (eventData.event.__typename) {
-    // In case of ERC20 'Transfer' event.
-    case 'TransferEvent': {
-      // On a transfer, balances for both parties change.
-      // Therefore, trigger indexing for both sender and receiver.
-
-      // Get event fields from eventData.
-      // const { from, to } = eventData.event;
-
-      // Update balance entry for sender in the database.
-      // await indexer.balanceOf(eventData.block.hash, eventData.contract, from);
-
-      // Update balance entry for receiver in the database.
-      // await indexer.balanceOf(eventData.block.hash, eventData.contract, to);
-
-      break;
-    }
-    // In case of ERC20 'Approval' event.
-    case 'ApprovalEvent': {
-      // On an approval, allowance for (owner, spender) combination changes.
-
-      // Get event fields from eventData.
-      // const { owner, spender } = eventData.event;
-
-      // Update allowance entry for (owner, spender) combination in the database.
-      // await indexer.allowance(eventData.block.hash, eventData.contract, owner, spender);
-
-      break;
-    }
-  }
-}
@@ -6,14 +6,33 @@ import assert from 'assert';

 import { Indexer, ResultEvent } from './indexer';

-/**
- * Event hook function.
- * @param indexer Indexer instance that contains methods to fetch and update the contract values in the database.
- * @param eventData ResultEvent object containing necessary information.
- */
+export async function createInitialCheckpoint (indexer: Indexer, contractAddress: string, blockHash: string): Promise<void> {
+  assert(indexer);
+  assert(blockHash);
+  assert(contractAddress);
+
+  // Store an empty state in an IPLDBlock.
+  const ipldBlockData: any = {
+    state: {}
+  };
+
+  await indexer.createCheckpoint(contractAddress, blockHash, ipldBlockData);
+}
+
+export async function createStateDiff (indexer: Indexer, blockHash: string): Promise<void> {
+  assert(indexer);
+  assert(blockHash);
+}
+
+export async function createStateCheckpoint (indexer: Indexer, contractAddress: string, blockHash: string): Promise<boolean> {
+  assert(indexer);
+  assert(blockHash);
+  assert(contractAddress);
+
+  return false;
+}
+
 export async function handleEvent (indexer: Indexer, eventData: ResultEvent): Promise<void> {
  assert(indexer);
  assert(eventData);
-
-  // Perform indexing based on the type of event.
 }
@@ -7,21 +7,28 @@ import debug from 'debug';
 import { DeepPartial } from 'typeorm';
 import JSONbig from 'json-bigint';
 import { ethers } from 'ethers';
+import { sha256 } from 'multiformats/hashes/sha2';
+import { CID } from 'multiformats/cid';
+import _ from 'lodash';

 import { JsonFragment } from '@ethersproject/abi';
 import { BaseProvider } from '@ethersproject/providers';
+import * as codec from '@ipld/dag-cbor';
 import { EthClient } from '@vulcanize/ipld-eth-client';
 import { StorageLayout } from '@vulcanize/solidity-mapper';
-import { EventInterface, Indexer as BaseIndexer, ValueResult, UNKNOWN_EVENT_NAME } from '@vulcanize/util';
+import { EventInterface, Indexer as BaseIndexer, IndexerInterface, ValueResult, UNKNOWN_EVENT_NAME, ServerConfig, updateStateForElementaryType } from '@vulcanize/util';
 import { GraphWatcher } from '@vulcanize/graph-node';

 import { Database } from './database';
 import { Contract } from './entity/Contract';
 import { Event } from './entity/Event';
 import { SyncStatus } from './entity/SyncStatus';
+import { HookStatus } from './entity/HookStatus';
 import { BlockProgress } from './entity/BlockProgress';
+import { IPLDBlock } from './entity/IPLDBlock';
 import artifacts from './artifacts/Example.json';
-import { handleEvent } from './hooks';
+import { createInitialCheckpoint, handleEvent, createStateDiff, createStateCheckpoint } from './hooks';
+import { IPFSClient } from './ipfs';

 const log = debug('vulcanize:indexer');

@@ -29,6 +36,7 @@ const TEST_EVENT = 'Test';

 export type ResultEvent = {
  block: {
+    cid: string;
    hash: string;
    number: number;
    timestamp: number;
@@ -48,30 +56,49 @@ export type ResultEvent = {
  event: any;

  proof: string;
-}
+};

-export class Indexer {
+export type ResultIPLDBlock = {
+  block: {
+    cid: string;
+    hash: string;
+    number: number;
+    timestamp: number;
+    parentHash: string;
+  };
+  contractAddress: string;
+  cid: string;
+  kind: string;
+  data: string;
+};
+
+export class Indexer implements IndexerInterface {
  _db: Database
  _ethClient: EthClient
  _ethProvider: BaseProvider
-  _postgraphileClient: EthClient;
+  _postgraphileClient: EthClient
+  _baseIndexer: BaseIndexer
+  _serverConfig: ServerConfig
  _graphWatcher: GraphWatcher;
-  _baseIndexer: BaseIndexer;

  _abi: JsonFragment[]
  _storageLayout: StorageLayout
  _contract: ethers.utils.Interface

-  constructor (db: Database, ethClient: EthClient, postgraphileClient: EthClient, ethProvider: BaseProvider, graphWatcher: GraphWatcher) {
+  _ipfsClient: IPFSClient
+
+  constructor (serverConfig: ServerConfig, db: Database, ethClient: EthClient, postgraphileClient: EthClient, ethProvider: BaseProvider, graphWatcher: GraphWatcher) {
    assert(db);
    assert(ethClient);
+    assert(postgraphileClient);

    this._db = db;
    this._ethClient = ethClient;
    this._postgraphileClient = postgraphileClient;
    this._ethProvider = ethProvider;
+    this._serverConfig = serverConfig;
+    this._baseIndexer = new BaseIndexer(this._db, this._ethClient, this._postgraphileClient, this._ethProvider);
    this._graphWatcher = graphWatcher;
-    this._baseIndexer = new BaseIndexer(this._db, this._ethClient, this._ethProvider);

    const { abi, storageLayout } = artifacts;

@@ -82,6 +109,8 @@ export class Indexer {
    this._storageLayout = storageLayout;

    this._contract = new ethers.utils.Interface(this._abi);
+
+    this._ipfsClient = new IPFSClient(this._serverConfig.ipfsApiAddr);
  }

  getResultEvent (event: Event): ResultEvent {
@@ -91,6 +120,7 @@ export class Indexer {

    return {
      block: {
+        cid: block.cid,
        hash: block.blockHash,
        number: block.blockNumber,
        timestamp: block.blockTimestamp,
@@ -118,6 +148,26 @@ export class Indexer {
    };
  }

+  getResultIPLDBlock (ipldBlock: IPLDBlock): ResultIPLDBlock {
+    const block = ipldBlock.block;
+
+    const data = codec.decode(Buffer.from(ipldBlock.data)) as any;
+
+    return {
+      block: {
+        cid: block.cid,
+        hash: block.blockHash,
+        number: block.blockNumber,
+        timestamp: block.blockTimestamp,
+        parentHash: block.parentHash
+      },
+      contractAddress: ipldBlock.contractAddress,
+      cid: ipldBlock.cid,
+      kind: ipldBlock.kind,
+      data: JSON.stringify(data)
+    };
+  }
+
  async getMethod (blockHash: string, contractAddress: string): Promise<ValueResult> {
    const entity = await this._db.getGetMethod({ blockHash, contractAddress });
    if (entity) {
@@ -131,19 +181,21 @@ export class Indexer {

    log('getMethod: db miss, fetching from upstream server');

-    const contract = new ethers.Contract(contractAddress, this._abi, this._ethProvider);
+    const { block: { number } } = await this._ethClient.getBlockByHash(blockHash);
+    const blockNumber = ethers.BigNumber.from(number).toNumber();

+    const contract = new ethers.Contract(contractAddress, this._abi, this._ethProvider);
    const value = await contract.getMethod({ blockTag: blockHash });

    const result: ValueResult = { value };

-    await this._db.saveGetMethod({ blockHash, contractAddress, value: result.value, proof: JSONbig.stringify(result.proof) });
+    await this._db.saveGetMethod({ blockHash, blockNumber, contractAddress, value: result.value, proof: JSONbig.stringify(result.proof) });

    return result;
  }

-  async _test (blockHash: string, contractAddress: string): Promise<ValueResult> {
-    const entity = await this._db.get_test({ blockHash, contractAddress });
+  async _test (blockHash: string, contractAddress: string, diff = false): Promise<ValueResult> {
+    const entity = await this._db._getTest({ blockHash, contractAddress });
    if (entity) {
      log('_test: db hit.');

@@ -155,6 +207,9 @@ export class Indexer {

    log('_test: db miss, fetching from upstream server');

+    const { block: { number } } = await this._ethClient.getBlockByHash(blockHash);
+    const blockNumber = ethers.BigNumber.from(number).toNumber();
+
    const result = await this._baseIndexer.getStorageValue(
      this._storageLayout,
      blockHash,
@@ -162,13 +217,325 @@ export class Indexer {
      '_test'
    );

-    await this._db.save_test({ blockHash, contractAddress, value: result.value, proof: JSONbig.stringify(result.proof) });
+    await this._db._saveTest({ blockHash, blockNumber, contractAddress, value: result.value, proof: JSONbig.stringify(result.proof) });
+
+    if (diff) {
+      const stateUpdate = updateStateForElementaryType({}, '_test', result.value.toString());
+      await this.createDiffStaged(contractAddress, blockHash, stateUpdate);
+    }

    return result;
  }

-  async getExampleEntity (blockHash: string, id: string): Promise<string> {
-    return this._graphWatcher.getEntity(blockHash, 'ExampleEntity', id);
+  async processCanonicalBlock (job: any): Promise<void> {
+    const { data: { blockHash } } = job;
+
+    // Finalize staged diff blocks if any.
+    await this.finalizeDiffStaged(blockHash);
+
+    // Call custom stateDiff hook.
+    await createStateDiff(this, blockHash);
+  }
+
+  async createDiffStaged (contractAddress: string, blockHash: string, data: any): Promise<void> {
+    const block = await this.getBlockProgress(blockHash);
+    assert(block);
+
+    // Create a staged diff block.
+    const ipldBlock = await this.prepareIPLDBlock(block, contractAddress, data, 'diff_staged');
+    await this.saveOrUpdateIPLDBlock(ipldBlock);
+  }
+
+  async finalizeDiffStaged (blockHash: string): Promise<void> {
+    const block = await this.getBlockProgress(blockHash);
+    assert(block);
+
+    // Get all the staged diff blocks for the given blockHash.
+    const stagedBlocks = await this._db.getIPLDBlocks({ block, kind: 'diff_staged' });
+
+    // For each staged block, create a diff block.
+    for (const stagedBlock of stagedBlocks) {
+      const data = codec.decode(Buffer.from(stagedBlock.data));
+      await this.createDiff(stagedBlock.contractAddress, stagedBlock.block.blockHash, data);
+    }
+
+    // Remove all the staged diff blocks for current blockNumber.
+    await this.removeStagedIPLDBlocks(block.blockNumber);
+  }
+
+  async createDiff (contractAddress: string, blockHash: string, data: any): Promise<void> {
+    const block = await this.getBlockProgress(blockHash);
+    assert(block);
+
+    // Fetch the latest checkpoint for the contract.
+    const checkpoint = await this.getLatestIPLDBlock(contractAddress, 'checkpoint');
+
+    // There should be an initial checkpoint at least.
+    // Assumption: There should be no events for the contract at the starting block.
+    assert(checkpoint, 'Initial checkpoint doesn\'t exist');
+
+    // Check if the latest checkpoint is in the same block.
+    assert(checkpoint.block.blockHash !== block.blockHash, 'Checkpoint already created for the block hash.');
+
+    const ipldBlock = await this.prepareIPLDBlock(block, contractAddress, data, 'diff');
+    await this.saveOrUpdateIPLDBlock(ipldBlock);
+  }
+
+  async processCheckpoint (job: any): Promise<void> {
+    // Return if checkpointInterval is <= 0.
+    const checkpointInterval = this._serverConfig.checkpointInterval;
+    if (checkpointInterval <= 0) return;
+
+    const { data: { blockHash, blockNumber } } = job;
+
+    // Get all the contracts.
+    const contracts = await this._db.getContracts({});
+
+    // For each contract, merge the diff till now to create a checkpoint.
+    for (const contract of contracts) {
+      // Check if contract has checkpointing on.
+      if (contract.checkpoint) {
+        // If a checkpoint doesn't already exist and blockNumber is equal to startingBlock, create an initial checkpoint.
+        const checkpointBlock = await this.getLatestIPLDBlock(contract.address, 'checkpoint');
+
+        if (!checkpointBlock) {
+          if (blockNumber === contract.startingBlock) {
+            // Call initial checkpoint hook.
+            await createInitialCheckpoint(this, contract.address, blockHash);
+          }
+        } else {
+          await this.createCheckpoint(contract.address, blockHash, null, checkpointInterval);
+        }
+      }
+    }
+  }
+
+  async processCLICheckpoint (contractAddress: string, blockHash?: string): Promise<string | undefined> {
+    const checkpointBlockHash = await this.createCheckpoint(contractAddress, blockHash);
+    assert(checkpointBlockHash);
+
+    const block = await this.getBlockProgress(checkpointBlockHash);
+    const checkpointIPLDBlocks = await this._db.getIPLDBlocks({ block, contractAddress, kind: 'checkpoint' });
+
+    // There can be at most one IPLDBlock for a (block, contractAddress, kind) combination.
+    assert(checkpointIPLDBlocks.length <= 1);
+    const checkpointIPLDBlock = checkpointIPLDBlocks[0];
+
+    const checkpointData = this.getIPLDData(checkpointIPLDBlock);
+
+    await this.pushToIPFS(checkpointData);
+
+    return checkpointBlockHash;
+  }
+
+  async createCheckpoint (contractAddress: string, blockHash?: string, data?: any, checkpointInterval?: number): Promise<string | undefined> {
+    const syncStatus = await this.getSyncStatus();
+    assert(syncStatus);
+
+    // Getting the current block.
+    let currentBlock;
+
+    if (blockHash) {
+      currentBlock = await this.getBlockProgress(blockHash);
+    } else {
+      // In case of empty blockHash from checkpoint CLI, get the latest canonical block for the checkpoint.
+      currentBlock = await this.getBlockProgress(syncStatus.latestCanonicalBlockHash);
+    }
+
+    assert(currentBlock);
+
+    // Data is passed in case of initial checkpoint and checkpoint hook.
+    // Assumption: There should be no events for the contract at the starting block.
+    if (data) {
+      const ipldBlock = await this.prepareIPLDBlock(currentBlock, contractAddress, data, 'checkpoint');
+      await this.saveOrUpdateIPLDBlock(ipldBlock);
+
+      return;
+    }
+
+    // If data is not passed, create from previous checkpoint and diffs after that.
+
+    // Make sure the block is marked complete.
+    assert(currentBlock.isComplete, 'Block for a checkpoint should be marked as complete');
+
+    // Make sure the block is in the pruned region.
+    assert(currentBlock.blockNumber <= syncStatus.latestCanonicalBlockNumber, 'Block for a checkpoint should be in the pruned region');
+
+    // Fetch the latest checkpoint for the contract.
+    const checkpointBlock = await this.getLatestIPLDBlock(contractAddress, 'checkpoint', currentBlock.blockNumber);
+    assert(checkpointBlock);
+
+    // Check (only if checkpointInterval is passed) if it is time for a new checkpoint.
+    if (checkpointInterval && checkpointBlock.block.blockNumber > (currentBlock.blockNumber - checkpointInterval)) {
+      return;
+    }
+
+    // Call state checkpoint hook and check if default checkpoint is disabled.
+    const disableDefaultCheckpoint = await createStateCheckpoint(this, contractAddress, currentBlock.blockHash);
+
+    if (disableDefaultCheckpoint) {
+      // Return if default checkpoint is disabled.
+      // Return block hash for checkpoint CLI.
+      return currentBlock.blockHash;
+    }
+
+    const { block: { blockNumber: checkpointBlockNumber } } = checkpointBlock;
+
+    // Fetching all diff blocks after checkpoint.
+    const diffBlocks = await this.getDiffIPLDBlocksByCheckpoint(contractAddress, checkpointBlockNumber);
+
+    const checkpointBlockData = codec.decode(Buffer.from(checkpointBlock.data)) as any;
+    data = {
+      state: checkpointBlockData.state
+    };
+
+    for (const diffBlock of diffBlocks) {
+      const diff = codec.decode(Buffer.from(diffBlock.data)) as any;
+      data.state = _.merge(data.state, diff.state);
+    }
+
+    const ipldBlock = await this.prepareIPLDBlock(currentBlock, contractAddress, data, 'checkpoint');
+    await this.saveOrUpdateIPLDBlock(ipldBlock);
+
+    return currentBlock.blockHash;
+  }
+
+  getIPLDData (ipldBlock: IPLDBlock): any {
+    return codec.decode(Buffer.from(ipldBlock.data));
+  }
+
+  async getIPLDBlocksByHash (blockHash: string): Promise<IPLDBlock[]> {
+    const block = await this.getBlockProgress(blockHash);
+    assert(block);
+
+    return this._db.getIPLDBlocks({ block });
+  }
+
+  async getIPLDBlockByCid (cid: string): Promise<IPLDBlock | undefined> {
+    const ipldBlocks = await this._db.getIPLDBlocks({ cid });
+
+    // There can be only one IPLDBlock with a particular cid.
+    assert(ipldBlocks.length <= 1);
+
+    return ipldBlocks[0];
+  }
+
+  async getLatestIPLDBlock (contractAddress: string, kind: string | null, blockNumber?: number): Promise<IPLDBlock | undefined> {
+    return this._db.getLatestIPLDBlock(contractAddress, kind, blockNumber);
+  }
+
+  async getPrevIPLDBlock (blockHash: string, contractAddress: string, kind?: string): Promise<IPLDBlock | undefined> {
+    const dbTx = await this._db.createTransactionRunner();
+    let res;
+
+    try {
+      res = await this._db.getPrevIPLDBlock(dbTx, blockHash, contractAddress, kind);
+      await dbTx.commitTransaction();
+    } catch (error) {
+      await dbTx.rollbackTransaction();
+      throw error;
+    } finally {
+      await dbTx.release();
+    }
+    return res;
+  }
+
+  async getDiffIPLDBlocksByCheckpoint (contractAddress: string, checkpointBlockNumber: number): Promise<IPLDBlock[]> {
+    return this._db.getDiffIPLDBlocksByCheckpoint(contractAddress, checkpointBlockNumber);
+  }
+
+  async prepareIPLDBlock (block: BlockProgress, contractAddress: string, data: any, kind: string):Promise<any> {
+    assert(_.includes(['diff', 'checkpoint', 'diff_staged'], kind));
+
+    // Get an existing 'diff' | 'diff_staged' | 'checkpoint' IPLDBlock for current block, contractAddress.
+    const currentIPLDBlocks = await this._db.getIPLDBlocks({ block, contractAddress, kind });
+
+    // There can be at most one IPLDBlock for a (block, contractAddress, kind) combination.
+    assert(currentIPLDBlocks.length <= 1);
+    const currentIPLDBlock = currentIPLDBlocks[0];
+
+    // Update currentIPLDBlock if it exists and is of same kind.
+    let ipldBlock;
+    if (currentIPLDBlock) {
+      ipldBlock = currentIPLDBlock;
+
+      // Update the data field.
+      const oldData = codec.decode(Buffer.from(currentIPLDBlock.data));
+      data = _.merge(oldData, data);
+    } else {
+      ipldBlock = new IPLDBlock();
+
+      // Fetch the parent IPLDBlock.
+      const parentIPLDBlock = await this.getLatestIPLDBlock(contractAddress, null, block.blockNumber);
+
+      // Setting the meta-data for an IPLDBlock (done only once per block).
+      data.meta = {
+        id: contractAddress,
+        kind,
+        parent: {
+          '/': parentIPLDBlock ? parentIPLDBlock.cid : null
+        },
+        ethBlock: {
+          cid: {
+            '/': block.cid
+          },
+          num: block.blockNumber
+        }
+      };
+    }
+
+    // Encoding the data using dag-cbor codec.
+    const bytes = codec.encode(data);
+
+    // Calculating sha256 (multi)hash of the encoded data.
+    const hash = await sha256.digest(bytes);
+
+    // Calculating the CID: v1, code: dag-cbor, hash.
+    const cid = CID.create(1, codec.code, hash);
+
+    // Update ipldBlock with new data.
+    ipldBlock = Object.assign(ipldBlock, {
+      block,
+      contractAddress,
+      cid: cid.toString(),
+      kind: data.meta.kind,
+      data: Buffer.from(bytes)
+    });
+
+    return ipldBlock;
+  }
+
+  async saveOrUpdateIPLDBlock (ipldBlock: IPLDBlock): Promise<IPLDBlock> {
+    return this._db.saveOrUpdateIPLDBlock(ipldBlock);
+  }
+
+  async removeStagedIPLDBlocks (blockNumber: number): Promise<void> {
+    const dbTx = await this._db.createTransactionRunner();
+
+    try {
+      await this._db.removeEntities(dbTx, IPLDBlock, { relations: ['block'], where: { block: { blockNumber }, kind: 'diff_staged' } });
+      await dbTx.commitTransaction();
+    } catch (error) {
+      await dbTx.rollbackTransaction();
+      throw error;
+    } finally {
+      await dbTx.release();
+    }
+  }
+
+  async pushToIPFS (data: any): Promise<void> {
+    await this._ipfsClient.push(data);
+  }
+
+  isIPFSConfigured (): boolean {
+    const ipfsAddr = this._serverConfig.ipfsApiAddr;
+
+    // Return false if ipfsAddr is undefined | null | empty string.
+    return (ipfsAddr !== undefined && ipfsAddr !== null && ipfsAddr !== '');
+  }
+
+  async getSubgraphEntity<Entity> (entity: new () => Entity, id: string, blockHash: string): Promise<Entity | undefined> {
+    return this._graphWatcher.getEntity(entity, id, blockHash);
  }

  async triggerIndexingOnEvent (event: Event): Promise<void> {
@@ -181,11 +548,6 @@ export class Indexer {
    await handleEvent(this, resultEvent);
  }

-  async processBlock (blockHash: string): Promise<void> {
-    // Call subgraph handler for block.
-    await this._graphWatcher.handleBlock(blockHash);
-  }
-
  async processEvent (event: Event): Promise<void> {
    // Trigger indexing of data based on the event.
    await this.triggerIndexingOnEvent(event);
@@ -218,14 +580,66 @@ export class Indexer {
    };
  }

-  async watchContract (address: string, startingBlock: number): Promise<boolean> {
-    // Always use the checksum address (https://docs.ethers.io/v5/api/utils/address/#utils-getAddress).
-    await this._db.saveContract(ethers.utils.getAddress(address), 'Example', startingBlock);
+  async watchContract (address: string, kind: string, checkpoint: boolean, startingBlock?: number): Promise<boolean> {
+    // Use the checksum address (https://docs.ethers.io/v5/api/utils/address/#utils-getAddress) if input to address is a contract address.
+    // If a contract identifier is passed as address instead, no need to convert to checksum address.
+    // Customize: use the kind input to filter out non-contract-address input to address.
+    const formattedAddress = (kind === '__protocol__') ? address : ethers.utils.getAddress(address);
+
+    if (!startingBlock) {
+      const syncStatus = await this.getSyncStatus();
+      assert(syncStatus);
+
+      startingBlock = syncStatus.latestIndexedBlockNumber;
+    }
+
+    await this._db.saveContract(formattedAddress, kind, checkpoint, startingBlock);

    return true;
  }

-  async getEventsByFilter (blockHash: string, contract: string, name: string | null): Promise<Array<Event>> {
+  async getHookStatus (): Promise<HookStatus | undefined> {
+    const dbTx = await this._db.createTransactionRunner();
+    let res;
+
+    try {
+      res = await this._db.getHookStatus(dbTx);
+      await dbTx.commitTransaction();
+    } catch (error) {
+      await dbTx.rollbackTransaction();
+      throw error;
+    } finally {
+      await dbTx.release();
+    }
+
+    return res;
+  }
+
+  async updateHookStatusProcessedBlock (blockNumber: number, force?: boolean): Promise<HookStatus> {
+    const dbTx = await this._db.createTransactionRunner();
+    let res;
+
+    try {
+      res = await this._db.updateHookStatusProcessedBlock(dbTx, blockNumber, force);
+      await dbTx.commitTransaction();
+    } catch (error) {
+      await dbTx.rollbackTransaction();
+      throw error;
+    } finally {
+      await dbTx.release();
+    }
+
+    return res;
+  }
+
+  async getLatestCanonicalBlock (): Promise<BlockProgress | undefined> {
+    const syncStatus = await this.getSyncStatus();
+    assert(syncStatus);
+
+    return this.getBlockProgress(syncStatus.latestCanonicalBlockHash);
+  }
+
+  async getEventsByFilter (blockHash: string, contract?: string, name?: string): Promise<Array<Event>> {
    return this._baseIndexer.getEventsByFilter(blockHash, contract, name);
  }

@@ -245,16 +659,16 @@ export class Indexer {
    return this._baseIndexer.getSyncStatus();
  }

-  async updateSyncStatusIndexedBlock (blockHash: string, blockNumber: number): Promise<SyncStatus> {
-    return this._baseIndexer.updateSyncStatusIndexedBlock(blockHash, blockNumber);
+  async updateSyncStatusIndexedBlock (blockHash: string, blockNumber: number, force = false): Promise<SyncStatus> {
+    return this._baseIndexer.updateSyncStatusIndexedBlock(blockHash, blockNumber, force);
  }

  async updateSyncStatusChainHead (blockHash: string, blockNumber: number): Promise<SyncStatus> {
    return this._baseIndexer.updateSyncStatusChainHead(blockHash, blockNumber);
  }

-  async updateSyncStatusCanonicalBlock (blockHash: string, blockNumber: number): Promise<SyncStatus> {
-    return this._baseIndexer.updateSyncStatusCanonicalBlock(blockHash, blockNumber);
+  async updateSyncStatusCanonicalBlock (blockHash: string, blockNumber: number, force = false): Promise<SyncStatus> {
+    return this._baseIndexer.updateSyncStatusCanonicalBlock(blockHash, blockNumber, force);
  }

  async getBlock (blockHash: string): Promise<any> {
@@ -297,7 +711,7 @@ export class Indexer {
    return this._baseIndexer.getAncestorAtDepth(blockHash, depth);
  }

-  async _fetchAndSaveEvents ({ blockHash }: DeepPartial<BlockProgress>): Promise<void> {
+  async _fetchAndSaveEvents ({ cid: blockCid, blockHash }: DeepPartial<BlockProgress>): Promise<void> {
    assert(blockHash);
    let { block, logs } = await this._ethClient.getLogs({ blockHash });

@@ -381,6 +795,7 @@ export class Indexer {

    try {
      block = {
+        cid: blockCid,
        blockHash,
        blockNumber: block.number,
        blockTimestamp: block.timestamp,
@@ -0,0 +1,17 @@
+//
+// Copyright 2021 Vulcanize, Inc.
+//
+
+import { create, IPFSHTTPClient } from 'ipfs-http-client';
+
+export class IPFSClient {
+  _client: IPFSHTTPClient
+
+  constructor (url: string) {
+    this._client = create({ url });
+  }
+
+  async push (data: any): Promise<void> {
+    await this._client.dag.put(data, { format: 'dag-cbor', hashAlg: 'sha2-256' });
+  }
+}
@@ -2,25 +2,26 @@
 // Copyright 2021 Vulcanize, Inc.
 //

+import path from 'path';
 import assert from 'assert';
 import 'reflect-metadata';
 import yargs from 'yargs';
 import { hideBin } from 'yargs/helpers';
 import debug from 'debug';
-import path from 'path';

-import { getCache } from '@vulcanize/cache';
-import { EthClient } from '@vulcanize/ipld-eth-client';
 import {
  getConfig,
+  Config,
  JobQueue,
  JobRunner as BaseJobRunner,
  QUEUE_BLOCK_PROCESSING,
  QUEUE_EVENT_PROCESSING,
+  QUEUE_BLOCK_CHECKPOINT,
+  QUEUE_HOOKS,
+  QUEUE_IPFS,
  JobQueueConfig,
  DEFAULT_CONFIG_PATH,
-  getCustomProvider,
-  JOB_KIND_INDEX
+  initClients
 } from '@vulcanize/util';
 import { GraphWatcher, Database as GraphDatabase } from '@vulcanize/graph-node';

@@ -45,18 +46,17 @@ export class JobRunner {
  async start (): Promise<void> {
    await this.subscribeBlockProcessingQueue();
    await this.subscribeEventProcessingQueue();
+    await this.subscribeBlockCheckpointQueue();
+    await this.subscribeHooksQueue();
+    await this.subscribeIPFSQueue();
  }

  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, blockHash } } = job;
-
-      if (kind === JOB_KIND_INDEX) {
-        await this._indexer.processBlock(blockHash);
-      }
-
      await this._jobQueue.markComplete(job);
    });
  }
@@ -73,6 +73,43 @@ export class JobRunner {
      await this._jobQueue.markComplete(job);
    });
  }
+
+  async subscribeHooksQueue (): Promise<void> {
+    await this._jobQueue.subscribe(QUEUE_HOOKS, async (job) => {
+      const { data: { blockNumber } } = job;
+
+      const hookStatus = await this._indexer.getHookStatus();
+
+      if (hookStatus && hookStatus.latestProcessedBlockNumber < (blockNumber - 1)) {
+        const message = `Hooks for blockNumber ${blockNumber - 1} not processed yet, aborting`;
+        log(message);
+
+        throw new Error(message);
+      }
+
+      await this._indexer.processCanonicalBlock(job);
+
+      await this._jobQueue.markComplete(job);
+    });
+  }
+
+  async subscribeBlockCheckpointQueue (): Promise<void> {
+    await this._jobQueue.subscribe(QUEUE_BLOCK_CHECKPOINT, async (job) => {
+      await this._indexer.processCheckpoint(job);
+
+      await this._jobQueue.markComplete(job);
+    });
+  }
+
+  async subscribeIPFSQueue (): Promise<void> {
+    await this._jobQueue.subscribe(QUEUE_IPFS, async (job) => {
+      const { data: { data } } = job;
+
+      await this._indexer.pushToIPFS(data);
+
+      await this._jobQueue.markComplete(job);
+    });
+  }
 }

 export const main = async (): Promise<any> => {
@@ -86,44 +123,23 @@ export const main = async (): Promise<any> => {
    })
    .argv;

-  const config = await getConfig(argv.f);
+  const config: Config = await getConfig(argv.f);
+  const { ethClient, postgraphileClient, ethProvider } = await initClients(config);

-  assert(config.server, 'Missing server config');
-
-  const { upstream, database: dbConfig, jobQueue: jobQueueConfig, server: { subgraphPath } } = config;
-
-  assert(dbConfig, 'Missing database config');
-
-  const db = new Database(dbConfig);
+  const db = new Database(config.database);
  await db.init();

-  assert(upstream, 'Missing upstream config');
-  const { ethServer: { gqlApiEndpoint, gqlPostgraphileEndpoint, rpcProviderEndpoint }, cache: cacheConfig } = upstream;
-  assert(gqlApiEndpoint, 'Missing upstream ethServer.gqlApiEndpoint');
-  assert(gqlPostgraphileEndpoint, 'Missing upstream ethServer.gqlPostgraphileEndpoint');
-
-  const cache = await getCache(cacheConfig);
-
-  const ethClient = new EthClient({
-    gqlEndpoint: gqlApiEndpoint,
-    gqlSubscriptionEndpoint: gqlPostgraphileEndpoint,
-    cache
-  });
-
-  const postgraphileClient = new EthClient({
-    gqlEndpoint: gqlPostgraphileEndpoint,
-    cache
-  });
-
-  const graphDb = new GraphDatabase(dbConfig, path.resolve(__dirname, 'entity/*'));
+  const graphDb = new GraphDatabase(config.database, path.resolve(__dirname, 'entity/*'));
  await graphDb.init();

-  const graphWatcher = new GraphWatcher(graphDb, postgraphileClient, subgraphPath);
+  const graphWatcher = new GraphWatcher(graphDb, postgraphileClient, config.server.subgraphPath);
+
+  const indexer = new Indexer(config.server, db, ethClient, postgraphileClient, ethProvider, graphWatcher);
+
+  graphWatcher.setIndexer(indexer);
  await graphWatcher.init();

-  const ethProvider = getCustomProvider(rpcProviderEndpoint);
-  const indexer = new Indexer(db, ethClient, postgraphileClient, ethProvider, graphWatcher);
-
+  const jobQueueConfig = config.jobQueue;
  assert(jobQueueConfig, 'Missing job queue config');

  const { dbConnectionString, maxCompletionLagInSecs } = jobQueueConfig;
@@ -11,6 +11,8 @@ import { ValueResult } from '@vulcanize/util';
 import { Indexer } from './indexer';
 import { EventWatcher } from './events';

+import { ExampleEntity } from './entity/ExampleEntity';
+
 const log = debug('vulcanize:resolver');

 export const createResolvers = async (indexer: Indexer, eventWatcher: EventWatcher): Promise<any> => {
@@ -34,9 +36,10 @@ export const createResolvers = async (indexer: Indexer, eventWatcher: EventWatch
    },

    Mutation: {
-      watchContract: (_: any, { contractAddress, startingBlock = 1 }: { contractAddress: string, startingBlock: number }): Promise<boolean> => {
-        log('watchContract', contractAddress, startingBlock);
-        return indexer.watchContract(contractAddress, startingBlock);
+      watchContract: (_: any, { address, kind, checkpoint, startingBlock }: { address: string, kind: string, checkpoint: boolean, startingBlock: number }): Promise<boolean> => {
+        log('watchContract', address, kind, checkpoint, startingBlock);
+
+        return indexer.watchContract(address, kind, checkpoint, startingBlock);
      }
    },

@@ -51,8 +54,14 @@ export const createResolvers = async (indexer: Indexer, eventWatcher: EventWatch
        return indexer._test(blockHash, contractAddress);
      },

-      events: async (_: any, { blockHash, contractAddress, name }: { blockHash: string, contractAddress: string, name: string }) => {
-        log('events', blockHash, contractAddress, name || '');
+      exampleEntity: async (_: any, { id, blockHash }: { id: string, blockHash: string }): Promise<ExampleEntity | undefined> => {
+        log('exampleEntity', id, blockHash);
+
+        return indexer.getSubgraphEntity(ExampleEntity, id, blockHash);
+      },
+
+      events: async (_: any, { blockHash, contractAddress, name }: { blockHash: string, contractAddress: string, name?: string }) => {
+        log('events', blockHash, contractAddress, name);

        const block = await indexer.getBlockProgress(blockHash);
        if (!block || !block.isComplete) {
@@ -75,12 +84,20 @@ export const createResolvers = async (indexer: Indexer, eventWatcher: EventWatch
        return events.map(event => indexer.getResultEvent(event));
      },

-      exampleEntity: async (_: any, { blockHash, id }: { blockHash: string, id: string }) => {
-        log('exampleEntity', blockHash, id);
+      getStateByCID: async (_: any, { cid }: { cid: string }) => {
+        log('getStateByCID', cid);

-        const exampleEntity = await indexer.getExampleEntity(blockHash, id);
+        const ipldBlock = await indexer.getIPLDBlockByCid(cid);

-        return JSON.stringify(exampleEntity, undefined, 2);
+        return ipldBlock && ipldBlock.block.isComplete ? indexer.getResultIPLDBlock(ipldBlock) : undefined;
+      },
+
+      getState: async (_: any, { blockHash, contractAddress, kind = 'diff' }: { blockHash: string, contractAddress: string, kind: string }) => {
+        log('getState', blockHash, contractAddress, kind);
+
+        const ipldBlock = await indexer.getPrevIPLDBlock(blockHash, contractAddress, kind);
+
+        return ipldBlock && ipldBlock.block.isComplete ? indexer.getResultIPLDBlock(ipldBlock) : undefined;
      }
    }
  };
@@ -1,22 +1,31 @@
-type Query {
-  events(blockHash: String!, contractAddress: String!, name: String): [ResultEvent!]
-  eventsInRange(fromBlockNumber: Int!, toBlockNumber: Int!): [ResultEvent!]
-  getMethod(blockHash: String!, contractAddress: String!): ResultString!
-  _test(blockHash: String!, contractAddress: String!): ResultBigInt!
-  exampleEntity(blockHash: String!, id: String!): String!
+scalar BigInt
+
+type Proof {
+  data: String!
 }

-type ResultEvent {
-  block: Block!
-  tx: Transaction!
-  contract: String!
-  eventIndex: Int!
-  eventSignature: String!
-  event: Event!
+type ResultBoolean {
+  value: Boolean!
+  proof: Proof
+}
+
+type ResultString {
+  value: String!
+  proof: Proof
+}
+
+type ResultInt {
+  value: Int!
+  proof: Proof
+}
+
+type ResultBigInt {
+  value: BigInt!
  proof: Proof
 }

 type Block {
+  cid: String!
  hash: String!
  number: Int!
  timestamp: Int!
@@ -30,6 +39,15 @@ type Transaction {
  to: String!
 }

+type ResultEvent {
+  block: Block!
+  tx: Transaction!
+  contract: String!
+  eventIndex: Int!
+  event: Event!
+  proof: Proof
+}
+
 union Event = TestEvent

 type TestEvent {
@@ -37,24 +55,33 @@ type TestEvent {
  param2: Int!
 }

-type Proof {
+type ResultIPLDBlock {
+  block: Block!
+  contractAddress: String!
+  cid: String!
+  kind: String!
  data: String!
 }

-type ResultString {
-  value: String!
-  proof: Proof
+type Query {
+  events(blockHash: String!, contractAddress: String!, name: String): [ResultEvent!]
+  eventsInRange(fromBlockNumber: Int!, toBlockNumber: Int!): [ResultEvent!]
+  getMethod(blockHash: String!, contractAddress: String!): ResultString!
+  _test(blockHash: String!, contractAddress: String!): ResultBigInt!
+  exampleEntity(id: String!, blockHash: String!): ExampleEntity!
+  getStateByCID(cid: String!): ResultIPLDBlock
+  getState(blockHash: String!, contractAddress: String!, kind: String): ResultIPLDBlock
 }

-type ResultBigInt {
-  value: BigInt!
-  proof: Proof
+type ExampleEntity {
+  id: ID!
+  count: BigInt!
+  param1: String!
+  param2: Int!
 }

-scalar BigInt
-
 type Mutation {
-  watchContract(contractAddress: String!, startingBlock: Int): Boolean!
+  watchContract(address: String!, kind: String!, checkpoint: Boolean!, startingBlock: Int): Boolean!
 }

 type Subscription {
@@ -14,9 +14,7 @@ import debug from 'debug';
 import 'graphql-import-node';
 import { createServer } from 'http';

-import { getCache } from '@vulcanize/cache';
-import { EthClient } from '@vulcanize/ipld-eth-client';
-import { DEFAULT_CONFIG_PATH, getConfig, JobQueue, KIND_ACTIVE, getCustomProvider } from '@vulcanize/util';
+import { DEFAULT_CONFIG_PATH, getConfig, Config, JobQueue, KIND_ACTIVE, initClients } from '@vulcanize/util';
 import { GraphWatcher, Database as GraphDatabase } from '@vulcanize/graph-node';

 import { createResolvers } from './resolvers';
@@ -37,58 +35,36 @@ export const main = async (): Promise<any> => {
    })
    .argv;

-  const config = await getConfig(argv.f);
+  const config: Config = await getConfig(argv.f);
+  const { ethClient, postgraphileClient, ethProvider } = await initClients(config);

-  assert(config.server, 'Missing server config');
+  const { host, port, kind: watcherKind } = config.server;

-  const { host, port, kind: watcherKind, subgraphPath } = config.server;
-
-  const { upstream, database: dbConfig, jobQueue: jobQueueConfig } = config;
-
-  assert(dbConfig, 'Missing database config');
-
-  const db = new Database(dbConfig);
+  const db = new Database(config.database);
  await db.init();

-  assert(upstream, 'Missing upstream config');
-  const { ethServer: { gqlApiEndpoint, gqlPostgraphileEndpoint, rpcProviderEndpoint }, cache: cacheConfig } = upstream;
-  assert(gqlApiEndpoint, 'Missing upstream ethServer.gqlApiEndpoint');
-  assert(gqlPostgraphileEndpoint, 'Missing upstream ethServer.gqlPostgraphileEndpoint');
-
-  const cache = await getCache(cacheConfig);
-
-  const ethClient = new EthClient({
-    gqlEndpoint: gqlApiEndpoint,
-    gqlSubscriptionEndpoint: gqlPostgraphileEndpoint,
-    cache
-  });
-
-  const postgraphileClient = new EthClient({
-    gqlEndpoint: gqlPostgraphileEndpoint,
-    cache
-  });
-
-  const ethProvider = getCustomProvider(rpcProviderEndpoint);
-
-  const graphDb = new GraphDatabase(dbConfig, path.resolve(__dirname, 'entity/*'));
+  const graphDb = new GraphDatabase(config.database, path.resolve(__dirname, 'entity/*'));
  await graphDb.init();

-  const graphWatcher = new GraphWatcher(graphDb, postgraphileClient, subgraphPath);
-  await graphWatcher.init();
-
-  const indexer = new Indexer(db, ethClient, postgraphileClient, ethProvider, graphWatcher);
+  const graphWatcher = new GraphWatcher(graphDb, postgraphileClient, config.server.subgraphPath);

  // 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
  const pubsub = new PubSub();
+  const indexer = new Indexer(config.server, db, ethClient, postgraphileClient, ethProvider, graphWatcher);

+  graphWatcher.setIndexer(indexer);
+  await graphWatcher.init();
+
+  const jobQueueConfig = config.jobQueue;
  assert(jobQueueConfig, 'Missing job queue config');
+
  const { dbConnectionString, maxCompletionLagInSecs } = jobQueueConfig;
  assert(dbConnectionString, 'Missing job queue db connection string');

  const jobQueue = new JobQueue({ dbConnectionString, maxCompletionLag: maxCompletionLagInSecs });

-  const eventWatcher = new EventWatcher(upstream, ethClient, postgraphileClient, indexer, pubsub, jobQueue);
+  const eventWatcher = new EventWatcher(config.upstream, ethClient, postgraphileClient, indexer, pubsub, jobQueue);

  if (watcherKind === KIND_ACTIVE) {
    await jobQueue.start();