Watch upstream ERC20 events to trigger indexing (#43)

* Move to apollo client, enables subscriptions. * Watch logs and trigger other indexer methods. * Refactoring config loading, watched contracts table. * Check event sync progress inside transaction. * Refactoring server startup.
2024-11-19 20:36:19 +00:00 · 2021-06-08 16:07:52 +05:30 · 2021-06-08 16:07:52 +05:30 · 84e1927402
commit 84e1927402
parent a13a909a85
15 changed files with 543 additions and 139 deletions
--- a/packages/ipld-eth-client/package.json
+++ b/packages/ipld-eth-client/package.json
@ -20,13 +20,17 @@
  },
  "homepage": "https://github.com/vulcanize/erc20-watcher#readme",
  "dependencies": {
+    "@apollo/client": "^3.3.19",
    "@vulcanize/cache": "^0.1.0",
+    "cross-fetch": "^3.1.4",
    "ethers": "^5.2.0",
    "graphql": "^15.5.0",
-    "graphql-request": "^3.4.0",
-    "left-pad": "^1.3.0"
+    "left-pad": "^1.3.0",
+    "subscriptions-transport-ws": "^0.9.18",
+    "ws": "^7.4.6"
  },
  "devDependencies": {
+    "@types/ws": "^7.4.4",
    "@typescript-eslint/eslint-plugin": "^4.25.0",
    "@typescript-eslint/parser": "^4.25.0",
    "eslint": "^7.27.0",
--- a/packages/ipld-eth-client/src/eth-client.ts
+++ b/packages/ipld-eth-client/src/eth-client.ts
@ -1,14 +1,23 @@
 import assert from 'assert';
-import { GraphQLClient } from 'graphql-request';
+import debug from 'debug';
+import fetch from 'cross-fetch';
+import { SubscriptionClient } from 'subscriptions-transport-ws';
+import ws from 'ws';

+import { ApolloClient, NormalizedCacheObject, split, HttpLink, InMemoryCache } from '@apollo/client/core';
+import { getMainDefinition } from '@apollo/client/utilities';
+import { WebSocketLink } from '@apollo/client/link/ws';
 import { Cache } from '@vulcanize/cache';

 import ethQueries from './eth-queries';
 import { padKey } from './utils';

+const log = debug('vulcanize:eth-client');
+
 interface Config {
  gqlEndpoint: string;
-  cache: Cache;
+  gqlSubscriptionEndpoint: string;
+  cache: Cache | undefined;
 }

 interface Vars {
@ -19,16 +28,55 @@ interface Vars {

 export class EthClient {
  _config: Config;
-  _client: GraphQLClient;
-  _cache: Cache;
+  _client: ApolloClient<NormalizedCacheObject>;
+  _cache: Cache | undefined;

  constructor (config: Config) {
    this._config = config;

-    const { gqlEndpoint, cache } = config;
-    assert(gqlEndpoint, 'Missing gql endpoint');
+    const { gqlEndpoint, gqlSubscriptionEndpoint, cache } = config;
+
+    assert(gqlEndpoint, 'Missing gql endpoint');
+    assert(gqlSubscriptionEndpoint, 'Missing gql subscription endpoint');
+
+    // https://www.apollographql.com/docs/react/data/subscriptions/
+    const subscriptionClient = new SubscriptionClient(gqlSubscriptionEndpoint, {
+      reconnect: true,
+      connectionCallback: (error: Error[]) => {
+        if (error) {
+          log('subscription client connection', error[0].message);
+        }
+      }
+    }, ws);
+
+    subscriptionClient.onError(error => {
+      log('subscription client error', error.message);
+    });
+
+    const httpLink = new HttpLink({
+      uri: gqlEndpoint,
+      fetch
+    });
+
+    const wsLink = new WebSocketLink(subscriptionClient);
+
+    const splitLink = split(
+      ({ query }) => {
+        const definition = getMainDefinition(query);
+        return (
+          definition.kind === 'OperationDefinition' &&
+          definition.operation === 'subscription'
+        );
+      },
+      wsLink,
+      httpLink
+    );
+
+    this._client = new ApolloClient({
+      link: splitLink,
+      cache: new InMemoryCache()
+    });

-    this._client = new GraphQLClient(gqlEndpoint);
    this._cache = cache;
  }

@ -63,6 +111,18 @@ export class EthClient {
    return logs;
  }

+  async watchLogs (onNext: (value: any) => void): Promise<ZenObservable.Subscription> {
+    const observable = await this._client.subscribe({
+      query: ethQueries.subscribeLogs
+    });
+
+    return observable.subscribe({
+      next (data) {
+        onNext(data);
+      }
+    });
+  }
+
  async _getCachedOrFetch (queryName: keyof typeof ethQueries, vars: Vars): Promise<any> {
    const keyObj = {
      queryName,
@ -77,8 +137,8 @@ export class EthClient {
      }
    }

-    // Not cached or cache disabled, need to perform an upstream GQL query.
-    const result = await this._client.request(ethQueries[queryName], vars);
+    // Result not cached or cache disabled, need to perform an upstream GQL query.
+    const { data: result } = await this._client.query({ query: ethQueries[queryName], variables: vars });

    // Cache the result and return it, if cache is enabled.
    if (this._cache) {
--- a/packages/ipld-eth-client/src/eth-queries.ts
+++ b/packages/ipld-eth-client/src/eth-queries.ts
@ -1,4 +1,4 @@
-import { gql } from 'graphql-request';
+import { gql } from '@apollo/client/core';

 export const getStorageAt = gql`
 query getStorageAt($blockHash: Bytes32!, $contract: Address!, $slot: Bytes32!) {
@ -24,7 +24,31 @@ query getLogs($blockHash: Bytes32!, $contract: Address!) {
 }
 `;

+export const subscribeLogs = gql`
+subscription SubscriptionReceipt {
+  listen(topic: "receipt_cids") {
+    relatedNode {
+      ... on ReceiptCid {
+        logContracts
+        topic0S
+        topic1S
+        topic2S
+        topic3S
+        contract
+        ethTransactionCidByTxId {
+          ethHeaderCidByHeaderId {
+            blockHash
+            blockNumber
+          }
+        }
+      }
+    }
+  }
+}
+`;
+
 export default {
  getStorageAt,
-  getLogs
+  getLogs,
+  subscribeLogs
 };
--- a/packages/watcher/environments/local.toml
+++ b/packages/watcher/environments/local.toml
@ -12,10 +12,20 @@
  synchronize = true
  logging = true

+  entities = [ "src/entity/**/*.ts" ]
+  migrations = [ "src/migration/**/*.ts" ]
+  subscribers = [ "src/subscriber/**/*.ts" ]
+
+  [database.cli]
+    entitiesDir = "src/entity"
+    migrationsDir = "src/migration"
+    subscribersDir = "src/subscriber"
+
 [upstream]
  gqlEndpoint = "http://127.0.0.1:8083/graphql"
+  gqlSubscriptionEndpoint = "http://127.0.0.1:5000/graphql"

  [upstream.cache]
    name = "requests"
-    enabled = true
+    enabled = false
    deleteOnStart = false
--- a/packages/watcher/package.json
+++ b/packages/watcher/package.json
@ -28,6 +28,7 @@
    "@vulcanize/solidity-mapper": "^0.1.0",
    "apollo-type-bigint": "^0.1.3",
    "debug": "^4.3.1",
+    "ethers": "^5.2.0",
    "express": "^4.17.1",
    "express-graphql": "^0.12.0",
    "fs-extra": "^10.0.0",
--- a/packages/watcher/src/cli/watch-contract.ts
+++ b/packages/watcher/src/cli/watch-contract.ts
@ -0,0 +1,45 @@
+import assert from 'assert';
+import yargs from 'yargs';
+import 'reflect-metadata';
+import { ethers } from 'ethers';
+
+import { Config, getConfig } from '../config';
+import { Database } from '../database';
+
+(async () => {
+  const argv = await yargs.parserConfiguration({
+    'parse-numbers': false
+  }).options({
+    configFile: {
+      type: 'string',
+      require: true,
+      demandOption: true,
+      describe: 'configuration file path (toml)'
+    },
+    address: {
+      type: 'string',
+      require: true,
+      demandOption: true,
+      describe: 'Address of the deployed contract'
+    },
+    startingBlock: {
+      type: 'number',
+      default: 1,
+      describe: 'Starting block'
+    }
+  }).argv;
+
+  const config: Config = await getConfig(argv.configFile);
+  const { database: dbConfig } = config;
+
+  assert(dbConfig);
+
+  const db = new Database(dbConfig);
+  await db.init();
+
+  // Always use the checksum address (https://docs.ethers.io/v5/api/utils/address/#utils-getAddress).
+  const address = ethers.utils.getAddress(argv.address);
+
+  await db.saveContract(address, argv.startingBlock);
+  await db.close();
+})();
--- a/packages/watcher/src/config.ts
+++ b/packages/watcher/src/config.ts
@ -0,0 +1,35 @@
+import fs from 'fs-extra';
+import path from 'path';
+import toml from 'toml';
+import debug from 'debug';
+import { ConnectionOptions } from 'typeorm';
+
+import { Config as CacheConfig } from '@vulcanize/cache';
+
+const log = debug('vulcanize:config');
+
+export interface Config {
+  server: {
+    host: string;
+    port: number;
+  };
+  database: ConnectionOptions;
+  upstream: {
+    gqlEndpoint: string;
+    gqlSubscriptionEndpoint: string;
+    cache: CacheConfig
+  }
+}
+
+export const getConfig = async (configFile: string): Promise<Config> => {
+  const configFilePath = path.resolve(configFile);
+  const fileExists = await fs.pathExists(configFilePath);
+  if (!fileExists) {
+    throw new Error(`Config file not found: ${configFilePath}`);
+  }
+
+  const config = toml.parse(await fs.readFile(configFilePath, 'utf8'));
+  log('config', JSON.stringify(config, null, 2));
+
+  return config;
+};
--- a/packages/watcher/src/database.ts
+++ b/packages/watcher/src/database.ts
@ -4,6 +4,7 @@ import { SnakeNamingStrategy } from 'typeorm-naming-strategies';

 import { Allowance } from './entity/Allowance';
 import { Balance } from './entity/Balance';
+import { Contract } from './entity/Contract';
 import { Event } from './entity/Event';
 import { EventSyncProgress } from './entity/EventProgress';

@ -25,11 +26,11 @@ export class Database {
    });
  }

-  async getBalance ({ blockHash, token, owner }: { blockHash: string, token: string, owner: string }): Promise<Balance | undefined> {
-    if (!this._conn) {
-      return;
-    }
+  async close (): Promise<void> {
+    return this._conn.close();
+  }

+  async getBalance ({ blockHash, token, owner }: { blockHash: string, token: string, owner: string }): Promise<Balance | undefined> {
    return this._conn.getRepository(Balance)
      .createQueryBuilder('balance')
      .where('block_hash = :blockHash AND token = :token AND owner = :owner', {
@ -41,10 +42,6 @@ export class Database {
  }

  async getAllowance ({ blockHash, token, owner, spender }: { blockHash: string, token: string, owner: string, spender: string }): Promise<Allowance | undefined> {
-    if (!this._conn) {
-      return;
-    }
-
    return this._conn.getRepository(Allowance)
      .createQueryBuilder('allowance')
      .where('block_hash = :blockHash AND token = :token AND owner = :owner AND spender = :spender', {
@ -56,32 +53,20 @@ export class Database {
      .getOne();
  }

-  async saveBalance ({ blockHash, token, owner, value, proof }: DeepPartial<Balance>): Promise<Balance | undefined> {
-    if (!this._conn) {
-      return;
-    }
-
+  async saveBalance ({ blockHash, token, owner, value, proof }: DeepPartial<Balance>): Promise<Balance> {
    const repo = this._conn.getRepository(Balance);
    const entity = repo.create({ blockHash, token, owner, value, proof });
    return repo.save(entity);
  }

-  async saveAllowance ({ blockHash, token, owner, spender, value, proof }: DeepPartial<Allowance>): Promise<Allowance | undefined> {
-    if (!this._conn) {
-      return;
-    }
-
+  async saveAllowance ({ blockHash, token, owner, spender, value, proof }: DeepPartial<Allowance>): Promise<Allowance> {
    const repo = this._conn.getRepository(Allowance);
    const entity = repo.create({ blockHash, token, owner, spender, value, proof });
    return repo.save(entity);
  }

  // Returns true if events have already been synced for the (block, token) combination.
-  async didSyncEvents ({ blockHash, token }: { blockHash: string, token: string }): Promise<boolean | undefined> {
-    if (!this._conn) {
-      return;
-    }
-
+  async didSyncEvents ({ blockHash, token }: { blockHash: string, token: string }): Promise<boolean> {
    const numRows = await this._conn.getRepository(EventSyncProgress)
      .createQueryBuilder()
      .where('block_hash = :blockHash AND token = :token', {
@ -104,10 +89,6 @@ export class Database {
  }

  async getEventsByName ({ blockHash, token, eventName }: { blockHash: string, token: string, eventName: string }): Promise<Event[] | undefined> {
-    if (!this._conn) {
-      return;
-    }
-
    return this._conn.getRepository(Event)
      .createQueryBuilder('event')
      .where('block_hash = :blockHash AND token = :token AND :eventName = :eventName', {
@ -119,28 +100,59 @@ export class Database {
  }

  async saveEvents ({ blockHash, token, events }: { blockHash: string, token: string, events: DeepPartial<Event>[] }): Promise<void> {
-    if (!this._conn) {
-      return;
-    }
-
-    // TODO: Using the same connection doesn't work when > 1 inserts are attempted at the same time (e.g. simultaneous GQL requests).
-
    // In a transaction:
    // (1) Save all the events in the database.
    // (2) Add an entry to the event progress table.

    await this._conn.transaction(async (tx) => {
-      // Bulk insert events.
-      await tx.createQueryBuilder()
-        .insert()
-        .into(Event)
-        .values(events)
-        .execute();
-
-      // Update event sync progress.
      const repo = tx.getRepository(EventSyncProgress);
-      const progress = repo.create({ blockHash, token });
-      await repo.save(progress);
+
+      // Check sync progress inside the transaction.
+      const numRows = await repo
+        .createQueryBuilder()
+        .where('block_hash = :blockHash AND token = :token', {
+          blockHash,
+          token
+        })
+        .getCount();
+
+      if (numRows === 0) {
+        // Bulk insert events.
+        await tx.createQueryBuilder()
+          .insert()
+          .into(Event)
+          .values(events)
+          .execute();
+
+        // Update event sync progress.
+        const progress = repo.create({ blockHash, token });
+        await repo.save(progress);
+      }
+    });
+  }
+
+  async isWatchedContract (address: string): Promise<boolean> {
+    const numRows = await this._conn.getRepository(Contract)
+      .createQueryBuilder()
+      .where('address = :address', { address })
+      .getCount();
+
+    return numRows > 0;
+  }
+
+  async saveContract (address: string, startingBlock: number): Promise<void> {
+    await this._conn.transaction(async (tx) => {
+      const repo = tx.getRepository(Contract);
+
+      const numRows = await repo
+        .createQueryBuilder()
+        .where('address = :address', { address })
+        .getCount();
+
+      if (numRows === 0) {
+        const entity = repo.create({ address, startingBlock });
+        await repo.save(entity);
+      }
    });
  }
 }
--- a/packages/watcher/src/entity/Contract.ts
+++ b/packages/watcher/src/entity/Contract.ts
@ -0,0 +1,14 @@
+import { Entity, PrimaryGeneratedColumn, Column, Index } from 'typeorm';
+
+@Entity()
+@Index(['address'], { unique: true })
+export class Contract {
+  @PrimaryGeneratedColumn()
+  id!: number;
+
+  @Column('varchar', { length: 42 })
+  address!: string;
+
+  @Column('numeric')
+  startingBlock!: BigInt;
+}
--- a/packages/watcher/src/events.ts
+++ b/packages/watcher/src/events.ts
@ -0,0 +1,59 @@
+import assert from 'assert';
+import debug from 'debug';
+import _ from 'lodash';
+
+import { EthClient } from '@vulcanize/ipld-eth-client';
+
+import { Indexer } from './indexer';
+
+const log = debug('vulcanize:events');
+
+export class EventWatcher {
+  _ethClient: EthClient
+  _indexer: Indexer
+  _subscription: ZenObservable.Subscription | undefined
+
+  constructor (ethClient: EthClient, indexer: Indexer) {
+    assert(ethClient);
+    assert(indexer);
+
+    this._ethClient = ethClient;
+    this._indexer = indexer;
+  }
+
+  async start (): Promise<void> {
+    assert(!this._subscription, 'subscription already started');
+
+    log('Started watching upstream logs...');
+
+    this._subscription = await this._ethClient.watchLogs(async (value) => {
+      const receipt = _.get(value, 'data.listen.relatedNode');
+      log('watchLogs', JSON.stringify(receipt, null, 2));
+
+      // Check if this log is for a contract we care about.
+      const { logContracts } = receipt;
+      if (logContracts && logContracts.length) {
+        for (let logIndex = 0; logIndex < logContracts.length; logIndex++) {
+          const contractAddress = logContracts[logIndex];
+          const isWatchedContract = await this._indexer.isWatchedContract(contractAddress);
+          if (isWatchedContract) {
+            // TODO: Move processing to background task runner.
+
+            const { ethTransactionCidByTxId: { ethHeaderCidByHeaderId: { blockHash } } } = receipt;
+            await this._indexer.getEvents(blockHash, contractAddress, null);
+
+            // Trigger other indexer methods based on event topic.
+            await this._indexer.processEvent(blockHash, contractAddress, receipt, logIndex);
+          }
+        }
+      }
+    });
+  }
+
+  async stop (): Promise<void> {
+    if (this._subscription) {
+      log('Stopped watching upstream logs');
+      this._subscription.unsubscribe();
+    }
+  }
+}
--- a/packages/watcher/src/gql.ts
+++ b/packages/watcher/src/gql.ts
@ -3,11 +3,12 @@ import { makeExecutableSchema } from '@graphql-tools/schema';
 import { GraphQLSchema } from 'graphql';

 import * as typeDefs from './erc20.graphql';
+import { Indexer } from './indexer';
 import { createResolvers as createMockResolvers } from './mock/resolvers';
-import { Config, createResolvers } from './resolvers';
+import { createResolvers } from './resolvers';

-export const createSchema = async (config: Config): Promise<GraphQLSchema> => {
-  const resolvers = process.env.MOCK ? await createMockResolvers() : await createResolvers(config);
+export const createSchema = async (indexer: Indexer): Promise<GraphQLSchema> => {
+  const resolvers = process.env.MOCK ? await createMockResolvers() : await createResolvers(indexer);

  return makeExecutableSchema({
    typeDefs,
--- a/packages/watcher/src/indexer.ts
+++ b/packages/watcher/src/indexer.ts
@ -4,6 +4,7 @@ import { invert } from 'lodash';
 import { JsonFragment } from '@ethersproject/abi';
 import { DeepPartial } from 'typeorm';
 import JSONbig from 'json-bigint';
+import { ethers } from 'ethers';

 import { EthClient, getMappingSlot, topictoAddress } from '@vulcanize/ipld-eth-client';
 import { getStorageInfo, getEventNameTopics, getStorageValue, GetStorageAt, StorageLayout } from '@vulcanize/solidity-mapper';
@ -44,6 +45,7 @@ export class Indexer {

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

  constructor (db: Database, ethClient: EthClient, artifacts: Artifacts) {
    assert(db);
@ -61,6 +63,8 @@ export class Indexer {

    this._abi = abi;
    this._storageLayout = storageLayout;
+
+    this._contract = new ethers.utils.Interface(this._abi);
  }

  async totalSupply (blockHash: string, token: string): Promise<ValueResult> {
@ -151,7 +155,41 @@ export class Indexer {
    throw new Error('Not implemented.');
  }

-  async getEvents (blockHash: string, token: string, name: string): Promise<EventsResult> {
+  async processEvent (blockHash: string, token: string, receipt: any, logIndex: number): Promise<void> {
+    const topics = [];
+
+    // We only care about the event type for now.
+    const data = '0x0000000000000000000000000000000000000000000000000000000000000000';
+
+    topics.push(receipt.topic0S[logIndex]);
+    topics.push(receipt.topic1S[logIndex]);
+    topics.push(receipt.topic2S[logIndex]);
+
+    const { name: eventName, args } = this._contract.parseLog({ topics, data });
+    log(`process event ${eventName} ${args}`);
+
+    switch (eventName) {
+      case 'Transfer': {
+        const [from, to] = args;
+
+        // Update balance for sender and receiver.
+        await this.balanceOf(blockHash, token, from);
+        await this.balanceOf(blockHash, token, to);
+
+        break;
+      }
+      case 'Approval': {
+        const [owner, spender] = args;
+
+        // Update allowance.
+        await this.allowance(blockHash, token, owner, spender);
+
+        break;
+      }
+    }
+  }
+
+  async getEvents (blockHash: string, token: string, name: string | null): Promise<EventsResult> {
    const didSyncEvents = await this._db.didSyncEvents({ blockHash, token });
    if (!didSyncEvents) {
      // Fetch and save events first and make a note in the event sync progress table.
@ -205,6 +243,12 @@ export class Indexer {
    return result;
  }

+  async isWatchedContract (address : string): Promise<boolean> {
+    assert(address);
+
+    return this._db.isWatchedContract(address);
+  }
+
  // TODO: Move into base/class or framework package.
  async _getStorageValue (blockHash: string, token: string, variable: string): Promise<ValueResult> {
    return getStorageValue(
--- a/packages/watcher/src/resolvers.ts
+++ b/packages/watcher/src/resolvers.ts
@ -1,67 +1,13 @@
 import assert from 'assert';
 import BigInt from 'apollo-type-bigint';
 import debug from 'debug';
-import 'reflect-metadata';
-import { ConnectionOptions } from 'typeorm';

-import { getCache, Config as CacheConfig } from '@vulcanize/cache';
-import { EthClient } from '@vulcanize/ipld-eth-client';
-
-import artifacts from './artifacts/ERC20.json';
 import { Indexer, ValueResult } from './indexer';
-import { Database } from './database';
-
-export interface Config {
-  server: {
-    host: string;
-    port: string;
-  };
-  database: ConnectionOptions;
-  upstream: {
-    gqlEndpoint: string;
-    cache: CacheConfig
-  }
-}

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

-export const createResolvers = async (config: Config): Promise<any> => {
-  const { upstream, database } = config;
-
-  assert(database, 'Missing database config');
-
-  const ormConfig: ConnectionOptions = {
-    ...database,
-    entities: [
-      'src/entity/**/*.ts'
-    ],
-    migrations: [
-      'src/migration/**/*.ts'
-    ],
-    subscribers: [
-      'src/subscriber/**/*.ts'
-    ],
-    cli: {
-      entitiesDir: 'src/entity',
-      migrationsDir: 'src/migration',
-      subscribersDir: 'src/subscriber'
-    }
-  };
-
-  const db = new Database(ormConfig);
-  await db.init();
-
-  assert(upstream, 'Missing upstream config');
-
-  const { gqlEndpoint, cache: cacheConfig } = upstream;
-  assert(upstream, 'Missing upstream gqlEndpoint');
-
-  const cache = await getCache(cacheConfig);
-  assert(cache, 'Missing cache');
-
-  const ethClient = new EthClient({ gqlEndpoint, cache });
-
-  const indexer = new Indexer(db, ethClient, artifacts);
+export const createResolvers = async (indexer: Indexer): Promise<any> => {
+  assert(indexer);

  return {
    BigInt: new BigInt('bigInt'),
--- a/packages/watcher/src/server.ts
+++ b/packages/watcher/src/server.ts
@ -1,14 +1,19 @@
 import assert from 'assert';
+import 'reflect-metadata';
 import express, { Application, Request, Response } from 'express';
 import { graphqlHTTP } from 'express-graphql';
-import fs from 'fs-extra';
-import path from 'path';
-import toml from 'toml';
 import yargs from 'yargs';
 import { hideBin } from 'yargs/helpers';
 import debug from 'debug';
-import JSONbig from 'json-bigint';

+import { getCache } from '@vulcanize/cache';
+import { EthClient } from '@vulcanize/ipld-eth-client';
+
+import artifacts from './artifacts/ERC20.json';
+import { Indexer } from './indexer';
+import { Database } from './database';
+import { EventWatcher } from './events';
+import { getConfig } from './config';
 import { createSchema } from './gql';

 const log = debug('vulcanize:server');
@ -23,23 +28,36 @@ export const createServer = async (): Promise<Application> => {
    })
    .argv;

-  const configFile = argv.f;
-  const configFilePath = path.resolve(configFile);
-  const fileExists = await fs.pathExists(configFilePath);
-  if (!fileExists) {
-    throw new Error(`Config file not found: ${configFilePath}`);
-  }
-
-  const config = toml.parse(await fs.readFile(configFilePath, 'utf8'));
-  log('config', JSONbig.stringify(config, null, 2));
+  const config = await getConfig(argv.f);

  assert(config.server, 'Missing server config');

  const { host, port } = config.server;

-  const app: Application = express();
+  const { upstream, database: dbConfig } = config;

-  const schema = await createSchema(config);
+  assert(dbConfig, 'Missing database config');
+
+  const db = new Database(dbConfig);
+  await db.init();
+
+  assert(upstream, 'Missing upstream config');
+  const { gqlEndpoint, gqlSubscriptionEndpoint, cache: cacheConfig } = upstream;
+  assert(gqlEndpoint, 'Missing upstream gqlEndpoint');
+  assert(gqlSubscriptionEndpoint, 'Missing upstream gqlSubscriptionEndpoint');
+
+  const cache = await getCache(cacheConfig);
+
+  const ethClient = new EthClient({ gqlEndpoint, gqlSubscriptionEndpoint, cache });
+
+  const indexer = new Indexer(db, ethClient, artifacts);
+
+  const eventWatcher = new EventWatcher(ethClient, indexer);
+  await eventWatcher.start();
+
+  const schema = await createSchema(indexer);
+
+  const app: Application = express();

  app.use(
    '/graphql',
--- a/yarn.lock
+++ b/yarn.lock
@ -2,6 +2,25 @@
 # yarn lockfile v1


+"@apollo/client@^3.3.19":
+  version "3.3.19"
+  resolved "https://registry.yarnpkg.com/@apollo/client/-/client-3.3.19.tgz#f1172dc9b9d7eae04c8940b047fd3b452ef92d2c"
+  integrity sha512-vzljWLPP0GwocfBhUopzDCUwsiaNTtii1eu8qDybAXqwj4/ZhnIM46c6dNQmnVcJpAIFRIsNCOxM4OlMDySJug==
+  dependencies:
+    "@graphql-typed-document-node/core" "^3.0.0"
+    "@types/zen-observable" "^0.8.0"
+    "@wry/context" "^0.6.0"
+    "@wry/equality" "^0.4.0"
+    fast-json-stable-stringify "^2.0.0"
+    graphql-tag "^2.12.0"
+    hoist-non-react-statics "^3.3.2"
+    optimism "^0.16.0"
+    prop-types "^15.7.2"
+    symbol-observable "^2.0.0"
+    ts-invariant "^0.7.0"
+    tslib "^1.10.0"
+    zen-observable "^0.8.14"
+
 "@ardatan/aggregate-error@0.0.6":
  version "0.0.6"
  resolved "https://registry.yarnpkg.com/@ardatan/aggregate-error/-/aggregate-error-0.0.6.tgz#fe6924771ea40fc98dc7a7045c2e872dc8527609"
@ -894,6 +913,11 @@
    camel-case "4.1.2"
    tslib "~2.2.0"

+"@graphql-typed-document-node/core@^3.0.0":
+  version "3.1.0"
+  resolved "https://registry.yarnpkg.com/@graphql-typed-document-node/core/-/core-3.1.0.tgz#0eee6373e11418bfe0b5638f654df7a4ca6a3950"
+  integrity sha512-wYn6r8zVZyQJ6rQaALBEln5B1pzxb9shV5Ef97kTvn6yVGrqyXVnDqnU24MXnFubR+rZjBY9NWuxX3FB2sTsjg==
+
 "@nodelib/fs.scandir@2.1.4":
  version "2.1.4"
  resolved "https://registry.yarnpkg.com/@nodelib/fs.scandir/-/fs.scandir-2.1.4.tgz#d4b3549a5db5de2683e0c1071ab4f140904bbf69"
@ -1328,6 +1352,13 @@
    "@types/bn.js" "*"
    "@types/underscore" "*"

+"@types/ws@^7.4.4":
+  version "7.4.4"
+  resolved "https://registry.yarnpkg.com/@types/ws/-/ws-7.4.4.tgz#93e1e00824c1de2608c30e6de4303ab3b4c0c9bc"
+  integrity sha512-d/7W23JAXPodQNbOZNXvl2K+bqAQrCMwlh/nuQsPSQk6Fq0opHoPrUw43aHsvSbIiQPr8Of2hkFbnz1XBFVyZQ==
+  dependencies:
+    "@types/node" "*"
+
 "@types/yargs-parser@*":
  version "20.2.0"
  resolved "https://registry.yarnpkg.com/@types/yargs-parser/-/yargs-parser-20.2.0.tgz#dd3e6699ba3237f0348cd085e4698780204842f9"
@ -1340,7 +1371,7 @@
  dependencies:
    "@types/yargs-parser" "*"

-"@types/zen-observable@^0.8.2":
+"@types/zen-observable@^0.8.0", "@types/zen-observable@^0.8.2":
  version "0.8.2"
  resolved "https://registry.yarnpkg.com/@types/zen-observable/-/zen-observable-0.8.2.tgz#808c9fa7e4517274ed555fa158f2de4b4f468e71"
  integrity sha512-HrCIVMLjE1MOozVoD86622S7aunluLb2PJdPfb3nYiEtohm8mIB/vyv0Fd37AdeMFrTUQXEunw78YloMA3Qilg==
@ -1420,6 +1451,27 @@
  resolved "https://registry.yarnpkg.com/@ungap/promise-all-settled/-/promise-all-settled-1.1.2.tgz#aa58042711d6e3275dd37dc597e5d31e8c290a44"
  integrity sha512-sL/cEvJWAnClXw0wHk85/2L0G6Sj8UB0Ctc1TEMbKSsmpRosqhwj9gWgFRZSrBr2f9tiXISwNhCPmlfqUqyb9Q==

+"@wry/context@^0.6.0":
+  version "0.6.0"
+  resolved "https://registry.yarnpkg.com/@wry/context/-/context-0.6.0.tgz#f903eceb89d238ef7e8168ed30f4511f92d83e06"
+  integrity sha512-sAgendOXR8dM7stJw3FusRxFHF/ZinU0lffsA2YTyyIOfic86JX02qlPqPVqJNZJPAxFt+2EE8bvq6ZlS0Kf+Q==
+  dependencies:
+    tslib "^2.1.0"
+
+"@wry/equality@^0.4.0":
+  version "0.4.0"
+  resolved "https://registry.yarnpkg.com/@wry/equality/-/equality-0.4.0.tgz#474491869a8d0590f4a33fd2a4850a77a0f63408"
+  integrity sha512-DxN/uawWfhRbgYE55zVCPOoe+jvsQ4m7PT1Wlxjyb/LCCLuU1UsucV2BbCxFAX8bjcSueFBbB5Qfj1Zfe8e7Fw==
+  dependencies:
+    tslib "^2.1.0"
+
+"@wry/trie@^0.3.0":
+  version "0.3.0"
+  resolved "https://registry.yarnpkg.com/@wry/trie/-/trie-0.3.0.tgz#3245e74988c4e3033299e479a1bf004430752463"
+  integrity sha512-Yw1akIogPhAT6XPYsRHlZZIS0tIGmAl9EYXHi2scf7LPKKqdqmow/Hu4kEqP2cJR3EjaU/9L0ZlAjFf3hFxmug==
+  dependencies:
+    tslib "^2.1.0"
+
 "@yarnpkg/lockfile@^1.1.0":
  version "1.1.0"
  resolved "https://registry.yarnpkg.com/@yarnpkg/lockfile/-/lockfile-1.1.0.tgz#e77a97fbd345b76d83245edcd17d393b1b41fb31"
@ -2332,6 +2384,11 @@ babylon@^6.18.0:
  resolved "https://registry.yarnpkg.com/babylon/-/babylon-6.18.0.tgz#af2f3b88fa6f5c1e4c634d1a0f8eac4f55b395e3"
  integrity sha512-q/UEjfGJ2Cm3oKV71DJz9d25TPnq5rhBVL2Q4fA5wcC3jcrdn7+SssEybFIxwAvvP+YCsCYNKughoF33GxgycQ==

+backo2@^1.0.2:
+  version "1.0.2"
+  resolved "https://registry.yarnpkg.com/backo2/-/backo2-1.0.2.tgz#31ab1ac8b129363463e35b3ebb69f4dfcfba7947"
+  integrity sha1-MasayLEpNjRj41s+u2n038+6eUc=
+
 backoff@^2.5.0:
  version "2.5.0"
  resolved "https://registry.yarnpkg.com/backoff/-/backoff-2.5.0.tgz#f616eda9d3e4b66b8ca7fca79f695722c5f8e26f"
@ -3151,7 +3208,7 @@ cross-fetch@^2.1.0, cross-fetch@^2.1.1:
    node-fetch "2.1.2"
    whatwg-fetch "2.0.4"

-cross-fetch@^3.0.6:
+cross-fetch@^3.0.6, cross-fetch@^3.1.4:
  version "3.1.4"
  resolved "https://registry.yarnpkg.com/cross-fetch/-/cross-fetch-3.1.4.tgz#9723f3a3a247bf8b89039f3a380a9244e8fa2f39"
  integrity sha512-1eAtFWdIubi6T4XPy6ei9iUFoKpUkIF971QLN8lIvvvwueI65+Nw5haMNKUwfJxabqlIIDODJKGrQ66gxC0PbQ==
@ -4358,6 +4415,11 @@ eventemitter3@4.0.4:
  resolved "https://registry.yarnpkg.com/eventemitter3/-/eventemitter3-4.0.4.tgz#b5463ace635a083d018bdc7c917b4c5f10a85384"
  integrity sha512-rlaVLnVxtxvoyLsQQFBx53YmXHDxRIzzTLbdfxqi4yocpSjAxXwkU0cScM5JgSKMqEhrZpnvQ2D9gjylR0AimQ==

+eventemitter3@^3.1.0:
+  version "3.1.2"
+  resolved "https://registry.yarnpkg.com/eventemitter3/-/eventemitter3-3.1.2.tgz#2d3d48f9c346698fce83a85d7d664e98535df6e7"
+  integrity sha512-tvtQIeLVHjDkJYnzf2dgVMxfuSGJeM/7UCG17TT4EumTfNtF+0nebF/4zWOIkCreAbtNqhGEboB6BWrwqNaw4Q==
+
 events@^3.0.0:
  version "3.3.0"
  resolved "https://registry.yarnpkg.com/events/-/events-3.3.0.tgz#31a95ad0a924e2d2c419a813aeb2c4e878ea7400"
@ -5038,6 +5100,13 @@ graphql-request@^3.4.0:
    extract-files "^9.0.0"
    form-data "^3.0.0"

+graphql-tag@^2.12.0:
+  version "2.12.4"
+  resolved "https://registry.yarnpkg.com/graphql-tag/-/graphql-tag-2.12.4.tgz#d34066688a4f09e72d6f4663c74211e9b4b7c4bf"
+  integrity sha512-VV1U4O+9x99EkNpNmCUV5RZwq6MnK4+pGbRYWG+lA/m3uo7TSqJF81OkcOP148gFP6fzdl7JWYBrwWVTS9jXww==
+  dependencies:
+    tslib "^2.1.0"
+
 graphql@^15.5.0:
  version "15.5.0"
  resolved "https://registry.yarnpkg.com/graphql/-/graphql-15.5.0.tgz#39d19494dbe69d1ea719915b578bf920344a69d5"
@ -5243,6 +5312,13 @@ hmac-drbg@^1.0.1:
    minimalistic-assert "^1.0.0"
    minimalistic-crypto-utils "^1.0.1"

+hoist-non-react-statics@^3.3.2:
+  version "3.3.2"
+  resolved "https://registry.yarnpkg.com/hoist-non-react-statics/-/hoist-non-react-statics-3.3.2.tgz#ece0acaf71d62c2969c2ec59feff42a4b1a85b45"
+  integrity sha512-/gGivxi8JPKWNm/W0jSmzcMPpfpPLc3dY/6GxhX2hQ9iGj3aDfklV4ET7NjKpSinLpJ5vafa9iiGIEZg10SfBw==
+  dependencies:
+    react-is "^16.7.0"
+
 home-or-tmp@^2.0.0:
  version "2.0.0"
  resolved "https://registry.yarnpkg.com/home-or-tmp/-/home-or-tmp-2.0.0.tgz#e36c3f2d2cae7d746a857e38d18d5f32a7882db8"
@ -5795,6 +5871,11 @@ isurl@^1.0.0-alpha5:
    has-to-string-tag-x "^1.2.0"
    is-object "^1.0.1"

+iterall@^1.2.1:
+  version "1.3.0"
+  resolved "https://registry.yarnpkg.com/iterall/-/iterall-1.3.0.tgz#afcb08492e2915cbd8a0884eb93a8c94d0d72fea"
+  integrity sha512-QZ9qOMdF+QLHxy1QIpUHUU1D5pS2CG2P69LF6L6CPjPYA/XMOmKV3PZpawHoAjHNyB0swdVTRxdYT4tbBbxqwg==
+
 js-sha3@0.5.7, js-sha3@^0.5.7:
  version "0.5.7"
  resolved "https://registry.yarnpkg.com/js-sha3/-/js-sha3-0.5.7.tgz#0d4ffd8002d5333aabaf4a23eed2f6374c9f28e7"
@ -6438,7 +6519,7 @@ looper@^3.0.0:
  resolved "https://registry.yarnpkg.com/looper/-/looper-3.0.0.tgz#2efa54c3b1cbaba9b94aee2e5914b0be57fbb749"
  integrity sha1-LvpUw7HLq6m5Su4uWRSwvlf7t0k=

-loose-envify@^1.0.0:
+loose-envify@^1.0.0, loose-envify@^1.4.0:
  version "1.4.0"
  resolved "https://registry.yarnpkg.com/loose-envify/-/loose-envify-1.4.0.tgz#71ee51fa7be4caec1a63839f7e682d8132d30caf"
  integrity sha512-lyuxPGr/Wfhrlem2CL/UcnUc1zcqKAImBDzukY7Y5F/yQiNdko6+fRLevlw1HgMySw7f611UIY408EtxRSoK3Q==
@ -7192,6 +7273,14 @@ open@^7.4.2:
    is-docker "^2.0.0"
    is-wsl "^2.1.1"

+optimism@^0.16.0:
+  version "0.16.1"
+  resolved "https://registry.yarnpkg.com/optimism/-/optimism-0.16.1.tgz#7c8efc1f3179f18307b887e18c15c5b7133f6e7d"
+  integrity sha512-64i+Uw3otrndfq5kaoGNoY7pvOhSsjFEN4bdEFh80MWVk/dbgJfMv7VFDeCT8LxNAlEVhQmdVEbfE7X2nWNIIg==
+  dependencies:
+    "@wry/context" "^0.6.0"
+    "@wry/trie" "^0.3.0"
+
 optionator@^0.9.1:
  version "0.9.1"
  resolved "https://registry.yarnpkg.com/optionator/-/optionator-0.9.1.tgz#4f236a6373dae0566a6d43e1326674f50c291499"
@ -7696,6 +7785,15 @@ promise-to-callback@^1.0.0:
    is-fn "^1.0.0"
    set-immediate-shim "^1.0.1"

+prop-types@^15.7.2:
+  version "15.7.2"
+  resolved "https://registry.yarnpkg.com/prop-types/-/prop-types-15.7.2.tgz#52c41e75b8c87e72b9d9360e0206b99dcbffa6c5"
+  integrity sha512-8QQikdH7//R2vurIJSutZ1smHYTcLpRWEOlHnzcWHmBYrOGUysKwSsrC89BCiFj3CbrfJ/nXFdJepOVrY1GCHQ==
+  dependencies:
+    loose-envify "^1.4.0"
+    object-assign "^4.1.1"
+    react-is "^16.8.1"
+
 proxy-addr@~2.0.5:
  version "2.0.6"
  resolved "https://registry.yarnpkg.com/proxy-addr/-/proxy-addr-2.0.6.tgz#fdc2336505447d3f2f2c638ed272caf614bbb2bf"
@ -7900,6 +7998,11 @@ rc@^1.2.8:
    minimist "^1.2.0"
    strip-json-comments "~2.0.1"

+react-is@^16.7.0, react-is@^16.8.1:
+  version "16.13.1"
+  resolved "https://registry.yarnpkg.com/react-is/-/react-is-16.13.1.tgz#789729a4dc36de2999dc156dd6c1d9c18cea56a4"
+  integrity sha512-24e6ynE2H+OKt4kqsOvNd8kBpV65zoxbA4BVsEOB3ARVWQki/DHzaUoC5KuON/BiccDaCCTZBuOcfZs70kR8bQ==
+
 read-pkg-up@^1.0.1:
  version "1.0.1"
  resolved "https://registry.yarnpkg.com/read-pkg-up/-/read-pkg-up-1.0.1.tgz#9d63c13276c065918d57f002a57f40a1b643fb02"
@ -8841,6 +8944,17 @@ strip-json-comments@3.1.1, strip-json-comments@^3.1.0, strip-json-comments@^3.1.
  resolved "https://registry.yarnpkg.com/strip-json-comments/-/strip-json-comments-3.1.1.tgz#31f1281b3832630434831c310c01cccda8cbe006"
  integrity sha512-6fPc+R4ihwqP6N/aIv2f1gMH8lOVtWQHoqC4yK6oSDVVocumAsfCqjkXnqiYMhmMwS/mEHLp7Vehlt3ql6lEig==

+subscriptions-transport-ws@^0.9.18:
+  version "0.9.18"
+  resolved "https://registry.yarnpkg.com/subscriptions-transport-ws/-/subscriptions-transport-ws-0.9.18.tgz#bcf02320c911fbadb054f7f928e51c6041a37b97"
+  integrity sha512-tztzcBTNoEbuErsVQpTN2xUNN/efAZXyCyL5m3x4t6SKrEiTL2N8SaKWBFWM4u56pL79ULif3zjyeq+oV+nOaA==
+  dependencies:
+    backo2 "^1.0.2"
+    eventemitter3 "^3.1.0"
+    iterall "^1.2.1"
+    symbol-observable "^1.0.4"
+    ws "^5.2.0"
+
 supports-color@6.0.0:
  version "6.0.0"
  resolved "https://registry.yarnpkg.com/supports-color/-/supports-color-6.0.0.tgz#76cfe742cf1f41bb9b1c29ad03068c05b4c0e40a"
@ -8891,6 +9005,16 @@ swarm-js@^0.1.40:
    tar "^4.0.2"
    xhr-request "^1.0.1"

+symbol-observable@^1.0.4:
+  version "1.2.0"
+  resolved "https://registry.yarnpkg.com/symbol-observable/-/symbol-observable-1.2.0.tgz#c22688aed4eab3cdc2dfeacbb561660560a00804"
+  integrity sha512-e900nM8RRtGhlV36KGEU9k65K3mPb1WV70OdjfxlG2EAuM1noi/E/BaW/uMhL7bPEssK8QV57vN3esixjUvcXQ==
+
+symbol-observable@^2.0.0:
+  version "2.0.3"
+  resolved "https://registry.yarnpkg.com/symbol-observable/-/symbol-observable-2.0.3.tgz#5b521d3d07a43c351055fa43b8355b62d33fd16a"
+  integrity sha512-sQV7phh2WCYAn81oAkakC5qjq2Ml0g8ozqz03wOGnx9dDlG1de6yrF+0RAzSJD8fPUow3PTSMf2SAbOGxb93BA==
+
 table@^6.0.9:
  version "6.7.1"
  resolved "https://registry.yarnpkg.com/table/-/table-6.7.1.tgz#ee05592b7143831a8c94f3cee6aae4c1ccef33e2"
@ -9108,6 +9232,13 @@ ts-generator@^0.1.1:
    resolve "^1.8.1"
    ts-essentials "^1.0.0"

+ts-invariant@^0.7.0:
+  version "0.7.3"
+  resolved "https://registry.yarnpkg.com/ts-invariant/-/ts-invariant-0.7.3.tgz#13aae22a4a165393aaf5cecdee45ef4128d358b8"
+  integrity sha512-UWDDeovyUTIMWj+45g5nhnl+8oo+GhxL5leTaHn5c8FkQWfh8v66gccLd2/YzVmV5hoQUjCEjhrXnQqVDJdvKA==
+  dependencies:
+    tslib "^2.1.0"
+
 ts-node@^10.0.0:
  version "10.0.0"
  resolved "https://registry.yarnpkg.com/ts-node/-/ts-node-10.0.0.tgz#05f10b9a716b0b624129ad44f0ea05dac84ba3be"
@ -9134,7 +9265,7 @@ tsconfig-paths@^3.9.0:
    minimist "^1.2.0"
    strip-bom "^3.0.0"

-tslib@^1.8.1, tslib@^1.9.3:
+tslib@^1.10.0, tslib@^1.8.1, tslib@^1.9.3:
  version "1.14.1"
  resolved "https://registry.yarnpkg.com/tslib/-/tslib-1.14.1.tgz#cf2d38bdc34a134bcaf1091c41f6619e2f672d00"
  integrity sha512-Xni35NKzjgMrwevysHTCArtLDpPvye8zV/0E4EyYn43P7/7qvQwPh9BGkHewbMulVntbigmcT7rdX3BNo9wRJg==
@ -9956,7 +10087,7 @@ ws@7.2.3:
  resolved "https://registry.yarnpkg.com/ws/-/ws-7.2.3.tgz#a5411e1fb04d5ed0efee76d26d5c46d830c39b46"
  integrity sha512-HTDl9G9hbkNDk98naoR/cHDws7+EyYMOdL1BmjsZXRUjf7d+MficC4B7HLUPlSiho0vg+CWKrGIt/VJBd1xunQ==

-ws@7.4.6, ws@^7.2.1:
+ws@7.4.6, ws@^7.2.1, ws@^7.4.6:
  version "7.4.6"
  resolved "https://registry.yarnpkg.com/ws/-/ws-7.4.6.tgz#5654ca8ecdeee47c33a9a4bf6d28e2be2980377c"
  integrity sha512-YmhHDO4MzaDLB+M9ym/mDA5z0naX8j7SIlT8f8z+I0VtzsRbekxEutHSme7NPS2qE8StCYQNUnfWdXta/Yu85A==
@ -9970,7 +10101,7 @@ ws@^3.0.0:
    safe-buffer "~5.1.0"
    ultron "~1.1.0"

-ws@^5.1.1:
+ws@^5.1.1, ws@^5.2.0:
  version "5.2.2"
  resolved "https://registry.yarnpkg.com/ws/-/ws-5.2.2.tgz#dffef14866b8e8dc9133582514d1befaf96e980f"
  integrity sha512-jaHFD6PFv6UgoIVda6qZllptQsMlDEJkTQcybzzXDYM1XO9Y8em691FGMPmM46WGyLU4z9KMgQN+qrux/nhlHA==
@ -10208,7 +10339,7 @@ zen-observable-ts@^1.0.0:
    "@types/zen-observable" "^0.8.2"
    zen-observable "^0.8.15"

-zen-observable@^0.8.15:
+zen-observable@^0.8.14, zen-observable@^0.8.15:
  version "0.8.15"
  resolved "https://registry.yarnpkg.com/zen-observable/-/zen-observable-0.8.15.tgz#96415c512d8e3ffd920afd3889604e30b9eaac15"
  integrity sha512-PQ2PC7R9rslx84ndNBZB/Dkv8V8fZEpk83RLgXtYd0fwUgEjseMn1Dgajh2x6S8QbZAFa9p2qVCEuYZNgve0dQ==