@vuer-ai/vuer-rtc-server

Real-time collaborative library for vuer.ai. Server is built with MongoDB persistence and Fastify.

The CRDT logic is in @vuer-ai/vuer-rtc. This package provides server-specific functionality.

Setup

Prerequisites

MongoDB with replica set support (required for Prisma transactions).

MongoDB Setup with Docker

1. Start MongoDB and Redis containers:

   cd ../../docker
   docker compose up -d

   This starts:

   • MongoDB 7 with replica set rs0 on port 27017
   • Redis 7 on port 6379

2. Verify replica set is initialized:

   docker exec vuer-rtc-mongo mongosh --quiet --eval "rs.status().set"

   Should output: rs0

3. Configure database connection:

   Create .env file in packages/vuer-rtc-server/:

   DATABASE_URL="mongodb://localhost:27017/vuer?replicaSet=rs0"

4. Push Prisma schema to MongoDB:

   cd packages/vuer-rtc-server
   npx prisma db push

5. Run tests:

   pnpm test

Stopping Services

cd ../../docker
docker compose down      # Stop containers (keep data)
docker compose down -v   # Stop containers and remove volumes (delete all data)

Note for macOS users: If you have MongoDB installed via Homebrew, stop it first to avoid port conflicts:

brew services stop mongodb-community

How CRDT Works

The Core Concept

CRDT (Conflict-free Replicated Data Types) allows multiple users to edit the same data concurrently without conflicts. The key insight: instead of locking or resolving conflicts, design operations that always merge correctly.

Architecture

┌─────────────┐     CRDTMessage      ┌─────────────┐
│   Client A  │ ──────────────────▶  │   Server    │
│  (Alice)    │                      │             │
└─────────────┘                      │  1. Append  │
                                     │     to      │
┌─────────────┐     CRDTMessage      │    Journal  │
│   Client B  │ ──────────────────▶  │             │
│  (Bob)      │                      │  2. Apply   │
└─────────────┘                      │     to      │
                                     │    Graph    │
                                     └─────────────┘
                                           │
                                           ▼
                                     Broadcast to all

State Model

The server maintains two data structures:

interface ServerState {
  graph: SceneGraph;        // Materialized view (fast reads)
  journal: CRDTMessage[];   // Append-only log (source of truth)
  snapshot?: Snapshot;      // Periodic checkpoint (optional)
}

| Component | Purpose | Persistence | |-----------|---------|-------------| | graph | Current state for queries | In-memory (rebuilt from journal) | | journal | Complete history of all messages | MongoDB / disk | | snapshot | Checkpoint to speed up recovery | MongoDB / disk |

Why both?

• graph = fast reads, but lost on restart
• journal = durable, enables replay, sync, and debugging

Key Components

1. CRDTMessage (Envelope) - Contains metadata for ordering:

   • lamportTime - Logical clock for total ordering (LWW)
   • clock - Vector clock for causal ordering
   • sessionId - Who sent this message

2. Operations - Explicit otype determines merge behavior:

   • vector3.set → Last-Write-Wins (absolute)
   • vector3.add → Sum values (relative/additive)
   • number.add → Sum values (counters)

Why Explicit otype Matters

Scenario: Two users drag the same object simultaneously

❌ Without explicit otype (ambiguous):

Alice: position = [5, 0, 0]  // Did she SET or ADD?
Bob:   position = [0, 3, 0]  // Did he SET or ADD?
// Result: ??? (unpredictable)

✅ With explicit otype (unambiguous):

Alice: { otype: 'vector3.add', value: [5, 0, 0] }  // += [5,0,0]
Bob:   { otype: 'vector3.add', value: [0, 3, 0] }  // += [0,3,0]
// Result: position += [5, 3, 0] ✅ (both movements apply)

Merge Rules by otype

| otype | Merge | Use Case | |-------|-------|----------| | *.set | Last-Write-Wins (higher lamport wins) | Absolute values | | *.add | Sum all values | Counters, drag deltas | | *.multiply | Product | Scale gestures | | array.push | Append all | Adding children | | array.remove | Remove from all | Removing children |

Journal Lifecycle

The journal is the source of truth. The graph is just a materialized view.

Message Flow

Client sends CRDTMessage
         │
         ▼
┌─────────────────────────┐
│  1. Deduplicate         │  ← Skip if msg.id already in journal
│  2. Append to journal   │  ← Persist to MongoDB
│  3. Apply to graph      │  ← Update in-memory state
│  4. Broadcast           │  ← Send to other clients
└─────────────────────────┘

Idempotency

Critical: Not all operations are idempotent on replay!

| Operation Type | Idempotent? | Why | |---------------|-------------|-----| | *.set (LWW) | ✅ Yes | Compares lamportTime, same result on replay | | *.add | ❌ No | Blindly adds value, doubles on replay | | *.multiply | ❌ No | Blindly multiplies, compounds on replay | | node.insert | ✅ Yes | Checks if node exists |

Solution: Track applied message IDs per node to skip duplicates:

interface SceneNode {
  // ... existing fields
  appliedMsgIds: Set<string>;  // Track which messages contributed
}

