wip: feat: add types and start setting up SSE response handler in server

harness/helpers/helpers.ts

@@ -11,8 +11,10 @@ import {
     callVariableValue,
     sendCommand,
 } from './commands'
-import { config_sse } from '../mockData'
+import { config_sse, SSEMessage } from '../mockData'
 import { ProxyClientOptions } from './proxyClientOptions'
+import SseStream from 'ssestream'
+import { sseStreams } from '../mockServer'
 
 const oldFetch = fetch
 
@@ -439,6 +441,17 @@ export class LocalTestClient extends BaseTestClient {
     getValidConfig() {
         return config_sse(getMockServerUrl(), `/client/${this.clientId}/sse`)
     }
+
+    getSSEStream(): SseStream {
+        return sseStreams[this.clientId]
+    }
+    sendSSEMessage(message: SSEMessage) {
+        const sseStream = this.getSSEStream()
+        if (!sseStream) {
+            throw new Error('SSE Stream not initialized')
+        }
+        sseStream.write(JSON.stringify(message))
+    }
 }
 
 export class CloudTestClient extends BaseTestClient {

harness/helpers/proxyClientOptions.ts

@@ -1,5 +1,3 @@
-import { getMockServerUrl } from './helpers'
-
 export type ProxyClientOptions = {
     eventsAPIURI?: string
     configCDNURI?: string
@@ -11,4 +9,5 @@ export type ProxyClientOptions = {
     enableRealtimeUpdates?: boolean
     logLevel?: string
     enableClientBootstrapping?: boolean
+    enableSSE?: boolean
 }

harness/mockData/index.ts

@@ -2,3 +2,4 @@ export * from './config'
 export * from './features'
 export * from './variables'
 export * from './users'
+export * from './sse'

harness/mockData/sse.ts

@@ -0,0 +1,14 @@
+
+export const SSEMessageExample = "{\"id\":\"yATzPE/mOzgY:0\",\"timestamp\":1712853334259,\"channel\":\"dvc_server_4fedfbd7a1aef0848768c8fad8f4536ca57e0ba0_v1\",\"data\":\"{\\\"etag\\\":\\\"\\\\\\\"714bc6a9acb038971923289ee6ce665b\\\\\\\"\\\",\\\"lastModified\\\":1712853333000}\",\"name\":\"change\"}"
+
+export class SSEMessageData {
+    etag: string
+    lastModified: number
+}
+export class SSEMessage {
+    id: string
+    timestamp: number
+    channel: string
+    data: string
+    name: string
+}

harness/mockServer/index.ts

@@ -2,14 +2,15 @@
 import Router from '@koa/router'
 import axios from 'axios'
 import bodyParser from 'koa-bodyparser'
+import SseStream from 'ssestream'
 
 // NOTE: This file is excecuted by the jest environment, which does not share a memory space / module cache with the
 // jest test files. This means that any things you want to be able to access from test context
 // (like the unmatchedRequests) will need to be bound to the global scope inside the jest-environment and accessed
 // through that
 
 let unmatchedRequests = []
-
+export const sseStreams: Map<string, SseStream> = new Map()
 export const assertNoUnmatchedRequests = async (
     currentClientId,
     testNameMap,
@@ -22,18 +23,18 @@
 
         if (url.includes(currentClientId)) {
             console.error('Unmatched requests: ' + currentUnmatchedRequests)
             throw new Error(
                 'Unexpected requests received: ' + currentUnmatchedRequests,
             )
         } else {
             const testName = testNameMap[clientId]
             console.error(
                 `Unmatched requests from test case ${testName} ` +
-                    currentUnmatchedRequests,
+                currentUnmatchedRequests,
             )
             throw new Error(
                 `Unexpected requests received from test case ${testName} ` +
-                    currentUnmatchedRequests,
+                currentUnmatchedRequests,
             )
         }
     }
@@ -43,6 +44,21 @@
     const app = new Koa()
     const router = new Router()
 
+    router.param('clientId', (clientId, ctx, next) => {
+        ctx.clientId = clientId
+        return next()
+    }).all('/client/:clientId/sse', async (ctx) => {
+        console.log(ctx.body)
+        let sseStream = sseStreams[ctx.clientId]
+        if (!sseStream) {
+            sseStream = new SseStream(ctx.req)
+            sseStreams[ctx.clientId] = sseStream
+        }
+        sseStream.pipe(ctx.res)
+        ctx.res.on('close', () => {
+            sseStream.unpipe(ctx.res)
+        })
+    })
     router.all('/(.*)', async (ctx) => {
         const { headers, request } = ctx
 
@@ -66,7 +82,7 @@
             } else {
                 response = await axios[request.method.toLowerCase()](
                     `https://myfakenockurl${ctx.request.url}`,
-                    request.body,
+                    ctx.request.body,
                     {
                         headers,
                     },

package.json

@@ -23,8 +23,10 @@
     "devDependencies": {
         "@devcycle/types": "^1.14.0",
         "@eresearchqut/jest-testcontainers": "^3.2.0",
-        "@koa/router": "^12.0.0",
+        "@koa/router": "^12.0.1",
         "@types/jest": "^29.2.3",
+        "@types/koa-bodyparser": "^4.3.12",
+        "@types/koa__router": "^12.0.4",
         "@typescript-eslint/eslint-plugin": "^5.44.0",
         "@typescript-eslint/parser": "^5.44.0",
         "axios": "^1.1.3",
@@ -50,6 +52,7 @@
     "dependencies": {
         "dotenv": "^16.0.3",
         "object-path-immutable": "^4.1.2",
+        "ssestream": "^1.1.0",
         "uuid": "^9.0.0"
     },
     "lint-staged": {