Gemini API × TanStack Query でストリーミングUI を本番運用する — キャンセル・再試行・キャッシュ整合の設計

// app/api/chat/stream/route.ts
import { GoogleGenerativeAI } from "@google/generative-ai";
 
export const runtime = "nodejs";
export const dynamic = "force-dynamic"; // キャッシュ抑止
 
const genAI = new GoogleGenerativeAI(process.env.GEMINI_API_KEY!);
 
export async function POST(req: Request) {
  const { messages, conversationId } = await req.json();
 
  // クライアント切断検知用
  const abortSignal = req.signal;
 
  const model = genAI.getGenerativeModel({
    model: "gemini-2.5-flash",
    generationConfig: { temperature: 0.7, maxOutputTokens: 2048 },
  });
 
  const encoder = new TextEncoder();
  const stream = new ReadableStream({
    async start(controller) {
      try {
        const result = await model.generateContentStream({
          contents: messages.map((m: { role: string; content: string }) => ({
            role: m.role === "assistant" ? "model" : "user",
            parts: [{ text: m.content }],
          })),
        });
 
        for await (const chunk of result.stream) {
          // クライアント側がキャンセルしていたら即終了
          if (abortSignal.aborted) {
            controller.close();
            return;
          }
          const text = chunk.text();
          if (!text) continue;
          // SSE フォーマット: "data: <JSON>\n\n"
          const payload = JSON.stringify({ type: "chunk", text });
          controller.enqueue(encoder.encode(`data: ${payload}\n\n`));
        }
        // 完了シグナル
        controller.enqueue(
          encoder.encode(`data: ${JSON.stringify({ type: "done", conversationId })}\n\n`)
        );
        controller.close();
      } catch (err: unknown) {
        const message = err instanceof Error ? err.message : "stream error";
        controller.enqueue(
          encoder.encode(`data: ${JSON.stringify({ type: "error", message })}\n\n`)
        );
        controller.close();
      }
    },
  });
 
  return new Response(stream, {
    headers: {
      "Content-Type": "text/event-stream",
      "Cache-Control": "no-cache, no-transform",
      Connection: "keep-alive",
      "X-Accel-Buffering": "no", // Nginx/Cloudflare 経由でもバッファされない
    },
  });
}

// hooks/useStreamingChat.ts
import { useMutation, useQueryClient } from "@tanstack/react-query";
import { useState, useRef, useCallback, useEffect } from "react";
 
type Message = { role: "user" | "assistant"; content: string };
 
type StreamRequest = { conversationId: string; messages: Message[] };
 
async function* readSSE(response: Response, signal: AbortSignal) {
  if (!response.body) throw new Error("No response body");
  const reader = response.body.getReader();
  const decoder = new TextDecoder();
  let buffer = "";
  try {
    while (true) {
      if (signal.aborted) return;
      const { done, value } = await reader.read();
      if (done) return;
      buffer += decoder.decode(value, { stream: true });
      // "\n\n" 区切りのSSEイベントを切り出す
      const events = buffer.split("\n\n");
      buffer = events.pop() ?? "";
      for (const ev of events) {
        const line = ev.split("\n").find((l) => l.startsWith("data: "));
        if (!line) continue;
        try {
          yield JSON.parse(line.slice(6));
        } catch {
          // 壊れたイベントはスキップ（ネットワーク途絶時など）
        }
      }
    }
  } finally {
    reader.releaseLock();
  }
}
 
export function useStreamingChat(conversationId: string) {
  const qc = useQueryClient();
  const [streamingText, setStreamingText] = useState("");
  const abortRef = useRef<AbortController | null>(null);
 
  const mutation = useMutation({
    mutationKey: ["chat-stream", conversationId],
    mutationFn: async (req: StreamRequest) => {
      // 既存のストリームがあればキャンセル（連打対策）
      abortRef.current?.abort();
      const ac = new AbortController();
      abortRef.current = ac;
      setStreamingText("");
 
      const res = await fetch("/api/chat/stream", {
        method: "POST",
        headers: { "Content-Type": "application/json" },
        body: JSON.stringify(req),
        signal: ac.signal,
      });
      if (!res.ok) throw new Error(`HTTP ${res.status}`);
 
      let full = "";
      for await (const event of readSSE(res, ac.signal)) {
        if (event.type === "chunk") {
          full += event.text;
          // 部分結果を React state で表示
          setStreamingText(full);
        } else if (event.type === "error") {
          throw new Error(event.message);
        } else if (event.type === "done") {
          return { conversationId: event.conversationId, text: full };
        }
      }
      // ループが done を見ずに抜けた場合（接続切断など）
      throw new Error("stream ended unexpectedly");
    },
    onSuccess: (data) => {
      // 確定した結果を queryClient のキャッシュに同期
      qc.setQueryData<Message[]>(["conversation", conversationId], (old = []) => [
        ...old,
        { role: "assistant", content: data.text },
      ]);
      setStreamingText("");
    },
    onError: () => {
      // 中途半端な部分結果を保持するか破棄するかは UX 設計次第
      // 本実装では「途中まで表示されたものを履歴に残す」方針
      if (streamingText) {
        qc.setQueryData<Message[]>(["conversation", conversationId], (old = []) => [
          ...old,
          { role: "assistant", content: streamingText + "\n\n[応答が中断されました]" },
        ]);
      }
      setStreamingText("");
    },
  });
 
  // アンマウント時の必須クリーンアップ（メモリリーク・無駄課金対策）
  useEffect(() => {
    return () => {
      abortRef.current?.abort();
    };
  }, []);
 
  const cancel = useCallback(() => {
    abortRef.current?.abort();
  }, []);
 
  return { ...mutation, streamingText, cancel };
}

