razeri_plugin_api/

alerting.rs

// Copyright (C) 2026 Gabriel Shea
// Licensed under either of the Apache License, Version 2.0 or the MIT
// license, at your option.

//! Alerting and health summary model for the canonical event stream.
//!
//! This module generalizes alerting beyond storage-specific metrics,
//! operating on [`RazeriEvent`] data with deterministic, pure evaluation
//! functions. Alert rules support threshold, windowed, and absence
//! conditions with optional deduplication.
//!
//! Health summaries aggregate event streams into operator-friendly status
//! indicators for workflows, categories, sources, entities, plugins, and
//! sinks.

use crate::event::{EventCategory, EventSeverity, EventSubjectKind, RazeriEvent};
use chrono::{DateTime, TimeDelta, Utc};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;

// ── Alert types ────────────────────────────────────────────────────────

/// Scope of events an alert rule monitors.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case", tag = "type")]
#[non_exhaustive]
pub enum AlertScope {
    /// All events regardless of category.
    Global,
    /// Events matching a specific category.
    Category {
        /// The category to filter on.
        category: EventCategory,
    },
}

/// Comparison operator for alert condition evaluation.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ComparisonOp {
    /// `>`
    GreaterThan,
    /// `>=`
    GreaterThanOrEqual,
    /// `<`
    LessThan,
    /// `<=`
    LessThanOrEqual,
    /// `==`
    Equal,
}

/// What aspect of the event stream to measure.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case", tag = "type")]
#[non_exhaustive]
pub enum AlertMetric {
    /// Total count of matching events.
    EventCount,
    /// Count of events with Error or Critical severity.
    ErrorCount,
    /// Count of events with a specific payload type name (e.g. `"materialization_failed"`).
    PayloadCount {
        /// Payload type name to count (matches `EventPayload::name()`).
        payload_name: String,
    },
}

/// How an alert rule evaluates the event stream.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case", tag = "type")]
#[non_exhaustive]
pub enum AlertCondition {
    /// Fire when absolute event count matches the comparison.
    Threshold {
        /// What to measure.
        metric: AlertMetric,
        /// Comparison operator.
        op: ComparisonOp,
        /// Threshold value.
        value: u64,
    },
    /// Fire when event count within a time window matches the comparison.
    Rate {
        /// What to measure.
        metric: AlertMetric,
        /// Rolling window size in seconds.
        window_secs: u64,
        /// Comparison operator.
        op: ComparisonOp,
        /// Threshold value.
        value: u64,
    },
    /// Fire when no matching events appear within the time window.
    Absence {
        /// Silence window in seconds.
        window_secs: u64,
    },
}

/// A named alerting rule definition.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AlertRule {
    /// Unique rule name.
    pub name: String,
    /// Which events this rule monitors.
    pub scope: AlertScope,
    /// How the rule evaluates the event stream.
    pub condition: AlertCondition,
    /// Severity assigned when the rule fires.
    pub severity: EventSeverity,
    /// Human-readable message describing the alert.
    pub message: String,
    /// Whether this rule is active.
    pub enabled: bool,
    /// Deduplication window in seconds. If set, the rule will not fire again
    /// if it already fired within this window.
    pub dedup_window_secs: Option<u64>,
}

/// Result of evaluating a single alert rule against the event stream.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AlertResult {
    /// Name of the rule that was evaluated.
    pub rule_name: String,
    /// Whether the alert condition was met and not deduplicated.
    pub fired: bool,
    /// Severity from the rule definition.
    pub severity: EventSeverity,
    /// Message from the rule definition.
    pub message: String,
    /// The measured value from the event stream.
    pub current_value: f64,
    /// The threshold value from the rule condition.
    pub threshold: f64,
    /// When this evaluation occurred.
    pub evaluated_at: DateTime<Utc>,
    /// Key used for deduplication.
    pub dedup_key: String,
}

// ── Health summary types ───────────────────────────────────────────────

/// Domain for which health is being summarized.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case", tag = "kind", content = "name")]
#[non_exhaustive]
pub enum HealthDomain {
    /// Health across all events.
    Overall,
    /// Health for a specific event category.
    Category(EventCategory),
    /// Health for a specific workflow.
    Workflow(String),
    /// Health for a named source.
    Source(String),
    /// Health for a named entity.
    Entity(String),
    /// Health for a named plugin.
    Plugin(String),
    /// Health for a named event sink.
    Sink(String),
}

/// Overall health status indicator.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum HealthStatus {
    /// No errors or warnings detected.
    Healthy,
    /// Warnings detected but no errors.
    Degraded,
    /// Errors or critical events detected.
    Unhealthy,
    /// No events observed (insufficient data).
    Unknown,
}

/// Aggregated health for a specific domain.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HealthSummary {
    /// Which domain this summary covers.
    pub domain: HealthDomain,
    /// Derived health status.
    pub status: HealthStatus,
    /// Total events counted.
    pub total_events: usize,
    /// Events with Error or Critical severity.
    pub error_count: usize,
    /// Events with Warning severity.
    pub warning_count: usize,
    /// Timestamp of the most recent event, if any.
    pub last_event_at: Option<DateTime<Utc>>,
    /// Names of active alerts relevant to this domain.
    pub active_alerts: Vec<String>,
}

