busya/ss-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(translate): add WebSocket endpoint for real-time run progress

Browse Source 
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
This commit is contained in:
busya
2026-05-29 16:42:59 +03:00
parent 1aceba465b
commit db7e5e3863
3 changed files with 153 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
backend/src/app.py
View File

@@ -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"

19
frontend/src/lib/api.js
View File

@@ -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') {

91
frontend/src/lib/stores/translationRun.js
View File

@@ -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;