/**
 * Realtime integration (TASK-0094 R1/R7 — the comment + notification
 * fan-out over the `factory:events` stream).
 *
 * Boots the REAL `createRealtimeServer` (rt-helpers) with an INJECTED
 * generic `domainEventSource` (a fake the test drives — the production
 * `factory:events` XREAD poll is Docker-gated; the seam is the same). The
 * server's routing sink maps each event to its frame:
 *   - `comment.*`            → `comment.event` (to `comment:{pageId}`
 *     doc-room sockets AND roomless control sockets with a `page:{pageId}`
 *     channel — the (b) web architecture),
 *   - `notification.created` → `notification.push` (to the RECIPIENT's
 *     control room only — the isolation assert),
 *   - `page.updated`         → `block:changed` (the TASK-0092 6b
 *     regression, now through the same generic source).
 *
 * Docker-free (no Redis: the source is a fake).
 */
import { afterAll, beforeAll, describe, expect, it } from "vitest";

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

interface FixtureWithSource {
  fx: RealtimeFixture;
  source: ReturnType<typeof makeFakeDomainEventSource>;
}

let suite: FixtureWithSource | null = null;

beforeAll(async () => {
  const source = makeFakeDomainEventSource();
  const fx = await buildRealtimeFixture({
    pages: new Set(["pg_1", "pg_2"]),
    workspaces: { pg_1: "w1", pg_2: "w1" },
    userNames: { u1: "Ana", u2: "Ben" },
    domainEventSource: source.factory,
  });
  suite = { fx, source };
});

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

/** Connect a `comment:{pageId}` doc-room socket. */
async function connectCommentRoom(fx: RealtimeFixture, user: string, pageId: string) {
  const token = mintToken({ sub: user, workspaces: ["w1"] });
  const url = `${fx.wsUrl}/realtime?token=${encodeURIComponent(token)}&room=${encodeURIComponent(`comment:${pageId}`)}`;
  return connectClient(fx, { url });
}

/** Connect a TOKEN-ONLY roomless control socket (the web handshake). */
async function connectControl(fx: RealtimeFixture, user: string) {
  const token = mintToken({ sub: user, workspaces: ["w1"] });
  const url = `${fx.wsUrl}/realtime?token=${encodeURIComponent(token)}`;
  return connectClient(fx, { url });
}

describe("comment.event fan-out (TASK-0094 R1)", () => {
  it("reaches a comment:{pageId} doc-room socket (kind 'created' + preview)", async () => {
    const { fx, source } = suite!;
    const c = await connectCommentRoom(fx, "u1", "pg_1");
    await fx.settle(20);
    const p = c.nextFrame("comment.event");
    source.deliver("comment.created", {
      workspaceId: "w1",
      commentId: "c_1",
      pageId: "pg_1",
      actorId: "u2",
      body: "hello from Ben",
    });
    const frame = await p;
    expect(frame).toMatchObject({
      type: "comment.event",
      kind: "created",
      commentId: "c_1",
      pageId: "pg_1",
      actorId: "u2",
      preview: "hello from Ben",
    });
    c.close();
  });

  it("reaches a roomless CONTROL socket with a page:{pageId} channel (the (b) web client)", async () => {
    const { fx, source } = suite!;
    const c = await connectControl(fx, "u1");
    c.sendText({ type: "subscribe", channel: "page:pg_1" });
    await fx.settle(20);
    const p = c.nextFrame("comment.event");
    source.deliver("comment.created", {
      workspaceId: "w1",
      commentId: "c_2",
      replyId: "r_2",
      pageId: "pg_1",
      actorId: "u2",
      body: "a reply",
    });
    const frame = await p;
    // A reply (comment.created WITH a replyId) → kind 'replied'.
    expect(frame).toMatchObject({ type: "comment.event", kind: "replied", commentId: "c_2" });
    c.close();
  });

  it("does NOT reach a socket subscribed to a DIFFERENT page's channel", async () => {
    const { fx, source } = suite!;
    const c = await connectControl(fx, "u1");
    c.sendText({ type: "subscribe", channel: "page:pg_2" });
    await fx.settle(20);
    source.deliver("comment.created", {
      workspaceId: "w1",
      commentId: "c_3",
      pageId: "pg_1",
      actorId: "u2",
      body: "for pg_1 only",
    });
    await fx.settle(50);
    // No comment.event for pg_1 arrived on the pg_2 channel.
    const got = c.textFrames().filter((f) => f.type === "comment.event");
    expect(got).toHaveLength(0);
    c.close();
  });

  it("does NOT reach a PAGE doc-room socket (comment rooms + page channels only — recorded)", async () => {
    const { fx, source } = suite!;
    // A `?room=page:pg_1` doc-room socket (the TASK-0136 provider shape).
    const c = await connectClient(fx, {
      token: mintToken({ sub: "u1", workspaces: ["w1"] }),
      room: "page:pg_1",
    });
    await fx.settle(20);
    source.deliver("comment.created", {
      workspaceId: "w1",
      commentId: "c_4",
      pageId: "pg_1",
      actorId: "u2",
      body: "not for doc-room sockets",
    });
    await fx.settle(50);
    const got = c.textFrames().filter((f) => f.type === "comment.event");
    expect(got).toHaveLength(0);
    c.close();
  });

  it("fans resolved/reopened/reacted/deleted kinds to the page channel", async () => {
    const { fx, source } = suite!;
    const c = await connectControl(fx, "u1");
    c.sendText({ type: "subscribe", channel: "page:pg_1" });
    await fx.settle(20);
    const pResolved = c.nextFrameWhere((f) => f.type === "comment.event" && f.kind === "resolved");
    const pReacted = c.nextFrameWhere((f) => f.type === "comment.event" && f.kind === "reacted");
    source.deliver("comment.resolved", {
      workspaceId: "w1",
      commentId: "c_5",
      pageId: "pg_1",
      actorId: "u1",
    });
    source.deliver("comment.reaction", {
      workspaceId: "w1",
      commentId: "c_5",
      pageId: "pg_1",
      actorId: "u2",
      emoji: "👍",
      active: true,
    });
    expect(await pResolved).toMatchObject({ commentId: "c_5", kind: "resolved", preview: "" });
    expect(await pReacted).toMatchObject({ commentId: "c_5", kind: "reacted" });
    c.close();
  });
});

