⚠️ HISTORICAL DESIGN. The body below describes the pre-byte-relay broker design (op validation at ingest, DO SQLite persistence, D1 cold backup). It was superseded by the live thin byte-relay: the WorkspaceDoc DO (relay/cf/src/workspace-doc.ts) does NOT parse or validate ops — it fans out raw Yjs sync frames verbatim, and D1 (workspace_sub_doc) is the durable source of truth written by the ui worker. For the current contract, read the frontmatter summary above, REALTIME-ARCHITECTURE-V3.md, and CLAUDE.md §3. The body is retained as design history; do not implement from it.

Realtime Broker Design (Yjs over Durable Object)

The implementation spec for relay/cf/src/workspace-doc.ts. One Durable Object per workspace. Holds the canonical Y.Doc, validates ops at ingest, broadcasts Yjs deltas, persists to DO SQLite, cold-backs-up to D1.

This document is the contract. Do not deviate without updating this file first.

References: ARCHITECTURE.md (Y.Doc structure, persistence layout), PRINCIPLES.md #1 (compute economics), #3 (no fallbacks), #8 (client compute default), #12 (one source of truth). See also REALTIME-ARCHITECTURE-V3.md for the multi-pathway resilience model that wraps this broker.

0. Scope

In scope:

• Lifecycle of one WorkspaceDoc DO instance.
• Ingest, validate, apply, persist, broadcast for Yjs document updates.
• Awareness routing (presence, cursors). Awareness is NOT persisted.
• D1 cold backup cadence.
• Failure handling for the cases listed in §7.

Out of scope (handled elsewhere):

