import { logger } from '$lib/core/utils/logger'; import { error, json } from '@sveltejs/kit'; import type { RequestEvent } from '@sveltejs/kit'; import { verifyAuthToken } from '$lib/server/auth'; // In-memory store for streamed catalog data // Format: { batchId: { chunks: [...], status: 'collecting'|'complete'|'error' } } const streamCache = new Map(); const MAX_CACHE_SIZE = 200; // max number of batches in cache const MAX_CHUNKS_PER_BATCH = 500; export async function POST({ request }: RequestEvent) { try { await verifyAuthToken(request); const data = await request.json(); const { batch_id, msg, content, current_chunk, total_chunks } = data.payload ?? {}; // Input validation if (!batch_id || typeof batch_id !== 'string') { throw error(400, 'Missing or invalid batch_id'); } if (!msg || typeof msg !== 'string') { throw error(400, 'Missing or invalid msg'); } if (content !== undefined && content !== null && typeof content !== 'string') { throw error(400, 'content must be a string'); } // Enforce cache size limit — evict oldest batches when over capacity if (!streamCache.has(batch_id) && streamCache.size >= MAX_CACHE_SIZE) { const oldestKey = streamCache.keys().next().value; if (oldestKey !== undefined) { streamCache.delete(oldestKey); logger.warn(`[API] Evicted oldest batch "${oldestKey}" — cache full`); } } // Initialize or update batch if (!streamCache.has(batch_id)) { const cappedTotal = typeof total_chunks === 'number' && total_chunks > 0 ? Math.min(total_chunks, MAX_CHUNKS_PER_BATCH) : MAX_CHUNKS_PER_BATCH; streamCache.set(batch_id, { chunks: [], status: 'collecting', total_chunks: cappedTotal, createdAt: Date.now() }); } const batch = streamCache.get(batch_id); // Handle different message types if (msg === 'start') { logger.info(`[API] Stream started for batch ${batch_id}`); } else if (msg === 'chunk') { if (batch.chunks.length >= MAX_CHUNKS_PER_BATCH) { logger.warn(`[API] Max chunks reached for batch ${batch_id}, ignoring chunk`); return json({ status: 'received', batch_id, warning: 'max_chunks_reached' }); } batch.chunks.push(content); logger.info(`[API] Received chunk ${current_chunk}/${total_chunks} for batch ${batch_id}`); } else if (msg === 'end') { batch.status = 'complete'; logger.info(`[API] Stream complete for batch ${batch_id}, total chunks: ${batch.chunks.length}`); } else if (msg === 'error') { batch.status = 'error'; batch.error = content; logger.info(`[API] Stream error for batch ${batch_id}:`, content); } return json({ status: 'received', batch_id }); } catch (error) { logger.error('[API] Error processing stream:', error); return json({ status: 'error', message: String(error) }, { status: 500 }); } } export async function GET({ request, url }: RequestEvent) { // Verify authentication await verifyAuthToken(request); const batchId = url.searchParams.get('batch_id'); // Clean up old cache entries (older than 5 minutes) const now = Date.now(); for (const [key, value] of streamCache.entries()) { if (now - value.createdAt > 5 * 60 * 1000) { streamCache.delete(key); } } // If batch_id specified, return that specific batch if (batchId) { const batch = streamCache.get(batchId); if (!batch) { return json({ status: 'not_found' }, { status: 404 }); } return json({ batch_id: batchId, status: batch.status, chunks: batch.chunks, total_chunks: batch.total_chunks, error: batch.error || null, createdAt: new Date(batch.createdAt).toISOString() }); } // Otherwise return list of all recent batches const batches = Array.from(streamCache.entries()).map(([batchId, batch]) => ({ batch_id: batchId, status: batch.status, chunks_count: batch.chunks.length, total_chunks: batch.total_chunks, error: batch.error || null, createdAt: new Date(batch.createdAt).toISOString() })); return json({ status: 'success', batches: batches.sort( (a, b) => new Date(b.createdAt).getTime() - new Date(a.createdAt).getTime() ) }); }