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/database/schema/core.ts

@@ -745,12 +745,12 @@ export const oemSuggestions = pgTable(
 
 // ─── OEM Expert Rewards (monthly leaderboard prizes) ──
 // One row per (TR-month, rank): the expert-rewards cron (1st of month 00:00
-// Europe/Istanbul) closes the finished season and grants the top 3 voters a
-// subscription extension (30/15/7 days — EXPERT_REWARD_LADDER). The unique
+// Europe/Istanbul) closes the finished season, e-mails the top 3 voters to
+// admin@sase.tr (name/e-mail/points + prize 30/15/7 days — EXPERT_REWARD_
+// LADDER) and records them here. Subscription days are NOT granted
+// automatically — the admin applies them manually from the mail. The unique
 // index doubles as the run-once guard: a second run for the same period
-// inserts nothing, so days are never granted twice. Granting mirrors the
-// referral mechanic: extend a live active/trial sub, else bank the days in
-// users.referral_credit_days (consumed at next trial/activation).
+// inserts nothing and sends no second mail.
 export const oemExpertRewards = pgTable(
   "oem_expert_rewards",
   {

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,35 +104,22 @@ 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 },
-          },
-        },
-      );
-      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 oylayıcısına üyelik uzatması (30/15/7 gün). Ödül
-    // yalnız DB'ye yazar (dış yan etki yok), o yüzden dev'de de çalışır;
-    // (period, rank) unique index'i çift vermeyi zaten engeller.
+    // biten ayın ilk 3'ünü admin@sase.tr'ye mailler (üyelik günlerini admin
+    // elle tanımlar) + denetim satırı yazar. Dev'de de çalışır; dev maili
+    // "[dev]" konu önekiyle ayrışır, (period, rank) unique index'i aynı
+    // dönem için ikinci maili/satırı engeller.
     await this.expertRewardsQueue.upsertJobScheduler(
       "expert-rewards-monthly",
       { pattern: "0 0 1 * *", tz: "Europe/Istanbul" },

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

@@ -0,0 +1,178 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { PrefetchWorkerService } from "./prefetch-worker.service";
+
+/**
+ * Chainable drizzle mock: select/from/where/orderBy return the chain; the
+ * terminal limit() yields the next queued result array. Mirrors the trick the
+ * other api specs use (await on a plain array resolves to the array itself).
+ */
+function makeDb(limitResults: unknown[][]) {
+  const queued = [...limitResults];
+  const chain: Record<string, unknown> = {};
+  for (const m of ["select", "from", "where", "orderBy"]) {
+    chain[m] = vi.fn(() => chain);
+  }
+  chain.limit = vi.fn(() => queued.shift() ?? []);
+  return chain;
+}
+
+function makeDeps(opts: { waiting: number; limitResults: unknown[][] }) {
+  const queue = {
+    // typed args so `.mock.calls[i]` is `unknown[]` (not a 0-length tuple) — the
+    // production `nest build` compiles spec files and rejects tuple-index access.
+    add: vi.fn((..._args: unknown[]) => Promise.resolve(undefined)),
+    getJobCounts: vi.fn(async () => ({ waiting: opts.waiting, delayed: 0, active: 0 })),
+  };
+  const redis = {
+    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),
+  };
+  const posthog = { payload: vi.fn(async () => ({})) }; // compiled-in defaults
+  const db = makeDb(opts.limitResults);
+  const service = new PrefetchWorkerService(
+    queue as never,
+    {} as never, // categoriesService — unused by the scan
+    redis as never,
+    posthog as never,
+    db as never,
+  );
+  return { service, queue, redis, posthog, db };
+}
+
+describe("PrefetchWorkerService — fast lane (lifo) + backlog gating", () => {
+  beforeEach(() => {
+    vi.clearAllMocks();
+    process.env.CATALOG_BACKFILL_ENABLED = "true"; // force the prod-host gate on
+  });
+  afterEach(() => {
+    process.env.CATALOG_BACKFILL_ENABLED = undefined;
+  });
+
+  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);
+      const [name, data, jobOpts] = queue.add.mock.calls[0];
+      expect(name).toBe("prefetch-init");
+      expect(data).toMatchObject({ vehicleId: "v1", source: "emex", fast: true });
+      expect(jobOpts).toMatchObject({ lifo: true });
+    });
+
+    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);
+      const [, data, jobOpts] = queue.add.mock.calls[0];
+      expect(data).toMatchObject({ fast: false });
+      expect((jobOpts as { lifo?: boolean }).lifo).toBeUndefined();
+    });
+  });
+
+  describe("addJob (child jobs)", () => {
+    it("sets lifo when data.fast is true", async () => {
+      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,
+        },
+      );
+      const [, , jobOpts] = queue.add.mock.calls[0];
+      expect(jobOpts).toMatchObject({ lifo: true });
+    });
+
+    it("omits lifo when data.fast is falsy", async () => {
+      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 },
+      );
+      const [, , jobOpts] = queue.add.mock.calls[0];
+      expect((jobOpts as { lifo?: boolean }).lifo).toBeUndefined();
+    });
+  });
+
+  describe("processBackfillScan", () => {
+    it("runs Phase-1 in the fast lane even when the backlog is far over the ceiling", async () => {
+      // waiting 50k ≫ default maxBacklog 1000 → Phase-2 must be suspended, but a
+      // zero-parts vehicle must still be onboarded with lifo.
+      const { service, queue, redis } = makeDeps({
+        waiting: 50_000,
+        limitResults: [[{ id: "empty1", source: "emex" }]], // Phase-1 query result
+      });
+      await (
+        service as never as { processBackfillScan: () => Promise<void> }
+      ).processBackfillScan();
+
+      expect(queue.add).toHaveBeenCalledTimes(1);
+      const [name, data, jobOpts] = queue.add.mock.calls[0];
+      expect(name).toBe("prefetch-init");
+      expect(data).toMatchObject({ vehicleId: "empty1", fast: true });
+      expect(jobOpts).toMatchObject({ lifo: true });
+      // Phase-2 suspended → the rolling cursor is never read or advanced.
+      expect(redis.getJson).not.toHaveBeenCalled();
+      expect(redis.setJson).not.toHaveBeenCalled();
+    });
+
+    it("runs Phase-2 (normal lane, no lifo) when the backlog is under the ceiling", async () => {
+      // waiting 10 < 1000 → Phase-1 empty, Phase-2 picks a vehicle with fast=false.
+      const { service, queue, redis } = makeDeps({
+        waiting: 10,
+        limitResults: [
+          [], // Phase-1: no zero-parts vehicles
+          [{ id: "stale1", source: "emex", createdAt: new Date("2026-01-01T00:00:00Z") }], // Phase-2
+        ],
+      });
+      await (
+        service as never as { processBackfillScan: () => Promise<void> }
+      ).processBackfillScan();
+
+      expect(queue.add).toHaveBeenCalledTimes(1);
+      const [, data, jobOpts] = queue.add.mock.calls[0];
+      expect(data).toMatchObject({ vehicleId: "stale1", fast: false });
+      expect((jobOpts as { lifo?: boolean }).lifo).toBeUndefined();
+      // Phase-2 ran → cursor advanced.
+      expect(redis.setJson).toHaveBeenCalled();
+    });
+
+    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();
+      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 };
@@ -167,7 +210,7 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
    * Init job: walk the category tree for a vehicle and queue sub-jobs.
    */
   private async processInit(job: Job<PrefetchInitJobData>): Promise<void> {
-    const { vehicleId, source } = job.data;
+    const { vehicleId, source, fast = false } = job.data;
     this.logger.log(`[prefetch] Init for vehicle=${vehicleId}, source=${source}`);
 
     await checkCooldown(this.redis, source);
@@ -244,7 +287,7 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
 
         for (const child of children) {
           if (child.unavailable) continue;
-          await this.queueCategoryJob(child, vehicleId, source, 1);
+          await this.queueCategoryJob(child, vehicleId, source, 1, fast);
           queued++;
         }
       } else if (this.isLeafLinkPath(cat.linkPath, cat.source, cat.hasSubgroups)) {
@@ -262,6 +305,7 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
             source,
             action: "parts" as const,
             depth: 0,
+            fast,
           });
           queued++;
         }
