Refactor util code to be reused (#241)

* Ignore watch contract jobs in event processing complete handler

* Update job-queue config and handle errors on job completion hook

* Update graph decimal implementation

* Return generic type from method to read watcher config

* Export fill prefetch batch size default value

packages/util/src/config.ts

@@ -73,14 +73,6 @@ export interface UpstreamConfig {
     rpcProviderEndpoint: string;
   }
   traceProviderEndpoint: string;
-  uniWatcher: {
-    gqlEndpoint: string;
-    gqlSubscriptionEndpoint: string;
-  };
-  tokenWatcher: {
-    gqlEndpoint: string;
-    gqlSubscriptionEndpoint: string;
-  }
 }
 
 export interface GQLMetricsConfig {
@@ -101,7 +93,7 @@ export interface Config {
   metrics: MetricsConfig,
 }
 
-export const getConfig = async (configFile: string): Promise<Config> => {
+export const getConfig = async<ConfigType> (configFile: string): Promise<ConfigType> => {
   const configFilePath = path.resolve(configFile);
   const fileExists = await fs.pathExists(configFilePath);
   if (!fileExists) {

packages/util/src/constants.ts

@@ -24,4 +24,6 @@ export const UNKNOWN_EVENT_NAME = '__unknown__';
 export const KIND_ACTIVE = 'active';
 export const KIND_LAZY = 'lazy';
 
+export const DEFAULT_PREFETCH_BATCH_SIZE = 10;
+
 export const DEFAULT_MAX_GQL_CACHE_SIZE = Math.pow(2, 20) * 8; // 8 MB

packages/util/src/events.ts

@@ -10,7 +10,7 @@ import { EthClient } from '@cerc-io/ipld-eth-client';
 
 import { JobQueue } from './job-queue';
 import { BlockProgressInterface, EventInterface, IndexerInterface } from './types';
-import { MAX_REORG_DEPTH, JOB_KIND_PRUNE, JOB_KIND_INDEX, UNKNOWN_EVENT_NAME } from './constants';
+import { MAX_REORG_DEPTH, JOB_KIND_PRUNE, JOB_KIND_INDEX, UNKNOWN_EVENT_NAME, JOB_KIND_EVENTS } from './constants';
 import { createPruningJob, processBlockByNumberWithCache } from './common';
 import { UpstreamConfig } from './config';
 import { OrderDirection } from './database';
@@ -98,7 +98,12 @@ export class EventWatcher {
   }
 
   async eventProcessingCompleteHandler (job: any): Promise<EventInterface[]> {
-    const { data: { request: { data: { blockHash } } } } = job;
+    const { data: { request: { data: { kind, blockHash } } } } = job;
+
+    // Ignore jobs other than JOB_KIND_EVENTS
+    if (kind !== JOB_KIND_EVENTS) {
+      return [];
+    }
     assert(blockHash);
 
     const blockProgress = await this._indexer.getBlockProgress(blockHash);

packages/util/src/fill.ts

@@ -8,11 +8,10 @@ import { JobQueue } from './job-queue';
 import { EventWatcherInterface, IndexerInterface } from './types';
 import { wait } from './misc';
 import { processBlockByNumberWithCache } from './common';
+import { DEFAULT_PREFETCH_BATCH_SIZE } from './constants';
 
 const log = debug('vulcanize:fill');
 
-const DEFAULT_PREFETCH_BATCH_SIZE = 10;
-
 export const fillBlocks = async (
   jobQueue: JobQueue,
   indexer: IndexerInterface,

packages/util/src/graph-decimal.ts

@@ -29,6 +29,11 @@ export class GraphDecimal {
     return this.value.toString();
   }
 
+  toJSON (): string {
+    // Using fixed-point notation in preparing entity state.
+    return this.toFixed();
+  }
+
   toFixed (): string {
     this._checkOutOfRange(this);
 
@@ -91,6 +96,13 @@ export class GraphDecimal {
     return new GraphDecimal(this.value.div(param));
   }
 
+  pow (n: Decimal.Value | GraphDecimal): GraphDecimal {
+    this._checkOutOfRange(this);
+    const param = this._checkOutOfRange(n);
+
+    return new GraphDecimal(this.value.pow(param));
+  }
+
   isZero (): boolean {
     this._checkOutOfRange(this);
 
@@ -115,14 +127,14 @@ export class GraphDecimal {
     this._checkOutOfRange(this);
     const param = this._checkOutOfRange(n);
 
-    return this.value.lessThan(param);
+    return this.value.greaterThan(param);
   }
 
   gt (n: Decimal.Value | GraphDecimal): boolean {
     this._checkOutOfRange(this);
     const param = this._checkOutOfRange(n);
 
-    return this.value.lessThan(param);
+    return this.value.gt(param);
   }
 
   comparedTo (n: Decimal.Value | GraphDecimal): number {

packages/util/src/indexer.ts

@@ -3,7 +3,7 @@
 //
 
 import assert from 'assert';
-import { DeepPartial, EntityTarget, FindConditions, FindManyOptions, LessThanOrEqual, MoreThan } from 'typeorm';
+import { DeepPartial, EntityTarget, FindConditions, FindManyOptions, MoreThan } from 'typeorm';
 import debug from 'debug';
 import JSONbig from 'json-bigint';
 import { ethers } from 'ethers';

packages/util/src/job-queue.ts

@@ -26,7 +26,7 @@ export class JobQueue {
   constructor (config: Config) {
     this._config = config;
     this._boss = new PgBoss({
-      // https://github.com/timgit/pg-boss/blob/master/docs/configuration.md
+      // https://github.com/timgit/pg-boss/blob/6.1.0/docs/configuration.md
 
       connectionString: this._config.dbConnectionString,
       onComplete: true,
@@ -37,9 +37,11 @@ export class JobQueue {
       retryBackoff: true,
 
       // Time before active job fails by expiration.
-      expireInHours: 24 * 7, // 7 days
+      expireInHours: 24 * 1, // 1 day
 
-      retentionDays: 30, // 30 days
+      retentionDays: 1, // 1 day
+
+      deleteAfterHours: 1, // 1 hour
 
       newJobCheckInterval: 100,
 
@@ -106,11 +108,24 @@ export class JobQueue {
   }
 
   async onComplete (queue: string, callback: JobCallback): Promise<string> {
-    return await this._boss.onComplete(queue, { teamSize: JOBS_PER_INTERVAL, teamConcurrency: 1 }, async (job: any) => {
-      const { id, data: { failed, createdOn } } = job;
-      log(`Job onComplete for queue ${queue} job ${id} created ${createdOn} success ${!failed}`);
-      await callback(job);
-    });
+    return await this._boss.onComplete(
+      queue,
+      {
+        teamSize: JOBS_PER_INTERVAL,
+        teamConcurrency: 1
+      },
+      async (job: any) => {
+        try {
+          const { id, data: { failed, createdOn } } = job;
+          log(`Job onComplete for queue ${queue} job ${id} created ${createdOn} success ${!failed}`);
+          await callback(job);
+        } catch (error) {
+          log(`Error in onComplete handler for ${queue} job ${job.id}`);
+          log(error);
+          throw error;
+        }
+      }
+    );
   }
 
   async markComplete (job: any): Promise<void> {