ARIA-AGENT/rvs/server.js

"use strict";

const { WebSocketServer } = require("ws");
const http = require("http");
const fs = require("fs");
const path = require("path");

// ── Konfiguration aus Umgebungsvariablen ────────────────────────────
const PORT = parseInt(process.env.PORT || "3000", 10);
const MAX_SESSIONS = parseInt(process.env.MAX_SESSIONS || "10", 10);
const UPDATES_DIR = process.env.UPDATES_DIR || "/updates";
// Kein Polling — APK wird manuell per git pull bereitgestellt

// Erlaubte Nachrichtentypen — alles andere wird verworfen
const ALLOWED_TYPES = new Set([
  "chat", "audio", "file", "location", "mode", "log", "event", "heartbeat",
  "file_request", "file_response", "file_saved", "stt_result", "config", "tts_request",
  "xtts_request", "xtts_response", "xtts_list_voices", "xtts_voices_list", "voice_upload", "xtts_voice_saved",
  "update_check", "update_available", "update_download", "update_data",
  "agent_activity", "cancel_request", "interject",
  "audio_pcm",
  "file_from_aria",
  "container_restart",
  "file_list_request", "file_list_response",
  "file_delete_request", "file_deleted",
  "xtts_export_voice", "xtts_voice_exported",
  "xtts_import_voice", "xtts_voice_imported",
  "skill_created",
  "trigger_created",
  "memory_saved",
  "location_update", "location_tracking",
  "chat_history_request", "chat_history_response", "chat_cleared",
  "delete_message_request", "chat_message_deleted",
  "brain_request", "brain_response",
  "app_log",
  "file_delete_batch_request", "file_delete_batch_response",
  "file_zip_request", "file_zip_response",
  "xtts_delete_voice",
  "voice_preload", "voice_ready",
  "stt_request", "stt_response",
  // Streaming-STT (Phase 1+2): App schickt PCM live an whisper-bridge,
  // die feuert stt_endpoint mit dem finalen Text — kein Audio-Roundtrip.
  "stt_stream_start", "stt_audio_chunk", "stt_stream_end",
  "stt_partial", "stt_endpoint", "stt_stream_done",
  // Speaker-ID / Voice-Enrollment (Phase 1+2): App schickt 5-10 Samples zur
  // whisper-bridge, die berechnet einen Voice-Fingerprint (Embedding-Vektor)
  // und nutzt ihn um nur Stefans Stimme an Whisper STT durchzulassen.
  "voice_id_status_request", "voice_id_status_response",
  "voice_id_enroll_request", "voice_id_enroll_response",
  "voice_id_delete_request", "voice_id_delete_response",
  // Projekte (Stefan-Konzept: Threads im Hauptchat verankert) — Side-Channel-
  // Event vom Brain → Bridge → App/Diagnostic, damit beide Clients ihren
  // aktiven-Projekt-Banner refreshen wenn ARIA via Tool was aendert.
  "project_changed",
  // File-Versioning (Datei-Manager in App): Versionen pro Datei listen,
  // alte Versionen herunterladen, Restore = non-destructive neuer Commit.
  "file_version_list_request", "file_version_list_response",
  "file_version_download_request", "file_version_download_response",
  "file_version_restore_request", "file_version_restore_response",
  "service_status",
  "config_request",
  "flux_request", "flux_response",
  "agent_stream",
  "oauth_callback",
  // Lokales LLM (Plan B) — Router im Brain schickt einfache Turns an das
  // Qwen3 auf der AI-Box (via Bridge → RVS → llm-adapter → llama.cpp).
  // llm_partial ist fuer B2 (Token-Streaming) reserviert, noch ungenutzt.
  "llm_request", "llm_response", "llm_partial",
  // Workspace-Desktop (Code-Projekte): Live-Code-Editor (CodeMirror in der App)
  // spiegelt ARIAs Datei-Writes, und QEMU-VNC wird als RFB-Bytes durch RVS
  // getunnelt (Base64-in-JSON wie audio_pcm — kein Binaer-Handling noetig).
  "code_file", "code_file_edit",
  // M1 Generatives Cockpit: ARIA komponiert via present_view eine View-Spec
  // (Orb + Karten), die App/Web/Diagnostic mit ihrem jeweiligen Renderer
  // materialisieren. Brain → Bridge → RVS → Clients.
  "aria_view",
  "check_desktop", "desktop_status",
  "vnc_open", "vnc_close", "vnc_data", "vnc_input",
  // Satelliten (Info-/Gateway-Aussenposten in fremden Netzen): melden sich mit
  // sat_hello, liefern Geraete-Inventar (sat_devices) auf sat_discover und
  // fuehren Aktionen aus (sat_command → sat_result).
  "sat_hello", "sat_discover", "sat_devices", "sat_command", "sat_result",
  // Satelliten-Credential-Store: Diagnostic legt pro Geraet Zugangsdaten ab
  // (SNMP/HTTP/FritzBox), der Satellit speichert sie verschluesselt.
  "sat_creds_set", "sat_creds_delete", "sat_creds_list",
  "sat_creds_result", "sat_creds_list_result",
  // Host-Agenten: ein Agent laeuft direkt auf einem Rechner, meldet sich mit
  // host_hello/host_ping und fuehrt host_command aus (exec/read/write/info/
  // screenshot) -> host_result. ARIA steuert so Rechner auch hinter NAT.
  "host_hello", "host_ping", "host_command", "host_result",
  // Agent-Historie: Brain schickt pro Host-Aktion ein agent_history-Event (via
  // Bridge), Diagnostic fragt die gespeicherte Historie per agent_history_query
  // ab, die Bridge antwortet mit agent_history_list. OHNE diese Typen verwirft
  // der RVS sie still an der Allow-List und die Historie bleibt leer.
  "agent_history", "agent_history_query", "agent_history_list",
  // STT-Lease: die App fragt vor einer Aufnahme, WELCHE STT-Instanz sie nehmen
  // soll (stt_lease_request -> stt_lease). Fehlten diese Typen hier, lief die
  // Anfrage IMMER in den 250ms-Timeout: jede Aufnahme startete verspaetet (erste
  // Silben weg) und ohne targetInstance nahmen ALLE STT-Worker dieselbe Session
  // an -> doppelte stt_endpoint. chat_ack quittiert getippte Nachrichten; ohne
  // ihn liefen sie in der App nach 3 Versuchen auf "fehlgeschlagen", obwohl sie
  // laengst verarbeitet waren. stt_transcribe_* ist die Wake-Wort-Bestaetigung.
  "stt_lease", "stt_lease_request", "stt_transcribe_blob", "stt_transcribe_result",
  "chat_ack",
  // Raum-Diagnose: Diagnostic fragt die aktuellen RVS-Raeume ab (rooms_query),
  // RVS antwortet direkt mit rooms_info (Fingerprint + Laenge + Client-Zahl je
  // Raum). Deckt Token-Prefix-Kollisionen auf (2 Raeume, gleicher 8-Zeichen-Log).
  "rooms_query", "rooms_info",
  // Compute-Flotte (AI-Boxen): Worker (f5tts/whisper/voxtral/llm-adapter) melden
  // sich per worker_hello an und pingen per worker_ping; der Diagnostic-Server
  // aggregiert das und broadcastet worker_update/worker_list an die Browser-UI.
  // node_stats_* speisen den Auslastungs-Monitor (live nvidia-smi + Historie).
  // OHNE diese Typen verwirft der RVS die Meldungen an der Allow-List (Z. 312),
  // und die Box bleibt in der Flotte unsichtbar, obwohl sie sendet.
  "worker_hello", "worker_ping", "worker_update", "worker_list",
  "node_stats", "node_stats_stream_start", "node_stats_stream_stop",
  "node_stats_history_request", "node_stats_history",
  "node_stats_reset", "node_stats_reset_done",
]);

