Khi xây dựng ứng dụng chatbot thời gian thực, vấn đề mất kết nối và tự động reconnect là thách thức lớn nhất mà các developer gặp phải. Bài viết này sẽ hướng dẫn bạn implement Server-Sent Events (SSE) với cơ chế断线重连 (reconnection) hoàn chỉnh, giúp ứng dụng của bạn hoạt động ổn định như sản phẩm thương mại.

Tôi đã implement tính năng này cho 3 dự án thực tế và rút ra được nhiều bài học quý giá. Quan trọng nhất: với HolySheep AI, độ trễ chỉ dưới 50ms và tiết kiệm đến 85%+ chi phí so với API chính thức.

Bảng So Sánh Chi Phí và Hiệu Suất

Tiêu chí HolySheep AI API Chính thức Đối thủ A
Claude 4 Opus $15/MTok $15/MTok $18/MTok
Độ trễ trung bình <50ms ✓ 120-200ms 80-150ms
Thanh toán WeChat/Alipay/Visa Credit Card Credit Card
Tín dụng miễn phí Có ✓ Không Không
Stream SSE support Đầy đủ ✓ Đầy đủ Giới hạn
Phương thức thanh toán ¥1=$1 USD USD
Group phù hợp Startup, Dev, Enterprise Enterprise lớn Medium business

Tại Sao Cần SSE Reconnection?

Khi người dùng chat với AI, kết nối mạng có thể bị ngắt do:

Không có cơ chế reconnect, người dùng phải refresh trang và mất toàn bộ context. Điều này gây trải nghiệm rất tệ.

Implementation Hoàn Chỉnh

1. Client-Side SSE Manager Class

/**
 * HolySheep AI - SSE Stream Manager with Auto-Reconnection
 * Author: HolySheep AI Technical Team
 * Version: 2.0.0
 * 
 * Tính năng:
 * - Auto-reconnect với exponential backoff
 * - Message buffering trong lúc reconnecting
 * - Heartbeat detection
 * - Context preservation
 */

class HolySheepSSEClient {
    constructor(config) {
        this.baseURL = 'https://api.holysheep.ai/v1';
        this.apiKey = config.apiKey || 'YOUR_HOLYSHEEP_API_KEY';
        this.model = config.model || 'claude-opus-4-5';
        
        // Connection state
        this.eventSource = null;
        this.isConnected = false;
        this.reconnectAttempts = 0;
        this.maxReconnectAttempts = 5;
        this.reconnectDelay = 1000; // ms
        
        // Message tracking
        this.lastEventId = null;
        this.pendingMessages = [];
        this.messageBuffer = [];
        
        // Callbacks
        this.onMessage = config.onMessage || (() => {});
        this.onError = config.onError || (() => {});
        this.onConnected = config.onConnected || (() => {});
        this.onDisconnected = config.onDisconnected || (() => {});
        
        // Heartbeat
        this.heartbeatInterval = null;
        this.heartbeatTimeout = 30000; // 30s
        
        // Request cancellation
        this.abortController = null;
    }

    /**
     * Khởi tạo streaming request đến HolySheep AI
     */
    async startStream(messages, conversationId = null) {
        this.abortController = new AbortController();
        
        const requestBody = {
            model: this.model,
            messages: messages,
            stream: true,
            max_tokens: 4096,
            temperature: 0.7
        };

        // Thêm conversation_id để preserve context khi reconnect
        if (conversationId) {
            requestBody.conversation_id = conversationId;
        }

        const url = ${this.baseURL}/chat/completions;
        
        try {
            const response = await fetch(url, {
                method: 'POST',
                headers: {
                    'Content-Type': 'application/json',
                    'Authorization': Bearer ${this.apiKey},
                    'Accept': 'text/event-stream',
                    'Cache-Control': 'no-cache',
                    'Connection': 'keep-alive',
                    'X-Request-ID': this.generateRequestId()
                },
                body: JSON.stringify(requestBody),
                signal: this.abortController.signal
            });

            if (!response.ok) {
                throw new Error(HTTP ${response.status}: ${response.statusText});
            }

            this.setupEventSource(response.body);
            this.startHeartbeat();
            this.onConnected();
            
        } catch (error) {
            this.handleError(error);
        }
    }

