CRDT 详解

// 用户 A 和用户 B 同时编辑
// 场景：两人都想修改同一段文字

用户A: "Hello World" → "Hello Universe"
用户B: "Hello World" → "Hello Everyone"

// 传统方式会出现冲突，需要服务器裁决
// 结果不确定，取决于谁先提交

// CRDT 知道每个字符的来源和顺序
// 即使同时编辑，也能自动合并

最终结果: "Hello Universe and Everyone"
// 或自动排序: "Hello Everyone and Universe"

// 传统方式：按位置操作（有问题）
{
  operation: "insert",
  position: 5,
  text: "Beautiful "
}

// CRDT：按 ID 操作（无冲突）
{
  operation: "insert",
  id: { clientId: 1, clock: 15 },  // 全局唯一
  text: "Beautiful "
}

// 每个客户端维护自己的时钟
interface ClientState {
  clientId: number;      // 客户端 ID
  clock: number;         // 逻辑时钟
  
  // 每次操作前先递增时钟
  nextClock(): number {
    return ++this.clock;
  }
}

// 客户端 A: ID=1, clock=10
// 客户端 B: ID=2, clock=10

// A 插入: { clientId: 1, clock: 10 }
// B 插入: { clientId: 2, clock: 10 }

// 时钟比较: (1, 10) vs (2, 10)
// 如果 ID 不同，按 ID 排序
// 结果: A 的内容在前，B 的内容在后

// CRDT 使用偏序关系确定操作顺序
function compare(a: OpId, b: OpId): number {
  // 1. 时钟小的在前
  if (a.clock !== b.clock) return a.clock - b.clock;
  
  // 2. 时钟相同，ID 小的在前
  return a.clientId - b.clientId;
}

// 示例
const op1 = { clientId: 1, clock: 5 };
const op2 = { clientId: 2, clock: 5 };
const op3 = { clientId: 1, clock: 6 };

// 排序结果: op1 < op2 < op3

// 传输的操作
interface Operation {
  type: 'insert' | 'delete';
  id: OpId;
  value?: string;        // insert 时需要
  length?: number;       // delete 时需要
}

// 发送方
function sendOperation(op: Operation) {
  broadcast(op);  // 广播给所有客户端
}

// 接收方
function receiveOperation(op: Operation) {
  apply(op);      // 应用操作
}

// 状态合并
interface State {
  version: number;
  data: Map<string, string>;
}

// 合并两个状态
function merge(state1: State, state2: State): State {
  const result = { ...state1 };
  
  // 取版本号最大的
  result.version = Math.max(state1.version, state2.version);
  
  // 合并数据（后写入覆盖先写入）
  for (const [key, value] of state2.data) {
    result.data.set(key, value);
  }
  
  return result;
}

class GSet<T> {
  private items = new Set<T>();
  
  add(item: T) {
    this.items.add(item);
    return this;
  }
  
  merge(other: GSet<T>): GSet<T> {
    const result = new GSet<T>();
    // 合并两个集合
    for (const item of this.items) result.add(item);
    for (const item of other.items) result.add(item);
    return result;
  }
}

class TwoPSet<T> {
  private added = new Set<T>();      // 添加集
  private removed = new Set<T>();    // 删除集
  
  add(item: T) {
    this.added.add(item);
  }
  
  remove(item: T) {
    this.removed.add(item);
  }
  
  contains(item: T): boolean {
    return this.added.has(item) && !this.removed.has(item);
  }
  
  merge(other: TwoPSet<T>): TwoPSet<T> {
    const result = new TwoPSet<T>();
    result.added = new Set([...this.added, ...other.added]);
    result.removed = new Set([...this.removed, ...other.removed]);
    return result;
  }
}

class LWWRegister<T> {
  private value: T;
  private timestamp: number;
  private owner: string;
  
  set(value: T, owner: string) {
    this.timestamp = Date.now();
    this.value = value;
    this.owner = owner;
  }
  
  merge(other: LWWRegister<T>): LWWRegister<T> {
    if (other.timestamp > this.timestamp) {
      return other;
    }
    return this;
  }
}

// Yjs 的文本实现
class YText {
  private chars = new Map<string, Char>();
  
  insert(index: number, text: string, clientId: number) {
    // 生成唯一 ID
    const id = {
      clientId,
      clock: this.nextClock++,
    };
    
    // 创建字符对象
    for (const char of text) {
      const charId = { ...id, offset: index++ };
      this.chars.set(toString(charId), { id: charId, content: char });
    }
    
    // 按 ID 排序存储
    this.reindex();
  }
  
  delete(index: number, length: number) {
    // 软删除：标记删除
    const ids = this.getIdsAtIndex(index, length);
    for (const id of ids) {
      this.chars.get(toString(id)).deleted = true;
    }
  }
  
