/**
 * Realtime integration test harness (TASK-0091).
 *
 * Boots the REAL `createRealtimeServer` (the production wiring minus the
 * DB/Redis seams — injectable fake store/log/permissions/meta/audit) on an
 * ephemeral 127.0.0.1 port pair and drives it with a real `ws` client.
 * No Docker is needed: the suite validates the wire protocol, handshake
 * matrix, heartbeat/prune, rate limits, resume, room GC, drain, and the
 * probe endpoints against the exact server code that ships.
 *
 * Time model: the server runs on REAL time (its heartbeat interval, flush
 * debounce, and room GC are real timers — the same as production; only
 * the clock readings are injectable). Timer-driven tests therefore use
 * short real cadences (`pingIntervalMs`, `pruneSilenceMs`, `roomGcMs`)
 * and wall-time waits, which is what a client experiences.
 */
import net from "node:net";
import { createHmac } from "node:crypto";
import WebSocket from "ws";
import { createRealtimeServer, type RealtimeServer } from "@factory/realtime/src/server";
import type { PageUpdatedSource } from "@factory/realtime/src/page-updated-source";
import type { DomainEventSource } from "@factory/realtime/src/event-source";
import type { RealtimeConfig } from "@factory/realtime/src/config";
import { InMemoryDocStore, type DocStore } from "@factory/realtime/src/rooms";
import { InMemoryUpdateLog } from "@factory/realtime/src/update-log";
import { WS_CLOSE } from "@factory/shared/realtime/frames";
import * as Y from "yjs";

import { DomainEventName, type DomainEvent, type BlockDelta } from "@factory/shared";

// --- fake audit sink ------------------------------------------------------------------

/** Records every published audit event (asserted for the abuse audit). */
export class FakeAudit {
  readonly records: { event: string; payload: Record<string, unknown> }[] = [];
  publish(event: string, payload: Record<string, unknown>): void {
    this.records.push({ event, payload });
  }
  abuse(cause?: string): number {
    return this.records.filter(
      (r) =>
        r.event === "realtime.connection.abuse" &&
        (cause === undefined || r.payload.cause === cause),
    ).length;
  }
}

/** Records every domain event the server emits (the flush's `page.updated`,
 *  TASK-0092 R3 — the `blockDelta` additive field is asserted against). */
export class FakeEventBus {
  readonly records: DomainEvent[] = [];
  publish(event: DomainEvent): void {
    this.records.push(event);
  }
  /** The `page.updated` events for a page (the flush's emission). */
  pageUpdated(pageId: string): DomainEvent<{
    pageId: string;
    version: number;
    blockDelta: BlockDelta;
    ops: unknown[];
  }>[] {
    return this.records.filter(
      (e) =>
        e.name === DomainEventName.PAGE_UPDATED &&
        (e.payload as { pageId?: string }).pageId === pageId,
    ) as DomainEvent<{ pageId: string; version: number; blockDelta: BlockDelta; ops: unknown[] }>[];
  }
}

// --- the fixture ------------------------------------------------------------------------

export interface RealtimeFixture {
  readonly server: RealtimeServer;
  readonly wsUrl: string;
  readonly healthUrl: string;
  readonly store: InMemoryDocStore;
  readonly updateLog: InMemoryUpdateLog;
  readonly audit: FakeAudit;
  /** Records the server's domain events (the flush's `page.updated`). */
  readonly events: FakeEventBus;
  /** The server's clock (the injected fake when `now0` is set, else real). */
  readonly now: () => number;
  /** Advance the fake clock (no-op on a real clock). */
  readonly advance: (ms: number) => void;
  /** Sleep wall time (let real timers + the network settle). */
  readonly settle: (ms: number) => Promise<void>;
  /** Always on in the fixture: the test clients identify their source IP
   *  via X-Forwarded-For (see `connectClient`'s auto-IP). */
  readonly trustProxy: boolean;
  stop: () => Promise<void>;
}