function applyOperation(node, op, meta) {
  if (node.appliedMsgIds.has(meta.messageId)) {
    return; // Already applied, skip
  }
  node.appliedMsgIds.add(meta.messageId);
  // ... apply operation
}

Recovery Flow

When server restarts:

1. Load snapshot from DB (if exists)
   └─ snapshot = { graph, journalIndex }

2. Load journal entries after snapshot
   └─ journal.slice(snapshot.journalIndex)

3. Replay journal onto snapshot.graph
   └─ for (msg of journal) graph = applyMessage(graph, msg)

4. Server ready

Client Sync Flow

When a new client connects:

┌────────────┐                      ┌────────────┐
│ New Client │                      │   Server   │
└─────┬──────┘                      └─────┬──────┘
      │                                   │
      │  1. Connect                       │
      │──────────────────────────────────▶│
      │                                   │
      │  2. Send current graph            │
      │◀──────────────────────────────────│
      │     (or snapshot + journal tail)  │
      │                                   │
      │  3. Client applies, catches up    │
      │                                   │
      │  4. Subscribe to live updates     │
      │◀─────────────────────────────────▶│
      │                                   │

Option A: Send full graph (simple, but large) Option B: Send snapshot + journalTail (smaller, incremental)

Compaction

The journal grows unbounded. Periodically compact:

Before compaction:
  journal: [msg1, msg2, msg3, ..., msg1000]
  snapshot: null

After compaction:
  journal: [msg901, msg902, ..., msg1000]  ← Keep recent
  snapshot: { graph: <state at msg900>, journalIndex: 900 }

When to compact:

• Journal exceeds N messages (e.g., 1000)
• Periodic timer (e.g., every hour)
• On graceful shutdown

Usage

import { createEmptyGraph, applyMessage } from '@vuer-rtc/server/operations';
import type { CRDTMessage } from '@vuer-rtc/server/operations';

// Create empty scene
let graph = createEmptyGraph();

// Apply a message with operations
const msg: CRDTMessage = {
  id: 'msg-001',
  sessionId: 'alice',
  clock: { alice: 1 },
  lamportTime: 1,
  timestamp: Date.now(),
  ops: [
    // Create scene root
    {
      key: 'scene',
      otype: 'node.insert',
      path: 'scene',
      value: { key: 'uuid-scene', tag: 'Scene', name: 'My Scene' },
    },
    // Create cube with parent (automatically adds to parent's children)
    {
      key: 'cube-1',
      otype: 'node.insert',
      path: 'cube-1',
      parent: 'scene', // Automatically adds to scene's children
      value: {
        key: 'uuid-001',
        tag: 'Mesh',
        name: 'Red Cube',
        color: '#ff0000',
        'transform.position': [0, 0, 0],
      },
    },
  ],
};

graph = applyMessage(graph, msg);

CRDTMessage Structure

interface CRDTMessage {
  id: string;           // Message ID
  sessionId: string;    // Who sent this
  clock: VectorClock;   // For causal ordering
  lamportTime: number;  // For total ordering (LWW)
  timestamp: number;    // Wall-clock time
  ops: Operation[];     // Batch of operations
}

Operation Types

Node Operations

| otype | Description | |-------|-------------| | node.insert | Create new node (idempotent). Use parent field to auto-add to parent's children. | | node.remove | Delete node (tombstone) |

Number Operations

| otype | Merge | Example | |-------|-------|---------| | number.set | LWW | opacity = 0.5 | | number.add | Sum | score += 10 | | number.multiply | Product | scale *= 2 | | number.min | Minimum | min(current, new) | | number.max | Maximum | max(current, new) |

Vector3 Operations

| otype | Merge | Example | |-------|-------|---------| | vector3.set | LWW | position = [0, 5, 0] | | vector3.add | Component sum | position += [5, 0, 0] | | vector3.multiply | Component product | scale *= [2, 2, 2] |

Array Operations

| otype | Merge | Example | |-------|-------|---------| | array.set | LWW | children = ['a', 'b'] | | array.push | Append | children.push('c') | | array.remove | Remove | children.remove('a') | | array.union | Union | Merge sets |

Other Operations

| otype | Merge | Example | |-------|-------|---------| | color.set | LWW | color = '#ff0000' | | string.set | LWW | name = 'Cube' | | boolean.set | LWW | visible = true | | quaternion.set | LWW | rotation = [0, 0, 0, 1] |

Additive vs LWW

Additive Operations (*.add)

Order doesn't matter, values accumulate:

Alice: position += [5, 0, 0]
Bob:   position += [0, 3, 0]
Result: position += [5, 3, 0] ✅

LWW Operations (*.set)

Higher lamportTime wins:

