dataprof-metrics 0.12.0

use std::collections::HashMap;

use dataprof_core::{ColumnProfile, QualityDimension, QualityScoreWeights};
use serde::{Deserialize, Serialize};

use crate::core::errors::DataProfilerError;

/// Completeness metrics (ISO 8000-8).
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, schemars::JsonSchema)]
pub struct CompletenessMetrics {
    #[serde(serialize_with = "crate::serde_helpers::round_2")]
    pub missing_values_ratio: f64,
    /// Percentage of records in which every field is present, counted over
    /// every row the engine read. Strict by design: a record missing one
    /// optional field is not complete, so read it beside `null_columns`,
    /// which names the columns responsible.
    #[serde(serialize_with = "crate::serde_helpers::round_2")]
    pub complete_records_ratio: f64,
    pub null_columns: Vec<String>,
    /// Total cells examined (rows × columns). 0 means the dimension had
    /// nothing to assess and it is excluded from the overall score.
    #[serde(default)]
    pub total_cells: usize,
}

/// Consistency metrics (ISO 8000-61).
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, schemars::JsonSchema)]
pub struct ConsistencyMetrics {
    #[serde(serialize_with = "crate::serde_helpers::round_2")]
    pub data_type_consistency: f64,
    pub format_violations: usize,
    pub encoding_issues: usize,
    /// Non-null values examined for type consistency. 0 means the dimension
    /// had nothing to assess and it is excluded from the overall score.
    #[serde(default)]
    pub values_checked: usize,
}

/// Uniqueness metrics (ISO 8000-110).
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, schemars::JsonSchema)]
pub struct UniquenessMetrics {
    pub duplicate_rows: usize,
    #[serde(serialize_with = "crate::serde_helpers::round_2")]
    pub key_uniqueness: f64,
    pub high_cardinality_warning: bool,
    /// Rows scanned for exact duplicates. 0 means the dimension had nothing
    /// to assess and it is excluded from the overall score.
    #[serde(default)]
    pub rows_checked: usize,
    /// Column whose uniqueness `key_uniqueness` describes. `None` means no
    /// key column was identified; `key_uniqueness` then carries no signal
    /// and does not contribute to the dimension score.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub key_column: Option<String>,
    /// True when `duplicate_rows` comes from the full-stream distinct-count
    /// estimator after it spilled to its HLL sketch (~1% relative error on
    /// the distinct count), rather than an exact count.
    #[serde(default, skip_serializing_if = "is_false")]
    pub duplicate_rows_approximate: bool,
}

/// Full-stream row-duplicate counts produced by an engine's row tracker.
///
/// Engines that see whole records (CSV, JSON, streaming readers) count
/// duplicates over *every* row with bounded memory: exact below the
/// distinct-row threshold, HLL-estimated (and flagged) beyond it. When
/// available this supersedes the sample-based duplicate scan, which cannot
/// run at all on misaligned per-column samples.
#[derive(Debug, Clone, Copy)]
pub struct RowDuplicateSummary {
    pub duplicate_rows: usize,
    pub rows_checked: usize,
    pub approximate: bool,
    /// The most duplicate rows there can be: `rows_checked` minus the distinct
    /// rows the tracker's exact set held when it was dropped. Equal to
    /// `duplicate_rows` while the count is exact (#819).
    pub max_duplicate_rows: usize,
}

/// Full-stream complete-record counts produced by an engine's row tracker.
///
/// A "complete" record is one whose every field is present. That is a
/// property of a row, so it can only be counted while whole rows are in
/// hand; per-column null totals cannot recover it, because they do not say
/// whether two nulls landed in the same record. Engines that see whole
/// records accumulate this exactly over the full stream with one counter,
/// independent of sampling.
#[derive(Debug, Clone, Copy)]
pub struct RowCompletenessSummary {
    pub complete_rows: usize,
    pub rows_checked: usize,
}

/// Accuracy metrics (ISO 25012).
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, schemars::JsonSchema)]
pub struct AccuracyMetrics {
    #[serde(serialize_with = "crate::serde_helpers::round_2")]
    pub outlier_ratio: f64,
    pub range_violations: usize,
    pub negative_values_in_positive: usize,
    /// Finite numeric values examined across all columns. 0 means the
    /// dimension had nothing to assess and it is excluded from the overall
    /// score.
    #[serde(default)]
    pub numeric_values_checked: usize,
}

/// Timeliness metrics (ISO 8000-8).
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, schemars::JsonSchema)]
pub struct TimelinessMetrics {
    pub future_dates_count: usize,
    #[serde(serialize_with = "crate::serde_helpers::round_2")]
    pub stale_data_ratio: f64,
    pub temporal_violations: usize,
    /// Non-null values in inferred or explicitly configured temporal columns
    /// that failed calendar-date parsing.
    #[serde(default)]
    pub invalid_date_values: usize,
    /// Non-null values examined in inferred or explicitly configured temporal
    /// columns. 0 means
    /// the dimension had nothing to assess and it is excluded from the
    /// overall score.
    #[serde(default)]
    pub date_values_checked: usize,
    /// Start/end value pairs actually compared for temporal ordering.
    /// `temporal_violations` is bounded by this, not by
    /// `date_values_checked` — the pair scan may cover columns the date
    /// scan does not.
    #[serde(default)]
    pub temporal_pairs_checked: usize,
}

/// Validity metrics derived from confidently detected semantic patterns.
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, schemars::JsonSchema)]
pub struct ValidityMetrics {
    #[serde(serialize_with = "crate::serde_helpers::round_2")]
    pub valid_values_ratio: f64,
    pub invalid_values: usize,
    /// Non-null values checked against a dominant semantic pattern.
    #[serde(default)]
    pub values_checked: usize,
}

/// Precision metrics for effective decimal-scale consistency.
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, schemars::JsonSchema)]
pub struct PrecisionMetrics {
    #[serde(serialize_with = "crate::serde_helpers::round_2")]
    pub decimal_places_consistency: f64,
    pub inconsistent_precision_values: usize,
    /// Parseable finite values examined in floating-point columns.
    #[serde(default)]
    pub numeric_values_checked: usize,
}

/// Whether a dimension's metrics were computed against anything.
///
/// Every dimension carries the counter it divides by, and a zero there means
/// the dimension had nothing to assess: no cells, no non-null values, no
/// numeric values, no dates, no confidently detected pattern. Its ratios are
/// then defaults rather than measurements, so they are withheld instead of
/// reported. `None` means "not analyzed"; it is not a score of 100.
///
/// This is the single definition of "assessed" behind the dimension scores,
/// the serialized report and the Python evidence accessors.
///
/// [`QualityMetrics::assessed_dimensions`] applies it and then narrows further,
/// to the dimensions carrying a positive score weight, because it answers what
/// is behind [`QualityMetrics::overall_score`] rather than what was measured. A
/// dimension weighted `0.0` is assessed and reported while contributing nothing
/// to the aggregate.
pub(crate) trait Assessed {
    fn assessed(&self) -> bool;
}

/// Serde predicate: skip a dimension that assessed nothing.
fn is_unassessed<T: Assessed>(metrics: &Option<T>) -> bool {
    metrics.as_ref().is_none_or(|metrics| !metrics.assessed())
}

impl CompletenessMetrics {
    /// True when at least one cell was examined.
    pub fn is_assessed(&self) -> bool {
        self.total_cells > 0
    }
}

impl ConsistencyMetrics {
    /// True when at least one non-null value was examined.
    pub fn is_assessed(&self) -> bool {
        self.values_checked > 0
    }
}

impl UniquenessMetrics {
    /// True when either component had data: rows scanned for duplicates, or an
    /// identified key column whose uniqueness `key_uniqueness` describes.
    pub fn is_assessed(&self) -> bool {
        self.rows_checked > 0 || self.key_column.is_some()
    }
}

