glo/helper/server.js

import crypto from "crypto";
import express from "express";
import http from "http";
import { WebSocketServer } from "ws";

const PORT = Number(process.env.PORT || 8080);
const HOST = process.env.HOST || "0.0.0.0";
const CALL_HELPER_SHARED_SECRET = String(
  process.env.CALL_HELPER_SHARED_SECRET || ""
).trim();
const PUBLIC_BASE = String(
  process.env.PUBLIC_BASE || `http://127.0.0.1:${PORT}`
).replace(/\/+$/, "");

const HELPER_LOG_LEVEL = String(process.env.HELPER_LOG_LEVEL || "info").toLowerCase();
const HELPER_DEFAULT_ROUTE_MODE = String(
  process.env.HELPER_DEFAULT_ROUTE_MODE || "direct"
).toLowerCase();

const ENABLE_OVERLAY_METADATA = String(
  process.env.ENABLE_OVERLAY_METADATA || "1"
) === "1";

const OVERLAY_YGG_ENABLED = String(
  process.env.OVERLAY_YGG_ENABLED || "0"
) === "1";
const OVERLAY_YGG_BRIDGE_WSS = String(
  process.env.OVERLAY_YGG_BRIDGE_WSS || ""
).trim();
const OVERLAY_YGG_BRIDGE_LABEL = String(
  process.env.OVERLAY_YGG_BRIDGE_LABEL || "ygg-bridge"
).trim();

const OVERLAY_I2P_ENABLED = String(
  process.env.OVERLAY_I2P_ENABLED || "0"
) === "1";
const OVERLAY_I2P_BRIDGE_HTTPS = String(
  process.env.OVERLAY_I2P_BRIDGE_HTTPS || ""
).trim();
const OVERLAY_I2P_BRIDGE_LABEL = String(
  process.env.OVERLAY_I2P_BRIDGE_LABEL || "i2p-bridge"
).trim();

const RELAY_ENABLED = String(process.env.RELAY_ENABLED || "1") === "1";
const RELAY_WSS = String(process.env.RELAY_WSS || "").trim();

if (!CALL_HELPER_SHARED_SECRET) {
  console.error("missing CALL_HELPER_SHARED_SECRET");
  process.exit(1);
}

const app = express();
app.use(express.json({ limit: "64kb" }));

const server = http.createServer(app);
const wss = new WebSocketServer({ noServer: true });

/**
 * in-memory state
 *
 * calls:
 *   callId -> {
 *     callId,
 *     room,
 *     routeMode,
 *     participants: Map(peerId -> participant),
 *     createdAt,
 *     updatedAt
 *   }
 *
 * tickets:
 *   ticket -> {
 *     ticket,
 *     callId,
 *     peerId,
 *     sub,
 *     username,
 *     role,
 *     room,
 *     exp,
 *     routeProfile
 *   }
 */
const calls = new Map();
const tickets = new Map();
const recentLogs = [];
const MAX_LOGS = 500;

function nowMs() {
  return Date.now();
}

function nowSeconds() {
  return Math.floor(Date.now() / 1000);
}

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

function log(level, message, meta = {}) {
  const allowed = ["debug", "info", "warn", "error"];
  const currentIdx = allowed.indexOf(HELPER_LOG_LEVEL);
  const entryIdx = allowed.indexOf(level);
  const item = {
    ts: isoNow(),
    level,
    message,
    meta
  };

  recentLogs.push(item);
  if (recentLogs.length > MAX_LOGS) {
    recentLogs.shift();
  }

  if (entryIdx >= currentIdx) {
    const method =
      level === "error" ? "error" :
      level === "warn" ? "warn" :
      "log";
    console[method](`[${item.ts}] [${level}] ${message}`, meta);
  }
}

function b64urlDecode(value) {
  const pad = value.length % 4 === 0 ? "" : "=".repeat(4 - (value.length % 4));
  const normalized = value.replace(/-/g, "+").replace(/_/g, "/") + pad;
  return Buffer.from(normalized, "base64");
}

