use super::*;

#[test]
fn new_creates_healthy_state() {
    let state = MetricsState::new();
    assert!(state.health_ok.load(Ordering::Relaxed));
}

#[test]
fn uptime_is_non_negative() {
    let state = MetricsState::new();
    assert!(state.uptime_seconds() >= 0.0);
}

#[test]
fn health_ok_value_returns_one_when_healthy() {
    let state = MetricsState::new();
    assert!((state.health_ok_value() - 1.0).abs() < f64::EPSILON);
}

#[test]
fn health_ok_value_returns_zero_when_degraded() {
    let state = MetricsState::new();
    state.health_ok.store(false, Ordering::Relaxed);
    assert!(state.health_ok_value().abs() < f64::EPSILON);
}

#[test]
fn render_prometheus_contains_expected_metrics() {
    let state = MetricsState::new();
    let output = state.render_prometheus();
    assert!(output.contains("tidaldb_uptime_seconds"));
    assert!(output.contains("tidaldb_health_ok"));
    assert!(output.contains("tidaldb_info"));
    assert!(output.contains("partition_id=\"0\""));
}

/// Parse the rendered exposition text and assert it is well-formed per the
/// Prometheus text format: every `# HELP <name>` / `# TYPE <name> <type>` pair
/// names the same metric and is followed by sample lines for that metric, and
/// every sample line has a parseable `name{labels} value` (or `name value`)
/// shape with a numeric value. This guards against malformed output that the
/// substring-only tests above would silently pass.
#[test]
fn render_prometheus_parses_as_well_formed_exposition() {
    let state = MetricsState::new();
    let output = state.render_prometheus();

    let mut seen_metric_line = false;
    let mut current_help: Option<String> = None;
    let mut current_type: Option<(String, String)> = None;

    for raw in output.lines() {
        let line = raw.trim_end();
        if line.is_empty() {
            continue;
        }

        if let Some(rest) = line.strip_prefix("# HELP ") {
            // "# HELP <name> <free text>"
            let name = rest
                .split_whitespace()
                .next()
                .expect("HELP line must name a metric");
            assert!(
                is_valid_metric_name(name),
                "HELP names invalid metric: {line:?}"
            );
            current_help = Some(name.to_string());
            continue;
        }

        if let Some(rest) = line.strip_prefix("# TYPE ") {
            // "# TYPE <name> <type>"
            let mut parts = rest.split_whitespace();
            let name = parts.next().expect("TYPE line must name a metric");
            let ty = parts.next().expect("TYPE line must declare a type");
            assert!(
                is_valid_metric_name(name),
                "TYPE names invalid metric: {line:?}"
            );
            assert!(
                matches!(
                    ty,
                    "counter" | "gauge" | "histogram" | "summary" | "untyped"
                ),
                "unknown metric type {ty:?} in {line:?}"
            );
            // HELP and TYPE for a metric must agree on the name.
            if let Some(h) = &current_help {
                assert_eq!(h, name, "HELP/TYPE name mismatch: {line:?}");
            }
            current_type = Some((name.to_string(), ty.to_string()));
            continue;
        }

        // Any other non-comment line is a sample line: "<name>[{labels}] <value>".
        assert!(
            !line.starts_with('#'),
            "unexpected comment line shape: {line:?}"
        );
        let (metric, value) = line
            .rsplit_once(' ')
            .unwrap_or_else(|| panic!("sample line has no value field: {line:?}"));
        // The base name precedes any '{' label block; histogram lines append
        // a "_bucket"/"_sum"/"_count" suffix to the declared TYPE name.
        let base = metric.split('{').next().unwrap_or(metric);
        assert!(
            is_valid_metric_name(base),
            "sample line has invalid metric name: {line:?}"
        );
        // Value must parse as a finite f64 (Prometheus values are f64), allowing
        // the "+Inf" le-label edge only inside the label block, never the value.
        let parsed: f64 = value
            .parse()
            .unwrap_or_else(|_| panic!("sample value is not numeric: {line:?}"));
        assert!(parsed.is_finite(), "sample value not finite: {line:?}");
        // A sample line must be covered by a preceding TYPE declaration whose
        // name is a prefix of the sample's base name.
        if let Some((tname, _)) = &current_type {
            assert!(
                base == *tname || base.starts_with(&format!("{tname}_")),
                "sample {base:?} not covered by TYPE {tname:?}: {line:?}"
            );
        } else {
            panic!("sample line before any TYPE declaration: {line:?}");
        }
        seen_metric_line = true;
    }

    assert!(seen_metric_line, "no metric sample lines rendered");
}

