feat(ws): enable websocket log streaming

package.json

@@ -17,40 +17,43 @@
     "lint:fix": "eslint . --fix"
   },
   "dependencies": {
+    "@fastify/websocket": "^11.0.2",
     "@inquirer/prompts": "^7.3.3",
     "@platformatic/composer": "^2.60.0",
+    "@platformatic/control": "^2.60.0",
     "@platformatic/runtime": "^2.60.0",
     "@platformatic/service": "^2.60.0",
-    "@platformatic/control": "^2.60.0",
     "@platformatic/vite": "^2.60.0",
+    "fastify": "^5.0.0",
     "proxyquire": "^2.1.3",
+    "react-use-websocket": "^4.13.0",
     "split2": "4.2.0",
-    "fastify": "^5.0.0",
     "wattpm": "^2.60.0"
   },
   "devDependencies": {
     "@fastify/type-provider-json-schema-to-ts": "^5.0.0",
-    "@types/proxyquire": "1.3.31",
-    "fastify-tsconfig": "^2.0.0",
-    "borp": "^0.19.0",
     "@inquirer/testing": "^2.1.45",
-    "@types/split2": "4.2.3",
-    "eslint": "^9.22.0",
-    "typescript": "^5.5.4",
-    "neostandard": "^0.12.1",
     "@platformatic/client-cli": "^2.58.0",
     "@platformatic/ui-components": "^0.15.2",
     "@types/d3": "^7.4.3",
+    "@types/proxyquire": "1.3.31",
     "@types/react-dom": "^19.0.4",
+    "@types/split2": "4.2.3",
+    "@types/ws": "^8.18.1",
     "@vitejs/plugin-react": "^4.3.3",
     "autoprefixer": "^10.4.20",
+    "borp": "^0.19.0",
     "d3": "~7.9.0",
     "dayjs": "^1.11.13",
+    "eslint": "^9.22.0",
+    "fastify-tsconfig": "^2.0.0",
+    "neostandard": "^0.12.1",
     "react": "^19.0.0",
     "react-dom": "^19.0.0",
     "react-router-dom": "^7.0.0",
     "source-map-support": "^0.5.21",
     "tailwindcss": "^3.4.15",
+    "typescript": "^5.5.4",
     "use-error-boundary": "^2.0.6",
     "vite": "^5.4.11",
     "vitest": "^3.0.8",

watt.json

@@ -8,6 +8,12 @@
     "level": "info"
   },
   "autoload": {
-    "path": "web"
+    "path": "web",
+    "mappings": {
+      "backend": {
+        "id": "backend",
+        "useHttp": true
+      }
+    }
   }
 }

web/backend/openapi.json

@@ -722,25 +722,6 @@
         }
       }
     },
