fifteen/apps/web/app/api/admin/jobs/[id]/stream/route.ts

import { NextRequest } from "next/server";
import { Job } from "bullmq";
import { getPipelineQueue, getValhallaQueue } from "@/lib/queue";
import type { PipelineJobData, JobProgress, ComputeScoresJobData, RefreshCityJobData } from "@/lib/queue";
import type { SSEEvent } from "@transportationer/shared";
import { CATEGORY_IDS } from "@transportationer/shared";

export const runtime = "nodejs";

function fmt(event: SSEEvent): string {
  return `data: ${JSON.stringify(event)}\n\n`;
}

export async function GET(
  req: NextRequest,
  { params }: { params: Promise<{ id: string }> },
): Promise<Response> {
  const { id } = await params;
  const queue = getPipelineQueue();
  const encoder = new TextEncoder();
  let timer: ReturnType<typeof setInterval> | null = null;

  // Resolve citySlug and creation timestamp from the refresh-city job.
  // We track progress by citySlug across all pipeline stages because
  // refresh-city itself completes almost immediately after enqueueing children.
  // jobCreatedAt gates failed lookups so we never match results from a
  // previous ingest of the same city.
  // computeScoresJobId is captured after flow.add() by the worker; once
  // available it allows exact-ID matching for the completion check,
  // eliminating false positives from previous runs.
  let citySlug: string;
  let jobCreatedAt: number;
  let computeScoresJobId: string | undefined;
  try {
    const job = await Job.fromId<PipelineJobData>(queue, id);
    if (!job) {
      return new Response(fmt({ type: "failed", jobId: id, error: "Job not found" }), {
        headers: { "Content-Type": "text/event-stream" },
      });
    }
    citySlug = "citySlug" in job.data ? (job.data.citySlug ?? "") : "";
    jobCreatedAt = job.timestamp;
    computeScoresJobId = (job.data as RefreshCityJobData).computeScoresJobId;
  } catch {
    return new Response(fmt({ type: "failed", jobId: id, error: "Queue unavailable" }), {
      headers: { "Content-Type": "text/event-stream" },
    });
  }

  const valhallaQueue = getValhallaQueue();

  const stream = new ReadableStream({
    start(ctrl) {
      const enqueue = (e: SSEEvent) => {
        try {
          ctrl.enqueue(encoder.encode(fmt(e)));
        } catch {
          /* controller already closed */
        }
      };

      const cleanup = () => {
        if (timer) { clearInterval(timer); timer = null; }
        try { ctrl.close(); } catch { /* already closed */ }
      };

      const poll = async () => {
        try {
          // If computeScoresJobId wasn't set when the stream opened (race with
          // the worker updating job data), re-read the job once to pick it up.
          if (!computeScoresJobId) {
            const refreshJob = await Job.fromId<PipelineJobData>(queue, id);
            computeScoresJobId = refreshJob
              ? (refreshJob.data as RefreshCityJobData).computeScoresJobId
              : undefined;
          }

          // 1. Fetch active jobs and waiting-children jobs in parallel.
          const [pipelineActive, valhallaActive, waitingChildren] = await Promise.all([
            queue.getActive(0, 100),
            valhallaQueue.getActive(0, 100),
            queue.getWaitingChildren(0, 100),
          ]);

          // 1a. Parallel routing phase: compute-scores is waiting for its routing
          //     children to finish. Report aggregate progress instead of one job's pct.
          // Only enter this branch when routingDispatched=true (Phase 1 has run).
          // Before that, compute-scores is in waiting-children while generate-grid
          // is running — fall through to the sequential active-job check instead.
          // Match by job ID (exact) when available; fall back to citySlug for the
          // brief window before computeScoresJobId is written to the job record.
          const csWaiting = waitingChildren.find(
            (j) =>
              j.data.type === "compute-scores" &&
              (j.data as ComputeScoresJobData).routingDispatched === true &&
              (computeScoresJobId ? j.id === computeScoresJobId : j.data.citySlug === citySlug),
          );
          if (csWaiting) {
            const csData = csWaiting.data as ComputeScoresJobData;
            // Transit uses a single compute-transit child, not per-category routing jobs.
            const routingModes = csData.modes.filter((m) => m !== "transit");
            const totalRoutingJobs = routingModes.length * CATEGORY_IDS.length;
            const hasTransit = csData.modes.includes("transit");

            // Count jobs that haven't finished yet (active or still waiting in queue)
            const pipelineWaiting = await queue.getWaiting(0, 200);
            const stillRoutingActive = pipelineActive.filter(
              (j) => j.data.citySlug === citySlug && j.data.type === "compute-routing",
            ).length;
            const stillRoutingWaiting = pipelineWaiting.filter(
              (j) => j.data.citySlug === citySlug && j.data.type === "compute-routing",
            ).length;
            const completedRouting = Math.max(0, totalRoutingJobs - stillRoutingActive - stillRoutingWaiting);

            // Check if compute-transit is still running
            const transitRunning =
              hasTransit &&
              (pipelineActive.some((j) => j.data.citySlug === citySlug && j.data.type === "compute-transit") ||
               pipelineWaiting.some((j) => j.data.citySlug === citySlug && j.data.type === "compute-transit"));

            // compute-transit job also shows its own progress when active — prefer that
            const transitActiveJob = pipelineActive.find(
              (j) => j.data.citySlug === citySlug && j.data.type === "compute-transit",
            );
            if (transitActiveJob) {
              const p = transitActiveJob.progress as JobProgress | undefined;
              if (p?.stage) {
                enqueue({ type: "progress", stage: p.stage, pct: p.pct, message: p.message });
                return;
              }
            }

            const pct = totalRoutingJobs > 0
              ? Math.round((completedRouting / totalRoutingJobs) * 100)
              : transitRunning ? 99 : 100;
            const message = transitRunning && completedRouting >= totalRoutingJobs
              ? "Routing done — computing transit isochrones…"
              : `${completedRouting} / ${totalRoutingJobs} routing jobs`;

            enqueue({ type: "progress", stage: "Computing scores", pct, message });
            return;
          }

          // 1b. Sequential phase: report whichever single job is currently active.
          const activeJob = [...pipelineActive, ...valhallaActive].find(
            (j) => j.data.citySlug === citySlug && j.data.type !== "refresh-city",
          );

          if (activeJob) {
            const p = activeJob.progress as JobProgress | undefined;
            if (p?.stage) {
              enqueue({
                type: "progress",
                stage: p.stage,
                pct: p.pct,
                message: p.message,
                bytesDownloaded: p.bytesDownloaded,
                totalBytes: p.totalBytes,
              });
            } else {
              enqueue({ type: "heartbeat" });
            }
            return;
          }

          // 2. No active stage — check for a failure that occurred after this refresh started.
          const [pipelineFailed, valhallaFailed] = await Promise.all([
            queue.getFailed(0, 50),
            valhallaQueue.getFailed(0, 50),
          ]);
          const recentFail = [...pipelineFailed, ...valhallaFailed].find(
            (j) =>
              j.data.citySlug === citySlug &&
              j.data.type !== "refresh-city" &&
              (j.finishedOn ?? 0) > jobCreatedAt,
          );
          if (recentFail) {
            enqueue({
              type: "failed",
              jobId: recentFail.id ?? "",
              error: recentFail.failedReason ?? "Pipeline stage failed",
            });
            cleanup();
            return;
          }

          // 3. Check if the specific compute-scores job completed → pipeline done.
          // Use exact job ID match (computeScoresJobId) to avoid false positives
          // from a previous run's completed record still in BullMQ's retention window.
          const completed = await queue.getCompleted(0, 100);
          const finalDone = completed.find((j) =>
            computeScoresJobId
              ? j.id === computeScoresJobId
              : j.data.citySlug === citySlug &&
                j.data.type === "compute-scores" &&
                (j.finishedOn ?? 0) > jobCreatedAt,
          );
          if (finalDone) {
            enqueue({ type: "completed", jobId: finalDone.id ?? "" });
            cleanup();
            return;
          }

          // 4. Still pending — heartbeat.
          enqueue({ type: "heartbeat" });
        } catch {
          enqueue({ type: "heartbeat" });
        }
      };

      poll();
      timer = setInterval(poll, 1000);

      req.signal.addEventListener("abort", cleanup);
    },

    cancel() {
      if (timer) clearInterval(timer);
    },
  });

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