aether-framework/resources/[aether]/aether-admin/media-server/server.js

/* =====================================================================
   aether-admin — Media-Server (WebRTC-SFU + Signaling)

   Zwei Ohren:
     • HTTP (GRANT_PORT)  — nur für den FiveM-Server. Der stellt hier
       kurzlebige Zugangstoken aus (/grant) und räumt Räume ab (/revoke),
       abgesichert über das gemeinsame Geheimnis ADMIN_SECRET.
     • WebSocket (WS_PORT) — für die NUIs. Publisher senden ihren Stream,
       Viewer empfangen ihn. Der Server routet die Medien (SFU): er nimmt
       die RTP-Pakete eines Publishers und schreibt sie in Ausgangs-Tracks,
       die an alle Viewer desselben Raums gehen.

   Signaling-Protokoll (JSON über WS), passend zu html/js/webrtc.js:
     Client → Server:  { t:"join", token }
                       { t:"offer",  sdp }   (Publisher ist Offerer)
                       { t:"answer", sdp }   (Viewer antwortet auf Server-Offer)
                       { t:"ice",    candidate }
     Server → Client:  { t:"ready" }         (Publisher darf sein Offer senden)
                       { t:"offer",  sdp }   (Server offeriert dem Viewer)
                       { t:"answer", sdp }   (Antwort an den Publisher)
                       { t:"ice",    candidate }

   HINWEIS: WebRTC-SFUs lassen sich nur im laufenden Betrieb sinnvoll
   prüfen. Dieser Server ist die Referenzimplementierung — je nach
   werift-Version können Kleinigkeiten anzupassen sein (siehe README).
===================================================================== */
"use strict";

const http = require("http");
const { WebSocketServer } = require("ws");
const { RTCPeerConnection, MediaStreamTrack } = require("werift");

const WS_PORT     = parseInt(process.env.AETHER_MEDIA_WS_PORT   || "8787", 10);
const GRANT_PORT  = parseInt(process.env.AETHER_MEDIA_HTTP_PORT || "8788", 10);
const ADMIN_SECRET = process.env.AETHER_MEDIA_SECRET || "";
const STUN = [{ urls: "stun:stun.l.google.com:19302" }];

if (!ADMIN_SECRET) {
  console.warn("[media] WARNUNG: AETHER_MEDIA_SECRET ist leer — /grant lehnt alles ab.");
}

/* ---- Zustand ------------------------------------------------------- */
// token -> { raum, rolle, exp }
const grants = new Map();
// raum -> { publishers:Set, viewers:Set, tracks:MediaStreamTrack[] }
const raeume = new Map();

function raumHolen(id) {
  if (!raeume.has(id)) raeume.set(id, { publishers: new Set(), viewers: new Set(), tracks: [] });
  return raeume.get(id);
}

// Abgelaufene Token regelmäßig entfernen
setInterval(() => {
  const jetzt = Date.now();
  for (const [t, g] of grants) if (g.exp < jetzt) grants.delete(t);
}, 10000);

/* =====================================================================
   HTTP: Token-Ausgabe (nur vom FiveM-Server)
===================================================================== */
const httpServer = http.createServer((req, res) => {
  if (req.method !== "POST") { res.writeHead(405).end(); return; }
  if (req.headers["x-admin-secret"] !== ADMIN_SECRET || !ADMIN_SECRET) {
    res.writeHead(403).end('{"ok":false}'); return;
  }

  let body = "";
  req.on("data", (c) => { body += c; if (body.length > 1e5) req.destroy(); });
  req.on("end", () => {
    let daten = {};
    try { daten = JSON.parse(body || "{}"); } catch (e) {}

    if (req.url === "/grant") {
      const gueltig = Math.max(5, Math.min(300, +daten.gueltigSek || 30));
      grants.set(daten.token, {
        raum: String(daten.raum || ""),
        rolle: daten.rolle === "publisher" ? "publisher" : "viewer",
        exp: Date.now() + gueltig * 1000,
      });
      res.writeHead(200, { "Content-Type": "application/json" }).end('{"ok":true}');
    } else if (req.url === "/revoke") {
      raumAufloesen(String(daten.raum || ""));
      res.writeHead(200, { "Content-Type": "application/json" }).end('{"ok":true}');
    } else {
      res.writeHead(404).end('{"ok":false}');
    }
  });
});
httpServer.listen(GRANT_PORT, () => console.log(`[media] HTTP /grant auf :${GRANT_PORT}`));

