yah-tower 0.8.20

//! Supervisor integration tests.
//!
//! Covers:
//!   - N rules each open a subscription (`load_*` family)
//!   - Rule disable closes the subscription (`disable_*`)
//!   - Rule edit restarts the subscription (`update_*`)
//!   - Dedup behaviour (mod `dedup`)
//!   - Rate window behaviour (mod `dedup`)
//!   - Backpressure lag health events
//!   - Federation fan-out to peers

use std::collections::HashMap;
use std::sync::{Arc, Mutex};

use tower::event::{Event, EventScope, Level};
use tower::supervisor::{
    EventStream, NoPeers, PeerRegistry, SubscriptionHealthKind, SubscriptionSource, Supervisor,
};
use tower::scryer_filter::ScryerFilter;
use tower_rules::*;

// ── Test infrastructure ───────────────────────────────────────────────────────

/// Channel-backed event stream for tests.
struct TestStream(std::sync::mpsc::Receiver<Event>);

impl EventStream for TestStream {
    fn try_next(&mut self) -> Option<Event> {
        self.0.try_recv().ok()
    }
}

/// Test subscription source: creates one mpsc channel per rule on `subscribe()`.
/// Tests inject events via `send_to(rule_id, event)`.
struct TestSource {
    senders: Mutex<HashMap<RuleId, std::sync::mpsc::Sender<Event>>>,
    subscribe_count: Mutex<usize>,
}

impl TestSource {
    fn new() -> Arc<Self> {
        Arc::new(Self {
            senders: Mutex::new(HashMap::new()),
            subscribe_count: Mutex::new(0),
        })
    }

    fn send_to(&self, rule_id: &RuleId, event: Event) {
        if let Some(tx) = self.senders.lock().unwrap().get(rule_id) {
            let _ = tx.send(event);
        }
    }

    /// Number of times `subscribe()` has been called (open_count tracks
    /// lifetime subscriptions, not just currently-open ones).
    fn open_count(&self) -> usize {
        *self.subscribe_count.lock().unwrap()
    }
}

impl SubscriptionSource for TestSource {
    fn subscribe(&self, _filter: ScryerFilter, rule_id: &RuleId) -> Box<dyn EventStream> {
        let (tx, rx) = std::sync::mpsc::channel();
        self.senders.lock().unwrap().insert(rule_id.clone(), tx);
        *self.subscribe_count.lock().unwrap() += 1;
        Box::new(TestStream(rx))
    }
}

/// Test peer registry: holds one `TestSource` per named peer.
struct TestPeerRegistry {
    peers: Vec<String>,
    sources: HashMap<String, Arc<TestSource>>,
}

impl TestPeerRegistry {
    fn new(peers: Vec<(&str, Arc<TestSource>)>) -> Arc<Self> {
        let sources =
            peers.iter().map(|(name, src)| (name.to_string(), src.clone())).collect();
        let peer_names = peers.into_iter().map(|(n, _)| n.to_string()).collect();
        Arc::new(Self { peers: peer_names, sources })
    }

    fn send_to_peer(&self, peer: &str, rule_id: &RuleId, event: Event) {
        if let Some(src) = self.sources.get(peer) {
            src.send_to(rule_id, event);
        }
    }
}

impl PeerRegistry for TestPeerRegistry {
    fn peers(&self) -> Vec<String> {
        self.peers.clone()
    }
    fn source_for(&self, peer: &str) -> Option<Arc<dyn SubscriptionSource>> {
        self.sources.get(peer).map(|s| s.clone() as Arc<dyn SubscriptionSource>)
    }
}

// ── Builder helpers ───────────────────────────────────────────────────────────

fn simple_rule(id: &str) -> TowerRule {
    TowerRule {
        schema_version: SchemaVersion::V1,
        id: RuleId(id.into()),
        name: id.into(),
        predicate: Predicate::EventMatch {
            scope: ScopeFilter::Any,
            level: None,
            target: None,
            fields: vec![],
            rate: None,
        },
        trigger: Trigger::Notification {
            channels: vec![NotificationChannel::DesktopBadge],
            severity: Severity::Info,
        },
        debounce_ms: None,
        federation: FederationPolicy::LocalOnly,
        enabled: true,
    }
}

