feat: SSE live ping stream for monitors

apps/web/src/routes/pings.ts

@@ -1,8 +1,48 @@
 import { Elysia, t } from "elysia";
 import sql from "../db";
 
-// Internal-only: called by the Rust monitor runner
+// ── SSE bus ───────────────────────────────────────────────────────────────────
+type SSEController = ReadableStreamDefaultController<Uint8Array>;
+const bus = new Map<string, Set<SSEController>>();
+const enc = new TextEncoder();
+
+function publish(monitorId: string, data: object) {
+  const subs = bus.get(monitorId);
+  if (!subs?.size) return;
+  const msg = enc.encode(`data: ${JSON.stringify(data)}\n\n`);
+  for (const ctrl of subs) {
+    try { ctrl.enqueue(msg); } catch { subs.delete(ctrl); }
+  }
+}
+
+function makeSSEStream(monitorId: string): Response {
+  let ctrl: SSEController;
+  const stream = new ReadableStream<Uint8Array>({
+    start(c) {
+      ctrl = c;
+      if (!bus.has(monitorId)) bus.set(monitorId, new Set());
+      bus.get(monitorId)!.add(ctrl);
+      ctrl.enqueue(enc.encode(": connected\n\n"));
+    },
+    cancel() {
+      bus.get(monitorId)?.delete(ctrl);
+      if (bus.get(monitorId)?.size === 0) bus.delete(monitorId);
+    },
+  });
+  return new Response(stream, {
+    headers: {
+      "Content-Type":      "text/event-stream",
+      "Cache-Control":     "no-cache",
+      "Connection":        "keep-alive",
+      "X-Accel-Buffering": "no",
+    },
+  });
+}
+
+// ── Routes ────────────────────────────────────────────────────────────────────
 export const ingest = new Elysia()
+
+  // Internal: called by Rust monitor runner
   .post("/internal/ingest", async ({ body, headers, error }) => {
     const token = headers["x-monitor-token"];
     if (token !== process.env.MONITOR_TOKEN) return error(401, { error: "Unauthorized" });
@@ -10,7 +50,7 @@ export const ingest = new Elysia()
     const meta = body.meta ? { ...body.meta } : {};
     if (body.cert_expiry_days != null) meta.cert_expiry_days = body.cert_expiry_days;
 
-    await sql`
+    const [ping] = await sql`
       INSERT INTO pings (monitor_id, status_code, latency_ms, up, error, meta)
       VALUES (
         ${body.monitor_id},
@@ -20,7 +60,10 @@ export const ingest = new Elysia()
         ${body.error ?? null},
         ${Object.keys(meta).length > 0 ? sql.json(meta) : null}
       )
+      RETURNING *
     `;
+
+    publish(body.monitor_id, ping);
     return { ok: true };
   }, {
     body: t.Object({
@@ -33,4 +76,28 @@ export const ingest = new Elysia()
       meta:             t.Optional(t.Any()),
     }),
     detail: { hide: true },
-  });
+  })
+
+  // SSE: stream live pings — auth via Bearer header OR ?auth= query param
+  // (EventSource doesn't support custom headers, hence the query param fallback)
+  .get("/monitors/:id/stream", async ({ params, headers, query, error }) => {
+    const key = headers["authorization"]?.replace("Bearer ", "").trim()
+      ?? (query.auth as string | undefined);
+
+    if (!key) return error(401, { error: "Unauthorized" });
+
+    // Resolve account from primary key or sub-key
+    const [acc] = await sql`SELECT id FROM accounts WHERE id = ${key}`;
+    const accountId = acc?.id ?? (
+      await sql`SELECT account_id FROM api_keys WHERE id = ${key}`.then(r => r[0]?.account_id)
+    );
+    if (!accountId) return error(401, { error: "Unauthorized" });
+
+    // Verify ownership
+    const [monitor] = await sql`
+      SELECT id FROM monitors WHERE id = ${params.id} AND account_id = ${accountId}
+    `;
+    if (!monitor) return error(404, { error: "Not found" });
+
+    return makeSSEStream(params.id);
+  }, { detail: { hide: true } });