ARIA-AGENT/rvs/server.js

"use strict";

const { WebSocketServer } = require("ws");

// ── Konfiguration aus Umgebungsvariablen ────────────────────────────
const PORT = parseInt(process.env.PORT || "3000", 10);
const MAX_SESSIONS = parseInt(process.env.MAX_SESSIONS || "10", 10);

// 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",
]);

// 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)}...`);
    }
  }
}

// ── WebSocket-Server starten ────────────────────────────────────────

const wss = new WebSocketServer({ port: PORT });

wss.on("listening", () => {
  log(`RVS läuft auf Port ${PORT} | Max Sessions: ${MAX_SESSIONS}`);
});

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

    // 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);
});

// ── 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));
});