use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering::Relaxed};
use std::time::{Duration, Instant};
use hdrhistogram::Histogram;
use indicatif::{ProgressBar, ProgressStyle};
use tokio::sync::{Mutex, Semaphore};
use tokio::time::sleep;
use crate::types::*;
use crate::rpc;
use crate::VERSION;

pub struct RunConfig {
    pub method: String,
    pub rates: Vec<u64>,
    pub duration: Option<u64>,
    pub durations: Option<Vec<u64>>,
    pub network: String,
    pub seed: Option<u64>,
    pub output_dir: Option<String>,
    pub verbose: bool,
    pub dry_run: bool,
    pub deep_check: bool,
    pub save_raw_output: bool,
}

impl Default for RunConfig {
    fn default() -> Self {
        Self {
            method: String::new(),
            rates: vec![10],
            duration: Some(10),
            durations: None,
            network: "mainnet".to_string(),
            seed: None,
            output_dir: None,
            verbose: true,
            dry_run: false,
            deep_check: false,
            save_raw_output: false,
        }
    }
}

pub async fn run(nodes: &Nodes, config: RunConfig) -> anyhow::Result<RunOutput> {
    let output_dir = config.output_dir.clone()
        .unwrap_or_else(|| std::env::temp_dir().join(format!("aqua_{}", chrono::Utc::now().timestamp())).to_string_lossy().to_string());
    std::fs::create_dir_all(&output_dir)?;

    let test = rpc::create_test(
        &config.method,
        &config.rates,
        config.duration,
        config.durations.as_deref(),
        &config.network,
        config.seed,
    )?;

    if config.verbose {
        println!("\naqua\n====\ntest: {}\nrates: {:?}\noutput: {}\n", config.method, config.rates, output_dir);
    }

    save_test(&output_dir, &test)?;

    if config.dry_run {
        println!("\n[dry run] test preview");
        println!("------------------------");
        println!("method: {}", config.method);
        println!("nodes: {}", nodes.keys().map(|n| n.as_str()).collect::<Vec<_>>().join(", "));
        println!("rates: {} rps", config.rates.iter().map(|r| r.to_string()).collect::<Vec<_>>().join(", "));
        println!("duration: {}s per rate", config.duration.unwrap_or(10));
        let total_calls: usize = test.attacks.iter().map(|a| a.calls.len()).sum();
        println!("total calls: {}", total_calls);
        println!("network: {}", config.network);
        if let Some(seed) = config.seed {
            println!("seed: {}", seed);
        }
        println!("\nrun without --dry to execute");
        return Ok(RunOutput {
            output_dir,
            test: Some(test.clone()),
            test_parameters: test.test_parameters.clone(),
            payload: SingleRunResultsPayload {
                aqua_version: VERSION.to_string(),
                cli_args: vec![],
                run_type: "single".to_string(),
                t_run_start: 0.0,
                t_run_end: 0.0,
                nodes: nodes.clone(),
                results: HashMap::new(),
            },
        });
    }

    let start = Instant::now();
    if config.verbose { println!("starting load test\n"); }

    let mut results = HashMap::new();
    for (name, node) in nodes {
        if config.verbose { println!("node: {}", name); }
        let res = run_test(&node.url, &test, config.verbose).await?;
        results.insert(name.clone(), res);
    }

    let payload = SingleRunResultsPayload {
        aqua_version: VERSION.to_string(),
        cli_args: std::env::args().collect(),
        run_type: "single".to_string(),
        t_run_start: 0.0,
        t_run_end: start.elapsed().as_secs_f64(),
        nodes: nodes.clone(),
        results: results.clone(),
    };

    save_results(&output_dir, &payload)?;
    if config.verbose { print_results(&results); }

    Ok(RunOutput {
        output_dir,
        test: Some(test.clone()),
        test_parameters: test.test_parameters,
        payload,
    })
}