fn service_event(ident: &str, seq: u64) -> Event {
    Event {
        scope: EventScope::Service(MeshIdent(ident.into())),
        level: Level::Info,
        target: "test.event".into(),
        msg: String::new(),
        fields: HashMap::new(),
        seq,
    }
}

fn make_supervisor(source: Arc<TestSource>) -> Supervisor {
    Supervisor::new(source, Arc::new(NoPeers), 256, 100)
}

// ── Lifecycle tests ───────────────────────────────────────────────────────────

#[test]
fn load_rule_opens_subscription() {
    let source = TestSource::new();
    let mut sup = make_supervisor(source.clone());

    assert_eq!(source.open_count(), 0);
    let loaded = sup.load_rule(simple_rule("rule-1")).unwrap();
    assert!(loaded);
    assert_eq!(source.open_count(), 1);
    assert_eq!(sup.active_count(), 1);
    assert!(sup.is_active(&RuleId("rule-1".into())));
}

#[test]
fn load_n_rules_opens_n_subscriptions() {
    let source = TestSource::new();
    let mut sup = make_supervisor(source.clone());

    for i in 0..5 {
        sup.load_rule(simple_rule(&format!("rule-{i}"))).unwrap();
    }
    assert_eq!(source.open_count(), 5);
    assert_eq!(sup.active_count(), 5);
}

#[test]
fn disabled_rule_skipped_no_subscription() {
    let source = TestSource::new();
    let mut sup = make_supervisor(source.clone());

    let mut rule = simple_rule("disabled");
    rule.enabled = false;
    let loaded = sup.load_rule(rule).unwrap();

    assert!(!loaded);
    assert_eq!(source.open_count(), 0);
    assert_eq!(sup.active_count(), 0);
}

#[test]
fn disable_rule_closes_subscription() {
    let source = TestSource::new();
    let mut sup = make_supervisor(source.clone());

    sup.load_rule(simple_rule("rule-1")).unwrap();
    assert_eq!(sup.active_count(), 1);

    let removed = sup.disable_rule(&RuleId("rule-1".into()));
    assert!(removed);
    assert_eq!(sup.active_count(), 0);
    assert!(!sup.is_active(&RuleId("rule-1".into())));
}

#[test]
fn disable_unknown_rule_returns_false() {
    let source = TestSource::new();
    let mut sup = make_supervisor(source);

    assert!(!sup.disable_rule(&RuleId("nonexistent".into())));
}

#[test]
fn update_rule_restarts_subscription() {
    let source = TestSource::new();
    let mut sup = make_supervisor(source.clone());

    sup.load_rule(simple_rule("rule-1")).unwrap();
    assert_eq!(source.open_count(), 1, "first load opens 1 subscription");

    sup.update_rule(simple_rule("rule-1")).unwrap();
    assert_eq!(source.open_count(), 2, "update opens a second subscription");
    assert_eq!(sup.active_count(), 1, "still exactly one active rule");
}

#[test]
fn update_to_disabled_closes_subscription() {
    let source = TestSource::new();
    let mut sup = make_supervisor(source.clone());

    sup.load_rule(simple_rule("rule-1")).unwrap();
    assert_eq!(sup.active_count(), 1);

    let mut disabled = simple_rule("rule-1");
    disabled.enabled = false;
    sup.update_rule(disabled).unwrap();

    assert_eq!(sup.active_count(), 0, "disabling via update should close subscription");
}

#[test]
fn poll_fires_matching_event() {
    let source = TestSource::new();
    let mut sup = make_supervisor(source.clone());

    sup.load_rule(simple_rule("rule-1")).unwrap();
    source.send_to(&RuleId("rule-1".into()), service_event("api.pdx", 1));

    let (fired, health) = sup.poll();
    assert_eq!(fired.len(), 1);
    assert_eq!(fired[0].rule_id, RuleId("rule-1".into()));
    assert_eq!(fired[0].event.seq, 1);
    assert!(fired[0].peer.is_none());
    assert!(health.is_empty());
}

#[test]
fn poll_no_events_returns_empty() {
    let source = TestSource::new();
    let mut sup = make_supervisor(source.clone());

    sup.load_rule(simple_rule("rule-1")).unwrap();
    let (fired, health) = sup.poll();
    assert!(fired.is_empty());
    assert!(health.is_empty());
}

