/**
 * Realtime integration (TASK-0091, acceptance criterion 4): the resume
 * protocol — the load-bearing test.
 *
 * Wire behavior under test (server.ts `handleSyncRequest`, R5):
 *  - reconnect with `lastSeq` covered by the update log → the server
 *    replays the logged updates as binary `update` frames, acks the last
 *    seq, THEN always finishes with the state-vector delta (idempotent
 *    convergence — the log is a speed optimization, the state vector is
 *    the correctness guarantee).
 *  - log gap (truncated log) → state-vector path only.
 *  - invariant: both paths converge to the byte-identical doc.
 *
 * The "server doc" is observed through a fresh client's full sync
 * (sync.request without a state vector → the complete doc), never
 * through the in-memory store (which only sees the debounced flush).
 */
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import * as Y from "yjs";
import {
  buildRealtimeFixture,
  connectClient,
  makeTextUpdate,
  type RealtimeFixture,
  type RtClient,
} from "./rt-helpers";

/** Apply every Yjs update to `doc` (sync.state chunks + binary replays). */
function applyAll(doc: Y.Doc, c: RtClient): void {
  for (const f of c.textFrames()) {
    if (f.type === "sync.state" && typeof f.chunk === "string" && f.chunk.length > 0) {
      Y.applyUpdate(doc, Buffer.from(f.chunk, "base64"), "it");
    }
  }
  for (const b of c.binaryFrames()) {
    if (b.kind === "update") Y.applyUpdate(doc, b.bytes, "it");
  }
}

/** A fresh client that full-syncs the room (the authoritative server view). */
async function serverDoc(fx: RealtimeFixture, room = "page:p-it"): Promise<Y.Doc> {
  const c = await connectClient(fx, { room });
  try {
    await c.nextFrame("hello");
    c.sendText({ type: "sync.request" });
    await c.nextFrameWhere((f) => f.type === "sync.state" && f.done === true, 8_000);
    const doc = new Y.Doc();
    applyAll(doc, c);
    return doc;
  } finally {
    c.close();
  }
}

/**
 * Join a page room, full-sync, send `n` one-char updates (awaiting each
 * ack so the log seqs are settled), return the client + the last acked
 * seq + the local doc.
 */
async function editSession(
  fx: RealtimeFixture,
  room: string,
  n: number,
  prefix: string,
  localDoc: Y.Doc,
): Promise<{ client: RtClient; lastSeq: number }> {
  const client = await connectClient(fx, { room });
  await client.nextFrame("hello");
  client.sendText({ type: "sync.request" });
  await client.nextFrameWhere((f) => f.type === "sync.state" && f.done === true, 8_000);
  applyAll(localDoc, client);
  let lastSeq = 0;
  for (let i = 0; i < n; i += 1) {
    const bytes = makeTextUpdate(localDoc, `${prefix}${i}`);
    client.sendBinary(1, bytes);
    const ack = await client.nextFrame("ack");
    lastSeq = Number(ack.seq);
  }
  return { client, lastSeq };
}

