/**
 * AI Software Factory — k6: realtime WebSocket scale (10,000 connections).
 *
 * Scenario (PRODUCT-SPEC §Performance, realtime < 200 ms):
 *   - 10,000 concurrent WebSocket connections (one per VU)
 *   - every VU joins the shared room "load" (configurable via K6_ROOM)
 *   - every 500 ms each VU sends a Yjs update (small base64 doc delta)
 *   - the realtime server acks each update (echoing the client seq); the
 *     round-trip time is the measured metric
 *
 * Pass criteria (thresholds below, fail the run when violated):
 *   rtt_ms          p(95) < 200 ms
 *   ws_disconnects  rate  < 0.5% during the hold phase
 *   join_failures   count < 1
 *
 * Protocol assumption (Phase 7 realtime contract — keep in sync with
 * apps/realtime):
 *   connect:  <K6_REALTIME_URL>/?token=<jwt>
 *   join:     {"type":"join","roomId":"load"}            → {"type":"join:ack","roomId":"load"}
 *   update:   {"type":"yjs-update","roomId":"load","docId":"<pageId>","seq":N,"ts":<ms>}
 *   ack:      {"type":"yjs-update:ack","roomId":"load","docId":"<pageId>","seq":N,"ts":<ms>}
 */
import http from "k6/http";
import { check } from "k6";
import WS from "k6/ws";
import { Trend, Rate } from "k6/metrics";

const RTT_MS = new Trend("rtt_ms", true);
const disconnects = new Rate("ws_disconnects");
const joinFailures = new Rate("join_failures");

const API_URL = __ENV.K6_API_URL || "http://localhost:4000";
const WS_URL = __ENV.K6_REALTIME_URL || "ws://localhost:4001";
const ROOM = __ENV.K6_ROOM || "load";
const EMAIL = __ENV.K6_EMAIL || "demo@factory.local";
const PASSWORD = __ENV.K6_PASSWORD || "factory-demo-123";

const HOLD_SECONDS = parseInt(__ENV.K6_DURATION || "300", 10); // default 5 min

export const options = {
  scenarios: {
    realtime_scale: {
      executor: "constant-vusers",
      vus: parseInt(__ENV.K6_VUS || "10000", 10),
      duration: `${HOLD_SECONDS}s`,
    },
  },
  thresholds: {
    rtt_ms: ["p(95)<200", "p(99)<400"],
    ws_disconnects: ["rate<0.005"],
    join_failures: ["count<1"],
  },
};

export function setup() {
  // Authenticate (seeded demo user; register as fallback) and mint a token
  // shared by all VUs (the token is a constant load attribute, not a
  // per-connection identity — the server's presence count is what matters).
  let login = http.post(`${API_URL}/api/v1/auth/login`, JSON.stringify({ email: EMAIL, password: PASSWORD }), {
    headers: { "Content-Type": "application/json" },
  });
  if (login.status !== 200) {
    const reg = http.post(`${API_URL}/api/v1/auth/register`, JSON.stringify({ email: EMAIL, password: PASSWORD, name: "K6 Realtime" }), {
      headers: { "Content-Type": "application/json" },
    });
    if (reg.status === 201 || reg.status === 200) {
      login = http.post(`${API_URL}/api/v1/auth/login`, JSON.stringify({ email: EMAIL, password: PASSWORD }), {
        headers: { "Content-Type": "application/json" },
      });
    }
  }
  if (login.status !== 200) throw new Error(`setup: login failed (${login.status})`);
  const body = login.json();
  const token = body.accessToken || body.access_token || body.token;
  if (!token) throw new Error("setup: no access token in login response");
  return { token, docId: `k6-realtime-doc-${Date.now()}` };
}

export default function (data) {
  const url = `${WS_URL}/?token=${encodeURIComponent(data.token)}`;
  const deadline = Date.now() + HOLD_SECONDS * 1000;
  let seq = 0;

  // In-flight update → sent-at map (for RTT measurement on ack).
  const pending = new Map();
  let joined = false;

  const ws = new WS(url, {
    // The shared token authenticates the socket; no extra headers needed.
    compression: false,
  });

  ws.on("open", () => {
    ws.send(JSON.stringify({ type: "join", roomId: ROOM }));
  });

  ws.on("message", (raw) => {
    let msg;
    try {
      msg = JSON.parse(raw.toString());
    } catch {
      return; // non-JSON frame (e.g. binary Yjs sync) — ignore for RTT.
    }
    if (msg.type === "join:ack" || msg.type === "welcome") {
      joined = true;
      return;
    }
    if (msg.type === "yjs-update:ack" && typeof msg.seq === "number" && pending.has(msg.seq)) {
      const sentAt = pending.get(msg.seq);
      pending.delete(msg.seq);
      RTT_MS.add(Date.now() - sentAt);
    }
  });

  ws.on("close", (e) => {
    // A close BEFORE the deadline is a mid-test disconnect (a clean close
    // at the deadline is expected and must not count).
    if (Date.now() < deadline - 1000) disconnects.add(1);
  });

  // Block the VU (yielding the event loop) until the hold phase ends.
  while (Date.now() < deadline) {
    // Wait ~500 ms between updates (ws.wait yields to the k6 runtime and
    // processes queued socket events).
    ws.wait(500);
    if (ws.closed) break;
    if (!joined) {
      // Give the join a bounded number of attempts; after that, count it as
      // a join failure (the threshold fails the run).
      ws.send(JSON.stringify({ type: "join", roomId: ROOM }));
      continue;
    }
    seq += 1;
    pending.set(seq, Date.now());
    ws.send(
      JSON.stringify({
        type: "yjs-update",
        roomId: ROOM,
        docId: data.docId,
        seq,
        ts: Date.now(),
        update: "AA==", // tiny Yjs delta (server acks; content is irrelevant)
      }),
    );
  }

  // Join must have succeeded by the end of the hold phase.
  if (!joined) joinFailures.add(1);

  // Drain: let in-flight acks land, then close cleanly.
  ws.wait(1500);
  ws.close();
}