Real-Time Systems

┌─────────────────────────────────────────────────────────────────┐
│                REAL-TIME COMMUNICATION METHODS                  │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  1. HTTP Polling                                                │
│     Client → Server (every N seconds)                          │
│     Simple but wasteful                                         │
│                                                                 │
│  2. Long Polling                                                │
│     Client → Server (waits for response)                       │
│     Better, still overhead per request                         │
│                                                                 │
│  3. Server-Sent Events (SSE)                                   │
│     Server → Client (one-way stream)                           │
│     Simple, HTTP-based, auto-reconnect                         │
│                                                                 │
│  4. WebSockets                                                  │
│     Client ↔ Server (bidirectional)                            │
│     Full duplex, low latency, complex                          │
│                                                                 │
│  5. WebRTC                                                      │
│     Client ↔ Client (peer-to-peer)                             │
│     Audio/video, lowest latency                                │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────┐
│                    WEBSOCKET LIFECYCLE                          │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│  1. HTTP Upgrade Request                                        │
│     GET /chat HTTP/1.1                                         │
│     Upgrade: websocket                                          │
│     Connection: Upgrade                                         │
│     Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==                │
│                                                                 │
│  2. Server Accepts                                              │
│     HTTP/1.1 101 Switching Protocols                           │
│     Upgrade: websocket                                          │
│     Connection: Upgrade                                         │
│     Sec-WebSocket-Accept: s3pPLMBiTxaQ9kYGzzhZRbK+xOo=         │
│                                                                 │
│  3. WebSocket Connection Established                           │
│     ┌────────────────────────────────────────────────────────┐ │
│     │ Client ◄══════════════════════════════════════► Server │ │
│     │        Full duplex binary/text frames                  │ │
│     └────────────────────────────────────────────────────────┘ │
│                                                                 │
│  4. Either side can close                                      │
│     Client sends Close frame → Server acknowledges → Done     │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────────┐
│              SCALABLE WEBSOCKET ARCHITECTURE                    │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│                         Clients                                 │
│                    ┌───┐ ┌───┐ ┌───┐                          │
│                    │ C │ │ C │ │ C │                          │
│                    └─┬─┘ └─┬─┘ └─┬─┘                          │
│                      │     │     │                              │
│                    ┌─┴─────┴─────┴─┐                          │
│                    │  Load Balancer │ (sticky sessions)        │
│                    └───────┬───────┘                          │
│           ┌────────────────┼────────────────┐                  │
│           │                │                │                   │
│      ┌────▼────┐      ┌────▼────┐      ┌────▼────┐            │
│      │  WS     │      │  WS     │      │  WS     │            │
│      │ Server 1│      │ Server 2│      │ Server 3│            │
│      └────┬────┘      └────┬────┘      └────┬────┘            │
│           │                │                │                   │
│           └────────────────┼────────────────┘                  │
│                            │                                    │
│                    ┌───────▼───────┐                           │
│                    │ Message Broker│ Redis Pub/Sub             │
│                    │   (Pub/Sub)   │ or Kafka                  │
│                    └───────────────┘                           │
│                                                                 │
│  Challenge: User A on Server 1 messages User B on Server 2    │
│  Solution: Pub/Sub to broadcast across servers                 │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from typing import AsyncGenerator
import asyncio
import json

app = FastAPI()

class EventBroker:
    """Simple in-memory event broker for SSE"""
    
    def __init__(self):
        self.subscribers: Dict[str, asyncio.Queue] = {}
    
    async def subscribe(self, user_id: str) -> asyncio.Queue:
        queue = asyncio.Queue()
        self.subscribers[user_id] = queue
        return queue
    
    def unsubscribe(self, user_id: str):
        self.subscribers.pop(user_id, None)
    
    async def publish(self, user_id: str, event: dict):
        if user_id in self.subscribers:
            await self.subscribers[user_id].put(event)
    
    async def broadcast(self, event: dict):
        for queue in self.subscribers.values():
            await queue.put(event)

broker = EventBroker()

async def event_stream(user_id: str) -> AsyncGenerator[str, None]:
    """Generate SSE stream for a user"""
    queue = await broker.subscribe(user_id)
    
    try:
        # Send initial connection event
        yield f"event: connected\ndata: {json.dumps({'user_id': user_id})}\n\n"
        
        while True:
            try:
                # Wait for event with timeout (for keepalive)
                event = await asyncio.wait_for(queue.get(), timeout=30)
                
                event_type = event.get("type", "message")
                data = json.dumps(event)
                
                yield f"event: {event_type}\ndata: {data}\n\n"
                
            except asyncio.TimeoutError:
                # Send keepalive comment
                yield ": keepalive\n\n"
    
    finally:
        broker.unsubscribe(user_id)

@app.get("/events/{user_id}")
async def sse_endpoint(user_id: str):
    """SSE endpoint for real-time notifications"""
    return StreamingResponse(
        event_stream(user_id),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no"  # Disable nginx buffering
        }
    )

@app.post("/notify/{user_id}")
async def send_notification(user_id: str, message: dict):
    """Send notification to a user"""
    await broker.publish(user_id, {
        "type": "notification",
        **message
    })
    return {"status": "sent"}

from fastapi import FastAPI
from typing import Optional
import asyncio

app = FastAPI()

