update-log.ts · Source

update-log.tsapps/realtime/src/update-log.tsraw ← src9,490 B← Back to the app
1/**
2 * The per-room update log for reconnect resume (TASK-0091 R5).
3 *
4 * The doc §4.5 "in-memory ring buffer" is implemented in Redis (task
5 * resolution, header note: Redis adds restart safety, identical bounds) —
6 * `realtime:updates:{roomKey}` as a Redis list, capped at the LAST 2,000
7 * entries OR ~10 MB (whichever hits first). On a successful Postgres
8 * flush the log is trimmed to post-flush entries.
9 *
10 * A seq is a monotonic integer per room, starting at 1. A client that
11 * reports `lastSeq` gets entries `(lastSeq, +∞)` when the log still
12 * covers `lastSeq + 1` (fast path); a gap falls back to the state-vector
13 * delta (always correct — the log is a speed optimization only, R5).
14 *
15 * {@link InMemoryUpdateLog} is the same contract without Redis — unit/
16 * integration tests and the no-Redis dev mode use it; {@link
17 * RedisUpdateLog} delegates to a minimal backend surface that ioredis
18 * satisfies structurally (injectable — the tests pass a fake).
19 */
20 
21import { REALTIME_LIMITS } from "@factory/shared/realtime/frames";
22 
23export interface UpdateLogEntry {
24  readonly seq: number;
25  /** The Yjs update bytes (base64 — Redis-safe). */
26  readonly updateB64: string;
27  /** The page version this update preceded (0 = before any flush). */
28  readonly pageVersion: number;
29}
30 
31export interface UpdateLog {
32  /** Append an update; returns the assigned seq. */
33  append(roomKey: string, updateB64: string, pageVersion: number): Promise<number>;
34  /**
35   * Entries with seq in `(afterSeq, +∞)`. `null` when the log does NOT
36   * cover `afterSeq + 1` (gap — the caller falls back to state vectors).
37   */
38  entriesAfter(roomKey: string, afterSeq: number): Promise<UpdateLogEntry[] | null>;
39  /** The highest seq ever appended (0 = empty). */
40  maxSeq(roomKey: string): Promise<number>;
41  /** Trim to entries strictly after `keepSeq` (post-flush). */
42  trimAfter(roomKey: string, keepSeq: number): Promise<void>;
43  /** Drop a room's log entirely (room GC). */
44  clear(roomKey: string): Promise<void>;
45}
46 
47// --- in-memory implementation (tests + no-Redis dev mode) -----------------------
48 
49interface MemEntry {
50  seq: number;
51  updateB64: string;
52  pageVersion: number;
53}
54 
55export class InMemoryUpdateLog implements UpdateLog {
56  private readonly maxEntries: number;
57  private readonly maxBytes: number;
58  private readonly logs = new Map<string, { entries: MemEntry[]; maxSeq: number }>();
59 
60  constructor(opts?: { maxEntries?: number; maxBytes?: number }) {
61    this.maxEntries = opts?.maxEntries ?? REALTIME_LIMITS.updateLogMaxEntries;
62    this.maxBytes = opts?.maxBytes ?? REALTIME_LIMITS.updateLogMaxBytes;
63  }
64 
65  private log(roomKey: string): { entries: MemEntry[]; maxSeq: number } {
66    let l = this.logs.get(roomKey);
67    if (l === undefined) {
68      l = { entries: [], maxSeq: 0 };
69      this.logs.set(roomKey, l);
70    }
71    return l;
72  }
73 
74  async append(roomKey: string, updateB64: string, pageVersion: number): Promise<number> {
75    const l = this.log(roomKey);
76    const seq = l.maxSeq + 1;
77    l.entries.push({ seq, updateB64, pageVersion });
78    l.maxSeq = seq;
79    this.evict(l);
80    return seq;
81  }
82 
83  private evict(l: { entries: MemEntry[]; maxSeq: number }): void {
84    // Entry cap (drop oldest).
85    while (l.entries.length > this.maxEntries) {
86      l.entries.shift();
87    }
88    // Byte cap (base64 length is the on-the-wire size; the bound is soft
89    // — a single oversized update may overshoot, recorded).
90    let bytes = l.entries.reduce((n, e) => n + e.updateB64.length, 0);
91    while (bytes > this.maxBytes && l.entries.length > 1) {
92      bytes -= l.entries.shift()!.updateB64.length;
93    }
94  }
95 
96  async entriesAfter(roomKey: string, afterSeq: number): Promise<UpdateLogEntry[] | null> {
97    const l = this.logs.get(roomKey);
98    if (l === undefined) return afterSeq === 0 ? [] : null;
99    if (l.entries.length === 0) return [];
100    const oldest = l.entries[0]!.seq;
101    // Gap: the client's next needed seq is no longer in the log.
102    if (afterSeq + 1 < oldest) return null;
103    return l.entries
104      .filter((e) => e.seq > afterSeq)
105      .map((e) => ({ seq: e.seq, updateB64: e.updateB64, pageVersion: e.pageVersion }));
106  }
107 
108  async maxSeq(roomKey: string): Promise<number> {
109    return this.logs.get(roomKey)?.maxSeq ?? 0;
110  }
111 
112  async trimAfter(roomKey: string, keepSeq: number): Promise<void> {
113    const l = this.logs.get(roomKey);
114    if (l === undefined) return;
115    l.entries = l.entries.filter((e) => e.seq > keepSeq);
116  }
117 
118  async clear(roomKey: string): Promise<void> {
119    this.logs.delete(roomKey);
120  }
121 
122  /** Test/diagnostic: raw log size (entries). */
123  size(roomKey: string): number {
124    return this.logs.get(roomKey)?.entries.length ?? 0;
125  }
126}
127 
128// --- Redis implementation (production) ------------------------------------------
129 
130/**
131 * The minimal Redis surface {@link RedisUpdateLog} needs. ioredis'
132 * `Redis` satisfies it structurally (all methods are promise-based);
133 * the integration tests inject a fake.
134 */
135export interface UpdateLogBackend {
136  incr(key: string): Promise<number>;
137  incrBy(key: string, delta: number): Promise<number>;
138  get(key: string): Promise<string | null>;
139  set(key: string, value: string): Promise<unknown>;
140  rpush(key: string, value: string): Promise<number>;
141  lrange(key: string, start: number, stop: number): Promise<string[]>;
142  ltrim(key: string, start: number, stop: number): Promise<unknown>;
143  llen(key: string): Promise<number>;
144  del(...keys: string[]): Promise<number>;
145}
146 
147const KEY = (roomKey: string): string => `realtime:updates:${roomKey}`;
148const SEQ_KEY = (roomKey: string): string => `realtime:updates:${roomKey}:seq`;
149const BYTES_KEY = (roomKey: string): string => `realtime:updates:${roomKey}:bytes`;
150 
151function parseEntry(raw: string): { seq: number; update: string; version: number } | null {
152  try {
153    const e = JSON.parse(raw) as { seq?: unknown; update?: unknown; version?: unknown };
154    if (typeof e.seq !== "number" || typeof e.update !== "string") return null;
155    return { seq: e.seq, update: e.update, version: typeof e.version === "number" ? e.version : 0 };
156  } catch {
157    return null;
158  }
159}
160 
161/**
162 * Redis-backed update log. One JSON string per list element
163 * (`{seq, update, version}`), one `RPUSH` per update; the list is
164 * left-trimmed to the entry cap and (best-effort) the byte cap via a
165 * stored byte counter.
166 */
167export class RedisUpdateLog implements UpdateLog {
168  private readonly redis: UpdateLogBackend;
169  private readonly maxEntries: number;
170  private readonly maxBytes: number;
171 
172  constructor(opts: { redis: UpdateLogBackend; maxEntries?: number; maxBytes?: number }) {
173    this.redis = opts.redis;
174    this.maxEntries = opts.maxEntries ?? REALTIME_LIMITS.updateLogMaxEntries;
175    this.maxBytes = opts.maxBytes ?? REALTIME_LIMITS.updateLogMaxBytes;
176  }
177 
178  async append(roomKey: string, updateB64: string, pageVersion: number): Promise<number> {
179    const seq = await this.redis.incr(SEQ_KEY(roomKey));
180    const entry = JSON.stringify({ seq, update: updateB64, version: pageVersion });
181    await this.redis.rpush(KEY(roomKey), entry);
182    await this.redis.incrBy(BYTES_KEY(roomKey), entry.length);
183    // Entry cap: keep the LAST maxEntries.
184    const len = await this.redis.llen(KEY(roomKey));
185    if (len > this.maxEntries) {
186      await this.redis.ltrim(KEY(roomKey), len - this.maxEntries, -1);
187      await this.remeasureBytes(roomKey);
188    }
189    // Byte cap (soft — a single giant update may overshoot; recorded).
190    const bytes = Number((await this.redis.get(BYTES_KEY(roomKey))) ?? 0);
191    if (bytes > this.maxBytes) {
192      const len2 = await this.redis.llen(KEY(roomKey));
193      if (len2 > 1) {
194        await this.redis.ltrim(KEY(roomKey), 1, -1);
195        await this.remeasureBytes(roomKey);
196      }
197    }
198    return seq;
199  }
200 
201  /** Re-sum the stored element lengths into the byte counter (post-trim). */
202  private async remeasureBytes(roomKey: string): Promise<void> {
203    const all = await this.redis.lrange(KEY(roomKey), 0, -1);
204    const total = all.reduce((n, s) => n + s.length, 0);
205    await this.redis.set(BYTES_KEY(roomKey), String(total));
206  }
207 
208  async entriesAfter(roomKey: string, afterSeq: number): Promise<UpdateLogEntry[] | null> {
209    const maxSeq = await this.maxSeq(roomKey);
210    if (maxSeq === 0) return afterSeq === 0 ? [] : null;
211    const all = await this.redis.lrange(KEY(roomKey), 0, -1);
212    if (all.length === 0) return [];
213    const first = parseEntry(all[0]!);
214    if (first === null || afterSeq + 1 < first.seq) return null; // gap
215    const out: UpdateLogEntry[] = [];
216    for (const raw of all) {
217      const e = parseEntry(raw);
218      if (e !== null && e.seq > afterSeq) {
219        out.push({ seq: e.seq, updateB64: e.update, pageVersion: e.version });
220      }
221    }
222    return out;
223  }
224 
225  async maxSeq(roomKey: string): Promise<number> {
226    const v = await this.redis.get(SEQ_KEY(roomKey));
227    return v === null ? 0 : Number(v);
228  }
229 
230  async trimAfter(roomKey: string, keepSeq: number): Promise<void> {
231    const all = await this.redis.lrange(KEY(roomKey), 0, -1);
232    let drop = 0;
233    for (const raw of all) {
234      const e = parseEntry(raw);
235      if (e === null || e.seq <= keepSeq) drop += 1;
236      else break;
237    }
238    if (drop > 0) {
239      await this.redis.ltrim(KEY(roomKey), drop, -1);
240      await this.remeasureBytes(roomKey);
241    }
242  }
243 
244  async clear(roomKey: string): Promise<void> {
245    await this.redis.del(KEY(roomKey), SEQ_KEY(roomKey), BYTES_KEY(roomKey));
246  }
247}
248