feat: support DRPC protocol (openwallet-foundation#1753)

Signed-off-by: wadeking98 <wkingnumber2@gmail.com>

packages/drpc/README.md

@@ -0,0 +1,73 @@
+<p align="center">
+  <br />
+  <img
+    alt="Credo Logo"
+    src="https://github.com/openwallet-foundation/credo-ts/blob/c7886cb8377ceb8ee4efe8d264211e561a75072d/images/credo-logo.png"
+    height="250px"
+  />
+</p>
+<h1 align="center"><b>Credo DRPC Module</b></h1>
+<p align="center">
+  <a
+    href="https://raw.githubusercontent.com/openwallet-foundation/credo-ts/main/LICENSE"
+    ><img
+      alt="License"
+      src="https://img.shields.io/badge/License-Apache%202.0-blue.svg"
+  /></a>
+  <a href="https://www.typescriptlang.org/"
+    ><img
+      alt="typescript"
+      src="https://img.shields.io/badge/%3C%2F%3E-TypeScript-%230074c1.svg"
+  /></a>
+    <a href="https://www.npmjs.com/package/@credo-ts/question-answer"
+    ><img
+      alt="@credo-ts/question-answer version"
+      src="https://img.shields.io/npm/v/@credo-ts/question-answer"
+  /></a>
+
+</p>
+<br />
+
+DRPC module for [Credo](https://github.com/openwallet-foundation/credo-ts.git). Implements [Aries RFC 0804](https://github.com/hyperledger/aries-rfcs/blob/ea87d2e37640ef944568e3fa01df1f36fe7f0ff3/features/0804-didcomm-rpc/README.md).
+
+### Quick start
+
+In order for this module to work, we have to inject it into the agent to access agent functionality. See the example for more information.
+
+### Example of usage
+
+```ts
+import { DrpcModule } from '@credo-ts/drpc'
+
+const agent = new Agent({
+  config: {
+    /* config */
+  },
+  dependencies: agentDependencies,
+  modules: {
+    drpc: new DrpcModule(),
+    /* other custom modules */
+  },
+})
+
+await agent.initialize()
+
+// Send a request to the specified connection
+const responseListener = await senderAgent.modules.drpc.sendRequest(connectionId, {
+  jsonrpc: '2.0',
+  method: 'hello',
+  id: 1,
+})
+
+// Listen for any incoming requests
+const { request, sendResponse } = await receiverAgent.modules.drpc.recvRequest()
+
+// Process the received request and create a response
+const result =
+  request.method === 'hello'
+    ? { jsonrpc: '2.0', result: 'Hello world!', id: request.id }
+    : { jsonrpc: '2.0', error: { code: DrpcErrorCode.METHOD_NOT_FOUND, message: 'Method not found' } }
+
+// Send the response back
+await sendResponse(result)
+```

packages/drpc/jest.config.ts

@@ -0,0 +1,13 @@
+import type { Config } from '@jest/types'
+
+import base from '../../jest.config.base'
+
+import packageJson from './package.json'
+
+const config: Config.InitialOptions = {
+  ...base,
+  displayName: packageJson.name,
+  setupFilesAfterEnv: ['./tests/setup.ts'],
+}
+
+export default config

packages/drpc/package.json

@@ -0,0 +1,37 @@
+{
+  "name": "@credo-ts/drpc",
+  "main": "build/index",
+  "types": "build/index",
+  "version": "0.4.2",
+  "files": [
+    "build"
+  ],
+  "license": "Apache-2.0",
+  "publishConfig": {
+    "access": "public"
+  },
+  "homepage": "https://github.com/openwallet-foundation/credo-ts/tree/main/packages/drpc",
+  "repository": {
+    "type": "git",
+    "url": "https://github.com/openwallet-foundation/credo-ts",
+    "directory": "packages/drpc"
+  },
+  "scripts": {
+    "build": "yarn run clean && yarn run compile",
+    "clean": "rimraf ./build",
+    "compile": "tsc -p tsconfig.build.json",
+    "prepublishOnly": "yarn run build",
+    "test": "jest"
+  },
+  "dependencies": {
+    "@credo-ts/core": "0.4.2",
+    "class-transformer": "^0.5.1",
+    "class-validator": "0.14.1"
+  },
+  "devDependencies": {
+    "@credo-ts/node": "0.4.2",
+    "reflect-metadata": "^0.1.13",
+    "rimraf": "^4.4.0",
+    "typescript": "~4.9.5"
+  }
+}

packages/drpc/src/DrpcApi.ts

@@ -0,0 +1,184 @@
+import type { DrpcRequest, DrpcResponse, DrpcRequestMessage, DrpcResponseMessage } from './messages'
+import type { DrpcRecord } from './repository/DrpcRecord'
+import type { ConnectionRecord } from '@credo-ts/core'
+
+import {
+  AgentContext,
+  MessageHandlerRegistry,
+  MessageSender,
+  OutboundMessageContext,
+  injectable,
+  ConnectionService,
+} from '@credo-ts/core'
+
+import { DrpcRequestHandler, DrpcResponseHandler } from './handlers'
+import { DrpcService } from './services'
+
+@injectable()
+export class DrpcApi {
+  private drpcMessageService: DrpcService
+  private messageSender: MessageSender
+  private connectionService: ConnectionService
+  private agentContext: AgentContext
+
+  public constructor(
+    messageHandlerRegistry: MessageHandlerRegistry,
+    drpcMessageService: DrpcService,
+    messageSender: MessageSender,
+    connectionService: ConnectionService,
+    agentContext: AgentContext
+  ) {
+    this.drpcMessageService = drpcMessageService
+    this.messageSender = messageSender
+    this.connectionService = connectionService
+    this.agentContext = agentContext
+    this.registerMessageHandlers(messageHandlerRegistry)
+  }
+
+  /**
+   * sends the request object to the connection and returns a function that will resolve to the response
+   * @param connectionId the connection to send the request to
+   * @param request the request object
+   * @returns curried function that waits for the response with an optional timeout in seconds
+   */
+  public async sendRequest(
+    connectionId: string,
+    request: DrpcRequest
+  ): Promise<() => Promise<DrpcResponse | undefined>> {
+    const connection = await this.connectionService.getById(this.agentContext, connectionId)
+    const { requestMessage: drpcMessage, record: drpcMessageRecord } =
+      await this.drpcMessageService.createRequestMessage(this.agentContext, request, connection.id)
+    const messageId = drpcMessage.id
+    await this.sendMessage(connection, drpcMessage, drpcMessageRecord)
+    return async (timeout?: number) => {
+      return await this.recvResponse(messageId, timeout)
+    }
+  }
+
+  /**
+   * Listen for a response that has a thread id matching the provided messageId
+   * @param messageId the id to match the response to
+   * @param timeoutMs the time in milliseconds to wait for a response
+   * @returns the response object
+   */
+  private async recvResponse(messageId: string, timeoutMs?: number): Promise<DrpcResponse | undefined> {
+    return new Promise((resolve) => {
+      const listener = ({
+        drpcMessageRecord,
+        removeListener,
+      }: {
+        drpcMessageRecord: DrpcRecord
+        removeListener: () => void
+      }) => {
+        const response = drpcMessageRecord.response
+        if (drpcMessageRecord.threadId === messageId) {
+          removeListener()
+          resolve(response)
+        }
+      }
+
+      const cancelListener = this.drpcMessageService.createResponseListener(listener)
+      if (timeoutMs) {
+        const handle = setTimeout(() => {
+          clearTimeout(handle)
+          cancelListener()
+          resolve(undefined)
+        }, timeoutMs)
+      }
+    })
+  }
+
+  /**
+   * Listen for a request and returns the request object and a function to send the response
+   * @param timeoutMs the time in seconds to wait for a request
+   * @returns the request object and a function to send the response
+   */
+  public async recvRequest(timeoutMs?: number): Promise<
+    | {
+        request: DrpcRequest
+        sendResponse: (response: DrpcResponse) => Promise<void>
+      }
+    | undefined
+  > {
+    return new Promise((resolve) => {
+      const listener = ({
+        drpcMessageRecord,
+        removeListener,
+      }: {
+        drpcMessageRecord: DrpcRecord
+        removeListener: () => void
+      }) => {
+        const request = drpcMessageRecord.request
+        if (request) {
+          removeListener()
+          resolve({
+            sendResponse: async (response: DrpcResponse) => {
+              await this.sendResponse({
+                connectionId: drpcMessageRecord.connectionId,
+                threadId: drpcMessageRecord.threadId,
+                response,
+              })
+            },
+            request,
+          })
+        }
+      }
+
+      const cancelListener = this.drpcMessageService.createRequestListener(listener)
+
+      if (timeoutMs) {
+        const handle = setTimeout(() => {
+          clearTimeout(handle)
+          cancelListener()
+          resolve(undefined)
+        }, timeoutMs)
+      }
+    })
+  }
+
+  /**
+   * Sends a drpc response to a connection
+   * @param connectionId the connection id to use
+   * @param threadId the thread id to respond to
+   * @param response the drpc response object to send
+   */
+  private async sendResponse(options: {
+    connectionId: string
+    threadId: string
+    response: DrpcResponse
+  }): Promise<void> {
+    const connection = await this.connectionService.getById(this.agentContext, options.connectionId)
+    const drpcMessageRecord = await this.drpcMessageService.findByThreadAndConnectionId(
+      this.agentContext,
+      options.connectionId,
+      options.threadId
+    )
+    if (!drpcMessageRecord) {
+      throw new Error(`No request found for threadId ${options.threadId}`)
+    }
+    const { responseMessage, record } = await this.drpcMessageService.createResponseMessage(
+      this.agentContext,
+      options.response,
+      drpcMessageRecord
+    )
+    await this.sendMessage(connection, responseMessage, record)
+  }
+
+  private async sendMessage(
+    connection: ConnectionRecord,
+    message: DrpcRequestMessage | DrpcResponseMessage,
+    messageRecord: DrpcRecord
+  ): Promise<void> {
+    const outboundMessageContext = new OutboundMessageContext(message, {
+      agentContext: this.agentContext,
+      connection,
+      associatedRecord: messageRecord,
+    })
+    await this.messageSender.sendMessage(outboundMessageContext)
+  }
+
+  private registerMessageHandlers(messageHandlerRegistry: MessageHandlerRegistry) {
+    messageHandlerRegistry.registerMessageHandler(new DrpcRequestHandler(this.drpcMessageService))
+    messageHandlerRegistry.registerMessageHandler(new DrpcResponseHandler(this.drpcMessageService))
+  }
+}

packages/drpc/src/DrpcModule.ts

@@ -0,0 +1,38 @@
+import type { FeatureRegistry, DependencyManager, Module } from '@credo-ts/core'
+
+import { Protocol, AgentConfig } from '@credo-ts/core'
+
+import { DrpcApi } from './DrpcApi'
+import { DrpcRole } from './models/DrpcRole'
+import { DrpcRepository } from './repository'
+import { DrpcService } from './services'
+
+export class DrpcModule implements Module {
+  public readonly api = DrpcApi
+
+  /**
+   * Registers the dependencies of the drpc message module on the dependency manager.
+   */
+  public register(dependencyManager: DependencyManager, featureRegistry: FeatureRegistry) {
+    // Warn about experimental module
+    dependencyManager
+      .resolve(AgentConfig)
+      .logger.warn(
+        "The '@credo-ts/drpc' module is experimental and could have unexpected breaking changes. When using this module, make sure to use strict versions for all @credo-ts packages."
+      )
+
+    // Services
+    dependencyManager.registerSingleton(DrpcService)
+
+    // Repositories
+    dependencyManager.registerSingleton(DrpcRepository)
+
+    // Features
+    featureRegistry.register(
+      new Protocol({
+        id: 'https://didcomm.org/drpc/1.0',
+        roles: [DrpcRole.Client, DrpcRole.Server],
+      })
+    )
+  }
+}

packages/drpc/src/DrpcRequestEvents.ts

@@ -0,0 +1,12 @@
+import type { DrpcRecord } from './repository'
+import type { BaseEvent } from '@credo-ts/core'
+
+export enum DrpcRequestEventTypes {
+  DrpcRequestStateChanged = 'DrpcRequestStateChanged',
+}
+export interface DrpcRequestStateChangedEvent extends BaseEvent {
+  type: typeof DrpcRequestEventTypes.DrpcRequestStateChanged
+  payload: {
+    drpcMessageRecord: DrpcRecord
+  }
+}

packages/drpc/src/DrpcResponseEvents.ts

@@ -0,0 +1,12 @@
+import type { DrpcRecord } from './repository'
+import type { BaseEvent } from '@credo-ts/core'
+
+export enum DrpcResponseEventTypes {
+  DrpcResponseStateChanged = 'DrpcResponseStateChanged',
+}
+export interface DrpcResponseStateChangedEvent extends BaseEvent {
+  type: typeof DrpcResponseEventTypes.DrpcResponseStateChanged
+  payload: {
+    drpcMessageRecord: DrpcRecord
+  }
+}