diff --git a/apps/api/src/jobs/jobs.module.ts b/apps/api/src/jobs/jobs.module.ts index 8c7ac09..ef92b7d 100644 --- a/apps/api/src/jobs/jobs.module.ts +++ b/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 diff --git a/apps/api/src/jobs/prefetch-worker.service.spec.ts b/apps/api/src/jobs/prefetch-worker.service.spec.ts index ccbac52..6cc60a5 100644 --- a/apps/api/src/jobs/prefetch-worker.service.spec.ts +++ b/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 }) - .enqueueInit("v1", "emex", true); + await ( + service as never as { enqueueInit: (a: string, b: string, c: boolean) => Promise } + ).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 }) - .enqueueInit("v1", "emex", false); + await ( + service as never as { enqueueInit: (a: string, b: string, c: boolean) => Promise } + ).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 }).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 }).processBackfillScan(); + await ( + service as never as { processBackfillScan: () => Promise } + ).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 }).processBackfillScan(); + await ( + service as never as { processBackfillScan: () => Promise } + ).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 }).processBackfillScan(); + await ( + service as never as { processBackfillScan: () => Promise } + ).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 }).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 }).runBackfillScan(); expect(queue.add).not.toHaveBeenCalled(); }); }); diff --git a/apps/api/src/jobs/prefetch-worker.service.ts b/apps/api/src/jobs/prefetch-worker.service.ts index 5846851..095f615 100644 --- a/apps/api/src/jobs/prefetch-worker.service.ts +++ b/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; + private backfillInterval?: ReturnType; 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 { + 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 { try { const data = job.data as { source?: string };