Merge pull request #26 from ora-io/dev

Dev

Author: murongg

package.json

@@ -1,6 +1,6 @@
 {
   "name": "ora-stack",
-  "version": "0.2.4",
+  "version": "0.2.5",
   "private": true,
   "packageManager": "[email protected]",
   "description": "",

packages/orap/README.md

@@ -78,7 +78,7 @@ const store = redisStore()
 const sm = new StoreManager(store)
 
 // example event: erc20 transfer
-const handle1 = (from: string, to: string, amount: number, event: ContractEventPayload, task: TaskRaplized, next: NextFunction) => {
+const handle1 = (from: string, to: string, amount: number, event: ContractEventPayload, next: NextFunction, task: TaskRaplized) => {
   console.log(`handle task 1: from ${from} to ${to} amount ${amount} task ${task}`)
   next()
 }

packages/orap/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@ora-io/orap",
   "type": "module",
-  "version": "0.2.4",
+  "version": "0.2.5",
   "packageManager": "[email protected]",
   "description": "",
   "author": "Norman (nom4dv3), MuRong",

packages/orap/src/flow/event.ts

@@ -1,7 +1,8 @@
 import type { AutoCrossCheckParam, Providers } from '@ora-io/reku'
 import type { Context } from '../task'
+import type { TaskFlowParams } from '../flow/task'
 import { TaskFlow } from '../flow/task'
-import type { StoreManager } from '../store'
+import { StoreManager } from '../store'
 import type { EventSignalRegisterParams } from '../signal'
 import { EventVerse } from '../verse/event'
 import type { TaskVerse } from '../verse/task'
@@ -37,13 +38,19 @@ export class EventFlow implements Flow {
     return this
   }
 
-  // task(store: Store, context?: Context): TaskFlow {
-  task(sm?: StoreManager, context?: Context): TaskFlow {
-    const tf = new TaskFlow(this)
-    if (sm)
-      tf.cache(sm)
-    if (context)
-      tf.context(context)
+  task(params?: TaskFlowParams): TaskFlow
+  task(sm?: StoreManager | TaskFlowParams, context?: Context): TaskFlow {
+    let tf: TaskFlow
+    if (sm instanceof StoreManager) {
+      tf = new TaskFlow(this)
+      if (sm)
+        tf.cache(sm)
+      if (context)
+        tf.context(context)
+    }
+    else {
+      tf = new TaskFlow(this, sm)
+    }
     this._taskFlows.push(tf)
     return tf
   }

packages/orap/src/flow/task.ts

@@ -18,13 +18,26 @@ const defaultHandleFn: HandleFn = () => {
 const defaultToKeyFn: ToKeyFn = _ => randomStr(8, alphabetHex)
 export interface TaskFlowTTL { taskTtl: Milliseconds; doneTtl: Milliseconds }
 
+export interface TaskFlowParams {
+  context?: Context
+  taskPrefix?: Prefix
+  donePrefix?: Prefix
+  taskTtl?: Milliseconds
+  doneTtl?: Milliseconds
+  cache?: StoreManager
+  toKeyFn?: ToKeyFn
+  handleFn?: HandleFn
+  successFn?: HandleResultFn
+  failFn?: HandleResultFn
+}
+
 // TODO: add 'Failed-Task:' ?
 export class TaskFlow implements Flow {
   sm: StoreManager = new StoreManager(memoryStore())
   taskPrefix: Prefix = 'Task:'
   donePrefix: Prefix = 'Done-Task:'
-  taskTtl?: Milliseconds
-  doneTtl?: Milliseconds
+  taskTtl: Milliseconds = 60 * 1000
+  doneTtl: Milliseconds = 60 * 1000
   toKeyFn: ToKeyFn = defaultToKeyFn
   handleFn: HandleFn = defaultHandleFn
   successFn: HandleResultFn = defaultSuccessFn
@@ -39,7 +52,21 @@ export class TaskFlow implements Flow {
 
   constructor(
     private parentFlow: EventFlow,
-  ) { }
+    params?: TaskFlowParams,
+  ) {
+    params?.context && this.context(params?.context)
+    const taskPrefix = params?.taskPrefix ?? this.taskPrefix
+    const donePrefix = params?.donePrefix ?? this.donePrefix
+    this.prefix(taskPrefix, donePrefix)
+    const taskTtl = params?.taskTtl ?? this.taskTtl
+    const doneTtl = params?.doneTtl ?? this.doneTtl
+    this.ttl(taskTtl, doneTtl)
+    params?.cache && this.cache(params?.cache)
+    params?.toKeyFn && this.key(params?.toKeyFn)
+    params?.handleFn && this.handle(params?.handleFn)
+    params?.successFn && this.success(params?.successFn)
+    params?.failFn && this.fail(params?.failFn)
+  }
 
   get middlewares() {
     return this._middlewares

packages/orap/src/task/verse.ts

@@ -1,5 +1,5 @@
 import type { Milliseconds } from '@ora-io/utils'
-import { composeFns, isJsonString, isString, stripPrefix } from '@ora-io/utils'
+import { argParser, composeFns, isJsonString, isString, stripPrefix } from '@ora-io/utils'
 import type { TaskFlow } from '../flow'
 import { HandleSuccessMiddleware } from '../middlewares/HandleSuccessMiddleware'
 import { HandleFailedMiddleware } from '../middlewares/private'
@@ -97,12 +97,13 @@ export class TaskRaplized extends TaskStorable {
    * @returns
    */
   toString() {
-    return this.stringify(this.eventLog)
+    const res = argParser.parse(this.eventLog)
+    return this.stringify(res)
   }
 
   fromString(jsonString: string) {
     if (isJsonString(jsonString))
-      this.eventLog = JSON.parse(jsonString)
+      this.eventLog = argParser.serialize(JSON.parse(jsonString))
 
     return this
   }

packages/reku/README.md

@@ -68,3 +68,139 @@ await acc.start({
   pollingInterval: 3000,
 })
 ```
+
+
+## Provider Manager
+
+`RekuProviderManager` is a TypeScript class designed to manage Ethereum providers and contracts. **It supports both WebSocket and JSON-RPC providers, includes heartbeat functionality to maintain the connection, and provides event listening and management features**.
+
+### Usage
+
+#### Import and Initialization
+
+First, import the `RekuProviderManager` class:
+
+```ts
+import { RekuProviderManager } from '@ora-io/reku'
+```
+
+Then, create an instance of `RekuProviderManager`:
+
+```ts
+const providerManager = new RekuProviderManager('wss://your-ethereum-node-url', {
+  heartbeatInterval: 5000, // Optional, default is 10 seconds
+  disabledHeartbeat: false // Optional, whether to disable heartbeat
+})
+```
+#### Adding Contracts
+
+You can add a contract using its address and ABI:
+
+```ts
+const contractAddress = '0xYourContractAddress'
+const contractAbi = [] // Your contract ABI
+
+providerManager.addContract(contractAddress, contractAbi)
+```
+
+Or add a contract using an `ethers.Contract` instance:
+
+```ts
+const contract = new ethers.Contract(contractAddress, contractAbi, providerManager.provider)
+providerManager.addContract(contractAddress, contract)
+```
+
+#### Event Listening
+
+Adding Event Listeners
+
+```ts
+providerManager.addListener(contractAddress, 'EventName', (event) => {
+  console.log('Event received:', event)
+})
+```
+
+Removing Event Listeners
+
+```ts
+providerManager.removeListener(contractAddress, 'EventName', listener)
+```
+
+Removing All Event Listeners
+
+```ts
+providerManager.removeAllListeners()
+```
+
+#### Event Management
+
+You can listen for `RekuProviderManager` errors and close events:
+
+```ts
+providerManager.on('error', (error) => {
+  console.error('Provider error:', error)
+})
+
+providerManager.on('close', (code, reason) => {
+  console.log(`Provider closed: ${code} - ${reason}`)
+})
+```
+
+#### Reconnecting
+
+You can manually reconnect to the provider:
+
+```ts 
+providerManager.reconnect()
+```
+
+#### Destroying
+
+When you no longer need the `RekuProviderManager`, you can destroy it to free up resources:
+
+```ts
+providerManager.destroy()
+```
+
+### Configuration Options
+
+The `RekuProviderManager` constructor accepts an optional configuration object:
+
+**`heartbeatInterval`**: Heartbeat interval time (in milliseconds), default is 10 seconds.  
+**`disabledHeartbeat`**: Whether to disable the heartbeat, default is false.
+
+
+## Contract Manager
+
+
+The `RekuContractManager` class is designed to manage Ethereum smart contracts using the ethers library. It provides methods to add, remove, and manage event listeners for contract events.
+
+### Example Usage
+
+```ts
+import { ethers } from 'ethers'
+import { RekuContractManager } from '@ora-io/reku'
+
+const provider = new ethers.providers.WebSocketProvider('wss://your-ethereum-node-url')
+const contractAddress = '0xYourContractAddress'
+const contractAbi = [] // Your contract ABI
+
+const manager = new RekuContractManager(contractAddress, contractAbi, provider)
+
+// Adding an event listener
+manager.addListener('EventName', (event) => {
+  console.log('Event received:', event)
+})
+
+// Removing an event listener
+manager.removeListener('EventName', listener)
+
+// Removing all event listeners
+manager.removeAllListeners()
+
+// Retrying all event listeners
+manager.retryAllListeners()
+
+// Retrying a specific event listener
+manager.retryListener('EventName')
+```

packages/reku/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@ora-io/reku",
   "type": "module",
-  "version": "0.2.3",
+  "version": "0.2.4",
   "packageManager": "[email protected]",
   "description": "",
   "author": "Norman (nom4dv3), MuRong",

packages/reku/src/event/crosschecker/autochecker.ts

@@ -58,10 +58,10 @@ export class AutoCrossChecker extends BaseCrossChecker {
 
     this.cache = new CrossCheckerCacheManager(options?.store, { keyPrefix: options?.storeKeyPrefix, ttl: options?.storeTtl })
 
-    let latestblocknum = await timeoutWithRetry(() => this.provider.provider?.getBlockNumber(), 15 * 1000, 3)
+    let latestBlockNum = await timeoutWithRetry(() => this.provider.provider?.getBlockNumber(), 15 * 1000, 3)
 
-    // resume checkpoint priority: options.fromBlock > cache > latestblocknum + 1
-    const defaultInitCheckpoint = await this.cache.getCheckpoint() ?? (latestblocknum)
+    // resume checkpoint priority: options.fromBlock > cache > latestBlockNum + 1
+    const defaultInitCheckpoint = await this.cache.getCheckpoint() ?? (latestBlockNum)
 
     const {
       fromBlock = defaultInitCheckpoint,
@@ -87,10 +87,10 @@ export class AutoCrossChecker extends BaseCrossChecker {
     }
 
     const waitNextCrosscheck = async (): Promise<boolean> => {
-      latestblocknum = await timeoutWithRetry(() => this.provider.provider?.getBlockNumber(), 15 * 1000, 3)
-      if (ccrOptions.toBlock + delayBlockFromLatest > latestblocknum) {
+      latestBlockNum = await timeoutWithRetry(() => this.provider.provider?.getBlockNumber(), 15 * 1000, 3)
+      if (ccrOptions.toBlock + delayBlockFromLatest > latestBlockNum) {
         // sleep until the toBlock
-        // await sleep((ccrOptions.toBlock + delayBlockFromLatest - latestblocknum) * blockInterval)
+        // await sleep((ccrOptions.toBlock + delayBlockFromLatest - latestBlockNum) * blockInterval)
         return false
       }
       return true
@@ -130,7 +130,7 @@ export class AutoCrossChecker extends BaseCrossChecker {
         await updateCCROptions(ccrOptions)
       }
       else {
-        debug('Because the latest block %d is too old, skip this cross check', latestblocknum)
+        debug('Because the latest block %d is too old, skip this cross check', latestBlockNum)
       }
       return endingCondition()
     }, pollingInterval)

packages/reku/src/provider/contract.ts

@@ -2,7 +2,7 @@ import type { ContractAddress, Fn } from '@ora-io/utils'
 import type { Interface, InterfaceAbi } from 'ethers'
 import { ethers } from 'ethers'
 
-export class ContractManager {
+export class RekuContractManager {
   private _contract?: ethers.Contract
   private _listeners: Map<ethers.ContractEventName, Fn> = new Map()