#!/usr/bin/env node
// [int->REQ-DEP-08] [int->REQ-DEP-06]
// scripts/soak-staging.mjs
// Source: 05-CONTEXT.md D-11 + 05-RESEARCH.md §Pattern 7
//         + 05-PATTERNS.md §"scripts/soak-staging.mjs"
// 30-min 2-client soak against wss://rebno-staging.fly.dev. Asserts WS stability
// (DEP-08) and OTLP ingestion (DEP-06 live verification, best-effort).
//
// Env:
//   STAGING_WSS_URL              (default: wss://rebno-staging.fly.dev)
//   STAGING_INVITE_TOKEN         (required when STAGING_MODE=1 on server)
//   SOAK_DURATION_MINUTES        (default: 30)
//   PROTOCOL_VERSION             (default: read from packages/protocol dist)
//   OBS_URL                      (default: http://rebno-obs.flycast:5080; reachable only via wireguard)
//   OBS_USER, OBS_PASS           (for OpenObserve REST auth — optional)
//
// Exit: 0 PASS, 1 FAIL.

import { setTimeout as sleep } from 'node:timers/promises';
import { fileURLToPath } from 'node:url';
import path from 'node:path';

const __dirname = path.dirname(fileURLToPath(import.meta.url));

// ---------------------------------------------------------------------------
// Config from env
// ---------------------------------------------------------------------------
const STAGING_WSS_URL = process.env.STAGING_WSS_URL ?? 'wss://rebno-staging.fly.dev';
const STAGING_INVITE_TOKEN = process.env.STAGING_INVITE_TOKEN ?? '';
const SOAK_DURATION_MINUTES = Number(process.env.SOAK_DURATION_MINUTES ?? '30');
const OBS_URL = process.env.OBS_URL ?? 'http://rebno-obs.flycast:5080';
const OBS_USER = process.env.OBS_USER ?? '';
const OBS_PASS = process.env.OBS_PASS ?? '';

// ---------------------------------------------------------------------------
// PROTOCOL_VERSION — read from the built package, fallback to literal 1
// ---------------------------------------------------------------------------
async function loadProtocolVersion() {
  const distPath = path.resolve(__dirname, '../packages/protocol/dist/index.js');
  try {
    const mod = await import(distPath);
    if (typeof mod.PROTOCOL_VERSION === 'number') {
      return mod.PROTOCOL_VERSION;
    }
  } catch {
    // dist not built — try workspace alias
  }
  try {
    const mod = await import('@rebno/protocol');
    if (typeof mod.PROTOCOL_VERSION === 'number') {
      return mod.PROTOCOL_VERSION;
    }
  } catch {
    // not resolvable from this script — use hardcoded value
  }
  return 1; // current PROTOCOL_VERSION constant (packages/protocol/src/version.ts)
}

// ---------------------------------------------------------------------------
// Shared stats accumulator
// ---------------------------------------------------------------------------
const stats = {
  /** Count of onLeave events with code !== 1000 (spurious disconnects). */
  disconnects: 0,
  /** Count of rate_limited / RATE_LIMITED messages received during baseline cadence. */
  rateLimited: 0,
  /** Round-trip latency samples (ms): time from input send → first onStateChange. */
  latencies: [],
};

// ---------------------------------------------------------------------------
// joinClient — mirrors authority.integ.test.ts:36-54, adds invite support
// ---------------------------------------------------------------------------
async function joinClient(name, protocolVersion) {
  const { Client } = await import('@colyseus/sdk');
  const client = new Client(STAGING_WSS_URL);

  const room = await client.joinOrCreate('rebno', {
    protocol_version: protocolVersion,
    invite: STAGING_INVITE_TOKEN,
  });

  let lastInputAt = 0;

  room.onMessage('*', (type, _msg) => {
    if (type === 'rate_limited' || type === 'RATE_LIMITED') {
      stats.rateLimited++;
      console.warn(`soak:staging [${name}]: received RATE_LIMITED (total=${stats.rateLimited})`);
    }
  });

  room.onStateChange(() => {
    if (lastInputAt > 0) {
      const latency = Date.now() - lastInputAt;
      stats.latencies.push(latency);
      lastInputAt = 0;
    }
  });

  room.onLeave((code) => {
    if (code !== 1000) {
      stats.disconnects++;
      console.warn(`soak:staging [${name}]: unexpected onLeave code=${code} (total disconnects=${stats.disconnects})`);
    }
  });

  let seqCounter = 0;

  return {
    name,
    room,
    sendMove() {
      lastInputAt = Date.now();
      room.send('input', {
        type: 'input',
        seq: ++seqCounter,
        dt_ms: 100,
        axis_x: 0,
        axis_y: 0,
        jump: false,
        action_btns: 0,
      });
    },
    sendChat(text) {
      room.send('chat_send', { type: 'chat_send', text });
    },
    async close() {
      try {
        await room.leave(true);
      } catch {
        /* best-effort */
      }
    },
  };
}

// ---------------------------------------------------------------------------
// p95 latency
// ---------------------------------------------------------------------------
function p95(arr) {
  if (!arr.length) return 0;
  const sorted = [...arr].sort((a, b) => a - b);
  const idx = Math.floor(sorted.length * 0.95);
  return sorted[idx] ?? sorted[sorted.length - 1];
}

