headplane/app/routes/util/live.ts

import { headscaleLiveStoreContext, requestApiContext } from "~/server/context";
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 getRequestApi = context.get(requestApiContext);
  const headscaleLiveStore = context.get(headscaleLiveStoreContext);

  const { api } = await getRequestApi(request);

  // Ensure resources are loaded before streaming
  await Promise.all([
    headscaleLiveStore.get(nodesResource, api),
    headscaleLiveStore.get(usersResource, api),
  ]);

  const stream = new ReadableStream({
    start(controller) {
      let closed = false;
      const encoder = new TextEncoder();
      const send = (event: string, data: unknown) => {
        if (closed) return;
        try {
          controller.enqueue(encoder.encode(`event: ${event}\ndata: ${JSON.stringify(data)}\n\n`));
        } catch {
          closed = true;
        }
      };

      const versions = headscaleLiveStore.getVersions();
      log.debug("sse", "Client connected, sending hello with versions: %o", versions);
      send("hello", versions);

      const unsubscribe = headscaleLiveStore.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");
        closed = true;
        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",
    },
  });
}