function verifySignedToken(token) {
  if (!token || typeof token !== "string" || !token.includes(".")) {
    throw new Error("invalid token format");
  }

  const [payloadPart, sigPart] = token.split(".", 2);
  const payloadBytes = b64urlDecode(payloadPart);
  const sigBytes = b64urlDecode(sigPart);

  const expected = crypto
    .createHmac("sha256", CALL_HELPER_SHARED_SECRET)
    .update(payloadBytes)
    .digest();

  if (
    sigBytes.length !== expected.length ||
    !crypto.timingSafeEqual(sigBytes, expected)
  ) {
    throw new Error("invalid token signature");
  }

  let payload;
  try {
    payload = JSON.parse(payloadBytes.toString("utf8"));
  } catch {
    throw new Error("invalid token payload json");
  }

  if (!payload || typeof payload !== "object") {
    throw new Error("invalid token payload");
  }

  if (!payload.exp || Number(payload.exp) <= nowSeconds()) {
    throw new Error("token expired");
  }

  if (!payload.can_call) {
    throw new Error("token not authorized for calls");
  }

  if (!payload.call_id || !payload.sub) {
    throw new Error("token missing required fields");
  }

  return payload;
}

function randomId(prefix = "") {
  return `${prefix}${crypto.randomBytes(8).toString("hex")}`;
}

function chooseRouteMode(payload) {
  const requested = String(payload.route_mode || "").trim().toLowerCase();

  if (requested === "overlay_ygg" && OVERLAY_YGG_ENABLED) {
    return "overlay_ygg";
  }

  if (requested === "overlay_i2p" && OVERLAY_I2P_ENABLED) {
    return "overlay_i2p";
  }

  if (HELPER_DEFAULT_ROUTE_MODE === "overlay_ygg" && OVERLAY_YGG_ENABLED) {
    return "overlay_ygg";
  }

  if (HELPER_DEFAULT_ROUTE_MODE === "overlay_i2p" && OVERLAY_I2P_ENABLED) {
    return "overlay_i2p";
  }

  return "direct";
}

function buildRouteCandidates(callId) {
  const routes = [];

  routes.push({
    kind: "direct",
    target: callId,
    priority: 100,
    available: true,
    label: "direct"
  });

  if (ENABLE_OVERLAY_METADATA && OVERLAY_YGG_ENABLED) {
    routes.push({
      kind: "overlay_ygg",
      target: OVERLAY_YGG_BRIDGE_WSS || PUBLIC_BASE.replace(/^http/, "ws"),
      priority: 90,
      available: Boolean(OVERLAY_YGG_BRIDGE_WSS),
      label: OVERLAY_YGG_BRIDGE_LABEL
    });
  }

  if (ENABLE_OVERLAY_METADATA && OVERLAY_I2P_ENABLED) {
    routes.push({
      kind: "overlay_i2p",
      target: OVERLAY_I2P_BRIDGE_HTTPS || PUBLIC_BASE,
      priority: 40,
      available: Boolean(OVERLAY_I2P_BRIDGE_HTTPS),
      label: OVERLAY_I2P_BRIDGE_LABEL
    });
  }

  if (RELAY_ENABLED) {
    routes.push({
      kind: "relay",
      target: RELAY_WSS || PUBLIC_BASE.replace(/^http/, "ws"),
      priority: 20,
      available: true,
      label: "relay"
    });
  }

  return routes;
}

function buildTransportPolicy(routeMode) {
  const order = [];
  if (routeMode === "overlay_ygg") {
    order.push("overlay_ygg", "direct", "relay");
  } else if (routeMode === "overlay_i2p") {
    order.push("direct", "relay");
  } else {
    order.push("direct", "relay");
  }

  return {
    preferred_order: order,
    allow_relay_fallback: RELAY_ENABLED,
    private_service_transport: OVERLAY_I2P_ENABLED ? "overlay_i2p" : null
  };
}

function getOrCreateCall(callId, room, routeMode = "direct") {
  let call = calls.get(callId);
  if (!call) {
    call = {
      callId,
      room,
      routeMode,
      participants: new Map(),
      createdAt: nowMs(),
      updatedAt: nowMs()
    };
    calls.set(callId, call);
    log("info", "created call", { callId, room, routeMode });
  }
  return call;
}

function buildWsUrl(ticket) {
  const publicBase = PUBLIC_BASE.replace(/^http:/, "ws:").replace(/^https:/, "wss:");
  return `${publicBase}/ws?ticket=${encodeURIComponent(ticket)}`;
}

function sendJson(ws, payload) {
  if (!ws || ws.readyState !== ws.OPEN) return;
  ws.send(JSON.stringify(payload));
}

function broadcast(call, payload, exceptPeerId = null) {
  for (const [peerId, participant] of call.participants.entries()) {
    if (exceptPeerId && peerId === exceptPeerId) continue;
    sendJson(participant.ws, payload);
  }
}

