import { describe, it, expect, beforeEach } from "vitest";
import * as Y from "yjs";
import {
  InMemoryDocStore,
  RoomManager,
  makePeer,
  realClock,
  type Room,
  type RoomClock,
  type RoomManagerEvents,
  type DocStore,
  type PersistResult,
} from "../src/rooms.js";
import {
  blocksToYDoc,
  yDocToBlockList,
  FRAGMENT_NAME,
  type BlockLike,
  type BlockOperation,
} from "@factory/collaboration/codec";

// --- deterministic clock -----------------------------------------------------------

/**
 * A fake clock where `advance(ms)` fires every scheduled timer in
 * chronological order up to `t + ms`. Timers registered with
 * `setTimeout(fn, delay)` fire at `now + delay`.
 */
function driveClock(): { clock: RoomClock; run: (ms: number) => void; now: () => number } {
  let t = 0;
  const timers = new Map<number, { at: number; fn: () => void }>();
  let nextId = 1;
  const clock: RoomClock = {
    now: () => t,
    setTimeout: (fn: () => void, ms: number) => {
      const id = nextId++;
      timers.set(id, { at: t + Math.max(0, ms), fn });
      return id;
    },
    clearTimeout: (h: unknown) => {
      if (typeof h === "number") timers.delete(h);
    },
  };
  return {
    clock,
    now: () => t,
    run: (ms: number) => {
      const end = t + ms;
      while (true) {
        let earliestId = -1;
        let earliestAt = Infinity;
        for (const [id, entry] of timers) {
          if (entry.at <= end && entry.at < earliestAt) {
            earliestAt = entry.at;
            earliestId = id;
          }
        }
        if (earliestId === -1) break;
        const entry = timers.get(earliestId)!;
        t = entry.at;
        timers.delete(earliestId);
        entry.fn();
      }
      t = end;
    },
  };
}

// --- fixtures ----------------------------------------------------------------------

const PARA = (id: string, text: string, position: string, parentId: string | null = null): BlockLike => ({
  id,
  type: "paragraph",
  content: { text },
  parentId,
  position,
});

describe("InMemoryDocStore (the persistence seam, note 17)", () => {
  it("seeds + hydrates blocks + version (a copy — mutations don't leak back)", async () => {
    const store = new InMemoryDocStore();
    store.seed("p1", [PARA("a", "hello", "1")]);
    const hydrated = await store.hydrate("p1");
    expect(hydrated.blocks).toHaveLength(1);
    expect(hydrated.blocks![0].id).toBe("a");
    expect(hydrated.version).toBe(1); // seed = version 1
    // mutating the hydrated copy must not corrupt the store
    (hydrated.blocks![0] as BlockLike).content = { text: "changed" };
    const again = await store.hydrate("p1");
    expect(again.blocks![0].content).toEqual({ text: "hello" });
  });

  it("hydrates { blocks: null, version: 1 } for an unknown page", async () => {
    const store = new InMemoryDocStore();
    expect(await store.hydrate("missing")).toEqual({ blocks: null, version: 1 });
  });

  it("persist bumps the version from the CAS base", async () => {
    const store = new InMemoryDocStore();
    const r1 = await store.persist("p1", [PARA("a", "x", "1")], 1);
    expect(r1).toEqual({ version: 2 });
    const r2 = await store.persist("p1", [PARA("a", "y", "1")], 2);
    expect(r2).toEqual({ version: 3 });
    expect(store.versionOf("p1")).toBe(3);
  });

  it("persist with a stale base is a NO-OP and returns the external state (the CAS conflict)", async () => {
    const store = new InMemoryDocStore();
    store.seed("p1", [PARA("a", "base", "1")]);
    // A concurrent REST writer replaces the blocks + bumps the version.
    const externalVersion = store.writeExternal("p1", [PARA("a", "rest-wrote", "1")]);
    expect(externalVersion).toBe(2);
    // The room's flush, based on the version it hydrated at (1), conflicts.
    const r = await store.persist("p1", [PARA("a", "room-edit", "1")], 1);
    expect(r.version).toBe(2); // the store's CURRENT version
    expect(r.conflict).toBeDefined();
    expect(r.conflict!.externalBlocks.map((b) => b.content)).toEqual([{ text: "rest-wrote" }]);
    // The persist was a no-op — the store still holds the external writer's state.
    expect((await store.hydrate("p1")).blocks![0].content).toEqual({ text: "rest-wrote" });
  });

  it("ping reflects setPingOk (drives /readyz)", async () => {
    const store = new InMemoryDocStore();
    expect(await store.ping()).toBe(true);
    store.setPingOk(false);
    expect(await store.ping()).toBe(false);
  });
});

