How to Build Real-Time Collaborative Applications: 8 Proven Techniques With Code

class ReliableWebSocket {
  constructor(url) {
    this.url = url;
    this.socket = null;
    this.isConnected = false;
    this.reconnectTimer = null;
    this.messageQueue = [];
    this.heartbeatInterval = null;

    // Track reconnect attempts with increasing delay
    this.reconnectAttempts = 0;
    this.maxReconnectDelay = 30000; // 30 seconds max

    this.connect();
  }

  connect() {
    try {
      this.socket = new WebSocket(this.url);
      this.setupSocketHandlers();
    } catch (error) {
      console.error('Connection setup failed:', error);
      this.scheduleReconnect();
    }
  }

  setupSocketHandlers() {
    this.socket.onopen = () => {
      console.log('Connected to server.');
      this.isConnected = true;
      this.reconnectAttempts = 0; // Reset on successful connection
      this.startHeartbeat();
      this.flushMessageQueue(); // Send any queued messages
    };

    this.socket.onclose = (event) => {
      console.log(`Connection closed. Code: ${event.code}, Reason: ${event.reason}`);
      this.isConnected = false;
      this.stopHeartbeat();
      this.scheduleReconnect();
    };

    this.socket.onerror = (error) => {
      console.error('WebSocket error:', error);
    };

    this.socket.onmessage = (event) => {
      this.handleIncomingMessage(event.data);
    };
  }

  send(data) {
    const message = JSON.stringify(data);
    if (this.isConnected && this.socket.readyState === WebSocket.OPEN) {
      this.socket.send(message);
    } else {
      // Store message to send when we reconnect
      this.messageQueue.push(message);
    }
  }

  flushMessageQueue() {
    while (this.messageQueue.length > 0 && this.isConnected) {
      const msg = this.messageQueue.shift();
      this.socket.send(msg);
    }
  }

  scheduleReconnect() {
    if (this.reconnectTimer) clearTimeout(this.reconnectTimer);

    // Calculate delay: 1s, then 1.5s, then 2.25s, etc., capped at 30s
    const baseDelay = 1000;
    const delay = Math.min(
      baseDelay * Math.pow(1.5, this.reconnectAttempts),
      this.maxReconnectDelay
    );

    this.reconnectAttempts++;
    console.log(`Reconnecting in ${Math.round(delay)}ms...`);

    this.reconnectTimer = setTimeout(() => this.connect(), delay);
  }

  startHeartbeat() {
    // Send a ping every 20 seconds to keep connection alive
    this.heartbeatInterval = setInterval(() => {
      if (this.isConnected) {
        this.send({ type: 'ping', timestamp: Date.now() });
      }
    }, 20000);
  }

  stopHeartbeat() {
    if (this.heartbeatInterval) {
      clearInterval(this.heartbeatInterval);
      this.heartbeatInterval = null;
    }
  }

  handleIncomingMessage(rawData) {
    try {
      const data = JSON.parse(rawData);
      // Ignore ping/pong messages internally, or handle them
      if (data.type === 'pong') {
        // Server responded to our ping, connection is alive
        return;
      }
      // ... process other message types like 'update', 'cursor', etc.
      console.log('Received:', data);
    } catch (e) {
      console.error('Failed to parse message:', e);
    }
  }

  disconnect() {
    this.stopHeartbeat();
    if (this.reconnectTimer) clearTimeout(this.reconnectTimer);
    if (this.socket) {
      this.socket.close(1000, 'Client closed connection');
    }
  }
}

// Usage
const ws = new ReliableWebSocket('wss://collab.example.com/socket');
ws.send({ type: 'join', documentId: 'doc_123' });

class GrowOnlySet {
  constructor() {
    this.items = new Set();
  }

  add(element) {
    this.items.add(element);
  }

  contains(element) {
    return this.items.has(element);
  }

  merge(otherSet) {
    // To merge, just add all items from the other set
    for (let element of otherSet.items) {
      this.items.add(element);
    }
  }

  getValues() {
    return Array.from(this.items);
  }
}

// User A's browser
const setA = new GrowOnlySet();
setA.add('Task 1');

// User B's browser, offline initially
const setB = new GrowOnlySet();
setB.add('Task 2');

// They connect and merge
setA.merge(setB);
setB.merge(setA);

console.log(setA.getValues()); // ['Task 1', 'Task 2']
console.log(setB.getValues()); // ['Task 1', 'Task 2'] - Now identical!