    /**
     * Thiết lập EventSource từ fetch response
     */
    setupEventSource(body) {
        const reader = body.getReader();
        const decoder = new TextDecoder();
        let buffer = '';

        const readStream = () => {
            reader.read().then(({ done, value }) => {
                if (done) {
                    this.handleDisconnect();
                    return;
                }

                buffer += decoder.decode(value, { stream: true });
                const lines = buffer.split('\n');
                buffer = lines.pop() || '';

                for (const line of lines) {
                    this.processSSELine(line);
                }

                if (this.isConnected) {
                    readStream();
                }
            }).catch(error => {
                this.handleError(error);
            });
        };

        readStream();
        this.isConnected = true;
    }

    /**
     * Xử lý từng dòng SSE
     */
    processSSELine(line) {
        // Parse: data: {"id":"...","choices":[{"delta":{"content":"..."}}]}
        
        if (!line.startsWith('data: ')) return;
        
        const data = line.slice(6).trim();
        
        if (data === '[DONE]') {
            this.onMessage({ type: 'done', content: '' });
            return;
        }

        try {
            const parsed = JSON.parse(data);
            const content = parsed.choices?.[0]?.delta?.content;
            
            if (content) {
                this.lastEventId = parsed.id;
                this.onMessage({
                    type: 'content',
                    content: content,
                    id: parsed.id,
                    done: false
                });
            }
            
            // Xử lý error từ server
            if (parsed.error) {
                this.onError({
                    type: 'server_error',
                    message: parsed.error.message,
                    code: parsed.error.code
                });
            }
            
        } catch (parseError) {
            console.warn('SSE parse error:', parseError);
        }
    }

    /**
     * Bắt đầu heartbeat để detect connection death
     */
    startHeartbeat() {
        this.stopHeartbeat();
        
        this.heartbeatInterval = setInterval(() => {
            if (!this.isConnected) return;
            
            // Gửi ping để check connection
            this.sendPing();
        }, this.heartbeatTimeout);
    }

    /**
     * Dừng heartbeat
     */
    stopHeartbeat() {
        if (this.heartbeatInterval) {
            clearInterval(this.heartbeatInterval);
            this.heartbeatInterval = null;
        }
    }

    /**
     * Gửi ping đến server
     */
    async sendPing() {
        try {
            const response = await fetch(${this.baseURL}/ping, {
                method: 'GET',
                headers: {
                    'Authorization': Bearer ${this.apiKey}
                }
            });
            
            if (!response.ok) {
                this.triggerReconnect();
            }
        } catch (error) {
            this.triggerReconnect();
        }
    }

    /**
     * Xử lý disconnect - tự động reconnect
     */
    handleDisconnect() {
        if (!this.isConnected) return;
        
        this.isConnected = false;
        this.stopHeartbeat();
        this.onDisconnected();
        
        if (this.reconnectAttempts < this.maxReconnectAttempts) {
            this.scheduleReconnect();
        }
    }

    /**
     * Schedule reconnect với exponential backoff
     */
    scheduleReconnect() {
        const delay = this.reconnectDelay * Math.pow(2, this.reconnectAttempts);
        const jitter = Math.random() * 1000; // Add randomness
        
        console.log(Scheduling reconnect in ${delay + jitter}ms (attempt ${this.reconnectAttempts + 1}));
        
        setTimeout(() => {
            this.reconnectAttempts++;
            this.attemptReconnect();
        }, delay + jitter);
    }

    /**
     * Thử reconnect
     */
    async attemptReconnect() {
        try {
            console.log(Reconnect attempt ${this.reconnectAttempts}/${this.maxReconnectAttempts});
            
            // Restore pending messages from buffer
            const pendingMessages = [...this.messageBuffer];
            
            await this.startStream(pendingMessages, this.lastEventId);
            
            // Reset state on successful reconnect
            this.reconnectAttempts = 0;
            this.messageBuffer = [];
            
        } catch (error) {
            console.error('Reconnect failed:', error);
            this.handleError(error);
            
            if (this.reconnectAttempts < this.maxReconnectAttempts) {
                this.scheduleReconnect();
            } else {
                this.onError({
                    type: 'max_reconnect_exceeded',
                    message: 'Unable to reconnect after maximum attempts'
                });
            }
        }
    }