/* =====================================================================
   WebSocket: Signaling + Medien
===================================================================== */
const wss = new WebSocketServer({ port: WS_PORT });
wss.on("listening", () => console.log(`[media] WebSocket auf :${WS_PORT}`));

wss.on("connection", (ws) => {
  let sitzung = null;   // wird nach gültigem join gesetzt

  ws.on("message", async (roh) => {
    let m; try { m = JSON.parse(roh.toString()); } catch (e) { return; }

    // Erster Schritt: Beitritt mit gültigem Token
    if (!sitzung) {
      if (m.t !== "join") return;
      const g = grants.get(m.token);
      if (!g || g.exp < Date.now()) { ws.close(4001, "ungueltiges Token"); return; }
      grants.delete(m.token);   // Einmal-Token

      const raum = raumHolen(g.raum);
      const pc = new RTCPeerConnection({ iceServers: STUN });
      sitzung = { rolle: g.rolle, raum: g.raum, pc, ws };

      pc.onIceCandidate.subscribe((candidate) => {
        sende(ws, { t: "ice", candidate: alsJson(candidate) });
      });

      if (g.rolle === "publisher") {
        raum.publishers.add(sitzung);
        // Publisher ist Offerer → wir bitten ihn, sein Offer zu schicken
        pc.onTrack.subscribe((track) => publisherTrack(g.raum, track));
        sende(ws, { t: "ready" });
      } else {
        raum.viewers.add(sitzung);
        // Viewer bekommt alle bereits laufenden Tracks und ein Offer
        for (const track of raum.tracks) pc.addTransceiver(track, { direction: "sendonly" });
        const offer = await pc.createOffer();
        await pc.setLocalDescription(offer);
        sende(ws, { t: "offer", sdp: pc.localDescription });
      }
      return;
    }

    // Laufende Verhandlung
    const pc = sitzung.pc;
    if (m.t === "offer") {                    // von einem Publisher
      await pc.setRemoteDescription(m.sdp);
      const answer = await pc.createAnswer();
      await pc.setLocalDescription(answer);
      sende(ws, { t: "answer", sdp: pc.localDescription });
    } else if (m.t === "answer") {            // von einem Viewer
      await pc.setRemoteDescription(m.sdp);
    } else if (m.t === "ice" && m.candidate) {
      try { await pc.addIceCandidate(m.candidate); } catch (e) {}
    }
  });

  ws.on("close", () => {
    if (!sitzung) return;
    const raum = raeume.get(sitzung.raum);
    try { sitzung.pc.close(); } catch (e) {}
    if (raum) { raum.publishers.delete(sitzung); raum.viewers.delete(sitzung); }
  });
});

/* ---- Ein Publisher-Track kam an → an alle Viewer weiterreichen ----- */
async function publisherTrack(raumId, track) {
  const raum = raeume.get(raumId);
  if (!raum) return;

  // Ausgangs-Track, in den wir die RTP-Pakete des Publishers schreiben
  const fwd = new MediaStreamTrack({ kind: track.kind });
  track.onReceiveRtp.subscribe((rtp) => { try { fwd.writeRtp(rtp); } catch (e) {} });
  raum.tracks.push(fwd);

  // Jedem laufenden Viewer den neuen Track zusätzlich anbieten (Renegotiation)
  for (const viewer of raum.viewers) {
    try {
      viewer.pc.addTransceiver(fwd, { direction: "sendonly" });
      const offer = await viewer.pc.createOffer();
      await viewer.pc.setLocalDescription(offer);
      sende(viewer.ws, { t: "offer", sdp: viewer.pc.localDescription });
    } catch (e) { /* Viewer könnte gerade weg sein */ }
  }
}

function raumAufloesen(raumId) {
  const raum = raeume.get(raumId);
  if (!raum) return;
  for (const s of raum.publishers) try { s.pc.close(); } catch (e) {}
  for (const s of raum.viewers) try { s.pc.close(); } catch (e) {}
  raeume.delete(raumId);
}

/* ---- Kleine Helfer ------------------------------------------------- */
function sende(ws, obj) {
  if (ws && ws.readyState === 1) ws.send(JSON.stringify(obj));
}
function alsJson(candidate) {
  if (!candidate) return null;
  if (typeof candidate.toJSON === "function") return candidate.toJSON();
  return {
    candidate: candidate.candidate,
    sdpMid: candidate.sdpMid,
    sdpMLineIndex: candidate.sdpMLineIndex,
  };
}