Merge pull request 'Faz 6: anti-poison (generic-model skip + per-arac kategori tavani)' (#147) from dev into main

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

@@ -28,11 +28,13 @@ function makeDeps(opts: { waiting: number; limitResults: unknown[][] }) {
     // the production nest build, which compiles spec files.
     exists: vi.fn(async (..._args: unknown[]) => false), // no cooldown / guard / complete marker
     get: vi.fn(async () => null), // no no-result residue
-    set: vi.fn(async () => undefined),
+    set: vi.fn(async (..._args: unknown[]) => undefined),
+    del: vi.fn(async (..._args: unknown[]) => undefined),
     incr: vi.fn(async (..._args: unknown[]) => 1), // per-source rate window counter
     expire: vi.fn(async () => undefined),
+    ttl: vi.fn(async () => -2), // checkCooldown: no cooldown key
     setNx: vi.fn(async () => true), // scan lock acquired by default
-    getJson: vi.fn(async () => null), // Phase-2 cursor empty
+    getJson: vi.fn(async (..._args: unknown[]): Promise<unknown> => null), // cursor / progress empty
     setJson: vi.fn(async () => undefined),
   };
   const posthog = { payload: vi.fn(async () => ({})) }; // compiled-in defaults
@@ -222,6 +224,58 @@ describe("PrefetchWorkerService — fast lane (lifo) + backlog gating", () => {
     });
   });
 
+  describe("poison guards (Faz 6: generic-model + category cap)", () => {
+    type PI = { processInit: (j: unknown) => Promise<void> };
+    type PC = { processChildren: (j: unknown) => Promise<void> };
+
+    it("processInit skips + marks poison for a generic-model vehicle (model == brand)", async () => {
+      const { service, queue, redis } = makeDeps({
+        waiting: 0,
+        limitResults: [[{ id: "op1", brandName: "Opel", model: "Opel" }]], // vehicle lookup
+      });
+      await (service as never as PI).processInit({
+        data: { vehicleId: "op1", source: "pl24" },
+      } as never);
+      const poisonSet = redis.set.mock.calls.some((c) =>
+        String(c[0]).includes("prefetch:poison:op1"),
+      );
+      expect(poisonSet).toBe(true);
+      expect(queue.add).not.toHaveBeenCalled(); // tree never drilled
+    });
+
+    it("processChildren marks poison once progress.total exceeds CATEGORY_CAP", async () => {
+      const { service, queue, redis } = makeDeps({ waiting: 0, limitResults: [] });
+      redis.getJson.mockImplementation(async (...a: unknown[]) =>
+        String(a[0]).includes("progress") ? { total: 5000 } : null,
+      );
+      await (service as never as PC).processChildren({
+        data: { vehicleId: "op1", categoryId: "c1", source: "pl24", depth: 1 },
+      } as never);
+      const poisonSet = redis.set.mock.calls.some((c) =>
+        String(c[0]).includes("prefetch:poison:op1"),
+      );
+      expect(poisonSet).toBe(true);
+      expect(queue.add).not.toHaveBeenCalled(); // stopped drilling
+    });
+
+    it("scan skips a poison-marked vehicle", async () => {
+      const { service, queue, redis } = makeDeps({
+        waiting: 10,
+        limitResults: [
+          [],
+          [{ id: "op1", source: "pl24", createdAt: new Date("2026-01-01T00:00:00Z") }],
+        ],
+      });
+      redis.exists.mockImplementation(async (...a: unknown[]) =>
+        String(a[0]).includes("prefetch:poison:"),
+      );
+      await (
+        service as never as { processBackfillScan: () => Promise<void> }
+      ).processBackfillScan();
+      expect(queue.add).not.toHaveBeenCalled();
+    });
+  });
+
   describe("queueCategoryJob — depth cap (no-op guard)", () => {
     // pl24 linkPath with no leaf marker (/bom/, /partinfo/ …) → treated as a
     // non-leaf folder that would normally queue a prefetch-children job.

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

@@ -113,6 +113,18 @@ const NORESULT_TTL_S = (Number(process.env.PREFETCH_NORESULT_TTL_DAYS) || 7) * 8
  *  forever). The TTL re-validates periodically so a later upstream catalog change
  *  is eventually picked up (default 21 days). Env-tunable. */
 const COMPLETE_TTL_S = (Number(process.env.PREFETCH_COMPLETE_TTL_DAYS) || 21) * 86_400;
+/**
+ * Per-vehicle category ceiling — anti-poison guard. A correctly-decoded PL24
+ * vehicle has ~100-800 categories (p95 744). A vehicle whose tree blows past
+ * this is almost always a decode that failed to resolve the model and landed on
+ * the generic ROOT catalog (the whole brand universe — e.g. 5 Opels with
+ * model="Opel" produced 420k/102k/21k categories and ZERO parts). Once a
+ * vehicle's discovered tree exceeds this, stop drilling it and mark it poison so
+ * it never gets re-picked. Env-tunable. */
+const CATEGORY_CAP = Number(process.env.PREFETCH_CATEGORY_CAP) || 3000;
+/** TTL for the poison marker (generic-model / over-cap vehicles). Long, because
+ *  the underlying cause is a decode bug — re-validate monthly. */
+const POISON_TTL_S = (Number(process.env.PREFETCH_POISON_TTL_DAYS) || 30) * 86_400;
 
 @Injectable()
 export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
@@ -263,9 +275,15 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
     await checkCooldown(this.redis, source);
     checkTimeWindow(source);
 
+    // Already flagged as poison (over-cap / generic model on a prior run) — skip.
+    if (await this.redis.exists(this.poisonKey(vehicleId))) {
+      this.logger.debug(`[prefetch] Skip poison vehicle=${vehicleId}`);
+      return;
+    }
+
     // Verify vehicle still exists
     const [vehicle] = await this.db
-      .select({ id: vehicles.id })
+      .select({ id: vehicles.id, brandName: vehicles.brandName, model: vehicles.model })
       .from(vehicles)
       .where(eq(vehicles.id, vehicleId))
       .limit(1);
@@ -275,7 +293,20 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
       return;
     }
 