    /**
     * Trigger manual reconnect
     */
    triggerReconnect() {
        if (!this.isConnected) return;
        
        console.log('Manual reconnect triggered');
        this.isConnected = false;
        this.stopHeartbeat();
        this.scheduleReconnect();
    }

    /**
     * Xử lý lỗi tổng quát
     */
    handleError(error) {
        if (error.name === 'AbortError') {
            console.log('Request was aborted');
            return;
        }
        
        this.onError({
            type: 'connection_error',
            message: error.message,
            recoverable: this.reconnectAttempts < this.maxReconnectAttempts
        });
        
        this.handleDisconnect();
    }

    /**
     * Dừng stream
     */
    stop() {
        this.isConnected = false;
        this.stopHeartbeat();
        
        if (this.abortController) {
            this.abortController.abort();
        }
        
        if (this.eventSource) {
            this.eventSource.close();
            this.eventSource = null;
        }
        
        console.log('Stream stopped');
    }

    /**
     * Tạm dừng stream (không disconnect)
     */
    pause() {
        this.stopHeartbeat();
    }

    /**
     * Tiếp tục stream
     */
    resume() {
        this.startHeartbeat();
    }

    /**
     * Utility: Generate unique request ID
     */
    generateRequestId() {
        return hs_${Date.now()}_${Math.random().toString(36).substr(2, 9)};
    }
}

// Export cho module system
if (typeof module !== 'undefined' && module.exports) {
    module.exports = HolySheepSSEClient;
}

2. React Hook với Auto-Reconnection

/**
 * HolySheep AI - React Hook for SSE Streaming
 * Hỗ trợ auto-reconnection, message history, và error recovery
 */

import { useState, useCallback, useRef, useEffect } from 'react';

