benjivers/notes Public Public: anyone can read this; nobody can change it.

/**

 * Realtime without a framework. Actions `emit` what changed; `GET /events`

 * streams it to every open page as server-sent events, numbered, with the

 * last hundred kept so a client that reconnects with `Last-Event-ID` (or

 * `?since=`) hears what it missed. `static/live.js` turns the stream into

 * DOM events. The bus is this process's: the terminal is another process,

 * and its writes show on the next load rather than live.

 *

 * A vault is private, so an event can be addressed: `emit(type, data, to)`

 * reaches only the streams opened by the person whose id is `to`. An event

 * with no `to` reaches everyone.

 */

import type { Who } from "./who.ts";

export interface LiveEvent {

  readonly id: number;

  readonly type: string;

  readonly data: unknown;

  /** Whose streams hear it; everyone's when absent. */

  readonly to?: string;

}

const KEEP = 100;

const HEARTBEAT_MS = 25_000;

const kept: LiveEvent[] = [];

const listeners = new Set<(event: LiveEvent) => void>();

let last = 0;

export function emit(type: string, data: unknown, to?: string): LiveEvent {

  const event: LiveEvent = { id: ++last, type, data, ...(to === undefined ? {} : { to }) };

  kept.push(event);

  if (kept.length > KEEP) kept.shift();

  for (const listen of listeners) listen(event);

  return event;

}

export function subscribe(listen: (event: LiveEvent) => void): () => void {

  listeners.add(listen);

  return () => listeners.delete(listen);

}

/** `GET /events`: a stream that replays what `since` missed, then follows. */

export function events(request: Request, who: Who): Response {

  const hears = (event: LiveEvent): boolean =>

    event.to === undefined || (who.role !== "anon" && event.to === who.id);

  const since = Number(

    request.headers.get("last-event-id") ??

      new URL(request.url).searchParams.get("since") ??

      0,

  );

  const encoder = new TextEncoder();

  let stop = (): void => {};

  const stream = new ReadableStream<Uint8Array>({

    start(controller) {

      const write = (text: string): void => {

        try {

          controller.enqueue(encoder.encode(text));

        } catch {

          stop();

        }

      };

      const send = (event: LiveEvent): void => {

        if (!hears(event)) return;

        write(

          `id: ${event.id}\ndata: ${JSON.stringify({ type: event.type, data: event.data })}\n\n`,

        );

      };

      const heartbeat = setInterval(() => write(":\n\n"), HEARTBEAT_MS);

      const unsubscribe = subscribe(send);

      stop = () => {

        clearInterval(heartbeat);

        unsubscribe();

        try {

          controller.close();

        } catch {}

      };

      request.signal.addEventListener("abort", stop);

      write("retry: 2000\n\n");

      for (const event of kept) if (event.id > since) send(event);

    },

    cancel() {

      stop();

    },

  });

  return new Response(stream, {

    headers: {

      "content-type": "text/event-stream",

      "cache-control": "no-store",

      "x-accel-buffering": "no",

    },

  });

}