技術文章 » 初探富文字之CRDT協同範例

初探富文字之CRDT協同範例

2023-03-05 18:00:18

初探富文字之CRDT協同範例

在前邊初探富文字之CRDT協同演演算法一文中我們探討了為什麼需要協同、分散式的最終一致性理論、偏序集與半格的概念、為什麼需要有偏序關係、如何通過資料結構避免衝突、分散式系統如何進行同步排程等等，這些屬於完成協同所需要了解的基礎知識，實際上當前有很多成熟的協同實現，例如automerge、yjs等等，本文就是關注於以yjs為CRDT協同框架來實現協同的範例。

描述

接入協同框架實際上並不是一件簡單的事情，當然相對於接入OT協同而言接入CRDT協同已經是比較簡單的了，因為我們只需要聚焦於資料結構的使用就好，而不需要對變換有過多的關注。當前我們更加關注的是Op-based CRDT，本文所說的CRDT也是特指的Op-based CRDT，畢竟State-baed CRDT需要將全量資料進行傳輸，每次都要完整傳輸狀態來完成同步讓它比較難變成通用的解決方案。因此與OT演演算法一樣，我們依然需要Operation，在富文字領域，最經典的Operation有quill的delta模型，通過retain、insert、delete三個操作完成整篇檔案的描述與操作，還有slate的JSON模型，通過insert_text、split_node、remove_text等等操作來完成整篇檔案的描述與操作。假如此時是OT的話，接下來我們就要聊到變換Transformation了，但是使用CRDT演演算法的情況下，我們的關注點變了，我們需要做的是關注於如何將我們現在的資料結構轉換為CRDT框架的資料結構，比如通過框架提供的Array、Map、Text等型別構建我們自己的JSON資料，並且我們的Op也需要對映到對框架提供的資料結構進行的操作，這樣框架便可以幫我們進行協同，當框架完成協同之後把框架的資料結構的改變返回，此時我們需要再將這部分改變對映到我們自己的Op，然後我們只需要在本地應用這些遠端同步並在本地轉換的Op，就可以做到協同了。

上邊這個資料轉換聽起來是不是有點耳熟，在前邊初探富文字之OT協同範例中，我們介紹了json0，我們也提到了一個可行的操作，我們讓變換Transformation這部分讓json0去做，我們需要關注的是從我們自己定義的資料結構轉換到json0，在json0進行變換操作之後我們同樣地將Op轉換後應用到我們原生的資料就好。雖然原理是完全不同的，但是我們在已有成熟框架的情況下似乎並不需要關注這點，我們更側重於使用，實際上在使用起來是很像的。此時假設我們有一個自研的思維導圖功能需要實現協同，而儲存的資料結構都是自定義的，沒有直接可以呼叫的實現方案，我們就需要進行轉換適配，那麼如果使用OT的話，並且藉助json0做變換，那麼我們需要做的是把Op轉換為json0的Op，傳送的資料也將會是這個json0的Op，那麼如果直接使用CRDT的話，我們更像是通過框架定義的資料結構將Op應用到資料結構上，傳送的資料是框架定義的資料，類似於將Op應用到資料結構上了，其他的操作都由框架給予完整的支援了。實際上通過框架提供的例子後，接入CRDT協同就主要是理解並且實現的問題了，這樣就有一個大體的實現方向了，而不是毫無頭緒不知道應該從哪裡開始做協同。另外還是那個宗旨，合適的才是最好的，要考慮到實現的成本問題，沒有必要硬套資料結構的實現，OT有OT的優點，CRDT有CRDT的優點，CRDT這類方法相比OT還比較年輕，還是在不斷髮展過程中的，實際上有些問題例如記憶體佔用、速度等問題最近幾年才被比較好的解決，ShareDB作者在關注CRDT不斷髮展的過程中也說了CRDTs are the future。此外從技術上講，CRDT型別是OT型別的子集，也就是說，CRDT實際上是不需要轉換函數的OT型別，因此任何可以處理這些OT型別的東西也應該能夠使用CRDT。

