yah-tower 0.8.20

//! `ScryerFilter` — the subscription filter shape tower hands to scryer.
//!
//! Tower compiles a `TowerRule.predicate` into a `ScryerFilter` and passes it
//! to `scryer.subscribe`. Scryer uses this to pre-filter events server-side so
//! only matching events flow to tower over the subscription stream.
//!
//! The `matches` method implements the same logic in-process so the compiler
//! and test fixtures can verify filter behaviour without a running scryer.

use tower_rules::{FieldFilter, FederationPolicy, FieldOp, HealthSignal, LevelFilter,
    RatePredicate, ScopeFilter, StringFilter,
};

use crate::event::{health_signal_name, Event, EventScope, Level};

/// The filter shape tower hands to `scryer.subscribe`.
///
/// `Fields` captures the indexable constraints that scryer can push down to
/// its SQL indexes. Boolean variants allow compound predicates to be evaluated
/// entirely inside scryer when the substrate supports it; in the current P2
/// stub tower evaluates them in-process via `matches`.
#[derive(Debug, Clone, PartialEq)]
pub enum ScryerFilter {
    /// Indexable field constraints — the fast path scryer evaluates first.
    Fields(FieldSet),
    /// All children must match.
    And(Vec<ScryerFilter>),
    /// At least one child must match.
    Or(Vec<ScryerFilter>),
    /// The single child must NOT match.
    Not(Box<ScryerFilter>),
    /// Matches every event regardless of content.
    Any,
    /// Subscribe to scryer's own health signal stream for this specific signal.
    /// Opens a parallel subscription, not against the main event store.
    HealthSignal(HealthSignal),
}

/// Indexable field constraints for a single `EventMatch` predicate.
///
/// All constraints are ANDed together; absent constraints are wildcards.
/// The `rate` constraint is noted here but evaluated by the supervisor (F3)
/// which maintains the sliding window — `matches` ignores it.
#[derive(Debug, Clone, PartialEq)]
pub struct FieldSet {
    pub scope: ScopeFilter,
    pub level: Option<LevelFilter>,
    pub target: Option<StringFilter>,
    pub fields: Vec<FieldFilter>,
    /// Rate constraint from the predicate. Stored for the supervisor but
    /// NOT evaluated in `FieldSet::matches` — rate requires a sliding window.
    pub rate: Option<RatePredicate>,
    /// Federation from the rule, applied when scryer fans out to peers.
    pub federation: FederationPolicy,
}

// ── matches ───────────────────────────────────────────────────────────────────

impl ScryerFilter {
    /// Returns `true` if `event` satisfies this filter.
    ///
    /// Rate predicates inside `Fields` are not evaluated here — they require a
    /// sliding window maintained by the supervisor (F3). All other constraints
    /// are checked in full.
    pub fn matches(&self, event: &Event) -> bool {
        match self {
            ScryerFilter::Any => true,
            ScryerFilter::HealthSignal(signal) => {
                event.scope == EventScope::Health
                    && event.target
                        == format!("scryer.health.{}", health_signal_name(*signal))
            }
            ScryerFilter::Fields(fs) => fs.matches(event),
            ScryerFilter::And(children) => children.iter().all(|f| f.matches(event)),
            ScryerFilter::Or(children) => children.iter().any(|f| f.matches(event)),
            ScryerFilter::Not(child) => !child.matches(event),
        }
    }
}

impl FieldSet {
    fn matches(&self, event: &Event) -> bool {
        if !scope_matches(&self.scope, &event.scope) {
            return false;
        }
        if let Some(min) = self.level {
            if event.level < level_filter_to_level(min) {
                return false;
            }
        }
        if let Some(target_filter) = &self.target {
            if !string_filter_matches(target_filter, &event.target) {
                return false;
            }
        }
        for ff in &self.fields {
            if !field_filter_matches(ff, event) {
                return false;
            }
        }
        true
        // rate: not evaluated here — supervisor handles sliding window
    }
}

// ── scope matching ────────────────────────────────────────────────────────────

fn scope_matches(filter: &ScopeFilter, scope: &EventScope) -> bool {
    match filter {
        ScopeFilter::Any => true,
        ScopeFilter::Federated => true, // treated as Any for in-process matching
        ScopeFilter::TaskRun { id: None } => matches!(scope, EventScope::TaskRun(_)),
        ScopeFilter::TaskRun { id: Some(want) } => {
            matches!(scope, EventScope::TaskRun(got) if got == want)
        }
        ScopeFilter::Service { ident: None } => matches!(scope, EventScope::Service(_)),
        ScopeFilter::Service { ident: Some(want) } => {
            matches!(scope, EventScope::Service(got) if got == want)
        }
        ScopeFilter::Forge { id: None } => matches!(scope, EventScope::Forge(_)),
        ScopeFilter::Forge { id: Some(want) } => {
            matches!(scope, EventScope::Forge(got) if got == want)
        }
    }
}