class TwoPhaseSet {
  constructor() {
    this.added = new GrowOnlySet();
    this.removed = new GrowOnlySet();
  }

  add(element) {
    this.added.add(element);
  }

  remove(element) {
    // You can only remove an element if it has been added
    if (this.added.contains(element)) {
      this.removed.add(element);
    }
  }

  contains(element) {
    return this.added.contains(element) && !this.removed.contains(element);
  }

  merge(otherSet) {
    this.added.merge(otherSet.added);
    this.removed.merge(otherSet.removed);
  }

  getValues() {
    return this.added.getValues().filter(item => !this.removed.contains(item));
  }
}

class TextCRDT {
  constructor(siteId) {
    this.siteId = siteId || this.generateId();
    this.sequence = []; // Array of { position: [...], char: 'a' }
  }

  generateId() {
    return Math.random().toString(36).substring(2, 15);
  }

  // Creates a position between 'before' and 'after'
  createPosition(before, after) {
    // Simplified: Use a fractional index.
    // In reality, you'd use a list of tuples with a site identifier.
    let low = before ? before.index : 0;
    let high = after ? after.index : 1;
    let newIndex = (low + high) / 2;

    // Include siteId to break ties if two users pick the same fraction
    return { index: newIndex, site: this.siteId };
  }

  insertAt(localIndex, char) {
    // localIndex is where the user *thinks* they're inserting (like cursor position 5)
    // We need to find the global positions before and after that point.
    let beforePos = localIndex > 0 ? this.sequence[localIndex - 1].position : null;
    let afterPos = localIndex < this.sequence.length ? this.sequence[localIndex].position : null;

    let newPosition = this.createPosition(beforePos, afterPos);
    let newEntry = { position: newPosition, char: char };

    // Insert, maintaining order by position
    let insertIndex = 0;
    while (insertIndex < this.sequence.length &&
           this.comparePositions(this.sequence[insertIndex].position, newPosition) < 0) {
      insertIndex++;
    }
    this.sequence.splice(insertIndex, 0, newEntry);

    // Return an operation to send to others
    return { type: 'insert', pos: newPosition, char: char, site: this.siteId };
  }

  comparePositions(posA, posB) {
    // First compare the numeric index
    if (posA.index !== posB.index) {
      return posA.index - posB.index;
    }
    // If equal, compare site IDs for a deterministic tie-breaker
    return posA.site.localeCompare(posB.site);
  }

  applyRemoteOperation(op) {
    if (op.type === 'insert') {
      let newEntry = { position: op.pos, char: op.char };
      let insertIndex = 0;
      while (insertIndex < this.sequence.length &&
             this.comparePositions(this.sequence[insertIndex].position, op.pos) < 0) {
        insertIndex++;
      }
      // Only insert if not already present (idempotent)
      if (this.sequence[insertIndex]?.position !== op.pos) {
        this.sequence.splice(insertIndex, 0, newEntry);
      }
    }
    // ... handle 'delete' operations similarly
  }

  getText() {
    return this.sequence.map(entry => entry.char).join('');
  }
}

// Simulating collaboration
const user1CRDT = new TextCRDT('user1');
const user2CRDT = new TextCRDT('user2');

// User 1 types 'Hello' at the start
let op1 = user1CRDT.insertAt(0, 'H');
user2CRDT.applyRemoteOperation(op1); // Sync

let op2 = user1CRDT.insertAt(1, 'e');
user2CRDT.applyRemoteOperation(op2);

// User 2 types 'i' at position 1 (between 'H' and 'e')
let op3 = user2CRDT.insertAt(1, 'i');
user1CRDT.applyRemoteOperation(op3);

console.log(user1CRDT.getText()); // 'Hie'
console.log(user2CRDT.getText()); // 'Hie' - Consistent!

class LWWRegister {
  constructor(replicaId) {
    this.replicaId = replicaId;
    this.value = null;
    this.timestamp = 0; // Simplified scalar timestamp for demo
  }

  set(newValue, newTimestamp = Date.now()) {
    if (newTimestamp > this.timestamp) {
      this.value = newValue;
      this.timestamp = newTimestamp;
    } else if (newTimestamp === this.timestamp && this.replicaId < replicaIdOfNew) {
      // Tie-breaker: replica with lower ID wins, ensuring consistency
      this.value = newValue;
    }
  }

