/**
 * AI Software Factory — k6: realtime 10k load (TASK-0135, the §26 matrix).
 *
 * Scenario (the spec, verbatim):
 *   - connect   10,000 concurrent ws over 5 min (100 rooms × 100 conns,
 *               jittered join), then held 15 min steady state.
 *   - edit      all 10k send block-edit (Yjs update) ops, 200 msg/s aggregate
 *               (rate-limited per-conn: 200/10000 = 0.02 msg/s = 1 op/50 s).
 *   - presence  1,000 presence identities, join/leave churn 10/min.
 *   - (verify)  a dedicated room the zero-dropped-op hash check targets.
 *
 * Thresholds (staging-hard):
 *   op_ack_ms           p(95) < 200 ms   (client-observed send-update → ack)
 *   connect_ms          p(95) < 1 000 ms (connect → hello)
 *   protocol_errors     count  < 1       (0 FORBIDDEN / bad_update / error)
 *   ws_unexpected_closes rate  < 0.005   (disconnects < 0.5%, excl. the
 *                                          scripted clean close at deadline)
 *
 * Zero-dropped Yjs ops (the spec's "final doc state hash equals the
 * sequential-replay hash") is a HOST-SIDE check, not a k6 threshold — k6's
 * goja VM has no Yjs to hash a doc. `run-realtime-load.sh` (this dir) does it:
 * the k6 `verify-writers` scenario sends a fixed, known set of Yjs updates to
 * the `page:verify` room; the runner connects, full-syncs that room, and asserts
 * the server's final doc state hashes to the same value as a fresh Yjs doc with
 * exactly those updates applied in order. A lost update changes the hash.
 *
 * Protocol (the real TASK-0091 contract — packages/shared frames.ts), reusing
 * tests/performance/realtime-1k.js conventions:
 *   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  → server logs it and replies
 *   ack:      {type:"ack", seq}         (the op-ack — measured here)
 *   awareness: binary 0x02 + JSON state
 *   ping:     {type:"ping", ts}         → {type:"ping-pong", ts}
 *
 * Why 10,000 distinct identities + distinct source IPs (MANDATORY): the
 * server caps 100 live sockets/user + 10/user control + 30/min connect per
 * source IP. One shared token can never reach 10,000 (the 101st is 429). Each
 * VU mints its own HS256 JWT (k6/crypto) for a distinct seeded user id and
 * presents a per-socket X-Forwarded-For, so the server must run
 * REALTIME_TRUST_PROXY=1. The identities + EDIT grants come from the seeder
 * (helpers/seed-1k-load.mjs with K6_USERS=10000 + K6_PAGE_IDS = 100 rooms).
 *
 * Env:
 *   K6_REALTIME_URL    ws base (default ws://localhost:4001)
 *   K6_JWT_SECRET      shared HS256 secret (required — must match the server)
 *   K6_WORKSPACE_ID    the loaded workspace (required)
 *   K6_PAGE_IDS        100 room page ids, comma-separated (required)
 *   K6_VUS             total edit VUs (default 10000)
 *   K6_RAMP_SECONDS    ramp 0→VUS (default 300 = 5 min)
 *   K6_HOLD_SECONDS    steady state (default 900 = 15 min)
 *   K6_PRESENCE_VUS    presence identities (default 1000)
 *   K6_OP_ACK_P95_MS   op-ack p95 budget (default 200)
 *   K6_CONNECT_P95_MS  connect p95 budget (default 1000)
 *   K6_STRICT          "1" hard (default) / "0" report-only (trivially-true
 *                      thresholds; the run still prints real values)
 *
 * Run (staging):  K6_STRICT=1 … run-realtime-load.sh
 * Run (smoke):    K6_PROFILE=smoke pnpm --dir tests/performance realtime-load
 */
import ws from "k6/ws";
import crypto from "k6/crypto";
import encoding from "k6/encoding";
import { sleep } from "k6";
import { Trend, Rate, Counter } from "k6/metrics";