-    await initProgress(this.redis, vehicleId);
+    // Generic-model guard: when VIN decode fails to resolve the model it stays
+    // equal to the brand (e.g. model "Opel" for brand "Opel") and the catalog
+    // fetch lands on the brand's ROOT tree — the whole model universe, hundreds
+    // of thousands of part-less categories. Don't drill it; mark poison until the
+    // decode is fixed. A real model is always more specific than the brand.
+    if (this.isGenericModel(vehicle.brandName, vehicle.model)) {
+      this.logger.warn(
+        `[prefetch] Skip generic-model vehicle=${vehicleId} (brand=${vehicle.brandName}, model=${vehicle.model}) — marking poison`,
+      );
+      await this.markPoison(vehicleId);
+      return;
+    }
+
+    await initProgress(this.redis, vehicleId); // resets progress.total — the cap's tree-size proxy
 
     // Get all top-level categories for this vehicle
     let topCategories = await this.db
@@ -389,6 +420,19 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
       return;
     }
 
+    // Anti-poison cap: progress.total tracks every sub-job queued for this vehicle
+    // (≈ its discovered tree size). Once it blows past the ceiling the vehicle is a
+    // generic-catalog explosion — stop drilling and mark it poison so the rest of
+    // its (part-less) tree is never fetched and it's never re-picked.
+    const prog = await this.redis.getJson<{ total: number }>(`prefetch:progress:${vehicleId}`);
+    if ((prog?.total ?? 0) >= CATEGORY_CAP) {
+      this.logger.warn(
+        `[prefetch] Category cap ${CATEGORY_CAP} hit for vehicle=${vehicleId} (tree≈${prog?.total}) — marking poison, stop drilling`,
+      );
+      await this.markPoison(vehicleId);
+      return;
+    }
+
     try {
       const children = await this.categoriesService.getChildren(categoryId);
 
@@ -514,6 +558,8 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
       // every wave; the marker's TTL re-validates them periodically. Phase-1
       // (zero-parts) vehicles never carry this marker, so this is a no-op there.
       if (await this.redis.exists(this.completeKey(v.id))) return;
+      // Skip poison vehicles (generic-model / over-cap catalog explosions).
+      if (await this.redis.exists(this.poisonKey(v.id))) return;
       seen.add(v.id);
       picked.push({ id: v.id, source: v.source, fast });
     };
@@ -773,6 +819,29 @@ export class PrefetchWorkerService implements OnModuleInit, OnModuleDestroy {
     return `prefetch:complete:${vehicleId}`;
   }
 
+  private poisonKey(vehicleId: string): string {
+    return `prefetch:poison:${vehicleId}`;
+  }
+
+  /**
+   * True when VIN decode failed to resolve a model: it stays equal to the brand
+   * (case-insensitive) or is empty. Such a vehicle can't be catalog-scoped and
+   * fetches the brand's whole root tree — see CATEGORY_CAP.
+   */
+  private isGenericModel(brandName: string | null, model: string | null): boolean {
+    const m = (model ?? "").trim();
+    if (!m) return true;
+    return m.toLowerCase() === (brandName ?? "").trim().toLowerCase();
+  }
+
+  /** Flag a vehicle as poison (generic-model / over-cap) so init/scan skip it,
+   *  and clear its in-flight + progress state. */
+  private async markPoison(vehicleId: string): Promise<void> {
+    await this.redis.set(this.poisonKey(vehicleId), "1", POISON_TTL_S);
+    await this.redis.del(`prefetch:scheduled:${vehicleId}`);
+    await this.redis.del(`prefetch:progress:${vehicleId}`);
+  }
+
   /**
    * Per-source fixed-window (60s) rate limit. Increments the source's window
    * counter; once it exceeds the source's ceiling, throws RateLimitError