impl AccuracyMetrics {
    /// True when at least one finite numeric value was examined.
    pub fn is_assessed(&self) -> bool {
        self.numeric_values_checked > 0
    }
}

impl TimelinessMetrics {
    /// True when at least one value in a temporal column was examined.
    pub fn is_assessed(&self) -> bool {
        self.date_values_checked > 0
    }
}

impl ValidityMetrics {
    /// True when at least one value was checked against a dominant pattern.
    pub fn is_assessed(&self) -> bool {
        self.values_checked > 0
    }
}

impl PrecisionMetrics {
    /// True when at least one parseable float was examined.
    pub fn is_assessed(&self) -> bool {
        self.numeric_values_checked > 0
    }
}

macro_rules! impl_assessed {
    ($($metrics:ty),+ $(,)?) => {
        $(impl Assessed for $metrics {
            fn assessed(&self) -> bool {
                self.is_assessed()
            }
        })+
    };
}

impl_assessed!(
    CompletenessMetrics,
    ConsistencyMetrics,
    UniquenessMetrics,
    AccuracyMetrics,
    TimelinessMetrics,
    ValidityMetrics,
    PrecisionMetrics,
);

/// Comprehensive data quality metrics following industry standards.
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize, schemars::JsonSchema)]
pub struct QualityMetrics {
    #[serde(skip_serializing_if = "is_unassessed")]
    pub completeness: Option<CompletenessMetrics>,
    #[serde(skip_serializing_if = "is_unassessed")]
    pub consistency: Option<ConsistencyMetrics>,
    #[serde(skip_serializing_if = "is_unassessed")]
    pub uniqueness: Option<UniquenessMetrics>,
    #[serde(skip_serializing_if = "is_unassessed")]
    pub accuracy: Option<AccuracyMetrics>,
    #[serde(skip_serializing_if = "is_unassessed")]
    pub timeliness: Option<TimelinessMetrics>,
    #[serde(skip_serializing_if = "is_unassessed")]
    pub validity: Option<ValidityMetrics>,
    #[serde(skip_serializing_if = "is_unassessed")]
    pub precision: Option<PrecisionMetrics>,
    /// True when the sample used to compute these metrics was below the
    /// minimum recommended size (10 rows). When set, the quality scores and
    /// per-dimension ratios should be treated as directional rather than
    /// reliable. Backwards-compatible: defaults to `false`.
    #[serde(default, skip_serializing_if = "is_false")]
    pub low_sample_warning: bool,
    /// Weights used to aggregate dimension scores. Default weights are omitted
    /// from serialized reports; custom weights are retained for reproducible
    /// score calculation after a round trip.
    #[serde(default, skip_serializing_if = "QualityScoreWeights::is_default")]
    pub score_weights: QualityScoreWeights,
}

fn is_false(b: &bool) -> bool {
    !*b
}

impl QualityMetrics {
    pub fn empty() -> Self {
        Self {
            completeness: Some(CompletenessMetrics {
                missing_values_ratio: 0.0,
                complete_records_ratio: 100.0,
                null_columns: vec![],
                total_cells: 0,
            }),
            consistency: Some(ConsistencyMetrics {
                data_type_consistency: 100.0,
                format_violations: 0,
                encoding_issues: 0,
                values_checked: 0,
            }),
            uniqueness: Some(UniquenessMetrics {
                duplicate_rows: 0,
                key_uniqueness: 100.0,
                high_cardinality_warning: false,
                rows_checked: 0,
                key_column: None,
                duplicate_rows_approximate: false,
            }),
            accuracy: Some(AccuracyMetrics {
                outlier_ratio: 0.0,
                range_violations: 0,
                negative_values_in_positive: 0,
                numeric_values_checked: 0,
            }),
            timeliness: Some(TimelinessMetrics {
                future_dates_count: 0,
                stale_data_ratio: 0.0,
                temporal_violations: 0,
                invalid_date_values: 0,
                date_values_checked: 0,
                temporal_pairs_checked: 0,
            }),
            validity: Some(ValidityMetrics {
                valid_values_ratio: 100.0,
                invalid_values: 0,
                values_checked: 0,
            }),
            precision: Some(PrecisionMetrics {
                decimal_places_consistency: 100.0,
                inconsistent_precision_values: 0,
                numeric_values_checked: 0,
            }),
            low_sample_warning: false,
            score_weights: QualityScoreWeights::default(),
        }
    }

    pub fn calculate_from_data(
        data: &HashMap<String, Vec<String>>,
        column_profiles: &[ColumnProfile],
    ) -> Result<Self, DataProfilerError> {
        let calculator = crate::analysis::MetricsCalculator::new();
        calculator.calculate_comprehensive_metrics(data, column_profiles, None)
    }

    /// Score for the completeness dimension (0-100), or `None` when the
    /// dimension was not computed or had no cells to assess.
    ///
    /// Mean of cell-level completeness (`100 - missing_values_ratio`) and
    /// row-level completeness (`complete_records_ratio`).
    pub fn completeness_score(&self) -> Option<f64> {
        let c = self.completeness.as_ref().filter(|c| c.is_assessed())?;
        let cell_level = 100.0 - c.missing_values_ratio;
        Some(((cell_level + c.complete_records_ratio) / 2.0).clamp(0.0, 100.0))
    }

    /// Score for the consistency dimension (0-100), or `None` when the
    /// dimension was not computed or had no non-null values to assess.
    ///
    /// Type consistency, penalized by format violations and encoding issues
    /// as a share of the values checked.
    pub fn consistency_score(&self) -> Option<f64> {
        let c = self.consistency.as_ref().filter(|c| c.is_assessed())?;
        let violation_ratio =
            (c.format_violations + c.encoding_issues) as f64 / c.values_checked as f64;
        Some((c.data_type_consistency - violation_ratio * 100.0).clamp(0.0, 100.0))
    }

    /// Score for the uniqueness dimension (0-100), or `None` when the
    /// dimension was not computed or neither component had data.
    ///
    /// Mean of the available components: share of non-duplicate rows (when
    /// a row tracker or an aligned sample scan produced a count) and
    /// `key_uniqueness` (when a key column was identified). Engines without
    /// a row tracker whose samples cannot be proven row-aligned contribute
    /// only the key component.
    pub fn uniqueness_score(&self) -> Option<f64> {
        let u = self.uniqueness.as_ref().filter(|u| u.is_assessed())?;
        let duplicate_score = (u.rows_checked > 0)
            .then(|| (1.0 - u.duplicate_rows as f64 / u.rows_checked as f64) * 100.0);
        let key_score = u.key_column.is_some().then_some(u.key_uniqueness);

        let (sum, count) = [duplicate_score, key_score]
            .iter()
            .flatten()
            .fold((0.0, 0u32), |(sum, count), score| (sum + score, count + 1));
        if count == 0 {
            return None;
        }
        Some((sum / count as f64).clamp(0.0, 100.0))
    }

    /// Score for the accuracy dimension (0-100), or `None` when the
    /// dimension was not computed or no numeric values were found.
    ///
    /// `100 - outlier_ratio`, penalized by range violations and negative
    /// values in positive-only columns as a share of the numeric values
    /// checked.
    pub fn accuracy_score(&self) -> Option<f64> {
        let a = self.accuracy.as_ref().filter(|a| a.is_assessed())?;
        let violation_ratio = (a.range_violations + a.negative_values_in_positive) as f64
            / a.numeric_values_checked as f64;
        Some((100.0 - a.outlier_ratio - violation_ratio * 100.0).clamp(0.0, 100.0))
    }