const OP_ACK_MS = new Trend("op_ack_ms", true);
const CONNECT_MS = new Trend("connect_ms", true);
const RTT_MS = new Trend("rtt_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 presenceJoins = new Counter("presence_joins");

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 SMOKE = __ENV.K6_PROFILE === "smoke";
const STRICT = (__ENV.K6_STRICT || "1") !== "0";

const VUS = parseInt(__ENV.K6_VUS || (SMOKE ? "100" : "10000"), 10);
const RAMP_SECONDS = parseInt(__ENV.K6_RAMP_SECONDS || (SMOKE ? "10" : "300"), 10);
const HOLD_SECONDS = parseInt(__ENV.K6_HOLD_SECONDS || (SMOKE ? "20" : "900"), 10);
const PRESENCE_VUS = parseInt(__ENV.K6_PRESENCE_VUS || (SMOKE ? "10" : "1000"), 10);
const OP_ACK_P95_MS = parseInt(__ENV.K6_OP_ACK_P95_MS || "200", 10);
const CONNECT_P95_MS = parseInt(__ENV.K6_CONNECT_P95_MS || "1000", 10);

// 200 msg/s aggregate across VUS ⇒ per-VU cadence (rate-limited per-conn).
const MSG_INTERVAL_MS = Math.max(100, Math.round((VUS * 1000) / 200));
// Presence churn: 10 join/leave cycles per minute fleet-wide. A small active
// set of `PRESENCE_ACTIVE` VUs (constant-vus) each runs one join→awareness→
// leave cycle per minute; each wave draws PRESENCE_ACTIVE DISTINCT identities
// (see presenceChurn), advancing the pool by that many — so all PRESENCE_VUS
// identities are exercised over the run. Smoke caps the pool to 10 (the
// seeded smoke fixture has 100 load users; the pool only needs a slice).
const PRESENCE_ACTIVE = 10;

// The zero-dropped-op verify room + its fixed update set (host-replayed).
// A dedicated seeded page id for the verify room (the server only joins
// rooms for real pages). The runner/seeder creates it (K6_PAGES includes
// load-p-verify); overridable.
const VERIFY_ROOM = __ENV.K6_VERIFY_PAGE_ID || "load-p-verify";
// 4 valid Yjs updates (base64) — each inserts a distinct marker into the Y.Text
// "k6". Re-applying is an idempotent no-op, but a LOST one is not: the final
// doc is missing that insert, so its state hash differs from the replay.
// (Same regenerated set as realtime-1k.js — see its header note.)
const YJS_UPDATES = [
  "AQGCprqECwAEAQJrNhlrNiBsb2FkIHVwZGF0ZSB2YXJpYW50IDAgAA==",
  "AQHgk6jsCgAEAQJrNhlrNiBsb2FkIHVwZGF0ZSB2YXJpYW50IDEgAA==",
  "AQGf2MjCBAAEAQJrNhlrNiBsb2FkIHVwZGF0ZSB2YXJpYW50IDIgAA==",
  "AQHzmP0+AAQBAms2GWs2IGxvYWQgdXBkYXRlIHZhcmlhbnQgMyAA",
];
const BINARY_KIND = { Update: 0x01, Awareness: 0x02 };

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

/**
 * Mint an access JWT (HS256, shared secret) — the exact claims shape
 * @factory/auth/tokens + apps/realtime/src/auth.ts accept: sub, workspaces[],
 * iat, exp (unix seconds), jti. (Reused from realtime-1k.js.)
 */
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,
        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. */
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 (100 room page ids)");
  if (!SMOKE && PAGE_IDS.length < 100) {
    throw new Error(`setup: full profile needs 100 room page ids, got ${PAGE_IDS.length}`);
  }
  return { ok: true };
}

/**
 * The 10k edit scenario: connect (jittered), hold, send Yjs update ops at the
 * per-conn rate-limited cadence, clean close. Measures connect_ms + op_ack_ms.
 */
