SSEストリーミングにJSON フォールバックを組み合わせる：プロダクションで安定動作させる実装パターン

import json
from typing import Any

def sse_event(name: str, payload: Any) -> str:
    """SSEイベントをフォーマットする。"""
    data = json.dumps(payload) if not isinstance(payload, str) else payload
    return f"event: {name}\ndata: {data}\n\n"

def sse_error(message: str, code: str | None = None) -> str:
    payload = {"message": message}
    if code:
        payload["code"] = code
    return sse_event("error", payload)

def sse_meta(**kwargs: Any) -> str:
    return sse_event("meta", {k: v for k, v in kwargs.items() if v is not None})

headers = {
    "content-type": "text/event-stream",
    "cache-control": "no-cache",
    "connection": "keep-alive",
    "transfer-encoding": "chunked",
    "X-Accel-Buffering": "no",
}
return StreamingResponse(event_gen(), headers=headers, media_type="text/event-stream")

import { NextRequest, NextResponse } from "next/server";

export async function POST(req: NextRequest) {
  const body = await req.json();
  const traceId = req.headers.get("x-trace-id") ?? crypto.randomUUID();

  const res = await fetch("http://127.0.0.1:8765/v1/chat/stream", {
    method: "POST",
    headers: {
      "content-type": "application/json",
      "x-trace-id": traceId,
    },
    body: JSON.stringify(body),
  });

  if (!res.ok || !res.body) {
    return NextResponse.json(
      { error: "Backend error" },
      { status: res.status },
    );
  }

  // ストリーミングボディをそのまま転送（読み取らない）
  return new Response(res.body, {
    headers: {
      "content-type": "text/event-stream",
      "cache-control": "no-cache",
      "X-Accel-Buffering": "no",
      "x-trace-id": traceId,
    },
  });
}

function useConvert(url: string) {
  const esRef = useRef<EventSource | null>(null);
  const [markdown, setMarkdown] = useState("");
  const [busy, setBusy] = useState(false);

  const start = useCallback(() => {
    setBusy(true);
    setMarkdown("");

    // SSEストリーミング接続
    const es = new EventSource(`/api/confluence/convert/stream?url=${encodeURIComponent(url)}`);
    esRef.current = es;

    es.addEventListener("token", (ev) => {
      const { content } = JSON.parse((ev as MessageEvent).data);
      setMarkdown((prev) => prev + content);
    });

    es.addEventListener("complete", () => {
      es.close();
      esRef.current = null;
      setBusy(false);
    });

    // SSE失敗時 → JSONフォールバック
    es.addEventListener("error", () => {
      es.close();
      esRef.current = null;
      fallbackJson();
    });
  }, [url]);

  // JSONフォールバック（ユーザー操作不要で自動実行）
  async function fallbackJson() {
    try {
      const res = await fetch("/api/confluence/convert", {
        method: "POST",
        headers: { "content-type": "application/json" },
        body: JSON.stringify({ url }),
      });
      const data = await res.json();
      setMarkdown(data.markdown || "");
    } finally {
      setBusy(false);
    }
  }

  // クリーンアップ
  useEffect(() => {
    return () => {
      try { esRef.current?.close(); } catch {}
      esRef.current = null;
    };
  }, []);

  return { markdown, busy, start };
}

async function searchMultiple(tasks: SearchTask[]) {
  let remaining = tasks.length;
  const errors: string[] = [];

  const doneOne = () => {
    remaining -= 1;
    if (remaining <= 0) {
      setBusy(false);
      if (errors.length) setError(errors.join(" | "));
    }
  };

  for (const task of tasks) {
    const es = new EventSource(
      `/api/jira/search/stream?fixVersion=${encodeURIComponent(task.version)}&instance=${task.instance}`
    );

    es.addEventListener("issue", (ev) => {
      const issue = JSON.parse((ev as MessageEvent).data);
      addIssue({ ...issue, source: task.label });
    });

    es.addEventListener("complete", () => {
      es.close();
      doneOne();
    });

    es.addEventListener("error", () => {
      es.close();
      // SSE失敗 → JSONフォールバック → 完了カウント
      void fallbackJson(task)
        .then((errMsg) => {
          if (errMsg) errors.push(`${task.label}: ${errMsg}`);
        })
        .finally(doneOne);
    });
  }
}

useEffect(() => {
  return () => {
    // 各クリーンアップを個別にtry-catch
    // 1つの失敗が他のクリーンアップをブロックしない
    try { esRef.current?.close(); } catch {}
    esRef.current = null;

    try { ocrAbortRef.current?.abort(); } catch {}
    ocrAbortRef.current = null;

    try {
      if (objectUrlRef.current) {
        URL.revokeObjectURL(objectUrlRef.current);
      }
    } catch {}
    objectUrlRef.current = null;
  };
}, []);

X-Accel-Buffering: no
cache-control: no-cache
transfer-encoding: chunked