export interface BuildFixtureOpts {
  /** Inject a fake server clock starting at this epoch ms (deterministic
   *  prune/GC threshold asserts; the timers themselves stay real). */
  now0?: number;
  /** Heartbeat interval (REAL timer, ms). Default 30 s. */
  pingIntervalMs?: number;
  /** Silence threshold (ms, REAL time). Default 90 s. */
  pruneSilenceMs?: number;
  /** Empty-room GC delay (ms, REAL timer). Default 5 min. */
  roomGcMs?: number;
  /** `page:<id>` / `database:<id>` / `comment:<id>` → effective level.
   *  Default: pages → `edit` (comment rooms follow), databases → `view`. */
  levels?: Record<string, "none" | "view" | "comment" | "edit" | "full_access">;
  /** Existing page ids (the meta resolver's existence check). */
  pages?: Set<string>;
  /** Existing database ids. */
  databases?: Set<string>;
  /** Resource id → workspace id (the token's workspaces[] must contain it). */
  workspaces?: Record<string, string>;
  /** Honor X-Forwarded-For as the client IP (the user-cap test spreads
   *  100 conns over distinct IPs so the 30/min/IP limit doesn't fire first). */
  trustProxy?: boolean;
  /** The doc store the server persists to. Defaults to the fixture's
   *  seedable in-memory store (returned as `store`). Slice 4 injects a
   *  real `HttpDocStore` (pointed at a fake `/internal/*` API) to test
   *  the PRODUCTION flush path — room `flush()` → HTTP flush → the CAS
   *  conflict → the rebase → the retry — over real HTTP, Docker-free. When
   *  set, the returned `store` handle is a separate unused in-memory store
   *  (assert against your fake instead). */
  store?: DocStore;
  /** The `page.updated` CONTROL source factory (TASK-0092 6b): given the
   *  server's sink, returns the source to poll (tests = a fake the test
   *  drives; production = the `factory:events` XREAD poll). Omitted → no
   *  control fan-out. */
  pageUpdatedSource?: (onPageUpdated: (event: DomainEvent) => void) => PageUpdatedSource;
  /** The GENERIC domain-event CONTROL source factory (TASK-0094): given the
   *  server's sink, returns the source to poll (tests = a fake the test
   *  drives; production = the `factory:events` XREAD poll). When provided it
   *  supplants `pageUpdatedSource`. Omitted → no control fan-out. */
  domainEventSource?: (onEvent: (event: DomainEvent) => void) => DomainEventSource;
  /** Mirror every domain event the server emits (the flush's
   *  `page.updated`) into a test-owned sink — the loopback "Redis stream"
   *  for the 6c convergence test (the in-memory stand-in for the API's
   *  `factory:events` XADD). `fx.events` still records the same events. */
  eventsTap?: (event: DomainEvent) => void;
  /** userId → display name (the meta seam's `userName` — the peer roster
   *  + the presence identity). Default: every user is "Test User". */
  userNames?: Record<string, string>;
  /** Presence prune override (ms). Default: the doc §5.1 90 s
   *  (`REALTIME_LIMITS.awarenessPruneMs`). The presence prune test uses a
   *  short window + the fake clock (`now0`). */
  presencePruneMs?: number;
}

/**
 * Boot an in-process realtime server with fake seams. Seed page content
 * with `fixture.store.seed(pageId, blocks)` before connecting clients.
 */
