//! PERF-01 duplicate-heavy replay for `/gateway/insert`.
//!
//! Runs 25/50/100 worker scenarios (configurable), captures p50/p95/p99 latency,
//! status/code distribution, and metrics deltas for duplicate and admission/fallback labels.
//!
//! Example:
//! cargo run --example perf_insert_duplicate_replay -- \
//!   --base-url http://localhost:4052 \
//!   --client athena_logging \
//!   --table cz_companies \
//!   --duplicate-key registration_number \
//!   --duplicate-value PERF-01-DUP-001 \
//!   --seed-before-each-scenario \
//!   --label after

use reqwest::Client;
use serde_json::{Value, json};
use std::collections::BTreeMap;
use std::env;
use std::error::Error;
use std::fs;
use std::path::PathBuf;
use std::time::Instant;
use tokio::task::JoinSet;

#[derive(Clone)]
struct ReplayConfig {
    base_url: String,
    insert_path: String,
    metrics_path: String,
    client_name: String,
    payload: Value,
    workers: Vec<usize>,
    requests_per_worker: usize,
    timeout_secs: f64,
    seed_before_each_scenario: bool,
    label: String,
    output_path: Option<PathBuf>,
}

#[derive(Debug, Clone)]
struct HttpResult {
    latency_ms: f64,
    status: u16,
    code: String,
}

#[derive(Debug, Clone)]
struct ScenarioReport {
    workers: usize,
    total_requests: usize,
    elapsed_s: f64,
    p50_ms: f64,
    p95_ms: f64,
    p99_ms: f64,
    mean_ms: f64,
    min_ms: f64,
    max_ms: f64,
    status_counts: BTreeMap<u16, usize>,
    code_counts: BTreeMap<String, usize>,
    metric_deltas: BTreeMap<String, f64>,
}

#[derive(Debug, Clone)]
struct MetricSample {
    name: String,
    labels: BTreeMap<String, String>,
    value: f64,
}

fn usage() -> String {
    [
        "Usage:",
        "  cargo run --example perf_insert_duplicate_replay -- --client <X-Athena-Client> [options]",
        "",
        "Required:",
        "  --client <name>",
        "",
        "Payload options:",
        "  --body-json <json>            Full /gateway/insert payload",
        "  --body-file <path>            Full /gateway/insert payload JSON file",
        "  --table <name>                Used when body-json/body-file are omitted",
        "  --duplicate-key <column>      Default: registration_number",
        "  --duplicate-value <value>     Default: PERF-01-DUPLICATE",
        "",
        "Scenario options:",
        "  --workers <csv>               Default: 25,50,100",
        "  --requests-per-worker <n>     Default: 20",
        "  --request-timeout-seconds <f> Default: 15",
        "  --seed-before-each-scenario   Send one seed insert before each scenario",
        "",
        "Endpoint options:",
        "  --base-url <url>              Default: http://localhost:4052",
        "  --insert-path <path>          Default: /gateway/insert",
        "  --metrics-path <path>         Default: /metrics",
        "",
        "Report options:",
        "  --label <name>                Default: current",
        "  --output <path>               Default: perf-01-report-<label>-<timestamp>.md",
    ]
    .join("\n")
}

fn parse_workers(raw: &str) -> Result<Vec<usize>, Box<dyn Error>> {
    let mut workers = Vec::new();
    for part in raw.split(',') {
        let trimmed = part.trim();
        if trimmed.is_empty() {
            continue;
        }
        let value: usize = trimmed.parse()?;
        if value == 0 {
            return Err("worker count must be > 0".into());
        }
        workers.push(value);
    }
    if workers.is_empty() {
        return Err("at least one worker count is required".into());
    }
    Ok(workers)
}