function callSnapshot(call) {
  return {
    call_id: call.callId,
    room: call.room,
    route_mode: call.routeMode,
    participant_count: call.participants.size,
    created_at: new Date(call.createdAt).toISOString(),
    updated_at: new Date(call.updatedAt).toISOString(),
    participants: [...call.participants.values()].map(participant => ({
      peer_id: participant.peerId,
      sub: participant.sub,
      username: participant.username,
      role: participant.role,
      route_mode: participant.routeMode,
      joined_at: new Date(participant.joinedAt).toISOString()
    }))
  };
}

function statusSnapshot() {
  return {
    ok: true,
    ts: isoNow(),
    public_base: PUBLIC_BASE,
    route_defaults: {
      default_route_mode: HELPER_DEFAULT_ROUTE_MODE,
      overlay_metadata_enabled: ENABLE_OVERLAY_METADATA,
      relay_enabled: RELAY_ENABLED
    },
    overlays: {
      ygg: {
        enabled: OVERLAY_YGG_ENABLED,
        bridge_label: OVERLAY_YGG_BRIDGE_LABEL,
        bridge_wss: OVERLAY_YGG_BRIDGE_WSS || null
      },
      i2p: {
        enabled: OVERLAY_I2P_ENABLED,
        bridge_label: OVERLAY_I2P_BRIDGE_LABEL,
        bridge_https: OVERLAY_I2P_BRIDGE_HTTPS || null
      }
    },
    counts: {
      active_calls: calls.size,
      active_tickets: tickets.size
    }
  };
}

function removeParticipant(callId, peerId, notify = true) {
  const call = calls.get(callId);
  if (!call) return;

  const participant = call.participants.get(peerId);
  if (!participant) return;

  call.participants.delete(peerId);
  call.updatedAt = nowMs();

  log("info", "participant removed", {
    callId,
    peerId,
    username: participant.username
  });

  if (notify) {
    broadcast(call, {
      type: "participant-left",
      participantId: peerId,
      roomId: callId
    });
  }

  if (call.participants.size === 0) {
    calls.delete(callId);
    log("info", "call removed because empty", { callId });
  }
}

app.get("/healthz", (_req, res) => {
  res.json(statusSnapshot());
});

app.get("/status", (_req, res) => {
  res.json({
    ...statusSnapshot(),
    calls: [...calls.values()].map(callSnapshot)
  });
});

app.get("/logs", (req, res) => {
  const limit = Math.max(1, Math.min(200, Number(req.query.limit || 50)));
  res.json({
    ok: true,
    logs: recentLogs.slice(-limit)
  });
});

app.post("/api/call/session", (req, res) => {
  try {
    const token = req.body?.token;
    const payload = verifySignedToken(token);

    const callId = String(payload.call_id);
    const room = String(payload.room || callId);
    const peerId = randomId("peer_");
    const ticket = randomId("ticket_");
    const routeMode = chooseRouteMode(payload);
    const routeCandidates = buildRouteCandidates(callId);
    const transportPolicy = buildTransportPolicy(routeMode);

    getOrCreateCall(callId, room, routeMode);

    tickets.set(ticket, {
      ticket,
      callId,
      peerId,
      sub: String(payload.sub),
      username: String(payload.username || peerId),
      role: String(payload.role || "user"),
      room,
      exp: Number(payload.exp),
      routeProfile: {
        routeMode,
        routeCandidates,
        transportPolicy
      }
    });

    log("info", "created call session ticket", {
      callId,
      peerId,
      username: payload.username,
      routeMode
    });

    return res.json({
      ok: true,
      call_id: callId,
      peer_id: peerId,
      ws_url: buildWsUrl(ticket),
      iceServers: [
        {
          urls: ["stun:stun.l.google.com:19302"]
        }
      ],
      transport_policy: transportPolicy,
      route_candidates: routeCandidates
    });
  } catch (error) {
    log("warn", "call session rejected", {
      error: error.message
    });

    return res.status(401).json({
      ok: false,
      error: error.message || "invalid call token"
    });
  }
});

server.on("upgrade", (req, socket, head) => {
  const url = new URL(req.url, `http://${req.headers.host}`);
  if (url.pathname !== "/ws") {
    socket.destroy();
    return;
  }

  const ticket = url.searchParams.get("ticket");
  const ticketData = ticket ? tickets.get(ticket) : null;

  if (!ticketData || ticketData.exp <= nowSeconds()) {
    socket.destroy();
    return;
  }

  wss.handleUpgrade(req, socket, head, ws => {
    ws._ticketData = ticketData;
    wss.emit("connection", ws, req);
  });
});

