use once_cell::sync::Lazy;
use prometheus::{
    Gauge, HistogramOpts, HistogramVec, IntCounter, IntCounterVec, IntGauge, IntGaugeVec, Opts,
    default_registry,
};

static TSO_REQUESTS_TOTAL: Lazy<IntCounterVec> = Lazy::new(|| {
    let metric = IntCounterVec::new(
        Opts::new(
            "tso_requests_total",
            "Total number of tso requests grouped by outcome",
        ),
        &["outcome"],
    )
    .expect("build tso_requests_total");
    let _ = default_registry().register(Box::new(metric.clone()));
    metric
});

static TSO_TIMESTAMPS_TOTAL: Lazy<IntCounter> = Lazy::new(|| {
    let metric = IntCounter::new(
        "tso_timestamps_total",
        "Total number of allocated timestamps",
    )
    .expect("build tso_timestamps_total");
    let _ = default_registry().register(Box::new(metric.clone()));
    metric
});

static TSO_REQUEST_LATENCY_SECONDS: Lazy<HistogramVec> = Lazy::new(|| {
    let metric = HistogramVec::new(
        HistogramOpts::new(
            "tso_request_latency_seconds",
            "Latency distribution for tso request handling",
        ),
        &["outcome"],
    )
    .expect("build tso_request_latency_seconds");
    let _ = default_registry().register(Box::new(metric.clone()));
    metric
});

static CONFIG_APPLY_ATTEMPTS_TOTAL: Lazy<IntCounterVec> = Lazy::new(|| {
    let metric = IntCounterVec::new(
        Opts::new(
            "config_apply_attempts_total",
            "Total number of config apply attempts grouped by endpoint",
        ),
        &["endpoint"],
    )
    .expect("build config_apply_attempts_total");
    let _ = default_registry().register(Box::new(metric.clone()));
    metric
});

static CONFIG_APPLY_FAILURES_TOTAL: Lazy<IntCounterVec> = Lazy::new(|| {
    let metric = IntCounterVec::new(
        Opts::new(
            "config_apply_failures_total",
            "Total number of config apply failures grouped by reason",
        ),
        &["reason"],
    )
    .expect("build config_apply_failures_total");
    let _ = default_registry().register(Box::new(metric.clone()));
    metric
});

static ACTIVE_CONFIG_EPOCH: Lazy<IntGauge> = Lazy::new(|| {
    let metric = IntGauge::new("active_config_epoch", "Current active config epoch")
        .expect("build active_config_epoch");
    let _ = default_registry().register(Box::new(metric.clone()));
    metric
});

static CONFIG_APPLY_PENDING_ACTORS: Lazy<IntGauge> = Lazy::new(|| {
    let metric = IntGauge::new(
        "config_apply_pending_actors",
        "Number of runtime actors pending config apply",
    )
    .expect("build config_apply_pending_actors");
    let _ = default_registry().register(Box::new(metric.clone()));
    metric
});

static ETCD_REQUEST_LATENCY_SECONDS: Lazy<HistogramVec> = Lazy::new(|| {
    let metric = HistogramVec::new(
        HistogramOpts::new(
            "etcd_request_latency_seconds",
            "Latency distribution for etcd operations",
        ),
        &["operation", "outcome"],
    )
    .expect("build etcd_request_latency_seconds");
    let _ = default_registry().register(Box::new(metric.clone()));
    metric
});

static ETCD_REQUESTS_FAILED_TOTAL: Lazy<IntCounterVec> = Lazy::new(|| {
    let metric = IntCounterVec::new(
        Opts::new(
            "etcd_requests_failed_total",
            "Failed etcd requests by operation",
        ),
        &["operation"],
    )
    .expect("build etcd_requests_failed_total");
    let _ = default_registry().register(Box::new(metric.clone()));
    metric
});

static GROUP_ACTOR_MAILBOX_SIZE: Lazy<IntGauge> = Lazy::new(|| {
    let metric = IntGauge::new(
        "group_actor_mailbox_size",
        "Aggregated group actor mailbox depth on this node",
    )
    .expect("build group_actor_mailbox_size");
    let _ = default_registry().register(Box::new(metric.clone()));
    metric
});

