/**
 * AI Software Factory — k6: realtime 1k load baseline (TASK-0091 acceptance
 * criterion "Load", first-run; TASK-0095 re-runs it as the final contract).
 *
 * Scenario (the task spec, verbatim):
 *   - 1,000 connections over 30 s (100/s connect rate), held 2 min
 *   - 200 msg/s mixed across the fleet: 50 % small doc updates,
 *     30 % awareness, 20 % ping (per VU: 0.2 msg/s — one frame every 5 s)
 *   - one node, p95 client→server→client round-trip < 200 ms,
 *     0 protocol errors, 0 join failures
 *
 * RTT measurement (documented per the task spec):
 *   The client sends the app-level `ping {ts}` frame; the server echoes
 *   `ping-pong {ts}` with the SAME timestamp (server.ts `case "ping"`) —
 *   so the round-trip is computed entirely on the client clock:
 *   rtt = now - sentTs, immune to clock skew. The `hello.serverTime`
 *   captured once at connect is recorded as the `clock_skew_ms` diagnostic
 *   (server - client) but does NOT enter the RTT math.
 *
 * Protocol (the real TASK-0091 contract — packages/shared frames.ts):
 *   connect:  <K6_REALTIME_URL>/realtime?token=<jwt>&room=page:<pageId>
 *             header: X-Forwarded-For: 10.100.<a>.<b>   (per-VU identity)
 *   hello:    {type:"hello", connId, canEdit, serverTime, roomPeers}
 *   update:   binary 0x01 + YjsUpdate   (valid frames — generated with yjs;
 *             an invalid update is a `bad_update` strike, 10 → 4008)
 *   awareness: binary 0x02 + JSON state (the server relays, no content check)
 *   ping:     {type:"ping", ts}                → {type:"ping-pong", ts}
 *   ack:      {type:"ack", seq}                (logged updates)
 *
 * Why 1,000 distinct identities + 1,000 distinct source IPs (both MANDATORY):
 *   - the server counts live sockets per user (100/user cap) and per-user
 *     control sockets (10/user cap, the multi-tab guard) — ONE shared token
 *     can never reach 1,000 connections (the 101st is rejected 429);
 *   - the connect limiter is 30/min per source IP — the VUs present
 *     per-socket `X-Forwarded-For` identities, so the server must run with
 *     REALTIME_TRUST_PROXY=1 (the production reality: nginx terminates the
 *     WS, the pod sees the proxy IP).
 *   The seeder (helpers/seed-1k-load.mjs) creates the 1,000 users +
 *   memberships + workspace-scoped EDIT grants; this script mints each VU's
 *   HS256 JWT with k6/crypto (claims: sub, workspaces[], iat, exp, jti —
 *   the exact shape @factory/auth/tokens verifies).
 *
 * Rooms: the 1,000 VUs are spread over K6_PAGE_IDS (default 10 pages ×
 * 100 peers) so the update fan-out stays at ~100-way per room (a
 * 1,000-peer single room would multiply every update 10×; the awareness
 * fan-out cap of 50 peers/room in REALTIME_LIMITS is the design ceiling
 * the server documents for presence).
 *
 * Thresholds (fail the run when violated):
 *   rtt_ms              p(95) < 200 ms   (the DoD)
 *   protocol_errors     count < 1        (the DoD: 0 protocol errors)
 *   join_failures       rate  < 0.5 %
 *   ws_unexpected_closes rate  < 0.5 %   (mid-hold server-side drops)
 *
 * Host-side DoD checks (NOT k6 thresholds — the runner run-1k.sh does them):
 *   memory < 2 GB (RSS peak of the server process) and /health 200
 *   throughout the run.
 *
 * Run (CI perf runner — see README.md):
 *   node tests/performance/helpers/seed-1k-load.mjs      # once
 *   K6_REALTIME_URL=ws://localhost:4001 K6_JWT_SECRET=… \
 *   K6_WORKSPACE_ID=… K6_PAGE_IDS=p1,p2,…,p10 \
 *   pnpm --dir tests/performance realtime:1k
 *
 * Artifact: realtime-1k-final.json (k6 --summary-export: all samples +
 * percentiles; the VU/connection curve is the `vus` metric over time). The
 * per-PR smoke tier (`realtime:smoke` — 100 conns / 30 s) reuses this script
 * + the same seeder with a 2× flake-superset RTT threshold and a short ramp.
 *
 * k6 API note: this script uses the CURRENT k6/ws socket API
 * (ws.connect(url, params, setup) + socket.setInterval / socket.setTimeout
 * + socket.sendBinary — verified against internal/js/modules/k6/ws).
 * The pre-existing realtime.js (10k) still targets the legacy `new WS()`
 * + ws.wait() API; treat its frames/protocol as a placeholder until it is
 * re-pointed at this contract (recorded TASK-0091 note).
 */
