fix(prefetch): PL24 bütçe/iş-saati kilitlenmesini kır

Prod'da PL24 prefetch tamamen durmuştu: günlük bütçe 600/600 dolu
görünürken gün boyunca SIFIR katalog isteği ve SIFIR yeni kategori
üretiliyordu (ölçüm 2026-09-20).

İki hata birlikte kapalı bir döngü kuruyordu:

1. Günlük bütçe `process()` içinde düşülüyor, iş-saati (ve cooldown)
   kapısı ise her handler'ın başında duruyordu. Pencere dışında uyanan
   bir iş önce bütçeden bir birim yiyor, sonra `time-window` fırlatıp
   hiçbir iş yapmadan erteleniyordu.

2. Bütçesi biten kaynak "bir sonraki UTC gece yarısı"na erteleniyordu.
   PREFETCH_PL24_START=9 ile bu an 03:00 Europe/Istanbul'a denk gelir —
   pencere açılmadan altı saat önce. Uyanan iş yine pencereye takılıyor,
   yine erteleniyor; taze günlük bütçe daha pencere açılmadan bu boş
   uyanmalarla tükeniyordu.

Düzeltme:
- cooldown + iş-saati kapıları `process()` içinde, bütçe düşülmeden
  ÖNCE çalışıyor; handler'lardaki kopyaları kaldırıldı.
- `alignToWindow()` eklendi: bütçe ertelemesi pencerenin içine
  hizalanıyor. Pencere tanımlı değilse (varsayılan 0–24) no-op.
- `currentIstanbulHour` artık `istanbulHourAt`'e deleg ediyor ve h24
  döngüsünün gece yarısı için ürettiği "24" değeri `% 24` ile
  normalleniyor (aksi halde saat hiçbir pencereye düşmez).

8 yeni regresyon testi; api paketi 602 test geçiyor.

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

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

@@ -74,14 +74,21 @@ export async function checkCooldown(redis: RedisService, source: string): Promis
   }
 }
 
-/** Current hour (0–23) in Europe/Istanbul. */
-function currentIstanbulHour(): number {
+/** Europe/Istanbul hour (0–23) at an arbitrary instant. */
+function istanbulHourAt(ms: number): number {
   const hourStr = new Intl.DateTimeFormat("en-US", {
     timeZone: "Europe/Istanbul",
     hour: "numeric",
     hour12: false,
-  }).format(new Date());
-  return Number.parseInt(hourStr, 10);
+  }).format(new Date(ms));
+  // `% 24` because the h24 hour cycle renders midnight as "24", which would put
+  // the hour outside every window and silently park the source forever.
+  return Number.parseInt(hourStr, 10) % 24;
+}
+
+/** Current hour (0–23) in Europe/Istanbul. */
+function currentIstanbulHour(): number {
+  return istanbulHourAt(Date.now());
 }
 
 /**
@@ -115,6 +122,35 @@ export function checkTimeWindow(source: string): void {
   }
 }
 
+/**
+ * The first instant at or after `fromMs` that falls inside `source`'s scrape
+ * window. Returns `fromMs` unchanged when the window is disabled (the default
+ * 0–24) or when `fromMs` is already inside it.
+ *
+ * WHY (plv2.md, finding consumers_jobs-03 — the budget/window deadlock):
+ * `checkSourceDailyBudget` defers a spent source to the next UTC midnight. With
+ * PREFETCH_PL24_START=9 that midnight lands at 03:00 Europe/Istanbul — six hours
+ * BEFORE the window opens. The woken job therefore did no work, threw
+ * `time-window`, and was deferred again to 09:00 — by which point the fresh
+ * daily budget had already been spent by the same stampede of no-op wake-ups.
+ * Measured on prod 2026-09-20: 600/600 pl24 budget consumed, 0 catalog requests
+ * and 0 new categories for the whole day. Landing the deferral inside the window
+ * breaks the cycle.
+ */
+export function alignToWindow(source: string, fromMs: number): number {
+  if (source !== "pl24") return fromMs;
+  if (PL24_WINDOW_START <= 0 && PL24_WINDOW_END >= 24) return fromMs;
+  let t = fromMs;
+  // Step by the hour rather than constructing a local-midnight date: DST-safe and
+  // free of month/year rollover edge cases. 48 steps covers any window shape.
+  for (let i = 0; i < 48; i++) {
+    const h = istanbulHourAt(t);
+    if (h >= PL24_WINDOW_START && h < PL24_WINDOW_END) return t;
+    t += 3_600_000;
+  }
+  return t;
+}
+
 /**
  * Milliseconds until the next 09:00 Europe/Istanbul.
  */

apps/api/src/jobs/prefetch-window-budget.spec.ts

