feat: socketdock implementation

.env.example

@@ -13,3 +13,6 @@ POSTGRES_ADMIN_PASSWORD=85a52c9283b0
 WALLET_KEY=a67d8724cb01c7c0b5d92f01fb052885
 WALLET_NAME=mediator-dev # Will be database name.
 
+USE_PUSH_NOTIFICATIONS='true'
+NOTIFICATION_WEBHOOK_URL=
+USE_SOCKETDOCK='false'

src/agent.ts

@@ -16,6 +16,7 @@ import { ariesAskar } from '@hyperledger/aries-askar-nodejs'
 
 import express from 'express'
 import { Server } from 'ws'
+import { SocketDockInboundTransport } from './transport/SocketDockInboundTransport'
 
 import config from './config'
 import { askarPostgresConfig } from './database'
@@ -90,14 +91,19 @@ export async function createAgent() {
   // Create all transports
   const httpInboundTransport = new HttpInboundTransport({ app, port: config.get('agent:port') })
   const httpOutboundTransport = new HttpOutboundTransport()
-  const wsInboundTransport = new WsInboundTransport({ server: socketServer })
-  const wsOutboundTransport = new WsOutboundTransport()
 
   // Register all Transports
   agent.registerInboundTransport(httpInboundTransport)
   agent.registerOutboundTransport(httpOutboundTransport)
-  agent.registerInboundTransport(wsInboundTransport)
-  agent.registerOutboundTransport(wsOutboundTransport)
+
+  if (!config.get('agent:useSocketDock')) {
+    const wsInboundTransport = new WsInboundTransport({ server: socketServer })
+    const wsOutboundTransport = new WsOutboundTransport()
+    agent.registerInboundTransport(wsInboundTransport)
+    agent.registerOutboundTransport(wsOutboundTransport)
+  } else {
+    agent.registerInboundTransport(new SocketDockInboundTransport({ app }))
+  }
 
   // Added health check endpoint
   httpInboundTransport.app.get('/health', async (_req, res) => {

src/config.ts

@@ -38,6 +38,7 @@ nconf.overrides({
     logLevel: process.env.LOG_LEVEL ?? LogLevel.debug,
     usePushNotifications: process.env.USE_PUSH_NOTIFICATIONS === 'true',
     notificationWebhookUrl: process.env.NOTIFICATION_WEBHOOK_URL,
+    useSocketDock: process.env.USE_SOCKETDOCK === 'true',
   },
   wallet: {
     name: process.env.WALLET_NAME ?? 'mediator-dev',

src/transport/SocketDockInboundTransport.ts

@@ -0,0 +1,98 @@
+import {
+  type Agent,
+  AgentEventTypes,
+  type AgentMessageReceivedEvent,
+  type DidCommMimeType,
+  type InboundTransport,
+  TransportService,
+} from '@credo-ts/core'
+import type { Express } from 'express'
+import express from 'express'
+import { SocketDockTransportSession } from './SocketDockTransportSession'
+
+export class SocketDockInboundTransport implements InboundTransport {
+  private app: Express
+  private activeConnections: Record<string, string> = {}
+
+  public constructor({ app }: { app: Express }) {
+    this.app = app
+
+    this.app.use(express.json())
+  }
+
+  public async start(agent: Agent) {
+    this.app.post('/connect', async (req, res) => {
+      agent.config.logger.info('SocketDockInboundTransport.connect')
+      const { connection_id: connectionId } = req.body.meta
+      if (!connectionId) {
+        throw new Error('ConnectionId is not sent from socketDock server')
+      }
+
+      const socketId = this.activeConnections[connectionId]
+      if (!socketId) {
+        this.activeConnections[connectionId] = connectionId
+        agent.config.logger.debug(`Saving new socketId : ${connectionId}`)
+      }
+
+      try {
+        res.status(200).send(`connection with socketId : ${connectionId} added successfully`)
+      } catch (error) {
+        res.status(500).send('Error sending response to send URL')
+      }
+    })
+
+    this.app.post('/message', async (req, res) => {
+      agent.config.logger.info('SocketDockInboundTransport.message')
+
+      const { connection_id: connectionId } = req.body.meta
+      if (!connectionId) {
+        throw new Error('ConnectionId is not sent from socketDock server')
+      }
+
+      try {
+        const socketId = this.activeConnections[connectionId]
+        agent.config.logger.debug(`activeConnections transport session : ${socketId} ${connectionId}`)
+        const sendUrl = req.body.meta.send
+        const requestMimeType = req.headers['content-type'] as DidCommMimeType
+        const session = new SocketDockTransportSession(socketId, res, sendUrl, requestMimeType)
+        const message = req.body.message
+        const encryptedMessage = JSON.parse(message)
+
+        agent.config.logger.debug(`Session value for transport session : ${session.id}`)
+
+        agent.events.emit<AgentMessageReceivedEvent>(agent.context, {
+          type: AgentEventTypes.AgentMessageReceived,
+          payload: {
+            message: encryptedMessage,
+            session: session,
+          },
+        })
+      } catch (error) {
+        if (!res.headersSent) {
+          res.status(500).send('Error processing message')
+        }
+      }
+    })
+
+    this.app.post('/disconnect', async (req, res) => {
+      agent.config.logger.info('SocketDockInboundTransport.disconnect')
+      const transportService = agent.dependencyManager.resolve(TransportService)
+      const { connection_id } = req.body
+      if (!connection_id) {
+        throw new Error('ConnectionId is not sent from socketDock server')
+      }
+
+      delete this.activeConnections[connection_id]
+      const session = transportService.findSessionById(connection_id)
+      agent.config.logger.debug(`Got session from transportService ${session?.id}`)
+      if (session) {
+        transportService.removeSession(session)
+      }
+
+      agent.config.logger.debug(`removed mapping of socketId : ${connection_id}`)
+      res.status(200).send(`Mapping with socketId : ${connection_id} removed successfully`)
+    })
+  }
+
+  public async stop(): Promise<void> {}
+}

src/transport/SocketDockTransportSession.ts

@@ -0,0 +1,49 @@
+import { type AgentContext, DidCommMimeType, type EncryptedMessage, type TransportSession } from '@credo-ts/core'
+import { CredoError } from '@credo-ts/core'
+import { agentDependencies } from '@credo-ts/node'
+import type { Response } from 'express'
+
+export const supportedContentTypes: string[] = [DidCommMimeType.V0, DidCommMimeType.V1]
+
+export class SocketDockTransportSession implements TransportSession {
+  public id: string
+  public readonly type = 'socketdock'
+  public res: Response
+  public sendUrl: string
+  public requestMimeType: DidCommMimeType
+
+  public constructor(id: string, res: Response, sendUrl: string, requestMimeType: DidCommMimeType) {
+    this.id = id
+    this.res = res
+    this.sendUrl = sendUrl
+    this.requestMimeType = requestMimeType
+  }
+
+  public async close() {
+    if (!this.res.headersSent) {
+      this.res.status(200).end()
+    }
+  }
+
+  public async send(agentContext: AgentContext, encryptedMessage: EncryptedMessage): Promise<void> {
+    if (this.res.headersSent) {
+      throw new CredoError(`${this.type} transport session has been closed.`)
+    }
+
+    // By default we take the agent config's default DIDComm content-type
+    let responseMimeType = agentContext.config.didCommMimeType
+
+    if (this.requestMimeType && supportedContentTypes.includes(this.requestMimeType)) {
+      responseMimeType = this.requestMimeType
+    }
+
+    const requestOptions = {
+      method: 'POST',
+      headers: {
+        'Content-Type': responseMimeType,
+      },
+      body: JSON.stringify(encryptedMessage),
+    }
+    await agentDependencies.fetch(this.sendUrl, requestOptions)
+  }
+}