或許上邊的一些概念可能一時間讓人難以理解，所以下面的Counter與Quill兩個範例就是介紹瞭如何使用yjs實現協同，究竟如何通過資料結構完成協同的接入工作，當然具體的API呼叫還是還是需要看yjs的檔案，本文只涉及到最基本的協同操作，所有的程式碼都在https://github.com/WindrunnerMax/Collab中，注意這是個pnpm的workspace monorepo專案，要注意使用pnpm安裝依賴。

Counter

首先我們執行一個基礎的協同範例Counter，實現的主要功能是在多個使用者端可以+1的情況下我們可以維護同一份計數器總數，該範例的地址是https://github.com/WindrunnerMax/Collab/tree/master/packages/crdt-counter，首先簡單看一下目錄結構(tree --dirsfirst -I node_modules):

crdt-counter
├── public
│   ├── favicon.ico
│   └── index.html
├── server
│   └── index.ts
├── src
│   ├── client.ts
│   ├── counter.tsx
│   └── index.tsx
├── babel.config.js
├── package.json
├── rollup.config.js
├── rollup.server.js
└── tsconfig.json

先簡略說明下各個資料夾和檔案的作用，public儲存了靜態資原始檔，在使用者端打包時將會把內容移動到build資料夾，server資料夾中儲存了CRDT伺服器端的實現，在執行時同樣會編譯為js檔案放置於build資料夾下，src資料夾是使用者端的程式碼，主要是檢視與CRDT使用者端的實現，babel.config.js是babel的設定資訊，rollup.config.js是打包使用者端的組態檔，rollup.server.js是打包伺服器端的組態檔，package.json與tsconfig.json大家都懂，就不贅述了。

在前邊CRDT協同演演算法實現一文中，我們經常提到的就是無需中央伺服器的分散式協同，那麼在這個例子中我們就來實現一個peer-to-peer的範例。yjs提供了一個y-webrtc的信令伺服器，甚至還有公共的信令伺服器可以用，當然可能因為網路的關係這個公共的信令伺服器在國內不是很適用。在繼續完成協同之前，我們還需要了解一下WebRTC以及信令的相關概念。

WebRTC是一種實時通訊技術，重點在於可以對等即P2P通訊，其允許瀏覽器和應用程式直接在網際網路上傳輸音訊、視訊和資料流，無需使用中間伺服器進行中轉。WebRTC利用瀏覽器內建的標準API和協定來提供這些功能，並且支援多種編解碼器和平臺，WebRTC可以用於開發各種實時通訊應用，例如線上會議、遠端共同作業、實時廣播、線上遊戲和IoT應用等。但是在多級NAT網路環境下，P2P連線可能會受到限制，簡單來說就是一臺裝置無法直接發現另一臺裝置，自然也就沒有辦法進行P2P通訊，這時需要使用特殊的技術來繞過NAT並建立P2P連線。

NAT Network Address Translation網路地址轉換是一種在IP網路中廣泛使用的技術，主要是將一個IP地址轉換為另一個IP地址，具體來說其工作原理是將一個私有IP地址(如在家庭網路或企業內部網路中使用的地址)對映到一個公共IP地址(如網際網路上的IP地址)。當一個裝置從私有網路向公共網路傳送封包時，NAT裝置會將源IP地址從私有地址轉換為公共地址，並且在返回封包時將目標IP地址從公共地址轉換為私有地址。NAT可以通過多種方式實現，例如靜態NAT、動態NAT和埠地址轉換PAT等，靜態NAT將一個私有IP地址對映到一個公共IP地址，而動態NAT則動態地為每個私有地址分配一個公共地址，PAT是一種特殊的動態NAT，在將私有IP地址轉換為公共IP地址時，還會將源埠號或目標埠號轉換為不同的埠號，以支援多個裝置使用同一個公共IP地址。NAT最初是為了解決IPv4地址空間的短缺而設計的，後來也為提高網路安全性並簡化網路管理提供了基礎。

在網際網路上大多數裝置都是通過路由器或防火牆連線到網路的，這些裝置通常使用網路地址轉換NAT將內部IP地址對映到一個公共的IP地址上，這個公共IP地址可以被其他裝置用來存取，但是這些裝置內部的IP地址是隱藏的，其他的裝置不能直接通過它們的內部IP地址建立P2P連線。因此，直接進行P2P連線可能會受到網路地址轉換NAT的限制，導致連線無法建立。為了解決這個問題，需要使用一些技術來繞過NAT並建立P2P連線。另外，P2P連線也需要一些控制和協調機制，以確保連線的可靠性和安全性。

