//! Cross-source NOTAM assembly: deduplicate by a normalized `(location,
//! number)` key, keep the most structured record, surface every source's own
//! version, and classify agreement. The one place NOTAMs from any mix of
//! sources are merged.

use std::cmp::Ordering;
use std::collections::BTreeMap;

use crate::compare::Agreement;

use serde::Serialize;

use super::{NotamRecord, NotamSource};

/// One NOTAM after reconciliation: the chosen record, every source that
/// carried it, each source's own richest version, and whether they agreed.
#[derive(Debug, Clone, PartialEq, Serialize)]
#[non_exhaustive]
pub struct ReconciledNotam {
    /// The most structured record for this `(location, number)`.
    pub notam: NotamRecord,
    /// Every source that reported it, sorted by slug.
    pub sources: Vec<NotamSource>,
    /// Each reporting source's richest record, sorted by slug — so a
    /// disagreement's conflicting content survives rather than being dropped.
    pub records: Vec<NotamRecord>,
    /// Agreement across the reporting sources.
    pub agreement: Agreement,
}

/// The reconciled NOTAM set, deterministically ordered by the normalized
/// `(location, number)` key.
#[derive(Debug, Clone, PartialEq, Serialize)]
#[non_exhaustive]
pub struct NotamReconciliation {
    /// One entry per distinct `(location, number)`.
    pub notams: Vec<ReconciledNotam>,
}

impl NotamReconciliation {
    /// Reconciled NOTAMs only one source carried — worth surfacing, because a
    /// missing NOTAM is a safety gap, not noise.
    pub fn single_source(&self) -> impl Iterator<Item = &ReconciledNotam> {
        self.notams
            .iter()
            .filter(|r| r.agreement == Agreement::Single)
    }

    /// Reconciled NOTAMs whose sources disagreed on the substance.
    pub fn disagreements(&self) -> impl Iterator<Item = &ReconciledNotam> {
        self.notams
            .iter()
            .filter(|r| r.agreement == Agreement::Disagree)
    }
}

/// Assemble records from any mix of sources into one attributed set:
/// deduplicate by a normalized `(location, number)`, keep the most structured
/// record, retain each source's own version, and classify agreement.
/// Duplicates from overlapping queries collapse; a NOTAM two sources both
/// carry is `Agree`/`Disagree`, one source `Single`.
pub fn reconcile(records: Vec<NotamRecord>) -> NotamReconciliation {
    let mut groups: BTreeMap<(String, String), Vec<NotamRecord>> = BTreeMap::new();
    for record in records {
        groups.entry(key(&record)).or_default().push(record);
    }

    // `filter_map` drops an empty group — which `into_values` never yields —
    // rather than `expect`/panic on the always-`Some` reconciliation.
    let notams = groups.into_values().filter_map(reconcile_group).collect();
    NotamReconciliation { notams }
}

/// The normalized cross-source merge key. Location case/whitespace, the
/// contiguous-US `K` ICAO prefix, and the leading zeros of the NOTAM number are
/// folded so the same NOTAM from two sources (`KATL`/`ATL`, `09/186`/`9/186`)
/// lands in one group; the displayed record keeps its raw values.
fn key(record: &NotamRecord) -> (String, String) {
    (
        normalize_location(&record.location),
        normalize_number(&record.number),
    )
}

/// Fold a location for keying: uppercase, trim, and reduce a 4-letter
/// contiguous-US ICAO identifier to its 3-letter FAA form (`KBOS` → `BOS`), so
/// a NOTAM keyed `BOS` by FAA NMS and `KBOS` by Leidos merges. Only the
/// unambiguous `K` prefix is folded; the Alaska/Hawaii/Pacific `P`/`A` prefixes
/// have non-positional LID mappings (`PAJN` → `JNU`) and are left as-is.
fn normalize_location(location: &str) -> String {
    let upper = location.trim().to_ascii_uppercase();
    if upper.len() == 4
        && upper.starts_with('K')
        && upper.as_bytes()[1..].iter().all(u8::is_ascii_alphanumeric)
    {
        upper[1..].to_owned()
    } else {
        upper
    }
}

