/**
 * Streaming response primitives.
 *
 * `streamResponse` turns a (sync or async) iterable of strings/bytes into a
 * streaming `Response` — the basis for streaming SSR (send the shell, then flush
 * page chunks as they render) and any progressively-generated output. `sse`
 * builds a Server-Sent Events stream from an async iterable of events.
 *
 * API routes and pages can already return a `Response` with a `ReadableStream`
 * body and the framework streams it unbuffered; these helpers just make the
 * common cases ergonomic.
 */

export interface StreamResponseInit {
  status?: number;
  headers?: HeadersInit;
  /** Content-Type; default "text/html; charset=utf-8". */
  contentType?: string;
}

type Chunk = string | Uint8Array;
type ChunkSource = Iterable<Chunk> | AsyncIterable<Chunk>;

/** Build a streaming Response from an (async) iterable of chunks. */
export function streamResponse(source: ChunkSource, init: StreamResponseInit = {}): Response {
  const encoder = new TextEncoder();
  const iterator = getIterator(source);

  const stream = new ReadableStream<Uint8Array>({
    async pull(controller) {
      try {
        const { done, value } = await iterator.next();
        if (done) {
          controller.close();
          return;
        }
        controller.enqueue(typeof value === "string" ? encoder.encode(value) : value);
      } catch (err) {
        controller.error(err);
      }
    },
    async cancel() {
      await iterator.return?.(undefined);
    },
  });

  const headers = new Headers(init.headers);
  if (!headers.has("content-type")) {
    headers.set("content-type", init.contentType ?? "text/html; charset=utf-8");
  }
  // Tell the server's compressor (and proxies) not to buffer/transform a stream.
  if (!headers.has("cache-control")) headers.set("cache-control", "no-transform");
  return new Response(stream, { status: init.status ?? 200, headers });
}

export interface ServerSentEvent {
  data: string;
  event?: string;
  id?: string;
  /** Client reconnection hint in milliseconds. */
  retry?: number;
}

/** Build a Server-Sent Events (text/event-stream) Response from events. */
export function sse(source: Iterable<ServerSentEvent> | AsyncIterable<ServerSentEvent>): Response {
  const iterator = getIterator(source);
  async function* frames(): AsyncGenerator<string> {
    for (;;) {
      const { done, value } = await iterator.next();
      if (done) return;
      yield formatEvent(value);
    }
  }
  return streamResponse(frames(), {
    contentType: "text/event-stream",
    headers: { "cache-control": "no-cache, no-transform", connection: "keep-alive" },
  });
}

function formatEvent(e: ServerSentEvent): string {
  let out = "";
  if (e.event) out += `event: ${e.event}\n`;
  if (e.id) out += `id: ${e.id}\n`;
  if (e.retry !== undefined) out += `retry: ${Math.floor(e.retry)}\n`;
  for (const line of e.data.split("\n")) out += `data: ${line}\n`;
  return out + "\n";
}

function getIterator<T>(source: Iterable<T> | AsyncIterable<T>): AsyncIterator<T> | Iterator<T> {
  const asAsync = (source as AsyncIterable<T>)[Symbol.asyncIterator];
  if (typeof asAsync === "function") return asAsync.call(source);
  return (source as Iterable<T>)[Symbol.iterator]();
}