信令可以用來解決多級NAT環境下的P2P連線問題，當兩個裝置嘗試建立P2P連線時，可以使用信令伺服器來交換網路資訊，例如IP地址、埠和協定型別等，以便裝置之間可以彼此發現並建立連線。當然信令伺服器並不是繞過NAT的唯一解決方案，STUN、TURN和ICE等技術也可以幫助解決這個問題。信令伺服器的主要作用是協調不同裝置之間的連線，以確保裝置可以正確地發現和通訊。在實際應用中，通常需要同時使用多種技術和工具來解決多級NAT環境下的P2P連線問題。

那麼回到WebRTC，我們即使是使用了P2P的技術，但是不可避免的需要一個信令伺服器來交換WebRTC對談描述和控制資訊。當然這些資訊不包括實際通訊的資料流本身，而是用於描述和控制這些流的方式和引數，這些資料流本身是通過對等連線在兩個瀏覽器之間直接傳輸的。主要資料流的通訊不經過中央伺服器，這就使得WebRTC有著低延遲和高頻寬等優點，但是同樣的因為每個對等點相互連線，不適合單個檔案上的大量共同作業者。

接下來我們要進行資料結構的設計，目前在yjs中是沒有Y.Number這個資料結構的，也就是說yjs沒有自增自減的操作，這點就與前邊OT範例不一樣了，所以在這裡我們需要設計資料結構。網路是不可靠的，我們不能夠在本地模擬+1的操作，就是說本地先取得值，然後進行+1操作之後再把值推到其他的使用者端上，這樣的設計雖然在本地測試應該是可行的，但是由於網路不可靠，我們不能保證本地取值的時候獲得的是最新的值，所以這個方案是不可靠的。

那麼我們思考幾種方案來實現這一點，有一種可行的方案是類似於我們之前介紹的CRDT資料結構，我們可以構造一個集合Y.Array，當我們點+1的時候，就向集合中push一個新的值，這樣再取和的時候直接取集合長度即可。

Y.Array: [] => +1 => [1] => +1 => [1, 1] => ...
Counter: [1, 1].size = N

另一種方案是使用Y.Map來完成，當用戶加入我們的P2P組的時候，我們通過其身份資訊為其分配一個id，然後這個id只記錄與自增自己的值，也就是說當某個使用者端點選+1的時候，操作的只有其id對應的數，而不能影響組網內其他的使用者的值。

Y.Map: {} => +1 => {"id": 1} => +1 => {"id": 2} => ...
Counter: Object.values({"id": 2}).reduce((a, b) => a + b) = N

在這裡我們使用的是Y.Map的方案，畢竟如果是Y.Array的話佔用資源會是比較大的，當然因為範例中並沒有身份資訊，每次進入的時候都是會隨機分配id的，當然這不會影響到我們的Counter。此外還有比較重要的一點是，因為我們是直接進行P2P通訊的，當所有的裝置都離線的時候，由於沒有設計實際的資料儲存機制，所以資料會丟失，這點也是需要注意的。

接下來我們看看程式碼的實現，首先我們來看看伺服器端，這裡主要實現是呼叫了一下y-webrtc-signaling來啟動一個信令伺服器，這是y-webrtc給予的開箱即用的功能，也可以基於這些內容進行改寫，不過因為是信令伺服器，除非有著很高的穩定性、客製化化等要求，否則直接當作開箱即用的信令伺服器就好。後邊主要是使用了express啟動了一個靜態資源伺服器，因為直接在瀏覽器開啟檔案的file協定有很多的安全限制，所以需要一個HTTP Server。

import { exec } from "child_process";
import express from "express";

// https://github.com/yjs/y-webrtc/blob/master/bin/server.js
exec("PORT=3001 npx y-webrtc-signaling", (err, stdout, stderr) => { // 呼叫`y-webrtc-signaling`
  console.log(stdout, stderr);
});

