import assert from 'assert' ; import path from 'path' ; import { Inject , Injectable } from '@nestjs/common' ; import { EventEmitter2 } from '@nestjs/event-emitter' ; import { ApiPromise } from '@polkadot/api' ; import { hexToU8a , u8aEq } from '@polkadot/util' ; import { buildSchema , getAllEntitiesRelations , isBlockHandlerProcessor , isCallHandlerProcessor , isEventHandlerProcessor , isCustomDs , isRuntimeDs , } from '@subql/common' ; import { RuntimeHandlerInputMap , SecondLayerHandlerProcessor , SubqlCustomDatasource , SubqlCustomHandler , SubqlDatasource , SubqlHandlerKind , SubqlNetworkFilter , SubqlRuntimeHandler , } from '@subql/types' ; import { QueryTypes , Sequelize } from 'sequelize' ; import { NodeConfig } from '../configure/NodeConfig' ; import { SubqueryProject } from '../configure/project.model' ; import { SubqueryRepo } from '../entities' ; import { getLogger } from '../utils/logger' ; import { profiler } from '../utils/profiler' ; import * as SubstrateUtil from '../utils/substrate' ; import { getYargsOption } from '../yargs' ; import { ApiService } from './api.service' ; import { DsProcessorService } from './ds-processor.service' ; import { MetadataFactory , MetadataRepo } from './entities/Metadata.entity' ; import { IndexerEvent } from './events' ; import { FetchService } from './fetch.service' ; import { MmrService } from './mmr.service' ; import { PoiService } from './poi.service' ; import { PoiBlock } from './PoiBlock' ; import { IndexerSandbox , SandboxService } from './sandbox.service' ; import { StoreService } from './store.service' ; import { BlockContent } from './types' ; const { version : packageVersion } = require ( '../../package.json' ) ; const DEFAULT_DB_SCHEMA = 'public' ; const NULL_MERKEL_ROOT = hexToU8a ( '0x00' ) ; const logger = getLogger ( 'indexer' ) ; const { argv } = getYargsOption ( ) ; @ Injectable ( ) export class IndexerManager { private api : ApiPromise ; private prevSpecVersion ? : number ; private filteredDataSources : SubqlDatasource [ ] ; protected metadataRepo : MetadataRepo ; constructor ( private storeService , private apiService , private fetchService , private poiService , protected mmrService , private sequelize , private project , private nodeConfig , private sandboxService , private dsProcessorService , @ Inject ( 'Subquery' ) protected subqueryRepo , private eventEmitter ) { } @ profiler ( argv . profiler ) async indexBlock ( blockContent ) { const { block } = blockContent ; const blockHeight = block . block . header . number . toNumber ( ) ; this . eventEmitter . emit ( IndexerEvent . BlockProcessing , { height : blockHeight , timestamp : Date . now ( ) , } ) ; const tx = await this . sequelize . transaction ( ) ; this . storeService . setTransaction ( tx ) ; let poiBlockHash ; try { const isUpgraded = block . specVersion !== this . prevSpecVersion ; const apiAt = await this . apiService . getPatchedApi ( block . block . hash , isUpgraded ? block . block . header . parentHash : undefined ) ; for ( const ds of this . filteredDataSources ) { const vm = this . sandboxService . getDsProcessor ( ds , apiAt ) ; if ( isRuntimeDs ( ds ) ) { await this . indexBlockForRuntimeDs ( vm , ds . mapping . handlers , blockContent ) ; } else if ( isCustomDs ( ds ) ) { await this . indexBlockForCustomDs ( ds , vm , blockContent ) ; } } await this . storeService . setMetadataBatch ( [ { key : 'lastProcessedHeight' , value : blockHeight } , { key : 'lastProcessedTimestamp' , value : Date . now ( ) } , ] , { transaction : tx } ) ; if ( this . nodeConfig . proofOfIndex ) { const operationHash = this . storeService . getOperationMerkleRoot ( ) ; if ( ! u8aEq ( operationHash , NULL_MERKEL_ROOT ) ) { const poiBlock = PoiBlock . create ( blockHeight , block . block . header . hash . toHex ( ) , operationHash , await this . poiService . getLatestPoiBlockHash ( ) , this . project . path ) ; poiBlockHash = poiBlock . hash ; await this . storeService . setPoi ( poiBlock , { transaction : tx } ) ; } } } catch ( e ) { await tx . rollback ( ) ; throw e ; } await tx . commit ( ) ; this . fetchService . latestProcessed ( block . block . header . number . toNumber ( ) ) ; this . prevSpecVersion = block . specVersion ; if ( this . nodeConfig . proofOfIndex ) { this . poiService . setLatestPoiBlockHash ( poiBlockHash ) ; } } async start ( ) { await this . dsProcessorService . validateCustomDs ( ) ; await this . fetchService . init ( ) ; this . api = this . apiService . getApi ( ) ; const schema = await this . ensureProject ( ) ; await this . initDbSchema ( schema ) ; this . metadataRepo = await this . ensureMetadata ( schema ) ; if ( this . nodeConfig . proofOfIndex ) { await Promise . all ( [ this . poiService . init ( schema ) , this . mmrService . init ( schema ) , ] ) ; } let startHeight ; const lastProcessedHeight = await this . metadataRepo . findOne ( { where : { key : 'lastProcessedHeight' } , } ) ; if ( lastProcessedHeight !== null && lastProcessedHeight . value !== null ) { startHeight = Number ( lastProcessedHeight . value ) + 1 ; } else { const project = await this . subqueryRepo . findOne ( { where : { name : this . nodeConfig . subqueryName } , } ) ; if ( project !== null ) { startHeight = project . nextBlockHeight ; } else { startHeight = this . getStartBlockFromDataSources ( ) ; } } void this . fetchService . startLoop ( startHeight ) . catch ( ( err ) => { logger . error ( err , 'failed to fetch block' ) ; process . exit ( 1 ) ; } ) ; this . filteredDataSources = this . filterDataSources ( startHeight ) ; this . fetchService . register ( ( block ) => this . indexBlock ( block ) ) ; if ( this . nodeConfig . proofOfIndex ) { void this . mmrService . syncFileBaseFromPoi ( ) . catch ( ( err ) => { logger . error ( err , 'failed to sync poi to mmr' ) ; process . exit ( 1 ) ; } ) ; } } private async ensureProject ( ) { let schema = await this . getExistingProjectSchema ( ) ; if ( ! schema ) { schema = await this . createProjectSchema ( ) ; } else { if ( argv [ 'force-clean' ] ) { try { await this . sequelize . dropSchema ( `"${ schema }"` , { logging : false , benchmark : false , } ) ; await this . sequelize . query ( ` DELETE
              FROM public.subqueries
              WHERE name = :name` , { replacements : { name : this . nodeConfig . subqueryName } , type : QueryTypes . DELETE , } ) ; logger . info ( 'force cleaned schema and tables' ) ; } catch ( err ) { logger . error ( err , 'failed to force clean schema and tables' ) ; } schema = await this . createProjectSchema ( ) ; } } this . eventEmitter . emit ( IndexerEvent . Ready , { value : true , } ) ; return schema ; } private async getExistingProjectSchema ( ) { let schema = this . nodeConfig . localMode ? DEFAULT_DB_SCHEMA : this . nodeConfig . dbSchema ; const schemas = ( await this . sequelize . query ( `SELECT schema_name FROM information_schema.schemata` , { type : QueryTypes . SELECT , } ) . then ( ( xs ) => xs . map ( ( x ) => x . schema_name ) ) . catch ( ( err ) => { logger . error ( `Unable to fetch all schemas: ${ err }` ) ; process . exit ( 1 ) ; } ) ) as [ string ] ; if ( ! schemas . includes ( schema ) ) { const subqueryModel = await this . subqueryRepo . findOne ( { where : { name : this . nodeConfig . subqueryName } , } ) ; if ( subqueryModel ) { schema = subqueryModel . dbSchema ; } else { schema = undefined ; } } return schema ; } private async createProjectSchema ( ) { let schema ; if ( this . nodeConfig . localMode ) { schema = DEFAULT_DB_SCHEMA ; } else { schema = this . nodeConfig . dbSchema ; const schemas = await this . sequelize . showAllSchemas ( undefined ) ; if ( ! ( schemas as unknown as string [ ] ) . includes ( schema ) ) { await this . sequelize . createSchema ( `"${ schema }"` , undefined ) ; } } return schema ; } private async initDbSchema ( schema ) { const graphqlSchema = buildSchema ( path . join ( this . project . path , this . project . schema ) ) ; const modelsRelations = getAllEntitiesRelations ( graphqlSchema ) ; await this . storeService . init ( modelsRelations , schema ) ; } private async ensureMetadata ( schema ) { const metadataRepo = MetadataFactory ( this . sequelize , schema ) ; const project = await this . subqueryRepo . findOne ( { where : { name : this . nodeConfig . subqueryName } , } ) ; this . eventEmitter . emit ( IndexerEvent . NetworkMetadata , this . apiService . networkMeta ) ; const keys = [ 'lastProcessedHeight' , 'blockOffset' , 'indexerNodeVersion' , 'chain' , 'specName' , 'genesisHash' , ] as const ; const entries = await metadataRepo . findAll ( { where : { key : keys , } , } ) ; const keyValue = entries . reduce ( ( arr , curr ) => { arr [ curr . key ] = curr . value ; return arr ; } , { } as { [ key in typeof keys [ number ] ] : string | boolean | number ; } ) ; const { chain , genesisHash , specName } = this . apiService . networkMeta ; if ( ! keyValue . blockOffset ) { const offsetValue = ( this . getStartBlockFromDataSources ( ) - 1 ) . toString ( ) ; await metadataRepo . upsert ( { key : 'blockOffset' , value : offsetValue } ) ; } if ( ! keyValue . genesisHash ) { if ( project ) { await metadataRepo . upsert ( { key : 'genesisHash' , value : project . networkGenesis , } ) ; } else { await metadataRepo . upsert ( { key : 'genesisHash' , value : genesisHash } ) ; } } else { assert ( this . project . network . genesisHash === keyValue . genesisHash , 'Specified project manifest genesis hash does not match database stored genesis hash, consider cleaning project schema using --force-clean' ) ; } if ( keyValue . chain !== chain ) { await metadataRepo . upsert ( { key : 'chain' , value : chain } ) ; } if ( keyValue . specName !== specName ) { await metadataRepo . upsert ( { key : 'specName' , value : specName } ) ; } if ( keyValue . indexerNodeVersion !== packageVersion ) { await metadataRepo . upsert ( { key : 'indexerNodeVersion' , value : packageVersion , } ) ; } return metadataRepo ; } private filterDataSources ( processedHeight ) { let filteredDs = this . getDataSourcesForSpecName ( ) ; if ( filteredDs . length === 0 ) { logger . error ( `Did not find any dataSource match with network specName ${ this . api . runtimeVersion . specName }` ) ; process . exit ( 1 ) ; } filteredDs = filteredDs . filter ( ( ds ) => ds . startBlock <= processedHeight ) ; if ( filteredDs . length === 0 ) { logger . error ( `Your start block is greater than the current indexed block height in your database. Either change your startBlock (project.yaml) to <= ${ processedHeight } 
         or delete your database and start again from the currently specified startBlock` ) ; process . exit ( 1 ) ; } filteredDs = filteredDs . filter ( ( ds ) => { if ( isCustomDs ( ds ) ) { return this . dsProcessorService . getDsProcessor ( ds ) . dsFilterProcessor ( ds , this . api ) ; } else { return true ; } } ) ; if ( ! filteredDs . length ) { logger . error ( `Did not find any datasources with associated processor` ) ; process . exit ( 1 ) ; } return filteredDs ; } private getStartBlockFromDataSources ( ) { const startBlocksList = this . getDataSourcesForSpecName ( ) . map ( ( item ) => item . startBlock ?? 1 ) ; if ( startBlocksList . length === 0 ) { logger . error ( `Failed to find a valid datasource, Please check your endpoint if specName filter is used.` ) ; process . exit ( 1 ) ; } else { return Math . min ( ... startBlocksList ) ; } } private getDataSourcesForSpecName ( ) { return this . project . dataSources . filter ( ( ds ) => ! ds . filter ?. specName || ds . filter . specName === this . api . runtimeVersion . specName . toString ( ) ) ; } private async indexBlockForRuntimeDs ( vm , handlers , { block , events , extrinsics } ) { for ( const handler of handlers ) { switch ( handler . kind ) { case SubqlHandlerKind . Block : if ( SubstrateUtil . filterBlock ( block , handler . filter ) ) { await vm . securedExec ( handler . handler , [ block ] ) ; } break ; case SubqlHandlerKind . Call : { const filteredExtrinsics = SubstrateUtil . filterExtrinsics ( extrinsics , handler . filter ) ; for ( const e of filteredExtrinsics ) { await vm . securedExec ( handler . handler , [ e ] ) ; } break ; } case SubqlHandlerKind . Event : { const filteredEvents = SubstrateUtil . filterEvents ( events , handler . filter ) ; for ( const e of filteredEvents ) { await vm . securedExec ( handler . handler , [ e ] ) ; } break ; } default : } } } private async indexBlockForCustomDs ( ds , vm , { block , events , extrinsics } ) { const plugin = this . dsProcessorService . getDsProcessor ( ds ) ; const assets = await this . dsProcessorService . getAssets ( ds ) ; const processData = async < K extends SubqlHandlerKind > ( processor , handler , filteredData ) : Promise < void > => { const transformedData = await Promise . all ( filteredData . filter ( ( data ) => processor . filterProcessor ( handler . filter , data , ds ) ) . map ( ( data ) => processor . transformer ( data , ds , this . api , assets ) ) ) ; for ( const data of transformedData ) { await vm . securedExec ( handler . handler , [ data ] ) ; } } ; for ( const handler of ds . mapping . handlers ) { const processor = plugin . handlerProcessors [ handler . kind ] ; if ( isBlockHandlerProcessor ( processor ) ) { await processData ( processor , handler , [ block ] ) ; } else if ( isCallHandlerProcessor ( processor ) ) { const filteredExtrinsics = SubstrateUtil . filterExtrinsics ( extrinsics , processor . baseFilter ) ; await processData ( processor , handler , filteredExtrinsics ) ; } else if ( isEventHandlerProcessor ( processor ) ) { const filteredEvents = SubstrateUtil . filterEvents ( events , processor . baseFilter ) ; await processData ( processor , handler , filteredEvents ) ; } } } }