# 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:

```bash
pnpm add @vuer-ai/vuer-rtc-server
```

The package requires Node.js 18+ and the following dependencies (installed automatically):
- `@vuer-ai/vuer-rtc` - Core CRDT operations
- `@prisma/client` - Database ORM
- `ws` - WebSocket server

## Environment Configuration

Create a `.env` file in your project root:

```bash
# Required: MongoDB connection string with replica set
DATABASE_URL="mongodb://localhost:27017/vuer-rtc?replicaSet=rs0"

# Optional: Server port (default: 8080)
PORT=8080

# Optional: Redis URL (for multi-instance scaling)
REDIS_URL="redis://localhost:6379"
```

### 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:

```bash
# Clone or download docker-compose.yml from the repository
curl -O https://raw.githubusercontent.com/vuer-ai/vuer-rtc/main/docker/docker-compose.yml

# Start MongoDB and Redis
docker compose up -d

# Wait for MongoDB replica set initialization (about 30 seconds)
docker compose logs -f mongo
```

The `docker-compose.yml` configuration:
- MongoDB 7 with replica set `rs0` on 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):
```bash
DATABASE_URL="mongodb+srv://username:password@cluster.mongodb.net/vuer-rtc?retryWrites=true&w=majority"
```

**Self-Hosted Replica Set**:
```bash
# Initialize replica set on primary server
mongosh
> rs.initiate({
  _id: "rs0",
  members: [
    { _id: 0, host: "mongo1.example.com:27017" },
    { _id: 1, host: "mongo2.example.com:27017" },
    { _id: 2, host: "mongo3.example.com:27017" }
  ]
})

# Connection string
DATABASE_URL="mongodb://mongo1.example.com:27017,mongo2.example.com:27017,mongo3.example.com:27017/vuer-rtc?replicaSet=rs0"
```

## Database Schema Initialization

After configuring `DATABASE_URL`, initialize the Prisma schema:

```bash
# Generate Prisma client
pnpm exec prisma generate

# Push schema to database (creates collections)
pnpm exec prisma db push
```

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:

```bash
docker compose up -d redis
```

### Production Setup

Use a managed Redis service:

```bash
# Redis Cloud, AWS ElastiCache, etc.
REDIS_URL="redis://username:password@redis.example.com:6379"

# Redis Cluster
REDIS_URL="redis://redis-cluster.example.com:6379?cluster=true"
```

### 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

```typescript

const PORT = Number(process.env.PORT) || 8080;
const prisma = createPrismaClient();

const broker = new InMemoryBroker();
const journalService = new JournalService(prisma);

// Wire broker clocks into journal for safe compaction
journalService.setMemberClockProvider(async (docId: string) => {
  const doc = await prisma.document.findUnique({ where: { id: docId } });
  if (!doc) return [];
  const members = await broker.getMembers(doc.name);
  return Array.from(members.values())
    .filter(m => m.connected)
    .map(m => m.vectorClock);
});

journalService.startCompactionLoop();

const rtcServer = new RTCServer(broker, {
  async processMessage(roomId: string, msg: any) {
    const docId = await ensureDocument(roomId);
    return journalService.processMessage(docId, msg);
  },
  async getStateForClient(roomId: string) {
    const docId = await ensureDocument(roomId);
    return journalService.getStateForClient(docId);
  },
});

const server = createServer((req, res) => {
  res.writeHead(200).end('OK');
});

const wss = new WebSocketServer({ server });

wss.on('connection', (ws, req) => {
  const url = new URL(req.url ?? '/', `http://${req.headers.host}`);
  const match = url.pathname.match(/^\/ws\/([^/]+)$/);
  if (!match) {
    ws.close(4000, 'Invalid path');
    return;
  }
  const roomId = decodeURIComponent(match[1]);
  const sessionId = url.searchParams.get('sessionId');
  if (!sessionId) {
    ws.close(4001, 'Missing sessionId');
    return;
  }
  rtcServer.handleConnection(ws, roomId, sessionId);
});

server.listen(PORT, () => {
  console.log(`Server listening on http://localhost:${PORT}`);
});
```

### Using the Built-in Server

The package includes a complete production-ready server:

```bash
# Build and run
pnpm build
pnpm serve
```

WebSocket URL format:
```
ws://localhost:8080/ws/{roomId}?sessionId={sessionId}
```

REST API endpoints:
- `GET /api/stats` - Database statistics
- `GET /api/documents` - List all documents
- `GET /api/documents/:id` - Get document details
- `GET /api/documents/:id/journal` - Get journal batches
- `GET /api/rooms/:roomId/state` - Get room state
- `DELETE /api/rooms/:roomId` - Clear room (keeps connections)

## Production Deployment

### Docker Deployment

Create a `Dockerfile`:

```dockerfile
FROM node:18-alpine AS base
RUN corepack enable

