CRDT と AI で支えるリアルタイム共同編集 — 競合を数学で解消するデータ同期設計

# crdt/yjs_bridge.py
"""
Python バックエンドと Yjs（JavaScript CRDT）の間のブリッジ。
本番では Yjs はクライアントで実行；サーバーストレージに同期します。
"""
from typing import Any, Dict, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
import json
 
@dataclass
class YjsUpdate:
    """Yjs ドキュメント内の変更を表現"""
    origin_client_id: str
    clock: int  # 順序付けのためのラムポート時計
    content: bytes  # エンコードされた Yjs 更新
    timestamp: datetime = field(default_factory=datetime.utcnow)
    parent_clock: int = 0  # 因果関係の順序付け
 
    def to_dict(self) -> Dict[str, Any]:
        return {
            "origin_client_id": self.origin_client_id,
            "clock": self.clock,
            "content": self.content.hex(),  # JSON 用 Base64
            "timestamp": self.timestamp.isoformat(),
            "parent_clock": self.parent_clock
        }
 
@dataclass
class CollaborativeDocument:
    """CRDT 同期で管理されるドキュメント"""
    doc_id: str
    client_id: str
    updates: List[YjsUpdate] = field(default_factory=list)
    vector_clock: Dict[str, int] = field(default_factory=dict)
    last_synced: Optional[datetime] = None
 
    def add_update(self, update: YjsUpdate) -> None:
        """ローカル変更をドキュメントに追加"""
        self.updates.append(update)
        # このクライアント用にベクトル時計を更新
        self.vector_clock[self.client_id] = (
            self.vector_clock.get(self.client_id, 0) + 1
        )
 
    def get_missing_updates(
        self,
        remote_vector_clock: Dict[str, int]
    ) -> List[YjsUpdate]:
        """リモートクライアントが見ていない更新を決定"""
        missing = []
 
        for update in self.updates:
            remote_clock = remote_vector_clock.get(
                update.origin_client_id, 0
            )
            if update.clock > remote_clock:
                missing.append(update)
 
        return missing
 
    def merge_remote_update(self, update: YjsUpdate) -> None:
        """別のクライアントからの変更を組み込む"""
        # 時計順序で更新を挿入
        insert_pos = len(self.updates)
        for i, existing in enumerate(self.updates):
            if existing.clock > update.clock:
                insert_pos = i
                break
 
        self.updates.insert(insert_pos, update)
 
        # ベクトル時計を更新
        self.vector_clock[update.origin_client_id] = max(
            self.vector_clock.get(update.origin_client_id, 0),
            update.clock
        )
 
# ドキュメント同期プロトコル
class CRDTSyncManager:
    """クライアント全体で CRDT ドキュメント同期を管理"""
 
    def __init__(self, storage_backend, max_batch_size: int = 100):
        self.storage = storage_backend
        self.max_batch_size = max_batch_size
        self.documents: Dict[str, CollaborativeDocument] = {}
 
    async def sync_document(
        self,
        doc_id: str,
        client_id: str,
        local_updates: List[YjsUpdate],
        remote_vector_clock: Dict[str, int]
    ) -> Dict[str, Any]:
        """
        クライアントとサーバー間でドキュメントを同期。
 
        サーバーは「因果関係のシーケンサー」として機能しますが、調整者ではありません。
        すべてのマージ決定はベクトル時計に基づいて決定的です。
        """
        doc = self.documents.get(doc_id)
        if not doc:
            doc = CollaborativeDocument(doc_id=doc_id, client_id=client_id)
            self.documents[doc_id] = doc
 
        # 受信更新をマージ
        for update in local_updates:
            doc.merge_remote_update(update)
 
        # 送信する更新を検索
        missing_updates = doc.get_missing_updates(remote_vector_clock)
 
        # 効率のためバッチ化
        update_batches = [
            missing_updates[i:i + self.max_batch_size]
            for i in range(0, len(missing_updates), self.max_batch_size)
        ]
 
        # ストレージに永続化
        await self.storage.save_document(doc)
 
        return {
            "doc_id": doc_id,
            "acknowledged": len(local_updates),
            "updates_to_apply": update_batches,
            "server_vector_clock": doc.vector_clock,
            "needs_full_sync": self._should_trigger_full_sync(doc)
        }
 
    def _should_trigger_full_sync(self, doc: CollaborativeDocument) -> bool:
        """完全なドキュメント状態を必要とする状況を検出"""
        # ドキュメントが非常に大きくなった場合、操作の代わりに状態を送信
        total_update_size = sum(
            len(u.content) for u in doc.updates
        )
        return total_update_size > 10_000_000  # 10MB 閾値