// Token-Raum: token -> { clients: Set<ws> }
const rooms = new Map();

// ── Hilfsfunktionen ─────────────────────────────────────────────────

function timestamp() {
  return new Date().toISOString();
}

function log(msg) {
  console.log(`[${timestamp()}] ${msg}`);
}

// Leere Räume und tote Clients aufräumen
function cleanupRooms() {
  for (const [token, room] of rooms) {
    // Tote Clients entfernen
    for (const client of room.clients) {
      if (client.readyState > 1) room.clients.delete(client);
    }
    // Raum löschen wenn leer
    if (room.clients.size === 0) {
      rooms.delete(token);
      log(`Leerer Raum entfernt: ${token.slice(0, 8)}...`);
    }
  }
}

// ── HTTP + WebSocket Server (hybrid) ────────────────────────────────
//
// Der gleiche Port handelt jetzt sowohl WebSocket-Upgrades (App, Bridges,
// Diagnostic) als auch normale HTTP-Requests (OAuth-Callbacks von Spotify,
// Google etc.). TLS-Termination passiert wie bisher vor dem RVS-Container
// (Caddy/Nginx); RVS selber bleibt plain HTTP. Wichtig fuer OAuth: aus
// Provider-Sicht ist die Callback-URL `https://{RVS_HOST}:{PORT_oeffentlich}
// /oauth/callback/{service}` — RVS schnappt den ?code=..&state=.., broadcastet
// als WS-Message `oauth_callback` und antwortet dem Browser mit einer
// schoenen "Tab schliessen"-Seite.
//
// maxPayload 1500MB: TTS-Streaming + Voice-Upload (WAV als base64) +
// audio_pcm Chunks koennen die ws-Library Default 1MB ueberschreiten.
// Plus: file_request/file_response fuer Re-Download von Anhaengen.
// 40 MB MP4 → ~53 MB base64 → vorher mit 50 MB Limit zerschossen
// (Code 1009 message too big, Bridge crashed im cleanup). 1500 MB
// deckt bis ~1 GB binaer ab (mit base64 ~33% Overhead + WS-Frame-
// Margin); groessere Files werden Bridge-seitig abgewiesen (siehe
// file_request-Handler) bevor die WS abreisst.
//
// WICHTIG: Node-Default-Heap ist ~1.5 GB. Fuer 1 GB-Files muss der
// Container mit --max-old-space-size=4096 (oder NODE_OPTIONS env var)
// gestartet werden, sonst OOM-Crash beim Empfang.
const httpServer = http.createServer(handleHttpRequest);
const wss = new WebSocketServer({ noServer: true, maxPayload: 1500 * 1024 * 1024 });

