mirror of
https://github.com/cerc-io/watcher-ts
synced 2024-11-19 20:36:19 +00:00
Implement IPLD refactoring in codegen (#85)
This commit is contained in:
prathamesh0
parent
1e7d84879b
commit
3ed4ab95af
@ -35,9 +35,13 @@ columns:
|
||||
tsType: string
|
||||
columnType: Column
|
||||
- name: kind
|
||||
pgType: varchar
|
||||
tsType: string
|
||||
tsType: StateKind
|
||||
columnType: Column
|
||||
columnOptions:
|
||||
- option: type
|
||||
value: "'enum'"
|
||||
- option: enum
|
||||
value: StateKind
|
||||
- name: data
|
||||
pgType: bytea
|
||||
tsType: Buffer
|
||||
@ -50,6 +54,9 @@ imports:
|
||||
- Index
|
||||
- ManyToOne
|
||||
from: typeorm
|
||||
- toImport:
|
||||
- StateKind
|
||||
from: '@vulcanize/util'
|
||||
- toImport:
|
||||
- BlockProgress
|
||||
from: ./BlockProgress
|
||||
|
||||
@ -32,7 +32,6 @@ import { exportFill } from './fill';
|
||||
import { exportCheckpoint } from './checkpoint';
|
||||
import { exportState } from './export-state';
|
||||
import { importState } from './import-state';
|
||||
import { exportIPFS } from './ipfs';
|
||||
import { exportInspectCID } from './inspect-cid';
|
||||
|
||||
const main = async (): Promise<void> => {
|
||||
@ -305,11 +304,6 @@ function generateWatcher (contractStrings: string[], visitor: Visitor, argv: any
|
||||
: process.stdout;
|
||||
importState(outStream);
|
||||
|
||||
outStream = outputDir
|
||||
? fs.createWriteStream(path.join(outputDir, 'src/ipfs.ts'))
|
||||
: process.stdout;
|
||||
exportIPFS(outStream);
|
||||
|
||||
outStream = outputDir
|
||||
? fs.createWriteStream(path.join(outputDir, 'src/cli/inspect-cid.ts'))
|
||||
: process.stdout;
|
||||
|
||||
@ -1,21 +0,0 @@
|
||||
//
|
||||
// Copyright 2021 Vulcanize, Inc.
|
||||
//
|
||||
|
||||
import fs from 'fs';
|
||||
import path from 'path';
|
||||
import Handlebars from 'handlebars';
|
||||
import { Writable } from 'stream';
|
||||
|
||||
const TEMPLATE_FILE = './templates/ipfs-template.handlebars';
|
||||
|
||||
/**
|
||||
* Writes the ipfs.ts file generated from a template to a stream.
|
||||
* @param outStream A writable output stream to write the ipfs.ts file to.
|
||||
*/
|
||||
export function exportIPFS (outStream: Writable): void {
|
||||
const templateString = fs.readFileSync(path.resolve(__dirname, TEMPLATE_FILE)).toString();
|
||||
const template = Handlebars.compile(templateString);
|
||||
const ipfsString = template({});
|
||||
outStream.write(ipfsString);
|
||||
}
|
||||
@ -3,10 +3,10 @@
|
||||
//
|
||||
|
||||
import assert from 'assert';
|
||||
import { Connection, ConnectionOptions, DeepPartial, FindConditions, QueryRunner, FindManyOptions, MoreThan } from 'typeorm';
|
||||
import { Connection, ConnectionOptions, DeepPartial, FindConditions, QueryRunner, FindManyOptions } from 'typeorm';
|
||||
import path from 'path';
|
||||
|
||||
import { Database as BaseDatabase, QueryOptions, Where, MAX_REORG_DEPTH, DatabaseInterface } from '@vulcanize/util';
|
||||
import { IPLDDatabase as BaseDatabase, IPLDDatabaseInterface, QueryOptions, StateKind, Where } from '@vulcanize/util';
|
||||
|
||||
import { Contract } from './entity/Contract';
|
||||
import { Event } from './entity/Event';
|
||||
@ -17,9 +17,11 @@ import { IPLDBlock } from './entity/IPLDBlock';
|
||||
|
||||
{{#each queries as | query |}}
|
||||
import { {{query.entityName}} } from './entity/{{query.entityName}}';
|
||||
{{/each}}
|
||||
{{#unless @last}}
|
||||
|
||||
export class Database implements DatabaseInterface {
|
||||
{{/unless}}
|
||||
{{/each}}
|
||||
export class Database implements IPLDDatabaseInterface {
|
||||
_config: ConnectionOptions;
|
||||
_conn!: Connection;
|
||||
_baseDatabase: BaseDatabase;
|
||||
@ -78,156 +80,57 @@ export class Database implements DatabaseInterface {
|
||||
}
|
||||
|
||||
{{/each}}
|
||||
getNewIPLDBlock (): IPLDBlock {
|
||||
return new IPLDBlock();
|
||||
}
|
||||
|
||||
async getIPLDBlocks (where: FindConditions<IPLDBlock>): Promise<IPLDBlock[]> {
|
||||
const repo = this._conn.getRepository(IPLDBlock);
|
||||
return repo.find({ where, relations: ['block'] });
|
||||
|
||||
return this._baseDatabase.getIPLDBlocks(repo, where);
|
||||
}
|
||||
|
||||
async getLatestIPLDBlock (contractAddress: string, kind: string | null, blockNumber?: number): Promise<IPLDBlock | undefined> {
|
||||
async getLatestIPLDBlock (contractAddress: string, kind: StateKind | null, blockNumber?: number): Promise<IPLDBlock | undefined> {
|
||||
const repo = this._conn.getRepository(IPLDBlock);
|
||||
|
||||
let queryBuilder = repo.createQueryBuilder('ipld_block')
|
||||
.leftJoinAndSelect('ipld_block.block', 'block')
|
||||
.where('block.is_pruned = false')
|
||||
.andWhere('ipld_block.contract_address = :contractAddress', { contractAddress })
|
||||
.orderBy('block.block_number', 'DESC');
|
||||
|
||||
// Filter out blocks after the provided block number.
|
||||
if (blockNumber) {
|
||||
queryBuilder.andWhere('block.block_number <= :blockNumber', { blockNumber });
|
||||
}
|
||||
|
||||
// Filter using kind if specified else order by id to give preference to checkpoint.
|
||||
queryBuilder = kind
|
||||
? queryBuilder.andWhere('ipld_block.kind = :kind', { kind })
|
||||
: queryBuilder.andWhere('ipld_block.kind != :kind', { kind: 'diff_staged' })
|
||||
.addOrderBy('ipld_block.id', 'DESC');
|
||||
|
||||
return queryBuilder.getOne();
|
||||
return this._baseDatabase.getLatestIPLDBlock(repo, contractAddress, kind, blockNumber);
|
||||
}
|
||||
|
||||
async getPrevIPLDBlock (queryRunner: QueryRunner, blockHash: string, contractAddress: string, kind?: string): Promise<IPLDBlock | undefined> {
|
||||
const heirerchicalQuery = `
|
||||
WITH RECURSIVE cte_query AS
|
||||
(
|
||||
SELECT
|
||||
b.block_hash,
|
||||
b.block_number,
|
||||
b.parent_hash,
|
||||
1 as depth,
|
||||
i.id,
|
||||
i.kind
|
||||
FROM
|
||||
block_progress b
|
||||
LEFT JOIN
|
||||
ipld_block i ON i.block_id = b.id
|
||||
AND i.contract_address = $2
|
||||
WHERE
|
||||
b.block_hash = $1
|
||||
UNION ALL
|
||||
SELECT
|
||||
b.block_hash,
|
||||
b.block_number,
|
||||
b.parent_hash,
|
||||
c.depth + 1,
|
||||
i.id,
|
||||
i.kind
|
||||
FROM
|
||||
block_progress b
|
||||
LEFT JOIN
|
||||
ipld_block i
|
||||
ON i.block_id = b.id
|
||||
AND i.contract_address = $2
|
||||
INNER JOIN
|
||||
cte_query c ON c.parent_hash = b.block_hash
|
||||
WHERE
|
||||
c.depth < $3
|
||||
)
|
||||
SELECT
|
||||
block_number, id, kind
|
||||
FROM
|
||||
cte_query
|
||||
ORDER BY block_number DESC, id DESC
|
||||
`;
|
||||
|
||||
// Fetching block and id for previous IPLDBlock in frothy region.
|
||||
const queryResult = await queryRunner.query(heirerchicalQuery, [blockHash, contractAddress, MAX_REORG_DEPTH]);
|
||||
const latestRequiredResult = kind
|
||||
? queryResult.find((obj: any) => obj.kind === kind)
|
||||
: queryResult.find((obj: any) => obj.id);
|
||||
|
||||
let result: IPLDBlock | undefined;
|
||||
if (latestRequiredResult) {
|
||||
result = await queryRunner.manager.findOne(IPLDBlock, { id: latestRequiredResult.id }, { relations: ['block'] });
|
||||
} else {
|
||||
// If IPLDBlock not found in frothy region get latest IPLDBlock in the pruned region.
|
||||
// Filter out IPLDBlocks from pruned blocks.
|
||||
const canonicalBlockNumber = queryResult.pop().block_number + 1;
|
||||
|
||||
let queryBuilder = queryRunner.manager.createQueryBuilder(IPLDBlock, 'ipld_block')
|
||||
.leftJoinAndSelect('ipld_block.block', 'block')
|
||||
.where('block.is_pruned = false')
|
||||
.andWhere('ipld_block.contract_address = :contractAddress', { contractAddress })
|
||||
.andWhere('block.block_number <= :canonicalBlockNumber', { canonicalBlockNumber })
|
||||
.orderBy('block.block_number', 'DESC');
|
||||
|
||||
// Filter using kind if specified else order by id to give preference to checkpoint.
|
||||
queryBuilder = kind
|
||||
? queryBuilder.andWhere('ipld_block.kind = :kind', { kind })
|
||||
: queryBuilder.addOrderBy('ipld_block.id', 'DESC');
|
||||
|
||||
result = await queryBuilder.getOne();
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
|
||||
// Fetch all diff IPLDBlocks after the specified checkpoint.
|
||||
async getDiffIPLDBlocksByCheckpoint (contractAddress: string, checkpointBlockNumber: number): Promise<IPLDBlock[]> {
|
||||
async getPrevIPLDBlock (blockHash: string, contractAddress: string, kind?: string): Promise<IPLDBlock | undefined> {
|
||||
const repo = this._conn.getRepository(IPLDBlock);
|
||||
|
||||
return repo.find({
|
||||
relations: ['block'],
|
||||
where: {
|
||||
contractAddress,
|
||||
kind: 'diff',
|
||||
block: {
|
||||
isPruned: false,
|
||||
blockNumber: MoreThan(checkpointBlockNumber)
|
||||
}
|
||||
},
|
||||
order: {
|
||||
block: 'ASC'
|
||||
}
|
||||
});
|
||||
return this._baseDatabase.getPrevIPLDBlock(repo, blockHash, contractAddress, kind);
|
||||
}
|
||||
|
||||
async saveOrUpdateIPLDBlock (ipldBlock: IPLDBlock): Promise<IPLDBlock> {
|
||||
// Fetch all diff IPLDBlocks after the specified block number.
|
||||
async getDiffIPLDBlocksByBlocknumber (contractAddress: string, blockNumber: number): Promise<IPLDBlock[]> {
|
||||
const repo = this._conn.getRepository(IPLDBlock);
|
||||
return repo.save(ipldBlock);
|
||||
|
||||
return this._baseDatabase.getDiffIPLDBlocksByBlocknumber(repo, contractAddress, blockNumber);
|
||||
}
|
||||
|
||||
async getHookStatus (queryRunner: QueryRunner): Promise<HookStatus | undefined> {
|
||||
const repo = queryRunner.manager.getRepository(HookStatus);
|
||||
async saveOrUpdateIPLDBlock (dbTx: QueryRunner, ipldBlock: IPLDBlock): Promise<IPLDBlock> {
|
||||
const repo = dbTx.manager.getRepository(IPLDBlock);
|
||||
|
||||
return repo.findOne();
|
||||
return this._baseDatabase.saveOrUpdateIPLDBlock(repo, ipldBlock);
|
||||
}
|
||||
|
||||
async removeIPLDBlocks (dbTx: QueryRunner, blockNumber: number, kind: string): Promise<void> {
|
||||
const repo = dbTx.manager.getRepository(IPLDBlock);
|
||||
|
||||
await this._baseDatabase.removeIPLDBlocks(repo, blockNumber, kind);
|
||||
}
|
||||
|
||||
async getHookStatus (): Promise<HookStatus | undefined> {
|
||||
const repo = this._conn.getRepository(HookStatus);
|
||||
|
||||
return this._baseDatabase.getHookStatus(repo);
|
||||
}
|
||||
|
||||
async updateHookStatusProcessedBlock (queryRunner: QueryRunner, blockNumber: number, force?: boolean): Promise<HookStatus> {
|
||||
const repo = queryRunner.manager.getRepository(HookStatus);
|
||||
let entity = await repo.findOne();
|
||||
|
||||
if (!entity) {
|
||||
entity = repo.create({
|
||||
latestProcessedBlockNumber: blockNumber
|
||||
});
|
||||
}
|
||||
|
||||
if (force || blockNumber > entity.latestProcessedBlockNumber) {
|
||||
entity.latestProcessedBlockNumber = blockNumber;
|
||||
}
|
||||
|
||||
return repo.save(entity);
|
||||
return this._baseDatabase.updateHookStatusProcessedBlock(repo, blockNumber, force);
|
||||
}
|
||||
|
||||
async getContracts (): Promise<Contract[]> {
|
||||
@ -236,12 +139,6 @@ export class Database implements DatabaseInterface {
|
||||
return this._baseDatabase.getContracts(repo);
|
||||
}
|
||||
|
||||
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();
|
||||
}
|
||||
@ -276,7 +173,7 @@ export class Database implements DatabaseInterface {
|
||||
return this._baseDatabase.saveEvents(blockRepo, eventRepo, block, events);
|
||||
}
|
||||
|
||||
async saveContract (address: string, kind: string, checkpoint: boolean, startingBlock: number): Promise<Contract> {
|
||||
async saveContract (queryRunner: QueryRunner, address: string, kind: string, checkpoint: boolean, startingBlock: number): Promise<Contract> {
|
||||
const repo = queryRunner.manager.getRepository(Contract);
|
||||
|
||||
return this._baseDatabase.saveContract(repo, address, kind, checkpoint, startingBlock);
|
||||
|
||||
@ -20,12 +20,14 @@ export class {{className}} {{~#if implements}} implements {{implements}} {{~/if}
|
||||
{{#each columns as | column |}}
|
||||
{{#if (compare column.columnType 'ManyToOne')}}
|
||||
@{{column.columnType}}({{column.lhs}} => {{column.rhs}}
|
||||
{{~#if column.columnOptions}}, {{/if}}
|
||||
{{~else}}
|
||||
@{{column.columnType}}(
|
||||
{{~#if column.pgType~}} '{{column.pgType}}'
|
||||
{{~#if column.columnOptions}}, {{/if}}
|
||||
{{~/if}}
|
||||
{{~/if}}
|
||||
{{~#if column.columnOptions}}, {
|
||||
{{~#if column.columnOptions}}{
|
||||
{{~#each column.columnOptions}} {{this.option}}: {{{this.value}}}
|
||||
{{~#unless @last}},{{/unless}}
|
||||
{{~/each}} }
|
||||
|
||||
@ -80,7 +80,10 @@ export class EventWatcher implements EventWatcherInterface {
|
||||
|
||||
await this._baseEventWatcher.blockProcessingCompleteHandler(job);
|
||||
|
||||
await this.createHooksJob(kind);
|
||||
// If it's a pruning job: Create a hooks job.
|
||||
if (kind === JOB_KIND_PRUNE) {
|
||||
await this.createHooksJob();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@ -148,20 +151,18 @@ export class EventWatcher implements EventWatcherInterface {
|
||||
}
|
||||
}
|
||||
|
||||
async createHooksJob (kind: string): Promise<void> {
|
||||
// If it's a pruning job: Create a hook job for the latest canonical block.
|
||||
if (kind === JOB_KIND_PRUNE) {
|
||||
const latestCanonicalBlock = await this._indexer.getLatestCanonicalBlock();
|
||||
assert(latestCanonicalBlock);
|
||||
async createHooksJob (): Promise<void> {
|
||||
// Get the latest canonical block
|
||||
const latestCanonicalBlock = await this._indexer.getLatestCanonicalBlock();
|
||||
|
||||
await this._jobQueue.pushJob(
|
||||
QUEUE_HOOKS,
|
||||
{
|
||||
blockHash: latestCanonicalBlock.blockHash,
|
||||
blockNumber: latestCanonicalBlock.blockNumber
|
||||
}
|
||||
);
|
||||
}
|
||||
// Create a hooks job for parent block of latestCanonicalBlock because pruning for first block is skipped as it is assumed to be a canonical block.
|
||||
await this._jobQueue.pushJob(
|
||||
QUEUE_HOOKS,
|
||||
{
|
||||
blockHash: latestCanonicalBlock.parentHash,
|
||||
blockNumber: latestCanonicalBlock.blockNumber - 1
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
async createCheckpointJob (blockHash: string, blockNumber: number): Promise<void> {
|
||||
|
||||
@ -61,7 +61,7 @@ export const main = async (): Promise<any> => {
|
||||
const graphDb = new GraphDatabase(config.database, path.resolve(__dirname, 'entity/*'));
|
||||
await graphDb.init();
|
||||
|
||||
const graphWatcher = new GraphWatcher(graphDb, postgraphileClient, ethProvider, config.server.subgraphPath);
|
||||
const graphWatcher = new GraphWatcher(graphDb, postgraphileClient, ethProvider, config.server);
|
||||
|
||||
// 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
|
||||
|
||||
@ -14,12 +14,13 @@ const ACCOUNTS = [
|
||||
];
|
||||
|
||||
/**
|
||||
* Hook function to create an initial checkpoint.
|
||||
* Hook function to store an initial state.
|
||||
* @param indexer Indexer instance.
|
||||
* @param block Concerned block.
|
||||
* @param blockHash Hash of the concerned block.
|
||||
* @param contractAddress Address of the concerned contract.
|
||||
* @returns Data block to be stored.
|
||||
*/
|
||||
export async function createInitialCheckpoint (indexer: Indexer, contractAddress: string, blockHash: string): Promise<void> {
|
||||
export async function createInitialState (indexer: Indexer, contractAddress: string, blockHash: string): Promise<any> { assert(indexer);
|
||||
assert(indexer);
|
||||
assert(blockHash);
|
||||
assert(contractAddress);
|
||||
@ -33,11 +34,12 @@ export async function createInitialCheckpoint (indexer: Indexer, contractAddress
|
||||
ipldBlockData = updateStateForMappingType(ipldBlockData, '_balances', [account], balance.value.toString());
|
||||
}
|
||||
|
||||
await indexer.createCheckpoint(contractAddress, blockHash, ipldBlockData);
|
||||
// Return initial state data to be saved.
|
||||
return ipldBlockData;
|
||||
}
|
||||
|
||||
/**
|
||||
* Hook function to create state diffs.
|
||||
* Hook function to create state diff.
|
||||
* @param indexer Indexer instance that contains methods to fetch the contract varaiable values.
|
||||
* @param blockHash Block hash of the concerned block.
|
||||
*/
|
||||
@ -100,6 +102,7 @@ export async function createStateDiff (indexer: Indexer, blockHash: string): Pro
|
||||
}
|
||||
}
|
||||
|
||||
// Use indexer.createStateDiff() method to create a custom state.
|
||||
await indexer.createDiff(contractAddress, blockHash, ipldBlockData);
|
||||
}
|
||||
}
|
||||
@ -124,6 +127,7 @@ export async function createStateCheckpoint (indexer: Indexer, contractAddress:
|
||||
ipldBlockData = updateStateForMappingType(ipldBlockData, '_balances', [account], balance.value.toString());
|
||||
}
|
||||
|
||||
// Use indexer.createStateCheckpoint() method to create a custom checkpoint.
|
||||
await indexer.createCheckpoint(contractAddress, blockHash, ipldBlockData);
|
||||
|
||||
return false;
|
||||
@ -132,7 +136,7 @@ export async function createStateCheckpoint (indexer: Indexer, contractAddress:
|
||||
/**
|
||||
* 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.
|
||||
* @param eventData ResultEvent object containing event information.
|
||||
*/
|
||||
export async function handleEvent (indexer: Indexer, eventData: ResultEvent): Promise<void> {
|
||||
assert(indexer);
|
||||
|
||||
@ -7,19 +7,34 @@ import debug from 'debug';
|
||||
import { DeepPartial, FindConditions, FindManyOptions } from 'typeorm';
|
||||
import JSONbig from 'json-bigint';
|
||||
import { ethers } from 'ethers';
|
||||
import { sha256 } from 'multiformats/hashes/sha2';
|
||||
import { CID } from 'multiformats/cid';
|
||||
import _ from 'lodash';
|
||||
|
||||
import { JsonFragment } from '@ethersproject/abi';
|
||||
import { BaseProvider } from '@ethersproject/providers';
|
||||
import * as codec from '@ipld/dag-cbor';
|
||||
import { EthClient } from '@vulcanize/ipld-eth-client';
|
||||
import { StorageLayout } from '@vulcanize/solidity-mapper';
|
||||
import { Indexer as BaseIndexer, IndexerInterface, ValueResult, UNKNOWN_EVENT_NAME, ServerConfig, Where, QueryOptions, updateStateForElementaryType, updateStateForMappingType, JobQueue } from '@vulcanize/util';
|
||||
import {
|
||||
IPLDIndexer as BaseIndexer,
|
||||
IndexerInterface,
|
||||
ValueResult,
|
||||
UNKNOWN_EVENT_NAME,
|
||||
ServerConfig,
|
||||
JobQueue,
|
||||
Where,
|
||||
QueryOptions,
|
||||
updateStateForElementaryType,
|
||||
updateStateForMappingType,
|
||||
BlockHeight,
|
||||
IPFSClient,
|
||||
StateKind
|
||||
} from '@vulcanize/util';
|
||||
import { GraphWatcher } from '@vulcanize/graph-node';
|
||||
|
||||
{{#each inputFileNames as | inputFileName |}}
|
||||
import artifacts from './artifacts/{{inputFileName}}.json';
|
||||
{{/each}}
|
||||
import { Database } from './database';
|
||||
import { createInitialState, handleEvent, createStateDiff, createStateCheckpoint } from './hooks';
|
||||
import { Contract } from './entity/Contract';
|
||||
import { Event } from './entity/Event';
|
||||
import { SyncStatus } from './entity/SyncStatus';
|
||||
@ -27,13 +42,6 @@ import { HookStatus } from './entity/HookStatus';
|
||||
import { BlockProgress } from './entity/BlockProgress';
|
||||
import { IPLDBlock } from './entity/IPLDBlock';
|
||||
|
||||
{{#each inputFileNames as | inputFileName |}}
|
||||
import artifacts from './artifacts/{{inputFileName}}.json';
|
||||
{{/each}}
|
||||
|
||||
import { createInitialCheckpoint, handleEvent, createStateDiff, createStateCheckpoint } from './hooks';
|
||||
import { IPFSClient } from './ipfs';
|
||||
|
||||
const log = debug('vulcanize:indexer');
|
||||
|
||||
{{#each events as | event |}}
|
||||
@ -93,6 +101,9 @@ export class Indexer implements IndexerInterface {
|
||||
|
||||
_ipfsClient: IPFSClient
|
||||
|
||||
_entityTypesMap: Map<string, { [key: string]: string }>
|
||||
_relationsMap: Map<any, { [key: string]: any }>
|
||||
|
||||
constructor (serverConfig: ServerConfig, db: Database, ethClient: EthClient, postgraphileClient: EthClient, ethProvider: BaseProvider, jobQueue: JobQueue, graphWatcher: GraphWatcher) {
|
||||
assert(db);
|
||||
assert(ethClient);
|
||||
@ -103,7 +114,8 @@ export class Indexer implements IndexerInterface {
|
||||
this._postgraphileClient = postgraphileClient;
|
||||
this._ethProvider = ethProvider;
|
||||
this._serverConfig = serverConfig;
|
||||
this._baseIndexer = new BaseIndexer(this._db, this._ethClient, this._postgraphileClient, this._ethProvider, jobQueue);
|
||||
this._ipfsClient = new IPFSClient(this._serverConfig.ipfsApiAddr);
|
||||
this._baseIndexer = new BaseIndexer(this._serverConfig, this._db, this._ethClient, this._postgraphileClient, this._ethProvider, jobQueue, this._ipfsClient);
|
||||
this._graphWatcher = graphWatcher;
|
||||
|
||||
const { abi, storageLayout } = artifacts;
|
||||
@ -116,11 +128,13 @@ export class Indexer implements IndexerInterface {
|
||||
|
||||
this._contract = new ethers.utils.Interface(this._abi);
|
||||
|
||||
this._ipfsClient = new IPFSClient(this._serverConfig.ipfsApiAddr);
|
||||
this._entityTypesMap = new Map();
|
||||
this._relationsMap = new Map();
|
||||
}
|
||||
|
||||
async init (): Promise<void> {
|
||||
await this._baseIndexer.fetchContracts();
|
||||
await this._baseIndexer.fetchIPLDStatus();
|
||||
}
|
||||
|
||||
getResultEvent (event: Event): ResultEvent {
|
||||
@ -258,315 +272,110 @@ export class Indexer implements IndexerInterface {
|
||||
}
|
||||
|
||||
{{/each}}
|
||||
async pushToIPFS (data: any): Promise<void> {
|
||||
await this._baseIndexer.pushToIPFS(data);
|
||||
}
|
||||
|
||||
async processInitialState (contractAddress: string, blockHash: string): Promise<any> {
|
||||
// Call initial state hook.
|
||||
return createInitialState(this, contractAddress, blockHash);
|
||||
}
|
||||
|
||||
async processStateCheckpoint (contractAddress: string, blockHash: string): Promise<boolean> {
|
||||
// Call checkpoint hook.
|
||||
return createStateCheckpoint(this, contractAddress, blockHash);
|
||||
}
|
||||
|
||||
async processCanonicalBlock (job: any): Promise<void> {
|
||||
const { data: { blockHash } } = job;
|
||||
|
||||
// Finalize staged diff blocks if any.
|
||||
await this.finalizeDiffStaged(blockHash);
|
||||
await this._baseIndexer.finalizeDiffStaged(blockHash);
|
||||
|
||||
// Call custom stateDiff hook.
|
||||
await createStateDiff(this, blockHash);
|
||||
}
|
||||
|
||||
async createDiffStaged (contractAddress: string, blockHash: string, data: any): Promise<void> {
|
||||
const block = await this.getBlockProgress(blockHash);
|
||||
assert(block);
|
||||
|
||||
// Create a staged diff block.
|
||||
const ipldBlock = await this.prepareIPLDBlock(block, contractAddress, data, 'diff_staged');
|
||||
await this.saveOrUpdateIPLDBlock(ipldBlock);
|
||||
}
|
||||
|
||||
async finalizeDiffStaged (blockHash: string): Promise<void> {
|
||||
const block = await this.getBlockProgress(blockHash);
|
||||
assert(block);
|
||||
|
||||
// Get all the staged diff blocks for the given blockHash.
|
||||
const stagedBlocks = await this._db.getIPLDBlocks({ block, kind: 'diff_staged' });
|
||||
|
||||
// For each staged block, create a diff block.
|
||||
for (const stagedBlock of stagedBlocks) {
|
||||
const data = codec.decode(Buffer.from(stagedBlock.data));
|
||||
await this.createDiff(stagedBlock.contractAddress, stagedBlock.block.blockHash, data);
|
||||
}
|
||||
|
||||
// Remove all the staged diff blocks for current blockNumber.
|
||||
await this.removeStagedIPLDBlocks(block.blockNumber);
|
||||
}
|
||||
|
||||
async createDiff (contractAddress: string, blockHash: string, data: any): Promise<void> {
|
||||
const block = await this.getBlockProgress(blockHash);
|
||||
assert(block);
|
||||
|
||||
// Fetch the latest checkpoint for the contract.
|
||||
const checkpoint = await this.getLatestIPLDBlock(contractAddress, 'checkpoint');
|
||||
|
||||
// There should be an initial checkpoint at least.
|
||||
// Assumption: There should be no events for the contract at the starting block.
|
||||
assert(checkpoint, 'Initial checkpoint doesn\'t exist');
|
||||
|
||||
// Check if the latest checkpoint is in the same block.
|
||||
assert(checkpoint.block.blockHash !== block.blockHash, 'Checkpoint already created for the block hash.');
|
||||
|
||||
const ipldBlock = await this.prepareIPLDBlock(block, contractAddress, data, 'diff');
|
||||
await this.saveOrUpdateIPLDBlock(ipldBlock);
|
||||
}
|
||||
|
||||
async processCheckpoint (job: any): Promise<void> {
|
||||
// Return if checkpointInterval is <= 0.
|
||||
const checkpointInterval = this._serverConfig.checkpointInterval;
|
||||
if (checkpointInterval <= 0) return;
|
||||
|
||||
const { data: { blockHash, blockNumber } } = job;
|
||||
|
||||
// Get all the contracts.
|
||||
const contracts = await this._db.getContracts({});
|
||||
|
||||
// For each contract, merge the diff till now to create a checkpoint.
|
||||
for (const contract of contracts) {
|
||||
// Check if contract has checkpointing on.
|
||||
if (contract.checkpoint) {
|
||||
// If a checkpoint doesn't already exist and blockNumber is equal to startingBlock, create an initial checkpoint.
|
||||
const checkpointBlock = await this.getLatestIPLDBlock(contract.address, 'checkpoint');
|
||||
|
||||
if (!checkpointBlock) {
|
||||
if (blockNumber === contract.startingBlock) {
|
||||
// Call initial checkpoint hook.
|
||||
await createInitialCheckpoint(this, contract.address, blockHash);
|
||||
}
|
||||
} else {
|
||||
await this.createCheckpoint(contract.address, blockHash, null, checkpointInterval);
|
||||
}
|
||||
}
|
||||
}
|
||||
const { data: { blockHash } } = job;
|
||||
await this._baseIndexer.processCheckpoint(this, blockHash, checkpointInterval);
|
||||
}
|
||||
|
||||
async processCLICheckpoint (contractAddress: string, blockHash?: string): Promise<string | undefined> {
|
||||
const checkpointBlockHash = await this.createCheckpoint(contractAddress, blockHash);
|
||||
assert(checkpointBlockHash);
|
||||
|
||||
const block = await this.getBlockProgress(checkpointBlockHash);
|
||||
const checkpointIPLDBlocks = await this._db.getIPLDBlocks({ block, contractAddress, kind: 'checkpoint' });
|
||||
|
||||
// There can be at most one IPLDBlock for a (block, contractAddress, kind) combination.
|
||||
assert(checkpointIPLDBlocks.length <= 1);
|
||||
const checkpointIPLDBlock = checkpointIPLDBlocks[0];
|
||||
|
||||
const checkpointData = this.getIPLDData(checkpointIPLDBlock);
|
||||
|
||||
await this.pushToIPFS(checkpointData);
|
||||
|
||||
return checkpointBlockHash;
|
||||
}
|
||||
|
||||
async createCheckpoint (contractAddress: string, blockHash?: string, data?: any, checkpointInterval?: number): Promise<string | undefined> {
|
||||
const syncStatus = await this.getSyncStatus();
|
||||
assert(syncStatus);
|
||||
|
||||
// Getting the current block.
|
||||
let currentBlock;
|
||||
|
||||
if (blockHash) {
|
||||
currentBlock = await this.getBlockProgress(blockHash);
|
||||
} else {
|
||||
// In case of empty blockHash from checkpoint CLI, get the latest canonical block for the checkpoint.
|
||||
currentBlock = await this.getBlockProgress(syncStatus.latestCanonicalBlockHash);
|
||||
}
|
||||
|
||||
assert(currentBlock);
|
||||
|
||||
// Data is passed in case of initial checkpoint and checkpoint hook.
|
||||
// Assumption: There should be no events for the contract at the starting block.
|
||||
if (data) {
|
||||
const ipldBlock = await this.prepareIPLDBlock(currentBlock, contractAddress, data, 'checkpoint');
|
||||
await this.saveOrUpdateIPLDBlock(ipldBlock);
|
||||
|
||||
return;
|
||||
}
|
||||
|
||||
// If data is not passed, create from previous checkpoint and diffs after that.
|
||||
|
||||
// Make sure the block is marked complete.
|
||||
assert(currentBlock.isComplete, 'Block for a checkpoint should be marked as complete');
|
||||
|
||||
// Make sure the block is in the pruned region.
|
||||
assert(currentBlock.blockNumber <= syncStatus.latestCanonicalBlockNumber, 'Block for a checkpoint should be in the pruned region');
|
||||
|
||||
// Fetch the latest checkpoint for the contract.
|
||||
const checkpointBlock = await this.getLatestIPLDBlock(contractAddress, 'checkpoint', currentBlock.blockNumber);
|
||||
assert(checkpointBlock);
|
||||
|
||||
// Check (only if checkpointInterval is passed) if it is time for a new checkpoint.
|
||||
if (checkpointInterval && checkpointBlock.block.blockNumber > (currentBlock.blockNumber - checkpointInterval)) {
|
||||
return;
|
||||
}
|
||||
|
||||
// Call state checkpoint hook and check if default checkpoint is disabled.
|
||||
const disableDefaultCheckpoint = await createStateCheckpoint(this, contractAddress, currentBlock.blockHash);
|
||||
|
||||
if (disableDefaultCheckpoint) {
|
||||
// Return if default checkpoint is disabled.
|
||||
// Return block hash for checkpoint CLI.
|
||||
return currentBlock.blockHash;
|
||||
}
|
||||
|
||||
const { block: { blockNumber: checkpointBlockNumber } } = checkpointBlock;
|
||||
|
||||
// Fetching all diff blocks after checkpoint.
|
||||
const diffBlocks = await this.getDiffIPLDBlocksByCheckpoint(contractAddress, checkpointBlockNumber);
|
||||
|
||||
const checkpointBlockData = codec.decode(Buffer.from(checkpointBlock.data)) as any;
|
||||
data = {
|
||||
state: checkpointBlockData.state
|
||||
};
|
||||
|
||||
for (const diffBlock of diffBlocks) {
|
||||
const diff = codec.decode(Buffer.from(diffBlock.data)) as any;
|
||||
data.state = _.merge(data.state, diff.state);
|
||||
}
|
||||
|
||||
const ipldBlock = await this.prepareIPLDBlock(currentBlock, contractAddress, data, 'checkpoint');
|
||||
await this.saveOrUpdateIPLDBlock(ipldBlock);
|
||||
|
||||
return currentBlock.blockHash;
|
||||
}
|
||||
|
||||
getIPLDData (ipldBlock: IPLDBlock): any {
|
||||
return codec.decode(Buffer.from(ipldBlock.data));
|
||||
}
|
||||
|
||||
async getIPLDBlocksByHash (blockHash: string): Promise<IPLDBlock[]> {
|
||||
const block = await this.getBlockProgress(blockHash);
|
||||
assert(block);
|
||||
|
||||
return this._db.getIPLDBlocks({ block });
|
||||
}
|
||||
|
||||
async getIPLDBlockByCid (cid: string): Promise<IPLDBlock | undefined> {
|
||||
const ipldBlocks = await this._db.getIPLDBlocks({ cid });
|
||||
|
||||
// There can be only one IPLDBlock with a particular cid.
|
||||
assert(ipldBlocks.length <= 1);
|
||||
|
||||
return ipldBlocks[0];
|
||||
}
|
||||
|
||||
async getLatestIPLDBlock (contractAddress: string, kind: string | null, blockNumber?: number): Promise<IPLDBlock | undefined> {
|
||||
return this._db.getLatestIPLDBlock(contractAddress, kind, blockNumber);
|
||||
return this._baseIndexer.processCLICheckpoint(this, contractAddress, blockHash);
|
||||
}
|
||||
|
||||
async getPrevIPLDBlock (blockHash: string, contractAddress: string, kind?: string): Promise<IPLDBlock | undefined> {
|
||||
const dbTx = await this._db.createTransactionRunner();
|
||||
let res;
|
||||
|
||||
try {
|
||||
res = await this._db.getPrevIPLDBlock(dbTx, blockHash, contractAddress, kind);
|
||||
await dbTx.commitTransaction();
|
||||
} catch (error) {
|
||||
await dbTx.rollbackTransaction();
|
||||
throw error;
|
||||
} finally {
|
||||
await dbTx.release();
|
||||
}
|
||||
return res;
|
||||
return this._db.getPrevIPLDBlock(blockHash, contractAddress, kind);
|
||||
}
|
||||
|
||||
async getDiffIPLDBlocksByCheckpoint (contractAddress: string, checkpointBlockNumber: number): Promise<IPLDBlock[]> {
|
||||
return this._db.getDiffIPLDBlocksByCheckpoint(contractAddress, checkpointBlockNumber);
|
||||
async getLatestIPLDBlock (contractAddress: string, kind: StateKind | null, blockNumber?: number): Promise<IPLDBlock | undefined> {
|
||||
return this._db.getLatestIPLDBlock(contractAddress, kind, blockNumber);
|
||||
}
|
||||
|
||||
async prepareIPLDBlock (block: BlockProgress, contractAddress: string, data: any, kind: string):Promise<any> {
|
||||
assert(_.includes(['diff', 'checkpoint', 'diff_staged'], kind));
|
||||
|
||||
// Get an existing 'diff' | 'diff_staged' | 'checkpoint' IPLDBlock for current block, contractAddress.
|
||||
const currentIPLDBlocks = await this._db.getIPLDBlocks({ block, contractAddress, kind });
|
||||
|
||||
// There can be at most one IPLDBlock for a (block, contractAddress, kind) combination.
|
||||
assert(currentIPLDBlocks.length <= 1);
|
||||
const currentIPLDBlock = currentIPLDBlocks[0];
|
||||
|
||||
// Update currentIPLDBlock if it exists and is of same kind.
|
||||
let ipldBlock;
|
||||
if (currentIPLDBlock) {
|
||||
ipldBlock = currentIPLDBlock;
|
||||
|
||||
// Update the data field.
|
||||
const oldData = codec.decode(Buffer.from(currentIPLDBlock.data));
|
||||
data = _.merge(oldData, data);
|
||||
} else {
|
||||
ipldBlock = new IPLDBlock();
|
||||
|
||||
// Fetch the parent IPLDBlock.
|
||||
const parentIPLDBlock = await this.getLatestIPLDBlock(contractAddress, null, block.blockNumber);
|
||||
|
||||
// Setting the meta-data for an IPLDBlock (done only once per block).
|
||||
data.meta = {
|
||||
id: contractAddress,
|
||||
kind,
|
||||
parent: {
|
||||
'/': parentIPLDBlock ? parentIPLDBlock.cid : null
|
||||
},
|
||||
ethBlock: {
|
||||
cid: {
|
||||
'/': block.cid
|
||||
},
|
||||
num: block.blockNumber
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
// Encoding the data using dag-cbor codec.
|
||||
const bytes = codec.encode(data);
|
||||
|
||||
// Calculating sha256 (multi)hash of the encoded data.
|
||||
const hash = await sha256.digest(bytes);
|
||||
|
||||
// Calculating the CID: v1, code: dag-cbor, hash.
|
||||
const cid = CID.create(1, codec.code, hash);
|
||||
|
||||
// Update ipldBlock with new data.
|
||||
ipldBlock = Object.assign(ipldBlock, {
|
||||
block,
|
||||
contractAddress,
|
||||
cid: cid.toString(),
|
||||
kind: data.meta.kind,
|
||||
data: Buffer.from(bytes)
|
||||
});
|
||||
|
||||
return ipldBlock;
|
||||
async getIPLDBlocksByHash (blockHash: string): Promise<IPLDBlock[]> {
|
||||
return this._baseIndexer.getIPLDBlocksByHash(blockHash);
|
||||
}
|
||||
|
||||
async saveOrUpdateIPLDBlock (ipldBlock: IPLDBlock): Promise<IPLDBlock> {
|
||||
return this._db.saveOrUpdateIPLDBlock(ipldBlock);
|
||||
async getIPLDBlockByCid (cid: string): Promise<IPLDBlock | undefined> {
|
||||
return this._baseIndexer.getIPLDBlockByCid(cid);
|
||||
}
|
||||
|
||||
async removeStagedIPLDBlocks (blockNumber: number): Promise<void> {
|
||||
const dbTx = await this._db.createTransactionRunner();
|
||||
|
||||
try {
|
||||
await this._db.removeEntities(dbTx, IPLDBlock, { relations: ['block'], where: { block: { blockNumber }, kind: 'diff_staged' } });
|
||||
await dbTx.commitTransaction();
|
||||
} catch (error) {
|
||||
await dbTx.rollbackTransaction();
|
||||
throw error;
|
||||
} finally {
|
||||
await dbTx.release();
|
||||
}
|
||||
}
|
||||
|
||||
async pushToIPFS (data: any): Promise<void> {
|
||||
await this._ipfsClient.push(data);
|
||||
getIPLDData (ipldBlock: IPLDBlock): any {
|
||||
return this._baseIndexer.getIPLDData(ipldBlock);
|
||||
}
|
||||
|
||||
isIPFSConfigured (): boolean {
|
||||
const ipfsAddr = this._serverConfig.ipfsApiAddr;
|
||||
|
||||
// Return false if ipfsAddr is undefined | null | empty string.
|
||||
return (ipfsAddr !== undefined && ipfsAddr !== null && ipfsAddr !== '');
|
||||
return this._baseIndexer.isIPFSConfigured();
|
||||
}
|
||||
|
||||
async getSubgraphEntity<Entity> (entity: new () => Entity, id: string, blockHash: string): Promise<Entity | undefined> {
|
||||
return this._graphWatcher.getEntity(entity, id, blockHash);
|
||||
// Method used to create auto diffs (diff_staged).
|
||||
async createDiffStaged (contractAddress: string, blockHash: string, data: any): Promise<void> {
|
||||
await this._baseIndexer.createDiffStaged(contractAddress, blockHash, data);
|
||||
}
|
||||
|
||||
// Method to be used by createStateDiff hook.
|
||||
async createDiff (contractAddress: string, blockHash: string, data: any): Promise<void> {
|
||||
const block = await this.getBlockProgress(blockHash);
|
||||
assert(block);
|
||||
|
||||
await this._baseIndexer.createDiff(contractAddress, block, data);
|
||||
}
|
||||
|
||||
// Method to be used by createStateCheckpoint hook.
|
||||
async createStateCheckpoint (contractAddress: string, blockHash: string, data: any): Promise<void> {
|
||||
const block = await this.getBlockProgress(blockHash);
|
||||
assert(block);
|
||||
|
||||
return this._baseIndexer.createStateCheckpoint(contractAddress, block, data);
|
||||
}
|
||||
|
||||
// Method to be used by checkpoint CLI.
|
||||
async createCheckpoint (contractAddress: string, blockHash: string): Promise<string | undefined> {
|
||||
const block = await this.getBlockProgress(blockHash);
|
||||
assert(block);
|
||||
|
||||
return this._baseIndexer.createCheckpoint(this, contractAddress, block);
|
||||
}
|
||||
|
||||
async saveOrUpdateIPLDBlock (ipldBlock: IPLDBlock): Promise<IPLDBlock> {
|
||||
return this._baseIndexer.saveOrUpdateIPLDBlock(ipldBlock);
|
||||
}
|
||||
|
||||
async removeIPLDBlocks (blockNumber: number, kind: StateKind): Promise<void> {
|
||||
await this._baseIndexer.removeIPLDBlocks(blockNumber, kind);
|
||||
}
|
||||
|
||||
async getSubgraphEntity<Entity> (entity: new () => Entity, id: string, block?: BlockHeight): Promise<any> {
|
||||
const relations = this._relationsMap.get(entity) || {};
|
||||
|
||||
const data = await this._graphWatcher.getEntity(entity, id, relations, block);
|
||||
|
||||
return data;
|
||||
}
|
||||
|
||||
async triggerIndexingOnEvent (event: Event): Promise<void> {
|
||||
@ -584,6 +393,14 @@ export class Indexer implements IndexerInterface {
|
||||
await this.triggerIndexingOnEvent(event);
|
||||
}
|
||||
|
||||
async processBlock (blockHash: string, blockNumber: number): Promise<void> {
|
||||
// Call a function to create initial state for contracts.
|
||||
await this._baseIndexer.createInit(this, blockHash, blockNumber);
|
||||
|
||||
// Call subgraph handler for block.
|
||||
await this._graphWatcher.handleBlock(blockHash);
|
||||
}
|
||||
|
||||
parseEventNameAndArgs (kind: string, logObj: any): any {
|
||||
let eventName = UNKNOWN_EVENT_NAME;
|
||||
let eventInfo = {};
|
||||
@ -595,7 +412,9 @@ export class Indexer implements IndexerInterface {
|
||||
{{#each events as | event |}}
|
||||
case {{capitalize event.name}}_EVENT: {
|
||||
eventName = logDescription.name;
|
||||
{{#if event.params}}
|
||||
const { {{#each event.params~}} {{this.name}} {{~#unless @last}}, {{/unless}} {{~/each}} } = logDescription.args;
|
||||
{{/if}}
|
||||
eventInfo = {
|
||||
{{#each event.params}}
|
||||
{{#if (compare this.type 'bigint')}}
|
||||
@ -619,31 +438,8 @@ export class Indexer implements IndexerInterface {
|
||||
};
|
||||
}
|
||||
|
||||
async watchContract (address: string, kind: string, checkpoint: boolean, startingBlock: number): Promise<void> {
|
||||
this._baseIndexer.updateIPLDStatusMap(address, {});
|
||||
|
||||
return this._baseIndexer.watchContract(address, kind, checkpoint, startingBlock);
|
||||
}
|
||||
|
||||
cacheContract (contract: Contract): void {
|
||||
return this._baseIndexer.cacheContract(contract);
|
||||
}
|
||||
|
||||
async getHookStatus (): Promise<HookStatus | undefined> {
|
||||
const dbTx = await this._db.createTransactionRunner();
|
||||
let res;
|
||||
|
||||
try {
|
||||
res = await this._db.getHookStatus(dbTx);
|
||||
await dbTx.commitTransaction();
|
||||
} catch (error) {
|
||||
await dbTx.rollbackTransaction();
|
||||
throw error;
|
||||
} finally {
|
||||
await dbTx.release();
|
||||
}
|
||||
|
||||
return res;
|
||||
return this._db.getHookStatus();
|
||||
}
|
||||
|
||||
async updateHookStatusProcessedBlock (blockNumber: number, force?: boolean): Promise<HookStatus> {
|
||||
@ -663,11 +459,32 @@ export class Indexer implements IndexerInterface {
|
||||
return res;
|
||||
}
|
||||
|
||||
async getLatestCanonicalBlock (): Promise<BlockProgress | undefined> {
|
||||
async getLatestCanonicalBlock (): Promise<BlockProgress> {
|
||||
const syncStatus = await this.getSyncStatus();
|
||||
assert(syncStatus);
|
||||
|
||||
return this.getBlockProgress(syncStatus.latestCanonicalBlockHash);
|
||||
const latestCanonicalBlock = await this.getBlockProgress(syncStatus.latestCanonicalBlockHash);
|
||||
assert(latestCanonicalBlock);
|
||||
|
||||
return latestCanonicalBlock;
|
||||
}
|
||||
|
||||
async getLatestHooksProcessedBlock (): Promise<BlockProgress> {
|
||||
return this._baseIndexer.getLatestHooksProcessedBlock();
|
||||
}
|
||||
|
||||
async watchContract (address: string, kind: string, checkpoint: boolean, startingBlock: number): Promise<void> {
|
||||
this._baseIndexer.updateIPLDStatusMap(address, {});
|
||||
|
||||
return this._baseIndexer.watchContract(address, kind, checkpoint, startingBlock);
|
||||
}
|
||||
|
||||
cacheContract (contract: Contract): void {
|
||||
return this._baseIndexer.cacheContract(contract);
|
||||
}
|
||||
|
||||
async saveEventEntity (dbEvent: Event): Promise<Event> {
|
||||
return this._baseIndexer.saveEventEntity(dbEvent);
|
||||
}
|
||||
|
||||
async getEventsByFilter (blockHash: string, contract?: string, name?: string): Promise<Array<Event>> {
|
||||
@ -690,6 +507,10 @@ export class Indexer implements IndexerInterface {
|
||||
return this._baseIndexer.getSyncStatus();
|
||||
}
|
||||
|
||||
async getBlocks (blockFilter: { blockHash?: string, blockNumber?: number }): Promise<any> {
|
||||
return this._baseIndexer.getBlocks(blockFilter);
|
||||
}
|
||||
|
||||
async updateSyncStatusIndexedBlock (blockHash: string, blockNumber: number, force = false): Promise<SyncStatus> {
|
||||
return this._baseIndexer.updateSyncStatusIndexedBlock(blockHash, blockNumber, force);
|
||||
}
|
||||
@ -702,10 +523,6 @@ export class Indexer implements IndexerInterface {
|
||||
return this._baseIndexer.updateSyncStatusCanonicalBlock(blockHash, blockNumber, force);
|
||||
}
|
||||
|
||||
async getBlocks (blockFilter: { blockNumber?: number, blockHash?: string }): Promise<any> {
|
||||
return this._baseIndexer.getBlocks(blockFilter);
|
||||
}
|
||||
|
||||
async getEvent (id: string): Promise<Event | undefined> {
|
||||
return this._baseIndexer.getEvent(id);
|
||||
}
|
||||
|
||||
@ -1,17 +0,0 @@
|
||||
//
|
||||
// Copyright 2021 Vulcanize, Inc.
|
||||
//
|
||||
|
||||
import { create, IPFSHTTPClient } from 'ipfs-http-client';
|
||||
|
||||
export class IPFSClient {
|
||||
_client: IPFSHTTPClient
|
||||
|
||||
constructor (url: string) {
|
||||
this._client = create({ url });
|
||||
}
|
||||
|
||||
async push (data: any): Promise<void> {
|
||||
await this._client.dag.put(data, { format: 'dag-cbor', hashAlg: 'sha2-256' });
|
||||
}
|
||||
}
|
||||
@ -71,11 +71,19 @@ export class JobRunner {
|
||||
|
||||
const hookStatus = await this._indexer.getHookStatus();
|
||||
|
||||
if (hookStatus && hookStatus.latestProcessedBlockNumber < (blockNumber - 1)) {
|
||||
const message = `Hooks for blockNumber ${blockNumber - 1} not processed yet, aborting`;
|
||||
log(message);
|
||||
if (hookStatus) {
|
||||
if (hookStatus.latestProcessedBlockNumber < (blockNumber - 1)) {
|
||||
const message = `Hooks for blockNumber ${blockNumber - 1} not processed yet, aborting`;
|
||||
log(message);
|
||||
|
||||
throw new Error(message);
|
||||
throw new Error(message);
|
||||
}
|
||||
|
||||
if (hookStatus.latestProcessedBlockNumber > (blockNumber - 1)) {
|
||||
log(`Hooks for blockNumber ${blockNumber} already processed`);
|
||||
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
await this._indexer.processCanonicalBlock(job);
|
||||
@ -123,7 +131,7 @@ export const main = async (): Promise<any> => {
|
||||
const graphDb = new GraphDatabase(config.database, path.resolve(__dirname, 'entity/*'));
|
||||
await graphDb.init();
|
||||
|
||||
const graphWatcher = new GraphWatcher(graphDb, postgraphileClient, ethProvider, config.server.subgraphPath);
|
||||
const graphWatcher = new GraphWatcher(graphDb, postgraphileClient, ethProvider, config.server);
|
||||
|
||||
const jobQueueConfig = config.jobQueue;
|
||||
assert(jobQueueConfig, 'Missing job queue config');
|
||||
|
||||
@ -6,7 +6,7 @@ import assert from 'assert';
|
||||
import BigInt from 'apollo-type-bigint';
|
||||
import debug from 'debug';
|
||||
|
||||
import { ValueResult } from '@vulcanize/util';
|
||||
import { ValueResult, BlockHeight, StateKind } from '@vulcanize/util';
|
||||
|
||||
import { Indexer } from './indexer';
|
||||
import { EventWatcher } from './events';
|
||||
@ -38,10 +38,11 @@ export const createResolvers = async (indexer: Indexer, eventWatcher: EventWatch
|
||||
},
|
||||
|
||||
Mutation: {
|
||||
watchContract: (_: any, { address, kind, checkpoint, startingBlock }: { address: string, kind: string, checkpoint: boolean, startingBlock: number }): Promise<boolean> => {
|
||||
watchContract: async (_: any, { address, kind, checkpoint, startingBlock = 1 }: { address: string, kind: string, checkpoint: boolean, startingBlock: number }): Promise<boolean> => {
|
||||
log('watchContract', address, kind, checkpoint, startingBlock);
|
||||
await indexer.watchContract(address, kind, checkpoint, startingBlock);
|
||||
|
||||
return indexer.watchContract(address, kind, checkpoint, startingBlock);
|
||||
return true;
|
||||
}
|
||||
},
|
||||
|
||||
@ -59,10 +60,10 @@ export const createResolvers = async (indexer: Indexer, eventWatcher: EventWatch
|
||||
{{/each}}
|
||||
|
||||
{{~#each subgraphQueries}}
|
||||
{{this.queryName}}: async (_: any, { id, blockHash }: { id: string, blockHash: string }): Promise<{{this.entityName}} | undefined> => {
|
||||
log('{{this.queryName}}', id, blockHash);
|
||||
{{this.queryName}}: async (_: any, { id, block = {} }: { id: string, block: BlockHeight }): Promise<{{this.entityName}} | undefined> => {
|
||||
log('{{this.queryName}}', id, block);
|
||||
|
||||
return indexer.getSubgraphEntity({{this.entityName}}, id, blockHash);
|
||||
return indexer.getSubgraphEntity({{this.entityName}}, id, block);
|
||||
},
|
||||
|
||||
{{/each}}
|
||||
@ -98,7 +99,7 @@ export const createResolvers = async (indexer: Indexer, eventWatcher: EventWatch
|
||||
return ipldBlock && ipldBlock.block.isComplete ? indexer.getResultIPLDBlock(ipldBlock) : undefined;
|
||||
},
|
||||
|
||||
getState: async (_: any, { blockHash, contractAddress, kind = 'diff' }: { blockHash: string, contractAddress: string, kind: string }) => {
|
||||
getState: async (_: any, { blockHash, contractAddress, kind = StateKind.Diff }: { blockHash: string, contractAddress: string, kind: string }) => {
|
||||
log('getState', blockHash, contractAddress, kind);
|
||||
|
||||
const ipldBlock = await indexer.getPrevIPLDBlock(blockHash, contractAddress, kind);
|
||||
|
||||
@ -46,7 +46,7 @@ export const main = async (): Promise<any> => {
|
||||
const graphDb = new GraphDatabase(config.database, path.resolve(__dirname, 'entity/*'));
|
||||
await graphDb.init();
|
||||
|
||||
const graphWatcher = new GraphWatcher(graphDb, postgraphileClient, ethProvider, config.server.subgraphPath);
|
||||
const graphWatcher = new GraphWatcher(graphDb, postgraphileClient, ethProvider, config.server);
|
||||
|
||||
// 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
|
||||
|
||||
@ -45,8 +45,16 @@ export class Visitor {
|
||||
return { name: item.name, type: item.typeName.name };
|
||||
});
|
||||
|
||||
const typeName = node.returnParameters[0].typeName;
|
||||
|
||||
// TODO Handle user defined type return.
|
||||
if (typeName.type === 'UserDefinedTypeName') {
|
||||
// Skip in case of UserDefinedTypeName.
|
||||
return;
|
||||
}
|
||||
|
||||
// TODO Handle multiple return parameters and array return type.
|
||||
const returnType = node.returnParameters[0].typeName.name;
|
||||
const returnType = typeName.name;
|
||||
|
||||
this._schema.addQuery(name, params, returnType);
|
||||
this._resolvers.addQuery(name, params, returnType);
|
||||
@ -72,6 +80,13 @@ export class Visitor {
|
||||
const params: Param[] = [];
|
||||
|
||||
let typeName = variable.typeName;
|
||||
|
||||
// TODO Handle user defined type.
|
||||
if (typeName.type === 'UserDefinedTypeName') {
|
||||
// Skip in case of UserDefinedTypeName.
|
||||
return;
|
||||
}
|
||||
|
||||
let numParams = 0;
|
||||
|
||||
// If the variable type is mapping, extract key as a param:
|
||||
|
||||
Loading…
Reference in New Issue
Block a user