implementing-websocket-realtime

Original source / Raw SKILL.md

---
name: implementing-websocket-realtime
description: Use this skill when implementing real-time features with WebSockets, including connection lifecycle management, pub/sub event patterns, state synchronization between client and server, event broadcasting, and integration with frontend state management (React Query, SWR, etc.). Covers FastAPI/Node.js/Express backends and React/Next.js/Vue frontends with fallback strategies and error recovery.
---

# WebSocket Real-Time Implementation

## Decision Tree: Real-Time Protocol Selection

| Requirement | Protocol | Why |
|-------------|----------|-----|
| Bidirectional, instant updates | **WebSocket** | Full duplex, low latency |
| Server → client only | **SSE** | Simpler, auto-reconnect |
| Infrequent updates (<1/min) | **Polling** | Simpler, no persistent connection |
| Mobile/unstable network | **WS + Fallback** | Graceful degradation |

**Decision**: Use WebSocket for real-time collaboration, SSE for notifications, polling as last resort.

## Generic Event Structure

```typescript
interface WebSocketEvent {
  topic: string;              // "resource-type:identifier" (e.g., "gift-list:family-123")
  event: EventType;           // ADDED | UPDATED | DELETED | STATUS_CHANGED | CUSTOM
  data: {
    entity_id: string;        // Resource ID
    payload: unknown;         // Event-specific data (DTO)
    user_id?: string;         // Who triggered (optional)
    tenant_id?: string;       // Multi-tenant context (optional)
    timestamp: string;        // ISO 8601
  };
  trace_id?: string;          // For observability
}

type EventType = "ADDED" | "UPDATED" | "DELETED" | "STATUS_CHANGED" | string;
```

**Validation**: See `./scripts/validate-ws-event.js`

## Connection Lifecycle

### 1. Initial Connection

```typescript
// Client
const ws = new WebSocket(`${WS_URL}?token=${authToken}`);

ws.onopen = () => {
  console.log('Connected');
  subscribeToTopics(['gift-list:123', 'user:456']);
};
```

```python
# Server (FastAPI)
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket, token: str):
    await websocket.accept()
    user = authenticate(token)  # Validate on connect
    connection_manager.connect(websocket, user.id)
```

### 2. Authentication

**Options**:
- **Query param**: `?token=jwt` (simplest)
- **First message**: Send auth message after connect
- **Cookie**: Use existing HTTP session

**Recommendation**: Query param for stateless, first message for flexible auth.

### 3. Subscription Management

```typescript
// Subscribe to topics
function subscribe(topics: string[]) {
  ws.send(JSON.stringify({
    type: 'subscribe',
    topics: ['gift-list:123', 'user:456']
  }));
}

// Unsubscribe on unmount
function unsubscribe(topics: string[]) {
  ws.send(JSON.stringify({
    type: 'unsubscribe',
    topics: ['gift-list:123']
  }));
}
```

### 4. Heartbeat/Keepalive

```typescript
// Client: Ping every 30s
const heartbeat = setInterval(() => {
  if (ws.readyState === WebSocket.OPEN) {
    ws.send(JSON.stringify({ type: 'ping' }));
  }
}, 30000);

// Server responds with pong
ws.onmessage = (event) => {
  const msg = JSON.parse(event.data);
  if (msg.type === 'pong') {
    lastPong = Date.now();
  }
};
```

### 5. Reconnection Logic

```typescript
let reconnectAttempts = 0;
const MAX_RECONNECT_ATTEMPTS = 5;
const INITIAL_DELAY = 1000;

function reconnect() {
  if (reconnectAttempts >= MAX_RECONNECT_ATTEMPTS) {
    console.error('Max reconnection attempts reached');
    return;
  }

  const delay = INITIAL_DELAY * Math.pow(2, reconnectAttempts); // Exponential backoff
  setTimeout(() => {
    reconnectAttempts++;
    connect();
  }, delay);
}

ws.onclose = () => {
  console.log('Connection closed, reconnecting...');
  reconnect();
};
```

### 6. Cleanup

```typescript
// On component unmount
useEffect(() => {
  return () => {
    if (ws) {
      ws.close();
      clearInterval(heartbeat);
    }
  };
}, []);
```

**Details**: See `./connection-lifecycle.md`

## State Synchronization Pattern

### Flow

```
1. Load initial data    → REST API (React Query)
2. Subscribe to updates → WebSocket (on mount)
3. Receive event        → Invalidate cache → React Query refetches
4. Optimistic update    → Update UI immediately, rollback on error
5. Unsubscribe          → WebSocket (on unmount)
6. Fallback             → Poll every 10s if WS fails
```

### Implementation

