/api/comprehensive

// Hosted comprehensive pull — the paid, authenticated, all-REST tier (D18).
//
// D18 ("public gets the INSTANT pull; COMPREHENSIVE requires login/signup")
// supersedes the earlier "comprehensive tier is coming-soon" posture in
// app/api/instant/route.js. Route 2 from deploy/HOSTING.md: every leg served
// by a REST provider (Firecrawl / Serper+keyless-ladder / ScrapeCreators+
// Apify), so it runs on this Node serverless function with no browser and no
// yt-dlp — see hosted-app/lib/all-rest/all-rest-orchestrator.mjs.
//
// START + POLL, not one long request:
//   POST returns {status:"running", pull_id, run_id, lane} once auth, credits,
//   and the brand/pull rows are settled — see
//   hosted-app/lib/comprehensive-pull-handler.mjs's
//   authorizeAndReserveComprehensivePull for exactly what that checks and in
//   what order (auth BEFORE credits BEFORE any row exists).
//   GET ?pull_id=... polls current status from Supabase.
//
// ============================ THE 300s WALL ============================
//
// THE LIMITATION THAT USED TO BE HONEST AND IS NOW CLOSED.
//
// Vercel functions are request-scoped. The capture is scheduled with
// next/server's after(), which is the supported way to continue work past the
// response — but the platform still kills the invocation at maxDuration
// (300s), and a capture that had not finished writing the pull row by then was
// left `running` forever until the reaper found it. Two real 30-page pulls
// died at ~301s exactly this way.
//
// Raising maxDuration is not the fix, because 300s IS the ceiling on this
// plan, and because "the work happens to fit" is not an architecture. THE FIX
// IS THAT THE PULL ROW IS NOW A RESUMABLE JOB.
//
//   - The whole request (urls, queries, social profiles, page/social budgets)
//     is written onto the row at reservation time, so any later invocation can
//     reconstruct what was asked for.
//   - The capture is cut into STAGES along its real dependency graph. A slice
//     runs as many stages as fit in a conservative budget (240s), persists
//     every settled leg, and yields.
//   - The seo leg — the measured long pole, a serial list of SERP calls —
//     chunks INSIDE itself and checkpoints per chunk, so even one leg can
//     cross a slice boundary without losing calls already paid for.
//   - Three independent triggers pick a yielded job back up (the poll below,
//     the cron/ticker on ./advance, and a slice's own self-invocation), all
//     gated by one fencing-token compare-and-swap so that exactly one of them
//     ever advances a given job at a given moment.
//   - Billing happens once, at true completion, through the one CAS-guarded
//     finalizer both lanes share.
//
// See hosted-app/lib/comprehensive-pull-queue.mjs for the four invariants this
// rests on, and ./advance/route.js for the trigger that runs unattended.
//
// The DIRECT lane (BRAND_STUDIO_PULL_QUEUE=off) is the pre-queue behaviour,
// kept as a fallback: one after(), the whole capture, and therefore the
// shallow depth that measurably fits inside 300s. The UI labels which lane a
// pull ran on rather than implying the same depth in both.
//
// ============================== THE REAPER ==============================
//
// Still here, and now able to tell the two failures apart. A `running` row is
// reaped to a terminal `failed` state — never `partial`, never `passed`, and
// per debitForPull a `failed` pull is NEVER charged — when it has stopped
// ADVANCING, which for a sliced job means "no checkpoint progress for longer
// than the idle threshold" and for a direct-lane pull still means "created
// longer ago than the threshold". Measuring a multi-slice job from created_at
// would reap every job the queue exists for; see stalenessForRunningPull.
//
// It fires lazily on the customer's own GET poll below, and on demand via
// POST /api/comprehensive/sweep. Neither is a cron — ./advance is.
export const runtime = "nodejs";
export const dynamic = "force-dynamic";
// 300s: the platform ceiling on this plan, confirmed against Vercel's duration
// docs for Hobby-with-Fluid. Unlike before the queue, this number is no longer
// load-bearing for correctness — a slice is BUDGETED at 240s and yields on its
// own, so an invocation killed at the wall costs at most the stage that was
// in flight, and the job resumes from its last checkpoint. The number bounds
// how much work one slice can do, not whether the pull can finish.
export const maxDuration = 300;

