yah-tower 0.8.20

//! Subscription supervisor: per-rule scryer subscriptions, dedup, rate limiting.
//!
//! For each enabled `TowerRule`, the supervisor opens one subscription to the
//! local scryer and (when the rule's federation policy requests it) one per
//! remote peer in the yubaba mesh. Events flow through the compiled `ScryerFilter`,
//! a dedup ring keyed by `(rule_id, scope_id, seq)`, and an optional sliding-
//! window rate gate before being surfaced as `FiredEvent`s.
//!
//! The supervisor is driven by polling (`poll()`). The P2 runtime wraps this in
//! an async task; the test harness calls `poll()` synchronously from a single
//! thread.
//!
//! Architecture: `.yah/docs/architecture/A052-yah-tower.md`

use std::collections::{HashMap, VecDeque};
use std::sync::Arc;
use std::time::{Duration, Instant};

use tower_rules::{FederationPolicy, Predicate, RatePredicate, RuleId, TowerRule};

use crate::event::{Event, EventScope};
use crate::rules::compiler::{self, CompileError};
use crate::scryer_filter::ScryerFilter;

// ── Public traits ─────────────────────────────────────────────────────────────

/// Non-blocking event stream from a scryer subscription.
///
/// In production this wraps an async stream from the scryer client (R093-F1+F2).
/// In tests a channel-backed implementation provides deterministic event feeds.
/// Subscriptions are closed by dropping the `Box<dyn EventStream>`.
pub trait EventStream: Send {
    /// Return the next available event without blocking.
    /// Returns `None` if no event is currently available.
    fn try_next(&mut self) -> Option<Event>;
}

/// Source of scryer subscriptions. One implementation per machine (local + one per remote peer).
///
/// Tower opens one subscription per rule by calling `subscribe`. The returned
/// stream is owned by the supervisor and dropped when the rule is disabled.
pub trait SubscriptionSource: Send + Sync {
    fn subscribe(&self, filter: ScryerFilter, rule_id: &RuleId) -> Box<dyn EventStream>;
}

/// Known remote peers in the yubaba mesh.
///
/// Injected into the supervisor at boot. On a local-only camp use `NoPeers`.
/// Federated subscriptions are opened ONCE per peer per rule — not once per
/// event — per the arch doc §Subscriptions efficiency story.
pub trait PeerRegistry: Send + Sync {
    fn peers(&self) -> Vec<String>;
    fn source_for(&self, peer: &str) -> Option<Arc<dyn SubscriptionSource>>;
}

/// No-op registry for local-only camps (the common case in tower-1).
pub struct NoPeers;

impl PeerRegistry for NoPeers {
    fn peers(&self) -> Vec<String> {
        vec![]
    }
    fn source_for(&self, _peer: &str) -> Option<Arc<dyn SubscriptionSource>> {
        None
    }
}

// ── Dedup ─────────────────────────────────────────────────────────────────────

/// Key for the dedup ring: identifies a (rule, event-scope, sequence-position) triple.
///
/// Because `seq` is unique within a scope per the scryer substrate contract
/// (R093-F1), and the ring includes `rule_id`, two rules matching the same
/// underlying event have different keys and both fire.
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct DedupKey {
    pub rule_id: RuleId,
    pub scope_id: String,
    pub seq: u64,
}

pub(crate) fn scope_id(scope: &EventScope) -> String {
    match scope {
        EventScope::TaskRun(id) => format!("task:{id}"),
        EventScope::Service(ident) => format!("service:{}", ident.0),
        EventScope::Forge(id) => format!("forge:{}", id.0),
        EventScope::Health => "health".to_string(),
    }
}

/// Bounded circular buffer of recently-fired `DedupKey`s with their fire timestamp.
///
/// When capacity is reached the oldest entry is evicted. The debounce window
/// allows expired entries to be matched again after the window elapses.
struct DedupRing {
    capacity: usize,
    entries: VecDeque<(DedupKey, Instant)>,
}

impl DedupRing {
    fn new(capacity: usize) -> Self {
        Self { capacity, entries: VecDeque::with_capacity(capacity) }
    }

    /// Returns `true` if `key` fired within `debounce`, or at all if debounce is `None`.
    fn contains(&self, key: &DedupKey, debounce: Option<Duration>) -> bool {
        self.entries.iter().any(|(k, fired_at)| {
            k == key && debounce.map_or(true, |d| fired_at.elapsed() < d)
        })
    }

    fn insert(&mut self, key: DedupKey) {
        if self.entries.len() >= self.capacity {
            self.entries.pop_front();
        }
        self.entries.push_back((key, Instant::now()));
    }
}

// ── Rate window ───────────────────────────────────────────────────────────────

/// Sliding-window counter for `EventMatch` predicates with a `rate` constraint.
///
/// Records match timestamps; `record()` returns `true` only after `min_count`
/// matches have accumulated within the `window_ms` window.
struct RateWindow {
    window_ms: u64,
    min_count: u32,
    timestamps: VecDeque<Instant>,
}