// ── level ─────────────────────────────────────────────────────────────────────

fn level_filter_to_level(f: LevelFilter) -> Level {
    match f {
        LevelFilter::Debug => Level::Debug,
        LevelFilter::Info => Level::Info,
        LevelFilter::Warn => Level::Warn,
        LevelFilter::Error => Level::Error,
    }
}

// ── string filter ─────────────────────────────────────────────────────────────

pub(crate) fn string_filter_matches(filter: &StringFilter, s: &str) -> bool {
    match filter {
        StringFilter::Exact { value } => s == value,
        StringFilter::Glob { pattern } => glob_matches(pattern, s),
        StringFilter::Regex { pattern } => {
            regex::Regex::new(pattern).map(|re| re.is_match(s)).unwrap_or(false)
        }
    }
}

/// Minimal glob matcher supporting `*` (any sequence) and `?` (any char).
/// Tower-1 globs are simple; the full glob spec is a future tower-2 feature.
fn glob_matches(pattern: &str, s: &str) -> bool {
    glob_matches_inner(pattern.as_bytes(), s.as_bytes())
}

fn glob_matches_inner(pat: &[u8], s: &[u8]) -> bool {
    match (pat.split_first(), s.split_first()) {
        (None, None) => true,
        (None, Some(_)) => false,
        (Some((&b'*', rest_pat)), _) => {
            // * matches zero or more chars
            (0..=s.len()).any(|i| glob_matches_inner(rest_pat, &s[i..]))
        }
        (Some((&b'?', rest_pat)), Some((_, rest_s))) => {
            glob_matches_inner(rest_pat, rest_s)
        }
        (Some((pc, rest_pat)), Some((sc, rest_s))) if pc == sc => {
            glob_matches_inner(rest_pat, rest_s)
        }
        _ => false,
    }
}

// ── field filter ──────────────────────────────────────────────────────────────

fn field_filter_matches(ff: &FieldFilter, event: &Event) -> bool {
    let value = resolve_path(&event.fields, &ff.path);
    match &ff.op {
        FieldOp::Exists => value.is_some(),
        FieldOp::Eq { value: want } => value.map(|v| v == want).unwrap_or(false),
        FieldOp::Contains { substring } => value
            .and_then(|v| v.as_str())
            .map(|s| s.contains(substring.as_str()))
            .unwrap_or(false),
        FieldOp::Matches { filter } => value
            .and_then(|v| v.as_str())
            .map(|s| string_filter_matches(filter, s))
            .unwrap_or(false),
    }
}

/// Walk a dot-separated path into an event's fields map.
///
/// Returns a reference into the original `fields` map for the first segment,
/// then walks into nested JSON objects via `serde_json::Value::get`.
fn resolve_path<'a>(
    fields: &'a std::collections::HashMap<String, serde_json::Value>,
    path: &str,
) -> Option<&'a serde_json::Value> {
    let mut parts = path.splitn(2, '.');
    let key = parts.next()?;
    let value = fields.get(key)?;
    match parts.next() {
        None => Some(value),
        Some(rest) => resolve_json_path(value, rest),
    }
}

/// Walk a dot-separated path into a `serde_json::Value`.
fn resolve_json_path<'a>(value: &'a serde_json::Value, path: &str) -> Option<&'a serde_json::Value> {
    let mut parts = path.splitn(2, '.');
    let key = parts.next()?;
    let child = value.get(key)?;
    match parts.next() {
        None => Some(child),
        Some(rest) => resolve_json_path(child, rest),
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn glob_star_matches_segment() {
        assert!(glob_matches("database.*", "database.query_slow"));
        assert!(glob_matches("database.*", "database.connection_failed"));
        assert!(!glob_matches("database.*", "redis.timeout"));
    }

    #[test]
    fn glob_star_matches_empty() {
        assert!(glob_matches("raft*", "raft"));
        assert!(glob_matches("raft*", "raft.term_change"));
    }

    #[test]
    fn glob_question_mark() {
        assert!(glob_matches("ra?t", "raft"));
        assert!(!glob_matches("ra?t", "rat"));
    }

    #[test]
    fn glob_exact_no_wildcards() {
        assert!(glob_matches("raft.quorum_lost", "raft.quorum_lost"));
        assert!(!glob_matches("raft.quorum_lost", "raft.quorum_los"));
    }
}