export function useHolySheepStream(apiKey) {
    const [messages, setMessages] = useState([]);
    const [isStreaming, setIsStreaming] = useState(false);
    const [error, setError] = useState(null);
    const [connectionStatus, setConnectionStatus] = useState('disconnected');
    
    const clientRef = useRef(null);
    const conversationIdRef = useRef(null);

    /**
     * Khởi tạo client với config
     */
    const initClient = useCallback(() => {
        if (clientRef.current) {
            clientRef.current.stop();
        }

        const client = new HolySheepSSEClient({
            apiKey: apiKey,
            model: 'claude-opus-4-5',
            
            onMessage: (data) => {
                if (data.type === 'done') {
                    setIsStreaming(false);
                    setConnectionStatus('done');
                } else if (data.type === 'content') {
                    setMessages(prev => {
                        const lastIndex = prev.length - 1;
                        const lastMessage = prev[lastIndex];
                        
                        if (lastMessage && lastMessage.role === 'assistant') {
                            // Append to existing assistant message
                            const updated = [...prev];
                            updated[lastIndex] = {
                                ...lastMessage,
                                content: lastMessage.content + data.content,
                                id: data.id
                            };
                            return updated;
                        } else {
                            // Create new assistant message
                            return [...prev, {
                                id: data.id || crypto.randomUUID(),
                                role: 'assistant',
                                content: data.content
                            }];
                        }
                    });
                }
            },
            
            onError: (error) => {
                console.error('Stream error:', error);
                setError(error);
                setConnectionStatus('error');
                
                if (!error.recoverable) {
                    setIsStreaming(false);
                }
            },
            
            onConnected: () => {
                setConnectionStatus('connected');
                setError(null);
            },
            
            onDisconnected: () => {
                setConnectionStatus('disconnected');
                setIsStreaming(false);
            }
        });

        clientRef.current = client;
        return client;
    }, [apiKey]);

    /**
     * Gửi message và nhận stream response
     */
    const sendMessage = useCallback(async (userMessage) => {
        // Add user message to history
        const userMsg = {
            id: crypto.randomUUID(),
            role: 'user',
            content: userMessage
        };
        
        setMessages(prev => [...prev, userMsg]);
        setError(null);
        setIsStreaming(true);
        setConnectionStatus('connecting');

        // Prepare messages for API (chỉ gửi last 10 messages để tiết kiệm token)
        const apiMessages = [
            ...messages,
            userMsg
        ].slice(-10);

        const client = initClient();
        
        try {
            await client.startStream(apiMessages, conversationIdRef.current);
        } catch (err) {
            setError({ message: err.message, type: 'init_error' });
            setIsStreaming(false);
            setConnectionStatus('error');
        }
    }, [messages, initClient]);

    /**
     * Xóa tất cả messages
     */
    const clearMessages = useCallback(() => {
        setMessages([]);
        setError(null);
        conversationIdRef.current = null;
    }, []);

    /**
     * Retry last request
     */
    const retryLast = useCallback(async () => {
        if (messages.length === 0) return;
        
        // Remove last assistant response if exists
        const msgsWithoutLastAssistant = [...messages];
        if (msgsWithoutLastAssistant.length > 0 && 
            msgsWithoutLastAssistant[msgsWithoutLastAssistant.length - 1].role === 'assistant') {
            msgsWithoutLastAssistant.pop();
        }
        
        // Find last user message
        const lastUserIndex = msgsWithoutLastAssistant.findLastIndex(m => m.role === 'user');
        if (lastUserIndex !== -1) {
            const lastUserMessage = msgsWithoutLastAssistant[lastUserIndex];
            // Remove everything after last user message
            const msgsToRetry = msgsWithoutLastAssistant.slice(0, lastUserIndex + 1);
            setMessages(msgsToRetry);
            await sendMessage(lastUserMessage.content);
        }
    }, [messages, sendMessage]);

    /**
     * Dừng stream hiện tại
     */
    const stopStream = useCallback(() => {
        if (clientRef.current) {
            clientRef.current.stop();
            setIsStreaming(false);
            setConnectionStatus('stopped');
        }
    }, []);

    /**
     * Manual reconnect
     */
    const reconnect = useCallback(() => {
        if (clientRef.current) {
            clientRef.current.triggerReconnect();
            setConnectionStatus('reconnecting');
        }
    }, []);

    // Cleanup on unmount
    useEffect(() => {
        return () => {
            if (clientRef.current) {
                clientRef.current.stop();
            }
        };
    }, []);

    return {
        messages,
        isStreaming,
        error,
        connectionStatus,
        sendMessage,
        clearMessages,
        retryLast,
        stopStream,
        reconnect
    };
}

// Component demo sử dụng hook
export function ChatInterface() {
    const { 
        messages, 
        isStreaming, 
        error, 
        connectionStatus,
        sendMessage, 
        clearMessages,
        retryLast 
    } = useHolySheepStream('YOUR_HOLYSHEEP_API_KEY');

    const [input, setInput] = useState('');
    const messagesEndRef = useRef(null);

    const handleSubmit = async (e) => {
        e.preventDefault();
        if (!input.trim() || isStreaming) return;
        
        await sendMessage(input);
        setInput('');
    };

    // Auto scroll to bottom
    useEffect(() => {
        messagesEndRef.current?.scrollIntoView({ behavior: 'smooth' });
    }, [messages]);

    const statusColor = {
        'connected': 'green',
        'connecting': 'yellow',
        'disconnected': 'gray',
        'reconnecting': 'orange',
        'error': 'red',
        'done': 'blue',
        'stopped': 'gray'
    };

    return (
        <div className="chat-container">
            <div className="status-bar">
                Status: <span style={{ color: statusColor[connectionStatus] }}>
                    {connectionStatus}
                </span>
            </div>
            
            <div className="messages">
                {messages.map(msg => (
                    <div key={msg.id} className={message ${msg.role}}>
                        <strong>{msg.role}: </strong>
                        {msg.content}
                    </div>
                ))}
                <div ref={messagesEndRef} />
            </div>
            
            {error && (
                <div className="error-banner">
                    Error: {error.message}
                    <button onClick={retryLast}>Retry</button>
                </div>
            )}
            
            <form onSubmit={handleSubmit}>
                <input
                    value={input}
                    onChange={e => setInput(e.target.value)}
                    placeholder="Type your message..."
                    disabled={isStreaming}
                />
                <button type="submit" disabled={isStreaming || !input.trim()}>
                    {isStreaming ? 'Streaming...' : 'Send'}
                </button>
                <button type="button" onClick={clearMessages}>
                    Clear
                </button>
            </form>
        </div>
    );
}