fn parse_payload(
    body_json: Option<String>,
    body_file: Option<PathBuf>,
    table: Option<String>,
    duplicate_key: &str,
    duplicate_value: &str,
) -> Result<Value, Box<dyn Error>> {
    let payload = match (body_json, body_file) {
        (Some(_), Some(_)) => return Err("provide only one of --body-json or --body-file".into()),
        (Some(raw), None) => serde_json::from_str::<Value>(&raw)?,
        (None, Some(path)) => {
            let text = fs::read_to_string(path)?;
            serde_json::from_str::<Value>(&text)?
        }
        (None, None) => {
            let table_name =
                table.ok_or("--table is required when --body-json/--body-file are not provided")?;
            json!({
                "table_name": table_name,
                "insert_body": {
                    duplicate_key: duplicate_value
                }
            })
        }
    };

    if payload.get("table_name").is_none() || payload.get("insert_body").is_none() {
        return Err("insert payload must contain table_name and insert_body".into());
    }

    Ok(payload)
}

fn parse_args() -> Result<ReplayConfig, Box<dyn Error>> {
    let args: Vec<String> = env::args().skip(1).collect();

    if args.iter().any(|arg| arg == "-h" || arg == "--help") {
        println!("{}", usage());
        std::process::exit(0);
    }

    let mut base_url = "http://localhost:4052".to_string();
    let mut insert_path = "/gateway/insert".to_string();
    let mut metrics_path = "/metrics".to_string();
    let mut client_name: Option<String> = None;

    let mut table: Option<String> = None;
    let mut duplicate_key = "registration_number".to_string();
    let mut duplicate_value = "PERF-01-DUPLICATE".to_string();
    let mut body_json: Option<String> = None;
    let mut body_file: Option<PathBuf> = None;

    let mut workers_raw = "25,50,100".to_string();
    let mut requests_per_worker: usize = 20;
    let mut timeout_secs: f64 = 15.0;
    let mut seed_before_each_scenario = false;

    let mut label = "current".to_string();
    let mut output_path: Option<PathBuf> = None;

    let mut i = 0usize;
    while i < args.len() {
        let key = args[i].as_str();
        match key {
            "--base-url" => {
                i += 1;
                base_url = args.get(i).ok_or("missing value for --base-url")?.clone();
            }
            "--insert-path" => {
                i += 1;
                insert_path = args
                    .get(i)
                    .ok_or("missing value for --insert-path")?
                    .clone();
            }
            "--metrics-path" => {
                i += 1;
                metrics_path = args
                    .get(i)
                    .ok_or("missing value for --metrics-path")?
                    .clone();
            }
            "--client" => {
                i += 1;
                client_name = Some(args.get(i).ok_or("missing value for --client")?.clone());
            }
            "--table" => {
                i += 1;
                table = Some(args.get(i).ok_or("missing value for --table")?.clone());
            }
            "--duplicate-key" => {
                i += 1;
                duplicate_key = args
                    .get(i)
                    .ok_or("missing value for --duplicate-key")?
                    .clone();
            }
            "--duplicate-value" => {
                i += 1;
                duplicate_value = args
                    .get(i)
                    .ok_or("missing value for --duplicate-value")?
                    .clone();
            }
            "--body-json" => {
                i += 1;
                body_json = Some(args.get(i).ok_or("missing value for --body-json")?.clone());
            }
            "--body-file" => {
                i += 1;
                body_file = Some(PathBuf::from(
                    args.get(i).ok_or("missing value for --body-file")?,
                ));
            }
            "--workers" => {
                i += 1;
                workers_raw = args.get(i).ok_or("missing value for --workers")?.clone();
            }
            "--requests-per-worker" => {
                i += 1;
                requests_per_worker = args
                    .get(i)
                    .ok_or("missing value for --requests-per-worker")?
                    .parse()?;
            }
            "--request-timeout-seconds" => {
                i += 1;
                timeout_secs = args
                    .get(i)
                    .ok_or("missing value for --request-timeout-seconds")?
                    .parse()?;
            }
            "--seed-before-each-scenario" => {
                seed_before_each_scenario = true;
            }
            "--label" => {
                i += 1;
                label = args.get(i).ok_or("missing value for --label")?.clone();
            }
            "--output" | "-o" => {
                i += 1;
                output_path = Some(PathBuf::from(
                    args.get(i).ok_or("missing value for --output")?,
                ));
            }
            other => {
                return Err(format!("unknown argument: {}\n\n{}", other, usage()).into());
            }
        }
        i += 1;
    }

    let client_name = client_name.ok_or("--client is required")?;
    let workers = parse_workers(&workers_raw)?;
    let payload = parse_payload(
        body_json,
        body_file,
        table,
        &duplicate_key,
        &duplicate_value,
    )?;

    Ok(ReplayConfig {
        base_url,
        insert_path,
        metrics_path,
        client_name,
        payload,
        workers,
        requests_per_worker,
        timeout_secs,
        seed_before_each_scenario,
        label,
        output_path,
    })
}