static GROUP_ACTOR_RESTARTS_TOTAL: Lazy<IntCounterVec> = Lazy::new(|| {
    let metric = IntCounterVec::new(
        Opts::new(
            "group_actor_restarts_total",
            "Group actor restarts grouped by bounded reason",
        ),
        &["reason"],
    )
    .expect("build group_actor_restarts_total");
    let _ = default_registry().register(Box::new(metric.clone()));
    metric
});

static TSO_ELECTION_STATUS: Lazy<IntGauge> = Lazy::new(|| {
    let metric = IntGauge::new(
        "tso_election_status",
        "Node-level election status (1 when at least one primary is served)",
    )
    .expect("build tso_election_status");
    let _ = default_registry().register(Box::new(metric.clone()));
    metric
});

static TSO_LEASE_STATUS: Lazy<IntGauge> = Lazy::new(|| {
    let metric = IntGauge::new(
        "tso_lease_status",
        "Node-level lease status (1 when at least one live lease exists)",
    )
    .expect("build tso_lease_status");
    let _ = default_registry().register(Box::new(metric.clone()));
    metric
});

static TSO_ELECTION_CAMPAIGNS_TOTAL: Lazy<IntCounter> = Lazy::new(|| {
    let metric = IntCounter::new(
        "tso_election_campaigns_total",
        "Number of successful election campaign transitions",
    )
    .expect("build tso_election_campaigns_total");
    let _ = default_registry().register(Box::new(metric.clone()));
    metric
});

static TSO_PRIMARY_TAKEOVER_DURATION_SECONDS: Lazy<HistogramVec> = Lazy::new(|| {
    let metric = HistogramVec::new(
        HistogramOpts::new(
            "tso_primary_takeover_duration_seconds",
            "Duration of successful local primary takeovers",
        ),
        &["outcome"],
    )
    .expect("build tso_primary_takeover_duration_seconds");
    let _ = default_registry().register(Box::new(metric.clone()));
    metric
});

static TSO_FAILOVER_EVENTS_TOTAL: Lazy<IntCounterVec> = Lazy::new(|| {
    let metric = IntCounterVec::new(
        Opts::new("tso_failover_events_total", "Failover events by reason"),
        &["reason"],
    )
    .expect("build tso_failover_events_total");
    let _ = default_registry().register(Box::new(metric.clone()));
    metric
});

static TSO_DRAIN_HANDOFF_DURATION_SECONDS: Lazy<HistogramVec> = Lazy::new(|| {
    let metric = HistogramVec::new(
        HistogramOpts::new(
            "tso_drain_handoff_duration_seconds",
            "Duration of drain handoff attempts",
        ),
        &["outcome"],
    )
    .expect("build tso_drain_handoff_duration_seconds");
    let _ = default_registry().register(Box::new(metric.clone()));
    metric
});

pub(crate) static TSO_PREFETCH_STALL_TOTAL: Lazy<IntCounter> = Lazy::new(|| {
    let metric = IntCounter::new(
        "tso_prefetch_stall_total",
        "Prefetch stalls that fall back to sync request-path updates",
    )
    .expect("build tso_prefetch_stall_total");
    let _ = default_registry().register(Box::new(metric.clone()));
    metric
});

pub(crate) static TSO_PREFETCH_DURATION_SECONDS: Lazy<HistogramVec> = Lazy::new(|| {
    let metric = HistogramVec::new(
        HistogramOpts::new(
            "tso_prefetch_duration_seconds",
            "Latency distribution for prefetch execution",
        ),
        &["outcome"],
    )
    .expect("build tso_prefetch_duration_seconds");
    let _ = default_registry().register(Box::new(metric.clone()));
    metric
});

pub(crate) static TSO_PREFETCH_ERROR_TOTAL: Lazy<IntCounterVec> = Lazy::new(|| {
    let metric = IntCounterVec::new(
        Opts::new(
            "tso_prefetch_error_total",
            "Prefetch errors grouped by reason",
        ),
        &["reason"],
    )
    .expect("build tso_prefetch_error_total");
    let _ = default_registry().register(Box::new(metric.clone()));
    metric
});

