feat: Improve SSE connection resilience and error handling
- Add connection timeout and reconnection management for SSE - Implement exponential backoff for reconnection attempts - Enhance error handling with more robust connection state tracking - Add logging for connection status and reconnection events - Prevent multiple simultaneous reconnection attempts - Improve user status timeout detection and reporting
- commit
- d24297160343724d995eed5de3b783757280bdf0
- author
- Kit · · 2025-02-14 14:04 UTC
- parent
- cce43cf
- signature
- SSH Unverified
SHA256:H91DFuEG4jpk5edzLA4aUKkpv4zv9Q339ipRmkph6TkThe key is unknown
3 files · +137 -62
M src/lib/states/chat.svelte.ts +91 -48
| @@ -21,6 +21,8 @@ class ChatState { | |||
| 21 | 21 | private userCache = $state<Record<string, User>>({}); | |
| 22 | 22 | private sseError = $state<string | null>(null); | |
| 23 | 23 | private sseRetryAfter = $state<number | null>(null); | |
| 24 | + | private connectionTimeout: ReturnType<typeof setTimeout> | null = null; | |
| 25 | + | private isReconnecting = $state(false); | |
| 24 | 26 | ||
| 25 | 27 | public async reinitialize() { | |
| 26 | 28 | console.debug('Reinitializing chat state...'); | |
| @@ -355,68 +357,95 @@ class ChatState { | |||
| 355 | 357 | ||
| 356 | 358 | private setupSSE() { | |
| 357 | 359 | console.debug('Setting up SSE connection for chat messages'); | |
| 358 | - | this.connectSSE(); | |
| 360 | + | // Clear any existing connection timeout | |
| 361 | + | if (this.connectionTimeout) { | |
| 362 | + | clearTimeout(this.connectionTimeout); | |
| 363 | + | this.connectionTimeout = null; | |
| 364 | + | } | |
| 365 | + | ||
| 366 | + | // Add a small delay before connecting to prevent rapid reconnection attempts | |
| 367 | + | this.connectionTimeout = setTimeout(() => { | |
| 368 | + | this.connectSSE(); | |
| 369 | + | }, 500); | |
| 359 | 370 | } | |
| 360 | 371 | ||
| 361 | 372 | private connectSSE() { | |
| 362 | 373 | if (this.eventSource) { | |
| 363 | 374 | this.eventSource.close(); | |
| 375 | + | this.eventSource = null; | |
| 364 | 376 | } | |
| 377 | + | ||
| 378 | + | if (this.isReconnecting) { | |
| 379 | + | console.debug('Already attempting to reconnect, skipping new connection attempt'); | |
| 380 | + | return; | |
| 381 | + | } | |
| 382 | + | ||
| 383 | + | this.isReconnecting = true; | |
| 384 | + | console.debug('Connecting to SSE...'); | |
| 365 | 385 | ||
| 366 | - | // Connect to the unified SSE endpoint | |
| 367 | - | this.eventSource = new EventSource('/api/sse', { withCredentials: true }); | |
| 368 | - | ||
| 369 | - | // Default message event for handshake | |
| 370 | - | this.eventSource.onmessage = (event) => { | |
| 371 | - | if (event.data === 'Connected') { | |
| 372 | - | console.debug('SSE connection established'); | |
| 373 | - | this.reconnectAttempts = 0; | |
| 374 | - | this.reconnectDelay = 1000; | |
| 375 | - | } else { | |
| 376 | - | console.debug('Received unexpected default message:', event.data); | |
| 377 | - | } | |
| 378 | - | }; | |
| 386 | + | try { | |
| 387 | + | // Connect to the unified SSE endpoint | |
| 388 | + | this.eventSource = new EventSource('/api/sse', { withCredentials: true }); | |
| 389 | + | ||
| 390 | + | // Default message event for handshake | |
| 391 | + | this.eventSource.onmessage = (event) => { | |
| 392 | + | if (event.data === 'Connected') { | |
| 393 | + | console.debug('SSE connection established'); | |
| 394 | + | this.reconnectAttempts = 0; | |
| 395 | + | this.reconnectDelay = 1000; | |
| 396 | + | this.isReconnecting = false; | |
| 397 | + | this.sseError = null; | |
| 398 | + | this.sseRetryAfter = null; | |
| 399 | + | } else { | |
| 400 | + | console.debug('Received unexpected default message:', event.data); | |
| 401 | + | } | |
| 402 | + | }; | |
| 379 | 403 | ||
| 380 | - | // Listen for chat messages | |
| 381 | - | this.eventSource.addEventListener('chatMessage', async (event: MessageEvent) => { | |
| 382 | - | try { | |
| 383 | - | const messageData = JSON.parse(event.data) as Message; | |
| 384 | - | // Ensure the sender's data is available | |
| 385 | - | if (!this.userCache[messageData.senderId]) { | |
| 386 | - | console.debug('Fetching user data for new message:', messageData.senderId); | |
| 387 | - | const response = await fetch(`/api/users/${messageData.senderId}`, { credentials: 'include' }); | |
| 388 | - | if (response.ok) { | |
| 389 | - | const userData = await response.json(); | |
| 390 | - | if (userData.success && userData.user) { | |
| 391 | - | this.userCache[userData.user.id] = userData.user; | |
| 392 | - | if (!this.users.some(u => u.id === userData.user.id)) { | |
| 393 | - | this.users = [...this.users, userData.user]; | |
| 404 | + | // Listen for chat messages | |
| 405 | + | this.eventSource.addEventListener('chatMessage', async (event: MessageEvent) => { | |
| 406 | + | try { | |
| 407 | + | const messageData = JSON.parse(event.data) as Message; | |
| 408 | + | // Ensure the sender's data is available | |
| 409 | + | if (!this.userCache[messageData.senderId]) { | |
| 410 | + | console.debug('Fetching user data for new message:', messageData.senderId); | |
| 411 | + | const response = await fetch(`/api/users/${messageData.senderId}`, { credentials: 'include' }); | |
| 412 | + | if (response.ok) { | |
| 413 | + | const userData = await response.json(); | |
| 414 | + | if (userData.success && userData.user) { | |
| 415 | + | this.userCache[userData.user.id] = userData.user; | |
| 416 | + | if (!this.users.some(u => u.id === userData.user.id)) { | |
| 417 | + | this.users = [...this.users, userData.user]; | |
| 418 | + | } | |
| 394 | 419 | } | |
| 395 | 420 | } | |
| 396 | 421 | } | |
| 422 | + | this.messages = [...this.messages, messageData]; | |
| 423 | + | await invalidate('chat:messages'); | |
| 424 | + | } catch (error) { | |
| 425 | + | console.debug('Error parsing chatMessage SSE data:', error); | |
| 397 | 426 | } | |
| 398 | - | this.messages = [...this.messages, messageData]; | |
| 399 | - | await invalidate('chat:messages'); | |
| 400 | - | } catch (error) { | |
| 401 | - | console.debug('Error parsing chatMessage SSE data:', error); | |
| 402 | - | } | |
| 403 | - | }); | |
| 427 | + | }); | |
| 404 | 428 | ||
| 405 | - | // Listen for user status updates | |
| 406 | - | this.eventSource.addEventListener('userStatusUpdate', (event: MessageEvent) => { | |
| 407 | - | try { | |
| 408 | - | const userUpdate = JSON.parse(event.data); | |
| 409 | - | console.debug('Received SSE user status update:', userUpdate); | |
| 410 | - | this.updateUserStatus(userUpdate.userId, userUpdate.status, userUpdate.lastSeen); | |
| 411 | - | } catch (error) { | |
| 412 | - | console.debug('Error processing userStatusUpdate event:', error); | |
| 413 | - | } | |
| 414 | - | }); | |
| 429 | + | // Listen for user status updates | |
| 430 | + | this.eventSource.addEventListener('userStatusUpdate', (event: MessageEvent) => { | |
| 431 | + | try { | |
| 432 | + | const userUpdate = JSON.parse(event.data); | |
| 433 | + | console.debug('Received SSE user status update:', userUpdate); | |
| 434 | + | this.updateUserStatus(userUpdate.userId, userUpdate.status, userUpdate.lastSeen); | |
| 435 | + | } catch (error) { | |
| 436 | + | console.debug('Error processing userStatusUpdate event:', error); | |
| 437 | + | } | |
| 438 | + | }); | |
| 415 | 439 | ||
| 416 | - | this.eventSource.onerror = (event) => { | |
| 417 | - | console.debug('SSE encountered an error:', event); | |
| 440 | + | this.eventSource.onerror = (event) => { | |
| 441 | + | console.debug('SSE encountered an error:', event); | |
| 442 | + | this.handleSSEError(); | |
| 443 | + | }; | |
| 444 | + | } catch (error) { | |
| 445 | + | console.debug('Error setting up SSE connection:', error); | |
| 446 | + | this.isReconnecting = false; | |
| 418 | 447 | this.handleSSEError(); | |
| 419 | - | }; | |
| 448 | + | } | |
| 420 | 449 | } | |
| 421 | 450 | ||
| 422 | 451 | private async handleSSEError() { | |
| @@ -425,6 +454,12 @@ class ChatState { | |||
| 425 | 454 | this.eventSource = null; | |
| 426 | 455 | } | |
| 427 | 456 | ||
| 457 | + | // If we're already reconnecting, don't start another reconnection attempt | |
| 458 | + | if (this.isReconnecting) { | |
| 459 | + | console.debug('Already attempting to reconnect, skipping error handler'); | |
| 460 | + | return; | |
| 461 | + | } | |
| 462 | + | ||
| 428 | 463 | // Check if the error was due to rate limiting | |
| 429 | 464 | try { | |
| 430 | 465 | const response = await fetch('/api/sse'); | |
| @@ -432,13 +467,14 @@ class ChatState { | |||
| 432 | 467 | const data = await response.json(); | |
| 433 | 468 | this.sseError = data.error; | |
| 434 | 469 | this.sseRetryAfter = data.retryAfter; | |
| 470 | + | this.isReconnecting = false; | |
| 435 | 471 | ||
| 436 | 472 | // Set up automatic retry after the rate limit expires | |
| 437 | 473 | setTimeout(() => { | |
| 438 | 474 | this.sseError = null; | |
| 439 | 475 | this.sseRetryAfter = null; | |
| 440 | 476 | this.reconnectSSE(); | |
| 441 | - | }, data.retryAfter * 1000); | |
| 477 | + | }, (data.retryAfter + 1) * 1000); // Add 1 second buffer | |
| 442 | 478 | ||
| 443 | 479 | return; | |
| 444 | 480 | } | |
| @@ -454,15 +490,22 @@ class ChatState { | |||
| 454 | 490 | setTimeout(() => { | |
| 455 | 491 | this.reconnectAttempts++; | |
| 456 | 492 | this.reconnectDelay *= 2; // Exponential backoff | |
| 493 | + | this.isReconnecting = false; | |
| 457 | 494 | this.connectSSE(); | |
| 458 | 495 | }, this.reconnectDelay); | |
| 459 | 496 | } else { | |
| 497 | + | this.isReconnecting = false; | |
| 460 | 498 | this.sseError = 'Unable to connect to chat. Please refresh the page to try again.'; | |
| 461 | 499 | console.debug('Max reconnection attempts reached'); | |
| 462 | 500 | } | |
| 463 | 501 | } | |
| 464 | 502 | ||
| 465 | 503 | private reconnectSSE() { | |
| 504 | + | if (this.isReconnecting) { | |
| 505 | + | console.debug('Already attempting to reconnect, skipping reconnection request'); | |
| 506 | + | return; | |
| 507 | + | } | |
| 508 | + | ||
| 466 | 509 | this.reconnectAttempts = 0; | |
| 467 | 510 | this.reconnectDelay = 1000; | |
| 468 | 511 | this.sseError = null; | |
| @@ -21,6 +21,8 @@ class ChatState { | |||||
| 21 | private userCache = $state<Record<string, User>>({}); | 21 | private userCache = $state<Record<string, User>>({}); | ||
| 22 | private sseError = $state<string | null>(null); | 22 | private sseError = $state<string | null>(null); | ||
| 23 | private sseRetryAfter = $state<number | null>(null); | 23 | private sseRetryAfter = $state<number | null>(null); | ||
| 24 | + | private connectionTimeout: ReturnType<typeof setTimeout> | null = null; | |||
| 25 | + | private isReconnecting = $state(false); | |||
| 24 | 26 | ||||
| 25 | public async reinitialize() { | 27 | public async reinitialize() { | ||
| 26 | console.debug('Reinitializing chat state...'); | 28 | console.debug('Reinitializing chat state...'); | ||
| @@ -355,68 +357,95 @@ class ChatState { | |||||
| 355 | 357 | ||||
| 356 | private setupSSE() { | 358 | private setupSSE() { | ||
| 357 | console.debug('Setting up SSE connection for chat messages'); | 359 | console.debug('Setting up SSE connection for chat messages'); | ||
| 358 | - | this.connectSSE(); | 360 | + | // Clear any existing connection timeout |
| 361 | + | if (this.connectionTimeout) { | |||
| 362 | + | clearTimeout(this.connectionTimeout); | |||
| 363 | + | this.connectionTimeout = null; | |||
| 364 | + | } | |||
| 365 | + | ||||
| 366 | + | // Add a small delay before connecting to prevent rapid reconnection attempts | |||
| 367 | + | this.connectionTimeout = setTimeout(() => { | |||
| 368 | + | this.connectSSE(); | |||
| 369 | + | }, 500); | |||
| 359 | } | 370 | } | ||
| 360 | 371 | ||||
| 361 | private connectSSE() { | 372 | private connectSSE() { | ||
| 362 | if (this.eventSource) { | 373 | if (this.eventSource) { | ||
| 363 | this.eventSource.close(); | 374 | this.eventSource.close(); | ||
| 375 | + | this.eventSource = null; | |||
| 364 | } | 376 | } | ||
| 377 | + | ||||
| 378 | + | if (this.isReconnecting) { | |||
| 379 | + | console.debug('Already attempting to reconnect, skipping new connection attempt'); | |||
| 380 | + | return; | |||
| 381 | + | } | |||
| 382 | + | ||||
| 383 | + | this.isReconnecting = true; | |||
| 384 | + | console.debug('Connecting to SSE...'); | |||
| 365 | 385 | ||||
| 366 | - | // Connect to the unified SSE endpoint | 386 | + | try { |
| 367 | - | this.eventSource = new EventSource('/api/sse', { withCredentials: true }); | 387 | + | // Connect to the unified SSE endpoint |
| 368 | - | 388 | + | this.eventSource = new EventSource('/api/sse', { withCredentials: true }); | |
| 369 | - | // Default message event for handshake | 389 | + | |
| 370 | - | this.eventSource.onmessage = (event) => { | 390 | + | // Default message event for handshake |
| 371 | - | if (event.data === 'Connected') { | 391 | + | this.eventSource.onmessage = (event) => { |
| 372 | - | console.debug('SSE connection established'); | 392 | + | if (event.data === 'Connected') { |
| 373 | - | this.reconnectAttempts = 0; | 393 | + | console.debug('SSE connection established'); |
| 374 | - | this.reconnectDelay = 1000; | 394 | + | this.reconnectAttempts = 0; |
| 375 | - | } else { | 395 | + | this.reconnectDelay = 1000; |
| 376 | - | console.debug('Received unexpected default message:', event.data); | 396 | + | this.isReconnecting = false; |
| 377 | - | } | 397 | + | this.sseError = null; |
| 378 | - | }; | 398 | + | this.sseRetryAfter = null; |
| 399 | + | } else { | |||
| 400 | + | console.debug('Received unexpected default message:', event.data); | |||
| 401 | + | } | |||
| 402 | + | }; | |||
| 379 | 403 | ||||
| 380 | - | // Listen for chat messages | 404 | + | // Listen for chat messages |
| 381 | - | this.eventSource.addEventListener('chatMessage', async (event: MessageEvent) => { | 405 | + | this.eventSource.addEventListener('chatMessage', async (event: MessageEvent) => { |
| 382 | - | try { | 406 | + | try { |
| 383 | - | const messageData = JSON.parse(event.data) as Message; | 407 | + | const messageData = JSON.parse(event.data) as Message; |
| 384 | - | // Ensure the sender's data is available | 408 | + | // Ensure the sender's data is available |
| 385 | - | if (!this.userCache[messageData.senderId]) { | 409 | + | if (!this.userCache[messageData.senderId]) { |
| 386 | - | console.debug('Fetching user data for new message:', messageData.senderId); | 410 | + | console.debug('Fetching user data for new message:', messageData.senderId); |
| 387 | - | const response = await fetch(`/api/users/${messageData.senderId}`, { credentials: 'include' }); | 411 | + | const response = await fetch(`/api/users/${messageData.senderId}`, { credentials: 'include' }); |
| 388 | - | if (response.ok) { | 412 | + | if (response.ok) { |
| 389 | - | const userData = await response.json(); | 413 | + | const userData = await response.json(); |
| 390 | - | if (userData.success && userData.user) { | 414 | + | if (userData.success && userData.user) { |
| 391 | - | this.userCache[userData.user.id] = userData.user; | 415 | + | this.userCache[userData.user.id] = userData.user; |
| 392 | - | if (!this.users.some(u => u.id === userData.user.id)) { | 416 | + | if (!this.users.some(u => u.id === userData.user.id)) { |
| 393 | - | this.users = [...this.users, userData.user]; | 417 | + | this.users = [...this.users, userData.user]; |
| 418 | + | } | |||
| 394 | } | 419 | } | ||
| 395 | } | 420 | } | ||
| 396 | } | 421 | } | ||
| 422 | + | this.messages = [...this.messages, messageData]; | |||
| 423 | + | await invalidate('chat:messages'); | |||
| 424 | + | } catch (error) { | |||
| 425 | + | console.debug('Error parsing chatMessage SSE data:', error); | |||
| 397 | } | 426 | } | ||
| 398 | - | this.messages = [...this.messages, messageData]; | 427 | + | }); |
| 399 | - | await invalidate('chat:messages'); | |||
| 400 | - | } catch (error) { | |||
| 401 | - | console.debug('Error parsing chatMessage SSE data:', error); | |||
| 402 | - | } | |||
| 403 | - | }); | |||
| 404 | 428 | ||||
| 405 | - | // Listen for user status updates | 429 | + | // Listen for user status updates |
| 406 | - | this.eventSource.addEventListener('userStatusUpdate', (event: MessageEvent) => { | 430 | + | this.eventSource.addEventListener('userStatusUpdate', (event: MessageEvent) => { |
| 407 | - | try { | 431 | + | try { |
| 408 | - | const userUpdate = JSON.parse(event.data); | 432 | + | const userUpdate = JSON.parse(event.data); |
| 409 | - | console.debug('Received SSE user status update:', userUpdate); | 433 | + | console.debug('Received SSE user status update:', userUpdate); |
| 410 | - | this.updateUserStatus(userUpdate.userId, userUpdate.status, userUpdate.lastSeen); | 434 | + | this.updateUserStatus(userUpdate.userId, userUpdate.status, userUpdate.lastSeen); |
| 411 | - | } catch (error) { | 435 | + | } catch (error) { |
| 412 | - | console.debug('Error processing userStatusUpdate event:', error); | 436 | + | console.debug('Error processing userStatusUpdate event:', error); |
| 413 | - | } | 437 | + | } |
| 414 | - | }); | 438 | + | }); |
| 415 | 439 | ||||
| 416 | - | this.eventSource.onerror = (event) => { | 440 | + | this.eventSource.onerror = (event) => { |
| 417 | - | console.debug('SSE encountered an error:', event); | 441 | + | console.debug('SSE encountered an error:', event); |
| 442 | + | this.handleSSEError(); | |||
| 443 | + | }; | |||
| 444 | + | } catch (error) { | |||
| 445 | + | console.debug('Error setting up SSE connection:', error); | |||
| 446 | + | this.isReconnecting = false; | |||
| 418 | this.handleSSEError(); | 447 | this.handleSSEError(); | ||
| 419 | - | }; | 448 | + | } |
| 420 | } | 449 | } | ||
| 421 | 450 | ||||
| 422 | private async handleSSEError() { | 451 | private async handleSSEError() { | ||
| @@ -425,6 +454,12 @@ class ChatState { | |||||
| 425 | this.eventSource = null; | 454 | this.eventSource = null; | ||
| 426 | } | 455 | } | ||
| 427 | 456 | ||||
| 457 | + | // If we're already reconnecting, don't start another reconnection attempt | |||
| 458 | + | if (this.isReconnecting) { | |||
| 459 | + | console.debug('Already attempting to reconnect, skipping error handler'); | |||
| 460 | + | return; | |||
| 461 | + | } | |||
| 462 | + | ||||
| 428 | // Check if the error was due to rate limiting | 463 | // Check if the error was due to rate limiting | ||
| 429 | try { | 464 | try { | ||
| 430 | const response = await fetch('/api/sse'); | 465 | const response = await fetch('/api/sse'); | ||
| @@ -432,13 +467,14 @@ class ChatState { | |||||
| 432 | const data = await response.json(); | 467 | const data = await response.json(); | ||
| 433 | this.sseError = data.error; | 468 | this.sseError = data.error; | ||
| 434 | this.sseRetryAfter = data.retryAfter; | 469 | this.sseRetryAfter = data.retryAfter; | ||
| 470 | + | this.isReconnecting = false; | |||
| 435 | 471 | ||||
| 436 | // Set up automatic retry after the rate limit expires | 472 | // Set up automatic retry after the rate limit expires | ||
| 437 | setTimeout(() => { | 473 | setTimeout(() => { | ||
| 438 | this.sseError = null; | 474 | this.sseError = null; | ||
| 439 | this.sseRetryAfter = null; | 475 | this.sseRetryAfter = null; | ||
| 440 | this.reconnectSSE(); | 476 | this.reconnectSSE(); | ||
| 441 | - | }, data.retryAfter * 1000); | 477 | + | }, (data.retryAfter + 1) * 1000); // Add 1 second buffer |
| 442 | 478 | ||||
| 443 | return; | 479 | return; | ||
| 444 | } | 480 | } | ||
| @@ -454,15 +490,22 @@ class ChatState { | |||||
| 454 | setTimeout(() => { | 490 | setTimeout(() => { | ||
| 455 | this.reconnectAttempts++; | 491 | this.reconnectAttempts++; | ||
| 456 | this.reconnectDelay *= 2; // Exponential backoff | 492 | this.reconnectDelay *= 2; // Exponential backoff | ||
| 493 | + | this.isReconnecting = false; | |||
| 457 | this.connectSSE(); | 494 | this.connectSSE(); | ||
| 458 | }, this.reconnectDelay); | 495 | }, this.reconnectDelay); | ||
| 459 | } else { | 496 | } else { | ||
| 497 | + | this.isReconnecting = false; | |||
| 460 | this.sseError = 'Unable to connect to chat. Please refresh the page to try again.'; | 498 | this.sseError = 'Unable to connect to chat. Please refresh the page to try again.'; | ||
| 461 | console.debug('Max reconnection attempts reached'); | 499 | console.debug('Max reconnection attempts reached'); | ||
| 462 | } | 500 | } | ||
| 463 | } | 501 | } | ||
| 464 | 502 | ||||
| 465 | private reconnectSSE() { | 503 | private reconnectSSE() { | ||
| 504 | + | if (this.isReconnecting) { | |||
| 505 | + | console.debug('Already attempting to reconnect, skipping reconnection request'); | |||
| 506 | + | return; | |||
| 507 | + | } | |||
| 508 | + | ||||
| 466 | this.reconnectAttempts = 0; | 509 | this.reconnectAttempts = 0; | ||
| 467 | this.reconnectDelay = 1000; | 510 | this.reconnectDelay = 1000; | ||
| 468 | this.sseError = null; | 511 | this.sseError = null; | ||
M src/routes/api/sse/+server.ts +31 -11
| @@ -8,15 +8,18 @@ console.log('[SSE] Server initialized'); | |||
| 8 | 8 | ||
| 9 | 9 | // Constants for timeout checks | |
| 10 | 10 | const ONLINE_TIMEOUT_MS = 2 * 60 * 1000; // 2 minutes | |
| 11 | - | const CHECK_INTERVAL_MS = 60 * 1000; // Check every minute | |
| 11 | + | const CHECK_INTERVAL_MS = 30 * 1000; // Check every 30 seconds | |
| 12 | 12 | ||
| 13 | 13 | // Initialize periodic check for timed out users | |
| 14 | 14 | const timeoutInterval = setInterval(async () => { | |
| 15 | 15 | const timeoutThreshold = Date.now() - ONLINE_TIMEOUT_MS; | |
| 16 | - | console.log('[SSE] Checking for timed out users', { checkTime: new Date().toISOString() }); | |
| 16 | + | console.log('[SSE] Checking for timed out users', { | |
| 17 | + | checkTime: new Date().toISOString(), | |
| 18 | + | timeoutThreshold: new Date(timeoutThreshold).toISOString() | |
| 19 | + | }); | |
| 17 | 20 | ||
| 18 | 21 | try { | |
| 19 | - | // Find and update users who are marked as online but haven't been seen recently | |
| 22 | + | // Find and update users who haven't been seen recently | |
| 20 | 23 | const now = Date.now(); | |
| 21 | 24 | const result = await db.update(users) | |
| 22 | 25 | .set({ | |
| @@ -25,15 +28,26 @@ const timeoutInterval = setInterval(async () => { | |||
| 25 | 28 | }) | |
| 26 | 29 | .where( | |
| 27 | 30 | and( | |
| 28 | - | eq(users.status, 'online'), | |
| 29 | - | lt(users.lastSeen, timeoutThreshold) | |
| 31 | + | eq(users.status, 'online'), // Only update online users | |
| 32 | + | lt(users.lastSeen, timeoutThreshold) // Who haven't been seen recently | |
| 30 | 33 | ) | |
| 31 | 34 | ) | |
| 32 | - | .returning({ id: users.id }); | |
| 35 | + | .returning({ | |
| 36 | + | id: users.id, | |
| 37 | + | nickname: users.nickname, | |
| 38 | + | lastSeen: users.lastSeen | |
| 39 | + | }); | |
| 40 | + | ||
| 41 | + | // Log timeout results | |
| 42 | + | if (result.length > 0) { | |
| 43 | + | console.log('[SSE] Found timed out users:', result.map(user => ({ | |
| 44 | + | nickname: user.nickname, | |
| 45 | + | lastSeen: new Date(user.lastSeen || 0).toISOString() | |
| 46 | + | }))); | |
| 47 | + | } | |
| 33 | 48 | ||
| 34 | 49 | // Broadcast status updates for timed out users | |
| 35 | 50 | result.forEach(user => { | |
| 36 | - | const maskedUserId = `${user.id.slice(0, 4)}...${user.id.slice(-4)}`; | |
| 37 | 51 | sseEmitter.emit('sse', { | |
| 38 | 52 | type: 'userStatusUpdate', | |
| 39 | 53 | data: { | |
| @@ -42,10 +56,13 @@ const timeoutInterval = setInterval(async () => { | |||
| 42 | 56 | lastSeen: now | |
| 43 | 57 | } | |
| 44 | 58 | }); | |
| 45 | - | console.log('[SSE] User timed out and marked offline:', maskedUserId); | |
| 59 | + | console.log('[SSE] User timed out and marked offline:', { | |
| 60 | + | nickname: user.nickname, | |
| 61 | + | lastSeen: new Date(now).toISOString() | |
| 62 | + | }); | |
| 46 | 63 | }); | |
| 47 | - | } catch { | |
| 48 | - | console.log('[SSE] Error during timeout check'); | |
| 64 | + | } catch (error) { | |
| 65 | + | console.log('[SSE] Error during timeout check:', error); | |
| 49 | 66 | } | |
| 50 | 67 | }, CHECK_INTERVAL_MS); | |
| 51 | 68 | ||
| @@ -117,7 +134,10 @@ export const GET: RequestHandler = async ({ locals }) => { | |||
| 117 | 134 | // Update lastSeen timestamp on each keepalive | |
| 118 | 135 | await db.update(users) | |
| 119 | 136 | .set({ lastSeen: Date.now() }) | |
| 120 | - | .where(eq(users.id, userId)); | |
| 137 | + | .where(eq(users.id, userId)) | |
| 138 | + | .catch(error => { | |
| 139 | + | console.log('[SSE] Error updating lastSeen on keepalive:', error); | |
| 140 | + | }); | |
| 121 | 141 | }, 20000); | |
| 122 | 142 | }, | |
| 123 | 143 | async cancel(reason) { | |
| @@ -8,15 +8,18 @@ console.log('[SSE] Server initialized'); | |||||
| 8 | 8 | ||||
| 9 | // Constants for timeout checks | 9 | // Constants for timeout checks | ||
| 10 | const ONLINE_TIMEOUT_MS = 2 * 60 * 1000; // 2 minutes | 10 | const ONLINE_TIMEOUT_MS = 2 * 60 * 1000; // 2 minutes | ||
| 11 | - | const CHECK_INTERVAL_MS = 60 * 1000; // Check every minute | 11 | + | const CHECK_INTERVAL_MS = 30 * 1000; // Check every 30 seconds |
| 12 | 12 | ||||
| 13 | // Initialize periodic check for timed out users | 13 | // Initialize periodic check for timed out users | ||
| 14 | const timeoutInterval = setInterval(async () => { | 14 | const timeoutInterval = setInterval(async () => { | ||
| 15 | const timeoutThreshold = Date.now() - ONLINE_TIMEOUT_MS; | 15 | const timeoutThreshold = Date.now() - ONLINE_TIMEOUT_MS; | ||
| 16 | - | console.log('[SSE] Checking for timed out users', { checkTime: new Date().toISOString() }); | 16 | + | console.log('[SSE] Checking for timed out users', { |
| 17 | + | checkTime: new Date().toISOString(), | |||
| 18 | + | timeoutThreshold: new Date(timeoutThreshold).toISOString() | |||
| 19 | + | }); | |||
| 17 | 20 | ||||
| 18 | try { | 21 | try { | ||
| 19 | - | // Find and update users who are marked as online but haven't been seen recently | 22 | + | // Find and update users who haven't been seen recently |
| 20 | const now = Date.now(); | 23 | const now = Date.now(); | ||
| 21 | const result = await db.update(users) | 24 | const result = await db.update(users) | ||
| 22 | .set({ | 25 | .set({ | ||
| @@ -25,15 +28,26 @@ const timeoutInterval = setInterval(async () => { | |||||
| 25 | }) | 28 | }) | ||
| 26 | .where( | 29 | .where( | ||
| 27 | and( | 30 | and( | ||
| 28 | - | eq(users.status, 'online'), | 31 | + | eq(users.status, 'online'), // Only update online users |
| 29 | - | lt(users.lastSeen, timeoutThreshold) | 32 | + | lt(users.lastSeen, timeoutThreshold) // Who haven't been seen recently |
| 30 | ) | 33 | ) | ||
| 31 | ) | 34 | ) | ||
| 32 | - | .returning({ id: users.id }); | 35 | + | .returning({ |
| 36 | + | id: users.id, | |||
| 37 | + | nickname: users.nickname, | |||
| 38 | + | lastSeen: users.lastSeen | |||
| 39 | + | }); | |||
| 40 | + | ||||
| 41 | + | // Log timeout results | |||
| 42 | + | if (result.length > 0) { | |||
| 43 | + | console.log('[SSE] Found timed out users:', result.map(user => ({ | |||
| 44 | + | nickname: user.nickname, | |||
| 45 | + | lastSeen: new Date(user.lastSeen || 0).toISOString() | |||
| 46 | + | }))); | |||
| 47 | + | } | |||
| 33 | 48 | ||||
| 34 | // Broadcast status updates for timed out users | 49 | // Broadcast status updates for timed out users | ||
| 35 | result.forEach(user => { | 50 | result.forEach(user => { | ||
| 36 | - | const maskedUserId = `${user.id.slice(0, 4)}...${user.id.slice(-4)}`; | |||
| 37 | sseEmitter.emit('sse', { | 51 | sseEmitter.emit('sse', { | ||
| 38 | type: 'userStatusUpdate', | 52 | type: 'userStatusUpdate', | ||
| 39 | data: { | 53 | data: { | ||
| @@ -42,10 +56,13 @@ const timeoutInterval = setInterval(async () => { | |||||
| 42 | lastSeen: now | 56 | lastSeen: now | ||
| 43 | } | 57 | } | ||
| 44 | }); | 58 | }); | ||
| 45 | - | console.log('[SSE] User timed out and marked offline:', maskedUserId); | 59 | + | console.log('[SSE] User timed out and marked offline:', { |
| 60 | + | nickname: user.nickname, | |||
| 61 | + | lastSeen: new Date(now).toISOString() | |||
| 62 | + | }); | |||
| 46 | }); | 63 | }); | ||
| 47 | - | } catch { | 64 | + | } catch (error) { |
| 48 | - | console.log('[SSE] Error during timeout check'); | 65 | + | console.log('[SSE] Error during timeout check:', error); |
| 49 | } | 66 | } | ||
| 50 | }, CHECK_INTERVAL_MS); | 67 | }, CHECK_INTERVAL_MS); | ||
| 51 | 68 | ||||
| @@ -117,7 +134,10 @@ export const GET: RequestHandler = async ({ locals }) => { | |||||
| 117 | // Update lastSeen timestamp on each keepalive | 134 | // Update lastSeen timestamp on each keepalive | ||
| 118 | await db.update(users) | 135 | await db.update(users) | ||
| 119 | .set({ lastSeen: Date.now() }) | 136 | .set({ lastSeen: Date.now() }) | ||
| 120 | - | .where(eq(users.id, userId)); | 137 | + | .where(eq(users.id, userId)) |
| 138 | + | .catch(error => { | |||
| 139 | + | console.log('[SSE] Error updating lastSeen on keepalive:', error); | |||
| 140 | + | }); | |||
| 121 | }, 20000); | 141 | }, 20000); | ||
| 122 | }, | 142 | }, | ||
| 123 | async cancel(reason) { | 143 | async cancel(reason) { | ||
M src/routes/api/status/+server.ts +15 -3
| @@ -21,9 +21,15 @@ export async function POST({ request, cookies, locals }) { | |||
| 21 | 21 | const maskedUserId = `${userId.slice(0, 4)}...${userId.slice(-4)}`; | |
| 22 | 22 | console.log('[Status] Updating user status:', { userId: maskedUserId, status }); | |
| 23 | 23 | ||
| 24 | + | const now = Date.now(); | |
| 25 | + | ||
| 24 | 26 | // Update the user status in the database | |
| 27 | + | // Always update lastSeen to support timeout detection | |
| 25 | 28 | await db.update(users) | |
| 26 | - | .set({ status, lastSeen: status === 'offline' ? Date.now() : null }) | |
| 29 | + | .set({ | |
| 30 | + | status, | |
| 31 | + | lastSeen: now // Always update lastSeen for proper timeout detection | |
| 32 | + | }) | |
| 27 | 33 | .where(eq(users.id, userId)); | |
| 28 | 34 | console.log('[Status] Database updated successfully'); | |
| 29 | 35 | ||
| @@ -33,7 +39,6 @@ export async function POST({ request, cookies, locals }) { | |||
| 33 | 39 | if (token) { | |
| 34 | 40 | const result = await validateSessionToken(token); | |
| 35 | 41 | if (result.session) { | |
| 36 | - | const now = Date.now(); | |
| 37 | 42 | const remaining = result.session.expiresAt - now; | |
| 38 | 43 | const threshold = 15 * 60 * 1000; // 15 minutes in milliseconds | |
| 39 | 44 | if (remaining < threshold) { | |
| @@ -47,7 +52,14 @@ export async function POST({ request, cookies, locals }) { | |||
| 47 | 52 | } | |
| 48 | 53 | ||
| 49 | 54 | // Broadcast the status update via SSE | |
| 50 | - | sseEmitter.emit('sse', { type: 'userStatusUpdate', data: { userId, status, lastSeen: status === 'offline' ? Date.now() : null } }); | |
| 55 | + | sseEmitter.emit('sse', { | |
| 56 | + | type: 'userStatusUpdate', | |
| 57 | + | data: { | |
| 58 | + | userId, | |
| 59 | + | status, | |
| 60 | + | lastSeen: now // Always send the current timestamp | |
| 61 | + | } | |
| 62 | + | }); | |
| 51 | 63 | ||
| 52 | 64 | return new Response(JSON.stringify({ success: true }), { status: 200, headers: { 'Content-Type': 'application/json' } }); | |
| 53 | 65 | }⏎̸ | |
| @@ -21,9 +21,15 @@ export async function POST({ request, cookies, locals }) { | |||||
| 21 | const maskedUserId = `${userId.slice(0, 4)}...${userId.slice(-4)}`; | 21 | const maskedUserId = `${userId.slice(0, 4)}...${userId.slice(-4)}`; | ||
| 22 | console.log('[Status] Updating user status:', { userId: maskedUserId, status }); | 22 | console.log('[Status] Updating user status:', { userId: maskedUserId, status }); | ||
| 23 | 23 | ||||
| 24 | + | const now = Date.now(); | |||
| 25 | + | ||||
| 24 | // Update the user status in the database | 26 | // Update the user status in the database | ||
| 27 | + | // Always update lastSeen to support timeout detection | |||
| 25 | await db.update(users) | 28 | await db.update(users) | ||
| 26 | - | .set({ status, lastSeen: status === 'offline' ? Date.now() : null }) | 29 | + | .set({ |
| 30 | + | status, | |||
| 31 | + | lastSeen: now // Always update lastSeen for proper timeout detection | |||
| 32 | + | }) | |||
| 27 | .where(eq(users.id, userId)); | 33 | .where(eq(users.id, userId)); | ||
| 28 | console.log('[Status] Database updated successfully'); | 34 | console.log('[Status] Database updated successfully'); | ||
| 29 | 35 | ||||
| @@ -33,7 +39,6 @@ export async function POST({ request, cookies, locals }) { | |||||
| 33 | if (token) { | 39 | if (token) { | ||
| 34 | const result = await validateSessionToken(token); | 40 | const result = await validateSessionToken(token); | ||
| 35 | if (result.session) { | 41 | if (result.session) { | ||
| 36 | - | const now = Date.now(); | |||
| 37 | const remaining = result.session.expiresAt - now; | 42 | const remaining = result.session.expiresAt - now; | ||
| 38 | const threshold = 15 * 60 * 1000; // 15 minutes in milliseconds | 43 | const threshold = 15 * 60 * 1000; // 15 minutes in milliseconds | ||
| 39 | if (remaining < threshold) { | 44 | if (remaining < threshold) { | ||
| @@ -47,7 +52,14 @@ export async function POST({ request, cookies, locals }) { | |||||
| 47 | } | 52 | } | ||
| 48 | 53 | ||||
| 49 | // Broadcast the status update via SSE | 54 | // Broadcast the status update via SSE | ||
| 50 | - | sseEmitter.emit('sse', { type: 'userStatusUpdate', data: { userId, status, lastSeen: status === 'offline' ? Date.now() : null } }); | 55 | + | sseEmitter.emit('sse', { |
| 56 | + | type: 'userStatusUpdate', | |||
| 57 | + | data: { | |||
| 58 | + | userId, | |||
| 59 | + | status, | |||
| 60 | + | lastSeen: now // Always send the current timestamp | |||
| 61 | + | } | |||
| 62 | + | }); | |||
| 51 | 63 | ||||
| 52 | return new Response(JSON.stringify({ success: true }), { status: 200, headers: { 'Content-Type': 'application/json' } }); | 64 | return new Response(JSON.stringify({ success: true }), { status: 200, headers: { 'Content-Type': 'application/json' } }); | ||
| 53 | }⏎̸ | 65 | }⏎̸ | ||