/**
 * Realtime integration (TASK-0092 slice 6b — the (b) producer): the
 * roomless CONTROL socket + the `page.updated` → `block:changed`
 * refetch-hint fan-out.
 *
 * Boots the REAL `createRealtimeServer` on an ephemeral port (rt-helpers)
 * with an INJECTED `pageUpdatedSource` (a fake the test drives — the
 * production `factory:events` XREAD poll is Docker-gated; the seam is the
 * same). A real `ws` client connects TOKEN-ONLY (no `?room=` — the web
 * `RealtimeClient`'s handshake), opens a `page:<id>` channel in-band via
 * `subscribe`, and receives the `block:changed` refetch-hint when the
 * source delivers a `page.updated`. Docker-free.
 */
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import {
  buildRealtimeFixture,
  connectClient,
  mintToken,
  probeUpgrade,
  type RealtimeFixture,
} from "./rt-helpers";
import type { DomainEvent } from "@factory/shared";
import type { PageUpdatedSource } from "@factory/realtime/src/page-updated-source";

/** A fake `page.updated` source the test drives directly (no Redis). */
function makeFakeSource() {
  let sink: ((event: DomainEvent) => void) | null = null;
  const source: PageUpdatedSource = {
    start(): void {
      /* the test drives deliver(); nothing to poll. */
    },
    async stop(): Promise<void> {
      sink = null;
    },
  };
  return {
    source,
    factory: (onPageUpdated: (event: DomainEvent) => void): PageUpdatedSource => {
      sink = onPageUpdated;
      return source;
    },
    /** Deliver a `page.updated` through the server's sink (the fan-out). */
    deliver(pageId: string, version?: number): void {
      expect(sink).not.toBeNull();
      sink!({
        id: `evt_${Math.random().toString(36).slice(2)}`,
        name: "page.updated",
        payload: version !== undefined ? { pageId, version } : { pageId },
        ts: new Date().toISOString(),
      });
    },
  };
}

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

describe("TASK-0092 6b — the roomless control socket + block:changed fan-out", () => {
  let fx: RealtimeFixture;
  let src: ReturnType<typeof makeFakeSource>;

  beforeAll(async () => {
    src = makeFakeSource();
    fx = await buildRealtimeFixture({
      pages: new Set(["p-ctl", "p-other"]),
      workspaces: { "p-ctl": "w1", "p-other": "w1" },
      pageUpdatedSource: src.factory,
    });
  });

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

  it("admits a token-only (roomless) upgrade and sends a control hello", async () => {
    const c = await connectControl(fx);
    try {
      const hello = await c.nextFrame("hello");
      expect(hello.type).toBe("hello");
      expect(typeof hello.connId).toBe("string");
    } finally {
      c.close();
    }
  });

  it("fans a block:changed refetch-hint to the page's control channel on a page.updated", async () => {
    const c = await connectControl(fx);
    try {
      await c.nextFrame("hello");
      c.sendText({ type: "subscribe", channel: "page:p-ctl" });
      await fx.settle(50);

      src.deliver("p-ctl", 7);

      const hint = await c.nextFrame("block:changed");
      expect(hint).toEqual({ type: "block:changed", pageId: "p-ctl", version: 7 });
    } finally {
      c.close();
    }
  });

  it("sends the hint WITHOUT a version when the page.updated has none (a REST write)", async () => {
    const c = await connectControl(fx);
    try {
      await c.nextFrame("hello");
      c.sendText({ type: "subscribe", channel: "page:p-ctl" });
      await fx.settle(50);

      src.deliver("p-ctl");

      const hint = await c.nextFrame("block:changed");
      expect(hint).toEqual({ type: "block:changed", pageId: "p-ctl" });
    } finally {
      c.close();
    }
  });

  it("only fans to the subscribed channel (page A's update does not reach page B's subscriber)", async () => {
    const other = await connectControl(fx);
    try {
      await other.nextFrame("hello");
      other.sendText({ type: "subscribe", channel: "page:p-other" });
      await fx.settle(50);

      src.deliver("p-ctl", 3);

      // The subscriber of p-other must NOT receive p-ctl's hint.
      await expect(other.nextFrame("block:changed", 400)).rejects.toThrow();
    } finally {
      other.close();
    }
  });

  it("a page:<id> channel and a workspace:<id> channel are both accepted (workspace rides no frames yet)", async () => {
    const c = await connectControl(fx);
    try {
      await c.nextFrame("hello");
      c.sendText({ type: "subscribe", channel: "page:p-ctl" });
      c.sendText({ type: "subscribe", channel: "workspace:w1" });
      await fx.settle(50);

      // Only the page channel fans; the workspace channel is accepted but
      // no frame is delivered for it (no strike, socket still open).
      src.deliver("p-ctl", 1);
      const hint = await c.nextFrame("block:changed");
      expect(hint.pageId).toBe("p-ctl");
      expect(c.isOpen()).toBe(true);
    } finally {
      c.close();
    }
  });

  it("unsubscribe stops the fan-out for that channel", async () => {
    const c = await connectControl(fx);
    try {
      await c.nextFrame("hello");
      c.sendText({ type: "subscribe", channel: "page:p-ctl" });
      await fx.settle(50);
      c.sendText({ type: "unsubscribe", channel: "page:p-ctl" });
      await fx.settle(50);

      src.deliver("p-ctl", 5);
      await expect(c.nextFrame("block:changed", 400)).rejects.toThrow();
    } finally {
      c.close();
    }
  });

  it("a control socket's sync.request is answered with an empty sync.state (no doc)", async () => {
    const c = await connectControl(fx);
    try {
      await c.nextFrame("hello");
      c.sendText({ type: "sync.request" });
      const st = await c.nextFrameWhere((f) => f.type === "sync.state" && f.done === true);
      expect(st.type).toBe("sync.state");
    } finally {
      c.close();
    }
  });

  it("a malformed room key is STILL rejected (400) — only a MISSING room is a control socket", async () => {
    const token = mintToken({ sub: "u-badroom", workspaces: ["w1"] });
    const r = await probeUpgrade(fx, { token, room: "nonsense" });
    expect(r.status).toBe(400);
  });
});