实时协作画布系统：Yjs + Hocuspocus + 持久化

概述

在设计工具、白板应用和文档编辑器中，多用户同时编辑同一文档的需求日益增长。传统的客户端-服务器模型在这种场景下存在诸多挑战，例如冲突解决、网络延迟和离线支持等。

为了支持多用户同时编辑画布内容，并具备离线编辑能力，我们采用了 Yjs（一种 CRDT 实现）配合 IndexedDB 和 Hocuspocus 的架构方案。

核心技术组件

Yjs (Y.Map)

Yjs 是一个用于创建实时协作应用程序的库，它实现了 Conflict-free Replicated Data Types (CRDTs) 算法。CRDTs 是一种特殊的数据结构，可以在多个副本之间同步，而不需要中央协调，从而保证最终一致性。

Y.Map 是 Yjs 提供的一种共享数据类型，类似于 JavaScript 中的 Map。它的关键特性包括：

• 自动冲突解决：当多个用户同时修改数据时，Yjs 自动解决冲突
• 分布式一致性：保证所有客户端看到相同的数据状态
• 高效同步：只传输变更部分，减少网络流量

持久化选项

持久化是协作系统的关键组件，不仅需要在客户端存储数据以支持离线使用，还需要在服务端存储数据以实现长期保存和共享。本项目实际实现的持久化策略包括：

1. IndexedDB（客户端）

IndexedDB 是浏览器内置的数据库，用于存储大量结构化数据。在协作系统中，它用于：

• 使用 y-indexeddb 库创建 IndexeddbPersistence 实例
• 为每个房间创建独立的 IndexedDB 存储 (canvas-local-db-${roomId})
• 提供 getYDocForRoom、getYElementsForRoom 和 getIndexedDBProviderForRoom 等函数来管理不同房间的数据

import * as Y from 'yjs';
import { IndexeddbPersistence } from 'y-indexeddb';
import { HocuspocusProvider } from '@hocuspocus/provider'; // 如需实时同步时导入

// 使用 Map 存储不同房间的 Yjs 文档及相关提供者，确保单例管理和数据隔离
const roomDocuments = new Map<
  string,
  {
    yDoc: Y.Doc;
    yElements: Y.Map<any>;
    indexeddbProvider: IndexeddbPersistence;
    wsProvider: HocuspocusProvider | null;
  }
>();

/**
 * 获取或创建指定房间的 Yjs 文档实例
 * 
 * @param roomId 协作房间的唯一标识符
 * @returns 该房间对应的 Y.Doc 实例
 */
export const getYDocForRoom = (roomId: string): Y.Doc => {
  // 若该房间的文档已存在，直接复用以避免重复创建
  if (roomDocuments.has(roomId)) {
    return roomDocuments.get(roomId)!.yDoc;
  }

  // 创建新的 Yjs 文档实例
  const yDoc = new Y.Doc();

  // 获取画布元素的核心数据结构（Y.Map，用于存储所有 CanvasElement）
  const yElements = yDoc.getMap<any>('elements');

  // 初始化 IndexedDB 持久化提供者
  // 数据库名称动态包含 roomId，确保不同房间的数据互不干扰
  const indexeddbProvider = new IndexeddbPersistence(`canvas-local-db-${roomId}`, yDoc);

  // 将文档相关信息存入 Map，便于后续访问和管理
  roomDocuments.set(roomId, {
    yDoc,
    yElements,
    indexeddbProvider,
    wsProvider: null, // 初始为空，后续可动态绑定 HocuspocusProvider 以实现实时协作
  });

  return yDoc;
};

• 离线数据存储：即使用户断网，数据也不会丢失
• 快速本地访问：减少对服务器的依赖
• 大容量存储：相比 localStorage，支持更大的数据量

2. SQLite（服务端）

SQLite 作为服务端数据库，用于持久化存储画布内容：

• 关系型结构：提供 SQL 查询能力
• 轻量级：无需单独的服务器进程
• 跨平台：可在多种环境中运行
• 服务端存储：确保数据在服务端持久化

// 服务端数据库实现 (ALD_Backend/src/db.ts)

import { Database } from "bun:sqlite";
const db = new Database("collab.sqlite");