/// External sink status for inclusion in health summaries.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SinkStatus {
    /// Sink name.
    pub name: String,
    /// Number of errors encountered.
    pub errors: usize,
    /// Most recent error message, if any.
    pub last_error: Option<String>,
}

// ── Pure evaluation functions ──────────────────────────────────────────

/// Evaluate alert rules against a canonical event stream.
///
/// This is a pure, deterministic function: identical inputs always produce
/// identical outputs. Deduplication uses `previously_fired` results to
/// suppress repeated firings within the dedup window.
#[must_use]
pub fn evaluate_rules(
    rules: &[AlertRule],
    events: &[RazeriEvent],
    now: DateTime<Utc>,
    previously_fired: &[AlertResult],
) -> Vec<AlertResult> {
    let mut results = Vec::new();

    for rule in rules {
        if !rule.enabled {
            continue;
        }

        let scoped: Vec<&RazeriEvent> =
            events.iter().filter(|e| matches_scope(&rule.scope, e)).collect();

        let (condition_met, current_value, threshold) =
            evaluate_condition(&rule.condition, &scoped, now);

        let dedup_key = rule.name.clone();
        let is_deduped = is_deduplicated(&dedup_key, rule.dedup_window_secs, now, previously_fired);

        results.push(AlertResult {
            rule_name: rule.name.clone(),
            fired: condition_met && !is_deduped,
            severity: rule.severity,
            message: rule.message.clone(),
            current_value,
            threshold,
            evaluated_at: now,
            dedup_key,
        });
    }

    results
}

/// Compute a health summary for a specific domain from the event stream.
#[must_use]
pub fn health_for_domain(
    domain: &HealthDomain,
    events: &[RazeriEvent],
    active_alerts: &[AlertResult],
) -> HealthSummary {
    let domain_events: Vec<&RazeriEvent> =
        events.iter().filter(|e| event_matches_domain(domain, e)).collect();

    let fired_names: Vec<String> =
        active_alerts.iter().filter(|a| a.fired).map(|a| a.rule_name.clone()).collect();

    summarize_events(domain.clone(), &domain_events, &fired_names)
}

/// Auto-discover health domains from the event stream and produce summaries.
///
/// Returns summaries for: Overall, each event category, each subject
/// (Source, Entity, Plugin), each workflow, and each sink (from external status).
/// Output order is deterministic for the same inputs.
#[must_use]
pub fn discover_health(
    events: &[RazeriEvent],
    active_alerts: &[AlertResult],
    sink_statuses: &[SinkStatus],
) -> Vec<HealthSummary> {
    let mut summaries = vec![build_overall_summary(events, active_alerts)];
    let fired_names = fired_alert_names(active_alerts);

    summaries.extend(build_category_summaries(events, &fired_names));
    summaries.extend(build_subject_summaries(events, &fired_names));
    summaries.extend(build_workflow_summaries(events, &fired_names));
    summaries.extend(build_sink_summaries(sink_statuses));

    summaries
}

// ── Internal helpers ───────────────────────────────────────────────────

/// Build the overall health summary from all events.
/// Parameters:
/// - `events`: Full canonical event stream.
/// - `active_alerts`: Evaluated alert results.
fn build_overall_summary(events: &[RazeriEvent], active_alerts: &[AlertResult]) -> HealthSummary {
    let fired_names = fired_alert_names(active_alerts);
    summarize_events(HealthDomain::Overall, &events.iter().collect::<Vec<_>>(), &fired_names)
}

/// Collect the names of currently firing alerts.
/// Parameters:
/// - `active_alerts`: Evaluated alert results.
fn fired_alert_names(active_alerts: &[AlertResult]) -> Vec<String> {
    active_alerts.iter().filter(|a| a.fired).map(|a| a.rule_name.clone()).collect()
}

/// Build per-category health summaries in deterministic category order.
/// Parameters:
/// - `events`: Full canonical event stream.
/// - `fired_names`: Names of currently firing alerts.
fn build_category_summaries(events: &[RazeriEvent], fired_names: &[String]) -> Vec<HealthSummary> {
    let mut by_category: HashMap<EventCategory, Vec<&RazeriEvent>> = HashMap::new();
    for event in events {
        by_category.entry(event.category()).or_default().push(event);
    }

    let mut categories: Vec<EventCategory> = by_category.keys().copied().collect();
    categories.sort_by_key(|c| format!("{c:?}"));

    categories
        .into_iter()
        .filter_map(|cat| {
            by_category.get(&cat).map(|cat_events| {
                summarize_events(HealthDomain::Category(cat), cat_events, fired_names)
            })
        })
        .collect()
}

/// Build Source/Entity/Plugin subject summaries in deterministic key order.
/// Parameters:
/// - `events`: Full canonical event stream.
/// - `fired_names`: Names of currently firing alerts.
fn build_subject_summaries(events: &[RazeriEvent], fired_names: &[String]) -> Vec<HealthSummary> {
    let mut by_subject: HashMap<(EventSubjectKind, String), Vec<&RazeriEvent>> = HashMap::new();
    for event in events {
        let key = (event.subject().kind, event.subject().name.clone());
        by_subject.entry(key).or_default().push(event);
    }

    let mut subject_keys: Vec<(EventSubjectKind, String)> = by_subject.keys().cloned().collect();
    subject_keys
        .sort_by(|a, b| format!("{:?}", a.0).cmp(&format!("{:?}", b.0)).then(a.1.cmp(&b.1)));

    subject_keys
        .into_iter()
        .filter_map(|(kind, name)| {
            let domain = match kind {
                EventSubjectKind::Source => HealthDomain::Source(name.clone()),
                EventSubjectKind::Entity => HealthDomain::Entity(name.clone()),
                EventSubjectKind::Plugin => HealthDomain::Plugin(name.clone()),
                _ => return None,
            };
            by_subject
                .get(&(kind, name))
                .map(|subj_events| summarize_events(domain, subj_events, fired_names))
        })
        .collect()
}

