Reset watcher to previous indexed block on start (#207)

* Reset watcher to previous indexed block before start

* Implement changes in other watchers

* Save successfully fetched blocks and events to prefetch cache

* Add unique query for transaction table

* Check db for blocks before fetching from eth-server

* Show all mismatches at a block

Co-authored-by: prathamesh0 <prathamesh.musale0@gmail.com>

packages/graph-node/src/cli/compare/compare-blocks.ts

@@ -200,7 +200,6 @@ export const main = async (): Promise<void> => {
       for (const [queryName, entityName] of Object.entries(queryNames)) {
         try {
           log(`At block ${blockNumber} for query ${queryName}:`);
-          let resultDiff = '';
 
           if (fetchIds) {
             const queryLimit = config.queries.queryLimits[queryName];
@@ -230,11 +229,15 @@ export const main = async (): Promise<void> => {
               }
 
               if (diff) {
-                resultDiff = diff;
+                log('Results mismatch:', diff);
+                diffFound = true;
+              } else {
+                log('Results match.');
               }
             }
           } else {
             if (updatedEntities.has(entityName)) {
+              let resultDiff;
               let result;
               let skip = 0;
 
@@ -271,14 +274,14 @@ export const main = async (): Promise<void> => {
                 // eslint-disable-next-line no-unmodified-loop-condition
                 paginate
               );
-            }
-          }
 
-          if (resultDiff) {
-            log('Results mismatch:', resultDiff);
-            diffFound = true;
-          } else {
-            log('Results match.');
+              if (resultDiff) {
+                log('Results mismatch:', resultDiff);
+                diffFound = true;
+              } else {
+                log('Results match.');
+              }
+            }
           }
         } catch (err: any) {
           log('Error:', err.message);

packages/graph-node/src/database.ts

@@ -41,7 +41,8 @@ const DEFAULT_CLEAR_ENTITIES_CACHE_INTERVAL = 1000;
 export enum ENTITY_QUERY_TYPE {
   SINGULAR,
   DISTINCT_ON,
-  GROUP_BY
+  GROUP_BY,
+  UNIQUE
 }
 
 interface CachedEntities {
@@ -378,6 +379,10 @@ export class Database {
         entities = await this.getEntitiesDistinctOn(queryRunner, entity, block, where, queryOptions);
         break;
 
+      case ENTITY_QUERY_TYPE.UNIQUE:
+        entities = await this.getEntitiesUnique(queryRunner, entity, block, where, queryOptions);
+        break;
+
       default:
         // Use group by query if entity query type is not specified in map.
         entities = await this.getEntitiesGroupBy(queryRunner, entity, block, where, queryOptions);
@@ -413,6 +418,11 @@ export class Database {
       .andWhere('blockProgress.is_pruned = :isPruned', { isPruned: false })
       .groupBy('subTable.id');
 
+    if (where.id) {
+      subQuery = this._baseDatabase.buildQuery(repo, subQuery, { id: where.id });
+      delete where.id;
+    }
+
     if (block.hash) {
       const { canonicalBlockNumber, blockHashes } = await this._baseDatabase.getFrothyRegion(queryRunner, block.hash);
 
@@ -473,6 +483,11 @@ export class Database {
       .addOrderBy('subTable.id', 'ASC')
       .addOrderBy('subTable.block_number', 'DESC');
 
+    if (where.id) {
+      subQuery = this._baseDatabase.buildQuery(repo, subQuery, { id: where.id });
+      delete where.id;
+    }
+
     if (block.hash) {
       const { canonicalBlockNumber, blockHashes } = await this._baseDatabase.getFrothyRegion(queryRunner, block.hash);
 
@@ -554,6 +569,56 @@ export class Database {
     return entities as Entity[];
   }
 
+  async getEntitiesUnique<Entity> (
+    queryRunner: QueryRunner,
+    entity: new () => Entity,
+    block: BlockHeight,
+    where: Where = {},
+    queryOptions: QueryOptions = {}
+  ): Promise<Entity[]> {
+    const repo = queryRunner.manager.getRepository(entity);
+    const { tableName } = repo.metadata;
+
+    let selectQueryBuilder = repo.createQueryBuilder(tableName)
+      .addFrom('block_progress', 'blockProgress')
+      .where(`${tableName}.block_hash = blockProgress.block_hash`)
+      .andWhere('blockProgress.is_pruned = :isPruned', { isPruned: false });
+
+    if (block.hash) {
+      const { canonicalBlockNumber, blockHashes } = await this._baseDatabase.getFrothyRegion(queryRunner, block.hash);
+
+      selectQueryBuilder = selectQueryBuilder
+        .andWhere(new Brackets(qb => {
+          qb.where(`${tableName}.block_hash IN (:...blockHashes)`, { blockHashes })
+            .orWhere(`${tableName}.block_number <= :canonicalBlockNumber`, { canonicalBlockNumber });
+        }));
+    }
+
+    if (block.number) {
+      selectQueryBuilder = selectQueryBuilder.andWhere(`${tableName}.block_number <= :blockNumber`, { blockNumber: block.number });
+    }
+
+    selectQueryBuilder = this._baseDatabase.buildQuery(repo, selectQueryBuilder, where);
+
+    if (queryOptions.orderBy) {
+      selectQueryBuilder = this._baseDatabase.orderQuery(repo, selectQueryBuilder, queryOptions);
+    }
+
+    selectQueryBuilder = this._baseDatabase.orderQuery(repo, selectQueryBuilder, { ...queryOptions, orderBy: 'id' });
+
+    if (queryOptions.skip) {
+      selectQueryBuilder = selectQueryBuilder.offset(queryOptions.skip);
+    }
+
+    if (queryOptions.limit) {
+      selectQueryBuilder = selectQueryBuilder.limit(queryOptions.limit);
+    }
+
+    const entities = await selectQueryBuilder.getMany();
+
+    return entities as Entity[];
+  }
+
   async loadEntitiesRelations<Entity> (
     queryRunner: QueryRunner,
     block: BlockHeight,

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

@@ -197,6 +197,10 @@ export class Indexer implements IndexerInterface {
   updateStateStatusMap (address: string, stateStatus: StateStatus): void {
     return undefined;
   }
+
+  async resetWatcherToBlock (blockNumber: number): Promise<void> {
+    return undefined;
+  }
 }
 
 class ServerConfig implements ServerConfigInterface {