#!/usr/bin/env node
/**
 * EventSnap load / stress driver.
 *
 * Simulates ~100 guests joining an event and uploading ~1000 images in bursts
 * (10-20 at a time) spread across a compressed time window, plus a pool of
 * viewers holding live SSE connections (like phones with the feed open) and one
 * "diashow" connection (the showcase display). Measures the metrics that matter
 * for a live event — especially the *pipeline backlog*: how long between an
 * upload succeeding and its preview being ready (SSE `upload-processed`), which
 * is exactly "how long until the photo shows up on the diashow".
 *
 * This is an HTTP-level driver on purpose: 100 real browsers would bottleneck
 * the test box, not the server. One real browser watches /diashow separately
 * (see diashow-watch.mjs).
 *
 * METHODOLOGY NOTE — what we change vs. the shipping config:
 *   We ONLY disable rate limits. They are per-IP/per-user anti-abuse guards; a
 *   synthetic test from one IP would trip them in a way real guests (distinct
 *   IPs, phones) never would, so leaving them on would measure the rate limiter
 *   instead of the pipeline. We DELIBERATELY leave compression concurrency,
 *   DB pool size and quotas at their real defaults — validating those is the
 *   whole point. (Runtime can't change compression concurrency anyway; it's a
 *   boot-time env var.)
 *
 *   Related real-world finding to keep in mind: the default upload_rate_per_hour
 *   is 10, so a real guest uploading a burst of 10-20 would be throttled by the
 *   SHIPPING config too. That's a genuine event-day issue worth its own report,
 *   independent of this pipeline test.
 *
 * Usage:
 *   node driver.mjs                      # full run (100 guests / 1000 imgs / 15 min)
 *   LT_GUESTS=5 LT_IMAGES=50 LT_WINDOW_SEC=60 node driver.mjs   # smoke
 *
 * Env knobs (all optional; defaults target the full run):
 *   LT_BASE            http://localhost:3101   frontend/caddy base URL
 *   LT_GUESTS          100                     number of virtual guests
 *   LT_IMAGES          1000                    total uploads to perform
 *   LT_WINDOW_SEC      900                     spread uploads over this window
 *   LT_BURST_MIN/MAX   10 / 20                 images per burst
 *   LT_BURST_CONC      3                       parallel uploads within one burst (phone-like)
 *   LT_VIEWERS         20                      extra guests holding SSE + polling feed
 *   LT_PHOTOS_DIR      /tmp/eventsnap-loadtest/photos
 *   LT_TRUNCATE        1                       wipe event data before the run
 *   LT_DRAIN_TIMEOUT_SEC 600                   max wait for compression backlog to drain
 *   LT_APP_CONTAINER   e2e-app-1               docker container for CPU/mem sampling
 *   LT_DB_CONTAINER    e2e-db-1                docker container for psql ground-truth
 *   LT_ADMIN_PW        admin-test-pw
 *   LT_KEEP_RATELIMITS 0                       set 1 to NOT disable rate limits
 *   LT_OUT_DIR         ./results
 */

import { readFileSync, readdirSync, writeFileSync, mkdirSync, statSync } from 'node:fs';
import { execFile } from 'node:child_process';
import { promisify } from 'node:util';
import { fileURLToPath } from 'node:url';
import { dirname, join } from 'node:path';

const execFileAsync = promisify(execFile);
const __dirname = dirname(fileURLToPath(import.meta.url));

// ── Config ──────────────────────────────────────────────────────────────────
const cfg = {
  base: process.env.LT_BASE ?? 'http://localhost:3101',
  guests: int('LT_GUESTS', 100),
  images: int('LT_IMAGES', 1000),
  windowSec: int('LT_WINDOW_SEC', 900),
  burstMin: int('LT_BURST_MIN', 10),
  burstMax: int('LT_BURST_MAX', 20),
  burstConc: int('LT_BURST_CONC', 3),
  viewers: int('LT_VIEWERS', 20),
  photosDir: process.env.LT_PHOTOS_DIR ?? '/tmp/eventsnap-loadtest/photos',
  truncate: process.env.LT_TRUNCATE !== '0',
  drainTimeoutSec: int('LT_DRAIN_TIMEOUT_SEC', 600),
  appContainer: process.env.LT_APP_CONTAINER ?? 'e2e-app-1',
  dbContainer: process.env.LT_DB_CONTAINER ?? 'e2e-db-1',
  adminPw: process.env.LT_ADMIN_PW ?? 'admin-test-pw',
  keepRateLimits: process.env.LT_KEEP_RATELIMITS === '1',
  outDir: process.env.LT_OUT_DIR ?? join(__dirname, 'results'),
};

