cerc-io/watcher-ts
16
1
Fork 0
mirror of https://github.com/cerc-io/watcher-ts synced 2024-11-19 20:36:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

Integrate generated watcher to invoke handlers in graph-node (#33)

Browse Source 
* Invoke handlers based on watcher-ts events

* Read subgraph yaml and listen to events from watcher-ts

* Create GraphWatcher class to use in generated example contract watcher

* Call graph-node event handler from generated watcher
This commit is contained in:
nikugogoi 2021-10-28 16:31:56 +05:30 committed by nabarun
parent 1c15c1eedb
commit 6cca55a1ab
51 changed files with 2533 additions and 384 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
packages/graph-node/package.json
View File

@ -1,10 +1,11 @@
{
"name": "@vulcanize/graph-node",
"version": "0.1.0",
"main": "index.js",
"main": "dist/index.js",
"license": "AGPL-3.0",
"devDependencies": {
"@graphprotocol/graph-ts": "^0.22.0",
"@types/js-yaml": "^4.0.4",
"@typescript-eslint/eslint-plugin": "^4.25.0",
"@typescript-eslint/parser": "^4.25.0",
"eslint": "^7.27.0",
@ -21,14 +22,16 @@
},
"scripts": {
"lint": "eslint .",
"build": "yarn asbuild && yarn build:example",
"build": "tsc",
"asbuild:debug": "asc assembly/index.ts --lib ./node_modules --exportRuntime --target debug --runPasses asyncify",
"asbuild:release": "asc assembly/index.ts --lib ./node_modules --exportRuntime --target release --runPasses asyncify",
"asbuild": "yarn asbuild:debug && yarn asbuild:release",
"test": "yarn asbuild:debug && mocha src/**/*.test.ts",
"build:example": "cd test/subgraph/example1 && yarn && yarn build"
"build:example": "cd test/subgraph/example1 && yarn && yarn build",
"watch": "DEBUG=vulcanize:* nodemon --watch src src/watcher.ts"
},
"dependencies": {
"@vulcanize/assemblyscript": "0.0.1"
"@vulcanize/assemblyscript": "0.0.1",
"js-yaml": "^4.1.0"
}
}

2
packages/graph-node/src/call-handler.test.ts
View File

@ -4,7 +4,7 @@
import path from 'path';
import { instantiate } from './index';
import { instantiate } from './loader';
import { createEvent } from './utils';
describe('call handler in mapping code', () => {

2
packages/graph-node/src/eden.test.ts
View File

@ -6,7 +6,7 @@ import assert from 'assert';
import { ethers } from 'ethers';
import path from 'path';
import { instantiate } from './index';
import { instantiate } from './loader';
import { createEvent } from './utils';
import edenNetworkAbi from '../test/subgraph/eden/EdenNetwork/abis/EdenNetwork.json';
import merkleDistributorAbi from '../test/subgraph/eden/EdenNetworkDistribution/abis/MerkleDistributor.json';

2
packages/graph-node/src/eth-call.test.ts
View File

@ -5,7 +5,7 @@
import assert from 'assert';
import path from 'path';
import { instantiate } from './index';
import { instantiate } from './loader';
import exampleAbi from '../test/subgraph/example1/build/Example1/abis/Example1.json';
describe('eth-call wasm tests', () => {

364
packages/graph-node/src/index.ts
View File

@ -1,363 +1 @@
//
// Copyright 2021 Vulcanize, Inc.
//
import assert from 'assert';
import fs from 'fs/promises';
import loader from '@vulcanize/assemblyscript/lib/loader';
import {
utils,
BigNumber,
getDefaultProvider,
Contract,
ContractInterface
} from 'ethers';
import { TypeId } from './types';
import { fromEthereumValue, toEthereumValue } from './utils';
const NETWORK_URL = 'http://127.0.0.1:8081';
type idOfType = (TypeId: number) => number
interface DataSource {
address: string
}
interface GraphData {
abis?: {[key: string]: ContractInterface};
dataSource?: DataSource;
}
export const instantiate = async (filePath: string, data: GraphData = {}): Promise<loader.ResultObject & { exports: any }> => {
const { abis = {}, dataSource } = data;
const buffer = await fs.readFile(filePath);
const provider = getDefaultProvider(NETWORK_URL);
const imports: WebAssembly.Imports = {
index: {
'store.get': async (entity: number, id: number) => {
console.log('store.get');
const entityString = __getString(entity);
console.log('entity:', entityString);
const idString = __getString(id);
console.log('id:', idString);
// TODO: Implement store get to fetch from DB using entity and id.
// TODO: Fill entity with field values.
// return Entity.__new()
return null;
},
'store.set': async (entity: number, id: number, data: number) => {
console.log('store.set');
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.
},
'typeConversion.stringToH160': () => {
console.log('index typeConversion.stringToH160');
},
'typeConversion.bytesToHex': () => {
console.log('index typeConversion.bytesToHex');
},
// 'typeConversion.bytesToString': () => {
// console.log('typeConversion.bytesToString');
// },
'typeConversion.bigIntToString': () => {
console.log('index typeConversion.bigIntToString');
},
// 'bigDecimal.fromString': () => {
// console.log('bigDecimal.fromString');
// },
// 'bigDecimal.times': () => {
// console.log('bigDecimal.times');
// },
'bigDecimal.dividedBy': () => {
console.log('bigDecimal.dividedBy');
},
// 'bigDecimal.plus': () => {
// console.log('bigDecimal.plus');
// },
// 'bigDecimal.minus': () => {
// console.log('bigDecimal.minus');
// },
'bigInt.plus': () => {
console.log('bigInt.plus');
},
'bigInt.minus': () => {
console.log('bigInt.minus');
},
'bigInt.times': () => {
console.log('bigInt.times');
},
'bigInt.dividedBy': () => {
console.log('bigInt.dividedBy');
},
// 'bigInt.mod': () => {
// console.log('bigInt.mod');
// },
'bigInt.fromString': () => {
console.log('bigInt.fromString');
},
'log.log': (_: number, msg: number) => {
console.log('log.log', __getString(msg));
},
// 'dataSource.create': () => {
// console.log('dataSource.create');
// },
'dataSource.address': () => {
console.log('dataSource.address');
},
'test.asyncMethod': async () => {
console.log('before timer start');
await new Promise(resolve => {
setTimeout(() => {
resolve(1);
}, 3000);
});
console.log('after timer complete');
return 123;
}
},
ethereum: {
'ethereum.call': async (call: number) => {
const smartContractCall = await ethereum.SmartContractCall.wrap(call);
const contractAddress = await Address.wrap(await smartContractCall.contractAddress);
const contractName = __getString(await smartContractCall.contractName);
const functionName = __getString(await smartContractCall.functionName);
const functionSignature = __getString(await smartContractCall.functionSignature);
let functionParams = __getArray(await smartContractCall.functionParams);
console.log('ethereum.call params');
console.log('functionSignature:', functionSignature);
const abi = abis[contractName];
const contract = new Contract(__getString(await contractAddress.toHexString()), abi, provider);
try {
const functionParamsPromise = functionParams.map(async param => {
const ethereumValue = await ethereum.Value.wrap(param);
return fromEthereumValue(exports, ethereumValue);
});
functionParams = await Promise.all(functionParamsPromise);
// TODO: Check for function overloading.
let result = await contract[functionName](...functionParams);
if (!Array.isArray(result)) {
result = [result];
}
// TODO: Check for function overloading.
// Using function signature does not work.
const outputs = contract.interface.getFunction(functionName).outputs;
const resultPtrArrayPromise = result.map(async (value: any, index: number) => {
assert(outputs);
return toEthereumValue(exports, value, outputs[index].type);
});
const resultPtrArray: any[] = await Promise.all(resultPtrArrayPromise);
const res = await __newArray(await getIdOfType(TypeId.ArrayEthereumValue), resultPtrArray);
return res;
} catch (err) {
console.log('eth_call error', err);
return null;
}
}
},
conversion: {
'typeConversion.stringToH160': async (s: number) => {
const string = __getString(s);
const address = utils.getAddress(string);
const byteArray = utils.arrayify(address);
const uint8ArrayId = await getIdOfType(TypeId.Uint8Array);
const ptr = __newArray(uint8ArrayId, byteArray);
return ptr;
},
'typeConversion.bigIntToString': (bigInt: number) => {
const bigIntByteArray = __getArray(bigInt);
const bigNumber = BigNumber.from(bigIntByteArray);
const ptr = __newString(bigNumber.toString());
return ptr;
},
'typeConversion.bigIntToHex': () => {
console.log('index typeConversion.bigIntToHex');
},
'typeConversion.bytesToHex': async (bytes: number) => {
const byteArray = __getArray(bytes);
const hexString = utils.hexlify(byteArray);
const ptr = await __newString(hexString);
return ptr;
},
'typeConversion.bytesToString': () => {
console.log('index typeConversion.bytesToString');
},
'typeConversion.bytesToBase58': () => {
console.log('index typeConversion.bytesToBase58');
}
},
numbers: {
'bigDecimal.dividedBy': async (x: number, y: number) => {
console.log('numbers bigDecimal.dividedBy');
const bigDecimaly = BigDecimal.wrap(y);
const yDigitsBigIntArray = __getArray(await bigDecimaly.digits);
const yDigits = BigNumber.from(yDigitsBigIntArray);
const yExpBigIntArray = __getArray(await bigDecimaly.exp);
const yExp = BigNumber.from(yExpBigIntArray);
console.log('y digits and exp', yDigits, yExp);
},
'bigDecimal.toString': () => {
console.log('numbers bigDecimal.toString');
},
'bigDecimal.fromString': () => {
console.log('numbers bigDecimal.toString');
},
'bigDecimal.plus': () => {
console.log('bigDecimal.plus');
},
'bigDecimal.minus': () => {
console.log('bigDecimal.minus');
},
'bigDecimal.times': () => {
console.log('bigDecimal.times');
},
'bigInt.fromString': async (s: number) => {
const string = __getString(s);
const bigNumber = BigNumber.from(string);
const hex = bigNumber.toHexString();
const bytes = utils.arrayify(hex);
const uint8ArrayId = await getIdOfType(TypeId.Uint8Array);
const ptr = await __newArray(uint8ArrayId, bytes);
const bigInt = await BigInt.fromSignedBytes(ptr);
return bigInt;
},
'bigInt.plus': async (x: number, y: number) => {
const xBigInt = await BigInt.wrap(x);
const xBigNumber = BigNumber.from(__getString(await xBigInt.toString()));
const yBigInt = await BigInt.wrap(y);
const yBigNumber = BigNumber.from(__getString(await yBigInt.toString()));
const sum = xBigNumber.add(yBigNumber);
const ptr = await __newString(sum.toString());
const sumBigInt = await BigInt.fromString(ptr);
return sumBigInt;
},
'bigInt.minus': async (x: number, y: number) => {
const xBigInt = await BigInt.wrap(x);
const xBigNumber = BigNumber.from(__getString(await xBigInt.toString()));
const yBigInt = await BigInt.wrap(y);
const yBigNumber = BigNumber.from(__getString(await yBigInt.toString()));
const diff = xBigNumber.sub(yBigNumber);
const ptr = await __newString(diff.toString());
const diffBigInt = BigInt.fromString(ptr);
return diffBigInt;
},
'bigInt.times': async (x: number, y: number) => {
const xBigInt = await BigInt.wrap(x);
const xBigNumber = BigNumber.from(__getString(await xBigInt.toString()));
const yBigInt = await BigInt.wrap(y);
const yBigNumber = BigNumber.from(__getString(await yBigInt.toString()));
const product = xBigNumber.mul(yBigNumber);
const ptr = await __newString(product.toString());
const productBigInt = BigInt.fromString(ptr);
return productBigInt;
},
'bigInt.dividedBy': async (x: number, y: number) => {
const xBigInt = await BigInt.wrap(x);
const xBigNumber = BigNumber.from(__getString(await xBigInt.toString()));
const yBigInt = await BigInt.wrap(y);
const yBigNumber = BigNumber.from(__getString(await yBigInt.toString()));
const quotient = xBigNumber.div(yBigNumber);
const ptr = await __newString(quotient.toString());
const quotientBigInt = BigInt.fromString(ptr);
return quotientBigInt;
},
'bigInt.dividedByDecimal': () => {
console.log('bigInt.dividedByDecimal');
},
'bigInt.mod': () => {
console.log('bigInt.mod');
},
'bigInt.bitOr': () => {
console.log('bigInt.bitOr');
},
'bigInt.bitAnd': () => {
console.log('bigInt.bitAnd');
},
'bigInt.leftShift': () => {
console.log('bigInt.leftShift');
},
'bigInt.rightShift': () => {
console.log('bigInt.rightShift');
},
'bigInt.pow': () => {
console.log('bigInt.pow');
}
},
datasource: {
'dataSource.address': async () => {
assert(dataSource);
return Address.fromString(await __newString(dataSource.address));
}
}
};
const instance = await loader.instantiate(buffer, imports);
const { exports } = instance;
const { __getString, __newString, __getArray, __newArray } = exports;
// TODO: Assign from types file generated by graph-cli
const getIdOfType: idOfType = exports.id_of_type as idOfType;
const BigDecimal: any = exports.BigDecimal as any;
const BigInt: any = exports.BigInt as any;
const Address: any = exports.Address as any;
const ethereum: any = exports.ethereum as any;
const Entity: any = exports.Entity as any;
return instance;
};
export * from './watcher';

2
packages/graph-node/src/index.test.ts → packages/graph-node/src/loader.test.ts
View File

@ -5,7 +5,7 @@
import path from 'path';
import { expect } from 'chai';
import { instantiate } from './index';
import { instantiate } from './loader';
const WASM_FILE_PATH = '../build/debug.wasm';

363
packages/graph-node/src/loader.ts Normal file
View File

@ -0,0 +1,363 @@
//
// Copyright 2021 Vulcanize, Inc.
//
import assert from 'assert';
import fs from 'fs/promises';
import loader from '@vulcanize/assemblyscript/lib/loader';
import {
utils,
BigNumber,
getDefaultProvider,
Contract,
ContractInterface
} from 'ethers';
import { TypeId } from './types';
import { fromEthereumValue, toEthereumValue } from './utils';
const NETWORK_URL = 'http://127.0.0.1:8081';
type idOfType = (TypeId: number) => number
interface DataSource {
address: string
}
interface GraphData {
abis?: {[key: string]: ContractInterface};
dataSource?: DataSource;
}
export const instantiate = async (filePath: string, data: GraphData = {}): Promise<loader.ResultObject & { exports: any }> => {
const { abis = {}, dataSource } = data;
const buffer = await fs.readFile(filePath);
const provider = getDefaultProvider(NETWORK_URL);
const imports: WebAssembly.Imports = {
index: {
'store.get': async (entity: number, id: number) => {
console.log('store.get');
const entityString = __getString(entity);
console.log('entity:', entityString);
const idString = __getString(id);
console.log('id:', idString);
// TODO: Implement store get to fetch from DB using entity and id.
// TODO: Fill entity with field values.
// return Entity.__new()
return null;
},
'store.set': async (entity: number, id: number, data: number) => {
console.log('store.set');
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.
},
'typeConversion.stringToH160': () => {
console.log('index typeConversion.stringToH160');
},
'typeConversion.bytesToHex': () => {
console.log('index typeConversion.bytesToHex');
},
// 'typeConversion.bytesToString': () => {
// console.log('typeConversion.bytesToString');
// },
'typeConversion.bigIntToString': () => {
console.log('index typeConversion.bigIntToString');
},
// 'bigDecimal.fromString': () => {
// console.log('bigDecimal.fromString');
// },
// 'bigDecimal.times': () => {
// console.log('bigDecimal.times');
// },
'bigDecimal.dividedBy': () => {
console.log('bigDecimal.dividedBy');
},
// 'bigDecimal.plus': () => {
// console.log('bigDecimal.plus');
// },
// 'bigDecimal.minus': () => {
// console.log('bigDecimal.minus');
// },
'bigInt.plus': () => {
console.log('bigInt.plus');
},
'bigInt.minus': () => {
console.log('bigInt.minus');
},
'bigInt.times': () => {
console.log('bigInt.times');
},
'bigInt.dividedBy': () => {
console.log('bigInt.dividedBy');
},
// 'bigInt.mod': () => {
// console.log('bigInt.mod');
// },
'bigInt.fromString': () => {
console.log('bigInt.fromString');
},
'log.log': (_: number, msg: number) => {
console.log('log.log', __getString(msg));
},
// 'dataSource.create': () => {
// console.log('dataSource.create');
// },
'dataSource.address': () => {
console.log('dataSource.address');
},
'test.asyncMethod': async () => {
console.log('before timer start');
await new Promise(resolve => {
setTimeout(() => {
resolve(1);
}, 3000);
});
console.log('after timer complete');
return 123;
}
},
ethereum: {
'ethereum.call': async (call: number) => {
const smartContractCall = await ethereum.SmartContractCall.wrap(call);
const contractAddress = await Address.wrap(await smartContractCall.contractAddress);
const contractName = __getString(await smartContractCall.contractName);
const functionName = __getString(await smartContractCall.functionName);
const functionSignature = __getString(await smartContractCall.functionSignature);
let functionParams = __getArray(await smartContractCall.functionParams);
console.log('ethereum.call params');
console.log('functionSignature:', functionSignature);
const abi = abis[contractName];
const contract = new Contract(__getString(await contractAddress.toHexString()), abi, provider);
try {
const functionParamsPromise = functionParams.map(async param => {
const ethereumValue = await ethereum.Value.wrap(param);
return fromEthereumValue(exports, ethereumValue);
});
functionParams = await Promise.all(functionParamsPromise);
// TODO: Check for function overloading.
let result = await contract[functionName](...functionParams);
if (!Array.isArray(result)) {
result = [result];
}
// TODO: Check for function overloading.
// Using function signature does not work.
const outputs = contract.interface.getFunction(functionName).outputs;
const resultPtrArrayPromise = result.map(async (value: any, index: number) => {
assert(outputs);
return toEthereumValue(exports, value, outputs[index].type);
});
const resultPtrArray: any[] = await Promise.all(resultPtrArrayPromise);
const res = await __newArray(await getIdOfType(TypeId.ArrayEthereumValue), resultPtrArray);
return res;
} catch (err) {
console.log('eth_call error', err);
return null;
}
}
},
conversion: {
'typeConversion.stringToH160': async (s: number) => {
const string = __getString(s);
const address = utils.getAddress(string);
const byteArray = utils.arrayify(address);
const uint8ArrayId = await getIdOfType(TypeId.Uint8Array);
const ptr = __newArray(uint8ArrayId, byteArray);
return ptr;
},
'typeConversion.bigIntToString': (bigInt: number) => {
const bigIntByteArray = __getArray(bigInt);
const bigNumber = BigNumber.from(bigIntByteArray);
const ptr = __newString(bigNumber.toString());
return ptr;
},
'typeConversion.bigIntToHex': () => {
console.log('index typeConversion.bigIntToHex');
},
'typeConversion.bytesToHex': async (bytes: number) => {
const byteArray = __getArray(bytes);
const hexString = utils.hexlify(byteArray);
const ptr = await __newString(hexString);
return ptr;
},
'typeConversion.bytesToString': () => {
console.log('index typeConversion.bytesToString');
},
'typeConversion.bytesToBase58': () => {
console.log('index typeConversion.bytesToBase58');
}
},
numbers: {
'bigDecimal.dividedBy': async (x: number, y: number) => {
console.log('numbers bigDecimal.dividedBy');
const bigDecimaly = BigDecimal.wrap(y);
const yDigitsBigIntArray = __getArray(await bigDecimaly.digits);
const yDigits = BigNumber.from(yDigitsBigIntArray);
const yExpBigIntArray = __getArray(await bigDecimaly.exp);
const yExp = BigNumber.from(yExpBigIntArray);
console.log('y digits and exp', yDigits, yExp);
},
'bigDecimal.toString': () => {
console.log('numbers bigDecimal.toString');
},
'bigDecimal.fromString': () => {
console.log('numbers bigDecimal.toString');
},
'bigDecimal.plus': () => {
console.log('bigDecimal.plus');
},
'bigDecimal.minus': () => {
console.log('bigDecimal.minus');
},
'bigDecimal.times': () => {
console.log('bigDecimal.times');
},
'bigInt.fromString': async (s: number) => {
const string = __getString(s);
const bigNumber = BigNumber.from(string);
const hex = bigNumber.toHexString();
const bytes = utils.arrayify(hex);
const uint8ArrayId = await getIdOfType(TypeId.Uint8Array);
const ptr = await __newArray(uint8ArrayId, bytes);
const bigInt = await BigInt.fromSignedBytes(ptr);
return bigInt;
},
'bigInt.plus': async (x: number, y: number) => {
const xBigInt = await BigInt.wrap(x);
const xBigNumber = BigNumber.from(__getString(await xBigInt.toString()));
const yBigInt = await BigInt.wrap(y);
const yBigNumber = BigNumber.from(__getString(await yBigInt.toString()));
const sum = xBigNumber.add(yBigNumber);
const ptr = await __newString(sum.toString());
const sumBigInt = await BigInt.fromString(ptr);
return sumBigInt;
},
'bigInt.minus': async (x: number, y: number) => {
const xBigInt = await BigInt.wrap(x);
const xBigNumber = BigNumber.from(__getString(await xBigInt.toString()));
const yBigInt = await BigInt.wrap(y);
const yBigNumber = BigNumber.from(__getString(await yBigInt.toString()));
const diff = xBigNumber.sub(yBigNumber);
const ptr = await __newString(diff.toString());
const diffBigInt = BigInt.fromString(ptr);
return diffBigInt;
},
'bigInt.times': async (x: number, y: number) => {
const xBigInt = await BigInt.wrap(x);
const xBigNumber = BigNumber.from(__getString(await xBigInt.toString()));
const yBigInt = await BigInt.wrap(y);
const yBigNumber = BigNumber.from(__getString(await yBigInt.toString()));
const product = xBigNumber.mul(yBigNumber);
const ptr = await __newString(product.toString());
const productBigInt = BigInt.fromString(ptr);
return productBigInt;
},
'bigInt.dividedBy': async (x: number, y: number) => {
const xBigInt = await BigInt.wrap(x);
const xBigNumber = BigNumber.from(__getString(await xBigInt.toString()));
const yBigInt = await BigInt.wrap(y);
const yBigNumber = BigNumber.from(__getString(await yBigInt.toString()));
const quotient = xBigNumber.div(yBigNumber);
const ptr = await __newString(quotient.toString());
const quotientBigInt = BigInt.fromString(ptr);
return quotientBigInt;
},
'bigInt.dividedByDecimal': () => {
console.log('bigInt.dividedByDecimal');
},
'bigInt.mod': () => {
console.log('bigInt.mod');
},
'bigInt.bitOr': () => {
console.log('bigInt.bitOr');
},
'bigInt.bitAnd': () => {
console.log('bigInt.bitAnd');
},
'bigInt.leftShift': () => {
console.log('bigInt.leftShift');
},
'bigInt.rightShift': () => {
console.log('bigInt.rightShift');
},
'bigInt.pow': () => {
console.log('bigInt.pow');
}
},
datasource: {
'dataSource.address': async () => {
assert(dataSource);
return Address.fromString(await __newString(dataSource.address));
}
}
};
const instance = await loader.instantiate(buffer, imports);
const { exports } = instance;
const { __getString, __newString, __getArray, __newArray } = exports;
// TODO: Assign from types file generated by graph-cli
const getIdOfType: idOfType = exports.id_of_type as idOfType;
const BigDecimal: any = exports.BigDecimal as any;
const BigInt: any = exports.BigInt as any;
const Address: any = exports.Address as any;
const ethereum: any = exports.ethereum as any;
const Entity: any = exports.Entity as any;
return instance;
};

2
packages/graph-node/src/numbers.test.ts
View File

@ -5,7 +5,7 @@
import path from 'path';
import { expect } from 'chai';
import { instantiate } from './index';
import { instantiate } from './loader';
const EXAMPLE_WASM_FILE_PATH = '../test/subgraph/example1/build/Example1/Example1.wasm';

2
packages/graph-node/src/type-conversion.test.ts
View File

@ -5,7 +5,7 @@
import path from 'path';
import { expect } from 'chai';
import { instantiate } from './index';
import { instantiate } from './loader';
const EXAMPLE_WASM_FILE_PATH = '../test/subgraph/example1/build/Example1/Example1.wasm';

22
packages/graph-node/src/utils.ts
View File

@ -1,6 +1,13 @@
import { BigNumber } from 'ethers';
import path from 'path';
import fs from 'fs-extra';
import debug from 'debug';
import yaml from 'js-yaml';
import { TypeId, ValueKind } from './types';
const log = debug('vulcanize:utils');
interface EventParam {
name: string;
value: any;
@ -215,3 +222,18 @@ export const toEthereumValue = async (exports: any, value: any, type: string): P
// For string type.
return ethereum.Value.fromString(await __newString(value));
};
export const getSubgraphConfig = async (subgraphPath: string): Promise<any> => {
const configFilePath = path.resolve(path.join(subgraphPath, 'subgraph.yaml'));
const fileExists = await fs.pathExists(configFilePath);
if (!fileExists) {
throw new Error(`Config file not found: ${configFilePath}`);
}
console.log(configFilePath);
const config = yaml.load(await fs.readFile(configFilePath, 'utf8'));
log('config', JSON.stringify(config, null, 2));
return config;
};

76
packages/graph-node/src/watcher.ts Normal file
View File

@ -0,0 +1,76 @@
//
// Copyright 2021 Vulcanize, Inc.
//
import 'reflect-metadata';
import debug from 'debug';
import path from 'path';
import fs from 'fs';
import { ContractInterface } from 'ethers';
import { getSubgraphConfig } from './utils';
import { instantiate } from './loader';
import { ResultObject } from '@vulcanize/assemblyscript/lib/loader';
const log = debug('vulcanize:graph-watcher');
export class GraphWatcher {
_subgraphPath: string;
_dataSources: any[] = []
_instanceMap: { [key: string]: ResultObject & { exports: any } } = {};
constructor (subgraphPath: string) {
this._subgraphPath = subgraphPath;
}
async init () {
const { dataSources } = await getSubgraphConfig(this._subgraphPath);
this._dataSources = dataSources;
this._instanceMap = this._dataSources.reduce(async (acc: { [key: string]: ResultObject & { exports: any } }, dataSource: any) => {
const { source: { address }, mapping } = dataSource;
const { abis, file } = mapping;
const data = {
abis: abis.reduce((acc: {[key: string]: ContractInterface}, abi: any) => {
const { name, file } = abi;
const abiFilePath = path.join(this._subgraphPath, file);
acc[name] = JSON.parse(fs.readFileSync(abiFilePath).toString());
return acc;
}, {}),
dataSource: {
address
}
};
const filePath = path.join(this._subgraphPath, file);
const instance = await instantiate(filePath, data);
acc[address] = instance;
return acc;
}, {});
}
async handleEvent (eventData: any) {
const { contract } = eventData;
const dataSource = this._dataSources.find(dataSource => dataSource.source.address === contract);
if (!dataSource) {
log(`Subgraph doesnt have configuration for contract ${contract}`);
return;
}
// TODO: Call instance methods based on event signature.
// value should contain event signature.
const [{ handler }] = dataSource.mapping.eventHandlers;
const { exports } = this._instanceMap[contract];
// Create ethereum event to be passed to handler.
// TODO: Create ethereum event to be passed to handler.
// const ethereumEvent = await createEvent(exports, address, event);
await exports[handler]();
}
}

15
packages/graph-node/test/contracts/Example.sol
View File

@ -1,10 +1,19 @@
pragma solidity >=0.4.22 <0.8.0;
// SPDX-License-Identifier: UNLICENSED
contract Example {
event Test(string param1, uint param2);
pragma solidity ^0.8.0;
contract Example {
uint256 private _test;
event Test(string param1, uint8 param2);
function getMethod() public view virtual returns (string memory)
{
return 'test';
}
function emitEvent() public virtual returns (bool) {
emit Test('abc', 123);
return true;
}
}

6
packages/graph-node/tsconfig.json
View File

@ -42,10 +42,10 @@
// "maxNodeModuleJsDepth": 1, /* Specify the maximum folder depth used for checking JavaScript files from `node_modules`. Only applicable with `allowJs`. */
/* Emit */
// "declaration": true, /* Generate .d.ts files from TypeScript and JavaScript files in your project. */
"declaration": true, /* Generate .d.ts files from TypeScript and JavaScript files in your project. */
// "declarationMap": true, /* Create sourcemaps for d.ts files. */
// "emitDeclarationOnly": true, /* Only output d.ts files and not JavaScript files. */
// "sourceMap": true, /* Create source map files for emitted JavaScript files. */
"sourceMap": true, /* Create source map files for emitted JavaScript files. */
// "outFile": "./", /* Specify a file that bundles all outputs into one JavaScript file. If `declaration` is true, also designates a file that bundles all .d.ts output. */
"outDir": "dist", /* Specify an output folder for all emitted files. */
// "removeComments": true, /* Disable emitting comments. */
@ -98,5 +98,5 @@
"skipLibCheck": true /* Skip type checking all .d.ts files. */
},
"include": ["src"],
"exclude": ["assembly", "dist"]
"exclude": ["assembly", "dist", "src/**/*.test.ts"]
}

2
packages/graph-test-watcher/.eslintignore Normal file
View File

@ -0,0 +1,2 @@
# Don't lint build output.
dist

27
packages/graph-test-watcher/.eslintrc.json Normal file
View File

@ -0,0 +1,27 @@
{
"env": {
"browser": true,
"es2021": true
},
"extends": [
"semistandard",
"plugin:@typescript-eslint/recommended"
],
"parser": "@typescript-eslint/parser",
"parserOptions": {
"ecmaVersion": 12,
"sourceType": "module"
},
"plugins": [
"@typescript-eslint"
],
"rules": {
"@typescript-eslint/no-explicit-any": "off",
"@typescript-eslint/explicit-module-boundary-types": [
"warn",
{
"allowArgumentsExplicitlyTypedAsAny": true
}
]
}
}

6
packages/graph-test-watcher/.gitignore vendored Normal file
View File

@ -0,0 +1,6 @@
.idea/
.vscode/
node_modules/
build/
tmp/
temp/

78
packages/graph-test-watcher/README.md Normal file
View File

@ -0,0 +1,78 @@
# Example Watcher
## Setup
* Run the following command to install required packages:
```bash
yarn
```
* Create a postgres12 database for the watcher:
```bash
sudo su - postgres
createdb graph-test-watcher
```
* If the watcher is an `active` watcher:
Create database for the job queue and enable the `pgcrypto` extension on them (https://github.com/timgit/pg-boss/blob/master/docs/usage.md#intro):
```
createdb graph-test-watcher-job-queue
```
```
postgres@tesla:~$ psql -U postgres -h localhost graph-test-watcher-job-queue
Password for user postgres:
psql (12.7 (Ubuntu 12.7-1.pgdg18.04+1))
SSL connection (protocol: TLSv1.3, cipher: TLS_AES_256_GCM_SHA384, bits: 256, compression: off)
Type "help" for help.
graph-test-watcher-job-queue=# CREATE EXTENSION pgcrypto;
CREATE EXTENSION
graph-test-watcher-job-queue=# exit
```
* Update the [config](./environments/local.toml) with database connection settings.
* Update the `upstream` config in the [config file](./environments/local.toml) and provide the `ipld-eth-server` GQL API and the `indexer-db` postgraphile endpoints.
## Customize
* Indexing on an event:
* Edit the custom hook function `handleEvent` (triggered on an event) in [hooks.ts](./src/hooks.ts) to perform corresponding indexing using the `Indexer` object.
* Refer to [hooks.example.ts](./src/hooks.example.ts) for an example hook function for events in an ERC20 contract.
## Run
* Run the watcher:
```bash
yarn server
```
GQL console: http://localhost:3008/graphql
* If the watcher is an `active` watcher:
* Run the job-runner:
```bash
yarn job-runner
```
* To watch a contract:
```bash
yarn watch:contract --address <contract-address> --kind Example --starting-block [block-number]
```
* To fill a block range:
```bash
yarn fill --startBlock <from-block> --endBlock <to-block>
```

31
packages/graph-test-watcher/environments/local.toml Normal file
View File

@ -0,0 +1,31 @@
[server]
host = "127.0.0.1"
port = 3008
kind = "active"
subgraphPath = "../graph-node/test/subgraph/example1/build"
[database]
type = "postgres"
host = "localhost"
port = 5432
database = "graph-test-watcher"
username = "postgres"
password = "postgres"
synchronize = true
logging = false
[upstream]
[upstream.ethServer]
gqlApiEndpoint = "http://127.0.0.1:8082/graphql"
gqlPostgraphileEndpoint = "http://127.0.0.1:5000/graphql"
rpcProviderEndpoint = "http://127.0.0.1:8081"
[upstream.cache]
name = "requests"
enabled = false
deleteOnStart = false
[jobQueue]
dbConnectionString = "postgres://postgres:postgres@localhost/graph-test-watcher-job-queue"
maxCompletionLagInSecs = 300
jobDelayInMilliSecs = 100

60
packages/graph-test-watcher/package.json Normal file
View File

@ -0,0 +1,60 @@
{
"name": "@vulcanize/graph-test-watcher",
"version": "0.1.0",
"description": "graph-test-watcher",
"private": true,
"main": "dist/index.js",
"scripts": {
"lint": "eslint .",
"build": "tsc",
"server": "DEBUG=vulcanize:* ts-node src/server.ts",
"job-runner": "DEBUG=vulcanize:* ts-node src/job-runner.ts",
"watch:contract": "DEBUG=vulcanize:* ts-node src/cli/watch-contract.ts",
"fill": "DEBUG=vulcanize:* ts-node src/fill.ts"
},
"repository": {
"type": "git",
"url": "git+https://github.com/vulcanize/watcher-ts.git"
},
"author": "",
"license": "AGPL-3.0",
"bugs": {
"url": "https://github.com/vulcanize/watcher-ts/issues"
},
"homepage": "https://github.com/vulcanize/watcher-ts#readme",
"dependencies": {
"@ethersproject/providers": "5.3.0",
"@vulcanize/cache": "^0.1.0",
"@vulcanize/graph-node": "^0.1.0",
"@vulcanize/ipld-eth-client": "^0.1.0",
"@vulcanize/solidity-mapper": "^0.1.0",
"@vulcanize/util": "^0.1.0",
"apollo-server-express": "^2.25.0",
"apollo-type-bigint": "^0.1.3",
"debug": "^4.3.1",
"ethers": "^5.2.0",
"express": "^4.17.1",
"graphql": "^15.5.0",
"graphql-import-node": "^0.0.4",
"json-bigint": "^1.0.0",
"reflect-metadata": "^0.1.13",
"typeorm": "^0.2.32",
"yargs": "^17.0.1"
},
"devDependencies": {
"@ethersproject/abi": "^5.3.0",
"@types/express": "^4.17.11",
"@types/yargs": "^17.0.0",
"@typescript-eslint/eslint-plugin": "^4.25.0",
"@typescript-eslint/parser": "^4.25.0",
"eslint": "^7.27.0",
"eslint-config-semistandard": "^15.0.1",
"eslint-config-standard": "^16.0.3",
"eslint-plugin-import": "^2.23.3",
"eslint-plugin-node": "^11.1.0",
"eslint-plugin-promise": "^5.1.0",
"eslint-plugin-standard": "^5.0.0",
"ts-node": "^10.0.0",
"typescript": "^4.3.2"
}
}

68
packages/graph-test-watcher/src/artifacts/Example.json Normal file
View File

@ -0,0 +1,68 @@
{
"abi": [
{
"anonymous": false,
"inputs": [
{
"indexed": false,
"internalType": "string",
"name": "param1",
"type": "string"
},
{
"indexed": false,
"internalType": "uint8",
"name": "param2",
"type": "uint8"
}
],
"name": "Test",
"type": "event"
},
{
"inputs": [],
"name": "emitEvent",
"outputs": [
{
"internalType": "bool",
"name": "",
"type": "bool"
}
],
"stateMutability": "nonpayable",
"type": "function"
},
{
"inputs": [],
"name": "getMethod",
"outputs": [
{
"internalType": "string",
"name": "",
"type": "string"
}
],
"stateMutability": "view",
"type": "function"
}
],
"storageLayout": {
"storage": [
{
"astId": 3,
"contract": "Example.sol:Example",
"label": "_test",
"offset": 0,
"slot": "0",
"type": "t_uint256"
}
],
"types": {
"t_uint256": {
"encoding": "inplace",
"label": "uint256",
"numberOfBytes": "32"
}
}
}
}

55
packages/graph-test-watcher/src/cli/watch-contract.ts Normal file
View File

@ -0,0 +1,55 @@
//
// Copyright 2021 Vulcanize, Inc.
//
import assert from 'assert';
import yargs from 'yargs';
import 'reflect-metadata';
import { Config, DEFAULT_CONFIG_PATH, getConfig } from '@vulcanize/util';
import { Database } from '../database';
(async () => {
const argv = await yargs.parserConfiguration({
'parse-numbers': false
}).options({
configFile: {
alias: 'f',
type: 'string',
require: true,
demandOption: true,
describe: 'configuration file path (toml)',
default: DEFAULT_CONFIG_PATH
},
address: {
type: 'string',
require: true,
demandOption: true,
describe: 'Address of the deployed contract'
},
kind: {
type: 'string',
require: true,
demandOption: true,
describe: 'Kind of contract'
},
startingBlock: {
type: 'number',
default: 1,
describe: 'Starting block'
}
}).argv;
const config: Config = await getConfig(argv.configFile);
const { database: dbConfig } = config;
assert(dbConfig);
const db = new Database(dbConfig);
await db.init();
await db.saveContract(argv.address, argv.kind, argv.startingBlock);
await db.close();
})();

76
packages/graph-test-watcher/src/client.ts Normal file
View File

@ -0,0 +1,76 @@
//
// Copyright 2021 Vulcanize, Inc.
//
import { gql } from '@apollo/client/core';
import { GraphQLClient, GraphQLConfig } from '@vulcanize/ipld-eth-client';
import { queries, mutations, subscriptions } from './gql';
export class Client {
_config: GraphQLConfig;
_client: GraphQLClient;
constructor (config: GraphQLConfig) {
this._config = config;
this._client = new GraphQLClient(config);
}
// eslint-disable-next-line camelcase
async getGetMethod (blockHash: string, contractAddress: string): Promise<any> {
const { getMethod } = await this._client.query(
gql(queries.getMethod),
{ blockHash, contractAddress }
);
return getMethod;
}
// eslint-disable-next-line camelcase
async get_test (blockHash: string, contractAddress: string): Promise<any> {
const { _test } = await this._client.query(
gql(queries._test),
{ blockHash, contractAddress }
);
return _test;
}
async getEvents (blockHash: string, contractAddress: string, name: string): Promise<any> {
const { events } = await this._client.query(
gql(queries.events),
{ blockHash, contractAddress, name }
);
return events;
}
async getEventsInRange (fromBlockNumber: number, toBlockNumber: number): Promise<any> {
const { eventsInRange } = await this._client.query(
gql(queries.eventsInRange),
{ fromBlockNumber, toBlockNumber }
);
return eventsInRange;
}
async watchContract (contractAddress: string, startingBlock?: number): Promise<any> {
const { watchContract } = await this._client.mutate(
gql(mutations.watchContract),
{ contractAddress, startingBlock }
);
return watchContract;
}
async watchEvents (onNext: (value: any) => void): Promise<ZenObservable.Subscription> {
return this._client.subscribe(
gql(subscriptions.onEvent),
({ data }) => {
onNext(data.onEvent);
}
);
}
}

197
packages/graph-test-watcher/src/database.ts Normal file
View File

@ -0,0 +1,197 @@
//
// Copyright 2021 Vulcanize, Inc.
//
import assert from 'assert';
import { Connection, ConnectionOptions, DeepPartial, FindConditions, QueryRunner, FindManyOptions } from 'typeorm';
import path from 'path';
import { Database as BaseDatabase } from '@vulcanize/util';
import { Contract } from './entity/Contract';
import { Event } from './entity/Event';
import { SyncStatus } from './entity/SyncStatus';
import { BlockProgress } from './entity/BlockProgress';
import { GetMethod } from './entity/GetMethod';
import { _Test } from './entity/_Test';
export class Database {
_config: ConnectionOptions;
_conn!: Connection;
_baseDatabase: BaseDatabase;
_propColMaps: { [key: string]: Map<string, string>; }
constructor (config: ConnectionOptions) {
assert(config);
this._config = {
...config,
entities: [path.join(__dirname, 'entity/*')]
};
this._baseDatabase = new BaseDatabase(this._config);
this._propColMaps = {};
}
async init (): Promise<void> {
this._conn = await this._baseDatabase.init();
this._setPropColMaps();
}
async close (): Promise<void> {
return this._baseDatabase.close();
}
// eslint-disable-next-line camelcase
async getGetMethod ({ blockHash, contractAddress }: { blockHash: string, contractAddress: string }): Promise<GetMethod | undefined> {
return this._conn.getRepository(GetMethod)
.findOne({
blockHash,
contractAddress
});
}
// eslint-disable-next-line camelcase
async get_test ({ blockHash, contractAddress }: { blockHash: string, contractAddress: string }): Promise<_Test | undefined> {
return this._conn.getRepository(_Test)
.findOne({
blockHash,
contractAddress
});
}
// eslint-disable-next-line camelcase
async saveGetMethod ({ blockHash, contractAddress, value, proof }: DeepPartial<GetMethod>): Promise<GetMethod> {
const repo = this._conn.getRepository(GetMethod);
const entity = repo.create({ blockHash, contractAddress, value, proof });
return repo.save(entity);
}
// eslint-disable-next-line camelcase
async save_test ({ blockHash, contractAddress, value, proof }: DeepPartial<_Test>): Promise<_Test> {
const repo = this._conn.getRepository(_Test);
const entity = repo.create({ blockHash, contractAddress, value, proof });
return repo.save(entity);
}
async getContract (address: string): Promise<Contract | undefined> {
const repo = this._conn.getRepository(Contract);
return this._baseDatabase.getContract(repo, address);
}
async createTransactionRunner (): Promise<QueryRunner> {
return this._baseDatabase.createTransactionRunner();
}
async getProcessedBlockCountForRange (fromBlockNumber: number, toBlockNumber: number): Promise<{ expected: number, actual: number }> {
const repo = this._conn.getRepository(BlockProgress);
return this._baseDatabase.getProcessedBlockCountForRange(repo, fromBlockNumber, toBlockNumber);
}
async getEventsInRange (fromBlockNumber: number, toBlockNumber: number): Promise<Array<Event>> {
const repo = this._conn.getRepository(Event);
return this._baseDatabase.getEventsInRange(repo, fromBlockNumber, toBlockNumber);
}
async saveEventEntity (queryRunner: QueryRunner, entity: Event): Promise<Event> {
const repo = queryRunner.manager.getRepository(Event);
return this._baseDatabase.saveEventEntity(repo, entity);
}
async getBlockEvents (blockHash: string, where: FindConditions<Event>): Promise<Event[]> {
const repo = this._conn.getRepository(Event);
return this._baseDatabase.getBlockEvents(repo, blockHash, where);
}
async saveEvents (queryRunner: QueryRunner, block: DeepPartial<BlockProgress>, events: DeepPartial<Event>[]): Promise<void> {
const blockRepo = queryRunner.manager.getRepository(BlockProgress);
const eventRepo = queryRunner.manager.getRepository(Event);
return this._baseDatabase.saveEvents(blockRepo, eventRepo, block, events);
}
async saveContract (address: string, kind: string, startingBlock: number): Promise<void> {
await this._conn.transaction(async (tx) => {
const repo = tx.getRepository(Contract);
return this._baseDatabase.saveContract(repo, address, startingBlock, kind);
});
}
async updateSyncStatusIndexedBlock (queryRunner: QueryRunner, blockHash: string, blockNumber: number): Promise<SyncStatus> {
const repo = queryRunner.manager.getRepository(SyncStatus);
return this._baseDatabase.updateSyncStatusIndexedBlock(repo, blockHash, blockNumber);
}
async updateSyncStatusCanonicalBlock (queryRunner: QueryRunner, blockHash: string, blockNumber: number): Promise<SyncStatus> {
const repo = queryRunner.manager.getRepository(SyncStatus);
return this._baseDatabase.updateSyncStatusCanonicalBlock(repo, blockHash, blockNumber);
}
async updateSyncStatusChainHead (queryRunner: QueryRunner, blockHash: string, blockNumber: number): Promise<SyncStatus> {
const repo = queryRunner.manager.getRepository(SyncStatus);
return this._baseDatabase.updateSyncStatusChainHead(repo, blockHash, blockNumber);
}
async getSyncStatus (queryRunner: QueryRunner): Promise<SyncStatus | undefined> {
const repo = queryRunner.manager.getRepository(SyncStatus);
return this._baseDatabase.getSyncStatus(repo);
}
async getEvent (id: string): Promise<Event | undefined> {
const repo = this._conn.getRepository(Event);
return this._baseDatabase.getEvent(repo, id);
}
async getBlocksAtHeight (height: number, isPruned: boolean): Promise<BlockProgress[]> {
const repo = this._conn.getRepository(BlockProgress);
return this._baseDatabase.getBlocksAtHeight(repo, height, isPruned);
}
async markBlocksAsPruned (queryRunner: QueryRunner, blocks: BlockProgress[]): Promise<void> {
const repo = queryRunner.manager.getRepository(BlockProgress);
return this._baseDatabase.markBlocksAsPruned(repo, blocks);
}
async getBlockProgress (blockHash: string): Promise<BlockProgress | undefined> {
const repo = this._conn.getRepository(BlockProgress);
return this._baseDatabase.getBlockProgress(repo, blockHash);
}
async updateBlockProgress (queryRunner: QueryRunner, blockHash: string, lastProcessedEventIndex: number): Promise<void> {
const repo = queryRunner.manager.getRepository(BlockProgress);
return this._baseDatabase.updateBlockProgress(repo, blockHash, lastProcessedEventIndex);
}
async removeEntities<Entity> (queryRunner: QueryRunner, entity: new () => Entity, findConditions?: FindManyOptions<Entity> | FindConditions<Entity>): Promise<void> {
return this._baseDatabase.removeEntities(queryRunner, entity, findConditions);
}
async getAncestorAtDepth (blockHash: string, depth: number): Promise<string> {
return this._baseDatabase.getAncestorAtDepth(blockHash, depth);
}
_getPropertyColumnMapForEntity (entityName: string): Map<string, string> {
return this._conn.getMetadata(entityName).ownColumns.reduce((acc, curr) => {
return acc.set(curr.propertyName, curr.databaseName);
}, new Map<string, string>());
}
_setPropColMaps (): void {
this._propColMaps.GetMethod = this._getPropertyColumnMapForEntity('GetMethod');
this._propColMaps._Test = this._getPropertyColumnMapForEntity('_Test');
}
}

42
packages/graph-test-watcher/src/entity/BlockProgress.ts Normal file
View File

@ -0,0 +1,42 @@
//
// Copyright 2021 Vulcanize, Inc.
//
import { Entity, PrimaryGeneratedColumn, Column, Index } from 'typeorm';
import { BlockProgressInterface } from '@vulcanize/util';
@Entity()
@Index(['blockHash'], { unique: true })
@Index(['blockNumber'])
@Index(['parentHash'])
export class BlockProgress implements BlockProgressInterface {
@PrimaryGeneratedColumn()
id!: number;
@Column('varchar', { length: 66 })
blockHash!: string;
@Column('varchar', { length: 66 })
parentHash!: string;
@Column('integer')
blockNumber!: number;
@Column('integer')
blockTimestamp!: number;
@Column('integer')
numEvents!: number;
@Column('integer')
numProcessedEvents!: number;
@Column('integer')
lastProcessedEventIndex!: number;
@Column('boolean')
isComplete!: boolean;
@Column('boolean', { default: false })
isPruned!: boolean;
}

21
packages/graph-test-watcher/src/entity/Contract.ts Normal file
View File

@ -0,0 +1,21 @@
//
// Copyright 2021 Vulcanize, Inc.
//
import { Entity, PrimaryGeneratedColumn, Column, Index } from 'typeorm';
@Entity()
@Index(['address'], { unique: true })
export class Contract {
@PrimaryGeneratedColumn()
id!: number;
@Column('varchar', { length: 42 })
address!: string;
@Column('varchar', { length: 8 })
kind!: string;
@Column('integer')
startingBlock!: number;
}

38
packages/graph-test-watcher/src/entity/Event.ts Normal file
View File

@ -0,0 +1,38 @@
//
// Copyright 2021 Vulcanize, Inc.
//
import { Entity, PrimaryGeneratedColumn, Column, Index, ManyToOne } from 'typeorm';
import { BlockProgress } from './BlockProgress';
@Entity()
@Index(['block', 'contract'])
@Index(['block', 'contract', 'eventName'])
export class Event {
@PrimaryGeneratedColumn()
id!: number;
@ManyToOne(() => BlockProgress)
block!: BlockProgress;
@Column('varchar', { length: 66 })
txHash!: string;
@Column('integer')
index!: number;
@Column('varchar', { length: 42 })
contract!: string;
@Column('varchar', { length: 256 })
eventName!: string;
@Column('text')
eventInfo!: string;
@Column('text')
extraInfo!: string;
@Column('text')
proof!: string;
}

24
packages/graph-test-watcher/src/entity/GetMethod.ts Normal file
View File

@ -0,0 +1,24 @@
//
// Copyright 2021 Vulcanize, Inc.
//
import { Entity, PrimaryGeneratedColumn, Column, Index } from 'typeorm';
@Entity()
@Index(['blockHash', 'contractAddress'], { unique: true })
export class GetMethod {
@PrimaryGeneratedColumn()
id!: number;
@Column('varchar', { length: 66 })
blockHash!: string;
@Column('varchar', { length: 42 })
contractAddress!: string;
@Column('varchar')
value!: string;
@Column('text', { nullable: true })
proof!: string;
}

30
packages/graph-test-watcher/src/entity/SyncStatus.ts Normal file
View File

@ -0,0 +1,30 @@
//
// Copyright 2021 Vulcanize, Inc.
//
import { Entity, PrimaryGeneratedColumn, Column } from 'typeorm';
import { SyncStatusInterface } from '@vulcanize/util';
@Entity()
export class SyncStatus implements SyncStatusInterface {
@PrimaryGeneratedColumn()
id!: number;
@Column('varchar', { length: 66 })
chainHeadBlockHash!: string;
@Column('integer')
chainHeadBlockNumber!: number;
@Column('varchar', { length: 66 })
latestIndexedBlockHash!: string;
@Column('integer')
latestIndexedBlockNumber!: number;
@Column('varchar', { length: 66 })
latestCanonicalBlockHash!: string;
@Column('integer')
latestCanonicalBlockNumber!: number;
}

25
packages/graph-test-watcher/src/entity/_Test.ts Normal file
View File

@ -0,0 +1,25 @@
//
// Copyright 2021 Vulcanize, Inc.
//
import { Entity, PrimaryGeneratedColumn, Column, Index } from 'typeorm';
import { bigintTransformer } from '@vulcanize/util';
@Entity()
@Index(['blockHash', 'contractAddress'], { unique: true })
export class _Test {
@PrimaryGeneratedColumn()
id!: number;
@Column('varchar', { length: 66 })
blockHash!: string;
@Column('varchar', { length: 42 })
contractAddress!: string;
@Column('numeric', { transformer: bigintTransformer })
value!: bigint;
@Column('text', { nullable: true })
proof!: string;
}

120
packages/graph-test-watcher/src/events.ts Normal file
View File

@ -0,0 +1,120 @@
//
// Copyright 2021 Vulcanize, Inc.
//
import assert from 'assert';
import debug from 'debug';
import { PubSub } from 'apollo-server-express';
import { EthClient } from '@vulcanize/ipld-eth-client';
import {
JobQueue,
EventWatcher as BaseEventWatcher,
QUEUE_BLOCK_PROCESSING,
QUEUE_EVENT_PROCESSING,
UNKNOWN_EVENT_NAME
} from '@vulcanize/util';
import { Indexer } from './indexer';
import { Event } from './entity/Event';
const EVENT = 'event';
const log = debug('vulcanize:events');
export class EventWatcher {
_ethClient: EthClient
_indexer: Indexer
_subscription: ZenObservable.Subscription | undefined
_baseEventWatcher: BaseEventWatcher
_pubsub: PubSub
_jobQueue: JobQueue
constructor (ethClient: EthClient, indexer: Indexer, pubsub: PubSub, jobQueue: JobQueue) {
assert(ethClient);
assert(indexer);
this._ethClient = ethClient;
this._indexer = indexer;
this._pubsub = pubsub;
this._jobQueue = jobQueue;
this._baseEventWatcher = new BaseEventWatcher(this._ethClient, this._indexer, this._pubsub, this._jobQueue);
}
getEventIterator (): AsyncIterator<any> {
return this._pubsub.asyncIterator([EVENT]);
}
getBlockProgressEventIterator (): AsyncIterator<any> {
return this._baseEventWatcher.getBlockProgressEventIterator();
}
async start (): Promise<void> {
assert(!this._subscription, 'subscription already started');
await this.watchBlocksAtChainHead();
await this.initBlockProcessingOnCompleteHandler();
await this.initEventProcessingOnCompleteHandler();
}
async stop (): Promise<void> {
this._baseEventWatcher.stop();
}
async watchBlocksAtChainHead (): Promise<void> {
log('Started watching upstream blocks...');
this._subscription = await this._ethClient.watchBlocks(async (value) => {
await this._baseEventWatcher.blocksHandler(value);
});
}
async initBlockProcessingOnCompleteHandler (): Promise<void> {
this._jobQueue.onComplete(QUEUE_BLOCK_PROCESSING, async (job) => {
const { id, data: { failed } } = job;
if (failed) {
log(`Job ${id} for queue ${QUEUE_BLOCK_PROCESSING} failed`);
return;
}
await this._baseEventWatcher.blockProcessingCompleteHandler(job);
});
}
async initEventProcessingOnCompleteHandler (): Promise<void> {
await this._jobQueue.onComplete(QUEUE_EVENT_PROCESSING, async (job) => {
const { id, data: { request, failed, state, createdOn } } = job;
if (failed) {
log(`Job ${id} for queue ${QUEUE_EVENT_PROCESSING} failed`);
return;
}
const dbEvent = await this._baseEventWatcher.eventProcessingCompleteHandler(job);
const timeElapsedInSeconds = (Date.now() - Date.parse(createdOn)) / 1000;
log(`Job onComplete event ${request.data.id} publish ${!!request.data.publish}`);
if (!failed && state === 'completed' && request.data.publish) {
// Check for max acceptable lag time between request and sending results to live subscribers.
if (timeElapsedInSeconds <= this._jobQueue.maxCompletionLag) {
await this.publishEventToSubscribers(dbEvent, timeElapsedInSeconds);
} else {
log(`event ${request.data.id} is too old (${timeElapsedInSeconds}s), not broadcasting to live subscribers`);
}
}
});
}
async publishEventToSubscribers (dbEvent: Event, timeElapsedInSeconds: number): Promise<void> {
if (dbEvent && dbEvent.eventName !== UNKNOWN_EVENT_NAME) {
const resultEvent = this._indexer.getResultEvent(dbEvent);
log(`pushing event to GQL subscribers (${timeElapsedInSeconds}s elapsed): ${resultEvent.event.__typename}`);
// Publishing the event here will result in pushing the payload to GQL subscribers for `onEvent`.
await this._pubsub.publish(EVENT, {
onEvent: resultEvent
});
}
}
}

97
packages/graph-test-watcher/src/fill.ts Normal file
View File

@ -0,0 +1,97 @@
//
// Copyright 2021 Vulcanize, Inc.
//
import assert from 'assert';
import 'reflect-metadata';
import yargs from 'yargs';
import { hideBin } from 'yargs/helpers';
import debug from 'debug';
import { PubSub } from 'apollo-server-express';
import { getCache } from '@vulcanize/cache';
import { EthClient } from '@vulcanize/ipld-eth-client';
import { getConfig, fillBlocks, JobQueue, DEFAULT_CONFIG_PATH, getCustomProvider } from '@vulcanize/util';
import { Database } from './database';
import { Indexer } from './indexer';
import { EventWatcher } from './events';
const log = debug('vulcanize:server');
export const main = async (): Promise<any> => {
const argv = await yargs(hideBin(process.argv)).parserConfiguration({
'parse-numbers': false
}).options({
configFile: {
alias: 'f',
type: 'string',
demandOption: true,
describe: 'configuration file path (toml)',
default: DEFAULT_CONFIG_PATH
},
startBlock: {
type: 'number',
demandOption: true,
describe: 'Block number to start processing at'
},
endBlock: {
type: 'number',
demandOption: true,
describe: 'Block number to stop processing at'
}
}).argv;
const config = await getConfig(argv.configFile);
assert(config.server, 'Missing server config');
const { upstream, database: dbConfig, jobQueue: jobQueueConfig } = config;
assert(dbConfig, 'Missing database config');
const db = new Database(dbConfig);
await db.init();
assert(upstream, 'Missing upstream config');
const { ethServer: { gqlPostgraphileEndpoint, rpcProviderEndpoint }, cache: cacheConfig } = upstream;
assert(gqlPostgraphileEndpoint, 'Missing upstream ethServer.gqlPostgraphileEndpoint');
const cache = await getCache(cacheConfig);
const ethClient = new EthClient({
gqlEndpoint: gqlPostgraphileEndpoint,
gqlSubscriptionEndpoint: gqlPostgraphileEndpoint,
cache
});
const postgraphileClient = new EthClient({
gqlEndpoint: gqlPostgraphileEndpoint,
cache
});
const ethProvider = getCustomProvider(rpcProviderEndpoint);
// Note: In-memory pubsub works fine for now, as each watcher is a single process anyway.
// Later: https://www.apollographql.com/docs/apollo-server/data/subscriptions/#production-pubsub-libraries
const pubsub = new PubSub();
const indexer = new Indexer(db, ethClient, postgraphileClient, ethProvider);
const { dbConnectionString, maxCompletionLagInSecs } = jobQueueConfig;
assert(dbConnectionString, 'Missing job queue db connection string');
const jobQueue = new JobQueue({ dbConnectionString, maxCompletionLag: maxCompletionLagInSecs });
await jobQueue.start();
const eventWatcher = new EventWatcher(ethClient, indexer, pubsub, jobQueue);
assert(jobQueueConfig, 'Missing job queue config');
await fillBlocks(jobQueue, indexer, ethClient, eventWatcher, argv);
};
main().catch(err => {
log(err);
}).finally(() => {
process.exit();
});

3
packages/graph-test-watcher/src/gql/index.ts Normal file
View File

@ -0,0 +1,3 @@
export * as mutations from './mutations';
export * as queries from './queries';
export * as subscriptions from './subscriptions';

4
packages/graph-test-watcher/src/gql/mutations/index.ts Normal file
View File

@ -0,0 +1,4 @@
import fs from 'fs';
import path from 'path';
export const watchContract = fs.readFileSync(path.join(__dirname, 'watchContract.gql'), 'utf8');

3
packages/graph-test-watcher/src/gql/mutations/watchContract.gql Normal file
View File

@ -0,0 +1,3 @@
mutation watchContract($contractAddress: String!, $startingBlock: Int){
watchContract(contractAddress: $contractAddress, startingBlock: $startingBlock)
}

8
packages/graph-test-watcher/src/gql/queries/_test.gql Normal file
View File

@ -0,0 +1,8 @@
query _test($blockHash: String!, $contractAddress: String!){
_test(blockHash: $blockHash, contractAddress: $contractAddress){
value
proof{
data
}
}
}

27
packages/graph-test-watcher/src/gql/queries/events.gql Normal file
View File

@ -0,0 +1,27 @@
query events($blockHash: String!, $contractAddress: String!, $name: String){
events(blockHash: $blockHash, contractAddress: $contractAddress, name: $name){
block{
hash
number
timestamp
parentHash
}
tx{
hash
index
from
to
}
contract
eventIndex
event{
... on TestEvent {
param1
param2
}
}
proof{
data
}
}
}

27
packages/graph-test-watcher/src/gql/queries/eventsInRange.gql Normal file
View File

@ -0,0 +1,27 @@
query eventsInRange($fromBlockNumber: Int!, $toBlockNumber: Int!){
eventsInRange(fromBlockNumber: $fromBlockNumber, toBlockNumber: $toBlockNumber){
block{
hash
number
timestamp
parentHash
}
tx{
hash
index
from
to
}
contract
eventIndex
event{
... on TestEvent {
param1
param2
}
}
proof{
data
}
}
}

8
packages/graph-test-watcher/src/gql/queries/getMethod.gql Normal file
View File

@ -0,0 +1,8 @@
query getMethod($blockHash: String!, $contractAddress: String!){
getMethod(blockHash: $blockHash, contractAddress: $contractAddress){
value
proof{
data
}
}
}

7
packages/graph-test-watcher/src/gql/queries/index.ts Normal file
View File

@ -0,0 +1,7 @@
import fs from 'fs';
import path from 'path';
export const events = fs.readFileSync(path.join(__dirname, 'events.gql'), 'utf8');
export const eventsInRange = fs.readFileSync(path.join(__dirname, 'eventsInRange.gql'), 'utf8');
export const getMethod = fs.readFileSync(path.join(__dirname, 'getMethod.gql'), 'utf8');
export const _test = fs.readFileSync(path.join(__dirname, '_test.gql'), 'utf8');

4
packages/graph-test-watcher/src/gql/subscriptions/index.ts Normal file
View File

@ -0,0 +1,4 @@
import fs from 'fs';
import path from 'path';
export const onEvent = fs.readFileSync(path.join(__dirname, 'onEvent.gql'), 'utf8');

27
packages/graph-test-watcher/src/gql/subscriptions/onEvent.gql Normal file
View File

@ -0,0 +1,27 @@
subscription onEvent{
onEvent{
block{
hash
number
timestamp
parentHash
}
tx{
hash
index
from
to
}
contract
eventIndex
event{
... on TestEvent {
param1
param2
}
}
proof{
data
}
}
}

51
packages/graph-test-watcher/src/hooks.example.ts Normal file
View File

@ -0,0 +1,51 @@
//
// Copyright 2021 Vulcanize, Inc.
//
import assert from 'assert';
import { Indexer, ResultEvent } from './indexer';
/**
* Event hook function.
* @param indexer Indexer instance that contains methods to fetch and update the contract values in the database.
* @param eventData ResultEvent object containing necessary information.
*/
export async function handleEvent (indexer: Indexer, eventData: ResultEvent): Promise<void> {
assert(indexer);
assert(eventData);
// The following code is for ERC20 contract implementation.
// Perform indexing based on the type of event.
switch (eventData.event.__typename) {
// In case of ERC20 'Transfer' event.
case 'TransferEvent': {
// On a transfer, balances for both parties change.
// Therefore, trigger indexing for both sender and receiver.
// Get event fields from eventData.
// const { from, to } = eventData.event;
// Update balance entry for sender in the database.
// await indexer.balanceOf(eventData.block.hash, eventData.contract, from);
// Update balance entry for receiver in the database.
// await indexer.balanceOf(eventData.block.hash, eventData.contract, to);
break;
}
// In case of ERC20 'Approval' event.
case 'ApprovalEvent': {
// On an approval, allowance for (owner, spender) combination changes.
// Get event fields from eventData.
// const { owner, spender } = eventData.event;
// Update allowance entry for (owner, spender) combination in the database.
// await indexer.allowance(eventData.block.hash, eventData.contract, owner, spender);
break;
}
}
}

19
packages/graph-test-watcher/src/hooks.ts Normal file
View File

@ -0,0 +1,19 @@
//
// Copyright 2021 Vulcanize, Inc.
//
import assert from 'assert';
import { Indexer, ResultEvent } from './indexer';
/**
* Event hook function.
* @param indexer Indexer instance that contains methods to fetch and update the contract values in the database.
* @param eventData ResultEvent object containing necessary information.
*/
export async function handleEvent (indexer: Indexer, eventData: ResultEvent): Promise<void> {
assert(indexer);
assert(eventData);
// Perform indexing based on the type of event.
}

382
packages/graph-test-watcher/src/indexer.ts Normal file
View File

@ -0,0 +1,382 @@
//
// Copyright 2021 Vulcanize, Inc.
//
import assert from 'assert';
import debug from 'debug';
import { DeepPartial } from 'typeorm';
import JSONbig from 'json-bigint';
import { ethers } from 'ethers';
import { JsonFragment } from '@ethersproject/abi';
import { BaseProvider } from '@ethersproject/providers';
import { EthClient } from '@vulcanize/ipld-eth-client';
import { StorageLayout } from '@vulcanize/solidity-mapper';
import { EventInterface, Indexer as BaseIndexer, ValueResult, UNKNOWN_EVENT_NAME } from '@vulcanize/util';
import { GraphWatcher } from '@vulcanize/graph-node';
import { Database } from './database';
import { Contract } from './entity/Contract';
import { Event } from './entity/Event';
import { SyncStatus } from './entity/SyncStatus';
import { BlockProgress } from './entity/BlockProgress';
import artifacts from './artifacts/Example.json';
import { handleEvent } from './hooks';
const log = debug('vulcanize:indexer');
const TEST_EVENT = 'Test';
export type ResultEvent = {
block: {
hash: string;
number: number;
timestamp: number;
parentHash: string;
};
tx: {
hash: string;
from: string;
to: string;
index: number;
};
contract: string;
eventIndex: number;
event: any;
proof: string;
}
export class Indexer {
_db: Database
_ethClient: EthClient
_ethProvider: BaseProvider
_postgraphileClient: EthClient;
_graphWatcher: GraphWatcher;
_baseIndexer: BaseIndexer;
_abi: JsonFragment[]
_storageLayout: StorageLayout
_contract: ethers.utils.Interface
constructor (db: Database, ethClient: EthClient, postgraphileClient: EthClient, ethProvider: BaseProvider, graphWatcher: GraphWatcher) {
assert(db);
assert(ethClient);
this._db = db;
this._ethClient = ethClient;
this._ethProvider = ethProvider;
this._postgraphileClient = postgraphileClient;
this._graphWatcher = graphWatcher;
this._baseIndexer = new BaseIndexer(this._db, this._ethClient, this._ethProvider);
const { abi, storageLayout } = artifacts;
assert(abi);
assert(storageLayout);
this._abi = abi;
this._storageLayout = storageLayout;
this._contract = new ethers.utils.Interface(this._abi);
}
getResultEvent (event: Event): ResultEvent {
const block = event.block;
const eventFields = JSONbig.parse(event.eventInfo);
const { tx } = JSON.parse(event.extraInfo);
return {
block: {
hash: block.blockHash,
number: block.blockNumber,
timestamp: block.blockTimestamp,
parentHash: block.parentHash
},
tx: {
hash: event.txHash,
from: tx.src,
to: tx.dst,
index: tx.index
},
contract: event.contract,
eventIndex: event.index,
event: {
__typename: `${event.eventName}Event`,
...eventFields
},
// TODO: Return proof only if requested.
proof: JSON.parse(event.proof)
};
}
async getMethod (blockHash: string, contractAddress: string): Promise<ValueResult> {
const entity = await this._db.getGetMethod({ blockHash, contractAddress });
if (entity) {
log('getMethod: db hit.');
return {
value: entity.value,
proof: JSON.parse(entity.proof)
};
}
log('getMethod: db miss, fetching from upstream server');
const contract = new ethers.Contract(contractAddress, this._abi, this._ethProvider);
const value = await contract.getMethod({ blockTag: blockHash });
const result: ValueResult = { value };
await this._db.saveGetMethod({ blockHash, contractAddress, value: result.value, proof: JSONbig.stringify(result.proof) });
return result;
}
async _test (blockHash: string, contractAddress: string): Promise<ValueResult> {
const entity = await this._db.get_test({ blockHash, contractAddress });
if (entity) {
log('_test: db hit.');
return {
value: entity.value,
proof: JSON.parse(entity.proof)
};
}
log('_test: db miss, fetching from upstream server');
const result = await this._baseIndexer.getStorageValue(
this._storageLayout,
blockHash,
contractAddress,
'_test'
);
await this._db.save_test({ blockHash, contractAddress, value: result.value, proof: JSONbig.stringify(result.proof) });
return result;
}
async triggerIndexingOnEvent (event: Event): Promise<void> {
const resultEvent = this.getResultEvent(event);
this._graphWatcher.handleEvent(resultEvent);
// Call custom hook function for indexing on event.
await handleEvent(this, resultEvent);
}
async processEvent (event: Event): Promise<void> {
// Trigger indexing of data based on the event.
await this.triggerIndexingOnEvent(event);
}
parseEventNameAndArgs (kind: string, logObj: any): any {
let eventName = UNKNOWN_EVENT_NAME;
let eventInfo = {};
const { topics, data } = logObj;
const logDescription = this._contract.parseLog({ data, topics });
switch (logDescription.name) {
case TEST_EVENT: {
eventName = logDescription.name;
const { param1, param2 } = logDescription.args;
eventInfo = {
param1,
param2
};
break;
}
}
return { eventName, eventInfo };
}
async watchContract (address: string, startingBlock: number): Promise<boolean> {
// Always use the checksum address (https://docs.ethers.io/v5/api/utils/address/#utils-getAddress).
await this._db.saveContract(ethers.utils.getAddress(address), 'Example', startingBlock);
return true;
}
async getEventsByFilter (blockHash: string, contract: string, name: string | null): Promise<Array<Event>> {
return this._baseIndexer.getEventsByFilter(blockHash, contract, name);
}
async isWatchedContract (address : string): Promise<Contract | undefined> {
return this._baseIndexer.isWatchedContract(address);
}
async getProcessedBlockCountForRange (fromBlockNumber: number, toBlockNumber: number): Promise<{ expected: number, actual: number }> {
return this._baseIndexer.getProcessedBlockCountForRange(fromBlockNumber, toBlockNumber);
}
async getEventsInRange (fromBlockNumber: number, toBlockNumber: number): Promise<Array<Event>> {
return this._baseIndexer.getEventsInRange(fromBlockNumber, toBlockNumber);
}
async getSyncStatus (): Promise<SyncStatus | undefined> {
return this._baseIndexer.getSyncStatus();
}
async updateSyncStatusIndexedBlock (blockHash: string, blockNumber: number): Promise<SyncStatus> {
return this._baseIndexer.updateSyncStatusIndexedBlock(blockHash, blockNumber);
}
async updateSyncStatusChainHead (blockHash: string, blockNumber: number): Promise<SyncStatus> {
return this._baseIndexer.updateSyncStatusChainHead(blockHash, blockNumber);
}
async updateSyncStatusCanonicalBlock (blockHash: string, blockNumber: number): Promise<SyncStatus> {
return this._baseIndexer.updateSyncStatusCanonicalBlock(blockHash, blockNumber);
}
async getBlock (blockHash: string): Promise<any> {
return this._baseIndexer.getBlock(blockHash);
}
async getEvent (id: string): Promise<Event | undefined> {
return this._baseIndexer.getEvent(id);
}
async getBlockProgress (blockHash: string): Promise<BlockProgress | undefined> {
return this._baseIndexer.getBlockProgress(blockHash);
}
async getBlocksAtHeight (height: number, isPruned: boolean): Promise<BlockProgress[]> {
return this._baseIndexer.getBlocksAtHeight(height, isPruned);
}
async getOrFetchBlockEvents (block: DeepPartial<BlockProgress>): Promise<Array<EventInterface>> {
return this._baseIndexer.getOrFetchBlockEvents(block, this._fetchAndSaveEvents.bind(this));
}
async getBlockEvents (blockHash: string): Promise<Array<Event>> {
return this._baseIndexer.getBlockEvents(blockHash);
}
async removeUnknownEvents (block: BlockProgress): Promise<void> {
return this._baseIndexer.removeUnknownEvents(Event, block);
}
async markBlocksAsPruned (blocks: BlockProgress[]): Promise<void> {
return this._baseIndexer.markBlocksAsPruned(blocks);
}
async updateBlockProgress (blockHash: string, lastProcessedEventIndex: number): Promise<void> {
return this._baseIndexer.updateBlockProgress(blockHash, lastProcessedEventIndex);
}
async getAncestorAtDepth (blockHash: string, depth: number): Promise<string> {
return this._baseIndexer.getAncestorAtDepth(blockHash, depth);
}
async _fetchAndSaveEvents ({ blockHash }: DeepPartial<BlockProgress>): Promise<void> {
assert(blockHash);
let { block, logs } = await this._ethClient.getLogs({ blockHash });
const {
allEthHeaderCids: {
nodes: [
{
ethTransactionCidsByHeaderId: {
nodes: transactions
}
}
]
}
} = await this._postgraphileClient.getBlockWithTransactions({ blockHash });
const transactionMap = transactions.reduce((acc: {[key: string]: any}, transaction: {[key: string]: any}) => {
acc[transaction.txHash] = transaction;
return acc;
}, {});
const dbEvents: Array<DeepPartial<Event>> = [];
for (let li = 0; li < logs.length; li++) {
const logObj = logs[li];
const {
topics,
data,
index: logIndex,
cid,
ipldBlock,
account: {
address
},
transaction: {
hash: txHash
},
receiptCID,
status
} = logObj;
if (status) {
let eventName = UNKNOWN_EVENT_NAME;
let eventInfo = {};
const tx = transactionMap[txHash];
const extraInfo = { topics, data, tx };
const contract = ethers.utils.getAddress(address);
const watchedContract = await this.isWatchedContract(contract);
if (watchedContract) {
const eventDetails = this.parseEventNameAndArgs(watchedContract.kind, logObj);
eventName = eventDetails.eventName;
eventInfo = eventDetails.eventInfo;
}
dbEvents.push({
index: logIndex,
txHash,
contract,
eventName,
eventInfo: JSONbig.stringify(eventInfo),
extraInfo: JSONbig.stringify(extraInfo),
proof: JSONbig.stringify({
data: JSONbig.stringify({
blockHash,
receiptCID,
log: {
cid,
ipldBlock
}
})
})
});
} else {
log(`Skipping event for receipt ${receiptCID} due to failed transaction.`);
}
}
const dbTx = await this._db.createTransactionRunner();
try {
block = {
blockHash,
blockNumber: block.number,
blockTimestamp: block.timestamp,
parentHash: block.parent.hash
};
await this._db.saveEvents(dbTx, block, dbEvents);
await dbTx.commitTransaction();
} catch (error) {
await dbTx.rollbackTransaction();
throw error;
} finally {
await dbTx.release();
}
}
}

136
packages/graph-test-watcher/src/job-runner.ts Normal file
View File

@ -0,0 +1,136 @@
//
// Copyright 2021 Vulcanize, Inc.
//
import assert from 'assert';
import 'reflect-metadata';
import yargs from 'yargs';
import { hideBin } from 'yargs/helpers';
import debug from 'debug';
import { getCache } from '@vulcanize/cache';
import { EthClient } from '@vulcanize/ipld-eth-client';
import {
getConfig,
JobQueue,
JobRunner as BaseJobRunner,
QUEUE_BLOCK_PROCESSING,
QUEUE_EVENT_PROCESSING,
JobQueueConfig,
DEFAULT_CONFIG_PATH,
getCustomProvider
} from '@vulcanize/util';
import { GraphWatcher } from '@vulcanize/graph-node';
import { Indexer } from './indexer';
import { Database } from './database';
const log = debug('vulcanize:job-runner');
export class JobRunner {
_indexer: Indexer
_jobQueue: JobQueue
_baseJobRunner: BaseJobRunner
_jobQueueConfig: JobQueueConfig
constructor (jobQueueConfig: JobQueueConfig, indexer: Indexer, jobQueue: JobQueue) {
this._indexer = indexer;
this._jobQueue = jobQueue;
this._jobQueueConfig = jobQueueConfig;
this._baseJobRunner = new BaseJobRunner(this._jobQueueConfig, this._indexer, this._jobQueue);
}
async start (): Promise<void> {
await this.subscribeBlockProcessingQueue();
await this.subscribeEventProcessingQueue();
}
async subscribeBlockProcessingQueue (): Promise<void> {
await this._jobQueue.subscribe(QUEUE_BLOCK_PROCESSING, async (job) => {
await this._baseJobRunner.processBlock(job);
await this._jobQueue.markComplete(job);
});
}
async subscribeEventProcessingQueue (): Promise<void> {
await this._jobQueue.subscribe(QUEUE_EVENT_PROCESSING, async (job) => {
const event = await this._baseJobRunner.processEvent(job);
const watchedContract = await this._indexer.isWatchedContract(event.contract);
if (watchedContract) {
await this._indexer.processEvent(event);
}
await this._jobQueue.markComplete(job);
});
}
}
export const main = async (): Promise<any> => {
const argv = await yargs(hideBin(process.argv))
.option('f', {
alias: 'config-file',
demandOption: true,
describe: 'configuration file path (toml)',
type: 'string',
default: DEFAULT_CONFIG_PATH
})
.argv;
const config = await getConfig(argv.f);
assert(config.server, 'Missing server config');
const { upstream, database: dbConfig, jobQueue: jobQueueConfig, server: { subgraphPath } } = config;
assert(dbConfig, 'Missing database config');
const db = new Database(dbConfig);
await db.init();
assert(upstream, 'Missing upstream config');
const { ethServer: { gqlApiEndpoint, gqlPostgraphileEndpoint, rpcProviderEndpoint }, cache: cacheConfig } = upstream;
assert(gqlApiEndpoint, 'Missing upstream ethServer.gqlApiEndpoint');
assert(gqlPostgraphileEndpoint, 'Missing upstream ethServer.gqlPostgraphileEndpoint');
const cache = await getCache(cacheConfig);
const ethClient = new EthClient({
gqlEndpoint: gqlApiEndpoint,
gqlSubscriptionEndpoint: gqlPostgraphileEndpoint,
cache
});
const postgraphileClient = new EthClient({
gqlEndpoint: gqlPostgraphileEndpoint,
cache
});
const graphWatcher = new GraphWatcher(subgraphPath);
await graphWatcher.init();
const ethProvider = getCustomProvider(rpcProviderEndpoint);
const indexer = new Indexer(db, ethClient, postgraphileClient, ethProvider, graphWatcher);
assert(jobQueueConfig, 'Missing job queue config');
const { dbConnectionString, maxCompletionLagInSecs } = jobQueueConfig;
assert(dbConnectionString, 'Missing job queue db connection string');
const jobQueue = new JobQueue({ dbConnectionString, maxCompletionLag: maxCompletionLagInSecs });
await jobQueue.start();
const jobRunner = new JobRunner(jobQueueConfig, indexer, jobQueue);
await jobRunner.start();
};
main().then(() => {
log('Starting job runner...');
}).catch(err => {
log(err);
});
process.on('uncaughtException', err => {
log('uncaughtException', err);
});

79
packages/graph-test-watcher/src/resolvers.ts Normal file
View File

@ -0,0 +1,79 @@
//
// Copyright 2021 Vulcanize, Inc.
//
import assert from 'assert';
import BigInt from 'apollo-type-bigint';
import debug from 'debug';
import { ValueResult } from '@vulcanize/util';
import { Indexer } from './indexer';
import { EventWatcher } from './events';
const log = debug('vulcanize:resolver');
export const createResolvers = async (indexer: Indexer, eventWatcher: EventWatcher): Promise<any> => {
assert(indexer);
return {
BigInt: new BigInt('bigInt'),
Event: {
__resolveType: (obj: any) => {
assert(obj.__typename);
return obj.__typename;
}
},
Subscription: {
onEvent: {
subscribe: () => eventWatcher.getEventIterator()
}
},
Mutation: {
watchContract: (_: any, { contractAddress, startingBlock = 1 }: { contractAddress: string, startingBlock: number }): Promise<boolean> => {
log('watchContract', contractAddress, startingBlock);
return indexer.watchContract(contractAddress, startingBlock);
}
},
Query: {
getMethod: (_: any, { blockHash, contractAddress }: { blockHash: string, contractAddress: string }): Promise<ValueResult> => {
log('getMethod', blockHash, contractAddress);
return indexer.getMethod(blockHash, contractAddress);
},
_test: (_: any, { blockHash, contractAddress }: { blockHash: string, contractAddress: string }): Promise<ValueResult> => {
log('_test', blockHash, contractAddress);
return indexer._test(blockHash, contractAddress);
},
events: async (_: any, { blockHash, contractAddress, name }: { blockHash: string, contractAddress: string, name: string }) => {
log('events', blockHash, contractAddress, name || '');
const block = await indexer.getBlockProgress(blockHash);
if (!block || !block.isComplete) {
throw new Error(`Block hash ${blockHash} number ${block?.blockNumber} not processed yet`);
}
const events = await indexer.getEventsByFilter(blockHash, contractAddress, name);
return events.map(event => indexer.getResultEvent(event));
},
eventsInRange: async (_: any, { fromBlockNumber, toBlockNumber }: { fromBlockNumber: number, toBlockNumber: number }) => {
log('eventsInRange', fromBlockNumber, toBlockNumber);
const { expected, actual } = await indexer.getProcessedBlockCountForRange(fromBlockNumber, toBlockNumber);
if (expected !== actual) {
throw new Error(`Range not available, expected ${expected}, got ${actual} blocks in range`);
}
const events = await indexer.getEventsInRange(fromBlockNumber, toBlockNumber);
return events.map(event => indexer.getResultEvent(event));
}
}
};
};

60
packages/graph-test-watcher/src/schema.gql Normal file
View File

@ -0,0 +1,60 @@
type Query {
events(blockHash: String!, contractAddress: String!, name: String): [ResultEvent!]
eventsInRange(fromBlockNumber: Int!, toBlockNumber: Int!): [ResultEvent!]
getMethod(blockHash: String!, contractAddress: String!): ResultString!
_test(blockHash: String!, contractAddress: String!): ResultBigInt!
}
type ResultEvent {
block: Block!
tx: Transaction!
contract: String!
eventIndex: Int!
event: Event!
proof: Proof
}
type Block {
hash: String!
number: Int!
timestamp: Int!
parentHash: String!
}
type Transaction {
hash: String!
index: Int!
from: String!
to: String!
}
union Event = TestEvent
type TestEvent {
param1: String!
param2: Int!
}
type Proof {
data: String!
}
type ResultString {
value: String!
proof: Proof
}
type ResultBigInt {
value: BigInt!
proof: Proof
}
scalar BigInt
type Mutation {
watchContract(contractAddress: String!, startingBlock: Int): Boolean!
}
type Subscription {
onEvent: ResultEvent!
}

121
packages/graph-test-watcher/src/server.ts Normal file
View File

@ -0,0 +1,121 @@
//
// Copyright 2021 Vulcanize, Inc.
//
import fs from 'fs';
import path from 'path';
import assert from 'assert';
import 'reflect-metadata';
import express, { Application } from 'express';
import { ApolloServer, PubSub } from 'apollo-server-express';
import yargs from 'yargs';
import { hideBin } from 'yargs/helpers';
import debug from 'debug';
import 'graphql-import-node';
import { createServer } from 'http';
import { getCache } from '@vulcanize/cache';
import { EthClient } from '@vulcanize/ipld-eth-client';
import { DEFAULT_CONFIG_PATH, getConfig, JobQueue, KIND_ACTIVE, getCustomProvider } from '@vulcanize/util';
import { GraphWatcher } from '@vulcanize/graph-node';
import { createResolvers } from './resolvers';
import { Indexer } from './indexer';
import { Database } from './database';
import { EventWatcher } from './events';
const log = debug('vulcanize:server');
export const main = async (): Promise<any> => {
const argv = await yargs(hideBin(process.argv))
.option('f', {
alias: 'config-file',
demandOption: true,
describe: 'configuration file path (toml)',
type: 'string',
default: DEFAULT_CONFIG_PATH
})
.argv;
const config = await getConfig(argv.f);
assert(config.server, 'Missing server config');
const { host, port, kind: watcherKind, subgraphPath } = config.server;
const { upstream, database: dbConfig, jobQueue: jobQueueConfig } = config;
assert(dbConfig, 'Missing database config');
const db = new Database(dbConfig);
await db.init();
assert(upstream, 'Missing upstream config');
const { ethServer: { gqlApiEndpoint, gqlPostgraphileEndpoint, rpcProviderEndpoint }, cache: cacheConfig } = upstream;
assert(gqlApiEndpoint, 'Missing upstream ethServer.gqlApiEndpoint');
assert(gqlPostgraphileEndpoint, 'Missing upstream ethServer.gqlPostgraphileEndpoint');
const cache = await getCache(cacheConfig);
const ethClient = new EthClient({
gqlEndpoint: gqlApiEndpoint,
gqlSubscriptionEndpoint: gqlPostgraphileEndpoint,
cache
});
const postgraphileClient = new EthClient({
gqlEndpoint: gqlPostgraphileEndpoint,
cache
});
const ethProvider = getCustomProvider(rpcProviderEndpoint);
const graphWatcher = new GraphWatcher(subgraphPath);
await graphWatcher.init();
const indexer = new Indexer(db, ethClient, postgraphileClient, ethProvider, graphWatcher);
// Note: In-memory pubsub works fine for now, as each watcher is a single process anyway.
// Later: https://www.apollographql.com/docs/apollo-server/data/subscriptions/#production-pubsub-libraries
const pubsub = new PubSub();
assert(jobQueueConfig, 'Missing job queue config');
const { dbConnectionString, maxCompletionLagInSecs } = jobQueueConfig;
assert(dbConnectionString, 'Missing job queue db connection string');
const jobQueue = new JobQueue({ dbConnectionString, maxCompletionLag: maxCompletionLagInSecs });
const eventWatcher = new EventWatcher(ethClient, indexer, pubsub, jobQueue);
if (watcherKind === KIND_ACTIVE) {
await jobQueue.start();
await eventWatcher.start();
}
const resolvers = await createResolvers(indexer, eventWatcher);
const app: Application = express();
const typeDefs = fs.readFileSync(path.join(__dirname, 'schema.gql')).toString();
const server = new ApolloServer({
typeDefs,
resolvers
});
await server.start();
server.applyMiddleware({ app });
const httpServer = createServer(app);
server.installSubscriptionHandlers(httpServer);
httpServer.listen(port, host, () => {
log(`Server is listening on host ${host} port ${port}`);
});
return { app, server };
};
main().then(() => {
log('Starting server...');
}).catch(err => {
log(err);
});

74
packages/graph-test-watcher/tsconfig.json Normal file
View File

@ -0,0 +1,74 @@
{
"compilerOptions": {
/* Visit https://aka.ms/tsconfig.json to read more about this file */
/* Basic Options */
// "incremental": true, /* Enable incremental compilation */
"target": "es5", /* Specify ECMAScript target version: 'ES3' (default), 'ES5', 'ES2015', 'ES2016', 'ES2017', 'ES2018', 'ES2019', 'ES2020', 'ES2021', or 'ESNEXT'. */
"module": "commonjs", /* Specify module code generation: 'none', 'commonjs', 'amd', 'system', 'umd', 'es2015', 'es2020', or 'ESNext'. */
// "lib": [], /* Specify library files to be included in the compilation. */
// "allowJs": true, /* Allow javascript files to be compiled. */
// "checkJs": true, /* Report errors in .js files. */
// "jsx": "preserve", /* Specify JSX code generation: 'preserve', 'react-native', 'react', 'react-jsx' or 'react-jsxdev'. */
// "declaration": true, /* Generates corresponding '.d.ts' file. */
// "declarationMap": true, /* Generates a sourcemap for each corresponding '.d.ts' file. */
// "sourceMap": true, /* Generates corresponding '.map' file. */
// "outFile": "./", /* Concatenate and emit output to single file. */
"outDir": "dist", /* Redirect output structure to the directory. */
// "rootDir": "./", /* Specify the root directory of input files. Use to control the output directory structure with --outDir. */
// "composite": true, /* Enable project compilation */
// "tsBuildInfoFile": "./", /* Specify file to store incremental compilation information */
// "removeComments": true, /* Do not emit comments to output. */
// "noEmit": true, /* Do not emit outputs. */
// "importHelpers": true, /* Import emit helpers from 'tslib'. */
// "downlevelIteration": true, /* Provide full support for iterables in 'for-of', spread, and destructuring when targeting 'ES5' or 'ES3'. */
// "isolatedModules": true, /* Transpile each file as a separate module (similar to 'ts.transpileModule'). */
/* Strict Type-Checking Options */
"strict": true, /* Enable all strict type-checking options. */
// "noImplicitAny": true, /* Raise error on expressions and declarations with an implied 'any' type. */
// "strictNullChecks": true, /* Enable strict null checks. */
// "strictFunctionTypes": true, /* Enable strict checking of function types. */
// "strictBindCallApply": true, /* Enable strict 'bind', 'call', and 'apply' methods on functions. */
// "strictPropertyInitialization": true, /* Enable strict checking of property initialization in classes. */
// "noImplicitThis": true, /* Raise error on 'this' expressions with an implied 'any' type. */
// "alwaysStrict": true, /* Parse in strict mode and emit "use strict" for each source file. */
/* Additional Checks */
// "noUnusedLocals": true, /* Report errors on unused locals. */
// "noUnusedParameters": true, /* Report errors on unused parameters. */
// "noImplicitReturns": true, /* Report error when not all code paths in function return a value. */
// "noFallthroughCasesInSwitch": true, /* Report errors for fallthrough cases in switch statement. */
// "noUncheckedIndexedAccess": true, /* Include 'undefined' in index signature results */
// "noImplicitOverride": true, /* Ensure overriding members in derived classes are marked with an 'override' modifier. */
// "noPropertyAccessFromIndexSignature": true, /* Require undeclared properties from index signatures to use element accesses. */
/* Module Resolution Options */
// "moduleResolution": "node", /* Specify module resolution strategy: 'node' (Node.js) or 'classic' (TypeScript pre-1.6). */
// "baseUrl": "./", /* Base directory to resolve non-absolute module names. */
// "paths": {}, /* A series of entries which re-map imports to lookup locations relative to the 'baseUrl'. */
// "rootDirs": [], /* List of root folders whose combined content represents the structure of the project at runtime. */
// "typeRoots": [], /* List of folders to include type definitions from. */
// "types": [], /* Type declaration files to be included in compilation. */
// "allowSyntheticDefaultImports": true, /* Allow default imports from modules with no default export. This does not affect code emit, just typechecking. */
"esModuleInterop": true, /* Enables emit interoperability between CommonJS and ES Modules via creation of namespace objects for all imports. Implies 'allowSyntheticDefaultImports'. */
// "preserveSymlinks": true, /* Do not resolve the real path of symlinks. */
// "allowUmdGlobalAccess": true, /* Allow accessing UMD globals from modules. */
/* Source Map Options */
// "sourceRoot": "", /* Specify the location where debugger should locate TypeScript files instead of source locations. */
// "mapRoot": "", /* Specify the location where debugger should locate map files instead of generated locations. */
// "inlineSourceMap": true, /* Emit a single file with source maps instead of having a separate file. */
// "inlineSources": true, /* Emit the source alongside the sourcemaps within a single file; requires '--inlineSourceMap' or '--sourceMap' to be set. */
/* Experimental Options */
"experimentalDecorators": true, /* Enables experimental support for ES7 decorators. */
// "emitDecoratorMetadata": true, /* Enables experimental support for emitting type metadata for decorators. */
/* Advanced Options */
"skipLibCheck": true, /* Skip type checking of declaration files. */
"forceConsistentCasingInFileNames": true, /* Disallow inconsistently-cased references to the same file. */
"resolveJsonModule": true /* Enabling the option allows importing JSON, and validating the types in that JSON file. */
},
"include": ["src/**/*"]
}

1
packages/util/src/config.ts
View File

@ -32,6 +32,7 @@ export interface ServerConfig {
checkpointing: boolean;
checkpointInterval: number;
ipfsApiAddr: string;
subgraphPath: string;
}
export interface UpstreamConfig {

10
yarn.lock
View File

@ -2417,10 +2417,10 @@
resolved "https://registry.yarnpkg.com/@types/http-errors/-/http-errors-1.8.0.tgz#682477dbbbd07cd032731cb3b0e7eaee3d026b69"
integrity sha512-2aoSC4UUbHDj2uCsCxcG/vRMXey/m17bC7UwitVm5hn22nI8O8Y9iDpA76Orc+DWkQ4zZrOKEshCqR/jSuXAHA==
"@types/js-yaml@^4.0.3":
version "4.0.3"
resolved "https://registry.yarnpkg.com/@types/js-yaml/-/js-yaml-4.0.3.tgz#9f33cd6fbf0d5ec575dc8c8fc69c7fec1b4eb200"
integrity sha512-5t9BhoORasuF5uCPr+d5/hdB++zRFUTMIZOzbNkr+jZh3yQht4HYbRDyj9fY8n2TZT30iW9huzav73x4NikqWg==
"@types/js-yaml@^4.0.3", "@types/js-yaml@^4.0.4":
version "4.0.4"
resolved "https://registry.yarnpkg.com/@types/js-yaml/-/js-yaml-4.0.4.tgz#cc38781257612581a1a0eb25f1709d2b06812fce"
integrity sha512-AuHubXUmg0AzkXH0Mx6sIxeY/1C110mm/EkE/gB1sTRz3h2dao2W/63q42SlVST+lICxz5Oki2hzYA6+KnnieQ==
"@types/json-bigint@^1.0.0":
version "1.0.0"
@ -8765,7 +8765,7 @@ js-yaml@^3.13.1, js-yaml@^3.14.0:
argparse "^1.0.7"
esprima "^4.0.0"
js-yaml@^4.0.0:
js-yaml@^4.0.0, js-yaml@^4.1.0:
version "4.1.0"
resolved "https://registry.yarnpkg.com/js-yaml/-/js-yaml-4.1.0.tgz#c1fb65f8f5017901cdd2c951864ba18458a10602"
integrity sha512-wpxZs9NoxZaJESJGIZTyDEaYpl0FKSA+FB9aJiyemKhMwkxQg63h4T1KJgUGHpTqPDNRcmmYLugrRjJlBtWvRA==