FROM base AS deps
WORKDIR /app
COPY package.json pnpm-lock.yaml ./
RUN pnpm install --frozen-lockfile

FROM base AS builder
WORKDIR /app
COPY --from=deps /app/node_modules ./node_modules
COPY . .
RUN pnpm exec prisma generate
RUN pnpm build

FROM base AS runner
WORKDIR /app
ENV NODE_ENV=production

COPY --from=builder /app/node_modules ./node_modules
COPY --from=builder /app/dist ./dist
COPY --from=builder /app/prisma ./prisma
COPY --from=builder /app/package.json ./

EXPOSE 8080
CMD ["node", "dist/serve.js"]
```

Build and run:

```bash
docker build -t vuer-rtc-server .
docker run -p 8080:8080 \
  -e DATABASE_URL="mongodb://mongo:27017/vuer-rtc?replicaSet=rs0" \
  vuer-rtc-server
```

### Docker Compose Example

Complete production setup with MongoDB and Redis:

```yaml
version: '3.8'

services:
  server:
    build: .
    ports:
      - "8080:8080"
    environment:
      DATABASE_URL: mongodb://mongo:27017/vuer-rtc?replicaSet=rs0
      REDIS_URL: redis://redis:6379
      PORT: 8080
    depends_on:
      mongo:
        condition: service_healthy
      redis:
        condition: service_started
    restart: unless-stopped

  mongo:
    image: mongo:7
    command: ["--replSet", "rs0", "--bind_ip_all"]
    volumes:
      - mongo-data:/data/db
    healthcheck:
      test: echo "try { rs.status() } catch (err) { rs.initiate({_id:'rs0',members:[{_id:0,host:'localhost:27017'}]}) }" | mongosh --quiet
      interval: 5s
      timeout: 30s
      retries: 30

  redis:
    image: redis:7-alpine
    command: redis-server --appendonly yes
    volumes:
      - redis-data:/data

volumes:
  mongo-data:
  redis-data:
```

### 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**:
```bash
DATABASE_URL="mongodb+srv://cluster.mongodb.net/vuer-rtc"
REDIS_URL="redis://redis.cloud:6379"
PORT=8080
NODE_ENV=production
```

### Horizontal Scaling

For multi-instance deployments:

1. **Use Redis-backed broker** (when available)
2. **Configure sticky sessions** on load balancer (WebSocket affinity)
3. **Share MongoDB cluster** across all instances
4. **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:

```typescript
server.on('request', (req, res) => {
  if (req.url === '/health') {
    prisma.$queryRaw`SELECT 1`
      .then(() => res.writeHead(200).end('OK'))
      .catch(() => res.writeHead(503).end('Database unavailable'));
  }
});
```

### 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

```typescript

const messageCounter = new Counter({
  name: 'vuer_rtc_messages_total',
  help: 'Total CRDT messages processed',
});

const connectionGauge = new Gauge({
  name: 'vuer_rtc_connections',
  help: 'Current WebSocket connections',
});

rtcServer.on('message', () => messageCounter.inc());
wss.on('connection', () => connectionGauge.inc());
wss.on('close', () => connectionGauge.dec());

server.on('request', (req, res) => {
  if (req.url === '/metrics') {
    res.setHeader('Content-Type', register.contentType);
    register.metrics().then(metrics => res.end(metrics));
  }
});
```

### Logging

Configure structured logging for production:

```typescript

const logger = pino({
  level: process.env.LOG_LEVEL || 'info',
  transport: process.env.NODE_ENV === 'development'
    ? { target: 'pino-pretty' }
    : undefined,
});

rtcServer.on('message', (msg) => {
  logger.debug({ msgId: msg.id, roomId: msg.roomId }, 'Message processed');
});
```

## Performance Tuning

### Journal Compaction

Configure compaction frequency in `JournalService`:

```typescript
// Default: compact every 5 minutes
journalService.startCompactionLoop({
  intervalMs: 5 * 60 * 1000,
  minOperations: 1000, // Only compact if > 1000 operations
});
```

### Connection Limits

MongoDB and WebSocket limits:

```typescript
// MongoDB connection pool (in DATABASE_URL)
DATABASE_URL="mongodb://localhost:27017/vuer-rtc?replicaSet=rs0&maxPoolSize=50"

