Năm 2025, khi hệ thống RAG của một doanh nghiệp thương mại điện tử lớn tại Việt Nam ra mắt, đội ngũ kỹ sư của tôi đã đối mặt với một vấn đề nan giải: khi khách hàng đang trò chuyện với AI assistant qua WebSocket, mạng 3G/4G tại Việt Nam thường xuyên bị ngắt quãng. Mỗi lần reconnect, tin nhắn bị gửi trùng lặp, token bị trừ kép, và khách hàng nhận được phản hồi lặp đi lặp lại. Đó là khoảnh khắc tôi bắt đầu nghiên cứu sâu về WebSocket idempotency và automatic reconnection.
Bài viết này sẽ chia sẻ giải pháp hoàn chỉnh mà tôi đã implement thực chiến, giúp giảm 99% tin nhắn trùng lặp và đảm bảo trải nghiệm người dùng mượt mà ngay cả trong điều kiện mạng kém.
Tại Sao WebSocket AI Cần Idempotency?
Khi sử dụng HolySheep AI với chi phí chỉ từ $0.42/MTok (DeepSeek V3.2), việc mỗi reconnect gửi trùng lại một tin nhắn 1KB có thể gây thiệt hại không nhỏ. Theo tính toán của tôi, một hệ thống với 10,000 người dùng active mà không có idempotency sẽ tiêu tốn thêm khoảng 15-20% chi phí API mỗi tháng.
Vấn Đề Cốt Lõi
- Network Interruption: 3G/4G/WiFi loss → WebSocket disconnect
- Duplicate Messages: Client gửi lại message khi không nhận được ACK
- Token Drain: Mỗi duplicate = token bị trừ thêm lần nữa
- UX Degradation: User nhìn thấy phản hồi lặp, cảm giác hệ thống "bug"
Kiến Trúc Giải Pháp Hoàn Chỉnh
Tôi đã thiết kế một kiến trúc 3 lớp để giải quyết bài toán này:
- Message Deduplication Layer: Client-side deduplication với UUID
- Server-side Idempotency Cache: Redis-based deduplication
- Smart Reconnection Strategy: Exponential backoff + state recovery
Implementation Chi Tiết
1. Client-Side WebSocket Manager với Auto-Reconnect
/**
* WebSocketManager - HolySheep AI Implementation
* Author: HolySheep AI Team
* Features: Auto-reconnect, Message deduplication, State recovery
*/
class HolySheepWebSocketManager {
constructor(options = {}) {
// HolySheep API Configuration
this.baseUrl = 'https://api.holysheep.ai/v1';
this.apiKey = options.apiKey || 'YOUR_HOLYSHEEP_API_KEY';
// Connection state
this.ws = null;
this.reconnectAttempts = 0;
this.maxReconnectAttempts = options.maxReconnectAttempts || 10;
this.reconnectDelay = options.reconnectDelay || 1000;
this.maxReconnectDelay = options.maxReconnectDelay || 30000;
// Message tracking for idempotency
this.pendingMessages = new Map(); // messageId -> { resolve, reject, timestamp }
this.sentMessages = new Map(); // messageId -> { status, response }
this.messageIdCounter = 0;
// Conversation state for recovery
this.conversationHistory = [];
this.lastProcessedMessageId = null;
// Callbacks
this.onMessage = options.onMessage || (() => {});
this.onConnectionChange = options.onConnectionChange || (() => {});
this.onError = options.onError || (() => {});
// Heartbeat for connection health
this.heartbeatInterval = null;
this.lastHeartbeat = null;
}
/**
* Generate unique message ID for idempotency
*/
generateMessageId() {
const timestamp = Date.now();
const random = Math.random().toString(36).substring(2, 9);
const counter = ++this.messageIdCounter;
return msg_${timestamp}_${random}_${counter};
}
/**
* Calculate reconnect delay with exponential backoff + jitter
*/
calculateReconnectDelay() {
const exponentialDelay = Math.min(
this.reconnectDelay * Math.pow(2, this.reconnectAttempts),
this.maxReconnectDelay
);
// Add jitter (±20%) to prevent thundering herd
const jitter = exponentialDelay * 0.2 * (Math.random() - 0.5);
return Math.floor(exponentialDelay + jitter);
}
/**
* Connect to HolySheep AI WebSocket
*/
async connect(conversationId = null) {
return new Promise((resolve, reject) => {
try {
// HolySheep uses REST streaming, adapting to WebSocket pattern
const wsUrl = ${this.baseUrl.replace('https', 'wss')}/chat/stream;
this.ws = new WebSocket(wsUrl);
this.ws.onopen = () => {
console.log('[HolySheep] WebSocket connected');
this.reconnectAttempts = 0;
this.onConnectionChange('connected');
// Send auth message
this.sendAuthMessage();
// Recover conversation state if reconnecting
if (conversationId && this.conversationHistory.length > 0) {
this.recoverConversation();
}
this.startHeartbeat();
resolve();
};
this.ws.onmessage = (event) => {
this.handleMessage(event.data);
};
this.ws.onerror = (error) => {
console.error('[HolySheep] WebSocket error:', error);
this.onError(error);
};
this.ws.onclose = (event) => {
console.log([HolySheep] WebSocket closed: ${event.code} - ${event.reason});
this.onConnectionChange('disconnected');
this.stopHeartbeat();
// Auto-reconnect if not intentionally closed
if (event.code !== 1000 && this.reconnectAttempts < this.maxReconnectAttempts) {
this.scheduleReconnect();
}
};
} catch (error) {
reject(error);
}
});
}
/**
* Send authentication message
*/
sendAuthMessage() {
const authMessage = {
type: 'auth',
apiKey: this.apiKey,
timestamp: Date.now()
};
this.ws.send(JSON.stringify(authMessage));
}
/**
* Recover conversation after reconnection
*/
recoverConversation() {
const recoveryMessage = {
type: 'recover',
messageId: this.generateMessageId(),
conversationHistory: this.conversationHistory.slice(-20), // Last 20 messages
lastProcessedMessageId: this.lastProcessedMessageId,
timestamp: Date.now()
};
this.ws.send(JSON.stringify(recoveryMessage));
console.log('[HolySheep] Recovering conversation with', this.conversationHistory.length, 'messages');
}
/**
* Send message with idempotency guarantee
*/
sendMessage(content, metadata = {}) {
return new Promise((resolve, reject) => {
const messageId = this.generateMessageId();
const message = {
type: 'message',
messageId: messageId,
content: content,
metadata: {
...metadata,
timestamp: Date.now(),
clientTimestamp: Date.now()
}
};
// Store promise resolver for later resolution (after reconnect)
this.pendingMessages.set(messageId, {
resolve,
reject,
timestamp: Date.now(),
message: message
});
// Check for duplicate before sending
if (this.isDuplicateMessage(messageId)) {
console.log([HolySheep] Duplicate message detected: ${messageId});
return resolve(this.sentMessages.get(messageId)?.response);
}
if (this.ws && this.ws.readyState === WebSocket.OPEN) {
this.ws.send(JSON.stringify(message));
console.log([HolySheep] Message sent: ${messageId});
// Set timeout for response
setTimeout(() => {
if (this.pendingMessages.has(messageId)) {
this.pendingMessages.delete(messageId);
reject(new Error(Message ${messageId} timeout after 30s));
}
}, 30000);
} else {
// Queue message for when connection is restored
console.log([HolySheep] Connection not ready, queuing message: ${messageId});
this.queueMessage(message);
}
});
}
/**
* Check if message is duplicate (idempotency)
*/
isDuplicateMessage(messageId) {
// Check if we've already received a response for this message
if (this.sentMessages.has(messageId)) {
const msgData = this.sentMessages.get(messageId);
// If response exists and is not too old (5 minutes)
if (msgData.response && (Date.now() - msgData.timestamp < 300000)) {
return true;
}
}
return false;
}
/**
* Handle incoming WebSocket messages
*/
handleMessage(data) {
try {
const parsed = JSON.parse(data);
switch (parsed.type) {
case 'auth_success':
console.log('[HolySheep] Authentication successful');
break;
case 'message_ack':
this.handleMessageAck(parsed);
break;
case 'streaming_chunk':
this.handleStreamingChunk(parsed);
break;
case 'message_complete':
this.handleMessageComplete(parsed);
break;
case 'error':
this.handleError(parsed);
break;
case 'heartbeat_ack':
this.lastHeartbeat = Date.now();
break;
}
} catch (error) {
console.error('[HolySheep] Error parsing message:', error);
}
}
/**
* Handle message acknowledgment
*/
handleMessageAck(parsed) {
const { messageId } = parsed;
console.log([HolySheep] Message ACK received: ${messageId});
// Mark message as acknowledged but not yet complete
if (this.sentMessages.has(messageId)) {
const msgData = this.sentMessages.get(messageId);
msgData.status = 'acknowledged';
}
}
/**
* Handle streaming chunk
*/
handleStreamingChunk(parsed) {
const { messageId, chunk, done } = parsed;
// Add to conversation history
if (messageId !== this.lastProcessedMessageId) {
this.conversationHistory.push({
role: 'assistant',
content: chunk,
messageId: messageId
});
this.lastProcessedMessageId = messageId;
}
// Emit streaming update
this.onMessage({
type: 'streaming',
messageId,
chunk,
done
});
}
/**
* Handle message complete
*/
handleMessageComplete(parsed) {
const { messageId, content, usage, metadata } = parsed;
const response = {
content,
usage,
metadata,
messageId
};
// Store completed response for idempotency
this.sentMessages.set(messageId, {
status: 'completed',
response,
timestamp: Date.now()
});
// Clean up old messages (keep last 100)
if (this.sentMessages.size > 100) {
const oldEntries = Array.from(this.sentMessages.entries())
.sort((a, b) => a[1].timestamp - b[1].timestamp)
.slice(0, 50);
oldEntries.forEach(([key]) => this.sentMessages.delete(key));
}
// Resolve pending promise
const pending = this.pendingMessages.get(messageId);
if (pending) {
pending.resolve(response);
this.pendingMessages.delete(messageId);
}
this.onMessage({
type: 'complete',
messageId,
...response
});
}
/**
* Handle errors
*/
handleError(parsed) {
const { messageId, error, code } = parsed;
console.error([HolySheep] Error for message ${messageId}:, error);
// Reject pending promise
const pending = this.pendingMessages.get(messageId);
if (pending) {
pending.reject(new Error(error));
this.pendingMessages.delete(messageId);
}
this.onError({ messageId, error, code });
}
/**
* Schedule reconnection with exponential backoff
*/
scheduleReconnect() {
this.reconnectAttempts++;
const delay = this.calculateReconnectDelay();
console.log([HolySheep] Scheduling reconnect attempt ${this.reconnectAttempts}/${this.maxReconnectAttempts} in ${delay}ms);
setTimeout(async () => {
try {
await this.connect();
// Replay pending messages after successful reconnection
await this.replayPendingMessages();
} catch (error) {
console.error('[HolySheep] Reconnect failed:', error);
if (this.reconnectAttempts < this.maxReconnectAttempts) {
this.scheduleReconnect();
}
}
}, delay);
}
/**
* Replay queued messages after reconnection
*/
async replayPendingMessages() {
console.log([HolySheep] Replaying ${this.pendingMessages.size} pending messages);
for (const [messageId, pending] of this.pendingMessages) {
// Check if message is too old (5 minutes)
if (Date.now() - pending.timestamp > 300000) {
pending.reject(new Error('Message expired while waiting for reconnection'));
this.pendingMessages.delete(messageId);
continue;
}
// Generate new message ID to avoid confusion
const newMessageId = this.generateMessageId();
const newMessage = {
...pending.message,
messageId: newMessageId,
isReplay: true,
originalMessageId: messageId
};
// Update pending tracker with new ID
this.pendingMessages.set(newMessageId, {
...pending,
message: newMessage
});
this.pendingMessages.delete(messageId);
// Send the message
if (this.ws && this.ws.readyState === WebSocket.OPEN) {
this.ws.send(JSON.stringify(newMessage));
console.log([HolySheep] Replayed message: ${messageId} -> ${newMessageId});
}
}
}
/**
* Queue message when connection is down
*/
queueMessage(message) {
this.pendingMessages.set(message.messageId, {
resolve: null,
reject: null,
timestamp: Date.now(),
message: message
});
}
/**
* Start heartbeat mechanism
*/
startHeartbeat() {
this.heartbeatInterval = setInterval(() => {
if (this.ws && this.ws.readyState === WebSocket.OPEN) {
const heartbeat = {
type: 'heartbeat',
timestamp: Date.now()
};
this.ws.send(JSON.stringify(heartbeat));
// Check if last heartbeat was too long ago
if (this.lastHeartbeat && Date.now() - this.lastHeartbeat > 60000) {
console.warn('[HolySheep] Heartbeat timeout, reconnecting...');
this.ws.close();
}
}
}, 30000);
}
/**
* Stop heartbeat
*/
stopHeartbeat() {
if (this.heartbeatInterval) {
clearInterval(this.heartbeatInterval);
this.heartbeatInterval = null;
}
}
/**
* Gracefully close connection
*/
close() {
this.stopHeartbeat();
if (this.ws) {
this.ws.close(1000, 'Client initiated close');
}
this.pendingMessages.clear();
console.log('[HolySheep] WebSocket connection closed gracefully');
}
}
// Export for different module systems
if (typeof module !== 'undefined' && module.exports) {
module.exports = { HolySheepWebSocketManager };
}2. Server-Side Idempotency Service (Node.js + Redis)
/**
* HolySheep Idempotency Service
* Server-side duplicate message detection
*/
const Redis = require('ioredis');
const crypto = require('crypto');
class IdempotencyService {
constructor(redisOptions = {}) {
this.redis = new Redis(redisOptions);
this.defaultTTL = 300; // 5 minutes
// Key prefixes
this.MESSAGE_KEY_PREFIX = 'idempotency:message:';
this.SESSION_KEY_PREFIX = 'idempotency:session:';
this.LOCK_KEY_PREFIX = 'idempotency:lock:';
}
/**
* Generate idempotency key from message
*/
generateIdempotencyKey(message) {
const data = ${message.sessionId}:${message.content}:${message.clientTimestamp};
return crypto.createHash('sha256').update(data).digest('hex').substring(0, 32);
}
/**
* Check if message is duplicate (idempotency check)
* @returns {Object} { isDuplicate: boolean, existingResponse: Object | null }
*/
async checkIdempotency(sessionId, messageId, content) {
const key = ${this.MESSAGE_KEY_PREFIX}${sessionId}:${messageId};
try {
// Try to get existing response
const existing = await this.redis.get(key);
if (existing) {
const parsed = JSON.parse(existing);
console.log([Idempotency] Duplicate detected: ${messageId});
return {
isDuplicate: true,
existingResponse: parsed.response,
originalTimestamp: parsed.timestamp
};
}
return { isDuplicate: false, existingResponse: null };
} catch (error) {
console.error('[Idempotency] Redis error:', error);
return { isDuplicate: false, existingResponse: null };
}
}
/**
* Store message response for idempotency
*/
async storeResponse(sessionId, messageId, response) {
const key = ${this.MESSAGE_KEY_PREFIX}${sessionId}:${messageId};
try {
await this.redis.setex(
key,
this.defaultTTL,
JSON.stringify({
response,
timestamp: Date.now(),
sessionId
})
);
console.log([Idempotency] Stored response for: ${messageId});
} catch (error) {
console.error('[Idempotency] Failed to store response:', error);
}
}
/**
* Acquire lock for processing (prevents race conditions)
*/
async acquireLock(sessionId, messageId, ttlSeconds = 30) {
const lockKey = ${this.LOCK_KEY_PREFIX}${sessionId}:${messageId};
try {
const result = await this.redis.set(
lockKey,
Date.now().toString(),
'EX',
ttlSeconds,
'NX'
);
return result === 'OK';
} catch (error) {
console.error('[Idempotency] Lock acquisition error:', error);
return false;
}
}
/**
* Release lock
*/
async releaseLock(sessionId, messageId) {
const lockKey = ${this.LOCK_KEY_PREFIX}${sessionId}:${messageId};
await this.redis.del(lockKey);
}
/**
* Track active session
*/
async trackSession(sessionId, metadata = {}) {
const key = ${this.SESSION_KEY_PREFIX}${sessionId};
try {
await this.redis.hset(key, {
...metadata,
lastActivity: Date.now(),
active: 'true'
});
await this.redis.expire(key, 3600); // 1 hour TTL
} catch (error) {
console.error('[Idempotency] Session tracking error:', error);
}
}
/**
* Get session state for recovery
*/
async getSessionState(sessionId) {
const key = ${this.SESSION_KEY_PREFIX}${sessionId};
try {
const state = await this.redis.hgetall(key);
return state && Object.keys(state).length > 0 ? state : null;
} catch (error) {
console.error('[Idempotency] Get session error:', error);
return null;
}
}
/**
* Full idempotency check with lock acquisition
* Returns cached response if duplicate, null otherwise
*/
async processWithIdempotency(sessionId, messageId, content, processFn) {
// Step 1: Check for existing response
const { isDuplicate, existingResponse } = await this.checkIdempotency(
sessionId,
messageId,
content
);
if (isDuplicate) {
return existingResponse;
}
// Step 2: Try to acquire processing lock
const lockAcquired = await this.acquireLock(sessionId, messageId, 60);
if (!lockAcquired) {
// Another process is handling this message
// Wait and retry
await new Promise(resolve => setTimeout(resolve, 500));
const retry = await this.checkIdempotency(sessionId, messageId, content);
if (retry.isDuplicate) {
return retry.existingResponse;
}
throw new Error('Failed to acquire lock after retry');
}
try {
// Step 3: Process the message
const response = await processFn();
// Step 4: Store response for future idempotency
await this.storeResponse(sessionId, messageId, response);
return response;
} finally {
// Always release lock
await this.releaseLock(sessionId, messageId);
}
}
}
/**
* HolySheep Chat Service - Complete Implementation
*/
const axios = require('axios');
class HolySheepChatService {
constructor(apiKey, idempotencyService) {
this.apiKey = apiKey;
this.baseUrl = 'https://api.holysheep.ai/v1';
this.idempotency = idempotencyService;
this.sessions = new Map();
}
/**
* Stream chat with HolySheep AI (using REST streaming)
* Compatible with WebSocket pattern for real-time updates
*/
async *streamChat(sessionId, messages, options = {}) {
const model = options.model || 'gpt-4.1';
// Transform messages to OpenAI format for HolySheep
const formattedMessages = messages.map(msg => ({
role: msg.role,
content: msg.content
}));
try {
const response = await axios.post(
${this.baseUrl}/chat/completions,
{
model: model,
messages: formattedMessages,
stream: true,
temperature: options.temperature || 0.7,
max_tokens: options.maxTokens || 2000
},
{
headers: {
'Authorization': Bearer ${this.apiKey},
'Content-Type': 'application/json'
},
responseType: 'stream'
}
);
let fullContent = '';
let usage = null;
let messageId = msg_${Date.now()}_${Math.random().toString(36).substr(2, 9)};
for await (const chunk of response.data) {
const lines = chunk.toString().split('\n');
for (const line of lines) {
if (line.startsWith('data: ')) {
const data = line.slice(6);
if (data === '[DONE]') {
// Store final response for idempotency
await this.idempotency.storeResponse(sessionId, messageId, {
content: fullContent,
usage: usage,
messageId: messageId
});
yield {
type: 'complete',
messageId: messageId,
content: fullContent,
usage: usage
};
return;
}
try {
const parsed = JSON.parse(data);
if (parsed.choices && parsed.choices[0].delta.content) {
const content = parsed.choices[0].delta.content;
fullContent += content;
yield {
type: 'streaming',
messageId: messageId,
chunk: content
};
}
if (parsed.usage) {
usage = parsed.usage;
}
} catch (e) {
// Skip invalid JSON chunks
}
}
}
}
} catch (error) {
console.error('[HolySheep] API Error:', error.response?.data || error.message);
yield {
type: 'error',
error: error.message,
code: error.response?.status || 500
};
}
}
/**
* Non-streaming chat with full idempotency
*/
async chat(sessionId, messages, options = {}) {
const messageId = msg_${Date.now()}_${Math.random().toString(36).substr(2, 9)};
// Check idempotency
const lastMessage = messages[messages.length - 1];
const { isDuplicate, existingResponse } = await this.idempotency.checkIdempotency(
sessionId,
messageId,
lastMessage.content
);
if (isDuplicate) {
console.log([HolySheep] Returning cached response for: ${messageId});
return existingResponse;
}
// Acquire lock and process
return await this.idempotency.processWithIdempotency(
sessionId,
messageId,
lastMessage.content,
async () => {
const model = options.model || 'gpt-4.1';
const formattedMessages = messages.map(msg => ({
role: msg.role,
content: msg.content
}));
const response = await axios.post(
${this.baseUrl}/chat/completions,
{
model: model,
messages: formattedMessages,
temperature: options.temperature || 0.7,
max_tokens: options.maxTokens || 2000
},
{
headers: {
'Authorization': Bearer ${this.apiKey},
'Content-Type': 'application/json'
}
}
);
return {
content: response.data.choices[0].message.content,
usage: response.data.usage,
messageId: messageId,
model: response.data.model
};
}
);
}
}
// Usage Example
async function main() {
// Initialize services
const idempotency = new IdempotencyService({
host: process.env.REDIS_HOST || 'localhost',
port: process.env.REDIS_PORT || 6379
});
const chatService = new HolySheepChatService(
'YOUR_HOLYSHEEP_API_KEY',
idempotency
);
const sessionId = 'user_123_session_456';
// Track session
await idempotency.trackSession(sessionId, {
userId: 'user_123',
platform: 'web'
});
// First request - will call API
console.log('=== First Request ===');
const response1 = await chatService.chat(
sessionId,
[{ role: 'user', content: 'Xin chào, tôi muốn tìm hiểu về AI' }]
);
console.log('Response 1:', response1.content.substring(0, 100) + '...');
// Second request with same content - returns cached
console.log('\n=== Duplicate Request (should be cached) ===');
const response2 = await chatService.chat(
sessionId,
[{ role: 'user', content: 'Xin chào, tôi muốn tìm hiểu về AI' }]
);
console.log('Response 2 (cached):', response2.content.substring(0, 100) + '...');
console.log('Same response:', response1.messageId === response2.messageId);
}
main().catch(console.error);
module.exports = { IdempotencyService, HolySheepChatService };3. Frontend Integration Example (React)
/**
* HolySheep AI Chat Hook - React Integration
* Complete with auto-reconnect and idempotency
*/
import { useState, useEffect, useRef, useCallback } from 'react';
// Assuming HolySheepWebSocketManager is available (from previous code block)
const { HolySheepWebSocketManager } = require('./HolySheepWebSocketManager');
export function useHolySheepChat(apiKey, options = {}) {
const [messages, setMessages] = useState([]);
const [isConnected, setIsConnected] = useState(false);
const [isLoading, setIsLoading] = useState(false);
const [error, setError] = useState(null);
const wsManagerRef = useRef(null);
const sessionIdRef = useRef(session_${Date.now()}_${Math.random().toString(36).substr(2, 9)});
const messageBufferRef = useRef([]);
// Initialize WebSocket connection
useEffect(() => {
if (!apiKey || apiKey === 'YOUR_HOLYSHEEP_API_KEY') {
setError('Vui lòng cung cấp API key hợp lệ');
return;
}
const wsManager = new HolySheepWebSocketManager({
apiKey,
maxReconnectAttempts: 10,
reconnectDelay: 1000,
maxReconnectDelay: 30000,
onMessage: handleIncomingMessage,
onConnectionChange: handleConnectionChange,
onError: handleError
});
wsManagerRef.current = wsManager;
// Connect
wsManager.connect(sessionIdRef.current).catch(err => {
console.error('Connection failed:', err);
setError('Kết nối thất bại');
});
return () => {
wsManager.close();
};
}, [apiKey]);
// Handle incoming messages
const handleIncomingMessage = useCallback((msg) => {
switch (msg.type) {
case 'streaming':
// Update streaming message
setMessages(prev => {
const lastMsg = prev[prev.length - 1];
if (lastMsg?.role === 'assistant' && lastMsg.isStreaming) {
return [
...prev.slice(0, -1),
{
...lastMsg,
content: lastMsg.content + msg.chunk,
messageId: msg.messageId
}
];
}
return prev;
});
break;
case 'complete':
// Mark streaming as complete
setMessages(prev => {
return prev.map(msg =>
msg.messageId === msg.messageId
? { ...msg, isStreaming: false }
: msg
);
});
setIsLoading(false);
break;
case 'error':
setError(msg.error);
setIsLoading(false);
break;
}
}, []);
// Handle connection state changes
const handleConnectionChange = useCallback((state) => {
console.log('Connection state:', state);
setIsConnected(state === 'connected');
if (state === 'connected') {
setError(null);
}
}, []);
// Handle errors
const handleError = useCallback((err) => {
console.error('WebSocket error:', err);
setError(err.error || 'Đã xảy ra lỗi');
setIsLoading(false);
}, []);
// Send message
const sendMessage = useCallback(async (content) => {
if (!wsManagerRef.current) {
setError('Chưa kết nối');
return;
}
if (!content.trim()) return;
// Add user message to state
const userMessage = {
id: user_${Date.now()},
role: 'user',
content: content.trim(),
timestamp: Date.now()
};
setMessages(prev => [...prev, userMessage]);
setIsLoading(true);
setError(null);
// Add placeholder for assistant
const assistantMessage = {
id: assistant_${Date.now()},
role: 'assistant',
content: '',
isStreaming: true,
timestamp: Date.now()
};
setMessages(prev => [...prev, assistantMessage]);
try {
// Send via WebSocket (with built-in idempotency)
const response = await wsManagerRef.current.sendMessage(
JSON.stringify({
sessionId: sessionIdRef.current,
messages: [
...messages.map(m => ({ role: m.role, content: m.content })),
{ role: 'user', content: content.trim() }
],
model: options.model || 'gpt-4.1',
temperature: options.temperature || 0.7
}),
{
userId: options.userId,
timestamp: Date.now()
}
);
// If response comes through WS, it will be handled by onMessage
// If it comes through REST, update here
if (response && response.content) {
setMessages(prev => {
const lastMsg = prev[prev.length - 1];
if (lastMsg?.role === 'assistant') {
return [
...prev.slice(0, -1),
{
...lastMsg,
content: response.content,
usage: response.usage,
isStreaming: false
}
];
}
return prev;
});
}
} catch (err) {
console.error('Send message error:', err);
setError(err.message || 'Gửi tin nhắn thất bại');
setMessages(prev => prev.filter(m => m.id !== assistantMessage.id));
setIsLoading(false);
}
}, [messages, options]);
// Retry failed message
const retryMessage = useCallback((messageId) => {
const failedMessage = messages.find(m => m.id === messageId);
if (failedMessage) {
// Remove failed message
setMessages(prev => prev.filter(m => m.id !== messageId));
// Resend
sendMessage(failedMessage.content.replace('/retry ', ''));
}
}, [messages, sendMessage]);
// Clear conversation
const clearConversation = useCallback(() => {
setMessages([]);
setError(null);
}, []);
return {
messages,
isConnected,
isLoading,
error,
sendMessage,
retryMessage,
clearConversation,
sessionId: sessionIdRef.current
};
}
/**
* Chat Component Example
*/
export function HolySheepChatComponent({ apiKey }) {
const {
messages,
isConnected,
isLoading,
error,
sendMessage,
clearConversation
} = useHolySheepChat(apiKey, { model: 'gpt-4.1' });
const [input, setInput] =