import { getLogger , logError , Md5Hash } from '@pigi/core-utils' ; import { ethers , Contract } from 'ethers' ; import { LogDescription } from 'ethers/utils' ; import { Filter , Log , Provider } from 'ethers/providers' ; import { EthereumEvent , EthereumListener } from '../../types/ethereum' ; import { DB } from '../../types/db' ; const log = getLogger ( 'ethereum- event-processor' ) ; interface SyncStatus { syncCompleted : boolean ; syncInProgress : boolean ; } /**
 * Ethereum EthereumEvent Processor
 * The single class to process and disseminate all Ethereum EthereumEvent subscriptions.
 */ export class EthereumEventProcessor { private readonly subscriptions : Map < string , Set < EthereumListener < EthereumEvent > > > ; private currentBlockNumber : number ; private syncStatuses : Map < string , SyncStatus > ; constructor ( private readonly db , private readonly earliestBlock = 0 ) { this . subscriptions = new Map < string , Set < EthereumListener < EthereumEvent > > > ( ) ; this . currentBlockNumber = 0 ; this . syncStatuses = new Map < string , SyncStatus > ( ) ; } param contract param eventName param handler param syncPastEvents public async subscribe ( contract , eventName , handler , syncPastEvents = true ) { const eventId = this . getEventID ( contract . address , eventName ) ; log . debug ( `Received subscriber for event ${ eventName }, ID: ${ eventId }` ) ; if ( ! this . subscriptions . has ( eventId ) ) { this . subscriptions . set ( eventId , new Set < EthereumListener < EthereumEvent > > ( [ handler ] ) ) ; } else { this . subscriptions . get ( eventId ) . add ( handler ) ; return ; } contract . on ( contract . filters [ eventName ] ( ) , async ( ... data ) => { log . debug ( `Received live event: ${ JSON . stringify ( data ) }` ) ; const ethersEvent = data [ data . length - 1 ] ; const event = this . createEventFromEthersEvent ( ethersEvent ) ; await this . handleEvent ( event ) ; try { await this . db . put ( Buffer . from ( event . eventID ) , Buffer . from ( ethersEvent . blockNumber . toString ( 10 ) ) ) ; } catch ( e ) { logError ( log , `Error storing most recent events block [${ ethersEvent . blockNumber }]!` , e ) ; } } ) ; if ( syncPastEvents ) { if ( ! this . syncStatuses . has ( eventId ) ) { this . syncStatuses . set ( eventId , { syncInProgress : true , syncCompleted : false , } ) ; await this . syncPastEvents ( contract , eventName , eventId ) ; return ; } const syncStatus = this . syncStatuses . get ( eventId ) ; if ( syncStatus . syncCompleted ) { await handler . onSyncCompleted ( eventName ) ; } } } param contract param eventName param eventId private async syncPastEvents ( contract , eventName , eventId ) { log . debug ( `Syncing events for event ${ eventName }` ) ; const blockNumber = await this . getBlockNumber ( contract . provider ) ; const lastSyncedBlockBuffer = await this . db . get ( Buffer . from ( eventId ) ) ; const lastSyncedNumber = ! ! lastSyncedBlockBuffer ? parseInt ( lastSyncedBlockBuffer . toString ( ) , 10 ) : this . earliestBlock ; if ( blockNumber === lastSyncedNumber ) { log . debug ( `Up to date, not syncing.` ) ; this . finishSync ( eventId , eventName , 0 ) ; return ; } const filter = contract . filters [ eventName ] ( ) ; filter . fromBlock = lastSyncedNumber + 1 ; filter . toBlock = 'latest' ; const logs = await contract . provider . getLogs ( filter ) ; const events = logs . map ( ( l ) => { const logDesc = contract . interface . parseLog ( l ) ; return EthereumEventProcessor . createEventFromLogDesc ( logDesc , eventId ) ; } ) ; for ( const event of events ) { await this . handleEvent ( event ) ; } this . finishSync ( eventId , eventName , events . length ) ; } private finishSync ( eventId , eventName , numEvents ) { const status = this . syncStatuses . get ( eventId ) ; status . syncCompleted = true ; status . syncInProgress = false ; log . debug ( `Synced events for event ${ eventName }, ${ eventId }. Found ${ numEvents } events` ) ; for ( const subscription of this . subscriptions . get ( eventId ) ) { subscription . onSyncCompleted ( eventId ) . catch ( ( e ) => { logError ( log , 'Error calling EthereumEvent sync callback' , e ) ; } ) ; } } param event private async handleEvent ( event ) { log . debug ( `Handling event ${ JSON . stringify ( event ) }` ) ; const subscribers = this . subscriptions . get ( event . eventID ) ; subscribers . forEach ( ( s ) => { try { s . handle ( event ) ; } catch ( e ) { } } ) ; } param provider returns private async getBlockNumber ( provider ) { if ( this . currentBlockNumber === 0 ) { this . currentBlockNumber = await provider . getBlockNumber ( ) ; } return this . currentBlockNumber ; } param logDesc param eventID returns private static createEventFromLogDesc ( logDesc , eventID ) { const values = EthereumEventProcessor . getLogValues ( logDesc . values ) ; return { eventID , name : logDesc . name , signature : logDesc . signature , values , } ; } param event returns private createEventFromEthersEvent ( event ) { const values = EthereumEventProcessor . getLogValues ( event . args ) ; return { eventID : this . getEventID ( event . address , event . event ) , name : event . event , signature : event . eventSignature , values , } ; } param logArgs returns private static getLogValues ( logArgs ) { const values = { ... logArgs } ; for ( let i = 0 ; i < logArgs [ 'length' ] ; i ++ ) { delete values [ i . toString ( ) ] ; } delete values [ 'length' ] ; return values ; } param address param eventName returns private getEventID ( address , eventName ) { return Md5Hash ( `${ address }${ eventName }` ) ; } }