implementing-realtime-sync

リアルタイム同期

アプリケーション間でライブ更新、コラボレーション、プレゼンス認識を実現するリアルタイム通信を実装します。

使用する場面

次の機能を構築する場合に、このスキルを使用します。

• LLM ストリーミングインターフェース - トークンを段階的にストリーミングします（ai-chat との統合）。
• ライブダッシュボード - クライアントにメトリクスや更新をプッシュします。
• 共同編集 - CRDT を使った複数ユーザーでのドキュメントやスプレッドシートの編集。
• チャットアプリケーション - プレゼンス機能を備えたリアルタイムメッセージング。
• マルチプレイヤー機能 - カーソル追跡、ライブ更新、プレゼンス認識。
• オフライン優先アプリ - 再接続時に同期するモバイルアプリ／PWA。

プロトコル選択の指針

通信パターンに基づいて、トランスポートプロトコルを選択します。

判断フロー

ONE-WAY (Server → Client only)
├─ LLM streaming, notifications, live feeds
└─ Use SSE (Server-Sent Events)
   ├─ Automatic reconnection (browser-native)
   ├─ Event IDs for resumption
   └─ Simple HTTP implementation

BIDIRECTIONAL (Client ↔ Server)
├─ Chat, games, collaborative editing
└─ Use WebSocket
   ├─ Manual reconnection required
   ├─ Binary + text support
   └─ Lower latency for two-way

COLLABORATIVE EDITING
├─ Multi-user documents/spreadsheets
└─ Use WebSocket + CRDT (Yjs or Automerge)
   ├─ CRDT handles conflict resolution
   ├─ WebSocket for transport
   └─ Offline-first with sync

PEER-TO-PEER MEDIA
├─ Video, screen sharing, voice calls
└─ Use WebRTC
   ├─ WebSocket for signaling
   ├─ Direct P2P connection
   └─ STUN/TURN for NAT traversal

プロトコルの比較

プロトコル	方向	再接続	複雑さ	最適な用途
SSE	サーバー → クライアント	自動	低	ライブフィード、LLM ストリーミング
WebSocket	双方向	手動	中	チャット、ゲーム、コラボレーション
WebRTC	P2P	複雑	高	ビデオ、画面共有、音声

実装パターン

パターン 1：SSE による LLM ストリーミング

LLM のトークンをフロントエンドに段階的にストリーミングします（ai-chat との統合）。

Python（FastAPI）：

from sse_starlette.sse import EventSourceResponse

@app.post("/chat/stream")
async def stream_chat(prompt: str):
    async def generate():
        async for chunk in llm_stream:
            yield {"event": "token", "data": chunk.content}
        yield {"event": "done", "data": "[DONE]"}
    return EventSourceResponse(generate())

フロントエンド：

const es = new EventSource('/chat/stream')
es.addEventListener('token', (e) => appendToken(e.data))

完全な実装、再接続、イベント ID による再開については、references/sse.md を参照してください。

パターン 2：WebSocket チャット

チャットアプリケーション向けの双方向通信です。

Python（FastAPI）：

