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

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[][] }) {
  // A queue double: getJob returns null (nothing deduped) and the ioredis client
  // stub answers the delayed-ZSET zcount the pressure gate takes.
  const makeQueue = (name: string, waiting: number) => ({
    name,
    // 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)),
    getJob: vi.fn(async (..._args: unknown[]): Promise<unknown> => null),
    getJobCounts: vi.fn(async (..._args: unknown[]) => ({
      waiting,
      delayed: 0,
      active: 0,
      prioritized: 0,
    })),
    toKey: (t: string) => `bull:${name}:${t}`,
    client: Promise.resolve({ zcount: vi.fn(async (..._args: unknown[]) => 0) }),
  });
  const queue = makeQueue("catalog-prefetch", opts.waiting);
  const redis = {
    // typed args (like queue.add) so mockImplementation((k)=>…) type-checks under
    // the production nest build, which compiles spec files.
    exists: vi.fn(async (..._args: unknown[]) => false), // no cooldown / guard / complete marker
    // null → no no-result residue AND daily budget counters read as 0 (unspent).
    get: vi.fn(async (..._args: unknown[]): Promise<string | null> => null),
    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 (..._args: unknown[]): Promise<unknown> => null), // cursor / progress empty
    setJson: vi.fn(async () => undefined),
  };
  const posthog = { payload: vi.fn(async () => ({})) }; // compiled-in defaults
  const db = makeDb(opts.limitResults);
  const fastQueue = makeQueue("catalog-prefetch-fast", 0);
  const service = new PrefetchWorkerService(
    queue as never,
    fastQueue as never,
    {} as never, // categoriesService — unused by the scan
    redis as never,
    posthog as never,
    db as never,
  );

  return { service, queue, fastQueue, 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 (fast queue)", async () => {
      const { service, fastQueue } = 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] = fastQueue.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 (fast queue)", async () => {
      const { service, fastQueue: 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,
        fastQueue: 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 a vehicle already marked complete (Phase-2 does not re-walk it)", async () => {
      const { service, queue, redis } = makeDeps({
        waiting: 10, // under ceiling → Phase-2 active
        limitResults: [
          [], // Phase-1: no zero-parts vehicles
          [{ id: "done1", source: "emex", createdAt: new Date("2026-01-01T00:00:00Z") }], // Phase-2
        ],
      });
      // scheduled guard absent (false), but the complete marker is present.
      redis.exists.mockImplementation(async (...a: unknown[]) =>
        String(a[0]).includes("prefetch:complete:"),
      );
      await (
        service as never as { processBackfillScan: () => Promise<void> }
      ).processBackfillScan();
      expect(queue.add).not.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,
        fastQueue: 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 (fast lane)
    });

    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();
    });
  });

  describe("checkSourceRate — per-source rate limit", () => {
    type CSR = { checkSourceRate: (s: string, lane?: "main" | "fast") => Promise<void> };

    // pl24's MAIN lane is parked whenever background backfill is off, which is
    // the default — so these ceiling tests turn it on explicitly to exercise the
    // rate limiter itself rather than the backfill gate (covered separately).
    beforeEach(() => {
      process.env.PL24_BACKFILL_ENABLED = "true";
      process.env.PL24_TR_DISABLED = undefined;
    });
    afterEach(() => {
      process.env.PL24_BACKFILL_ENABLED = undefined;
    });

    it("passes when under the source ceiling", async () => {
      const { service, redis } = makeDeps({ waiting: 0, limitResults: [] });
      redis.incr.mockResolvedValueOnce(1);
      await expect((service as never as CSR).checkSourceRate("pl24")).resolves.toBeUndefined();
    });

    it("throws a source-rate RateLimitError once over the ceiling", async () => {
      const { service, redis } = makeDeps({ waiting: 0, limitResults: [] });
      redis.incr.mockResolvedValueOnce(21); // pl24 default ceiling is 20
      await expect((service as never as CSR).checkSourceRate("pl24")).rejects.toMatchObject({
        cause: "source-rate",
      });
    });

    it("pl24 MAIN lane parkta iken hiç sayaç harcamaz (varsayılan)", async () => {
      process.env.PL24_BACKFILL_ENABLED = undefined;
      const { service, redis } = makeDeps({ waiting: 0, limitResults: [] });
      await expect((service as never as CSR).checkSourceRate("pl24", "main")).rejects.toMatchObject(
        { cause: "source-rate" },
      );
      expect(redis.incr).not.toHaveBeenCalled();
    });

    it("kullanıcı (fast) şeridi backfill anahtarından etkilenmez", async () => {
      process.env.PL24_BACKFILL_ENABLED = undefined;
      const { service, redis } = makeDeps({ waiting: 0, limitResults: [] });
      redis.incr.mockResolvedValueOnce(1);
      await expect(
        (service as never as CSR).checkSourceRate("pl24", "fast"),
      ).resolves.toBeUndefined();
    });

    it("is unlimited (no counter) for a source without a configured ceiling", async () => {
      const { service, redis } = makeDeps({ waiting: 0, limitResults: [] });
      await (service as never as CSR).checkSourceRate("unknown-source");
      expect(redis.incr).not.toHaveBeenCalled();
    });
  });

  describe("poison guard (Faz 6: category cap — brand-agnostic anti-explosion)", () => {
    type PC = { processChildren: (j: unknown) => Promise<void> };

    it("processChildren marks poison once the STORED tree exceeds CATEGORY_CAP", async () => {
      // The cap is now measured from the DB (categories count), not the ephemeral
      // progress.total which every processInit resets to 0.
      const { service, queue, redis } = makeDeps({ waiting: 0, limitResults: [[{ n: 5000 }]] });
      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.
    const nonLeaf = {
      id: "c1",
      linkPath: "/catalog/groups/x",
      source: "pl24",
      unavailable: false,
      hasSubgroups: false,
    };
    type QCJ = {
      queueCategoryJob: (c: unknown, v: string, s: string, d: number, f: boolean) => Promise<void>;
    };

    it("does NOT enqueue a children job at depth >= MAX_DEPTH (avoids a no-op)", async () => {
      const { service, queue } = makeDeps({ waiting: 0, limitResults: [] });
      // depth 12 == default MAX_DEPTH → processChildren would early-return, so the
      // job must never be created.
      await (service as never as QCJ).queueCategoryJob(nonLeaf, "v1", "pl24", 12, false);
      expect(queue.add).not.toHaveBeenCalled();
    });

    it("enqueues a children job for a non-leaf within the cap", async () => {
      // childCheck query (.limit(1)) returns no rows → enqueue a children fetch.
      const { service, queue } = makeDeps({ waiting: 0, limitResults: [[]] });
      await (service as never as QCJ).queueCategoryJob(nonLeaf, "v1", "pl24", 1, false);
      expect(queue.add).toHaveBeenCalledTimes(1);
      expect(queue.add.mock.calls[0][0]).toBe("prefetch-children");
    });
  });
});