impl RateWindow {
    fn new(min_count: u32, window_ms: u64) -> Self {
        Self { window_ms, min_count, timestamps: VecDeque::new() }
    }

    /// Record a matching event. Returns `true` once the rate threshold is met.
    fn record(&mut self) -> bool {
        let now = Instant::now();
        let window = Duration::from_millis(self.window_ms);
        self.timestamps.retain(|t| now.duration_since(*t) < window);
        self.timestamps.push_back(now);
        self.timestamps.len() >= self.min_count as usize
    }
}

// ── Rule entry ────────────────────────────────────────────────────────────────

struct RuleEntry {
    rule: TowerRule,
    filter: ScryerFilter,
    local_stream: Box<dyn EventStream>,
    peer_streams: HashMap<String, Box<dyn EventStream>>,
    rate_window: Option<RateWindow>,
}

// ── Public output types ───────────────────────────────────────────────────────

/// An event that has passed filter, dedup, and rate checks for a rule.
///
/// Handed to the dispatch engine (F4) which captures context, renders the
/// trigger, and records the audit trail.
#[derive(Debug)]
pub struct FiredEvent {
    /// The rule whose predicate matched.
    pub rule_id: RuleId,
    /// Snapshot of the full rule at fire time, so the dispatch engine (F4) can
    /// render the trigger without needing access to the supervisor's rule map.
    pub rule: TowerRule,
    /// The matching event.
    pub event: Event,
    /// `None` = from the local scryer; `Some(peer)` = federated from a remote peer.
    pub peer: Option<String>,
}

/// Health event emitted by the supervisor itself.
///
/// Surfaced as a `tower.subscription.*` scryer event so tower can dogfood its
/// own observability substrate per arch doc §Dispatch step 3.
#[derive(Debug)]
pub struct SubscriptionHealth {
    pub rule_id: RuleId,
    pub kind: SubscriptionHealthKind,
}

#[derive(Debug)]
pub enum SubscriptionHealthKind {
    /// A new subscription was opened for this rule.
    Opened,
    /// A rule's subscription was closed (rule disabled or updated).
    Closed,
    /// Events were dropped due to backpressure (local stream channel full).
    Lag { dropped: u64 },
}

// ── Supervisor ────────────────────────────────────────────────────────────────

/// Per-rule subscription supervisor.
///
/// Manages one scryer subscription per enabled rule plus federation fan-out.
/// Drives dedup and rate filtering before surfacing `FiredEvent`s to the
/// dispatch engine (F4).
///
/// The supervisor is driven by polling (`poll()`). The P2 async runtime calls
/// `poll()` in a tokio task; tests call it synchronously.
pub struct Supervisor {
    local_source: Arc<dyn SubscriptionSource>,
    peer_registry: Arc<dyn PeerRegistry>,
    rules: HashMap<RuleId, RuleEntry>,
    dedup: DedupRing,
    /// Max events read from a single rule's local stream per `poll()` call.
    /// Excess events are drained-and-dropped; a `Lag` health event is emitted.
    backpressure_limit: usize,
}

impl Supervisor {
    pub fn new(
        local_source: Arc<dyn SubscriptionSource>,
        peer_registry: Arc<dyn PeerRegistry>,
        dedup_capacity: usize,
        backpressure_limit: usize,
    ) -> Self {
        Self {
            local_source,
            peer_registry,
            rules: HashMap::new(),
            dedup: DedupRing::new(dedup_capacity),
            backpressure_limit,
        }
    }

    /// Load a rule and open its subscription(s).
    ///
    /// Returns `Ok(true)` if the rule was loaded (enabled), `Ok(false)` if
    /// skipped (disabled), or `Err(CompileError)` if the predicate is invalid.
    pub fn load_rule(&mut self, rule: TowerRule) -> Result<bool, CompileError> {
        if !rule.enabled {
            return Ok(false);
        }
        let filter = compiler::compile(&rule)?;
        let local_stream = self.local_source.subscribe(filter.clone(), &rule.id);
        let peer_streams = open_peer_streams(&rule, &filter, &*self.peer_registry);
        let rate_window = make_rate_window(&rule.predicate);
        self.rules.insert(
            rule.id.clone(),
            RuleEntry { rule, filter, local_stream, peer_streams, rate_window },
        );
        Ok(true)
    }

    /// Disable a rule and close its subscription(s).
    ///
    /// Subscriptions are closed by dropping the `RuleEntry` (which drops the
    /// `Box<dyn EventStream>` handles). Returns `true` if the rule was found.
    pub fn disable_rule(&mut self, id: &RuleId) -> bool {
        self.rules.remove(id).is_some()
    }

