/**
 * Realtime integration (TASK-0093 slice 2 — presence + live cursors): the
 * server's presence producer over the (b) wire — JSON control frames
 * (`presence` client → server; `presence` / `presence-leave` server →
 * client), NOT the binary 0x02 awareness relay (that stays for the
 * TASK-0136 provider path; the conflict is recorded in the task file).
 *
 * Boots the REAL `createRealtimeServer` (rt-helpers) with real `ws`
 * clients. Coverage (the task spec R1–R5 + the (b) producer):
 *  - R1: A joins a page room → B's join re-broadcast carries A's full
 *    state; A's `presence` frame (cursor move) reaches B within 500 ms;
 *    a third client on a DIFFERENT page receives nothing.
 *  - R2: identity is server-stamped (the client frame carries ONLY
 *    cursor/viewing — the broadcast's `user` comes from the meta seam's
 *    display name + the shared presenceColor hash; a client-supplied
 *    name is impossible by schema).
 *  - R3: > 50 peers → avatars-only (cursor/viewing stripped); the
 *    boundary is 50 full / 51 stripped.
 *  - R4: leave propagation < 500 ms — an explicit `presence-leave` AND a
 *    socket close both announce `presence-leave` to the room.
 *  - R5: the 90 s prune (a short `presencePruneMs` + the fake clock)
 *    drops a silent peer and announces it.
 *  - (b) fan-out: a roomless CONTROL socket subscribed to `page:<id>`
 *    receives the room's broadcasts (and its own presence updates reach
 *    the doc-room peers) — the two transports interoperate.
 *
 * Docker-free (the spec's 3-client E2E twin is Docker-gated under
 * `tests/e2e` — same scenario, real browser).
 */
import { afterAll, beforeAll, describe, expect, it } from "vitest";

import { presenceColor } from "@factory/shared/presence";

import { buildRealtimeFixture, connectClient, mintToken, type RealtimeFixture } from "./rt-helpers";

/** The fixture's distinct display names (the meta seam's userNames opt). */
const NAMES: Record<string, string> = {
  "u-ana": "Ana",
  "u-bob": "Bob",
  "u-carol": "Carol",
};

async function connectUser(fx: RealtimeFixture, sub: string, room: string) {
  const token = mintToken({ sub, workspaces: ["w1"] });
  return connectClient(fx, {
    token,
    url: `${fx.wsUrl}/realtime?token=${encodeURIComponent(token)}&room=${encodeURIComponent(room)}`,
  });
}

/** A token-only (roomless) control client — the (b) web handshake. */
async function connectControl(fx: RealtimeFixture, sub: string) {
  const token = mintToken({ sub, workspaces: ["w1"] });
  return connectClient(fx, {
    token,
    url: `${fx.wsUrl}/realtime?token=${encodeURIComponent(token)}`,
  });
}

const presenceOf = (f: Record<string, unknown>) =>
  f as {
    type: string;
    pageId?: string;
    user?: { id: string; name: string; color: string };
    cursor?: { blockId: string; offset: number } | null;
    viewing?: boolean;
    lastSeen?: number;
  };