export async function buildRealtimeFixture(opts: BuildFixtureOpts = {}): Promise<RealtimeFixture> {
  // Fake clock when `now0` is set (deterministic threshold asserts), real
  // clock otherwise. Timers stay real (the server wires real setTimeout /
  // setInterval — the injected clock drives READINGS, not scheduling).
  let t = opts.now0 ?? Date.now();
  const fake = opts.now0 !== undefined;
  const now = (): number => (fake ? t : Date.now());
  const audit = new FakeAudit();
  const events = new FakeEventBus();
  // The seedable in-memory handle (returned as `store`). The server
  // persists to `opts.store` when provided (slice 4's HttpDocStore), else
  // to this one (every pre-existing test).
  const seedStore = new InMemoryDocStore();
  const updateLog = new InMemoryUpdateLog();

  // The server's domain-event sink. `events` always records (the pre-existing
  // assert target); when an `eventsTap` is provided (6c), each emission is
  // also mirrored into it — the loopback "Redis stream" the REAL
  // `createPageUpdatedSource` polls (the in-memory `factory:events`
  // XADD stand-in).
  const eventsSink = {
    publish: (event: DomainEvent): void => {
      events.publish(event);
      opts.eventsTap?.(event);
    },
  };

  const levels = opts.levels ?? {};
  const pages = opts.pages ?? new Set<string>();
  const databases = opts.databases ?? new Set<string>();
  const workspaces = opts.workspaces ?? {};
  const levelMap: Record<string, "none" | "view" | "comment" | "edit" | "full_access"> = {};
  for (const [key, level] of Object.entries(levels)) levelMap[key] = level;
  for (const pageId of pages) {
    if (!(`page:${pageId}` in levelMap)) levelMap[`page:${pageId}`] = "edit";
    if (!(`comment:${pageId}` in levelMap)) levelMap[`comment:${pageId}`] = "edit";
  }
  for (const dbId of databases) {
    if (!(`database:${dbId}` in levelMap)) levelMap[`database:${dbId}`] = "view";
  }

  const ports = await freePorts(2);
  const config: RealtimeConfig = {
    port: ports[0]!,
    healthPort: ports[1]!,
    jwtSecret: IT_SECRET,
    redisUrl: "redis://unused",
    apiUrl: "http://unused",
    logLevel: "silent",
    rateLimitWindowMs: 60_000,
    rateLimitMax: 300,
    maxRoomSize: 200,
    useRedis: false,
    useDatabase: false,
    // Empty token + useDatabase:false → the in-memory store (buildStore
    // never consults the token on this path).
    internalServiceToken: "",
    // Always on: the test clients identify their source IP via
    // X-Forwarded-For (connectClient auto-assigns a unique synthetic IP
    // per connection — a single 127.0.0.1 would exhaust the 30/min/IP
    // connect budget in the second test of any file).
    trustProxy: true,
  };

  const server = createRealtimeServer(config, {
    now,
    pingIntervalMs: opts.pingIntervalMs ?? 30_000,
    pruneSilenceMs: opts.pruneSilenceMs ?? 90_000,
    roomGcMs: opts.roomGcMs ?? 5 * 60_000,
    store: opts.store ?? seedStore,
    log: updateLog,
    audit,
    events: eventsSink,
    pageUpdatedSource: opts.pageUpdatedSource,
    domainEventSource: opts.domainEventSource,
    presencePruneMs: opts.presencePruneMs,
    permissions: {
      resolveLevel: async ({ resourceType, resourceId }) =>
        levelMap[`${resourceType}:${resourceId}`] ?? "none",
    },
    meta: {
      resolve: async ({ kind, id, userId }) => {
        const name = (uid: string) => opts.userNames?.[uid] ?? "Test User";
        if (kind === "user") {
          return userId.length > 0 ? { workspaceId: "__own__", userName: name(userId) } : null;
        }
        if (kind === "page" || kind === "comment") {
          if (!pages.has(id)) return null;
          return { workspaceId: workspaces[id] ?? "w1", userName: name(userId) };
        }
        if (!databases.has(id)) return null;
        return { workspaceId: workspaces[id] ?? "w1", userName: name(userId) };
      },
    },
  });

  await server.start();
  return {
    server,
    wsUrl: `ws://127.0.0.1:${ports[0]}`,
    healthUrl: `http://127.0.0.1:${ports[1]}`,
    store: seedStore,
    updateLog,
    audit,
    events,
    now,
    advance: (ms) => {
      if (fake) t += ms;
    },
    settle: (ms) => new Promise((r) => setTimeout(r, ms)),
    trustProxy: true,
    stop: () => server.stop(),
  };
}

/**
 * A fake GENERIC domain-event source the test drives directly (no Redis) —
 * the in-memory stand-in for the production `factory:events` XREAD poll
 * (TASK-0094; the same pattern as `control-channel.test.ts`'s
 * `pageUpdatedSource` fake, now routing every event through the server's
 * sink). Pass `domainEventSource: factory` to {@link buildRealtimeFixture}.
 */
