/**
 * Realtime integration (TASK-0092 slice 6c — the (b) CONVERGENCE loop,
 * Docker-free): the full producer chain with NO fakes of the decision
 * path. A room client edits a block → the room's REAL flush emits
 * `page.updated` on the server's event sink → a loopback "Redis stream"
 * (the in-memory stand-in for the API's `factory:events` XADD, recorded
 * via `eventsTap`) → the REAL `createPageUpdatedSource` poll loop (fake
 * `XREAD` honoring the raw keyword form + BLOCK pacing) → the REAL
 * `blockChangedForPageUpdated` derivation → the REAL
 * `publishBlockChanged` → the `block:changed` refetch-hint arrives on
 * TWO independent roomless control sockets (the two-tab (b) story: every
 * tab's overlay re-reads the source of truth at rest after a remote
 * change).
 *
 * What is faked and why (recorded): the `XREAD` transport (Docker-gated —
 * a real Redis) and the API's REST/flush EMIT side (the server's own
 * `EventSink` is the EMIT seam the API mirrors). Everything from the event
 * on (parsing, derivation, fan-out, socket delivery) is the production
 * code path.
 */
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import * as Y from "yjs";
import {
  buildRealtimeFixture,
  connectClient,
  mintToken,
  type RealtimeFixture,
  type RtClient,
} from "./rt-helpers";
import {
  createPageUpdatedSource,
  type PageUpdatedStreamClient,
  type PageUpdatedSource,
} from "@factory/realtime/src/page-updated-source";
import { FRAGMENT_NAME, type BlockLike } from "@factory/collaboration/codec";
import type { DomainEvent } from "@factory/shared";

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

/** A loopback stream: the server's emitted events (via `eventsTap`) are the
 *  XADD side; `xread` is the (fake) XREAD side honoring the raw keyword
 *  form the real source sends + BLOCK pacing (sleep `blockMs` when there
 *  is nothing new — the same hot-loop guard the unit-test fake uses). */
function makeLoopbackStream(blockMs = 5) {
  const entries: Array<[string, string[]]> = [];
  let marker = -1; // the index "now" resolves to once the source starts ($ = tail)
  let counter = 0;
  const client: PageUpdatedStreamClient = {
    async xread(...args: (string | number)[]) {
      // XREAD COUNT n BLOCK ms STREAMS key lastId
      const cursor = args[args.length - 1] as string;
      let from: number;
      if (cursor === "$") {
        if (marker === -1) {
          marker = entries.length; // first $ = start from the current tail
          await new Promise((r) => setTimeout(r, blockMs));
          return [];
        }
        from = marker;
      } else {
        const idx = entries.findIndex((e) => e[0] === cursor);
        from = idx === -1 ? entries.length : idx + 1;
      }
      if (from >= entries.length) {
        await new Promise((r) => setTimeout(r, blockMs)); // BLOCK — nothing new
        return [];
      }
      const slice = entries.slice(from);
      return [[String(args[5]!), slice]]; // the stream key + its entries
    },
  };
  return {
    client,
    /** The XADD side (the API's EventBus mirrors this shape). */
    xadd(event: DomainEvent): void {
      const [event_, payload, ts] = [
        event.name,
        JSON.stringify(event.payload),
        event.ts,
      ];
      entries.push([`${String(counter++).padStart(6, "0")}`, ["event", event_, "payload", payload, "ts", ts]]);
    },
  };
}

/** Join + full-sync a page room into `localDoc`. */
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();
  for (const f of client.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 client.binaryFrames()) {
    if (b.kind === "update") Y.applyUpdate(doc, b.bytes, "it");
  }
  return { client, doc };
}

/** The block's `text` slot (find the element 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 found = walk(el.toArray().filter((c): c is Y.XmlElement => c instanceof Y.XmlElement));
      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;
}

/** Append to a block's text (a real Yjs edit) + send as a 0x01 frame + await the ack. */
async function editBlock(client: RtClient, doc: Y.Doc, blockId: string, append: string): Promise<void> {
  const before = Y.encodeStateAsUpdate(doc);
  const slot = textSlotOf(doc, blockId);
  slot.insert(slot.length, append);
  client.sendBinary(1, Y.encodeStateAsUpdate(doc, before));
  await client.nextFrame("ack");
}

