RERUM Sockets

See #113

Overview

This playbook provides step-by-step instructions for adding WebSocket-based real-time collaboration to the RERUM API. The implementation enables multiple users to work on the same JSON documents simultaneously with Operational Transformation (OT) for conflict resolution.

Architecture Decisions

• WebSocket Library: Socket.IO (mature, handles reconnection, fallback support)
• Conflict Resolution: Operational Transformation (OT) - Google Docs style
• Update Granularity: Document-level (broadcast full document on changes)
• Cluster Support: Redis adapter for PM2 cluster mode
• Authentication: Auth0 Bearer token validation

---

Table of Contents

1. Prerequisites
2. Redis Installation on RHEL
3. Package Dependencies
4. Socket.IO Server Setup
5. PM2 Configuration for Sticky Sessions
6. WebSocket Authentication
7. Document Room Management
8. Real-Time Change Broadcasting
9. Operational Transformation Implementation
10. Integration with Existing Endpoints
11. Client-Side Implementation Guide
12. Testing Strategy
13. Deployment Checklist
14. Troubleshooting

---

1. Prerequisites

Required Infrastructure

• RHEL VM with Node.js 24 LTS
• PM2 process manager (already installed)
• MongoDB (already configured)
• Auth0 account (already configured)
• Ports 80/443 open (already configured)

New Requirements

• Redis server (installation instructions below)
• Additional npm packages (listed in Section 3)

Estimated Implementation Time

• Redis setup: 30 minutes
• Socket.IO integration: 2-4 hours
• OT implementation: 4-8 hours
• Testing & debugging: 2-4 hours
• Total: 1-2 days

---

2. Redis Installation on RHEL

Step 2.1: Install Redis

# For RHEL 8/9 with dnf
sudo dnf install redis -y

# OR for older RHEL versions with yum
sudo yum install epel-release -y
sudo yum install redis -y

Step 2.2: Configure Redis

# Edit Redis configuration
sudo vi /etc/redis/redis.conf

Make these changes in redis.conf:

# Bind to localhost only (secure for same-server access)
bind 127.0.0.1

# Set a password (IMPORTANT for security)
requirepass YOUR_SECURE_REDIS_PASSWORD

# Enable persistence (optional but recommended)
appendonly yes

# Set max memory (adjust based on your VM resources)
maxmemory 256mb
maxmemory-policy allkeys-lru

Step 2.3: Start and Enable Redis

# Start Redis
sudo systemctl start redis

# Enable Redis to start on boot
sudo systemctl enable redis

# Verify Redis is running
sudo systemctl status redis

# Test Redis connection
redis-cli -a YOUR_SECURE_REDIS_PASSWORD ping
# Should return: PONG

Step 2.4: Add Redis Environment Variables

Add to your .env file:

# Redis Configuration
REDIS_HOST=127.0.0.1
REDIS_PORT=6379
REDIS_PASSWORD=YOUR_SECURE_REDIS_PASSWORD

---

3. Package Dependencies

Step 3.1: Install Required Packages

cd /path/to/rerum_server_nodejs

# Socket.IO server
npm install socket.io@4.7.5

# Redis adapter for Socket.IO (cluster mode support)
npm install @socket.io/redis-adapter@8.3.0

# Redis client
npm install redis@4.6.14

# JSON patch for operational transformation
npm install fast-json-patch@3.1.1

# Optional: For more sophisticated OT
npm install ot-json1@1.0.2

Step 3.2: Update package.json

Your package.json dependencies should now include:

{
  \"dependencies\": {
    \"socket.io\": \"^4.7.5\",
    \"@socket.io/redis-adapter\": \"^8.3.0\",
    \"redis\": \"^4.6.14\",
    \"fast-json-patch\": \"^3.1.1\",
    \"ot-json1\": \"^1.0.2\"
  }
}

---

4. Socket.IO Server Setup

Step 4.1: Create WebSocket Module

Create a new file websocket/index.js:

#!/usr/bin/env node

/**
 * WebSocket server setup for RERUM real-time collaboration
 * Uses Socket.IO with Redis adapter for PM2 cluster mode support
 */

import { Server } from 'socket.io'
import { createAdapter } from '@socket.io/redis-adapter'
import { createClient } from 'redis'
import dotenv from 'dotenv'
dotenv.config()

let io = null
let pubClient = null
let subClient = null

/**
 * Initialize Socket.IO server with Redis adapter
 * @param {http.Server} httpServer - The HTTP server instance
 * @returns {Server} Socket.IO server instance
 */
