laser_wire/filter/
expr.rs

1use crate::codes::FILTER_OP_VERSION;
2use crate::error::InvalidError;
3use crate::filter::coerce::{ExactDecimal, TimestampFormat};
4use crate::filter::path::FieldPath;
5use crate::filter::text::{MAX_REGEX_PREDICATES, TextMatch, TextPredicate};
6use crate::limits::{
7    MAX_FILTER_BYTES, MAX_FILTER_DEPTH, MAX_FILTER_LIST_ITEMS, MAX_FILTER_NODES,
8    MAX_FILTER_PATH_BYTES, MAX_FILTER_STRING_BYTES,
9};
10use crate::query::{CmpOp, Predicate};
11use crate::schema::{Digest32, TypedValue};
12use crate::validate::Validate;
13use serde::{Deserialize, Serialize};
14use sha2::{Digest as _, Sha256};
15
16/// Version of the evaluation rules a filter was written against. Part of the
17/// digest, so a rule change can never silently reinterpret a saved revision.
18pub const FILTER_EVALUATOR_VERSION: u32 = 1;
19
20const DIGEST_DOMAIN: &[u8] = b"agdx.consumer-filter.v1\0";
21
22/// A declarative predicate over one record: payload fields, typed user
23/// headers, presence tests, and boolean composition. It never runs user code.
24///
25/// Comparison truth is three-valued. A missing field or a type mismatch is
26/// unknown, `not` keeps unknown, and an unknown root selects nothing. A decode
27/// fault is separate from truth and follows the filter's [`FaultPolicy`].
28#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
29#[serde(rename_all = "snake_case")]
30pub enum FilterExpr {
31    All(Vec<FilterExpr>),
32    Any(Vec<FilterExpr>),
33    Not(Box<FilterExpr>),
34    /// A payload comparison. The field is a [`FieldPath`] text form.
35    Pred(Predicate),
36    /// A payload comparison after an explicit value coercion.
37    PredAs(CoercedPredicate),
38    /// The payload path exists. An explicit `null` value is present.
39    Present(FieldPath),
40    /// The payload path does not exist.
41    Absent(FieldPath),
42    /// A typed user-header comparison. The key is matched exactly.
43    Header(HeaderPredicate),
44    /// A text match on a payload field: equals, prefix, suffix, contains,
45    /// glob, or regex, optionally case-insensitive.
46    Text(TextPredicate),
47    /// A text match on one user header, keyed exactly.
48    HeaderText(TextPredicate),
49}
50
51/// A payload comparison whose field and literal are both coerced before the
52/// compare, so `"2026-09-21T18:04:12Z"` orders as an instant and `"12.50"` as a
53/// number. The coercion is part of the expression and its digest.
54#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
55pub struct CoercedPredicate {
56    pub pred: Predicate,
57    pub coerce: Coerce,
58}
59
60/// An explicit value coercion. A value that does not parse under it compares
61/// as unknown.
62#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
63#[serde(tag = "kind", rename_all = "snake_case")]
64pub enum Coerce {
65    /// Compare as an instant in microseconds.
66    Timestamp { format: TimestampFormat },
67    /// Compare as an exact decimal. Accepts numbers and decimal strings.
68    Number,
69}
70
71/// A comparison against one user header, keyed exactly (`agdx.ct` is one key).
72#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
73pub struct HeaderPredicate {
74    pub key: String,
75    pub op: CmpOp,
76    pub value: TypedValue,
77}
78
79/// How the record payload is decoded before payload predicates run. Chosen
80/// explicitly by the filter: a content-type header never overrides it. A
81/// record whose `agdx.ct` header names another codec is never decoded, it
82/// follows the filter's `foreign_policy` instead.
83#[derive(
84    Clone,
85    Copy,
86    Debug,
87    PartialEq,
88    Eq,
89    Serialize,
90    Deserialize,
91    strum::Display,
92    strum::EnumString,
93    strum::VariantArray,
94)]
95#[serde(rename_all = "snake_case")]
96#[strum(serialize_all = "snake_case")]
97pub enum FilterCodec {
98    /// Decode the payload as JSON.
99    Json,
100    /// Decode one CBOR data item.
101    Cbor,
102    /// Decode a raw Avro datum under its registered writer schema.
103    Avro,
104    /// Decode a Protobuf message under its registered descriptor.
105    Protobuf,
106    /// Never decode the payload. Only header predicates are allowed, so the
107    /// payload may be in any format.
108    HeadersOnly,
109    /// A codec this build does not know, for example one a newer catalog
110    /// saved. It decodes so that listings stay readable, but it never
111    /// validates or compiles.
112    #[serde(other)]
113    Unknown,
114}
115
116/// What happens to a record the filter cannot judge on its own terms.
117///
118/// `foreign_policy` covers a record in another format: its `agdx.ct` header
119/// names another codec, or its writer schema is missing, not listed by the
120/// filter, or of another schema family. `mismatch_policy` covers a record
121/// where a predicate's value exists but has a type the predicate cannot
122/// compare, so the filter could not decide. A mixed log keeps both at
123/// `reject`. A consumer that wants to see edge cases sets `pass`.
124#[derive(
125    Clone,
126    Copy,
127    Debug,
128    Default,
129    PartialEq,
130    Eq,
131    Serialize,
132    Deserialize,
133    strum::Display,
134    strum::EnumString,
135    strum::VariantArray,
136)]
137#[serde(rename_all = "snake_case")]
138#[strum(serialize_all = "snake_case")]
139pub enum RecordPolicy {
140    /// Skip the record. It is not selected and never a fault.
141    #[default]
142    Reject,
143    /// Deliver the record marked unevaluated, so the consumer decides.
144    Pass,
145}
146
147impl RecordPolicy {
148    pub const fn is_reject(&self) -> bool {
149        matches!(self, RecordPolicy::Reject)
150    }
151}
152
153/// What happens to a record whose payload cannot be decoded (malformed, over
154/// the size or depth limit). A valid record missing a field is not a fault.
155#[derive(
156    Clone,
157    Copy,
158    Debug,
159    Default,
160    PartialEq,
161    Eq,
162    Serialize,
163    Deserialize,
164    strum::Display,
165    strum::EnumString,
166    strum::VariantArray,
167)]
168#[serde(rename_all = "snake_case")]
169#[strum(serialize_all = "snake_case")]
170pub enum FaultPolicy {
171    /// Stop the page before the record. It stays unacknowledged until the
172    /// reader changes policy or the record ages out.
173    #[default]
174    Stop,
175    /// Deliver the record marked unevaluated.
176    Pass,
177    /// Skip the record, counted as a fault.
178    Drop,
179}
180
181/// One consumer filter: the expression plus everything that changes what it
182/// selects. Its [`digest`](Self::digest) identifies these semantics.
183#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
184pub struct ConsumerFilter {
185    pub v: u32,
186    pub evaluator_version: u32,
187    pub expr: FilterExpr,
188    pub codec: FilterCodec,
189    #[serde(default)]
190    pub fault_policy: FaultPolicy,
191    #[serde(default, skip_serializing_if = "RecordPolicy::is_reject")]
192    pub foreign_policy: RecordPolicy,
193    #[serde(default, skip_serializing_if = "RecordPolicy::is_reject")]
194    pub mismatch_policy: RecordPolicy,
195    /// Immutable writer schema IDs permitted for Avro or Protobuf payloads.
196    /// A record selects one of these with its agdx.sid header.
197    #[serde(default, skip_serializing_if = "Vec::is_empty")]
198    pub schema_refs: Vec<u32>,
199}
200
201impl FilterExpr {
202    pub fn all(children: impl IntoIterator<Item = FilterExpr>) -> Self {
203        Self::All(children.into_iter().collect())
204    }
205
206    pub fn any(children: impl IntoIterator<Item = FilterExpr>) -> Self {
207        Self::Any(children.into_iter().collect())
208    }
209
210    pub fn negate(child: FilterExpr) -> Self {
211        Self::Not(Box::new(child))
212    }
213
214    pub fn pred(field: impl Into<String>, op: CmpOp, value: impl Into<TypedValue>) -> Self {
215        Self::Pred(Predicate {
216            field: field.into(),
217            op,
218            value: value.into(),
219        })
220    }
221
222    pub fn pred_as(
223        field: impl Into<String>,
224        op: CmpOp,
225        value: impl Into<TypedValue>,
226        coerce: Coerce,
227    ) -> Self {
228        Self::PredAs(CoercedPredicate {
229            pred: Predicate {
230                field: field.into(),
231                op,
232                value: value.into(),
233            },
234            coerce,
235        })
236    }
237
238    /// Presence of a payload path. Returns the parse error of an invalid
239    /// path, so it suits untrusted input where [`Self::present`] would panic.
240    pub fn try_present(path: &str) -> Result<Self, InvalidError> {
241        FieldPath::parse(path).map(Self::Present)
242    }
243
244    /// Presence of a trusted path. Panics if it is invalid.
245    pub fn present(path: &str) -> Self {
246        Self::Present(FieldPath::parse(path).expect("a valid field path"))
247    }
248
249    /// Absence of a payload path. Returns the parse error of an invalid
250    /// path, so it suits untrusted input where [`Self::absent`] would panic.
251    pub fn try_absent(path: &str) -> Result<Self, InvalidError> {
252        FieldPath::parse(path).map(Self::Absent)
253    }
254
255    /// Absence of a trusted path. Panics if it is invalid.
256    pub fn absent(path: &str) -> Self {
257        Self::Absent(FieldPath::parse(path).expect("a valid field path"))
258    }
259
260    pub fn header(key: impl Into<String>, op: CmpOp, value: impl Into<TypedValue>) -> Self {
261        Self::Header(HeaderPredicate {
262            key: key.into(),
263            op,
264            value: value.into(),
265        })
266    }
267
268    pub fn text(field: impl Into<String>, kind: TextMatch, pattern: impl Into<String>) -> Self {
269        Self::Text(TextPredicate {
270            field: field.into(),
271            kind,
272            pattern: pattern.into(),
273            case_insensitive: false,
274        })
275    }
276
277    pub fn header_text(
278        key: impl Into<String>,
279        kind: TextMatch,
280        pattern: impl Into<String>,
281    ) -> Self {
282        Self::HeaderText(TextPredicate {
283            field: key.into(),
284            kind,
285            pattern: pattern.into(),
286            case_insensitive: false,
287        })
288    }
289
290    /// Make a text match case-insensitive. Any other node is returned as is.
291    #[must_use]
292    pub fn case_insensitive(mut self) -> Self {
293        if let Self::Text(predicate) | Self::HeaderText(predicate) = &mut self {
294            predicate.case_insensitive = true;
295        }
296        self
297    }
298
299    /// Whether evaluating this node can require the decoded payload.
300    pub fn reads_payload(&self) -> bool {
301        match self {
302            Self::All(children) | Self::Any(children) => children.iter().any(Self::reads_payload),
303            Self::Not(child) => child.reads_payload(),
304            Self::Pred(_)
305            | Self::PredAs(_)
306            | Self::Present(_)
307            | Self::Absent(_)
308            | Self::Text(_) => true,
309            Self::Header(_) | Self::HeaderText(_) => false,
310        }
311    }
312
313    /// Whether evaluating this node can require the record headers.
314    pub fn reads_headers(&self) -> bool {
315        match self {
316            Self::All(children) | Self::Any(children) => children.iter().any(Self::reads_headers),
317            Self::Not(child) => child.reads_headers(),
318            Self::Header(_) | Self::HeaderText(_) => true,
319            Self::Pred(_)
320            | Self::PredAs(_)
321            | Self::Present(_)
322            | Self::Absent(_)
323            | Self::Text(_) => false,
324        }
325    }
326}
327
328impl ConsumerFilter {
329    /// A JSON-decoding filter with the default `stop` fault policy.
330    pub fn json(expr: FilterExpr) -> Self {
331        Self::new(expr, FilterCodec::Json)
332    }
333
334    /// Decode a self-describing CBOR payload.
335    pub fn cbor(expr: FilterExpr) -> Self {
336        Self::new(expr, FilterCodec::Cbor)
337    }
338
339    /// Decode a raw Avro datum with one of the listed writer schemas.
340    pub fn avro(expr: FilterExpr, schema_refs: impl IntoIterator<Item = u32>) -> Self {
341        Self::schema_codec(expr, FilterCodec::Avro, schema_refs)
342    }
343
344    /// Decode a Protobuf message with one of the listed writer schemas.
345    pub fn protobuf(expr: FilterExpr, schema_refs: impl IntoIterator<Item = u32>) -> Self {
346        Self::schema_codec(expr, FilterCodec::Protobuf, schema_refs)
347    }
348
349    fn schema_codec(
350        expr: FilterExpr,
351        codec: FilterCodec,
352        schema_refs: impl IntoIterator<Item = u32>,
353    ) -> Self {
354        let mut filter = Self::new(expr, codec);
355        filter.schema_refs = schema_refs.into_iter().collect();
356        filter.schema_refs.sort_unstable();
357        filter.schema_refs.dedup();
358        filter
359    }
360
361    /// A header-only filter. The payload is never decoded.
362    pub fn headers_only(expr: FilterExpr) -> Self {
363        Self::new(expr, FilterCodec::HeadersOnly)
364    }
365
366    fn new(expr: FilterExpr, codec: FilterCodec) -> Self {
367        Self {
368            v: FILTER_OP_VERSION,
369            evaluator_version: FILTER_EVALUATOR_VERSION,
370            expr,
371            codec,
372            fault_policy: FaultPolicy::Stop,
373            foreign_policy: RecordPolicy::Reject,
374            mismatch_policy: RecordPolicy::Reject,
375            schema_refs: Vec::new(),
376        }
377    }
378
379    #[must_use]
380    pub fn with_fault_policy(mut self, fault_policy: FaultPolicy) -> Self {
381        self.fault_policy = fault_policy;
382        self
383    }
384
385    #[must_use]
386    pub fn with_foreign_policy(mut self, foreign_policy: RecordPolicy) -> Self {
387        self.foreign_policy = foreign_policy;
388        self
389    }
390
391    #[must_use]
392    pub fn with_mismatch_policy(mut self, mismatch_policy: RecordPolicy) -> Self {
393        self.mismatch_policy = mismatch_policy;
394        self
395    }
396
397    /// SHA-256 over a domain tag and the JSON encoding of the filter. Every
398    /// semantic field is covered. Two filters get the same digest exactly when
399    /// their encodings are equal: logically equivalent but differently written
400    /// expressions keep distinct digests.
401    pub fn digest(&self) -> Digest32 {
402        let encoded = serde_json::to_vec(self).expect("a consumer filter always encodes as JSON");
403        let mut hasher = Sha256::new();
404        hasher.update(DIGEST_DOMAIN);
405        hasher.update(&encoded);
406        Digest32::new(hasher.finalize().into())
407    }
408}
409
410impl Validate for ConsumerFilter {
411    fn validate(&self) -> Result<(), InvalidError> {
412        if self.v != FILTER_OP_VERSION {
413            return Err(InvalidError::new(format!(
414                "filter version {} is not supported, expected {FILTER_OP_VERSION}",
415                self.v
416            )));
417        }
418        if self.evaluator_version != FILTER_EVALUATOR_VERSION {
419            return Err(InvalidError::new(format!(
420                "evaluator version {} is not supported, expected {FILTER_EVALUATOR_VERSION}",
421                self.evaluator_version
422            )));
423        }
424        let schema_codec = matches!(self.codec, FilterCodec::Avro | FilterCodec::Protobuf);
425        if schema_codec == self.schema_refs.is_empty() {
426            return Err(InvalidError::new(
427                "Avro and Protobuf require schema_refs, other codecs require no schema_refs",
428            ));
429        }
430        if self.schema_refs.windows(2).any(|ids| ids[0] >= ids[1]) {
431            return Err(InvalidError::new(
432                "schema_refs must be sorted and contain no duplicates",
433            ));
434        }
435        let encoded_bytes = serde_json::to_vec(self)
436            .map_err(|error| InvalidError::new(format!("filter does not encode: {error}")))?
437            .len();
438        if encoded_bytes > MAX_FILTER_BYTES {
439            return Err(InvalidError::new(format!(
440                "filter is {encoded_bytes}B, exceeds cap {MAX_FILTER_BYTES}B"
441            )));
442        }
443        if self.codec == FilterCodec::Unknown {
444            return Err(InvalidError::new(
445                "the filter codec is not supported by this build",
446            ));
447        }
448        if self.codec == FilterCodec::HeadersOnly && self.expr.reads_payload() {
449            return Err(InvalidError::new(
450                "a headers_only filter cannot use payload predicates",
451            ));
452        }
453        let mut nodes = 0;
454        validate_node(&self.expr, 1, &mut nodes)?;
455        if regex_predicates(&self.expr) > MAX_REGEX_PREDICATES {
456            return Err(InvalidError::new(format!(
457                "a filter holds at most {MAX_REGEX_PREDICATES} compiled text predicates (glob or regex)"
458            )));
459        }
460        Ok(())
461    }
462}
463
464// Every regex compiles to a bounded program, and this bounds their number, so
465// one filter and the per-shard cache of compiled filters stay bounded too.
466fn regex_predicates(expr: &FilterExpr) -> usize {
467    match expr {
468        FilterExpr::All(children) | FilterExpr::Any(children) => {
469            children.iter().map(regex_predicates).sum()
470        }
471        FilterExpr::Not(child) => regex_predicates(child),
472        FilterExpr::Text(predicate) | FilterExpr::HeaderText(predicate) => {
473            usize::from(matches!(predicate.kind, TextMatch::Regex | TextMatch::Glob))
474        }
475        FilterExpr::Pred(_)
476        | FilterExpr::PredAs(_)
477        | FilterExpr::Present(_)
478        | FilterExpr::Absent(_)
479        | FilterExpr::Header(_) => 0,
480    }
481}
482
483fn validate_node(expr: &FilterExpr, depth: usize, nodes: &mut usize) -> Result<(), InvalidError> {
484    if depth > MAX_FILTER_DEPTH {
485        return Err(InvalidError::new(format!(
486            "filter nesting exceeds depth {MAX_FILTER_DEPTH}"
487        )));
488    }
489    count_nodes(nodes, 1)?;
490    match expr {
491        FilterExpr::All(children) | FilterExpr::Any(children) => {
492            if children.is_empty() {
493                return Err(InvalidError::new("`all` and `any` need at least one child"));
494            }
495            children
496                .iter()
497                .try_for_each(|child| validate_node(child, depth + 1, nodes))
498        }
499        FilterExpr::Not(child) => validate_node(child, depth + 1, nodes),
500        FilterExpr::Pred(predicate) => {
501            FieldPath::parse(&predicate.field)?;
502            validate_literal(predicate.op, &predicate.value, LiteralSite::Payload, nodes)
503        }
504        FilterExpr::PredAs(coerced) => {
505            FieldPath::parse(&coerced.pred.field)?;
506            validate_coerced(coerced, nodes)
507        }
508        FilterExpr::Present(_) | FilterExpr::Absent(_) => Ok(()),
509        FilterExpr::Header(header) => {
510            validate_header_key(&header.key)?;
511            validate_literal(header.op, &header.value, LiteralSite::Header, nodes)
512        }
513        FilterExpr::Text(predicate) => {
514            FieldPath::parse(&predicate.field)?;
515            predicate.validate()
516        }
517        FilterExpr::HeaderText(predicate) => {
518            validate_header_key(&predicate.field)?;
519            predicate.validate()
520        }
521    }
522}
523
524fn validate_header_key(key: &str) -> Result<(), InvalidError> {
525    if key.is_empty() || key.len() > MAX_FILTER_PATH_BYTES {
526        return Err(InvalidError::new(format!(
527            "header key must be 1..={MAX_FILTER_PATH_BYTES} bytes"
528        )));
529    }
530    Ok(())
531}
532
533fn count_nodes(nodes: &mut usize, added: usize) -> Result<(), InvalidError> {
534    *nodes += added;
535    if *nodes > MAX_FILTER_NODES {
536        return Err(InvalidError::new(format!(
537            "filter has more than {MAX_FILTER_NODES} nodes"
538        )));
539    }
540    Ok(())
541}
542
543#[derive(Clone, Copy, PartialEq, Eq)]
544enum LiteralSite {
545    Payload,
546    Header,
547}
548
549fn validate_literal(
550    op: CmpOp,
551    value: &TypedValue,
552    site: LiteralSite,
553    nodes: &mut usize,
554) -> Result<(), InvalidError> {
555    match (op, value) {
556        (CmpOp::Eq | CmpOp::Ne, TypedValue::Null) if site == LiteralSite::Payload => Ok(()),
557        (_, TypedValue::Null) => Err(InvalidError::new(match site {
558            LiteralSite::Payload => "a null literal only supports eq and ne",
559            LiteralSite::Header => "a header predicate cannot compare against null",
560        })),
561        (CmpOp::In, TypedValue::List(items)) => {
562            if items.is_empty() || items.len() > MAX_FILTER_LIST_ITEMS {
563                return Err(InvalidError::new(format!(
564                    "an `in` list needs 1..={MAX_FILTER_LIST_ITEMS} items"
565                )));
566            }
567            count_nodes(nodes, items.len())?;
568            items.iter().try_for_each(|item| {
569                if matches!(item, TypedValue::Null | TypedValue::List(_)) {
570                    return Err(InvalidError::new(
571                        "`in` list items must be non-null scalars",
572                    ));
573                }
574                validate_scalar(item)
575            })
576        }
577        (CmpOp::In, _) => Err(InvalidError::new("`in` expects a list literal")),
578        (_, TypedValue::List(_)) => Err(InvalidError::new("a list literal only supports `in`")),
579        (CmpOp::Lt | CmpOp::Lte | CmpOp::Gt | CmpOp::Gte, TypedValue::Boolean(_)) => Err(
580            InvalidError::new("ordered comparisons need a number or string literal"),
581        ),
582        (CmpOp::Prefix, value) if !matches!(value, TypedValue::String(_)) => {
583            Err(InvalidError::new("`prefix` expects a string literal"))
584        }
585        (_, value) => validate_scalar(value),
586    }
587}
588
589fn validate_scalar(value: &TypedValue) -> Result<(), InvalidError> {
590    match value {
591        TypedValue::Boolean(_) | TypedValue::Int(_) | TypedValue::Long(_) => Ok(()),
592        TypedValue::Double(number) if number.is_finite() => Ok(()),
593        TypedValue::Double(_) => Err(InvalidError::new("a double literal must be finite")),
594        TypedValue::String(text) if text.len() <= MAX_FILTER_STRING_BYTES => Ok(()),
595        TypedValue::String(_) => Err(InvalidError::new(format!(
596            "a string literal exceeds {MAX_FILTER_STRING_BYTES}B"
597        ))),
598        _ => Err(InvalidError::new(
599            "supported literals are null, boolean, int, long, double, string, and list",
600        )),
601    }
602}
603
604fn validate_coerced(coerced: &CoercedPredicate, nodes: &mut usize) -> Result<(), InvalidError> {
605    let op = coerced.pred.op;
606    if matches!(op, CmpOp::Contains | CmpOp::Prefix) {
607        return Err(InvalidError::new(
608            "a coerced predicate supports eq, ne, lt, lte, gt, gte, and in",
609        ));
610    }
611    let items: Vec<&TypedValue> = match (&coerced.pred.value, op) {
612        (TypedValue::List(items), CmpOp::In) => {
613            if items.is_empty() || items.len() > MAX_FILTER_LIST_ITEMS {
614                return Err(InvalidError::new(format!(
615                    "an `in` list needs 1..={MAX_FILTER_LIST_ITEMS} items"
616                )));
617            }
618            count_nodes(nodes, items.len())?;
619            items.iter().collect()
620        }
621        (_, CmpOp::In) => return Err(InvalidError::new("`in` expects a list literal")),
622        (TypedValue::List(_), _) => {
623            return Err(InvalidError::new("a list literal only supports `in`"));
624        }
625        (value, _) => vec![value],
626    };
627    items.into_iter().try_for_each(|item| {
628        if coerced_literal_parses(coerced.coerce, item) {
629            Ok(())
630        } else {
631            Err(InvalidError::new(format!(
632                "literal {} does not parse under the {} coercion",
633                item.diagnostic_text(),
634                coerce_label(coerced.coerce)
635            )))
636        }
637    })
638}
639
640pub(crate) fn coerced_literal_parses(coerce: Coerce, value: &TypedValue) -> bool {
641    match coerce {
642        Coerce::Timestamp { format } => match value {
643            TypedValue::String(text) => format.micros_from_text(text).is_some(),
644            TypedValue::Int(number) => format.micros_from_integer(i128::from(*number)).is_some(),
645            TypedValue::Long(number) => format.micros_from_integer(i128::from(*number)).is_some(),
646            _ => false,
647        },
648        Coerce::Number => match value {
649            TypedValue::String(text) => ExactDecimal::parse(text).is_some(),
650            TypedValue::Int(_) | TypedValue::Long(_) => true,
651            _ => false,
652        },
653    }
654}
655
656fn coerce_label(coerce: Coerce) -> String {
657    match coerce {
658        Coerce::Timestamp { format } => format!("timestamp ({format})"),
659        Coerce::Number => "number".to_owned(),
660    }
661}
662
663#[cfg(test)]
664mod tests {
665    use super::*;
666
667    fn safe_mode_filter() -> ConsumerFilter {
668        ConsumerFilter::json(FilterExpr::all([
669            FilterExpr::pred("table", CmpOp::Eq, "satellites"),
670            FilterExpr::pred("changed", CmpOp::Contains, "mode"),
671            FilterExpr::pred("after.mode", CmpOp::Eq, "safe"),
672        ]))
673    }
674
675    #[test]
676    fn given_a_well_formed_filter_when_validated_then_should_pass() {
677        safe_mode_filter().validate().expect("the filter is valid");
678    }
679
680    #[test]
681    fn given_compiled_text_predicates_when_the_filter_budget_is_exceeded_then_should_reject() {
682        for kind in [TextMatch::Glob, TextMatch::Regex] {
683            let children = (0..4).map(|_| FilterExpr::text("name", kind, "safe"));
684            ConsumerFilter::json(FilterExpr::all(children))
685                .validate()
686                .expect("four programs fit");
687            let children = (0..5).map(|_| FilterExpr::text("name", kind, "safe"));
688            assert!(
689                ConsumerFilter::json(FilterExpr::all(children))
690                    .validate()
691                    .is_err()
692            );
693        }
694    }
695
696    #[test]
697    fn given_equal_filters_when_digested_then_should_match_and_differ_on_any_change() {
698        let digest = safe_mode_filter().digest();
699        assert_eq!(digest, safe_mode_filter().digest());
700        assert_eq!(digest.as_bytes().len(), Digest32::BYTES);
701        let dropped = safe_mode_filter().with_fault_policy(FaultPolicy::Drop);
702        assert_ne!(digest, dropped.digest(), "the fault policy is semantic");
703        let reordered = ConsumerFilter::json(FilterExpr::all([
704            FilterExpr::pred("changed", CmpOp::Contains, "mode"),
705            FilterExpr::pred("table", CmpOp::Eq, "satellites"),
706            FilterExpr::pred("after.mode", CmpOp::Eq, "safe"),
707        ]));
708        assert_ne!(digest, reordered.digest(), "expression order is preserved");
709    }
710
711    #[test]
712    fn given_payload_nodes_in_a_headers_only_filter_when_validated_then_should_be_rejected() {
713        let filter = ConsumerFilter::headers_only(FilterExpr::any([
714            FilterExpr::header("table", CmpOp::Eq, "satellites"),
715            FilterExpr::present("kind"),
716        ]));
717        assert!(filter.validate().is_err());
718        let filter =
719            ConsumerFilter::headers_only(FilterExpr::header("table", CmpOp::Eq, "satellites"));
720        filter.validate().expect("header predicates are allowed");
721    }
722
723    #[test]
724    fn given_invalid_literals_when_validated_then_should_be_rejected() {
725        let cases = [
726            FilterExpr::pred("a", CmpOp::Gt, TypedValue::Null),
727            FilterExpr::pred("a", CmpOp::In, "x"),
728            FilterExpr::pred("a", CmpOp::Eq, TypedValue::List(vec!["x".into()])),
729            FilterExpr::pred("a", CmpOp::In, TypedValue::List(Vec::new())),
730            FilterExpr::pred("a", CmpOp::In, TypedValue::List(vec![TypedValue::Null])),
731            FilterExpr::pred("a", CmpOp::Prefix, 5),
732            FilterExpr::pred("a", CmpOp::Lt, true),
733            FilterExpr::pred("a", CmpOp::Eq, f64::NAN),
734            FilterExpr::pred("a", CmpOp::Eq, TypedValue::Date(3)),
735            FilterExpr::pred("a..b", CmpOp::Eq, 1),
736            FilterExpr::header("", CmpOp::Eq, "x"),
737            FilterExpr::header("kind", CmpOp::Eq, TypedValue::Null),
738            FilterExpr::all([]),
739        ];
740        for expr in cases {
741            let filter = ConsumerFilter::json(expr.clone());
742            assert!(filter.validate().is_err(), "{expr:?} must be rejected");
743        }
744    }
745
746    #[test]
747    fn given_coerced_literals_when_validated_then_should_parse_under_their_coercion() {
748        let valid = [
749            FilterExpr::pred_as(
750                "date",
751                CmpOp::Gte,
752                "2026-09-21T00:00:00Z",
753                Coerce::Timestamp {
754                    format: TimestampFormat::Rfc3339,
755                },
756            ),
757            FilterExpr::pred_as(
758                "timestamp_ms",
759                CmpOp::Lt,
760                1_758_470_652_331_i64,
761                Coerce::Timestamp {
762                    format: TimestampFormat::EpochMillis,
763                },
764            ),
765            FilterExpr::pred_as("amount", CmpOp::Gt, "12.50", Coerce::Number),
766            FilterExpr::pred_as(
767                "amount",
768                CmpOp::In,
769                TypedValue::List(vec!["1".into(), 2_i64.into()]),
770                Coerce::Number,
771            ),
772        ];
773        for expr in valid {
774            ConsumerFilter::json(expr.clone())
775                .validate()
776                .unwrap_or_else(|error| panic!("{expr:?} must be valid: {error}"));
777        }
778        let invalid = [
779            FilterExpr::pred_as(
780                "date",
781                CmpOp::Gte,
782                "2026-09-21T00:00:00",
783                Coerce::Timestamp {
784                    format: TimestampFormat::Rfc3339,
785                },
786            ),
787            FilterExpr::pred_as("amount", CmpOp::Gt, "1e5", Coerce::Number),
788            FilterExpr::pred_as("amount", CmpOp::Contains, "1", Coerce::Number),
789            FilterExpr::pred_as("amount", CmpOp::Gt, 1.5, Coerce::Number),
790        ];
791        for expr in invalid {
792            assert!(
793                ConsumerFilter::json(expr.clone()).validate().is_err(),
794                "{expr:?} must be rejected"
795            );
796        }
797    }
798
799    #[test]
800    fn given_limits_when_exceeded_then_should_be_rejected() {
801        let mut expr = FilterExpr::pred("a", CmpOp::Eq, 1);
802        for _ in 0..MAX_FILTER_DEPTH {
803            expr = FilterExpr::negate(expr);
804        }
805        assert!(ConsumerFilter::json(expr).validate().is_err(), "too deep");
806        let wide = FilterExpr::any(
807            (0..MAX_FILTER_NODES).map(|index| FilterExpr::pred(format!("f{index}"), CmpOp::Eq, 1)),
808        );
809        assert!(
810            ConsumerFilter::json(wide).validate().is_err(),
811            "too many nodes"
812        );
813        let mut filter = safe_mode_filter();
814        filter.schema_refs = vec![7];
815        assert!(
816            filter.validate().is_err(),
817            "schema references are not served yet"
818        );
819        let mut filter = safe_mode_filter();
820        filter.evaluator_version = 2;
821        assert!(filter.validate().is_err(), "unknown evaluator version");
822    }
823
824    #[test]
825    fn given_a_filter_when_json_round_tripped_then_should_use_typed_value_tags() {
826        let filter = ConsumerFilter::json(FilterExpr::all([
827            FilterExpr::pred("op", CmpOp::Eq, "d"),
828            FilterExpr::present("fields.mode"),
829        ]));
830        let json = serde_json::to_value(&filter).expect("encodes");
831        assert_eq!(
832            json,
833            serde_json::json!({
834                "v": 1,
835                "evaluator_version": 1,
836                "expr": {"all": [
837                    {"pred": {"field": "op", "op": "eq", "value": {"kind": "string", "value": "d"}}},
838                    {"present": "fields.mode"}
839                ]},
840                "codec": "json",
841                "fault_policy": "stop"
842            })
843        );
844        let back: ConsumerFilter = serde_json::from_value(json).expect("decodes");
845        assert_eq!(back, filter);
846    }
847}
laser_wire/filter/expr.rs

laser_wire/filter/
expr.rs
