feat(api): hourly catalog backfill for decoded-but-unfetched vehicles

prefetch was reactive-only (on decode) and processInit read top-level
categories straight from DB, so a vehicle decoded but never viewed got
no catalog. Add a production-only sweep so no decoded vehicle is left
without catalog data.

- processInit self-seeds top categories via getCategoryTree when DB has
  none (fetches+inserts top groups from PL24/PSA/EMEX), closing the
  never-viewed gap for both reactive and backfill paths
- new backfill-scan job + hourly cron: phase 1 queues decoded vehicles
  with zero parts, phase 2 rolling createdAt-cursor rescan of all decoded
  vehicles (prefetch-init is idempotent → gap-fills partial ones)
- guardrails: skip wave if queue backlog > 1000, per-source cooldown,
  business-hours window (isWithinTimeWindow), batch <=20, in-flight guard
- PRODUCTION ONLY: gated on NODE_ENV both at cron registration and in
  processBackfillScan; dev has a separate DB and must not scrape

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

apps/api/src/jobs/jobs.module.ts

@@ -88,6 +88,30 @@ export class JobsModule implements OnModuleInit, OnModuleDestroy {
       },
     );
     console.log("[jobs] Registered lifecycle-email cron: 0 9 * * *");
+
+    // Catalog backfill: every hour at :00. Scans for decoded vehicles whose catalog
+    // isn't fully prefetched and queues them. Self-throttled (queue-depth guard),
+    // cooldown- and business-hours-aware — handled by PrefetchWorkerService.
+    // PRODUCTION ONLY: dev runs against a separate DB and must not sweep/scrape.
+    if (process.env.NODE_ENV === "production") {
+      await this.catalogPrefetchQueue.upsertJobScheduler(
+        "catalog-backfill-hourly",
+        { pattern: "0 * * * *" },
+        {
+          name: "backfill-scan",
+          data: {},
+          opts: {
+            removeOnComplete: { count: 48 },
+            removeOnFail: { count: 100 },
+          },
+        },
+      );
+      console.log("[jobs] Registered catalog-backfill cron: 0 * * * *");
+    } else {
+      // Clean up any stale scheduler (e.g. if NODE_ENV changed) and stay idle.
+      await this.catalogPrefetchQueue.removeJobScheduler("catalog-backfill-hourly").catch(() => {});
+      console.log("[jobs] Skipped catalog-backfill cron (NODE_ENV != production)");
+    }
   }
 
   async onModuleDestroy() {

apps/api/src/jobs/prefetch-utils.ts

@@ -24,22 +24,34 @@ export async function checkCooldown(redis: RedisService, source: string): Promis
   }
 }
 