export function makeFakeDomainEventSource() {
  let sink: ((event: DomainEvent) => void) | null = null;
  const source: DomainEventSource = {
    start(): void {
      /* the test drives deliver(); nothing to poll. */
    },
    async stop(): Promise<void> {
      sink = null;
    },
  };
  return {
    source,
    factory: (onEvent: (event: DomainEvent) => void): DomainEventSource => {
      sink = onEvent;
      return source;
    },
    /** Deliver a raw domain event through the server's routing sink. */
    deliver(name: string, payload: Record<string, unknown>): void {
      if (sink === null)
        throw new Error("source not wired (the server was built without the factory)");
      sink({
        id: `evt_${Math.random().toString(36).slice(2)}`,
        name: name as DomainEvent["name"],
        payload,
        ts: new Date().toISOString(),
      });
    },
  };
}

/** Free ephemeral 127.0.0.1 ports (the server binds them). */
export async function freePorts(n: number): Promise<number[]> {
  const out: number[] = [];
  for (let i = 0; i < n; i += 1) {
    const srv = net.createServer();
    await new Promise<void>((resolve, reject) => {
      srv.once("error", reject);
      srv.listen(0, "127.0.0.1", () => resolve());
    });
    const addr = srv.address();
    if (addr === null || typeof addr === "string") throw new Error("no free port");
    out.push(addr.port);
    await new Promise<void>((resolve) => srv.close(() => resolve()));
  }
  return out;
}

// --- JWT (the API's claim shape: `sub`, `workspaces[]`) ----------------------------------

export const IT_SECRET = "it-secret-0123456789";

function b64url(s: string): string {
  return Buffer.from(s, "utf8").toString("base64url");
}

/** Mint a handshake JWT the server accepts (HS256, shared secret). */
export function mintToken(
  claims: { sub: string; workspaces: string[]; expSec?: number; iss?: string },
  secret: string = IT_SECRET,
): string {
  const expSec = claims.expSec ?? Math.floor(Date.now() / 1000) + 3600;
  const body: Record<string, unknown> = {
    sub: claims.sub,
    iat: Math.floor(Date.now() / 1000),
    exp: expSec,
    jti: `j-${Math.random().toString(36).slice(2, 10)}`,
    workspaces: claims.workspaces,
  };
  if (claims.iss !== undefined) body.iss = claims.iss;
  const h = b64url(JSON.stringify({ alg: "HS256", typ: "JWT" }));
  const c = b64url(JSON.stringify(body));
  const sig = createHmac("sha256", secret).update(`${h}.${c}`).digest().toString("base64url");
  return `${h}.${c}.${sig}`;
}

// --- the ws client ----------------------------------------------------------------------

export interface RtClient {
  readonly ws: WebSocket;
  textFrames: () => Record<string, unknown>[];
  binaryFrames: () => { kind: "update" | "awareness"; bytes: Uint8Array }[];
  /** Await the next `n` frames of `type` (already-received ones count). */
  nextFrames: (type: string, n: number, timeoutMs?: number) => Promise<Record<string, unknown>[]>;
  nextFrame: (type: string, timeoutMs?: number) => Promise<Record<string, unknown>>;
  /** Await the first text frame matching a predicate (e.g. `done === true`). */
  nextFrameWhere: (
    pred: (f: Record<string, unknown>) => boolean,
    timeoutMs?: number,
  ) => Promise<Record<string, unknown>>;
  /** Await the next binary frame (the 0x01/0x02-prefixed Yjs payloads). */
  nextBinary: (timeoutMs?: number) => Promise<{ kind: "update" | "awareness"; bytes: Uint8Array }>;
  sendText: (obj: unknown) => void;
  sendBinary: (kind: 1 | 2, bytes: Uint8Array) => void;
  /** The observed close (null while open). */
  closed: () => { code: number; reason: string } | null;
  /** Resolve when the socket closes (already closed → immediate). */
  onClose: (timeoutMs?: number) => Promise<{ code: number; reason: string }>;
  isOpen: () => boolean;
  close: (code?: number) => void;
  terminate: () => void;
}