/// Build per-workflow summaries in deterministic workflow-id order.
/// Parameters:
/// - `events`: Full canonical event stream.
/// - `fired_names`: Names of currently firing alerts.
fn build_workflow_summaries(events: &[RazeriEvent], fired_names: &[String]) -> Vec<HealthSummary> {
    let mut by_workflow: HashMap<String, Vec<&RazeriEvent>> = HashMap::new();
    for event in events {
        if let Some(wf) = event.workflow() {
            by_workflow.entry(wf.workflow_id.clone()).or_default().push(event);
        }
    }

    let mut workflow_ids: Vec<String> = by_workflow.keys().cloned().collect();
    workflow_ids.sort();

    workflow_ids
        .into_iter()
        .filter_map(|wf_id| {
            by_workflow.get(&wf_id).map(|wf_events| {
                summarize_events(HealthDomain::Workflow(wf_id), wf_events, fired_names)
            })
        })
        .collect()
}

/// Convert external sink statuses into health summaries.
/// Parameters:
/// - `sink_statuses`: Sink health state emitted by adapters.
fn build_sink_summaries(sink_statuses: &[SinkStatus]) -> Vec<HealthSummary> {
    sink_statuses
        .iter()
        .map(|sink| {
            let status =
                if sink.errors > 0 { HealthStatus::Unhealthy } else { HealthStatus::Healthy };
            HealthSummary {
                domain: HealthDomain::Sink(sink.name.clone()),
                status,
                total_events: 0,
                error_count: sink.errors,
                warning_count: 0,
                last_event_at: None,
                active_alerts: Vec::new(),
            }
        })
        .collect()
}

fn matches_scope(scope: &AlertScope, event: &RazeriEvent) -> bool {
    match scope {
        AlertScope::Global => true,
        AlertScope::Category { category } => event.category() == *category,
    }
}

#[allow(clippy::cast_precision_loss, clippy::cast_possible_wrap)]
fn evaluate_condition(
    condition: &AlertCondition,
    events: &[&RazeriEvent],
    now: DateTime<Utc>,
) -> (bool, f64, f64) {
    match condition {
        AlertCondition::Threshold { metric, op, value } => {
            let count = count_metric(metric, events) as f64;
            let threshold = *value as f64;
            (compare(*op, count, threshold), count, threshold)
        }
        AlertCondition::Rate { metric, window_secs, op, value } => {
            let cutoff = now - TimeDelta::seconds(*window_secs as i64);
            let windowed: Vec<&RazeriEvent> =
                events.iter().filter(|e| e.timestamp() >= cutoff).copied().collect();
            let count = count_metric(metric, &windowed) as f64;
            let threshold = *value as f64;
            (compare(*op, count, threshold), count, threshold)
        }
        AlertCondition::Absence { window_secs } => {
            let cutoff = now - TimeDelta::seconds(*window_secs as i64);
            let has_recent = events.iter().any(|e| e.timestamp() >= cutoff);
            let last_seen_secs = events
                .iter()
                .map(|e| (now - e.timestamp()).num_seconds().max(0) as f64)
                .fold(f64::INFINITY, f64::min);
            // If no events at all, report window+1 as the gap
            let gap = if last_seen_secs.is_infinite() {
                *window_secs as f64 + 1.0
            } else {
                last_seen_secs
            };
            (!has_recent, gap, *window_secs as f64)
        }
    }
}

fn count_metric(metric: &AlertMetric, events: &[&RazeriEvent]) -> usize {
    match metric {
        AlertMetric::EventCount => events.len(),
        AlertMetric::ErrorCount => {
            events.iter().filter(|e| e.severity() >= EventSeverity::Error).count()
        }
        AlertMetric::PayloadCount { payload_name } => {
            events.iter().filter(|e| e.payload().name() == payload_name.as_str()).count()
        }
    }
}

fn compare(op: ComparisonOp, left: f64, right: f64) -> bool {
    match op {
        ComparisonOp::GreaterThan => left > right,
        ComparisonOp::GreaterThanOrEqual => left >= right,
        ComparisonOp::LessThan => left < right,
        ComparisonOp::LessThanOrEqual => left <= right,
        ComparisonOp::Equal => (left - right).abs() < f64::EPSILON,
    }
}

#[allow(clippy::cast_possible_wrap)]
fn is_deduplicated(
    dedup_key: &str,
    dedup_window_secs: Option<u64>,
    now: DateTime<Utc>,
    previously_fired: &[AlertResult],
) -> bool {
    let Some(window_secs) = dedup_window_secs else {
        return false;
    };
    let cutoff = now - TimeDelta::seconds(window_secs as i64);
    previously_fired
        .iter()
        .any(|prev| prev.fired && prev.dedup_key == dedup_key && prev.evaluated_at >= cutoff)
}