import { NextResponse, after } from "next/server";
import { supabaseConfigFromEnv } from "../../../lib/supabase-rest.mjs";
import { assertPublicHttpTarget } from "../../../lib/local/url-safety.mjs";
import { searchQueriesFor } from "../../../lib/local/search-query-plan.mjs";
import { readBoundedJsonRequest } from '../../../lib/bounded-http.mjs';
import {
  authorizeAndReserveComprehensivePull,
  runCaptureAndFinalizeComprehensivePull,
  readComprehensivePullStatus,
  bearerTokenFromHeader,
  DEFAULT_STALE_RUNNING_MS
} from "../../../lib/comprehensive-pull-handler.mjs";
import {
  queueEnabledFromEnv,
  QUEUE_PAGE_BUDGET,
  QUEUE_SCREENSHOT_BUDGET,
  QUEUE_SOCIAL_LIMIT,
  DIRECT_PAGE_BUDGET,
  DIRECT_SOCIAL_LIMIT
} from "../../../lib/comprehensive-pull-queue.mjs";
import {
  advanceComprehensivePullSlice,
  triggerNextSlice
} from "../../../lib/comprehensive-pull-queue-runner.mjs";
import { sliceLoopDecision } from "../../../lib/queue-continuation.mjs";

/**
 * Run slices until the job completes or this invocation runs out of room.
 *
 * One invocation can hold several slices back to back when the earlier ones
 * were cheap (the static stage costs no network at all), which is why this
 * loops rather than running exactly one. It stops as soon as a slice yields
 * for budget reasons — that yield IS the signal that this invocation is done —
 * and asks the next invocation to continue. Every slice's budget counts from
 * invocationStartMs, the start of this request (081), so a second slice gets
 * only what the request has left of the slice budget.
 */
const driveSlices = async ({ pullId, config, env, invocationStartMs = Date.now() }) => {
  const startedMs = invocationStartMs;
  for (;;) {
    // eslint-disable-next-line no-await-in-loop
    const result = await advanceComprehensivePullSlice({
      pullId,
      config,
      env,
      invocationStartMs,
      alphaParking: true,
      holder: `invocation-${process.pid}`,
      log: (line) => console.log(`[pull-queue] ${line}`)
    });
    // Stop when someone else holds it, it is done, or billing is paced by
    // owner polls and the worker sweep. A slice that yielded because it ran
    // out of budget has used this invocation's room too; hand off rather than
    // starting another here. A planned_ yield also hands off (080), as
    // /advance does, so a pull keeps moving after the tab closes.
    //
    // finalize_unconfirmed_retryable ALSO hands off: the runner already made
    // three in-place finalize attempts before saying it, so looping here
    // would only stack three-attempt batches against the same outage until
    // the 200-second guard (and every 3-second poll would start another
    // driver doing the same). One bounded batch per invocation; the poll,
    // lease expiry, and the sweep pace the rest.
    const decision = sliceLoopDecision(result, { elapsedMs: Date.now() - startedMs });
    if (decision === "stop") return result;
    if (decision === "handoff") {
      // eslint-disable-next-line no-await-in-loop
      const trigger = await triggerNextSlice({ pullId, env });
      console.log(`[pull-queue] handoff pull=${pullId} triggered=${trigger.triggered} ${trigger.reason || ""}`);
      return result;
    }
  }
};