    /// Score for the timeliness dimension (0-100), or `None` when the
    /// dimension was not computed or no values were found in inferred or
    /// explicitly configured temporal columns.
    ///
    /// `100 - stale_data_ratio`, penalized by future dates as a share of
    /// the date values checked and by temporal ordering violations as a
    /// share of the pairs actually compared.
    pub fn timeliness_score(&self) -> Option<f64> {
        let t = self.timeliness.as_ref().filter(|t| t.is_assessed())?;
        let value_violation_ratio =
            (t.future_dates_count + t.invalid_date_values) as f64 / t.date_values_checked as f64;
        let temporal_ratio = if t.temporal_pairs_checked > 0 {
            t.temporal_violations as f64 / t.temporal_pairs_checked as f64
        } else {
            0.0
        };
        Some(
            (100.0 - t.stale_data_ratio - (value_violation_ratio + temporal_ratio) * 100.0)
                .clamp(0.0, 100.0),
        )
    }

    /// Score for semantic-pattern validity (0-100), or `None` when no column
    /// had a confidently detected pattern to validate.
    pub fn validity_score(&self) -> Option<f64> {
        let validity = self.validity.as_ref().filter(|v| v.is_assessed())?;
        Some(validity.valid_values_ratio.clamp(0.0, 100.0))
    }

    /// Score for decimal-scale precision consistency (0-100), or `None` when
    /// no floating-point values were available to assess.
    pub fn precision_score(&self) -> Option<f64> {
        let precision = self.precision.as_ref().filter(|p| p.is_assessed())?;
        Some(precision.decimal_places_consistency.clamp(0.0, 100.0))
    }

    /// Weighted components of the overall score: `(dimension, weight, score)`.
    fn weighted_scores(&self) -> [(QualityDimension, f64, Option<f64>); 7] {
        [
            (
                QualityDimension::Completeness,
                self.score_weights.completeness,
                self.completeness_score(),
            ),
            (
                QualityDimension::Consistency,
                self.score_weights.consistency,
                self.consistency_score(),
            ),
            (
                QualityDimension::Uniqueness,
                self.score_weights.uniqueness,
                self.uniqueness_score(),
            ),
            (
                QualityDimension::Accuracy,
                self.score_weights.accuracy,
                self.accuracy_score(),
            ),
            (
                QualityDimension::Timeliness,
                self.score_weights.timeliness,
                self.timeliness_score(),
            ),
            (
                QualityDimension::Validity,
                self.score_weights.validity,
                self.validity_score(),
            ),
            (
                QualityDimension::Precision,
                self.score_weights.precision,
                self.precision_score(),
            ),
        ]
    }

    /// One dimension's score (0-100), or `None` when it had nothing to
    /// assess. Weight-independent: a dimension weighted `0.0` still reports
    /// the score it measured, it just does not move the aggregate.
    pub fn dimension_score(&self, dimension: QualityDimension) -> Option<f64> {
        match dimension {
            QualityDimension::Completeness => self.completeness_score(),
            QualityDimension::Consistency => self.consistency_score(),
            QualityDimension::Uniqueness => self.uniqueness_score(),
            QualityDimension::Accuracy => self.accuracy_score(),
            QualityDimension::Timeliness => self.timeliness_score(),
            QualityDimension::Validity => self.validity_score(),
            QualityDimension::Precision => self.precision_score(),
        }
    }

    /// Dimensions that were computed *and* had data to assess. Only these
    /// contribute to [`overall_score`](Self::overall_score).
    pub fn assessed_dimensions(&self) -> Vec<QualityDimension> {
        self.weighted_scores()
            .iter()
            .filter(|(_, weight, score)| *weight > 0.0 && score.is_some())
            .map(|(dim, _, _)| *dim)
            .collect()
    }

    /// Overall quality score (0-100): weighted average of the assessed
    /// dimension scores, with weights renormalized over the assessed
    /// dimensions. A dimension with nothing to assess (no numeric values,
    /// no date columns, ...) is excluded instead of counting as perfect.
    ///
    /// `None` when no dimension was assessable — the renormalization has no
    /// remainder to average, and an empty set of evidence is "not analyzed",
    /// not a score of zero. Equivalent to
    /// [`assessed_dimensions`](Self::assessed_dimensions) being empty.
    pub fn overall_score(&self) -> Option<f64> {
        let mut total_weight = 0.0;
        let mut score = 0.0;

        for (_, weight, dimension_score) in self.weighted_scores() {
            if let Some(value) = dimension_score {
                total_weight += weight;
                score += value * weight;
            }
        }

        (total_weight > 0.0).then(|| (score / total_weight).min(100.0))
    }

    pub fn missing_values_ratio(&self) -> f64 {
        self.completeness
            .as_ref()
            .map_or(0.0, |c| c.missing_values_ratio)
    }

    pub fn complete_records_ratio(&self) -> f64 {
        self.completeness
            .as_ref()
            .map_or(100.0, |c| c.complete_records_ratio)
    }

    pub fn null_columns(&self) -> &[String] {
        self.completeness.as_ref().map_or(&[], |c| &c.null_columns)
    }

    pub fn data_type_consistency(&self) -> f64 {
        self.consistency
            .as_ref()
            .map_or(100.0, |c| c.data_type_consistency)
    }

    pub fn format_violations(&self) -> usize {
        self.consistency.as_ref().map_or(0, |c| c.format_violations)
    }

    pub fn encoding_issues(&self) -> usize {
        self.consistency.as_ref().map_or(0, |c| c.encoding_issues)
    }

    pub fn duplicate_rows(&self) -> usize {
        self.uniqueness.as_ref().map_or(0, |u| u.duplicate_rows)
    }

    pub fn key_uniqueness(&self) -> f64 {
        self.uniqueness.as_ref().map_or(100.0, |u| u.key_uniqueness)
    }

    pub fn high_cardinality_warning(&self) -> bool {
        self.uniqueness
            .as_ref()
            .is_some_and(|u| u.high_cardinality_warning)
    }

    pub fn outlier_ratio(&self) -> f64 {
        self.accuracy.as_ref().map_or(0.0, |a| a.outlier_ratio)
    }

    pub fn range_violations(&self) -> usize {
        self.accuracy.as_ref().map_or(0, |a| a.range_violations)
    }

    pub fn negative_values_in_positive(&self) -> usize {
        self.accuracy
            .as_ref()
            .map_or(0, |a| a.negative_values_in_positive)
    }

    pub fn future_dates_count(&self) -> usize {
        self.timeliness.as_ref().map_or(0, |t| t.future_dates_count)
    }

    pub fn stale_data_ratio(&self) -> f64 {
        self.timeliness.as_ref().map_or(0.0, |t| t.stale_data_ratio)
    }

    pub fn temporal_violations(&self) -> usize {
        self.timeliness
            .as_ref()
            .map_or(0, |t| t.temporal_violations)
    }

    pub fn invalid_date_values(&self) -> usize {
        self.timeliness
            .as_ref()
            .map_or(0, |t| t.invalid_date_values)
    }

    pub fn valid_values_ratio(&self) -> f64 {
        self.validity
            .as_ref()
            .map_or(100.0, |v| v.valid_values_ratio)
    }

    pub fn invalid_values(&self) -> usize {
        self.validity.as_ref().map_or(0, |v| v.invalid_values)
    }

    pub fn decimal_places_consistency(&self) -> f64 {
        self.precision
            .as_ref()
            .map_or(100.0, |p| p.decimal_places_consistency)
    }

    pub fn inconsistent_precision_values(&self) -> usize {
        self.precision
            .as_ref()
            .map_or(0, |p| p.inconsistent_precision_values)
    }

    pub fn supports_dimension(&self, dimension: QualityDimension) -> bool {
        match dimension {
            QualityDimension::Completeness => self.completeness.is_some(),
            QualityDimension::Consistency => self.consistency.is_some(),
            QualityDimension::Uniqueness => self.uniqueness.is_some(),
            QualityDimension::Accuracy => self.accuracy.is_some(),
            QualityDimension::Timeliness => self.timeliness.is_some(),
            QualityDimension::Validity => self.validity.is_some(),
            QualityDimension::Precision => self.precision.is_some(),
        }
    }
}