fn event_matches_domain(domain: &HealthDomain, event: &RazeriEvent) -> bool {
    match domain {
        HealthDomain::Overall => true,
        HealthDomain::Category(cat) => event.category() == *cat,
        HealthDomain::Workflow(wf_id) => {
            event.workflow().is_some_and(|wf| wf.workflow_id == *wf_id)
        }
        HealthDomain::Source(name) => {
            event.subject().kind == EventSubjectKind::Source && event.subject().name == *name
        }
        HealthDomain::Entity(name) => {
            event.subject().kind == EventSubjectKind::Entity && event.subject().name == *name
        }
        HealthDomain::Plugin(name) => {
            event.subject().kind == EventSubjectKind::Plugin && event.subject().name == *name
        }
        HealthDomain::Sink(_) => false,
    }
}

fn summarize_events(
    domain: HealthDomain,
    events: &[&RazeriEvent],
    active_alert_names: &[String],
) -> HealthSummary {
    let error_count = events.iter().filter(|e| e.severity() >= EventSeverity::Error).count();
    let warning_count = events.iter().filter(|e| e.severity() == EventSeverity::Warning).count();
    let last_event_at = events.iter().map(|e| e.timestamp()).max();

    let status = if events.is_empty() {
        HealthStatus::Unknown
    } else if error_count > 0 {
        HealthStatus::Unhealthy
    } else if warning_count > 0 {
        HealthStatus::Degraded
    } else {
        HealthStatus::Healthy
    };

    HealthSummary {
        domain,
        status,
        total_events: events.len(),
        error_count,
        warning_count,
        last_event_at,
        active_alerts: active_alert_names.to_vec(),
    }
}

#[cfg(test)]
#[allow(clippy::float_cmp)]
mod tests {
    use super::*;
    use crate::event::WorkflowContext;

    fn ts(offset_secs: i64) -> DateTime<Utc> {
        let base = DateTime::parse_from_rfc3339("2026-06-01T12:00:00Z")
            .expect("valid base timestamp")
            .with_timezone(&Utc);
        base + TimeDelta::seconds(offset_secs)
    }

    fn now() -> DateTime<Utc> {
        ts(0)
    }

    fn wf(id: &str) -> WorkflowContext {
        WorkflowContext::new(id).expect("valid workflow id")
    }

    fn info_event(source: &str, offset_secs: i64) -> RazeriEvent {
        RazeriEvent::profiling_completed(
            format!("evt-info-{source}-{offset_secs}"),
            ts(offset_secs),
            wf("wf-1"),
            source,
            10,
            1000,
            42,
        )
        .expect("valid info event")
    }

    fn error_event(design: &str, offset_secs: i64) -> RazeriEvent {
        RazeriEvent::materialization_failed(
            format!("evt-err-{design}-{offset_secs}"),
            ts(offset_secs),
            wf("wf-1"),
            design,
            "PostgreSQL",
            "connection refused",
        )
        .expect("valid error event")
    }

    fn warning_event(design: &str, offset_secs: i64) -> RazeriEvent {
        RazeriEvent::design_validated(
            format!("evt-warn-{design}-{offset_secs}"),
            ts(offset_secs),
            wf("wf-1"),
            design,
            0,
            3,
        )
        .expect("valid warning event")
    }

    fn plugin_event(name: &str, offset_secs: i64) -> RazeriEvent {
        RazeriEvent::plugin_loaded(
            format!("evt-plugin-{name}-{offset_secs}"),
            ts(offset_secs),
            name,
            "dialect",
            "builtin",
        )
        .expect("valid plugin event")
    }

    fn threshold_rule(name: &str, metric: AlertMetric, value: u64) -> AlertRule {
        AlertRule {
            name: name.to_string(),
            scope: AlertScope::Global,
            condition: AlertCondition::Threshold { metric, op: ComparisonOp::GreaterThan, value },
            severity: EventSeverity::Error,
            message: format!("{name} triggered"),
            enabled: true,
            dedup_window_secs: None,
        }
    }

    fn rate_rule(name: &str, metric: AlertMetric, window_secs: u64, value: u64) -> AlertRule {
        AlertRule {
            name: name.to_string(),
            scope: AlertScope::Global,
            condition: AlertCondition::Rate {
                metric,
                window_secs,
                op: ComparisonOp::GreaterThan,
                value,
            },
            severity: EventSeverity::Warning,
            message: format!("{name} triggered"),
            enabled: true,
            dedup_window_secs: None,
        }
    }

    fn absence_rule(name: &str, window_secs: u64) -> AlertRule {
        AlertRule {
            name: name.to_string(),
            scope: AlertScope::Global,
            condition: AlertCondition::Absence { window_secs },
            severity: EventSeverity::Critical,
            message: format!("{name} triggered"),
            enabled: true,
            dedup_window_secs: None,
        }
    }

    // ── Alert evaluation tests ─────────────────────────────────────────

    #[test]
    fn threshold_fires_for_error_count_above_value() {
        let events = vec![
            error_event("design-a", -10),
            error_event("design-b", -20),
            error_event("design-c", -30),
        ];
        let rules = [threshold_rule("high-errors", AlertMetric::ErrorCount, 2)];
        let results = evaluate_rules(&rules, &events, now(), &[]);

        assert_eq!(results.len(), 1);
        assert!(results[0].fired);
        assert_eq!(results[0].current_value, 3.0);
        assert_eq!(results[0].threshold, 2.0);
    }

