Implement subgraph store host API (#35)

* Implement store get api without blockHash and blockNumber

* Pass database instance to GraphWatcher

* Implement store set without block data

* Store blockHash and blockNumber in database entity table

* Implement getting entity in subgraph from store.get

* Add block data present in postgraphile

* Pass db and context to instantiate method in tests

* GQL API in graph-test-watcher to test store.set

* Remove contract address from subgraph file

* Fix block in dummy event data

* Pass just blockHash to get an entity from the database

* Review changes and add TODOs

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

packages/graph-node/src/call-handler.test.ts

@@ -3,17 +3,54 @@
 //
 
 import path from 'path';
+import chai, { assert, expect } from 'chai';
+import spies from 'chai-spies';
 
-import { getDummyEventData } from '../test/utils';
+import { getDummyEventData, getTestDatabase } from '../test/utils';
 import { instantiate } from './loader';
-import { createEvent } from './utils';
+import { createEvent, Block } from './utils';
+import { Database } from './database';
+
+chai.use(spies);
+
+const sandbox = chai.spy.sandbox();
 
 describe('call handler in mapping code', () => {
   let exports: any;
+  let db: Database;
+
+  const eventData = getDummyEventData();
+
+  before(async () => {
+    db = getTestDatabase();
+
+    sandbox.on(db, 'getEntity', (blockHash: string, entityString: string, idString: string) => {
+      assert(blockHash);
+      assert(entityString);
+      assert(idString);
+    });
+
+    sandbox.on(db, 'fromGraphEntity', async (instanceExports: any, block: Block, entity: string, entityInstance: any) => {
+      const entityFields = [
+        { type: 'varchar', propertyName: 'blockHash' },
+        { type: 'integer', propertyName: 'blockNumber' },
+        { type: 'bigint', propertyName: 'count' },
+        { type: 'varchar', propertyName: 'param1' },
+        { type: 'integer', propertyName: 'param2' }
+      ];
+
+      return db.getEntityValues(instanceExports, block, entityInstance, entityFields);
+    });
+
+    sandbox.on(db, 'saveEntity', (entity: string, data: any) => {
+      assert(entity);
+      assert(data);
+    });
+  });
 
   it('should load the subgraph example wasm', async () => {
     const filePath = path.resolve(__dirname, '../test/subgraph/example1/build/Example1/Example1.wasm');
-    const instance = await instantiate(filePath);
+    const instance = await instantiate(db, { event: { block: eventData.block } }, filePath);
     exports = instance.exports;
   });
 
@@ -27,8 +64,6 @@ describe('call handler in mapping code', () => {
     // TODO: Check api version https://github.com/graphprotocol/graph-node/blob/6098daa8955bdfac597cec87080af5449807e874/runtime/wasm/src/module/mod.rs#L533
     _start();
 
-    const eventData = getDummyEventData();
-
     // Create event params data.
     eventData.eventParams = [
       {
@@ -50,5 +85,13 @@ describe('call handler in mapping code', () => {
     const test = await createEvent(exports, contractAddress, eventData);
 
     await handleTest(test);
+
+    expect(db.getEntity).to.have.been.called();
+    expect(db.fromGraphEntity).to.have.been.called();
+    expect(db.saveEntity).to.have.been.called();
+  });
+
+  after(() => {
+    sandbox.restore();
   });
 });

packages/graph-node/src/database.ts

@@ -0,0 +1,136 @@
+//
+// Copyright 2021 Vulcanize, Inc.
+//
+
+import assert from 'assert';
+import {
+  Connection,
+  ConnectionOptions,
+  FindOneOptions
+} from 'typeorm';
+
+import {
+  Database as BaseDatabase
+} from '@vulcanize/util';
+
+import { Block, fromEntityValue, toEntityValue } from './utils';
+
+export class Database {
+  _config: ConnectionOptions
+  _conn!: Connection
+  _baseDatabase: BaseDatabase
+
+  constructor (config: ConnectionOptions, entitiesPath: string) {
+    assert(config);
+
+    this._config = {
+      name: 'subgraph',
+      ...config,
+      entities: [entitiesPath]
+    };
+
+    this._baseDatabase = new BaseDatabase(this._config);
+  }
+
+  async init (): Promise<void> {
+    this._conn = await this._baseDatabase.init();
+  }
+
+  async close (): Promise<void> {
+    return this._baseDatabase.close();
+  }
+
+  async getEntity (blockHash: string, entity: string, id: string): Promise<any> {
+    const queryRunner = this._conn.createQueryRunner();
+    const repo = queryRunner.manager.getRepository(entity);
+    const whereOptions: { [key: string]: any } = { id };
+
+    if (blockHash) {
+      whereOptions.blockHash = blockHash;
+    }
+
+    const findOptions = {
+      where: whereOptions,
+      order: {
+        blockNumber: 'DESC'
+      }
+    };
+
+    let entityData = await repo.findOne(findOptions as FindOneOptions<any>);
+
+    if (!entityData && findOptions.where.blockHash) {
+      entityData = await this._baseDatabase.getPrevEntityVersion(queryRunner, repo, findOptions);
+    }
+
+    return entityData;
+  }
+
+  async saveEntity (entity: string, data: any): Promise<void> {
+    const repo = this._conn.getRepository(entity);
+
+    const dbEntity: any = repo.create(data);
+    await repo.save(dbEntity);
+  }
+
+  async toGraphEntity (instanceExports: any, entity: string, data: any): Promise<any> {
+    // TODO: Cache schema/columns.
+    const repo = this._conn.getRepository(entity);
+    const entityFields = repo.metadata.columns;
+
+    const { Entity } = instanceExports;
+    const entityInstance = await Entity.__new();
+
+    const entityValuePromises = entityFields.filter(field => {
+      const { propertyName } = field;
+
+      // TODO: Will clash if entity has blockHash and blockNumber fields.
+      if (propertyName === 'blockHash' || propertyName === 'blockNumber') {
+        return false;
+      }
+
+      return true;
+    }).map(async (field) => {
+      const { type, propertyName } = field;
+
+      return toEntityValue(instanceExports, entityInstance, data, type.toString(), propertyName);
+    }, {});
+
+    await Promise.all(entityValuePromises);
+
+    return entityInstance;
+  }
+
+  async fromGraphEntity (instanceExports: any, block: Block, entity: string, entityInstance: any): Promise<{ [key: string]: any } > {
+    // TODO: Cache schema/columns.
+    const repo = this._conn.getRepository(entity);
+    const entityFields = repo.metadata.columns;
+
+    return this.getEntityValues(instanceExports, block, entityInstance, entityFields);
+  }
+
+  async getEntityValues (instanceExports: any, block: Block, entityInstance: any, entityFields: any): Promise<{ [key: string]: any } > {
+    const entityValuePromises = entityFields.map(async (field: any) => {
+      const { type, propertyName } = field;
+
+      // TODO: Will clash if entity has blockHash and blockNumber fields.
+      if (propertyName === 'blockHash') {
+        return block.blockHash;
+      }
+
+      if (propertyName === 'blockNumber') {
+        return block.blockNumber;
+      }
+
+      return fromEntityValue(instanceExports, entityInstance, type.toString(), propertyName);
+    }, {});
+
+    const entityValues = await Promise.all(entityValuePromises);
+
+    return entityFields.reduce((acc: { [key: string]: any }, field: any, index: number) => {
+      const { propertyName } = field;
+      acc[propertyName] = entityValues[index];
+
+      return acc;
+    }, {});
+  }
+}

packages/graph-node/src/eden.test.ts

@@ -5,19 +5,48 @@
 import assert from 'assert';
 import { ethers } from 'ethers';
 import path from 'path';
+import chai from 'chai';
+import spies from 'chai-spies';
 
 import { instantiate } from './loader';
-import { createEvent } from './utils';
+import { createEvent, Block } from './utils';
 import edenNetworkAbi from '../test/subgraph/eden/EdenNetwork/abis/EdenNetwork.json';
 import merkleDistributorAbi from '../test/subgraph/eden/EdenNetworkDistribution/abis/MerkleDistributor.json';
 import distributorGovernanceAbi from '../test/subgraph/eden/EdenNetworkGovernance/abis/DistributorGovernance.json';
-import { getDummyEventData } from '../test/utils';
+import { getDummyEventData, getTestDatabase } from '../test/utils';
+import { Database } from './database';
 
 const ZERO_ADDRESS = '0x0000000000000000000000000000000000000000';
 
-describe('eden wasm loader tests', () => {
+chai.use(spies);
+
+const sandbox = chai.spy.sandbox();
+
+describe('eden wasm loader tests', async () => {
+  let db: Database;
   const eventData = getDummyEventData();
 
+  before(async () => {
+    db = getTestDatabase();
+
+    sandbox.on(db, 'getEntity', (blockHash: string, entityString: string, idString: string) => {
+      assert(blockHash);
+      assert(entityString);
+      assert(idString);
+    });
+
+    sandbox.on(db, 'fromGraphEntity', async (instanceExports: any, block: Block, entity: string, entityInstance: any) => {
+      const entityFields: any = [];
+
+      return db.getEntityValues(instanceExports, block, entityInstance, entityFields);
+    });
+
+    sandbox.on(db, 'saveEntity', (entity: string, data: any) => {
+      assert(entity);
+      assert(data);
+    });
+  });
+
   describe('EdenNetwork wasm', () => {
     let exports: any;
 
@@ -36,7 +65,7 @@ describe('eden wasm loader tests', () => {
 
     it('should load the subgraph network wasm', async () => {
       const filePath = path.resolve(__dirname, '../test/subgraph/eden/EdenNetwork/EdenNetwork.wasm');
-      ({ exports } = await instantiate(filePath, data));
+      ({ exports } = await instantiate(db, { event: { block: eventData.block } }, filePath, data));
       const { _start } = exports;
       _start();
     });
@@ -195,7 +224,7 @@ describe('eden wasm loader tests', () => {
 
     it('should load the subgraph network distribution wasm', async () => {
       const filePath = path.resolve(__dirname, '../test/subgraph/eden/EdenNetworkDistribution/EdenNetworkDistribution.wasm');
-      ({ exports } = await instantiate(filePath, data));
+      ({ exports } = await instantiate(db, { event: { block: eventData.block } }, filePath, data));
       const { _start } = exports;
       _start();
     });
@@ -339,7 +368,7 @@ describe('eden wasm loader tests', () => {
 
     it('should load the subgraph network governance wasm', async () => {
       const filePath = path.resolve(__dirname, '../test/subgraph/eden/EdenNetworkGovernance/EdenNetworkGovernance.wasm');
-      ({ exports } = await instantiate(filePath, data));
+      ({ exports } = await instantiate(db, { event: { block: eventData.block } }, filePath, data));
       const { _start } = exports;
       _start();
     });
@@ -427,4 +456,8 @@ describe('eden wasm loader tests', () => {
       await rewardScheduleChanged(rewardScheduleChangedEvent);
     });
   });
+
+  after(() => {
+    sandbox.restore();
+  });
 });

packages/graph-node/src/eth-call.test.ts

@@ -7,9 +7,12 @@ import path from 'path';
 
 import { instantiate } from './loader';
 import exampleAbi from '../test/subgraph/example1/build/Example1/abis/Example1.json';
+import { getTestDatabase } from '../test/utils';
+import { Database } from './database';
 
 describe('eth-call wasm tests', () => {
   let exports: any;
+  let db: Database;
 
   const contractAddress = process.env.EXAMPLE_CONTRACT_ADDRESS;
   assert(contractAddress);
@@ -23,9 +26,13 @@ describe('eth-call wasm tests', () => {
     }
   };
 
+  before(async () => {
+    db = getTestDatabase();
+  });
+
   it('should load the subgraph example wasm', async () => {
     const filePath = path.resolve(__dirname, '../test/subgraph/example1/build/Example1/Example1.wasm');
-    const instance = await instantiate(filePath, data);
+    const instance = await instantiate(db, { event: {} }, filePath, data);
     exports = instance.exports;
   });
 

packages/graph-node/src/index.ts

@@ -1 +1,2 @@
 export * from './watcher';
+export * from './database';

packages/graph-node/src/loader.test.ts

@@ -6,15 +6,21 @@ import path from 'path';
 import { expect } from 'chai';
 
 import { instantiate } from './loader';
+import { getTestDatabase } from '../test/utils';
+import { Database } from './database';
 
 const WASM_FILE_PATH = '../build/debug.wasm';
 
 describe('wasm loader tests', () => {
   let exports: any;
+  let db: Database;
 
   before(async () => {
+    db = getTestDatabase();
+
     const filePath = path.resolve(__dirname, WASM_FILE_PATH);
-    const instance = await instantiate(filePath);
+    const instance = await instantiate(db, { event: {} }, filePath);
+
     exports = instance.exports;
   });
 

packages/graph-node/src/loader.ts

@@ -14,7 +14,8 @@ import {
 } from 'ethers';
 
 import { TypeId } from './types';
-import { fromEthereumValue, toEthereumValue } from './utils';
+import { Block, fromEthereumValue, toEthereumValue } from './utils';
+import { Database } from './database';
 
 const NETWORK_URL = 'http://127.0.0.1:8081';
 
@@ -29,7 +30,13 @@ interface GraphData {
   dataSource?: DataSource;
 }
 
-export const instantiate = async (filePath: string, data: GraphData = {}): Promise<loader.ResultObject & { exports: any }> => {
+export interface Context {
+  event: {
+    block?: Block
+  }
+}
+
+export const instantiate = async (database: Database, context: Context, filePath: string, data: GraphData = {}): Promise<loader.ResultObject & { exports: any }> => {
   const { abis = {}, dataSource } = data;
   const buffer = await fs.readFile(filePath);
   const provider = getDefaultProvider(NETWORK_URL);
@@ -37,31 +44,26 @@ export const instantiate = async (filePath: string, data: GraphData = {}): Promi
   const imports: WebAssembly.Imports = {
     index: {
       'store.get': async (entity: number, id: number) => {
-        console.log('store.get');
+        const entityName = __getString(entity);
+        const entityId = __getString(id);
 
-        const entityString = __getString(entity);
-        console.log('entity:', entityString);
-        const idString = __getString(id);
-        console.log('id:', idString);
+        assert(context.event.block);
+        const entityData = await database.getEntity(context.event.block.blockHash, entityName, entityId);
 
-        // TODO: Implement store get to fetch from DB using entity and id.
+        if (!entityData) {
+          return null;
+        }
 
-        // TODO: Fill entity with field values.
-        // return Entity.__new()
-        return null;
+        return database.toGraphEntity(exports, entityName, entityData);
       },
       'store.set': async (entity: number, id: number, data: number) => {
-        console.log('store.set');
+        const entityName = __getString(entity);
 
-        const entityString = __getString(entity);
-        console.log('entity:', entityString);
-        const idString = __getString(id);
-        console.log('id:', idString);
         const entityInstance = await Entity.wrap(data);
-        const entityInstanceId = __getString(await entityInstance.getString(await __newString('id')));
-        console.log('entity instance id:', entityInstanceId);
 
-        // TODO: Implement store set to save entity in db with values from entityInstance.
+        assert(context.event.block);
+        const dbData = await database.fromGraphEntity(exports, context.event.block, entityName, entityInstance);
+        await database.saveEntity(entityName, dbData);
       },
 
       'typeConversion.stringToH160': () => {

packages/graph-node/src/numbers.test.ts

@@ -6,15 +6,20 @@ import path from 'path';
 import { expect } from 'chai';
 
 import { instantiate } from './loader';
+import { getTestDatabase } from '../test/utils';
+import { Database } from './database';
 
 const EXAMPLE_WASM_FILE_PATH = '../test/subgraph/example1/build/Example1/Example1.wasm';
 
 describe('numbers wasm tests', () => {
   let exports: any;
+  let db: Database;
 
   before(async () => {
+    db = getTestDatabase();
+
     const filePath = path.resolve(__dirname, EXAMPLE_WASM_FILE_PATH);
-    const instance = await instantiate(filePath);
+    const instance = await instantiate(db, { event: {} }, filePath);
     exports = instance.exports;
     const { _start } = exports;
 

packages/graph-node/src/type-conversion.test.ts

@@ -6,15 +6,20 @@ import path from 'path';
 import { expect } from 'chai';
 
 import { instantiate } from './loader';
+import { getTestDatabase } from '../test/utils';
+import { Database } from './database';
 
 const EXAMPLE_WASM_FILE_PATH = '../test/subgraph/example1/build/Example1/Example1.wasm';
 
 describe('typeConversion wasm tests', () => {
   let exports: any;
+  let db: Database;
 
   before(async () => {
+    db = getTestDatabase();
+
     const filePath = path.resolve(__dirname, EXAMPLE_WASM_FILE_PATH);
-    const instance = await instantiate(filePath);
+    const instance = await instantiate(db, { event: {} }, filePath);
     exports = instance.exports;
     const { _start } = exports;
 

packages/graph-node/src/utils.ts

@@ -14,13 +14,6 @@ interface EventParam {
   kind: string;
 }
 
-interface Block {
-  hash: string;
-  number: number;
-  timestamp: number;
-  parentHash: string;
-}
-
 interface Transaction {
   hash: string;
   index: number;
@@ -28,6 +21,17 @@ interface Transaction {
   to: string;
 }
 
+export interface Block {
+  blockHash: string;
+  blockNumber: string;
+  timestamp: string;
+  parentHash: string;
+  stateRoot: string;
+  td: string;
+  txRoot: string;
+  receiptRoot: string;
+}
+
 export interface EventData {
   block: Block;
   tx: Transaction;
@@ -161,41 +165,48 @@ export const createEvent = async (instanceExports: any, contractAddress: string,
   } = instanceExports;
 
   // Fill block data.
-  const blockHashByteArray = await ByteArray.fromHexString(await __newString(blockData.hash));
+  const blockHashByteArray = await ByteArray.fromHexString(await __newString(blockData.blockHash));
   const blockHash = await Bytes.fromByteArray(blockHashByteArray);
 
   const parentHashByteArray = await ByteArray.fromHexString(await __newString(blockData.parentHash));
   const parentHash = await Bytes.fromByteArray(parentHashByteArray);
 
-  const blockNumber = await BigInt.fromI32(blockData.number);
+  const blockNumber = await BigInt.fromString(await __newString(blockData.blockNumber));
 
-  const blockTimestamp = await BigInt.fromI32(blockData.timestamp);
+  const blockTimestamp = await BigInt.fromString(await __newString(blockData.timestamp));
+
+  const stateRootByteArray = await ByteArray.fromHexString(await __newString(blockData.stateRoot));
+  const stateRoot = await Bytes.fromByteArray(stateRootByteArray);
+
+  const transactionsRootByteArray = await ByteArray.fromHexString(await __newString(blockData.txRoot));
+  const transactionsRoot = await Bytes.fromByteArray(transactionsRootByteArray);
+
+  const receiptsRootByteArray = await ByteArray.fromHexString(await __newString(blockData.receiptRoot));
+  const receiptsRoot = await Bytes.fromByteArray(receiptsRootByteArray);
+
+  const totalDifficulty = await BigInt.fromString(await __newString(blockData.td));
 
   // Missing fields from watcher in block data:
   // unclesHash
   // author
-  // stateRoot
-  // transactionsRoot
-  // receiptsRoot
   // gasUsed
   // gasLimit
   // difficulty
-  // totalDifficulty
   // size
   const block = await ethereum.Block.__new(
     blockHash,
     parentHash,
     await Bytes.empty(),
     await Address.zero(),
-    await Bytes.empty(),
-    await Bytes.empty(),
-    await Bytes.empty(),
+    stateRoot,
+    transactionsRoot,
+    receiptsRoot,
     blockNumber,
     await BigInt.fromI32(0),
     await BigInt.fromI32(0),
     blockTimestamp,
     await BigInt.fromI32(0),
-    await BigInt.fromI32(0),
+    totalDifficulty,
     null
   );
 
@@ -266,3 +277,55 @@ export const getSubgraphConfig = async (subgraphPath: string): Promise<any> => {
 
   return config;
 };
+
+export const toEntityValue = async (instanceExports: any, entityInstance: any, data: any, type: string, key: string) => {
+  const { __newString, BigInt: ExportBigInt } = instanceExports;
+  const entityKey = await __newString(key);
+  const value = data[key];
+
+  switch (type) {
+    case 'varchar': {
+      const entityValue = await __newString(value);
+
+      return entityInstance.setString(entityKey, entityValue);
+    }
+
+    case 'integer': {
+      return entityInstance.setI32(entityKey, value);
+    }
+
+    case 'bigint': {
+      const bigInt = await ExportBigInt.fromString(await __newString(value.toString()));
+
+      return entityInstance.setBigInt(entityKey, bigInt);
+    }
+
+    // TODO: Support more types.
+    default:
+      throw new Error(`Unsupported type: ${type}`);
+  }
+};
+
+export const fromEntityValue = async (instanceExports: any, entityInstance: any, type: string, key: string): Promise<any> => {
+  const { __newString, __getString, BigInt: ExportBigInt } = instanceExports;
+  const entityKey = await __newString(key);
+
+  switch (type) {
+    case 'varchar': {
+      return __getString(await entityInstance.getString(entityKey));
+    }
+
+    case 'integer': {
+      return entityInstance.getI32(entityKey);
+    }
+
+    case 'bigint': {
+      const bigInt = ExportBigInt.wrap(await entityInstance.getBigInt(entityKey));
+      return BigInt(__getString(await bigInt.toString()));
+    }
+
+    // TODO: Support more types.
+    default:
+      throw new Error(`Unsupported type: ${type}`);
+  }
+};

packages/graph-node/src/watcher.ts

@@ -9,9 +9,11 @@ import fs from 'fs';
 import { ContractInterface, utils } from 'ethers';
 
 import { ResultObject } from '@vulcanize/assemblyscript/lib/loader';
+import { EthClient } from '@vulcanize/ipld-eth-client';
 
 import { createEvent, getSubgraphConfig } from './utils';
-import { instantiate } from './loader';
+import { Context, instantiate } from './loader';
+import { Database } from './database';
 
 const log = debug('vulcanize:graph-watcher');
 
@@ -21,11 +23,19 @@ interface DataSource {
 }
 
 export class GraphWatcher {
+  _database: Database;
+  _postgraphileClient: EthClient;
   _subgraphPath: string;
   _dataSources: any[] = [];
   _dataSourceMap: { [key: string]: DataSource } = {};
 
-  constructor (subgraphPath: string) {
+  _context: Context = {
+    event: {}
+  }
+
+  constructor (database: Database, postgraphileClient: EthClient, subgraphPath: string) {
+    this._database = database;
+    this._postgraphileClient = postgraphileClient;
     this._subgraphPath = subgraphPath;
   }
 
@@ -58,7 +68,7 @@ export class GraphWatcher {
       const filePath = path.join(this._subgraphPath, file);
 
       return {
-        instance: await instantiate(filePath, data),
+        instance: await instantiate(this._database, this._context, filePath, data),
         contractInterface
       };
     }, {});
@@ -83,6 +93,16 @@ export class GraphWatcher {
   async handleEvent (eventData: any) {
     const { contract, event, eventSignature, block, tx, eventIndex } = eventData;
 
+    const {
+      allEthHeaderCids: {
+        nodes: [
+          blockData
+        ]
+      }
+    } = await this._postgraphileClient.getBlocks({ blockHash: block.hash });
+
+    this._context.event.block = blockData;
+
     // Get dataSource in subgraph yaml based on contract address.
     const dataSource = this._dataSources.find(dataSource => dataSource.source.address === contract);
 
@@ -113,7 +133,7 @@ export class GraphWatcher {
 
     const data = {
       eventParams: eventParams,
-      block,
+      block: blockData,
       tx,
       eventIndex
     };
@@ -123,4 +143,8 @@ export class GraphWatcher {
 
     await exports[eventHandler.handler](ethereumEvent);
   }
+
+  async getEntity (blockHash: string, entity: string, id: string): Promise<any> {
+    return this._database.getEntity(blockHash, entity, id);
+  }
 }