fix(sse): resolve real-time log rendering on re-extraction using thread-safe loop.call_soon_threadsafe queue and resilient line-by-line SSE parsing

Author: sharriffajar

frontend/app.js

@@ -693,19 +693,32 @@ document.addEventListener('DOMContentLoaded', () => {
         if (done) break;
 
         buffer += decoder.decode(value, { stream: true });
-        const lines = buffer.split('\n\n');
-        buffer = lines.pop();
+        const lines = buffer.split(/\r?\n/);
+        buffer = lines.pop(); // keep last incomplete line in buffer
 
-        for (const block of lines) {
-          if (block.startsWith('data: ')) {
-            const jsonStr = block.replace('data: ', '').trim();
+        for (const line of lines) {
+          const trimmed = line.trim();
+          if (trimmed.startsWith('data:')) {
+            const jsonStr = trimmed.replace(/^data:\s*/, '').trim();
             if (jsonStr) {
-              const event = JSON.parse(jsonStr);
-              handleExtractionEvent(event);
+              try {
+                const event = JSON.parse(jsonStr);
+                handleExtractionEvent(event);
+              } catch (err) {
+                console.warn('Failed to parse SSE JSON:', jsonStr, err);
+              }
             }
           }
         }
       }
+
+      // Flush any remaining buffer on stream close
+      if (buffer && buffer.trim().startsWith('data:')) {
+        try {
+          const event = JSON.parse(buffer.trim().replace(/^data:\s*/, ''));
+          handleExtractionEvent(event);
+        } catch (e) {}
+      }
     } catch (e) {
       if (e.name === 'AbortError') {
         appendTerminalLog('⏹️ Extraction canceled by user.');

routes/extraction.py

@@ -17,6 +17,7 @@
     get_embedder,
     get_qdrant,
     sanitize_error_message,
+    get_persisted_workspace_files,
     get_persisted_document,
     save_persisted_document,
 )
@@ -42,24 +43,26 @@ class ExtractRequest(BaseModel):
 @router.post("/api/extract-jsonld-stream")
 async def extract_jsonld_stream(req: ExtractRequest):
     file_name = req.file_name
-    if file_name not in WORKSPACE_FILES:
+    workspace_files = get_persisted_workspace_files()
+    if file_name not in workspace_files:
         raise HTTPException(status_code=404, detail="File belum diunggah ke workspace.")
     if req.base_url and not is_safe_custom_endpoint(req.base_url):
         raise HTTPException(status_code=400, detail="Disallowed or unsafe custom base_url parameter.")
 
+    loop = asyncio.get_running_loop()
+    log_queue = asyncio.Queue()
+
     # SSE Generator for real-time extraction logs
     async def event_generator():
-        log_queue = asyncio.Queue()
-        
-        def sync_logger(msg: str):
+        def thread_safe_logger(msg: str):
             clean_msg = sanitize_error_message(msg)
-            log_queue.put_nowait({"type": "log", "message": clean_msg})
+            loop.call_soon_threadsafe(log_queue.put_nowait, {"type": "log", "message": clean_msg})
 
         async def run_extraction():
             try:
                 # Ensure chunks exist safely with workspace lock
                 async with _WORKSPACE_LOCK:
-                    fpath = WORKSPACE_FILES[file_name]
+                    fpath = workspace_files[file_name]
                     file_chunks = [c for c in EXTRACTED_CHUNKS if c.get("metadata", {}).get("source") == file_name]
                     if not file_chunks:
                         file_chunks = parse_document(fpath, file_name)
@@ -77,7 +80,7 @@ async def run_extraction():
                     chunks=file_chunks,
                     qdrant_client=qdrant if IS_INDEXED else None,
                     embedder=embedder,
-                    progress_callback=sync_logger,
+                    progress_callback=thread_safe_logger,
                     llm_provider=req.llm_provider,
                     llm_model=req.llm_model,
                     api_key=req.api_key,
@@ -87,16 +90,16 @@ async def run_extraction():
                 existing_record = get_persisted_document(file_name)
                 if existing_record:
                     final_res = merge_and_enrich_json_ld(existing_record, res)
-                    sync_logger("🔄 [Database Optimization] Menggabungkan field & struktur baru dengan data terverifikasi sebelumnya secara non-destruktif.")
+                    thread_safe_logger("🔄 [Database Optimization] Menggabungkan field & struktur baru dengan data terverifikasi sebelumnya secara non-destruktif.")
                 else:
                     final_res = res
 
                 async with _WORKSPACE_LOCK:
                     save_persisted_document(file_name, final_res)
-                await log_queue.put({"type": "complete", "result": final_res})
+                loop.call_soon_threadsafe(log_queue.put_nowait, {"type": "complete", "result": final_res})
             except Exception as e:
                 clean_err = sanitize_error_message(str(e))
-                await log_queue.put({"type": "error", "error": clean_err})
+                loop.call_soon_threadsafe(log_queue.put_nowait, {"type": "error", "error": clean_err})
 
         # Launch extraction task
         task = asyncio.create_task(run_extraction())
@@ -117,7 +120,13 @@ async def run_extraction():
                 pass
             raise
 
-    return StreamingResponse(event_generator(), media_type="text/event-stream")
+    headers = {
+        "Cache-Control": "no-cache",
+        "Connection": "keep-alive",
+        "X-Accel-Buffering": "no",
+        "Content-Type": "text/event-stream; charset=utf-8",
+    }
+    return StreamingResponse(event_generator(), media_type="text/event-stream", headers=headers)
 
 
 @router.get("/api/jsonld/{file_name}")