/// A Prometheus metric name: `[a-zA-Z_:][a-zA-Z0-9_:]*`.
fn is_valid_metric_name(name: &str) -> bool {
    let mut chars = name.chars();
    match chars.next() {
        Some(c) if c.is_ascii_alphabetic() || c == '_' || c == ':' => {}
        _ => return false,
    }
    chars.all(|c| c.is_ascii_alphanumeric() || c == '_' || c == ':')
}

#[test]
fn render_healthz_contains_expected_fields() {
    let state = MetricsState::new();
    let output = state.render_healthz();
    assert!(output.contains("\"status\":\"ok\""));
    assert!(output.contains("\"uptime_seconds\":"));
    assert!(output.contains("\"version\":"));
    assert!(output.contains("\"build_hash\":"));
}

#[test]
fn render_healthz_degraded() {
    let state = MetricsState::new();
    state.health_ok.store(false, Ordering::Relaxed);
    let output = state.render_healthz();
    assert!(output.contains("\"status\":\"degraded\""));
}

#[cfg(feature = "metrics")]
#[test]
fn metrics_state_renders_signal_metrics() {
    let state = MetricsState::new();
    state.signal_writes_total.store(42, Ordering::Relaxed);
    state.signal_hot_entries.store(100, Ordering::Relaxed);
    state.wal_lag_bytes.store(8192, Ordering::Relaxed);
    state
        .wal_compacted_segments_total
        .store(3, Ordering::Relaxed);

    let output = state.render_prometheus();

    assert!(
        output.contains("tidaldb_signal_writes_total"),
        "missing signal_writes_total: {output}"
    );
    assert!(output.contains("42"), "missing value 42: {output}");
    assert!(
        output.contains("tidaldb_signal_hot_entries"),
        "missing signal_hot_entries: {output}"
    );
    assert!(output.contains("100"), "missing value 100: {output}");
    assert!(
        output.contains("tidaldb_wal_lag_bytes"),
        "missing wal_lag_bytes: {output}"
    );
    assert!(output.contains("8192"), "missing value 8192: {output}");
    assert!(
        output.contains("tidaldb_wal_compacted_segments_total"),
        "missing wal_compacted_segments_total: {output}"
    );
    assert!(
        output.contains("tidaldb_checkpoint_age_seconds"),
        "missing checkpoint_age_seconds: {output}"
    );
    assert!(
        output.contains("tidaldb_signal_write_latency_us"),
        "missing signal_write_latency_us histogram: {output}"
    );
}

#[cfg(feature = "metrics")]
#[test]
fn render_prometheus_contains_index_metrics() {
    let m = MetricsState::new();
    m.tantivy_segment_count.store(3, Ordering::Relaxed);
    m.tantivy_indexed_docs.store(10000, Ordering::Relaxed);
    m.usearch_vector_count.store(500, Ordering::Relaxed);
    m.usearch_index_size_bytes
        .store(1_048_576, Ordering::Relaxed);
    m.bitmap_index_cardinality.store(42, Ordering::Relaxed);

    let prom = m.render_prometheus();
    assert!(
        prom.contains("tidaldb_tantivy_segment_count 3"),
        "missing tantivy_segment_count: {prom}"
    );
    assert!(
        prom.contains("tidaldb_tantivy_indexed_docs 10000"),
        "missing tantivy_indexed_docs: {prom}"
    );
    assert!(
        prom.contains("tidaldb_usearch_vector_count 500"),
        "missing usearch_vector_count: {prom}"
    );
    assert!(
        prom.contains("tidaldb_usearch_index_size_bytes 1048576"),
        "missing usearch_index_size_bytes: {prom}"
    );
    assert!(
        prom.contains("tidaldb_bitmap_index_cardinality 42"),
        "missing bitmap_index_cardinality: {prom}"
    );
}

