1import { WebSocketServer } from 'ws';
2import { createServer } from 'http';
3import jwt from 'jsonwebtoken';
4
5const server = createServer();
6const wss = new WebSocketServer({ 
7  server,
8  perMessageDeflate: {
9    zlibDeflateOptions: {
10      chunkSize: 1024,
11      memLevel: 7,
12      level: 3
13    },
14    zlibInflateOptions: {
15      chunkSize: 10 * 1024
16    },
17    threshold: 1024 // Only compress messages > 1KB
18  }
19});
20
21// Connection handling with authentication
22wss.on('connection', async (ws, req) => {
23  // Extract token from query params or headers
24  const token = new URLSearchParams(req.url?.split('?')[1]).get('token');
25  
26  if (!token) {
27    ws.close(4001, 'Missing authentication token');
28    return;
29  }
30  
31  try {
32    // Verify JWT token (Clerk, Auth0, etc.)
33    const user = jwt.verify(token, process.env.JWT_SECRET!) as { userId: string };
34    (ws as any).userId = user.userId;
35    
36    console.log(`✅ User ${user.userId} connected`);
37    
38    // Send welcome message
39    ws.send(JSON.stringify({ type: 'connected', userId: user.userId }));
40    
41    // Handle incoming messages
42    ws.on('message', (data) => handleMessage(ws, data));
43    
44    // Handle disconnection
45    ws.on('close', () => {
46      console.log(`❌ User ${user.userId} disconnected`);
47    });
48    
49    // Handle errors
50    ws.on('error', (error) => {
51      console.error('WebSocket error:', error);
52    });
53    
54  } catch (error) {
55    ws.close(4002, 'Invalid authentication token');
56  }
57});
58
59server.listen(3001, () => {
60  console.log('WebSocket server running on ws://localhost:3001');
61});

1import { Redis } from '@upstash/redis';
2import { WebSocket } from 'ws';
3
4const redis = new Redis({
5  url: process.env.UPSTASH_REDIS_REST_URL!,
6  token: process.env.UPSTASH_REDIS_REST_TOKEN!,
7});
8
9// Subscribe to Redis channels for horizontal scaling
10const subscriber = redis.duplicate();
11const publisher = redis.duplicate();
12
13interface ConnectedClient {
14  ws: WebSocket;
15  userId: string;
16  channels: Set<string>;
17}
18
19const clients = new Map<string, ConnectedClient>();
20
21// Subscribe to Redis Pub/Sub
22subscriber.subscribe('notifications', 'messages', (message) => {
23  const data = JSON.parse(message);
24  
25  // Broadcast to all connected clients subscribed to this channel
26  clients.forEach((client) => {
27    if (client.channels.has(data.channel)) {
28      client.ws.send(JSON.stringify(data));
29    }
30  });
31});
32
33// Publish message to Redis (works across multiple servers)
34export async function broadcastToChannel(channel: string, message: any) {
35  await publisher.publish(channel, JSON.stringify({ channel, ...message }));
36}
37
38// Example: notify all users about a new post
39export async function notifyNewPost(postId: string, title: string) {
40  await broadcastToChannel('notifications', {
41    type: 'new_post',
42    postId,
43    title,
44    timestamp: Date.now(),
45  });
46}
47
48// Client subscribes to specific channels
49export function subscribeClient(userId: string, channel: string) {
50  const client = clients.get(userId);
51  if (client) {
52    client.channels.add(channel);
53  }
54}

1import { useEffect, useRef, useState } from 'react';
2
3interface UseWebSocketOptions {
4  url: string;
5  token: string;
6  onMessage?: (data: any) => void;
7  reconnectInterval?: number;
8  maxReconnectAttempts?: number;
9}
10
11export function useWebSocket({
12  url,
13  token,
14  onMessage,
15  reconnectInterval = 3000,
16  maxReconnectAttempts = 5,
17}: UseWebSocketOptions) {
18  const [isConnected, setIsConnected] = useState(false);
19  const [reconnectAttempt, setReconnectAttempt] = useState(0);
20  const wsRef = useRef<WebSocket | null>(null);
21  const reconnectTimeoutRef = useRef<NodeJS.Timeout>();
22  
23  const connect = () => {
24    const ws = new WebSocket(`${url}?token=${token}`);
25    
26    ws.onopen = () => {
27      console.log('✅ WebSocket connected');
28      setIsConnected(true);
29      setReconnectAttempt(0);
30      
31      // Send heartbeat every 30s to keep connection alive
32      const heartbeat = setInterval(() => {
33        if (ws.readyState === WebSocket.OPEN) {
34          ws.send(JSON.stringify({ type: 'ping' }));
35        }
36      }, 30000);
37      
38      ws.onclose = () => {
39        clearInterval(heartbeat);
40      };
41    };
42    
43    ws.onmessage = (event) => {
44      try {
45        const data = JSON.parse(event.data);
46        if (data.type !== 'pong') { // Ignore pong responses
47          onMessage?.(data);
48        }
49      } catch (error) {
50        console.error('Failed to parse message:', error);
51      }
52    };
53    
54    ws.onerror = (error) => {
55      console.error('WebSocket error:', error);
56    };
57    
58    ws.onclose = () => {
59      console.log('❌ WebSocket disconnected');
60      setIsConnected(false);
61      wsRef.current = null;
62      
63      // Attempt reconnection with exponential backoff
64      if (reconnectAttempt < maxReconnectAttempts) {
65        const delay = reconnectInterval * Math.pow(2, reconnectAttempt);
66        console.log(`Reconnecting in ${delay}ms... (attempt ${reconnectAttempt + 1}/${maxReconnectAttempts})`);
67        
68        reconnectTimeoutRef.current = setTimeout(() => {
69          setReconnectAttempt(prev => prev + 1);
70          connect();
71        }, delay);
72      } else {
73        console.error('Max reconnection attempts reached');
74      }
75    };
76    
77    wsRef.current = ws;
78  };
79  
80  useEffect(() => {
81    connect();
82    
83    return () => {
84      clearTimeout(reconnectTimeoutRef.current);
85      wsRef.current?.close();
86    };
87  }, [url, token]);
88  
89  const send = (data: any) => {
90    if (wsRef.current?.readyState === WebSocket.OPEN) {
91      wsRef.current.send(JSON.stringify(data));
92    } else {
93      console.warn('WebSocket is not connected');
94    }
95  };
96  
97  return { isConnected, send };
98}

