presence.test.ts

presence.test.tstests/integration/realtime/presence.test.tsraw ← realtime19,425 B← Back to the app
1/**
2 * Realtime integration (TASK-0093 slice 2 — presence + live cursors): the
3 * server's presence producer over the (b) wire — JSON control frames
4 * (`presence` client → server; `presence` / `presence-leave` server →
5 * client), NOT the binary 0x02 awareness relay (that stays for the
6 * TASK-0136 provider path; the conflict is recorded in the task file).
7 *
8 * Boots the REAL `createRealtimeServer` (rt-helpers) with real `ws`
9 * clients. Coverage (the task spec R1–R5 + the (b) producer):
10 *  - R1: A joins a page room → B's join re-broadcast carries A's full
11 *    state; A's `presence` frame (cursor move) reaches B within 500 ms;
12 *    a third client on a DIFFERENT page receives nothing.
13 *  - R2: identity is server-stamped (the client frame carries ONLY
14 *    cursor/viewing — the broadcast's `user` comes from the meta seam's
15 *    display name + the shared presenceColor hash; a client-supplied
16 *    name is impossible by schema).
17 *  - R3: > 50 peers → avatars-only (cursor/viewing stripped); the
18 *    boundary is 50 full / 51 stripped.
19 *  - R4: leave propagation < 500 ms — an explicit `presence-leave` AND a
20 *    socket close both announce `presence-leave` to the room.
21 *  - R5: the 90 s prune (a short `presencePruneMs` + the fake clock)
22 *    drops a silent peer and announces it.
23 *  - (b) fan-out: a roomless CONTROL socket subscribed to `page:<id>`
24 *    receives the room's broadcasts (and its own presence updates reach
25 *    the doc-room peers) — the two transports interoperate.
26 *
27 * Docker-free (the spec's 3-client E2E twin is Docker-gated under
28 * `tests/e2e` — same scenario, real browser).
29 */
30import { afterAll, beforeAll, describe, expect, it } from "vitest";
31 
32import { presenceColor } from "@factory/shared/presence";
33 
34import { buildRealtimeFixture, connectClient, mintToken, type RealtimeFixture } from "./rt-helpers";
35 
36/** The fixture's distinct display names (the meta seam's userNames opt). */
37const NAMES: Record<string, string> = {
38  "u-ana": "Ana",
39  "u-bob": "Bob",
40  "u-carol": "Carol",
41};
42 
43async function connectUser(fx: RealtimeFixture, sub: string, room: string) {
44  const token = mintToken({ sub, workspaces: ["w1"] });
45  return connectClient(fx, {
46    token,
47    url: `${fx.wsUrl}/realtime?token=${encodeURIComponent(token)}&room=${encodeURIComponent(room)}`,
48  });
49}
50 
51/** A token-only (roomless) control client — the (b) web handshake. */
52async function connectControl(fx: RealtimeFixture, sub: string) {
53  const token = mintToken({ sub, workspaces: ["w1"] });
54  return connectClient(fx, {
55    token,
56    url: `${fx.wsUrl}/realtime?token=${encodeURIComponent(token)}`,
57  });
58}
59 
60const presenceOf = (f: Record<string, unknown>) =>
61  f as {
62    type: string;
63    pageId?: string;
64    user?: { id: string; name: string; color: string };
65    cursor?: { blockId: string; offset: number } | null;
66    viewing?: boolean;
67    lastSeen?: number;
68  };
69 
70describe("TASK-0093 — presence fan-out (doc-room transport)", () => {
71  let fx: RealtimeFixture;
72 
73  beforeAll(async () => {
74    fx = await buildRealtimeFixture({
75      pages: new Set(["p-pres", "p-other", "p-cap", "p-ctl"]),
76      userNames: NAMES,
77    });
78  });
79 
80  afterAll(async () => {
81    await fx.stop();
82  });
83 
84  it("R1+R2: join re-broadcast + cursor move < 500 ms, other pages isolated, identity server-stamped", async () => {
85    const a = await connectUser(fx, "u-ana", "page:p-pres");
86    const b = await connectUser(fx, "u-bob", "page:p-pres");
87    const c = await connectUser(fx, "u-carol", "page:p-other");
88    try {
89      const ha = await a.nextFrame("hello");
90      expect(ha.roomPeers).toEqual([]);
91      const hb = await b.nextFrame("hello");
92      // B's hello: A is already in the room (the roster, pre-existing).
93      expect(hb.roomPeers).toHaveLength(1);
94 
95      // A's join re-broadcast reached B (the joiner's initial state — no
96      // cursor, viewing=true, server-stamped identity).
97      const aSeenByB = presenceOf(
98        await b.nextFrameWhere(
99          (f) => f.type === "presence" && (f as { user?: { id?: string } }).user?.id === "u-ana",
100        ),
101      );
102      expect(aSeenByB.type).toBe("presence");
103      expect(aSeenByB.pageId).toBe("page:p-pres");
104      const aUser = aSeenByB.user!;
105      // R2: the identity is SERVER-STAMPED — the display name from the
106      // meta seam (the client frame carries no user field at all) + the
107      // shared hash's color.
108      expect(aUser.name).toBe("Ana");
109      expect(aUser.color).toBe(presenceColor("u-ana"));
110      expect(aSeenByB.cursor).toBeNull();
111      expect(aSeenByB.viewing).toBe(true);
112      expect(typeof aSeenByB.lastSeen).toBe("number");
113 
114      // R1: A moves its cursor → B receives it within the 500 ms
115      // contract (in-process the round-trip is single-digit ms).
116      const t0 = Date.now();
117      a.sendText({ type: "presence", cursor: { blockId: "blk_1", offset: 4 }, viewing: true });
118      const moved = await b.nextFrameWhere(
119        (f) => f.type === "presence" && (f as { user?: { id?: string } }).user?.id === "u-ana",
120      );
121      const latencyMs = Date.now() - t0;
122      expect(latencyMs).toBeLessThan(500);
123      expect(moved.cursor).toEqual({ blockId: "blk_1", offset: 4 });
124      expect(moved.viewing).toBe(true);
125 
126      // R2 (b): the SENDER gets no echo of its own state (it knows it).
127      expect(
128        a
129          .textFrames()
130          .filter(
131            (f) => f.type === "presence" && (f as { user?: { id?: string } }).user?.id === "u-ana",
132          ),
133      ).toHaveLength(0);
134 
135      // Isolation: C (a different page) received NO presence frames.
136      expect(
137        c.textFrames().filter((f) => f.type === "presence" || f.type === "presence-leave"),
138      ).toHaveLength(0);
139    } finally {
140      a.close();
141      b.close();
142      c.close();
143      await Promise.all([a.onClose(), b.onClose(), c.onClose()]);
144    }
145  });
146 
147  it("R4: an explicit presence-leave reaches the room in < 500 ms", async () => {
148    const a = await connectUser(fx, "u-ana", "page:p-pres");
149    const b = await connectUser(fx, "u-bob", "page:p-pres");
150    try {
151      await a.nextFrame("hello");
152      await b.nextFrame("hello");
153      // B's join re-broadcast: B received A's state (the joiner hears the
154      // existing peers).
155      await b.nextFrameWhere(
156        (f) => f.type === "presence" && (f as { user?: { id?: string } }).user?.id === "u-ana",
157      );
158      const t0 = Date.now();
159      a.sendText({ type: "presence-leave" });
160      const leave = await b.nextFrame("presence-leave");
161      const latencyMs = Date.now() - t0;
162      expect(latencyMs).toBeLessThan(500);
163      expect(leave.pageId).toBe("page:p-pres");
164      expect(leave.userId).toBe("u-ana");
165    } finally {
166      a.close();
167      b.close();
168      await Promise.all([a.onClose(), b.onClose()]);
169    }
170  });
171 
172  it("R4: a socket close announces the peer's leave", async () => {
173    const a = await connectUser(fx, "u-ana", "page:p-pres");
174    const b = await connectUser(fx, "u-bob", "page:p-pres");
175    try {
176      await a.nextFrame("hello");
177      await b.nextFrame("hello");
178      // B's join re-broadcast (so B has A's state first — the leave must
179      // be a distinct frame).
180      await b.nextFrameWhere(
181        (f) => f.type === "presence" && (f as { user?: { id?: string } }).user?.id === "u-ana",
182      );
183      a.close();
184      const leave = await b.nextFrame("presence-leave");
185      expect(leave.userId).toBe("u-ana");
186      expect(leave.pageId).toBe("page:p-pres");
187    } finally {
188      b.close();
189      await b.onClose();
190    }
191  });
192 
193  it("R3: the 50-peer cap — 50 peers carry full cursors, the 51st is avatars-only", async () => {
194    // 49 peers join (u-cap-0..48) + the observer (u-ana) = 50 = the cap.
195    const peers: Awaited<ReturnType<typeof connectUser>>[] = [];
196    const observer = await connectUser(fx, "u-ana", "page:p-cap");
197    await observer.nextFrame("hello");
198    try {
199      for (let i = 0; i < 49; i += 1) {
200        const p = await connectUser(fx, `u-cap-${i}`, "page:p-cap");
201        await p.nextFrame("hello");
202        peers.push(p);
203      }
204      // The observer's JOIN re-broadcast (initial state, cursor null)
205      // reached the peers — consume it so the cursor update below is the
206      // FIRST remaining u-ana frame (the discriminator is the cursor).
207      await peers[0]!.nextFrameWhere(
208        (f) => f.type === "presence" && (f as { user?: { id?: string } }).user?.id === "u-ana",
209      );
210 
211      // A cursor update from the observer (50 peers — exactly at the
212      // cap) is carried FULL (the cap strips only ABOVE 50).
213      observer.sendText({ type: "presence", cursor: { blockId: "blk_9", offset: 1 } });
214      const full = await peers[0]!.nextFrameWhere(
215        (f) =>
216          f.type === "presence" &&
217          (f as { user?: { id?: string } }).user?.id === "u-ana" &&
218          (f as { cursor?: { blockId?: string } | null }).cursor?.blockId === "blk_9",
219      );
220      expect(full.cursor).toEqual({ blockId: "blk_9", offset: 1 });
221 
222      // The 51st peer joins → the room is over the cap: its INITIAL
223      // state (the join re-broadcast) arrives avatars-only.
224      const over = await connectUser(fx, "u-cap-51", "page:p-cap");
225      await over.nextFrame("hello");
226      const stripped = presenceOf(
227        await over.nextFrameWhere(
228          (f) => f.type === "presence" && (f as { user?: { id?: string } }).user?.id === "u-ana",
229        ),
230      );
231      expect(stripped.cursor).toBeNull();
232      expect(stripped.viewing).toBeUndefined();
233      expect(stripped.user?.name).toBe("Ana"); // the avatar stays
234 
235      // …and a subsequent cursor update from the observer is also
236      // stripped (the cap is live, not just at join). The sleep keeps the
237      // upsert's `lastSeen` on a later millisecond (the frame
238      // discriminator below).
239      await new Promise((r) => setTimeout(r, 10));
240      observer.sendText({ type: "presence", cursor: { blockId: "blk_9", offset: 2 } });
241      const stripped2 = await over.nextFrameWhere(
242        (f) =>
243          f.type === "presence" &&
244          (f as { user?: { id?: string } }).user?.id === "u-ana" &&
245          (f as { lastSeen?: number }).lastSeen !== stripped.lastSeen,
246      );
247      expect(stripped2.cursor).toBeNull();
248 
249      // Dropping back to ≤ 50 (the 51st leaves) restores full cursors.
250      over.close();
251      await over.onClose();
252      await new Promise((r) => setTimeout(r, 10));
253      observer.sendText({ type: "presence", cursor: { blockId: "blk_9", offset: 3 } });
254      const full2 = await peers[0]!.nextFrameWhere(
255        (f) =>
256          f.type === "presence" &&
257          (f as { user?: { id?: string } }).user?.id === "u-ana" &&
258          (f as { lastSeen?: number }).lastSeen !== stripped2.lastSeen,
259      );
260      expect(full2.cursor).toEqual({ blockId: "blk_9", offset: 3 });
261    } finally {
262      for (const p of peers) {
263        p.close();
264      }
265      observer.close();
266      await Promise.all([...peers.map((p) => p.onClose()), observer.onClose()]);
267    }
268  });
269});
270 
271describe("TASK-0093 — the (b) control-socket transport", () => {
272  let fx: RealtimeFixture;
273 
274  beforeAll(async () => {
275    fx = await buildRealtimeFixture({
276      pages: new Set(["p-ctl"]),
277      userNames: NAMES,
278    });
279  });
280 
281  afterAll(async () => {
282    await fx.stop();
283  });
284 
285  it("a roomless control socket subscribed to page:<id> receives the room's presence + its own updates reach the doc-room peers", async () => {
286    const doc = await connectUser(fx, "u-bob", "page:p-ctl");
287    const ctl = await connectControl(fx, "u-ana");
288    try {
289      const hdoc = await doc.nextFrame("hello");
290      expect(hdoc.roomPeers).toEqual([]);
291      const hctl = await ctl.nextFrame("hello");
292      expect(hctl.roomPeers).toEqual([]);
293 
294      // The control socket opens the page channel (the (b) handshake).
295      ctl.sendText({ type: "subscribe", channel: "page:p-ctl" });
296 
297      // 1) The doc-room peer's state reaches the control socket: Bob's
298      //    join re-broadcast (his initial state) lands on Ana's control
299      //    socket (it's subscribed to the room).
300      const bobSeenByCtl = presenceOf(
301        await ctl.nextFrameWhere(
302          (f) => f.type === "presence" && (f as { user?: { id?: string } }).user?.id === "u-bob",
303        ),
304      );
305      expect(bobSeenByCtl.pageId).toBe("page:p-ctl");
306      expect(bobSeenByCtl.user?.name).toBe("Bob");
307      expect(bobSeenByCtl.cursor).toBeNull();
308 
309      // 2) The control socket's OWN presence reaches the doc-room peer
310      //    (the server stamps Ana's identity from the meta seam's
311      //    `kind: "user"` path — no handshake meta for a roomless
312      //    socket). Ana's JOIN re-broadcast (initial state, cursor
313      //    null) reached the doc peer first — consume it, then assert
314      //    the cursor update (the discriminator is the cursor).
315      await doc.nextFrameWhere(
316        (f) => f.type === "presence" && (f as { user?: { id?: string } }).user?.id === "u-ana",
317      );
318      const t0 = Date.now();
319      ctl.sendText({ type: "presence", cursor: { blockId: "blk_c", offset: 2 }, viewing: true });
320      const anaSeenByDoc = presenceOf(
321        await doc.nextFrameWhere(
322          (f) =>
323            f.type === "presence" &&
324            (f as { user?: { id?: string } }).user?.id === "u-ana" &&
325            (f as { cursor?: { blockId?: string } | null }).cursor?.blockId === "blk_c",
326        ),
327      );
328      expect(Date.now() - t0).toBeLessThan(500);
329      expect(anaSeenByDoc.user?.name).toBe("Ana");
330      expect(anaSeenByDoc.user?.color).toBe(presenceColor("u-ana"));
331      expect(anaSeenByDoc.cursor).toEqual({ blockId: "blk_c", offset: 2 });
332 
333      // 3) Ana's leave (the explicit frame) is announced to the doc peer.
334      ctl.sendText({ type: "presence-leave" });
335      const leave = await doc.nextFrame("presence-leave");
336      expect(leave.userId).toBe("u-ana");
337      expect(leave.pageId).toBe("page:p-ctl");
338 
339      // 4) A control socket with NO page channel gets nothing (a
340      //    workspace-only subscription is not a presence room).
341      const ctl2 = await connectControl(fx, "u-carol");
342      try {
343        await ctl2.nextFrame("hello");
344        ctl2.sendText({ type: "subscribe", channel: "workspace:w1" });
345        ctl.sendText({ type: "presence", cursor: { blockId: "blk_c", offset: 3 } });
346        // Give any (buggy) cross-room frame a chance to arrive.
347        await new Promise((r) => setTimeout(r, 150));
348        expect(ctl2.textFrames().filter((f) => f.type === "presence")).toHaveLength(0);
349      } finally {
350        ctl2.close();
351        await ctl2.onClose();
352      }
353    } finally {
354      doc.close();
355      ctl.close();
356      await Promise.all([doc.onClose(), ctl.onClose()]);
357    }
358  });
359 
360  it("a control socket's presence follows the LAST subscribed page channel (page switch)", async () => {
361    const ctl = await connectControl(fx, "u-ana");
362    const doc = await connectUser(fx, "u-bob", "page:p-ctl");
363    try {
364      await ctl.nextFrame("hello");
365      await doc.nextFrame("hello");
366      ctl.sendText({ type: "subscribe", channel: "page:p-ctl" });
367      // Let Ana's join re-broadcast settle.
368      await doc.nextFrameWhere(
369        (f) => f.type === "presence" && (f as { user?: { id?: string } }).user?.id === "u-ana",
370      );
371 
372      // Ana switches pages: unsubscribe + resubscribe (the (b) client's
373      // page navigation). Her presence room moves; the old room gets a
374      // leave.
375      ctl.sendText({ type: "unsubscribe", channel: "page:p-ctl" });
376      const leave = await doc.nextFrame("presence-leave");
377      expect(leave.userId).toBe("u-ana");
378    } finally {
379      ctl.close();
380      doc.close();
381      await Promise.all([ctl.onClose(), doc.onClose()]);
382    }
383  });
384});
385 
386describe("TASK-0093 — R5 the 90 s prune (the fake clock)", () => {
387  let fx: RealtimeFixture;
388 
389  beforeAll(async () => {
390    // Short prune window (5 s on the fake clock) + a fast heartbeat
391    // cadence so the prune loop fires quickly on REAL time. The R4
392    // socket prune (pruneSilenceMs) stays long so the silent peer's
393    // SOCKET survives — only its PRESENCE entry is dropped (doc §5.1:
394    // the client heals a pruned peer by dropping it; the socket's
395    // `presence` re-asserts it).
396    // `now0` = the REAL current time (the fake clock starts "now" — the
397    // mitted tokens' `exp` (real now + 1 h) stays valid; `advance()`
398    // moves only the fake clock, so the prune window math is relative).
399    fx = await buildRealtimeFixture({
400      pages: new Set(["p-prune"]),
401      now0: Date.now(),
402      userNames: NAMES,
403      presencePruneMs: 5_000,
404      pingIntervalMs: 200,
405      pruneSilenceMs: 3600_000, // the socket prune is effectively off
406    });
407  });
408 
409  afterAll(async () => {
410    await fx.stop();
411  });
412 
413  it("a peer silent past the prune window is dropped + announced", async () => {
414    const a = await connectUser(fx, "u-ana", "page:p-prune");
415    const b = await connectUser(fx, "u-bob", "page:p-prune");
416    try {
417      await a.nextFrame("hello");
418      await b.nextFrame("hello");
419      // A's join re-broadcast (B now has A's state — the leave must be
420      // a distinct, later frame).
421      await b.nextFrameWhere(
422        (f) => f.type === "presence" && (f as { user?: { id?: string } }).user?.id === "u-ana",
423      );
424      // A asserts its cursor (lastSeen = now on the fake clock).
425      a.sendText({ type: "presence", cursor: { blockId: "blk_p", offset: 1 } });
426      await b.nextFrameWhere(
427        (f) =>
428          f.type === "presence" &&
429          (f as { user?: { id?: string } }).user?.id === "u-ana" &&
430          (f as { cursor?: unknown }).cursor !== null,
431      );
432 
433      // Keep A's entry FRESH while we set up (a live client re-asserts
434      // its presence periodically — the real web client sends it on
435      // selection changes; the prune loop runs every 200 ms on real
436      // time, so without this the 5 s window could expire mid-test).
437      fx.advance(1_000);
438      a.sendText({ type: "presence", cursor: { blockId: "blk_p", offset: 1 } });
439      await b.nextFrameWhere(
440        (f) =>
441          f.type === "presence" &&
442          (f as { user?: { id?: string } }).user?.id === "u-ana" &&
443          (f as { cursor?: unknown }).cursor !== null,
444      );
445 
446      // Now A goes silent: age it past the 5 s prune window (B keeps
447      // heartbeating, so only A's PRESENCE entry goes stale).
448      fx.advance(5_001);
449      b.sendText({ type: "ping", ts: 1 });
450      const leave = await b.nextFrame("presence-leave");
451      expect(leave.userId).toBe("u-ana");
452      expect(leave.pageId).toBe("page:p-prune");
453 
454      // A's socket SURVIVED the presence prune (the R4 socket prune is
455      // at the long threshold) — and a re-asserted `presence` puts A
456      // back in the roster (the doc §5.2 best-effort heal).
457      expect(a.isOpen()).toBe(true);
458      a.sendText({ type: "presence", cursor: { blockId: "blk_p", offset: 2 } });
459      const back = await b.nextFrameWhere(
460        (f) =>
461          f.type === "presence" &&
462          (f as { user?: { id?: string } }).user?.id === "u-ana" &&
463          (f as { cursor?: { blockId?: string } | null }).cursor?.blockId === "blk_p",
464      );
465      expect(back.cursor).toEqual({ blockId: "blk_p", offset: 2 });
466    } finally {
467      a.close();
468      b.close();
469      await Promise.all([a.onClose(), b.onClose()]);
470    }
471  });
472});
473