Enhancement/event based form integration

agent-manager/package.json

@@ -24,7 +24,7 @@
         "express": "^4.19.2",
         "kafkajs": "^2.2.4",
         "kuber-client": "^3.0.3",
-        "libcardano": "1.4.3",
+        "libcardano": "1.4.11",
         "luxon": "^3.4.4",
         "prisma": "^5.13.0",
         "swagger-jsdoc": "^6.2.8",

agent-manager/src/controller/health.ts

@@ -0,0 +1,16 @@
+import { Router, Request, Response } from 'express'
+import { handlerWrapper } from '../utils/asyncWrapper'
+import { checkKafkaStatus } from '../service/healthCheck/kafka'
+
+const router = Router()
+
+function healthCheck(req: Request, res: Response) {
+    checkKafkaStatus()
+        .then((a) => console.log('Success : ', a))
+        .catch((err) => console.log('Error: ', err))
+    return res.status(200).send({ Msg: 'OK ' })
+}
+
+router.get('/', handlerWrapper(healthCheck))
+
+export default router

agent-manager/src/index.ts

@@ -6,11 +6,14 @@ import { createBlockchainInstance } from './service/Listeners/BlockchainService'
 import { ManagerWalletService } from './service/Manager/ManagerWallet'
 import { TxListener } from './service/Listeners/TxListener'
 import { parseRawBlockBody } from 'libcardano/cardano/ledger-serialization/transaction'
-import environments from "./config/environments";
+import environments from './config/environments'
+import healthCheck from './controller/health'
 
 const app = express()
 const port = environments.serverPort
 
+app.use('/health', healthCheck)
+
 const server = app.listen(port, async () => {
     console.log(`Server is running on http://localhost:${port}`)
 
@@ -35,6 +38,7 @@ const server = app.listen(port, async () => {
     })
     await initKafkaConsumers(manager)
 })