// ── HEALTH-1 / CONCURRENCY-2: degraded derivation ───────────────────────────

#[test]
fn fresh_state_is_not_degraded() {
    let state = MetricsState::new();
    assert!(!state.is_degraded());
    assert!((state.health_ok_value() - 1.0).abs() < f64::EPSILON);
    assert!(state.render_healthz().contains("\"status\":\"ok\""));
}

#[test]
fn dead_checkpoint_thread_flips_health_degraded() {
    // Simulate the checkpoint thread dying (panic supervisor or a finished
    // join handle observed by health_check). Health must flip even though the
    // explicit close flag is still `true` (HEALTH-1, CONCURRENCY-2).
    let state = MetricsState::new();
    assert!(!state.is_degraded(), "precondition: healthy");

    state.checkpoint_thread_died.store(true, Ordering::Release);

    assert!(
        state.is_degraded(),
        "dead checkpoint thread must report degraded"
    );
    assert!(state.health_ok_value().abs() < f64::EPSILON);
    assert!(
        state.render_healthz().contains("\"status\":\"degraded\""),
        "/healthz must report degraded on a dead checkpoint thread"
    );
    assert!(
        state
            .render_prometheus()
            .contains("tidaldb_health_ok{partition_id=\"0\"} 0"),
        "tidaldb_health_ok must read 0 on a dead checkpoint thread"
    );
}

#[cfg(feature = "metrics")]
#[test]
fn stale_checkpoint_flips_health_degraded() {
    // A checkpoint that succeeded once but is now older than the staleness
    // limit means durability has silently stalled (stuck/failing thread or
    // disk-full). Health must flip from "ok" to "degraded".
    let state = MetricsState::new();

    // Fresh checkpoint (now): not stale.
    let now_ns = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| u64::try_from(d.as_nanos()).unwrap_or(u64::MAX))
        .unwrap_or(0);
    state.last_checkpoint_ns.store(now_ns, Ordering::Release);
    assert!(!state.is_degraded(), "a just-written checkpoint is healthy");

    // Now back-date the last checkpoint past the staleness limit.
    let stale_ns = now_ns.saturating_sub(MetricsState::CHECKPOINT_STALENESS_LIMIT_NS + 1);
    state.last_checkpoint_ns.store(stale_ns, Ordering::Release);
    assert!(
        state.is_degraded(),
        "a checkpoint older than the staleness limit must report degraded"
    );
    assert!(
        state.render_healthz().contains("\"status\":\"degraded\""),
        "/healthz must report degraded on a stale checkpoint"
    );
}

#[cfg(feature = "metrics")]
#[test]
fn zero_checkpoint_timestamp_is_not_stale() {
    // A database that has not yet run its first checkpoint (last == 0) must
    // not be flagged stale on age alone.
    let state = MetricsState::new();
    state.last_checkpoint_ns.store(0, Ordering::Release);
    assert!(!state.is_degraded());
}

#[cfg(feature = "metrics")]
#[test]
fn active_sessions_tracks_lifecycle() {
    let state = MetricsState::new();
    state.active_sessions.fetch_add(1, Ordering::Relaxed);
    state.active_sessions.fetch_add(1, Ordering::Relaxed);
    assert_eq!(state.active_sessions.load(Ordering::Relaxed), 2);
    state.active_sessions.fetch_sub(1, Ordering::Relaxed);
    assert_eq!(state.active_sessions.load(Ordering::Relaxed), 1);
}

#[cfg(feature = "metrics")]
#[test]
fn degradation_level_renders_correctly() {
    let state = MetricsState::new();
    state.degradation_level.store(2, Ordering::Relaxed);
    let output = state.render_prometheus();
    assert!(
        output.contains("tidaldb_degradation_level"),
        "missing tidaldb_degradation_level: {output}"
    );
}

