real-time-collaboration-engine

实时协作引擎

擅长构建类似 Google Docs 的协作编辑功能，支持 WebSocket、冲突解决和在线状态感知。

适用场景

✅ 适用于：

• 协作式文本／代码编辑器
• 共享白板和设计工具
• 多人视频编辑时间线
• 实时数据仪表板
• 多人游戏状态同步

❌ 不适用于：

• 简单聊天应用（使用基础 WebSocket）
• 请求-响应式 API（使用 REST／GraphQL）
• 单用户应用
• 只读数据流（使用 Server-Sent Events）

快速决策树

Need real-time collaboration?
├── Text editing? → Operational Transform (OT)
├── JSON data structures? → CRDTs
├── Cursor tracking only? → Simple WebSocket + presence
├── Offline-first? → CRDTs (better offline merge)
└── No conflicts possible? → Basic broadcast

---

技术选型

冲突解决策略（2024）

策略	最适合	复杂度	离线支持
操作转换（OT）	文本、有序序列	高	有限
CRDT	JSON 对象、集合	中	出色
最后写入优先	简单状态	低	基础
三方合并	Git 风格编辑	高	良好

时间线：

• 2010：Google Wave 使用 OT
• 2014：Figma 采用 CRDT
• 2019：Yjs（CRDT 库）发布
• 2022：Automerge 2.0（CRDT 库）发布
• 2024：PartyKit 简化实时基础设施

---

常见反模式

反模式 1：广播每次按键

新手想法：“立即发送每次更改，以获得实时体验。”

问题：网络中充斥着细小消息，性能不佳。

错误做法：

// ❌ Sends message on every keystroke
function Editor() {
  const handleChange = (text: string) => {
    socket.emit('text-change', { text });  // Every keystroke!
  };

  return <textarea onChange={(e) => handleChange(e.target.value)} />;
}

错误原因：每分钟输入 100 个词 = 每分钟 500 条消息 = 网络拥塞。

正确做法：

// ✅ Batches changes every 200ms
function Editor() {
  const [pendingChanges, setPendingChanges] = useState<Change[]>([]);

  useEffect(() => {
    const interval = setInterval(() => {
      if (pendingChanges.length > 0) {
        socket.emit('text-batch', { changes: pendingChanges });
        setPendingChanges([]);
      }
    }, 200);

    return () => clearInterval(interval);
  }, [pendingChanges]);

  const handleChange = (change: Change) => {
    setPendingChanges(prev => [...prev, change]);
  };

  return <textarea onChange={handleChange} />;
}

影响：每分钟 500 条消息 → 每秒 5 条消息（减少 90%）。

---

反模式 2：没有冲突解决策略

问题：并发编辑会导致数据丢失或损坏。

表现：用户发现自己的更改消失，文档变得不一致。

错误做法：

// ❌ Last write wins, overwrites concurrent changes
socket.on('text-change', ({ userId, text }) => {
  setDocument(text);  // Loses concurrent edits!
});

错误原因：如果用户 A 和 B 同时编辑，其中一方的更改会丢失。

正确做法（OT）：

// ✅ Operational Transform for text
import { TextOperation } from 'ot.js';

socket.on('operation', ({ userId, operation, revision }) => {
  const transformed = transformOperation(
    operation,
    pendingOperations,
    revision
  );

  applyOperation(transformed);
  incrementRevision();
});

function transformOperation(
  incoming: Operation,
  pending: Operation[],
  baseRevision: number
): Operation {
  // Transform incoming against pending operations
  let transformed = incoming;
  for (const op of pending) {
    transformed = TextOperation.transform(transformed, op)[0];
  }
  return transformed;
}

正确做法（CRDT）：

// ✅ CRDT for JSON objects
import * as Y from 'yjs';

const ydoc = new Y.Doc();
const ytext = ydoc.getText('document');

// Automatically handles conflicts
ytext.insert(0, 'Hello');

// Sync with peers
const provider = new WebsocketProvider('ws://localhost:1234', 'room', ydoc);

影响：并发编辑可正确合并，不会丢失数据。

---

反模式 3：未处理断开连接

问题：用户离线后会丢失工作或看到过时状态。

错误做法：

// ❌ No offline handling
socket.on('disconnect', () => {
  console.log('Disconnected');  // That's it?!
});

