Merge pull request #87 from nevermined-io/fix/task-log-socket

Fix/task log socket

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@nevermined-io/payments",
-  "version": "0.7.3",
+  "version": "0.7.4",
   "description": "Typescript SDK to interact with the Nevermined Payments Protocol",
   "main": "./dist/index.js",
   "types": "./dist/index.d.ts",

src/api/nvm-backend.ts

@@ -214,7 +214,7 @@ export class NVMBackendApi {
       })
 
       await this.socketClient.emit('_join-tasks', JSON.stringify({ tasks, history }))
-      await this.socketClient.on('task-log', this.handleTaskLog.bind(this))
+      await this.socketClient.on('task-log', this.handleTaskLog.bind(this, tasks))
     } catch (error) {
       throw new PaymentsError(
         `Unable to initialize websocket client: ${this.opts.webSocketHost} - ${(error as Error).message}`,
@@ -223,15 +223,22 @@ export class NVMBackendApi {
   }
 
   /**
-   * Handles the 'task-log' event from the websocket.
+   * Handles the '_task-log' event from the websocket.
    * Parses the incoming data, retrieves the corresponding callback,
    * executes it, and removes the callback if the task is completed or failed.
    *
+   * @param boundTasks - The list of task IDs that the callback is bound to.
    * @param data - The data received from the websocket event.
    */
-  private handleTaskLog(data: any): void {
+  private handleTaskLog(boundTasks: string[], data: any): void {
     const parsedData = JSON.parse(data)
     const { task_id: taskId } = parsedData
+
+    // If the task ID is not in the list of bound tasks, ignore the event
+    if (!boundTasks.includes(taskId)) {
+      return
+    }
+
     const callback = this.taskCallbacks.get(taskId)
     if (callback) {
       // Execute the stored callback

tests/e2e/payments.e2e.test.ts

@@ -38,7 +38,6 @@ describe('Payments API (e2e)', () => {
   let completedTaskDID: string
   let failedTaskId: string
   let failedTaskDID: string
-  let logsReceived = 0
 
   describe('Payments Setup', () => {
     it('The Payments client can be initialized correctly', () => {
@@ -238,7 +237,6 @@ describe('Payments API (e2e)', () => {
           subscriberQueryOpts,
           async (data) => {
             console.log('Task Log received', data)
-            logsReceived++
           }
         )
 
@@ -353,17 +351,38 @@ describe('Payments API (e2e)', () => {
     )
 
     it('Subscriber should be able to receive logs', async () => {
-
+
+      const aiTask: CreateTaskDto = {
+        query: 'https://www.youtube.com/watch?v=0tZFQs7qBfQ',
+        name: 'transcribe',
+        additional_params: [],
+        artifacts: [],
+      }
+      const accessConfig = await paymentsSubscriber.getServiceAccessConfig(agentDID)
+      const queryOpts = {
+        accessToken: accessConfig.accessToken,
+        proxyHost: accessConfig.neverminedProxyUri,
+      }
+
+      let logsReceived = 0
+      const taskResult = await paymentsSubscriber.query.createTask(agentDID, aiTask, queryOpts, async (data) => {
+        console.log('New Task Log received', data)
+        logsReceived++
+      })
+      const taskId = taskResult.data.task.task_id
       const logMessage: TaskLogMessage = {
         level: 'info',
         task_status: AgentExecutionStatus.Completed,
-        task_id: createdTaskId,
+        task_id: taskId,
         message: 'This is a log message',
       }
+
       console.log(`Sending log message for task ${logMessage.task_id}`)
+
       await paymentsBuilder.query.logTask(logMessage)
 
-      await sleep(2_000)
+      await sleep(5_000)
+      console.log(`# Logs received: ${logsReceived}`)
       expect(logsReceived).toBeGreaterThan(0)
 
     }, TEST_TIMEOUT)