Merge pull request #19 from ora-io/feat/middleware

feat(orap): middleware

Author: murongg

packages/orap/examples/declarativeDemo/app.ts

@@ -1,7 +1,8 @@
+/* eslint-disable prefer-rest-params */
 import type { ContractEventPayload } from 'ethers'
-import { Logger, randomStr, redisStore } from '@ora-io/utils'
+import { Logger, objectKeys, randomStr, redisStore } from '@ora-io/utils'
 import type { ListenOptions, ToKeyFn } from '../../src'
-import { Orap, StoreManager } from '../../src'
+import { Orap, StoreManager, getTaskContext } from '../../src'
 import ABI from './erc20.abi.json'
 
 const MAINNET_USDT_ADDR = '0xdAC17F958D2ee523a2206206994597C13D831ec7'
@@ -59,7 +60,10 @@ export function startDemo(options: ListenOptions, storeConfig?: any) {
 
 async function handleTask(from: string, to: string, amount: number, _event: ContractEventPayload) {
   logger.log('[+] handleTask: from =', from, 'to =', to, 'amount =', amount)
-  return true
+  const args = objectKeys(arguments).map(k => arguments[k])
+
+  const { next } = getTaskContext(...args)
+  await next()
 }
 
 async function newEventSignalHook(from: string, to: string, amount: number, event: ContractEventPayload) {
@@ -70,5 +74,8 @@ async function newEventSignalHook(from: string, to: string, amount: number, even
 
 async function handleTask_2(from: string, to: string, amount: number) {
   logger.log('[+] handleTask_2: from =', from, 'to =', to, 'amount =', amount)
-  return true
+  const args = objectKeys(arguments).map(k => arguments[k])
+
+  const { next } = getTaskContext(...args)
+  await next()
 }

packages/orap/src/flow/interface.ts

@@ -3,7 +3,8 @@ import type { Context, TaskRaplized } from '../task'
 import type { Verse } from '../verse/interface'
 
 export type ToKeyFn = (...eventLog: Array<any>) => Awaitable<string>
-export type HandleFn = (...eventLog: Array<any>) => Awaitable<boolean>
+export type NextFunction = (...args: any[]) => void
+export type HandleFn = (...args: any[]) => Awaitable<any>
 export type HandleResultFn = (task: TaskRaplized) => Awaitable<void>
 export type PrefixFn = ((...eventLog: Array<any>) => Awaitable<string>) | ((context?: Context) => string) // the later context one is kind of useless, can rm
 export type Prefix = PrefixFn | string

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

@@ -54,6 +54,7 @@ describe('TaskFlow', () => {
     const handler = vi.fn()
     const result = taskFlow.handle(handler)
     expect(result).toBe(taskFlow)
+    expect(taskFlow.middlewares).toContain(handler)
     expect(taskFlow.handleFn).toBe(handler)
   })
 

packages/orap/src/flow/task.ts

@@ -28,6 +28,8 @@ export class TaskFlow implements Flow {
   handleFn: HandleFn = defaultHandleFn
   successFn: HandleResultFn = defaultSuccessFn
 
+  private _middlewares: Array<HandleFn> = []
+
   failFn: HandleResultFn = async (task: TaskRaplized) => {
     await task.remove()
   }
@@ -38,6 +40,10 @@ export class TaskFlow implements Flow {
     private parentFlow: EventFlow,
   ) { }
 
+  get middlewares() {
+    return this._middlewares
+  }
+
   cache(sm: StoreManager) {
     this.sm = sm
     return this
@@ -73,6 +79,12 @@ export class TaskFlow implements Flow {
 
   handle(handler: HandleFn): this {
     this.handleFn = handler
+    this._middlewares.push(handler)
+    return this
+  }
+
+  use(middleware: HandleFn): this {
+    this._middlewares.push(middleware)
     return this
   }
 

packages/orap/src/index.ts

@@ -6,5 +6,7 @@ export * from './verse'
 export * from './signal'
 export * from './store'
 export * from './task'
+export * from './middlewares'
+export * from './utils'
 
 export default Orap

packages/orap/src/middlewares/CheckTransactionStatus.ts

@@ -0,0 +1,22 @@
+import { ContractEventPayload } from 'ethers'
+import type { Providers } from '@ora-io/reku'
+import { getTaskContext } from '../utils'
+
+export const CheckTransactionStatus = (provider: Providers) => {
+  if (!provider)
+    throw new Error('provider is required')
+
+  return async (...args: any[]) => {
+    const { next } = getTaskContext(...args)
+    const contractEventPayload = args.at(-3) as ContractEventPayload
+    if (contractEventPayload instanceof ContractEventPayload) {
+      const tx = await provider.provider.getTransactionReceipt(contractEventPayload.log.transactionHash)
+      if (!tx || tx?.status === 0)
+        throw new Error('Transaction failed')
+      await next()
+    }
+    else {
+      throw new TypeError('Invalid contract event payload')
+    }
+  }
+}

packages/orap/src/middlewares/HandleFailedMiddleware.ts

@@ -0,0 +1,13 @@
+import type { HandleFn } from '../flow'
+import { getTaskContext } from '../utils'
+
+export const HandleFailedMiddleware: HandleFn = async (...args: any[]) => {
+  const { next, task } = getTaskContext(...args)
+
+  try {
+    await next()
+  }
+  catch (error) {
+    await task.flow.failFn(task)
+  }
+}

packages/orap/src/middlewares/HandleSuccessMiddleware.ts

@@ -0,0 +1,10 @@
+import type { HandleFn } from '../flow'
+import { getTaskContext } from '../utils'
+
+export const HandleSuccessMiddleware: HandleFn = async (...args: any[]) => {
+  const { next, task } = getTaskContext(...args)
+  await task.done()
+  await task.remove()
+  await task.flow.successFn(task)
+  await next()
+}

packages/orap/src/middlewares/index.ts

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

packages/orap/src/middlewares/private.ts

@@ -0,0 +1,2 @@
+export * from './HandleFailedMiddleware'
+export * from './HandleSuccessMiddleware'