From 9919b12aab0c151796f6bc0c894311aafbaee002 Mon Sep 17 00:00:00 2001
From: Jamie Brunton <jamie@jbrunton.co.uk>
Date: Sat, 8 Aug 2026 15:00:02 +0100
Subject: [PATCH] feat(web): bridge logs to typed event stream

---
 src/web/log-broadcaster.ts    | 16 +++++++++++++++-
 src/web/routes/logs.ts        |  1 +
 tests/log-broadcaster.test.ts | 25 +++++++++++++++++++++++++
 3 files changed, 41 insertions(+), 1 deletion(-)
 create mode 100644 tests/log-broadcaster.test.ts

diff --git a/src/web/log-broadcaster.ts b/src/web/log-broadcaster.ts
index a8a31e0..034210f 100644
--- a/src/web/log-broadcaster.ts
+++ b/src/web/log-broadcaster.ts
@@ -1,5 +1,7 @@
-import type { Request, Response } from 'express';
+import type { Request, RequestHandler, Response } from 'express';
 import type { LogEntry } from './types.js';
+import { RealtimeEventHub } from './events/index.js';
+import { createExpressSseHandler } from './events/express-sse-adapter.js';
 
 const MAX_LOGS = 500;
 const INITIAL_LOGS_ON_CONNECT = 50;
@@ -11,6 +13,13 @@ const INITIAL_LOGS_ON_CONNECT = 50;
 export class LogBroadcaster {
   private entries: LogEntry[] = [];
   private clients = new Set<Response>();
+  private readonly eventHub: RealtimeEventHub;
+  readonly events: RequestHandler;
+
+  constructor(eventHub?: RealtimeEventHub) {
+    this.eventHub = eventHub ?? new RealtimeEventHub({ capacity: MAX_LOGS });
+    this.events = createExpressSseHandler({ hub: this.eventHub });
+  }
 
   push(level: LogEntry['level'], message: string, data?: unknown): void {
     const entry: LogEntry = { timestamp: Date.now(), level, message, data };
@@ -19,6 +28,10 @@ export class LogBroadcaster {
       this.entries = this.entries.slice(-MAX_LOGS);
     }
 
+    if (!this.eventHub.closed) {
+      this.eventHub.publish('log.created', { level, message, data });
+    }
+
     this.fanout(JSON.stringify({ type: 'log', ...entry }));
   }
 
@@ -76,5 +89,6 @@ export class LogBroadcaster {
       }
     }
     this.clients.clear();
+    this.eventHub.close();
   }
 }
diff --git a/src/web/routes/logs.ts b/src/web/routes/logs.ts
index ec4c138..de12588 100644
--- a/src/web/routes/logs.ts
+++ b/src/web/routes/logs.ts
@@ -4,6 +4,7 @@ import type { WebContext } from '../context.js';
 export function createLogsRouter(ctx: WebContext): Router {
   const router = Router();
   router.get('/logs/stream', ctx.broadcaster.stream);
+  router.get('/events/stream', ctx.broadcaster.events);
   router.get('/logs', ctx.broadcaster.recent);
   return router;
 }
diff --git a/tests/log-broadcaster.test.ts b/tests/log-broadcaster.test.ts
new file mode 100644
index 0000000..4d914c5
--- /dev/null
+++ b/tests/log-broadcaster.test.ts
@@ -0,0 +1,25 @@
+import { describe, expect, it } from 'vitest';
+import { RealtimeEventHub, type RealtimeEvent } from '../src/web/events/index.js';
+import { LogBroadcaster } from '../src/web/log-broadcaster.js';
+
+describe('LogBroadcaster realtime bridge', () => {
+  it('publishes each log through the typed event hub', () => {
+    const hub = new RealtimeEventHub({ capacity: 10 });
+    const received: RealtimeEvent[] = [];
+    hub.subscribe((event) => received.push(event));
+    const broadcaster = new LogBroadcaster(hub);
+
+    broadcaster.push('warn', 'Connection is stale', { attempts: 2 });
+
+    expect(received).toEqual([
+      expect.objectContaining({
+        type: 'log.created',
+        payload: {
+          level: 'warn',
+          message: 'Connection is stale',
+          data: { attempts: 2 },
+        },
+      }),
+    ]);
+  });
+});