use std::collections::{HashMap, HashSet};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use super::collector::ProviderMeta;
use super::config::{GeyserConfig, GrpcSettingsSummary};
use super::types::*;
use crate::VERSION;

type SlotKey = (u64, SlotStatus);
type AcctKey = (u64, String, String);

pub fn process_results(
    cfg: &GeyserConfig,
    data1: EndpointData,
    data2: EndpointData,
    meta1: ProviderMeta,
    meta2: ProviderMeta,
    duration_secs: f64,
    track_accounts: bool,
    owner_filter: Option<String>,
) -> GeyserBenchmarkResult {
    let duration_ms = (duration_secs * 1000.0) as u64;
    let started_at = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis() as u64
        - duration_ms;

    let map1 = build_slot_map(&data1.slot_updates);
    let map2 = build_slot_map(&data2.slot_updates);

    let slots1: HashSet<u64> = map1.keys().map(|(s, _)| *s).collect();
    let slots2: HashSet<u64> = map2.keys().map(|(s, _)| *s).collect();

    let total_slots = (slots1.len() + slots2.len()) as u64;
    let common: Vec<u64> = slots1.intersection(&slots2).copied().collect();
    let common_count = common.len() as u64;

    let accts1_by_slot = group_accounts(&data1.account_updates);
    let accts2_by_slot = group_accounts(&data2.account_updates);
    let acct_matches = if track_accounts {
        build_account_matches(&data1.account_updates, &data2.account_updates)
    } else {
        HashMap::new()
    };

    let mut compared = 0u64;
    let mut dropped = 0u64;

    let mut p1_first_shred = Vec::new();
    let mut p1_processing = Vec::new();
    let mut p1_confirmation = Vec::new();
    let mut p1_finalization = Vec::new();
    let mut p1_download = Vec::new();
    let mut p1_replay = Vec::new();
    let mut p1_confirm = Vec::new();
    let mut p1_finalize = Vec::new();
    let mut p1_acct_delay = Vec::new();

    let mut p2_first_shred = Vec::new();
    let mut p2_processing = Vec::new();
    let mut p2_confirmation = Vec::new();
    let mut p2_finalization = Vec::new();
    let mut p2_download = Vec::new();
    let mut p2_replay = Vec::new();
    let mut p2_confirm = Vec::new();
    let mut p2_finalize = Vec::new();
    let mut p2_acct_delay = Vec::new();

    let mut slot_comparisons = Vec::new();

    let mut sorted_common = common;
    sorted_common.sort_unstable();

    for slot in sorted_common {
        if map1.contains_key(&(slot, SlotStatus::Dead)) || map2.contains_key(&(slot, SlotStatus::Dead)) {
            dropped += 1;
            continue;
        }

        let stages = SlotStatus::required_stages();
        let ep1_ok = stages.iter().all(|s| map1.contains_key(&(slot, *s)));
        let ep2_ok = stages.iter().all(|s| map2.contains_key(&(slot, *s)));

        if !ep1_ok || !ep2_ok {
            dropped += 1;
            continue;
        }

        if track_accounts {
            let cnt1 = accts1_by_slot.get(&slot).map(|v| v.len()).unwrap_or(0);
            let cnt2 = accts2_by_slot.get(&slot).map(|v| v.len()).unwrap_or(0);
            if cnt1 != cnt2 {
                dropped += 1;
                continue;
            }
        }

        let (d1_fs, d2_fs) = calc_delay(&map1, &map2, slot, SlotStatus::FirstShredReceived);
        let (d1_pr, d2_pr) = calc_delay(&map1, &map2, slot, SlotStatus::Processed);
        let (d1_cf, d2_cf) = calc_delay(&map1, &map2, slot, SlotStatus::Confirmed);
        let (d1_fn, d2_fn) = calc_delay(&map1, &map2, slot, SlotStatus::Finalized);

        if let Some(d) = d1_fs { p1_first_shred.push(d); }
        if let Some(d) = d2_fs { p2_first_shred.push(d); }
        if let Some(d) = d1_pr { p1_processing.push(d); }
        if let Some(d) = d2_pr { p2_processing.push(d); }
        if let Some(d) = d1_cf { p1_confirmation.push(d); }
        if let Some(d) = d2_cf { p2_confirmation.push(d); }
        if let Some(d) = d1_fn { p1_finalization.push(d); }
        if let Some(d) = d2_fn { p2_finalization.push(d); }

        let dl1 = calc_download(&map1, slot);
        let dl2 = calc_download(&map2, slot);
        let rp1 = calc_replay(&map1, slot);
        let rp2 = calc_replay(&map2, slot);
        let cn1 = calc_confirm(&map1, slot);
        let cn2 = calc_confirm(&map2, slot);
        let fi1 = calc_finalize(&map1, slot);
        let fi2 = calc_finalize(&map2, slot);

        if let Some(d) = dl1 { p1_download.push(d); }
        if let Some(d) = dl2 { p2_download.push(d); }
        if let Some(d) = rp1 { p1_replay.push(d); }
        if let Some(d) = rp2 { p2_replay.push(d); }
        if let Some(d) = cn1 { p1_confirm.push(d); }
        if let Some(d) = cn2 { p2_confirm.push(d); }
        if let Some(d) = fi1 { p1_finalize.push(d); }
        if let Some(d) = fi2 { p2_finalize.push(d); }

        let mut detail1 = build_slot_detail(&map1, slot);
        let mut detail2 = build_slot_detail(&map2, slot);

        detail1.first_shred_delay_ms = d1_fs.map(|d| d.as_secs_f64() * 1000.0);
        detail2.first_shred_delay_ms = d2_fs.map(|d| d.as_secs_f64() * 1000.0);
        detail1.processing_delay_ms = d1_pr.map(|d| d.as_secs_f64() * 1000.0);
        detail2.processing_delay_ms = d2_pr.map(|d| d.as_secs_f64() * 1000.0);
        detail1.confirmation_delay_ms = d1_cf.map(|d| d.as_secs_f64() * 1000.0);
        detail2.confirmation_delay_ms = d2_cf.map(|d| d.as_secs_f64() * 1000.0);
        detail1.finalization_delay_ms = d1_fn.map(|d| d.as_secs_f64() * 1000.0);
        detail2.finalization_delay_ms = d2_fn.map(|d| d.as_secs_f64() * 1000.0);

        if track_accounts {
            detail1.accounts = build_acct_details(accts1_by_slot.get(&slot), &acct_matches, 1);
            detail2.accounts = build_acct_details(accts2_by_slot.get(&slot), &acct_matches, 2);

            for a in &detail1.accounts {
                if let Some(d) = a.delay_ms {
                    if d > 0.0 { p1_acct_delay.push(Duration::from_secs_f64(d / 1000.0)); }
                }
            }
            for a in &detail2.accounts {
                if let Some(d) = a.delay_ms {
                    if d > 0.0 { p2_acct_delay.push(Duration::from_secs_f64(d / 1000.0)); }
                }
            }
        }

        slot_comparisons.push(SlotComparison {
            slot,
            provider1: detail1,
            provider2: detail2,
        });

        compared += 1;
    }

    let p1_metrics = ProviderMetrics {
        first_shred_delay: Percentiles::from_durations(p1_first_shred),
        processing_delay: Percentiles::from_durations(p1_processing),
        confirmation_delay: Percentiles::from_durations(p1_confirmation),
        finalization_delay: Percentiles::from_durations(p1_finalization),
        download_time: Percentiles::from_durations(p1_download),
        replay_time: Percentiles::from_durations(p1_replay),
        confirm_time: Percentiles::from_durations(p1_confirm),
        finalize_time: Percentiles::from_durations(p1_finalize),
        account_delay: if track_accounts && !p1_acct_delay.is_empty() {
            Some(Percentiles::from_durations(p1_acct_delay))
        } else {
            None
        },
    };

    let p2_metrics = ProviderMetrics {
        first_shred_delay: Percentiles::from_durations(p2_first_shred),
        processing_delay: Percentiles::from_durations(p2_processing),
        confirmation_delay: Percentiles::from_durations(p2_confirmation),
        finalization_delay: Percentiles::from_durations(p2_finalization),
        download_time: Percentiles::from_durations(p2_download),
        replay_time: Percentiles::from_durations(p2_replay),
        confirm_time: Percentiles::from_durations(p2_confirm),
        finalize_time: Percentiles::from_durations(p2_finalize),
        account_delay: if track_accounts && !p2_acct_delay.is_empty() {
            Some(Percentiles::from_durations(p2_acct_delay))
        } else {
            None
        },
    };

    GeyserBenchmarkResult {
        version: VERSION.to_string(),
        with_accounts: track_accounts,
        account_filter: owner_filter,
        grpc_settings: GrpcSettingsSummary::from(cfg),
        meta: BenchmarkMeta {
            total_slots,
            common_slots: common_count,
            compared_slots: compared,
            dropped_slots: dropped,
            duration_ms,
            started_at,
            account_totals: if track_accounts {
                Some((data1.account_updates.len() as u64, data2.account_updates.len() as u64))
            } else {
                None
            },
        },
        providers: [
            ProviderInfo {
                url: data1.name,
                plugin_name: meta1.plugin_name,
                plugin_version: meta1.version,
                avg_ping_ms: meta1.avg_ping.as_secs_f64() * 1000.0,
                total_updates: data1.slot_updates.len() as u64,
                unique_slots: slots1.len() as u64,
                account_count: if track_accounts { Some(data1.account_updates.len() as u64) } else { None },
            },
            ProviderInfo {
                url: data2.name,
                plugin_name: meta2.plugin_name,
                plugin_version: meta2.version,
                avg_ping_ms: meta2.avg_ping.as_secs_f64() * 1000.0,
                total_updates: data2.slot_updates.len() as u64,
                unique_slots: slots2.len() as u64,
                account_count: if track_accounts { Some(data2.account_updates.len() as u64) } else { None },
            },
        ],
        provider1_metrics: p1_metrics,
        provider2_metrics: p2_metrics,
        slots: slot_comparisons,
    }
}

