sase.tr/apps/api/src/jobs/prefetch-worker.service.ts

import {
  Inject,
  Injectable,
  Logger,
  type OnModuleDestroy,
  type OnModuleInit,
} from "@nestjs/common";
import { DelayedError, type Job, type Queue, Worker } from "bullmq";
import { and, asc, eq, gt, inArray, isNull, notExists, sql } from "drizzle-orm";
import { CategoriesService } from "../categories/categories.service";
import { DATABASE, type Database } from "../database/database.provider";
import { categories, parts, vehicles } from "../database/schema/core";
import { isPl24LeafNode, isPl24PartDetailNode } from "../integrations/pl24/pl24-tree";
import { PostHogService } from "../posthog/posthog.service";
import { RedisService } from "../redis/redis.service";
import { QUEUE_NAMES, getBullConnection } from "./bull.config";
import { backfillContext } from "./prefetch-context";
import {
  RateLimitError,
  alignToWindow,
  checkCooldown,
  checkTimeWindow,
  initProgress,
  isCatalogBackfillEnabled,
  isWithinTimeWindow,
  updateProgress,
} from "./prefetch-utils";
import { PrefetchCategoryJobData, PrefetchInitJobData } from "./prefetch.types";
import {
  CATALOG_PREFETCH_FAST_QUEUE,
  CATALOG_PREFETCH_QUEUE,
} from "./queues/catalog-prefetch.queue";

/**
 * Depth cap for the prefetch BFS — a safety ceiling, NOT the target depth.
 *
 * Goal is COMPLETE coverage: every decoded vehicle's full category tree + all
 * parts in the DB so viewing is instant (DB read). The BFS terminates naturally
 * at real leaves (isLeafLinkPath); this cap only guards against pathological /
 * cyclic trees running away. Real PL24/EMEX/pcat trees are ~5-7 deep, so 12 is
 * comfortably beyond every genuine leaf while still bounding a runaway.
 * Efficiency is handled separately: queueCategoryJob never enqueues a children
 * job past the cap (no early-return no-ops), and completed vehicles are skipped
 * by the scan. Env-tunable (PREFETCH_MAX_DEPTH).
 */
const MAX_DEPTH = Number(process.env.PREFETCH_MAX_DEPTH) || 12;

// ── Backfill scan (hourly cron) tuning ──
/**
 * Vehicles queued per scan wave. Was 20 while pcat needed business-hours
 * gating; doubled to 40 once the warm JWT pool + Redis hydration made the
 * scrape stack 24/7. The MAX_BACKLOG guard below still gates run-away
 * pileups, and per-source cooldown still yields to live users.
 */
const BACKFILL_BATCH_SIZE = 40;
/** Skip the wave entirely if the queue already has more than this many jobs pending. */
const BACKFILL_MAX_BACKLOG = 1000;
/**
 * In-flight guard TTL (seconds) — safety net if a run dies without clearing.
 *
 * 36h, NOT 6h: a chain deferred on the daily budget lives until the next UTC
 * midnight (~24h). With a 6h TTL the guard expired mid-chain, the scan re-picked
 * the vehicle, processInit reset progress.total, and the in-flight jobs'
 * incrementCompleted then compared `completed >= total` against the NEW total —
 * corrupting completion accounting for a still-partial vehicle.
 */
const BACKFILL_SCHEDULED_TTL = Number(process.env.PREFETCH_SCHEDULED_TTL_H || 36) * 60 * 60;
/**
 * Only delayed jobs due within this horizon count as queue PRESSURE.
 *
 * `delayed` lumps two very different things together: per-minute source-rate
 * defers (<60s) and retry backoff (30/60s), which ARE load, versus daily-budget
 * and off-hours defers (hours, parked on the next UTC midnight), which are merely
 * THROTTLED FUTURE WORK. Counting the latter as backlog froze Phase-2 for most of
 * the day (2026-08-01: delayed=11495, all pcat, all parked at 00:00:01 UTC; the
 * partial drain fell to ~25 vehicles/day, ETA 79 days).
 */
const PRESSURE_HORIZON_MS = Number(process.env.PREFETCH_PRESSURE_HORIZON_MS) || 10 * 60_000;
/**
 * Absolute ceiling on TOTAL pending jobs (waiting+active+delayed+prioritized)
 * across BOTH lanes — the runaway backstop that `delayed` used to provide by
 * accident. Excluding long defers from the pressure gate removes the only
 * negative feedback on production, so the pool needs its own hard stop: Phase-2
 * can otherwise produce ~288k jobs/day against ~65k/day of budget. 50k ≈ 15h of
 * the combined daily budgets, so the pool always drains inside a window; 1/9 of
 * the guard-less BFS runaway (470k, months to drain). Set 0 to disable Phase-2.
 */
const HARD_MAX_TOTAL_JOBS = Number(process.env.PREFETCH_MAX_TOTAL_JOBS ?? 50_000);
/**
 * Rough fan-out of one Phase-2 re-drill. Admission control is done in JOB units,
 * not vehicle units: 40 vehicles/wave is meaningless when one vehicle is 100-3000
 * jobs (p95 tree = 744 categories).
 */
const EST_JOBS_PER_VEHICLE = Number(process.env.PREFETCH_EST_JOBS_PER_VEHICLE) || 400;
/**
 * Share of each source's daily budget reserved for the FAST lane. The daily
 * counter has no lane component, so backfill spending the budget also parked the
 * user's fresh-decode chain until midnight. One shared counter (the total
 * upstream/bandwidth ceiling stays exactly SOURCE_DAILY_MAX) but two thresholds:
 * the main (backfill) lane stops at 80%, the fast lane may use 100%.
 */
const DAILY_FAST_RESERVE = 0.2;
/** Only these decode sources have catalogs worth prefetching. */
/**
 * Whether the PL24 *background* backfill lane may run. The user-triggered fast
 * lane is never gated by this.
 *
 * Default is OFF. Two PL24 accounts were banned while bulk background load ran
 * against them, so the background lane has to be switched on deliberately and
 * watched, never inherited from an unset variable.
 *
 * `PL24_BACKFILL_ENABLED` replaces the old `PL24_TR_DISABLED`, whose name said
 * "the tr account is dead" while its actual job was "keep bulk load off the one
 * surviving account". The old variable is still honoured so a half-applied
 * deploy cannot silently open the tap: it can only keep the lane closed.
 */
function isPl24BackfillEnabled(): boolean {
  if (process.env.PL24_BACKFILL_ENABLED !== "true") return false;
  // Legacy kill switch still wins while it is explicitly set.
  return process.env.PL24_TR_DISABLED !== "true";
}

const BACKFILL_SOURCES = ["pl24", "emex", "parts-catalogs"];
/** Redis key holding the rolling rescan cursor (last createdAt seen). */
const BACKFILL_CURSOR_KEY = "prefetch:backfill:cursor";

// ── In-process scan timer (replaces the BullMQ cron, which gets buried) ──
/** How often the in-process backfill scan runs. */
const BACKFILL_SCAN_INTERVAL_MS = 60 * 60 * 1000;
/** Delay before the first scan after boot, so startup settles first. */
const BACKFILL_SCAN_KICKOFF_MS = 60 * 1000;
/** NX-lock TTL (seconds) — just under the interval so each cycle re-acquires. */
const BACKFILL_SCAN_LOCK_TTL_S = 55 * 60;

// ── Throughput knobs (env-tunable so prod can ramp without a redeploy) ──
/**
 * Concurrent jobs the worker runs. emex/pl24 fetches parallelise well, so >1
 * lets a slow parts-catalogs job stop head-of-line-blocking emex. A user-active
 * cooldown still pauses the whole worker, so this only speeds up idle periods.
 */
