import { Block , Provider } from 'ethers/providers' ; import { getLogger , logError } from '@pigi/core-utils' ; import { EthereumListener } from '../../types/ethereum' ; import { DB } from '../../types/db' ; const log = getLogger ( 'ethereum-block-processor' ) ; const blockKey = Buffer . from ( 'latestBlock' ) ; /**
 * Ethereum Block Processor
 * Single place through which all block subscriptions are handled.
 */ export class EthereumBlockProcessor { private readonly subscriptions : Set < EthereumListener < Block > > ; private currentBlockNumber : number ; private syncInProgress : boolean ; private syncCompleted : boolean ; constructor ( private readonly db , private readonly earliestBlock = 0 ) { this . subscriptions = new Set < EthereumListener < Block > > ( ) ; this . currentBlockNumber = 0 ; this . syncInProgress = false ; this . syncCompleted = false ; } param provider param handler param syncPastBlocks public async subscribe ( provider , handler , syncPastBlocks = true ) { this . subscriptions . add ( handler ) ; provider . on ( 'block' , async ( blockNumber ) => { log . debug ( `Block [${ blockNumber }] was mined!` ) ; await this . fetchAndDisseminateBlock ( provider , blockNumber ) ; this . currentBlockNumber = blockNumber ; try { await this . db . put ( blockKey , Buffer . from ( this . currentBlockNumber . toString ( ) ) ) ; } catch ( e ) { logError ( log , `Error storing most recent block received [${ blockNumber }]!` , e ) ; } } ) ; if ( syncPastBlocks ) { if ( this . syncCompleted ) { await handler . onSyncCompleted ( ) ; return ; } if ( ! this . syncInProgress ) { this . syncInProgress = true ; await this . syncBlocks ( provider ) ; } } } param provider param blockNumber private async fetchAndDisseminateBlock ( provider , blockNumber ) { log . debug ( `Fetching block [${ blockNumber }].` ) ; const block = await provider . getBlock ( blockNumber , true ) ; log . debug ( `Received block: [${ JSON . stringify ( block ) }].` ) ; this . subscriptions . forEach ( ( h ) => { try { h . handle ( block ) ; } catch ( e ) { } } ) ; } param provider private async syncBlocks ( provider ) { log . debug ( `Syncing blocks` ) ; const blockNumber = await this . getBlockNumber ( provider ) ; const lastSyncedBlockBuffer = await this . db . get ( blockKey ) ; const lastSyncedNumber = ! ! lastSyncedBlockBuffer ? parseInt ( lastSyncedBlockBuffer . toString ( ) , 10 ) : this . earliestBlock - 1 ; if ( blockNumber === lastSyncedNumber ) { log . debug ( `Up to date, not syncing.` ) ; this . finishSync ( blockNumber , blockNumber ) ; return ; } for ( let i = lastSyncedNumber + 1 ; i <= blockNumber ; i ++ ) { await this . fetchAndDisseminateBlock ( provider , i ) ; } this . finishSync ( lastSyncedNumber + 1 , blockNumber ) ; } private finishSync ( syncStart , currentBlock ) { this . syncCompleted = true ; this . syncInProgress = false ; log . debug ( `Synced from block [${ syncStart }] to [${ currentBlock }]!` ) ; for ( const callback of this . subscriptions ) { callback . onSyncCompleted ( ) . catch ( ( e ) => { logError ( log , 'Error calling Block sync callback' , e ) ; } ) ; } } param provider returns private async getBlockNumber ( provider ) { if ( this . currentBlockNumber === 0 ) { this . currentBlockNumber = await provider . getBlockNumber ( ) ; } log . debug ( `Current block number: ${ this . currentBlockNumber }` ) ; return this . currentBlockNumber ; } }