#!/usr/bin/env node
/**
 * AI Software Factory — host-side zero-dropped-Yjs-ops verification (TASK-0135).
 *
 * k6's goja VM has no Yjs, so the spec's "zero dropped Yjs ops (the final doc
 * state hash equals the sequential-replay hash)" cannot be a k6 threshold.
 * This script does it host-side, after `run-realtime-load.sh` finishes the k6
 * run:
 *
 *   1. The k6 `verify` scenario's writers (seeded load users) each sent the 4
 *      known Yjs updates (YJS_UPDATES, the same fixed set as realtime-1k.js —
 *      distinct inserts into the Y.Text "k6") to the dedicated verify room
 *      `page:<K6_VERIFY_PAGE_ID>`.
 *   2. This script joins that room as a seeded load user, sends a
 *      `sync.request` (no state vector → the server replies with the FULL doc
 *      state as base64 `sync.state` chunks, the R5 resume path), and
 *      reconstructs the server's final doc.
 *   3. It builds the EXPECTED doc: a fresh `Y.Doc()` + the room baseline
 *      (`yMeta.pageId` — exactly what apps/realtime/src/rooms.ts builds for a
 *      page room with no seeded blocks) + the 4 YJS_UPDATES applied in order
 *      (the sequential replay).
 *   4. It compares the canonical full-state encodings
 *      (`Y.encodeStateAsUpdate` of each). Equal ⇒ zero ops dropped. A lost
 *      update changes the "k6" text and therefore the hash ⇒ FAIL.
 *
 * Timing (IMPORTANT): a room is evicted (and later re-hydrated from the doc
 * store, where the "k6" text is NOT represented as blocks) only after its GC
 * warm window (~5 min after the last peer leaves). The runner invokes this
 * checker IMMEDIATELY after k6 exits, well inside the window, so it sees the
 * live in-memory doc. If the hash mismatches, the first suspect is that the
 * room was re-hydrated — re-run the check within 5 min of the k6 run.
 *
 * Env:
 *   K6_REALTIME_URL    ws base (default ws://localhost:4001)
 *   K6_JWT_SECRET      shared HS256 secret (required — mints the checker's token)
 *   K6_WORKSPACE_ID    the loaded workspace (required)
 *   K6_VERIFY_PAGE_ID  the verify room page (default load-p-verify)
 *
 * Exit code: 0 = hashes match (zero dropped ops), 1 = mismatch or error.
 *
 * Run:  node tests/performance/verify-zero-op-loss.mjs
 */
import crypto from "node:crypto";
import { createRequire } from "node:module";

// Resolve yjs/ws from this package's own deps (pnpm links them here).
const require = createRequire(import.meta.url);
const Y = require("yjs");
const WebSocket = require("ws");

const WS_URL = process.env.K6_REALTIME_URL || "ws://localhost:4001";
const SECRET = process.env.K6_JWT_SECRET || "";
const WORKSPACE_ID = process.env.K6_WORKSPACE_ID || "";
const VERIFY_PAGE_ID = process.env.K6_VERIFY_PAGE_ID || "load-p-verify";

// MUST stay in sync with realtime-load.js's YJS_UPDATES (the fixed set the
// verify writers send — distinct inserts of "k6 load update variant N" into
// the Y.Text "k6").
const YJS_UPDATES = [
  "AQGCprqECwAEAQJrNhlrNiBsb2FkIHVwZGF0ZSB2YXJpYW50IDAgAA==",
  "AQHgk6jsCgAEAQJrNhlrNiBsb2FkIHVwZGF0ZSB2YXJpYW50IDEgAA==",
  "AQGf2MjCBAAEAQJrNhlrNiBsb2FkIHVwZGF0ZSB2YXJpYW50IDIgAA==",
  "AQHzmP0+AAQBAms2GWs2IGxvYWQgdXBkYXRlIHZhcmlhbnQgMyAA",
];

