sql-server-query-ai/app/api/analyze/route.ts

import { requireUser } from "@/lib/auth";
import { analystConfig, jevKey, jevModel } from "@/lib/settings";
import { digestPlan, digestForModel, PlanError } from "@/lib/digest";
import { proposeCandidates, streamReport, AnalystError } from "@/lib/analyst";
import { judgeCandidates, jevFallback, makeJevClient } from "@/lib/jev";
import { sse } from "@/lib/stream";
import { newAnalysis, patchAnalysis } from "@/lib/db";
import { reportPrompt } from "@/lib/report-spec";

export const runtime = "nodejs";
export const dynamic = "force-dynamic";
const MAX_BODY_BYTES = 16_000_000;

export async function POST(req: Request) {
  const user = await requireUser().catch(() => null);
  if (!user) return Response.json({ error: "Sign in first." }, { status: 401 });

  const analyst = analystConfig(user.id);
  const jev = jevKey(user.id);
  if (!analyst) return Response.json({ error: "Set the analyst model in Settings first." }, { status: 412 });
  if (!jev) return Response.json({ error: "Set your Jev (TypeSafe) API key in Settings first." }, { status: 412 });

  const raw = await req.text();
  if (raw.length > MAX_BODY_BYTES) return Response.json({ error: "Request too large." }, { status: 413 });
  let xml = ""; let note = "";
  try { const body = JSON.parse(raw); xml = String(body.xml ?? ""); note = String(body.note ?? "").slice(0, 500); }
  catch { return Response.json({ error: "Expected JSON with an xml field." }, { status: 400 }); }

  const id = newAnalysis(user.id, note.trim().slice(0, 120) || "Plan analysis", xml);
  const enc = new TextEncoder();
  const prompt = reportPrompt();

  // The analyst model needs minutes per stage. If the DBA closes the tab, enqueue() does not throw, so the
  // producer would keep firing model calls into a dead stream and queue behind its own ghost. cancel() is the
  // only signal that the reader is gone: stop launching further calls and abort the one in flight.
  const ac = new AbortController();
  let abandoned = false;

  const stream = new ReadableStream<Uint8Array>({ async start(ctrl) {
    const send = (event: string, data: unknown) => ctrl.enqueue(enc.encode(sse(event, data)));
    let oul = "";
    const close = (status: string, error?: string) => {
      patchAnalysis(id, { status, ...(error ? { error } : {}), ...(oul ? { oul } : {}) });
      send("done", { id }); ctrl.close();
    };
    try {
      send("stage", { stage: "digesting" });
      let digest;
      try { digest = digestPlan(xml); }
      catch (e) { const m = e instanceof PlanError ? e.message : "Could not read that plan.";
        send("error", { stage: "digesting", message: m }); patchAnalysis(id, { status: "failed", error: m });
        send("done", { id }); return ctrl.close(); }
      if (abandoned) return giveUp(id, oul);
      patchAnalysis(id, { digest: JSON.stringify(digest) });

      send("stage", { stage: "proposing" });
      let candidates;
      try { candidates = await proposeCandidates(analyst, digest, note, fetch, ac.signal); }
      catch (e) { const m = e instanceof AnalystError ? e.message : "The analyst model call failed.";
        send("error", { stage: "proposing", message: m }); patchAnalysis(id, { status: "failed", error: m });
        send("done", { id }); return ctrl.close(); }
      if (abandoned) return giveUp(id, oul);
      candidates.forEach((c) => send("candidate", { key: c.key, title: c.title }));
      patchAnalysis(id, { candidates: JSON.stringify(candidates) });

      send("stage", { stage: "judging" });
      let verdict;
      try { verdict = await judgeCandidates(digest, candidates, makeJevClient(jev, jevModel(user.id))); }
      catch (e) { verdict = jevFallback(candidates, `Jev call failed: ${(e as Error).message}`); }
      send("verdict", verdict);
      patchAnalysis(id, { verdict: JSON.stringify(verdict) });

      send("stage", { stage: "writing" });
      for await (const chunk of streamReport(analyst,
          { plan: digestForModel(digest), options: candidates, verdict }, prompt, fetch, ac.signal)) {
        oul += chunk; send("oul", { chunk });
      }
      if (abandoned) return giveUp(id, oul);
      close(oul ? "done" : "partial", oul ? undefined : "The analyst model returned an empty report.");
    } catch (e) {
      const m = abandoned ? "Browser stopped listening." : ((e as Error)?.message ?? "Analysis failed.");
      if (!abandoned) send("error", { stage: "stream", message: m });
      patchAnalysis(id, { status: "failed", error: m, ...(oul ? { oul } : {}) });
      send("done", { id }); ctrl.close();
    }
    function giveUp(id: string, oul: string) {
      patchAnalysis(id, { status: oul ? "partial" : "abandoned",
        error: "Closed before the run finished.", ...(oul ? { oul } : {}) });
      try { ctrl.close(); } catch { /* already closed */ }
    }
  }, cancel() { abandoned = true; ac.abort(); } });

  return new Response(stream, { headers: { "content-type": "text/event-stream; charset=utf-8",
    "cache-control": "no-store", connection: "keep-alive", "x-accel-buffering": "no" } });
}