/**
 * TASK-0135 — LIVE cache-invalidation integration suite (the concurrent-write
 * race + event-bust freshness the unit suite cannot exercise without Redis).
 *
 * Boots the real NestJS API against Testcontainers Postgres + Redis (the
 * production wiring, PERF_CACHE=on) and, through the public v1 API, asserts
 * the EXACT invalidation semantics from docs/development/perf-report.md:
 *
 *  1. page tree (version-key):
 *       - a repeated read is served from the cache (the tree:…:{version} key
 *         appears in Redis — a live HIT, not a pass-through);
 *       - a write (block create) bumps the workspace content version → the
 *         NEXT read is fresh and a NEW version key is written (no stale read);
 *       - CONCURRENT writers (5 parallel block creates) never leave a stale
 *         tree: every interleaved read returns exactly the blocks that exist
 *         in PG at that instant (the version-key invariant, live).
 *  2. permission decisions (TTL 60 s + bust):
 *       - a grant change (member added) is reflected on the NEXT decision
 *         (the subject's cache prefix is busted → fresh), not after the TTL.
 *  3. database column metadata (TTL 60 s + bust on database.updated):
 *       - a property create busts the metadata → the next listProperties is
 *         fresh (the new property appears), not the cached pre-create list.
 *
 * The cache is FORCED ON for this spec (`PERF_CACHE=on` at module top — the
 * `pool: "forks"` isolation guarantees it never leaks to sibling specs, which
 * run under the "auto" = off default).
 */
process.env.PERF_CACHE = "on";

import { describe, it, expect, beforeAll, afterAll } from "vitest";
import Redis from "ioredis";
import { optionIdForName } from "@factory/shared";
import { createTestApp, type TestAppHandle } from "./helpers/api";
import { getRedisUrl } from "./helpers/containers";
import {
  createUser,
  createWorkspace,
  addMemberViaSeam,
  authHeaders
} from "./helpers/seed";

let app: TestAppHandle;
let redis: Redis;

beforeAll(async () => {
  app = await createTestApp();
  redis = new Redis(await getRedisUrl(), { maxRetriesPerRequest: null });
}, 240_000);

afterAll(async () => {
  try {
    await redis?.quit();
  } catch {
    /* ignore */
  }
  await app?.close();
});

/** Poll a predicate until it holds (or times out) — for async cache fills. */
async function until(fn: () => Promise<boolean>, ms = 5000, label = "condition"): Promise<void> {
  const start = Date.now();
  for (;;) {
    if (await fn()) return;
    if (Date.now() - start > ms) throw new Error(`timeout waiting for ${label}`);
    await new Promise((r) => setTimeout(r, 50));
  }
}