// HTTP-Upgrade-Pfad → an WebSocket-Server reichen
httpServer.on("upgrade", (req, socket, head) => {
  wss.handleUpgrade(req, socket, head, (ws) => {
    wss.emit("connection", ws, req);
  });
});

httpServer.listen(PORT, () => {
  log(`RVS läuft auf Port ${PORT} (HTTP + WS) | Max Sessions: ${MAX_SESSIONS}`);
  // Beim Start pruefen ob eine APK da ist
  const apkInfo = getLatestAPK();
  if (apkInfo) log(`APK bereit: v${apkInfo.version} (${(fs.statSync(apkInfo.path).size / 1024 / 1024).toFixed(1)}MB)`);
});

// ── HTTP Route-Handler ──────────────────────────────────────────────

function handleHttpRequest(req, res) {
  try {
    const url = new URL(req.url, `http://${req.headers.host || "localhost"}`);
    const pathname = url.pathname;

    // OAuth-Callback: GET /oauth/callback/{service}?code=...&state=...&error=...
    // Pattern fuer Spotify, Google, Strava, GitHub, ... — alle OAuth2 Auth-Code-Flow.
    // Wir broadcasten an alle Raeume (App ist nicht im selben Raum wie Bridge,
    // aber Bridge schon — sie picks-up und forwardet ans Brain).
    const oauthMatch = pathname.match(/^\/oauth\/callback\/([a-zA-Z0-9_-]+)\/?$/);
    if (req.method === "GET" && oauthMatch) {
      const service = oauthMatch[1];
      const code = url.searchParams.get("code") || "";
      const state = url.searchParams.get("state") || "";
      const err = url.searchParams.get("error") || "";
      const errDesc = url.searchParams.get("error_description") || "";

      log(`OAuth-Callback: service=${service} code=${code.slice(0, 8)}... state=${state.slice(0, 8)}... err=${err}`);

      const payload = { service, code, state };
      if (err) {
        payload.error = err;
        if (errDesc) payload.errorDescription = errDesc;
      }

      // An alle Clients in allen Raeumen broadcasten — Bridge picks-up.
      const msg = JSON.stringify({
        type: "oauth_callback",
        payload,
        timestamp: Date.now(),
      });
      let receivers = 0;
      for (const [, room] of rooms) {
        for (const client of room.clients) {
          if (client.readyState === 1) {
            try { client.send(msg); receivers++; } catch (_) {}
          }
        }
      }
      log(`OAuth-Callback gebroadcastet an ${receivers} Client(s)`);

      // Browser-Antwort: schoene HTML-Seite (auch bei Error)
      const ok = !err;
      const title = ok ? "OAuth erfolgreich" : "OAuth fehlgeschlagen";
      const bodyColor = ok ? "#34C759" : "#FF3B30";
      const icon = ok ? "✅" : "❌";
      const subtitle = ok
        ? "Du kannst dieses Tab schliessen — ARIA hat den Zugang erhalten."
        : `Fehler: ${escapeHtml(err)} ${errDesc ? "— " + escapeHtml(errDesc) : ""}`;
      const html = `<!doctype html>
<html lang="de"><head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1">
<title>${title} — ${escapeHtml(service)}</title>
<style>
  html,body{margin:0;padding:0;height:100%;font-family:-apple-system,BlinkMacSystemFont,'Segoe UI',sans-serif;background:#0D0D1A;color:#E0E0F0;}
  body{display:flex;align-items:center;justify-content:center;}
  .card{background:#1E1E2E;border:1px solid #2A2A3E;border-radius:12px;padding:32px;max-width:420px;text-align:center;box-shadow:0 4px 24px rgba(0,0,0,0.4);}
  .icon{font-size:64px;line-height:1;margin-bottom:16px;}
  .title{font-size:20px;font-weight:600;color:${bodyColor};margin-bottom:8px;}
  .service{font-size:13px;color:#8888AA;margin-bottom:20px;text-transform:uppercase;letter-spacing:0.1em;}
  .sub{font-size:14px;color:#C0C0D0;line-height:1.5;}
  .hint{font-size:11px;color:#666680;margin-top:24px;}
</style></head><body>
<div class="card">
  <div class="icon">${icon}</div>
  <div class="title">${title}</div>
  <div class="service">${escapeHtml(service)}</div>
  <div class="sub">${subtitle}</div>
  <div class="hint">Du kannst zur ARIA-App zurueckkehren.</div>
</div>
<script>setTimeout(()=>{try{window.close();}catch(e){}}, 4000);</script>
</body></html>`;
      res.writeHead(ok ? 200 : 400, {
        "Content-Type": "text/html; charset=utf-8",
        "Cache-Control": "no-store",
      });
      res.end(html);
      return;
    }

    // Health-Endpoint
    if (req.method === "GET" && pathname === "/health") {
      res.writeHead(200, { "Content-Type": "application/json" });
      res.end(JSON.stringify({ ok: true, rooms: rooms.size }));
      return;
    }

    // Default: 404
    res.writeHead(404, { "Content-Type": "text/plain" });
    res.end("Not Found\n");
  } catch (e) {
    log(`HTTP handler error: ${e.message}`);
    try { res.writeHead(500).end("Internal Server Error"); } catch (_) {}
  }
}