async fn run_test(url: &str, test: &LoadTest, verbose: bool) -> anyhow::Result<LoadTestOutput> {
    let mut output = LoadTestOutput::default();
    for attack in &test.attacks {
        if verbose { println!("  rate={} rps, duration={}s", attack.rate, attack.duration); }
        let res = run_attack(url, attack, verbose).await?;
        output.target_rate.push(res.target_rate);
        output.actual_rate.push(res.actual_rate);
        output.target_duration.push(res.target_duration);
        output.actual_duration.push(res.actual_duration);
        output.requests.push(res.requests);
        output.throughput.push(res.throughput);
        output.success.push(res.success);
        output.min.push(res.min);
        output.mean.push(res.mean);
        output.p50.push(res.p50);
        output.p90.push(res.p90);
        output.p95.push(res.p95);
        output.p99.push(res.p99);
        output.max.push(res.max);
        output.status_codes.push(res.status_codes);
        output.errors.push(res.errors);
    }
    Ok(output)
}

async fn run_attack(url: &str, attack: &VegetaAttack, verbose: bool) -> anyhow::Result<LoadTestOutputDatum> {
    let client = reqwest::Client::builder()
        .timeout(Duration::from_secs(30))
        .pool_max_idle_per_host(attack.rate as usize)
        .build()?;

    let start = Instant::now();
    let end = start + Duration::from_secs(attack.duration);
    let hist = Arc::new(Mutex::new(Histogram::<u64>::new_with_bounds(1, 60_000_000, 3)?));
    let success = Arc::new(AtomicU64::new(0));
    let errors = Arc::new(AtomicU64::new(0));
    let status_codes = Arc::new(Mutex::new(HashMap::<String, u64>::new()));
    let error_msgs = Arc::new(Mutex::new(Vec::<String>::new()));
    let delay = if attack.rate > 0 { 1_000_000 / attack.rate } else { 1_000_000 };
    let sem = Arc::new(Semaphore::new(attack.rate.max(100) as usize));

    let pb = if verbose {
        let pb = ProgressBar::new(attack.rate * attack.duration);
        if let Ok(style) = ProgressStyle::default_bar()
            .template("    {spinner:.green} [{elapsed_precise}] [{bar:40.cyan/blue}] {pos}/{len}") {
            pb.set_style(style.progress_chars("#>-"));
        }
        Some(pb)
    } else { None };

    let mut handles = Vec::new();
    let mut idx = 0;
    while Instant::now() < end {
        let body = if attack.calls.is_empty() {
            serde_json::json!({"jsonrpc": "2.0", "id": 1, "method": "getSlot", "params": []})
        } else {
            let c: Call = serde_json::from_value(attack.calls[idx % attack.calls.len()].clone())?;
            idx += 1;
            serde_json::json!({"jsonrpc": "2.0", "id": idx, "method": c.method, "params": c.params})
        };

        let (client, url, hist, success, errors) = (client.clone(), url.to_string(), hist.clone(), success.clone(), errors.clone());
        let (status_codes, error_msgs, pb) = (status_codes.clone(), error_msgs.clone(), pb.clone());
        let permit = sem.clone().acquire_owned().await?;

        handles.push(tokio::spawn(async move {
            let _p = permit;
            let t = Instant::now();
            match client.post(&url).json(&body).send().await {
                Ok(r) => {
                    { status_codes.lock().await.entry(r.status().as_u16().to_string()).and_modify(|c| *c += 1).or_insert(1); }
                    if r.status().is_success() {
                        if let Ok(json) = r.json::<serde_json::Value>().await {
                            if json.get("error").is_some() {
                                errors.fetch_add(1, Relaxed);
                                let mut e = error_msgs.lock().await;
                                if e.len() < 10 { e.push(json["error"].to_string()); }
                            } else {
                                success.fetch_add(1, Relaxed);
                            }
                        } else {
                            success.fetch_add(1, Relaxed);
                        }
                    } else {
                        errors.fetch_add(1, Relaxed);
                    }
                    let _ = hist.lock().await.record(t.elapsed().as_micros() as u64);
                }
                Err(e) => {
                    errors.fetch_add(1, Relaxed);
                    let mut errs = error_msgs.lock().await;
                    if errs.len() < 10 { errs.push(e.to_string()); }
                }
            }
            if let Some(pb) = pb { pb.inc(1); }
        }));
        sleep(Duration::from_micros(delay)).await;
    }

    for h in handles { let _ = h.await; }
    if let Some(pb) = pb { pb.finish_and_clear(); }

    let dur = start.elapsed().as_secs_f64();
    let h = hist.lock().await;
    let succ = success.load(Relaxed);
    let errs = errors.load(Relaxed);
    let total = succ + errs;

    let pct = |q| if !h.is_empty() { Some(h.value_at_quantile(q) as f64 / 1e6) } else { None };
    let final_status = status_codes.lock().await.clone();
    let final_errors = error_msgs.lock().await.clone();

    Ok(LoadTestOutputDatum {
        target_rate: attack.rate,
        actual_rate: Some(total as f64 / dur),
        target_duration: attack.duration,
        actual_duration: Some(dur),
        requests: total,
        throughput: Some(total as f64 / dur),
        success: Some(succ as f64 / total.max(1) as f64),
        min: if !h.is_empty() { Some(h.min() as f64 / 1e6) } else { None },
        mean: if !h.is_empty() { Some(h.mean() / 1e6) } else { None },
        p50: pct(0.50), p90: pct(0.90), p95: pct(0.95), p99: pct(0.99),
        max: if !h.is_empty() { Some(h.max() as f64 / 1e6) } else { None },
        status_codes: final_status,
        errors: final_errors,
        first_request_timestamp: None,
        last_request_timestamp: None,
        last_response_timestamp: None,
        final_wait_time: None,
    })
}

