server.js

/**
 * Minimal load test server for Azure OpenAI Realtime (WebRTC) connections.
 * Responsibilities:
 *  - /start: ramp up many peer connections (configurable batches)
 *  - /stop: stop spawning + close all open connections
 *  - /events: SSE stream of metrics every second
 *  - /status: one-shot metrics
 * Uses createRealtimeConnection() for each individual handshake and Metrics for aggregation.
 */
import express from "express";
import dotenv from "dotenv";
import { createRealtimeConnection } from "./webrtcSession.js";
import { Metrics } from "./metrics.js";

dotenv.config();
const app = express();
app.use(express.static("public"));

const config = {
  resource: process.env.AZURE_OPENAI_RESOURCE,
  key: process.env.AZURE_OPENAI_KEY,
  deployment: process.env.AZURE_OPENAI_DEPLOYMENT,
  region: process.env.AZURE_OPENAI_REGION,
  apiVersion: process.env.API_VERSION,
  dcTimeoutMs: Number(process.env.DC_TIMEOUT_MS) || 8000,
  verbose: /^true$/i.test(process.env.VERBOSE_LOG || "")
};

const launchJitterMs = Number(process.env.LAUNCH_JITTER_MS) || 0;
// Mask first 10 characters of the key when logging
const maskedKey = config.key ? '*'.repeat(Math.min(10, config.key.length)) + config.key.slice(10) : undefined;
console.log("Loaded env:", { ...config, key: maskedKey });
if (!config.resource || !config.key) { console.error("Missing required env"); process.exit(1); }

const metrics = new Metrics();
let runController = null; // null => idle
let openConnections = [];

// POST /start  { target, rampStep, rampIntervalMs, batchDelayMs }
app.post("/start", express.json(), async (req, res) => {
  if (runController) return res.status(400).json({ error: "already running" });
  const { target = 50, rampStep = 10, rampIntervalMs = 5000, batchDelayMs = 200 } = req.body || {};
  const errs = [];
  if (!(rampStep >= 1)) errs.push("rampStep >= 1");
  if (!(rampIntervalMs >= 5000)) errs.push("rampIntervalMs >= 5000");
  if (!(batchDelayMs >= 200)) errs.push("batchDelayMs >= 200");
  if (errs.length) return res.status(400).json({ error: "invalid params", details: errs });

  metrics.reset();
  runController = { stop: false };
  res.json({ status: "started", target, rampStep });

  (async () => {
    for (let current = 0; current < target && !runController.stop; current += rampStep) {
      const toLaunch = Math.min(rampStep, target - current);
      await launchBatch(toLaunch, batchDelayMs, runController);
      await sleep(rampIntervalMs);
    }
  })();
});

// POST /stop
app.post("/stop", (req, res) => {
  if (!runController) return res.json({ status: "idle" });
  runController.stop = true;
  openConnections.forEach(c => c.close?.());
  openConnections = [];
  runController = null;
  res.json({ status: "stopping" });
});

// GET /events (SSE every second)
app.get("/events", (req, res) => {
  res.writeHead(200, { "Content-Type": "text/event-stream", "Cache-Control": "no-cache", "Connection": "keep-alive" });
  const interval = setInterval(() => { res.write(`data: ${JSON.stringify(metrics.snapshot())}\n\n`); }, 1000);
  req.on("close", () => clearInterval(interval));
});

// GET /status
app.get("/status", (req, res) => { res.json(metrics.snapshot()); });

function sleep(ms) { return new Promise(r => setTimeout(r, ms)); }

async function launchBatch(n, spacingMs, controller) {
  for (let i = 0; i < n && !controller.stop; i++) {
    const jitter = launchJitterMs ? Math.random() * launchJitterMs : 0;
    if (jitter) await sleep(jitter);
    spawnOne(controller);
    if (spacingMs) await sleep(spacingMs);
  }
}

async function spawnOne(controller) {
  metrics.recordStart();
  let out;
  try {
    out = await createRealtimeConnection({ ...config });
  } catch (e) {
    metrics.recordFailure(e?.message || 'UNCAUGHT');
    return;
  }
  if (controller.stop) { out?.close?.(); return; }
  if (out && out.ok) {
    metrics.recordSuccess(out.handshakeMs);
    openConnections.push(out);
  } else {
    metrics.recordFailure(out?.error || 'UNKNOWN');
  }
}

const port = process.env.PORT || 3000;
app.listen(port, () => console.log("LISTEN " + port));