/**
 * GET /api/projects/.../agent/sessions/.../events/stream?afterSeq=0
 * Server-Sent Events: tail agent_session_events while the session is active.
 */
import { authSession } from "@/lib/auth/session-server";
import { query, queryOne } from "@/lib/db-postgres";

export const dynamic = "force-dynamic";
export const runtime = "nodejs";
/** Long-lived SSE — raise if your host defaults to a shorter limit (e.g. Vercel). */
export const maxDuration = 300;

const TERMINAL = new Set(["done", "approved", "failed", "stopped"]);

export async function GET(
  req: Request,
  { params }: { params: Promise<{ projectId: string; sessionId: string }> }
) {
  const session = await authSession();
  if (!session?.user?.email) {
    return new Response("Unauthorized", { status: 401 });
  }

  const { projectId, sessionId } = await params;
  let afterSeq = Math.max(0, parseInt(new URL(req.url).searchParams.get("afterSeq") ?? "0", 10) || 0);

  const allowed = await queryOne<{ n: string }>(
    `SELECT 1 AS n FROM agent_sessions s
     JOIN fs_projects p ON p.id::text = s.project_id::text
     JOIN fs_users u ON u.id = p.user_id
     WHERE s.id = $1::uuid AND s.project_id::text = $2 AND u.data->>'email' = $3
     LIMIT 1`,
    [sessionId, projectId, session.user.email]
  );
  if (!allowed) {
    return new Response("Not found", { status: 404 });
  }

  const encoder = new TextEncoder();
  const signal = req.signal;

  const stream = new ReadableStream({
    async start(controller) {
      const send = (obj: object) => {
        controller.enqueue(encoder.encode(`data: ${JSON.stringify(obj)}\n\n`));
      };

      let idleAfterTerminal = 0;
      let lastHeartbeat = Date.now();

      try {
        while (!signal.aborted) {
          const rows = await query<{ seq: number; ts: string; type: string; payload: Record<string, unknown> }>(
            `SELECT e.seq, e.ts::text, e.type, e.payload
             FROM agent_session_events e
             WHERE e.session_id = $1::uuid AND e.seq > $2
             ORDER BY e.seq ASC
             LIMIT 200`,
            [sessionId, afterSeq]
          );

          for (const row of rows) {
            afterSeq = row.seq;
            send({ seq: row.seq, ts: row.ts, type: row.type, payload: row.payload });
          }

          const st = await queryOne<{ status: string }>(
            `SELECT status FROM agent_sessions WHERE id = $1::uuid LIMIT 1`,
            [sessionId]
          );
          const status = st?.status ?? "";
          const terminal = TERMINAL.has(status);

          if (rows.length === 0) {
            if (terminal) {
              idleAfterTerminal++;
              if (idleAfterTerminal >= 3) {
                send({ type: "_stream.end", seq: afterSeq });
                break;
              }
            } else {
              idleAfterTerminal = 0;
            }
          } else {
            idleAfterTerminal = 0;
          }

          const now = Date.now();
          if (now - lastHeartbeat > 20000) {
            send({ type: "_heartbeat", t: now });
            lastHeartbeat = now;
          }

          await new Promise((r) => setTimeout(r, 750));
        }
      } catch (e) {
        console.error("[events/stream]", e);
        try {
          send({ type: "_stream.error", message: "stream failed" });
        } catch {
          /* ignore */
        }
      } finally {
        try {
          controller.close();
        } catch {
          /* ignore */
        }
      }
    },
  });

  return new Response(stream, {
    headers: {
      "Content-Type": "text/event-stream; charset=utf-8",
      "Cache-Control": "no-cache, no-transform",
      Connection: "keep-alive",
      "X-Accel-Buffering": "no",
    },
  });
}