@@ -0,0 +1,283 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+
+/**
+ * Regression lock for the PL24 budget/window DEADLOCK (plv2.md, consumers_jobs-03).
+ *
+ * Two bugs combined to kill PL24 prefetch outright on prod:
+ *   1. `checkSourceDailyBudget` was debited in `process()` while the
+ *      business-hours gate still sat at the top of each handler — so a job that
+ *      woke outside the window paid a budget unit to do nothing.
+ *   2. A spent source was deferred to the next UTC midnight, which is 03:00
+ *      Europe/Istanbul — six hours BEFORE a 09:00 window opens. Every deferred
+ *      job therefore woke early, burned a unit of the fresh daily budget, hit
+ *      the window gate and was deferred again.
+ *
+ * Measured on prod 2026-09-20 before the fix: `prefetch:daily:pl24` at 600/600
+ * with ZERO catalog requests and ZERO new pl24 categories for the whole day.
+ *
+ * The window constants are read at module load, so every case imports the
+ * modules fresh with the env already set.
+ */
+
+const REAL_ENV = { ...process.env };
+
+/** 2026-09-20 00:00 UTC = 03:00 Europe/Istanbul — outside a 09:00–18:00 window. */
+const OUTSIDE_MS = Date.UTC(2026, 8, 20, 0, 0, 0);
+/** 2026-09-20 09:00 UTC = 12:00 Europe/Istanbul — inside it. */
+const INSIDE_MS = Date.UTC(2026, 8, 20, 9, 0, 0);
+
+function istanbulHour(ms: number): number {
+  return (
+    Number.parseInt(
+      new Intl.DateTimeFormat("en-US", {
+        timeZone: "Europe/Istanbul",
+        hour: "numeric",
+        hour12: false,
+      }).format(new Date(ms)),
+      10,
+    ) % 24
+  );
+}
+
+async function loadModules(env: Record<string, string | undefined>) {
+  vi.resetModules();
+  for (const [k, v] of Object.entries(env)) {
+    if (v === undefined) delete process.env[k];
+    else process.env[k] = v;
+  }
+  const utils = await import("./prefetch-utils");
+  const worker = await import("./prefetch-worker.service");
+  return { utils, worker };
+}
+
+/** Minimal deps: only what `process()` touches before dispatching a job. */
+function makeDeps(redisOverrides: Record<string, unknown> = {}) {
+  const incrCalls: string[] = [];
+  const redis = {
+    exists: vi.fn(async (..._a: unknown[]) => false),
+    get: vi.fn(async (..._a: unknown[]): Promise<string | null> => null),
+    set: vi.fn(async (..._a: unknown[]) => undefined),
+    del: vi.fn(async (..._a: unknown[]) => undefined),
+    incr: vi.fn(async (k: string) => {
+      incrCalls.push(k);
+      return 1;
+    }),
+    expire: vi.fn(async (..._a: unknown[]) => undefined),
+    ttl: vi.fn(async (..._a: unknown[]) => -2), // no cooldown key
+    setNx: vi.fn(async (..._a: unknown[]) => true),
+    getJson: vi.fn(async (..._a: unknown[]): Promise<unknown> => null),
+    setJson: vi.fn(async (..._a: unknown[]) => undefined),
+    ...redisOverrides,
+  };
+  const queue = {
+    name: "catalog-prefetch",
+    add: vi.fn(async (..._a: unknown[]) => undefined),
+    getJob: vi.fn(async (..._a: unknown[]): Promise<unknown> => null),
+  };
+  const categoriesService = {
+    getCategoryWithParts: vi.fn(async (..._a: unknown[]) => ({ parts: [] })),
+    getChildren: vi.fn(async (..._a: unknown[]) => []),
+  };
+  return { redis, queue, categoriesService, incrCalls };
+}
+
+function makeJob(name: string, data: Record<string, unknown>) {
+  return {
+    name,
+    data,
+    moveToDelayed: vi.fn(async (..._a: unknown[]) => undefined),
+    attemptsMade: 0,
+  };
+}
+
+const dailyIncrs = (keys: string[]) => keys.filter((k) => k.startsWith("prefetch:daily:pl24"));
+
+describe("alignToWindow", () => {
+  afterEach(() => {
+    process.env = { ...REAL_ENV };
+  });
+
+  it("pushes a pre-window instant into the configured window", async () => {
+    const { utils } = await loadModules({ PREFETCH_PL24_START: "9", PREFETCH_PL24_END: "18" });
+    const aligned = utils.alignToWindow("pl24", OUTSIDE_MS);
+    expect(aligned).toBeGreaterThan(OUTSIDE_MS);
+    const h = istanbulHour(aligned);
+    expect(h).toBeGreaterThanOrEqual(9);
+    expect(h).toBeLessThan(18);
+  });
+
+  it("leaves an in-window instant untouched", async () => {
+    const { utils } = await loadModules({ PREFETCH_PL24_START: "9", PREFETCH_PL24_END: "18" });
+    expect(utils.alignToWindow("pl24", INSIDE_MS)).toBe(INSIDE_MS);
+  });
+
+  it("is a no-op for sources that have no window", async () => {
+    const { utils } = await loadModules({ PREFETCH_PL24_START: "9", PREFETCH_PL24_END: "18" });
+    expect(utils.alignToWindow("emex", OUTSIDE_MS)).toBe(OUTSIDE_MS);
+    expect(utils.alignToWindow("parts-catalogs", OUTSIDE_MS)).toBe(OUTSIDE_MS);
+  });
+
+  it("is a no-op when the window is disabled (the default 0–24)", async () => {
+    const { utils } = await loadModules({
+      PREFETCH_PL24_START: undefined,
+      PREFETCH_PL24_END: undefined,
+    });
+    expect(utils.alignToWindow("pl24", OUTSIDE_MS)).toBe(OUTSIDE_MS);
+  });
+});
+
+describe("process() gate order — window before daily budget", () => {
+  beforeEach(() => {
+    vi.useFakeTimers();
+  });
+  afterEach(() => {
+    vi.useRealTimers();
+    process.env = { ...REAL_ENV };
+  });
+
+  it("does NOT debit the daily budget for a job deferred by the window", async () => {
+    const { worker } = await loadModules({
+      PREFETCH_PL24_START: "9",
+      PREFETCH_PL24_END: "18",
+      PL24_TR_DISABLED: undefined,
+    });
+    vi.setSystemTime(OUTSIDE_MS);
+    const { redis, queue, categoriesService, incrCalls } = makeDeps();
+    const svc = new worker.PrefetchWorkerService(
+      queue as never,
+      queue as never,
+      categoriesService as never,
+      redis as never,
+      { payload: vi.fn(async () => ({})) } as never,
+      {} as never,
+    );
+    const job = makeJob("prefetch-parts", {
+      vehicleId: "v1",
+      categoryId: "c1",
+      source: "pl24",
+      fast: true,
+    });
+
+    await expect(
+      (svc as never as { process: (j: unknown, t?: string) => Promise<void> }).process(job, "tok"),
+    ).rejects.toThrow();
+
+    expect(job.moveToDelayed).toHaveBeenCalled();
+    expect(dailyIncrs(incrCalls)).toHaveLength(0);
+    // …and the handler never ran, so nothing was fetched upstream either.
+    expect(categoriesService.getCategoryWithParts).not.toHaveBeenCalled();
+  });
+
+  it("defers the window-blocked job to an instant inside the window", async () => {
+    const { worker } = await loadModules({
+      PREFETCH_PL24_START: "9",
+      PREFETCH_PL24_END: "18",
+      PL24_TR_DISABLED: undefined,
+    });
+    vi.setSystemTime(OUTSIDE_MS);
+    const { redis, queue, categoriesService } = makeDeps();
+    const svc = new worker.PrefetchWorkerService(
+      queue as never,
+      queue as never,
+      categoriesService as never,
+      redis as never,
+      { payload: vi.fn(async () => ({})) } as never,
+      {} as never,
+    );
+    const job = makeJob("prefetch-parts", {
+      vehicleId: "v1",
+      categoryId: "c1",
+      source: "pl24",
+      fast: true,
+    });
+    await expect(
+      (svc as never as { process: (j: unknown, t?: string) => Promise<void> }).process(job, "tok"),
+    ).rejects.toThrow();
+
+    const [when] = job.moveToDelayed.mock.calls[0] as [number];
+    const h = istanbulHour(when);
+    expect(h).toBeGreaterThanOrEqual(9);
+    expect(h).toBeLessThan(18);
+  });
+
+  it("debits the daily budget once the window is open", async () => {
+    const { worker } = await loadModules({
+      PREFETCH_PL24_START: "9",
+      PREFETCH_PL24_END: "18",
+      PL24_TR_DISABLED: undefined,
+    });
+    vi.setSystemTime(INSIDE_MS);
+    const { redis, queue, categoriesService, incrCalls } = makeDeps();
+    const svc = new worker.PrefetchWorkerService(
+      queue as never,
+      queue as never,
+      categoriesService as never,
+      redis as never,
+      { payload: vi.fn(async () => ({})) } as never,
+      {} as never,
+    );
+    const job = makeJob("prefetch-parts", {
+      vehicleId: "v1",
+      categoryId: "c1",
+      source: "pl24",
+      fast: true,
+    });
+
+    await (svc as never as { process: (j: unknown, t?: string) => Promise<void> }).process(
+      job,
+      "tok",
+    );
+
+    expect(dailyIncrs(incrCalls)).toHaveLength(1);
+    expect(categoriesService.getCategoryWithParts).toHaveBeenCalledWith("c1");
+  });
+});
+
+describe("checkSourceDailyBudget — spent source retries inside the window", () => {
+  beforeEach(() => {
+    vi.useFakeTimers();
+  });
+  afterEach(() => {
+    vi.useRealTimers();
+    process.env = { ...REAL_ENV };
+  });
+
+  it("never parks a spent source at 03:00 Istanbul again", async () => {
+    const { worker } = await loadModules({
+      PREFETCH_PL24_START: "9",
+      PREFETCH_PL24_END: "18",
+      PREFETCH_DAILY_PL24: "600",
+    });
+    vi.setSystemTime(INSIDE_MS);
+    // Counter already at the fast-lane ceiling → the next job must be deferred.
+    const { redis, queue, categoriesService } = makeDeps({
+      get: vi.fn(async (k: string) => (String(k).startsWith("prefetch:daily:pl24") ? "600" : null)),
+    });
+    const svc = new worker.PrefetchWorkerService(
+      queue as never,
+      queue as never,
+      categoriesService as never,
+      redis as never,
+      { payload: vi.fn(async () => ({})) } as never,
+      {} as never,
+    );
+    const job = makeJob("prefetch-parts", {
+      vehicleId: "v1",
+      categoryId: "c1",
+      source: "pl24",
+      fast: true,
+    });
+
+    await expect(
+      (svc as never as { process: (j: unknown, t?: string) => Promise<void> }).process(job, "tok"),
+    ).rejects.toThrow();
+
+    const [when] = job.moveToDelayed.mock.calls[0] as [number];
+    // Past the UTC rollover…
+    expect(when).toBeGreaterThan(Date.UTC(2026, 8, 21, 0, 0, 0));
+    // …and inside the window, not at 03:00 Istanbul like the old deferral.
+    const h = istanbulHour(when);
+    expect(h).toBeGreaterThanOrEqual(9);
+    expect(h).toBeLessThan(18);
+  });
+});

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

