initial commit

apps/backend/src/modules/quotes/handlers/quoteEvents.ts

@@ -0,0 +1,49 @@
+import type { Context } from "hono";
+import { eventBus } from "../../../lib/event-bus";
+
+export function quoteEvents(c: Context) {
+  const cleanups: (() => void)[] = [];
+  let isCancelled = false;
+
+  function cleanup() {
+    if (isCancelled) return;
+    isCancelled = true;
+    cleanups.forEach((fn) => fn());
+  }
+
+  const stream = new ReadableStream({
+    start(controller) {
+      const unsubscribe = eventBus.on("quotes-updated", () => {
+        const data = "event: quotes-updated\ndata: updated\n\n";
+        try {
+          controller.enqueue(new TextEncoder().encode(data));
+        } catch {
+          /* stream closed */
+        }
+      });
+      cleanups.push(unsubscribe);
+
+      const keepAlive = setInterval(() => {
+        try {
+          controller.enqueue(new TextEncoder().encode(": keepalive\n\n"));
+        } catch {
+          clearInterval(keepAlive);
+        }
+      }, 5_000);
+      cleanups.push(() => clearInterval(keepAlive));
+
+      c.req.raw.signal.addEventListener("abort", cleanup);
+    },
+    cancel() {
+      cleanup();
+    },
+  });
+
+  return new Response(stream, {
+    headers: {
+      "Content-Type": "text/event-stream",
+      "Cache-Control": "no-cache",
+      "Connection": "keep-alive",
+    },
+  });
+}