Implement block filter for plural queries on subgraph entities (#444)

* Fix bigint values transformation

* Fix starts and ends filter operator resolution

* Support case insensitive filters for string fields

* Add support for global filter _change_block

* Handle _change_block filter in all query types

packages/graph-node/src/watcher.ts

@@ -27,7 +27,8 @@ import {
   getSubgraphConfig,
   Transaction,
   EthClient,
-  DEFAULT_LIMIT
+  DEFAULT_LIMIT,
+  FILTER_CHANGE_BLOCK
 } from '@cerc-io/util';
 
 import { Context, GraphData, instantiate } from './loader';
@@ -322,6 +323,17 @@ export class GraphWatcher {
 
     try {
       where = Object.entries(where).reduce((acc: { [key: string]: any }, [fieldWithSuffix, value]) => {
+        if (fieldWithSuffix === FILTER_CHANGE_BLOCK) {
+          assert(value.number_gte && typeof value.number_gte === 'number');
+
+          // Maintain util.Where type
+          acc[FILTER_CHANGE_BLOCK] = [{
+            value: value.number_gte
+          }];
+
+          return acc;
+        }
+
         const [field, ...suffix] = fieldWithSuffix.split('_');
 
         if (!acc[field]) {
@@ -345,6 +357,11 @@ export class GraphWatcher {
           filter.operator = operator;
         }
 
+        // If filter field ends with "nocase", use case insensitive version of the operator
+        if (suffix[suffix.length - 1] === 'nocase') {
+          filter.operator = `${operator}_nocase`;
+        }
+
         acc[field].push(filter);
 
         return acc;

packages/util/src/database.ts

@@ -35,7 +35,10 @@ export const OPERATOR_MAP = {
   in: 'IN',
   contains: 'LIKE',
   starts: 'LIKE',
-  ends: 'LIKE'
+  ends: 'LIKE',
+  contains_nocase: 'ILIKE',
+  starts_nocase: 'ILIKE',
+  ends_nocase: 'ILIKE'
 };
 
 const INSERT_EVENTS_BATCH = 100;
@@ -875,11 +878,11 @@ export class Database {
           }
         }
 
-        if (['contains', 'starts'].some(el => el === operator)) {
+        if (['contains', 'contains_nocase', 'ends', 'ends_nocase'].some(el => el === operator)) {
           value = `%${value}`;
         }
 
-        if (['contains', 'ends'].some(el => el === operator)) {
+        if (['contains', 'contains_nocase', 'starts', 'starts_nocase'].some(el => el === operator)) {
           value += '%';
         }
 
@@ -927,19 +930,22 @@ export class Database {
     eventCount.set(res);
   }
 
+  // TODO: Transform in the GQL type BigInt parsing itself
   _transformBigIntValues (value: any): any {
+    // Handle array of bigints
     if (Array.isArray(value)) {
       if (value.length > 0 && typeof value[0] === 'bigint') {
         return value.map(val => {
           return val.toString();
         });
       }
-
-      return value;
     }
 
+    // Handle bigint
     if (typeof value === 'bigint') {
       return value.toString();
     }
+
+    return value;
   }
 }

packages/util/src/graph/database.ts

@@ -31,6 +31,8 @@ import { fromStateEntityValues } from './state-utils';
 
 const log = debug('vulcanize:graph-database');
 
+export const FILTER_CHANGE_BLOCK = '_change_block';
+
 export const DEFAULT_LIMIT = 100;
 const DEFAULT_CLEAR_ENTITIES_CACHE_INTERVAL = 1000;
 
@@ -433,6 +435,11 @@ export class GraphDatabase {
       delete where.id;
     }
 
+    if (where[FILTER_CHANGE_BLOCK]) {
+      subQuery = subQuery.andWhere('subTable.block_number >= :changeBlockNumber', { changeBlockNumber: where[FILTER_CHANGE_BLOCK][0].value });
+      delete where[FILTER_CHANGE_BLOCK];
+    }
+
     if (block.hash) {
       const { canonicalBlockNumber, blockHashes } = await this._baseDatabase.getFrothyRegion(queryRunner, block.hash);
 
@@ -496,6 +503,11 @@ export class GraphDatabase {
       delete where.id;
     }
 
+    if (where[FILTER_CHANGE_BLOCK]) {
+      subQuery = subQuery.andWhere('subTable.block_number >= :changeBlockNumber', { changeBlockNumber: where[FILTER_CHANGE_BLOCK][0].value });
+      delete where[FILTER_CHANGE_BLOCK];
+    }
+
     if (block.hash) {
       const { canonicalBlockNumber, blockHashes } = await this._baseDatabase.getFrothyRegion(queryRunner, block.hash);
 
@@ -554,6 +566,11 @@ export class GraphDatabase {
       .addOrderBy(`${tableName}.block_number`, 'DESC')
       .limit(1);
 
+    if (where[FILTER_CHANGE_BLOCK]) {
+      selectQueryBuilder = selectQueryBuilder.andWhere(`${tableName}.block_number >= :changeBlockNumber`, { changeBlockNumber: where[FILTER_CHANGE_BLOCK][0].value });
+      delete where[FILTER_CHANGE_BLOCK];
+    }
+
     if (block.hash) {
       const { canonicalBlockNumber, blockHashes } = await this._baseDatabase.getFrothyRegion(queryRunner, block.hash);
 
@@ -588,6 +605,11 @@ export class GraphDatabase {
     let selectQueryBuilder = repo.createQueryBuilder(tableName)
       .where('is_pruned = :isPruned', { isPruned: false });
 
+    if (where[FILTER_CHANGE_BLOCK]) {
+      selectQueryBuilder = selectQueryBuilder.andWhere(`${tableName}.block_number >= :changeBlockNumber`, { changeBlockNumber: where[FILTER_CHANGE_BLOCK][0].value });
+      delete where[FILTER_CHANGE_BLOCK];
+    }
+
     if (block.hash) {
       const { canonicalBlockNumber, blockHashes } = await this._baseDatabase.getFrothyRegion(queryRunner, block.hash);
 
@@ -658,6 +680,11 @@ export class GraphDatabase {
         );
     }
 
+    if (where[FILTER_CHANGE_BLOCK]) {
+      selectQueryBuilder = selectQueryBuilder.andWhere('latest.block_number >= :changeBlockNumber', { changeBlockNumber: where[FILTER_CHANGE_BLOCK][0].value });
+      delete where[FILTER_CHANGE_BLOCK];
+    }
+
     selectQueryBuilder = this._baseDatabase.buildQuery(repo, selectQueryBuilder, where, 'latest');
 
     if (queryOptions.orderBy) {
@@ -694,6 +721,11 @@ export class GraphDatabase {
       .orderBy('subTable.block_number', 'DESC')
       .limit(1);
 
+    if (where[FILTER_CHANGE_BLOCK]) {
+      subQuery = subQuery.andWhere('subTable.block_number >= :changeBlockNumber', { changeBlockNumber: where[FILTER_CHANGE_BLOCK][0].value });
+      delete where[FILTER_CHANGE_BLOCK];
+    }
+
     if (block.hash) {
       const { canonicalBlockNumber, blockHashes } = await this._baseDatabase.getFrothyRegion(queryRunner, block.hash);