/** Connect a real `ws` client to the gateway. Rejects on a refused upgrade.
 *
 * Source IP: every call gets a UNIQUE synthetic IP via X-Forwarded-For
 * (the fixture always trusts the proxy). Without this, one 127.0.0.1
 * would exhaust the 30-connects/min/IP budget in the second test of any
 * file — the rate-limit tests pass an explicit `headers["x-forwarded-for"]`
 * when they deliberately share an IP. */
let testIpCounter = 0;
function autoTestIp(): string {
  testIpCounter = (testIpCounter + 1) % 0xffffff;
  return `10.${(testIpCounter >> 16) & 0xff}.${(testIpCounter >> 8) & 0xff}.${testIpCounter & 0xff}`;
}

export async function connectClient(
  base: { wsUrl: string },
  opts: { token?: string; room?: string; url?: string; headers?: Record<string, string> } = {},
): Promise<RtClient> {
  const token = opts.token ?? mintToken({ sub: "u-it", workspaces: ["w1"] });
  const room = opts.room ?? "page:p-it";
  const url =
    opts.url ??
    `${base.wsUrl}/realtime?token=${encodeURIComponent(token)}&room=${encodeURIComponent(room)}`;
  const headers = { "x-forwarded-for": autoTestIp(), ...opts.headers };
  const ws = new WebSocket(url, { headers });

  /**
   * Two consuming queues (text frames, binary frames). `nextFrame`/
   * `nextFrames`/`nextFrameWhere` return the NEXT unconsumed matching
   * text frames and mark them consumed, so two successive
   * `nextFrame("ack")` calls never see the same ack (the resume test's
   * seq accounting depends on this). `textFrames()`/`binaryFrames()` are
   * NON-consuming history views (e.g. counting the replayed updates).
   * The `consumed` flag alone provides the "don't return a frame twice"
   * guarantee — there is no cursor, so a predicate that skips frames
   * never hides earlier frames from a different waiter.
   */
  interface Bin {
    kind: "update" | "awareness";
    bytes: Uint8Array;
  }
  const textQueue: { frame: Record<string, unknown>; consumed: boolean }[] = [];
  const binQueue: { bin: Bin; consumed: boolean }[] = [];
  interface Waiter {
    label: string;
    pred: ((f: Record<string, unknown>) => boolean) | null;
    need: number;
    timer: ReturnType<typeof setTimeout>;
    resolve: (v: Record<string, unknown>[]) => void;
    reject: (e: Error) => void;
  }
  const waiters: Waiter[] = [];
  let closeInfo: { code: number; reason: string } | null = null;
  const closeWaiters: {
    resolve: (v: { code: number; reason: string }) => void;
    timer: ReturnType<typeof setTimeout>;
  }[] = [];

  /** The first `need` unconsumed text frames matching `pred` (indexes). */
  const findMatches = (
    pred: ((f: Record<string, unknown>) => boolean) | null,
    need: number,
  ): number[] => {
    const out: number[] = [];
    for (let j = 0; j < textQueue.length && out.length < need; j += 1) {
      const qe = textQueue[j]!;
      if (qe.consumed) continue;
      if (pred === null || pred(qe.frame)) out.push(j);
    }
    return out;
  };
  const consume = (idxs: number[]): Record<string, unknown>[] => {
    for (const j of idxs) textQueue[j]!.consumed = true;
    return idxs.map((j) => textQueue[j]!.frame);
  };

  /** Try to satisfy every waiter with the next unconsumed matching text
   *  frames. Returns true when at least one waiter was satisfied. */
  const settleWaiters = (): boolean => {
    let satisfied = false;
    for (let i = waiters.length - 1; i >= 0; i -= 1) {
      const w = waiters[i]!;
      const matched = findMatches(w.pred, w.need);
      if (matched.length >= w.need) {
        const frames = consume(matched);
        waiters.splice(i, 1);
        clearTimeout(w.timer);
        w.resolve(frames);
        satisfied = true;
      }
    }
    return satisfied;
  };

  ws.on("message", (data: Buffer | ArrayBuffer | Buffer[], isBinary: boolean) => {
    if (isBinary) {
      const b = Buffer.isBuffer(data) ? new Uint8Array(data) : new Uint8Array(data as ArrayBuffer);
      binQueue.push({
        bin: { kind: b[0] === 2 ? "awareness" : "update", bytes: b.subarray(1) },
        consumed: false,
      });
    } else {
      let parsed: Record<string, unknown>;
      try {
        parsed = JSON.parse(Buffer.isBuffer(data) ? data.toString("utf8") : String(data)) as Record<
          string,
          unknown
        >;
      } catch {
        return;
      }
      textQueue.push({ frame: parsed, consumed: false });
    }
    settleWaiters();
  });
  ws.on("close", (code, reason) => {
    closeInfo = { code, reason: reason.toString() };
    for (const w of closeWaiters.splice(0)) {
      clearTimeout(w.timer);
      w.resolve(closeInfo!);
    }
    settleWaiters();
  });

  await new Promise<void>((resolve, reject) => {
    const t = setTimeout(() => reject(new Error("connect timeout")), 5_000);
    ws.once("open", () => {
      clearTimeout(t);
      resolve();
    });
    ws.once("error", (err) => {
      clearTimeout(t);
      reject(err instanceof Error ? err : new Error(String(err)));
    });
  });

  const addWaiter = (
    pred: ((f: Record<string, unknown>) => boolean) | null,
    need: number,
    timeoutMs: number,
    label: string,
  ): Promise<Record<string, unknown>[]> =>
    new Promise((resolve, reject) => {
      // Fast path: already enough unconsumed matches in the queue.
      const matched = findMatches(pred, need);
      if (matched.length >= need) {
        resolve(consume(matched));
        return;
      }
      const w: Waiter = {
        label,
        pred,
        need,
        timer: setTimeout(() => {
          const idx = waiters.indexOf(w);
          if (idx >= 0) waiters.splice(idx, 1);
          const got = textQueue.filter((q) => !q.consumed).length;
          reject(
            new Error(
              `timeout waiting for ${label} (${got} unconsumed frames, socket readyState=${ws.readyState})`,
            ),
          );
        }, timeoutMs),
        resolve,
        reject,
      };
      waiters.push(w);
      settleWaiters();
    });

  const nextFrames = (
    type: string,
    n: number,
    timeoutMs = 8_000,
  ): Promise<Record<string, unknown>[]> =>
    addWaiter((f) => f.type === type, n, timeoutMs, `${n} × "${type}"`);
  const nextFrame = (type: string, timeoutMs = 8_000): Promise<Record<string, unknown>> =>
    addWaiter((f) => f.type === type, 1, timeoutMs, `"${type}"`).then(
      (frames) => frames[frames.length - 1]!,
    );
  const nextFrameWhere = (
    pred: (f: Record<string, unknown>) => boolean,
    timeoutMs = 8_000,
  ): Promise<Record<string, unknown>> =>
    addWaiter(pred, 1, timeoutMs, "frame matching predicate").then(
      (frames) => frames[frames.length - 1]!,
    );

  return {
    ws,
    textFrames: () => textQueue.map((q) => q.frame),
    binaryFrames: () => binQueue.map((q) => q.bin),
    nextFrames,
    nextFrame,
    nextFrameWhere,
    nextBinary: (timeoutMs = 8_000) =>
      new Promise((resolve, reject) => {
        // The next unconsumed binary frame (never the same one twice).
        const already = binQueue.find((q) => !q.consumed);
        if (already) {
          already.consumed = true;
          resolve(already.bin);
          return;
        }
        const timer = setTimeout(
          () => reject(new Error("timeout waiting for a binary frame")),
          timeoutMs,
        );
        const onMsg = (data: Buffer | ArrayBuffer | Buffer[], isBinary: boolean): void => {
          if (!isBinary) return;
          const b = Buffer.isBuffer(data)
            ? new Uint8Array(data)
            : new Uint8Array(data as ArrayBuffer);
          clearTimeout(timer);
          ws.off("message", onMsg);
          resolve({ kind: b[0] === 2 ? "awareness" : "update", bytes: b.subarray(1) });
        };
        ws.on("message", onMsg);
      }),
    sendText: (obj) => ws.send(JSON.stringify(obj)),
    sendBinary: (kind, bytes) => {
      const out = new Uint8Array(bytes.byteLength + 1);
      out[0] = kind;
      out.set(bytes, 1);
      ws.send(out);
    },
    closed: () => closeInfo,
    onClose: (timeoutMs = 8_000) =>
      new Promise((resolve, reject) => {
        if (closeInfo !== null) {
          resolve(closeInfo);
          return;
        }
        const w = {
          resolve,
          timer: setTimeout(
            () => reject(new Error("timeout waiting for close (socket still open)")),
            timeoutMs,
          ),
        };
        closeWaiters.push(w);
      }),
    isOpen: () => ws.readyState === WebSocket.OPEN,
    close: (code = 1000) => ws.close(code),
    terminate: () => ws.terminate(),
  };
}

