feat: socketdock implementation

.env.example

@@ -5,4 +5,5 @@ POSTGRES_ADMIN_USER=
 POSTGRES_ADMIN_PASSWORD=
 
 USE_PUSH_NOTIFICATIONS='true'
-NOTIFICATION_WEBHOOK_URL=
+NOTIFICATION_WEBHOOK_URL=
+USE_SOCKETDOCK='false'

package.json

@@ -54,4 +54,4 @@
       "@credo-ts/core": "patches/@credo-ts__core.patch"
     }
   }
-}
+}

src/agent.ts

@@ -16,8 +16,18 @@ import { ariesAskar } from '@hyperledger/aries-askar-nodejs'
 
 import express from 'express'
 import { Server } from 'ws'
+import { SocketDockInboundTransport } from './transport/SocketDockInboundTransport'
 
-import { AGENT_ENDPOINTS, AGENT_NAME, AGENT_PORT, LOG_LEVEL, POSTGRES_HOST, WALLET_KEY, WALLET_NAME } from './constants'
+import {
+  AGENT_ENDPOINTS,
+  AGENT_NAME,
+  AGENT_PORT,
+  LOG_LEVEL,
+  POSTGRES_HOST,
+  USE_SOCKETDOCK,
+  WALLET_KEY,
+  WALLET_NAME,
+} from './constants'
 import { askarPostgresConfig } from './database'
 import { Logger } from './logger'
 import { PushNotificationsFcmModule } from './push-notifications/fcm'
@@ -90,14 +100,19 @@ export async function createAgent() {
   // Create all transports
   const httpInboundTransport = new HttpInboundTransport({ app, port: 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 (USE_SOCKETDOCK === 'false') {
+    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/constants.ts

@@ -17,6 +17,8 @@ export const POSTGRES_ADMIN_PASSWORD = process.env.POSTGRES_ADMIN_PASSWORD
 
 export const INVITATION_URL = process.env.INVITATION_URL
 
+export const USE_SOCKETDOCK = process.env.USE_SOCKETDOCK || 'false'
+
 export const LOG_LEVEL = LogLevel.debug
 
 export const IS_DEV = process.env.NODE_ENV === 'development'

src/transport/SocketDockInboundTransport.ts

@@ -0,0 +1,91 @@
+import type { Express } from 'express'
+import { Agent, AgentEventTypes, AgentMessageReceivedEvent, InboundTransport, TransportService } from '@credo-ts/core'
+import { SocketDockTransportSession } from './SocketDockTransportSession'
+import express from 'express'
+
+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 string
+        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,50 @@
+import { AgentContext, DidCommMimeType, EncryptedMessage, TransportSession } from '@credo-ts/core'
+import type { Response } from 'express'
+import { CredoError } from '@credo-ts/core'
+
+import { agentDependencies } from '@credo-ts/node'
+
+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: any
+
+  public constructor(id: string, res: Response, sendUrl: string, requestMimeType: string) {
+    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)
+  }
+}