/**
 * The realtime server (TASK-0091) — the production WebSocket server of
 * `docs/architecture/realtime-architecture.md` §1–2 + §4.5, with the R3
 * wire contract (`packages/shared/realtime/frames.ts` + the
 * `packages/collaboration` client mirror).
 *
 *  - R1 handshake: upgrade on `/realtime?token=<JWT>&room=<roomKey>`
 *    (the `Authorization: Bearer` header path is kept for proxies).
 *    Validation order: IP connect rate (30/min) → JWT (4001) →
 *    room key (4000 bad_room) → user cap (100) → workspace
 *    membership (4002) → room permission via the shared engine (4002)
 *    → join → `hello {connId, canEdit, serverTime, roomPeers}`.
 *  - R2 RoomManager: one authoritative Y.Doc per room (page/database),
 *    comment rooms are fan-out (no doc), `user:{userId}` control rooms
 *    ride the same socket (note 16, ≤ 10 live sockets/user).
 *  - R3 frames: text JSON control frames + binary 0x01/0x02; corrupt
 *    updates + unknown frames count strikes (10 in 10 s → 4008 + audit
 *    `realtime.connection.abuse`).
 *  - R4 heartbeat: server `ws.ping()` every 30 s; 90 s silence → 4000.
 *  - R5 resume: state-vector delta (paged `sync.state` ≤ 64 KB) OR the
 *    Redis update log (`lastSeq` fast path; log gap → state vector).
 *  - R6 lifecycle: close codes 4000–4010, SIGTERM drain (4010 +
 *    `retryAfterMs`, flush all dirty rooms, ≤ 10 s), `/health` +
 *    `/readyz` + `/metrics` on BOTH ports (note 15).
 *
 * The persistence seam (note 17): `DocStore` — the production Prisma
 * implementation lives in `./prisma-store.js` (wired by `index.ts`);
 * the default is the in-memory store (dev + unit/integration tests).
 * The permission/meta seams are the same shared engine + a Prisma
 * assembly (`./prisma-store.js`); the defaults are fail-closed.
 */

import http from "node:http";
import type net from "node:net";
import { randomUUID } from "node:crypto";
import pino, { type Logger } from "pino";
import * as Y from "yjs";
import { WebSocketServer, type WebSocket } from "ws";

import { verifyHandshakeToken, type SessionIdentity } from "./auth.js";
import type { RealtimeConfig } from "./config.js";
import { Metrics, mergeExposition } from "./metrics.js";
import {
  encodeBinaryAwareness,
  encodeBinaryUpdate,
  encodeServerFrame,
  decodeClientFrame,
  MAX_FRAME_BYTES,
  StrikeTracker,
} from "./protocol.js";
import { parseRoomKey, type ParsedRoom } from "./room-key.js";
import {
  InMemoryDocStore,
  makePeer,
  RoomManager,
  type DocStore,
  type Room,
  type RoomManagerEvents,
} from "./rooms.js";
import { ConnectLimiter, TokenBucket } from "./rate-limit.js";
import { PresenceTracker, type PresenceUpdate } from "./presence.js";
import { InMemoryUpdateLog, type UpdateLog } from "./update-log.js";
import {
  REALTIME_LIMITS,
  WS_CLOSE,
  type BlockChangedFrame,
  type CommentEventFrame,
  type NotificationPushFrame,
  type ServerFrame,
  type SyncStateFrame,
} from "@factory/shared/realtime/frames";
import {
  DomainEventName,
  createDomainEvent,
  type DomainEvent,
  type BlockOperation,
} from "@factory/shared";
import { blockChangedForPageUpdated, type PageUpdatedSource } from "./page-updated-source.js";
import type { DomainEventSource } from "./event-source.js";
import { commentEventFrameFor } from "./comments.js";
import { notificationPushFor } from "./notifications.js";
import { m, renderPrometheus, startSpan } from "@factory/observability";

// --- injectable seams (tests + the TASK-0092 production wiring) -----------------

/**
 * The permission resolver — the SAME engine as the API guard (the
 * shared `@factory/permissions` per the architecture doc; the API's
 * Prisma assembly is mirrored in `./prisma-store.js` — the conflict is
 * recorded in the task file). `level` is the effective level.
 */
export interface PermissionResolver {
  resolveLevel(input: {
    userId: string;
    workspaceId: string;
    resourceType: "page" | "database";
    resourceId: string;
  }): Promise<"none" | "view" | "comment" | "edit" | "full_access">;
}

/**
 * Resolves a room's workspace (note 14 — the room key carries none; the
 * server resolves it server-side) + the joining user's display name
 * (the peer roster). `null` = unknown resource (4002 no-existence).
 */
export interface RoomMetaResolver {
  resolve(input: {
    kind: ParsedRoom["kind"];
    id: string;
    userId: string;
  }): Promise<{ workspaceId: string; userName?: string } | null>;
}

/** The audit/event seam — the api's audit consumer writes the row. */
export interface AuditSink {
  publish(event: "realtime.connection.abuse", payload: Record<string, unknown>): void;
}

/**
 * The domain-event seam (TASK-0092 R3): the flush emits `page.updated`
 * (the Phase-3 event shape, the `blockDelta` field additive — the Phase-8
 * indexer + the computed-recompute consumer both listen). Production wires
 * this to the event bus (the same `factory:events` stream the API
 * publishes; the realtime app has no Redis dependency of its own, so the
 * seam is injected — the worker consumes the stream). The default (no
 * seam) is a no-op, so the server stays testable without a bus.
 */
export interface EventSink {
  publish(event: DomainEvent): void;
}

export interface RealtimeServerDeps {
  readonly logger?: Logger;
  readonly store?: DocStore;
  readonly log?: UpdateLog & { ping?: () => Promise<boolean> };
  readonly permissions?: PermissionResolver;
  readonly meta?: RoomMetaResolver;
  readonly audit?: AuditSink;
  /** The domain-event seam (the flush's `page.updated`, TASK-0092 R3). */
  readonly events?: EventSink;
  /** The `page.updated` CONTROL source (TASK-0092 6b): a FACTORY that,
   *  given the server's `page.updated` sink, returns the source to poll
   *  (production = the `factory:events` stream poll in index.ts; tests =
   *  a fake the test drives). The server starts it in `start()` and
   *  stops it in `stop()`. Omitted → no control fan-out. */
  readonly pageUpdatedSource?: (onPageUpdated: (event: DomainEvent) => void) => PageUpdatedSource;
  /** The GENERIC domain-event CONTROL source (TASK-0094): a FACTORY that,
   *  given the server's domain-event sink, returns the source to poll
   *  (production = the `factory:events` stream poll in index.ts; tests =
   *  a fake the test drives). Routes EVERY event: `page.updated` →
   *  `block:changed`, `comment.*` → `comment.event`,
   *  `notification.created` → `notification.push`. When provided it
   *  SUPPLANTS `pageUpdatedSource` (one stream, one poller — the legacy
   *  seam is kept only so pre-TASK-0094 tests keep passing). Omitted → no
   *  control fan-out. */
  readonly domainEventSource?: (onEvent: (event: DomainEvent) => void) => DomainEventSource;
  readonly now?: () => number;
  /** Heartbeat cadence override (tests use short intervals). */
  readonly pingIntervalMs?: number;
  readonly pruneSilenceMs?: number;
  /** Empty-room GC delay override (the R2 5 min warm room; the
   *  integration test uses a short delay to exercise the evict path). */
  readonly roomGcMs?: number;
  /** Presence prune override (ms; default 90 s — the doc §5.1 awareness
   *  timeout). The integration prune test uses a short window. */
  readonly presencePruneMs?: number;
}

interface SocketState {
  connId: string;
  session: SessionIdentity;
  /** The joined room (`page:`/`database:`/`comment:`/`user:`) — NULL for
   *  a roomless CONTROL socket (TASK-0092 6b: the web `RealtimeClient`
   *  connects token-only, no `?room=`, and opens `page:<id>` channels
   *  in-band via `subscribe`). */
  room: ParsedRoom | null;
  /** The control channels this socket has opened in-band
   *  (`subscribe page:<id>` / `workspace:<id>`). Only non-empty on
   *  roomless control sockets. */
  pageChannels: Set<string>;
  /** The doc room (page/database) or null (comment/user). */
  docRoom: Room | null;
  ip: string;
  canEdit: boolean;
  strikes: StrikeTracker;
  lastActivity: number;
  lastAckedSeq: number;
  closed: boolean;
  dropped: boolean;
  /** The display name resolved at handshake (the presence identity — the
   *  meta seam's `userName`; control sockets are named lazily on first
   *  `presence` frame, then cached). */
  peerName?: string;
  /** The page/database channel this control socket's presence currently
   *  tracks (roomless sockets only — the LAST subscribed
   *  `page:<id>`/`database:<id>` channel; presence frames before any
   *  such subscription are ignored). */
  lastPresenceRoom?: string;
}