const app = express(); // 範例化`express`
app.use(express.static("build")); // 使用者端打包過後的靜態資源路徑
app.listen(3000);
console.log("Listening on http://localhost:3000");

在使用者端方面主要是定義了一個定義了一個共用的連結，通過id來加入我們的P2P組，並且還有密碼的保護，這裡需要連結的信令伺服器也就是上邊啟動的y-webrtc的3001埠的信令服務。之後我們通過observe定義的Y.Map資料結構的變化來執行回撥，在這裡實際上就是將回撥過後的整個Map資料傳回回撥函數，然後在檢視層進行Counter的計算，這裡還有一個transaction.origin判斷是為了防止我們原生的呼叫觸發回撥。最後我們定義了一個increase函數，在這裡我們通過transact作為事務來執行set操作，因為我們之前的設計只會處理我們當前使用者端對應的id的那個值，原生的值是可信的，直接自增即可，transact最後一個引數也就是上邊提到了的transaction.origin，可以用來判斷事件的來源。

import { Doc, Map as YMap } from "yjs";
import { WebrtcProvider } from "y-webrtc";

const getRandomId = () => Math.floor(Math.random() * 10000).toString();
export type ClientCallback = (record: Record<string, number>) => void;

class Connection {
  private doc: Doc;
  private map: YMap<number>;
  public id: string = getRandomId(); // 當前使用者端生成的唯一`id`
  public counter = 0; // 當前使用者端的初始值

  constructor() {
    const doc = new Doc();
    new WebrtcProvider("crdt-example", doc, { // `P2P`組名稱 // `Y.Doc`範例
      password: "room-password", // `P2P`組密碼
      signaling: ["ws://localhost:3001"], // 信令伺服器
    });
    const yMapDoc = doc.getMap<number>("counter"); // 獲取資料結構
    this.doc = doc;
    this.map = yMapDoc;
  }

  bind(cb: ClientCallback) {
    this.map.observe(event => { // 監聽資料結構變化 // 如果是多層巢狀需要`observeDeep`
      if (event.transaction.origin !== this) { // 防止本地修改時觸發
        const record = [...this.map.entries()].reduce( // 獲取`Y.Map`定義中的所有資料
          (cur, [key, value]) => ({ ...cur, [key]: value }),
          {} as Record<string, number>
        );
        cb(record); // 執行回撥
      }
    });
  }

  public increase() {
    this.doc.transact(() => { // 事務
      this.map.set(this.id, ++this.counter); // 自增本地`id`對應的值
    }, this); // 來源
  }
}

export default new Connection();

Quill

在執行富文字的範例Quill之前，我們不妨先來簡單討論一下是如何在富文字上應用的CRDT，在前文CRDT協同演演算法中主要討論的是分散式與CRDT的原理，並沒有涉及具體的富文字該如何設計資料結構，那麼在這裡我們簡單討論下yjs在富文字上應用CRDT的設計。看之前描述那一節的時候我們可能會產生一些有趣的想法，或許我們可以這麼來做，可以通過底層來實現OT，之後在上層封裝一層資料結構供外部使用的方式，從而對外看起來像是CRDT。當然原理上是不會這麼做的，因為這樣失去了擁抱CRDT的意義，可能會有部分借鑑實現的思路，但是不會直接這麼做的。

首先我們可以回憶一下CRDT在集合這個資料結構上的設計，我們主要考慮到了集合的新增和刪除如何完整的保證交換律、結合律、冪等律，那麼現在在富文字的實現上，我們不僅需要考慮到插入和刪除，需要考慮到順序的問題，並且我們還需要保證CCI，即最終一致性、因果一致性、意圖一致性，當然還需要考慮到Undo/Redo、遊標同步等相關的問題。

那麼我們首先來看看如何保證插入資料的順序，對於OT而言是通過索引得知使用者要操作的位置，並且通過變換來確保最終一致性，那麼CRDT是不需要這麼做的，上邊也提到過完全靠OT的話可能就失去了擁抱CRDT的意義，那麼如何確保要插入的位置正確呢，CRDT不靠索引的話就需要靠資料結構來完成這點，我們可以通過相對位置來完成，例如我們目前有AB字串，此時在中間插入了C字元，那麼這個字元就需要被標記為在A之後，在B之前，那麼很顯然，我們需要為每個字元都分配唯一的id，否則我們是無法做到這一點的，當然這塊實際上還有優化空間，在這裡就先不談這點，那麼由此我們通過相對位置保證了插入的順序。