# sync/websocket_handler.py
import asyncio
from typing import Set
from fastapi import WebSocket, WebSocketDisconnect
import json
 
class CollaborativeDocumentManager:
    """ドキュメント同期のリアルタイムの WebSocket ハンドラー"""
 
    def __init__(self, crdt_sync: CRDTSyncManager):
        self.crdt_sync = crdt_sync
        # ルーム -> 接続されたクライアントのセット
        self.active_connections: Dict[str, Set[WebSocket]] = {}
        # どのクライアントがどのルームに接続されているかを追跡
        self.client_rooms: Dict[WebSocket, str] = {}
 
    async def connect(self, websocket: WebSocket, doc_id: str, client_id: str):
        """クライアントがドキュメント編集セッションに参加"""
        await websocket.accept()
 
        if doc_id not in self.active_connections:
            self.active_connections[doc_id] = set()
 
        self.active_connections[doc_id].add(websocket)
        self.client_rooms[websocket] = doc_id
 
        # 初期ドキュメント状態を送信
        doc = self.crdt_sync.documents.get(doc_id)
        await websocket.send_json({
            "type": "initial_state",
            "updates": [u.to_dict() for u in (doc.updates if doc else [])],
            "vector_clock": doc.vector_clock if doc else {}
        })
 
    async def handle_client_updates(
        self,
        websocket: WebSocket,
        doc_id: str,
        client_id: str,
        updates: List[Dict[str, Any]]
    ):
        """クライアントからの更新を処理して他のクライアントへブロードキャスト"""
        # JSON を YjsUpdate オブジェクトに変換
        yjs_updates = [
            YjsUpdate(
                origin_client_id=u["origin_client_id"],
                clock=u["clock"],
                content=bytes.fromhex(u["content"]),
                parent_clock=u.get("parent_clock", 0)
            )
            for u in updates
        ]
 
        # サーバーと同期
        sync_result = await self.crdt_sync.sync_document(
            doc_id=doc_id,
            client_id=client_id,
            local_updates=yjs_updates,
            remote_vector_clock={}
        )
 
        # 他のクライアントへブロードキャスト
        await self.broadcast_updates(
            doc_id,
            exclude_client=websocket,
            updates=yjs_updates,
            server_vector_clock=sync_result["server_vector_clock"]
        )
 
        # 受信を確認
        await websocket.send_json({
            "type": "ack",
            "acknowledged_count": sync_result["acknowledged"],
            "server_vector_clock": sync_result["server_vector_clock"]
        })
 
    async def broadcast_updates(
        self,
        doc_id: str,
        exclude_client: WebSocket,
        updates: List[YjsUpdate],
        server_vector_clock: Dict[str, int]
    ):
        """すべての接続されたクライアントに更新を送信"""
        if doc_id not in self.active_connections:
            return
 
        disconnected = set()
 
        for client in self.active_connections[doc_id]:
            if client == exclude_client:
                continue
 
            try:
                await client.send_json({
                    "type": "remote_updates",
                    "updates": [u.to_dict() for u in updates],
                    "server_vector_clock": server_vector_clock
                })
            except Exception:
                disconnected.add(client)
 
        # 切断されたクライアントをクリーンアップ
        for client in disconnected:
            await self.disconnect(client)
 
    async def disconnect(self, websocket: WebSocket):
        """クライアントがセッションから切断"""
        doc_id = self.client_rooms.pop(websocket, None)
        if doc_id and doc_id in self.active_connections:
            self.active_connections[doc_id].discard(websocket)
            if not self.active_connections[doc_id]:
                del self.active_connections[doc_id]