feat: improve sending error handling (#1045)

Signed-off-by: Ariel Gentile <[email protected]>

packages/core/src/agent/MessageSender.ts

@@ -10,7 +10,7 @@ import type { TransportSession } from './TransportService'
 
 import { DID_COMM_TRANSPORT_QUEUE, InjectionSymbols } from '../constants'
 import { ReturnRouteTypes } from '../decorators/transport/TransportDecorator'
-import { AriesFrameworkError } from '../error'
+import { AriesFrameworkError, MessageSendingError } from '../error'
 import { Logger } from '../logger'
 import { DidCommDocumentService } from '../modules/didcomm'
 import { getKeyDidMappingByVerificationMethod } from '../modules/dids/domain/key-type'
@@ -209,8 +209,9 @@ export class MessageSender {
 
     if (!connection.did) {
       this.logger.error(`Unable to send message using connection '${connection.id}' that doesn't have a did`)
-      throw new AriesFrameworkError(
-        `Unable to send message using connection '${connection.id}' that doesn't have a did`
+      throw new MessageSendingError(
+        `Unable to send message using connection '${connection.id}' that doesn't have a did`,
+        { outboundMessage }
       )
     }
 
@@ -277,7 +278,10 @@ export class MessageSender {
       errors,
       connection,
     })
-    throw new AriesFrameworkError(`Message is undeliverable to connection ${connection.id} (${connection.theirLabel})`)
+    throw new MessageSendingError(
+      `Message is undeliverable to connection ${connection.id} (${connection.theirLabel})`,
+      { outboundMessage }
+    )
   }
 
   public async sendMessageToService({

packages/core/src/error/MessageSendingError.ts

@@ -0,0 +1,11 @@
+import type { OutboundMessage } from '../types'
+
+import { AriesFrameworkError } from './AriesFrameworkError'
+
+export class MessageSendingError extends AriesFrameworkError {
+  public outboundMessage: OutboundMessage
+  public constructor(message: string, { outboundMessage, cause }: { outboundMessage: OutboundMessage; cause?: Error }) {
+    super(message, { cause })
+    this.outboundMessage = outboundMessage
+  }
+}

packages/core/src/error/index.ts

@@ -3,3 +3,4 @@ export * from './RecordNotFoundError'
 export * from './RecordDuplicateError'
 export * from './IndySdkError'
 export * from './ClassValidationError'
+export * from './MessageSendingError'

packages/core/src/modules/basic-messages/BasicMessagesModule.ts

@@ -30,18 +30,61 @@ export class BasicMessagesModule {
     this.registerHandlers(dispatcher)
   }
 
+  /**
+   * Send a message to an active connection
+   *
+   * @param connectionId Connection Id
+   * @param message Message contents
+   * @throws {RecordNotFoundError} If connection is not found
+   * @throws {MessageSendingError} If message is undeliverable
+   * @returns the created record
+   */
   public async sendMessage(connectionId: string, message: string) {
     const connection = await this.connectionService.getById(connectionId)
 
-    const basicMessage = await this.basicMessageService.createMessage(message, connection)
+    const { message: basicMessage, record: basicMessageRecord } = await this.basicMessageService.createMessage(
+      message,
+      connection
+    )
     const outboundMessage = createOutboundMessage(connection, basicMessage)
+    outboundMessage.associatedRecord = basicMessageRecord
+
     await this.messageSender.sendMessage(outboundMessage)
+    return basicMessageRecord
   }
 
+  /**
+   * Retrieve all basic messages matching a given query
+   *
+   * @param query The query
+   * @returns array containing all matching records
+   */
   public async findAllByQuery(query: Partial<BasicMessageTags>) {
     return this.basicMessageService.findAllByQuery(query)
   }
 
+  /**
+   * Retrieve a basic message record by id
+   *
+   * @param basicMessageRecordId The basic message record id
+   * @throws {RecordNotFoundError} If no record is found
+   * @return The basic message record
+   *
+   */
+  public async getById(basicMessageRecordId: string) {
+    return this.basicMessageService.getById(basicMessageRecordId)
+  }
+
+  /**
+   * Delete a basic message record by id
+   *
+   * @param connectionId the basic message record id
+   * @throws {RecordNotFoundError} If no record is found
+   */
+  public async deleteById(basicMessageRecordId: string) {
+    await this.basicMessageService.deleteById(basicMessageRecordId)
+  }
+
   private registerHandlers(dispatcher: Dispatcher) {
     dispatcher.registerHandler(new BasicMessageHandler(this.basicMessageService))
   }

packages/core/src/modules/basic-messages/__tests__/basic-messages.e2e.test.ts

@@ -0,0 +1,110 @@
+/* eslint-disable @typescript-eslint/no-non-null-assertion */
+import type { SubjectMessage } from '../../../../../../tests/transport/SubjectInboundTransport'
+import type { ConnectionRecord } from '../../../modules/connections'
+
+import { Subject } from 'rxjs'
+
+import { SubjectInboundTransport } from '../../../../../../tests/transport/SubjectInboundTransport'
+import { SubjectOutboundTransport } from '../../../../../../tests/transport/SubjectOutboundTransport'
+import { getBaseConfig, makeConnection, waitForBasicMessage } from '../../../../tests/helpers'
+import testLogger from '../../../../tests/logger'
+import { Agent } from '../../../agent/Agent'
+import { MessageSendingError, RecordNotFoundError } from '../../../error'
+import { BasicMessage } from '../messages'
+import { BasicMessageRecord } from '../repository'
+
+const faberConfig = getBaseConfig('Faber Basic Messages', {
+  endpoints: ['rxjs:faber'],
+})
+
+const aliceConfig = getBaseConfig('Alice Basic Messages', {
+  endpoints: ['rxjs:alice'],
+})
+
+describe('Basic Messages E2E', () => {
+  let faberAgent: Agent
+  let aliceAgent: Agent
+  let faberConnection: ConnectionRecord
+  let aliceConnection: ConnectionRecord
+
+  beforeEach(async () => {
+    const faberMessages = new Subject<SubjectMessage>()
+    const aliceMessages = new Subject<SubjectMessage>()
+    const subjectMap = {
+      'rxjs:faber': faberMessages,
+      'rxjs:alice': aliceMessages,
+    }
+
+    faberAgent = new Agent(faberConfig.config, faberConfig.agentDependencies)
+    faberAgent.registerInboundTransport(new SubjectInboundTransport(faberMessages))
+    faberAgent.registerOutboundTransport(new SubjectOutboundTransport(subjectMap))
+    await faberAgent.initialize()
+
+    aliceAgent = new Agent(aliceConfig.config, aliceConfig.agentDependencies)
+    aliceAgent.registerInboundTransport(new SubjectInboundTransport(aliceMessages))
+    aliceAgent.registerOutboundTransport(new SubjectOutboundTransport(subjectMap))
+    await aliceAgent.initialize()
+    ;[aliceConnection, faberConnection] = await makeConnection(aliceAgent, faberAgent)
+  })
+
+  afterEach(async () => {
+    await faberAgent.shutdown()
+    await faberAgent.wallet.delete()
+    await aliceAgent.shutdown()
+    await aliceAgent.wallet.delete()
+  })
+
+  test('Alice and Faber exchange messages', async () => {
+    testLogger.test('Alice sends message to Faber')
+    const helloRecord = await aliceAgent.basicMessages.sendMessage(aliceConnection.id, 'Hello')
+
+    expect(helloRecord.content).toBe('Hello')
+
+    testLogger.test('Faber waits for message from Alice')
+    await waitForBasicMessage(faberAgent, {
+      content: 'Hello',
+    })
+
+    testLogger.test('Faber sends message to Alice')
+    const replyRecord = await faberAgent.basicMessages.sendMessage(faberConnection.id, 'How are you?')
+    expect(replyRecord.content).toBe('How are you?')
+
+    testLogger.test('Alice waits until she receives message from faber')
+    await waitForBasicMessage(aliceAgent, {
+      content: 'How are you?',
+    })
+  })
+
+  test('Alice is unable to send a message', async () => {
+    testLogger.test('Alice sends message to Faber that is undeliverable')
+
+    const spy = jest.spyOn(aliceAgent.outboundTransports[0], 'sendMessage').mockRejectedValue(new Error('any error'))
+
+    await expect(aliceAgent.basicMessages.sendMessage(aliceConnection.id, 'Hello')).rejects.toThrowError(
+      MessageSendingError
+    )
+    try {
+      await aliceAgent.basicMessages.sendMessage(aliceConnection.id, 'Hello undeliverable')
+    } catch (error) {
+      const thrownError = error as MessageSendingError
+      expect(thrownError.message).toEqual(
+        `Message is undeliverable to connection ${aliceConnection.id} (${aliceConnection.theirLabel})`
+      )
+      testLogger.test('Error thrown includes the outbound message and recently created record id')
+      expect(thrownError.outboundMessage.associatedRecord).toBeInstanceOf(BasicMessageRecord)
+      expect(thrownError.outboundMessage.payload).toBeInstanceOf(BasicMessage)
+      expect((thrownError.outboundMessage.payload as BasicMessage).content).toBe('Hello undeliverable')
+
+      testLogger.test('Created record can be found and deleted by id')
+      const storedRecord = await aliceAgent.basicMessages.getById(thrownError.outboundMessage.associatedRecord!.id)
+      expect(storedRecord).toBeInstanceOf(BasicMessageRecord)
+      expect(storedRecord.content).toBe('Hello undeliverable')
+
+      await aliceAgent.basicMessages.deleteById(storedRecord.id)
+      await expect(
+        aliceAgent.basicMessages.getById(thrownError.outboundMessage.associatedRecord!.id)
+      ).rejects.toThrowError(RecordNotFoundError)
+    }
+    spy.mockClear()
+  })
+})

packages/core/src/modules/basic-messages/services/BasicMessageService.ts

@@ -35,7 +35,7 @@ export class BasicMessageService {
     await this.basicMessageRepository.save(basicMessageRecord)
     this.emitStateChangedEvent(basicMessageRecord, basicMessage)
 
-    return basicMessage
+    return { message: basicMessage, record: basicMessageRecord }
   }
 
   /**
@@ -64,4 +64,13 @@ export class BasicMessageService {
   public async findAllByQuery(query: Partial<BasicMessageTags>) {
     return this.basicMessageRepository.findByQuery(query)
   }
+
+  public async getById(basicMessageRecordId: string) {
+    return this.basicMessageRepository.getById(basicMessageRecordId)
+  }
+
+  public async deleteById(basicMessageRecordId: string) {
+    const basicMessageRecord = await this.getById(basicMessageRecordId)
+    return this.basicMessageRepository.delete(basicMessageRecord)
+  }
 }

packages/core/src/types.ts

@@ -8,6 +8,7 @@ import type { IndyPoolConfig } from './modules/ledger/IndyPool'
 import type { OutOfBandRecord } from './modules/oob/repository'
 import type { AutoAcceptProof } from './modules/proofs'
 import type { MediatorPickupStrategy } from './modules/routing'
+import type { BaseRecord } from './storage/BaseRecord'
 
 export enum KeyDerivationMethod {
   /** default value in indy-sdk. Will be used when no value is provided */
@@ -96,6 +97,7 @@ export interface OutboundMessage<T extends AgentMessage = AgentMessage> {
   connection: ConnectionRecord
   sessionId?: string
   outOfBand?: OutOfBandRecord
+  associatedRecord?: BaseRecord
 }
 
 export interface OutboundServiceMessage<T extends AgentMessage = AgentMessage> {