protocol.ts · Source

protocol.tsapps/realtime/src/protocol.tsraw ← src6,150 B← Back to the app
1/**
2 * Frame protocol (TASK-0091 R3) — the wire shapes for `apps/realtime`.
3 *
4 * The schemas themselves live in `packages/shared/realtime/frames.ts`
5 * (client + server share them); this module is the server's framing
6 * layer: encode/decode of text + binary frames, the hostile-frame
7 * validation result codes, and the 10-strike policy (R3: 10 bad frames
8 * in a sliding 10 s window → close 4008 + audit).
9 *
10 * Framing (frozen — `packages/collaboration/src/types.ts` mirror):
11 *  - **Text**: one JSON object per frame, `type` discriminant.
12 *  - **Binary**: 1-byte kind prefix + raw Yjs bytes
13 *      `0x01` doc update · `0x02` awareness update.
14 */
15 
16import {
17  BINARY_KIND,
18  ClientFrameSchema,
19  REALTIME_LIMITS,
20  type ClientFrame,
21  type ServerFrame,
22} from "@factory/shared/realtime/frames";
23 
24// --- close codes (R6) — re-exported so server handlers import one module -------
25 
26export { WS_CLOSE } from "@factory/shared/realtime/frames";
27import { WS_CLOSE } from "@factory/shared/realtime/frames";
28 
29export type CloseCode = (typeof WS_CLOSE)[keyof typeof WS_CLOSE];
30 
31// --- client frame decoding -------------------------------------------------------
32 
33/** A decoded inbound frame. */
34export type DecodedFrame =
35  | { readonly kind: "text"; readonly frame: ClientFrame }
36  | {
37      readonly kind: "binary";
38      readonly binary: "update" | "awareness";
39      readonly bytes: Uint8Array;
40    };
41 
42/** Why a frame was rejected (all count as a strike except `unknown_frame`
43 *  policy is identical — R3: any bad frame is a strike). */
44export type FrameRejectCode =
45  | "invalid_json"
46  | "unknown_frame"
47  | "invalid_frame"
48  | "bad_binary_kind"
49  | "bad_update"
50  | "frame_too_large";
51 
52/**
53 * Decode one inbound WS frame (text = JSON control frame, binary =
54 * prefixed Yjs bytes). Pure — validation of a binary UPDATE's Yjs
55 * well-formedness happens at apply time (a corrupt update is a
56 * `bad_update` strike; the throwaway-doc probe is in server.ts because
57 * it allocates a `Y.Doc`).
58 */
59export function decodeClientFrame(
60  data: string | Uint8Array,
61): { ok: true; frame: DecodedFrame } | { ok: false; code: FrameRejectCode } {
62  if (typeof data === "string") {
63    let parsed: unknown;
64    try {
65      parsed = JSON.parse(data);
66    } catch {
67      return { ok: false, code: "invalid_json" };
68    }
69    const result = ClientFrameSchema.safeParse(parsed);
70    if (!result.success) {
71      // Distinguish a wholly-unknown `type` (unknown_frame) from a known
72      // type with a bad payload (invalid_frame) — the test matrix asserts
73      // both codes.
74      const t = (parsed as { type?: unknown } | null)?.type;
75      const known =
76        t === "sync.request" ||
77        t === "ping" ||
78        t === "outbox.pending" ||
79        t === "subscribe" ||
80        t === "unsubscribe" ||
81        t === "presence" ||
82        t === "presence-leave";
83      return { ok: false, code: known ? "invalid_frame" : "unknown_frame" };
84    }
85    return { ok: true, frame: { kind: "text", frame: result.data } };
86  }
87  // Binary: prefix byte + payload.
88  if (data.byteLength < 1) {
89    return { ok: false, code: "bad_binary_kind" };
90  }
91  const kind = data[0];
92  if (kind === BINARY_KIND.Update) {
93    return { ok: true, frame: { kind: "binary", binary: "update", bytes: data.subarray(1) } };
94  }
95  if (kind === BINARY_KIND.Awareness) {
96    return { ok: true, frame: { kind: "binary", binary: "awareness", bytes: data.subarray(1) } };
97  }
98  return { ok: false, code: "bad_binary_kind" };
99}
100 
101/** Encode an outbound server frame (always a text/JSON frame — binary
102 *  server→client frames don't exist in the R3 contract: doc updates are
103 *  relayed as binary 0x01 frames built here from raw bytes). */
104export function encodeServerFrame(frame: ServerFrame): string {
105  return JSON.stringify(frame);
106}
107 
108/** Encode an outbound binary doc update (the relay of an inbound 0x01). */
109export function encodeBinaryUpdate(bytes: Uint8Array): Uint8Array {
110  const out = new Uint8Array(bytes.byteLength + 1);
111  out[0] = BINARY_KIND.Update;
112  out.set(bytes, 1);
113  return out;
114}
115 
116/** Encode an outbound binary awareness update (0x02). */
117export function encodeBinaryAwareness(bytes: Uint8Array): Uint8Array {
118  const out = new Uint8Array(bytes.byteLength + 1);
119  out[0] = BINARY_KIND.Awareness;
120  out.set(bytes, 1);
121  return out;
122}
123 
124// --- the 10-strike policy (R3) -----------------------------------------------------
125 
126/**
127 * Per-socket bad-frame strike tracker. 10 strikes inside a sliding 10 s
128 * window → policy violation (the socket closes 4008 + audit
129 * `realtime.connection.abuse`). A strike that falls out of the window no
130 * longer counts (the counter is a timestamp list, not a running total).
131 */
132export class StrikeTracker {
133  private readonly limit: number;
134  private readonly windowMs: number;
135  private readonly now: () => number;
136  private strikes: number[] = [];
137 
138  constructor(opts?: { limit?: number; windowMs?: number; now?: () => number }) {
139    this.limit = opts?.limit ?? REALTIME_LIMITS.strikeLimit;
140    this.windowMs = opts?.windowMs ?? REALTIME_LIMITS.strikeWindowMs;
141    this.now = opts?.now ?? Date.now;
142  }
143 
144  /**
145   * Record a bad frame. Returns `violated` when the sliding window now
146   * holds `limit` strikes (the caller closes + audits exactly once —
147   * subsequent frames are ignored by the server for that socket).
148   */
149  record(): { violated: boolean; strikes: number } {
150    const t = this.now();
151    const cutoff = t - this.windowMs;
152    this.strikes = this.strikes.filter((s) => s > cutoff);
153    this.strikes.push(t);
154    const violated = this.strikes.length >= this.limit;
155    return { violated, strikes: this.strikes.length };
156  }
157 
158  /** Current in-window strike count (metrics/tests). */
159  count(): number {
160    const t = this.now();
161    const cutoff = t - this.windowMs;
162    return this.strikes.filter((s) => s > cutoff).length;
163  }
164 
165  reset(): void {
166    this.strikes = [];
167  }
168}
169 
170// --- frame size guard (R6: 256 KB cap) ----------------------------------------------
171 
172/** The server-side max inbound frame size (ws `maxPayload`). */
173export const MAX_FRAME_BYTES = REALTIME_LIMITS.maxFrameBytes;
174