feat: enhance EventFlow and EventSignal to support multiple addresses; refactor test cases and improve debugging capabilities

Author: murongg

package.json

@@ -47,7 +47,8 @@
     "publish": "nr prepublishOnly && esno scripts/publish.ts",
     "release": "esno scripts/release.ts",
     "start": "esno src/index.ts",
-    "test": "DEBUG=DEBUG:ora-stack:* vitest",
+    "test": "vitest",
+    "test:debug": "DEBUG=DEBUG:ora-stack:* vitest",
     "typecheck": "tsc --noEmit"
   },
   "devDependencies": {

packages/orap/examples/declarativeDemo/app.ts

@@ -2,7 +2,7 @@
 import type { ContractEventPayload } from 'ethers'
 import { Logger, objectKeys, randomStr, redisStore } from '@ora-io/utils'
 import type { ListenOptions, ToKeyFn } from '../../src'
-import { CheckTransactionStatus, Orap, StoreManager, getMiddlewareContext } from '../../src'
+import { Orap, StoreManager, getMiddlewareContext } from '../../src'
 import ABI from './erc20.abi.json'
 
 const MAINNET_USDT_ADDR = '0xdAC17F958D2ee523a2206206994597C13D831ec7'
@@ -24,7 +24,12 @@ export function startDemo(options: ListenOptions, storeConfig?: any) {
 
   const toKey: ToKeyFn = (from: string, _to: string, _amount: number) => `${from}_${randomStr(4)}`
 
-  const event = orap.event(eventSignalParam.address, eventSignalParam.abi, eventSignalParam.eventName)
+  const event = orap.event({
+    address: eventSignalParam.address,
+    abi: eventSignalParam.abi,
+    eventName: eventSignalParam.eventName,
+    enableSubscribe: false,
+  })
     .crosscheck({
       store,
       storeKeyPrefix: 'ora-stack:orap:demo:cc:',
@@ -43,27 +48,34 @@ export function startDemo(options: ListenOptions, storeConfig?: any) {
     .key(toKey)
     .prefix('ora-stack:orap:demo:TransferTask:', 'ora-stack:orap:demo:Done-TransferTask:')
     .ttl({ taskTtl: 120000, doneTtl: 60000 })
-    .use(CheckTransactionStatus(options.wsProvider))
+    // .use(CheckTransactionStatus(options.wsProvider))
     .handle(handleTask)
     // add another task
-    .another()
-    .task()
-    .prefix('ora-stack:orap:demo:AnotherTask:', 'ora-stack:orap:demo:Done-AnotherTask:')
-    .cache(sm) // rm to use mem by default
-    .ttl({ taskTtl: 20000, doneTtl: 20000 })
-    .handle(handleTask_2)
-
-  event.stop()
+    // .another()
+    // .task()
+    // .prefix('ora-stack:orap:demo:AnotherTask:', 'ora-stack:orap:demo:Done-AnotherTask:')
+    // .cache(sm) // rm to use mem by default
+    // .ttl({ taskTtl: 20000, doneTtl: 20000 })
+    // .handle(handleTask_2)
 
   // start signal listener
   orap.listen(
     options,
     () => { logger.log('listening on provider.network') },
   )
+
+  setTimeout(() => {
+    logger.log('[+] add another address')
+    event.addresses([
+      eventSignalParam.address,
+      '0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48',
+    ])
+    event.restart()
+  }, 10 * 1000)
 }
 
-async function handleTask(from: string, to: string, amount: number, _event: ContractEventPayload) {
-  logger.log('[+] handleTask: from =', from, 'to =', to, 'amount =', amount)
+async function handleTask(from: string, to: string, amount: number, event: ContractEventPayload) {
+  logger.log('[+] handleTask: from =', from, 'to =', to, 'amount =', amount, 'address =', event.log.address)
   const args = objectKeys(arguments).map(k => arguments[k])
 
   const { next } = getMiddlewareContext(...args)
@@ -76,10 +88,10 @@ async function newEventSignalHook(from: string, to: string, amount: number, even
   return true // true to continue handle tasks, false to hijack the process.
 }
 
-async function handleTask_2(from: string, to: string, amount: number) {
-  logger.log('[+] handleTask_2: from =', from, 'to =', to, 'amount =', amount)
-  const args = objectKeys(arguments).map(k => arguments[k])
+// async function handleTask_2(from: string, to: string, amount: number) {
+//   logger.log('[+] handleTask_2: from =', from, 'to =', to, 'amount =', amount)
+//   const args = objectKeys(arguments).map(k => arguments[k])
 
-  const { next } = getMiddlewareContext(...args)
-  await next()
-}
+//   const { next } = getMiddlewareContext(...args)
+//   await next()
+// }

packages/orap/src/flow/event.test.ts

@@ -1,6 +1,7 @@
 import { ethers } from 'ethers'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 import { EventVerse } from '../verse/event'
+import { ERC20_ABI, USDT_ADDRESS } from '../../tests/config'
 import { OrapFlow } from './orap'
 import { EventFlow } from './event'
 
@@ -10,7 +11,7 @@ describe('EventFlow', () => {
 
   beforeEach(() => {
     orapFlow = new OrapFlow()
-    eventFlow = new EventFlow(orapFlow)
+    eventFlow = new EventFlow(orapFlow, { address: USDT_ADDRESS, abi: ERC20_ABI, eventName: 'Transfer' })
   })
 
   it('should create a task flow', () => {
@@ -56,4 +57,36 @@ describe('EventFlow', () => {
     eventFlow.stop()
     expect(stopFn).toHaveBeenCalled()
   })
+
+  describe('address', () => {
+    it('should set the address', () => {
+      const eventFlow = new EventFlow(orapFlow, { address: USDT_ADDRESS, abi: ERC20_ABI, eventName: 'Transfer' })
+      eventFlow.address('0x1234567890123456789012345678901234567890')
+      expect(eventFlow.params.address).toEqual(['0x1234567890123456789012345678901234567890'])
+    })
+
+    it.only('should set the array address', () => {
+      const eventFlow = new EventFlow(orapFlow, { address: [USDT_ADDRESS], abi: ERC20_ABI, eventName: 'Transfer' })
+      eventFlow.address(1, '0x1234567890123456789012345678901234567890')
+      expect(eventFlow.params.address).toContainEqual('0x1234567890123456789012345678901234567890')
+    })
+
+    it('should set the address with number', () => {
+      const eventFlow = new EventFlow(orapFlow, { address: USDT_ADDRESS, abi: ERC20_ABI, eventName: 'Transfer' })
+      eventFlow.address(1, '0x1234567890123456789012345678901234567890')
+      expect(eventFlow.params.address).toEqual(['0x1234567890123456789012345678901234567890'])
+    })
+
+    it('should set the addresses with array', () => {
+      const eventFlow = new EventFlow(orapFlow, { address: USDT_ADDRESS, abi: ERC20_ABI, eventName: 'Transfer' })
+      eventFlow.addresses(['0x1234567890123456789012345678901234567890'])
+      expect(eventFlow.params.address).toEqual(['0x1234567890123456789012345678901234567890'])
+    })
+
+    it('should set the addresses with array and number', () => {
+      const eventFlow = new EventFlow(orapFlow, { address: USDT_ADDRESS, abi: ERC20_ABI, eventName: 'Transfer' })
+      eventFlow.addresses([USDT_ADDRESS, '0x1234567890123456789012345678901234567890'])
+      expect(eventFlow.params.address).toEqual([USDT_ADDRESS, '0x1234567890123456789012345678901234567890'])
+    })
+  })
 })

packages/orap/src/flow/event.ts

@@ -1,4 +1,5 @@
 import type { AutoCrossCheckParam, Providers } from '@ora-io/reku'
+import type { ContractAddress } from '@ora-io/utils/src'
 import type { Context } from '../task'
 import type { TaskFlowParams } from '../flow/task'
 import { TaskFlow } from '../flow/task'
@@ -19,16 +20,24 @@ export class EventFlow implements Flow {
   private _crosscheckProvider?: Providers
 
   private _verse: EventVerse = new EventVerse(this)
+  private _addresses: ContractAddress[] = []
+  private _params: EventSignalRegisterParams
 
   get verse() {
     return this._verse
   }
 
+  get params() {
+    return this._params
+  }
+
   constructor(
-    private parentFlow?: OrapFlow,
-    public params?: EventSignalRegisterParams,
+    private parentFlow: OrapFlow,
+    params: EventSignalRegisterParams,
     handleFn?: HandleFn, // return: succ & continue if true, stop if false
   ) {
+    this._params = params
+    this._addresses = Array.isArray(this.params.address) ? this.params.address : [this.params.address]
     // Default handleFn
     this.handleFn = handleFn ?? (async (..._args: Array<any>) => {
       return true
@@ -58,6 +67,7 @@ export class EventFlow implements Flow {
       tf = new TaskFlow(this, sm)
     }
     this._taskFlows.push(tf)
+    this._verse.setTaskVerses(this._taskFlows.map(flow => flow.verse))
     return tf
   }
 
@@ -95,14 +105,40 @@ export class EventFlow implements Flow {
   }
 
   another(): OrapFlow {
-    return this.parentFlow!
+    return this.parentFlow
   }
 
-  stop() {
+  stop(): this {
     this._verse.stop()
+    return this
   }
 
-  restart() {
+  restart(): this {
     this._verse.restart()
+    return this
+  }
+
+  address(_index: number, _address: ContractAddress): this
+  address(address: ContractAddress): this
+  address(_first: ContractAddress | number, _second?: ContractAddress): this {
+    if (typeof _first === 'number') {
+      if (!_second)
+        throw new Error('address is required')
+      if (Array.isArray(this._params.address))
+        Reflect.set(this._addresses, _first, _second)
+
+      else this._addresses = [_second!]
+    }
+    else
+      if (Array.isArray(this._params.address)) { this._addresses = [...new Set([...this._params.address, _first])] }
+      else { this._addresses = [_first] }
+    this._params.address = this._addresses
+    return this
+  }
+
+  addresses(addresses: ContractAddress[]): this {
+    this._addresses = addresses
+    this._params.address = this._addresses
+    return this
   }
 }

packages/orap/src/flow/orap.ts

@@ -51,7 +51,7 @@ export class OrapFlow implements Flow {
    * @param options
    * @param onListenFn
    */
-  listen(options: ListenOptions, onListenFn?: Fn) {
+  listen(options: ListenOptions, onListenFn?: Fn): this {
     for (const eventFlow of this.subflows.event) {
       eventFlow.setSubscribeProvider(options.wsProvider)
       if (options.httpProvider)
@@ -65,14 +65,17 @@ export class OrapFlow implements Flow {
 
     this._verse.play()
     this.onListenFn()
+    return this
   }
 
-  stop() {
+  stop(): this {
     this._verse.stop()
+    return this
   }
 
-  restart() {
+  restart(): this {
     this._verse.restart()
+    return this
   }
 
   assemble(): OrapVerse {

packages/orap/src/flow/task.test.ts

@@ -1,14 +1,15 @@
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 import { memoryStore } from '@ora-io/utils'
 import { StoreManager } from '../store'
-import { EventFlow, TaskFlow } from '.'
+import { ERC20_ABI, USDT_ADDRESS } from '../../tests/config'
+import { EventFlow, OrapFlow, TaskFlow } from '.'
 
 describe('TaskFlow', () => {
   let parentFlow: any
   let taskFlow: TaskFlow
 
   beforeEach(() => {
-    parentFlow = new EventFlow()
+    parentFlow = new EventFlow(new OrapFlow(), { address: USDT_ADDRESS, abi: ERC20_ABI, eventName: 'Transfer' })
     taskFlow = new TaskFlow(parentFlow)
   })
 

packages/orap/src/flow/task.ts

@@ -144,11 +144,13 @@ export class TaskFlow implements Flow {
     return new TaskVerse(this)
   }
 
-  stop() {
+  stop(): this {
     this._verse.stop()
+    return this
   }
 
-  restart() {
+  restart(): this {
     this._verse.restart()
+    return this
   }
 }

packages/orap/src/signal/event.test.ts

@@ -2,6 +2,7 @@ import { ethers } from 'ethers'
 import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 import type { AutoCrossCheckParam } from '@ora-io/reku'
 import { AutoCrossChecker, RekuProviderManager } from '@ora-io/reku'
+import type { ContractAddress } from '@ora-io/utils/src'
 import { ERC20_ABI, MAINNET_WSS, USDT_ADDRESS } from '../../tests/config'
 import type { EventSignalCallback, EventSignalRegisterParams } from './event'
 import { EventSignal } from './event'
@@ -30,11 +31,6 @@ describe('EventSignal', () => {
       expect(eventSignal).toBeInstanceOf(EventSignal)
     })
 
-    it('should set the contract property', () => {
-      expect(eventSignal.contract).toBeInstanceOf(ethers.Contract)
-      expect(eventSignal.contract.interface).toEqual(new ethers.Interface(ERC20_ABI))
-    })
-
     it('should set the eventFragment property', () => {
       expect(eventSignal.eventFragment).toBeInstanceOf(ethers.EventFragment)
       expect(eventSignal.eventFragment.name).toBe(params.eventName)
@@ -91,29 +87,37 @@ describe('EventSignal', () => {
     const provider = {} as any
 
     beforeEach(() => {
-      eventSignal.contract.connect = vi.fn().mockReturnValue({
-        on: vi.fn(),
-      })
+      // Mock the contract in the contractMap
+      const mockContract = {
+        connect: vi.fn().mockReturnValue({
+          on: vi.fn(),
+        }),
+        removeListener: vi.fn(),
+      } as any
+      eventSignal.contractMap.set(params.address as ContractAddress, mockContract)
     })
 
     it('should add the contract to the provider if provider is an instance of RekuProviderManager', () => {
       rekuProviderManager.addContract = vi.fn()
       rekuProviderManager.addListener = vi.fn()
       eventSignal.startEventListener(rekuProviderManager)
-      expect(rekuProviderManager.addContract).toHaveBeenCalledWith(params.address, eventSignal.contract)
+      expect(rekuProviderManager.addContract).toHaveBeenCalledWith(params.address, eventSignal.contractMap.get(params.address as ContractAddress))
       expect(rekuProviderManager.addListener).toHaveBeenCalledWith(params.address, params.eventName, eventSignal.subscribeCallback)
     })
 
     it('should connect the contract to the provider if provider is not an instance of RekuProviderManager', () => {
       eventSignal.startEventListener(provider)
-      expect(eventSignal.contract.connect).toHaveBeenCalledWith(provider)
+      expect(eventSignal.contractMap.get(params.address as ContractAddress)?.connect).toHaveBeenCalledWith(provider)
     })
 
     it('should call the on method of the listener with the eventName and subscribeCallback', () => {
       const listener = {
         on: vi.fn(),
       }
-      eventSignal.contract.connect = vi.fn().mockReturnValue(listener)
+      const mockContract = eventSignal.contractMap.get(params.address as ContractAddress)
+      if (mockContract)
+        mockContract.connect = vi.fn().mockReturnValue(listener)
+
       eventSignal.startEventListener(provider)
       expect(listener.on).toHaveBeenCalledWith(params.eventName, eventSignal.subscribeCallback)
     })