static TSO_PREFETCH_EPOCH_MISMATCH_TOTAL: Lazy<IntCounter> = Lazy::new(|| {
    let metric = IntCounter::new(
        "tso_prefetch_epoch_mismatch_total",
        "Prefetch runs skipped due to config-epoch mismatch during execution",
    )
    .expect("build tso_prefetch_epoch_mismatch_total");
    let _ = default_registry().register(Box::new(metric.clone()));
    metric
});

static TSO_PHYSICAL_TIME_CLOCK_SKEW_SECONDS: Lazy<Gauge> = Lazy::new(|| {
    let metric = Gauge::new(
        "tso_physical_time_clock_skew_seconds",
        "Absolute skew between in-memory physical timestamp and host wall clock",
    )
    .expect("build tso_physical_time_clock_skew_seconds");
    let _ = default_registry().register(Box::new(metric.clone()));
    metric
});

static TSO_PHYSICAL_TIME_EXTERNAL_CLOCK_SKEW_SECONDS: Lazy<Gauge> = Lazy::new(|| {
    let metric = Gauge::new(
        "tso_physical_time_external_clock_skew_seconds",
        "Absolute skew between in-memory physical timestamp and external NTP source",
    )
    .expect("build tso_physical_time_external_clock_skew_seconds");
    let _ = default_registry().register(Box::new(metric.clone()));
    metric
});

static TSO_PHYSICAL_TIME_EXTERNAL_CLOCK_SOURCE_ERRORS_TOTAL: Lazy<IntCounter> = Lazy::new(|| {
    let metric = IntCounter::new(
        "tso_physical_time_external_clock_source_errors_total",
        "Errors while polling external NTP clock source for skew monitoring",
    )
    .expect("build tso_physical_time_external_clock_source_errors_total");
    let _ = default_registry().register(Box::new(metric.clone()));
    metric
});

static ETCD_WATCH_CURRENT_REVISION: Lazy<IntGaugeVec> = Lazy::new(|| {
    let metric = IntGaugeVec::new(
        Opts::new(
            "etcd_watch_current_revision",
            "Latest etcd revision observed by watch path",
        ),
        &["path"],
    )
    .expect("build etcd_watch_current_revision");
    let _ = default_registry().register(Box::new(metric.clone()));
    metric
});

static ASSIGNED_GROUP_QPS_ON_NODE: Lazy<Gauge> = Lazy::new(|| {
    let metric = Gauge::new(
        "assigned_group_qps_on_node",
        "Estimated aggregate TSO request rate across groups assigned to this node",
    )
    .expect("build assigned_group_qps_on_node");
    let _ = default_registry().register(Box::new(metric.clone()));
    metric
});

static ASSIGNED_GROUP_QPS_STALE_GROUPS: Lazy<IntGauge> = Lazy::new(|| {
    let metric = IntGauge::new(
        "assigned_group_qps_stale_groups",
        "Number of assigned groups with stale or missing request-rate samples",
    )
    .expect("build assigned_group_qps_stale_groups");
    let _ = default_registry().register(Box::new(metric.clone()));
    metric
});

static GROUP_REBALANCE_OPERATIONS_TOTAL: Lazy<IntCounter> = Lazy::new(|| {
    let metric = IntCounter::new(
        "group_rebalance_operations_total",
        "Total number of accepted group rebalance operations",
    )
    .expect("build group_rebalance_operations_total");
    let _ = default_registry().register(Box::new(metric.clone()));
    metric
});

static GET_MINTS_REQUESTS_TOTAL: Lazy<IntCounterVec> = Lazy::new(|| {
    let metric = IntCounterVec::new(
        Opts::new(
            "get_mints_requests_total",
            "Total number of GetMinTS requests grouped by outcome",
        ),
        &["outcome"],
    )
    .expect("build get_mints_requests_total");
    let _ = default_registry().register(Box::new(metric.clone()));
    metric
});

static GET_MINTS_RATE_LIMITED_TOTAL: Lazy<IntCounter> = Lazy::new(|| {
    let metric = IntCounter::new(
        "get_mints_rate_limited_total",
        "GetMinTS requests denied by rate limiter",
    )
    .expect("build get_mints_rate_limited_total");
    let _ = default_registry().register(Box::new(metric.clone()));
    metric
});