import ws from "k6/ws";
import crypto from "k6/crypto";
import encoding from "k6/encoding";
import { Trend, Rate, Counter } from "k6/metrics";

const RTT_MS = new Trend("rtt_ms", true);
const CLOCK_SKEW_MS = new Trend("clock_skew_ms", true);
const joinFailures = new Rate("join_failures");
const protocolErrors = new Counter("protocol_errors");
const unexpectedCloses = new Rate("ws_unexpected_closes");
const updateAcks = new Counter("update_acks");

const WS_URL = __ENV.K6_REALTIME_URL || "ws://localhost:4001";
const SECRET = __ENV.K6_JWT_SECRET || "";
const WORKSPACE_ID = __ENV.K6_WORKSPACE_ID || "";
const PAGE_IDS = String(__ENV.K6_PAGE_IDS || "").split(",").filter(Boolean);
const VUS = parseInt(__ENV.K6_VUS || "1000", 10);
const RAMP_SECONDS = parseInt(__ENV.K6_RAMP_SECONDS || "30", 10);
const HOLD_SECONDS = parseInt(__ENV.K6_HOLD_SECONDS || "120", 10);
// RTT thresholds are env-overridable so the per-PR SMOKE tier (100 conns /
// 30 s — see the `realtime:smoke` script) can run a flake-superset (2× the
// DoD: p95<400 / p99<800) on noisy shared runners while the FULL 1k merge-
// to-develop lane enforces the strict DoD (p95<200 / p99<400). The protocol-
// correctness thresholds below (0 protocol errors, <0.5% joins/closes) are
// NEVER flake-superseted — a protocol drop is a hard regression, not jitter.
const RTT_P95_MS = parseInt(__ENV.K6_RTT_P95_MS || "200", 10);
const RTT_P99_MS = parseInt(__ENV.K6_RTT_P99_MS || "400", 10);

/**
 * Per-VU send cadence: 200 msg/s fleet / VUS. The spec pins the 200/VUS
 * pair (1,000); keep the two in lockstep so the fleet rate stays 200/s.
 */
const MSG_INTERVAL_MS = Math.max(100, Math.round((VUS * 1000) / 200));

/**
 * Valid Yjs updates (base64), each inserting a few bytes into Y.Text "k6"
 * under a DEDICATED Yjs client id — re-applying the same frame is an
 * idempotent no-op (the room never grows across the 2-min hold), and the
 * server's throwaway-doc `applyUpdate` validation passes.
 * Regenerate (not by hand!): see the TASK-0091 note in the seeder header.
 */
const YJS_UPDATES = [
  "AQGCprqECwAEAQJrNhlrNiBsb2FkIHVwZGF0ZSB2YXJpYW50IDAgAA==",
  "AQHgk6jsCgAEAQJrNhlrNiBsb2FkIHVwZGF0ZSB2YXJpYW50IDEgAA==",
  "AQGf2MjCBAAEAQJrNhlrNiBsb2FkIHVwZGF0ZSB2YXJpYW50IDIgAA==",
  "AQHzmP0+AAQBAms2GWs2IGxvYWQgdXBkYXRlIHZhcmlhbnQgMyAA",
];

const BINARY_KIND = { Update: 0x01, Awareness: 0x02 }; // frames.ts BINARY_KIND

