feat: outbound message send via session

packages/core/src/agent/Agent.ts

@@ -1,9 +1,9 @@
-import type { AgentDependencies } from './AgentDependencies'
-import type { AgentModulesInput } from './AgentModules'
-import type { AgentMessageReceivedEvent } from './Events'
 import type { InboundTransport } from '../transport/InboundTransport'
 import type { OutboundTransport } from '../transport/OutboundTransport'
 import type { InitConfig } from '../types'
+import type { AgentDependencies } from './AgentDependencies'
+import type { AgentModulesInput } from './AgentModules'
+import type { AgentMessageReceivedEvent } from './Events'
 import type { Subscription } from 'rxjs'
 
 import { Subject } from 'rxjs'
@@ -141,6 +141,10 @@ export class Agent<AgentModules extends AgentModulesInput = any> extends BaseAge
     this.messageSender.registerOutboundTransport(outboundTransport)
   }
 
+  public async resetOutboundTransport() {
+    await this.messageSender.resetOutboundTransport()
+  }
+
   public get outboundTransports() {
     return this.messageSender.outboundTransports
   }

packages/core/src/agent/Dispatcher.ts

@@ -55,7 +55,6 @@ class Dispatcher {
       outboundMessage = await messageHandler.handle(messageContext)
     } catch (error) {
       const problemReportMessage = error.problemReport
-
       if (problemReportMessage instanceof ProblemReportMessage && messageContext.connection) {
         problemReportMessage.setThread({
           threadId: message.threadId,
@@ -78,6 +77,9 @@ class Dispatcher {
     }
 
     if (outboundMessage) {
+      // Store the sessionId of the inbound message, if there is one, so messages can later be send without
+      // outbound transport.
+      outboundMessage.sessionIdFromInbound = messageContext.sessionId
       if (outboundMessage.isOutboundServiceMessage()) {
         await this.messageSender.sendMessageToService(outboundMessage)
       } else {

packages/core/src/agent/MessageReceiver.ts

@@ -1,10 +1,10 @@
-import type { AgentMessage } from './AgentMessage'
-import type { DecryptedMessageContext } from './EnvelopeService'
-import type { TransportSession } from './TransportService'
-import type { AgentContext } from './context'
 import type { ConnectionRecord } from '../modules/connections'
 import type { InboundTransport } from '../transport'
 import type { EncryptedMessage, PlaintextMessage } from '../types'
+import type { AgentMessage } from './AgentMessage'
+import type { DecryptedMessageContext, EnvelopeKeys } from './EnvelopeService'
+import type { TransportSession } from './TransportService'
+import type { AgentContext } from './context'
 
 import { InjectionSymbols } from '../constants'
 import { AriesFrameworkError } from '../error'
@@ -85,7 +85,7 @@ export class MessageReceiver {
       if (this.isEncryptedMessage(inboundMessage)) {
         await this.receiveEncryptedMessage(agentContext, inboundMessage as EncryptedMessage, session)
       } else if (this.isPlaintextMessage(inboundMessage)) {
-        await this.receivePlaintextMessage(agentContext, inboundMessage, connection)
+        await this.receivePlaintextMessage(agentContext, inboundMessage, connection, session)
       } else {
         throw new AriesFrameworkError('Unable to parse incoming message: unrecognized format')
       }
@@ -98,10 +98,12 @@ export class MessageReceiver {
   private async receivePlaintextMessage(
     agentContext: AgentContext,
     plaintextMessage: PlaintextMessage,
-    connection?: ConnectionRecord
+    connection?: ConnectionRecord,
+    session?: TransportSession
   ) {
     const message = await this.transformAndValidate(agentContext, plaintextMessage)
-    const messageContext = new InboundMessageContext(message, { connection, agentContext })
+    const messageContext = new InboundMessageContext(message, { connection, agentContext, sessionId: session?.id })
+
     await this.dispatcher.dispatch(messageContext)
   }
 
@@ -150,11 +152,24 @@ export class MessageReceiver {
       session.connectionId = connection?.id
       messageContext.sessionId = session.id
       this.transportService.saveSession(session)
+    } else if (senderKey && recipientKey && message.service?.serviceEndpoint && session) {
+      const keys = {
+        recipientKeys: [senderKey],
+        routingKeys: [],
+        senderKey: recipientKey,
+      }
+      session.keys = keys
+      session.inboundMessage = message
+      // We allow unready connections to be attached to the session as we want to be able to
+      // use return routing to make connections. This is especially useful for creating connections
+      // with mediators when you don't have a public endpoint yet.
+      session.connectionId = connection?.id
+      messageContext.sessionId = session.id
+      this.transportService.saveSession(session)
     } else if (session) {
       // No need to wait for session to stay open if we're not actually going to respond to the message.
       await session.close()
     }
-
     await this.dispatcher.dispatch(messageContext)
   }
 

packages/core/src/agent/MessageSender.ts

@@ -43,7 +43,7 @@ export class MessageSender {
   private didResolverService: DidResolverService
   private didCommDocumentService: DidCommDocumentService
   private eventEmitter: EventEmitter
-  public readonly outboundTransports: OutboundTransport[] = []
+  public outboundTransports: OutboundTransport[] = []
 
   public constructor(
     envelopeService: EnvelopeService,
@@ -68,6 +68,15 @@ export class MessageSender {
     this.outboundTransports.push(outboundTransport)
   }
 
+  public async resetOutboundTransport() {
+    // NOTE: we could also make this void instead of an async fct.
+    // Maybe void is a bit dirty?
+    this.outboundTransports.forEach(async (obt) => {
+      await obt.stop()
+    })
+    this.outboundTransports = []
+  }
+
   public async packMessage(
     agentContext: AgentContext,
     {
@@ -185,7 +194,7 @@ export class MessageSender {
       transportPriority?: TransportPriorityOptions
     }
   ) {
-    const { agentContext, connection, outOfBand, sessionId, message } = outboundMessageContext
+    const { agentContext, connection, outOfBand, sessionId, message, sessionIdFromInbound } = outboundMessageContext
     const errors: Error[] = []
 
     if (!connection) {
@@ -203,8 +212,10 @@ export class MessageSender {
 
     let session: TransportSession | undefined
 
-    if (sessionId) {
-      session = this.transportService.findSessionById(sessionId)
+    if (sessionIdFromInbound) {
+      session = this.transportService.findSessionById(sessionIdFromInbound)
+    } else if (sessionId) {
+      this.transportService.findSessionById(sessionId)
     }
     if (!session) {
       // Try to send to already open session
@@ -348,7 +359,7 @@ export class MessageSender {
       this.emitMessageSentEvent(outboundMessageContext, OutboundMessageSendStatus.SentToTransport)
     } catch (error) {
       this.logger.error(
-        `Message is undeliverable to service with id ${outboundMessageContext.serviceParams?.service.id}: ${error.message}`,
+        `Message is undeliverable to service with id ${outboundMessageContext.serviceParams?.service.id}: ${error}`,
         {
           message: outboundMessageContext.message,
           error,
@@ -364,14 +375,14 @@ export class MessageSender {
   }
 
   private async sendToService(outboundMessageContext: OutboundMessageContext) {
-    const { agentContext, message, serviceParams, connection } = outboundMessageContext
+    const { agentContext, message, serviceParams, connection, sessionIdFromInbound, sessionId } = outboundMessageContext
 
     if (!serviceParams) {
       throw new AriesFrameworkError('No service parameters found in outbound message context')
     }
     const { service, senderKey, returnRoute } = serviceParams
 
-    if (this.outboundTransports.length === 0) {
+    if (this.outboundTransports.length === 0 && !sessionIdFromInbound && !sessionId) {
       throw new AriesFrameworkError('Agent has no outbound transport!')
     }
 
@@ -388,7 +399,7 @@ export class MessageSender {
 
     // Set return routing for message if requested
     if (returnRoute) {
-      message.setReturnRouting(ReturnRouteTypes.all)
+      message.setReturnRouting(ReturnRouteTypes.all, message.threadId)
     }
 
     try {
@@ -408,13 +419,61 @@ export class MessageSender {
     const outboundPackage = await this.packMessage(agentContext, { message, keys, endpoint: service.serviceEndpoint })
     outboundPackage.endpoint = service.serviceEndpoint
     outboundPackage.connectionId = connection?.id
-    for (const transport of this.outboundTransports) {
-      const protocolScheme = getProtocolScheme(service.serviceEndpoint)
-      if (!protocolScheme) {
-        this.logger.warn('Service does not have valid protocolScheme.')
-      } else if (transport.supportedSchemes.includes(protocolScheme)) {
-        await transport.sendMessage(outboundPackage)
-        return
+
+    // Outbound transport is present
+    if (this.outboundTransports.length !== 0) {
+      for (const transport of this.outboundTransports) {
+        const protocolScheme = getProtocolScheme(service.serviceEndpoint)
+        if (!protocolScheme) {
+          this.logger.warn('Service does not have valid protocolScheme.')
+        } else if (transport.supportedSchemes.includes(protocolScheme)) {
+          await transport.sendMessage(outboundPackage)
+          return
+        }
+      }
+      // No outbound transport: Try retrieving session-based connection instead
+    } else if (sessionId || sessionIdFromInbound) {
+      const session = sessionIdFromInbound
+        ? this.transportService.findSessionById(sessionIdFromInbound)
+        : sessionId
+        ? this.transportService.findSessionById(sessionId)
+        : undefined
+
+      if (session?.inboundMessage?.hasReturnRouting(message.threadId)) {
+        this.logger.debug(
+          `Found session with return routing for message '${message.id}' (connection '${connection?.id}'`
+        )
+        try {
+          await this.sendMessageToService(outboundMessageContext)
+          this.emitMessageSentEvent(outboundMessageContext, OutboundMessageSendStatus.SentToSession)
+          return
+        } catch (error) {
+          this.logger.debug(`Sending an outbound message via session failed with error: ${error.message}.`, error)
+          throw new MessageSendingError(
+            `Unable to send message to service: ${service.serviceEndpoint} ${JSON.stringify(
+              message.threadId
+            )} ${JSON.stringify(session)}`,
+            {
+              outboundMessageContext,
+            }
+          )
+        }
+      } else if (outboundMessageContext.serviceParams?.service.serviceEndpoint && session) {
+        try {
+          await this.sendMessageToSession(agentContext, session, message)
+          this.emitMessageSentEvent(outboundMessageContext, OutboundMessageSendStatus.SentToSession)
+          return
+        } catch (error) {
+          this.logger.debug(`Sending an outbound message via session failed with error: ${error.message}.`, error)
+          throw new MessageSendingError(
+            `Unable to send message to service: ${
+              outboundMessageContext.serviceParams?.service.serviceEndpoint
+            } with threadId ${JSON.stringify(message.threadId)} `,
+            {
+              outboundMessageContext,
+            }
+          )
+        }
       }
     }
     throw new MessageSendingError(`Unable to send message to service: ${service.serviceEndpoint}`, {

packages/core/src/agent/TransportService.ts

@@ -1,7 +1,7 @@
-import type { AgentMessage } from './AgentMessage'
-import type { EnvelopeKeys } from './EnvelopeService'
 import type { DidDocument } from '../modules/dids'
 import type { EncryptedMessage } from '../types'
+import type { AgentMessage } from './AgentMessage'
+import type { EnvelopeKeys } from './EnvelopeService'
 
 import { DID_COMM_TRANSPORT_QUEUE } from '../constants'
 import { injectable } from '../plugins'

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

@@ -6,6 +6,7 @@ import type { OutOfBandRecord } from '../../modules/oob'
 import type { BaseRecord } from '../../storage/BaseRecord'
 import type { AgentMessage } from '../AgentMessage'
 import type { AgentContext } from '../context'
+import type { InboundMessageContext } from './InboundMessageContext'
 
 import { AriesFrameworkError } from '../../error'
 
@@ -17,6 +18,7 @@ export interface ServiceMessageParams {
 
 export interface OutboundMessageContextParams {
   agentContext: AgentContext
+  inboundMessageContext?: InboundMessageContext
   associatedRecord?: BaseRecord<any, any, any>
   connection?: ConnectionRecord
   serviceParams?: ServiceMessageParams
@@ -26,15 +28,17 @@ export interface OutboundMessageContextParams {
 
 export class OutboundMessageContext<T extends AgentMessage = AgentMessage> {
   public message: T
+  public sessionIdFromInbound?: string | undefined
   public connection?: ConnectionRecord
   public serviceParams?: ServiceMessageParams
   public outOfBand?: OutOfBandRecord
   public associatedRecord?: BaseRecord<any, any, any>
   public sessionId?: string
   public readonly agentContext: AgentContext
 
-  public constructor(message: T, context: OutboundMessageContextParams) {
+  public constructor(message: T, context: OutboundMessageContextParams, inboundContext?: InboundMessageContext) {
     this.message = message
+    this.sessionIdFromInbound = inboundContext?.sessionId ? inboundContext.sessionId : undefined
     this.connection = context.connection
     this.sessionId = context.sessionId
     this.outOfBand = context.outOfBand
@@ -67,6 +71,7 @@ export class OutboundMessageContext<T extends AgentMessage = AgentMessage> {
     return {
       message: this.message,
       outOfBand: this.outOfBand,
+      sessionIdFromInbound: this.sessionIdFromInbound,
       associatedRecord: this.associatedRecord,
       sessionId: this.sessionId,
       serviceParams: this.serviceParams,

packages/core/src/modules/proofs/ProofsApi.ts

@@ -1,3 +1,5 @@
+import type { AgentMessage } from '../../agent/AgentMessage'
+import type { Query } from '../../storage/StorageService'
 import type {
   AcceptProofOptions,
   AcceptProofProposalOptions,
@@ -21,8 +23,6 @@ import type {
 import type { ProofProtocol } from './protocol/ProofProtocol'
 import type { ProofFormatsFromProtocols } from './protocol/ProofProtocolOptions'
 import type { ProofExchangeRecord } from './repository/ProofExchangeRecord'
-import type { AgentMessage } from '../../agent/AgentMessage'
-import type { Query } from '../../storage/StorageService'
 
 import { injectable } from 'tsyringe'
 

packages/core/src/modules/proofs/protocol/BaseProofProtocol.ts

@@ -1,3 +1,13 @@
+import type { AgentMessage } from '../../../agent/AgentMessage'
+import type { FeatureRegistry } from '../../../agent/FeatureRegistry'
+import type { AgentContext } from '../../../agent/context/AgentContext'
+import type { InboundMessageContext } from '../../../agent/models/InboundMessageContext'
+import type { DependencyManager } from '../../../plugins'
+import type { Query } from '../../../storage/StorageService'
+import type { ProblemReportMessage } from '../../problem-reports'
+import type { ProofStateChangedEvent } from '../ProofEvents'
+import type { ExtractProofFormats, ProofFormatService } from '../formats'
+import type { ProofExchangeRecord } from '../repository'
 import type { ProofProtocol } from './ProofProtocol'
 import type {
   CreateProofProposalOptions,
@@ -16,16 +26,6 @@ import type {
   SelectCredentialsForRequestOptions,
   SelectCredentialsForRequestReturn,
 } from './ProofProtocolOptions'
-import type { AgentMessage } from '../../../agent/AgentMessage'
-import type { FeatureRegistry } from '../../../agent/FeatureRegistry'
-import type { AgentContext } from '../../../agent/context/AgentContext'
-import type { InboundMessageContext } from '../../../agent/models/InboundMessageContext'
-import type { DependencyManager } from '../../../plugins'
-import type { Query } from '../../../storage/StorageService'
-import type { ProblemReportMessage } from '../../problem-reports'
-import type { ProofStateChangedEvent } from '../ProofEvents'
-import type { ExtractProofFormats, ProofFormatService } from '../formats'
-import type { ProofExchangeRecord } from '../repository'
 
 import { EventEmitter } from '../../../agent/EventEmitter'
 import { DidCommMessageRepository } from '../../../storage'