frames.ts · Source

frames.tspackages/shared/src/realtime/frames.tsraw ← realtime17,728 B← Back to the app
1/**
2 * @factory/shared/realtime — the WS frame contract (TASK-0091 R3).
3 *
4 * The single discriminated union for `apps/realtime` (server) and
5 * `packages/collaboration` (client mirror) — both sides import the Zod
6 * schemas from here; the client's structural types in
7 * `packages/collaboration/src/types.ts` are a lockstep mirror and must
8 * never drift (the codec/provider tests pin the shape).
9 *
10 * Framing (recorded in the task file, notes 12/13):
11 *  - **Text frames**: one JSON object per frame, `type` discriminant.
12 *  - **Binary frames**: 1-byte kind prefix + raw Yjs bytes.
13 *      `0x01` — doc update (`Y.applyUpdate`-compatible)
14 *      `0x02` — awareness update (y-protocols `applyAwarenessUpdate`)
15 *
16 * Base64 Yjs payloads live INSIDE the JSON control frames
17 * (`sync.request.stateVector`, `sync.state.chunk`); the `update` /
18 * `awareness` frames are binary (prefix + raw bytes).
19 *
20 * Close codes: `4000` protocol/heartbeat, `4001` unauthorized, `4002`
21 * forbidden, `4008` rate/policy, `4009` payload, `4010` server shutdown
22 * (the scaffold's 44xx set is superseded — task header note 3).
23 */
24 
25import { z } from "zod";
26 
27// --- close codes (TASK-0091 R6) ----------------------------------------------
28 
29export const WS_CLOSE = {
30  /** Protocol error / heartbeat timeout (no pong in 90 s). */
31  ProtocolOrHeartbeat: 4000,
32  /** Unauthorized: invalid/expired JWT (no room join happened). */
33  Unauthorized: 4001,
34  /** Forbidden: wrong workspace / not the socket's own control room / no VIEW. */
35  Forbidden: 4002,
36  /** Rate limit / policy violation (connect rate, msg rate, 10 strikes). */
37  RateOrPolicy: 4008,
38  /** Frame larger than the 256 KB cap. */
39  PayloadTooLarge: 4009,
40  /** Graceful server shutdown (SIGTERM drain — retry after 5 s). */
41  ServerShutdown: 4010,
42} as const;
43 
44export type WsCloseCode = (typeof WS_CLOSE)[keyof typeof WS_CLOSE];
45 
46// --- binary framing ------------------------------------------------------------
47 
48/** Binary frame kind prefix (the first byte of a binary WS frame). */
49export const BINARY_KIND = {
50  Update: 0x01,
51  Awareness: 0x02,
52} as const;
53 
54export type BinaryKind = (typeof BINARY_KIND)[keyof typeof BINARY_KIND];
55 
56// --- client → server (JSON text frames) ----------------------------------------
57 
58/** Join-sync request. Empty `stateVector` = full doc; set = state-vector diff. */
59export const SyncRequestFrameSchema = z.object({
60  type: z.literal("sync.request"),
61  /** Base64 state vector; omitted = the local doc is empty (full sync). */
62  stateVector: z.string().optional(),
63  /** Last acked server seq (the Redis update-log resume fast path, R5). */
64  lastSeq: z.number().int().nonnegative().optional(),
65});
66 
67/**
68 * Application-level heartbeat. Carries the client's last-ack seq so a
69 * reconnect can use the update-log resume even between pings (R5).
70 */
71export const PingFrameSchema = z.object({
72  type: z.literal("ping"),
73  ts: z.number(),
74  lastSeq: z.number().int().nonnegative().optional(),
75});
76 
77/**
78 * Non-editor mutation count pending in the local outbox (sent after `hello`
79 * on (re)connect so the server can estimate the drain load).
80 */
81export const OutboxPendingFrameSchema = z.object({
82  type: z.literal("outbox.pending"),
83  count: z.number().int().nonnegative(),
84});
85 
86/**
87 * Control-channel join (TASK-0092 (b) / 6b): open a page's control room on
88 * THIS socket (the web `RealtimeClient` is one-socket-many-channels — it
89 * connects token-only, no `?room=`, and opens channels in-band). `page:<id>`
90 * channels receive the server's control frames (the `block:changed`
91 * refetch-hint; comments/notifications in TASK-0094); `workspace:<id>` is
92 * accepted (the web client may (re)subscribe it) but no frames ride it yet.
93 * Idempotent — re-subscribing an open channel is a no-op (reconnect).
94 */
95export const SubscribeFrameSchema = z.object({
96  type: z.literal("subscribe"),
97  channel: z
98    .string()
99    .refine((c) => /^page:[^:\s]{1,}$/.test(c) || /^workspace:[^:\s]{1,}$/.test(c), {
100      message: "channel must be page:<id> or workspace:<id>",
101    }),
102});
103 
104/** Control-channel leave (the inverse of {@link SubscribeFrameSchema}). */
105export const UnsubscribeFrameSchema = z.object({
106  type: z.literal("unsubscribe"),
107  channel: z
108    .string()
109    .refine((c) => /^page:[^:\s]{1,}$/.test(c) || /^workspace:[^:\s]{1,}$/.test(c), {
110      message: "channel must be page:<id> or workspace:<id>",
111    }),
112});
113 
114/**
115 * A peer's cursor anchor (doc §5.1): a text offset inside one block.
116 * The `selection` sub-object carries a (possibly multi-block) range so the
117 * remote tab can render a translucent highlight (TASK-0093 R4).
118 */
119export const PresenceCursorSchema = z.object({
120  blockId: z.string().min(1),
121  offset: z.number().int().nonnegative(),
122  selection: z
123    .object({
124      fromBlockId: z.string().min(1),
125      fromOffset: z.number().int().nonnegative(),
126      toBlockId: z.string().min(1),
127      toOffset: z.number().int().nonnegative(),
128    })
129    .optional(),
130});
131 
132/**
133 * Presence UPDATE (client → server, TASK-0093): the sender's cursor /
134 * viewing state. The peer identity is NEVER carried by the client — the
135 * server stamps `user` (id/name/color) from the socket's authenticated
136 * handshake + the shared `presenceColor` hash before broadcasting (the
137 * server-stamped-names decision: a client cannot impersonate a name via
138 * presence; the frame is pure state). `cursor: null` clears the cursor
139 * (tab hidden / blur → `viewing: false`); `cursor` absent = no change.
140 * Rides both socket kinds: the doc-room socket (the TASK-0136 provider)
141 * and the roomless control socket over its `page:<id>` channel (the (b)
142 * web client).
143 */
144export const PresenceFrameSchema = z.object({
145  type: z.literal("presence"),
146  cursor: PresenceCursorSchema.nullable().optional(),
147  viewing: z.boolean().optional(),
148});
149 
150/** Presence LEAVE (client → server): the explicit "I'm done" (page
151 *  unmount / `dispose`) — the server drops the peer immediately instead
152 *  of waiting for the 90 s prune. Best-effort: a dropped leave is healed
153 *  by the prune + the next join's re-broadcast (doc §5.2). */
154export const PresenceLeaveFrameSchema = z.object({
155  type: z.literal("presence-leave"),
156});
157 
158/** JSON client frames. Binary client frames: `update` / `awareness` (prefixed). */
159export const ClientFrameSchema = z.discriminatedUnion("type", [
160  SyncRequestFrameSchema,
161  PingFrameSchema,
162  OutboxPendingFrameSchema,
163  SubscribeFrameSchema,
164  UnsubscribeFrameSchema,
165  PresenceFrameSchema,
166  PresenceLeaveFrameSchema,
167]);
168 
169// --- server → client (JSON text frames) -----------------------------------------
170 
171/** A roster peer (the `hello` payload — users only, no cursors; cursors
172 *  arrive via the join re-broadcast + `presence` frames, TASK-0093 R2). */
173export const RoomPeerSchema = z.object({
174  user: z.object({ id: z.string(), name: z.string(), color: z.string() }),
175});
176 
177/**
178 * Emitted on a successful room join (R1): identity + capability + clock.
179 * `canEdit` is the defense-in-depth flag the client also enforces locally
180 * (a VIEW-only peer's local updates are dropped, mirroring the server).
181 */
182export const HelloFrameSchema = z.object({
183  type: z.literal("hello"),
184  connId: z.string(),
185  canEdit: z.boolean(),
186  serverTime: z.number(),
187  roomPeers: z.array(RoomPeerSchema),
188});
189 
190/**
191 * Paged doc state: one base64 Yjs update per chunk (≤ 64 KB server-side),
192 * `done` marks the end of the transfer (R3; paged hydration, TASK-0092).
193 */
194export const SyncStateFrameSchema = z.object({
195  type: z.literal("sync.state"),
196  chunk: z.string(),
197  done: z.boolean(),
198});
199 
200/**
201 * Server seq ack for a logged update. When the server sends no acks the
202 * client falls back to pure state-vector resume (always correct — the log
203 * is a speed optimization only, R5).
204 */
205export const AckFrameSchema = z.object({
206  type: z.literal("ack"),
207  seq: z.number().int().nonnegative(),
208});
209 
210/** A persisted doc state reached Postgres (the "Saved" indicator). */
211export const DocPersistedFrameSchema = z.object({
212  type: z.literal("doc.persisted"),
213  pageId: z.string(),
214  version: z.number().int(),
215});
216 
217/**
218 * Page-block change REFETCH HINT over the `page:{pageId}` control room
219 * (TASK-0092 R6 / (b) — the user-chosen additive bridge). The producer
220 * (the realtime server) pushes this on a `page.updated` domain event
221 * (REST write or a flush) instead of replaying per-block patches: a web
222 * `Block` carries `createdAt`/`updatedAt` (rendered in the context menu +
223 * page footer) which the flush's minimal `BlockOperation`s do NOT, so a
224 * lossless `block:upsert` would need a DB row read per block. The hint
225 * says "this page's blocks changed → refetch" (the client re-reads the DB,
226 * the source of truth at rest) — lossless, correct timestamps, no per-
227 * block fan-out. `version` (the new `pages.version`) is advisory: the
228 * client may use it for dedup/last-write-wins but MUST NOT gate on it.
229 */
230export const BlockChangedFrameSchema = z.object({
231  type: z.literal("block:changed"),
232  pageId: z.string().min(1),
233  /** The new `pages.version` (advisory — refetch anyway). */
234  version: z.number().int().min(1).optional(),
235});
236 
237/**
238 * Presence BROADCAST (server → client, TASK-0093): the full current state
239 * of ONE peer in a page/database room. The server stamps `user`
240 * (server-stamped names — id from the socket's handshake, name from the
241 * identity resolver, color from the shared `presenceColor` hash); the
242 * client never supplies it. Fans out on:
243 *   - a peer's `presence` frame (its cursor/viewing update),
244 *   - a room's JOIN re-broadcast (every peer's state, cursors included —
245 *     the deterministic catch-up for joiners, doc §5.1),
246 *   - a prune (the pruned peer is removed via `presence-leave`),
247 *   - > 50 peers (the cap): `cursor`/`viewing` are stripped, `user` stays
248 *     (avatars-only mode, doc §5.1).
249 * Over doc-room sockets the frame carries the room's id in `pageId` (a
250 * database room carries `databaseId` in the same field); over roomless
251 * control sockets it rides the subscribed `page:<id>` channel (the (b)
252 * web client — one socket, many channels).
253 */
254export const PresenceBroadcastSchema = z.object({
255  type: z.literal("presence"),
256  /** The room the state belongs to (`page:<id>` or `database:<id>`). */
257  pageId: z.string().min(1),
258  /** The peer's server-stamped identity. */
259  user: z.object({ id: z.string().min(1), name: z.string(), color: z.string() }),
260  /** The peer's cursor (null = cleared / hidden / avatars-only cap). */
261  cursor: PresenceCursorSchema.nullable().optional(),
262  /** Whether the peer is actively viewing (false = background tab — the
263   *  cursor is hidden, the avatar stays). */
264  viewing: z.boolean().optional(),
265  /** Server epoch ms (the client's prune basis + join-latency metric). */
266  lastSeen: z.number().int().nonnegative(),
267});
268 
269/**
270 * Presence LEAVE broadcast (server → client, TASK-0093): the peer left
271 * the room (explicit `presence-leave`, the socket closed, or the 90 s
272 * prune). The client removes the peer from its roster (the bar updates,
273 * the "X left" toast fires). Best-effort — a missed leave is healed by
274 * the client's own prune window + the next join's re-broadcast (doc §5.2).
275 */
276export const PresenceLeaveBroadcastSchema = z.object({
277  type: z.literal("presence-leave"),
278  /** The room the peer left (`page:<id>` or `database:<id>`). */
279  pageId: z.string().min(1),
280  userId: z.string().min(1),
281});
282 
283/** Comment fan-out over the page room (TASK-0094 R1). */
284export const CommentEventFrameSchema = z.object({
285  type: z.literal("comment.event"),
286  kind: z.enum(["created", "replied", "resolved", "reopened", "reacted", "deleted"]),
287  commentId: z.string(),
288  pageId: z.string(),
289  actorId: z.string(),
290  /** First ~200 chars + render hint (the full comment is fetched on demand). */
291  preview: z.string(),
292});
293 
294/** Notification push over the `user:{userId}` control room (TASK-0094 R7). */
295export const NotificationPushFrameSchema = z.object({
296  type: z.literal("notification.push"),
297  notificationId: z.string(),
298  /** The notification kind (comment.reply, mention, …) — `type` is the
299   *  frame discriminator, so the payload field is `kind`. */
300  kind: z.string(),
301  title: z.string(),
302  deepLink: z.object({ type: z.string(), id: z.string() }),
303  actorName: z.string().optional(),
304});
305 
306/** Long-hydration notice (the client shows "Loading page…", TASK-0092 R2). */
307export const HydratingFrameSchema = z.object({
308  type: z.literal("hydrating"),
309  pageId: z.string(),
310});
311 
312/** Protocol error (a close is NOT implied — the socket stays open). */
313export const ErrorFrameSchema = z.object({
314  type: z.literal("error"),
315  code: z.string(),
316  message: z.string(),
317});
318 
319/** Heartbeat echo. */
320export const PingPongFrameSchema = z.object({
321  type: z.literal("ping-pong"),
322  ts: z.number(),
323});
324 
325export const ServerFrameSchema = z.discriminatedUnion("type", [
326  HelloFrameSchema,
327  SyncStateFrameSchema,
328  AckFrameSchema,
329  DocPersistedFrameSchema,
330  BlockChangedFrameSchema,
331  PresenceBroadcastSchema,
332  PresenceLeaveBroadcastSchema,
333  CommentEventFrameSchema,
334  NotificationPushFrameSchema,
335  HydratingFrameSchema,
336  ErrorFrameSchema,
337  PingPongFrameSchema,
338]);
339 
340// --- inferred types (client mirror + server handlers) ---------------------------
341 
342export type SyncRequestFrame = z.infer<typeof SyncRequestFrameSchema>;
343export type PingFrame = z.infer<typeof PingFrameSchema>;
344export type OutboxPendingFrame = z.infer<typeof OutboxPendingFrameSchema>;
345export type SubscribeFrame = z.infer<typeof SubscribeFrameSchema>;
346export type UnsubscribeFrame = z.infer<typeof UnsubscribeFrameSchema>;
347export type PresenceCursor = z.infer<typeof PresenceCursorSchema>;
348export type PresenceFrame = z.infer<typeof PresenceFrameSchema>;
349export type PresenceLeaveFrame = z.infer<typeof PresenceLeaveFrameSchema>;
350export type ClientFrame = z.infer<typeof ClientFrameSchema>;
351 
352export type RoomPeer = z.infer<typeof RoomPeerSchema>;
353export type HelloFrame = z.infer<typeof HelloFrameSchema>;
354export type SyncStateFrame = z.infer<typeof SyncStateFrameSchema>;
355export type AckFrame = z.infer<typeof AckFrameSchema>;
356export type DocPersistedFrame = z.infer<typeof DocPersistedFrameSchema>;
357export type BlockChangedFrame = z.infer<typeof BlockChangedFrameSchema>;
358export type PresenceBroadcast = z.infer<typeof PresenceBroadcastSchema>;
359export type PresenceLeaveBroadcast = z.infer<typeof PresenceLeaveBroadcastSchema>;
360export type CommentEventFrame = z.infer<typeof CommentEventFrameSchema>;
361export type NotificationPushFrame = z.infer<typeof NotificationPushFrameSchema>;
362export type HydratingFrame = z.infer<typeof HydratingFrameSchema>;
363export type ErrorFrame = z.infer<typeof ErrorFrameSchema>;
364export type PingPongFrame = z.infer<typeof PingPongFrameSchema>;
365export type ServerFrame = z.infer<typeof ServerFrameSchema>;
366 
367/** One decoded wire frame (text = JSON object, binary = prefixed Yjs bytes). */
368export type WireFrame =
369  | { readonly kind: "text"; readonly frame: ClientFrame | ServerFrame }
370  | {
371      readonly kind: "binary";
372      readonly binary: "update" | "awareness";
373      readonly bytes: Uint8Array;
374    };
375 
376// --- close-code hints (the client toast text) ------------------------------------
377 
378/** Close code → user-facing reconnect hint (the client toast text). */
379export const CLOSE_HINTS: Readonly<Record<number, string>> = {
380  [WS_CLOSE.ProtocolOrHeartbeat]: "Connection lost — reconnecting…",
381  [WS_CLOSE.Unauthorized]: "Session expired — signing you back in…",
382  [WS_CLOSE.Forbidden]: "You no longer have access to this page.",
383  [WS_CLOSE.RateOrPolicy]: "Too many changes — giving the server a moment…",
384  [WS_CLOSE.PayloadTooLarge]: "A change was too large to send — retrying…",
385  [WS_CLOSE.ServerShutdown]: "The server is restarting — reconnecting in a moment…",
386};
387 
388// --- server-side limit constants (TASK-0091, doc §2.1/§6.2) -----------------------
389 
390export const REALTIME_LIMITS = {
391  /** New connections per IP per minute (connect-time limiter). */
392  connectPerIpPerMin: 30,
393  /** Live sockets per user (global, the server counts live sockets). */
394  maxConnsPerUser: 100,
395  /** Max frame size in bytes (a bigger frame → close 4009). */
396  maxFrameBytes: 256 * 1024,
397  /** Sustained messages per socket per second (token bucket). */
398  maxMsgsPerSocketPerSec: 100,
399  /** Server→client sync.state chunk cap (R3 paged transfer). */
400  maxSyncChunkBytes: 64 * 1024,
401  /** Update log: last N entries (Redis list cap, R5). */
402  updateLogMaxEntries: 2000,
403  /** Update log: total byte cap (R5, whichever bound hits first). */
404  updateLogMaxBytes: 10 * 1024 * 1024,
405  /** Server ping cadence (ms, R4). */
406  pingIntervalMs: 30_000,
407  /** Silence before a socket is pruned with 4000 (ms, R4). */
408  pruneSilenceMs: 90_000,
409  /** Awareness silence before a peer's state is pruned (ms, doc §5.1). */
410  awarenessPruneMs: 90_000,
411  /** Presence fan-out cap: > this many peers → avatars only, no cursors. */
412  awarenessPeerCap: 50,
413  /** Live control sockets per user (multi-tab abuse guard, note 16). */
414  maxControlSocketsPerUser: 10,
415  /** Bad frames in 10 s before the socket is dropped (R3 10-strike). */
416  strikeLimit: 10,
417  /** The strike window (ms, R3). */
418  strikeWindowMs: 10_000,
419  /** Empty-room eviction (the "warm room" lifetime, R2). */
420  roomGcMs: 5 * 60_000,
421  /** Max buffered updates while a room is hydrating (doc §4.2). */
422  hydrateBufferLimit: 5_000,
423  /** Hydration deadline (ms; buffer flush + error on overflow, doc §4.2). */
424  hydrateMaxMs: 5_000,
425  /** SIGTERM drain budget (ms, R6). */
426  drainMaxMs: 10_000,
427  /** retryAfterMs advertised on the 4010 close frame (R6). */
428  shutdownRetryAfterMs: 5_000,
429} as const;
430