    /// Hot-reload a rule: close the existing subscription and open a new one.
    ///
    /// Called by the config-file watcher (F6) when a rule file is edited on
    /// disk. If the updated rule is disabled, the old subscription is closed
    /// without opening a replacement.
    pub fn update_rule(&mut self, rule: TowerRule) -> Result<(), CompileError> {
        let id = rule.id.clone();
        self.disable_rule(&id);
        self.load_rule(rule)?;
        Ok(())
    }

    /// Poll all active subscriptions and surface `FiredEvent`s.
    ///
    /// Reads up to `backpressure_limit` events per rule stream, applies the
    /// compiled filter, dedup ring, and rate window. Returns
    /// `(fired_events, health_events)` — both may be empty.
    pub fn poll(&mut self) -> (Vec<FiredEvent>, Vec<SubscriptionHealth>) {
        // Phase 1: drain streams into filter-passing candidates.
        // We borrow `self.rules` mutably here (for stream + rate_window),
        // deferring `self.dedup` mutations to phase 2 to avoid simultaneous
        // mutable borrows.
        let mut candidates: Vec<(RuleId, Event, Option<String>, Option<Duration>)> = Vec::new();
        let mut health: Vec<SubscriptionHealth> = Vec::new();

        for (rule_id, entry) in self.rules.iter_mut() {
            let debounce = entry.rule.debounce_ms.map(Duration::from_millis);
            let mut read_count = 0usize;
            let mut dropped = 0u64;

            // Local stream — bounded by backpressure_limit
            loop {
                if read_count >= self.backpressure_limit {
                    while entry.local_stream.try_next().is_some() {
                        dropped += 1;
                    }
                    break;
                }
                match entry.local_stream.try_next() {
                    None => break,
                    Some(event) => {
                        read_count += 1;
                        if entry.filter.matches(&event) {
                            candidates.push((rule_id.clone(), event, None, debounce));
                        }
                    }
                }
            }

            if dropped > 0 {
                health.push(SubscriptionHealth {
                    rule_id: rule_id.clone(),
                    kind: SubscriptionHealthKind::Lag { dropped },
                });
            }

            // Peer streams (federated) — no backpressure limit applied to peers
            for (peer, stream) in entry.peer_streams.iter_mut() {
                while let Some(event) = stream.try_next() {
                    if entry.filter.matches(&event) {
                        candidates.push((
                            rule_id.clone(),
                            event,
                            Some(peer.clone()),
                            debounce,
                        ));
                    }
                }
            }
        }

        // Phase 2: dedup + rate filtering.
        // self.rules borrow from phase 1 has ended; we can now borrow self.dedup
        // and self.rules independently (sequentially, not simultaneously).
        let mut fired = Vec::new();
        for (rule_id, event, peer, debounce) in candidates {
            let key = DedupKey {
                rule_id: rule_id.clone(),
                scope_id: scope_id(&event.scope),
                seq: event.seq,
            };

            if self.dedup.contains(&key, debounce) {
                continue;
            }

            let rate_ok = self
                .rules
                .get_mut(&rule_id)
                .and_then(|e| e.rate_window.as_mut())
                .map_or(true, |rw| rw.record());

            if rate_ok {
                self.dedup.insert(key);
                let rule = self.rules[&rule_id].rule.clone();
                fired.push(FiredEvent { rule_id, rule, event, peer });
            }
        }

        (fired, health)
    }

    /// Number of currently active (enabled + subscribed) rules.
    pub fn active_count(&self) -> usize {
        self.rules.len()
    }

    /// Whether a rule is currently active.
    pub fn is_active(&self, id: &RuleId) -> bool {
        self.rules.contains_key(id)
    }
}

// ── Helpers ───────────────────────────────────────────────────────────────────

fn open_peer_streams(
    rule: &TowerRule,
    filter: &ScryerFilter,
    peers: &dyn PeerRegistry,
) -> HashMap<String, Box<dyn EventStream>> {
    let peer_names: Vec<String> = match &rule.federation {
        FederationPolicy::LocalOnly => return HashMap::new(),
        FederationPolicy::MeshWide => peers.peers(),
        FederationPolicy::MirrorSet { mirrors } => mirrors.clone(),
    };

    peer_names
        .into_iter()
        .filter_map(|peer| {
            peers
                .source_for(&peer)
                .map(|src| (peer.clone(), src.subscribe(filter.clone(), &rule.id)))
        })
        .collect()
}

/// Extract a rate window from the top-level predicate if it is an `EventMatch`
/// with a rate constraint. Tower-1 evaluates rate at the rule level for simple
/// predicates; compound predicates with per-leaf rate windows are tower-2.
fn make_rate_window(predicate: &Predicate) -> Option<RateWindow> {
    match predicate {
        Predicate::EventMatch { rate: Some(RatePredicate { min_count, window_ms }), .. } => {
            Some(RateWindow::new(*min_count, *window_ms))
        }
        _ => None,
    }
}