#[cfg(feature = "metrics")]
#[test]
fn render_prometheus_contains_session_metrics() {
    let state = MetricsState::new();
    let prom = state.render_prometheus();
    assert!(
        prom.contains("tidaldb_active_sessions"),
        "missing active_sessions: {prom}"
    );
    assert!(
        prom.contains("tidaldb_closed_sessions_total"),
        "missing closed_sessions_total: {prom}"
    );
    assert!(
        prom.contains("tidaldb_session_auto_closed_total"),
        "missing session_auto_closed_total: {prom}"
    );
    assert!(
        prom.contains("tidaldb_rate_limited_total"),
        "missing rate_limited_total: {prom}"
    );
    assert!(
        prom.contains("tidaldb_degradation_level"),
        "missing degradation_level: {prom}"
    );
}

#[cfg(feature = "metrics")]
#[test]
fn metrics_state_checkpoint_age_zero_when_no_checkpoint() {
    let state = MetricsState::new();
    // last_checkpoint_ns is 0 (default) -- checkpoint_age should be 0.
    let output = state.render_prometheus();
    // Find the checkpoint_age_seconds line and verify it's 0.
    let age_line = output
        .lines()
        .find(|l| l.starts_with("tidaldb_checkpoint_age_seconds "))
        .expect("missing checkpoint_age_seconds line");
    assert!(
        age_line.contains(" 0"),
        "checkpoint age should be 0 when no checkpoint: {age_line}"
    );
}

// ── Feature-flag verification tests (m7p4, task-07) ─────────────────

/// `QueryStats` is NOT feature-gated -- always available regardless of
/// whether the `metrics` feature is enabled.
#[test]
fn query_stats_always_available() {
    use crate::query::QueryStats;
    let stats = QueryStats::new("test".to_owned());
    assert_eq!(stats.profile_name, "test");
    assert_eq!(stats.total_time_us, 0);
}

/// Base `MetricsState` fields (`uptime_seconds`, `health_ok_value`) work
/// without the `metrics` feature -- they are unconditionally compiled.
#[test]
fn metrics_state_base_always_available() {
    let state = MetricsState::new();
    assert!(state.uptime_seconds() >= 0.0);
    assert!((state.health_ok_value() - 1.0).abs() < f64::EPSILON);
}

/// Feature-gated counters (`signal_writes_total`, etc.) only exist when
/// the `metrics` feature is enabled -- this test proves they compile and
/// are functional.
#[cfg(feature = "metrics")]
#[test]
fn metrics_feature_counters_exist() {
    let state = MetricsState::new();
    state.signal_writes_total.fetch_add(1, Ordering::Relaxed);
    assert_eq!(state.signal_writes_total.load(Ordering::Relaxed), 1);
}

/// Every co-located shard group must get its own `tidaldb_usearch_vector_count`.
///
/// Before 2026-08-30 only the metrics OWNER's group had this series, so on the
/// 3-group RF3 cluster two thirds of each node's corpus had no vector count at
/// all and a divergence there was unobservable. The owner stays UNLABELED for
/// wire compatibility, so a cross-node alert grouped `by (shard)` puts each
/// replica set in its own bucket without mixing groups or double-counting.
#[test]
fn every_colocated_group_exposes_its_own_vector_count() {
    let owner = MetricsState::new();
    owner.usearch_vector_count.store(33335, Ordering::Relaxed);

    for (shard, count) in [(1u16, 111u64), (2u16, 222u64)] {
        let sib = std::sync::Arc::new(MetricsState::new());
        sib.usearch_vector_count.store(count, Ordering::Relaxed);
        owner.register_node_sibling(shard, sib);
    }

    let out = owner.render_prometheus();
    assert!(
        out.contains("\ntidaldb_usearch_vector_count 33335\n"),
        "owner series must stay unlabeled for wire compatibility:\n{out}"
    );
    assert!(
        out.contains("tidaldb_usearch_vector_count{shard=\"1\"} 111"),
        "group 1 count missing:\n{out}"
    );
    assert!(
        out.contains("tidaldb_usearch_vector_count{shard=\"2\"} 222"),
        "group 2 count missing:\n{out}"
    );
    // Exactly one series per hosted group: 1 owner + 2 siblings. More would mean
    // a node double-counts itself and any sum()/max() over it is wrong.
    assert_eq!(
        out.lines()
            .filter(|l| l.starts_with("tidaldb_usearch_vector_count"))
            .count(),
        3,
        "one vector-count series per hosted group, no duplicates:\n{out}"
    );
}