fn make_url(base_url: &str, path: &str) -> String {
    format!(
        "{}/{}",
        base_url.trim_end_matches('/'),
        path.trim_start_matches('/')
    )
}

fn percentile(latencies: &[f64], p: f64) -> f64 {
    if latencies.is_empty() {
        return 0.0;
    }
    if p <= 0.0 {
        return latencies
            .iter()
            .fold(f64::INFINITY, |acc, value| acc.min(*value));
    }
    if p >= 100.0 {
        return latencies
            .iter()
            .fold(f64::NEG_INFINITY, |acc, value| acc.max(*value));
    }

    let mut sorted = latencies.to_vec();
    sorted.sort_by(|a, b| a.total_cmp(b));
    let rank = ((p / 100.0) * sorted.len() as f64).ceil() as usize;
    let idx = rank.saturating_sub(1).min(sorted.len() - 1);
    sorted[idx]
}

fn parse_metric_line(line: &str) -> Option<MetricSample> {
    let trimmed = line.trim();
    if trimmed.is_empty() || trimmed.starts_with('#') {
        return None;
    }

    let (left, right) = trimmed.rsplit_once(' ')?;
    let value: f64 = right.parse().ok()?;

    if let Some((name, labels_blob)) = left.split_once('{') {
        let labels_raw = labels_blob.strip_suffix('}')?;
        let mut labels = BTreeMap::new();
        if !labels_raw.trim().is_empty() {
            for chunk in labels_raw.split(',') {
                let (k, v) = match chunk.split_once('=') {
                    Some(pair) => pair,
                    None => continue,
                };
                let key = k.trim().to_string();
                let mut value_text = v.trim().to_string();
                if value_text.starts_with('"') && value_text.ends_with('"') && value_text.len() >= 2
                {
                    value_text = value_text[1..value_text.len() - 1].to_string();
                }
                labels.insert(key, value_text.replace("\\\"", "\""));
            }
        }

        Some(MetricSample {
            name: name.to_string(),
            labels,
            value,
        })
    } else {
        Some(MetricSample {
            name: left.to_string(),
            labels: BTreeMap::new(),
            value,
        })
    }
}

fn metric_sum(samples: &[MetricSample], name: &str, expected_labels: &[(&str, &str)]) -> f64 {
    let mut total = 0.0f64;
    for sample in samples {
        if sample.name != name {
            continue;
        }

        let mut labels_match = true;
        for (key, value) in expected_labels {
            if sample.labels.get(*key).map(String::as_str) != Some(*value) {
                labels_match = false;
                break;
            }
        }

        if labels_match {
            total += sample.value;
        }
    }
    total
}