    #[test]
    fn threshold_does_not_fire_below_value() {
        let events = vec![error_event("design-a", -10)];
        let rules = [threshold_rule("high-errors", AlertMetric::ErrorCount, 2)];
        let results = evaluate_rules(&rules, &events, now(), &[]);

        assert_eq!(results.len(), 1);
        assert!(!results[0].fired);
        assert_eq!(results[0].current_value, 1.0);
    }

    #[test]
    fn rate_filters_events_by_window() {
        let events = vec![
            info_event("a.csv", -10),   // within 60s window
            info_event("b.csv", -30),   // within 60s window
            info_event("c.csv", -50),   // within 60s window
            info_event("d.csv", -120),  // outside 60s window
            info_event("e.csv", -3600), // far outside window
        ];
        let rules = [rate_rule("high-event-rate", AlertMetric::EventCount, 60, 2)];
        let results = evaluate_rules(&rules, &events, now(), &[]);

        assert_eq!(results.len(), 1);
        assert!(results[0].fired);
        assert_eq!(results[0].current_value, 3.0);
    }

    #[test]
    fn rate_does_not_fire_below_threshold_in_window() {
        let events = vec![
            info_event("a.csv", -10),   // within 60s window
            info_event("b.csv", -3600), // outside 60s window
        ];
        let rules = [rate_rule("high-event-rate", AlertMetric::EventCount, 60, 2)];
        let results = evaluate_rules(&rules, &events, now(), &[]);

        assert_eq!(results.len(), 1);
        assert!(!results[0].fired);
        assert_eq!(results[0].current_value, 1.0);
    }

    #[test]
    fn absence_fires_when_no_recent_events() {
        let events = vec![
            info_event("a.csv", -7200), // 2 hours ago
        ];
        let rules = [absence_rule("heartbeat", 3600)]; // window = 1 hour
        let results = evaluate_rules(&rules, &events, now(), &[]);

        assert_eq!(results.len(), 1);
        assert!(results[0].fired);
    }

    #[test]
    fn absence_does_not_fire_with_recent_events() {
        let events = vec![
            info_event("a.csv", -300), // 5 minutes ago
        ];
        let rules = [absence_rule("heartbeat", 3600)]; // window = 1 hour
        let results = evaluate_rules(&rules, &events, now(), &[]);

        assert_eq!(results.len(), 1);
        assert!(!results[0].fired);
    }

    #[test]
    fn absence_fires_when_no_events_at_all() {
        let rules = [absence_rule("heartbeat", 3600)];
        let results = evaluate_rules(&rules, &[], now(), &[]);

        assert_eq!(results.len(), 1);
        assert!(results[0].fired);
        assert!(results[0].current_value > 3600.0);
    }

    #[test]
    fn dedup_suppresses_repeated_firing() {
        let events = vec![
            error_event("design-a", -10),
            error_event("design-b", -20),
            error_event("design-c", -30),
        ];
        let rules = [AlertRule {
            dedup_window_secs: Some(300),
            ..threshold_rule("high-errors", AlertMetric::ErrorCount, 2)
        }];

        let previous = vec![AlertResult {
            rule_name: "high-errors".to_string(),
            fired: true,
            severity: EventSeverity::Error,
            message: "high-errors triggered".to_string(),
            current_value: 3.0,
            threshold: 2.0,
            evaluated_at: ts(-60), // fired 60s ago, within 300s dedup window
            dedup_key: "high-errors".to_string(),
        }];

        let results = evaluate_rules(&rules, &events, now(), &previous);
        assert_eq!(results.len(), 1);
        assert!(!results[0].fired); // deduplicated
    }

    #[test]
    fn dedup_allows_firing_after_window_expires() {
        let events = vec![
            error_event("design-a", -10),
            error_event("design-b", -20),
            error_event("design-c", -30),
        ];
        let rules = [AlertRule {
            dedup_window_secs: Some(300),
            ..threshold_rule("high-errors", AlertMetric::ErrorCount, 2)
        }];

        let previous = vec![AlertResult {
            rule_name: "high-errors".to_string(),
            fired: true,
            severity: EventSeverity::Error,
            message: "high-errors triggered".to_string(),
            current_value: 3.0,
            threshold: 2.0,
            evaluated_at: ts(-600), // fired 600s ago, outside 300s dedup window
            dedup_key: "high-errors".to_string(),
        }];

        let results = evaluate_rules(&rules, &events, now(), &previous);
        assert_eq!(results.len(), 1);
        assert!(results[0].fired); // not deduplicated
    }

    #[test]
    fn disabled_rules_are_skipped() {
        let events = vec![error_event("design-a", -10)];
        let rules = [AlertRule {
            enabled: false,
            ..threshold_rule("disabled-rule", AlertMetric::ErrorCount, 0)
        }];

        let results = evaluate_rules(&rules, &events, now(), &[]);
        assert!(results.is_empty());
    }

    #[test]
    fn category_scope_filters_events() {
        let events = vec![
            error_event("design-a", -10),    // Materialization
            info_event("source.csv", -20),   // Profiling
            plugin_event("pg-dialect", -30), // Plugin
        ];
        let rules = [AlertRule {
            scope: AlertScope::Category { category: EventCategory::Materialization },
            ..threshold_rule("mat-errors", AlertMetric::ErrorCount, 0)
        }];
        let results = evaluate_rules(&rules, &events, now(), &[]);

        assert_eq!(results.len(), 1);
        assert!(results[0].fired);
        assert_eq!(results[0].current_value, 1.0); // only the materialization error
    }

