Invoke subgraph handler in watcher event processing (#34)

* Invoke subgraph handler in watcher event processing

* Fix error when invoking subgraph handler

* Parse events using event signature specified in subgraph yaml

* Use contract abi to parse event params

* Invoke event handler based on event signature

* Fill event with block and transaction data

* Comment missing fields in block and transaction data

packages/graph-test-watcher/environments/local.toml

@@ -19,6 +19,7 @@
     gqlApiEndpoint = "http://127.0.0.1:8082/graphql"
     gqlPostgraphileEndpoint = "http://127.0.0.1:5000/graphql"
     rpcProviderEndpoint = "http://127.0.0.1:8081"
+    blockDelayInMilliSecs = 2000
 
   [upstream.cache]
     name = "requests"

packages/graph-test-watcher/src/events.ts

@@ -12,7 +12,8 @@ import {
   EventWatcher as BaseEventWatcher,
   QUEUE_BLOCK_PROCESSING,
   QUEUE_EVENT_PROCESSING,
-  UNKNOWN_EVENT_NAME
+  UNKNOWN_EVENT_NAME,
+  UpstreamConfig
 } from '@vulcanize/util';
 
 import { Indexer } from './indexer';
@@ -30,7 +31,7 @@ export class EventWatcher {
   _pubsub: PubSub
   _jobQueue: JobQueue
 
-  constructor (ethClient: EthClient, indexer: Indexer, pubsub: PubSub, jobQueue: JobQueue) {
+  constructor (upstreamConfig: UpstreamConfig, ethClient: EthClient, postgraphileClient: EthClient, indexer: Indexer, pubsub: PubSub, jobQueue: JobQueue) {
     assert(ethClient);
     assert(indexer);
 
@@ -38,7 +39,7 @@ export class EventWatcher {
     this._indexer = indexer;
     this._pubsub = pubsub;
     this._jobQueue = jobQueue;
-    this._baseEventWatcher = new BaseEventWatcher(this._ethClient, this._indexer, this._pubsub, this._jobQueue);
+    this._baseEventWatcher = new BaseEventWatcher(upstreamConfig, this._ethClient, postgraphileClient, this._indexer, this._pubsub, this._jobQueue);
   }
 
   getEventIterator (): AsyncIterator<any> {
@@ -52,22 +53,15 @@ export class EventWatcher {
   async start (): Promise<void> {
     assert(!this._subscription, 'subscription already started');
 
-    await this.watchBlocksAtChainHead();
     await this.initBlockProcessingOnCompleteHandler();
     await this.initEventProcessingOnCompleteHandler();
+    this._baseEventWatcher.startBlockProcessing();
   }
 
   async stop (): Promise<void> {
     this._baseEventWatcher.stop();
   }
 
-  async watchBlocksAtChainHead (): Promise<void> {
-    log('Started watching upstream blocks...');
-    this._subscription = await this._ethClient.watchBlocks(async (value) => {
-      await this._baseEventWatcher.blocksHandler(value);
-    });
-  }
-
   async initBlockProcessingOnCompleteHandler (): Promise<void> {
     this._jobQueue.onComplete(QUEUE_BLOCK_PROCESSING, async (job) => {
       const { id, data: { failed } } = job;

packages/graph-test-watcher/src/fill.ts

@@ -54,13 +54,13 @@ export const main = async (): Promise<any> => {
   await db.init();
 
   assert(upstream, 'Missing upstream config');
-  const { ethServer: { gqlPostgraphileEndpoint, rpcProviderEndpoint }, cache: cacheConfig } = upstream;
+  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: gqlPostgraphileEndpoint,
+    gqlEndpoint: gqlApiEndpoint,
     gqlSubscriptionEndpoint: gqlPostgraphileEndpoint,
     cache
   });
@@ -83,11 +83,11 @@ export const main = async (): Promise<any> => {
   const jobQueue = new JobQueue({ dbConnectionString, maxCompletionLag: maxCompletionLagInSecs });
   await jobQueue.start();
 
-  const eventWatcher = new EventWatcher(ethClient, indexer, pubsub, jobQueue);
+  const eventWatcher = new EventWatcher(upstream, ethClient, postgraphileClient, indexer, pubsub, jobQueue);
 
   assert(jobQueueConfig, 'Missing job queue config');
 
-  await fillBlocks(jobQueue, indexer, ethClient, eventWatcher, argv);
+  await fillBlocks(jobQueue, indexer, postgraphileClient, eventWatcher, blockDelayInMilliSecs, argv);
 };
 
 main().catch(err => {

packages/graph-test-watcher/src/indexer.ts

@@ -44,6 +44,7 @@ export type ResultEvent = {
   contract: string;
 
   eventIndex: number;
+  eventSignature: string;
   event: any;
 
   proof: string;
@@ -67,8 +68,8 @@ export class Indexer {
 
     this._db = db;
     this._ethClient = ethClient;
-    this._ethProvider = ethProvider;
     this._postgraphileClient = postgraphileClient;
+    this._ethProvider = ethProvider;
     this._graphWatcher = graphWatcher;
     this._baseIndexer = new BaseIndexer(this._db, this._ethClient, this._ethProvider);
 
@@ -86,7 +87,7 @@ export class Indexer {
   getResultEvent (event: Event): ResultEvent {
     const block = event.block;
     const eventFields = JSONbig.parse(event.eventInfo);
-    const { tx } = JSON.parse(event.extraInfo);
+    const { tx, eventSignature } = JSON.parse(event.extraInfo);
 
     return {
       block: {
@@ -106,6 +107,7 @@ export class Indexer {
       contract: event.contract,
 
       eventIndex: event.index,
+      eventSignature,
       event: {
         __typename: `${event.eventName}Event`,
         ...eventFields
@@ -168,7 +170,8 @@ export class Indexer {
   async triggerIndexingOnEvent (event: Event): Promise<void> {
     const resultEvent = this.getResultEvent(event);
 
-    this._graphWatcher.handleEvent(resultEvent);
+    // Call subgraph handler for event.
+    await this._graphWatcher.handleEvent(resultEvent);
 
     // Call custom hook function for indexing on event.
     await handleEvent(this, resultEvent);
@@ -199,7 +202,11 @@ export class Indexer {
       }
     }
 
-    return { eventName, eventInfo };
+    return {
+      eventName,
+      eventInfo,
+      eventSignature: logDescription.signature
+    };
   }
 
   async watchContract (address: string, startingBlock: number): Promise<boolean> {
@@ -326,7 +333,7 @@ export class Indexer {
         let eventName = UNKNOWN_EVENT_NAME;
         let eventInfo = {};
         const tx = transactionMap[txHash];
-        const extraInfo = { topics, data, tx };
+        const extraInfo: { [key: string]: any } = { topics, data, tx };
 
         const contract = ethers.utils.getAddress(address);
         const watchedContract = await this.isWatchedContract(contract);
@@ -335,6 +342,7 @@ export class Indexer {
           const eventDetails = this.parseEventNameAndArgs(watchedContract.kind, logObj);
           eventName = eventDetails.eventName;
           eventInfo = eventDetails.eventInfo;
+          extraInfo.eventSignature = eventDetails.eventSignature;
         }
 
         dbEvents.push({

packages/graph-test-watcher/src/job-runner.ts

@@ -34,9 +34,9 @@ export class JobRunner {
   _jobQueueConfig: JobQueueConfig
 
   constructor (jobQueueConfig: JobQueueConfig, indexer: Indexer, jobQueue: JobQueue) {
+    this._jobQueueConfig = jobQueueConfig;
     this._indexer = indexer;
     this._jobQueue = jobQueue;
-    this._jobQueueConfig = jobQueueConfig;
     this._baseJobRunner = new BaseJobRunner(this._jobQueueConfig, this._indexer, this._jobQueue);
   }
 

packages/graph-test-watcher/src/schema.gql

@@ -10,6 +10,7 @@ type ResultEvent {
   tx: Transaction!
   contract: String!
   eventIndex: Int!
+  eventSignature: String!
   event: Event!
   proof: Proof
 }

packages/graph-test-watcher/src/server.ts

@@ -85,7 +85,7 @@ export const main = async (): Promise<any> => {
 
   const jobQueue = new JobQueue({ dbConnectionString, maxCompletionLag: maxCompletionLagInSecs });
 
-  const eventWatcher = new EventWatcher(ethClient, indexer, pubsub, jobQueue);
+  const eventWatcher = new EventWatcher(upstream, ethClient, postgraphileClient, indexer, pubsub, jobQueue);
 
   if (watcherKind === KIND_ACTIVE) {
     await jobQueue.start();