错误原因：待处理的更改会丢失，没有重新连接策略，用户体验差。

正确做法：

// ✅ Queue changes offline, sync on reconnect
const [isOnline, setIsOnline] = useState(true);
const [offlineQueue, setOfflineQueue] = useState<Change[]>([]);

socket.on('disconnect', () => {
  setIsOnline(false);
  showToast('Offline - changes will sync when reconnected');
});

socket.on('connect', () => {
  setIsOnline(true);

  // Send queued changes
  if (offlineQueue.length > 0) {
    socket.emit('sync-offline-changes', { changes: offlineQueue });
    setOfflineQueue([]);
  }
});

const handleChange = (change: Change) => {
  if (isOnline) {
    socket.emit('change', change);
  } else {
    setOfflineQueue(prev => [...prev, change]);
  }
};

时间线背景：

• 2015：离线优先应用很少见
• 2020：PWA 让离线体验成为标准
• 2024：用户期待无缝离线编辑

---

反模式 4：仅同步客户端状态

问题：服务器没有权威性，客户端会失去同步。

错误做法：

// ❌ Clients broadcast to each other directly
socket.on('peer-change', ({ userId, change }) => {
  applyChange(change);  // No validation, no server state
});

错误原因：恶意客户端可以发送无效数据，失去同步后也无法恢复。

正确做法：

// ✅ Server is source of truth
// Client
socket.emit('operation', { operation, clientRevision });

socket.on('ack', ({ serverRevision }) => {
  if (serverRevision !== expectedRevision) {
    // Desync detected, request full state
    socket.emit('request-full-state');
  }
});

// Server
io.on('connection', (socket) => {
  socket.on('operation', ({ operation, clientRevision }) => {
    // Validate operation
    if (!isValid(operation)) {
      socket.emit('error', { message: 'Invalid operation' });
      return;
    }

    // Apply to server state
    const serverRevision = applyOperation(operation);

    // Broadcast to all clients
    io.emit('operation', { operation, serverRevision });
  });
});

影响：确保数据完整性，并且可以从客户端错误中恢复。

---

反模式 5：没有在线状态感知

问题：用户无法看到谁在编辑什么，导致编辑冲突。

表现：两个人在不知情的情况下编辑同一部分。

错误做法：

// ❌ No awareness of other users
function Editor() {
  return <textarea />;  // Flying blind!
}

正确做法：

// ✅ Show active users and cursors
import { usePresence } from './usePresence';

function Editor() {
  const { users, updateCursor } = usePresence();

  const handleCursorMove = (position: number) => {
    socket.emit('cursor-move', { userId: myId, position });
  };

  return (
    <div>
      {/* Show who's online */}
      <UserList users={users} />

      {/* Show remote cursors */}
      <EditorWithCursors
        content={content}
        cursors={users.map(u => u.cursor)}
        onCursorMove={handleCursorMove}
      />
    </div>
  );
}

功能：

• 显示带头像的活跃用户列表
• 按用户以不同颜色标示光标位置
• 高亮显示选区
• 显示“用户 X 正在输入……”提示

---

实现模式

模式 1：配置带重连功能的 WebSocket

import { io } from 'socket.io-client';

const socket = io('ws://localhost:3000', {
  reconnection: true,
  reconnectionDelay: 1000,
  reconnectionDelayMax: 5000,
  reconnectionAttempts: Infinity,
  transports: ['websocket', 'polling']  // Fallback
});

socket.on('connect', () => {
  console.log('Connected:', socket.id);
});

socket.on('disconnect', (reason) => {
  if (reason === 'io server disconnect') {
    // Server disconnected, manually reconnect
    socket.connect();
  }
});

socket.on('connect_error', (error) => {
  console.error('Connection error:', error);
});

模式 2：操作转换（文本）

import { TextOperation } from 'ot.js';

class OTEditor {
  private revision = 0;
  private pendingOperations: TextOperation[] = [];

  applyLocalOperation(op: TextOperation): void {
    // Apply immediately (optimistic update)
    this.applyToEditor(op);

    // Send to server
    this.sendOperation(op);

    // Store as pending
    this.pendingOperations.push(op);
  }