describe("notification.push control-room fan-out (TASK-0094 R7)", () => {
  it("reaches the RECIPIENT's control socket (all their tabs) with the frame fields", async () => {
    const { fx, source } = suite!;
    // Two tabs for u1 (a user with multiple control sockets gets both).
    const c1 = await connectControl(fx, "u1");
    const c2 = await connectControl(fx, "u1");
    await fx.settle(20);
    const p1 = c1.nextFrame("notification.push");
    const p2 = c2.nextFrame("notification.push");
    source.deliver("notification.created", {
      workspaceId: "w1",
      notificationId: "n_1",
      userId: "u1",
      type: "MENTION",
      title: "You were mentioned in a comment",
      deepLink: { type: "comment", id: "c_9" },
      actorId: "u2",
      actorName: "Ben",
    });
    const f1 = await p1;
    const f2 = await p2;
    for (const f of [f1, f2]) {
      expect(f).toMatchObject({
        type: "notification.push",
        notificationId: "n_1",
        kind: "mention",
        title: "You were mentioned in a comment",
        deepLink: { type: "comment", id: "c_9" },
        actorName: "Ben",
      });
    }
    c1.close();
    c2.close();
  });

  it("ISOLATION: a notification for user X never appears on user Y's control socket", async () => {
    const { fx, source } = suite!;
    const other = await connectControl(fx, "u2");
    await fx.settle(20);
    source.deliver("notification.created", {
      workspaceId: "w1",
      notificationId: "n_2",
      userId: "u1",
      type: "COMMENT",
      title: "Ben replied in a thread",
      deepLink: { type: "comment", id: "c_10" },
      actorName: "Ben",
    });
    await fx.settle(60);
    const got = other.textFrames().filter((f) => f.type === "notification.push");
    expect(got).toHaveLength(0);
    other.close();
  });

  it("a user with NO live sockets → no push error (the fan-out is a no-op)", async () => {
    const { fx, source } = suite!;
    // u9 has no socket. The deliver must not throw (the row is the truth).
    expect(() =>
      source.deliver("notification.created", {
        workspaceId: "w1",
        notificationId: "n_3",
        userId: "u9",
        type: "ASSIGNMENT",
        title: "Assigned to you",
        deepLink: { type: "record", id: "r_1" },
      }),
    ).not.toThrow();
  });

  it("unknown types still push (conservative kind — no crash)", async () => {
    const { fx, source } = suite!;
    const c = await connectControl(fx, "u1");
    await fx.settle(20);
    const p = c.nextFrame("notification.push");
    source.deliver("notification.created", {
      workspaceId: "w1",
      notificationId: "n_4",
      userId: "u1",
      type: "SOME_FUTURE_TYPE",
      title: "t",
      deepLink: { type: "page", id: "pg_1" },
    });
    expect(await p).toMatchObject({ notificationId: "n_4", kind: "some_future_type" });
    c.close();
  });
});

describe("the generic source routes page.updated too (TASK-0092 6b regression)", () => {
  it("page.updated → block:changed on the page channel (same source as comments)", async () => {
    const { fx, source } = suite!;
    const c = await connectControl(fx, "u1");
    c.sendText({ type: "subscribe", channel: "page:pg_1" });
    await fx.settle(20);
    const p = c.nextFrame("block:changed");
    source.deliver("page.updated", { pageId: "pg_1", version: 12 });
    expect(await p).toMatchObject({ type: "block:changed", pageId: "pg_1", version: 12 });
    c.close();
  });

  it("an unrelated event (record.created) produces NO control frames", async () => {
    const { fx, source } = suite!;
    const c = await connectControl(fx, "u1");
    c.sendText({ type: "subscribe", channel: "page:pg_1" });
    await fx.settle(20);
    source.deliver("record.created", { workspaceId: "w1", recordId: "r_9", databaseId: "d_1" });
    await fx.settle(50);
    const controlFrames = c
      .textFrames()
      .filter(
        (f) =>
          f.type === "comment.event" ||
          f.type === "notification.push" ||
          f.type === "block:changed",
      );
    expect(controlFrames).toHaveLength(0);
    c.close();
  });
});