feat(phase3c): SSE /api/events/stream + live /events page

apps/web/src/app/api/events/stream/route.ts

@@ -0,0 +1,73 @@
+import { headers } from "next/headers";
+import { auth } from "@/lib/auth";
+import { prisma } from "@/lib/db";
+
+export const dynamic = "force-dynamic";
+export const runtime = "nodejs";
+
+export async function GET() {
+  const session = await auth.api.getSession({ headers: await headers() });
+  if (!session) return new Response("unauthorized", { status: 401 });
+
+  const encoder = new TextEncoder();
+  let cursor = new Date(Date.now() - 60 * 1000);
+  let alive = true;
+
+  const stream = new ReadableStream({
+    async start(controller) {
+      const send = (event: string, data: unknown) => {
+        controller.enqueue(encoder.encode(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`));
+      };
+      send("hello", { ts: Date.now() });
+
+      const tick = async () => {
+        if (!alive) return;
+        try {
+          const rows = await prisma.event.findMany({
+            where: { receivedAt: { gt: cursor } },
+            orderBy: { receivedAt: "asc" },
+            take: 50,
+          });
+          for (const r of rows) {
+            send("event", {
+              id: r.id,
+              streamId: r.streamId,
+              projectKey: r.projectKey,
+              eventType: r.eventType,
+              version: r.version,
+              occurredAt: r.occurredAt.toISOString(),
+              receivedAt: r.receivedAt.toISOString(),
+              payload: r.payload,
+            });
+            cursor = r.receivedAt;
+          }
+        } catch (e) {
+          send("error", { message: e instanceof Error ? e.message : "tick failed" });
+        }
+      };
+
+      await tick();
+      const interval = setInterval(tick, 2000);
+      const heartbeat = setInterval(() => send("ping", { ts: Date.now() }), 25_000);
+
+      const close = () => {
+        alive = false;
+        clearInterval(interval);
+        clearInterval(heartbeat);
+        try { controller.close(); } catch {}
+      };
+      // close after 5 min — client will reconnect via EventSource
+      setTimeout(close, 5 * 60 * 1000);
+    },
+    cancel() { alive = false; },
+  });
+
+  return new Response(stream, {
+    headers: {
+      "Content-Type": "text/event-stream; charset=utf-8",
+      "Cache-Control": "no-cache, no-transform",
+      "Connection": "keep-alive",
+      "X-Accel-Buffering": "no",
+    },
+  });
+}

apps/web/src/app/events/_live.tsx

@@ -0,0 +1,86 @@
+"use client";
+
+import { useEffect, useState } from "react";
+import { Badge } from "@/components/ui/badge";
+import {
+  Table,
+  TableBody,
+  TableCell,
+  TableHead,
+  TableHeader,
+  TableRow,
+} from "@/components/ui/table";
+
+type EventRow = {
+  id: string;
+  streamId: string;
+  projectKey: string;
+  eventType: string;
+  version: number;
+  occurredAt: string;
+  receivedAt: string;
+  payload: unknown;
+};
+
+export function EventsLive({ initial }: { initial: EventRow[] }) {
+  const [rows, setRows] = useState<EventRow[]>(initial);
+  const [status, setStatus] = useState<"connecting" | "live" | "down">("connecting");
+
+  useEffect(() => {
+    const es = new EventSource("/api/events/stream");
+    es.addEventListener("hello", () => setStatus("live"));
+    es.addEventListener("event", (e) => {
+      const ev = JSON.parse((e as MessageEvent).data) as EventRow;
+      setRows((cur) => [ev, ...cur.filter((r) => r.id !== ev.id)].slice(0, 200));
+    });
+    es.onerror = () => setStatus("down");
+    return () => es.close();
+  }, []);
+
+  return (
+    <>
+      <div className="flex items-center justify-between">
+        <p className="text-sm text-muted-foreground">
+          Streaming from Redis (worker → panel DB → SSE). Latest 200 events.
+        </p>
+        <Badge variant={status === "live" ? "default" : "outline"}>{status}</Badge>
+      </div>
+
+      <div className="rounded-md border">
+        <Table>
+          <TableHeader>
+            <TableRow>
+              <TableHead className="w-[180px]">Received</TableHead>
+              <TableHead className="w-[120px]">Project</TableHead>
+              <TableHead>Type</TableHead>
+              <TableHead className="w-[60px]">v</TableHead>
+              <TableHead>Payload</TableHead>
+            </TableRow>
+          </TableHeader>
+          <TableBody>
+            {rows.map((r) => (
+              <TableRow key={r.id}>
+                <TableCell className="font-mono text-xs">
+                  {r.receivedAt.replace("T", " ").slice(0, 19)}
+                </TableCell>
+                <TableCell><Badge variant="outline">{r.projectKey}</Badge></TableCell>
+                <TableCell className="font-mono text-xs">{r.eventType}</TableCell>
+                <TableCell className="font-mono text-xs">{r.version}</TableCell>
+                <TableCell className="max-w-[600px] truncate font-mono text-xs text-muted-foreground">
+                  {JSON.stringify(r.payload)}
+                </TableCell>
+              </TableRow>
+            ))}
+            {rows.length === 0 && (
+              <TableRow>
+                <TableCell colSpan={5} className="text-center text-sm text-muted-foreground">
+                  No events yet. When a spoke publishes to <code>&lt;key&gt;:events</code>, they appear here.
+                </TableCell>
+              </TableRow>
+            )}
+          </TableBody>
+        </Table>
+      </div>
+    </>
+  );
+}

apps/web/src/app/events/page.tsx

@@ -1,15 +1,28 @@
+import { prisma } from "@/lib/db";
 import { PanelShell } from "@/components/panel-shell";
+import { EventsLive } from "./_live";
+
+export const dynamic = "force-dynamic";
+
+export default async function EventsPage() {
+  const initial = await prisma.event.findMany({
+    orderBy: { receivedAt: "desc" },
+    take: 100,
+  });
+  const serialized = initial.map((r) => ({
+    id: r.id,
+    streamId: r.streamId,
+    projectKey: r.projectKey,
+    eventType: r.eventType,
+    version: r.version,
+    occurredAt: r.occurredAt.toISOString(),
+    receivedAt: r.receivedAt.toISOString(),
+    payload: r.payload as unknown,
+  }));
 
-export default function EventsPage() {
   return (
     <PanelShell title="Events">
-      <div className="rounded-lg border p-8">
-        <h2 className="text-lg font-semibold">Cross-project event timeline</h2>
-        <p className="mt-1 text-sm text-muted-foreground">
-          Phase 3 — Redis Streams consumer + SSE push. Spokes publish to their own stream
-          (e.g. <code>sase:events</code>); panel worker persists and broadcasts here.
-        </p>
-      </div>
+      <EventsLive initial={serialized} />
     </PanelShell>
   );
 }