fn build_slot_map(updates: &[SlotUpdate]) -> HashMap<SlotKey, SlotUpdate> {
    updates.iter().map(|u| ((u.slot, u.status), u.clone())).collect()
}

fn group_accounts(updates: &[AccountUpdate]) -> HashMap<u64, Vec<AccountUpdate>> {
    let mut by_slot = HashMap::new();
    for u in updates {
        by_slot.entry(u.slot).or_insert_with(Vec::new).push(u.clone());
    }
    by_slot
}

fn build_account_matches(
    updates1: &[AccountUpdate],
    updates2: &[AccountUpdate],
) -> HashMap<AcctKey, (Vec<(u64, Instant)>, Vec<(u64, Instant)>)> {
    let mut map: HashMap<AcctKey, (Vec<(u64, Instant)>, Vec<(u64, Instant)>)> = HashMap::new();

    for u in updates1 {
        let key = (u.slot, u.pubkey.clone(), u.txn_signature.clone().unwrap_or_default());
        map.entry(key).or_insert((Vec::new(), Vec::new())).0.push((u.write_version, u.timestamp));
    }

    for u in updates2 {
        let key = (u.slot, u.pubkey.clone(), u.txn_signature.clone().unwrap_or_default());
        map.entry(key).or_insert((Vec::new(), Vec::new())).1.push((u.write_version, u.timestamp));
    }

    for (_, (v1, v2)) in map.iter_mut() {
        v1.sort_by_key(|(wv, _)| *wv);
        v2.sort_by_key(|(wv, _)| *wv);
    }

    map
}