static GET_MINTS_AUTHZ_DENIED_TOTAL: Lazy<IntCounter> = Lazy::new(|| {
    let metric = IntCounter::new(
        "get_mints_authz_denied_total",
        "GetMinTS requests denied by authorization guardrails",
    )
    .expect("build get_mints_authz_denied_total");
    let _ = default_registry().register(Box::new(metric.clone()));
    metric
});

static METRICS_DROPPED_SERIES_TOTAL: Lazy<IntCounterVec> = Lazy::new(|| {
    let metric = IntCounterVec::new(
        Opts::new(
            "metrics_dropped_series_total",
            "Dropped metrics series due to cardinality guardrails",
        ),
        &["metric"],
    )
    .expect("build metrics_dropped_series_total");
    let _ = default_registry().register(Box::new(metric.clone()));
    metric
});

fn observe_dropped_series(metric: &'static str) {
    METRICS_DROPPED_SERIES_TOTAL
        .with_label_values(&[metric])
        .inc();
}

fn normalize_label<'a>(metric: &'static str, label: &'a str, allowed: &[&str]) -> &'a str {
    if allowed.contains(&label) {
        label
    } else {
        observe_dropped_series(metric);
        "other"
    }
}

fn normalize_watch_path(path: &str) -> &'static str {
    if path.contains("/tso/keyspace_groups/membership/") {
        "membership"
    } else {
        observe_dropped_series("etcd_watch_current_revision");
        "other"
    }
}

/// Records one TSO request outcome and latency sample.
pub fn observe_tso_request(success: bool, count: u32, latency_seconds: f64) {
    let outcome = if success { "success" } else { "error" };
    TSO_REQUESTS_TOTAL.with_label_values(&[outcome]).inc();
    if success {
        TSO_TIMESTAMPS_TOTAL.inc_by(u64::from(count));
    }
    TSO_REQUEST_LATENCY_SECONDS
        .with_label_values(&[outcome])
        .observe(latency_seconds);
}

/// Records one config-apply attempt.
pub fn config_apply_attempt(endpoint: &str) {
    const ALLOWED_ENDPOINTS: [&str; 1] = ["apply_config"];
    let endpoint = normalize_label("config_apply_attempts_total", endpoint, &ALLOWED_ENDPOINTS);
    CONFIG_APPLY_ATTEMPTS_TOTAL
        .with_label_values(&[endpoint])
        .inc();
}

/// Records one config-apply failure.
pub fn config_apply_failure(reason: &str) {
    const ALLOWED_REASONS: [&str; 6] = [
        "sync_store",
        "validation",
        "restart_required",
        "epoch_apply",
        "persist",
        "other",
    ];
    let reason = normalize_label("config_apply_failures_total", reason, &ALLOWED_REASONS);
    CONFIG_APPLY_FAILURES_TOTAL
        .with_label_values(&[reason])
        .inc();
}

/// Sets current active config epoch gauge.
pub fn set_active_config_epoch(epoch: u64) {
    let value = i64::try_from(epoch).unwrap_or(i64::MAX);
    ACTIVE_CONFIG_EPOCH.set(value);
}

/// Sets current pending actor count for in-flight config apply.
pub fn set_config_apply_pending_actors(count: usize) {
    let value = i64::try_from(count).unwrap_or(i64::MAX);
    CONFIG_APPLY_PENDING_ACTORS.set(value);
}

/// RAII helper that resets pending actor gauge when config-apply scope exits.
pub struct ConfigApplyPendingGuard;

impl ConfigApplyPendingGuard {
    /// Sets pending-actor gauge for current config-apply scope.
    #[must_use]
    pub fn new(count: usize) -> Self {
        set_config_apply_pending_actors(count);
        Self
    }
}

impl Drop for ConfigApplyPendingGuard {
    fn drop(&mut self) {
        set_config_apply_pending_actors(0);
    }
}

