/**
 * 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",
    },
  });
}