/// Confidence level for quality metrics.
#[derive(Debug, Clone, Serialize, Deserialize, schemars::JsonSchema)]
pub enum MetricConfidence {
    Exact,
    Approximate {
        sample_size: usize,
        population_size: Option<usize>,
    },
    Mixed {
        exact_dimensions: Vec<String>,
        sampled_dimensions: Vec<String>,
        sample_size: usize,
    },
    /// No dimension had anything to assess, so there is no score for the
    /// other variants to describe. Reporting `Exact` here would claim
    /// certainty about a number that was never computed.
    NotAssessed,
    /// The document this assessment was read from does not say how its
    /// numbers were obtained. Only produced by deserializing a report written
    /// before dataprof recorded confidence; a profiling run always records it.
    ///
    /// Distinct from every other variant on purpose. Reading an unrecorded
    /// provenance as `Exact` would let a decision about the whole source rest
    /// on numbers that may have come from a sample.
    Unrecorded,
}

/// Aggregate scores retained alongside rounded metric inputs in saved reports.
/// Recomputing these from rounded ratios can change a threshold decision.
#[derive(Debug, Clone, Serialize, Deserialize, schemars::JsonSchema)]
pub struct QualityScores {
    #[serde(serialize_with = "dataprof_core::serde_helpers::round_2_opt")]
    pub overall_score: Option<f64>,
    #[serde(serialize_with = "serialize_dimension_scores")]
    pub dimension_scores: std::collections::BTreeMap<String, Option<f64>>,
}

fn serialize_dimension_scores<S: serde::Serializer>(
    scores: &std::collections::BTreeMap<String, Option<f64>>,
    serializer: S,
) -> Result<S::Ok, S::Error> {
    #[derive(Serialize)]
    struct Score(
        #[serde(serialize_with = "dataprof_core::serde_helpers::round_2_opt")] Option<f64>,
    );
    scores
        .iter()
        .map(|(name, value)| (name, Score(*value)))
        .collect::<std::collections::BTreeMap<_, _>>()
        .serialize(serializer)
}

impl QualityScores {
    fn from_metrics(metrics: &QualityMetrics) -> Self {
        Self {
            overall_score: metrics.overall_score(),
            dimension_scores: QualityDimension::all()
                .into_iter()
                .map(|dimension| (dimension.to_string(), metrics.dimension_score(dimension)))
                .collect(),
        }
    }
}

/// A range a score lies in, on the 0-100 scale.
#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize, schemars::JsonSchema)]
pub struct ScoreInterval {
    #[serde(serialize_with = "dataprof_core::serde_helpers::round_2")]
    pub lower: f64,
    #[serde(serialize_with = "dataprof_core::serde_helpers::round_2")]
    pub upper: f64,
}

impl ScoreInterval {
    fn is_valid(&self) -> bool {
        self.lower.is_finite()
            && self.upper.is_finite()
            && 0.0 <= self.lower
            && self.lower <= self.upper
            && self.upper <= 100.0
    }
}

/// Where the scores over every scanned row lie, for an assessment whose
/// scores were partly computed over a retained sample of those rows.
///
/// The sample fixes each check: a column's type, dominant form and decimal
/// scale, detected pattern and outlier fences. The intervals cover the scores
/// those same checks would give over every scanned value, and all of them
/// hold at once with probability at least `confidence_level`. When the scan
/// read the whole source, that is the whole source's score, which is the
/// only case a quality gate decides on them. Intervals
/// are rounded outward to two decimals, so a saved one still contains what
/// the computed one did.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize, schemars::JsonSchema)]
pub struct ScoreBounds {
    /// Probability that every interval here holds at once.
    pub confidence_level: f64,
    /// Interval for the overall score. `None` when there is no overall score,
    /// or when a sampled dimension behind it has no bound.
    pub overall_score: Option<ScoreInterval>,
    /// Interval per dimension, keyed like `QualityScores::dimension_scores`.
    /// An exactly computed dimension's interval is its score. `None` for a
    /// dimension that was not assessed, or a sampled one with no bound: an
    /// estimated key count or a duplicate scan over a sample. Start/end
    /// ordering is compared only between date columns whose samples hold the
    /// same rows, so a pair where either column has nulls is not compared.
    pub dimension_scores: std::collections::BTreeMap<String, Option<ScoreInterval>>,
}

impl ScoreBounds {
    fn is_valid(&self) -> bool {
        self.confidence_level.is_finite()
            && 0.0 < self.confidence_level
            && self.confidence_level < 1.0
            && self
                .overall_score
                .is_none_or(|interval| interval.is_valid())
            && self
                .dimension_scores
                .values()
                .all(|interval| interval.is_none_or(|interval| interval.is_valid()))
    }
}

/// Wraps quality metrics with confidence information.
#[derive(Debug, Clone, schemars::JsonSchema)]
pub struct QualityAssessment {
    pub metrics: QualityMetrics,
    pub confidence: MetricConfidence,
    /// Scores from a saved document. Absent on older documents; those derive
    /// their scores from the metric values they actually recorded.
    #[schemars(default)]
    scores: Option<QualityScores>,
    #[schemars(skip)]
    score_inputs: Option<Box<QualityMetrics>>,
    /// Bounds on the scores over every scanned row, present only when some score was
    /// computed over a retained sample.
    #[schemars(default)]
    score_bounds: Option<ScoreBounds>,
    #[schemars(skip)]
    bounds_inputs: Option<Box<QualityMetrics>>,
}

impl Serialize for QualityAssessment {
    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
        #[derive(Serialize)]
        struct Document<'a> {
            metrics: &'a QualityMetrics,
            confidence: &'a MetricConfidence,
            scores: QualityScores,
            #[serde(skip_serializing_if = "Option::is_none")]
            score_bounds: Option<&'a ScoreBounds>,
        }
        Document {
            metrics: &self.metrics,
            confidence: &self.confidence,
            scores: self.scores(),
            score_bounds: self.score_bounds(),
        }
        .serialize(serializer)
    }
}

impl<'de> Deserialize<'de> for QualityAssessment {
    fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
        #[derive(Deserialize)]
        struct Document {
            metrics: QualityMetrics,
            confidence: MetricConfidence,
            #[serde(default)]
            scores: Option<QualityScores>,
            #[serde(default)]
            score_bounds: Option<ScoreBounds>,
        }
        let document = Document::deserialize(deserializer)?;
        if document
            .score_bounds
            .as_ref()
            .is_some_and(|bounds| !bounds.is_valid())
        {
            return Err(serde::de::Error::custom(
                "quality score bounds must lie in 0..=100 with lower <= upper, at a confidence level between 0 and 1",
            ));
        }
        if let Some(scores) = &document.scores {
            let in_range = |score: Option<f64>| {
                score.is_none_or(|value| value.is_finite() && (0.0..=100.0).contains(&value))
            };
            let valid_score =
                |score: Option<f64>, assessed: bool| score.is_some() == assessed && in_range(score);
            // Scores for dimensions this build does not know are retained, so
            // the range applies to every entry, not only the known ones.
            if !scores
                .dimension_scores
                .values()
                .all(|score| in_range(*score))
                || !valid_score(
                    scores.overall_score,
                    document.metrics.overall_score().is_some(),
                )
                || QualityDimension::all().into_iter().any(|dimension| {
                    scores
                        .dimension_scores
                        .get(&dimension.to_string())
                        .is_none_or(|score| {
                            !valid_score(
                                *score,
                                document.metrics.dimension_score(dimension).is_some(),
                            )
                        })
                })
            {
                return Err(serde::de::Error::custom(
                    "quality scores must include every dimension, preserve unassessed scores as null, and lie in 0..=100",
                ));
            }
        }
        let score_inputs = if document.scores.is_some() {
            Some(Box::new(document.metrics.clone()))
        } else {
            None
        };
        let bounds_inputs = document
            .score_bounds
            .as_ref()
            .map(|_| Box::new(document.metrics.clone()));
        Ok(Self {
            metrics: document.metrics,
            confidence: document.confidence,
            scores: document.scores,
            score_inputs,
            score_bounds: document.score_bounds,
            bounds_inputs,
        })
    }
}

