/**
 * Realtime integration (TASK-0092, acceptance criteria 2 + 3): the
 * flush-rebase (a concurrent REST write vs a dirty room → both survive, no
 * clobber, the next flush is a no-op) and the persistence diff minimality
 * (the flush's `page.updated` `blockDelta` is the MINIMAL op set; unchanged
 * blocks are untouched). Reuses TASK-0091's harness (the real server + the
 * in-memory store, driven by a real `ws` client over real block updates).
 */
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import * as Y from "yjs";
import {
  buildRealtimeFixture,
  connectClient,
  type RealtimeFixture,
  type RtClient,
} from "./rt-helpers";
import {
  blocksToYDoc,
  yDocToBlockList,
  FRAGMENT_NAME,
  type BlockLike,
} from "@factory/collaboration/codec";

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

/** Apply every Yjs update (sync.state chunks + binary) to `doc`. */
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");
  }
}

/**
 * Join + full-sync a page room into `localDoc` (the client now shares the
 * server's Yjs item ids). Returns the client + the local doc.
 */
async function joinAndSync(fx: RealtimeFixture, room: string): Promise<{ client: RtClient; doc: Y.Doc }> {
  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);
  const doc = new Y.Doc();
  applyAll(doc, client);
  return { client, doc };
}

/** Find a block's `text` slot in a doc by its `id` attr. */
function textSlotOf(doc: Y.Doc, blockId: string): Y.XmlText {
  const fragment = 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[]);
  if (el === null) throw new Error(`block "${blockId}" not found in the doc`);
  const text = el.toArray().find((c): c is Y.XmlText => c instanceof Y.XmlText && c.getAttribute("name") === "text");
  if (text === undefined) throw new Error(`block "${blockId}" has no text slot`);
  return text;
}

/** Edit a block's text (a real Yjs text edit on the shared item) + return the
 *  delta bytes to send as a 0x01 frame. */
function blockTextUpdate(doc: Y.Doc, blockId: string, append: string): Uint8Array {
  const before = Y.encodeStateAsUpdate(doc);
  const slot = textSlotOf(doc, blockId);
  slot.insert(slot.length, append);
  return Y.encodeStateAsUpdate(doc, before);
}

/** Edit a block + await the server's ack (the update is applied + logged). */
async function editBlock(fx: RealtimeFixture, client: RtClient, doc: Y.Doc, blockId: string, append: string): Promise<void> {
  void fx;
  client.sendBinary(1, blockTextUpdate(doc, blockId, append));
  await client.nextFrame("ack");
}

describe("flush-rebase (TASK-0092 R2, acceptance criterion 2)", () => {
  let fx: RealtimeFixture;
  beforeAll(async () => {
    fx = await buildRealtimeFixture({
      pages: new Set(["p-rb"]),
      workspaces: { "p-rb": "w1" },
      levels: { "page:p-rb": "edit" },
    });
  });
  afterAll(async () => {
    await fx.stop();
  });

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

      // Wait for the flush (1 s debounce → the rebase + the retry persist).
      const persisted = await client.nextFrame("doc.persisted", 8_000);
      expect(String(persisted.pageId)).toBe("p-rb");

      // The FINAL store state has BOTH edits (the acceptance criterion).
      const final = (await fx.store.hydrate("p-rb")).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" }); // the room's edit
      expect(b.content).toEqual({ text: "base b+rest edit" }); // the REST edit

      // The persist retried at the external version and won (v3).
      expect(Number(persisted.version)).toBe(3);
      // A rejoiner's doc == the store (the rebase converged the CRDT).
      const ref = await joinAndSync(fx, "page:p-rb");
      ref.doc.destroy();
      expect(ref.client.textFrames().some((f) => f.type === "error")).toBe(false);

      // The flush emitted `page.updated` with the additive `blockDelta`
      // (both real changes vs the base → 2 updates).
      const evts = fx.events.pageUpdated("p-rb");
      expect(evts.length).toBeGreaterThanOrEqual(1);
      const delta = evts[evts.length - 1]!.payload.blockDelta;
      expect(delta).toMatchObject({ inserted: 0, updated: 2, moved: 0, deleted: 0 });

      // The room is now CLEAN + in sync — the NEXT flush is a no-op (the
      // idempotency assert): a clean persist emits a ZERO blockDelta + the
      // store content is unchanged.
      const room = fx.server.manager.get("page:p-rb")!;
      expect(room.isDirty).toBe(false);
      await room.flush();
      const noop = fx.events.pageUpdated("p-rb")[fx.events.pageUpdated("p-rb").length - 1]!;
      expect(noop.payload.blockDelta).toEqual({ inserted: 0, updated: 0, moved: 0, deleted: 0 });
      const unchanged = (await fx.store.hydrate("p-rb")).blocks!;
      expect(unchanged.find((b) => b.id === "a")!.content).toEqual({ text: "base a+room edit" });
      expect(unchanged.find((b) => b.id === "b")!.content).toEqual({ text: "base b+rest edit" });
    } finally {
      client.close();
      doc.destroy();
    }
  });
});

describe("persistence diff minimality (TASK-0092 R3, acceptance criterion 3)", () => {
  let fx: RealtimeFixture;
  beforeAll(async () => {
    fx = await buildRealtimeFixture({
      pages: new Set(["p-dm"]),
      workspaces: { "p-dm": "w1" },
      levels: { "page:p-dm": "edit" },
    });
  });
  afterAll(async () => {
    await fx.stop();
  });

  it("editing ONE block → the blockDelta is exactly that 1 update; the other blocks are untouched", async () => {
    fx.store.seed("p-dm", [PARA("a", "alpha", "1"), PARA("b", "beta", "2"), PARA("c", "gamma", "3")]);
    const { client, doc } = await joinAndSync(fx, "page:p-dm");
    try {
      // Edit ONLY block "a".
      await editBlock(fx, client, doc, "a", " edited");
      await client.nextFrame("doc.persisted", 8_000);

      // The blockDelta is MINIMAL: exactly 1 update (block "a"), nothing else.
      const evts = fx.events.pageUpdated("p-dm");
      const delta = evts[evts.length - 1]!.payload.blockDelta;
      expect(delta).toEqual({ inserted: 0, updated: 1, moved: 0, deleted: 0 });
      // The emitted ops are exactly the one changed block.
      const ops = evts[evts.length - 1]!.payload.ops as Array<{ op: string; id?: string }>;
      expect(ops).toHaveLength(1);
      expect(ops[0]).toMatchObject({ op: "update", id: "a" });

      // The store: block "a" changed; "b" and "c" UNCHANGED (same content).
      const final = (await fx.store.hydrate("p-dm")).blocks!;
      expect(final.find((b) => b.id === "a")!.content).toEqual({ text: "alpha edited" });
      expect(final.find((b) => b.id === "b")!.content).toEqual({ text: "beta" });
      expect(final.find((b) => b.id === "c")!.content).toEqual({ text: "gamma" });
      // The room doc agrees (the CRDT state == the persisted state).
      expect(yDocToBlockList(doc).map((b) => b.id).sort()).toEqual(["a", "b", "c"]);
      void blocksToYDoc; // (imported for the round-trip type check)
    } finally {
      client.close();
      doc.destroy();
    }
  });
});