/**
 * Live transcript streaming for the dashboard (epic #292, slice 4/5 — #296).
 *
 * The run detail view could show what a run *did*; this makes it possible to
 * watch one happening, which is the difference between reading a postmortem and
 * noticing a worker is stuck at turn eight.
 *
 * ## It follows the file, not the daemon
 *
 * `tailRun` is the follow — the same function `omp-conductor tail` runs, with
 * the same 1s re-stat cadence, the same read-from-byte-zero behaviour (attaching
 * to a run ten turns in and showing nothing until turn eleven is not watching
 * it), and the same advisor-transcript discovery. This module only turns its
 * lines into Server-Sent Events.
 *
 * That is what makes the acceptance criterion "neither blocks the daemon" true by
 * construction rather than by measurement: nothing here talks to the daemon at
 * all. Two browsers on one run are two read-only file follows, and the daemon
 * does not know either of them exists.
 *
 * ## Why SSE and not a websocket
 *
 * The data is one-directional, text, and append-only, which is exactly what SSE
 * is. A websocket would add a handshake, a framing layer and a dependency for a
 * feature whose whole payload is lines of text — and the epic's constraint is no
 * runtime dependencies. `EventSource` is in every browser and reconnects on its
 * own.
 */

import { tailRun } from "../commands/tail.ts";
import type { Store } from "../types.ts";

/** How often a comment frame is written when nothing else is. */
export const KEEPALIVE_MS = 15_000;

/** Injected so a test drives the whole stream with no file and no clock. */
export interface TranscriptStreamDeps {
  /**
   * The follow itself. Defaults to `tailRun`, and a test replaces it — the point
   * of the seam is that this module owns the *framing*, not the following.
   */
  follow?: (
    project: string,
    issue: number,
    opts: { write: (line: string) => void; signal: AbortSignal; store?: Store },
  ) => Promise<void>;
  /**
   * A store for the follow to read the run row through. Optional, and NOT closed
   * here: `tailRun` closes whatever store it is given, so opening one in this
   * module would double-close it. Production passes none and the follow opens
   * (and closes) its own, exactly as `omp-conductor tail` does.
   */
  store?: Store;
  keepaliveMs?: number;
}

/**
 * One SSE frame.
 *
 * Exported because the framing is the contract a browser parses, and a test that
 * asserts on assembled frames is asserting on the thing `EventSource` sees. Data
 * is split per line so a payload containing a newline cannot terminate the frame
 * early — the one way a text stream corrupts an SSE connection.
 */
export function sseFrame(event: string | undefined, data: string): string {
  const lines = data.split("\n").map((line) => `data: ${line}`);
  return `${event === undefined ? "" : `event: ${event}\n`}${lines.join("\n")}\n\n`;
}

/**
 * Stream one run's transcript as Server-Sent Events.
 *
 * Three terminal shapes, and each is said out loud rather than inferred from a
 * closed connection:
 *
 *  - **`error`** — there is no transcript to follow (no run, or a claimed run
 *    whose session never opened one). `tail` throws in exactly that case, and
 *    the message is its message. A silent empty stream would read as "a quiet
 *    worker", which is the wrong conclusion to invite.
 *  - **`end`** — the follow returned, which `tailRun` does once the run row has
 *    left the live states and stayed quiet. Its last written line is
 *    `run ended: <state>`, so the terminal state is already in the stream.
 *  - **the client vanishing** — the request's signal aborts the follow, which is
 *    why `tailRun` takes one: a closed tab must not leave a poll loop holding
 *    file descriptors until the run happens to finish.
 */
export function transcriptStreamResponse(
  project: string,
  issue: number,
  req: Request,
  deps: TranscriptStreamDeps = {},
): Response {
  const follow = deps.follow ?? defaultFollow;
  const keepaliveMs = deps.keepaliveMs ?? KEEPALIVE_MS;
  const controller = new AbortController();
  // Both directions matter: the client going away aborts the follow, and the
  // follow finishing means nobody is waiting on the client any more.
  req.signal.addEventListener("abort", () => controller.abort(), { once: true });

  const encoder = new TextEncoder();
  const body = new ReadableStream<Uint8Array>({
    start(sink) {
      let closed = false;
      const send = (frame: string): void => {
        if (closed) return;
        try {
          sink.enqueue(encoder.encode(frame));
        } catch {
          // The client hung up between our abort check and this write. Stop
          // following rather than throwing inside a stream nobody reads.
          closed = true;
          controller.abort();
        }
      };
      // A comment frame — `EventSource` ignores it, proxies and load balancers
      // do not, which is the whole reason it exists.
      const keepalive = setInterval(() => send(": keepalive\n\n"), keepaliveMs);
      const finish = (): void => {
        if (closed) return;
        closed = true;
        clearInterval(keepalive);
        try {
          sink.close();
        } catch {
          // Already closed by the client; nothing to do.
        }
      };

      void (async () => {
        try {
          await follow(project, issue, {
            write: (line) => send(sseFrame(undefined, line)),
            signal: controller.signal,
            ...(deps.store === undefined ? {} : { store: deps.store }),
          });
          if (!controller.signal.aborted) send(sseFrame("end", "stream closed"));
        } catch (err) {
          // `tail`'s own wording: "no run recorded for #N", or
          // "no transcript yet (state: claimed)". Both are answers, not faults.
          send(sseFrame("error", err instanceof Error ? err.message : String(err)));
        } finally {
          finish();
        }
      })();
    },
    cancel() {
      controller.abort();
    },
  });

  return new Response(body, {
    headers: {
      "content-type": "text/event-stream; charset=utf-8",
      // No buffering anywhere in between, or the 1s cadence becomes minutes.
      "cache-control": "no-cache, no-transform",
      connection: "keep-alive",
      "x-accel-buffering": "no",
    },
  });
}

/** `tailRun` with the stream's store, so one connection opens one reader. */
async function defaultFollow(
  project: string,
  issue: number,
  opts: { write: (line: string) => void; signal: AbortSignal; store?: Store },
): Promise<void> {
  await tailRun(project, issue, {
    write: opts.write,
    signal: opts.signal,
    ...(opts.store === undefined ? {} : { store: opts.store }),
  });
}