describe("RoomManager — cold hydration (R2)", () => {
  let store: InMemoryDocStore;
  let events: RoomManagerEvents;
  let manager: RoomManager;

  beforeEach(() => {
    store = new InMemoryDocStore();
    events = { onPersisted: () => {}, onRoomEvicted: () => {} };
    manager = new RoomManager({ store, clock: realClock() }, events);
  });

  it("hydrates a cold page room from the store before publishing it", async () => {
    store.seed("p1", [PARA("a", "seeded", "1"), PARA("b", "two", "2")]);
    const room = await manager.getOrCreate("page:p1", "page", "p1");
    const blocks = yDocToBlockList(room.doc);
    expect(blocks.map((b) => b.id)).toEqual(["a", "b"]);
    expect(blocks[0].content).toEqual({ text: "seeded" });
  });

  it("a warm room is returned as-is (no re-hydration)", async () => {
    store.seed("p1", [PARA("a", "seeded", "1")]);
    const r1 = await manager.getOrCreate("page:p1", "page", "p1");
    const r2 = await manager.getOrCreate("page:p1", "page", "p1");
    expect(r1).toBe(r2);
  });

  it("get returns the live room; remove drops it from the map", async () => {
    const room = await manager.getOrCreate("page:p1", "page", "p1");
    expect(manager.get("page:p1")).toBe(room);
    manager.remove("page:p1");
    expect(manager.get("page:p1")).toBeUndefined();
  });
});