export const options = {
  scenarios: {
    realtime_1k: {
      // NOTE: the executor is `ramping-vus` (the real k6 name in every
      // version). An earlier draft spelled it `ramping-vusers` (unknown
      // executor) — the script was never archive-validated; TASK-0095 S14
      // caught + fixed it (and pinned k6 v0.57.0 in the CI perf lane, the
      // first version where the `1_000` numeric separator below parses).
      executor: "ramping-vus",
      startVUs: 0,
      stages: [
        // The spec: 1,000 connections over 30 s (100/s connect rate).
        { duration: `${RAMP_SECONDS}s`, target: VUS },
        // …sustained 2 min at full fleet.
        { duration: `${HOLD_SECONDS}s`, target: VUS },
      ],
      // A VU's iteration (connect + hold + clean close) takes exactly the
      // scenario duration, so no iteration overlap.
      gracefulStop: "10s",
    },
  },
  thresholds: {
    // The RTT DoD (full lane: p95<200 / p99<400); the smoke tier overrides
    // via K6_RTT_P95_MS/K6_RTT_P99_MS (2× flake-superset — see above).
    rtt_ms: [`p(95)<${RTT_P95_MS}`, `p(99)<${RTT_P99_MS}`],
    protocol_errors: ["count<1"],
    join_failures: ["rate<0.005"],
    ws_unexpected_closes: ["rate<0.005"],
  },
};