-    "/runtimes/{pid}/logs": {
-      "get": {
-        "parameters": [
-          {
-            "schema": {
-              "type": "number"
-            },
-            "in": "path",
-            "name": "pid",
-            "required": true
-          }
-        ],
-        "responses": {
-          "200": {
-            "description": "Default Response"
-          }
-        }
-      }
-    },
     "/runtimes/{pid}/openapi/{serviceId}": {
       "get": {
         "parameters": [

web/backend/plugins/websocket.ts

@@ -0,0 +1,6 @@
+import websocket from '@fastify/websocket'
+import { FastifyInstance } from 'fastify'
+
+export default async function (fastify: FastifyInstance) {
+  await fastify.register(websocket)
+}

web/backend/routes/metrics.ts

@@ -0,0 +1,47 @@
+import { FastifyInstance } from 'fastify'
+import { JsonSchemaToTsProvider } from '@fastify/type-provider-json-schema-to-ts'
+import { metricResponseSchema, pidParamSchema } from '../schemas'
+
+export default async function (fastify: FastifyInstance) {
+  const typedFastify = fastify.withTypeProvider<JsonSchemaToTsProvider>()
+  const emptyMetrics = { dataCpu: [], dataLatency: [], dataMem: [], dataReq: [] }
+
+  typedFastify.get('/runtimes/:pid/metrics', {
+    schema: { params: pidParamSchema, response: { 200: metricResponseSchema } },
+  }, async ({ params: { pid } }) => {
+    return typedFastify.mappedMetrics[pid]?.aggregated || emptyMetrics
+  })
+
+  typedFastify.get('/runtimes/:pid/metrics/:serviceId', {
+    schema: {
+      params: {
+        type: 'object',
+        properties: {
+          pid: { type: 'number' },
+          serviceId: { type: 'string' }
+        },
+        required: ['pid', 'serviceId']
+      },
+      response: { 200: metricResponseSchema }
+    }
+  }, async ({ params: { pid, serviceId } }) => {
+    return fastify.mappedMetrics[pid]?.services[serviceId]?.all || emptyMetrics
+  })
+
+  typedFastify.get('/runtimes/:pid/metrics/:serviceId/:workerId', {
+    schema: {
+      params: {
+        type: 'object',
+        properties: {
+          pid: { type: 'number' },
+          serviceId: { type: 'string' },
+          workerId: { type: 'number' },
+        },
+        required: ['pid', 'serviceId', 'workerId']
+      },
+      response: { 200: metricResponseSchema }
+    }
+  }, async ({ params: { pid, serviceId, workerId } }) => {
+    return fastify.mappedMetrics[pid]?.services[serviceId]?.[workerId] || emptyMetrics
+  })
+}

web/backend/routes/root.ts

@@ -1,13 +1,11 @@
 import { FastifyInstance } from 'fastify'
 import { JsonSchemaToTsProvider } from '@fastify/type-provider-json-schema-to-ts'
 import { RuntimeApiClient } from '@platformatic/control'
-import { getLogsFromReadable } from '../utils/log'
-import { metricResponseSchema, SelectableRuntime, selectableRuntimeSchema } from '../schemas'
+import { pidParamSchema, SelectableRuntime, selectableRuntimeSchema } from '../schemas'
 
 export default async function (fastify: FastifyInstance) {
   const typedFastify = fastify.withTypeProvider<JsonSchemaToTsProvider>()
   const api = new RuntimeApiClient()
-  const emptyMetrics = { dataCpu: [], dataLatency: [], dataMem: [], dataReq: [] }
 
   typedFastify.get('/runtimes', {
     schema: {
@@ -42,7 +40,7 @@ export default async function (fastify: FastifyInstance) {
 
   typedFastify.get('/runtimes/:pid/health', {
     schema: {
-      params: { type: 'object', properties: { pid: { type: 'number' } }, required: ['pid'] },
+      params: pidParamSchema,
       response: {
         200: {
           type: 'object',
@@ -70,51 +68,9 @@ export default async function (fastify: FastifyInstance) {
     }
   })
 
-  typedFastify.get('/runtimes/:pid/metrics', {
-    schema: {
-      params: { type: 'object', properties: { pid: { type: 'number' } }, required: ['pid'] },
-      response: { 200: metricResponseSchema }
-    },
-  }, async ({ params: { pid } }) => {
-    return typedFastify.mappedMetrics[pid]?.aggregated || emptyMetrics
-  })
-
-  typedFastify.get('/runtimes/:pid/metrics/:serviceId', {
-    schema: {
-      params: {
-        type: 'object',
-        properties: {
-          pid: { type: 'number' },
-          serviceId: { type: 'string' }
-        },
-        required: ['pid', 'serviceId']
-      },
-      response: { 200: metricResponseSchema }
-    }
-  }, async ({ params: { pid, serviceId } }) => {
-    return fastify.mappedMetrics[pid]?.services[serviceId]?.all || emptyMetrics
-  })
-
-  typedFastify.get('/runtimes/:pid/metrics/:serviceId/:workerId', {
-    schema: {
-      params: {
-        type: 'object',
-        properties: {
-          pid: { type: 'number' },
-          serviceId: { type: 'string' },
-          workerId: { type: 'number' },
-        },
-        required: ['pid', 'serviceId', 'workerId']
-      },
-      response: { 200: metricResponseSchema }
-    }
-  }, async ({ params: { pid, serviceId, workerId } }) => {
-    return fastify.mappedMetrics[pid]?.services[serviceId]?.[workerId] || emptyMetrics
-  })
-
   typedFastify.get('/runtimes/:pid/services', {
     schema: {
-      params: { type: 'object', properties: { pid: { type: 'number' } }, required: ['pid'] },
+      params: pidParamSchema,
       response: {
         200: {
           type: 'object',
@@ -202,13 +158,6 @@ export default async function (fastify: FastifyInstance) {
     return api.getRuntimeServices(request.params.pid)
   })
 
-  typedFastify.get('/runtimes/:pid/logs', {
-    schema: {
-      params: { type: 'object', properties: { pid: { type: 'number' } }, required: ['pid'] }
-    }
-  }, async ({ params: { pid }, log }) =>
-    getLogsFromReadable(await api.getRuntimeAllLogsStream(pid), log))
-
   typedFastify.get('/runtimes/:pid/openapi/:serviceId', {
     schema: {
       params: { type: 'object', properties: { pid: { type: 'number' }, serviceId: { type: 'string' } }, required: ['pid', 'serviceId'] }
@@ -218,15 +167,11 @@ export default async function (fastify: FastifyInstance) {
   })
 
   typedFastify.post('/runtimes/:pid/restart', {
-    schema: {
-      params: { type: 'object', properties: { pid: { type: 'number' } }, required: ['pid'] },
-      body: { type: 'object' }
-    }
+    schema: { params: pidParamSchema, body: { type: 'object' } }
   }, async (request) => {
     try {
       await api.restartRuntime(request.params.pid)
     } catch (err) {
-      // TODO: restart is currently not working. Apparently, the call to `this.start()` on `@platformatic/runtime/lib/runtime.js` fails. Once it will be fixed, we can remove the catch and the warning log (and leave the function throw as it was before). Monitor this issue to check if it's fixed or not (https://github.com/platformatic/platformatic/issues/3928)
       fastify.log.warn({ err }, 'Issue restarting the runtime')
     }
   })

web/backend/routes/ws.ts

@@ -0,0 +1,44 @@
+import type { WebSocket } from 'ws'
+import split2 from 'split2'
+import { FastifyInstance } from 'fastify'
+import { JsonSchemaToTsProvider } from '@fastify/type-provider-json-schema-to-ts'
+import { RuntimeApiClient } from '@platformatic/control'
+import { PidParam, pidParamSchema } from '../schemas'
+import { pipeline } from 'node:stream/promises'
+
+export default async function (fastify: FastifyInstance) {
+  const typedFastify = fastify.withTypeProvider<JsonSchemaToTsProvider>()
+  const api = new RuntimeApiClient()
+
+  const wsSendAsync = (socket: WebSocket, data: string): Promise<void> => new Promise((resolve, reject) => setTimeout(() => socket.send(data, (err) => (err)
+    ? reject(err)
+    : resolve()
+  ), 100)
+  )
+
+  typedFastify.get<{ Params: PidParam }>('/runtimes/:pid/logs/ws', {
+    schema: { params: pidParamSchema, hide: true },
+    websocket: true
+  }, async (socket, { params: { pid } }) => {
+    try {
+      const clientStream = api.getRuntimeLiveLogsStream(pid)
+
+      socket.on('close', () => {
+        clientStream.destroy()
+      })
+
+      await pipeline(
+        clientStream,
+        split2(),
+        async function * (source: AsyncIterable<string>) {
+          for await (const line of source) {
+            await wsSendAsync(socket, line)
+          }
+        }
+      )
+    } catch (error) {
+      fastify.log.error({ error }, 'fatal error on runtime logs ws')
+      socket.close()
+    }
+  })
+}

web/backend/schemas/index.ts

@@ -67,6 +67,9 @@ export const metricResponseSchema = {
 } as const
 export type MetricsResponse = FromSchema<typeof metricResponseSchema>
 
+export const pidParamSchema = { type: 'object', additionalProperties: false, properties: { pid: { type: 'number' } }, required: ['pid'] } as const
+export type PidParam = FromSchema<typeof pidParamSchema>
+
 export const selectableRuntimeSchema = {
   type: 'object',
   additionalProperties: false,

web/backend/test/helper.ts

@@ -60,7 +60,7 @@ export async function startWatt (t: TestContext): Promise<string> {
       const input = data.toString()
       if (input.includes('Platformatic is now listening at ')) {
         removeListeners()
-        resolve(input)
+        resolve(input.match(/http:\/\/[^:]+:(\d+)/)?.[1] || '0')
       }
     }