• Workspace creation in D1 (web Worker). DO assumes the workspace row exists; the Worker validates before routing (lift v1's zombie gate).
• LLM calls, Typst rendering, exports — all client-side per principle #8.
• Auth — link = identity; no auth surface in the broker.
• MCP server — separate Worker; it talks to this DO via service binding for ops it needs to inject server-side.

1. DO lifecycle

1.1 Instantiation

A WorkspaceDoc instance is materialized the first time any code calls env.WORKSPACE_DOC.get(idFromName(wsId)) and then invokes a method on the stub. The Worker entry already validates the workspace exists in D1 before this call (zombie gate, lifted from v1).

On constructor(state, env) the DO does the minimum:

• Stash state and env.
• Initialize the in-memory Y.Doc.
• Restore session metadata for any hibernated WebSockets via state.getWebSockets() + deserializeAttachment().
• Defer SQLite reads (snapshot + ops log) until the first fetch or webSocketMessage. The constructor must stay cheap because it runs on every wake.

1.2 First-touch hydration (hydrate())

Called lazily (idempotent, guarded by a hydrated: boolean flag). Reads the latest snapshot from SQLite, applies it to the Y.Doc, then replays the ops log on top. Pseudocode:

private async hydrate(): Promise<void> {
  if (this.hydrated) return;
  const snap = this.sql.exec<{ data: ArrayBuffer; clock: number }>(
    'SELECT data, clock FROM snapshot ORDER BY clock DESC LIMIT 1'
  ).one();
  if (snap) Y.applyUpdate(this.doc, new Uint8Array(snap.data), 'hydrate');

  const cursor = this.sql.exec<{ data: ArrayBuffer }>(
    'SELECT data FROM ops WHERE clock > ? ORDER BY clock ASC',
    snap?.clock ?? 0,
  );
  for (const row of cursor) {
    Y.applyUpdate(this.doc, new Uint8Array(row.data), 'hydrate');
  }
  this.hydrated = true;
}

Errors during hydrate are fatal — return 503 to the client. Per principle #3, no fallbacks; corruption surfaces, doesn't get papered over.

1.3 Hibernation

The runtime hibernates the DO when there are no active timers, no pending I/O, and all open WebSockets were acceptWebSocket()-ed (hibernatable mode). What survives:

• DO SQLite (storage class new_sqlite_classes per wrangler.toml): snapshot + ops log + meta rows persist across hibernation and crashes.
• Hibernated WebSockets: connections stay open at the runtime level; sockets are re-bound to a fresh JS object on wake.
• Per-socket attachments (serializeAttachment()): session metadata (userId, joinedAt) survives.

What does NOT survive:

• The in-memory Y.Doc. It must be reconstructed on wake from snapshot + ops log (§1.2).
• The Map<WebSocket, Session> view. Rebuild from state.getWebSockets() + deserializeAttachment() in the constructor.
• Awareness state. Treated as ephemeral by Yjs; clients re-broadcast their awareness on reconnect (and on wake clients discover each other again via fresh sync step 1).

1.4 Wake path

1. Constructor runs, restores socket metadata, sets hydrated = false.
2. First inbound message (or fetch) calls hydrate(). Latency budget: see §10.
3. While hydrating, inbound messages are queued (one per socket, bounded — see §10).
4. After hydrate, drain the queue in arrival order. From now on, normal path.

2. Y.Doc canonical state

2.1 Where it lives

The Y.Doc lives in DO memory (this.doc: Y.Doc) while the DO is awake. There is exactly one Y.Doc per workspace per DO instance. Per principle #12, this is the source of truth for the workspace's editable state during the live window. D1 is cold backup, not a parallel writer.

The shape inside the Y.Doc matches ARCHITECTURE.md §"Y.Doc structure per workspace": settings, layout, order_state, cv_profile, cl_profile, jobs, cloud_keys. The broker does not care about substructure beyond schema validation (§8); it stores opaque updates.

2.2 Persistence model

Two SQLite tables in the DO's storage:

-- Append-only log of validated updates since the last snapshot.
CREATE TABLE IF NOT EXISTS ops (
  clock     INTEGER PRIMARY KEY,   -- monotonic op counter, NOT a Yjs clock
  data      BLOB    NOT NULL,      -- raw Yjs update bytes
  origin    TEXT,                  -- userId from the WS session, or 'mcp', 'system'
  ts        INTEGER NOT NULL       -- ms epoch when applied
);

-- One snapshot row at a time after compaction; older rows kept until R2 archive (out of scope here).
CREATE TABLE IF NOT EXISTS snapshot (
  clock     INTEGER PRIMARY KEY,   -- ops.clock at the moment of snapshot
  data      BLOB    NOT NULL,      -- Y.encodeStateAsUpdate(doc)
  ts        INTEGER NOT NULL,
  byte_size INTEGER NOT NULL
);

CREATE TABLE IF NOT EXISTS meta (
  key       TEXT PRIMARY KEY,
  value     TEXT NOT NULL
);
-- Keys: 'op_counter', 'last_d1_backup_clock', 'last_d1_backup_ts'.

clock is a DO-local monotonic counter, NOT a Yjs vector clock. It exists so the broker can replay ops in the order they were applied without trusting client-supplied ordering. Yjs handles real causal ordering inside the update bytes.

2.3 Concrete numbers (N, M)

Snapshot when EITHER condition fires:

• N = 200 ops since the last snapshot, OR
• M = 5 minutes of wall-clock time since the last snapshot if any ops were appended in that window.

Reasoning:

• Cold-start cost on wake is dominated by replaying the ops log. 200 small CRDT updates apply in <50ms; that's the pain-free ceiling.
• A typical edit session bursts ~30-80 ops in a minute (typing in fields, reordering sections). N=200 means a snapshot every few minutes during active editing, not every keystroke.
• M=5min catches the long-tail case where a workspace gets one op every 90 seconds — without M, the log would grow unbounded. The 5-minute upper bound on staleness keeps wake latency predictable.
• These numbers are tunable. See §11 open questions.

A snapshot pass is also forced before D1 backup (§6) so the cold backup is always consistent.

3. Connection lifecycle

The broker speaks the standard Yjs sync protocol over a single WebSocket. Reference: y-protocols (specifically y-protocols/sync and y-protocols/awareness), encoded with lib0. Messages are framed as binary Uint8Arrays; the first varint is the message type.

3.1 Message types (Yjs convention)

Inside messageSync, the second varint distinguishes sub-types: 0=SyncStep1, 1=SyncStep2, 2=Update.

3.2 Connect flow

1. Client picks a relay entry from PUBLIC_RELAY_URLS via its RelayWalker (ui/src/lib/realtime/relay-walker.ts) and connects to <wsUrl>/<wsId>?userId=<deviceId> (today wss://cf.relay.careervector.corbet.ch/ws/<wsId>?userId=<deviceId> for the CF entry, wss://deno.relay.careervector.corbet.ch/ws/<wsId>?userId=<deviceId> for the Deno entry). See REALTIME-ARCHITECTURE-V3.md §"Resilience: multi-vendor relay" for the walker contract.
2. Worker validates wsId format and existence in D1 (lifted from v1 — keep the zombie gate verbatim).
3. Worker forwards the upgrade to the DO stub. DO calls state.acceptWebSocket(server) and stores Session { userId, connectedAt } via serializeAttachment.
4. DO triggers hydrate() if not already done.
5. Server initiates sync: DO sends SyncStep1(stateVector(this.doc)) to the new client. (Yjs allows either side to start; server-initiated keeps the catch-up payload deterministic.)
6. Client responds with SyncStep2(diff) containing all updates the server is missing, plus its own SyncStep1 requesting the server's state.
7. DO answers with its SyncStep2. New client now has full state; server has client's offline-accumulated ops.
8. DO sends current awareness snapshot (messageAwareness with all known peer states).
9. Client sends its initial awareness; DO routes per §4.

3.3 Live phase

While the connection is open:

• Any messageSync(Update) from the client is validated (§8), applied to the canonical Y.Doc, persisted (§2), and broadcast (§4).
• Any messageAwareness is applied to the awareness instance and broadcast (§4).
• Any messageQueryAwareness is answered with the current awareness snapshot back to the asking socket.

3.4 Catch-up after disconnect

Yjs handles this natively via vector clocks — there is no separate "missed ops" channel. On reconnect, the client repeats §3.2 from step 1, the server's SyncStep1 carries its current vector clock, the client's SyncStep2 sends only the deltas the server is missing, and vice versa. No full snapshot transfer unless §7.4 triggers.

4. Broadcast topology

One DO, N WebSockets. Star topology — every applied update fans out to the other N-1 sockets.

4.1 Document updates

When a messageSync(Update) from socket A is validated and applied:

// Pseudocode
const update = readSyncUpdate(decoder);
validate(update);                           // §8
Y.applyUpdate(this.doc, update, sessionA);  // origin tag = sender socket
appendToOpsLog(update, sessionA.userId);    // §2.2
maybeCompact();                             // §5
broadcastUpdate(update, exceptSocket=A);    // fan-out

The fan-out re-encodes the update inside a messageSync(Update) envelope and sends the binary frame to every other open socket. Sender is excluded — it already has the change locally (CRDT property: applying your own update is a no-op).

4.2 Awareness

Awareness updates use messageAwareness framing and Awareness (y-protocols/awareness) state-merge semantics. Awareness is NEVER applied to the Y.Doc and NEVER persisted. It's strictly in-DO-memory + broadcast.

Behavior:

• Apply incoming awareness update to this.awareness (so the DO can answer messageQueryAwareness from late joiners).
• Broadcast verbatim to all other sockets.
• On webSocketClose, remove the disconnecting client's awareness state and broadcast a removal update.

This is the data path that powers v1's upper-right presence indicator. The UI is feature-complete (principle #13); we're only swapping the wire format.

4.3 Why server-applies (vs pure relay)

v1 was a pure relay: binary frames forwarded blind. v2 applies updates server-side because:

1. We need a canonical Y.Doc to persist (snapshot/ops log).
2. We need to validate against Zod schemas (§8) before broadcast — invalid ops never reach other clients.
3. Late joiners need the server to answer SyncStep1 from a real state, not from "whatever the last client to talk had".

Cost: applying a Yjs update is microseconds. The DO stays well under the 10ms-per-request principle.

5. Snapshot compaction

5.1 Trigger

maybeCompact() runs after every appended op. It checks:

• Op count since last snapshot ≥ N (200), OR
• Wall-clock since last snapshot ≥ M (5 minutes) AND at least one op since.

A setAlarm() covers the M case when the DO would otherwise hibernate before the timer fires; the alarm handler runs maybeCompact() and re-arms if needed.

5.2 Procedure

private async compact(): Promise<void> {
  const fullState = Y.encodeStateAsUpdate(this.doc);
  const clock = this.opCounter;
  const ts = Date.now();
  this.sql.transactionSync(() => {
    this.sql.exec(
      'INSERT INTO snapshot (clock, data, ts, byte_size) VALUES (?, ?, ?, ?)',
      clock, fullState, ts, fullState.byteLength,
    );
    this.sql.exec('DELETE FROM ops WHERE clock <= ?', clock);
    this.sql.exec(
      'INSERT OR REPLACE INTO meta (key, value) VALUES (?, ?)',
      'last_snapshot_clock', String(clock),
    );
  });
}

transactionSync makes the snapshot insert + ops delete atomic. If the DO crashes between the insert and delete, the next hydrate sees a snapshot at clock K and ops at clock ≤ K; replaying those ops is a no-op for the Y.Doc (CRDT idempotence) but wastes time. We tolerate this — see §7.1 (soft-fail).

5.3 Older snapshots

Older snapshot rows are kept in DO SQLite for 24h then archived to R2 (per ARCHITECTURE.md's "Older snapshots… R2"). The R2 archival cron is OUT OF SCOPE for this design — open question §11.5.

For now: compact() keeps only the most recent snapshot row (the INSERT could be INSERT OR REPLACE keyed on clock if we collapse to single row; we keep the multi-row table to leave room for the archival job to read, then delete). Implementer choice — start with single-row, expand only when archival lands.

6. D1 cold backup

D1 is the durable source of truth per ARCHITECTURE.md and REALTIME-ARCHITECTURE-V3.md. Per principle #1, we don't burn writes that aren't needed.

6.1 Cadence

Backup to D1 ONCE PER HOUR while the DO is awake AND has unsaved changes. Implementation: a DO alarm (state.storage.setAlarm) set on first op after the previous backup, scheduled for lastBackupTs + 1h (or immediately if more than 1h has elapsed). The handler:

1. Forces a compact() so the snapshot row reflects the latest state.
2. Reads the current snapshot bytes.
3. Writes to D1: UPDATE workspaces SET y_snapshot = ?, y_snapshot_clock = ?, y_snapshot_ts = ? WHERE id = ?.
4. Records last_d1_backup_clock + last_d1_backup_ts in meta.
5. Re-arms the alarm only if new ops have arrived since the backup completed.

The DO may also hibernate between backups; that's fine because DO SQLite IS durable. D1 is purely the "DO got nuked / migration disaster / human inspection" backup.

6.2 Justification (principle #1)

A typical workspace edited daily: ~24 D1 writes/day for backup, vs the alternative options:

D1 free tier is 100k writes/day across the whole account. Hourly per workspace scales to 4000 active workspaces before we'd need to revisit, which is far beyond v1's reality.

6.3 D1 schema delta

Add to existing workspaces table (cold-backup columns; nullable to coexist with v1):

ALTER TABLE workspaces ADD COLUMN y_snapshot BLOB;
ALTER TABLE workspaces ADD COLUMN y_snapshot_clock INTEGER;
ALTER TABLE workspaces ADD COLUMN y_snapshot_ts INTEGER;

Migration tracked separately. The web Worker reads y_snapshot to bootstrap a fresh client load before WebSocket sync (per ARCHITECTURE.md §"Read-only views").

7. Failure modes

7.1 DO crash mid-write (soft-fail)

Yjs updates are CRDT — applying the same update twice is idempotent, applying overlapping updates converges. So:

• If the DO crashes between Y.applyUpdate and appendToOpsLog: the in-memory Y.Doc is gone anyway; on next hydrate the update is lost UNLESS the client retries. Clients DO retry: Yjs's IndexedDB persistence holds the update locally and the next sync round re-sends it.
• If the DO crashes between appendToOpsLog and broadcast: persisted but other clients didn't see it live. They learn it on their next sync round (reconnect or next ping).
• If the DO crashes mid-compact(): the transactionSync ensures atomicity. Worst case is a snapshot row with overlapping ops in the log; replay is idempotent.

Log this category as soft-fail: structured log entry { kind: 'soft_fail', phase: 'mid-apply'|'mid-append'|'mid-broadcast'|'mid-compact', wsId, ts }. Don't surface to the user. Don't throw.

7.2 Client clock skew

Yjs uses Lamport-style logical clocks per client (clientID + monotonic seq), not wall clocks. Skew is invisible at the protocol layer. ts in the ops log is server-side Date.now(), so the broker never trusts client wall clocks. No mitigation needed. Documented here so reviewers know why we don't validate client timestamps.

7.3 Concurrent same-field edits

Yjs converges deterministically:

• Y.Map<string, primitive> field set by two clients concurrently: the update with the higher (clientID, clock) tuple wins, applied identically on every replica. No data loss in the sense of inconsistency; one user's value is overwritten by another's, exactly as a single-user LWW would behave for that field.
• Y.Text concurrent edits at the same position: character-level merge. Both contributions are preserved.
• Y.Array concurrent inserts at the same index: causal ordering produces a deterministic order on every replica.

The broker does NOT need to mediate. We document the property here so the implementer knows not to add "conflict resolution" logic — Yjs has it.

7.4 Long disconnect → catch-up payload too large

If a client has been offline for a long time and the server's "diff since your state vector" exceeds a hard cap, refuse the incremental delta and send a full state snapshot instead.

const CATCHUP_CAP_BYTES = 2_000_000; // 2 MB
const diff = Y.encodeStateAsUpdate(this.doc, clientStateVector);
if (diff.byteLength > CATCHUP_CAP_BYTES) {
  // Send full snapshot (smaller in pathological history-heavy cases) and
  // a fresh empty state vector, OR simply send the full snapshot if it's smaller.
  const full = Y.encodeStateAsUpdate(this.doc);
  send(sock, syncStep2(full.byteLength < diff.byteLength ? full : diff));
} else {
  send(sock, syncStep2(diff));
}

The 2 MB cap is conservative; on Workers we have plenty of headroom but the client's IndexedDB / memory budget is the real constraint. Open question §11.4.

7.5 Malformed op from client

Per principle #3 (no fallbacks), the broker validates EVERY incoming Yjs update against lib/schemas Zod definitions BEFORE applying. Validation strategy in §8.

On validation failure:

• Do NOT apply the update.
• Do NOT broadcast.
• Do NOT log to ops table.
• Send a messageAuth-style deny to the offender? No. Yjs auth messages are for permission denial, which doesn't apply here. Instead, define a new envelope:

// messageType = 100 (custom; out of y-protocols range)
// payload: { code: 'schema_invalid', issues: ZodIssue[], updateClock: clientClock }

The client logs and surfaces. Repeated violations from one socket → close socket with code 1008 (policy violation). Threshold: 5 violations within 60s.

A successfully validated but semantically suspect update (e.g., creates a section with a kind field not in the registry) — the schema should catch that at parse time. If the schema doesn't catch it, that's a lib/schemas bug, not a broker concern.

7.6 D1 unavailable during cold backup

Log { kind: 'd1_backup_failed', wsId, error } and re-arm the alarm with backoff (next attempt: 5min, 15min, 60min). DO SQLite is unaffected — the DO continues serving live traffic. Backup is best-effort.

8. Schema validation at ingest

Every Yjs update is validated against lib/schemas before being applied to the canonical Y.Doc. Reject invalid ops without applying. Principle #3.

8.1 The challenge

A Yjs update is opaque binary — it doesn't directly say "set jobs[abc].title = 'X'". To validate semantics we have to:

1. Decode the update against a temp Y.Doc snapshotted from current state.
2. Diff the resulting state against current state to extract the logical changes (which paths in the doc tree changed, what the new values are).
3. Run each change through the corresponding Zod schema.
4. If all changes validate, apply the original update bytes to the real this.doc. If not, reject.

8.2 Implementation sketch

async validate(update: Uint8Array): Promise<ValidationResult> {
  const probe = new Y.Doc();
  Y.applyUpdate(probe, Y.encodeStateAsUpdate(this.doc));     // clone current
  Y.applyUpdate(probe, update);                              // apply candidate

  const changes = diffYDocs(this.doc, probe);                // returns Path[]
  for (const change of changes) {
    const schema = pickSchema(change.path);                  // routes to lib/schemas
    const parsed = schema.safeParse(change.newValue);
    if (!parsed.success) return { ok: false, issues: parsed.error.issues, path: change.path };
  }
  return { ok: true };
}

pickSchema(path) is a dispatcher in lib/mutations that maps paths like ['jobs', jobId] → JobSchema, ['settings'] → WorkspaceSettingsSchema, etc.

8.3 Performance

Cloning the Y.Doc on every update is the obvious cost. Mitigations:

• For small workspaces (~95% of cases) the clone+diff costs <1ms.
• For large workspaces, switch to delta-decoding the update via Y.decodeUpdate and walking only the touched structures. Optimization deferred until measured.

Open question §11.2.

8.4 What schemas exist

Validated paths (each maps to a Zod schema in lib/schemas):

If pickSchema returns no schema for a path, that's a bug — throw, don't pass. Principle #3.

9. Ops the broker accepts

The broker doesn't have RPC-shaped ops; it accepts Yjs updates that touch any of the documented paths. At a high level the catalog is:

• Workspace settings: settings.*, layout.*, order_state.*.
• Profiles: cv_profile.<lang>.tree (recursive Node tree with variant-pool wrappers), cl_profile.<lang>.tree. Plus format per language.
• Jobs lifecycle: create (jobs.<id> = {...}), update (any sub-path), delete (jobs.<id> removed).
• Per-job tailored output: jobs.<id>.tailored_cv, jobs.<id>.tailored_cl (full Node-tree snapshots).
• Per-job derived data: jobs.<id>.evaluations, jobs.<id>.origins.
• Cloud keys / cascade chains: cloud_keys.providers, cloud_keys.chains.

The typed op catalog that produces these updates is in lib/mutations/src/opsCatalog.ts; current OP_CATALOG_VERSION is exported from lib/mutations/src/index.ts. See CLAUDE.md §24 for the canonical op contract.

Awareness (cursor positions, presence) is separate — see §4.2.

This is descriptive, not prescriptive: any Yjs update touching the documented Y.Doc shape is valid input. Schemas in lib/schemas define what's accepted under each path; the broker is path-agnostic beyond schema dispatch.

10. Hibernation costs

10.1 Wake-up budget

This is well within the user-perceived "instant" budget for a reconnect.

10.2 Behavior during wake

While hydrate() is in-flight:

• webSocketMessage handler enqueues incoming frames into a per-socket pendingFrames: Uint8Array[].
• Per-socket queue capped at 32 frames; on overflow, close the socket with code 1011 (server overload). Client reconnects, re-syncs. Better than memory-bombing the DO.
• After hydrate() resolves, drain the queues in arrival order per socket. Awareness frames are processed first within each queue (cheap, no validation), then sync frames.

10.3 What we explicitly don't do

• Don't pre-warm. Cloudflare runtime decides; we cooperate.
• Don't snapshot-on-hibernate. Hibernation is silent; we can't run code in the hibernate path. The M=5min timer plus the alarm-fired backup pattern covers durability.

11. Open questions

These need user input before implementation. Default behavior is listed; flag them at decision time.

1. Snapshot frequency tradeoff (N, M). Current proposal: N=200 ops, M=5min. Higher N means longer wake replay but fewer compaction events; lower N means faster wake but more SQLite churn. Want to gut-check with realistic editing telemetry before locking. Default: ship with N=200, M=300s, instrument and revisit.

2. Schema validation cost on large workspaces. Cloning the full Y.Doc per incoming update is O(state size). For workspaces with many jobs (hundreds), this could exceed the 10ms-per-request principle. Alternative: parse the update bytes directly and validate just the touched subtree. More complex; harder to be sure we caught everything. Default: ship with full-clone, measure, optimize if needed.

3. D1 backup cadence. Current proposal: 1 hour while active. Should we also force a backup right before announced maintenance windows or DO migrations? Default: hourly only; rely on operational discipline.

4. Catch-up payload cap (CATCHUP_CAP_BYTES). 2 MB is a guess. Real constraint is client IndexedDB write speed + initial render budget. Default: ship 2 MB, watch for "long disconnect" telemetry.

5. R2 archival of old snapshots. ARCHITECTURE.md mentions R2 holds older snapshots. This design doesn't specify the archival job. Open: is it a cron Worker that scans D1 for old workspace_sub_doc rows, or does the DO itself push to R2 when a new snapshot supersedes the old? The latter couples the DO to R2 writes (more failure surface). Default: out of scope; design separately when archival becomes necessary.

Appendix A — Type sketches

// relay/cf/src/workspace-doc.ts — target shape

import * as Y from 'yjs';
import { Awareness, encodeAwarenessUpdate, applyAwarenessUpdate } from 'y-protocols/awareness';
import * as syncProtocol from 'y-protocols/sync';
import * as encoding from 'lib0/encoding';
import * as decoding from 'lib0/decoding';
import { validateUpdate } from '@cv/mutations';   // §8

const MSG_SYNC = 0;
const MSG_AWARENESS = 1;
const MSG_QUERY_AWARENESS = 3;
const MSG_SCHEMA_REJECT = 100;             // §7.5

type Session = { userId: string; connectedAt: number };

interface Env {
  CV_DB: D1Database;
}

export class WorkspaceDoc {
  private state: DurableObjectState;
  private env: Env;
  private sql: SqlStorage;
  private doc = new Y.Doc();
  private awareness = new Awareness(this.doc);
  private sessions = new Map<WebSocket, Session>();
  private hydrated = false;
  private opCounter = 0;
  private opsSinceSnapshot = 0;
  private lastSnapshotTs = 0;
  private violationsBySocket = new WeakMap<WebSocket, number[]>();

  constructor(state: DurableObjectState, env: Env) {
    this.state = state;
    this.env = env;
    this.sql = state.storage.sql;
    this.bootstrapSchema();
    for (const ws of state.getWebSockets()) {
      const meta = ws.deserializeAttachment() as Session | null;
      if (meta) this.sessions.set(ws, meta);
    }
  }

  private bootstrapSchema(): void { /* CREATE TABLE IF NOT EXISTS … */ }

  private async hydrate(): Promise<void> { /* §1.2 */ }

  async fetch(req: Request): Promise<Response> {
    const upgrade = req.headers.get('Upgrade');
    if (upgrade !== 'websocket') return new Response('expected websocket', { status: 426 });
    await this.hydrate();
    const url = new URL(req.url);
    const userId = url.searchParams.get('userId') ?? `anon_${crypto.randomUUID()}`;

    const pair = new WebSocketPair();
    const [client, server] = [pair[0], pair[1]];
    this.state.acceptWebSocket(server);
    const session: Session = { userId, connectedAt: Date.now() };
    server.serializeAttachment(session);
    this.sessions.set(server, session);

    // Server-initiated sync step 1
    this.sendSyncStep1(server);
    // Send current awareness snapshot
    this.sendAwarenessSnapshot(server);

    return new Response(null, { status: 101, webSocket: client });
  }

  async webSocketMessage(ws: WebSocket, msg: ArrayBuffer | string): Promise<void> {
    if (typeof msg === 'string') return; // we don't speak JSON anymore
    await this.hydrate();
    const decoder = decoding.createDecoder(new Uint8Array(msg));
    const messageType = decoding.readVarUint(decoder);
    switch (messageType) {
      case MSG_SYNC:        return this.onSync(ws, decoder);
      case MSG_AWARENESS:   return this.onAwareness(ws, decoder);
      case MSG_QUERY_AWARENESS: return this.sendAwarenessSnapshot(ws);
      default: /* unknown — ignore */ return;
    }
  }

  async webSocketClose(ws: WebSocket): Promise<void> {
    const session = this.sessions.get(ws);
    this.sessions.delete(ws);
    if (session) {
      // Remove this client's awareness state and broadcast removal
      const clientID = (this.awareness as any).clientIDsBySocket?.get(ws); // tracked at connect
      if (clientID != null) {
        this.awareness.removeAwarenessStates([clientID], ws);
      }
    }
  }

  async alarm(): Promise<void> {
    await this.hydrate();
    await this.maybeCompact(/* force */ true);
    await this.backupToD1();
    if (this.opsSinceSnapshot > 0 || /* new ops since backup */ false) {
      this.state.storage.setAlarm(Date.now() + 60 * 60 * 1000);
    }
  }

  // — internal —
  private onSync(ws: WebSocket, decoder: decoding.Decoder): void { /* §3, §4.1, §8 */ }
  private onAwareness(ws: WebSocket, decoder: decoding.Decoder): void { /* §4.2 */ }
  private sendSyncStep1(ws: WebSocket): void { /* §3.2 */ }
  private sendAwarenessSnapshot(ws: WebSocket): void { /* §4.2 */ }
  private appendOp(update: Uint8Array, origin: string): void { /* §2.2 */ }
  private async maybeCompact(force?: boolean): Promise<void> { /* §5 */ }
  private async backupToD1(): Promise<void> { /* §6 */ }
  private rejectUpdate(ws: WebSocket, issues: unknown, clock: number): void { /* §7.5 */ }
  private broadcastUpdate(update: Uint8Array, except: WebSocket): void { /* §4.1 */ }
}

Appendix B — Differences from v1 broker