const WORKER_CONCURRENCY = Number(process.env.PREFETCH_CONCURRENCY) || 8;
/**
 * Global queue rate ceiling (jobs/min) across ALL sources — now a SAFETY CAP,
 * not the real throttle. Per-source limits (below) do the actual pacing so a
 * slow/limited source can't starve the others. Defaults to the sum of the
 * per-source ceilings + headroom. Was a hard 5/min, then a shared 20/min.
 */
const WORKER_RATE_MAX = Number(process.env.PREFETCH_RATE_MAX) || 180;
/**
 * Per-source rate ceilings (jobs/min). Each source is throttled INDEPENDENTLY
 * (Redis fixed-window) so PL24, EMEX and parts-catalogs run concurrently at
 * their own safe rates instead of fighting over one global budget. 0 = unlimited
 * for that source. PL24 stays conservative (upstream ban risk); raise it only
 * with rotating residential proxies. pcat is already paced by PCAT_PACE_MS.
 */
const SOURCE_RATE_MAX: Record<string, number> = {
  pl24: Number(process.env.PREFETCH_RATE_PL24) || 20,
  emex: Number(process.env.PREFETCH_RATE_EMEX) || 20,
  // pcat runs through a DataImpulse rotating gateway (74.81.81.81:10000-10999,
  // ~1000 port-rotated exit IPs — verified 2026-06-25), so per-IP ban load is
  // spread; safe to run well above the old shared cap. Ramp step 2 = 90/min
  // (2026-07-08: 93k backlog %89 pcat idi — 20/min + 5s pace drenaja yetmiyordu).
  "parts-catalogs": Number(process.env.PREFETCH_RATE_PCAT) || 90,
};
/**
 * Per-source ROLLING-DAY budget (backfill jobs/source/UTC-day). A storm guard on
 * top of the per-minute ceilings: those cap burst rate but not the daily TOTAL,
 * so a source running near its ceiling for many hours drains the proxy budget
 * (2026-07-09: a backlog drain pushed pcat to ~74k calls / ~10 GB in one day).
 * When a source hits its daily budget, further backfill jobs defer until the
 * window rolls. User-facing decodes are unaffected — they don't pass through the
 * worker. Only counts jobs that already cleared the per-minute gate, so
 * rate-limited retries don't inflate it. 0 = unlimited. Env-tunable: raise to
 * drain a backlog faster, lower to conserve proxy budget harder.
 */
const SOURCE_DAILY_MAX: Record<string, number> = {
  pl24: Number(process.env.PREFETCH_DAILY_PL24) || 15_000,
  emex: Number(process.env.PREFETCH_DAILY_EMEX) || 20_000,
  // 45k (1.5x): the 2026-07-09 incident that created this guard was ~74k calls /
  // ~10 GB in a day, so 45k ≈ 61% of that — a reversible step at ~6 GB/day. With
  // DAILY_FAST_RESERVE the backfill (main) lane gets 36k, i.e. +20% over the old
  // effective 30k while still leaving 9k for user decodes. Burst rate is
  // UNCHANGED (SOURCE_RATE_MAX pcat 90/min), so per-IP ban pressure is the same —
  // this only sustains that pace for more of the day. Rollback signal: any hour
  // with >2 pcat HTTP 402 (DataImpulse quota) → back to 30_000.
  "parts-catalogs": Number(process.env.PREFETCH_DAILY_PCAT) || 45_000,
};
/**
 * Per-job pacing for parts-catalogs only (its browser/JWT capture is heavy).
 * Set 0 to disable. Other sources are paced by the limiter + cooldown alone.
 */
const PCAT_PACE_MS = Number(process.env.PREFETCH_PCAT_DELAY_MS) || 1_500;

/**
 * Per-job pacing for PL24 BACKFILL jobs (main lane only — a user waiting on a
 * fresh decode must never be slowed down). Both account bans followed days of
 * thousands of back-to-back PL24 calls; a paced, jittered stream looks nothing
 * like that. Set 0 to disable.
 */
const PL24_PACE_MS = Number(process.env.PREFETCH_PL24_DELAY_MS) || 8_000;

/**
 * How deep the REACTIVE (fast-lane) drill may go for PL24.
 *
 * A freshly decoded vehicle used to be walked to the bottom immediately: one
 * Passat produced 1,251 categories, one L200 2,323 — 1.4k-6.8k categories/day
 * from 3-17 decodes, which is exactly the volume that preceded both bans
 * (plv2.md §2.2). Depth 1 = top groups and their direct children; anything
 * deeper is fetched lazily when the user actually opens that node, or by the
 * budgeted backfill lane. Other sources keep MAX_DEPTH.
 */
const PL24_FAST_MAX_DEPTH = Number(process.env.PREFETCH_PL24_FAST_DEPTH) || 1;

/** Depth ceiling for this source+lane. */
function maxDepthFor(source: string, fast: boolean): number {
  if (source === "pl24" && fast) return PL24_FAST_MAX_DEPTH;
  return MAX_DEPTH;
}

/** Jittered pace so our request stream is not a metronome. */
function jitter(ms: number): number {
  return Math.round(ms * (0.5 + Math.random()));
}

/** Test-only surface for the pure helpers above. */
export const __testables = { maxDepthFor, jitter, isPl24BackfillEnabled };

// ── Phase-1 residue exclusion ──
/**
 * Skip a zero-parts vehicle once this many backfill attempts have completed with
 * still-zero parts (genuine upstream-data gaps — e.g. a VIN in no catalog).
 * Re-picking these every wave wastes capacity and starves the Phase-2 rescan.
 */
const NORESULT_MAX_ATTEMPTS = Number(process.env.PREFETCH_NORESULT_MAX) || 2;
/** TTL for the no-result counter — excluded vehicles retry after this, so a later
 *  catalog fix eventually re-fills them (default 7 days). */
const NORESULT_TTL_S = (Number(process.env.PREFETCH_NORESULT_TTL_DAYS) || 7) * 86_400;
/** TTL for the "fully fetched" marker. A vehicle whose whole prefetch chain
 *  finished WITH parts is marked complete and skipped by the Phase-2 rescan, so
 *  the scan stops re-walking finished trees (the old behaviour churned the queue
 *  forever). The TTL re-validates periodically so a later upstream catalog change
 *  is eventually picked up (default 21 days). Env-tunable. */
const COMPLETE_TTL_S = (Number(process.env.PREFETCH_COMPLETE_TTL_DAYS) || 21) * 86_400;
/**
 * Per-vehicle category ceiling — anti-poison guard. A correctly-decoded PL24
 * vehicle has ~100-800 categories (p95 744). A vehicle whose tree blows past
 * this is almost always a decode that failed to resolve the model and landed on
 * the generic ROOT catalog (the whole brand universe — e.g. 5 Opels with
 * model="Opel" produced 420k/102k/21k categories and ZERO parts). Once a
 * vehicle's discovered tree exceeds this, stop drilling it and mark it poison so
 * it never gets re-picked. Env-tunable. */
const CATEGORY_CAP = Number(process.env.PREFETCH_CATEGORY_CAP) || 3000;
/** TTL for the poison marker (generic-model / over-cap vehicles). Long, because
 *  the underlying cause is a decode bug — re-validate monthly. */
const POISON_TTL_S = (Number(process.env.PREFETCH_POISON_TTL_DAYS) || 30) * 86_400;