impl QualityAssessment {
    /// Pair metrics with the confidence of the run that produced them.
    ///
    /// `confidence` is downgraded to [`MetricConfidence::NotAssessed`] when
    /// no dimension was assessable: how the numbers were obtained is only
    /// meaningful when there are numbers.
    ///
    /// "Assessable" here is weight-independent. `assessed_dimensions()`
    /// narrows to the dimensions carrying a positive score weight, because it
    /// answers what is behind the aggregate; provenance is not about the
    /// aggregate. A dimension weighted `0.0` is still measured and still
    /// reported, and a consumer deciding on it needs to know whether its
    /// number came from a sample.
    pub fn new(metrics: QualityMetrics, confidence: MetricConfidence) -> Self {
        let measured = QualityDimension::all()
            .into_iter()
            .any(|dimension| metrics.dimension_score(dimension).is_some());
        let confidence = if measured {
            confidence
        } else {
            MetricConfidence::NotAssessed
        };
        Self {
            metrics,
            confidence,
            scores: None,
            score_inputs: None,
            score_bounds: None,
            bounds_inputs: None,
        }
    }

    /// Attach bounds on the scores over every scanned row, computed from the same
    /// sample as these metrics.
    pub fn with_score_bounds(mut self, bounds: Option<ScoreBounds>) -> Self {
        self.bounds_inputs = bounds.as_ref().map(|_| Box::new(self.metrics.clone()));
        self.score_bounds = bounds;
        self
    }

    /// Bounds on the scores over every scanned row, when some score was computed over a
    /// retained sample and the bounds were recorded.
    ///
    /// `None` once `metrics` has been changed: bounds describe the numbers
    /// they were computed with, and would otherwise vouch for edited ones.
    pub fn score_bounds(&self) -> Option<&ScoreBounds> {
        match (&self.score_bounds, &self.bounds_inputs) {
            (Some(bounds), Some(inputs)) if &self.metrics == inputs.as_ref() => Some(bounds),
            _ => None,
        }
    }

    pub fn exact(metrics: QualityMetrics) -> Self {
        Self::new(metrics, MetricConfidence::Exact)
    }

    pub fn approximate(
        metrics: QualityMetrics,
        sample_size: usize,
        population_size: Option<usize>,
    ) -> Self {
        Self::new(
            metrics,
            MetricConfidence::Approximate {
                sample_size,
                population_size,
            },
        )
    }

    /// Overall quality score, or `None` when no dimension was assessable.
    /// See [`QualityMetrics::overall_score`].
    pub fn score(&self) -> Option<f64> {
        self.scores().overall_score
    }

    /// Original aggregate scores for unchanged restored metrics; otherwise
    /// compute from the current metrics. The input snapshot prevents a caller
    /// mutating public `metrics` from accidentally retaining stale scores.
    pub fn scores(&self) -> QualityScores {
        if let (Some(scores), Some(inputs)) = (&self.scores, &self.score_inputs)
            && &self.metrics == inputs.as_ref()
        {
            return scores.clone();
        }
        QualityScores::from_metrics(&self.metrics)
    }

    /// Labels of the metric components computed from a retained sample of the
    /// scanned rows rather than from every one of them.
    ///
    /// A bounded quality sample is how a profiler keeps memory flat over a
    /// large source, so a fully scanned file can still carry sampled quality
    /// numbers. Anything deciding about the *whole source* needs to know
    /// which: a ratio over a sample bounds nothing about the rows it left out.
    /// An empty list means every computed component saw every scanned row.
    /// `None` means the assessment does not record this at all, which only a
    /// report read back from a document written before dataprof recorded it
    /// can be: unknown coverage is a third answer, not "nothing was sampled".
    ///
    /// The labels are the ones [`MetricConfidence::Mixed`] already records, so
    /// uniqueness appears as its two components, `key_uniqueness` and
    /// `duplicate_rows`, rather than as one dimension name. They carry
    /// different provenance: a full-stream row tracker counts duplicates
    /// exactly while the key scan reads the sample.
    pub fn sampled_dimensions(&self) -> Option<Vec<String>> {
        match &self.confidence {
            MetricConfidence::Exact | MetricConfidence::NotAssessed => Some(Vec::new()),
            MetricConfidence::Mixed {
                sampled_dimensions, ..
            } => Some(sampled_dimensions.clone()),
            // One sample produced everything, so every component carrying
            // signal came out of it.
            MetricConfidence::Approximate { .. } => Some(
                QualityDimension::all()
                    .into_iter()
                    .filter(|dimension| self.metrics.supports_dimension(*dimension))
                    .flat_map(|dimension| self.component_labels(dimension))
                    .collect(),
            ),
            MetricConfidence::Unrecorded => None,
        }
    }

    /// The component labels a dimension contributes, skipping components that
    /// were computed but carry no signal.
    fn component_labels(&self, dimension: QualityDimension) -> Vec<String> {
        let QualityDimension::Uniqueness = dimension else {
            return vec![dimension.to_string()];
        };
        let mut labels = Vec::new();
        if let Some(uniqueness) = &self.metrics.uniqueness {
            if uniqueness.key_column.is_some() {
                labels.push("key_uniqueness".to_string());
            }
            if uniqueness.rows_checked > 0 {
                labels.push("duplicate_rows".to_string());
            }
        }
        labels
    }
}

