feat: Enhance user status tracking and SSE integration
- Implement comprehensive user status management with visibility and activity tracking - Add last seen functionality and granular status updates (online, away, busy, offline) - Refactor SSE handling to support multiple event types and unified connection - Improve chat room and user initialization with more robust error handling - Update chat state management to preserve user details across status changes
- commit
- 6b9673c9b5ceea1241e4ce88114c926662a5e9d6
- author
- Kit · · 2025-02-13 16:31 UTC
- parent
- 6ce47eb
- signature
- SSH Unverified
SHA256:H91DFuEG4jpk5edzLA4aUKkpv4zv9Q339ipRmkph6TkThe key is unknown
11 files · +611 -158
M src/hooks.server.ts +7 -0
| @@ -1,6 +1,13 @@ | |||
| 1 | 1 | import type { Handle } from "@sveltejs/kit"; | |
| 2 | 2 | import { validateSessionToken } from "$lib/api/session.server"; | |
| 3 | 3 | import { setSessionTokenCookie, deleteSessionTokenCookie } from "$lib/api/session.cookie"; | |
| 4 | + | import { initializeDatabase } from "$lib/db/db.server"; | |
| 5 | + | ||
| 6 | + | // Initialize database when the server starts | |
| 7 | + | console.debug("Initializing database on server start..."); | |
| 8 | + | initializeDatabase().catch(error => { | |
| 9 | + | console.error('Failed to initialize database:', error); | |
| 10 | + | }); | |
| 4 | 11 | ||
| 5 | 12 | export const handle: Handle = async ({ event, resolve }) => { | |
| 6 | 13 | console.debug("Handling request:", event.url.pathname); | |
| @@ -1,6 +1,13 @@ | |||||
| 1 | import type { Handle } from "@sveltejs/kit"; | 1 | import type { Handle } from "@sveltejs/kit"; | ||
| 2 | import { validateSessionToken } from "$lib/api/session.server"; | 2 | import { validateSessionToken } from "$lib/api/session.server"; | ||
| 3 | import { setSessionTokenCookie, deleteSessionTokenCookie } from "$lib/api/session.cookie"; | 3 | import { setSessionTokenCookie, deleteSessionTokenCookie } from "$lib/api/session.cookie"; | ||
| 4 | + | import { initializeDatabase } from "$lib/db/db.server"; | |||
| 5 | + | ||||
| 6 | + | // Initialize database when the server starts | |||
| 7 | + | console.debug("Initializing database on server start..."); | |||
| 8 | + | initializeDatabase().catch(error => { | |||
| 9 | + | console.error('Failed to initialize database:', error); | |||
| 10 | + | }); | |||
| 4 | 11 | ||||
| 5 | export const handle: Handle = async ({ event, resolve }) => { | 12 | export const handle: Handle = async ({ event, resolve }) => { | ||
| 6 | console.debug("Handling request:", event.url.pathname); | 13 | console.debug("Handling request:", event.url.pathname); | ||
M src/lib/components/chat-room.svelte +51 -2
| @@ -7,6 +7,7 @@ import { draggable } from '$lib/actions/draggable'; | |||
| 7 | 7 | import { resizable } from '$lib/actions/resizable'; | |
| 8 | 8 | import { maximizable } from '$lib/actions/maximizable'; | |
| 9 | 9 | import LoadingButton from './ui/button-loading.svelte'; | |
| 10 | + | import { DEFAULT_CHAT_ROOM_ID } from '$lib/db/schema'; | |
| 10 | 11 | ||
| 11 | 12 | // Initialize with default values for SSR | |
| 12 | 13 | let windowWidth = $state(400); | |
| @@ -80,7 +81,7 @@ $effect(() => { | |||
| 80 | 81 | ||
| 81 | 82 | // Fetch both in parallel | |
| 82 | 83 | Promise.all([ | |
| 83 | - | fetch('/api/rooms/default?public=true').then(r => r.json()), | |
| 84 | + | fetch(`/api/rooms/${DEFAULT_CHAT_ROOM_ID}?public=true`).then(r => r.json()), | |
| 84 | 85 | fetch('/api/chat/messages?public=true').then(r => r.json()) | |
| 85 | 86 | ]) | |
| 86 | 87 | .then(([roomData, messagesData]) => { | |
| @@ -166,6 +167,39 @@ function getStatusIcon(status: User['status']) { | |||
| 166 | 167 | } | |
| 167 | 168 | } | |
| 168 | 169 | ||
| 170 | + | function formatLastSeen(lastSeen: number | null | undefined) { | |
| 171 | + | if (!lastSeen) return 'Never'; | |
| 172 | + | ||
| 173 | + | const now = Date.now(); | |
| 174 | + | const diff = now - lastSeen; | |
| 175 | + | ||
| 176 | + | // Less than a minute | |
| 177 | + | if (diff < 60000) { | |
| 178 | + | return 'Just now'; | |
| 179 | + | } | |
| 180 | + | ||
| 181 | + | // Less than an hour | |
| 182 | + | if (diff < 3600000) { | |
| 183 | + | const minutes = Math.floor(diff / 60000); | |
| 184 | + | return `${minutes} minute${minutes > 1 ? 's' : ''} ago`; | |
| 185 | + | } | |
| 186 | + | ||
| 187 | + | // Less than a day | |
| 188 | + | if (diff < 86400000) { | |
| 189 | + | const hours = Math.floor(diff / 3600000); | |
| 190 | + | return `${hours} hour${hours > 1 ? 's' : ''} ago`; | |
| 191 | + | } | |
| 192 | + | ||
| 193 | + | // Less than a week | |
| 194 | + | if (diff < 604800000) { | |
| 195 | + | const days = Math.floor(diff / 86400000); | |
| 196 | + | return `${days} day${days > 1 ? 's' : ''} ago`; | |
| 197 | + | } | |
| 198 | + | ||
| 199 | + | // More than a week | |
| 200 | + | return new Date(lastSeen).toLocaleDateString(); | |
| 201 | + | } | |
| 202 | + | ||
| 169 | 203 | function handleDragMove(event: CustomEvent<{ x: number; y: number }>) { | |
| 170 | 204 | windowX = event.detail.x; | |
| 171 | 205 | windowY = event.detail.y; | |
| @@ -285,12 +319,16 @@ function handleMaximizeEvent(event: CustomEvent<{ | |||
| 285 | 319 | > | |
| 286 | 320 | <p style="margin: 0 0 0.5rem 0;"><strong>Buddy List</strong></p> | |
| 287 | 321 | {#each onlineUsers as user} | |
| 288 | - | <div class="user"> | |
| 322 | + | <div class="user" class:offline={user.status === 'offline'}> | |
| 289 | 323 | <span class="status-icon">{getStatusIcon(user.status)}</span> | |
| 290 | 324 | <div class="user-info"> | |
| 291 | 325 | <div class="nickname">{user.nickname}</div> | |
| 292 | 326 | {#if user.status === 'away'} | |
| 293 | 327 | <div class="status-message">Away</div> | |
| 328 | + | {:else if user.status === 'busy'} | |
| 329 | + | <div class="status-message">Busy</div> | |
| 330 | + | {:else if user.status === 'offline'} | |
| 331 | + | <div class="status-message">Last seen: {formatLastSeen(user.lastSeen)}</div> | |
| 294 | 332 | {/if} | |
| 295 | 333 | </div> | |
| 296 | 334 | </div> | |
| @@ -350,10 +388,16 @@ function handleMaximizeEvent(event: CustomEvent<{ | |||
| 350 | 388 | align-items: flex-start; | |
| 351 | 389 | margin-bottom: 0.75rem; | |
| 352 | 390 | gap: 0.375rem; | |
| 391 | + | transition: opacity 0.3s ease; | |
| 392 | + | } | |
| 393 | + | ||
| 394 | + | .user.offline { | |
| 395 | + | opacity: 0.6; | |
| 353 | 396 | } | |
| 354 | 397 | ||
| 355 | 398 | .status-icon { | |
| 356 | 399 | font-size: 0.875rem; | |
| 400 | + | transition: color 0.3s ease; | |
| 357 | 401 | } | |
| 358 | 402 | ||
| 359 | 403 | .user-info { | |
| @@ -364,6 +408,11 @@ function handleMaximizeEvent(event: CustomEvent<{ | |||
| 364 | 408 | .user .nickname { | |
| 365 | 409 | font-weight: bold; | |
| 366 | 410 | margin-bottom: 0.125rem; | |
| 411 | + | transition: color 0.3s ease; | |
| 412 | + | } | |
| 413 | + | ||
| 414 | + | .user.offline .nickname { | |
| 415 | + | color: #666; | |
| 367 | 416 | } | |
| 368 | 417 | ||
| 369 | 418 | .user .status-message { | |
| @@ -7,6 +7,7 @@ import { draggable } from '$lib/actions/draggable'; | |||||
| 7 | import { resizable } from '$lib/actions/resizable'; | 7 | import { resizable } from '$lib/actions/resizable'; | ||
| 8 | import { maximizable } from '$lib/actions/maximizable'; | 8 | import { maximizable } from '$lib/actions/maximizable'; | ||
| 9 | import LoadingButton from './ui/button-loading.svelte'; | 9 | import LoadingButton from './ui/button-loading.svelte'; | ||
| 10 | + | import { DEFAULT_CHAT_ROOM_ID } from '$lib/db/schema'; | |||
| 10 | 11 | ||||
| 11 | // Initialize with default values for SSR | 12 | // Initialize with default values for SSR | ||
| 12 | let windowWidth = $state(400); | 13 | let windowWidth = $state(400); | ||
| @@ -80,7 +81,7 @@ $effect(() => { | |||||
| 80 | 81 | ||||
| 81 | // Fetch both in parallel | 82 | // Fetch both in parallel | ||
| 82 | Promise.all([ | 83 | Promise.all([ | ||
| 83 | - | fetch('/api/rooms/default?public=true').then(r => r.json()), | 84 | + | fetch(`/api/rooms/${DEFAULT_CHAT_ROOM_ID}?public=true`).then(r => r.json()), |
| 84 | fetch('/api/chat/messages?public=true').then(r => r.json()) | 85 | fetch('/api/chat/messages?public=true').then(r => r.json()) | ||
| 85 | ]) | 86 | ]) | ||
| 86 | .then(([roomData, messagesData]) => { | 87 | .then(([roomData, messagesData]) => { | ||
| @@ -166,6 +167,39 @@ function getStatusIcon(status: User['status']) { | |||||
| 166 | } | 167 | } | ||
| 167 | } | 168 | } | ||
| 168 | 169 | ||||
| 170 | + | function formatLastSeen(lastSeen: number | null | undefined) { | |||
| 171 | + | if (!lastSeen) return 'Never'; | |||
| 172 | + | ||||
| 173 | + | const now = Date.now(); | |||
| 174 | + | const diff = now - lastSeen; | |||
| 175 | + | ||||
| 176 | + | // Less than a minute | |||
| 177 | + | if (diff < 60000) { | |||
| 178 | + | return 'Just now'; | |||
| 179 | + | } | |||
| 180 | + | ||||
| 181 | + | // Less than an hour | |||
| 182 | + | if (diff < 3600000) { | |||
| 183 | + | const minutes = Math.floor(diff / 60000); | |||
| 184 | + | return `${minutes} minute${minutes > 1 ? 's' : ''} ago`; | |||
| 185 | + | } | |||
| 186 | + | ||||
| 187 | + | // Less than a day | |||
| 188 | + | if (diff < 86400000) { | |||
| 189 | + | const hours = Math.floor(diff / 3600000); | |||
| 190 | + | return `${hours} hour${hours > 1 ? 's' : ''} ago`; | |||
| 191 | + | } | |||
| 192 | + | ||||
| 193 | + | // Less than a week | |||
| 194 | + | if (diff < 604800000) { | |||
| 195 | + | const days = Math.floor(diff / 86400000); | |||
| 196 | + | return `${days} day${days > 1 ? 's' : ''} ago`; | |||
| 197 | + | } | |||
| 198 | + | ||||
| 199 | + | // More than a week | |||
| 200 | + | return new Date(lastSeen).toLocaleDateString(); | |||
| 201 | + | } | |||
| 202 | + | ||||
| 169 | function handleDragMove(event: CustomEvent<{ x: number; y: number }>) { | 203 | function handleDragMove(event: CustomEvent<{ x: number; y: number }>) { | ||
| 170 | windowX = event.detail.x; | 204 | windowX = event.detail.x; | ||
| 171 | windowY = event.detail.y; | 205 | windowY = event.detail.y; | ||
| @@ -285,12 +319,16 @@ function handleMaximizeEvent(event: CustomEvent<{ | |||||
| 285 | > | 319 | > | ||
| 286 | <p style="margin: 0 0 0.5rem 0;"><strong>Buddy List</strong></p> | 320 | <p style="margin: 0 0 0.5rem 0;"><strong>Buddy List</strong></p> | ||
| 287 | {#each onlineUsers as user} | 321 | {#each onlineUsers as user} | ||
| 288 | - | <div class="user"> | 322 | + | <div class="user" class:offline={user.status === 'offline'}> |
| 289 | <span class="status-icon">{getStatusIcon(user.status)}</span> | 323 | <span class="status-icon">{getStatusIcon(user.status)}</span> | ||
| 290 | <div class="user-info"> | 324 | <div class="user-info"> | ||
| 291 | <div class="nickname">{user.nickname}</div> | 325 | <div class="nickname">{user.nickname}</div> | ||
| 292 | {#if user.status === 'away'} | 326 | {#if user.status === 'away'} | ||
| 293 | <div class="status-message">Away</div> | 327 | <div class="status-message">Away</div> | ||
| 328 | + | {:else if user.status === 'busy'} | |||
| 329 | + | <div class="status-message">Busy</div> | |||
| 330 | + | {:else if user.status === 'offline'} | |||
| 331 | + | <div class="status-message">Last seen: {formatLastSeen(user.lastSeen)}</div> | |||
| 294 | {/if} | 332 | {/if} | ||
| 295 | </div> | 333 | </div> | ||
| 296 | </div> | 334 | </div> | ||
| @@ -350,10 +388,16 @@ function handleMaximizeEvent(event: CustomEvent<{ | |||||
| 350 | align-items: flex-start; | 388 | align-items: flex-start; | ||
| 351 | margin-bottom: 0.75rem; | 389 | margin-bottom: 0.75rem; | ||
| 352 | gap: 0.375rem; | 390 | gap: 0.375rem; | ||
| 391 | + | transition: opacity 0.3s ease; | |||
| 392 | + | } | |||
| 393 | + | ||||
| 394 | + | .user.offline { | |||
| 395 | + | opacity: 0.6; | |||
| 353 | } | 396 | } | ||
| 354 | 397 | ||||
| 355 | .status-icon { | 398 | .status-icon { | ||
| 356 | font-size: 0.875rem; | 399 | font-size: 0.875rem; | ||
| 400 | + | transition: color 0.3s ease; | |||
| 357 | } | 401 | } | ||
| 358 | 402 | ||||
| 359 | .user-info { | 403 | .user-info { | ||
| @@ -364,6 +408,11 @@ function handleMaximizeEvent(event: CustomEvent<{ | |||||
| 364 | .user .nickname { | 408 | .user .nickname { | ||
| 365 | font-weight: bold; | 409 | font-weight: bold; | ||
| 366 | margin-bottom: 0.125rem; | 410 | margin-bottom: 0.125rem; | ||
| 411 | + | transition: color 0.3s ease; | |||
| 412 | + | } | |||
| 413 | + | ||||
| 414 | + | .user.offline .nickname { | |||
| 415 | + | color: #666; | |||
| 367 | } | 416 | } | ||
| 368 | 417 | ||||
| 369 | .user .status-message { | 418 | .user .status-message { | ||
M src/lib/db/schema.ts +38 -18
| @@ -5,6 +5,19 @@ import { v4 as uuidv4 } from "uuid"; | |||
| 5 | 5 | import { eq } from "drizzle-orm"; | |
| 6 | 6 | import type { LibSQLDatabase } from "drizzle-orm/libsql"; | |
| 7 | 7 | ||
| 8 | + | // Default chat room ID - using a UUID to avoid collisions | |
| 9 | + | export const DEFAULT_CHAT_ROOM_ID = '00000000-0000-0000-0000-000000000001'; | |
| 10 | + | ||
| 11 | + | // Type guard for LibSQL errors | |
| 12 | + | interface LibSQLError { | |
| 13 | + | code: string; | |
| 14 | + | message: string; | |
| 15 | + | } | |
| 16 | + | ||
| 17 | + | function isLibSQLError(error: unknown): error is LibSQLError { | |
| 18 | + | return typeof error === 'object' && error !== null && 'code' in error; | |
| 19 | + | } | |
| 20 | + | ||
| 8 | 21 | export const users = table( | |
| 9 | 22 | "users", | |
| 10 | 23 | { | |
| @@ -109,25 +122,32 @@ const schema = { | |||
| 109 | 122 | export async function ensureDefaultChatRoom(db: LibSQLDatabase<typeof schema>) { | |
| 110 | 123 | console.debug('Ensuring default chat room exists...'); | |
| 111 | 124 | try { | |
| 112 | - | // Check if default room exists | |
| 113 | - | const defaultRoom = await db.select() | |
| 114 | - | .from(chatRooms) | |
| 115 | - | .where(eq(chatRooms.id, 'default')) | |
| 116 | - | .get(); | |
| 117 | - | ||
| 118 | - | if (!defaultRoom) { | |
| 119 | - | console.debug('Creating default chat room...'); | |
| 120 | - | await db.insert(chatRooms).values({ | |
| 121 | - | id: 'default', | |
| 122 | - | name: 'General', | |
| 123 | - | type: 'group', | |
| 124 | - | createdAt: Date.now() | |
| 125 | - | }); | |
| 126 | - | console.debug('Default chat room created successfully'); | |
| 127 | - | } else { | |
| 128 | - | console.debug('Default chat room already exists'); | |
| 125 | + | // Use a transaction to handle concurrent creation attempts | |
| 126 | + | await db.transaction(async (tx) => { | |
| 127 | + | const defaultRoom = await tx.select() | |
| 128 | + | .from(chatRooms) | |
| 129 | + | .where(eq(chatRooms.id, DEFAULT_CHAT_ROOM_ID)) | |
| 130 | + | .get(); | |
| 131 | + | ||
| 132 | + | if (!defaultRoom) { | |
| 133 | + | console.debug('Creating default chat room...'); | |
| 134 | + | await tx.insert(chatRooms).values({ | |
| 135 | + | id: DEFAULT_CHAT_ROOM_ID, | |
| 136 | + | name: 'General', | |
| 137 | + | type: 'group', | |
| 138 | + | createdAt: Date.now() | |
| 139 | + | }); | |
| 140 | + | console.debug('Default chat room created successfully'); | |
| 141 | + | } else { | |
| 142 | + | console.debug('Default chat room already exists'); | |
| 143 | + | } | |
| 144 | + | }); | |
| 145 | + | } catch (error: unknown) { | |
| 146 | + | // If the error is a unique constraint violation, another process probably created the room | |
| 147 | + | if (isLibSQLError(error) && error.code === 'SQLITE_CONSTRAINT') { | |
| 148 | + | console.debug('Default chat room was created by another process'); | |
| 149 | + | return; | |
| 129 | 150 | } | |
| 130 | - | } catch (error) { | |
| 131 | 151 | console.error('Error ensuring default chat room:', error); | |
| 132 | 152 | throw error; | |
| 133 | 153 | } | |
| @@ -5,6 +5,19 @@ import { v4 as uuidv4 } from "uuid"; | |||||
| 5 | import { eq } from "drizzle-orm"; | 5 | import { eq } from "drizzle-orm"; | ||
| 6 | import type { LibSQLDatabase } from "drizzle-orm/libsql"; | 6 | import type { LibSQLDatabase } from "drizzle-orm/libsql"; | ||
| 7 | 7 | ||||
| 8 | + | // Default chat room ID - using a UUID to avoid collisions | |||
| 9 | + | export const DEFAULT_CHAT_ROOM_ID = '00000000-0000-0000-0000-000000000001'; | |||
| 10 | + | ||||
| 11 | + | // Type guard for LibSQL errors | |||
| 12 | + | interface LibSQLError { | |||
| 13 | + | code: string; | |||
| 14 | + | message: string; | |||
| 15 | + | } | |||
| 16 | + | ||||
| 17 | + | function isLibSQLError(error: unknown): error is LibSQLError { | |||
| 18 | + | return typeof error === 'object' && error !== null && 'code' in error; | |||
| 19 | + | } | |||
| 20 | + | ||||
| 8 | export const users = table( | 21 | export const users = table( | ||
| 9 | "users", | 22 | "users", | ||
| 10 | { | 23 | { | ||
| @@ -109,25 +122,32 @@ const schema = { | |||||
| 109 | export async function ensureDefaultChatRoom(db: LibSQLDatabase<typeof schema>) { | 122 | export async function ensureDefaultChatRoom(db: LibSQLDatabase<typeof schema>) { | ||
| 110 | console.debug('Ensuring default chat room exists...'); | 123 | console.debug('Ensuring default chat room exists...'); | ||
| 111 | try { | 124 | try { | ||
| 112 | - | // Check if default room exists | 125 | + | // Use a transaction to handle concurrent creation attempts |
| 113 | - | const defaultRoom = await db.select() | 126 | + | await db.transaction(async (tx) => { |
| 114 | - | .from(chatRooms) | 127 | + | const defaultRoom = await tx.select() |
| 115 | - | .where(eq(chatRooms.id, 'default')) | 128 | + | .from(chatRooms) |
| 116 | - | .get(); | 129 | + | .where(eq(chatRooms.id, DEFAULT_CHAT_ROOM_ID)) |
| 117 | - | 130 | + | .get(); | |
| 118 | - | if (!defaultRoom) { | 131 | + | |
| 119 | - | console.debug('Creating default chat room...'); | 132 | + | if (!defaultRoom) { |
| 120 | - | await db.insert(chatRooms).values({ | 133 | + | console.debug('Creating default chat room...'); |
| 121 | - | id: 'default', | 134 | + | await tx.insert(chatRooms).values({ |
| 122 | - | name: 'General', | 135 | + | id: DEFAULT_CHAT_ROOM_ID, |
| 123 | - | type: 'group', | 136 | + | name: 'General', |
| 124 | - | createdAt: Date.now() | 137 | + | type: 'group', |
| 125 | - | }); | 138 | + | createdAt: Date.now() |
| 126 | - | console.debug('Default chat room created successfully'); | 139 | + | }); |
| 127 | - | } else { | 140 | + | console.debug('Default chat room created successfully'); |
| 128 | - | console.debug('Default chat room already exists'); | 141 | + | } else { |
| 142 | + | console.debug('Default chat room already exists'); | |||
| 143 | + | } | |||
| 144 | + | }); | |||
| 145 | + | } catch (error: unknown) { | |||
| 146 | + | // If the error is a unique constraint violation, another process probably created the room | |||
| 147 | + | if (isLibSQLError(error) && error.code === 'SQLITE_CONSTRAINT') { | |||
| 148 | + | console.debug('Default chat room was created by another process'); | |||
| 149 | + | return; | |||
| 129 | } | 150 | } | ||
| 130 | - | } catch (error) { | |||
| 131 | console.error('Error ensuring default chat room:', error); | 151 | console.error('Error ensuring default chat room:', error); | ||
| 132 | throw error; | 152 | throw error; | ||
| 133 | } | 153 | } | ||
A src/lib/sseEmitter.ts +5 -0
| @@ -0,0 +1,5 @@ | |||
| 1 | + | import { EventEmitter } from 'events'; | |
| 2 | + | ||
| 3 | + | console.debug('Initializing shared SSE emitter'); | |
| 4 | + | export const sseEmitter = new EventEmitter(); | |
| 5 | + | sseEmitter.setMaxListeners(100); ⏎̸ | |
| @@ -0,0 +1,5 @@ | |||||
| 1 | + | import { EventEmitter } from 'events'; | |||
| 2 | + | ||||
| 3 | + | console.debug('Initializing shared SSE emitter'); | |||
| 4 | + | export const sseEmitter = new EventEmitter(); | |||
| 5 | + | sseEmitter.setMaxListeners(100); ⏎̸ | |||
M src/lib/states/chat.svelte.ts +171 -64
| @@ -5,6 +5,7 @@ import type { | |||
| 5 | 5 | GetMessagesResponse | |
| 6 | 6 | } from '../types/payloads'; | |
| 7 | 7 | import { invalidate } from '$app/navigation'; | |
| 8 | + | import { DEFAULT_CHAT_ROOM_ID } from '$lib/db/schema'; | |
| 8 | 9 | ||
| 9 | 10 | ||
| 10 | 11 | class ChatState { | |
| @@ -16,7 +17,7 @@ class ChatState { | |||
| 16 | 17 | private maxReconnectAttempts = 5; | |
| 17 | 18 | private reconnectDelay = 1000; // Start with 1 second | |
| 18 | 19 | private isInitializing = $state(false); | |
| 19 | - | private currentRoomId = $state<string>('default'); | |
| 20 | + | private currentRoomId = $state<string>(DEFAULT_CHAT_ROOM_ID); | |
| 20 | 21 | private userCache = $state<Record<string, User>>({}); | |
| 21 | 22 | ||
| 22 | 23 | public async reinitialize() { | |
| @@ -29,11 +30,13 @@ class ChatState { | |||
| 29 | 30 | this.reconnectDelay = 1000; | |
| 30 | 31 | this.messages = []; | |
| 31 | 32 | if (this.currentUser) { | |
| 33 | + | // First set up SSE to receive status updates | |
| 34 | + | this.setupSSE(); | |
| 35 | + | // Then initialize messages and room users | |
| 32 | 36 | await Promise.all([ | |
| 33 | 37 | this.initializeMessages(), | |
| 34 | 38 | this.initializeRoomUsers() | |
| 35 | 39 | ]); | |
| 36 | - | this.setupSSE(); | |
| 37 | 40 | } | |
| 38 | 41 | } | |
| 39 | 42 | ||
| @@ -47,11 +50,15 @@ class ChatState { | |||
| 47 | 50 | this.currentUser = user; | |
| 48 | 51 | ||
| 49 | 52 | if (user) { | |
| 53 | + | // Add the current user to the cache immediately | |
| 54 | + | this.userCache[user.id] = user; | |
| 55 | + | // First set up SSE to receive status updates | |
| 56 | + | this.setupSSE(); | |
| 57 | + | // Then initialize messages and room users | |
| 50 | 58 | await Promise.all([ | |
| 51 | 59 | this.initializeMessages(), | |
| 52 | 60 | this.initializeRoomUsers() | |
| 53 | 61 | ]); | |
| 54 | - | this.setupSSE(); | |
| 55 | 62 | } else { | |
| 56 | 63 | // If no global session, clear SSE and state | |
| 57 | 64 | if (this.eventSource) { | |
| @@ -60,6 +67,7 @@ class ChatState { | |||
| 60 | 67 | } | |
| 61 | 68 | this.messages = []; | |
| 62 | 69 | this.users = []; | |
| 70 | + | this.userCache = {}; | |
| 63 | 71 | } | |
| 64 | 72 | ||
| 65 | 73 | // Invalidate both session and messages to trigger reloads | |
| @@ -86,8 +94,19 @@ class ChatState { | |||
| 86 | 94 | } | |
| 87 | 95 | ||
| 88 | 96 | console.debug('Fetched room buddy list:', data.buddyList); | |
| 89 | - | this.users = data.buddyList; | |
| 90 | - | // Update cache with new buddy list users | |
| 97 | + | // Merge fetched buddy list with existing cached users to preserve user details even if offline | |
| 98 | + | const mergedUsers = new Map<string, typeof data.buddyList[0]>(); | |
| 99 | + | // Add existing cached users | |
| 100 | + | Object.values(this.userCache).forEach((user) => mergedUsers.set(user.id, user)); | |
| 101 | + | ||
| 102 | + | // Overwrite/add with fetched buddy list | |
| 103 | + | data.buddyList.forEach((user: typeof data.buddyList[0]) => { | |
| 104 | + | mergedUsers.set(user.id, user); | |
| 105 | + | }); | |
| 106 | + | ||
| 107 | + | this.users = Array.from(mergedUsers.values()); | |
| 108 | + | ||
| 109 | + | // Update cache with merged users | |
| 91 | 110 | this.users.forEach(user => { | |
| 92 | 111 | this.userCache[user.id] = user; | |
| 93 | 112 | }); | |
| @@ -98,7 +117,16 @@ class ChatState { | |||
| 98 | 117 | } | |
| 99 | 118 | ||
| 100 | 119 | getOnlineUsers() { | |
| 101 | - | return this.users.filter((user: User) => user.status !== 'offline'); | |
| 120 | + | // Sort users by status: online first, then away, then busy, then offline | |
| 121 | + | return [...this.users].sort((a: User, b: User) => { | |
| 122 | + | const statusOrder: Record<User['status'], number> = { | |
| 123 | + | 'online': 0, | |
| 124 | + | 'away': 1, | |
| 125 | + | 'busy': 2, | |
| 126 | + | 'offline': 3 | |
| 127 | + | }; | |
| 128 | + | return statusOrder[a.status] - statusOrder[b.status]; | |
| 129 | + | }); | |
| 102 | 130 | } | |
| 103 | 131 | ||
| 104 | 132 | getUserById(userId: string) { | |
| @@ -106,29 +134,94 @@ class ChatState { | |||
| 106 | 134 | if (this.userCache[userId]) { | |
| 107 | 135 | return this.userCache[userId]; | |
| 108 | 136 | } | |
| 137 | + | ||
| 138 | + | // Return a fallback object and cache it to avoid repeated fetches | |
| 139 | + | const fallback = { | |
| 140 | + | id: userId, | |
| 141 | + | nickname: 'Unknown User', | |
| 142 | + | status: 'offline', | |
| 143 | + | password: '', // system user, no password needed | |
| 144 | + | createdAt: Date.now(), | |
| 145 | + | lastSeen: null, | |
| 146 | + | avatarUrl: null | |
| 147 | + | }; | |
| 148 | + | this.userCache[userId] = fallback; | |
| 149 | + | console.debug('getUserById: returning fallback for missing user', { userId, fallback }); | |
| 150 | + | return fallback; | |
| 151 | + | } | |
| 152 | + | ||
| 153 | + | updateUserStatus(userId: string, status: User['status'], lastSeen?: number) { | |
| 154 | + | console.debug('Updating user status with:', { userId, status, lastSeen, | |
| 155 | + | currentCache: this.userCache[userId], | |
| 156 | + | currentUsers: this.users.find(u => u.id === userId) | |
| 157 | + | }); | |
| 158 | + | ||
| 159 | + | // Try to get the user from cache first | |
| 160 | + | let user: User | undefined = this.userCache[userId]; | |
| 109 | 161 | ||
| 110 | - | // If not in cache, look up in users array | |
| 111 | - | const user = this.users.find((user: User) => user.id === userId); | |
| 112 | - | if (user) { | |
| 113 | - | // Update cache | |
| 114 | - | this.userCache[userId] = user; | |
| 115 | - | console.debug('getUserById: cached user data', { userId, user }); | |
| 116 | - | } else { | |
| 117 | - | console.debug('getUserById: user not found', { userId }); | |
| 162 | + | // If not in cache, try to find in users array | |
| 163 | + | if (!user) { | |
| 164 | + | const foundUser = this.users.find((u: User) => u.id === userId); | |
| 165 | + | if (foundUser) { | |
| 166 | + | user = foundUser; | |
| 167 | + | } | |
| 118 | 168 | } | |
| 119 | - | return user; | |
| 120 | - | } | |
| 121 | 169 | ||
| 122 | - | updateUserStatus(userId: string, status: User['status']) { | |
| 123 | - | this.users = this.users.map((user: User) => { | |
| 124 | - | if (user.id === userId) { | |
| 125 | - | const updatedUser = { ...user, status }; | |
| 126 | - | // Update cache | |
| 127 | - | this.userCache[userId] = updatedUser; | |
| 128 | - | return updatedUser; | |
| 170 | + | if (user) { | |
| 171 | + | const updatedUser: User = { | |
| 172 | + | ...user, | |
| 173 | + | status, | |
| 174 | + | lastSeen: status === 'offline' ? (lastSeen ?? Date.now()) : user.lastSeen | |
| 175 | + | }; | |
| 176 | + | ||
| 177 | + | // Update cache first | |
| 178 | + | this.userCache[userId] = updatedUser; | |
| 179 | + | ||
| 180 | + | // Then update or add to users array | |
| 181 | + | const index = this.users.findIndex((u: User) => u.id === userId); | |
| 182 | + | if (index !== -1) { | |
| 183 | + | // Create new array with updated user | |
| 184 | + | this.users = [ | |
| 185 | + | ...this.users.slice(0, index), | |
| 186 | + | updatedUser, | |
| 187 | + | ...this.users.slice(index + 1) | |
| 188 | + | ]; | |
| 189 | + | } else { | |
| 190 | + | this.users = [...this.users, updatedUser]; | |
| 129 | 191 | } | |
| 130 | - | return user; | |
| 131 | - | }); | |
| 192 | + | ||
| 193 | + | console.debug('User status updated:', { | |
| 194 | + | userId, | |
| 195 | + | newStatus: status, | |
| 196 | + | updatedInCache: this.userCache[userId].status === status, | |
| 197 | + | updatedInList: this.users.find(u => u.id === userId)?.status === status | |
| 198 | + | }); | |
| 199 | + | } else { | |
| 200 | + | // If user not found anywhere, fetch from API | |
| 201 | + | console.debug('User not found in cache or list, fetching from API:', userId); | |
| 202 | + | fetch(`/api/users/${userId}`, { credentials: 'include' }) | |
| 203 | + | .then(response => response.json()) | |
| 204 | + | .then(data => { | |
| 205 | + | if (data.success && data.user) { | |
| 206 | + | const newUser: User = { | |
| 207 | + | ...data.user, | |
| 208 | + | status, | |
| 209 | + | lastSeen: status === 'offline' ? (lastSeen ?? Date.now()) : data.user.lastSeen | |
| 210 | + | }; | |
| 211 | + | // Update both cache and users array | |
| 212 | + | this.userCache[userId] = newUser; | |
| 213 | + | this.users = [...this.users, newUser]; | |
| 214 | + | ||
| 215 | + | console.debug('User fetched and status updated:', { | |
| 216 | + | userId, | |
| 217 | + | status: newUser.status, | |
| 218 | + | addedToCache: !!this.userCache[userId], | |
| 219 | + | addedToList: this.users.some(u => u.id === userId) | |
| 220 | + | }); | |
| 221 | + | } | |
| 222 | + | }) | |
| 223 | + | .catch(error => console.error('Error fetching user data:', error)); | |
| 224 | + | } | |
| 132 | 225 | } | |
| 133 | 226 | ||
| 134 | 227 | // Message methods | |
| @@ -139,11 +232,20 @@ class ChatState { | |||
| 139 | 232 | ||
| 140 | 233 | // Add method to update user cache for public access | |
| 141 | 234 | updateUserCache(users: User[]) { | |
| 142 | - | users.forEach(user => { | |
| 143 | - | this.userCache[user.id] = user; | |
| 144 | - | }); | |
| 145 | - | // Also update the users array | |
| 146 | - | this.users = users; | |
| 235 | + | // Merge incoming users with the existing cache | |
| 236 | + | const mergedUsers = new Map<string, User>(); | |
| 237 | + | ||
| 238 | + | // Add all users from the existing cache | |
| 239 | + | Object.values(this.userCache).forEach(user => mergedUsers.set(user.id, user)); | |
| 240 | + | ||
| 241 | + | // Merge with the new list of users | |
| 242 | + | users.forEach(user => mergedUsers.set(user.id, user)); | |
| 243 | + | ||
| 244 | + | // Update this.users with all merged users | |
| 245 | + | this.users = Array.from(mergedUsers.values()); | |
| 246 | + | ||
| 247 | + | // Update the cache as well | |
| 248 | + | mergedUsers.forEach((user, id) => { this.userCache[id] = user; }); | |
| 147 | 249 | } | |
| 148 | 250 | ||
| 149 | 251 | // Add method to update messages for public access | |
| @@ -210,7 +312,7 @@ class ChatState { | |||
| 210 | 312 | content, | |
| 211 | 313 | type, | |
| 212 | 314 | userId: user.id, | |
| 213 | - | chatRoomId: 'default' | |
| 315 | + | chatRoomId: DEFAULT_CHAT_ROOM_ID | |
| 214 | 316 | }; | |
| 215 | 317 | ||
| 216 | 318 | console.debug('Sending message with payload:', payload); | |
| @@ -251,51 +353,56 @@ class ChatState { | |||
| 251 | 353 | if (this.eventSource) { | |
| 252 | 354 | this.eventSource.close(); | |
| 253 | 355 | } | |
| 254 | - | ||
| 255 | - | this.eventSource = new EventSource('/api/chat/messages', { | |
| 256 | - | withCredentials: true | |
| 257 | - | }); | |
| 258 | 356 | ||
| 259 | - | this.eventSource.onmessage = async (event) => { | |
| 260 | - | try { | |
| 261 | - | if (event.data === 'Connected') { | |
| 262 | - | console.debug('SSE connection established'); | |
| 263 | - | this.reconnectAttempts = 0; | |
| 264 | - | this.reconnectDelay = 1000; | |
| 265 | - | return; | |
| 266 | - | } | |
| 357 | + | // Connect to the unified SSE endpoint | |
| 358 | + | this.eventSource = new EventSource('/api/sse', { withCredentials: true }); | |
| 359 | + | ||
| 360 | + | // Default message event for handshake | |
| 361 | + | this.eventSource.onmessage = (event) => { | |
| 362 | + | if (event.data === 'Connected') { | |
| 363 | + | console.debug('SSE connection established'); | |
| 364 | + | this.reconnectAttempts = 0; | |
| 365 | + | this.reconnectDelay = 1000; | |
| 366 | + | } else { | |
| 367 | + | console.debug('Received unexpected default message:', event.data); | |
| 368 | + | } | |
| 369 | + | }; | |
| 267 | 370 | ||
| 371 | + | // Listen for chat messages | |
| 372 | + | this.eventSource.addEventListener('chatMessage', async (event: MessageEvent) => { | |
| 373 | + | try { | |
| 268 | 374 | const messageData = JSON.parse(event.data) as Message; | |
| 269 | - | ||
| 270 | - | // Ensure we have the sender's data before adding the message | |
| 375 | + | // Ensure the sender's data is available | |
| 271 | 376 | if (!this.userCache[messageData.senderId]) { | |
| 272 | - | try { | |
| 273 | - | console.debug('Fetching user data for new message:', messageData.senderId); | |
| 274 | - | const response = await fetch(`/api/users/${messageData.senderId}`, { | |
| 275 | - | credentials: 'include' | |
| 276 | - | }); | |
| 277 | - | if (response.ok) { | |
| 278 | - | const userData = await response.json(); | |
| 279 | - | if (userData.success && userData.user) { | |
| 280 | - | this.userCache[userData.user.id] = userData.user; | |
| 281 | - | // Only add to users array if not already present | |
| 282 | - | if (!this.users.some(u => u.id === userData.user.id)) { | |
| 283 | - | this.users = [...this.users, userData.user]; | |
| 284 | - | } | |
| 377 | + | console.debug('Fetching user data for new message:', messageData.senderId); | |
| 378 | + | const response = await fetch(`/api/users/${messageData.senderId}`, { credentials: 'include' }); | |
| 379 | + | if (response.ok) { | |
| 380 | + | const userData = await response.json(); | |
| 381 | + | if (userData.success && userData.user) { | |
| 382 | + | this.userCache[userData.user.id] = userData.user; | |
| 383 | + | if (!this.users.some(u => u.id === userData.user.id)) { | |
| 384 | + | this.users = [...this.users, userData.user]; | |
| 285 | 385 | } | |
| 286 | 386 | } | |
| 287 | - | } catch (error) { | |
| 288 | - | console.debug('Error fetching user data for new message:', error); | |
| 289 | 387 | } | |
| 290 | 388 | } | |
| 291 | - | ||
| 292 | - | // Update messages state | |
| 293 | 389 | this.messages = [...this.messages, messageData]; | |
| 294 | 390 | await invalidate('chat:messages'); | |
| 295 | 391 | } catch (error) { | |
| 296 | - | console.debug('Error parsing SSE message data:', error); | |
| 392 | + | console.debug('Error parsing chatMessage SSE data:', error); | |
| 297 | 393 | } | |
| 298 | - | }; | |
| 394 | + | }); | |
| 395 | + | ||
| 396 | + | // Listen for user status updates | |
| 397 | + | this.eventSource.addEventListener('userStatusUpdate', (event: MessageEvent) => { | |
| 398 | + | try { | |
| 399 | + | const userUpdate = JSON.parse(event.data); | |
| 400 | + | console.debug('Received SSE user status update:', userUpdate); | |
| 401 | + | this.updateUserStatus(userUpdate.userId, userUpdate.status, userUpdate.lastSeen); | |
| 402 | + | } catch (error) { | |
| 403 | + | console.debug('Error processing userStatusUpdate event:', error); | |
| 404 | + | } | |
| 405 | + | }); | |
| 299 | 406 | ||
| 300 | 407 | this.eventSource.onerror = (event) => { | |
| 301 | 408 | console.debug('SSE encountered an error:', event); | |
| @@ -5,6 +5,7 @@ import type { | |||||
| 5 | GetMessagesResponse | 5 | GetMessagesResponse | ||
| 6 | } from '../types/payloads'; | 6 | } from '../types/payloads'; | ||
| 7 | import { invalidate } from '$app/navigation'; | 7 | import { invalidate } from '$app/navigation'; | ||
| 8 | + | import { DEFAULT_CHAT_ROOM_ID } from '$lib/db/schema'; | |||
| 8 | 9 | ||||
| 9 | 10 | ||||
| 10 | class ChatState { | 11 | class ChatState { | ||
| @@ -16,7 +17,7 @@ class ChatState { | |||||
| 16 | private maxReconnectAttempts = 5; | 17 | private maxReconnectAttempts = 5; | ||
| 17 | private reconnectDelay = 1000; // Start with 1 second | 18 | private reconnectDelay = 1000; // Start with 1 second | ||
| 18 | private isInitializing = $state(false); | 19 | private isInitializing = $state(false); | ||
| 19 | - | private currentRoomId = $state<string>('default'); | 20 | + | private currentRoomId = $state<string>(DEFAULT_CHAT_ROOM_ID); |
| 20 | private userCache = $state<Record<string, User>>({}); | 21 | private userCache = $state<Record<string, User>>({}); | ||
| 21 | 22 | ||||
| 22 | public async reinitialize() { | 23 | public async reinitialize() { | ||
| @@ -29,11 +30,13 @@ class ChatState { | |||||
| 29 | this.reconnectDelay = 1000; | 30 | this.reconnectDelay = 1000; | ||
| 30 | this.messages = []; | 31 | this.messages = []; | ||
| 31 | if (this.currentUser) { | 32 | if (this.currentUser) { | ||
| 33 | + | // First set up SSE to receive status updates | |||
| 34 | + | this.setupSSE(); | |||
| 35 | + | // Then initialize messages and room users | |||
| 32 | await Promise.all([ | 36 | await Promise.all([ | ||
| 33 | this.initializeMessages(), | 37 | this.initializeMessages(), | ||
| 34 | this.initializeRoomUsers() | 38 | this.initializeRoomUsers() | ||
| 35 | ]); | 39 | ]); | ||
| 36 | - | this.setupSSE(); | |||
| 37 | } | 40 | } | ||
| 38 | } | 41 | } | ||
| 39 | 42 | ||||
| @@ -47,11 +50,15 @@ class ChatState { | |||||
| 47 | this.currentUser = user; | 50 | this.currentUser = user; | ||
| 48 | 51 | ||||
| 49 | if (user) { | 52 | if (user) { | ||
| 53 | + | // Add the current user to the cache immediately | |||
| 54 | + | this.userCache[user.id] = user; | |||
| 55 | + | // First set up SSE to receive status updates | |||
| 56 | + | this.setupSSE(); | |||
| 57 | + | // Then initialize messages and room users | |||
| 50 | await Promise.all([ | 58 | await Promise.all([ | ||
| 51 | this.initializeMessages(), | 59 | this.initializeMessages(), | ||
| 52 | this.initializeRoomUsers() | 60 | this.initializeRoomUsers() | ||
| 53 | ]); | 61 | ]); | ||
| 54 | - | this.setupSSE(); | |||
| 55 | } else { | 62 | } else { | ||
| 56 | // If no global session, clear SSE and state | 63 | // If no global session, clear SSE and state | ||
| 57 | if (this.eventSource) { | 64 | if (this.eventSource) { | ||
| @@ -60,6 +67,7 @@ class ChatState { | |||||
| 60 | } | 67 | } | ||
| 61 | this.messages = []; | 68 | this.messages = []; | ||
| 62 | this.users = []; | 69 | this.users = []; | ||
| 70 | + | this.userCache = {}; | |||
| 63 | } | 71 | } | ||
| 64 | 72 | ||||
| 65 | // Invalidate both session and messages to trigger reloads | 73 | // Invalidate both session and messages to trigger reloads | ||
| @@ -86,8 +94,19 @@ class ChatState { | |||||
| 86 | } | 94 | } | ||
| 87 | 95 | ||||
| 88 | console.debug('Fetched room buddy list:', data.buddyList); | 96 | console.debug('Fetched room buddy list:', data.buddyList); | ||
| 89 | - | this.users = data.buddyList; | 97 | + | // Merge fetched buddy list with existing cached users to preserve user details even if offline |
| 90 | - | // Update cache with new buddy list users | 98 | + | const mergedUsers = new Map<string, typeof data.buddyList[0]>(); |
| 99 | + | // Add existing cached users | |||
| 100 | + | Object.values(this.userCache).forEach((user) => mergedUsers.set(user.id, user)); | |||
| 101 | + | ||||
| 102 | + | // Overwrite/add with fetched buddy list | |||
| 103 | + | data.buddyList.forEach((user: typeof data.buddyList[0]) => { | |||
| 104 | + | mergedUsers.set(user.id, user); | |||
| 105 | + | }); | |||
| 106 | + | ||||
| 107 | + | this.users = Array.from(mergedUsers.values()); | |||
| 108 | + | ||||
| 109 | + | // Update cache with merged users | |||
| 91 | this.users.forEach(user => { | 110 | this.users.forEach(user => { | ||
| 92 | this.userCache[user.id] = user; | 111 | this.userCache[user.id] = user; | ||
| 93 | }); | 112 | }); | ||
| @@ -98,7 +117,16 @@ class ChatState { | |||||
| 98 | } | 117 | } | ||
| 99 | 118 | ||||
| 100 | getOnlineUsers() { | 119 | getOnlineUsers() { | ||
| 101 | - | return this.users.filter((user: User) => user.status !== 'offline'); | 120 | + | // Sort users by status: online first, then away, then busy, then offline |
| 121 | + | return [...this.users].sort((a: User, b: User) => { | |||
| 122 | + | const statusOrder: Record<User['status'], number> = { | |||
| 123 | + | 'online': 0, | |||
| 124 | + | 'away': 1, | |||
| 125 | + | 'busy': 2, | |||
| 126 | + | 'offline': 3 | |||
| 127 | + | }; | |||
| 128 | + | return statusOrder[a.status] - statusOrder[b.status]; | |||
| 129 | + | }); | |||
| 102 | } | 130 | } | ||
| 103 | 131 | ||||
| 104 | getUserById(userId: string) { | 132 | getUserById(userId: string) { | ||
| @@ -106,29 +134,94 @@ class ChatState { | |||||
| 106 | if (this.userCache[userId]) { | 134 | if (this.userCache[userId]) { | ||
| 107 | return this.userCache[userId]; | 135 | return this.userCache[userId]; | ||
| 108 | } | 136 | } | ||
| 137 | + | ||||
| 138 | + | // Return a fallback object and cache it to avoid repeated fetches | |||
| 139 | + | const fallback = { | |||
| 140 | + | id: userId, | |||
| 141 | + | nickname: 'Unknown User', | |||
| 142 | + | status: 'offline', | |||
| 143 | + | password: '', // system user, no password needed | |||
| 144 | + | createdAt: Date.now(), | |||
| 145 | + | lastSeen: null, | |||
| 146 | + | avatarUrl: null | |||
| 147 | + | }; | |||
| 148 | + | this.userCache[userId] = fallback; | |||
| 149 | + | console.debug('getUserById: returning fallback for missing user', { userId, fallback }); | |||
| 150 | + | return fallback; | |||
| 151 | + | } | |||
| 152 | + | ||||
| 153 | + | updateUserStatus(userId: string, status: User['status'], lastSeen?: number) { | |||
| 154 | + | console.debug('Updating user status with:', { userId, status, lastSeen, | |||
| 155 | + | currentCache: this.userCache[userId], | |||
| 156 | + | currentUsers: this.users.find(u => u.id === userId) | |||
| 157 | + | }); | |||
| 158 | + | ||||
| 159 | + | // Try to get the user from cache first | |||
| 160 | + | let user: User | undefined = this.userCache[userId]; | |||
| 109 | 161 | ||||
| 110 | - | // If not in cache, look up in users array | 162 | + | // If not in cache, try to find in users array |
| 111 | - | const user = this.users.find((user: User) => user.id === userId); | 163 | + | if (!user) { |
| 112 | - | if (user) { | 164 | + | const foundUser = this.users.find((u: User) => u.id === userId); |
| 113 | - | // Update cache | 165 | + | if (foundUser) { |
| 114 | - | this.userCache[userId] = user; | 166 | + | user = foundUser; |
| 115 | - | console.debug('getUserById: cached user data', { userId, user }); | 167 | + | } |
| 116 | - | } else { | |||
| 117 | - | console.debug('getUserById: user not found', { userId }); | |||
| 118 | } | 168 | } | ||
| 119 | - | return user; | |||
| 120 | - | } | |||
| 121 | 169 | ||||
| 122 | - | updateUserStatus(userId: string, status: User['status']) { | 170 | + | if (user) { |
| 123 | - | this.users = this.users.map((user: User) => { | 171 | + | const updatedUser: User = { |
| 124 | - | if (user.id === userId) { | 172 | + | ...user, |
| 125 | - | const updatedUser = { ...user, status }; | 173 | + | status, |
| 126 | - | // Update cache | 174 | + | lastSeen: status === 'offline' ? (lastSeen ?? Date.now()) : user.lastSeen |
| 127 | - | this.userCache[userId] = updatedUser; | 175 | + | }; |
| 128 | - | return updatedUser; | 176 | + | |
| 177 | + | // Update cache first | |||
| 178 | + | this.userCache[userId] = updatedUser; | |||
| 179 | + | ||||
| 180 | + | // Then update or add to users array | |||
| 181 | + | const index = this.users.findIndex((u: User) => u.id === userId); | |||
| 182 | + | if (index !== -1) { | |||
| 183 | + | // Create new array with updated user | |||
| 184 | + | this.users = [ | |||
| 185 | + | ...this.users.slice(0, index), | |||
| 186 | + | updatedUser, | |||
| 187 | + | ...this.users.slice(index + 1) | |||
| 188 | + | ]; | |||
| 189 | + | } else { | |||
| 190 | + | this.users = [...this.users, updatedUser]; | |||
| 129 | } | 191 | } | ||
| 130 | - | return user; | 192 | + | |
| 131 | - | }); | 193 | + | console.debug('User status updated:', { |
| 194 | + | userId, | |||
| 195 | + | newStatus: status, | |||
| 196 | + | updatedInCache: this.userCache[userId].status === status, | |||
| 197 | + | updatedInList: this.users.find(u => u.id === userId)?.status === status | |||
| 198 | + | }); | |||
| 199 | + | } else { | |||
| 200 | + | // If user not found anywhere, fetch from API | |||
| 201 | + | console.debug('User not found in cache or list, fetching from API:', userId); | |||
| 202 | + | fetch(`/api/users/${userId}`, { credentials: 'include' }) | |||
| 203 | + | .then(response => response.json()) | |||
| 204 | + | .then(data => { | |||
| 205 | + | if (data.success && data.user) { | |||
| 206 | + | const newUser: User = { | |||
| 207 | + | ...data.user, | |||
| 208 | + | status, | |||
| 209 | + | lastSeen: status === 'offline' ? (lastSeen ?? Date.now()) : data.user.lastSeen | |||
| 210 | + | }; | |||
| 211 | + | // Update both cache and users array | |||
| 212 | + | this.userCache[userId] = newUser; | |||
| 213 | + | this.users = [...this.users, newUser]; | |||
| 214 | + | ||||
| 215 | + | console.debug('User fetched and status updated:', { | |||
| 216 | + | userId, | |||
| 217 | + | status: newUser.status, | |||
| 218 | + | addedToCache: !!this.userCache[userId], | |||
| 219 | + | addedToList: this.users.some(u => u.id === userId) | |||
| 220 | + | }); | |||
| 221 | + | } | |||
| 222 | + | }) | |||
| 223 | + | .catch(error => console.error('Error fetching user data:', error)); | |||
| 224 | + | } | |||
| 132 | } | 225 | } | ||
| 133 | 226 | ||||
| 134 | // Message methods | 227 | // Message methods | ||
| @@ -139,11 +232,20 @@ class ChatState { | |||||
| 139 | 232 | ||||
| 140 | // Add method to update user cache for public access | 233 | // Add method to update user cache for public access | ||
| 141 | updateUserCache(users: User[]) { | 234 | updateUserCache(users: User[]) { | ||
| 142 | - | users.forEach(user => { | 235 | + | // Merge incoming users with the existing cache |
| 143 | - | this.userCache[user.id] = user; | 236 | + | const mergedUsers = new Map<string, User>(); |
| 144 | - | }); | 237 | + | |
| 145 | - | // Also update the users array | 238 | + | // Add all users from the existing cache |
| 146 | - | this.users = users; | 239 | + | Object.values(this.userCache).forEach(user => mergedUsers.set(user.id, user)); |
| 240 | + | ||||
| 241 | + | // Merge with the new list of users | |||
| 242 | + | users.forEach(user => mergedUsers.set(user.id, user)); | |||
| 243 | + | ||||
| 244 | + | // Update this.users with all merged users | |||
| 245 | + | this.users = Array.from(mergedUsers.values()); | |||
| 246 | + | ||||
| 247 | + | // Update the cache as well | |||
| 248 | + | mergedUsers.forEach((user, id) => { this.userCache[id] = user; }); | |||
| 147 | } | 249 | } | ||
| 148 | 250 | ||||
| 149 | // Add method to update messages for public access | 251 | // Add method to update messages for public access | ||
| @@ -210,7 +312,7 @@ class ChatState { | |||||
| 210 | content, | 312 | content, | ||
| 211 | type, | 313 | type, | ||
| 212 | userId: user.id, | 314 | userId: user.id, | ||
| 213 | - | chatRoomId: 'default' | 315 | + | chatRoomId: DEFAULT_CHAT_ROOM_ID |
| 214 | }; | 316 | }; | ||
| 215 | 317 | ||||
| 216 | console.debug('Sending message with payload:', payload); | 318 | console.debug('Sending message with payload:', payload); | ||
| @@ -251,51 +353,56 @@ class ChatState { | |||||
| 251 | if (this.eventSource) { | 353 | if (this.eventSource) { | ||
| 252 | this.eventSource.close(); | 354 | this.eventSource.close(); | ||
| 253 | } | 355 | } | ||
| 254 | - | ||||
| 255 | - | this.eventSource = new EventSource('/api/chat/messages', { | |||
| 256 | - | withCredentials: true | |||
| 257 | - | }); | |||
| 258 | 356 | ||||
| 259 | - | this.eventSource.onmessage = async (event) => { | 357 | + | // Connect to the unified SSE endpoint |
| 260 | - | try { | 358 | + | this.eventSource = new EventSource('/api/sse', { withCredentials: true }); |
| 261 | - | if (event.data === 'Connected') { | 359 | + | |
| 262 | - | console.debug('SSE connection established'); | 360 | + | // Default message event for handshake |
| 263 | - | this.reconnectAttempts = 0; | 361 | + | this.eventSource.onmessage = (event) => { |
| 264 | - | this.reconnectDelay = 1000; | 362 | + | if (event.data === 'Connected') { |
| 265 | - | return; | 363 | + | console.debug('SSE connection established'); |
| 266 | - | } | 364 | + | this.reconnectAttempts = 0; |
| 365 | + | this.reconnectDelay = 1000; | |||
| 366 | + | } else { | |||
| 367 | + | console.debug('Received unexpected default message:', event.data); | |||
| 368 | + | } | |||
| 369 | + | }; | |||
| 267 | 370 | ||||
| 371 | + | // Listen for chat messages | |||
| 372 | + | this.eventSource.addEventListener('chatMessage', async (event: MessageEvent) => { | |||
| 373 | + | try { | |||
| 268 | const messageData = JSON.parse(event.data) as Message; | 374 | const messageData = JSON.parse(event.data) as Message; | ||
| 269 | - | 375 | + | // Ensure the sender's data is available | |
| 270 | - | // Ensure we have the sender's data before adding the message | |||
| 271 | if (!this.userCache[messageData.senderId]) { | 376 | if (!this.userCache[messageData.senderId]) { | ||
| 272 | - | try { | 377 | + | console.debug('Fetching user data for new message:', messageData.senderId); |
| 273 | - | console.debug('Fetching user data for new message:', messageData.senderId); | 378 | + | const response = await fetch(`/api/users/${messageData.senderId}`, { credentials: 'include' }); |
| 274 | - | const response = await fetch(`/api/users/${messageData.senderId}`, { | 379 | + | if (response.ok) { |
| 275 | - | credentials: 'include' | 380 | + | const userData = await response.json(); |
| 276 | - | }); | 381 | + | if (userData.success && userData.user) { |
| 277 | - | if (response.ok) { | 382 | + | this.userCache[userData.user.id] = userData.user; |
| 278 | - | const userData = await response.json(); | 383 | + | if (!this.users.some(u => u.id === userData.user.id)) { |
| 279 | - | if (userData.success && userData.user) { | 384 | + | this.users = [...this.users, userData.user]; |
| 280 | - | this.userCache[userData.user.id] = userData.user; | |||
| 281 | - | // Only add to users array if not already present | |||
| 282 | - | if (!this.users.some(u => u.id === userData.user.id)) { | |||
| 283 | - | this.users = [...this.users, userData.user]; | |||
| 284 | - | } | |||
| 285 | } | 385 | } | ||
| 286 | } | 386 | } | ||
| 287 | - | } catch (error) { | |||
| 288 | - | console.debug('Error fetching user data for new message:', error); | |||
| 289 | } | 387 | } | ||
| 290 | } | 388 | } | ||
| 291 | - | ||||
| 292 | - | // Update messages state | |||
| 293 | this.messages = [...this.messages, messageData]; | 389 | this.messages = [...this.messages, messageData]; | ||
| 294 | await invalidate('chat:messages'); | 390 | await invalidate('chat:messages'); | ||
| 295 | } catch (error) { | 391 | } catch (error) { | ||
| 296 | - | console.debug('Error parsing SSE message data:', error); | 392 | + | console.debug('Error parsing chatMessage SSE data:', error); |
| 297 | } | 393 | } | ||
| 298 | - | }; | 394 | + | }); |
| 395 | + | ||||
| 396 | + | // Listen for user status updates | |||
| 397 | + | this.eventSource.addEventListener('userStatusUpdate', (event: MessageEvent) => { | |||
| 398 | + | try { | |||
| 399 | + | const userUpdate = JSON.parse(event.data); | |||
| 400 | + | console.debug('Received SSE user status update:', userUpdate); | |||
| 401 | + | this.updateUserStatus(userUpdate.userId, userUpdate.status, userUpdate.lastSeen); | |||
| 402 | + | } catch (error) { | |||
| 403 | + | console.debug('Error processing userStatusUpdate event:', error); | |||
| 404 | + | } | |||
| 405 | + | }); | |||
| 299 | 406 | ||||
| 300 | this.eventSource.onerror = (event) => { | 407 | this.eventSource.onerror = (event) => { | ||
| 301 | console.debug('SSE encountered an error:', event); | 408 | console.debug('SSE encountered an error:', event); | ||
M src/routes/+layout.svelte +109 -1
| @@ -3,10 +3,117 @@ | |||
| 3 | 3 | import DesktopIcons from '$lib/components/desktop-icons.svelte'; | |
| 4 | 4 | import ChatRoom from '$lib/components/chat-room.svelte'; | |
| 5 | 5 | import { chatState } from '$lib/states/chat.svelte'; | |
| 6 | + | import { onMount, onDestroy } from 'svelte'; | |
| 7 | + | import { browser } from '$app/environment'; | |
| 6 | 8 | ||
| 7 | 9 | let { children, data } = $props(); | |
| 10 | + | let isVisible = $state(true); | |
| 11 | + | let statusInterval = $state<number | undefined>(undefined); | |
| 12 | + | ||
| 13 | + | async function updateUserStatus(status: 'online' | 'offline' | 'busy') { | |
| 14 | + | if (!data.user?.id) return; | |
| 15 | + | ||
| 16 | + | const timestamp = new Date().toISOString(); | |
| 17 | + | console.debug('Updating user status:', { | |
| 18 | + | userId: data.user.id, | |
| 19 | + | status, | |
| 20 | + | timestamp, | |
| 21 | + | trigger: new Error().stack?.split('\n')[2]?.trim() | |
| 22 | + | }); | |
| 23 | + | ||
| 24 | + | try { | |
| 25 | + | const response = await fetch('/api/status', { | |
| 26 | + | method: 'POST', | |
| 27 | + | headers: { 'Content-Type': 'application/json' }, | |
| 28 | + | body: JSON.stringify({ | |
| 29 | + | status, | |
| 30 | + | lastSeen: timestamp | |
| 31 | + | }) | |
| 32 | + | }); | |
| 33 | + | ||
| 34 | + | if (!response.ok) { | |
| 35 | + | console.error('Failed to update user status:', await response.json()); | |
| 36 | + | } | |
| 37 | + | } catch (error) { | |
| 38 | + | console.error('Error updating user status:', error); | |
| 39 | + | } | |
| 40 | + | } | |
| 41 | + | ||
| 42 | + | function startStatusTicker() { | |
| 43 | + | if (statusInterval) return; | |
| 44 | + | ||
| 45 | + | console.debug('Starting status ticker'); | |
| 46 | + | const tickHandler = () => { | |
| 47 | + | if (isVisible) { | |
| 48 | + | updateUserStatus('online'); | |
| 49 | + | } | |
| 50 | + | }; | |
| 51 | + | ||
| 52 | + | statusInterval = window.setInterval(tickHandler, 60000); // Every minute | |
| 53 | + | // Run immediately on start | |
| 54 | + | tickHandler(); | |
| 55 | + | } | |
| 56 | + | ||
| 57 | + | function stopStatusTicker() { | |
| 58 | + | if (!statusInterval) return; | |
| 59 | + | console.debug('Stopping status ticker'); | |
| 60 | + | window.clearInterval(statusInterval); | |
| 61 | + | statusInterval = undefined; | |
| 62 | + | } | |
| 63 | + | ||
| 64 | + | function handleVisibilityChange() { | |
| 65 | + | if (!browser) return; | |
| 66 | + | isVisible = document.visibilityState === 'visible'; | |
| 67 | + | updateUserStatus(isVisible ? 'online' : 'busy'); | |
| 68 | + | ||
| 69 | + | // Start or stop the ticker based on visibility | |
| 70 | + | if (isVisible) { | |
| 71 | + | startStatusTicker(); | |
| 72 | + | } else { | |
| 73 | + | stopStatusTicker(); | |
| 74 | + | } | |
| 75 | + | } | |
| 76 | + | ||
| 77 | + | function cleanup() { | |
| 78 | + | if (!browser) return; | |
| 79 | + | console.debug('Running cleanup - Setting user status to offline'); | |
| 80 | + | document.removeEventListener('visibilitychange', handleVisibilityChange); | |
| 81 | + | stopStatusTicker(); | |
| 82 | + | updateUserStatus('offline'); | |
| 83 | + | } | |
| 84 | + | ||
| 85 | + | onMount(() => { | |
| 86 | + | if (!browser) return; | |
| 87 | + | ||
| 88 | + | // Set initial online status | |
| 89 | + | updateUserStatus('online'); | |
| 90 | + | ||
| 91 | + | // Start the status ticker | |
| 92 | + | startStatusTicker(); | |
| 93 | + | ||
| 94 | + | // Add visibility change listener | |
| 95 | + | document.addEventListener('visibilitychange', handleVisibilityChange); | |
| 96 | + | ||
| 97 | + | // Add beforeunload listener | |
| 98 | + | window.addEventListener('beforeunload', cleanup); | |
| 99 | + | ||
| 100 | + | // Cleanup on unmount | |
| 101 | + | return () => { | |
| 102 | + | window.removeEventListener('beforeunload', cleanup); | |
| 103 | + | cleanup(); | |
| 104 | + | }; | |
| 105 | + | }); | |
| 106 | + | ||
| 107 | + | // Additional safety net for clean navigation within the app | |
| 108 | + | onDestroy(() => { | |
| 109 | + | if (!browser) return; | |
| 110 | + | console.debug('Component being destroyed - Running cleanup'); | |
| 111 | + | window.removeEventListener('beforeunload', cleanup); | |
| 112 | + | cleanup(); | |
| 113 | + | }); | |
| 8 | 114 | ||
| 9 | 115 | $effect(() => { | |
| 116 | + | if (!browser) return; | |
| 10 | 117 | console.debug('Layout reactive update - Session state:', { | |
| 11 | 118 | timestamp: new Date().toISOString(), | |
| 12 | 119 | hasUser: !!data.user, | |
| @@ -20,7 +127,8 @@ | |||
| 20 | 127 | id: data.session.id, | |
| 21 | 128 | expiresAt: new Date(data.session.expiresAt) | |
| 22 | 129 | } : null, | |
| 23 | - | cookiePresent: document.cookie.includes('session=') | |
| 130 | + | cookiePresent: document.cookie.includes('session='), | |
| 131 | + | isVisible | |
| 24 | 132 | }); | |
| 25 | 133 | }); | |
| 26 | 134 | ||
| @@ -3,10 +3,117 @@ | |||||
| 3 | import DesktopIcons from '$lib/components/desktop-icons.svelte'; | 3 | import DesktopIcons from '$lib/components/desktop-icons.svelte'; | ||
| 4 | import ChatRoom from '$lib/components/chat-room.svelte'; | 4 | import ChatRoom from '$lib/components/chat-room.svelte'; | ||
| 5 | import { chatState } from '$lib/states/chat.svelte'; | 5 | import { chatState } from '$lib/states/chat.svelte'; | ||
| 6 | + | import { onMount, onDestroy } from 'svelte'; | |||
| 7 | + | import { browser } from '$app/environment'; | |||
| 6 | 8 | ||||
| 7 | let { children, data } = $props(); | 9 | let { children, data } = $props(); | ||
| 10 | + | let isVisible = $state(true); | |||
| 11 | + | let statusInterval = $state<number | undefined>(undefined); | |||
| 12 | + | ||||
| 13 | + | async function updateUserStatus(status: 'online' | 'offline' | 'busy') { | |||
| 14 | + | if (!data.user?.id) return; | |||
| 15 | + | ||||
| 16 | + | const timestamp = new Date().toISOString(); | |||
| 17 | + | console.debug('Updating user status:', { | |||
| 18 | + | userId: data.user.id, | |||
| 19 | + | status, | |||
| 20 | + | timestamp, | |||
| 21 | + | trigger: new Error().stack?.split('\n')[2]?.trim() | |||
| 22 | + | }); | |||
| 23 | + | ||||
| 24 | + | try { | |||
| 25 | + | const response = await fetch('/api/status', { | |||
| 26 | + | method: 'POST', | |||
| 27 | + | headers: { 'Content-Type': 'application/json' }, | |||
| 28 | + | body: JSON.stringify({ | |||
| 29 | + | status, | |||
| 30 | + | lastSeen: timestamp | |||
| 31 | + | }) | |||
| 32 | + | }); | |||
| 33 | + | ||||
| 34 | + | if (!response.ok) { | |||
| 35 | + | console.error('Failed to update user status:', await response.json()); | |||
| 36 | + | } | |||
| 37 | + | } catch (error) { | |||
| 38 | + | console.error('Error updating user status:', error); | |||
| 39 | + | } | |||
| 40 | + | } | |||
| 41 | + | ||||
| 42 | + | function startStatusTicker() { | |||
| 43 | + | if (statusInterval) return; | |||
| 44 | + | ||||
| 45 | + | console.debug('Starting status ticker'); | |||
| 46 | + | const tickHandler = () => { | |||
| 47 | + | if (isVisible) { | |||
| 48 | + | updateUserStatus('online'); | |||
| 49 | + | } | |||
| 50 | + | }; | |||
| 51 | + | ||||
| 52 | + | statusInterval = window.setInterval(tickHandler, 60000); // Every minute | |||
| 53 | + | // Run immediately on start | |||
| 54 | + | tickHandler(); | |||
| 55 | + | } | |||
| 56 | + | ||||
| 57 | + | function stopStatusTicker() { | |||
| 58 | + | if (!statusInterval) return; | |||
| 59 | + | console.debug('Stopping status ticker'); | |||
| 60 | + | window.clearInterval(statusInterval); | |||
| 61 | + | statusInterval = undefined; | |||
| 62 | + | } | |||
| 63 | + | ||||
| 64 | + | function handleVisibilityChange() { | |||
| 65 | + | if (!browser) return; | |||
| 66 | + | isVisible = document.visibilityState === 'visible'; | |||
| 67 | + | updateUserStatus(isVisible ? 'online' : 'busy'); | |||
| 68 | + | ||||
| 69 | + | // Start or stop the ticker based on visibility | |||
| 70 | + | if (isVisible) { | |||
| 71 | + | startStatusTicker(); | |||
| 72 | + | } else { | |||
| 73 | + | stopStatusTicker(); | |||
| 74 | + | } | |||
| 75 | + | } | |||
| 76 | + | ||||
| 77 | + | function cleanup() { | |||
| 78 | + | if (!browser) return; | |||
| 79 | + | console.debug('Running cleanup - Setting user status to offline'); | |||
| 80 | + | document.removeEventListener('visibilitychange', handleVisibilityChange); | |||
| 81 | + | stopStatusTicker(); | |||
| 82 | + | updateUserStatus('offline'); | |||
| 83 | + | } | |||
| 84 | + | ||||
| 85 | + | onMount(() => { | |||
| 86 | + | if (!browser) return; | |||
| 87 | + | ||||
| 88 | + | // Set initial online status | |||
| 89 | + | updateUserStatus('online'); | |||
| 90 | + | ||||
| 91 | + | // Start the status ticker | |||
| 92 | + | startStatusTicker(); | |||
| 93 | + | ||||
| 94 | + | // Add visibility change listener | |||
| 95 | + | document.addEventListener('visibilitychange', handleVisibilityChange); | |||
| 96 | + | ||||
| 97 | + | // Add beforeunload listener | |||
| 98 | + | window.addEventListener('beforeunload', cleanup); | |||
| 99 | + | ||||
| 100 | + | // Cleanup on unmount | |||
| 101 | + | return () => { | |||
| 102 | + | window.removeEventListener('beforeunload', cleanup); | |||
| 103 | + | cleanup(); | |||
| 104 | + | }; | |||
| 105 | + | }); | |||
| 106 | + | ||||
| 107 | + | // Additional safety net for clean navigation within the app | |||
| 108 | + | onDestroy(() => { | |||
| 109 | + | if (!browser) return; | |||
| 110 | + | console.debug('Component being destroyed - Running cleanup'); | |||
| 111 | + | window.removeEventListener('beforeunload', cleanup); | |||
| 112 | + | cleanup(); | |||
| 113 | + | }); | |||
| 8 | 114 | ||||
| 9 | $effect(() => { | 115 | $effect(() => { | ||
| 116 | + | if (!browser) return; | |||
| 10 | console.debug('Layout reactive update - Session state:', { | 117 | console.debug('Layout reactive update - Session state:', { | ||
| 11 | timestamp: new Date().toISOString(), | 118 | timestamp: new Date().toISOString(), | ||
| 12 | hasUser: !!data.user, | 119 | hasUser: !!data.user, | ||
| @@ -20,7 +127,8 @@ | |||||
| 20 | id: data.session.id, | 127 | id: data.session.id, | ||
| 21 | expiresAt: new Date(data.session.expiresAt) | 128 | expiresAt: new Date(data.session.expiresAt) | ||
| 22 | } : null, | 129 | } : null, | ||
| 23 | - | cookiePresent: document.cookie.includes('session=') | 130 | + | cookiePresent: document.cookie.includes('session='), |
| 131 | + | isVisible | |||
| 24 | }); | 132 | }); | ||
| 25 | }); | 133 | }); | ||
| 26 | 134 | ||||
M src/routes/api/chat/messages/+server.ts +50 -68
| @@ -9,68 +9,28 @@ import type { | |||
| 9 | 9 | } from '$lib/types/payloads'; | |
| 10 | 10 | import { desc } from 'drizzle-orm'; | |
| 11 | 11 | import { error } from '@sveltejs/kit'; | |
| 12 | + | import { sseEmitter } from '$lib/sseEmitter'; | |
| 13 | + | import { eq } from 'drizzle-orm'; | |
| 14 | + | import { chatRooms, DEFAULT_CHAT_ROOM_ID } from '$lib/db/schema'; | |
| 15 | + | import { users } from '$lib/db/schema'; | |
| 16 | + | import { ensureDefaultChatRoom } from '$lib/db/schema'; | |
| 12 | 17 | ||
| 13 | - | // Global set to store SSE controllers | |
| 14 | - | const sseClients = new Set<ReadableStreamDefaultController<Uint8Array>>(); | |
| 15 | - | ||
| 16 | - | // GET endpoint: handle both SSE connections and initial message fetches | |
| 18 | + | // GET endpoint: fetch messages from the DB | |
| 17 | 19 | export async function GET({ request, locals }) { | |
| 18 | - | const acceptHeader = request.headers.get('Accept'); | |
| 19 | 20 | const isPublic = new URL(request.url).searchParams.get('public') === 'true'; | |
| 20 | - | ||
| 21 | - | // If Accept header is event-stream, handle as SSE connection | |
| 22 | - | if (acceptHeader?.includes('text/event-stream')) { | |
| 23 | - | // SSE connections require authentication | |
| 24 | - | if (!locals.user) { | |
| 25 | - | throw error(401, 'Authentication required for real-time updates'); | |
| 26 | - | } | |
| 27 | - | ||
| 28 | - | console.debug('Establishing new SSE connection for chat messages'); | |
| 29 | - | let thisController: ReadableStreamDefaultController<Uint8Array>; | |
| 30 | - | let heartbeatInterval: NodeJS.Timeout; | |
| 31 | - | ||
| 32 | - | const stream = new ReadableStream({ | |
| 33 | - | start(controller) { | |
| 34 | - | console.debug('SSE controller started'); | |
| 35 | - | thisController = controller; | |
| 36 | - | sseClients.add(controller); | |
| 37 | - | const encoder = new TextEncoder(); | |
| 38 | - | // Send initial connection message | |
| 39 | - | controller.enqueue(encoder.encode(`data: Connected\n\n`)); | |
| 40 | - | ||
| 41 | - | // Heartbeat to keep the connection alive every 15 seconds | |
| 42 | - | heartbeatInterval = setInterval(() => { | |
| 43 | - | controller.enqueue(encoder.encode(`: heartbeat\n\n`)); | |
| 44 | - | }, 15000); | |
| 45 | - | }, | |
| 46 | - | cancel() { | |
| 47 | - | console.debug('SSE connection cancelled'); | |
| 48 | - | sseClients.delete(thisController); | |
| 49 | - | clearInterval(heartbeatInterval); | |
| 50 | - | } | |
| 51 | - | }); | |
| 52 | - | ||
| 53 | - | return new Response(stream, { | |
| 54 | - | headers: { | |
| 55 | - | 'Content-Type': 'text/event-stream', | |
| 56 | - | 'Cache-Control': 'no-cache', | |
| 57 | - | 'Connection': 'keep-alive' | |
| 58 | - | } | |
| 59 | - | }); | |
| 60 | - | } | |
| 61 | - | ||
| 62 | - | // For regular GET requests, check authentication unless it's a public request | |
| 63 | 21 | if (!isPublic && !locals.user) { | |
| 64 | 22 | throw error(401, 'Authentication required'); | |
| 65 | 23 | } | |
| 66 | 24 | ||
| 67 | 25 | try { | |
| 26 | + | // Ensure default chat room exists | |
| 27 | + | await ensureDefaultChatRoom(db); | |
| 28 | + | ||
| 68 | 29 | console.debug('Fetching messages', { isPublic }); | |
| 69 | 30 | const query = db.select() | |
| 70 | 31 | .from(messages) | |
| 71 | 32 | .orderBy(desc(messages.timestamp)); | |
| 72 | 33 | ||
| 73 | - | // For public requests, limit to 10 most recent messages | |
| 74 | 34 | if (isPublic) { | |
| 75 | 35 | query.limit(10); | |
| 76 | 36 | } else { | |
| @@ -78,8 +38,7 @@ export async function GET({ request, locals }) { | |||
| 78 | 38 | } | |
| 79 | 39 | ||
| 80 | 40 | const fetchedMessages = await query; | |
| 81 | - | ||
| 82 | - | // Return messages in chronological order with success response | |
| 41 | + | ||
| 83 | 42 | const response: GetMessagesResponse = { | |
| 84 | 43 | success: true, | |
| 85 | 44 | messages: fetchedMessages.reverse() | |
| @@ -99,7 +58,7 @@ export async function GET({ request, locals }) { | |||
| 99 | 58 | error: 'Failed to fetch messages' | |
| 100 | 59 | }; | |
| 101 | 60 | return new Response( | |
| 102 | - | JSON.stringify(errorResponse), | |
| 61 | + | JSON.stringify(errorResponse), | |
| 103 | 62 | { | |
| 104 | 63 | status: 500, | |
| 105 | 64 | headers: { 'Content-Type': 'application/json' } | |
| @@ -110,15 +69,16 @@ export async function GET({ request, locals }) { | |||
| 110 | 69 | ||
| 111 | 70 | // POST endpoint: receive a new message, save it in the DB, and broadcast via SSE | |
| 112 | 71 | export async function POST({ request, locals }: { request: Request, locals: App.Locals }) { | |
| 113 | - | // Require authentication for posting messages | |
| 114 | 72 | if (!locals.user) { | |
| 115 | 73 | throw error(401, 'Authentication required'); | |
| 116 | 74 | } | |
| 117 | 75 | ||
| 118 | 76 | console.debug('Received new message POST'); | |
| 119 | 77 | try { | |
| 78 | + | // Ensure default chat room exists | |
| 79 | + | await ensureDefaultChatRoom(db); | |
| 80 | + | ||
| 120 | 81 | const data = await request.json() as SendMessageRequest; | |
| 121 | - | // Validate input payload | |
| 122 | 82 | if (!data.content || !data.userId) { | |
| 123 | 83 | console.debug('Invalid payload', data); | |
| 124 | 84 | const errorResponse: SendMessageResponse = { | |
| @@ -128,35 +88,57 @@ export async function POST({ request, locals }: { request: Request, locals: App. | |||
| 128 | 88 | return new Response(JSON.stringify(errorResponse), { status: 400 }); | |
| 129 | 89 | } | |
| 130 | 90 | ||
| 131 | - | // Verify the user is posting as themselves | |
| 132 | 91 | if (data.userId !== locals.user.id) { | |
| 133 | 92 | throw error(403, 'Cannot post messages as another user'); | |
| 134 | 93 | } | |
| 135 | 94 | ||
| 136 | - | // Create a new message object | |
| 95 | + | // Validate that both the chat room and user exist | |
| 96 | + | const chatRoomId = data.chatRoomId || DEFAULT_CHAT_ROOM_ID; | |
| 97 | + | ||
| 98 | + | // Check chat room existence | |
| 99 | + | const chatRoom = await db.select() | |
| 100 | + | .from(chatRooms) | |
| 101 | + | .where(eq(chatRooms.id, chatRoomId)) | |
| 102 | + | .get(); | |
| 103 | + | ||
| 104 | + | if (!chatRoom) { | |
| 105 | + | console.debug('Chat room not found:', chatRoomId); | |
| 106 | + | const errorResponse: SendMessageResponse = { | |
| 107 | + | success: false, | |
| 108 | + | error: 'Chat room not found' | |
| 109 | + | }; | |
| 110 | + | return new Response(JSON.stringify(errorResponse), { status: 404 }); | |
| 111 | + | } | |
| 112 | + | ||
| 113 | + | // Check user existence | |
| 114 | + | const user = await db.select() | |
| 115 | + | .from(users) | |
| 116 | + | .where(eq(users.id, data.userId)) | |
| 117 | + | .get(); | |
| 118 | + | ||
| 119 | + | if (!user) { | |
| 120 | + | console.debug('User not found:', data.userId); | |
| 121 | + | const errorResponse: SendMessageResponse = { | |
| 122 | + | success: false, | |
| 123 | + | error: 'User not found' | |
| 124 | + | }; | |
| 125 | + | return new Response(JSON.stringify(errorResponse), { status: 404 }); | |
| 126 | + | } | |
| 127 | + | ||
| 137 | 128 | const newMessage: Message = { | |
| 138 | 129 | id: uuidv4(), | |
| 139 | - | chatRoomId: data.chatRoomId || 'default', | |
| 130 | + | chatRoomId: data.chatRoomId || DEFAULT_CHAT_ROOM_ID, | |
| 140 | 131 | senderId: data.userId, | |
| 141 | 132 | content: data.content, | |
| 142 | 133 | type: data.type || 'chat', | |
| 143 | 134 | timestamp: Date.now() | |
| 144 | 135 | }; | |
| 145 | 136 | ||
| 146 | - | // Save the message in the database | |
| 147 | 137 | await db.insert(messages).values(newMessage); | |
| 148 | 138 | console.debug('Message saved in DB', newMessage); | |
| 149 | 139 | ||
| 150 | - | // Broadcast the new message to all connected SSE clients | |
| 151 | - | const encoder = new TextEncoder(); | |
| 152 | - | const sseData = `data: ${JSON.stringify(newMessage)}\n\n`; | |
| 153 | - | sseClients.forEach(controller => { | |
| 154 | - | try { | |
| 155 | - | controller.enqueue(encoder.encode(sseData)); | |
| 156 | - | } catch (error) { | |
| 157 | - | console.debug('Error broadcasting to a client:', error); | |
| 158 | - | } | |
| 159 | - | }); | |
| 140 | + | // Broadcast the new message via the unified SSE emitter | |
| 141 | + | sseEmitter.emit('sse', { type: 'chatMessage', data: newMessage }); | |
| 160 | 142 | ||
| 161 | 143 | const successResponse: SendMessageResponse = { | |
| 162 | 144 | success: true, | |
| @@ -9,68 +9,28 @@ import type { | |||||
| 9 | } from '$lib/types/payloads'; | 9 | } from '$lib/types/payloads'; | ||
| 10 | import { desc } from 'drizzle-orm'; | 10 | import { desc } from 'drizzle-orm'; | ||
| 11 | import { error } from '@sveltejs/kit'; | 11 | import { error } from '@sveltejs/kit'; | ||
| 12 | + | import { sseEmitter } from '$lib/sseEmitter'; | |||
| 13 | + | import { eq } from 'drizzle-orm'; | |||
| 14 | + | import { chatRooms, DEFAULT_CHAT_ROOM_ID } from '$lib/db/schema'; | |||
| 15 | + | import { users } from '$lib/db/schema'; | |||
| 16 | + | import { ensureDefaultChatRoom } from '$lib/db/schema'; | |||
| 12 | 17 | ||||
| 13 | - | // Global set to store SSE controllers | 18 | + | // GET endpoint: fetch messages from the DB |
| 14 | - | const sseClients = new Set<ReadableStreamDefaultController<Uint8Array>>(); | |||
| 15 | - | ||||
| 16 | - | // GET endpoint: handle both SSE connections and initial message fetches | |||
| 17 | export async function GET({ request, locals }) { | 19 | export async function GET({ request, locals }) { | ||
| 18 | - | const acceptHeader = request.headers.get('Accept'); | |||
| 19 | const isPublic = new URL(request.url).searchParams.get('public') === 'true'; | 20 | const isPublic = new URL(request.url).searchParams.get('public') === 'true'; | ||
| 20 | - | ||||
| 21 | - | // If Accept header is event-stream, handle as SSE connection | |||
| 22 | - | if (acceptHeader?.includes('text/event-stream')) { | |||
| 23 | - | // SSE connections require authentication | |||
| 24 | - | if (!locals.user) { | |||
| 25 | - | throw error(401, 'Authentication required for real-time updates'); | |||
| 26 | - | } | |||
| 27 | - | ||||
| 28 | - | console.debug('Establishing new SSE connection for chat messages'); | |||
| 29 | - | let thisController: ReadableStreamDefaultController<Uint8Array>; | |||
| 30 | - | let heartbeatInterval: NodeJS.Timeout; | |||
| 31 | - | ||||
| 32 | - | const stream = new ReadableStream({ | |||
| 33 | - | start(controller) { | |||
| 34 | - | console.debug('SSE controller started'); | |||
| 35 | - | thisController = controller; | |||
| 36 | - | sseClients.add(controller); | |||
| 37 | - | const encoder = new TextEncoder(); | |||
| 38 | - | // Send initial connection message | |||
| 39 | - | controller.enqueue(encoder.encode(`data: Connected\n\n`)); | |||
| 40 | - | ||||
| 41 | - | // Heartbeat to keep the connection alive every 15 seconds | |||
| 42 | - | heartbeatInterval = setInterval(() => { | |||
| 43 | - | controller.enqueue(encoder.encode(`: heartbeat\n\n`)); | |||
| 44 | - | }, 15000); | |||
| 45 | - | }, | |||
| 46 | - | cancel() { | |||
| 47 | - | console.debug('SSE connection cancelled'); | |||
| 48 | - | sseClients.delete(thisController); | |||
| 49 | - | clearInterval(heartbeatInterval); | |||
| 50 | - | } | |||
| 51 | - | }); | |||
| 52 | - | ||||
| 53 | - | return new Response(stream, { | |||
| 54 | - | headers: { | |||
| 55 | - | 'Content-Type': 'text/event-stream', | |||
| 56 | - | 'Cache-Control': 'no-cache', | |||
| 57 | - | 'Connection': 'keep-alive' | |||
| 58 | - | } | |||
| 59 | - | }); | |||
| 60 | - | } | |||
| 61 | - | ||||
| 62 | - | // For regular GET requests, check authentication unless it's a public request | |||
| 63 | if (!isPublic && !locals.user) { | 21 | if (!isPublic && !locals.user) { | ||
| 64 | throw error(401, 'Authentication required'); | 22 | throw error(401, 'Authentication required'); | ||
| 65 | } | 23 | } | ||
| 66 | 24 | ||||
| 67 | try { | 25 | try { | ||
| 26 | + | // Ensure default chat room exists | |||
| 27 | + | await ensureDefaultChatRoom(db); | |||
| 28 | + | ||||
| 68 | console.debug('Fetching messages', { isPublic }); | 29 | console.debug('Fetching messages', { isPublic }); | ||
| 69 | const query = db.select() | 30 | const query = db.select() | ||
| 70 | .from(messages) | 31 | .from(messages) | ||
| 71 | .orderBy(desc(messages.timestamp)); | 32 | .orderBy(desc(messages.timestamp)); | ||
| 72 | 33 | ||||
| 73 | - | // For public requests, limit to 10 most recent messages | |||
| 74 | if (isPublic) { | 34 | if (isPublic) { | ||
| 75 | query.limit(10); | 35 | query.limit(10); | ||
| 76 | } else { | 36 | } else { | ||
| @@ -78,8 +38,7 @@ export async function GET({ request, locals }) { | |||||
| 78 | } | 38 | } | ||
| 79 | 39 | ||||
| 80 | const fetchedMessages = await query; | 40 | const fetchedMessages = await query; | ||
| 81 | - | 41 | + | ||
| 82 | - | // Return messages in chronological order with success response | |||
| 83 | const response: GetMessagesResponse = { | 42 | const response: GetMessagesResponse = { | ||
| 84 | success: true, | 43 | success: true, | ||
| 85 | messages: fetchedMessages.reverse() | 44 | messages: fetchedMessages.reverse() | ||
| @@ -99,7 +58,7 @@ export async function GET({ request, locals }) { | |||||
| 99 | error: 'Failed to fetch messages' | 58 | error: 'Failed to fetch messages' | ||
| 100 | }; | 59 | }; | ||
| 101 | return new Response( | 60 | return new Response( | ||
| 102 | - | JSON.stringify(errorResponse), | 61 | + | JSON.stringify(errorResponse), |
| 103 | { | 62 | { | ||
| 104 | status: 500, | 63 | status: 500, | ||
| 105 | headers: { 'Content-Type': 'application/json' } | 64 | headers: { 'Content-Type': 'application/json' } | ||
| @@ -110,15 +69,16 @@ export async function GET({ request, locals }) { | |||||
| 110 | 69 | ||||
| 111 | // POST endpoint: receive a new message, save it in the DB, and broadcast via SSE | 70 | // POST endpoint: receive a new message, save it in the DB, and broadcast via SSE | ||
| 112 | export async function POST({ request, locals }: { request: Request, locals: App.Locals }) { | 71 | export async function POST({ request, locals }: { request: Request, locals: App.Locals }) { | ||
| 113 | - | // Require authentication for posting messages | |||
| 114 | if (!locals.user) { | 72 | if (!locals.user) { | ||
| 115 | throw error(401, 'Authentication required'); | 73 | throw error(401, 'Authentication required'); | ||
| 116 | } | 74 | } | ||
| 117 | 75 | ||||
| 118 | console.debug('Received new message POST'); | 76 | console.debug('Received new message POST'); | ||
| 119 | try { | 77 | try { | ||
| 78 | + | // Ensure default chat room exists | |||
| 79 | + | await ensureDefaultChatRoom(db); | |||
| 80 | + | ||||
| 120 | const data = await request.json() as SendMessageRequest; | 81 | const data = await request.json() as SendMessageRequest; | ||
| 121 | - | // Validate input payload | |||
| 122 | if (!data.content || !data.userId) { | 82 | if (!data.content || !data.userId) { | ||
| 123 | console.debug('Invalid payload', data); | 83 | console.debug('Invalid payload', data); | ||
| 124 | const errorResponse: SendMessageResponse = { | 84 | const errorResponse: SendMessageResponse = { | ||
| @@ -128,35 +88,57 @@ export async function POST({ request, locals }: { request: Request, locals: App. | |||||
| 128 | return new Response(JSON.stringify(errorResponse), { status: 400 }); | 88 | return new Response(JSON.stringify(errorResponse), { status: 400 }); | ||
| 129 | } | 89 | } | ||
| 130 | 90 | ||||
| 131 | - | // Verify the user is posting as themselves | |||
| 132 | if (data.userId !== locals.user.id) { | 91 | if (data.userId !== locals.user.id) { | ||
| 133 | throw error(403, 'Cannot post messages as another user'); | 92 | throw error(403, 'Cannot post messages as another user'); | ||
| 134 | } | 93 | } | ||
| 135 | 94 | ||||
| 136 | - | // Create a new message object | 95 | + | // Validate that both the chat room and user exist |
| 96 | + | const chatRoomId = data.chatRoomId || DEFAULT_CHAT_ROOM_ID; | |||
| 97 | + | ||||
| 98 | + | // Check chat room existence | |||
| 99 | + | const chatRoom = await db.select() | |||
| 100 | + | .from(chatRooms) | |||
| 101 | + | .where(eq(chatRooms.id, chatRoomId)) | |||
| 102 | + | .get(); | |||
| 103 | + | ||||
| 104 | + | if (!chatRoom) { | |||
| 105 | + | console.debug('Chat room not found:', chatRoomId); | |||
| 106 | + | const errorResponse: SendMessageResponse = { | |||
| 107 | + | success: false, | |||
| 108 | + | error: 'Chat room not found' | |||
| 109 | + | }; | |||
| 110 | + | return new Response(JSON.stringify(errorResponse), { status: 404 }); | |||
| 111 | + | } | |||
| 112 | + | ||||
| 113 | + | // Check user existence | |||
| 114 | + | const user = await db.select() | |||
| 115 | + | .from(users) | |||
| 116 | + | .where(eq(users.id, data.userId)) | |||
| 117 | + | .get(); | |||
| 118 | + | ||||
| 119 | + | if (!user) { | |||
| 120 | + | console.debug('User not found:', data.userId); | |||
| 121 | + | const errorResponse: SendMessageResponse = { | |||
| 122 | + | success: false, | |||
| 123 | + | error: 'User not found' | |||
| 124 | + | }; | |||
| 125 | + | return new Response(JSON.stringify(errorResponse), { status: 404 }); | |||
| 126 | + | } | |||
| 127 | + | ||||
| 137 | const newMessage: Message = { | 128 | const newMessage: Message = { | ||
| 138 | id: uuidv4(), | 129 | id: uuidv4(), | ||
| 139 | - | chatRoomId: data.chatRoomId || 'default', | 130 | + | chatRoomId: data.chatRoomId || DEFAULT_CHAT_ROOM_ID, |
| 140 | senderId: data.userId, | 131 | senderId: data.userId, | ||
| 141 | content: data.content, | 132 | content: data.content, | ||
| 142 | type: data.type || 'chat', | 133 | type: data.type || 'chat', | ||
| 143 | timestamp: Date.now() | 134 | timestamp: Date.now() | ||
| 144 | }; | 135 | }; | ||
| 145 | 136 | ||||
| 146 | - | // Save the message in the database | |||
| 147 | await db.insert(messages).values(newMessage); | 137 | await db.insert(messages).values(newMessage); | ||
| 148 | console.debug('Message saved in DB', newMessage); | 138 | console.debug('Message saved in DB', newMessage); | ||
| 149 | 139 | ||||
| 150 | - | // Broadcast the new message to all connected SSE clients | 140 | + | // Broadcast the new message via the unified SSE emitter |
| 151 | - | const encoder = new TextEncoder(); | 141 | + | sseEmitter.emit('sse', { type: 'chatMessage', data: newMessage }); |
| 152 | - | const sseData = `data: ${JSON.stringify(newMessage)}\n\n`; | |||
| 153 | - | sseClients.forEach(controller => { | |||
| 154 | - | try { | |||
| 155 | - | controller.enqueue(encoder.encode(sseData)); | |||
| 156 | - | } catch (error) { | |||
| 157 | - | console.debug('Error broadcasting to a client:', error); | |||
| 158 | - | } | |||
| 159 | - | }); | |||
| 160 | 142 | ||||
| 161 | const successResponse: SendMessageResponse = { | 143 | const successResponse: SendMessageResponse = { | ||
| 162 | success: true, | 144 | success: true, | ||
M src/routes/api/rooms/[roomId]/+server.ts +3 -4
| @@ -1,4 +1,4 @@ | |||
| 1 | - | import { desc, eq, not } from 'drizzle-orm'; | |
| 1 | + | import { desc, eq } from 'drizzle-orm'; | |
| 2 | 2 | import db from '$lib/db/db.server'; | |
| 3 | 3 | import { messages, users } from '$lib/db/schema'; | |
| 4 | 4 | import type { PublicRoomResponse } from '$lib/types/payloads'; | |
| @@ -29,10 +29,9 @@ export async function GET({ params, url, locals }): Promise<Response> { | |||
| 29 | 29 | messageQuery.limit(50); | |
| 30 | 30 | } | |
| 31 | 31 | ||
| 32 | - | // Fetch buddy list: users that are not offline | |
| 32 | + | // Fetch buddy list: get all users in the room (including offline) | |
| 33 | 33 | const fetchedUsers = await db.select() | |
| 34 | - | .from(users) | |
| 35 | - | .where(not(eq(users.status, 'offline'))); | |
| 34 | + | .from(users); | |
| 36 | 35 | ||
| 37 | 36 | // Sanitize user data by removing sensitive fields | |
| 38 | 37 | // eslint-disable-next-line @typescript-eslint/no-unused-vars | |
| @@ -1,4 +1,4 @@ | |||||
| 1 | - | import { desc, eq, not } from 'drizzle-orm'; | 1 | + | import { desc, eq } from 'drizzle-orm'; |
| 2 | import db from '$lib/db/db.server'; | 2 | import db from '$lib/db/db.server'; | ||
| 3 | import { messages, users } from '$lib/db/schema'; | 3 | import { messages, users } from '$lib/db/schema'; | ||
| 4 | import type { PublicRoomResponse } from '$lib/types/payloads'; | 4 | import type { PublicRoomResponse } from '$lib/types/payloads'; | ||
| @@ -29,10 +29,9 @@ export async function GET({ params, url, locals }): Promise<Response> { | |||||
| 29 | messageQuery.limit(50); | 29 | messageQuery.limit(50); | ||
| 30 | } | 30 | } | ||
| 31 | 31 | ||||
| 32 | - | // Fetch buddy list: users that are not offline | 32 | + | // Fetch buddy list: get all users in the room (including offline) |
| 33 | const fetchedUsers = await db.select() | 33 | const fetchedUsers = await db.select() | ||
| 34 | - | .from(users) | 34 | + | .from(users); |
| 35 | - | .where(not(eq(users.status, 'offline'))); | |||
| 36 | 35 | ||||
| 37 | // Sanitize user data by removing sensitive fields | 36 | // Sanitize user data by removing sensitive fields | ||
| 38 | // eslint-disable-next-line @typescript-eslint/no-unused-vars | 37 | // eslint-disable-next-line @typescript-eslint/no-unused-vars | ||
A src/routes/api/sse/+server.ts +147 -0
| @@ -0,0 +1,147 @@ | |||
| 1 | + | import type { RequestHandler } from './$types'; | |
| 2 | + | import { sseEmitter } from '$lib/sseEmitter'; | |
| 3 | + | import db from '$lib/db/db.server'; | |
| 4 | + | import { users } from '$lib/db/schema'; | |
| 5 | + | import { eq, and, lt } from 'drizzle-orm'; | |
| 6 | + | ||
| 7 | + | console.debug('SSE connection initiated'); | |
| 8 | + | ||
| 9 | + | // Constants for timeout checks | |
| 10 | + | const ONLINE_TIMEOUT_MS = 2 * 60 * 1000; // 2 minutes | |
| 11 | + | const CHECK_INTERVAL_MS = 60 * 1000; // Check every minute | |
| 12 | + | ||
| 13 | + | // Initialize periodic check for timed out users | |
| 14 | + | const timeoutInterval = setInterval(async () => { | |
| 15 | + | const timeoutThreshold = Date.now() - ONLINE_TIMEOUT_MS; | |
| 16 | + | console.debug('Checking for timed out users, threshold:', new Date(timeoutThreshold)); | |
| 17 | + | ||
| 18 | + | try { | |
| 19 | + | // Find and update users who are marked as online but haven't been seen recently | |
| 20 | + | const now = Date.now(); | |
| 21 | + | const result = await db.update(users) | |
| 22 | + | .set({ | |
| 23 | + | status: 'offline', | |
| 24 | + | lastSeen: now | |
| 25 | + | }) | |
| 26 | + | .where( | |
| 27 | + | and( | |
| 28 | + | eq(users.status, 'online'), | |
| 29 | + | lt(users.lastSeen, timeoutThreshold) | |
| 30 | + | ) | |
| 31 | + | ) | |
| 32 | + | .returning({ id: users.id }); | |
| 33 | + | ||
| 34 | + | // Broadcast status updates for timed out users | |
| 35 | + | result.forEach(user => { | |
| 36 | + | sseEmitter.emit('sse', { | |
| 37 | + | type: 'userStatusUpdate', | |
| 38 | + | data: { | |
| 39 | + | userId: user.id, | |
| 40 | + | status: 'offline', | |
| 41 | + | lastSeen: now | |
| 42 | + | } | |
| 43 | + | }); | |
| 44 | + | console.debug('User timed out and marked offline:', user.id); | |
| 45 | + | }); | |
| 46 | + | } catch (error) { | |
| 47 | + | console.debug('Error during timeout check:', error); | |
| 48 | + | } | |
| 49 | + | }, CHECK_INTERVAL_MS); | |
| 50 | + | ||
| 51 | + | // Ensure the interval is cleared if the module is reloaded | |
| 52 | + | process.on('beforeExit', () => { | |
| 53 | + | if (timeoutInterval) clearInterval(timeoutInterval); | |
| 54 | + | }); | |
| 55 | + | ||
| 56 | + | export const GET: RequestHandler = async ({ locals }) => { | |
| 57 | + | if (!locals.user) { | |
| 58 | + | return new Response('Unauthorized', { status: 401 }); | |
| 59 | + | } | |
| 60 | + | ||
| 61 | + | const userId = locals.user.id; | |
| 62 | + | const encoder = new TextEncoder(); | |
| 63 | + | let keepAliveInterval: ReturnType<typeof setInterval>; | |
| 64 | + | let onSSE: ((event: { type: string; data: unknown }) => void) | undefined; | |
| 65 | + | ||
| 66 | + | // Update user status to online and lastSeen timestamp when establishing connection | |
| 67 | + | const now = Date.now(); | |
| 68 | + | await db.update(users) | |
| 69 | + | .set({ | |
| 70 | + | status: 'online', | |
| 71 | + | lastSeen: now | |
| 72 | + | }) | |
| 73 | + | .where(eq(users.id, userId)); | |
| 74 | + | ||
| 75 | + | // Broadcast the status update | |
| 76 | + | sseEmitter.emit('sse', { | |
| 77 | + | type: 'userStatusUpdate', | |
| 78 | + | data: { | |
| 79 | + | userId, | |
| 80 | + | status: 'online', | |
| 81 | + | lastSeen: now | |
| 82 | + | } | |
| 83 | + | }); | |
| 84 | + | ||
| 85 | + | const stream = new ReadableStream({ | |
| 86 | + | start(controller) { | |
| 87 | + | const sendEvent = (event: { type: string; data: unknown }) => { | |
| 88 | + | try { | |
| 89 | + | const payload = `event: ${event.type}\n` + | |
| 90 | + | `data: ${JSON.stringify(event.data)}\n\n`; | |
| 91 | + | controller.enqueue(encoder.encode(payload)); | |
| 92 | + | console.debug('Sent SSE event:', event); | |
| 93 | + | } catch (err) { | |
| 94 | + | console.debug('Error sending SSE event:', err); | |
| 95 | + | } | |
| 96 | + | }; | |
| 97 | + | ||
| 98 | + | onSSE = (event: { type: string; data: unknown }) => { | |
| 99 | + | sendEvent(event); | |
| 100 | + | }; | |
| 101 | + | ||
| 102 | + | sseEmitter.addListener('sse', onSSE); | |
| 103 | + | console.debug('Added SSE listener'); | |
| 104 | + | ||
| 105 | + | // Send initial connection message | |
| 106 | + | controller.enqueue(encoder.encode('data: Connected\n\n')); | |
| 107 | + | ||
| 108 | + | // Keep connection alive and update lastSeen: send a comment every 20 seconds | |
| 109 | + | keepAliveInterval = setInterval(async () => { | |
| 110 | + | controller.enqueue(encoder.encode(':\n\n')); | |
| 111 | + | // Update lastSeen timestamp on each keepalive | |
| 112 | + | await db.update(users) | |
| 113 | + | .set({ lastSeen: Date.now() }) | |
| 114 | + | .where(eq(users.id, userId)); | |
| 115 | + | }, 20000); | |
| 116 | + | }, | |
| 117 | + | async cancel(reason) { | |
| 118 | + | if (keepAliveInterval) clearInterval(keepAliveInterval); | |
| 119 | + | if (onSSE) sseEmitter.removeListener('sse', onSSE); | |
| 120 | + | console.debug('SSE connection cancelled:', reason); | |
| 121 | + | ||
| 122 | + | // Update user status to offline in the database | |
| 123 | + | const now = Date.now(); | |
| 124 | + | await db.update(users) | |
| 125 | + | .set({ status: 'offline', lastSeen: now }) | |
| 126 | + | .where(eq(users.id, userId)); | |
| 127 | + | ||
| 128 | + | // Broadcast the status update | |
| 129 | + | sseEmitter.emit('sse', { | |
| 130 | + | type: 'userStatusUpdate', | |
| 131 | + | data: { | |
| 132 | + | userId, | |
| 133 | + | status: 'offline', | |
| 134 | + | lastSeen: now | |
| 135 | + | } | |
| 136 | + | }); | |
| 137 | + | } | |
| 138 | + | }); | |
| 139 | + | ||
| 140 | + | return new Response(stream, { | |
| 141 | + | headers: { | |
| 142 | + | 'Content-Type': 'text/event-stream', | |
| 143 | + | 'Cache-Control': 'no-cache', | |
| 144 | + | 'Connection': 'keep-alive' | |
| 145 | + | } | |
| 146 | + | }); | |
| 147 | + | }; ⏎̸ | |
| @@ -0,0 +1,147 @@ | |||||
| 1 | + | import type { RequestHandler } from './$types'; | |||
| 2 | + | import { sseEmitter } from '$lib/sseEmitter'; | |||
| 3 | + | import db from '$lib/db/db.server'; | |||
| 4 | + | import { users } from '$lib/db/schema'; | |||
| 5 | + | import { eq, and, lt } from 'drizzle-orm'; | |||
| 6 | + | ||||
| 7 | + | console.debug('SSE connection initiated'); | |||
| 8 | + | ||||
| 9 | + | // Constants for timeout checks | |||
| 10 | + | const ONLINE_TIMEOUT_MS = 2 * 60 * 1000; // 2 minutes | |||
| 11 | + | const CHECK_INTERVAL_MS = 60 * 1000; // Check every minute | |||
| 12 | + | ||||
| 13 | + | // Initialize periodic check for timed out users | |||
| 14 | + | const timeoutInterval = setInterval(async () => { | |||
| 15 | + | const timeoutThreshold = Date.now() - ONLINE_TIMEOUT_MS; | |||
| 16 | + | console.debug('Checking for timed out users, threshold:', new Date(timeoutThreshold)); | |||
| 17 | + | ||||
| 18 | + | try { | |||
| 19 | + | // Find and update users who are marked as online but haven't been seen recently | |||
| 20 | + | const now = Date.now(); | |||
| 21 | + | const result = await db.update(users) | |||
| 22 | + | .set({ | |||
| 23 | + | status: 'offline', | |||
| 24 | + | lastSeen: now | |||
| 25 | + | }) | |||
| 26 | + | .where( | |||
| 27 | + | and( | |||
| 28 | + | eq(users.status, 'online'), | |||
| 29 | + | lt(users.lastSeen, timeoutThreshold) | |||
| 30 | + | ) | |||
| 31 | + | ) | |||
| 32 | + | .returning({ id: users.id }); | |||
| 33 | + | ||||
| 34 | + | // Broadcast status updates for timed out users | |||
| 35 | + | result.forEach(user => { | |||
| 36 | + | sseEmitter.emit('sse', { | |||
| 37 | + | type: 'userStatusUpdate', | |||
| 38 | + | data: { | |||
| 39 | + | userId: user.id, | |||
| 40 | + | status: 'offline', | |||
| 41 | + | lastSeen: now | |||
| 42 | + | } | |||
| 43 | + | }); | |||
| 44 | + | console.debug('User timed out and marked offline:', user.id); | |||
| 45 | + | }); | |||
| 46 | + | } catch (error) { | |||
| 47 | + | console.debug('Error during timeout check:', error); | |||
| 48 | + | } | |||
| 49 | + | }, CHECK_INTERVAL_MS); | |||
| 50 | + | ||||
| 51 | + | // Ensure the interval is cleared if the module is reloaded | |||
| 52 | + | process.on('beforeExit', () => { | |||
| 53 | + | if (timeoutInterval) clearInterval(timeoutInterval); | |||
| 54 | + | }); | |||
| 55 | + | ||||
| 56 | + | export const GET: RequestHandler = async ({ locals }) => { | |||
| 57 | + | if (!locals.user) { | |||
| 58 | + | return new Response('Unauthorized', { status: 401 }); | |||
| 59 | + | } | |||
| 60 | + | ||||
| 61 | + | const userId = locals.user.id; | |||
| 62 | + | const encoder = new TextEncoder(); | |||
| 63 | + | let keepAliveInterval: ReturnType<typeof setInterval>; | |||
| 64 | + | let onSSE: ((event: { type: string; data: unknown }) => void) | undefined; | |||
| 65 | + | ||||
| 66 | + | // Update user status to online and lastSeen timestamp when establishing connection | |||
| 67 | + | const now = Date.now(); | |||
| 68 | + | await db.update(users) | |||
| 69 | + | .set({ | |||
| 70 | + | status: 'online', | |||
| 71 | + | lastSeen: now | |||
| 72 | + | }) | |||
| 73 | + | .where(eq(users.id, userId)); | |||
| 74 | + | ||||
| 75 | + | // Broadcast the status update | |||
| 76 | + | sseEmitter.emit('sse', { | |||
| 77 | + | type: 'userStatusUpdate', | |||
| 78 | + | data: { | |||
| 79 | + | userId, | |||
| 80 | + | status: 'online', | |||
| 81 | + | lastSeen: now | |||
| 82 | + | } | |||
| 83 | + | }); | |||
| 84 | + | ||||
| 85 | + | const stream = new ReadableStream({ | |||
| 86 | + | start(controller) { | |||
| 87 | + | const sendEvent = (event: { type: string; data: unknown }) => { | |||
| 88 | + | try { | |||
| 89 | + | const payload = `event: ${event.type}\n` + | |||
| 90 | + | `data: ${JSON.stringify(event.data)}\n\n`; | |||
| 91 | + | controller.enqueue(encoder.encode(payload)); | |||
| 92 | + | console.debug('Sent SSE event:', event); | |||
| 93 | + | } catch (err) { | |||
| 94 | + | console.debug('Error sending SSE event:', err); | |||
| 95 | + | } | |||
| 96 | + | }; | |||
| 97 | + | ||||
| 98 | + | onSSE = (event: { type: string; data: unknown }) => { | |||
| 99 | + | sendEvent(event); | |||
| 100 | + | }; | |||
| 101 | + | ||||
| 102 | + | sseEmitter.addListener('sse', onSSE); | |||
| 103 | + | console.debug('Added SSE listener'); | |||
| 104 | + | ||||
| 105 | + | // Send initial connection message | |||
| 106 | + | controller.enqueue(encoder.encode('data: Connected\n\n')); | |||
| 107 | + | ||||
| 108 | + | // Keep connection alive and update lastSeen: send a comment every 20 seconds | |||
| 109 | + | keepAliveInterval = setInterval(async () => { | |||
| 110 | + | controller.enqueue(encoder.encode(':\n\n')); | |||
| 111 | + | // Update lastSeen timestamp on each keepalive | |||
| 112 | + | await db.update(users) | |||
| 113 | + | .set({ lastSeen: Date.now() }) | |||
| 114 | + | .where(eq(users.id, userId)); | |||
| 115 | + | }, 20000); | |||
| 116 | + | }, | |||
| 117 | + | async cancel(reason) { | |||
| 118 | + | if (keepAliveInterval) clearInterval(keepAliveInterval); | |||
| 119 | + | if (onSSE) sseEmitter.removeListener('sse', onSSE); | |||
| 120 | + | console.debug('SSE connection cancelled:', reason); | |||
| 121 | + | ||||
| 122 | + | // Update user status to offline in the database | |||
| 123 | + | const now = Date.now(); | |||
| 124 | + | await db.update(users) | |||
| 125 | + | .set({ status: 'offline', lastSeen: now }) | |||
| 126 | + | .where(eq(users.id, userId)); | |||
| 127 | + | ||||
| 128 | + | // Broadcast the status update | |||
| 129 | + | sseEmitter.emit('sse', { | |||
| 130 | + | type: 'userStatusUpdate', | |||
| 131 | + | data: { | |||
| 132 | + | userId, | |||
| 133 | + | status: 'offline', | |||
| 134 | + | lastSeen: now | |||
| 135 | + | } | |||
| 136 | + | }); | |||
| 137 | + | } | |||
| 138 | + | }); | |||
| 139 | + | ||||
| 140 | + | return new Response(stream, { | |||
| 141 | + | headers: { | |||
| 142 | + | 'Content-Type': 'text/event-stream', | |||
| 143 | + | 'Cache-Control': 'no-cache', | |||
| 144 | + | 'Connection': 'keep-alive' | |||
| 145 | + | } | |||
| 146 | + | }); | |||
| 147 | + | }; ⏎̸ | |||
A src/routes/api/status/+server.ts +29 -0
| @@ -0,0 +1,29 @@ | |||
| 1 | + | import { error } from '@sveltejs/kit'; | |
| 2 | + | import { sseEmitter } from '$lib/sseEmitter'; | |
| 3 | + | import db from '$lib/db/db.server'; | |
| 4 | + | import { users } from '$lib/db/schema'; | |
| 5 | + | import { eq } from 'drizzle-orm'; | |
| 6 | + | ||
| 7 | + | export async function POST({ request, locals }) { | |
| 8 | + | if (!locals.user) { | |
| 9 | + | throw error(401, 'Authentication required'); | |
| 10 | + | } | |
| 11 | + | ||
| 12 | + | const { status } = await request.json(); | |
| 13 | + | const validStatuses = ['online', 'away', 'busy', 'offline']; | |
| 14 | + | if (!validStatuses.includes(status)) { | |
| 15 | + | throw error(400, 'Invalid status'); | |
| 16 | + | } | |
| 17 | + | ||
| 18 | + | const userId = locals.user.id; | |
| 19 | + | console.debug('Updating status for user', userId, 'to', status); | |
| 20 | + | ||
| 21 | + | // Update the user status in the database | |
| 22 | + | await db.update(users).set({ status, lastSeen: status === 'offline' ? Date.now() : null }).where(eq(users.id, userId)); | |
| 23 | + | console.debug('Status updated in database'); | |
| 24 | + | ||
| 25 | + | // Broadcast the status update via SSE | |
| 26 | + | sseEmitter.emit('sse', { type: 'userStatusUpdate', data: { userId, status, lastSeen: status === 'offline' ? Date.now() : null } }); | |
| 27 | + | ||
| 28 | + | return new Response(JSON.stringify({ success: true }), { status: 200, headers: { 'Content-Type': 'application/json' } }); | |
| 29 | + | } ⏎̸ | |
| @@ -0,0 +1,29 @@ | |||||
| 1 | + | import { error } from '@sveltejs/kit'; | |||
| 2 | + | import { sseEmitter } from '$lib/sseEmitter'; | |||
| 3 | + | import db from '$lib/db/db.server'; | |||
| 4 | + | import { users } from '$lib/db/schema'; | |||
| 5 | + | import { eq } from 'drizzle-orm'; | |||
| 6 | + | ||||
| 7 | + | export async function POST({ request, locals }) { | |||
| 8 | + | if (!locals.user) { | |||
| 9 | + | throw error(401, 'Authentication required'); | |||
| 10 | + | } | |||
| 11 | + | ||||
| 12 | + | const { status } = await request.json(); | |||
| 13 | + | const validStatuses = ['online', 'away', 'busy', 'offline']; | |||
| 14 | + | if (!validStatuses.includes(status)) { | |||
| 15 | + | throw error(400, 'Invalid status'); | |||
| 16 | + | } | |||
| 17 | + | ||||
| 18 | + | const userId = locals.user.id; | |||
| 19 | + | console.debug('Updating status for user', userId, 'to', status); | |||
| 20 | + | ||||
| 21 | + | // Update the user status in the database | |||
| 22 | + | await db.update(users).set({ status, lastSeen: status === 'offline' ? Date.now() : null }).where(eq(users.id, userId)); | |||
| 23 | + | console.debug('Status updated in database'); | |||
| 24 | + | ||||
| 25 | + | // Broadcast the status update via SSE | |||
| 26 | + | sseEmitter.emit('sse', { type: 'userStatusUpdate', data: { userId, status, lastSeen: status === 'offline' ? Date.now() : null } }); | |||
| 27 | + | ||||
| 28 | + | return new Response(JSON.stringify({ success: true }), { status: 200, headers: { 'Content-Type': 'application/json' } }); | |||
| 29 | + | } ⏎̸ | |||
M src/routes/api/users/[userId]/+server.ts +1 -1
| @@ -33,4 +33,4 @@ export const GET: RequestHandler = async ({ params }) => { | |||
| 33 | 33 | console.debug('Error fetching user:', err); | |
| 34 | 34 | throw error(500, 'Failed to fetch user information'); | |
| 35 | 35 | } | |
| 36 | - | }; ⏎̸ | |
| 36 | + | };⏎̸ | |
| @@ -33,4 +33,4 @@ export const GET: RequestHandler = async ({ params }) => { | |||||
| 33 | console.debug('Error fetching user:', err); | 33 | console.debug('Error fetching user:', err); | ||
| 34 | throw error(500, 'Failed to fetch user information'); | 34 | throw error(500, 'Failed to fetch user information'); | ||
| 35 | } | 35 | } | ||
| 36 | - | }; ⏎̸ | 36 | + | };⏎̸ |