describe("TASK-0093 — presence fan-out (doc-room transport)", () => {
  let fx: RealtimeFixture;

  beforeAll(async () => {
    fx = await buildRealtimeFixture({
      pages: new Set(["p-pres", "p-other", "p-cap", "p-ctl"]),
      userNames: NAMES,
    });
  });

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

  it("R1+R2: join re-broadcast + cursor move < 500 ms, other pages isolated, identity server-stamped", async () => {
    const a = await connectUser(fx, "u-ana", "page:p-pres");
    const b = await connectUser(fx, "u-bob", "page:p-pres");
    const c = await connectUser(fx, "u-carol", "page:p-other");
    try {
      const ha = await a.nextFrame("hello");
      expect(ha.roomPeers).toEqual([]);
      const hb = await b.nextFrame("hello");
      // B's hello: A is already in the room (the roster, pre-existing).
      expect(hb.roomPeers).toHaveLength(1);

      // A's join re-broadcast reached B (the joiner's initial state — no
      // cursor, viewing=true, server-stamped identity).
      const aSeenByB = presenceOf(
        await b.nextFrameWhere(
          (f) => f.type === "presence" && (f as { user?: { id?: string } }).user?.id === "u-ana",
        ),
      );
      expect(aSeenByB.type).toBe("presence");
      expect(aSeenByB.pageId).toBe("page:p-pres");
      const aUser = aSeenByB.user!;
      // R2: the identity is SERVER-STAMPED — the display name from the
      // meta seam (the client frame carries no user field at all) + the
      // shared hash's color.
      expect(aUser.name).toBe("Ana");
      expect(aUser.color).toBe(presenceColor("u-ana"));
      expect(aSeenByB.cursor).toBeNull();
      expect(aSeenByB.viewing).toBe(true);
      expect(typeof aSeenByB.lastSeen).toBe("number");

      // R1: A moves its cursor → B receives it within the 500 ms
      // contract (in-process the round-trip is single-digit ms).
      const t0 = Date.now();
      a.sendText({ type: "presence", cursor: { blockId: "blk_1", offset: 4 }, viewing: true });
      const moved = await b.nextFrameWhere(
        (f) => f.type === "presence" && (f as { user?: { id?: string } }).user?.id === "u-ana",
      );
      const latencyMs = Date.now() - t0;
      expect(latencyMs).toBeLessThan(500);
      expect(moved.cursor).toEqual({ blockId: "blk_1", offset: 4 });
      expect(moved.viewing).toBe(true);

      // R2 (b): the SENDER gets no echo of its own state (it knows it).
      expect(
        a
          .textFrames()
          .filter(
            (f) => f.type === "presence" && (f as { user?: { id?: string } }).user?.id === "u-ana",
          ),
      ).toHaveLength(0);

      // Isolation: C (a different page) received NO presence frames.
      expect(
        c.textFrames().filter((f) => f.type === "presence" || f.type === "presence-leave"),
      ).toHaveLength(0);
    } finally {
      a.close();
      b.close();
      c.close();
      await Promise.all([a.onClose(), b.onClose(), c.onClose()]);
    }
  });

  it("R4: an explicit presence-leave reaches the room in < 500 ms", async () => {
    const a = await connectUser(fx, "u-ana", "page:p-pres");
    const b = await connectUser(fx, "u-bob", "page:p-pres");
    try {
      await a.nextFrame("hello");
      await b.nextFrame("hello");
      // B's join re-broadcast: B received A's state (the joiner hears the
      // existing peers).
      await b.nextFrameWhere(
        (f) => f.type === "presence" && (f as { user?: { id?: string } }).user?.id === "u-ana",
      );
      const t0 = Date.now();
      a.sendText({ type: "presence-leave" });
      const leave = await b.nextFrame("presence-leave");
      const latencyMs = Date.now() - t0;
      expect(latencyMs).toBeLessThan(500);
      expect(leave.pageId).toBe("page:p-pres");
      expect(leave.userId).toBe("u-ana");
    } finally {
      a.close();
      b.close();
      await Promise.all([a.onClose(), b.onClose()]);
    }
  });

  it("R4: a socket close announces the peer's leave", async () => {
    const a = await connectUser(fx, "u-ana", "page:p-pres");
    const b = await connectUser(fx, "u-bob", "page:p-pres");
    try {
      await a.nextFrame("hello");
      await b.nextFrame("hello");
      // B's join re-broadcast (so B has A's state first — the leave must
      // be a distinct frame).
      await b.nextFrameWhere(
        (f) => f.type === "presence" && (f as { user?: { id?: string } }).user?.id === "u-ana",
      );
      a.close();
      const leave = await b.nextFrame("presence-leave");
      expect(leave.userId).toBe("u-ana");
      expect(leave.pageId).toBe("page:p-pres");
    } finally {
      b.close();
      await b.onClose();
    }
  });

  it("R3: the 50-peer cap — 50 peers carry full cursors, the 51st is avatars-only", async () => {
    // 49 peers join (u-cap-0..48) + the observer (u-ana) = 50 = the cap.
    const peers: Awaited<ReturnType<typeof connectUser>>[] = [];
    const observer = await connectUser(fx, "u-ana", "page:p-cap");
    await observer.nextFrame("hello");
    try {
      for (let i = 0; i < 49; i += 1) {
        const p = await connectUser(fx, `u-cap-${i}`, "page:p-cap");
        await p.nextFrame("hello");
        peers.push(p);
      }
      // The observer's JOIN re-broadcast (initial state, cursor null)
      // reached the peers — consume it so the cursor update below is the
      // FIRST remaining u-ana frame (the discriminator is the cursor).
      await peers[0]!.nextFrameWhere(
        (f) => f.type === "presence" && (f as { user?: { id?: string } }).user?.id === "u-ana",
      );

      // A cursor update from the observer (50 peers — exactly at the
      // cap) is carried FULL (the cap strips only ABOVE 50).
      observer.sendText({ type: "presence", cursor: { blockId: "blk_9", offset: 1 } });
      const full = await peers[0]!.nextFrameWhere(
        (f) =>
          f.type === "presence" &&
          (f as { user?: { id?: string } }).user?.id === "u-ana" &&
          (f as { cursor?: { blockId?: string } | null }).cursor?.blockId === "blk_9",
      );
      expect(full.cursor).toEqual({ blockId: "blk_9", offset: 1 });

      // The 51st peer joins → the room is over the cap: its INITIAL
      // state (the join re-broadcast) arrives avatars-only.
      const over = await connectUser(fx, "u-cap-51", "page:p-cap");
      await over.nextFrame("hello");
      const stripped = presenceOf(
        await over.nextFrameWhere(
          (f) => f.type === "presence" && (f as { user?: { id?: string } }).user?.id === "u-ana",
        ),
      );
      expect(stripped.cursor).toBeNull();
      expect(stripped.viewing).toBeUndefined();
      expect(stripped.user?.name).toBe("Ana"); // the avatar stays

      // …and a subsequent cursor update from the observer is also
      // stripped (the cap is live, not just at join). The sleep keeps the
      // upsert's `lastSeen` on a later millisecond (the frame
      // discriminator below).
      await new Promise((r) => setTimeout(r, 10));
      observer.sendText({ type: "presence", cursor: { blockId: "blk_9", offset: 2 } });
      const stripped2 = await over.nextFrameWhere(
        (f) =>
          f.type === "presence" &&
          (f as { user?: { id?: string } }).user?.id === "u-ana" &&
          (f as { lastSeen?: number }).lastSeen !== stripped.lastSeen,
      );
      expect(stripped2.cursor).toBeNull();

      // Dropping back to ≤ 50 (the 51st leaves) restores full cursors.
      over.close();
      await over.onClose();
      await new Promise((r) => setTimeout(r, 10));
      observer.sendText({ type: "presence", cursor: { blockId: "blk_9", offset: 3 } });
      const full2 = await peers[0]!.nextFrameWhere(
        (f) =>
          f.type === "presence" &&
          (f as { user?: { id?: string } }).user?.id === "u-ana" &&
          (f as { lastSeen?: number }).lastSeen !== stripped2.lastSeen,
      );
      expect(full2.cursor).toEqual({ blockId: "blk_9", offset: 3 });
    } finally {
      for (const p of peers) {
        p.close();
      }
      observer.close();
      await Promise.all([...peers.map((p) => p.onClose()), observer.onClose()]);
    }
  });
});

