perf: switch to SSE dispatched changes

Previously we would use naive revalidators which would invalidate EVERY SINGLE action loader every 3 seconds, resulting in several fetches. It would also bubble fetches across the layout actions into the individual pages. This new approach selectively has live stores of resources which then poll for changes on the server side and then dispatches updates to the client via a new /events/live SSE endpoint.
2026-08-28 16:07:07 +00:00 · 2026-03-16 23:30:05 -04:00
parent 27f8fa0b42
commit 25dc09e025
21 changed files with 719 additions and 395 deletions
@@ -0,0 +1,60 @@
+import { nodesResource, usersResource } from "~/server/headscale/live-store";
+import log from "~/utils/log";
+
+import type { Route } from "./+types/live";
+
+export async function loader({ request, context }: Route.LoaderArgs) {
+  const principal = await context.auth.require(request);
+  const apiKey = context.auth.getHeadscaleApiKey(principal, context.oidc?.apiKey);
+  const api = context.hsApi.getRuntimeClient(apiKey);
+
+  // Ensure resources are loaded before streaming
+  await Promise.all([
+    context.hsLive.get(nodesResource, api),
+    context.hsLive.get(usersResource, api),
+  ]);
+
+  const stream = new ReadableStream({
+    start(controller) {
+      const encoder = new TextEncoder();
+      const send = (event: string, data: unknown) => {
+        controller.enqueue(encoder.encode(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`));
+      };
+
+      const versions = context.hsLive.getVersions();
+      log.debug("sse", "Client connected, sending hello with versions: %o", versions);
+      send("hello", versions);
+
+      const unsubscribe = context.hsLive.subscribe((resource, version) => {
+        log.debug("sse", "Sending change event: %s v%s", resource, version);
+        send("changed", { resource, version });
+      });
+
+      const heartbeat = setInterval(() => {
+        try {
+          controller.enqueue(encoder.encode(": heartbeat\n\n"));
+        } catch {
+          clearInterval(heartbeat);
+        }
+      }, 30_000);
+
+      request.signal.addEventListener("abort", () => {
+        log.debug("sse", "Client disconnected");
+        unsubscribe();
+        clearInterval(heartbeat);
+        try {
+          controller.close();
+        } catch {}
+      });
+    },
+  });
+
+  return new Response(stream, {
+    headers: {
+      "Content-Type": "text/event-stream",
+      "Cache-Control": "no-cache",
+      Connection: "keep-alive",
+      "X-Accel-Buffering": "no",
+    },
+  });
+}