describe("resume protocol (R5)", () => {
  let fx: RealtimeFixture;
  beforeAll(async () => {
    fx = await buildRealtimeFixture({
      pages: new Set(["p-it", "p-seed"]),
      workspaces: { "p-it": "w1", "p-seed": "w1" },
      levels: { "page:p-it": "edit", "page:p-seed": "edit" },
    });
  });
  afterAll(async () => {
    await fx.stop();
  });

  it("a client that was briefly offline resumes from the update log (exactly the missed frames) and converges", async () => {
    // Client A: joins, receives the full doc, applies 20 updates, leaves.
    const docA = new Y.Doc();
    const a = await editSession(fx, "page:p-it", 20, "A", docA);
    a.client.close();
    const aLastSeq = a.lastSeq;
    expect(aLastSeq).toBe(20); // the log is per-room, this is the only writer so far

    // Client B: joins, applies 5 updates (the "server applies 5 more").
    const docB = new Y.Doc();
    const b = await editSession(fx, "page:p-it", 5, "B", docB);
    const serverSeq = b.lastSeq;
    expect(serverSeq).toBe(25);
    b.client.close();

    // The reference: what the server's doc looks like right now.
    const ref = await serverDoc(fx);
    const refVec = Y.encodeStateVector(ref);

    // A reconnects with its last-ack seq → it must receive EXACTLY the 5
    // logged updates (binary frames before the state-delta sync.state),
    // the ack of the last logged seq, then the (idempotent) state delta.
    const docA2 = new Y.Doc();
    // Re-apply A's local state (its editor kept the doc in memory).
    const docA3 = new Y.Doc();
    Y.applyUpdate(docA3, Y.encodeStateAsUpdate(docA), "it");
    void docA2;

    const a2 = await connectClient(fx, { room: "page:p-it" });
    try {
      await a2.nextFrame("hello");
      a2.sendText({ type: "sync.request", lastSeq: aLastSeq });
      await a2.nextFrameWhere((f) => f.type === "sync.state" && f.done === true, 8_000);

      // Exactly 5 binary update frames were replayed from the log
      // (the state delta arrives as sync.state TEXT frames, not binary).
      const replayed = a2.binaryFrames().filter((b2) => b2.kind === "update");
      expect(replayed).toHaveLength(5);
      // The ack covers the last logged seq.
      const acks = a2.textFrames().filter((f) => f.type === "ack");
      expect(acks.length).toBeGreaterThanOrEqual(1);
      expect(Number(acks[acks.length - 1]!.seq)).toBe(serverSeq);

      // Converge the local doc: replay + the state delta.
      applyAll(docA3, a2);
      expect(Y.encodeStateVector(docA3)).toEqual(refVec); // byte-identical
      const txt = (d: Y.Doc) => d.getText("yText").toString();
      expect(txt(docA3)).toBe(txt(ref));
    } finally {
      a2.close();
      ref.destroy();
      docA3.destroy();
    }
  });

  it("with the log truncated (a gap), the state-vector path returns the same final doc", async () => {
    // Same room, current log maxSeq is 25 — truncate it (the "Redis log
    // killed" scenario; InMemoryUpdateLog.clear is the seam).
    void (await fx.updateLog.maxSeq("page:p-it"));
    await fx.updateLog.clear("page:p-it");

    // A fresh client reconnecting with a lastSeq the log no longer covers
    // → no binary replay, the state-vector delta alone.
    const doc = new Y.Doc();
    const c = await connectClient(fx, { room: "page:p-it" });
    try {
      await c.nextFrame("hello");
      c.sendText({ type: "sync.request", lastSeq: 25 });
      await c.nextFrameWhere((f) => f.type === "sync.state" && f.done === true, 8_000);
      expect(c.binaryFrames().filter((b2) => b2.kind === "update")).toHaveLength(0);
      applyAll(doc, c);

      const ref = await serverDoc(fx);
      expect(Y.encodeStateVector(doc)).toEqual(Y.encodeStateVector(ref));
      expect(doc.getText("yText").toString()).toBe(ref.getText("yText").toString());
      ref.destroy();
    } finally {
      c.close();
      doc.destroy();
    }
  });

  it("invariant (property): 20 random disconnect/reconnect windows always converge", async () => {
    // Deterministic pseudo-random (mulberry32) — the property is over
    // random window shapes, not the RNG itself.
    let seed = 0x20260926;
    const rnd = (): number => {
      seed |= 0;
      seed = (seed + 0x6d2b79f5) | 0;
      let t = Math.imul(seed ^ (seed >>> 15), 1 | seed);
      t = (t + Math.imul(t ^ (t >>> 7), 61 | t)) ^ t;
      return ((t ^ (t >>> 14)) >>> 0) / 4294967296;
    };

    const room = "page:p-seed";
    const local = new Y.Doc();
    let lastSeq = 0;
    let useLogResume = true;

    for (let i = 0; i < 20; i += 1) {
      // Window shape: 1–4 local edits before the disconnect.
      const client = await connectClient(fx, { room });
      const edits = 1 + Math.floor(rnd() * 4);
      for (let e = 0; e < edits; e += 1) {
        const bytes = makeTextUpdate(local, `w${i}-${e}`);
        client.sendBinary(1, bytes);
        const ack = await client.nextFrame("ack");
        lastSeq = Number(ack.seq);
      }
      // The client drops (no clean close — a network partition).
      client.terminate();
      await new Promise((r) => setTimeout(r, 30));

      // Meanwhile the room advances 1–3 updates from another peer.
      const other = await connectClient(fx, { room });
      const remoteEdits = 1 + Math.floor(rnd() * 3);
      const otherDoc = new Y.Doc();
      for (let e = 0; e < remoteEdits; e += 1) {
        // otherDoc is empty — its updates are fresh content, still
        // valid CRDT merges.
        const bytes = makeTextUpdate(otherDoc, `r${i}-${e}`);
        other.sendBinary(1, bytes);
        await other.nextFrame("ack");
      }
      other.close();

      // Reconnect: 50/50 log-resume vs state-vector resume.
      useLogResume = rnd() > 0.5;
      const back = await connectClient(fx, { room });
      back.sendText(useLogResume ? { type: "sync.request", lastSeq } : { type: "sync.request" });
      await back.nextFrameWhere((f) => f.type === "sync.state" && f.done === true, 8_000);
      applyAll(local, back);

      const ref = await serverDoc(fx, room);
      expect(Y.encodeStateVector(local), `window ${i} diverged`).toEqual(
        Y.encodeStateVector(ref),
      );
      ref.destroy();
      back.close();
    }
    local.destroy();
  });

  it("a resume for a different room's log is rejected to the state-vector path (tenant scoping)", async () => {
    // The log is keyed per room: lastSeq from room X on room Y is a gap
    // (the security requirement — a client can only ever resume ITS room).
    const doc = new Y.Doc();
    const c = await connectClient(fx, { room: "page:p-it" });
    try {
      await c.nextFrame("hello");
      // 999 does not exist in p-it's log (it's at ~25) → gap → SV path.
      c.sendText({ type: "sync.request", lastSeq: 999 });
      await c.nextFrameWhere((f) => f.type === "sync.state" && f.done === true, 8_000);
      expect(c.binaryFrames().filter((b2) => b2.kind === "update")).toHaveLength(0);
      applyAll(doc, c);
      const ref = await serverDoc(fx);
      expect(Y.encodeStateVector(doc)).toEqual(Y.encodeStateVector(ref));
      ref.destroy();
    } finally {
      c.close();
      doc.destroy();
    }
  });
});