fn collect_perf_metrics(samples: &[MetricSample]) -> BTreeMap<String, f64> {
    let mut out = BTreeMap::new();
    out.insert(
        "db_unique_violation".to_string(),
        metric_sum(
            samples,
            "athena_gateway_insert_window_events_total",
            &[("label", "db_unique_violation")],
        ),
    );
    out.insert(
        "admission_defer_high_watermark".to_string(),
        metric_sum(
            samples,
            "athena_gateway_insert_window_events_total",
            &[("label", "admission_defer_high_watermark")],
        ),
    );
    out.insert(
        "admission_reject_queue_full".to_string(),
        metric_sum(
            samples,
            "athena_gateway_insert_window_events_total",
            &[("label", "admission_reject_queue_full")],
        ),
    );
    out.insert(
        "admission_reject_enqueue_failed".to_string(),
        metric_sum(
            samples,
            "athena_gateway_insert_window_events_total",
            &[("label", "admission_reject_enqueue_failed")],
        ),
    );
    out.insert(
        "window_wait_timeout".to_string(),
        metric_sum(
            samples,
            "athena_gateway_insert_window_events_total",
            &[("label", "window_wait_timeout")],
        ),
    );
    out.insert(
        "db_timeout".to_string(),
        metric_sum(
            samples,
            "athena_gateway_insert_window_events_total",
            &[("label", "db_timeout")],
        ),
    );
    out.insert(
        "bulk_fallback_sql_error".to_string(),
        metric_sum(
            samples,
            "athena_gateway_insert_window_events_total",
            &[("label", "bulk_fallback_sql_error")],
        ),
    );
    out.insert(
        "deferred_insert_queued".to_string(),
        metric_sum(
            samples,
            "athena_gateway_deferred_events_total",
            &[("kind", "gateway_insert"), ("status", "queued")],
        ),
    );
    out.insert(
        "backend_unavailable_insert".to_string(),
        metric_sum(
            samples,
            "athena_gateway_backend_unavailable_total",
            &[("route", "insert")],
        ),
    );
    out
}

async fn fetch_metrics(
    client: &Client,
    base_url: &str,
    metrics_path: &str,
) -> Result<BTreeMap<String, f64>, Box<dyn Error>> {
    let url = make_url(base_url, metrics_path);
    let text = client
        .get(url)
        .send()
        .await?
        .error_for_status()?
        .text()
        .await?;

    let samples: Vec<MetricSample> = text.lines().filter_map(parse_metric_line).collect();
    Ok(collect_perf_metrics(&samples))
}

async fn send_insert(client: &Client, cfg: &ReplayConfig) -> HttpResult {
    let started = Instant::now();
    let url = make_url(&cfg.base_url, &cfg.insert_path);
    let mut status = 0u16;
    let mut code = "transport_error".to_string();

    let result = client
        .put(url)
        .header("Content-Type", "application/json")
        .header("X-Athena-Client", cfg.client_name.as_str())
        .json(&cfg.payload)
        .send()
        .await;

    if let Ok(response) = result {
        status = response.status().as_u16();
        let body_text = response.text().await.unwrap_or_default();
        if let Ok(parsed) = serde_json::from_str::<Value>(&body_text) {
            code = parsed
                .get("code")
                .and_then(Value::as_str)
                .map(|s| s.to_string())
                .or_else(|| {
                    parsed
                        .get("status")
                        .and_then(Value::as_str)
                        .map(|s| s.to_string())
                })
                .unwrap_or_else(|| status.to_string());
        } else {
            code = status.to_string();
        }
    }

    HttpResult {
        latency_ms: started.elapsed().as_secs_f64() * 1000.0,
        status,
        code,
    }
}

fn subtract_metrics(
    after: &BTreeMap<String, f64>,
    before: &BTreeMap<String, f64>,
) -> BTreeMap<String, f64> {
    let mut keys = BTreeMap::<String, ()>::new();
    for key in after.keys() {
        keys.insert(key.clone(), ());
    }
    for key in before.keys() {
        keys.insert(key.clone(), ());
    }

    let mut out = BTreeMap::new();
    for key in keys.keys() {
        let a = after.get(key).copied().unwrap_or(0.0);
        let b = before.get(key).copied().unwrap_or(0.0);
        out.insert(key.clone(), a - b);
    }
    out
}