/// Records one etcd operation latency sample and failure count when applicable.
pub fn observe_etcd_request(operation: &str, success: bool, latency_seconds: f64) {
    const ALLOWED_OPERATIONS: [&str; 11] = [
        "connect",
        "watch",
        "get_membership_snapshot",
        "lease_grant",
        "txn_campaign",
        "lease_keep_alive_open",
        "lease_keep_alive_send",
        "lease_keep_alive_recv",
        "get_value",
        "save_timestamp",
        "load_timestamp",
    ];
    let operation = normalize_label(
        "etcd_request_latency_seconds",
        operation,
        &ALLOWED_OPERATIONS,
    );
    let outcome = if success { "success" } else { "error" };
    ETCD_REQUEST_LATENCY_SECONDS
        .with_label_values(&[operation, outcome])
        .observe(latency_seconds);
    if !success {
        ETCD_REQUESTS_FAILED_TOTAL
            .with_label_values(&[operation])
            .inc();
    }
}

/// Updates node-level actor mailbox depth gauge.
pub fn set_group_actor_mailbox_size(total_depth: u32) {
    GROUP_ACTOR_MAILBOX_SIZE.set(i64::from(total_depth));
}

/// Records one actor restart event.
pub fn observe_group_actor_restart(reason: &str) {
    const ALLOWED_REASONS: [&str; 13] = [
        "expected_primary_sync_error",
        "expected_primary_mismatch",
        "lease_grant_failed",
        "campaign_state_update_failed",
        "campaign_refresh_failed",
        "lost_campaign",
        "leader_read_failed",
        "campaign_txn_failed",
        "local_primary_check_failed",
        "lease_rejected",
        "keepalive_failed",
        "keepalive_error",
        "other",
    ];
    let reason = normalize_label("group_actor_restarts_total", reason, &ALLOWED_REASONS);
    GROUP_ACTOR_RESTARTS_TOTAL
        .with_label_values(&[reason])
        .inc();
}

/// Updates node-level election and lease gauges.
pub fn set_election_and_lease_status(has_primary: bool, has_lease: bool) {
    TSO_ELECTION_STATUS.set(if has_primary { 1 } else { 0 });
    TSO_LEASE_STATUS.set(if has_lease { 1 } else { 0 });
}

/// Records successful campaign execution.
pub fn observe_election_campaign() {
    TSO_ELECTION_CAMPAIGNS_TOTAL.inc();
}

/// Records one primary takeover duration sample.
pub fn observe_primary_takeover_duration(success: bool, latency_seconds: f64) {
    let outcome = if success { "success" } else { "error" };
    TSO_PRIMARY_TAKEOVER_DURATION_SECONDS
        .with_label_values(&[outcome])
        .observe(latency_seconds);
}

/// Records a failover event by reason.
pub fn observe_failover_event(reason: &str) {
    const ALLOWED_REASONS: [&str; 5] = [
        "expected_primary_mismatch",
        "lost_campaign",
        "local_primary_check_failed",
        "lease_rejected",
        "other",
    ];
    let reason = normalize_label("tso_failover_events_total", reason, &ALLOWED_REASONS);
    TSO_FAILOVER_EVENTS_TOTAL.with_label_values(&[reason]).inc();
}

/// Records one drain handoff duration sample.
pub fn observe_drain_handoff_duration(success: bool, latency_seconds: f64) {
    let outcome = if success { "success" } else { "timeout" };
    TSO_DRAIN_HANDOFF_DURATION_SECONDS
        .with_label_values(&[outcome])
        .observe(latency_seconds);
}

/// Records one prefetch execution duration sample.
pub fn observe_prefetch_duration(success: bool, latency_seconds: f64) {
    let outcome = if success { "success" } else { "error" };
    TSO_PREFETCH_DURATION_SECONDS
        .with_label_values(&[outcome])
        .observe(latency_seconds);
}

/// Records one or more prefetch stall events.
pub fn observe_prefetch_stall(count: u32) {
    if count == 0 {
        return;
    }
    TSO_PREFETCH_STALL_TOTAL.inc_by(u64::from(count));
}

