1import { v4 as uuidv4 } from 'uuid';
2import db from '$lib/db/db.server';
3import { messages } from '$lib/db/schema';
4import type { Message } from '$lib/types/chat';
5import type {
6 SendMessageRequest,
7 SendMessageResponse,
8 GetMessagesResponse
9} from '$lib/types/payloads';
10import { desc } from 'drizzle-orm';
11import { error } from '@sveltejs/kit';
12
13// Global set to store SSE controllers
14const sseClients = new Set<ReadableStreamDefaultController<Uint8Array>>();
15
16// GET endpoint: handle both SSE connections and initial message fetches
17export async function GET({ request, locals }) {
18 const acceptHeader = request.headers.get('Accept');
19 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) {
64 throw error(401, 'Authentication required');
65 }
66
67 try {
68 console.debug('Fetching messages', { isPublic });
69 const query = db.select()
70 .from(messages)
71 .orderBy(desc(messages.timestamp));
72
73 // For public requests, limit to 10 most recent messages
74 if (isPublic) {
75 query.limit(10);
76 } else {
77 query.limit(50);
78 }
79
80 const fetchedMessages = await query;
81
82 // Return messages in chronological order with success response
83 const response: GetMessagesResponse = {
84 success: true,
85 messages: fetchedMessages.reverse()
86 };
87
88 return new Response(
89 JSON.stringify(response),
90 {
91 status: 200,
92 headers: { 'Content-Type': 'application/json' }
93 }
94 );
95 } catch (error) {
96 console.debug('Error fetching messages:', error);
97 const errorResponse: GetMessagesResponse = {
98 success: false,
99 error: 'Failed to fetch messages'
100 };
101 return new Response(
102 JSON.stringify(errorResponse),
103 {
104 status: 500,
105 headers: { 'Content-Type': 'application/json' }
106 }
107 );
108 }
109}
110
111// POST endpoint: receive a new message, save it in the DB, and broadcast via SSE
112export async function POST({ request, locals }: { request: Request, locals: App.Locals }) {
113 // Require authentication for posting messages
114 if (!locals.user) {
115 throw error(401, 'Authentication required');
116 }
117
118 console.debug('Received new message POST');
119 try {
120 const data = await request.json() as SendMessageRequest;
121 // Validate input payload
122 if (!data.content || !data.userId) {
123 console.debug('Invalid payload', data);
124 const errorResponse: SendMessageResponse = {
125 success: false,
126 error: 'Invalid message payload'
127 };
128 return new Response(JSON.stringify(errorResponse), { status: 400 });
129 }
130
131 // Verify the user is posting as themselves
132 if (data.userId !== locals.user.id) {
133 throw error(403, 'Cannot post messages as another user');
134 }
135
136 // Create a new message object
137 const newMessage: Message = {
138 id: uuidv4(),
139 chatRoomId: data.chatRoomId || 'default',
140 senderId: data.userId,
141 content: data.content,
142 type: data.type || 'chat',
143 timestamp: Date.now()
144 };
145
146 // Save the message in the database
147 await db.insert(messages).values(newMessage);
148 console.debug('Message saved in DB', newMessage);
149
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 });
160
161 const successResponse: SendMessageResponse = {
162 success: true,
163 message: newMessage
164 };
165
166 return new Response(
167 JSON.stringify(successResponse),
168 {
169 status: 201,
170 headers: { 'Content-Type': 'application/json' }
171 }
172 );
173 } catch (error) {
174 console.debug('Error processing POST:', error);
175 const errorResponse: SendMessageResponse = {
176 success: false,
177 error: 'Failed to save message'
178 };
179 return new Response(
180 JSON.stringify(errorResponse),
181 {
182 status: 500,
183 headers: { 'Content-Type': 'application/json' }
184 }
185 );
186 }
187}