describe("TASK-0093 — the (b) control-socket transport", () => {
  let fx: RealtimeFixture;

  beforeAll(async () => {
    fx = await buildRealtimeFixture({
      pages: new Set(["p-ctl"]),
      userNames: NAMES,
    });
  });

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

  it("a roomless control socket subscribed to page:<id> receives the room's presence + its own updates reach the doc-room peers", async () => {
    const doc = await connectUser(fx, "u-bob", "page:p-ctl");
    const ctl = await connectControl(fx, "u-ana");
    try {
      const hdoc = await doc.nextFrame("hello");
      expect(hdoc.roomPeers).toEqual([]);
      const hctl = await ctl.nextFrame("hello");
      expect(hctl.roomPeers).toEqual([]);

      // The control socket opens the page channel (the (b) handshake).
      ctl.sendText({ type: "subscribe", channel: "page:p-ctl" });

      // 1) The doc-room peer's state reaches the control socket: Bob's
      //    join re-broadcast (his initial state) lands on Ana's control
      //    socket (it's subscribed to the room).
      const bobSeenByCtl = presenceOf(
        await ctl.nextFrameWhere(
          (f) => f.type === "presence" && (f as { user?: { id?: string } }).user?.id === "u-bob",
        ),
      );
      expect(bobSeenByCtl.pageId).toBe("page:p-ctl");
      expect(bobSeenByCtl.user?.name).toBe("Bob");
      expect(bobSeenByCtl.cursor).toBeNull();

      // 2) The control socket's OWN presence reaches the doc-room peer
      //    (the server stamps Ana's identity from the meta seam's
      //    `kind: "user"` path — no handshake meta for a roomless
      //    socket). Ana's JOIN re-broadcast (initial state, cursor
      //    null) reached the doc peer first — consume it, then assert
      //    the cursor update (the discriminator is the cursor).
      await doc.nextFrameWhere(
        (f) => f.type === "presence" && (f as { user?: { id?: string } }).user?.id === "u-ana",
      );
      const t0 = Date.now();
      ctl.sendText({ type: "presence", cursor: { blockId: "blk_c", offset: 2 }, viewing: true });
      const anaSeenByDoc = presenceOf(
        await doc.nextFrameWhere(
          (f) =>
            f.type === "presence" &&
            (f as { user?: { id?: string } }).user?.id === "u-ana" &&
            (f as { cursor?: { blockId?: string } | null }).cursor?.blockId === "blk_c",
        ),
      );
      expect(Date.now() - t0).toBeLessThan(500);
      expect(anaSeenByDoc.user?.name).toBe("Ana");
      expect(anaSeenByDoc.user?.color).toBe(presenceColor("u-ana"));
      expect(anaSeenByDoc.cursor).toEqual({ blockId: "blk_c", offset: 2 });

      // 3) Ana's leave (the explicit frame) is announced to the doc peer.
      ctl.sendText({ type: "presence-leave" });
      const leave = await doc.nextFrame("presence-leave");
      expect(leave.userId).toBe("u-ana");
      expect(leave.pageId).toBe("page:p-ctl");

      // 4) A control socket with NO page channel gets nothing (a
      //    workspace-only subscription is not a presence room).
      const ctl2 = await connectControl(fx, "u-carol");
      try {
        await ctl2.nextFrame("hello");
        ctl2.sendText({ type: "subscribe", channel: "workspace:w1" });
        ctl.sendText({ type: "presence", cursor: { blockId: "blk_c", offset: 3 } });
        // Give any (buggy) cross-room frame a chance to arrive.
        await new Promise((r) => setTimeout(r, 150));
        expect(ctl2.textFrames().filter((f) => f.type === "presence")).toHaveLength(0);
      } finally {
        ctl2.close();
        await ctl2.onClose();
      }
    } finally {
      doc.close();
      ctl.close();
      await Promise.all([doc.onClose(), ctl.onClose()]);
    }
  });

  it("a control socket's presence follows the LAST subscribed page channel (page switch)", async () => {
    const ctl = await connectControl(fx, "u-ana");
    const doc = await connectUser(fx, "u-bob", "page:p-ctl");
    try {
      await ctl.nextFrame("hello");
      await doc.nextFrame("hello");
      ctl.sendText({ type: "subscribe", channel: "page:p-ctl" });
      // Let Ana's join re-broadcast settle.
      await doc.nextFrameWhere(
        (f) => f.type === "presence" && (f as { user?: { id?: string } }).user?.id === "u-ana",
      );

      // Ana switches pages: unsubscribe + resubscribe (the (b) client's
      // page navigation). Her presence room moves; the old room gets a
      // leave.
      ctl.sendText({ type: "unsubscribe", channel: "page:p-ctl" });
      const leave = await doc.nextFrame("presence-leave");
      expect(leave.userId).toBe("u-ana");
    } finally {
      ctl.close();
      doc.close();
      await Promise.all([ctl.onClose(), doc.onClose()]);
    }
  });
});

