/**
 * SSE endpoint adapter:
 * writes the runtime/channel event stream as an SSE response, shared by both the Session
 * channel and the user channel.
 *
 * - Response headers: text/event-stream, no-cache, `X-Accel-Buffering: no` (disables
 *   buffering on reverse proxies);
 * - Heartbeat: writes a `: ping` comment line every 20s; the connection is torn down on write failure;
 * - Replay protocol: a fresh subscription without Last-Event-ID does not replay the buffer
 *   (history is served by the messages endpoint) — it only sends the initial events the
 *   caller supplied (pending approvals / hello). With a Last-Event-ID that hits the buffer,
 *   replay resumes from there; on a miss, `resync_required` is sent first, then the
 *   connection continues.
 * Docs: /docs/server-api § "Delivery Guarantees".
 */
import type { Context } from "hono";
import { streamSSE } from "hono/streaming";
import type { ServerEvent } from "../api/types.js";
import type { Channel, ChannelEvent, ChannelListener } from "../runtime/channel.js";

const HEARTBEAT_MS = 20_000;

export interface SseEndpointOptions {
  /** Initial server events to send privately after the subscription is established (in order): pending-approval replay / user channel hello. */
  initialEvents?: ServerEvent[];
}

/** Stream out a Channel as an SSE response. */
export function sseEndpoint(c: Context, channel: Channel, opts: SseEndpointOptions = {}): Response {
  const lastEventIdHeader = c.req.header("Last-Event-ID");
  c.header("X-Accel-Buffering", "no");
  c.header("Cache-Control", "no-cache");

  return streamSSE(c, async (stream) => {
    let closed = false;
    let finish: () => void = () => {};
    const done = new Promise<void>((resolve) => {
      finish = () => {
        if (closed) return;
        closed = true;
        resolve();
      };
    });

    // Write serialization: SSE events must be written fully and in order; any write failure tears down the connection.
    let chain: Promise<void> = Promise.resolve();
    const enqueue = (write: () => Promise<unknown>): void => {
      chain = chain
        .then(async () => {
          if (!closed) await write();
        })
        .catch(() => finish());
    };
    const listener: ChannelListener = (evt: ChannelEvent) => {
      enqueue(() =>
        stream.writeSSE({
          data: evt.data,
          // Event id is an opaque string generated by the channel (`<epoch>-<seq>`), passed through as-is.
          id: evt.id,
          ...(evt.event !== undefined ? { event: evt.event } : {}),
        }),
      );
    };

    const unsubscribe = channel.subscribe(listener);
    // Subscribe first (to avoid dropping events in a race with broadcasts), then replay
    // synchronously — event order: buffered replay (or resync_required) -> initial events
    // (task_state snapshot / pending approvals / hello) -> live stream.
    if (lastEventIdHeader !== undefined) {
      const replay = channel.replayAfter(lastEventIdHeader);
      if (!replay.hit) {
        const resync: ServerEvent = { type: "resync_required" };
        channel.sendTo(listener, resync, "server_event");
      } else {
        for (const evt of replay.events) listener(evt);
      }
    }
    for (const event of opts.initialEvents ?? []) {
      channel.sendTo(listener, event, "server_event");
    }

    const heartbeat = setInterval(() => {
      enqueue(() => stream.write(": ping\n\n"));
    }, HEARTBEAT_MS);

    stream.onAbort(() => finish());
    try {
      await done;
    } finally {
      clearInterval(heartbeat);
      unsubscribe();
    }
  });
}