fn build_acct_details(
    updates: Option<&Vec<AccountUpdate>>,
    matches: &HashMap<AcctKey, (Vec<(u64, Instant)>, Vec<(u64, Instant)>)>,
    endpoint_num: usize,
) -> Vec<AccountDetail> {
    let mut out = Vec::new();
    if let Some(updates) = updates {
        for u in updates {
            let key = (u.slot, u.pubkey.clone(), u.txn_signature.clone().unwrap_or_default());
            let delay_ms = matches.get(&key).and_then(|(v1, v2)| {
                match (v1.last(), v2.last()) {
                    (Some((_, i1)), Some((_, i2))) => {
                        if endpoint_num == 1 {
                            if i1 < i2 { Some(0.0) } else { Some(i1.duration_since(*i2).as_secs_f64() * 1000.0) }
                        } else {
                            if i2 < i1 { Some(0.0) } else { Some(i2.duration_since(*i1).as_secs_f64() * 1000.0) }
                        }
                    }
                    _ => None,
                }
            });
            out.push(AccountDetail {
                pubkey: u.pubkey.clone(),
                write_version: u.write_version,
                signature: u.txn_signature.clone().unwrap_or_default(),
                delay_ms,
                timestamp_ms: to_epoch_ms(u.wall_time),
            });
        }
    }
    out
}