impl From<QualityMetrics> for QualityAssessment {
    fn from(metrics: QualityMetrics) -> Self {
        Self::exact(metrics)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn sample_bounds() -> ScoreBounds {
        let interval = |lower, upper| Some(ScoreInterval { lower, upper });
        ScoreBounds {
            confidence_level: 0.999,
            overall_score: interval(97.25, 99.5),
            dimension_scores: QualityDimension::all()
                .into_iter()
                .map(|dimension| (dimension.to_string(), interval(96.0, 100.0)))
                .collect(),
        }
    }

    #[test]
    fn score_bounds_survive_a_saved_document() {
        let assessment = QualityAssessment::approximate(perfect_assessed(), 10, Some(100))
            .with_score_bounds(Some(sample_bounds()));
        let text = serde_json::to_string(&assessment).unwrap();
        let restored: QualityAssessment = serde_json::from_str(&text).unwrap();

        assert_eq!(restored.score_bounds(), Some(&sample_bounds()));
    }

    #[test]
    fn an_exact_assessment_writes_no_bounds() {
        let text = serde_json::to_string(&QualityAssessment::exact(perfect_assessed())).unwrap();
        let document: serde_json::Value = serde_json::from_str(&text).unwrap();
        assert!(document.get("score_bounds").is_none());
    }

    #[test]
    fn edited_metrics_lose_the_bounds_computed_for_the_old_ones() {
        let mut assessment = QualityAssessment::approximate(perfect_assessed(), 10, Some(100))
            .with_score_bounds(Some(sample_bounds()));
        assessment
            .metrics
            .consistency
            .as_mut()
            .unwrap()
            .format_violations = 50;

        assert_eq!(assessment.score_bounds(), None);
        let text = serde_json::to_string(&assessment).unwrap();
        assert!(!text.contains("score_bounds"));
    }

    #[test]
    fn a_document_with_inverted_or_out_of_range_bounds_is_refused() {
        let assessment = QualityAssessment::approximate(perfect_assessed(), 10, Some(100))
            .with_score_bounds(Some(sample_bounds()));
        let document = serde_json::to_value(&assessment).unwrap();
        for (pointer, value) in [
            ("/score_bounds/overall_score/lower", serde_json::json!(99.9)),
            (
                "/score_bounds/overall_score/upper",
                serde_json::json!(100.5),
            ),
            ("/score_bounds/confidence_level", serde_json::json!(1.0)),
            (
                "/score_bounds/dimension_scores/validity/lower",
                serde_json::json!(-1.0),
            ),
        ] {
            let mut broken = document.clone();
            *broken.pointer_mut(pointer).unwrap() = value;
            assert!(
                serde_json::from_value::<QualityAssessment>(broken).is_err(),
                "{pointer} was accepted"
            );
        }
    }

    #[test]
    fn saved_aggregate_scores_survive_rounding_their_inputs() {
        let metrics = QualityMetrics {
            completeness: Some(CompletenessMetrics {
                missing_values_ratio: 32.77465638193188,
                complete_records_ratio: 67.22534361806812,
                null_columns: vec![],
                total_cells: 100,
            }),
            ..QualityMetrics::default()
        };
        let assessment = QualityAssessment::exact(metrics);
        let document = serde_json::to_value(&assessment).unwrap();
        assert_eq!(document["scores"]["overall_score"], 67.23);
        let mut restored: QualityAssessment = serde_json::from_value(document.clone()).unwrap();
        assert_eq!(restored.score(), Some(67.23));
        assert_eq!(
            restored.scores().dimension_scores["completeness"],
            Some(67.23)
        );
        assert_eq!(serde_json::to_value(&restored).unwrap(), document);

        // Public metrics may be edited by Rust callers, including below the
        // serialization precision. Any actual edit invalidates retained scores.
        restored
            .metrics
            .completeness
            .as_mut()
            .unwrap()
            .missing_values_ratio += 0.0001;
        assert_eq!(restored.score(), restored.metrics.overall_score());
        assert_ne!(restored.score(), Some(67.23));
    }

    #[test]
    fn legacy_quality_without_saved_scores_remains_readable() {
        let assessment = QualityAssessment::exact(perfect_assessed());
        let mut document = serde_json::to_value(&assessment).unwrap();
        document.as_object_mut().unwrap().remove("scores");
        let restored: QualityAssessment = serde_json::from_value(document).unwrap();
        assert_eq!(restored.score(), restored.metrics.overall_score());
    }

    #[test]
    fn saved_scores_cannot_fabricate_assessed_dimensions() {
        let assessment = QualityAssessment::exact(QualityMetrics::default());
        let mut document = serde_json::to_value(&assessment).unwrap();
        document["scores"]["dimension_scores"]["accuracy"] = serde_json::json!(100.0);
        assert!(serde_json::from_value::<QualityAssessment>(document).is_err());
        let mut document = serde_json::to_value(&assessment).unwrap();
        document["scores"]["dimension_scores"]
            .as_object_mut()
            .unwrap()
            .remove("accuracy");
        assert!(serde_json::from_value::<QualityAssessment>(document).is_err());
    }

    #[test]
    fn saved_scores_for_unknown_dimensions_are_range_checked() {
        let assessment = QualityAssessment::exact(perfect_assessed());
        let mut document = serde_json::to_value(&assessment).unwrap();
        document["scores"]["dimension_scores"]["lineage"] = serde_json::json!(55.0);
        let restored: QualityAssessment = serde_json::from_value(document.clone()).unwrap();
        assert_eq!(restored.scores().dimension_scores["lineage"], Some(55.0));
        document["scores"]["dimension_scores"]["lineage"] = serde_json::json!(1000.0);
        assert!(serde_json::from_value::<QualityAssessment>(document).is_err());
    }

    /// Metrics where every dimension has data to assess and a perfect score.
    fn perfect_assessed() -> QualityMetrics {
        QualityMetrics {
            completeness: Some(CompletenessMetrics {
                missing_values_ratio: 0.0,
                complete_records_ratio: 100.0,
                null_columns: vec![],
                total_cells: 100,
            }),
            consistency: Some(ConsistencyMetrics {
                data_type_consistency: 100.0,
                format_violations: 0,
                encoding_issues: 0,
                values_checked: 100,
            }),
            uniqueness: Some(UniquenessMetrics {
                duplicate_rows: 0,
                key_uniqueness: 100.0,
                high_cardinality_warning: false,
                rows_checked: 100,
                key_column: None,
                duplicate_rows_approximate: false,
            }),
            accuracy: Some(AccuracyMetrics {
                outlier_ratio: 0.0,
                range_violations: 0,
                negative_values_in_positive: 0,
                numeric_values_checked: 100,
            }),
            timeliness: Some(TimelinessMetrics {
                future_dates_count: 0,
                stale_data_ratio: 0.0,
                temporal_violations: 0,
                invalid_date_values: 0,
                date_values_checked: 100,
                temporal_pairs_checked: 100,
            }),
            validity: Some(ValidityMetrics {
                valid_values_ratio: 100.0,
                invalid_values: 0,
                values_checked: 100,
            }),
            precision: Some(PrecisionMetrics {
                decimal_places_consistency: 100.0,
                inconsistent_precision_values: 0,
                numeric_values_checked: 100,
            }),
            low_sample_warning: false,
            score_weights: QualityScoreWeights::default(),
        }
    }

    #[test]
    fn test_custom_weights_change_and_survive_serialized_score() {
        let mut metrics = perfect_assessed();
        if let Some(ref mut c) = metrics.completeness {
            c.missing_values_ratio = 100.0;
            c.complete_records_ratio = 0.0;
        }
        metrics.score_weights = QualityScoreWeights {
            completeness: 1.0,
            consistency: 0.0,
            uniqueness: 0.0,
            accuracy: 0.0,
            timeliness: 0.0,
            validity: 0.0,
            precision: 0.0,
        };

        assert_eq!(metrics.overall_score(), Some(0.0));

        let json = serde_json::to_string(&metrics).expect("serialize custom weights");
        assert!(json.contains("score_weights"));
        let restored: QualityMetrics =
            serde_json::from_str(&json).expect("deserialize custom weights");
        assert_eq!(restored.score_weights, metrics.score_weights);
        assert_eq!(restored.overall_score(), metrics.overall_score());
        assert_eq!(
            restored.assessed_dimensions(),
            vec![QualityDimension::Completeness]
        );
    }

    #[test]
    fn test_empty_metrics_nothing_assessed() {
        let metrics = QualityMetrics::empty();
        assert!(metrics.assessed_dimensions().is_empty());
        // Every denominator is zero, so there is nothing to average. A 0.0
        // here would read as "this data is terrible" (#571).
        assert_eq!(metrics.overall_score(), None);
    }

    #[test]
    fn test_unassessable_assessment_reports_no_score_and_no_confidence() {
        let assessment = QualityAssessment::exact(QualityMetrics::empty());
        assert_eq!(assessment.score(), None);
        assert!(
            matches!(assessment.confidence, MetricConfidence::NotAssessed),
            "confidence must not claim exactness about a score that was never computed, got {:?}",
            assessment.confidence
        );

        let sampled = QualityAssessment::approximate(QualityMetrics::empty(), 0, Some(0));
        assert!(matches!(sampled.confidence, MetricConfidence::NotAssessed));
    }

    #[test]
    fn test_assessed_assessment_keeps_its_confidence() {
        let assessment = QualityAssessment::exact(perfect_assessed());
        assert_eq!(assessment.score(), Some(100.0));
        assert!(matches!(assessment.confidence, MetricConfidence::Exact));
    }

    #[test]
    fn test_perfect_assessed_scores_100() {
        let metrics = perfect_assessed();
        assert_eq!(metrics.assessed_dimensions().len(), 7);
        assert_eq!(metrics.overall_score(), Some(100.0));
    }

    #[test]
    fn test_quality_score_completeness_weight() {
        let mut metrics = perfect_assessed();
        if let Some(ref mut c) = metrics.completeness {
            c.missing_values_ratio = 100.0;
            c.complete_records_ratio = 0.0;
        }
        assert_eq!(metrics.overall_score(), Some(75.0));
    }

    #[test]
    fn test_quality_score_all_bad() {
        let mut metrics = perfect_assessed();
        if let Some(ref mut c) = metrics.completeness {
            c.missing_values_ratio = 100.0;
            c.complete_records_ratio = 0.0;
        }
        if let Some(ref mut c) = metrics.consistency {
            c.data_type_consistency = 0.0;
        }
        if let Some(ref mut u) = metrics.uniqueness {
            u.duplicate_rows = 100;
        }
        if let Some(ref mut a) = metrics.accuracy {
            a.outlier_ratio = 100.0;
        }
        if let Some(ref mut t) = metrics.timeliness {
            t.stale_data_ratio = 100.0;
        }
        if let Some(ref mut v) = metrics.validity {
            v.valid_values_ratio = 0.0;
        }
        if let Some(ref mut p) = metrics.precision {
            p.decimal_places_consistency = 0.0;
        }

        // Assessed and genuinely bad: 0.0, not None.
        assert_eq!(metrics.overall_score(), Some(0.0));
    }

    #[test]
    fn test_vacuous_dimensions_drop_out() {
        // Text-only dataset shape: nothing numeric, no dates, no rows scanned
        // for duplicates. Under the old aggregation these dimensions counted
        // as perfect and floored the score at 70; now they are excluded and
        // the weights renormalize over what was actually assessed.
        let mut metrics = perfect_assessed();
        if let Some(ref mut c) = metrics.completeness {
            c.missing_values_ratio = 50.0;
            c.complete_records_ratio = 50.0;
        }
        if let Some(ref mut u) = metrics.uniqueness {
            u.rows_checked = 0;
        }
        if let Some(ref mut a) = metrics.accuracy {
            a.numeric_values_checked = 0;
        }
        if let Some(ref mut t) = metrics.timeliness {
            t.date_values_checked = 0;
        }
        if let Some(ref mut v) = metrics.validity {
            v.values_checked = 0;
        }
        if let Some(ref mut p) = metrics.precision {
            p.numeric_values_checked = 0;
        }

        assert_eq!(
            metrics.assessed_dimensions(),
            vec![
                QualityDimension::Completeness,
                QualityDimension::Consistency
            ]
        );
        // (0.25 * 50 + 0.20 * 100) / 0.45 = 72.22..
        assert!((metrics.overall_score().expect("assessed") - 72.2222).abs() < 0.01);
    }

    #[test]
    fn test_duplicate_rows_lower_uniqueness_score() {
        let mut metrics = perfect_assessed();
        if let Some(ref mut u) = metrics.uniqueness {
            u.duplicate_rows = 30;
        }
        let score = metrics
            .uniqueness_score()
            .expect("uniqueness should be assessed");
        assert!((score - 70.0).abs() < 0.01);
    }

    #[test]
    fn test_key_only_uniqueness_when_duplicate_scan_not_assessable() {
        // Streaming shape: per-column samples are misaligned so the
        // duplicate scan did not run, but key uniqueness is exact.
        let mut metrics = perfect_assessed();
        if let Some(ref mut u) = metrics.uniqueness {
            u.rows_checked = 0;
            u.key_column = Some("order_id".to_string());
            u.key_uniqueness = 90.0;
        }
        let score = metrics
            .uniqueness_score()
            .expect("key component alone should keep uniqueness assessed");
        assert!((score - 90.0).abs() < 0.01);
    }

    #[test]
    fn test_key_column_blends_into_uniqueness_score() {
        let mut metrics = perfect_assessed();
        if let Some(ref mut u) = metrics.uniqueness {
            u.key_column = Some("order_id".to_string());
            u.key_uniqueness = 60.0;
        }
        // mean(duplicate-free score 100, key uniqueness 60)
        let score = metrics
            .uniqueness_score()
            .expect("uniqueness should be assessed");
        assert!((score - 80.0).abs() < 0.01);
    }

    #[test]
    fn test_format_and_encoding_violations_lower_consistency_score() {
        let mut metrics = perfect_assessed();
        if let Some(ref mut c) = metrics.consistency {
            c.format_violations = 5;
            c.encoding_issues = 5;
        }
        let score = metrics
            .consistency_score()
            .expect("consistency should be assessed");
        assert!((score - 90.0).abs() < 0.01);
    }

    #[test]
    fn test_range_and_negative_violations_lower_accuracy_score() {
        let mut metrics = perfect_assessed();
        if let Some(ref mut a) = metrics.accuracy {
            a.outlier_ratio = 10.0;
            a.range_violations = 5;
            a.negative_values_in_positive = 5;
        }
        let score = metrics
            .accuracy_score()
            .expect("accuracy should be assessed");
        assert!((score - 80.0).abs() < 0.01);
    }

    #[test]
    fn test_future_dates_and_temporal_violations_lower_timeliness_score() {
        let mut metrics = perfect_assessed();
        if let Some(ref mut t) = metrics.timeliness {
            t.stale_data_ratio = 20.0;
            t.future_dates_count = 5;
            t.temporal_violations = 5;
        }
        let score = metrics
            .timeliness_score()
            .expect("timeliness should be assessed");
        assert!((score - 70.0).abs() < 0.01);
    }

    #[test]
    fn test_legacy_json_without_denominators_is_not_assessed() {
        // Reports serialized before the denominator fields existed
        // deserialize with zero denominators: facts remain readable, but no
        // dimension is assessable and no score is fabricated.
        let json = r#"{
            "completeness": {
                "missing_values_ratio": 5.0,
                "complete_records_ratio": 95.0,
                "null_columns": []
            }
        }"#;
        let metrics: QualityMetrics = serde_json::from_str(json).unwrap();

        assert!((metrics.missing_values_ratio() - 5.0).abs() < 0.01);
        assert!(metrics.completeness_score().is_none());
        assert!(metrics.assessed_dimensions().is_empty());
    }

