Server Setup and Implementation
The @vuer-ai/vuer-rtc-server package provides production-ready server infrastructure for real-time
collaborative applications using CRDT operations. It includes MongoDB persistence, WebSocket transport,
and comprehensive state management.
Installation
Install the server package in your Node.js project:
The package requires Node.js 18+ and the following dependencies (installed automatically):
@vuer-ai/vuer-rtc- Core CRDT operations@prisma/client- Database ORMws- WebSocket server
Environment Configuration
Create a .env file in your project root:
Required Environment Variables
| Variable | Description | Example |
|---|---|---|
DATABASE_URL | MongoDB connection string with replica set | mongodb://localhost:27017/vuer-rtc?replicaSet=rs0 |
Optional Environment Variables
| Variable | Default | Description |
|---|---|---|
PORT | 8080 | HTTP/WebSocket server port |
REDIS_URL | - | Redis connection URL (required for multi-instance deployments) |
MongoDB Replica Set Setup
The server requires MongoDB with replica set support for transactions and change streams.
Development (Local)
Use Docker Compose for quick local setup:
The docker-compose.yml configuration:
- MongoDB 7 with replica set
rs0on port 27017 - Redis 7 on port 6379
- Automatic replica set initialization via healthcheck
Production Setup
For production deployments, use a managed MongoDB service:
MongoDB Atlas (Recommended):
Self-Hosted Replica Set:
Database Schema Initialization
After configuring DATABASE_URL, initialize the Prisma schema:
This creates 4 MongoDB collections:
| Collection | Purpose |
|---|---|
Document | Scene metadata and current state snapshots |
Operation | Individual CRDT operations with vector clocks |
JournalBatch | Batched write-ahead log (33ms batching) |
Session | Client connections, presence, and clock state |
Redis Configuration (Optional)
Redis is required for horizontal scaling across multiple server instances. The broker uses Redis pub/sub to synchronize room state and member presence.
Development (Local)
Redis is included in the Docker Compose setup:
Production Setup
Use a managed Redis service:
Scaling Considerations
- Single Instance: No Redis required, uses
InMemoryBroker - Multi-Instance: Redis required for
RedisBroker(not yet implemented) - Load Balancing: Use sticky sessions or implement Redis-backed broker
Starting the Server
Basic Usage
Using the Built-in Server
The package includes a complete production-ready server:
WebSocket URL format:
REST API endpoints:
GET /api/stats- Database statisticsGET /api/documents- List all documentsGET /api/documents/:id- Get document detailsGET /api/documents/:id/journal- Get journal batchesGET /api/rooms/:roomId/state- Get room stateDELETE /api/rooms/:roomId- Clear room (keeps connections)
Production Deployment
Docker Deployment
Create a Dockerfile:
Build and run:
Docker Compose Example
Complete production setup with MongoDB and Redis:
Cloud Deployment
Recommended Stack:
- Application: AWS ECS, Google Cloud Run, or Railway
- Database: MongoDB Atlas (M10+ with replica set)
- Cache: Redis Cloud or AWS ElastiCache
- Load Balancer: AWS ALB with WebSocket support
Environment Variables:
Horizontal Scaling
For multi-instance deployments:
- Use Redis-backed broker (when available)
- Configure sticky sessions on load balancer (WebSocket affinity)
- Share MongoDB cluster across all instances
- Monitor connection limits on MongoDB and Redis
Current limitation: InMemoryBroker only supports single-instance deployment.
RedisBroker implementation is planned for horizontal scaling.
Health Checks and Monitoring
Health Check Endpoint
Add a health check route:
Monitoring Metrics
Track these key metrics:
| Metric | Endpoint/Method | Description |
|---|---|---|
| Active connections | wss.clients.size | Current WebSocket connections |
| Room count | broker.getRooms() | Number of active rooms |
| Database size | GET /api/stats/sizes | Storage usage per collection |
| Operation throughput | Custom counter | Messages/second processed |
| Journal batch size | Monitor JournalBatch.operations.length | Average operations per batch |
Example Prometheus Metrics
Logging
Configure structured logging for production:
Performance Tuning
Journal Compaction
Configure compaction frequency in JournalService:
Connection Limits
MongoDB and WebSocket limits:
Memory Management
Monitor and limit room history:
Architecture Overview
vuer-rtc uses explicit operations where each message specifies its merge behavior via ot.
This simplifies server implementation significantly -- the server just applies operations and
broadcasts.
The server also relays ephemeral awareness (presence and cursors) per room: it keeps an in-memory state map, hands newcomers the current roster on connect, and broadcasts a leave on disconnect. Awareness is never journaled or persisted — see Awareness.
Architecture Overview
Database Schema
The server persists to 4 MongoDB collections:
| Collection | Purpose |
|---|---|
| Document | Scene metadata + currentState (schema-free JSON snapshot) |
| Operation | Individual CRDT operations with vector clocks |
| JournalBatch | Batched write-ahead log (33ms batching) |
| Session | Client connections, presence, and clock state |
Data Flow
Minimal Server
The server only needs to:
- Receive messages from clients
- Apply operations using the dispatcher
- Broadcast to other clients
- Send acknowledgements back
Why No Schema Needed?
With explicit operations like number.add, vector3.set, the operation itself specifies merge behavior:
The server doesn't need a schema to know score is additive - the ot: 'number.add' already says so!
Vector Clocks (Optional)
Vector clocks help detect concurrent operations for advanced conflict handling. For most use cases, Lamport timestamps (already in each message) are sufficient.
Handling Undo/Redo
Meta operations (meta.undo, meta.redo) work by referencing a target message:
Complete Example
Source Files
dispatcher.ts (Operation Application)
/**
* Operation Dispatcher
*
* Applies CRDTMessage operations to a SceneGraph.
* Each operation is dispatched to its corresponding "apply" function.
*
* Uses shallow cloning for immutability without immer overhead.
*/
import type { SceneGraph, CRDTMessage, Operation } from './OperationTypes.js';
import type { OpMeta } from './apply/types.js';
import * as registry from './apply/index.js';
/**
* Handler map: ot -> apply function
*/
const handlers: Record<string, (graph: SceneGraph, op: Operation, meta: OpMeta) => void> = {
// Number operations
'number.set': registry.NumberSet as (graph: SceneGraph, op: Operation, meta: OpMeta) => void,
'number.add': registry.NumberAdd as (graph: SceneGraph, op: Operation, meta: OpMeta) => void,
'number.multiply': registry.NumberMultiply as (graph: SceneGraph, op: Operation, meta: OpMeta) => void,
'number.min': registry.NumberMin as (graph: SceneGraph, op: Operation, meta: OpMeta) => void,
'number.max': registry.NumberMax as (graph: SceneGraph, op: Operation, meta: OpMeta) => void,
// String operations
'string.set': registry.StringSet as (graph: SceneGraph, op: Operation, meta: OpMeta) => void,
'string.concat': registry.StringConcat as (graph: SceneGraph, op: Operation, meta: OpMeta) => void,
// Text CRDT operations (character-level collaborative editing)
'text.init': registry.TextInit as (graph: SceneGraph, op: Operation, meta: OpMeta) => void,
'text.insert': registry.TextInsert as (graph: SceneGraph, op: Operation, meta: OpMeta) => void,
'text.delete': registry.TextDelete as (graph: SceneGraph, op: Operation, meta: OpMeta) => void,
'text.replace': registry.TextReplace as (graph: SceneGraph, op: Operation, meta: OpMeta) => void,
// Boolean operations
'boolean.set': registry.BooleanSet as (graph: SceneGraph, op: Operation, meta: OpMeta) => void,
'boolean.or': registry.BooleanOr as (graph: SceneGraph, op: Operation, meta: OpMeta) => void,
'boolean.and': registry.BooleanAnd as (graph: SceneGraph, op: Operation, meta: OpMeta) => void,
// Vector3 operations
'vector3.set': registry.Vector3Set as (graph: SceneGraph, op: Operation, meta: OpMeta) => void,
'vector3.add': registry.Vector3Add as (graph: SceneGraph, op: Operation, meta: OpMeta) => void,
'vector3.multiply': registry.Vector3Multiply as (graph: SceneGraph, op: Operation, meta: OpMeta) => void,
'vector3.applyEuler': registry.Vector3ApplyEuler as (graph: SceneGraph, op: Operation, meta: OpMeta) => void,
'vector3.applyQuaternion': registry.Vector3ApplyQuaternion as (graph: SceneGraph, op: Operation, meta: OpMeta) => void,
// Euler operations
'euler.set': registry.EulerSet as (graph: SceneGraph, op: Operation, meta: OpMeta) => void,
'euler.add': registry.EulerAdd as (graph: SceneGraph, op: Operation, meta: OpMeta) => void,
// Quaternion operations
'quaternion.set': registry.QuaternionSet as (graph: SceneGraph, op: Operation, meta: OpMeta) => void,
'quaternion.multiply': registry.QuaternionMultiply as (graph: SceneGraph, op: Operation, meta: OpMeta) => void,
// Color operations
'color.set': registry.ColorSet as (graph: SceneGraph, op: Operation, meta: OpMeta) => void,
'color.blend': registry.ColorBlend as (graph: SceneGraph, op: Operation, meta: OpMeta) => void,
// Array operations
'array.set': registry.ArraySet as (graph: SceneGraph, op: Operation, meta: OpMeta) => void,
'array.push': registry.ArrayPush as (graph: SceneGraph, op: Operation, meta: OpMeta) => void,
'array.remove': registry.ArrayRemove as (graph: SceneGraph, op: Operation, meta: OpMeta) => void,
'array.union': registry.ArrayUnion as (graph: SceneGraph, op: Operation, meta: OpMeta) => void,
// Object operations
'object.set': registry.ObjectSet as (graph: SceneGraph, op: Operation, meta: OpMeta) => void,
'object.merge': registry.ObjectMerge as (graph: SceneGraph, op: Operation, meta: OpMeta) => void,
// Node operations
'node.insert': registry.NodeInsert as (graph: SceneGraph, op: Operation, meta: OpMeta) => void,
'node.remove': registry.NodeRemove as (graph: SceneGraph, op: Operation, meta: OpMeta) => void,
'node.move': registry.NodeMove as (graph: SceneGraph, op: Operation, meta: OpMeta) => void,
};
/**
* Apply a single operation to a SceneGraph (mutates in place)
*/
export function applyOperation(
graph: SceneGraph,
op: Operation,
meta: OpMeta
): void {
const handler = handlers[op.ot];
if (handler) {
handler(graph, op, meta);
} else {
console.warn(`Unknown ot: ${op.ot}`);
}
}
/**
* Compare operations for sorting by Lamport timestamp.
* CRDT invariant: operations must be applied in causal order (seq → ts → id).
*
* This ensures that:
* 1. Operations with lower Lamport clocks are applied first
* 2. If Lamport clocks are equal, wall-clock time breaks the tie
* 3. If both are equal, lexicographic ID order ensures determinism
*
* @param a - First operation
* @param b - Second operation
* @returns Negative if a < b, positive if a > b, 0 if equal
*/
function compareOperations(a: Operation, b: Operation): number {
const aOp = a as any;
const bOp = b as any;
// Compare by Lamport clock (seq)
if (aOp.seq !== undefined && bOp.seq !== undefined) {
if (aOp.seq !== bOp.seq) return aOp.seq - bOp.seq;
}
// If equal or missing, compare by wall-clock time (ts)
if (aOp.ts !== undefined && bOp.ts !== undefined) {
if (aOp.ts !== bOp.ts) return aOp.ts - bOp.ts;
}
// Final fallback: compare by ID (lexicographic)
if (aOp.id && bOp.id) {
return String(aOp.id).localeCompare(String(bOp.id));
}
// If no metadata, preserve original order
return 0;
}
/**
* Shallow clone the graph and modified nodes
*/
function shallowCloneGraph(graph: SceneGraph): SceneGraph {
return {
...graph,
nodes: { ...graph.nodes },
};
}
/**
* Apply a CRDTMessage to a SceneGraph (immutable)
*
* Creates a shallow clone of the graph before applying operations.
*
* @param graph - Current scene graph state
* @param msg - CRDT message containing operations
* @returns New scene graph state with operations applied
*/
export function applyMessage(graph: SceneGraph, msg: CRDTMessage): SceneGraph {
// Shallow clone for immutability
const newGraph = shallowCloneGraph(graph);
const meta: OpMeta = {
client: msg.client,
clock: msg.clock,
lt: msg.lt,
ts: msg.ts,
};
// Sort operations by Lamport timestamp to ensure causal order
// This is critical for CRDT correctness, especially for text operations
// where replace ops may reference IDs from insert ops
const sortedOps = [...msg.ops].sort(compareOperations);
for (const op of sortedOps) {
// Determine all node keys that this operation will mutate
const keysToClone: string[] = [];
if (op.key && newGraph.nodes[op.key]) {
keysToClone.push(op.key);
}
if (op.ot === 'node.move') {
const { nodeKey, newParent } = (op as any).value;
if (nodeKey && newGraph.nodes[nodeKey]) keysToClone.push(nodeKey);
if (newParent && newGraph.nodes[newParent]) keysToClone.push(newParent);
}
if (op.ot === 'node.remove') {
const nodeKey = (op as any).value;
if (typeof nodeKey === 'string' && newGraph.nodes[nodeKey]) keysToClone.push(nodeKey);
}
for (const k of new Set(keysToClone)) {
const orig = newGraph.nodes[k];
newGraph.nodes[k] = {
...orig,
children: orig.children ? [...orig.children] : [],
};
}
applyOperation(newGraph, op, meta);
}
return newGraph;
}
/**
* Apply a CRDTMessage to a SceneGraph (mutable)
*
* Mutates the graph directly for better performance.
*
* @param graph - Scene graph state to mutate
* @param msg - CRDT message containing operations
*/
export function applyMessageMut(graph: SceneGraph, msg: CRDTMessage): void {
const meta: OpMeta = {
client: msg.client,
clock: msg.clock,
lt: msg.lt,
ts: msg.ts,
};
// Sort operations by Lamport timestamp to ensure causal order
const sortedOps = [...msg.ops].sort(compareOperations);
for (const op of sortedOps) {
applyOperation(graph, op, meta);
}
}
/**
* Apply multiple CRDTMessages to a SceneGraph (immutable)
*
* @param graph - Current scene graph state
* @param messages - Array of CRDT messages to apply
* @returns New scene graph state with all operations applied
*/
export function applyMessages(graph: SceneGraph, messages: CRDTMessage[]): SceneGraph {
let current = graph;
for (const msg of messages) {
current = applyMessage(current, msg);
}
return current;
}
/**
* Apply multiple CRDTMessages to a SceneGraph (mutable)
*
* @param graph - Scene graph state to mutate
* @param messages - Array of CRDT messages to apply
*/
export function applyMessagesMut(graph: SceneGraph, messages: CRDTMessage[]): void {
for (const msg of messages) {
applyMessageMut(graph, msg);
}
}
/**
* Create an empty scene graph
*/
export function createEmptyGraph(): SceneGraph {
return {
nodes: {},
rootKey: '',
lww: {},
tombstones: {},
};
}VectorClock.ts
/**
* VectorClock - CRDT-inspired vector clock implementation
*
* Vector clocks provide causal ordering of operations in a distributed system.
* Each session maintains a counter, and clocks are compared to detect:
* - Causal ordering (A happened before B)
* - Concurrent operations (A and B are independent)
*/
export type VectorClock = Record<string, number>;
export class VectorClockManager {
/**
* Create a new vector clock for a client
* Initializes the client's counter to 0
*/
create(client: string): VectorClock {
return { [client]: 0 };
}
/**
* Increment the counter for a client
* Returns a new clock (immutable)
*/
increment(clock: VectorClock, client: string): VectorClock {
const currentValue = clock[client] || 0;
return {
...clock,
[client]: currentValue + 1,
};
}
/**
* Merge two vector clocks
* Takes the maximum value for each client
* Used when receiving remote operations
*/
merge(clock1: VectorClock, clock2: VectorClock): VectorClock {
const merged: VectorClock = { ...clock1 };
Object.entries(clock2).forEach(([client, count]) => {
merged[client] = Math.max(merged[client] || 0, count);
});
return merged;
}
/**
* Compare two vector clocks
*
* Returns:
* 1 if clock1 > clock2 (clock1 causally after clock2)
* -1 if clock1 < clock2 (clock1 causally before clock2)
* 0 if concurrent (neither causally precedes the other)
*/
compare(clock1: VectorClock, clock2: VectorClock): number {
const allSessionIds = new Set([
...Object.keys(clock1),
...Object.keys(clock2),
]);
let clock1Greater = false;
let clock2Greater = false;
allSessionIds.forEach((client) => {
const val1 = clock1[client] || 0;
const val2 = clock2[client] || 0;
if (val1 > val2) {
clock1Greater = true;
}
if (val2 > val1) {
clock2Greater = true;
}
});
// If clock1 is greater in all dimensions, it causally follows clock2
if (clock1Greater && !clock2Greater) {
return 1;
}
// If clock2 is greater in all dimensions, it causally follows clock1
if (clock2Greater && !clock1Greater) {
return -1;
}
// Otherwise, they are concurrent (or identical)
return 0;
}
/**
* Check if two operations are concurrent
* (neither causally precedes the other)
*/
areConcurrent(clock1: VectorClock, clock2: VectorClock): boolean {
return this.compare(clock1, clock2) === 0;
}
}