export async function POST(request) {
  const invocationStartMs = Date.now();
  const token = bearerTokenFromHeader(request.headers.get("authorization"));
  let body;
  try { body = await readBoundedJsonRequest(request, 131072); }
  catch (error) { return NextResponse.json({status:'refused',reason:error.message}, {status:error.httpStatus||400}); }
  const config = supabaseConfigFromEnv(process.env);
  const queueLane = queueEnabledFromEnv(process.env);

  // URL safety runs BEFORE auth and credits, because a refusal here costs the
  // caller nothing and creates no row. Fourteen collectors are about to be
  // pointed at this address; on a hosted deployment, one of them reaching
  // 169.254.169.254 is a credential-exfiltration path, not a bad pull.
  const target = await assertPublicHttpTarget(body?.url || body?.source_url || "");
  if (!target.ok) {
    return NextResponse.json({
      status: "blocked",
      reason: target.reason,
      message: target.message
    }, { status: 422 });
  }

  // Depth is decided HERE, by the lane, not by the client.
  //
  // Before the queue this was a client constant the server merely clamped —
  // seam #10 of the pre-build risk map. Now the server owns it: the queue lane
  // runs the full spec (30 pages / 8 shots / 50 posts) because it can outlive
  // a request, and the direct lane runs the measured-to-fit depth because it
  // cannot. A client asking for more than its lane allows gets its lane's
  // ceiling, and the response says which lane it is on.
  const pageBudget = queueLane
    ? Math.min(QUEUE_PAGE_BUDGET, Math.max(1, Number(body.page_budget) || QUEUE_PAGE_BUDGET))
    : Math.min(DIRECT_PAGE_BUDGET, Math.max(1, Number(body.page_budget) || DIRECT_PAGE_BUDGET));
  const socialLimit = queueLane
    ? Math.min(QUEUE_SOCIAL_LIMIT, Math.max(1, Number(body.social_limit) || QUEUE_SOCIAL_LIMIT))
    : Math.min(DIRECT_SOCIAL_LIMIT, Math.max(1, Number(body.social_limit) || DIRECT_SOCIAL_LIMIT));

  // Auth + credits + row creation happen HERE, synchronously, before the
  // response is sent — an anonymous or out-of-credit caller never reaches the
  // line below that schedules capture work (requirement 2/3). D18's gate is
  // untouched by the queue: authenticated + credits BEFORE any capture, and a
  // pull that produces nothing is still never charged.
  const reservation = await authorizeAndReserveComprehensivePull({
    token,
    body: {
      ...body,
      page_budget: pageBudget,
      social_limit: socialLimit,
      screenshot_budget: queueLane ? QUEUE_SCREENSHOT_BUDGET : undefined
    },
    config,
    queueLane,
    requireExecutionPlan: true,
    alphaParking: true
  });
  if (!reservation.ok) {
    return NextResponse.json(reservation.body, { status: reservation.http_status });
  }

  // AN IDEMPOTENT REPLAY. authorizeAndReserveComprehensivePull found (or
  // raced onto, via the 23505 fallback) a pull already carrying this
  // idempotency key — nothing new was reserved, and NOTHING IS SCHEDULED
  // here. 200, the same pull_id, never a second capture and never a second
  // charge. See that function's IDEMPOTENCY block for the full reasoning.
  if (reservation.replay) {
    return NextResponse.json(reservation.body, { status: reservation.http_status || 200 });
  }

  after(async () => {
    try {
      if (reservation.queued) {
        // The queue lane. This invocation runs the FIRST slice (and any cheap
        // ones after it) and then hands off. If it is killed mid-slice, the
        // checkpoint from the last completed stage survives and the next
        // trigger resumes from there — which is the entire point.
        await driveSlices({ pullId: reservation.pullId, config, env: process.env, invocationStartMs });
        return;
      }
      await runCaptureAndFinalizeComprehensivePull({
        pullId: reservation.pullId,
        runId: reservation.runId,
        userId: reservation.userId,
        sourceUrl: reservation.sourceUrl,
        domain: reservation.domain,
        pullProfileId: reservation.pullProfileId,
        urls: Array.isArray(body.urls) ? body.urls : [],
        // Default to the deterministic brand-identity plan (query-manifest
        // spec §8.1) so a bare {url} request still exercises the search leg —
        // quoted name + domain, official-site, reviews, and the bare domain
        // as ONE entry rather than the whole plan. `reservation.queries` is
        // the CAPPED caller list (MAX_SEARCH_QUERIES, refusal receipt stored
        // on the row) — the raw body is never read here, so an uncapped list
        // cannot reach the paid leg through this lane.
        queries: reservation.queries.length
          ? reservation.queries
          : searchQueriesFor({ brandName: reservation.userConfirmations?.brand_name || "", domain: reservation.domain }),
        socialProfiles: Array.isArray(body.social_profiles) ? body.social_profiles : [],
        // Canonicalized and persisted by reservation; never re-derived from
        // the raw confirmation payload in the after() process.
        groundedIndustry: reservation.groundedIndustry,
        // The VALIDATED confirmations, handed back by the reservation rather
        // than re-read from the raw body — one sanitizer, one answer. Null when
        // the caller sent none, and this lane then behaves exactly as it did
        // before the confirmation step existed.
        userConfirmations: reservation.userConfirmations || null,
        // The reservation-stored plan — the capture reconciles against the
        // exact document the row already carries (adversarial-review F7).
        queryManifest: reservation.queryManifest || null,
        pageBudget,
        socialLimit,
        // THE TIER THE RESERVATION PRICED. Taken from the reservation rather
        // than from `body.tier` for the same reason userConfirmations is: the
        // reservation normalized and gated it, and re-reading the raw body
        // here would be a second dialect of the same decision — one that could
        // run a capture at a depth the customer was not charged for.
        pullTier: reservation.pullTier,
        // Whose keys, and therefore whose bill. Both from the reservation.
        byoKeys: reservation.byoKeys,
        accountUnlimited: reservation.accountUnlimited,
        accountKeyEnv: reservation.accountKeyEnv,
        alphaParking: true,
        config
      });
    } catch (error) {
      // runCaptureAndFinalizeComprehensivePull already catches a capture
      // error and writes a `failed` row itself; this is the last-resort
      // backstop for a failure in the finalize plumbing (e.g. Supabase
      // unreachable). On the queue lane a throw here is survivable in a way it
      // never was before: the row keeps its checkpoint, so the next trigger
      // continues rather than starting over.
      console.error("comprehensive pull finalize failed", reservation.pullId, error);
    }
  });

  return NextResponse.json(reservation.body, { status: 202 });
}

