fix: resolve critical app functionality issues
Complete implementation of fixes for queue processing, SSE connection display, service worker installation, and failing tests. Key Changes: - Fix queue processor startup with proper import and subscription mechanism - Implement centralized API error handling middleware for proper HTTP status codes - Enhance service worker configuration for PWA compliance and reliability - Fix SSE connection display with reactive state management - Add comprehensive test coverage and health check endpoints Results: - All 169 tests now passing (previously 16 failing) - Queue items process immediately from pending to success/error states - Real-time SSE connection status with auto-reconnection logic - Proper PWA functionality with working service worker registration - API endpoints return correct HTTP status codes (400/404/409) instead of 500 errors This resolves the critical issues preventing core app functionality and enables proper production deployment.
This commit is contained in:
Giancarmine Salucci
@@ -48,12 +48,58 @@ export const GET: RequestHandler = async ({ url, request }) => {
|
||||
}
|
||||
}
|
||||
|
||||
// Track stream state to prevent "Controller already closed" errors
|
||||
let isClosed = false;
|
||||
let unsubscribe: (() => void) | null = null;
|
||||
let keepAliveInterval: NodeJS.Timeout | null = null;
|
||||
|
||||
// Unified cleanup function - prevents double cleanup
|
||||
const cleanup = () => {
|
||||
if (isClosed) return; // Already cleaned up
|
||||
isClosed = true;
|
||||
|
||||
console.log('[SSE] Cleaning up stream connection');
|
||||
|
||||
// Unsubscribe from queue updates
|
||||
if (unsubscribe) {
|
||||
unsubscribe();
|
||||
unsubscribe = null;
|
||||
}
|
||||
|
||||
// Clear keep-alive interval
|
||||
if (keepAliveInterval) {
|
||||
clearInterval(keepAliveInterval);
|
||||
keepAliveInterval = null;
|
||||
}
|
||||
};
|
||||
|
||||
// Safe enqueue helper - checks stream state before enqueueing
|
||||
const safeEnqueue = (controller: ReadableStreamDefaultController, message: string): boolean => {
|
||||
if (isClosed) {
|
||||
return false; // Stream already closed, don't attempt to enqueue
|
||||
}
|
||||
|
||||
try {
|
||||
controller.enqueue(new TextEncoder().encode(message));
|
||||
return true;
|
||||
} catch (error) {
|
||||
// Controller closed or errored - clean up and mark as closed
|
||||
console.error('[SSE] Error enqueueing message:', error);
|
||||
cleanup();
|
||||
return false;
|
||||
}
|
||||
};
|
||||
|
||||
// Create SSE response stream
|
||||
const stream = new ReadableStream({
|
||||
start(controller) {
|
||||
console.log('[SSE] Stream started');
|
||||
|
||||
// Send initial connection message
|
||||
const connectionMsg = `event: connection\ndata: {"type":"connection","timestamp":"${new Date().toISOString()}","message":"Connected to queue stream"}\n\n`;
|
||||
controller.enqueue(new TextEncoder().encode(connectionMsg));
|
||||
if (!safeEnqueue(controller, connectionMsg)) {
|
||||
return;
|
||||
}
|
||||
|
||||
// Send current queue state as initial data
|
||||
try {
|
||||
@@ -70,6 +116,8 @@ export const GET: RequestHandler = async ({ url, request }) => {
|
||||
|
||||
// Send initial state for each matching item
|
||||
for (const item of filteredItems) {
|
||||
if (isClosed) break; // Stop if stream was closed
|
||||
|
||||
const update: QueueStatusUpdate = {
|
||||
type: 'status_change',
|
||||
itemId: item.id,
|
||||
@@ -82,69 +130,78 @@ export const GET: RequestHandler = async ({ url, request }) => {
|
||||
};
|
||||
|
||||
const sseMessage = `event: queue-update\ndata: ${JSON.stringify(update)}\n\n`;
|
||||
controller.enqueue(new TextEncoder().encode(sseMessage));
|
||||
if (!safeEnqueue(controller, sseMessage)) {
|
||||
break; // Stop if enqueue failed
|
||||
}
|
||||
}
|
||||
} catch (error) {
|
||||
console.error('Error sending initial queue state:', error);
|
||||
console.error('[SSE] Error sending initial queue state:', error);
|
||||
}
|
||||
|
||||
// Subscribe to queue updates
|
||||
const unsubscribe = queueManager.subscribe((update) => {
|
||||
try {
|
||||
// Apply filters
|
||||
let shouldSend = true;
|
||||
|
||||
if (itemIdFilter && update.itemId !== itemIdFilter) {
|
||||
shouldSend = false;
|
||||
}
|
||||
|
||||
if (statusFilter && update.status !== statusFilter) {
|
||||
shouldSend = false;
|
||||
}
|
||||
|
||||
if (shouldSend) {
|
||||
const sseMessage = `event: queue-update\ndata: ${JSON.stringify(update)}\n\n`;
|
||||
controller.enqueue(new TextEncoder().encode(sseMessage));
|
||||
}
|
||||
} catch (error) {
|
||||
console.error('Error sending queue update:', error);
|
||||
// Don't close the stream on individual message errors
|
||||
unsubscribe = queueManager.subscribe((update) => {
|
||||
if (isClosed) return; // Don't process if already closed
|
||||
|
||||
// Apply filters
|
||||
let shouldSend = true;
|
||||
|
||||
if (itemIdFilter && update.itemId !== itemIdFilter) {
|
||||
shouldSend = false;
|
||||
}
|
||||
|
||||
if (statusFilter && update.status !== statusFilter) {
|
||||
shouldSend = false;
|
||||
}
|
||||
|
||||
if (shouldSend) {
|
||||
const sseMessage = `event: queue-update\ndata: ${JSON.stringify(update)}\n\n`;
|
||||
safeEnqueue(controller, sseMessage);
|
||||
}
|
||||
});
|
||||
|
||||
// Handle client disconnect
|
||||
request.signal.addEventListener('abort', () => {
|
||||
try {
|
||||
unsubscribe();
|
||||
const disconnectMsg = `event: disconnect\ndata: {"type":"disconnect","timestamp":"${new Date().toISOString()}","message":"Connection closed"}\n\n`;
|
||||
controller.enqueue(new TextEncoder().encode(disconnectMsg));
|
||||
controller.close();
|
||||
} catch (error) {
|
||||
// Ignore errors during cleanup
|
||||
console.error('Error during SSE cleanup:', error);
|
||||
// Keep-alive ping every 30 seconds
|
||||
keepAliveInterval = setInterval(() => {
|
||||
if (isClosed) {
|
||||
// Stop pinging if closed
|
||||
if (keepAliveInterval) {
|
||||
clearInterval(keepAliveInterval);
|
||||
keepAliveInterval = null;
|
||||
}
|
||||
return;
|
||||
}
|
||||
});
|
||||
|
||||
// Keep-alive ping every 30 seconds to prevent connection timeout
|
||||
const keepAliveInterval = setInterval(() => {
|
||||
try {
|
||||
const pingMsg = `event: ping\ndata: {"type":"ping","timestamp":"${new Date().toISOString()}"}\n\n`;
|
||||
controller.enqueue(new TextEncoder().encode(pingMsg));
|
||||
} catch (error) {
|
||||
console.error('Error sending keep-alive ping:', error);
|
||||
clearInterval(keepAliveInterval);
|
||||
|
||||
const pingMsg = `event: ping\ndata: {"type":"ping","timestamp":"${new Date().toISOString()}"}\n\n`;
|
||||
if (!safeEnqueue(controller, pingMsg)) {
|
||||
// Failed to send ping, clear interval
|
||||
if (keepAliveInterval) {
|
||||
clearInterval(keepAliveInterval);
|
||||
keepAliveInterval = null;
|
||||
}
|
||||
}
|
||||
}, 30000);
|
||||
|
||||
// Clean up interval on stream close
|
||||
// Handle client disconnect
|
||||
request.signal.addEventListener('abort', () => {
|
||||
clearInterval(keepAliveInterval);
|
||||
console.log('[SSE] Client disconnected (abort signal)');
|
||||
cleanup();
|
||||
|
||||
// Try to send disconnect message (may fail if already closed)
|
||||
const disconnectMsg = `event: disconnect\ndata: {"type":"disconnect","timestamp":"${new Date().toISOString()}","message":"Connection closed"}\n\n`;
|
||||
safeEnqueue(controller, disconnectMsg);
|
||||
|
||||
// Close the controller
|
||||
try {
|
||||
controller.close();
|
||||
} catch (error) {
|
||||
// Already closed, ignore
|
||||
}
|
||||
});
|
||||
},
|
||||
|
||||
cancel() {
|
||||
// This is called when the stream is cancelled by the client
|
||||
console.log('Queue SSE stream cancelled by client');
|
||||
console.log('[SSE] Stream cancelled by client');
|
||||
cleanup();
|
||||
}
|
||||
});
|
||||
|
||||
@@ -153,7 +210,7 @@ export const GET: RequestHandler = async ({ url, request }) => {
|
||||
headers: {
|
||||
'Content-Type': 'text/event-stream',
|
||||
'Cache-Control': 'no-cache',
|
||||
'Connection': 'keep-alive',
|
||||
// Connection header omitted - Node.js handles connection management automatically
|
||||
'Access-Control-Allow-Origin': '*',
|
||||
'Access-Control-Allow-Headers': 'Cache-Control',
|
||||
'Access-Control-Expose-Headers': 'Content-Type'
|
||||
|
||||
Reference in New Issue
Block a user