/** base64url for JWT segments (no padding). */
function b64url(input) {
  return input.replace(/\+/g, "-").replace(/\//g, "_").replace(/=+$/, "");
}

/**
 * Mint this VU's access JWT (HS256, shared secret) — the exact claims
 * shape @factory/auth/tokens `verifyAccessToken` + the handshake verify
 * (apps/realtime/src/auth.ts) accept: sub (string), exp (unix seconds),
 * iat (unix seconds, not in the future), jti, workspaces (string[]).
 * `iss` is optional — omitted (checked only when present, note 13).
 */
function mintToken(sub, workspaces) {
  const nowSec = Math.floor(Date.now() / 1000);
  const header = b64url(encoding.b64encode(JSON.stringify({ alg: "HS256", typ: "JWT" }), "base64"));
  const claims = b64url(
    encoding.b64encode(
      JSON.stringify({
        sub: sub,
        workspaces: workspaces,
        iat: nowSec,
        exp: nowSec + 900, // 15 min (the API's ACCESS_TOKEN lifetime)
        jti: `k6-${sub}-${nowSec}`,
      }),
      "base64",
    ),
  );
  const sig = b64url(crypto.hmac("sha256", SECRET, `${header}.${claims}`, "base64"));
  return `${header}.${claims}.${sig}`;
}

/** binary 0x01 + a valid Yjs update → ArrayBuffer for ws.sendBinary. */
function updateFrame(variant) {
  const bytes = encoding.b64decode(YJS_UPDATES[variant % YJS_UPDATES.length], "base64");
  const out = new Uint8Array(bytes.length + 1);
  out[0] = BINARY_KIND.Update;
  out.set(bytes, 1);
  return out.buffer;
}

/** binary 0x02 + a small awareness state (the server relays it as-is). */
function awarenessFrame(userId) {
  const payload = JSON.stringify({
    user: { id: userId, name: "k6", color: "slate" },
    cursor: { x: (userId.length * 7) % 800, y: (userId.length * 13) % 600 },
  });
  const data = encoding.b64encode(payload, "utf8");
  const out = new Uint8Array(data.length + 1);
  out[0] = BINARY_KIND.Awareness;
  out.set(data, 1);
  return out.buffer;
}

export function setup() {
  if (!SECRET) throw new Error("setup: K6_JWT_SECRET is required (the shared HS256 secret)");
  if (!WORKSPACE_ID) throw new Error("setup: K6_WORKSPACE_ID is required (the seeder prints it)");
  if (PAGE_IDS.length === 0) throw new Error("setup: K6_PAGE_IDS is required (the seeder prints them)");
  return { ok: true };
}

export default function () {
  // One identity per VU (1-based __VU → stable across the ramp):
  // distinct user (100/user socket cap), distinct IP (30/min/IP connect
  // limiter — the server must run REALTIME_TRUST_PROXY=1), page round-robin
  // (100 peers/room fan-out).
  const vu = __VU;
  const pad4 = String(vu - 1).padStart(4, "0");
  const userId = `k6-load-u-${pad4}`;
  const ip = `10.100.${Math.floor((vu - 1) / 256)}.${(vu - 1) % 256}`;
  const pageId = PAGE_IDS[(vu - 1) % PAGE_IDS.length];
  const token = mintToken(userId, [WORKSPACE_ID]);

  const url = `${WS_URL}/realtime?token=${encodeURIComponent(token)}&room=${encodeURIComponent(`page:${pageId}`)}`;
  // Each VU connects when its ramp slot opens, then HOLDS until the
  // scenario end (__END_TIME) — so the fleet holds the FULL 2 minutes at
  // 1,000 (the ramp's 30 s stagger would otherwise taper the last
  // 30 s as early VUs finish first). ws.connect blocks the iteration
  // until the socket closes, so the hold is the iteration's duration.
  const holdMs = Math.max(1_000, __END_TIME - (Date.now() - __START_TIME));
  const deadline = Date.now() + holdMs;

  let joined = false;
  const sentTs = {}; // ping ts → true (cleared on the echo)

  const resp = ws.connect(
    url,
    { headers: { "x-forwarded-for": ip } },
    (socket) => {
      socket.on("message", (raw) => {
        let msg;
        try {
          msg = JSON.parse(raw);
        } catch {
          return; // non-JSON (shouldn't happen for text frames)
        }
        switch (msg.type) {
          case "hello":
            joined = true;
            if (msg.canEdit !== true) {
              // The seeded identity has a workspace-scoped EDIT grant; a
              // VIEW hello means the seed/permission wiring is wrong —
              // every update this VU sends would be a FORBIDDEN protocol
              // error, so count it NOW (the 0-protocol-errors DoD).
              protocolErrors.add(1);
            }
            if (typeof msg.serverTime === "number") {
              // One-time skew diagnostic (server - client), per the spec.
              CLOCK_SKEW_MS.add(msg.serverTime - Date.now());
            }
            break;
          case "ping-pong":
            if (typeof msg.ts === "number" && sentTs[msg.ts]) {
              RTT_MS.add(Date.now() - msg.ts);
              delete sentTs[msg.ts];
            }
            break;
          case "ack":
            updateAcks.add(1);
            break;
          case "error":
            // A server-side protocol error frame (FORBIDDEN / bad_update /
            // unknown_frame / …). The DoD: zero of these.
            protocolErrors.add(1);
            break;
          // sync.state / doc.persisted / comment.event / … — ignore.
        }
      });
      // Binary frames (relayed updates / awareness) — not RTT samples.
      socket.on("binaryMessage", () => {});
      socket.on("close", () => {
        // A close BEFORE the hold window ends is a mid-test disconnect.
        if (!joined || Date.now() < deadline - 5000) unexpectedCloses.add(1);
      });
      socket.on("error", () => {
        protocolErrors.add(1);
      });

      // The mixed 200 msg/s fleet (0.2 msg/s per VU), spec mix 50/30/20.
      // socket.setInterval runs the callback in the socket event loop.
      let tick = 0;
      socket.setInterval(() => {
        const r = Math.random();
        if (r < 0.5) {
          socket.sendBinary(updateFrame(tick));
        } else if (r < 0.8) {
          socket.sendBinary(awarenessFrame(userId));
        } else {
          const ts = Date.now();
          sentTs[ts] = true;
          socket.send(JSON.stringify({ type: "ping", ts: ts }));
        }
        tick += 1;
      }, MSG_INTERVAL_MS);

      // End the hold with a clean 1000 close (the close handler's window
      // check keeps a deadline close from counting as an unexpected close).
      socket.setTimeout(() => {
        if (joined) socket.close(1000);
      }, holdMs);
    },
  );

  if (resp.Status !== 101) {
    // The handshake was rejected at the upgrade (401/403/429/400 — the
    // pre-join rejections are HTTP statuses, no WS close frame exists).
    joinFailures.add(1);
  }
}