/// Fold a NOTAM number for keying: uppercase, trim, and strip the leading
/// zeros of each purely-numeric `/`-separated segment (`09/186` → `9/186`).
fn normalize_number(number: &str) -> String {
    number
        .trim()
        .to_ascii_uppercase()
        .split('/')
        .map(|segment| {
            let segment = segment.trim();
            if !segment.is_empty() && segment.bytes().all(|b| b.is_ascii_digit()) {
                let trimmed = segment.trim_start_matches('0');
                if trimmed.is_empty() {
                    "0".to_owned()
                } else {
                    trimmed.to_owned()
                }
            } else {
                segment.to_owned()
            }
        })
        .collect::<Vec<_>>()
        .join("/")
}

/// Collapse one key-group into a [`ReconciledNotam`]: one richest
/// representative per source, the overall richest as the headline, and the
/// agreement across sources. `None` only for an empty group, which
/// `into_values` never yields.
fn reconcile_group(group: Vec<NotamRecord>) -> Option<ReconciledNotam> {
    let mut per_source: BTreeMap<String, NotamRecord> = BTreeMap::new();
    for record in group {
        let slug = record.source.slug().to_owned();
        let replace = per_source
            .get(&slug)
            .is_none_or(|existing| richness_ordering(&record, existing) == Ordering::Greater);
        if replace {
            per_source.insert(slug, record);
        }
    }
    // BTreeMap iterates in slug order, so `records`/`sources` are slug-sorted.
    let records: Vec<NotamRecord> = per_source.into_values().collect();
    let sources: Vec<NotamSource> = records.iter().map(|r| r.source.clone()).collect();
    let agreement = classify(&records);
    let notam = records
        .iter()
        .max_by(|a, b| richness_ordering(a, b))
        .cloned()?;
    Some(ReconciledNotam {
        notam,
        sources,
        records,
        agreement,
    })
}

/// A total order by "richness": lowest source fidelity rank, then most
/// populated fields, then the longest then lexically-greatest text, then
/// source slug. Total (not just partial) so the winner never depends on input
/// order. `max_by` picks the greatest.
fn richness_ordering(a: &NotamRecord, b: &NotamRecord) -> Ordering {
    b.source
        .fidelity_rank()
        .cmp(&a.source.fidelity_rank())
        .then_with(|| a.richness().cmp(&b.richness()))
        .then_with(|| a.raw_text.len().cmp(&b.raw_text.len()))
        .then_with(|| a.raw_text.cmp(&b.raw_text))
        .then_with(|| a.source.slug().cmp(b.source.slug()))
}

/// `Single` for one source; otherwise `Agree` unless two sources that both
/// populate a structured field conflict on it.
///
/// The `(location, number)` key already proves the records are the same NOTAM,
/// so free text — whose framing differs by source (NMS carries the bare body;
/// Leidos prefixes the coded `!acct number location` header) — is not compared.
/// Only a concrete conflict in a field both sources carry (classification,
/// validity, flight levels, Q-code) is a disagreement; a field absent from a
/// source is no opinion, not a conflict. A source that carries no structured
/// fields (Leidos) therefore confirms rather than contradicts.
fn classify(records: &[NotamRecord]) -> Agreement {
    if records.len() < 2 {
        return Agreement::Single;
    }
    let conflict = conflicts(records.iter().filter_map(|r| r.classification.clone()))
        || conflicts(records.iter().filter_map(|r| r.valid))
        || conflicts(records.iter().filter_map(|r| r.min_fl))
        || conflicts(records.iter().filter_map(|r| r.max_fl))
        || conflicts(records.iter().filter_map(|r| r.q_code.clone()));
    if conflict {
        Agreement::Disagree
    } else {
        Agreement::Agree
    }
}

/// Whether the supplied (present-only) values disagree: more than one distinct
/// value among the sources that populated the field.
fn conflicts<T: PartialEq>(values: impl Iterator<Item = T>) -> bool {
    let mut seen: Option<T> = None;
    for value in values {
        match &seen {
            Some(first) if *first != value => return true,
            None => seen = Some(value),
            _ => {}
        }
    }
    false
}
