Handle template create in subgraph watcher during reorgs (#533)

* Handle template create events processing order during reorgs

* Add removeWatcher method in graph-node dummy indexer for test

* Apply GQL and RPC server middlewares ordered on requested paths

* Increment package version

* Remove console logs

---------

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

lerna.json

@@ -2,7 +2,7 @@
   "packages": [
     "packages/*"
   ],
-  "version": "0.2.107",
+  "version": "0.2.108",
   "npmClient": "yarn",
   "useWorkspaces": true,
   "command": {

packages/cache/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/cache",
-  "version": "0.2.107",
+  "version": "0.2.108",
   "description": "Generic object cache",
   "main": "dist/index.js",
   "scripts": {

packages/cli/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/cli",
-  "version": "0.2.107",
+  "version": "0.2.108",
   "main": "dist/index.js",
   "license": "AGPL-3.0",
   "scripts": {
@@ -15,13 +15,13 @@
   },
   "dependencies": {
     "@apollo/client": "^3.7.1",
-    "@cerc-io/cache": "^0.2.107",
+    "@cerc-io/cache": "^0.2.108",
-    "@cerc-io/ipld-eth-client": "^0.2.107",
+    "@cerc-io/ipld-eth-client": "^0.2.108",
     "@cerc-io/libp2p": "^0.42.2-laconic-0.1.4",
     "@cerc-io/nitro-node": "^0.1.15",
-    "@cerc-io/peer": "^0.2.107",
+    "@cerc-io/peer": "^0.2.108",
-    "@cerc-io/rpc-eth-client": "^0.2.107",
+    "@cerc-io/rpc-eth-client": "^0.2.108",
-    "@cerc-io/util": "^0.2.107",
+    "@cerc-io/util": "^0.2.108",
     "@ethersproject/providers": "^5.4.4",
     "@graphql-tools/utils": "^9.1.1",
     "@ipld/dag-cbor": "^8.0.0",

packages/codegen/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/codegen",
-  "version": "0.2.107",
+  "version": "0.2.108",
   "description": "Code generator",
   "private": true,
   "main": "index.js",
@@ -20,7 +20,7 @@
   },
   "homepage": "https://github.com/cerc-io/watcher-ts#readme",
   "dependencies": {
-    "@cerc-io/util": "^0.2.107",
+    "@cerc-io/util": "^0.2.108",
     "@graphql-tools/load-files": "^6.5.2",
     "@npmcli/package-json": "^5.0.0",
     "@poanet/solidity-flattener": "https://github.com/vulcanize/solidity-flattener.git",

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

@@ -635,6 +635,10 @@ export class Indexer implements IndexerInterface {
     return this._baseIndexer.watchContract(address, kind, checkpoint, startingBlock, context);
   }
 
+  async removeContract (address: string, kind: string): Promise<void> {
+    return this._baseIndexer.removeContract(address, kind);
+  }
+
   updateStateStatusMap (address: string, stateStatus: StateStatus): void {
     this._baseIndexer.updateStateStatusMap(address, stateStatus);
   }

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

@@ -41,12 +41,12 @@
   "homepage": "https://github.com/cerc-io/watcher-ts#readme",
   "dependencies": {
     "@apollo/client": "^3.3.19",
-    "@cerc-io/cli": "^0.2.107",
+    "@cerc-io/cli": "^0.2.108",
-    "@cerc-io/ipld-eth-client": "^0.2.107",
+    "@cerc-io/ipld-eth-client": "^0.2.108",
-    "@cerc-io/solidity-mapper": "^0.2.107",
+    "@cerc-io/solidity-mapper": "^0.2.108",
-    "@cerc-io/util": "^0.2.107",
+    "@cerc-io/util": "^0.2.108",
     {{#if (subgraphPath)}}
-    "@cerc-io/graph-node": "^0.2.107",
+    "@cerc-io/graph-node": "^0.2.108",
     {{/if}}
     "@ethersproject/providers": "^5.4.4",
     "debug": "^4.3.1",

packages/graph-node/package.json

@@ -1,10 +1,10 @@
 {
   "name": "@cerc-io/graph-node",
-  "version": "0.2.107",
+  "version": "0.2.108",
   "main": "dist/index.js",
   "license": "AGPL-3.0",
   "devDependencies": {
-    "@cerc-io/solidity-mapper": "^0.2.107",
+    "@cerc-io/solidity-mapper": "^0.2.108",
     "@ethersproject/providers": "^5.4.4",
     "@graphprotocol/graph-ts": "^0.22.0",
     "@nomiclabs/hardhat-ethers": "^2.0.2",
@@ -51,9 +51,9 @@
   "dependencies": {
     "@apollo/client": "^3.3.19",
     "@cerc-io/assemblyscript": "0.19.10-watcher-ts-0.1.2",
-    "@cerc-io/cache": "^0.2.107",
+    "@cerc-io/cache": "^0.2.108",
-    "@cerc-io/ipld-eth-client": "^0.2.107",
+    "@cerc-io/ipld-eth-client": "^0.2.108",
-    "@cerc-io/util": "^0.2.107",
+    "@cerc-io/util": "^0.2.108",
     "@types/json-diff": "^0.5.2",
     "@types/yargs": "^17.0.0",
     "bn.js": "^4.11.9",

packages/graph-node/src/loader.ts

@@ -734,25 +734,13 @@ export const instantiate = async (
         return __newString(dataSource.network);
       },
       'dataSource.create': async (name: number, params: number) => {
-        const [addressStringPtr] = __getArray(params);
+        await handleDataSourceCreate(name, params);
-        const addressString = __getString(addressStringPtr);
-        const contractKind = __getString(name);
-
-        assert(indexer.watchContract);
-        assert(context.block);
-        await indexer.watchContract(utils.getAddress(addressString), contractKind, true, Number(context.block.blockNumber));
       },
       'dataSource.createWithContext': async (name: number, params: number, dataSourceContext: number) => {
-        const [addressStringPtr] = __getArray(params);
-        const addressString = __getString(addressStringPtr);
-        const contractKind = __getString(name);
-
         const contextInstance = await Entity.wrap(dataSourceContext);
         const dbData = await database.fromGraphContext(instanceExports, contextInstance);
 
-        assert(indexer.watchContract);
+        await handleDataSourceCreate(name, params, dbData);
-        assert(context.block);
-        await indexer.watchContract(utils.getAddress(addressString), contractKind, true, Number(context.block.blockNumber), dbData);
       }
     },
     json: {
@@ -786,6 +774,34 @@ export const instantiate = async (
     }
   };
 
+  const handleDataSourceCreate = async (name: number, params: number, dbData?: {[key: string]: any}) => {
+    const [addressStringPtr] = __getArray(params);
+    const addressString = __getString(addressStringPtr);
+    const contractKind = __getString(name);
+
+    assert(context.block);
+    const contractAddress = utils.getAddress(addressString);
+    const watchedContracts = indexer.isContractAddressWatched(contractAddress);
+
+    // If template contract is already watched (incase of reorgs)
+    // Remove from watched contracts and throw error to reprocess block with correct order of template contract events
+    if (
+      watchedContracts &&
+      watchedContracts.some(watchedContract => watchedContract.kind === contractKind)
+    ) {
+      await indexer.removeContract(contractAddress, contractKind);
+      throw new Error(`Template contract ${contractAddress} of kind ${contractKind} already exists; removed from watched contracts`);
+    }
+
+    await indexer.watchContract(
+      contractAddress,
+      contractKind,
+      true,
+      Number(context.block.blockNumber),
+      dbData
+    );
+  };
+
   const instance = await loader.instantiate(source, imports);
   const { exports: instanceExports } = instance;
 

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

@@ -276,6 +276,10 @@ export class Indexer implements IndexerInterface {
     return undefined;
   }
 
+  async removeContract (address: string, kind: string): Promise<void> {
+    return undefined;
+  }
+
   async processBlock (blockProgress: BlockProgressInterface): Promise<void> {
     return undefined;
   }

packages/ipld-eth-client/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/ipld-eth-client",
-  "version": "0.2.107",
+  "version": "0.2.108",
   "description": "IPLD ETH Client",
   "main": "dist/index.js",
   "scripts": {
@@ -20,8 +20,8 @@
   "homepage": "https://github.com/cerc-io/watcher-ts#readme",
   "dependencies": {
     "@apollo/client": "^3.7.1",
-    "@cerc-io/cache": "^0.2.107",
+    "@cerc-io/cache": "^0.2.108",
-    "@cerc-io/util": "^0.2.107",
+    "@cerc-io/util": "^0.2.108",
     "cross-fetch": "^3.1.4",
     "debug": "^4.3.1",
     "ethers": "^5.4.4",

packages/peer/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/peer",
-  "version": "0.2.107",
+  "version": "0.2.108",
   "description": "libp2p module",
   "main": "dist/index.js",
   "exports": "./dist/index.js",

packages/rpc-eth-client/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/rpc-eth-client",
-  "version": "0.2.107",
+  "version": "0.2.108",
   "description": "RPC ETH Client",
   "main": "dist/index.js",
   "scripts": {
@@ -19,9 +19,9 @@
   },
   "homepage": "https://github.com/cerc-io/watcher-ts#readme",
   "dependencies": {
-    "@cerc-io/cache": "^0.2.107",
+    "@cerc-io/cache": "^0.2.108",
-    "@cerc-io/ipld-eth-client": "^0.2.107",
+    "@cerc-io/ipld-eth-client": "^0.2.108",
-    "@cerc-io/util": "^0.2.107",
+    "@cerc-io/util": "^0.2.108",
     "chai": "^4.3.4",
     "ethers": "^5.4.4",
     "left-pad": "^1.3.0",

packages/solidity-mapper/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/solidity-mapper",
-  "version": "0.2.107",
+  "version": "0.2.108",
   "main": "dist/index.js",
   "license": "AGPL-3.0",
   "devDependencies": {

packages/test/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/test",
-  "version": "0.2.107",
+  "version": "0.2.108",
   "main": "dist/index.js",
   "license": "AGPL-3.0",
   "private": true,

packages/tracing-client/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@cerc-io/tracing-client",
-  "version": "0.2.107",
+  "version": "0.2.108",
   "description": "ETH VM tracing client",
   "main": "dist/index.js",
   "scripts": {

packages/util/package.json

@@ -1,13 +1,13 @@
 {
   "name": "@cerc-io/util",
-  "version": "0.2.107",
+  "version": "0.2.108",
   "main": "dist/index.js",
   "license": "AGPL-3.0",
   "dependencies": {
     "@apollo/utils.keyvaluecache": "^1.0.1",
     "@cerc-io/nitro-node": "^0.1.15",
-    "@cerc-io/peer": "^0.2.107",
+    "@cerc-io/peer": "^0.2.108",
-    "@cerc-io/solidity-mapper": "^0.2.107",
+    "@cerc-io/solidity-mapper": "^0.2.108",
     "@cerc-io/ts-channel": "1.0.3-ts-nitro-0.1.1",
     "@ethersproject/properties": "^5.7.0",
     "@ethersproject/providers": "^5.4.4",
@@ -55,7 +55,7 @@
     "yargs": "^17.0.1"
   },
   "devDependencies": {
-    "@cerc-io/cache": "^0.2.107",
+    "@cerc-io/cache": "^0.2.108",
     "@nomiclabs/hardhat-waffle": "^2.0.1",
     "@types/bunyan": "^1.8.8",
     "@types/express": "^4.17.14",

packages/util/src/indexer.ts

@@ -920,6 +920,22 @@ export class Indexer {
     }
   }
 
+  async removeContract (address: string, kind: string): Promise<void> {
+    const dbTx = await this._db.createTransactionRunner();
+
+    try {
+      await this._db.deleteEntitiesByConditions(dbTx, 'contract', { kind, address });
+      this._clearWatchedContracts(
+        watchedContract => watchedContract.kind === kind && watchedContract.address === address
+      );
+    } catch (error) {
+      await dbTx.rollbackTransaction();
+      throw error;
+    } finally {
+      await dbTx.release();
+    }
+  }
+
   cacheContract (contract: ContractInterface): void {
     if (!this._watchedContractsByAddressMap[contract.address]) {
       this._watchedContractsByAddressMap[contract.address] = [];

packages/util/src/server.ts

@@ -97,23 +97,32 @@ export const createAndStartServer = async (
 
   await server.start();
 
-  server.applyMiddleware({
-    app,
-    path: gqlPath
-  });
-
   const rpcPath = serverConfig.ethRPC?.path ?? DEFAULT_ETH_RPC_PATH;
+  const rpcEnabled = serverConfig.ethRPC?.enabled;
+
+  // Apply GraphQL middleware
+  const applyGraphQLMiddleware = () => {
+    server.applyMiddleware({
+      app,
+      path: gqlPath
+    });
+  };
+
+  // Apply RPC middleware
+  const applyRPCMiddleware = () => {
+    if (!rpcEnabled) {
+      return;
+    }
 
-  if (serverConfig.ethRPC?.enabled) {
     // Create a JSON-RPC server to handle ETH RPC calls
     const rpcServer = jayson.Server(ethRPCHandlers);
 
-    // Mount the JSON-RPC server to ETH_RPC_PATH
+    // Mount the JSON-RPC server to rpcPath
     app.use(
       rpcPath,
       jsonParser(),
       (req: any, res: any, next: () => void) => {
-        // Convert all GET requests to POST to avoid getting rejected from jayson server middleware
+        // Convert all GET requests to POST to avoid getting rejected by jayson server middleware
         if (jayson.Utils.isMethod(req, 'GET')) {
           req.method = 'POST';
         }
@@ -121,15 +130,32 @@ export const createAndStartServer = async (
       },
       rpcServer.middleware()
     );
+  };
+
+  // Apply middlewares based on path specificity
+  if (isPathMoreSpecific(rpcPath, gqlPath)) {
+    applyRPCMiddleware();
+    applyGraphQLMiddleware();
+  } else {
+    applyGraphQLMiddleware();
+    applyRPCMiddleware();
   }
 
   httpServer.listen(port, host, () => {
-    log(`GQL server is listening on http://${host}:${port}${server.graphqlPath}`);
+    log(`GQL server is listening on http://${host}:${port}${gqlPath}`);
 
-    if (serverConfig.ethRPC?.enabled) {
+    if (rpcEnabled) {
       log(`ETH JSON RPC server is listening on http://${host}:${port}${rpcPath}`);
     }
   });
 
   return server;
 };
+
+// Determine which path is more specific (more segments)
+function isPathMoreSpecific (path1: string, path2: string) {
+  const path1Segments = path1.split('/').filter(segment => segment !== '');
+  const path2Segments = path2.split('/').filter(segment => segment !== '');
+
+  return path1Segments.length > path2Segments.length;
+}

packages/util/src/types.ts

@@ -217,6 +217,7 @@ export interface IndexerInterface {
   addContracts?: () => Promise<void>
   cacheContract: (contract: ContractInterface) => void;
   watchContract: (address: string, kind: string, checkpoint: boolean, startingBlock: number, context?: any) => Promise<void>
+  removeContract: (address: string, kind: string) => Promise<void>;
   getEntityTypesMap?: () => Map<string, { [key: string]: string }>
   getRelationsMap?: () => Map<any, { [key: string]: any }>
   processInitialState: (contractAddress: string, blockHash: string) => Promise<any>