#[test]
fn compile_error_propagates() {
    let source = TestSource::new();
    let mut sup = make_supervisor(source);

    let bad_rule = TowerRule {
        schema_version: SchemaVersion::V1,
        id: RuleId("bad".into()),
        name: "bad".into(),
        predicate: Predicate::Compound {
            op: CompoundOp::Not,
            children: vec![
                Predicate::ScryerHealth { signal: HealthSignal::IngestionLag },
                Predicate::ScryerHealth { signal: HealthSignal::RingOverflow },
            ],
        },
        trigger: Trigger::Notification {
            channels: vec![NotificationChannel::DesktopBadge],
            severity: Severity::Info,
        },
        debounce_ms: None,
        federation: FederationPolicy::LocalOnly,
        enabled: true,
    };
    assert!(sup.load_rule(bad_rule).is_err());
    assert_eq!(sup.active_count(), 0);
}

#[test]
fn backpressure_emits_lag_health_event() {
    let source = TestSource::new();
    let mut sup = Supervisor::new(source.clone(), Arc::new(NoPeers), 256, 2);

    sup.load_rule(simple_rule("rule-1")).unwrap();

    for i in 0..5u64 {
        source.send_to(&RuleId("rule-1".into()), service_event("api.pdx", i));
    }
    let (fired, health) = sup.poll();
    assert_eq!(fired.len(), 2, "only backpressure_limit events should fire");
    assert_eq!(health.len(), 1, "one lag health event expected");
    assert!(matches!(health[0].kind, SubscriptionHealthKind::Lag { dropped: _ }));
}

#[test]
fn federation_mesh_wide_opens_peer_subscriptions() {
    let peer_source = TestSource::new();
    let peers = TestPeerRegistry::new(vec![("peer-1", peer_source.clone())]);
    let local_source = TestSource::new();

    let mut sup = Supervisor::new(
        local_source.clone(),
        peers.clone(),
        256,
        100,
    );

    let rule = TowerRule {
        schema_version: SchemaVersion::V1,
        id: RuleId("fed-rule".into()),
        name: "federated".into(),
        predicate: Predicate::EventMatch {
            scope: ScopeFilter::Any,
            level: None,
            target: None,
            fields: vec![],
            rate: None,
        },
        trigger: Trigger::Notification {
            channels: vec![NotificationChannel::DesktopBadge],
            severity: Severity::Info,
        },
        debounce_ms: None,
        federation: FederationPolicy::MeshWide,
        enabled: true,
    };
    sup.load_rule(rule).unwrap();

    assert_eq!(local_source.open_count(), 1, "one local subscription");
    assert_eq!(peer_source.open_count(), 1, "one subscription per peer");

    peers.send_to_peer("peer-1", &RuleId("fed-rule".into()), service_event("db.remote", 1));
    let (fired, _) = sup.poll();
    assert_eq!(fired.len(), 1);
    assert_eq!(fired[0].peer, Some("peer-1".into()), "federated event tagged with peer name");
}

#[test]
fn federation_mirror_set_opens_named_peers_only() {
    let peer_a = TestSource::new();
    let peer_b = TestSource::new();
    let peers = TestPeerRegistry::new(vec![
        ("mirror-a", peer_a.clone()),
        ("mirror-b", peer_b.clone()),
    ]);
    let local_source = TestSource::new();

    let mut sup = Supervisor::new(local_source.clone(), peers, 256, 100);

    let rule = TowerRule {
        schema_version: SchemaVersion::V1,
        id: RuleId("mirror-set-rule".into()),
        name: "mirror set".into(),
        predicate: Predicate::EventMatch {
            scope: ScopeFilter::Any,
            level: None,
            target: None,
            fields: vec![],
            rate: None,
        },
        trigger: Trigger::Notification {
            channels: vec![NotificationChannel::DesktopBadge],
            severity: Severity::Info,
        },
        debounce_ms: None,
        federation: FederationPolicy::MirrorSet { mirrors: vec!["mirror-a".into()] },
        enabled: true,
    };
    sup.load_rule(rule).unwrap();

    assert_eq!(peer_a.open_count(), 1, "mirror-a subscribed");
    assert_eq!(peer_b.open_count(), 0, "mirror-b not in the mirror set");
}

// ── Dedup tests ───────────────────────────────────────────────────────────────

mod dedup {
    use super::*;