fn calc_delay(
    map1: &HashMap<SlotKey, SlotUpdate>,
    map2: &HashMap<SlotKey, SlotUpdate>,
    slot: u64,
    status: SlotStatus,
) -> (Option<Duration>, Option<Duration>) {
    let key = (slot, status);
    match (map1.get(&key), map2.get(&key)) {
        (Some(u1), Some(u2)) => {
            if u1.timestamp < u2.timestamp {
                (Some(Duration::ZERO), Some(u2.timestamp.duration_since(u1.timestamp)))
            } else if u2.timestamp < u1.timestamp {
                (Some(u1.timestamp.duration_since(u2.timestamp)), Some(Duration::ZERO))
            } else {
                (Some(Duration::ZERO), Some(Duration::ZERO))
            }
        }
        _ => (None, None),
    }
}

fn calc_download(map: &HashMap<SlotKey, SlotUpdate>, slot: u64) -> Option<Duration> {
    let first = map.get(&(slot, SlotStatus::FirstShredReceived))?;
    let completed = map.get(&(slot, SlotStatus::Completed))?;
    Some(completed.timestamp.duration_since(first.timestamp))
}

fn calc_replay(map: &HashMap<SlotKey, SlotUpdate>, slot: u64) -> Option<Duration> {
    let bank = map.get(&(slot, SlotStatus::CreatedBank))?;
    let processed = map.get(&(slot, SlotStatus::Processed))?;
    Some(processed.timestamp.duration_since(bank.timestamp))
}

fn calc_confirm(map: &HashMap<SlotKey, SlotUpdate>, slot: u64) -> Option<Duration> {
    let processed = map.get(&(slot, SlotStatus::Processed))?;
    let confirmed = map.get(&(slot, SlotStatus::Confirmed))?;
    Some(confirmed.timestamp.duration_since(processed.timestamp))
}

fn calc_finalize(map: &HashMap<SlotKey, SlotUpdate>, slot: u64) -> Option<Duration> {
    let confirmed = map.get(&(slot, SlotStatus::Confirmed))?;
    let finalized = map.get(&(slot, SlotStatus::Finalized))?;
    Some(finalized.timestamp.duration_since(confirmed.timestamp))
}