/** Connect a TOKEN-ONLY (roomless) control client + open its hello. */
async function connectControl(fx: RealtimeFixture, sub: string): Promise<RtClient> {
  const token = mintToken({ sub, workspaces: ["w1"] });
  const url = `${fx.wsUrl}/realtime?token=${encodeURIComponent(token)}`;
  const c = await connectClient(fx, { url });
  await c.nextFrame("hello");
  return c;
}

describe("TASK-0092 6c — the (b) convergence loop (real flush → block:changed → 2 control clients)", () => {
  let fx: RealtimeFixture;
  let stream: ReturnType<typeof makeLoopbackStream>;

  beforeAll(async () => {
    stream = makeLoopbackStream(5);
    // The REAL production source, fed by the loopback stream (the in-memory
    // factory:events). The fixture mirrors the server's EMIT side into it.
    // A FACTORY (not an instance) because the server builds the sink first
    // and starts/stops the source in start()/stop().
    const factory = (on: (e: DomainEvent) => void): PageUpdatedSource =>
      createPageUpdatedSource(on, { client: stream.client });
    fx = await buildRealtimeFixture({
      pages: new Set(["p-cv", "p-other"]),
      workspaces: { "p-cv": "w1", "p-other": "w1" },
      levels: { "page:p-cv": "edit", "page:p-other": "edit" },
      eventsTap: (event) => stream.xadd(event),
      pageUpdatedSource: factory,
    });
    // Let the source's FIRST XREAD (cursor "$" → set the tail marker) run
    // before the test emits anything, so a flush event is never behind the
    // marker (the real source starts from "$" = the current tail).
    await fx.settle(30);
  });

  afterAll(async () => {
    await fx.stop();
  });

  it("a room edit → the REAL flush → the REAL source → the hint reaches BOTH control clients of the page (a third client on another page gets nothing)", async () => {
    fx.store.seed("p-cv", [PARA("a", "base a", "1"), PARA("b", "base b", "2")]);

    // Two control tabs on the page (different users/tokens — the two-tab
    // story), plus a control tab on an UNRELATED page.
    const tab1 = await connectControl(fx, "u-tab1");
    const tab2 = await connectControl(fx, "u-tab2");
    const otherTab = await connectControl(fx, "u-other");
    try {
      tab1.sendText({ type: "subscribe", channel: "page:p-cv" });
      tab2.sendText({ type: "subscribe", channel: "page:p-cv" });
      otherTab.sendText({ type: "subscribe", channel: "page:p-other" });
      await fx.settle(50);

      // A real room client edits block "a" → the room is dirty → the 1 s
      // debounce flush fires → `page.updated` on the sink → the loopback
      // stream → the real source's XREAD → block:changed → fan-out.
      const { client, doc } = await joinAndSync(fx, "page:p-cv");
      try {
        await editBlock(client, doc, "a", "+edit");
        const persisted = await client.nextFrame("doc.persisted", 8_000);
        expect(String(persisted.pageId)).toBe("p-cv");

        // BOTH control tabs get the refetch-hint (the (b) convergence
        // signal — each tab's overlay clears + re-reads the blocks).
        const h1 = await tab1.nextFrame("block:changed", 8_000);
        const h2 = await tab2.nextFrame("block:changed", 8_000);
        expect(h1.pageId).toBe("p-cv");
        expect(h2.pageId).toBe("p-cv");
        // The version is advisory (the flush's store version ≥ 2).
        expect(Number(h1.version)).toBeGreaterThanOrEqual(2);
        expect(h2).toEqual(h1);

        // The UNRELATED page's control tab gets NOTHING for this flush
        // (channel isolation through the real path).
        await expect(otherTab.nextFrame("block:changed", 400)).rejects.toThrow();

        // The loopback really carried the server's own emission (not a
        // separately injected event): the sink recorded a page.updated with
        // the minimal blockDelta (1 update) for THIS flush.
        const evts = fx.events.pageUpdated("p-cv");
        expect(evts.length).toBeGreaterThanOrEqual(1);
        const delta = evts[evts.length - 1]!.payload.blockDelta;
        expect(delta).toEqual({ inserted: 0, updated: 1, moved: 0, deleted: 0 });

        // The store converged (the refetch the hint triggers would read
        // exactly this).
        const final = (await fx.store.hydrate("p-cv")).blocks!;
        expect(final.find((b) => b.id === "a")!.content).toEqual({ text: "base a+edit" });
        expect(final.find((b) => b.id === "b")!.content).toEqual({ text: "base b" });
      } finally {
        client.close();
        doc.destroy();
      }
    } finally {
      tab1.close();
      tab2.close();
      otherTab.close();
    }
  });
});