    #[test]
    fn same_event_two_rules_fires_both() {
        let source = TestSource::new();
        let mut sup = make_supervisor(source.clone());

        sup.load_rule(simple_rule("rule-a")).unwrap();
        sup.load_rule(simple_rule("rule-b")).unwrap();

        // Same underlying event (scope + seq) arrives on both rule streams.
        // Dedup key is (rule_id, scope_id, seq), so the two are distinct keys.
        let event = service_event("api.pdx", 42);
        source.send_to(&RuleId("rule-a".into()), event.clone());
        source.send_to(&RuleId("rule-b".into()), event.clone());

        let (fired, _) = sup.poll();
        assert_eq!(fired.len(), 2, "both rules should fire");

        let rule_ids: Vec<_> = fired.iter().map(|f| f.rule_id.0.as_str()).collect();
        assert!(rule_ids.contains(&"rule-a"));
        assert!(rule_ids.contains(&"rule-b"));
    }

    #[test]
    fn same_event_replay_does_not_refire() {
        let source = TestSource::new();
        let mut sup = make_supervisor(source.clone());

        sup.load_rule(simple_rule("rule-1")).unwrap();

        let event = service_event("api.pdx", 99);

        // First arrival
        source.send_to(&RuleId("rule-1".into()), event.clone());
        let (fired, _) = sup.poll();
        assert_eq!(fired.len(), 1, "first occurrence should fire");

        // Replay: exact same event arrives again (e.g. after reconnect)
        source.send_to(&RuleId("rule-1".into()), event.clone());
        let (fired, _) = sup.poll();
        assert_eq!(fired.len(), 0, "replay should be suppressed by dedup ring");
    }

    #[test]
    fn different_seq_fires_again() {
        let source = TestSource::new();
        let mut sup = make_supervisor(source.clone());

        sup.load_rule(simple_rule("rule-1")).unwrap();

        source.send_to(&RuleId("rule-1".into()), service_event("api.pdx", 10));
        source.send_to(&RuleId("rule-1".into()), service_event("api.pdx", 11));

        let (fired, _) = sup.poll();
        assert_eq!(fired.len(), 2, "distinct seq values should both fire");
    }

    #[test]
    fn rate_window_suppresses_below_threshold() {
        let source = TestSource::new();
        let mut sup = make_supervisor(source.clone());

        let rule = TowerRule {
            schema_version: SchemaVersion::V1,
            id: RuleId("rate-rule".into()),
            name: "rate".into(),
            predicate: Predicate::EventMatch {
                scope: ScopeFilter::Any,
                level: None,
                target: None,
                fields: vec![],
                rate: Some(RatePredicate { min_count: 3, window_ms: 5000 }),
            },
            trigger: Trigger::Notification {
                channels: vec![NotificationChannel::DesktopBadge],
                severity: Severity::Warning,
            },
            debounce_ms: None,
            federation: FederationPolicy::LocalOnly,
            enabled: true,
        };
        sup.load_rule(rule).unwrap();

        // Only 2 matching events — below min_count=3
        source.send_to(&RuleId("rate-rule".into()), service_event("api.pdx", 1));
        source.send_to(&RuleId("rate-rule".into()), service_event("api.pdx", 2));

        let (fired, _) = sup.poll();
        assert_eq!(fired.len(), 0, "should not fire before rate threshold is met");
    }

    #[test]
    fn rate_window_fires_at_threshold() {
        let source = TestSource::new();
        let mut sup = make_supervisor(source.clone());

        let rule = TowerRule {
            schema_version: SchemaVersion::V1,
            id: RuleId("rate-rule".into()),
            name: "rate".into(),
            predicate: Predicate::EventMatch {
                scope: ScopeFilter::Any,
                level: None,
                target: None,
                fields: vec![],
                rate: Some(RatePredicate { min_count: 3, window_ms: 5000 }),
            },
            trigger: Trigger::Notification {
                channels: vec![NotificationChannel::DesktopBadge],
                severity: Severity::Warning,
            },
            debounce_ms: None,
            federation: FederationPolicy::LocalOnly,
            enabled: true,
        };
        sup.load_rule(rule).unwrap();

        // Exactly min_count=3 matching events
        for i in 0..3u64 {
            source.send_to(&RuleId("rate-rule".into()), service_event("api.pdx", i));
        }

        let (fired, _) = sup.poll();
        assert_eq!(fired.len(), 1, "should fire exactly once at the rate threshold");
    }
}