feat(crosscheck): support provider manager

Author: murongg

packages/orap/index.ts

@@ -2,4 +2,3 @@ export * from './signal'
 export * from './store'
 export * from './task'
 export * from './orap'
-export * from './types'

packages/orap/orap.ts

@@ -1,6 +1,6 @@
-import { ProviderManager } from '@ora-io/rek'
+import type { Providers } from '@ora-io/rek'
+import { RekProviderManager } from '@ora-io/rek'
 import { EventSignal } from './signal/event'
-import type { Providers } from './types'
 
 export interface ListenOptions {
   wsProvider: Providers
@@ -26,9 +26,9 @@ export class Orap {
 
   _listenChain(wsProvider: Providers | string, httpProvider?: Providers | string) {
     if (typeof wsProvider === 'string')
-      wsProvider = new ProviderManager(wsProvider)
+      wsProvider = new RekProviderManager(wsProvider)
     if (httpProvider && typeof httpProvider === 'string')
-      httpProvider = new ProviderManager(httpProvider)
+      httpProvider = new RekProviderManager(httpProvider)
     this.routes.event.forEach(es => es.listen(wsProvider as Providers, httpProvider as Providers))
   }
 

packages/orap/signal/event.ts

@@ -1,8 +1,7 @@
 import type { EventFragment, Log } from 'ethers'
 import { ethers } from 'ethers'
-import { AutoCrossChecker, ONE_MINUTE_MS, ProviderManager } from '@ora-io/rek'
-import type { AutoCrossCheckParam } from '@ora-io/rek'
-import type { Providers } from '../types'
+import { AutoCrossChecker, ONE_MINUTE_MS, RekProviderManager } from '@ora-io/rek'
+import type { AutoCrossCheckParam, Providers } from '@ora-io/rek'
 import type { Signal } from './type'
 
 export interface EventSignalRegisterParams {
@@ -62,7 +61,7 @@ export class EventSignal implements Signal {
     this.provider = provider
 
     // start event listener
-    if (provider instanceof ProviderManager) {
+    if (provider instanceof RekProviderManager) {
       provider.addContract(this.params.address, this.contract)
       provider.addListener(this.params.address, this.params.eventName, this.subscribeCallback)
     }
@@ -79,15 +78,13 @@ export class EventSignal implements Signal {
     if (this.crosscheckerOptions) {
       if (!crosscheckProvider)
         throw new Error('crosschecker set, please provide crosscheckProvider to listen function')
-      const _crosscheckProvider = crosscheckProvider instanceof ProviderManager ? crosscheckProvider.provider : crosscheckProvider
-      if (_crosscheckProvider)
-        this.startCrossChecker(_crosscheckProvider)
+      this.startCrossChecker(crosscheckProvider)
     }
 
     return this
   }
 
-  async startCrossChecker(provider: ethers.WebSocketProvider | ethers.JsonRpcProvider) {
+  async startCrossChecker(provider: Providers) {
     if (!this.crosscheckerOptions)
       throw new Error('no crosscheck set, can\'t start crosschecker')
     this.crosschecker = new AutoCrossChecker(provider, this.crosscheckerOptions)

packages/rek/event/crosschecker/autochecker.ts

@@ -1,6 +1,7 @@
 import type { ethers } from 'ethers'
 import { logger, polling, retryOnNull, sleep } from '@ora-io/utils'
 import { ETH_BLOCK_INTERVAL_MS } from '../../constants'
+import type { Providers } from '../../types/w3'
 import { CrossCheckerCacheManager } from './cache/manager'
 import type { AutoCrossCheckParam, CrossCheckRangeParam } from './interface'
 import { BaseCrossChecker } from './basechecker'
@@ -11,7 +12,7 @@ export class AutoCrossChecker extends BaseCrossChecker {
   checkpointBlockNumber = 0
 
   constructor(
-    provider: ethers.Provider,
+    provider: Providers,
     options?: AutoCrossCheckParam,
   ) {
     super(provider)
@@ -51,7 +52,7 @@ export class AutoCrossChecker extends BaseCrossChecker {
   async start(options: AutoCrossCheckParam) {
     this.validate(options)
     // TODO: use blockNumber for performance
-    const latestblocknum = await retryOnNull(async () => await this.provider.getBlockNumber())
+    const latestblocknum = await retryOnNull(async () => await this.provider.provider?.getBlockNumber())
     const {
       fromBlock = latestblocknum,
       batchBlocksCount = 10,
@@ -90,7 +91,7 @@ export class AutoCrossChecker extends BaseCrossChecker {
 
     const waitNextCrosscheck = async (): Promise<boolean> => {
       // TODO: use blockNumber for performance
-      const latestblocknum = (await retryOnNull(async () => await this.provider.getBlock('latest'))).number
+      const latestblocknum = (await retryOnNull(async () => await this.provider.provider?.getBlock('latest'))).number
       logger.info(`[*] ccrOptions: fromBlock ${ccrOptions.fromBlock}, toBlock ${ccrOptions.toBlock}, latestblocknum ${latestblocknum}`)
       if (ccrOptions.toBlock + delayBlockFromLatest > latestblocknum) {
         // sleep until the toBlock

packages/rek/event/crosschecker/basechecker.ts

@@ -1,11 +1,12 @@
 import type { ethers } from 'ethers'
 import { logger, retryOnNull } from '@ora-io/utils'
 import { ETH_BLOCK_COUNT_ONE_HOUR } from '../../constants'
+import type { Providers } from '../../types/w3'
 import type { CrossCheckFromParam, CrossCheckRangeParam, CrossCheckRetroParam, SimpleLog } from './interface'
 
 export class BaseCrossChecker {
-  provider: ethers.Provider
-  constructor(provider: ethers.Provider) {
+  provider: Providers
+  constructor(provider: Providers) {
     this.provider = provider
   }
 
@@ -33,7 +34,7 @@ export class BaseCrossChecker {
 
     // define from, to
     // TODO: use blockNumber for performance
-    const block = await retryOnNull(async () => await this.provider.getBlock('latest'))
+    const block = await retryOnNull(async () => await this.provider.provider?.getBlock('latest'))
     const options: CrossCheckRangeParam = {
       ...ccrOptions,
       fromBlock: block.number - retroBlockCount,
@@ -52,7 +53,7 @@ export class BaseCrossChecker {
     ccfOptions: CrossCheckFromParam,
   ) {
     // TODO: use blockNumber for performance
-    const block = await retryOnNull(async () => await this.provider.getBlock('latest'))
+    const block = await retryOnNull(async () => await this.provider.provider?.getBlock('latest'))
 
     // suggest use large retroBlockCount
     if (block.number - ccfOptions.fromBlock < ETH_BLOCK_COUNT_ONE_HOUR)
@@ -103,16 +104,18 @@ export class BaseCrossChecker {
       ...(address && { address }),
       ...(topics && { topics }),
     }
-    const logs = await this.provider.getLogs(params)
-    // get ignoreLogs keys
-    const ignoreLogs = options.ignoreLogs
+    if (this.provider.provider) {
+      const logs = await this.provider.provider?.getLogs(params)
+      // get ignoreLogs keys
+      const ignoreLogs = options.ignoreLogs
 
-    // crosscheck missing logs
-    const missingLogs = ignoreLogs ? await this.diff(logs, ignoreLogs) : logs
+      // crosscheck missing logs
+      const missingLogs = ignoreLogs ? await this.diff(logs, ignoreLogs) : logs
 
-    // callback on missing logs
-    for (const log of missingLogs)
-      await options.onMissingLog(log)
+      // callback on missing logs
+      for (const log of missingLogs)
+        await options.onMissingLog(log)
+    }
   }
 }
 

packages/rek/index.ts

@@ -2,4 +2,5 @@ export * from './event/crosschecker/basechecker'
 export * from './event/crosschecker/autochecker'
 export * from './event/crosschecker/interface'
 export * from './constants'
-export * from './managers'
+export * from './provider'
+export * from './types'

packages/rek/managers/contract.ts renamed to packages/rek/provider/contract.ts

@@ -4,16 +4,16 @@ import { Interface, WebSocketProvider, ethers } from 'ethers'
 import type { ErrorEvent, WebSocket } from 'ws'
 import { ContractManager } from './contract'
 
-export interface ProviderManagerOptions {
+export interface RekProviderManagerOptions {
   /**
    * The interval of heartbeat, default is 10s
    */
   heartbeatInterval?: number
 }
 
-export type ProviderManagerEvent = 'error' | 'close'
+export type RekProviderManagerEvent = 'error' | 'close'
 
-export class ProviderManager {
+export class RekProviderManager {
   private _provider?: ethers.JsonRpcProvider | ethers.WebSocketProvider
   private _contracts: Map<string, ContractManager> = new Map()
 
@@ -23,7 +23,7 @@ export class ProviderManager {
 
   private _event?: EventEmitter
 
-  constructor(public providerUrl: string, options?: ProviderManagerOptions) {
+  constructor(public providerUrl: string, options?: RekProviderManagerOptions) {
     this.connect()
     if (options?.heartbeatInterval)
       this._heartbeatInterval = options.heartbeatInterval
@@ -99,21 +99,21 @@ export class ProviderManager {
     })
   }
 
-  on(event: ProviderManagerEvent, listener: (...args: any[]) => void) {
+  on(event: RekProviderManagerEvent, listener: (...args: any[]) => void) {
     if (!this._event)
       this._event = new EventEmitter()
 
     this._event?.on(event, listener)
   }
 
-  once(event: ProviderManagerEvent, listener: (...args: any[]) => void) {
+  once(event: RekProviderManagerEvent, listener: (...args: any[]) => void) {
     if (!this._event)
       this._event = new EventEmitter()
 
     this._event?.once(event, listener)
   }
 
-  removeEvent(event: ProviderManagerEvent, listener: (...args: any[]) => void) {
+  removeEvent(event: RekProviderManagerEvent, listener: (...args: any[]) => void) {
     this._event?.removeListener(event, listener)
   }
 

packages/rek/managers/index.ts renamed to packages/rek/provider/index.ts

@@ -0,0 +1 @@
+export * from './w3'