Implement remaining graph-node host APIs for templates (#475)

* Implement createWithContext graph-node dataSource host API

* Parse individual context entries

* Implement graph-node dataSource context host API

* Handle array type fields and refactor code

* Resolve all values when parsing an array field

* Fix wasm entity creation with array type fields

* Required codegen changes

packages/codegen/src/data/entities/Contract.yaml

@@ -26,6 +26,13 @@ columns:
     pgType: integer
     tsType: number
     columnType: Column
+  - name: context
+    pgType: jsonb
+    tsType: 'Record<string, { data: any, type: number }>'
+    columnType: Column
+    columnOptions:
+      - option: nullable
+        value: true
 imports:
   - toImport:
       - Entity

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

@@ -229,10 +229,10 @@ export class Database implements DatabaseInterface {
     return this._baseDatabase.saveBlockProgress(repo, block);
   }
 
-  async saveContract (queryRunner: QueryRunner, address: string, kind: string, checkpoint: boolean, startingBlock: number): Promise<Contract> {
+  async saveContract (queryRunner: QueryRunner, address: string, kind: string, checkpoint: boolean, startingBlock: number, context?: any): Promise<Contract> {
     const repo = queryRunner.manager.getRepository(Contract);
 
-    return this._baseDatabase.saveContract(repo, address, kind, checkpoint, startingBlock);
+    return this._baseDatabase.saveContract(repo, address, kind, checkpoint, startingBlock, context);
   }
 
   async updateSyncStatusIndexedBlock (queryRunner: QueryRunner, blockHash: string, blockNumber: number, force = false): Promise<SyncStatus> {

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

@@ -32,7 +32,7 @@ export class {{className}} {{~#if implements}} implements {{implements}} {{~/if}
   {{~#unless @last}},{{/unless}}
   {{~/each}} }
   {{~/if}})
-    {{column.name}}!: {{column.tsType}};
+    {{column.name}}!: {{{column.tsType}}};
   {{~#unless @last}}
 
   {{/unless}}

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

@@ -595,8 +595,8 @@ export class Indexer implements IndexerInterface {
   }
 
   {{/if}}
-  async watchContract (address: string, kind: string, checkpoint: boolean, startingBlock: number): Promise<void> {
-    return this._baseIndexer.watchContract(address, kind, checkpoint, startingBlock);
+  async watchContract (address: string, kind: string, checkpoint: boolean, startingBlock: number, context?: any): Promise<void> {
+    return this._baseIndexer.watchContract(address, kind, checkpoint, startingBlock, context);
   }
 
   updateStateStatusMap (address: string, stateStatus: StateStatus): void {

packages/graph-node/src/loader.ts

@@ -101,7 +101,7 @@ export const instantiate = async (
         const dbEntity = await database.saveEntity(entityName, dbData);
         database.cacheUpdatedEntityByName(entityName, dbEntity);
 
-        // Update the in-memory subgraph state enabled
+        // Update the in-memory subgraph state if enabled
         if (indexer.serverConfig.enableState) {
           // Prepare diff data for the entity update
           assert(indexer.getRelationsMap);
@@ -698,10 +698,14 @@ export const instantiate = async (
         return Address.fromString(addressStringPtr);
       },
       'dataSource.context': async () => {
-        // TODO: Implement use in data source templates.
-        // https://thegraph.com/docs/en/developer/create-subgraph-hosted/#data-source-context
+        assert(context.contractAddress);
+        const contract = indexer.isWatchedContract(context.contractAddress);
 
-        return Entity.__new();
+        if (!contract) {
+          return null;
+        }
+
+        return database.toGraphContext(instanceExports, contract.context);
       },
       'dataSource.network': async () => {
         assert(dataSource);
@@ -715,6 +719,18 @@ export const instantiate = async (
         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);
+        assert(context.block);
+        await indexer.watchContract(utils.getAddress(addressString), contractKind, true, Number(context.block.blockNumber), dbData);
       }
     },
     json: {

packages/util/src/database.ts

@@ -601,13 +601,13 @@ export class Database {
       .getMany();
   }
 
-  async saveContract (repo: Repository<ContractInterface>, address: string, kind: string, checkpoint: boolean, startingBlock: number): Promise<ContractInterface> {
+  async saveContract (repo: Repository<ContractInterface>, address: string, kind: string, checkpoint: boolean, startingBlock: number, context?: any): Promise<ContractInterface> {
     const contract = await repo
       .createQueryBuilder()
       .where('address = :address', { address })
       .getOne();
 
-    const entity = repo.create({ address, kind, checkpoint, startingBlock });
+    const entity = repo.create({ address, kind, checkpoint, startingBlock, context });
 
     // If contract already present, overwrite fields.
     if (contract) {

packages/util/src/graph/database.ts

@@ -20,16 +20,20 @@ import { RawSqlResultsToEntityTransformer } from 'typeorm/query-builder/transfor
 import { SelectionNode } from 'graphql';
 import _ from 'lodash';
 import debug from 'debug';
+import JSONbig from 'json-bigint';
 
 import { Database as BaseDatabase, QueryOptions, Where, CanonicalBlockHeight } from '../database';
 import { BlockProgressInterface } from '../types';
 import { cachePrunedEntitiesCount, eventProcessingLoadEntityCacheHitCount, eventProcessingLoadEntityCount, eventProcessingLoadEntityDBQueryDuration } from '../metrics';
 import { ServerConfig } from '../config';
-import { Block, fromEntityValue, getLatestEntityFromEntity, resolveEntityFieldConflicts, toEntityValue } from './utils';
+import { Block, formatValue, fromEntityValue, getLatestEntityFromEntity, parseEntityValue, resolveEntityFieldConflicts, toEntityValue } from './utils';
 import { fromStateEntityValues } from './state-utils';
+import { ValueKind } from './types';
 
 const log = debug('vulcanize:graph-database');
 
+const JSONbigNative = JSONbig({ useNativeBigInt: true });
+
 export const FILTER_CHANGE_BLOCK = '_change_block';
 
 export const DEFAULT_LIMIT = 100;
@@ -953,6 +957,50 @@ export class GraphDatabase {
     return entityInstance;
   }
 
+  async toGraphContext (instanceExports: any, contextData: any): Promise<any> {
+    const { Entity } = instanceExports;
+    const contextInstance = await Entity.__new();
+
+    const { __newString } = instanceExports;
+    const contextValuePromises = Object.entries(contextData as Record<string, { type: ValueKind, data: any }>).map(async ([key, { type, data }]) => {
+      const contextKey = await __newString(key);
+
+      const value = JSONbigNative.parse(data);
+      const contextValue = await formatValue(instanceExports, type, value);
+
+      return contextInstance.set(contextKey, contextValue);
+    });
+
+    await Promise.all(contextValuePromises);
+
+    return contextInstance;
+  }
+
+  async fromGraphContext (instanceExports: any, contextInstance: any): Promise<{ [key: string]: any }> {
+    const { __getString, __getArray, ValueTypedMapEntry } = instanceExports;
+
+    const contextInstanceEntries = __getArray(await contextInstance.entries);
+    const contextValuePromises = contextInstanceEntries.map(async (entryPtr: any) => {
+      const entry = await ValueTypedMapEntry.wrap(entryPtr);
+      const contextKeyPtr = await entry.key;
+      const contextValuePtr = await entry.value;
+
+      const key = await __getString(contextKeyPtr);
+      const parsedValue = await parseEntityValue(instanceExports, contextValuePtr);
+
+      return { key, ...parsedValue };
+    });
+
+    const contextValues = await Promise.all(contextValuePromises);
+
+    return contextValues.reduce((acc: { [key: string]: any }, contextValue: any) => {
+      const { key, type, data } = contextValue;
+      acc[key] = { type, data: JSONbigNative.stringify(data) };
+
+      return acc;
+    }, {});
+  }
+
   async fromGraphEntity (instanceExports: any, block: Block, entityName: string, entityInstance: any): Promise<{ [key: string]: any }> {
     // TODO: Cache schema/columns.
     const repo = this._conn.getRepository(entityName);
@@ -981,10 +1029,17 @@ export class GraphDatabase {
 
       // Get blockNumber as _blockNumber and blockHash as _blockHash from the entityInstance (wasm).
       if (['_blockNumber', '_blockHash'].includes(propertyName)) {
-        return fromEntityValue(instanceExports, entityInstance, propertyName.slice(1));
+        const entityValue = await fromEntityValue(instanceExports, entityInstance, propertyName.slice(1));
+        return entityValue.data;
       }
 
-      return fromEntityValue(instanceExports, entityInstance, propertyName);
+      const entityValue = await fromEntityValue(instanceExports, entityInstance, propertyName);
+
+      if (entityValue.type === ValueKind.ARRAY) {
+        return entityValue.data.map((el: any) => el.data);
+      }
+
+      return entityValue.data;
     }, {});
 
     const entityValues = await Promise.all(entityValuePromises);

packages/util/src/graph/types.ts

@@ -83,6 +83,17 @@ export enum ValueKind {
   BIGINT = 7,
 }
 
+export const TypeNameToValueKind: Record<string, ValueKind> = {
+  String: ValueKind.STRING,
+  Int: ValueKind.INT,
+  BigDecimal: ValueKind.BIGDECIMAL,
+  Boolean: ValueKind.BOOL,
+  Array: ValueKind.ARRAY,
+  Null: ValueKind.NULL,
+  Bytes: ValueKind.BYTES,
+  BigInt: ValueKind.BIGINT
+};
+
 export enum Level {
   CRITICAL = 0,
   ERROR = 1,

packages/util/src/graph/utils.ts

@@ -11,7 +11,7 @@ import _ from 'lodash';
 import { MappingKey, StorageLayout } from '@cerc-io/solidity-mapper';
 
 import { GraphDecimal } from './graph-decimal';
-import { EthereumValueKind, TypeId, ValueKind } from './types';
+import { EthereumValueKind, TypeId, TypeNameToValueKind, ValueKind } from './types';
 
 const log = debug('vulcanize:utils');
 
@@ -539,12 +539,10 @@ export const getSubgraphConfig = async (subgraphPath: string): Promise<any> => {
 };
 
 export const toEntityValue = async (instanceExports: any, entityInstance: any, data: any, field: ColumnMetadata, type: string): Promise<any> => {
-  const { __newString, Value } = instanceExports;
+  const { __newString } = instanceExports;
   const { isArray, propertyName, isNullable } = field;
 
   const entityKey = await __newString(propertyName);
-  const entityValuePtr = await entityInstance.get(entityKey);
-  const subgraphValue = Value.wrap(entityValuePtr);
   const value = data[propertyName];
 
   // Check if the entity property is nullable.
@@ -553,12 +551,12 @@ export const toEntityValue = async (instanceExports: any, entityInstance: any, d
     return value;
   }
 
-  const entityValue = await formatEntityValue(instanceExports, subgraphValue, type, value, isArray);
+  const entityValue = await formatEntityValue(instanceExports, type, value, isArray);
 
   return entityInstance.set(entityKey, entityValue);
 };
 
-export const fromEntityValue = async (instanceExports: any, entityInstance: any, key: string): Promise<any> => {
+export const fromEntityValue = async (instanceExports: any, entityInstance: any, key: string): Promise<{ type: ValueKind, data: any }> => {
   const { __newString } = instanceExports;
   const entityKey = await __newString(key);
   const entityValuePtr = await entityInstance.get(entityKey);
@@ -566,7 +564,7 @@ export const fromEntityValue = async (instanceExports: any, entityInstance: any,
   return parseEntityValue(instanceExports, entityValuePtr);
 };
 
-const parseEntityValue = async (instanceExports: any, valuePtr: number) => {
+export const parseEntityValue = async (instanceExports: any, valuePtr: number): Promise<{ type: ValueKind, data: any }> => {
   const {
     __getString,
     __getArray,
@@ -582,7 +580,7 @@ const parseEntityValue = async (instanceExports: any, valuePtr: number) => {
   switch (kind) {
     case ValueKind.STRING: {
       const stringValue = await value.toString();
-      return __getString(stringValue);
+      return { type: kind, data: __getString(stringValue) };
     }
 
     case ValueKind.BYTES: {
@@ -590,17 +588,17 @@ const parseEntityValue = async (instanceExports: any, valuePtr: number) => {
       const bytes = await Bytes.wrap(bytesPtr);
       const bytesStringPtr = await bytes.toHexString();
 
-      return __getString(bytesStringPtr);
+      return { type: kind, data: __getString(bytesStringPtr) };
     }
 
     case ValueKind.BOOL: {
       const bool = await value.toBoolean();
-
-      return Boolean(bool);
+      return { type: kind, data: Boolean(bool) };
     }
 
     case ValueKind.INT: {
-      return value.toI32();
+      const data = await value.toI32();
+      return { type: kind, data };
     }
 
     case ValueKind.BIGINT: {
@@ -609,7 +607,7 @@ const parseEntityValue = async (instanceExports: any, valuePtr: number) => {
       const bigIntStringPtr = await bigInt.toString();
       const bigIntString = __getString(bigIntStringPtr);
 
-      return BigInt(bigIntString);
+      return { type: kind, data: BigInt(bigIntString) };
     }
 
     case ValueKind.BIGDECIMAL: {
@@ -617,19 +615,22 @@ const parseEntityValue = async (instanceExports: any, valuePtr: number) => {
       const bigDecimal = BigDecimal.wrap(bigDecimalPtr);
       const bigDecimalStringPtr = await bigDecimal.toString();
 
-      return new GraphDecimal(__getString(bigDecimalStringPtr)).toFixed();
+      return { type: kind, data: new GraphDecimal(__getString(bigDecimalStringPtr)).toFixed() };
     }
 
     case ValueKind.ARRAY: {
       const arrayPtr = await value.toArray();
       const arr = await __getArray(arrayPtr);
-      const arrDataPromises = arr.map((arrValuePtr: any) => parseEntityValue(instanceExports, arrValuePtr));
+      const arrDataPromises = arr.map(async (arrValuePtr: any) => {
+        return parseEntityValue(instanceExports, arrValuePtr);
+      });
+      const data = await Promise.all(arrDataPromises);
 
-      return Promise.all(arrDataPromises);
+      return { type: kind, data };
     }
 
     case ValueKind.NULL: {
-      return null;
+      return { type: kind, data: null };
     }
 
     default:
@@ -637,49 +638,50 @@ const parseEntityValue = async (instanceExports: any, valuePtr: number) => {
   }
 };
 
-const formatEntityValue = async (instanceExports: any, subgraphValue: any, type: string, value: any, isArray: boolean): Promise<any> => {
-  const { __newString, __newArray, BigInt: ASBigInt, Value, ByteArray, Bytes, BigDecimal, id_of_type: getIdOfType } = instanceExports;
-
+export const formatEntityValue = async (instanceExports: any, type: string, value: any, isArray: boolean): Promise<any> => {
+  let valueToFormat = value;
+  let typeName = type;
   if (isArray) {
-    const dataArrayPromises = value.map((el: any) => formatEntityValue(instanceExports, subgraphValue, type, el, false));
-    const dataArray = await Promise.all(dataArrayPromises);
-    const arrayStoreValueId = await getIdOfType(TypeId.ArrayStoreValue);
-    const valueArray = await __newArray(arrayStoreValueId, dataArray);
-
-    return Value.fromArray(valueArray);
+    typeName = 'Array';
+    valueToFormat = value.map((el: any) => { return { type: TypeNameToValueKind[type] ?? ValueKind.STRING, data: el }; });
   }
 
-  switch (type) {
-    case 'ID':
-    case 'String': {
+  return formatValue(instanceExports, TypeNameToValueKind[typeName] ?? ValueKind.STRING, valueToFormat);
+};
+
+export const formatValue = async (instanceExports: any, kind: ValueKind, value: any): Promise<any> => {
+  const { __newString, __newArray, BigInt: ASBigInt, Value, ByteArray, Bytes, BigDecimal, id_of_type: getIdOfType } = instanceExports;
+
+  switch (kind) {
+    case ValueKind.STRING: {
       const entityValue = await __newString(value);
 
       return Value.fromString(entityValue);
     }
 
-    case 'Boolean': {
+    case ValueKind.BOOL: {
       return Value.fromBoolean(value ? 1 : 0);
     }
 
-    case 'Int': {
+    case ValueKind.INT: {
       return Value.fromI32(value);
     }
 
-    case 'BigInt': {
+    case ValueKind.BIGINT: {
       const valueStringPtr = await __newString(value.toString());
       const bigInt = await ASBigInt.fromString(valueStringPtr);
 
       return Value.fromBigInt(bigInt);
     }
 
-    case 'BigDecimal': {
+    case ValueKind.BIGDECIMAL: {
       const valueStringPtr = await __newString(value.toString());
       const bigDecimal = await BigDecimal.fromString(valueStringPtr);
 
       return Value.fromBigDecimal(bigDecimal);
     }
 
-    case 'Bytes': {
+    case ValueKind.BYTES: {
       const entityValue = await __newString(value);
       const byteArray = await ByteArray.fromHexString(entityValue);
       const bytes = await Bytes.fromByteArray(byteArray);
@@ -687,11 +689,17 @@ const formatEntityValue = async (instanceExports: any, subgraphValue: any, type:
       return Value.fromBytes(bytes);
     }
 
-    // Return default as string for enum or custom type.
-    default: {
-      const entityValue = await __newString(value);
+    case ValueKind.ARRAY: {
+      const dataArrayPromises = value.map((el: any) => formatValue(instanceExports, el.type, el.data));
+      const dataArray = await Promise.all(dataArrayPromises);
+      const arrayStoreValueId = await getIdOfType(TypeId.ArrayStoreValue);
+      const valueArray = await __newArray(arrayStoreValueId, dataArray);
 
-      return Value.fromString(entityValue);
+      return Value.fromArray(valueArray);
+    }
+
+    case ValueKind.NULL: {
+      return Value.fromNull();
     }
   }
 };

packages/util/src/indexer.ts

@@ -758,7 +758,7 @@ export class Indexer {
     return Object.values(this._watchedContracts);
   }
 
-  async watchContract (address: string, kind: string, checkpoint: boolean, startingBlock: number): Promise<void> {
+  async watchContract (address: string, kind: string, checkpoint: boolean, startingBlock: number, context?: any): Promise<void> {
     assert(this._db.saveContract);
 
     // Use the checksum address (https://docs.ethers.io/v5/api/utils/address/#utils-getAddress) if input to address is a contract address.
@@ -770,7 +770,7 @@ export class Indexer {
     const dbTx = await this._db.createTransactionRunner();
 
     try {
-      const contract = await this._db.saveContract(dbTx, contractAddress, kind, checkpoint, startingBlock);
+      const contract = await this._db.saveContract(dbTx, contractAddress, kind, checkpoint, startingBlock, context);
       this.cacheContract(contract);
       await dbTx.commitTransaction();
 

packages/util/src/types.ts

@@ -73,6 +73,7 @@ export interface ContractInterface {
   startingBlock: number;
   kind: string;
   checkpoint: boolean;
+  context: Record<string, { type: number, data: any }>;
 }
 
 export interface StateInterface {
@@ -204,7 +205,7 @@ export interface IndexerInterface {
   getContractsByKind?: (kind: string) => ContractInterface[]
   addContracts?: () => Promise<void>
   cacheContract: (contract: ContractInterface) => void;
-  watchContract: (address: string, kind: string, checkpoint: boolean, startingBlock: number) => Promise<void>
+  watchContract: (address: string, kind: string, checkpoint: boolean, startingBlock: number, context?: any) => Promise<void>
   getEntityTypesMap?: () => Map<string, { [key: string]: string }>
   getRelationsMap?: () => Map<any, { [key: string]: any }>
   processInitialState: (contractAddress: string, blockHash: string) => Promise<any>
@@ -263,7 +264,7 @@ export interface DatabaseInterface {
   removeEntities<Entity> (queryRunner: QueryRunner, entity: new () => Entity, findConditions?: FindManyOptions<Entity> | FindConditions<Entity>): Promise<void>;
   deleteEntitiesByConditions<Entity> (queryRunner: QueryRunner, entity: EntityTarget<Entity>, findConditions: FindConditions<Entity>): Promise<void>
   getContracts: () => Promise<ContractInterface[]>
-  saveContract: (queryRunner: QueryRunner, contractAddress: string, kind: string, checkpoint: boolean, startingBlock: number) => Promise<ContractInterface>
+  saveContract: (queryRunner: QueryRunner, contractAddress: string, kind: string, checkpoint: boolean, startingBlock: number, context?: any) => Promise<ContractInterface>
   getLatestState (contractAddress: string, kind: StateKind | null, blockNumber?: number): Promise<StateInterface | undefined>
   getStates (where: FindConditions<StateInterface>): Promise<StateInterface[]>
   getDiffStatesInRange (contractAddress: string, startBlock: number, endBlock: number): Promise<StateInterface[]>