server.ts · Source

server.tsapps/realtime/src/server.tsraw ← src67,911 B← Back to the app
1/**
2 * The realtime server (TASK-0091) — the production WebSocket server of
3 * `docs/architecture/realtime-architecture.md` §1–2 + §4.5, with the R3
4 * wire contract (`packages/shared/realtime/frames.ts` + the
5 * `packages/collaboration` client mirror).
6 *
7 *  - R1 handshake: upgrade on `/realtime?token=<JWT>&room=<roomKey>`
8 *    (the `Authorization: Bearer` header path is kept for proxies).
9 *    Validation order: IP connect rate (30/min) → JWT (4001) →
10 *    room key (4000 bad_room) → user cap (100) → workspace
11 *    membership (4002) → room permission via the shared engine (4002)
12 *    → join → `hello {connId, canEdit, serverTime, roomPeers}`.
13 *  - R2 RoomManager: one authoritative Y.Doc per room (page/database),
14 *    comment rooms are fan-out (no doc), `user:{userId}` control rooms
15 *    ride the same socket (note 16, ≤ 10 live sockets/user).
16 *  - R3 frames: text JSON control frames + binary 0x01/0x02; corrupt
17 *    updates + unknown frames count strikes (10 in 10 s → 4008 + audit
18 *    `realtime.connection.abuse`).
19 *  - R4 heartbeat: server `ws.ping()` every 30 s; 90 s silence → 4000.
20 *  - R5 resume: state-vector delta (paged `sync.state` ≤ 64 KB) OR the
21 *    Redis update log (`lastSeq` fast path; log gap → state vector).
22 *  - R6 lifecycle: close codes 4000–4010, SIGTERM drain (4010 +
23 *    `retryAfterMs`, flush all dirty rooms, ≤ 10 s), `/health` +
24 *    `/readyz` + `/metrics` on BOTH ports (note 15).
25 *
26 * The persistence seam (note 17): `DocStore` — the production Prisma
27 * implementation lives in `./prisma-store.js` (wired by `index.ts`);
28 * the default is the in-memory store (dev + unit/integration tests).
29 * The permission/meta seams are the same shared engine + a Prisma
30 * assembly (`./prisma-store.js`); the defaults are fail-closed.
31 */
32 
33import http from "node:http";
34import type net from "node:net";
35import { randomUUID } from "node:crypto";
36import pino, { type Logger } from "pino";
37import * as Y from "yjs";
38import { WebSocketServer, type WebSocket } from "ws";
39 
40import { verifyHandshakeToken, type SessionIdentity } from "./auth.js";
41import type { RealtimeConfig } from "./config.js";
42import { Metrics, mergeExposition } from "./metrics.js";
43import {
44  encodeBinaryAwareness,
45  encodeBinaryUpdate,
46  encodeServerFrame,
47  decodeClientFrame,
48  MAX_FRAME_BYTES,
49  StrikeTracker,
50} from "./protocol.js";
51import { parseRoomKey, type ParsedRoom } from "./room-key.js";
52import {
53  InMemoryDocStore,
54  makePeer,
55  RoomManager,
56  type DocStore,
57  type Room,
58  type RoomManagerEvents,
59} from "./rooms.js";
60import { ConnectLimiter, TokenBucket } from "./rate-limit.js";
61import { PresenceTracker, type PresenceUpdate } from "./presence.js";
62import { InMemoryUpdateLog, type UpdateLog } from "./update-log.js";
63import {
64  REALTIME_LIMITS,
65  WS_CLOSE,
66  type BlockChangedFrame,
67  type CommentEventFrame,
68  type NotificationPushFrame,
69  type ServerFrame,
70  type SyncStateFrame,
71} from "@factory/shared/realtime/frames";
72import {
73  DomainEventName,
74  createDomainEvent,
75  type DomainEvent,
76  type BlockOperation,
77} from "@factory/shared";
78import { blockChangedForPageUpdated, type PageUpdatedSource } from "./page-updated-source.js";
79import type { DomainEventSource } from "./event-source.js";
80import { commentEventFrameFor } from "./comments.js";
81import { notificationPushFor } from "./notifications.js";
82import { m, renderPrometheus, startSpan } from "@factory/observability";
83 
84// --- injectable seams (tests + the TASK-0092 production wiring) -----------------
85 
86/**
87 * The permission resolver — the SAME engine as the API guard (the
88 * shared `@factory/permissions` per the architecture doc; the API's
89 * Prisma assembly is mirrored in `./prisma-store.js` — the conflict is
90 * recorded in the task file). `level` is the effective level.
91 */
92export interface PermissionResolver {
93  resolveLevel(input: {
94    userId: string;
95    workspaceId: string;
96    resourceType: "page" | "database";
97    resourceId: string;
98  }): Promise<"none" | "view" | "comment" | "edit" | "full_access">;
99}
100 
101/**
102 * Resolves a room's workspace (note 14 — the room key carries none; the
103 * server resolves it server-side) + the joining user's display name
104 * (the peer roster). `null` = unknown resource (4002 no-existence).
105 */
106export interface RoomMetaResolver {
107  resolve(input: {
108    kind: ParsedRoom["kind"];
109    id: string;
110    userId: string;
111  }): Promise<{ workspaceId: string; userName?: string } | null>;
112}
113 
114/** The audit/event seam — the api's audit consumer writes the row. */
115export interface AuditSink {
116  publish(event: "realtime.connection.abuse", payload: Record<string, unknown>): void;
117}
118 
119/**
120 * The domain-event seam (TASK-0092 R3): the flush emits `page.updated`
121 * (the Phase-3 event shape, the `blockDelta` field additive — the Phase-8
122 * indexer + the computed-recompute consumer both listen). Production wires
123 * this to the event bus (the same `factory:events` stream the API
124 * publishes; the realtime app has no Redis dependency of its own, so the
125 * seam is injected — the worker consumes the stream). The default (no
126 * seam) is a no-op, so the server stays testable without a bus.
127 */
128export interface EventSink {
129  publish(event: DomainEvent): void;
130}
131 
132export interface RealtimeServerDeps {
133  readonly logger?: Logger;
134  readonly store?: DocStore;
135  readonly log?: UpdateLog & { ping?: () => Promise<boolean> };
136  readonly permissions?: PermissionResolver;
137  readonly meta?: RoomMetaResolver;
138  readonly audit?: AuditSink;
139  /** The domain-event seam (the flush's `page.updated`, TASK-0092 R3). */
140  readonly events?: EventSink;
141  /** The `page.updated` CONTROL source (TASK-0092 6b): a FACTORY that,
142   *  given the server's `page.updated` sink, returns the source to poll
143   *  (production = the `factory:events` stream poll in index.ts; tests =
144   *  a fake the test drives). The server starts it in `start()` and
145   *  stops it in `stop()`. Omitted → no control fan-out. */
146  readonly pageUpdatedSource?: (onPageUpdated: (event: DomainEvent) => void) => PageUpdatedSource;
147  /** The GENERIC domain-event CONTROL source (TASK-0094): a FACTORY that,
148   *  given the server's domain-event sink, returns the source to poll
149   *  (production = the `factory:events` stream poll in index.ts; tests =
150   *  a fake the test drives). Routes EVERY event: `page.updated` →
151   *  `block:changed`, `comment.*` → `comment.event`,
152   *  `notification.created` → `notification.push`. When provided it
153   *  SUPPLANTS `pageUpdatedSource` (one stream, one poller — the legacy
154   *  seam is kept only so pre-TASK-0094 tests keep passing). Omitted → no
155   *  control fan-out. */
156  readonly domainEventSource?: (onEvent: (event: DomainEvent) => void) => DomainEventSource;
157  readonly now?: () => number;
158  /** Heartbeat cadence override (tests use short intervals). */
159  readonly pingIntervalMs?: number;
160  readonly pruneSilenceMs?: number;
161  /** Empty-room GC delay override (the R2 5 min warm room; the
162   *  integration test uses a short delay to exercise the evict path). */
163  readonly roomGcMs?: number;
164  /** Presence prune override (ms; default 90 s — the doc §5.1 awareness
165   *  timeout). The integration prune test uses a short window. */
166  readonly presencePruneMs?: number;
167}
168 
169interface SocketState {
170  connId: string;
171  session: SessionIdentity;
172  /** The joined room (`page:`/`database:`/`comment:`/`user:`) — NULL for
173   *  a roomless CONTROL socket (TASK-0092 6b: the web `RealtimeClient`
174   *  connects token-only, no `?room=`, and opens `page:<id>` channels
175   *  in-band via `subscribe`). */
176  room: ParsedRoom | null;
177  /** The control channels this socket has opened in-band
178   *  (`subscribe page:<id>` / `workspace:<id>`). Only non-empty on
179   *  roomless control sockets. */
180  pageChannels: Set<string>;
181  /** The doc room (page/database) or null (comment/user). */
182  docRoom: Room | null;
183  ip: string;
184  canEdit: boolean;
185  strikes: StrikeTracker;
186  lastActivity: number;
187  lastAckedSeq: number;
188  closed: boolean;
189  dropped: boolean;
190  /** The display name resolved at handshake (the presence identity — the
191   *  meta seam's `userName`; control sockets are named lazily on first
192   *  `presence` frame, then cached). */
193  peerName?: string;
194  /** The page/database channel this control socket's presence currently
195   *  tracks (roomless sockets only — the LAST subscribed
196   *  `page:<id>`/`database:<id>` channel; presence frames before any
197   *  such subscription are ignored). */
198  lastPresenceRoom?: string;
199}
200 
201export interface RealtimeServer {
202  start(): Promise<void>;
203  stop(): Promise<void>;
204  /** SIGTERM drain (R6): 4010 to all sockets, flush all rooms, ≤ 10 s. */
205  drain(): Promise<void>;
206  /** Push a comment event to a page's comment room (TASK-0094 seam). */
207  publishCommentEvent(pageId: string, event: CommentEventFrame): void;
208  /** Push a notification to a user's control room (TASK-0094 seam). */
209  publishNotification(userId: string, event: NotificationPushFrame): void;
210  /** Push a `block:changed` refetch-hint to every control socket that has
211   *  opened the page's `page:<id>` channel (TASK-0092 6b / the (b)
212   *  producer). Mirrors the TASK-0094 comment/notification fan-out seams;
213   *  no-op when no control socket has the channel. */
214  publishBlockChanged(frame: BlockChangedFrame): void;
215  /** Test/diagnostic: the live socket states. */
216  sockets(): readonly SocketState[];
217  readonly manager: RoomManager;
218  readonly metrics: Metrics;
219}
220 
221export function createRealtimeServer(
222  config: RealtimeConfig,
223  deps: RealtimeServerDeps = {},
224): RealtimeServer {
225  const logger = deps.logger ?? pino({ level: config.logLevel, enabled: false });
226  const now = deps.now ?? Date.now;
227  const pingIntervalMs = deps.pingIntervalMs ?? REALTIME_LIMITS.pingIntervalMs;
228  const pruneSilenceMs = deps.pruneSilenceMs ?? REALTIME_LIMITS.pruneSilenceMs;
229  const roomGcMs = deps.roomGcMs ?? REALTIME_LIMITS.roomGcMs; // R2: 5 min warm room
230  const presencePruneMs = deps.presencePruneMs ?? REALTIME_LIMITS.awarenessPruneMs; // doc §5.1: 90 s
231 
232  const metrics = new Metrics(now);
233  const store = deps.store ?? new InMemoryDocStore();
234  const updateLog = deps.log ?? new InMemoryUpdateLog();
235  const permissions = deps.permissions ?? failClosedPermissions(logger);
236  const meta = deps.meta ?? failClosedMeta(logger);
237  const audit = deps.audit ?? { publish: () => undefined };
238  const events = deps.events;
239 
240  const socketsByUser = new Map<string, Set<WebSocket>>();
241  const userControlSockets = new Map<string, WebSocket[]>();
242  const states = new Map<WebSocket, SocketState>();
243  /** The room's max logged update seq (the flush snapshot's trim
244   *  watermark, R5 — the trim basis is the LOG seq, not the page
245   *  version; the seq domain restarts per room lifetime). */
246  const logSeqByRoom = new Map<string, number>();
247  /** roomKey → the room's workspaceId (the `page.updated` event's
248   *  `workspaceId`, recorded at join from the handshake meta). */
249  const workspaceIdByRoom = new Map<string, string>();
250  const msgRate = new TokenBucket({
251    capacity: REALTIME_LIMITS.maxMsgsPerSocketPerSec,
252    rate: REALTIME_LIMITS.maxMsgsPerSocketPerSec,
253    now,
254  });
255  const connectLimiter = new ConnectLimiter({
256    windowMs: 60_000,
257    max: REALTIME_LIMITS.connectPerIpPerMin,
258    now,
259  });
260  /** Presence (TASK-0093, doc §5) — one room-scoped tracker for the
261   *  process. Peers are keyed (roomKey, userId): a user's two tabs in the
262   *  same room share one avatar; identity is server-stamped (the handshake
263   *  id + the meta seam's display name + the shared presenceColor hash). */
264  const presence = new PresenceTracker({ now, pruneAfterMs: presencePruneMs });
265  /** roomKey → the live doc-room sockets currently in that room (the
266   *  presence fan-out target; `comment`/`user` rooms are never tracked —
267   *  presence is page/database only, doc §1). Updated on join/close. */
268  const roomSockets = new Map<string, Set<WebSocket>>();
269 
270  let wss: WebSocketServer | null = null;
271  let httpServer: http.Server | null = null;
272  let probeServer: http.Server | null = null;
273  let heartbeat: ReturnType<typeof setInterval> | undefined;
274  let presencePrune: ReturnType<typeof setInterval> | undefined;
275  let draining = false;
276  let startedAt = 0;
277 
278  const findWsByConn = (connId: string): WebSocket | null => {
279    for (const [ws, st] of states) {
280      if (st.connId === connId) return ws;
281    }
282    return null;
283  };
284 
285  const sendText = (ws: WebSocket, frame: ServerFrame): void => {
286    if (ws.readyState === 1) ws.send(encodeServerFrame(frame));
287  };
288 
289  /** Count the flush's minimal ops into the `page.updated` `blockDelta`
290   *  shape (the spec's `{inserted, updated, moved, deleted}` counts). */
291  const blockDeltaCounts = (
292    ops: readonly BlockOperation[],
293  ): {
294    inserted: number;
295    updated: number;
296    moved: number;
297    deleted: number;
298  } => {
299    let inserted = 0;
300    let updated = 0;
301    let moved = 0;
302    let deleted = 0;
303    for (const op of ops) {
304      if (op.op === "insert") inserted += 1;
305      else if (op.op === "update") updated += 1;
306      else if (op.op === "move") moved += 1;
307      else if (op.op === "delete") deleted += 1;
308    }
309    return { inserted, updated, moved, deleted };
310  };
311 
312  const sendBinary = (ws: WebSocket, bytes: Uint8Array): void => {
313    if (ws.readyState === 1) ws.send(bytes);
314  };
315 
316  const dropSocket = (
317    ws: WebSocket,
318    code: number,
319    cause: string,
320    opts?: { auditCause?: string },
321  ): void => {
322    const st = states.get(ws);
323    if (st === undefined) {
324      // Defensive: no state (already cleaned up) — count + close only.
325      metrics.inc("ws_drops_total", { cause });
326      try {
327        ws.close(code, cause);
328      } catch {
329        ws.terminate();
330      }
331      return;
332    }
333    if (st.dropped) return;
334    st.dropped = true;
335    const { connId, ip } = st;
336    const userId = st.session.userId;
337    states.delete(ws);
338    m.ws.setConnections(states.size); // TASK-0135 — connection gauge
339    msgRate.drop(connId);
340    const userSet = socketsByUser.get(userId);
341    if (userSet !== undefined) {
342      userSet.delete(ws);
343      if (userSet.size === 0) socketsByUser.delete(userId);
344    }
345    const controlArr = userControlSockets.get(userId);
346    if (controlArr !== undefined) {
347      const i = controlArr.indexOf(ws);
348      if (i >= 0) controlArr.splice(i, 1);
349      if (controlArr.length === 0) userControlSockets.delete(userId);
350    }
351    metrics.inc("ws_drops_total", { cause });
352    if (opts?.auditCause !== undefined) {
353      audit.publish("realtime.connection.abuse", {
354        connId,
355        userId,
356        ip,
357        room: st.room === null ? "control" : `${st.room.kind}:${st.room.id}`,
358        cause: opts.auditCause,
359        at: now(),
360      });
361    }
362    logger.warn({ connId, code, cause, ip }, "socket dropped");
363    try {
364      ws.close(code, cause);
365    } catch {
366      ws.terminate();
367    }
368  };
369 
370  const roomEvents: RoomManagerEvents = {
371    onPersisted: (roomKey, resourceId, version, logSeqAtFlush, blockDelta) => {
372      // Trim the update log to post-flush entries (R5): everything up to
373      // the flush snapshot's watermark is persisted; keep the rest.
374      void updateLog.trimAfter(roomKey, logSeqAtFlush).catch((err) => {
375        logger.warn({ err: String(err), roomKey }, "log trim failed");
376      });
377      const room = manager.get(roomKey);
378      if (room !== undefined) {
379        const text = encodeServerFrame({
380          type: "doc.persisted",
381          pageId: resourceId,
382          version,
383        });
384        for (const [connId] of room.peers) {
385          const ws = findWsByConn(connId);
386          if (ws !== null && ws.readyState === 1) ws.send(text);
387        }
388      }
389      // TASK-0092 R3: emit `page.updated` (the Phase-3 event shape) with
390      // the ADDITIVE `blockDelta` counts — the Phase-8 indexer + the
391      // computed-recompute consumer listen. The flush is a system write,
392      // attributed `actor: 'realtime-flush'` (auditable, not impersonable
393      // — the public API has no path to set that actor).
394      const delta = blockDeltaCounts(blockDelta);
395      events?.publish(
396        createDomainEvent(
397          DomainEventName.PAGE_UPDATED,
398          {
399            pageId: resourceId,
400            version,
401            blockDelta: delta,
402            ops: blockDelta,
403          },
404          {
405            workspaceId: workspaceIdByRoom.get(roomKey),
406            actorId: "realtime-flush",
407          },
408        ),
409      );
410      metrics.inc("events_processed_total", { type: "persist" });
411      logger.info({ roomKey, resourceId, version, delta }, "doc persisted");
412    },
413    onRebased: (roomKey, resourceId, appliedExternalOps) => {
414      // The flush-rebase ran (a concurrent REST write was absorbed into the
415      // doc) — observable for the rebase integration test + ops.
416      metrics.inc("events_processed_total", { type: "rebase" });
417      logger.info(
418        { roomKey, resourceId, appliedExternalOps },
419        "doc rebased on external (REST) write",
420      );
421    },
422    onRoomEvicted: (roomKey) => {
423      logger.info({ roomKey }, "room evicted");
424      logSeqByRoom.delete(roomKey);
425      workspaceIdByRoom.delete(roomKey);
426      for (const [ws, st] of [...states]) {
427        if (st.docRoom !== null && st.docRoom.key === roomKey && !st.closed) {
428          sendText(ws, {
429            type: "error",
430            code: "room_evicted",
431            message: "page reloaded — reconnecting",
432          });
433          dropSocket(ws, WS_CLOSE.ProtocolOrHeartbeat, "room_evicted");
434        }
435      }
436      manager.remove(roomKey);
437      void updateLog.clear(roomKey);
438    },
439  };
440 
441  const manager = new RoomManager(
442    {
443      store,
444      clock: {
445        now,
446        setTimeout: (fn, ms) => setTimeout(fn, ms),
447        clearTimeout: (h) => clearTimeout(h as ReturnType<typeof setTimeout>),
448      },
449      roomGcMs,
450      flushSeqProvider: (roomKey) => logSeqByRoom.get(roomKey) ?? 0,
451    },
452    roomEvents,
453  );
454 
455  // --- the upgrade handshake (R1) --------------------------------------------------
456 
457  const onUpgrade = (req: http.IncomingMessage, socket: net.Socket, head: Buffer): void => {
458    const url = req.url ?? "/";
459    const pathname = url.split("?")[0];
460    if (pathname !== "/realtime" && pathname !== "/realtime/") {
461      socket.destroy();
462      return;
463    }
464    if (draining) {
465      socket.write("HTTP/1.1 503 Service Unavailable\r\n\r\n");
466      socket.destroy();
467      return;
468    }
469    const params = new URL(url, "http://localhost").searchParams;
470    // The token: query (the client path) or the Authorization header
471    // (proxy compatibility, R1). NEVER logged.
472    const queryToken = params.get("token");
473    const headerAuth = req.headers.authorization;
474    const headerToken = headerAuth?.startsWith("Bearer ") ? headerAuth.slice(7).trim() : null;
475    const token = queryToken ?? headerToken;
476    const roomParam =
477      params.get("room") ?? (req.headers["x-realtime-room"] as string | undefined) ?? null;
478 
479    const ip = clientIp(req, config.trustProxy);
480    const ua = (req.headers["user-agent"] ?? "").slice(0, 200);
481 
482    // 1. connect rate (30/min/IP) — BEFORE auth (an unauthenticated
483    //    flood must not burn the JWT path).
484    const admit = connectLimiter.tryAdmit(ip);
485    if (!admit.ok) {
486      metrics.inc("ws_drops_total", { cause: "connect_rate_limited" });
487      audit.publish("realtime.connection.abuse", { ip, cause: "connect_rate_limited", at: now() });
488      socket.write("HTTP/1.1 429 Too Many Requests\r\n\r\n");
489      socket.destroy();
490      return;
491    }
492 
493    // 2. JWT verify (4001 — no room join).
494    const verified = verifyHandshakeToken(token, config.jwtSecret, now());
495    if (!verified.ok) {
496      metrics.inc("ws_handshake_failures_total");
497      logger.warn({ reason: verified.reason, ip, ua }, "handshake: unauthorized");
498      socket.write("HTTP/1.1 401 Unauthorized\r\n\r\n");
499      socket.destroy();
500      return;
501    }
502    const session = verified.session;
503 
504    // 3. parse the room key. A MALFORMED key → 400 bad_room (R1). A
505    //    MISSING/EMPTY key is a ROOMLESS CONTROL connection (TASK-0092
506    //    6b — the web RealtimeClient connects token-only, no ?room=, and
507    //    opens `page:<id>` channels in-band via `subscribe`); the socket
508    //    is admitted with `room: null` (no doc, no permission check —
509    //    the token was already verified above).
510    const room = parseRoomKey(roomParam ?? "");
511    if (room === null && (roomParam ?? "").length > 0) {
512      metrics.inc("ws_handshake_failures_total");
513      logger.warn({ ip, ua, room: roomParam }, "handshake: bad_room");
514      socket.write("HTTP/1.1 400 Bad Request\r\n\r\n");
515      socket.destroy();
516      return;
517    }
518 
519    // 4. user cap (100 live sockets/user).
520    const userSockets = socketsByUser.get(session.userId) ?? new Set<WebSocket>();
521    if (userSockets.size >= REALTIME_LIMITS.maxConnsPerUser) {
522      metrics.inc("ws_drops_total", { cause: "user_conn_cap" });
523      audit.publish("realtime.connection.abuse", {
524        userId: session.userId,
525        ip,
526        cause: "user_conn_cap",
527        at: now(),
528      });
529      socket.write("HTTP/1.1 429 Too Many Requests\r\n\r\n");
530      socket.destroy();
531      return;
532    }
533 
534    // 5. workspace + permission (async) → upgrade + join.
535    void (async () => {
536      try {
537        let canEdit = false;
538        let peerName: string | undefined;
539        let resolvedWorkspaceId: string | undefined;
540        if (room !== null) {
541          // user: control rooms are authorized to the socket's own userId
542          // ONLY (note 16 — joining user:{otherId} → 4002). No workspace
543          // or resource check applies (pure fan-out, header note 4).
544          if (room.kind === "user" && room.id !== session.userId) {
545            logger.warn(
546              { userId: session.userId, room: `user:${room.id}` },
547              "handshake: forbidden control room",
548            );
549            socket.write("HTTP/1.1 403 Forbidden\r\n\r\n");
550            socket.destroy();
551            return;
552          }
553          if (room.kind !== "user") {
554            const info = await meta.resolve({
555              kind: room.kind,
556              id: room.id,
557              userId: session.userId,
558            });
559            if (info === null) {
560              logger.warn(
561                { userId: session.userId, room: `${room.kind}:${room.id}` },
562                "handshake: unknown resource",
563              );
564              socket.write("HTTP/1.1 403 Forbidden\r\n\r\n");
565              socket.destroy();
566              return;
567            }
568            const workspaceId = info.workspaceId;
569            if (!session.workspaces.includes(workspaceId)) {
570              logger.warn(
571                { userId: session.userId, workspaceId, room: `${room.kind}:${room.id}` },
572                "handshake: wrong_workspace",
573              );
574              socket.write("HTTP/1.1 403 Forbidden\r\n\r\n");
575              socket.destroy();
576              return;
577            }
578            peerName = info.userName;
579            resolvedWorkspaceId = workspaceId;
580            // Doc rooms: VIEW to join, EDIT to write (the shared engine).
581            if (room.kind === "page" || room.kind === "database") {
582              const level = await permissions.resolveLevel({
583                userId: session.userId,
584                workspaceId,
585                resourceType: room.kind,
586                resourceId: room.id,
587              });
588              if (level === "none") {
589                logger.warn(
590                  { userId: session.userId, room: `${room.kind}:${room.id}` },
591                  "handshake: no VIEW",
592                );
593                socket.write("HTTP/1.1 403 Forbidden\r\n\r\n");
594                socket.destroy();
595                return;
596              }
597              canEdit = level === "edit" || level === "full_access";
598            }
599            // comment:{pageId} rooms fan out over the page — the join
600            // permission is the page's VIEW (the comment panel is part of
601            // the page; comment writes are REST, TASK-0094).
602            if (room.kind === "comment") {
603              const level = await permissions.resolveLevel({
604                userId: session.userId,
605                workspaceId,
606                resourceType: "page",
607                resourceId: room.id,
608              });
609              if (level === "none") {
610                logger.warn(
611                  { userId: session.userId, room: `comment:${room.id}` },
612                  "handshake: no VIEW",
613                );
614                socket.write("HTTP/1.1 403 Forbidden\r\n\r\n");
615                socket.destroy();
616                return;
617              }
618            }
619          }
620        }
621        // Roomless CONTROL socket (6b): the token was verified in step 2;
622        // no doc, no resource check — the page channels open in-band via
623        // `subscribe` after the upgrade.
624        if (wss === null) {
625          socket.destroy();
626          return;
627        }
628        wss.handleUpgrade(req, socket, head, (ws) => {
629          void onConnection(ws, session, room, canEdit, peerName, ip, ua, resolvedWorkspaceId);
630        });
631      } catch (err) {
632        logger.error({ err: String(err), ip }, "handshake failure");
633        socket.destroy();
634      }
635    })();
636  };
637 
638  // --- presence (TASK-0093, doc §5) ----------------------------------------------------
639 
640  /** The sockets that should receive a room's presence frames: every live
641   *  doc-room socket in the room + every roomless control socket that has
642   *  opened the room's `page:<id>`/`database:<id>` channel (the (b) web
643   *  client — one token-only socket, many in-band channels). */
644  const roomPresenceSubscribers = (roomKey: string): WebSocket[] => {
645    const out: WebSocket[] = [];
646    const inRoom = roomSockets.get(roomKey);
647    if (inRoom !== undefined) for (const ws of inRoom) if (ws.readyState === 1) out.push(ws);
648    for (const [ws, st] of states) {
649      if (st.room !== null) continue;
650      if (st.pageChannels.has(roomKey) && ws.readyState === 1) out.push(ws);
651    }
652    return out;
653  };
654 
655  /** Resolve a socket's display name for presence stamping. Doc-room
656   *  sockets carry it from the handshake meta (the page/database
657   *  existence resolve returns `userName`); roomless CONTROL sockets
658   *  resolve lazily once (the `meta` seam's `kind: "user"` path — the
659   *  Prisma store maps it to the `users` table) and cache in
660   *  `st.peerName`. Unresolvable → the shared fallback (the peer is
661   *  still identified by name, never color-only — the task spec). */
662  const resolvePeerName = async (st: SocketState): Promise<void> => {
663    if (st.peerName !== undefined) return;
664    const info = await meta.resolve({
665      kind: "user",
666      id: st.session.userId,
667      userId: st.session.userId,
668    });
669    st.peerName = info?.userName;
670  };
671 
672  /** Upsert a socket's own presence (the server stamps the identity — the
673   *  client frame carries ONLY `cursor`/`viewing`) and fan the broadcast
674   *  to the room's other subscribers (doc §5.1 best-effort; the > 50
675   *  cap strips `cursor`/`viewing` — the tracker shapes the frame).
676   *  Idempotent per (room, user): a second tab upserts the same key. */
677  const applyPresence = (ws: WebSocket, st: SocketState, update: PresenceUpdate): void => {
678    const room = presenceRoomFor(st);
679    if (room === null) return;
680    // TASK-0135: the spec's "span per ws room op" — presence fan-out is a
681    // room op (join/rejoin re-broadcasts ride the same path). The sync span
682    // is free when OTel is disabled (no-op) and adds no I/O.
683    // workspaceId ABSENT (not "unknown") when the room has no recorded
684    // tenant scope (contract shared with the API's span attributes).
685    const workspaceId = workspaceIdByRoom.get(room);
686    startSpan(
687      "ws.presence.update",
688      { room, userId: st.session.userId, ...(workspaceId !== undefined ? { workspaceId } : {}) },
689      () => {
690        const entry = presence.upsert(room, { id: st.session.userId, name: st.peerName }, update);
691        const frame = presence.broadcastFor(room, entry);
692        const text = encodeServerFrame(frame);
693        for (const target of roomPresenceSubscribers(room)) {
694          if (target === ws) continue; // the sender's own state is its echo — it knows it
695          target.send(text);
696        }
697      }
698    );
699  };
700 
701  /** Broadcast one peer's leave (explicit `presence-leave`, a socket
702   *  close, or the 90 s prune) — the removal signal (doc §5.2
703   *  best-effort; a missed leave heals on the next join re-broadcast +
704   *  the client's own prune window). */
705  const announcePresenceLeave = (roomKey: string, userId: string): void => {
706    // TASK-0135: span per ws room op (presence leave fan-out).
707    const workspaceId = workspaceIdByRoom.get(roomKey);
708    startSpan(
709      "ws.presence.leave",
710      { room: roomKey, userId, ...(workspaceId !== undefined ? { workspaceId } : {}) },
711      () => {
712        presence.remove(roomKey, userId);
713        const frame = { type: "presence-leave", pageId: roomKey, userId } as const;
714        const text = encodeServerFrame(frame);
715        for (const ws of roomPresenceSubscribers(roomKey)) ws.send(text);
716        logger.debug({ room: roomKey, userId }, "presence leave");
717      }
718    );
719  };
720 
721  /** The doc room this socket's presence belongs to — page/database rooms
722   *  only (comment/user rooms have no presence, doc §1). A roomless
723   *  control socket's presence is the LAST page/database channel it
724   *  subscribed (the (b) client switches pages by re-subscribing; the
725   *  previous page's entry leaves with the `unsubscribe`-driven
726   *  `presence-leave`). */
727  const presenceRoomFor = (st: SocketState): string | null => {
728    if (st.room !== null) {
729      return st.room.kind === "page" || st.room.kind === "database"
730        ? `${st.room.kind}:${st.room.id}`
731        : null;
732    }
733    return st.lastPresenceRoom ?? null;
734  };
735 
736  // --- a live socket -----------------------------------------------------------------
737 
738  const onConnection = async (
739    ws: WebSocket,
740    session: SessionIdentity,
741    room: ParsedRoom | null,
742    canEdit: boolean,
743    peerName: string | undefined,
744    ip: string,
745    ua: string,
746    resolvedWorkspaceId: string | undefined,
747  ): Promise<void> => {
748    const connId = randomUUID().replace(/-/g, "").slice(0, 16);
749    const st: SocketState = {
750      connId,
751      session,
752      room,
753      pageChannels: new Set(),
754      docRoom: null,
755      ip,
756      canEdit,
757      strikes: new StrikeTracker({ now }),
758      lastActivity: now(),
759      lastAckedSeq: 0,
760      closed: false,
761      dropped: false,
762    };
763    states.set(ws, st);
764    const userSockets = socketsByUser.get(session.userId) ?? new Set<WebSocket>();
765    userSockets.add(ws);
766    socketsByUser.set(session.userId, userSockets);
767    metrics.inc("events_processed_total", { type: "connection" });
768    // A roomless control socket has no doc room to count (6b).
769    if (room !== null) {
770      metrics.set("ws_connections", countConnectionsByRoom(room.kind), { room: room.kind });
771    }
772    // TASK-0135: the spec's `ws_active_connections` gauge — the PROCESS-wide
773    // total (states.size; the local per-room `ws_connections` is kept below
774    // it, the spec asks for the global gauge). Free when OTel is disabled.
775    m.ws.setConnections(states.size);
776    logger.info(
777      {
778        connId,
779        userId: session.userId,
780        room: room === null ? "control" : `${room.kind}:${room.id}`,
781        ip,
782        ua,
783      },
784      "connected",
785    );
786 
787    // The control room rides the same socket (header note 4 / note 16):
788    // ≤ 10 live control sockets per user — the OLDEST is evicted with a
789    // `control_evicted` error frame + close 4008.
790    const userArr = userControlSockets.get(session.userId) ?? [];
791    userArr.push(ws);
792    if (userArr.length > REALTIME_LIMITS.maxControlSocketsPerUser) {
793      const oldest = userArr.shift()!;
794      if (oldest !== ws) {
795        sendText(oldest, {
796          type: "error",
797          code: "control_evicted",
798          message: "too many tabs — reconnect this tab",
799        });
800        dropSocket(oldest, WS_CLOSE.RateOrPolicy, "control_evicted", {
801          auditCause: "control_evicted",
802        });
803      }
804    }
805    userControlSockets.set(session.userId, userArr);
806 
807    try {
808      if (room !== null && (room.kind === "page" || room.kind === "database")) {
809        const key = `${room.kind}:${room.id}`;
810        const docRoom = await manager.getOrCreate(key, room.kind, room.id);
811        // The room's workspace (for the flush's `page.updated` event) —
812        // recorded at join from the handshake meta (the room key carries
813        // no tenant scope, note 14).
814        if (resolvedWorkspaceId !== undefined) workspaceIdByRoom.set(key, resolvedWorkspaceId);
815        const peer = makePeer(connId, session.userId, canEdit, peerName);
816        docRoom.peers.set(connId, peer);
817        docRoom.notePeersChanged();
818        st.docRoom = docRoom;
819        const inRoom = roomSockets.get(key) ?? new Set<WebSocket>();
820        inRoom.add(ws);
821        roomSockets.set(key, inRoom);
822        sendText(ws, {
823          type: "hello",
824          connId,
825          canEdit,
826          serverTime: now(),
827          roomPeers: docRoom.roster().filter((p) => p.user.id !== session.userId),
828        });
829        // Presence (TASK-0093, doc §5.1): the join re-broadcast — a late
830        // joiner hears every existing peer's full state (cursors
831        // included; the deterministic catch-up), and the existing peers
832        // hear the joiner's initial state (no cursor, viewing=true —
833        // the avatar appears, the cursor shows when the joiner's first
834        // `presence` frame lands). Idempotent per (room, user): a second
835        // tab of the same user upserts the same presence key.
836        const joined = presence.upsert(key, { id: session.userId, name: peerName }, {});
837        for (const target of roomPresenceSubscribers(key)) {
838          if (target === ws) continue;
839          sendText(target, presence.broadcastFor(key, joined));
840        }
841        for (const other of presence.peers(key)) {
842          if (other.userId === session.userId) continue;
843          sendText(ws, presence.broadcastFor(key, other));
844        }
845        logger.debug(
846          { connId, room: key, peers: presence.count(key) },
847          "presence join re-broadcast",
848        );
849      } else {
850        // comment / user rooms: no doc — hello with the user's other
851        // tabs as peers (the control room's roster).
852        sendText(ws, {
853          type: "hello",
854          connId,
855          canEdit: false,
856          serverTime: now(),
857          roomPeers: userArr
858            .filter((o) => o !== ws)
859            .map((o, i) => {
860              const os = states.get(o);
861              return {
862                user: {
863                  id: os?.connId ?? `tab-${i}`,
864                  name: os !== undefined ? "Guest (shared)" : "tab",
865                  color: "slate",
866                },
867              };
868            }),
869        });
870      }
871    } catch (err) {
872      logger.error({ err: String(err), connId }, "join failure");
873      dropSocket(ws, WS_CLOSE.ProtocolOrHeartbeat, "join_failed");
874      return;
875    }
876 
877    // --- inbound frames (R3) -------------------------------------------------------
878 
879    ws.on("message", (data: Buffer | ArrayBuffer | Buffer[], isBinary: boolean) => {
880      const stNow = states.get(ws);
881      if (stNow === undefined || stNow.closed) return;
882      stNow.lastActivity = now();
883 
884      // Per-socket sustained rate (100 msg/s → 4008 + audit), keyed by connId.
885      const rate = msgRate.consume(stNow.connId);
886      if (!rate.ok) {
887        dropSocket(ws, WS_CLOSE.RateOrPolicy, "rate_limited", { auditCause: "msg_rate_limited" });
888        return;
889      }
890 
891      // The `isBinary` flag is the authoritative discriminator (the ws
892      // library delivers TEXT frames as Buffers too — `typeof data` would
893      // parse every JSON control frame as binary and strike it as
894      // `bad_binary_kind`; the flag routes text → JSON, binary → 0x01/0x02).
895      const asBytes = (): Uint8Array =>
896        Buffer.isBuffer(data)
897          ? new Uint8Array(data)
898          : Array.isArray(data)
899            ? new Uint8Array(Buffer.concat(data))
900            : new Uint8Array(data as ArrayBuffer);
901      const raw: Uint8Array | string = isBinary
902        ? asBytes()
903        : typeof data === "string"
904          ? data
905          : Buffer.from(asBytes()).toString("utf8");
906 
907      if (typeof raw !== "string" && raw.byteLength > MAX_FRAME_BYTES) {
908        dropSocket(ws, WS_CLOSE.PayloadTooLarge, "payload_too_large", {
909          auditCause: "frame_too_large",
910        });
911        return;
912      }
913 
914      const decoded = decodeClientFrame(raw);
915      if (!decoded.ok) {
916        const strike = stNow.strikes.record();
917        metrics.inc("ws_drops_total", { cause: `strike:${decoded.code}` });
918        sendText(ws, { type: "error", code: decoded.code, message: "frame rejected" });
919        if (strike.violated) {
920          dropSocket(ws, WS_CLOSE.RateOrPolicy, "strike_limit", {
921            auditCause: `strikes:${decoded.code}`,
922          });
923        }
924        return;
925      }
926 
927      if (decoded.frame.kind === "text") {
928        metrics.inc("ws_messages_total", { frame: decoded.frame.frame.type });
929        switch (decoded.frame.frame.type) {
930          case "sync.request": {
931            void handleSyncRequest(ws, stNow, decoded.frame.frame);
932            break;
933          }
934          case "ping": {
935            if (decoded.frame.frame.lastSeq !== undefined) {
936              stNow.lastAckedSeq = decoded.frame.frame.lastSeq;
937            }
938            sendText(ws, { type: "ping-pong", ts: decoded.frame.frame.ts });
939            break;
940          }
941          case "outbox.pending": {
942            logger.debug(
943              { connId: stNow.connId, count: decoded.frame.frame.count },
944              "outbox.pending",
945            );
946            break;
947          }
948          case "subscribe": {
949            // Control-channel join (6b): a roomless control socket opens a
950            // `page:<id>` (or `workspace:<id>`) channel in-band. Idempotent
951            // (reconnect re-subscribes). A doc-room socket ignores it (its
952            // room is the handshake's; control frames ride control sockets).
953            if (stNow.room === null) {
954              const channel = decoded.frame.frame.channel;
955              stNow.pageChannels.add(channel);
956              // Presence room tracking (TASK-0093): a page/database channel
957              // becomes the socket's presence room (the LAST one wins — the
958              // (b) client re-subscribes when it switches pages; the
959              // previous page's entry leaves with the `unsubscribe`).
960              if (channel.startsWith("page:") || channel.startsWith("database:")) {
961                if (stNow.lastPresenceRoom !== undefined && stNow.lastPresenceRoom !== channel) {
962                  announcePresenceLeave(stNow.lastPresenceRoom, stNow.session.userId);
963                }
964                stNow.lastPresenceRoom = channel;
965                // Join re-broadcast (doc §5.1): a late joiner hears every
966                // existing peer's full state (cursors included) — the
967                // deterministic catch-up — and the existing peers hear the
968                // joiner's initial state (no cursor, viewing=true). The
969                // name is resolved lazily for control sockets (no
970                // handshake meta for a roomless socket).
971                void (async () => {
972                  await resolvePeerName(stNow);
973                  const room = stNow.lastPresenceRoom;
974                  if (room === undefined) return;
975                  const joined = presence.upsert(
976                    room,
977                    { id: stNow.session.userId, name: stNow.peerName },
978                    {},
979                  );
980                  for (const target of roomPresenceSubscribers(room)) {
981                    if (target === ws) continue;
982                    sendText(target, presence.broadcastFor(room, joined));
983                  }
984                  for (const peer of presence.peers(room)) {
985                    if (peer.userId === stNow.session.userId) continue;
986                    sendText(ws, presence.broadcastFor(room, peer));
987                  }
988                  logger.debug(
989                    { connId: stNow.connId, channel, peers: presence.count(room) },
990                    "presence join re-broadcast",
991                  );
992                })();
993              }
994            }
995            logger.debug(
996              { connId: stNow.connId, channel: decoded.frame.frame.channel },
997              "control subscribe",
998            );
999            break;
1000          }
1001          case "unsubscribe": {
1002            if (stNow.room === null) {
1003              const channel = decoded.frame.frame.channel;
1004              stNow.pageChannels.delete(channel);
1005              // Presence room tracking: unsubscribing the socket's current
1006              // presence room leaves that room's roster (the (b) client
1007              // leaves a page by unsubscribing its channel).
1008              if (stNow.lastPresenceRoom === channel) {
1009                stNow.lastPresenceRoom = undefined;
1010                announcePresenceLeave(channel, stNow.session.userId);
1011              }
1012            }
1013            logger.debug(
1014              { connId: stNow.connId, channel: decoded.frame.frame.channel },
1015              "control unsubscribe",
1016            );
1017            break;
1018          }
1019          case "presence": {
1020            // Presence upsert (TASK-0093, doc §5): the client carries ONLY
1021            // `cursor`/`viewing` (the wire schema forbids identity). The
1022            // server stamps `user {id, name, color}` (id from the
1023            // handshake, name from the meta seam, color from the shared
1024            // `presenceColor` hash) and fans the broadcast out. A control
1025            // socket must have a page/database presence room (subscribed)
1026            // or the frame is a no-op (it has no room to state).
1027            const presenceUpdate: PresenceUpdate = {
1028              cursor: decoded.frame.frame.cursor,
1029              viewing: decoded.frame.frame.viewing,
1030            };
1031            if (stNow.room === null) {
1032              void (async () => {
1033                await resolvePeerName(stNow);
1034                if (stNow.lastPresenceRoom === undefined) return; // no room yet
1035                applyPresence(ws, stNow, presenceUpdate);
1036              })();
1037            } else {
1038              applyPresence(ws, stNow, presenceUpdate);
1039            }
1040            break;
1041          }
1042          case "presence-leave": {
1043            // Explicit leave (TASK-0093, doc §5): the client says it's
1044            // going (tab blur / page switch / socket close is implicit via
1045            // the `close` handler). Drops the peer + announces.
1046            const room = presenceRoomFor(stNow);
1047            if (room !== null) {
1048              announcePresenceLeave(room, stNow.session.userId);
1049            }
1050            break;
1051          }
1052        }
1053        return;
1054      }
1055 
1056      if (decoded.frame.binary === "update") {
1057        metrics.inc("ws_messages_total", { frame: "update" });
1058        void handleUpdate(ws, stNow, decoded.frame.bytes);
1059        return;
1060      }
1061      metrics.inc("ws_messages_total", { frame: "awareness" });
1062      handleAwareness(ws, stNow, decoded.frame.bytes);
1063    });
1064 
1065    ws.on("pong", () => {
1066      const stNow = states.get(ws);
1067      if (stNow !== undefined) stNow.lastActivity = now();
1068    });
1069 
1070    ws.on("close", () => onSocketClose(ws));
1071    ws.on("error", (err) => {
1072      const msg = err instanceof Error ? err.message : String(err);
1073      if (/payload|1009|max/i.test(msg)) {
1074        dropSocket(ws, WS_CLOSE.PayloadTooLarge, "payload_too_large", {
1075          auditCause: "frame_too_large",
1076        });
1077        return;
1078      }
1079      logger.debug({ connId: st.connId, err: msg }, "socket error");
1080      onSocketClose(ws);
1081    });
1082  };
1083 
1084  // --- sync.request (R5) --------------------------------------------------------------
1085 
1086  const handleSyncRequest = async (
1087    ws: WebSocket,
1088    st: SocketState,
1089    frame: { stateVector?: string; lastSeq?: number },
1090  ): Promise<void> => {
1091    const room = st.docRoom;
1092    if (room === null) {
1093      // comment/user/control rooms have no doc — nothing to sync.
1094      sendText(ws, { type: "sync.state", chunk: "", done: true });
1095      return;
1096    }
1097    const key = room.key;
1098    const svB64 = frame.stateVector;
1099    // R5 fast path: the update log covers lastSeq+1 → replay the logged
1100    // updates (cheap — no state-vector encode of a large doc).
1101    if (frame.lastSeq !== undefined) {
1102      const entries = await updateLog.entriesAfter(key, frame.lastSeq);
1103      if (entries !== null) {
1104        for (const e of entries) {
1105          sendBinary(ws, encodeBinaryUpdate(Buffer.from(e.updateB64, "base64")));
1106        }
1107        const last = entries[entries.length - 1];
1108        if (last !== undefined) {
1109          sendText(ws, { type: "ack", seq: last.seq });
1110        }
1111        // The logged updates may NOT cover the client's full state for a
1112        // fresh client — always finish with the state-vector delta so the
1113        // client converges (idempotent merge, R5 invariant).
1114        await sendStateDelta(ws, room, svB64);
1115        return;
1116      }
1117      // Log gap (or a lastSeq from another room's log — the log is
1118      // room-scoped, Security section) → state-vector fallback.
1119      logger.debug(
1120        { connId: st.connId, key, lastSeq: frame.lastSeq },
1121        "resume: log gap → state vector",
1122      );
1123    }
1124    await sendStateDelta(ws, room, svB64);
1125  };
1126 
1127  const sendStateDelta = async (
1128    ws: WebSocket,
1129    room: Room,
1130    svB64: string | undefined,
1131  ): Promise<void> => {
1132    const sv = svB64 !== undefined && svB64.length > 0 ? Buffer.from(svB64, "base64") : null;
1133    const delta = room.stateDelta(sv);
1134    // Paged ≤ 64 KB chunks (R3): base64 expands 3 bytes → 4 chars, so a
1135    // 48 KB raw chunk stays ≤ 64 KB on the wire.
1136    const CHUNK = Math.floor(REALTIME_LIMITS.maxSyncChunkBytes * 0.75);
1137    if (delta.byteLength === 0) {
1138      sendText(ws, { type: "sync.state", chunk: "", done: true });
1139      return;
1140    }
1141    for (let off = 0; off < delta.byteLength; off += CHUNK) {
1142      const slice = delta.subarray(off, Math.min(off + CHUNK, delta.byteLength));
1143      const done = off + CHUNK >= delta.byteLength;
1144      const frame: SyncStateFrame = {
1145        type: "sync.state",
1146        chunk: Buffer.from(slice).toString("base64"),
1147        done,
1148      };
1149      sendText(ws, frame);
1150    }
1151  };
1152 
1153  // --- binary update (R3) ---------------------------------------------------------------
1154 
1155  const handleUpdate = async (ws: WebSocket, st: SocketState, bytes: Uint8Array): Promise<void> => {
1156    const room = st.docRoom;
1157    if (room === null) {
1158      strikeFor(ws, st, "bad_update");
1159      return;
1160    }
1161    // Defense in depth (R1): the canEdit flag is re-checked per frame —
1162    // a VIEW-only peer's update is dropped with `error FORBIDDEN`
1163    // (the guard AND the per-socket flag — both).
1164    if (!st.canEdit) {
1165      metrics.inc("ws_drops_total", { cause: "view_write_dropped" });
1166      sendText(ws, { type: "error", code: "FORBIDDEN", message: "this room is read-only for you" });
1167      logger.warn({ connId: st.connId, room: room.key }, "update dropped: VIEW-only peer");
1168      return;
1169    }
1170    // Validate the Yjs update on a throwaway doc (a corrupt frame →
1171    // `bad_update` strike; R3).
1172    const probe = new Y.Doc();
1173    try {
1174      Y.applyUpdate(probe, bytes, "remote");
1175    } catch {
1176      probe.destroy();
1177      strikeFor(ws, st, "bad_update");
1178      return;
1179    }
1180    probe.destroy();
1181 
1182    try {
1183      const applied = room.applyUpdate(bytes);
1184      const key = room.key;
1185      if (!applied) {
1186        // Buffered during a re-hydration (doc §4.2) — applied in order by
1187        // the room's `completeHydration`. It must still be logged + acked
1188        // NOW (its seq is part of the room's log continuity; a client's
1189        // lastSeq accounting assumes every applied update got a seq).
1190        const b64 = Buffer.from(bytes).toString("base64");
1191        const seq = await updateLog.append(key, b64, room.lastVersion);
1192        logSeqByRoom.set(key, Math.max(logSeqByRoom.get(key) ?? 0, seq));
1193        st.lastAckedSeq = Math.max(st.lastAckedSeq, seq);
1194        sendText(ws, { type: "ack", seq });
1195        return;
1196      }
1197      const b64 = Buffer.from(bytes).toString("base64");
1198      const seq = await updateLog.append(key, b64, room.lastVersion);
1199      logSeqByRoom.set(key, Math.max(logSeqByRoom.get(key) ?? 0, seq));
1200      st.lastAckedSeq = Math.max(st.lastAckedSeq, seq);
1201      // Relay to the OTHER peers (binary 0x01) — the spec's "span per ws
1202      // room op" (a broadcast fan-out; TASK-0135).
1203      const workspaceId = workspaceIdByRoom.get(key);
1204      startSpan(
1205        "ws.update.broadcast",
1206        { room: key, peers: room.peers.size, ...(workspaceId !== undefined ? { workspaceId } : {}) },
1207        () => {
1208          const relay = encodeBinaryUpdate(bytes);
1209          for (const [connId] of room.peers) {
1210            if (connId === st.connId) continue;
1211            const peerWs = findWsByConn(connId);
1212            if (peerWs !== null) sendBinary(peerWs, relay);
1213          }
1214        }
1215      );
1216      // Ack the log seq (the client's resume fast path).
1217      sendText(ws, { type: "ack", seq });
1218    } catch (err) {
1219      logger.error({ err: String(err), connId: st.connId }, "applyUpdate failed");
1220      strikeFor(ws, st, "bad_update");
1221    }
1222  };
1223 
1224  const strikeFor = (
1225    ws: WebSocket,
1226    st: SocketState,
1227    code: "bad_update" | "unknown_frame",
1228  ): void => {
1229    const strike = st.strikes.record();
1230    metrics.inc("ws_drops_total", { cause: `strike:${code}` });
1231    sendText(ws, { type: "error", code, message: "frame rejected" });
1232    if (strike.violated) {
1233      dropSocket(ws, WS_CLOSE.RateOrPolicy, "strike_limit", { auditCause: `strikes:${code}` });
1234    }
1235  };
1236 
1237  // --- awareness (R4 fan-out) -------------------------------------------------------------
1238 
1239  const handleAwareness = (ws: WebSocket, st: SocketState, bytes: Uint8Array): void => {
1240    const room = st.docRoom;
1241    if (room === null) return; // no awareness on control rooms
1242    const relay = encodeBinaryAwareness(bytes);
1243    const peers = room.peers;
1244    // doc §5.1: presence fan-out is capped at 50 peers/room — beyond
1245    // the cap the relay goes to the NEWEST 50 peers (the client renders
1246    // avatars-only past the cap; the cursor-strip is client-side per the
1247    // frozen contract — the server's cap is the fan-out limit, recorded).
1248    const cap = REALTIME_LIMITS.awarenessPeerCap;
1249    const targets = peers.size > cap ? [...peers.keys()].slice(-cap) : [...peers.keys()];
1250    for (const connId of targets) {
1251      if (connId === st.connId) continue;
1252      const peerWs = findWsByConn(connId);
1253      if (peerWs !== null) sendBinary(peerWs, relay);
1254    }
1255  };
1256 
1257  // --- close / leave -------------------------------------------------------------------------
1258 
1259  const onSocketClose = (ws: WebSocket): void => {
1260    const st = states.get(ws);
1261    if (st === undefined) return;
1262    st.closed = true;
1263    states.delete(ws);
1264    m.ws.setConnections(states.size); // TASK-0135 — connection gauge
1265    msgRate.drop(st.connId);
1266    const userSet = socketsByUser.get(st.session.userId);
1267    if (userSet !== undefined) {
1268      userSet.delete(ws);
1269      if (userSet.size === 0) socketsByUser.delete(st.session.userId);
1270    }
1271    const controlArr = userControlSockets.get(st.session.userId);
1272    if (controlArr !== undefined) {
1273      const i = controlArr.indexOf(ws);
1274      if (i >= 0) controlArr.splice(i, 1);
1275      if (controlArr.length === 0) userControlSockets.delete(st.session.userId);
1276    }
1277    if (st.docRoom !== null) {
1278      st.docRoom.peers.delete(st.connId);
1279      st.docRoom.notePeersChanged();
1280      // Presence (TASK-0093): a doc-room socket's leave is implicit (the
1281      // roster entry dies with the socket) — drop the socket from the
1282      // room's presence fan-out set + announce the (room, user) leave,
1283      // but ONLY if no OTHER live socket of the same user is still in the
1284      // room (a second tab keeps the avatar — the presence key is shared
1285      // per (room, user)).
1286      const roomKey = st.docRoom.key;
1287      const inRoom = roomSockets.get(roomKey);
1288      if (inRoom !== undefined) {
1289        inRoom.delete(ws);
1290        if (inRoom.size === 0) roomSockets.delete(roomKey);
1291      }
1292      let otherTabInRoom = false;
1293      for (const [otherWs, other] of states) {
1294        if (
1295          otherWs !== ws &&
1296          other.docRoom !== null &&
1297          other.docRoom.key === roomKey &&
1298          other.session.userId === st.session.userId
1299        ) {
1300          otherTabInRoom = true;
1301          break;
1302        }
1303      }
1304      if (!otherTabInRoom) announcePresenceLeave(roomKey, st.session.userId);
1305    } else if (st.lastPresenceRoom !== undefined) {
1306      // Roomless control socket: its tracked presence room leaves with the
1307      // socket (the (b) client's page channel dies with the ws).
1308      announcePresenceLeave(st.lastPresenceRoom, st.session.userId);
1309      st.lastPresenceRoom = undefined;
1310    }
1311    // Drop the control channels (6b) — no control socket outlives its ws.
1312    st.pageChannels.clear();
1313    if (st.room !== null) {
1314      metrics.set("ws_connections", countConnectionsByRoom(st.room.kind), { room: st.room.kind });
1315    }
1316    logger.info(
1317      {
1318        connId: st.connId,
1319        userId: st.session.userId,
1320        room: st.room === null ? "control" : `${st.room.kind}:${st.room.id}`,
1321      },
1322      "disconnected",
1323    );
1324  };
1325 
1326  const countConnectionsByRoom = (kind: string): number => {
1327    let n = 0;
1328    for (const st of states.values()) {
1329      if (st.room !== null && st.room.kind === kind) n += 1;
1330    }
1331    return n;
1332  };
1333 
1334  // --- TASK-0094 seams (fan-out) -------------------------------------------------------------
1335 
1336  const publishCommentEvent = (pageId: string, event: CommentEventFrame): void => {
1337    const text = encodeServerFrame(event);
1338    const channel = `page:${pageId}`;
1339    for (const [ws, st] of states) {
1340      if (ws.readyState !== 1) continue;
1341      // Two fan-out targets (the (b) web architecture, recorded in the task
1342      // file):
1343      //   1. `comment:{pageId}` doc-room sockets (the `?room=` handshake
1344      //      kind — a dedicated comment-panel socket),
1345      //   2. roomless control sockets that opened the page's `page:{pageId}`
1346      //      channel in-band (the web `RealtimeClient` — it connects
1347      //      token-only and never opens a `comment:` room).
1348      const inCommentRoom = st.room !== null && st.room.kind === "comment" && st.room.id === pageId;
1349      const hasPageChannel = st.room === null && st.pageChannels.has(channel);
1350      if (inCommentRoom || hasPageChannel) {
1351        ws.send(text);
1352      }
1353    }
1354  };
1355 
1356  /** Push a `block:changed` refetch-hint to every roomless control socket
1357   *  that opened the page's `page:<id>` channel (6b). A doc-room socket
1358   *  gets the change through its own CRDT relay (the flush rebase), so
1359   *  the hint targets control sockets only — no double-signal. */
1360  const publishBlockChanged = (frame: BlockChangedFrame): void => {
1361    const channel = `page:${frame.pageId}`;
1362    const text = encodeServerFrame(frame);
1363    for (const [ws, st] of states) {
1364      if (st.room === null && st.pageChannels.has(channel) && ws.readyState === 1) {
1365        ws.send(text);
1366      }
1367    }
1368  };
1369 
1370  // --- TASK-0094: the domain-event control source (the (b) producer,
1371  //   generalized from TASK-0092 6b) ---
1372  // One stream, one poller. The injected FACTORY hands back the live source
1373  // (production = the `factory:events` XREAD poll, index.ts; tests = a fake
1374  // the test drives). Its sink routes EVERY event to the matching fan-out:
1375  //   - `page.updated`         → `block:changed` refetch-hint (6b)
1376  //   - `comment.*`            → `comment.event` room frames (R1)
1377  //   - `notification.created` → `notification.push` control-room push (R7)
1378  //   - anything else          → dropped (pure fan-out; the worker owns the
1379  //                              other consumers of the same stream).
1380  // When `domainEventSource` is provided it SUPPLANTS `pageUpdatedSource`
1381  // (the legacy single-purpose seam, kept only so pre-TASK-0094 tests keep
1382  // passing without edits). Omitted → no control fan-out (dev mode).
1383  const routeDomainEvent = (event: DomainEvent): void => {
1384    const blockFrame = blockChangedForPageUpdated(event);
1385    if (blockFrame !== null) {
1386      publishBlockChanged(blockFrame);
1387      return;
1388    }
1389    const commentFrame = commentEventFrameFor(event);
1390    if (commentFrame !== null) {
1391      publishCommentEvent(commentFrame.pageId, commentFrame);
1392      return;
1393    }
1394    const push = notificationPushFor(event);
1395    if (push !== null) {
1396      publishNotification(push.userId, push.frame);
1397      return;
1398    }
1399    // Not a control fan-out event (search-index / webhook / automation …) —
1400    // the worker consumes those; the realtime app drops them.
1401  };
1402 
1403  const domainEventSourceImpl: DomainEventSource | null = deps.domainEventSource
1404    ? deps.domainEventSource(routeDomainEvent)
1405    : null;
1406  const pageUpdatedSourceImpl: PageUpdatedSource | null =
1407    domainEventSourceImpl !== null || deps.pageUpdatedSource === undefined
1408      ? null
1409      : deps.pageUpdatedSource(routeDomainEvent);
1410 
1411  const publishNotification = (userId: string, event: NotificationPushFrame): void => {
1412    const text = encodeServerFrame(event);
1413    const arr = userControlSockets.get(userId);
1414    if (arr === undefined) return;
1415    for (const ws of arr) {
1416      if (ws.readyState === 1) ws.send(text);
1417    }
1418  };
1419 
1420  // --- heartbeat (R4) --------------------------------------------------------------------------
1421 
1422  const startHeartbeat = (): void => {
1423    heartbeat = setInterval(() => {
1424      const t = now();
1425      for (const ws of wss?.clients ?? []) {
1426        const st = states.get(ws);
1427        if (st === undefined || st.closed) continue;
1428        if (ws.readyState === 1) {
1429          try {
1430            ws.ping();
1431          } catch {
1432            /* closed mid-ping */
1433          }
1434        }
1435        if (t - st.lastActivity > pruneSilenceMs) {
1436          dropSocket(ws, WS_CLOSE.ProtocolOrHeartbeat, "heartbeat_timeout");
1437        }
1438      }
1439    }, pingIntervalMs);
1440    heartbeat.unref?.();
1441    // Presence prune (doc §5.1: 90 s silence → drop; the R4 socket prune
1442    // at 90 s already disconnects a silent socket — this drops the
1443    // PRESENCE entry for a user whose last tab left without a clean
1444    // `presence-leave`, and announces the removal). Runs on the same
1445    // cadence as the heartbeat; the threshold is the doc's
1446    // `awarenessPruneMs` (injectable for the deterministic prune test).
1447    presencePrune = setInterval(() => {
1448      const pruned = presence.prune();
1449      for (const [roomKey, entries] of pruned) {
1450        for (const entry of entries) announcePresenceLeave(roomKey, entry.userId);
1451        logger.debug({ room: roomKey, pruned: entries.length }, "presence prune");
1452      }
1453    }, pingIntervalMs);
1454    presencePrune.unref?.();
1455  };
1456 
1457  // --- health / readyz / metrics (R6) ----------------------------------------------------------
1458 
1459  const healthHandler = (req: http.IncomingMessage, res: http.ServerResponse): void => {
1460    const url = (req.url ?? "").split("?")[0];
1461    if (req.method !== "GET") {
1462      res.writeHead(405).end();
1463      return;
1464    }
1465    if (url === "/health" || url === "/health/") {
1466      res.writeHead(200, { "content-type": "application/json" }).end(
1467        JSON.stringify({
1468          status: "ok",
1469          connections: states.size,
1470          rooms: manager.rooms().length,
1471          uptime: Math.round((now() - startedAt) / 1000),
1472          // RSS in KB (the load baseline's "memory < 2 GB" reads this —
1473          // the probe port has no other way to see the process heap).
1474          memoryRss: Math.round(process.memoryUsage().rss / 1024),
1475        }),
1476      );
1477      return;
1478    }
1479    // Legacy scaffold path (the Phase-0 compose probes it — note 15).
1480    if (url === "/health/ready") {
1481      void readyz(res);
1482      return;
1483    }
1484    if (url === "/readyz" || url === "/readyz/") {
1485      void readyz(res);
1486      return;
1487    }
1488    if (url === "/metrics" || url === "/metrics/") {
1489      // TASK-0135: MERGE the OTel exposition (target_info + the spec's
1490      // `ws_active_connections`) with the local registry (per-room
1491      // `ws_connections{room}`, `ws_messages_total{frame}`, drops, …). The
1492      // local registry stays authoritative for its series — when OTel is ON
1493      // we DROP its overlapping `ws_messages_total` (no label-set collision
1494      // is legal in one exposition; the local {frame} form is richer) and
1495      // `target_info` is OTel-only. Local-only → behavior unchanged (the
1496      // merge is a no-op on empty OTel text, which is what a disabled OTel
1497      // yields).
1498      void (async () => {
1499        const local = metrics.render();
1500        let otelText = "";
1501        try {
1502          otelText = await renderPrometheus();
1503        } catch {
1504          otelText = "";
1505        }
1506        if (!otelText) {
1507          res.writeHead(200, { "content-type": "text/plain; version=0.0.4" }).end(local);
1508          return;
1509        }
1510        const merged = mergeExposition(local, otelText);
1511        res.writeHead(200, { "content-type": "text/plain; version=0.0.4" }).end(merged);
1512      })();
1513      return;
1514    }
1515    res.writeHead(404).end();
1516  };
1517 
1518  const readyz = async (res: http.ServerResponse): Promise<void> => {
1519    const checks: Record<string, string> = { ws: draining ? "draining" : "ok" };
1520    let ok = true;
1521    const redisPing = (updateLog as { ping?: () => Promise<boolean> }).ping;
1522    if (config.useRedis && redisPing !== undefined) {
1523      try {
1524        const r = await redisPing.call(updateLog);
1525        checks.redis = r ? "ok" : "error";
1526        if (!r) ok = false;
1527      } catch {
1528        checks.redis = "error";
1529        ok = false;
1530      }
1531    }
1532    try {
1533      const d = await store.ping();
1534      checks.database = d ? "ok" : "error";
1535      if (!d) ok = false;
1536    } catch {
1537      checks.database = "error";
1538      ok = false;
1539    }
1540    res
1541      .writeHead(ok ? 200 : 503, { "content-type": "application/json" })
1542      .end(JSON.stringify({ status: ok ? "ready" : "not_ready", checks }));
1543  };
1544 
1545  // --- lifecycle ---------------------------------------------------------------------------------
1546 
1547  const listen = (server: http.Server, port: number, what: string): Promise<void> =>
1548    new Promise<void>((resolve, reject) => {
1549      const onErr = (err: Error): void => reject(err);
1550      server.once("error", onErr);
1551      server.listen(port, () => {
1552        server.removeListener("error", onErr);
1553        logger.info({ port, what }, "listening");
1554        resolve();
1555      });
1556    });
1557 
1558  const closeServer = (server: http.Server): Promise<void> =>
1559    new Promise<void>((resolve) => {
1560      server.close(() => resolve());
1561      server.closeAllConnections?.();
1562    });
1563 
1564  const start = async (): Promise<void> => {
1565    startedAt = now();
1566    const server = http.createServer(healthHandler);
1567    // The ws-layer maxPayload must sit ABOVE the app-level MAX_FRAME_BYTES
1568    // (256 KB): a frame in between reaches the message handler and is
1569    // closed with the app's 4009 payload_too_large (+ audit). The ws
1570    // layer's 1009 is the second-line backstop for frames above 300 KB.
1571    // With maxPayload == MAX_FRAME_BYTES the 1009 always won the race and
1572    // the 4009 path was unreachable (recorded TASK-0091 fix).
1573    const wsMaxPayload = MAX_FRAME_BYTES + 48 * 1024;
1574    const wsServer = new WebSocketServer({ noServer: true, maxPayload: wsMaxPayload });
1575    wss = wsServer;
1576    httpServer = server;
1577    server.on("upgrade", onUpgrade);
1578    await listen(server, config.port, "ws");
1579 
1580    // The second probe port (note 15 — the compose exposes 4002).
1581    const probe = http.createServer(healthHandler);
1582    probeServer = probe;
1583    await listen(probe, config.healthPort, "probe");
1584 
1585    startHeartbeat();
1586 
1587    // The control source (6b / TASK-0094) starts with the server; an event
1588    // before any control socket exists is a no-op (no channels).
1589    domainEventSourceImpl?.start();
1590    pageUpdatedSourceImpl?.start();
1591  };
1592 
1593  const stop = async (): Promise<void> => {
1594    draining = true;
1595    if (heartbeat !== undefined) {
1596      clearInterval(heartbeat);
1597      heartbeat = undefined;
1598    }
1599    if (presencePrune !== undefined) {
1600      clearInterval(presencePrune);
1601      presencePrune = undefined;
1602    }
1603    if (domainEventSourceImpl !== null) {
1604      await domainEventSourceImpl.stop();
1605    }
1606    if (pageUpdatedSourceImpl !== null) {
1607      await pageUpdatedSourceImpl.stop();
1608    }
1609    for (const ws of wss?.clients ?? []) {
1610      try {
1611        ws.close(1001, "server shutting down");
1612      } catch {
1613        /* already closed */
1614      }
1615    }
1616    if (httpServer !== null) {
1617      const s = httpServer;
1618      httpServer = null;
1619      await closeServer(s);
1620    }
1621    if (probeServer !== null) {
1622      const s = probeServer;
1623      probeServer = null;
1624      await closeServer(s);
1625    }
1626    if (wss !== null) {
1627      const w = wss;
1628      wss = null;
1629      await new Promise<void>((resolve) => w.close(() => resolve()));
1630    }
1631  };
1632 
1633  /** SIGTERM drain (R6): 4010 + retryAfterMs to all sockets, flush all
1634   *  dirty rooms (≤ 10 s), exit. */
1635  const drain = async (): Promise<void> => {
1636    draining = true;
1637    const notice = encodeServerFrame({
1638      type: "error",
1639      code: "server_shutdown",
1640      message: `retry after ${REALTIME_LIMITS.shutdownRetryAfterMs} ms`,
1641    });
1642    for (const ws of wss?.clients ?? []) {
1643      if (ws.readyState === 1) {
1644        ws.send(notice);
1645      }
1646      try {
1647        ws.close(WS_CLOSE.ServerShutdown, `retryAfterMs=${REALTIME_LIMITS.shutdownRetryAfterMs}`);
1648      } catch {
1649        /* already closed */
1650      }
1651    }
1652    const result = await manager.drainAll(REALTIME_LIMITS.drainMaxMs);
1653    logger.info({ ...result }, "drain complete");
1654    await stop();
1655  };
1656 
1657  return {
1658    start,
1659    stop,
1660    drain,
1661    publishCommentEvent,
1662    publishBlockChanged,
1663    publishNotification,
1664    sockets: (): readonly SocketState[] => [...states.values()],
1665    manager,
1666    metrics,
1667  };
1668}
1669 
1670// --- defaults (dev + tests; index.ts wires the production seams) -------------------------
1671 
1672function clientIp(req: http.IncomingMessage, trustProxy: boolean): string {
1673  if (trustProxy) {
1674    const fwd = req.headers["x-forwarded-for"];
1675    const first = Array.isArray(fwd) ? fwd[0] : fwd;
1676    if (first !== undefined && first !== null) {
1677      const head = first.split(",")[0];
1678      if (head !== undefined && head.trim().length > 0) return head.trim();
1679    }
1680  }
1681  return req.socket.remoteAddress ?? "unknown";
1682}
1683 
1684/** Dev default: no writes allowed (canEdit=false), joins allowed. The
1685 *  production resolver is `createPrismaPermissionResolver` (index.ts). */
1686function failClosedPermissions(logger: Logger): PermissionResolver {
1687  logger.debug({}, "realtime: using fail-closed permission resolver (no DB wired)");
1688  return {
1689    resolveLevel: async () => "view",
1690  };
1691}
1692 
1693/** Dev default: unknown resources (no DB wired) → 4002. The production
1694 *  resolver is `createPrismaMetaResolver` (index.ts). */
1695function failClosedMeta(logger: Logger): RoomMetaResolver {
1696  logger.debug({}, "realtime: using fail-closed meta resolver (no DB wired)");
1697  return {
1698    resolve: async () => null, // unknown resource → 4002 (fail closed)
1699  };
1700}
1701