wss.on("connection", ws => {
  const ticketData = ws._ticketData;
  if (!ticketData) {
    ws.close();
    return;
  }

  tickets.delete(ticketData.ticket);

  const call = getOrCreateCall(
    ticketData.callId,
    ticketData.room,
    ticketData.routeProfile?.routeMode || "direct"
  );

  const participant = {
    peerId: ticketData.peerId,
    sub: ticketData.sub,
    username: ticketData.username,
    role: ticketData.role,
    room: ticketData.room,
    routeMode: ticketData.routeProfile?.routeMode || "direct",
    routeCandidates: ticketData.routeProfile?.routeCandidates || [],
    transportPolicy: ticketData.routeProfile?.transportPolicy || {},
    ws,
    joinedAt: nowMs()
  };

  call.participants.set(ticketData.peerId, participant);
  call.updatedAt = nowMs();

  log("info", "participant joined websocket", {
    callId: call.callId,
    peerId: participant.peerId,
    username: participant.username,
    routeMode: participant.routeMode
  });

  sendJson(ws, {
    type: "joined",
    participantId: ticketData.peerId,
    roomId: call.callId,
    transportPolicy: participant.transportPolicy,
    routeCandidates: participant.routeCandidates
  });

  for (const [peerId, existing] of call.participants.entries()) {
    if (peerId === ticketData.peerId) continue;
    sendJson(ws, {
      type: "participant-joined",
      participantId: peerId,
      roomId: call.callId,
      username: existing.username,
      routeMode: existing.routeMode
    });
  }

  broadcast(
    call,
    {
      type: "participant-joined",
      participantId: ticketData.peerId,
      roomId: call.callId,
      username: participant.username,
      routeMode: participant.routeMode
    },
    ticketData.peerId
  );

  ws.on("message", raw => {
    let msg;
    try {
      msg = JSON.parse(String(raw));
    } catch {
      log("warn", "invalid ws json from peer", {
        callId: call.callId,
        peerId: participant.peerId
      });
      return;
    }

    const senderPeerId = ticketData.peerId;
    call.updatedAt = nowMs();

    if (!msg || typeof msg !== "object" || !msg.type) {
      return;
    }

    if (msg.type === "hangup") {
      log("info", "peer requested hangup", {
        callId: call.callId,
        peerId: senderPeerId
      });
      removeParticipant(call.callId, senderPeerId, true);
      try {
        ws.close();
      } catch {}
      return;
    }

    if (msg.type === "offer" || msg.type === "answer") {
      const to = String(msg.to || "").trim();
      if (!to) return;
      const target = call.participants.get(to);
      if (!target) {
        log("warn", "signal target missing", {
          callId: call.callId,
          from: senderPeerId,
          to,
          type: msg.type
        });
        return;
      }

      sendJson(target.ws, {
        type: msg.type,
        from: senderPeerId,
        sdp: msg.sdp,
        username: participant.username,
        roomId: call.callId,
        routeMode: participant.routeMode
      });
      return;
    }

    if (msg.type === "ice") {
      const to = String(msg.to || "").trim();
      if (!to) return;
      const target = call.participants.get(to);
      if (!target) {
        log("warn", "ice target missing", {
          callId: call.callId,
          from: senderPeerId,
          to
        });
        return;
      }

      sendJson(target.ws, {
        type: "ice",
        from: senderPeerId,
        candidate: msg.candidate,
        sdpMid: msg.sdpMid,
        sdpMLineIndex: msg.sdpMLineIndex,
        username: participant.username,
        roomId: call.callId,
        routeMode: participant.routeMode
      });
      return;
    }
  });

  ws.on("close", () => {
    log("info", "websocket closed", {
      callId: call.callId,
      peerId: participant.peerId
    });
    removeParticipant(call.callId, ticketData.peerId, true);
  });

  ws.on("error", error => {
    log("warn", "websocket error", {
      callId: call.callId,
      peerId: participant.peerId,
      error: error?.message || "unknown"
    });
    removeParticipant(call.callId, ticketData.peerId, true);
  });
});

server.listen(PORT, HOST, () => {
  log("info", "calls helper listening", {
    host: HOST,
    port: PORT,
    publicBase: PUBLIC_BASE,
    overlays: {
      ygg: OVERLAY_YGG_ENABLED,
      i2p: OVERLAY_I2P_ENABLED
    }
  });
});