detached d483b27

refactor(cleanup): remove SSE implementation (replaced by Convex real-time)

commit
d483b27d840751617444380083c4902b69d406c1
author
bnema · · 2026-01-18 18:35 UTC
parent
59801ed
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
9 files · +0 -1965
Split
UnifiedSplit
D src/lib/sse/MessageQueue.ts +0 -227
@@ -1,227 +0,0 @@
1-/**
2- * Server-side Message Queue for SSE Catch-up
3- *
4- * Stores recent messages to allow clients to catch up after reconnection
5- * using the Last-Event-ID mechanism.
6- */
7-
8-import { SSE_CONFIG } from './config';
9-
10-/**
11- * Queued message structure
12- */
13-export interface QueuedMessage {
14- /** Unique event ID for Last-Event-ID support */
15- id: string;
16- /** Event type (chatMessage, userStatusUpdate, etc.) */
17- event: string;
18- /** Event payload */
19- data: unknown;
20- /** Timestamp for TTL expiration */
21- timestamp: number;
22- /** Chat room ID for filtering */
23- roomId: string;
24-}
25-
26-/**
27- * Message Queue for SSE catch-up support
28- *
29- * Maintains a queue of recent messages per room to allow clients
30- * to catch up after reconnection using Last-Event-ID.
31- */
32-class MessageQueue {
33- private queues: Map<string, QueuedMessage[]> = new Map();
34- private readonly ttl: number;
35- private readonly maxSize: number;
36- private cleanupInterval: ReturnType<typeof setInterval> | null = null;
37-
38- constructor(
39- ttl: number = SSE_CONFIG.MESSAGE_QUEUE_TTL,
40- maxSize: number = SSE_CONFIG.MESSAGE_QUEUE_MAX_SIZE
41- ) {
42- this.ttl = ttl;
43- this.maxSize = maxSize;
44-
45- // Start periodic cleanup
46- this.startCleanup();
47- console.log('[MessageQueue] Initialized:', { ttl, maxSize });
48- }
49-
50- /**
51- * Add a message to the queue
52- * @returns The generated event ID
53- */
54- enqueue(roomId: string, event: string, data: unknown): string {
55- const id = crypto.randomUUID();
56- const message: QueuedMessage = {
57- id,
58- event,
59- data,
60- timestamp: Date.now(),
61- roomId
62- };
63-
64- let queue = this.queues.get(roomId);
65- if (!queue) {
66- queue = [];
67- this.queues.set(roomId, queue);
68- }
69-
70- queue.push(message);
71-
72- // Trim if over max size (keep newest)
73- if (queue.length > this.maxSize) {
74- const trimmed = queue.length - this.maxSize;
75- this.queues.set(roomId, queue.slice(-this.maxSize));
76- console.log('[MessageQueue] Queue trimmed:', { roomId, trimmed });
77- }
78-
79- return id;
80- }
81-
82- /**
83- * Get messages since a given event ID
84- * If the ID is not found, returns the most recent messages
85- */
86- getMessagesSince(roomId: string, lastEventId: string): QueuedMessage[] {
87- const queue = this.queues.get(roomId);
88- if (!queue || queue.length === 0) {
89- return [];
90- }
91-
92- const lastIndex = queue.findIndex((m) => m.id === lastEventId);
93-
94- if (lastIndex === -1) {
95- // ID not found - return recent messages as fallback
96- console.log('[MessageQueue] Last-Event-ID not found, returning recent:', {
97- roomId,
98- lastEventId,
99- returning: Math.min(50, queue.length)
100- });
101- return queue.slice(-50);
102- }
103-
104- // Return messages after the last seen ID
105- const missed = queue.slice(lastIndex + 1);
106- if (missed.length > 0) {
107- console.log('[MessageQueue] Catch-up messages:', {
108- roomId,
109- lastEventId,
110- missedCount: missed.length
111- });
112- }
113- return missed;
114- }
115-
116- /**
117- * Get all messages for a room (for initial load)
118- */
119- getMessages(roomId: string, limit: number = 50): QueuedMessage[] {
120- const queue = this.queues.get(roomId);
121- if (!queue) {
122- return [];
123- }
124- return queue.slice(-limit);
125- }
126-
127- /**
128- * Get the latest event ID for a room
129- */
130- getLatestEventId(roomId: string): string | null {
131- const queue = this.queues.get(roomId);
132- if (!queue || queue.length === 0) {
133- return null;
134- }
135- return queue[queue.length - 1].id;
136- }
137-
138- /**
139- * Check if an event ID exists in the queue
140- */
141- hasEventId(roomId: string, eventId: string): boolean {
142- const queue = this.queues.get(roomId);
143- if (!queue) {
144- return false;
145- }
146- return queue.some((m) => m.id === eventId);
147- }
148-
149- /**
150- * Remove expired messages from all queues
151- */
152- cleanup(): number {
153- const now = Date.now();
154- let totalRemoved = 0;
155-
156- for (const [roomId, queue] of this.queues.entries()) {
157- const originalLength = queue.length;
158- const filtered = queue.filter((m) => now - m.timestamp < this.ttl);
159-
160- if (filtered.length !== originalLength) {
161- totalRemoved += originalLength - filtered.length;
162- if (filtered.length === 0) {
163- this.queues.delete(roomId);
164- } else {
165- this.queues.set(roomId, filtered);
166- }
167- }
168- }
169-
170- if (totalRemoved > 0) {
171- console.log('[MessageQueue] Cleanup completed:', {
172- removed: totalRemoved,
173- activeRooms: this.queues.size
174- });
175- }
176-
177- return totalRemoved;
178- }
179-
180- /**
181- * Start periodic cleanup interval
182- */
183- private startCleanup(): void {
184- // Cleanup every minute
185- this.cleanupInterval = setInterval(() => {
186- this.cleanup();
187- }, 60_000);
188- }
189-
190- /**
191- * Stop cleanup interval (for shutdown)
192- */
193- stopCleanup(): void {
194- if (this.cleanupInterval) {
195- clearInterval(this.cleanupInterval);
196- this.cleanupInterval = null;
197- }
198- }
199-
200- /**
201- * Clear all queues (for testing or reset)
202- */
203- clear(): void {
204- this.queues.clear();
205- console.log('[MessageQueue] All queues cleared');
206- }
207-
208- /**
209- * Get statistics about the queue
210- */
211- getStats(): { rooms: number; totalMessages: number } {
212- let totalMessages = 0;
213- for (const queue of this.queues.values()) {
214- totalMessages += queue.length;
215- }
216- return {
217- rooms: this.queues.size,
218- totalMessages
219- };
220- }
221-}
222-
223-// Singleton instance
224-export const messageQueue = new MessageQueue();
225-
226-// Export class for testing
227-export { MessageQueue };
@@ -1,227 +0,0 @@
1-/**
2- * Server-side Message Queue for SSE Catch-up
3- *
4- * Stores recent messages to allow clients to catch up after reconnection
5- * using the Last-Event-ID mechanism.
6- */
7-
8-import { SSE_CONFIG } from './config';
9-
10-/**
11- * Queued message structure
12- */
13-export interface QueuedMessage {
14- /** Unique event ID for Last-Event-ID support */
15- id: string;
16- /** Event type (chatMessage, userStatusUpdate, etc.) */
17- event: string;
18- /** Event payload */
19- data: unknown;
20- /** Timestamp for TTL expiration */
21- timestamp: number;
22- /** Chat room ID for filtering */
23- roomId: string;
24-}
25-
26-/**
27- * Message Queue for SSE catch-up support
28- *
29- * Maintains a queue of recent messages per room to allow clients
30- * to catch up after reconnection using Last-Event-ID.
31- */
32-class MessageQueue {
33- private queues: Map<string, QueuedMessage[]> = new Map();
34- private readonly ttl: number;
35- private readonly maxSize: number;
36- private cleanupInterval: ReturnType<typeof setInterval> | null = null;
37-
38- constructor(
39- ttl: number = SSE_CONFIG.MESSAGE_QUEUE_TTL,
40- maxSize: number = SSE_CONFIG.MESSAGE_QUEUE_MAX_SIZE
41- ) {
42- this.ttl = ttl;
43- this.maxSize = maxSize;
44-
45- // Start periodic cleanup
46- this.startCleanup();
47- console.log('[MessageQueue] Initialized:', { ttl, maxSize });
48- }
49-
50- /**
51- * Add a message to the queue
52- * @returns The generated event ID
53- */
54- enqueue(roomId: string, event: string, data: unknown): string {
55- const id = crypto.randomUUID();
56- const message: QueuedMessage = {
57- id,
58- event,
59- data,
60- timestamp: Date.now(),
61- roomId
62- };
63-
64- let queue = this.queues.get(roomId);
65- if (!queue) {
66- queue = [];
67- this.queues.set(roomId, queue);
68- }
69-
70- queue.push(message);
71-
72- // Trim if over max size (keep newest)
73- if (queue.length > this.maxSize) {
74- const trimmed = queue.length - this.maxSize;
75- this.queues.set(roomId, queue.slice(-this.maxSize));
76- console.log('[MessageQueue] Queue trimmed:', { roomId, trimmed });
77- }
78-
79- return id;
80- }
81-
82- /**
83- * Get messages since a given event ID
84- * If the ID is not found, returns the most recent messages
85- */
86- getMessagesSince(roomId: string, lastEventId: string): QueuedMessage[] {
87- const queue = this.queues.get(roomId);
88- if (!queue || queue.length === 0) {
89- return [];
90- }
91-
92- const lastIndex = queue.findIndex((m) => m.id === lastEventId);
93-
94- if (lastIndex === -1) {
95- // ID not found - return recent messages as fallback
96- console.log('[MessageQueue] Last-Event-ID not found, returning recent:', {
97- roomId,
98- lastEventId,
99- returning: Math.min(50, queue.length)
100- });
101- return queue.slice(-50);
102- }
103-
104- // Return messages after the last seen ID
105- const missed = queue.slice(lastIndex + 1);
106- if (missed.length > 0) {
107- console.log('[MessageQueue] Catch-up messages:', {
108- roomId,
109- lastEventId,
110- missedCount: missed.length
111- });
112- }
113- return missed;
114- }
115-
116- /**
117- * Get all messages for a room (for initial load)
118- */
119- getMessages(roomId: string, limit: number = 50): QueuedMessage[] {
120- const queue = this.queues.get(roomId);
121- if (!queue) {
122- return [];
123- }
124- return queue.slice(-limit);
125- }
126-
127- /**
128- * Get the latest event ID for a room
129- */
130- getLatestEventId(roomId: string): string | null {
131- const queue = this.queues.get(roomId);
132- if (!queue || queue.length === 0) {
133- return null;
134- }
135- return queue[queue.length - 1].id;
136- }
137-
138- /**
139- * Check if an event ID exists in the queue
140- */
141- hasEventId(roomId: string, eventId: string): boolean {
142- const queue = this.queues.get(roomId);
143- if (!queue) {
144- return false;
145- }
146- return queue.some((m) => m.id === eventId);
147- }
148-
149- /**
150- * Remove expired messages from all queues
151- */
152- cleanup(): number {
153- const now = Date.now();
154- let totalRemoved = 0;
155-
156- for (const [roomId, queue] of this.queues.entries()) {
157- const originalLength = queue.length;
158- const filtered = queue.filter((m) => now - m.timestamp < this.ttl);
159-
160- if (filtered.length !== originalLength) {
161- totalRemoved += originalLength - filtered.length;
162- if (filtered.length === 0) {
163- this.queues.delete(roomId);
164- } else {
165- this.queues.set(roomId, filtered);
166- }
167- }
168- }
169-
170- if (totalRemoved > 0) {
171- console.log('[MessageQueue] Cleanup completed:', {
172- removed: totalRemoved,
173- activeRooms: this.queues.size
174- });
175- }
176-
177- return totalRemoved;
178- }
179-
180- /**
181- * Start periodic cleanup interval
182- */
183- private startCleanup(): void {
184- // Cleanup every minute
185- this.cleanupInterval = setInterval(() => {
186- this.cleanup();
187- }, 60_000);
188- }
189-
190- /**
191- * Stop cleanup interval (for shutdown)
192- */
193- stopCleanup(): void {
194- if (this.cleanupInterval) {
195- clearInterval(this.cleanupInterval);
196- this.cleanupInterval = null;
197- }
198- }
199-
200- /**
201- * Clear all queues (for testing or reset)
202- */
203- clear(): void {
204- this.queues.clear();
205- console.log('[MessageQueue] All queues cleared');
206- }
207-
208- /**
209- * Get statistics about the queue
210- */
211- getStats(): { rooms: number; totalMessages: number } {
212- let totalMessages = 0;
213- for (const queue of this.queues.values()) {
214- totalMessages += queue.length;
215- }
216- return {
217- rooms: this.queues.size,
218- totalMessages
219- };
220- }
221-}
222-
223-// Singleton instance
224-export const messageQueue = new MessageQueue();
225-
226-// Export class for testing
227-export { MessageQueue };
D src/lib/sse/SSEClient.svelte.ts +0 -322
@@ -1,322 +0,0 @@
1-/**
2- * SSE Client - Client-side SSE connection manager
3- *
4- * Handles EventSource connections with:
5- * - Unlimited reconnection with exponential backoff + jitter
6- * - Heartbeat monitoring and acknowledgment
7- * - Last-Event-ID tracking for catch-up
8- * - Reactive state using Svelte 5 runes
9- */
10-
11-import { SSE_CONFIG, type ConnectionStatus } from './config';
12-import type { SSEEventMap } from './SSEEventEmitter';
13-
14-/**
15- * Configuration for SSE Client
16- */
17-export interface SSEClientConfig {
18- /** SSE endpoint URL */
19- url: string;
20- /** Callback for incoming messages */
21- onMessage: <T extends keyof SSEEventMap>(type: T, data: SSEEventMap[T]) => void;
22- /** Callback for connection status changes */
23- onStatusChange?: (status: ConnectionStatus) => void;
24- /** Callback for errors */
25- onError?: (error: string) => void;
26-}
27-
28-/**
29- * SSE Client with Svelte 5 reactive state
30- */
31-export class SSEClient {
32- // Configuration
33- private config: SSEClientConfig;
34-
35- // Connection state
36- private eventSource: EventSource | null = null;
37- private _connectionId: string | null = null;
38-
39- // Reactive state using Svelte 5 runes
40- private _status = $state<ConnectionStatus>('disconnected');
41- private _lastEventId = $state<string | null>(null);
42- private _error = $state<string | null>(null);
43- private _reconnectAttempts = $state(0);
44-
45- // Timers
46- private heartbeatTimer: ReturnType<typeof setTimeout> | null = null;
47- private reconnectTimer: ReturnType<typeof setTimeout> | null = null;
48-
49- // Derived state
50- public status = $derived(this._status);
51- public lastEventId = $derived(this._lastEventId);
52- public error = $derived(this._error);
53- public reconnectAttempts = $derived(this._reconnectAttempts);
54- public isConnected = $derived(this._status === 'connected');
55- public isReconnecting = $derived(this._status === 'reconnecting');
56-
57- constructor(config: SSEClientConfig) {
58- this.config = config;
59- console.debug('[SSEClient] Initialized');
60- }
61-
62- /**
63- * Connect to the SSE endpoint
64- */
65- connect(): void {
66- if (this.eventSource && this.eventSource.readyState !== EventSource.CLOSED) {
67- console.debug('[SSEClient] Already connected or connecting');
68- return;
69- }
70-
71- this.setStatus('connecting');
72- this._error = null;
73-
74- try {
75- // EventSource automatically sends Last-Event-ID header on reconnection
76- this.eventSource = new EventSource(this.config.url, { withCredentials: true });
77-
78- this.eventSource.onopen = () => {
79- console.debug('[SSEClient] Connection established');
80- this.setStatus('connected');
81- this._reconnectAttempts = 0;
82- this._error = null;
83- this.startHeartbeatMonitor();
84- };
85-
86- this.eventSource.onerror = (event) => {
87- console.debug('[SSEClient] Connection error:', event);
88-
89- if (this.eventSource?.readyState === EventSource.CLOSED) {
90- this.handleDisconnection();
91- } else if (this.eventSource?.readyState === EventSource.CONNECTING) {
92- this.setStatus('reconnecting');
93- }
94- };
95-
96- // Set up event listeners
97- this.setupEventListeners();
98- } catch (error) {
99- console.error('[SSEClient] Failed to create EventSource:', error);
100- this._error = 'Failed to connect';
101- this.handleDisconnection();
102- }
103- }
104-
105- /**
106- * Disconnect from SSE
107- */
108- disconnect(): void {
109- console.debug('[SSEClient] Disconnecting');
110- this.cleanup();
111- this.setStatus('disconnected');
112- this._connectionId = null;
113- }
114-
115- /**
116- * Get the current connection ID
117- */
118- getConnectionId(): string | null {
119- return this._connectionId;
120- }
121-
122- /**
123- * Set up event listeners for SSE events
124- */
125- private setupEventListeners(): void {
126- if (!this.eventSource) return;
127-
128- // Track last event ID from all events
129- const trackEventId = (event: MessageEvent) => {
130- if (event.lastEventId) {
131- this._lastEventId = event.lastEventId;
132- }
133- };
134-
135- // Chat message events
136- this.eventSource.addEventListener('chatMessage', (event: MessageEvent) => {
137- trackEventId(event);
138- try {
139- const data = JSON.parse(event.data);
140- this.config.onMessage('chatMessage', data);
141- } catch (error) {
142- console.error('[SSEClient] Error parsing chatMessage:', error);
143- }
144- });
145-
146- // User status update events
147- this.eventSource.addEventListener('userStatusUpdate', (event: MessageEvent) => {
148- trackEventId(event);
149- try {
150- const data = JSON.parse(event.data);
151- this.config.onMessage('userStatusUpdate', data);
152- } catch (error) {
153- console.error('[SSEClient] Error parsing userStatusUpdate:', error);
154- }
155- });
156-
157- // Buddy list update events
158- this.eventSource.addEventListener('buddyListUpdate', (event: MessageEvent) => {
159- trackEventId(event);
160- try {
161- const data = JSON.parse(event.data);
162- this.config.onMessage('buddyListUpdate', data);
163- } catch (error) {
164- console.error('[SSEClient] Error parsing buddyListUpdate:', error);
165- }
166- });
167-
168- // Heartbeat events
169- this.eventSource.addEventListener('heartbeat', (event: MessageEvent) => {
170- trackEventId(event);
171- try {
172- const data = JSON.parse(event.data);
173- this._connectionId = data.connectionId;
174- this.acknowledgeHeartbeat(data.connectionId);
175- this.resetHeartbeatMonitor();
176- } catch (error) {
177- console.error('[SSEClient] Error parsing heartbeat:', error);
178- }
179- });
180- }
181-
182- /**
183- * Start monitoring for heartbeat timeout
184- */
185- private startHeartbeatMonitor(): void {
186- this.clearHeartbeatTimer();
187-
188- this.heartbeatTimer = setTimeout(() => {
189- console.warn('[SSEClient] Heartbeat timeout - connection may be stale');
190- // Force reconnection if no heartbeat received
191- if (this.eventSource) {
192- this.eventSource.close();
193- }
194- this.handleDisconnection();
195- }, SSE_CONFIG.HEARTBEAT_TIMEOUT);
196- }
197-
198- /**
199- * Reset the heartbeat monitor (called when heartbeat received)
200- */
201- private resetHeartbeatMonitor(): void {
202- this.startHeartbeatMonitor();
203- }
204-
205- /**
206- * Clear heartbeat timer
207- */
208- private clearHeartbeatTimer(): void {
209- if (this.heartbeatTimer) {
210- clearTimeout(this.heartbeatTimer);
211- this.heartbeatTimer = null;
212- }
213- }
214-
215- /**
216- * Send heartbeat acknowledgment to server
217- */
218- private async acknowledgeHeartbeat(connectionId: string): Promise<void> {
219- try {
220- const response = await fetch('/api/sse/heartbeat', {
221- method: 'POST',
222- headers: { 'Content-Type': 'application/json' },
223- credentials: 'include',
224- body: JSON.stringify({ connectionId })
225- });
226-
227- if (!response.ok) {
228- console.warn('[SSEClient] Heartbeat acknowledgment failed:', response.status);
229- }
230- } catch (error) {
231- console.warn('[SSEClient] Failed to acknowledge heartbeat:', error);
232- }
233- }
234-
235- /**
236- * Handle disconnection with reconnection logic
237- */
238- private handleDisconnection(): void {
239- this.cleanup();
240- this.setStatus('reconnecting');
241- this._reconnectAttempts++;
242-
243- const delay = this.calculateBackoff();
244- console.debug(`[SSEClient] Reconnecting in ${delay}ms (attempt ${this._reconnectAttempts})`);
245-
246- this.reconnectTimer = setTimeout(() => {
247- this.connect();
248- }, delay);
249- }
250-
251- /**
252- * Calculate reconnection delay with exponential backoff and jitter
253- * Never gives up - always tries to reconnect
254- */
255- private calculateBackoff(): number {
256- const base = SSE_CONFIG.INITIAL_RECONNECT_DELAY;
257- const max = SSE_CONFIG.MAX_RECONNECT_DELAY;
258-
259- // Exponential backoff: 2^attempts * base, capped at max
260- const exponentialDelay = Math.min(base * Math.pow(2, this._reconnectAttempts), max);
261-
262- // Add random jitter (0-25% of delay) to prevent thundering herd
263- const jitter = exponentialDelay * Math.random() * 0.25;
264-
265- return Math.floor(exponentialDelay + jitter);
266- }
267-
268- /**
269- * Get the estimated next reconnect delay in seconds
270- * Returns null if not reconnecting
271- */
272- getReconnectDelay(): number | null {
273- if (this._status !== 'reconnecting') {
274- return null;
275- }
276- // Calculate the expected delay for the current attempt
277- const base = SSE_CONFIG.INITIAL_RECONNECT_DELAY;
278- const max = SSE_CONFIG.MAX_RECONNECT_DELAY;
279- const exponentialDelay = Math.min(base * Math.pow(2, this._reconnectAttempts - 1), max);
280- return Math.ceil(exponentialDelay / 1000); // Convert to seconds
281- }
282-
283- /**
284- * Clean up resources
285- */
286- private cleanup(): void {
287- this.clearHeartbeatTimer();
288-
289- if (this.reconnectTimer) {
290- clearTimeout(this.reconnectTimer);
291- this.reconnectTimer = null;
292- }
293-
294- if (this.eventSource) {
295- this.eventSource.close();
296- this.eventSource = null;
297- }
298- }
299-
300- /**
301- * Set connection status and notify listener
302- */
303- private setStatus(status: ConnectionStatus): void {
304- this._status = status;
305- this.config.onStatusChange?.(status);
306- }
307-
308- /**
309- * Destroy the client completely
310- */
311- destroy(): void {
312- console.debug('[SSEClient] Destroying');
313- this.disconnect();
314- }
315-}
316-
317-/**
318- * Factory function to create an SSE client
319- */
320-export function createSSEClient(config: SSEClientConfig): SSEClient {
321- return new SSEClient(config);
322-}
@@ -1,322 +0,0 @@
1-/**
2- * SSE Client - Client-side SSE connection manager
3- *
4- * Handles EventSource connections with:
5- * - Unlimited reconnection with exponential backoff + jitter
6- * - Heartbeat monitoring and acknowledgment
7- * - Last-Event-ID tracking for catch-up
8- * - Reactive state using Svelte 5 runes
9- */
10-
11-import { SSE_CONFIG, type ConnectionStatus } from './config';
12-import type { SSEEventMap } from './SSEEventEmitter';
13-
14-/**
15- * Configuration for SSE Client
16- */
17-export interface SSEClientConfig {
18- /** SSE endpoint URL */
19- url: string;
20- /** Callback for incoming messages */
21- onMessage: <T extends keyof SSEEventMap>(type: T, data: SSEEventMap[T]) => void;
22- /** Callback for connection status changes */
23- onStatusChange?: (status: ConnectionStatus) => void;
24- /** Callback for errors */
25- onError?: (error: string) => void;
26-}
27-
28-/**
29- * SSE Client with Svelte 5 reactive state
30- */
31-export class SSEClient {
32- // Configuration
33- private config: SSEClientConfig;
34-
35- // Connection state
36- private eventSource: EventSource | null = null;
37- private _connectionId: string | null = null;
38-
39- // Reactive state using Svelte 5 runes
40- private _status = $state<ConnectionStatus>('disconnected');
41- private _lastEventId = $state<string | null>(null);
42- private _error = $state<string | null>(null);
43- private _reconnectAttempts = $state(0);
44-
45- // Timers
46- private heartbeatTimer: ReturnType<typeof setTimeout> | null = null;
47- private reconnectTimer: ReturnType<typeof setTimeout> | null = null;
48-
49- // Derived state
50- public status = $derived(this._status);
51- public lastEventId = $derived(this._lastEventId);
52- public error = $derived(this._error);
53- public reconnectAttempts = $derived(this._reconnectAttempts);
54- public isConnected = $derived(this._status === 'connected');
55- public isReconnecting = $derived(this._status === 'reconnecting');
56-
57- constructor(config: SSEClientConfig) {
58- this.config = config;
59- console.debug('[SSEClient] Initialized');
60- }
61-
62- /**
63- * Connect to the SSE endpoint
64- */
65- connect(): void {
66- if (this.eventSource && this.eventSource.readyState !== EventSource.CLOSED) {
67- console.debug('[SSEClient] Already connected or connecting');
68- return;
69- }
70-
71- this.setStatus('connecting');
72- this._error = null;
73-
74- try {
75- // EventSource automatically sends Last-Event-ID header on reconnection
76- this.eventSource = new EventSource(this.config.url, { withCredentials: true });
77-
78- this.eventSource.onopen = () => {
79- console.debug('[SSEClient] Connection established');
80- this.setStatus('connected');
81- this._reconnectAttempts = 0;
82- this._error = null;
83- this.startHeartbeatMonitor();
84- };
85-
86- this.eventSource.onerror = (event) => {
87- console.debug('[SSEClient] Connection error:', event);
88-
89- if (this.eventSource?.readyState === EventSource.CLOSED) {
90- this.handleDisconnection();
91- } else if (this.eventSource?.readyState === EventSource.CONNECTING) {
92- this.setStatus('reconnecting');
93- }
94- };
95-
96- // Set up event listeners
97- this.setupEventListeners();
98- } catch (error) {
99- console.error('[SSEClient] Failed to create EventSource:', error);
100- this._error = 'Failed to connect';
101- this.handleDisconnection();
102- }
103- }
104-
105- /**
106- * Disconnect from SSE
107- */
108- disconnect(): void {
109- console.debug('[SSEClient] Disconnecting');
110- this.cleanup();
111- this.setStatus('disconnected');
112- this._connectionId = null;
113- }
114-
115- /**
116- * Get the current connection ID
117- */
118- getConnectionId(): string | null {
119- return this._connectionId;
120- }
121-
122- /**
123- * Set up event listeners for SSE events
124- */
125- private setupEventListeners(): void {
126- if (!this.eventSource) return;
127-
128- // Track last event ID from all events
129- const trackEventId = (event: MessageEvent) => {
130- if (event.lastEventId) {
131- this._lastEventId = event.lastEventId;
132- }
133- };
134-
135- // Chat message events
136- this.eventSource.addEventListener('chatMessage', (event: MessageEvent) => {
137- trackEventId(event);
138- try {
139- const data = JSON.parse(event.data);
140- this.config.onMessage('chatMessage', data);
141- } catch (error) {
142- console.error('[SSEClient] Error parsing chatMessage:', error);
143- }
144- });
145-
146- // User status update events
147- this.eventSource.addEventListener('userStatusUpdate', (event: MessageEvent) => {
148- trackEventId(event);
149- try {
150- const data = JSON.parse(event.data);
151- this.config.onMessage('userStatusUpdate', data);
152- } catch (error) {
153- console.error('[SSEClient] Error parsing userStatusUpdate:', error);
154- }
155- });
156-
157- // Buddy list update events
158- this.eventSource.addEventListener('buddyListUpdate', (event: MessageEvent) => {
159- trackEventId(event);
160- try {
161- const data = JSON.parse(event.data);
162- this.config.onMessage('buddyListUpdate', data);
163- } catch (error) {
164- console.error('[SSEClient] Error parsing buddyListUpdate:', error);
165- }
166- });
167-
168- // Heartbeat events
169- this.eventSource.addEventListener('heartbeat', (event: MessageEvent) => {
170- trackEventId(event);
171- try {
172- const data = JSON.parse(event.data);
173- this._connectionId = data.connectionId;
174- this.acknowledgeHeartbeat(data.connectionId);
175- this.resetHeartbeatMonitor();
176- } catch (error) {
177- console.error('[SSEClient] Error parsing heartbeat:', error);
178- }
179- });
180- }
181-
182- /**
183- * Start monitoring for heartbeat timeout
184- */
185- private startHeartbeatMonitor(): void {
186- this.clearHeartbeatTimer();
187-
188- this.heartbeatTimer = setTimeout(() => {
189- console.warn('[SSEClient] Heartbeat timeout - connection may be stale');
190- // Force reconnection if no heartbeat received
191- if (this.eventSource) {
192- this.eventSource.close();
193- }
194- this.handleDisconnection();
195- }, SSE_CONFIG.HEARTBEAT_TIMEOUT);
196- }
197-
198- /**
199- * Reset the heartbeat monitor (called when heartbeat received)
200- */
201- private resetHeartbeatMonitor(): void {
202- this.startHeartbeatMonitor();
203- }
204-
205- /**
206- * Clear heartbeat timer
207- */
208- private clearHeartbeatTimer(): void {
209- if (this.heartbeatTimer) {
210- clearTimeout(this.heartbeatTimer);
211- this.heartbeatTimer = null;
212- }
213- }
214-
215- /**
216- * Send heartbeat acknowledgment to server
217- */
218- private async acknowledgeHeartbeat(connectionId: string): Promise<void> {
219- try {
220- const response = await fetch('/api/sse/heartbeat', {
221- method: 'POST',
222- headers: { 'Content-Type': 'application/json' },
223- credentials: 'include',
224- body: JSON.stringify({ connectionId })
225- });
226-
227- if (!response.ok) {
228- console.warn('[SSEClient] Heartbeat acknowledgment failed:', response.status);
229- }
230- } catch (error) {
231- console.warn('[SSEClient] Failed to acknowledge heartbeat:', error);
232- }
233- }
234-
235- /**
236- * Handle disconnection with reconnection logic
237- */
238- private handleDisconnection(): void {
239- this.cleanup();
240- this.setStatus('reconnecting');
241- this._reconnectAttempts++;
242-
243- const delay = this.calculateBackoff();
244- console.debug(`[SSEClient] Reconnecting in ${delay}ms (attempt ${this._reconnectAttempts})`);
245-
246- this.reconnectTimer = setTimeout(() => {
247- this.connect();
248- }, delay);
249- }
250-
251- /**
252- * Calculate reconnection delay with exponential backoff and jitter
253- * Never gives up - always tries to reconnect
254- */
255- private calculateBackoff(): number {
256- const base = SSE_CONFIG.INITIAL_RECONNECT_DELAY;
257- const max = SSE_CONFIG.MAX_RECONNECT_DELAY;
258-
259- // Exponential backoff: 2^attempts * base, capped at max
260- const exponentialDelay = Math.min(base * Math.pow(2, this._reconnectAttempts), max);
261-
262- // Add random jitter (0-25% of delay) to prevent thundering herd
263- const jitter = exponentialDelay * Math.random() * 0.25;
264-
265- return Math.floor(exponentialDelay + jitter);
266- }
267-
268- /**
269- * Get the estimated next reconnect delay in seconds
270- * Returns null if not reconnecting
271- */
272- getReconnectDelay(): number | null {
273- if (this._status !== 'reconnecting') {
274- return null;
275- }
276- // Calculate the expected delay for the current attempt
277- const base = SSE_CONFIG.INITIAL_RECONNECT_DELAY;
278- const max = SSE_CONFIG.MAX_RECONNECT_DELAY;
279- const exponentialDelay = Math.min(base * Math.pow(2, this._reconnectAttempts - 1), max);
280- return Math.ceil(exponentialDelay / 1000); // Convert to seconds
281- }
282-
283- /**
284- * Clean up resources
285- */
286- private cleanup(): void {
287- this.clearHeartbeatTimer();
288-
289- if (this.reconnectTimer) {
290- clearTimeout(this.reconnectTimer);
291- this.reconnectTimer = null;
292- }
293-
294- if (this.eventSource) {
295- this.eventSource.close();
296- this.eventSource = null;
297- }
298- }
299-
300- /**
301- * Set connection status and notify listener
302- */
303- private setStatus(status: ConnectionStatus): void {
304- this._status = status;
305- this.config.onStatusChange?.(status);
306- }
307-
308- /**
309- * Destroy the client completely
310- */
311- destroy(): void {
312- console.debug('[SSEClient] Destroying');
313- this.disconnect();
314- }
315-}
316-
317-/**
318- * Factory function to create an SSE client
319- */
320-export function createSSEClient(config: SSEClientConfig): SSEClient {
321- return new SSEClient(config);
322-}
D src/lib/sse/SSEEventEmitter.ts +0 -188
@@ -1,188 +0,0 @@
1-/**
2- * Type-safe SSE Event Emitter
3- *
4- * Provides typed event emission for SSE broadcasts with automatic
5- * listener cleanup and memory leak prevention.
6- */
7-
8-import type { Message, SafeUser, UserStatus } from '../types/chat';
9-
10-/**
11- * SSE Event Type Map - defines the payload for each event type
12- */
13-export interface SSEEventMap {
14- chatMessage: Message;
15- userStatusUpdate: {
16- userId: string;
17- status: UserStatus;
18- lastSeen: number;
19- };
20- buddyListUpdate: SafeUser[];
21- heartbeat: {
22- connectionId: string;
23- timestamp: number;
24- serverTime: string;
25- };
26-}
27-
28-/**
29- * Generic SSE event structure sent over the wire
30- */
31-export interface SSEEvent<T extends keyof SSEEventMap = keyof SSEEventMap> {
32- type: T;
33- data: SSEEventMap[T];
34- id?: string; // Event ID for Last-Event-ID support
35-}
36-
37-/**
38- * Listener function type for SSE events
39- */
40-type SSEListener = (event: SSEEvent) => void;
41-
42-/**
43- * Type-safe SSE Event Emitter with automatic cleanup
44- */
45-class SSEEventEmitter {
46- private listeners: Map<string, Set<SSEListener>> = new Map();
47- private maxListeners: number = 100;
48- private listenerCount: number = 0;
49-
50- constructor() {
51- console.log('[SSE-Emitter] Initialized');
52- }
53-
54- /**
55- * Emit an SSE event to all listeners
56- */
57- emit<T extends keyof SSEEventMap>(type: T, data: SSEEventMap[T], id?: string): void {
58- const event: SSEEvent<T> = { type, data, id };
59-
60- // Get listeners for the 'sse' channel (main broadcast channel)
61- const eventListeners = this.listeners.get('sse');
62- if (eventListeners && eventListeners.size > 0) {
63- // Only log non-frequent event types
64- if (!['userStatusUpdate', 'chatMessage', 'heartbeat'].includes(type)) {
65- console.log('[SSE-Emitter] Broadcasting event:', {
66- type,
67- listenersCount: eventListeners.size
68- });
69- }
70-
71- // Iterate over a copy to allow removal during iteration
72- for (const listener of [...eventListeners]) {
73- try {
74- listener(event as SSEEvent);
75- } catch (error) {
76- console.error('[SSE-Emitter] Error in listener:', error);
77- // Remove broken listener
78- this.removeListener('sse', listener);
79- }
80- }
81- }
82- }
83-
84- /**
85- * Add a listener for SSE events
86- * Returns a cleanup function for easy removal
87- */
88- addListener(channel: string, listener: SSEListener): () => void {
89- if (!this.listeners.has(channel)) {
90- this.listeners.set(channel, new Set());
91- }
92-
93- const channelListeners = this.listeners.get(channel)!;
94-
95- if (channelListeners.size >= this.maxListeners) {
96- console.warn(
97- `[SSE-Emitter] Warning: Channel '${channel}' has reached max listeners (${this.maxListeners})`
98- );
99- }
100-
101- channelListeners.add(listener);
102- this.listenerCount++;
103-
104- console.log('[SSE-Emitter] Listener added:', {
105- channel,
106- totalListeners: channelListeners.size,
107- globalTotal: this.listenerCount
108- });
109-
110- // Return cleanup function
111- return () => this.removeListener(channel, listener);
112- }
113-
114- /**
115- * Remove a specific listener
116- */
117- removeListener(channel: string, listener: SSEListener): boolean {
118- const channelListeners = this.listeners.get(channel);
119- if (channelListeners) {
120- const removed = channelListeners.delete(listener);
121- if (removed) {
122- this.listenerCount--;
123- console.log('[SSE-Emitter] Listener removed:', {
124- channel,
125- remainingListeners: channelListeners.size,
126- globalTotal: this.listenerCount
127- });
128-
129- // Clean up empty channels
130- if (channelListeners.size === 0) {
131- this.listeners.delete(channel);
132- }
133- return true;
134- }
135- }
136- return false;
137- }
138-
139- /**
140- * Remove all listeners for a channel
141- */
142- removeAllListeners(channel?: string): void {
143- if (channel) {
144- const channelListeners = this.listeners.get(channel);
145- if (channelListeners) {
146- this.listenerCount -= channelListeners.size;
147- this.listeners.delete(channel);
148- console.log('[SSE-Emitter] All listeners removed for channel:', channel);
149- }
150- } else {
151- this.listeners.clear();
152- this.listenerCount = 0;
153- console.log('[SSE-Emitter] All listeners removed');
154- }
155- }
156-
157- /**
158- * Get the count of listeners for a channel
159- */
160- getListenerCount(channel?: string): number {
161- if (channel) {
162- return this.listeners.get(channel)?.size ?? 0;
163- }
164- return this.listenerCount;
165- }
166-
167- /**
168- * Set the maximum number of listeners per channel
169- */
170- setMaxListeners(n: number): void {
171- this.maxListeners = n;
172- console.log('[SSE-Emitter] Max listeners updated:', { newLimit: n });
173- }
174-
175- /**
176- * Check if a channel has any listeners
177- */
178- hasListeners(channel: string): boolean {
179- const channelListeners = this.listeners.get(channel);
180- return channelListeners !== undefined && channelListeners.size > 0;
181- }
182-}
183-
184-// Singleton instance
185-export const sseEmitter = new SSEEventEmitter();
186-
187-// Export type for use in other modules
188-export type { SSEListener };
@@ -1,188 +0,0 @@
1-/**
2- * Type-safe SSE Event Emitter
3- *
4- * Provides typed event emission for SSE broadcasts with automatic
5- * listener cleanup and memory leak prevention.
6- */
7-
8-import type { Message, SafeUser, UserStatus } from '../types/chat';
9-
10-/**
11- * SSE Event Type Map - defines the payload for each event type
12- */
13-export interface SSEEventMap {
14- chatMessage: Message;
15- userStatusUpdate: {
16- userId: string;
17- status: UserStatus;
18- lastSeen: number;
19- };
20- buddyListUpdate: SafeUser[];
21- heartbeat: {
22- connectionId: string;
23- timestamp: number;
24- serverTime: string;
25- };
26-}
27-
28-/**
29- * Generic SSE event structure sent over the wire
30- */
31-export interface SSEEvent<T extends keyof SSEEventMap = keyof SSEEventMap> {
32- type: T;
33- data: SSEEventMap[T];
34- id?: string; // Event ID for Last-Event-ID support
35-}
36-
37-/**
38- * Listener function type for SSE events
39- */
40-type SSEListener = (event: SSEEvent) => void;
41-
42-/**
43- * Type-safe SSE Event Emitter with automatic cleanup
44- */
45-class SSEEventEmitter {
46- private listeners: Map<string, Set<SSEListener>> = new Map();
47- private maxListeners: number = 100;
48- private listenerCount: number = 0;
49-
50- constructor() {
51- console.log('[SSE-Emitter] Initialized');
52- }
53-
54- /**
55- * Emit an SSE event to all listeners
56- */
57- emit<T extends keyof SSEEventMap>(type: T, data: SSEEventMap[T], id?: string): void {
58- const event: SSEEvent<T> = { type, data, id };
59-
60- // Get listeners for the 'sse' channel (main broadcast channel)
61- const eventListeners = this.listeners.get('sse');
62- if (eventListeners && eventListeners.size > 0) {
63- // Only log non-frequent event types
64- if (!['userStatusUpdate', 'chatMessage', 'heartbeat'].includes(type)) {
65- console.log('[SSE-Emitter] Broadcasting event:', {
66- type,
67- listenersCount: eventListeners.size
68- });
69- }
70-
71- // Iterate over a copy to allow removal during iteration
72- for (const listener of [...eventListeners]) {
73- try {
74- listener(event as SSEEvent);
75- } catch (error) {
76- console.error('[SSE-Emitter] Error in listener:', error);
77- // Remove broken listener
78- this.removeListener('sse', listener);
79- }
80- }
81- }
82- }
83-
84- /**
85- * Add a listener for SSE events
86- * Returns a cleanup function for easy removal
87- */
88- addListener(channel: string, listener: SSEListener): () => void {
89- if (!this.listeners.has(channel)) {
90- this.listeners.set(channel, new Set());
91- }
92-
93- const channelListeners = this.listeners.get(channel)!;
94-
95- if (channelListeners.size >= this.maxListeners) {
96- console.warn(
97- `[SSE-Emitter] Warning: Channel '${channel}' has reached max listeners (${this.maxListeners})`
98- );
99- }
100-
101- channelListeners.add(listener);
102- this.listenerCount++;
103-
104- console.log('[SSE-Emitter] Listener added:', {
105- channel,
106- totalListeners: channelListeners.size,
107- globalTotal: this.listenerCount
108- });
109-
110- // Return cleanup function
111- return () => this.removeListener(channel, listener);
112- }
113-
114- /**
115- * Remove a specific listener
116- */
117- removeListener(channel: string, listener: SSEListener): boolean {
118- const channelListeners = this.listeners.get(channel);
119- if (channelListeners) {
120- const removed = channelListeners.delete(listener);
121- if (removed) {
122- this.listenerCount--;
123- console.log('[SSE-Emitter] Listener removed:', {
124- channel,
125- remainingListeners: channelListeners.size,
126- globalTotal: this.listenerCount
127- });
128-
129- // Clean up empty channels
130- if (channelListeners.size === 0) {
131- this.listeners.delete(channel);
132- }
133- return true;
134- }
135- }
136- return false;
137- }
138-
139- /**
140- * Remove all listeners for a channel
141- */
142- removeAllListeners(channel?: string): void {
143- if (channel) {
144- const channelListeners = this.listeners.get(channel);
145- if (channelListeners) {
146- this.listenerCount -= channelListeners.size;
147- this.listeners.delete(channel);
148- console.log('[SSE-Emitter] All listeners removed for channel:', channel);
149- }
150- } else {
151- this.listeners.clear();
152- this.listenerCount = 0;
153- console.log('[SSE-Emitter] All listeners removed');
154- }
155- }
156-
157- /**
158- * Get the count of listeners for a channel
159- */
160- getListenerCount(channel?: string): number {
161- if (channel) {
162- return this.listeners.get(channel)?.size ?? 0;
163- }
164- return this.listenerCount;
165- }
166-
167- /**
168- * Set the maximum number of listeners per channel
169- */
170- setMaxListeners(n: number): void {
171- this.maxListeners = n;
172- console.log('[SSE-Emitter] Max listeners updated:', { newLimit: n });
173- }
174-
175- /**
176- * Check if a channel has any listeners
177- */
178- hasListeners(channel: string): boolean {
179- const channelListeners = this.listeners.get(channel);
180- return channelListeners !== undefined && channelListeners.size > 0;
181- }
182-}
183-
184-// Singleton instance
185-export const sseEmitter = new SSEEventEmitter();
186-
187-// Export type for use in other modules
188-export type { SSEListener };
D src/lib/sse/SSEManager.ts +0 -381
@@ -1,381 +0,0 @@
1-/**
2- * SSE Manager - Central server-side manager for SSE connections
3- *
4- * Coordinates connection tracking, message queuing, event broadcasting,
5- * and heartbeat management.
6- */
7-
8-import { SSE_CONFIG } from './config';
9-import { sseEmitter, type SSEEventMap, type SSEEvent } from './SSEEventEmitter';
10-import { messageQueue } from './MessageQueue';
11-import { createLogger } from '../utils/logger.server';
12-
13-const log = createLogger('sse-manager');
14-
15-/**
16- * Enhanced connection info with heartbeat tracking
17- */
18-export interface SSEConnection {
19- /** Unique connection ID */
20- connectionId: string;
21- /** User ID */
22- userId: string;
23- /** When the connection was established */
24- connectedAt: number;
25- /** Last activity timestamp */
26- lastActivity: number;
27- /** Last heartbeat acknowledgment */
28- lastHeartbeat: number;
29- /** Current room ID */
30- roomId: string;
31- /** Stream controller for sending events */
32- controller: ReadableStreamDefaultController<Uint8Array> | null;
33-}
34-
35-/**
36- * SSE Manager singleton
37- */
38-class SSEManager {
39- /** Map of connectionId -> SSEConnection */
40- private connections: Map<string, SSEConnection> = new Map();
41- /** Map of userId -> Set of connectionIds (for multi-tab support) */
42- private userConnections: Map<string, Set<string>> = new Map();
43- /** Text encoder for SSE output */
44- private encoder = new TextEncoder();
45- /** Intervals for periodic tasks */
46- private heartbeatInterval: ReturnType<typeof setInterval> | null = null;
47- private cleanupInterval: ReturnType<typeof setInterval> | null = null;
48-
49- constructor() {
50- log.info('SSE Manager initialized');
51- }
52-
53- /**
54- * Start background tasks (heartbeat, cleanup)
55- */
56- startBackgroundTasks(): void {
57- if (this.heartbeatInterval || this.cleanupInterval) {
58- log.debug('Background tasks already running');
59- return;
60- }
61-
62- // Send heartbeats to all connections
63- this.heartbeatInterval = setInterval(() => {
64- this.sendHeartbeats();
65- }, SSE_CONFIG.HEARTBEAT_INTERVAL);
66-
67- // Cleanup stale connections
68- this.cleanupInterval = setInterval(() => {
69- this.cleanupStaleConnections();
70- }, SSE_CONFIG.STALE_CHECK_INTERVAL);
71-
72- log.info('Background tasks started');
73- }
74-
75- /**
76- * Stop background tasks
77- */
78- stopBackgroundTasks(): void {
79- if (this.heartbeatInterval) {
80- clearInterval(this.heartbeatInterval);
81- this.heartbeatInterval = null;
82- }
83- if (this.cleanupInterval) {
84- clearInterval(this.cleanupInterval);
85- this.cleanupInterval = null;
86- }
87- log.info('Background tasks stopped');
88- }
89-
90- /**
91- * Add a new SSE connection
92- * @returns Connection ID and whether this is a new user connection
93- */
94- addConnection(
95- userId: string,
96- roomId: string,
97- controller: ReadableStreamDefaultController<Uint8Array>
98- ): { connectionId: string; isNewUser: boolean } {
99- const connectionId = crypto.randomUUID();
100- const now = Date.now();
101-
102- const connection: SSEConnection = {
103- connectionId,
104- userId,
105- connectedAt: now,
106- lastActivity: now,
107- lastHeartbeat: now,
108- roomId,
109- controller
110- };
111-
112- this.connections.set(connectionId, connection);
113-
114- // Track user -> connections mapping
115- let userConns = this.userConnections.get(userId);
116- const isNewUser = !userConns || userConns.size === 0;
117-
118- if (!userConns) {
119- userConns = new Set();
120- this.userConnections.set(userId, userConns);
121- }
122- userConns.add(connectionId);
123-
124- log.info('Connection added', {
125- connectionId: connectionId.slice(0, 8),
126- userId: userId.slice(0, 8),
127- isNewUser,
128- userConnectionCount: userConns.size,
129- totalConnections: this.connections.size
130- });
131-
132- return { connectionId, isNewUser };
133- }
134-
135- /**
136- * Remove a connection
137- * @returns Whether this was the last connection for the user
138- */
139- removeConnection(connectionId: string): {
140- removed: boolean;
141- isLastConnection: boolean;
142- userId: string | null;
143- } {
144- const connection = this.connections.get(connectionId);
145- if (!connection) {
146- return { removed: false, isLastConnection: false, userId: null };
147- }
148-
149- const { userId } = connection;
150- this.connections.delete(connectionId);
151-
152- // Update user connections map
153- const userConns = this.userConnections.get(userId);
154- if (userConns) {
155- userConns.delete(connectionId);
156- if (userConns.size === 0) {
157- this.userConnections.delete(userId);
158- }
159- }
160-
161- const isLastConnection = !userConns || userConns.size === 0;
162-
163- log.info('Connection removed', {
164- connectionId: connectionId.slice(0, 8),
165- userId: userId.slice(0, 8),
166- isLastConnection,
167- remainingConnections: this.connections.size
168- });
169-
170- return { removed: true, isLastConnection, userId };
171- }
172-
173- /**
174- * Get a connection by ID
175- */
176- getConnection(connectionId: string): SSEConnection | undefined {
177- return this.connections.get(connectionId);
178- }
179-
180- /**
181- * Check if a user has any active connections
182- */
183- hasUserConnection(userId: string): boolean {
184- const userConns = this.userConnections.get(userId);
185- return userConns !== undefined && userConns.size > 0;
186- }
187-
188- /**
189- * Get all connection IDs for a user
190- */
191- getUserConnectionIds(userId: string): string[] {
192- const userConns = this.userConnections.get(userId);
193- return userConns ? Array.from(userConns) : [];
194- }
195-
196- /**
197- * Update activity timestamp for a connection
198- */
199- updateActivity(connectionId: string): void {
200- const connection = this.connections.get(connectionId);
201- if (connection) {
202- connection.lastActivity = Date.now();
203- }
204- }
205-
206- /**
207- * Acknowledge heartbeat for a connection
208- */
209- acknowledgeHeartbeat(connectionId: string): boolean {
210- const connection = this.connections.get(connectionId);
211- if (connection) {
212- connection.lastHeartbeat = Date.now();
213- connection.lastActivity = Date.now();
214- log.debug('Heartbeat acknowledged', { connectionId: connectionId.slice(0, 8) });
215- return true;
216- }
217- return false;
218- }
219-
220- /**
221- * Send an event to a specific connection
222- */
223- sendToConnection(connectionId: string, event: SSEEvent, eventId?: string): boolean {
224- const connection = this.connections.get(connectionId);
225- if (!connection || !connection.controller) {
226- return false;
227- }
228-
229- try {
230- const id = eventId ?? crypto.randomUUID();
231- const payload = this.formatSSEEvent(event, id);
232- connection.controller.enqueue(this.encoder.encode(payload));
233-
234- // Queue for catch-up
235- messageQueue.enqueue(connection.roomId, event.type, event.data);
236-
237- return true;
238- } catch (error) {
239- log.error('Error sending to connection', { connectionId: connectionId.slice(0, 8), error });
240- return false;
241- }
242- }
243-
244- /**
245- * Send an event to a specific user (all their connections)
246- */
247- sendToUser(userId: string, event: SSEEvent): number {
248- const connectionIds = this.getUserConnectionIds(userId);
249- let sent = 0;
250- for (const connId of connectionIds) {
251- if (this.sendToConnection(connId, event)) {
252- sent++;
253- }
254- }
255- return sent;
256- }
257-
258- /**
259- * Broadcast an event to all connections
260- * Uses the event emitter for distribution
261- */
262- broadcast<T extends keyof SSEEventMap>(type: T, data: SSEEventMap[T], roomId?: string): void {
263- const eventId = crypto.randomUUID();
264-
265- // Queue the message first
266- if (roomId) {
267- messageQueue.enqueue(roomId, type, data);
268- }
269-
270- // Emit through the event emitter (listeners on SSE endpoint will pick this up)
271- sseEmitter.emit(type, data, eventId);
272- }
273-
274- /**
275- * Get messages since a specific event ID (for catch-up)
276- */
277- getMessagesSince(roomId: string, lastEventId: string) {
278- return messageQueue.getMessagesSince(roomId, lastEventId);
279- }
280-
281- /**
282- * Format an SSE event for transmission
283- */
284- formatSSEEvent(event: SSEEvent, eventId: string): string {
285- const lines = [
286- `id: ${eventId}`,
287- `event: ${event.type}`,
288- `data: ${JSON.stringify(event.data)}`,
289- `retry: ${SSE_CONFIG.RETRY_INTERVAL}`,
290- '',
291- ''
292- ];
293- return lines.join('\n');
294- }
295-
296- /**
297- * Format a keep-alive comment
298- */
299- formatKeepAlive(): string {
300- return ':\n\n';
301- }
302-
303- /**
304- * Send heartbeats to all connections
305- */
306- private sendHeartbeats(): void {
307- const now = Date.now();
308- let sent = 0;
309-
310- for (const [connectionId, connection] of this.connections) {
311- if (connection.controller) {
312- try {
313- const heartbeatEvent: SSEEvent<'heartbeat'> = {
314- type: 'heartbeat',
315- data: {
316- connectionId,
317- timestamp: now,
318- serverTime: new Date(now).toISOString()
319- }
320- };
321- const payload = this.formatSSEEvent(heartbeatEvent, crypto.randomUUID());
322- connection.controller.enqueue(this.encoder.encode(payload));
323- sent++;
324- } catch {
325- // Connection might be closed, will be cleaned up later
326- }
327- }
328- }
329-
330- if (sent > 0) {
331- log.debug('Heartbeats sent', { count: sent });
332- }
333- }
334-
335- /**
336- * Clean up stale connections
337- */
338- private cleanupStaleConnections(): void {
339- const now = Date.now();
340- const staleConnections: string[] = [];
341-
342- for (const [connectionId, connection] of this.connections) {
343- const timeSinceHeartbeat = now - connection.lastHeartbeat;
344- if (timeSinceHeartbeat > SSE_CONFIG.HEARTBEAT_TIMEOUT) {
345- staleConnections.push(connectionId);
346- }
347- }
348-
349- if (staleConnections.length > 0) {
350- log.info('Cleaning up stale connections', { count: staleConnections.length });
351- for (const connId of staleConnections) {
352- this.removeConnection(connId);
353- }
354- }
355- }
356-
357- /**
358- * Get all active user IDs
359- */
360- getActiveUserIds(): string[] {
361- return Array.from(this.userConnections.keys());
362- }
363-
364- /**
365- * Get connection statistics
366- */
367- getStats(): {
368- totalConnections: number;
369- uniqueUsers: number;
370- queueStats: { rooms: number; totalMessages: number };
371- } {
372- return {
373- totalConnections: this.connections.size,
374- uniqueUsers: this.userConnections.size,
375- queueStats: messageQueue.getStats()
376- };
377- }
378-}
379-
380-// Singleton instance
381-export const sseManager = new SSEManager();
@@ -1,381 +0,0 @@
1-/**
2- * SSE Manager - Central server-side manager for SSE connections
3- *
4- * Coordinates connection tracking, message queuing, event broadcasting,
5- * and heartbeat management.
6- */
7-
8-import { SSE_CONFIG } from './config';
9-import { sseEmitter, type SSEEventMap, type SSEEvent } from './SSEEventEmitter';
10-import { messageQueue } from './MessageQueue';
11-import { createLogger } from '../utils/logger.server';
12-
13-const log = createLogger('sse-manager');
14-
15-/**
16- * Enhanced connection info with heartbeat tracking
17- */
18-export interface SSEConnection {
19- /** Unique connection ID */
20- connectionId: string;
21- /** User ID */
22- userId: string;
23- /** When the connection was established */
24- connectedAt: number;
25- /** Last activity timestamp */
26- lastActivity: number;
27- /** Last heartbeat acknowledgment */
28- lastHeartbeat: number;
29- /** Current room ID */
30- roomId: string;
31- /** Stream controller for sending events */
32- controller: ReadableStreamDefaultController<Uint8Array> | null;
33-}
34-
35-/**
36- * SSE Manager singleton
37- */
38-class SSEManager {
39- /** Map of connectionId -> SSEConnection */
40- private connections: Map<string, SSEConnection> = new Map();
41- /** Map of userId -> Set of connectionIds (for multi-tab support) */
42- private userConnections: Map<string, Set<string>> = new Map();
43- /** Text encoder for SSE output */
44- private encoder = new TextEncoder();
45- /** Intervals for periodic tasks */
46- private heartbeatInterval: ReturnType<typeof setInterval> | null = null;
47- private cleanupInterval: ReturnType<typeof setInterval> | null = null;
48-
49- constructor() {
50- log.info('SSE Manager initialized');
51- }
52-
53- /**
54- * Start background tasks (heartbeat, cleanup)
55- */
56- startBackgroundTasks(): void {
57- if (this.heartbeatInterval || this.cleanupInterval) {
58- log.debug('Background tasks already running');
59- return;
60- }
61-
62- // Send heartbeats to all connections
63- this.heartbeatInterval = setInterval(() => {
64- this.sendHeartbeats();
65- }, SSE_CONFIG.HEARTBEAT_INTERVAL);
66-
67- // Cleanup stale connections
68- this.cleanupInterval = setInterval(() => {
69- this.cleanupStaleConnections();
70- }, SSE_CONFIG.STALE_CHECK_INTERVAL);
71-
72- log.info('Background tasks started');
73- }
74-
75- /**
76- * Stop background tasks
77- */
78- stopBackgroundTasks(): void {
79- if (this.heartbeatInterval) {
80- clearInterval(this.heartbeatInterval);
81- this.heartbeatInterval = null;
82- }
83- if (this.cleanupInterval) {
84- clearInterval(this.cleanupInterval);
85- this.cleanupInterval = null;
86- }
87- log.info('Background tasks stopped');
88- }
89-
90- /**
91- * Add a new SSE connection
92- * @returns Connection ID and whether this is a new user connection
93- */
94- addConnection(
95- userId: string,
96- roomId: string,
97- controller: ReadableStreamDefaultController<Uint8Array>
98- ): { connectionId: string; isNewUser: boolean } {
99- const connectionId = crypto.randomUUID();
100- const now = Date.now();
101-
102- const connection: SSEConnection = {
103- connectionId,
104- userId,
105- connectedAt: now,
106- lastActivity: now,
107- lastHeartbeat: now,
108- roomId,
109- controller
110- };
111-
112- this.connections.set(connectionId, connection);
113-
114- // Track user -> connections mapping
115- let userConns = this.userConnections.get(userId);
116- const isNewUser = !userConns || userConns.size === 0;
117-
118- if (!userConns) {
119- userConns = new Set();
120- this.userConnections.set(userId, userConns);
121- }
122- userConns.add(connectionId);
123-
124- log.info('Connection added', {
125- connectionId: connectionId.slice(0, 8),
126- userId: userId.slice(0, 8),
127- isNewUser,
128- userConnectionCount: userConns.size,
129- totalConnections: this.connections.size
130- });
131-
132- return { connectionId, isNewUser };
133- }
134-
135- /**
136- * Remove a connection
137- * @returns Whether this was the last connection for the user
138- */
139- removeConnection(connectionId: string): {
140- removed: boolean;
141- isLastConnection: boolean;
142- userId: string | null;
143- } {
144- const connection = this.connections.get(connectionId);
145- if (!connection) {
146- return { removed: false, isLastConnection: false, userId: null };
147- }
148-
149- const { userId } = connection;
150- this.connections.delete(connectionId);
151-
152- // Update user connections map
153- const userConns = this.userConnections.get(userId);
154- if (userConns) {
155- userConns.delete(connectionId);
156- if (userConns.size === 0) {
157- this.userConnections.delete(userId);
158- }
159- }
160-
161- const isLastConnection = !userConns || userConns.size === 0;
162-
163- log.info('Connection removed', {
164- connectionId: connectionId.slice(0, 8),
165- userId: userId.slice(0, 8),
166- isLastConnection,
167- remainingConnections: this.connections.size
168- });
169-
170- return { removed: true, isLastConnection, userId };
171- }
172-
173- /**
174- * Get a connection by ID
175- */
176- getConnection(connectionId: string): SSEConnection | undefined {
177- return this.connections.get(connectionId);
178- }
179-
180- /**
181- * Check if a user has any active connections
182- */
183- hasUserConnection(userId: string): boolean {
184- const userConns = this.userConnections.get(userId);
185- return userConns !== undefined && userConns.size > 0;
186- }
187-
188- /**
189- * Get all connection IDs for a user
190- */
191- getUserConnectionIds(userId: string): string[] {
192- const userConns = this.userConnections.get(userId);
193- return userConns ? Array.from(userConns) : [];
194- }
195-
196- /**
197- * Update activity timestamp for a connection
198- */
199- updateActivity(connectionId: string): void {
200- const connection = this.connections.get(connectionId);
201- if (connection) {
202- connection.lastActivity = Date.now();
203- }
204- }
205-
206- /**
207- * Acknowledge heartbeat for a connection
208- */
209- acknowledgeHeartbeat(connectionId: string): boolean {
210- const connection = this.connections.get(connectionId);
211- if (connection) {
212- connection.lastHeartbeat = Date.now();
213- connection.lastActivity = Date.now();
214- log.debug('Heartbeat acknowledged', { connectionId: connectionId.slice(0, 8) });
215- return true;
216- }
217- return false;
218- }
219-
220- /**
221- * Send an event to a specific connection
222- */
223- sendToConnection(connectionId: string, event: SSEEvent, eventId?: string): boolean {
224- const connection = this.connections.get(connectionId);
225- if (!connection || !connection.controller) {
226- return false;
227- }
228-
229- try {
230- const id = eventId ?? crypto.randomUUID();
231- const payload = this.formatSSEEvent(event, id);
232- connection.controller.enqueue(this.encoder.encode(payload));
233-
234- // Queue for catch-up
235- messageQueue.enqueue(connection.roomId, event.type, event.data);
236-
237- return true;
238- } catch (error) {
239- log.error('Error sending to connection', { connectionId: connectionId.slice(0, 8), error });
240- return false;
241- }
242- }
243-
244- /**
245- * Send an event to a specific user (all their connections)
246- */
247- sendToUser(userId: string, event: SSEEvent): number {
248- const connectionIds = this.getUserConnectionIds(userId);
249- let sent = 0;
250- for (const connId of connectionIds) {
251- if (this.sendToConnection(connId, event)) {
252- sent++;
253- }
254- }
255- return sent;
256- }
257-
258- /**
259- * Broadcast an event to all connections
260- * Uses the event emitter for distribution
261- */
262- broadcast<T extends keyof SSEEventMap>(type: T, data: SSEEventMap[T], roomId?: string): void {
263- const eventId = crypto.randomUUID();
264-
265- // Queue the message first
266- if (roomId) {
267- messageQueue.enqueue(roomId, type, data);
268- }
269-
270- // Emit through the event emitter (listeners on SSE endpoint will pick this up)
271- sseEmitter.emit(type, data, eventId);
272- }
273-
274- /**
275- * Get messages since a specific event ID (for catch-up)
276- */
277- getMessagesSince(roomId: string, lastEventId: string) {
278- return messageQueue.getMessagesSince(roomId, lastEventId);
279- }
280-
281- /**
282- * Format an SSE event for transmission
283- */
284- formatSSEEvent(event: SSEEvent, eventId: string): string {
285- const lines = [
286- `id: ${eventId}`,
287- `event: ${event.type}`,
288- `data: ${JSON.stringify(event.data)}`,
289- `retry: ${SSE_CONFIG.RETRY_INTERVAL}`,
290- '',
291- ''
292- ];
293- return lines.join('\n');
294- }
295-
296- /**
297- * Format a keep-alive comment
298- */
299- formatKeepAlive(): string {
300- return ':\n\n';
301- }
302-
303- /**
304- * Send heartbeats to all connections
305- */
306- private sendHeartbeats(): void {
307- const now = Date.now();
308- let sent = 0;
309-
310- for (const [connectionId, connection] of this.connections) {
311- if (connection.controller) {
312- try {
313- const heartbeatEvent: SSEEvent<'heartbeat'> = {
314- type: 'heartbeat',
315- data: {
316- connectionId,
317- timestamp: now,
318- serverTime: new Date(now).toISOString()
319- }
320- };
321- const payload = this.formatSSEEvent(heartbeatEvent, crypto.randomUUID());
322- connection.controller.enqueue(this.encoder.encode(payload));
323- sent++;
324- } catch {
325- // Connection might be closed, will be cleaned up later
326- }
327- }
328- }
329-
330- if (sent > 0) {
331- log.debug('Heartbeats sent', { count: sent });
332- }
333- }
334-
335- /**
336- * Clean up stale connections
337- */
338- private cleanupStaleConnections(): void {
339- const now = Date.now();
340- const staleConnections: string[] = [];
341-
342- for (const [connectionId, connection] of this.connections) {
343- const timeSinceHeartbeat = now - connection.lastHeartbeat;
344- if (timeSinceHeartbeat > SSE_CONFIG.HEARTBEAT_TIMEOUT) {
345- staleConnections.push(connectionId);
346- }
347- }
348-
349- if (staleConnections.length > 0) {
350- log.info('Cleaning up stale connections', { count: staleConnections.length });
351- for (const connId of staleConnections) {
352- this.removeConnection(connId);
353- }
354- }
355- }
356-
357- /**
358- * Get all active user IDs
359- */
360- getActiveUserIds(): string[] {
361- return Array.from(this.userConnections.keys());
362- }
363-
364- /**
365- * Get connection statistics
366- */
367- getStats(): {
368- totalConnections: number;
369- uniqueUsers: number;
370- queueStats: { rooms: number; totalMessages: number };
371- } {
372- return {
373- totalConnections: this.connections.size,
374- uniqueUsers: this.userConnections.size,
375- queueStats: messageQueue.getStats()
376- };
377- }
378-}
379-
380-// Singleton instance
381-export const sseManager = new SSEManager();
D src/lib/sse/__tests__/MessageQueue.test.ts +0 -156
@@ -1,156 +0,0 @@
1-/**
2- * Unit tests for MessageQueue
3- */
4-
5-import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest';
6-import { MessageQueue } from '../MessageQueue';
7-
8-describe('MessageQueue', () => {
9- let queue: MessageQueue;
10-
11- beforeEach(() => {
12- // Create a queue with short TTL for testing
13- queue = new MessageQueue(1000, 10); // 1 second TTL, max 10 messages
14- });
15-
16- afterEach(() => {
17- queue.stopCleanup();
18- queue.clear();
19- });
20-
21- describe('enqueue', () => {
22- it('should add a message to the queue and return an ID', () => {
23- const id = queue.enqueue('room1', 'chatMessage', { text: 'hello' });
24-
25- expect(id).toBeDefined();
26- expect(typeof id).toBe('string');
27- expect(id.length).toBeGreaterThan(0);
28- });
29-
30- it('should store messages by room ID', () => {
31- queue.enqueue('room1', 'chatMessage', { text: 'hello' });
32- queue.enqueue('room2', 'chatMessage', { text: 'world' });
33-
34- const room1Messages = queue.getMessages('room1');
35- const room2Messages = queue.getMessages('room2');
36-
37- expect(room1Messages).toHaveLength(1);
38- expect(room2Messages).toHaveLength(1);
39- expect(room1Messages[0].data).toEqual({ text: 'hello' });
40- expect(room2Messages[0].data).toEqual({ text: 'world' });
41- });
42-
43- it('should enforce max queue size', () => {
44- // Add more than max (10) messages
45- for (let i = 0; i < 15; i++) {
46- queue.enqueue('room1', 'chatMessage', { index: i });
47- }
48-
49- const messages = queue.getMessages('room1');
50- expect(messages).toHaveLength(10);
51-
52- // Should keep the newest messages
53- expect(messages[0].data).toEqual({ index: 5 });
54- expect(messages[9].data).toEqual({ index: 14 });
55- });
56- });
57-
58- describe('getMessagesSince', () => {
59- it('should return messages after a given ID', () => {
60- const id1 = queue.enqueue('room1', 'chatMessage', { index: 1 });
61- const id2 = queue.enqueue('room1', 'chatMessage', { index: 2 });
62- queue.enqueue('room1', 'chatMessage', { index: 3 });
63-
64- const messagesSinceId1 = queue.getMessagesSince('room1', id1);
65- expect(messagesSinceId1).toHaveLength(2);
66- expect(messagesSinceId1[0].data).toEqual({ index: 2 });
67- expect(messagesSinceId1[1].data).toEqual({ index: 3 });
68-
69- const messagesSinceId2 = queue.getMessagesSince('room1', id2);
70- expect(messagesSinceId2).toHaveLength(1);
71- expect(messagesSinceId2[0].data).toEqual({ index: 3 });
72- });
73-
74- it('should return recent messages if ID not found', () => {
75- for (let i = 0; i < 5; i++) {
76- queue.enqueue('room1', 'chatMessage', { index: i });
77- }
78-
79- const messages = queue.getMessagesSince('room1', 'unknown-id');
80- expect(messages).toHaveLength(5);
81- });
82-
83- it('should return empty array for non-existent room', () => {
84- const messages = queue.getMessagesSince('nonexistent', 'some-id');
85- expect(messages).toEqual([]);
86- });
87- });
88-
89- describe('getLatestEventId', () => {
90- it('should return the ID of the last message', () => {
91- queue.enqueue('room1', 'chatMessage', { index: 1 });
92- const id2 = queue.enqueue('room1', 'chatMessage', { index: 2 });
93-
94- expect(queue.getLatestEventId('room1')).toBe(id2);
95- });
96-
97- it('should return null for empty room', () => {
98- expect(queue.getLatestEventId('nonexistent')).toBeNull();
99- });
100- });
101-
102- describe('hasEventId', () => {
103- it('should return true if event ID exists', () => {
104- const id = queue.enqueue('room1', 'chatMessage', { text: 'hello' });
105- expect(queue.hasEventId('room1', id)).toBe(true);
106- });
107-
108- it('should return false if event ID does not exist', () => {
109- expect(queue.hasEventId('room1', 'unknown')).toBe(false);
110- });
111- });
112-
113- describe('cleanup', () => {
114- it('should remove expired messages', async () => {
115- vi.useFakeTimers();
116-
117- queue.enqueue('room1', 'chatMessage', { text: 'old' });
118-
119- // Fast-forward past TTL
120- vi.advanceTimersByTime(2000);
121-
122- const removed = queue.cleanup();
123- expect(removed).toBe(1);
124- expect(queue.getMessages('room1')).toHaveLength(0);
125-
126- vi.useRealTimers();
127- });
128-
129- it('should keep non-expired messages', async () => {
130- vi.useFakeTimers();
131-
132- queue.enqueue('room1', 'chatMessage', { text: 'message' });
133-
134- // Fast-forward but not past TTL
135- vi.advanceTimersByTime(500);
136-
137- const removed = queue.cleanup();
138- expect(removed).toBe(0);
139- expect(queue.getMessages('room1')).toHaveLength(1);
140-
141- vi.useRealTimers();
142- });
143- });
144-
145- describe('getStats', () => {
146- it('should return correct statistics', () => {
147- queue.enqueue('room1', 'chatMessage', { text: '1' });
148- queue.enqueue('room1', 'chatMessage', { text: '2' });
149- queue.enqueue('room2', 'chatMessage', { text: '3' });
150-
151- const stats = queue.getStats();
152- expect(stats.rooms).toBe(2);
153- expect(stats.totalMessages).toBe(3);
154- });
155- });
156-});
@@ -1,156 +0,0 @@
1-/**
2- * Unit tests for MessageQueue
3- */
4-
5-import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest';
6-import { MessageQueue } from '../MessageQueue';
7-
8-describe('MessageQueue', () => {
9- let queue: MessageQueue;
10-
11- beforeEach(() => {
12- // Create a queue with short TTL for testing
13- queue = new MessageQueue(1000, 10); // 1 second TTL, max 10 messages
14- });
15-
16- afterEach(() => {
17- queue.stopCleanup();
18- queue.clear();
19- });
20-
21- describe('enqueue', () => {
22- it('should add a message to the queue and return an ID', () => {
23- const id = queue.enqueue('room1', 'chatMessage', { text: 'hello' });
24-
25- expect(id).toBeDefined();
26- expect(typeof id).toBe('string');
27- expect(id.length).toBeGreaterThan(0);
28- });
29-
30- it('should store messages by room ID', () => {
31- queue.enqueue('room1', 'chatMessage', { text: 'hello' });
32- queue.enqueue('room2', 'chatMessage', { text: 'world' });
33-
34- const room1Messages = queue.getMessages('room1');
35- const room2Messages = queue.getMessages('room2');
36-
37- expect(room1Messages).toHaveLength(1);
38- expect(room2Messages).toHaveLength(1);
39- expect(room1Messages[0].data).toEqual({ text: 'hello' });
40- expect(room2Messages[0].data).toEqual({ text: 'world' });
41- });
42-
43- it('should enforce max queue size', () => {
44- // Add more than max (10) messages
45- for (let i = 0; i < 15; i++) {
46- queue.enqueue('room1', 'chatMessage', { index: i });
47- }
48-
49- const messages = queue.getMessages('room1');
50- expect(messages).toHaveLength(10);
51-
52- // Should keep the newest messages
53- expect(messages[0].data).toEqual({ index: 5 });
54- expect(messages[9].data).toEqual({ index: 14 });
55- });
56- });
57-
58- describe('getMessagesSince', () => {
59- it('should return messages after a given ID', () => {
60- const id1 = queue.enqueue('room1', 'chatMessage', { index: 1 });
61- const id2 = queue.enqueue('room1', 'chatMessage', { index: 2 });
62- queue.enqueue('room1', 'chatMessage', { index: 3 });
63-
64- const messagesSinceId1 = queue.getMessagesSince('room1', id1);
65- expect(messagesSinceId1).toHaveLength(2);
66- expect(messagesSinceId1[0].data).toEqual({ index: 2 });
67- expect(messagesSinceId1[1].data).toEqual({ index: 3 });
68-
69- const messagesSinceId2 = queue.getMessagesSince('room1', id2);
70- expect(messagesSinceId2).toHaveLength(1);
71- expect(messagesSinceId2[0].data).toEqual({ index: 3 });
72- });
73-
74- it('should return recent messages if ID not found', () => {
75- for (let i = 0; i < 5; i++) {
76- queue.enqueue('room1', 'chatMessage', { index: i });
77- }
78-
79- const messages = queue.getMessagesSince('room1', 'unknown-id');
80- expect(messages).toHaveLength(5);
81- });
82-
83- it('should return empty array for non-existent room', () => {
84- const messages = queue.getMessagesSince('nonexistent', 'some-id');
85- expect(messages).toEqual([]);
86- });
87- });
88-
89- describe('getLatestEventId', () => {
90- it('should return the ID of the last message', () => {
91- queue.enqueue('room1', 'chatMessage', { index: 1 });
92- const id2 = queue.enqueue('room1', 'chatMessage', { index: 2 });
93-
94- expect(queue.getLatestEventId('room1')).toBe(id2);
95- });
96-
97- it('should return null for empty room', () => {
98- expect(queue.getLatestEventId('nonexistent')).toBeNull();
99- });
100- });
101-
102- describe('hasEventId', () => {
103- it('should return true if event ID exists', () => {
104- const id = queue.enqueue('room1', 'chatMessage', { text: 'hello' });
105- expect(queue.hasEventId('room1', id)).toBe(true);
106- });
107-
108- it('should return false if event ID does not exist', () => {
109- expect(queue.hasEventId('room1', 'unknown')).toBe(false);
110- });
111- });
112-
113- describe('cleanup', () => {
114- it('should remove expired messages', async () => {
115- vi.useFakeTimers();
116-
117- queue.enqueue('room1', 'chatMessage', { text: 'old' });
118-
119- // Fast-forward past TTL
120- vi.advanceTimersByTime(2000);
121-
122- const removed = queue.cleanup();
123- expect(removed).toBe(1);
124- expect(queue.getMessages('room1')).toHaveLength(0);
125-
126- vi.useRealTimers();
127- });
128-
129- it('should keep non-expired messages', async () => {
130- vi.useFakeTimers();
131-
132- queue.enqueue('room1', 'chatMessage', { text: 'message' });
133-
134- // Fast-forward but not past TTL
135- vi.advanceTimersByTime(500);
136-
137- const removed = queue.cleanup();
138- expect(removed).toBe(0);
139- expect(queue.getMessages('room1')).toHaveLength(1);
140-
141- vi.useRealTimers();
142- });
143- });
144-
145- describe('getStats', () => {
146- it('should return correct statistics', () => {
147- queue.enqueue('room1', 'chatMessage', { text: '1' });
148- queue.enqueue('room1', 'chatMessage', { text: '2' });
149- queue.enqueue('room2', 'chatMessage', { text: '3' });
150-
151- const stats = queue.getStats();
152- expect(stats.rooms).toBe(2);
153- expect(stats.totalMessages).toBe(3);
154- });
155- });
156-});
D src/lib/sse/__tests__/SSEEventEmitter.test.ts +0 -187
@@ -1,187 +0,0 @@
1-/**
2- * Unit tests for SSEEventEmitter
3- */
4-
5-import { describe, it, expect, beforeEach, vi } from 'vitest';
6-
7-// We need to import the class, not the singleton for testing
8-// First, let's create a test-friendly version
9-
10-describe('SSEEventEmitter', () => {
11- // Import the module fresh for each test
12- let SSEEventEmitter: typeof import('../SSEEventEmitter');
13-
14- beforeEach(async () => {
15- // Clear module cache and re-import
16- vi.resetModules();
17- SSEEventEmitter = await import('../SSEEventEmitter');
18- });
19-
20- describe('emit', () => {
21- it('should broadcast events to all listeners on the sse channel', () => {
22- const { sseEmitter } = SSEEventEmitter;
23- const listener = vi.fn();
24-
25- sseEmitter.addListener('sse', listener);
26-
27- sseEmitter.emit('chatMessage', {
28- id: '123',
29- chatRoomId: 'room1',
30- senderId: 'user1',
31- content: 'Hello',
32- type: 'chat',
33- timestamp: Date.now()
34- });
35-
36- expect(listener).toHaveBeenCalledTimes(1);
37- expect(listener).toHaveBeenCalledWith(
38- expect.objectContaining({
39- type: 'chatMessage',
40- data: expect.objectContaining({
41- id: '123',
42- content: 'Hello'
43- })
44- })
45- );
46- });
47-
48- it('should include event ID when provided', () => {
49- const { sseEmitter } = SSEEventEmitter;
50- const listener = vi.fn();
51-
52- sseEmitter.addListener('sse', listener);
53-
54- sseEmitter.emit(
55- 'chatMessage',
56- {
57- id: '123',
58- chatRoomId: 'room1',
59- senderId: 'user1',
60- content: 'Hello',
61- type: 'chat',
62- timestamp: Date.now()
63- },
64- 'event-id-456'
65- );
66-
67- expect(listener).toHaveBeenCalledWith(
68- expect.objectContaining({
69- id: 'event-id-456'
70- })
71- );
72- });
73- });
74-
75- describe('addListener', () => {
76- it('should return a cleanup function', () => {
77- const { sseEmitter } = SSEEventEmitter;
78- const listener = vi.fn();
79-
80- const cleanup = sseEmitter.addListener('sse', listener);
81-
82- expect(typeof cleanup).toBe('function');
83- expect(sseEmitter.getListenerCount('sse')).toBe(1);
84-
85- cleanup();
86- expect(sseEmitter.getListenerCount('sse')).toBe(0);
87- });
88-
89- it('should support multiple listeners', () => {
90- const { sseEmitter } = SSEEventEmitter;
91- const listener1 = vi.fn();
92- const listener2 = vi.fn();
93-
94- sseEmitter.addListener('sse', listener1);
95- sseEmitter.addListener('sse', listener2);
96-
97- sseEmitter.emit('buddyListUpdate', []);
98-
99- expect(listener1).toHaveBeenCalledTimes(1);
100- expect(listener2).toHaveBeenCalledTimes(1);
101- });
102- });
103-
104- describe('removeListener', () => {
105- it('should remove a specific listener', () => {
106- const { sseEmitter } = SSEEventEmitter;
107- const listener = vi.fn();
108-
109- sseEmitter.addListener('sse', listener);
110- expect(sseEmitter.getListenerCount('sse')).toBe(1);
111-
112- const removed = sseEmitter.removeListener('sse', listener);
113- expect(removed).toBe(true);
114- expect(sseEmitter.getListenerCount('sse')).toBe(0);
115- });
116-
117- it('should return false for non-existent listener', () => {
118- const { sseEmitter } = SSEEventEmitter;
119- const listener = vi.fn();
120-
121- const removed = sseEmitter.removeListener('sse', listener);
122- expect(removed).toBe(false);
123- });
124- });
125-
126- describe('removeAllListeners', () => {
127- it('should remove all listeners for a channel', () => {
128- const { sseEmitter } = SSEEventEmitter;
129- const listener1 = vi.fn();
130- const listener2 = vi.fn();
131-
132- sseEmitter.addListener('sse', listener1);
133- sseEmitter.addListener('sse', listener2);
134- expect(sseEmitter.getListenerCount('sse')).toBe(2);
135-
136- sseEmitter.removeAllListeners('sse');
137- expect(sseEmitter.getListenerCount('sse')).toBe(0);
138- });
139-
140- it('should remove all listeners when no channel specified', () => {
141- const { sseEmitter } = SSEEventEmitter;
142- const listener = vi.fn();
143-
144- sseEmitter.addListener('sse', listener);
145- sseEmitter.addListener('other', listener);
146-
147- sseEmitter.removeAllListeners();
148- expect(sseEmitter.getListenerCount()).toBe(0);
149- });
150- });
151-
152- describe('hasListeners', () => {
153- it('should return true when channel has listeners', () => {
154- const { sseEmitter } = SSEEventEmitter;
155- const listener = vi.fn();
156-
157- sseEmitter.addListener('sse', listener);
158- expect(sseEmitter.hasListeners('sse')).toBe(true);
159- });
160-
161- it('should return false when channel has no listeners', () => {
162- const { sseEmitter } = SSEEventEmitter;
163- expect(sseEmitter.hasListeners('sse')).toBe(false);
164- });
165- });
166-
167- describe('error handling', () => {
168- it('should continue broadcasting even if a listener throws', () => {
169- const { sseEmitter } = SSEEventEmitter;
170- const badListener = vi.fn(() => {
171- throw new Error('Listener error');
172- });
173- const goodListener = vi.fn();
174-
175- sseEmitter.addListener('sse', badListener);
176- sseEmitter.addListener('sse', goodListener);
177-
178- // Should not throw
179- expect(() => {
180- sseEmitter.emit('buddyListUpdate', []);
181- }).not.toThrow();
182-
183- // Good listener should still be called
184- expect(goodListener).toHaveBeenCalledTimes(1);
185- });
186- });
187-});
@@ -1,187 +0,0 @@
1-/**
2- * Unit tests for SSEEventEmitter
3- */
4-
5-import { describe, it, expect, beforeEach, vi } from 'vitest';
6-
7-// We need to import the class, not the singleton for testing
8-// First, let's create a test-friendly version
9-
10-describe('SSEEventEmitter', () => {
11- // Import the module fresh for each test
12- let SSEEventEmitter: typeof import('../SSEEventEmitter');
13-
14- beforeEach(async () => {
15- // Clear module cache and re-import
16- vi.resetModules();
17- SSEEventEmitter = await import('../SSEEventEmitter');
18- });
19-
20- describe('emit', () => {
21- it('should broadcast events to all listeners on the sse channel', () => {
22- const { sseEmitter } = SSEEventEmitter;
23- const listener = vi.fn();
24-
25- sseEmitter.addListener('sse', listener);
26-
27- sseEmitter.emit('chatMessage', {
28- id: '123',
29- chatRoomId: 'room1',
30- senderId: 'user1',
31- content: 'Hello',
32- type: 'chat',
33- timestamp: Date.now()
34- });
35-
36- expect(listener).toHaveBeenCalledTimes(1);
37- expect(listener).toHaveBeenCalledWith(
38- expect.objectContaining({
39- type: 'chatMessage',
40- data: expect.objectContaining({
41- id: '123',
42- content: 'Hello'
43- })
44- })
45- );
46- });
47-
48- it('should include event ID when provided', () => {
49- const { sseEmitter } = SSEEventEmitter;
50- const listener = vi.fn();
51-
52- sseEmitter.addListener('sse', listener);
53-
54- sseEmitter.emit(
55- 'chatMessage',
56- {
57- id: '123',
58- chatRoomId: 'room1',
59- senderId: 'user1',
60- content: 'Hello',
61- type: 'chat',
62- timestamp: Date.now()
63- },
64- 'event-id-456'
65- );
66-
67- expect(listener).toHaveBeenCalledWith(
68- expect.objectContaining({
69- id: 'event-id-456'
70- })
71- );
72- });
73- });
74-
75- describe('addListener', () => {
76- it('should return a cleanup function', () => {
77- const { sseEmitter } = SSEEventEmitter;
78- const listener = vi.fn();
79-
80- const cleanup = sseEmitter.addListener('sse', listener);
81-
82- expect(typeof cleanup).toBe('function');
83- expect(sseEmitter.getListenerCount('sse')).toBe(1);
84-
85- cleanup();
86- expect(sseEmitter.getListenerCount('sse')).toBe(0);
87- });
88-
89- it('should support multiple listeners', () => {
90- const { sseEmitter } = SSEEventEmitter;
91- const listener1 = vi.fn();
92- const listener2 = vi.fn();
93-
94- sseEmitter.addListener('sse', listener1);
95- sseEmitter.addListener('sse', listener2);
96-
97- sseEmitter.emit('buddyListUpdate', []);
98-
99- expect(listener1).toHaveBeenCalledTimes(1);
100- expect(listener2).toHaveBeenCalledTimes(1);
101- });
102- });
103-
104- describe('removeListener', () => {
105- it('should remove a specific listener', () => {
106- const { sseEmitter } = SSEEventEmitter;
107- const listener = vi.fn();
108-
109- sseEmitter.addListener('sse', listener);
110- expect(sseEmitter.getListenerCount('sse')).toBe(1);
111-
112- const removed = sseEmitter.removeListener('sse', listener);
113- expect(removed).toBe(true);
114- expect(sseEmitter.getListenerCount('sse')).toBe(0);
115- });
116-
117- it('should return false for non-existent listener', () => {
118- const { sseEmitter } = SSEEventEmitter;
119- const listener = vi.fn();
120-
121- const removed = sseEmitter.removeListener('sse', listener);
122- expect(removed).toBe(false);
123- });
124- });
125-
126- describe('removeAllListeners', () => {
127- it('should remove all listeners for a channel', () => {
128- const { sseEmitter } = SSEEventEmitter;
129- const listener1 = vi.fn();
130- const listener2 = vi.fn();
131-
132- sseEmitter.addListener('sse', listener1);
133- sseEmitter.addListener('sse', listener2);
134- expect(sseEmitter.getListenerCount('sse')).toBe(2);
135-
136- sseEmitter.removeAllListeners('sse');
137- expect(sseEmitter.getListenerCount('sse')).toBe(0);
138- });
139-
140- it('should remove all listeners when no channel specified', () => {
141- const { sseEmitter } = SSEEventEmitter;
142- const listener = vi.fn();
143-
144- sseEmitter.addListener('sse', listener);
145- sseEmitter.addListener('other', listener);
146-
147- sseEmitter.removeAllListeners();
148- expect(sseEmitter.getListenerCount()).toBe(0);
149- });
150- });
151-
152- describe('hasListeners', () => {
153- it('should return true when channel has listeners', () => {
154- const { sseEmitter } = SSEEventEmitter;
155- const listener = vi.fn();
156-
157- sseEmitter.addListener('sse', listener);
158- expect(sseEmitter.hasListeners('sse')).toBe(true);
159- });
160-
161- it('should return false when channel has no listeners', () => {
162- const { sseEmitter } = SSEEventEmitter;
163- expect(sseEmitter.hasListeners('sse')).toBe(false);
164- });
165- });
166-
167- describe('error handling', () => {
168- it('should continue broadcasting even if a listener throws', () => {
169- const { sseEmitter } = SSEEventEmitter;
170- const badListener = vi.fn(() => {
171- throw new Error('Listener error');
172- });
173- const goodListener = vi.fn();
174-
175- sseEmitter.addListener('sse', badListener);
176- sseEmitter.addListener('sse', goodListener);
177-
178- // Should not throw
179- expect(() => {
180- sseEmitter.emit('buddyListUpdate', []);
181- }).not.toThrow();
182-
183- // Good listener should still be called
184- expect(goodListener).toHaveBeenCalledTimes(1);
185- });
186- });
187-});
D src/lib/sse/config.ts +0 -53
@@ -1,53 +0,0 @@
1-/**
2- * Centralized SSE Configuration Constants
3- *
4- * All timing constants for SSE connection management are defined here
5- * to ensure consistency between server and client implementations.
6- */
7-
8-export const SSE_CONFIG = {
9- // Server-side timing
10- /** Heartbeat interval - server sends ping every 30 seconds */
11- HEARTBEAT_INTERVAL: 30_000,
12-
13- /** Connection timeout - mark user offline after 5 minutes of inactivity */
14- CONNECTION_TIMEOUT: 5 * 60_000,
15-
16- /** Stale connection check interval - cleanup every 60 seconds */
17- STALE_CHECK_INTERVAL: 60_000,
18-
19- /** Message queue TTL - keep messages for 5 minutes for catch-up */
20- MESSAGE_QUEUE_TTL: 5 * 60_000,
21-
22- /** Maximum messages to queue per room */
23- MESSAGE_QUEUE_MAX_SIZE: 1000,
24-
25- /** Buddy list broadcast interval */
26- BUDDY_LIST_BROADCAST_INTERVAL: 10_000,
27-
28- // Client-side timing
29- /** Initial reconnect delay - 1 second */
30- INITIAL_RECONNECT_DELAY: 1_000,
31-
32- /** Maximum reconnect delay - 60 seconds cap */
33- MAX_RECONNECT_DELAY: 60_000,
34-
35- /** Heartbeat timeout - 45 seconds (1.5x server interval) */
36- HEARTBEAT_TIMEOUT: 45_000,
37-
38- // Shared timing
39- /** Idle timeout - mark as idle after 2 minutes */
40- IDLE_TIMEOUT: 2 * 60_000,
41-
42- /** Status update interval */
43- STATUS_UPDATE_INTERVAL: 30_000,
44-
45- /** EventSource retry hint (ms) - sent to client */
46- RETRY_INTERVAL: 3_000
47-} as const;
48-
49-/** SSE Event Types */
50-export type SSEEventType = 'chatMessage' | 'userStatusUpdate' | 'buddyListUpdate' | 'heartbeat';
51-
52-/** Connection status */
53-export type ConnectionStatus = 'connecting' | 'connected' | 'reconnecting' | 'disconnected';
@@ -1,53 +0,0 @@
1-/**
2- * Centralized SSE Configuration Constants
3- *
4- * All timing constants for SSE connection management are defined here
5- * to ensure consistency between server and client implementations.
6- */
7-
8-export const SSE_CONFIG = {
9- // Server-side timing
10- /** Heartbeat interval - server sends ping every 30 seconds */
11- HEARTBEAT_INTERVAL: 30_000,
12-
13- /** Connection timeout - mark user offline after 5 minutes of inactivity */
14- CONNECTION_TIMEOUT: 5 * 60_000,
15-
16- /** Stale connection check interval - cleanup every 60 seconds */
17- STALE_CHECK_INTERVAL: 60_000,
18-
19- /** Message queue TTL - keep messages for 5 minutes for catch-up */
20- MESSAGE_QUEUE_TTL: 5 * 60_000,
21-
22- /** Maximum messages to queue per room */
23- MESSAGE_QUEUE_MAX_SIZE: 1000,
24-
25- /** Buddy list broadcast interval */
26- BUDDY_LIST_BROADCAST_INTERVAL: 10_000,
27-
28- // Client-side timing
29- /** Initial reconnect delay - 1 second */
30- INITIAL_RECONNECT_DELAY: 1_000,
31-
32- /** Maximum reconnect delay - 60 seconds cap */
33- MAX_RECONNECT_DELAY: 60_000,
34-
35- /** Heartbeat timeout - 45 seconds (1.5x server interval) */
36- HEARTBEAT_TIMEOUT: 45_000,
37-
38- // Shared timing
39- /** Idle timeout - mark as idle after 2 minutes */
40- IDLE_TIMEOUT: 2 * 60_000,
41-
42- /** Status update interval */
43- STATUS_UPDATE_INTERVAL: 30_000,
44-
45- /** EventSource retry hint (ms) - sent to client */
46- RETRY_INTERVAL: 3_000
47-} as const;
48-
49-/** SSE Event Types */
50-export type SSEEventType = 'chatMessage' | 'userStatusUpdate' | 'buddyListUpdate' | 'heartbeat';
51-
52-/** Connection status */
53-export type ConnectionStatus = 'connecting' | 'connected' | 'reconnecting' | 'disconnected';
D src/routes/api/sse/+server.ts +0 -385
@@ -1,385 +0,0 @@
1-/**
2- * SSE Server Endpoint
3- *
4- * Handles Server-Sent Events connections with:
5- * - Event IDs for Last-Event-ID catch-up support
6- * - Heartbeat mechanism for connection health
7- * - Proper connection lifecycle management
8- */
9-
10-declare global {
11- var __sseIntervalsStarted: boolean;
12-}
13-
14-import type { RequestHandler } from './$types';
15-import { SSE_CONFIG } from '$lib/sse/config';
16-import { sseEmitter, type SSEEvent } from '$lib/sse/SSEEventEmitter';
17-import { sseManager } from '$lib/sse/SSEManager';
18-import { messageQueue } from '$lib/sse/MessageQueue';
19-import db from '$lib/db/db.server';
20-import { users } from '$lib/db/schema';
21-import { eq, and, lt, ne } from 'drizzle-orm';
22-import { createSafeUser } from '$lib/types/chat';
23-import { createLogger } from '$lib/utils/logger.server';
24-import { buddyListCache } from '$lib/buddyListCache';
25-import { env } from '$env/dynamic/public';
26-
27-const log = createLogger('sse-server');
28-
29-console.log('[SSE] Server initialized with new SSE modules');
30-
31-// Default chat room ID
32-const DEFAULT_CHAT_ROOM_ID =
33- env.PUBLIC_DEFAULT_CHAT_ROOM_ID || '00000000-0000-0000-0000-000000000001';
34-
35-// Intervals for background tasks
36-let timeoutInterval: ReturnType<typeof setInterval>;
37-let buddyListInterval: ReturnType<typeof setInterval>;
38-let connectionCleanupInterval: ReturnType<typeof setInterval>;
39-
40-// Timestamp tracker for buddy list updates
41-let lastBuddyListUpdate = 0;
42-let lastBuddyListHash = '';
43-
44-/**
45- * Start periodic check for timed out users
46- */
47-function startTimeoutCheck() {
48- if (timeoutInterval) {
49- clearInterval(timeoutInterval);
50- }
51-
52- timeoutInterval = setInterval(async () => {
53- const timeoutThreshold = Date.now() - SSE_CONFIG.CONNECTION_TIMEOUT;
54- log.debug('Checking for timed out users', {
55- checkTime: new Date().toISOString(),
56- timeoutThreshold: new Date(timeoutThreshold).toISOString()
57- });
58-
59- try {
60- const now = Date.now();
61- const result = await db
62- .update(users)
63- .set({
64- status: 'offline',
65- lastSeen: now
66- })
67- .where(and(ne(users.status, 'offline'), lt(users.lastSeen, timeoutThreshold)))
68- .returning({
69- id: users.id,
70- nickname: users.nickname,
71- lastSeen: users.lastSeen
72- });
73-
74- if (result.length > 0) {
75- log.info('User status updated to offline due to timeout', {
76- usersUpdated: result.length,
77- timeoutThreshold: new Date(timeoutThreshold).toISOString()
78- });
79-
80- buddyListCache.invalidate();
81-
82- // Broadcast status updates for each timed out user
83- for (const user of result) {
84- sseEmitter.emit('userStatusUpdate', {
85- userId: user.id,
86- status: 'offline',
87- lastSeen: now
88- });
89- }
90- }
91- } catch (error) {
92- log.error('Error during timeout check', { error });
93- }
94- }, SSE_CONFIG.STALE_CHECK_INTERVAL);
95-}
96-
97-/**
98- * Start buddy list broadcast interval
99- */
100-function startBuddyListUpdateInterval() {
101- if (buddyListInterval) {
102- clearInterval(buddyListInterval);
103- }
104-
105- buddyListInterval = setInterval(async () => {
106- try {
107- const now = Date.now();
108- if (now - lastBuddyListUpdate >= SSE_CONFIG.BUDDY_LIST_BROADCAST_INTERVAL) {
109- if (buddyListCache.needsRefresh()) {
110- log.debug('Fetching buddy list from database (cache miss)...');
111- const buddyList = await db.select().from(users);
112- const safeBuddyList = buddyList.map((user) => createSafeUser(user));
113-
114- const hasChanged = buddyListCache.update(safeBuddyList);
115-
116- if (hasChanged) {
117- log.debug('Broadcasting buddy list update (data changed)...');
118- sseEmitter.emit('buddyListUpdate', safeBuddyList);
119- }
120- } else {
121- const cachedData = buddyListCache.get();
122- if (cachedData) {
123- const currentHash = JSON.stringify(cachedData);
124- if (currentHash !== lastBuddyListHash) {
125- log.debug('Broadcasting buddy list update from cache...');
126- lastBuddyListHash = currentHash;
127- sseEmitter.emit('buddyListUpdate', cachedData);
128- }
129- }
130- }
131- lastBuddyListUpdate = now;
132- }
133- } catch (error) {
134- log.error('Error fetching buddy list for update', { error });
135- }
136- }, 1000);
137-}
138-
139-/**
140- * Start periodic cleanup for stale connections
141- */
142-function startConnectionCleanup() {
143- if (connectionCleanupInterval) {
144- clearInterval(connectionCleanupInterval);
145- }
146-
147- connectionCleanupInterval = setInterval(async () => {
148- // Get stats and log them periodically
149- const stats = sseManager.getStats();
150- if (stats.totalConnections > 0) {
151- log.debug('SSE connection stats', stats);
152- }
153- }, SSE_CONFIG.STALE_CHECK_INTERVAL);
154-}
155-
156-// Start background tasks only once
157-if (!globalThis.__sseIntervalsStarted) {
158- startTimeoutCheck();
159- startBuddyListUpdateInterval();
160- startConnectionCleanup();
161- sseManager.startBackgroundTasks();
162- globalThis.__sseIntervalsStarted = true;
163- log.info('Started global intervals for SSE management');
164-} else {
165- log.debug('Global intervals already started, skipping initialization');
166-}
167-
168-// Cleanup on process exit
169-if (typeof process !== 'undefined') {
170- const cleanup = () => {
171- if (timeoutInterval) clearInterval(timeoutInterval);
172- if (buddyListInterval) clearInterval(buddyListInterval);
173- if (connectionCleanupInterval) clearInterval(connectionCleanupInterval);
174- sseManager.stopBackgroundTasks();
175- };
176-
177- process.on('beforeExit', cleanup);
178- process.on('SIGTERM', () => {
179- cleanup();
180- process.exit(0);
181- });
182- process.on('SIGINT', () => {
183- cleanup();
184- process.exit(0);
185- });
186-}
187-
188-/**
189- * Format an SSE event with ID and retry field
190- */
191-function formatSSEEvent(event: SSEEvent, eventId: string): string {
192- const lines = [
193- `id: ${eventId}`,
194- `event: ${event.type}`,
195- `data: ${JSON.stringify(event.data)}`,
196- `retry: ${SSE_CONFIG.RETRY_INTERVAL}`,
197- '',
198- ''
199- ];
200- return lines.join('\n');
201-}
202-
203-/**
204- * GET /api/sse
205- *
206- * Establishes an SSE connection for real-time updates.
207- */
208-export const GET: RequestHandler = async ({ request, locals }) => {
209- log.debug('Processing SSE request', {
210- url: request.url,
211- userId: locals.user?.id
212- });
213-
214- if (!locals.user) {
215- log.warn('Authentication failed - no user in locals');
216- return new Response('Unauthorized', {
217- status: 401,
218- headers: { 'Content-Type': 'application/json' }
219- });
220- }
221-
222- const userId = locals.user.id;
223- const roomId = DEFAULT_CHAT_ROOM_ID;
224- const encoder = new TextEncoder();
225-
226- // Check for Last-Event-ID header for catch-up
227- const lastEventId = request.headers.get('Last-Event-ID');
228- if (lastEventId) {
229- log.debug('Client reconnecting with Last-Event-ID', {
230- userId: userId.slice(0, 8),
231- lastEventId: lastEventId.slice(0, 8)
232- });
233- }
234-
235- let connectionId: string;
236- let keepAliveInterval: ReturnType<typeof setInterval>;
237- let cleanupListener: (() => void) | undefined;
238-
239- const stream = new ReadableStream({
240- async start(controller) {
241- // Add connection to manager
242- const { connectionId: connId, isNewUser } = sseManager.addConnection(
243- userId,
244- roomId,
245- controller
246- );
247- connectionId = connId;
248-
249- // Update user status if new connection
250- if (isNewUser) {
251- const now = Date.now();
252- try {
253- await db
254- .update(users)
255- .set({ status: 'online', lastSeen: now })
256- .where(eq(users.id, userId));
257-
258- log.info('User status updated to online', {
259- userId: userId.slice(0, 8),
260- connectionId: connectionId.slice(0, 8)
261- });
262-
263- buddyListCache.invalidate();
264-
265- sseEmitter.emit('userStatusUpdate', {
266- userId,
267- status: 'online',
268- lastSeen: now
269- });
270- } catch (error) {
271- log.error('Error updating user status', { error });
272- }
273- }
274-
275- // Send initial connection event
276- const connectEvent = formatSSEEvent(
277- { type: 'userStatusUpdate', data: { userId, status: 'connected', lastSeen: Date.now() } },
278- crypto.randomUUID()
279- );
280- controller.enqueue(encoder.encode(connectEvent));
281-
282- // Send catch-up messages if reconnecting with Last-Event-ID
283- if (lastEventId) {
284- const missedMessages = messageQueue.getMessagesSince(roomId, lastEventId);
285- if (missedMessages.length > 0) {
286- log.info('Sending catch-up messages', {
287- userId: userId.slice(0, 8),
288- count: missedMessages.length
289- });
290- for (const msg of missedMessages) {
291- // Cast to SSEEvent - the data was already validated when queued
292- const event = { type: msg.event, data: msg.data } as SSEEvent;
293- const payload = formatSSEEvent(event, msg.id);
294- controller.enqueue(encoder.encode(payload));
295- }
296- }
297- }
298-
299- // Listen for SSE events
300- const onSSE = (event: SSEEvent) => {
301- try {
302- const eventId = crypto.randomUUID();
303- const payload = formatSSEEvent(event, eventId);
304- controller.enqueue(encoder.encode(payload));
305-
306- // Queue message for catch-up (if not heartbeat)
307- if (event.type !== 'heartbeat') {
308- messageQueue.enqueue(roomId, event.type, event.data);
309- }
310- } catch (error) {
311- log.error('Error sending event', { error });
312- }
313- };
314-
315- cleanupListener = sseEmitter.addListener('sse', onSSE);
316-
317- // Handle abort signal
318- request.signal.addEventListener('abort', () => {
319- if (cleanupListener) {
320- cleanupListener();
321- log.info('Removed listener due to abort signal');
322- }
323- });
324-
325- log.info('SSE connection established', {
326- connectionId: connectionId.slice(0, 8),
327- userId: userId.slice(0, 8)
328- });
329-
330- // Keep-alive pings (in addition to heartbeats from manager)
331- keepAliveInterval = setInterval(() => {
332- try {
333- controller.enqueue(encoder.encode(':\n\n'));
334- } catch {
335- // Connection might be closed
336- }
337- }, SSE_CONFIG.HEARTBEAT_INTERVAL);
338- },
339-
340- async cancel(reason) {
341- if (keepAliveInterval) clearInterval(keepAliveInterval);
342- if (cleanupListener) cleanupListener();
343-
344- const maskedUserId = userId.slice(0, 8);
345- log.info('Connection closed', {
346- userId: maskedUserId,
347- connectionId: connectionId?.slice(0, 8),
348- reason
349- });
350-
351- // Remove connection from manager
352- const { isLastConnection } = sseManager.removeConnection(connectionId);
353-
354- // Update database if last connection for this user
355- if (isLastConnection) {
356- const now = Date.now();
357- try {
358- await db
359- .update(users)
360- .set({ status: 'offline', lastSeen: now })
361- .where(eq(users.id, userId));
362-
363- buddyListCache.invalidate();
364-
365- sseEmitter.emit('userStatusUpdate', {
366- userId,
367- status: 'offline',
368- lastSeen: now
369- });
370- } catch (error) {
371- log.error('Error updating user status on disconnect', { error });
372- }
373- }
374- }
375- });
376-
377- return new Response(stream, {
378- headers: {
379- 'Content-Type': 'text/event-stream',
380- 'Cache-Control': 'no-cache',
381- Connection: 'keep-alive',
382- 'X-Accel-Buffering': 'no' // Disable nginx buffering
383- }
384- });
385-};
@@ -1,385 +0,0 @@
1-/**
2- * SSE Server Endpoint
3- *
4- * Handles Server-Sent Events connections with:
5- * - Event IDs for Last-Event-ID catch-up support
6- * - Heartbeat mechanism for connection health
7- * - Proper connection lifecycle management
8- */
9-
10-declare global {
11- var __sseIntervalsStarted: boolean;
12-}
13-
14-import type { RequestHandler } from './$types';
15-import { SSE_CONFIG } from '$lib/sse/config';
16-import { sseEmitter, type SSEEvent } from '$lib/sse/SSEEventEmitter';
17-import { sseManager } from '$lib/sse/SSEManager';
18-import { messageQueue } from '$lib/sse/MessageQueue';
19-import db from '$lib/db/db.server';
20-import { users } from '$lib/db/schema';
21-import { eq, and, lt, ne } from 'drizzle-orm';
22-import { createSafeUser } from '$lib/types/chat';
23-import { createLogger } from '$lib/utils/logger.server';
24-import { buddyListCache } from '$lib/buddyListCache';
25-import { env } from '$env/dynamic/public';
26-
27-const log = createLogger('sse-server');
28-
29-console.log('[SSE] Server initialized with new SSE modules');
30-
31-// Default chat room ID
32-const DEFAULT_CHAT_ROOM_ID =
33- env.PUBLIC_DEFAULT_CHAT_ROOM_ID || '00000000-0000-0000-0000-000000000001';
34-
35-// Intervals for background tasks
36-let timeoutInterval: ReturnType<typeof setInterval>;
37-let buddyListInterval: ReturnType<typeof setInterval>;
38-let connectionCleanupInterval: ReturnType<typeof setInterval>;
39-
40-// Timestamp tracker for buddy list updates
41-let lastBuddyListUpdate = 0;
42-let lastBuddyListHash = '';
43-
44-/**
45- * Start periodic check for timed out users
46- */
47-function startTimeoutCheck() {
48- if (timeoutInterval) {
49- clearInterval(timeoutInterval);
50- }
51-
52- timeoutInterval = setInterval(async () => {
53- const timeoutThreshold = Date.now() - SSE_CONFIG.CONNECTION_TIMEOUT;
54- log.debug('Checking for timed out users', {
55- checkTime: new Date().toISOString(),
56- timeoutThreshold: new Date(timeoutThreshold).toISOString()
57- });
58-
59- try {
60- const now = Date.now();
61- const result = await db
62- .update(users)
63- .set({
64- status: 'offline',
65- lastSeen: now
66- })
67- .where(and(ne(users.status, 'offline'), lt(users.lastSeen, timeoutThreshold)))
68- .returning({
69- id: users.id,
70- nickname: users.nickname,
71- lastSeen: users.lastSeen
72- });
73-
74- if (result.length > 0) {
75- log.info('User status updated to offline due to timeout', {
76- usersUpdated: result.length,
77- timeoutThreshold: new Date(timeoutThreshold).toISOString()
78- });
79-
80- buddyListCache.invalidate();
81-
82- // Broadcast status updates for each timed out user
83- for (const user of result) {
84- sseEmitter.emit('userStatusUpdate', {
85- userId: user.id,
86- status: 'offline',
87- lastSeen: now
88- });
89- }
90- }
91- } catch (error) {
92- log.error('Error during timeout check', { error });
93- }
94- }, SSE_CONFIG.STALE_CHECK_INTERVAL);
95-}
96-
97-/**
98- * Start buddy list broadcast interval
99- */
100-function startBuddyListUpdateInterval() {
101- if (buddyListInterval) {
102- clearInterval(buddyListInterval);
103- }
104-
105- buddyListInterval = setInterval(async () => {
106- try {
107- const now = Date.now();
108- if (now - lastBuddyListUpdate >= SSE_CONFIG.BUDDY_LIST_BROADCAST_INTERVAL) {
109- if (buddyListCache.needsRefresh()) {
110- log.debug('Fetching buddy list from database (cache miss)...');
111- const buddyList = await db.select().from(users);
112- const safeBuddyList = buddyList.map((user) => createSafeUser(user));
113-
114- const hasChanged = buddyListCache.update(safeBuddyList);
115-
116- if (hasChanged) {
117- log.debug('Broadcasting buddy list update (data changed)...');
118- sseEmitter.emit('buddyListUpdate', safeBuddyList);
119- }
120- } else {
121- const cachedData = buddyListCache.get();
122- if (cachedData) {
123- const currentHash = JSON.stringify(cachedData);
124- if (currentHash !== lastBuddyListHash) {
125- log.debug('Broadcasting buddy list update from cache...');
126- lastBuddyListHash = currentHash;
127- sseEmitter.emit('buddyListUpdate', cachedData);
128- }
129- }
130- }
131- lastBuddyListUpdate = now;
132- }
133- } catch (error) {
134- log.error('Error fetching buddy list for update', { error });
135- }
136- }, 1000);
137-}
138-
139-/**
140- * Start periodic cleanup for stale connections
141- */
142-function startConnectionCleanup() {
143- if (connectionCleanupInterval) {
144- clearInterval(connectionCleanupInterval);
145- }
146-
147- connectionCleanupInterval = setInterval(async () => {
148- // Get stats and log them periodically
149- const stats = sseManager.getStats();
150- if (stats.totalConnections > 0) {
151- log.debug('SSE connection stats', stats);
152- }
153- }, SSE_CONFIG.STALE_CHECK_INTERVAL);
154-}
155-
156-// Start background tasks only once
157-if (!globalThis.__sseIntervalsStarted) {
158- startTimeoutCheck();
159- startBuddyListUpdateInterval();
160- startConnectionCleanup();
161- sseManager.startBackgroundTasks();
162- globalThis.__sseIntervalsStarted = true;
163- log.info('Started global intervals for SSE management');
164-} else {
165- log.debug('Global intervals already started, skipping initialization');
166-}
167-
168-// Cleanup on process exit
169-if (typeof process !== 'undefined') {
170- const cleanup = () => {
171- if (timeoutInterval) clearInterval(timeoutInterval);
172- if (buddyListInterval) clearInterval(buddyListInterval);
173- if (connectionCleanupInterval) clearInterval(connectionCleanupInterval);
174- sseManager.stopBackgroundTasks();
175- };
176-
177- process.on('beforeExit', cleanup);
178- process.on('SIGTERM', () => {
179- cleanup();
180- process.exit(0);
181- });
182- process.on('SIGINT', () => {
183- cleanup();
184- process.exit(0);
185- });
186-}
187-
188-/**
189- * Format an SSE event with ID and retry field
190- */
191-function formatSSEEvent(event: SSEEvent, eventId: string): string {
192- const lines = [
193- `id: ${eventId}`,
194- `event: ${event.type}`,
195- `data: ${JSON.stringify(event.data)}`,
196- `retry: ${SSE_CONFIG.RETRY_INTERVAL}`,
197- '',
198- ''
199- ];
200- return lines.join('\n');
201-}
202-
203-/**
204- * GET /api/sse
205- *
206- * Establishes an SSE connection for real-time updates.
207- */
208-export const GET: RequestHandler = async ({ request, locals }) => {
209- log.debug('Processing SSE request', {
210- url: request.url,
211- userId: locals.user?.id
212- });
213-
214- if (!locals.user) {
215- log.warn('Authentication failed - no user in locals');
216- return new Response('Unauthorized', {
217- status: 401,
218- headers: { 'Content-Type': 'application/json' }
219- });
220- }
221-
222- const userId = locals.user.id;
223- const roomId = DEFAULT_CHAT_ROOM_ID;
224- const encoder = new TextEncoder();
225-
226- // Check for Last-Event-ID header for catch-up
227- const lastEventId = request.headers.get('Last-Event-ID');
228- if (lastEventId) {
229- log.debug('Client reconnecting with Last-Event-ID', {
230- userId: userId.slice(0, 8),
231- lastEventId: lastEventId.slice(0, 8)
232- });
233- }
234-
235- let connectionId: string;
236- let keepAliveInterval: ReturnType<typeof setInterval>;
237- let cleanupListener: (() => void) | undefined;
238-
239- const stream = new ReadableStream({
240- async start(controller) {
241- // Add connection to manager
242- const { connectionId: connId, isNewUser } = sseManager.addConnection(
243- userId,
244- roomId,
245- controller
246- );
247- connectionId = connId;
248-
249- // Update user status if new connection
250- if (isNewUser) {
251- const now = Date.now();
252- try {
253- await db
254- .update(users)
255- .set({ status: 'online', lastSeen: now })
256- .where(eq(users.id, userId));
257-
258- log.info('User status updated to online', {
259- userId: userId.slice(0, 8),
260- connectionId: connectionId.slice(0, 8)
261- });
262-
263- buddyListCache.invalidate();
264-
265- sseEmitter.emit('userStatusUpdate', {
266- userId,
267- status: 'online',
268- lastSeen: now
269- });
270- } catch (error) {
271- log.error('Error updating user status', { error });
272- }
273- }
274-
275- // Send initial connection event
276- const connectEvent = formatSSEEvent(
277- { type: 'userStatusUpdate', data: { userId, status: 'connected', lastSeen: Date.now() } },
278- crypto.randomUUID()
279- );
280- controller.enqueue(encoder.encode(connectEvent));
281-
282- // Send catch-up messages if reconnecting with Last-Event-ID
283- if (lastEventId) {
284- const missedMessages = messageQueue.getMessagesSince(roomId, lastEventId);
285- if (missedMessages.length > 0) {
286- log.info('Sending catch-up messages', {
287- userId: userId.slice(0, 8),
288- count: missedMessages.length
289- });
290- for (const msg of missedMessages) {
291- // Cast to SSEEvent - the data was already validated when queued
292- const event = { type: msg.event, data: msg.data } as SSEEvent;
293- const payload = formatSSEEvent(event, msg.id);
294- controller.enqueue(encoder.encode(payload));
295- }
296- }
297- }
298-
299- // Listen for SSE events
300- const onSSE = (event: SSEEvent) => {
301- try {
302- const eventId = crypto.randomUUID();
303- const payload = formatSSEEvent(event, eventId);
304- controller.enqueue(encoder.encode(payload));
305-
306- // Queue message for catch-up (if not heartbeat)
307- if (event.type !== 'heartbeat') {
308- messageQueue.enqueue(roomId, event.type, event.data);
309- }
310- } catch (error) {
311- log.error('Error sending event', { error });
312- }
313- };
314-
315- cleanupListener = sseEmitter.addListener('sse', onSSE);
316-
317- // Handle abort signal
318- request.signal.addEventListener('abort', () => {
319- if (cleanupListener) {
320- cleanupListener();
321- log.info('Removed listener due to abort signal');
322- }
323- });
324-
325- log.info('SSE connection established', {
326- connectionId: connectionId.slice(0, 8),
327- userId: userId.slice(0, 8)
328- });
329-
330- // Keep-alive pings (in addition to heartbeats from manager)
331- keepAliveInterval = setInterval(() => {
332- try {
333- controller.enqueue(encoder.encode(':\n\n'));
334- } catch {
335- // Connection might be closed
336- }
337- }, SSE_CONFIG.HEARTBEAT_INTERVAL);
338- },
339-
340- async cancel(reason) {
341- if (keepAliveInterval) clearInterval(keepAliveInterval);
342- if (cleanupListener) cleanupListener();
343-
344- const maskedUserId = userId.slice(0, 8);
345- log.info('Connection closed', {
346- userId: maskedUserId,
347- connectionId: connectionId?.slice(0, 8),
348- reason
349- });
350-
351- // Remove connection from manager
352- const { isLastConnection } = sseManager.removeConnection(connectionId);
353-
354- // Update database if last connection for this user
355- if (isLastConnection) {
356- const now = Date.now();
357- try {
358- await db
359- .update(users)
360- .set({ status: 'offline', lastSeen: now })
361- .where(eq(users.id, userId));
362-
363- buddyListCache.invalidate();
364-
365- sseEmitter.emit('userStatusUpdate', {
366- userId,
367- status: 'offline',
368- lastSeen: now
369- });
370- } catch (error) {
371- log.error('Error updating user status on disconnect', { error });
372- }
373- }
374- }
375- });
376-
377- return new Response(stream, {
378- headers: {
379- 'Content-Type': 'text/event-stream',
380- 'Cache-Control': 'no-cache',
381- Connection: 'keep-alive',
382- 'X-Accel-Buffering': 'no' // Disable nginx buffering
383- }
384- });
385-};
D src/routes/api/sse/heartbeat/+server.ts +0 -66
@@ -1,66 +0,0 @@
1-/**
2- * Heartbeat Acknowledgment Endpoint
3- *
4- * Receives heartbeat acknowledgments from clients to track
5- * active connections and detect stale ones.
6- */
7-
8-import type { RequestHandler } from './$types';
9-import { sseManager } from '$lib/sse/SSEManager';
10-import { createLogger } from '$lib/utils/logger.server';
11-import { json } from '@sveltejs/kit';
12-
13-const log = createLogger('sse-heartbeat');
14-
15-/**
16- * POST /api/sse/heartbeat
17- *
18- * Acknowledges a heartbeat from a client, updating the connection's
19- * last heartbeat timestamp.
20- */
21-export const POST: RequestHandler = async ({ request, locals }) => {
22- // Require authentication
23- if (!locals.user) {
24- return json({ success: false, error: 'Unauthorized' }, { status: 401 });
25- }
26-
27- try {
28- const body = await request.json();
29- const { connectionId } = body;
30-
31- if (!connectionId || typeof connectionId !== 'string') {
32- return json({ success: false, error: 'Invalid connectionId' }, { status: 400 });
33- }
34-
35- // Verify the connection belongs to this user
36- const connection = sseManager.getConnection(connectionId);
37- if (!connection) {
38- log.debug('Heartbeat for unknown connection', {
39- connectionId: connectionId.slice(0, 8),
40- userId: locals.user.id.slice(0, 8)
41- });
42- return json({ success: false, error: 'Connection not found' }, { status: 404 });
43- }
44-
45- if (connection.userId !== locals.user.id) {
46- log.warn('Heartbeat userId mismatch', {
47- connectionId: connectionId.slice(0, 8),
48- expectedUserId: connection.userId.slice(0, 8),
49- actualUserId: locals.user.id.slice(0, 8)
50- });
51- return json({ success: false, error: 'Unauthorized' }, { status: 403 });
52- }
53-
54- // Acknowledge the heartbeat
55- const acknowledged = sseManager.acknowledgeHeartbeat(connectionId);
56-
57- if (acknowledged) {
58- return json({ success: true });
59- } else {
60- return json({ success: false, error: 'Failed to acknowledge' }, { status: 500 });
61- }
62- } catch (error) {
63- log.error('Heartbeat error', { error });
64- return json({ success: false, error: 'Internal server error' }, { status: 500 });
65- }
66-};
@@ -1,66 +0,0 @@
1-/**
2- * Heartbeat Acknowledgment Endpoint
3- *
4- * Receives heartbeat acknowledgments from clients to track
5- * active connections and detect stale ones.
6- */
7-
8-import type { RequestHandler } from './$types';
9-import { sseManager } from '$lib/sse/SSEManager';
10-import { createLogger } from '$lib/utils/logger.server';
11-import { json } from '@sveltejs/kit';
12-
13-const log = createLogger('sse-heartbeat');
14-
15-/**
16- * POST /api/sse/heartbeat
17- *
18- * Acknowledges a heartbeat from a client, updating the connection's
19- * last heartbeat timestamp.
20- */
21-export const POST: RequestHandler = async ({ request, locals }) => {
22- // Require authentication
23- if (!locals.user) {
24- return json({ success: false, error: 'Unauthorized' }, { status: 401 });
25- }
26-
27- try {
28- const body = await request.json();
29- const { connectionId } = body;
30-
31- if (!connectionId || typeof connectionId !== 'string') {
32- return json({ success: false, error: 'Invalid connectionId' }, { status: 400 });
33- }
34-
35- // Verify the connection belongs to this user
36- const connection = sseManager.getConnection(connectionId);
37- if (!connection) {
38- log.debug('Heartbeat for unknown connection', {
39- connectionId: connectionId.slice(0, 8),
40- userId: locals.user.id.slice(0, 8)
41- });
42- return json({ success: false, error: 'Connection not found' }, { status: 404 });
43- }
44-
45- if (connection.userId !== locals.user.id) {
46- log.warn('Heartbeat userId mismatch', {
47- connectionId: connectionId.slice(0, 8),
48- expectedUserId: connection.userId.slice(0, 8),
49- actualUserId: locals.user.id.slice(0, 8)
50- });
51- return json({ success: false, error: 'Unauthorized' }, { status: 403 });
52- }
53-
54- // Acknowledge the heartbeat
55- const acknowledged = sseManager.acknowledgeHeartbeat(connectionId);
56-
57- if (acknowledged) {
58- return json({ success: true });
59- } else {
60- return json({ success: false, error: 'Failed to acknowledge' }, { status: 500 });
61- }
62- } catch (error) {
63- log.error('Heartbeat error', { error });
64- return json({ success: false, error: 'Internal server error' }, { status: 500 });
65- }
66-};

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 3

Tags 8

↑↓ moveenter openesc close