// WebSocket max connections (OS limits apply)
wss.setMaxListeners(10000);
```

### Memory Management

Monitor and limit room history:

```typescript
// Clear old rooms periodically
setInterval(async () => {
  const rooms = await broker.getRooms();
  for (const roomId of rooms) {
    const members = await broker.getMembers(roomId);
    if (members.size === 0) {
      await rtcServer.clearRoom(roomId);
    }
  }
}, 60 * 60 * 1000); // Every hour
```

---

## 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](/awareness.md).

## Architecture Overview

```
┌────────────────────────────────────────────────────────────────────────────┐
│                              Server                                        │
│  ┌──────────────────────────────────────────────────────────────────────┐  │
│  │                         WebSocket Handler                            │  │
│  │  ┌─────────────┐    ┌─────────────┐    ┌─────────────────────────┐   │  │
│  │  │  Receive    │───▶│  Validate   │───▶│  Apply Operations       │   │  │
│  │  │  Message    │    │  & Dedup    │    │  (dispatcher.ts)        │   │  │
│  │  └─────────────┘    └─────────────┘    └───────────┬─────────────┘   │  │
│  │                                                    │                 │  │
│  │                     ┌──────────────────────────────┼─────────────┐   │  │
│  │                     │                              ▼             │   │  │
│  │                     │  ┌─────────────────────────────────────┐   │   │  │
│  │                     │  │           SceneGraph                │   │   │  │
│  │                     │  │  ┌───────────────────────────────┐  │   │   │  │
│  │                     │  │  │ nodes: { [key]: SceneNode }   │  │   │   │  │
│  │                     │  │  │   - position, rotation, etc.  │  │   │   │  │
│  │                     │  │  │   - schema-free properties    │  │   │   │  │
│  │                     │  │  └───────────────────────────────┘  │   │   │  │
│  │                     │  └─────────────────────────────────────┘   │   │  │
│  │                     │                    ▲                       │   │  │
│  │                     │                    │ rebuild               │   │  │
│  │                     │                    │                       │   │  │
│  │                     │  ┌─────────────────┴───────────────────┐   │   │  │
│  │                     │  │        Journal (Event Log)          │   │   │  │
│  │                     │  │  ┌───────────────────────────────┐  │   │   │  │
│  │                     │  │  │ Entry { msg, deletedAt? }     │  │   │   │  │
│  │                     │  │  │ Entry { msg, deletedAt? }     │  │   │   │  │
│  │                     │  │  │ ...                           │  │   │   │  │
│  │                     │  │  └───────────────────────────────┘  │   │   │  │
│  │                     │  └─────────────────────────────────────┘   │   │  │
│  │                     │                In-Memory                   │   │  │
│  │                     └────────────────────────────────────────────┘   │  │
│  └──────────────────────────────────────────────────────────────────────┘  │
│                                        │                                   │
│            ┌───────────────────────────┼───────────────────────────┐       │
│            ▼                           ▼                           ▼       │
│  ┌─────────────────┐         ┌─────────────────┐         ┌──────────────┐  │
│  │   Broadcast     │         │   Persist to    │         │    Send      │  │
│  │   to Clients    │         │    MongoDB      │         │    Ack       │  │
│  └─────────────────┘         └─────────────────┘         └──────────────┘  │
└────────────────────────────────────────────────────────────────────────────┘
                                        │
        ┌───────────────────────────────┼───────────────────────────┐
        ▼                               ▼                           ▼
┌───────────────┐             ┌───────────────┐             ┌───────────────┐
│   Client A    │             │   Client B    │             │   Client C    │
│  (WebSocket)  │             │  (WebSocket)  │             │  (WebSocket)  │
└───────────────┘             └───────────────┘             └───────────────┘
```

## 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 |

```
┌─────────────────────────────────────────────────────────────────┐
│                         MongoDB                                 │
│  ┌─────────────┐  ┌─────────────┐  ┌────────────┐  ┌─────────┐  │
│  │  Document   │  │  Operation  │  │ JournalBatch│ │ Session │  │
│  │             │  │             │  │             │  │         │  │
│  │ currentState│  │ vectorClock │  │ operations[]│  │presence │  │
│  │ (JSON)      │  │ lamportTime │  │ startTime   │  │clockValue│ │
│  │ version     │  │ data        │  │ endTime     │  │connected│  │
│  └─────────────┘  └─────────────┘  └────────────┘  └─────────┘  │
└─────────────────────────────────────────────────────────────────┘
```

## Data Flow

```
Client                          Server                         MongoDB
  │                               │                               │
  │  CRDTMessage                  │                               │
  │  {id, ops[], timestamp}       │                               │
  │──────────────────────────────▶│                               │
  │                               │                               │
  │                               │  1. Deduplicate (processedIds)│
  │                               │  2. Apply ops to SceneGraph   │
  │                               │  3. Add to Journal            │
  │                               │                               │
  │                               │  Persist                      │
  │                               │──────────────────────────────▶│
  │                               │                               │
  │              Ack {msgId}      │                               │
  │◀──────────────────────────────│                               │
  │                               │                               │
  │                               │  Broadcast to other clients   │
  │                               │──────────────────────────────▶│ (Client B, C)
  │                               │                               │