Alice: color = red   (lamport: 10)
Bob:   color = blue  (lamport: 11)
Result: color = blue ✅ (Bob's lamport is higher)

Client Reconciliation

When a client receives a message from the server, it applies it using the same applyMessage function:

1. Client makes local edit     →  applies locally (optimistic)
2. Client sends to server      →  server applies & broadcasts
3. Client receives broadcast   →  applies with deduplication

Important: Clients must track applied message IDs to avoid double-applying their own messages:

// Client-side state
const appliedMsgIds = new Set<string>();

function onServerMessage(msg: CRDTMessage) {
  if (appliedMsgIds.has(msg.id)) {
    return; // Already applied locally, skip
  }
  appliedMsgIds.add(msg.id);
  graph = applyMessage(graph, msg);
}

For LWW operations: If the server's lamport time is higher, the server value wins:

// Client has: color = '#ff0000' (lamport: 5)
// Server sends: color = '#0000ff' (lamport: 8)
graph = applyMessage(graph, serverMsg);  // color becomes '#0000ff'

For additive operations: Each unique message applies once:

// Client applied locally: position += [5, 0, 0] (msg-alice-1)
// Server broadcasts same message back
// Client skips (already in appliedMsgIds)

// Server sends Bob's edit: position += [0, 3, 0] (msg-bob-1)
// Client applies (new message ID)
graph = applyMessage(graph, serverMsg);  // position += [0, 3, 0]

Key insight: Convergence requires both CRDT merge rules AND message deduplication. Without deduplication, additive operations would double-apply.

Conflict Resolution Example

import { createEmptyGraph, applyMessage } from '@vuer-rtc/server/operations';

// Setup: cube at position [0, 0, 0]
let graph = applyMessage(createEmptyGraph(), {
  id: 'setup', sessionId: 'server', clock: { server: 1 }, lamportTime: 0, timestamp: Date.now(),
  ops: [{ key: 'cube', otype: 'node.insert', path: 'cube', value: { key: 'uuid', tag: 'Mesh', name: 'Cube', 'position': [0, 0, 0], color: '#fff' }}],
});

// Alice drags right, Bob drags up (concurrent - both use additive)
graph = applyMessage(graph, {
  id: 'alice', sessionId: 'alice', clock: { alice: 1 }, lamportTime: 1, timestamp: Date.now(),
  ops: [{ key: 'cube', otype: 'vector3.add', path: 'position', value: [5, 0, 0] }],
});
graph = applyMessage(graph, {
  id: 'bob', sessionId: 'bob', clock: { bob: 1 }, lamportTime: 2, timestamp: Date.now(),
  ops: [{ key: 'cube', otype: 'vector3.add', path: 'position', value: [0, 3, 0] }],
});

console.log(graph.nodes['cube'].position); // [5, 3, 0] - both movements applied!

// Bob sets blue (lamport 11), then Alice's earlier edit arrives (lamport 10)
graph = applyMessage(graph, {
  id: 'bob-color', sessionId: 'bob', clock: { bob: 2 }, lamportTime: 11, timestamp: Date.now(),
  ops: [{ key: 'cube', otype: 'color.set', path: 'color', value: '#0000ff' }],
});
graph = applyMessage(graph, {
  id: 'alice-color', sessionId: 'alice', clock: { alice: 2 }, lamportTime: 10, timestamp: Date.now(),
  ops: [{ key: 'cube', otype: 'color.set', path: 'color', value: '#ff0000' }],
});

console.log(graph.nodes['cube'].color); // '#0000ff' - Bob still wins (lamport 11 > 10)

Examples

See the examples/ folder for runnable examples:

npx tsx examples/01-basic-usage.ts
npx tsx examples/02-concurrent-edits.ts
npx tsx examples/03-scene-building.ts
npx tsx examples/04-conflict-resolution.ts

Project Structure

src/operations/
├── OperationTypes.ts   # Type definitions
├── dispatcher.ts       # applyMessage(), applyMessages()
├── apply/
│   ├── index.ts        # Registry exports
│   ├── types.ts        # OpMeta, ApplyFn
│   ├── number.ts       # NumberSet, NumberAdd, ...
│   ├── vector3.ts      # Vector3Set, Vector3Add, ...
│   ├── array.ts        # ArraySet, ArrayPush, ...
│   ├── node.ts         # NodeInsert, NodeRemove
│   └── ...

Lossless checkpoints and note sync verification

Server 0.4.2 and SDK 0.8.10 preserve character IDs and deleted anchors alongside plain graph text in Snapshot.textRopes. Keep this field intact when storing or forwarding snapshots. Graph stores hydrate it automatically in initialSnapshot and loadServerState; custom consumers can use encodeTextSnapshot and hydrateTextSnapshot from @vuer-ai/vuer-rtc.

A client sends { mtype: 'sync-check', requestId } on its existing socket. The server answers only that socket with { mtype: 'sync-status', requestId, status }, where status contains revision, committed clock, algorithm: 'sha256', and the content.text checksum. A null status means no verification is available. This is a read; it emits no CRDT operations and does not lock editing.

Compare text only when the response is fresh, committed clocks match, and local edits have been acknowledged. Different clocks mean catch-up; slow or missing replies are not divergence. Preserve local drafts when a proven mismatch pauses sending and require an explicit recovery choice. Upgrade API/CLI bridges as well as browser editors. String-only legacy checkpoints cannot reconstruct discarded anchors, and this update does not repair existing garbled text.