Timestamp block filters (subquery#1310)

* add block timestamp filter

* bug fixes

* bug fixes

* clean code and write test for timestamp filter

* move cron generation to SubqueryProject

* fix next timestamp log bug

* remove modifiedDataSources

.gitpod.dockerfile

@@ -0,0 +1 @@
+FROM gitpod/workspace-postgresql

.gitpod.yml

@@ -0,0 +1,11 @@
+# This configuration file was automatically generated by Gitpod.
+# Please adjust to your needs (see https://www.gitpod.io/docs/config-gitpod-file)
+# and commit this file to your remote git repository to share the goodness with others.
+image:
+  file: .gitpod.dockerfile
+
+tasks:
+  - init: yarn install && yarn run build
+
+
+

package.json

@@ -8,6 +8,7 @@
     "@actions/core": "^1.6.0",
     "@babel/preset-env": "^7.16.11",
     "@octokit/request": "^5.6.3",
+    "@types/cron-converter": "^1",
     "@types/node": "^14.18.10",
     "@typescript-eslint/eslint-plugin": "^5.10.2",
     "@typescript-eslint/parser": "^5.10.2",

packages/common-substrate/src/project/models.ts

@@ -42,6 +42,9 @@ export class BlockFilter implements SubstrateBlockFilter {
   @IsOptional()
   @IsInt()
   modulo?: number;
+  @IsOptional()
+  @IsString()
+  timestamp?: string;
 }
 
 export class EventFilter extends BlockFilter implements SubstrateEventFilter {

packages/node/package.json

@@ -33,6 +33,7 @@
     "@subql/x-merkle-mountain-range": "2.0.0-0.1.2",
     "@willsoto/nestjs-prometheus": "^4.4.0",
     "app-module-path": "^2.2.0",
+    "cron-converter": "^1.0.2",
     "dayjs": "^1.10.7",
     "eventemitter2": "^6.4.5",
     "fetch-h2": "3.0.2",

packages/node/src/configure/SubqueryProject.ts

@@ -1,6 +1,7 @@
 // Copyright 2020-2022 OnFinality Limited authors & contributors
 // SPDX-License-Identifier: Apache-2.0
 
+import { ApiPromise } from '@polkadot/api';
 import { RegisteredTypes } from '@polkadot/types/types';
 import {
   ReaderFactory,
@@ -18,19 +19,31 @@ import {
   SubstrateDataSource,
   FileType,
   ProjectManifestV1_0_0Impl,
+  SubstrateBlockFilter,
+  isRuntimeDs,
+  SubstrateHandlerKind,
 } from '@subql/common-substrate';
 import { buildSchemaFromString } from '@subql/utils';
+import Cron from 'cron-converter';
 import { GraphQLSchema } from 'graphql';
 import {
   getChainTypes,
   getProjectRoot,
   updateDataSourcesV0_2_0,
 } from '../utils/project';
+import { getBlockByHeight, getTimestamp } from '../utils/substrate';
 
 export type SubqlProjectDs = SubstrateDataSource & {
   mapping: SubstrateDataSource['mapping'] & { entryScript: string };
 };
 
+export type SubqlProjectBlockFilter = SubstrateBlockFilter & {
+  cronSchedule?: {
+    schedule: Cron.Seeker;
+    next: number;
+  };
+};
+
 export type SubqlProjectDsTemplate = Omit<SubqlProjectDs, 'startBlock'> & {
   name: string;
 };
@@ -228,3 +241,53 @@ async function loadProjectTemplates(
     }));
   }
 }
+
+// eslint-disable-next-line @typescript-eslint/require-await
+export async function generateTimestampReferenceForBlockFilters(
+  dataSources: SubqlProjectDs[],
+  api: ApiPromise,
+): Promise<SubqlProjectDs[]> {
+  const cron = new Cron();
+
+  dataSources = await Promise.all(
+    dataSources.map(async (ds) => {
+      if (isRuntimeDs(ds)) {
+        const startBlock = ds.startBlock ?? 1;
+        let block;
+        let timestampReference;
+
+        ds.mapping.handlers = await Promise.all(
+          ds.mapping.handlers.map(async (handler) => {
+            if (handler.kind === SubstrateHandlerKind.Block) {
+              if (handler.filter?.timestamp) {
+                if (!block) {
+                  block = await getBlockByHeight(api, startBlock);
+                  timestampReference = getTimestamp(block);
+                }
+                try {
+                  cron.fromString(handler.filter.timestamp);
+                } catch (e) {
+                  throw new Error(
+                    `Invalid Cron string: ${handler.filter.timestamp}`,
+                  );
+                }
+
+                const schedule = cron.schedule(timestampReference);
+                (handler.filter as SubqlProjectBlockFilter).cronSchedule = {
+                  schedule: schedule,
+                  get next() {
+                    return Date.parse(this.schedule.next().format());
+                  },
+                };
+              }
+            }
+            return handler;
+          }),
+        );
+      }
+      return ds;
+    }),
+  );
+
+  return dataSources;
+}

