fix(api): defer rate-limited prefetch jobs instead of failing them

The worker threw a custom RateLimitError for active-user cooldown and off-hours
windows, but BullMQ treated that as a normal failure: it retried with the
queue's exponential backoff (ignoring retryAfterMs) and dropped the job to
"failed" after attempts:3. So during busy hours sub-jobs were discarded en
masse and the requested defer time (e.g. "wait until 09:00") never applied.

Convert RateLimitError into job.moveToDelayed(now + retryAfterMs, token) +
DelayedError — BullMQ's contract for "not done, not failed, retry later". This
does not consume an attempt and honours the exact delay, so cooldown jobs wait
~60-120s and off-hours jobs wait until the window reopens, then resume. Genuine
errors still fail/retry normally.

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

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

@@ -5,7 +5,7 @@ import {
   type OnModuleDestroy,
   type OnModuleInit,
 } from "@nestjs/common";
-import { type Job, type Queue, Worker } from "bullmq";
+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";
@@ -52,20 +52,20 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
   ) {}
 
   onModuleInit() {
-    this.worker = new Worker(QUEUE_NAMES.CATALOG_PREFETCH, (job) => this.process(job), {
+    this.worker = new Worker(
-      connection: getBullConnection(),
+      QUEUE_NAMES.CATALOG_PREFETCH,
-      concurrency: 1,
+      (job, token) => this.process(job, token),
-      limiter: { max: 5, duration: 60_000 },
+      {
-    });
+        connection: getBullConnection(),
+        concurrency: 1,
+        limiter: { max: 5, duration: 60_000 },
+      },
+    );
 
     this.worker.on("failed", (job, err) => {
-      if (err instanceof RateLimitError) {
+      // RateLimitError is handled in process() via moveToDelayed (a DelayedError,
-        this.logger.debug(
+      // not a failure), so anything reaching here is a genuine failure.
-          `[prefetch] Job ${job?.name} rate-limited, will retry in ${err.retryAfterMs}ms`,
+      this.logger.warn(`[prefetch] Job ${job?.name} failed: ${err.message}`);
-        );
-      } else {
-        this.logger.warn(`[prefetch] Job ${job?.name} failed: ${err.message}`);
-      }
     });
 
     this.worker.on("error", (err) => {
@@ -82,25 +82,38 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
     }
   }
 
-  private async process(job: Job): Promise<void> {
+  private async process(job: Job, token?: string): Promise<void> {
-    const data = job.data as { source?: string };
+    try {
-    if (data.source === "parts-catalogs") {
+      const data = job.data as { source?: string };
-      await new Promise((r) => setTimeout(r, 15_000));
+      if (data.source === "parts-catalogs") {
-    }
+        await new Promise((r) => setTimeout(r, 15_000));
+      }
 
-    if (job.name === "backfill-scan") {
+      if (job.name === "backfill-scan") {
-      return this.processBackfillScan();
+        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) {
+      // Rate-limited (active user cooldown or off-hours window): defer the job to
+      // its requested time WITHOUT consuming a retry attempt. moveToDelayed +
+      // DelayedError is BullMQ's contract for "not done, not failed — try later".
+      if (err instanceof RateLimitError) {
+        const delayMs = Math.max(1000, err.retryAfterMs);
+        await job.moveToDelayed(Date.now() + delayMs, token);
+        this.logger.debug(`[prefetch] Job ${job.name} deferred ${delayMs}ms (rate-limited)`);
+        throw new DelayedError();
+      }
+      throw err;
     }
-    if (job.name === "prefetch-init") {
-      return this.processInit(job as Job<PrefetchInitJobData>);
-    }
-    if (job.name === "prefetch-children") {
-      return this.processChildren(job as Job<PrefetchCategoryJobData>);
-    }
-    if (job.name === "prefetch-parts") {
-      return this.processParts(job as Job<PrefetchCategoryJobData>);
-    }
-    this.logger.warn(`[prefetch] Unknown job name: ${job.name}`);
   }
 
   /**