function b64url(input) {
  return input.replace(/\+/g, "-").replace(/\//g, "_").replace(/=+$/, "");
}

/** Mint an HS256 access token (same claims shape the realtime server accepts). */
function mintToken(sub) {
  const nowSec = Math.floor(Date.now() / 1000);
  const header = b64url(Buffer.from(JSON.stringify({ alg: "HS256", typ: "JWT" })).toString("base64"));
  const claims = b64url(
    Buffer.from(
      JSON.stringify({ sub, workspaces: [WORKSPACE_ID], iat: nowSec, exp: nowSec + 600, jti: `verify-${sub}-${nowSec}` }),
    ).toString("base64"),
  );
  const sig = b64url(crypto.createHmac("sha256", SECRET).update(`${header}.${claims}`).digest("base64"));
  return `${header}.${claims}.${sig}`;
}

function sha256(b64) {
  return crypto.createHash("sha256").update(Buffer.from(b64, "base64")).digest("hex");
}

function fullStateB64(doc) {
  return Buffer.from(Y.encodeStateAsUpdate(doc)).toString("base64");
}

/**
 * Join the verify room and pull the full doc state (R5 sync.request →
 * sync.state chunks). Resolves the concatenated state bytes, or rejects.
 */
function fetchRoomState(token) {
  return new Promise((resolve, reject) => {
    const url = `${WS_URL}/realtime?token=${encodeURIComponent(token)}&room=${encodeURIComponent(`page:${VERIFY_PAGE_ID}`)}`;
    const ws = new WebSocket(url, { headers: { "x-forwarded-for": "10.100.250.250" } });
    let chunks = [];
    let gotHello = false;
    const timer = setTimeout(() => {
      reject(new Error(`timeout: no full sync within 30 s (joined=${gotHello})`));
      ws.close();
    }, 30_000);
    ws.on("open", () => {
      // Wait for hello before requesting the sync (the server may not have
      // joined us to the room yet).
    });
    ws.on("message", (data) => {
      const buf = Buffer.isBuffer(data) ? data : Buffer.from(data);
      if (buf.length === 0) return;
      const kind = buf[0];
      if (kind === 0x01 || kind === 0x02) return; // binary (updates/awareness) — not ours
      let msg;
      try {
        msg = JSON.parse(buf.toString("utf8"));
      } catch {
        return;
      }
      if (msg.type === "hello") {
        gotHello = true;
        // No stateVector / lastSeq → the server replies with the FULL doc
        // state (sendStateDelta with sv=null → encodeStateAsUpdate(doc)).
        ws.send(JSON.stringify({ type: "sync.request" }));
      } else if (msg.type === "sync.state") {
        if (msg.chunk) chunks.push(msg.chunk);
        if (msg.done === true) {
          clearTimeout(timer);
          ws.close();
          resolve(Buffer.concat(chunks.map((c) => Buffer.from(c, "base64"))));
        }
      } else if (msg.type === "error") {
        clearTimeout(timer);
        ws.close();
        reject(new Error(`server error frame: ${JSON.stringify(msg)}`));
      }
    });
    ws.on("error", (err) => {
      clearTimeout(timer);
      reject(err);
    });
  });
}

async function main() {
  if (!SECRET) throw new Error("K6_JWT_SECRET is required");
  if (!WORKSPACE_ID) throw new Error("K6_WORKSPACE_ID is required");

  // 1. The server's final doc state for the verify room.
  const stateBytes = await fetchRoomState(mintToken("k6-load-u-0000"));
  const serverDoc = new Y.Doc();
  Y.applyUpdate(serverDoc, stateBytes, "remote");
  const serverState = fullStateB64(serverDoc);

  // 2. The expected doc: fresh baseline + the 4 updates in order (the
  //    sequential replay). The baseline mirrors rooms.ts: a page room with no
  //    seeded blocks = new Y.Doc() + yMeta.pageId.
  const expectedDoc = new Y.Doc();
  expectedDoc.getMap("yMeta").set("pageId", VERIFY_PAGE_ID);
  for (const b64 of YJS_UPDATES) {
    Y.applyUpdate(expectedDoc, Buffer.from(b64, "base64"), "remote");
  }
  const expectedState = fullStateB64(expectedDoc);

  const serverHash = sha256(serverState);
  const expectedHash = sha256(expectedState);
  const serverText = serverDoc.getText("k6").toString();
  const expectedText = expectedDoc.getText("k6").toString();
  const missing = YJS_UPDATES.map((_, i) => `k6 load update variant ${i}`).filter(
    (marker) => !serverText.includes(marker),
  );

  console.log("zero-op-loss verification:");
  console.log(`  room           page:${VERIFY_PAGE_ID}`);
  console.log(`  server state   ${stateBytes.byteLength} bytes, sha256=${serverHash}`);
  console.log(`  replay state   ${Buffer.from(expectedState, "base64").byteLength} bytes, sha256=${expectedHash}`);
  console.log(`  server "k6"    ${JSON.stringify(serverText)}`);
  console.log(`  replay "k6"    ${JSON.stringify(expectedText)}`);

  serverDoc.destroy();
  expectedDoc.destroy();

  if (serverHash === expectedHash) {
    console.log("PASS — final doc state == sequential-replay hash (zero dropped Yjs ops)");
    process.exit(0);
  }

  console.error("FAIL — final doc state != sequential-replay hash (dropped or extra ops)");
  if (missing.length) console.error(`  missing markers: ${missing.join(", ")}`);
  console.error(
    "  hint: if the room was evicted + re-hydrated from the doc store (GC warm window, ~5 min),\n" +
      "  the 'k6' text is not represented there — re-run this check within 5 min of the k6 run.",
  );
  process.exit(1);
}

main().catch((err) => {
  console.error("verify-zero-op-loss failed:", err.message || err);
  process.exit(1);
});