实时协作 CRDT 原理

class GCounter {
  constructor(replicaId) {
    this.id = replicaId;
    this.counts = new Map([[replicaId, 0]]);
  }

  increment(n = 1) {
    this.counts.set(this.id, (this.counts.get(this.id) || 0) + n);
  }

  value() {
    let sum = 0;
    for (const v of this.counts.values()) sum += v;
    return sum;
  }

  merge(other) {
    for (const [id, v] of other.counts) {
      this.counts.set(id, Math.max(this.counts.get(id) || 0, v));
    }
  }
}

const a = new GCounter("A");
const b = new GCounter("B");
a.increment(3);
b.increment(2);
a.merge(b);
b.merge(a);
console.log(a.value(), b.value()); // 5 5，收敛一致

class LWWRegister {
  constructor() {
    this.value = null;
    this.timestamp = -Infinity;
  }

  set(value, timestamp) {
    if (timestamp > this.timestamp) {
      this.value = value;
      this.timestamp = timestamp;
    }
  }

  merge(other) {
    if (other.timestamp > this.timestamp) {
      this.value = other.value;
      this.timestamp = other.timestamp;
    }
  }
}

// RGA 节点的简化表示
class RGANode {
  constructor(id, char, originId) {
    this.id = id;            // 全局唯一，如 "A:5"
    this.char = char;
    this.originId = originId; // 插入位置的左邻居
    this.deleted = false;
  }
}

function mergeRGA(local, remote) {
  const seen = new Set(local.map((n) => n.id));
  for (const node of remote) {
    if (!seen.has(node.id)) {
      local.push(node);
      seen.add(node.id);
    }
  }
  // 按 originId 与 id 排序，保证所有副本顺序一致
  local.sort((a, b) => {
    if (a.originId === b.originId) return a.id < b.id ? -1 : 1;
    return a.originId < b.originId ? -1 : 1;
  });
  return local;
}

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

const doc = new Y.Doc();
const text = doc.getText("content");

// 本地编辑立即生效，无需等待网络
text.insert(0, "Hello ");

// 远端更新自动合并
text.observe((event) => {
  console.log("文档变化，变更长度:", event.delta);
});

// 接入 WebSocket 同步
const provider = new WebsocketProvider("wss://example.com", "room-1", doc);
provider.on("status", (event) => {
  console.log("连接状态:", event.status); // connected / disconnected
});

// 手动编码状态，用于持久化或通过其他通道传输
const update = Y.encodeStateAsUpdate(doc);
Y.applyUpdate(doc, update);

const doc = new Y.Doc({ gc: true }); // 默认开启

// 生成快照
const snapshot = Y.snapshot(doc);
const stateVector = Y.encodeStateVector(doc);

// 从快照恢复
const restored = Y.createDocFromSnapshot(doc, snapshot);

// 状态向量：记录每个副本已见过的最新时钟
function encodeStateVector(doc) {
  return Y.encodeStateVector(doc);
}

// 对端根据我的状态向量，生成我缺失的差量更新
function diffForPeer(doc, remoteStateVector) {
  return Y.encodeStateAsUpdate(doc, remoteStateVector);
}

// 应用差量更新
function applyRemoteUpdate(doc, update) {
  Y.applyUpdate(doc, update, "remote");
}

// 监听本地更新并广播
doc.on("update", (update, origin) => {
  if (origin === "remote") return; // 避免回环
  broadcast(update);
});

const pc = new RTCPeerConnection();
const channel = pc.createDataChannel("crdt", { ordered: true });

channel.onopen = () => {
  // 连接建立后先交换状态向量
  channel.send(JSON.stringify({ type: "sv", sv: Array.from(Y.encodeStateVector(doc)) }));
};

channel.onmessage = (event) => {
  const msg = JSON.parse(event.data);
  if (msg.type === "sv") {
    const remoteSV = new Uint8Array(msg.sv);
    const diff = Y.encodeStateAsUpdate(doc, remoteSV);
    channel.send(JSON.stringify({ type: "update", data: Array.from(diff) }));
  } else if (msg.type === "update") {
    Y.applyUpdate(doc, new Uint8Array(msg.data), "remote");
  }
};

doc.on("update", (update) => {
  if (channel.readyState === "open") {
    channel.send(JSON.stringify({ type: "update", data: Array.from(update) }));
  }
});