/**
 * 统一 SSE（Server-Sent Events）流封装
 *
 * 用法：
 *   import { sseStream } from "../_shared/sse.ts";
 *
 *   return sseStream(ctx, async ({ send, close }) => {
 *     send({ type: "delta", text: "hello " });
 *     send({ type: "delta", text: "world" });
 *     send({ type: "done" });
 *     close();
 *   });
 *
 * 自动处理：
 *  - SSE 标准 headers（Content-Type / Cache-Control / Connection）
 *  - CORS（与 _shared/cors.ts 一致）
 *  - 30s 心跳防止网关断流
 *  - 异常捕获 → 以 SSE event=error 帧上报，避免 Response Body 中段截断
 *  - x-trace-id 透传
 */

import { withCorsHeaders } from "./cors.ts";
import type { FnContext } from "./middleware.ts";

const HEARTBEAT_MS = 30_000;

export interface SseHandle {
  /** 推送一个 JSON 数据帧（自动 stringify） */
  send: (data: unknown, event?: string) => void;
  /** 推送原始字符串（已自带 SSE 帧格式） */
  raw: (chunk: string) => void;
  /** 主动结束 */
  close: () => void;
  /** 是否已关闭 */
  readonly closed: boolean;
}

export type SseProducer = (handle: SseHandle) => Promise<void> | void;

export function sseStream(ctx: FnContext, producer: SseProducer): Response {
  const encoder = new TextEncoder();

  let closed = false;
  let heartbeat: number | undefined;

  const stream = new ReadableStream<Uint8Array>({
    async start(controller) {
      const safeEnqueue = (chunk: string) => {
        if (closed) return;
        try {
          controller.enqueue(encoder.encode(chunk));
        } catch (e) {
          ctx.error("sseStream enqueue failed:", e);
          cleanup();
        }
      };

      const handle: SseHandle = {
        send: (data, event) => {
          const payload = typeof data === "string" ? data : JSON.stringify(data);
          const prefix = event ? `event: ${event}\n` : "";
          safeEnqueue(`${prefix}data: ${payload}\n\n`);
        },
        raw: (chunk) => safeEnqueue(chunk),
        close: () => cleanup(),
        get closed() { return closed; },
      };

      const cleanup = () => {
        if (closed) return;
        closed = true;
        if (heartbeat !== undefined) clearInterval(heartbeat);
        try { controller.close(); } catch { /* already closed */ }
      };

      // 心跳：注释帧（: ping），不会作为 data 触发 onmessage
      heartbeat = setInterval(() => {
        if (closed) return;
        safeEnqueue(`: ping ${Date.now()}\n\n`);
      }, HEARTBEAT_MS) as unknown as number;

      try {
        await producer(handle);
      } catch (err) {
        ctx.error("sseStream producer error:", err);
        handle.send(
          {
            error: true,
            code: "STREAM_ERROR",
            message: err instanceof Error ? err.message : String(err),
            trace_id: ctx.traceId,
          },
          "error",
        );
      } finally {
        cleanup();
      }
    },
    cancel() {
      closed = true;
      if (heartbeat !== undefined) clearInterval(heartbeat);
    },
  });

  return new Response(stream, {
    status: 200,
    headers: withCorsHeaders({
      "Content-Type": "text/event-stream; charset=utf-8",
      "Cache-Control": "no-cache, no-transform",
      "Connection": "keep-alive",
      "X-Accel-Buffering": "no",
      "x-trace-id": ctx.traceId,
    }),
  });
}