```typescript
// 1. Load initial data
const { data, isLoading } = useQuery({
  queryKey: ['gift-list', listId],
  queryFn: () => fetchGiftList(listId),
});

// 2. Subscribe to WebSocket updates
useEffect(() => {
  const ws = connectWebSocket();

  ws.onmessage = (event) => {
    const wsEvent: WebSocketEvent = JSON.parse(event.data);

    // 3. Invalidate cache on event
    if (wsEvent.topic === `gift-list:${listId}`) {
      queryClient.invalidateQueries(['gift-list', listId]);
    }
  };

  ws.send(JSON.stringify({
    type: 'subscribe',
    topics: [`gift-list:${listId}`]
  }));

  return () => {
    ws.send(JSON.stringify({
      type: 'unsubscribe',
      topics: [`gift-list:${listId}`]
    }));
    ws.close();
  };
}, [listId]);

// 4. Optimistic update
const mutation = useMutation({
  mutationFn: updateGift,
  onMutate: async (newGift) => {
    await queryClient.cancelQueries(['gift-list', listId]);
    const previous = queryClient.getQueryData(['gift-list', listId]);
    queryClient.setQueryData(['gift-list', listId], (old) => ({
      ...old,
      gifts: old.gifts.map(g => g.id === newGift.id ? newGift : g)
    }));
    return { previous };
  },
  onError: (err, newGift, context) => {
    queryClient.setQueryData(['gift-list', listId], context.previous);
  },
});
```

**Details**: See `./state-sync-strategies.md`

## Implementation Checklist

### Backend Setup

- [ ] WebSocket server endpoint
  - [ ] FastAPI: `@app.websocket("/ws")`
  - [ ] Node.js: `ws` or `socket.io` library
- [ ] Authentication on connect
- [ ] Connection manager (track active connections)
- [ ] Topic subscription logic
- [ ] Event broadcasting
  - [ ] Per-topic subscribers
  - [ ] Per-user filtering (if multi-tenant)
- [ ] Heartbeat/pong handler
- [ ] Error handling & logging
- [ ] Trace IDs for observability

### Frontend Setup

- [ ] WebSocket connection hook
- [ ] Auto-reconnection logic
- [ ] Subscription management
- [ ] Event handlers
- [ ] State management integration
  - [ ] React Query invalidation
  - [ ] SWR revalidation
  - [ ] Custom state updates
- [ ] Optimistic updates
- [ ] Connection status UI
- [ ] Fallback polling (if WS fails)
- [ ] Cleanup on unmount

### Testing

- [ ] Connection establishment
- [ ] Authentication flow
- [ ] Subscription/unsubscription
- [ ] Event delivery
- [ ] Reconnection logic
- [ ] Fallback behavior
- [ ] Load testing (concurrent connections)
- [ ] Network failure scenarios

**Details**: See `./backend-patterns.md` and `./frontend-patterns.md`

## Backend Patterns

### FastAPI Connection Manager

```python
from fastapi import WebSocket
from typing import Dict, Set

class ConnectionManager:
    def __init__(self):
        self.active_connections: Dict[str, WebSocket] = {}
        self.subscriptions: Dict[str, Set[str]] = {}  # topic -> set of user_ids

    async def connect(self, websocket: WebSocket, user_id: str):
        self.active_connections[user_id] = websocket

    def disconnect(self, user_id: str):
        if user_id in self.active_connections:
            del self.active_connections[user_id]

    def subscribe(self, user_id: str, topic: str):
        if topic not in self.subscriptions:
            self.subscriptions[topic] = set()
        self.subscriptions[topic].add(user_id)

    async def broadcast(self, topic: str, event: dict):
        if topic not in self.subscriptions:
            return

        for user_id in self.subscriptions[topic]:
            if user_id in self.active_connections:
                ws = self.active_connections[user_id]
                await ws.send_json(event)

manager = ConnectionManager()
```

**Full examples**: See `./backend-patterns.md`

## Frontend Patterns

### React Hook: useWebSocket

```typescript
function useWebSocket(topics: string[]) {
  const [isConnected, setIsConnected] = useState(false);
  const wsRef = useRef<WebSocket | null>(null);
  const queryClient = useQueryClient();

  useEffect(() => {
    const ws = new WebSocket(`${WS_URL}?token=${getToken()}`);

    ws.onopen = () => {
      setIsConnected(true);
      ws.send(JSON.stringify({ type: 'subscribe', topics }));
    };

    ws.onmessage = (event) => {
      const wsEvent: WebSocketEvent = JSON.parse(event.data);

      // Invalidate relevant queries
      queryClient.invalidateQueries([wsEvent.topic.split(':')[0]]);
    };

    ws.onclose = () => {
      setIsConnected(false);
      // Reconnect logic here
    };

    wsRef.current = ws;

    return () => {
      ws.send(JSON.stringify({ type: 'unsubscribe', topics }));
      ws.close();
    };
  }, [topics.join(',')]);

  return { isConnected };
}
```