export function editLoad() {
  const vu = __VU;
  // 4-digit ids — the seeder (helpers/seed-1k-load.mjs) creates
  // k6-load-u-0000 … k6-load-u-9999 (10,000 fit exactly in 4 digits).
  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}`;
  // 100 rooms × 100 conns: VU i → room (i-1) mod 100.
  const roomIdx = (vu - 1) % Math.max(1, Math.min(100, PAGE_IDS.length));
  const pageId = PAGE_IDS[roomIdx];
  const token = mintToken(userId, [WORKSPACE_ID]);

  // Real thundering-herd de-correlation: a per-VU 0..~1.05 s pre-DELAY (an
  // actual sleep, not a timestamp offset) so the 100 conns of a room don't all
  // hit the upgrade in the same instant. The ramp already spreads joins; this
  // adds room-level de-correlation. connectStart is the TRUE connect time
  // (after the delay) so CONNECT_MS (connect→hello) is a real, non-negative
  // latency.
  sleep((vu % 7) * 0.15);
  const connectStart = Date.now();

  const url = `${WS_URL}/realtime?token=${encodeURIComponent(token)}&room=${encodeURIComponent(`page:${pageId}`)}`;
  const holdMs = Math.max(1000, HOLD_SECONDS * 1000);
  let joined = false;
  let lastSentTs = 0; // for op-ack (one outstanding update per VU at this rate)
  const sentTs = {};

  // NOTE: ws.connect's return value is NOT used for success detection. On
  // k6 v0.57 the returned object has no reliable `.Status` (measured
  // `undefined` on a successful join, and property access on the native
  // value throws in goja) — so `resp.Status !== 101` would false-positive
  // (or throw) on every HEALTHY join. The only robust success signal is the
  // server's `hello` frame, which it sends only after the full auth + room +
  // permission chain resolves. `joined` (set in the on-message handler below)
  // is therefore the join-success oracle; `!joined` after the hold = a join
  // failure (covers 401/403/400/429 upgrade rejects AND the 500-permission
  // path, which upgrades OK but never sends hello).
  ws.connect(
    url,
    { headers: { "x-forwarded-for": ip } },
    (socket) => {
      socket.on("message", (raw) => {
        let msg;
        try {
          msg = JSON.parse(raw);
        } catch {
          return;
        }
        switch (msg.type) {
          case "hello":
            joined = true;
            CONNECT_MS.add(Date.now() - connectStart); // connect → hello
            if (msg.canEdit !== true) protocolErrors.add(1); // VIEW hello = wiring bug
            break;
          case "ack":
            // op-ack: the client-observed latency of a sent update.
            updateAcks.add(1);
            if (lastSentTs) OP_ACK_MS.add(Date.now() - lastSentTs);
            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 "error":
            protocolErrors.add(1);
            break;
          // sync.state / doc.persisted / … — ignore.
        }
      });
      socket.on("binaryMessage", () => {}); // relayed updates/awareness
      socket.on("close", () => {
        // "Unexpected" = we HAD a good join (hello arrived) and the socket
        // dropped before the hold window. A failed join (rejected upgrade, or
        // the 500-permission path — no hello) is a join_failure, not an
        // unexpected close, so gate on `joined`.
        if (joined && Date.now() < connectStart + holdMs - 5000) unexpectedCloses.add(1);
      });
      socket.on("error", () => protocolErrors.add(1));

      // The rate-limited 200 msg/s aggregate: one op per MSG_INTERVAL_MS,
      // alternating update / awareness / ping (the block-edit op is the update).
      let tick = 0;
      socket.setInterval(() => {
        const r = Math.random();
        if (r < 0.6) {
          lastSentTs = Date.now();
          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);

      socket.setTimeout(() => {
        if (joined) socket.close(1000); // scripted clean close (not a disconnect)
      }, holdMs);
    },
  );

  // HOLD the VU. k6 v0.57's ws.connect is NON-BLOCKING: it returns as soon as
  // the upgrade is initiated, so without an explicit hold the iteration
  // returns instantly and k6 fires the next one — a tight connect-retry loop
  // (measured: ~1,068 iters/s, 42k joins in a 30 s smoke). The socket's
  // setInterval (edit cadence) + setTimeout (scripted close at holdMs) keep
  // the connection alive and drive the metrics; each 1 s sleep YIELDS to the
  // socket event loop (k6 timers need the VU to yield), so the edit cadence
  // keeps ticking. The socket's own close ends this iteration (k6 returns a
  // VU to the pool when its socket closes).
  const editDeadline = Date.now() + holdMs + 5_000; // grace past the scripted close
  while (Date.now() < editDeadline) sleep(1);

  // Join-failure detection — the `hello` oracle (see the note above ws.connect).
  // No hello by the end of the hold = the join never established (bad token /
  // 403 / 429 / 400, or the 500-permission path that upgrades but never
  // sends hello). On a healthy post-fix server this stays ~0; on the current
  // buggy server it is ~1.0 (the dev report prints it without blocking).
  if (!joined) joinFailures.add(1);
}

/**
 * Presence churn: 10 active VUs, each a join→awareness→leave cycle once a
 * minute (10/min fleet), round-robining the PRESENCE_VUS identity pool.
 */
export function presenceChurn() {
  const cycle = __ITER + 1;
  // 10 active VUs × one cycle per minute = 10 join/leave cycles per minute
  // fleet-wide. The `+ (__VU - 1)` offset makes each wave draw 10 DISTINCT
  // identities (constant-vus gives every VU its own __ITER, so without the
  // offset all 10 VUs of a wave would reuse the same identity); each wave
  // advances the pool by PRESENCE_ACTIVE, so all PRESENCE_VUS identities are
  // exercised over the run.
  const userIdx = ((cycle - 1) * PRESENCE_ACTIVE + (__VU - 1)) % PRESENCE_VUS;
  // Presence identities reuse the seeded load-user pool (k6-load-u-XXXX —
  // they carry the workspace EDIT grant; a k6-pres-u-* id would have no user
  // row and 403 the join).
  const pad4 = String(userIdx).padStart(4, "0");
  const userId = `k6-load-u-${pad4}`;
  const ip = `10.100.${200 + Math.floor(userIdx / 256)}.${userIdx % 256}`;
  const pageId = PAGE_IDS[userIdx % PAGE_IDS.length];
  const token = mintToken(userId, [WORKSPACE_ID]);

  const url = `${WS_URL}/realtime?token=${encodeURIComponent(token)}&room=${encodeURIComponent(`page:${pageId}`)}`;
  // joined is the join-success oracle (hello frame) — see the note in
  // editLoad. Presence joins are short; a join that never hellos within the
  // 5 s dwell is a join_failure.
  let joined = false;
  ws.connect(
    url,
    { headers: { "x-forwarded-for": ip } },
    (socket) => {
      socket.on("message", (raw) => {
        let msg;
        try {
          msg = JSON.parse(raw);
        } catch {
          return;
        }
        if (msg.type === "hello") {
          joined = true;
          presenceJoins.add(1);
          socket.sendBinary(awarenessFrame(userId));
        } else if (msg.type === "error") {
          protocolErrors.add(1);
        }
      });
      socket.on("binaryMessage", () => {});
      // Short presence dwell (5 s) then leave — the join/leave churn.
      socket.setTimeout(() => {
        if (joined) socket.close(1000);
      }, 5000);
    },
  );

  // HOLD the VU (ws.connect is non-blocking on k6 v0.57 — same as editLoad):
  // hold ~7 s past the scripted 5 s presence dwell, then let the next cycle
  // draw a fresh identity.
  const presenceDeadline = Date.now() + 7_000;
  while (Date.now() < presenceDeadline) sleep(1);
  if (!joined) joinFailures.add(1);
}

/**
 * The zero-dropped-op verify writers: a small set of VUs in the dedicated
 * `page:verify` room, each sending the 4 known Yjs updates exactly once. The
 * host runner (run-realtime-load.sh) full-syncs this room and asserts its doc
 * state hash == a fresh doc replayed with these 4 updates (loss ⇒ mismatch).
 */
export function verifyWriters() {
  // Reuse the first seeded load users (k6-load-u-0000… — they exist in both
  // the 100-user smoke seed and the 10,000-user full seed and carry the
  // workspace EDIT grant).
  const userId = `k6-load-u-${String(__VU - 1).padStart(4, "0")}`;
  const ip = `10.100.250.${__VU}`;
  const token = mintToken(userId, [WORKSPACE_ID]);
  const url = `${WS_URL}/realtime?token=${encodeURIComponent(token)}&room=${encodeURIComponent(`page:${VERIFY_ROOM}`)}`;
  // joined is the join-success oracle (hello frame) — see the note in
  // editLoad. A verify writer that never hellos sends no updates (the update
  // send is gated on `joined`), so the host-side hash check would already
  // catch the loss; this metric records the join failure explicitly.
  let joined = false;
  ws.connect(
    url,
    { headers: { "x-forwarded-for": ip } },
    (socket) => {
      socket.on("message", (raw) => {
        let msg;
        try {
          msg = JSON.parse(raw);
        } catch {
          return;
        }
        if (msg.type === "hello") joined = true;
        else if (msg.type === "error") protocolErrors.add(1);
      });
      socket.on("binaryMessage", () => {});
      socket.on("close", () => {});
      // Send all 4 verify updates once the room is joined, then hold.
      socket.setTimeout(() => {
        if (!joined) return;
        for (let v = 0; v < YJS_UPDATES.length; v++) {
          socket.sendBinary(updateFrame(v));
        }
      }, 2000);
      socket.setTimeout(() => {
        if (joined) socket.close(1000);
      }, Math.max(4000, HOLD_SECONDS * 1000));
    },
  );

  // HOLD the VU (ws.connect is non-blocking on k6 v0.57): hold through the
  // scripted close (max(4 s, HOLD) + the 2 s send delay) so this VU's verify
  // updates are actually delivered before the iteration ends.
  const verifyDeadline = Date.now() + Math.max(4000, HOLD_SECONDS * 1000) + 5_000;
  while (Date.now() < verifyDeadline) sleep(1);
  if (!joined) joinFailures.add(1);
}

const HARD_THRESHOLDS = {
  op_ack_ms: [`p(95)<${OP_ACK_P95_MS}`],
  connect_ms: [`p(95)<${CONNECT_P95_MS}`],
  // A load run that establishes ZERO connections has measured nothing — the
  // op-ack / connect Trends stay empty (avg 0) and would otherwise pass the
  // p95 budgets vacuously while 100% of handshakes fail. Gate on the
  // establishment rate so the hard tier actually requires healthy joins.
  // (Soft is `rate<=1` = always-pass, so report-only runs stay green.)
  join_failures: ["rate<0.001"],
  protocol_errors: ["count<1"],
  ws_unexpected_closes: ["rate<0.005"],
};
// Report-only (K6_STRICT=0): thresholds that can NEVER fail, so the dev/main
// tier always stays green while the summary still prints the real values.
// NOTE: k6 Rate thresholds are a FRACTION (0..1), so `rate<=1` is the true
// "always-pass" bound (a 100% failure rate is exactly 1.0, which would fail
// `rate<1` — that red-on-soft bug is what the `<=` fixes).
const SOFT_THRESHOLDS = {
  op_ack_ms: ["avg<1e15"],
  connect_ms: ["avg<1e15"],
  join_failures: ["rate<=1"],
  protocol_errors: ["count<1e15"],
  ws_unexpected_closes: ["rate<=1"],
};

export const options = {
  scenarios: {
    // 10k concurrent over RAMP (5 min), held HOLD (15 min) steady.
    edit: {
      executor: "ramping-vus",
      startVUs: 0,
      stages: [
        { duration: `${RAMP_SECONDS}s`, target: VUS },
        { duration: `${HOLD_SECONDS}s`, target: VUS },
      ],
      gracefulStop: "10s",
      exec: "editLoad",
    },
    // Presence churn 10/min (PRESENCE_ACTIVE VUs × 1 cycle/min).
    presence: {
      executor: "constant-vus",
      startTime: "0s",
      duration: `${RAMP_SECONDS + HOLD_SECONDS}s`,
      vus: PRESENCE_ACTIVE,
      exec: "presenceChurn",
    },
    // Zero-dropped-op verify writers (dedicated room).
    verify: {
      executor: "constant-vus",
      startTime: "0s",
      duration: `${RAMP_SECONDS + HOLD_SECONDS + 10}s`,
      vus: SMOKE ? 1 : 4,
      exec: "verifyWriters",
    },
  },
  thresholds: STRICT ? HARD_THRESHOLDS : SOFT_THRESHOLDS,
  summaryTrendStats: ["avg", "min", "med", "p(90)", "p(95)", "p(99)", "max"],
};