// // Copyright 2021 Vulcanize, Inc. // import assert from 'assert'; import { DeepPartial, FindConditions, FindManyOptions, ObjectLiteral } from 'typeorm'; import debug from 'debug'; import { ethers, constants } from 'ethers'; import { GraphQLResolveInfo } from 'graphql'; import { JsonFragment } from '@ethersproject/abi'; import { BaseProvider } from '@ethersproject/providers'; import { MappingKey, StorageLayout } from '@cerc-io/solidity-mapper'; import { Indexer as BaseIndexer, IndexerInterface, ValueResult, ServerConfig, JobQueue, Where, QueryOptions, BlockHeight, ResultMeta, updateSubgraphState, dumpSubgraphState, GraphWatcherInterface, StateKind, StateStatus, ResultEvent, getResultEvent, DatabaseInterface, Clients, EthClient, UpstreamConfig, EthFullBlock, EthFullTransaction, ExtraEventData } from '@cerc-io/util'; import { GraphWatcher } from '@cerc-io/graph-node'; import UniswapV2FactoryArtifacts from './artifacts/UniswapV2Factory.json'; import { Database, ENTITIES, SUBGRAPH_ENTITIES } from './database'; import { createInitialState, handleEvent, createStateDiff, createStateCheckpoint } from './hooks'; import { Contract } from './entity/Contract'; import { Event } from './entity/Event'; import { SyncStatus } from './entity/SyncStatus'; import { StateSyncStatus } from './entity/StateSyncStatus'; import { BlockProgress } from './entity/BlockProgress'; import { State } from './entity/State'; /* eslint-disable @typescript-eslint/no-unused-vars */ import { Block } from './entity/Block'; /* eslint-enable @typescript-eslint/no-unused-vars */ import { FrothyEntity } from './entity/FrothyEntity'; // eslint-disable-next-line @typescript-eslint/no-unused-vars const log = debug('vulcanize:indexer'); const KIND_UNISWAPV2FACTORY = 'UniswapV2Factory'; export class Indexer implements IndexerInterface { _db: Database; _ethClient: EthClient; _ethProvider: BaseProvider; _baseIndexer: BaseIndexer; _serverConfig: ServerConfig; _upstreamConfig: UpstreamConfig; _graphWatcher: GraphWatcher; _abiMap: Map; _storageLayoutMap: Map; _contractMap: Map; eventSignaturesMap: Map; _entityTypesMap: Map; _relationsMap: Map; _subgraphStateMap: Map; constructor ( config: { server: ServerConfig; upstream: UpstreamConfig; }, db: DatabaseInterface, clients: Clients, ethProvider: BaseProvider, jobQueue: JobQueue, graphWatcher?: GraphWatcherInterface ) { assert(db); assert(clients.ethClient); this._db = db as Database; this._ethClient = clients.ethClient; this._ethProvider = ethProvider; this._serverConfig = config.server; this._upstreamConfig = config.upstream; this._baseIndexer = new BaseIndexer(config, this._db, this._ethClient, this._ethProvider, jobQueue); assert(graphWatcher); this._graphWatcher = graphWatcher as GraphWatcher; this._abiMap = new Map(); this._storageLayoutMap = new Map(); this._contractMap = new Map(); this.eventSignaturesMap = new Map(); const { abi: UniswapV2FactoryABI } = UniswapV2FactoryArtifacts; assert(UniswapV2FactoryABI); this._abiMap.set(KIND_UNISWAPV2FACTORY, UniswapV2FactoryABI); const UniswapV2FactoryContractInterface = new ethers.utils.Interface(UniswapV2FactoryABI); this._contractMap.set(KIND_UNISWAPV2FACTORY, UniswapV2FactoryContractInterface); const UniswapV2FactoryEventSignatures = Object.values(UniswapV2FactoryContractInterface.events).map(value => { return UniswapV2FactoryContractInterface.getEventTopic(value); }); this.eventSignaturesMap.set(KIND_UNISWAPV2FACTORY, UniswapV2FactoryEventSignatures); this._entityTypesMap = new Map(); this._populateEntityTypesMap(); this._relationsMap = new Map(); this._populateRelationsMap(); this._subgraphStateMap = new Map(); } get serverConfig (): ServerConfig { return this._serverConfig; } get upstreamConfig (): UpstreamConfig { return this._upstreamConfig; } get storageLayoutMap (): Map { return this._storageLayoutMap; } get graphWatcher (): GraphWatcher { return this._graphWatcher; } async init (): Promise { await this._baseIndexer.fetchContracts(); await this._baseIndexer.fetchStateStatus(); } switchClients ({ ethClient, ethProvider }: { ethClient: EthClient, ethProvider: BaseProvider }): void { this._ethClient = ethClient; this._ethProvider = ethProvider; this._baseIndexer.switchClients({ ethClient, ethProvider }); this._graphWatcher.switchClients({ ethClient, ethProvider }); } async getMetaData (block: BlockHeight): Promise { return this._baseIndexer.getMetaData(block); } getResultEvent (event: Event): ResultEvent { return getResultEvent(event); } async getStorageValue (storageLayout: StorageLayout, blockHash: string, contractAddress: string, variable: string, ...mappingKeys: MappingKey[]): Promise { return this._baseIndexer.getStorageValue( storageLayout, blockHash, contractAddress, variable, ...mappingKeys ); } async getEntitiesForBlock (blockHash: string, tableName: string): Promise { return this._db.getEntitiesForBlock(blockHash, tableName); } async processInitialState (contractAddress: string, blockHash: string): Promise { // Call initial state hook. return createInitialState(this, contractAddress, blockHash); } async processStateCheckpoint (contractAddress: string, blockHash: string): Promise { // Call checkpoint hook. return createStateCheckpoint(this, contractAddress, blockHash); } async processCanonicalBlock (blockHash: string, blockNumber: number): Promise { console.time('time:indexer#processCanonicalBlock-finalize_auto_diffs'); // Finalize staged diff blocks if any. await this._baseIndexer.finalizeDiffStaged(blockHash); console.timeEnd('time:indexer#processCanonicalBlock-finalize_auto_diffs'); // Call custom stateDiff hook. await createStateDiff(this, blockHash); this._graphWatcher.pruneEntityCacheFrothyBlocks(blockHash, blockNumber); } async processCheckpoint (blockHash: string): Promise { // Return if checkpointInterval is <= 0. const checkpointInterval = this._serverConfig.checkpointInterval; if (checkpointInterval <= 0) return; console.time('time:indexer#processCheckpoint-checkpoint'); await this._baseIndexer.processCheckpoint(this, blockHash, checkpointInterval); console.timeEnd('time:indexer#processCheckpoint-checkpoint'); } async processCLICheckpoint (contractAddress: string, blockHash?: string): Promise { return this._baseIndexer.processCLICheckpoint(this, contractAddress, blockHash); } async getPrevState (blockHash: string, contractAddress: string, kind?: string): Promise { return this._db.getPrevState(blockHash, contractAddress, kind); } async getLatestState (contractAddress: string, kind: StateKind | null, blockNumber?: number): Promise { return this._db.getLatestState(contractAddress, kind, blockNumber); } async getStatesByHash (blockHash: string): Promise { return this._baseIndexer.getStatesByHash(blockHash); } async getStateByCID (cid: string): Promise { return this._baseIndexer.getStateByCID(cid); } async getStates (where: FindConditions): Promise { return this._db.getStates(where); } getStateData (state: State): any { return this._baseIndexer.getStateData(state); } // Method used to create auto diffs (diff_staged). async createDiffStaged (contractAddress: string, blockHash: string, data: any): Promise { console.time('time:indexer#createDiffStaged-auto_diff'); await this._baseIndexer.createDiffStaged(contractAddress, blockHash, data); console.timeEnd('time:indexer#createDiffStaged-auto_diff'); } // Method to be used by createStateDiff hook. async createDiff (contractAddress: string, blockHash: string, data: any): Promise { 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 { const block = await this.getBlockProgress(blockHash); assert(block); return this._baseIndexer.createStateCheckpoint(contractAddress, block, data); } // Method to be used by export-state CLI. async createCheckpoint (contractAddress: string, blockHash: string): Promise { const block = await this.getBlockProgress(blockHash); assert(block); return this._baseIndexer.createCheckpoint(this, contractAddress, block); } // Method to be used by fill-state CLI. async createInit (blockHash: string, blockNumber: number): Promise { // Create initial state for contracts. await this._baseIndexer.createInit(this, blockHash, blockNumber); } async saveOrUpdateState (state: State): Promise { return this._baseIndexer.saveOrUpdateState(state); } async removeStates (blockNumber: number, kind: StateKind): Promise { await this._baseIndexer.removeStates(blockNumber, kind); } async getSubgraphEntity ( entity: new () => Entity, id: string, block: BlockHeight, queryInfo: GraphQLResolveInfo ): Promise { const data = await this._graphWatcher.getEntity(entity, id, this._relationsMap, block, queryInfo); return data; } async getSubgraphEntities ( entity: new () => Entity, block: BlockHeight, where: { [key: string]: any } = {}, queryOptions: QueryOptions = {}, queryInfo: GraphQLResolveInfo ): Promise { return this._graphWatcher.getEntities(entity, this._relationsMap, block, where, queryOptions, queryInfo); } // eslint-disable-next-line @typescript-eslint/no-unused-vars async triggerIndexingOnEvent (event: Event, extraData: ExtraEventData): Promise { const resultEvent = this.getResultEvent(event); console.time('time:indexer#processEvent-mapping_code'); // Call subgraph handler for event. await this._graphWatcher.handleEvent(resultEvent, extraData); console.timeEnd('time:indexer#processEvent-mapping_code'); // Call custom hook function for indexing on event. await handleEvent(this, resultEvent); } async processEvent (event: Event, extraData: ExtraEventData): Promise { // Trigger indexing of data based on the event. await this.triggerIndexingOnEvent(event, extraData); } async processBlock (blockProgress: BlockProgress): Promise { console.time('time:indexer#processBlock-init_state'); // Call a function to create initial state for contracts. await this._baseIndexer.createInit(this, blockProgress.blockHash, blockProgress.blockNumber); console.timeEnd('time:indexer#processBlock-init_state'); this._graphWatcher.updateEntityCacheFrothyBlocks(blockProgress); } async processBlockAfterEvents (blockHash: string, blockNumber: number, extraData: ExtraEventData): Promise { console.time('time:indexer#processBlockAfterEvents-mapping_code'); // Call subgraph handler for block. await this._graphWatcher.handleBlock(blockHash, blockNumber, extraData); console.timeEnd('time:indexer#processBlockAfterEvents-mapping_code'); console.time('time:indexer#processBlockAfterEvents-dump_subgraph_state'); // Persist subgraph state to the DB. await this.dumpSubgraphState(blockHash); console.timeEnd('time:indexer#processBlockAfterEvents-dump_subgraph_state'); } parseEventNameAndArgs (kind: string, logObj: any): { eventParsed: boolean, eventDetails: any } { const { topics, data } = logObj; const contract = this._contractMap.get(kind); assert(contract); let logDescription: ethers.utils.LogDescription; try { logDescription = contract.parseLog({ data, topics }); } catch (err) { // Return if no matching event found if ((err as Error).message.includes('no matching event')) { log(`WARNING: Skipping event for contract ${kind} as no matching event found in the ABI`); return { eventParsed: false, eventDetails: {} }; } throw err; } const { eventName, eventInfo, eventSignature } = this._baseIndexer.parseEvent(logDescription); return { eventParsed: true, eventDetails: { eventName, eventInfo, eventSignature } }; } async getStateSyncStatus (): Promise { return this._db.getStateSyncStatus(); } async updateStateSyncStatusIndexedBlock (blockNumber: number, force?: boolean): Promise { if (!this._serverConfig.enableState) { return; } const dbTx = await this._db.createTransactionRunner(); let res; try { res = await this._db.updateStateSyncStatusIndexedBlock(dbTx, blockNumber, force); await dbTx.commitTransaction(); } catch (error) { await dbTx.rollbackTransaction(); throw error; } finally { await dbTx.release(); } return res; } async updateStateSyncStatusCheckpointBlock (blockNumber: number, force?: boolean): Promise { const dbTx = await this._db.createTransactionRunner(); let res; try { res = await this._db.updateStateSyncStatusCheckpointBlock(dbTx, blockNumber, force); await dbTx.commitTransaction(); } catch (error) { await dbTx.rollbackTransaction(); throw error; } finally { await dbTx.release(); } return res; } async getLatestCanonicalBlock (): Promise { const syncStatus = await this.getSyncStatus(); assert(syncStatus); if (syncStatus.latestCanonicalBlockHash === constants.HashZero) { return; } const latestCanonicalBlock = await this.getBlockProgress(syncStatus.latestCanonicalBlockHash); assert(latestCanonicalBlock); return latestCanonicalBlock; } async getLatestStateIndexedBlock (): Promise { return this._baseIndexer.getLatestStateIndexedBlock(); } async addContracts (): Promise { // Watching all the contracts in the subgraph. await this._graphWatcher.addContracts(); } async watchContract (address: string, kind: string, checkpoint: boolean, startingBlock: number, context?: any): Promise { return this._baseIndexer.watchContract(address, kind, checkpoint, startingBlock, context); } updateStateStatusMap (address: string, stateStatus: StateStatus): void { this._baseIndexer.updateStateStatusMap(address, stateStatus); } cacheContract (contract: Contract): void { return this._baseIndexer.cacheContract(contract); } async saveEventEntity (dbEvent: Event): Promise { return this._baseIndexer.saveEventEntity(dbEvent); } async saveEvents (dbEvents: Event[]): Promise { return this._baseIndexer.saveEvents(dbEvents); } async getEventsByFilter (blockHash: string, contract?: string, name?: string): Promise> { return this._baseIndexer.getEventsByFilter(blockHash, contract, name); } isWatchedContract (address : string): Contract | undefined { return this._baseIndexer.isWatchedContract(address); } getWatchedContracts (): Contract[] { return this._baseIndexer.getWatchedContracts(); } getContractsByKind (kind: string): Contract[] { return this._baseIndexer.getContractsByKind(kind); } async getProcessedBlockCountForRange (fromBlockNumber: number, toBlockNumber: number): Promise<{ expected: number, actual: number }> { return this._baseIndexer.getProcessedBlockCountForRange(fromBlockNumber, toBlockNumber); } async getEventsInRange (fromBlockNumber: number, toBlockNumber: number): Promise> { return this._baseIndexer.getEventsInRange(fromBlockNumber, toBlockNumber, this._serverConfig.gql.maxEventsBlockRange); } async getSyncStatus (): Promise { return this._baseIndexer.getSyncStatus(); } async getBlocks (blockFilter: { blockHash?: string, blockNumber?: number }): Promise { return this._baseIndexer.getBlocks(blockFilter); } async updateSyncStatusIndexedBlock (blockHash: string, blockNumber: number, force = false): Promise { return this._baseIndexer.updateSyncStatusIndexedBlock(blockHash, blockNumber, force); } async updateSyncStatusChainHead (blockHash: string, blockNumber: number, force = false): Promise { return this._baseIndexer.updateSyncStatusChainHead(blockHash, blockNumber, force); } async updateSyncStatusCanonicalBlock (blockHash: string, blockNumber: number, force = false): Promise { const syncStatus = this._baseIndexer.updateSyncStatusCanonicalBlock(blockHash, blockNumber, force); await this.pruneFrothyEntities(blockNumber); return syncStatus; } async updateSyncStatusProcessedBlock (blockHash: string, blockNumber: number, force = false): Promise { return this._baseIndexer.updateSyncStatusProcessedBlock(blockHash, blockNumber, force); } async updateSyncStatusIndexingError (hasIndexingError: boolean): Promise { return this._baseIndexer.updateSyncStatusIndexingError(hasIndexingError); } async updateSyncStatus (syncStatus: DeepPartial): Promise { return this._baseIndexer.updateSyncStatus(syncStatus); } async getEvent (id: string): Promise { return this._baseIndexer.getEvent(id); } async getBlockProgress (blockHash: string): Promise { return this._baseIndexer.getBlockProgress(blockHash); } async getBlockProgressEntities (where: FindConditions, options: FindManyOptions): Promise { return this._baseIndexer.getBlockProgressEntities(where, options); } async getBlocksAtHeight (height: number, isPruned: boolean): Promise { return this._baseIndexer.getBlocksAtHeight(height, isPruned); } async fetchAndSaveFilteredEventsAndBlocks (startBlock: number, endBlock: number): Promise<{ blockProgress: BlockProgress, events: DeepPartial[], ethFullBlock: EthFullBlock; ethFullTransactions: EthFullTransaction[]; }[]> { return this._baseIndexer.fetchAndSaveFilteredEventsAndBlocks(startBlock, endBlock, this.eventSignaturesMap, this.parseEventNameAndArgs.bind(this)); } async fetchEventsForContracts (blockHash: string, blockNumber: number, addresses: string[]): Promise[]> { return this._baseIndexer.fetchEventsForContracts(blockHash, blockNumber, addresses, this.eventSignaturesMap, this.parseEventNameAndArgs.bind(this)); } async saveBlockAndFetchEvents (block: DeepPartial): Promise<[ BlockProgress, DeepPartial[], EthFullTransaction[] ]> { return this._saveBlockAndFetchEvents(block); } async getBlockEvents (blockHash: string, where: Where, queryOptions: QueryOptions): Promise> { return this._baseIndexer.getBlockEvents(blockHash, where, queryOptions); } async removeUnknownEvents (block: BlockProgress): Promise { return this._baseIndexer.removeUnknownEvents(Event, block); } async markBlocksAsPruned (blocks: BlockProgress[]): Promise { await this._baseIndexer.markBlocksAsPruned(blocks); await this._graphWatcher.pruneEntities(FrothyEntity, blocks, SUBGRAPH_ENTITIES); } async pruneFrothyEntities (blockNumber: number): Promise { await this._graphWatcher.pruneFrothyEntities(FrothyEntity, blockNumber); } async resetLatestEntities (blockNumber: number): Promise { await this._graphWatcher.resetLatestEntities(blockNumber); } async updateBlockProgress (block: BlockProgress, lastProcessedEventIndex: number): Promise { return this._baseIndexer.updateBlockProgress(block, lastProcessedEventIndex); } async getAncestorAtHeight (blockHash: string, height: number): Promise { return this._baseIndexer.getAncestorAtHeight(blockHash, height); } async resetWatcherToBlock (blockNumber: number): Promise { const entities = [...ENTITIES, FrothyEntity]; await this._baseIndexer.resetWatcherToBlock(blockNumber, entities); await this.resetLatestEntities(blockNumber); } async clearProcessedBlockData (block: BlockProgress): Promise { const entities = [...ENTITIES, FrothyEntity]; await this._baseIndexer.clearProcessedBlockData(block, entities); await this.resetLatestEntities(block.blockNumber); } getEntityTypesMap (): Map { return this._entityTypesMap; } getRelationsMap (): Map { return this._relationsMap; } updateSubgraphState (contractAddress: string, data: any): void { return updateSubgraphState(this._subgraphStateMap, contractAddress, data); } async dumpSubgraphState (blockHash: string, isStateFinalized = false): Promise { return dumpSubgraphState(this, this._subgraphStateMap, blockHash, isStateFinalized); } _populateEntityTypesMap (): void { this._entityTypesMap.set('Block', { id: 'ID', number: 'BigInt', timestamp: 'BigInt', parentHash: 'String', author: 'String', difficulty: 'BigInt', totalDifficulty: 'BigInt', gasUsed: 'BigInt', gasLimit: 'BigInt', receiptsRoot: 'String', transactionsRoot: 'String', stateRoot: 'String', size: 'BigInt', unclesHash: 'String' }); } // eslint-disable-next-line @typescript-eslint/no-empty-function _populateRelationsMap (): void { } async _saveBlockAndFetchEvents ({ cid: blockCid, blockHash, blockNumber, blockTimestamp, parentHash }: DeepPartial): Promise<[ BlockProgress, DeepPartial[], EthFullTransaction[] ]> { assert(blockHash); assert(blockNumber); const { events: dbEvents, transactions } = await this._baseIndexer.fetchEvents(blockHash, blockNumber, this.eventSignaturesMap, this.parseEventNameAndArgs.bind(this)); const dbTx = await this._db.createTransactionRunner(); try { const block = { cid: blockCid, blockHash, blockNumber, blockTimestamp, parentHash }; console.time(`time:indexer#_saveBlockAndFetchEvents-db-save-${blockNumber}`); const blockProgress = await this._db.saveBlockWithEvents(dbTx, block, dbEvents); await dbTx.commitTransaction(); console.timeEnd(`time:indexer#_saveBlockAndFetchEvents-db-save-${blockNumber}`); return [blockProgress, [], transactions]; } catch (error) { await dbTx.rollbackTransaction(); throw error; } finally { await dbTx.release(); } } async getFullTransactions (txHashList: string[]): Promise { return this._baseIndexer.getFullTransactions(txHashList); } }