  merge(other: YText): YText {
    // 合并两个文本
    const merged = new YText();
    
    // 遍历所有字符
    for (const [id, char] of this.chars) {
      merged.chars.set(id, char);
    }
    
    for (const [id, char] of other.chars) {
      // 冲突解决：ID 大的在后
      if (merged.chars.has(id)) {
        // 保留更新的版本
        if (char.timestamp > merged.chars.get(id).timestamp) {
          merged.chars.set(id, char);
        }
      } else {
        merged.chars.set(id, char);
      }
    }
    
    return merged;
  }
  
  toString(): string {
    // 按 ID 排序后拼接
    return Array.from(this.chars.values())
      .filter(c => !c.deleted)
      .sort((a, b) => compareIds(a.id, b.id))
      .map(c => c.content)
      .join('');
  }
}

import * as Y from 'yjs';

// 创建文档
const ydoc = new Y.Doc();

// 获取共享类型
const yText = ydoc.getText('content');      // 共享文本
const yArray = ydoc.getArray('items');      // 共享数组
const yMap = ydoc.getMap('metadata');       // 共享 Map

// 监听变更
ydoc.on('update', (update: Uint8Array, origin: unknown) => {
  console.log('文档更新:', update);
});

// 事务
ydoc.transact(() => {
  yText.insert(0, 'Hello');
  yArray.push([1, 2, 3]);
  yMap.set('key', 'value');
});

const yText = ydoc.getText('content');

// 插入
yText.insert(0, 'Hello World');

// 删除
yText.delete(0, 5);

// 获取内容
console.log(yText.toString()); // "World"

// 监听
yText.observe(event => {
  console.log('操作类型:', event.transaction.local ? '本地' : '远程');
  console.log('插入:', event.delta.filter(d => d.insert));
  console.log('删除:', event.delta.filter(d => d.delete));
});

// 撤销
const undoManager = new Y.UndoManager(yText);
undoManager.undo();
undoManager.redo();

const yArray = ydoc.getArray('items');

// 添加
yArray.push([1, 2, 3]);        // 末尾添加
yArray.unshift(0);             // 开头插入

// 获取
console.log(yArray.toArray()); // [0, 1, 2, 3]

// 遍历
yArray.forEach(item => console.log(item));

// 监听
yArray.observe(event => {
  console.log('插入:', event.changes.added);
  console.log('删除:', event.changes.deleted);
});

const yMap = ydoc.getMap('metadata');

// 设置
yMap.set('title', 'My Doc');
yMap.set('author', 'John');

// 获取
console.log(yMap.get('title')); // "My Doc"

// 删除
yMap.delete('author');

// 遍历
yMap.forEach((value, key) => {
  console.log(`${key}: ${value}`);
});

// Google Docs 使用的 OT
// 需要服务器按顺序处理操作

// 操作序列必须严格排序
OpA → OpB → OpC

// 如果顺序错了，结果就错了
OpA → OpC → OpB  // 可能导致不一致

// CRDT：操作顺序不影响最终结果
// 两个操作 A 和 B，无论以什么顺序应用，结果都相同

// 应用 A 然后 B
apply(A);
apply(B);  // 结果: "AB"

// 应用 B 然后 A
apply(B);
apply(A);  // 结果: "AB"

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

// 1. 创建文档
const ydoc = new Y.Doc();

// 2. 连接 WebSocket
const provider = new WebsocketProvider(
  'wss://demos.yjs.dev',
  'my-room-name',
  ydoc
);

// 3. 获取共享文本
const yText = ydoc.getText('editor');

// 4. 创建编辑器绑定
const editor = new Editor({
  extensions: [
    // 将 Yjs 绑定到编辑器
    Collaboration.configure({
      document: ydoc,
    }),
  ],
});

// 5. 监听用户状态
provider.awareness.setLocalState({
  user: {
    name: 'Alice',
    color: '#ff0000',
  },
});

provider.awareness.on('change', () => {
  console.log('在线用户:', provider.awareness.getStates());
});

import * as Y from 'yjs';
import { IndexeddbPersistence } from 'y-indexeddb';

// 创建文档
const ydoc = new Y.Doc();

// 本地持久化
const persistence = new IndexeddbPersistence('my-app', ydoc);

persistence.on('synced', () => {
  console.log('本地数据加载完成');
});

// 离线编辑
ydoc.getText('notes').insert(0, '这是在离线时编辑的');

// 重新上线后同步
// IndexeddbPersistence 会自动处理同步

// Yjs 压缩
const ydoc = new Y.Doc();

// 获取状态向量
const stateVector = Y.encodeStateVector(ydoc);

// 压缩到特定版本
const compressed = Y.encodeStateAsUpdate(ydoc, stateVector);

// 合并更新
Y.applyUpdate(ydoc, compressed);

// 清理已删除的内容
const gc = new Y.Doc();
Y.gc(gc);

// 清理特定文档
Y.mergeUpdates([
  Y.encodeStateAsUpdate(ydoc1),
  Y.encodeStateAsUpdate(ydoc2),
]);