async fn run_scenario(
    cfg: &ReplayConfig,
    workers: usize,
) -> Result<ScenarioReport, Box<dyn Error>> {
    let timeout = std::time::Duration::from_secs_f64(cfg.timeout_secs);
    let client = Client::builder().timeout(timeout).build()?;

    let before = fetch_metrics(&client, &cfg.base_url, &cfg.metrics_path).await?;

    if cfg.seed_before_each_scenario {
        let _ = send_insert(&client, cfg).await;
    }

    let started = Instant::now();
    let mut set = JoinSet::new();
    for _ in 0..workers {
        let per_worker = cfg.requests_per_worker;
        let scenario_cfg = cfg.clone();
        let worker_client = client.clone();
        set.spawn(async move {
            let mut local = Vec::with_capacity(per_worker);
            for _ in 0..per_worker {
                local.push(send_insert(&worker_client, &scenario_cfg).await);
            }
            local
        });
    }

    let mut results = Vec::<HttpResult>::with_capacity(workers * cfg.requests_per_worker);
    while let Some(joined) = set.join_next().await {
        let mut worker_results = joined?;
        results.append(&mut worker_results);
    }

    let elapsed_s = started.elapsed().as_secs_f64();
    let after = fetch_metrics(&client, &cfg.base_url, &cfg.metrics_path).await?;
    let metric_deltas = subtract_metrics(&after, &before);

    let latencies: Vec<f64> = results.iter().map(|r| r.latency_ms).collect();
    let mut status_counts = BTreeMap::<u16, usize>::new();
    let mut code_counts = BTreeMap::<String, usize>::new();
    for result in &results {
        *status_counts.entry(result.status).or_insert(0) += 1;
        *code_counts.entry(result.code.clone()).or_insert(0) += 1;
    }

    let mean_ms = if latencies.is_empty() {
        0.0
    } else {
        latencies.iter().sum::<f64>() / latencies.len() as f64
    };
    let min_ms = latencies
        .iter()
        .fold(f64::INFINITY, |acc, value| acc.min(*value));
    let max_ms = latencies
        .iter()
        .fold(f64::NEG_INFINITY, |acc, value| acc.max(*value));

    Ok(ScenarioReport {
        workers,
        total_requests: workers * cfg.requests_per_worker,
        elapsed_s,
        p50_ms: percentile(&latencies, 50.0),
        p95_ms: percentile(&latencies, 95.0),
        p99_ms: percentile(&latencies, 99.0),
        mean_ms,
        min_ms: if min_ms.is_finite() { min_ms } else { 0.0 },
        max_ms: if max_ms.is_finite() { max_ms } else { 0.0 },
        status_counts,
        code_counts,
        metric_deltas,
    })
}

fn render_counts_u16(map: &BTreeMap<u16, usize>) -> String {
    if map.is_empty() {
        return "-".to_string();
    }
    map.iter()
        .map(|(k, v)| format!("{}:{}", k, v))
        .collect::<Vec<_>>()
        .join(", ")
}

fn render_counts_string(map: &BTreeMap<String, usize>) -> String {
    if map.is_empty() {
        return "-".to_string();
    }
    let mut pairs: Vec<(&String, &usize)> = map.iter().collect();
    pairs.sort_by(|a, b| b.1.cmp(a.1).then_with(|| a.0.cmp(b.0)));
    pairs
        .into_iter()
        .take(6)
        .map(|(k, v)| format!("{}:{}", k, v))
        .collect::<Vec<_>>()
        .join(", ")
}

fn metric_delta(report: &ScenarioReport, key: &str) -> f64 {
    report.metric_deltas.get(key).copied().unwrap_or(0.0)
}

