feat: adding trust ping events and trust ping command (#1182)

Signed-off-by: Kim Ebert <[email protected]>

packages/core/src/modules/connections/ConnectionsApi.ts

@@ -32,6 +32,11 @@ import { HandshakeProtocol } from './models'
 import { ConnectionService } from './services/ConnectionService'
 import { TrustPingService } from './services/TrustPingService'
 
+export interface SendPingOptions {
+  responseRequested?: boolean
+  withReturnRouting?: boolean
+}
+
 @injectable()
 export class ConnectionsApi {
   /**
@@ -227,6 +232,41 @@ export class ConnectionsApi {
     return connectionRecord
   }
 
+  /**
+   * Send a trust ping to an established connection
+   *
+   * @param connectionId the id of the connection for which to accept the response
+   * @param responseRequested do we want a response to our ping
+   * @param withReturnRouting do we want a response at the time of posting
+   * @returns TurstPingMessage
+   */
+  public async sendPing(
+    connectionId: string,
+    { responseRequested = true, withReturnRouting = undefined }: SendPingOptions
+  ) {
+    const connection = await this.getById(connectionId)
+
+    const { message } = await this.connectionService.createTrustPing(this.agentContext, connection, {
+      responseRequested: responseRequested,
+    })
+
+    if (withReturnRouting === true) {
+      message.setReturnRouting(ReturnRouteTypes.all)
+    }
+
+    // Disable return routing as we don't want to receive a response for this message over the same channel
+    // This has led to long timeouts as not all clients actually close an http socket if there is no response message
+    if (withReturnRouting === false) {
+      message.setReturnRouting(ReturnRouteTypes.none)
+    }
+
+    await this.messageSender.sendMessage(
+      new OutboundMessageContext(message, { agentContext: this.agentContext, connection })
+    )
+
+    return message
+  }
+
   public async returnWhenIsConnected(connectionId: string, options?: { timeoutMs: number }): Promise<ConnectionRecord> {
     return this.connectionService.returnWhenIsConnected(this.agentContext, connectionId, options?.timeoutMs)
   }

packages/core/src/modules/connections/TrustPingEvents.ts

@@ -0,0 +1,24 @@
+import type { BaseEvent } from '../../agent/Events'
+import type { TrustPingMessage, TrustPingResponseMessage } from './messages'
+import type { ConnectionRecord } from './repository/ConnectionRecord'
+
+export enum TrustPingEventTypes {
+  TrustPingReceivedEvent = 'TrustPingReceivedEvent',
+  TrustPingResponseReceivedEvent = 'TrustPingResponseReceivedEvent',
+}
+
+export interface TrustPingReceivedEvent extends BaseEvent {
+  type: typeof TrustPingEventTypes.TrustPingReceivedEvent
+  payload: {
+    connectionRecord: ConnectionRecord
+    message: TrustPingMessage
+  }
+}
+
+export interface TrustPingResponseReceivedEvent extends BaseEvent {
+  type: typeof TrustPingEventTypes.TrustPingResponseReceivedEvent
+  payload: {
+    connectionRecord: ConnectionRecord
+    message: TrustPingResponseMessage
+  }
+}

packages/core/src/modules/connections/index.ts

@@ -3,6 +3,7 @@ export * from './models'
 export * from './repository'
 export * from './services'
 export * from './ConnectionEvents'
+export * from './TrustPingEvents'
 export * from './ConnectionsApi'
 export * from './DidExchangeProtocol'
 export * from './ConnectionsModuleConfig'

packages/core/src/modules/connections/services/TrustPingService.ts

@@ -1,14 +1,31 @@
 import type { InboundMessageContext } from '../../../agent/models/InboundMessageContext'
+import type { TrustPingReceivedEvent, TrustPingResponseReceivedEvent } from '../TrustPingEvents'
 import type { TrustPingMessage } from '../messages'
 import type { ConnectionRecord } from '../repository/ConnectionRecord'
 
+import { EventEmitter } from '../../../agent/EventEmitter'
 import { OutboundMessageContext } from '../../../agent/models'
 import { injectable } from '../../../plugins'
+import { TrustPingEventTypes } from '../TrustPingEvents'
 import { TrustPingResponseMessage } from '../messages'
 
 @injectable()
 export class TrustPingService {
+  private eventEmitter: EventEmitter
+
+  public constructor(eventEmitter: EventEmitter) {
+    this.eventEmitter = eventEmitter
+  }
+
   public processPing({ message, agentContext }: InboundMessageContext<TrustPingMessage>, connection: ConnectionRecord) {
+    this.eventEmitter.emit<TrustPingReceivedEvent>(agentContext, {
+      type: TrustPingEventTypes.TrustPingReceivedEvent,
+      payload: {
+        connectionRecord: connection,
+        message: message,
+      },
+    })
+
     if (message.responseRequested) {
       const response = new TrustPingResponseMessage({
         threadId: message.id,
@@ -18,8 +35,17 @@ export class TrustPingService {
     }
   }
 
-  // eslint-disable-next-line @typescript-eslint/no-unused-vars
   public processPingResponse(inboundMessage: InboundMessageContext<TrustPingResponseMessage>) {
-    // TODO: handle ping response message
+    const { agentContext, message } = inboundMessage
+
+    const connection = inboundMessage.assertReadyConnection()
+
+    this.eventEmitter.emit<TrustPingResponseReceivedEvent>(agentContext, {
+      type: TrustPingEventTypes.TrustPingResponseReceivedEvent,
+      payload: {
+        connectionRecord: connection,
+        message: message,
+      },
+    })
   }
 }

packages/core/tests/connections.test.ts

@@ -18,7 +18,7 @@ import { Agent } from '../src/agent/Agent'
 import { didKeyToVerkey } from '../src/modules/dids/helpers'
 import { OutOfBandState } from '../src/modules/oob/domain/OutOfBandState'
 
-import { getAgentOptions } from './helpers'
+import { getAgentOptions, waitForTrustPingResponseReceivedEvent } from './helpers'
 
 describe('connections', () => {
   let faberAgent: Agent
@@ -85,6 +85,25 @@ describe('connections', () => {
     await mediatorAgent.wallet.delete()
   })
 
+  it('one agent should be able to send and receive a ping', async () => {
+    const faberOutOfBandRecord = await faberAgent.oob.createInvitation({
+      handshakeProtocols: [HandshakeProtocol.Connections],
+      multiUseInvitation: true,
+    })
+
+    const invitation = faberOutOfBandRecord.outOfBandInvitation
+    const invitationUrl = invitation.toUrl({ domain: 'https://example.com' })
+
+    // Receive invitation with alice agent
+    let { connectionRecord: aliceFaberConnection } = await aliceAgent.oob.receiveInvitationFromUrl(invitationUrl)
+    aliceFaberConnection = await aliceAgent.connections.returnWhenIsConnected(aliceFaberConnection!.id)
+    expect(aliceFaberConnection.state).toBe(DidExchangeState.Completed)
+
+    const ping = await aliceAgent.connections.sendPing(aliceFaberConnection.id, {})
+
+    await waitForTrustPingResponseReceivedEvent(aliceAgent, { threadId: ping.threadId })
+  })
+
   it('one should be able to make multiple connections using a multi use invite', async () => {
     const faberOutOfBandRecord = await faberAgent.oob.createInvitation({
       handshakeProtocols: [HandshakeProtocol.Connections],

packages/core/tests/helpers.ts

@@ -8,6 +8,7 @@ import type {
   ConnectionRecordProps,
   CredentialDefinitionTemplate,
   CredentialStateChangedEvent,
+  TrustPingResponseReceivedEvent,
   InitConfig,
   InjectionToken,
   ProofStateChangedEvent,
@@ -45,6 +46,7 @@ import {
   ConnectionRecord,
   CredentialEventTypes,
   CredentialState,
+  TrustPingEventTypes,
   DependencyManager,
   DidExchangeRole,
   DidExchangeState,
@@ -240,6 +242,50 @@ export function waitForProofExchangeRecordSubject(
   )
 }
 
+export async function waitForTrustPingResponseReceivedEvent(
+  agent: Agent,
+  options: {
+    threadId?: string
+    parentThreadId?: string
+    state?: ProofState
+    previousState?: ProofState | null
+    timeoutMs?: number
+  }
+) {
+  const observable = agent.events.observable<TrustPingResponseReceivedEvent>(
+    TrustPingEventTypes.TrustPingResponseReceivedEvent
+  )
+
+  return waitForTrustPingResponseReceivedEventSubject(observable, options)
+}
+
+export function waitForTrustPingResponseReceivedEventSubject(
+  subject: ReplaySubject<TrustPingResponseReceivedEvent> | Observable<TrustPingResponseReceivedEvent>,
+  {
+    threadId,
+    timeoutMs = 10000,
+  }: {
+    threadId?: string
+    timeoutMs?: number
+  }
+) {
+  const observable = subject instanceof ReplaySubject ? subject.asObservable() : subject
+  return firstValueFrom(
+    observable.pipe(
+      filter((e) => threadId === undefined || e.payload.message.threadId === threadId),
+      timeout(timeoutMs),
+      catchError(() => {
+        throw new Error(
+          `TrustPingResponseReceivedEvent event not emitted within specified timeout: {
+  threadId: ${threadId},
+}`
+        )
+      }),
+      map((e) => e.payload.message)
+    )
+  )
+}
+
 export function waitForCredentialRecordSubject(
   subject: ReplaySubject<CredentialStateChangedEvent> | Observable<CredentialStateChangedEvent>,
   {