rooms.ts · Source

rooms.tsapps/realtime/src/rooms.tsraw ← src23,104 B← Back to the app
1/**
2 * RoomManager (TASK-0091 R2) — one authoritative room per resource.
3 *
4 * A room owns: its `Y.Doc` (page/database rooms only), the peer set,
5 * the debounced persist state, and the hydrate-on-write buffer
6 * (doc §4.2). The per-room update log (R5) lives in the server, keyed
7 * by the room key.
8 *
9 * Lifecycle (doc §3 / R2):
10 *  - cold room: hydrate from the {@link DocStore} BEFORE publishing the
11 *    room (concurrent joins share the one hydration);
12 *  - update while hydrating: buffered (max 5 s / 5,000 updates), applied
13 *    to the hydrated doc in order (doc §4.2 hydrate-on-write race);
14 *  - leave: 0 peers + dirty → the flush is awaited before eviction; a
15 *    room never evicts with unsaved updates (GC after 5 min warm);
16 *  - SIGTERM: `drainAll()` flushes every dirty room (≤ 10 s budget, R6).
17 *
18 * The persistence seam is {@link DocStore} (note 17 — the Prisma
19 * implementation lands in TASK-0092 via the api's internal flush route;
20 * tests + no-DB dev use {@link InMemoryDocStore}). The Yjs codec is the
21 * shared `@factory/collaboration` codec (the same blocksToYDoc/
22 * yDocToBlockList the client canvas uses — one codec, both sides).
23 */
24 
25import * as Y from "yjs";
26import {
27  type BlockLike,
28  blocksToYDoc,
29  yDocToBlockList,
30  applyBlockOperationsToYDoc,
31  diffBlockSets,
32} from "@factory/collaboration/codec";
33import type { BlockOperation } from "@factory/shared";
34import { presenceColor } from "@factory/shared/presence";
35import type { RoomPeer } from "@factory/shared/realtime/frames";
36 
37/** The block id an op refers to (insert: the inserted block's id; the
38 *  rest: the target's id) — the 3-way rebase's "who touched what" basis.
39 *  An insert without an explicit id is a server-minted block: no external
40 *  (REST) op can ever target it, so it never collides and is filtered out
41 *  here (`undefined`). */
42function opIdOf(op: BlockOperation): string | undefined {
43  return op.op === "insert" ? op.block.id : op.id;
44}
45 
46// --- the persistence seam (note 17) ---------------------------------------------
47 
48/**
49 * The doc persistence seam. Production: Postgres (TASK-0092, the
50 * {@link PrismaDocStore} with the `pages.version` CAS + flush-rebase).
51 * Tests/dev: {@link InMemoryDocStore}.
52 *
53 * The seam returns the page's `pages.version` alongside the blocks,
54 * atomically, because the flush-rebase (R2 of TASK-0092) needs the version
55 * the doc was hydrated at (its CAS base) and the version a persist reached
56 * (the new CAS base) in one consistent read — a two-call `hydrate()` +
57 * `getVersion()` pair could race a concurrent REST write in between.
58 */
59export interface HydrateResult {
60  /** Position-ordered blocks; `null` = the page has no blocks. */
61  readonly blocks: readonly BlockLike[] | null;
62  /** The `pages.version` at hydration time — the flush's CAS base. */
63  readonly version: number;
64}
65 
66export interface PersistResult {
67  /** The new `pages.version` (the `doc.persisted` frame carries it). */
68  readonly version: number;
69  /**
70   * Only set on a version conflict (the doc's CAS base was stale — a
71   * concurrent non-Yjs writer bumped `pages.version` since the doc was
72   * hydrated/last-flushed). `externalBlocks` is the store's current state
73   * (the writer's result — the REST batch that ran concurrently); the room
74   * rebases its CRDT on it (3-way merge, the doc §4.3 "flush wins" rule) and
75   * retries the persist once.
76   */
77  readonly conflict?: {
78    readonly externalBlocks: readonly BlockLike[];
79  };
80}
81 
82export interface DocStore {
83  /** Hydrate blocks + the page version for a page (see {@link HydrateResult}). */
84  hydrate(pageId: string): Promise<HydrateResult>;
85  /**
86   * Persist the page's full block set under the `pages.version` CAS
87   * (soft-deletes are the store's concern). `baseVersion` is the version the
88   * doc was hydrated at / last flushed to: if the current version differs,
89   * the persist is a no-op and {@link PersistResult.conflict} carries the
90   * store's external state (the concurrent writer) for the room to rebase on.
91   * A successful persist returns the new version (no conflict).
92   */
93  persist(
94    pageId: string,
95    blocks: readonly BlockLike[],
96    baseVersion: number,
97  ): Promise<PersistResult>;
98  /** Whether the store backend is reachable (`/readyz`). */
99  ping(): Promise<boolean>;
100}
101 
102/** A trivial in-memory store (tests + no-DB dev mode). */
103export class InMemoryDocStore implements DocStore {
104  private readonly pages = new Map<string, BlockLike[]>();
105  private readonly versions = new Map<string, number>();
106  private pingOk = true;
107 
108  setPingOk(ok: boolean): void {
109    this.pingOk = ok;
110  }
111 
112  seed(pageId: string, blocks: readonly BlockLike[]): void {
113    this.pages.set(pageId, blocks.map((b) => ({ ...b })));
114    // A seed is a "the page exists at version N" — the version bumps only on
115    // persist, so the seed's version is whatever was already recorded (the
116    // rebase tests simulate a concurrent external writer by bumping it).
117    if (!this.versions.has(pageId)) this.versions.set(pageId, 1);
118  }
119 
120  /**
121   * Simulate a concurrent NON-Yjs writer (a REST `POST /v1/blocks` batch)
122   * against the store: replace the page's blocks and bump `pages.version`.
123   * The rebase test drives the conflict this way. (Production: the API's
124   * block service does exactly this on the `blocks` table.)
125   */
126  writeExternal(pageId: string, blocks: readonly BlockLike[]): number {
127    this.pages.set(pageId, blocks.map((b) => ({ ...b })));
128    const v = (this.versions.get(pageId) ?? 1) + 1;
129    this.versions.set(pageId, v);
130    return v;
131  }
132 
133  /** The page's current `pages.version` (tests / the rebase assertion). */
134  versionOf(pageId: string): number {
135    return this.versions.get(pageId) ?? 1;
136  }
137 
138  async hydrate(pageId: string): Promise<HydrateResult> {
139    const b = this.pages.get(pageId);
140    return {
141      blocks: b === undefined ? null : b.map((x) => ({ ...x })),
142      version: this.versions.get(pageId) ?? 1,
143    };
144  }
145 
146  async persist(
147    pageId: string,
148    blocks: readonly BlockLike[],
149    baseVersion: number,
150  ): Promise<PersistResult> {
151    const current = this.versions.get(pageId) ?? 1;
152    if (current !== baseVersion) {
153      // CAS failure: a concurrent writer bumped the version. Read the store's
154      // current state back (the external writer's result) for the rebase.
155      const ext = this.pages.get(pageId);
156      return { version: current, conflict: { externalBlocks: ext === undefined ? [] : ext.map((x) => ({ ...x })) } };
157    }
158    this.pages.set(pageId, blocks.map((b) => ({ ...b })));
159    const v = current + 1;
160    this.versions.set(pageId, v);
161    return { version: v };
162  }
163 
164  async ping(): Promise<boolean> {
165    return this.pingOk;
166  }
167}
168 
169// --- rooms -----------------------------------------------------------------------
170 
171/** A connected peer in a room. */
172export interface Peer {
173  readonly connId: string;
174  readonly userId: string;
175  readonly name: string;
176  readonly color: string;
177  /** The socket's `canEdit` flag (defense in depth — re-checked per frame). */
178  canEdit: boolean;
179  /** The peer's room roster entry (the `hello.roomPeers` payload). */
180  roster: RoomPeer;
181}
182 
183/** The injectable clock + timers (deterministic flush/GC tests). */
184export interface RoomClock {
185  readonly now: () => number;
186  readonly setTimeout: (fn: () => void, ms: number) => unknown;
187  readonly clearTimeout: (h: unknown) => void;
188}
189 
190export function realClock(): RoomClock {
191  return {
192    now: () => Date.now(),
193    setTimeout: (fn, ms) => setTimeout(fn, ms),
194    clearTimeout: (h) => clearTimeout(h as ReturnType<typeof setTimeout>),
195  };
196}
197 
198export interface RoomManagerOpts {
199  readonly store: DocStore;
200  readonly clock?: RoomClock;
201  /** Debounce before a flush (1 s, doc §4.3). */
202  readonly persistDebounceMs?: number;
203  /** Hard cap on dirty time (10 s, doc §4.3). */
204  readonly persistHardCapMs?: number;
205  /** Empty-room GC delay (5 min warm room, R2). */
206  readonly roomGcMs?: number;
207  /** Hydrate buffer cap (doc §4.2). */
208  readonly hydrateBufferLimit?: number;
209  /** Reports the room's current max logged update seq (the server's
210   *  update log) — captured at the flush snapshot so the post-persist
211   *  trim keeps exactly the post-flush entries (R5: "truncated after a
212   *  successful Postgres persistence"). The seq domain is the LOG seq
213   *  (per-room, restarts after eviction), NOT the page version. */
214  readonly flushSeqProvider?: (roomKey: string) => number;
215}
216 
217/** Events the room manager reports (metrics/frames). */
218export interface RoomManagerEvents {
219  /** A doc persisted (the server broadcasts `doc.persisted` + trims the log
220   *  to post-flush entries, R5). `logSeqAtFlush` is the room's max logged
221   *  update seq captured at the flush snapshot — the trim basis.
222   *  `blockDelta` is the minimal op diff of this flush vs the last persisted
223   *  state (the `page.updated` `blockDelta` additive field, TASK-0092 R2/R6). */
224  onPersisted: (
225    roomKey: string,
226    resourceId: string,
227    version: number,
228    logSeqAtFlush: number,
229    blockDelta: readonly BlockOperation[],
230  ) => void;
231  /** A room rebased its doc on a concurrent external (REST) write after a
232   *  version conflict (the flush-rebase path, TASK-0092 R2). */
233  onRebased?: (roomKey: string, resourceId: string, appliedExternalOps: number) => void;
234  /** A room was evicted (GC / drain). */
235  onRoomEvicted: (roomKey: string) => void;
236}
237 
238/**
239 * One room. (Exported for the unit tests; the server works through
240 * {@link RoomManager}.)
241 */
242export class Room {
243  readonly key: string;
244  readonly kind: "page" | "database";
245  readonly resourceId: string;
246 
247  /** The authoritative doc (page/database rooms). Mutable: hydration
248   *  swaps it (a destroyed doc is never re-used — the yjs rule). */
249  doc: Y.Doc;
250 
251  readonly peers = new Map<string, Peer>();
252 
253  /** Inbound updates buffered while a re-hydration is in flight
254   *  (doc §4.2 hydrate-on-write race; the cold path hydrates before the
255   *  room is published, so peers never see a half-built doc). */
256  private hydrateBuffer: Uint8Array[] = [];
257  private hydrating = false;
258 
259  private dirtySince: number | null = null;
260  private flushTimer: unknown = null;
261  private flushing: Promise<void> | null = null;
262  /** The `pages.version` the doc was hydrated at / last flushed to — the
263   *  persist CAS base (the flush-rebase guard, TASK-0092 R2). Also the
264   *  per-update `pageVersion` the server records in the update log. */
265  lastVersion = 0;
266  /** The block set the doc last persisted as — the 3-way merge BASE for the
267   *  flush-rebase + the `blockDelta` diff baseline. */
268  private lastPersistedBlocks: readonly BlockLike[] = [];
269 
270  private gcTimer: unknown = null;
271  private _evicted = false;
272 
273  constructor(
274    key: string,
275    kind: "page" | "database",
276    resourceId: string,
277    private readonly opts: Required<RoomManagerOpts>,
278    private readonly events: RoomManagerEvents,
279    seed: HydrateResult,
280  ) {
281    this.key = key;
282    this.kind = kind;
283    this.resourceId = resourceId;
284    this.doc = this.buildDoc(seed.blocks);
285    this.lastVersion = seed.version;
286    // A hydrated page persists its baseline (so the first dirty flush diffs
287    // against the real starting state, not an empty one). An empty page has
288    // an empty baseline.
289    this.lastPersistedBlocks = seed.blocks === null ? [] : seed.blocks.map((b) => ({ ...b }));
290  }
291 
292  get isDirty(): boolean {
293    return this.dirtySince !== null;
294  }
295 
296  get isEmpty(): boolean {
297    return this.peers.size === 0;
298  }
299 
300  get isEvicted(): boolean {
301    return this._evicted;
302  }
303 
304  /** Number of buffered (not yet applied) hydration updates (tests). */
305  get bufferedUpdates(): number {
306    return this.hydrateBuffer.length;
307  }
308 
309  private buildDoc(seed: readonly BlockLike[] | null): Y.Doc {
310    const doc =
311      seed === null || seed.length === 0
312        ? new Y.Doc()
313        : blocksToYDoc(seed, { pageId: this.kind === "page" ? this.resourceId : undefined });
314    if (this.kind === "page") {
315      doc.getMap("yMeta").set("pageId", this.resourceId);
316    }
317    return doc;
318  }
319 
320  /**
321   * Apply a remote (client) update. While a re-hydration is in flight,
322   * the update is BUFFERED (doc §4.2) — applied in order once the doc is
323   * ready. Returns `true` when applied now, `false` when buffered.
324   */
325  applyUpdate(bytes: Uint8Array): boolean {
326    if (this.hydrating) {
327      if (this.hydrateBuffer.length >= this.opts.hydrateBufferLimit) {
328        throw new Error("hydrate buffer overflow");
329      }
330      this.hydrateBuffer.push(bytes);
331      return false;
332    }
333    Y.applyUpdate(this.doc, bytes, "remote");
334    this.markDirty();
335    return true;
336  }
337 
338  /** Encode the state delta for a client's state vector (R5 resume). */
339  stateDelta(sinceVectorBytes: Uint8Array | null): Uint8Array {
340    if (sinceVectorBytes === null || sinceVectorBytes.byteLength === 0) {
341      return Y.encodeStateAsUpdate(this.doc);
342    }
343    // `encodeStateAsUpdate` takes the RAW state-vector bytes (not the decoded map).
344    return Y.encodeStateAsUpdate(this.doc, sinceVectorBytes);
345  }
346 
347  private markDirty(): void {
348    if (this.dirtySince === null) {
349      this.dirtySince = this.opts.clock.now();
350    }
351    this.scheduleFlush();
352  }
353 
354  private scheduleFlush(): void {
355    if (this.flushTimer !== null) {
356      this.opts.clock.clearTimeout(this.flushTimer);
357    }
358    const age = this.dirtySince === null ? 0 : this.opts.clock.now() - this.dirtySince;
359    const delay = Math.max(
360      0,
361      Math.min(this.opts.persistDebounceMs, this.opts.persistHardCapMs - age),
362    );
363    this.flushTimer = this.opts.clock.setTimeout(() => {
364      void this.flush();
365    }, delay);
366  }
367 
368  /**
369   * Flush the current doc state (debounce tick, room GC, or SIGTERM drain).
370   * Concurrent flushes coalesce into the one in flight.
371   *
372   * TASK-0092 R2 (the flush-rebase, doc §4.3 "flush wins"): the persist is a
373   * `pages.version` CAS. If a concurrent non-Yjs writer (a REST
374   * `POST /v1/blocks` batch) bumped the version since the doc was hydrated
375   * / last flushed, the persist is a no-op and the store returns its
376   * external state. We rebase the CRDT on it — a 3-way merge whose BASE is
377   * the last persisted block set: the external (REST) ops are applied to the
378   * doc, so on any block the ROOM also touched the doc's CRDT edit wins
379   * (flush-wins) and on blocks only REST touched the REST change is absorbed.
380   * The persist then retries ONCE at the new version. A second conflict
381   * (two concurrent external writes inside one flush window) is outside the
382   * single-retry contract: the room stays dirty with its user edits intact
383   * (a full-set persist, so nothing is lost) and the next flush converges.
384   */
385  flush(): Promise<void> {
386    if (this.flushing !== null) {
387      return this.flushing;
388    }
389    const p = (async () => {
390      if (this.flushTimer !== null) {
391        this.opts.clock.clearTimeout(this.flushTimer);
392        this.flushTimer = null;
393      }
394      // Capture the log watermark BEFORE the (async) persist: entries logged
395      // up to here are covered by this flush's doc snapshot.
396      const logSeqAtFlush = this.opts.flushSeqProvider?.(this.key) ?? 0;
397 
398      // OURS (pre-rebase) + the set of blocks the ROOM's user edits touched
399      // (the flush-wins basis). Captured BEFORE the rebase loop so "room
400      // touched X" reflects user edits only — never the blocks a rebase pass
401      // absorbed (absorbed external blocks must stay absorbable by LATER
402      // external writes, not be mistaken for user edits).
403      let blocks = yDocToBlockList(this.doc);
404      const roomTouched = new Set(
405        diffBlockSets(this.lastPersistedBlocks, blocks).map(opIdOf),
406      );
407      // The external state the doc is already reconciled to (starts at the
408      // last persisted state; each rebase pass advances it).
409      let externalBase = this.lastPersistedBlocks;
410 
411      let result = await this.opts.store.persist(this.resourceId, blocks, this.lastVersion);
412      let attempts = 1;
413 
414      // Version conflict → rebase onto the external (REST) state + retry ONCE
415      // (the spec's "retries once"). The rebase is idempotent + convergent:
416      // each pass applies only the external changes since `externalBase`, so
417      // re-applying is a no-op and later external writes are never clobbered.
418      while (result.conflict !== undefined) {
419        const externalOps = diffBlockSets(externalBase, result.conflict.externalBlocks);
420        const absorb = externalOps.filter((op) => !roomTouched.has(opIdOf(op)));
421        if (absorb.length > 0) {
422          Y.transact(this.doc, () => {
423            applyBlockOperationsToYDoc(this.doc, absorb);
424          }, "realtime-flush-rebase");
425        }
426        this.events.onRebased?.(this.key, this.resourceId, absorb.length);
427        externalBase = result.conflict.externalBlocks;
428        blocks = yDocToBlockList(this.doc);
429 
430        if (attempts >= 2) {
431          // The single retry was used up (two external writers inside one
432          // flush window — outside the single-retry contract). The doc is
433          // now in-sync with the LATEST external (the next flush's full-set
434          // persist clobbers nothing); keep the room dirty with its user
435          // edits intact, advance the CAS base (so the next flush doesn't
436          // re-conflict on the same stale version), and re-arm the debounce.
437          this.lastVersion = result.version;
438          this.scheduleFlush();
439          return;
440        }
441        attempts += 1;
442        result = await this.opts.store.persist(this.resourceId, blocks, result.version);
443      }
444 
445      const blockDelta = diffBlockSets(this.lastPersistedBlocks, blocks);
446      this.lastVersion = result.version;
447      this.lastPersistedBlocks = blocks.map((b) => ({ ...b }));
448      this.dirtySince = null;
449      this.events.onPersisted(this.key, this.resourceId, result.version, logSeqAtFlush, blockDelta);
450    })();
451    this.flushing = p;
452    void p.finally(() => {
453      this.flushing = null;
454    });
455    return p;
456  }
457 
458  /** Evict (room GC / drain): flush-if-dirty, then drop the doc. */
459  async evict(): Promise<void> {
460    if (this._evicted) return;
461    this._evicted = true;
462    if (this.gcTimer !== null) {
463      this.opts.clock.clearTimeout(this.gcTimer);
464      this.gcTimer = null;
465    }
466    if (this.isDirty) {
467      await this.flush();
468    }
469    this.doc.destroy();
470    this.events.onRoomEvicted(this.key);
471  }
472 
473  /**
474   * Mark the room mid re-hydration (the manager, after an eviction
475   * re-join — normally unreachable: evicted rooms are removed from the
476   * manager map; the cold path hydrates before publishing).
477   */
478  setHydrating(): void {
479    this.hydrating = true;
480  }
481 
482  /** Finish a re-hydration: swap the doc + apply buffered updates. */
483  completeHydration(hydrated: Y.Doc): void {
484    if (!this.hydrating) return;
485    this.doc.destroy();
486    this.doc = hydrated;
487    for (const u of this.hydrateBuffer) {
488      Y.applyUpdate(this.doc, u, "remote");
489    }
490    this.hydrateBuffer = [];
491    this.hydrating = false;
492  }
493 
494  /** Arm (or disarm) the empty-room GC timer (R2: 5 min warm room). */
495  notePeersChanged(): void {
496    if (this._evicted) return;
497    if (this.peers.size === 0) {
498      if (this.gcTimer === null) {
499        this.gcTimer = this.opts.clock.setTimeout(() => {
500          this.gcTimer = null;
501          void this.evict();
502        }, this.opts.roomGcMs);
503      }
504    } else if (this.gcTimer !== null) {
505      this.opts.clock.clearTimeout(this.gcTimer);
506      this.gcTimer = null;
507    }
508  }
509 
510  roster(): readonly RoomPeer[] {
511    return [...this.peers.values()].map((p) => p.roster);
512  }
513 
514  /** A peer's display color (userId hash, the shared 20-color palette). */
515  static colorFor(userId: string): string {
516    return presenceColor(userId);
517  }
518}
519 
520// --- the manager -------------------------------------------------------------------
521 
522export class RoomManager {
523  private readonly roomMap = new Map<string, Room>();
524  private readonly events: RoomManagerEvents;
525  private readonly opts: Required<RoomManagerOpts>;
526 
527  constructor(opts: RoomManagerOpts, events: RoomManagerEvents) {
528    this.opts = {
529      store: opts.store,
530      clock: opts.clock ?? realClock(),
531      persistDebounceMs: opts.persistDebounceMs ?? 1000,
532      persistHardCapMs: opts.persistHardCapMs ?? 10_000,
533      roomGcMs: opts.roomGcMs ?? 5 * 60_000,
534      hydrateBufferLimit: opts.hydrateBufferLimit ?? 5_000,
535      // No log wired (unit tests / no-Redis dev) → a 0 watermark: the
536      // trim keeps everything (nothing is trimmed) — the safe default.
537      flushSeqProvider: opts.flushSeqProvider ?? (() => 0),
538    };
539    this.events = events;
540  }
541 
542  /**
543   * Get-or-create a doc room. Cold rooms hydrate BEFORE the room is
544   * published (a joining peer never syncs against an empty doc); warm
545   * rooms are returned as-is.
546   */
547  async getOrCreate(
548    key: string,
549    kind: "page" | "database",
550    resourceId: string,
551  ): Promise<Room> {
552    const existing = this.roomMap.get(key);
553    if (existing !== undefined) return existing;
554    // Cold start: hydrate BEFORE constructing/publishing the room — a
555    // joining peer never syncs against an empty doc (note 14: the
556    // resource id is globally unique, the room key IS the tenant scope;
557    // the workspace is recorded by the server for metrics only).
558    const seed: HydrateResult =
559      kind === "page"
560        ? await this.opts.store.hydrate(resourceId)
561        : { blocks: null, version: 0 }; // database rooms: the DocStore is page-shaped
562    const room = new Room(key, kind, resourceId, this.opts, this.events, seed);
563    this.roomMap.set(key, room);
564    return room;
565  }
566 
567  get(key: string): Room | undefined {
568    return this.roomMap.get(key);
569  }
570 
571  /** Drop a room from the map (after `evict()`). */
572  remove(key: string): void {
573    this.roomMap.delete(key);
574  }
575 
576  /** All live rooms (drain/metrics). */
577  rooms(): Room[] {
578    return [...this.roomMap.values()];
579  }
580 
581  /**
582   * SIGTERM drain (R6): flush every dirty room, evict all, within the
583   * budget (overflow is logged by the caller). Returns the counts.
584   */
585  async drainAll(budgetMs: number): Promise<{ flushed: number; evicted: number }> {
586    const deadline = this.opts.clock.now() + budgetMs;
587    let flushed = 0;
588    let evicted = 0;
589    for (const room of this.rooms()) {
590      if (room.isDirty) {
591        if (this.opts.clock.now() > deadline) break;
592        await room.flush();
593        flushed += 1;
594      }
595      await room.evict();
596      this.roomMap.delete(room.key);
597      evicted += 1;
598    }
599    return { flushed, evicted };
600  }
601}
602 
603/** Build a peer from a handshake identity (color = the shared palette). */
604export function makePeer(
605  connId: string,
606  userId: string,
607  canEdit: boolean,
608  name?: string,
609): Peer {
610  const roster: RoomPeer = {
611    user: { id: userId, name: name ?? "Guest (shared)", color: Room.colorFor(userId) },
612  };
613  return {
614    connId,
615    userId,
616    name: roster.user.name,
617    color: roster.user.color,
618    canEdit,
619    roster,
620  };
621}
622