// --- a RAW socket (the doc's "never pongs" heartbeat scenario) ----------------------------

export interface RawSocket {
  readonly socket: net.Socket;
  /** Unmasked server→client close-frame codes, in order. */
  closeCodes: () => number[];
  isOpen: () => boolean;
  destroy: () => void;
}

/**
 * Hand-roll the WS upgrade on a bare TCP socket. Unlike the `ws` client
 * (which auto-pongs), this socket NEVER answers the server's protocol
 * pings — exactly the doc's "a socket silent 90 s is terminated with
 * code 4000" scenario.
 */
export async function connectRawSocket(
  base: { wsUrl: string },
  opts: { token?: string; room?: string } = {},
): Promise<RawSocket> {
  const token = opts.token ?? mintToken({ sub: "u-raw", workspaces: ["w1"] });
  const room = opts.room ?? "page:p-it";
  const port = Number(new URL(base.wsUrl).port);
  const socket = net.connect(port, "127.0.0.1");
  const closeCodes: number[] = [];
  let upgraded = false;

  const scanForCloseFrames = (buf: Buffer): void => {
    for (let i = 0; i + 4 <= buf.length; i += 1) {
      // A real unmasked close frame: FIN bit set, opcode 0x8, no mask
      // bit (server→client is unmasked), small payload, and a code in
      // the valid close range (1000–4999). Without the FIN + range
      // checks the scanner false-positives on the 101-Upgrade response
      // body (recorded test bug).
      const b0 = buf[i]!;
      const b1 = buf[i + 1]!;
      if ((b0 & 0x80) !== 0x80) continue; // FIN not set — not a frame start
      if ((b0 & 0x0f) !== 0x8) continue; // not a close frame
      if ((b1 & 0x80) !== 0) continue; // masked — not a server frame
      if (b1 >= 126) continue; // payload > 125 bytes — not a close frame
      const code = (buf[i + 2]! << 8) | buf[i + 3]!;
      if (code >= 1000 && code <= 4999) closeCodes.push(code);
    }
  };

  await new Promise<void>((resolve, reject) => {
    const t = setTimeout(() => reject(new Error("raw upgrade timeout")), 5_000);
    socket.on("connect", () => {
      socket.write(
        [
          `GET /realtime?token=${encodeURIComponent(token)}&room=${encodeURIComponent(room)} HTTP/1.1`,
          `Host: 127.0.0.1:${port}`,
          "Upgrade: websocket",
          "Connection: Upgrade",
          "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==",
          "Sec-WebSocket-Version: 13",
          "",
          "",
        ].join("\r\n"),
      );
    });
    socket.on("data", (buf) => {
      scanForCloseFrames(buf);
      if (!upgraded) {
        const head = buf.subarray(0, 60).toString("utf8");
        if (head.includes("101")) {
          upgraded = true;
          clearTimeout(t);
          resolve();
        } else if (/^HTTP\/1\.[01] [45]\d\d/.test(head)) {
          clearTimeout(t);
          reject(new Error(`raw upgrade refused: ${head.split("\r\n")[0]}`));
        }
      }
    });
    socket.on("error", (err) => {
      clearTimeout(t);
      reject(err);
    });
  });

  return {
    socket,
    closeCodes: () => [...closeCodes],
    isOpen: () => !socket.destroyed,
    destroy: () => socket.destroy(),
  };
}