    #[test]
    fn payload_count_metric_filters_by_name() {
        let events = vec![
            error_event("design-a", -10),  // materialization_failed
            error_event("design-b", -20),  // materialization_failed
            info_event("source.csv", -30), // profiling_completed
        ];
        let metric =
            AlertMetric::PayloadCount { payload_name: "materialization_failed".to_string() };
        let rules = [threshold_rule("mat-failures", metric, 1)];
        let results = evaluate_rules(&rules, &events, now(), &[]);

        assert_eq!(results.len(), 1);
        assert!(results[0].fired);
        assert_eq!(results[0].current_value, 2.0);
    }

    #[test]
    fn deterministic_evaluation_produces_identical_output() {
        let events = vec![
            error_event("design-a", -10),
            info_event("source.csv", -20),
            warning_event("design-b", -30),
        ];
        let rules = [
            threshold_rule("errors", AlertMetric::ErrorCount, 0),
            rate_rule("events", AlertMetric::EventCount, 60, 1),
        ];

        let result1 = evaluate_rules(&rules, &events, now(), &[]);
        let result2 = evaluate_rules(&rules, &events, now(), &[]);

        assert_eq!(result1.len(), result2.len());
        for (a, b) in result1.iter().zip(result2.iter()) {
            assert_eq!(a.rule_name, b.rule_name);
            assert_eq!(a.fired, b.fired);
            assert_eq!(a.current_value, b.current_value);
            assert_eq!(a.threshold, b.threshold);
            assert_eq!(a.evaluated_at, b.evaluated_at);
        }
    }

    #[test]
    fn severity_mapping_is_stable() {
        let events = [
            error_event("design-a", -10), // Error severity from MaterializationFailed
            warning_event("design-b", -20), // Warning from DesignValidated with warnings
            info_event("source.csv", -30), // Info from ProfilingCompleted
        ];

        assert_eq!(events[0].severity(), EventSeverity::Error);
        assert_eq!(events[1].severity(), EventSeverity::Warning);
        assert_eq!(events[2].severity(), EventSeverity::Info);
    }

    // ── Health summary tests ───────────────────────────────────────────

    #[test]
    fn health_summary_healthy_for_info_events() {
        let events = vec![info_event("a.csv", -10), info_event("b.csv", -20)];
        let summary = health_for_domain(&HealthDomain::Overall, &events, &[]);

        assert_eq!(summary.status, HealthStatus::Healthy);
        assert_eq!(summary.total_events, 2);
        assert_eq!(summary.error_count, 0);
        assert_eq!(summary.warning_count, 0);
    }

    #[test]
    fn health_summary_degraded_for_warnings() {
        let events = vec![info_event("a.csv", -10), warning_event("design-a", -20)];
        let summary = health_for_domain(&HealthDomain::Overall, &events, &[]);

        assert_eq!(summary.status, HealthStatus::Degraded);
        assert_eq!(summary.warning_count, 1);
        assert_eq!(summary.error_count, 0);
    }

    #[test]
    fn health_summary_unhealthy_for_errors() {
        let events = vec![info_event("a.csv", -10), error_event("design-a", -20)];
        let summary = health_for_domain(&HealthDomain::Overall, &events, &[]);

        assert_eq!(summary.status, HealthStatus::Unhealthy);
        assert_eq!(summary.error_count, 1);
    }

    #[test]
    fn health_summary_unknown_for_no_events() {
        let summary = health_for_domain(&HealthDomain::Overall, &[], &[]);

        assert_eq!(summary.status, HealthStatus::Unknown);
        assert_eq!(summary.total_events, 0);
    }

    #[test]
    fn health_summary_includes_last_event_timestamp() {
        let events = vec![info_event("a.csv", -100), info_event("b.csv", -10)];
        let summary = health_for_domain(&HealthDomain::Overall, &events, &[]);

        assert_eq!(summary.last_event_at, Some(ts(-10)));
    }

    #[test]
    fn health_summary_includes_active_alert_names() {
        let events = vec![error_event("design-a", -10)];
        let alerts = vec![AlertResult {
            rule_name: "high-errors".to_string(),
            fired: true,
            severity: EventSeverity::Error,
            message: "too many errors".to_string(),
            current_value: 1.0,
            threshold: 0.0,
            evaluated_at: now(),
            dedup_key: "high-errors".to_string(),
        }];
        let summary = health_for_domain(&HealthDomain::Overall, &events, &alerts);

        assert_eq!(summary.active_alerts, vec!["high-errors"]);
    }

    #[test]
    fn health_for_domain_filters_by_category() {
        let events = vec![
            info_event("a.csv", -10),     // Profiling
            error_event("design-a", -20), // Materialization
        ];
        let summary =
            health_for_domain(&HealthDomain::Category(EventCategory::Profiling), &events, &[]);

        assert_eq!(summary.status, HealthStatus::Healthy);
        assert_eq!(summary.total_events, 1);
        assert_eq!(summary.error_count, 0);
    }

