use super::codecs::PayloadDecoders;
use crate::content::ContentType;
use crate::control::SchemaDef;
use crate::error::InvalidError;
use crate::filter::coerce::{ExactDecimal, TimestampFormat};
use crate::filter::expr::{
Coerce, ConsumerFilter, FaultPolicy, FilterCodec, FilterExpr, RecordPolicy,
};
use crate::filter::path::{FieldPath, PathSegment};
use crate::filter::read::{
ExplainNode, FaultReason, FilterExplanation, FilterHeader, HeaderScalar, Truth, Verdict,
};
use crate::filter::text::TextMatcher;
use crate::headers::CONTENT_TYPE;
use crate::limits::MAX_FILTER_PARSE_DEPTH;
use crate::query::CmpOp;
use crate::schema::{Digest32, TypedValue};
use crate::validate::Validate;
use serde::de::{DeserializeSeed, MapAccess, SeqAccess, Visitor};
use serde::{Deserialize, Deserializer};
use serde_json::{Map, Value};
use std::borrow::Cow;
use std::cmp::Ordering;
use std::fmt;
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct DecodeLimits {
pub max_payload_bytes: usize,
pub max_depth: usize,
}
impl Default for DecodeLimits {
fn default() -> Self {
Self {
max_payload_bytes: 1024 * 1024,
max_depth: 64,
}
}
}
#[derive(Clone, Copy, Debug)]
pub struct FilterRecord<'a> {
pub payload: &'a [u8],
pub headers: &'a [HeaderRef<'a>],
}
#[derive(Clone, Copy, Debug, PartialEq)]
pub struct HeaderRef<'a> {
pub key: &'a str,
pub value: HeaderValueRef<'a>,
}
#[derive(Clone, Copy, Debug, PartialEq)]
pub enum HeaderValueRef<'a> {
Bool(bool),
Int(i64),
Uint(u64),
Float(f64),
String(&'a str),
Raw(&'a [u8]),
}
impl<'a> From<&'a FilterHeader> for HeaderRef<'a> {
fn from(header: &'a FilterHeader) -> Self {
Self {
key: &header.key,
value: match &header.value {
HeaderScalar::Bool(value) => HeaderValueRef::Bool(*value),
HeaderScalar::Int(value) => HeaderValueRef::Int(*value),
HeaderScalar::Uint(value) => HeaderValueRef::Uint(*value),
HeaderScalar::Float(value) => HeaderValueRef::Float(*value),
HeaderScalar::String(value) => HeaderValueRef::String(value),
HeaderScalar::Raw(bytes) => HeaderValueRef::Raw(bytes),
},
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum HeaderNeed {
None,
ContentType,
All,
}
#[derive(Clone, Debug)]
pub struct CompiledFilter {
filter: ConsumerFilter,
digest: Digest32,
root: Node,
paths: PathTrie,
reads_headers: bool,
decoders: PayloadDecoders,
}
#[derive(Clone, Debug, Default)]
pub(super) struct PathTrie {
pub(super) whole: bool,
indices: Vec<(u32, PathTrie)>,
keys: Vec<(String, PathTrie)>,
}
#[derive(Clone, Debug)]
struct Node {
kind: NodeKind,
reads_payload: bool,
}
#[derive(Clone, Debug)]
enum NodeKind {
All(Vec<Node>),
Any(Vec<Node>),
Not(Box<Node>),
Compare {
path: FieldPath,
op: CmpOp,
literal: Literal,
},
Coerced {
path: FieldPath,
op: CmpOp,
coerce: Coerce,
literal: CoercedLiteral,
},
Present(FieldPath),
Absent(FieldPath),
Header {
key: String,
op: CmpOp,
literal: Literal,
},
Text {
path: FieldPath,
matcher: TextMatcher,
},
HeaderText {
key: String,
matcher: TextMatcher,
},
}
#[derive(Clone, Debug, PartialEq)]
enum Literal {
Null,
Bool(bool),
Int(i128),
Float(f64),
Text(String),
List(Vec<Literal>),
}
#[derive(Clone, Debug, PartialEq)]
enum CoercedLiteral {
Instant(i64),
Decimal(ExactDecimal),
List(Vec<CoercedLiteral>),
}
#[derive(Clone, Debug, PartialEq)]
enum Comparable {
Instant(i64),
Decimal(ExactDecimal),
}
enum Scalar<'a> {
Null,
Bool(bool),
Int(i128),
Float(f64),
Text(&'a str),
Composite,
}
impl CompiledFilter {
pub fn compile(filter: &ConsumerFilter) -> Result<Self, InvalidError> {
Self::compile_with_schemas(filter, &[])
}
pub fn compile_with_schemas(
filter: &ConsumerFilter,
schemas: &[SchemaDef],
) -> Result<Self, InvalidError> {
filter.validate()?;
let decoders = PayloadDecoders::compile(filter, schemas)?;
let mut paths = PathTrie::default();
collect_paths(&filter.expr, &mut paths)?;
Ok(Self {
filter: filter.clone(),
digest: filter.digest(),
root: compile_node(&filter.expr)?,
paths,
reads_headers: filter.expr.reads_headers() || !filter.schema_refs.is_empty(),
decoders,
})
}
pub fn digest(&self) -> &Digest32 {
&self.digest
}
pub fn filter(&self) -> &ConsumerFilter {
&self.filter
}
pub fn fault_policy(&self) -> FaultPolicy {
self.filter.fault_policy
}
pub fn policy_for(&self, reason: FaultReason) -> FaultPolicy {
match self.record_policy(reason) {
Some(RecordPolicy::Reject) => FaultPolicy::Drop,
Some(RecordPolicy::Pass) => FaultPolicy::Pass,
None => self.filter.fault_policy,
}
}
pub fn record_policy(&self, reason: FaultReason) -> Option<RecordPolicy> {
if reason.is_foreign() {
Some(self.filter.foreign_policy)
} else if reason == FaultReason::TypeMismatch {
Some(self.filter.mismatch_policy)
} else {
None
}
}
pub fn reads_payload(&self) -> bool {
self.root.reads_payload
}
pub fn reads_headers(&self) -> bool {
self.reads_headers
}
pub fn header_need(&self) -> HeaderNeed {
if self.reads_headers {
HeaderNeed::All
} else if self.root.reads_payload {
HeaderNeed::ContentType
} else {
HeaderNeed::None
}
}
pub fn evaluate(&self, record: &FilterRecord<'_>, limits: &DecodeLimits) -> Verdict {
self.evaluate_with_fault(record, limits).0
}
pub fn evaluate_with_fault(
&self,
record: &FilterRecord<'_>,
limits: &DecodeLimits,
) -> (Verdict, Option<FaultReason>) {
let mut context = Context::new(
record,
limits,
self.filter.codec,
&self.paths,
&self.decoders,
);
let truth = context.truth(&self.root);
self.evaluated(truth, context.mismatched)
}
pub fn explain(&self, record: &FilterRecord<'_>, limits: &DecodeLimits) -> FilterExplanation {
let mut context = Context::new(
record,
limits,
self.filter.codec,
&self.paths,
&self.decoders,
);
let truth = context.truth(&self.root);
let (verdict, fault) = self.evaluated(truth, context.mismatched);
let root = context.explain(&self.root);
FilterExplanation {
verdict,
fault,
root,
}
}
}
impl CompiledFilter {
fn evaluated(
&self,
truth: Result<Truth, FaultReason>,
mismatched: bool,
) -> (Verdict, Option<FaultReason>) {
let reason = match truth {
Ok(Truth::Match) => return (Verdict::Selected, None),
Ok(Truth::NoMatch) => return (Verdict::Rejected, None),
Ok(Truth::Unknown) if !mismatched => return (Verdict::Rejected, None),
Ok(Truth::Unknown) => FaultReason::TypeMismatch,
Err(reason) => reason,
};
match self.record_policy(reason) {
Some(RecordPolicy::Reject) => (Verdict::Rejected, Some(reason)),
Some(RecordPolicy::Pass) | None => (Verdict::Fault, Some(reason)),
}
}
}
fn compile_node(expr: &FilterExpr) -> Result<Node, InvalidError> {
let kind = match expr {
FilterExpr::All(children) => NodeKind::All(
children
.iter()
.map(compile_node)
.collect::<Result<_, _>>()?,
),
FilterExpr::Any(children) => NodeKind::Any(
children
.iter()
.map(compile_node)
.collect::<Result<_, _>>()?,
),
FilterExpr::Not(child) => NodeKind::Not(Box::new(compile_node(child)?)),
FilterExpr::Pred(predicate) => NodeKind::Compare {
path: FieldPath::parse(&predicate.field)?,
op: predicate.op,
literal: compile_literal(&predicate.value)?,
},
FilterExpr::PredAs(coerced) => NodeKind::Coerced {
path: FieldPath::parse(&coerced.pred.field)?,
op: coerced.pred.op,
coerce: coerced.coerce,
literal: compile_coerced_literal(coerced.coerce, &coerced.pred.value)?,
},
FilterExpr::Present(path) => NodeKind::Present(path.clone()),
FilterExpr::Absent(path) => NodeKind::Absent(path.clone()),
FilterExpr::Header(header) => NodeKind::Header {
key: header.key.clone(),
op: header.op,
literal: compile_literal(&header.value)?,
},
FilterExpr::Text(predicate) => NodeKind::Text {
path: FieldPath::parse(&predicate.field)?,
matcher: TextMatcher::compile(predicate)?,
},
FilterExpr::HeaderText(predicate) => NodeKind::HeaderText {
key: predicate.field.clone(),
matcher: TextMatcher::compile(predicate)?,
},
};
Ok(Node {
kind,
reads_payload: expr.reads_payload(),
})
}
fn compile_literal(value: &TypedValue) -> Result<Literal, InvalidError> {
Ok(match value {
TypedValue::Null => Literal::Null,
TypedValue::Boolean(value) => Literal::Bool(*value),
TypedValue::Int(value) => Literal::Int(i128::from(*value)),
TypedValue::Long(value) => Literal::Int(i128::from(*value)),
TypedValue::Double(value) => Literal::Float(*value),
TypedValue::String(value) => Literal::Text(value.clone()),
TypedValue::List(items) => Literal::List(
items
.iter()
.map(compile_literal)
.collect::<Result<_, _>>()?,
),
other => {
return Err(InvalidError::new(format!(
"unsupported filter literal {}",
other.diagnostic_text()
)));
}
})
}
fn compile_coerced_literal(
coerce: Coerce,
value: &TypedValue,
) -> Result<CoercedLiteral, InvalidError> {
if let TypedValue::List(items) = value {
return items
.iter()
.map(|item| compile_coerced_literal(coerce, item))
.collect::<Result<_, _>>()
.map(CoercedLiteral::List);
}
let comparable = match (coerce, value) {
(Coerce::Timestamp { format }, TypedValue::String(text)) => {
format.micros_from_text(text).map(Comparable::Instant)
}
(Coerce::Timestamp { format }, TypedValue::Int(number)) => format
.micros_from_integer(i128::from(*number))
.map(Comparable::Instant),
(Coerce::Timestamp { format }, TypedValue::Long(number)) => format
.micros_from_integer(i128::from(*number))
.map(Comparable::Instant),
(Coerce::Number, TypedValue::String(text)) => {
ExactDecimal::parse(text).map(Comparable::Decimal)
}
(Coerce::Number, TypedValue::Int(number)) => Some(Comparable::Decimal(
ExactDecimal::from_integer(i128::from(*number)),
)),
(Coerce::Number, TypedValue::Long(number)) => Some(Comparable::Decimal(
ExactDecimal::from_integer(i128::from(*number)),
)),
_ => None,
};
match comparable {
Some(Comparable::Instant(micros)) => Ok(CoercedLiteral::Instant(micros)),
Some(Comparable::Decimal(decimal)) => Ok(CoercedLiteral::Decimal(decimal)),
None => Err(InvalidError::new(format!(
"literal {} does not parse under its coercion",
value.diagnostic_text()
))),
}
}
struct Context<'a> {
record: &'a FilterRecord<'a>,
limits: &'a DecodeLimits,
codec: FilterCodec,
paths: &'a PathTrie,
decoders: &'a PayloadDecoders,
payload: Option<Result<Value, FaultReason>>,
mismatched: bool,
}
impl<'a> Context<'a> {
fn new(
record: &'a FilterRecord<'a>,
limits: &'a DecodeLimits,
codec: FilterCodec,
paths: &'a PathTrie,
decoders: &'a PayloadDecoders,
) -> Self {
Self {
record,
limits,
codec,
paths,
decoders,
payload: None,
mismatched: false,
}
}
fn payload(&mut self) -> Result<&Value, FaultReason> {
if self.payload.is_none() {
let decoded = if self.declares_foreign_codec() {
Err(FaultReason::ForeignCodec)
} else {
self.decode()
};
self.payload = Some(decoded);
}
match self.payload.as_ref().expect("the payload was just decoded") {
Ok(value) => Ok(value),
Err(reason) => Err(*reason),
}
}
fn declares_foreign_codec(&self) -> bool {
let Some(header) = self.header(CONTENT_TYPE) else {
return false;
};
let code = match header.value {
HeaderValueRef::Uint(code) => u8::try_from(code).ok(),
HeaderValueRef::Int(code) => u8::try_from(code).ok(),
_ => None,
};
let declared = match code.and_then(ContentType::from_code) {
None | Some(ContentType::Any) => return false,
Some(declared) => declared,
};
let expected = match self.codec {
FilterCodec::Json => ContentType::Json,
FilterCodec::Cbor => ContentType::Cbor,
FilterCodec::Avro => ContentType::Avro,
FilterCodec::Protobuf => ContentType::Protobuf,
FilterCodec::HeadersOnly | FilterCodec::Unknown => return false,
};
declared != expected
}
fn decode(&self) -> Result<Value, FaultReason> {
match self.codec {
FilterCodec::Json => decode_json_paths(self.record.payload, self.limits, self.paths),
FilterCodec::Cbor | FilterCodec::Avro | FilterCodec::Protobuf => self
.decoders
.decode_paths(self.codec, self.record, self.limits, self.paths),
FilterCodec::HeadersOnly | FilterCodec::Unknown => Err(FaultReason::Malformed),
}
}
fn truth(&mut self, node: &Node) -> Result<Truth, FaultReason> {
match &node.kind {
NodeKind::All(children) => self.combine(children, Truth::NoMatch, Truth::Match),
NodeKind::Any(children) => self.combine(children, Truth::Match, Truth::NoMatch),
NodeKind::Not(child) => self.truth(child).map(negate),
NodeKind::Header { key, op, literal } => {
let (truth, mismatched) = self.header_truth(key, *op, literal);
self.mismatched |= mismatched;
Ok(truth)
}
NodeKind::HeaderText { key, matcher } => {
let (truth, mismatched) = self.header_text_truth(key, matcher);
self.mismatched |= mismatched;
Ok(truth)
}
leaf => {
let value = self.payload()?;
let (truth, mismatched) = payload_truth(leaf, value);
self.mismatched |= mismatched;
Ok(truth)
}
}
}
fn combine(
&mut self,
children: &[Node],
decisive: Truth,
otherwise: Truth,
) -> Result<Truth, FaultReason> {
let mut unknown = false;
let header_first = children
.iter()
.filter(|child| !child.reads_payload)
.chain(children.iter().filter(|child| child.reads_payload));
for child in header_first {
match self.truth(child)? {
truth if truth == decisive => return Ok(decisive),
Truth::Unknown => unknown = true,
_ => {}
}
}
Ok(if unknown { Truth::Unknown } else { otherwise })
}
fn header_truth(&self, key: &str, op: CmpOp, literal: &Literal) -> (Truth, bool) {
let Some(header) = self.header(key) else {
return (Truth::Unknown, false);
};
let scalar = header_scalar(header.value);
let truth = compare_scalar(&scalar, None, op, literal);
(
truth,
truth == Truth::Unknown && !comparable(&scalar, None, op, literal),
)
}
fn header_text_truth(&self, key: &str, matcher: &TextMatcher) -> (Truth, bool) {
let Some(header) = self.header(key) else {
return (Truth::Unknown, false);
};
match header_scalar(header.value) {
Scalar::Text(text) => (truth_of(matcher.matches(text)), false),
_ => (Truth::Unknown, true),
}
}
fn header(&self, key: &str) -> Option<&HeaderRef<'a>> {
self.record
.headers
.iter()
.rev()
.find(|header| header.key == key)
}
fn explain(&mut self, node: &Node) -> ExplainNode {
let (label, children) = match &node.kind {
NodeKind::All(children) => ("all".to_owned(), children.as_slice()),
NodeKind::Any(children) => ("any".to_owned(), children.as_slice()),
NodeKind::Not(child) => ("not".to_owned(), std::slice::from_ref(child.as_ref())),
leaf => {
let truth = match leaf {
NodeKind::Header { key, op, literal } => {
Some(self.header_truth(key, *op, literal).0)
}
NodeKind::HeaderText { key, matcher } => {
Some(self.header_text_truth(key, matcher).0)
}
_ => self
.payload()
.ok()
.map(|value| payload_truth(leaf, value).0),
};
return ExplainNode {
label: leaf_label(leaf),
truth,
children: Vec::new(),
};
}
};
let explained: Vec<ExplainNode> =
children.iter().map(|child| self.explain(child)).collect();
let truths: Option<Vec<Truth>> = explained.iter().map(|child| child.truth).collect();
let truth = truths.map(|truths| match &node.kind {
NodeKind::All(_) => fold(&truths, Truth::NoMatch, Truth::Match),
NodeKind::Any(_) => fold(&truths, Truth::Match, Truth::NoMatch),
_ => negate(truths[0]),
});
ExplainNode {
label,
truth,
children: explained,
}
}
}
fn fold(truths: &[Truth], decisive: Truth, otherwise: Truth) -> Truth {
if truths.contains(&decisive) {
decisive
} else if truths.contains(&Truth::Unknown) {
Truth::Unknown
} else {
otherwise
}
}
const fn negate(truth: Truth) -> Truth {
match truth {
Truth::Match => Truth::NoMatch,
Truth::NoMatch => Truth::Match,
Truth::Unknown => Truth::Unknown,
}
}
fn payload_truth(leaf: &NodeKind, value: &Value) -> (Truth, bool) {
match leaf {
NodeKind::Compare { path, op, literal } => match resolve(value, path) {
None => (missing_truth(*op, literal), false),
Some(field) => {
let scalar = scalar_of(field);
let truth = compare_scalar(&scalar, Some(field), *op, literal);
let mismatched =
truth == Truth::Unknown && !comparable(&scalar, Some(field), *op, literal);
(truth, mismatched)
}
},
NodeKind::Coerced {
path,
op,
coerce,
literal,
} => match resolve(value, path) {
None => (Truth::Unknown, false),
Some(field) => match coerce_value(*coerce, field) {
Some(comparable) => (compare_coerced(&comparable, *op, literal), false),
None => (Truth::Unknown, !field.is_null()),
},
},
NodeKind::Text { path, matcher } => match resolve(value, path) {
None | Some(Value::Null) => (Truth::Unknown, false),
Some(Value::String(text)) => (truth_of(matcher.matches(text)), false),
Some(_) => (Truth::Unknown, true),
},
NodeKind::Present(path) => (truth_of(resolve(value, path).is_some()), false),
NodeKind::Absent(path) => (truth_of(resolve(value, path).is_none()), false),
NodeKind::All(_)
| NodeKind::Any(_)
| NodeKind::Not(_)
| NodeKind::Header { .. }
| NodeKind::HeaderText { .. } => unreachable!("only payload leaves read the payload"),
}
}
fn header_scalar(value: HeaderValueRef<'_>) -> Scalar<'_> {
match value {
HeaderValueRef::Bool(value) => Scalar::Bool(value),
HeaderValueRef::Int(value) => Scalar::Int(i128::from(value)),
HeaderValueRef::Uint(value) => Scalar::Int(i128::from(value)),
HeaderValueRef::Float(value) => Scalar::Float(value),
HeaderValueRef::String(value) => Scalar::Text(value),
HeaderValueRef::Raw(bytes) => match std::str::from_utf8(bytes) {
Ok(text) => Scalar::Text(text),
Err(_) => Scalar::Composite,
},
}
}
const fn truth_of(matched: bool) -> Truth {
if matched {
Truth::Match
} else {
Truth::NoMatch
}
}
fn comparable(scalar: &Scalar<'_>, field: Option<&Value>, op: CmpOp, literal: &Literal) -> bool {
match (scalar, literal) {
(Scalar::Null, _) | (_, Literal::Null) => true,
(Scalar::Text(_), Literal::Text(_)) | (Scalar::Bool(_), Literal::Bool(_)) => true,
(Scalar::Int(_) | Scalar::Float(_), Literal::Int(_) | Literal::Float(_)) => true,
(scalar, Literal::List(items)) => {
items.iter().any(|item| comparable(scalar, field, op, item))
}
(Scalar::Composite, _) => op == CmpOp::Contains && matches!(field, Some(Value::Array(_))),
_ => false,
}
}
fn missing_truth(op: CmpOp, literal: &Literal) -> Truth {
match (op, literal) {
(CmpOp::Eq, Literal::Null) => Truth::Match,
(CmpOp::Ne, Literal::Null) => Truth::NoMatch,
_ => Truth::Unknown,
}
}
fn compare_scalar(
scalar: &Scalar<'_>,
field: Option<&Value>,
op: CmpOp,
literal: &Literal,
) -> Truth {
match (scalar, literal) {
(Scalar::Null, Literal::Null) => return truth_of(op == CmpOp::Eq),
(Scalar::Null, _) => return Truth::Unknown,
(_, Literal::Null) => return truth_of(op == CmpOp::Ne),
_ => {}
}
match op {
CmpOp::Eq => scalar_eq(scalar, literal).map_or(Truth::Unknown, truth_of),
CmpOp::Ne => scalar_eq(scalar, literal).map_or(Truth::Unknown, |equal| truth_of(!equal)),
CmpOp::Lt | CmpOp::Lte | CmpOp::Gt | CmpOp::Gte => match scalar_cmp(scalar, literal) {
Some(ordering) => truth_of(match op {
CmpOp::Lt => ordering == Ordering::Less,
CmpOp::Lte => ordering != Ordering::Greater,
CmpOp::Gt => ordering == Ordering::Greater,
_ => ordering != Ordering::Less,
}),
None => Truth::Unknown,
},
CmpOp::In => {
let Literal::List(items) = literal else {
return Truth::Unknown;
};
let mut unknown = false;
for item in items {
match scalar_eq(scalar, item) {
Some(true) => return Truth::Match,
Some(false) => {}
None => unknown = true,
}
}
if unknown {
Truth::Unknown
} else {
Truth::NoMatch
}
}
CmpOp::Contains => match (scalar, literal, field) {
(Scalar::Text(text), Literal::Text(needle), _) => {
truth_of(text.contains(needle.as_str()))
}
(Scalar::Composite, _, Some(Value::Array(elements))) => truth_of(
elements
.iter()
.any(|element| scalar_eq(&scalar_of(element), literal) == Some(true)),
),
_ => Truth::Unknown,
},
CmpOp::Prefix => match (scalar, literal) {
(Scalar::Text(text), Literal::Text(prefix)) => {
truth_of(text.starts_with(prefix.as_str()))
}
_ => Truth::Unknown,
},
}
}
fn scalar_eq(scalar: &Scalar<'_>, literal: &Literal) -> Option<bool> {
match (scalar, literal) {
(Scalar::Bool(left), Literal::Bool(right)) => Some(left == right),
(Scalar::Text(left), Literal::Text(right)) => Some(*left == right.as_str()),
_ => scalar_cmp(scalar, literal).map(|ordering| ordering == Ordering::Equal),
}
}
fn scalar_cmp(scalar: &Scalar<'_>, literal: &Literal) -> Option<Ordering> {
match (scalar, literal) {
(Scalar::Int(left), Literal::Int(right)) => Some(left.cmp(right)),
(Scalar::Float(left), Literal::Float(right)) => left.partial_cmp(right),
(Scalar::Int(left), Literal::Float(right)) => exact_double(*left)?.partial_cmp(right),
(Scalar::Float(left), Literal::Int(right)) => left.partial_cmp(&exact_double(*right)?),
(Scalar::Text(left), Literal::Text(right)) => Some(left.as_bytes().cmp(right.as_bytes())),
_ => None,
}
}
#[allow(clippy::cast_precision_loss)]
fn exact_double(value: i128) -> Option<f64> {
let magnitude = value.unsigned_abs();
let significant_bits = u128::BITS - magnitude.leading_zeros();
(significant_bits <= 53 || magnitude.trailing_zeros() >= significant_bits - 53)
.then_some(value as f64)
}
fn compare_coerced(comparable: &Comparable, op: CmpOp, literal: &CoercedLiteral) -> Truth {
let ordering = |literal: &CoercedLiteral| match (comparable, literal) {
(Comparable::Instant(left), CoercedLiteral::Instant(right)) => Some(left.cmp(right)),
(Comparable::Decimal(left), CoercedLiteral::Decimal(right)) => Some(left.cmp(right)),
_ => None,
};
match (op, literal) {
(CmpOp::In, CoercedLiteral::List(items)) => truth_of(
items
.iter()
.any(|item| ordering(item) == Some(Ordering::Equal)),
),
(_, literal) => match ordering(literal) {
Some(ordering) => truth_of(match op {
CmpOp::Eq => ordering == Ordering::Equal,
CmpOp::Ne => ordering != Ordering::Equal,
CmpOp::Lt => ordering == Ordering::Less,
CmpOp::Lte => ordering != Ordering::Greater,
CmpOp::Gt => ordering == Ordering::Greater,
CmpOp::Gte => ordering != Ordering::Less,
CmpOp::In | CmpOp::Contains | CmpOp::Prefix => return Truth::Unknown,
}),
None => Truth::Unknown,
},
}
}
fn coerce_value(coerce: Coerce, field: &Value) -> Option<Comparable> {
match coerce {
Coerce::Timestamp { format } => {
let micros = match (format, field) {
(_, Value::String(text)) => format.micros_from_text(text),
(TimestampFormat::Rfc3339, _) => None,
(_, Value::Number(number)) => {
let integer = number
.as_i64()
.map(i128::from)
.or_else(|| number.as_u64().map(i128::from))?;
format.micros_from_integer(integer)
}
_ => None,
}?;
Some(Comparable::Instant(micros))
}
Coerce::Number => {
let decimal = match field {
Value::String(text) => ExactDecimal::parse(text),
Value::Number(number) => number
.as_i64()
.map(|value| ExactDecimal::from_integer(i128::from(value)))
.or_else(|| {
number
.as_u64()
.map(|value| ExactDecimal::from_integer(i128::from(value)))
})
.or_else(|| number.as_f64().and_then(ExactDecimal::from_f64)),
_ => None,
}?;
Some(Comparable::Decimal(decimal))
}
}
}
fn scalar_of(value: &Value) -> Scalar<'_> {
match value {
Value::Null => Scalar::Null,
Value::Bool(value) => Scalar::Bool(*value),
Value::Number(number) => number
.as_i64()
.map(|value| Scalar::Int(i128::from(value)))
.or_else(|| number.as_u64().map(|value| Scalar::Int(i128::from(value))))
.or_else(|| number.as_f64().map(Scalar::Float))
.unwrap_or(Scalar::Composite),
Value::String(text) => Scalar::Text(text),
Value::Array(_) | Value::Object(_) => Scalar::Composite,
}
}
fn resolve<'v>(value: &'v Value, path: &FieldPath) -> Option<&'v Value> {
path.segments()
.iter()
.try_fold(value, |current, segment| match (segment, current) {
(PathSegment::Key(key), Value::Object(map)) => map.get(key),
(PathSegment::Index(index), Value::Array(items)) => items.get(*index as usize),
_ => None,
})
}
pub fn decode_json(payload: &[u8], limits: &DecodeLimits) -> Result<Value, FaultReason> {
decode_json_paths(payload, limits, &PathTrie::whole())
}
fn decode_json_paths(
payload: &[u8],
limits: &DecodeLimits,
paths: &PathTrie,
) -> Result<Value, FaultReason> {
if payload.len() > limits.max_payload_bytes {
return Err(FaultReason::TooLarge);
}
if !json_depth_within(payload, limits.max_depth.min(MAX_FILTER_PARSE_DEPTH)) {
return Err(FaultReason::TooDeep);
}
let mut deserializer = serde_json::Deserializer::from_slice(payload);
let value = Pruned(paths)
.deserialize(&mut deserializer)
.map_err(|_| FaultReason::Malformed)?;
deserializer.end().map_err(|_| FaultReason::Malformed)?;
Ok(value)
}
impl PathTrie {
pub(super) fn whole() -> Self {
Self {
whole: true,
..Self::default()
}
}
pub(super) fn child(&self, key: &str) -> Option<&PathTrie> {
self.keys
.iter()
.find_map(|(name, child)| (name == key).then_some(child))
}
fn child_mut(&mut self, key: &str) -> &mut PathTrie {
let index = match self.keys.iter().position(|(name, _)| name == key) {
Some(index) => index,
None => {
self.keys.push((key.to_owned(), PathTrie::default()));
self.keys.len() - 1
}
};
&mut self.keys[index].1
}
pub(super) fn index(&self, index: u32) -> Option<&PathTrie> {
self.indices
.iter()
.find_map(|(number, child)| (*number == index).then_some(child))
}
fn index_mut(&mut self, index: u32) -> &mut PathTrie {
let position = match self.indices.iter().position(|(number, _)| *number == index) {
Some(position) => position,
None => {
self.indices.push((index, Self::default()));
self.indices.len() - 1
}
};
&mut self.indices[position].1
}
fn insert(&mut self, path: &FieldPath) {
let mut node = self;
for segment in path.segments() {
if node.whole {
return;
}
match segment {
PathSegment::Key(key) => node = node.child_mut(key),
PathSegment::Index(index) => node = node.index_mut(*index),
}
}
node.whole = true;
node.keys.clear();
node.indices.clear();
}
}
fn collect_paths(expr: &FilterExpr, paths: &mut PathTrie) -> Result<(), InvalidError> {
match expr {
FilterExpr::All(children) | FilterExpr::Any(children) => {
for child in children {
collect_paths(child, paths)?;
}
}
FilterExpr::Not(child) => collect_paths(child, paths)?,
FilterExpr::Pred(predicate) => paths.insert(&FieldPath::parse(&predicate.field)?),
FilterExpr::PredAs(coerced) => paths.insert(&FieldPath::parse(&coerced.pred.field)?),
FilterExpr::Present(path) | FilterExpr::Absent(path) => paths.insert(path),
FilterExpr::Text(predicate) => paths.insert(&FieldPath::parse(&predicate.field)?),
FilterExpr::Header(_) | FilterExpr::HeaderText(_) => {}
}
Ok(())
}
struct Pruned<'t>(&'t PathTrie);
impl<'de> DeserializeSeed<'de> for Pruned<'_> {
type Value = Value;
fn deserialize<D: Deserializer<'de>>(self, deserializer: D) -> Result<Value, D::Error> {
if self.0.whole {
return Value::deserialize(deserializer);
}
deserializer.deserialize_any(self)
}
}
impl<'de> Visitor<'de> for Pruned<'_> {
type Value = Value;
fn expecting(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str("a JSON value")
}
fn visit_bool<E>(self, value: bool) -> Result<Value, E> {
Ok(Value::Bool(value))
}
fn visit_i64<E>(self, value: i64) -> Result<Value, E> {
Ok(Value::from(value))
}
fn visit_u64<E>(self, value: u64) -> Result<Value, E> {
Ok(Value::from(value))
}
fn visit_f64<E>(self, value: f64) -> Result<Value, E> {
Ok(serde_json::Number::from_f64(value).map_or(Value::Null, Value::Number))
}
fn visit_str<E>(self, _value: &str) -> Result<Value, E> {
Ok(Value::Null)
}
fn visit_unit<E>(self) -> Result<Value, E> {
Ok(Value::Null)
}
fn visit_seq<A: SeqAccess<'de>>(self, mut seq: A) -> Result<Value, A::Error> {
let mut items = Vec::new();
let ignored = PathTrie::default();
let mut index = 0_u32;
while let Some(item) =
seq.next_element_seed(Pruned(self.0.index(index).unwrap_or(&ignored)))?
{
if self.0.index(index).is_some() {
items.resize(index as usize, Value::Null);
items.push(item);
}
index = index.saturating_add(1);
}
Ok(Value::Array(items))
}
fn visit_map<A: MapAccess<'de>>(self, mut map: A) -> Result<Value, A::Error> {
let mut object = Map::new();
let ignored = PathTrie::default();
while let Some(JsonKey(key)) = map.next_key::<JsonKey<'_>>()? {
match self.0.child(key.as_ref()) {
Some(child) => {
let value = map.next_value_seed(Pruned(child))?;
object.insert(key.into_owned(), value);
}
None => {
map.next_value_seed(Pruned(&ignored))?;
}
}
}
Ok(Value::Object(object))
}
}
struct JsonKey<'a>(Cow<'a, str>);
impl<'de> Deserialize<'de> for JsonKey<'de> {
fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
struct KeyVisitor;
impl<'de> Visitor<'de> for KeyVisitor {
type Value = JsonKey<'de>;
fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str("an object key")
}
fn visit_borrowed_str<E>(self, value: &'de str) -> Result<Self::Value, E> {
Ok(JsonKey(Cow::Borrowed(value)))
}
fn visit_str<E>(self, value: &str) -> Result<Self::Value, E> {
Ok(JsonKey(Cow::Owned(value.to_owned())))
}
fn visit_string<E>(self, value: String) -> Result<Self::Value, E> {
Ok(JsonKey(Cow::Owned(value)))
}
}
deserializer.deserialize_str(KeyVisitor)
}
}
fn json_depth_within(payload: &[u8], max_depth: usize) -> bool {
let mut depth = 0usize;
let mut index = 0;
while index < payload.len() {
match payload[index] {
b'"' => {
index += 1;
loop {
let Some(found) = memchr::memchr2(b'"', b'\\', &payload[index..]) else {
return true;
};
index += found;
if payload[index] == b'"' {
break;
}
index += 2;
if index >= payload.len() {
return true;
}
}
}
b'[' | b'{' => {
depth += 1;
if depth > max_depth {
return false;
}
}
b']' | b'}' => depth = depth.saturating_sub(1),
_ => {}
}
index += 1;
}
true
}
fn leaf_label(leaf: &NodeKind) -> String {
match leaf {
NodeKind::Compare { path, op, literal } => {
format!("{path} {op} {}", literal_label(literal))
}
NodeKind::Coerced {
path,
op,
coerce,
literal,
} => {
let coerce = match coerce {
Coerce::Timestamp { format } => format!("timestamp({format})"),
Coerce::Number => "number".to_owned(),
};
format!("{coerce}({path}) {op} {}", coerced_label(literal))
}
NodeKind::Present(path) => format!("{path} is present"),
NodeKind::Absent(path) => format!("{path} is absent"),
NodeKind::Header { key, op, literal } => {
format!("header {key} {op} {}", literal_label(literal))
}
NodeKind::Text { path, matcher } => {
format!("{path} {} {}", matcher.kind(), matcher.label())
}
NodeKind::HeaderText { key, matcher } => {
format!("header {key} {} {}", matcher.kind(), matcher.label())
}
NodeKind::All(_) | NodeKind::Any(_) | NodeKind::Not(_) => unreachable!("not a leaf"),
}
}
fn literal_label(literal: &Literal) -> String {
match literal {
Literal::Null => "null".to_owned(),
Literal::Bool(value) => value.to_string(),
Literal::Int(value) => value.to_string(),
Literal::Float(value) => value.to_string(),
Literal::Text(value) => format!("{value:?}"),
Literal::List(items) => format!(
"[{}]",
items
.iter()
.map(literal_label)
.collect::<Vec<_>>()
.join(", ")
),
}
}
fn coerced_label(literal: &CoercedLiteral) -> String {
match literal {
CoercedLiteral::Instant(micros) => format!("{micros}us"),
CoercedLiteral::Decimal(decimal) => format!("{decimal:?}"),
CoercedLiteral::List(items) => format!(
"[{}]",
items
.iter()
.map(coerced_label)
.collect::<Vec<_>>()
.join(", ")
),
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::filter::text::TextMatch;
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}}"#;
const SAFE_MODE_VALUES: &str = r#"{"op":"u","table":"satellites","after":{"id":"sat-042","name":"Kestrel-42","mode":"safe","orbit":"leo","battery_pct":61}}"#;
const DECOMMISSION: &str = r#"{"op":"d","table":"satellites","before":{"id":"sat-042"}}"#;
const TELEMETRY: &str = r#"{"event":"satellite.telemetry_changed","satellite_id":"sat-042","fields":{"mode":null}}"#;
const GROUND_STATION: &str = r#"{"op":"u","table":"ground_stations","changed":["status"],"after":{"id":"svalbard","status":"online"}}"#;
fn verdict(filter: &ConsumerFilter, payload: &str) -> Verdict {
CompiledFilter::compile(filter).expect("compiles").evaluate(
&FilterRecord {
payload: payload.as_bytes(),
headers: &[],
},
&DecodeLimits::default(),
)
}
fn strict() -> ConsumerFilter {
ConsumerFilter::json(FilterExpr::any([
FilterExpr::all([
FilterExpr::pred("table", CmpOp::Eq, "satellites"),
FilterExpr::pred("op", CmpOp::Eq, "u"),
FilterExpr::pred("changed", CmpOp::Contains, "mode"),
FilterExpr::pred("after.mode", CmpOp::Eq, "safe"),
]),
FilterExpr::all([
FilterExpr::pred("table", CmpOp::Eq, "satellites"),
FilterExpr::pred("op", CmpOp::Eq, "d"),
]),
FilterExpr::all([
FilterExpr::pred("event", CmpOp::Eq, "satellite.telemetry_changed"),
FilterExpr::present("fields.mode"),
]),
]))
}
fn legacy() -> ConsumerFilter {
ConsumerFilter::json(FilterExpr::all([
FilterExpr::pred("table", CmpOp::Eq, "satellites"),
FilterExpr::pred("op", CmpOp::Eq, "u"),
FilterExpr::pred("after.mode", CmpOp::Eq, "safe"),
]))
}
#[test]
fn given_an_indexed_array_path_when_decoding_then_should_retain_only_the_requested_field_and_validate_the_rest()
{
let mut paths = PathTrie::default();
paths.insert(&FieldPath::parse("items[1].sku").expect("path"));
let value = decode_json_paths(
br#"{"items":[{"other":"unused"},{"sku":"chosen","unused":[1,2,3]},{"sku":"unread"}]}"#,
&DecodeLimits::default(),
&paths,
)
.expect("decoded");
assert_eq!(value, serde_json::json!({"items":[null,{"sku":"chosen"}]}));
assert_eq!(
decode_json_paths(
br#"{"items":[{"bad":"\ud800"},{"sku":"chosen"}]}"#,
&DecodeLimits::default(),
&paths
),
Err(FaultReason::Malformed)
);
}
#[test]
fn given_the_safe_mode_fixtures_when_evaluated_then_should_match_the_scenario_table() {
let expected = [
(SAFE_MODE, Verdict::Selected, Verdict::Selected),
(SAFE_MODE_VALUES, Verdict::Rejected, Verdict::Selected),
(DECOMMISSION, Verdict::Selected, Verdict::Rejected),
(TELEMETRY, Verdict::Selected, Verdict::Rejected),
(GROUND_STATION, Verdict::Rejected, Verdict::Rejected),
];
for (payload, strict_verdict, legacy_verdict) in expected {
assert_eq!(
verdict(&strict(), payload),
strict_verdict,
"strict on {payload}"
);
assert_eq!(
verdict(&legacy(), payload),
legacy_verdict,
"legacy on {payload}"
);
}
}
#[test]
fn given_missing_and_null_fields_when_compared_then_should_follow_the_null_rules() {
let eq_null =
ConsumerFilter::json(FilterExpr::pred("missing", CmpOp::Eq, TypedValue::Null));
let ne_null =
ConsumerFilter::json(FilterExpr::pred("missing", CmpOp::Ne, TypedValue::Null));
let not_eq = ConsumerFilter::json(FilterExpr::negate(FilterExpr::pred(
"missing",
CmpOp::Eq,
"x",
)));
let present_null = ConsumerFilter::json(FilterExpr::present("value"));
let ne_null_on_null =
ConsumerFilter::json(FilterExpr::pred("value", CmpOp::Ne, TypedValue::Null));
assert_eq!(verdict(&eq_null, r#"{"value":null}"#), Verdict::Selected);
assert_eq!(verdict(&ne_null, r#"{"value":null}"#), Verdict::Rejected);
assert_eq!(
verdict(¬_eq, r#"{}"#),
Verdict::Rejected,
"not(unknown) is unknown"
);
assert_eq!(
verdict(&present_null, r#"{"value":null}"#),
Verdict::Selected
);
assert_eq!(
verdict(&ne_null_on_null, r#"{"value":null}"#),
Verdict::Rejected
);
}
#[test]
fn given_numbers_when_compared_then_should_stay_exact() {
let big =
ConsumerFilter::json(FilterExpr::pred("id", CmpOp::Eq, 9_007_199_254_740_993_i64));
assert_eq!(
verdict(&big, r#"{"id":9007199254740993}"#),
Verdict::Selected
);
assert_eq!(
verdict(&big, r#"{"id":9007199254740992}"#),
Verdict::Rejected
);
let mixed = ConsumerFilter::json(FilterExpr::pred("port", CmpOp::Gte, 5061.0));
assert_eq!(verdict(&mixed, r#"{"port":5061}"#), Verdict::Selected);
let inexact = ConsumerFilter::json(FilterExpr::pred("id", CmpOp::Eq, 1.0));
assert_eq!(
verdict(&inexact, r#"{"id":9007199254740993}"#),
Verdict::Rejected
);
let text_vs_number = ConsumerFilter::json(FilterExpr::pred("port", CmpOp::Eq, "5061"));
assert_eq!(
verdict(&text_vs_number, r#"{"port":5061}"#),
Verdict::Rejected
);
}
#[test]
fn given_contains_and_prefix_when_compared_then_should_be_literal() {
let contains = ConsumerFilter::json(FilterExpr::pred("name", CmpOp::Contains, "%"));
assert_eq!(verdict(&contains, r#"{"name":"100%"}"#), Verdict::Selected);
assert_eq!(verdict(&contains, r#"{"name":"100"}"#), Verdict::Rejected);
let array = ConsumerFilter::json(FilterExpr::pred("tags", CmpOp::Contains, 2));
assert_eq!(verdict(&array, r#"{"tags":[1,2,3]}"#), Verdict::Selected);
assert_eq!(verdict(&array, r#"{"tags":["2"]}"#), Verdict::Rejected);
let prefix = ConsumerFilter::json(FilterExpr::pred("name", CmpOp::Prefix, "eci_"));
assert_eq!(verdict(&prefix, r#"{"name":"eci_x"}"#), Verdict::Selected);
assert_eq!(verdict(&prefix, r#"{"name":"ecix"}"#), Verdict::Rejected);
}
#[test]
fn given_in_lists_when_compared_then_should_match_any_item() {
let filter = ConsumerFilter::json(FilterExpr::pred(
"table",
CmpOp::In,
TypedValue::List(vec!["satellites".into(), "ground_stations".into()]),
));
assert_eq!(
verdict(&filter, r#"{"table":"ground_stations"}"#),
Verdict::Selected
);
assert_eq!(
verdict(&filter, r#"{"table":"launches"}"#),
Verdict::Rejected
);
}
#[test]
fn given_coercions_when_compared_then_should_use_instants_and_exact_decimals() {
let after = ConsumerFilter::json(FilterExpr::pred_as(
"at",
CmpOp::Gt,
"2026-09-21T18:00:00Z",
Coerce::Timestamp {
format: TimestampFormat::Rfc3339,
},
));
assert_eq!(
verdict(&after, r#"{"at":"2026-09-21T20:04:12+02:00"}"#),
Verdict::Selected
);
assert_eq!(
verdict(&after, r#"{"at":"2026-09-21T19:59:59+02:00"}"#),
Verdict::Rejected
);
assert_eq!(
verdict(&after, r#"{"at":"2026-09-21T18:04:12"}"#),
Verdict::Rejected,
"no zone is unknown"
);
let millis = ConsumerFilter::json(FilterExpr::pred_as(
"timestamp_ms",
CmpOp::Lt,
"2026-09-21T16:04:13Z",
Coerce::Timestamp {
format: TimestampFormat::EpochMillis,
},
));
assert!(
CompiledFilter::compile(&millis).is_err(),
"an epoch literal must be an integer"
);
let amount = ConsumerFilter::json(FilterExpr::pred_as(
"amount",
CmpOp::Gt,
"9007199254740992",
Coerce::Number,
));
assert_eq!(
verdict(&amount, r#"{"amount":"9007199254740993"}"#),
Verdict::Selected
);
assert_eq!(
verdict(&amount, r#"{"amount":"9007199254740992.0"}"#),
Verdict::Rejected
);
assert_eq!(verdict(&amount, r#"{"amount":"abc"}"#), Verdict::Rejected);
}
#[test]
fn given_header_predicates_when_they_decide_then_should_skip_the_payload() {
let filter = ConsumerFilter::json(FilterExpr::all([
FilterExpr::pred("op", CmpOp::Eq, "d"),
FilterExpr::header("table", CmpOp::Eq, "satellites"),
]));
let compiled = CompiledFilter::compile(&filter).expect("compiles");
let ground_station = [HeaderRef {
key: "table",
value: HeaderValueRef::String("ground_stations"),
}];
let rejected = compiled.evaluate(
&FilterRecord {
payload: b"not json at all",
headers: &ground_station,
},
&DecodeLimits::default(),
);
assert_eq!(
rejected,
Verdict::Rejected,
"the header decides before any decode"
);
let satellite = [HeaderRef {
key: "table",
value: HeaderValueRef::String("satellites"),
}];
let fault = compiled.evaluate(
&FilterRecord {
payload: b"not json at all",
headers: &satellite,
},
&DecodeLimits::default(),
);
assert_eq!(
fault,
Verdict::Fault,
"the payload is needed and does not decode"
);
}
#[test]
fn given_payload_limits_when_exceeded_then_should_fault_without_parsing() {
let filter = ConsumerFilter::json(FilterExpr::present("a"));
let compiled = CompiledFilter::compile(&filter).expect("compiles");
let limits = DecodeLimits {
max_payload_bytes: 64,
max_depth: 3,
};
let deep = r#"{"a":{"b":{"c":{"d":1}}}}"#;
let record = FilterRecord {
payload: deep.as_bytes(),
headers: &[],
};
assert_eq!(compiled.evaluate(&record, &limits), Verdict::Fault);
assert_eq!(
compiled.evaluate_with_fault(&record, &limits).1,
Some(FaultReason::TooDeep)
);
let braces_in_strings = r#"{"a":"{{{{{{{{"}"#;
let record = FilterRecord {
payload: braces_in_strings.as_bytes(),
headers: &[],
};
assert_eq!(compiled.evaluate(&record, &limits), Verdict::Selected);
let large = format!(r#"{{"a":"{}"}}"#, "x".repeat(100));
let record = FilterRecord {
payload: large.as_bytes(),
headers: &[],
};
assert_eq!(
compiled.evaluate_with_fault(&record, &limits).1,
Some(FaultReason::TooLarge)
);
}
#[test]
fn given_a_record_when_explained_then_should_report_every_node() {
let compiled = CompiledFilter::compile(&strict()).expect("compiles");
let explanation = compiled.explain(
&FilterRecord {
payload: SAFE_MODE_VALUES.as_bytes(),
headers: &[],
},
&DecodeLimits::default(),
);
assert_eq!(explanation.verdict, Verdict::Rejected);
assert_eq!(explanation.root.label, "any");
assert_eq!(explanation.root.children.len(), 3);
let snapshot_branch = &explanation.root.children[0];
assert_eq!(snapshot_branch.truth, Some(Truth::Unknown));
assert_eq!(
snapshot_branch.children[2].label,
r#"changed contains "mode""#
);
assert_eq!(snapshot_branch.children[2].truth, Some(Truth::Unknown));
let malformed = compiled.explain(
&FilterRecord {
payload: b"{",
headers: &[],
},
&DecodeLimits::default(),
);
assert_eq!(malformed.verdict, Verdict::Fault);
assert_eq!(malformed.fault, Some(FaultReason::Malformed));
}
#[test]
fn given_duplicate_keys_when_decoded_then_should_keep_the_last_value() {
let filter = ConsumerFilter::json(FilterExpr::pred("kind", CmpOp::Eq, "B"));
assert_eq!(
verdict(&filter, r#"{"kind":"A","kind":"B"}"#),
Verdict::Selected
);
}
fn decided(
filter: &ConsumerFilter,
payload: &[u8],
headers: &[HeaderRef<'_>],
) -> (Verdict, Option<FaultReason>) {
CompiledFilter::compile(filter)
.expect("compiles")
.evaluate_with_fault(&FilterRecord { payload, headers }, &DecodeLimits::default())
}
fn content_type(code: u64) -> HeaderRef<'static> {
HeaderRef {
key: CONTENT_TYPE,
value: HeaderValueRef::Uint(code),
}
}
#[test]
fn given_a_record_declared_in_another_codec_when_a_json_filter_reads_it_then_should_reject_without_decoding()
{
let filter = ConsumerFilter::json(FilterExpr::pred("type", CmpOp::Eq, "billing"));
let protobuf = content_type(u64::from(ContentType::Protobuf.code()));
assert_eq!(
decided(&filter, b"\x08\x01", &[protobuf]),
(Verdict::Rejected, Some(FaultReason::ForeignCodec))
);
let passing = filter.clone().with_foreign_policy(RecordPolicy::Pass);
assert_eq!(
decided(&passing, b"\x08\x01", &[protobuf]),
(Verdict::Fault, Some(FaultReason::ForeignCodec))
);
let compiled = CompiledFilter::compile(&passing).expect("compiles");
assert_eq!(
compiled.policy_for(FaultReason::ForeignCodec),
FaultPolicy::Pass
);
let json = content_type(u64::from(ContentType::Json.code()));
assert_eq!(
decided(&filter, br#"{"type":"billing"}"#, &[json]),
(Verdict::Selected, None)
);
assert_eq!(
decided(&filter, br#"{"type":"billing"}"#, &[content_type(255)]),
(Verdict::Selected, None)
);
assert_eq!(
decided(&filter, br#"{"type":"billing"}"#, &[content_type(200)]),
(Verdict::Selected, None)
);
assert_eq!(
decided(&filter, b"not json", &[]),
(Verdict::Fault, Some(FaultReason::Malformed))
);
}
#[test]
fn given_a_value_of_another_type_when_compared_then_should_follow_the_mismatch_policy() {
let filter = ConsumerFilter::json(FilterExpr::pred("xyz", CmpOp::Eq, "Abc"));
for payload in [
r#"{"xyz":1}"#,
r#"{"xyz":{"a":1}}"#,
r#"{"xyz":[1]}"#,
r#"{"xyz":true}"#,
] {
assert_eq!(
decided(&filter, payload.as_bytes(), &[]),
(Verdict::Rejected, Some(FaultReason::TypeMismatch)),
"{payload}"
);
}
let passing = filter.with_mismatch_policy(RecordPolicy::Pass);
assert_eq!(
decided(&passing, br#"{"xyz":1}"#, &[]),
(Verdict::Fault, Some(FaultReason::TypeMismatch))
);
assert_eq!(
decided(&passing, br#"{"xyz":"Abc"}"#, &[]),
(Verdict::Selected, None)
);
assert_eq!(
decided(&passing, br#"{"xyz":"Abd"}"#, &[]),
(Verdict::Rejected, None)
);
assert_eq!(
decided(&passing, br#"{"other":1}"#, &[]),
(Verdict::Rejected, None)
);
assert_eq!(
decided(&passing, br#"{"xyz":null}"#, &[]),
(Verdict::Rejected, None)
);
}
#[test]
fn given_text_predicates_on_a_mixed_log_when_evaluated_then_should_select_the_subdomain() {
let filter = ConsumerFilter::json(FilterExpr::any([
FilterExpr::text("type", TextMatch::Contains, ".v1."),
FilterExpr::text("type", TextMatch::Glob, "SHIPPING.*").case_insensitive(),
]));
assert_eq!(
verdict(&filter, r#"{"type":"billing.invoice.v1.created"}"#),
Verdict::Selected
);
assert_eq!(
verdict(&filter, r#"{"type":"shipping.parcel.v2.sent"}"#),
Verdict::Selected
);
assert_eq!(
verdict(&filter, r#"{"type":"billing.invoice.v2.created"}"#),
Verdict::Rejected
);
let headers = ConsumerFilter::headers_only(FilterExpr::header_text(
"event.type",
TextMatch::Regex,
r"^billing\.[a-z]+\.v1\.",
));
let typed = [HeaderRef {
key: "event.type",
value: HeaderValueRef::String("billing.invoice.v1.created"),
}];
assert_eq!(
decided(&headers, b"\xff\xfe anything", &typed),
(Verdict::Selected, None)
);
let numeric = [HeaderRef {
key: "event.type",
value: HeaderValueRef::Uint(7),
}];
assert_eq!(
decided(&headers, b"", &numeric),
(Verdict::Rejected, Some(FaultReason::TypeMismatch))
);
let passing = headers.with_mismatch_policy(RecordPolicy::Pass);
assert_eq!(
decided(&passing, b"", &numeric),
(Verdict::Fault, Some(FaultReason::TypeMismatch))
);
}
#[test]
fn given_escaped_quotes_and_deep_nesting_when_depth_is_checked_then_should_count_only_structure()
{
assert!(json_depth_within(br#"{"a":"[[[[\"{{{{"}"#, 1));
assert!(!json_depth_within(br#"{"a":[[1]]}"#, 2));
assert!(json_depth_within(br#"{"a":"\\"}"#, 1));
assert!(json_depth_within(br#"{"a":"unterminated"#, 1));
}
}