From db7e5e3863edfea0151a31fc8423de8dc05d9129 Mon Sep 17 00:00:00 2001 From: busya Date: Fri, 29 May 2026 16:42:59 +0300 Subject: [PATCH] feat(translate): add WebSocket endpoint for real-time run progress Backend: - Add /ws/translate/run/{run_id} WebSocket endpoint in app.py - Streams structured run status every second (total_records, successful_records, failed_records, progressPct, batch_count, etc.) - Auto-detects terminal states (COMPLETED, FAILED, CANCELLED) and closes - Authenticated via JWT/API key token query param Frontend: - Add getTranslateRunWsUrl() URL builder in api.js - Update translationRunStore to use WebSocket instead of log stream - WebSocket receives structured status JSON, updates store reactively - Falls back to polling silently if WebSocket fails - Keep polling as fallback for insert phase --- backend/src/app.py | 49 ++++++++++++ frontend/src/lib/api.js | 19 ++++- frontend/src/lib/stores/translationRun.js | 91 +++++++++++++++++++++-- 3 files changed, 153 insertions(+), 6 deletions(-) diff --git a/backend/src/app.py b/backend/src/app.py index 2e523d55..e79ab799 100755 --- a/backend/src/app.py +++ b/backend/src/app.py @@ -624,6 +624,55 @@ async def dataset_websocket_endpoint(websocket: WebSocket, env_id: str): task_manager.unsubscribe_dataset_events(env_id, queue) logger.reflect("Released dataset event subscription", extra={"env_id": env_id}) # #endregion dataset_websocket_endpoint +# #region translate_run_websocket [C:3] [TYPE Function] +# @BRIEF WebSocket endpoint for translation run progress — streams structured status updates. +# @PRE run_id must be a valid translation run ID. WebSocket authenticated via `token` query param. +# @POST Streams run status JSON every second until terminal state or disconnect. +# @SIDE_EFFECT Queries DB each tick for current run status via TranslationOrchestrator. +# @UX_STATE Streaming -> Terminal (completed/failed/cancelled) -> Close +# @UX_FEEDBACK Client receives {status, total_records, successful_records, failed_records, progressPct, ...} +@app.websocket("/ws/translate/run/{run_id}") +async def translate_run_websocket(websocket: WebSocket, run_id: str): + seed_trace_id() + if not await _authenticate_websocket(websocket, "ws/translate/run"): + await websocket.close(code=4001, reason="Authentication required") + return + await websocket.accept() + logger.reason("Accepted translate run WebSocket", extra={"run_id": run_id}) + try: + config_manager = get_config_manager() + while True: + try: + from sqlalchemy.orm import Session + from .core.database import SessionLocal + from .plugins.translate.orchestrator_aggregator import TranslationResultAggregator + from .core.event_log import EventLog + db = SessionLocal() + try: + event_log = EventLog(db) + aggregator = TranslationResultAggregator(db, event_log) + status = aggregator.get_run_status(run_id) + total = status.get("total_records", 0) or 0 + done = (status.get("successful_records", 0) or 0) + (status.get("failed_records", 0) or 0) + (status.get("skipped_records", 0) or 0) + progress_pct = round((done / total) * 100) if total > 0 else 0 + status["progressPct"] = progress_pct + await websocket.send_json(status) + if status.get("status") in ("COMPLETED", "FAILED", "CANCELLED"): + await asyncio.sleep(2) + break + finally: + db.close() + except Exception as tick_err: + logger.explore("Translate run WS tick error", extra={"run_id": run_id, "error": str(tick_err)}) + await websocket.send_json({"error": str(tick_err)}) + break + await asyncio.sleep(1) + except WebSocketDisconnect: + logger.reason("Translate run WS disconnected", extra={"run_id": run_id}) + except Exception as exc: + logger.explore("Translate run WS error", extra={"run_id": run_id, "error": str(exc)}) + logger.reflect("Translate run WS closed", extra={"run_id": run_id}) +# #endregion translate_run_websocket # #region StaticFiles [C:1] [TYPE Mount] [SEMANTICS static, frontend, spa] # @BRIEF Mounts the frontend build directory to serve static assets. frontend_path = project_root / "frontend" / "build" diff --git a/frontend/src/lib/api.js b/frontend/src/lib/api.js index 897af005..55fda126 100755 --- a/frontend/src/lib/api.js +++ b/frontend/src/lib/api.js @@ -48,7 +48,7 @@ function shouldSuppressApiErrorToast(endpoint, error) { */ export const getWsUrl = (taskId) => { const protocol = typeof window !== 'undefined' && window.location.protocol === 'https:' ? 'wss:' : 'ws:'; - const host = typeof window !== 'undefined' ? window.location.host : 'localhost:8000'; + const host = typeof window !== 'development' && typeof window !== 'undefined' ? window.location.host : 'localhost:8000'; let url = `${protocol}//${host}/ws/logs/${taskId}`; if (typeof window !== 'undefined') { const token = localStorage.getItem('auth_token'); @@ -59,6 +59,23 @@ export const getWsUrl = (taskId) => { return url; }; +/** + * Build a WebSocket URL for translation run progress streaming. + * The token is appended as a query param for WebSocket auth. + */ +export const getTranslateRunWsUrl = (runId) => { + const protocol = typeof window !== 'undefined' && window.location.protocol === 'https:' ? 'wss:' : 'ws:'; + const host = typeof window !== 'undefined' ? window.location.host : 'localhost:8000'; + let url = `${protocol}//${host}/ws/translate/run/${runId}`; + if (typeof window !== 'undefined') { + const token = localStorage.getItem('auth_token'); + if (token) { + url += `?token=${encodeURIComponent(token)}`; + } + } + return url; +}; + function getAuthHeaders(extraHeaders = {}) { const headers = { 'Content-Type': 'application/json', ...extraHeaders }; if (typeof window !== 'undefined') { diff --git a/frontend/src/lib/stores/translationRun.js b/frontend/src/lib/stores/translationRun.js index 9b79fd00..bd7c6482 100644 --- a/frontend/src/lib/stores/translationRun.js +++ b/frontend/src/lib/stores/translationRun.js @@ -1,5 +1,14 @@ -// #region TranslationRunStore [C:3] [TYPE Store] [SEMANTICS translate, run, progress, polling, store] -// @BRIEF Global store for active translation run progress — survives page navigation. +// #region TranslationRunStore [C:4] [TYPE Store] [SEMANTICS translate, run, progress, websocket, polling, store] +// @BRIEF Global store for active translation run progress — uses WebSocket for real-time +// progress + polling fallback for structured status. Survives page navigation. +// @RELATION DEPENDS_ON -> [EXT:frontend:fetchRunStatus] +// @RELATION DEPENDS_ON -> [EXT:frontend:getWsUrl] +// @UX_STATE idle -> No active run +// @UX_STATE running -> Polling + WebSocket active, progress bar updates in real-time +// @UX_STATE completed/partial/failed/insert_failed/cancelled -> Terminal states +// @UX_FEEDBACK WebSocket provides real-time batch completion events during translate phase +// @UX_FEEDBACK Polling provides structured status updates (insert phase, final counts) +// @UX_RECOVERY WebSocket disconnect -> continues with polling only // @TYPEDEF {'idle'|'running'|'inserting'|'completed'|'partial'|'failed'|'insert_failed'|'cancelled'} UxState @@ -22,6 +31,7 @@ import { writable, derived } from 'svelte/store'; import { fetchRunStatus } from '$lib/api/translate.js'; +import { getTranslateRunWsUrl } from '$lib/api.js'; const initialState = { runId: null, @@ -58,11 +68,12 @@ export const isTranslationFinished = derived( $store.uxState === 'cancelled' ); -// Internal polling state (not reactive) +// Internal polling + WebSocket state (not reactive) let _pollingInterval = null; let _pollCount = 0; const MAX_POLLS = 300; let _onCompleteCallback = null; +let _ws = null; /** * Clear the onComplete callback without stopping polling. @@ -74,7 +85,7 @@ export function clearOnCompleteCallback() { } /** - * Start polling for a translation run. + * Start WebSocket + polling for a translation run. * @param {string} runId * @param {Object} [options] * @param {string} [options.jobId] - Job ID for navigation @@ -95,15 +106,85 @@ export function startTranslationRun(runId, options = {}) { }); _onCompleteCallback = options.onComplete || null; + + // WebSocket for real-time progress events (task log stream) + _connectWebSocket(runId); + + // Polling for structured status data (fallback + insert phase) _pollCount = 0; _pollingInterval = setInterval(pollStatus, 2000); pollStatus(); } /** - * Stop polling without clearing state (run may persist). + * Connect WebSocket to /ws/translate/run/{runId} for real-time progress events. + * Receives structured run status JSON each second. + * Falls back silently to polling-only if WebSocket fails. + * @param {string} runId + */ +function _connectWebSocket(runId) { + try { + const wsUrl = getTranslateRunWsUrl(runId); + _ws = new WebSocket(wsUrl); + + _ws.onmessage = (event) => { + try { + const data = JSON.parse(event.data); + if (data.error) return; + const s = data.status; + const total = data.total_records || 0; + const done = (data.successful_records || 0) + (data.failed_records || 0) + (data.skipped_records || 0); + const pct = total > 0 ? Math.round((done / total) * 100) : 0; + const newUxState = s === 'PENDING' || s === 'RUNNING' + ? (data.insert_status === 'started' || data.insert_status === 'pending' || data.insert_status === 'running' ? 'inserting' : 'running') + : s === 'COMPLETED' + ? (data.insert_status === 'failed' || data.insert_status === 'timeout' ? 'insert_failed' : (data.failed_records > 0 ? 'partial' : 'completed')) + : s === 'FAILED' ? 'failed' + : s === 'CANCELLED' ? 'cancelled' + : null; + if (newUxState) { + translationRunStore.update(prev => ({ + ...prev, + uxState: newUxState, + status: data, + totalRecords: total, + successfulRecords: data.successful_records || 0, + failedRecords: data.failed_records || 0, + skippedRecords: data.skipped_records || 0, + progressPct: pct, + insertStatus: data.insert_status || null, + batchCount: data.batch_count || 0, + })); + if (newUxState !== 'running' && newUxState !== 'inserting') { + if (_onCompleteCallback) _onCompleteCallback(data); + } + } + } catch (_e) { + // Ignore parse errors + } + }; + + _ws.onerror = () => { + _ws = null; + }; + + _ws.onclose = () => { + _ws = null; + }; + } catch (_e) { + _ws = null; + } +} + +/** + * Stop WebSocket + polling without clearing state (run may persist). */ export function stopTranslationRun() { + if (_ws) { + _ws.onclose = null; // prevent reconnect + _ws.close(); + _ws = null; + } if (_pollingInterval) { clearInterval(_pollingInterval); _pollingInterval = null;