function int(name, def) {
  const v = process.env[name];
  return v === undefined ? def : parseInt(v, 10);
}

const API = `${cfg.base}/api/v1`;
const now = () => Date.now();
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
const rand = (min, max) => min + Math.floor(Math.random() * (max - min + 1));
const pick = (arr) => arr[Math.floor(Math.random() * arr.length)];

// ── HTTP helpers ──────────────────────────────────────────────────────────────
async function api(path, { method = 'GET', token, json, expect } = {}) {
  const headers = {};
  if (token) headers.Authorization = `Bearer ${token}`;
  if (json !== undefined) headers['Content-Type'] = 'application/json';
  const res = await fetch(`${API}${path}`, {
    method,
    headers,
    body: json !== undefined ? JSON.stringify(json) : undefined,
  });
  let body;
  if (res.status !== 204) {
    const text = await res.text();
    try {
      body = text.length ? JSON.parse(text) : undefined;
    } catch {
      body = text;
    }
  }
  if (expect && !expect.includes(res.status)) {
    throw new Error(`${method} ${path} → ${res.status}: ${JSON.stringify(body)}`);
  }
  return { status: res.status, body };
}

const adminLogin = () =>
  api('/admin/login', { method: 'POST', json: { password: cfg.adminPw }, expect: [200] }).then(
    (r) => r.body.jwt
  );

const joinGuest = (name) =>
  api('/join', { method: 'POST', json: { display_name: name }, expect: [201] }).then((r) => r.body);

const patchConfig = (adminJwt, patch) =>
  api('/admin/config', { method: 'PATCH', token: adminJwt, json: patch, expect: [204] });

const getConfig = (adminJwt) => api('/admin/config', { token: adminJwt }).then((r) => r.body);

const truncate = (adminJwt) =>
  api('/admin/__truncate', { method: 'POST', token: adminJwt, expect: [204] });

const CAPTIONS = [
  'Was für ein magischer Tag 💍', 'Der erste Tanz 🕺', 'Prost! 🥂', 'Die Torte 🍰',
  'Feuerwerk 🎆', 'Beste Freunde 💕', 'Was für eine Stimmung! 🎉', 'Details 🌸',
  'Sonnenuntergang 🌅', 'Tanzfläche brennt 🔥', null, null, null,
];
const TAGS = ['hochzeit', 'liebe', 'party', 'tanzen', 'natur', 'feier', 'freunde', 'dessert'];

async function uploadPhoto(jwt, file, buf) {
  const form = new FormData();
  form.append('file', new Blob([buf], { type: 'image/jpeg' }), 'photo.jpg');
  const cap = pick(CAPTIONS);
  if (cap) form.append('caption', cap);
  form.append('hashtags', `${pick(TAGS)},${pick(TAGS)}`);
  const t0 = now();
  const res = await fetch(`${API}/upload`, {
    method: 'POST',
    headers: { Authorization: `Bearer ${jwt}` },
    body: form,
  });
  const t1 = now();
  let id, errText;
  if (res.status === 201) {
    id = (await res.json()).id;
  } else {
    errText = (await res.text()).slice(0, 200);
  }
  return { status: res.status, id, ms: t1 - t0, endTs: t1, bytes: buf.length, errText };
}