describe("TASK-0093 — R5 the 90 s prune (the fake clock)", () => {
  let fx: RealtimeFixture;

  beforeAll(async () => {
    // Short prune window (5 s on the fake clock) + a fast heartbeat
    // cadence so the prune loop fires quickly on REAL time. The R4
    // socket prune (pruneSilenceMs) stays long so the silent peer's
    // SOCKET survives — only its PRESENCE entry is dropped (doc §5.1:
    // the client heals a pruned peer by dropping it; the socket's
    // `presence` re-asserts it).
    // `now0` = the REAL current time (the fake clock starts "now" — the
    // mitted tokens' `exp` (real now + 1 h) stays valid; `advance()`
    // moves only the fake clock, so the prune window math is relative).
    fx = await buildRealtimeFixture({
      pages: new Set(["p-prune"]),
      now0: Date.now(),
      userNames: NAMES,
      presencePruneMs: 5_000,
      pingIntervalMs: 200,
      pruneSilenceMs: 3600_000, // the socket prune is effectively off
    });
  });

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

  it("a peer silent past the prune window is dropped + announced", async () => {
    const a = await connectUser(fx, "u-ana", "page:p-prune");
    const b = await connectUser(fx, "u-bob", "page:p-prune");
    try {
      await a.nextFrame("hello");
      await b.nextFrame("hello");
      // A's join re-broadcast (B now has A's state — the leave must be
      // a distinct, later frame).
      await b.nextFrameWhere(
        (f) => f.type === "presence" && (f as { user?: { id?: string } }).user?.id === "u-ana",
      );
      // A asserts its cursor (lastSeen = now on the fake clock).
      a.sendText({ type: "presence", cursor: { blockId: "blk_p", offset: 1 } });
      await b.nextFrameWhere(
        (f) =>
          f.type === "presence" &&
          (f as { user?: { id?: string } }).user?.id === "u-ana" &&
          (f as { cursor?: unknown }).cursor !== null,
      );

      // Keep A's entry FRESH while we set up (a live client re-asserts
      // its presence periodically — the real web client sends it on
      // selection changes; the prune loop runs every 200 ms on real
      // time, so without this the 5 s window could expire mid-test).
      fx.advance(1_000);
      a.sendText({ type: "presence", cursor: { blockId: "blk_p", offset: 1 } });
      await b.nextFrameWhere(
        (f) =>
          f.type === "presence" &&
          (f as { user?: { id?: string } }).user?.id === "u-ana" &&
          (f as { cursor?: unknown }).cursor !== null,
      );

      // Now A goes silent: age it past the 5 s prune window (B keeps
      // heartbeating, so only A's PRESENCE entry goes stale).
      fx.advance(5_001);
      b.sendText({ type: "ping", ts: 1 });
      const leave = await b.nextFrame("presence-leave");
      expect(leave.userId).toBe("u-ana");
      expect(leave.pageId).toBe("page:p-prune");

      // A's socket SURVIVED the presence prune (the R4 socket prune is
      // at the long threshold) — and a re-asserted `presence` puts A
      // back in the roster (the doc §5.2 best-effort heal).
      expect(a.isOpen()).toBe(true);
      a.sendText({ type: "presence", cursor: { blockId: "blk_p", offset: 2 } });
      const back = await b.nextFrameWhere(
        (f) =>
          f.type === "presence" &&
          (f as { user?: { id?: string } }).user?.id === "u-ana" &&
          (f as { cursor?: { blockId?: string } | null }).cursor?.blockId === "blk_p",
      );
      expect(back.cursor).toEqual({ blockId: "blk_p", offset: 2 });
    } finally {
      a.close();
      b.close();
      await Promise.all([a.onClose(), b.onClose()]);
    }
  });
});