Реализация CRDT/OT для real-time синхронизации на сайте

Реализация CRDT/OT для real-time синхронизации на сайте

Real-time синхронизация нескольких клиентов без конфликтов — задача, которая на практике оказывается сложнее, чем выглядит. Простая трансляция событий через WebSocket работает до тех пор, пока два пользователя не редактируют один фрагмент одновременно. Дальше без формальной модели данных начинается каша.

OT против CRDT: принципиальная разница

Operational Transformation (OT) — алгоритм, который трансформирует операции с учётом конкурентных изменений. Применялся в Google Docs с 2006 года. Суть: если пользователь A вставил символ на позицию 5, а пользователь B удалил символ на позиции 3, то операция A должна быть трансформирована перед применением на стороне B.

Алгоритм работает корректно только при наличии центрального сервера-арбитра, который упорядочивает операции. Без него реализация OT для более двух клиентов становится экспоненциально сложной.

CRDT (Conflict-free Replicated Data Types) — структуры данных, которые математически гарантируют eventual consistency без координации. Операции коммутативны и идемпотентны — порядок применения не влияет на результат.

CRDT на практике: Yjs

Yjs — наиболее зрелая CRDT-библиотека для браузера и Node.js. Структура данных строится из Y.Doc, который содержит shared types: Y.Text, Y.Map, Y.Array.

npm install yjs y-websocket y-protocols

Серверная часть (y-websocket):

// server.js
const { WebSocketServer } = require('ws');
const { setupWSConnection } = require('y-websocket/bin/utils');
const http = require('http');

const server = http.createServer((req, res) => {
  res.writeHead(200);
  res.end('ok');
});

const wss = new WebSocketServer({ server });

wss.on('connection', (ws, req) => {
  // roomName извлекается из URL: /doc/my-document-id
  setupWSConnection(ws, req, {
    docName: req.url.slice(1),
    gc: true, // garbage collection для удалённых элементов
  });
});

server.listen(1234);

Клиентская часть с интеграцией в редактор:

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

const ydoc = new Y.Doc();

const provider = new WebsocketProvider(
  'ws://localhost:1234',
  'my-document-room',
  ydoc,
  { connect: true }
);

// Статус подключения
provider.on('status', ({ status }) => {
  console.log('WS status:', status); // 'connected' | 'disconnected'
});

// Синхронизация начального состояния
provider.on('sync', (isSynced: boolean) => {
  if (isSynced) {
    console.log('Document synced from server');
  }
});

const ytext = ydoc.getText('quill-content');

const quill = new Quill('#editor', { theme: 'snow' });
const binding = new QuillBinding(ytext, quill, provider.awareness);

// Awareness: присутствие и курсоры пользователей
provider.awareness.setLocalStateField('user', {
  name: 'Иван Петров',
  color: '#4a9eff',
});

Y.Map и Y.Array для структурированных данных

Текстовые редакторы — частный случай. CRDT применяется шире: синхронизация состояния форм, досок задач, диаграмм.

// Синхронизированная карта задач (Kanban-доска)
const ytasks = ydoc.getMap<Y.Map<unknown>>('tasks');

// Создание задачи
function createTask(id: string, title: string, status: string) {
  const task = new Y.Map<unknown>();
  task.set('id', id);
  task.set('title', title);
  task.set('status', status);
  task.set('createdAt', Date.now());
  ytasks.set(id, task);
}

// Перемещение задачи
function moveTask(id: string, newStatus: string) {
  const task = ytasks.get(id);
  if (task) {
    task.set('status', newStatus);
  }
}

// Реактивное обновление UI
ytasks.observe((event) => {
  event.changes.keys.forEach((change, key) => {
    if (change.action === 'add') {
      renderTask(ytasks.get(key));
    } else if (change.action === 'delete') {
      removeTaskFromUI(key);
    } else if (change.action === 'update') {
      updateTaskInUI(key, ytasks.get(key));
    }
  });
});