fn build_slot_detail(map: &HashMap<SlotKey, SlotUpdate>, slot: u64) -> SlotDetail {
    let stages = SlotStatus::required_stages();
    let mut transitions = Vec::new();

    for s in stages {
        if let Some(u) = map.get(&(slot, *s)) {
            transitions.push(SlotTransition {
                status: format!("{:?}", s),
                timestamp_ms: to_epoch_ms(u.wall_time),
            });
        }
    }

    let dl = calc_download(map, slot).map(|d| d.as_secs_f64() * 1000.0).unwrap_or(0.0);
    let rp = calc_replay(map, slot).map(|d| d.as_secs_f64() * 1000.0).unwrap_or(0.0);
    let cn = calc_confirm(map, slot).map(|d| d.as_secs_f64() * 1000.0).unwrap_or(0.0);
    let fi = calc_finalize(map, slot).map(|d| d.as_secs_f64() * 1000.0).unwrap_or(0.0);

    SlotDetail {
        first_shred_delay_ms: None,
        processing_delay_ms: None,
        confirmation_delay_ms: None,
        finalization_delay_ms: None,
        transitions,
        durations: StageDurations {
            download_ms: dl,
            replay_ms: rp,
            confirm_ms: cn,
            finalize_ms: fi,
        },
        accounts: Vec::new(),
    }
}

pub fn print_summary(result: &GeyserBenchmarkResult) {
    println!("\n=== GEYSER BENCHMARK SUMMARY ===");
    println!("version: {}", result.version);
    println!("with accounts: {}", result.with_accounts);
    println!("slots collected: {}", result.meta.total_slots);
    println!("common slots: {}", result.meta.common_slots);
    println!("compared slots: {}", result.meta.compared_slots);
    println!("dropped slots: {}", result.meta.dropped_slots);
    println!("duration: {}ms", result.meta.duration_ms);

    for (i, p) in result.providers.iter().enumerate() {
        println!("\nProvider {} ({} v{})", i + 1, p.plugin_name, p.plugin_version);
        println!("  url: {}", p.url);
        println!("  ping: {:.2}ms", p.avg_ping_ms);
        println!("  updates: {}", p.total_updates);
        println!("  unique slots: {}", p.unique_slots);
        if let Some(cnt) = p.account_count {
            println!("  account updates: {}", cnt);
        }
    }

    println!("\n=== PROVIDER 1 METRICS (ms) ===");
    print_metrics(&result.provider1_metrics);

    println!("\n=== PROVIDER 2 METRICS (ms) ===");
    print_metrics(&result.provider2_metrics);
}

fn print_metrics(m: &ProviderMetrics) {
    println!("  first shred delay: p50={:.2}, p90={:.2}, p99={:.2}", m.first_shred_delay.p50, m.first_shred_delay.p90, m.first_shred_delay.p99);
    println!("  processing delay:  p50={:.2}, p90={:.2}, p99={:.2}", m.processing_delay.p50, m.processing_delay.p90, m.processing_delay.p99);
    println!("  confirmation delay: p50={:.2}, p90={:.2}, p99={:.2}", m.confirmation_delay.p50, m.confirmation_delay.p90, m.confirmation_delay.p99);
    println!("  finalization delay: p50={:.2}, p90={:.2}, p99={:.2}", m.finalization_delay.p50, m.finalization_delay.p90, m.finalization_delay.p99);
    println!("  download time:     p50={:.2}, p90={:.2}, p99={:.2}", m.download_time.p50, m.download_time.p90, m.download_time.p99);
    println!("  replay time:       p50={:.2}, p90={:.2}, p99={:.2}", m.replay_time.p50, m.replay_time.p90, m.replay_time.p99);
    println!("  confirm time:      p50={:.2}, p90={:.2}, p99={:.2}", m.confirm_time.p50, m.confirm_time.p90, m.confirm_time.p99);
    println!("  finalize time:     p50={:.2}, p90={:.2}, p99={:.2}", m.finalize_time.p50, m.finalize_time.p90, m.finalize_time.p99);
    if let Some(ref ad) = m.account_delay {
        println!("  account delay:     p50={:.2}, p90={:.2}, p99={:.2}", ad.p50, ad.p90, ad.p99);
    }
}