接下來我們再看看刪除的問題，在前文的Observed-Remove Set集合資料結構中我們是可以真正的進行刪除操作的，而在這裡由於我們是通過相對位置來實現完整的順序，所以實際上我們是不能夠真正地將我們標記的Item進行刪除的，Item可以理解為插入的字元，也就是所謂的軟刪除。舉個例子，目前我們有AB字串，其中一個使用者端刪除了B，另一個使用者端同時在A與B之間增加了C，那麼此時這兩個Op同步到了第三個使用者端，那麼假如增加了C這個操作先到並且執行了，再刪除了B，那麼沒有問題，可是假設我們先刪除了B，再增加了C，那麼這個C我們就不能夠找到他要插入的位置，因為B已經被刪除了，我們是要在A與B之間去插入C的，那麼這樣這個操作就無法執行下去了，由此這樣其實就導致了操作不滿足交換律，那麼這就不能真的作為CRDT的資料結構設計了。其實我們可能會想，為什麼需要兩個位置來保證插入的字元位置，完全可以用B的左側或者A的右側來完成，實際上思考一下這是同樣的問題，多個使用者端來操作的話假如一個刪除了A另一個刪除了B，那麼便無論如何也找不到插入的位置了，這是不滿足交換律和結合律的操作，就不能作為CRDT的實現了。因此為了衝突的解決yjs並沒有真正的刪除Item，而是採用了標記的形式，即刪除的Item會被加入一個deleted標記，那麼不刪除會造成一個明顯的問題，空間的佔用會無限增長，因此yjs引入了墓碑機制，當確認了內容不會再被幹涉之後，將物件的內容替換為空的墓碑物件。

上邊也提到了衝突的問題，很明顯在設計上是存在衝突的問題的，因為CRDT實際上並不是完全為了協同編輯的場景而專門設計的，其主要是為了解決分散式場景中的一致性問題，所以在應用到協同編輯的場景中，不可避免地會出現衝突的問題，實際上這個衝突主要是為了集合順序的引入而導致的，要是不關心順序，那麼自然就不會出現衝突問題了。那麼為了使資料能夠滿足三律，在前文我們引入了一個偏序的概念，但是在協同編輯設計中，使用偏序不能夠保證資料同步的正確性和一致性，因為其無法處理一些關鍵的衝突情況，舉一個簡單的例子，假設我們此時有AB字串，如果一個使用者端在AB中加入了C，另一個加入了D，那麼究竟誰在前呢，所以我們需要引入全序的方法，即任意兩個Item都是可以比較的。那麼很明顯的，如果我們為每個Item附加上時間戳的元資訊，便可以引入全序了，但是實際上由於不同的使用者端可能具有不同的時鐘偏差，網路延遲和時鐘不同步等問題也可能導致時間戳不可靠。那麼相比之下，邏輯時鐘或者邏輯時間戳可以使用更簡單和可靠的方式來維護事件的順序:

  • 每次發生本地事件時，clock = clocl + 1。
  • 每次接收到遠端事件時，clock = max(clock, remoteClock) + 1。

看起來依舊會有發生衝突的可能，那麼我們可以再引入一個使用者端的唯一id，也就是clientID。這種機制看似簡單，但實際上使我們獲得了數學上性質良好的全序結構，這意味著我們可以在任意兩個Item之間對比獲得邏輯上的先後關係，這對保證CRDT演演算法的正確性相當重要。此外，通過這種方式我們也可以保證因果一致性，假如此時我們有兩個操作a、b如果有因果關係，那麼a.clock一定大於b.clock，這樣的得到的順序一定是滿足因果關係的，當然如果沒有因果關係，就可以取任意的順序執行了。舉個例子，我們有三個使用者端A、B、C以及字串SE，A在SE中間新增了a字元，此時這個操作同步到了B，B將a字元給刪除了，假設此時C先收到了B的刪除操作，因為這個操作依賴於A的操作，需要進行因果依賴關係的檢查，這個操作的邏輯時鐘和位移大於C本地檔案中已經應用的操作的邏輯時鐘和位移，需要等待先前的操作被應用後再應用這個操作，當然這並不是在yjs中的實現，因為yjs不會存在真正的刪除操作，並且在刪除操作的時候實際上並不會導致時鐘的增加，只是增加一個標記，上邊這個例子其實可以換個說法，兩個相同的插入操作，因為我們是相對位置，所以後一個插入操作是依賴前一個插入操作的，因此就需要因果檢查，其實這也是件有意思的事情，當收到在同一個位置編輯的不同使用者端操作時候，如果時鐘相同就是衝突操作，不相同就是因果關係。