```

## Minimal Server

The server only needs to:
1. **Receive messages** from clients
2. **Apply operations** using the dispatcher
3. **Broadcast** to other clients
4. **Send acknowledgements** back

```typescript

let state: SceneGraph = { nodes: {}, rootKey: '' };
const processedIds = new Set<string>();

function handleMessage(ws: WebSocket, msg: CRDTMessage) {
  // Idempotent - skip duplicates
  if (processedIds.has(msg.id)) {
    ws.send(JSON.stringify({ mtype: 'ack', msgId: msg.id }));
    return;
  }
  processedIds.add(msg.id);

  // Apply operations to state
  state = applyMessage(state, msg);

  // Send ack to sender
  ws.send(JSON.stringify({ mtype: 'ack', msgId: msg.id }));

  // Broadcast to other clients
  broadcast(msg, ws);
}
```

## Why No Schema Needed?

With explicit operations like `number.add`, `vector3.set`, the **operation itself specifies merge behavior**:

```typescript
// Client sends:
{
  ot: 'number.add',  // <-- This tells the server HOW to apply
  key: 'player',
  path: 'score',
  value: 10
}
```

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.

```typescript

const clockManager = new VectorClockManager();

// Compare two clocks
const comparison = clockManager.compare(clock1, clock2);
// 1 = clock1 > clock2 (happened after)
// -1 = clock1 < clock2 (happened before)
// 0 = concurrent (conflict!)

if (comparison === 0) {
  // Concurrent operations - both valid, apply in lamportTime order
}
```

## Handling Undo/Redo

Meta operations (`meta.undo`, `meta.redo`) work by referencing a target message:

```typescript
function handleMetaOps(msg: CRDTMessage, journal: JournalEntry[]) {
  for (const op of msg.ops) {
    if (op.ot === 'meta.undo') {
      const target = journal.find(e => e.msg.id === op.targetMsgId);
      if (target) target.deletedAt = msg.timestamp;
    } else if (op.ot === 'meta.redo') {
      const target = journal.find(e => e.msg.id === op.targetMsgId);
      if (target) delete target.deletedAt;
    }
  }
}
```

## Complete Example

```typescript

interface JournalEntry {
  msg: CRDTMessage;
  deletedAt?: number;
}

class RTCServer {
  private state: SceneGraph = createEmptyGraph();
  private journal: JournalEntry[] = [];
  private processedIds = new Set<string>();
  private clients = new Set();

  handleConnection(ws: WebSocket) {
    this.clients.add(ws);

    // Send current state to new client
    ws.send(JSON.stringify({
      mtype: 'init',
      state: this.state,
      journal: this.journal.map(e => e.msg),
    }));

    ws.on('message', (data) => {
      const msg = JSON.parse(data.toString()) as CRDTMessage;
      this.handleMessage(ws, msg);
    });

    ws.on('close', () => this.clients.delete(ws));
  }

  private handleMessage(sender: WebSocket, msg: CRDTMessage) {
    // Idempotent
    if (this.processedIds.has(msg.id)) {
      sender.send(JSON.stringify({ mtype: 'ack', msgId: msg.id }));
      return;
    }
    this.processedIds.add(msg.id);

    // Handle meta ops (undo/redo)
    for (const op of msg.ops) {
      if (op.ot === 'meta.undo') {
        const target = this.journal.find(e => e.msg.id === (op as any).targetMsgId);
        if (target) target.deletedAt = msg.timestamp;
      } else if (op.ot === 'meta.redo') {
        const target = this.journal.find(e => e.msg.id === (op as any).targetMsgId);
        if (target) delete target.deletedAt;
      }
    }

    // Add to journal
    this.journal.push({ msg });

    // Rebuild state from journal (skip deleted entries)
    this.rebuildState();

    // Ack sender
    sender.send(JSON.stringify({ mtype: 'ack', msgId: msg.id }));

    // Broadcast to others
    for (const client of this.clients) {
      if (client !== sender && client.readyState === WebSocket.OPEN) {
        client.send(JSON.stringify({ mtype: 'message', msg }));
      }
    }
  }

  private rebuildState() {
    let state = createEmptyGraph();
    for (const entry of this.journal) {
      if (entry.deletedAt) continue;
      const realOps = entry.msg.ops.filter(op => !op.ot.startsWith('meta.'));
      if (realOps.length > 0) {
        state = applyMessage(state, { ...entry.msg, ops: realOps });
      }
    }
    this.state = state;
  }
}
```

---

## Source Files

### dispatcher.ts (Operation Application)

### VectorClock.ts