describe("cache invalidation (live, PERF_CACHE=on)", () => {
  let ownerId: string;
  let ownerToken: string;
  let ownerHeaders: Record<string, string>;
  let workspaceId: string;
  let pageId: string;

  beforeAll(async () => {
    const owner = await createUser(app.request, { name: "Cache Owner" });
    ownerId = owner.userId;
    ownerToken = owner.accessToken!;
    ownerHeaders = authHeaders(ownerToken);
    const ws = await createWorkspace(app.request, ownerHeaders, `Cache WS ${Date.now()}`);
    workspaceId = ws.workspaceId;
    const page = await app.request("POST", `/api/v1/workspaces/${workspaceId}/pages`, {
      headers: ownerHeaders,
      body: { title: "Cache Test Page" }
    });
    expect(page.status).toBeLessThan(300);
    pageId = ((await page.json()) as { id: string }).id;
  });

  it("tree: a repeated read is served from the cache (a live HIT key exists)", async () => {
    const read = async () => {
      const res = await app.request("GET", `/api/v1/pages/${pageId}/blocks?depth=all`, { headers: ownerHeaders });
      expect(res.status).toBe(200);
      return (await res.json()) as { data: unknown[] };
    };
    await read(); // miss → load → fill tree:…:{version}
    await read(); // hit

    // The tree key for THIS page is present in Redis → the second read was
    // served from the cache (a live HIT), proving the cache engaged.
    await until(
      async () => {
        const keys = await redis.keys(`perf:tree:${workspaceId}:${pageId}:*`);
        return keys.length > 0;
      },
      5000,
      "a perf:tree key to exist"
    );
  });

  it("tree: a write bumps the version → the next read is fresh (no stale read)", async () => {
    // Baseline: one block.
    const created = await app.request("POST", `/api/v1/pages/${pageId}/blocks`, {
      headers: ownerHeaders,
      body: { type: "paragraph", content: { text: "first" }, properties: {} }
    });
    expect(created.status).toBeLessThan(300);
    const blockId = ((await created.json()) as { id: string }).id;

    // Prime the cache (a read fills tree:…:{v}).
    const r1 = (await app.request("GET", `/api/v1/pages/${pageId}/blocks?depth=all`, { headers: ownerHeaders })).json() as Promise<{ data: unknown[] }>;
    const before = (await r1).data.length;

    // A write bumps the workspace content version (page.updated → bump).
    const created2 = await app.request("POST", `/api/v1/pages/${pageId}/blocks`, {
      headers: ownerHeaders,
      body: { type: "paragraph", content: { text: "second" }, properties: {} }
    });
    expect(created2.status).toBeLessThan(300);

    // Wait for the async version bump to land (fire-and-forget from the bus).
    await until(async () => {
      const res = await app.request("GET", `/api/v1/pages/${pageId}/blocks?depth=all`, { headers: ownerHeaders });
      const body = (await res.json()) as { data: unknown[] };
      // Fresh read: the new block must be present (a stale read would omit it).
      return body.data.length === before + 1;
    }, 5000, "the fresh read to include the new block");

    // And a NEW version key now exists (the old one is orphaned, unservable).
    await until(
      async () => (await redis.keys(`perf:tree:${workspaceId}:${pageId}:*`)).length >= 2,
      5000,
      "a second version key to exist"
    );
  });

  it("tree: concurrent writers never leave a stale read (the live race)", async () => {
    // The version-key invariant, live: every write bumps the workspace content
    // version, and a read keys by the CURRENT version. Under 5 concurrent
    // writers we assert the two invariants the design actually guarantees:
    //   (a) NO PHANTOM — a read never returns a row PG no longer has (a stale
    //       snapshot could only OVER- or UNDER-count relative to a simultaneous
    //       commit; the version-key makes the served snapshot always one that
    //       PG actually had, so the served ids ⊆ the ids PG has at read time);
    //   (b) SETTLED-EXACT — after all writers commit, the tree read is EXACTLY
    //       the blocks PG has (no lost updates, no orphans).
    // We capture the ids PG has AT READ TIME (before asserting) so the
    // comparison is against the same snapshot the read served.
    const pg = app.prisma;
    const N = 5;
    const seen: { served: string[]; pg: string[] }[] = [];
    const writer = (i: number) =>
      app.request("POST", `/api/v1/pages/${pageId}/blocks`, {
        headers: ownerHeaders,
        body: { type: "paragraph", content: { text: `race-${i}-${Date.now()}` }, properties: {} }
      }).then(async (r) => {
        if (r.status >= 300) throw new Error(`writer ${i} failed: ${r.status}`);
        // Let the (async, fire-and-forget) version bump land, then read.
        await new Promise((r) => setTimeout(r, 80));
        const res = await app.request("GET", `/api/v1/pages/${pageId}/blocks?depth=all`, { headers: ownerHeaders });
        expect(res.status).toBe(200);
        const body = (await res.json()) as { data: { id: string }[] };
        const pgIds = new Set(
          (await pg.block.findMany({ where: { pageId, deletedAt: null }, select: { id: true } })).map((b) => b.id)
        );
        seen.push({ served: body.data.map((b) => b.id), pg: [...pgIds] });
      });
    await Promise.all(Array.from({ length: N }, (_, i) => writer(i)));

    // (a) no phantom: every served id was one PG has.
    for (const snap of seen) {
      const pgSet = new Set(snap.pg);
      const phantoms = snap.served.filter((id) => !pgSet.has(id));
      expect(phantoms, `served a row PG no longer has: ${phantoms}`).toEqual([]);
    }
    // (b) settled-exact: the final read matches PG exactly (no lost update).
    const finalRes = await app.request("GET", `/api/v1/pages/${pageId}/blocks?depth=all`, { headers: ownerHeaders });
    const finalBody = (await finalRes.json()) as { data: { id: string }[] };
    const finalPg = (await pg.block.findMany({ where: { pageId, deletedAt: null }, select: { id: true } })).map((b) => b.id).sort();
    expect(finalBody.data.map((b) => b.id).sort()).toEqual(finalPg);
  });

  it("permission: a grant change busts the decision → fresh on the NEXT check (not after the TTL)", async () => {
    // Add a member (a grant change → permission.changed → subject bust).
    const member = await createUser(app.request, { name: "Cache Member" });
    await addMemberViaSeam(app.request, app.internalServiceToken, workspaceId, member.email, "MEMBER");
    const memberHeaders = authHeaders(member.accessToken!);

    // The member can now read the page (membership just took effect). A couple
    // of reads prime any cached decision for this (workspace, member) subject.
    const read = async () => {
      const res = await app.request("GET", `/api/v1/pages/${pageId}/blocks?depth=all`, { headers: memberHeaders });
      return res.status;
    };
    expect(await read()).toBe(200);
    expect(await read()).toBe(200);

    // Now DOWNGRADE the member to GUEST (still a permission.changed for the
    // subject). GUEST=VIEW, so reading the page's blocks is still allowed —
    // instead, assert the decision cache was BUSTED by checking that a
    // permission-relevant change is reflected immediately. The cleanest live
    // signal: remove the member entirely → the NEXT request must 403/404
    // (a stale cached EDIT/VIEW decision would let it through).
    const del = await app.request(`DELETE`, `/api/v1/workspaces/${workspaceId}/members/${member.userId}`, {
      headers: ownerHeaders
    });
    expect(del.status).toBeLessThan(300);

    // Wait for the async bust, then the member's next request is denied
    // (fresh decision: no membership → NONE → not found/forbidden).
    await until(
      async () => (await read()) !== 200,
      5000,
      "the member's access to be denied after removal"
    );
  });

  it("metadata: a property change busts the column list → fresh on the NEXT read", async () => {
    // Create a DATABASE page (the only public path to a Database) — it seeds
    // the title property. The database id differs from the page id, so resolve
    // it from the page via Prisma.
    const dbPage = await app.request("POST", `/api/v1/workspaces/${workspaceId}/pages`, {
      headers: ownerHeaders,
      body: { title: "Cache Meta DB", type: "database" }
    });
    expect(dbPage.status).toBeLessThan(300);
    const pageObj = (await dbPage.json()) as { id: string };
    const db = await app.prisma.database.findFirst({ where: { pageId: pageObj.id } });
    expect(db).not.toBeNull();
    const dbId = db!.id;

    const listProps = async () => {
      const res = await app.request("GET", `/api/v1/databases/${dbId}/properties`, { headers: ownerHeaders });
      expect(res.status).toBe(200);
      return ((await res.json()) as { data: { name: string }[] }).data;
    };

    const before = await listProps(); // miss → fill metadata:…:properties
    const beforeCount = before.length;
    expect(beforeCount).toBeGreaterThan(0); // the seeded title property
    await listProps(); // hit

    // Add a property → database.updated (propertyId) → busts the metadata.
    const prop = await app.request("POST", `/api/v1/databases/${dbId}/properties`, {
      headers: ownerHeaders,
      body: { name: "Cached Col", type: "TEXT" }
    });
    expect(prop.status).toBeLessThan(300);

    // The NEXT listProperties is FRESH (includes the new property) — a cached
    // pre-create list would not. (The metadata cache is 60 s; this must be a
    // bust, not a TTL expiry.)
    await until(
      async () => (await listProps()).length === beforeCount + 1,
      5000,
      "the property list to include the new column"
    );
  });

  it("metadata: record-open and listProperties share ONE normalized entry (no id-less pollution)", async () => {
    // Regression (the "select works but the cell renders blank" report):
    // `getRecord` (record-open) and `listProperties` BOTH fill the SAME
    // `perf:metadata:{db}:properties` cache key, but only `listProperties`
    // normalized the id-less legacy select options. A record-open that
    // filled the cache FIRST left the RAW (id-less) entry in place for the
    // 60 s TTL, so the table view's `findOption` missed every canonical
    // `{ optionId }` value → blank select/status cells. Both loaders must now
    // store the SAME normalized shape.
    //
    // A fresh database page SEEDS a "Status" (SELECT) property whose options
    // are id-less (`DEFAULT_DB_PROPERTIES` — `apps/api/src/constants.ts`):
    // this IS the legacy state in the wild (no data migration needed).
    const dbPage = await app.request("POST", `/api/v1/workspaces/${workspaceId}/pages`, {
      headers: ownerHeaders,
      body: { title: "Cache Polluted DB", type: "database" }
    });
    expect(dbPage.status).toBeLessThan(300);
    const dbPageObj = (await dbPage.json()) as { id: string };
    const db = await app.prisma.database.findFirst({ where: { pageId: dbPageObj.id } });
    expect(db).not.toBeNull();
    const dbId = db!.id;
    const statusProp = await app.prisma.databaseProperty.findFirst({
      where: { databaseId: dbId, name: "Status" }
    });
    expect(statusProp).not.toBeNull();
    // Guard the premise: the seeded options really are id-less (otherwise the
    // test would pass vacuously).
    expect(
      ((statusProp!.config as { options?: Array<{ id?: string }> }).options ?? []).every(
        (o) => !o.id,
      ),
    ).toBe(true);

    interface PropDto {
      id: string;
      name: string;
      config?: { options?: Array<{ id?: string; name?: string }> };
    }
    const listProps = async (): Promise<PropDto[]> => {
      const res = await app.request("GET", `/api/v1/databases/${dbId}/properties`, { headers: ownerHeaders });
      expect(res.status).toBe(200);
      return ((await res.json()) as { data: PropDto[] }).data;
    };
    const statusOptions = (props: PropDto[]) => props.find((p) => p.id === statusProp.id)?.config?.options ?? [];

    // A record so the record-open has something to serve.
    const rec = await app.request("POST", `/api/v1/databases/${dbId}/records`, {
      headers: ownerHeaders,
      body: { title: "Polluted Row", properties: {}, links: {} }
    });
    expect(rec.status).toBeLessThan(300);
    // The create response IS the record DTO (top-level id).
    const recordId = ((await rec.json()) as { id: string }).id;

    // THE repro: a record-open FILLS the shared cache FIRST (record-create
    // does not bust the column-metadata cache — only `database.updated` does).
    // The record-open route is `GET /api/v1/records/:id` (the web's
    // `useDatabaseRecord` fetches exactly this URL).
    const open = await app.request("GET", `/api/v1/records/${recordId}`, { headers: ownerHeaders });
    expect(open.status).toBe(200);

    // The NEXT listProperties must be served from that entry — and it must
    // be NORMALIZED (deterministic option ids), not the raw id-less config.
    await until(
      async () => {
        const opts = statusOptions(await listProps());
        return (
          opts.length === 3 &&
          opts.map((o) => o.id).join(",") ===
            [optionIdForName("To Do"), optionIdForName("In Progress"), optionIdForName("Done")].join(",")
        );
      },
      5000,
      "the shared metadata entry to carry normalized option ids"
    );
  });
});