/**
 * TASK-0135 — cache-layer unit tests (pure, fake-Redis-backed).
 *
 * The CI `unit` job has NO Redis/Postgres services, so these tests exercise
 * the cache's EXACT invalidation semantics against a deterministic in-memory
 * fake (get/set/del/incr/scan with TTL semantics + a virtual clock):
 *
 *  - version-keyed tree freshness (a write that bumps the version invalidates
 *    the previous key — no stale read is possible);
 *  - permission TTL = 60 s (the "no event → stale ≤ 60 s" bound) + bust on
 *    `permission.changed` (the next decision is fresh);
 *  - search TTL = 5 s, keyed by principal (two principals never share);
 *  - metadata TTL = 60 s + bust on `database.updated`;
 *  - every entry is zod-validated on read: corrupt JSON or a shape mismatch
 *    is a cache MISS (never a throw, never a 500);
 *  - Redis failure degrades to pass-through.
 *
 * The LIVE concurrent-write race (real Redis + real API writes) is the
 * integration suite (`tests/integration/cache-invalidation.integration.test.ts`).
 */
import { describe, expect, it, beforeEach } from "vitest";
import { z } from "zod";
import { EventEmitter } from "node:events";
import { CacheService, CACHE_TTLS } from "./cache.js";
import { EVENTS } from "../constants.js";
import { resetEnvCache } from "../config/env.js";

// ---------------------------------------------------------------------------
// deterministic fake Redis (TTL + virtual clock)
// ---------------------------------------------------------------------------

interface FakeEntry {
  value: string;
  expireAt: number | null; // virtual ms
}

class FakeRedis {
  store = new Map<string, FakeEntry>();
  now = 0; // virtual clock (ms)
  fail = false; // when true, every op rejects (simulates a blip)

  get nowMs(): number {
    return this.now;
  }
  /** Advance the virtual clock by `ms` and expire any entries past their TTL. */
  tick(ms: number): void {
    this.now += ms;
    for (const [k, v] of [...this.store]) if (v.expireAt !== null && v.expireAt <= this.now) this.store.delete(k);
  }

  private guard(): void {
    if (this.fail) throw new Error("ECONNRESET (fake)");
  }

  async get(key: string): Promise<string | null> {
    this.guard();
    const e = this.store.get(key);
    if (!e) return null;
    if (e.expireAt !== null && e.expireAt <= this.now) {
      this.store.delete(key);
      return null;
    }
    return e.value;
  }
  async set(key: string, value: string, ...args: string[]): Promise<"OK"> {
    this.guard();
    let ttlMs: number | null = null;
    // ioredis: SET key value PX ms
    for (let i = 0; i < args.length; i++) {
      if (args[i] === "PX") ttlMs = Number(args[i + 1]);
    }
    this.store.set(key, { value, expireAt: ttlMs === null ? null : this.now + ttlMs });
    return "OK";
  }
  async del(...keys: string[]): Promise<number> {
    this.guard();
    let n = 0;
    for (const k of keys) if (this.store.delete(k)) n++;
    return n;
  }
  async incr(key: string): Promise<number> {
    this.guard();
    const cur = this.store.get(key);
    const n = (cur ? parseInt(cur.value, 10) : 0) + 1;
    this.store.set(key, { value: String(n), expireAt: null });
    return n;
  }
  async scan(cursor: string, ...args: string[]): Promise<[string, string[]]> {
    this.guard();
    let pattern = "*";
    for (let i = 0; i < args.length; i++) if (args[i] === "MATCH") pattern = args[i + 1];
    const rx = new RegExp("^" + pattern.split("*").map(escapeRe).join(".*") + "$");
    const keys: string[] = [];
    for (const [k, v] of this.store) {
      if (v.expireAt !== null && v.expireAt <= this.now) continue;
      if (rx.test(k)) keys.push(k);
    }
    return ["0", keys];
  }
}

function escapeRe(s: string): string {
  return s.replace(/[.*+?^${}()|[\]\\]/g, "\\$&");
}

// A fake EventBus with the same onLocal/publish surface the cache consumes.
class FakeEvents {
  private emitter = new EventEmitter();
  onLocal(event: string, handler: (body: { event: string; payload: string }) => void): () => void {
    this.emitter.on(event, handler);
    return () => this.emitter.off(event, handler);
  }
  async publish(event: string, payload: Record<string, unknown>): Promise<void> {
    this.emitter.emit(event, { event, payload: JSON.stringify(payload) });
  }
}

const LEVEL = z.enum(["NONE", "VIEW", "COMMENT", "EDIT", "FULL_ACCESS"]);
const TREE_ITEM = z.object({ id: z.string(), position: z.string() });

