Add support for meta query in watcher GQL API

packages/cli/src/job-runner.ts

@@ -115,6 +115,7 @@ export class JobRunnerCmd {
     // Delete all active and pending (before completed) jobs to start job-runner without old queued jobs
     await jobRunner.jobQueue.deleteAllJobs('completed');
     await jobRunner.resetToPrevIndexedBlock();
+    await indexer.updateSyncStatusIndexingError(false);
 
     await startJobRunner(jobRunner);
     jobRunner.handleShutdown();

packages/codegen/src/data/entities/SyncStatus.yaml

@@ -49,6 +49,13 @@ columns:
     pgType: integer
     tsType: number
     columnType: Column
+  - name: hasIndexingError
+    pgType: boolean
+    tsType: boolean
+    columnType: Column
+    columnOptions:
+      - option: default
+        value: false
 imports:
   - toImport:
       - Entity

packages/codegen/src/schema.ts

@@ -3,7 +3,7 @@
 //
 
 import assert from 'assert';
-import { GraphQLSchema, parse, printSchema, print, GraphQLDirective, GraphQLInt, GraphQLBoolean, GraphQLEnumType, DefinitionNode } from 'graphql';
+import { GraphQLSchema, parse, printSchema, print, GraphQLDirective, GraphQLInt, GraphQLBoolean, GraphQLEnumType, DefinitionNode, GraphQLString, GraphQLNonNull } from 'graphql';
 import { ObjectTypeComposer, NonNullComposer, ObjectTypeComposerDefinition, ObjectTypeComposerFieldConfigMapDefinition, SchemaComposer } from 'graphql-compose';
 import { Writable } from 'stream';
 import { utils } from 'ethers';
@@ -98,13 +98,16 @@ export class Schema {
     // Add a mutation for watching a contract.
     this._addWatchContractMutation();
 
-    // Add type and query for SyncStatus.
-    this._addSyncStatus();
-
     // Add State type and queries.
     this._addStateType();
     this._addStateQuery();
 
+    // Add type and query for SyncStatus.
+    this._addSyncStatus();
+
+    // Add type and query for meta data
+    this._addMeta();
+
     // Build the schema.
     return this._composer.buildSchema();
   }
@@ -269,7 +272,7 @@ export class Schema {
     typeComposer = this._composer.createObjectTC({
       name: '_Block_',
       fields: {
-        cid: 'String!',
+        cid: 'String',
         hash: 'String!',
         number: 'Int!',
         timestamp: 'Int!',
@@ -456,6 +459,28 @@ export class Schema {
     });
   }
 
+  _addMeta (): void {
+    const typeComposer = this._composer.createObjectTC({
+      name: '_Meta_',
+      fields: {
+        block: this._composer.getOTC('_Block_').NonNull,
+        deployment: { type: new GraphQLNonNull(GraphQLString) },
+        hasIndexingErrors: { type: new GraphQLNonNull(GraphQLBoolean) }
+      }
+    });
+
+    this._composer.addSchemaMustHaveType(typeComposer);
+
+    this._composer.Query.addFields({
+      _meta: {
+        type: this._composer.getOTC('_Meta_'),
+        args: {
+          block: BlockHeight
+        }
+      }
+    });
+  }
+
   _addStateType (): void {
     const typeComposer = this._composer.createObjectTC({
       name: 'ResultState',

packages/codegen/src/templates/database-template.handlebars

@@ -259,6 +259,12 @@ export class Database implements DatabaseInterface {
     return this._baseDatabase.forceUpdateSyncStatus(repo, blockHash, blockNumber);
   }
 
+  async updateSyncStatusIndexingError (queryRunner: QueryRunner, hasIndexingError: boolean): Promise<SyncStatus> {
+    const repo = queryRunner.manager.getRepository(SyncStatus);
+
+    return this._baseDatabase.updateSyncStatusIndexingError(repo, hasIndexingError);
+  }
+
   async getSyncStatus (queryRunner: QueryRunner): Promise<SyncStatus | undefined> {
     const repo = queryRunner.manager.getRepository(SyncStatus);
 

packages/codegen/src/templates/indexer-template.handlebars

@@ -43,7 +43,8 @@ import {
   DatabaseInterface,
   Clients,
   EthClient,
-  UpstreamConfig
+  UpstreamConfig,
+  ResultMeta
 } from '@cerc-io/util';
 {{#if (subgraphPath)}}
 import { GraphWatcher } from '@cerc-io/graph-node';
@@ -197,6 +198,10 @@ export class Indexer implements IndexerInterface {
     await this._baseIndexer.fetchStateStatus();
   }
 
+  async getMetaData (block: BlockHeight): Promise<ResultMeta | null> {
+    return this._baseIndexer.getMetaData(block);
+  }
+
   getResultEvent (event: Event): ResultEvent {
     return getResultEvent(event);
   }
@@ -660,6 +665,10 @@ export class Indexer implements IndexerInterface {
     return this._baseIndexer.forceUpdateSyncStatus(blockHash, blockNumber);
   }
 
+  async updateSyncStatusIndexingError (hasIndexingError: boolean): Promise<SyncStatus> {
+    return this._baseIndexer.updateSyncStatusIndexingError(hasIndexingError);
+  }
+
   async getEvent (id: string): Promise<Event | undefined> {
     return this._baseIndexer.getEvent(id);
   }

packages/codegen/src/templates/resolvers-template.handlebars

@@ -168,6 +168,17 @@ export const createResolvers = async (indexerArg: IndexerInterface, eventWatcher
         gqlQueryCount.labels('getSyncStatus').inc(1);
 
         return indexer.getSyncStatus();
+      },
+
+      _meta: async (
+        _: any,
+        { block = {} }: { block: BlockHeight }
+      ) => {
+        log('_meta');
+        gqlTotalQueryCount.inc(1);
+        gqlQueryCount.labels('_meta').inc(1);
+
+        return indexer.getMetaData(block);
       }
     }
   };

packages/graph-node/test/utils/indexer.ts

@@ -177,6 +177,12 @@ export class Indexer implements IndexerInterface {
     return {} as SyncStatusInterface;
   }
 
+  async updateSyncStatusIndexingError (hasIndexingError: boolean): Promise<SyncStatusInterface> {
+    assert(hasIndexingError);
+
+    return {} as SyncStatusInterface;
+  }
+
   async markBlocksAsPruned (blocks: BlockProgressInterface[]): Promise<void> {
     assert(blocks);
 

packages/util/src/database.ts

@@ -202,6 +202,15 @@ export class Database {
     return await repo.save(entity);
   }
 
+  async updateSyncStatusIndexingError (repo: Repository<SyncStatusInterface>, hasIndexingError: boolean): Promise<SyncStatusInterface> {
+    const entity = await repo.findOne();
+    assert(entity);
+
+    entity.hasIndexingError = hasIndexingError;
+
+    return repo.save(entity);
+  }
+
   async getBlockProgress (repo: Repository<BlockProgressInterface>, blockHash: string): Promise<BlockProgressInterface | undefined> {
     return repo.findOne({ where: { blockHash } });
   }
@@ -1099,7 +1108,6 @@ export class Database {
     eventCount.set(res);
   }
 
-  // TODO: Transform in the GQL type BigInt parsing itself
   _transformBigValues (value: any): any {
     // Handle array of bigints
     if (Array.isArray(value)) {

packages/util/src/events.ts

@@ -9,8 +9,8 @@ import PgBoss from 'pg-boss';
 import { constants } from 'ethers';
 
 import { JobQueue } from './job-queue';
-import { BlockProgressInterface, EventInterface, IndexerInterface, EthClient } from './types';
-import { MAX_REORG_DEPTH, JOB_KIND_PRUNE, JOB_KIND_INDEX, UNKNOWN_EVENT_NAME, JOB_KIND_EVENTS, QUEUE_BLOCK_PROCESSING, QUEUE_EVENT_PROCESSING, QUEUE_HISTORICAL_PROCESSING } from './constants';
+import { BlockProgressInterface, EventInterface, IndexerInterface, EthClient, EventsJobData, EventsQueueJobKind } from './types';
+import { MAX_REORG_DEPTH, JOB_KIND_PRUNE, JOB_KIND_INDEX, UNKNOWN_EVENT_NAME, QUEUE_BLOCK_PROCESSING, QUEUE_EVENT_PROCESSING, QUEUE_HISTORICAL_PROCESSING } from './constants';
 import { createPruningJob, processBlockByNumber } from './common';
 import { OrderDirection } from './database';
 import { HistoricalJobData, HistoricalJobResponseData } from './job-runner';
@@ -258,27 +258,26 @@ export class EventWatcher {
     );
   }
 
-  async eventProcessingCompleteHandler (job: any): Promise<void> {
-    const { id, data: { request, failed, state, createdOn } } = job;
+  async eventProcessingCompleteHandler (job: PgBoss.Job<any>): Promise<void> {
+    const { id, data: { request: { data }, failed, state, createdOn } } = job;
 
     if (failed) {
       log(`Job ${id} for queue ${QUEUE_EVENT_PROCESSING} failed`);
       return;
     }
 
-    const { data: { kind, blockHash, publish } } = request;
-
-    // Ignore jobs other than JOB_KIND_EVENTS
-    if (kind !== JOB_KIND_EVENTS) {
+    // Ignore jobs other than event processsing
+    const { kind } = data;
+    if (kind !== EventsQueueJobKind.EVENTS) {
       return;
     }
 
+    const { blockHash, publish }: EventsJobData = data;
+
     // Check if publish is set to true
     // Events and blocks are not published in historical processing
     // GQL subscription events will not be triggered if publish is set to false
     if (publish) {
-      assert(blockHash);
-
       const blockProgress = await this._indexer.getBlockProgress(blockHash);
       assert(blockProgress);
 
@@ -303,7 +302,7 @@ export class EventWatcher {
       // TODO: Use a different pubsub to publish event from job-runner.
       // https://www.apollographql.com/docs/apollo-server/data/subscriptions/#production-pubsub-libraries
       for (const dbEvent of dbEvents) {
-        log(`Job onComplete event ${dbEvent.id} publish ${!!request.data.publish}`);
+        log(`Job onComplete event ${dbEvent.id} publish ${publish}`);
 
         if (!failed && state === 'completed') {
           // Check for max acceptable lag time between request and sending results to live subscribers.

packages/util/src/indexer.ts

@@ -22,11 +22,13 @@ import {
   SyncStatusInterface,
   StateInterface,
   StateKind,
-  EthClient
+  EthClient,
+  ContractJobData,
+  EventsQueueJobKind
 } from './types';
-import { UNKNOWN_EVENT_NAME, JOB_KIND_CONTRACT, QUEUE_EVENT_PROCESSING, DIFF_MERGE_BATCH_SIZE } from './constants';
+import { UNKNOWN_EVENT_NAME, QUEUE_EVENT_PROCESSING, DIFF_MERGE_BATCH_SIZE } from './constants';
 import { JobQueue } from './job-queue';
-import { Where, QueryOptions } from './database';
+import { Where, QueryOptions, BlockHeight } from './database';
 import { ServerConfig, UpstreamConfig } from './config';
 import { createOrUpdateStateData, StateDataMeta } from './state-helper';
 
@@ -88,6 +90,18 @@ export type ResultEvent = {
   proof: string;
 };
 
+export type ResultMeta = {
+  block: {
+    cid: string | null;
+    hash: string;
+    number: number;
+    timestamp: number;
+    parentHash: string;
+  };
+  deployment: string;
+  hasIndexingErrors: boolean;
+};
+
 export class Indexer {
   _serverConfig: ServerConfig;
   _upstreamConfig: UpstreamConfig;
@@ -131,6 +145,40 @@ export class Indexer {
     }, {});
   }
 
+  async getMetaData (block: BlockHeight): Promise<ResultMeta | null> {
+    let resultBlock: BlockProgressInterface | undefined;
+
+    const syncStatus = await this.getSyncStatus();
+    assert(syncStatus);
+
+    if (block.hash) {
+      resultBlock = await this.getBlockProgress(block.hash);
+    } else {
+      const blockHeight = block.number ? block.number : syncStatus.latestIndexedBlockNumber - 1;
+
+      // Get all the blocks at a height
+      const blocksAtHeight = await this.getBlocksAtHeight(blockHeight, false);
+
+      if (blocksAtHeight.length) {
+        resultBlock = blocksAtHeight[0];
+      }
+    }
+
+    return resultBlock
+      ? {
+        block: {
+          cid: resultBlock.cid,
+          number: resultBlock.blockNumber,
+          hash: resultBlock.blockHash,
+          timestamp: resultBlock.blockTimestamp,
+          parentHash: resultBlock.parentHash
+        },
+        deployment: '',
+        hasIndexingErrors: syncStatus.hasIndexingError
+      }
+      : null;
+  }
+
   async getSyncStatus (): Promise<SyncStatusInterface | undefined> {
     const dbTx = await this._db.createTransactionRunner();
     let res;
@@ -216,6 +264,23 @@ export class Indexer {
     return res;
   }
 
+  async updateSyncStatusIndexingError (hasIndexingError: boolean): Promise<SyncStatusInterface> {
+    const dbTx = await this._db.createTransactionRunner();
+    let res;
+
+    try {
+      res = await this._db.updateSyncStatusIndexingError(dbTx, hasIndexingError);
+      await dbTx.commitTransaction();
+    } catch (error) {
+      await dbTx.rollbackTransaction();
+      throw error;
+    } finally {
+      await dbTx.release();
+    }
+
+    return res;
+  }
+
   async getBlocks (blockFilter: { blockNumber?: number, blockHash?: string }): Promise<any> {
     assert(blockFilter.blockHash || blockFilter.blockNumber);
     const result = await this._ethClient.getBlocks(blockFilter);
@@ -760,12 +825,10 @@ export class Indexer {
       this.cacheContract(contract);
       await dbTx.commitTransaction();
 
+      const contractJob: ContractJobData = { kind: EventsQueueJobKind.CONTRACT, contract };
       await this._jobQueue.pushJob(
         QUEUE_EVENT_PROCESSING,
-        {
-          kind: JOB_KIND_CONTRACT,
-          contract
-        },
+        contractJob,
         { priority: 1 }
       );
     } catch (error) {