/// Records one or more prefetch errors by reason.
pub fn observe_prefetch_error(reason: &str, count: u32) {
    if count == 0 {
        return;
    }
    const ALLOWED_REASONS: [&str; 2] = ["prefetch_failed", "other"];
    let reason = normalize_label("tso_prefetch_error_total", reason, &ALLOWED_REASONS);
    TSO_PREFETCH_ERROR_TOTAL
        .with_label_values(&[reason])
        .inc_by(u64::from(count));
}

/// Records one prefetch config-epoch mismatch.
pub fn observe_prefetch_epoch_mismatch() {
    TSO_PREFETCH_EPOCH_MISMATCH_TOTAL.inc();
}

/// Updates absolute physical clock skew gauge.
pub fn set_physical_time_clock_skew_seconds(skew_seconds: f64) {
    TSO_PHYSICAL_TIME_CLOCK_SKEW_SECONDS.set(skew_seconds.max(0.0));
}

/// Updates absolute physical clock skew gauge against external time source.
pub fn set_physical_time_external_clock_skew_seconds(skew_seconds: f64) {
    TSO_PHYSICAL_TIME_EXTERNAL_CLOCK_SKEW_SECONDS.set(skew_seconds.max(0.0));
}

/// Records one external time-source polling failure.
pub fn observe_physical_time_external_clock_source_error() {
    TSO_PHYSICAL_TIME_EXTERNAL_CLOCK_SOURCE_ERRORS_TOTAL.inc();
}

/// Updates latest observed etcd watch revision for one path.
pub fn set_etcd_watch_current_revision(path: &str, revision: i64) {
    let path = normalize_watch_path(path);
    ETCD_WATCH_CURRENT_REVISION
        .with_label_values(&[path])
        .set(revision.max(0));
}

/// Updates estimated aggregate assigned-group request rate gauge.
pub fn set_assigned_group_qps_on_node(qps: f64) {
    ASSIGNED_GROUP_QPS_ON_NODE.set(qps.max(0.0));
}

/// Updates count of groups with stale/missing request-rate samples.
pub fn set_assigned_group_qps_stale_groups(stale_groups: u64) {
    ASSIGNED_GROUP_QPS_STALE_GROUPS.set(i64::try_from(stale_groups).unwrap_or(i64::MAX));
}

/// Records one accepted rebalance operation.
pub fn observe_group_rebalance_operation() {
    GROUP_REBALANCE_OPERATIONS_TOTAL.inc();
}

/// Records one GetMinTS request outcome.
pub fn observe_get_min_ts_request(outcome: &str) {
    const ALLOWED_OUTCOMES: [&str; 5] =
        ["success", "error", "rate_limited", "authz_denied", "other"];
    let outcome = normalize_label("get_mints_requests_total", outcome, &ALLOWED_OUTCOMES);
    GET_MINTS_REQUESTS_TOTAL.with_label_values(&[outcome]).inc();
}

/// Records one GetMinTS rate-limiter rejection.
pub fn observe_get_min_ts_rate_limited() {
    GET_MINTS_RATE_LIMITED_TOTAL.inc();
}

/// Records one GetMinTS authorization rejection.
pub fn observe_get_min_ts_authz_denied() {
    GET_MINTS_AUTHZ_DENIED_TOTAL.inc();
}

#[cfg(test)]
pub fn gather_metric_names() -> Vec<String> {
    prometheus::gather()
        .into_iter()
        .map(|family| family.get_name().to_string())
        .collect()
}

#[cfg(test)]
mod tests {
    use super::{
        TSO_PREFETCH_DURATION_SECONDS, TSO_PREFETCH_ERROR_TOTAL, TSO_PREFETCH_STALL_TOTAL,
        gather_metric_names, observe_drain_handoff_duration, observe_get_min_ts_authz_denied,
        observe_group_actor_restart, observe_group_rebalance_operation,
        observe_physical_time_external_clock_source_error, observe_prefetch_duration,
        observe_prefetch_epoch_mismatch, observe_prefetch_error, observe_prefetch_stall,
        set_assigned_group_qps_on_node, set_assigned_group_qps_stale_groups,
        set_etcd_watch_current_revision, set_physical_time_clock_skew_seconds,
        set_physical_time_external_clock_skew_seconds,
    };