    #[test]
    fn health_for_domain_filters_by_source() {
        let events =
            vec![info_event("a.csv", -10), info_event("b.csv", -20), error_event("design-a", -30)];
        let summary = health_for_domain(&HealthDomain::Source("a.csv".to_string()), &events, &[]);

        assert_eq!(summary.total_events, 1);
        assert_eq!(summary.status, HealthStatus::Healthy);
    }

    #[test]
    fn sink_failures_visible_in_health_summaries() {
        let sinks = vec![SinkStatus {
            name: "webhook-prod".to_string(),
            errors: 5,
            last_error: Some("connection timeout".to_string()),
        }];
        let summaries = discover_health(&[], &[], &sinks);

        let sink_summary =
            summaries.iter().find(|s| s.domain == HealthDomain::Sink("webhook-prod".to_string()));
        assert!(sink_summary.is_some());
        let sink = sink_summary.unwrap();
        assert_eq!(sink.status, HealthStatus::Unhealthy);
        assert_eq!(sink.error_count, 5);
    }

    #[test]
    fn discover_health_finds_all_domains() {
        let events = vec![
            info_event("customers.csv", -10), // Profiling, Source
            error_event("orders", -20),       // Materialization, Design subject
            plugin_event("pg-dialect", -30),  // Plugin
        ];
        let summaries = discover_health(&events, &[], &[]);

        // Should include: Overall, Category(Profiling), Category(Materialization),
        // Category(Plugin), Source(customers.csv), Plugin(pg-dialect),
        // Workflow(wf-1), plus Design(orders) as Entity
        assert!(summaries.iter().any(|s| s.domain == HealthDomain::Overall));
        assert!(
            summaries.iter().any(|s| s.domain == HealthDomain::Category(EventCategory::Profiling))
        );
        assert!(
            summaries
                .iter()
                .any(|s| s.domain == HealthDomain::Category(EventCategory::Materialization))
        );
        assert!(
            summaries.iter().any(|s| s.domain == HealthDomain::Source("customers.csv".to_string()))
        );
        assert!(
            summaries.iter().any(|s| s.domain == HealthDomain::Plugin("pg-dialect".to_string()))
        );
        assert!(summaries.iter().any(|s| s.domain == HealthDomain::Workflow("wf-1".to_string())));
    }

    #[test]
    fn discover_health_deterministic_for_same_input() {
        let events = vec![
            info_event("a.csv", -10),
            error_event("design-a", -20),
            warning_event("design-b", -30),
            plugin_event("dialect-1", -40),
        ];

        let result1 = discover_health(&events, &[], &[]);
        let result2 = discover_health(&events, &[], &[]);

        assert_eq!(result1.len(), result2.len());
        for (a, b) in result1.iter().zip(result2.iter()) {
            assert_eq!(a.domain, b.domain);
            assert_eq!(a.status, b.status);
            assert_eq!(a.total_events, b.total_events);
            assert_eq!(a.error_count, b.error_count);
        }
    }

    // ── Fixture tests: normal, degraded, and failing systems ───────────

    /// Normal system: all info events, no errors or warnings.
    fn normal_fixture() -> Vec<RazeriEvent> {
        vec![
            info_event("customers.csv", -10),
            info_event("orders.csv", -20),
            plugin_event("pg-dialect", -30),
        ]
    }

    /// Degraded system: some warnings but no errors.
    fn degraded_fixture() -> Vec<RazeriEvent> {
        vec![
            info_event("customers.csv", -10),
            warning_event("orders-design", -20),
            warning_event("products-design", -30),
            plugin_event("pg-dialect", -40),
        ]
    }

    /// Failing system: errors across multiple categories.
    fn failing_fixture() -> Vec<RazeriEvent> {
        vec![
            error_event("orders-design", -10),
            error_event("customers-design", -20),
            warning_event("products-design", -30),
            info_event("source.csv", -40),
        ]
    }

    #[test]
    fn fixture_normal_system_is_healthy() {
        let summary = health_for_domain(&HealthDomain::Overall, &normal_fixture(), &[]);
        assert_eq!(summary.status, HealthStatus::Healthy);
        assert_eq!(summary.error_count, 0);
        assert_eq!(summary.warning_count, 0);
    }

    #[test]
    fn fixture_degraded_system_is_degraded() {
        let summary = health_for_domain(&HealthDomain::Overall, &degraded_fixture(), &[]);
        assert_eq!(summary.status, HealthStatus::Degraded);
        assert_eq!(summary.error_count, 0);
        assert!(summary.warning_count > 0);
    }

    #[test]
    fn fixture_failing_system_is_unhealthy() {
        let summary = health_for_domain(&HealthDomain::Overall, &failing_fixture(), &[]);
        assert_eq!(summary.status, HealthStatus::Unhealthy);
        assert!(summary.error_count > 0);
    }

    #[test]
    fn fixture_failing_system_alerts_fire() {
        let rules = [threshold_rule("mat-errors", AlertMetric::ErrorCount, 1)];
        let results = evaluate_rules(&rules, &failing_fixture(), now(), &[]);

        assert_eq!(results.len(), 1);
        assert!(results[0].fired);
        assert_eq!(results[0].current_value, 2.0);
    }

    #[test]
    fn fixture_normal_system_alerts_do_not_fire() {
        let rules = [threshold_rule("mat-errors", AlertMetric::ErrorCount, 0)];
        let results = evaluate_rules(&rules, &normal_fixture(), now(), &[]);

        assert_eq!(results.len(), 1);
        assert!(!results[0].fired);
    }