    #[test]
    fn test_partial_dimensions_only_completeness() {
        let metrics = QualityMetrics {
            completeness: Some(CompletenessMetrics {
                complete_records_ratio: 100.0,
                missing_values_ratio: 0.0,
                null_columns: vec![],
                total_cells: 10,
            }),
            ..QualityMetrics::default()
        };

        assert!(metrics.completeness.is_some());
        assert!(metrics.consistency.is_none());
        assert!(metrics.uniqueness.is_none());
        assert!(metrics.accuracy.is_none());
        assert!(metrics.timeliness.is_none());
        assert_eq!(metrics.overall_score(), Some(100.0));
    }

    #[test]
    fn test_partial_dimensions_two_dimensions() {
        let metrics = QualityMetrics {
            completeness: Some(CompletenessMetrics {
                missing_values_ratio: 50.0,
                complete_records_ratio: 50.0,
                null_columns: vec![],
                total_cells: 100,
            }),
            uniqueness: Some(UniquenessMetrics {
                duplicate_rows: 20,
                key_uniqueness: 100.0,
                high_cardinality_warning: false,
                rows_checked: 100,
                key_column: None,
                duplicate_rows_approximate: false,
            }),
            ..QualityMetrics::default()
        };

        // (0.25 * 50 + 0.15 * 80) / 0.40 = 61.25
        assert!((metrics.overall_score().expect("assessed") - 61.25).abs() < 0.01);
    }

    #[test]
    fn test_all_dimensions_none_score_is_none() {
        let metrics = QualityMetrics::default();

        assert_eq!(metrics.overall_score(), None);
        assert!(metrics.assessed_dimensions().is_empty());
    }

    #[test]
    fn test_partial_dimensions_json_skips_none() {
        let metrics = QualityMetrics {
            completeness: Some(CompletenessMetrics {
                total_cells: 10,
                ..CompletenessMetrics::default()
            }),
            ..QualityMetrics::default()
        };

        let json = serde_json::to_string(&metrics).unwrap();
        assert!(json.contains("completeness"));
        assert!(!json.contains("consistency"));
        assert!(!json.contains("uniqueness"));
        assert!(!json.contains("accuracy"));
        assert!(!json.contains("timeliness"));
    }