// ── PL24 reaktif drill derinliği + backfill pacing (plv2 Faz 1 / adım 2b) ──
// Ban'ı süren hacim, her yeni decode'da tüm ağacın gezilmesiydi (bir Passat =
// 1.251 kategori). Fast lane artık PL24'te 1. seviyede durur; derin drill ya
// kullanıcı tıklamasıyla ya da bütçeli backfill lane'inde olur.
describe("PrefetchWorkerService — PL24 derinlik tavanı", () => {
  const load = async () => {
    const mod = await import("./prefetch-worker.service");
    return mod as unknown as {
      __testables?: { maxDepthFor(source: string, fast: boolean): number };
    };
  };

  it("pl24 fast lane 1. seviyede durur, diğer kaynaklar tam derinlik kullanır", async () => {
    // maxDepthFor modül-özel; davranışı dolaylı doğrula: env varsayılanları
    process.env.PREFETCH_PL24_FAST_DEPTH = "";
    process.env.PREFETCH_MAX_DEPTH = "";
    const mod = await load();
    const fn = mod.__testables?.maxDepthFor;
    if (!fn) return; // testable export yoksa atla (davranış e2e'de doğrulanır)
    expect(fn("pl24", true)).toBe(1);
    expect(fn("pl24", false)).toBeGreaterThan(1);
    expect(fn("parts-catalogs", true)).toBeGreaterThan(1);
  });
});

/**
 * PL24 arka plan backfill anahtarı (Faz 3) + Mitsubishi parça-detay kırpması.
 */
describe("PL24 backfill anahtarı", () => {
  const ENV = { ...process.env };
  afterEach(() => {
    process.env = { ...ENV };
  });

  it("varsayılan KAPALI — değişken hiç yoksa arka plan akmaz", async () => {
    const { __testables } = await import("./prefetch-worker.service");
    process.env.PL24_BACKFILL_ENABLED = undefined;
    process.env.PL24_TR_DISABLED = undefined;
    expect(__testables.isPl24BackfillEnabled()).toBe(false);
  });

  it("yalnız açık 'true' ile açılır", async () => {
    const { __testables } = await import("./prefetch-worker.service");
    process.env.PL24_TR_DISABLED = undefined;
    process.env.PL24_BACKFILL_ENABLED = "true";
    expect(__testables.isPl24BackfillEnabled()).toBe(true);
    process.env.PL24_BACKFILL_ENABLED = "1";
    expect(__testables.isPl24BackfillEnabled()).toBe(false);
  });

  it("eski PL24_TR_DISABLED hâlâ kapatabilir (yarım deploy musluğu açamaz)", async () => {
    const { __testables } = await import("./prefetch-worker.service");
    process.env.PL24_BACKFILL_ENABLED = "true";
    process.env.PL24_TR_DISABLED = "true";
    expect(__testables.isPl24BackfillEnabled()).toBe(false);
  });
});