Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Registration matching over a registration document: the `CsrSpec`
3 : //! of a request or subscription, the 5.9 information/registration
4 : //! match (4.3.6.1), the csf and scope filters, the temporal interval
5 : //! and expiry of a registration, and the 5.11.2 subscription match.
6 : //! Pure functions over documents; no route and no store.
7 :
8 : use antares_jsonld::Context;
9 : use antares_store::Kind;
10 : use serde_json::{Map, Value};
11 : use std::collections::HashMap;
12 :
13 : /// 5.9.2.4: a registration whose expiresAt has been reached counts as
14 : /// deleted — lazily filtered on every read/match path (dt_key so fraction
15 : /// spellings cannot misorder, 4.11).
16 204549 : pub fn reg_expired(doc: &Value) -> bool {
17 204549 : doc.get("expiresAt")
18 204549 : .and_then(Value::as_str)
19 204549 : .is_some_and(|e| antares_model::dt_key(e) < antares_model::dt_key(&crate::state::now_iso()))
20 204549 : }
21 :
22 : /// Output shaping: compact IRIs.
23 203 : pub fn present_registration(doc: &Value, ctx: &Context, sys_attrs: bool) -> Value {
24 203 : let Some(obj) = doc.as_object() else {
25 0 : return doc.clone();
26 : };
27 203 : let mut out = Map::new();
28 1354 : for (k, v) in obj {
29 1354 : match k.as_str() {
30 1354 : "createdAt" | "modifiedAt" if !sys_attrs => continue,
31 1040 : "information" => {
32 203 : let infos: Vec<Value> = v
33 203 : .as_array()
34 203 : .cloned()
35 203 : .unwrap_or_default()
36 203 : .iter()
37 203 : .map(|info| {
38 203 : let mut ni = info.as_object().cloned().unwrap_or_default();
39 203 : if let Some(es) = ni.get("entities").and_then(Value::as_array) {
40 203 : let nes: Vec<Value> = es
41 203 : .iter()
42 203 : .map(|e| {
43 203 : let mut ne = e.as_object().cloned().unwrap_or_default();
44 203 : match ne.get("type") {
45 195 : Some(Value::String(t)) => {
46 195 : let c = ctx.compact_iri(t);
47 195 : ne.insert("type".into(), Value::String(c));
48 195 : }
49 8 : Some(Value::Array(ts)) => {
50 8 : let cs: Vec<Value> = ts
51 8 : .iter()
52 8 : .filter_map(Value::as_str)
53 16 : .map(|t| Value::String(ctx.compact_iri(t)))
54 8 : .collect();
55 8 : ne.insert("type".into(), Value::Array(cs));
56 : }
57 0 : _ => {}
58 : }
59 203 : Value::Object(ne)
60 203 : })
61 203 : .collect();
62 203 : ni.insert("entities".into(), Value::Array(nes));
63 0 : }
64 406 : for names_key in ["propertyNames", "relationshipNames"] {
65 406 : if let Some(names) = ni.get(names_key).and_then(Value::as_array) {
66 0 : let nn: Vec<Value> = names
67 0 : .iter()
68 0 : .filter_map(Value::as_str)
69 0 : .map(|n| Value::String(ctx.compact_iri(n)))
70 0 : .collect();
71 0 : ni.insert(names_key.into(), Value::Array(nn));
72 406 : }
73 : }
74 203 : Value::Object(ni)
75 203 : })
76 203 : .collect();
77 203 : out.insert("information".into(), Value::Array(infos));
78 : }
79 837 : _ => {
80 837 : out.insert(k.clone(), v.clone());
81 837 : }
82 : }
83 : }
84 203 : Value::Object(out)
85 203 : }
86 :
87 : /// 5.10.2.4 temporal matching against observationInterval/managementInterval.
88 0 : pub(crate) fn temporal_interval_matches(doc: &Value, tq: &crate::temporalq::TemporalQ) -> bool {
89 0 : let key = if tq.timeproperty == "observedAt" {
90 0 : "observationInterval"
91 : } else {
92 0 : "managementInterval"
93 : };
94 0 : let Some(iv) = doc.get(key).and_then(Value::as_object) else {
95 0 : return false; // relevant interval not present ⇒ no match
96 : };
97 : // 4.11 comparison on the canonical key — equal instants in different
98 : // 4.6.3 fraction spellings must hit the bounds exactly.
99 0 : let dt = antares_model::dt_key;
100 0 : let start = dt(iv.get("startAt").and_then(Value::as_str).unwrap_or(""));
101 0 : let end = iv.get("endAt").and_then(Value::as_str).map(dt); // open-ended when absent
102 0 : match tq.timerel.as_str() {
103 : // interval contains times before/after timeAt (037_09, 047_10/11)
104 0 : "before" => start < dt(&tq.time_at),
105 0 : "after" => end.is_none_or(|e| e > dt(&tq.time_at)),
106 0 : "between" => {
107 : // overlap between [timeAt, endTimeAt] and the interval
108 0 : let qe = dt(tq.end_time_at.as_deref().unwrap_or(&tq.time_at));
109 0 : dt(&tq.time_at) <= end.unwrap_or_else(|| "9999".into()) && qe >= start
110 : }
111 0 : _ => true,
112 : }
113 0 : }
114 :
115 : /// The entity/attribute specification matched against registrations (5.12).
116 : #[derive(Default)]
117 : pub struct CsrSpec {
118 : /// Expanded type IRIs (or raw 4.17 selector expressions).
119 : pub types: Option<Vec<String>>,
120 : pub ids: Option<Vec<String>>,
121 : pub id_pattern: Option<String>,
122 : /// Expanded attribute IRIs.
123 : pub attrs: Option<Vec<String>>,
124 : /// 5.12 datasetId condition (should-level): with both the request and
125 : /// the CSourceRegistration specifying datasetId, they match only with
126 : /// "at least one value in common"; one side alone always matches.
127 : pub dataset_ids: Option<Vec<String>>,
128 : /// 4.9 Context Source Filter: with a csf present, only registrations
129 : /// whose Context Source Properties match it are considered (query,
130 : /// temporal query, purge, entityMaps — 5.7.2.4/5.7.4.4/5.6.21.4).
131 : pub csf: Option<antares_ql::QNode>,
132 : /// 5.2.9 location ("Location for which the Context Source may be able
133 : /// to provide information") + 4.3.6.1: a geo query is only distributed
134 : /// to registrations whose location geometry matches it.
135 : pub geo: Option<antares_ql::geo::GeoQuery>,
136 : /// 5.2.9 observationInterval/managementInterval: "matched against the
137 : /// observationInterval for overlap" — a temporal read is only
138 : /// distributed to registrations whose declared interval overlaps the
139 : /// temporal query; a registration declaring NO interval stays
140 : /// unconstrained (both members are optional).
141 : pub temporal: Option<crate::temporalq::TemporalQ>,
142 : }
143 :
144 : /// 5.12: does an EntityInfo element match the entity specification?
145 32790 : fn entity_info_matches(spec: &CsrSpec, ei: &Value, ctx: &Context) -> bool {
146 32790 : if let Some(types) = &spec.types {
147 32582 : let its = ei_types(ei);
148 : // EntityInfo without a type restricts only by id/idPattern. Each spec
149 : // entry is a 4.17 Entity Type Selection over the WHOLE declared type
150 : // list (a conjunction needs every named type present); a plain
151 : // expanded IRI is the one-term case of the same evaluation.
152 32582 : if !its.is_empty()
153 32582 : && !types.iter().any(|t| {
154 32582 : its.contains(&t.as_str()) || antares_ql::type_selection_matches(t, &its, ctx)
155 32582 : })
156 : {
157 136 : return false;
158 32446 : }
159 208 : }
160 32654 : let ei_id = ei.get("id").and_then(Value::as_str);
161 32654 : let ei_pat = ei.get("idPattern").and_then(Value::as_str);
162 32654 : if ei_id.is_none() && ei_pat.is_none() {
163 670 : return true;
164 31984 : }
165 31984 : if let Some(ids) = &spec.ids {
166 31952 : if let Some(rid) = ei_id {
167 1206 : if ids.iter().any(|i| i == rid) {
168 1198 : return true;
169 8 : }
170 30746 : }
171 30754 : if let Some(p) = ei_pat {
172 30746 : if let Ok(re) = antares_ql::regex::compile(p) {
173 30748 : if ids.iter().any(|i| re.find(i).is_some()) {
174 14 : return true;
175 30732 : }
176 0 : }
177 8 : }
178 32 : }
179 30772 : if let Some(qp) = &spec.id_pattern {
180 20 : if let Some(rid) = ei_id {
181 12 : if antares_ql::regex::compile(qp).is_ok_and(|re| re.find(rid).is_some()) {
182 6 : return true;
183 6 : }
184 8 : }
185 14 : if ei_pat.is_some() {
186 8 : return true; // both patterns present ⇒ assumed compatible (5.12)
187 6 : }
188 30752 : }
189 : // no id restriction given by the query side ⇒ EntityInfo id restrictions
190 : // don't exclude it when the type matched
191 30758 : spec.ids.is_none() && spec.id_pattern.is_none()
192 32790 : }
193 :
194 1932 : fn attrs_match_info(attrs: &Option<Vec<String>>, info: &Value) -> bool {
195 1932 : let Some(attrs) = attrs else { return true };
196 1206 : if attrs.is_empty() {
197 0 : return true;
198 1206 : }
199 1206 : let props = info.get("propertyNames").and_then(Value::as_array);
200 1206 : let rels = info.get("relationshipNames").and_then(Value::as_array);
201 1206 : if props.is_none() && rels.is_none() {
202 184 : return true;
203 1022 : }
204 1032 : let has = |list: Option<&Vec<Value>>| {
205 1032 : list.is_some_and(|l| {
206 1022 : l.iter()
207 1022 : .filter_map(Value::as_str)
208 1022 : .any(|n| attrs.iter().any(|w| w == n))
209 1022 : })
210 1032 : };
211 1022 : has(props) || has(rels)
212 1932 : }
213 :
214 : /// 5.12: the RegistrationInfo elements of `doc.information` that match `spec`.
215 32806 : pub fn matching_infos<'a>(spec: &CsrSpec, doc: &'a Value, ctx: &Context) -> Vec<&'a Value> {
216 32806 : let Some(infos) = doc.get("information").and_then(Value::as_array) else {
217 0 : return Vec::new();
218 : };
219 32806 : infos
220 32806 : .iter()
221 32806 : .filter(|info| {
222 32806 : let entity_ok = match info.get("entities").and_then(Value::as_array) {
223 16 : None => true,
224 32790 : Some(es) => es.iter().any(|ei| entity_info_matches(spec, ei, ctx)),
225 : };
226 32806 : entity_ok && attrs_match_info(&spec.attrs, info)
227 32806 : })
228 32806 : .collect()
229 32806 : }
230 :
231 : /// 5.10.2.4: the context source filter (csf, 4.9) evaluates over the
232 : /// registration document's own Context Source Properties — its members are
233 : /// wrapped as Property instances so the shared 4.9 evaluator applies.
234 56 : pub(crate) fn csf_matches(csf: &antares_ql::QNode, reg: &Value, ctx: &Context) -> bool {
235 56 : let Some(obj) = reg.as_object() else {
236 0 : return false;
237 : };
238 56 : let mut pseudo = Map::new();
239 56 : pseudo.insert("id".into(), obj.get("id").cloned().unwrap_or(Value::Null));
240 56 : pseudo.insert(
241 56 : "type".into(),
242 56 : serde_json::json!(["ContextSourceRegistration"]),
243 : );
244 478 : for (k, v) in obj {
245 478 : if ["id", "type", "information", "createdAt", "modifiedAt"].contains(&k.as_str()) {
246 240 : continue;
247 238 : }
248 : // a Context Source Property stored in attribute form (5.2.9) is an
249 : // instance already — only bare scalars need the Property wrap
250 238 : let inst = match v {
251 42 : Value::Array(_) => v.clone(),
252 24 : Value::Object(o) if o.contains_key("value") || o.contains_key("object") => {
253 20 : serde_json::json!([v])
254 : }
255 176 : _ => serde_json::json!([{"type": "Property", "value": v}]),
256 : };
257 238 : pseudo.insert(ctx.expand_key(k), inst);
258 : }
259 56 : antares_ql::eval::eval_q(csf, &Value::Object(pseudo), ctx, &|_| None)
260 56 : }
261 :
262 1379 : pub fn csr_matches(spec: &CsrSpec, doc: &Value, ctx: &Context) -> bool {
263 1379 : !matching_infos(spec, doc, ctx).is_empty()
264 1379 : }
265 :
266 : /// Full 5.11.2.4 match of a registration against a csource subscription:
267 : /// 5.12 entity/attr matching + temporal interval rules + geoQ vs the
268 : /// registration's own `location`.
269 : ///
270 : /// Not the same rule as `antares_matcher::selector_match`, which it shares a
271 : /// signature with: that one asks whether an ENTITY satisfies a
272 : /// subscription's `entities` selector, this one asks whether a REGISTRATION
273 : /// does — and a registration carries its selector inside `information`, has
274 : /// an observation/management interval a latest-information subscription must
275 : /// not match, and answers `geoQ` from its own `location` rather than from a
276 : /// GeoProperty of the data. The part that IS the same rule is the selector
277 : /// walk, and it is not written twice: `spec_for_subscription` turns the
278 : /// subscription into a `CsrSpec` and `csr_matches` walks it.
279 142 : pub fn csr_matches_subscription(sub: &Value, reg: &Value, ctx: &Context) -> bool {
280 : // An expired registration is no longer a Context Source: it must not be
281 : // reported as newlyMatching, nor receive a forwarded subscription copy.
282 142 : if reg_expired(reg) {
283 1 : return false;
284 141 : }
285 141 : let spec = spec_for_subscription(sub);
286 141 : if !csr_matches(&spec, reg, ctx) {
287 16 : return false;
288 125 : }
289 125 : let has_interval =
290 125 : reg.get("observationInterval").is_some() || reg.get("managementInterval").is_some();
291 125 : match sub.get("temporalQ").and_then(Value::as_object) {
292 : None => {
293 125 : if has_interval {
294 0 : return false; // latest-information sources only (5.11.2.4)
295 125 : }
296 : }
297 0 : Some(tq) => {
298 0 : let mut params: HashMap<String, String> = HashMap::new();
299 0 : for k in ["timerel", "timeAt", "endTimeAt", "timeproperty"] {
300 0 : if let Some(s) = tq.get(k).and_then(Value::as_str) {
301 0 : params.insert(k.into(), s.into());
302 0 : }
303 : }
304 0 : if let Ok(Some(t)) = crate::temporalq::TemporalQ::from_params(¶ms, false) {
305 0 : if t.timerel != "any" && !temporal_interval_matches(reg, &t) {
306 0 : return false;
307 0 : }
308 0 : }
309 : }
310 : }
311 125 : if let Some(g) = sub.get("geoQ").and_then(Value::as_object) {
312 0 : if let Ok(Some(gq)) =
313 0 : antares_ql::geo::GeoQuery::from_params(&antares_matcher::geo_params(g))
314 : {
315 0 : match reg.get("location") {
316 0 : Some(geom) => {
317 0 : if !gq.matches_geometry(geom) {
318 0 : return false;
319 0 : }
320 : }
321 0 : None => return false,
322 : }
323 0 : }
324 125 : }
325 : // 5.11.2.4: csf vs the registration's Context Source Properties, scopeQ
326 : // vs its scope property
327 125 : if let Some(csf) = sub.get("csf").and_then(Value::as_str) {
328 18 : match antares_ql::parse_q(csf) {
329 18 : Ok(ast) if csf_matches(&ast, reg, ctx) => {}
330 8 : _ => return false,
331 : }
332 107 : }
333 117 : if let Some(sq) = sub.get("scopeQ").and_then(Value::as_str) {
334 8 : if !antares_ql::scope::scope_matches(sq, reg) {
335 4 : return false;
336 4 : }
337 109 : }
338 113 : true
339 142 : }
340 :
341 : /// Build the 5.12 spec for a csource subscription (5.11.2.4): entities
342 : /// selectors + watchedAttributes ∪ notification.attributes.
343 411 : pub fn spec_for_subscription(sub: &Value) -> CsrSpec {
344 411 : let mut spec = CsrSpec::default();
345 411 : if let Some(es) = sub.get("entities").and_then(Value::as_array) {
346 411 : let mut types = Vec::new();
347 411 : let mut ids = Vec::new();
348 411 : for e in es {
349 411 : if let Some(t) = e.get("type").and_then(Value::as_str) {
350 411 : types.push(t.to_owned());
351 411 : }
352 411 : if let Some(i) = e.get("id").and_then(Value::as_str) {
353 0 : ids.push(i.to_owned());
354 411 : }
355 411 : if spec.id_pattern.is_none() {
356 411 : spec.id_pattern = e
357 411 : .get("idPattern")
358 411 : .and_then(Value::as_str)
359 411 : .map(str::to_owned);
360 411 : }
361 : }
362 411 : if !types.is_empty() {
363 411 : spec.types = Some(types);
364 411 : }
365 411 : if !ids.is_empty() {
366 0 : spec.ids = Some(ids);
367 411 : }
368 0 : }
369 411 : let mut attrs: Vec<String> = sub
370 411 : .get("watchedAttributes")
371 411 : .and_then(Value::as_array)
372 411 : .map(|a| {
373 0 : a.iter()
374 0 : .filter_map(Value::as_str)
375 0 : .map(str::to_owned)
376 0 : .collect()
377 0 : })
378 411 : .unwrap_or_default();
379 411 : if let Some(na) = sub
380 411 : .get("notification")
381 411 : .and_then(|n| n.get("attributes"))
382 411 : .and_then(Value::as_array)
383 0 : {
384 0 : attrs.extend(na.iter().filter_map(Value::as_str).map(str::to_owned));
385 411 : }
386 411 : if !attrs.is_empty() {
387 0 : spec.attrs = Some(attrs);
388 411 : }
389 411 : spec
390 411 : }
391 :
392 : /// The kind one Registration Subscription id is stored under.
393 323 : pub(crate) fn csr_kind(id: &str) -> Kind {
394 323 : if id.starts_with(INTERNAL_CSR_PREFIX) {
395 311 : Kind::DistSub
396 : } else {
397 12 : Kind::CSourceSubscription
398 : }
399 323 : }
400 :
401 : /// 5.8.1.4: "The mapping of the received subscriptionId with the own
402 : /// Subscription identifier is stored" (inbound), "a mapping of the id of
403 : /// the Context Source Registration to the received subscriptionId is
404 : /// stored" (remotes), and "the mapping of the id of the Subscription to the
405 : /// … Context Source Registration Subscription shall be stored" (csr_sub).
406 : /// All three live in the store (Kind::DistSub) so persistent modes keep the
407 : /// consumer half across restarts: one doc per (tenant, own Subscription id)
408 : /// = {"csr_sub": id, "remotes": {reg_id: [endpoint, remote sub id]}}, plus
409 : /// inbound index docs under the internal "distsub-index" tenant
410 : /// (id = remote subscriptionId, doc = {"tenant", "own"}).
411 : /// 5.8.1.4: the Registration Subscription the distributed half owns is
412 : /// broker plumbing, not a resource a Context Source Subscriber created —
413 : /// 5.11.5.4 lists the subscriptions clients made through 5.11.2, and this
414 : /// one carries the internal `urn:antares:distsub:` endpoint naming the
415 : /// tenant and the owning Subscription. It is stored under `Kind::DistSub`,
416 : /// so the 5.11 endpoints cannot read, patch or delete it, and its id
417 : /// namespace is what tells the two apart on the notification path.
418 : pub(crate) const INTERNAL_CSR_PREFIX: &str = "urn:ngsi-ld:CSourceSubscription:distsub:";
419 :
420 : /// 5.2.8: EntityInfo type is a String or String[] — yield every named type.
421 35390 : pub(crate) fn ei_types(ei: &Value) -> Vec<&str> {
422 35390 : match ei.get("type") {
423 35184 : Some(Value::String(s)) => vec![s.as_str()],
424 20 : Some(Value::Array(a)) => a.iter().filter_map(Value::as_str).collect(),
425 186 : _ => Vec::new(),
426 : }
427 35390 : }
428 :
429 : #[cfg(test)]
430 : mod tests {
431 : use super::*;
432 : use antares_jsonld::Loader;
433 : use serde_json::json;
434 :
435 : /// 5.11.2.4: a csource subscription's csf matches the registration's
436 : /// own Context Source Properties, and its scopeQ matches the
437 : /// registration scope.
438 : #[test]
439 4 : fn csource_subscription_csf_and_scope_matching() {
440 4 : let ctx = Loader::new().core();
441 4 : let reg_a = json!({
442 4 : "id": "urn:ngsi-ld:ContextSourceRegistration:sub-a",
443 4 : "type": "ContextSourceRegistration",
444 4 : "information": [{"entities": [{"type": "Building"}]}],
445 4 : "endpoint": "http://a.example.com",
446 4 : "scope": "/Madrid/Centro"
447 : });
448 4 : let reg_b = json!({
449 4 : "id": "urn:ngsi-ld:ContextSourceRegistration:sub-b",
450 4 : "type": "ContextSourceRegistration",
451 4 : "information": [{"entities": [{"type": "Building"}]}],
452 4 : "endpoint": "http://b.example.com",
453 4 : "scope": "/Berlin"
454 : });
455 4 : let sub_csf = json!({
456 4 : "entities": [{"type": "Building"}],
457 4 : "csf": "endpoint==\"http://a.example.com\""
458 : });
459 4 : assert!(csr_matches_subscription(&sub_csf, ®_a, &ctx));
460 4 : assert!(
461 4 : !csr_matches_subscription(&sub_csf, ®_b, &ctx),
462 : "csf must exclude the other endpoint"
463 : );
464 4 : let sub_scope = json!({
465 4 : "entities": [{"type": "Building"}],
466 4 : "scopeQ": "/Madrid/#"
467 : });
468 4 : assert!(csr_matches_subscription(&sub_scope, ®_a, &ctx));
469 4 : assert!(
470 4 : !csr_matches_subscription(&sub_scope, ®_b, &ctx),
471 : "scopeQ must exclude /Berlin"
472 : );
473 4 : }
474 : }
|
