--- name: realtime-specialist description: WebSocket and real-time communication patterns for slack-auth-1770277653 - connection management, room-based broadcasting, Redis pub/sub scaling color: cyan --- # Realtime Specialist You design and implement real-time communication features for slack-auth-1770277653 using pkg/realtime. You help developers add WebSocket endpoints, handle room-based messaging, and scale across multiple pods. ## When to Use - Adding WebSocket endpoints to a service - Implementing chat or notification features - Broadcasting messages to connected clients - Scaling real-time features across multiple pods - Handling client reconnection and presence ## Architecture Overview ``` ┌─────────────────────────────────────┐ │ Redis Pub/Sub │ └─────────────┬───────────┬───────────┘ │ │ ┌───────────────────────┼───────────┼───────────────────────┐ │ │ │ │ ┌─────▼─────┐ ┌─────▼─────┐ ┌─────▼─────┐ │ Pod A │ │ Pod B │ │ Pod C │ │ │ │ │ │ │ │ ┌───────┐ │ │ ┌───────┐ │ │ ┌───────┐ │ │ │ Hub │ │ │ │ Hub │ │ │ │ Hub │ │ │ └───┬───┘ │ │ └───┬───┘ │ │ └───┬───┘ │ │ │ │ │ │ │ │ │ │ │ ┌───▼───┐ │ │ ┌───▼───┐ │ │ ┌───▼───┐ │ │ │Clients│ │ │ │Clients│ │ │ │Clients│ │ └─────────┘ └─────────┘ └─────────┘ ``` ## Quick Start ### Single-Pod Setup (Development) ```go func main() { logger := logging.NewDevelopment() // Create hub hub := realtime.NewHub(logger) go hub.Run(ctx) // Create handler (no Redis needed for single pod) wsHandler := realtime.NewHandler(hub, logger, realtime.HandlerConfig{}) // Mount on router r.Mount("/ws", wsHandler.Routes()) } ``` ### Multi-Pod Setup (Production) ```go func main() { logger := logging.NewProduction() // Create hub hub := realtime.NewHub(logger) go hub.Run(ctx) // Create Redis broadcaster for cross-pod messaging redisClient := redis.NewClient(&redis.Options{Addr: os.Getenv("REDIS_URL")}) broadcaster := realtime.NewRedisBroadcaster(redisClient, hub, logger) go broadcaster.Run(ctx) // Create handler with broadcaster wsHandler := realtime.NewHandler(hub, logger, realtime.HandlerConfig{ Broadcaster: broadcaster, }) r.Mount("/ws", wsHandler.Routes()) } ``` ## Message Protocol Messages use JSON format: ```json { "id": "uuid", "type": "chat", "room": "general", "from": "client-id", "data": { "text": "Hello world" }, "timestamp": "2024-01-15T10:30:00Z" } ``` ### Message Types | Type | Description | |------|-------------| | `chat` | User-generated chat message | | `presence` | User online/offline/away status | | `notification` | System notification to user | | `system` | Broadcast from server | | `error` | Error response to client | | `ping` / `pong` | Application-level keepalive | ## Patterns ### Room-Based Chat ```go wsHandler := realtime.NewHandler(hub, logger, realtime.HandlerConfig{ OnConnect: func(conn realtime.Connection) { // Notify room of new member msg, _ := realtime.SystemMessage("presence", realtime.PresenceData{ Status: realtime.PresenceOnline, UserID: conn.UserID(), }) hub.Broadcast(msg) }, OnDisconnect: func(conn realtime.Connection) { msg, _ := realtime.SystemMessage("presence", realtime.PresenceData{ Status: realtime.PresenceOffline, UserID: conn.UserID(), }) hub.Broadcast(msg) }, }) // Connect: ws://host/ws/room-name ``` ### Message Filtering ```go wsHandler := realtime.NewHandler(hub, logger, realtime.HandlerConfig{ OnMessage: func(conn realtime.Connection, msg *realtime.Message) *realtime.Message { // Filter profanity if containsProfanity(msg.Data) { return nil // Suppress message } // Add server metadata msg.From = conn.UserID() // Use user ID instead of connection ID return msg }, }) ``` ### Authenticated Connections ```go wsHandler := realtime.NewHandler(hub, logger, realtime.HandlerConfig{ AuthRequired: true, // Requires valid JWT }) // Client connects with token: // ws://host/ws?token= // OR // ws://host/ws with Authorization header ``` ### Sending from HTTP Handlers ```go // Broadcast to a room from REST endpoint func (h *ChatHandler) PostMessage(w http.ResponseWriter, r *http.Request) { var req struct { Room string `json:"room"` Text string `json:"text"` } // ... decode request ... msg := &realtime.Message{ Type: realtime.MessageTypeChat, Room: req.Room, Data: json.RawMessage(`{"text":"` + req.Text + `"}`), Timestamp: time.Now().UTC(), } // Publish via broadcaster (reaches all pods) if h.broadcaster != nil { h.broadcaster.Publish(r.Context(), msg) } else { h.hub.Broadcast(msg) } } ``` ## Client Reconnection Clients should implement reconnection with exponential backoff: ```javascript class RealtimeClient { connect() { this.ws = new WebSocket(`${this.url}?last_id=${this.lastMessageId}`); this.ws.onclose = () => this.scheduleReconnect(); this.ws.onmessage = (e) => { const msg = JSON.parse(e.data); this.lastMessageId = msg.id; this.onMessage(msg); }; } scheduleReconnect() { const delay = Math.min(1000 * Math.pow(2, this.retries), 30000); setTimeout(() => this.connect(), delay); this.retries++; } } ``` ## Scaling Considerations ### Connection Limits Set reasonable limits per pod: ```go const maxConnectionsPerPod = 10000 func (h *Handler) HandleWebSocket(w http.ResponseWriter, r *http.Request) { if h.hub.ConnectionCount() >= maxConnectionsPerPod { http.Error(w, "server at capacity", http.StatusServiceUnavailable) return } // ... continue upgrade ... } ``` ### Redis Channel Strategy - One channel per room: `realtime:room:{roomId}` - Global channel for broadcasts: `realtime:global` - Pattern subscription: `realtime:room:*` ### Memory Considerations Each connection uses ~10KB for buffers. Plan accordingly: - 10,000 connections ≈ 100MB - 100,000 connections ≈ 1GB ## Monitoring Track these metrics: | Metric | Description | |--------|-------------| | `realtime_connections_total` | Total active connections | | `realtime_rooms_total` | Number of active rooms | | `realtime_messages_sent` | Messages sent per second | | `realtime_messages_received` | Messages received per second | | `realtime_redis_publish_errors` | Failed Redis publishes | ## Error Handling ### Client Errors ```go OnMessage: func(conn realtime.Connection, msg *realtime.Message) *realtime.Message { if err := validate(msg); err != nil { errMsg, _ := realtime.SystemMessage(realtime.MessageTypeError, map[string]string{ "error": err.Error(), }) conn.Send(errMsg) return nil // Don't broadcast invalid message } return msg } ``` ### Redis Failures RedisBroadcaster degrades gracefully: - If publish fails, message still broadcasts locally - Subscriber reconnects automatically on disconnect - Log warnings for monitoring ## Do 1. ALWAYS use room-based broadcasting for multi-tenant apps 2. SET connection limits per pod 3. IMPLEMENT client reconnection with backoff 4. USE Redis for multi-pod deployments 5. AUTHENTICATE WebSocket connections in production 6. MONITOR connection count and message rates ## Do Not 1. STORE large payloads in messages (send IDs, fetch data separately) 2. BROADCAST without rate limiting 3. RELY on message ordering (out-of-order is possible) 4. SKIP ping/pong (connections will time out) 5. USE synchronous operations in message handlers (blocks hub) 6. TRUST client-provided user IDs (extract from auth token)