function escapeHtml(s) {
  return String(s || "").replace(/[&<>"']/g, (c) =>
    ({ "&": "&amp;", "<": "&lt;", ">": "&gt;", '"': "&quot;", "'": "&#39;" }[c]));
}

wss.on("connection", (ws, req) => {
  // Token aus URL-Query lesen: ws://host:port/?token=abc123
  const url = new URL(req.url, `http://${req.headers.host}`);
  let token = url.searchParams.get("token");

  // Wenn kein Token in der URL, auf erste Nachricht warten
  if (!token) {
    ws.once("message", (raw) => {
      try {
        const msg = JSON.parse(raw);
        if (msg.token) {
          registerClient(ws, msg.token);
        } else {
          ws.close(4000, "Kein Token angegeben");
        }
      } catch {
        ws.close(4000, "Ungültige erste Nachricht — Token erwartet");
      }
    });
    return;
  }

  registerClient(ws, token);
});

function registerClient(ws, token) {
  // Maximale Anzahl aktiver Sessions prüfen
  if (!rooms.has(token) && rooms.size >= MAX_SESSIONS) {
    ws.close(4002, "Maximale Anzahl aktiver Sessions erreicht");
    log(`Abgelehnt: Session-Limit (${MAX_SESSIONS}) erreicht`);
    return;
  }

  // Raum erstellen oder betreten
  if (!rooms.has(token)) {
    rooms.set(token, { clients: new Set() });
    log(`Neuer Raum: ${token.slice(0, 8)}...`);
  }

  const room = rooms.get(token);
  room.clients.add(ws);
  ws._token = token;

  log(`Client verbunden: ${token.slice(0, 8)}... (${room.clients.size} im Raum)`);

  // Nachrichten an alle anderen Clients im selben Raum weiterleiten
  ws.on("message", (raw) => {
    let msg;
    try {
      msg = JSON.parse(raw);
    } catch {
      return; // Keine gültige JSON-Nachricht — ignorieren
    }

    // Nur erlaubte Nachrichtentypen durchlassen
    if (!msg.type || !ALLOWED_TYPES.has(msg.type)) {
      return;
    }

    // Update-Check: direkt an den anfragenden Client antworten (nicht relay'en)
    if (msg.type === "update_check") {
      const clientVersion = msg.payload?.version || "0.0.0.0";
      const apkInfo = getLatestAPK();
      if (apkInfo && compareVersions(apkInfo.version, clientVersion) > 0) {
        ws.send(JSON.stringify({
          type: "update_available",
          payload: {
            version: apkInfo.version,
            downloadUrl: `/update/latest.apk`,
            size: fs.statSync(apkInfo.path).size,
          },
          timestamp: Date.now(),
        }));
      }
      return;
    }

    // Raum-Diagnose: direkt an den anfragenden Client antworten (nicht relay'en).
    // Zeigt ALLE Raeume mit token8 (wie im Log), einem laengeren Fingerprint und
    // der Token-Laenge — so werden Prefix-Kollisionen (2 Raeume, gleicher 8-Zeichen-
    // Log, aber verschiedene volle Tokens) sofort sichtbar. KEINE vollen Tokens.
    if (msg.type === "rooms_query") {
      const crypto = require("crypto");
      const out = [];
      for (const [tok, room] of rooms) {
        let live = 0;
        for (const c of room.clients) if (c.readyState === 1) live++;
        out.push({
          token8: tok.slice(0, 8),
          fp: crypto.createHash("sha256").update(tok).digest("hex").slice(0, 12),
          len: tok.length,
          clients: live,
          you: tok === ws._token,
        });
      }
      out.sort((a, b) => b.clients - a.clients);
      ws.send(JSON.stringify({ type: "rooms_info", payload: { rooms: out }, timestamp: Date.now() }));
      return;
    }

    // Update-Download: APK als Base64 ueber WebSocket senden
    if (msg.type === "update_download") {
      const apkInfo = getLatestAPK();
      if (!apkInfo) {
        ws.send(JSON.stringify({ type: "update_data", payload: { error: "Keine APK verfuegbar" }, timestamp: Date.now() }));
        return;
      }
      try {
        const data = fs.readFileSync(apkInfo.path);
        const base64 = data.toString("base64");
        const sizeMB = (data.length / 1024 / 1024).toFixed(1);
        log(`APK sende: v${apkInfo.version} (${sizeMB}MB) an Client`);
        ws.send(JSON.stringify({
          type: "update_data",
          payload: {
            version: apkInfo.version,
            base64,
            size: data.length,
            fileName: `ARIA-v${apkInfo.version}.apk`,
          },
          timestamp: Date.now(),
        }));
      } catch (err) {
        ws.send(JSON.stringify({ type: "update_data", payload: { error: err.message }, timestamp: Date.now() }));
      }
      return;
    }

    // An alle anderen Clients im Raum weiterleiten
    for (const client of room.clients) {
      if (client !== ws && client.readyState === 1) {
        client.send(raw.toString());
      }
    }
  });

  ws.on("close", () => {
    room.clients.delete(ws);
    log(`Client getrennt: ${token.slice(0, 8)}... (${room.clients.size} verbleibend)`);
    if (room.clients.size === 0) {
      rooms.delete(token);
      log(`Raum geschlossen: ${token.slice(0, 8)}...`);
    }
  });

  ws.on("error", (err) => {
    log(`Fehler: ${err.message}`);
    room.clients.delete(ws);
  });
}

// ── Heartbeat — hält Verbindungen am Leben, räumt tote auf ──────────

const HEARTBEAT_INTERVAL = 15_000;

const heartbeat = setInterval(() => {
  for (const client of wss.clients) {
    if (client.isAlive === false) {
      log(`Toter Client entfernt (kein Pong)`);
      client.terminate();
      continue;
    }
    client.isAlive = false;
    client.ping();
  }
}, HEARTBEAT_INTERVAL);

wss.on("connection", (ws) => {
  ws.isAlive = true;
  ws.on("pong", () => { ws.isAlive = true; });
  // App-seitiger Heartbeat (JSON) zaehlt auch als lebendig
  const origOnMessage = ws._events?.message;
  ws.on("message", (raw) => {
    try {
      const msg = JSON.parse(raw);
      if (msg.type === "heartbeat") ws.isAlive = true;
    } catch {}
  });
});

// Aufräumen alle 30 Sekunden (statt 60)
const cleanup = setInterval(cleanupRooms, 30_000);

wss.on("close", () => {
  clearInterval(heartbeat);
  clearInterval(cleanup);
});

// ── Auto-Update: APK-Erkennung + Push ──────────────────────────────

let latestVersion = null;

function getLatestAPK() {
  try {
    if (!fs.existsSync(UPDATES_DIR)) return null;
    const files = fs.readdirSync(UPDATES_DIR)
      .filter(f => f.endsWith(".apk"))
      .map(f => {
        // ARIA-v0.0.2.3.apk oder ARIA-Cockpit-release.apk
        const match = f.match(/(\d+\.\d+\.\d+[\.\d]*)/);
        return { file: f, path: path.join(UPDATES_DIR, f), version: match ? match[1] : null };
      })
      .filter(f => f.version)
      .sort((a, b) => compareVersions(b.version, a.version)); // Neueste zuerst

    return files[0] || null;
  } catch {
    return null;
  }
}

function compareVersions(a, b) {
  const pa = a.split(".").map(Number);
  const pb = b.split(".").map(Number);
  for (let i = 0; i < Math.max(pa.length, pb.length); i++) {
    const diff = (pa[i] || 0) - (pb[i] || 0);
    if (diff !== 0) return diff;
  }
  return 0;
}

function notifyClientsAboutUpdate(apkInfo) {
  const msg = JSON.stringify({
    type: "update_available",
    payload: {
      version: apkInfo.version,
      downloadUrl: `/update/latest.apk`,
      size: fs.statSync(apkInfo.path).size,
    },
    timestamp: Date.now(),
  });

  // An alle Clients in allen Rooms senden
  for (const [, room] of rooms) {
    for (const client of room.clients) {
      if (client.readyState === 1) {
        client.send(msg);
      }
    }
  }
  log(`Update-Benachrichtigung gesendet: v${apkInfo.version} (${rooms.size} Raum/Raeume)`);
}

// Kein Polling — Update-Check passiert on-demand (update_check Message von App)

// ── Sauberes Herunterfahren ─────────────────────────────────────────

process.on("SIGTERM", () => {
  log("SIGTERM empfangen — fahre herunter");
  wss.close(() => process.exit(0));
});

process.on("SIGINT", () => {
  log("SIGINT empfangen — fahre herunter");
  wss.close(() => process.exit(0));
});