@@ -17,6 +17,7 @@ import { QUEUE_NAMES, getBullConnection } from "./bull.config";
 import { backfillContext } from "./prefetch-context";
 import {
   RateLimitError,
+  alignToWindow,
   checkCooldown,
   checkTimeWindow,
   initProgress,
@@ -355,6 +356,17 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
       ) {
         const lane = (job.data as { fast?: boolean }).fast ? "fast" : "main";
         await this.checkSourceRate(data.source, lane);
+        // Cooldown + business-hours window BEFORE the daily charge. These two
+        // gates used to live at the top of each handler, i.e. AFTER the budget
+        // was already debited, so every 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 formed a closed 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. Order matters here.
+        await checkCooldown(this.redis, data.source);
+        checkTimeWindow(data.source);
         // Daily budget AFTER the per-minute gate: a job deferred on the minute
         // ceiling above never reaches here, so rate-limited retries don't inflate
         // the daily counter — only jobs about to do real work are counted. The
@@ -421,8 +433,8 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
     const { vehicleId, source, fast = false } = job.data;
     this.logger.log(`[prefetch] Init for vehicle=${vehicleId}, source=${source}`);
 
-    await checkCooldown(this.redis, source);
-    checkTimeWindow(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))) {
@@ -563,8 +575,8 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
     const { vehicleId, categoryId, source, depth, fast = false } = job.data;
     this.logger.log(`[prefetch] Children for category=${categoryId}, depth=${depth}`);
 
-    await checkCooldown(this.redis, source);
-    checkTimeWindow(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.
 
     const depthCeiling = maxDepthFor(source, fast);
     if (depth >= depthCeiling) {
@@ -631,8 +643,8 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
     const { vehicleId, categoryId, source } = job.data;
     this.logger.log(`[prefetch] Parts for category=${categoryId}`);
 
-    await checkCooldown(this.redis, source);
-    checkTimeWindow(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.
 
     try {
       await this.categoriesService.getCategoryWithParts(categoryId);
@@ -1214,11 +1226,17 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
       // 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 msLeft = dayMs - (now % dayMs) + 1000 + Math.floor(Math.random() * 45 * 60_000);
+      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 until the window rolls`,
+            `deferring ~${Math.round(msLeft / 3_600_000)}h to the next in-window slot`,
         );
       }
       throw new RateLimitError(msLeft, "source-rate");