    #[test]
    fn a_dimension_with_no_denominator_is_not_serialized() {
        // `empty()` builds every dimension with zero counters, which is what an
        // engine produces for a header-only file. Serializing those defaults
        // published `valid_values_ratio: 100.0` from `values_checked: 0` (#622).
        let metrics = QualityMetrics::empty();
        assert!(metrics.assessed_dimensions().is_empty());

        let json = serde_json::to_string(&metrics).unwrap();
        for dimension in [
            "completeness",
            "consistency",
            "uniqueness",
            "accuracy",
            "timeliness",
            "validity",
            "precision",
        ] {
            assert!(
                !json.contains(dimension),
                "{dimension} was serialized without having assessed anything: {json}"
            );
        }
    }

    #[test]
    fn a_zero_weighted_dimension_keeps_its_evidence() {
        // Serialization asks "did this measure anything", which is a wider
        // question than "is this behind the overall score". A weight of 0.0
        // removes a dimension from `assessed_dimensions` and from the
        // aggregate, and removes nothing from what it actually observed.
        let mut metrics = QualityMetrics::empty();
        metrics.validity = Some(ValidityMetrics {
            valid_values_ratio: 80.0,
            invalid_values: 2,
            values_checked: 10,
        });
        metrics.score_weights = QualityScoreWeights {
            validity: 0.0,
            ..QualityScoreWeights::default()
        };

        assert!(metrics.validity.as_ref().expect("present").is_assessed());
        assert!(metrics.validity_score().is_some());
        assert!(
            !metrics
                .assessed_dimensions()
                .contains(&QualityDimension::Validity)
        );
        assert_eq!(metrics.overall_score(), None);

        let json = serde_json::to_string(&metrics).unwrap();
        assert!(
            json.contains("valid_values_ratio"),
            "a measured dimension was dropped for carrying no weight: {json}"
        );
    }

    #[test]
    fn serialized_dimensions_are_exactly_the_ones_with_a_denominator() {
        let mut metrics = QualityMetrics::empty();
        metrics.completeness = Some(CompletenessMetrics {
            missing_values_ratio: 10.0,
            complete_records_ratio: 90.0,
            null_columns: vec!["a".to_string()],
            total_cells: 100,
        });
        metrics.validity = Some(ValidityMetrics {
            valid_values_ratio: 80.0,
            invalid_values: 2,
            values_checked: 10,
        });

        let assessed = metrics.assessed_dimensions();
        assert_eq!(
            assessed,
            vec![QualityDimension::Completeness, QualityDimension::Validity]
        );

        let json = serde_json::to_string(&metrics).unwrap();
        assert!(json.contains("completeness"));
        assert!(json.contains("validity"));
        assert!(!json.contains("consistency"));
        assert!(!json.contains("precision"));
    }

    #[test]
    fn is_assessed_agrees_with_the_dimension_score() {
        // One definition of "assessed" behind both, so a score can never be
        // None while its evidence claims to have measured something.
        let assessed = QualityMetrics {
            uniqueness: Some(UniquenessMetrics {
                duplicate_rows: 0,
                key_uniqueness: 100.0,
                high_cardinality_warning: false,
                rows_checked: 0,
                // No rows scanned, but a key column was identified, so the key
                // component alone still carries a measurement.
                key_column: Some("id".to_string()),
                duplicate_rows_approximate: false,
            }),
            ..QualityMetrics::default()
        };
        let uniqueness = assessed.uniqueness.as_ref().expect("present");
        assert!(uniqueness.is_assessed());
        assert!(assessed.uniqueness_score().is_some());

        let unassessed = QualityMetrics::empty();
        for metrics in [
            unassessed.completeness.as_ref().map(|m| m.is_assessed()),
            unassessed.consistency.as_ref().map(|m| m.is_assessed()),
            unassessed.uniqueness.as_ref().map(|m| m.is_assessed()),
            unassessed.accuracy.as_ref().map(|m| m.is_assessed()),
            unassessed.timeliness.as_ref().map(|m| m.is_assessed()),
            unassessed.validity.as_ref().map(|m| m.is_assessed()),
            unassessed.precision.as_ref().map(|m| m.is_assessed()),
        ] {
            assert_eq!(metrics, Some(false));
        }
    }

    #[test]
    fn test_partial_dimensions_flat_accessors_return_defaults() {
        let metrics = QualityMetrics::default();

        assert!((metrics.complete_records_ratio() - 100.0).abs() < 0.01);
        assert!((metrics.data_type_consistency() - 100.0).abs() < 0.01);
        assert!((metrics.key_uniqueness() - 100.0).abs() < 0.01);
        assert!((metrics.missing_values_ratio() - 0.0).abs() < 0.01);
        assert_eq!(metrics.duplicate_rows(), 0);
        assert!(!metrics.high_cardinality_warning());
    }

    #[test]
    fn test_partial_dimension_flat_defaults_table() {
        struct Case {
            name: &'static str,
            metrics: QualityMetrics,
            has_completeness: bool,
            has_uniqueness: bool,
            has_accuracy: bool,
            missing_values_ratio: f64,
            key_uniqueness: f64,
            outlier_ratio: f64,
        }

        let cases = [
            Case {
                name: "only completeness",
                metrics: QualityMetrics {
                    completeness: Some(CompletenessMetrics {
                        missing_values_ratio: 12.5,
                        complete_records_ratio: 87.5,
                        null_columns: vec!["email".to_string()],
                        total_cells: 16,
                    }),
                    ..QualityMetrics::default()
                },
                has_completeness: true,
                has_uniqueness: false,
                has_accuracy: false,
                missing_values_ratio: 12.5,
                key_uniqueness: 100.0,
                outlier_ratio: 0.0,
            },
            Case {
                name: "only uniqueness",
                metrics: QualityMetrics {
                    uniqueness: Some(UniquenessMetrics {
                        duplicate_rows: 2,
                        key_uniqueness: 92.0,
                        high_cardinality_warning: true,
                        rows_checked: 25,
                        key_column: Some("user_id".to_string()),
                        duplicate_rows_approximate: false,
                    }),
                    ..QualityMetrics::default()
                },
                has_completeness: false,
                has_uniqueness: true,
                has_accuracy: false,
                missing_values_ratio: 0.0,
                key_uniqueness: 92.0,
                outlier_ratio: 0.0,
            },
            Case {
                name: "only accuracy",
                metrics: QualityMetrics {
                    accuracy: Some(AccuracyMetrics {
                        outlier_ratio: 6.25,
                        range_violations: 1,
                        negative_values_in_positive: 1,
                        numeric_values_checked: 16,
                    }),
                    ..QualityMetrics::default()
                },
                has_completeness: false,
                has_uniqueness: false,
                has_accuracy: true,
                missing_values_ratio: 0.0,
                key_uniqueness: 100.0,
                outlier_ratio: 6.25,
            },
        ];

        for case in cases {
            assert_eq!(
                case.metrics.completeness.is_some(),
                case.has_completeness,
                "{} completeness presence",
                case.name
            );
            assert_eq!(
                case.metrics.uniqueness.is_some(),
                case.has_uniqueness,
                "{} uniqueness presence",
                case.name
            );
            assert_eq!(
                case.metrics.accuracy.is_some(),
                case.has_accuracy,
                "{} accuracy presence",
                case.name
            );
            assert!(
                (case.metrics.missing_values_ratio() - case.missing_values_ratio).abs() < 0.01,
                "{} missing_values_ratio",
                case.name
            );
            assert!(
                (case.metrics.key_uniqueness() - case.key_uniqueness).abs() < 0.01,
                "{} key_uniqueness",
                case.name
            );
            assert!(
                (case.metrics.outlier_ratio() - case.outlier_ratio).abs() < 0.01,
                "{} outlier_ratio",
                case.name
            );
        }
    }
}