  receiveRemoteOperation(op: TextOperation, serverRevision: number): void {
    // Transform against pending operations
    let transformed = op;
    for (const pending of this.pendingOperations) {
      [transformed, pending] = TextOperation.transform(transformed, pending);
    }

    // Apply transformed operation
    this.applyToEditor(transformed);
    this.revision = serverRevision;
  }

  acknowledgeOperation(serverRevision: number): void {
    // Remove acknowledged operation from pending
    this.pendingOperations.shift();
    this.revision = serverRevision;
  }
}

模式 3：使用 Yjs 的 CRDT

import * as Y from 'yjs';
import { WebsocketProvider } from 'y-websocket';

// Create shared document
const ydoc = new Y.Doc();

// Define shared types
const ytext = ydoc.getText('content');
const ymap = ydoc.getMap('metadata');
const yarray = ydoc.getArray('users');

// Connect to sync server
const provider = new WebsocketProvider(
  'ws://localhost:1234',
  'room-name',
  ydoc
);

// Listen to changes
ytext.observe(event => {
  console.log('Text changed:', event.changes);
});

// Make changes (automatically synced)
ytext.insert(0, 'Hello ');
ytext.insert(6, 'World!');

// Undo/redo support
const undoManager = new Y.UndoManager(ytext);
undoManager.undo();
undoManager.redo();

模式 4：在线状态感知

import { Awareness } from 'y-protocols/awareness';

const awareness = provider.awareness;

// Set local state
awareness.setLocalState({
  user: {
    name: 'Alice',
    color: '#ff0000',
    cursor: { line: 10, ch: 5 }
  }
});

// Listen to changes
awareness.on('change', ({ added, updated, removed }) => {
  // Update UI with user cursors/selections
  const states = awareness.getStates();
  states.forEach((state, clientId) => {
    if (clientId !== awareness.clientID) {
      renderCursor(state.user.cursor, state.user.color);
    }
  });
});

模式 5：带回滚的乐观更新

class OptimisticEditor {
  private optimisticChanges = new Map<string, Change>();

  async applyChange(change: Change): Promise<void> {
    const changeId = generateId();

    // Apply immediately (optimistic)
    this.applyToUI(change);
    this.optimisticChanges.set(changeId, change);

    try {
      // Send to server
      const result = await this.sendToServer(change);

      // Success - remove from optimistic
      this.optimisticChanges.delete(changeId);

    } catch (error) {
      // Failed - rollback
      this.rollback(changeId);
      this.showError('Could not apply change');
    }
  }

  private rollback(changeId: string): void {
    const change = this.optimisticChanges.get(changeId);
    if (change) {
      this.revertInUI(change);
      this.optimisticChanges.delete(changeId);
    }
  }
}

---

生产环境检查清单

□ WebSocket connection with auto-reconnect
□ Offline queue for pending changes
□ Conflict resolution strategy (OT or CRDT)
□ Server authority (clients can't desync)
□ Presence awareness (cursors, active users)
□ Optimistic updates with rollback
□ Change batching (not per-keystroke)
□ Message compression for large payloads
□ Authentication and authorization
□ Rate limiting (prevent spam)
□ Heartbeat/ping-pong to detect dead connections
□ Graceful degradation (falls back to polling if WebSocket fails)

---

适用与不适用场景

场景	策略
文本编辑（Google Docs）	✅ 操作转换
JSON 对象（Figma）	✅ CRDT（Yjs、Automerge）
简单的光标共享	✅ 基础 WebSocket + 在线状态
聊天消息	✅ 简单的仅追加模式（无需 OT／CRDT）
视频时间线编辑	✅ 时间线使用 CRDT，文本使用 OT
只读仪表板	❌ 改用 Server-Sent Events

---

参考资料

• /references/ot-vs-crdt.md - 冲突解决策略的深入比较
• /references/websocket-scaling.md - 扩展到数百万个并发连接
• /references/presence-patterns.md - 光标跟踪、用户感知、活动指示器

脚本

• scripts/collaboration_tester.ts - 模拟并发编辑，测试冲突解决
• scripts/latency_simulator.ts - 测试高延迟／丢包情况下的行为

---

此技能适用于：实时协作 | WebSocket 架构 | 操作转换 | CRDT | 在线状态感知 | 冲突解决