implementing-realtime-sync

Real-Time Sync

Implement real-time communication for live updates, collaboration, and presence awareness across applications.

When to Use

Use this skill when building:

• LLM streaming interfaces - Stream tokens progressively (ai-chat integration)
• Live dashboards - Push metrics and updates to clients
• Collaborative editing - Multi-user document/spreadsheet editing with CRDTs
• Chat applications - Real-time messaging with presence
• Multiplayer features - Cursor tracking, live updates, presence awareness
• Offline-first apps - Mobile/PWA with sync-on-reconnect

Protocol Selection Framework

Choose the transport protocol based on communication pattern:

Decision Tree

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

Protocol Comparison

Protocol	Direction	Reconnection	Complexity	Best For
SSE	Server → Client	Automatic	Low	Live feeds, LLM streaming
WebSocket	Bidirectional	Manual	Medium	Chat, games, collaboration
WebRTC	P2P	Complex	High	Video, screen share, voice

Implementation Patterns

Pattern 1: LLM Streaming with SSE

Stream LLM tokens progressively to frontend (ai-chat integration).

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())