Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Subscription matching (CIM 009 5.8.6) against one in-memory entity: the
3 : //! predicates the broker's notification path applies, usable without a
4 : //! broker — a gateway or an edge component can answer "would this change
5 : //! notify subscription X" with the broker's own semantics.
6 : //!
7 : //! Inputs are the stored (internal, expanded) forms: the subscription
8 : //! document as created (5.2.12, selector entity types expanded) and the
9 : //! entity document as the broker stores it (expanded attribute IRIs,
10 : //! `type` as an array). Matching is
11 : //! index-shaped in the broker (candidate lookup by (tenant, type) /
12 : //! (tenant, watched attribute)); every predicate here evaluates one
13 : //! candidate self-contained.
14 : #![cfg_attr(not(test), warn(clippy::expect_used))]
15 : #![deny(missing_docs)]
16 :
17 : use antares_jsonld::Context;
18 : use antares_model::dt_key;
19 : use antares_ql::eval::EntityLookup;
20 : use antares_ql::geo::GeoQuery;
21 : use antares_ql::type_selection_matches;
22 : use serde_json::{Map, Value};
23 :
24 1656 : fn sub_str<'a>(sub: &'a Value, key: &str) -> Option<&'a str> {
25 1656 : sub.get(key).and_then(Value::as_str)
26 1656 : }
27 :
28 32 : fn now_iso() -> String {
29 32 : chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true)
30 32 : }
31 :
32 : /// Every predicate at once: active, entities selector (5.2.33), `q` /
33 : /// `scopeQ` / `geoQ` conditions, throttling window. `lookup` resolves 4.9
34 : /// linked-entity terms (`attr{…}`); `&|_| None` when no store is at hand.
35 16 : pub fn would_notify(sub: &Value, doc: &Value, ctx: &Context, lookup: EntityLookup) -> bool {
36 16 : is_active(sub)
37 12 : && selector_match(sub, doc, ctx)
38 8 : && conditions_match(sub, doc, ctx, lookup)
39 4 : && !throttled(sub)
40 16 : }
41 :
42 : /// 5.8.1.4 / 5.2.12: `isActive` false or an `expiresAt` in the past means
43 : /// the subscription notifies nothing.
44 1521 : pub fn is_active(sub: &Value) -> bool {
45 1521 : if sub.get("isActive") == Some(&Value::Bool(false)) {
46 2 : return false;
47 1519 : }
48 : // 5.8.1.4 auto-expiry; dt_key so fraction spellings cannot misorder
49 : // around the boundary second (4.11)
50 1519 : !sub.get("expiresAt")
51 1519 : .and_then(Value::as_str)
52 1519 : .is_some_and(|e| dt_key(e) < dt_key(&now_iso()))
53 1521 : }
54 :
55 : /// entities selector (5.2.33) against an internal entity doc.
56 970 : pub fn selector_match(sub: &Value, doc: &Value, ctx: &Context) -> bool {
57 970 : let Some(sel) = sub.get("entities").and_then(Value::as_array) else {
58 18 : return true; // watchedAttributes-only subscription
59 : };
60 952 : let types: Vec<&str> = doc
61 952 : .get("type")
62 952 : .and_then(Value::as_array)
63 952 : .map(|a| a.iter().filter_map(Value::as_str).collect())
64 952 : .unwrap_or_default();
65 952 : let id = doc.get("id").and_then(Value::as_str).unwrap_or("");
66 960 : sel.iter().any(|e| {
67 960 : let t_ok = e.get("type").and_then(Value::as_str).is_none_or(|t| {
68 : // Table 5.2.33-1: "*" indicates "a request for all Entities" —
69 : // neither a term nor a 4.17 expression, but the absence of a type
70 : // predicate, exactly as `type=*` is on the query side.
71 902 : if t == "*" {
72 14 : true
73 888 : } else if t.contains(['|', ',', ';', '(']) {
74 16 : type_selection_matches(t, &types, ctx)
75 : } else {
76 872 : types.contains(&t)
77 : }
78 902 : });
79 : // Table 5.2.33-1: id is String or String[]; "id takes precedence
80 : // over idPattern" — a selector carrying id ignores its idPattern.
81 960 : let id_ok = match e.get("id") {
82 856 : None => true,
83 82 : Some(Value::String(i)) => i == id,
84 30 : Some(Value::Array(a)) => a.iter().filter_map(Value::as_str).any(|i| i == id),
85 0 : Some(_) => false,
86 : };
87 960 : let pat_ok = e.get("id").is_some()
88 856 : || e.get("idPattern").and_then(Value::as_str).is_none_or(|p| {
89 42 : antares_ql::regex::compile(p).is_ok_and(|re| re.find(id).is_some())
90 42 : });
91 960 : t_ok && id_ok && pat_ok
92 960 : })
93 970 : }
94 :
95 : /// A subscription's `geoQ` (Table 5.2.13-1) in the parameter shape the 4.10
96 : /// GeoQuery parser takes. The one reading of that table: every validator,
97 : /// matcher and forwarder in the broker turns a `geoQ` object into query
98 : /// parameters here, so `coordinates` is spelled the same way on all of them.
99 34 : pub fn geo_params(g: &Map<String, Value>) -> std::collections::HashMap<String, String> {
100 34 : let mut params: std::collections::HashMap<String, String> = Default::default();
101 102 : for k in ["georel", "geometry", "geoproperty"] {
102 102 : if let Some(s) = g.get(k).and_then(Value::as_str) {
103 68 : params.insert(k.into(), s.to_owned());
104 68 : }
105 : }
106 34 : if let Some(c) = g.get("coordinates") {
107 34 : params.insert(
108 34 : "coordinates".into(),
109 34 : match c {
110 6 : Value::String(s) => s.clone(),
111 28 : other => other.to_string(),
112 : },
113 : );
114 0 : }
115 34 : params
116 34 : }
117 :
118 : /// 5.8.6 notification matching: the subscription's q (4.9), scopeQ (4.19)
119 : /// and geoQ (4.10) conditions against an internal entity doc; all present
120 : /// conditions must hold.
121 778 : pub fn conditions_match(sub: &Value, doc: &Value, ctx: &Context, lookup: EntityLookup) -> bool {
122 778 : if let Some(q) = sub_str(sub, "q") {
123 : // q values in subscription bodies may be percent-encoded (4.9, 046_05)
124 70 : let q = antares_ql::percent_decode(q.as_bytes());
125 : // parsed once per distinct q text, not once per event per candidate
126 70 : match antares_ql::regex::q_node(&q) {
127 70 : Some(node) => {
128 : // Table 5.2.12-1 gives the Subscription the `expandValues`
129 : // and `jsonKeys` pair 4.9 gives the query, and this
130 : // condition IS that query: a term the first list names is
131 : // compared after JSON-LD type coercion against the @context
132 : // the Subscription was supplied with, less the names the
133 : // second declares uninterpretable.
134 : // ponytail: the coercion rebuilds the tree, so it costs one
135 : // clone per event for a Subscription that names the list and
136 : // nothing at all for one that does not; cache the coerced
137 : // tree beside the parsed one if a deployment measures it.
138 : let coerced;
139 70 : let node: &antares_ql::QNode = match antares_ql::eval::expansion_list(
140 70 : sub_str(sub, "expandValues"),
141 70 : sub_str(sub, "jsonKeys"),
142 70 : ) {
143 64 : None => &node,
144 6 : Some(list) => {
145 6 : coerced = antares_ql::eval::apply_expand_values(
146 6 : (*node).clone(),
147 6 : Some(&list),
148 6 : ctx,
149 : );
150 6 : &coerced
151 : }
152 : };
153 70 : if !antares_ql::eval::eval_q(node, doc, ctx, lookup) {
154 40 : return false;
155 30 : }
156 : }
157 0 : None => return false,
158 : }
159 708 : }
160 738 : if let Some(sq) = sub_str(sub, "scopeQ") {
161 4 : if !antares_ql::scope::scope_matches(sq, doc) {
162 2 : return false;
163 2 : }
164 734 : }
165 736 : if let Some(g) = sub.get("geoQ").and_then(Value::as_object) {
166 : // the geometry parse is shared per distinct geoQ member; the
167 : // serialization of the stored member is the key
168 10 : let key = serde_json::to_string(g).unwrap_or_default();
169 10 : let gq = antares_ql::regex::geo_query(&key, || {
170 2 : GeoQuery::from_params(&geo_params(g)).ok().flatten()
171 2 : });
172 10 : match gq {
173 10 : Some(gq) => {
174 10 : if !gq.matches(doc, ctx) {
175 2 : return false;
176 8 : }
177 : }
178 0 : None => return false,
179 : }
180 726 : }
181 734 : true
182 778 : }
183 :
184 : /// 5.2.12 `throttling`: true while the last notification is younger than the
185 : /// throttling window, so no further notification is due yet.
186 408 : pub fn throttled(sub: &Value) -> bool {
187 408 : let Some(secs) = sub.get("throttling").and_then(Value::as_f64) else {
188 394 : return false;
189 : };
190 14 : let Some(last) = sub
191 14 : .get("notification")
192 14 : .and_then(|n| n.get("lastNotification"))
193 14 : .and_then(Value::as_str)
194 : else {
195 2 : return false;
196 : };
197 12 : chrono::DateTime::parse_from_rfc3339(last).is_ok_and(|t| {
198 12 : (chrono::Utc::now() - t.with_timezone(&chrono::Utc)).num_milliseconds()
199 12 : < (secs * 1000.0) as i64
200 12 : })
201 408 : }
202 :
203 : #[cfg(test)]
204 : mod tests {
205 : use super::*;
206 : use serde_json::json;
207 : use std::sync::Arc;
208 :
209 : const DC: &str = "https://uri.etsi.org/ngsi-ld/default-context";
210 :
211 : /// The core context once per process, with no loader behind it: the
212 : /// interpreter-run of these tests (Miri) must not pay for a cache and an
213 : /// HTTP client it never uses.
214 48 : fn ctx() -> Arc<Context> {
215 : static CORE: std::sync::OnceLock<Arc<Context>> = std::sync::OnceLock::new();
216 48 : CORE.get_or_init(|| Arc::new(antares_jsonld::core_context()))
217 48 : .clone()
218 48 : }
219 :
220 : /// A stored entity in its internal form: the broker's own expansion of
221 : /// the API payload (expanded IRIs, `type` as an array, instance arrays).
222 26 : fn expand(doc: Value) -> Value {
223 26 : antares_jsonld::expand_entity(
224 26 : doc.as_object().expect("object"),
225 26 : &ctx(),
226 26 : antares_jsonld::ExpandOpts::default(),
227 : )
228 26 : .expect("a valid entity")
229 26 : }
230 :
231 22 : fn vehicle(speed: f64, lon: f64) -> Value {
232 22 : expand(json!({
233 22 : "id": "urn:ngsi-ld:Vehicle:A1",
234 22 : "type": "Vehicle",
235 22 : "speed": {"type": "Property", "value": speed},
236 22 : "driver": {"type": "Relationship", "object": "urn:ngsi-ld:Person:P1"},
237 22 : "location": {"type": "GeoProperty",
238 22 : "value": {"type": "Point", "coordinates": [lon, 40.4]}}
239 : }))
240 22 : }
241 :
242 : /// A subscription as the broker stores it: entity types in the selector
243 : /// are already expanded (5.2.33), `q`/`geoQ` as sent.
244 22 : fn sub() -> Value {
245 22 : json!({
246 22 : "id": "urn:ngsi-ld:Subscription:1",
247 22 : "type": "Subscription",
248 22 : "entities": [{"type": format!("{DC}/Vehicle"), "idPattern": "^urn:ngsi-ld:Vehicle:.*"}],
249 22 : "q": "speed>25",
250 22 : "geoQ": {"georel": "near;maxDistance==2000", "geometry": "Point",
251 22 : "coordinates": "[-3.7,40.4]"},
252 22 : "notification": {"endpoint": {"uri": "http://x/n"}}
253 : })
254 22 : }
255 :
256 : /// Table 5.2.12-1 gives a Subscription the same `expandValues` and
257 : /// `jsonKeys` pair 4.9 gives a query, and the notification condition IS
258 : /// that query: a term whose Attribute `expandValues` names is compared
259 : /// after JSON-LD type coercion, and a name `jsonKeys` carries as well is
260 : /// left uninterpreted (6.4.3.2, the broker's precedence).
261 : #[test]
262 2 : fn clause_5_2_12_expand_values_coerces_the_condition_value() {
263 2 : let c = ctx();
264 2 : let none = |_: &str| None;
265 2 : let doc = expand(json!({
266 2 : "id": "urn:ngsi-ld:Vehicle:A1",
267 2 : "type": "Vehicle",
268 2 : "category": {"type": "Property", "value": format!("{DC}/Camping")}
269 : }));
270 2 : let mut s = sub();
271 2 : s.as_object_mut().expect("object").remove("geoQ");
272 2 : s["q"] = json!("category==\"Camping\"");
273 2 : assert!(
274 2 : !conditions_match(&s, &doc, &c, &none),
275 : "the literal cannot match the expanded value on its own"
276 : );
277 2 : s["expandValues"] = json!("category");
278 2 : assert!(
279 2 : conditions_match(&s, &doc, &c, &none),
280 : "expandValues coerces the term value"
281 : );
282 2 : s["jsonKeys"] = json!("category");
283 2 : assert!(
284 2 : !conditions_match(&s, &doc, &c, &none),
285 : "a name in both lists is left uninterpreted"
286 : );
287 2 : }
288 :
289 : /// 5.8.6: every predicate holds → the change would notify.
290 : #[test]
291 2 : fn matching_change_notifies() {
292 2 : assert!(would_notify(&sub(), &vehicle(30.0, -3.7), &ctx(), &|_| {
293 0 : None
294 0 : }));
295 2 : }
296 :
297 : /// Each predicate alone refuses: q (4.9), geoQ (4.10), selector type,
298 : /// idPattern (5.2.33), `isActive`, `expiresAt` (5.8.1.4), throttling.
299 : #[test]
300 2 : fn each_failing_predicate_refuses() {
301 2 : let c = ctx();
302 2 : let none = |_: &str| None;
303 2 : assert!(!would_notify(&sub(), &vehicle(20.0, -3.7), &c, &none), "q");
304 2 : assert!(
305 2 : !would_notify(&sub(), &vehicle(30.0, -4.7), &c, &none),
306 : "geoQ"
307 : );
308 2 : let mut s = sub();
309 2 : s["entities"][0]["type"] = json!(format!("{DC}/Bicycle"));
310 2 : assert!(!would_notify(&s, &vehicle(30.0, -3.7), &c, &none), "type");
311 2 : let mut s = sub();
312 2 : s["entities"][0]["idPattern"] = json!("^urn:ngsi-ld:Bicycle:.*");
313 2 : assert!(
314 2 : !would_notify(&s, &vehicle(30.0, -3.7), &c, &none),
315 : "idPattern"
316 : );
317 2 : let mut s = sub();
318 2 : s["isActive"] = json!(false);
319 2 : assert!(
320 2 : !would_notify(&s, &vehicle(30.0, -3.7), &c, &none),
321 : "isActive"
322 : );
323 2 : let mut s = sub();
324 2 : s["expiresAt"] = json!("2000-01-01T00:00:00Z");
325 2 : assert!(
326 2 : !would_notify(&s, &vehicle(30.0, -3.7), &c, &none),
327 : "expiresAt"
328 : );
329 2 : let mut s = sub();
330 2 : s["throttling"] = json!(3600);
331 2 : s["notification"]["lastNotification"] = json!(now_iso());
332 2 : assert!(
333 2 : !would_notify(&s, &vehicle(30.0, -3.7), &c, &none),
334 : "throttling"
335 : );
336 2 : assert!(throttled(&s));
337 2 : s["notification"]["lastNotification"] = json!("2000-01-01T00:00:00Z");
338 2 : assert!(
339 2 : !throttled(&s),
340 : "an old lastNotification is outside the window"
341 : );
342 2 : }
343 :
344 : /// 5.2.33 Table 5.2.33-1: `id` takes precedence over `idPattern`, and a
345 : /// selector without `type` matches every type.
346 : #[test]
347 2 : fn selector_id_precedence_and_typeless_selector() {
348 2 : let c = ctx();
349 2 : let doc = vehicle(30.0, -3.7);
350 2 : let s = json!({"entities": [{"id": "urn:ngsi-ld:Vehicle:A1", "idPattern": "^nothing$"}]});
351 2 : assert!(selector_match(&s, &doc, &c));
352 2 : let s = json!({"entities": [{"id": ["urn:ngsi-ld:Vehicle:B2"], "type": format!("{DC}/Vehicle")}]});
353 2 : assert!(!selector_match(&s, &doc, &c));
354 2 : let s = json!({"watchedAttributes": ["speed"]});
355 2 : assert!(
356 2 : selector_match(&s, &doc, &c),
357 : "watchedAttributes-only subscription"
358 : );
359 2 : }
360 :
361 : /// 4.9 EXAMPLE 13/14: a linked-entity term (`driver{name}`) resolves the
362 : /// Relationship object through `lookup`; without a store it cannot match.
363 : #[test]
364 2 : fn linked_entity_term_uses_the_lookup() {
365 2 : let c = ctx();
366 2 : let mut s = sub();
367 2 : s["q"] = json!("driver{name}==\"Ann\"");
368 2 : let doc = vehicle(30.0, -3.7);
369 2 : let store = |uri: &str| {
370 2 : (uri == "urn:ngsi-ld:Person:P1").then(|| {
371 2 : expand(json!({"id": uri, "type": "Person",
372 2 : "name": {"type": "Property", "value": "Ann"}}))
373 2 : })
374 2 : };
375 2 : assert!(conditions_match(&s, &doc, &c, &store));
376 2 : assert!(
377 2 : !conditions_match(&s, &doc, &c, &|_| None),
378 : "no store, no match"
379 : );
380 2 : }
381 :
382 : /// 4.19: `scopeQ` is a condition like `q`; a percent-encoded `q` (as a
383 : /// subscription body may carry it) decodes before evaluation.
384 : #[test]
385 2 : fn scope_q_and_percent_encoded_q() {
386 2 : let c = ctx();
387 2 : let mut doc = vehicle(30.0, -3.7);
388 2 : doc["scope"] = json!(["/Madrid/Centre"]);
389 2 : let mut s = sub();
390 2 : s["scopeQ"] = json!("/Madrid/#");
391 2 : s["q"] = json!("speed%3E25");
392 2 : assert!(conditions_match(&s, &doc, &c, &|_| None));
393 2 : s["scopeQ"] = json!("/Paris/#");
394 2 : assert!(!conditions_match(&s, &doc, &c, &|_| None));
395 2 : }
396 :
397 : /// Table 5.2.33-1 `type`: "A valid type selection string as per clause
398 : /// 4.17. To indicate a request for all Entities (with implied local
399 : /// scope), \"*\" is also allowed as a value." 5.2.33 scopes EntitySelector
400 : /// to what is "queried or subscribed to", so a Subscription carrying it
401 : /// selects every Entity Type — it is not a 4.17 expression and not a term
402 : /// to expand, but the absence of a type predicate.
403 : #[test]
404 2 : fn a_star_selector_type_selects_every_entity_type() {
405 2 : let sub = json!({"entities": [{"type": "*"}]});
406 6 : for doc in [
407 2 : json!({"id": "urn:ngsi-ld:Vehicle:1", "type": [format!("{DC}/Vehicle")]}),
408 2 : json!({"id": "urn:ngsi-ld:Parking:1", "type": [format!("{DC}/Parking")]}),
409 2 : json!({"id": "urn:ngsi-ld:Odd:1", "type": ["urn:example:Odd"]}),
410 2 : ] {
411 6 : assert!(
412 6 : selector_match(&sub, &doc, &ctx()),
413 : "a \"*\" selector must match every type: {doc}"
414 : );
415 : }
416 : // "*" is the type predicate only — id and idPattern still narrow.
417 2 : let pinned = json!({"entities": [{"type": "*", "id": "urn:ngsi-ld:Vehicle:1"}]});
418 2 : assert!(selector_match(
419 2 : &pinned,
420 2 : &json!({"id": "urn:ngsi-ld:Vehicle:1", "type": [format!("{DC}/Vehicle")]}),
421 2 : &ctx()
422 : ));
423 2 : assert!(
424 2 : !selector_match(
425 2 : &pinned,
426 2 : &json!({"id": "urn:ngsi-ld:Vehicle:2", "type": [format!("{DC}/Vehicle")]}),
427 2 : &ctx()
428 2 : ),
429 : "a \"*\" type does not widen an id that was given"
430 : );
431 2 : }
432 : }
|
