detached 0270c67

feat(api): add heartbeat endpoint and refactor SSE handler

- Add POST /api/sse/heartbeat endpoint for connection keep-alive
- Refactor main SSE handler (/api/sse) to use new modular architecture
- Improve error handling and connection lifecycle management
- Simplify message routing and event handling
commit
0270c678830ef72fc5d4f9e77857e64cedb2a74d
author
Kit · · 2025-12-15 10:21 UTC
parent
78e1985
signature
SSH UnverifiedSHA256:H91DFuEG4jpk5edzLA4aUKkpv4zv9Q339ipRmkph6Tk The key is unknown
2 files · +270 -145
Unified
UnifiedSplit
M src/routes/api/sse/+server.ts +204 -145
@@ -1,44 +1,56 @@
1-// At the top of the file, before any code
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+
210 declare global {
311 var __sseIntervalsStarted: boolean;
412 }
513
614 import type { RequestHandler } from './$types';
7-import { sseEmitter } from '$lib/sseEmitter';
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';
819 import db from '$lib/db/db.server';
920 import { users } from '$lib/db/schema';
1021 import { eq, and, lt, ne } from 'drizzle-orm';
1122 import { createSafeUser } from '$lib/types/chat';
1223 import { createLogger } from '$lib/utils/logger.server';
13-import { sseConnectionTracker } from '$lib/sseConnectionTracker';
1424 import { buddyListCache } from '$lib/buddyListCache';
25+import { env } from '$env/dynamic/public';
1526
1627 const log = createLogger('sse-server');
1728
18-console.log('[SSE] Server initialized');
19-
20-// Constants for timeout checks
21-const ONLINE_TIMEOUT_MS = 2 * 60 * 1000; // 2 minutes
22-const CHECK_INTERVAL_MS = 30 * 1000; // Check every 30 seconds
29+console.log('[SSE] Server initialized with new SSE modules');
2330
24-// Update the interval constant
25-const BUDDY_LIST_BROADCAST_INTERVAL = 10000; // 10 seconds
31+// Default chat room ID
32+const DEFAULT_CHAT_ROOM_ID =
33+ env.PUBLIC_DEFAULT_CHAT_ROOM_ID || '00000000-0000-0000-0000-000000000001';
2634
35+// Intervals for background tasks
2736 let timeoutInterval: ReturnType<typeof setInterval>;
2837 let buddyListInterval: ReturnType<typeof setInterval>;
38+let connectionCleanupInterval: ReturnType<typeof setInterval>;
2939
30-// Add a timestamp tracker for buddy list updates
40+// Timestamp tracker for buddy list updates
3141 let lastBuddyListUpdate = 0;
32-let lastBuddyListData: unknown[] = [];
42+let lastBuddyListHash = '';
3343
34-// Initialize periodic check for timed out users
44+/**
45+ * Start periodic check for timed out users
46+ */
3547 function startTimeoutCheck() {
3648 if (timeoutInterval) {
3749 clearInterval(timeoutInterval);
3850 }
3951
4052 timeoutInterval = setInterval(async () => {
41- const timeoutThreshold = Date.now() - ONLINE_TIMEOUT_MS;
53+ const timeoutThreshold = Date.now() - SSE_CONFIG.CONNECTION_TIMEOUT;
4254 log.debug('Checking for timed out users', {
4355 checkTime: new Date().toISOString(),
4456 timeoutThreshold: new Date(timeoutThreshold).toISOString()
@@ -54,8 +66,8 @@ function startTimeoutCheck() {
5466 })
5567 .where(
5668 and(
57- ne(users.status, 'offline'), // Update any non-offline users
58- lt(users.lastSeen, timeoutThreshold) // Who haven't been seen recently
69+ ne(users.status, 'offline'),
70+ lt(users.lastSeen, timeoutThreshold)
5971 )
6072 )
6173 .returning({
@@ -65,20 +77,31 @@ function startTimeoutCheck() {
6577 });
6678
6779 if (result.length > 0) {
68- log.info('User status updated to offline', {
80+ log.info('User status updated to offline due to timeout', {
6981 usersUpdated: result.length,
7082 timeoutThreshold: new Date(timeoutThreshold).toISOString()
7183 });
7284
73- // Invalidate buddy list cache when users timeout
7485 buddyListCache.invalidate();
86+
87+ // Broadcast status updates for each timed out user
88+ for (const user of result) {
89+ sseEmitter.emit('userStatusUpdate', {
90+ userId: user.id,
91+ status: 'offline',
92+ lastSeen: now
93+ });
94+ }
7595 }
7696 } catch (error) {
7797 log.error('Error during timeout check', { error });
7898 }
79- }, CHECK_INTERVAL_MS);
99+ }, SSE_CONFIG.STALE_CHECK_INTERVAL);
80100 }
81101
102+/**
103+ * Start buddy list broadcast interval
104+ */
82105 function startBuddyListUpdateInterval() {
83106 if (buddyListInterval) {
84107 clearInterval(buddyListInterval);
@@ -87,36 +110,27 @@ function startBuddyListUpdateInterval() {
87110 buddyListInterval = setInterval(async () => {
88111 try {
89112 const now = Date.now();
90- // Only fetch and broadcast if 10 seconds have passed since last update
91- if (now - lastBuddyListUpdate >= BUDDY_LIST_BROADCAST_INTERVAL) {
92- // Only query database if cache is invalid or doesn't exist
113+ if (now - lastBuddyListUpdate >= SSE_CONFIG.BUDDY_LIST_BROADCAST_INTERVAL) {
93114 if (buddyListCache.needsRefresh()) {
94115 log.debug('Fetching buddy list from database (cache miss)...');
95- // Query the complete buddy list from the database.
96116 const buddyList = await db.select().from(users);
97- // Use createSafeUser to properly sanitize user data
98117 const safeBuddyList = buddyList.map((user) => createSafeUser(user));
99118
100- // Update cache and check if data changed
101119 const hasChanged = buddyListCache.update(safeBuddyList);
102120
103121 if (hasChanged) {
104122 log.debug('Broadcasting buddy list update (data changed)...');
105- sseEmitter.emit('sse', {
106- type: 'buddyListUpdate',
107- data: safeBuddyList
108- });
123+ sseEmitter.emit('buddyListUpdate', safeBuddyList);
109124 }
110125 } else {
111- // Use cached data
112126 const cachedData = buddyListCache.get();
113- if (cachedData && JSON.stringify(cachedData) !== JSON.stringify(lastBuddyListData)) {
114- log.debug('Broadcasting buddy list update from cache...');
115- lastBuddyListData = cachedData;
116- sseEmitter.emit('sse', {
117- type: 'buddyListUpdate',
118- data: cachedData
119- });
127+ if (cachedData) {
128+ const currentHash = JSON.stringify(cachedData);
129+ if (currentHash !== lastBuddyListHash) {
130+ log.debug('Broadcasting buddy list update from cache...');
131+ lastBuddyListHash = currentHash;
132+ sseEmitter.emit('buddyListUpdate', cachedData);
133+ }
120134 }
121135 }
122136 lastBuddyListUpdate = now;
@@ -124,72 +138,78 @@ function startBuddyListUpdateInterval() {
124138 } catch (error) {
125139 log.error('Error fetching buddy list for update', { error });
126140 }
127- }, 1000); // Check every second but only broadcast every 10 seconds
141+ }, 1000);
128142 }
129143
130-// Add periodic cleanup for stale connections
131-let connectionCleanupInterval: ReturnType<typeof setInterval>;
132-
144+/**
145+ * Start periodic cleanup for stale connections
146+ */
133147 function startConnectionCleanup() {
134148 if (connectionCleanupInterval) {
135149 clearInterval(connectionCleanupInterval);
136150 }
137151
138- connectionCleanupInterval = setInterval(() => {
139- const staleUserIds = sseConnectionTracker.cleanupStaleConnections();
140- if (staleUserIds.length > 0) {
141- // Mark stale connections as offline
142- staleUserIds.forEach(async (userId) => {
143- try {
144- await db
145- .update(users)
146- .set({ status: 'offline', lastSeen: Date.now() })
147- .where(eq(users.id, userId));
148-
149- buddyListCache.invalidate();
150- } catch (error) {
151- log.error('Error updating stale user status', { userId, error });
152- }
153- });
152+ connectionCleanupInterval = setInterval(async () => {
153+ // Get stats and log them periodically
154+ const stats = sseManager.getStats();
155+ if (stats.totalConnections > 0) {
156+ log.debug('SSE connection stats', stats);
154157 }
155- }, 60000); // Run every minute
158+ }, SSE_CONFIG.STALE_CHECK_INTERVAL);
156159 }
157160
158-// <<< NEW: Ensure intervals are started only once >>>
161+// Start background tasks only once
159162 if (!globalThis.__sseIntervalsStarted) {
160163 startTimeoutCheck();
161164 startBuddyListUpdateInterval();
162165 startConnectionCleanup();
166+ sseManager.startBackgroundTasks();
163167 globalThis.__sseIntervalsStarted = true;
164- log.info('Started global intervals for timeout and buddy list updates');
168+ log.info('Started global intervals for SSE management');
165169 } else {
166170 log.debug('Global intervals already started, skipping initialization');
167171 }
168172
169-// Ensure the interval is cleared if the module is reloaded
173+// Cleanup on process exit
170174 if (typeof process !== 'undefined') {
171- process.on('beforeExit', () => {
175+ const cleanup = () => {
172176 if (timeoutInterval) clearInterval(timeoutInterval);
173177 if (buddyListInterval) clearInterval(buddyListInterval);
174178 if (connectionCleanupInterval) clearInterval(connectionCleanupInterval);
175- });
179+ sseManager.stopBackgroundTasks();
180+ };
176181
177- // Also handle SIGTERM and SIGINT
182+ process.on('beforeExit', cleanup);
178183 process.on('SIGTERM', () => {
179- if (timeoutInterval) clearInterval(timeoutInterval);
180- if (buddyListInterval) clearInterval(buddyListInterval);
181- if (connectionCleanupInterval) clearInterval(connectionCleanupInterval);
184+ cleanup();
182185 process.exit(0);
183186 });
184-
185187 process.on('SIGINT', () => {
186- if (timeoutInterval) clearInterval(timeoutInterval);
187- if (buddyListInterval) clearInterval(buddyListInterval);
188- if (connectionCleanupInterval) clearInterval(connectionCleanupInterval);
188+ cleanup();
189189 process.exit(0);
190190 });
191191 }
192192
193+/**
194+ * Format an SSE event with ID and retry field
195+ */
196+function formatSSEEvent(event: SSEEvent, eventId: string): string {
197+ const lines = [
198+ `id: ${eventId}`,
199+ `event: ${event.type}`,
200+ `data: ${JSON.stringify(event.data)}`,
201+ `retry: ${SSE_CONFIG.RETRY_INTERVAL}`,
202+ '',
203+ ''
204+ ];
205+ return lines.join('\n');
206+}
207+
208+/**
209+ * GET /api/sse
210+ *
211+ * Establishes an SSE connection for real-time updates.
212+ */
193213 export const GET: RequestHandler = async ({ request, locals }) => {
194214 log.debug('Processing SSE request', {
195215 url: request.url,
@@ -200,119 +220,157 @@ export const GET: RequestHandler = async ({ request, locals }) => {
200220 log.warn('Authentication failed - no user in locals');
201221 return new Response('Unauthorized', {
202222 status: 401,
203- headers: {
204- 'Content-Type': 'application/json'
205- }
223+ headers: { 'Content-Type': 'application/json' }
206224 });
207225 }
208226
209227 const userId = locals.user.id;
228+ const roomId = DEFAULT_CHAT_ROOM_ID;
210229 const encoder = new TextEncoder();
211- let keepAliveInterval: ReturnType<typeof setInterval>;
212- let onSSE: ((event: { type: string; data: unknown }) => void) | undefined;
213230
214- const now = Date.now();
231+ // Check for Last-Event-ID header for catch-up
232+ const lastEventId = request.headers.get('Last-Event-ID');
233+ if (lastEventId) {
234+ log.debug('Client reconnecting with Last-Event-ID', {
235+ userId: userId.slice(0, 8),
236+ lastEventId: lastEventId.slice(0, 8)
237+ });
238+ }
215239
216- // Only update status if this is a new connection
217- const isNewConnection = sseConnectionTracker.addConnection(userId);
240+ let connectionId: string;
241+ let keepAliveInterval: ReturnType<typeof setInterval>;
242+ let cleanupListener: (() => void) | undefined;
218243
219- if (isNewConnection) {
220- try {
221- await db
222- .update(users)
223- .set({
224- status: 'online',
225- lastSeen: now
226- })
227- .where(eq(users.id, userId));
244+ const stream = new ReadableStream({
245+ async start(controller) {
246+ // Add connection to manager
247+ const { connectionId: connId, isNewUser } = sseManager.addConnection(
248+ userId,
249+ roomId,
250+ controller
251+ );
252+ connectionId = connId;
253+
254+ // Update user status if new connection
255+ if (isNewUser) {
256+ const now = Date.now();
257+ try {
258+ await db
259+ .update(users)
260+ .set({ status: 'online', lastSeen: now })
261+ .where(eq(users.id, userId));
228262
229- log.info('User status updated to online', {
230- userId: `${userId.slice(0, 4)}...${userId.slice(-4)}`,
231- timestamp: new Date(now).toISOString()
232- });
263+ log.info('User status updated to online', {
264+ userId: userId.slice(0, 8),
265+ connectionId: connectionId.slice(0, 8)
266+ });
233267
234- // Invalidate buddy list cache when user status changes
235- buddyListCache.invalidate();
268+ buddyListCache.invalidate();
236269
237- sseEmitter.emit('sse', {
238- type: 'userStatusUpdate',
239- data: {
240- userId,
241- status: 'online',
242- lastSeen: now
270+ sseEmitter.emit('userStatusUpdate', {
271+ userId,
272+ status: 'online',
273+ lastSeen: now
274+ });
275+ } catch (error) {
276+ log.error('Error updating user status', { error });
243277 }
244- });
245- } catch (error) {
246- log.error('Error updating user status', { error });
247- sseConnectionTracker.removeConnection(userId);
248- }
249- } else {
250- log.debug('User already has active SSE connection, skipping status update', {
251- userId: `${userId.slice(0, 4)}...${userId.slice(-4)}`
252- });
253- }
278+ }
254279
255- const stream = new ReadableStream({
256- start(controller) {
257- const sendEvent = (event: { type: string; data: unknown }) => {
280+ // Send initial connection event
281+ const connectEvent = formatSSEEvent(
282+ { type: 'userStatusUpdate', data: { userId, status: 'connected', lastSeen: Date.now() } },
283+ crypto.randomUUID()
284+ );
285+ controller.enqueue(encoder.encode(connectEvent));
286+
287+ // Send catch-up messages if reconnecting with Last-Event-ID
288+ if (lastEventId) {
289+ const missedMessages = messageQueue.getMessagesSince(roomId, lastEventId);
290+ if (missedMessages.length > 0) {
291+ log.info('Sending catch-up messages', {
292+ userId: userId.slice(0, 8),
293+ count: missedMessages.length
294+ });
295+ for (const msg of missedMessages) {
296+ // Cast to SSEEvent - the data was already validated when queued
297+ const event = { type: msg.event, data: msg.data } as SSEEvent;
298+ const payload = formatSSEEvent(event, msg.id);
299+ controller.enqueue(encoder.encode(payload));
300+ }
301+ }
302+ }
303+
304+ // Listen for SSE events
305+ const onSSE = (event: SSEEvent) => {
258306 try {
259- const payload = `event: ${event.type}\n` + `data: ${JSON.stringify(event.data)}\n\n`;
307+ const eventId = crypto.randomUUID();
308+ const payload = formatSSEEvent(event, eventId);
260309 controller.enqueue(encoder.encode(payload));
261- if (!['userStatusUpdate', 'chatMessage'].includes(event.type)) {
262- log.info('Event sent', { type: event.type });
310+
311+ // Queue message for catch-up (if not heartbeat)
312+ if (event.type !== 'heartbeat') {
313+ messageQueue.enqueue(roomId, event.type, event.data);
263314 }
264315 } catch (error) {
265316 log.error('Error sending event', { error });
266317 }
267318 };
268319
269- onSSE = (event: { type: string; data: unknown }) => {
270- sendEvent(event);
271- };
320+ cleanupListener = sseEmitter.addListener('sse', onSSE);
272321
273- sseEmitter.addListener('sse', onSSE);
322+ // Handle abort signal
274323 request.signal.addEventListener('abort', () => {
275- if (onSSE) {
276- sseEmitter.removeListener('sse', onSSE);
324+ if (cleanupListener) {
325+ cleanupListener();
277326 log.info('Removed listener due to abort signal');
278327 }
279328 });
280- log.info('Added new listener');
281329
282- controller.enqueue(encoder.encode('data: Connected\n\n'));
330+ log.info('SSE connection established', {
331+ connectionId: connectionId.slice(0, 8),
332+ userId: userId.slice(0, 8)
333+ });
283334
335+ // Keep-alive pings (in addition to heartbeats from manager)
284336 keepAliveInterval = setInterval(() => {
285- controller.enqueue(encoder.encode(':\n\n'));
286- }, 20000);
337+ try {
338+ controller.enqueue(encoder.encode(':\n\n'));
339+ } catch {
340+ // Connection might be closed
341+ }
342+ }, SSE_CONFIG.HEARTBEAT_INTERVAL);
287343 },
344+
288345 async cancel(reason) {
289346 if (keepAliveInterval) clearInterval(keepAliveInterval);
290- if (onSSE) sseEmitter.removeListener('sse', onSSE);
291- const maskedUserId = `${userId.slice(0, 4)}...${userId.slice(-4)}`;
292- log.info('Connection closed', { userId: maskedUserId, reason });
347+ if (cleanupListener) cleanupListener();
293348
294- // Remove from connection tracker
295- const wasTracked = sseConnectionTracker.removeConnection(userId);
349+ const maskedUserId = userId.slice(0, 8);
350+ log.info('Connection closed', { userId: maskedUserId, connectionId: connectionId?.slice(0, 8), reason });
296351
297- // Only update database if this was the last connection
298- if (wasTracked && !sseConnectionTracker.hasConnection(userId)) {
352+ // Remove connection from manager
353+ const { isLastConnection } = sseManager.removeConnection(connectionId);
354+
355+ // Update database if last connection for this user
356+ if (isLastConnection) {
299357 const now = Date.now();
300- await db
301- .update(users)
302- .set({ status: 'offline', lastSeen: now })
303- .where(eq(users.id, userId));
358+ try {
359+ await db
360+ .update(users)
361+ .set({ status: 'offline', lastSeen: now })
362+ .where(eq(users.id, userId));
304363
305- // Invalidate buddy list cache when user goes offline
306- buddyListCache.invalidate();
364+ buddyListCache.invalidate();
307365
308- sseEmitter.emit('sse', {
309- type: 'userStatusUpdate',
310- data: {
366+ sseEmitter.emit('userStatusUpdate', {
311367 userId,
312368 status: 'offline',
313369 lastSeen: now
314- }
315- });
370+ });
371+ } catch (error) {
372+ log.error('Error updating user status on disconnect', { error });
373+ }
316374 }
317375 }
318376 });
@@ -321,7 +379,8 @@ export const GET: RequestHandler = async ({ request, locals }) => {
321379 headers: {
322380 'Content-Type': 'text/event-stream',
323381 'Cache-Control': 'no-cache',
324- Connection: 'keep-alive'
382+ 'Connection': 'keep-alive',
383+ 'X-Accel-Buffering': 'no' // Disable nginx buffering
325384 }
326385 });
327386 };
@@ -1,44 +1,56 @@
1-// At the top of the file, before any code1+/**
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+
2 declare global {10 declare global {
3 var __sseIntervalsStarted: boolean;11 var __sseIntervalsStarted: boolean;
4 }12 }
5 13
6 import type { RequestHandler } from './$types';14 import type { RequestHandler } from './$types';
7-import { sseEmitter } from '$lib/sseEmitter';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';
8 import db from '$lib/db/db.server';19 import db from '$lib/db/db.server';
9 import { users } from '$lib/db/schema';20 import { users } from '$lib/db/schema';
10 import { eq, and, lt, ne } from 'drizzle-orm';21 import { eq, and, lt, ne } from 'drizzle-orm';
11 import { createSafeUser } from '$lib/types/chat';22 import { createSafeUser } from '$lib/types/chat';
12 import { createLogger } from '$lib/utils/logger.server';23 import { createLogger } from '$lib/utils/logger.server';
13-import { sseConnectionTracker } from '$lib/sseConnectionTracker';
14 import { buddyListCache } from '$lib/buddyListCache';24 import { buddyListCache } from '$lib/buddyListCache';
25+import { env } from '$env/dynamic/public';
15 26
16 const log = createLogger('sse-server');27 const log = createLogger('sse-server');
17 28
18-console.log('[SSE] Server initialized');29+console.log('[SSE] Server initialized with new SSE modules');
19-
20-// Constants for timeout checks
21-const ONLINE_TIMEOUT_MS = 2 * 60 * 1000; // 2 minutes
22-const CHECK_INTERVAL_MS = 30 * 1000; // Check every 30 seconds
23 30
24-// Update the interval constant31+// Default chat room ID
25-const BUDDY_LIST_BROADCAST_INTERVAL = 10000; // 10 seconds32+const DEFAULT_CHAT_ROOM_ID =
33+ env.PUBLIC_DEFAULT_CHAT_ROOM_ID || '00000000-0000-0000-0000-000000000001';
26 34
35+// Intervals for background tasks
27 let timeoutInterval: ReturnType<typeof setInterval>;36 let timeoutInterval: ReturnType<typeof setInterval>;
28 let buddyListInterval: ReturnType<typeof setInterval>;37 let buddyListInterval: ReturnType<typeof setInterval>;
38+let connectionCleanupInterval: ReturnType<typeof setInterval>;
29 39
30-// Add a timestamp tracker for buddy list updates40+// Timestamp tracker for buddy list updates
31 let lastBuddyListUpdate = 0;41 let lastBuddyListUpdate = 0;
32-let lastBuddyListData: unknown[] = [];42+let lastBuddyListHash = '';
33 43
34-// Initialize periodic check for timed out users44+/**
45+ * Start periodic check for timed out users
46+ */
35 function startTimeoutCheck() {47 function startTimeoutCheck() {
36 if (timeoutInterval) {48 if (timeoutInterval) {
37 clearInterval(timeoutInterval);49 clearInterval(timeoutInterval);
38 }50 }
39 51
40 timeoutInterval = setInterval(async () => {52 timeoutInterval = setInterval(async () => {
41- const timeoutThreshold = Date.now() - ONLINE_TIMEOUT_MS;53+ const timeoutThreshold = Date.now() - SSE_CONFIG.CONNECTION_TIMEOUT;
42 log.debug('Checking for timed out users', {54 log.debug('Checking for timed out users', {
43 checkTime: new Date().toISOString(),55 checkTime: new Date().toISOString(),
44 timeoutThreshold: new Date(timeoutThreshold).toISOString()56 timeoutThreshold: new Date(timeoutThreshold).toISOString()
@@ -54,8 +66,8 @@ function startTimeoutCheck() {
54 })66 })
55 .where(67 .where(
56 and(68 and(
57- ne(users.status, 'offline'), // Update any non-offline users69+ ne(users.status, 'offline'),
58- lt(users.lastSeen, timeoutThreshold) // Who haven't been seen recently70+ lt(users.lastSeen, timeoutThreshold)
59 )71 )
60 )72 )
61 .returning({73 .returning({
@@ -65,20 +77,31 @@ function startTimeoutCheck() {
65 });77 });
66 78
67 if (result.length > 0) {79 if (result.length > 0) {
68- log.info('User status updated to offline', {80+ log.info('User status updated to offline due to timeout', {
69 usersUpdated: result.length,81 usersUpdated: result.length,
70 timeoutThreshold: new Date(timeoutThreshold).toISOString()82 timeoutThreshold: new Date(timeoutThreshold).toISOString()
71 });83 });
72 84
73- // Invalidate buddy list cache when users timeout
74 buddyListCache.invalidate();85 buddyListCache.invalidate();
86+
87+ // Broadcast status updates for each timed out user
88+ for (const user of result) {
89+ sseEmitter.emit('userStatusUpdate', {
90+ userId: user.id,
91+ status: 'offline',
92+ lastSeen: now
93+ });
94+ }
75 }95 }
76 } catch (error) {96 } catch (error) {
77 log.error('Error during timeout check', { error });97 log.error('Error during timeout check', { error });
78 }98 }
79- }, CHECK_INTERVAL_MS);99+ }, SSE_CONFIG.STALE_CHECK_INTERVAL);
80 }100 }
81 101
102+/**
103+ * Start buddy list broadcast interval
104+ */
82 function startBuddyListUpdateInterval() {105 function startBuddyListUpdateInterval() {
83 if (buddyListInterval) {106 if (buddyListInterval) {
84 clearInterval(buddyListInterval);107 clearInterval(buddyListInterval);
@@ -87,36 +110,27 @@ function startBuddyListUpdateInterval() {
87 buddyListInterval = setInterval(async () => {110 buddyListInterval = setInterval(async () => {
88 try {111 try {
89 const now = Date.now();112 const now = Date.now();
90- // Only fetch and broadcast if 10 seconds have passed since last update113+ if (now - lastBuddyListUpdate >= SSE_CONFIG.BUDDY_LIST_BROADCAST_INTERVAL) {
91- if (now - lastBuddyListUpdate >= BUDDY_LIST_BROADCAST_INTERVAL) {
92- // Only query database if cache is invalid or doesn't exist
93 if (buddyListCache.needsRefresh()) {114 if (buddyListCache.needsRefresh()) {
94 log.debug('Fetching buddy list from database (cache miss)...');115 log.debug('Fetching buddy list from database (cache miss)...');
95- // Query the complete buddy list from the database.
96 const buddyList = await db.select().from(users);116 const buddyList = await db.select().from(users);
97- // Use createSafeUser to properly sanitize user data
98 const safeBuddyList = buddyList.map((user) => createSafeUser(user));117 const safeBuddyList = buddyList.map((user) => createSafeUser(user));
99 118
100- // Update cache and check if data changed
101 const hasChanged = buddyListCache.update(safeBuddyList);119 const hasChanged = buddyListCache.update(safeBuddyList);
102 120
103 if (hasChanged) {121 if (hasChanged) {
104 log.debug('Broadcasting buddy list update (data changed)...');122 log.debug('Broadcasting buddy list update (data changed)...');
105- sseEmitter.emit('sse', {123+ sseEmitter.emit('buddyListUpdate', safeBuddyList);
106- type: 'buddyListUpdate',
107- data: safeBuddyList
108- });
109 }124 }
110 } else {125 } else {
111- // Use cached data
112 const cachedData = buddyListCache.get();126 const cachedData = buddyListCache.get();
113- if (cachedData && JSON.stringify(cachedData) !== JSON.stringify(lastBuddyListData)) {127+ if (cachedData) {
114- log.debug('Broadcasting buddy list update from cache...');128+ const currentHash = JSON.stringify(cachedData);
115- lastBuddyListData = cachedData;129+ if (currentHash !== lastBuddyListHash) {
116- sseEmitter.emit('sse', {130+ log.debug('Broadcasting buddy list update from cache...');
117- type: 'buddyListUpdate',131+ lastBuddyListHash = currentHash;
118- data: cachedData132+ sseEmitter.emit('buddyListUpdate', cachedData);
119- });133+ }
120 }134 }
121 }135 }
122 lastBuddyListUpdate = now;136 lastBuddyListUpdate = now;
@@ -124,72 +138,78 @@ function startBuddyListUpdateInterval() {
124 } catch (error) {138 } catch (error) {
125 log.error('Error fetching buddy list for update', { error });139 log.error('Error fetching buddy list for update', { error });
126 }140 }
127- }, 1000); // Check every second but only broadcast every 10 seconds141+ }, 1000);
128 }142 }
129 143
130-// Add periodic cleanup for stale connections144+/**
131-let connectionCleanupInterval: ReturnType<typeof setInterval>;145+ * Start periodic cleanup for stale connections
132-146+ */
133 function startConnectionCleanup() {147 function startConnectionCleanup() {
134 if (connectionCleanupInterval) {148 if (connectionCleanupInterval) {
135 clearInterval(connectionCleanupInterval);149 clearInterval(connectionCleanupInterval);
136 }150 }
137 151
138- connectionCleanupInterval = setInterval(() => {152+ connectionCleanupInterval = setInterval(async () => {
139- const staleUserIds = sseConnectionTracker.cleanupStaleConnections();153+ // Get stats and log them periodically
140- if (staleUserIds.length > 0) {154+ const stats = sseManager.getStats();
141- // Mark stale connections as offline155+ if (stats.totalConnections > 0) {
142- staleUserIds.forEach(async (userId) => {156+ log.debug('SSE connection stats', stats);
143- try {
144- await db
145- .update(users)
146- .set({ status: 'offline', lastSeen: Date.now() })
147- .where(eq(users.id, userId));
148-
149- buddyListCache.invalidate();
150- } catch (error) {
151- log.error('Error updating stale user status', { userId, error });
152- }
153- });
154 }157 }
155- }, 60000); // Run every minute158+ }, SSE_CONFIG.STALE_CHECK_INTERVAL);
156 }159 }
157 160
158-// <<< NEW: Ensure intervals are started only once >>>161+// Start background tasks only once
159 if (!globalThis.__sseIntervalsStarted) {162 if (!globalThis.__sseIntervalsStarted) {
160 startTimeoutCheck();163 startTimeoutCheck();
161 startBuddyListUpdateInterval();164 startBuddyListUpdateInterval();
162 startConnectionCleanup();165 startConnectionCleanup();
166+ sseManager.startBackgroundTasks();
163 globalThis.__sseIntervalsStarted = true;167 globalThis.__sseIntervalsStarted = true;
164- log.info('Started global intervals for timeout and buddy list updates');168+ log.info('Started global intervals for SSE management');
165 } else {169 } else {
166 log.debug('Global intervals already started, skipping initialization');170 log.debug('Global intervals already started, skipping initialization');
167 }171 }
168 172
169-// Ensure the interval is cleared if the module is reloaded173+// Cleanup on process exit
170 if (typeof process !== 'undefined') {174 if (typeof process !== 'undefined') {
171- process.on('beforeExit', () => {175+ const cleanup = () => {
172 if (timeoutInterval) clearInterval(timeoutInterval);176 if (timeoutInterval) clearInterval(timeoutInterval);
173 if (buddyListInterval) clearInterval(buddyListInterval);177 if (buddyListInterval) clearInterval(buddyListInterval);
174 if (connectionCleanupInterval) clearInterval(connectionCleanupInterval);178 if (connectionCleanupInterval) clearInterval(connectionCleanupInterval);
175- });179+ sseManager.stopBackgroundTasks();
180+ };
176 181
177- // Also handle SIGTERM and SIGINT182+ process.on('beforeExit', cleanup);
178 process.on('SIGTERM', () => {183 process.on('SIGTERM', () => {
179- if (timeoutInterval) clearInterval(timeoutInterval);184+ cleanup();
180- if (buddyListInterval) clearInterval(buddyListInterval);
181- if (connectionCleanupInterval) clearInterval(connectionCleanupInterval);
182 process.exit(0);185 process.exit(0);
183 });186 });
184-
185 process.on('SIGINT', () => {187 process.on('SIGINT', () => {
186- if (timeoutInterval) clearInterval(timeoutInterval);188+ cleanup();
187- if (buddyListInterval) clearInterval(buddyListInterval);
188- if (connectionCleanupInterval) clearInterval(connectionCleanupInterval);
189 process.exit(0);189 process.exit(0);
190 });190 });
191 }191 }
192 192
193+/**
194+ * Format an SSE event with ID and retry field
195+ */
196+function formatSSEEvent(event: SSEEvent, eventId: string): string {
197+ const lines = [
198+ `id: ${eventId}`,
199+ `event: ${event.type}`,
200+ `data: ${JSON.stringify(event.data)}`,
201+ `retry: ${SSE_CONFIG.RETRY_INTERVAL}`,
202+ '',
203+ ''
204+ ];
205+ return lines.join('\n');
206+}
207+
208+/**
209+ * GET /api/sse
210+ *
211+ * Establishes an SSE connection for real-time updates.
212+ */
193 export const GET: RequestHandler = async ({ request, locals }) => {213 export const GET: RequestHandler = async ({ request, locals }) => {
194 log.debug('Processing SSE request', {214 log.debug('Processing SSE request', {
195 url: request.url,215 url: request.url,
@@ -200,119 +220,157 @@ export const GET: RequestHandler = async ({ request, locals }) => {
200 log.warn('Authentication failed - no user in locals');220 log.warn('Authentication failed - no user in locals');
201 return new Response('Unauthorized', {221 return new Response('Unauthorized', {
202 status: 401,222 status: 401,
203- headers: {223+ headers: { 'Content-Type': 'application/json' }
204- 'Content-Type': 'application/json'
205- }
206 });224 });
207 }225 }
208 226
209 const userId = locals.user.id;227 const userId = locals.user.id;
228+ const roomId = DEFAULT_CHAT_ROOM_ID;
210 const encoder = new TextEncoder();229 const encoder = new TextEncoder();
211- let keepAliveInterval: ReturnType<typeof setInterval>;
212- let onSSE: ((event: { type: string; data: unknown }) => void) | undefined;
213 230
214- const now = Date.now();231+ // Check for Last-Event-ID header for catch-up
232+ const lastEventId = request.headers.get('Last-Event-ID');
233+ if (lastEventId) {
234+ log.debug('Client reconnecting with Last-Event-ID', {
235+ userId: userId.slice(0, 8),
236+ lastEventId: lastEventId.slice(0, 8)
237+ });
238+ }
215 239
216- // Only update status if this is a new connection240+ let connectionId: string;
217- const isNewConnection = sseConnectionTracker.addConnection(userId);241+ let keepAliveInterval: ReturnType<typeof setInterval>;
242+ let cleanupListener: (() => void) | undefined;
218 243
219- if (isNewConnection) {244+ const stream = new ReadableStream({
220- try {245+ async start(controller) {
221- await db246+ // Add connection to manager
222- .update(users)247+ const { connectionId: connId, isNewUser } = sseManager.addConnection(
223- .set({248+ userId,
224- status: 'online',249+ roomId,
225- lastSeen: now250+ controller
226- })251+ );
227- .where(eq(users.id, userId));252+ connectionId = connId;
253+
254+ // Update user status if new connection
255+ if (isNewUser) {
256+ const now = Date.now();
257+ try {
258+ await db
259+ .update(users)
260+ .set({ status: 'online', lastSeen: now })
261+ .where(eq(users.id, userId));
228 262
229- log.info('User status updated to online', {263+ log.info('User status updated to online', {
230- userId: `${userId.slice(0, 4)}...${userId.slice(-4)}`,264+ userId: userId.slice(0, 8),
231- timestamp: new Date(now).toISOString()265+ connectionId: connectionId.slice(0, 8)
232- });266+ });
233 267
234- // Invalidate buddy list cache when user status changes268+ buddyListCache.invalidate();
235- buddyListCache.invalidate();
236 269
237- sseEmitter.emit('sse', {270+ sseEmitter.emit('userStatusUpdate', {
238- type: 'userStatusUpdate',271+ userId,
239- data: {272+ status: 'online',
240- userId,273+ lastSeen: now
241- status: 'online',274+ });
242- lastSeen: now275+ } catch (error) {
276+ log.error('Error updating user status', { error });
243 }277 }
244- });278+ }
245- } catch (error) {
246- log.error('Error updating user status', { error });
247- sseConnectionTracker.removeConnection(userId);
248- }
249- } else {
250- log.debug('User already has active SSE connection, skipping status update', {
251- userId: `${userId.slice(0, 4)}...${userId.slice(-4)}`
252- });
253- }
254 279
255- const stream = new ReadableStream({280+ // Send initial connection event
256- start(controller) {281+ const connectEvent = formatSSEEvent(
257- const sendEvent = (event: { type: string; data: unknown }) => {282+ { type: 'userStatusUpdate', data: { userId, status: 'connected', lastSeen: Date.now() } },
283+ crypto.randomUUID()
284+ );
285+ controller.enqueue(encoder.encode(connectEvent));
286+
287+ // Send catch-up messages if reconnecting with Last-Event-ID
288+ if (lastEventId) {
289+ const missedMessages = messageQueue.getMessagesSince(roomId, lastEventId);
290+ if (missedMessages.length > 0) {
291+ log.info('Sending catch-up messages', {
292+ userId: userId.slice(0, 8),
293+ count: missedMessages.length
294+ });
295+ for (const msg of missedMessages) {
296+ // Cast to SSEEvent - the data was already validated when queued
297+ const event = { type: msg.event, data: msg.data } as SSEEvent;
298+ const payload = formatSSEEvent(event, msg.id);
299+ controller.enqueue(encoder.encode(payload));
300+ }
301+ }
302+ }
303+
304+ // Listen for SSE events
305+ const onSSE = (event: SSEEvent) => {
258 try {306 try {
259- const payload = `event: ${event.type}\n` + `data: ${JSON.stringify(event.data)}\n\n`;307+ const eventId = crypto.randomUUID();
308+ const payload = formatSSEEvent(event, eventId);
260 controller.enqueue(encoder.encode(payload));309 controller.enqueue(encoder.encode(payload));
261- if (!['userStatusUpdate', 'chatMessage'].includes(event.type)) {310+
262- log.info('Event sent', { type: event.type });311+ // Queue message for catch-up (if not heartbeat)
312+ if (event.type !== 'heartbeat') {
313+ messageQueue.enqueue(roomId, event.type, event.data);
263 }314 }
264 } catch (error) {315 } catch (error) {
265 log.error('Error sending event', { error });316 log.error('Error sending event', { error });
266 }317 }
267 };318 };
268 319
269- onSSE = (event: { type: string; data: unknown }) => {320+ cleanupListener = sseEmitter.addListener('sse', onSSE);
270- sendEvent(event);
271- };
272 321
273- sseEmitter.addListener('sse', onSSE);322+ // Handle abort signal
274 request.signal.addEventListener('abort', () => {323 request.signal.addEventListener('abort', () => {
275- if (onSSE) {324+ if (cleanupListener) {
276- sseEmitter.removeListener('sse', onSSE);325+ cleanupListener();
277 log.info('Removed listener due to abort signal');326 log.info('Removed listener due to abort signal');
278 }327 }
279 });328 });
280- log.info('Added new listener');
281 329
282- controller.enqueue(encoder.encode('data: Connected\n\n'));330+ log.info('SSE connection established', {
331+ connectionId: connectionId.slice(0, 8),
332+ userId: userId.slice(0, 8)
333+ });
283 334
335+ // Keep-alive pings (in addition to heartbeats from manager)
284 keepAliveInterval = setInterval(() => {336 keepAliveInterval = setInterval(() => {
285- controller.enqueue(encoder.encode(':\n\n'));337+ try {
286- }, 20000);338+ controller.enqueue(encoder.encode(':\n\n'));
339+ } catch {
340+ // Connection might be closed
341+ }
342+ }, SSE_CONFIG.HEARTBEAT_INTERVAL);
287 },343 },
344+
288 async cancel(reason) {345 async cancel(reason) {
289 if (keepAliveInterval) clearInterval(keepAliveInterval);346 if (keepAliveInterval) clearInterval(keepAliveInterval);
290- if (onSSE) sseEmitter.removeListener('sse', onSSE);347+ if (cleanupListener) cleanupListener();
291- const maskedUserId = `${userId.slice(0, 4)}...${userId.slice(-4)}`;
292- log.info('Connection closed', { userId: maskedUserId, reason });
293 348
294- // Remove from connection tracker349+ const maskedUserId = userId.slice(0, 8);
295- const wasTracked = sseConnectionTracker.removeConnection(userId);350+ log.info('Connection closed', { userId: maskedUserId, connectionId: connectionId?.slice(0, 8), reason });
296 351
297- // Only update database if this was the last connection352+ // Remove connection from manager
298- if (wasTracked && !sseConnectionTracker.hasConnection(userId)) {353+ const { isLastConnection } = sseManager.removeConnection(connectionId);
354+
355+ // Update database if last connection for this user
356+ if (isLastConnection) {
299 const now = Date.now();357 const now = Date.now();
300- await db358+ try {
301- .update(users)359+ await db
302- .set({ status: 'offline', lastSeen: now })360+ .update(users)
303- .where(eq(users.id, userId));361+ .set({ status: 'offline', lastSeen: now })
362+ .where(eq(users.id, userId));
304 363
305- // Invalidate buddy list cache when user goes offline364+ buddyListCache.invalidate();
306- buddyListCache.invalidate();
307 365
308- sseEmitter.emit('sse', {366+ sseEmitter.emit('userStatusUpdate', {
309- type: 'userStatusUpdate',
310- data: {
311 userId,367 userId,
312 status: 'offline',368 status: 'offline',
313 lastSeen: now369 lastSeen: now
314- }370+ });
315- });371+ } catch (error) {
372+ log.error('Error updating user status on disconnect', { error });
373+ }
316 }374 }
317 }375 }
318 });376 });
@@ -321,7 +379,8 @@ export const GET: RequestHandler = async ({ request, locals }) => {
321 headers: {379 headers: {
322 'Content-Type': 'text/event-stream',380 'Content-Type': 'text/event-stream',
323 'Cache-Control': 'no-cache',381 'Cache-Control': 'no-cache',
324- Connection: 'keep-alive'382+ 'Connection': 'keep-alive',
383+ 'X-Accel-Buffering': 'no' // Disable nginx buffering
325 }384 }
326 });385 });
327 };386 };
A src/routes/api/sse/heartbeat/+server.ts +66 -0
@@ -0,0 +1,66 @@
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+};
@@ -0,0 +1,66 @@
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