Refactor watchers for util package changes

packages/eden-watcher/src/cli/reset-cmds/state.ts

@@ -107,6 +107,8 @@ export const handler = async (argv: any): Promise<void> => {
       await indexer.updateHookStatusProcessedBlock(blockProgress.blockNumber, true);
     }
 
+    await indexer.updateSyncStatusChainHead(blockProgress.blockHash, blockProgress.blockNumber, true);
+
     dbTx.commitTransaction();
   } catch (error) {
     await dbTx.rollbackTransaction();

packages/eden-watcher/src/database.ts

@@ -150,10 +150,10 @@ export class Database implements IPLDDatabaseInterface {
     return this._baseDatabase.updateSyncStatusCanonicalBlock(repo, blockHash, blockNumber, force);
   }
 
-  async updateSyncStatusChainHead (queryRunner: QueryRunner, blockHash: string, blockNumber: number): Promise<SyncStatus> {
+  async updateSyncStatusChainHead (queryRunner: QueryRunner, blockHash: string, blockNumber: number, force = false): Promise<SyncStatus> {
     const repo = queryRunner.manager.getRepository(SyncStatus);
 
-    return this._baseDatabase.updateSyncStatusChainHead(repo, blockHash, blockNumber);
+    return this._baseDatabase.updateSyncStatusChainHead(repo, blockHash, blockNumber, force);
   }
 
   async getSyncStatus (queryRunner: QueryRunner): Promise<SyncStatus | undefined> {

packages/eden-watcher/src/fill.ts

@@ -39,6 +39,16 @@ export const main = async (): Promise<any> => {
       type: 'number',
       demandOption: true,
       describe: 'Block number to stop processing at'
+    },
+    prefetch: {
+      type: 'boolean',
+      default: false,
+      describe: 'Block and events prefetch mode'
+    },
+    batchBlocks: {
+      type: 'number',
+      default: 10,
+      describe: 'Number of blocks prefetched in batch'
     }
   }).argv;
 

packages/eden-watcher/src/indexer.ts

@@ -910,8 +910,8 @@ export class Indexer implements IndexerInterface {
     return this._baseIndexer.updateSyncStatusIndexedBlock(blockHash, blockNumber, force);
   }
 
-  async updateSyncStatusChainHead (blockHash: string, blockNumber: number): Promise<SyncStatus> {
+  async updateSyncStatusChainHead (blockHash: string, blockNumber: number, force = false): Promise<SyncStatus> {
-    return this._baseIndexer.updateSyncStatusChainHead(blockHash, blockNumber);
+    return this._baseIndexer.updateSyncStatusChainHead(blockHash, blockNumber, force);
   }
 
   async updateSyncStatusCanonicalBlock (blockHash: string, blockNumber: number, force = false): Promise<SyncStatus> {
@@ -1301,8 +1301,6 @@ export class Indexer implements IndexerInterface {
   async _fetchAndSaveEvents ({ cid: blockCid, blockHash }: DeepPartial<BlockProgress>): Promise<BlockProgress> {
     assert(blockHash);
 
-    console.time('time:indexer#_fetchAndSaveEvents-logs_txs');
-
     const logsPromise = this._ethClient.getLogs({ blockHash });
     const transactionsPromise = this._postgraphileClient.getBlockWithTransactions({ blockHash });
 
@@ -1321,8 +1319,6 @@ export class Indexer implements IndexerInterface {
       }
     ] = await Promise.all([logsPromise, transactionsPromise]);
 
-    console.timeEnd('time:indexer#_fetchAndSaveEvents-logs_txs');
-
     const transactionMap = transactions.reduce((acc: {[key: string]: any}, transaction: {[key: string]: any}) => {
       acc[transaction.txHash] = transaction;
       return acc;

packages/graph-test-watcher/src/cli/reset-cmds/state.ts

@@ -95,6 +95,8 @@ export const handler = async (argv: any): Promise<void> => {
       await indexer.updateHookStatusProcessedBlock(blockProgress.blockNumber, true);
     }
 
+    await indexer.updateSyncStatusChainHead(blockProgress.blockHash, blockProgress.blockNumber, true);
+
     dbTx.commitTransaction();
   } catch (error) {
     await dbTx.rollbackTransaction();

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

@@ -184,10 +184,10 @@ export class Database implements IPLDDatabaseInterface {
     return this._baseDatabase.updateSyncStatusCanonicalBlock(repo, blockHash, blockNumber, force);
   }
 
-  async updateSyncStatusChainHead (queryRunner: QueryRunner, blockHash: string, blockNumber: number): Promise<SyncStatus> {
+  async updateSyncStatusChainHead (queryRunner: QueryRunner, blockHash: string, blockNumber: number, force = false): Promise<SyncStatus> {
     const repo = queryRunner.manager.getRepository(SyncStatus);
 
-    return this._baseDatabase.updateSyncStatusChainHead(repo, blockHash, blockNumber);
+    return this._baseDatabase.updateSyncStatusChainHead(repo, blockHash, blockNumber, force);
   }
 
   async getSyncStatus (queryRunner: QueryRunner): Promise<SyncStatus | undefined> {

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

@@ -39,6 +39,16 @@ export const main = async (): Promise<any> => {
       type: 'number',
       demandOption: true,
       describe: 'Block number to stop processing at'
+    },
+    prefetch: {
+      type: 'boolean',
+      default: false,
+      describe: 'Block and events prefetch mode'
+    },
+    batchBlocks: {
+      type: 'number',
+      default: 10,
+      describe: 'Number of blocks prefetched in batch'
     }
   }).argv;
 

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

@@ -486,8 +486,8 @@ export class Indexer implements IndexerInterface {
     return this._baseIndexer.updateSyncStatusIndexedBlock(blockHash, blockNumber, force);
   }
 
-  async updateSyncStatusChainHead (blockHash: string, blockNumber: number): Promise<SyncStatus> {
+  async updateSyncStatusChainHead (blockHash: string, blockNumber: number, force = false): Promise<SyncStatus> {
-    return this._baseIndexer.updateSyncStatusChainHead(blockHash, blockNumber);
+    return this._baseIndexer.updateSyncStatusChainHead(blockHash, blockNumber, force);
   }
 
   async updateSyncStatusCanonicalBlock (blockHash: string, blockNumber: number, force = false): Promise<SyncStatus> {

packages/util/src/fill.ts

@@ -123,11 +123,11 @@ const prefetchBlocks = async (
     }
 
     const fetchBlockPromises = blocks.map(async block => {
-      const { blockHash, blockNumber, parentHash, timestamp } = block;
+      const { cid, blockHash, blockNumber, parentHash, timestamp } = block;
       const blockProgress = await indexer.getBlockProgress(blockHash);
 
       if (!blockProgress) {
-        await indexer.fetchBlockEvents({ blockHash, blockNumber, parentHash, blockTimestamp: timestamp });
+        await indexer.fetchBlockEvents({ cid, blockHash, blockNumber, parentHash, blockTimestamp: timestamp });
       }
     });