interface Deps {
  cache: CacheService;
  redis: FakeRedis;
  events: FakeEvents;
}

/**
 * Build a CacheService wired to a fake Redis + events, and force the cache
 * ON. The service reads `PERF_CACHE` at construction — set the env before
 * building (these run in a vitest worker with no live Redis, so the app
 * default "auto" would be off).
 */
async function build(): Promise<Deps> {
  const prev = process.env.PERF_CACHE;
  process.env.PERF_CACHE = "on";
  resetEnvCache();
  try {
    const redis = new FakeRedis();
    const events = new FakeEvents();
    const fakeRedisService = { getRedis: () => redis as unknown as import("ioredis").default } as unknown as {
      getRedis(): import("ioredis").default;
    };
    const cache = new CacheService(fakeRedisService as never, events as never);
    cache.onModuleInit();
    return { cache, redis, events };
  } finally {
    if (prev === undefined) delete process.env.PERF_CACHE;
    else process.env.PERF_CACHE = prev;
    resetEnvCache();
  }
}

describe("cache (TASK-0135 invalidation semantics)", () => {
  beforeEach(() => {
    // ensure a clean env between tests
    delete process.env.PERF_CACHE;
  });

  describe("enablement gate (PERF_CACHE)", () => {
    it("auto (default) is OFF outside production → pass-through, no Redis writes", async () => {
      const prev = process.env.PERF_CACHE;
      const prevNode = process.env.NODE_ENV;
      process.env.PERF_CACHE = "auto";
      process.env.NODE_ENV = "test";
      resetEnvCache();
      try {
        const redis = new FakeRedis();
        const events = new FakeEvents();
        const cache = new CacheService({ getRedis: () => redis as never } as never, events as never);
        cache.onModuleInit();
        expect(cache.isOn()).toBe(false);
        const { data, cached } = await cache.getMetadata("db1", "properties", async () => ({ ok: true }), z.object({ ok: z.boolean() }));
        expect(cached).toBe(false);
        expect(data).toEqual({ ok: true });
        expect(redis.store.size).toBe(0); // nothing written
      } finally {
        if (prev === undefined) delete process.env.PERF_CACHE;
        else process.env.PERF_CACHE = prev;
        if (prevNode === undefined) delete process.env.NODE_ENV;
        else process.env.NODE_ENV = prevNode;
        resetEnvCache();
      }
    });

    it("`on` is ON regardless of NODE_ENV", async () => {
      const prev = process.env.PERF_CACHE;
      process.env.PERF_CACHE = "on";
      resetEnvCache();
      try {
        const redis = new FakeRedis();
        const events = new FakeEvents();
        const cache = new CacheService({ getRedis: () => redis as never } as never, events as never);
        cache.onModuleInit();
        expect(cache.isOn()).toBe(true);
      } finally {
        if (prev === undefined) delete process.env.PERF_CACHE;
        else process.env.PERF_CACHE = prev;
        resetEnvCache();
      }
    });
  });

  describe("tree (version-key) freshness", () => {
    it("a write that bumps the version invalidates the previous key (no stale read)", async () => {
      const { cache, events, redis } = await build();
      const ws = "ws1";
      const loads: string[] = [];
      const loader = async () => {
        loads.push("load");
        return [{ id: "b1", position: "a0" }, { id: "b2", position: "a1" }];
      };

      // First read: miss → load → store under version 1 (bump happens via a
      // prior write, so seed the version).
      await cache.bumpVersion(ws); // v1 (a write happened)
      const r1 = await cache.getTree(ws, "p1", loader, z.array(TREE_ITEM));
      expect(r1.cached).toBe(false);
      expect(loads).toHaveLength(1);

      // Second read, no write in between: HIT (same version key).
      const r2 = await cache.getTree(ws, "p1", loader, z.array(TREE_ITEM));
      expect(r2.cached).toBe(true);
      expect(loads).toHaveLength(1); // no second load

      // A concurrent write bumps the version (page.updated → bumpVersion).
      await events.publish(EVENTS.PAGE_UPDATED, { workspaceId: ws, pageId: "p1" });
      // The bump is async (fire-and-forget from the bus); wait one tick.
      await new Promise((r) => setTimeout(r, 10));
      expect(await cache.getVersion(ws)).toBe(2);

      // The NEXT read uses the new version key → MISS → fresh load.
      const r3 = await cache.getTree(ws, "p1", loader, z.array(TREE_ITEM));
      expect(r3.cached).toBe(false);
      expect(loads).toHaveLength(2);
      // The stale v1 key is still in Redis (orphaned) but can never be
      // served again (reads key by the CURRENT version).
      expect(redis.store.has("perf:tree:ws1:p1:1")).toBe(true);
      expect(redis.store.has("perf:tree:ws1:p1:2")).toBe(true);
    });

    it("version 0 (no write yet / Redis cold) never caches — always loads", async () => {
      const { cache } = await build();
      const loads: string[] = [];
      const loader = async () => {
        loads.push("x");
        return [{ id: "b1", position: "a0" }];
      };
      const r = await cache.getTree("wsX", "p1", loader, z.array(TREE_ITEM));
      expect(r.cached).toBe(false);
      expect(loads).toHaveLength(1);
      const r2 = await cache.getTree("wsX", "p1", loader, z.array(TREE_ITEM));
      expect(r2.cached).toBe(false);
      expect(loads).toHaveLength(2); // never cached at version 0
    });
  });

  describe("permission (TTL 60 s + bust)", () => {
    it("caches a decision and serves it within the TTL (no event → stale ≤ 60 s)", async () => {
      const { cache, redis } = await build();
      let loads = 0;
      const loader = async () => {
        loads++;
        return "EDIT" as const;
      };
      const scope = { pageId: "p1" };
      const r1 = await cache.getPermission("ws1", "u1", scope, loader, LEVEL);
      expect(r1.cached).toBe(false);
      expect(loads).toBe(1);
      const r2 = await cache.getPermission("ws1", "u1", scope, loader, LEVEL);
      expect(r2.cached).toBe(true);
      expect(loads).toBe(1); // served from cache

      // The TTL is exactly 60 s: at 59.9 s the entry is still live, at 60.1 s
      // it is gone (the next decision is fresh).
      expect(CACHE_TTLS.PERMISSION_MS).toBe(60_000);
      const key = [...redis.store.keys()].find((k) => k.startsWith("perf:perm:ws1:u1:"));
      expect(key).toBeDefined();
      redis.tick(CACHE_TTLS.PERMISSION_MS - 100);
      const stillLive = (await redis.get(key!)) !== null;
      expect(stillLive).toBe(true);
      redis.tick(200); // now 60.1 s elapsed
      const r3 = await cache.getPermission("ws1", "u1", scope, loader, LEVEL);
      expect(r3.cached).toBe(false);
      expect(loads).toBe(2); // fresh after TTL
    });

    it("a permission.changed event busts the subject's entries → next decision fresh", async () => {
      const { cache, events } = await build();
      let loads = 0;
      const loader = async () => {
        loads++;
        return (loads === 1 ? "EDIT" : "NONE") as "EDIT" | "NONE";
      };
      const scope = { pageId: "p1" };
      const r1 = await cache.getPermission("ws1", "u2", scope, loader, LEVEL);
      expect(r1.level).toBe("EDIT");
      // The grant is revoked → the API publishes permission.changed.
      await events.publish(EVENTS.PERMISSION_CHANGED, { workspaceId: "ws1", userId: "u2", removed: true });
      await new Promise((r) => setTimeout(r, 15)); // async scanDel
      // The NEXT decision is FRESH (the bust removed the cached EDIT).
      const r2 = await cache.getPermission("ws1", "u2", scope, loader, LEVEL);
      expect(r2.cached).toBe(false);
      expect(r2.level).toBe("NONE");
      expect(loads).toBe(2);
    });

    it("different users never share a permission entry", async () => {
      const { cache, redis } = await build();
      const scope = { pageId: "p1" };
      await cache.getPermission("ws1", "uA", scope, async () => "EDIT" as const, LEVEL);
      await cache.getPermission("ws1", "uB", scope, async () => "NONE" as const, LEVEL);
      const keys = [...redis.store.keys()].filter((k) => k.startsWith("perf:perm:ws1:"));
      expect(keys).toHaveLength(2); // one per user
    });
  });

  describe("search (TTL 5 s, permission-keyed)", () => {
    it("caches for exactly 5 s and is keyed by principal (no cross-principal sharing)", async () => {
      const { cache, redis } = await build();
      expect(CACHE_TTLS.SEARCH_MS).toBe(5_000);
      let loads = 0;
      const loader = async () => {
        loads++;
        return { data: ["hit"] };
      };
      const input = { q: "notion", types: ["page"], workspaceId: "ws1", cursor: null, limit: 20 };
      const r1 = await cache.getSearch("ws1", "principal-A", input, loader, z.object({ data: z.array(z.string()) }));
      expect(r1.cached).toBe(false);
      const r2 = await cache.getSearch("ws1", "principal-A", input, loader, z.object({ data: z.array(z.string()) }));
      expect(r2.cached).toBe(true);
      expect(loads).toBe(1);

      // A DIFFERENT principal → a different key → a fresh load (permission
      // relevant: two principals must never see each other's results).
      const r3 = await cache.getSearch("ws1", "principal-B", input, loader, z.object({ data: z.array(z.string()) }));
      expect(r3.cached).toBe(false);
      expect(loads).toBe(2);
      const keys = [...redis.store.keys()].filter((k) => k.startsWith("perf:search:ws1:"));
      expect(keys).toHaveLength(2);

      // The 5 s TTL: alive at 4.9 s, gone at 5.1 s.
      const keyA = keys.find((k) => k.includes("principal-A"))!;
      redis.tick(4_900);
      expect(await redis.get(keyA)).not.toBeNull();
      redis.tick(300);
      const r4 = await cache.getSearch("ws1", "principal-A", input, loader, z.object({ data: z.array(z.string()) }));
      expect(r4.cached).toBe(false);
      expect(loads).toBe(3);
    });
  });

  describe("metadata (TTL 60 s + bust on database.updated)", () => {
    it("busts on database.updated → next read is fresh", async () => {
      const { cache, events } = await build();
      let loads = 0;
      const loader = async () => {
        loads++;
        return loads === 1 ? [{ id: "prop1" }] : [{ id: "prop1" }, { id: "prop2" }];
      };
      const schema = z.array(z.object({ id: z.string() }));
      const r1 = await cache.getMetadata("db1", "properties", loader, schema);
      expect(r1.cached).toBe(false);
      expect((r1.data as { id: string }[])).toHaveLength(1);
      // A property is added → database.updated carries databaseId.
      await events.publish(EVENTS.DATABASE_UPDATED, { workspaceId: "ws1", databaseId: "db1", propertyId: "prop2" });
      await new Promise((r) => setTimeout(r, 10));
      const r2 = await cache.getMetadata("db1", "properties", loader, schema);
      expect(r2.cached).toBe(false);
      expect((r2.data as { id: string }[])).toHaveLength(2);
      expect(loads).toBe(2);
    });
  });

  describe("zod-on-read: corrupt entry = miss, never a throw", () => {
    it("corrupt JSON is a miss (and is replaced with a valid entry), never a throw", async () => {
      const { cache, redis } = await build();
      // Write a corrupt value directly (simulating a bad partial write).
      const key = "perf:metadata:db1:properties";
      await redis.set(key, "{not-json", "PX", "60000");
      const { data, cached } = await cache.getMetadata("db1", "properties", async () => ({ ok: true }), z.object({ ok: z.boolean() }));
      expect(cached).toBe(false);
      expect(data).toEqual({ ok: true });
      // The corrupt entry is gone: the key now holds the VALID refilled value
      // (a corrupt entry can never be served, and is never left in place).
      const nowStored = await redis.get(key);
      expect(nowStored).toBe(JSON.stringify({ ok: true }));
    });

    it("a shape mismatch is a miss (and is replaced with a valid entry), never a throw", async () => {
      const { cache, redis } = await build();
      const key = "perf:metadata:db1:properties";
      await redis.set(key, JSON.stringify(["wrong-shape"]), "PX", "60000");
      const { data, cached } = await cache.getMetadata("db1", "properties", async () => ({ ok: true }), z.object({ ok: z.boolean() }));
      expect(cached).toBe(false);
      expect(data).toEqual({ ok: true });
      const nowStored = await redis.get(key);
      expect(nowStored).toBe(JSON.stringify({ ok: true }));
    });

    it("a corrupt permission entry is a miss, not a thrown 500", async () => {
      const { cache, redis } = await build();
      // Find the perm key the cache would use and corrupt it.
      const scope = { pageId: "p1" };
      await cache.getPermission("ws1", "u9", scope, async () => "EDIT" as const, LEVEL);
      const key = [...redis.store.keys()].find((k) => k.startsWith("perf:perm:ws1:u9:"))!;
      await redis.set(key, '"BOGUS-LEVEL"', "PX", "60000"); // not a valid Level
      const { level, cached } = await cache.getPermission("ws1", "u9", scope, async () => "VIEW" as const, LEVEL);
      expect(cached).toBe(false);
      expect(level).toBe("VIEW"); // fresh, not the corrupt value
    });
  });

  describe("Redis failure degrades to pass-through (never a 500)", () => {
    it("a Redis blip makes reads fall through to the loader without throwing", async () => {
      const { cache, redis } = await build();
      const loads: string[] = [];
      const loader = async () => {
        loads.push("x");
        return [{ id: "b1", position: "a0" }];
      };
      await cache.bumpVersion("ws1");
      redis.fail = true; // simulate the blip
      const { data, cached } = await cache.getTree("ws1", "p1", loader, z.array(TREE_ITEM));
      expect(cached).toBe(false);
      expect(loads).toHaveLength(1); // fell through to the real loader
      expect(data).toHaveLength(1);
      redis.fail = false;
    });
  });
});