// 启用 WAL 模式以提高并发性能
db.exec("PRAGMA journal_mode = WAL;");
// 房间表，包含 content BLOB 字段存储 Yjs 二进制数据
db.run(`
  CREATE TABLE IF NOT EXISTS rooms (
    id TEXT PRIMARY KEY,
    name TEXT NOT NULL,
    creator_id TEXT NOT NULL,
    created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
    content BLOB, -- Yjs 二进制数据
    FOREIGN KEY (creator_id) REFERENCES users(id)
  )
`);

// Hocuspocus 数据库扩展实现 (ALD_Backend/src/collab.ts)

const dbExtension = new HocuspocusDB({
  fetch: async ({ documentName }) => {
    const roomId = getRoomId(documentName);
    console.log(
      `[Yjs] Fetching data for RoomID: ${roomId}, Original documentName: ${documentName}`
    );

    const query = db.query("SELECT content FROM rooms WHERE id = $id");
    const row = query.get({ $id: roomId }) as {
      content: Uint8Array | null;
    } | null;

    if (row && row.content !== null && row.content !== undefined) {
      if (row.content.length > 0) {
        console.log(
          `[Yjs] Returning data with size: ${row.content.length} bytes`
        );
        return new Uint8Array(row.content);
      } else {
        console.log(`[Yjs] content is empty, creating new Yjs document`);
        const ydoc = new Y.Doc();
        ydoc.getMap("elements"); // 存储图形元素
        return Y.encodeStateAsUpdate(ydoc);
      }
    }

    console.log(`[Yjs] No valid data found, creating new Yjs document`);
    const ydoc = new Y.Doc();
    ydoc.getMap("elements"); // 存储图形元素
    return Y.encodeStateAsUpdate(ydoc);
  },

  store: async ({ documentName, state }) => {
    const roomId = getRoomId(documentName);
    try {
      console.log(
        `[Yjs] Saving data for RoomID: ${roomId}, State size: ${state.length} bytes, Original documentName: ${documentName}`
      );

      if (state.length > 0) {
        const roomCheck = db.query("SELECT id FROM rooms WHERE id = $id");

        const roomExists = roomCheck.get({ $id: roomId });

        if (!roomExists) {
          console.error(
            `[Yjs] Room ${roomId} does not exist, cannot save data`
          );

          return;
        }

        const update = db.query(
          "UPDATE rooms SET content = $blob WHERE id = $id"
        );

        update.run({ $blob: state, $id: roomId });

        console.log(`[Yjs] Data saved successfully for RoomID: ${roomId}`);
      } else {
        console.log(
          `[Yjs] Skipping save for RoomID: ${roomId} as state is empty`
        );
      }
    } catch (error) {
      console.error(`[Yjs] Save failed for ${roomId}:`, error);
    }
  },
});

Hocuspocus Provider

Hocuspocus 是一个协作编辑框架，提供了 Yjs 的服务器端实现。它负责：

• 多客户端同步：协调多个客户端之间的数据同步
• WebSocket 连接管理：建立持久连接
• 房间管理：隔离不同协作空间的数据
• 与多种持久化后端集成：可连接到数据库、文件系统等

系统架构设计

graph TB
    subgraph "Frontend Application (React/Vue)"
        A[Canvas UI Components]
        B[State Management<br/>Zustand/Pinia]
        C[Yjs Document<br/>Shared Data]
        D[IndexedDB<br/>Local Persistence]
    end
    subgraph "Backend Services"
        E[Hono Server<br/>Port 3000]
        F[RESTful API<br/>Auth/RM]
        G[SQLite DB<br/>Persistence]
    end
    subgraph "WebSocket Services"
        H[Hocuspocus Server<br/>Port 1234]
        I[Yjs Extensions<br/>DB/Storage]
    end
    A <--> C
    B <--> C
    C <--> D
    C <--> H
    E <--> F
    E <--> G
    F <--> H
    H <--> I
    I <--> G
    style A fill:#87CEEB
    style B fill:#98FB98
    style C fill:#FFD700
    style D fill:#DDA0DD
    style E fill:#F0E68C
    style F fill:#FFA07A
    style G fill:#BA55D3
    style H fill:#20B2AA
    style I fill:#FF69B4

