fix(backfill): run hourly scan on an in-process timer (cron marker gets buried)

The BullMQ cron scan never fired on prod (ic frozen at 28 for days, 0
scans completed) even after the lifo change. Root cause, confirmed live:
a cron scheduler enqueues a DELAYED marker, and BullMQ promotes delayed
jobs to the wait-list HEAD via LPUSH — `promoteDelayedJobs` ignores lifo —
while the worker pops from the tail. Behind the ~444k deep-drill backlog
the scan marker sat at wait position 1 (head) with ~444k jobs ahead of it
→ unreachable for weeks. So scheduler-level lifo cannot work; the marker
must not go through the queue at all.

Run the scan on an in-process hourly timer in PrefetchWorkerService
instead (kickoff 60s after boot, then hourly), prod-host gated and
single-fired via a Redis NX lock. The scan's Phase-1 still enqueues
NON-delayed `lifo` init jobs, which DO jump the wait list (RPUSH tail →
popped next), so newly-decoded / zero-parts vehicles get onboarded ahead
of the backlog while it drains. jobs.module now removes the old cron
scheduler so stale markers stop being produced; the backfill-scan job
handler stays so any already-queued legacy marker runs harmlessly.

Tested: spec covers the NX lock (run when acquired / skip when held) plus
the existing lifo + Phase-1/Phase-2 gating; full api suite 272 passed,
`pnpm build` + biome green.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

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

@@ -1,7 +1,7 @@
 import { Inject, Module, type OnModuleDestroy, type OnModuleInit } from "@nestjs/common";
 import { Queue } from "bullmq";
 import { CategoriesModule } from "../categories/categories.module";