connections: set[WebSocket] = set()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    connections.add(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            for conn in connections:
                await conn.send_text(data)
    except WebSocketDisconnect:
        connections.remove(websocket)

多言語の例、認証、ハートビート、スケーリングについては、references/websockets.md を参照してください。

パターン 3：CRDT による共同編集

Yjs を使った、競合のない複数ユーザーでの編集です。

TypeScript（Yjs）：

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

const doc = new Y.Doc()
const provider = new WebsocketProvider('ws://localhost:1234', 'doc-id', doc)
const ytext = doc.getText('content')

ytext.observe(event => console.log('Changes:', event.changes))
ytext.insert(0, 'Hello collaborative world!')

競合解決、Yjs と Automerge の比較、高度なパターンについては、references/crdts.md を参照してください。

パターン 4：プレゼンス認識

オンラインユーザー、カーソル位置、入力中インジケーターを追跡します。

Yjs Awareness API：

const awareness = provider.awareness
awareness.setLocalState({ user: { name: 'Alice' }, cursor: { x: 100, y: 200 } })
awareness.on('change', () => {
  awareness.getStates().forEach((state, clientId) => {
    renderCursor(state.cursor, state.user)
  })
})

カーソル追跡、入力中インジケーター、オンライン状態については、references/presence-patterns.md を参照してください。

パターン 5：オフライン同期（モバイル／PWA）

変更をローカルにキューイングし、接続が復旧したら同期します。

TypeScript（Yjs + IndexedDB）：

import { IndexeddbPersistence } from 'y-indexeddb'
import { WebsocketProvider } from 'y-websocket'

const doc = new Y.Doc()
const indexeddbProvider = new IndexeddbPersistence('my-doc', doc)
const wsProvider = new WebsocketProvider('wss://api.example.com/sync', 'my-doc', doc)

wsProvider.on('status', (e) => {
  console.log(e.status === 'connected' ? 'Online' : 'Offline')
})

競合解決と同期戦略については、references/offline-sync.md を参照してください。

ライブラリの推奨事項

Python

WebSocket：

• websockets 13.x - AsyncIO ベースで、本番環境対応
• FastAPI WebSocket - 組み込み、依存性注入に対応
• Flask-SocketIO - フォールバック機能を備えた Socket.IO プロトコル

SSE：

• sse-starlette - FastAPI／Starlette 対応、非同期、ジェネレーターベース
• Flask-SSE - Pub/Sub 用の Redis バックエンド

Rust

WebSocket：

• tokio-tungstenite 0.23 - Tokio と統合され、本番環境対応
• axum WebSocket - 組み込みのエクストラクター、tower ミドルウェア

SSE：

• axum SSE - ネイティブ対応、非同期ストリーム

Go

WebSocket：

• gorilla/websocket - 実績があり、圧縮に対応
• nhooyr/websocket - モダンな API、コンテキストに対応

SSE：

• net/http（ネイティブ） - Flusher インターフェースを使用し、依存関係なし

TypeScript

WebSocket：

• ws - ネイティブ WebSocket サーバー、軽量
• Socket.io 4.x - 自動再接続、フォールバック、ルーム機能
• Hono WebSocket - エッジランタイム（Cloudflare Workers、Deno）に対応

SSE：

• EventSource（ネイティブ） - ブラウザー標準で、自動再試行
• Node.js http（ネイティブ） - サーバーサイドで使用でき、依存関係なし

CRDT：

• Yjs - 成熟しており、TypeScript／Rust に対応。リッチテキスト編集に対応
• Automerge - Rust／JS 対応、JSON に似たデータ、時間をさかのぼる機能

再接続戦略

SSE： ブラウザーの EventSource が指数バックオフを使って再接続を自動的に処理します。 WebSocket： 接続の集中を防ぐため、ジッターを加えた指数バックオフを手動で実装します。

完全な実装パターンについては、references/sse.md と references/websockets.md を参照してください。

セキュリティパターン

認証： 同一オリジンでは Cookie ベース、または Sec-WebSocket-Protocol ヘッダー内のトークンを使用します。 レート制限： スライディングウィンドウを使用して、ユーザーごとにメッセージを制限します。

認証とレート制限の実装については、references/websockets.md を参照してください。

Redis Pub/Sub によるスケーリング

水平スケーリングには、Redis pub/sub を使用して複数のバックエンドサーバー間でメッセージをブロードキャストします。

Redis によるスケーリングの完全な実装については、references/websockets.md を参照してください。

フロントエンドとの統合

React Hooks パターン

LLM ストリーミング用 SSE（ai-chat）：

useEffect(() => {
  const es = new EventSource(`/api/chat/stream?prompt=${prompt}`)
  es.addEventListener('token', (e) => setContent(prev => prev + e.data))
  return () => es.close()
}, [prompt])

ライブメトリクス用 WebSocket（ダッシュボード）：

useEffect(() => {
  const ws = new WebSocket('ws://localhost:8000/metrics')
  ws.onmessage = (e) => setMetrics(JSON.parse(e.data))
  return () => ws.close()
}, [])

共同編集テーブル用 Yjs：

useEffect(() => {
  const doc = new Y.Doc()
  const provider = new WebsocketProvider('ws://localhost:1234', docId, doc)
  const yarray = doc.getArray('rows')
  yarray.observe(() => setRows(yarray.toArray()))
  return () => provider.destroy()
}, [docId])

リファレンスドキュメント

詳しい実装パターンについては、以下を参照してください。

• references/sse.md - SSE プロトコル、再接続、イベント ID
• references/websockets.md - WebSocket の認証、ハートビート、スケーリング
• references/crdts.md - Yjs と Automerge の比較、競合解決
• references/presence-patterns.md - カーソル追跡、入力インジケーター
• references/offline-sync.md - モバイルのパターン、競合対応の戦略

サンプルプロジェクト

以下で動作する実装を利用できます。

• examples/llm-streaming-sse/ - LLM ストリーミング用 FastAPI SSE（実行可能）
• examples/chat-websocket/ - Python FastAPI と TypeScript によるチャット
• examples/collaborative-yjs/ - Yjs による共同編集エディター

テストツール

スクリプトを使用して実装を検証します。

• scripts/test_websocket_connection.py - WebSocket 接続のテスト