describe("Room — updates, dirty state, debounced flush (R2)", () => {
  let drive: ReturnType<typeof driveClock>;
  let store: InMemoryDocStore;
  let evicted: string[];
  let persisted: Array<{ roomKey: string; version: number }>;
  let manager: RoomManager;

  beforeEach(() => {
    drive = driveClock();
    store = new InMemoryDocStore();
    evicted = [];
    persisted = [];
    manager = new RoomManager(
      { store, clock: drive.clock, persistDebounceMs: 1000, persistHardCapMs: 10_000 },
      {
        onPersisted: (roomKey, _resourceId, version) => persisted.push({ roomKey, version }),
        onRoomEvicted: (roomKey) => evicted.push(roomKey),
      },
    );
  });

  /** Simulate a client's update: build it in a second doc (via the codec)
   *  and apply the encoded update into the room's authoritative doc — the
   *  same path the server takes for an inbound 0x01 frame. */
  function clientUpdate(room: Awaited<ReturnType<RoomManager["getOrCreate"]>>): void {
    const clientDoc = blocksToYDoc([PARA("newblock", "from client", "9")], { pageId: "p1" });
    const update = Y.encodeStateAsUpdate(clientDoc);
    clientDoc.destroy();
    expect(room.applyUpdate(update)).toBe(true); // applied now (not hydrating)
  }

  it("applying an update marks the room dirty", async () => {
    const room = await manager.getOrCreate("page:p1", "page", "p1");
    expect(room.isDirty).toBe(false);
    clientUpdate(room);
    expect(room.isDirty).toBe(true);
  });

  it("persists after the debounce (1 s) — the store sees the new block", async () => {
    const room = await manager.getOrCreate("page:p1", "page", "p1");
    clientUpdate(room);
    expect(persisted).toHaveLength(0); // not yet
    drive.run(1000); // fire the debounce timer
    await new Promise((r) => setTimeout(r, 0)); // the flush is async (store round-trip)
    expect(persisted).toHaveLength(1);
    const stored = await store.hydrate("p1");
    expect(stored.blocks?.some((b) => b.id === "newblock")).toBe(true);
    expect(room.isDirty).toBe(false);
  });

  it("coalesces repeated updates into one flush", async () => {
    const room = await manager.getOrCreate("page:p1", "page", "p1");
    clientUpdate(room);
    clientUpdate(room);
    clientUpdate(room);
    drive.run(1000);
    await new Promise((r) => setTimeout(r, 0));
    expect(persisted).toHaveLength(1); // one flush, not three
  });

  it("stateDelta with no state vector returns the full doc", async () => {
    store.seed("p1", [PARA("a", "full", "1")]);
    const room = await manager.getOrCreate("page:p1", "page", "p1");
    const full = room.stateDelta(null);
    expect(full.byteLength).toBeGreaterThan(0);
    // applying the full update to a fresh doc reproduces the blocks
    const fresh = new Y.Doc();
    Y.applyUpdate(fresh, full, "tester");
    expect(yDocToBlockList(fresh).map((b) => b.id)).toEqual(["a"]);
    fresh.destroy();
  });

  it("stateDelta with a client state vector returns only the missing update (R5 resume)", async () => {
    store.seed("p1", [PARA("a", "seed", "1")]);
    const room = await manager.getOrCreate("page:p1", "page", "p1");
    // The client joins, receives the full doc, and applies it locally — its
    // state vector now matches the room's current state (the real join path).
    const clientDoc = new Y.Doc();
    Y.applyUpdate(clientDoc, room.stateDelta(null), "server");
    const clientVector = Y.encodeStateVector(clientDoc);
    // The server then receives one more update (another client).
    clientUpdate(room);
    const delta = room.stateDelta(clientVector);
    const full = room.stateDelta(null);
    // The delta is non-empty (the new block) and smaller than the full doc.
    expect(delta.byteLength).toBeGreaterThan(0);
    expect(delta.byteLength).toBeLessThan(full.byteLength);
    // Applying the delta to the client doc converges it to the server state.
    Y.applyUpdate(clientDoc, delta, "server");
    expect(yDocToBlockList(clientDoc).some((b) => b.id === "newblock")).toBe(true);
    clientDoc.destroy();
  });

  it("evict flushes a dirty room and reports the eviction", async () => {
    const room = await manager.getOrCreate("page:p1", "page", "p1");
    clientUpdate(room);
    await room.evict();
    expect(evicted).toEqual(["page:p1"]);
    expect(persisted).toHaveLength(1); // flushed before eviction
    const stored = await store.hydrate("p1");
    expect(stored.blocks?.some((b) => b.id === "newblock")).toBe(true);
  });

  it("evict of a clean room does not persist", async () => {
    const room = await manager.getOrCreate("page:p1", "page", "p1");
    await room.evict();
    expect(evicted).toEqual(["page:p1"]);
    expect(persisted).toHaveLength(0);
  });

  it("empty-room GC evicts after the warm lifetime (R2)", async () => {
    const gcManager = new RoomManager(
      { store, clock: drive.clock, roomGcMs: 100 },
      { onPersisted: () => {}, onRoomEvicted: (k) => evicted.push(k) },
    );
    const room = await gcManager.getOrCreate("page:p1", "page", "p1");
    const peer = makePeer("c1", "u1", true, "Ada");
    room.peers.set("c1", peer);
    room.notePeersChanged(); // has a peer → GC armed off
    room.peers.delete("c1");
    room.notePeersChanged(); // empty → GC armed for 100 ms
    drive.run(99);
    expect(evicted).toHaveLength(0);
    drive.run(1);
    expect(evicted).toEqual(["page:p1"]);
  });

  it("a peer re-joining cancels the pending GC", async () => {
    const gcManager = new RoomManager(
      { store, clock: drive.clock, roomGcMs: 100 },
      { onPersisted: () => {}, onRoomEvicted: (k) => evicted.push(k) },
    );
    const room = await gcManager.getOrCreate("page:p1", "page", "p1");
    room.notePeersChanged(); // empty → GC armed
    room.peers.set("c1", makePeer("c1", "u1", true));
    room.notePeersChanged(); // re-join → disarmed
    drive.run(1000);
    expect(evicted).toHaveLength(0);
    expect(room.isEvicted).toBe(false);
  });

  it("drainAll flushes every dirty room and evicts all (R6 SIGTERM)", async () => {
    const r1 = await manager.getOrCreate("page:p1", "page", "p1");
    void (await manager.getOrCreate("page:p2", "page", "p2"));
    clientUpdate(r1);
    const result = await manager.drainAll(10_000);
    expect(result).toEqual({ flushed: 1, evicted: 2 });
    expect(evicted.sort()).toEqual(["page:p1", "page:p2"]);
    expect(persisted).toHaveLength(1);
    expect(manager.rooms()).toHaveLength(0);
  });
});