+
 server.on('error', (e) => {
     console.error('Server error:', e)
     process.exit(1)

agent-manager/src/service/Listeners/KafkaMessageConsumer.ts

@@ -7,14 +7,14 @@ import environments from '../../config/environments'
 const config = environments.kafka
 const configTopic = `${config.topicPrefix || config.prefix || 'agent'}-updates`
 const triggerTopic = `${config.topicPrefix || config.prefix || 'agent'}-triggers`
-const topicList = [configTopic, triggerTopic]
+export const topicList = [configTopic, triggerTopic]
 const brokers = config.brokers
     .split(',')
     .map((x) => x.trim())
     .filter((x) => x && x.length > 0)
 const groupId = config.consumerGroup || `${config.prefix || 'agent'}-manager`
 
-const kafka = new Kafka({
+export const kafka = new Kafka({
     clientId: config.clientId ? config.clientId : `${config.prefix || 'agent'}-manager`,
     brokers, // Update with your Kafka broker address
 })

agent-manager/src/service/healthCheck/kafka.ts

@@ -0,0 +1,34 @@
+import { kafka, topicList } from '../Listeners/KafkaMessageConsumer'
+
+export async function checkKafkaStatus() {
+    const admin = kafka.admin()
+    try {
+        // Connect to Kafka
+        await admin.connect()
+
+        // Fetch metadata
+        const metadata = await admin.fetchTopicMetadata({ topics: [] })
+        console.log('Metadata:', metadata)
+
+        const clusterInfo = await admin.describeCluster()
+        console.log('ClusterInfo : ', clusterInfo)
+
+        // Optionally check for specific topics
+        const missingTopics = topicList.filter((topic) => !metadata.topics.find((t: any) => t.name === topic))
+
+        if (missingTopics.length > 0) {
+            console.error('Missing topics:', missingTopics)
+            return { status: 'error', message: `Missing topics: ${missingTopics.join(', ')}` }
+        }
+
+        return { status: 'ok', message: 'Kafka is in sync and working.' }
+    } catch (error: any) {
+        console.error('Error connecting to Kafka:', error)
+        return { status: 'error', message: `Error: ${error.message}` }
+    } finally {
+        await admin.disconnect()
+    }
+}
+
+// Example usage
+checkKafkaStatus().then((status) => console.log('Kafka Status:', status))

agent-manager/src/utils/asyncWrapper.ts

@@ -0,0 +1,15 @@
+import {NextFunction, Request, Response} from "express";
+
+export const handlerWrapper = (fn: any) => (req: Request, res: Response, next: NextFunction) => {
+    try {
+        const result = fn(req, res, next)
+        if (result instanceof Promise) {
+            result.catch((e) => {
+                next(e)
+            })
+        }
+    } catch (error) {
+        // Handle sync errors
+        next(error)
+    }
+}

agent-manager/yarn.lock

@@ -2684,10 +2684,10 @@ levn@^0.4.1:
     prelude-ls "^1.2.1"
     type-check "~0.4.0"
 
-[email protected].3:
-  version "1.4.3"
-  resolved "https://registry.yarnpkg.com/libcardano/-/libcardano-1.4.3.tgz#a3dc20452e1520b65fa32621337de62413a3cd5c"
-  integrity sha512-5R5aE+WHwC1wxFLH4fVjXeQh0WbOLCBZKcFmbVYvGwWy2TMaqspncy8QMclqD03x8GtdK7yWEFTp9buGP9/eTw==
+[email protected].11:
+  version "1.4.11"
+  resolved "https://registry.yarnpkg.com/libcardano/-/libcardano-1.4.11.tgz#ffe0f1f63a4a9bb940e87a5b5b68284cbbbaa1a7"
+  integrity sha512-rl/okJH27NBGwOx7VTgBkOJe4q6vBbBE4FbPrXXDhgLgI55004K0ODK0R4+3HCHAL8TtnXPqPRJXE5vvO++lIQ==
   dependencies:
     "@cardano-sdk/crypto" "^0.1.30"
     "@emurgo/cardano-serialization-lib-nodejs" "^11.5.0"

agent-node/package.json

@@ -23,7 +23,7 @@
         "bech32": "^2.0.0",
         "dotenv": "^16.4.5",
         "kuber-client": "^2.0.0",
-        "libcardano": "1.4.2",
+        "libcardano": "1.4.11",
         "luxon": "^3.4.4",
         "node-cron": "^3.0.3",
         "ws": "^8.18.0"

agent-node/src/constants/global.ts

@@ -1,13 +1,10 @@
-import { EventTriggerTypeDetails } from '../types/types'
+import { IEventBasedAction } from '../types/eventTriger'
 
 export const globalState: {
-    eventTriggerTypeDetails: EventTriggerTypeDetails
+    eventTypeDetails: IEventBasedAction[]
     agentName: string
 } = {
-    eventTriggerTypeDetails: {
-        eventType: false,
-        function_name: '',
-    },
+    eventTypeDetails: [],
     agentName: '',
 }
 

agent-node/src/executor/AgentFunctions.ts

@@ -7,6 +7,7 @@ export interface FunctionHolder {
 export interface FunctionGroup {
     functions: FunctionHolder
     builtins: FunctionHolder
+    filters: FunctionHolder
 }
 // Helper function to require modules with extensions automatically
 function requireModule(filePath: string) {
@@ -27,6 +28,7 @@ function requireModule(filePath: string) {
 function loadHandlersSync(directory: string): FunctionGroup {
     const handlers: FunctionHolder = {}
     const builtins: FunctionHolder = {}
+    const filters: FunctionHolder = {}
 
     // Get the list of files in the directory
     const files = fs.readdirSync(directory)
@@ -44,12 +46,16 @@ function loadHandlersSync(directory: string): FunctionGroup {
                 const module: any = requireModule(filePath)
                 // Check if the module exports a `handler` function
                 const handler = module.handler || module.default
+                const filter = module.filter
                 if (typeof handler === 'function') {
                     handlers[baseFileName] = handler
                 }
                 if (module.builtin || handler.name == 'builtin') {
                     builtins[baseFileName] = module.builtin || module.default
                 }
+                if (filter && typeof filter == "function"){
+                    filters[baseFileName] = filter
+                }
             } catch (error) {
                 console.error(
                     `Failed to load handler from file ${filePath}:`,
@@ -62,6 +68,7 @@ function loadHandlersSync(directory: string): FunctionGroup {
     return {
         functions: handlers,
         builtins: builtins,
+        filters: filters
     }
 }
 

agent-node/src/executor/AgentRunner.ts

@@ -7,10 +7,12 @@ import { loadRootKeyFromBuffer } from '../utils/cardano'
 import { HdWallet } from 'libcardano'
 import { AgentWalletDetails } from '../types/types'
 import { globalState } from '../constants/global'
+import { EventContext } from './BaseFunction'
 
 export class AgentRunner {
     executor: Executor
     managerInterface: ManagerInterface
+
     constructor(managerInterface: ManagerInterface, txListener: TxListener) {
         this.managerInterface = managerInterface
         this.executor = new Executor(null, managerInterface, txListener)
@@ -27,6 +29,36 @@ export class AgentRunner {
         })
     }
 
+    async invokeFunctionWithEventContext(
+        eventFilterContext: any,
+        context: EventContext,
+        triggerType: TriggerType,
+        instanceIndex: number,
+        method: string,
+        ...args: any[]
+    ) {
+        const eventContext = {
+            event: context,
+            filter: eventFilterContext,
+        }
+        const params = await this.executor
+            .filterFunctionParams(method, eventContext)
+            .catch((err) => console.error('Function Invocation Error: ', err))
+
+        if (params) {
+            this.executor
+                .invokeFunctionWithContext(eventContext, method, ...params)
+                .then((result) => {
+                    saveTxLog(
+                        result,
+                        this.managerInterface,
+                        triggerType,
+                        instanceIndex
+                    )
+                })
+        }
+    }
+
     async remakeContext(index: number) {
         const rootKey = loadRootKeyFromBuffer()
         const hdWallet = HdWallet.fromHdKey(rootKey)

agent-node/src/executor/BaseFunction.ts

@@ -1,3 +1,6 @@
+import {Transaction} from "libcardano/cardano/ledger-serialization/transaction";
+import {DecodedBlock} from "./TxListener";
+
 export interface Key {
     private: string
     public: string
@@ -75,12 +78,20 @@ export interface Helpers {
     generateVoteMetadataContent: () => string
 }
 
+export interface EventContext{
+    tx: Transaction;
+    block: DecodedBlock;
+    confirmation: number
+}
+
 export interface FunctionContext {
     wallet: Wallet
     kuber: KuberApi
     builtins: Builtins
     agentName: string
-    helpers: Helpers
+    helpers: Helpers,
+    event ?: EventContext,
+    filter?: any
 }
 
 // Create a restricted execution environment

agent-node/src/executor/Executor.ts

@@ -14,12 +14,14 @@ export interface CallLog {
     return?: any
     error?: Error | string
 }
+
 export class Executor {
     wallet: any
     readonly rpcInterface: ManagerInterface
     readonly functions: FunctionGroup
     readonly functionContext: FunctionContext
     txListener: TxListener
+
     constructor(
         wallet: any,
         rpcInterface: ManagerInterface,
@@ -122,6 +124,7 @@ export class Executor {
                         processQueue(this.rpcInterface, this.txListener)
                     }
                 })
+
                 async function processQueue(
                     rpcInterface: ManagerInterface,
                     txListener: TxListener
@@ -285,7 +288,29 @@ export class Executor {
         } as Builtins
     }
 
-    invokeFunction(name: string, ...args: any): Promise<CallLog[]> {
+    async invokeFunction(name: string, ...args: any): Promise<CallLog[]> {
+        return await this.invokeFunctionWithContext({}, name, ...args)
+    }
+
+    filterFunctionParams(name: string, context: any): Promise<any> {
+        const f: any | undefined = this.functions.filters[name]
+        if (f === undefined) {
+            return Promise.reject(new Error('Filter not defined'))
+        }
+        const newContext = { ...this.functionContext, ...context }
+        try {
+            const result: any = f(newContext)
+            return Promise.resolve(result)
+        } catch (err) {
+            return Promise.reject(err)
+        }
+    }
+
+    invokeFunctionWithContext(
+        context: any,
+        name: string,
+        ...args: any
+    ): Promise<CallLog[]> {
         const f: any | undefined = this.functions.functions[name]
         const log: CallLog = {
             function: name,
@@ -296,7 +321,7 @@ export class Executor {
             return Promise.resolve([log])
         }
 
-        const newContext = { ...this.functionContext }
+        const newContext = { ...this.functionContext, ...context }
         const builtinsProxy = this.makeProxy(newContext.builtins)
         newContext.builtins = builtinsProxy.proxy
         builtinsProxy.callLog.push(log)
@@ -327,5 +352,6 @@ export class Executor {
         }
         return Promise.resolve(builtinsProxy.callLog)
     }
+
     // newBlock(block) {}
 }

agent-node/src/functions/transferADA.ts

@@ -19,3 +19,5 @@ export default async function handler(
             throw err
         })
 }
+
+

agent-node/src/functions/voteOnProposal.ts

@@ -21,6 +21,7 @@ export default async function handler(
             anchor: anchorData,
         },
     }
+
     return await context.wallet
         .buildAndSubmit(req, true)
         .then((v) => v)
@@ -38,3 +39,20 @@ export default async function handler(
             }
         })
 }
+
+export function filter(context: FunctionContext) {
+    if (!context.filter || !context.event) return null
+    const tx = context.event.tx
+    try {
+        const filters = context.filter['proposalProcedures']
+        const matchedFilterIndexes = filters.map((i: any) => !!i.matchedIndex)
+        return matchedFilterIndexes.length
+            ? matchedFilterIndexes.map((p: any, index: number) => ({
+                  name: 'proposalId',
+                  value: `${tx.hash.toString('hex')}#${index}`,
+              }))
+            : null
+    } catch (err) {
+        return null
+    }
+}