/** Fetch a JSON probe endpoint. */
export async function fetchJson(
  url: string,
): Promise<{ status: number; body: Record<string, unknown> }> {
  const res = await fetch(url);
  let body: Record<string, unknown> = {};
  try {
    body = (await res.json()) as Record<string, unknown>;
  } catch {
    /* non-JSON body */
  }
  return { status: res.status, body };
}

// --- raw HTTP upgrade probe (pre-join handshake rejections) ------------------------

/**
 * Issue the WS upgrade request over a bare TCP socket and return the HTTP
 * response status (the handshake rejections happen AT the upgrade —
 * 401/403/400/429 — before any WebSocket close frame exists, R1).
 */
export async function probeUpgrade(
  base: { wsUrl: string },
  opts: { token?: string; room?: string; path?: string; headers?: Record<string, string> } = {},
): Promise<{ status: number; line: string }> {
  const token = opts.token ?? mintToken({ sub: "u-probe", workspaces: ["w1"] });
  const room = opts.room ?? "page:p-it";
  const path =
    opts.path ?? `/realtime?token=${encodeURIComponent(token)}&room=${encodeURIComponent(room)}`;
  const port = Number(new URL(base.wsUrl).port);
  const socket = net.connect(port, "127.0.0.1");
  try {
    const allHeaders = { "x-forwarded-for": autoTestIp(), ...opts.headers };
    const extra = Object.entries(allHeaders)
      .map(([k, v]) => `${k}: ${v}`)
      .join("\r\n");
    await new Promise<void>((resolve) => {
      socket.once("connect", () => resolve());
    });
    // The two trailing empty strings are MANDATORY: they form the
    // \r\n\r\n that terminates the HTTP header block (a .filter that
    // drops empty lines would leave the request unterminated — the
    // server waits for more data forever; recorded test bug).
    const lines = [
      `GET ${path} HTTP/1.1`,
      `Host: 127.0.0.1:${port}`,
      "Upgrade: websocket",
      "Connection: Upgrade",
      "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==",
      "Sec-WebSocket-Version: 13",
    ];
    if (extra !== "") lines.push(extra);
    lines.push("", "");
    socket.write(lines.join("\r\n"));
    let buf = "";
    await new Promise<void>((resolve, reject) => {
      const t = setTimeout(() => {
        socket.destroy();
        reject(new Error("probe timeout"));
      }, 5_000);
      const finish = (): void => {
        clearTimeout(t);
        resolve();
      };
      socket.once("data", (d) => {
        buf += d.toString("utf8");
        finish();
      });
      // The wrong-path case: the server destroys the socket WITHOUT an
      // HTTP response (EOF — no 'data', but 'end'/'close' still fire).
      socket.once("end", finish);
      socket.once("close", finish);
      socket.once("error", reject);
    });
    const line = buf.split("\r\n")[0] ?? "";
    const m = line.match(/HTTP\/1\.[01] (\d{3})/);
    return { status: Number(m?.[1] ?? 0), line };
  } finally {
    socket.destroy();
  }
}

// --- Yjs test updates ----------------------------------------------------------------

/** Append `text` to the Yjs doc's `yText` type and return the update bytes
 *  (the `update` binary frame payload for the 0x01 prefix). The delta
 *  against the pre-edit state vector keeps the log small + the relay
 *  honest (a full-state update would be idempotent too, but wasteful). */
export function makeTextUpdate(doc: Y.Doc, text: string): Uint8Array {
  const yText = doc.getText("yText");
  const before = Y.encodeStateAsUpdate(doc);
  yText.insert(0, text);
  // `encodeStateAsUpdate(doc, sv)` is already binary — no re-encode.
  return Y.encodeStateAsUpdate(doc, before);
}

export { WS_CLOSE };