describe("Room — flush-rebase (TASK-0092 R2, the 3-way merge)", () => {
  let drive: ReturnType<typeof driveClock>;
  let store: InMemoryDocStore;
  let events: {
    persisted: Array<{ version: number; blockDelta: BlockOperation[] }>;
    rebased: number;
  };
  let manager: RoomManager;

  beforeEach(() => {
    drive = driveClock();
    store = new InMemoryDocStore();
    events = { persisted: [], rebased: 0 };
    manager = new RoomManager(
      { store, clock: drive.clock, persistDebounceMs: 1000, persistHardCapMs: 10_000 },
      {
        onPersisted: (_roomKey, _resourceId, version, _logSeq, blockDelta) =>
          events.persisted.push({ version, blockDelta }),
        onRebased: () => {
          events.rebased += 1;
        },
        onRoomEvicted: () => {},
      },
    );
  });

  /**
   * Edit a block's text content in the room's authoritative doc, the way a
   * client's 0x01 update would (a real Yjs text edit, not a codec round-
   * trip): finds the block element by its `id` attr and appends to its
   * `text` slot. Also marks the room dirty (the server's applyUpdate path
   * does this; the test bypasses the frame decode, so it calls the room's
   * dirty bookkeeping via a subsequent flush — the assertion is on the
   * persisted content, which only lands if the edit is in the doc).
   */
  function editBlockText(room: Room, blockId: string, append: string): void {
    const fragment = room.doc.getXmlFragment(FRAGMENT_NAME);
    const walk = (els: Y.XmlElement[]): Y.XmlElement | null => {
      for (const el of els) {
        if (el.getAttribute("id") === blockId) return el;
        const nested = el.toArray().filter((c): c is Y.XmlElement => c instanceof Y.XmlElement);
        const found = walk(nested);
        if (found !== null) return found;
      }
      return null;
    };
    const el = walk(fragment.toArray() as Y.XmlElement[]);
    expect(el, `block "${blockId}" found in the doc`).not.toBeNull();
    const text = el
      .toArray()
      .find((c): c is Y.XmlText => c instanceof Y.XmlText && c.getAttribute("name") === "text");
    expect(text, "a paragraph has a `text` slot").toBeDefined();
    text.insert(text.length, append);
  }

  it("concurrent REST write + room edit on DIFFERENT blocks → both survive (no clobber), next flush is a no-op", async () => {
    store.seed("p1", [PARA("a", "base a", "1"), PARA("b", "base b", "2")]);
    const room = await manager.getOrCreate("page:p1", "page", "p1");
    // The room's client edits block "a".
    editBlockText(room, "a", "+room edit on a");
    // A concurrent REST write (bypassing the room) edits block "b" and bumps
    // the version — exactly the race the flush's CAS detects.
    store.writeExternal(
      "p1",
      [PARA("a", "base a", "1"), PARA("b", "base b+rest edit on b", "2")],
    );
    await room.flush();

    // The rebase ran once.
    expect(events.rebased).toBe(1);
    // The FINAL store state has BOTH edits (the acceptance criterion).
    const final = (await store.hydrate("p1")).blocks!;
    const a = final.find((b) => b.id === "a")!;
    const b = final.find((b) => b.id === "b")!;
    expect(a.content).toEqual({ text: "base a+room edit on a" });
    expect(b.content).toEqual({ text: "base b+rest edit on b" });

    // The page version advanced past the external writer (the persist
    // retried at the external version and won).
    expect(store.versionOf("p1")).toBe(3); // seed=1, rest=2, flush=3
    // The persisted blockDelta is non-trivial (real changes landed).
    const last = events.persisted[events.persisted.length - 1]!;
    expect(last.blockDelta.length).toBeGreaterThan(0);

    // The rebase converged: the doc now matches the store exactly, so a
    // further persist of the same blocks at the current version is content-
    // identical (the idempotency assert — no clobber, nothing to re-merge).
    const blocksNow = yDocToBlockList(room.doc);
    const noop = await store.persist("p1", blocksNow, store.versionOf("p1"));
    expect(noop.conflict).toBeUndefined();
    expect(noop.version).toBe(4); // a version bump (the store's CAS)
    const after = (await store.hydrate("p1")).blocks!;
    expect(after).toEqual(blocksNow); // same content — the rebase converged
  });

  it("BOTH sides edit the SAME block → the FLUSH (room) wins (the spec's conflict rule)", async () => {
    store.seed("p1", [PARA("a", "base a", "1"), PARA("b", "keep b", "2")]);
    const room = await manager.getOrCreate("page:p1", "page", "p1");
    // The room's client edits block "a".
    editBlockText(room, "a", "+room edit WINS");
    // A concurrent REST write edits the SAME block "a" (and touches "b").
    store.writeExternal(
      "p1",
      [PARA("a", "base a+rest edit loses", "1"), PARA("b", "keep b+rest", "2")],
    );
    await room.flush();
    const final = (await store.hydrate("p1")).blocks!;
    const a = final.find((b) => b.id === "a")!;
    const b = final.find((b) => b.id === "b")!;
    // Flush wins on the shared block; REST's non-shared edit survives.
    expect(a.content).toEqual({ text: "base a+room edit WINS" });
    expect(b.content).toEqual({ text: "keep b+rest" });
  });

  it("one rebase absorbs MULTIPLE concurrent external writes (all survive, single persist)", async () => {
    store.seed("p1", [PARA("a", "base", "1")]);
    const room = await manager.getOrCreate("page:p1", "page", "p1");
    editBlockText(room, "a", "+room");
    // Two external writers land before the flush — the rebase must absorb
    // BOTH (each is a real change vs the room's base) in one pass.
    store.writeExternal("p1", [PARA("a", "base", "1"), PARA("e1", "external 1", "3")]);
    store.writeExternal("p1", [PARA("a", "base", "1"), PARA("e1", "external 1", "3"), PARA("e2", "external 2", "4")]);
    await room.flush();
    expect(events.rebased).toBe(1);
    expect(events.persisted).toHaveLength(1);
    const final = (await store.hydrate("p1")).blocks!;
    expect(final.map((b) => b.id).sort()).toEqual(["a", "e1", "e2"]);
    expect(final.find((b) => b.id === "a")!.content).toEqual({ text: "base+room" });
    expect(final.find((b) => b.id === "e1")!.content).toEqual({ text: "external 1" });
    expect(final.find((b) => b.id === "e2")!.content).toEqual({ text: "external 2" });
  });

  it("a version bump DURING the retry → stays dirty (no clobber), converges on the next flush", async () => {
    // A purpose-built store that simulates a SECOND concurrent writer firing
    // between the conflicted persist and the rebase retry — the only way to
    // reach the second-conflict branch synchronously (a real race is
    // asynchronous; this deterministically exercises the single-retry limit).
    const base = new InMemoryDocStore();
    base.seed("p1", [PARA("a", "base", "1")]);
    let persistCalls = 0;
    const raceStore: DocStore = {
      hydrate: () => base.hydrate("p1"),
      ping: () => base.ping(),
      persist: async (pageId, blocks, bVersion): Promise<PersistResult> => {
        persistCalls += 1;
        const r = await base.persist(pageId, blocks, bVersion);
        if (r.conflict !== undefined && persistCalls === 1) {
          // The second external writer lands right after the first conflict
          // is reported — before the rebase retry re-checks the CAS.
          base.writeExternal("p1", [
            PARA("a", "base", "1"),
            PARA("e1", "external 1", "3"),
            PARA("e2", "external 2", "4"),
          ]);
        }
        return r;
      },
    };
    const raceManager = new RoomManager(
      { store: raceStore, clock: drive.clock, persistDebounceMs: 1000, persistHardCapMs: 10_000 },
      {
        onPersisted: (_k, _r, _v, _s, delta) => events.persisted.push({ version: _v, blockDelta: delta }),
        onRebased: () => {
          events.rebased += 1;
        },
        onRoomEvicted: () => {},
      },
    );
    const room = await raceManager.getOrCreate("page:p1", "page", "p1");
    editBlockText(room, "a", "+room");
    // External write #1 (conflicts the first persist).
    base.writeExternal("p1", [PARA("a", "base", "1"), PARA("e1", "external 1", "3")]);
    await room.flush();
    // The single retry was used up (the second writer landed during it):
    // no persist landed, the room keeps its user edits, no clobber.
    expect(events.persisted).toHaveLength(0);
    expect(base.versionOf("p1")).toBe(3); // e1 then e2 → v2, v3
    // The next flush converges (absorbs both externals + the room's edit).
    await room.flush();
    expect(events.persisted).toHaveLength(1);
    const final = (await base.hydrate("p1")).blocks!;
    expect(final.map((b) => b.id).sort()).toEqual(["a", "e1", "e2"]);
    expect(final.find((b) => b.id === "a")!.content).toEqual({ text: "base+room" });
  });

  it("the doc's origin for the rebase update is 'realtime-flush-rebase' (observable)", async () => {
    store.seed("p1", [PARA("a", "base", "1")]);
    const room = await manager.getOrCreate("page:p1", "page", "p1");
    const origins: string[] = [];
    room.doc.on("update", (_u: Uint8Array, origin: string | null) => {
      if (typeof origin === "string") origins.push(origin);
    });
    store.writeExternal("p1", [PARA("a", "base", "1"), PARA("x", "ext", "2")]);
    await room.flush();
    expect(origins).toContain("realtime-flush-rebase");
  });
});

describe("makePeer (roster entry)", () => {
  it("builds a peer with a deterministic per-user color + the provided name", () => {
    const a = makePeer("c1", "u1", true, "Ada");
    const b = makePeer("c2", "u1", false, "Ada");
    expect(a.userId).toBe("u1");
    expect(a.canEdit).toBe(true);
    expect(b.canEdit).toBe(false);
    expect(a.roster.user.id).toBe("u1");
    expect(a.roster.user.name).toBe("Ada");
    // same user → same color (the shared 20-color palette)
    expect(a.roster.user.color).toBe(b.roster.user.color);
    expect(typeof a.roster.user.color).toBe("string");
  });

  it("defaults the name when none is provided", () => {
    const p = makePeer("c1", "u9", true);
    expect(p.roster.user.name.length).toBeGreaterThan(0);
  });
});