    #[test]
    fn newly_added_metric_families_register_on_first_use() {
        observe_group_actor_restart("test");
        observe_prefetch_stall(1);
        observe_prefetch_duration(true, 0.001);
        observe_prefetch_error("test", 1);
        observe_prefetch_epoch_mismatch();
        set_physical_time_clock_skew_seconds(0.25);
        set_physical_time_external_clock_skew_seconds(0.125);
        observe_physical_time_external_clock_source_error();
        set_etcd_watch_current_revision("/pd/1/tso/keyspace_groups/membership/", 42);
        observe_drain_handoff_duration(true, 0.01);
        set_assigned_group_qps_on_node(123.4);
        set_assigned_group_qps_stale_groups(2);
        observe_group_rebalance_operation();
        observe_get_min_ts_authz_denied();
        observe_prefetch_error("not-allowed-token", 1);
        set_etcd_watch_current_revision("/not/allowed/path", 1);

        let names = gather_metric_names();
        assert!(
            names
                .iter()
                .any(|name| name == "group_actor_restarts_total")
        );
        assert!(names.iter().any(|name| name == "tso_prefetch_stall_total"));
        assert!(
            names
                .iter()
                .any(|name| name == "tso_prefetch_duration_seconds")
        );
        assert!(names.iter().any(|name| name == "tso_prefetch_error_total"));
        assert!(
            names
                .iter()
                .any(|name| name == "tso_prefetch_epoch_mismatch_total")
        );
        assert!(
            names
                .iter()
                .any(|name| name == "tso_physical_time_clock_skew_seconds")
        );
        assert!(
            names
                .iter()
                .any(|name| name == "tso_physical_time_external_clock_skew_seconds")
        );
        assert!(
            names
                .iter()
                .any(|name| { name == "tso_physical_time_external_clock_source_errors_total" })
        );
        assert!(
            names
                .iter()
                .any(|name| name == "etcd_watch_current_revision")
        );
        assert!(
            names
                .iter()
                .any(|name| name == "tso_drain_handoff_duration_seconds")
        );
        assert!(
            names
                .iter()
                .any(|name| name == "assigned_group_qps_on_node")
        );
        assert!(
            names
                .iter()
                .any(|name| name == "assigned_group_qps_stale_groups")
        );
        assert!(
            names
                .iter()
                .any(|name| name == "group_rebalance_operations_total")
        );
        assert!(
            names
                .iter()
                .any(|name| name == "get_mints_authz_denied_total")
        );
        assert!(
            names
                .iter()
                .any(|name| name == "metrics_dropped_series_total")
        );
        assert!(TSO_PREFETCH_STALL_TOTAL.get() > 0);
        assert!(
            TSO_PREFETCH_DURATION_SECONDS
                .with_label_values(&["success"])
                .get_sample_count()
                > 0
        );
        assert!(TSO_PREFETCH_ERROR_TOTAL.with_label_values(&["other"]).get() > 0);
    }

    #[test]
    fn runtime_restart_reasons_are_allowlisted() {
        const RUNTIME_RESTART_REASONS: [&str; 12] = [
            "expected_primary_sync_error",
            "expected_primary_mismatch",
            "lease_grant_failed",
            "campaign_state_update_failed",
            "campaign_refresh_failed",
            "lost_campaign",
            "leader_read_failed",
            "campaign_txn_failed",
            "local_primary_check_failed",
            "lease_rejected",
            "keepalive_failed",
            "keepalive_error",
        ];

        for reason in RUNTIME_RESTART_REASONS {
            observe_group_actor_restart(reason);
        }

        let group_actor_restart_metric = prometheus::gather()
            .into_iter()
            .find(|family| family.get_name() == "group_actor_restarts_total")
            .expect("group_actor_restarts_total metric should be registered");

        for expected_reason in RUNTIME_RESTART_REASONS {
            let seen = group_actor_restart_metric
                .get_metric()
                .iter()
                .any(|metric| {
                    metric.get_label().iter().any(|label| {
                        label.get_name() == "reason" && label.get_value() == expected_reason
                    })
                });
            assert!(
                seen,
                "runtime restart reason '{expected_reason}' is not allowlisted in group_actor_restarts_total"
            );
        }
    }
}