那麼由此我們通過CRDT資料結構與演演算法設計解決了最終一致性和因果一致性，對於意圖一致性的問題，當不存在衝突的時候我們是能夠保證意圖的，即插入檔案的Item的順序，在衝突的時候我們實際上會比較clientID決定究竟誰在前在後，其實實際上無論誰在前還是在後都可以認為是一種烏龍，我們在衝突的時候只保證最終一致性，對於意圖一致性則需要做額外的設計才可以實現，在這裡就不做過多探討了。實際上yjs還有大量的設計與優化操作，以及基於YATA的衝突解決演演算法等，比如通過雙向連結串列來儲存檔案結構順序，通過Map為每個使用者端儲存的扁平的 Item陣列，優化本地插入的速度而設計的快取機制(連結串列的查詢O(N)與跟隨遊標的位置快取)，傾向於State-based的刪除，Undo/Redo，遊標同步，壓縮資料網路傳輸等等，還是很值得研究的。

我們再回到富文字的範例Quill中，實現的主要功能是在quill富文字編輯器中接入協同，並支援編輯遊標的同步，該範例的地址是https://github.com/WindrunnerMax/Collab/tree/master/packages/crdt-quill，首先簡單看一下目錄結構(tree --dirsfirst -I node_modules):

crdt-quill
├── public
│   └── favicon.ico
├── server
│   └── index.ts
├── src
│   ├── client.ts
│   ├── index.css
│   ├── index.ts
│   └── quill.ts
├── package.json
├── rollup.config.js
├── rollup.server.js
└── tsconfig.json

依舊簡略說明下各個資料夾和檔案的作用，public儲存了靜態資原始檔，在使用者端打包時將會把內容移動到build資料夾，server資料夾中儲存了CRDT伺服器端的實現，在執行時同樣會編譯為js檔案放置於build資料夾下，src資料夾是使用者端的程式碼，主要是檢視與CRDT使用者端的實現，rollup.config.js是打包使用者端的組態檔，rollup.server.js是打包伺服器端的組態檔，package.json與tsconfig.json大家都懂，就不贅述了。

quill的資料結構並不是JSON而是Delta，Delta是通過retain、insert、delete三個操作完成整篇檔案的描述與操作，我們試想一下描述一段字串的操作需要什麼，是不是通過這三種操作就能夠完全覆蓋了，所以通過Delta來描述文字增刪改是完全可行的，而且12年quill的開源可以說是富文字發展的一個里程碑，於是yjs是直接原生支援Delta資料結構的。

接下來我們看看來看看伺服器端，這裡主要實現是呼叫了一下y-websocket來啟動一個websocket伺服器，這是y-websocket給予的開箱即用的功能，也可以基於這些內容進行改寫，yjs還提供了y-mongodb-provider等伺服器端服務可以使用。後邊主要是使用了express啟動了一個靜態資源伺服器，因為直接在瀏覽器開啟檔案的file協定有很多的安全限制，所以需要一個HTTP Server。

import { exec } from "child_process";
import express from "express";

// https://github.com/yjs/y-websocket/blob/master/bin/server.js
exec("PORT=3001 npx y-websocket", (err, stdout, stderr) => { // 呼叫`y-websocket`
  console.log(stdout, stderr);
});

const app = express(); // 範例化`express`
app.use(express.static("build")); // 使用者端打包過後的靜態資源路徑
app.use(express.static("node_modules/quill/dist")); // `quill`靜態資源路徑
app.listen(3000);
console.log("Listening on http://localhost:3000");

