feat: queued messages reception time (#1824)

Signed-off-by: Ariel Gentile <gentilester@gmail.com>

packages/core/src/agent/Agent.ts

@@ -157,6 +157,7 @@ export class Agent<AgentModules extends AgentModulesInput = any> extends BaseAge
             .receiveMessage(e.payload.message, {
               connection: e.payload.connection,
               contextCorrelationId: e.payload.contextCorrelationId,
+              receivedAt: e.payload.receivedAt,
             })
             .catch((error) => {
               this.logger.error('Failed to process message', { error })

packages/core/src/agent/Dispatcher.ts

@@ -97,6 +97,7 @@ class Dispatcher {
       payload: {
         message,
         connection,
+        receivedAt: messageContext.receivedAt,
       },
     })
   }

packages/core/src/agent/Events.ts

@@ -33,6 +33,7 @@ export interface AgentMessageReceivedEvent extends BaseEvent {
     message: unknown
     connection?: ConnectionRecord
     contextCorrelationId?: string
+    receivedAt?: Date
   }
 }
 
@@ -41,6 +42,7 @@ export interface AgentMessageProcessedEvent extends BaseEvent {
   payload: {
     message: AgentMessage
     connection?: ConnectionRecord
+    receivedAt?: Date
   }
 }
 

packages/core/src/agent/MessageReceiver.ts

@@ -82,7 +82,13 @@ export class MessageReceiver {
       session,
       connection,
       contextCorrelationId,
-    }: { session?: TransportSession; connection?: ConnectionRecord; contextCorrelationId?: string } = {}
+      receivedAt,
+    }: {
+      session?: TransportSession
+      connection?: ConnectionRecord
+      contextCorrelationId?: string
+      receivedAt?: Date
+    } = {}
   ) {
     this.logger.debug(`Agent received message`)
 
@@ -93,9 +99,9 @@ export class MessageReceiver {
 
     try {
       if (this.isEncryptedMessage(inboundMessage)) {
-        await this.receiveEncryptedMessage(agentContext, inboundMessage as EncryptedMessage, session)
+        await this.receiveEncryptedMessage(agentContext, inboundMessage as EncryptedMessage, session, receivedAt)
       } else if (this.isPlaintextMessage(inboundMessage)) {
-        await this.receivePlaintextMessage(agentContext, inboundMessage, connection)
+        await this.receivePlaintextMessage(agentContext, inboundMessage, connection, receivedAt)
       } else {
         throw new CredoError('Unable to parse incoming message: unrecognized format')
       }
@@ -108,17 +114,19 @@ export class MessageReceiver {
   private async receivePlaintextMessage(
     agentContext: AgentContext,
     plaintextMessage: PlaintextMessage,
-    connection?: ConnectionRecord
+    connection?: ConnectionRecord,
+    receivedAt?: Date
   ) {
     const message = await this.transformAndValidate(agentContext, plaintextMessage)
-    const messageContext = new InboundMessageContext(message, { connection, agentContext })
+    const messageContext = new InboundMessageContext(message, { connection, agentContext, receivedAt })
     await this.dispatcher.dispatch(messageContext)
   }
 
   private async receiveEncryptedMessage(
     agentContext: AgentContext,
     encryptedMessage: EncryptedMessage,
-    session?: TransportSession
+    session?: TransportSession,
+    receivedAt?: Date
   ) {
     const decryptedMessage = await this.decryptMessage(agentContext, encryptedMessage)
     const { plaintextMessage, senderKey, recipientKey } = decryptedMessage
@@ -140,6 +148,7 @@ export class MessageReceiver {
       senderKey,
       recipientKey,
       agentContext,
+      receivedAt,
     })
 
     // We want to save a session if there is a chance of returning outbound message via inbound transport.

packages/core/src/agent/models/InboundMessageContext.ts

@@ -11,6 +11,7 @@ export interface MessageContextParams {
   senderKey?: Key
   recipientKey?: Key
   agentContext: AgentContext
+  receivedAt?: Date
 }
 
 export class InboundMessageContext<T extends AgentMessage = AgentMessage> {
@@ -19,6 +20,7 @@ export class InboundMessageContext<T extends AgentMessage = AgentMessage> {
   public sessionId?: string
   public senderKey?: Key
   public recipientKey?: Key
+  public receivedAt: Date
   public readonly agentContext: AgentContext
 
   public constructor(message: T, context: MessageContextParams) {
@@ -28,6 +30,7 @@ export class InboundMessageContext<T extends AgentMessage = AgentMessage> {
     this.connection = context.connection
     this.sessionId = context.sessionId
     this.agentContext = context.agentContext
+    this.receivedAt = context.receivedAt ?? new Date()
   }
 
   /**

packages/core/src/modules/message-pickup/protocol/v2/V2MessagePickupProtocol.ts

@@ -118,6 +118,7 @@ export class V2MessagePickupProtocol extends BaseMessagePickupProtocol {
       (msg) =>
         new Attachment({
           id: msg.id,
+          lastmodTime: msg.receivedAt,
           data: {
             json: msg.encryptedMessage,
           },
@@ -190,6 +191,7 @@ export class V2MessagePickupProtocol extends BaseMessagePickupProtocol {
       (msg) =>
         new Attachment({
           id: msg.id,
+          lastmodTime: msg.receivedAt,
           data: {
             json: msg.encryptedMessage,
           },
@@ -323,6 +325,7 @@ export class V2MessagePickupProtocol extends BaseMessagePickupProtocol {
         payload: {
           message: attachment.getDataAsJson<EncryptedMessage>(),
           contextCorrelationId: messageContext.agentContext.contextCorrelationId,
+          receivedAt: attachment.lastmodTime,
         },
       })
     }

packages/core/src/modules/message-pickup/storage/QueuedMessage.ts

@@ -1,6 +1,13 @@
 import type { EncryptedMessage } from '../../../types'
 
+/**
+ * Basic representation of an encrypted message in a Message Pickup Queue
+ * - id: Message Pickup repository's specific queued message id (unrelated to DIDComm message id)
+ * - receivedAt: reception time (i.e. time when the message has been added to the queue)
+ * - encryptedMessage: packed message
+ */
 export type QueuedMessage = {
   id: string
+  receivedAt?: Date
   encryptedMessage: EncryptedMessage
 }