Build Real-Time Collaboration with CRDTs

// src/crdt/document.ts
import * as Y from 'yjs';
import { z } from 'zod';

// Each document is a Yjs doc with typed sections
export function createTaskDocument(): Y.Doc {
  const doc = new Y.Doc();

  // Shared types for different parts of the task
  const title = doc.getText('title');
  const description = doc.getText('description');
  const checklist = doc.getArray<Y.Map<any>>('checklist');
  const comments = doc.getArray<Y.Map<any>>('comments');
  const metadata = doc.getMap('metadata');

  return doc;
}

export function addChecklistItem(doc: Y.Doc, text: string, assignee?: string): void {
  const checklist = doc.getArray<Y.Map<any>>('checklist');
  const item = new Y.Map<any>();
  item.set('id', crypto.randomUUID());
  item.set('text', text);
  item.set('completed', false);
  item.set('assignee', assignee ?? null);
  item.set('createdAt', new Date().toISOString());
  checklist.push([item]);
}

export function toggleChecklistItem(doc: Y.Doc, index: number): void {
  const checklist = doc.getArray<Y.Map<any>>('checklist');
  const item = checklist.get(index);
  if (item) {
    item.set('completed', !item.get('completed'));
  }
}

// Awareness: who's editing what right now
export const AwarenessState = z.object({
  userId: z.string(),
  userName: z.string(),
  color: z.string(),
  cursor: z.object({
    field: z.string(),          // which field they're in
    index: z.number().int(),    // cursor position
    length: z.number().int(),   // selection length
  }).optional(),
});

// src/server/ws-sync.ts
import { Hono } from 'hono';
import { createNodeWebSocket } from '@hono/node-ws';
import * as Y from 'yjs';
import { Redis } from 'ioredis';

const redis = new Redis(process.env.REDIS_URL!);

// In-memory document cache
const docs = new Map<string, Y.Doc>();
const connections = new Map<string, Set<WebSocket>>();

export function createSyncServer() {
  const app = new Hono();
  const { injectWebSocket, upgradeWebSocket } = createNodeWebSocket({ app });

  app.get('/ws/doc/:docId', upgradeWebSocket((c) => {
    const docId = c.req.param('docId');

    return {
      async onOpen(_event, ws) {
        // Load or create document
        let doc = docs.get(docId);
        if (!doc) {
          doc = new Y.Doc();
          const saved = await redis.getBuffer(`doc:${docId}`);
          if (saved) {
            Y.applyUpdate(doc, saved);
          }
          docs.set(docId, doc);
        }

        // Track connection
        if (!connections.has(docId)) connections.set(docId, new Set());
        connections.get(docId)!.add(ws.raw as any);

        // Send current state
        const state = Y.encodeStateAsUpdate(doc);
        (ws.raw as any).send(state);
      },

      onMessage(event, ws) {
        const doc = docs.get(docId)!;
        const update = new Uint8Array(event.data as ArrayBuffer);

        // Apply update to server doc
        Y.applyUpdate(doc, update);

        // Broadcast to other clients
        const peers = connections.get(docId);
        if (peers) {
          for (const peer of peers) {
            if (peer !== ws.raw && peer.readyState === 1) {
              peer.send(update);
            }
          }
        }

        // Persist to Redis (debounced in production)
        const encoded = Y.encodeStateAsUpdate(doc);
        redis.setBuffer(`doc:${docId}`, Buffer.from(encoded));
      },

      onClose(_event, ws) {
        connections.get(docId)?.delete(ws.raw as any);
      },
    };
  }));

  return { app, injectWebSocket };
}

// src/client/offline-sync.ts
import * as Y from 'yjs';
import { IndexeddbPersistence } from 'y-indexeddb';
import { WebsocketProvider } from 'y-websocket';

export class OfflineFirstDoc {
  doc: Y.Doc;
  private indexeddb: IndexeddbPersistence;
  private wsProvider: WebsocketProvider | null = null;
  private syncUrl: string;

  constructor(docId: string, syncUrl: string) {
    this.doc = new Y.Doc();
    this.syncUrl = syncUrl;

    // Local persistence — works offline
    this.indexeddb = new IndexeddbPersistence(docId, this.doc);

    // When IndexedDB loads, connect to server
    this.indexeddb.on('synced', () => {
      this.connectWebSocket(docId);
    });

    // Auto-reconnect on network change
    window.addEventListener('online', () => this.connectWebSocket(docId));
    window.addEventListener('offline', () => this.disconnect());
  }

  private connectWebSocket(docId: string): void {
    if (this.wsProvider) return;
    this.wsProvider = new WebsocketProvider(this.syncUrl, docId, this.doc, {
      connect: true,
      maxBackoffTime: 10000,
    });

    this.wsProvider.on('status', (event: { status: string }) => {
      console.log(`Sync status: ${event.status}`);
    });
  }

  private disconnect(): void {
    this.wsProvider?.destroy();
    this.wsProvider = null;
  }

  // Get text field as observable
  getText(field: string): Y.Text {
    return this.doc.getText(field);
  }

  // Observe changes (for React/Vue binding)
  onChange(callback: () => void): () => void {
    const handler = () => callback();
    this.doc.on('update', handler);
    return () => this.doc.off('update', handler);
  }

  // Undo/redo support
  createUndoManager(fields: string[]): Y.UndoManager {
    const trackedTypes = fields.map(f => this.doc.getText(f));
    return new Y.UndoManager(trackedTypes);
  }
}

// src/client/checklist.ts
import * as Y from 'yjs';

// Reorder items (CRDT-safe: uses fractional indexing)
export function reorderChecklist(
  doc: Y.Doc,
  fromIndex: number,
  toIndex: number
): void {
  doc.transact(() => {
    const checklist = doc.getArray<Y.Map<any>>('checklist');
    const item = checklist.get(fromIndex);
    if (!item) return;

    // Clone the item data
    const data: Record<string, any> = {};
    for (const [key, value] of item.entries()) {
      data[key] = value;
    }

    // Remove from old position, insert at new
    checklist.delete(fromIndex);
    const newItem = new Y.Map<any>();
    for (const [key, value] of Object.entries(data)) {
      newItem.set(key, value);
    }
    checklist.insert(Math.min(toIndex, checklist.length), [newItem]);
  });
}

// Concurrent edits demo: two users toggle the same checkbox
// User A: toggles item 3 to "completed"
// User B: toggles item 3 to "completed" (independently, offline)
// Result: item 3 ends up "completed" (last-writer-wins on the boolean field)
// No conflict, no data loss, no manual resolution needed