// ── SSE client (auto-reconnecting) ────────────────────────────────────────────
class SseClient {
  constructor(jwt, label, onEvent) {
    this.jwt = jwt;
    this.label = label;
    this.onEvent = onEvent;
    this.stop = false;
    this.reconnects = 0;
    this.resyncs = 0;
    this.controller = null;
  }
  start() {
    this._loop();
    return this;
  }
  async _loop() {
    while (!this.stop) {
      try {
        const tkt = await api('/stream/ticket', { method: 'POST', token: this.jwt, expect: [200] });
        this.controller = new AbortController();
        const res = await fetch(`${API}/stream?ticket=${tkt.body.ticket}`, {
          headers: { Accept: 'text/event-stream' },
          signal: this.controller.signal,
        });
        if (!res.ok || !res.body) throw new Error(`stream ${res.status}`);
        const reader = res.body.getReader();
        const dec = new TextDecoder();
        let buf = '';
        while (!this.stop) {
          const { value, done } = await reader.read();
          if (done) break;
          buf += dec.decode(value, { stream: true });
          let idx;
          while ((idx = buf.indexOf('\n\n')) !== -1) {
            const frame = buf.slice(0, idx);
            buf = buf.slice(idx + 2);
            this._parseFrame(frame);
          }
        }
      } catch (e) {
        if (this.stop) return;
        this.reconnects++;
        await sleep(500 + Math.random() * 500);
      }
    }
  }
  _parseFrame(frame) {
    let event = 'message';
    let data = '';
    for (const line of frame.split('\n')) {
      if (line.startsWith('event:')) event = line.slice(6).trim();
      else if (line.startsWith('data:')) data += line.slice(5).trim();
    }
    if (event === 'resync') this.resyncs++;
    if (!data) return;
    let payload;
    try {
      payload = JSON.parse(data);
    } catch {
      payload = data;
    }
    this.onEvent(event, payload, this.label);
  }
  close() {
    this.stop = true;
    try {
      this.controller?.abort();
    } catch {
      /* noop */
    }
  }
}

// ── Resource sampler (docker stats + psql ground truth) ───────────────────────
async function sampleResources() {
  const out = { ts: now() };
  try {
    const { stdout } = await execFileAsync('docker', [
      'stats', '--no-stream', '--format',
      '{{.Name}}|{{.CPUPerc}}|{{.MemUsage}}',
      cfg.appContainer, cfg.dbContainer,
    ]);
    out.docker = stdout.trim();
  } catch (e) {
    out.dockerErr = String(e).slice(0, 120);
  }
  try {
    const { stdout } = await psql(
      `select count(*) from pg_stat_activity where datname='eventsnap_test'`
    );
    out.dbConns = parseInt(stdout.trim(), 10);
  } catch {
    /* ignore */
  }
  return out;
}

function psql(sql) {
  return execFileAsync('docker', [
    'exec', cfg.dbContainer, 'psql', '-U', 'eventsnap_test', '-d', 'eventsnap_test',
    '-tAc', sql,
  ]);
}

async function compressionCounts() {
  try {
    const { stdout } = await psql(
      `select compression_status, count(*) from upload where deleted_at is null group by 1`
    );
    const map = {};
    for (const line of stdout.trim().split('\n')) {
      if (!line) continue;
      const [status, n] = line.split('|');
      map[status] = parseInt(n, 10);
    }
    return map;
  } catch (e) {
    return { error: String(e).slice(0, 120) };
  }
}

// ── Stats helpers ─────────────────────────────────────────────────────────────
function pct(sorted, p) {
  if (!sorted.length) return null;
  const i = Math.min(sorted.length - 1, Math.floor((p / 100) * sorted.length));
  return sorted[i];
}
function summarize(nums) {
  if (!nums.length) return null;
  const s = [...nums].sort((a, b) => a - b);
  const sum = s.reduce((a, b) => a + b, 0);
  return {
    n: s.length, min: s[0], max: s[s.length - 1], mean: Math.round(sum / s.length),
    p50: pct(s, 50), p95: pct(s, 95), p99: pct(s, 99),
  };
}

// ── Image pool ────────────────────────────────────────────────────────────────
function loadPool() {
  let files;
  try {
    files = readdirSync(cfg.photosDir).filter((f) => f.endsWith('.jpg'));
  } catch {
    files = [];
  }
  if (!files.length) {
    console.error(
      `\n✗ No images in ${cfg.photosDir}. Run first:\n` +
        `    LT_PHOTOS_DIR=${cfg.photosDir} python3 ${join(__dirname, 'gen-images.py')}\n`
    );
    process.exit(1);
  }
  const pool = files.map((f) => {
    const p = join(cfg.photosDir, f);
    return { buf: readFileSync(p), size: statSync(p).size, name: f };
  });
  return pool;
}