-import { isCatalogBackfillEnabled, isLifecycleEmailEnabled } from "./prefetch-utils";
+import { isLifecycleEmailEnabled } from "./prefetch-utils";
 import { PrefetchWorkerService } from "./prefetch-worker.service";
 import {
   CATALOG_PREFETCH_QUEUE,
@@ -104,36 +104,16 @@ export class JobsModule implements OnModuleInit, OnModuleDestroy {
       console.log("[jobs] Skipped lifecycle-email cron (not prod host)");
     }
 
-    // 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 (sase.tr) ONLY: dev.sase.tr is a separate DB and must not
-    // sweep/scrape. NODE_ENV is "production" on BOTH, so gate on the prod host.
-    if (isCatalogBackfillEnabled()) {
-      await this.catalogPrefetchQueue.upsertJobScheduler(
-        "catalog-backfill-hourly",
-        { pattern: "0 * * * *" },
-        {
-          name: "backfill-scan",
-          data: {},
-          opts: {
-            removeOnComplete: { count: 48 },
-            removeOnFail: { count: 100 },
-            // lifo so the scan job lands at the tail of the wait list and runs on
-            // the next tick instead of queueing behind the deep-drill backlog —
-            // otherwise the hourly scan is buried for days and never fires. BullMQ
-            // drains wait (RPOPLPUSH from the tail) before the prioritized ZSET, so
-            // lifo — not `priority` — is what jumps an already-deep wait queue.
-            lifo: true,
-          },
-        },
-      );
-      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 (not prod host)");
-    }
+    // Catalog backfill now runs on an IN-PROCESS hourly timer inside
+    // PrefetchWorkerService — NOT a BullMQ cron. A cron marker is a delayed job that
+    // BullMQ promotes to the wait-list HEAD with LPUSH (lifo ignored), and the worker
+    // pops from the tail, so behind a deep prefetch backlog the scan was buried for
+    // days and never fired. Remove any previously-registered cron scheduler here so
+    // stale markers stop being produced (the in-process timer self-gates on the prod
+    // host). The "backfill-scan" job handler is kept so any already-queued legacy
+    // marker still runs harmlessly when reached.
+    await this.catalogPrefetchQueue.removeJobScheduler("catalog-backfill-hourly").catch(() => {});
+    console.log("[jobs] catalog-backfill cron removed (scan runs in-process)");
 
     // Parça Uzmanları sezon kapanışı: her ayın 1'i 00:00 Türkiye saati —
     // biten ayın ilk 3'ünü admin@sase.tr'ye mailler (üyelik günlerini admin

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

@@ -27,6 +27,7 @@ function makeDeps(opts: { waiting: number; limitResults: unknown[][] }) {
     exists: vi.fn(async () => false), // no source cooldown, no in-flight guard
     get: vi.fn(async () => null), // no no-result residue
     set: vi.fn(async () => undefined),
+    setNx: vi.fn(async () => true), // scan lock acquired by default
     getJson: vi.fn(async () => null), // Phase-2 cursor empty
     setJson: vi.fn(async () => undefined),
   };
@@ -54,8 +55,9 @@ describe("PrefetchWorkerService — fast lane (lifo) + backlog gating", () => {
   describe("enqueueInit", () => {
     it("sets lifo + data.fast when fast", async () => {
       const { service, queue } = makeDeps({ waiting: 0, limitResults: [] });
-      await (service as never as { enqueueInit: (a: string, b: string, c: boolean) => Promise<void> })
-        .enqueueInit("v1", "emex", true);
+      await (
+        service as never as { enqueueInit: (a: string, b: string, c: boolean) => Promise<void> }
+      ).enqueueInit("v1", "emex", true);
       const [name, data, jobOpts] = queue.add.mock.calls[0];
       expect(name).toBe("prefetch-init");
       expect(data).toMatchObject({ vehicleId: "v1", source: "emex", fast: true });
@@ -64,8 +66,9 @@ describe("PrefetchWorkerService — fast lane (lifo) + backlog gating", () => {
 
     it("omits lifo when not fast", async () => {
       const { service, queue } = makeDeps({ waiting: 0, limitResults: [] });
-      await (service as never as { enqueueInit: (a: string, b: string, c: boolean) => Promise<void> })
-        .enqueueInit("v1", "emex", false);
+      await (
+        service as never as { enqueueInit: (a: string, b: string, c: boolean) => Promise<void> }
+      ).enqueueInit("v1", "emex", false);
       const [, data, jobOpts] = queue.add.mock.calls[0];
       expect(data).toMatchObject({ fast: false });
       expect((jobOpts as { lifo?: boolean }).lifo).toBeUndefined();
@@ -77,7 +80,14 @@ describe("PrefetchWorkerService — fast lane (lifo) + backlog gating", () => {
       const { service, queue } = makeDeps({ waiting: 0, limitResults: [] });
       await (service as never as { addJob: (n: string, d: unknown) => Promise<void> }).addJob(
         "prefetch-children",
-        { vehicleId: "v1", categoryId: "c1", source: "emex", action: "children", depth: 1, fast: true },
+        {
+          vehicleId: "v1",
+          categoryId: "c1",
+          source: "emex",
+          action: "children",
+          depth: 1,
+          fast: true,
+        },
       );
       const [, , jobOpts] = queue.add.mock.calls[0];
       expect(jobOpts).toMatchObject({ lifo: true });
@@ -102,7 +112,9 @@ describe("PrefetchWorkerService — fast lane (lifo) + backlog gating", () => {
         waiting: 50_000,
         limitResults: [[{ id: "empty1", source: "emex" }]], // Phase-1 query result
       });
-      await (service as never as { processBackfillScan: () => Promise<void> }).processBackfillScan();
+      await (
+        service as never as { processBackfillScan: () => Promise<void> }
+      ).processBackfillScan();
 
       expect(queue.add).toHaveBeenCalledTimes(1);
       const [name, data, jobOpts] = queue.add.mock.calls[0];
@@ -123,7 +135,9 @@ describe("PrefetchWorkerService — fast lane (lifo) + backlog gating", () => {
           [{ id: "stale1", source: "emex", createdAt: new Date("2026-01-01T00:00:00Z") }], // Phase-2
         ],
       });
-      await (service as never as { processBackfillScan: () => Promise<void> }).processBackfillScan();
+      await (
+        service as never as { processBackfillScan: () => Promise<void> }
+      ).processBackfillScan();
 
       expect(queue.add).toHaveBeenCalledTimes(1);
       const [, data, jobOpts] = queue.add.mock.calls[0];
@@ -136,7 +150,28 @@ describe("PrefetchWorkerService — fast lane (lifo) + backlog gating", () => {
     it("skips entirely when the backfill gate is off (not prod host)", async () => {
       process.env.CATALOG_BACKFILL_ENABLED = "false";
       const { service, queue } = makeDeps({ waiting: 0, limitResults: [] });
-      await (service as never as { processBackfillScan: () => Promise<void> }).processBackfillScan();
+      await (
+        service as never as { processBackfillScan: () => Promise<void> }
+      ).processBackfillScan();
+      expect(queue.add).not.toHaveBeenCalled();
+    });
+  });
+
+  describe("runBackfillScan (in-process timer entry, NX-locked)", () => {
+    it("runs the scan when the lock is acquired", async () => {
+      const { service, queue, redis } = makeDeps({
+        waiting: 50_000,
+        limitResults: [[{ id: "empty1", source: "emex" }]],
+      });
+      await (service as never as { runBackfillScan: () => Promise<void> }).runBackfillScan();
+      expect(redis.setNx).toHaveBeenCalledWith("prefetch:backfill:lock", "1", expect.any(Number));
+      expect(queue.add).toHaveBeenCalledTimes(1); // Phase-1 init enqueued
+    });
+
+    it("skips the scan when the lock is already held", async () => {
+      const { service, queue, redis } = makeDeps({ waiting: 50_000, limitResults: [] });
+      redis.setNx.mockResolvedValueOnce(false);
+      await (service as never as { runBackfillScan: () => Promise<void> }).runBackfillScan();
       expect(queue.add).not.toHaveBeenCalled();
     });
   });

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

@@ -45,6 +45,14 @@ 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
@@ -75,6 +83,8 @@ const NORESULT_TTL_S = (Number(process.env.PREFETCH_NORESULT_TTL_DAYS) || 7) * 8
 export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
   private readonly logger = new Logger(PrefetchWorkerService.name);
   private worker: Worker | null = null;
+  private backfillKickoff?: ReturnType<typeof setTimeout>;
+  private backfillInterval?: ReturnType<typeof setInterval>;
 
   constructor(
     @Inject(CATALOG_PREFETCH_QUEUE) private queue: Queue,
@@ -110,15 +120,48 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
     this.logger.log(
       `[prefetch] Worker started (concurrency=${WORKER_CONCURRENCY}, ${WORKER_RATE_MAX} jobs/min, 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.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 };