1import { Redis } from '@upstash/redis';
2import { WebSocket } from 'ws';
3
4const redis = new Redis({
5  url: process.env.UPSTASH_REDIS_REST_URL!,
6  token: process.env.UPSTASH_REDIS_REST_TOKEN!,
7});
8
9interface RateLimitConfig {
10  maxMessages: number;  // Max messages per window
11  windowMs: number;     // Time window in milliseconds
12}
13
14const DEFAULT_RATE_LIMIT: RateLimitConfig = {
15  maxMessages: 10,
16  windowMs: 1000, // 10 messages per second
17};
18
19export async function checkRateLimit(
20  userId: string,
21  config: RateLimitConfig = DEFAULT_RATE_LIMIT
22): Promise<boolean> {
23  const key = `ratelimit:ws:${userId}`;
24  const now = Date.now();
25  const windowStart = now - config.windowMs;
26  
27  // Remove old entries
28  await redis.zremrangebyscore(key, 0, windowStart);
29  
30  // Count messages in current window
31  const count = await redis.zcard(key);
32  
33  if (count >= config.maxMessages) {
34    return false; // Rate limit exceeded
35  }
36  
37  // Add current message
38  await redis.zadd(key, { score: now, member: `${now}-${Math.random()}` });
39  await redis.expire(key, Math.ceil(config.windowMs / 1000));
40  
41  return true; // Allow message
42}
43
44// Usage in WebSocket handler
45export async function handleMessage(ws: WebSocket, data: Buffer) {
46  const userId = (ws as any).userId;
47  
48  // Check rate limit
49  const allowed = await checkRateLimit(userId);
50  
51  if (!allowed) {
52    ws.send(JSON.stringify({
53      type: 'error',
54      message: 'Rate limit exceeded. Please slow down.',
55    }));
56    return;
57  }
58  
59  // Process message...
60  const message = JSON.parse(data.toString());
61  console.log(`Message from ${userId}:`, message);
62}

1import { WebSocket } from 'ws';
2
3interface Message {
4  type: string;
5  data: any;
6  priority?: number;
7}
8
9class MessageQueue {
10  private queue: Message[] = [];
11  private processing = false;
12  
13  async add(ws: WebSocket, message: Message) {
14    this.queue.push(message);
15    this.queue.sort((a, b) => (b.priority || 0) - (a.priority || 0));
16    
17    if (!this.processing) {
18      this.process(ws);
19    }
20  }
21  
22  private async process(ws: WebSocket) {
23    this.processing = true;
24    
25    while (this.queue.length > 0) {
26      const message = this.queue.shift();
27      if (!message) break;
28      
29      // Check if socket is ready (bufferedAmount indicates backpressure)
30      if (ws.bufferedAmount > 1024 * 1024) { // 1MB buffer
31        console.warn('Backpressure detected, pausing...');
32        
33        // Wait for buffer to drain
34        await new Promise(resolve => {
35          const checkBuffer = setInterval(() => {
36            if (ws.bufferedAmount < 512 * 1024) { // Resume at 512KB
37              clearInterval(checkBuffer);
38              resolve(null);
39            }
40          }, 100);
41        });
42      }
43      
44      // Send message
45      if (ws.readyState === WebSocket.OPEN) {
46        ws.send(JSON.stringify(message));
47      }
48    }
49    
50    this.processing = false;
51  }
52}
53
54const queues = new Map<string, MessageQueue>();
55
56export function sendWithBackpressure(
57  ws: WebSocket,
58  message: Message,
59  userId: string
60) {
61  if (!queues.has(userId)) {
62    queues.set(userId, new MessageQueue());
63  }
64  
65  const queue = queues.get(userId)!;
66  queue.add(ws, message);
67}
68
69// Example usage: send bulk notifications
70export async function sendBulkNotifications(
71  userId: string,
72  ws: WebSocket,
73  notifications: any[]
74) {
75  for (const notification of notifications) {
76    sendWithBackpressure(ws, {
77      type: 'notification',
78      data: notification,
79      priority: notification.urgent ? 10 : 1,
80    }, userId);
81  }
82}