vulcanize · nikugogoi · Nov 23, 2022 · Nov 23, 2022 · Nov 23, 2022
diff --git a/packages/erc20-watcher/src/events.ts b/packages/erc20-watcher/src/events.ts
@@ -9,7 +9,8 @@ import {
   EventWatcher as BaseEventWatcher,
   QUEUE_BLOCK_PROCESSING,
   QUEUE_EVENT_PROCESSING,
-  JobQueue
+  JobQueue,
+  IndexerInterface
 } from '@cerc-io/util';
 import { EthClient } from '@cerc-io/ipld-eth-client';
 
@@ -23,12 +24,12 @@ export class EventWatcher {
   _pubsub: PubSub
   _jobQueue: JobQueue
 
-  constructor (ethClient: EthClient, indexer: Indexer, pubsub: PubSub, jobQueue: JobQueue) {
+  constructor (ethClient: EthClient, indexer: IndexerInterface, pubsub: PubSub, jobQueue: JobQueue) {
     assert(ethClient);
     assert(indexer);
 
     this._ethClient = ethClient;
-    this._indexer = indexer;
+    this._indexer = indexer as Indexer;
     this._pubsub = pubsub;
     this._jobQueue = jobQueue;
     this._baseEventWatcher = new BaseEventWatcher(this._ethClient, this._indexer, this._pubsub, this._jobQueue);

diff --git a/packages/erc20-watcher/src/resolvers.ts b/packages/erc20-watcher/src/resolvers.ts
@@ -6,15 +6,16 @@ import assert from 'assert';
 import BigInt from 'apollo-type-bigint';
 import debug from 'debug';
 
-import { gqlTotalQueryCount, gqlQueryCount, ValueResult } from '@cerc-io/util';
+import { gqlTotalQueryCount, gqlQueryCount, ValueResult, IndexerInterface, EventWatcherInterface } from '@cerc-io/util';
 
 import { CONTRACT_KIND, Indexer } from './indexer';
 import { EventWatcher } from './events';
 
 const log = debug('vulcanize:resolver');
 
-export const createResolvers = async (indexer: Indexer, eventWatcher: EventWatcher): Promise<any> => {
-  assert(indexer);
+export const createResolvers = async (indexerArg: IndexerInterface, eventWatcherArg: EventWatcherInterface): Promise<any> => {
+  const indexer = indexerArg as Indexer;
+  const eventWatcher = eventWatcherArg as EventWatcher;
 
   return {
     BigInt: new BigInt('bigInt'),

diff --git a/packages/erc20-watcher/src/server.ts b/packages/erc20-watcher/src/server.ts
@@ -2,30 +2,12 @@
 // Copyright 2021 Vulcanize, Inc.
 //
 
-import assert from 'assert';
-import 'reflect-metadata';
-import express, { Application } from 'express';
-import { PubSub } from 'graphql-subscriptions';
-import yargs from 'yargs';
-import { hideBin } from 'yargs/helpers';
 import debug from 'debug';
 import 'graphql-import-node';
 
-import { getCache } from '@cerc-io/cache';
-import {
-  KIND_ACTIVE,
-  DEFAULT_CONFIG_PATH,
-  createAndStartServer,
-  JobQueue,
-  getCustomProvider,
-  startGQLMetricsServer,
-  getConfig
-} from '@cerc-io/util';
-import { Config } from '@vulcanize/util';
-import { EthClient } from '@cerc-io/ipld-eth-client';
+import { ServerCmd } from '@cerc-io/cli';
 
 import typeDefs from './schema';
-
 import { createResolvers as createMockResolvers } from './mock/resolvers';
 import { createResolvers } from './resolvers';
 import { Indexer } from './indexer';
@@ -35,73 +17,10 @@ import { EventWatcher } from './events';
 const log = debug('vulcanize:server');
 
 export const main = async (): Promise<any> => {
-  const argv = await yargs(hideBin(process.argv))
-    .option('f', {
-      alias: 'config-file',
-      demandOption: true,
-      describe: 'configuration file path (toml)',
-      type: 'string',
-      default: DEFAULT_CONFIG_PATH
-    })
-    .argv;
-
-  const config: Config = await getConfig(argv.f);
-
-  assert(config.server, 'Missing server config');
-
-  const { kind: watcherKind } = config.server;
-
-  const { upstream, database: dbConfig, jobQueue: jobQueueConfig } = config;
-
-  assert(dbConfig, 'Missing database config');
-
-  const db = new Database(dbConfig);
-  await db.init();
-
-  assert(upstream, 'Missing upstream config');
-  const { ethServer: { gqlApiEndpoint, rpcProviderEndpoint }, cache: cacheConfig } = upstream;
-  assert(gqlApiEndpoint, 'Missing upstream ethServer.gqlApiEndpoint');
-
-  const cache = await getCache(cacheConfig);
-  const ethClient = new EthClient({
-    gqlEndpoint: gqlApiEndpoint,
-    cache
-  });
-
-  const ethProvider = getCustomProvider(rpcProviderEndpoint);
-
-  // 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();
-
-  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 indexer = new Indexer(config.server, db, { ethClient }, ethProvider, jobQueue);
-  await indexer.init();
-
-  const eventWatcher = new EventWatcher(ethClient, indexer, pubsub, jobQueue);
-
-  if (watcherKind === KIND_ACTIVE) {
-    await jobQueue.start();
-    // Delete jobs to prevent creating jobs after completion of processing previous block.
-    await jobQueue.deleteAllJobs();
-    await eventWatcher.start();
-  }
-
-  const resolvers = process.env.MOCK ? await createMockResolvers() : await createResolvers(indexer, eventWatcher);
-
-  // Create an Express app and server
-  const app: Application = express();
-  const server = createAndStartServer(app, typeDefs, resolvers, config.server);
-
-  await startGQLMetricsServer(config);
+  const serverCmd = new ServerCmd();
+  await serverCmd.init(Database, Indexer, EventWatcher);
 
-  return { app, server };
+  return process.env.MOCK ? serverCmd.exec(createMockResolvers, typeDefs) : serverCmd.exec(createResolvers, typeDefs);
 };
 
 main().then(() => {

diff --git a/packages/uni-info-watcher/environments/local.toml b/packages/uni-info-watcher/environments/local.toml
@@ -4,6 +4,7 @@
   # Use mode demo when running watcher locally.
   # Mode demo whitelists all tokens so that entity values get updated.
   mode = "prod"
+  kind = "active"
 
   # Checkpointing state.
   checkpointing = true

diff --git a/packages/uni-info-watcher/src/resolvers.ts b/packages/uni-info-watcher/src/resolvers.ts
@@ -8,7 +8,17 @@ import debug from 'debug';
 import { GraphQLResolveInfo, GraphQLScalarType } from 'graphql';
 import JSONbig from 'json-bigint';
 
-import { BlockHeight, OrderDirection, getResultState, setGQLCacheHints, GraphDecimal, gqlQueryCount, gqlTotalQueryCount } from '@cerc-io/util';
+import {
+  BlockHeight,
+  OrderDirection,
+  getResultState,
+  setGQLCacheHints,
+  GraphDecimal,
+  gqlQueryCount,
+  gqlTotalQueryCount,
+  IndexerInterface,
+  EventWatcherInterface
+} from '@cerc-io/util';
 
 import { Indexer } from './indexer';
 import { Burn } from './entity/Burn';
@@ -38,8 +48,9 @@ const log = debug('vulcanize:resolver');
 
 export { BlockHeight };
 
-export const createResolvers = async (indexer: Indexer, eventWatcher: EventWatcher): Promise<any> => {
-  assert(indexer);
+export const createResolvers = async (indexerArg: IndexerInterface, eventWatcherArg: EventWatcherInterface): Promise<any> => {
+  const indexer = indexerArg as Indexer;
+  const eventWatcher = eventWatcherArg as EventWatcher;
 
   const gqlCacheConfig = indexer.serverConfig.gqlCache;
 

diff --git a/packages/uni-info-watcher/src/server.ts b/packages/uni-info-watcher/src/server.ts
@@ -2,24 +2,16 @@
 // Copyright 2021 Vulcanize, Inc.
 //
 
-import assert from 'assert';
 import 'reflect-metadata';
-import express, { Application } from 'express';
-import { PubSub } from 'graphql-subscriptions';
-import yargs from 'yargs';
-import { hideBin } from 'yargs/helpers';
 import debug from 'debug';
 import 'graphql-import-node';
 
+import { ServerCmd } from '@cerc-io/cli';
+import { Config } from '@vulcanize/util';
 import { Client as ERC20Client } from '@vulcanize/erc20-watcher';
 import { Client as UniClient } from '@vulcanize/uni-watcher';
-import { Config } from '@vulcanize/util';
-import { createAndStartServer, DEFAULT_CONFIG_PATH, JobQueue, getCustomProvider, startGQLMetricsServer, getConfig } from '@cerc-io/util';
-import { getCache } from '@cerc-io/cache';
-import { EthClient } from '@cerc-io/ipld-eth-client';
 
 import typeDefs from './schema';
-
 import { createResolvers as createMockResolvers } from './mock/resolvers';
 import { createResolvers } from './resolvers';
 import { Indexer } from './indexer';
@@ -29,75 +21,20 @@ import { EventWatcher } from './events';
 const log = debug('vulcanize:server');
 
 export const main = async (): Promise<any> => {
-  const argv = await yargs(hideBin(process.argv))
-    .option('f', {
-      alias: 'config-file',
-      demandOption: true,
-      describe: 'configuration file path (toml)',
-      type: 'string',
-      default: DEFAULT_CONFIG_PATH
-    })
-    .argv;
-
-  const config: Config = await getConfig(argv.f);
-
-  assert(config.server, 'Missing server config');
+  const serverCmd = new ServerCmd();
 
-  const { upstream, database: dbConfig, jobQueue: jobQueueConfig } = config;
-
-  assert(dbConfig, 'Missing database config');
-
-  const db = new Database(dbConfig, config.server);
-  await db.init();
-
-  assert(upstream, 'Missing upstream config');
+  const config: Config = await serverCmd.initConfig();
   const {
-    ethServer: {
-      gqlApiEndpoint,
-      rpcProviderEndpoint
-    },
     uniWatcher,
-    tokenWatcher,
-    cache: cacheConfig
-  } = upstream;
-
-  assert(gqlApiEndpoint, 'Missing upstream ethServer.gqlApiEndpoint');
-
-  const cache = await getCache(cacheConfig);
-  const ethClient = new EthClient({
-    gqlEndpoint: gqlApiEndpoint,
-    cache
-  });
+    tokenWatcher
+  } = config.upstream;
 
   const uniClient = new UniClient(uniWatcher);
   const erc20Client = new ERC20Client(tokenWatcher);
-  const ethProvider = getCustomProvider(rpcProviderEndpoint);
-
-  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 indexer = new Indexer(config.server, db, { uniClient, erc20Client, ethClient }, ethProvider, jobQueue);
-
-  const pubSub = new PubSub();
-  const eventWatcher = new EventWatcher(ethClient, indexer, pubSub, jobQueue);
-  // Delete jobs to prevent creating jobs after completion of processing previous block.
-  await jobQueue.deleteAllJobs();
-  await eventWatcher.start();
-
-  const resolvers = process.env.MOCK ? await createMockResolvers() : await createResolvers(indexer, eventWatcher);
-
-  // Create an Express app and server
-  const app: Application = express();
-  const server = createAndStartServer(app, typeDefs, resolvers, config.server);
 
-  await startGQLMetricsServer(config);
+  await serverCmd.init(Database, Indexer, EventWatcher, { uniClient, erc20Client });
 
-  return { app, server };
+  return process.env.MOCK ? serverCmd.exec(createMockResolvers, typeDefs) : serverCmd.exec(createResolvers, typeDefs);
 };
 
 main().then(() => {

diff --git a/packages/uni-watcher/environments/local.toml b/packages/uni-watcher/environments/local.toml
@@ -1,6 +1,7 @@
 [server]
   host = "127.0.0.1"
   port = 3003
+  kind = "active"
 
 [metrics]
   host = "127.0.0.1"

diff --git a/packages/uni-watcher/src/events.ts b/packages/uni-watcher/src/events.ts
@@ -11,7 +11,8 @@ import {
   QUEUE_BLOCK_PROCESSING,
   QUEUE_EVENT_PROCESSING,
   JobQueue,
-  EventWatcherInterface
+  EventWatcherInterface,
+  IndexerInterface
 } from '@cerc-io/util';
 
 import { Indexer } from './indexer';
@@ -24,9 +25,9 @@ export class EventWatcher implements EventWatcherInterface {
   _jobQueue: JobQueue
   _baseEventWatcher: BaseEventWatcher
 
-  constructor (ethClient: EthClient, indexer: Indexer, pubsub: PubSub, jobQueue: JobQueue) {
+  constructor (ethClient: EthClient, indexer: IndexerInterface, pubsub: PubSub, jobQueue: JobQueue) {
     this._ethClient = ethClient;
-    this._indexer = indexer;
+    this._indexer = indexer as Indexer;
     this._pubsub = pubsub;
     this._jobQueue = jobQueue;
     this._baseEventWatcher = new BaseEventWatcher(this._ethClient, this._indexer, this._pubsub, this._jobQueue);

diff --git a/packages/uni-watcher/src/resolvers.ts b/packages/uni-watcher/src/resolvers.ts
@@ -6,14 +6,23 @@ import assert from 'assert';
 import BigInt from 'apollo-type-bigint';
 import debug from 'debug';
 
-import { gqlTotalQueryCount, gqlQueryCount, ValueResult } from '@cerc-io/util';
+import {
+  gqlTotalQueryCount,
+  gqlQueryCount,
+  ValueResult,
+  IndexerInterface,
+  EventWatcherInterface
+} from '@cerc-io/util';
 
 import { Indexer } from './indexer';
 import { EventWatcher } from './events';
 
 const log = debug('vulcanize:resolver');
 
-export const createResolvers = async (indexer: Indexer, eventWatcher: EventWatcher): Promise<any> => {
+export const createResolvers = async (indexerArg: IndexerInterface, eventWatcherArg: EventWatcherInterface): Promise<any> => {
+  const indexer = indexerArg as Indexer;
+  const eventWatcher = eventWatcherArg as EventWatcher;
+
   return {
     BigInt: new BigInt('bigInt'),