provider.ts · Source

provider.tspackages/collaboration/src/provider.tsraw ← src18,113 B← Back to the app
1/**
2 * @factory/collaboration — the realtime provider (TASK-0092 R4).
3 *
4 * A plain TS class (NOT a React component — the canvas's `EditorCanvas`
5 * instantiates it): connects to `wsUrl/realtime?token=<JWT>&room=page:{pageId}`,
6 * runs the TASK-0091 R3 protocol, drives the {@link YjsDocBinding}
7 * (y-prosemirror on the canvas, a fake in tests), and exposes the state
8 * machine the top-bar chip + offline banner consume.
9 *
10 * Protocol (client side — the authoritative schemas land in
11 * `packages/shared/src/realtime/frames.ts`, TASK-0091; this client is the
12 * lockstep mirror, see `types.ts`):
13 *
14 *   connect → `hello {connId, canEdit, serverTime, roomPeers}`
15 *     → `sync.request {}` (local doc empty = full) or
16 *       `sync.request {stateVector}` (local state = diff)
17 *       + one binary `0x01` update of the FULL local state (the
18 *       "ship all pending" step — the CRDT merge is idempotent)
19 *       + `outbox.pending {count}` (the drain-load signal, TASK-0095 R4.2)
20 *     → paged `sync.state {chunk, done}` (base64 Yjs updates, ≤ 64 KB)
21 *       applied via the binding; `done` → state `connected`
22 *   steady state:
23 *     → local binding updates → binary `0x01` + raw Yjs bytes (dropped
24 *       locally when `canEdit` is false → `onLocalUpdateBlocked`)
25 *     → 30 s app-level `ping {ts, lastSeq}` → `ping-pong`
26 *     → inbound binary `0x01` → `binding.applyUpdate`; `0x02` → the
27 *       canvas's y-protocols awareness handling (`onAwarenessUpdate`)
28 *
29 * Close codes (TASK-0091 R6): 4001 → refresh the access token (cookie
30 * refresh BFF) and reconnect ONCE (a second 4001 = terminal); 4002 →
31 * terminal (no access — `onClosed`, no reconnect); 4000/4008/4009 →
32 * backoff reconnect `min(1 s · 2^n, maxReconnectDelayMs ~2 s)` (the
33 * < 3 s reconnect budget: first attempt ~1 s, second ~2 s); 4010 → retry
34 * after a fixed 5 s (server shutdown drain).
35 *
36 * Offline (TASK-0092 R5): the offline persistence (y-indexeddb via
37 * `offlineFactory`) is created BEFORE the socket opens, so a returning
38 * tab has local state for a state-vector sync. `setOnline(false)` →
39 * state `offline` (local edits keep applying to the doc — no edit is
40 * ever blocked offline); `setOnline(true)` → reconnect.
41 *
42 * Determinism: clock + timers are injectable (AGENTS.md rule 25).
43 */
44 
45import * as Y from "yjs";
46 
47import { AwarenessStore } from "./awareness.js";
48import { bytesToBase64, blocksToYDoc, decodeBase64Yjs, type BlockLike } from "./codec.js";
49import {
50  OFFLINE_AUTOSAVE_INTERVAL_MS,
51  createIndexeddbOfflineFactory,
52  offlineDocName,
53} from "./offline.js";
54import {
55  BINARY_KIND,
56  CLOSE_HINTS,
57  WS_CLOSE,
58  type ClientFrame,
59  type ConnectionState,
60  type HelloFrame,
61  type OfflinePersistence,
62  type RealtimeProviderConfig,
63  type ServerFrame,
64  type WSAdapter,
65} from "./types.js";
66 
67/** Reconnect backoff cap (default ~2 s → the < 3 s budget: first attempt
68 *  ~1 s, second ~2 s — both inside the exit criterion). */
69const DEFAULT_MAX_RECONNECT_DELAY_MS = 2000;
70/** First backoff step (doubles per attempt, capped). */
71const FIRST_BACKOFF_MS = 1000;
72/** Graceful-shutdown retry delay (4010 — the server is draining). */
73const SHUTDOWN_RETRY_MS = 5000;
74/** App-level heartbeat cadence (the server's own ping period is 30 s). */
75const DEFAULT_PING_INTERVAL_MS = 30_000;
76 
77/**
78 * The realtime provider. Instantiate via {@link createRealtimeProvider};
79 * call {@link RealtimeProvider.connect} on mount and
80 * {@link RealtimeProvider.dispose} on unmount.
81 */
82export class RealtimeProvider {
83  private readonly config: RealtimeProviderConfig;
84  private readonly now: () => number;
85  private readonly setTimeout: typeof globalThis.setTimeout;
86  private readonly clearTimeout: typeof globalThis.clearTimeout;
87 
88  private state: ConnectionState = "closed";
89  private socket: WSAdapter | null = null;
90  private canEdit = false;
91  private connId: string | null = null;
92  private lastSeq: number | null = null;
93  private reconnectAttempts = 0;
94  private refreshedOnce = false;
95  private closedByUs = false;
96  private pingTimer: ReturnType<typeof globalThis.setTimeout> | null = null;
97  private reconnectTimer: ReturnType<typeof globalThis.setTimeout> | null = null;
98  private offline: OfflinePersistence | null = null;
99  private localUpdateUnsub: (() => void) | null = null;
100  private readonly awareness: AwarenessStore | null;
101 
102  constructor(config: RealtimeProviderConfig) {
103    this.config = config;
104    this.now = config.now ?? (() => Date.now());
105    // `typeof globalThis.setTimeout` is an overload set (node `Timer` vs DOM
106    // `number`); a plain lambda is not assignable to it, and no parameter
107    // types flow through the `??` — so type the params explicitly and cast
108    // the default (the runtime call is the same either way).
109    this.setTimeout = (config.setTimeout ?? ((fn: unknown, ms?: number) =>
110      globalThis.setTimeout(fn as Parameters<typeof globalThis.setTimeout>[0], ms))) as typeof globalThis.setTimeout;
111    this.clearTimeout = (config.clearTimeout ?? ((id: unknown) =>
112      globalThis.clearTimeout(id as Parameters<typeof globalThis.clearTimeout>[0]))) as typeof globalThis.clearTimeout;
113    this.awareness =
114      config.presence !== undefined
115        ? new AwarenessStore({
116            self: config.presence,
117            ...(config.hasBlock !== undefined ? { hasBlock: config.hasBlock } : {}),
118            onPeersChange: (peers) => this.config.onPeerChange?.(peers),
119            now: this.now,
120          })
121        : null;
122  }
123 
124  /** The current connection state (the top-bar StatusChip input). */
125  get connectionState(): ConnectionState {
126    return this.state;
127  }
128 
129  /** The join capability flag (VIEW-only rooms drop local updates). */
130  get isCanEdit(): boolean {
131    return this.canEdit;
132  }
133 
134  /** The connection identity assigned by the server (null before hello). */
135  get connectionId(): string | null {
136    return this.connId;
137  }
138 
139  /** The awareness store (null when presence is disabled). */
140  get awarenessStore(): AwarenessStore | null {
141    return this.awareness;
142  }
143 
144  /**
145   * The pending outbox count (the banner's sum input — the CRDT-pending
146   * proxy; the app adds its own outbox count, TASK-0095 R2).
147   */
148  get localPendingCount(): number {
149    return this.config.getOutboxPendingCount?.() ?? 0;
150  }
151 
152  /**
153   * Open the offline persistence (BEFORE the socket — R5), seed the doc
154   * if needed, and connect. No-op when already connecting/syncing/
155   * connected; `offline` when the network is down (reconnects on
156   * {@link setOnline}).
157   */
158  async connect(): Promise<void> {
159    if (this.closedByUs) return;
160    if (
161      this.state === "connecting" ||
162      this.state === "syncing" ||
163      this.state === "connected"
164    ) {
165      return;
166    }
167    if (!this.isOnline()) {
168      this.setState("offline");
169      return;
170    }
171    this.setState("connecting");
172 
173    // R5: load the offline doc BEFORE the sync.request — a returning tab
174    // gets a state-vector sync (diff) instead of a full doc.
175    const factory = this.config.offlineFactory ?? createIndexeddbOfflineFactory();
176    try {
177      this.offline = await factory.create(
178        offlineDocName(this.config.workspaceId, this.config.pageId),
179        this.config.binding.doc,
180        { autoSaveInterval: OFFLINE_AUTOSAVE_INTERVAL_MS },
181      );
182    } catch {
183      this.offline = null; // no IndexedDB (e.g. private mode) — online-only
184    }
185 
186    // Seed the doc from `initialBlocks` only when nothing else provided
187    // state (no offline doc content, no ProseMirror view state yet).
188    if (!this.config.binding.hasLocalState() && this.config.initialBlocks !== undefined) {
189      this.seedInitialBlocks();
190    }
191 
192    this.localUpdateUnsub ??= this.config.binding.onLocalUpdate((update) => {
193      this.handleLocalUpdate(update);
194    });
195 
196    this.openSocket();
197  }
198 
199  /** The app's connectivity listener (online/offline events). */
200  setOnline(online: boolean): void {
201    if (online) {
202      if (this.state === "offline") void this.connect();
203    } else {
204      this.setState("offline");
205      this.stopPing();
206      // Close the socket explicitly — an offline connection is a zombie
207      // (the reconnect loop is off until setOnline(true)).
208      const socket = this.socket;
209      if (socket !== null && (socket.readyState === 0 || socket.readyState === 1)) {
210        socket.close(1000, "client-offline");
211      }
212    }
213  }
214 
215  /** Flush pending offline state + release (pagehide / unmount). */
216  async dispose(): Promise<void> {
217    this.closedByUs = true;
218    this.stopPing();
219    this.clearReconnect();
220    const socket = this.socket;
221    if (socket !== null) {
222      socket.onopen = null;
223      socket.onmessage = null;
224      socket.onclose = null;
225      socket.onerror = null;
226      if (socket.readyState === 0 || socket.readyState === 1) {
227        socket.close(1000, "dispose");
228      }
229      this.socket = null;
230    }
231    this.localUpdateUnsub?.();
232    this.localUpdateUnsub = null;
233    await this.offline?.dispose();
234    this.offline = null;
235    this.setState("closed");
236  }
237 
238  // --- internals -------------------------------------------------------------
239 
240  private isOnline(): boolean {
241    return this.config.isOnline?.() ?? true;
242  }
243 
244  private setState(state: ConnectionState): void {
245    if (this.state === state) return;
246    this.state = state;
247    this.config.onStateChange?.(state);
248  }
249 
250  private socketUrl(): string {
251    const base = this.config.wsUrl.replace(/\/+$/, "");
252    const token = encodeURIComponent(this.config.getToken());
253    const room = encodeURIComponent(`page:${this.config.pageId}`);
254    return `${base}/realtime?token=${token}&room=${room}`;
255  }
256 
257  private openSocket(): void {
258    const socket = this.config.createSocket(this.socketUrl());
259    this.socket = socket;
260 
261    socket.onopen = () => {
262      // The room join is confirmed by `hello` — nothing is sent until
263      // then (the server authorizes the join at handshake).
264    };
265 
266    socket.onmessage = (event) => {
267      const data = event.data;
268      if (typeof data === "string") {
269        this.handleTextFrame(data);
270      } else {
271        this.handleBinaryFrame(data instanceof Uint8Array ? data : new Uint8Array(data));
272      }
273    };
274 
275    socket.onclose = (event) => {
276      this.handleClose(event.code);
277    };
278 
279    socket.onerror = () => {
280      // The close event follows (the WS contract); nothing to do here.
281    };
282  }
283 
284  private handleTextFrame(raw: string): void {
285    let frame: ServerFrame;
286    try {
287      frame = JSON.parse(raw) as ServerFrame;
288    } catch {
289      this.config.onProtocolError?.({ code: "bad_frame", message: raw.slice(0, 200) });
290      return;
291    }
292    switch (frame.type) {
293      case "hello":
294        this.handleHello(frame);
295        break;
296      case "sync.state": {
297        this.config.binding.applyUpdate(decodeBase64Yjs(frame.chunk));
298        if (frame.done) this.setState("connected");
299        break;
300      }
301      case "ack":
302        if (frame.seq > (this.lastSeq ?? 0)) this.lastSeq = frame.seq;
303        break;
304      case "doc.persisted":
305        this.config.onDocPersisted?.({ pageId: frame.pageId, version: frame.version });
306        break;
307      case "comment.event":
308        this.config.onCommentEvent?.(frame);
309        break;
310      case "notification.push":
311        this.config.onNotificationPush?.(frame);
312        break;
313      case "hydrating":
314        this.config.onHydrating?.(frame.pageId);
315        break;
316      case "error":
317        this.config.onProtocolError?.({ code: frame.code, message: frame.message });
318        break;
319      case "ping-pong":
320        break;
321    }
322  }
323 
324  private handleHello(frame: HelloFrame): void {
325    this.connId = frame.connId;
326    this.canEdit = frame.canEdit;
327    this.reconnectAttempts = 0;
328    this.refreshedOnce = false;
329    this.awareness?.applyRoster(frame.roomPeers);
330    this.setState("syncing");
331 
332    // R4: empty local doc → full sync; local state → state-vector diff.
333    const hasLocal = this.config.binding.hasLocalState();
334    const request: ClientFrame = hasLocal
335      ? {
336          type: "sync.request",
337          stateVector: bytesToBase64(this.config.binding.stateVector()),
338          ...(this.lastSeq !== null ? { lastSeq: this.lastSeq } : {}),
339        }
340      : { type: "sync.request" };
341    this.send(request);
342 
343    // "Then any local-pending updates" — ship the full local state once
344    // (the CRDT merge is idempotent; the server discards duplicates).
345    if (hasLocal) {
346      this.sendBinary(BINARY_KIND.Update, this.config.binding.encodeStateAsUpdate());
347    }
348 
349    // The outbox drain-load signal (TASK-0095 R4.2).
350    if (this.config.getOutboxPendingCount !== undefined) {
351      this.send({ type: "outbox.pending", count: this.config.getOutboxPendingCount() });
352    }
353 
354    this.startPing();
355  }
356 
357  private handleBinaryFrame(bytes: Uint8Array): void {
358    if (bytes.length === 0) return;
359    const kind = bytes[0] ?? 0;
360    const payload = bytes.slice(1);
361    if (kind === BINARY_KIND.Update) {
362      this.config.binding.applyUpdate(payload);
363    } else if (kind === BINARY_KIND.Awareness) {
364      this.config.onAwarenessUpdate?.(payload);
365    }
366  }
367 
368  private handleLocalUpdate(update: Uint8Array): void {
369    if (!this.canEdit) {
370      // Defense in depth (mirrors the server): a VIEW-only local update
371      // is dropped, never sent.
372      this.config.onLocalUpdateBlocked?.();
373      return;
374    }
375    const socket = this.socket;
376    if (
377      socket !== null &&
378      socket.readyState === 1 &&
379      (this.state === "syncing" || this.state === "connected")
380    ) {
381      this.sendBinary(BINARY_KIND.Update, update);
382    }
383    // Offline / reconnecting: the update stays in the local doc; the
384    // reconnect ships the full local state (idempotent merge).
385  }
386 
387  private handleClose(code: number): void {
388    this.stopPing();
389    this.socket = null;
390    if (this.closedByUs) return;
391 
392    const hint = CLOSE_HINTS[code] ?? "Connection lost — reconnecting…";
393 
394    // 4001 unauthorized: refresh the access token and reconnect ONCE
395    // (a second 4001 after the refresh = terminal — not a token problem).
396    if (code === WS_CLOSE.Unauthorized) {
397      if (!this.refreshedOnce) {
398        this.refreshedOnce = true;
399        void this.refreshAndReconnect(hint);
400        return;
401      }
402      this.setState("closed");
403      this.config.onClosed?.(code, hint);
404      this.closedByUs = true; // stop the reconnect loop
405      return;
406    }
407 
408    // 4002 forbidden: terminal — the user lost access (the app navigates).
409    if (code === WS_CLOSE.Forbidden) {
410      this.setState("closed");
411      this.config.onClosed?.(code, hint);
412      this.closedByUs = true;
413      return;
414    }
415 
416    // Offline: no reconnect loop — wait for setOnline(true).
417    if (!this.isOnline()) {
418      this.setState("offline");
419      return;
420    }
421 
422    this.setState("reconnecting");
423    if (code === WS_CLOSE.ServerShutdown) {
424      // Graceful drain: a fixed 5 s wait (the server is restarting).
425      this.scheduleReconnect(SHUTDOWN_RETRY_MS);
426      return;
427    }
428 
429    const delay = Math.min(
430      FIRST_BACKOFF_MS * 2 ** this.reconnectAttempts,
431      this.config.maxReconnectDelayMs ?? DEFAULT_MAX_RECONNECT_DELAY_MS,
432    );
433    this.reconnectAttempts += 1;
434    this.scheduleReconnect(delay);
435  }
436 
437  private async refreshAndReconnect(hint: string): Promise<void> {
438    try {
439      await this.config.refreshAccessToken();
440      this.setState("reconnecting");
441      this.scheduleReconnect(0);
442    } catch {
443      this.setState("closed");
444      this.config.onClosed?.(WS_CLOSE.Unauthorized, hint);
445      this.closedByUs = true;
446    }
447  }
448 
449  private scheduleReconnect(delayMs: number): void {
450    this.clearReconnect();
451    this.reconnectTimer = this.setTimeout(() => {
452      this.reconnectTimer = null;
453      if (this.closedByUs) return;
454      this.setState("connecting");
455      this.openSocket();
456    }, delayMs);
457  }
458 
459  private clearReconnect(): void {
460    if (this.reconnectTimer !== null) {
461      this.clearTimeout(this.reconnectTimer);
462      this.reconnectTimer = null;
463    }
464  }
465 
466  private startPing(): void {
467    this.stopPing();
468    const interval = this.config.pingIntervalMs ?? DEFAULT_PING_INTERVAL_MS;
469    const tick = () => {
470      const socket = this.socket;
471      if (
472        socket !== null &&
473        socket.readyState === 1 &&
474        (this.state === "syncing" || this.state === "connected")
475      ) {
476        const ping: ClientFrame = {
477          type: "ping",
478          ts: this.now(),
479          ...(this.lastSeq !== null ? { lastSeq: this.lastSeq } : {}),
480        };
481        this.send(ping);
482      }
483      this.pingTimer = this.setTimeout(tick, interval);
484    };
485    this.pingTimer = this.setTimeout(tick, interval);
486  }
487 
488  private stopPing(): void {
489    if (this.pingTimer !== null) {
490      this.clearTimeout(this.pingTimer);
491      this.pingTimer = null;
492    }
493  }
494 
495  private send(frame: ClientFrame): void {
496    const socket = this.socket;
497    if (socket !== null && socket.readyState === 1) {
498      socket.send(JSON.stringify(frame));
499    }
500  }
501 
502  private sendBinary(kind: number, bytes: Uint8Array): void {
503    const socket = this.socket;
504    if (socket === null || socket.readyState !== 1) return;
505    const out = new Uint8Array(1 + bytes.length);
506    out[0] = kind;
507    out.set(bytes, 1);
508    socket.send(out);
509  }
510 
511  private seedInitialBlocks(): void {
512    // The canvas normally seeds the doc through the binding (the
513    // ProseMirror view). `initialBlocks` is the task's `pageBlocks`
514    // fallback: a plain block set the codec converts into the doc.
515    const blocks = this.config.initialBlocks;
516    if (blocks === undefined) return;
517    const seeded = blocksToYDoc(blocks as unknown as BlockLike[], {
518      pageId: this.config.pageId,
519    });
520    Y.applyUpdate(this.config.binding.doc, Y.encodeStateAsUpdate(seeded));
521    seeded.destroy();
522  }
523}
524 
525/** The task's `createRealtimeProvider({pageId, workspaceId, …})` factory. */
526export function createRealtimeProvider(config: RealtimeProviderConfig): RealtimeProvider {
527  return new RealtimeProvider(config);
528}
529