perf(backfill): raise prefetch worker throughput, env-tunable (Tier 2)

After Tier 1 removed the self-throttle, the 5 jobs/min limiter + concurrency 1
became the bottleneck. Make throughput env-tunable so prod can ramp while
watching the fail rate:
- concurrency 1 -> 3 (PREFETCH_CONCURRENCY): parallelises emex/pl24 so a slow
  parts-catalogs job no longer head-of-line-blocks the queue.
- rate ceiling 5 -> 20 jobs/min (PREFETCH_RATE_MAX).
- parts-catalogs: drop the pathological cumulative index*20s enqueue delay (the
  Nth leaf of a vehicle waited N*20s); keep one per-job pace (PCAT_PACE_MS,
  default 15s, 0 to disable).
- PL24 09-18 scrape window now env-tunable (PREFETCH_PL24_START / _END).

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

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

@@ -77,10 +77,15 @@ function currentIstanbulHour(): number {
  * captures alive 24/7 so backfill can sweep through the night when the
  * upstream is most idle.
  */
+/** PL24 scrape window (Europe/Istanbul hours), env-tunable so it can be widened
+ *  without a redeploy (e.g. PREFETCH_PL24_START=8 PREFETCH_PL24_END=22). */
+const PL24_WINDOW_START = Number(process.env.PREFETCH_PL24_START) || 9;
+const PL24_WINDOW_END = Number(process.env.PREFETCH_PL24_END) || 18;
+
 export function isWithinTimeWindow(source: string): boolean {
   if (source !== "pl24") return true;
   const h = currentIstanbulHour();
-  return h >= 9 && h < 18;
+  return h >= PL24_WINDOW_START && h < PL24_WINDOW_END;
 }
 
 /**
@@ -118,13 +123,12 @@ export function msUntilNext9AM(): number {
   const minute = get("minute");
   const second = get("second");
 
-  // If before 09:00 → wait until 09:00 today
-  // If 18:00+ → wait until 09:00 tomorrow
+  // Before the window opens → wait until it opens today; otherwise → tomorrow.
   let hoursToWait: number;
-  if (hour < 9) {
-    hoursToWait = 9 - hour;
+  if (hour < PL24_WINDOW_START) {
+    hoursToWait = PL24_WINDOW_START - hour;
   } else {
-    hoursToWait = 24 - hour + 9;
+    hoursToWait = 24 - hour + PL24_WINDOW_START;
   }
 
   const ms = hoursToWait * 3600_000 - minute * 60_000 - second * 1000;

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

@@ -44,11 +44,25 @@ const BACKFILL_SOURCES = ["pl24", "emex", "parts-catalogs"];
 /** Redis key holding the rolling rescan cursor (last createdAt seen). */
 const BACKFILL_CURSOR_KEY = "prefetch:backfill:cursor";
 
+// ── 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) || 3;
+/** Global queue rate ceiling (jobs per minute). Was a hard 5/min. */
+const WORKER_RATE_MAX = Number(process.env.PREFETCH_RATE_MAX) || 20;
+/**
+ * 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) || 15_000;
+
 @Injectable()
 export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
   private readonly logger = new Logger(PrefetchWorkerService.name);
   private worker: Worker | null = null;
-  private pcatJobIndex = 0;
 
   constructor(
     @Inject(CATALOG_PREFETCH_QUEUE) private queue: Queue,
@@ -65,8 +79,8 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
       (job, token) => backfillContext.run(true, () => this.process(job, token)),
       {
         connection: getBullConnection(),
-        concurrency: 1,
-        limiter: { max: 5, duration: 60_000 },
+        concurrency: WORKER_CONCURRENCY,
+        limiter: { max: WORKER_RATE_MAX, duration: 60_000 },
       },
     );
 
@@ -80,7 +94,9 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
       this.logger.error(`[prefetch] Worker error: ${err.message}`);
     });
 
-    this.logger.log("[prefetch] Worker started (concurrency=1, 5 jobs/min)");
+    this.logger.log(
+      `[prefetch] Worker started (concurrency=${WORKER_CONCURRENCY}, ${WORKER_RATE_MAX} jobs/min, pcatPace=${PCAT_PACE_MS}ms)`,
+    );
   }
 
   async onModuleDestroy() {
@@ -93,8 +109,8 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
   private async process(job: Job, token?: string): Promise<void> {
     try {
       const data = job.data as { source?: string };
-      if (data.source === "parts-catalogs") {
-        await new Promise((r) => setTimeout(r, 15_000));
+      if (data.source === "parts-catalogs" && PCAT_PACE_MS > 0) {
+        await new Promise((r) => setTimeout(r, PCAT_PACE_MS));
       }
 
       if (job.name === "backfill-scan") {
@@ -141,8 +157,6 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
     const { vehicleId, source } = job.data;
     this.logger.log(`[prefetch] Init for vehicle=${vehicleId}, source=${source}`);
 
-    this.pcatJobIndex = 0;
-
     await checkCooldown(this.redis, source);
     checkTimeWindow(source);
 
@@ -515,9 +529,9 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
       jobId: `prefetch-${data.vehicleId}-${data.categoryId}-${data.action}`,
     };
 
-    if (data.source === "parts-catalogs") {
-      opts.delay = ++this.pcatJobIndex * 20_000;
-    }
+    // 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 this.queue.add(name, data, opts);
   }