Add a table for entites in frothy region for subgraph watchers (#231)

* Add a table for entites in frothy region and update it in a subscriber

* Accommodate changes to other watchers and codegen

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

@@ -0,0 +1,31 @@
+className: FrothyEntity
+indexOn:
+  - columns:
+      - blockNumber
+columns:
+  - name: id
+    pgType: varchar
+    tsType: string
+    columnType: PrimaryColumn
+  - name: name
+    pgType: varchar
+    tsType: string
+    columnType: PrimaryColumn
+  - name: blockHash
+    pgType: varchar
+    tsType: string
+    columnType: PrimaryColumn
+    columnOptions:
+      - option: length
+        value: 66
+  - name: blockNumber
+    pgType: integer
+    tsType: number
+    columnType: Column
+imports:
+  - toImport:
+      - Entity
+      - PrimaryColumn
+      - Column
+      - Index
+    from: typeorm

packages/codegen/src/database.ts

@@ -16,10 +16,12 @@ const TEMPLATE_FILE = './templates/database-template.handlebars';
 
 export class Database {
   _queries: Array<any>;
+  _subgraphEntities: Array<any>;
   _templateString: string;
 
   constructor () {
     this._queries = [];
+    this._subgraphEntities = [];
     this._templateString = fs.readFileSync(path.resolve(__dirname, TEMPLATE_FILE)).toString();
   }
 
@@ -72,6 +74,23 @@ export class Database {
     this._queries.push(queryObject);
   }
 
+  addSubgraphEntities (subgraphSchemaDocument: any): void {
+    // Add subgraph entities for adding them to the entities list.
+    const subgraphTypeDefs = subgraphSchemaDocument.definitions;
+
+    subgraphTypeDefs.forEach((def: any) => {
+      if (def.kind !== 'ObjectTypeDefinition') {
+        return;
+      }
+
+      const entityObject: any = {
+        className: def.name.value
+      };
+
+      this._subgraphEntities.push(entityObject);
+    });
+  }
+
   /**
    * Writes the database file generated from a template to a stream.
    * @param outStream A writable output stream to write the database file to.
@@ -79,7 +98,8 @@ export class Database {
   exportDatabase (outStream: Writable): void {
     const template = Handlebars.compile(this._templateString);
     const obj = {
-      queries: this._queries
+      queries: this._queries,
+      subgraphEntities: this._subgraphEntities
     };
     const database = template(obj);
     outStream.write(database);

packages/codegen/src/entity.ts

@@ -173,7 +173,7 @@ export class Entity {
    * Writes the generated entity files in the given directory.
    * @param entityDir Directory to write the entities to.
    */
-  exportEntities (entityDir: string): void {
+  exportEntities (entityDir: string, subgraphPath: string): void {
     this._addEventEntity();
     this._addSyncStatusEntity();
     this._addContractEntity();
@@ -181,6 +181,11 @@ export class Entity {
     this._addStateEntity();
     this._addStateSyncStatusEntity();
 
+    // Add FrothyEntity table only for subgraph watchers
+    if (subgraphPath) {
+      this._addFrothyEntity();
+    }
+
     const template = Handlebars.compile(this._templateString);
     this._entities.forEach(entityObj => {
       const entity = template(entityObj);
@@ -288,6 +293,11 @@ export class Entity {
     this._entities.push(entity);
   }
 
+  _addFrothyEntity (): void {
+    const entity = yaml.load(fs.readFileSync(path.resolve(__dirname, TABLES_DIR, 'FrothyEntity.yaml'), 'utf8'));
+    this._entities.push(entity);
+  }
+
   _addBigIntTransformerOption (entityObject: any): void {
     let importObject = entityObject.imports.find((element: any) => {
       return element.from === '@cerc-io/util';

packages/codegen/src/generate-code.ts

@@ -37,6 +37,7 @@ import { importState } from './import-state';
 import { exportInspectCID } from './inspect-cid';
 import { getSubgraphConfig } from './utils/subgraph';
 import { exportIndexBlock } from './index-block';
+import { exportSubscriber } from './subscriber';
 
 const main = async (): Promise<void> => {
   const argv = await yargs(hideBin(process.argv))
@@ -217,7 +218,7 @@ function generateWatcher (visitor: Visitor, contracts: any[], config: any) {
   const entityDir = outputDir
     ? path.join(outputDir, 'src/entity')
     : '';
-  visitor.exportEntities(entityDir);
+  visitor.exportEntities(entityDir, config.subgraphPath);
 
   outStream = outputDir
     ? fs.createWriteStream(path.join(outputDir, 'README.md'))
@@ -323,6 +324,13 @@ function generateWatcher (visitor: Visitor, contracts: any[], config: any) {
     ? fs.createWriteStream(path.join(outputDir, 'src/cli/index-block.ts'))
     : process.stdout;
   exportIndexBlock(outStream);
+
+  if (config.subgraphPath) {
+    outStream = outputDir
+      ? fs.createWriteStream(path.join(outputDir, 'src/entity/Subscriber.ts'))
+      : process.stdout;
+    exportSubscriber(outStream);
+  }
 }
 
 function getConfig (configFile: string): any {

packages/codegen/src/subscriber.ts

@@ -0,0 +1,21 @@
+//
+// Copyright 2022 Vulcanize, Inc.
+//
+
+import fs from 'fs';
+import path from 'path';
+import Handlebars from 'handlebars';
+import { Writable } from 'stream';
+
+const SUBSCRIBER_TEMPLATE_FILE = './templates/subscriber-template.handlebars';
+
+/**
+ * Writes the subscriber file generated from template to a stream.
+ * @param outStream A writable output stream to write the subscriber file to.
+ */
+export function exportSubscriber (subscriberOutStream: Writable): void {
+  const subscriberTemplateString = fs.readFileSync(path.resolve(__dirname, SUBSCRIBER_TEMPLATE_FILE)).toString();
+  const subscriberTemplate = Handlebars.compile(subscriberTemplateString);
+  const subscriber = subscriberTemplate({});
+  subscriberOutStream.write(subscriber);
+}

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

@@ -17,6 +17,15 @@ import { State } from './entity/State';
 {{#each queries as | query |}}
 import { {{query.entityName}} } from './entity/{{query.entityName}}';
 {{/each}}
+{{#each subgraphEntities as | subgraphEntity |}}
+import { {{subgraphEntity.className}} } from './entity/{{subgraphEntity.className}}';
+{{/each}}
+
+export const ENTITIES = [
+  {{~#each queries as | query |}}{{query.entityName}}, {{/each}}
+  {{~#each subgraphEntities as | subgraphEntity |}}{{subgraphEntity.className}}
+  {{~#unless @last}}, {{/unless}}
+  {{~/each}}];
 
 export class Database implements DatabaseInterface {
   _config: ConnectionOptions;

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

@@ -41,7 +41,7 @@ import { GraphWatcher } from '@cerc-io/graph-node';
 {{#each contracts as | contract |}}
 import {{contract.contractName}}Artifacts from './artifacts/{{contract.contractName}}.json';
 {{/each}}
-import { Database } from './database';
+import { Database, ENTITIES } from './database';
 import { createInitialState, handleEvent, createStateDiff, createStateCheckpoint } from './hooks';
 import { Contract } from './entity/Contract';
 import { Event } from './entity/Event';
@@ -49,13 +49,12 @@ import { SyncStatus } from './entity/SyncStatus';
 import { StateSyncStatus } from './entity/StateSyncStatus';
 import { BlockProgress } from './entity/BlockProgress';
 import { State } from './entity/State';
-
-{{#each queries as | query |}}
-import { {{query.entityName}} } from './entity/{{query.entityName}}';
-{{/each}}
 {{#each subgraphEntities as | subgraphEntity |}}
 import { {{subgraphEntity.className}} } from './entity/{{subgraphEntity.className}}';
 {{/each}}
+{{#if (subgraphPath)}}
+import { FrothyEntity } from './entity/FrothyEntity';
+{{/if}}
 
 const log = debug('vulcanize:indexer');
 const JSONbigNative = JSONbig({ useNativeBigInt: true });
@@ -521,7 +520,12 @@ export class Indexer implements IndexerInterface {
   }
 
   async updateSyncStatusCanonicalBlock (blockHash: string, blockNumber: number, force = false): Promise<SyncStatus> {
-    return this._baseIndexer.updateSyncStatusCanonicalBlock(blockHash, blockNumber, force);
+    const syncStatus = this._baseIndexer.updateSyncStatusCanonicalBlock(blockHash, blockNumber, force);
+    {{#if (subgraphPath)}}
+    await this.pruneFrothyEntities(blockNumber);
+    {{/if}}
+
+    return syncStatus;
   }
 
   async getEvent (id: string): Promise<Event | undefined> {
@@ -556,6 +560,12 @@ export class Indexer implements IndexerInterface {
     return this._baseIndexer.markBlocksAsPruned(blocks);
   }
 
+  {{#if (subgraphPath)}}
+  async pruneFrothyEntities (blockNumber: number): Promise<void> {
+    await this._graphWatcher.pruneFrothyEntities(FrothyEntity, blockNumber);
+  }
+  {{/if}}
+
   async updateBlockProgress (block: BlockProgress, lastProcessedEventIndex: number): Promise<BlockProgress> {
     return this._baseIndexer.updateBlockProgress(block, lastProcessedEventIndex);
   }
@@ -565,14 +575,11 @@ export class Indexer implements IndexerInterface {
   }
 
   async resetWatcherToBlock (blockNumber: number): Promise<void> {
-    const entities = [
-      {{#each queries as | query |}}
-      {{query.entityName}},
-      {{/each}}
-      {{#each subgraphEntities as | subgraphEntity |}}
-      {{subgraphEntity.className}},
-      {{/each}}
-    ];
+    {{#if (subgraphPath)}}
+    const entities = [...ENTITIES, FrothyEntity];
+    {{else}}
+    const entities = [...ENTITIES];
+    {{/if}}
     await this._baseIndexer.resetWatcherToBlock(blockNumber, entities);
   }
 

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

@@ -0,0 +1,20 @@
+//
+// Copyright 2022 Vulcanize, Inc.
+//
+
+import { EventSubscriber, EntitySubscriberInterface, InsertEvent, UpdateEvent } from 'typeorm';
+
+import { FrothyEntity } from './FrothyEntity';
+import { ENTITIES } from '../database';
+import { afterEntityInsertOrUpdate } from '@cerc-io/graph-node';
+
+@EventSubscriber()
+export class EntitySubscriber implements EntitySubscriberInterface {
+  async afterInsert (event: InsertEvent<any>): Promise<void> {
+    await afterEntityInsertOrUpdate(FrothyEntity, ENTITIES, event);
+  }
+
+  async afterUpdate (event: UpdateEvent<any>): Promise<void> {
+    await afterEntityInsertOrUpdate(FrothyEntity, ENTITIES, event);
+  }
+}

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

@@ -21,7 +21,7 @@
     // "removeComments": true,                      /* Do not emit comments to output. */
     // "noEmit": true,                              /* Do not emit outputs. */
     // "importHelpers": true,                       /* Import emit helpers from 'tslib'. */
-    // "downlevelIteration": true,                  /* Provide full support for iterables in 'for-of', spread, and destructuring when targeting 'ES5' or 'ES3'. */
+    "downlevelIteration": true,                  /* Provide full support for iterables in 'for-of', spread, and destructuring when targeting 'ES5' or 'ES3'. */
     // "isolatedModules": true,                     /* Transpile each file as a separate module (similar to 'ts.transpileModule'). */
 
     /* Strict Type-Checking Options */

packages/codegen/src/visitor.ts

@@ -155,6 +155,7 @@ export class Visitor {
     this._resolvers.addSubgraphResolvers(subgraphSchemaDocument);
     this._reset.addSubgraphEntities(subgraphSchemaDocument);
     this._indexer.addSubgraphEntities(subgraphSchemaDocument);
+    this._database.addSubgraphEntities(subgraphSchemaDocument);
   }
 
   /**
@@ -187,8 +188,8 @@ export class Visitor {
    * Writes the generated entity files in the given directory.
    * @param entityDir Directory to write the entities to.
    */
-  exportEntities (entityDir: string): void {
-    this._entity.exportEntities(entityDir);
+  exportEntities (entityDir: string, subgraphPath: string): void {
+    this._entity.exportEntities(entityDir, subgraphPath);
   }
 
   /**

packages/eden-watcher/src/entity/FrothyEntity.ts

@@ -0,0 +1,21 @@
+//
+// Copyright 2022 Vulcanize, Inc.
+//
+
+import { Entity, PrimaryColumn, Column, Index } from 'typeorm';
+
+@Entity()
+@Index(['blockNumber'])
+export class FrothyEntity {
+  @PrimaryColumn('varchar')
+  id!: string;
+
+  @PrimaryColumn('varchar')
+  name!: string;
+
+  @PrimaryColumn('varchar', { length: 66 })
+  blockHash!: string;
+
+  @Column('integer')
+  blockNumber!: number;
+}

packages/eden-watcher/src/entity/Subscriber.ts

@@ -0,0 +1,20 @@
+//
+// Copyright 2022 Vulcanize, Inc.
+//
+
+import { EventSubscriber, EntitySubscriberInterface, InsertEvent, UpdateEvent } from 'typeorm';
+
+import { FrothyEntity } from './FrothyEntity';
+import { ENTITIES } from '../database';
+import { afterEntityInsertOrUpdate } from '@cerc-io/graph-node';
+
+@EventSubscriber()
+export class EntitySubscriber implements EntitySubscriberInterface {
+  async afterInsert (event: InsertEvent<any>): Promise<void> {
+    await afterEntityInsertOrUpdate(FrothyEntity, ENTITIES, event);
+  }
+
+  async afterUpdate (event: UpdateEvent<any>): Promise<void> {
+    await afterEntityInsertOrUpdate(FrothyEntity, ENTITIES, event);
+  }
+}

packages/eden-watcher/src/indexer.ts

@@ -55,6 +55,7 @@ import { Distribution } from './entity/Distribution';
 import { Claim } from './entity/Claim';
 import { Account } from './entity/Account';
 import { Slash } from './entity/Slash';
+import { FrothyEntity } from './entity/FrothyEntity';
 
 const KIND_EDENNETWORK = 'EdenNetwork';
 const KIND_MERKLEDISTRIBUTOR = 'EdenNetworkDistribution';
@@ -444,7 +445,10 @@ export class Indexer implements IndexerInterface {
   }
 
   async updateSyncStatusCanonicalBlock (blockHash: string, blockNumber: number, force = false): Promise<SyncStatus> {
-    return this._baseIndexer.updateSyncStatusCanonicalBlock(blockHash, blockNumber, force);
+    const syncStatus = this._baseIndexer.updateSyncStatusCanonicalBlock(blockHash, blockNumber, force);
+    await this.pruneFrothyEntities(blockNumber);
+
+    return syncStatus;
   }
 
   async getEvent (id: string): Promise<Event | undefined> {
@@ -481,6 +485,10 @@ export class Indexer implements IndexerInterface {
     await this._graphWatcher.pruneEntities(blocks, ENTITIES);
   }
 
+  async pruneFrothyEntities (blockNumber: number): Promise<void> {
+    await this._graphWatcher.pruneFrothyEntities(FrothyEntity, blockNumber);
+  }
+
   async updateBlockProgress (block: BlockProgress, lastProcessedEventIndex: number): Promise<BlockProgress> {
     return this._baseIndexer.updateBlockProgress(block, lastProcessedEventIndex);
   }
@@ -522,7 +530,7 @@ export class Indexer implements IndexerInterface {
   }
 
   async resetWatcherToBlock (blockNumber: number): Promise<void> {
-    const entities = [ProducerSet, Producer, RewardSchedule, RewardScheduleEntry, Network, Staker, ProducerEpoch, Epoch, Block, SlotClaim, Slot, Distributor, Distribution, Claim, Account, Slash];
+    const entities = [...ENTITIES, FrothyEntity];
     await this._baseIndexer.resetWatcherToBlock(blockNumber, entities);
   }
 

packages/eden-watcher/tsconfig.json

@@ -21,7 +21,7 @@
     // "removeComments": true,                      /* Do not emit comments to output. */
     // "noEmit": true,                              /* Do not emit outputs. */
     // "importHelpers": true,                       /* Import emit helpers from 'tslib'. */
-    // "downlevelIteration": true,                  /* Provide full support for iterables in 'for-of', spread, and destructuring when targeting 'ES5' or 'ES3'. */
+    "downlevelIteration": true,                  /* Provide full support for iterables in 'for-of', spread, and destructuring when targeting 'ES5' or 'ES3'. */
     // "isolatedModules": true,                     /* Transpile each file as a separate module (similar to 'ts.transpileModule'). */
 
     /* Strict Type-Checking Options */

packages/erc20-watcher/src/database.ts

@@ -17,6 +17,8 @@ import { BlockProgress } from './entity/BlockProgress';
 import { State } from './entity/State';
 import { StateSyncStatus } from './entity/StateSyncStatus';
 
+export const ENTITIES = new Set([Allowance, Balance]);
+
 export class Database implements DatabaseInterface {
   _config: ConnectionOptions
   _conn!: Connection

packages/erc20-watcher/src/indexer.ts

@@ -14,7 +14,7 @@ import { EthClient } from '@cerc-io/ipld-eth-client';
 import { MappingKey, StorageLayout } from '@cerc-io/solidity-mapper';
 import { IndexerInterface, Indexer as BaseIndexer, ValueResult, JobQueue, Where, QueryOptions, ServerConfig, StateStatus } from '@cerc-io/util';
 
-import { Database } from './database';
+import { Database, ENTITIES } from './database';
 import { Event } from './entity/Event';
 import { fetchTokenDecimals, fetchTokenName, fetchTokenSymbol, fetchTokenTotalSupply } from './utils';
 import { SyncStatus } from './entity/SyncStatus';
@@ -23,8 +23,6 @@ import artifacts from './artifacts/ERC20.json';
 import { BlockProgress } from './entity/BlockProgress';
 import { Contract } from './entity/Contract';
 import { State } from './entity/State';
-import { Allowance } from './entity/Allowance';
-import { Balance } from './entity/Balance';
 
 const log = debug('vulcanize:indexer');
 const JSONbigNative = JSONbig({ useNativeBigInt: true });
@@ -430,7 +428,7 @@ export class Indexer implements IndexerInterface {
   }
 
   async resetWatcherToBlock (blockNumber: number): Promise<void> {
-    const entities = [Allowance, Balance];
+    const entities = [...ENTITIES];
     await this._baseIndexer.resetWatcherToBlock(blockNumber, entities);
   }
 

packages/erc721-watcher/src/database.ts

@@ -30,6 +30,8 @@ import { _TokenApprovals } from './entity/_TokenApprovals';
 import { _OperatorApprovals } from './entity/_OperatorApprovals';
 import { TransferCount } from './entity/TransferCount';
 
+export const ENTITIES = new Set([_Balances, _Name, _OperatorApprovals, _Owners, _Symbol, _TokenApprovals, BalanceOf, GetApproved, IsApprovedForAll, Name, OwnerOf, SupportsInterface, Symbol, TokenURI, TransferCount]);
+
 export class Database implements DatabaseInterface {
   _config: ConnectionOptions;
   _conn!: Connection;

packages/erc721-watcher/src/indexer.ts

@@ -30,7 +30,7 @@ import {
 } from '@cerc-io/util';
 
 import ERC721Artifacts from './artifacts/ERC721.json';
-import { Database } from './database';
+import { Database, ENTITIES } from './database';
 import { createInitialState, handleEvent, createStateDiff, createStateCheckpoint } from './hooks';
 import { Contract } from './entity/Contract';
 import { Event } from './entity/Event';
@@ -38,20 +38,6 @@ import { SyncStatus } from './entity/SyncStatus';
 import { StateSyncStatus } from './entity/StateSyncStatus';
 import { BlockProgress } from './entity/BlockProgress';
 import { State } from './entity/State';
-import { SupportsInterface } from './entity/SupportsInterface';
-import { BalanceOf } from './entity/BalanceOf';
-import { OwnerOf } from './entity/OwnerOf';
-import { GetApproved } from './entity/GetApproved';
-import { IsApprovedForAll } from './entity/IsApprovedForAll';
-import { Name } from './entity/Name';
-import { Symbol } from './entity/Symbol';
-import { TokenURI } from './entity/TokenURI';
-import { _Name } from './entity/_Name';
-import { _Symbol } from './entity/_Symbol';
-import { _Owners } from './entity/_Owners';
-import { _Balances } from './entity/_Balances';
-import { _TokenApprovals } from './entity/_TokenApprovals';
-import { _OperatorApprovals } from './entity/_OperatorApprovals';
 import { TransferCount } from './entity/TransferCount';
 
 const log = debug('vulcanize:indexer');
@@ -878,7 +864,7 @@ export class Indexer implements IndexerInterface {
   }
 
   async resetWatcherToBlock (blockNumber: number): Promise<void> {
-    const entities = [SupportsInterface, BalanceOf, OwnerOf, GetApproved, IsApprovedForAll, Name, Symbol, TokenURI, _Name, _Symbol, _Owners, _Balances, _TokenApprovals, _OperatorApprovals, TransferCount];
+    const entities = [...ENTITIES];
     await this._baseIndexer.resetWatcherToBlock(blockNumber, entities);
   }
 

packages/erc721-watcher/tsconfig.json

@@ -21,7 +21,7 @@
     // "removeComments": true,                      /* Do not emit comments to output. */
     // "noEmit": true,                              /* Do not emit outputs. */
     // "importHelpers": true,                       /* Import emit helpers from 'tslib'. */
-    // "downlevelIteration": true,                  /* Provide full support for iterables in 'for-of', spread, and destructuring when targeting 'ES5' or 'ES3'. */
+    "downlevelIteration": true,                  /* Provide full support for iterables in 'for-of', spread, and destructuring when targeting 'ES5' or 'ES3'. */
     // "isolatedModules": true,                     /* Transpile each file as a separate module (similar to 'ts.transpileModule'). */
 
     /* Strict Type-Checking Options */

packages/graph-node/src/database.ts

@@ -1256,6 +1256,11 @@ export class Database {
     await Promise.all(updatePromises);
   }
 
+  async pruneFrothyEntities<Entity> (queryRunner: QueryRunner, frothyEntityType: new () => Entity, blockNumber: number): Promise<void> {
+    // Remove frothy entity entries at | below the prune block height
+    return this._baseDatabase.removeEntities(queryRunner, frothyEntityType, { where: { blockNumber: LessThanOrEqual(blockNumber) } });
+  }
+
   _measureCachedPrunedEntities () {
     const totalEntities = Array.from(this.cachedEntities.latestPrunedEntities.values())
       .reduce((acc, idEntitiesMap) => acc + idEntitiesMap.size, 0);

packages/graph-node/src/index.ts

@@ -3,5 +3,6 @@ export * from './database';
 export {
   prepareEntityState,
   updateEntitiesFromState,
-  resolveEntityFieldConflicts
+  resolveEntityFieldConflicts,
+  afterEntityInsertOrUpdate
 } from './utils';

packages/graph-node/src/utils.ts

@@ -3,9 +3,10 @@ import path from 'path';
 import fs from 'fs-extra';
 import debug from 'debug';
 import yaml from 'js-yaml';
-import { ValueTransformer } from 'typeorm';
+import { EntityTarget, InsertEvent, UpdateEvent, ValueTransformer } from 'typeorm';
 import { ColumnMetadata } from 'typeorm/metadata/ColumnMetadata';
 import assert from 'assert';
+import _ from 'lodash';
 
 import { GraphDecimal, IndexerInterface, jsonBigIntStringReplacer, StateInterface } from '@cerc-io/util';
 import { MappingKey, StorageLayout } from '@cerc-io/solidity-mapper';
@@ -897,3 +898,29 @@ export const updateEntitiesFromState = async (database: Database, indexer: Index
     console.timeEnd(`time:watcher#GraphWatcher-updateEntitiesFromState-update-entity-${entityName}`);
   }
 };
+
+export const afterEntityInsertOrUpdate = async<Entity> (frothyEntityType: EntityTarget<Entity>, entities: Set<any>, event: InsertEvent<any> | UpdateEvent<any>): Promise<void> => {
+  const entity = event.entity;
+
+  // TODO: Check and return if entity is being pruned (is_pruned flag update)
+
+  // Insert the entity details in FrothyEntity table
+  if (entities.has(entity.constructor)) {
+    const frothyEntity = event.manager.create(
+      frothyEntityType,
+      {
+        ..._.pick(entity, ['id', 'blockHash', 'blockNumber']),
+        ...{ name: entity.constructor.name }
+      }
+    );
+
+    await event.manager.createQueryBuilder()
+      .insert()
+      .into(frothyEntityType)
+      .values(frothyEntity as any)
+      .orIgnore()
+      .execute();
+  }
+
+  // TOOD: Update latest entity tables
+};

packages/graph-node/src/watcher.ts

@@ -372,6 +372,20 @@ export class GraphWatcher {
     }
   }
 
+  async pruneFrothyEntities<Entity> (frothyEntityType: new () => Entity, blockNumber: number): Promise<void> {
+    const dbTx = await this._database.createTransactionRunner();
+    try {
+      await this._database.pruneFrothyEntities(dbTx, frothyEntityType, blockNumber);
+
+      dbTx.commitTransaction();
+    } catch (error) {
+      await dbTx.rollbackTransaction();
+      throw error;
+    } finally {
+      await dbTx.release();
+    }
+  }
+
   pruneEntityCacheFrothyBlocks (canonicalBlockHash: string, canonicalBlockNumber: number) {
     this._database.pruneEntityCacheFrothyBlocks(canonicalBlockHash, canonicalBlockNumber);
   }

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

@@ -17,6 +17,11 @@ import { State } from './entity/State';
 
 import { GetMethod } from './entity/GetMethod';
 import { _Test } from './entity/_Test';
+import { Author } from './entity/Author';
+import { Blog } from './entity/Blog';
+import { Category } from './entity/Category';
+
+export const ENTITIES = new Set([_Test, Author, Blog, Category, GetMethod]);
 
 export class Database implements DatabaseInterface {
   _config: ConnectionOptions;

packages/graph-test-watcher/src/entity/FrothyEntity.ts

@@ -0,0 +1,21 @@
+//
+// Copyright 2022 Vulcanize, Inc.
+//
+
+import { Entity, PrimaryColumn, Column, Index } from 'typeorm';
+
+@Entity()
+@Index(['blockNumber'])
+export class FrothyEntity {
+  @PrimaryColumn('varchar')
+  id!: string;
+
+  @PrimaryColumn('varchar')
+  name!: string;
+
+  @PrimaryColumn('varchar', { length: 66 })
+  blockHash!: string;
+
+  @Column('integer')
+  blockNumber!: number;
+}

packages/graph-test-watcher/src/entity/Subscriber.ts

@@ -0,0 +1,20 @@
+//
+// Copyright 2022 Vulcanize, Inc.
+//
+
+import { EventSubscriber, EntitySubscriberInterface, InsertEvent, UpdateEvent } from 'typeorm';
+
+import { FrothyEntity } from './FrothyEntity';
+import { ENTITIES } from '../database';
+import { afterEntityInsertOrUpdate } from '@cerc-io/graph-node';
+
+@EventSubscriber()
+export class EntitySubscriber implements EntitySubscriberInterface {
+  async afterInsert (event: InsertEvent<any>): Promise<void> {
+    await afterEntityInsertOrUpdate(FrothyEntity, ENTITIES, event);
+  }
+
+  async afterUpdate (event: UpdateEvent<any>): Promise<void> {
+    await afterEntityInsertOrUpdate(FrothyEntity, ENTITIES, event);
+  }
+}

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

@@ -31,7 +31,7 @@ import {
 } from '@cerc-io/util';
 import { GraphWatcher } from '@cerc-io/graph-node';
 
-import { Database } from './database';
+import { Database, ENTITIES } from './database';
 import { Contract } from './entity/Contract';
 import { Event } from './entity/Event';
 import { SyncStatus } from './entity/SyncStatus';
@@ -43,6 +43,7 @@ import { createInitialState, handleEvent, createStateDiff, createStateCheckpoint
 import { Author } from './entity/Author';
 import { Blog } from './entity/Blog';
 import { Category } from './entity/Category';
+import { FrothyEntity } from './entity/FrothyEntity';
 
 const log = debug('vulcanize:indexer');
 const JSONbigNative = JSONbig({ useNativeBigInt: true });
@@ -449,7 +450,10 @@ export class Indexer implements IndexerInterface {
   }
 
   async updateSyncStatusCanonicalBlock (blockHash: string, blockNumber: number, force = false): Promise<SyncStatus> {
-    return this._baseIndexer.updateSyncStatusCanonicalBlock(blockHash, blockNumber, force);
+    const syncStatus = this._baseIndexer.updateSyncStatusCanonicalBlock(blockHash, blockNumber, force);
+    await this.pruneFrothyEntities(blockNumber);
+
+    return syncStatus;
   }
 
   async getEvent (id: string): Promise<Event | undefined> {
@@ -484,6 +488,10 @@ export class Indexer implements IndexerInterface {
     return this._baseIndexer.markBlocksAsPruned(blocks);
   }
 
+  async pruneFrothyEntities (blockNumber: number): Promise<void> {
+    await this._graphWatcher.pruneFrothyEntities(FrothyEntity, blockNumber);
+  }
+
   async updateBlockProgress (block: BlockProgress, lastProcessedEventIndex: number): Promise<BlockProgress> {
     return this._baseIndexer.updateBlockProgress(block, lastProcessedEventIndex);
   }
@@ -525,7 +533,7 @@ export class Indexer implements IndexerInterface {
   }
 
   async resetWatcherToBlock (blockNumber: number): Promise<void> {
-    const entities = [Author, Blog, Category];
+    const entities = [...ENTITIES, FrothyEntity];
     await this._baseIndexer.resetWatcherToBlock(blockNumber, entities);
   }
 

packages/graph-test-watcher/tsconfig.json

@@ -21,7 +21,7 @@
     // "removeComments": true,                      /* Do not emit comments to output. */
     // "noEmit": true,                              /* Do not emit outputs. */
     // "importHelpers": true,                       /* Import emit helpers from 'tslib'. */
-    // "downlevelIteration": true,                  /* Provide full support for iterables in 'for-of', spread, and destructuring when targeting 'ES5' or 'ES3'. */
+    "downlevelIteration": true,                  /* Provide full support for iterables in 'for-of', spread, and destructuring when targeting 'ES5' or 'ES3'. */
     // "isolatedModules": true,                     /* Transpile each file as a separate module (similar to 'ts.transpileModule'). */
 
     /* Strict Type-Checking Options */

packages/mobymask-watcher/src/database.ts

@@ -20,6 +20,8 @@ import { IsRevoked } from './entity/IsRevoked';
 import { IsPhisher } from './entity/IsPhisher';
 import { IsMember } from './entity/IsMember';
 
+export const ENTITIES = new Set([_Owner, IsMember, IsPhisher, IsRevoked, MultiNonce]);
+
 export class Database implements DatabaseInterface {
   _config: ConnectionOptions;
   _conn!: Connection;

packages/mobymask-watcher/src/indexer.ts

@@ -31,7 +31,7 @@ import {
 } from '@cerc-io/util';
 
 import PhisherRegistryArtifacts from './artifacts/PhisherRegistry.json';
-import { Database } from './database';
+import { Database, ENTITIES } from './database';
 import { createInitialState, handleEvent, createStateDiff, createStateCheckpoint } from './hooks';
 import { Contract } from './entity/Contract';
 import { Event } from './entity/Event';
@@ -608,7 +608,7 @@ export class Indexer implements IndexerInterface {
   }
 
   async resetWatcherToBlock (blockNumber: number): Promise<void> {
-    const entities = [MultiNonce, _Owner, IsRevoked, IsPhisher, IsMember];
+    const entities = [...ENTITIES];
     await this._baseIndexer.resetWatcherToBlock(blockNumber, entities);
   }
 

packages/mobymask-watcher/tsconfig.json

@@ -21,7 +21,7 @@
     // "removeComments": true,                      /* Do not emit comments to output. */
     // "noEmit": true,                              /* Do not emit outputs. */
     // "importHelpers": true,                       /* Import emit helpers from 'tslib'. */
-    // "downlevelIteration": true,                  /* Provide full support for iterables in 'for-of', spread, and destructuring when targeting 'ES5' or 'ES3'. */
+    "downlevelIteration": true,                  /* Provide full support for iterables in 'for-of', spread, and destructuring when targeting 'ES5' or 'ES3'. */
     // "isolatedModules": true,                     /* Transpile each file as a separate module (similar to 'ts.transpileModule'). */
 
     /* Strict Type-Checking Options */

packages/util/src/indexer.ts

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