packages/node/src/indexer/fetch.service.ts

@@ -569,7 +569,9 @@ export class FetchService implements OnApplicationShutdown {
       const { _metadata: metaData } = dictionary;
 
       if (metaData.genesisHash !== this.api.genesisHash.toString()) {
-        logger.error('The dictionary that you have specified does not match the chain you are indexing, it will be ignored. Please update your project manifest to reference the correct dictionary');
+        logger.error(
+          'The dictionary that you have specified does not match the chain you are indexing, it will be ignored. Please update your project manifest to reference the correct dictionary',
+        );
         this.useDictionary = false;
         this.eventEmitter.emit(IndexerEvent.UsingDictionary, {
           value: Number(this.useDictionary),

packages/node/src/indexer/indexer.manager.ts

@@ -33,7 +33,11 @@ import {
   SubstrateExtrinsic,
 } from '@subql/types';
 import { Sequelize } from 'sequelize';
-import { SubqlProjectDs, SubqueryProject } from '../configure/SubqueryProject';
+import {
+  generateTimestampReferenceForBlockFilters,
+  SubqlProjectDs,
+  SubqueryProject,
+} from '../configure/SubqueryProject';
 import * as SubstrateUtil from '../utils/substrate';
 import { yargsOptions } from '../yargs';
 import { ApiService } from './api.service';
@@ -176,11 +180,13 @@ export class IndexerManager {
 
   async start(): Promise<void> {
     await this.projectService.init();
+    logger.info('indexer manager started');
   }
 
   private filterDataSources(nextProcessingHeight: number): SubqlProjectDs[] {
     let filteredDs: SubqlProjectDs[];
-    filteredDs = this.project.dataSources.filter(
+
+    filteredDs = this.projectService.dataSources.filter(
       (ds) => ds.startBlock <= nextProcessingHeight,
     );
 

packages/node/src/indexer/project.service.ts

@@ -18,7 +18,11 @@ import {
   getMetaDataInfo,
 } from '@subql/node-core';
 import { Sequelize } from 'sequelize';
-import { SubqlProjectDs, SubqueryProject } from '../configure/SubqueryProject';
+import {
+  generateTimestampReferenceForBlockFilters,
+  SubqlProjectDs,
+  SubqueryProject,
+} from '../configure/SubqueryProject';
 import { initDbSchema } from '../utils/project';
 import { ApiService } from './api.service';
 import { DsProcessorService } from './ds-processor.service';
@@ -55,6 +59,10 @@ export class ProjectService {
     return this._schema;
   }
 
+  get dataSources(): SubqlProjectDs[] {
+    return this.project.dataSources;
+  }
+
   get blockOffset(): number {
     return this._blockOffset;
   }
@@ -71,6 +79,10 @@ export class ProjectService {
     // Used to load assets into DS-processor, has to be done in any thread
     await this.dsProcessorService.validateProjectCustomDatasources();
     // Do extra work on main thread to setup stuff
+    this.project.dataSources = await generateTimestampReferenceForBlockFilters(
+      this.project.dataSources,
+      this.apiService.getApi(),
+    );
     if (isMainThread) {
       this._schema = await this.ensureProject();
       await this.initDbSchema();

packages/node/src/utils/substrate.test.ts

@@ -2,7 +2,9 @@
 // SPDX-License-Identifier: Apache-2.0
 
 import { ApiPromise, WsProvider } from '@polkadot/api';
-import { fetchBlocks } from './substrate';
+import Cron from 'cron-converter';
+import { SubqlProjectBlockFilter } from '../configure/SubqueryProject';
+import { fetchBlocks, filterBlock } from './substrate';
 
 const endpoint = 'wss://polkadot.api.onfinality.io/public-ws';
 
@@ -29,6 +31,45 @@ describe('substrate utils', () => {
     }
   });
 
+  it('filters blocks based on timestamp', async () => {
+    const cronString = '*/5 * * * *';
+    const cron = new Cron();
+    try {
+      cron.fromString(cronString);
+    } catch (e) {
+      throw new Error(`invalid cron expression: ${cronString}`);
+    }
+    const blocks = await fetchBlocks(api, 100000, 100100);
+    const reference = blocks[0].block.timestamp;
+    const schedule = cron.schedule(reference);
+    const filter: SubqlProjectBlockFilter = {
+      timestamp: cronString,
+      cronSchedule: {
+        schedule: schedule,
+        get next() {
+          return Date.parse(this.schedule.next().format());
+        },
+      },
+    };
+    const filteredBlocks = blocks.filter((block) => {
+      return filterBlock(block.block, filter) !== undefined;
+    });
+
+    expect(filteredBlocks).toHaveLength(2);
+  });
+
+  it('invalid timestamp throws error on cron creation', () => {
+    const cronString = 'invalid cron';
+    const cron = new Cron();
+    expect(() => {
+      try {
+        cron.fromString(cronString);
+      } catch (e) {
+        throw new Error(`invalid cron expression: ${cronString}`);
+      }
+    }).toThrow(Error);
+  });
+
   it.skip('when failed to fetch, log block height and re-throw error', async () => {
     //some large number of block height
     await expect(fetchBlocks(api, 100000000, 100000019)).rejects.toThrow(

packages/node/src/utils/substrate.ts

@@ -23,6 +23,7 @@ import {
   SubstrateExtrinsic,
 } from '@subql/types';
 import { last, merge, range } from 'lodash';
+import { SubqlProjectBlockFilter } from '../configure/SubqueryProject';
 import { BlockContent } from '../indexer/types';
 const logger = getLogger('fetch');
 const INTERVAL_THRESHOLD = BN_THOUSAND.div(BN_TWO);
@@ -41,7 +42,7 @@ export function wrapBlock(
   });
 }
 
-function getTimestamp({ block: { extrinsics } }: SignedBlock): Date {
+export function getTimestamp({ block: { extrinsics } }: SignedBlock): Date {
   for (const e of extrinsics) {
     const {
       method: { method, section },
@@ -123,6 +124,11 @@ export function filterBlock(
 ): SubstrateBlock | undefined {
   if (!filter) return block;
   if (!filterBlockModulo(block, filter)) return;
+  if (filter.timestamp) {
+    if (!filterBlockTimestamp(block, filter as SubqlProjectBlockFilter)) {
+      return;
+    }
+  }
   return filter.specVersion === undefined ||
     block.specVersion === undefined ||
     checkSpecRange(filter.specVersion, block.specVersion)
@@ -139,6 +145,30 @@ export function filterBlockModulo(
   return block.block.header.number.toNumber() % modulo === 0;
 }
 
+export function filterBlockTimestamp(
+  block: SubstrateBlock,
+  filter: SubqlProjectBlockFilter,
+): boolean {
+  const unixTimestamp = block.timestamp.getTime();
+  if (unixTimestamp > filter.cronSchedule.next) {
+    logger.info(
+      `Block with timestamp ${new Date(
+        unixTimestamp,
+      ).toString()} is about to be indexed`,
+    );
+    logger.info(
+      `Next block will be indexed at ${new Date(
+        filter.cronSchedule.next,
+      ).toString()}`,
+    );
+    filter.cronSchedule.schedule.prev();
+    return true;
+  } else {
+    filter.cronSchedule.schedule.prev();
+    return false;
+  }
+}
+
 export function filterExtrinsic(
   { block, extrinsic, success }: SubstrateExtrinsic,
   filter?: SubstrateCallFilter,
@@ -252,7 +282,7 @@ export async function fetchBlocks(
   });
 }
 
-async function getBlockByHeight(
+export async function getBlockByHeight(
   api: ApiPromise,
   height: number,
 ): Promise<SignedBlock> {

packages/types/src/project.ts

@@ -51,6 +51,7 @@ interface SubstrateBaseHandlerFilter {
 
 export interface SubstrateBlockFilter extends SubstrateBaseHandlerFilter {
   modulo?: number;
+  timestamp?: string;
 }
 
 export interface SubstrateEventFilter extends SubstrateBaseHandlerFilter {