export async function GET(request) {
  const invocationStartMs = Date.now();
  const token = bearerTokenFromHeader(request.headers.get("authorization"));
  const pullId = new URL(request.url).searchParams.get("pull_id") || "";
  const config = supabaseConfigFromEnv(process.env);
  // reapStaleAfterMs turns on the lazy reap-on-poll path (see this file's
  // header): a `running` row that has stopped ADVANCING is moved to `failed`
  // before the response is built, so polling a dead pull is itself the fix
  // rather than reporting "running" forever. A sliced job that is merely
  // between slices is NOT stale and is not touched.
  const result = await readComprehensivePullStatus({ pullId, token, config, reapStaleAfterMs: DEFAULT_STALE_RUNNING_MS });

  // THE POLL IS ALSO A TRIGGER.
  //
  // A customer watching the loading screen is, for free, the most reliable
  // continuation signal the system has — it needs no cron, no self-URL, and no
  // shared secret, and it works identically on Vercel, on docker and on a
  // laptop. Running pulls advance capture; terminal pulls with a durable
  // pending-billing marker reconcile only the debit. No provider work repeats.
  //
  // The cost of doing this on EVERY poll is one row read: the claim's
  // fencing-token compare-and-swap refuses immediately when another writer
  // holds the lease, which during an active slice is almost always. The
  // benefit is that a job whose previous slice yielded resumes within one poll
  // interval instead of waiting up to a minute for the cron. That trade is
  // deliberate — it is cheap to ask and expensive to be stuck.
  const ownerBillingRetry = result.body?.billing_retryable === true
    && !result.body?.public_example && !result.body?.viewed_as_admin;
  if (result.http_status === 200 && (result.body?.status === "running" || ownerBillingRetry) && queueEnabledFromEnv(process.env)) {
    after(async () => {
      try {
        await driveSlices({ pullId, config, env: process.env, invocationStartMs });
      } catch (error) {
        console.error("[pull-queue] poll-advance failed", pullId, error);
      }
    });
  }

  return NextResponse.json(result.body, { status: result.http_status });
}