@@ -273,6 +317,7 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
           source,
           action: "children" as const,
           depth: 0,
+          fast,
         });
         queued++;
       }
@@ -286,7 +331,7 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
    * Fetch children (sub-categories) for a category.
    */
   private async processChildren(job: Job<PrefetchCategoryJobData>): Promise<void> {
-    const { vehicleId, categoryId, source, depth } = job.data;
+    const { vehicleId, categoryId, source, depth, fast = false } = job.data;
     this.logger.log(`[prefetch] Children for category=${categoryId}, depth=${depth}`);
 
     await checkCooldown(this.redis, source);
@@ -303,7 +348,7 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
       let queued = 0;
       for (const child of children) {
         if (child.unavailable) continue;
-        await this.queueCategoryJob(child, vehicleId, source, depth + 1);
+        await this.queueCategoryJob(child, vehicleId, source, depth + 1, fast);
         queued++;
       }
 
@@ -378,12 +423,16 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
         ? cfg.maxBacklog
         : BACKFILL_MAX_BACKLOG;
 
-    // Self-throttle: don't pile on if the queue is already deep — let it drain.
+    // Self-throttle: when the queue is already deep, suspend only the Phase-2
+    // rolling rescan (the part that piles on). Phase-1 still runs every wave so
+    // genuinely-empty vehicles keep getting onboarded through the fast lane even
+    // while a large deep-drill backlog is still draining — otherwise a single
+    // backlog spike freezes new-vehicle coverage until the whole queue clears.
     const counts = await this.queue.getJobCounts("waiting", "delayed", "active");
     const backlog = (counts.waiting ?? 0) + (counts.delayed ?? 0) + (counts.active ?? 0);
-    if (backlog > maxBacklog) {
-      this.logger.log(`[backfill] Skip — queue backlog ${backlog} > ${maxBacklog}`);
-      return;
+    const phase2Allowed = backlog <= maxBacklog;
+    if (!phase2Allowed) {
+      this.logger.log(`[backfill] Backlog ${backlog} > ${maxBacklog} — Phase-1 (fast lane) only`);
     }
 
     // Only target sources eligible right now: not in cooldown (user active) and
@@ -399,11 +448,14 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
       return;
     }
 
-    const picked: Array<{ id: string; source: string }> = [];
+    const picked: Array<{ id: string; source: string; fast: boolean }> = [];
     const seen = new Set<string>();
     const overfetch = batchSize * 4; // headroom for in-flight skips
 
-    const tryPick = async (v: { id: string; source: string | null }): Promise<void> => {
+    const tryPick = async (
+      v: { id: string; source: string | null },
+      fast: boolean,
+    ): Promise<void> => {
       if (picked.length >= batchSize || seen.has(v.id) || !v.source) return;
       if (await this.redis.exists(`prefetch:scheduled:${v.id}`)) return; // already in flight
       // Skip exhausted residue: vehicles whose prefetch keeps finishing with zero
@@ -411,7 +463,7 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
       const noResult = await this.redis.get(this.noResultKey(v.id));
       if (noResult && Number(noResult) >= NORESULT_MAX_ATTEMPTS) return;
       seen.add(v.id);
-      picked.push({ id: v.id, source: v.source });
+      picked.push({ id: v.id, source: v.source, fast });
     };
 
     // Phase 1 — clear the obvious backlog first: decoded vehicles with zero parts.
@@ -429,11 +481,12 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
       .orderBy(asc(vehicles.createdAt))
       .limit(overfetch);
 
-    for (const v of noParts) await tryPick(v);
+    for (const v of noParts) await tryPick(v, true);
 
     // 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 < batchSize) {
+    // ones. A createdAt cursor walks forward and wraps around at the end. Gated by
+    // the backlog ceiling (above) so it doesn't pile on while the queue is deep.
+    if (phase2Allowed && picked.length < batchSize) {
       const cursorObj = await this.redis.getJson<{ ts: string }>(BACKFILL_CURSOR_KEY);
       const cursor = cursorObj?.ts ? new Date(cursorObj.ts) : new Date(0);
 
@@ -456,7 +509,7 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
       let lastTs: Date | null = null;
       for (const v of rolling) {
         lastTs = v.createdAt;
-        await tryPick(v);
+        await tryPick(v, false);
       }
       if (lastTs) {
         await this.redis.setJson(BACKFILL_CURSOR_KEY, { ts: lastTs.toISOString() }, 30 * 86400);
@@ -468,18 +521,26 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
       return;
     }
 
-    for (const v of picked) await this.enqueueInit(v.id, v.source);
+    for (const v of picked) await this.enqueueInit(v.id, v.source, v.fast);
+    const fastCount = picked.filter((v) => v.fast).length;
     this.logger.log(
-      `[backfill] Queued ${picked.length} vehicle(s) (sources=${eligible.join(",")}, backlog=${backlog})`,
+      `[backfill] Queued ${picked.length} vehicle(s) (${fastCount} fast-lane, ` +
+        `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> {
+  private async enqueueInit(vehicleId: string, source: string, fast = false): Promise<void> {
     await this.queue.add(
       "prefetch-init",
-      { vehicleId, source: source as PrefetchInitJobData["source"] },
-      { removeOnComplete: { count: 1000 }, removeOnFail: { count: 5000 } },
+      { vehicleId, source: source as PrefetchInitJobData["source"], fast },
+      {
+        removeOnComplete: { count: 1000 },
+        removeOnFail: { count: 5000 },
+        // Fast lane (Phase-1): lifo so a zero-parts vehicle's whole chain jumps
+        // the deep-drill backlog instead of queueing behind it (see addJob).
+        ...(fast ? { lifo: true } : {}),
+      },
     );
     // Guard cleared on completion (incrementCompleted) or by TTL if the run dies.
     await this.redis.set(`prefetch:scheduled:${vehicleId}`, "1", BACKFILL_SCHEDULED_TTL);
@@ -498,6 +559,7 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
     vehicleId: string,
     source: string,
     depth: number,
+    fast = false,
   ): Promise<void> {
     if (cat.unavailable) return;
 
@@ -516,6 +578,7 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
           source: source as "pl24" | "emex",
           action: "parts" as const,
           depth,
+          fast,
         });
       }
     } else if (cat.linkPath) {
@@ -535,7 +598,7 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
 
         for (const child of children) {
           if (child.unavailable) continue;
-          await this.queueCategoryJob(child, vehicleId, source, depth + 1);
+          await this.queueCategoryJob(child, vehicleId, source, depth + 1, fast);
         }
       } else {
         await this.addJob("prefetch-children", {
@@ -544,6 +607,7 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
           source: source as "pl24" | "emex",
           action: "children" as const,
           depth,
+          fast,
         });
       }
     }
@@ -583,6 +647,12 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
       // "-" instead. The values are UUIDs — the ID only needs to be deterministic
       // (for dedup), not parseable.
       jobId: `prefetch-${data.vehicleId}-${data.categoryId}-${data.action}`,
+      // Fast lane (Phase-1 / reactive): add with `lifo` so the job RPUSHes to the
+      // TAIL of the wait list, where BullMQ's RPOPLPUSH picks it next — i.e. ahead
+      // of the deep deep-drill backlog already sitting in wait. (BullMQ 5 drains
+      // the wait list before the prioritized ZSET, so `priority` would do the
+      // OPPOSITE here and starve the job behind the backlog; lifo is correct.)
+      ...(data.fast ? { lifo: true } : {}),
     };
 
     // parts-catalogs pacing is handled per-job in process() (PCAT_PACE_MS) + the

apps/api/src/jobs/prefetch.types.ts

@@ -4,6 +4,13 @@ export type PrefetchSource = "pl24" | "emex" | "parts-catalogs";
 export interface PrefetchInitJobData {
   vehicleId: string;
   source: PrefetchSource;
+  /**
+   * Fast lane: enqueue this job (and its whole sub-job chain) with BullMQ `lifo`
+   * so it lands at the tail of the wait list and is picked before the deep
+   * deep-drill backlog. Set for Phase-1 (zero-parts) backfill so newly-decoded
+   * vehicles aren't starved behind the rolling rescan. See PrefetchWorkerService.
+   */
+  fast?: boolean;
 }
 
 /** Per-category job: fetches children OR parts */
@@ -13,4 +20,6 @@ export interface PrefetchCategoryJobData {
   source: PrefetchSource;
   action: "children" | "parts";
   depth: number;
+  /** Inherited from the init job — keeps the whole chain in the fast lane. */
+  fast?: boolean;
 }

apps/api/src/jobs/processors/expert-rewards.processor.ts

@@ -1,24 +1,34 @@
 import { Job } from "bullmq";
-import { and, asc, desc, eq, gte, lt, or, sql } from "drizzle-orm";
+import { and, asc, desc, eq, gte, lt, sql } from "drizzle-orm";
 import { PostgresJsDatabase } from "drizzle-orm/postgres-js";
-import {
-  oemExpertRewards,
-  oemVotePoints,
-  userSubscriptions,
-  users,
-} from "../../database/schema/core";
+import { oemExpertRewards, oemVotePoints, users } from "../../database/schema/core";
 import { EXPERT_REWARD_LADDER, previousTrMonthWindow } from "../../oem-votes/expert-period";
 
 type Database = PostgresJsDatabase<Record<string, unknown>>;
 
+const ADMIN_EMAIL = "admin@sase.tr";
+
+interface SeasonWinner {
+  rank: number;
+  days: number;
+  userId: string;
+  name: string;
+  email: string;
+  points: number;
+}
+
 // Parça Uzmanları sezon kapanışı: her ayın 1'i 00:00 TR'de biten ayın ilk 3
-// oylayıcısına üyelik uzatması (30/15/7 gün). oem_expert_rewards'taki
-// (period, rank) unique index run-once garantisidir — yeniden çalıştırma
-// (retry, elle tetik) hiçbir şeyi ikinci kez vermez.
+// oylayıcısını hesaplar, kazananları admin@sase.tr'ye MAİLLER ve
+// oem_expert_rewards'a denetim satırı yazar. Üyelik uzatması OTOMATİK
+// YAPILMAZ — günleri (30/15/7) admin maildeki listeye göre elle tanımlar.
+// (period, rank) unique index run-once garantisidir. Sıra bilinçli: önce
+// mail, sonra satırlar — satırlar önce yazılsaydı mail hatasındaki retry
+// alreadyGranted'a takılır, mail hiç gitmezdi (nadir çift mail, kayıp
+// mailden iyidir).
 export async function processExpertRewards(
   job: Job,
   db: Database,
-): Promise<{ period: string; granted: number; skipped: boolean }> {
+): Promise<{ period: string; winners: number; skipped: boolean }> {
   const { start, end } = previousTrMonthWindow(new Date());
   const periodLabel = start.toISOString();
   console.log(`[expert-rewards] Processing job ${job.id} for period ${periodLabel}`);
@@ -30,85 +40,119 @@ export async function processExpertRewards(
     .limit(1);
   if (alreadyGranted) {
     console.log(`[expert-rewards] Period ${periodLabel} already granted — skipping`);
-    return { period: periodLabel, granted: 0, skipped: true };
+    return { period: periodLabel, winners: 0, skipped: true };
   }
 
   // Biten sezonun ilk 3'ü — liderlik tablosuyla aynı sıralama: puan desc,
   // eşitlikte puana daha erken ulaşan önde.
   const totalPoints = sql<number>`sum(${oemVotePoints.points})::int`;
   const top = await db
-    .select({ userId: oemVotePoints.userId, points: totalPoints })
+    .select({
+      userId: oemVotePoints.userId,
+      name: users.name,
+      email: users.email,
+      points: totalPoints,
+    })
     .from(oemVotePoints)
+    .innerJoin(users, eq(users.id, oemVotePoints.userId))
     .where(and(gte(oemVotePoints.createdAt, start), lt(oemVotePoints.createdAt, end)))
-    .groupBy(oemVotePoints.userId)
+    .groupBy(oemVotePoints.userId, users.name, users.email)
     .orderBy(desc(totalPoints), asc(sql`min(${oemVotePoints.createdAt})`))
     .limit(EXPERT_REWARD_LADDER.length);
 
   if (top.length === 0) {
-    console.log(`[expert-rewards] No votes in period ${periodLabel} — nothing to grant`);
-    return { period: periodLabel, granted: 0, skipped: false };
+    console.log(`[expert-rewards] No votes in period ${periodLabel} — nothing to report`);
+    return { period: periodLabel, winners: 0, skipped: false };
   }
 
-  let granted = 0;
-  for (let i = 0; i < top.length; i++) {
-    const winner = top[i];
-    const { rank, days } = EXPERT_REWARD_LADDER[i];
+  const winners: SeasonWinner[] = top.map((row, i) => ({
+    rank: EXPERT_REWARD_LADDER[i].rank,
+    days: EXPERT_REWARD_LADDER[i].days,
+    userId: row.userId,
+    name: row.name,
+    email: row.email,
+    points: row.points,
+  }));
 
-    await db.transaction(async (tx) => {
-      const inserted = await tx
-        .insert(oemExpertRewards)
-        .values({
-          periodStart: start,
-          userId: winner.userId,
-          rank,
-          points: winner.points,
-          rewardDays: days,
-        })
-        .onConflictDoNothing()
-        .returning({ id: oemExpertRewards.id });
-      // Yarış/yeniden-deneme: kayıt zaten varsa gün de verilmiş demektir.
-      if (inserted.length === 0) return;
+  await sendWinnersMail(start, winners);
 
-      // Referral ödül mekaniğinin birebir kopyası (worker Nest DI'sız çalıştığı
-      // için ReferralsService.grantRewardDays buradan çağrılamıyor): canlı
-      // active/trial aboneliği uzat, yoksa günleri krediye banka et — kredi bir
-      // sonraki trial/aktivasyonda otomatik tüketilir.
-      const [sub] = await tx
-        .select({ id: userSubscriptions.id, endDate: userSubscriptions.endDate })
-        .from(userSubscriptions)
-        .where(
-          and(
-            eq(userSubscriptions.userId, winner.userId),
-            or(eq(userSubscriptions.status, "active"), eq(userSubscriptions.status, "trial")),
-          ),
-        )
-        .orderBy(desc(userSubscriptions.endDate))
-        .limit(1);
-
-      if (sub?.endDate) {
-        const newEnd = new Date(sub.endDate);
-        newEnd.setDate(newEnd.getDate() + days);
-        await tx
-          .update(userSubscriptions)
-          .set({ endDate: newEnd, updatedAt: new Date() })
-          .where(eq(userSubscriptions.id, sub.id));
-      } else {
-        await tx
-          .update(users)
-          .set({
-            referralCreditDays: sql`${users.referralCreditDays} + ${days}`,
-            updatedAt: new Date(),
-          })
-          .where(eq(users.id, winner.userId));
-      }
-
-      granted++;
-      console.log(
-        `[expert-rewards] rank=${rank} user=${winner.userId} points=${winner.points} +${days}d`,
-      );
-    });
+  for (const winner of winners) {
+    await db
+      .insert(oemExpertRewards)
+      .values({
+        periodStart: start,
+        userId: winner.userId,
+        rank: winner.rank,
+        points: winner.points,
+        rewardDays: winner.days,
+      })
+      .onConflictDoNothing();
   }
 
-  console.log(`[expert-rewards] Period ${periodLabel}: granted ${granted} reward(s)`);
-  return { period: periodLabel, granted, skipped: false };
+  console.log(
+    `[expert-rewards] Period ${periodLabel}: mailed ${winners.length} winner(s) to ${ADMIN_EMAIL}`,
+  );
+  return { period: periodLabel, winners: winners.length, skipped: false };
+}
+
+const escapeHtml = (value: string) =>
+  value.replace(/&/g, "&amp;").replace(/</g, "&lt;").replace(/>/g, "&gt;");
+
+// Worker Nest DI'sız çalıştığı için EmailService kullanılamıyor — aynı Postal
+// HTTP çağrısı (email.service.ts ile aynı payload) env üzerinden yapılır.
+async function sendWinnersMail(periodStart: Date, winners: SeasonWinner[]): Promise<void> {
+  const apiUrl = process.env.POSTAL_API_URL;
+  const apiKey = process.env.POSTAL_API_KEY;
+  if (!apiUrl || !apiKey) {
+    throw new Error("POSTAL_API_URL/POSTAL_API_KEY yok — kazanan maili gönderilemiyor");
+  }
+  const fromAddress = process.env.POSTAL_FROM_ADDRESS ?? "noreply@sase.tr";
+  const fromName = process.env.POSTAL_FROM_NAME ?? "Sase.tr";
+
+  const monthLabel = new Intl.DateTimeFormat("tr-TR", {
+    month: "long",
+    year: "numeric",
+    timeZone: "Europe/Istanbul",
+  }).format(periodStart);
+  // dev DB'den atılan deneme mailleri prod kapanışlarıyla karışmasın.
+  const isDev = (process.env.DATABASE_URL ?? "").includes("sase_dev");
+  const subject = `${isDev ? "[dev] " : ""}Parça Uzmanları ${monthLabel} kazananları — üyelik günlerini tanımlayın`;
+
+  const rows = winners
+    .map(
+      (w) =>
+        `<tr><td style="padding:6px 10px">${w.rank}.</td><td style="padding:6px 10px">${escapeHtml(w.name)}</td><td style="padding:6px 10px">${escapeHtml(w.email)}</td><td style="padding:6px 10px;text-align:right">${w.points}</td><td style="padding:6px 10px"><b>${w.days} gün</b></td><td style="padding:6px 10px;font-family:monospace;font-size:12px">${w.userId}</td></tr>`,
+    )
+    .join("");
+  const html = `
+    <h2>Parça Uzmanları — ${monthLabel} sezonu kapandı</h2>
+    <p>Kazananlara üyelik uzatmasını panelden tanımlayın:</p>
+    <table border="1" cellspacing="0" style="border-collapse:collapse;border-color:#ddd">
+      <tr><th style="padding:6px 10px">Sıra</th><th style="padding:6px 10px">Ad</th><th style="padding:6px 10px">E-posta</th><th style="padding:6px 10px">Puan</th><th style="padding:6px 10px">Ödül</th><th style="padding:6px 10px">Kullanıcı ID</th></tr>
+      ${rows}
+    </table>
+    <p style="color:#888;font-size:12px">Bu mail expert-rewards cron'undan otomatik gönderildi; üyelik günleri otomatik tanımlanmaz.</p>`;
+  const text = [
+    `Parça Uzmanları ${monthLabel} kazananları:`,
+    ...winners.map(
+      (w) => `${w.rank}. ${w.name} <${w.email}> — ${w.points} puan → ${w.days} gün (${w.userId})`,
+    ),
+  ].join("\n");
+
+  const response = await fetch(`${apiUrl}/api/v1/send/message`, {
+    method: "POST",
+    headers: { "Content-Type": "application/json", "X-Server-API-Key": apiKey },
+    body: JSON.stringify({
+      to: [ADMIN_EMAIL],
+      from: `${fromName} <${fromAddress}>`,
+      subject,
+      html_body: html,
+      plain_body: text,
+      tag: "expert-rewards",
+    }),
+  });
+  const result = (await response.json()) as { status?: string };
+  if (result.status !== "success") {
+    throw new Error(`Postal kazanan maili gönderemedi: ${JSON.stringify(result)}`);
+  }
 }

apps/web/src/routes/dashboard/uzmanlar.tsx

@@ -191,8 +191,7 @@ function ExpertsPage() {
         Oy ver <span className="font-semibold text-muted-foreground">+1</span> · çoğunluğu tuttur{" "}
         <span className="font-semibold text-muted-foreground">+2 bonus</span> · kodu ilk
         değerlendiren <span className="font-semibold text-muted-foreground">3 puanı</span> kapar ·
-        sıralama her ayın 1'i 00:00'da sıfırlanır, biten ayın ilk 3'üne üyelik uzatması otomatik
-        tanımlanır
+        sıralama her ayın 1'i 00:00'da sıfırlanır, biten ayın ilk 3'üne üyelik uzatması tanımlanır
       </p>
     </div>
   );