fix: AgentMessageProcessedEvent payload

Signed-off-by: Sai Ranjit Tummalapalli <sairanjit.tummalapalli@ayanworks.com>
openwallet-foundation · Sep 18, 2024 · 93eaf72 · 93eaf72
1 parent 696afdf
commit 93eaf72
Show file tree

Hide file tree

Showing 5 changed files with 14 additions and 8 deletions.
diff --git a/packages/core/src/agent/Dispatcher.ts b/packages/core/src/agent/Dispatcher.ts
@@ -8,6 +8,7 @@ import { CredoError } from '../error'
 import { Logger } from '../logger'
 import { ProblemReportError, ProblemReportReason } from '../modules/problem-reports'
 import { injectable, inject } from '../plugins'
+import { EncryptedMessage } from '../types'
 import { canHandleMessageType, parseMessageType } from '../utils/messageType'
 
 import { ProblemReportMessage } from './../modules/problem-reports/messages/ProblemReportMessage'
@@ -64,7 +65,7 @@ class Dispatcher {
     await next()
   }
 
-  public async dispatch(messageContext: InboundMessageContext): Promise<void> {
+  public async dispatch(messageContext: InboundMessageContext, encryptedMessage?: EncryptedMessage): Promise<void> {
     const { agentContext, connection, senderKey, recipientKey, message } = messageContext
 
     // Set default handler if available, middleware can still override the message handler
@@ -138,6 +139,7 @@ class Dispatcher {
         message,
         connection,
         receivedAt: messageContext.receivedAt,
+        encryptedMessage,
       },
     })
   }

diff --git a/packages/core/src/agent/Events.ts b/packages/core/src/agent/Events.ts
@@ -2,6 +2,7 @@ import type { AgentMessage } from './AgentMessage'
 import type { TransportSession } from './TransportService'
 import type { OutboundMessageContext, OutboundMessageSendStatus } from './models'
 import type { ConnectionRecord } from '../modules/connections'
+import type { EncryptedMessage } from '../types'
 import type { Observable } from 'rxjs'
 
 import { filter } from 'rxjs'
@@ -45,6 +46,7 @@ export interface AgentMessageProcessedEvent extends BaseEvent {
     message: AgentMessage
     connection?: ConnectionRecord
     receivedAt?: Date
+    encryptedMessage?: EncryptedMessage
   }
 }
 

diff --git a/packages/core/src/agent/MessageReceiver.ts b/packages/core/src/agent/MessageReceiver.ts
@@ -174,7 +174,7 @@ export class MessageReceiver {
       await session.close()
     }
 
-    await this.dispatcher.dispatch(messageContext)
+    await this.dispatcher.dispatch(messageContext, encryptedMessage)
   }
 
   /**

diff --git a/packages/node/src/transport/HttpInboundTransport.ts b/packages/node/src/transport/HttpInboundTransport.ts
@@ -10,9 +10,9 @@ import type {
 import type { Express, Request, Response } from 'express'
 import type { Server } from 'http'
 
-import { DidCommMimeType, CredoError, TransportService, utils, AgentEventTypes } from '@credo-ts/core'
+import { DidCommMimeType, CredoError, TransportService, utils, AgentEventTypes, deepEquality } from '@credo-ts/core'
 import express, { text } from 'express'
-import { first, firstValueFrom, ReplaySubject, timeout } from 'rxjs'
+import { filter, firstValueFrom, ReplaySubject, timeout } from 'rxjs'
 
 const supportedContentTypes: string[] = [DidCommMimeType.V0, DidCommMimeType.V1]
 
@@ -66,7 +66,8 @@ export class HttpInboundTransport implements InboundTransport {
 
         observable
           .pipe(
-            first((e) => e.type === AgentEventTypes.AgentMessageProcessed),
+            filter((e) => e.type === AgentEventTypes.AgentMessageProcessed),
+            filter((e) => deepEquality(e.payload.encryptedMessage, encryptedMessage)),
             timeout({
               first: 10000, // timeout after 10 seconds
               meta: 'HttpInboundTransport.start',

diff --git a/packages/node/src/transport/WsInboundTransport.ts b/packages/node/src/transport/WsInboundTransport.ts
@@ -9,8 +9,8 @@ import type {
   AgentMessageProcessedEvent,
 } from '@credo-ts/core'
 
-import { CredoError, TransportService, utils, AgentEventTypes } from '@credo-ts/core'
-import { first, firstValueFrom, ReplaySubject, timeout } from 'rxjs'
+import { CredoError, TransportService, utils, AgentEventTypes, deepEquality } from '@credo-ts/core'
+import { filter, firstValueFrom, ReplaySubject, timeout } from 'rxjs'
 // eslint-disable-next-line import/no-named-as-default
 import WebSocket, { Server } from 'ws'
 
@@ -79,7 +79,8 @@ export class WsInboundTransport implements InboundTransport {
 
         observable
           .pipe(
-            first((e) => e.type === AgentEventTypes.AgentMessageProcessed),
+            filter((e) => e.type === AgentEventTypes.AgentMessageProcessed),
+            filter((e) => deepEquality(e.payload.encryptedMessage, encryptedMessage)),
             timeout({
               first: 10000, // timeout after 10 seconds
               meta: 'WsInboundTransport.listenOnWebSocketMessages',