Refactor import-state CLI to cli package (#250)

* Remove unnecessary upstream config arg to event watcher

* Initialize event watcher in CLI package

* Refactor import-state CLI to cli package

packages/cli/src/base.ts

@@ -5,6 +5,7 @@
 import 'reflect-metadata';
 import assert from 'assert';
 import { ConnectionOptions } from 'typeorm';
+import { PubSub } from 'graphql-subscriptions';
 
 import { JsonRpcProvider } from '@ethersproject/providers';
 import { GraphWatcher, Database as GraphDatabase } from '@cerc-io/graph-node';
@@ -17,15 +18,44 @@ import {
   IndexerInterface,
   ServerConfig,
   Database as BaseDatabase,
-  Clients
+  Clients,
+  EventWatcherInterface
 } from '@cerc-io/util';
+import { EthClient } from '@cerc-io/ipld-eth-client';
 
 export class BaseCmd {
   _config?: Config;
   _clients?: Clients;
   _ethProvider?: JsonRpcProvider;
+  _jobQueue?: JobQueue
   _database?: DatabaseInterface;
   _indexer?: IndexerInterface;
+  _graphDb?: GraphDatabase
+  _eventWatcher?: EventWatcherInterface
+
+  get config (): Config | undefined {
+    return this._config;
+  }
+
+  get jobQueue (): JobQueue | undefined {
+    return this._jobQueue;
+  }
+
+  get database (): DatabaseInterface | undefined {
+    return this._database;
+  }
+
+  get graphDb (): GraphDatabase | undefined {
+    return this._graphDb;
+  }
+
+  get indexer (): IndexerInterface | undefined {
+    return this._indexer;
+  }
+
+  get eventWatcher (): EventWatcherInterface | undefined {
+    return this._eventWatcher;
+  }
 
   async initConfig<ConfigType> (configFile: string): Promise<ConfigType> {
     if (!this._config) {
@@ -49,10 +79,7 @@ export class BaseCmd {
       graphWatcher?: GraphWatcher
     ) => IndexerInterface,
     clients: { [key: string]: any } = {}
-  ): Promise<{
-    database: DatabaseInterface,
-    indexer: IndexerInterface
-  }> {
+  ): Promise<void> {
     assert(this._config);
 
     this._database = new Database(this._config.database, this._config.server);
@@ -64,8 +91,8 @@ export class BaseCmd {
     const { dbConnectionString, maxCompletionLagInSecs } = jobQueueConfig;
     assert(dbConnectionString, 'Missing job queue db connection string');
 
-    const jobQueue = new JobQueue({ dbConnectionString, maxCompletionLag: maxCompletionLagInSecs });
-    await jobQueue.start();
+    this._jobQueue = new JobQueue({ dbConnectionString, maxCompletionLag: maxCompletionLagInSecs });
+    await this._jobQueue.start();
 
     const { ethClient, ethProvider } = await initClients(this._config);
     this._ethProvider = ethProvider;
@@ -74,17 +101,33 @@ export class BaseCmd {
     // Check if subgraph watcher.
     if (this._config.server.subgraphPath) {
       const graphWatcher = await this._getGraphWatcher(this._database.baseDatabase);
-      this._indexer = new Indexer(this._config.server, this._database, this._clients, ethProvider, jobQueue, graphWatcher);
+      this._indexer = new Indexer(this._config.server, this._database, this._clients, ethProvider, this._jobQueue, graphWatcher);
       await this._indexer.init();
 
       graphWatcher.setIndexer(this._indexer);
       await graphWatcher.init();
     } else {
-      this._indexer = new Indexer(this._config.server, this._database, this._clients, ethProvider, jobQueue);
+      this._indexer = new Indexer(this._config.server, this._database, this._clients, ethProvider, this._jobQueue);
       await this._indexer.init();
     }
+  }
 
-    return { database: this._database, indexer: this._indexer };
+  async initEventWatcher (
+    EventWatcher: new(
+      ethClient: EthClient,
+      indexer: IndexerInterface,
+      pubsub: PubSub,
+      jobQueue: JobQueue
+    ) => EventWatcherInterface
+  ): Promise<void> {
+    assert(this._clients?.ethClient);
+    assert(this._indexer);
+    assert(this._jobQueue);
+
+    // 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();
+    this._eventWatcher = new EventWatcher(this._clients.ethClient, this._indexer, pubsub, this._jobQueue);
   }
 
   async _getGraphWatcher (baseDatabase: BaseDatabase): Promise<GraphWatcher> {
@@ -92,9 +135,9 @@ export class BaseCmd {
     assert(this._clients?.ethClient);
     assert(this._ethProvider);
 
-    const graphDb = new GraphDatabase(this._config.server, baseDatabase);
-    await graphDb.init();
+    this._graphDb = new GraphDatabase(this._config.server, baseDatabase);
+    await this._graphDb.init();
 
-    return new GraphWatcher(graphDb, this._clients.ethClient, this._ethProvider, this._config.server);
+    return new GraphWatcher(this._graphDb, this._clients.ethClient, this._ethProvider, this._config.server);
   }
 }

packages/cli/src/checkpoint/create.ts

@@ -60,17 +60,21 @@ export class CreateCheckpointCmd {
     this._argv = argv;
     await this.initConfig(argv.configFile);
 
-    ({ database: this._database, indexer: this._indexer } = await this._baseCmd.init(Database, Indexer, clients));
+    await this._baseCmd.init(Database, Indexer, clients);
   }
 
   async exec (): Promise<void> {
     assert(this._argv);
-    assert(this._database);
-    assert(this._indexer);
 
-    const blockHash = await this._indexer.processCLICheckpoint(this._argv.address, this._argv.blockHash);
+    const database = this._baseCmd.database;
+    const indexer = this._baseCmd.indexer;
 
-    await this._database.close();
+    assert(database);
+    assert(indexer);
+
+    const blockHash = await indexer.processCLICheckpoint(this._argv.address, this._argv.blockHash);
+
+    await database.close();
     log(`Created a checkpoint for contract ${this._argv.address} at block-hash ${blockHash}`);
   }
 }

packages/cli/src/import-state.ts

@@ -0,0 +1,181 @@
+//
+// Copyright 2022 Vulcanize, Inc.
+//
+
+import yargs from 'yargs';
+import 'reflect-metadata';
+import assert from 'assert';
+import path from 'path';
+import fs from 'fs';
+import debug from 'debug';
+import { ConnectionOptions } from 'typeorm';
+import { PubSub } from 'graphql-subscriptions';
+
+import { JsonRpcProvider } from '@ethersproject/providers';
+import { GraphWatcher, updateEntitiesFromState } from '@cerc-io/graph-node';
+import { EthClient } from '@cerc-io/ipld-eth-client';
+import {
+  DEFAULT_CONFIG_PATH,
+  JobQueue,
+  DatabaseInterface,
+  IndexerInterface,
+  ServerConfig,
+  Clients,
+  EventWatcherInterface,
+  fillBlocks,
+  StateKind
+} from '@cerc-io/util';
+import * as codec from '@ipld/dag-cbor';
+
+import { BaseCmd } from './base';
+
+const log = debug('vulcanize:import-state');
+
+interface Arguments {
+  configFile: string;
+  importFile: string;
+}
+
+export class ImportStateCmd {
+  _argv?: Arguments
+  _baseCmd: BaseCmd;
+
+  constructor () {
+    this._baseCmd = new BaseCmd();
+  }
+
+  async initConfig<ConfigType> (): Promise<ConfigType> {
+    this._argv = this._getArgv();
+    assert(this._argv);
+
+    return this._baseCmd.initConfig(this._argv.configFile);
+  }
+
+  async init (
+    Database: new (config: ConnectionOptions,
+      serverConfig?: ServerConfig
+    ) => DatabaseInterface,
+    Indexer: new (
+      serverConfig: ServerConfig,
+      db: DatabaseInterface,
+      clients: Clients,
+      ethProvider: JsonRpcProvider,
+      jobQueue: JobQueue,
+      graphWatcher?: GraphWatcher
+    ) => IndexerInterface,
+    EventWatcher: new(
+      ethClient: EthClient,
+      indexer: IndexerInterface,
+      pubsub: PubSub,
+      jobQueue: JobQueue
+    ) => EventWatcherInterface,
+    clients: { [key: string]: any } = {}
+  ): Promise<void> {
+    await this.initConfig();
+
+    await this._baseCmd.init(Database, Indexer, clients);
+    await this._baseCmd.initEventWatcher(EventWatcher);
+  }
+
+  async exec (State: new() => any): Promise<void> {
+    assert(this._argv);
+
+    const config = this._baseCmd.config;
+    const jobQueue = this._baseCmd.jobQueue;
+    const database = this._baseCmd.database;
+    const indexer = this._baseCmd.indexer;
+    const eventWatcher = this._baseCmd.eventWatcher;
+
+    assert(config);
+    assert(jobQueue);
+    assert(database);
+    assert(indexer);
+    assert(eventWatcher);
+
+    // Import data.
+    const importFilePath = path.resolve(this._argv.importFile);
+    const encodedImportData = fs.readFileSync(importFilePath);
+    const importData = codec.decode(Buffer.from(encodedImportData)) as any;
+
+    // Fill the snapshot block.
+    await fillBlocks(
+      jobQueue,
+      indexer,
+      eventWatcher,
+      config.jobQueue.blockDelayInMilliSecs,
+      {
+        prefetch: true,
+        startBlock: importData.snapshotBlock.blockNumber,
+        endBlock: importData.snapshotBlock.blockNumber
+      }
+    );
+
+    // Fill the Contracts.
+    for (const contract of importData.contracts) {
+      indexer.watchContract(contract.address, contract.kind, contract.checkpoint, contract.startingBlock);
+    }
+
+    // Get the snapshot block.
+    const block = await indexer.getBlockProgress(importData.snapshotBlock.blockHash);
+    assert(block);
+
+    // Fill the States.
+    for (const checkpoint of importData.stateCheckpoints) {
+      let state = new State();
+
+      state = Object.assign(state, checkpoint);
+      state.block = block;
+      state.data = Buffer.from(codec.encode(state.data));
+
+      state = await indexer.saveOrUpdateState(state);
+
+      // Fill entities using State if:
+      //  relationsMap defined for the watcher,
+      //  graphDb instance is avaiable
+      // TODO: Fill latest entity tables
+      if (indexer.getRelationsMap) {
+        if (this._baseCmd.graphDb) {
+          await updateEntitiesFromState(this._baseCmd.graphDb, indexer, state);
+        } else if (database.graphDatabase) {
+          await updateEntitiesFromState(database.graphDatabase, indexer, state);
+        }
+      }
+    }
+
+    // Mark snapshot block as completely processed.
+    block.isComplete = true;
+    await indexer.updateBlockProgress(block, block.lastProcessedEventIndex);
+    await indexer.updateSyncStatusChainHead(block.blockHash, block.blockNumber);
+    await indexer.updateSyncStatusIndexedBlock(block.blockHash, block.blockNumber);
+    await indexer.updateStateSyncStatusIndexedBlock(block.blockNumber);
+    await indexer.updateStateSyncStatusCheckpointBlock(block.blockNumber);
+
+    // The 'diff_staged' and 'init' State entries are unnecessary as checkpoints have been already created for the snapshot block.
+    await indexer.removeStates(block.blockNumber, StateKind.Init);
+    await indexer.removeStates(block.blockNumber, StateKind.DiffStaged);
+
+    log(`Import completed for snapshot block at height ${block.blockNumber}`);
+    await database.close();
+  }
+
+  _getArgv (): any {
+    return yargs.parserConfiguration({
+      'parse-numbers': false
+    }).options({
+      configFile: {
+        alias: 'f',
+        type: 'string',
+        require: true,
+        demandOption: true,
+        describe: 'Configuration file path (toml)',
+        default: DEFAULT_CONFIG_PATH
+      },
+      importFile: {
+        alias: 'i',
+        type: 'string',
+        demandOption: true,
+        describe: 'Import file path (JSON)'
+      }
+    }).argv;
+  }
+}

packages/cli/src/index.ts

@@ -7,3 +7,4 @@ export * from './reset/watcher';
 export * from './reset/state';
 export * from './checkpoint/create';
 export * from './inspect-cid';
+export * from './import-state';

packages/cli/src/inspect-cid.ts

@@ -32,8 +32,6 @@ interface Arguments {
 export class InspectCIDCmd {
   _argv?: Arguments
   _baseCmd: BaseCmd;
-  _database?: DatabaseInterface;
-  _indexer?: IndexerInterface;
 
   constructor () {
     this._baseCmd = new BaseCmd();
@@ -63,21 +61,25 @@ export class InspectCIDCmd {
   ): Promise<void> {
     await this.initConfig();
 
-    ({ database: this._database, indexer: this._indexer } = await this._baseCmd.init(Database, Indexer, clients));
+    await this._baseCmd.init(Database, Indexer, clients);
   }
 
   async exec (): Promise<void> {
     assert(this._argv);
-    assert(this._database);
-    assert(this._indexer);
 
-    const state = await this._indexer.getStateByCID(this._argv.cid);
+    const database = this._baseCmd.database;
+    const indexer = this._baseCmd.indexer;
+
+    assert(database);
+    assert(indexer);
+
+    const state = await indexer.getStateByCID(this._argv.cid);
     assert(state, 'State for the provided CID doesn\'t exist.');
 
-    const stateData = await this._indexer.getStateData(state);
+    const stateData = await indexer.getStateData(state);
     log(util.inspect(stateData, false, null));
 
-    await this._database.close();
+    await database.close();
   }
 
   _getArgv (): any {

packages/cli/src/reset/watcher.ts

@@ -29,8 +29,6 @@ interface Arguments {
 export class ResetWatcherCmd {
   _argv?: Arguments
   _baseCmd: BaseCmd;
-  _database?: DatabaseInterface;
-  _indexer?: IndexerInterface;
 
   constructor () {
     this._baseCmd = new BaseCmd();
@@ -59,17 +57,21 @@ export class ResetWatcherCmd {
     this._argv = argv;
     await this.initConfig(argv.configFile);
 
-    ({ database: this._database, indexer: this._indexer } = await this._baseCmd.init(Database, Indexer, clients));
+    await this._baseCmd.init(Database, Indexer, clients);
   }
 
   async exec (): Promise<void> {
     assert(this._argv);
-    assert(this._database);
-    assert(this._indexer);
 
-    await this._indexer.resetWatcherToBlock(this._argv.blockNumber);
+    const database = this._baseCmd.database;
+    const indexer = this._baseCmd.indexer;
 
-    await this._database.close();
+    assert(database);
+    assert(indexer);
+
+    await indexer.resetWatcherToBlock(this._argv.blockNumber);
+
+    await database.close();
     log('Reset watcher successfully');
   }
 }

packages/cli/src/watch-contract.ts

@@ -31,8 +31,6 @@ interface Arguments {
 export class WatchContractCmd {
   _argv?: Arguments;
   _baseCmd: BaseCmd;
-  _database?: DatabaseInterface;
-  _indexer?: IndexerInterface;
 
   constructor () {
     this._baseCmd = new BaseCmd();
@@ -62,17 +60,20 @@ export class WatchContractCmd {
   ): Promise<void> {
     await this.initConfig();
 
-    ({ database: this._database, indexer: this._indexer } = await this._baseCmd.init(Database, Indexer, clients));
+    await this._baseCmd.init(Database, Indexer, clients);
   }
 
   async exec (): Promise<void> {
     assert(this._argv);
-    assert(this._database);
-    assert(this._indexer);
-    assert(this._indexer.watchContract);
 
-    await this._indexer.watchContract(this._argv.address, this._argv.kind, this._argv.checkpoint, this._argv.startingBlock);
-    await this._database.close();
+    const database = this._baseCmd.database;
+    const indexer = this._baseCmd.indexer;
+
+    assert(database);
+    assert(indexer);
+
+    await indexer.watchContract(this._argv.address, this._argv.kind, this._argv.checkpoint, this._argv.startingBlock);
+    await database.close();
   }
 
   _getArgv (): any {