Skip to main content

laser_wire/filter/
eval.rs

1use super::codecs::PayloadDecoders;
2use crate::content::ContentType;
3use crate::control::SchemaDef;
4use crate::error::InvalidError;
5use crate::filter::coerce::{ExactDecimal, TimestampFormat};
6use crate::filter::expr::{
7    Coerce, ConsumerFilter, FaultPolicy, FilterCodec, FilterExpr, RecordPolicy,
8};
9use crate::filter::path::{FieldPath, PathSegment};
10use crate::filter::read::{
11    ExplainNode, FaultReason, FilterExplanation, FilterHeader, HeaderScalar, Truth, Verdict,
12};
13use crate::filter::text::TextMatcher;
14use crate::headers::CONTENT_TYPE;
15use crate::limits::MAX_FILTER_PARSE_DEPTH;
16use crate::query::CmpOp;
17use crate::schema::{Digest32, TypedValue};
18use crate::validate::Validate;
19use serde::de::{DeserializeSeed, MapAccess, SeqAccess, Visitor};
20use serde::{Deserialize, Deserializer};
21use serde_json::{Map, Value};
22use std::borrow::Cow;
23use std::cmp::Ordering;
24use std::fmt;
25
26/// Bounds on decoding one payload. Checked before the payload is parsed.
27#[derive(Clone, Copy, Debug, PartialEq, Eq)]
28pub struct DecodeLimits {
29    pub max_payload_bytes: usize,
30    pub max_depth: usize,
31}
32
33impl Default for DecodeLimits {
34    fn default() -> Self {
35        Self {
36            max_payload_bytes: 1024 * 1024,
37            max_depth: 64,
38        }
39    }
40}
41
42/// One record as the evaluator sees it. Headers are read only when the filter
43/// has header predicates, so a caller may pass them lazily decoded. Header
44/// keys and values are borrowed, so a caller does not copy them per record.
45#[derive(Clone, Copy, Debug)]
46pub struct FilterRecord<'a> {
47    pub payload: &'a [u8],
48    pub headers: &'a [HeaderRef<'a>],
49}
50
51/// One user header of a record, borrowed from wherever the caller holds it.
52#[derive(Clone, Copy, Debug, PartialEq)]
53pub struct HeaderRef<'a> {
54    pub key: &'a str,
55    pub value: HeaderValueRef<'a>,
56}
57
58/// A typed user header value, borrowed. Mirrors [`HeaderScalar`].
59#[derive(Clone, Copy, Debug, PartialEq)]
60pub enum HeaderValueRef<'a> {
61    Bool(bool),
62    Int(i64),
63    Uint(u64),
64    Float(f64),
65    String(&'a str),
66    Raw(&'a [u8]),
67}
68
69impl<'a> From<&'a FilterHeader> for HeaderRef<'a> {
70    fn from(header: &'a FilterHeader) -> Self {
71        Self {
72            key: &header.key,
73            value: match &header.value {
74                HeaderScalar::Bool(value) => HeaderValueRef::Bool(*value),
75                HeaderScalar::Int(value) => HeaderValueRef::Int(*value),
76                HeaderScalar::Uint(value) => HeaderValueRef::Uint(*value),
77                HeaderScalar::Float(value) => HeaderValueRef::Float(*value),
78                HeaderScalar::String(value) => HeaderValueRef::String(value),
79                HeaderScalar::Raw(bytes) => HeaderValueRef::Raw(bytes),
80            },
81        }
82    }
83}
84
85/// What a caller decodes from a record's header block before evaluating.
86#[derive(Clone, Copy, Debug, PartialEq, Eq)]
87pub enum HeaderNeed {
88    None,
89    /// Only `agdx.ct`, to keep a record in another codec out of the decoder.
90    ContentType,
91    All,
92}
93
94/// A validated filter compiled for repeated evaluation. Compile once per
95/// digest, evaluate per record.
96#[derive(Clone, Debug)]
97pub struct CompiledFilter {
98    filter: ConsumerFilter,
99    digest: Digest32,
100    root: Node,
101    paths: PathTrie,
102    reads_headers: bool,
103    decoders: PayloadDecoders,
104}
105
106/// The payload paths a filter reads. Decoding keeps only the values on these
107/// paths, so a filter that reads four fields of a wide change record does not
108/// build the rest of the document. Skipped values are still parsed, so a
109/// malformed payload stays a fault.
110// Children are a short vector searched in order. A filter reads a handful of
111// keys per level, so this beats hashing every key of every record.
112#[derive(Clone, Debug, Default)]
113pub(super) struct PathTrie {
114    pub(super) whole: bool,
115    indices: Vec<(u32, PathTrie)>,
116    keys: Vec<(String, PathTrie)>,
117}
118
119#[derive(Clone, Debug)]
120struct Node {
121    kind: NodeKind,
122    reads_payload: bool,
123}
124
125#[derive(Clone, Debug)]
126enum NodeKind {
127    All(Vec<Node>),
128    Any(Vec<Node>),
129    Not(Box<Node>),
130    Compare {
131        path: FieldPath,
132        op: CmpOp,
133        literal: Literal,
134    },
135    Coerced {
136        path: FieldPath,
137        op: CmpOp,
138        coerce: Coerce,
139        literal: CoercedLiteral,
140    },
141    Present(FieldPath),
142    Absent(FieldPath),
143    Header {
144        key: String,
145        op: CmpOp,
146        literal: Literal,
147    },
148    Text {
149        path: FieldPath,
150        matcher: TextMatcher,
151    },
152    HeaderText {
153        key: String,
154        matcher: TextMatcher,
155    },
156}
157
158#[derive(Clone, Debug, PartialEq)]
159enum Literal {
160    Null,
161    Bool(bool),
162    Int(i128),
163    Float(f64),
164    Text(String),
165    List(Vec<Literal>),
166}
167
168#[derive(Clone, Debug, PartialEq)]
169enum CoercedLiteral {
170    Instant(i64),
171    Decimal(ExactDecimal),
172    List(Vec<CoercedLiteral>),
173}
174
175#[derive(Clone, Debug, PartialEq)]
176enum Comparable {
177    Instant(i64),
178    Decimal(ExactDecimal),
179}
180
181/// A value read from the payload or a header, ready to compare.
182enum Scalar<'a> {
183    Null,
184    Bool(bool),
185    Int(i128),
186    Float(f64),
187    Text(&'a str),
188    Composite,
189}
190
191impl CompiledFilter {
192    /// Validate and compile.
193    pub fn compile(filter: &ConsumerFilter) -> Result<Self, InvalidError> {
194        Self::compile_with_schemas(filter, &[])
195    }
196
197    /// Compile against immutable writer schemas loaded by the caller.
198    pub fn compile_with_schemas(
199        filter: &ConsumerFilter,
200        schemas: &[SchemaDef],
201    ) -> Result<Self, InvalidError> {
202        filter.validate()?;
203        let decoders = PayloadDecoders::compile(filter, schemas)?;
204        let mut paths = PathTrie::default();
205        collect_paths(&filter.expr, &mut paths)?;
206        Ok(Self {
207            filter: filter.clone(),
208            digest: filter.digest(),
209            root: compile_node(&filter.expr)?,
210            paths,
211            reads_headers: filter.expr.reads_headers() || !filter.schema_refs.is_empty(),
212            decoders,
213        })
214    }
215
216    pub fn digest(&self) -> &Digest32 {
217        &self.digest
218    }
219
220    pub fn filter(&self) -> &ConsumerFilter {
221        &self.filter
222    }
223
224    pub fn fault_policy(&self) -> FaultPolicy {
225        self.filter.fault_policy
226    }
227
228    /// The policy a record that faulted for `reason` follows. A record in
229    /// another format follows the foreign policy, a type mismatch the mismatch
230    /// policy, and a payload broken in the filter's own codec the fault policy.
231    /// A `reject` record policy never reaches a fault, so it maps to `drop`.
232    pub fn policy_for(&self, reason: FaultReason) -> FaultPolicy {
233        match self.record_policy(reason) {
234            Some(RecordPolicy::Reject) => FaultPolicy::Drop,
235            Some(RecordPolicy::Pass) => FaultPolicy::Pass,
236            None => self.filter.fault_policy,
237        }
238    }
239
240    /// The record policy that covers `reason`, or `None` for a decode fault.
241    pub fn record_policy(&self, reason: FaultReason) -> Option<RecordPolicy> {
242        if reason.is_foreign() {
243            Some(self.filter.foreign_policy)
244        } else if reason == FaultReason::TypeMismatch {
245            Some(self.filter.mismatch_policy)
246        } else {
247            None
248        }
249    }
250
251    pub fn reads_payload(&self) -> bool {
252        self.root.reads_payload
253    }
254
255    pub fn reads_headers(&self) -> bool {
256        self.reads_headers
257    }
258
259    /// Which headers a caller has to decode before evaluating. A payload filter
260    /// without header predicates needs only `agdx.ct`, which
261    /// [`headers::content_type`](crate::filter::headers::content_type) reads
262    /// without decoding the rest of the block.
263    pub fn header_need(&self) -> HeaderNeed {
264        if self.reads_headers {
265            HeaderNeed::All
266        } else if self.root.reads_payload {
267            HeaderNeed::ContentType
268        } else {
269            HeaderNeed::None
270        }
271    }
272
273    /// Whether the record is selected, rejected, or could not be decoded.
274    /// Header predicates run before the payload is touched, and the payload is
275    /// decoded at most once, only when the headers do not already decide.
276    pub fn evaluate(&self, record: &FilterRecord<'_>, limits: &DecodeLimits) -> Verdict {
277        self.evaluate_with_fault(record, limits).0
278    }
279
280    /// Evaluate once and retain the decode fault without parsing the payload again.
281    pub fn evaluate_with_fault(
282        &self,
283        record: &FilterRecord<'_>,
284        limits: &DecodeLimits,
285    ) -> (Verdict, Option<FaultReason>) {
286        let mut context = Context::new(
287            record,
288            limits,
289            self.filter.codec,
290            &self.paths,
291            &self.decoders,
292        );
293        let truth = context.truth(&self.root);
294        self.evaluated(truth, context.mismatched)
295    }
296
297    /// Report each node's truth, for sample tests and previews. The verdict is
298    /// the delivery verdict of [`evaluate`](Self::evaluate), so headers that
299    /// decide the record still win over an undecodable payload. The tree
300    /// evaluates every node, and payload nodes carry no truth when the payload
301    /// does not decode.
302    pub fn explain(&self, record: &FilterRecord<'_>, limits: &DecodeLimits) -> FilterExplanation {
303        let mut context = Context::new(
304            record,
305            limits,
306            self.filter.codec,
307            &self.paths,
308            &self.decoders,
309        );
310        let truth = context.truth(&self.root);
311        let (verdict, fault) = self.evaluated(truth, context.mismatched);
312        let root = context.explain(&self.root);
313        FilterExplanation {
314            verdict,
315            fault,
316            root,
317        }
318    }
319}
320
321impl CompiledFilter {
322    // A record policy of `reject` decides here: the record is rejected, never
323    // counted as a fault, and the reason travels with the verdict so a preview
324    // or sample test can show why it was skipped. `pass` and every decode
325    // fault surface as a fault with the reason, which the caller resolves
326    // through `policy_for`.
327    fn evaluated(
328        &self,
329        truth: Result<Truth, FaultReason>,
330        mismatched: bool,
331    ) -> (Verdict, Option<FaultReason>) {
332        let reason = match truth {
333            Ok(Truth::Match) => return (Verdict::Selected, None),
334            Ok(Truth::NoMatch) => return (Verdict::Rejected, None),
335            Ok(Truth::Unknown) if !mismatched => return (Verdict::Rejected, None),
336            Ok(Truth::Unknown) => FaultReason::TypeMismatch,
337            Err(reason) => reason,
338        };
339        match self.record_policy(reason) {
340            Some(RecordPolicy::Reject) => (Verdict::Rejected, Some(reason)),
341            Some(RecordPolicy::Pass) | None => (Verdict::Fault, Some(reason)),
342        }
343    }
344}
345
346fn compile_node(expr: &FilterExpr) -> Result<Node, InvalidError> {
347    let kind = match expr {
348        FilterExpr::All(children) => NodeKind::All(
349            children
350                .iter()
351                .map(compile_node)
352                .collect::<Result<_, _>>()?,
353        ),
354        FilterExpr::Any(children) => NodeKind::Any(
355            children
356                .iter()
357                .map(compile_node)
358                .collect::<Result<_, _>>()?,
359        ),
360        FilterExpr::Not(child) => NodeKind::Not(Box::new(compile_node(child)?)),
361        FilterExpr::Pred(predicate) => NodeKind::Compare {
362            path: FieldPath::parse(&predicate.field)?,
363            op: predicate.op,
364            literal: compile_literal(&predicate.value)?,
365        },
366        FilterExpr::PredAs(coerced) => NodeKind::Coerced {
367            path: FieldPath::parse(&coerced.pred.field)?,
368            op: coerced.pred.op,
369            coerce: coerced.coerce,
370            literal: compile_coerced_literal(coerced.coerce, &coerced.pred.value)?,
371        },
372        FilterExpr::Present(path) => NodeKind::Present(path.clone()),
373        FilterExpr::Absent(path) => NodeKind::Absent(path.clone()),
374        FilterExpr::Header(header) => NodeKind::Header {
375            key: header.key.clone(),
376            op: header.op,
377            literal: compile_literal(&header.value)?,
378        },
379        FilterExpr::Text(predicate) => NodeKind::Text {
380            path: FieldPath::parse(&predicate.field)?,
381            matcher: TextMatcher::compile(predicate)?,
382        },
383        FilterExpr::HeaderText(predicate) => NodeKind::HeaderText {
384            key: predicate.field.clone(),
385            matcher: TextMatcher::compile(predicate)?,
386        },
387    };
388    Ok(Node {
389        kind,
390        reads_payload: expr.reads_payload(),
391    })
392}
393
394fn compile_literal(value: &TypedValue) -> Result<Literal, InvalidError> {
395    Ok(match value {
396        TypedValue::Null => Literal::Null,
397        TypedValue::Boolean(value) => Literal::Bool(*value),
398        TypedValue::Int(value) => Literal::Int(i128::from(*value)),
399        TypedValue::Long(value) => Literal::Int(i128::from(*value)),
400        TypedValue::Double(value) => Literal::Float(*value),
401        TypedValue::String(value) => Literal::Text(value.clone()),
402        TypedValue::List(items) => Literal::List(
403            items
404                .iter()
405                .map(compile_literal)
406                .collect::<Result<_, _>>()?,
407        ),
408        other => {
409            return Err(InvalidError::new(format!(
410                "unsupported filter literal {}",
411                other.diagnostic_text()
412            )));
413        }
414    })
415}
416
417fn compile_coerced_literal(
418    coerce: Coerce,
419    value: &TypedValue,
420) -> Result<CoercedLiteral, InvalidError> {
421    if let TypedValue::List(items) = value {
422        return items
423            .iter()
424            .map(|item| compile_coerced_literal(coerce, item))
425            .collect::<Result<_, _>>()
426            .map(CoercedLiteral::List);
427    }
428    let comparable = match (coerce, value) {
429        (Coerce::Timestamp { format }, TypedValue::String(text)) => {
430            format.micros_from_text(text).map(Comparable::Instant)
431        }
432        (Coerce::Timestamp { format }, TypedValue::Int(number)) => format
433            .micros_from_integer(i128::from(*number))
434            .map(Comparable::Instant),
435        (Coerce::Timestamp { format }, TypedValue::Long(number)) => format
436            .micros_from_integer(i128::from(*number))
437            .map(Comparable::Instant),
438        (Coerce::Number, TypedValue::String(text)) => {
439            ExactDecimal::parse(text).map(Comparable::Decimal)
440        }
441        (Coerce::Number, TypedValue::Int(number)) => Some(Comparable::Decimal(
442            ExactDecimal::from_integer(i128::from(*number)),
443        )),
444        (Coerce::Number, TypedValue::Long(number)) => Some(Comparable::Decimal(
445            ExactDecimal::from_integer(i128::from(*number)),
446        )),
447        _ => None,
448    };
449    match comparable {
450        Some(Comparable::Instant(micros)) => Ok(CoercedLiteral::Instant(micros)),
451        Some(Comparable::Decimal(decimal)) => Ok(CoercedLiteral::Decimal(decimal)),
452        None => Err(InvalidError::new(format!(
453            "literal {} does not parse under its coercion",
454            value.diagnostic_text()
455        ))),
456    }
457}
458
459struct Context<'a> {
460    record: &'a FilterRecord<'a>,
461    limits: &'a DecodeLimits,
462    codec: FilterCodec,
463    paths: &'a PathTrie,
464    decoders: &'a PayloadDecoders,
465    payload: Option<Result<Value, FaultReason>>,
466    // A predicate saw a value of a type it cannot compare.
467    mismatched: bool,
468}
469
470impl<'a> Context<'a> {
471    fn new(
472        record: &'a FilterRecord<'a>,
473        limits: &'a DecodeLimits,
474        codec: FilterCodec,
475        paths: &'a PathTrie,
476        decoders: &'a PayloadDecoders,
477    ) -> Self {
478        Self {
479            record,
480            limits,
481            codec,
482            paths,
483            decoders,
484            payload: None,
485            mismatched: false,
486        }
487    }
488
489    fn payload(&mut self) -> Result<&Value, FaultReason> {
490        if self.payload.is_none() {
491            let decoded = if self.declares_foreign_codec() {
492                Err(FaultReason::ForeignCodec)
493            } else {
494                self.decode()
495            };
496            self.payload = Some(decoded);
497        }
498        match self.payload.as_ref().expect("the payload was just decoded") {
499            Ok(value) => Ok(value),
500            Err(reason) => Err(*reason),
501        }
502    }
503
504    // `any` and codes this build does not know decode as usual: the content
505    // type dictionary grows, and an unknown code never rejects a record.
506    fn declares_foreign_codec(&self) -> bool {
507        let Some(header) = self.header(CONTENT_TYPE) else {
508            return false;
509        };
510        let code = match header.value {
511            HeaderValueRef::Uint(code) => u8::try_from(code).ok(),
512            HeaderValueRef::Int(code) => u8::try_from(code).ok(),
513            _ => None,
514        };
515        let declared = match code.and_then(ContentType::from_code) {
516            None | Some(ContentType::Any) => return false,
517            Some(declared) => declared,
518        };
519        let expected = match self.codec {
520            FilterCodec::Json => ContentType::Json,
521            FilterCodec::Cbor => ContentType::Cbor,
522            FilterCodec::Avro => ContentType::Avro,
523            FilterCodec::Protobuf => ContentType::Protobuf,
524            FilterCodec::HeadersOnly | FilterCodec::Unknown => return false,
525        };
526        declared != expected
527    }
528
529    fn decode(&self) -> Result<Value, FaultReason> {
530        match self.codec {
531            FilterCodec::Json => decode_json_paths(self.record.payload, self.limits, self.paths),
532            FilterCodec::Cbor | FilterCodec::Avro | FilterCodec::Protobuf => self
533                .decoders
534                .decode_paths(self.codec, self.record, self.limits, self.paths),
535            FilterCodec::HeadersOnly | FilterCodec::Unknown => Err(FaultReason::Malformed),
536        }
537    }
538
539    fn truth(&mut self, node: &Node) -> Result<Truth, FaultReason> {
540        match &node.kind {
541            NodeKind::All(children) => self.combine(children, Truth::NoMatch, Truth::Match),
542            NodeKind::Any(children) => self.combine(children, Truth::Match, Truth::NoMatch),
543            NodeKind::Not(child) => self.truth(child).map(negate),
544            NodeKind::Header { key, op, literal } => {
545                let (truth, mismatched) = self.header_truth(key, *op, literal);
546                self.mismatched |= mismatched;
547                Ok(truth)
548            }
549            NodeKind::HeaderText { key, matcher } => {
550                let (truth, mismatched) = self.header_text_truth(key, matcher);
551                self.mismatched |= mismatched;
552                Ok(truth)
553            }
554            leaf => {
555                let value = self.payload()?;
556                let (truth, mismatched) = payload_truth(leaf, value);
557                self.mismatched |= mismatched;
558                Ok(truth)
559            }
560        }
561    }
562
563    // `decisive` ends the combination at once (`no_match` for all, `match` for
564    // any). Children that read no payload run first, so headers can decide the
565    // node before the payload is ever decoded.
566    fn combine(
567        &mut self,
568        children: &[Node],
569        decisive: Truth,
570        otherwise: Truth,
571    ) -> Result<Truth, FaultReason> {
572        let mut unknown = false;
573        let header_first = children
574            .iter()
575            .filter(|child| !child.reads_payload)
576            .chain(children.iter().filter(|child| child.reads_payload));
577        for child in header_first {
578            match self.truth(child)? {
579                truth if truth == decisive => return Ok(decisive),
580                Truth::Unknown => unknown = true,
581                _ => {}
582            }
583        }
584        Ok(if unknown { Truth::Unknown } else { otherwise })
585    }
586
587    // The truth, and whether the header exists with a type the predicate
588    // cannot compare.
589    fn header_truth(&self, key: &str, op: CmpOp, literal: &Literal) -> (Truth, bool) {
590        let Some(header) = self.header(key) else {
591            return (Truth::Unknown, false);
592        };
593        let scalar = header_scalar(header.value);
594        let truth = compare_scalar(&scalar, None, op, literal);
595        (
596            truth,
597            truth == Truth::Unknown && !comparable(&scalar, None, op, literal),
598        )
599    }
600
601    fn header_text_truth(&self, key: &str, matcher: &TextMatcher) -> (Truth, bool) {
602        let Some(header) = self.header(key) else {
603            return (Truth::Unknown, false);
604        };
605        match header_scalar(header.value) {
606            Scalar::Text(text) => (truth_of(matcher.matches(text)), false),
607            _ => (Truth::Unknown, true),
608        }
609    }
610
611    // The last entry with a key wins, as the header decoder and the schema id
612    // lookup already resolve duplicates.
613    fn header(&self, key: &str) -> Option<&HeaderRef<'a>> {
614        self.record
615            .headers
616            .iter()
617            .rev()
618            .find(|header| header.key == key)
619    }
620
621    fn explain(&mut self, node: &Node) -> ExplainNode {
622        let (label, children) = match &node.kind {
623            NodeKind::All(children) => ("all".to_owned(), children.as_slice()),
624            NodeKind::Any(children) => ("any".to_owned(), children.as_slice()),
625            NodeKind::Not(child) => ("not".to_owned(), std::slice::from_ref(child.as_ref())),
626            leaf => {
627                let truth = match leaf {
628                    NodeKind::Header { key, op, literal } => {
629                        Some(self.header_truth(key, *op, literal).0)
630                    }
631                    NodeKind::HeaderText { key, matcher } => {
632                        Some(self.header_text_truth(key, matcher).0)
633                    }
634                    _ => self
635                        .payload()
636                        .ok()
637                        .map(|value| payload_truth(leaf, value).0),
638                };
639                return ExplainNode {
640                    label: leaf_label(leaf),
641                    truth,
642                    children: Vec::new(),
643                };
644            }
645        };
646        let explained: Vec<ExplainNode> =
647            children.iter().map(|child| self.explain(child)).collect();
648        let truths: Option<Vec<Truth>> = explained.iter().map(|child| child.truth).collect();
649        let truth = truths.map(|truths| match &node.kind {
650            NodeKind::All(_) => fold(&truths, Truth::NoMatch, Truth::Match),
651            NodeKind::Any(_) => fold(&truths, Truth::Match, Truth::NoMatch),
652            _ => negate(truths[0]),
653        });
654        ExplainNode {
655            label,
656            truth,
657            children: explained,
658        }
659    }
660}
661
662fn fold(truths: &[Truth], decisive: Truth, otherwise: Truth) -> Truth {
663    if truths.contains(&decisive) {
664        decisive
665    } else if truths.contains(&Truth::Unknown) {
666        Truth::Unknown
667    } else {
668        otherwise
669    }
670}
671
672const fn negate(truth: Truth) -> Truth {
673    match truth {
674        Truth::Match => Truth::NoMatch,
675        Truth::NoMatch => Truth::Match,
676        Truth::Unknown => Truth::Unknown,
677    }
678}
679
680// The truth, and whether the field exists, is not null, and has a type the
681// predicate cannot compare. A missing or null field is plain unknown.
682fn payload_truth(leaf: &NodeKind, value: &Value) -> (Truth, bool) {
683    match leaf {
684        NodeKind::Compare { path, op, literal } => match resolve(value, path) {
685            None => (missing_truth(*op, literal), false),
686            Some(field) => {
687                let scalar = scalar_of(field);
688                let truth = compare_scalar(&scalar, Some(field), *op, literal);
689                let mismatched =
690                    truth == Truth::Unknown && !comparable(&scalar, Some(field), *op, literal);
691                (truth, mismatched)
692            }
693        },
694        NodeKind::Coerced {
695            path,
696            op,
697            coerce,
698            literal,
699        } => match resolve(value, path) {
700            None => (Truth::Unknown, false),
701            Some(field) => match coerce_value(*coerce, field) {
702                Some(comparable) => (compare_coerced(&comparable, *op, literal), false),
703                None => (Truth::Unknown, !field.is_null()),
704            },
705        },
706        NodeKind::Text { path, matcher } => match resolve(value, path) {
707            None | Some(Value::Null) => (Truth::Unknown, false),
708            Some(Value::String(text)) => (truth_of(matcher.matches(text)), false),
709            Some(_) => (Truth::Unknown, true),
710        },
711        NodeKind::Present(path) => (truth_of(resolve(value, path).is_some()), false),
712        NodeKind::Absent(path) => (truth_of(resolve(value, path).is_none()), false),
713        NodeKind::All(_)
714        | NodeKind::Any(_)
715        | NodeKind::Not(_)
716        | NodeKind::Header { .. }
717        | NodeKind::HeaderText { .. } => unreachable!("only payload leaves read the payload"),
718    }
719}
720
721fn header_scalar(value: HeaderValueRef<'_>) -> Scalar<'_> {
722    match value {
723        HeaderValueRef::Bool(value) => Scalar::Bool(value),
724        HeaderValueRef::Int(value) => Scalar::Int(i128::from(value)),
725        HeaderValueRef::Uint(value) => Scalar::Int(i128::from(value)),
726        HeaderValueRef::Float(value) => Scalar::Float(value),
727        HeaderValueRef::String(value) => Scalar::Text(value),
728        // Raw bytes that spell UTF-8 compare as text, pinned by the shared
729        // corpus. Anything else has no scalar meaning here.
730        HeaderValueRef::Raw(bytes) => match std::str::from_utf8(bytes) {
731            Ok(text) => Scalar::Text(text),
732            Err(_) => Scalar::Composite,
733        },
734    }
735}
736
737const fn truth_of(matched: bool) -> Truth {
738    if matched {
739        Truth::Match
740    } else {
741        Truth::NoMatch
742    }
743}
744
745// Whether the value's kind can be compared with the literal's kind at all. A
746// comparison that stays unknown between comparable kinds, such as an integer
747// beyond 2^53 against a double or a float header holding NaN, is undecidable,
748// not a type mismatch. A null value is never a mismatch.
749fn comparable(scalar: &Scalar<'_>, field: Option<&Value>, op: CmpOp, literal: &Literal) -> bool {
750    match (scalar, literal) {
751        (Scalar::Null, _) | (_, Literal::Null) => true,
752        (Scalar::Text(_), Literal::Text(_)) | (Scalar::Bool(_), Literal::Bool(_)) => true,
753        (Scalar::Int(_) | Scalar::Float(_), Literal::Int(_) | Literal::Float(_)) => true,
754        (scalar, Literal::List(items)) => {
755            items.iter().any(|item| comparable(scalar, field, op, item))
756        }
757        (Scalar::Composite, _) => op == CmpOp::Contains && matches!(field, Some(Value::Array(_))),
758        _ => false,
759    }
760}
761
762// A missing field is unknown, except for the null tests: `eq null` matches a
763// missing field and `ne null` does not.
764fn missing_truth(op: CmpOp, literal: &Literal) -> Truth {
765    match (op, literal) {
766        (CmpOp::Eq, Literal::Null) => Truth::Match,
767        (CmpOp::Ne, Literal::Null) => Truth::NoMatch,
768        _ => Truth::Unknown,
769    }
770}
771
772fn compare_scalar(
773    scalar: &Scalar<'_>,
774    field: Option<&Value>,
775    op: CmpOp,
776    literal: &Literal,
777) -> Truth {
778    match (scalar, literal) {
779        (Scalar::Null, Literal::Null) => return truth_of(op == CmpOp::Eq),
780        (Scalar::Null, _) => return Truth::Unknown,
781        (_, Literal::Null) => return truth_of(op == CmpOp::Ne),
782        _ => {}
783    }
784    match op {
785        CmpOp::Eq => scalar_eq(scalar, literal).map_or(Truth::Unknown, truth_of),
786        CmpOp::Ne => scalar_eq(scalar, literal).map_or(Truth::Unknown, |equal| truth_of(!equal)),
787        CmpOp::Lt | CmpOp::Lte | CmpOp::Gt | CmpOp::Gte => match scalar_cmp(scalar, literal) {
788            Some(ordering) => truth_of(match op {
789                CmpOp::Lt => ordering == Ordering::Less,
790                CmpOp::Lte => ordering != Ordering::Greater,
791                CmpOp::Gt => ordering == Ordering::Greater,
792                _ => ordering != Ordering::Less,
793            }),
794            None => Truth::Unknown,
795        },
796        CmpOp::In => {
797            let Literal::List(items) = literal else {
798                return Truth::Unknown;
799            };
800            let mut unknown = false;
801            for item in items {
802                match scalar_eq(scalar, item) {
803                    Some(true) => return Truth::Match,
804                    Some(false) => {}
805                    None => unknown = true,
806                }
807            }
808            if unknown {
809                Truth::Unknown
810            } else {
811                Truth::NoMatch
812            }
813        }
814        CmpOp::Contains => match (scalar, literal, field) {
815            (Scalar::Text(text), Literal::Text(needle), _) => {
816                truth_of(text.contains(needle.as_str()))
817            }
818            (Scalar::Composite, _, Some(Value::Array(elements))) => truth_of(
819                elements
820                    .iter()
821                    .any(|element| scalar_eq(&scalar_of(element), literal) == Some(true)),
822            ),
823            _ => Truth::Unknown,
824        },
825        CmpOp::Prefix => match (scalar, literal) {
826            (Scalar::Text(text), Literal::Text(prefix)) => {
827                truth_of(text.starts_with(prefix.as_str()))
828            }
829            _ => Truth::Unknown,
830        },
831    }
832}
833
834fn scalar_eq(scalar: &Scalar<'_>, literal: &Literal) -> Option<bool> {
835    match (scalar, literal) {
836        (Scalar::Bool(left), Literal::Bool(right)) => Some(left == right),
837        (Scalar::Text(left), Literal::Text(right)) => Some(*left == right.as_str()),
838        _ => scalar_cmp(scalar, literal).map(|ordering| ordering == Ordering::Equal),
839    }
840}
841
842fn scalar_cmp(scalar: &Scalar<'_>, literal: &Literal) -> Option<Ordering> {
843    match (scalar, literal) {
844        (Scalar::Int(left), Literal::Int(right)) => Some(left.cmp(right)),
845        (Scalar::Float(left), Literal::Float(right)) => left.partial_cmp(right),
846        (Scalar::Int(left), Literal::Float(right)) => exact_double(*left)?.partial_cmp(right),
847        (Scalar::Float(left), Literal::Int(right)) => left.partial_cmp(&exact_double(*right)?),
848        (Scalar::Text(left), Literal::Text(right)) => Some(left.as_bytes().cmp(right.as_bytes())),
849        _ => None,
850    }
851}
852
853#[allow(clippy::cast_precision_loss)]
854fn exact_double(value: i128) -> Option<f64> {
855    let magnitude = value.unsigned_abs();
856    let significant_bits = u128::BITS - magnitude.leading_zeros();
857    (significant_bits <= 53 || magnitude.trailing_zeros() >= significant_bits - 53)
858        .then_some(value as f64)
859}
860
861fn compare_coerced(comparable: &Comparable, op: CmpOp, literal: &CoercedLiteral) -> Truth {
862    let ordering = |literal: &CoercedLiteral| match (comparable, literal) {
863        (Comparable::Instant(left), CoercedLiteral::Instant(right)) => Some(left.cmp(right)),
864        (Comparable::Decimal(left), CoercedLiteral::Decimal(right)) => Some(left.cmp(right)),
865        _ => None,
866    };
867    match (op, literal) {
868        (CmpOp::In, CoercedLiteral::List(items)) => truth_of(
869            items
870                .iter()
871                .any(|item| ordering(item) == Some(Ordering::Equal)),
872        ),
873        (_, literal) => match ordering(literal) {
874            Some(ordering) => truth_of(match op {
875                CmpOp::Eq => ordering == Ordering::Equal,
876                CmpOp::Ne => ordering != Ordering::Equal,
877                CmpOp::Lt => ordering == Ordering::Less,
878                CmpOp::Lte => ordering != Ordering::Greater,
879                CmpOp::Gt => ordering == Ordering::Greater,
880                CmpOp::Gte => ordering != Ordering::Less,
881                CmpOp::In | CmpOp::Contains | CmpOp::Prefix => return Truth::Unknown,
882            }),
883            None => Truth::Unknown,
884        },
885    }
886}
887
888fn coerce_value(coerce: Coerce, field: &Value) -> Option<Comparable> {
889    match coerce {
890        Coerce::Timestamp { format } => {
891            let micros = match (format, field) {
892                (_, Value::String(text)) => format.micros_from_text(text),
893                (TimestampFormat::Rfc3339, _) => None,
894                (_, Value::Number(number)) => {
895                    let integer = number
896                        .as_i64()
897                        .map(i128::from)
898                        .or_else(|| number.as_u64().map(i128::from))?;
899                    format.micros_from_integer(integer)
900                }
901                _ => None,
902            }?;
903            Some(Comparable::Instant(micros))
904        }
905        Coerce::Number => {
906            let decimal = match field {
907                Value::String(text) => ExactDecimal::parse(text),
908                Value::Number(number) => number
909                    .as_i64()
910                    .map(|value| ExactDecimal::from_integer(i128::from(value)))
911                    .or_else(|| {
912                        number
913                            .as_u64()
914                            .map(|value| ExactDecimal::from_integer(i128::from(value)))
915                    })
916                    .or_else(|| number.as_f64().and_then(ExactDecimal::from_f64)),
917                _ => None,
918            }?;
919            Some(Comparable::Decimal(decimal))
920        }
921    }
922}
923
924fn scalar_of(value: &Value) -> Scalar<'_> {
925    match value {
926        Value::Null => Scalar::Null,
927        Value::Bool(value) => Scalar::Bool(*value),
928        Value::Number(number) => number
929            .as_i64()
930            .map(|value| Scalar::Int(i128::from(value)))
931            .or_else(|| number.as_u64().map(|value| Scalar::Int(i128::from(value))))
932            .or_else(|| number.as_f64().map(Scalar::Float))
933            .unwrap_or(Scalar::Composite),
934        Value::String(text) => Scalar::Text(text),
935        Value::Array(_) | Value::Object(_) => Scalar::Composite,
936    }
937}
938
939fn resolve<'v>(value: &'v Value, path: &FieldPath) -> Option<&'v Value> {
940    path.segments()
941        .iter()
942        .try_fold(value, |current, segment| match (segment, current) {
943            (PathSegment::Key(key), Value::Object(map)) => map.get(key),
944            (PathSegment::Index(index), Value::Array(items)) => items.get(*index as usize),
945            _ => None,
946        })
947}
948
949/// Decode a JSON payload within `limits`. The depth is checked by a scan over
950/// the raw bytes before any value is built, so a hostile payload cannot make
951/// the parser recurse or allocate past the limit.
952pub fn decode_json(payload: &[u8], limits: &DecodeLimits) -> Result<Value, FaultReason> {
953    decode_json_paths(payload, limits, &PathTrie::whole())
954}
955
956// The depth scan also bounds subtrees that the filter does not retain.
957fn decode_json_paths(
958    payload: &[u8],
959    limits: &DecodeLimits,
960    paths: &PathTrie,
961) -> Result<Value, FaultReason> {
962    if payload.len() > limits.max_payload_bytes {
963        return Err(FaultReason::TooLarge);
964    }
965    if !json_depth_within(payload, limits.max_depth.min(MAX_FILTER_PARSE_DEPTH)) {
966        return Err(FaultReason::TooDeep);
967    }
968    let mut deserializer = serde_json::Deserializer::from_slice(payload);
969    let value = Pruned(paths)
970        .deserialize(&mut deserializer)
971        .map_err(|_| FaultReason::Malformed)?;
972    deserializer.end().map_err(|_| FaultReason::Malformed)?;
973    Ok(value)
974}
975
976impl PathTrie {
977    pub(super) fn whole() -> Self {
978        Self {
979            whole: true,
980            ..Self::default()
981        }
982    }
983
984    pub(super) fn child(&self, key: &str) -> Option<&PathTrie> {
985        self.keys
986            .iter()
987            .find_map(|(name, child)| (name == key).then_some(child))
988    }
989
990    fn child_mut(&mut self, key: &str) -> &mut PathTrie {
991        let index = match self.keys.iter().position(|(name, _)| name == key) {
992            Some(index) => index,
993            None => {
994                self.keys.push((key.to_owned(), PathTrie::default()));
995                self.keys.len() - 1
996            }
997        };
998        &mut self.keys[index].1
999    }
1000
1001    pub(super) fn index(&self, index: u32) -> Option<&PathTrie> {
1002        self.indices
1003            .iter()
1004            .find_map(|(number, child)| (*number == index).then_some(child))
1005    }
1006
1007    fn index_mut(&mut self, index: u32) -> &mut PathTrie {
1008        let position = match self.indices.iter().position(|(number, _)| *number == index) {
1009            Some(position) => position,
1010            None => {
1011                self.indices.push((index, Self::default()));
1012                self.indices.len() - 1
1013            }
1014        };
1015        &mut self.indices[position].1
1016    }
1017
1018    fn insert(&mut self, path: &FieldPath) {
1019        let mut node = self;
1020        for segment in path.segments() {
1021            if node.whole {
1022                return;
1023            }
1024            match segment {
1025                PathSegment::Key(key) => node = node.child_mut(key),
1026                PathSegment::Index(index) => node = node.index_mut(*index),
1027            }
1028        }
1029        node.whole = true;
1030        node.keys.clear();
1031        node.indices.clear();
1032    }
1033}
1034
1035fn collect_paths(expr: &FilterExpr, paths: &mut PathTrie) -> Result<(), InvalidError> {
1036    match expr {
1037        FilterExpr::All(children) | FilterExpr::Any(children) => {
1038            for child in children {
1039                collect_paths(child, paths)?;
1040            }
1041        }
1042        FilterExpr::Not(child) => collect_paths(child, paths)?,
1043        FilterExpr::Pred(predicate) => paths.insert(&FieldPath::parse(&predicate.field)?),
1044        FilterExpr::PredAs(coerced) => paths.insert(&FieldPath::parse(&coerced.pred.field)?),
1045        FilterExpr::Present(path) | FilterExpr::Absent(path) => paths.insert(path),
1046        FilterExpr::Text(predicate) => paths.insert(&FieldPath::parse(&predicate.field)?),
1047        FilterExpr::Header(_) | FilterExpr::HeaderText(_) => {}
1048    }
1049    Ok(())
1050}
1051
1052// Builds a `serde_json::Value` that holds only the values on the trie's paths.
1053// Values off the paths are validated without retention. Array positions stay
1054// stable, and a repeated key keeps its last value, as `Value` does.
1055struct Pruned<'t>(&'t PathTrie);
1056
1057impl<'de> DeserializeSeed<'de> for Pruned<'_> {
1058    type Value = Value;
1059
1060    fn deserialize<D: Deserializer<'de>>(self, deserializer: D) -> Result<Value, D::Error> {
1061        if self.0.whole {
1062            return Value::deserialize(deserializer);
1063        }
1064        deserializer.deserialize_any(self)
1065    }
1066}
1067
1068impl<'de> Visitor<'de> for Pruned<'_> {
1069    type Value = Value;
1070
1071    fn expecting(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1072        f.write_str("a JSON value")
1073    }
1074
1075    fn visit_bool<E>(self, value: bool) -> Result<Value, E> {
1076        Ok(Value::Bool(value))
1077    }
1078
1079    fn visit_i64<E>(self, value: i64) -> Result<Value, E> {
1080        Ok(Value::from(value))
1081    }
1082
1083    fn visit_u64<E>(self, value: u64) -> Result<Value, E> {
1084        Ok(Value::from(value))
1085    }
1086
1087    fn visit_f64<E>(self, value: f64) -> Result<Value, E> {
1088        Ok(serde_json::Number::from_f64(value).map_or(Value::Null, Value::Number))
1089    }
1090
1091    fn visit_str<E>(self, _value: &str) -> Result<Value, E> {
1092        Ok(Value::Null)
1093    }
1094
1095    fn visit_unit<E>(self) -> Result<Value, E> {
1096        Ok(Value::Null)
1097    }
1098
1099    fn visit_seq<A: SeqAccess<'de>>(self, mut seq: A) -> Result<Value, A::Error> {
1100        let mut items = Vec::new();
1101        let ignored = PathTrie::default();
1102        let mut index = 0_u32;
1103        while let Some(item) =
1104            seq.next_element_seed(Pruned(self.0.index(index).unwrap_or(&ignored)))?
1105        {
1106            if self.0.index(index).is_some() {
1107                items.resize(index as usize, Value::Null);
1108                items.push(item);
1109            }
1110            index = index.saturating_add(1);
1111        }
1112        Ok(Value::Array(items))
1113    }
1114
1115    fn visit_map<A: MapAccess<'de>>(self, mut map: A) -> Result<Value, A::Error> {
1116        let mut object = Map::new();
1117        let ignored = PathTrie::default();
1118        while let Some(JsonKey(key)) = map.next_key::<JsonKey<'_>>()? {
1119            match self.0.child(key.as_ref()) {
1120                Some(child) => {
1121                    let value = map.next_value_seed(Pruned(child))?;
1122                    object.insert(key.into_owned(), value);
1123                }
1124                None => {
1125                    // IgnoredAny accepts overflowing numbers and lone
1126                    // surrogates. Unused fields must obey the same JSON
1127                    // rules as fields the filter reads.
1128                    map.next_value_seed(Pruned(&ignored))?;
1129                }
1130            }
1131        }
1132        Ok(Value::Object(object))
1133    }
1134}
1135
1136struct JsonKey<'a>(Cow<'a, str>);
1137
1138impl<'de> Deserialize<'de> for JsonKey<'de> {
1139    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
1140        struct KeyVisitor;
1141        impl<'de> Visitor<'de> for KeyVisitor {
1142            type Value = JsonKey<'de>;
1143            fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
1144                formatter.write_str("an object key")
1145            }
1146            fn visit_borrowed_str<E>(self, value: &'de str) -> Result<Self::Value, E> {
1147                Ok(JsonKey(Cow::Borrowed(value)))
1148            }
1149            fn visit_str<E>(self, value: &str) -> Result<Self::Value, E> {
1150                Ok(JsonKey(Cow::Owned(value.to_owned())))
1151            }
1152            fn visit_string<E>(self, value: String) -> Result<Self::Value, E> {
1153                Ok(JsonKey(Cow::Owned(value)))
1154            }
1155        }
1156        deserializer.deserialize_str(KeyVisitor)
1157    }
1158}
1159
1160// String bodies are skipped with memchr, which finds the next quote or
1161// backslash many bytes at a time. Most payload bytes sit inside strings.
1162fn json_depth_within(payload: &[u8], max_depth: usize) -> bool {
1163    let mut depth = 0usize;
1164    let mut index = 0;
1165    while index < payload.len() {
1166        match payload[index] {
1167            b'"' => {
1168                index += 1;
1169                loop {
1170                    let Some(found) = memchr::memchr2(b'"', b'\\', &payload[index..]) else {
1171                        return true;
1172                    };
1173                    index += found;
1174                    if payload[index] == b'"' {
1175                        break;
1176                    }
1177                    index += 2;
1178                    if index >= payload.len() {
1179                        return true;
1180                    }
1181                }
1182            }
1183            b'[' | b'{' => {
1184                depth += 1;
1185                if depth > max_depth {
1186                    return false;
1187                }
1188            }
1189            b']' | b'}' => depth = depth.saturating_sub(1),
1190            _ => {}
1191        }
1192        index += 1;
1193    }
1194    true
1195}
1196
1197fn leaf_label(leaf: &NodeKind) -> String {
1198    match leaf {
1199        NodeKind::Compare { path, op, literal } => {
1200            format!("{path} {op} {}", literal_label(literal))
1201        }
1202        NodeKind::Coerced {
1203            path,
1204            op,
1205            coerce,
1206            literal,
1207        } => {
1208            let coerce = match coerce {
1209                Coerce::Timestamp { format } => format!("timestamp({format})"),
1210                Coerce::Number => "number".to_owned(),
1211            };
1212            format!("{coerce}({path}) {op} {}", coerced_label(literal))
1213        }
1214        NodeKind::Present(path) => format!("{path} is present"),
1215        NodeKind::Absent(path) => format!("{path} is absent"),
1216        NodeKind::Header { key, op, literal } => {
1217            format!("header {key} {op} {}", literal_label(literal))
1218        }
1219        NodeKind::Text { path, matcher } => {
1220            format!("{path} {} {}", matcher.kind(), matcher.label())
1221        }
1222        NodeKind::HeaderText { key, matcher } => {
1223            format!("header {key} {} {}", matcher.kind(), matcher.label())
1224        }
1225        NodeKind::All(_) | NodeKind::Any(_) | NodeKind::Not(_) => unreachable!("not a leaf"),
1226    }
1227}
1228
1229fn literal_label(literal: &Literal) -> String {
1230    match literal {
1231        Literal::Null => "null".to_owned(),
1232        Literal::Bool(value) => value.to_string(),
1233        Literal::Int(value) => value.to_string(),
1234        Literal::Float(value) => value.to_string(),
1235        Literal::Text(value) => format!("{value:?}"),
1236        Literal::List(items) => format!(
1237            "[{}]",
1238            items
1239                .iter()
1240                .map(literal_label)
1241                .collect::<Vec<_>>()
1242                .join(", ")
1243        ),
1244    }
1245}
1246
1247fn coerced_label(literal: &CoercedLiteral) -> String {
1248    match literal {
1249        CoercedLiteral::Instant(micros) => format!("{micros}us"),
1250        CoercedLiteral::Decimal(decimal) => format!("{decimal:?}"),
1251        CoercedLiteral::List(items) => format!(
1252            "[{}]",
1253            items
1254                .iter()
1255                .map(coerced_label)
1256                .collect::<Vec<_>>()
1257                .join(", ")
1258        ),
1259    }
1260}
1261
1262#[cfg(test)]
1263mod tests {
1264    use super::*;
1265    use crate::filter::text::TextMatch;
1266
1267    const SAFE_MODE: &str = r#"{"op":"u","table":"satellites","changed":["mode"],"after":{"id":"sat-042","name":"Kestrel-42","mode":"safe","orbit":"leo","battery_pct":61}}"#;
1268    const SAFE_MODE_VALUES: &str = r#"{"op":"u","table":"satellites","after":{"id":"sat-042","name":"Kestrel-42","mode":"safe","orbit":"leo","battery_pct":61}}"#;
1269    const DECOMMISSION: &str = r#"{"op":"d","table":"satellites","before":{"id":"sat-042"}}"#;
1270    const TELEMETRY: &str = r#"{"event":"satellite.telemetry_changed","satellite_id":"sat-042","fields":{"mode":null}}"#;
1271    const GROUND_STATION: &str = r#"{"op":"u","table":"ground_stations","changed":["status"],"after":{"id":"svalbard","status":"online"}}"#;
1272
1273    fn verdict(filter: &ConsumerFilter, payload: &str) -> Verdict {
1274        CompiledFilter::compile(filter).expect("compiles").evaluate(
1275            &FilterRecord {
1276                payload: payload.as_bytes(),
1277                headers: &[],
1278            },
1279            &DecodeLimits::default(),
1280        )
1281    }
1282
1283    fn strict() -> ConsumerFilter {
1284        ConsumerFilter::json(FilterExpr::any([
1285            FilterExpr::all([
1286                FilterExpr::pred("table", CmpOp::Eq, "satellites"),
1287                FilterExpr::pred("op", CmpOp::Eq, "u"),
1288                FilterExpr::pred("changed", CmpOp::Contains, "mode"),
1289                FilterExpr::pred("after.mode", CmpOp::Eq, "safe"),
1290            ]),
1291            FilterExpr::all([
1292                FilterExpr::pred("table", CmpOp::Eq, "satellites"),
1293                FilterExpr::pred("op", CmpOp::Eq, "d"),
1294            ]),
1295            FilterExpr::all([
1296                FilterExpr::pred("event", CmpOp::Eq, "satellite.telemetry_changed"),
1297                FilterExpr::present("fields.mode"),
1298            ]),
1299        ]))
1300    }
1301
1302    fn legacy() -> ConsumerFilter {
1303        ConsumerFilter::json(FilterExpr::all([
1304            FilterExpr::pred("table", CmpOp::Eq, "satellites"),
1305            FilterExpr::pred("op", CmpOp::Eq, "u"),
1306            FilterExpr::pred("after.mode", CmpOp::Eq, "safe"),
1307        ]))
1308    }
1309
1310    #[test]
1311    fn given_an_indexed_array_path_when_decoding_then_should_retain_only_the_requested_field_and_validate_the_rest()
1312     {
1313        let mut paths = PathTrie::default();
1314        paths.insert(&FieldPath::parse("items[1].sku").expect("path"));
1315        let value = decode_json_paths(
1316            br#"{"items":[{"other":"unused"},{"sku":"chosen","unused":[1,2,3]},{"sku":"unread"}]}"#,
1317            &DecodeLimits::default(),
1318            &paths,
1319        )
1320        .expect("decoded");
1321        assert_eq!(value, serde_json::json!({"items":[null,{"sku":"chosen"}]}));
1322        assert_eq!(
1323            decode_json_paths(
1324                br#"{"items":[{"bad":"\ud800"},{"sku":"chosen"}]}"#,
1325                &DecodeLimits::default(),
1326                &paths
1327            ),
1328            Err(FaultReason::Malformed)
1329        );
1330    }
1331
1332    #[test]
1333    fn given_the_safe_mode_fixtures_when_evaluated_then_should_match_the_scenario_table() {
1334        let expected = [
1335            (SAFE_MODE, Verdict::Selected, Verdict::Selected),
1336            (SAFE_MODE_VALUES, Verdict::Rejected, Verdict::Selected),
1337            (DECOMMISSION, Verdict::Selected, Verdict::Rejected),
1338            (TELEMETRY, Verdict::Selected, Verdict::Rejected),
1339            (GROUND_STATION, Verdict::Rejected, Verdict::Rejected),
1340        ];
1341        for (payload, strict_verdict, legacy_verdict) in expected {
1342            assert_eq!(
1343                verdict(&strict(), payload),
1344                strict_verdict,
1345                "strict on {payload}"
1346            );
1347            assert_eq!(
1348                verdict(&legacy(), payload),
1349                legacy_verdict,
1350                "legacy on {payload}"
1351            );
1352        }
1353    }
1354
1355    #[test]
1356    fn given_missing_and_null_fields_when_compared_then_should_follow_the_null_rules() {
1357        let eq_null =
1358            ConsumerFilter::json(FilterExpr::pred("missing", CmpOp::Eq, TypedValue::Null));
1359        let ne_null =
1360            ConsumerFilter::json(FilterExpr::pred("missing", CmpOp::Ne, TypedValue::Null));
1361        let not_eq = ConsumerFilter::json(FilterExpr::negate(FilterExpr::pred(
1362            "missing",
1363            CmpOp::Eq,
1364            "x",
1365        )));
1366        let present_null = ConsumerFilter::json(FilterExpr::present("value"));
1367        let ne_null_on_null =
1368            ConsumerFilter::json(FilterExpr::pred("value", CmpOp::Ne, TypedValue::Null));
1369        assert_eq!(verdict(&eq_null, r#"{"value":null}"#), Verdict::Selected);
1370        assert_eq!(verdict(&ne_null, r#"{"value":null}"#), Verdict::Rejected);
1371        assert_eq!(
1372            verdict(&not_eq, r#"{}"#),
1373            Verdict::Rejected,
1374            "not(unknown) is unknown"
1375        );
1376        assert_eq!(
1377            verdict(&present_null, r#"{"value":null}"#),
1378            Verdict::Selected
1379        );
1380        assert_eq!(
1381            verdict(&ne_null_on_null, r#"{"value":null}"#),
1382            Verdict::Rejected
1383        );
1384    }
1385
1386    #[test]
1387    fn given_numbers_when_compared_then_should_stay_exact() {
1388        let big =
1389            ConsumerFilter::json(FilterExpr::pred("id", CmpOp::Eq, 9_007_199_254_740_993_i64));
1390        assert_eq!(
1391            verdict(&big, r#"{"id":9007199254740993}"#),
1392            Verdict::Selected
1393        );
1394        assert_eq!(
1395            verdict(&big, r#"{"id":9007199254740992}"#),
1396            Verdict::Rejected
1397        );
1398        let mixed = ConsumerFilter::json(FilterExpr::pred("port", CmpOp::Gte, 5061.0));
1399        assert_eq!(verdict(&mixed, r#"{"port":5061}"#), Verdict::Selected);
1400        let inexact = ConsumerFilter::json(FilterExpr::pred("id", CmpOp::Eq, 1.0));
1401        assert_eq!(
1402            verdict(&inexact, r#"{"id":9007199254740993}"#),
1403            Verdict::Rejected
1404        );
1405        let text_vs_number = ConsumerFilter::json(FilterExpr::pred("port", CmpOp::Eq, "5061"));
1406        assert_eq!(
1407            verdict(&text_vs_number, r#"{"port":5061}"#),
1408            Verdict::Rejected
1409        );
1410    }
1411
1412    #[test]
1413    fn given_contains_and_prefix_when_compared_then_should_be_literal() {
1414        let contains = ConsumerFilter::json(FilterExpr::pred("name", CmpOp::Contains, "%"));
1415        assert_eq!(verdict(&contains, r#"{"name":"100%"}"#), Verdict::Selected);
1416        assert_eq!(verdict(&contains, r#"{"name":"100"}"#), Verdict::Rejected);
1417        let array = ConsumerFilter::json(FilterExpr::pred("tags", CmpOp::Contains, 2));
1418        assert_eq!(verdict(&array, r#"{"tags":[1,2,3]}"#), Verdict::Selected);
1419        assert_eq!(verdict(&array, r#"{"tags":["2"]}"#), Verdict::Rejected);
1420        let prefix = ConsumerFilter::json(FilterExpr::pred("name", CmpOp::Prefix, "eci_"));
1421        assert_eq!(verdict(&prefix, r#"{"name":"eci_x"}"#), Verdict::Selected);
1422        assert_eq!(verdict(&prefix, r#"{"name":"ecix"}"#), Verdict::Rejected);
1423    }
1424
1425    #[test]
1426    fn given_in_lists_when_compared_then_should_match_any_item() {
1427        let filter = ConsumerFilter::json(FilterExpr::pred(
1428            "table",
1429            CmpOp::In,
1430            TypedValue::List(vec!["satellites".into(), "ground_stations".into()]),
1431        ));
1432        assert_eq!(
1433            verdict(&filter, r#"{"table":"ground_stations"}"#),
1434            Verdict::Selected
1435        );
1436        assert_eq!(
1437            verdict(&filter, r#"{"table":"launches"}"#),
1438            Verdict::Rejected
1439        );
1440    }
1441
1442    #[test]
1443    fn given_coercions_when_compared_then_should_use_instants_and_exact_decimals() {
1444        let after = ConsumerFilter::json(FilterExpr::pred_as(
1445            "at",
1446            CmpOp::Gt,
1447            "2026-09-21T18:00:00Z",
1448            Coerce::Timestamp {
1449                format: TimestampFormat::Rfc3339,
1450            },
1451        ));
1452        assert_eq!(
1453            verdict(&after, r#"{"at":"2026-09-21T20:04:12+02:00"}"#),
1454            Verdict::Selected
1455        );
1456        assert_eq!(
1457            verdict(&after, r#"{"at":"2026-09-21T19:59:59+02:00"}"#),
1458            Verdict::Rejected
1459        );
1460        assert_eq!(
1461            verdict(&after, r#"{"at":"2026-09-21T18:04:12"}"#),
1462            Verdict::Rejected,
1463            "no zone is unknown"
1464        );
1465        let millis = ConsumerFilter::json(FilterExpr::pred_as(
1466            "timestamp_ms",
1467            CmpOp::Lt,
1468            "2026-09-21T16:04:13Z",
1469            Coerce::Timestamp {
1470                format: TimestampFormat::EpochMillis,
1471            },
1472        ));
1473        assert!(
1474            CompiledFilter::compile(&millis).is_err(),
1475            "an epoch literal must be an integer"
1476        );
1477        let amount = ConsumerFilter::json(FilterExpr::pred_as(
1478            "amount",
1479            CmpOp::Gt,
1480            "9007199254740992",
1481            Coerce::Number,
1482        ));
1483        assert_eq!(
1484            verdict(&amount, r#"{"amount":"9007199254740993"}"#),
1485            Verdict::Selected
1486        );
1487        assert_eq!(
1488            verdict(&amount, r#"{"amount":"9007199254740992.0"}"#),
1489            Verdict::Rejected
1490        );
1491        assert_eq!(verdict(&amount, r#"{"amount":"abc"}"#), Verdict::Rejected);
1492    }
1493
1494    #[test]
1495    fn given_header_predicates_when_they_decide_then_should_skip_the_payload() {
1496        let filter = ConsumerFilter::json(FilterExpr::all([
1497            FilterExpr::pred("op", CmpOp::Eq, "d"),
1498            FilterExpr::header("table", CmpOp::Eq, "satellites"),
1499        ]));
1500        let compiled = CompiledFilter::compile(&filter).expect("compiles");
1501        let ground_station = [HeaderRef {
1502            key: "table",
1503            value: HeaderValueRef::String("ground_stations"),
1504        }];
1505        let rejected = compiled.evaluate(
1506            &FilterRecord {
1507                payload: b"not json at all",
1508                headers: &ground_station,
1509            },
1510            &DecodeLimits::default(),
1511        );
1512        assert_eq!(
1513            rejected,
1514            Verdict::Rejected,
1515            "the header decides before any decode"
1516        );
1517        let satellite = [HeaderRef {
1518            key: "table",
1519            value: HeaderValueRef::String("satellites"),
1520        }];
1521        let fault = compiled.evaluate(
1522            &FilterRecord {
1523                payload: b"not json at all",
1524                headers: &satellite,
1525            },
1526            &DecodeLimits::default(),
1527        );
1528        assert_eq!(
1529            fault,
1530            Verdict::Fault,
1531            "the payload is needed and does not decode"
1532        );
1533    }
1534
1535    #[test]
1536    fn given_payload_limits_when_exceeded_then_should_fault_without_parsing() {
1537        let filter = ConsumerFilter::json(FilterExpr::present("a"));
1538        let compiled = CompiledFilter::compile(&filter).expect("compiles");
1539        let limits = DecodeLimits {
1540            max_payload_bytes: 64,
1541            max_depth: 3,
1542        };
1543        let deep = r#"{"a":{"b":{"c":{"d":1}}}}"#;
1544        let record = FilterRecord {
1545            payload: deep.as_bytes(),
1546            headers: &[],
1547        };
1548        assert_eq!(compiled.evaluate(&record, &limits), Verdict::Fault);
1549        assert_eq!(
1550            compiled.evaluate_with_fault(&record, &limits).1,
1551            Some(FaultReason::TooDeep)
1552        );
1553        let braces_in_strings = r#"{"a":"{{{{{{{{"}"#;
1554        let record = FilterRecord {
1555            payload: braces_in_strings.as_bytes(),
1556            headers: &[],
1557        };
1558        assert_eq!(compiled.evaluate(&record, &limits), Verdict::Selected);
1559        let large = format!(r#"{{"a":"{}"}}"#, "x".repeat(100));
1560        let record = FilterRecord {
1561            payload: large.as_bytes(),
1562            headers: &[],
1563        };
1564        assert_eq!(
1565            compiled.evaluate_with_fault(&record, &limits).1,
1566            Some(FaultReason::TooLarge)
1567        );
1568    }
1569
1570    #[test]
1571    fn given_a_record_when_explained_then_should_report_every_node() {
1572        let compiled = CompiledFilter::compile(&strict()).expect("compiles");
1573        let explanation = compiled.explain(
1574            &FilterRecord {
1575                payload: SAFE_MODE_VALUES.as_bytes(),
1576                headers: &[],
1577            },
1578            &DecodeLimits::default(),
1579        );
1580        assert_eq!(explanation.verdict, Verdict::Rejected);
1581        assert_eq!(explanation.root.label, "any");
1582        assert_eq!(explanation.root.children.len(), 3);
1583        let snapshot_branch = &explanation.root.children[0];
1584        assert_eq!(snapshot_branch.truth, Some(Truth::Unknown));
1585        assert_eq!(
1586            snapshot_branch.children[2].label,
1587            r#"changed contains "mode""#
1588        );
1589        assert_eq!(snapshot_branch.children[2].truth, Some(Truth::Unknown));
1590        let malformed = compiled.explain(
1591            &FilterRecord {
1592                payload: b"{",
1593                headers: &[],
1594            },
1595            &DecodeLimits::default(),
1596        );
1597        assert_eq!(malformed.verdict, Verdict::Fault);
1598        assert_eq!(malformed.fault, Some(FaultReason::Malformed));
1599    }
1600
1601    #[test]
1602    fn given_duplicate_keys_when_decoded_then_should_keep_the_last_value() {
1603        let filter = ConsumerFilter::json(FilterExpr::pred("kind", CmpOp::Eq, "B"));
1604        assert_eq!(
1605            verdict(&filter, r#"{"kind":"A","kind":"B"}"#),
1606            Verdict::Selected
1607        );
1608    }
1609
1610    fn decided(
1611        filter: &ConsumerFilter,
1612        payload: &[u8],
1613        headers: &[HeaderRef<'_>],
1614    ) -> (Verdict, Option<FaultReason>) {
1615        CompiledFilter::compile(filter)
1616            .expect("compiles")
1617            .evaluate_with_fault(&FilterRecord { payload, headers }, &DecodeLimits::default())
1618    }
1619
1620    fn content_type(code: u64) -> HeaderRef<'static> {
1621        HeaderRef {
1622            key: CONTENT_TYPE,
1623            value: HeaderValueRef::Uint(code),
1624        }
1625    }
1626
1627    #[test]
1628    fn given_a_record_declared_in_another_codec_when_a_json_filter_reads_it_then_should_reject_without_decoding()
1629     {
1630        let filter = ConsumerFilter::json(FilterExpr::pred("type", CmpOp::Eq, "billing"));
1631        let protobuf = content_type(u64::from(ContentType::Protobuf.code()));
1632        assert_eq!(
1633            decided(&filter, b"\x08\x01", &[protobuf]),
1634            (Verdict::Rejected, Some(FaultReason::ForeignCodec))
1635        );
1636        let passing = filter.clone().with_foreign_policy(RecordPolicy::Pass);
1637        assert_eq!(
1638            decided(&passing, b"\x08\x01", &[protobuf]),
1639            (Verdict::Fault, Some(FaultReason::ForeignCodec))
1640        );
1641        let compiled = CompiledFilter::compile(&passing).expect("compiles");
1642        assert_eq!(
1643            compiled.policy_for(FaultReason::ForeignCodec),
1644            FaultPolicy::Pass
1645        );
1646        let json = content_type(u64::from(ContentType::Json.code()));
1647        assert_eq!(
1648            decided(&filter, br#"{"type":"billing"}"#, &[json]),
1649            (Verdict::Selected, None)
1650        );
1651        assert_eq!(
1652            decided(&filter, br#"{"type":"billing"}"#, &[content_type(255)]),
1653            (Verdict::Selected, None)
1654        );
1655        assert_eq!(
1656            decided(&filter, br#"{"type":"billing"}"#, &[content_type(200)]),
1657            (Verdict::Selected, None)
1658        );
1659        assert_eq!(
1660            decided(&filter, b"not json", &[]),
1661            (Verdict::Fault, Some(FaultReason::Malformed))
1662        );
1663    }
1664
1665    #[test]
1666    fn given_a_value_of_another_type_when_compared_then_should_follow_the_mismatch_policy() {
1667        let filter = ConsumerFilter::json(FilterExpr::pred("xyz", CmpOp::Eq, "Abc"));
1668        for payload in [
1669            r#"{"xyz":1}"#,
1670            r#"{"xyz":{"a":1}}"#,
1671            r#"{"xyz":[1]}"#,
1672            r#"{"xyz":true}"#,
1673        ] {
1674            assert_eq!(
1675                decided(&filter, payload.as_bytes(), &[]),
1676                (Verdict::Rejected, Some(FaultReason::TypeMismatch)),
1677                "{payload}"
1678            );
1679        }
1680        let passing = filter.with_mismatch_policy(RecordPolicy::Pass);
1681        assert_eq!(
1682            decided(&passing, br#"{"xyz":1}"#, &[]),
1683            (Verdict::Fault, Some(FaultReason::TypeMismatch))
1684        );
1685        assert_eq!(
1686            decided(&passing, br#"{"xyz":"Abc"}"#, &[]),
1687            (Verdict::Selected, None)
1688        );
1689        assert_eq!(
1690            decided(&passing, br#"{"xyz":"Abd"}"#, &[]),
1691            (Verdict::Rejected, None)
1692        );
1693        assert_eq!(
1694            decided(&passing, br#"{"other":1}"#, &[]),
1695            (Verdict::Rejected, None)
1696        );
1697        assert_eq!(
1698            decided(&passing, br#"{"xyz":null}"#, &[]),
1699            (Verdict::Rejected, None)
1700        );
1701    }
1702
1703    #[test]
1704    fn given_text_predicates_on_a_mixed_log_when_evaluated_then_should_select_the_subdomain() {
1705        let filter = ConsumerFilter::json(FilterExpr::any([
1706            FilterExpr::text("type", TextMatch::Contains, ".v1."),
1707            FilterExpr::text("type", TextMatch::Glob, "SHIPPING.*").case_insensitive(),
1708        ]));
1709        assert_eq!(
1710            verdict(&filter, r#"{"type":"billing.invoice.v1.created"}"#),
1711            Verdict::Selected
1712        );
1713        assert_eq!(
1714            verdict(&filter, r#"{"type":"shipping.parcel.v2.sent"}"#),
1715            Verdict::Selected
1716        );
1717        assert_eq!(
1718            verdict(&filter, r#"{"type":"billing.invoice.v2.created"}"#),
1719            Verdict::Rejected
1720        );
1721        let headers = ConsumerFilter::headers_only(FilterExpr::header_text(
1722            "event.type",
1723            TextMatch::Regex,
1724            r"^billing\.[a-z]+\.v1\.",
1725        ));
1726        let typed = [HeaderRef {
1727            key: "event.type",
1728            value: HeaderValueRef::String("billing.invoice.v1.created"),
1729        }];
1730        assert_eq!(
1731            decided(&headers, b"\xff\xfe anything", &typed),
1732            (Verdict::Selected, None)
1733        );
1734        let numeric = [HeaderRef {
1735            key: "event.type",
1736            value: HeaderValueRef::Uint(7),
1737        }];
1738        assert_eq!(
1739            decided(&headers, b"", &numeric),
1740            (Verdict::Rejected, Some(FaultReason::TypeMismatch))
1741        );
1742        let passing = headers.with_mismatch_policy(RecordPolicy::Pass);
1743        assert_eq!(
1744            decided(&passing, b"", &numeric),
1745            (Verdict::Fault, Some(FaultReason::TypeMismatch))
1746        );
1747    }
1748
1749    #[test]
1750    fn given_escaped_quotes_and_deep_nesting_when_depth_is_checked_then_should_count_only_structure()
1751     {
1752        assert!(json_depth_within(br#"{"a":"[[[[\"{{{{"}"#, 1));
1753        assert!(!json_depth_within(br#"{"a":[[1]]}"#, 2));
1754        assert!(json_depth_within(br#"{"a":"\\"}"#, 1));
1755        assert!(json_depth_within(br#"{"a":"unterminated"#, 1));
1756    }
1757}