diff --git a/rvs/server.js b/rvs/server.js index 593c8a7..87e57b5 100644 --- a/rvs/server.js +++ b/rvs/server.js @@ -124,6 +124,86 @@ const ALLOWED_TYPES = new Set([ // Token-Raum: token -> { clients: Set } const rooms = new Map(); +// ── Punkt-zu-Punkt-Antworten ──────────────────────────────────────── +// +// Der RVS ist ein Broadcast-Relay: normalerweise bekommt JEDER Client im Raum +// jede Nachricht. Bei Datei-Antworten ist das fatal. Ein 150-MB-Video wird als +// base64 ~200 MB gross und ging bisher an ALLE Clients im Raum — auch an die +// STT/TTS-Worker, die daran mit Code 1009 (message too big) gestorben sind und +// die gerade laufende Sprachaufnahme mitgerissen haben ("erste Sprachnachricht +// kommt nicht durch"). Die Worker-max_size-Angleichung verhindert den Absturz, +// aber 200 MB an zehn Clients zu schicken bleibt Unsinn: nur EINER hat gefragt. +// +// Darum: wer eine Anfrage mit requestId stellt, wird gemerkt — die Antwort mit +// derselben requestId geht nur an ihn zurueck. Unbekannte requestId → wie +// bisher broadcasten (nichts geht verloren, falls ein Client mithoeren muss, +// dessen Anfrage der RVS nie gesehen hat). +const TARGETED = new Map([ + ["file_request", "file_response"], + ["file_zip_request", "file_zip_response"], + ["file_version_download_request", "file_version_download_response"], + ["brain_request", "brain_response"], +]); +const TARGETED_REPLIES = new Set(TARGETED.values()); + +// `antwortTyp \0 requestId` -> { clients: Set, at } +// Mehrere Clients koennen dieselbe requestId benutzen (die App nimmt die +// Nachrichten-ID) — darum ein Set, sonst bekaeme nur der letzte die Antwort. +const pendingReplies = new Map(); +const PENDING_TTL_MS = 180_000; + +function pendingKey(replyType, requestId) { + return `${replyType}${requestId}`; +} + +/** Merkt sich den Fragenden, wenn die Nachricht eine gezielte Anfrage ist. */ +function rememberRequester(ws, msg) { + const replyType = TARGETED.get(msg.type); + if (!replyType) return; + const rid = msg.payload && msg.payload.requestId; + if (!rid || typeof rid !== "string") return; + const key = pendingKey(replyType, rid); + let entry = pendingReplies.get(key); + if (!entry) { + entry = { clients: new Set(), at: Date.now() }; + pendingReplies.set(key, entry); + } + entry.clients.add(ws); + entry.at = Date.now(); +} + +/** + * Empfaenger fuer eine Antwort-Nachricht. + * null = keine gezielte Antwort → normal broadcasten. + * [] = der Fragende ist weg → verwerfen (NICHT broadcasten, sonst waere + * der ganze Sinn weg und die 200 MB gingen doch an alle). + */ +function targetedRecipients(msg) { + if (!TARGETED_REPLIES.has(msg.type)) return null; + const rid = msg.payload && msg.payload.requestId; + if (!rid || typeof rid !== "string") return null; + const key = pendingKey(msg.type, rid); + const entry = pendingReplies.get(key); + if (!entry) return null; + pendingReplies.delete(key); + return [...entry.clients].filter((c) => c.readyState === 1); +} + +/** Abgelaufene/tote Eintraege wegwerfen (laeuft mit cleanupRooms). */ +function cleanupPendingReplies() { + const now = Date.now(); + for (const [key, entry] of pendingReplies) { + if (now - entry.at > PENDING_TTL_MS) { + pendingReplies.delete(key); + continue; + } + for (const c of entry.clients) { + if (c.readyState > 1) entry.clients.delete(c); + } + if (entry.clients.size === 0) pendingReplies.delete(key); + } +} + // ── Hilfsfunktionen ───────────────────────────────────────────────── function timestamp() { @@ -147,6 +227,7 @@ function cleanupRooms() { log(`Leerer Raum entfernt: ${token.slice(0, 8)}...`); } } + cleanupPendingReplies(); } // ── HTTP + WebSocket Server (hybrid) ──────────────────────────────── @@ -421,6 +502,28 @@ function registerClient(ws, token) { return; } + // Anfrage merken, damit die Antwort gleich gezielt zurueckgehen kann + rememberRequester(ws, msg); + + // Gezielte Antwort? Dann NUR an den Fragenden — nicht an den ganzen Raum. + const targets = targetedRecipients(msg); + if (targets) { + const data = raw.toString(); + let sent = 0; + for (const client of targets) { + if (client !== ws) { client.send(data); sent++; } + } + // Grosse Payloads mitloggen: hier spart der RVS am meisten ein, und + // wenn mal eine Antwort verschwindet will man genau das hier sehen. + if (raw.length > 4 * 1024 * 1024 || sent === 0) { + const mb = (raw.length / 1024 / 1024).toFixed(1); + log(sent === 0 + ? `${msg.type}: Fragender nicht mehr verbunden — ${mb}MB verworfen` + : `${msg.type}: ${mb}MB gezielt an ${sent} Client(s) statt an ${room.clients.size - 1}`); + } + return; + } + // An alle anderen Clients im Raum weiterleiten for (const client of room.clients) { if (client !== ws && client.readyState === 1) { @@ -431,6 +534,9 @@ function registerClient(ws, token) { ws.on("close", () => { room.clients.delete(ws); + for (const [key, entry] of pendingReplies) { + if (entry.clients.delete(ws) && entry.clients.size === 0) pendingReplies.delete(key); + } log(`Client getrennt: ${token.slice(0, 8)}... (${room.clients.size} verbleibend)`); if (room.clients.size === 0) { rooms.delete(token);