在使用者端方面主要是定義了一個定義了一個共用的連結，通過crdt-quill作為RoomName進入組，這裡需要連結的websocket伺服器也就是上邊啟動的y-websocket的3001埠的服務。之後我們定義了頂層的資料結構為YText資料結構的變化來執行回撥，並且將一些資訊暴露了出去，doc就是這需要使用的yjs範例，type是我們定義的頂層資料結構，awareness意為感知，只要是用來完成實時資料同步，在這裡是用來同步遊標選區。

import { Doc, Text as YText } from "yjs";
import { WebsocketProvider } from "y-websocket";

class Connection {
  public doc: Doc; // `yjs`範例
  public type: YText; // 頂層資料結構
  private connection: WebsocketProvider; // `WebSocket`連結
  public awareness: WebsocketProvider["awareness"]; // 資料實時同步

  constructor() {
    const doc = new Doc(); // 範例化
    const provider = new WebsocketProvider("ws://localhost:3001", "crdt-quill", doc); // 連結`WebSocket`伺服器
    provider.on("status", (e: { status: string }) => {
      console.log("WebSocket", e.status); // 連結狀態
    });
    this.doc = doc; // `yjs`範例
    this.type = doc.getText("quill"); // 獲取頂層資料結構
    this.connection = provider; // 連結
    this.awareness = provider.awareness; // 資料實時同步
  }

  reconnect() {
    this.connection.connect(); // 重連
  }

  disconnect() {
    this.connection.disconnect(); // 斷線
  }
}

export default new Connection();

在使用者端主要分為了兩部分，分別是範例化quill的範例，以及quill與yjs使用者端通訊的實現。在quill的實現中主要是將quill範例化，註冊遊標的外掛，隨機生成id的方法，通過id獲取隨機顏色的方法，以及遊標同步的位置轉換。在quill與yjs使用者端通訊的實現中，主要是完成了對於quill與doc的事件監聽，主要是遠端資料變更的回撥，本地資料變化的回撥，遊標同步事件感知的回撥。

import Quill from "quill";
import QuillCursors from "quill-cursors";
import tinyColor from "tinycolor2";
import { Awareness } from "y-protocols/awareness.js";
import {
  Doc,
  Text as YText,
  createAbsolutePositionFromRelativePosition,
  createRelativePositionFromJSON,
} from "yjs";
export type { Sources } from "quill";

Quill.register("modules/cursors", QuillCursors); // 註冊遊標外掛

export default new Quill("#editor", { // 範例化`quill`
  theme: "snow",
  modules: { cursors: true },
});

const COLOR_MAP: Record<string, string> = {}; // `id => color`

export const getRandomId = () => Math.floor(Math.random() * 10000).toString(); // 隨機生成使用者`id`

export const getCursorColor = (id: string) => { // 根據`id`獲取顏色
  COLOR_MAP[id] = COLOR_MAP[id] || tinyColor.random().toHexString();
  return COLOR_MAP[id];
};

export const updateCursor = (
  cursor: QuillCursors,
  state: Awareness["states"] extends Map<number, infer I> ? I : never,
  clientId: number,
  doc: Doc,
  type: YText
) => {
  try {
    // 從`Awareness`中取得狀態
    if (state && state.cursor && clientId !== doc.clientID) {
      const user = state.user || {};
      const color = user.color || "#aaa";
      const name = user.name || `User: ${clientId}`;
      // 根據`clientId`建立遊標
      cursor.createCursor(clientId.toString(), name, color);
      // 相對位置轉換為絕對位置 // 選區為`focus --- anchor`
      const focus = createAbsolutePositionFromRelativePosition(
        createRelativePositionFromJSON(state.cursor.focus),
        doc
      );
      const anchor = createAbsolutePositionFromRelativePosition(
        createRelativePositionFromJSON(state.cursor.anchor),
        doc
      );
      if (focus && anchor && focus.type === type) {
        // 移動遊標位置
        cursor.moveCursor(clientId.toString(), {
          index: focus.index,
          length: anchor.index - focus.index,
        });
      }
    } else {
      // 根據`clientId`移除遊標
      cursor.removeCursor(clientId.toString());
    }
  } catch (err) {
    console.error(err);
  }
};