3. Backend Proxy với Rate Limiting và Health Check

/**
 * HolySheep AI - Backend Proxy Server
 * Xử lý rate limiting, caching, và failover
 * Node.js + Express implementation
 */

const express = require('express');
const cors = require('cors');
const { RateLimiterMemory } = require('rate-limiter-flexible');
const NodeCache = require('node-cache');

const app = express();
app.use(cors());
app.use(express.json());

// Configuration
const HOLYSHEEP_CONFIG = {
    baseURL: 'https://api.holysheep.ai/v1',
    apiKey: process.env.HOLYSHEEP_API_KEY,
    timeout: 60000, // 60s
    maxRetries: 3
};

// In-memory cache cho responses (5 phút)
const responseCache = new NodeCache({ stdTTL: 300 });

// Rate limiter: 100 requests/phút cho mỗi IP
const rateLimiter = new RateLimiterMemory({
    points: 100,
    duration: 60,
    blockDuration: 120
});

/**
 * Middleware: Rate limiting
 */
const rateLimitMiddleware = async (req, res, next) => {
    try {
        await rateLimiter.consume(req.ip);
        next();
    } catch (error) {
        res.status(429).json({
            error: 'Too many requests',
            retryAfter: Math.ceil(error.msBeforeNext / 1000)
        });
    }
};

/**
 * Middleware: Auth validation
 */
const authMiddleware = (req, res, next) => {
    const apiKey = req.headers.authorization?.replace('Bearer ', '');
    
    if (!apiKey) {
        return res.status(401).json({ error: 'API key required' });
    }
    
    // Validate key format (HolySheep format: hs_xxx)
    if (!apiKey.startsWith('hs_') && !apiKey.startsWith('sk-')) {
        return res.status(401).json({ error: 'Invalid API key format' });
    }
    
    req.apiKey = apiKey;
    next();
};

/**
 * Health check endpoint
 */
app.get('/health', async (req, res) => {
    try {
        const startTime = Date.now();
        
        const response = await fetch(${HOLYSHEEP_CONFIG.baseURL}/models, {
            headers: {
                'Authorization': Bearer ${HOLYSHEEP_CONFIG.apiKey}
            }
        });
        
        const latency = Date.now() - startTime;
        
        res.json({
            status: response.ok ? 'healthy' : 'degraded',
            holySheepStatus: response.ok ? 'up' : 'down',
            latency: ${latency}ms,
            timestamp: new Date().toISOString()
        });
    } catch (error) {
        res.status(503).json({
            status: 'unhealthy',
            error: error.message
        });
    }
});

/**
 * Stream endpoint - Proxy request đến HolySheep AI
 */
