rt-helpers.ts · Source

rt-helpers.tstests/integration/realtime/rt-helpers.tsraw ← realtime31,016 B← Back to the app
1/**
2 * Realtime integration test harness (TASK-0091).
3 *
4 * Boots the REAL `createRealtimeServer` (the production wiring minus the
5 * DB/Redis seams — injectable fake store/log/permissions/meta/audit) on an
6 * ephemeral 127.0.0.1 port pair and drives it with a real `ws` client.
7 * No Docker is needed: the suite validates the wire protocol, handshake
8 * matrix, heartbeat/prune, rate limits, resume, room GC, drain, and the
9 * probe endpoints against the exact server code that ships.
10 *
11 * Time model: the server runs on REAL time (its heartbeat interval, flush
12 * debounce, and room GC are real timers — the same as production; only
13 * the clock readings are injectable). Timer-driven tests therefore use
14 * short real cadences (`pingIntervalMs`, `pruneSilenceMs`, `roomGcMs`)
15 * and wall-time waits, which is what a client experiences.
16 */
17import net from "node:net";
18import { createHmac } from "node:crypto";
19import WebSocket from "ws";
20import { createRealtimeServer, type RealtimeServer } from "@factory/realtime/src/server";
21import type { PageUpdatedSource } from "@factory/realtime/src/page-updated-source";
22import type { DomainEventSource } from "@factory/realtime/src/event-source";
23import type { RealtimeConfig } from "@factory/realtime/src/config";
24import { InMemoryDocStore, type DocStore } from "@factory/realtime/src/rooms";
25import { InMemoryUpdateLog } from "@factory/realtime/src/update-log";
26import { WS_CLOSE } from "@factory/shared/realtime/frames";
27import * as Y from "yjs";
28 
29import { DomainEventName, type DomainEvent, type BlockDelta } from "@factory/shared";
30 
31// --- fake audit sink ------------------------------------------------------------------
32 
33/** Records every published audit event (asserted for the abuse audit). */
34export class FakeAudit {
35  readonly records: { event: string; payload: Record<string, unknown> }[] = [];
36  publish(event: string, payload: Record<string, unknown>): void {
37    this.records.push({ event, payload });
38  }
39  abuse(cause?: string): number {
40    return this.records.filter(
41      (r) =>
42        r.event === "realtime.connection.abuse" &&
43        (cause === undefined || r.payload.cause === cause),
44    ).length;
45  }
46}
47 
48/** Records every domain event the server emits (the flush's `page.updated`,
49 *  TASK-0092 R3 — the `blockDelta` additive field is asserted against). */
50export class FakeEventBus {
51  readonly records: DomainEvent[] = [];
52  publish(event: DomainEvent): void {
53    this.records.push(event);
54  }
55  /** The `page.updated` events for a page (the flush's emission). */
56  pageUpdated(pageId: string): DomainEvent<{
57    pageId: string;
58    version: number;
59    blockDelta: BlockDelta;
60    ops: unknown[];
61  }>[] {
62    return this.records.filter(
63      (e) =>
64        e.name === DomainEventName.PAGE_UPDATED &&
65        (e.payload as { pageId?: string }).pageId === pageId,
66    ) as DomainEvent<{ pageId: string; version: number; blockDelta: BlockDelta; ops: unknown[] }>[];
67  }
68}
69 
70// --- the fixture ------------------------------------------------------------------------
71 
72export interface RealtimeFixture {
73  readonly server: RealtimeServer;
74  readonly wsUrl: string;
75  readonly healthUrl: string;
76  readonly store: InMemoryDocStore;
77  readonly updateLog: InMemoryUpdateLog;
78  readonly audit: FakeAudit;
79  /** Records the server's domain events (the flush's `page.updated`). */
80  readonly events: FakeEventBus;
81  /** The server's clock (the injected fake when `now0` is set, else real). */
82  readonly now: () => number;
83  /** Advance the fake clock (no-op on a real clock). */
84  readonly advance: (ms: number) => void;
85  /** Sleep wall time (let real timers + the network settle). */
86  readonly settle: (ms: number) => Promise<void>;
87  /** Always on in the fixture: the test clients identify their source IP
88   *  via X-Forwarded-For (see `connectClient`'s auto-IP). */
89  readonly trustProxy: boolean;
90  stop: () => Promise<void>;
91}
92 
93export interface BuildFixtureOpts {
94  /** Inject a fake server clock starting at this epoch ms (deterministic
95   *  prune/GC threshold asserts; the timers themselves stay real). */
96  now0?: number;
97  /** Heartbeat interval (REAL timer, ms). Default 30 s. */
98  pingIntervalMs?: number;
99  /** Silence threshold (ms, REAL time). Default 90 s. */
100  pruneSilenceMs?: number;
101  /** Empty-room GC delay (ms, REAL timer). Default 5 min. */
102  roomGcMs?: number;
103  /** `page:<id>` / `database:<id>` / `comment:<id>` → effective level.
104   *  Default: pages → `edit` (comment rooms follow), databases → `view`. */
105  levels?: Record<string, "none" | "view" | "comment" | "edit" | "full_access">;
106  /** Existing page ids (the meta resolver's existence check). */
107  pages?: Set<string>;
108  /** Existing database ids. */
109  databases?: Set<string>;
110  /** Resource id → workspace id (the token's workspaces[] must contain it). */
111  workspaces?: Record<string, string>;
112  /** Honor X-Forwarded-For as the client IP (the user-cap test spreads
113   *  100 conns over distinct IPs so the 30/min/IP limit doesn't fire first). */
114  trustProxy?: boolean;
115  /** The doc store the server persists to. Defaults to the fixture's
116   *  seedable in-memory store (returned as `store`). Slice 4 injects a
117   *  real `HttpDocStore` (pointed at a fake `/internal/*` API) to test
118   *  the PRODUCTION flush path — room `flush()` → HTTP flush → the CAS
119   *  conflict → the rebase → the retry — over real HTTP, Docker-free. When
120   *  set, the returned `store` handle is a separate unused in-memory store
121   *  (assert against your fake instead). */
122  store?: DocStore;
123  /** The `page.updated` CONTROL source factory (TASK-0092 6b): given the
124   *  server's sink, returns the source to poll (tests = a fake the test
125   *  drives; production = the `factory:events` XREAD poll). Omitted → no
126   *  control fan-out. */
127  pageUpdatedSource?: (onPageUpdated: (event: DomainEvent) => void) => PageUpdatedSource;
128  /** The GENERIC domain-event CONTROL source factory (TASK-0094): given the
129   *  server's sink, returns the source to poll (tests = a fake the test
130   *  drives; production = the `factory:events` XREAD poll). When provided it
131   *  supplants `pageUpdatedSource`. Omitted → no control fan-out. */
132  domainEventSource?: (onEvent: (event: DomainEvent) => void) => DomainEventSource;
133  /** Mirror every domain event the server emits (the flush's
134   *  `page.updated`) into a test-owned sink — the loopback "Redis stream"
135   *  for the 6c convergence test (the in-memory stand-in for the API's
136   *  `factory:events` XADD). `fx.events` still records the same events. */
137  eventsTap?: (event: DomainEvent) => void;
138  /** userId → display name (the meta seam's `userName` — the peer roster
139   *  + the presence identity). Default: every user is "Test User". */
140  userNames?: Record<string, string>;
141  /** Presence prune override (ms). Default: the doc §5.1 90 s
142   *  (`REALTIME_LIMITS.awarenessPruneMs`). The presence prune test uses a
143   *  short window + the fake clock (`now0`). */
144  presencePruneMs?: number;
145}
146 
147/**
148 * Boot an in-process realtime server with fake seams. Seed page content
149 * with `fixture.store.seed(pageId, blocks)` before connecting clients.
150 */
151export async function buildRealtimeFixture(opts: BuildFixtureOpts = {}): Promise<RealtimeFixture> {
152  // Fake clock when `now0` is set (deterministic threshold asserts), real
153  // clock otherwise. Timers stay real (the server wires real setTimeout /
154  // setInterval — the injected clock drives READINGS, not scheduling).
155  let t = opts.now0 ?? Date.now();
156  const fake = opts.now0 !== undefined;
157  const now = (): number => (fake ? t : Date.now());
158  const audit = new FakeAudit();
159  const events = new FakeEventBus();
160  // The seedable in-memory handle (returned as `store`). The server
161  // persists to `opts.store` when provided (slice 4's HttpDocStore), else
162  // to this one (every pre-existing test).
163  const seedStore = new InMemoryDocStore();
164  const updateLog = new InMemoryUpdateLog();
165 
166  // The server's domain-event sink. `events` always records (the pre-existing
167  // assert target); when an `eventsTap` is provided (6c), each emission is
168  // also mirrored into it — the loopback "Redis stream" the REAL
169  // `createPageUpdatedSource` polls (the in-memory `factory:events`
170  // XADD stand-in).
171  const eventsSink = {
172    publish: (event: DomainEvent): void => {
173      events.publish(event);
174      opts.eventsTap?.(event);
175    },
176  };
177 
178  const levels = opts.levels ?? {};
179  const pages = opts.pages ?? new Set<string>();
180  const databases = opts.databases ?? new Set<string>();
181  const workspaces = opts.workspaces ?? {};
182  const levelMap: Record<string, "none" | "view" | "comment" | "edit" | "full_access"> = {};
183  for (const [key, level] of Object.entries(levels)) levelMap[key] = level;
184  for (const pageId of pages) {
185    if (!(`page:${pageId}` in levelMap)) levelMap[`page:${pageId}`] = "edit";
186    if (!(`comment:${pageId}` in levelMap)) levelMap[`comment:${pageId}`] = "edit";
187  }
188  for (const dbId of databases) {
189    if (!(`database:${dbId}` in levelMap)) levelMap[`database:${dbId}`] = "view";
190  }
191 
192  const ports = await freePorts(2);
193  const config: RealtimeConfig = {
194    port: ports[0]!,
195    healthPort: ports[1]!,
196    jwtSecret: IT_SECRET,
197    redisUrl: "redis://unused",
198    apiUrl: "http://unused",
199    logLevel: "silent",
200    rateLimitWindowMs: 60_000,
201    rateLimitMax: 300,
202    maxRoomSize: 200,
203    useRedis: false,
204    useDatabase: false,
205    // Empty token + useDatabase:false → the in-memory store (buildStore
206    // never consults the token on this path).
207    internalServiceToken: "",
208    // Always on: the test clients identify their source IP via
209    // X-Forwarded-For (connectClient auto-assigns a unique synthetic IP
210    // per connection — a single 127.0.0.1 would exhaust the 30/min/IP
211    // connect budget in the second test of any file).
212    trustProxy: true,
213  };
214 
215  const server = createRealtimeServer(config, {
216    now,
217    pingIntervalMs: opts.pingIntervalMs ?? 30_000,
218    pruneSilenceMs: opts.pruneSilenceMs ?? 90_000,
219    roomGcMs: opts.roomGcMs ?? 5 * 60_000,
220    store: opts.store ?? seedStore,
221    log: updateLog,
222    audit,
223    events: eventsSink,
224    pageUpdatedSource: opts.pageUpdatedSource,
225    domainEventSource: opts.domainEventSource,
226    presencePruneMs: opts.presencePruneMs,
227    permissions: {
228      resolveLevel: async ({ resourceType, resourceId }) =>
229        levelMap[`${resourceType}:${resourceId}`] ?? "none",
230    },
231    meta: {
232      resolve: async ({ kind, id, userId }) => {
233        const name = (uid: string) => opts.userNames?.[uid] ?? "Test User";
234        if (kind === "user") {
235          return userId.length > 0 ? { workspaceId: "__own__", userName: name(userId) } : null;
236        }
237        if (kind === "page" || kind === "comment") {
238          if (!pages.has(id)) return null;
239          return { workspaceId: workspaces[id] ?? "w1", userName: name(userId) };
240        }
241        if (!databases.has(id)) return null;
242        return { workspaceId: workspaces[id] ?? "w1", userName: name(userId) };
243      },
244    },
245  });
246 
247  await server.start();
248  return {
249    server,
250    wsUrl: `ws://127.0.0.1:${ports[0]}`,
251    healthUrl: `http://127.0.0.1:${ports[1]}`,
252    store: seedStore,
253    updateLog,
254    audit,
255    events,
256    now,
257    advance: (ms) => {
258      if (fake) t += ms;
259    },
260    settle: (ms) => new Promise((r) => setTimeout(r, ms)),
261    trustProxy: true,
262    stop: () => server.stop(),
263  };
264}
265 
266/**
267 * A fake GENERIC domain-event source the test drives directly (no Redis) —
268 * the in-memory stand-in for the production `factory:events` XREAD poll
269 * (TASK-0094; the same pattern as `control-channel.test.ts`'s
270 * `pageUpdatedSource` fake, now routing every event through the server's
271 * sink). Pass `domainEventSource: factory` to {@link buildRealtimeFixture}.
272 */
273export function makeFakeDomainEventSource() {
274  let sink: ((event: DomainEvent) => void) | null = null;
275  const source: DomainEventSource = {
276    start(): void {
277      /* the test drives deliver(); nothing to poll. */
278    },
279    async stop(): Promise<void> {
280      sink = null;
281    },
282  };
283  return {
284    source,
285    factory: (onEvent: (event: DomainEvent) => void): DomainEventSource => {
286      sink = onEvent;
287      return source;
288    },
289    /** Deliver a raw domain event through the server's routing sink. */
290    deliver(name: string, payload: Record<string, unknown>): void {
291      if (sink === null)
292        throw new Error("source not wired (the server was built without the factory)");
293      sink({
294        id: `evt_${Math.random().toString(36).slice(2)}`,
295        name: name as DomainEvent["name"],
296        payload,
297        ts: new Date().toISOString(),
298      });
299    },
300  };
301}
302 
303/** Free ephemeral 127.0.0.1 ports (the server binds them). */
304export async function freePorts(n: number): Promise<number[]> {
305  const out: number[] = [];
306  for (let i = 0; i < n; i += 1) {
307    const srv = net.createServer();
308    await new Promise<void>((resolve, reject) => {
309      srv.once("error", reject);
310      srv.listen(0, "127.0.0.1", () => resolve());
311    });
312    const addr = srv.address();
313    if (addr === null || typeof addr === "string") throw new Error("no free port");
314    out.push(addr.port);
315    await new Promise<void>((resolve) => srv.close(() => resolve()));
316  }
317  return out;
318}
319 
320// --- JWT (the API's claim shape: `sub`, `workspaces[]`) ----------------------------------
321 
322export const IT_SECRET = "it-secret-0123456789";
323 
324function b64url(s: string): string {
325  return Buffer.from(s, "utf8").toString("base64url");
326}
327 
328/** Mint a handshake JWT the server accepts (HS256, shared secret). */
329export function mintToken(
330  claims: { sub: string; workspaces: string[]; expSec?: number; iss?: string },
331  secret: string = IT_SECRET,
332): string {
333  const expSec = claims.expSec ?? Math.floor(Date.now() / 1000) + 3600;
334  const body: Record<string, unknown> = {
335    sub: claims.sub,
336    iat: Math.floor(Date.now() / 1000),
337    exp: expSec,
338    jti: `j-${Math.random().toString(36).slice(2, 10)}`,
339    workspaces: claims.workspaces,
340  };
341  if (claims.iss !== undefined) body.iss = claims.iss;
342  const h = b64url(JSON.stringify({ alg: "HS256", typ: "JWT" }));
343  const c = b64url(JSON.stringify(body));
344  const sig = createHmac("sha256", secret).update(`${h}.${c}`).digest().toString("base64url");
345  return `${h}.${c}.${sig}`;
346}
347 
348// --- the ws client ----------------------------------------------------------------------
349 
350export interface RtClient {
351  readonly ws: WebSocket;
352  textFrames: () => Record<string, unknown>[];
353  binaryFrames: () => { kind: "update" | "awareness"; bytes: Uint8Array }[];
354  /** Await the next `n` frames of `type` (already-received ones count). */
355  nextFrames: (type: string, n: number, timeoutMs?: number) => Promise<Record<string, unknown>[]>;
356  nextFrame: (type: string, timeoutMs?: number) => Promise<Record<string, unknown>>;
357  /** Await the first text frame matching a predicate (e.g. `done === true`). */
358  nextFrameWhere: (
359    pred: (f: Record<string, unknown>) => boolean,
360    timeoutMs?: number,
361  ) => Promise<Record<string, unknown>>;
362  /** Await the next binary frame (the 0x01/0x02-prefixed Yjs payloads). */
363  nextBinary: (timeoutMs?: number) => Promise<{ kind: "update" | "awareness"; bytes: Uint8Array }>;
364  sendText: (obj: unknown) => void;
365  sendBinary: (kind: 1 | 2, bytes: Uint8Array) => void;
366  /** The observed close (null while open). */
367  closed: () => { code: number; reason: string } | null;
368  /** Resolve when the socket closes (already closed → immediate). */
369  onClose: (timeoutMs?: number) => Promise<{ code: number; reason: string }>;
370  isOpen: () => boolean;
371  close: (code?: number) => void;
372  terminate: () => void;
373}
374 
375/** Connect a real `ws` client to the gateway. Rejects on a refused upgrade.
376 *
377 * Source IP: every call gets a UNIQUE synthetic IP via X-Forwarded-For
378 * (the fixture always trusts the proxy). Without this, one 127.0.0.1
379 * would exhaust the 30-connects/min/IP budget in the second test of any
380 * file — the rate-limit tests pass an explicit `headers["x-forwarded-for"]`
381 * when they deliberately share an IP. */
382let testIpCounter = 0;
383function autoTestIp(): string {
384  testIpCounter = (testIpCounter + 1) % 0xffffff;
385  return `10.${(testIpCounter >> 16) & 0xff}.${(testIpCounter >> 8) & 0xff}.${testIpCounter & 0xff}`;
386}
387 
388export async function connectClient(
389  base: { wsUrl: string },
390  opts: { token?: string; room?: string; url?: string; headers?: Record<string, string> } = {},
391): Promise<RtClient> {
392  const token = opts.token ?? mintToken({ sub: "u-it", workspaces: ["w1"] });
393  const room = opts.room ?? "page:p-it";
394  const url =
395    opts.url ??
396    `${base.wsUrl}/realtime?token=${encodeURIComponent(token)}&room=${encodeURIComponent(room)}`;
397  const headers = { "x-forwarded-for": autoTestIp(), ...opts.headers };
398  const ws = new WebSocket(url, { headers });
399 
400  /**
401   * Two consuming queues (text frames, binary frames). `nextFrame`/
402   * `nextFrames`/`nextFrameWhere` return the NEXT unconsumed matching
403   * text frames and mark them consumed, so two successive
404   * `nextFrame("ack")` calls never see the same ack (the resume test's
405   * seq accounting depends on this). `textFrames()`/`binaryFrames()` are
406   * NON-consuming history views (e.g. counting the replayed updates).
407   * The `consumed` flag alone provides the "don't return a frame twice"
408   * guarantee — there is no cursor, so a predicate that skips frames
409   * never hides earlier frames from a different waiter.
410   */
411  interface Bin {
412    kind: "update" | "awareness";
413    bytes: Uint8Array;
414  }
415  const textQueue: { frame: Record<string, unknown>; consumed: boolean }[] = [];
416  const binQueue: { bin: Bin; consumed: boolean }[] = [];
417  interface Waiter {
418    label: string;
419    pred: ((f: Record<string, unknown>) => boolean) | null;
420    need: number;
421    timer: ReturnType<typeof setTimeout>;
422    resolve: (v: Record<string, unknown>[]) => void;
423    reject: (e: Error) => void;
424  }
425  const waiters: Waiter[] = [];
426  let closeInfo: { code: number; reason: string } | null = null;
427  const closeWaiters: {
428    resolve: (v: { code: number; reason: string }) => void;
429    timer: ReturnType<typeof setTimeout>;
430  }[] = [];
431 
432  /** The first `need` unconsumed text frames matching `pred` (indexes). */
433  const findMatches = (
434    pred: ((f: Record<string, unknown>) => boolean) | null,
435    need: number,
436  ): number[] => {
437    const out: number[] = [];
438    for (let j = 0; j < textQueue.length && out.length < need; j += 1) {
439      const qe = textQueue[j]!;
440      if (qe.consumed) continue;
441      if (pred === null || pred(qe.frame)) out.push(j);
442    }
443    return out;
444  };
445  const consume = (idxs: number[]): Record<string, unknown>[] => {
446    for (const j of idxs) textQueue[j]!.consumed = true;
447    return idxs.map((j) => textQueue[j]!.frame);
448  };
449 
450  /** Try to satisfy every waiter with the next unconsumed matching text
451   *  frames. Returns true when at least one waiter was satisfied. */
452  const settleWaiters = (): boolean => {
453    let satisfied = false;
454    for (let i = waiters.length - 1; i >= 0; i -= 1) {
455      const w = waiters[i]!;
456      const matched = findMatches(w.pred, w.need);
457      if (matched.length >= w.need) {
458        const frames = consume(matched);
459        waiters.splice(i, 1);
460        clearTimeout(w.timer);
461        w.resolve(frames);
462        satisfied = true;
463      }
464    }
465    return satisfied;
466  };
467 
468  ws.on("message", (data: Buffer | ArrayBuffer | Buffer[], isBinary: boolean) => {
469    if (isBinary) {
470      const b = Buffer.isBuffer(data) ? new Uint8Array(data) : new Uint8Array(data as ArrayBuffer);
471      binQueue.push({
472        bin: { kind: b[0] === 2 ? "awareness" : "update", bytes: b.subarray(1) },
473        consumed: false,
474      });
475    } else {
476      let parsed: Record<string, unknown>;
477      try {
478        parsed = JSON.parse(Buffer.isBuffer(data) ? data.toString("utf8") : String(data)) as Record<
479          string,
480          unknown
481        >;
482      } catch {
483        return;
484      }
485      textQueue.push({ frame: parsed, consumed: false });
486    }
487    settleWaiters();
488  });
489  ws.on("close", (code, reason) => {
490    closeInfo = { code, reason: reason.toString() };
491    for (const w of closeWaiters.splice(0)) {
492      clearTimeout(w.timer);
493      w.resolve(closeInfo!);
494    }
495    settleWaiters();
496  });
497 
498  await new Promise<void>((resolve, reject) => {
499    const t = setTimeout(() => reject(new Error("connect timeout")), 5_000);
500    ws.once("open", () => {
501      clearTimeout(t);
502      resolve();
503    });
504    ws.once("error", (err) => {
505      clearTimeout(t);
506      reject(err instanceof Error ? err : new Error(String(err)));
507    });
508  });
509 
510  const addWaiter = (
511    pred: ((f: Record<string, unknown>) => boolean) | null,
512    need: number,
513    timeoutMs: number,
514    label: string,
515  ): Promise<Record<string, unknown>[]> =>
516    new Promise((resolve, reject) => {
517      // Fast path: already enough unconsumed matches in the queue.
518      const matched = findMatches(pred, need);
519      if (matched.length >= need) {
520        resolve(consume(matched));
521        return;
522      }
523      const w: Waiter = {
524        label,
525        pred,
526        need,
527        timer: setTimeout(() => {
528          const idx = waiters.indexOf(w);
529          if (idx >= 0) waiters.splice(idx, 1);
530          const got = textQueue.filter((q) => !q.consumed).length;
531          reject(
532            new Error(
533              `timeout waiting for ${label} (${got} unconsumed frames, socket readyState=${ws.readyState})`,
534            ),
535          );
536        }, timeoutMs),
537        resolve,
538        reject,
539      };
540      waiters.push(w);
541      settleWaiters();
542    });
543 
544  const nextFrames = (
545    type: string,
546    n: number,
547    timeoutMs = 8_000,
548  ): Promise<Record<string, unknown>[]> =>
549    addWaiter((f) => f.type === type, n, timeoutMs, `${n} × "${type}"`);
550  const nextFrame = (type: string, timeoutMs = 8_000): Promise<Record<string, unknown>> =>
551    addWaiter((f) => f.type === type, 1, timeoutMs, `"${type}"`).then(
552      (frames) => frames[frames.length - 1]!,
553    );
554  const nextFrameWhere = (
555    pred: (f: Record<string, unknown>) => boolean,
556    timeoutMs = 8_000,
557  ): Promise<Record<string, unknown>> =>
558    addWaiter(pred, 1, timeoutMs, "frame matching predicate").then(
559      (frames) => frames[frames.length - 1]!,
560    );
561 
562  return {
563    ws,
564    textFrames: () => textQueue.map((q) => q.frame),
565    binaryFrames: () => binQueue.map((q) => q.bin),
566    nextFrames,
567    nextFrame,
568    nextFrameWhere,
569    nextBinary: (timeoutMs = 8_000) =>
570      new Promise((resolve, reject) => {
571        // The next unconsumed binary frame (never the same one twice).
572        const already = binQueue.find((q) => !q.consumed);
573        if (already) {
574          already.consumed = true;
575          resolve(already.bin);
576          return;
577        }
578        const timer = setTimeout(
579          () => reject(new Error("timeout waiting for a binary frame")),
580          timeoutMs,
581        );
582        const onMsg = (data: Buffer | ArrayBuffer | Buffer[], isBinary: boolean): void => {
583          if (!isBinary) return;
584          const b = Buffer.isBuffer(data)
585            ? new Uint8Array(data)
586            : new Uint8Array(data as ArrayBuffer);
587          clearTimeout(timer);
588          ws.off("message", onMsg);
589          resolve({ kind: b[0] === 2 ? "awareness" : "update", bytes: b.subarray(1) });
590        };
591        ws.on("message", onMsg);
592      }),
593    sendText: (obj) => ws.send(JSON.stringify(obj)),
594    sendBinary: (kind, bytes) => {
595      const out = new Uint8Array(bytes.byteLength + 1);
596      out[0] = kind;
597      out.set(bytes, 1);
598      ws.send(out);
599    },
600    closed: () => closeInfo,
601    onClose: (timeoutMs = 8_000) =>
602      new Promise((resolve, reject) => {
603        if (closeInfo !== null) {
604          resolve(closeInfo);
605          return;
606        }
607        const w = {
608          resolve,
609          timer: setTimeout(
610            () => reject(new Error("timeout waiting for close (socket still open)")),
611            timeoutMs,
612          ),
613        };
614        closeWaiters.push(w);
615      }),
616    isOpen: () => ws.readyState === WebSocket.OPEN,
617    close: (code = 1000) => ws.close(code),
618    terminate: () => ws.terminate(),
619  };
620}
621 
622// --- a RAW socket (the doc's "never pongs" heartbeat scenario) ----------------------------
623 
624export interface RawSocket {
625  readonly socket: net.Socket;
626  /** Unmasked server→client close-frame codes, in order. */
627  closeCodes: () => number[];
628  isOpen: () => boolean;
629  destroy: () => void;
630}
631 
632/**
633 * Hand-roll the WS upgrade on a bare TCP socket. Unlike the `ws` client
634 * (which auto-pongs), this socket NEVER answers the server's protocol
635 * pings — exactly the doc's "a socket silent 90 s is terminated with
636 * code 4000" scenario.
637 */
638export async function connectRawSocket(
639  base: { wsUrl: string },
640  opts: { token?: string; room?: string } = {},
641): Promise<RawSocket> {
642  const token = opts.token ?? mintToken({ sub: "u-raw", workspaces: ["w1"] });
643  const room = opts.room ?? "page:p-it";
644  const port = Number(new URL(base.wsUrl).port);
645  const socket = net.connect(port, "127.0.0.1");
646  const closeCodes: number[] = [];
647  let upgraded = false;
648 
649  const scanForCloseFrames = (buf: Buffer): void => {
650    for (let i = 0; i + 4 <= buf.length; i += 1) {
651      // A real unmasked close frame: FIN bit set, opcode 0x8, no mask
652      // bit (server→client is unmasked), small payload, and a code in
653      // the valid close range (1000–4999). Without the FIN + range
654      // checks the scanner false-positives on the 101-Upgrade response
655      // body (recorded test bug).
656      const b0 = buf[i]!;
657      const b1 = buf[i + 1]!;
658      if ((b0 & 0x80) !== 0x80) continue; // FIN not set — not a frame start
659      if ((b0 & 0x0f) !== 0x8) continue; // not a close frame
660      if ((b1 & 0x80) !== 0) continue; // masked — not a server frame
661      if (b1 >= 126) continue; // payload > 125 bytes — not a close frame
662      const code = (buf[i + 2]! << 8) | buf[i + 3]!;
663      if (code >= 1000 && code <= 4999) closeCodes.push(code);
664    }
665  };
666 
667  await new Promise<void>((resolve, reject) => {
668    const t = setTimeout(() => reject(new Error("raw upgrade timeout")), 5_000);
669    socket.on("connect", () => {
670      socket.write(
671        [
672          `GET /realtime?token=${encodeURIComponent(token)}&room=${encodeURIComponent(room)} HTTP/1.1`,
673          `Host: 127.0.0.1:${port}`,
674          "Upgrade: websocket",
675          "Connection: Upgrade",
676          "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==",
677          "Sec-WebSocket-Version: 13",
678          "",
679          "",
680        ].join("\r\n"),
681      );
682    });
683    socket.on("data", (buf) => {
684      scanForCloseFrames(buf);
685      if (!upgraded) {
686        const head = buf.subarray(0, 60).toString("utf8");
687        if (head.includes("101")) {
688          upgraded = true;
689          clearTimeout(t);
690          resolve();
691        } else if (/^HTTP\/1\.[01] [45]\d\d/.test(head)) {
692          clearTimeout(t);
693          reject(new Error(`raw upgrade refused: ${head.split("\r\n")[0]}`));
694        }
695      }
696    });
697    socket.on("error", (err) => {
698      clearTimeout(t);
699      reject(err);
700    });
701  });
702 
703  return {
704    socket,
705    closeCodes: () => [...closeCodes],
706    isOpen: () => !socket.destroyed,
707    destroy: () => socket.destroy(),
708  };
709}
710 
711/** Fetch a JSON probe endpoint. */
712export async function fetchJson(
713  url: string,
714): Promise<{ status: number; body: Record<string, unknown> }> {
715  const res = await fetch(url);
716  let body: Record<string, unknown> = {};
717  try {
718    body = (await res.json()) as Record<string, unknown>;
719  } catch {
720    /* non-JSON body */
721  }
722  return { status: res.status, body };
723}
724 
725// --- raw HTTP upgrade probe (pre-join handshake rejections) ------------------------
726 
727/**
728 * Issue the WS upgrade request over a bare TCP socket and return the HTTP
729 * response status (the handshake rejections happen AT the upgrade —
730 * 401/403/400/429 — before any WebSocket close frame exists, R1).
731 */
732export async function probeUpgrade(
733  base: { wsUrl: string },
734  opts: { token?: string; room?: string; path?: string; headers?: Record<string, string> } = {},
735): Promise<{ status: number; line: string }> {
736  const token = opts.token ?? mintToken({ sub: "u-probe", workspaces: ["w1"] });
737  const room = opts.room ?? "page:p-it";
738  const path =
739    opts.path ?? `/realtime?token=${encodeURIComponent(token)}&room=${encodeURIComponent(room)}`;
740  const port = Number(new URL(base.wsUrl).port);
741  const socket = net.connect(port, "127.0.0.1");
742  try {
743    const allHeaders = { "x-forwarded-for": autoTestIp(), ...opts.headers };
744    const extra = Object.entries(allHeaders)
745      .map(([k, v]) => `${k}: ${v}`)
746      .join("\r\n");
747    await new Promise<void>((resolve) => {
748      socket.once("connect", () => resolve());
749    });
750    // The two trailing empty strings are MANDATORY: they form the
751    // \r\n\r\n that terminates the HTTP header block (a .filter that
752    // drops empty lines would leave the request unterminated — the
753    // server waits for more data forever; recorded test bug).
754    const lines = [
755      `GET ${path} HTTP/1.1`,
756      `Host: 127.0.0.1:${port}`,
757      "Upgrade: websocket",
758      "Connection: Upgrade",
759      "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==",
760      "Sec-WebSocket-Version: 13",
761    ];
762    if (extra !== "") lines.push(extra);
763    lines.push("", "");
764    socket.write(lines.join("\r\n"));
765    let buf = "";
766    await new Promise<void>((resolve, reject) => {
767      const t = setTimeout(() => {
768        socket.destroy();
769        reject(new Error("probe timeout"));
770      }, 5_000);
771      const finish = (): void => {
772        clearTimeout(t);
773        resolve();
774      };
775      socket.once("data", (d) => {
776        buf += d.toString("utf8");
777        finish();
778      });
779      // The wrong-path case: the server destroys the socket WITHOUT an
780      // HTTP response (EOF — no 'data', but 'end'/'close' still fire).
781      socket.once("end", finish);
782      socket.once("close", finish);
783      socket.once("error", reject);
784    });
785    const line = buf.split("\r\n")[0] ?? "";
786    const m = line.match(/HTTP\/1\.[01] (\d{3})/);
787    return { status: Number(m?.[1] ?? 0), line };
788  } finally {
789    socket.destroy();
790  }
791}
792 
793// --- Yjs test updates ----------------------------------------------------------------
794 
795/** Append `text` to the Yjs doc's `yText` type and return the update bytes
796 *  (the `update` binary frame payload for the 0x01 prefix). The delta
797 *  against the pre-edit state vector keeps the log small + the relay
798 *  honest (a full-state update would be idempotent too, but wasteful). */
799export function makeTextUpdate(doc: Y.Doc, text: string): Uint8Array {
800  const yText = doc.getText("yText");
801  const before = Y.encodeStateAsUpdate(doc);
802  yText.insert(0, text);
803  // `encodeStateAsUpdate(doc, sv)` is already binary — no re-encode.
804  return Y.encodeStateAsUpdate(doc, before);
805}
806 
807export { WS_CLOSE };
808