    // ── Serialization tests ────────────────────────────────────────────

    #[test]
    fn alert_rule_serializes_deterministically() {
        let rule = threshold_rule("test-rule", AlertMetric::ErrorCount, 5);
        let json1 = serde_json::to_string(&rule).expect("serialize");
        let json2 = serde_json::to_string(&rule).expect("serialize");
        assert_eq!(json1, json2);
    }

    #[test]
    fn alert_result_round_trips_through_json() {
        let result = AlertResult {
            rule_name: "test".to_string(),
            fired: true,
            severity: EventSeverity::Error,
            message: "test alert".to_string(),
            current_value: 5.0,
            threshold: 3.0,
            evaluated_at: now(),
            dedup_key: "test".to_string(),
        };
        let json = serde_json::to_string(&result).expect("serialize");
        let parsed: AlertResult = serde_json::from_str(&json).expect("deserialize");
        assert_eq!(parsed.rule_name, result.rule_name);
        assert_eq!(parsed.fired, result.fired);
        assert_eq!(parsed.current_value, result.current_value);
    }

    #[test]
    fn health_summary_round_trips_through_json() {
        let summary = HealthSummary {
            domain: HealthDomain::Category(EventCategory::Storage),
            status: HealthStatus::Degraded,
            total_events: 10,
            error_count: 0,
            warning_count: 3,
            last_event_at: Some(now()),
            active_alerts: vec!["slow-storage".to_string()],
        };
        let json = serde_json::to_string(&summary).expect("serialize");
        let parsed: HealthSummary = serde_json::from_str(&json).expect("deserialize");
        assert_eq!(parsed.domain, summary.domain);
        assert_eq!(parsed.status, summary.status);
        assert_eq!(parsed.total_events, summary.total_events);
    }

    // ── Snapshot tests ──────────────────────────────────────────────────

    #[test]
    fn snapshot_alert_rule_json_shape() {
        let rule = threshold_rule("high-errors", AlertMetric::ErrorCount, 5);
        let val: serde_json::Value = serde_json::to_value(&rule).unwrap();
        let obj = val.as_object().unwrap();
        assert!(obj.contains_key("name"));
        assert!(obj.contains_key("scope"));
        assert!(obj.contains_key("condition"));
        assert!(obj.contains_key("severity"));
        assert!(obj.contains_key("message"));
        assert!(obj.contains_key("enabled"));

        // Scope is internally tagged
        let scope = obj["scope"].as_object().unwrap();
        assert_eq!(scope["type"], "global");

        // Condition is internally tagged
        let cond = obj["condition"].as_object().unwrap();
        assert_eq!(cond["type"], "threshold");
        assert!(cond.contains_key("metric"));
        assert!(cond.contains_key("op"));
        assert!(cond.contains_key("value"));
    }

    #[test]
    fn snapshot_alert_result_json_shape() {
        let result = AlertResult {
            rule_name: "check".to_string(),
            fired: true,
            severity: EventSeverity::Warning,
            message: "test".to_string(),
            current_value: 3.0,
            threshold: 2.0,
            evaluated_at: now(),
            dedup_key: "check".to_string(),
        };
        let val: serde_json::Value = serde_json::to_value(&result).unwrap();
        let obj = val.as_object().unwrap();
        assert!(obj.contains_key("rule_name"));
        assert!(obj.contains_key("fired"));
        assert!(obj.contains_key("severity"));
        assert!(obj.contains_key("message"));
        assert!(obj.contains_key("current_value"));
        assert!(obj.contains_key("threshold"));
        assert!(obj.contains_key("evaluated_at"));
        assert!(obj.contains_key("dedup_key"));
    }

    #[test]
    fn snapshot_health_summary_json_shape() {
        let summary = HealthSummary {
            domain: HealthDomain::Category(EventCategory::Profiling),
            status: HealthStatus::Healthy,
            total_events: 5,
            error_count: 0,
            warning_count: 0,
            last_event_at: Some(now()),
            active_alerts: vec![],
        };
        let val: serde_json::Value = serde_json::to_value(&summary).unwrap();
        let obj = val.as_object().unwrap();

        // Domain is tagged with kind+name
        let domain = obj["domain"].as_object().unwrap();
        assert_eq!(domain["kind"], "category");
        assert_eq!(domain["name"], "profiling");

        assert_eq!(obj["status"], "healthy");
        assert!(obj.contains_key("total_events"));
        assert!(obj.contains_key("error_count"));
        assert!(obj.contains_key("warning_count"));
        assert!(obj.contains_key("last_event_at"));
        assert!(obj.contains_key("active_alerts"));
    }

    #[test]
    fn snapshot_discover_health_is_deterministic() {
        let events =
            vec![info_event("src1", 0), info_event("src2", -10), error_event("design1", -5)];
        let results = evaluate_rules(
            &[threshold_rule("errs", AlertMetric::ErrorCount, 0)],
            &events,
            now(),
            &[],
        );
        let h1 = discover_health(&events, &results, &[]);
        let h2 = discover_health(&events, &results, &[]);
        let j1 = serde_json::to_string(&h1).unwrap();
        let j2 = serde_json::to_string(&h2).unwrap();
        assert_eq!(j1, j2, "discover_health must be deterministic");
    }
}