app.post('/v1/chat/stream', authMiddleware, rateLimitMiddleware, async (req, res) => {
    const { messages, model, conversation_id, ...options } = req.body;
    
    // Validate messages
    if (!messages || !Array.isArray(messages) || messages.length === 0) {
        return res.status(400).json({ error: 'messages array required' });
    }
    
    // Generate cache key (chỉ cache cho non-streaming)
    const cacheKey = stream_${Buffer.from(JSON.stringify({ messages, model })).toString('base64')};
    
    // Set headers cho SSE
    res.writeHead(200, {
        'Content-Type': 'text/event-stream',
        'Cache-Control': 'no-cache',
        'Connection': 'keep-alive',
        'X-Accel-Buffering': 'no'
    });
    
    let retryCount = 0;
    const maxRetries = HOLYSHEEP_CONFIG.maxRetries;
    
    const streamFromHolySheep = async () => {
        try {
            const response = await fetch(${HOLYSHEEP_CONFIG.baseURL}/chat/completions, {
                method: 'POST',
                headers: {
                    'Content-Type': 'application/json',
                    'Authorization': Bearer ${HOLYSHEEP_CONFIG.apiKey},
                    'Accept': 'text/event-stream',
                    'X-Forwarded-For': req.ip,
                    'X-Conversation-ID': conversation_id || ''
                },
                body: JSON.stringify({
                    model: model || 'claude-opus-4-5',
                    messages: messages,
                    stream: true,
                    ...options
                })
            });
            
            if (!response.ok) {
                const error = await response.text();
                res.write(data: ${JSON.stringify({ error: { message: error } })}\n\n);
                res.end();
                return;
            }
            
            // Pipe response
            response.body.pipe(res);
            
            response.body.on('error', () => {
                console.log('Stream error, attempting reconnect...');
                if (retryCount < maxRetries) {
                    retryCount++;
                    setTimeout(streamFromHolySheep, 1000 * retryCount);
                }
            });
            
        } catch (error) {
            console.error('HolySheep API error:', error);
            
            if (retryCount < maxRetries) {
                retryCount++;
                console.log(Retrying (${retryCount}/${maxRetries})...);
                setTimeout(streamFromHolySheep, 1000 * retryCount);
            } else {
                res.write(`data: ${JSON.stringify({ 
                    error: { 
                        message: 'Service temporarily unavailable',
                        retryable: true 
                    } 
                })}\n\n`);
                res.end();
            }
        }
    };
    
    streamFromHolySheep();
    
    // Cleanup on client disconnect
    req.on('close', () => {
        console.log('Client disconnected');
    });
});

/**
 * Non-streaming chat endpoint (cached)
 */
app.post('/v1/chat/completions', authMiddleware, rateLimitMiddleware, async (req, res) => {
    const { messages, model } = req.body;
    const cacheKey = chat_${Buffer.from(JSON.stringify({ messages, model })).toString('base64')};
    
    // Check cache
    const cached = responseCache.get(cacheKey);
    if (cached) {
        return res.json(cached);
    }
    
    try {
        const response = await fetch(${HOLYSHEEP_CONFIG.baseURL}/chat/completions, {
            method: 'POST',
            headers: {
                'Content-Type': 'application/json',
                'Authorization': Bearer ${HOLYSHEEP_CONFIG.apiKey}
            },
            body: JSON.stringify({
                model: model || 'claude-opus-4-5',
                messages: messages,
                stream: false
            })
        });
        
        if (!response.ok) {
            const error = await response.json();
            return res.status(response.status).json(error);
        }
        
        const data = await response.json();
        
        // Cache successful response
        responseCache.set(cacheKey, data);
        
        res.json(data);
        
    } catch (error) {
        console.error('HolySheep API error:', error);
        res.status(500).json({ error: error.message });
    }
});

/**
 * Get available models
 */
app.get('/v1/models', authMiddleware, async (req, res) => {
    try {
        const response = await fetch(${HOLYSHEEP_CONFIG.baseURL}/models, {
            headers: {
                'Authorization': Bearer ${HOLYSHEEP_CONFIG.apiKey}
            }
        });
        
        const data = await response.json();
        res.json(data);
        
    } catch (error) {
        res.status(500).json({ error: error.message });
    }
});

/**
 * Get API usage
 */
app.get('/v1/usage', authMiddleware, async (req, res) => {
    try {
        const response = await fetch(${HOLYSHEEP_CONFIG.baseURL}/usage, {
            headers: {
                'Authorization': Bearer ${HOLYSHEEP_CONFIG.apiKey}
            }
        });
        
        const data = await response.json();
        res.json(data);
        
    } catch (error) {
        res.status(500).json({ error: error.message });
    }
});