fn render_report(cfg: &ReplayConfig, reports: &[ScenarioReport]) -> String {
    let generated_at = chrono::Utc::now().to_rfc3339();
    let workers = reports
        .iter()
        .map(|r| r.workers.to_string())
        .collect::<Vec<_>>()
        .join(", ");

    let mut out = String::new();
    out.push_str(&format!(
        "# PERF-01 Duplicate Insert Replay Report ({})\n\n",
        cfg.label
    ));
    out.push_str(&format!("- Generated: {}\n", generated_at));
    out.push_str(&format!("- Base URL: {}\n", cfg.base_url));
    out.push_str(&format!("- Insert path: {}\n", cfg.insert_path));
    out.push_str(&format!("- Metrics path: {}\n", cfg.metrics_path));
    out.push_str(&format!("- Client: {}\n", cfg.client_name));
    out.push_str(&format!("- Workers: {}\n", workers));
    out.push_str(&format!(
        "- Requests per worker: {}\n",
        cfg.requests_per_worker
    ));
    out.push_str(&format!("- Request timeout (s): {:.2}\n", cfg.timeout_secs));
    out.push_str(&format!(
        "- Seed before each scenario: {}\n\n",
        cfg.seed_before_each_scenario
    ));

    out.push_str("## Replay Payload\n\n");
    out.push_str("```json\n");
    out.push_str(&serde_json::to_string_pretty(&cfg.payload).unwrap_or_else(|_| "{}".to_string()));
    out.push_str("\n```\n\n");

    out.push_str("## Latency and Outcome Summary\n\n");
    out.push_str("| Workers | Requests | Elapsed(s) | p50(ms) | p95(ms) | p99(ms) | Mean(ms) | Min(ms) | Max(ms) | Status Counts | Top Codes |\n");
    out.push_str("| --- | --- | --- | --- | --- | --- | --- | --- | --- | --- | --- |\n");
    for report in reports {
        out.push_str(&format!(
            "| {} | {} | {:.2} | {:.2} | {:.2} | {:.2} | {:.2} | {:.2} | {:.2} | {} | {} |\n",
            report.workers,
            report.total_requests,
            report.elapsed_s,
            report.p50_ms,
            report.p95_ms,
            report.p99_ms,
            report.mean_ms,
            report.min_ms,
            report.max_ms,
            render_counts_u16(&report.status_counts),
            render_counts_string(&report.code_counts)
        ));
    }
    out.push_str("\n");

    out.push_str("## DB 23505 Proxy and Queue/Fallback Event Deltas\n\n");
    out.push_str("| Workers | db_unique_violation (proxy 23505) | admission_defer_high_watermark | admission_reject_queue_full | admission_reject_enqueue_failed | db_timeout | window_wait_timeout | bulk_fallback_sql_error | deferred_insert_queued | backend_unavailable_insert |\n");
    out.push_str("| --- | --- | --- | --- | --- | --- | --- | --- | --- | --- |\n");
    for report in reports {
        out.push_str(&format!(
            "| {} | {:.0} | {:.0} | {:.0} | {:.0} | {:.0} | {:.0} | {:.0} | {:.0} | {:.0} |\n",
            report.workers,
            metric_delta(report, "db_unique_violation"),
            metric_delta(report, "admission_defer_high_watermark"),
            metric_delta(report, "admission_reject_queue_full"),
            metric_delta(report, "admission_reject_enqueue_failed"),
            metric_delta(report, "db_timeout"),
            metric_delta(report, "window_wait_timeout"),
            metric_delta(report, "bulk_fallback_sql_error"),
            metric_delta(report, "deferred_insert_queued"),
            metric_delta(report, "backend_unavailable_insert"),
        ));
    }
    out.push_str("\n");

    out.push_str("## Notes\n\n");
    out.push_str("- db_unique_violation is a gateway metric proxy for observed DB sql_state=23505 conflict paths.\n");
    out.push_str("- admission_*, db_timeout, window_wait_timeout, and bulk_fallback_sql_error capture queue/fallback resilience behavior.\n");
    out.push_str("- For strict before/after comparison, run this example twice with --label before and --label after.\n");

    out
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let cfg = parse_args()?;

    let mut reports = Vec::new();
    for workers in &cfg.workers {
        let report = run_scenario(&cfg, *workers).await?;
        println!(
            "workers={} requests={} p50={:.2}ms p95={:.2}ms p99={:.2}ms statuses={}",
            report.workers,
            report.total_requests,
            report.p50_ms,
            report.p95_ms,
            report.p99_ms,
            render_counts_u16(&report.status_counts)
        );
        reports.push(report);
    }

    let report_text = render_report(&cfg, &reports);
    let output = cfg.output_path.unwrap_or_else(|| {
        let ts = chrono::Utc::now().format("%Y%m%d-%H%M%S").to_string();
        PathBuf::from(format!("perf-01-report-{}-{}.md", cfg.label, ts))
    });

    fs::write(&output, report_text)?;
    println!("Wrote report: {}", output.display());
    Ok(())
}