// CancelButton.tsx
import { useStreamingChat } from "@/hooks/useStreamingChat";
 
export function CancelButton({ conversationId }: { conversationId: string }) {
  const { cancel, isPending } = useStreamingChat(conversationId);
  if (!isPending) return null;
  return (
    <button
      type="button"
      onClick={cancel}
      className="rounded bg-red-500 px-3 py-1 text-white"
    >
      応答を中止
    </button>
  );
}

// ❌ ダメな例：ストリーム開始時しかチェックしていない
async function bad(stream: AsyncIterable<unknown>, abortSignal: AbortSignal) {
  if (abortSignal.aborted) return;
  for await (const chunk of stream) {
    // ここに入った後はキャンセル不可
    void chunk;
  }
}
 
// ✅ 正しい例：各チャンクで確認
async function good(stream: AsyncIterable<unknown>, abortSignal: AbortSignal) {
  for await (const chunk of stream) {
    if (abortSignal.aborted) return;
    void chunk;
  }
}

// useMutation のオプション部分のみ抜粋
{
  retry: (failureCount, error) => {
    // 既にストリームが始まっていた場合は再試行しない
    if (error instanceof Error && error.message === "stream ended unexpectedly") {
      return false; // 課金二重を避ける
    }
    // 接続前のエラーは3回まで
    return failureCount < 3;
  },
  retryDelay: (attempt) => Math.min(1000 * 2 ** attempt, 30000),
}

// hooks/useConversation.ts
import { useQuery } from "@tanstack/react-query";
 
type Message = { role: "user" | "assistant"; content: string };
 
export function useConversation(conversationId: string) {
  return useQuery<Message[]>({
    queryKey: ["conversation", conversationId],
    queryFn: async () => {
      const res = await fetch(`/api/conversations/${conversationId}`);
      if (!res.ok) throw new Error("Failed to load conversation");
      return res.json();
    },
    // ストリーミング中の重複取得を抑止
    staleTime: 60_000,
    // ウィンドウ復帰時の自動再取得は要注意（後述）
    refetchOnWindowFocus: false,
  });
}

// app/providers.tsx
import { QueryClient, QueryClientProvider } from "@tanstack/react-query";
 
const queryClient = new QueryClient({
  defaultOptions: {
    queries: {
      refetchOnWindowFocus: false,
      staleTime: 30_000,
    },
    mutations: {
      // mutation のリトライはデフォルト 0 なので明示しなくてもよいが、
      // 意図を明確にするため記述しておく
      retry: 0,
    },
  },
});

useEffect(() => {
  return () => {
    // アンマウント時に AbortController を発火
    abortRef.current?.abort();
  };
}, []);

// __tests__/streamingChat.test.tsx
import { setupServer } from "msw/node";
import { http, HttpResponse } from "msw";
import { renderHook, act, waitFor } from "@testing-library/react";
import { QueryClient, QueryClientProvider } from "@tanstack/react-query";
import { useStreamingChat } from "@/hooks/useStreamingChat";
 
const server = setupServer(
  http.post("/api/chat/stream", () => {
    const stream = new ReadableStream({
      async start(controller) {
        const enc = new TextEncoder();
        const events = [
          { type: "chunk", text: "こんにちは" },
          { type: "chunk", text: "、Gemini です。" },
          { type: "done", conversationId: "conv-1" },
        ];
        for (const e of events) {
          controller.enqueue(enc.encode(`data: ${JSON.stringify(e)}
 
`));
          await new Promise((r) => setTimeout(r, 10));
        }
        controller.close();
      },
    });
    return new HttpResponse(stream, {
      headers: { "Content-Type": "text/event-stream" },
    });
  })
);
 
beforeAll(() => server.listen());
afterAll(() => server.close());
 
const wrapper = ({ children }: { children: React.ReactNode }) => {
  const qc = new QueryClient({ defaultOptions: { mutations: { retry: 0 } } });
  return <QueryClientProvider client={qc}>{children}</QueryClientProvider>;
};
 
test("ストリーミング完了時にキャッシュへ書き込まれる", async () => {
  const { result } = renderHook(() => useStreamingChat("conv-1"), { wrapper });
  act(() => {
    result.current.mutate({
      conversationId: "conv-1",
      messages: [{ role: "user", content: "テスト" }],
    });
  });
  await waitFor(() => expect(result.current.isSuccess).toBe(true));
  expect(result.current.data?.text).toBe("こんにちは、Gemini です。");
});