const PORT = process.env.PORT || 3000;
app.listen(PORT, () => {
    console.log(HolySheep Proxy Server running on port ${PORT});
    console.log(Health check: http://localhost:${PORT}/health);
});

Lỗi Thường Gặp và Cách Khắc Phục

Lỗi 1: CORS Policy Error

Mô tả: Trình duyệt chặn request với lỗi "Access-Control-Allow-Origin missing"

// ❌ SAI: Gọi trực tiếp từ browser
const response = await fetch('https://api.holysheep.ai/v1/chat/completions', {
    method: 'POST',
    headers: { 'Authorization': 'Bearer YOUR_KEY' }
});

// ✅ ĐÚNG: Sử dụng backend proxy
const response = await fetch('/api/v1/chat/stream', {
    method: 'POST',
    headers: { 
        'Content-Type': 'application/json',
        'Authorization': 'Bearer YOUR_KEY' // Key được verify ở backend
    },
    body: JSON.stringify({ messages })
});

// Backend proxy (Express) - thêm CORS headers
app.use((req, res, next) => {
    res.header('Access-Control-Allow-Origin', 'https://your-domain.com');
    res.header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS');
    res.header('Access-Control-Allow-Headers', 'Content-Type, Authorization');
    next();
});

Lỗi 2: Stream bị cắt giữa chừng

Mô tả: Response bị dừng đột ngột, thiếu phần cuối hoặc bị truncate

// ❌ SAI: Không xử lý buffer đúng cách
process.stdout.write(data); // Có thể mất data nếu network chậm

// ✅ ĐÚNG: Buffer cho đến khi có complete line
let buffer = '';
response.body.on('data', (chunk) => {
    buffer += chunk.toString();
    
    // Xử lý từng dòng hoàn chỉnh
    let newlineIndex;
    while ((newlineIndex = buffer.indexOf('\n')) !== -1) {
        const line = buffer.slice(0, newlineIndex);
        buffer = buffer.slice(newlineIndex + 1);
        
        if (line.startsWith('data: ')) {
            const data = line.slice(6);
            if (data === '[DONE]') {
                console.log('Stream completed');
            } else {
                process.stdout.write(data);
            }
        }
    }
});

// Hoặc dùng readline module
import * as readline from 'readline';

const rl = readline.createInterface({
    input: response.body,
    crlfDelay: Infinity
});

for await (const line of rl) {
    if (line.startsWith('data: ')) {
        const data = line.slice(6);
        if (data !== '[DONE]') {
            console.log('Received:', data);
        }
    }
}

Lỗi 3: Reconnection loop không dừng

Mô tả: Client liên tục reconnect mà không có kết quả, gây spam server

// ❌ SAI: Không có giới hạn reconnect attempts
while (!connected) {
    try {
        await connect();
        connected = true;
    } catch (e) {
        await sleep(1000); // Vòng lặp vô hạn!
    }
}

// ✅ ĐÚNG: Exponential backoff với jitter và max attempts
class ReconnectionManager {
    constructor() {
        this.attempts = 0;
        this.maxAttempts = 5;
        this.baseDelay = 1000; // 1 second
        this.maxDelay = 30000; // 30 seconds
    }
    
    async reconnect(connectFn) {
        while (this.attempts < this.maxAttempts) {
            try {
                await connectFn();
                this.attempts = 0; // Reset on success
                return true;
            } catch (error) {
                this.attempts++;
                
                if (this.attempts >= this.maxAttempts) {
                    console.error('Max reconnect attempts reached');
                    this.notifyUserMaxRetriesExceeded();
                    return false;
                }
                
                // Exponential backoff: 1s, 2s, 4s, 8s, 16s
                const delay = Math.min(
                    this.baseDelay * Math.pow(2, this.attempts - 1),
                    this.maxDelay
                );
                
                // Thêm jitter (±25%) để tránh thundering herd
                const jitter = delay * 0.25 * (Math.random() - 0.5);
                const actualDelay = delay + jitter;
                
                console.log(Reconnecting in ${Math.round(actualDelay)}ms...);
                await this.sleep

Tài nguyên liên quan
Bài viết liên quan