export interface RealtimeServer {
  start(): Promise<void>;
  stop(): Promise<void>;
  /** SIGTERM drain (R6): 4010 to all sockets, flush all rooms, ≤ 10 s. */
  drain(): Promise<void>;
  /** Push a comment event to a page's comment room (TASK-0094 seam). */
  publishCommentEvent(pageId: string, event: CommentEventFrame): void;
  /** Push a notification to a user's control room (TASK-0094 seam). */
  publishNotification(userId: string, event: NotificationPushFrame): void;
  /** Push a `block:changed` refetch-hint to every control socket that has
   *  opened the page's `page:<id>` channel (TASK-0092 6b / the (b)
   *  producer). Mirrors the TASK-0094 comment/notification fan-out seams;
   *  no-op when no control socket has the channel. */
  publishBlockChanged(frame: BlockChangedFrame): void;
  /** Test/diagnostic: the live socket states. */
  sockets(): readonly SocketState[];
  readonly manager: RoomManager;
  readonly metrics: Metrics;
}

export function createRealtimeServer(
  config: RealtimeConfig,
  deps: RealtimeServerDeps = {},
): RealtimeServer {
  const logger = deps.logger ?? pino({ level: config.logLevel, enabled: false });
  const now = deps.now ?? Date.now;
  const pingIntervalMs = deps.pingIntervalMs ?? REALTIME_LIMITS.pingIntervalMs;
  const pruneSilenceMs = deps.pruneSilenceMs ?? REALTIME_LIMITS.pruneSilenceMs;
  const roomGcMs = deps.roomGcMs ?? REALTIME_LIMITS.roomGcMs; // R2: 5 min warm room
  const presencePruneMs = deps.presencePruneMs ?? REALTIME_LIMITS.awarenessPruneMs; // doc §5.1: 90 s

  const metrics = new Metrics(now);
  const store = deps.store ?? new InMemoryDocStore();
  const updateLog = deps.log ?? new InMemoryUpdateLog();
  const permissions = deps.permissions ?? failClosedPermissions(logger);
  const meta = deps.meta ?? failClosedMeta(logger);
  const audit = deps.audit ?? { publish: () => undefined };
  const events = deps.events;

  const socketsByUser = new Map<string, Set<WebSocket>>();
  const userControlSockets = new Map<string, WebSocket[]>();
  const states = new Map<WebSocket, SocketState>();
  /** The room's max logged update seq (the flush snapshot's trim
   *  watermark, R5 — the trim basis is the LOG seq, not the page
   *  version; the seq domain restarts per room lifetime). */
  const logSeqByRoom = new Map<string, number>();
  /** roomKey → the room's workspaceId (the `page.updated` event's
   *  `workspaceId`, recorded at join from the handshake meta). */
  const workspaceIdByRoom = new Map<string, string>();
  const msgRate = new TokenBucket({
    capacity: REALTIME_LIMITS.maxMsgsPerSocketPerSec,
    rate: REALTIME_LIMITS.maxMsgsPerSocketPerSec,
    now,
  });
  const connectLimiter = new ConnectLimiter({
    windowMs: 60_000,
    max: REALTIME_LIMITS.connectPerIpPerMin,
    now,
  });
  /** Presence (TASK-0093, doc §5) — one room-scoped tracker for the
   *  process. Peers are keyed (roomKey, userId): a user's two tabs in the
   *  same room share one avatar; identity is server-stamped (the handshake
   *  id + the meta seam's display name + the shared presenceColor hash). */
  const presence = new PresenceTracker({ now, pruneAfterMs: presencePruneMs });
  /** roomKey → the live doc-room sockets currently in that room (the
   *  presence fan-out target; `comment`/`user` rooms are never tracked —
   *  presence is page/database only, doc §1). Updated on join/close. */
  const roomSockets = new Map<string, Set<WebSocket>>();

  let wss: WebSocketServer | null = null;
  let httpServer: http.Server | null = null;
  let probeServer: http.Server | null = null;
  let heartbeat: ReturnType<typeof setInterval> | undefined;
  let presencePrune: ReturnType<typeof setInterval> | undefined;
  let draining = false;
  let startedAt = 0;

  const findWsByConn = (connId: string): WebSocket | null => {
    for (const [ws, st] of states) {
      if (st.connId === connId) return ws;
    }
    return null;
  };

  const sendText = (ws: WebSocket, frame: ServerFrame): void => {
    if (ws.readyState === 1) ws.send(encodeServerFrame(frame));
  };

  /** Count the flush's minimal ops into the `page.updated` `blockDelta`
   *  shape (the spec's `{inserted, updated, moved, deleted}` counts). */
  const blockDeltaCounts = (
    ops: readonly BlockOperation[],
  ): {
    inserted: number;
    updated: number;
    moved: number;
    deleted: number;
  } => {
    let inserted = 0;
    let updated = 0;
    let moved = 0;
    let deleted = 0;
    for (const op of ops) {
      if (op.op === "insert") inserted += 1;
      else if (op.op === "update") updated += 1;
      else if (op.op === "move") moved += 1;
      else if (op.op === "delete") deleted += 1;
    }
    return { inserted, updated, moved, deleted };
  };

  const sendBinary = (ws: WebSocket, bytes: Uint8Array): void => {
    if (ws.readyState === 1) ws.send(bytes);
  };

  const dropSocket = (
    ws: WebSocket,
    code: number,
    cause: string,
    opts?: { auditCause?: string },
  ): void => {
    const st = states.get(ws);
    if (st === undefined) {
      // Defensive: no state (already cleaned up) — count + close only.
      metrics.inc("ws_drops_total", { cause });
      try {
        ws.close(code, cause);
      } catch {
        ws.terminate();
      }
      return;
    }
    if (st.dropped) return;
    st.dropped = true;
    const { connId, ip } = st;
    const userId = st.session.userId;
    states.delete(ws);
    m.ws.setConnections(states.size); // TASK-0135 — connection gauge
    msgRate.drop(connId);
    const userSet = socketsByUser.get(userId);
    if (userSet !== undefined) {
      userSet.delete(ws);
      if (userSet.size === 0) socketsByUser.delete(userId);
    }
    const controlArr = userControlSockets.get(userId);
    if (controlArr !== undefined) {
      const i = controlArr.indexOf(ws);
      if (i >= 0) controlArr.splice(i, 1);
      if (controlArr.length === 0) userControlSockets.delete(userId);
    }
    metrics.inc("ws_drops_total", { cause });
    if (opts?.auditCause !== undefined) {
      audit.publish("realtime.connection.abuse", {
        connId,
        userId,
        ip,
        room: st.room === null ? "control" : `${st.room.kind}:${st.room.id}`,
        cause: opts.auditCause,
        at: now(),
      });
    }
    logger.warn({ connId, code, cause, ip }, "socket dropped");
    try {
      ws.close(code, cause);
    } catch {
      ws.terminate();
    }
  };

  const roomEvents: RoomManagerEvents = {
    onPersisted: (roomKey, resourceId, version, logSeqAtFlush, blockDelta) => {
      // Trim the update log to post-flush entries (R5): everything up to
      // the flush snapshot's watermark is persisted; keep the rest.
      void updateLog.trimAfter(roomKey, logSeqAtFlush).catch((err) => {
        logger.warn({ err: String(err), roomKey }, "log trim failed");
      });
      const room = manager.get(roomKey);
      if (room !== undefined) {
        const text = encodeServerFrame({
          type: "doc.persisted",
          pageId: resourceId,
          version,
        });
        for (const [connId] of room.peers) {
          const ws = findWsByConn(connId);
          if (ws !== null && ws.readyState === 1) ws.send(text);
        }
      }
      // TASK-0092 R3: emit `page.updated` (the Phase-3 event shape) with
      // the ADDITIVE `blockDelta` counts — the Phase-8 indexer + the
      // computed-recompute consumer listen. The flush is a system write,
      // attributed `actor: 'realtime-flush'` (auditable, not impersonable
      // — the public API has no path to set that actor).
      const delta = blockDeltaCounts(blockDelta);
      events?.publish(
        createDomainEvent(
          DomainEventName.PAGE_UPDATED,
          {
            pageId: resourceId,
            version,
            blockDelta: delta,
            ops: blockDelta,
          },
          {
            workspaceId: workspaceIdByRoom.get(roomKey),
            actorId: "realtime-flush",
          },
        ),
      );
      metrics.inc("events_processed_total", { type: "persist" });
      logger.info({ roomKey, resourceId, version, delta }, "doc persisted");
    },
    onRebased: (roomKey, resourceId, appliedExternalOps) => {
      // The flush-rebase ran (a concurrent REST write was absorbed into the
      // doc) — observable for the rebase integration test + ops.
      metrics.inc("events_processed_total", { type: "rebase" });
      logger.info(
        { roomKey, resourceId, appliedExternalOps },
        "doc rebased on external (REST) write",
      );
    },
    onRoomEvicted: (roomKey) => {
      logger.info({ roomKey }, "room evicted");
      logSeqByRoom.delete(roomKey);
      workspaceIdByRoom.delete(roomKey);
      for (const [ws, st] of [...states]) {
        if (st.docRoom !== null && st.docRoom.key === roomKey && !st.closed) {
          sendText(ws, {
            type: "error",
            code: "room_evicted",
            message: "page reloaded — reconnecting",
          });
          dropSocket(ws, WS_CLOSE.ProtocolOrHeartbeat, "room_evicted");
        }
      }
      manager.remove(roomKey);
      void updateLog.clear(roomKey);
    },
  };

  const manager = new RoomManager(
    {
      store,
      clock: {
        now,
        setTimeout: (fn, ms) => setTimeout(fn, ms),
        clearTimeout: (h) => clearTimeout(h as ReturnType<typeof setTimeout>),
      },
      roomGcMs,
      flushSeqProvider: (roomKey) => logSeqByRoom.get(roomKey) ?? 0,
    },
    roomEvents,
  );

  // --- the upgrade handshake (R1) --------------------------------------------------

  const onUpgrade = (req: http.IncomingMessage, socket: net.Socket, head: Buffer): void => {
    const url = req.url ?? "/";
    const pathname = url.split("?")[0];
    if (pathname !== "/realtime" && pathname !== "/realtime/") {
      socket.destroy();
      return;
    }
    if (draining) {
      socket.write("HTTP/1.1 503 Service Unavailable\r\n\r\n");
      socket.destroy();
      return;
    }
    const params = new URL(url, "http://localhost").searchParams;
    // The token: query (the client path) or the Authorization header
    // (proxy compatibility, R1). NEVER logged.
    const queryToken = params.get("token");
    const headerAuth = req.headers.authorization;
    const headerToken = headerAuth?.startsWith("Bearer ") ? headerAuth.slice(7).trim() : null;
    const token = queryToken ?? headerToken;
    const roomParam =
      params.get("room") ?? (req.headers["x-realtime-room"] as string | undefined) ?? null;

    const ip = clientIp(req, config.trustProxy);
    const ua = (req.headers["user-agent"] ?? "").slice(0, 200);

    // 1. connect rate (30/min/IP) — BEFORE auth (an unauthenticated
    //    flood must not burn the JWT path).
    const admit = connectLimiter.tryAdmit(ip);
    if (!admit.ok) {
      metrics.inc("ws_drops_total", { cause: "connect_rate_limited" });
      audit.publish("realtime.connection.abuse", { ip, cause: "connect_rate_limited", at: now() });
      socket.write("HTTP/1.1 429 Too Many Requests\r\n\r\n");
      socket.destroy();
      return;
    }

    // 2. JWT verify (4001 — no room join).
    const verified = verifyHandshakeToken(token, config.jwtSecret, now());
    if (!verified.ok) {
      metrics.inc("ws_handshake_failures_total");
      logger.warn({ reason: verified.reason, ip, ua }, "handshake: unauthorized");
      socket.write("HTTP/1.1 401 Unauthorized\r\n\r\n");
      socket.destroy();
      return;
    }
    const session = verified.session;

    // 3. parse the room key. A MALFORMED key → 400 bad_room (R1). A
    //    MISSING/EMPTY key is a ROOMLESS CONTROL connection (TASK-0092
    //    6b — the web RealtimeClient connects token-only, no ?room=, and
    //    opens `page:<id>` channels in-band via `subscribe`); the socket
    //    is admitted with `room: null` (no doc, no permission check —
    //    the token was already verified above).
    const room = parseRoomKey(roomParam ?? "");
    if (room === null && (roomParam ?? "").length > 0) {
      metrics.inc("ws_handshake_failures_total");
      logger.warn({ ip, ua, room: roomParam }, "handshake: bad_room");
      socket.write("HTTP/1.1 400 Bad Request\r\n\r\n");
      socket.destroy();
      return;
    }

    // 4. user cap (100 live sockets/user).
    const userSockets = socketsByUser.get(session.userId) ?? new Set<WebSocket>();
    if (userSockets.size >= REALTIME_LIMITS.maxConnsPerUser) {
      metrics.inc("ws_drops_total", { cause: "user_conn_cap" });
      audit.publish("realtime.connection.abuse", {
        userId: session.userId,
        ip,
        cause: "user_conn_cap",
        at: now(),
      });
      socket.write("HTTP/1.1 429 Too Many Requests\r\n\r\n");
      socket.destroy();
      return;
    }

    // 5. workspace + permission (async) → upgrade + join.
    void (async () => {
      try {
        let canEdit = false;
        let peerName: string | undefined;
        let resolvedWorkspaceId: string | undefined;
        if (room !== null) {
          // user: control rooms are authorized to the socket's own userId
          // ONLY (note 16 — joining user:{otherId} → 4002). No workspace
          // or resource check applies (pure fan-out, header note 4).
          if (room.kind === "user" && room.id !== session.userId) {
            logger.warn(
              { userId: session.userId, room: `user:${room.id}` },
              "handshake: forbidden control room",
            );
            socket.write("HTTP/1.1 403 Forbidden\r\n\r\n");
            socket.destroy();
            return;
          }
          if (room.kind !== "user") {
            const info = await meta.resolve({
              kind: room.kind,
              id: room.id,
              userId: session.userId,
            });
            if (info === null) {
              logger.warn(
                { userId: session.userId, room: `${room.kind}:${room.id}` },
                "handshake: unknown resource",
              );
              socket.write("HTTP/1.1 403 Forbidden\r\n\r\n");
              socket.destroy();
              return;
            }
            const workspaceId = info.workspaceId;
            if (!session.workspaces.includes(workspaceId)) {
              logger.warn(
                { userId: session.userId, workspaceId, room: `${room.kind}:${room.id}` },
                "handshake: wrong_workspace",
              );
              socket.write("HTTP/1.1 403 Forbidden\r\n\r\n");
              socket.destroy();
              return;
            }
            peerName = info.userName;
            resolvedWorkspaceId = workspaceId;
            // Doc rooms: VIEW to join, EDIT to write (the shared engine).
            if (room.kind === "page" || room.kind === "database") {
              const level = await permissions.resolveLevel({
                userId: session.userId,
                workspaceId,
                resourceType: room.kind,
                resourceId: room.id,
              });
              if (level === "none") {
                logger.warn(
                  { userId: session.userId, room: `${room.kind}:${room.id}` },
                  "handshake: no VIEW",
                );
                socket.write("HTTP/1.1 403 Forbidden\r\n\r\n");
                socket.destroy();
                return;
              }
              canEdit = level === "edit" || level === "full_access";
            }
            // comment:{pageId} rooms fan out over the page — the join
            // permission is the page's VIEW (the comment panel is part of
            // the page; comment writes are REST, TASK-0094).
            if (room.kind === "comment") {
              const level = await permissions.resolveLevel({
                userId: session.userId,
                workspaceId,
                resourceType: "page",
                resourceId: room.id,
              });
              if (level === "none") {
                logger.warn(
                  { userId: session.userId, room: `comment:${room.id}` },
                  "handshake: no VIEW",
                );
                socket.write("HTTP/1.1 403 Forbidden\r\n\r\n");
                socket.destroy();
                return;
              }
            }
          }
        }
        // Roomless CONTROL socket (6b): the token was verified in step 2;
        // no doc, no resource check — the page channels open in-band via
        // `subscribe` after the upgrade.
        if (wss === null) {
          socket.destroy();
          return;
        }
        wss.handleUpgrade(req, socket, head, (ws) => {
          void onConnection(ws, session, room, canEdit, peerName, ip, ua, resolvedWorkspaceId);
        });
      } catch (err) {
        logger.error({ err: String(err), ip }, "handshake failure");
        socket.destroy();
      }
    })();
  };

  // --- presence (TASK-0093, doc §5) ----------------------------------------------------

  /** The sockets that should receive a room's presence frames: every live
   *  doc-room socket in the room + every roomless control socket that has
   *  opened the room's `page:<id>`/`database:<id>` channel (the (b) web
   *  client — one token-only socket, many in-band channels). */
  const roomPresenceSubscribers = (roomKey: string): WebSocket[] => {
    const out: WebSocket[] = [];
    const inRoom = roomSockets.get(roomKey);
    if (inRoom !== undefined) for (const ws of inRoom) if (ws.readyState === 1) out.push(ws);
    for (const [ws, st] of states) {
      if (st.room !== null) continue;
      if (st.pageChannels.has(roomKey) && ws.readyState === 1) out.push(ws);
    }
    return out;
  };

  /** Resolve a socket's display name for presence stamping. Doc-room
   *  sockets carry it from the handshake meta (the page/database
   *  existence resolve returns `userName`); roomless CONTROL sockets
   *  resolve lazily once (the `meta` seam's `kind: "user"` path — the
   *  Prisma store maps it to the `users` table) and cache in
   *  `st.peerName`. Unresolvable → the shared fallback (the peer is
   *  still identified by name, never color-only — the task spec). */
  const resolvePeerName = async (st: SocketState): Promise<void> => {
    if (st.peerName !== undefined) return;
    const info = await meta.resolve({
      kind: "user",
      id: st.session.userId,
      userId: st.session.userId,
    });
    st.peerName = info?.userName;
  };

  /** Upsert a socket's own presence (the server stamps the identity — the
   *  client frame carries ONLY `cursor`/`viewing`) and fan the broadcast
   *  to the room's other subscribers (doc §5.1 best-effort; the > 50
   *  cap strips `cursor`/`viewing` — the tracker shapes the frame).
   *  Idempotent per (room, user): a second tab upserts the same key. */
  const applyPresence = (ws: WebSocket, st: SocketState, update: PresenceUpdate): void => {
    const room = presenceRoomFor(st);
    if (room === null) return;
    // TASK-0135: the spec's "span per ws room op" — presence fan-out is a
    // room op (join/rejoin re-broadcasts ride the same path). The sync span
    // is free when OTel is disabled (no-op) and adds no I/O.
    // workspaceId ABSENT (not "unknown") when the room has no recorded
    // tenant scope (contract shared with the API's span attributes).
    const workspaceId = workspaceIdByRoom.get(room);
    startSpan(
      "ws.presence.update",
      { room, userId: st.session.userId, ...(workspaceId !== undefined ? { workspaceId } : {}) },
      () => {
        const entry = presence.upsert(room, { id: st.session.userId, name: st.peerName }, update);
        const frame = presence.broadcastFor(room, entry);
        const text = encodeServerFrame(frame);
        for (const target of roomPresenceSubscribers(room)) {
          if (target === ws) continue; // the sender's own state is its echo — it knows it
          target.send(text);
        }
      }
    );
  };

  /** Broadcast one peer's leave (explicit `presence-leave`, a socket
   *  close, or the 90 s prune) — the removal signal (doc §5.2
   *  best-effort; a missed leave heals on the next join re-broadcast +
   *  the client's own prune window). */
  const announcePresenceLeave = (roomKey: string, userId: string): void => {
    // TASK-0135: span per ws room op (presence leave fan-out).
    const workspaceId = workspaceIdByRoom.get(roomKey);
    startSpan(
      "ws.presence.leave",
      { room: roomKey, userId, ...(workspaceId !== undefined ? { workspaceId } : {}) },
      () => {
        presence.remove(roomKey, userId);
        const frame = { type: "presence-leave", pageId: roomKey, userId } as const;
        const text = encodeServerFrame(frame);
        for (const ws of roomPresenceSubscribers(roomKey)) ws.send(text);
        logger.debug({ room: roomKey, userId }, "presence leave");
      }
    );
  };

  /** The doc room this socket's presence belongs to — page/database rooms
   *  only (comment/user rooms have no presence, doc §1). A roomless
   *  control socket's presence is the LAST page/database channel it
   *  subscribed (the (b) client switches pages by re-subscribing; the
   *  previous page's entry leaves with the `unsubscribe`-driven
   *  `presence-leave`). */
  const presenceRoomFor = (st: SocketState): string | null => {
    if (st.room !== null) {
      return st.room.kind === "page" || st.room.kind === "database"
        ? `${st.room.kind}:${st.room.id}`
        : null;
    }
    return st.lastPresenceRoom ?? null;
  };

  // --- a live socket -----------------------------------------------------------------

  const onConnection = async (
    ws: WebSocket,
    session: SessionIdentity,
    room: ParsedRoom | null,
    canEdit: boolean,
    peerName: string | undefined,
    ip: string,
    ua: string,
    resolvedWorkspaceId: string | undefined,
  ): Promise<void> => {
    const connId = randomUUID().replace(/-/g, "").slice(0, 16);
    const st: SocketState = {
      connId,
      session,
      room,
      pageChannels: new Set(),
      docRoom: null,
      ip,
      canEdit,
      strikes: new StrikeTracker({ now }),
      lastActivity: now(),
      lastAckedSeq: 0,
      closed: false,
      dropped: false,
    };
    states.set(ws, st);
    const userSockets = socketsByUser.get(session.userId) ?? new Set<WebSocket>();
    userSockets.add(ws);
    socketsByUser.set(session.userId, userSockets);
    metrics.inc("events_processed_total", { type: "connection" });
    // A roomless control socket has no doc room to count (6b).
    if (room !== null) {
      metrics.set("ws_connections", countConnectionsByRoom(room.kind), { room: room.kind });
    }
    // TASK-0135: the spec's `ws_active_connections` gauge — the PROCESS-wide
    // total (states.size; the local per-room `ws_connections` is kept below
    // it, the spec asks for the global gauge). Free when OTel is disabled.
    m.ws.setConnections(states.size);
    logger.info(
      {
        connId,
        userId: session.userId,
        room: room === null ? "control" : `${room.kind}:${room.id}`,
        ip,
        ua,
      },
      "connected",
    );

    // The control room rides the same socket (header note 4 / note 16):
    // ≤ 10 live control sockets per user — the OLDEST is evicted with a
    // `control_evicted` error frame + close 4008.
    const userArr = userControlSockets.get(session.userId) ?? [];
    userArr.push(ws);
    if (userArr.length > REALTIME_LIMITS.maxControlSocketsPerUser) {
      const oldest = userArr.shift()!;
      if (oldest !== ws) {
        sendText(oldest, {
          type: "error",
          code: "control_evicted",
          message: "too many tabs — reconnect this tab",
        });
        dropSocket(oldest, WS_CLOSE.RateOrPolicy, "control_evicted", {
          auditCause: "control_evicted",
        });
      }
    }
    userControlSockets.set(session.userId, userArr);

    try {
      if (room !== null && (room.kind === "page" || room.kind === "database")) {
        const key = `${room.kind}:${room.id}`;
        const docRoom = await manager.getOrCreate(key, room.kind, room.id);
        // The room's workspace (for the flush's `page.updated` event) —
        // recorded at join from the handshake meta (the room key carries
        // no tenant scope, note 14).
        if (resolvedWorkspaceId !== undefined) workspaceIdByRoom.set(key, resolvedWorkspaceId);
        const peer = makePeer(connId, session.userId, canEdit, peerName);
        docRoom.peers.set(connId, peer);
        docRoom.notePeersChanged();
        st.docRoom = docRoom;
        const inRoom = roomSockets.get(key) ?? new Set<WebSocket>();
        inRoom.add(ws);
        roomSockets.set(key, inRoom);
        sendText(ws, {
          type: "hello",
          connId,
          canEdit,
          serverTime: now(),
          roomPeers: docRoom.roster().filter((p) => p.user.id !== session.userId),
        });
        // Presence (TASK-0093, doc §5.1): the join re-broadcast — a late
        // joiner hears every existing peer's full state (cursors
        // included; the deterministic catch-up), and the existing peers
        // hear the joiner's initial state (no cursor, viewing=true —
        // the avatar appears, the cursor shows when the joiner's first
        // `presence` frame lands). Idempotent per (room, user): a second
        // tab of the same user upserts the same presence key.
        const joined = presence.upsert(key, { id: session.userId, name: peerName }, {});
        for (const target of roomPresenceSubscribers(key)) {
          if (target === ws) continue;
          sendText(target, presence.broadcastFor(key, joined));
        }
        for (const other of presence.peers(key)) {
          if (other.userId === session.userId) continue;
          sendText(ws, presence.broadcastFor(key, other));
        }
        logger.debug(
          { connId, room: key, peers: presence.count(key) },
          "presence join re-broadcast",
        );
      } else {
        // comment / user rooms: no doc — hello with the user's other
        // tabs as peers (the control room's roster).
        sendText(ws, {
          type: "hello",
          connId,
          canEdit: false,
          serverTime: now(),
          roomPeers: userArr
            .filter((o) => o !== ws)
            .map((o, i) => {
              const os = states.get(o);
              return {
                user: {
                  id: os?.connId ?? `tab-${i}`,
                  name: os !== undefined ? "Guest (shared)" : "tab",
                  color: "slate",
                },
              };
            }),
        });
      }
    } catch (err) {
      logger.error({ err: String(err), connId }, "join failure");
      dropSocket(ws, WS_CLOSE.ProtocolOrHeartbeat, "join_failed");
      return;
    }

    // --- inbound frames (R3) -------------------------------------------------------

    ws.on("message", (data: Buffer | ArrayBuffer | Buffer[], isBinary: boolean) => {
      const stNow = states.get(ws);
      if (stNow === undefined || stNow.closed) return;
      stNow.lastActivity = now();

      // Per-socket sustained rate (100 msg/s → 4008 + audit), keyed by connId.
      const rate = msgRate.consume(stNow.connId);
      if (!rate.ok) {
        dropSocket(ws, WS_CLOSE.RateOrPolicy, "rate_limited", { auditCause: "msg_rate_limited" });
        return;
      }

      // The `isBinary` flag is the authoritative discriminator (the ws
      // library delivers TEXT frames as Buffers too — `typeof data` would
      // parse every JSON control frame as binary and strike it as
      // `bad_binary_kind`; the flag routes text → JSON, binary → 0x01/0x02).
      const asBytes = (): Uint8Array =>
        Buffer.isBuffer(data)
          ? new Uint8Array(data)
          : Array.isArray(data)
            ? new Uint8Array(Buffer.concat(data))
            : new Uint8Array(data as ArrayBuffer);
      const raw: Uint8Array | string = isBinary
        ? asBytes()
        : typeof data === "string"
          ? data
          : Buffer.from(asBytes()).toString("utf8");

      if (typeof raw !== "string" && raw.byteLength > MAX_FRAME_BYTES) {
        dropSocket(ws, WS_CLOSE.PayloadTooLarge, "payload_too_large", {
          auditCause: "frame_too_large",
        });
        return;
      }

      const decoded = decodeClientFrame(raw);
      if (!decoded.ok) {
        const strike = stNow.strikes.record();
        metrics.inc("ws_drops_total", { cause: `strike:${decoded.code}` });
        sendText(ws, { type: "error", code: decoded.code, message: "frame rejected" });
        if (strike.violated) {
          dropSocket(ws, WS_CLOSE.RateOrPolicy, "strike_limit", {
            auditCause: `strikes:${decoded.code}`,
          });
        }
        return;
      }

      if (decoded.frame.kind === "text") {
        metrics.inc("ws_messages_total", { frame: decoded.frame.frame.type });
        switch (decoded.frame.frame.type) {
          case "sync.request": {
            void handleSyncRequest(ws, stNow, decoded.frame.frame);
            break;
          }
          case "ping": {
            if (decoded.frame.frame.lastSeq !== undefined) {
              stNow.lastAckedSeq = decoded.frame.frame.lastSeq;
            }
            sendText(ws, { type: "ping-pong", ts: decoded.frame.frame.ts });
            break;
          }
          case "outbox.pending": {
            logger.debug(
              { connId: stNow.connId, count: decoded.frame.frame.count },
              "outbox.pending",
            );
            break;
          }
          case "subscribe": {
            // Control-channel join (6b): a roomless control socket opens a
            // `page:<id>` (or `workspace:<id>`) channel in-band. Idempotent
            // (reconnect re-subscribes). A doc-room socket ignores it (its
            // room is the handshake's; control frames ride control sockets).
            if (stNow.room === null) {
              const channel = decoded.frame.frame.channel;
              stNow.pageChannels.add(channel);
              // Presence room tracking (TASK-0093): a page/database channel
              // becomes the socket's presence room (the LAST one wins — the
              // (b) client re-subscribes when it switches pages; the
              // previous page's entry leaves with the `unsubscribe`).
              if (channel.startsWith("page:") || channel.startsWith("database:")) {
                if (stNow.lastPresenceRoom !== undefined && stNow.lastPresenceRoom !== channel) {
                  announcePresenceLeave(stNow.lastPresenceRoom, stNow.session.userId);
                }
                stNow.lastPresenceRoom = channel;
                // Join re-broadcast (doc §5.1): a late joiner hears every
                // existing peer's full state (cursors included) — the
                // deterministic catch-up — and the existing peers hear the
                // joiner's initial state (no cursor, viewing=true). The
                // name is resolved lazily for control sockets (no
                // handshake meta for a roomless socket).
                void (async () => {
                  await resolvePeerName(stNow);
                  const room = stNow.lastPresenceRoom;
                  if (room === undefined) return;
                  const joined = presence.upsert(
                    room,
                    { id: stNow.session.userId, name: stNow.peerName },
                    {},
                  );
                  for (const target of roomPresenceSubscribers(room)) {
                    if (target === ws) continue;
                    sendText(target, presence.broadcastFor(room, joined));
                  }
                  for (const peer of presence.peers(room)) {
                    if (peer.userId === stNow.session.userId) continue;
                    sendText(ws, presence.broadcastFor(room, peer));
                  }
                  logger.debug(
                    { connId: stNow.connId, channel, peers: presence.count(room) },
                    "presence join re-broadcast",
                  );
                })();
              }
            }
            logger.debug(
              { connId: stNow.connId, channel: decoded.frame.frame.channel },
              "control subscribe",
            );
            break;
          }
          case "unsubscribe": {
            if (stNow.room === null) {
              const channel = decoded.frame.frame.channel;
              stNow.pageChannels.delete(channel);
              // Presence room tracking: unsubscribing the socket's current
              // presence room leaves that room's roster (the (b) client
              // leaves a page by unsubscribing its channel).
              if (stNow.lastPresenceRoom === channel) {
                stNow.lastPresenceRoom = undefined;
                announcePresenceLeave(channel, stNow.session.userId);
              }
            }
            logger.debug(
              { connId: stNow.connId, channel: decoded.frame.frame.channel },
              "control unsubscribe",
            );
            break;
          }
          case "presence": {
            // Presence upsert (TASK-0093, doc §5): the client carries ONLY
            // `cursor`/`viewing` (the wire schema forbids identity). The
            // server stamps `user {id, name, color}` (id from the
            // handshake, name from the meta seam, color from the shared
            // `presenceColor` hash) and fans the broadcast out. A control
            // socket must have a page/database presence room (subscribed)
            // or the frame is a no-op (it has no room to state).
            const presenceUpdate: PresenceUpdate = {
              cursor: decoded.frame.frame.cursor,
              viewing: decoded.frame.frame.viewing,
            };
            if (stNow.room === null) {
              void (async () => {
                await resolvePeerName(stNow);
                if (stNow.lastPresenceRoom === undefined) return; // no room yet
                applyPresence(ws, stNow, presenceUpdate);
              })();
            } else {
              applyPresence(ws, stNow, presenceUpdate);
            }
            break;
          }
          case "presence-leave": {
            // Explicit leave (TASK-0093, doc §5): the client says it's
            // going (tab blur / page switch / socket close is implicit via
            // the `close` handler). Drops the peer + announces.
            const room = presenceRoomFor(stNow);
            if (room !== null) {
              announcePresenceLeave(room, stNow.session.userId);
            }
            break;
          }
        }
        return;
      }

      if (decoded.frame.binary === "update") {
        metrics.inc("ws_messages_total", { frame: "update" });
        void handleUpdate(ws, stNow, decoded.frame.bytes);
        return;
      }
      metrics.inc("ws_messages_total", { frame: "awareness" });
      handleAwareness(ws, stNow, decoded.frame.bytes);
    });

    ws.on("pong", () => {
      const stNow = states.get(ws);
      if (stNow !== undefined) stNow.lastActivity = now();
    });

    ws.on("close", () => onSocketClose(ws));
    ws.on("error", (err) => {
      const msg = err instanceof Error ? err.message : String(err);
      if (/payload|1009|max/i.test(msg)) {
        dropSocket(ws, WS_CLOSE.PayloadTooLarge, "payload_too_large", {
          auditCause: "frame_too_large",
        });
        return;
      }
      logger.debug({ connId: st.connId, err: msg }, "socket error");
      onSocketClose(ws);
    });
  };

  // --- sync.request (R5) --------------------------------------------------------------

  const handleSyncRequest = async (
    ws: WebSocket,
    st: SocketState,
    frame: { stateVector?: string; lastSeq?: number },
  ): Promise<void> => {
    const room = st.docRoom;
    if (room === null) {
      // comment/user/control rooms have no doc — nothing to sync.
      sendText(ws, { type: "sync.state", chunk: "", done: true });
      return;
    }
    const key = room.key;
    const svB64 = frame.stateVector;
    // R5 fast path: the update log covers lastSeq+1 → replay the logged
    // updates (cheap — no state-vector encode of a large doc).
    if (frame.lastSeq !== undefined) {
      const entries = await updateLog.entriesAfter(key, frame.lastSeq);
      if (entries !== null) {
        for (const e of entries) {
          sendBinary(ws, encodeBinaryUpdate(Buffer.from(e.updateB64, "base64")));
        }
        const last = entries[entries.length - 1];
        if (last !== undefined) {
          sendText(ws, { type: "ack", seq: last.seq });
        }
        // The logged updates may NOT cover the client's full state for a
        // fresh client — always finish with the state-vector delta so the
        // client converges (idempotent merge, R5 invariant).
        await sendStateDelta(ws, room, svB64);
        return;
      }
      // Log gap (or a lastSeq from another room's log — the log is
      // room-scoped, Security section) → state-vector fallback.
      logger.debug(
        { connId: st.connId, key, lastSeq: frame.lastSeq },
        "resume: log gap → state vector",
      );
    }
    await sendStateDelta(ws, room, svB64);
  };

  const sendStateDelta = async (
    ws: WebSocket,
    room: Room,
    svB64: string | undefined,
  ): Promise<void> => {
    const sv = svB64 !== undefined && svB64.length > 0 ? Buffer.from(svB64, "base64") : null;
    const delta = room.stateDelta(sv);
    // Paged ≤ 64 KB chunks (R3): base64 expands 3 bytes → 4 chars, so a
    // 48 KB raw chunk stays ≤ 64 KB on the wire.
    const CHUNK = Math.floor(REALTIME_LIMITS.maxSyncChunkBytes * 0.75);
    if (delta.byteLength === 0) {
      sendText(ws, { type: "sync.state", chunk: "", done: true });
      return;
    }
    for (let off = 0; off < delta.byteLength; off += CHUNK) {
      const slice = delta.subarray(off, Math.min(off + CHUNK, delta.byteLength));
      const done = off + CHUNK >= delta.byteLength;
      const frame: SyncStateFrame = {
        type: "sync.state",
        chunk: Buffer.from(slice).toString("base64"),
        done,
      };
      sendText(ws, frame);
    }
  };

  // --- binary update (R3) ---------------------------------------------------------------

  const handleUpdate = async (ws: WebSocket, st: SocketState, bytes: Uint8Array): Promise<void> => {
    const room = st.docRoom;
    if (room === null) {
      strikeFor(ws, st, "bad_update");
      return;
    }
    // Defense in depth (R1): the canEdit flag is re-checked per frame —
    // a VIEW-only peer's update is dropped with `error FORBIDDEN`
    // (the guard AND the per-socket flag — both).
    if (!st.canEdit) {
      metrics.inc("ws_drops_total", { cause: "view_write_dropped" });
      sendText(ws, { type: "error", code: "FORBIDDEN", message: "this room is read-only for you" });
      logger.warn({ connId: st.connId, room: room.key }, "update dropped: VIEW-only peer");
      return;
    }
    // Validate the Yjs update on a throwaway doc (a corrupt frame →
    // `bad_update` strike; R3).
    const probe = new Y.Doc();
    try {
      Y.applyUpdate(probe, bytes, "remote");
    } catch {
      probe.destroy();
      strikeFor(ws, st, "bad_update");
      return;
    }
    probe.destroy();

    try {
      const applied = room.applyUpdate(bytes);
      const key = room.key;
      if (!applied) {
        // Buffered during a re-hydration (doc §4.2) — applied in order by
        // the room's `completeHydration`. It must still be logged + acked
        // NOW (its seq is part of the room's log continuity; a client's
        // lastSeq accounting assumes every applied update got a seq).
        const b64 = Buffer.from(bytes).toString("base64");
        const seq = await updateLog.append(key, b64, room.lastVersion);
        logSeqByRoom.set(key, Math.max(logSeqByRoom.get(key) ?? 0, seq));
        st.lastAckedSeq = Math.max(st.lastAckedSeq, seq);
        sendText(ws, { type: "ack", seq });
        return;
      }
      const b64 = Buffer.from(bytes).toString("base64");
      const seq = await updateLog.append(key, b64, room.lastVersion);
      logSeqByRoom.set(key, Math.max(logSeqByRoom.get(key) ?? 0, seq));
      st.lastAckedSeq = Math.max(st.lastAckedSeq, seq);
      // Relay to the OTHER peers (binary 0x01) — the spec's "span per ws
      // room op" (a broadcast fan-out; TASK-0135).
      const workspaceId = workspaceIdByRoom.get(key);
      startSpan(
        "ws.update.broadcast",
        { room: key, peers: room.peers.size, ...(workspaceId !== undefined ? { workspaceId } : {}) },
        () => {
          const relay = encodeBinaryUpdate(bytes);
          for (const [connId] of room.peers) {
            if (connId === st.connId) continue;
            const peerWs = findWsByConn(connId);
            if (peerWs !== null) sendBinary(peerWs, relay);
          }
        }
      );
      // Ack the log seq (the client's resume fast path).
      sendText(ws, { type: "ack", seq });
    } catch (err) {
      logger.error({ err: String(err), connId: st.connId }, "applyUpdate failed");
      strikeFor(ws, st, "bad_update");
    }
  };

  const strikeFor = (
    ws: WebSocket,
    st: SocketState,
    code: "bad_update" | "unknown_frame",
  ): void => {
    const strike = st.strikes.record();
    metrics.inc("ws_drops_total", { cause: `strike:${code}` });
    sendText(ws, { type: "error", code, message: "frame rejected" });
    if (strike.violated) {
      dropSocket(ws, WS_CLOSE.RateOrPolicy, "strike_limit", { auditCause: `strikes:${code}` });
    }
  };

  // --- awareness (R4 fan-out) -------------------------------------------------------------

  const handleAwareness = (ws: WebSocket, st: SocketState, bytes: Uint8Array): void => {
    const room = st.docRoom;
    if (room === null) return; // no awareness on control rooms
    const relay = encodeBinaryAwareness(bytes);
    const peers = room.peers;
    // doc §5.1: presence fan-out is capped at 50 peers/room — beyond
    // the cap the relay goes to the NEWEST 50 peers (the client renders
    // avatars-only past the cap; the cursor-strip is client-side per the
    // frozen contract — the server's cap is the fan-out limit, recorded).
    const cap = REALTIME_LIMITS.awarenessPeerCap;
    const targets = peers.size > cap ? [...peers.keys()].slice(-cap) : [...peers.keys()];
    for (const connId of targets) {
      if (connId === st.connId) continue;
      const peerWs = findWsByConn(connId);
      if (peerWs !== null) sendBinary(peerWs, relay);
    }
  };

  // --- close / leave -------------------------------------------------------------------------

  const onSocketClose = (ws: WebSocket): void => {
    const st = states.get(ws);
    if (st === undefined) return;
    st.closed = true;
    states.delete(ws);
    m.ws.setConnections(states.size); // TASK-0135 — connection gauge
    msgRate.drop(st.connId);
    const userSet = socketsByUser.get(st.session.userId);
    if (userSet !== undefined) {
      userSet.delete(ws);
      if (userSet.size === 0) socketsByUser.delete(st.session.userId);
    }
    const controlArr = userControlSockets.get(st.session.userId);
    if (controlArr !== undefined) {
      const i = controlArr.indexOf(ws);
      if (i >= 0) controlArr.splice(i, 1);
      if (controlArr.length === 0) userControlSockets.delete(st.session.userId);
    }
    if (st.docRoom !== null) {
      st.docRoom.peers.delete(st.connId);
      st.docRoom.notePeersChanged();
      // Presence (TASK-0093): a doc-room socket's leave is implicit (the
      // roster entry dies with the socket) — drop the socket from the
      // room's presence fan-out set + announce the (room, user) leave,
      // but ONLY if no OTHER live socket of the same user is still in the
      // room (a second tab keeps the avatar — the presence key is shared
      // per (room, user)).
      const roomKey = st.docRoom.key;
      const inRoom = roomSockets.get(roomKey);
      if (inRoom !== undefined) {
        inRoom.delete(ws);
        if (inRoom.size === 0) roomSockets.delete(roomKey);
      }
      let otherTabInRoom = false;
      for (const [otherWs, other] of states) {
        if (
          otherWs !== ws &&
          other.docRoom !== null &&
          other.docRoom.key === roomKey &&
          other.session.userId === st.session.userId
        ) {
          otherTabInRoom = true;
          break;
        }
      }
      if (!otherTabInRoom) announcePresenceLeave(roomKey, st.session.userId);
    } else if (st.lastPresenceRoom !== undefined) {
      // Roomless control socket: its tracked presence room leaves with the
      // socket (the (b) client's page channel dies with the ws).
      announcePresenceLeave(st.lastPresenceRoom, st.session.userId);
      st.lastPresenceRoom = undefined;
    }
    // Drop the control channels (6b) — no control socket outlives its ws.
    st.pageChannels.clear();
    if (st.room !== null) {
      metrics.set("ws_connections", countConnectionsByRoom(st.room.kind), { room: st.room.kind });
    }
    logger.info(
      {
        connId: st.connId,
        userId: st.session.userId,
        room: st.room === null ? "control" : `${st.room.kind}:${st.room.id}`,
      },
      "disconnected",
    );
  };

  const countConnectionsByRoom = (kind: string): number => {
    let n = 0;
    for (const st of states.values()) {
      if (st.room !== null && st.room.kind === kind) n += 1;
    }
    return n;
  };

  // --- TASK-0094 seams (fan-out) -------------------------------------------------------------

  const publishCommentEvent = (pageId: string, event: CommentEventFrame): void => {
    const text = encodeServerFrame(event);
    const channel = `page:${pageId}`;
    for (const [ws, st] of states) {
      if (ws.readyState !== 1) continue;
      // Two fan-out targets (the (b) web architecture, recorded in the task
      // file):
      //   1. `comment:{pageId}` doc-room sockets (the `?room=` handshake
      //      kind — a dedicated comment-panel socket),
      //   2. roomless control sockets that opened the page's `page:{pageId}`
      //      channel in-band (the web `RealtimeClient` — it connects
      //      token-only and never opens a `comment:` room).
      const inCommentRoom = st.room !== null && st.room.kind === "comment" && st.room.id === pageId;
      const hasPageChannel = st.room === null && st.pageChannels.has(channel);
      if (inCommentRoom || hasPageChannel) {
        ws.send(text);
      }
    }
  };

  /** Push a `block:changed` refetch-hint to every roomless control socket
   *  that opened the page's `page:<id>` channel (6b). A doc-room socket
   *  gets the change through its own CRDT relay (the flush rebase), so
   *  the hint targets control sockets only — no double-signal. */
  const publishBlockChanged = (frame: BlockChangedFrame): void => {
    const channel = `page:${frame.pageId}`;
    const text = encodeServerFrame(frame);
    for (const [ws, st] of states) {
      if (st.room === null && st.pageChannels.has(channel) && ws.readyState === 1) {
        ws.send(text);
      }
    }
  };

  // --- TASK-0094: the domain-event control source (the (b) producer,
  //   generalized from TASK-0092 6b) ---
  // One stream, one poller. The injected FACTORY hands back the live source
  // (production = the `factory:events` XREAD poll, index.ts; tests = a fake
  // the test drives). Its sink routes EVERY event to the matching fan-out:
  //   - `page.updated`         → `block:changed` refetch-hint (6b)
  //   - `comment.*`            → `comment.event` room frames (R1)
  //   - `notification.created` → `notification.push` control-room push (R7)
  //   - anything else          → dropped (pure fan-out; the worker owns the
  //                              other consumers of the same stream).
  // When `domainEventSource` is provided it SUPPLANTS `pageUpdatedSource`
  // (the legacy single-purpose seam, kept only so pre-TASK-0094 tests keep
  // passing without edits). Omitted → no control fan-out (dev mode).
  const routeDomainEvent = (event: DomainEvent): void => {
    const blockFrame = blockChangedForPageUpdated(event);
    if (blockFrame !== null) {
      publishBlockChanged(blockFrame);
      return;
    }
    const commentFrame = commentEventFrameFor(event);
    if (commentFrame !== null) {
      publishCommentEvent(commentFrame.pageId, commentFrame);
      return;
    }
    const push = notificationPushFor(event);
    if (push !== null) {
      publishNotification(push.userId, push.frame);
      return;
    }
    // Not a control fan-out event (search-index / webhook / automation …) —
    // the worker consumes those; the realtime app drops them.
  };

  const domainEventSourceImpl: DomainEventSource | null = deps.domainEventSource
    ? deps.domainEventSource(routeDomainEvent)
    : null;
  const pageUpdatedSourceImpl: PageUpdatedSource | null =
    domainEventSourceImpl !== null || deps.pageUpdatedSource === undefined
      ? null
      : deps.pageUpdatedSource(routeDomainEvent);

  const publishNotification = (userId: string, event: NotificationPushFrame): void => {
    const text = encodeServerFrame(event);
    const arr = userControlSockets.get(userId);
    if (arr === undefined) return;
    for (const ws of arr) {
      if (ws.readyState === 1) ws.send(text);
    }
  };

  // --- heartbeat (R4) --------------------------------------------------------------------------

  const startHeartbeat = (): void => {
    heartbeat = setInterval(() => {
      const t = now();
      for (const ws of wss?.clients ?? []) {
        const st = states.get(ws);
        if (st === undefined || st.closed) continue;
        if (ws.readyState === 1) {
          try {
            ws.ping();
          } catch {
            /* closed mid-ping */
          }
        }
        if (t - st.lastActivity > pruneSilenceMs) {
          dropSocket(ws, WS_CLOSE.ProtocolOrHeartbeat, "heartbeat_timeout");
        }
      }
    }, pingIntervalMs);
    heartbeat.unref?.();
    // Presence prune (doc §5.1: 90 s silence → drop; the R4 socket prune
    // at 90 s already disconnects a silent socket — this drops the
    // PRESENCE entry for a user whose last tab left without a clean
    // `presence-leave`, and announces the removal). Runs on the same
    // cadence as the heartbeat; the threshold is the doc's
    // `awarenessPruneMs` (injectable for the deterministic prune test).
    presencePrune = setInterval(() => {
      const pruned = presence.prune();
      for (const [roomKey, entries] of pruned) {
        for (const entry of entries) announcePresenceLeave(roomKey, entry.userId);
        logger.debug({ room: roomKey, pruned: entries.length }, "presence prune");
      }
    }, pingIntervalMs);
    presencePrune.unref?.();
  };

  // --- health / readyz / metrics (R6) ----------------------------------------------------------

  const healthHandler = (req: http.IncomingMessage, res: http.ServerResponse): void => {
    const url = (req.url ?? "").split("?")[0];
    if (req.method !== "GET") {
      res.writeHead(405).end();
      return;
    }
    if (url === "/health" || url === "/health/") {
      res.writeHead(200, { "content-type": "application/json" }).end(
        JSON.stringify({
          status: "ok",
          connections: states.size,
          rooms: manager.rooms().length,
          uptime: Math.round((now() - startedAt) / 1000),
          // RSS in KB (the load baseline's "memory < 2 GB" reads this —
          // the probe port has no other way to see the process heap).
          memoryRss: Math.round(process.memoryUsage().rss / 1024),
        }),
      );
      return;
    }
    // Legacy scaffold path (the Phase-0 compose probes it — note 15).
    if (url === "/health/ready") {
      void readyz(res);
      return;
    }
    if (url === "/readyz" || url === "/readyz/") {
      void readyz(res);
      return;
    }
    if (url === "/metrics" || url === "/metrics/") {
      // TASK-0135: MERGE the OTel exposition (target_info + the spec's
      // `ws_active_connections`) with the local registry (per-room
      // `ws_connections{room}`, `ws_messages_total{frame}`, drops, …). The
      // local registry stays authoritative for its series — when OTel is ON
      // we DROP its overlapping `ws_messages_total` (no label-set collision
      // is legal in one exposition; the local {frame} form is richer) and
      // `target_info` is OTel-only. Local-only → behavior unchanged (the
      // merge is a no-op on empty OTel text, which is what a disabled OTel
      // yields).
      void (async () => {
        const local = metrics.render();
        let otelText = "";
        try {
          otelText = await renderPrometheus();
        } catch {
          otelText = "";
        }
        if (!otelText) {
          res.writeHead(200, { "content-type": "text/plain; version=0.0.4" }).end(local);
          return;
        }
        const merged = mergeExposition(local, otelText);
        res.writeHead(200, { "content-type": "text/plain; version=0.0.4" }).end(merged);
      })();
      return;
    }
    res.writeHead(404).end();
  };

  const readyz = async (res: http.ServerResponse): Promise<void> => {
    const checks: Record<string, string> = { ws: draining ? "draining" : "ok" };
    let ok = true;
    const redisPing = (updateLog as { ping?: () => Promise<boolean> }).ping;
    if (config.useRedis && redisPing !== undefined) {
      try {
        const r = await redisPing.call(updateLog);
        checks.redis = r ? "ok" : "error";
        if (!r) ok = false;
      } catch {
        checks.redis = "error";
        ok = false;
      }
    }
    try {
      const d = await store.ping();
      checks.database = d ? "ok" : "error";
      if (!d) ok = false;
    } catch {
      checks.database = "error";
      ok = false;
    }
    res
      .writeHead(ok ? 200 : 503, { "content-type": "application/json" })
      .end(JSON.stringify({ status: ok ? "ready" : "not_ready", checks }));
  };

  // --- lifecycle ---------------------------------------------------------------------------------

  const listen = (server: http.Server, port: number, what: string): Promise<void> =>
    new Promise<void>((resolve, reject) => {
      const onErr = (err: Error): void => reject(err);
      server.once("error", onErr);
      server.listen(port, () => {
        server.removeListener("error", onErr);
        logger.info({ port, what }, "listening");
        resolve();
      });
    });

  const closeServer = (server: http.Server): Promise<void> =>
    new Promise<void>((resolve) => {
      server.close(() => resolve());
      server.closeAllConnections?.();
    });

  const start = async (): Promise<void> => {
    startedAt = now();
    const server = http.createServer(healthHandler);
    // The ws-layer maxPayload must sit ABOVE the app-level MAX_FRAME_BYTES
    // (256 KB): a frame in between reaches the message handler and is
    // closed with the app's 4009 payload_too_large (+ audit). The ws
    // layer's 1009 is the second-line backstop for frames above 300 KB.
    // With maxPayload == MAX_FRAME_BYTES the 1009 always won the race and
    // the 4009 path was unreachable (recorded TASK-0091 fix).
    const wsMaxPayload = MAX_FRAME_BYTES + 48 * 1024;
    const wsServer = new WebSocketServer({ noServer: true, maxPayload: wsMaxPayload });
    wss = wsServer;
    httpServer = server;
    server.on("upgrade", onUpgrade);
    await listen(server, config.port, "ws");

    // The second probe port (note 15 — the compose exposes 4002).
    const probe = http.createServer(healthHandler);
    probeServer = probe;
    await listen(probe, config.healthPort, "probe");

    startHeartbeat();

    // The control source (6b / TASK-0094) starts with the server; an event
    // before any control socket exists is a no-op (no channels).
    domainEventSourceImpl?.start();
    pageUpdatedSourceImpl?.start();
  };

  const stop = async (): Promise<void> => {
    draining = true;
    if (heartbeat !== undefined) {
      clearInterval(heartbeat);
      heartbeat = undefined;
    }
    if (presencePrune !== undefined) {
      clearInterval(presencePrune);
      presencePrune = undefined;
    }
    if (domainEventSourceImpl !== null) {
      await domainEventSourceImpl.stop();
    }
    if (pageUpdatedSourceImpl !== null) {
      await pageUpdatedSourceImpl.stop();
    }
    for (const ws of wss?.clients ?? []) {
      try {
        ws.close(1001, "server shutting down");
      } catch {
        /* already closed */
      }
    }
    if (httpServer !== null) {
      const s = httpServer;
      httpServer = null;
      await closeServer(s);
    }
    if (probeServer !== null) {
      const s = probeServer;
      probeServer = null;
      await closeServer(s);
    }
    if (wss !== null) {
      const w = wss;
      wss = null;
      await new Promise<void>((resolve) => w.close(() => resolve()));
    }
  };

  /** SIGTERM drain (R6): 4010 + retryAfterMs to all sockets, flush all
   *  dirty rooms (≤ 10 s), exit. */
  const drain = async (): Promise<void> => {
    draining = true;
    const notice = encodeServerFrame({
      type: "error",
      code: "server_shutdown",
      message: `retry after ${REALTIME_LIMITS.shutdownRetryAfterMs} ms`,
    });
    for (const ws of wss?.clients ?? []) {
      if (ws.readyState === 1) {
        ws.send(notice);
      }
      try {
        ws.close(WS_CLOSE.ServerShutdown, `retryAfterMs=${REALTIME_LIMITS.shutdownRetryAfterMs}`);
      } catch {
        /* already closed */
      }
    }
    const result = await manager.drainAll(REALTIME_LIMITS.drainMaxMs);
    logger.info({ ...result }, "drain complete");
    await stop();
  };

  return {
    start,
    stop,
    drain,
    publishCommentEvent,
    publishBlockChanged,
    publishNotification,
    sockets: (): readonly SocketState[] => [...states.values()],
    manager,
    metrics,
  };
}

// --- defaults (dev + tests; index.ts wires the production seams) -------------------------

function clientIp(req: http.IncomingMessage, trustProxy: boolean): string {
  if (trustProxy) {
    const fwd = req.headers["x-forwarded-for"];
    const first = Array.isArray(fwd) ? fwd[0] : fwd;
    if (first !== undefined && first !== null) {
      const head = first.split(",")[0];
      if (head !== undefined && head.trim().length > 0) return head.trim();
    }
  }
  return req.socket.remoteAddress ?? "unknown";
}

/** Dev default: no writes allowed (canEdit=false), joins allowed. The
 *  production resolver is `createPrismaPermissionResolver` (index.ts). */
function failClosedPermissions(logger: Logger): PermissionResolver {
  logger.debug({}, "realtime: using fail-closed permission resolver (no DB wired)");
  return {
    resolveLevel: async () => "view",
  };
}

/** Dev default: unknown resources (no DB wired) → 4002. The production
 *  resolver is `createPrismaMetaResolver` (index.ts). */
function failClosedMeta(logger: Logger): RoomMetaResolver {
  logger.debug({}, "realtime: using fail-closed meta resolver (no DB wired)");
  return {
    resolve: async () => null, // unknown resource → 4002 (fail closed)
  };
}