1// At the top of the file, before any code
2declare global {
3 // eslint-disable-next-line no-var
4 var __sseIntervalsStarted: boolean;
5}
6
7import type { RequestHandler } from './$types';
8import { sseEmitter } from '$lib/sseEmitter';
9import db from '$lib/db/db.server';
10import { users } from '$lib/db/schema';
11import { eq, and, lt, ne } from 'drizzle-orm';
12import { createSafeUser } from '$lib/types/chat';
13import { createLogger } from '$lib/utils/logger.server';
14import { sseConnectionTracker } from '$lib/sseConnectionTracker';
15import { buddyListCache } from '$lib/buddyListCache';
16
17const log = createLogger('sse-server');
18
19console.log('[SSE] Server initialized');
20
21// Constants for timeout checks
22const ONLINE_TIMEOUT_MS = 2 * 60 * 1000; // 2 minutes
23const CHECK_INTERVAL_MS = 30 * 1000; // Check every 30 seconds
24
25// Update the interval constant
26const BUDDY_LIST_BROADCAST_INTERVAL = 10000; // 10 seconds
27
28let timeoutInterval: ReturnType<typeof setInterval>;
29let buddyListInterval: ReturnType<typeof setInterval>;
30
31// Add a timestamp tracker for buddy list updates
32let lastBuddyListUpdate = 0;
33let lastBuddyListData: unknown[] = [];
34
35// Initialize periodic check for timed out users
36function startTimeoutCheck() {
37 if (timeoutInterval) {
38 clearInterval(timeoutInterval);
39 }
40
41 timeoutInterval = setInterval(async () => {
42 const timeoutThreshold = Date.now() - ONLINE_TIMEOUT_MS;
43 log.debug('Checking for timed out users', {
44 checkTime: new Date().toISOString(),
45 timeoutThreshold: new Date(timeoutThreshold).toISOString()
46 });
47
48 try {
49 const now = Date.now();
50 const result = await db.update(users)
51 .set({
52 status: 'offline',
53 lastSeen: now
54 })
55 .where(
56 and(
57 ne(users.status, 'offline'), // Update any non-offline users
58 lt(users.lastSeen, timeoutThreshold) // Who haven't been seen recently
59 )
60 )
61 .returning({
62 id: users.id,
63 nickname: users.nickname,
64 lastSeen: users.lastSeen
65 });
66
67 if (result.length > 0) {
68 log.info('User status updated to offline', {
69 usersUpdated: result.length,
70 timeoutThreshold: new Date(timeoutThreshold).toISOString()
71 });
72
73 // Invalidate buddy list cache when users timeout
74 buddyListCache.invalidate();
75 }
76 } catch (error) {
77 log.error('Error during timeout check', { error });
78 }
79 }, CHECK_INTERVAL_MS);
80}
81
82function startBuddyListUpdateInterval() {
83 if (buddyListInterval) {
84 clearInterval(buddyListInterval);
85 }
86
87 buddyListInterval = setInterval(async () => {
88 try {
89 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
93 if (buddyListCache.needsRefresh()) {
94 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);
97 // Use createSafeUser to properly sanitize user data
98 const safeBuddyList = buddyList.map(user => createSafeUser(user));
99
100 // Update cache and check if data changed
101 const hasChanged = buddyListCache.update(safeBuddyList);
102
103 if (hasChanged) {
104 log.debug('Broadcasting buddy list update (data changed)...');
105 sseEmitter.emit('sse', {
106 type: 'buddyListUpdate',
107 data: safeBuddyList
108 });
109 }
110 } else {
111 // Use cached data
112 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 });
120 }
121 }
122 lastBuddyListUpdate = now;
123 }
124 } catch (error) {
125 log.error('Error fetching buddy list for update', { error });
126 }
127 }, 1000); // Check every second but only broadcast every 10 seconds
128}
129
130// Add periodic cleanup for stale connections
131let connectionCleanupInterval: ReturnType<typeof setInterval>;
132
133function startConnectionCleanup() {
134 if (connectionCleanupInterval) {
135 clearInterval(connectionCleanupInterval);
136 }
137
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.update(users)
145 .set({ status: 'offline', lastSeen: Date.now() })
146 .where(eq(users.id, userId));
147
148 buddyListCache.invalidate();
149 } catch (error) {
150 log.error('Error updating stale user status', { userId, error });
151 }
152 });
153 }
154 }, 60000); // Run every minute
155}
156
157// <<< NEW: Ensure intervals are started only once >>>
158if (!globalThis.__sseIntervalsStarted) {
159 startTimeoutCheck();
160 startBuddyListUpdateInterval();
161 startConnectionCleanup();
162 globalThis.__sseIntervalsStarted = true;
163 log.info('Started global intervals for timeout and buddy list updates');
164} else {
165 log.debug('Global intervals already started, skipping initialization');
166}
167
168// Ensure the interval is cleared if the module is reloaded
169if (typeof process !== 'undefined') {
170 process.on('beforeExit', () => {
171 if (timeoutInterval) clearInterval(timeoutInterval);
172 if (buddyListInterval) clearInterval(buddyListInterval);
173 if (connectionCleanupInterval) clearInterval(connectionCleanupInterval);
174 });
175
176 // Also handle SIGTERM and SIGINT
177 process.on('SIGTERM', () => {
178 if (timeoutInterval) clearInterval(timeoutInterval);
179 if (buddyListInterval) clearInterval(buddyListInterval);
180 if (connectionCleanupInterval) clearInterval(connectionCleanupInterval);
181 process.exit(0);
182 });
183
184 process.on('SIGINT', () => {
185 if (timeoutInterval) clearInterval(timeoutInterval);
186 if (buddyListInterval) clearInterval(buddyListInterval);
187 if (connectionCleanupInterval) clearInterval(connectionCleanupInterval);
188 process.exit(0);
189 });
190}
191
192export const GET: RequestHandler = async ({ request, locals, }) => {
193 log.debug('Processing SSE request', {
194 url: request.url,
195 userId: locals.user?.id
196 });
197
198 if (!locals.user) {
199 log.warn('Authentication failed - no user in locals');
200 return new Response('Unauthorized', {
201 status: 401,
202 headers: {
203 'Content-Type': 'application/json'
204 }
205 });
206 }
207
208 const userId = locals.user.id;
209 const encoder = new TextEncoder();
210 let keepAliveInterval: ReturnType<typeof setInterval>;
211 let onSSE: ((event: { type: string; data: unknown }) => void) | undefined;
212
213 const now = Date.now();
214
215 // Only update status if this is a new connection
216 const isNewConnection = sseConnectionTracker.addConnection(userId);
217
218 if (isNewConnection) {
219 try {
220 await db.update(users)
221 .set({
222 status: 'online',
223 lastSeen: now
224 })
225 .where(eq(users.id, userId));
226
227 log.info('User status updated to online', {
228 userId: `${userId.slice(0, 4)}...${userId.slice(-4)}`,
229 timestamp: new Date(now).toISOString()
230 });
231
232 // Invalidate buddy list cache when user status changes
233 buddyListCache.invalidate();
234
235 sseEmitter.emit('sse', {
236 type: 'userStatusUpdate',
237 data: {
238 userId,
239 status: 'online',
240 lastSeen: now
241 }
242 });
243 } catch (error) {
244 log.error('Error updating user status', { error });
245 sseConnectionTracker.removeConnection(userId);
246 }
247 } else {
248 log.debug('User already has active SSE connection, skipping status update', {
249 userId: `${userId.slice(0, 4)}...${userId.slice(-4)}`
250 });
251 }
252
253 const stream = new ReadableStream({
254 start(controller) {
255 const sendEvent = (event: { type: string; data: unknown }) => {
256 try {
257 const payload = `event: ${event.type}\n` +
258 `data: ${JSON.stringify(event.data)}\n\n`;
259 controller.enqueue(encoder.encode(payload));
260 if (!['userStatusUpdate', 'chatMessage'].includes(event.type)) {
261 log.info('Event sent', { type: event.type });
262 }
263 } catch (error) {
264 log.error('Error sending event', { error });
265 }
266 };
267
268 onSSE = (event: { type: string; data: unknown }) => {
269 sendEvent(event);
270 };
271
272 sseEmitter.addListener('sse', onSSE);
273 request.signal.addEventListener('abort', () => {
274 if (onSSE) {
275 sseEmitter.removeListener('sse', onSSE);
276 log.info('Removed listener due to abort signal');
277 }
278 });
279 log.info('Added new listener');
280
281 controller.enqueue(encoder.encode('data: Connected\n\n'));
282
283 keepAliveInterval = setInterval(() => {
284 controller.enqueue(encoder.encode(':\n\n'));
285 }, 20000);
286 },
287 async cancel(reason) {
288 if (keepAliveInterval) clearInterval(keepAliveInterval);
289 if (onSSE) sseEmitter.removeListener('sse', onSSE);
290 const maskedUserId = `${userId.slice(0, 4)}...${userId.slice(-4)}`;
291 log.info('Connection closed', { userId: maskedUserId, reason });
292
293 // Remove from connection tracker
294 const wasTracked = sseConnectionTracker.removeConnection(userId);
295
296 // Only update database if this was the last connection
297 if (wasTracked && !sseConnectionTracker.hasConnection(userId)) {
298 const now = Date.now();
299 await db.update(users)
300 .set({ status: 'offline', lastSeen: now })
301 .where(eq(users.id, userId));
302
303 // Invalidate buddy list cache when user goes offline
304 buddyListCache.invalidate();
305
306 sseEmitter.emit('sse', {
307 type: 'userStatusUpdate',
308 data: {
309 userId,
310 status: 'offline',
311 lastSeen: now
312 }
313 });
314 }
315 }
316 });
317
318 return new Response(stream, {
319 headers: {
320 'Content-Type': 'text/event-stream',
321 'Cache-Control': 'no-cache',
322 'Connection': 'keep-alive'
323 }
324 });
325};