// ---------------------------------------------------------------------------
// OpenObserve ingest check (best-effort; fails gracefully if flycast unreachable)
// ---------------------------------------------------------------------------
async function checkObsIngest() {
  if (!OBS_USER || !OBS_PASS) {
    console.log('soak:staging: OpenObserve query SKIPPED (OBS_USER/OBS_PASS unset — expected in public CI)');
    return null;
  }
  try {
    const auth = Buffer.from(`${OBS_USER}:${OBS_PASS}`).toString('base64');
    const url = `${OBS_URL}/api/default/traces/_search?size=1`;
    const res = await fetch(url, {
      headers: { Authorization: `Basic ${auth}` },
      signal: AbortSignal.timeout(5_000),
    });
    if (!res.ok) {
      console.warn(`soak:staging: OpenObserve query returned ${res.status} — SKIPPED`);
      return null;
    }
    const body = await res.json();
    const hits = body?.hits?.total ?? body?.total ?? null;
    if (hits !== null && hits < 1) {
      console.warn(`soak:staging: WARN — OpenObserve reports 0 trace hits (DEP-06 ingest may be broken)`);
    } else if (hits !== null) {
      console.log(`soak:staging: OpenObserve traces hits=${hits} (DEP-06 ingest verified)`);
    }
    return hits;
  } catch (err) {
    console.warn(
      `soak:staging: OpenObserve query unreachable (expected if CI lacks wireguard/flycast): ${err?.message}`,
    );
    return null;
  }
}

// ---------------------------------------------------------------------------
// main
// ---------------------------------------------------------------------------
async function main() {
  const PROTOCOL_VERSION = await loadProtocolVersion();
  console.log(
    `soak:staging: connecting to ${STAGING_WSS_URL}, duration=${SOAK_DURATION_MINUTES}m, protocol_version=${PROTOCOL_VERSION}`,
  );

  if (!STAGING_INVITE_TOKEN) {
    console.warn('soak:staging: WARN — STAGING_INVITE_TOKEN is unset; will fail if server enforces STAGING_MODE');
  }

  const alice = await joinClient('alice', PROTOCOL_VERSION);
  const bob = await joinClient('bob', PROTOCOL_VERSION);

  console.log('soak:staging: both clients joined — beginning soak loop');

  const endAt = Date.now() + SOAK_DURATION_MINUTES * 60_000;
  let tick = 0;

  while (Date.now() < endAt) {
    // 10 Hz move input (one send per 100ms loop iteration)
    alice.sendMove();
    bob.sendMove();

    // 0.1 Hz chat (every 100 ticks × 100ms = every 10s)
    if (tick % 100 === 0) {
      alice.sendChat(`alice soak tick=${tick}`);
      bob.sendChat(`bob soak tick=${tick}`);
    }

    tick++;
    await sleep(100);
  }

  console.log(`soak:staging: soak loop complete after ${tick} ticks; closing clients`);

  await alice.close();
  await bob.close();

  // OpenObserve best-effort ingest check (DEP-06)
  const obsHits = await checkObsIngest();

  // Compute p95 tick latency
  const p95Latency = p95(stats.latencies);
  const sampleCount = stats.latencies.length;

  console.log(
    `soak:staging: results — disconnects=${stats.disconnects}, rate_limited=${stats.rateLimited}, ` +
      `p95=${p95Latency}ms, samples=${sampleCount}, obs_traces=${obsHits ?? 'skipped'}`,
  );

  // Assertions
  let failed = false;

  if (stats.disconnects > 0) {
    console.error(
      `soak:staging: FAIL — ${stats.disconnects} unexpected WS disconnects detected (DEP-08 violation: Fly idle-timeout vs Colyseus pingInterval unresolved)`,
    );
    failed = true;
  }

  if (stats.rateLimited > 0) {
    console.error(
      `soak:staging: FAIL — ${stats.rateLimited} RATE_LIMITED events received during baseline 10Hz/0.1Hz cadence (server rate-limit budget drift)`,
    );
    failed = true;
  }

  if (p95Latency > 100) {
    console.error(
      `soak:staging: FAIL — p95 tick latency ${p95Latency}ms exceeds 100ms hard cap (hard gate)`,
    );
    failed = true;
  } else if (p95Latency > 25) {
    console.warn(
      `soak:staging: WARN — p95 tick latency ${p95Latency}ms exceeds 25ms soft cap ` +
        `(acceptable for CI runner network jitter; record in 05-HUMAN-UAT.md if this is a CI-scheduled run)`,
    );
  }

  if (failed) {
    process.exit(1);
  }

  console.log(
    `soak:staging: PASS — duration=${SOAK_DURATION_MINUTES}m, p95=${p95Latency}ms, disconnects=0, rate_limited=0`,
  );
  process.exit(0);
}

// ---------------------------------------------------------------------------
// isMain guard — allows import by tests later if needed
// ---------------------------------------------------------------------------
const isMain =
  process.argv[1] &&
  (process.argv[1] === fileURLToPath(import.meta.url) ||
    process.argv[1].endsWith('soak-staging.mjs'));

if (isMain) {
  main().catch((err) => {
    console.error('soak:staging: FATAL ERROR', err);
    process.exit(1);
  });
}