class LongPollManager:
    def __init__(self):
        self.pending: Dict[str, asyncio.Future] = {}
        self.messages: Dict[str, list] = {}
    
    async def wait_for_messages(
        self, 
        user_id: str, 
        timeout: float = 30
    ) -> list:
        """Wait for new messages with timeout"""
        
        # Check for pending messages first
        if user_id in self.messages and self.messages[user_id]:
            messages = self.messages.pop(user_id)
            return messages
        
        # Create a future to wait on
        future = asyncio.get_event_loop().create_future()
        self.pending[user_id] = future
        
        try:
            messages = await asyncio.wait_for(future, timeout=timeout)
            return messages
        except asyncio.TimeoutError:
            return []
        finally:
            self.pending.pop(user_id, None)
    
    async def send_message(self, user_id: str, message: dict):
        """Send message to user, either immediately or queue"""
        if user_id in self.pending:
            # User is waiting, resolve immediately
            future = self.pending.pop(user_id)
            if not future.done():
                future.set_result([message])
        else:
            # Queue for next poll
            if user_id not in self.messages:
                self.messages[user_id] = []
            self.messages[user_id].append(message)

poll_manager = LongPollManager()

@app.get("/poll/{user_id}")
async def long_poll(user_id: str, timeout: float = 30):
    """Long polling endpoint"""
    messages = await poll_manager.wait_for_messages(user_id, timeout)
    return {"messages": messages}

@app.post("/send/{user_id}")
async def send_message(user_id: str, message: dict):
    """Send message to user"""
    await poll_manager.send_message(user_id, message)
    return {"status": "sent"}

from datetime import datetime, timedelta
from typing import Dict, Set, List
import asyncio
import redis.asyncio as redis

class PresenceSystem:
    """
    Track online/offline status across servers.
    Uses Redis with TTL for automatic cleanup.
    """
    
    HEARTBEAT_INTERVAL = 30  # seconds
    PRESENCE_TTL = 60  # seconds (2x heartbeat)
    
    def __init__(self, redis_client: redis.Redis, server_id: str):
        self.redis = redis_client
        self.server_id = server_id
    
    async def set_online(self, user_id: str):
        """Mark user as online"""
        key = f"presence:{user_id}"
        await self.redis.hset(key, mapping={
            "server_id": self.server_id,
            "status": "online",
            "last_seen": datetime.utcnow().isoformat()
        })
        await self.redis.expire(key, self.PRESENCE_TTL)
        
        # Add to online set
        await self.redis.sadd("online_users", user_id)
    
    async def heartbeat(self, user_id: str):
        """Refresh presence TTL"""
        key = f"presence:{user_id}"
        
        # Update last_seen and extend TTL
        await self.redis.hset(key, "last_seen", datetime.utcnow().isoformat())
        await self.redis.expire(key, self.PRESENCE_TTL)
    
    async def set_offline(self, user_id: str):
        """Mark user as offline"""
        await self.redis.delete(f"presence:{user_id}")
        await self.redis.srem("online_users", user_id)
    
    async def is_online(self, user_id: str) -> bool:
        """Check if user is online"""
        return await self.redis.exists(f"presence:{user_id}")
    
    async def get_status(self, user_id: str) -> Dict:
        """Get user's presence status"""
        data = await self.redis.hgetall(f"presence:{user_id}")
        
        if not data:
            return {"status": "offline"}
        
        return {
            "status": data.get("status", "offline"),
            "server_id": data.get("server_id"),
            "last_seen": data.get("last_seen")
        }
    
    async def get_online_users(self) -> Set[str]:
        """Get all online users"""
        return await self.redis.smembers("online_users")
    
    async def get_online_count(self) -> int:
        """Get count of online users"""
        return await self.redis.scard("online_users")
    
    async def get_friends_online(self, user_id: str, friend_ids: List[str]) -> List[str]:
        """Get which friends are online"""
        online_friends = []
        
        # Use pipeline for efficiency
        async with self.redis.pipeline() as pipe:
            for friend_id in friend_ids:
                pipe.exists(f"presence:{friend_id}")
            results = await pipe.execute()
        
        for friend_id, is_online in zip(friend_ids, results):
            if is_online:
                online_friends.append(friend_id)
        
        return online_friends
    
    async def cleanup_stale(self):
        """Remove stale presence entries (run periodically)"""
        online_users = await self.redis.smembers("online_users")
        
        for user_id in online_users:
            if not await self.redis.exists(f"presence:{user_id}"):
                await self.redis.srem("online_users", user_id)

class TypingIndicator:
    """
    Track and broadcast typing indicators.
    Uses short TTLs to auto-expire.
    """
    
    TYPING_TTL = 3  # seconds
    
    def __init__(self, redis_client: redis.Redis, broadcast_fn):
        self.redis = redis_client
        self.broadcast = broadcast_fn
    
    async def set_typing(self, user_id: str, room_id: str):
        """Mark user as typing in a room"""
        key = f"typing:{room_id}:{user_id}"
        
        # Only broadcast if not already typing
        is_new = not await self.redis.exists(key)
        
        await self.redis.setex(key, self.TYPING_TTL, "1")
        
        if is_new:
            await self.broadcast(room_id, {
                "type": "typing_start",
                "user_id": user_id,
                "room_id": room_id
            })
    
    async def stop_typing(self, user_id: str, room_id: str):
        """Mark user as stopped typing"""
        key = f"typing:{room_id}:{user_id}"
        
        if await self.redis.delete(key):
            await self.broadcast(room_id, {
                "type": "typing_stop",
                "user_id": user_id,
                "room_id": room_id
            })
    
    async def get_typing_users(self, room_id: str) -> List[str]:
        """Get all users currently typing in a room"""
        pattern = f"typing:{room_id}:*"
        keys = await self.redis.keys(pattern)
        
        return [key.split(":")[-1] for key in keys]