graph LR
    subgraph "Client A"
        A[Y.Map<br/>Shared Data]
        B[IndexedDB<br/>Local Persistence]
    end
    subgraph "Server"
        C[Hocuspocus<br/>Server]
        D[SQLite DB<br/>Persistence]
    end
    subgraph "Client B"
        E[Y.Map<br/>Shared Data]
        F[IndexedDB<br/>Local Persistence]
    end
    A <--> C
    E <--> C
    B <--> D
    F <--> D
    C <--> D
    style A fill:#FFD700
    style B fill:#DDA0DD
    style C fill:#20B2AA
    style D fill:#BA55D3
    style E fill:#FFD700
    style F fill:#DDA0DD

数据流向

1. 用户操作更新本地 Y.Map
2. 变更自动同步到 IndexedDB（本地持久化）
3. 变更通过 Hocuspocus 同步到服务器和其他客户端
4. 服务器将变更存储到 SQLite 数据库
5. 其他客户端接收变更并更新本地 Y.Map

实现细节

1. 初始化协作环境

import * as Y from "yjs";
import { IndexeddbPersistence } from "y-indexeddb";
import { HocuspocusProvider } from "@hocuspocus/provider";

// 创建 Yjs 文档
const ydoc = new Y.Doc();
// 获取共享的 Y.Map 用于存储画布元素
const yElements = ydoc.getMap("elements");
// 设置 IndexedDB 持久化
const indexeddbProvider = new IndexeddbPersistence("canvas-room", ydoc);
// 设置 Hocuspocus 提供者
export const initWsProvider = (roomId: string, token: string) => {
  // 如果房间不存在，先创建
  if (!roomDocuments.has(roomId)) {
    getYDocForRoom(roomId);
  }

  const roomData = roomDocuments.get(roomId)!;

  // 如果已存在 WebSocket 提供者，先销毁
  if (roomData.wsProvider) {
    roomData.wsProvider.destroy();
  }

  // 创建新的 WebSocket 提供者，并关联 Yjs 文档
  console.log(
    `[Room ${roomId}] Initializing WebSocket Provider with token: ${token}`
  );

  const wsProvider = new HocuspocusProvider({
    // 确保 URL 结尾规范，方便拼接
    url: `ws://localhost:3000/collaboration/${roomId}`,
    name: roomId, // Hocuspocus 会将其拼接为 /collaboration/{roomId}
    token: token,
    // 明确指定要同步的文档
    document: roomData.yDoc,
  });
  console.log(wsProvider);

  // 监听 WebSocket 连接状态
  wsProvider.on("status", (event: any) => {
    console.log(`[Room ${roomId}] WebSocket status:`, event.status); // 'connected' or 'disconnected'
  });

  // 更新房间数据中的 WebSocket 提供者
  roomData.wsProvider = wsProvider;
  return wsProvider;
};

2. 画布元素管理

interface CanvasElement {
  id: string;
  type: string;
  x: number;
  y: number;
  width: number;
  height: number;
  color: string;
}

// 添加元素

yElements.set(elementId, {
  id: elementId,
  type: "rectangle",
  x: 100,
  y: 100,
  width: 200,
  height: 150,
  color: "#ff0000",
});

// 监听元素变化

yElements.observeDeep((events) => {
  events.forEach((event) => {
    // 处理添加、更新、删除事件
  });
});

3. React 状态集成

import { create } from "zustand";
import { useEffect } from "react";
interface CanvasStore {
  ydoc: Y.Doc;
  yElements: Y.Map<CanvasElement>;
  elements: Map<string, CanvasElement>;
  addElement: (element: CanvasElement) => void;
  updateElement: (id: string, updates: Partial<CanvasElement>) => void;
  deleteElement: (id: string) => void;
}

export const useCanvasStore = create<CanvasStore>((set, get) => ({
  // ... store implementation
}));

4. 离线支持实现

离线支持是通过 IndexedDB 实现的：

• 当用户在线时，所有操作同步到服务器和其他客户端
• 当用户离线时，操作仅保存在本地 IndexedDB 中
• 当用户重新连接时，本地更改自动同步到服务器

// 等待 IndexedDB 数据加载
await indexeddbProvider.whenSynced;

// 监听连接状态
provider.on("synced", () => {
  console.log("Document synced with server");
});

优化

后续还可以实现批量更新，防抖，服务端数据验证等优化。