-/**
- * Check PL24 business hours (09:00–18:00 Europe/Istanbul).
- * Throws RateLimitError with delay until next 09:00 if outside window.
- */
-export function checkTimeWindow(source: string): void {
-  if (source !== "pl24" && source !== "parts-catalogs") return;
-
+/** Current hour (0–23) in Europe/Istanbul. */
+function currentIstanbulHour(): number {
   const hourStr = new Intl.DateTimeFormat("en-US", {
     timeZone: "Europe/Istanbul",
     hour: "numeric",
     hour12: false,
   }).format(new Date());
-  const h = Number.parseInt(hourStr, 10);
+  return Number.parseInt(hourStr, 10);
+}
 
+/**
+ * Whether `source` may be scraped right now.
+ * PL24 → 09:00–18:00, parts-catalogs → 09:00–19:00 (Europe/Istanbul).
+ * EMEX and everything else have no window (always true).
+ */
+export function isWithinTimeWindow(source: string): boolean {
+  if (source !== "pl24" && source !== "parts-catalogs") return true;
+  const h = currentIstanbulHour();
   const endHour = source === "parts-catalogs" ? 19 : 18;
-  if (h < 9 || h >= endHour) {
+  return h >= 9 && h < endHour;
+}
+
+/**
+ * Check PL24 business hours (09:00–18:00 Europe/Istanbul).
+ * Throws RateLimitError with delay until next 09:00 if outside window.
+ */
+export function checkTimeWindow(source: string): void {
+  if (!isWithinTimeWindow(source)) {
     throw new RateLimitError(msUntilNext9AM());
   }
 }

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

@@ -6,7 +6,7 @@ import {
   type OnModuleInit,
 } from "@nestjs/common";
 import { type Job, type Queue, Worker } from "bullmq";
-import { and, eq, isNull } from "drizzle-orm";
+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";
@@ -17,6 +17,7 @@ import {
   checkCooldown,
   checkTimeWindow,
   initProgress,
+  isWithinTimeWindow,
   updateProgress,
 } from "./prefetch-utils";
 import { PrefetchCategoryJobData, PrefetchInitJobData } from "./prefetch.types";
@@ -24,6 +25,18 @@ import { CATALOG_PREFETCH_QUEUE } from "./queues/catalog-prefetch.queue";
 
 const MAX_DEPTH = 5;
 
+// ── Backfill scan (hourly cron) tuning ──
+/** Vehicles queued per scan wave. */
+const BACKFILL_BATCH_SIZE = 20;
+/** 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. */
+const BACKFILL_SCHEDULED_TTL = 6 * 60 * 60;
+/** Only these decode sources have catalogs worth prefetching. */
+const BACKFILL_SOURCES = ["pl24", "emex", "parts-catalogs"];
+/** Redis key holding the rolling rescan cursor (last createdAt seen). */
+const BACKFILL_CURSOR_KEY = "prefetch:backfill:cursor";
+
 @Injectable()
 export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
   private readonly logger = new Logger(PrefetchWorkerService.name);
@@ -74,6 +87,9 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
       await new Promise((r) => setTimeout(r, 15_000));
     }
 
+    if (job.name === "backfill-scan") {
+      return this.processBackfillScan();
+    }
     if (job.name === "prefetch-init") {
       return this.processInit(job as Job<PrefetchInitJobData>);
     }
@@ -113,11 +129,29 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
     await initProgress(this.redis, vehicleId);
 
     // Get all top-level categories for this vehicle
-    const topCategories = await this.db
+    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}`);
       await updateProgress(this.redis, vehicleId, {
@@ -251,6 +285,124 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
     }
   }
 
+  /**
+   * 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 only — dev uses a separate DB and must never sweep/scrape.
+    // Defense-in-depth in case a scan job lands here via a shared Redis/queue.
+    if (process.env.NODE_ENV !== "production") {
+      this.logger.log("[backfill] Skip — NODE_ENV != production");
+      return;
+    }
+
+    // Self-throttle: don't pile on if the queue is already deep — let it drain.
+    const counts = await this.queue.getJobCounts("waiting", "delayed", "active");
+    const backlog = (counts.waiting ?? 0) + (counts.delayed ?? 0) + (counts.active ?? 0);
+    if (backlog > BACKFILL_MAX_BACKLOG) {
+      this.logger.log(`[backfill] Skip — queue backlog ${backlog} > ${BACKFILL_MAX_BACKLOG}`);
+      return;
+    }
+
+    // Only target sources eligible right now: not in cooldown (user active) and
+    // inside their scrape window (PL24/parts-catalogs office hours; EMEX always).
+    const eligible: string[] = [];
+    for (const s of BACKFILL_SOURCES) {
+      if (await this.redis.exists(`prefetch:activity:${s}`)) continue;
+      if (!isWithinTimeWindow(s)) 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 }> = [];
+    const seen = new Set<string>();
+    const overfetch = BACKFILL_BATCH_SIZE * 4; // headroom for in-flight skips
+
+    const tryPick = async (v: { id: string; source: string | null }): Promise<void> => {
+      if (picked.length >= BACKFILL_BATCH_SIZE || seen.has(v.id) || !v.source) return;
+      if (await this.redis.exists(`prefetch:scheduled:${v.id}`)) return; // already in flight
+      seen.add(v.id);
+      picked.push({ id: v.id, source: v.source });
+    };
+
+    // 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);
+
+    // Phase 2 — rolling rescan of ALL decoded vehicles to gap-fill partially-fetched
+    // ones. A createdAt cursor walks forward and wraps around at the end.
+    if (picked.length < BACKFILL_BATCH_SIZE) {
+      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)))
+        .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);
+      }
+      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);
+    this.logger.log(
+      `[backfill] Queued ${picked.length} vehicle(s) (sources=${eligible.join(",")}, backlog=${backlog})`,
+    );
+  }
+
+  /** Queue a prefetch-init for a vehicle and set the in-flight guard. */
+  private async enqueueInit(vehicleId: string, source: string): Promise<void> {
+    await this.queue.add(
+      "prefetch-init",
+      { vehicleId, source: source as PrefetchInitJobData["source"] },
+      { removeOnComplete: { count: 1000 }, removeOnFail: { count: 5000 } },
+    );
+    // Guard cleared on completion (incrementCompleted) or by TTL if the run dies.
+    await this.redis.set(`prefetch:scheduled:${vehicleId}`, "1", BACKFILL_SCHEDULED_TTL);
+  }
+
   // ==================== Helpers ====================
 
   private async queueCategoryJob(