provider.test.ts

provider.test.tspackages/collaboration/src/provider.test.tsraw ← src11,081 B← Back to the app
1/**
2 * TASK-0092 R4/R5 — the RealtimeProvider connectivity + token-refresh state
3 * machine (the spec's named unit test: "the connectivity state machine
4 * (online→offline→online transitions, the token-expiry reconnect)").
5 *
6 * A driven fake `WSAdapter` (the provider never touches a real socket) +
7 * an injectable timer queue (the reconnect backoff is real logic, not wall
8 * time — the test flushes it manually) + a no-op offline factory (IndexedDB
9 * is a browser concern; the state machine under test is connectivity, not
10 * persistence).
11 */
12import { describe, expect, it } from "vitest";
13import * as Y from "yjs";
14import { createRealtimeProvider } from "./provider.js";
15import {
16  WS_CLOSE,
17  type ConnectionState,
18  type RealtimeProviderConfig,
19  type WSAdapter,
20  type YjsDocBinding,
21} from "./types.js";
22 
23// --- fakes -----------------------------------------------------------------
24 
25/** A Yjs doc binding over a real (empty) Y.Doc — the provider's update
26 *  source/sink; the test only needs the lifecycle callbacks, not ProseMirror. */
27function makeBinding(): YjsDocBinding {
28  const doc = new Y.Doc();
29  let localCb: ((u: Uint8Array) => void) | null = null;
30  return {
31    doc,
32    applyUpdate: (u) => {
33      if (u.length > 0) Y.applyUpdate(doc, u, "net");
34    },
35    onLocalUpdate: (cb) => {
36      localCb = cb;
37      return () => {
38        localCb = null;
39      };
40    },
41    // Force the empty-doc path (a full `sync.request {}`, not a state-vector
42    // diff) — the connectivity tests don't care about seeding.
43    hasLocalState: () => false,
44    stateVector: () => Y.encodeStateVector(doc),
45    encodeStateAsUpdate: () => Y.encodeStateAsUpdate(doc)
46  };
47}
48 
49/** A controllable WebSocket: the test drives open/message/close. */
50class FakeSocket implements WSAdapter {
51  readyState = 0; // CONNECTING
52  onopen: (() => void) | null = null;
53  onmessage: ((e: { data: string | ArrayBuffer | Uint8Array }) => void) | null = null;
54  onclose: ((e: { code: number; reason: string }) => void) | null = null;
55  onerror: ((e: unknown) => void) | null = null;
56  readonly sent: Array<string | ArrayBuffer | Uint8Array> = [];
57  readonly url: string;
58  constructor(url: string) {
59    this.url = url;
60  }
61  send(data: string | ArrayBuffer | Uint8Array): void {
62    this.sent.push(data);
63  }
64  close(): void {
65    this.readyState = 3;
66  }
67  // test drivers
68  _open(): void {
69    this.readyState = 1;
70    this.onopen?.();
71  }
72  _text(obj: unknown): void {
73    this.onmessage?.({ data: JSON.stringify(obj) });
74  }
75  _close(code: number, reason = ""): void {
76    this.readyState = 3;
77    this.onclose?.({ code, reason });
78  }
79}
80 
81interface RigOverrides {
82  isOnline?: () => boolean;
83  refreshAccessToken?: () => Promise<string>;
84  onClosed?: (code: number, hint: string) => void;
85}
86 
87interface Rig {
88  provider: ReturnType<typeof createRealtimeProvider>;
89  sockets: FakeSocket[];
90  states: ConnectionState[];
91  /** Flush all pending reconnect timers (the injected queue). */
92  tick: () => void;
93  refreshCalls: () => number;
94  setOnline: (b: boolean) => void;
95}
96 
97/** Build the provider with a MANUAL timer queue (the test flushes it — the
98 *  reconnect backoff is exercised without wall-clock sleeps). */
99function makeRig(overrides: RigOverrides = {}): Rig {
100  const sockets: FakeSocket[] = [];
101  const states: ConnectionState[] = [];
102  let timerQueue: Array<() => void> = [];
103  let refreshCalls = 0;
104  let online = overrides.isOnline ? overrides.isOnline() : true;
105 
106  const config: RealtimeProviderConfig = {
107    pageId: "pg_1",
108    workspaceId: "ws_1",
109    wsUrl: "ws://realtime.test",
110    // After a refresh the handshake uses the new token (the 4001-retry test
111    // asserts the fresh socket URL carries it).
112    getToken: () => (refreshCalls > 0 ? "tok-refreshed" : "tok"),
113    refreshAccessToken: overrides.refreshAccessToken ?? (async () => {
114      refreshCalls += 1;
115      return "tok-refreshed";
116    }),
117    binding: makeBinding(),
118    createSocket: (url) => {
119      const s = new FakeSocket(url);
120      sockets.push(s);
121      return s;
122    },
123    isOnline: () => online,
124    // No IndexedDB in the state-machine test (the browser default would try).
125    offlineFactory: {
126      create: async (docName: string) => ({ docName, flush: async () => {}, dispose: async () => {} })
127    },
128    // A 1000-min app-level ping — the test never flushes it.
129    pingIntervalMs: 60_000_000,
130    onStateChange: (s) => {
131      states.push(s);
132    },
133    // exactOptionalPropertyTypes: only include onClosed when provided.
134    ...(overrides.onClosed !== undefined ? { onClosed: overrides.onClosed } : {}),
135    // The injectable timer queue (the reconnect backoff, flushed by tick).
136    setTimeout: ((fn: () => void) => {
137      timerQueue.push(fn);
138      return 0 as unknown as ReturnType<typeof globalThis.setTimeout>;
139    }) as unknown as typeof globalThis.setTimeout,
140    clearTimeout: (() => {}) as unknown as typeof globalThis.clearTimeout
141  };
142  const provider = createRealtimeProvider(config);
143  return {
144    provider,
145    sockets,
146    states,
147    refreshCalls: () => refreshCalls,
148    tick: () => {
149      const q = timerQueue;
150      timerQueue = [];
151      for (const fn of q) fn();
152    },
153    setOnline: (b: boolean) => {
154      online = b;
155      provider.setOnline(b);
156    }
157  };
158}
159 
160/** Connect → open → hello → sync.done (a live `connected` session). */
161async function reachConnected(rig: Rig): Promise<FakeSocket> {
162  await rig.provider.connect();
163  const s = rig.sockets[0]!;
164  s._open();
165  s._text({ type: "hello", connId: "c1", canEdit: true, roomPeers: [] });
166  s._text({ type: "sync.state", chunk: "", done: true });
167  expect(rig.provider.connectionState).toBe("connected");
168  return s;
169}
170 
171/** Flush microtasks (the 4001-refresh path is async: `await
172 *  refreshAccessToken()` runs before `scheduleReconnect`). The test's own
173 *  global setTimeout is the REAL one (only the provider's config timer is
174 *  injected), so a 0 ms real timer is the flush point. */
175function flushMicrotasks(): Promise<void> {
176  return new Promise((r) => globalThis.setTimeout(r, 0));
177}
178 
179describe("RealtimeProvider — connectivity + token-refresh state machine (R4/R5)", () => {
180  it("online → connecting → syncing → connected (the happy path, empty doc → full sync)", async () => {
181    const rig = makeRig();
182    await rig.provider.connect();
183    const s = rig.sockets[0]!;
184    s._open();
185    s._text({ type: "hello", connId: "c1", canEdit: true, roomPeers: [] });
186    // hello → the empty doc sends a FULL sync.request (no state vector to
187    // diff) before any sync.state arrives.
188    const req = s.sent.find((x) => typeof x === "string" && JSON.parse(x).type === "sync.request") as string;
189    expect(JSON.parse(req)).toEqual({ type: "sync.request" });
190    s._text({ type: "sync.state", chunk: "", done: true });
191    expect(rig.states).toEqual(["connecting", "syncing", "connected"]);
192    expect(rig.provider.connectionState).toBe("connected");
193    await rig.provider.dispose();
194  });
195 
196  it("offline → 'offline' (no socket); setOnline(true) → reconnects (the online/offline transition)", async () => {
197    const rig = makeRig({ isOnline: () => false });
198    await rig.provider.connect();
199    // No socket was opened while offline.
200    expect(rig.sockets).toHaveLength(0);
201    expect(rig.states).toEqual(["offline"]);
202    // Back online → the provider reconnects (a new socket). connect() is
203    // async (the offline factory resolves before openSocket) → flush.
204    rig.setOnline(true);
205    await flushMicrotasks();
206    expect(rig.sockets).toHaveLength(1);
207    expect(rig.states).toContain("connecting");
208    await rig.provider.dispose();
209  });
210 
211  it("a 4001 (expired token) mid-session → refreshAccessToken + reconnect ONCE, then connected", async () => {
212    const rig = makeRig();
213    await reachConnected(rig);
214    // The token expires → the server closes 4001.
215    rig.sockets[0]!._close(WS_CLOSE.Unauthorized);
216    await flushMicrotasks(); // the async refresh + scheduleReconnect(0)
217    expect(rig.states).toContain("reconnecting");
218    // The refresh was requested exactly once…
219    expect(rig.refreshCalls()).toBe(1);
220    // …and the reconnect timer (scheduleReconnect(0) after a successful
221    // refresh) fires a fresh socket.
222    rig.tick();
223    expect(rig.sockets).toHaveLength(2);
224    // The fresh socket uses the REFRESHED token (the handshake URL carries it).
225    expect(rig.sockets[1]!.url).toContain(encodeURIComponent("tok-refreshed"));
226    rig.sockets[1]!._open();
227    rig.sockets[1]!._text({ type: "hello", connId: "c2", canEdit: true, roomPeers: [] });
228    rig.sockets[1]!._text({ type: "sync.state", chunk: "", done: true });
229    expect(rig.provider.connectionState).toBe("connected");
230    await rig.provider.dispose();
231  });
232 
233  it("a SECOND 4001 after a refresh → terminal (not a token problem): closed, no reconnect loop", async () => {
234    let closedCode: number | null = null;
235    const rig = makeRig({ onClosed: (code) => (closedCode = code) });
236    await reachConnected(rig);
237    // First 4001 → refresh (hello reset reconnectAttempts/refreshedOnce).
238    rig.sockets[0]!._close(WS_CLOSE.Unauthorized);
239    await flushMicrotasks(); // the async refresh resolves
240    rig.tick(); // scheduleReconnect(0) → new socket
241    const s2 = rig.sockets[1]!;
242    // The refreshed token is STILL rejected (a non-token problem) → 4001 again.
243    s2._open();
244    s2._close(WS_CLOSE.Unauthorized);
245    // Terminal: closed, no third socket, no further refresh.
246    expect(rig.provider.connectionState).toBe("closed");
247    expect(rig.sockets).toHaveLength(2);
248    expect(rig.refreshCalls()).toBe(1);
249    expect(closedCode).toBe(WS_CLOSE.Unauthorized);
250    // Flushing timers does nothing (closedByUs stops the loop).
251    rig.tick();
252    expect(rig.sockets).toHaveLength(2);
253  });
254 
255  it("a transient close (4000 protocol/heartbeat) → reconnect with backoff, recovers", async () => {
256    const rig = makeRig();
257    await reachConnected(rig);
258    // A protocol/heartbeat close (4000) — a transient network blip.
259    rig.sockets[0]!._close(WS_CLOSE.ProtocolOrHeartbeat);
260    expect(rig.states).toContain("reconnecting");
261    // The backoff timer fires a new socket.
262    rig.tick();
263    expect(rig.sockets).toHaveLength(2);
264    rig.sockets[1]!._open();
265    rig.sockets[1]!._text({ type: "hello", connId: "c1b", canEdit: true, roomPeers: [] });
266    rig.sockets[1]!._text({ type: "sync.state", chunk: "", done: true });
267    expect(rig.provider.connectionState).toBe("connected");
268    await rig.provider.dispose();
269  });
270 
271  it("4002 (forbidden, lost access) → terminal closed (the app navigates away)", async () => {
272    let closedCode: number | null = null;
273    const rig = makeRig({ onClosed: (code) => (closedCode = code) });
274    await rig.provider.connect();
275    rig.sockets[0]!._open();
276    rig.sockets[0]!._close(WS_CLOSE.Forbidden);
277    expect(rig.provider.connectionState).toBe("closed");
278    expect(closedCode).toBe(WS_CLOSE.Forbidden);
279    rig.tick();
280    expect(rig.sockets).toHaveLength(1); // no reconnect
281  });
282});
283