316 lines
9.7 KiB
TypeScript
316 lines
9.7 KiB
TypeScript
/** Shared, authenticated and self-healing WebSocket transport. */
|
|
|
|
import { apiClient } from '@/lib/api';
|
|
import { AUTH_SESSION_EVENT, type AuthSessionEventDetail } from '@/lib/api/client';
|
|
import { getWebSocketUrl } from '@/lib/url';
|
|
|
|
export type RealtimeEvent = {
|
|
type: string;
|
|
[key: string]: unknown;
|
|
};
|
|
type MessageHandler = (data: RealtimeEvent) => void;
|
|
export type ConnectionStatus = {
|
|
isConnected: boolean;
|
|
isConnecting: boolean;
|
|
reconnectAttempts: number;
|
|
maxReconnectAttempts: number;
|
|
};
|
|
type StatusHandler = (status: ConnectionStatus) => void;
|
|
type WebSocketTicketResponse = { ticket: string; expires_in: number };
|
|
|
|
const RAPID_RECONNECT_LIMIT = 5;
|
|
const MAX_RECONNECT_DELAY_MS = 60_000;
|
|
const HEARTBEAT_TIMEOUT_MS = 70_000;
|
|
const HEARTBEAT_CHECK_MS = 15_000;
|
|
|
|
export class WebSocketManager {
|
|
private static instance: WebSocketManager | null = null;
|
|
private ws: WebSocket | null = null;
|
|
private reconnectTimeout: ReturnType<typeof setTimeout> | null = null;
|
|
private heartbeatTimer: ReturnType<typeof setInterval> | null = null;
|
|
private reconnectAttempts = 0;
|
|
private activeChatId: number | null = null;
|
|
private readonly messageHandlers = new Set<MessageHandler>();
|
|
private readonly statusHandlers = new Set<StatusHandler>();
|
|
private isConnecting = false;
|
|
private shouldReconnect = false;
|
|
private connectionGeneration = 0;
|
|
private lastServerMessageAt = 0;
|
|
|
|
private constructor() {
|
|
if (typeof window === 'undefined') return;
|
|
|
|
window.addEventListener(AUTH_SESSION_EVENT, (event) => {
|
|
const detail = (event as CustomEvent<AuthSessionEventDetail>).detail;
|
|
if (detail?.type === 'session-cleared') {
|
|
this.disconnect();
|
|
} else if (detail?.type === 'token-refreshed' && !this.ws && this.hasSubscribers()) {
|
|
this.reconnectAttempts = 0;
|
|
this.connect();
|
|
}
|
|
});
|
|
window.addEventListener('storage', (event) => {
|
|
if (event.key !== 'access_token') return;
|
|
if (!event.newValue) this.disconnect();
|
|
else if (!this.ws && this.hasSubscribers()) this.connectImmediately();
|
|
});
|
|
window.addEventListener('online', () => this.connectImmediately());
|
|
if (typeof document !== 'undefined') {
|
|
document.addEventListener('visibilitychange', () => {
|
|
if (document.visibilityState === 'visible') this.connectImmediately();
|
|
});
|
|
}
|
|
}
|
|
|
|
static getInstance(): WebSocketManager {
|
|
if (!WebSocketManager.instance) {
|
|
WebSocketManager.instance = new WebSocketManager();
|
|
}
|
|
return WebSocketManager.instance;
|
|
}
|
|
|
|
private hasSubscribers(): boolean {
|
|
return this.messageHandlers.size > 0 || this.statusHandlers.size > 0;
|
|
}
|
|
|
|
private getConnectionStatus(): ConnectionStatus {
|
|
return {
|
|
isConnected: this.ws?.readyState === WebSocket.OPEN,
|
|
isConnecting: this.isConnecting,
|
|
reconnectAttempts: this.reconnectAttempts,
|
|
maxReconnectAttempts: RAPID_RECONNECT_LIMIT,
|
|
};
|
|
}
|
|
|
|
private notifyStatus() {
|
|
const status = this.getConnectionStatus();
|
|
this.statusHandlers.forEach((handler) => {
|
|
try {
|
|
handler(status);
|
|
} catch (error) {
|
|
console.error('[WS Manager] Status handler error:', error);
|
|
}
|
|
});
|
|
}
|
|
|
|
send(payload: Record<string, unknown>): boolean {
|
|
if (this.ws?.readyState !== WebSocket.OPEN) return false;
|
|
this.ws.send(JSON.stringify(payload));
|
|
return true;
|
|
}
|
|
|
|
private syncActiveChatOnSocket() {
|
|
if (this.activeChatId === null) return;
|
|
this.send({
|
|
action: 'open_chat',
|
|
chat_id: this.activeChatId,
|
|
load_history: false,
|
|
});
|
|
}
|
|
|
|
private async requestTicket(): Promise<string> {
|
|
const response = await apiClient.request<WebSocketTicketResponse>(
|
|
'/api/auth/ws-ticket/',
|
|
{ method: 'POST' },
|
|
);
|
|
if (!response.ticket) throw new Error('WebSocket ticket is missing');
|
|
return response.ticket;
|
|
}
|
|
|
|
connect() {
|
|
if (this.ws?.readyState === WebSocket.OPEN || this.isConnecting) return;
|
|
if (!this.hasSubscribers()) {
|
|
this.shouldReconnect = false;
|
|
return;
|
|
}
|
|
if (typeof window === 'undefined' || !localStorage.getItem('access_token')) {
|
|
this.notifyStatus();
|
|
return;
|
|
}
|
|
|
|
this.clearReconnectTimer();
|
|
this.shouldReconnect = true;
|
|
this.isConnecting = true;
|
|
const generation = ++this.connectionGeneration;
|
|
this.notifyStatus();
|
|
void this.openConnection(generation);
|
|
}
|
|
|
|
private async openConnection(generation: number) {
|
|
try {
|
|
const ticket = await this.requestTicket();
|
|
if (
|
|
generation !== this.connectionGeneration
|
|
|| !this.shouldReconnect
|
|
|| !this.hasSubscribers()
|
|
) return;
|
|
|
|
const baseUrl = getWebSocketUrl();
|
|
const wsUrl = `${baseUrl}${baseUrl.includes('?') ? '&' : '?'}ticket=${encodeURIComponent(ticket)}`;
|
|
const socket = new WebSocket(wsUrl);
|
|
this.ws = socket;
|
|
|
|
socket.onopen = () => {
|
|
if (this.ws !== socket) return;
|
|
this.isConnecting = false;
|
|
this.reconnectAttempts = 0;
|
|
this.lastServerMessageAt = Date.now();
|
|
this.startHeartbeatWatchdog(socket);
|
|
this.syncActiveChatOnSocket();
|
|
this.notifyStatus();
|
|
};
|
|
|
|
socket.onmessage = (event) => {
|
|
if (this.ws !== socket) return;
|
|
this.lastServerMessageAt = Date.now();
|
|
try {
|
|
const data = JSON.parse(event.data) as RealtimeEvent;
|
|
if (data.type === 'ping') {
|
|
this.send({ action: 'pong', timestamp: data.timestamp });
|
|
return;
|
|
}
|
|
this.messageHandlers.forEach((handler) => {
|
|
try {
|
|
handler(data);
|
|
} catch (error) {
|
|
console.error('[WS Manager] Handler error:', error);
|
|
}
|
|
});
|
|
} catch (error) {
|
|
console.error('[WS Manager] Parse error:', error);
|
|
}
|
|
};
|
|
|
|
socket.onerror = () => {
|
|
if (this.ws !== socket) return;
|
|
this.isConnecting = false;
|
|
this.notifyStatus();
|
|
};
|
|
|
|
socket.onclose = () => {
|
|
if (this.ws !== socket) return;
|
|
this.stopHeartbeatWatchdog();
|
|
this.ws = null;
|
|
this.isConnecting = false;
|
|
this.notifyStatus();
|
|
this.scheduleReconnect();
|
|
};
|
|
} catch (error) {
|
|
if (generation !== this.connectionGeneration) return;
|
|
console.error('[WS Manager] Connection preparation failed:', error);
|
|
this.isConnecting = false;
|
|
this.notifyStatus();
|
|
this.scheduleReconnect();
|
|
}
|
|
}
|
|
|
|
private scheduleReconnect() {
|
|
if (!this.shouldReconnect || !this.hasSubscribers() || this.reconnectTimeout) return;
|
|
this.reconnectAttempts += 1;
|
|
this.notifyStatus();
|
|
|
|
const exponentialAttempt = Math.min(this.reconnectAttempts, RAPID_RECONNECT_LIMIT);
|
|
const baseDelay = this.reconnectAttempts > RAPID_RECONNECT_LIMIT
|
|
? MAX_RECONNECT_DELAY_MS
|
|
: Math.min(1000 * 2 ** exponentialAttempt, 30_000);
|
|
const jitteredDelay = Math.round(baseDelay * (0.8 + Math.random() * 0.4));
|
|
this.reconnectTimeout = setTimeout(() => {
|
|
this.reconnectTimeout = null;
|
|
this.connect();
|
|
}, jitteredDelay);
|
|
}
|
|
|
|
private connectImmediately() {
|
|
if (!this.shouldReconnect || !this.hasSubscribers() || this.ws || this.isConnecting) return;
|
|
this.clearReconnectTimer();
|
|
this.reconnectAttempts = 0;
|
|
this.connect();
|
|
}
|
|
|
|
private startHeartbeatWatchdog(socket: WebSocket) {
|
|
this.stopHeartbeatWatchdog();
|
|
this.heartbeatTimer = setInterval(() => {
|
|
if (
|
|
this.ws === socket
|
|
&& socket.readyState === WebSocket.OPEN
|
|
&& Date.now() - this.lastServerMessageAt > HEARTBEAT_TIMEOUT_MS
|
|
) {
|
|
socket.close(4408, 'Heartbeat timeout');
|
|
}
|
|
}, HEARTBEAT_CHECK_MS);
|
|
}
|
|
|
|
private stopHeartbeatWatchdog() {
|
|
if (!this.heartbeatTimer) return;
|
|
clearInterval(this.heartbeatTimer);
|
|
this.heartbeatTimer = null;
|
|
}
|
|
|
|
private clearReconnectTimer() {
|
|
if (!this.reconnectTimeout) return;
|
|
clearTimeout(this.reconnectTimeout);
|
|
this.reconnectTimeout = null;
|
|
}
|
|
|
|
disconnect() {
|
|
this.shouldReconnect = false;
|
|
this.connectionGeneration += 1;
|
|
this.clearReconnectTimer();
|
|
this.stopHeartbeatWatchdog();
|
|
const socket = this.ws;
|
|
this.ws = null;
|
|
if (socket) {
|
|
socket.onclose = null;
|
|
socket.close(1000, 'Client disconnect');
|
|
}
|
|
this.isConnecting = false;
|
|
this.reconnectAttempts = 0;
|
|
this.notifyStatus();
|
|
}
|
|
|
|
setActiveChat(chatId: number) {
|
|
if (!Number.isFinite(chatId) || chatId <= 0 || this.activeChatId === chatId) return;
|
|
const previousChatId = this.activeChatId;
|
|
this.activeChatId = chatId;
|
|
if (previousChatId !== null) {
|
|
this.send({ action: 'close_chat', chat_id: previousChatId });
|
|
}
|
|
this.syncActiveChatOnSocket();
|
|
}
|
|
|
|
clearActiveChat(chatId?: number) {
|
|
if (this.activeChatId === null) return;
|
|
if (typeof chatId === 'number' && this.activeChatId !== chatId) return;
|
|
const chatToClose = this.activeChatId;
|
|
this.activeChatId = null;
|
|
this.send({ action: 'close_chat', chat_id: chatToClose });
|
|
}
|
|
|
|
subscribe(handler: MessageHandler) {
|
|
this.messageHandlers.add(handler);
|
|
this.shouldReconnect = true;
|
|
if (this.hasSubscribers() && !this.ws && !this.isConnecting) this.connect();
|
|
return () => {
|
|
this.messageHandlers.delete(handler);
|
|
if (!this.hasSubscribers()) this.disconnect();
|
|
};
|
|
}
|
|
|
|
subscribeStatus(handler: StatusHandler) {
|
|
this.statusHandlers.add(handler);
|
|
handler(this.getConnectionStatus());
|
|
this.shouldReconnect = true;
|
|
if (this.hasSubscribers() && !this.ws && !this.isConnecting) this.connect();
|
|
return () => {
|
|
this.statusHandlers.delete(handler);
|
|
if (!this.hasSubscribers()) this.disconnect();
|
|
};
|
|
}
|
|
|
|
getState(): number {
|
|
return this.ws?.readyState ?? 3;
|
|
}
|
|
}
|
|
|
|
export default WebSocketManager.getInstance();
|