fn save_test(dir: &str, test: &LoadTest) -> anyhow::Result<()> {
    let payload = SingleRunTestPayload {
        aqua_version: VERSION.to_string(),
        run_type: "single".to_string(),
        name: test.test_parameters.test_name.clone(),
        test_parameters: test.test_parameters.clone(),
    };
    std::fs::write(PathBuf::from(dir).join("test.json"), serde_json::to_string_pretty(&payload)?)?;
    Ok(())
}

fn save_results(dir: &str, payload: &SingleRunResultsPayload) -> anyhow::Result<()> {
    std::fs::write(PathBuf::from(dir).join("results.json"), serde_json::to_string_pretty(payload)?)?;
    Ok(())
}

fn print_results(results: &HashMap<String, LoadTestOutput>) {
    println!("\nresults\n=======");
    for (name, out) in results {
        println!("\n{}", name);
        if out.target_rate.is_empty() { continue; }
        let total: u64 = out.requests.iter().sum();
        let avg_success = out.success.iter().filter_map(|s| *s).sum::<f64>() / out.success.len().max(1) as f64;
        println!("  requests: {}, success: {:.1}%", total, avg_success * 100.0);
        if let Some(Some(m)) = out.mean.last() { println!("  mean latency: {:.2}ms", m * 1000.0); }
        if let Some(Some(p)) = out.p99.last() { println!("  p99 latency: {:.2}ms", p * 1000.0); }
    }
}

pub fn load_test_payload(path: &str) -> anyhow::Result<SingleRunTestPayload> {
    let p = if std::path::Path::new(path).is_dir() {
        PathBuf::from(path).join("test.json")
    } else {
        PathBuf::from(path)
    };
    Ok(serde_json::from_str(&std::fs::read_to_string(p)?)?)
}

pub fn load_results_payload(path: &str) -> anyhow::Result<SingleRunResultsPayload> {
    let p = if std::path::Path::new(path).is_dir() {
        PathBuf::from(path).join("results.json")
    } else {
        PathBuf::from(path)
    };
    Ok(serde_json::from_str(&std::fs::read_to_string(p)?)?)
}

// Equality testing
#[derive(Debug, Clone, serde::Serialize)]
pub struct EqualityResult {
    pub all_equal: bool,
    pub differences: Vec<EqualityDiff>,
}

#[derive(Debug, Clone, serde::Serialize)]
pub struct EqualityDiff {
    pub call_index: usize,
    pub node1: String,
    pub node2: String,
    pub response1: serde_json::Value,
    pub response2: serde_json::Value,
}