// ── Burst schedule ────────────────────────────────────────────────────────────
// Build bursts until we reach the image target, assign each to a random guest at
// a random time in the window. Naturally gives some guests several bursts and
// some none — like a real event. Guests all join before their first burst.
function buildSchedule() {
  const bursts = [];
  let remaining = cfg.images;
  while (remaining > 0) {
    const size = Math.min(remaining, rand(cfg.burstMin, cfg.burstMax));
    bursts.push({
      guest: rand(0, cfg.guests - 1),
      atMs: Math.floor(Math.random() * cfg.windowSec * 1000),
      size,
    });
    remaining -= size;
  }
  bursts.sort((a, b) => a.atMs - b.atMs);
  return bursts;
}

// ── Main ──────────────────────────────────────────────────────────────────────
async function main() {
  console.log('━'.repeat(72));
  console.log('EventSnap load driver');
  console.log(
    `  target: ${cfg.guests} guests, ${cfg.images} images, ` +
      `bursts ${cfg.burstMin}-${cfg.burstMax}, window ${cfg.windowSec}s, ${cfg.viewers} viewers`
  );
  console.log(`  base: ${cfg.base}`);
  console.log('━'.repeat(72));

  const pool = loadPool();
  const avgMb = pool.reduce((a, p) => a + p.size, 0) / pool.length / 1e6;
  console.log(
    `[pool] ${pool.length} images, avg ${avgMb.toFixed(2)} MB, ` +
      `projected originals ~${((avgMb * cfg.images) / 1000).toFixed(1)} GB`
  );

  // Preflight: disk headroom
  try {
    const { stdout } = await execFileAsync('df', ['-BG', '--output=avail', '/']);
    console.log(`[preflight] disk avail:${stdout.trim().split('\n').pop().trim()}`);
  } catch {
    /* ignore */
  }

  // Admin: reset + config
  const admin = await adminLogin();
  const cfgBefore = await getConfig(admin);
  if (cfg.truncate) {
    console.log('[setup] truncating event data…');
    await truncate(admin);
  }
  const admin2 = await adminLogin();
  if (!cfg.keepRateLimits) {
    console.log('[setup] disabling rate limits (methodology: isolate pipeline, not limiter)');
    await patchConfig(admin2, {
      rate_limits_enabled: 'false',
      upload_rate_enabled: 'false',
      feed_rate_enabled: 'false',
      join_rate_enabled: 'false',
      recover_rate_enabled: 'false',
    });
  }
  console.log(
    `[setup] compression concurrency / quotas left at SHIPPING defaults ` +
      `(compression_worker_concurrency is a boot env var, not runtime)`
  );

  // Metrics stores
  const uploads = []; // {status, ms, endTs, bytes, id, guest}
  const processedAt = new Map(); // upload_id -> ts of upload-processed
  const uploadEndTs = new Map(); // upload_id -> endTs
  const errorEvents = []; // upload-error payloads
  const resources = [];
  let newUploadEvents = 0;

  // Join guests (staggered, small concurrency)
  console.log(`[join] joining ${cfg.guests} guests…`);
  const accounts = new Array(cfg.guests);
  const joinConc = 10;
  for (let i = 0; i < cfg.guests; i += joinConc) {
    await Promise.all(
      Array.from({ length: Math.min(joinConc, cfg.guests - i) }, (_, k) =>
        joinGuest(`LoadGuest ${i + k}`).then((a) => (accounts[i + k] = a))
      )
    );
  }
  console.log('[join] done');

  // Live connections: 1 diashow + N viewers
  const onEvent = (event, payload) => {
    if (event === 'new-upload') newUploadEvents++;
    else if (event === 'upload-processed' && payload?.upload_id) {
      if (!processedAt.has(payload.upload_id)) processedAt.set(payload.upload_id, now());
    } else if (event === 'upload-error' && payload?.upload_id) {
      errorEvents.push(payload);
    }
  };
  const sseClients = [];
  sseClients.push(new SseClient(accounts[0].jwt, 'diashow', onEvent).start());
  for (let i = 0; i < Math.min(cfg.viewers, cfg.guests); i++) {
    sseClients.push(new SseClient(accounts[i].jwt, `viewer${i}`, onEvent).start());
  }
  console.log(`[sse] opened ${sseClients.length} live connections (1 diashow + viewers)`);

  // Viewers also poll the feed periodically (viewing load)
  let feedPolls = 0;
  const feedPoller = setInterval(async () => {
    const g = accounts[rand(0, Math.min(cfg.viewers, cfg.guests) - 1)];
    try {
      await api('/feed', { token: g.jwt });
      feedPolls++;
    } catch {
      /* ignore */
    }
  }, 1500);

  // Resource sampler
  const sampler = setInterval(async () => resources.push(await sampleResources()), 5000);
  resources.push(await sampleResources());

  // Run the burst schedule
  const schedule = buildSchedule();
  console.log(
    `[run] ${schedule.length} bursts scheduled over ${cfg.windowSec}s ` +
      `(≈${(cfg.images / (cfg.windowSec / 60)).toFixed(0)} img/min avg). Starting…`
  );
  const startTs = now();
  let done = 0;

  const runBurst = async (burst) => {
    const jwt = accounts[burst.guest].jwt;
    const items = Array.from({ length: burst.size }, () => pick(pool));
    // upload with phone-like small concurrency inside the burst
    for (let i = 0; i < items.length; i += cfg.burstConc) {
      const chunk = items.slice(i, i + cfg.burstConc);
      const results = await Promise.all(chunk.map((it) => uploadPhoto(jwt, it.name, it.buf)));
      for (const r of results) {
        uploads.push({ ...r, guest: burst.guest });
        if (r.id) uploadEndTs.set(r.id, r.endTs);
        done++;
      }
    }
  };

  const timers = [];
  const burstPromises = [];
  for (const burst of schedule) {
    timers.push(
      setTimeout(() => {
        burstPromises.push(runBurst(burst));
      }, burst.atMs)
    );
  }

  // progress ticker
  const ticker = setInterval(() => {
    const elapsed = ((now() - startTs) / 1000).toFixed(0);
    const ok = uploads.filter((u) => u.status === 201).length;
    console.log(
      `[t+${elapsed}s] uploads ${done}/${cfg.images} (ok ${ok}), ` +
        `processed ${processedAt.size}, new-upload evts ${newUploadEvents}, ` +
        `feedPolls ${feedPolls}`
    );
  }, 10000);

  // wait for the window to elapse, then for all scheduled bursts to finish
  await sleep(cfg.windowSec * 1000 + 500);
  await Promise.all(burstPromises);
  clearInterval(ticker);
  console.log(`[run] all bursts issued. uploaded ${done}, ok ${uploads.filter((u) => u.status === 201).length}`);

  // Drain: wait for compression backlog to clear (SSE processed ⊇ successful ids)
  console.log('[drain] waiting for compression backlog to clear…');
  const drainStart = now();
  const successIds = new Set(uploads.filter((u) => u.id).map((u) => u.id));
  let lastLog = 0;
  let drainReason = 'timeout';
  while (now() - drainStart < cfg.drainTimeoutSec * 1000) {
    // SSE view (what the diashow actually "sees")
    const pendingSse = [...successIds].filter((id) => !processedAt.has(id)).length;
    if (pendingSse === 0) {
      console.log('[drain] backlog cleared (all upload-processed events received)');
      drainReason = 'sse-complete';
      break;
    }
    // DB ground truth — authoritative even if an SSE reconnect dropped events
    const counts = await compressionCounts();
    const dbPending = Object.entries(counts)
      .filter(([s]) => s !== 'done' && s !== 'error')
      .reduce((a, [, n]) => a + n, 0);
    if (!counts.error && dbPending === 0) {
      console.log(`[drain] backlog cleared per DB (db status: ${JSON.stringify(counts)})`);
      drainReason = 'db-complete';
      break;
    }
    if (now() - lastLog > 5000) {
      console.log(
        `[drain] pending(sse) ${pendingSse}, pending(db) ${dbPending} — db: ${JSON.stringify(counts)}`
      );
      lastLog = now();
    }
    await sleep(1000);
  }
  const drainMs = now() - drainStart;

  // Stop background work
  clearInterval(sampler);
  clearInterval(feedPoller);
  timers.forEach(clearTimeout);
  resources.push(await sampleResources());
  const finalCounts = await compressionCounts();
  await sleep(200);
  sseClients.forEach((c) => c.close());

  // ── Build report ────────────────────────────────────────────────────────────
  const byStatus = {};
  for (const u of uploads) byStatus[u.status] = (byStatus[u.status] ?? 0) + 1;
  const okUploads = uploads.filter((u) => u.status === 201);
  const uploadLatency = summarize(okUploads.map((u) => u.ms));
  const pipelineLatency = summarize(
    [...processedAt.entries()]
      .filter(([id]) => uploadEndTs.has(id))
      .map(([id, ts]) => ts - uploadEndTs.get(id))
  );
  const totalReconnects = sseClients.reduce((a, c) => a + c.reconnects, 0);
  const totalResyncs = sseClients.reduce((a, c) => a + c.resyncs, 0);

  const report = {
    config: cfg,
    startedAt: new Date(startTs).toISOString(),
    durationSec: Math.round((now() - startTs) / 1000),
    totals: {
      uploadsAttempted: uploads.length,
      uploadsOk: okUploads.length,
      byStatus,
      newUploadEvents,
      processedEvents: processedAt.size,
      uploadErrorEvents: errorEvents.length,
      feedPolls,
    },
    uploadLatencyMs: uploadLatency,
    pipelineLatencyMs: pipelineLatency,
    drain: {
      ms: drainMs,
      cleared: drainReason !== 'timeout',
      reason: drainReason,
      ssePending: [...successIds].filter((id) => !processedAt.has(id)).length,
      finalDbCounts: finalCounts,
    },
    sse: {
      connections: sseClients.length,
      reconnects: totalReconnects,
      resyncs: totalResyncs,
    },
    resources,
    rateLimitsDisabled: !cfg.keepRateLimits,
    configBefore: cfgBefore,
  };

  mkdirSync(cfg.outDir, { recursive: true });
  const stamp = new Date(startTs).toISOString().replace(/[:.]/g, '-');
  const outPath = join(cfg.outDir, `run-${stamp}.json`);
  writeFileSync(outPath, JSON.stringify(report, null, 2));

  // ── Print verdict ─────────────────────────────────────────────────────────
  console.log('\n' + '━'.repeat(72));
  console.log('RESULTS');
  console.log('━'.repeat(72));
  console.log(`uploads: ${okUploads.length}/${uploads.length} ok — byStatus ${JSON.stringify(byStatus)}`);
  console.log(`upload latency ms: ${JSON.stringify(uploadLatency)}`);
  console.log(`pipeline latency ms (upload→preview ready): ${JSON.stringify(pipelineLatency)}`);
  console.log(`backlog drain: ${(drainMs / 1000).toFixed(1)}s, cleared=${report.drain.cleared}`);
  console.log(`final db compression status: ${JSON.stringify(finalCounts)}`);
  console.log(`sse: ${sseClients.length} conns, ${totalReconnects} reconnects, ${totalResyncs} resyncs`);
  console.log(`\nfull report → ${outPath}`);

  // Heuristic pass/fail flags (validate shipping config)
  const flags = [];
  const err5xx = Object.entries(byStatus).filter(([s]) => +s >= 500).reduce((a, [, n]) => a + n, 0);
  if (err5xx > 0) flags.push(`✗ ${err5xx} server errors (5xx)`);
  if (byStatus['507']) flags.push(`✗ ${byStatus['507']} quota rejections (507)`);
  if (byStatus['413']) flags.push(`⚠ ${byStatus['413']} too-large (413)`);
  if (byStatus['429']) flags.push(`⚠ ${byStatus['429']} rate-limited (429) — unexpected with limits off`);
  if (!report.drain.cleared) flags.push(`✗ backlog did NOT drain within ${cfg.drainTimeoutSec}s`);
  if (totalResyncs > sseClients.length) flags.push(`⚠ ${totalResyncs} SSE resyncs (consumer lag)`);
  if (pipelineLatency && pipelineLatency.p95 > 60000)
    flags.push(`⚠ pipeline p95 ${(pipelineLatency.p95 / 1000).toFixed(0)}s (diashow lags >1min)`);
  console.log('\nflags:');
  if (flags.length) flags.forEach((f) => console.log('  ' + f));
  else console.log('  ✓ no red flags — default config handled the load');
  console.log('━'.repeat(72));
}

main().catch((e) => {
  console.error('\n✗ driver failed:', e);
  process.exit(1);
});