OpenAI SSE 스트리밍 끊김·중복 토큰 재시도 패턴

// app/api/stream/route.ts
import { NextRequest } from "next/server";

export const runtime = "nodejs";

type StreamEvent =
  | { type: "delta"; requestId: string; delta: string }
  | { type: "checkpoint"; requestId: string; cursor: number; text: string }
  | { type: "done"; requestId: string; cursor: number; text: string };

function ssePack(event: StreamEvent, id?: string) {
  const lines: string[] = [];
  if (id) lines.push(`id: ${id}`);
  lines.push(`event: ${event.type}`);
  lines.push(`data: ${JSON.stringify(event)}`);
  lines.push("\n");
  return lines.join("\n");
}

export async function POST(req: NextRequest) {
  const { requestId, prompt, resumeCursor } = await req.json();

  const stream = new ReadableStream({
    async start(controller) {
      const encoder = new TextEncoder();

      let text = "";
      let cursor = 0;
      let seq = 0;

      // resumeCursor가 있으면 서버 저장소에서 text를 복원하는 게 이상적입니다.
      // 여기서는 단순화를 위해 빈 문자열부터 시작합니다.
      if (typeof resumeCursor === "number" && resumeCursor > 0) {
        // TODO: requestId 기준으로 저장된 누적 결과를 불러와 cursor 위치까지 복원
      }

      // SSE keep-alive: 중간 장비 idle timeout 회피용
      const ka = setInterval(() => {
        controller.enqueue(encoder.encode(`event: ping\ndata: {}\n\n`));
      }, 15000);

      try {
        // TODO: OpenAI 스트리밍 호출로 교체
        // for await (const delta of openAIStream(prompt)) { ... }
        const fake = ["OpenAI ", "SSE ", "스트리밍 ", "예시입니다."];

        for (const delta of fake) {
          text += delta;
          cursor = text.length;
          seq += 1;

          controller.enqueue(
            encoder.encode(ssePack({ type: "delta", requestId, delta }, `${requestId}:${seq}`))
          );

          // 일정 주기마다 checkpoint를 흘려보냄
          if (seq % 2 === 0) {
            controller.enqueue(
              encoder.encode(
                ssePack(
                  { type: "checkpoint", requestId, cursor, text },
                  `${requestId}:cp:${cursor}`
                )
              )
            );
            // TODO: requestId 기준으로 text/cursor를 저장(Redis 등)
          }
        }

        controller.enqueue(
          encoder.encode(
            ssePack({ type: "done", requestId, cursor, text }, `${requestId}:done:${cursor}`)
          )
        );
      } finally {
        clearInterval(ka);
        controller.close();
      }
    },
  });

  return new Response(stream, {
    headers: {
      "Content-Type": "text/event-stream",
      "Cache-Control": "no-cache, no-transform",
      Connection: "keep-alive",
    },
  });
}

function appendWithOverlap(base: string, incoming: string, maxOverlap = 200) {
  if (!incoming) return base;
  if (!base) return incoming;

  // incoming이 base의 접미사/접두사와 겹치는 최대 길이를 찾는다.
  const start = Math.max(0, base.length - maxOverlap);
  const tail = base.slice(start);

  let best = 0;
  const limit = Math.min(tail.length, incoming.length);

  for (let k = 1; k <= limit; k += 1) {
    if (tail.slice(tail.length - k) === incoming.slice(0, k)) {
      best = k;
    }
  }

  return base + incoming.slice(best);
}

type Checkpoint = { cursor: number; text: string };

async function streamWithRetry(url: string, body: any, onText: (t: string) => void) {
  let attempt = 0;
  let checkpoint: Checkpoint = { cursor: 0, text: "" };

  while (attempt < 5) {
    const controller = new AbortController();
    const timeout = setTimeout(() => controller.abort(), 65000);

    try {
      const res = await fetch(url, {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify({ ...body, resumeCursor: checkpoint.cursor }),
        signal: controller.signal,
      });

      if (!res.ok || !res.body) throw new Error(`bad response: ${res.status}`);

      const reader = res.body.getReader();
      const decoder = new TextDecoder();
      let buf = "";

      while (true) {
        const { value, done } = await reader.read();
        if (done) break;

        buf += decoder.decode(value, { stream: true });

        // SSE는 빈 줄로 이벤트가 구분된다.
        let idx;
        while ((idx = buf.indexOf("\n\n")) !== -1) {
          const raw = buf.slice(0, idx);
          buf = buf.slice(idx + 2);

          const eventLine = raw.split("\n").find((l) => l.startsWith("event:"));
          const dataLine = raw.split("\n").find((l) => l.startsWith("data:"));
          if (!dataLine) continue;

          const event = eventLine ? eventLine.replace("event:", "").trim() : "message";
          const data = JSON.parse(dataLine.replace("data:", "").trim());

          if (event === "delta") {
            // delta는 중복 가능성이 높으므로 overlap 기반으로 붙인다.
            checkpoint.text = appendWithOverlap(checkpoint.text, data.delta);
            checkpoint.cursor = checkpoint.text.length;
            onText(checkpoint.text);
          }

          if (event === "checkpoint") {
            // 서버가 준 누적 텍스트가 있으면 그것을 신뢰(가장 안전)
            checkpoint.text = data.text;
            checkpoint.cursor = data.cursor;
            onText(checkpoint.text);
          }

          if (event === "done") {
            checkpoint.text = data.text;
            checkpoint.cursor = data.cursor;
            onText(checkpoint.text);
            clearTimeout(timeout);
            return;
          }
        }
      }

      // 여기로 오면 연결이 정상 종료됐는데 done을 못 받은 상태일 수 있다.
      throw new Error("stream ended without done");
    } catch (e) {
      attempt += 1;
      clearTimeout(timeout);

      const backoff = Math.min(2500, 300 * Math.pow(2, attempt - 1));
      await new Promise((r) => setTimeout(r, backoff));
      continue;
    }
  }

  throw new Error("stream failed after retries");
}