**Full examples**: See `./frontend-patterns.md`

## Progressive Disclosure References

For detailed patterns and examples:

- **Connection Lifecycle**: `./connection-lifecycle.md`
- **Event Structure & Validation**: `./event-structure-patterns.md`
- **State Sync Strategies**: `./state-sync-strategies.md`
- **Backend Implementations**: `./backend-patterns.md`
- **Frontend Implementations**: `./frontend-patterns.md`
- **Fallback & Recovery**: `./fallback-strategies.md`
- **Event Validator Script**: `./scripts/validate-ws-event.js`


---

## Referenced Files

> The following files are referenced in this skill and included for context.

### scripts/validate-ws-event.js

```javascript
#!/usr/bin/env node
/**
 * WebSocket Event Validator
 * Validates event structure against schema.
 *
 * Usage:
 *   node validate-ws-event.js '{"topic":"gift-list:123","event":"UPDATED",...}'
 *   echo '{"topic":"..."}' | node validate-ws-event.js
 */

const TOPIC_PATTERN = /^[a-z][a-z0-9-]*:[a-z0-9-]+$/;
const VALID_EVENTS = ['ADDED', 'UPDATED', 'DELETED', 'STATUS_CHANGED'];

function validate(event) {
  const errors = [];
  const warnings = [];

  // Required fields
  if (!event.topic) {
    errors.push('Missing required field: topic');
  } else if (!TOPIC_PATTERN.test(event.topic)) {
    errors.push(`Invalid topic format "${event.topic}". Expected: "{resource}:{id}" (e.g., "gift-list:123")`);
  }

  if (!event.event) {
    errors.push('Missing required field: event');
  } else if (!VALID_EVENTS.includes(event.event) && !event.event.match(/^[A-Z_]+$/)) {
    warnings.push(`Non-standard event type "${event.event}". Standard: ${VALID_EVENTS.join(', ')}`);
  }

  if (!event.data) {
    errors.push('Missing required field: data');
  } else {
    if (!event.data.entity_id) {
      errors.push('Missing required field: data.entity_id');
    }

    if (event.data.payload === undefined) {
      warnings.push('Missing data.payload (optional but recommended)');
    }

    if (!event.data.timestamp) {
      warnings.push('Missing data.timestamp (recommended for ordering)');
    } else {
      const ts = new Date(event.data.timestamp);
      if (isNaN(ts.getTime())) {
        errors.push(`Invalid timestamp "${event.data.timestamp}". Use ISO 8601 format.`);
      }
    }
  }

  // Optional fields validation
  if (event.trace_id && typeof event.trace_id !== 'string') {
    errors.push('trace_id must be a string');
  }

  if (event.version !== undefined && (!Number.isInteger(event.version) || event.version < 1)) {
    errors.push('version must be a positive integer');
  }

  return { valid: errors.length === 0, errors, warnings };
}

function formatResult(result, event) {
  const lines = [];

  if (result.valid) {
    lines.push('✓ Event is valid');
  } else {
    lines.push('✗ Event is invalid');
  }

  if (result.errors.length > 0) {
    lines.push('\nErrors:');
    result.errors.forEach(e => lines.push(`  - ${e}`));
  }

  if (result.warnings.length > 0) {
    lines.push('\nWarnings:');
    result.warnings.forEach(w => lines.push(`  - ${w}`));
  }

  if (result.valid) {
    lines.push('\nParsed:');
    lines.push(`  Topic: ${event.topic}`);
    lines.push(`  Event: ${event.event}`);
    lines.push(`  Entity: ${event.data?.entity_id}`);
  }

  return lines.join('\n');
}

async function main() {
  let input = process.argv[2];

  // Read from stdin if no argument
  if (!input) {
    const chunks = [];
    for await (const chunk of process.stdin) {
      chunks.push(chunk);
    }
    input = Buffer.concat(chunks).toString().trim();
  }

  if (!input) {
    console.log('Usage: node validate-ws-event.js \'{"topic":"...","event":"...","data":{...}}\'');
    console.log('\nExample valid event:');
    console.log(JSON.stringify({
      topic: 'gift-list:123',
      event: 'UPDATED',
      data: {
        entity_id: 'gift-456',
        payload: { name: 'Updated Gift' },
        user_id: 'user-789',
        timestamp: new Date().toISOString()
      }
    }, null, 2));
    process.exit(1);
  }

  try {
    const event = JSON.parse(input);
    const result = validate(event);
    console.log(formatResult(result, event));
    process.exit(result.valid ? 0 : 1);
  } catch (err) {
    console.error('✗ Invalid JSON:', err.message);
    process.exit(1);
  }
}

main();

```