Персистентность: хранение Y.Doc на сервере

Ephemeral y-websocket сервер теряет документ при перезапуске. Для production нужен persistence layer.

// Вариант 1: y-leveldb (для single-node)
const { LeveldbPersistence } = require('y-leveldb');
const persistence = new LeveldbPersistence('./data');

setupWSConnection(ws, req, {
  docName,
  gc: true,
  persistence, // автоматически сохраняет и восстанавливает
});

// Вариант 2: Redis (для multi-node через y-redis)
// npm install y-redis
import { createRedisStorage } from 'y-redis';

const redisStorage = createRedisStorage({
  host: 'localhost',
  port: 6379,
});

// Вариант 3: PostgreSQL с Hocuspocus
// npm install @hocuspocus/server @hocuspocus/extension-database
import { Server } from '@hocuspocus/server';
import { Database } from '@hocuspocus/extension-database';
import { Pool } from 'pg';

const pool = new Pool({ connectionString: process.env.DATABASE_URL });

const server = Server.configure({
  port: 1234,
  extensions: [
    new Database({
      fetch: async ({ documentName }) => {
        const { rows } = await pool.query(
          'SELECT data FROM documents WHERE name = $1',
          [documentName]
        );
        return rows[0]?.data ?? null;
      },
      store: async ({ documentName, state }) => {
        await pool.query(
          `INSERT INTO documents (name, data, updated_at)
           VALUES ($1, $2, NOW())
           ON CONFLICT (name)
           DO UPDATE SET data = $2, updated_at = NOW()`,
          [documentName, Buffer.from(state)]
        );
      },
    }),
  ],
});

server.listen();

Automerge как альтернатива

Automerge 2.x (написан на Rust/WASM) обеспечивает лучшую производительность для больших документов и нативно поддерживает JSON-подобные структуры:

import * as Automerge from '@automerge/automerge';

// Инициализация документа
let doc = Automerge.init<{ items: string[] }>();

// Локальное изменение
doc = Automerge.change(doc, 'Add item', (d) => {
  if (!d.items) d.items = [];
  d.items.push('новый элемент');
});

// Сериализация для передачи по сети
const binary = Automerge.save(doc);

// Получение изменений от другого клиента
const [newDoc, patch] = Automerge.applyChanges(doc, [remoteChange]);

Разрешение конфликтов в CRDT

CRDT не устраняет конфликты — он определяет детерминированный победитель. Для Last-Write-Wins Map (LWW-Map) побеждает операция с более поздним timestamp. Для текста Yjs использует алгоритм YATA, где позиция вставки определяется соседними элементами, а не индексом — это устойчиво к конкурентным вставкам.

Граничный случай: два пользователя одновременно удаляют и редактируют один элемент. В CRDT удаление победит при LWW, но Yjs сохранит контент как "надгробие" (tombstone) с информацией о том, что он удалён — это позволяет корректно применить изменения, сделанные до удаления.

Масштабирование: multi-node синхронизация

Один WebSocket-сервер не масштабируется горизонтально — каждый клиент подключается к конкретному процессу. Решения:

Sticky sessions на уровне nginx (по document_id → upstream):

upstream yjs_backend {
    hash $arg_room consistent;  # или cookie_room
    server yjs1:1234;
    server yjs2:1234;
    server yjs3:1234;
}

Pub/Sub через Redis — каждый сервер публикует обновления в Redis channel, остальные подписаны. Hocuspocus с расширением Redis поддерживает это из коробки.

Liveblocks/PartyKit — управляемая инфраструктура для CRDT, если нет желания поддерживать собственный кластер.

Сроки реализации

Базовая CRDT-синхронизация текстового редактора на Yjs + y-websocket: 3–5 дней. Добавление персистентности через PostgreSQL/Redis: ещё 2–3 дня. Multi-node с Redis pub/sub и тестированием под нагрузкой: плюс неделя. Кастомные типы данных (Kanban, диаграммы) поверх Y.Map: зависит от UI-сложности.