pub async fn run_equality_test(nodes: &Nodes, calls: &[Call], verbose: bool) -> anyhow::Result<EqualityResult> {
    let client = reqwest::Client::builder().timeout(Duration::from_secs(30)).build()?;
    let mut all_responses: HashMap<String, Vec<serde_json::Value>> = HashMap::new();

    for (name, node) in nodes {
        if verbose { println!("testing node: {}", name); }

        let pb = if verbose {
            let pb = ProgressBar::new(calls.len() as u64);
            if let Ok(style) = ProgressStyle::default_bar()
                .template("  {spinner:.green} [{elapsed_precise}] [{bar:40.cyan/blue}] {pos}/{len}") {
                pb.set_style(style.progress_chars("#>-"));
            }
            Some(pb)
        } else { None };

        let mut responses = Vec::new();
        for (i, call) in calls.iter().enumerate() {
            let body = serde_json::json!({"jsonrpc": "2.0", "id": i + 1, "method": call.method, "params": call.params});
            match client.post(&node.url).json(&body).send().await {
                Ok(resp) => {
                    match resp.json::<serde_json::Value>().await {
                        Ok(json) => responses.push(json),
                        Err(e) => responses.push(serde_json::json!({"error": e.to_string()})),
                    }
                }
                Err(e) => responses.push(serde_json::json!({"error": e.to_string()})),
            }
            if let Some(ref pb) = pb { pb.inc(1); }
        }
        if let Some(pb) = pb { pb.finish_and_clear(); }
        all_responses.insert(name.clone(), responses);
    }

    let mut differences = Vec::new();
    let node_names: Vec<_> = all_responses.keys().cloned().collect();

    if node_names.len() >= 2 {
        let first = &node_names[0];
        let first_results = &all_responses[first];

        for other in &node_names[1..] {
            let other_results = &all_responses[other];
            for (i, (r1, r2)) in first_results.iter().zip(other_results.iter()).enumerate() {
                if !results_equal(r1, r2) {
                    differences.push(EqualityDiff {
                        call_index: i,
                        node1: first.clone(),
                        node2: other.clone(),
                        response1: r1.clone(),
                        response2: r2.clone(),
                    });
                }
            }
        }
    }

    Ok(EqualityResult {
        all_equal: differences.is_empty(),
        differences,
    })
}

fn results_equal(r1: &serde_json::Value, r2: &serde_json::Value) -> bool {
    match (r1.get("result"), r2.get("result")) {
        (Some(a), Some(b)) => a == b,
        (None, None) => r1.get("error") == r2.get("error"),
        _ => false,
    }
}

pub fn print_metric_tables(results: &HashMap<String, LoadTestOutput>, metrics: &[&str]) {
    use comfy_table::{Table, ContentArrangement};

    for metric in metrics {
        println!("\n{}:", metric);
        let mut table = Table::new();
        table.set_content_arrangement(ContentArrangement::Dynamic);

        let mut header = vec!["Node".to_string()];
        if let Some((_, first)) = results.iter().next() {
            for rate in &first.target_rate {
                header.push(format!("{} rps", rate));
            }
        }
        table.set_header(header);

        for (name, out) in results {
            let mut row = vec![name.clone()];
            let values: &Vec<Option<f64>> = match *metric {
                "success" => &out.success,
                "throughput" => &out.throughput,
                "mean" => &out.mean,
                "p50" => &out.p50,
                "p90" => &out.p90,
                "p95" => &out.p95,
                "p99" => &out.p99,
                "min" => &out.min,
                "max" => &out.max,
                _ => &out.success,
            };
            for v in values {
                row.push(match v {
                    Some(val) if *metric == "success" => format!("{:.1}%", val * 100.0),
                    Some(val) if *metric == "throughput" => format!("{:.1}", val),
                    Some(val) => format!("{:.2}ms", val * 1000.0),
                    None => "-".to_string(),
                });
            }
            table.add_row(row);
        }
        println!("{}", table);
    }
}

pub fn print_results_summary(results: &HashMap<String, LoadTestOutput>) {
    println!("\n=== Results Summary ===\n");
    for (name, out) in results {
        println!("{}:", name);
        if out.target_rate.is_empty() {
            println!("  no results\n");
            continue;
        }
        let total: u64 = out.requests.iter().sum();
        let avg_success = out.success.iter().filter_map(|s| *s).sum::<f64>() / out.success.len().max(1) as f64;
        let avg_throughput = out.throughput.iter().filter_map(|t| *t).sum::<f64>() / out.throughput.len().max(1) as f64;
        println!("  total requests: {}", total);
        println!("  avg success: {:.1}%", avg_success * 100.0);
        println!("  avg throughput: {:.1} rps\n", avg_throughput);
    }
}
