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 } from 'drizzle-orm';
12import { createSafeUser } from '$lib/types/chat';
13import { createLogger } from '$lib/utils/logger.server';
14
15const log = createLogger('sse-server');
16
17console.log('[SSE] Server initialized');
18
19// Constants for timeout checks
20const ONLINE_TIMEOUT_MS = 2 * 60 * 1000; // 2 minutes
21const CHECK_INTERVAL_MS = 30 * 1000; // Check every 30 seconds
22
23// Update the interval constant
24const BUDDY_LIST_BROADCAST_INTERVAL = 10000; // 10 seconds
25
26let timeoutInterval: ReturnType<typeof setInterval>;
27let buddyListInterval: ReturnType<typeof setInterval>;
28
29// Add a timestamp tracker for buddy list updates
30let lastBuddyListUpdate = 0;
31let lastBuddyListData: unknown[] = [];
32
33// Initialize periodic check for timed out users
34function startTimeoutCheck() {
35 if (timeoutInterval) {
36 clearInterval(timeoutInterval);
37 }
38
39 timeoutInterval = setInterval(async () => {
40 const timeoutThreshold = Date.now() - ONLINE_TIMEOUT_MS;
41 log.debug('Checking for timed out users', {
42 checkTime: new Date().toISOString(),
43 timeoutThreshold: new Date(timeoutThreshold).toISOString()
44 });
45
46 try {
47 const now = Date.now();
48 const result = await db.update(users)
49 .set({
50 status: 'offline',
51 lastSeen: now
52 })
53 .where(
54 and(
55 eq(users.status, 'online'), // Only update online users
56 lt(users.lastSeen, timeoutThreshold) // Who haven't been seen recently
57 )
58 )
59 .returning({
60 id: users.id,
61 nickname: users.nickname,
62 lastSeen: users.lastSeen
63 });
64
65 if (result.length > 0) {
66 log.info('User status updated to offline', {
67 usersUpdated: result.length,
68 timeoutThreshold: new Date(timeoutThreshold).toISOString()
69 });
70 }
71 } catch (error) {
72 log.error('Error during timeout check', { error });
73 }
74 }, CHECK_INTERVAL_MS);
75}
76
77function startBuddyListUpdateInterval() {
78 if (buddyListInterval) {
79 clearInterval(buddyListInterval);
80 }
81
82 buddyListInterval = setInterval(async () => {
83 try {
84 const now = Date.now();
85 // Only fetch and broadcast if 10 seconds have passed since last update
86 if (now - lastBuddyListUpdate >= BUDDY_LIST_BROADCAST_INTERVAL) {
87 log.debug('Broadcasting buddy list update...');
88 // Query the complete buddy list from the database.
89 const buddyList = await db.select().from(users);
90 // Use createSafeUser to properly sanitize user data
91 const safeBuddyList = buddyList.map(user => createSafeUser(user));
92
93 // Only broadcast if data has changed
94 if (JSON.stringify(safeBuddyList) !== JSON.stringify(lastBuddyListData)) {
95 lastBuddyListData = safeBuddyList;
96 sseEmitter.emit('sse', {
97 type: 'buddyListUpdate',
98 data: safeBuddyList
99 });
100 }
101 lastBuddyListUpdate = now;
102 }
103 } catch (error) {
104 log.error('Error fetching buddy list for update', { error });
105 }
106 }, 1000); // Check every second but only broadcast every 10 seconds
107}
108
109// <<< NEW: Ensure intervals are started only once >>>
110if (!globalThis.__sseIntervalsStarted) {
111 startTimeoutCheck();
112 startBuddyListUpdateInterval();
113 globalThis.__sseIntervalsStarted = true;
114 log.info('Started global intervals for timeout and buddy list updates');
115} else {
116 log.debug('Global intervals already started, skipping initialization');
117}
118
119// Ensure the interval is cleared if the module is reloaded
120if (typeof process !== 'undefined') {
121 process.on('beforeExit', () => {
122 if (timeoutInterval) clearInterval(timeoutInterval);
123 if (buddyListInterval) clearInterval(buddyListInterval);
124 });
125
126 // Also handle SIGTERM and SIGINT
127 process.on('SIGTERM', () => {
128 if (timeoutInterval) clearInterval(timeoutInterval);
129 if (buddyListInterval) clearInterval(buddyListInterval);
130 process.exit(0);
131 });
132
133 process.on('SIGINT', () => {
134 if (timeoutInterval) clearInterval(timeoutInterval);
135 if (buddyListInterval) clearInterval(buddyListInterval);
136 process.exit(0);
137 });
138}
139
140export const GET: RequestHandler = async ({ request, locals, }) => {
141 log.debug('Processing SSE request', {
142 url: request.url,
143 userId: locals.user?.id
144 });
145
146 if (!locals.user) {
147 log.warn('Authentication failed - no user in locals');
148 return new Response('Unauthorized', {
149 status: 401,
150 headers: {
151 'Content-Type': 'application/json'
152 }
153 });
154 }
155
156 const userId = locals.user.id;
157 const encoder = new TextEncoder();
158 let keepAliveInterval: ReturnType<typeof setInterval>;
159 let onSSE: ((event: { type: string; data: unknown }) => void) | undefined;
160
161 const now = Date.now();
162
163 try {
164 await db.update(users)
165 .set({
166 status: 'online',
167 lastSeen: now
168 })
169 .where(eq(users.id, userId));
170
171 log.info('User status updated to online', {
172 userId: `${userId.slice(0, 4)}...${userId.slice(-4)}`,
173 timestamp: new Date(now).toISOString()
174 });
175
176 sseEmitter.emit('sse', {
177 type: 'userStatusUpdate',
178 data: {
179 userId,
180 status: 'online',
181 lastSeen: now
182 }
183 });
184 } catch (error) {
185 log.error('Error updating user status', { error });
186 }
187
188 const stream = new ReadableStream({
189 start(controller) {
190 const sendEvent = (event: { type: string; data: unknown }) => {
191 try {
192 const payload = `event: ${event.type}\n` +
193 `data: ${JSON.stringify(event.data)}\n\n`;
194 controller.enqueue(encoder.encode(payload));
195 if (!['userStatusUpdate', 'chatMessage'].includes(event.type)) {
196 log.info('Event sent', { type: event.type });
197 }
198 } catch (error) {
199 log.error('Error sending event', { error });
200 }
201 };
202
203 onSSE = (event: { type: string; data: unknown }) => {
204 sendEvent(event);
205 };
206
207 sseEmitter.addListener('sse', onSSE);
208 request.signal.addEventListener('abort', () => {
209 if (onSSE) {
210 sseEmitter.removeListener('sse', onSSE);
211 log.info('Removed listener due to abort signal');
212 }
213 });
214 log.info('Added new listener');
215
216 controller.enqueue(encoder.encode('data: Connected\n\n'));
217
218 keepAliveInterval = setInterval(() => {
219 controller.enqueue(encoder.encode(':\n\n'));
220 }, 20000);
221 },
222 async cancel(reason) {
223 if (keepAliveInterval) clearInterval(keepAliveInterval);
224 if (onSSE) sseEmitter.removeListener('sse', onSSE);
225 const maskedUserId = `${userId.slice(0, 4)}...${userId.slice(-4)}`;
226 log.info('Connection closed', { userId: maskedUserId, reason });
227
228 const now = Date.now();
229 await db.update(users)
230 .set({ status: 'offline', lastSeen: now })
231 .where(eq(users.id, userId));
232
233 sseEmitter.emit('sse', {
234 type: 'userStatusUpdate',
235 data: {
236 userId,
237 status: 'offline',
238 lastSeen: now
239 }
240 });
241 }
242 });
243
244 return new Response(stream, {
245 headers: {
246 'Content-Type': 'text/event-stream',
247 'Cache-Control': 'no-cache',
248 'Connection': 'keep-alive'
249 }
250 });
251};