  merge(otherRegister) {
    // Take the value with the highest timestamp
    if (otherRegister.timestamp > this.timestamp) {
      this.value = otherRegister.value;
      this.timestamp = otherRegister.timestamp;
    } else if (otherRegister.timestamp === this.timestamp) {
      // Tie-break by replicaId
      if (otherRegister.replicaId < this.replicaId) {
        this.value = otherRegister.value;
        this.replicaId = otherRegister.replicaId;
      }
    }
  }
}

class Awareness {
  constructor(localUserId) {
    this.localState = { cursor: null, name: 'Anonymous', color: '#ff0000' };
    this.states = new Map(); // userId -> { state, lastUpdated }
    this.localUserId = localUserId;

    // Update server every 2 seconds
    setInterval(() => this.broadcastUpdate(), 2000);
  }

  broadcastUpdate() {
    // Send our current state to others via WebSocket
    const message = {
      type: 'awareness',
      userId: this.localUserId,
      state: this.localState,
      timestamp: Date.now()
    };
    // ws.send(message);
  }

  processIncoming(message) {
    const { userId, state, timestamp } = message;
    const existing = this.states.get(userId);

    // Use last-write-wins based on timestamp
    if (!existing || timestamp > existing.lastUpdated) {
      this.states.set(userId, { state, lastUpdated: timestamp });
    }

    // Cleanup old entries (e.g., older than 10 seconds)
    const now = Date.now();
    for (let [id, data] of this.states.entries()) {
      if (now - data.lastUpdated > 10000 && id !== this.localUserId) {
        this.states.delete(id);
      }
    }
  }

  getOnlineUsers() {
    return Array.from(this.states.keys());
  }
}

class OperationBatcher {
  constructor(sendCallback, batchWindow = 100) {
    this.batch = [];
    this.sendCallback = sendCallback; // Function to send batched ops
    this.batchWindow = batchWindow;
    this.timer = null;
  }

  addOperation(op) {
    this.batch.push(op);
    if (!this.timer) {
      this.timer = setTimeout(() => this.flush(), this.batchWindow);
    }
  }

  flush() {
    if (this.batch.length > 0) {
      this.sendCallback(this.batch);
      this.batch = [];
    }
    if (this.timer) {
      clearTimeout(this.timer);
      this.timer = null;
    }
  }
}

// Usage in editor
const batcher = new OperationBatcher((ops) => {
  ws.send({ type: 'opsBatch', operations: ops });
});

// On each keystroke:
editor.on('change', (changeOp) => {
  batcher.addOperation(changeOp);
});

class OfflineManager {
  constructor() {
    this.pendingOperations = [];
    this.lastSyncedVersion = 0;
    this.db = this.initDB();
  }

  async initDB() {
    // Open IndexedDB database
    return new Promise((resolve) => {
      // ... IndexedDB setup code
    });
  }

  async saveOperationLocally(op) {
    this.pendingOperations.push(op);
    await this.db.save('pending_ops', op);
  }

  async syncOnReconnect(websocket) {
    // Send our last known version
    websocket.send({ type: 'sync', version: this.lastSyncedVersion });

    // Server will respond with missing ops
  }

  async processSyncResponse(missingOps) {
    for (let op of missingOps) {
      await this.applyOperation(op);
    }
    // Also send our pending ops to server
    for (let op of this.pendingOperations) {
      await this.sendOperation(op);
    }
    this.pendingOperations = [];
  }
}

async function testConvergence() {
  const clients = [];
  const network = new SimulatedNetwork(); // Delays, drops messages

  // Create 3 virtual clients with their own CRDT
  for (let i = 0; i < 3; i++) {
    clients.push(new TextCRDT(`client${i}`));
  }

  // Simulate random typing on each client
  for (let step = 0; step < 100; step++) {
    const clientIdx = Math.floor(Math.random() * clients.length);
    const client = clients[clientIdx];
    const op = client.insertAt(
      Math.floor(Math.random() * client.getText().length),
      String.fromCharCode(97 + Math.floor(Math.random() * 26))
    );

    // Broadcast to other clients via simulated network
    network.broadcast(op, clientIdx);
  }

  // Force final sync
  network.deliverAllMessages();

  // Check if all clients have identical text
  const firstText = clients[0].getText();
  const allMatch = clients.every(c => c.getText() === firstText);
  console.assert(allMatch, 'CRDTs did not converge!');
  console.log('Test passed:', allMatch);
}