import "./index.css";
import quill, { getRandomId, updateCursor, Sources, getCursorColor } from "./quill";
import client from "./client";
import Delta from "quill-delta";
import QuillCursors from "quill-cursors";
import { compareRelativePositions, createRelativePositionFromTypeIndex } from "yjs";

const userId = getRandomId(); // 本地使用者端的`id` 或者使用`awareness.clientID`
const doc = client.doc; // `yjs`範例
const type = client.type; // 頂層型別
const cursors = quill.getModule("cursors") as QuillCursors; // `quill`遊標模組
const awareness = client.awareness; // 實時通訊感知模組

// 設定當前使用者端的資訊 `State`的資料結構類似於`Record<string, unknown>`
awareness.setLocalStateField("user", {
  name: "User: " + userId,
  color: getCursorColor(userId),
});

// 頁面顯示的使用者資訊
const userNode = document.getElementById("user") as HTMLInputElement;
userNode && (userNode.value = "User: " + userId);

type.observe(event => {
  // 來源資訊 // 本地`UpdateContents`不應該再觸發`ApplyDelta'
  if (event.transaction.origin !== userId) {
    const delta = event.delta;
    quill.updateContents(new Delta(delta), "api"); // 應用遠端資料, 來源
  }
});

quill.on("editor-change", (_: string, delta: Delta, state: Delta, origin: Sources) => {
  if (delta && delta.ops) {
    // 來源資訊 // 本地`ApplyDelta`不應該再觸發`UpdateContents`
    if (origin !== "api") {
      doc.transact(() => {
        type.applyDelta(delta.ops); // 應用`Ops`到`yjs`
      }, userId); // 來源
    }
  }

  const sel = quill.getSelection(); // 選區
  const aw = awareness.getLocalState(); // 實時通訊狀態資料
  if (sel === null) { // 失去焦點
    if (awareness.getLocalState() !== null) {
      awareness.setLocalStateField("cursor", null); // 清除選區狀態
    }
  } else {
    // 卷對位置轉換為相對位置 // 選區為`focus --- anchor`
    const focus = createRelativePositionFromTypeIndex(type, sel.index);
    const anchor = createRelativePositionFromTypeIndex(type, sel.index + sel.length);
    if (
      !aw ||
      !aw.cursor ||
      !compareRelativePositions(focus, aw.cursor.focus) ||
      !compareRelativePositions(anchor, aw.cursor.anchor)
    ) {
      // 選區位置發生變化 設定位置資訊
      awareness.setLocalStateField("cursor", { focus, anchor });
    }
  }
  // 更新所有遊標狀態到本地
  awareness.getStates().forEach((aw, clientId) => {
    updateCursor(cursors, aw, clientId, doc, type);
  });
});

// 初始化更新所有遠端遊標狀態到本地
awareness.getStates().forEach((state, clientId) => {
  updateCursor(cursors, state, clientId, doc, type);
});
// 監聽遠端狀態變化的回撥
awareness.on(
  "change",
  ({ added, removed, updated }: { added: number[]; removed: number[]; updated: number[] }) => {
    const states = awareness.getStates();
    added.forEach(id => {
      const state = states.get(id);
      state && updateCursor(cursors, state, id, doc, type);
    });
    updated.forEach(id => {
      const state = states.get(id);
      state && updateCursor(cursors, state, id, doc, type);
    });
    removed.forEach(id => {
      cursors.removeCursor(id.toString());
    });
  }
);

每日一題

https://github.com/WindrunnerMax/EveryDay

參考

https://docs.yjs.dev/
https://github.com/yjs/yjs
https://github.com/automerge/automerge
https://zhuanlan.zhihu.com/p/425265438
https://zhuanlan.zhihu.com/p/452980520
https://josephg.com/blog/crdts-go-brrr/
https://www.npmjs.com/package/quill-delta
https://josephg.com/blog/crdts-are-the-future/
https://github.com/yjs/yjs/blob/main/INTERNALS.md
https://cloud.tencent.com/developer/article/2081651