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    /// The stream whose writer-schema registry holds `schema_refs`. Absent
200    /// resolves them in the deployment-wide registry and leaves the encoding
201    /// and the digest unchanged.
202    #[serde(default, skip_serializing_if = "Option::is_none")]
203    pub schema_stream: Option<String>,
204}
205
206impl FilterExpr {
207    pub fn all(children: impl IntoIterator<Item = FilterExpr>) -> Self {
208        Self::All(children.into_iter().collect())
209    }
210
211    pub fn any(children: impl IntoIterator<Item = FilterExpr>) -> Self {
212        Self::Any(children.into_iter().collect())
213    }
214
215    pub fn negate(child: FilterExpr) -> Self {
216        Self::Not(Box::new(child))
217    }
218
219    pub fn pred(field: impl Into<String>, op: CmpOp, value: impl Into<TypedValue>) -> Self {
220        Self::Pred(Predicate {
221            field: field.into(),
222            op,
223            value: value.into(),
224        })
225    }
226
227    pub fn pred_as(
228        field: impl Into<String>,
229        op: CmpOp,
230        value: impl Into<TypedValue>,
231        coerce: Coerce,
232    ) -> Self {
233        Self::PredAs(CoercedPredicate {
234            pred: Predicate {
235                field: field.into(),
236                op,
237                value: value.into(),
238            },
239            coerce,
240        })
241    }
242
243    /// Presence of a payload path. Returns the parse error of an invalid
244    /// path, so it suits untrusted input where [`Self::present`] would panic.
245    pub fn try_present(path: &str) -> Result<Self, InvalidError> {
246        FieldPath::parse(path).map(Self::Present)
247    }
248
249    /// Presence of a trusted path. Panics if it is invalid.
250    pub fn present(path: &str) -> Self {
251        Self::Present(FieldPath::parse(path).expect("a valid field path"))
252    }
253
254    /// Absence of a payload path. Returns the parse error of an invalid
255    /// path, so it suits untrusted input where [`Self::absent`] would panic.
256    pub fn try_absent(path: &str) -> Result<Self, InvalidError> {
257        FieldPath::parse(path).map(Self::Absent)
258    }
259
260    /// Absence of a trusted path. Panics if it is invalid.
261    pub fn absent(path: &str) -> Self {
262        Self::Absent(FieldPath::parse(path).expect("a valid field path"))
263    }
264
265    pub fn header(key: impl Into<String>, op: CmpOp, value: impl Into<TypedValue>) -> Self {
266        Self::Header(HeaderPredicate {
267            key: key.into(),
268            op,
269            value: value.into(),
270        })
271    }
272
273    pub fn text(field: impl Into<String>, kind: TextMatch, pattern: impl Into<String>) -> Self {
274        Self::Text(TextPredicate {
275            field: field.into(),
276            kind,
277            pattern: pattern.into(),
278            case_insensitive: false,
279        })
280    }
281
282    pub fn header_text(
283        key: impl Into<String>,
284        kind: TextMatch,
285        pattern: impl Into<String>,
286    ) -> Self {
287        Self::HeaderText(TextPredicate {
288            field: key.into(),
289            kind,
290            pattern: pattern.into(),
291            case_insensitive: false,
292        })
293    }
294
295    /// Make a text match case-insensitive. Any other node is returned as is.
296    #[must_use]
297    pub fn case_insensitive(mut self) -> Self {
298        if let Self::Text(predicate) | Self::HeaderText(predicate) = &mut self {
299            predicate.case_insensitive = true;
300        }
301        self
302    }
303
304    /// Whether evaluating this node can require the decoded payload.
305    pub fn reads_payload(&self) -> bool {
306        match self {
307            Self::All(children) | Self::Any(children) => children.iter().any(Self::reads_payload),
308            Self::Not(child) => child.reads_payload(),
309            Self::Pred(_)
310            | Self::PredAs(_)
311            | Self::Present(_)
312            | Self::Absent(_)
313            | Self::Text(_) => true,
314            Self::Header(_) | Self::HeaderText(_) => false,
315        }
316    }
317
318    /// Whether evaluating this node can require the record headers.
319    pub fn reads_headers(&self) -> bool {
320        match self {
321            Self::All(children) | Self::Any(children) => children.iter().any(Self::reads_headers),
322            Self::Not(child) => child.reads_headers(),
323            Self::Header(_) | Self::HeaderText(_) => true,
324            Self::Pred(_)
325            | Self::PredAs(_)
326            | Self::Present(_)
327            | Self::Absent(_)
328            | Self::Text(_) => false,
329        }
330    }
331}
332
333impl ConsumerFilter {
334    /// A JSON-decoding filter with the default `stop` fault policy.
335    pub fn json(expr: FilterExpr) -> Self {
336        Self::new(expr, FilterCodec::Json)
337    }
338
339    /// Decode a self-describing CBOR payload.
340    pub fn cbor(expr: FilterExpr) -> Self {
341        Self::new(expr, FilterCodec::Cbor)
342    }
343
344    /// Decode a raw Avro datum with one of the listed writer schemas.
345    pub fn avro(expr: FilterExpr, schema_refs: impl IntoIterator<Item = u32>) -> Self {
346        Self::schema_codec(expr, FilterCodec::Avro, schema_refs)
347    }
348
349    /// Decode a Protobuf message with one of the listed writer schemas.
350    pub fn protobuf(expr: FilterExpr, schema_refs: impl IntoIterator<Item = u32>) -> Self {
351        Self::schema_codec(expr, FilterCodec::Protobuf, schema_refs)
352    }
353
354    fn schema_codec(
355        expr: FilterExpr,
356        codec: FilterCodec,
357        schema_refs: impl IntoIterator<Item = u32>,
358    ) -> Self {
359        let mut filter = Self::new(expr, codec);
360        filter.schema_refs = schema_refs.into_iter().collect();
361        filter.schema_refs.sort_unstable();
362        filter.schema_refs.dedup();
363        filter
364    }
365
366    /// A header-only filter. The payload is never decoded.
367    pub fn headers_only(expr: FilterExpr) -> Self {
368        Self::new(expr, FilterCodec::HeadersOnly)
369    }
370
371    fn new(expr: FilterExpr, codec: FilterCodec) -> Self {
372        Self {
373            v: FILTER_OP_VERSION,
374            evaluator_version: FILTER_EVALUATOR_VERSION,
375            expr,
376            codec,
377            fault_policy: FaultPolicy::Stop,
378            foreign_policy: RecordPolicy::Reject,
379            mismatch_policy: RecordPolicy::Reject,
380            schema_refs: Vec::new(),
381            schema_stream: None,
382        }
383    }
384
385    #[must_use]
386    pub fn with_fault_policy(mut self, fault_policy: FaultPolicy) -> Self {
387        self.fault_policy = fault_policy;
388        self
389    }
390
391    #[must_use]
392    pub fn with_foreign_policy(mut self, foreign_policy: RecordPolicy) -> Self {
393        self.foreign_policy = foreign_policy;
394        self
395    }
396
397    #[must_use]
398    pub fn with_mismatch_policy(mut self, mismatch_policy: RecordPolicy) -> Self {
399        self.mismatch_policy = mismatch_policy;
400        self
401    }
402
403    /// Resolve `schema_refs` in the writer-schema registry of `stream`.
404    #[must_use]
405    pub fn with_schema_stream(mut self, stream: impl Into<String>) -> Self {
406        self.schema_stream = Some(stream.into());
407        self
408    }
409
410    /// SHA-256 over a domain tag and the JSON encoding of the filter. Every
411    /// semantic field is covered. Two filters get the same digest exactly when
412    /// their encodings are equal: logically equivalent but differently written
413    /// expressions keep distinct digests.
414    pub fn digest(&self) -> Digest32 {
415        let encoded = serde_json::to_vec(self).expect("a consumer filter always encodes as JSON");
416        let mut hasher = Sha256::new();
417        hasher.update(DIGEST_DOMAIN);
418        hasher.update(&encoded);
419        Digest32::new(hasher.finalize().into())
420    }
421}
422
423impl Validate for ConsumerFilter {
424    fn validate(&self) -> Result<(), InvalidError> {
425        if self.v != FILTER_OP_VERSION {
426            return Err(InvalidError::new(format!(
427                "filter version {} is not supported, expected {FILTER_OP_VERSION}",
428                self.v
429            )));
430        }
431        if self.evaluator_version != FILTER_EVALUATOR_VERSION {
432            return Err(InvalidError::new(format!(
433                "evaluator version {} is not supported, expected {FILTER_EVALUATOR_VERSION}",
434                self.evaluator_version
435            )));
436        }
437        let schema_codec = matches!(self.codec, FilterCodec::Avro | FilterCodec::Protobuf);
438        if schema_codec == self.schema_refs.is_empty() {
439            return Err(InvalidError::new(
440                "Avro and Protobuf require schema_refs, other codecs require no schema_refs",
441            ));
442        }
443        if self.schema_refs.windows(2).any(|ids| ids[0] >= ids[1]) {
444            return Err(InvalidError::new(
445                "schema_refs must be sorted and contain no duplicates",
446            ));
447        }
448        if let Some(stream) = &self.schema_stream {
449            if self.schema_refs.is_empty() {
450                return Err(InvalidError::new(
451                    "schema_stream applies only to a filter with schema_refs",
452                ));
453            }
454            if stream.is_empty() || stream.contains('/') || stream.chars().any(char::is_control) {
455                return Err(InvalidError::new(
456                    "schema_stream must be a non-empty stream name without '/' or control characters",
457                ));
458            }
459        }
460        let encoded_bytes = serde_json::to_vec(self)
461            .map_err(|error| InvalidError::new(format!("filter does not encode: {error}")))?
462            .len();
463        if encoded_bytes > MAX_FILTER_BYTES {
464            return Err(InvalidError::new(format!(
465                "filter is {encoded_bytes}B, exceeds cap {MAX_FILTER_BYTES}B"
466            )));
467        }
468        if self.codec == FilterCodec::Unknown {
469            return Err(InvalidError::new(
470                "the filter codec is not supported by this build",
471            ));
472        }
473        if self.codec == FilterCodec::HeadersOnly && self.expr.reads_payload() {
474            return Err(InvalidError::new(
475                "a headers_only filter cannot use payload predicates",
476            ));
477        }
478        let mut nodes = 0;
479        validate_node(&self.expr, 1, &mut nodes)?;
480        if regex_predicates(&self.expr) > MAX_REGEX_PREDICATES {
481            return Err(InvalidError::new(format!(
482                "a filter holds at most {MAX_REGEX_PREDICATES} compiled text predicates (glob or regex)"
483            )));
484        }
485        Ok(())
486    }
487}
488
489// Every regex compiles to a bounded program, and this bounds their number, so
490// one filter and the per-shard cache of compiled filters stay bounded too.
491fn regex_predicates(expr: &FilterExpr) -> usize {
492    match expr {
493        FilterExpr::All(children) | FilterExpr::Any(children) => {
494            children.iter().map(regex_predicates).sum()
495        }
496        FilterExpr::Not(child) => regex_predicates(child),
497        FilterExpr::Text(predicate) | FilterExpr::HeaderText(predicate) => {
498            usize::from(matches!(predicate.kind, TextMatch::Regex | TextMatch::Glob))
499        }
500        FilterExpr::Pred(_)
501        | FilterExpr::PredAs(_)
502        | FilterExpr::Present(_)
503        | FilterExpr::Absent(_)
504        | FilterExpr::Header(_) => 0,
505    }
506}
507
508fn validate_node(expr: &FilterExpr, depth: usize, nodes: &mut usize) -> Result<(), InvalidError> {
509    if depth > MAX_FILTER_DEPTH {
510        return Err(InvalidError::new(format!(
511            "filter nesting exceeds depth {MAX_FILTER_DEPTH}"
512        )));
513    }
514    count_nodes(nodes, 1)?;
515    match expr {
516        FilterExpr::All(children) | FilterExpr::Any(children) => {
517            if children.is_empty() {
518                return Err(InvalidError::new("`all` and `any` need at least one child"));
519            }
520            children
521                .iter()
522                .try_for_each(|child| validate_node(child, depth + 1, nodes))
523        }
524        FilterExpr::Not(child) => validate_node(child, depth + 1, nodes),
525        FilterExpr::Pred(predicate) => {
526            FieldPath::parse(&predicate.field)?;
527            validate_literal(predicate.op, &predicate.value, LiteralSite::Payload, nodes)
528        }
529        FilterExpr::PredAs(coerced) => {
530            FieldPath::parse(&coerced.pred.field)?;
531            validate_coerced(coerced, nodes)
532        }
533        FilterExpr::Present(_) | FilterExpr::Absent(_) => Ok(()),
534        FilterExpr::Header(header) => {
535            validate_header_key(&header.key)?;
536            validate_literal(header.op, &header.value, LiteralSite::Header, nodes)
537        }
538        FilterExpr::Text(predicate) => {
539            FieldPath::parse(&predicate.field)?;
540            predicate.validate()
541        }
542        FilterExpr::HeaderText(predicate) => {
543            validate_header_key(&predicate.field)?;
544            predicate.validate()
545        }
546    }
547}
548
549fn validate_header_key(key: &str) -> Result<(), InvalidError> {
550    if key.is_empty() || key.len() > MAX_FILTER_PATH_BYTES {
551        return Err(InvalidError::new(format!(
552            "header key must be 1..={MAX_FILTER_PATH_BYTES} bytes"
553        )));
554    }
555    Ok(())
556}
557
558fn count_nodes(nodes: &mut usize, added: usize) -> Result<(), InvalidError> {
559    *nodes += added;
560    if *nodes > MAX_FILTER_NODES {
561        return Err(InvalidError::new(format!(
562            "filter has more than {MAX_FILTER_NODES} nodes"
563        )));
564    }
565    Ok(())
566}
567
568#[derive(Clone, Copy, PartialEq, Eq)]
569enum LiteralSite {
570    Payload,
571    Header,
572}
573
574fn validate_literal(
575    op: CmpOp,
576    value: &TypedValue,
577    site: LiteralSite,
578    nodes: &mut usize,
579) -> Result<(), InvalidError> {
580    match (op, value) {
581        (CmpOp::Eq | CmpOp::Ne, TypedValue::Null) if site == LiteralSite::Payload => Ok(()),
582        (_, TypedValue::Null) => Err(InvalidError::new(match site {
583            LiteralSite::Payload => "a null literal only supports eq and ne",
584            LiteralSite::Header => "a header predicate cannot compare against null",
585        })),
586        (CmpOp::In, TypedValue::List(items)) => {
587            if items.is_empty() || items.len() > MAX_FILTER_LIST_ITEMS {
588                return Err(InvalidError::new(format!(
589                    "an `in` list needs 1..={MAX_FILTER_LIST_ITEMS} items"
590                )));
591            }
592            count_nodes(nodes, items.len())?;
593            items.iter().try_for_each(|item| {
594                if matches!(item, TypedValue::Null | TypedValue::List(_)) {
595                    return Err(InvalidError::new(
596                        "`in` list items must be non-null scalars",
597                    ));
598                }
599                validate_scalar(item)
600            })
601        }
602        (CmpOp::In, _) => Err(InvalidError::new("`in` expects a list literal")),
603        (_, TypedValue::List(_)) => Err(InvalidError::new("a list literal only supports `in`")),
604        (CmpOp::Lt | CmpOp::Lte | CmpOp::Gt | CmpOp::Gte, TypedValue::Boolean(_)) => Err(
605            InvalidError::new("ordered comparisons need a number or string literal"),
606        ),
607        (CmpOp::Prefix, value) if !matches!(value, TypedValue::String(_)) => {
608            Err(InvalidError::new("`prefix` expects a string literal"))
609        }
610        (_, value) => validate_scalar(value),
611    }
612}
613
614fn validate_scalar(value: &TypedValue) -> Result<(), InvalidError> {
615    match value {
616        TypedValue::Boolean(_) | TypedValue::Int(_) | TypedValue::Long(_) => Ok(()),
617        TypedValue::Double(number) if number.is_finite() => Ok(()),
618        TypedValue::Double(_) => Err(InvalidError::new("a double literal must be finite")),
619        TypedValue::String(text) if text.len() <= MAX_FILTER_STRING_BYTES => Ok(()),
620        TypedValue::String(_) => Err(InvalidError::new(format!(
621            "a string literal exceeds {MAX_FILTER_STRING_BYTES}B"
622        ))),
623        _ => Err(InvalidError::new(
624            "supported literals are null, boolean, int, long, double, string, and list",
625        )),
626    }
627}
628
629fn validate_coerced(coerced: &CoercedPredicate, nodes: &mut usize) -> Result<(), InvalidError> {
630    let op = coerced.pred.op;
631    if matches!(op, CmpOp::Contains | CmpOp::Prefix) {
632        return Err(InvalidError::new(
633            "a coerced predicate supports eq, ne, lt, lte, gt, gte, and in",
634        ));
635    }
636    let items: Vec<&TypedValue> = match (&coerced.pred.value, op) {
637        (TypedValue::List(items), CmpOp::In) => {
638            if items.is_empty() || items.len() > MAX_FILTER_LIST_ITEMS {
639                return Err(InvalidError::new(format!(
640                    "an `in` list needs 1..={MAX_FILTER_LIST_ITEMS} items"
641                )));
642            }
643            count_nodes(nodes, items.len())?;
644            items.iter().collect()
645        }
646        (_, CmpOp::In) => return Err(InvalidError::new("`in` expects a list literal")),
647        (TypedValue::List(_), _) => {
648            return Err(InvalidError::new("a list literal only supports `in`"));
649        }
650        (value, _) => vec![value],
651    };
652    items.into_iter().try_for_each(|item| {
653        if coerced_literal_parses(coerced.coerce, item) {
654            Ok(())
655        } else {
656            Err(InvalidError::new(format!(
657                "literal {} does not parse under the {} coercion",
658                item.diagnostic_text(),
659                coerce_label(coerced.coerce)
660            )))
661        }
662    })
663}
664
665pub(crate) fn coerced_literal_parses(coerce: Coerce, value: &TypedValue) -> bool {
666    match coerce {
667        Coerce::Timestamp { format } => match value {
668            TypedValue::String(text) => format.micros_from_text(text).is_some(),
669            TypedValue::Int(number) => format.micros_from_integer(i128::from(*number)).is_some(),
670            TypedValue::Long(number) => format.micros_from_integer(i128::from(*number)).is_some(),
671            _ => false,
672        },
673        Coerce::Number => match value {
674            TypedValue::String(text) => ExactDecimal::parse(text).is_some(),
675            TypedValue::Int(_) | TypedValue::Long(_) => true,
676            _ => false,
677        },
678    }
679}
680
681fn coerce_label(coerce: Coerce) -> String {
682    match coerce {
683        Coerce::Timestamp { format } => format!("timestamp ({format})"),
684        Coerce::Number => "number".to_owned(),
685    }
686}
687
688#[cfg(test)]
689mod tests {
690    use super::*;
691
692    fn safe_mode_filter() -> ConsumerFilter {
693        ConsumerFilter::json(FilterExpr::all([
694            FilterExpr::pred("table", CmpOp::Eq, "satellites"),
695            FilterExpr::pred("changed", CmpOp::Contains, "mode"),
696            FilterExpr::pred("after.mode", CmpOp::Eq, "safe"),
697        ]))
698    }
699
700    #[test]
701    fn given_a_well_formed_filter_when_validated_then_should_pass() {
702        safe_mode_filter().validate().expect("the filter is valid");
703    }
704
705    #[test]
706    fn given_compiled_text_predicates_when_the_filter_budget_is_exceeded_then_should_reject() {
707        for kind in [TextMatch::Glob, TextMatch::Regex] {
708            let children = (0..4).map(|_| FilterExpr::text("name", kind, "safe"));
709            ConsumerFilter::json(FilterExpr::all(children))
710                .validate()
711                .expect("four programs fit");
712            let children = (0..5).map(|_| FilterExpr::text("name", kind, "safe"));
713            assert!(
714                ConsumerFilter::json(FilterExpr::all(children))
715                    .validate()
716                    .is_err()
717            );
718        }
719    }
720
721    #[test]
722    fn given_a_schema_stream_when_encoded_then_should_change_only_a_filter_that_sets_it() {
723        let filter = ConsumerFilter::avro(FilterExpr::pred("mode", CmpOp::Eq, "safe"), [7]);
724        let json = serde_json::to_string(&filter).expect("filter encodes");
725        assert!(!json.contains("schema_stream"), "unset is omitted: {json}");
726        let scoped = filter.clone().with_schema_stream("acme");
727        scoped.validate().expect("a scoped schema filter is valid");
728        assert_ne!(scoped.digest(), filter.digest());
729        let back: ConsumerFilter =
730            serde_json::from_str(&serde_json::to_string(&scoped).expect("filter encodes"))
731                .expect("filter decodes");
732        assert_eq!(back.schema_stream.as_deref(), Some("acme"));
733    }
734
735    #[test]
736    fn given_an_invalid_schema_stream_when_validated_then_should_reject() {
737        let avro = ConsumerFilter::avro(FilterExpr::pred("mode", CmpOp::Eq, "safe"), [7]);
738        assert!(avro.clone().with_schema_stream("").validate().is_err());
739        assert!(avro.with_schema_stream("a/b").validate().is_err());
740        assert!(
741            safe_mode_filter()
742                .with_schema_stream("acme")
743                .validate()
744                .is_err(),
745            "a filter without schema_refs names no registry"
746        );
747    }
748
749    #[test]
750    fn given_equal_filters_when_digested_then_should_match_and_differ_on_any_change() {
751        let digest = safe_mode_filter().digest();
752        assert_eq!(digest, safe_mode_filter().digest());
753        assert_eq!(digest.as_bytes().len(), Digest32::BYTES);
754        let dropped = safe_mode_filter().with_fault_policy(FaultPolicy::Drop);
755        assert_ne!(digest, dropped.digest(), "the fault policy is semantic");
756        let reordered = ConsumerFilter::json(FilterExpr::all([
757            FilterExpr::pred("changed", CmpOp::Contains, "mode"),
758            FilterExpr::pred("table", CmpOp::Eq, "satellites"),
759            FilterExpr::pred("after.mode", CmpOp::Eq, "safe"),
760        ]));
761        assert_ne!(digest, reordered.digest(), "expression order is preserved");
762    }
763
764    #[test]
765    fn given_payload_nodes_in_a_headers_only_filter_when_validated_then_should_be_rejected() {
766        let filter = ConsumerFilter::headers_only(FilterExpr::any([
767            FilterExpr::header("table", CmpOp::Eq, "satellites"),
768            FilterExpr::present("kind"),
769        ]));
770        assert!(filter.validate().is_err());
771        let filter =
772            ConsumerFilter::headers_only(FilterExpr::header("table", CmpOp::Eq, "satellites"));
773        filter.validate().expect("header predicates are allowed");
774    }
775
776    #[test]
777    fn given_invalid_literals_when_validated_then_should_be_rejected() {
778        let cases = [
779            FilterExpr::pred("a", CmpOp::Gt, TypedValue::Null),
780            FilterExpr::pred("a", CmpOp::In, "x"),
781            FilterExpr::pred("a", CmpOp::Eq, TypedValue::List(vec!["x".into()])),
782            FilterExpr::pred("a", CmpOp::In, TypedValue::List(Vec::new())),
783            FilterExpr::pred("a", CmpOp::In, TypedValue::List(vec![TypedValue::Null])),
784            FilterExpr::pred("a", CmpOp::Prefix, 5),
785            FilterExpr::pred("a", CmpOp::Lt, true),
786            FilterExpr::pred("a", CmpOp::Eq, f64::NAN),
787            FilterExpr::pred("a", CmpOp::Eq, TypedValue::Date(3)),
788            FilterExpr::pred("a..b", CmpOp::Eq, 1),
789            FilterExpr::header("", CmpOp::Eq, "x"),
790            FilterExpr::header("kind", CmpOp::Eq, TypedValue::Null),
791            FilterExpr::all([]),
792        ];
793        for expr in cases {
794            let filter = ConsumerFilter::json(expr.clone());
795            assert!(filter.validate().is_err(), "{expr:?} must be rejected");
796        }
797    }
798
799    #[test]
800    fn given_coerced_literals_when_validated_then_should_parse_under_their_coercion() {
801        let valid = [
802            FilterExpr::pred_as(
803                "date",
804                CmpOp::Gte,
805                "2026-09-21T00:00:00Z",
806                Coerce::Timestamp {
807                    format: TimestampFormat::Rfc3339,
808                },
809            ),
810            FilterExpr::pred_as(
811                "timestamp_ms",
812                CmpOp::Lt,
813                1_758_470_652_331_i64,
814                Coerce::Timestamp {
815                    format: TimestampFormat::EpochMillis,
816                },
817            ),
818            FilterExpr::pred_as("cpu", CmpOp::Gt, "12.50", Coerce::Number),
819            FilterExpr::pred_as(
820                "cpu",
821                CmpOp::In,
822                TypedValue::List(vec!["1".into(), 2_i64.into()]),
823                Coerce::Number,
824            ),
825        ];
826        for expr in valid {
827            ConsumerFilter::json(expr.clone())
828                .validate()
829                .unwrap_or_else(|error| panic!("{expr:?} must be valid: {error}"));
830        }
831        let invalid = [
832            FilterExpr::pred_as(
833                "date",
834                CmpOp::Gte,
835                "2026-09-21T00:00:00",
836                Coerce::Timestamp {
837                    format: TimestampFormat::Rfc3339,
838                },
839            ),
840            FilterExpr::pred_as("cpu", CmpOp::Gt, "1e5", Coerce::Number),
841            FilterExpr::pred_as("cpu", CmpOp::Contains, "1", Coerce::Number),
842            FilterExpr::pred_as("cpu", CmpOp::Gt, 1.5, Coerce::Number),
843        ];
844        for expr in invalid {
845            assert!(
846                ConsumerFilter::json(expr.clone()).validate().is_err(),
847                "{expr:?} must be rejected"
848            );
849        }
850    }
851
852    #[test]
853    fn given_limits_when_exceeded_then_should_be_rejected() {
854        let mut expr = FilterExpr::pred("a", CmpOp::Eq, 1);
855        for _ in 0..MAX_FILTER_DEPTH {
856            expr = FilterExpr::negate(expr);
857        }
858        assert!(ConsumerFilter::json(expr).validate().is_err(), "too deep");
859        let wide = FilterExpr::any(
860            (0..MAX_FILTER_NODES).map(|index| FilterExpr::pred(format!("f{index}"), CmpOp::Eq, 1)),
861        );
862        assert!(
863            ConsumerFilter::json(wide).validate().is_err(),
864            "too many nodes"
865        );
866        let mut filter = safe_mode_filter();
867        filter.schema_refs = vec![7];
868        assert!(
869            filter.validate().is_err(),
870            "schema references are not served yet"
871        );
872        let mut filter = safe_mode_filter();
873        filter.evaluator_version = 2;
874        assert!(filter.validate().is_err(), "unknown evaluator version");
875    }
876
877    #[test]
878    fn given_a_filter_when_json_round_tripped_then_should_use_typed_value_tags() {
879        let filter = ConsumerFilter::json(FilterExpr::all([
880            FilterExpr::pred("op", CmpOp::Eq, "d"),
881            FilterExpr::present("fields.mode"),
882        ]));
883        let json = serde_json::to_value(&filter).expect("encodes");
884        assert_eq!(
885            json,
886            serde_json::json!({
887                "v": 1,
888                "evaluator_version": 1,
889                "expr": {"all": [
890                    {"pred": {"field": "op", "op": "eq", "value": {"kind": "string", "value": "d"}}},
891                    {"present": "fields.mode"}
892                ]},
893                "codec": "json",
894                "fault_policy": "stop"
895            })
896        );
897        let back: ConsumerFilter = serde_json::from_value(json).expect("decodes");
898        assert_eq!(back, filter);
899    }
900}
laser_wire/filter/expr.rs

laser_wire/filter/
expr.rs
