CRDT Overview

Alice's editor:
  "Hello"
  ↓ (adds " World")
  "Hello World"

Bob's editor (at same time):
  "Hello"
  ↓ (adds "!")
  "Hello!"

After sync (both converge to):
  "Hello World!"
  ↑ CRDT algorithm automatically merges

┌──────────────────────────────────────────────┐
│ Client Application                           │
│  - Yjs Document (Y.Doc)                      │
│  - Shared types (Y.Text, Y.Map, Y.Array)     │
│  - Awareness (presence, cursors)             │
└──────────────────────────────────────────────┘
               ↓ WebSocket (binary protocol)
┌──────────────────────────────────────────────┐
│ Cloudillo Server                             │
│  ┌────────────────────────────────────────┐  │
│  │ WebSocket Handler                      │  │
│  │  - Yrs sync-protocol decoding          │  │
│  │  - Authentication (at connection)      │  │
│  └────────────────────────────────────────┘  │
│                    ↓                         │
│  ┌────────────────────────────────────────┐  │
│  │ Document Registry (CRDT_DOCS)          │  │
│  │  - Live Y.Doc + broadcast channels     │  │
│  │  - Loaded on first connect             │  │
│  │  - Optimized + dropped on last close   │  │
│  └────────────────────────────────────────┘  │
│                    ↓                         │
│  ┌─────────────────────────────────────────┐ │
│  │ Storage Layer                           │ │
│  │  - CrdtAdapter (binary updates)         │ │
│  └─────────────────────────────────────────┘ │
└──────────────────────────────────────────────┘

struct DocState {
    awareness_tx: Arc<broadcast::Sender<(String, Vec<u8>)>>,  // (conn_id, data)
    sync_tx: Arc<broadcast::Sender<(String, Vec<u8>)>>,       // (conn_id, data)
    doc: Arc<Mutex<Doc>>,                                      // live Y.Doc
}

static CRDT_DOCS: LazyLock<RwLock<HashMap<String, DocState>>> = /* ... */;

Client                          Server
  |                               |
  |--- GET /ws/crdt/:docId ------>|
  |    (Authorization: Bearer...) |
  |                               |--- Validate token
  |                               |--- Load database instance
  |                               |--- Create session
  |<-- 101 Switching Protocols ---|
  |                               |
  |<====== WebSocket Open =======>|
  |                               |
  |<-- SyncStep1 (server SV) -----|  Server sends its state vector
  |--- SyncStep1 (client SV) ---->|  Client sends its state vector
  |<-- SyncStep2 (diff) ----------|  Server replies with missing updates
  |--- SyncStep2 (client diff) -->|  Client sends what server is missing
  |                               |
  |<====== Synchronized =========>|
  |                               |
  |--- Update (user edits) ------>|--- Apply to live Doc, store, broadcast
  |<-- Update (echo back) --------|--- Echo to sender (keepalive)
  |<-- Update (remote edits) -----|
  |                               |
  |--- Awareness Update --------->|--- Broadcast to others
  |<-- Awareness Update ----------|--- Echo to sender

Input: WebSocket, user_id, doc_id, app, tn_id, read_only
Output: ()

1. Connection Setup:
   - Generate unique conn_id
   - Get or create the DocState (live Doc + channels) in CRDT_DOCS.
     The first connection loads the Doc via load_or_init_doc():
       - If no stored updates: create a Doc with a "meta" map, persist it
       - Otherwise: replay all stored updates into a fresh Doc (worker pool)
   - Create CrdtConnection (clones channels + doc)
   - Record initial file access (throttled)
   - Send SyncStep1 with the live Doc's state vector

2. Spawn Concurrent Tasks:
   - Heartbeat task: sends ping frames every 15 seconds
   - Receive task: processes incoming WebSocket messages
   - Sync broadcast task: forwards CRDT updates from other clients
   - Awareness broadcast task: forwards awareness updates from other clients

3. Message Loop (receive task), per Sync message:

   a. SyncStep1 (client state vector):
      - Compute diff from live Doc, reply with SyncStep2. Return (no broadcast).

   b. SyncStep2 / Update:
      - If read_only or empty: reject, return
      - Apply to the live Doc; if decode/apply fails: reject
      - If the apply was a no-op (snapshot unchanged): skip persist, return
      - Persist via CrdtAdapter.store_update(); on failure: return
      - Broadcast to other clients (SyncStep2 re-encoded as Update)
      - Echo back to sender (keepalive)

   c. Awareness:
      - Broadcast to other clients, echo back to sender

   Broadcast tasks skip messages originating from the same conn_id.

4. Connection Close:
   - Record final file access/modification
   - Abort heartbeat, sync, and awareness tasks
   - If last connection: wait 2s grace period, re-check receivers;
     if still none, remove DocState and optimize the document

This pattern ensures:
- Live Doc enables an instant state-vector handshake
- Apply-before-persist with no-op detection (avoids storing redundant updates)
- Echo + broadcast (sender gets echo, others get broadcast)
- Automatic optimization when all clients disconnect

Input: app, tn_id, doc_id, live Doc
Output: ()

1. Last connection closes → wait 2s grace period, re-check receivers.
   If a new connection arrived: skip optimization.

2. Load stored updates (for seq numbers and size comparison).
   If 0 or 1 updates: skip.

3. Encode the live Doc's full state as one update via
   encode_state_as_update_v1() — instant, no reconstruction.

4. If the merged update is not smaller than the originals: skip.

5. Atomically replace all updates via CrdtAdapter.compact_updates():
   removes the old seqs and inserts the merged update (client_id = "system")
   in a single redb transaction.

Benefits: faster initial sync and lower storage for subsequent connections.