@Injectable()
export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
  private readonly logger = new Logger(PrefetchWorkerService.name);
  private worker: Worker | null = null;
  private fastWorker?: Worker;
  private backfillKickoff?: ReturnType<typeof setTimeout>;
  private backfillInterval?: ReturnType<typeof setInterval>;

  constructor(
    @Inject(CATALOG_PREFETCH_QUEUE) private queue: Queue,
    @Inject(CATALOG_PREFETCH_FAST_QUEUE) private fastQueue: Queue,
    private categoriesService: CategoriesService,
    private redis: RedisService,
    private posthog: PostHogService,
    @Inject(DATABASE) private db: Database,
  ) {}

  onModuleInit() {
    this.worker = new Worker(
      QUEUE_NAMES.CATALOG_PREFETCH,
      // Run every job inside the backfill context so source services' touchActivity()
      // skips the user-activity cooldown key — the worker must not throttle itself.
      (job, token) => backfillContext.run(true, () => this.process(job, token)),
      {
        connection: getBullConnection(),
        concurrency: WORKER_CONCURRENCY,
        limiter: { max: WORKER_RATE_MAX, duration: 60_000 },
      },
    );

    this.worker.on("failed", (job, err) => {
      // RateLimitError is handled in process() via moveToDelayed (a DelayedError,
      // not a failure), so anything reaching here is a genuine failure.
      this.logger.warn(`[prefetch] Job ${job?.name} failed: ${err.message}`);
    });

    this.worker.on("error", (err) => {
      this.logger.error(`[prefetch] Worker error: ${err.message}`);
    });

    // FAST lane: own queue + worker so fresh decodes never compete with the
    // deep-drill backlog (whose production>consumption regime starves lifo tails).
    // Same process fn; the shared per-source redis rate gate still protects
    // upstream across BOTH workers.
    this.fastWorker = new Worker(
      QUEUE_NAMES.CATALOG_PREFETCH_FAST,
      (job, token) => backfillContext.run(true, () => this.process(job, token)),
      {
        connection: getBullConnection(),
        concurrency: 4,
        limiter: { max: 120, duration: 60_000 },
      },
    );
    this.fastWorker.on("failed", (job, err) => {
      this.logger.warn(`[prefetch:fast] Job ${job?.name} failed: ${err.message}`);
    });
    this.fastWorker.on("error", (err) => {
      this.logger.error(`[prefetch:fast] Worker error: ${err.message}`);
    });
    this.logger.log("[prefetch] Fast worker started (concurrency=4, 120/min)");

    this.logger.log(
      `[prefetch] Worker started (concurrency=${WORKER_CONCURRENCY}, global ${WORKER_RATE_MAX}/min, ` +
        `per-source ${JSON.stringify(SOURCE_RATE_MAX)}, daily ${JSON.stringify(SOURCE_DAILY_MAX)}, pcatPace=${PCAT_PACE_MS}ms)`,
    );

    // Hourly backfill scan — run IN-PROCESS, not as a BullMQ cron job. A cron
    // scheduler enqueues a delayed marker; BullMQ promotes delayed jobs to the
    // wait-list HEAD with LPUSH (lifo is ignored on delayed promotion) while the
    // worker pops from the tail, so behind a deep prefetch backlog the scan marker
    // is buried for days and never fires. An in-process timer sidesteps the queue
    // entirely; the scan's Phase-1 then enqueues `lifo` init jobs that DO jump the
    // wait list. Prod-host gated (same as the scan body); single-fired via a Redis
    // NX lock so restarts / multiple instances don't double-scan.
    if (isCatalogBackfillEnabled()) {
      const run = () =>
        backfillContext.run(true, () =>
          this.runBackfillScan().catch((err) =>
            this.logger.error(`[backfill] scan crashed: ${(err as Error).message}`),
          ),
        );
      this.backfillKickoff = setTimeout(run, BACKFILL_SCAN_KICKOFF_MS);
      this.backfillInterval = setInterval(run, BACKFILL_SCAN_INTERVAL_MS);
      this.logger.log("[backfill] in-process hourly scan enabled");
    }
  }

  async onModuleDestroy() {
    if (this.backfillKickoff) clearTimeout(this.backfillKickoff);
    if (this.backfillInterval) clearInterval(this.backfillInterval);
    if (this.worker) {
      await this.fastWorker?.close();
      await this.worker.close();
      this.worker = null;
    }
  }

  /**
   * Hourly scan entry point used by the in-process timer. A Redis NX lock keeps it
   * single-fire across restarts and (future) multiple instances; processBackfillScan
   * is idempotent, so a missed cycle is harmless.
   */
  private async runBackfillScan(): Promise<void> {
    const got = await this.redis.setNx("prefetch:backfill:lock", "1", BACKFILL_SCAN_LOCK_TTL_S);
    if (!got) return;
    await this.processBackfillScan();
  }

  private async process(job: Job, token?: string): Promise<void> {
    try {
      const data = job.data as { source?: string };
      // Per-source rate gate FIRST (before the pcat pace) so we don't burn the 15s
      // sleep on a job we're about to defer. Scan jobs are exempt.
      if (
        data.source &&
        (job.name === "prefetch-init" ||
          job.name === "prefetch-children" ||
          job.name === "prefetch-parts")
      ) {
        const lane = (job.data as { fast?: boolean }).fast ? "fast" : "main";
        // GATE ORDER IS LOAD-BEARING — cheapest first, and every gate that can
        // reject the job must run BEFORE any counter is debited.
        //
        // The business-hours window and the cooldown used to sit at the top of
        // each handler, i.e. AFTER the per-minute counter and the daily budget
        // had already been charged. So a job that woke outside the window paid a
        // budget unit to do nothing. Combined with a deferral target of "next UTC
        // midnight" (= 03:00 Europe/Istanbul, six hours before a 09:00 window
        // opens) that closed a loop: the whole daily allowance was burned by
        // no-op wake-ups before the window ever opened, so the source never ran
        // again. Measured on prod 2026-09-20 — pl24 at 600/600 with 0 catalog
        // requests and 0 new categories for the day.
        //
        // 1. Window: pure clock arithmetic, no I/O, and an out-of-window job can
        //    never do useful work — so nothing else is worth spending on it.
        checkTimeWindow(data.source);
        // 2. Cooldown: one Redis TTL read. Pauses the whole worker (see catch).
        await checkCooldown(this.redis, data.source);
        // 3. Per-minute ceiling.
        await this.checkSourceRate(data.source, lane);
        // 4. Daily budget last: a job deferred on any gate above never reaches
        //    here, so only jobs about to do real work are counted. The lane
        //    decides which threshold applies (backfill stops at the main limit,
        //    the user's fast lane may use the full budget).
        await this.checkSourceDailyBudget(data.source, lane);
      }
      if (data.source === "parts-catalogs" && PCAT_PACE_MS > 0) {
        await new Promise((r) => setTimeout(r, PCAT_PACE_MS));
      }
      // PL24 backfill only: pace + jitter. The fast (user) lane is never delayed.
      if (
        data.source === "pl24" &&
        PL24_PACE_MS > 0 &&
        !(job.data as { fast?: boolean }).fast &&
        (job.name === "prefetch-children" || job.name === "prefetch-parts")
      ) {
        await new Promise((r) => setTimeout(r, jitter(PL24_PACE_MS)));
      }

      if (job.name === "backfill-scan") {
        return await this.processBackfillScan();
      }
      if (job.name === "prefetch-init") {
        return await this.processInit(job as Job<PrefetchInitJobData>);
      }
      if (job.name === "prefetch-children") {
        return await this.processChildren(job as Job<PrefetchCategoryJobData>);
      }
      if (job.name === "prefetch-parts") {
        return await this.processParts(job as Job<PrefetchCategoryJobData>);
      }
      this.logger.warn(`[prefetch] Unknown job name: ${job.name}`);
    } catch (err) {
      if (err instanceof RateLimitError) {
        const delayMs = Math.max(1000, err.retryAfterMs);
        if (err.cause === "cooldown") {
          // Pause the WHOLE worker for the cooldown window: the per-source
          // activity key is refreshed by every user request, so per-job defer
          // livelocks (job comes back, key is still set, defer again, …). One
          // global wait is correct: when it ends and cooldown is still active,
          // we'll pause again — at most ~1 cycle per retryAfterMs instead of
          // cycling every queued job through.
          const lane = (job.data as { fast?: boolean }).fast ? this.fastWorker : this.worker;
          await lane?.rateLimit(delayMs);
          this.logger.debug(`[prefetch] Worker paused ${delayMs}ms (cooldown ${job.data.source})`);
          throw Worker.RateLimitError();
        }
        // time-window (off-hours, up to hours) and source-rate (this source hit
        // its per-minute ceiling, ~sub-minute) both defer PER-JOB so other sources
        // keep flowing — only cooldown pauses the whole worker.
        await job.moveToDelayed(Date.now() + delayMs, token);
        this.logger.debug(`[prefetch] Job ${job.name} deferred ${delayMs}ms (${err.cause})`);
        throw new DelayedError();
      }
      throw err;
    }
  }

  /**
   * Init job: walk the category tree for a vehicle and queue sub-jobs.
   */
  private async processInit(job: Job<PrefetchInitJobData>): Promise<void> {
    const { vehicleId, source, fast = false } = job.data;
    this.logger.log(`[prefetch] Init for vehicle=${vehicleId}, source=${source}`);

    // Cooldown + time-window are enforced in process() before the daily budget
    // is debited — see the comment there; re-checking here would be a no-op.

    // Already flagged as poison (tree exceeded CATEGORY_CAP on a prior run) — skip.
    if (await this.redis.exists(this.poisonKey(vehicleId))) {
      this.logger.debug(`[prefetch] Skip poison vehicle=${vehicleId}`);
      return;
    }

    // Verify vehicle still exists
    const [vehicle] = await this.db
      .select({ id: vehicles.id })
      .from(vehicles)
      .where(eq(vehicles.id, vehicleId))
      .limit(1);

    if (!vehicle) {
      this.logger.warn(`[prefetch] Vehicle ${vehicleId} not found, skipping`);
      return;
    }

    await initProgress(this.redis, vehicleId); // resets progress.total — the cap's tree-size proxy

    // Get all top-level categories for this vehicle
    let topCategories = await this.db
      .select()
      .from(categories)
      .where(and(eq(categories.vehicleId, vehicleId), isNull(categories.parentId)));

    // Self-seed: a vehicle that was decoded but never viewed has no categories in
    // DB yet. getCategoryTree fetches & inserts the top-level groups from upstream
    // (PL24/PSA/EMEX) so backfill can proceed without a user opening the page.
    // Cooldown/time-window were already enforced above, so this only runs in-window.
    if (topCategories.length === 0) {
      try {
        await this.categoriesService.getCategoryTree(vehicleId);
        topCategories = await this.db
          .select()
          .from(categories)
          .where(and(eq(categories.vehicleId, vehicleId), isNull(categories.parentId)));
      } catch (err) {
        this.logger.warn(
          `[prefetch] Top-category seed failed for ${vehicleId}: ${(err as Error).message}`,
        );
      }
    }

    if (topCategories.length === 0) {
      this.logger.log(`[prefetch] No categories for vehicle=${vehicleId}`);
      // No categories after a self-seed attempt = no catalog data for this VIN.
      // Count it as residue so Phase-1 stops re-picking it every wave.
      await this.markNoResult(vehicleId);
      await updateProgress(this.redis, vehicleId, {
        status: "completed",
        total: 0,
      });
      return;
    }

    let queued = 0;

    for (const cat of topCategories) {
      if (cat.unavailable) continue;

      // Check if this category already has children in DB
      const [childCheck] = await this.db
        .select({ id: categories.id })
        .from(categories)
        .where(eq(categories.parentId, cat.id))
        .limit(1);

      if (childCheck) {
        // Has children — queue recursive exploration of children
        const children = await this.db
          .select()
          .from(categories)
          .where(eq(categories.parentId, cat.id));

        for (const child of children) {
          if (child.unavailable) continue;
          // Count jobs ACTUALLY queued, not nodes walked — see addJob's doc: an
          // inflated progress.total makes the chain never reach "finished".
          queued += await this.queueCategoryJob(child, vehicleId, source, 1, fast);
        }
      } else if (this.isLeafLinkPath(cat.linkPath, cat.source, cat.hasSubgroups, cat.linkWid)) {
        // Leaf — check if parts already fetched
        const [partCheck] = await this.db
          .select({ id: parts.id })
          .from(parts)
          .where(eq(parts.categoryId, cat.id))
          .limit(1);

        if (
          !partCheck &&
          cat.linkPath &&
          (await this.addJob("prefetch-parts", {
            vehicleId,
            categoryId: cat.id,
            source,
            action: "parts" as const,
            depth: 0,
            fast,
          }))
        ) {
          queued++;
        }
      } else if (
        cat.linkPath &&
        // Non-leaf without children — needs children fetch
        (await this.addJob("prefetch-children", {
          vehicleId,
          categoryId: cat.id,
          source,
          action: "children" as const,
          depth: 0,
          fast,
        }))
      ) {
        queued++;
      }
    }

    await updateProgress(this.redis, vehicleId, { total: queued });
    if (queued === 0) {
      // Nothing left to fetch — the tree is already complete in DB. Without this
      // there is no job to fire incrementCompleted, so progress sits at {0,0}, the
      // vehicle is never marked fullyFetched, and the scan re-picks it every wave
      // forever while burning a daily-budget slot each time.
      this.logger.log(`[prefetch] Nothing to queue for vehicle=${vehicleId} — already complete`);
      await this.finalizeVehicle(vehicleId);
      return;
    }
    this.logger.log(`[prefetch] Queued ${queued} sub-jobs for vehicle=${vehicleId}`);
  }

  /**
   * Fetch children (sub-categories) for a category.
   */
  private async processChildren(job: Job<PrefetchCategoryJobData>): Promise<void> {
    const { vehicleId, categoryId, source, depth, fast = false } = job.data;
    this.logger.log(`[prefetch] Children for category=${categoryId}, depth=${depth}`);

    // Cooldown + time-window are enforced in process() before the daily budget
    // is debited — see the comment there; re-checking here would be a no-op.

    const depthCeiling = maxDepthFor(source, fast);
    if (depth >= depthCeiling) {
      // For the PL24 fast lane this is the normal stopping point, not a problem:
      // deeper nodes are drilled lazily on user click or by the backfill lane.
      const level = source === "pl24" && fast ? "log" : "warn";
      this.logger[level](
        `[prefetch] Depth ceiling ${depthCeiling} reached for category=${categoryId} (source=${source}, lane=${fast ? "fast" : "main"})`,
      );
      return;
    }

    // Anti-poison cap measured from the DB, NOT from progress.total: the ephemeral
    // counter is reset to 0 by every processInit, so a re-picked vehicle could
    // drill another CATEGORY_CAP nodes per round and never trip the guard (the
    // generic-ROOT Opels: 420k/102k/21k categories, ZERO parts). The stored tree
    // is the real, cumulative size.
    const [capRow] = await this.db
      .select({ n: sql<number>`count(*)::int` })
      .from(categories)
      .where(eq(categories.vehicleId, vehicleId))
      .limit(1);
    if ((capRow?.n ?? 0) >= CATEGORY_CAP) {
      this.logger.warn(
        `[prefetch] Category cap ${CATEGORY_CAP} hit for vehicle=${vehicleId} (tree=${capRow?.n}) — marking poison, stop drilling`,
      );
      await this.markPoison(vehicleId);
      return;
    }

    try {
      const children = await this.categoriesService.getChildren(categoryId);

      let queued = 0;
      for (const child of children) {
        if (child.unavailable) continue;
        // Real queued-job count (see addJob) — walking a node that dedupes or
        // already has parts must not inflate progress.total.
        queued += await this.queueCategoryJob(child, vehicleId, source, depth + 1, fast);
      }

      if (queued > 0) {
        await updateProgress(this.redis, vehicleId, {
          total:
            ((await this.redis.getJson<{ total: number }>(`prefetch:progress:${vehicleId}`))
              ?.total || 0) + queued,
        });
      }

      await this.incrementCompleted(vehicleId);
    } catch (err) {
      if (err instanceof RateLimitError) throw err;
      this.logger.error(
        `[prefetch] Children fetch failed for ${categoryId}: ${(err as Error).message}`,
      );
      await this.incrementErrors(vehicleId);
    }
  }

  /**
   * Fetch parts + schema for a leaf category.
   */
  private async processParts(job: Job<PrefetchCategoryJobData>): Promise<void> {
    const { vehicleId, categoryId, source } = job.data;
    this.logger.log(`[prefetch] Parts for category=${categoryId}`);

    // Cooldown + time-window are enforced in process() before the daily budget
    // is debited — see the comment there; re-checking here would be a no-op.

    try {
      await this.categoriesService.getCategoryWithParts(categoryId);
      await this.incrementCompleted(vehicleId);
    } catch (err) {
      if (err instanceof RateLimitError) throw err;
      this.logger.error(
        `[prefetch] Parts fetch failed for ${categoryId}: ${(err as Error).message}`,
      );
      await this.incrementErrors(vehicleId);
    }
  }

  /**
   * Backfill scan (hourly cron): find decoded vehicles whose catalog isn't fully
   * prefetched and queue prefetch-init for them. prefetch-init is idempotent — it
   * only fetches missing categories/parts, so re-running over a complete vehicle is
   * cheap. Self-throttling via a queue-depth guard, per-source cooldown, and the
   * business-hours window. Goal: no decoded vehicle is left without catalog data.
   */
  private async processBackfillScan(): Promise<void> {
    // Production (sase.tr) only — dev.sase.tr uses a separate DB and must never
    // sweep/scrape. Defense-in-depth in case a scan job lands here via a stale
    // scheduler. NODE_ENV is "production" on both envs, so gate on the prod host.
    if (!isCatalogBackfillEnabled()) {
      this.logger.log("[backfill] Skip — not prod host");
      return;
    }

    // Live-tunable knobs via the `cfg-backfill-tuning` remote-config flag — change
    // batch size / backlog ceiling / business-hours gating from the PostHog UI
    // without a redeploy. Falls back to the compiled-in defaults on any miss.
    const cfg = await this.posthog.payload<{
      batchSize?: number;
      maxBacklog?: number;
      businessHoursOnly?: boolean;
    }>("cfg-backfill-tuning", {});
    const batchSize =
      typeof cfg.batchSize === "number" && cfg.batchSize > 0 ? cfg.batchSize : BACKFILL_BATCH_SIZE;
    const maxBacklog =
      typeof cfg.maxBacklog === "number" && cfg.maxBacklog > 0
        ? cfg.maxBacklog
        : BACKFILL_MAX_BACKLOG;

    // Self-throttle: when the queue is already deep, suspend only the Phase-2
    // rolling rescan (the part that piles on). Phase-1 still runs every wave so
    // genuinely-empty vehicles keep getting onboarded through the fast lane even
    // while a large deep-drill backlog is still draining — otherwise a single
    // backlog spike freezes new-vehicle coverage until the whole queue clears.
    // Two INDEPENDENT ceilings (Phase-1 fast lane is never gated, so genuinely
    // empty vehicles keep getting onboarded):
    //  1. pressure = waiting + active + delayed-due-within-PRESSURE_HORIZON_MS,
    //     both lanes, vs maxBacklog — "are the workers actually swamped?".
    //     Budget/off-hours defers are parked hours out and no longer count, so a
    //     spent daily budget stops freezing Phase-2 for the rest of the day.
    //  2. total = every pending job, both lanes, vs HARD_MAX_TOTAL_JOBS — the
    //     runaway backstop that `delayed` used to provide by accident.
    const depth = await this.getQueueDepth();
    const headroom = Math.max(0, HARD_MAX_TOTAL_JOBS - depth.total);
    // Admission control in JOB units: one wave fans out to batchSize * ~400 jobs
    // long before the next hourly scan can react, so shrink the wave as the pool
    // fills instead of stepping over the cap by a whole batch.
    const phase2Budget = Math.min(batchSize, Math.floor(headroom / EST_JOBS_PER_VEHICLE));
    const phase2Allowed = depth.pressure <= maxBacklog && phase2Budget > 0;
    if (!phase2Allowed) {
      this.logger.log(
        `[backfill] Phase-1 (fast lane) only — pressure=${depth.pressure}/${maxBacklog}, ` +
          `total=${depth.total}/${HARD_MAX_TOTAL_JOBS} (delayed=${depth.delayed}, imminent=${depth.imminent})`,
      );
    }
    if (HARD_MAX_TOTAL_JOBS > 0 && depth.total > HARD_MAX_TOTAL_JOBS * 1.5) {
      // Should be unreachable — admission control caps intake once per hour. If it
      // fires, real fan-out is far above EST_JOBS_PER_VEHICLE.
      this.logger.error(
        `[backfill] Queue pool ${depth.total} > 1.5x hard cap ${HARD_MAX_TOTAL_JOBS} — investigate fan-out`,
      );
    }

    // Only target sources eligible right now: not in cooldown (user active),
    // inside their scrape window, AND still inside today's main-lane budget.
    // The budget check is THE key guard: deferring at job level only MOVES work
    // into `delayed`, it does not stop PRODUCING it — and budget defers never
    // exhaust attempts (moveToDelayed skipAttempt), so nothing drops them. Once a
    // source's main-lane budget is spent the scan must stop feeding it for the
    // rest of the UTC day; feeding a throttled source is exactly how the 470k
    // runaway was built. This also preserves the fast-lane reserve for real users.
    const eligible: string[] = [];
    for (const s of BACKFILL_SOURCES) {
      // Background backfill is opt-in per source; pl24 defaults to OFF so bulk
      // load can never burn the one surviving account by accident.
      if (s === "pl24" && !isPl24BackfillEnabled()) continue;
      if (await this.redis.exists(`prefetch:activity:${s}`)) continue;
      if (cfg.businessHoursOnly !== false && !isWithinTimeWindow(s)) continue;
      const mainLimit = this.dailyMainLimit(s);
      if (mainLimit > 0) {
        const spent = Number((await this.redis.get(this.dailyKey(s))) ?? 0);
        if (spent >= mainLimit) {
          this.logger.log(
            `[backfill] ${s} daily budget spent (${spent}/${mainLimit}) — source skipped this wave`,
          );
          continue;
        }
      }
      eligible.push(s);
    }
    if (eligible.length === 0) {
      this.logger.log("[backfill] Skip — no eligible sources (cooldown / off-hours)");
      return;
    }

    const picked: Array<{ id: string; source: string; fast: boolean }> = [];
    const seen = new Set<string>();
    const overfetch = batchSize * 4; // headroom for in-flight skips

    const tryPick = async (
      v: { id: string; source: string | null },
      fast: boolean,
      limit: number = batchSize,
    ): Promise<void> => {
      if (picked.length >= limit || seen.has(v.id) || !v.source) return;
      // TWO different in-flight guards, both must be clear:
      //  - prefetch:scheduled:backfill:<id> — ours (36h, covers a budget-parked chain)
      //  - prefetch:scheduled:<id>          — the USER decode path's (vehicles.service.ts
      //    schedulePrefetch, short TTL). We deliberately stopped WRITING the bare
      //    key: our long TTL made schedulePrefetch() silently skip the user's
      //    fresh-decode fast-lane init.
      if (await this.redis.exists(this.scheduledKey(v.id))) return;
      if (await this.redis.exists(`prefetch:scheduled:${v.id}`)) return; // user chain in flight
      // Skip exhausted residue: vehicles whose prefetch keeps finishing with zero
      // parts (no catalog data). They'd otherwise be re-picked every wave forever.
      const noResult = await this.redis.get(this.noResultKey(v.id));
      if (noResult && Number(noResult) >= NORESULT_MAX_ATTEMPTS) return;
      // Skip vehicles whose full tree is already fetched (marker set on chain
      // completion with parts). Stops Phase-2 from re-walking finished vehicles
      // every wave; the marker's TTL re-validates them periodically. Phase-1
      // (zero-parts) vehicles never carry this marker, so this is a no-op there.
      if (await this.redis.exists(this.completeKey(v.id))) return;
      // Skip poison vehicles (generic-model / over-cap catalog explosions).
      if (await this.redis.exists(this.poisonKey(v.id))) return;
      seen.add(v.id);
      picked.push({ id: v.id, source: v.source, fast });
    };

    // Phase 1 — clear the obvious backlog first: decoded vehicles with zero parts.
    const noParts = await this.db
      .select({ id: vehicles.id, source: vehicles.source })
      .from(vehicles)
      .where(
        and(
          inArray(vehicles.source, eligible),
          notExists(
            this.db.select({ one: sql`1` }).from(parts).where(eq(parts.vehicleId, vehicles.id)),
          ),
        ),
      )
      .orderBy(asc(vehicles.createdAt))
      .limit(overfetch);

    for (const v of noParts) await tryPick(v, true);

    // Phase 2 — rolling rescan of vehicles that are NOT fully fetched, to gap-fill
    // partials. Targets the DURABLE `fullyFetched` flag (indexed) instead of
    // walking the whole fleet and relying only on the ephemeral 21-day
    // `prefetch:complete:` marker — a Redis flush would otherwise re-drill every
    // vehicle at once. The fullyFetchedAt clause keeps the periodic re-validation
    // the TTL used to provide. Gated by the ceilings above + a JOB-unit budget.
    const phase2Limit = Math.min(batchSize, picked.length + phase2Budget);
    if (phase2Allowed && picked.length < phase2Limit) {
      const cursorObj = await this.redis.getJson<{ ts: string }>(BACKFILL_CURSOR_KEY);
      const cursor = cursorObj?.ts ? new Date(cursorObj.ts) : new Date(0);

      const rolling = await this.db
        .select({ id: vehicles.id, source: vehicles.source, createdAt: vehicles.createdAt })
        .from(vehicles)
        .where(
          and(
            inArray(vehicles.source, eligible),
            gt(vehicles.createdAt, cursor),
            sql`(${vehicles.fullyFetched} = false OR ${vehicles.fullyFetchedAt} < now() - interval '21 days')`,
          ),
        )
        .orderBy(asc(vehicles.createdAt))
        .limit(overfetch);

      if (rolling.length === 0 && cursor.getTime() > 0) {
        // Reached the end — restart the rescan from the beginning next wave.
        await this.redis.setJson(
          BACKFILL_CURSOR_KEY,
          { ts: new Date(0).toISOString() },
          30 * 86400,
        );
      }

      let lastTs: Date | null = null;
      for (const v of rolling) {
        lastTs = v.createdAt;
        await tryPick(v, false, phase2Limit);
      }
      if (lastTs) {
        await this.redis.setJson(BACKFILL_CURSOR_KEY, { ts: lastTs.toISOString() }, 30 * 86400);
      }
    }

    if (picked.length === 0) {
      this.logger.log("[backfill] No candidates this wave");
      return;
    }

    for (const v of picked) await this.enqueueInit(v.id, v.source, v.fast);
    const fastCount = picked.filter((v) => v.fast).length;
    this.logger.log(
      `[backfill] Queued ${picked.length} vehicle(s) (${fastCount} fast-lane, ` +
        `sources=${eligible.join(",")}, pressure=${depth.pressure}, total=${depth.total}, ` +
        `phase2Budget=${phase2Budget})`,
    );
  }

  /** Queue a prefetch-init for a vehicle and set the in-flight guard. */
  private async enqueueInit(vehicleId: string, source: string, fast = false): Promise<void> {
    const q = fast ? this.fastQueue : this.queue;
    await q.add(
      "prefetch-init",
      { vehicleId, source: source as PrefetchInitJobData["source"], fast },
      {
        removeOnComplete: { count: 1000 },
        removeOnFail: { count: 5000 },
        // Fast lane (Phase-1): lifo so a zero-parts vehicle's whole chain jumps
        // the deep-drill backlog instead of queueing behind it (see addJob).
        ...(fast ? { lifo: true } : {}),
      },
    );
    // Guard cleared on completion (incrementCompleted) or by TTL if the run dies.
    await this.redis.set(this.scheduledKey(vehicleId), "1", BACKFILL_SCHEDULED_TTL);
  }

  // ==================== Helpers ====================

  private async queueCategoryJob(
    cat: {
      id: string;
      linkPath: string | null;
      linkWid?: string | null;
      source: string;
      unavailable: boolean;
      hasSubgroups?: boolean | null;
    },
    vehicleId: string,
    source: string,
    depth: number,
    fast = false,
  ): Promise<number> {
    if (cat.unavailable) return 0;

    // A per-part detail node (Mitsubishi `partInfoTable` /details/vinpartinfo) is
    // neither a group nor a listing: its parent's response already carried the
    // part. Queueing it costs one upstream request and returns nothing. Prod had
    // 19,576 of these, with 2 parts between them.
    if (
      cat.source === "pl24" &&
      isPl24PartDetailNode({ linkPath: cat.linkPath, linkWid: cat.linkWid })
    ) {
      return 0;
    }

    if (this.isLeafLinkPath(cat.linkPath, cat.source, cat.hasSubgroups, cat.linkWid)) {
      // Leaf — check if already has parts
      const [partCheck] = await this.db
        .select({ id: parts.id })
        .from(parts)
        .where(eq(parts.categoryId, cat.id))
        .limit(1);

      if (!partCheck && cat.linkPath) {
        return (await this.addJob("prefetch-parts", {
          vehicleId,
          categoryId: cat.id,
          source: source as "pl24" | "emex",
          action: "parts" as const,
          depth,
          fast,
        }))
          ? 1
          : 0;
      }
      return 0;
    }
    if (cat.linkPath && depth < MAX_DEPTH) {
      // Non-leaf within the depth cap — explore children. The `depth < MAX_DEPTH`
      // gate mirrors processChildren's early-return: without it we'd enqueue a
      // prefetch-children job that processChildren just drops, burning a rate-limit
      // slot on a no-op (this was ~85% of the queue at MAX_DEPTH=2).
      const [childCheck] = await this.db
        .select({ id: categories.id })
        .from(categories)
        .where(eq(categories.parentId, cat.id))
        .limit(1);

      if (childCheck) {
        // Already has children — explore them recursively
        const children = await this.db
          .select()
          .from(categories)
          .where(eq(categories.parentId, cat.id));

        let n = 0;
        for (const child of children) {
          if (child.unavailable) continue;
          n += await this.queueCategoryJob(child, vehicleId, source, depth + 1, fast);
        }
        return n;
      }
      return (await this.addJob("prefetch-children", {
        vehicleId,
        categoryId: cat.id,
        source: source as "pl24" | "emex",
        action: "children" as const,
        depth,
        fast,
      }))
        ? 1
        : 0;
    }
    return 0;
  }

  private isLeafLinkPath(
    linkPath: string | null,
    source: string,
    hasSubgroups?: boolean | null,
    linkWid?: string | null,
  ): boolean {
    if (!linkPath) return false;
    // EMEX: Vehicle.aspx group nodes are parents to drill; Unit.aspx (hierarchical
    // tree) and QuickDetails.aspx (legacy flat) leaves carry parts.
    if (source === "emex") return !linkPath.includes("Vehicle.aspx");
    // parts-catalogs: EVERY node has a pcat: linkPath, so the old "pcat: prefix =
    // leaf" rule mis-flagged every mid-group folder as a leaf — the worker fetched
    // its "parts" (a 1-level drill that just revealed sub-groups) and never
    // recursed, leaving deep pcat trees a single level shallow (no parts → empty
    // part counts, dead cross-tree search). Use the captured hasSubgroups flag
    // instead: a node is a parts leaf only when it is NOT a known parent group, so
    // folders queue a children job and the recursion drills to full depth. Unknown
    // flag (null, rare pre-migration rows) → treated as leaf, preserving the old
    // 1-level behaviour for those.
    if (source === "parts-catalogs") return hasSubgroups !== true;
    // PL24: one shared classifier (integrations/pl24/pl24-tree). The old inline
    // list was case-sensitive, so p5psa/p5volvo's camelCase `/details/vin/
    // bomDetails` was never recognised as a leaf and its parts were never
    // prefetched.
    // `linkWid` is passed through on purpose: it is the reliable cross-brand
    // marker and the read path has always used it, but this queueing path used
    // to drop it and fall back to path matching alone. That is why Mitsubishi's
    // `detailsTable` parts list was queued as a group here even after the shared
    // classifier learned about it.
    return isPl24LeafNode({ linkPath, hasSubgroups, linkWid });
  }

  /**
   * Queue one sub-job. Returns whether a NEW job was actually created.
   *
   * The return value is the fix for a completion-accounting bug: `progress.total`
   * is the chain's denominator (incrementCompleted fires "finished" at
   * `completed >= total`), but callers used to increment it for every node they
   * WALKED — including nodes where this deterministic jobId deduped the add, and
   * leaves that already had parts. An inflated total means the chain never
   * reaches "finished", so the vehicle is never marked fullyFetched and the scan
   * re-picks it every wave forever (the ~25 vehicles/day plateau). Queue
   * behaviour is UNCHANGED — BullMQ already no-op'd a duplicate jobId.
   */
  private async addJob(name: string, data: PrefetchCategoryJobData): Promise<boolean> {
    // BullMQ rejects custom job IDs containing ":" (its key separator), so use
    // "-" instead. The values are UUIDs — the ID only needs to be deterministic
    // (for dedup), not parseable.
    const jobId = `prefetch-${data.vehicleId}-${data.categoryId}-${data.action}`;
    const opts: Record<string, unknown> = {
      jobId,
      // Fast lane (Phase-1 / reactive): add with `lifo` so the job RPUSHes to the
      // TAIL of the wait list, where BullMQ's RPOPLPUSH picks it next — i.e. ahead
      // of the deep deep-drill backlog already sitting in wait. (BullMQ 5 drains
      // the wait list before the prioritized ZSET, so `priority` would do the
      // OPPOSITE here and starve the job behind the backlog; lifo is correct.)
      ...(data.fast ? { lifo: true } : {}),
    };
    const q = data.fast ? this.fastQueue : this.queue;
    if (await q.getJob(jobId)) return false;

    // parts-catalogs pacing is handled per-job in process() (PCAT_PACE_MS) + the
    // limiter. The old cumulative `index * 20s` delay was pathological (the Nth
    // leaf of a vehicle waited N*20s) and is gone.

    await q.add(name, data, opts);
    return true;
  }

  private async incrementCompleted(vehicleId: string): Promise<void> {
    const progress = await this.redis.getJson<{
      completed: number;
      total: number;
    }>(`prefetch:progress:${vehicleId}`);
    if (!progress) return;

    const completed = (progress.completed || 0) + 1;
    const isFinished = completed >= progress.total;

    await updateProgress(this.redis, vehicleId, {
      completed,
      ...(isFinished ? { status: "completed" } : {}),
    });

    if (isFinished) {
      this.logger.log(`[prefetch] Completed all jobs for vehicle=${vehicleId}`);
      await this.finalizeVehicle(vehicleId);
    }
  }

  /**
   * Settle a vehicle whose prefetch chain is done: mark it complete (with parts)
   * or count it as residue (still zero parts), then clear the run's Redis state.
   * Called both when the last sub-job finishes AND when processInit finds there
   * is nothing left to queue — otherwise an already-complete tree would never be
   * settled and the scan would re-pick it forever.
   */
  private async finalizeVehicle(vehicleId: string): Promise<void> {
    // Genuine residue: the whole chain finished but the vehicle still has no
    // parts (all leaves empty / no catalog data). Count it so Phase-1 stops
    // re-picking it every wave.
    const [hasPart] = await this.db
      .select({ id: parts.id })
      .from(parts)
      .where(eq(parts.vehicleId, vehicleId))
      .limit(1);
    if (hasPart) {
      // Fully fetched with parts → mark complete so the Phase-2 rescan skips it
      // (re-validates after the TTL). A chain with any failed job never reaches
      // isFinished, so partially-fetched vehicles are never marked — they keep
      // getting gap-filled.
      await this.redis.set(this.completeKey(vehicleId), "1", COMPLETE_TTL_S);
      // Durable completeness flag (never expires) — the reliable "% fully
      // fetched" measurement, independent of the ephemeral Redis marker.
      // Re-set on every re-drill completion so fullyFetchedAt tracks last-verified.
      await this.db
        .update(vehicles)
        .set({ fullyFetched: true, fullyFetchedAt: new Date() })
        .where(eq(vehicles.id, vehicleId));
    } else {
      await this.markNoResult(vehicleId);
    }
    // Clean up Redis keys — data is in PostgreSQL now
    await this.redis.del(this.scheduledKey(vehicleId));
    await this.redis.del(`prefetch:progress:${vehicleId}`);
  }

  private async incrementErrors(vehicleId: string): Promise<void> {
    const progress = await this.redis.getJson<{ errors: number }>(`prefetch:progress:${vehicleId}`);
    if (!progress) return;
    await updateProgress(this.redis, vehicleId, {
      errors: (progress.errors || 0) + 1,
    });
  }

  private noResultKey(vehicleId: string): string {
    return `prefetch:noresult:${vehicleId}`;
  }

  /**
   * Backfill's own in-flight guard — namespaced so it can't shadow the user
   * decode path's `prefetch:scheduled:<id>` (vehicles.service.ts reads that key
   * and skips the fresh-decode fast-lane init while it is set).
   */
  private scheduledKey(vehicleId: string): string {
    return `prefetch:scheduled:backfill:${vehicleId}`;
  }

  /**
   * Queue depth across BOTH lanes, split into "pressure" (work the workers will
   * pick up within PRESSURE_HORIZON_MS) and "total" (everything pending).
   *
   * BullMQ stores delayed jobs in a ZSET scored `dueMs * 0x1000 + seq` (12-bit
   * collision counter — addDelayedJob / moveToDelayed lua), so "due within X ms"
   * is a plain ZCOUNT with the bound encoded the same way. RedisService exposes
   * no zcount, so we borrow each queue's OWN ioredis connection and its toKey()
   * rather than hand-building `bull:<name>:delayed`.
   */
  private async getQueueDepth(): Promise<{
    pressure: number;
    total: number;
    imminent: number;
    delayed: number;
  }> {
    // String bound so ioredis can't render it in exponent notation.
    const maxScore = String((Date.now() + PRESSURE_HORIZON_MS + 1) * 0x1000 - 1);
    let pressure = 0;
    let total = 0;
    let imminent = 0;
    let delayed = 0;
    for (const q of [this.queue, this.fastQueue]) {
      const c = await q.getJobCounts("waiting", "active", "delayed", "prioritized");
      const live = (c.waiting ?? 0) + (c.active ?? 0);
      const d = c.delayed ?? 0;
      // Fail CLOSED: if the ZCOUNT can't be taken, count every delayed job as
      // pressure — fall back to the old over-conservative gate rather than
      // silently unlocking Phase-2 with no visibility. A repeating warn here
      // means Phase-2 is suspended again.
      let due = d;
      try {
        const client = await q.client;
        due = await client.zcount(q.toKey("delayed"), "-inf", maxScore);
      } catch (err) {
        this.logger.warn(
          `[backfill] delayed zcount failed on ${q.name} (${(err as Error).message}) — counting all delayed as pressure`,
        );
      }
      pressure += live + due;
      total += live + d + (c.prioritized ?? 0);
      imminent += due;
      delayed += d;
    }
    return { pressure, total, imminent, delayed };
  }

  private completeKey(vehicleId: string): string {
    return `prefetch:complete:${vehicleId}`;
  }

  private poisonKey(vehicleId: string): string {
    return `prefetch:poison:${vehicleId}`;
  }

  /** Flag a vehicle as poison (its tree blew past CATEGORY_CAP — a generic
   *  root-catalog explosion from a decode that didn't resolve the model) so
   *  init/scan skip it, and clear its in-flight + progress state. Brand-agnostic:
   *  only an actual explosion trips it, so small legit generic-model PSA vehicles
   *  (which stay ~200 categories and DO yield parts) are never affected. */
  private async markPoison(vehicleId: string): Promise<void> {
    await this.redis.set(this.poisonKey(vehicleId), "1", POISON_TTL_S);
    await this.redis.del(this.scheduledKey(vehicleId));
    await this.redis.del(`prefetch:progress:${vehicleId}`);
  }

  /**
   * Per-source fixed-window (60s) rate limit. Increments the source's window
   * counter; once it exceeds the source's ceiling, throws RateLimitError
   * ("source-rate") so process() defers THIS job to the window's end while other
   * sources keep flowing. 0 / unknown source ceiling = unlimited.
   */
  private async checkSourceRate(source: string, lane: "main" | "fast" = "main"): Promise<void> {
    // Per-LANE budgets: the main worker's deep-drill backlog would otherwise eat
    // the whole per-source window and starve the fast lane with 18s defers
    // (observed: fresh BMW init couldn't get a single pcat slot). Totals per
    // source stay the same as before, so upstream load is unchanged.
    const total = SOURCE_RATE_MAX[source] ?? 0;
    // Park already-queued pl24 MAIN-lane jobs while background backfill is off
    // (long defer, no attempt consumed) — the eligibility scan stops producing
    // new ones, this stops an existing backlog from draining through the account.
    if (source === "pl24" && lane === "main" && !isPl24BackfillEnabled()) {
      throw new RateLimitError(15 * 60_000, "source-rate");
    }
    if (total <= 0) return;
    const fastShare = Math.max(1, Math.floor(total / 3));
    const max = lane === "fast" ? fastShare : total - fastShare;
    const windowMs = 60_000;
    const now = Date.now();
    const key = `prefetch:rate:${lane}:${source}:${Math.floor(now / windowMs)}`;
    const n = await this.redis.incr(key);
    if (n === 1) await this.redis.expire(key, 61);
    if (n > max) {
      throw new RateLimitError(windowMs - (now % windowMs) + 50, "source-rate");
    }
  }

  /**
   * Per-source rolling-day budget (storm guard). UTC-day fixed window in Redis;
   * when the source's daily job count exceeds its ceiling, defer the job until
   * the window rolls so a long run can't drain the proxy budget. 0 = unlimited.
   */
  private async checkSourceDailyBudget(
    source: string,
    lane: "main" | "fast" = "main",
  ): Promise<void> {
    const max = SOURCE_DAILY_MAX[source] ?? 0;
    if (max <= 0) return;
    const limit = lane === "fast" ? max : this.dailyMainLimit(source);
    const dayMs = 86_400_000;
    const now = Date.now();
    const key = this.dailyKey(source, now);

    // READ-then-INCR (was INCR-then-check). A REJECTED attempt must not count:
    // the old order inflated the counter with every defer (observed 48531 against
    // a 30000 budget), which (a) made the number useless for capacity decisions
    // and (b) — now that the scan reads the same counter to stop feeding a spent
    // source — would let pure defer churn lock the source out. Worst-case
    // overshoot under the read/incr race is WORKER_CONCURRENCY jobs: acceptable.
    const n = Number((await this.redis.get(key)) ?? 0);
    if (n >= limit) {
      // Defer to the next UTC day, plus up to 45min of JITTER. Without jitter every
      // deferred job wakes in the SAME millisecond (observed: 11495 jobs all at
      // 00:00:01 UTC) — the promotion lands as one burst and the pressure signal
      // flaps. Other sources keep flowing (per-job defer, not a worker pause).
      const rollover = now + dayMs - (now % dayMs) + 1000 + Math.floor(Math.random() * 45 * 60_000);
      // Land the retry INSIDE the source's scrape window. The UTC rollover alone
      // is 03:00 Europe/Istanbul, so with a 09:00 window every deferred job woke
      // six hours early, failed the window check and was deferred again — the
      // other half of the deadlock fixed in process(). alignToWindow is a no-op
      // when no window is configured (the default).
      const msLeft = Math.max(1000, alignToWindow(source, rollover) - now);
      if (n === limit) {
        this.logger.warn(
          `[prefetch] ${source} daily budget hit (lane=${lane}, ${n}/${limit} of ${max}) — ` +
            `deferring ~${Math.round(msLeft / 3_600_000)}h to the next in-window slot`,
        );
      }
      throw new RateLimitError(msLeft, "source-rate");
    }
    const after = await this.redis.incr(key);
    if (after === 1) await this.redis.expire(key, 90_000); // ~25h, outlives the window
  }

  /** Redis key for a source's UTC-day budget counter (shared by both lanes). */
  private dailyKey(source: string, now = Date.now()): string {
    return `prefetch:daily:${source}:${Math.floor(now / 86_400_000)}`;
  }

  /** Main (backfill) lane threshold — the fast lane's reserve is never available
   *  to backfill, so a sweep can't park the user's fresh-decode chain. */
  private dailyMainLimit(source: string): number {
    const max = SOURCE_DAILY_MAX[source] ?? 0;
    return max <= 0 ? 0 : Math.floor(max * (1 - DAILY_FAST_RESERVE));
  }

  /** Record that a backfill attempt finished with the vehicle still at zero parts. */
  private async markNoResult(vehicleId: string): Promise<void> {
    const key = this.noResultKey(vehicleId);
    const n = await this.redis.incr(key);
    await this.redis.expire(key, NORESULT_TTL_S);
    this.logger.debug(`[prefetch] no-result #${n} for vehicle=${vehicleId}`);
  }
}