tower/
scryer_filter.rs

1//! `ScryerFilter` — the subscription filter shape tower hands to scryer.
2//!
3//! Tower compiles a `TowerRule.predicate` into a `ScryerFilter` and passes it
4//! to `scryer.subscribe`. Scryer uses this to pre-filter events server-side so
5//! only matching events flow to tower over the subscription stream.
6//!
7//! The `matches` method implements the same logic in-process so the compiler
8//! and test fixtures can verify filter behaviour without a running scryer.
9
10use tower_rules::{FieldFilter, FederationPolicy, FieldOp, HealthSignal, LevelFilter,
11    RatePredicate, ScopeFilter, StringFilter,
12};
13
14use crate::event::{health_signal_name, Event, EventScope, Level};
15
16/// The filter shape tower hands to `scryer.subscribe`.
17///
18/// `Fields` captures the indexable constraints that scryer can push down to
19/// its SQL indexes. Boolean variants allow compound predicates to be evaluated
20/// entirely inside scryer when the substrate supports it; in the current P2
21/// stub tower evaluates them in-process via `matches`.
22#[derive(Debug, Clone, PartialEq)]
23pub enum ScryerFilter {
24    /// Indexable field constraints — the fast path scryer evaluates first.
25    Fields(FieldSet),
26    /// All children must match.
27    And(Vec<ScryerFilter>),
28    /// At least one child must match.
29    Or(Vec<ScryerFilter>),
30    /// The single child must NOT match.
31    Not(Box<ScryerFilter>),
32    /// Matches every event regardless of content.
33    Any,
34    /// Subscribe to scryer's own health signal stream for this specific signal.
35    /// Opens a parallel subscription, not against the main event store.
36    HealthSignal(HealthSignal),
37}
38
39/// Indexable field constraints for a single `EventMatch` predicate.
40///
41/// All constraints are ANDed together; absent constraints are wildcards.
42/// The `rate` constraint is noted here but evaluated by the supervisor (F3)
43/// which maintains the sliding window — `matches` ignores it.
44#[derive(Debug, Clone, PartialEq)]
45pub struct FieldSet {
46    pub scope: ScopeFilter,
47    pub level: Option<LevelFilter>,
48    pub target: Option<StringFilter>,
49    pub fields: Vec<FieldFilter>,
50    /// Rate constraint from the predicate. Stored for the supervisor but
51    /// NOT evaluated in `FieldSet::matches` — rate requires a sliding window.
52    pub rate: Option<RatePredicate>,
53    /// Federation from the rule, applied when scryer fans out to peers.
54    pub federation: FederationPolicy,
55}
56
57// ── matches ───────────────────────────────────────────────────────────────────
58
59impl ScryerFilter {
60    /// Returns `true` if `event` satisfies this filter.
61    ///
62    /// Rate predicates inside `Fields` are not evaluated here — they require a
63    /// sliding window maintained by the supervisor (F3). All other constraints
64    /// are checked in full.
65    pub fn matches(&self, event: &Event) -> bool {
66        match self {
67            ScryerFilter::Any => true,
68            ScryerFilter::HealthSignal(signal) => {
69                event.scope == EventScope::Health
70                    && event.target
71                        == format!("scryer.health.{}", health_signal_name(*signal))
72            }
73            ScryerFilter::Fields(fs) => fs.matches(event),
74            ScryerFilter::And(children) => children.iter().all(|f| f.matches(event)),
75            ScryerFilter::Or(children) => children.iter().any(|f| f.matches(event)),
76            ScryerFilter::Not(child) => !child.matches(event),
77        }
78    }
79}
80
81impl FieldSet {
82    fn matches(&self, event: &Event) -> bool {
83        if !scope_matches(&self.scope, &event.scope) {
84            return false;
85        }
86        if let Some(min) = self.level {
87            if event.level < level_filter_to_level(min) {
88                return false;
89            }
90        }
91        if let Some(target_filter) = &self.target {
92            if !string_filter_matches(target_filter, &event.target) {
93                return false;
94            }
95        }
96        for ff in &self.fields {
97            if !field_filter_matches(ff, event) {
98                return false;
99            }
100        }
101        true
102        // rate: not evaluated here — supervisor handles sliding window
103    }
104}
105
106// ── scope matching ────────────────────────────────────────────────────────────
107
108fn scope_matches(filter: &ScopeFilter, scope: &EventScope) -> bool {
109    match filter {
110        ScopeFilter::Any => true,
111        ScopeFilter::Federated => true, // treated as Any for in-process matching
112        ScopeFilter::TaskRun { id: None } => matches!(scope, EventScope::TaskRun(_)),
113        ScopeFilter::TaskRun { id: Some(want) } => {
114            matches!(scope, EventScope::TaskRun(got) if got == want)
115        }
116        ScopeFilter::Service { ident: None } => matches!(scope, EventScope::Service(_)),
117        ScopeFilter::Service { ident: Some(want) } => {
118            matches!(scope, EventScope::Service(got) if got == want)
119        }
120        ScopeFilter::Forge { id: None } => matches!(scope, EventScope::Forge(_)),
121        ScopeFilter::Forge { id: Some(want) } => {
122            matches!(scope, EventScope::Forge(got) if got == want)
123        }
124    }
125}
126
127// ── level ─────────────────────────────────────────────────────────────────────
128
129fn level_filter_to_level(f: LevelFilter) -> Level {
130    match f {
131        LevelFilter::Debug => Level::Debug,
132        LevelFilter::Info => Level::Info,
133        LevelFilter::Warn => Level::Warn,
134        LevelFilter::Error => Level::Error,
135    }
136}
137
138// ── string filter ─────────────────────────────────────────────────────────────
139
140pub(crate) fn string_filter_matches(filter: &StringFilter, s: &str) -> bool {
141    match filter {
142        StringFilter::Exact { value } => s == value,
143        StringFilter::Glob { pattern } => glob_matches(pattern, s),
144        StringFilter::Regex { pattern } => {
145            regex::Regex::new(pattern).map(|re| re.is_match(s)).unwrap_or(false)
146        }
147    }
148}
149
150/// Minimal glob matcher supporting `*` (any sequence) and `?` (any char).
151/// Tower-1 globs are simple; the full glob spec is a future tower-2 feature.
152fn glob_matches(pattern: &str, s: &str) -> bool {
153    glob_matches_inner(pattern.as_bytes(), s.as_bytes())
154}
155
156fn glob_matches_inner(pat: &[u8], s: &[u8]) -> bool {
157    match (pat.split_first(), s.split_first()) {
158        (None, None) => true,
159        (None, Some(_)) => false,
160        (Some((&b'*', rest_pat)), _) => {
161            // * matches zero or more chars
162            (0..=s.len()).any(|i| glob_matches_inner(rest_pat, &s[i..]))
163        }
164        (Some((&b'?', rest_pat)), Some((_, rest_s))) => {
165            glob_matches_inner(rest_pat, rest_s)
166        }
167        (Some((pc, rest_pat)), Some((sc, rest_s))) if pc == sc => {
168            glob_matches_inner(rest_pat, rest_s)
169        }
170        _ => false,
171    }
172}
173
174// ── field filter ──────────────────────────────────────────────────────────────
175
176fn field_filter_matches(ff: &FieldFilter, event: &Event) -> bool {
177    let value = resolve_path(&event.fields, &ff.path);
178    match &ff.op {
179        FieldOp::Exists => value.is_some(),
180        FieldOp::Eq { value: want } => value.map(|v| v == want).unwrap_or(false),
181        FieldOp::Contains { substring } => value
182            .and_then(|v| v.as_str())
183            .map(|s| s.contains(substring.as_str()))
184            .unwrap_or(false),
185        FieldOp::Matches { filter } => value
186            .and_then(|v| v.as_str())
187            .map(|s| string_filter_matches(filter, s))
188            .unwrap_or(false),
189    }
190}
191
192/// Walk a dot-separated path into an event's fields map.
193///
194/// Returns a reference into the original `fields` map for the first segment,
195/// then walks into nested JSON objects via `serde_json::Value::get`.
196fn resolve_path<'a>(
197    fields: &'a std::collections::HashMap<String, serde_json::Value>,
198    path: &str,
199) -> Option<&'a serde_json::Value> {
200    let mut parts = path.splitn(2, '.');
201    let key = parts.next()?;
202    let value = fields.get(key)?;
203    match parts.next() {
204        None => Some(value),
205        Some(rest) => resolve_json_path(value, rest),
206    }
207}
208
209/// Walk a dot-separated path into a `serde_json::Value`.
210fn resolve_json_path<'a>(value: &'a serde_json::Value, path: &str) -> Option<&'a serde_json::Value> {
211    let mut parts = path.splitn(2, '.');
212    let key = parts.next()?;
213    let child = value.get(key)?;
214    match parts.next() {
215        None => Some(child),
216        Some(rest) => resolve_json_path(child, rest),
217    }
218}
219
220#[cfg(test)]
221mod tests {
222    use super::*;
223
224    #[test]
225    fn glob_star_matches_segment() {
226        assert!(glob_matches("database.*", "database.query_slow"));
227        assert!(glob_matches("database.*", "database.connection_failed"));
228        assert!(!glob_matches("database.*", "redis.timeout"));
229    }
230
231    #[test]
232    fn glob_star_matches_empty() {
233        assert!(glob_matches("raft*", "raft"));
234        assert!(glob_matches("raft*", "raft.term_change"));
235    }
236
237    #[test]
238    fn glob_question_mark() {
239        assert!(glob_matches("ra?t", "raft"));
240        assert!(!glob_matches("ra?t", "rat"));
241    }
242
243    #[test]
244    fn glob_exact_no_wildcards() {
245        assert!(glob_matches("raft.quorum_lost", "raft.quorum_lost"));
246        assert!(!glob_matches("raft.quorum_lost", "raft.quorum_los"));
247    }
248}
tower/scryer_filter.rs

tower/
scryer_filter.rs
