Purge unknown events after block processing complete (#280)

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

@@ -3,7 +3,7 @@
 //
 
 import assert from 'assert';
-import { Connection, ConnectionOptions, DeepPartial, FindConditions, QueryRunner } from 'typeorm';
+import { Connection, ConnectionOptions, DeepPartial, FindConditions, QueryRunner, FindManyOptions } from 'typeorm';
 import path from 'path';
 
 import { Database as BaseDatabase } from '@vulcanize/util';
@@ -179,7 +179,7 @@ export class Database {
     return this._baseDatabase.updateBlockProgress(repo, blockHash, lastProcessedEventIndex);
   }
 
-  async removeEntities<Entity> (queryRunner: QueryRunner, entity: new () => Entity, findConditions?: FindConditions<Entity>): Promise<void> {
+  async removeEntities<Entity> (queryRunner: QueryRunner, entity: new () => Entity, findConditions?: FindManyOptions<Entity> | FindConditions<Entity>): Promise<void> {
     return this._baseDatabase.removeEntities(queryRunner, entity, findConditions);
   }
 

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

@@ -91,8 +91,8 @@ export const main = async (): Promise<any> => {
   await fillBlocks(jobQueue, indexer, ethClient, eventWatcher, argv);
 };
 
-main().then(() => {
+main().catch(err => {
-  process.exit();
-}).catch(err => {
   log(err);
+}).finally(() => {
+  process.exit();
 });

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

@@ -275,6 +275,10 @@ export class Indexer {
     return this._baseIndexer.getBlockEvents(blockHash);
   }
 
+  async removeUnknownEvents (block: BlockProgress): Promise<void> {
+    return this._baseIndexer.removeUnknownEvents(Event, block);
+  }
+
   async markBlocksAsPruned (blocks: BlockProgress[]): Promise<void> {
     return this._baseIndexer.markBlocksAsPruned(blocks);
   }

packages/erc20-watcher/src/database.ts

@@ -3,7 +3,7 @@
 //
 
 import assert from 'assert';
-import { Connection, ConnectionOptions, DeepPartial, FindConditions, QueryRunner } from 'typeorm';
+import { Connection, ConnectionOptions, DeepPartial, FindConditions, FindManyOptions, QueryRunner } from 'typeorm';
 import path from 'path';
 
 import { Database as BaseDatabase } from '@vulcanize/util';
@@ -177,7 +177,7 @@ export class Database {
     return this._baseDatabase.updateBlockProgress(repo, blockHash, lastProcessedEventIndex);
   }
 
-  async removeEntities<Entity> (queryRunner: QueryRunner, entity: new () => Entity, findConditions?: FindConditions<Entity>): Promise<void> {
+  async removeEntities<Entity> (queryRunner: QueryRunner, entity: new () => Entity, findConditions?: FindManyOptions<Entity> | FindConditions<Entity>): Promise<void> {
     return this._baseDatabase.removeEntities(queryRunner, entity, findConditions);
   }
 

packages/erc20-watcher/src/fill.ts

@@ -88,8 +88,8 @@ export const main = async (): Promise<any> => {
   await fillBlocks(jobQueue, indexer, ethClient, eventWatcher, argv);
 };
 
-main().then(() => {
+main().catch(err => {
-  process.exit();
-}).catch(err => {
   log(err);
+}).finally(() => {
+  process.exit();
 });

packages/erc20-watcher/src/indexer.ts

@@ -357,6 +357,10 @@ export class Indexer {
     return this._baseIndexer.getBlockEvents(blockHash);
   }
 
+  async removeUnknownEvents (block: BlockProgress): Promise<void> {
+    return this._baseIndexer.removeUnknownEvents(Event, block);
+  }
+
   async markBlocksAsPruned (blocks: BlockProgress[]): Promise<void> {
     return this._baseIndexer.markBlocksAsPruned(blocks);
   }

packages/uni-info-watcher/src/database.ts

@@ -8,6 +8,7 @@ import {
   ConnectionOptions,
   DeepPartial,
   FindConditions,
+  FindManyOptions,
   FindOneOptions,
   LessThanOrEqual,
   QueryRunner
@@ -626,7 +627,7 @@ export class Database implements DatabaseInterface {
     return this._baseDatabase.getEntities(queryRunner, entity, findConditions);
   }
 
-  async removeEntities<Entity> (queryRunner: QueryRunner, entity: new () => Entity, findConditions?: FindConditions<Entity>): Promise<void> {
+  async removeEntities<Entity> (queryRunner: QueryRunner, entity: new () => Entity, findConditions?: FindManyOptions<Entity> | FindConditions<Entity>): Promise<void> {
     return this._baseDatabase.removeEntities(queryRunner, entity, findConditions);
   }
 

packages/uni-info-watcher/src/fill.ts

@@ -89,8 +89,8 @@ export const main = async (): Promise<any> => {
   await fillBlocks(jobQueue, indexer, ethClient, eventWatcher, argv);
 };
 
-main().then(() => {
+main().catch(err => {
-  process.exit();
-}).catch(err => {
   log(err);
+}).finally(() => {
+  process.exit();
 });

packages/uni-info-watcher/src/indexer.ts

@@ -308,6 +308,10 @@ export class Indexer implements IndexerInterface {
     return this._baseIndexer.getBlockEvents(blockHash);
   }
 
+  async removeUnknownEvents (block: BlockProgress): Promise<void> {
+    return this._baseIndexer.removeUnknownEvents(Event, block);
+  }
+
   async updateSyncStatusIndexedBlock (blockHash: string, blockNumber: number, force = false): Promise<SyncStatus> {
     return this._baseIndexer.updateSyncStatusIndexedBlock(blockHash, blockNumber, force);
   }

packages/uni-watcher/src/database.ts

@@ -3,7 +3,7 @@
 //
 
 import assert from 'assert';
-import { Connection, ConnectionOptions, DeepPartial, QueryRunner, FindConditions } from 'typeorm';
+import { Connection, ConnectionOptions, DeepPartial, QueryRunner, FindConditions, FindManyOptions } from 'typeorm';
 import path from 'path';
 
 import { Database as BaseDatabase, DatabaseInterface } from '@vulcanize/util';
@@ -148,7 +148,7 @@ export class Database implements DatabaseInterface {
     return this._baseDatabase.getEntities(queryRunner, entity, findConditions);
   }
 
-  async removeEntities<Entity> (queryRunner: QueryRunner, entity: new () => Entity, findConditions?: FindConditions<Entity>): Promise<void> {
+  async removeEntities<Entity> (queryRunner: QueryRunner, entity: new () => Entity, findConditions?: FindManyOptions<Entity> | FindConditions<Entity>): Promise<void> {
     return this._baseDatabase.removeEntities(queryRunner, entity, findConditions);
   }
 

packages/uni-watcher/src/fill.ts

@@ -90,8 +90,8 @@ export const main = async (): Promise<any> => {
   await fillBlocks(jobQueue, indexer, ethClient, eventWatcher, argv);
 };
 
-main().then(() => {
+main().catch(err => {
-  process.exit();
-}).catch(err => {
   log(err);
+}).finally(() => {
+  process.exit();
 });

packages/uni-watcher/src/indexer.ts

@@ -324,6 +324,10 @@ export class Indexer implements IndexerInterface {
     return this._baseIndexer.getBlockEvents(blockHash);
   }
 
+  async removeUnknownEvents (block: BlockProgress): Promise<void> {
+    return this._baseIndexer.removeUnknownEvents(Event, block);
+  }
+
   async updateSyncStatusIndexedBlock (blockHash: string, blockNumber: number, force = false): Promise<SyncStatus> {
     return this._baseIndexer.updateSyncStatusIndexedBlock(blockHash, blockNumber, force);
   }

packages/util/src/database.ts

@@ -10,6 +10,7 @@ import {
   createConnection,
   DeepPartial,
   FindConditions,
+  FindManyOptions,
   In,
   QueryRunner,
   Repository
@@ -261,7 +262,7 @@ export class Database {
     }
   }
 
-  async removeEntities<Entity> (queryRunner: QueryRunner, entity: new () => Entity, findConditions?: FindConditions<Entity>): Promise<void> {
+  async removeEntities<Entity> (queryRunner: QueryRunner, entity: new () => Entity, findConditions?: FindManyOptions<Entity> | FindConditions<Entity>): Promise<void> {
     const repo = queryRunner.manager.getRepository(entity);
 
     const entities = await repo.find(findConditions);

packages/util/src/events.ts

@@ -72,8 +72,13 @@ export class EventWatcher {
 
     await this._indexer.updateBlockProgress(dbEvent.block.blockHash, dbEvent.index);
     const blockProgress = await this._indexer.getBlockProgress(dbEvent.block.blockHash);
+
     if (blockProgress) {
       await this.publishBlockProgressToSubscribers(blockProgress);
+
+      if (blockProgress.isComplete) {
+        await this._indexer.removeUnknownEvents(blockProgress);
+      }
     }
 
     return dbEvent;

packages/util/src/indexer.ts

@@ -196,6 +196,31 @@ export class Indexer {
     return events;
   }
 
+  async removeUnknownEvents (eventEntityClass: new () => EventInterface, block: BlockProgressInterface): Promise<void> {
+    const dbTx = await this._db.createTransactionRunner();
+
+    try {
+      await this._db.removeEntities(
+        dbTx,
+        eventEntityClass,
+        {
+          where: {
+            block: { id: block.id },
+            eventName: UNKNOWN_EVENT_NAME
+          },
+          relations: ['block']
+        }
+      );
+
+      await dbTx.commitTransaction();
+    } catch (error) {
+      await dbTx.rollbackTransaction();
+      throw error;
+    } finally {
+      await dbTx.release();
+    }
+  }
+
   async getAncestorAtDepth (blockHash: string, depth: number): Promise<string> {
     return this._db.getAncestorAtDepth(blockHash, depth);
   }

packages/util/src/types.ts

@@ -2,7 +2,7 @@
 // Copyright 2021 Vulcanize, Inc.
 //
 
-import { DeepPartial, FindConditions, QueryRunner } from 'typeorm';
+import { DeepPartial, FindConditions, FindManyOptions, QueryRunner } from 'typeorm';
 
 export interface BlockProgressInterface {
   id: number;
@@ -55,6 +55,7 @@ export interface IndexerInterface {
   getBlockEvents (blockHash: string): Promise<Array<EventInterface>>
   getAncestorAtDepth (blockHash: string, depth: number): Promise<string>
   getOrFetchBlockEvents (block: DeepPartial<BlockProgressInterface>): Promise<Array<EventInterface>>
+  removeUnknownEvents (block: BlockProgressInterface): Promise<void>
   updateBlockProgress (blockHash: string, lastProcessedEventIndex: number): Promise<void>
   updateSyncStatusChainHead (blockHash: string, blockNumber: number): Promise<SyncStatusInterface>
   updateSyncStatusIndexedBlock (blockHash: string, blockNumber: number, force?: boolean): Promise<SyncStatusInterface>
@@ -85,6 +86,6 @@ export interface DatabaseInterface {
   updateSyncStatusCanonicalBlock (queryRunner: QueryRunner, blockHash: string, blockNumber: number, force?: boolean): Promise<SyncStatusInterface>;
   saveEvents (queryRunner: QueryRunner, block: DeepPartial<BlockProgressInterface>, events: DeepPartial<EventInterface>[]): Promise<void>;
   saveEventEntity (queryRunner: QueryRunner, entity: EventInterface): Promise<EventInterface>;
-  removeEntities<Entity> (queryRunner: QueryRunner, entity: new () => Entity, findConditions?: FindConditions<Entity>): Promise<void>;
+  removeEntities<Entity> (queryRunner: QueryRunner, entity: new () => Entity, findConditions?: FindManyOptions<Entity> | FindConditions<Entity>): Promise<void>;
   getContract?: (address: string) => Promise<ContractInterface | undefined>
 }