async function initializeWebSocket(httpServer) {
    // Create Socket.IO server
    io = new Server(httpServer, {
        cors: {
            origin: \"*\",
            methods: [\"GET\", \"POST\", \"PUT\", \"PATCH\", \"DELETE\"],
            allowedHeaders: [\"Authorization\", \"Content-Type\"],
            credentials: true
        },
        // Connection settings
        pingTimeout: 60000,
        pingInterval: 25000,
        // Transport settings
        transports: ['websocket', 'polling'],
        // Allow upgrades from polling to websocket
        allowUpgrades: true
    })

    // Set up Redis adapter for cluster mode
    if (process.env.REDIS_HOST) {
        try {
            pubClient = createClient({
                socket: {
                    host: process.env.REDIS_HOST,
                    port: parseInt(process.env.REDIS_PORT) || 6379
                },
                password: process.env.REDIS_PASSWORD
            })

            subClient = pubClient.duplicate()

            await Promise.all([pubClient.connect(), subClient.connect()])

            io.adapter(createAdapter(pubClient, subClient))
            console.log('Socket.IO Redis adapter connected successfully')
        } catch (error) {
            console.error('Failed to connect Redis adapter:', error)
            console.log('Falling back to in-memory adapter (single process mode)')
        }
    } else {
        console.log('Redis not configured - using in-memory adapter (single process only)')
    }

    // Set up connection handling
    setupConnectionHandlers(io)

    return io
}

/**
 * Set up Socket.IO connection handlers
 * @param {Server} io - Socket.IO server instance
 */
function setupConnectionHandlers(io) {
    io.on('connection', (socket) => {
        console.log(`Client connected: ${socket.id}`)

        // Document room management
        socket.on('document:join', (documentId) => {
            socket.join(`doc:${documentId}`)
            console.log(`Client ${socket.id} joined document: ${documentId}`)
            
            // Notify others in the room
            socket.to(`doc:${documentId}`).emit('document:user-joined', {
                socketId: socket.id,
                documentId: documentId,
                timestamp: new Date().toISOString()
            })
        })

        socket.on('document:leave', (documentId) => {
            socket.leave(`doc:${documentId}`)
            console.log(`Client ${socket.id} left document: ${documentId}`)
            
            // Notify others in the room
            socket.to(`doc:${documentId}`).emit('document:user-left', {
                socketId: socket.id,
                documentId: documentId,
                timestamp: new Date().toISOString()
            })
        })

        // Handle client-initiated changes (for OT)
        socket.on('document:local-change', (data) => {
            // This will be handled by the OT system
            // See Section 9 for OT implementation
        })

        socket.on('disconnect', (reason) => {
            console.log(`Client disconnected: ${socket.id}, reason: ${reason}`)
        })

        socket.on('error', (error) => {
            console.error(`Socket error for ${socket.id}:`, error)
        })
    })
}

/**
 * Broadcast document change to all clients in a document room
 * @param {string} documentId - The document ID
 * @param {Object} document - The updated document
 * @param {string} operation - The operation type (create, update, overwrite, etc.)
 * @param {string} excludeSocketId - Socket ID to exclude from broadcast (optional)
 */
function broadcastDocumentChange(documentId, document, operation, excludeSocketId = null) {
    if (!io) {
        console.warn('Socket.IO not initialized - cannot broadcast')
        return
    }

    const payload = {
        documentId: documentId,
        document: document,
        operation: operation,
        timestamp: new Date().toISOString(),
        version: document.__rerum?.isOverwritten || document.__rerum?.createdAt
    }

    if (excludeSocketId) {
        // Broadcast to all in room except the sender
        io.to(`doc:${documentId}`).except(excludeSocketId).emit('document:change', payload)
    } else {
        // Broadcast to all in room
        io.to(`doc:${documentId}`).emit('document:change', payload)
    }

    console.log(`Broadcasted ${operation} for document ${documentId}`)
}

/**
 * Broadcast conflict notification
 * @param {string} documentId - The document ID
 * @param {Object} currentVersion - The current version of the document
 * @param {Object} attemptedChange - The change that caused the conflict
 * @param {string} targetSocketId - Socket ID to send conflict notification to
 */
function broadcastConflict(documentId, currentVersion, attemptedChange, targetSocketId) {
    if (!io) return

    const payload = {
        documentId: documentId,
        currentVersion: currentVersion,
        attemptedChange: attemptedChange,
        timestamp: new Date().toISOString(),
        message: 'Document was modified by another user. Please review and retry.'
    }

    if (targetSocketId) {
        io.to(targetSocketId).emit('document:conflict', payload)
    } else {
        io.to(`doc:${documentId}`).emit('document:conflict', payload)
    }
}

/**
 * Get Socket.IO server instance
 * @returns {Server|null} Socket.IO server instance
 */
function getIO() {
    return io
}

/**
 * Graceful shutdown
 */
async function shutdown() {
    if (io) {
        io.close()
    }
    if (pubClient) {
        await pubClient.quit()
    }
    if (subClient) {
        await subClient.quit()
    }
}

export {
    initializeWebSocket,
    broadcastDocumentChange,
    broadcastConflict,
    getIO,
    shutdown
}

Step 4.2: Modify bin/rerum_v1.js

Update the server startup file to initialize WebSocket:

#!/usr/bin/env node

/**
 * Module dependencies.
 */

import app from '../app.js'
import debug from 'debug'
debug('rerum_server_nodejs:server')
import http from \"http\"
import dotenv from \"dotenv\"
dotenv.config()

// Import WebSocket module
import { initializeWebSocket, shutdown } from '../websocket/index.js'

/**
 * Get port from environment and store in Express.
 */

const port = process.env.PORT ?? 3001
app.set('port', port)

/**
 * Create HTTP server.
 */

const server = http.createServer(app)

/**
 * Initialize WebSocket server
 */
initializeWebSocket(server).then(() => {
    console.log('WebSocket server initialized')
}).catch(err => {
    console.error('Failed to initialize WebSocket:', err)
})

/**
 * Listen on provided port, on all network interfaces.
 */

server.listen(port)
server.on('error', onError)
server.on('listening', onListening)

/**
 * Control the keep alive header
 */ 
server.keepAliveTimeout = 8 * 1000
server.headersTimeout = 8.5 * 1000

/**
 * Graceful shutdown handling
 */
process.on('SIGTERM', async () => {
    console.log('SIGTERM received, shutting down gracefully')
    await shutdown()
    server.close(() => {
        console.log('Server closed')
        process.exit(0)
    })
})

process.on('SIGINT', async () => {
    console.log('SIGINT received, shutting down gracefully')
    await shutdown()
    server.close(() => {
        console.log('Server closed')
        process.exit(0)
    })
})

/**
 * Event listener for HTTP server \"error\" event.
 */

function onError(error) {
    if (error.syscall !== 'listen') {
        throw error
    }
    const bind = `Port ${port}`
    switch (error.code) {
        case 'EACCES':
            console.error(bind + ' requires elevated privileges')
            process.exit(1)
        case 'EADDRINUSE':
            console.error(bind + ' is already in use')
            process.exit(1)
        default:
            throw error
    }
}

/**
 * Event listener for HTTP server \"listening\" event.
 */

function onListening() {
    console.log(\"LISTENING ON \" + port)
    console.log(\"WebSocket server available at ws://localhost:\" + port)
    const addr = server.address()
    const bind = typeof addr === 'string'
        ? 'pipe ' + addr
        : 'port ' + addr.port
    debug('Listening on ' + bind)
}

---

5. PM2 Configuration for Sticky Sessions

Step 5.1: Create PM2 Ecosystem File

Create ecosystem.config.cjs in the project root:

module.exports = {
    apps: [{
        name: 'rerum-api',
        script: './bin/rerum_v1.js',
        
        // Cluster mode settings
        instances: 'max', // Or specify number: 4
        exec_mode: 'cluster',
        
        // Environment
        env: {
            NODE_ENV: 'production',
            PORT: 3001
        },
        
        // Logging
        log_date_format: 'YYYY-MM-DD HH:mm:ss Z',
        error_file: './logs/error.log',
        out_file: './logs/out.log',
        merge_logs: true,
        
        // Restart settings
        max_memory_restart: '1G',
        restart_delay: 4000,
        
        // Watch settings (disable in production)
        watch: false,
        
        // Graceful shutdown
        kill_timeout: 5000,
        listen_timeout: 10000,
        
        // Instance variance for port binding
        increment_var: 'PORT',
        
        // Node.js flags
        node_args: '--max-old-space-size=1024'
    }]
}

Step 5.2: Configure Nginx for Sticky Sessions (if using Nginx as reverse proxy)

If you're using Nginx in front of PM2, add sticky session support:

# /etc/nginx/conf.d/rerum.conf

upstream rerum_backend {
    ip_hash;  # Enables sticky sessions based on client IP
    
    server 127.0.0.1:3001;
    server 127.0.0.1:3002;
    server 127.0.0.1:3003;
    server 127.0.0.1:3004;
    
    keepalive 64;
}

server {
    listen 80;
    listen 443 ssl;
    server_name store.rerum.io;
    
    # SSL configuration
    ssl_certificate /path/to/cert.pem;
    ssl_certificate_key /path/to/key.pem;
    
    # WebSocket support
    location /socket.io/ {
        proxy_pass http://rerum_backend;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection \"upgrade\";
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
        
        # WebSocket specific timeouts
        proxy_connect_timeout 7d;
        proxy_send_timeout 7d;
        proxy_read_timeout 7d;
    }
    
    # Regular HTTP requests
    location / {
        proxy_pass http://rerum_backend;
        proxy_http_version 1.1;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
        proxy_set_header Connection \"\";
        
        # Keep-alive
        proxy_connect_timeout 60s;
        proxy_send_timeout 60s;
        proxy_read_timeout 60s;
    }
}

Step 5.3: Alternative - PM2 with @socket.io/sticky

If you prefer not to use Nginx sticky sessions, you can use PM2's sticky cluster module:

npm install @socket.io/sticky
npm install @socket.io/cluster-adapter

Then modify bin/rerum_v1.js for sticky cluster:

// Alternative approach using @socket.io/sticky
import cluster from 'cluster'
import http from 'http'
import { setupMaster, setupWorker } from '@socket.io/sticky'
import { createAdapter, setupPrimary } from '@socket.io/cluster-adapter'

if (cluster.isPrimary) {
    console.log(`Primary ${process.pid} is running`)
    
    const httpServer = http.createServer()
    setupMaster(httpServer, {
        loadBalancingMethod: 'least-connection'
    })
    setupPrimary()
    
    httpServer.listen(3001)
    
    for (let i = 0; i < numCPUs; i++) {
        cluster.fork()
    }
} else {
    // Worker process
    const httpServer = http.createServer(app)
    const io = new Server(httpServer)
    
    io.adapter(createAdapter())
    setupWorker(io)
    
    // ... rest of setup
}

---

6. WebSocket Authentication

Step 6.1: Create Authentication Middleware

Create websocket/auth.js:

#!/usr/bin/env node

/**
 * WebSocket authentication middleware
 * Validates Auth0 Bearer tokens for WebSocket connections
 */

import dotenv from 'dotenv'
dotenv.config()

// JWT verification (you may need to adjust based on your Auth0 setup)
import { auth } from 'express-oauth2-jwt-bearer'

/**
 * Validate Auth0 token from WebSocket handshake
 * @param {Object} socket - Socket.IO socket instance
 * @param {Function} next - Next middleware function
 */
async function authenticateSocket(socket, next) {
    try {
        const token = extractToken(socket)
        
        if (!token) {
            // Allow read-only connections without auth
            // Or reject: return next(new Error('Authentication required'))
            socket.data.authenticated = false
            socket.data.agent = null
            console.log(`Unauthenticated connection: ${socket.id}`)
            return next()
        }

        // Verify token with Auth0
        const decoded = await verifyAuth0Token(token)
        
        if (decoded) {
            socket.data.authenticated = true
            socket.data.agent = decoded.sub || decoded.azp
            socket.data.token = token
            console.log(`Authenticated connection: ${socket.id}, agent: ${socket.data.agent}`)
            return next()
        } else {
            return next(new Error('Invalid token'))
        }
    } catch (error) {
        console.error('Socket authentication error:', error)
        return next(new Error('Authentication failed'))
    }
}

/**
 * Extract Bearer token from socket handshake
 * @param {Object} socket - Socket.IO socket instance
 * @returns {string|null} Token or null
 */
function extractToken(socket) {
    // Try Authorization header
    const authHeader = socket.handshake.headers.authorization
    if (authHeader && authHeader.startsWith('Bearer ')) {
        return authHeader.substring(7)
    }
    
    // Try query parameter (for environments where headers aren't supported)
    if (socket.handshake.auth && socket.handshake.auth.token) {
        return socket.handshake.auth.token
    }
    
    // Try query string
    if (socket.handshake.query && socket.handshake.query.token) {
        return socket.handshake.query.token
    }
    
    return null
}

/**
 * Verify Auth0 token
 * This is a simplified version - adjust based on your Auth0 configuration
 * @param {string} token - JWT token
 * @returns {Object|null} Decoded token or null
 */
async function verifyAuth0Token(token) {
    try {
        // Use jose or jsonwebtoken for verification
        // This example uses a simple fetch to Auth0 userinfo endpoint
        // For production, use proper JWT verification
        
        const response = await fetch(`${process.env.ISSUER_BASE_URL}/userinfo`, {
            headers: {
                'Authorization': `Bearer ${token}`
            }
        })
        
        if (response.ok) {
            return await response.json()
        }
        
        // Alternative: Verify JWT locally
        // const { verify } = await import('jsonwebtoken')
        // const jwksClient = require('jwks-rsa')
        // ... implement proper JWT verification
        
        return null
    } catch (error) {
        console.error('Token verification error:', error)
        return null
    }
}

/**
 * Require authentication for write operations
 * @param {Object} socket - Socket.IO socket instance
 * @returns {boolean} Whether socket is authenticated
 */
function requireAuth(socket) {
    if (!socket.data.authenticated) {
        socket.emit('error', {
            code: 'UNAUTHORIZED',
            message: 'Authentication required for this operation'
        })
        return false
    }
    return true
}

/**
 * Check if socket's agent matches document generator
 * @param {Object} socket - Socket.IO socket instance
 * @param {Object} document - RERUM document
 * @returns {boolean} Whether agent is the generator
 */
function isGenerator(socket, document) {
    const socketAgent = socket.data.agent
    const docGenerator = document?.__rerum?.generatedBy
    return socketAgent && docGenerator && socketAgent === docGenerator
}

export {
    authenticateSocket,
    requireAuth,
    isGenerator,
    extractToken
}

Step 6.2: Apply Authentication Middleware

Update websocket/index.js to use authentication:

import { authenticateSocket, requireAuth } from './auth.js'

function setupConnectionHandlers(io) {
    // Apply authentication middleware
    io.use(authenticateSocket)
    
    io.on('connection', (socket) => {
        console.log(`Client connected: ${socket.id}, authenticated: ${socket.data.authenticated}`)
        
        // ... rest of handlers
        
        // Example: Protected operation
        socket.on('document:request-edit', (data) => {
            if (!requireAuth(socket)) return
            
            // Process edit request...
        })
    })
}

---

7. Document Room Management

Step 7.1: Create Room Manager

Create websocket/rooms.js:

#!/usr/bin/env node

/**
 * Document room management for real-time collaboration
 */

import { getIO } from './index.js'

// Track active rooms and their metadata
const roomMetadata = new Map()

/**
 * Join a document room
 * @param {Object} socket - Socket.IO socket instance
 * @param {string} documentId - The document ID to join
 * @param {Object} options - Join options
 */
function joinDocumentRoom(socket, documentId, options = {}) {
    const roomId = `doc:${documentId}`
    
    // Join the room
    socket.join(roomId)
    
    // Track room metadata
    if (!roomMetadata.has(roomId)) {
        roomMetadata.set(roomId, {
            documentId: documentId,
            clients: new Set(),
            createdAt: new Date().toISOString()
        })
    }
    
    const room = roomMetadata.get(roomId)
    room.clients.add(socket.id)
    
    // Store document ID on socket for cleanup
    if (!socket.data.rooms) {
        socket.data.rooms = new Set()
    }
    socket.data.rooms.add(documentId)
    
    console.log(`Socket ${socket.id} joined room ${roomId}. Total clients: ${room.clients.size}`)
    
    return {
        roomId: roomId,
        clientCount: room.clients.size
    }
}

/**
 * Leave a document room
 * @param {Object} socket - Socket.IO socket instance
 * @param {string} documentId - The document ID to leave
 */
function leaveDocumentRoom(socket, documentId) {
    const roomId = `doc:${documentId}`
    
    // Leave the room
    socket.leave(roomId)
    
    // Update metadata
    if (roomMetadata.has(roomId)) {
        const room = roomMetadata.get(roomId)
        room.clients.delete(socket.id)
        
        // Clean up empty rooms
        if (room.clients.size === 0) {
            roomMetadata.delete(roomId)
            console.log(`Room ${roomId} deleted (no clients)`)
        }
    }
    
    // Update socket data
    if (socket.data.rooms) {
        socket.data.rooms.delete(documentId)
    }
    
    console.log(`Socket ${socket.id} left room ${roomId}`)
}

/**
 * Leave all document rooms (on disconnect)
 * @param {Object} socket - Socket.IO socket instance
 */
function leaveAllRooms(socket) {
    if (socket.data.rooms) {
        for (const documentId of socket.data.rooms) {
            leaveDocumentRoom(socket, documentId)
            
            // Notify others
            const io = getIO()
            if (io) {
                io.to(`doc:${documentId}`).emit('document:user-left', {
                    socketId: socket.id,
                    documentId: documentId,
                    timestamp: new Date().toISOString()
                })
            }
        }
    }
}

/**
 * Get clients in a document room
 * @param {string} documentId - The document ID
 * @returns {Array} Array of socket IDs
 */
function getDocumentClients(documentId) {
    const roomId = `doc:${documentId}`
    const room = roomMetadata.get(roomId)
    return room ? Array.from(room.clients) : []
}

/**
 * Get room metadata
 * @param {string} documentId - The document ID
 * @returns {Object|null} Room metadata
 */
function getRoomMetadata(documentId) {
    const roomId = `doc:${documentId}`
    return roomMetadata.get(roomId) || null
}

/**
 * Check if a document has active collaborators
 * @param {string} documentId - The document ID
 * @returns {boolean} Whether document has active clients
 */
function hasActiveCollaborators(documentId) {
    const roomId = `doc:${documentId}`
    const room = roomMetadata.get(roomId)
    return room && room.clients.size > 0
}

/**
 * Get all active document rooms
 * @returns {Array} Array of document IDs with active rooms
 */
function getActiveDocuments() {
    return Array.from(roomMetadata.keys()).map(roomId => roomId.replace('doc:', ''))
}

export {
    joinDocumentRoom,
    leaveDocumentRoom,
    leaveAllRooms,
    getDocumentClients,
    getRoomMetadata,
    hasActiveCollaborators,
    getActiveDocuments
}

---

8. Real-Time Change Broadcasting

Step 8.1: Create Broadcast Service

Create websocket/broadcast.js:

#!/usr/bin/env node

/**
 * Real-time change broadcasting service
 * Handles document change notifications across WebSocket clients
 */

import { getIO } from './index.js'
import { hasActiveCollaborators, getDocumentClients } from './rooms.js'

/**
 * Operation types for RERUM
 */
const OperationType = {
    CREATE: 'create',
    UPDATE: 'update',
    OVERWRITE: 'overwrite',
    PATCH: 'patch',
    SET: 'set',
    UNSET: 'unset',
    DELETE: 'delete',
    RELEASE: 'release'
}

/**
 * Broadcast a document change to all subscribers
 * @param {Object} params - Broadcast parameters
 * @param {string} params.documentId - The document ID
 * @param {Object} params.document - The updated document
 * @param {string} params.operation - The operation type
 * @param {string} params.agentId - The agent who made the change
 * @param {string} params.excludeSocketId - Socket to exclude from broadcast
 * @param {Object} params.previousState - Previous document state (for OT)
 */
function broadcastChange({
    documentId,
    document,
    operation,
    agentId = null,
    excludeSocketId = null,
    previousState = null
}) {
    const io = getIO()
    if (!io) {
        console.warn('Socket.IO not initialized')
        return { broadcasted: false, reason: 'io_not_initialized' }
    }
    
    // Check if anyone is listening
    if (!hasActiveCollaborators(documentId)) {
        return { broadcasted: false, reason: 'no_collaborators' }
    }
    
    const payload = {
        type: 'document:change',
        documentId: documentId,
        document: document,
        operation: operation,
        agentId: agentId,
        version: extractVersion(document),
        timestamp: new Date().toISOString(),
        // Include previous state for OT conflict resolution
        previousVersion: previousState ? extractVersion(previousState) : null
    }
    
    const roomId = `doc:${documentId}`
    
    if (excludeSocketId) {
        io.to(roomId).except(excludeSocketId).emit('document:change', payload)
    } else {
        io.to(roomId).emit('document:change', payload)
    }
    
    const clientCount = getDocumentClients(documentId).length
    console.log(`Broadcasted ${operation} to ${clientCount} clients for document ${documentId}`)
    
    return { 
        broadcasted: true, 
        clientCount: clientCount,
        operation: operation
    }
}

/**
 * Broadcast a conflict notification
 * @param {Object} params - Conflict parameters
 * @param {string} params.documentId - The document ID
 * @param {Object} params.currentDocument - Current document state in DB
 * @param {Object} params.attemptedChange - The change that was attempted
 * @param {string} params.targetSocketId - Specific socket to notify
 * @param {string} params.conflictType - Type of conflict (version_mismatch, etc.)
 */
function broadcastConflict({
    documentId,
    currentDocument,
    attemptedChange,
    targetSocketId = null,
    conflictType = 'version_mismatch'
}) {
    const io = getIO()
    if (!io) return
    
    const payload = {
        type: 'document:conflict',
        documentId: documentId,
        currentDocument: currentDocument,
        currentVersion: extractVersion(currentDocument),
        attemptedChange: attemptedChange,
        conflictType: conflictType,
        timestamp: new Date().toISOString(),
        message: getConflictMessage(conflictType)
    }
    
    if (targetSocketId) {
        io.to(targetSocketId).emit('document:conflict', payload)
    } else {
        io.to(`doc:${documentId}`).emit('document:conflict', payload)
    }
    
    console.log(`Broadcasted conflict (${conflictType}) for document ${documentId}`)
}

/**
 * Broadcast document deletion
 * @param {string} documentId - The document ID
 * @param {string} agentId - The agent who deleted
 */
function broadcastDeletion(documentId, agentId) {
    const io = getIO()
    if (!io) return
    
    const payload = {
        type: 'document:deleted',
        documentId: documentId,
        agentId: agentId,
        timestamp: new Date().toISOString()
    }
    
    io.to(`doc:${documentId}`).emit('document:deleted', payload)
    console.log(`Broadcasted deletion for document ${documentId}`)
}

/**
 * Extract version identifier from document
 * @param {Object} document - RERUM document
 * @returns {string} Version identifier
 */
function extractVersion(document) {
    if (!document || !document.__rerum) return null
    
    // For overwrites, use isOverwritten timestamp
    if (document.__rerum.isOverwritten) {
        return document.__rerum.isOverwritten
    }
    
    // For versioned updates, use the document ID
    return document['@id'] || document.id || document._id
}

/**
 * Get human-readable conflict message
 * @param {string} conflictType - Type of conflict
 * @returns {string} Message
 */
function getConflictMessage(conflictType) {
    const messages = {
        'version_mismatch': 'The document was modified by another user. Your changes could not be applied.',
        'concurrent_edit': 'Another user is editing this document. Changes may conflict.',
        'deleted': 'This document has been deleted.',
        'released': 'This document has been released and cannot be modified.',
        'unauthorized': 'You are not authorized to modify this document.'
    }
    return messages[conflictType] || 'A conflict occurred.'
}

export {
    broadcastChange,
    broadcastConflict,
    broadcastDeletion,
    OperationType,
    extractVersion
}

---

9. Operational Transformation Implementation

Step 9.1: Create OT Manager

Create websocket/ot.js:

#!/usr/bin/env node

/**
 * Operational Transformation (OT) Manager for RERUM
 * Handles conflict resolution for concurrent document edits
 * 
 * This implementation uses JSON Patch (RFC 6902) operations
 * which are well-suited for JSON document transformation
 */

import * as jsonpatch from 'fast-json-patch'

/**
 * OT Document state tracker
 * Tracks pending operations and document versions
 */
class OTDocument {
    constructor(documentId, initialState) {
        this.documentId = documentId
        this.serverState = JSON.parse(JSON.stringify(initialState))
        this.version = 0
        this.pendingOperations = []
        this.history = []
    }
    
    /**
     * Apply an operation to the server state
     * @param {Array} patches - JSON Patch operations
     * @param {string} agentId - Agent applying the operation
     * @returns {Object} Result with new state and version
     */
    applyOperation(patches, agentId) {
        try {
            // Validate patches
            const errors = jsonpatch.validate(patches, this.serverState)
            if (errors) {
                return {
                    success: false,
                    error: 'Invalid patch operations',
                    details: errors
                }
            }
            
            // Apply patches
            const result = jsonpatch.applyPatch(
                this.serverState,
                patches,
                true, // validate
                true  // mutate in place
            )
            
            this.version++
            
            // Store in history
            this.history.push({
                version: this.version,
                patches: patches,
                agentId: agentId,
                timestamp: new Date().toISOString()
            })
            
            // Trim history if too long
            if (this.history.length > 100) {
                this.history = this.history.slice(-50)
            }
            
            return {
                success: true,
                state: this.serverState,
                version: this.version
            }
        } catch (error) {
            return {
                success: false,
                error: error.message
            }
        }
    }
    
    /**
     * Transform an operation against concurrent operations
     * @param {Array} clientPatches - Client's patches
     * @param {number} clientVersion - Client's base version
     * @returns {Object} Transformed patches
     */
    transformOperation(clientPatches, clientVersion) {
        if (clientVersion === this.version) {
            // No transformation needed
            return { patches: clientPatches, transformed: false }
        }
        
        // Get operations that happened since client's version
        const concurrentOps = this.history.filter(h => h.version > clientVersion)
        
        if (concurrentOps.length === 0) {
            return { patches: clientPatches, transformed: false }
        }
        
        // Transform client patches against each concurrent operation
        let transformedPatches = clientPatches
        
        for (const op of concurrentOps) {
            transformedPatches = this.transformPatches(transformedPatches, op.patches)
        }
        
        return {
            patches: transformedPatches,
            transformed: true,
            transformedAgainst: concurrentOps.length
        }
    }
    
    /**
     * Transform patches A against patches B
     * This is a simplified transformation - for production use consider ot-json1
     * @param {Array} patchesA - Patches to transform
     * @param {Array} patchesB - Patches to transform against
     * @returns {Array} Transformed patches
     */
    transformPatches(patchesA, patchesB) {
        // Simple transformation: adjust paths based on array operations
        return patchesA.map(patchA => {
            let transformed = { ...patchA }
            
            for (const patchB of patchesB) {
                // If B added/removed array elements, adjust A's indices
                if (patchB.op === 'add' && patchA.path.startsWith(patchB.path)) {
                    // Shift indices after the added element
                    transformed = this.adjustArrayIndex(transformed, patchB.path, 1)
                } else if (patchB.op === 'remove' && patchA.path.startsWith(patchB.path)) {
                    // Shift indices after the removed element
                    transformed = this.adjustArrayIndex(transformed, patchB.path, -1)
                }
            }
            
            return transformed
        })
    }
    
    /**
     * Adjust array indices in a patch path
     * @param {Object} patch - The patch to adjust
     * @param {string} basePath - The base path of the array operation
     * @param {number} delta - Amount to adjust (+1 for add, -1 for remove)
     * @returns {Object} Adjusted patch
     */
    adjustArrayIndex(patch, basePath, delta) {
        const pathParts = patch.path.split('/')
        const basePathParts = basePath.split('/')
        
        // Find the array index in the path
        for (let i = 0; i < pathParts.length; i++) {
            const index = parseInt(pathParts[i])
            if (!isNaN(index) && i === basePathParts.length - 1) {
                const baseIndex = parseInt(basePathParts[basePathParts.length - 1])
                if (index >= baseIndex) {
                    pathParts[i] = String(index + delta)
                }
            }
        }
        
        return { ...patch, path: pathParts.join('/') }
    }
    
    /**
     * Get current state
     * @returns {Object} Current document state
     */
    getState() {
        return JSON.parse(JSON.stringify(this.serverState))
    }
    
    /**
     * Get current version
     * @returns {number} Current version
     */
    getVersion() {
        return this.version
    }
}

/**
 * OT Manager - manages multiple documents
 */
class OTManager {
    constructor() {
        this.documents = new Map()
    }
    
    /**
     * Initialize or get an OT document
     * @param {string} documentId - Document ID
     * @param {Object} initialState - Initial document state
     * @returns {OTDocument} OT document instance
     */
    getOrCreateDocument(documentId, initialState) {
        if (!this.documents.has(documentId)) {
            this.documents.set(documentId, new OTDocument(documentId, initialState))
        }
        return this.documents.get(documentId)
    }
    
    /**
     * Apply a change from a client
     * @param {string} documentId - Document ID
     * @param {Array} patches - JSON Patch operations
     * @param {number} clientVersion - Client's base version
     * @param {string} agentId - Agent making the change
     * @returns {Object} Result
     */
    applyChange(documentId, patches, clientVersion, agentId) {
        const doc = this.documents.get(documentId)
        if (!doc) {
            return { success: false, error: 'Document not found in OT manager' }
        }
        
        // Transform if needed
        const { patches: transformedPatches, transformed } = doc.transformOperation(patches, clientVersion)
        
        // Apply the (potentially transformed) patches
        const result = doc.applyOperation(transformedPatches, agentId)
        
        if (result.success) {
            return {
                success: true,
                state: result.state,
                version: result.version,
                transformed: transformed,
                appliedPatches: transformedPatches
            }
        }
        
        return result
    }
    
    /**
     * Update document state from external source (e.g., direct API call)
     * @param {string} documentId - Document ID
     * @param {Object} newState - New document state
     */
    updateDocument(documentId, newState) {
        const doc = this.documents.get(documentId)
        if (doc) {
            // Generate patches from old to new state
            const patches = jsonpatch.compare(doc.serverState, newState)
            if (patches.length > 0) {
                doc.applyOperation(patches, 'external')
            }
        }
    }
    
    /**
     * Remove document from OT manager
     * @param {string} documentId - Document ID
     */
    removeDocument(documentId) {
        this.documents.delete(documentId)
    }
    
    /**
     * Check if document is being tracked
     * @param {string} documentId - Document ID
     * @returns {boolean}
     */
    hasDocument(documentId) {
        return this.documents.has(documentId)
    }
}

// Singleton instance
const otManager = new OTManager()

/**
 * Generate JSON Patch from two document states
 * @param {Object} oldDoc - Old document state
 * @param {Object} newDoc - New document state
 * @returns {Array} JSON Patch operations
 */
function generatePatch(oldDoc, newDoc) {
    return jsonpatch.compare(oldDoc, newDoc)
}

/**
 * Apply JSON Patch to a document
 * @param {Object} doc - Document to patch
 * @param {Array} patches - JSON Patch operations
 * @returns {Object} Patched document
 */
function applyPatch(doc, patches) {
    const cloned = JSON.parse(JSON.stringify(doc))
    jsonpatch.applyPatch(cloned, patches)
    return cloned
}

export {
    OTManager,
    OTDocument,
    otManager,
    generatePatch,
    applyPatch
}

---

10. Integration with Existing Endpoints

Step 10.1: Modify controllers/overwrite.js

Add WebSocket broadcasting to the overwrite controller:

// Add at the top of the file
import { broadcastChange, broadcastConflict, OperationType } from '../websocket/broadcast.js'
import { otManager } from '../websocket/ot.js'

// Inside the overwrite function, after successful overwrite:
// (After line: res.json(newObject))

// Add this block:
// Broadcast change to WebSocket clients
const broadcastResult = broadcastChange({
    documentId: id,
    document: newObject,
    operation: OperationType.OVERWRITE,
    agentId: agentRequestingOverwrite,
    excludeSocketId: req.headers['x-socket-id'], // Allow clients to exclude themselves
    previousState: originalObject
})

// Update OT manager
if (otManager.hasDocument(id)) {
    otManager.updateDocument(id, newObject)
}

// For conflict responses (409), add:
// (Inside the version mismatch block)

// Broadcast conflict notification
if (req.headers['x-socket-id']) {
    broadcastConflict({
        documentId: id,
        currentDocument: originalObject,
        attemptedChange: objectReceived,
        targetSocketId: req.headers['x-socket-id'],
        conflictType: 'version_mismatch'
    })
}

Step 10.2: Modify controllers/putUpdate.js

Add WebSocket broadcasting to versioned updates:

// Add at the top
import { broadcastChange, OperationType } from '../websocket/broadcast.js'

// After successful update (after res.json()):
broadcastChange({
    documentId: originalId,
    document: newObject,
    operation: OperationType.UPDATE,
    agentId: getAgentClaim(req),
    excludeSocketId: req.headers['x-socket-id']
})

// Also broadcast to the NEW document ID since it changed
broadcastChange({
    documentId: newObject._id,
    document: newObject,
    operation: OperationType.UPDATE,
    agentId: getAgentClaim(req),
    excludeSocketId: req.headers['x-socket-id']
})

Step 10.3: Modify controllers/delete.js

Add WebSocket broadcasting for deletions:

// Add at the top
import { broadcastDeletion } from '../websocket/broadcast.js'
import { otManager } from '../websocket/ot.js'

// After successful deletion:
broadcastDeletion(id, agentRequestingDelete)

// Remove from OT manager
otManager.removeDocument(id)

Step 10.4: Create WebSocket Header Middleware

Create middleware to handle WebSocket-related headers in websocket/middleware.js:

#!/usr/bin/env node

/**
 * Express middleware for WebSocket integration
 */

/**
 * Middleware to extract WebSocket client ID from request headers
 * Clients should include X-Socket-ID header with their socket.id
 */
function extractSocketId(req, res, next) {
    req.socketId = req.headers['x-socket-id'] || null
    next()
}

/**
 * Middleware to add WebSocket info to response
 * Useful for clients to know if real-time sync is available
 */
function addWebSocketInfo(req, res, next) {
    res.set('X-WebSocket-Available', 'true')
    res.set('X-WebSocket-Endpoint', '/socket.io/')
    next()
}

export { extractSocketId, addWebSocketInfo }

Add to app.js:

import { extractSocketId, addWebSocketInfo } from './websocket/middleware.js'

// Add after other middleware
app.use(extractSocketId)
app.use(addWebSocketInfo)

---

11. Client-Side Implementation Guide

Step 11.1: Basic Client Setup

// client-example.js
import { io } from 'socket.io-client'

class RerumRealtimeClient {
    constructor(serverUrl, options = {}) {
        this.serverUrl = serverUrl
        this.socket = null
        this.token = options.token || null
        this.documentId = null
        this.localVersion = 0
        this.pendingChanges = []
        this.callbacks = {
            onConnect: () => {},
            onDisconnect: () => {},
            onChange: () => {},
            onConflict: () => {},
            onError: () => {}
        }
    }
    
    /**
     * Connect to the WebSocket server
     */
    connect() {
        this.socket = io(this.serverUrl, {
            transports: ['websocket', 'polling'],
            auth: {
                token: this.token
            },
            reconnection: true,
            reconnectionAttempts: 5,
            reconnectionDelay: 1000
        })
        
        this.socket.on('connect', () => {
            console.log('Connected to RERUM WebSocket')
            this.callbacks.onConnect(this.socket.id)
        })
        
        this.socket.on('disconnect', (reason) => {
            console.log('Disconnected:', reason)
            this.callbacks.onDisconnect(reason)
        })
        
        this.socket.on('document:change', (data) => {
            console.log('Document changed:', data)
            this.localVersion = data.version
            this.callbacks.onChange(data)
        })
        
        this.socket.on('document:conflict', (data) => {
            console.log('Conflict detected:', data)
            this.callbacks.onConflict(data)
        })
        
        this.socket.on('document:deleted', (data) => {
            console.log('Document deleted:', data)
            this.callbacks.onChange({ ...data, operation: 'delete' })
        })
        
        this.socket.on('error', (error) => {
            console.error('Socket error:', error)
            this.callbacks.onError(error)
        })
        
        return this
    }
    
    /**
     * Join a document room for real-time updates
     * @param {string} documentId - The document ID to join
     */
    joinDocument(documentId) {
        this.documentId = documentId
        this.socket.emit('document:join', documentId)
        return this
    }
    
    /**
     * Leave a document room
     * @param {string} documentId - The document ID to leave
     */
    leaveDocument(documentId) {
        this.socket.emit('document:leave', documentId || this.documentId)
        if (documentId === this.documentId) {
            this.documentId = null
        }
        return this
    }
    
    /**
     * Make an API request with socket ID for broadcast exclusion
     * @param {string} endpoint - API endpoint
     * @param {Object} options - Fetch options
     */
    async apiRequest(endpoint, options = {}) {
        const headers = {
            'Content-Type': 'application/json',
            ...options.headers
        }
        
        // Add socket ID to exclude self from broadcasts
        if (this.socket && this.socket.id) {
            headers['X-Socket-ID'] = this.socket.id
        }
        
        // Add auth token
        if (this.token) {
            headers['Authorization'] = `Bearer ${this.token}`
        }
        
        const response = await fetch(`${this.serverUrl}${endpoint}`, {
            ...options,
            headers
        })
        
        return response
    }
    
    /**
     * Overwrite a document with optimistic locking
     * @param {Object} document - The document to save
     * @param {string} expectedVersion - Expected version for optimistic locking
     */
    async overwrite(document, expectedVersion = null) {
        const headers = {}
        if (expectedVersion) {
            headers['If-Overwritten-Version'] = expectedVersion
        }
        
        const response = await this.apiRequest('/v1/api/overwrite', {
            method: 'PUT',
            headers,
            body: JSON.stringify(document)
        })
        
        if (response.status === 409) {
            // Conflict - another user modified the document
            const conflictData = await response.json()
            this.callbacks.onConflict({
                type: 'version_mismatch',
                currentDocument: conflictData.currentVersion,
                attemptedChange: document
            })
            return { success: false, conflict: true, data: conflictData }
        }
        
        if (!response.ok) {
            throw new Error(`API error: ${response.status}`)
        }
        
        const result = await response.json()
        this.localVersion = result.__rerum?.isOverwritten
        return { success: true, data: result }
    }
    
    /**
     * Set event callbacks
     * @param {Object} callbacks - Callback functions
     */
    on(callbacks) {
        Object.assign(this.callbacks, callbacks)
        return this
    }
    
    /**
     * Disconnect from the server
     */
    disconnect() {
        if (this.socket) {
            this.socket.disconnect()
        }
    }
    
    /**
     * Get the socket ID (useful for API requests)
     */
    getSocketId() {
        return this.socket?.id
    }
}

export default RerumRealtimeClient

Step 11.2: Usage Example

// Example usage in a client application

import RerumRealtimeClient from './rerum-realtime-client.js'

// Initialize client
const client = new RerumRealtimeClient('https://store.rerum.io', {
    token: 'your-auth0-access-token'
})

// Set up event handlers
client.on({
    onConnect: (socketId) => {
        console.log('Connected with ID:', socketId)
        // Join the document you're editing
        client.joinDocument('abcdef1234567890')
    },
    
    onDisconnect: (reason) => {
        console.log('Disconnected:', reason)
        // Show offline indicator
    },
    
    onChange: (data) => {
        console.log('Document updated:', data)
        // Update your local state/UI with the new document
        updateLocalDocument(data.document)
    },
    
    onConflict: (data) => {
        console.log('Conflict!', data)
        // Handle conflict - show dialog, auto-merge, etc.
        handleConflict(data)
    },
    
    onError: (error) => {
        console.error('Error:', error)
    }
})

// Connect
client.connect()

// Later, when saving changes:
async function saveDocument(document) {
    const result = await client.overwrite(document, document.__rerum?.isOverwritten)
    
    if (result.conflict) {
        // Handle conflict
        showConflictDialog(result.data.currentVersion, document)
    } else {
        // Success
        console.log('Saved:', result.data)
    }
}

// When done editing:
client.leaveDocument('abcdef1234567890')
client.disconnect()

Step 11.3: React Hook Example

// useRerumRealtime.js
import { useEffect, useState, useCallback, useRef } from 'react'
import RerumRealtimeClient from './rerum-realtime-client.js'

export function useRerumRealtime(serverUrl, token) {
    const clientRef = useRef(null)
    const [connected, setConnected] = useState(false)
    const [document, setDocument] = useState(null)
    const [conflict, setConflict] = useState(null)
    
    useEffect(() => {
        const client = new RerumRealtimeClient(serverUrl, { token })
        
        client.on({
            onConnect: () => setConnected(true),
            onDisconnect: () => setConnected(false),
            onChange: (data) => setDocument(data.document),
            onConflict: (data) => setConflict(data)
        })
        
        client.connect()
        clientRef.current = client
        
        return () => {
            client.disconnect()
        }
    }, [serverUrl, token])
    
    const joinDocument = useCallback((docId) => {
        clientRef.current?.joinDocument(docId)
    }, [])
    
    const leaveDocument = useCallback((docId) => {
        clientRef.current?.leaveDocument(docId)
    }, [])
    
    const saveDocument = useCallback(async (doc) => {
        return clientRef.current?.overwrite(doc, doc.__rerum?.isOverwritten)
    }, [])
    
    const clearConflict = useCallback(() => {
        setConflict(null)
    }, [])
    
    return {
        connected,
        document,
        conflict,
        joinDocument,
        leaveDocument,
        saveDocument,
        clearConflict,
        socketId: clientRef.current?.getSocketId()
    }
}

---

12. Testing Strategy

Step 12.1: Unit Tests

Create __tests__/websocket.test.js:

import { jest } from '@jest/globals'
import { OTDocument, OTManager, generatePatch, applyPatch } from '../websocket/ot.js'

describe('Operational Transformation', () => {
    describe('OTDocument', () => {
        test('should apply simple patches', () => {
            const doc = new OTDocument('test-1', { name: 'Original' })
            
            const result = doc.applyOperation(
                [{ op: 'replace', path: '/name', value: 'Updated' }],
                'agent-1'
            )
            
            expect(result.success).toBe(true)
            expect(result.state.name).toBe('Updated')
            expect(result.version).toBe(1)
        })
        
        test('should transform concurrent operations', () => {
            const doc = new OTDocument('test-2', { items: ['a', 'b', 'c'] })
            
            // First operation: add 'd' at index 3
            doc.applyOperation(
                [{ op: 'add', path: '/items/3', value: 'd' }],
                'agent-1'
            )
            
            // Second operation: client at version 0 tries to add 'e' at index 3
            const transformed = doc.transformOperation(
                [{ op: 'add', path: '/items/3', value: 'e' }],
                0 // client was at version 0
            )
            
            expect(transformed.transformed).toBe(true)
            // The path should be adjusted to account for the first insert
            expect(transformed.patches[0].path).toBe('/items/4')
        })
    })
    
    describe('generatePatch', () => {
        test('should generate correct patches', () => {
            const oldDoc = { name: 'Old', count: 1 }
            const newDoc = { name: 'New', count: 1, extra: true }
            
            const patches = generatePatch(oldDoc, newDoc)
            
            expect(patches).toContainEqual({ op: 'replace', path: '/name', value: 'New' })
            expect(patches).toContainEqual({ op: 'add', path: '/extra', value: true })
        })
    })
})

Step 12.2: Integration Tests

Create __tests__/websocket-integration.test.js:

import { jest } from '@jest/globals'
import { createServer } from 'http'
import { Server } from 'socket.io'
import { io as Client } from 'socket.io-client'

describe('WebSocket Integration', () => {
    let httpServer, ioServer, clientSocket
    
    beforeAll((done) => {
        httpServer = createServer()
        ioServer = new Server(httpServer)
        httpServer.listen(() => {
            const port = httpServer.address().port
            clientSocket = Client(`http://localhost:${port}`)
            clientSocket.on('connect', done)
        })
    })
    
    afterAll(() => {
        ioServer.close()
        clientSocket.close()
        httpServer.close()
    })
    
    test('should join document room', (done) => {
        ioServer.on('connection', (socket) => {
            socket.on('document:join', (docId) => {
                expect(docId).toBe('test-doc-123')
                done()
            })
        })
        
        clientSocket.emit('document:join', 'test-doc-123')
    })
    
    test('should receive document changes', (done) => {
        clientSocket.on('document:change', (data) => {
            expect(data.documentId).toBe('test-doc-123')
            expect(data.document.name).toBe('Updated')
            done()
        })
        
        // Simulate server broadcasting a change
        ioServer.emit('document:change', {
            documentId: 'test-doc-123',
            document: { name: 'Updated' },
            operation: 'overwrite'
        })
    })
})

Step 12.3: Manual Testing Checklist

## Manual Testing Checklist

### Connection Tests
- [ ] Client can connect to WebSocket server
- [ ] Client reconnects automatically after disconnect
- [ ] Authentication works with valid Auth0 token
- [ ] Unauthenticated clients are handled appropriately

### Document Room Tests
- [ ] Client can join a document room
- [ ] Client can leave a document room
- [ ] Multiple clients can join the same room
- [ ] Client leaving room notifies others

### Real-Time Updates
- [ ] Changes via /overwrite are broadcast to room
- [ ] Changes via /update are broadcast to room
- [ ] Changes via /patch are broadcast to room
- [ ] Deletions are broadcast to room
- [ ] Sender is excluded from their own broadcasts

### Conflict Handling
- [ ] 409 conflict is detected on version mismatch
- [ ] Conflict notification is sent to client
- [ ] Current document state is provided in conflict

### Cluster Mode (PM2)
- [ ] Multiple instances can communicate via Redis
- [ ] Sticky sessions work correctly
- [ ] Broadcasts reach all relevant clients across instances

---

13. Deployment Checklist

Pre-Deployment

## Pre-Deployment Checklist

### Environment Configuration
- [ ] Redis installed and running on RHEL VM
- [ ] Redis password configured in .env
- [ ] All new npm packages installed
- [ ] PM2 ecosystem.config.cjs created

### Code Changes
- [ ] websocket/ directory created with all modules
- [ ] bin/rerum_v1.js updated for WebSocket init
- [ ] Controllers updated to broadcast changes
- [ ] Middleware added to app.js

### Nginx Configuration (if applicable)
- [ ] WebSocket location block added
- [ ] Sticky sessions configured
- [ ] Proxy timeouts adjusted for long-lived connections

### Testing
- [ ] Unit tests passing
- [ ] Integration tests passing
- [ ] Manual testing completed
- [ ] Load testing performed

Deployment Steps

# 1. Stop current PM2 processes
pm2 stop all

# 2. Pull latest code
git pull origin main

# 3. Install dependencies
npm install

# 4. Verify Redis is running
sudo systemctl status redis
redis-cli -a $REDIS_PASSWORD ping

# 5. Run tests
npm run runtest

# 6. Start with new PM2 config
pm2 start ecosystem.config.cjs

# 7. Verify processes are running
pm2 status
pm2 logs

# 8. Test WebSocket connection
# Use a WebSocket client to connect to wss://store.rerum.io/socket.io/

# 9. Monitor for errors
pm2 logs --err
tail -f /var/log/nginx/error.log

Rollback Plan

# If issues occur:

# 1. Stop PM2
pm2 stop all

# 2. Revert to previous version
git checkout HEAD~1

# 3. Restart without WebSocket
pm2 start ./bin/rerum_v1.js -i max --name rerum-api

# 4. Verify
pm2 status
curl https://store.rerum.io/v1/api

---

14. Troubleshooting

Common Issues

WebSocket Connection Fails

Symptom: Clients can't establish WebSocket connection

Check:

1. Nginx WebSocket configuration

   nginx -t
   grep -A 20 \"location /socket.io\" /etc/nginx/conf.d/rerum.conf
   

2. Firewall rules

   sudo firewall-cmd --list-all
   

3. PM2 process status

   pm2 status
   pm2 logs
   

Redis Connection Errors

Symptom: \"Failed to connect Redis adapter\" in logs

Check:

1. Redis service status

   sudo systemctl status redis
   

2. Redis connectivity

   redis-cli -a $REDIS_PASSWORD ping
   

3. Environment variables

   cat .env | grep REDIS
   

Broadcasts Not Reaching All Clients

Symptom: Changes made by one client not visible to others

Check:

1. Redis adapter connected

   pm2 logs | grep \"Redis adapter\"
   

2. Clients in same room

   // Add debug logging to rooms.js
   console.log('Room clients:', getDocumentClients(documentId))
   

3. Sticky sessions working

   # Multiple requests should hit same worker
   for i in {1..10}; do curl -s https://store.rerum.io/v1/api | grep -o \"worker-[0-9]*\"; done
   

High Memory Usage

Symptom: PM2 processes using excessive memory

Check:

1. Memory limits in ecosystem.config.cjs
2. History size in OT manager
3. Room cleanup on disconnect

Fix:

// Add to websocket/ot.js
// Periodic cleanup of inactive documents
setInterval(() => {
    for (const [docId, doc] of otManager.documents) {
        if (!hasActiveCollaborators(docId)) {
            otManager.removeDocument(docId)
        }
    }
}, 300000) // Every 5 minutes

---

Appendix A: Environment Variables Reference

# Existing RERUM variables
RERUM_API_VERSION=1.1.0
RERUM_BASE=https://store.rerum.io
RERUM_PREFIX=https://store.rerum.io/v1/
RERUM_ID_PREFIX=https://store.rerum.io/v1/id/
MONGO_CONNECTION_STRING=mongodb://...
MONGODBNAME=rerum
MONGODBCOLLECTION=objects

# Auth0 variables
AUDIENCE=...
ISSUER_BASE_URL=https://...auth0.com/
CLIENTID=...
RERUMSECRET=...

# NEW: Redis variables
REDIS_HOST=127.0.0.1
REDIS_PORT=6379
REDIS_PASSWORD=your-secure-password

# NEW: WebSocket variables (optional)
WS_PING_TIMEOUT=60000
WS_PING_INTERVAL=25000

---

Appendix B: API Changes Summary

New Headers

Header	Direction	Purpose
X-Socket-ID	Request	Client's socket ID for broadcast exclusion
X-WebSocket-Available	Response	Indicates WebSocket support
X-WebSocket-Endpoint	Response	WebSocket connection endpoint

New WebSocket Events

Event	Direction	Payload
document:join	Client → Server	documentId
document:leave	Client → Server	documentId
document:change	Server → Client	{ documentId, document, operation, timestamp, version }
document:conflict	Server → Client	{ documentId, currentDocument, attemptedChange, conflictType }
document:deleted	Server → Client	{ documentId, agentId, timestamp }
document:user-joined	Server → Client	{ socketId, documentId, timestamp }
document:user-left	Server → Client	{ socketId, documentId, timestamp }

---

Appendix C: Security Considerations

1. Token Validation: Always validate Auth0 tokens on WebSocket connections
2. Rate Limiting: Consider implementing rate limiting on WebSocket events
3. Input Validation: Validate all document IDs and payloads
4. Redis Security: Use password authentication, bind to localhost only
5. TLS: Ensure all WebSocket connections use WSS (TLS)
6. Room Authorization: Verify client has permission to join document rooms

---

Next Steps

After implementing this playbook:

1. Monitor Performance: Track WebSocket connection counts, message rates
2. Add Metrics: Integrate with your monitoring solution (Prometheus, etc.)
3. Consider Presence: If needed later, add user presence features
4. Optimize OT: For complex documents, consider using ot-json1 library
5. Add Cursor Sync: For collaborative editing, sync cursor positions

---

Document Version: 1.0 Last Updated: July 2025 Author: E1 (Emergent AI) " Observation: Create successful: /app/docs/WEBSOCKET_IMPLEMENTATION_PLAYBOOK.md