Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! /entities resource (CIM 009 6.4–6.7; operations 5.6.1–5.6.6, 5.6.17,
3 : //! 5.6.18, 5.6.19, 5.6.21, 5.7.1, 5.7.2).
4 :
5 : use crate::history::mirror_delete_entity;
6 : use crate::negotiate::*;
7 : use crate::paging::{attach_warnings, order_entities, page_params, paginate, paginate_pre};
8 : use crate::repr::{apply, parse_repr};
9 : use crate::repr::{
10 : collect_flat_beyond, compact_for, inline_join_beyond, to_geojson_collection,
11 : to_geojson_feature, MAX_JOIN_LOOKUPS,
12 : };
13 : use crate::stamp::stamp_new;
14 : use crate::state::{now_iso, AppState};
15 : use antares_jsonld::{expand_entity, is_ngsi_null, ExpandOpts};
16 : use antares_model::{NgsiError, TenantId};
17 : use antares_ql::eval::eval_q;
18 : use antares_ql::parse_q;
19 : use antares_store::CurrentStateDriverExt;
20 : use antares_store::Kind;
21 : use axum::body::Bytes;
22 : use axum::extract::{Path, State};
23 : use axum::http::{HeaderMap, StatusCode};
24 : use axum::response::{IntoResponse, Response};
25 : use serde_json::{json, Map, Value};
26 : use std::collections::HashMap;
27 :
28 : use crate::negotiate::CleanParams;
29 :
30 : use antares_model::is_meta;
31 :
32 : // ---------- temporal mirroring (auto-recording; Scorpio ENTITY-topic parity) ----------
33 : //
34 : // Append-side auto-recording (create/update/partial/merge/replace/batch) is
35 : // driven centrally off the store's change hook — see
36 : // `notify::record_temporal_change`. Only the DELETION mirrors below stay as
37 : // explicit handler calls (their typed-null deletion shape is not derivable
38 : // from a plain before/after append).
39 :
40 : // ---------- POST /entities/ (5.6.1) ----------
41 :
42 : /// 5.6.1.5: the output of a successful Create Entity is "the URI of the
43 : /// created Entity" — the resource URL carried in the Location header. The id
44 : /// is one path segment (RFC 3986 clause 3.3), so it is percent-encoded:
45 : /// spliced raw, a `#` in the id turns the rest of it into a fragment and the
46 : /// client is handed a URL addressing a different resource.
47 12250 : fn entity_location(id: &str) -> String {
48 12250 : format!(
49 : "/ngsi-ld/v1/entities/{}",
50 12250 : crate::federation::path_segment(id)
51 : )
52 12250 : }
53 :
54 : /// The selector of Entity types is input data of Delete (5.6.6.3), Merge
55 : /// (5.6.17.3) and Replace Entity (5.6.18.3), and each of those clauses
56 : /// forwards "matching input data ... to the Registration endpoint". A
57 : /// registration may cover several Entity types, so a forward that drops the
58 : /// selector lets the peer act on an entity the client's selector excluded.
59 170 : fn type_selector_query(params: &HashMap<String, String>) -> Vec<(String, String)> {
60 170 : params
61 170 : .get("type")
62 170 : .map(|v| ("type".to_owned(), v.clone()))
63 170 : .into_iter()
64 170 : .collect()
65 170 : }
66 :
67 12682 : pub async fn create_entity(
68 12682 : State(st): State<AppState>,
69 12682 : CleanParams(params): CleanParams,
70 12682 : headers: HeaderMap,
71 12682 : body: Bytes,
72 12682 : ) -> Response {
73 12682 : match create_entity_inner(&st, ¶ms, &headers, &body).await {
74 12250 : Ok(r) => r,
75 432 : Err(e) => e.into_response(),
76 : }
77 12682 : }
78 :
79 12682 : async fn create_entity_inner(
80 12682 : st: &AppState,
81 12682 : params: &HashMap<String, String>,
82 12682 : headers: &HeaderMap,
83 12682 : body: &[u8],
84 12682 : ) -> ApiResult<Response> {
85 12682 : let tenant = tenant_from(headers)?;
86 12682 : check_params(params, &["local"])?;
87 12678 : let parsed = parse_body(st, headers, body, BodyKind::Standard).await?;
88 12608 : let obj = parsed.object(NgsiError::InvalidRequest(
89 12608 : "entity document must be a JSON object".into(),
90 12608 : ))?;
91 12608 : let mut expanded = expand_entity(obj, &parsed.ctx, ExpandOpts::default())?;
92 12278 : let id = antares_jsonld::new_entity_id(&expanded)?.to_owned();
93 :
94 : // distributed create (4.3.6, 6.4.3.1)
95 12274 : let types: Option<Vec<String>> = expanded.get("type").and_then(Value::as_array).map(|a| {
96 12274 : a.iter()
97 12274 : .filter_map(Value::as_str)
98 12274 : .map(str::to_owned)
99 12274 : .collect()
100 12274 : });
101 12274 : let attr_iris: Vec<String> = expanded
102 12274 : .as_object()
103 46098 : .map(|o| o.keys().filter(|k| !is_meta(k)).cloned().collect())
104 12274 : .unwrap_or_default();
105 12274 : let spec = crate::registry::CsrSpec {
106 12274 : ids: Some(vec![id.clone()]),
107 12274 : types,
108 12274 : attrs: (!attr_iris.is_empty()).then_some(attr_iris),
109 12274 : ..Default::default()
110 12274 : };
111 : // ADR-0020: the policy seam, once per request, after expansion and
112 : // before the operation or any fan-out. Everything the engine is given
113 : // is the expanded form the store would see.
114 12274 : gate!(
115 : st, &tenant, headers, "5.6.1",
116 : ids: &[&id],
117 : types: spec.types.as_deref().unwrap_or(&[]),
118 : attrs: spec.attrs.as_deref().unwrap_or(&[]),
119 : body: Some(&expanded),
120 : )
121 12274 : .await?;
122 12266 : let regs = match crate::federation::write_plan(st, &tenant, &spec, &parsed.ctx, params, headers)
123 12266 : .await?
124 : {
125 0 : crate::federation::WritePlan::Answered(r) => return Ok(*r),
126 12266 : crate::federation::WritePlan::Forward(regs) => regs,
127 : };
128 12266 : if !regs.is_empty() {
129 : // 5.6.1.4: exclusive/redirect registrations not supporting the
130 : // Create Entity operation yield an error of type Conflict (and are
131 : // never contacted); an inclusive one is simply not forwarded.
132 160 : let (conflicts, fwd) =
133 160 : crate::federation::scoped_fragments(®s, "createEntity", obj, &parsed.ctx);
134 160 : let proxies: Vec<&crate::federation::FedReg> =
135 166 : regs.iter().filter(|r| r.is_proxy()).collect();
136 160 : let (rest, has_attrs) = crate::federation::strip_proxied(obj, &proxies, &parsed.ctx);
137 160 : let mut parts = Vec::new();
138 : // local part only when something non-proxied remains (4.3.6.3)
139 160 : if has_attrs || proxies.is_empty() {
140 106 : let mut local_exp = expand_entity(&rest, &parsed.ctx, ExpandOpts::default())?;
141 106 : stamp_new(&mut local_exp, &now_iso());
142 106 : if st
143 106 : .store
144 106 : .create(&tenant, Kind::Entity, &id, local_exp.clone())
145 106 : .await?
146 98 : {
147 98 : parts.push(crate::federation::Part::local(201, "created locally"));
148 98 : } else {
149 8 : parts.push(crate::federation::Part::local(
150 8 : 409,
151 8 : format!("entity {id} already exists"),
152 8 : ));
153 8 : }
154 54 : }
155 160 : parts.extend(conflicts);
156 160 : let ctx_url = crate::federation::ctx_link_url(headers, &parsed.ctx.source);
157 160 : for (reg, frag) in fwd {
158 78 : parts.push(
159 78 : crate::federation::forward_part(
160 78 : st,
161 78 : reqwest::Method::POST,
162 78 : format!("{}/ngsi-ld/v1/entities", reg.endpoint),
163 78 : &[],
164 78 : headers,
165 78 : &tenant,
166 78 : ®,
167 78 : &ctx_url,
168 78 : Some(frag),
169 78 : )
170 78 : .await,
171 : );
172 : }
173 160 : return Ok(crate::federation::combine(
174 160 : st.profile,
175 160 : Some(&id),
176 160 : parts,
177 160 : created(entity_location(&id), &tenant),
178 160 : &tenant,
179 160 : ));
180 12106 : }
181 :
182 12106 : stamp_new(&mut expanded, &now_iso());
183 12106 : if !st
184 12106 : .store
185 12106 : .create(&tenant, Kind::Entity, &id, expanded.clone())
186 12106 : .await?
187 : {
188 16 : return Err(NgsiError::AlreadyExists(format!("entity {id} already exists")).into());
189 12090 : }
190 12090 : Ok(created(entity_location(&id), &tenant))
191 12682 : }
192 :
193 : // ---------- GET /entities/{id} (5.7.1) ----------
194 :
195 2224 : pub async fn retrieve_entity(
196 2224 : State(st): State<AppState>,
197 2224 : Path(id): Path<String>,
198 2224 : CleanParams(params): CleanParams,
199 2224 : headers: HeaderMap,
200 2224 : ) -> Response {
201 2224 : match retrieve_entity_outer(&st, &id, ¶ms, &headers).await {
202 2056 : Ok(r) => r,
203 168 : Err(e) => e.into_response(),
204 : }
205 2224 : }
206 :
207 : /// 5.7.1.4 Retrieve Entity: the EntityMap half of the clause is the shared
208 : /// rule (`entity_maps::retrieve_with_map`); this is the retrieve it wraps.
209 2224 : async fn retrieve_entity_outer(
210 2224 : st: &AppState,
211 2224 : id: &str,
212 2224 : params: &HashMap<String, String>,
213 2224 : headers: &HeaderMap,
214 2224 : ) -> ApiResult<Response> {
215 2224 : crate::entity_map::retrieve_with_map(st, id, params, headers, false, |map| async move {
216 2222 : retrieve_entity_inner(st, id, params, headers, map.as_ref()).await
217 4444 : })
218 2224 : .await
219 2224 : }
220 :
221 2222 : async fn retrieve_entity_inner(
222 2222 : st: &AppState,
223 2222 : id: &str,
224 2222 : params: &HashMap<String, String>,
225 2222 : headers: &HeaderMap,
226 2222 : map: Option<&Value>,
227 2222 : ) -> ApiResult<Response> {
228 2222 : let tenant = tenant_from(headers)?;
229 2222 : check_params(
230 2222 : params,
231 2222 : &[
232 2222 : "attrs",
233 2222 : "pick",
234 2222 : "omit",
235 2222 : "options",
236 2222 : "format",
237 2222 : "lang",
238 2222 : "type",
239 2222 : "geometryProperty",
240 2222 : "datasetId",
241 2222 : "containedBy",
242 2222 : "join",
243 2222 : "joinLevel",
244 2222 : "local",
245 2222 : "entityMap",
246 2222 : ],
247 0 : )?;
248 2222 : let accept = parse_accept_geo(headers)?;
249 : // 5.7.1.4: geometryProperty is only meaningful for the GeoJSON
250 : // representation — any other Accept is BadRequestData
251 2218 : if params.contains_key("geometryProperty") && accept != Accept::GeoJson {
252 12 : return Err(NgsiError::BadRequestData(
253 12 : "geometryProperty requires Accept: application/geo+json (5.7.1.4)".into(),
254 12 : )
255 12 : .into());
256 2206 : }
257 2206 : let ctx = request_context(&st.loader, headers).await?;
258 2206 : let named = ReadMembers::with_ctx(st, params, &ctx);
259 2206 : let filter = gate!(
260 : st, &tenant, headers, "5.7.1",
261 : ids: &[id],
262 : types: &named.types,
263 : attrs: &named.attrs,
264 : )
265 2206 : .await?;
266 2202 : let mut repr = parse_repr(params, &ctx)?;
267 2146 : crate::repr::narrow_repr(&mut repr, &filter);
268 2146 : let join = parse_join(params)?;
269 2146 : check_linked_projection(&repr, &join)?;
270 2130 : antares_model::EntityId::new(id)?;
271 2126 : let local_doc = st.store.get(&tenant, Kind::Entity, id).await?;
272 2122 : let looped = crate::federation::via_loop(
273 2122 : headers,
274 2122 : &crate::federation::alias_for(&st.host_alias, &tenant),
275 : );
276 2122 : let fed_on = crate::federation::active(params) && !looped;
277 : // 6.3.17: abnormal distributed-GET outcomes surface as NGSILD-Warning
278 2122 : let mut warnings: Vec<String> = Vec::new();
279 2122 : if crate::federation::active(params) && looped {
280 8 : let spec = crate::registry::CsrSpec {
281 8 : ids: Some(vec![id.to_owned()]),
282 8 : ..Default::default()
283 8 : };
284 : // only a loop that suppressed a real forward is abnormal behaviour
285 8 : if !crate::federation::matching_regs(st, &tenant, &spec, &ctx, headers)
286 8 : .await?
287 8 : .is_empty()
288 8 : {
289 8 : warnings.push(crate::federation::warning(
290 8 : 199,
291 8 : &crate::federation::alias_for(&st.host_alias, &tenant),
292 8 : "a registration loop has been detected",
293 8 : ));
294 8 : }
295 2114 : }
296 2122 : let doc = if fed_on {
297 1982 : let mut sole_failure = None;
298 1982 : let fed = crate::federation::fed_retrieve(
299 1982 : st,
300 1982 : &tenant,
301 1982 : headers,
302 1982 : &ctx,
303 1982 : id,
304 1982 : map,
305 1982 : None,
306 1982 : &mut warnings,
307 1982 : &mut sole_failure,
308 1982 : )
309 1982 : .await?;
310 1980 : match local_doc {
311 1694 : Some(mut base) => {
312 3388 : for aux_pass in [false, true] {
313 3388 : for (aux, d) in &fed {
314 24 : if *aux == aux_pass {
315 12 : crate::federation::merge_docs(&mut base, d, *aux);
316 12 : }
317 : }
318 : }
319 1694 : base
320 : }
321 : None => {
322 286 : let first = fed
323 286 : .iter()
324 286 : .find(|(aux, _)| !aux)
325 286 : .map(|(_, d)| d.clone())
326 286 : .or_else(|| fed.first().map(|(_, d)| d.clone()));
327 286 : let Some(mut base) = first else {
328 : // 6.3.17 p.278: "In the case of an exclusive or redirect
329 : // registration, where all of the data is held outside of
330 : // the Context Broker and held in a single registered
331 : // source, the following errors shall be returned": 504
332 : // when it "fails to respond in time", 502 when "the
333 : // single forwarded request fails for any other reason".
334 : // Its 404 stays the plain 404 below.
335 234 : if let Some(p) = sole_failure.filter(|p| p.status != 404) {
336 10 : let mut resp = crate::federation::source_error(&p, &tenant);
337 10 : attach_warnings(&mut resp, &warnings);
338 10 : return Ok(resp);
339 224 : }
340 : // 6.3.17: abnormal distributed outcomes surface as
341 : // NGSILD-Warning even when the retrieve ends 404
342 224 : let mut resp = ApiError::from(NgsiError::ResourceNotFound(format!(
343 224 : "entity {id} not found"
344 224 : )))
345 224 : .into_response();
346 224 : attach_warnings(&mut resp, &warnings);
347 224 : echo_tenant(&tenant, &mut resp);
348 224 : return Ok(resp);
349 : };
350 104 : for aux_pass in [false, true] {
351 112 : for (aux, d) in &fed {
352 112 : if *aux == aux_pass {
353 56 : crate::federation::merge_docs(&mut base, d, *aux);
354 56 : }
355 : }
356 : }
357 52 : base
358 : }
359 : }
360 : } else {
361 140 : local_doc.ok_or_else(|| NgsiError::ResourceNotFound(format!("entity {id} not found")))?
362 : };
363 : // 5.7.1.4: no entity "whose id (URI), and where specified type, is
364 : // equivalent" — the optional ?type selector (4.17) narrows the target
365 1854 : if !crate::negotiate::matches_type_param(&doc, params, &ctx) {
366 8 : return Err(NgsiError::ResourceNotFound(format!(
367 8 : "entity {id} does not match the type selector"
368 8 : ))
369 8 : .into());
370 1846 : }
371 : // ADR-0020: an Entity outside the engine's narrowing answers the way
372 : // 5.7.1.4 answers an absent one — "If the NGSI-LD Entity does not
373 : // exist, an error of type ResourceNotFound shall be raised" — because a
374 : // refusal here would tell the caller the Entity is there.
375 1846 : if !crate::repr::policy_admits(st, &tenant, &ctx, repr.policy_reach.as_deref(), &doc).await {
376 8 : return Err(NgsiError::ResourceNotFound(format!("entity {id} not found")).into());
377 1838 : }
378 : // 5.7.1: attrs projection with no matching attribute ⇒ 404, read on the
379 : // Entity as the narrowing shows it: a hidden Attribute is absent, or
380 : // the 404 would tell the caller it is there
381 1838 : if let Some(want) = &repr.attrs {
382 12 : let shown = crate::repr::policy_view(&doc, &repr);
383 12 : if !want.iter().any(|a| shown.get(a).is_some()) {
384 6 : return Err(NgsiError::ResourceNotFound(format!(
385 6 : "entity {id} has none of the requested attributes"
386 6 : ))
387 6 : .into());
388 6 : }
389 1826 : }
390 1832 : let shaped = apply(&doc, &repr);
391 1832 : if (repr.pick.is_some() || repr.omit.is_some())
392 88 : && shaped.as_object().is_some_and(|o| o.is_empty())
393 : {
394 10 : return Err(NgsiError::ResourceNotFound(format!(
395 10 : "projection matches nothing on entity {id}"
396 10 : ))
397 10 : .into());
398 1822 : }
399 1822 : let mut payload = compact_for(&repr, &shaped, &ctx);
400 1822 : if let Some((mode, level)) = &join {
401 134 : let held = contained_by(params);
402 134 : let complete = match mode.as_str() {
403 134 : "inline" => {
404 72 : inline_join_beyond(
405 72 : st,
406 72 : &tenant,
407 72 : &ctx,
408 72 : &repr,
409 72 : &mut payload,
410 72 : &doc,
411 72 : *level,
412 72 : &held,
413 72 : &mut { MAX_JOIN_LOOKUPS },
414 72 : fed_on.then_some(headers),
415 72 : )
416 72 : .await
417 : }
418 62 : "flat" => {
419 62 : let mut linked = std::collections::BTreeMap::new();
420 62 : let complete = collect_flat_beyond(
421 62 : st,
422 62 : &tenant,
423 62 : &ctx,
424 62 : &repr,
425 62 : &doc,
426 62 : *level,
427 62 : &mut linked,
428 62 : &held,
429 62 : &mut { MAX_JOIN_LOOKUPS },
430 62 : )
431 62 : .await;
432 62 : if !linked.is_empty() {
433 56 : let mut arr = vec![payload];
434 84 : for (_, (ldoc, lrepr)) in linked {
435 84 : arr.push(compact_for(&lrepr, &apply(&ldoc, &lrepr), &ctx));
436 84 : }
437 56 : payload = Value::Array(arr);
438 6 : }
439 62 : complete
440 : }
441 0 : _ => true,
442 : };
443 134 : if !complete {
444 8 : warnings.push(crate::federation::warning(
445 8 : 199,
446 8 : &crate::federation::alias_for(&st.host_alias, &tenant),
447 8 : "the linked entity retrieval was truncated",
448 8 : ));
449 126 : }
450 1688 : }
451 1822 : let payload = if accept == Accept::GeoJson {
452 42 : to_geojson_feature(payload, params.get("geometryProperty"))
453 : } else {
454 1780 : payload
455 : };
456 1822 : let mut resp = respond_prefer(StatusCode::OK, payload, &ctx, accept, &tenant, headers);
457 1822 : attach_warnings(&mut resp, &warnings);
458 1822 : filter.mark(&st.host_alias, resp.headers_mut());
459 1822 : Ok(resp)
460 2222 : }
461 :
462 : /// 5.7.1.4 / 5.7.2.4: a `{…}` projection selects into Linked Entities —
463 : /// it must be requested via join, and may not select deeper than joinLevel.
464 4426 : pub(crate) fn check_linked_projection(
465 4426 : repr: &crate::repr::Repr,
466 4426 : join: &Option<(String, usize)>,
467 4426 : ) -> ApiResult<()> {
468 4426 : let depth = repr
469 4426 : .pick
470 4426 : .as_deref()
471 4426 : .map(crate::repr::proj_depth)
472 4426 : .unwrap_or(0)
473 4426 : .max(
474 4426 : repr.omit
475 4426 : .as_deref()
476 4426 : .map(crate::repr::proj_depth)
477 4426 : .unwrap_or(0),
478 : );
479 4426 : if depth == 0 {
480 4310 : return Ok(());
481 116 : }
482 96 : match join {
483 96 : Some((mode, level)) if mode != "@none" => {
484 96 : if depth > *level {
485 12 : return Err(NgsiError::BadRequestData(format!(
486 12 : "projected attribute depth {depth} exceeds joinLevel {level} (5.7.1.4/5.7.2.4)"
487 12 : ))
488 12 : .into());
489 84 : }
490 84 : Ok(())
491 : }
492 20 : _ => Err(NgsiError::BadRequestData(
493 20 : "projection uses Linked Entity selection but join is not specified (5.7.1.4/5.7.2.4)"
494 20 : .into(),
495 20 : )
496 20 : .into()),
497 : }
498 4426 : }
499 :
500 : /// join/joinLevel params (4.5.23). Returns (mode, level).
501 4434 : pub fn parse_join(params: &HashMap<String, String>) -> ApiResult<Option<(String, usize)>> {
502 4434 : let Some(mode) = params.get("join") else {
503 4094 : return Ok(None);
504 : };
505 340 : if !["inline", "flat", "@none"].contains(&mode.as_str()) {
506 4 : return Err(NgsiError::BadRequestData(format!("invalid join {mode:?}")).into());
507 336 : }
508 336 : let level = match params.get("joinLevel") {
509 312 : Some(l) => l
510 312 : .parse::<usize>()
511 312 : .ok()
512 : // Bounded traversal depth
513 312 : .filter(|l| *l >= 1 && *l <= crate::bounds::MAX_JOIN_LEVEL)
514 312 : .ok_or_else(|| {
515 4 : NgsiError::BadRequestData(format!(
516 4 : "invalid joinLevel {l:?} (1..={})",
517 4 : crate::bounds::MAX_JOIN_LEVEL
518 4 : ))
519 4 : })?,
520 24 : None => 1,
521 : };
522 332 : if mode == "@none" {
523 4 : return Ok(None);
524 328 : }
525 328 : Ok(Some((mode.clone(), level)))
526 4434 : }
527 :
528 : /// Table 6.4.3.2-1 `containedBy`: "List of entity ids which have previously
529 : /// been encountered whilst retrieving the Entity Graph. Only applicable if
530 : /// joinLevel is present." They are already in the graph the client is
531 : /// assembling, so 4.5.23.1's "avoid ... duplicates or loops" counts them as
532 : /// resolved and the walk does not follow them again.
533 308 : pub fn contained_by(params: &HashMap<String, String>) -> Vec<String> {
534 308 : params
535 308 : .get("containedBy")
536 308 : .map(|s| {
537 4 : s.split(',')
538 4 : .map(str::trim)
539 4 : .filter(|t| !t.is_empty())
540 4 : .map(str::to_owned)
541 4 : .collect()
542 4 : })
543 308 : .unwrap_or_default()
544 308 : }
545 :
546 : // ---------- GET /entities/ (5.7.2) ----------
547 :
548 2278 : pub async fn query_entities(
549 2278 : State(st): State<AppState>,
550 2278 : CleanParams(params): CleanParams,
551 2278 : headers: HeaderMap,
552 2278 : ) -> Response {
553 2278 : match query_entities_outer(&st, params, &headers).await {
554 1966 : Ok(r) => r,
555 312 : Err(e) => e.into_response(),
556 : }
557 2278 : }
558 :
559 : /// 5.5.14 / 5.5.9.3: a query referencing an EntityMap (NGSILD-EntityMap
560 : /// request header, 6.4.3.2-2) is fixed to the map's Entities; the filters
561 : /// are re-checked at processing time and local entries that no longer match
562 : /// are removed from the map by its creator. An expired or unknown map means
563 : /// "no inference can be made … a new one shall be created".
564 2278 : async fn query_entities_outer(
565 2278 : st: &AppState,
566 2278 : mut params: HashMap<String, String>,
567 2278 : headers: &HeaderMap,
568 2278 : ) -> ApiResult<Response> {
569 2278 : let tenant = tenant_from(headers)?;
570 2278 : let q_ast = params.get("q").map(|q| parse_q(q)).transpose()?;
571 2270 : let named = ReadMembers::of(st, headers, ¶ms).await;
572 2270 : let filter = gate!(
573 : st, &tenant, headers, "5.7.2",
574 : q: q_ast.as_ref(),
575 : scope_q: params.get("scopeQ").map(String::as_str),
576 : types: &named.types,
577 : attrs: &named.attrs,
578 : geo: named.geo.as_ref(),
579 : )
580 2270 : .await?;
581 2266 : let Some(map_ref) = single_header(headers, "NGSILD-EntityMap")? else {
582 2218 : return query_entities_inner(st, ¶ms, headers, &filter).await;
583 : };
584 : // 5.7.2.4: an unknown parameter and a too-wide query are BadRequestData
585 : // for this request, whether or not it carries an EntityMap reference —
586 : // and the paged fetch below walks the whole map, locally and forwarded,
587 : // before the inner call would reach these same two checks.
588 44 : check_params(¶ms, crate::negotiate::QUERY_PARAMS)?;
589 44 : if !qualifies_non_wide(¶ms, q_ast.as_ref()) {
590 0 : return Err(NgsiError::BadRequestData(
591 0 : "query needs at least one of type, attrs, q, georel (5.7.2)".into(),
592 0 : )
593 0 : .into());
594 44 : }
595 44 : let map_id = map_ref.rsplit('/').next().unwrap_or(&map_ref).to_owned();
596 44 : let Some(mut map) = crate::entity_map::map_if_accessible(st, &tenant, headers, &map_id).await
597 : else {
598 : // 5.5.14: expired or inaccessible → a new EntityMap is created
599 6 : params.insert("entityMap".into(), "true".into());
600 6 : return query_entities_inner(st, ¶ms, headers, &filter).await;
601 : };
602 38 : let ctx = request_context(&st.loader, headers).await?;
603 : // a request that references a live map does not create a new one
604 38 : params.remove("entityMap");
605 38 : let (offset, limit, count) = page_params(st, ¶ms)?;
606 : // pagination links carry the ORIGINAL query, never the page's id list
607 38 : let link_params = params.clone();
608 38 : let accept = parse_accept_geo(headers)?;
609 :
610 : // 5.5.9.3 paged fetch: the map fixes the candidate id set; candidates
611 : // are fetched (locally + forwarded) chunk by chunk, "filters shall be
612 : // rechecked before returning results" per chunk, and visited entries
613 : // that no longer match are removed from the map — "Entities not or no
614 : // longer fitting the query shall be removed from the Entity map during
615 : // pagination". Pruning is judgeable only for "@none" (local) entries: a
616 : // remote-backed id may merely have an unreachable source right now
617 : // (5.5.14). Memory per request is O(chunk), never O(map) — the reason
618 : // EntityMaps exist for the distributed case. count=true walks every
619 : // candidate (the total needs each id checked), still chunk-bounded.
620 38 : let ids: Vec<String> = crate::entity_map::candidate_ids(&map, ¶ms);
621 38 : let narrow = crate::repr::narrowing_reading(&filter, ¶ms, &ctx)?;
622 38 : let looped = crate::federation::via_loop(
623 38 : headers,
624 38 : &crate::federation::alias_for(&st.host_alias, &tenant),
625 : );
626 38 : let chunk_size = limit.max(20);
627 38 : let mut page_ids: Vec<String> = Vec::new();
628 38 : let mut stale: Vec<String> = Vec::new();
629 38 : let (mut skipped, mut total, mut more, mut visited) = (0usize, 0usize, false, 0usize);
630 38 : for chunk in ids.chunks(chunk_size) {
631 38 : let mut p = params.clone();
632 38 : p.remove("limit");
633 38 : p.remove("offset");
634 38 : p.remove("count");
635 38 : p.insert("id".into(), chunk.join(","));
636 : // the final page fetch below re-surfaces the same peers' warnings
637 38 : let mut chunk_warnings = Vec::new();
638 38 : let fed = if crate::federation::active(&p) && !looped {
639 38 : crate::federation::fed_query(st, &tenant, headers, &ctx, &p, &mut chunk_warnings)
640 38 : .await?
641 : } else {
642 0 : Vec::new()
643 : };
644 38 : let docs = filter_entities_fed(st, &tenant, &p, &ctx, fed, Some(&narrow)).await?;
645 38 : let matched: std::collections::HashSet<&str> = docs
646 38 : .iter()
647 172 : .filter_map(|d| d.get("id").and_then(Value::as_str))
648 38 : .collect();
649 38 : if let Some(emap) = map.get_mut("entityMap").and_then(Value::as_object_mut) {
650 180 : for id in chunk {
651 180 : let local_only = emap
652 180 : .get(id)
653 180 : .and_then(Value::as_array)
654 180 : .is_some_and(|a| matches!(a.as_slice(), [only] if only == "@none"));
655 180 : if local_only && !matched.contains(id.as_str()) {
656 8 : emap.remove(id);
657 8 : stale.push(id.clone());
658 172 : }
659 : }
660 0 : }
661 38 : visited = visited.saturating_add(chunk.len());
662 180 : for id in chunk {
663 180 : if !matched.contains(id.as_str()) {
664 8 : continue;
665 172 : }
666 172 : total = total.saturating_add(1);
667 172 : if skipped < offset {
668 24 : skipped = skipped.saturating_add(1);
669 148 : } else if page_ids.len() < limit {
670 76 : page_ids.push(id.clone());
671 80 : } else {
672 72 : more = true;
673 72 : }
674 : }
675 38 : if !count && page_ids.len() == limit {
676 : // page full — next exists if an extra match was seen or
677 : // unvisited candidates remain ("pages shall always be filled to
678 : // the maximum, as long as Entities are available")
679 14 : more = more || visited < ids.len();
680 14 : break;
681 24 : }
682 : }
683 38 : if count {
684 6 : more = total > offset.saturating_add(limit);
685 32 : }
686 38 : crate::entity_map::map_prune(st, &tenant, &map_id, &stale).await?;
687 : // fix the final fetch to exactly the page's survivors (5.5.14: an empty
688 : // set is fixed to nothing); one extra page-sized fetch keeps the whole
689 : // repr pipeline shared instead of forked
690 38 : params.insert(
691 38 : "id".into(),
692 38 : if page_ids.is_empty() {
693 0 : "urn:ngsi-ld:entitymap:empty".to_owned()
694 : } else {
695 38 : page_ids.join(",")
696 : },
697 : );
698 38 : params.remove("offset");
699 38 : params.remove("count");
700 38 : if limit == 0 {
701 0 : // count-only shape: the inner default limit is irrelevant against
702 0 : // the empty id sentinel
703 0 : params.remove("limit");
704 38 : }
705 : // The narrowing reaches the answer here; the map's own contents are
706 : // still the ones the query built, which is what P5's per-subject map is
707 : // for.
708 38 : let mut resp = query_entities_inner(st, ¶ms, headers, &filter).await?;
709 : // "The location of the EntityMap used in the query operation is
710 : // returned in the response" (6.4.3.2-2)
711 38 : if let Ok(v) = format!("/ngsi-ld/v1/entityMaps/{map_id}").parse() {
712 38 : resp.headers_mut().insert("NGSILD-EntityMap", v);
713 38 : }
714 38 : if count {
715 6 : if let Ok(v) = total.to_string().parse() {
716 6 : resp.headers_mut().insert("NGSILD-Results-Count", v);
717 6 : }
718 32 : }
719 : // 6.3.10 links from the original query — the inner call saw offset 0
720 : // over exactly one page, so it emitted none
721 76 : for (off, rel, cond) in [
722 38 : (offset.saturating_add(limit), "next", more && limit > 0),
723 38 : (offset.saturating_sub(limit.max(1)), "prev", offset > 0),
724 : ] {
725 76 : if !cond {
726 50 : continue;
727 26 : }
728 26 : let mut qp: Vec<String> = link_params
729 26 : .iter()
730 64 : .filter(|(k, _)| k.as_str() != "offset")
731 58 : .map(|(k, v)| format!("{k}={}", crate::paging::query_value(v)))
732 26 : .collect();
733 26 : qp.push(format!("offset={off}"));
734 26 : qp.sort(); // deterministic order — the suite string-compares links
735 26 : let ty = match accept {
736 0 : Accept::LdJson => ";type=\"application/ld+json\"",
737 26 : Accept::Json => ";type=\"application/json\"",
738 0 : Accept::GeoJson => ";type=\"application/geo+json\"",
739 : };
740 26 : if let Ok(v) = format!("</ngsi-ld/v1/entities?{}>; rel=\"{rel}\"{ty}", qp.join("&")).parse()
741 26 : {
742 26 : resp.headers_mut().append(axum::http::header::LINK, v);
743 26 : }
744 : }
745 38 : Ok(resp)
746 2278 : }
747 :
748 2262 : async fn query_entities_inner(
749 2262 : st: &AppState,
750 2262 : params: &HashMap<String, String>,
751 2262 : headers: &HeaderMap,
752 2262 : filter: &std::sync::Arc<crate::policy::Filter>,
753 2262 : ) -> ApiResult<Response> {
754 2262 : let tenant = tenant_from(headers)?;
755 2262 : check_params(params, crate::negotiate::QUERY_PARAMS)?;
756 2210 : let accept = parse_accept_geo(headers)?;
757 2206 : let ctx = request_context(&st.loader, headers).await?;
758 :
759 : // 5.7.2.4 a-e: id/idPattern alone are NOT sufficient, and the attrs
760 : // list / q must include "at least one non-system Attribute" to qualify.
761 : // The judgement is about what the CLIENT asked for, so it reads the
762 : // request's own `q` — a policy condition is not the client's filter and
763 : // does not make a wide query narrow (ADR-0020).
764 2206 : let has_filter = qualifies_non_wide(
765 2206 : params,
766 2206 : params.get("q").map(|q| parse_q(q)).transpose()?.as_ref(),
767 : );
768 : // what the client's own query reads, before the engine's q joins it
769 : //
770 2206 : let client_params = params;
771 : // Everything below reads the narrowed query: the store push-down, the
772 : // local re-check 5.7.2.4 runs over merged results, and the query the
773 : // request is forwarded with.
774 2206 : let narrowed = filter.narrow_params(params, &ctx)?;
775 2198 : let params = &narrowed;
776 2198 : let q_ast = params
777 2198 : .get("q")
778 2198 : .map(|q| antares_ql::parse_narrowed_q(q))
779 2198 : .transpose()?;
780 : // 5.7.2.4 validation bullets (p.201), in the spec's own order.
781 2198 : star_type_local(params)?;
782 2190 : if params.contains_key("geometryProperty") && accept != Accept::GeoJson {
783 8 : return Err(NgsiError::BadRequestData(
784 8 : "geometryProperty requires Accept: application/geo+json (5.7.2.4)".into(),
785 8 : )
786 8 : .into());
787 2182 : }
788 : // "If the ordering parameter is present and the execution of the operation
789 : // is not limited to the local scope then BadRequestData" — reinforced by
790 : // 4.23.1: "Sort ordering is never applied to distributed operations."
791 : // The subject is the EXECUTION: a query nothing federates to runs locally
792 : // regardless of `local=true`, which is why this asks would_federate rather
793 : // than active (ETSI 019_19 orders without local).
794 2182 : crate::paging::check_collation(params)?;
795 2174 : if params.contains_key("orderBy")
796 282 : && crate::federation::would_federate(st, &tenant, &ctx, params, headers).await?
797 : {
798 8 : return Err(NgsiError::BadRequestData(
799 8 : "orderBy requires local scope — ordering is never applied to \
800 8 : distributed operations (5.7.2.4, 4.23.1)"
801 8 : .into(),
802 8 : )
803 8 : .into());
804 2166 : }
805 2166 : if !has_filter {
806 32 : return Err(NgsiError::BadRequestData(
807 32 : "query needs at least one of type, attrs, q, georel (5.7.2)".into(),
808 32 : )
809 32 : .into());
810 2134 : }
811 :
812 2134 : let mut repr = parse_repr(params, &ctx)?;
813 2082 : crate::repr::narrow_repr(&mut repr, filter);
814 2082 : crate::repr::note_reads(&mut repr, client_params, &ctx)?;
815 2082 : let join = parse_join(params)?;
816 2076 : check_linked_projection(&repr, &join)?;
817 : // 5.7.2.4: filter conditions using Linked Entity attributes need join,
818 : // and their hop depth may not exceed joinLevel ("too deep query")
819 2060 : let link_depth = q_ast
820 2060 : .as_ref()
821 2060 : .map(antares_ql::QNode::max_link_depth)
822 2060 : .unwrap_or(0);
823 2060 : if link_depth > 0 {
824 32 : match &join {
825 32 : Some((mode, level)) if mode != "@none" => {
826 32 : if link_depth > *level {
827 8 : return Err(NgsiError::BadRequestData(format!(
828 8 : "linked attribute query depth {link_depth} exceeds joinLevel {level} \
829 8 : (5.7.2.4 — too deep query)"
830 8 : ))
831 8 : .into());
832 24 : }
833 : }
834 : _ => {
835 8 : return Err(NgsiError::BadRequestData(
836 8 : "q references Linked Entity attributes but join is not specified \
837 8 : (5.7.2.4 — too deep query)"
838 8 : .into(),
839 8 : )
840 8 : .into());
841 : }
842 : }
843 2020 : }
844 : // 5.7.2.4: a syntactically invalid context source filter is 400.
845 2044 : if let Some(csf) = params.get("csf") {
846 24 : parse_q(csf)?;
847 2020 : }
848 : // 6.3.17: abnormal distributed-GET outcomes surface as NGSILD-Warning
849 2032 : let mut warnings: Vec<String> = Vec::new();
850 2032 : let looped = crate::federation::via_loop(
851 2032 : headers,
852 2032 : &crate::federation::alias_for(&st.host_alias, &tenant),
853 : );
854 2032 : let fed = if crate::federation::active(params) && !looped {
855 1906 : crate::federation::fed_query(st, &tenant, headers, &ctx, params, &mut warnings).await?
856 : } else {
857 126 : if crate::federation::active(params)
858 2 : && looped
859 2 : && crate::federation::would_federate(st, &tenant, &ctx, params, headers).await?
860 2 : {
861 2 : warnings.push(crate::federation::warning(
862 2 : 199,
863 2 : &crate::federation::alias_for(&st.host_alias, &tenant),
864 2 : "a registration loop has been detected",
865 2 : ));
866 124 : }
867 126 : Vec::new()
868 : };
869 : // Pushdown gates: pagination per page_pushdown_allowed (no federation
870 : // candidates, no idPattern, no orderBy). Projection additionally excludes
871 : // join (linked-entity walks read page docs) and GeoJSON output.
872 2028 : let (p_offset, p_limit, _) = page_params(st, params)?;
873 2020 : let push_page = page_pushdown_allowed(fed.is_empty(), params);
874 2020 : let push_proj = join.is_none() && accept != Accept::GeoJson;
875 : // an ordered read sorts on what its keys read and reads its page whole
876 : // afterwards
877 2020 : let order_keys = params
878 2020 : .get("orderBy")
879 2020 : .map(|spec| crate::paging::order_heads(spec, &ctx));
880 2020 : let pushdown = Pushdown {
881 2020 : page: push_page.then_some((p_offset, p_limit)),
882 2020 : proj: push_proj.then_some(&repr),
883 2020 : order_keys: order_keys.as_deref(),
884 2020 : };
885 1972 : let mut filtered =
886 2020 : filter_entities_paged(st, &tenant, params, &ctx, fed, pushdown, Some(&repr)).await?;
887 : // the page is read back by `id`, a comma-separated list
888 3296 : let listable = |d: &Value| {
889 2888 : d.get("id")
890 2888 : .and_then(Value::as_str)
891 2888 : .is_some_and(|id| !id.contains(','))
892 2888 : };
893 1972 : if filtered.trimmed && !filtered.docs.iter().all(listable) {
894 80 : filtered = filter_entities_paged(
895 80 : st,
896 80 : &tenant,
897 80 : params,
898 80 : &ctx,
899 80 : Vec::new(),
900 80 : Pushdown {
901 80 : order_keys: None,
902 80 : ..pushdown
903 80 : },
904 80 : Some(&repr),
905 80 : )
906 80 : .await?;
907 1892 : }
908 1972 : let trimmed = filtered.trimmed;
909 1972 : let mut matches = filtered.docs;
910 1972 : if let Some(spec) = params.get("orderBy") {
911 274 : crate::repr::order_as_shown(&mut matches, Some(&repr), |m| {
912 274 : order_entities(m, spec, params, &ctx)
913 274 : })?;
914 1698 : }
915 1966 : let keyset = push_page && crate::paging::keyset_allowed(params);
916 1966 : let (page, count_hdr, links) = if keyset && params.contains_key(crate::paging::CURSOR) {
917 : // the rows start after the cursor's id, at the client's offset
918 4 : let total = match filtered.paged {
919 4 : true => filtered.total.unwrap_or(matches.len()),
920 0 : false => matches.len(),
921 : };
922 4 : matches.truncate(p_limit);
923 4 : let total = p_offset.saturating_add(total);
924 4 : paginate_pre(st, params, matches, "/ngsi-ld/v1/entities", total)?
925 1962 : } else if filtered.paged {
926 1398 : let total = filtered.total.unwrap_or(matches.len());
927 1398 : paginate_pre(st, params, matches, "/ngsi-ld/v1/entities", total)?
928 : } else {
929 564 : paginate(st, params, matches, "/ngsi-ld/v1/entities")?
930 : };
931 1966 : let page = match trimmed {
932 184 : true => whole_page(st, &tenant, params, &ctx, &repr, page).await?,
933 1782 : false => page,
934 : };
935 1966 : let links = match keyset {
936 1274 : true => crate::paging::with_next_cursor(
937 1274 : links,
938 1274 : &crate::paging::view_of(st),
939 1274 : tenant.as_str(),
940 1274 : params,
941 1274 : page.last()
942 1274 : .and_then(|d| d.get("id"))
943 1274 : .and_then(Value::as_str),
944 1274 : p_offset.saturating_add(p_limit),
945 : ),
946 692 : false => links,
947 : };
948 :
949 : // each rendered Entity beside the one it was rendered from, which the
950 : // inline join reads its Relationships off
951 1966 : let (rendered_from, mut payload): (Vec<&Value>, Vec<Value>) = page
952 1966 : .iter()
953 6516 : .filter_map(|doc| {
954 6516 : let shaped = apply(doc, &repr);
955 : // pick projections that match nothing drop the entity entirely
956 6516 : if repr.pick.is_some() && shaped.as_object().is_some_and(|o| o.is_empty()) {
957 4 : return None;
958 6512 : }
959 6512 : Some((doc, compact_for(&repr, &shaped, &ctx)))
960 6516 : })
961 1966 : .unzip();
962 1966 : if let Some((mode, level)) = &join {
963 156 : let mut complete = true;
964 156 : let held = contained_by(params);
965 : // 4.5.23.1 bounds the WIDTH of the retrieval per request, so one
966 : // allowance is spent across the whole page: minting a fresh budget per
967 : // payload Entity multiplied it by the page size, and a page of
968 : // max_limit densely linked Entities bought MAX_JOIN_LOOKUPS lookups
969 : // each.
970 156 : let mut budget = MAX_JOIN_LOOKUPS;
971 156 : match mode.as_str() {
972 156 : "inline" => {
973 140 : for (p, stored) in payload.iter_mut().zip(&rendered_from) {
974 140 : complete &= inline_join_beyond(
975 140 : st,
976 140 : &tenant,
977 140 : &ctx,
978 140 : &repr,
979 140 : p,
980 140 : stored,
981 140 : *level,
982 140 : &held,
983 140 : &mut budget,
984 140 : crate::federation::active(params).then_some(headers),
985 140 : )
986 140 : .await;
987 : }
988 : }
989 64 : "flat" => {
990 64 : let mut linked = std::collections::BTreeMap::new();
991 116 : for doc in &page {
992 116 : complete &= collect_flat_beyond(
993 116 : st,
994 116 : &tenant,
995 116 : &ctx,
996 116 : &repr,
997 116 : doc,
998 116 : *level,
999 116 : &mut linked,
1000 116 : &held,
1001 116 : &mut budget,
1002 116 : )
1003 116 : .await;
1004 : }
1005 116 : let page_ids: Vec<&str> = page.iter().filter_map(|d| d["id"].as_str()).collect();
1006 148 : for (id, (ldoc, lrepr)) in linked {
1007 148 : if !page_ids.contains(&id.as_str()) {
1008 148 : payload.push(compact_for(&lrepr, &apply(&ldoc, &lrepr), &ctx));
1009 148 : }
1010 : }
1011 : }
1012 0 : _ => {}
1013 : }
1014 156 : if !complete {
1015 4 : warnings.push(crate::federation::warning(
1016 4 : 199,
1017 4 : &crate::federation::alias_for(&st.host_alias, &tenant),
1018 4 : "the linked entity retrieval was truncated",
1019 4 : ));
1020 152 : }
1021 1810 : }
1022 1966 : let mut resp = if accept == Accept::GeoJson {
1023 24 : let fc = to_geojson_collection(payload, params.get("geometryProperty"));
1024 24 : respond_prefer(StatusCode::OK, fc, &ctx, accept, &tenant, headers)
1025 : } else {
1026 1942 : crate::negotiate::respond_list(StatusCode::OK, payload, &ctx, accept, &tenant)
1027 : };
1028 1966 : attach_paging(&mut resp, count_hdr, &links);
1029 1966 : attach_warnings(&mut resp, &warnings);
1030 : // 6.4.3.2: entityMap=true — the EntityMap for this query is (re)created;
1031 : // the response carries NGSILD-EntityMap and 201 Created.
1032 1966 : if params.get("entityMap").map(String::as_str) == Some("true") {
1033 28 : let map = build_query_map(st, &tenant, headers, &ctx, params, filter).await?;
1034 28 : *resp.status_mut() = StatusCode::CREATED;
1035 28 : if let Some(id) = map.get("id").and_then(Value::as_str) {
1036 28 : if let Ok(v) = format!("/ngsi-ld/v1/entityMaps/{id}").parse() {
1037 28 : resp.headers_mut().insert("NGSILD-EntityMap", v);
1038 28 : }
1039 0 : }
1040 1938 : }
1041 1966 : filter.mark(&st.host_alias, resp.headers_mut());
1042 1966 : Ok(resp)
1043 2262 : }
1044 :
1045 : /// Table 6.4.3.2-1: `type=*` implies local scope and shall not be combined
1046 : /// with `local=false`.
1047 3628 : pub(crate) fn star_type_local(params: &HashMap<String, String>) -> Result<(), NgsiError> {
1048 3628 : if params.get("type").map(String::as_str) == Some("*")
1049 24 : && params.get("local").map(String::as_str) == Some("false")
1050 : {
1051 12 : return Err(NgsiError::BadRequestData(
1052 12 : "type=* implies local and shall not be combined with local=false \
1053 12 : (Table 6.4.3.2-1)"
1054 12 : .into(),
1055 12 : ));
1056 3616 : }
1057 3616 : Ok(())
1058 3628 : }
1059 :
1060 : /// 5.7.2.4 / 5.7.4.4 / 5.14.4.4 a-e: a query qualifies (is not "too wide")
1061 : /// only with a type selector, an attrs list or q naming at least one
1062 : /// non-system Attribute, a geoquery, or local scope.
1063 3900 : pub(crate) fn qualifies_non_wide(
1064 3900 : params: &HashMap<String, String>,
1065 3900 : q_ast: Option<&antares_ql::QNode>,
1066 3900 : ) -> bool {
1067 3900 : let attrs_qualify = params.get("attrs").is_some_and(|a| {
1068 192 : a.split(',')
1069 196 : .any(|n| antares_ql::is_non_system_attr(n.trim()))
1070 192 : });
1071 3900 : let q_qualifies = q_ast.is_some_and(|ast| {
1072 758 : ast.attribute_paths()
1073 758 : .iter()
1074 758 : .any(|h| antares_ql::is_non_system_attr(h))
1075 758 : });
1076 3900 : params.contains_key("type")
1077 212 : || attrs_qualify
1078 160 : || q_qualifies
1079 116 : || params.contains_key("georel")
1080 112 : || params.get("local").map(String::as_str) == Some("true")
1081 3900 : }
1082 :
1083 : /// Same, with federated candidate docs merged in before filtering (4.3.6.7).
1084 264 : pub async fn filter_entities_fed(
1085 264 : st: &AppState,
1086 264 : tenant: &TenantId,
1087 264 : params: &HashMap<String, String>,
1088 264 : ctx: &antares_jsonld::Context,
1089 264 : fed: Vec<(bool, Value)>,
1090 264 : narrow: Option<&crate::repr::Repr>,
1091 264 : ) -> ApiResult<Vec<Value>> {
1092 : Ok(
1093 264 : filter_entities_paged(st, tenant, params, ctx, fed, Pushdown::default(), narrow)
1094 264 : .await?
1095 : .docs,
1096 : )
1097 264 : }
1098 :
1099 : /// What the paged variant produced. `paged` = the store already applied
1100 : /// ORDER BY id + LIMIT/OFFSET (and `total` is the pre-LIMIT match count), so
1101 : /// the caller must NOT slice again.
1102 : pub struct Filtered {
1103 : pub docs: Vec<Value>,
1104 : pub paged: bool,
1105 : pub total: Option<usize>,
1106 : /// How far a chunked walk steps past this page, in the unit its offset
1107 : /// counts. With a pushed page that is the store rows the SQL page read,
1108 : /// before the evaluator below dropped any — `docs` undercounts it
1109 : /// whenever idPattern (invisible to the store filter) removed rows. Where
1110 : /// the offset counts matches instead (a window or a count the loop
1111 : /// decides), it is the matches of the page: stepping by the rows read
1112 : /// there skipped matches, and a window of rows that never match kept the
1113 : /// walk going forever.
1114 : pub rows: usize,
1115 : /// The store decided the match set and kept only the core members and
1116 : /// [`Pushdown::order_keys`] of each document: enough to order and page
1117 : /// it, and the page is then read whole by id.
1118 : pub trimmed: bool,
1119 : }
1120 :
1121 : /// Whether the store may be asked to project members away. Only when its
1122 : /// answer is the final answer: 5.7.2.4 applies the query, geoquery, Scope
1123 : /// query and Attribute filters after remote parts have been aggregated, so
1124 : /// with federated candidates present those filters still run over documents
1125 : /// the store would have stripped; and 4.23 orders Entities by the value of
1126 : /// the ordering member, which a projection may have removed — leaving the
1127 : /// comparator nothing to compare and the client id order.
1128 1932 : fn proj_pushdown_allowed(fed_is_empty: bool, params: &HashMap<String, String>) -> bool {
1129 1932 : fed_is_empty && !params.contains_key("orderBy")
1130 1932 : }
1131 :
1132 : /// Whether `q` reads only Attributes that every rule of a type in `groups`
1133 : /// shows and the narrowing's own projection keeps, in the form the store
1134 : /// compiles exactly: one Attribute per term, no Linked Entity hop, no system
1135 : /// member. An Entity of those types shows at least what one such rule shows,
1136 : /// so on those Attributes it answers `q` as stored and as shown alike.
1137 32 : fn reads_only_shown(
1138 32 : q: &antares_ql::QNode,
1139 32 : rules: &crate::policy::Filter,
1140 32 : groups: &[Vec<String>],
1141 32 : narrow: Option<&crate::repr::Repr>,
1142 32 : expand: &dyn Fn(&str) -> String,
1143 32 : ) -> bool {
1144 32 : antares_ql::sql::compile_q(q, "entity", 1, expand).is_some()
1145 28 : && shows_alike(q.attribute_paths(), rules, groups, narrow, expand)
1146 32 : }
1147 :
1148 : /// Whether every rule of a type in `groups` shows each of `names` (terms) and
1149 : /// the narrowing's own projection keeps it: on those Attributes the Entity as
1150 : /// stored and as shown are alike.
1151 36 : fn shows_alike<'n>(
1152 36 : names: impl IntoIterator<Item = &'n str>,
1153 36 : rules: &crate::policy::Filter,
1154 36 : groups: &[Vec<String>],
1155 36 : narrow: Option<&crate::repr::Repr>,
1156 36 : expand: &dyn Fn(&str) -> String,
1157 36 : ) -> bool {
1158 36 : if !rules.omit_by_type.is_empty() {
1159 0 : return false;
1160 36 : }
1161 : // every rule naming a type the request selects
1162 36 : let selected: Vec<&crate::policy::Rule> = rules
1163 36 : .rules
1164 36 : .iter()
1165 58 : .filter(|r| {
1166 58 : r.types
1167 58 : .iter()
1168 76 : .any(|t| groups.iter().flatten().any(|g| g == t))
1169 58 : })
1170 36 : .collect();
1171 72 : let heads = |nodes: &Option<Vec<crate::repr::ProjNode>>| {
1172 72 : nodes
1173 72 : .as_ref()
1174 72 : .map(|n| n.iter().map(|n| n.iri.clone()).collect::<Vec<_>>())
1175 72 : };
1176 36 : let pick = narrow.and_then(|r| heads(&r.policy_pick));
1177 36 : let omit = narrow
1178 36 : .and_then(|r| heads(&r.policy_omit))
1179 36 : .unwrap_or_default();
1180 36 : !selected.is_empty()
1181 36 : && names.into_iter().all(|name| {
1182 36 : let iri = expand(name);
1183 36 : antares_ql::is_non_system_attr(name)
1184 36 : && !omit.contains(&iri)
1185 36 : && pick.as_ref().is_none_or(|p| p.contains(&iri))
1186 36 : && selected
1187 36 : .iter()
1188 54 : .all(|r| r.every_attribute || r.attributes.contains(&iri))
1189 36 : })
1190 36 : }
1191 :
1192 : /// The client's type selection (OR of AND groups, `None` = any type) with
1193 : /// one of the rules' types required in every group: `(a;b)|c` under rules
1194 : /// of `T` and `U` selects `(a;b;T)|(a;b;U)|(c;T)|(c;U)`, and a group that
1195 : /// already names one of them stays as it is (`(a;T)` selects no more than
1196 : /// `(a;T;U)`). `None` when the rules name no type, so a selection is never
1197 : /// dropped into "any".
1198 124 : fn judged_types<'a>(
1199 124 : client: Option<&'a [Vec<String>]>,
1200 124 : rules: &crate::policy::Filter,
1201 124 : ) -> Option<std::borrow::Cow<'a, [Vec<String>]>> {
1202 124 : let mut union: Vec<&String> = rules.rules.iter().flat_map(|r| &r.types).collect();
1203 124 : union.sort();
1204 124 : union.dedup();
1205 124 : if union.is_empty() {
1206 10 : return None;
1207 114 : }
1208 246 : let named = |g: &Vec<String>| g.iter().any(|t| union.contains(&t));
1209 114 : if let Some(groups) = client.filter(|c| c.iter().all(named)) {
1210 86 : return Some(std::borrow::Cow::Borrowed(groups));
1211 28 : }
1212 28 : let any = [Vec::new()];
1213 : Some(std::borrow::Cow::Owned(
1214 28 : client
1215 28 : .unwrap_or(&any)
1216 28 : .iter()
1217 58 : .flat_map(|g| -> Vec<Vec<String>> {
1218 58 : if named(g) {
1219 30 : return vec![g.clone()];
1220 28 : }
1221 28 : union
1222 28 : .iter()
1223 50 : .map(|t| {
1224 50 : let mut g = g.clone();
1225 50 : g.push((*t).clone());
1226 50 : g
1227 50 : })
1228 28 : .collect()
1229 58 : })
1230 28 : .collect(),
1231 : ))
1232 124 : }
1233 :
1234 : /// The documents of a page ordered over trimmed ones ([`Filtered::trimmed`]),
1235 : /// read whole by id under the same filter, in the page's order. An Entity
1236 : /// deleted between the two reads drops out of the page.
1237 184 : async fn whole_page(
1238 184 : st: &AppState,
1239 184 : tenant: &TenantId,
1240 184 : params: &HashMap<String, String>,
1241 184 : ctx: &antares_jsonld::Context,
1242 184 : repr: &crate::repr::Repr,
1243 184 : page: Vec<Value>,
1244 184 : ) -> ApiResult<Vec<Value>> {
1245 474 : let ids: Vec<&str> = page.iter().filter_map(|d| d.get("id")?.as_str()).collect();
1246 184 : if ids.is_empty() {
1247 36 : return Ok(Vec::new());
1248 148 : }
1249 148 : let mut by_id = params.clone();
1250 148 : by_id.insert("id".to_owned(), ids.join(","));
1251 148 : let whole = filter_entities_paged(
1252 148 : st,
1253 148 : tenant,
1254 148 : &by_id,
1255 148 : ctx,
1256 148 : Vec::new(),
1257 148 : Pushdown::default(),
1258 148 : Some(repr),
1259 148 : )
1260 148 : .await?;
1261 148 : let mut found: HashMap<String, Value> = whole
1262 148 : .docs
1263 148 : .into_iter()
1264 474 : .filter_map(|d| Some((d.get("id")?.as_str()?.to_owned(), d)))
1265 148 : .collect();
1266 474 : Ok(ids.iter().filter_map(|id| found.remove(*id)).collect())
1267 184 : }
1268 :
1269 : /// Whether the store may be asked to cut the page (ORDER BY id +
1270 : /// LIMIT/OFFSET). Only when nothing outside it still narrows or reorders the
1271 : /// match set:
1272 : ///
1273 : /// * federated candidates are merged in afterwards, and 5.7.2.4 applies the
1274 : /// query, geoquery, Scope query and Attribute filters only after that
1275 : /// aggregation — so a SQL page would be cut from the wrong set;
1276 : /// * `idPattern` is not part of the store filter, so it drops rows the SQL
1277 : /// page already counted;
1278 : /// * `orderBy` orders the whole match set by the value of the ordering member
1279 : /// (4.23) before the page is cut, and that comparison order is the
1280 : /// evaluator's.
1281 : ///
1282 : /// `limit=0` — the count-only shape of 6.3.10, where `count=true` is
1283 : /// mandatory — IS pushed: the store returns no rows and counts the match set,
1284 : /// which is the same count the scan derives from a materialized one, without
1285 : /// materializing 100 million documents to throw them away.
1286 2040 : fn page_pushdown_allowed(fed_is_empty: bool, params: &HashMap<String, String>) -> bool {
1287 2040 : fed_is_empty && !params.contains_key("idPattern") && !params.contains_key("orderBy")
1288 2040 : }
1289 :
1290 : /// Entity Type Selection Language (4.17): `,`/`|` = OR, `(a;b)` = AND.
1291 : /// Table 6.4.3.2-1: `"*"` selects every Entity Type, i.e. no type predicate
1292 : /// at all. Expanding it as a term yields an IRI nothing matches, which is
1293 : /// how `type=*` silently returned an empty array.
1294 8435 : pub(crate) fn type_selection(
1295 8435 : params: &HashMap<String, String>,
1296 8435 : ctx: &antares_jsonld::Context,
1297 8435 : ) -> Option<Vec<Vec<String>>> {
1298 8435 : params.get("type").filter(|s| *s != "*").map(|s| {
1299 4633 : s.split([',', '|'])
1300 5445 : .map(|alt| {
1301 5445 : alt.trim()
1302 5445 : .trim_start_matches('(')
1303 5445 : .trim_end_matches(')')
1304 5445 : .split(';')
1305 5465 : .map(|t| ctx.expand_key(t.trim()))
1306 5445 : .collect()
1307 5445 : })
1308 4633 : .collect()
1309 4633 : })
1310 8435 : }
1311 :
1312 : /// The expanded `attrs` list of a query.
1313 3859 : fn attr_list(
1314 3859 : params: &HashMap<String, String>,
1315 3859 : ctx: &antares_jsonld::Context,
1316 3859 : ) -> Option<Vec<String>> {
1317 3859 : params
1318 3859 : .get("attrs")
1319 3859 : .map(|s| s.split(',').map(|t| ctx.expand_key(t.trim())).collect())
1320 3859 : }
1321 :
1322 : /// What a read names, for the policy gate (ADR-0020: "`Operation` is the
1323 : /// expanded request"): the types of its type selector, its `attrs`, its
1324 : /// `q` and its GeoQuery, expanded with the request @context exactly as the
1325 : /// store query uses them. Built only with an engine attached. A member
1326 : /// that does not parse is left out, and the read refuses it after the gate
1327 : /// as before.
1328 : #[derive(Default)]
1329 : pub(crate) struct ReadMembers {
1330 : pub types: Vec<String>,
1331 : pub attrs: Vec<String>,
1332 : pub q: Option<antares_ql::QNode>,
1333 : pub geo: Option<antares_ql::geo::GeoQuery>,
1334 : }
1335 :
1336 : impl ReadMembers {
1337 3424 : pub(crate) fn with_ctx(
1338 3424 : st: &AppState,
1339 3424 : params: &HashMap<String, String>,
1340 3424 : ctx: &antares_jsonld::Context,
1341 3424 : ) -> Self {
1342 3424 : if st.policy.is_none() {
1343 2584 : return Self::default();
1344 840 : }
1345 840 : let mut types: Vec<String> = Vec::new();
1346 1004 : for t in type_selection(params, ctx).into_iter().flatten().flatten() {
1347 1004 : if !types.contains(&t) {
1348 1004 : types.push(t);
1349 1004 : }
1350 : }
1351 : Self {
1352 840 : types,
1353 840 : attrs: attr_list(params, ctx).unwrap_or_default(),
1354 840 : q: params
1355 840 : .get("q")
1356 840 : .and_then(|q| antares_ql::parse_narrowed_q(q).ok()),
1357 840 : geo: antares_ql::geo::GeoQuery::from_params(params)
1358 840 : .ok()
1359 840 : .flatten(),
1360 : }
1361 3424 : }
1362 :
1363 : /// Same, for a gate taken before the request @context is resolved.
1364 3514 : pub(crate) async fn of(
1365 3514 : st: &AppState,
1366 3514 : headers: &HeaderMap,
1367 3514 : params: &HashMap<String, String>,
1368 3514 : ) -> Self {
1369 3514 : if st.policy.is_none() {
1370 2904 : return Self::default();
1371 610 : }
1372 610 : match request_context(&st.loader, headers).await {
1373 610 : Ok(ctx) => Self::with_ctx(st, params, &ctx),
1374 0 : Err(_) => Self::default(),
1375 : }
1376 3514 : }
1377 : }
1378 :
1379 : /// What the store may be asked to do beyond filtering; see
1380 : /// [`filter_entities_paged`].
1381 : #[derive(Debug, Default, Clone, Copy)]
1382 : pub struct Pushdown<'a> {
1383 : pub page: Option<(usize, usize)>,
1384 : pub proj: Option<&'a crate::repr::Repr>,
1385 : /// The expanded Attributes an `orderBy` reads: an ordered read sorts
1386 : /// the whole match set, and these alone are what the sort needs
1387 : /// ([`Filtered::trimmed`]).
1388 : pub order_keys: Option<&'a [String]>,
1389 : }
1390 :
1391 : /// The full filtering path (5.7.2). `push.page` = (offset, limit) to push into the
1392 : /// store — pass it ONLY when every filter the store cannot see is absent
1393 : /// (idPattern, federation, orderBy); the store still refuses unless its own
1394 : /// predicates compiled exactly. `push.proj` = the parsed representation,
1395 : /// offered for projection pushdown (pick/omit/attrs top-level heads) under
1396 : /// the same exactness gate. `narrow` = the caller's policy narrowing, which
1397 : /// a linked q term reads its targets through (ADR-0020).
1398 3033 : pub async fn filter_entities_paged(
1399 3033 : st: &AppState,
1400 3033 : tenant: &TenantId,
1401 3033 : params: &HashMap<String, String>,
1402 3033 : ctx: &antares_jsonld::Context,
1403 3033 : fed: Vec<(bool, Value)>,
1404 3033 : push: Pushdown<'_>,
1405 3033 : narrow: Option<&crate::repr::Repr>,
1406 3033 : ) -> ApiResult<Filtered> {
1407 : let Pushdown {
1408 3033 : page,
1409 3033 : proj,
1410 3033 : order_keys,
1411 3033 : } = push;
1412 : // a keyset cursor: the store seeks past its id, so the rows
1413 : // already start at the client's offset and every page below is counted
1414 : // from there
1415 3033 : let after = crate::paging::cursor_after(&crate::paging::view_of(st), tenant.as_str(), params)?;
1416 3033 : let page = page.map(|(offset, limit)| (if after.is_some() { 0 } else { offset }, limit));
1417 : // a pushed page over local rows cannot be merged with federated
1418 : // candidates — refuse here so no caller can create that page
1419 3033 : let page = if fed.is_empty() { page } else { None };
1420 : // and neither can a pushed projection: the store's answer is only the
1421 : // final answer when nothing downstream still needs the stripped members
1422 3033 : let proj = proj.filter(|_| proj_pushdown_allowed(fed.is_empty(), params));
1423 : // under rules the store answers the policy's own
1424 : // predicates over the stored Entity, and the loop below decides which
1425 : // rules hold and evaluates the client's predicates on the Entity as it
1426 : // is shown, so the client's page is cut in the store only where the
1427 : // store can judge the rules itself (`judged` below).
1428 3033 : let rules = narrow.and_then(|r| r.policy_types.as_deref());
1429 3033 : let client_page = page;
1430 3033 : let page = if rules.is_some() { None } else { page };
1431 3033 : let proj = if rules.is_some() { None } else { proj };
1432 3033 : let ids: Option<Vec<&str>> = params.get("id").map(|s| s.split(',').collect());
1433 3033 : if let Some(ids) = &ids {
1434 1254 : for id in ids {
1435 1254 : antares_model::EntityId::new(id)?;
1436 : }
1437 2521 : }
1438 3027 : let id_pattern = match params.get("idPattern") {
1439 130 : Some(p) => {
1440 508 : if ["**", "++", "*+", "+*"].iter().any(|q| p.contains(q)) {
1441 4 : return Err(NgsiError::BadRequestData(format!("invalid idPattern {p:?}")).into());
1442 126 : }
1443 : Some(
1444 126 : antares_ql::regex::compile(p)
1445 126 : .map_err(|_| NgsiError::BadRequestData(format!("invalid idPattern {p:?}")))?,
1446 : )
1447 : }
1448 2897 : None => None,
1449 : };
1450 3019 : let type_sel = type_selection(params, ctx);
1451 3019 : let attr_filter = attr_list(params, ctx);
1452 3019 : let q_ast = match params.get("q") {
1453 : // 4.9 expandValues: "attributes whose values should be expanded
1454 : // against the supplied @context using JSON-LD type coercion prior to
1455 : // executing the query" (EXAMPLE 12), less the Attributes jsonKeys
1456 : // declares uninterpretable as JSON-LD.
1457 : // the query may be narrowed: every caller parsed its own `q` with
1458 : // `parse_q` before (threat V23)
1459 652 : Some(q) => Some(antares_ql::eval::apply_expand_values(
1460 652 : antares_ql::parse_narrowed_q(q)?,
1461 652 : antares_ql::eval::expansion_list(
1462 652 : params.get("expandValues").map(String::as_str),
1463 652 : params.get("jsonKeys").map(String::as_str),
1464 652 : )
1465 652 : .as_deref(),
1466 652 : ctx,
1467 : )),
1468 2367 : None => None,
1469 : };
1470 3019 : let scope_q = params.get("scopeQ");
1471 : // The request's own geoquery (7.2.5 allows one) and the narrowing's
1472 : // area: every one must hold, so a caller's area
1473 : // cannot widen the policy's.
1474 3019 : let own_geo = antares_ql::geo::GeoQuery::from_params(params)?;
1475 2983 : let policy_geo = narrow
1476 2983 : .and_then(|r| r.policy_reach.as_deref())
1477 2983 : .map(|reach| reach.areas.as_slice())
1478 2983 : .unwrap_or_default();
1479 2983 : let geos: Vec<&antares_ql::geo::GeoQuery> = own_geo.iter().chain(policy_geo).collect();
1480 : // under rules the client's q, attrs and geoQ read the shown Entity, and
1481 : // only the engine's own q and area reach the store and the stored one
1482 2983 : let client_q = narrow.and_then(|r| r.policy_client_q.as_ref());
1483 2983 : let engine_q = narrow
1484 2983 : .and_then(|r| r.policy_reach.as_deref())
1485 2983 : .and_then(|reach| reach.q.as_ref());
1486 2983 : let client_and_engine_q = q_ast.clone();
1487 2983 : let (q_ast, shown_q) = if rules.is_some() {
1488 310 : (engine_q.cloned(), client_q)
1489 : } else {
1490 2673 : (q_ast, None)
1491 : };
1492 2983 : let (geos, shown_geos): (Vec<_>, Vec<_>) = if rules.is_some() {
1493 310 : (policy_geo.iter().collect(), own_geo.iter().collect())
1494 : } else {
1495 2673 : (geos, Vec::new())
1496 : };
1497 2983 : let (attr_filter, shown_attrs) = if rules.is_some() {
1498 310 : (None, attr_filter)
1499 : } else {
1500 2673 : (attr_filter, None)
1501 : };
1502 2983 : let reads_shown = shown_attrs.is_some() || !shown_geos.is_empty() || shown_q.is_some();
1503 : // Rules that select by type alone admit exactly the Entities of their
1504 : // types. When the client reads nothing on the shown Entity either, the
1505 : // store takes those types as one more selection, and pages and counts
1506 : // as at the root (a page of one read every Entity the policy
1507 : // admits).
1508 : // So too rules that narrow their types only by one set of `idPattern`s
1509 : // the store applies exactly: it takes them as a selection too.
1510 2983 : let rule_patterns = rules.and_then(|f| crate::policy::shared_id_patterns(&f.rules));
1511 2983 : let by_type = rules
1512 2983 : .filter(|f| {
1513 310 : rule_patterns.is_some()
1514 296 : || f.rules
1515 296 : .iter()
1516 296 : .all(crate::policy::Rule::admits_by_type_alone)
1517 310 : })
1518 2983 : .and_then(|f| Some((f, judged_types(type_sel.as_deref(), f)?)));
1519 : // So too when the client's `q` reads only Attributes that every rule of
1520 : // the types it selects shows: on those, the Entity as stored and as shown
1521 : // answer it alike (`count=true` with a `q`, which every paged
1522 : // façade read sends, read every Entity the policy admits and hit the
1523 : // store's TooManyResults ceiling).
1524 : // So too the client's geoquery on the default GeoProperty when every rule
1525 : // of those types shows it: the store's spatial pushdown answers it as the
1526 : // shown Entity would (a geo read under
1527 : // a view read every Entity of the type in doubling pages).
1528 2983 : let geos_alike = shown_attrs.is_none()
1529 2977 : && by_type.as_ref().is_some_and(|(f, groups)| {
1530 98 : shown_geos.iter().all(|g| {
1531 8 : let name = Some(g.geoproperty.as_str()).filter(|n| !n.is_empty());
1532 8 : g.to_sql_spec(ctx).is_some()
1533 8 : && shows_alike(
1534 8 : [name.unwrap_or("location")],
1535 8 : f,
1536 8 : groups,
1537 8 : narrow,
1538 8 : &|t: &str| ctx.expand_key(t),
1539 : )
1540 8 : })
1541 98 : });
1542 2983 : let q_alike = geos_alike
1543 94 : && by_type
1544 94 : .as_ref()
1545 94 : .zip(shown_q)
1546 94 : .is_some_and(|((f, groups), q)| {
1547 60 : reads_only_shown(q, f, groups, narrow, &|t: &str| ctx.expand_key(t))
1548 32 : });
1549 2983 : let judged = by_type
1550 2983 : .filter(|_| !reads_shown || (geos_alike && (shown_q.is_none() || q_alike)))
1551 2983 : .map(|(_, groups)| groups);
1552 2983 : let (geos, shown_geos) = if judged.is_some() && geos_alike {
1553 68 : (geos.into_iter().chain(shown_geos).collect(), Vec::new())
1554 : } else {
1555 2915 : (geos, shown_geos)
1556 : };
1557 2983 : let reads_shown = shown_attrs.is_some() || !shown_geos.is_empty() || shown_q.is_some();
1558 : // the client's own `q` (with the engine's, as the parameters carry them)
1559 : // reaches the store only where it reads the shown Entity alike
1560 2983 : let q_ast = if judged.is_some() && q_alike {
1561 6 : client_and_engine_q
1562 : } else {
1563 2977 : q_ast
1564 : };
1565 2983 : let page = if judged.is_some() { client_page } else { page };
1566 : // An ordered read sorts every match, so the store keeps only what the
1567 : // sort reads: only where nothing below reads more of the stored
1568 : // Entity, that is no rules or rules that select by type alone (a client
1569 : // `q` they leave to the store is decided there), and no federated rows.
1570 2983 : let order_keys = order_keys.filter(|_| fed.is_empty() && (rules.is_none() || judged.is_some()));
1571 : // Otherwise, without `count` the answer needs the matches up to the
1572 : // page's end and one more (whether a next page exists): the store is
1573 : // read in pages of its own, doubling, until that many held.
1574 : // `count=true` under conditional rules needs every match counted: the
1575 : // store is read in pages of its own to the end, only the client's window
1576 : // is kept, and the scan stops at MAX_FOLD_DOCS. One flat read
1577 : // refused a dataset's count once its type held MAX_UNDECIDED_ROWS.
1578 : // Still reads every Entity the policy admits; compile the
1579 : // rules' OR and the shown-set predicates to SQL if that shows up
1580 2983 : let counting = client_page.filter(|_| {
1581 2063 : rules.is_some()
1582 302 : && judged.is_none()
1583 236 : && params.get("count").map(String::as_str) == Some("true")
1584 2063 : });
1585 : // So too a geoquery or a scopeQ the store narrows but cannot decide: it
1586 : // pages the candidates (`query_candidates`) and the loop below decides
1587 : // them, where one flat read took up to MAX_UNDECIDED_ROWS for a page of
1588 : // 20 (~1 000 rows, 0.6-1.1 s a statement).
1589 2983 : let undecidable = !geos.is_empty() || scope_q.is_some();
1590 2983 : let enough = match client_page {
1591 310 : Some((offset, limit))
1592 2063 : if ((rules.is_some() && judged.is_none()) || undecidable)
1593 442 : && params.get("count").map(String::as_str) != Some("true") =>
1594 : {
1595 310 : offset.checked_add(limit).and_then(|n| n.checked_add(1))
1596 : }
1597 2673 : _ => None,
1598 : };
1599 :
1600 : // Hand the store what it can filter on. A backend that can push
1601 : // the predicate down (postgres/timescale) returns fewer rows — and says
1602 : // via `decided` whether it applied EVERY present predicate exactly, which
1603 : // is what licenses pagination/projection pushdown and lets the loop below
1604 : // skip re-deciding. A backend that cannot (memory/file) returns the
1605 : // snapshot and the loop stays the arbiter.
1606 2983 : let expand = |t: &str| ctx.expand_key(t);
1607 2983 : let geo_specs: Vec<_> = geos.iter().filter_map(|g| g.to_sql_spec(ctx)).collect();
1608 : // A geo query whose spec declined to compile (non-default geoproperty) is
1609 : // INVISIBLE to the store — the store would truthfully claim `decided`
1610 : // about what it saw, projection would strip the very member the evaluator
1611 : // still needs, and a pushed page would page over the wrong set. Forfeit
1612 : // every pushdown up front and mask `decided` after. The specs that did
1613 : // compile still narrow the rows.
1614 2983 : let geo_uncompiled = geo_specs.len() < geos.len();
1615 2983 : let page = if geo_uncompiled { None } else { page };
1616 2983 : let enough = enough.filter(|_| !geo_uncompiled);
1617 2983 : let counting = counting.filter(|_| !geo_uncompiled);
1618 2983 : let proj = proj.filter(|_| !geo_uncompiled);
1619 2983 : let order_keys = order_keys.filter(|_| !geo_uncompiled);
1620 : // pick (or attrs) heads to keep / whole-attr omit heads to drop; core
1621 : // members are never SQL-dropped (only `://` IRIs qualify) — repr::apply
1622 : // stays the decider for those.
1623 2983 : let keep_attrs: Option<Vec<String>> = proj
1624 2983 : .and_then(|r| {
1625 1106 : r.pick
1626 1106 : .as_ref()
1627 1106 : .map(|nodes| nodes.iter().map(|n| n.iri.clone()).collect())
1628 1106 : .or_else(|| r.attrs.clone())
1629 1106 : })
1630 2983 : .or_else(|| order_keys.map(<[String]>::to_vec));
1631 2983 : let drop_attrs: Option<Vec<String>> = proj
1632 2983 : .and_then(|r| {
1633 1106 : r.omit.as_ref().map(|nodes| {
1634 16 : nodes
1635 16 : .iter()
1636 20 : .filter(|n| n.children.is_none() && n.iri.contains("://"))
1637 16 : .map(|n| n.iri.clone())
1638 16 : .collect::<Vec<_>>()
1639 16 : })
1640 1106 : })
1641 2983 : .filter(|v| !v.is_empty());
1642 : // 5.7.2.4 split entities (p.202): the filters (q, geoquery, Scope query,
1643 : // Attributes) apply only AFTER remote parts and local information have
1644 : // been aggregated — so with federated candidates present the store must
1645 : // not drop (or pre-project) the LOCAL half of a split entity. The
1646 : // post-merge loop below applies them instead (`decided` is already
1647 : // false whenever `fed` is non-empty).
1648 2983 : let split_agg = crate::federation::split_entities(params) && !fed.is_empty();
1649 : // the id list is client-sized (a POST query body carries an array with no
1650 : // count of its own) and the candidate set is store-sized, so the two are
1651 : // never multiplied together
1652 2983 : let id_set: Option<std::collections::HashSet<&str>> =
1653 2983 : ids.as_ref().map(|v| v.iter().copied().collect());
1654 2983 : let mut out = Vec::new();
1655 : let (mut paged, mut total): (bool, Option<usize>);
1656 : let mut trimmed: bool;
1657 2983 : let mut rows = 0usize;
1658 2983 : let mut fed_once = Some(fed);
1659 : // the store page under `enough` or `counting`: (offset, limit), the limit
1660 : // doubling up to COUNT_CHUNK. Under `enough` only the client's window is
1661 : // kept, as under `counting`: holding every match up to the page's end made
1662 : // a deep page of a large dataset hold the whole dataset, read in one
1663 : // statement, and four of them took the broker past its limit.
1664 : const COUNT_CHUNK: usize = 2048;
1665 2983 : let window = client_page.filter(|_| enough.is_some());
1666 2983 : let mut read = (
1667 : 0usize,
1668 2983 : enough
1669 2983 : .map(|n| n.min(COUNT_CHUNK))
1670 2983 : .unwrap_or(if counting.is_some() { 1024 } else { 0 }),
1671 : );
1672 2983 : let mut matched = 0usize;
1673 : // the walk's keyset: each store page starts after the last id
1674 : // the one before read, where a growing OFFSET re-read every row it
1675 : // skipped (80 pages over 164 000 Trees)
1676 2983 : let mut walk_after: Option<String> = None;
1677 : // Rules the loop below decides row by row, each narrowing by one
1678 : // `idPattern`, hand the store the id prefixes those patterns force: a
1679 : // dataset view of 70 KPIs walked the Tenant's 128 678 and timed out at
1680 : // the gateway, even for `limit=1`. The rules still decide.
1681 : // Rules the store decides by their patterns keep the ranges too: the
1682 : // regex filters only the rows the prefixes' index range found.
1683 2983 : let rule_prefixes =
1684 2983 : match rules.filter(|_| (judged.is_none() || rule_patterns.is_some()) && ids.is_none()) {
1685 256 : Some(f) => crate::policy::rules_id_prefixes(st, tenant, &f.rules).await?,
1686 2727 : None => None,
1687 : };
1688 : 'scan: loop {
1689 3011 : let f = antares_store::filter::EntityFilter {
1690 3011 : ids: ids.as_deref(),
1691 : // 5.2.33: id takes precedence over idPattern, so the literal
1692 : // narrows only when no id selector was given
1693 3011 : id_literal: if ids.is_none() {
1694 2505 : params
1695 2505 : .get("idPattern")
1696 2505 : .and_then(|p| antares_store::filter::id_pattern_literal(p))
1697 : } else {
1698 506 : None
1699 : },
1700 3011 : id_prefixes: rule_prefixes.as_deref(),
1701 3011 : id_patterns: rule_patterns,
1702 3011 : types: judged.as_deref().or(type_sel.as_deref()),
1703 3011 : attrs: if split_agg {
1704 6 : None
1705 : } else {
1706 3005 : attr_filter.as_deref()
1707 : },
1708 3011 : q: if split_agg { None } else { q_ast.as_ref() },
1709 3011 : scope_q: if split_agg {
1710 6 : None
1711 : } else {
1712 3005 : scope_q.map(String::as_str)
1713 : },
1714 3011 : geo: if split_agg { &[] } else { &geo_specs },
1715 3011 : expand: &expand,
1716 3011 : ctx: Some(ctx),
1717 3011 : page: match enough.or(counting.map(|_| 0)) {
1718 372 : Some(_) => Some(store_page(read)),
1719 2639 : None => page.map(|(offset, limit)| antares_store::filter::Page {
1720 1715 : offset: offset as i64,
1721 1715 : limit: limit as i64,
1722 1715 : count: params.get("count").map(String::as_str) == Some("true"),
1723 1715 : }),
1724 : },
1725 3011 : keep_attrs: keep_attrs.as_deref(),
1726 3011 : drop_attrs: drop_attrs.as_deref(),
1727 3011 : after_id: walk_after.as_deref().or(after.as_deref()),
1728 : };
1729 : // a walk in pages decides each row itself, so the store may page
1730 : // what it cannot decide
1731 3011 : let outcome = match enough.or(counting.map(|_| 0)) {
1732 372 : Some(_) => st.store.query_candidates(tenant, &f).await?,
1733 2639 : None => st.store.query_entities(tenant, &f).await?,
1734 : };
1735 3011 : let fed = fed_once.take().unwrap_or_default();
1736 3011 : let store_paged = outcome.paged;
1737 3011 : let decided = outcome.decided && fed.is_empty() && !geo_uncompiled;
1738 3011 : trimmed = order_keys.is_some() && decided;
1739 3011 : paged = store_paged && fed.is_empty() && enough.is_none();
1740 3011 : total = outcome.total.map(|t| t as usize);
1741 3011 : let got = outcome.rows.len();
1742 3011 : rows = rows.saturating_add(got);
1743 3011 : walk_after = outcome
1744 3011 : .rows
1745 3011 : .last()
1746 3011 : .and_then(|d| d.get("id")?.as_str())
1747 3011 : .map(str::to_owned);
1748 3011 : let all = crate::federation::merge_candidates(outcome.rows, fed);
1749 32476 : for doc in all {
1750 32476 : let id = doc.get("id").and_then(Value::as_str).unwrap_or("");
1751 32476 : if let Some(ids) = &id_set {
1752 1164 : if !decided && !ids.contains(id) {
1753 0 : continue;
1754 1164 : }
1755 31312 : }
1756 : // 5.2.33: "id takes precedence over idPattern" — the pattern only
1757 : // filters when no id selector was given.
1758 32476 : if ids.is_none() {
1759 31312 : if let Some(re) = &id_pattern {
1760 : // idPattern is invisible to the store — applied even when
1761 : // decided
1762 2628 : if !re.is_match(id) {
1763 1214 : continue;
1764 1414 : }
1765 28684 : }
1766 1164 : }
1767 31262 : if !decided {
1768 6872 : if let Some(sel) = &type_sel {
1769 6872 : let etypes: Vec<&str> = doc
1770 6872 : .get("type")
1771 6872 : .and_then(Value::as_array)
1772 6872 : .map(|a| a.iter().filter_map(Value::as_str).collect())
1773 6872 : .unwrap_or_default();
1774 6872 : let matched = sel
1775 6872 : .iter()
1776 6878 : .any(|and_group| and_group.iter().all(|w| etypes.contains(&w.as_str())));
1777 6872 : if !matched {
1778 0 : continue;
1779 6872 : }
1780 0 : }
1781 6872 : if let Some(attrs) = &attr_filter {
1782 16 : if !attrs.iter().any(|a| doc.get(a).is_some()) {
1783 0 : continue;
1784 16 : }
1785 6856 : }
1786 6872 : if let Some(ast) = &q_ast {
1787 : // 4.9 linked-entity subqueries (attr{path}) resolve through
1788 : // the local store, same tenant, under the caller's narrowing
1789 138 : let narrow = narrow.map(|r| (r, ctx));
1790 174 : if !crate::notify::linked_eval(st, tenant, narrow, |l| {
1791 174 : eval_q(ast, &doc, ctx, l)
1792 174 : })
1793 138 : .await
1794 : {
1795 68 : continue;
1796 70 : }
1797 6734 : }
1798 6804 : if let Some(sq) = scope_q {
1799 6340 : if !crate::scope_matches(sq, &doc) {
1800 2842 : continue;
1801 3498 : }
1802 464 : }
1803 3962 : if !geos.iter().all(|g| g.matches(&doc, ctx)) {
1804 164 : continue;
1805 3798 : }
1806 24390 : }
1807 28188 : if let (Some(rules), Some(r)) = (rules, narrow) {
1808 1676 : if !rules.admits(&doc, crate::repr::policy_context()) {
1809 440 : continue;
1810 1236 : }
1811 : // the Entity as shown is built only for a client predicate
1812 : // that reads it
1813 1236 : if reads_shown {
1814 176 : let view = crate::repr::policy_view(&doc, r);
1815 176 : if let Some(attrs) = &shown_attrs {
1816 12 : if !attrs.iter().any(|a| view.get(a).is_some()) {
1817 6 : continue;
1818 6 : }
1819 164 : }
1820 170 : if !shown_geos.iter().all(|g| g.matches(&view, ctx)) {
1821 10 : continue;
1822 160 : }
1823 : // a `q` the store took because it reads the shown Entity
1824 : // alike is decided there
1825 160 : if let Some(ast) = shown_q.filter(|_| !(q_alike && decided)) {
1826 148 : let narrow = narrow.map(|r| (r, ctx));
1827 148 : if !crate::notify::linked_eval(st, tenant, narrow, |l| {
1828 148 : eval_q(ast, &view, ctx, l)
1829 148 : })
1830 148 : .await
1831 : {
1832 64 : continue;
1833 84 : }
1834 12 : }
1835 1060 : }
1836 26512 : }
1837 27668 : if let Some((offset, limit)) = counting {
1838 : // ORDER BY id is the store's order, so the k-th match is the
1839 : // k-th Entity of the client's page
1840 88 : if matched.checked_sub(offset).is_some_and(|k| k < limit) {
1841 28 : out.push(doc);
1842 60 : }
1843 88 : matched = matched.saturating_add(1);
1844 88 : continue;
1845 27580 : }
1846 27580 : if let Some((offset, limit)) = window {
1847 : // the window and one match past it, which says a next page exists
1848 3880 : if matched >= offset {
1849 3840 : out.push(doc);
1850 3840 : }
1851 3880 : matched = matched.saturating_add(1);
1852 3880 : if out.len() > limit {
1853 14 : break 'scan;
1854 3866 : }
1855 3866 : continue;
1856 23700 : }
1857 23700 : out.push(doc);
1858 : }
1859 : // a store that could not page answered with every row it holds
1860 2997 : if (enough.is_none() && counting.is_none()) || !store_paged || got < read.1 {
1861 2967 : break;
1862 30 : }
1863 : // a walk over candidates is bounded as the flat read it replaces was,
1864 : // a window that never fills (rules admitting nothing) too
1865 30 : if rows >= *crate::bounds::MAX_FOLD_DOCS {
1866 2 : return Err(NgsiError::TooManyResults(format!(
1867 2 : "this query read more than {} entities; narrow it",
1868 2 : *crate::bounds::MAX_FOLD_DOCS
1869 2 : ))
1870 2 : .into());
1871 28 : }
1872 28 : read = (0, read.1.saturating_mul(2).min(COUNT_CHUNK));
1873 : }
1874 2981 : if let Some((offset, limit)) = window {
1875 : // the smallest total consistent with "a next page exists"
1876 308 : let more = out.len() > limit;
1877 308 : out.truncate(limit);
1878 308 : return Ok(Filtered {
1879 308 : total: Some(
1880 308 : offset
1881 308 : .saturating_add(out.len())
1882 308 : .saturating_add(usize::from(more)),
1883 308 : ),
1884 308 : rows: out.len(),
1885 308 : docs: out,
1886 308 : paged: true,
1887 308 : trimmed,
1888 308 : });
1889 2673 : }
1890 2673 : if counting.is_some() {
1891 : // the store paged to the end, or answered every row it holds at once
1892 34 : return Ok(Filtered {
1893 34 : rows: out.len(),
1894 34 : docs: out,
1895 34 : paged: true,
1896 34 : total: Some(matched),
1897 34 : trimmed,
1898 34 : });
1899 2639 : }
1900 2639 : Ok(Filtered {
1901 2639 : docs: out,
1902 2639 : paged,
1903 2639 : total,
1904 2639 : rows,
1905 2639 : trimmed,
1906 2639 : })
1907 3033 : }
1908 :
1909 : // ---------- DELETE /entities/{id} (5.6.6) ----------
1910 :
1911 3106 : pub async fn delete_entity(
1912 3106 : State(st): State<AppState>,
1913 3106 : Path(id): Path<String>,
1914 3106 : CleanParams(params): CleanParams,
1915 3106 : headers: HeaderMap,
1916 3106 : ) -> Response {
1917 3106 : let go = async {
1918 3106 : let tenant = tenant_from(&headers)?;
1919 3106 : antares_model::EntityId::new(&id)?;
1920 3102 : check_params(¶ms, &["local", "type"])?;
1921 : // 5.5.7: `type` is a term, so the request's own @context expands it.
1922 : // Under the core context the same word names a different type than
1923 : // the client meant, and the delete then removes an Entity the
1924 : // client's selector excluded.
1925 3102 : let ctx = request_context(&st.loader, &headers).await?;
1926 3102 : let pre = crate::policy::pre_image(&st, &tenant, &id, None).await?;
1927 3102 : let granted = gate!(
1928 : st, &tenant, &headers, "5.6.6",
1929 : ids: &[&id],
1930 : types: &pre.types,
1931 : stored: pre.stored.as_ref(),
1932 : )
1933 3102 : .await?;
1934 3092 : let cond = crate::policy::Condition::of(&granted);
1935 : // 4.17/5.6.6.4: the type selector gates the target — a registration
1936 : // for a different type must not receive the forwarded delete.
1937 : // every type the 4.17 selection names: the forward carries the
1938 : // selector and the Context Source judges it
1939 3092 : let spec = crate::registry::CsrSpec {
1940 3092 : ids: Some(vec![id.clone()]),
1941 3092 : types: type_selection(¶ms, &ctx).map(|sel| sel.into_iter().flatten().collect()),
1942 3092 : ..Default::default()
1943 : };
1944 3092 : let regs = match crate::federation::write_plan(&st, &tenant, &spec, &ctx, ¶ms, &headers)
1945 3092 : .await?
1946 : {
1947 20 : crate::federation::WritePlan::Answered(r) => return Ok(*r),
1948 3072 : crate::federation::WritePlan::Forward(regs) => regs,
1949 : };
1950 3072 : crate::policy::Condition::unforwarded(cond.as_ref(), !regs.is_empty())?;
1951 3072 : let keep = |d: &Value| {
1952 : // An Entity the engine's condition does not select is
1953 : // not found
1954 2844 : if cond.as_ref().is_some_and(|c| !c.holds(d)) {
1955 4 : return false;
1956 2840 : }
1957 2840 : crate::negotiate::matches_type_param(d, ¶ms, &ctx)
1958 2844 : };
1959 3072 : if !regs.is_empty() {
1960 126 : let proxy_match = regs.iter().any(|r| r.is_proxy());
1961 126 : let mut parts = Vec::new();
1962 126 : if st.store.delete_entity_if(&tenant, &id, &keep).await? {
1963 66 : mirror_delete_entity(&st, &tenant, &id).await;
1964 66 : parts.push(crate::federation::Part::local(204, "deleted locally"));
1965 60 : } else if !proxy_match {
1966 32 : // nothing local to delete, and no proxy that owns it: the
1967 32 : // local half of this operation is the 404
1968 32 : parts.push(crate::federation::Part::local(
1969 32 : 404,
1970 32 : format!("entity {id} not found locally"),
1971 32 : ));
1972 34 : }
1973 126 : let ctx_url = crate::federation::ctx_link_url(&headers, &ctx.source);
1974 126 : let seg = crate::federation::path_segment(&id);
1975 126 : let fwd_q = type_selector_query(¶ms);
1976 126 : for reg in ®s {
1977 : // 5.6.6.4: proxy modes not supporting Delete Entity are an
1978 : // error of type Conflict; inclusive ones are not forwarded.
1979 126 : if !reg.supports("deleteEntity") {
1980 12 : if reg.is_proxy() {
1981 4 : parts.push(crate::federation::conflict_part("deleteEntity", reg));
1982 8 : }
1983 12 : continue;
1984 114 : }
1985 114 : parts.push(
1986 114 : crate::federation::forward_part(
1987 114 : &st,
1988 114 : reqwest::Method::DELETE,
1989 114 : format!("{}/ngsi-ld/v1/entities/{seg}", reg.endpoint),
1990 114 : &fwd_q,
1991 114 : &headers,
1992 114 : &tenant,
1993 114 : reg,
1994 114 : &ctx_url,
1995 114 : None,
1996 114 : )
1997 114 : .await,
1998 : );
1999 : }
2000 126 : return Ok(crate::federation::combine(
2001 126 : st.profile,
2002 126 : Some(&id),
2003 126 : parts,
2004 126 : no_content(&tenant),
2005 126 : &tenant,
2006 126 : ));
2007 2946 : }
2008 2946 : if st.store.delete_entity_if(&tenant, &id, &keep).await? {
2009 2762 : mirror_delete_entity(&st, &tenant, &id).await;
2010 2762 : Ok::<_, ApiError>(no_content(&tenant))
2011 : } else {
2012 184 : Err(NgsiError::ResourceNotFound(format!("entity {id} not found")).into())
2013 : }
2014 3106 : };
2015 3106 : go.await.unwrap_or_else(|e| e.into_response())
2016 3106 : }
2017 :
2018 : // ---------- DELETE /entities/ — Purge (5.6.21) ----------
2019 :
2020 : /// 5.6.21.3 Input data of a Purge: the Entity type selector, the identifier
2021 : /// list and id pattern, the restrictive and exclusionary Attribute-name
2022 : /// lists, the NGSI-LD Query, the GeoQuery, the Scope query and the context
2023 : /// source filter. 5.6.21.4 forwards "matching input data ... to the
2024 : /// Registration endpoint", so every one of them travels: a forward that
2025 : /// carries fewer restrictions than the client issued makes the peer execute
2026 : /// a strictly wider purge than the one requested. `local` is absent by
2027 : /// design — it selects local scope, which is what stops the forward
2028 : /// happening at all (5.5.13).
2029 : const PURGE_FORWARD_PARAMS: &[&str] = &[
2030 : "type",
2031 : "id",
2032 : "idPattern",
2033 : "attrs",
2034 : "q",
2035 : "georel",
2036 : "geometry",
2037 : "coordinates",
2038 : "geoproperty",
2039 : "scopeQ",
2040 : "csf",
2041 : "keep",
2042 : "drop",
2043 : ];
2044 :
2045 30 : fn forwarded_purge_query(params: &HashMap<String, String>) -> Vec<(String, String)> {
2046 30 : PURGE_FORWARD_PARAMS
2047 30 : .iter()
2048 390 : .filter_map(|k| params.get(*k).map(|v| ((*k).to_owned(), v.clone())))
2049 30 : .collect()
2050 30 : }
2051 :
2052 : /// How many matched Entities one round of a Purge fetches and applies. The
2053 : /// match set of 5.6.21.4 is deliberately unbounded — `DELETE /entities?type=T`
2054 : /// matches every Entity of that type — so it is walked page by page and
2055 : /// applied in batches rather than materialized whole.
2056 : const PURGE_CHUNK: usize = 500;
2057 :
2058 : /// Where the next chunk of a walked match set starts, or None when the set is
2059 : /// exhausted. `rows` = documents the store returned for this chunk (not the
2060 : /// subset that survived the evaluator: idPattern is applied after the store,
2061 : /// so a full chunk can arrive narrowed or even empty and the walk must still
2062 : /// go on — 5.6.21.4 deletes ALL matched Entities, not the first page of
2063 : /// them). `left_the_set` = how many of those rows no longer match the query
2064 : /// afterwards: a Purge deletes or prunes them, a read leaves every row where
2065 : /// it was. The rows still in the set have to be stepped over, or the walk
2066 : /// re-reads them forever.
2067 : ///
2068 : /// Termination: every round either removes rows from the match set or
2069 : /// advances the offset by a full chunk, and both are bounded by the number of
2070 : /// stored Entities.
2071 539 : pub(crate) fn next_scan_offset(
2072 539 : offset: usize,
2073 539 : rows: usize,
2074 539 : left_the_set: usize,
2075 539 : paged: bool,
2076 539 : chunk: usize,
2077 539 : ) -> Option<usize> {
2078 : // an unpaged answer IS the whole match set, which this round just applied
2079 539 : if !paged || rows < chunk {
2080 475 : return None;
2081 64 : }
2082 64 : Some(offset.saturating_add(rows.saturating_sub(left_the_set)))
2083 539 : }
2084 :
2085 : /// 5.6.21.4 "And thereafter": with no Attribute-name list, delete every
2086 : /// matched Entity found locally; with a restrictive list, delete those
2087 : /// Attributes from them; with an exclusionary list, delete all but those.
2088 160 : async fn purge_locally(
2089 160 : st: &AppState,
2090 160 : tenant: &TenantId,
2091 160 : params: &HashMap<String, String>,
2092 160 : ctx: &antares_jsonld::Context,
2093 160 : keep: &Option<Vec<String>>,
2094 160 : drop: &Option<Vec<String>>,
2095 160 : ) -> ApiResult<()> {
2096 160 : let prune = keep.is_some() || drop.is_some();
2097 160 : let mut offset = 0usize;
2098 : loop {
2099 192 : let batch = filter_entities_paged(
2100 192 : st,
2101 192 : tenant,
2102 192 : params,
2103 192 : ctx,
2104 192 : Vec::new(),
2105 192 : Pushdown {
2106 192 : page: Some((offset, PURGE_CHUNK)),
2107 192 : proj: None,
2108 192 : order_keys: None,
2109 192 : },
2110 192 : // 5.6.21.4 refuses linked q terms before a purge runs
2111 192 : None,
2112 192 : )
2113 192 : .await?;
2114 192 : let rows = batch.rows;
2115 192 : let ids: Vec<String> = batch
2116 192 : .docs
2117 192 : .iter()
2118 16146 : .filter_map(|d| d["id"].as_str().map(str::to_owned))
2119 192 : .collect();
2120 192 : let left_the_set = if ids.is_empty() {
2121 100 : 0
2122 92 : } else if prune {
2123 28 : let mut changed = 0usize;
2124 28 : st.store
2125 8036 : .batch_mutate(tenant, &ids, |_, doc| {
2126 8036 : let target = antares_store::stored_object(doc)?;
2127 8036 : let attrs: Vec<String> =
2128 28176 : target.keys().filter(|k| !is_meta(k)).cloned().collect();
2129 8036 : let before = target.len();
2130 12088 : for a in attrs {
2131 12088 : let purge = match (keep, drop) {
2132 12072 : (Some(keep), _) => !keep.contains(&a),
2133 16 : (_, Some(drop)) => drop.contains(&a),
2134 0 : _ => true,
2135 : };
2136 12088 : if purge {
2137 4044 : target.remove(&a);
2138 8044 : }
2139 : }
2140 8036 : if target.len() != before {
2141 4036 : changed = changed.saturating_add(1);
2142 4036 : }
2143 8036 : Ok::<(), NgsiError>(())
2144 8036 : })
2145 28 : .await?;
2146 : // A prune keeps the Entity, but it may have removed the very
2147 : // Attribute the query matched on (`attrs=speed&drop=speed`), which
2148 : // takes it out of the match set and shifts the rest down. Whether
2149 : // it did is only observable by reading the window again, so a round
2150 : // that changed something re-reads it; the re-read finds those
2151 : // Entities unchanged and steps over them.
2152 28 : if changed > 0 {
2153 20 : rows
2154 : } else {
2155 8 : 0
2156 : }
2157 : } else {
2158 64 : let mut gone = 0usize;
2159 8110 : for (id, deleted) in ids.iter().zip(st.store.batch_delete(tenant, &ids).await?) {
2160 8110 : if deleted {
2161 8110 : gone = gone.saturating_add(1);
2162 8110 : mirror_delete_entity(st, tenant, id).await;
2163 0 : }
2164 : }
2165 64 : gone
2166 : };
2167 192 : match next_scan_offset(offset, rows, left_the_set, batch.paged, PURGE_CHUNK) {
2168 32 : Some(next) => offset = next,
2169 160 : None => return Ok(()),
2170 : }
2171 : }
2172 160 : }
2173 :
2174 : /// 5.6.21 Purge Entities: delete (or keep=/drop=-prune) all entities matched
2175 : /// by the query; output data is none — 204 (5.6.21.5). Too-wide queries,
2176 : /// Linked Entity paths and invalid id/q/geo/csf are BadRequestData
2177 : /// (5.6.21.4); matched registrations forward only with purgeEntity support.
2178 208 : pub async fn purge_entities(
2179 208 : State(st): State<AppState>,
2180 208 : CleanParams(params): CleanParams,
2181 208 : headers: HeaderMap,
2182 208 : ) -> Response {
2183 208 : match purge_inner(&st, ¶ms, &headers).await {
2184 140 : Ok(r) => r,
2185 68 : Err(e) => e.into_response(),
2186 : }
2187 208 : }
2188 :
2189 230 : async fn purge_inner(
2190 230 : st: &AppState,
2191 230 : params: &HashMap<String, String>,
2192 230 : headers: &HeaderMap,
2193 230 : ) -> ApiResult<Response> {
2194 230 : let tenant = tenant_from(headers)?;
2195 230 : check_params(
2196 230 : params,
2197 230 : &[
2198 230 : "id",
2199 230 : "idPattern",
2200 230 : "type",
2201 230 : "attrs",
2202 230 : "q",
2203 230 : "georel",
2204 230 : "geometry",
2205 230 : "coordinates",
2206 230 : "geoproperty",
2207 230 : "scopeQ",
2208 230 : "csf",
2209 230 : "keep",
2210 230 : "drop",
2211 230 : "local",
2212 230 : ],
2213 4 : )?;
2214 226 : let ctx = request_context(&st.loader, headers).await?;
2215 : // The engine sees the whole request; 5.6.21 stays one
2216 : // tenant-wide grant (ADR-0020), so nothing narrows the purge itself.
2217 226 : let named = ReadMembers::with_ctx(st, params, &ctx);
2218 226 : let ids: Vec<&str> = params
2219 226 : .get("id")
2220 226 : .map(|s| s.split(',').map(str::trim).collect())
2221 226 : .unwrap_or_default();
2222 226 : gate!(
2223 : st, &tenant, headers, "5.6.21",
2224 : ids: &ids,
2225 : q: named.q.as_ref(),
2226 : scope_q: params.get("scopeQ").map(String::as_str),
2227 : types: &named.types,
2228 : attrs: &named.attrs,
2229 : geo: named.geo.as_ref(),
2230 : )
2231 226 : .await?;
2232 : // 5.6.21.4: exactly five qualifying conditions —
2233 : // a) selector of Entity Types
2234 : // b) list of Attribute names, including at least one non-system Attribute
2235 : // c) NGSI-LD Query, including at least one non-system Attribute
2236 : // d) NGSI-LD GeoQuery
2237 : // e) local scope (5.5.13)
2238 : // "If none of the above is provided, then an error of type BadRequestData
2239 : // shall be raised (too wide query)."
2240 : //
2241 : // id/idPattern are legal input data (5.6.21.3) and DO filter, but they are
2242 : // never sufficient on their own: "it is not possible to purge a set of
2243 : // entities by only specifying desired Entity identifiers". Listing them
2244 : // here is how `DELETE /entities?idPattern=.*` became a tenant wipe.
2245 224 : let attrs_qualify = params.get("attrs").is_some_and(|a| {
2246 22 : a.split(',')
2247 22 : .any(|n| antares_ql::is_non_system_attr(n.trim()))
2248 22 : });
2249 224 : let q_ast = params.get("q").map(|q| parse_q(q)).transpose()?;
2250 224 : let q_qualifies = q_ast.as_ref().is_some_and(|ast| {
2251 26 : ast.attribute_paths()
2252 26 : .iter()
2253 26 : .any(|h| antares_ql::is_non_system_attr(h))
2254 26 : });
2255 : // 5.6.21.4: Linked Entity retrieval in the projection attributes, or
2256 : // Linked Entity attributes in the filter conditions → BadRequestData.
2257 224 : if q_ast
2258 224 : .as_ref()
2259 224 : .is_some_and(antares_ql::QNode::has_linked_paths)
2260 : {
2261 8 : return Err(NgsiError::BadRequestData(
2262 8 : "purge q must not reference Linked Entity attributes (5.6.21.4)".into(),
2263 8 : )
2264 8 : .into());
2265 216 : }
2266 216 : if params.get("attrs").is_some_and(|a| a.contains('{')) {
2267 8 : return Err(NgsiError::BadRequestData(
2268 8 : "purge attrs must not use Linked Entity retrieval (5.6.21.4)".into(),
2269 8 : )
2270 8 : .into());
2271 208 : }
2272 : // Table 6.4.3.2-1, as on the entity query (5.7.2.4)
2273 208 : star_type_local(params)?;
2274 : // 5.6.21.4: a syntactically invalid context source filter is
2275 : // BadRequestData.
2276 206 : if let Some(csf) = params.get("csf") {
2277 8 : parse_q(csf)?;
2278 198 : }
2279 202 : let has_filter = params.contains_key("type")
2280 96 : || attrs_qualify
2281 92 : || q_qualifies
2282 84 : || params.contains_key("georel")
2283 84 : || params.get("local").map(String::as_str) == Some("true");
2284 202 : if !has_filter {
2285 36 : return Err(NgsiError::BadRequestData(
2286 36 : "purge needs at least one of: type, attrs or q naming a non-system \
2287 36 : Attribute, georel, or local=true (5.6.21.4 — too wide query)"
2288 36 : .into(),
2289 36 : )
2290 36 : .into());
2291 166 : }
2292 166 : if params.contains_key("keep") && params.contains_key("drop") {
2293 4 : return Err(NgsiError::BadRequestData(
2294 4 : "keep and drop are mutually exclusive (5.6.21)".into(),
2295 4 : )
2296 4 : .into());
2297 162 : }
2298 162 : let keep: Option<Vec<String>> = params
2299 162 : .get("keep")
2300 162 : .map(|s| s.split(',').map(|t| ctx.expand_key(t.trim())).collect());
2301 162 : let drop: Option<Vec<String>> = params
2302 162 : .get("drop")
2303 162 : .map(|s| s.split(',').map(|t| ctx.expand_key(t.trim())).collect());
2304 : // distributed purge (5.6.21 / 6.4.3.3). Matching and the 6.3.17/6.3.18
2305 : // loop check come first: 508 Loop Detected is an error status, so the
2306 : // request it answers must not have deleted a page of Entities already.
2307 162 : let spec = crate::registry::CsrSpec {
2308 : // every type the 4.17 selection names, as for 5.6.6
2309 162 : types: type_selection(params, &ctx).map(|sel| sel.into_iter().flatten().collect()),
2310 162 : ids: params
2311 162 : .get("id")
2312 162 : .map(|s| s.split(',').map(str::to_owned).collect()),
2313 : // 5.12: the purge's idPattern is part of the Entity specification too
2314 162 : id_pattern: params.get("idPattern").cloned(),
2315 162 : csf: params.get("csf").and_then(|c| antares_ql::parse_q(c).ok()),
2316 162 : ..Default::default()
2317 : };
2318 160 : let regs =
2319 162 : match crate::federation::write_plan(st, &tenant, &spec, &ctx, params, headers).await? {
2320 2 : crate::federation::WritePlan::Answered(r) => return Ok(*r),
2321 160 : crate::federation::WritePlan::Forward(regs) => regs,
2322 : };
2323 160 : purge_locally(st, &tenant, params, &ctx, &keep, &drop).await?;
2324 160 : if !regs.is_empty() {
2325 22 : let mut parts = vec![crate::federation::Part::local(204, "purged locally")];
2326 22 : let ctx_url = crate::federation::ctx_link_url(headers, &ctx.source);
2327 22 : let query = forwarded_purge_query(params);
2328 26 : for reg in ®s {
2329 : // 5.6.21.4: matching input data is forwarded only when the
2330 : // registration supports Purge Entity; an unsupported matched
2331 : // registration — any mode — is an error of type Conflict
2332 : // (partial success when other parts succeeded).
2333 26 : if !reg.supports("purgeEntity") {
2334 2 : parts.push(crate::federation::conflict_part("purgeEntity", reg));
2335 2 : continue;
2336 24 : }
2337 24 : parts.push(
2338 24 : crate::federation::forward_part(
2339 24 : st,
2340 24 : reqwest::Method::DELETE,
2341 24 : format!("{}/ngsi-ld/v1/entities", reg.endpoint),
2342 24 : &query,
2343 24 : headers,
2344 24 : &tenant,
2345 24 : reg,
2346 24 : &ctx_url,
2347 24 : None,
2348 24 : )
2349 24 : .await,
2350 : );
2351 : }
2352 22 : return Ok(crate::federation::combine(
2353 22 : st.profile,
2354 22 : None,
2355 22 : parts,
2356 22 : no_content(&tenant),
2357 22 : &tenant,
2358 22 : ));
2359 138 : }
2360 138 : Ok(no_content(&tenant))
2361 230 : }
2362 :
2363 : // ---------- PATCH /entities/{id} — Merge (5.6.17 / 5.5.12) ----------
2364 :
2365 248 : pub async fn merge_entity(
2366 248 : State(st): State<AppState>,
2367 248 : Path(id): Path<String>,
2368 248 : CleanParams(params): CleanParams,
2369 248 : headers: HeaderMap,
2370 248 : body: Bytes,
2371 248 : ) -> Response {
2372 248 : match merge_entity_inner(&st, &id, ¶ms, &headers, &body).await {
2373 202 : Ok(r) => r,
2374 46 : Err(e) => e.into_response(),
2375 : }
2376 248 : }
2377 :
2378 248 : async fn merge_entity_inner(
2379 248 : st: &AppState,
2380 248 : id: &str,
2381 248 : params: &HashMap<String, String>,
2382 248 : headers: &HeaderMap,
2383 248 : body: &[u8],
2384 248 : ) -> ApiResult<Response> {
2385 248 : let tenant = tenant_from(headers)?;
2386 248 : antares_model::EntityId::new(id)?;
2387 244 : check_params(
2388 244 : params,
2389 244 : &["options", "format", "observedAt", "lang", "local", "type"],
2390 0 : )?;
2391 244 : let parsed = parse_body(st, headers, body, BodyKind::MergePatch).await?;
2392 244 : let obj = parsed.object(NgsiError::BadRequestData(
2393 244 : "fragment must be a JSON object".into(),
2394 244 : ))?;
2395 244 : if let Some(bid) = obj.get("id").and_then(Value::as_str) {
2396 116 : if bid != id {
2397 0 : return Err(NgsiError::BadRequestData("fragment id mismatch".into()).into());
2398 116 : }
2399 128 : }
2400 244 : let mut fragment = expand_entity(
2401 244 : obj,
2402 244 : &parsed.ctx,
2403 244 : ExpandOpts {
2404 244 : fragment: true,
2405 244 : allow_null: true,
2406 244 : merge: true,
2407 244 : temporal: false,
2408 244 : ..Default::default()
2409 244 : },
2410 6 : )?;
2411 238 : let pre = crate::policy::pre_image(st, &tenant, id, Some(&fragment)).await?;
2412 238 : let granted = gate!(
2413 : st, &tenant, headers, "5.6.17",
2414 : ids: &[id],
2415 : types: &pre.types,
2416 : stored: pre.stored.as_ref(),
2417 : attrs: &crate::attrs::attr_iris_of(&fragment),
2418 : body: Some(&fragment),
2419 : )
2420 238 : .await?;
2421 236 : let cond = crate::policy::Condition::of(&granted);
2422 : // 5.6.17.3: a common observedAt timestamp to use across merged
2423 : // Attributes, and a common language tag for merged LanguageMaps.
2424 236 : let observed_at = params.get("observedAt").map(String::as_str);
2425 236 : if let Some(t) = observed_at {
2426 8 : if !antares_jsonld::parse_datetime(t) {
2427 4 : return Err(
2428 4 : NgsiError::BadRequestData("observedAt must be a DateTime (4.8)".into()).into(),
2429 4 : );
2430 4 : }
2431 228 : }
2432 232 : let lang = params.get("lang").map(String::as_str);
2433 232 : let simplified = simplified_input(params)?;
2434 230 : apply_common_observed_at(&mut fragment, observed_at);
2435 230 : let ts = now_iso();
2436 :
2437 : // 5.6.17.4: the ?type selector narrows the Context Sources as it
2438 : // narrows the local target, as for 5.6.6
2439 230 : let spec = crate::registry::CsrSpec {
2440 230 : ids: Some(vec![id.to_owned()]),
2441 230 : types: type_selection(params, &parsed.ctx).map(|sel| sel.into_iter().flatten().collect()),
2442 230 : ..Default::default()
2443 : };
2444 230 : let regs = match crate::federation::write_plan(st, &tenant, &spec, &parsed.ctx, params, headers)
2445 230 : .await?
2446 : {
2447 0 : crate::federation::WritePlan::Answered(r) => return Ok(*r),
2448 230 : crate::federation::WritePlan::Forward(regs) => regs,
2449 : };
2450 230 : crate::policy::Condition::unforwarded(cond.as_ref(), !regs.is_empty())?;
2451 230 : if !regs.is_empty() {
2452 18 : let proxies: Vec<&crate::federation::FedReg> =
2453 22 : regs.iter().filter(|r| r.is_proxy()).collect();
2454 18 : let mut parts = Vec::new();
2455 18 : let (rest, has_attrs) = crate::federation::strip_proxied(obj, &proxies, &parsed.ctx);
2456 : // 5.6.17.4: the target is "an existing Entity whose id (URI), and
2457 : // where specified type, is equivalent held locally" — the ?type
2458 : // selector narrows it on this path exactly as on the local-only one.
2459 18 : let local_exists = st
2460 18 : .store
2461 18 : .get(&tenant, Kind::Entity, id)
2462 18 : .await?
2463 18 : .is_some_and(|d| crate::negotiate::matches_type_param(&d, params, &parsed.ctx));
2464 18 : if (local_exists || proxies.is_empty()) && has_attrs {
2465 10 : let mut local_frag = expand_entity(
2466 10 : &rest,
2467 10 : &parsed.ctx,
2468 10 : ExpandOpts {
2469 10 : fragment: true,
2470 10 : allow_null: true,
2471 10 : merge: true,
2472 10 : temporal: false,
2473 10 : ..Default::default()
2474 10 : },
2475 0 : )?;
2476 10 : apply_common_observed_at(&mut local_frag, observed_at);
2477 10 : let res = st
2478 10 : .store
2479 10 : .mutate(&tenant, Kind::Entity, id, |doc| {
2480 10 : crate::history::restage(id);
2481 10 : if !crate::negotiate::matches_type_param(doc, params, &parsed.ctx) {
2482 0 : return Err(NgsiError::ResourceNotFound(format!(
2483 0 : "entity {id} does not match the type selector"
2484 0 : )));
2485 10 : }
2486 10 : let mut frag = local_frag.clone();
2487 10 : let raw = KeptTypes {
2488 10 : raw: &rest,
2489 10 : ctx: &parsed.ctx,
2490 10 : simplified,
2491 10 : lang,
2492 10 : observed_at,
2493 10 : };
2494 10 : raw.apply(doc, &mut frag)?;
2495 10 : apply_common_lang(doc, &mut frag, lang);
2496 10 : merge_into(doc, &frag, &ts);
2497 10 : Ok::<(), NgsiError>(())
2498 10 : })
2499 10 : .await?;
2500 10 : crate::history::settle(id, matches!(res, Some(Ok(()))));
2501 10 : parts.push(match res {
2502 10 : Some(Ok(())) => crate::federation::Part::local(204, "merged locally"),
2503 0 : _ => crate::federation::Part::local(404, format!("entity {id} not found locally")),
2504 : });
2505 8 : }
2506 18 : let ctx_url = crate::federation::ctx_link_url(headers, &parsed.ctx.source);
2507 18 : let seg = crate::federation::path_segment(id);
2508 18 : let fwd_q = type_selector_query(params);
2509 22 : for reg in ®s {
2510 : // 5.6.17.4: proxy modes without Merge Entity support are an
2511 : // error of type Conflict; inclusive ones are not forwarded.
2512 22 : if !reg.supports("mergeEntity") {
2513 4 : if reg.is_proxy() {
2514 2 : parts.push(crate::federation::conflict_part("mergeEntity", reg));
2515 2 : }
2516 4 : continue;
2517 18 : }
2518 18 : let Some(frag) = crate::federation::reduce_to_scope(obj, reg, &parsed.ctx) else {
2519 0 : continue;
2520 : };
2521 18 : parts.push(
2522 18 : crate::federation::forward_part(
2523 18 : st,
2524 18 : reqwest::Method::PATCH,
2525 18 : format!("{}/ngsi-ld/v1/entities/{seg}", reg.endpoint),
2526 18 : &fwd_q,
2527 18 : headers,
2528 18 : &tenant,
2529 18 : reg,
2530 18 : &ctx_url,
2531 18 : Some(frag),
2532 18 : )
2533 18 : .await,
2534 : );
2535 : }
2536 18 : return Ok(crate::federation::combine(
2537 18 : st.profile,
2538 18 : Some(id),
2539 18 : parts,
2540 18 : no_content(&tenant),
2541 18 : &tenant,
2542 18 : ));
2543 212 : }
2544 :
2545 212 : let res = st
2546 212 : .store
2547 212 : .mutate(&tenant, Kind::Entity, id, |doc| {
2548 198 : crate::history::restage(id);
2549 198 : crate::policy::Condition::pre(cond.as_ref(), id, doc)?;
2550 : // 5.6.17.4: the ?type selector narrows the merge target
2551 194 : if !crate::negotiate::matches_type_param(doc, params, &parsed.ctx) {
2552 0 : return Err(NgsiError::ResourceNotFound(format!(
2553 0 : "entity {id} does not match the type selector"
2554 0 : )));
2555 194 : }
2556 194 : let mut frag = fragment.clone();
2557 194 : let raw = KeptTypes {
2558 194 : raw: obj,
2559 194 : ctx: &parsed.ctx,
2560 194 : simplified,
2561 194 : lang,
2562 194 : observed_at,
2563 194 : };
2564 194 : raw.apply(doc, &mut frag)?;
2565 184 : apply_common_lang(doc, &mut frag, lang);
2566 184 : merge_into(doc, &frag, &ts);
2567 184 : Ok::<(), NgsiError>(())
2568 198 : })
2569 212 : .await?;
2570 212 : crate::history::settle(id, matches!(res, Some(Ok(()))));
2571 198 : match res {
2572 14 : None => Err(NgsiError::ResourceNotFound(format!("entity {id} not found")).into()),
2573 14 : Some(Err(e)) => Err(e.into()),
2574 184 : Some(Ok(())) => Ok(no_content(&tenant)),
2575 : }
2576 248 : }
2577 :
2578 : /// 5.6.17.3: "An optional parameter indicating a common observedAt timestamp
2579 : /// to use across merged Attributes." It applies to the Attribute instances of
2580 : /// the Fragment that do not carry an observedAt of their own; a deletion
2581 : /// instance (5.5.12 NGSI-LD Null) removes the Attribute and takes none.
2582 268 : fn apply_common_observed_at(fragment: &mut Value, observed_at: Option<&str>) {
2583 268 : let (Some(ts), Some(obj)) = (observed_at, fragment.as_object_mut()) else {
2584 264 : return;
2585 : };
2586 8 : for (k, v) in obj.iter_mut() {
2587 8 : if is_meta(k) {
2588 0 : continue;
2589 8 : }
2590 8 : let Some(instances) = v.as_array_mut() else {
2591 0 : continue;
2592 : };
2593 8 : for inst in instances {
2594 8 : if antares_jsonld::is_deletion_instance(inst) {
2595 0 : continue;
2596 8 : }
2597 8 : if let Some(o) = inst.as_object_mut() {
2598 8 : o.entry("observedAt".to_owned())
2599 8 : .or_insert_with(|| Value::String(ts.to_owned()));
2600 0 : }
2601 : }
2602 : }
2603 268 : }
2604 :
2605 : /// 5.6.17.4: "If a common language tag is defined and a LanguageProperty
2606 : /// Attribute to be merged is represented as a string, the pre-existing
2607 : /// languageMap JSON object shall be preserved. The string value shall only
2608 : /// replace the value associated to the language tag key found within the
2609 : /// languageMap." The string instance is rewritten into a one-key languageMap
2610 : /// patch, which 5.5.12 then merges into the stored map key by key.
2611 194 : fn apply_common_lang(target: &Value, fragment: &mut Value, lang: Option<&str>) {
2612 194 : let (Some(lang), Some(frag)) = (lang, fragment.as_object_mut()) else {
2613 188 : return;
2614 : };
2615 6 : for (k, v) in frag.iter_mut() {
2616 6 : if is_meta(k) {
2617 0 : continue;
2618 6 : }
2619 6 : let pre_existing_langmap = target
2620 6 : .get(k)
2621 6 : .and_then(Value::as_array)
2622 6 : .is_some_and(|insts| insts.iter().any(|i| i.get("languageMap").is_some()));
2623 6 : if !pre_existing_langmap {
2624 0 : continue;
2625 6 : }
2626 6 : let Some(instances) = v.as_array_mut() else {
2627 0 : continue;
2628 : };
2629 6 : for inst in instances {
2630 6 : if antares_jsonld::is_deletion_instance(inst) {
2631 0 : continue;
2632 6 : }
2633 6 : let Some(o) = inst.as_object_mut() else {
2634 0 : continue;
2635 : };
2636 6 : let Some(s) = o.get("value").and_then(Value::as_str).map(str::to_owned) else {
2637 2 : continue;
2638 : };
2639 4 : o.remove("value");
2640 4 : o.insert("type".into(), Value::String("LanguageProperty".into()));
2641 4 : let mut map = Map::new();
2642 4 : map.insert(lang.to_owned(), Value::String(s));
2643 4 : o.insert("languageMap".into(), Value::Object(map));
2644 : }
2645 : }
2646 194 : }
2647 :
2648 : /// TS 104 176 7.3.3.2 Table 7.3.3.2-1: `format` "shall be one of:
2649 : /// "simplified" (or its synonym "keyValues")", and so is the deprecated
2650 : /// `options=keyValues` spelling (6.3.7). Either says the Fragment holds the
2651 : /// simplified representation of ETSI TS 104 175 clause 5.3.2.4.
2652 232 : fn simplified_input(params: &HashMap<String, String>) -> Result<bool, NgsiError> {
2653 232 : let by_options = params.get("options").is_some_and(|o| {
2654 12 : o.split(',')
2655 12 : .any(|o| matches!(o.trim(), "keyValues" | "simplified"))
2656 12 : });
2657 232 : match params.get("format").map(String::as_str) {
2658 208 : None => Ok(by_options),
2659 24 : Some("simplified" | "keyValues") => Ok(true),
2660 2 : Some("normalized" | "concise") => Ok(by_options),
2661 2 : Some(f) => Err(NgsiError::BadRequestData(format!(
2662 2 : "format {f:?}: Merge Entity takes only \"simplified\" or \"keyValues\" \
2663 2 : (TS 104 176 7.3.3.2)"
2664 2 : ))),
2665 : }
2666 232 : }
2667 :
2668 : /// The Fragment members a Merge Entity reads by the type of the Attribute it
2669 : /// lands on, re-expanded inside the store transaction against the stored
2670 : /// Entity:
2671 : /// - 10.2.9.4: "If the Attribute to be merged is represented in a simplified
2672 : /// representation, the type of any pre-existing Attribute in the target
2673 : /// entity shall be preserved" (TS 104 176 7.3.3.2: "any attempt to modify
2674 : /// the type of an Attribute shall result in a BadRequest error"). A bare
2675 : /// value goes into the member the stored type holds it in, and a value
2676 : /// that type cannot hold is BadRequestData from the same expansion a
2677 : /// normalized write runs.
2678 : /// - 4.7.1: a GeoProperty's Value "shall be a GeoJSON Geometry". A type-less
2679 : /// `{"value": ...}` onto a stored GeoProperty is validated as one; read as
2680 : /// a plain Property it stored `{"type":"Polygon"}` with no coordinates.
2681 : struct KeptTypes<'a> {
2682 : raw: &'a Map<String, Value>,
2683 : ctx: &'a antares_jsonld::Context,
2684 : simplified: bool,
2685 : lang: Option<&'a str>,
2686 : observed_at: Option<&'a str>,
2687 : }
2688 :
2689 : impl KeptTypes<'_> {
2690 204 : fn apply(&self, doc: &Value, frag: &mut Value) -> Result<(), NgsiError> {
2691 204 : let mut kept = Map::new();
2692 204 : let mut iris = Vec::new();
2693 558 : for (k, v) in self.raw {
2694 558 : if k == "@context" || is_meta(k) || is_ngsi_null(v) {
2695 382 : continue;
2696 176 : }
2697 176 : let iri = antares_jsonld::expand_attr_name(k, self.ctx)?;
2698 176 : let Some(stored) = doc
2699 176 : .get(&iri)
2700 176 : .and_then(Value::as_array)
2701 176 : .and_then(|a| a.first())
2702 176 : .and_then(|i| i.get("type"))
2703 176 : .and_then(Value::as_str)
2704 : else {
2705 12 : continue;
2706 : };
2707 164 : let typeless = v
2708 164 : .as_object()
2709 164 : .filter(|o| !o.contains_key("type") && !o.contains_key("@type"));
2710 164 : let normalized = if self.simplified {
2711 34 : self.simplified_as(&iri, stored, v)?
2712 4 : } else if let Some(o) =
2713 130 : typeless.filter(|o| stored == "GeoProperty" && o.contains_key("value"))
2714 : {
2715 4 : let mut o = o.clone();
2716 4 : o.insert("type".into(), Value::String(stored.to_owned()));
2717 4 : Value::Object(o)
2718 : } else {
2719 126 : continue;
2720 : };
2721 36 : kept.insert(k.clone(), normalized);
2722 36 : iris.push(iri);
2723 : }
2724 202 : if kept.is_empty() {
2725 166 : return Ok(());
2726 36 : }
2727 36 : let mut expanded = expand_entity(
2728 36 : &kept,
2729 36 : self.ctx,
2730 36 : ExpandOpts {
2731 36 : fragment: true,
2732 36 : allow_null: true,
2733 36 : merge: true,
2734 36 : temporal: false,
2735 36 : ..Default::default()
2736 36 : },
2737 8 : )?;
2738 28 : apply_common_observed_at(&mut expanded, self.observed_at);
2739 28 : if let (Some(f), Some(e)) = (frag.as_object_mut(), expanded.as_object_mut()) {
2740 28 : for iri in iris {
2741 28 : if let Some(v) = e.remove(&iri) {
2742 28 : f.insert(iri, v);
2743 28 : }
2744 : }
2745 0 : }
2746 28 : Ok(())
2747 204 : }
2748 :
2749 : /// The normalized Attribute a simplified value of a stored `stored`
2750 : /// Attribute stands for (TS 104 175 5.3.2.4: a LanguageProperty,
2751 : /// JsonProperty or VocabProperty may arrive as `{"languageMap": ...}`,
2752 : /// `{"json": ...}` or `{"vocab": ...}`; every other type as its bare
2753 : /// value). 10.2.9.4: a LanguageProperty sent "as a string" updates the
2754 : /// key of the common language tag, so without `lang` there is no key.
2755 34 : fn simplified_as(&self, iri: &str, stored: &str, v: &Value) -> Result<Value, NgsiError> {
2756 34 : let member = match stored {
2757 34 : "Relationship" => "object",
2758 28 : "ListProperty" => "valueList",
2759 22 : "ListRelationship" => "objectList",
2760 18 : "LanguageProperty" => "languageMap",
2761 14 : "VocabProperty" => "vocab",
2762 10 : "JsonProperty" => "json",
2763 6 : _ => "value",
2764 : };
2765 34 : let self_described = matches!(
2766 34 : stored,
2767 34 : "LanguageProperty" | "VocabProperty" | "JsonProperty"
2768 12 : ) && v
2769 12 : .as_object()
2770 12 : .is_some_and(|o| o.len() == 1 && o.contains_key(member));
2771 34 : let mut out = if self_described {
2772 4 : v.as_object().cloned().unwrap_or_default()
2773 30 : } else if stored == "LanguageProperty" {
2774 4 : let Some(tag) = self.lang else {
2775 2 : return Err(NgsiError::BadRequestData(format!(
2776 2 : "attribute {iri}: a simplified value for a LanguageProperty needs the \
2777 2 : lang URL parameter to name its languageMap key (TS 104 176 7.3.3.2)"
2778 2 : )));
2779 : };
2780 2 : let mut m = Map::new();
2781 2 : m.insert(member.into(), json!({ tag: v }));
2782 2 : m
2783 : } else {
2784 26 : let mut m = Map::new();
2785 26 : m.insert(member.into(), v.clone());
2786 26 : m
2787 : };
2788 32 : out.insert("type".into(), Value::String(stored.to_owned()));
2789 32 : Ok(Value::Object(out))
2790 34 : }
2791 : }
2792 :
2793 : /// JSON Merge-Patch over internal docs (5.5.12).
2794 302 : pub fn merge_into(doc: &mut Value, fragment: &Value, ts: &str) {
2795 302 : merge_with(doc, fragment, ts, false);
2796 302 : }
2797 :
2798 : /// 5.6.9 Update (5.6.3 per Entity): as [`merge_into`], but an instance the
2799 : /// fragment names replaces the stored one whole (its `createdAt` kept), so a
2800 : /// member the update leaves out (`unitCode`) is gone, never merged over.
2801 112 : pub fn update_into(doc: &mut Value, fragment: &Value, ts: &str) {
2802 112 : merge_with(doc, fragment, ts, true);
2803 112 : }
2804 :
2805 414 : fn merge_with(doc: &mut Value, fragment: &Value, ts: &str, replace: bool) {
2806 : // The Entity as a backfill names it: id and type, before the merge.
2807 414 : let shell = serde_json::json!({"id": doc.get("id"), "type": doc.get("type")});
2808 414 : let (Some(target), Some(frag)) = (doc.as_object_mut(), fragment.as_object()) else {
2809 0 : return;
2810 : };
2811 1188 : for (k, v) in frag {
2812 1188 : match k.as_str() {
2813 1188 : "id" | "createdAt" | "modifiedAt" => continue,
2814 900 : "type" => {
2815 : // union of types
2816 310 : let mut cur: Vec<Value> = target
2817 310 : .get("type")
2818 310 : .and_then(Value::as_array)
2819 310 : .cloned()
2820 310 : .unwrap_or_default();
2821 314 : for t in v.as_array().cloned().unwrap_or_default() {
2822 314 : if !cur.contains(&t) {
2823 20 : cur.push(t);
2824 294 : }
2825 : }
2826 310 : target.insert("type".into(), Value::Array(cur));
2827 : }
2828 : // 5.5.12: "For each member of the Fragment, whose value is an
2829 : // NGSI-LD Null, contained by the target, the target member is
2830 : // removed." 4.18 admits the sentinel as a scope for exactly this
2831 : // reason, and expansion hands scopes over as an array, so the
2832 : // deletion arrives as its single entry. Storing it instead left
2833 : // the Entity scoped to a string no 4.18 grammar accepts.
2834 590 : "scope" => {
2835 34 : if antares_jsonld::is_ngsi_null_list(v) {
2836 8 : target.remove("scope");
2837 26 : } else {
2838 26 : target.insert("scope".into(), v.clone());
2839 26 : }
2840 : }
2841 : // 4.22: expiresAt is a settable Entity member (5.2.4, not in the
2842 : // read-only Table 5.2.2-1) — merge updates the storage expiry;
2843 : // an NGSI-LD Null removes it (5.5.12). Without this arm it fell
2844 : // through to the attribute path, where a bare string has no
2845 : // instances and the member was silently dropped.
2846 556 : "expiresAt" => {
2847 32 : if is_ngsi_null(v) {
2848 8 : target.remove("expiresAt");
2849 24 : } else {
2850 24 : target.insert("expiresAt".into(), v.clone());
2851 24 : }
2852 : }
2853 : _ => {
2854 524 : let frag_instances = v.as_array().cloned().unwrap_or_default();
2855 524 : let mut cur: Vec<Value> = target
2856 524 : .get(k)
2857 524 : .and_then(Value::as_array)
2858 524 : .cloned()
2859 524 : .unwrap_or_default();
2860 540 : for fi in frag_instances {
2861 540 : let is_delete = antares_jsonld::is_deletion_instance(&fi);
2862 540 : let want_ds = fi.get("datasetId").and_then(Value::as_str);
2863 540 : let pos = cur
2864 540 : .iter()
2865 540 : .position(|ci| ci.get("datasetId").and_then(Value::as_str) == want_ds);
2866 540 : match (is_delete, pos) {
2867 68 : (true, Some(p)) => {
2868 68 : cur.remove(p);
2869 68 : }
2870 0 : (true, None) => {}
2871 354 : (false, Some(p)) => {
2872 354 : if let Some(ci) = cur.get_mut(p) {
2873 354 : if crate::history::backfilled(&shell, k, ci, &fi) {
2874 4 : // older than the stored one: history only
2875 350 : } else if replace {
2876 142 : let created = ci.get("createdAt").cloned();
2877 142 : let mut ni = Value::Array(vec![fi.clone()]);
2878 142 : crate::stamp::stamp_instances(&mut ni, ts);
2879 142 : if let Some(Value::Object(mut o)) =
2880 142 : ni.as_array_mut().and_then(Vec::pop)
2881 : {
2882 142 : if let Some(c) = created {
2883 142 : o.insert("createdAt".into(), c);
2884 142 : }
2885 142 : *ci = Value::Object(o);
2886 0 : }
2887 208 : } else {
2888 208 : merge_instance(ci, &fi, ts);
2889 208 : }
2890 0 : }
2891 : }
2892 : (false, None) => {
2893 118 : let mut ni = fi.clone();
2894 118 : if let Some(o) = ni.as_object_mut() {
2895 118 : o.insert("createdAt".into(), Value::String(ts.to_owned()));
2896 118 : o.insert("modifiedAt".into(), Value::String(ts.to_owned()));
2897 118 : }
2898 118 : cur.push(ni);
2899 : }
2900 : }
2901 : }
2902 524 : if cur.is_empty() {
2903 36 : target.remove(k);
2904 488 : } else {
2905 488 : target.insert(k.clone(), Value::Array(cur));
2906 488 : }
2907 : }
2908 : }
2909 : }
2910 414 : target.insert("modifiedAt".into(), Value::String(ts.to_owned()));
2911 414 : }
2912 :
2913 208 : fn merge_instance(target: &mut Value, frag: &Value, ts: &str) {
2914 208 : let (Some(t), Some(f)) = (target.as_object_mut(), frag.as_object()) else {
2915 0 : return;
2916 : };
2917 : // An instance of another attribute type (a LanguageProperty sent over a
2918 : // stored Property) is a new attribute, not a patch of the old one: a key
2919 : // by key merge kept the Property's value beside the new languageMap,
2920 : // a shape no NGSI-LD type allows. It replaces the instance and
2921 : // keeps only the instant the attribute was created (4.8).
2922 208 : let changes_type = f
2923 208 : .get("type")
2924 208 : .is_some_and(|ft| t.get("type").is_some_and(|tt| tt != ft));
2925 208 : if changes_type {
2926 8 : let created = t.get("createdAt").cloned();
2927 8 : t.clear();
2928 16 : for (k, v) in f {
2929 16 : if k != "createdAt" && k != "modifiedAt" && !antares_jsonld::removes_member(v) {
2930 16 : t.insert(k.clone(), v.clone());
2931 16 : }
2932 : }
2933 8 : if let Some(c) = created {
2934 8 : t.insert("createdAt".into(), c);
2935 8 : }
2936 8 : t.insert("modifiedAt".into(), Value::String(ts.to_owned()));
2937 8 : return;
2938 200 : }
2939 528 : for (k, v) in f {
2940 528 : if k == "createdAt" || k == "modifiedAt" {
2941 0 : continue;
2942 528 : }
2943 : // a sub-Attribute whose instances are all deletions leaves too (10.2.9.4)
2944 528 : if antares_jsonld::removes_member(v) {
2945 4 : t.remove(k);
2946 24 : } else if let (Some(cur), Some(patch)) =
2947 524 : (t.get_mut(k).and_then(Value::as_object_mut), v.as_object())
2948 24 : {
2949 24 : // 5.5.12: the merge goes "into JSON objects representing a
2950 24 : // Property value" — RFC 7396 with the NGSI-LD Null as removal.
2951 24 : merge_value_object(cur, patch);
2952 500 : } else {
2953 500 : t.insert(k.clone(), v.clone());
2954 500 : }
2955 : }
2956 200 : t.insert("modifiedAt".into(), Value::String(ts.to_owned()));
2957 208 : }
2958 :
2959 : /// RFC 7396 merge patch over a compound (JSON object) member value, with
2960 : /// "urn:ngsi-ld:null" / JSON null as the key-removal marker (5.5.12); the
2961 : /// sentinel itself is never stored (5.5.4).
2962 24 : fn merge_value_object(target: &mut Map<String, Value>, patch: &Map<String, Value>) {
2963 34 : for (k, v) in patch {
2964 34 : if v.is_null() || is_ngsi_null(v) {
2965 8 : target.remove(k);
2966 8 : } else if let (Some(cur), Some(po)) = (
2967 26 : target.get_mut(k).and_then(Value::as_object_mut),
2968 26 : v.as_object(),
2969 0 : ) {
2970 0 : merge_value_object(cur, po);
2971 26 : } else {
2972 26 : target.insert(k.clone(), v.clone());
2973 26 : }
2974 : }
2975 24 : }
2976 :
2977 : // ---------- PUT /entities/{id} — Replace (5.6.18) ----------
2978 :
2979 82 : pub async fn replace_entity(
2980 82 : State(st): State<AppState>,
2981 82 : Path(id): Path<String>,
2982 82 : CleanParams(params): CleanParams,
2983 82 : headers: HeaderMap,
2984 82 : body: Bytes,
2985 82 : ) -> Response {
2986 82 : let go = async {
2987 82 : let tenant = tenant_from(&headers)?;
2988 82 : antares_model::EntityId::new(&id)?;
2989 78 : check_params(¶ms, &["local", "type"])?;
2990 : // 5.5.7 again: the selector is expanded with the request's @context,
2991 : // not the core one. The target is judged before the body is parsed
2992 : // (5.6.18: an unknown target is 404 before body validation), so the
2993 : // header context is resolved on its own here.
2994 78 : let ctx0 = request_context(&st.loader, &headers).await?;
2995 : // The engine is shown the replacement. The body is read for it
2996 : // leniently: a body that does not parse is not the engine's to
2997 : // judge, and the replace below refuses it as before, after the
2998 : // 404 an unknown target answers first (057_03).
2999 78 : let shown = match &st.policy {
3000 30 : Some(_) => parse_body(&st, &headers, &body, BodyKind::Standard)
3001 30 : .await
3002 30 : .ok()
3003 30 : .and_then(|p| {
3004 30 : let obj = p.object(NgsiError::BadRequestData(String::new())).ok()?;
3005 30 : expand_entity(obj, &p.ctx, ExpandOpts::default()).ok()
3006 30 : }),
3007 48 : None => None,
3008 : };
3009 78 : let pre = crate::policy::pre_image(&st, &tenant, &id, shown.as_ref()).await?;
3010 78 : let granted = gate!(
3011 : st, &tenant, &headers, "5.6.18",
3012 : ids: &[&id],
3013 : types: &pre.types,
3014 : stored: pre.stored.as_ref(),
3015 : attrs: &shown.as_ref().map(crate::attrs::attr_iris_of).unwrap_or_default(),
3016 : body: shown.as_ref(),
3017 : )
3018 78 : .await?;
3019 68 : let cond = crate::policy::Condition::of(&granted);
3020 : // 5.6.18.4: the ?type selector narrows the target — a non-matching
3021 : // entity is "not known" for this replace.
3022 68 : let local_doc = st
3023 68 : .store
3024 68 : .get(&tenant, Kind::Entity, &id)
3025 68 : .await?
3026 68 : .filter(|d| crate::negotiate::matches_type_param(d, ¶ms, &ctx0))
3027 : // An Entity the engine's condition does not select is
3028 : // not found, before the body is read, as an absent target is
3029 68 : .filter(|d| cond.as_ref().is_none_or(|c| c.holds(d)));
3030 : // the same selector narrows the Context Sources the replace goes to
3031 68 : let spec = crate::registry::CsrSpec {
3032 68 : ids: Some(vec![id.clone()]),
3033 68 : types: type_selection(¶ms, &ctx0).map(|sel| sel.into_iter().flatten().collect()),
3034 68 : ..Default::default()
3035 : };
3036 68 : let regs =
3037 68 : match crate::federation::write_plan(&st, &tenant, &spec, &ctx0, ¶ms, &headers)
3038 68 : .await?
3039 : {
3040 0 : crate::federation::WritePlan::Answered(r) => return Ok(*r),
3041 68 : crate::federation::WritePlan::Forward(regs) => regs,
3042 : };
3043 68 : crate::policy::Condition::unforwarded(cond.as_ref(), !regs.is_empty())?;
3044 68 : if regs.is_empty() {
3045 : // 5.6.18: an unknown target is 404 before body validation (057_03).
3046 : // The read above answers that; the write below decides again
3047 : // under the row lock, because between the two the target can be
3048 : // deleted and a replace that writes anyway puts it back.
3049 50 : if local_doc.is_none() {
3050 24 : return Err(NgsiError::ResourceNotFound(format!("entity {id} not found")).into());
3051 26 : }
3052 26 : let parsed = parse_body(&st, &headers, &body, BodyKind::Standard).await?;
3053 26 : let obj = parsed.object(NgsiError::BadRequestData(
3054 26 : "entity must be a JSON object".into(),
3055 26 : ))?;
3056 26 : let mut expanded = expand_entity(obj, &parsed.ctx, ExpandOpts::default())?;
3057 26 : if expanded.get("id").and_then(Value::as_str) != Some(id.as_str()) {
3058 0 : return Err(NgsiError::BadRequestData("entity id mismatch".into()).into());
3059 26 : }
3060 26 : let ts = now_iso();
3061 26 : stamp_new(&mut expanded, &ts);
3062 26 : let res = st
3063 26 : .store
3064 26 : .mutate(&tenant, Kind::Entity, &id, |doc| {
3065 24 : crate::policy::Condition::pre(cond.as_ref(), &id, doc)?;
3066 : // 5.6.18.4: the ?type selector narrows the target here too —
3067 : // the type of the row being written is the one that counts
3068 24 : if !crate::negotiate::matches_type_param(doc, ¶ms, &ctx0) {
3069 0 : return Err(NgsiError::ResourceNotFound(format!(
3070 0 : "entity {id} does not match the type selector"
3071 0 : )));
3072 24 : }
3073 : // 4.8: "createdAt ... shall be the date and time at which the
3074 : // Entity was created" — the target's own stamp, read under
3075 : // the lock rather than from a snapshot another write may
3076 : // already have replaced.
3077 24 : if let (Some(o), Some(created)) =
3078 24 : (expanded.as_object_mut(), doc.get("createdAt").cloned())
3079 24 : {
3080 24 : o.insert("createdAt".into(), created);
3081 24 : }
3082 24 : *doc = expanded.clone();
3083 24 : Ok::<(), NgsiError>(())
3084 24 : })
3085 26 : .await?;
3086 24 : return match res {
3087 2 : None => Err(NgsiError::ResourceNotFound(format!("entity {id} not found")).into()),
3088 0 : Some(Err(e)) => Err(e.into()),
3089 24 : Some(Ok(())) => Ok::<_, ApiError>(no_content(&tenant)),
3090 : };
3091 18 : }
3092 18 : let parsed = parse_body(&st, &headers, &body, BodyKind::Standard).await?;
3093 18 : let obj = parsed.object(NgsiError::BadRequestData(
3094 18 : "entity must be a JSON object".into(),
3095 18 : ))?;
3096 18 : let expanded = expand_entity(obj, &parsed.ctx, ExpandOpts::default())?;
3097 18 : if expanded.get("id").and_then(Value::as_str) != Some(id.as_str()) {
3098 0 : return Err(NgsiError::BadRequestData("entity id mismatch".into()).into());
3099 18 : }
3100 18 : let mut parts = Vec::new();
3101 18 : let proxies: Vec<&crate::federation::FedReg> =
3102 22 : regs.iter().filter(|r| r.is_proxy()).collect();
3103 18 : let proxy_match = !proxies.is_empty();
3104 18 : if local_doc.is_some() || !proxy_match {
3105 12 : let gone =
3106 12 : crate::federation::Part::local(404, format!("entity {id} not found locally"));
3107 12 : if local_doc.is_none() {
3108 0 : parts.push(gone);
3109 0 : } else {
3110 12 : let (rest, _) = crate::federation::strip_proxied(obj, &proxies, &parsed.ctx);
3111 12 : let mut local_exp = expand_entity(&rest, &parsed.ctx, ExpandOpts::default())?;
3112 12 : let ts = now_iso();
3113 12 : stamp_new(&mut local_exp, &ts);
3114 : // the same row lock as the local-only path above: the read
3115 : // that found the target is not the write that replaces it
3116 12 : let res = st
3117 12 : .store
3118 12 : .mutate(&tenant, Kind::Entity, &id, |doc| {
3119 12 : if !crate::negotiate::matches_type_param(doc, ¶ms, &ctx0) {
3120 0 : return Err(NgsiError::ResourceNotFound(format!(
3121 0 : "entity {id} does not match the type selector"
3122 0 : )));
3123 12 : }
3124 12 : if let (Some(o), Some(created)) =
3125 12 : (local_exp.as_object_mut(), doc.get("createdAt").cloned())
3126 12 : {
3127 12 : o.insert("createdAt".into(), created);
3128 12 : }
3129 12 : *doc = local_exp.clone();
3130 12 : Ok::<(), NgsiError>(())
3131 12 : })
3132 12 : .await?;
3133 12 : match res {
3134 : Some(Ok(())) => {
3135 12 : parts.push(crate::federation::Part::local(204, "replaced locally"))
3136 : }
3137 0 : _ => parts.push(gone),
3138 : }
3139 : }
3140 6 : }
3141 18 : let ctx_url = crate::federation::ctx_link_url(&headers, &parsed.ctx.source);
3142 18 : let seg = crate::federation::path_segment(&id);
3143 18 : let fwd_q = type_selector_query(¶ms);
3144 22 : for reg in ®s {
3145 : // 5.6.18.4: proxy modes without Replace Entity support are an
3146 : // error of type Conflict; inclusive ones are not forwarded.
3147 22 : if !reg.supports("replaceEntity") {
3148 2 : if reg.is_proxy() {
3149 2 : parts.push(crate::federation::conflict_part("replaceEntity", reg));
3150 2 : }
3151 2 : continue;
3152 20 : }
3153 20 : let Some(frag) = crate::federation::reduce_to_scope(obj, reg, &parsed.ctx) else {
3154 0 : continue;
3155 : };
3156 20 : parts.push(
3157 20 : crate::federation::forward_part(
3158 20 : &st,
3159 20 : reqwest::Method::PUT,
3160 20 : format!("{}/ngsi-ld/v1/entities/{seg}", reg.endpoint),
3161 20 : &fwd_q,
3162 20 : &headers,
3163 20 : &tenant,
3164 20 : reg,
3165 20 : &ctx_url,
3166 20 : Some(frag),
3167 20 : )
3168 20 : .await,
3169 : );
3170 : }
3171 18 : Ok(crate::federation::combine(
3172 18 : st.profile,
3173 18 : Some(&id),
3174 18 : parts,
3175 18 : no_content(&tenant),
3176 18 : &tenant,
3177 18 : ))
3178 82 : };
3179 82 : go.await.unwrap_or_else(|e| e.into_response())
3180 82 : }
3181 :
3182 : // ---------- Entity Ordering (4.23) ----------
3183 :
3184 : // ---------- GeoJSON output (6.3.15) ----------
3185 :
3186 : // ---------- GET /entities/{id}/attrs/{attrId} [+ /value] ----------
3187 : // NGSI-LD 2.0 pre-adoptions #14/#15: retrieve a single
3188 : // attribute of an entity, and its bare value. Additive-only: 2.0 defines the
3189 : // resources, 1.9.1 clients never see them unless asked.
3190 :
3191 42 : pub async fn retrieve_entity_attr(
3192 42 : State(st): State<AppState>,
3193 42 : Path((id, attr)): Path<(String, String)>,
3194 42 : CleanParams(params): CleanParams,
3195 42 : headers: HeaderMap,
3196 42 : ) -> Response {
3197 42 : match retrieve_attr_inner(&st, &id, &attr, false, ¶ms, &headers).await {
3198 14 : Ok(r) => r,
3199 28 : Err(e) => e.into_response(),
3200 : }
3201 42 : }
3202 :
3203 30 : pub async fn retrieve_entity_attr_value(
3204 30 : State(st): State<AppState>,
3205 30 : Path((id, attr)): Path<(String, String)>,
3206 30 : CleanParams(params): CleanParams,
3207 30 : headers: HeaderMap,
3208 30 : ) -> Response {
3209 30 : match retrieve_attr_inner(&st, &id, &attr, true, ¶ms, &headers).await {
3210 8 : Ok(r) => r,
3211 22 : Err(e) => e.into_response(),
3212 : }
3213 30 : }
3214 :
3215 72 : async fn retrieve_attr_inner(
3216 72 : st: &AppState,
3217 72 : id: &str,
3218 72 : attr: &str,
3219 72 : value_only: bool,
3220 72 : params: &HashMap<String, String>,
3221 72 : headers: &HeaderMap,
3222 72 : ) -> ApiResult<Response> {
3223 72 : let tenant = tenant_from(headers)?;
3224 72 : check_params(params, &["options", "format", "lang", "datasetId", "local"])?;
3225 72 : let ctx = request_context(&st.loader, headers).await?;
3226 72 : let filter = gate!(
3227 : st, &tenant, headers, "5.7.1",
3228 : ids: &[id],
3229 : attrs: &antares_jsonld::expand_attr_name(attr, &ctx).ok().into_iter().collect::<Vec<_>>(),
3230 : )
3231 72 : .await?;
3232 72 : let mut repr = parse_repr(params, &ctx)?;
3233 72 : crate::repr::narrow_repr(&mut repr, &filter);
3234 72 : antares_model::EntityId::new(id)?;
3235 72 : antares_model::check_attr_name(attr)?;
3236 72 : let doc = st
3237 72 : .store
3238 72 : .get(&tenant, Kind::Entity, id)
3239 72 : .await?
3240 72 : .ok_or_else(|| NgsiError::ResourceNotFound(format!("entity {id} not found")))?;
3241 : // ADR-0020: the Attribute of an Entity outside the
3242 : // narrowing is the Attribute of an absent one, as Retrieve Entity
3243 : // answers it, whatever the Attribute's own visibility.
3244 36 : if !crate::repr::policy_admits(st, &tenant, &ctx, repr.policy_reach.as_deref(), &doc).await {
3245 4 : return Err(NgsiError::ResourceNotFound(format!("entity {id} not found")).into());
3246 32 : }
3247 32 : let attr_iri = antares_jsonld::expand_attr_name(attr, &ctx)?;
3248 : // One answer for an Attribute the Entity lacks and one the narrowing
3249 : // hides.
3250 28 : let lacks = || NgsiError::ResourceNotFound(format!("entity {id} has no attribute {attr}"));
3251 28 : let node = doc.get(&attr_iri).ok_or_else(lacks)?;
3252 : // Compact through the entity pipeline so the attribute serializes exactly
3253 : // as it would inside a full retrieve.
3254 22 : let mini = serde_json::json!({
3255 22 : "id": doc.get("id").cloned().unwrap_or_default(),
3256 22 : "type": doc.get("type").cloned().unwrap_or_default(),
3257 22 : attr_iri.clone(): node.clone(),
3258 : });
3259 22 : let shaped = crate::repr::apply(&mini, &repr);
3260 22 : let compacted = compact_for(&repr, &shaped, &ctx);
3261 22 : let key = ctx.compact_iri(&attr_iri);
3262 22 : let member = compacted.get(&key).cloned().ok_or_else(lacks)?;
3263 22 : let body = if value_only {
3264 : // #15: the bare value — value / object / languageMap, whichever the
3265 : // attribute type carries; multi-instance attributes yield an array.
3266 12 : fn bare(v: &Value) -> Value {
3267 12 : match v {
3268 2 : Value::Array(a) => Value::Array(a.iter().map(bare).collect()),
3269 10 : Value::Object(o) => o
3270 10 : .get("value")
3271 10 : .or_else(|| o.get("object"))
3272 10 : .or_else(|| o.get("languageMap"))
3273 10 : .or_else(|| o.get("json"))
3274 10 : .or_else(|| o.get("vocab"))
3275 10 : .or_else(|| o.get("valueList"))
3276 10 : .or_else(|| o.get("objectList"))
3277 10 : .cloned()
3278 10 : .unwrap_or(Value::Null),
3279 0 : other => other.clone(),
3280 : }
3281 12 : }
3282 8 : bare(&member)
3283 : } else {
3284 14 : member
3285 : };
3286 22 : let accept = parse_accept(headers)?;
3287 22 : let mut resp = respond(StatusCode::OK, body, &ctx, accept, &tenant);
3288 22 : filter.mark(&st.host_alias, resp.headers_mut());
3289 22 : Ok(resp)
3290 72 : }
3291 :
3292 : /// 5.14.4.4: run the (split-reduced when applicable) local query and record
3293 : /// each matching id under the "@none" local marker; forward to matching
3294 : /// registrations supporting createEntityMapQueryEntity and merge each
3295 : /// returned EntityMap (ids → registration id, linkedMaps → remote map id);
3296 : /// store the local EntityMap and return it.
3297 : /// Known ceiling: the local candidate ids are the first max_limit matches —
3298 : /// the query is paged into the store instead of materializing every matching
3299 : /// Entity document, so one request cannot pull a whole tenant into memory.
3300 : /// Raise the cap if local candidate sets outgrow it.
3301 192 : pub(crate) async fn build_query_map(
3302 192 : st: &AppState,
3303 192 : tenant: &TenantId,
3304 192 : headers: &HeaderMap,
3305 192 : ctx: &antares_jsonld::Context,
3306 192 : params: &HashMap<String, String>,
3307 192 : filter: &crate::policy::Filter,
3308 192 : ) -> ApiResult<Value> {
3309 192 : let q_ast = params
3310 192 : .get("q")
3311 192 : .map(|q| antares_ql::parse_q(q))
3312 192 : .transpose()?;
3313 : // 5.14.4.4 a-e: too wide query
3314 192 : if !qualifies_non_wide(params, q_ast.as_ref()) {
3315 6 : return Err(NgsiError::BadRequestData(
3316 6 : "EntityMap query needs at least one of type, attrs, q, georel, or local=true \
3317 6 : (5.14.4.4 — too wide query)"
3318 6 : .into(),
3319 6 : )
3320 6 : .into());
3321 186 : }
3322 : // the candidate set is the NARROWED query's (ADR-0020)
3323 186 : let narrowed = filter.narrow_params(params, ctx)?;
3324 186 : let narrow = crate::repr::narrowing_reading(filter, params, ctx)?;
3325 186 : let params = &narrowed;
3326 : // 5.14.4.4: invalid entity ids / csf are BadRequestData
3327 186 : if let Some(ids) = params.get("id") {
3328 0 : for id in ids.split(',') {
3329 0 : antares_model::EntityId::new(id.trim())?;
3330 : }
3331 186 : }
3332 186 : if let Some(csf) = params.get("csf") {
3333 0 : antares_ql::parse_q(csf)?;
3334 186 : }
3335 186 : let local_scope = params.get("local").map(String::as_str) == Some("true");
3336 186 : let split = params.get("splitEntities").map(String::as_str) == Some("true");
3337 : // Split entities: only id/type/idPattern narrow the local candidate set —
3338 : // value/geo/scope filters cannot be judged on a fragment (5.14.4.4).
3339 186 : let eff: HashMap<String, String> = if split && !local_scope {
3340 0 : params
3341 0 : .iter()
3342 0 : .filter(|(k, _)| ["id", "idPattern", "type", "local"].contains(&k.as_str()))
3343 0 : .map(|(k, v)| (k.clone(), v.clone()))
3344 0 : .collect()
3345 : } else {
3346 186 : params.clone()
3347 : };
3348 : // idPattern is invisible to the store, so a pushed page cannot BE the
3349 : // candidate set's page — it is still the window a walk steps through, and
3350 : // the literal the pattern forces has already narrowed the rows (5.2.33).
3351 : // Without the walk a tenant-wide pattern arrives as one allocation of
3352 : // every Entity the rest of the filter matched.
3353 186 : let mut emap = Map::new();
3354 186 : let mut offset = 0usize;
3355 196 : while emap.len() < st.max_limit {
3356 192 : let batch = filter_entities_paged(
3357 192 : st,
3358 192 : tenant,
3359 192 : &eff,
3360 192 : ctx,
3361 192 : Vec::new(),
3362 192 : Pushdown {
3363 192 : page: Some((offset, st.max_limit)),
3364 192 : proj: None,
3365 192 : order_keys: None,
3366 192 : },
3367 192 : Some(&narrow),
3368 192 : )
3369 192 : .await?;
3370 310 : for d in &batch.docs {
3371 310 : if emap.len() == st.max_limit {
3372 0 : break;
3373 310 : }
3374 310 : if let Some(id) = d.get("id").and_then(Value::as_str) {
3375 310 : // "@none" refers to an Entity held locally (5.2.39)
3376 310 : emap.insert(id.to_owned(), json!(["@none"]));
3377 310 : }
3378 : }
3379 192 : match next_scan_offset(offset, batch.rows, 0, batch.paged, st.max_limit) {
3380 10 : Some(next) => offset = next,
3381 182 : None => break,
3382 : }
3383 : }
3384 186 : crate::entity_map::merge_and_store_map(st, tenant, headers, ctx, params, false, emap).await
3385 192 : }
3386 :
3387 : /// The store page of the conditional-rules scan as the store's `i64` binds,
3388 : /// saturating: `enough` is the client's offset plus limit, so an offset near
3389 : /// `i64::MAX` (which `page_params` admits) and the doubling limit pass it,
3390 : /// and an `as` cast wraps negative, a LIMIT postgres refuses (500).
3391 380 : fn store_page((offset, limit): (usize, usize)) -> antares_store::filter::Page {
3392 380 : antares_store::filter::Page {
3393 380 : offset: i64::try_from(offset).unwrap_or(i64::MAX),
3394 380 : limit: i64::try_from(limit).unwrap_or(i64::MAX),
3395 380 : count: false,
3396 380 : }
3397 380 : }
3398 :
3399 : #[cfg(test)]
3400 : mod tests {
3401 : #[test]
3402 4 : fn the_scan_page_past_i64_saturates_and_never_wraps_negative() {
3403 4 : let past = usize::try_from(i64::MAX).unwrap_or(usize::MAX) + 11;
3404 4 : let p = super::store_page((past, usize::MAX));
3405 4 : assert_eq!((p.offset, p.limit), (i64::MAX, i64::MAX));
3406 4 : let p = super::store_page((20, 41));
3407 4 : assert_eq!((p.offset, p.limit, p.count), (20, 41, false));
3408 4 : }
3409 :
3410 : use super::merge_into;
3411 : use serde_json::json;
3412 :
3413 : /// 5.5.12 EXAMPLE 1 + the datasetId/type bullets: a merge updates the
3414 : /// named sub-attributes and leaves the others untouched; a fragment
3415 : /// instance with an unknown datasetId is ADDED (not replacing the
3416 : /// default); entity types are unioned.
3417 : #[test]
3418 4 : fn clause_5_5_12_merge_algorithm() {
3419 4 : let mut doc = json!({"id": "urn:x", "type": ["https://uri.etsi.org/ngsi-ld/default-context/T"],
3420 4 : "https://uri.etsi.org/ngsi-ld/default-context/temperature": [{
3421 4 : "type": "Property", "value": 25, "unitCode": "CEL",
3422 4 : "observedAt": "2022-03-14T01:59:26.535Z"}]});
3423 4 : merge_into(
3424 4 : &mut doc,
3425 4 : &json!({
3426 4 : "type": ["https://uri.etsi.org/ngsi-ld/default-context/T",
3427 4 : "https://uri.etsi.org/ngsi-ld/default-context/U"],
3428 4 : "https://uri.etsi.org/ngsi-ld/default-context/temperature": [
3429 4 : {"type": "Property", "value": 100,
3430 4 : "observedAt": "2022-03-14T13:00:00.000Z"},
3431 4 : {"type": "Property", "value": 7,
3432 4 : "datasetId": "urn:ngsi-ld:Dataset:extra"}
3433 4 : ]}),
3434 4 : "2026-08-11T00:00:00Z",
3435 : );
3436 : // EXAMPLE 1: value/observedAt updated, unitCode untouched
3437 4 : let t = &doc["https://uri.etsi.org/ngsi-ld/default-context/temperature"];
3438 4 : let default = t
3439 4 : .as_array()
3440 4 : .unwrap()
3441 4 : .iter()
3442 4 : .find(|i| i.get("datasetId").is_none())
3443 4 : .expect("default instance");
3444 4 : assert_eq!(default["value"], 100);
3445 4 : assert_eq!(default["observedAt"], "2022-03-14T13:00:00.000Z");
3446 4 : assert_eq!(
3447 4 : default["unitCode"], "CEL",
3448 : "unmentioned sub-attribute survives"
3449 : );
3450 : // unknown datasetId is added as a NEW instance
3451 4 : assert_eq!(t.as_array().unwrap().len(), 2);
3452 : // entity types are unioned, no duplicates
3453 4 : assert_eq!(
3454 4 : doc["type"],
3455 4 : json!([
3456 : "https://uri.etsi.org/ngsi-ld/default-context/T",
3457 : "https://uri.etsi.org/ngsi-ld/default-context/U"
3458 : ])
3459 : );
3460 4 : }
3461 :
3462 : /// 5.5.12: merge "merges the provided information with the existing
3463 : /// information up to an arbitrary depth, e.g. including going into JSON
3464 : /// objects representing a Property value" (RFC 7396 with the NGSI-LD
3465 : /// Null) — untouched keys survive, null-valued keys are removed, and the
3466 : /// null sentinel never lands in the stored document (5.5.4).
3467 : #[test]
3468 4 : fn merge_goes_into_compound_property_values() {
3469 4 : let mut doc = json!({"id": "urn:x", "type": ["T"],
3470 4 : "https://uri.etsi.org/ngsi-ld/default-context/address": [{
3471 4 : "type": "Property",
3472 4 : "value": {"street": "Straße des 17. Juni", "city": "Berlin",
3473 4 : "country": "Germany"}}]});
3474 4 : merge_into(
3475 4 : &mut doc,
3476 4 : &json!({"https://uri.etsi.org/ngsi-ld/default-context/address": [{
3477 4 : "type": "Property",
3478 4 : "value": {"street": "Pariser Platz",
3479 4 : "country": "urn:ngsi-ld:null"}}]}),
3480 4 : "2026-08-11T00:00:00Z",
3481 : );
3482 4 : let v = &doc["https://uri.etsi.org/ngsi-ld/default-context/address"][0]["value"];
3483 4 : assert_eq!(v["street"], "Pariser Platz");
3484 4 : assert_eq!(v["city"], "Berlin", "untouched keys survive the merge");
3485 4 : assert!(v.get("country").is_none(), "null removes the key");
3486 4 : assert!(
3487 4 : !doc.to_string().contains("urn:ngsi-ld:null"),
3488 : "the null sentinel must never be stored"
3489 : );
3490 4 : }
3491 :
3492 : /// An Entity kept `value` beside a `languageMap` after a
3493 : /// writer changed `name` from one attribute type to the other.
3494 : #[test]
3495 4 : fn merge_replaces_an_instance_whose_attribute_type_changes() {
3496 4 : let name = "https://schema.org/name";
3497 4 : let mut doc = json!({"id": "urn:x", "type": ["T"],
3498 4 : name: [{"type": "Property", "value": "M1020",
3499 4 : "createdAt": "2026-10-01T00:00:00Z", "modifiedAt": "2026-10-01T00:00:00Z"}]});
3500 4 : merge_into(
3501 4 : &mut doc,
3502 4 : &json!({name: [{"type": "LanguageProperty", "languageMap": {"sk": "Most"}}]}),
3503 4 : "2026-10-22T00:02:58Z",
3504 : );
3505 4 : assert_eq!(
3506 4 : doc[name],
3507 4 : json!([{"type": "LanguageProperty", "languageMap": {"sk": "Most"},
3508 4 : "createdAt": "2026-10-01T00:00:00Z", "modifiedAt": "2026-10-22T00:02:58Z"}])
3509 : );
3510 4 : merge_into(
3511 4 : &mut doc,
3512 4 : &json!({name: [{"type": "Property", "value": "M1020"}]}),
3513 4 : "2026-10-22T01:00:00Z",
3514 : );
3515 4 : assert_eq!(
3516 4 : doc[name],
3517 4 : json!([{"type": "Property", "value": "M1020",
3518 4 : "createdAt": "2026-10-01T00:00:00Z", "modifiedAt": "2026-10-22T01:00:00Z"}])
3519 : );
3520 4 : }
3521 :
3522 : /// The same attribute type still merges key by key (5.5.12).
3523 : #[test]
3524 4 : fn merge_keeps_the_keys_of_an_instance_of_the_same_type() {
3525 4 : let name = "https://schema.org/name";
3526 4 : let mut doc = json!({"id": "urn:x", "type": ["T"],
3527 4 : name: [{"type": "LanguageProperty", "languageMap": {"sk": "Most", "en": "Bridge"},
3528 4 : "observedAt": "2026-10-01T00:00:00Z"}]});
3529 4 : merge_into(
3530 4 : &mut doc,
3531 4 : &json!({name: [{"type": "LanguageProperty", "languageMap": {"sk": "Most M1020"}}]}),
3532 4 : "2026-10-22T00:00:00Z",
3533 : );
3534 4 : assert_eq!(
3535 4 : doc[name][0]["languageMap"],
3536 4 : json!({"sk": "Most M1020", "en": "Bridge"})
3537 : );
3538 4 : assert_eq!(doc[name][0]["observedAt"], "2026-10-01T00:00:00Z");
3539 4 : }
3540 :
3541 : #[test]
3542 4 : fn merge_sets_and_null_removes_expires_at() {
3543 4 : let mut doc = json!({"id": "urn:x", "type": ["T"]});
3544 4 : merge_into(
3545 4 : &mut doc,
3546 4 : &json!({"expiresAt": "2030-01-01T00:00:00Z"}),
3547 4 : "2026-08-08T00:00:00Z",
3548 : );
3549 4 : assert_eq!(doc["expiresAt"], "2030-01-01T00:00:00Z");
3550 4 : merge_into(
3551 4 : &mut doc,
3552 4 : &json!({"expiresAt": "urn:ngsi-ld:null"}),
3553 4 : "2026-08-08T00:00:01Z",
3554 : );
3555 4 : assert!(doc.get("expiresAt").is_none(), "NGSI-LD Null removes it");
3556 4 : }
3557 : }
3558 :
3559 : #[cfg(test)]
3560 : mod clause_4_16 {
3561 : use super::*;
3562 : use serde_json::json;
3563 :
3564 : /// 4.16: "Entity Types can be implicitly added by all operations that
3565 : /// update or append attributes. There is no operation to remove Entity
3566 : /// Types from an Entity."
3567 : #[test]
3568 4 : fn merge_unions_types_and_never_removes() {
3569 4 : let mut doc = json!({"id": "urn:x", "type": ["A"],
3570 4 : "https://ex/p": [{"type": "Property", "value": 1}]});
3571 4 : merge_into(
3572 4 : &mut doc,
3573 4 : &json!({"type": ["B"]}),
3574 4 : "2026-08-11T00:00:00.000Z",
3575 : );
3576 4 : assert_eq!(doc["type"], json!(["A", "B"]), "types union");
3577 : // a fragment naming FEWER types must not shrink the set
3578 4 : merge_into(
3579 4 : &mut doc,
3580 4 : &json!({"type": ["A"]}),
3581 4 : "2026-08-11T00:00:00.000Z",
3582 : );
3583 4 : assert_eq!(
3584 4 : doc["type"],
3585 4 : json!(["A", "B"]),
3586 : "no operation removes Entity Types"
3587 : );
3588 : // duplicates are not accumulated
3589 4 : merge_into(
3590 4 : &mut doc,
3591 4 : &json!({"type": ["B"]}),
3592 4 : "2026-08-11T00:00:00.000Z",
3593 : );
3594 4 : assert_eq!(doc["type"], json!(["A", "B"]));
3595 4 : }
3596 : }
3597 :
3598 : #[cfg(test)]
3599 : mod clause_4_17 {
3600 : use antares_jsonld::Loader;
3601 : use antares_ql::type_selection_matches;
3602 :
3603 : /// 4.17: disjunction via `|` or `,`, conjunction via `(a;b)`; short
3604 : /// names expand against the @context.
3605 : #[test]
3606 4 : fn selection_language_semantics() {
3607 4 : let ctx = Loader::new().core();
3608 : const D: &str = "https://uri.etsi.org/ngsi-ld/default-context/";
3609 4 : let home = format!("{D}Home");
3610 4 : let vehicle = format!("{D}Vehicle");
3611 4 : let motorhome = format!("{D}Motorhome");
3612 4 : let both: Vec<&str> = vec![&home, &vehicle];
3613 4 : let only_home: Vec<&str> = vec![&home];
3614 4 : let only_motor: Vec<&str> = vec![&motorhome];
3615 : // EXAMPLE 1: OR, both spellings
3616 4 : assert!(type_selection_matches("Building|Home", &only_home, &ctx));
3617 4 : assert!(type_selection_matches("Building,Home", &only_home, &ctx));
3618 4 : assert!(!type_selection_matches("Building|House", &only_home, &ctx));
3619 : // EXAMPLE 2: conjunction — ALL listed types required
3620 4 : assert!(type_selection_matches("(Home;Vehicle)", &both, &ctx));
3621 4 : assert!(
3622 4 : !type_selection_matches("(Home;Vehicle)", &only_home, &ctx),
3623 : "an entity with only Home must NOT match the conjunction"
3624 : );
3625 : // EXAMPLE 3: (Home;Vehicle)|Motorhome in both alternative spellings
3626 8 : for sel in ["(Home;Vehicle)|Motorhome", "(Home;Vehicle),Motorhome"] {
3627 8 : assert!(type_selection_matches(sel, &both, &ctx), "{sel}");
3628 8 : assert!(type_selection_matches(sel, &only_motor, &ctx), "{sel}");
3629 8 : assert!(!type_selection_matches(sel, &only_home, &ctx), "{sel}");
3630 : }
3631 4 : }
3632 : }
3633 :
3634 : #[cfg(test)]
3635 : mod clause_5_2_2 {
3636 : use super::*;
3637 : use antares_jsonld::{ExpandOpts, Loader};
3638 : use serde_json::json;
3639 :
3640 : /// 5.2.2: createdAt/modifiedAt/deletedAt "shall not be provided by
3641 : /// Context Producers. In the event that they are provided ... NGSI-LD
3642 : /// implementations shall ignore them" — server stamps win, no error.
3643 : #[test]
3644 4 : fn client_provided_system_timestamps_are_ignored() {
3645 4 : let ctx = Loader::new().core();
3646 4 : let doc = json!({"id": "urn:x", "type": "T",
3647 4 : "createdAt": "1999-01-01T00:00:00Z",
3648 4 : "modifiedAt": "1999-01-01T00:00:00Z",
3649 4 : "deletedAt": "1999-01-01T00:00:00Z",
3650 4 : "p": {"type": "Property", "value": 1,
3651 4 : "createdAt": "1999-01-01T00:00:00Z"}});
3652 4 : let mut out = antares_jsonld::expand_entity(
3653 4 : doc.as_object().expect("obj"),
3654 4 : &ctx,
3655 4 : ExpandOpts::default(),
3656 : )
3657 4 : .expect("providing common members is not an error");
3658 4 : stamp_new(&mut out, "2026-08-11T00:00:00.000Z");
3659 4 : assert_eq!(out["createdAt"], "2026-08-11T00:00:00.000Z");
3660 4 : assert_eq!(out["modifiedAt"], "2026-08-11T00:00:00.000Z");
3661 4 : assert!(out.get("deletedAt").is_none(), "deletedAt never creatable");
3662 4 : let inst = &out["https://uri.etsi.org/ngsi-ld/default-context/p"][0];
3663 4 : assert_eq!(
3664 4 : inst["createdAt"], "2026-08-11T00:00:00.000Z",
3665 : "instance-level client timestamp ignored too"
3666 : );
3667 4 : }
3668 :
3669 : /// 5.2.2: common members are only generated "when the Context Consumer
3670 : /// explicitly asks for their inclusion" (sysAttrs, 6.3.11).
3671 : #[test]
3672 4 : fn common_members_appear_only_on_request() {
3673 4 : let doc = json!({"id": "urn:x", "type": ["T"],
3674 4 : "createdAt": "2026-08-11T00:00:00.000Z",
3675 4 : "modifiedAt": "2026-08-11T00:00:00.000Z",
3676 4 : "https://uri.etsi.org/ngsi-ld/default-context/p": [
3677 4 : {"type": "Property", "value": 1,
3678 4 : "createdAt": "2026-08-11T00:00:00.000Z",
3679 4 : "modifiedAt": "2026-08-11T00:00:00.000Z"}]});
3680 4 : let plain = crate::repr::apply(&doc, &crate::repr::Repr::default());
3681 4 : assert!(plain.get("createdAt").is_none());
3682 4 : assert!(plain["https://uri.etsi.org/ngsi-ld/default-context/p"][0]
3683 4 : .get("modifiedAt")
3684 4 : .is_none());
3685 4 : let sys = crate::repr::apply(
3686 4 : &doc,
3687 4 : &crate::repr::Repr {
3688 4 : sys_attrs: true,
3689 4 : ..Default::default()
3690 4 : },
3691 : );
3692 4 : assert!(sys.get("createdAt").is_some());
3693 4 : assert!(sys["https://uri.etsi.org/ngsi-ld/default-context/p"][0]
3694 4 : .get("modifiedAt")
3695 4 : .is_some());
3696 4 : }
3697 : }
3698 :
3699 : /// 4.5.23.1: "When retrieving Linked Entities, it is necessary to limit
3700 : /// retrieval to avoid cascades of an excessive length, duplicates or loops."
3701 : #[cfg(test)]
3702 : mod clause_4_5_23_bounds {
3703 : use super::*;
3704 : use crate::repr::inline_join;
3705 : use axum::body::Body;
3706 : use axum::http::Request;
3707 : use http_body_util::BodyExt;
3708 : use serde_json::json;
3709 : use tower::ServiceExt;
3710 :
3711 8 : fn app() -> axum::Router {
3712 8 : crate::router(AppState::new("antares-test".into()))
3713 8 : }
3714 :
3715 20 : async fn create(app: &axum::Router, body: Value) {
3716 20 : let payload = body.to_string();
3717 20 : let resp = app
3718 20 : .clone()
3719 20 : .oneshot(
3720 20 : Request::post("/ngsi-ld/v1/entities")
3721 20 : .header("Content-Type", "application/json")
3722 20 : .header("Content-Length", payload.len().to_string())
3723 20 : .body(Body::from(payload))
3724 20 : .expect("req"),
3725 20 : )
3726 20 : .await
3727 20 : .expect("resp");
3728 20 : assert_eq!(resp.status(), StatusCode::CREATED, "create failed");
3729 20 : }
3730 :
3731 8 : async fn get(app: &axum::Router, uri: &str) -> (axum::http::response::Parts, Value) {
3732 8 : let resp = app
3733 8 : .clone()
3734 8 : .oneshot(Request::get(uri).body(Body::empty()).expect("req"))
3735 8 : .await
3736 8 : .expect("resp");
3737 8 : let (parts, body) = resp.into_parts();
3738 8 : let bytes = body.collect().await.expect("body").to_bytes();
3739 8 : let json: Value = serde_json::from_slice(&bytes).expect("json");
3740 8 : (parts, json)
3741 8 : }
3742 :
3743 : /// 4.5.23.1/4.5.23.3: the flattened array carries the Linking Entity and
3744 : /// its Linked Entities — a Relationship pointing back at the root is a
3745 : /// loop, so the root shall appear exactly ONCE, not once as the Linking
3746 : /// Entity and again as its own Linked Entity.
3747 : #[tokio::test]
3748 4 : async fn flat_join_never_repeats_the_root_entity() {
3749 4 : let app = app();
3750 4 : let root = "urn:ngsi-ld:Loop:root";
3751 4 : let leaf = "urn:ngsi-ld:Loop:leaf";
3752 4 : create(&app, json!({"id": leaf, "type": "Loop"})).await;
3753 4 : create(
3754 4 : &app,
3755 4 : json!({"id": root, "type": "Loop",
3756 4 : "self": {"type": "Relationship", "object": root},
3757 4 : "other": {"type": "Relationship", "object": leaf}}),
3758 4 : )
3759 4 : .await;
3760 :
3761 4 : let (_, body) = get(
3762 4 : &app,
3763 4 : &format!("/ngsi-ld/v1/entities/{root}?join=flat&joinLevel=3"),
3764 : )
3765 4 : .await;
3766 4 : let arr = match body {
3767 4 : Value::Array(a) => a,
3768 0 : other => vec![other],
3769 : };
3770 4 : let (mut roots, mut leaves) = (0usize, 0usize);
3771 8 : for e in &arr {
3772 8 : match e["id"].as_str() {
3773 8 : Some(id) if id == root => roots += 1,
3774 4 : Some(id) if id == leaf => leaves += 1,
3775 0 : _ => {}
3776 : }
3777 : }
3778 4 : assert_eq!(
3779 : roots, 1,
3780 : "the root must appear exactly once in the flattened array: {arr:?}"
3781 : );
3782 4 : assert_eq!(
3783 : leaves, 1,
3784 : "the genuine Linked Entity is still there once: {arr:?}"
3785 : );
3786 4 : assert_eq!(arr.len(), 2, "no other entity is in the array: {arr:?}");
3787 4 : }
3788 :
3789 : /// 4.5.23.1: a cyclic graph at a high joinLevel shall not cascade — the
3790 : /// walk stops at entities it already resolved and says so with an
3791 : /// NGSILD-Warning (6.3.17) instead of expanding fan-out^joinLevel.
3792 : #[tokio::test]
3793 4 : async fn cyclic_inline_join_stops_instead_of_cascading() {
3794 4 : let app = app();
3795 4 : let ids = [
3796 4 : "urn:ngsi-ld:Cyc:a",
3797 4 : "urn:ngsi-ld:Cyc:b",
3798 4 : "urn:ngsi-ld:Cyc:c",
3799 4 : ];
3800 : // complete graph: every entity links to every entity, itself included
3801 12 : for id in ids {
3802 12 : create(
3803 12 : &app,
3804 12 : json!({"id": id, "type": "Cyc",
3805 12 : "toA": {"type": "Relationship", "object": ids[0]},
3806 12 : "toB": {"type": "Relationship", "object": ids[1]},
3807 12 : "toC": {"type": "Relationship", "object": ids[2]}}),
3808 12 : )
3809 12 : .await;
3810 : }
3811 :
3812 4 : let (parts, body) = get(
3813 4 : &app,
3814 4 : &format!("/ngsi-ld/v1/entities/{}?join=inline&joinLevel=9", ids[0]),
3815 : )
3816 4 : .await;
3817 4 : assert_eq!(parts.status, StatusCode::OK);
3818 : // 3^9 embeddings if the walk is unbounded; a handful if it is not
3819 4 : let embedded = body.to_string().matches("urn:ngsi-ld:Cyc:").count();
3820 4 : assert!(
3821 4 : embedded < 64,
3822 : "the cyclic walk cascaded: {embedded} entity references embedded"
3823 : );
3824 4 : assert!(
3825 4 : parts.headers.get("NGSILD-Warning").is_some(),
3826 4 : "a truncated Linked Entity Retrieval must be reported (6.3.17)"
3827 4 : );
3828 4 : }
3829 :
3830 : /// 4.5.23.1: joinLevel bounds the depth, not the width — one retrieval
3831 : /// may only buy MAX_JOIN_LOOKUPS Linked Entity reads, and the truncation
3832 : /// is reported back to the caller.
3833 : #[tokio::test]
3834 4 : async fn wide_inline_join_stops_at_the_lookup_budget() {
3835 4 : let st = AppState::new("antares-test".into());
3836 4 : let tenant = TenantId::default();
3837 4 : let ctx = antares_jsonld::Loader::new().core();
3838 4 : let mut targets: Vec<Value> = Vec::new();
3839 4400 : for n in 0..MAX_JOIN_LOOKUPS + 100 {
3840 4400 : let id = format!("urn:ngsi-ld:Wide:{n}");
3841 4400 : let doc = json!({"id": &id, "type":
3842 : ["https://uri.etsi.org/ngsi-ld/default-context/Wide"]});
3843 4400 : st.store
3844 4400 : .upsert(&tenant, Kind::Entity, &id, doc)
3845 4400 : .await
3846 4400 : .expect("seed");
3847 4400 : targets.push(Value::String(id));
3848 : }
3849 4 : let mut compacted = json!({"id": "urn:ngsi-ld:Wide:root", "type": "Wide",
3850 4 : "links": {"type": "Relationship", "object": Value::Array(targets)}});
3851 4 : let complete = inline_join(
3852 4 : &st,
3853 4 : &tenant,
3854 4 : &ctx,
3855 4 : &crate::repr::Repr::default(),
3856 4 : &mut compacted,
3857 4 : &Value::Null,
3858 4 : 1,
3859 4 : )
3860 4 : .await;
3861 4 : assert!(!complete, "the budget was hit, so the walk is incomplete");
3862 4 : assert_eq!(
3863 4 : compacted["links"]["entity"]
3864 4 : .as_array()
3865 4 : .expect("entity array")
3866 4 : .len(),
3867 4 : MAX_JOIN_LOOKUPS,
3868 4 : "no more Linked Entities than the budget are resolved"
3869 4 : );
3870 4 : }
3871 : }
3872 :
3873 : /// 5.6.6.4 / 5.6.17.4 / 5.6.18.4 / 5.6.21.4: "matching input data is
3874 : /// forwarded to the Registration endpoint" — the forward may narrow what the
3875 : /// peer does, never widen it.
3876 : #[cfg(test)]
3877 : mod forwarded_input_data {
3878 : use super::*;
3879 :
3880 44 : fn params(pairs: &[(&str, &str)]) -> HashMap<String, String> {
3881 44 : pairs
3882 44 : .iter()
3883 128 : .map(|(k, v)| ((*k).to_owned(), (*v).to_owned()))
3884 44 : .collect()
3885 44 : }
3886 :
3887 : /// 5.6.21.3 lists the type selector, the id list, the id pattern, the
3888 : /// restrictive and exclusionary Attribute-name lists, the query, the
3889 : /// geoquery, the Scope query and the context source filter as Purge
3890 : /// input data. Every one of them restricts the purge, so every one of
3891 : /// them travels.
3892 : #[test]
3893 4 : fn purge_forward_carries_every_restriction_the_client_issued() {
3894 4 : let p = params(&[
3895 4 : ("type", "Vehicle"),
3896 4 : ("id", "urn:ngsi-ld:Vehicle:A1"),
3897 4 : ("idPattern", "^urn:ngsi-ld:Vehicle:"),
3898 4 : ("attrs", "speed"),
3899 4 : ("q", "speed>5"),
3900 4 : ("georel", "near;maxDistance==100"),
3901 4 : ("geometry", "Point"),
3902 4 : ("coordinates", "[0,0]"),
3903 4 : ("geoproperty", "location"),
3904 4 : ("scopeQ", "/x"),
3905 4 : ("csf", "name==p"),
3906 4 : ("keep", "name"),
3907 4 : ("local", "false"),
3908 4 : ]);
3909 4 : let q = forwarded_purge_query(&p);
3910 48 : for k in [
3911 4 : "type",
3912 4 : "id",
3913 4 : "idPattern",
3914 4 : "attrs",
3915 4 : "q",
3916 4 : "georel",
3917 4 : "geometry",
3918 4 : "coordinates",
3919 4 : "geoproperty",
3920 4 : "scopeQ",
3921 4 : "csf",
3922 4 : "keep",
3923 4 : ] {
3924 48 : assert!(
3925 312 : q.iter().any(|(a, _)| a == k),
3926 : "{k} was dropped, so the peer executes a wider purge than the \
3927 : client issued: {q:?}"
3928 : );
3929 : }
3930 4 : assert!(
3931 48 : !q.iter().any(|(k, _)| k == "local"),
3932 : "local scope is the reason a forward happens at all — it is not \
3933 : itself forwarded (5.5.13): {q:?}"
3934 : );
3935 4 : assert_eq!(q.len(), 12, "nothing else is invented: {q:?}");
3936 4 : }
3937 :
3938 : /// The exclusionary list travels on its own too — `drop=` alone must not
3939 : /// reach the peer as a bare, entity-deleting purge.
3940 : #[test]
3941 4 : fn purge_forward_carries_drop_and_omits_absent_members() {
3942 4 : let q = forwarded_purge_query(¶ms(&[("type", "Vehicle"), ("drop", "speed")]));
3943 4 : assert!(
3944 4 : q.contains(&("drop".to_owned(), "speed".to_owned())),
3945 : "{q:?}"
3946 : );
3947 4 : assert_eq!(q.len(), 2, "absent parameters are not forwarded: {q:?}");
3948 4 : }
3949 :
3950 : /// 5.6.6.3 / 5.6.17.3 / 5.6.18.3: the selector of Entity types is input
3951 : /// data of Delete, Merge and Replace Entity. A registration may cover
3952 : /// several types, so the peer needs the selector to reach the same
3953 : /// verdict this broker reached locally.
3954 : #[test]
3955 4 : fn write_forwards_carry_the_type_selector() {
3956 4 : assert_eq!(
3957 4 : type_selector_query(¶ms(&[("type", "Vehicle"), ("local", "false")])),
3958 4 : vec![("type".to_owned(), "Vehicle".to_owned())]
3959 : );
3960 4 : assert!(
3961 4 : type_selector_query(¶ms(&[("local", "false")])).is_empty(),
3962 : "no selector, nothing to forward"
3963 : );
3964 4 : }
3965 :
3966 : /// 5.7.2.4 applies q/geoquery/Scope query/Attributes only after remote
3967 : /// parts have been aggregated, and 4.23 orders by the value of a member —
3968 : /// neither survives a store-side projection of that member.
3969 : #[test]
3970 4 : fn projection_is_pushed_down_only_when_the_store_answer_is_final() {
3971 4 : let plain = params(&[("type", "T"), ("pick", "name")]);
3972 4 : assert!(
3973 4 : proj_pushdown_allowed(true, &plain),
3974 : "a purely local query keeps the pushdown"
3975 : );
3976 4 : assert!(
3977 4 : !proj_pushdown_allowed(false, &plain),
3978 : "federated candidates mean the filters run again after the merge"
3979 : );
3980 4 : assert!(
3981 4 : !proj_pushdown_allowed(true, ¶ms(&[("attrs", "name"), ("orderBy", "age")])),
3982 : "the ordering member must survive to be compared"
3983 : );
3984 4 : }
3985 :
3986 : /// 6.3.10 makes `limit=0` legal only with `count=true` — an answer that is
3987 : /// a count and no rows. That shape is pushed to the store; the filters the
3988 : /// store cannot see (idPattern, 4.23 ordering, federated candidates merged
3989 : /// per 5.7.2.4) still forfeit the page.
3990 : #[test]
3991 4 : fn the_count_only_page_is_pushed_down_but_store_blind_filters_are_not() {
3992 4 : assert!(
3993 4 : page_pushdown_allowed(
3994 : true,
3995 4 : ¶ms(&[("type", "T"), ("limit", "0"), ("count", "true")])
3996 : ),
3997 : "a count is answered by counting, not by materializing the match set"
3998 : );
3999 4 : assert!(
4000 4 : page_pushdown_allowed(true, ¶ms(&[("type", "T"), ("limit", "10")])),
4001 : "a plain local query keeps the pushdown"
4002 : );
4003 4 : assert!(
4004 4 : !page_pushdown_allowed(true, ¶ms(&[("type", "T"), ("idPattern", "^urn:")])),
4005 : "idPattern is applied after the store, so it drops rows the SQL \
4006 : page already counted"
4007 : );
4008 4 : assert!(
4009 4 : !page_pushdown_allowed(true, ¶ms(&[("type", "T"), ("orderBy", "speed")])),
4010 : "4.23 orders the whole match set before the page is cut"
4011 : );
4012 4 : assert!(
4013 4 : !page_pushdown_allowed(false, ¶ms(&[("type", "T")])),
4014 : "federated candidates are merged after the store answered"
4015 : );
4016 4 : }
4017 : }
4018 :
4019 : /// 5.6.1 Create Entity and 5.6.21 Purge Entities, end to end over the store.
4020 : #[cfg(test)]
4021 : mod clause_5_6_1_and_5_6_21 {
4022 : use super::*;
4023 : use axum::body::Body;
4024 : use axum::http::Request;
4025 : use http_body_util::BodyExt;
4026 : use serde_json::json;
4027 : use tower::ServiceExt;
4028 :
4029 : /// 5.6.1.5: the output is "the URI of the created Entity", returned in
4030 : /// the Location header. An id is one path segment (RFC 3986 clause 3.3),
4031 : /// so a `#` in it must not be able to end the segment.
4032 : #[tokio::test]
4033 4 : async fn location_header_percent_encodes_the_entity_id() {
4034 4 : let app = crate::router(AppState::new("antares-test".into()));
4035 4 : let id = "urn:ngsi-ld:Vehicle:A#4567";
4036 4 : let payload = json!({"id": id, "type": "Vehicle"}).to_string();
4037 4 : let resp = app
4038 4 : .oneshot(
4039 4 : Request::post("/ngsi-ld/v1/entities")
4040 4 : .header("Content-Type", "application/json")
4041 4 : .header("Content-Length", payload.len().to_string())
4042 4 : .body(Body::from(payload))
4043 4 : .expect("req"),
4044 4 : )
4045 4 : .await
4046 4 : .expect("resp");
4047 4 : assert_eq!(resp.status(), StatusCode::CREATED);
4048 4 : let loc = resp
4049 4 : .headers()
4050 4 : .get("Location")
4051 4 : .expect("Location header")
4052 4 : .to_str()
4053 4 : .expect("ascii");
4054 4 : assert_eq!(loc, "/ngsi-ld/v1/entities/urn:ngsi-ld:Vehicle:A%234567");
4055 4 : assert!(
4056 4 : !loc.contains('#'),
4057 4 : "a raw # truncates the URL at the fragment and addresses another \
4058 4 : resource: {loc}"
4059 4 : );
4060 4 : }
4061 :
4062 : /// 5.7.2 names Entities by a comma-separated `id` list, and
4063 : /// the EntityMap, temporal and federation pages rebuild that list, so an
4064 : /// id holding a comma is refused on every create door and nothing is
4065 : /// stored.
4066 : #[tokio::test]
4067 4 : async fn an_entity_id_holding_a_comma_is_refused_at_create() {
4068 4 : let app = crate::router(AppState::new("antares-test".into()));
4069 4 : let id = "urn:ngsi-ld:Vehicle:a,b";
4070 4 : let one = json!({"id": id, "type": "Vehicle"});
4071 4 : let temporal = json!({"id": id, "type": "Vehicle",
4072 4 : "speed": [{"type": "Property", "value": 1, "observedAt": "2026-10-26T00:00:00Z"}]});
4073 : // a batch answers 207 with the member's 400 in `errors` (5.6.7)
4074 16 : for (path, body, want) in [
4075 4 : ("/ngsi-ld/v1/entities", one.clone(), StatusCode::BAD_REQUEST),
4076 4 : (
4077 4 : "/ngsi-ld/v1/entityOperations/create",
4078 4 : json!([one.clone()]),
4079 4 : StatusCode::MULTI_STATUS,
4080 4 : ),
4081 4 : (
4082 4 : "/ngsi-ld/v1/entityOperations/upsert",
4083 4 : json!([one.clone()]),
4084 4 : StatusCode::MULTI_STATUS,
4085 4 : ),
4086 4 : (
4087 4 : "/ngsi-ld/v1/temporal/entities",
4088 4 : temporal,
4089 4 : StatusCode::BAD_REQUEST,
4090 4 : ),
4091 4 : ] {
4092 16 : let payload = body.to_string();
4093 16 : let resp = app
4094 16 : .clone()
4095 16 : .oneshot(
4096 16 : Request::post(path)
4097 16 : .header("Content-Type", "application/json")
4098 16 : .header("Content-Length", payload.len().to_string())
4099 16 : .body(Body::from(payload))
4100 16 : .expect("req"),
4101 16 : )
4102 16 : .await
4103 16 : .expect("resp");
4104 16 : let status = resp.status();
4105 16 : let bytes = resp.into_body().collect().await.expect("body").to_bytes();
4106 16 : let text = String::from_utf8_lossy(&bytes);
4107 16 : assert_eq!(status, want, "{path}: {text}");
4108 16 : assert!(text.contains("holds a comma"), "{path}: {text}");
4109 : }
4110 4 : let resp = app
4111 4 : .oneshot(
4112 4 : Request::get("/ngsi-ld/v1/entities/urn:ngsi-ld:Vehicle:a,b")
4113 4 : .body(Body::empty())
4114 4 : .expect("req"),
4115 4 : )
4116 4 : .await
4117 4 : .expect("resp");
4118 4 : assert_eq!(resp.status(), StatusCode::NOT_FOUND);
4119 4 : }
4120 :
4121 : /// an upsert refuses a comma only for an id it would create;
4122 : /// an Entity stored under one before the rule is still upserted, so a
4123 : /// source that keeps writing it does not turn red.
4124 : #[tokio::test]
4125 4 : async fn an_upsert_still_reaches_a_stored_comma_id() {
4126 4 : let st = AppState::new("antares-test".into());
4127 4 : let tenant = TenantId::default();
4128 4 : let id = "urn:ngsi-ld:Vehicle:a,b";
4129 4 : let vehicle = "https://uri.etsi.org/ngsi-ld/default-context/Vehicle";
4130 4 : let doc = json!({"id": id, "type": [vehicle]});
4131 4 : st.store
4132 4 : .upsert(&tenant, Kind::Entity, id, doc.clone())
4133 4 : .await
4134 4 : .expect("seed entity");
4135 4 : assert!(st
4136 4 : .temporal
4137 4 : .create(&tenant, id, doc)
4138 4 : .await
4139 4 : .expect("seed temporal"));
4140 4 : let app = crate::router(st);
4141 4 : let one = json!({"id": id, "type": "Vehicle",
4142 4 : "speed": {"type": "Property", "value": 2}});
4143 4 : let temporal = json!({"id": id, "type": "Vehicle",
4144 4 : "speed": [{"type": "Property", "value": 2, "observedAt": "2026-10-26T00:00:00Z"}]});
4145 8 : for (path, body) in [
4146 4 : ("/ngsi-ld/v1/entityOperations/upsert", json!([one])),
4147 4 : ("/ngsi-ld/v1/temporal/entities", temporal),
4148 4 : ] {
4149 8 : let payload = body.to_string();
4150 8 : let resp = app
4151 8 : .clone()
4152 8 : .oneshot(
4153 8 : Request::post(path)
4154 8 : .header("Content-Type", "application/json")
4155 8 : .header("Content-Length", payload.len().to_string())
4156 8 : .body(Body::from(payload))
4157 8 : .expect("req"),
4158 8 : )
4159 8 : .await
4160 8 : .expect("resp");
4161 8 : let status = resp.status();
4162 8 : let bytes = resp.into_body().collect().await.expect("body").to_bytes();
4163 8 : assert_eq!(
4164 4 : status,
4165 4 : StatusCode::NO_CONTENT,
4166 4 : "{path}: {}",
4167 4 : String::from_utf8_lossy(&bytes)
4168 4 : );
4169 4 : }
4170 4 : }
4171 :
4172 18 : async fn seed(st: &AppState, tenant: &TenantId, n: usize) {
4173 12580 : for i in 0..n {
4174 12580 : let id = format!("urn:ngsi-ld:Purge:{i:05}");
4175 12580 : let doc = json!({"id": &id,
4176 12580 : "type": ["https://uri.etsi.org/ngsi-ld/default-context/Purge"],
4177 12580 : "https://uri.etsi.org/ngsi-ld/default-context/name":
4178 12580 : [{"type": "Property", "value": "n"}],
4179 12580 : "https://uri.etsi.org/ngsi-ld/default-context/speed":
4180 12580 : [{"type": "Property", "value": 1}]});
4181 12580 : st.store
4182 12580 : .upsert(tenant, Kind::Entity, &id, doc)
4183 12580 : .await
4184 12580 : .expect("seed");
4185 : }
4186 18 : }
4187 :
4188 22 : fn purge_params(pairs: &[(&str, &str)]) -> HashMap<String, String> {
4189 22 : pairs
4190 22 : .iter()
4191 40 : .map(|(k, v)| ((*k).to_owned(), (*v).to_owned()))
4192 22 : .collect()
4193 22 : }
4194 :
4195 : /// 5.6.21.4: the implementation "shall delete all Entities that can be
4196 : /// found locally using retrieved list of Entity ids" — all of them, not
4197 : /// one page of them, however many rounds that takes.
4198 : #[tokio::test]
4199 4 : async fn purge_deletes_every_match_across_page_boundaries() {
4200 4 : let st = AppState::new("antares-test".into());
4201 4 : let tenant = TenantId::default();
4202 4 : let n = PURGE_CHUNK * 2 + 7;
4203 4 : seed(&st, &tenant, n).await;
4204 4 : let resp = purge_inner(&st, &purge_params(&[("type", "Purge")]), &HeaderMap::new())
4205 4 : .await
4206 4 : .expect("purge");
4207 4 : assert_eq!(resp.status(), StatusCode::NO_CONTENT);
4208 4 : assert!(
4209 4 : st.store
4210 4 : .list(&tenant, Kind::Entity)
4211 4 : .await
4212 4 : .expect("list")
4213 4 : .is_empty(),
4214 4 : "entities survived the purge"
4215 4 : );
4216 4 : }
4217 :
4218 : /// 5.6.21.4: with an exclusionary list the implementation "shall delete
4219 : /// all but the given set of Attributes" — the Entities themselves
4220 : /// survive, again for the whole match set and not just its first page.
4221 : #[tokio::test]
4222 4 : async fn purge_with_keep_prunes_every_match_and_deletes_no_entity() {
4223 4 : let st = AppState::new("antares-test".into());
4224 4 : let tenant = TenantId::default();
4225 4 : let n = PURGE_CHUNK * 2 + 7;
4226 4 : seed(&st, &tenant, n).await;
4227 4 : let resp = purge_inner(
4228 4 : &st,
4229 4 : &purge_params(&[("type", "Purge"), ("keep", "name")]),
4230 4 : &HeaderMap::new(),
4231 4 : )
4232 4 : .await
4233 4 : .expect("purge");
4234 4 : assert_eq!(resp.status(), StatusCode::NO_CONTENT);
4235 4 : let left = st.store.list(&tenant, Kind::Entity).await.expect("list");
4236 4 : assert_eq!(left.len(), n, "keep= prunes attributes, not entities");
4237 4028 : for doc in &left {
4238 4028 : assert!(
4239 4028 : doc.get("https://uri.etsi.org/ngsi-ld/default-context/name")
4240 4028 : .is_some(),
4241 4 : "the kept attribute is still there: {doc}"
4242 4 : );
4243 4028 : assert!(
4244 4028 : doc.get("https://uri.etsi.org/ngsi-ld/default-context/speed")
4245 4028 : .is_none(),
4246 4 : "every other attribute is gone: {doc}"
4247 4 : );
4248 4 : }
4249 4 : }
4250 :
4251 : /// 5.6.21.4: "id matches the id pattern passed as a parameter" — the
4252 : /// pattern narrows the match set, and only that set is deleted.
4253 : #[tokio::test]
4254 4 : async fn purge_deletes_the_id_pattern_matches_and_nothing_else() {
4255 4 : let st = AppState::new("antares-test".into());
4256 4 : let tenant = TenantId::default();
4257 4 : seed(&st, &tenant, 30).await;
4258 4 : let resp = purge_inner(
4259 4 : &st,
4260 4 : &purge_params(&[("type", "Purge"), ("idPattern", "^urn:ngsi-ld:Purge:0000")]),
4261 4 : &HeaderMap::new(),
4262 4 : )
4263 4 : .await
4264 4 : .expect("purge");
4265 4 : assert_eq!(resp.status(), StatusCode::NO_CONTENT);
4266 4 : let left: Vec<String> = st
4267 4 : .store
4268 4 : .list(&tenant, Kind::Entity)
4269 4 : .await
4270 4 : .expect("list")
4271 4 : .iter()
4272 80 : .filter_map(|d| d["id"].as_str().map(str::to_owned))
4273 4 : .collect();
4274 4 : assert_eq!(
4275 4 : left.len(),
4276 : 20,
4277 : "only the ten pattern matches went: {left:?}"
4278 : );
4279 4 : assert!(
4280 4 : !left
4281 4 : .iter()
4282 80 : .any(|id| id.starts_with("urn:ngsi-ld:Purge:0000")),
4283 : "a pattern match survived: {left:?}"
4284 : );
4285 4 : assert!(
4286 4 : left.contains(&"urn:ngsi-ld:Purge:00010".to_owned()),
4287 4 : "an Entity the pattern does not match must not be purged: {left:?}"
4288 4 : );
4289 4 : }
4290 :
4291 : /// 5.6.21.4 deletes "all Entities that can be found locally using
4292 : /// retrieved list of Entity ids". The retrieval is chunked, and idPattern
4293 : /// is applied after the store — so a chunk can come back narrowed, or
4294 : /// empty, while matches still wait behind it. The walk continues on the
4295 : /// store's row count, never on the surviving subset.
4296 : #[test]
4297 4 : fn purge_walks_on_the_rows_the_store_returned_not_the_narrowed_subset() {
4298 4 : assert_eq!(
4299 4 : next_scan_offset(0, 500, 0, true, 500),
4300 : Some(500),
4301 : "a full chunk the pattern narrowed to nothing still has a successor"
4302 : );
4303 4 : assert_eq!(
4304 4 : next_scan_offset(500, 500, 120, true, 500),
4305 : Some(880),
4306 : "the 380 rows the round did not remove are stepped over"
4307 : );
4308 4 : assert_ne!(
4309 4 : next_scan_offset(0, 500, 0, true, 500),
4310 : None,
4311 : "stopping here leaves every match behind the first chunk alive"
4312 : );
4313 4 : }
4314 :
4315 : /// 5.6.21.4 against a store that really pages (memory answers every query
4316 : /// with the whole match set, so it cannot exercise the walk): with an
4317 : /// idPattern spread across the match set, every chunk arrives narrowed —
4318 : /// and the purge still has to delete "all Entities that can be found
4319 : /// locally using retrieved list of Entity ids", not just the first chunk's
4320 : /// share. Skips without ANTARES_TEST_DATABASE_URL.
4321 : #[cfg(feature = "postgres")]
4322 : #[tokio::test(flavor = "multi_thread")]
4323 4 : async fn purge_over_a_paging_store_deletes_every_id_pattern_match() {
4324 4 : let url = match std::env::var("ANTARES_TEST_DATABASE_URL") {
4325 4 : Ok(u) => u,
4326 : Err(_) => {
4327 0 : eprintln!("SKIP: ANTARES_TEST_DATABASE_URL not set");
4328 0 : return;
4329 : }
4330 : };
4331 4 : let pool = antares_sql::store::pg::connect(&url, 5)
4332 4 : .await
4333 4 : .expect("connect");
4334 4 : let tenant = TenantId::new("purgepaging").expect("tenant");
4335 4 : antares_sql::store::pg::ensure_tenant(&pool, &tenant)
4336 4 : .await
4337 2 : .expect("tenant row");
4338 2 : let st = AppState::with_store(
4339 2 : "antares-test".into(),
4340 2 : std::sync::Arc::new(antares_sql::store::any::AnyStore::Pg(
4341 2 : antares_sql::store::any::PgBackend::new(pool),
4342 2 : )),
4343 2 : "postgres",
4344 : );
4345 2 : for doc in st.store.list(&tenant, Kind::Entity).await.expect("list") {
4346 0 : if let Some(id) = doc["id"].as_str() {
4347 0 : st.store
4348 0 : .delete(&tenant, Kind::Entity, id)
4349 0 : .await
4350 0 : .expect("clean");
4351 0 : }
4352 : }
4353 : // more than two chunks, and the pattern matches every second id — so
4354 : // no chunk the store pages is full once the pattern filtered it
4355 2 : let n = PURGE_CHUNK * 2 + 200;
4356 2 : seed(&st, &tenant, n).await;
4357 : // the purge must run AS the seeded tenant — with no NGSILD-Tenant
4358 : // header it correctly purges the default tenant's (empty) match set
4359 : // and every seeded row survives (5.5.10)
4360 2 : let mut headers = HeaderMap::new();
4361 2 : headers.insert("NGSILD-Tenant", "purgepaging".parse().expect("header"));
4362 2 : let resp = purge_inner(
4363 2 : &st,
4364 2 : &purge_params(&[("type", "Purge"), ("idPattern", "[02468]$")]),
4365 2 : &headers,
4366 2 : )
4367 2 : .await
4368 2 : .expect("purge");
4369 2 : assert_eq!(resp.status(), StatusCode::NO_CONTENT);
4370 2 : let left: Vec<String> = st
4371 2 : .store
4372 2 : .list(&tenant, Kind::Entity)
4373 2 : .await
4374 2 : .expect("list")
4375 2 : .iter()
4376 1200 : .filter_map(|d| d["id"].as_str().map(str::to_owned))
4377 2 : .collect();
4378 2 : assert!(
4379 2 : !left
4380 2 : .iter()
4381 1200 : .any(|id| id.ends_with(['0', '2', '4', '6', '8'])),
4382 : "{} of {} pattern matches survived the chunked walk",
4383 0 : left.iter()
4384 0 : .filter(|id| id.ends_with(['0', '2', '4', '6', '8']))
4385 0 : .count(),
4386 0 : n / 2
4387 : );
4388 2 : assert_eq!(
4389 2 : left.len(),
4390 2 : n / 2,
4391 : "an Entity the pattern does not match must not be purged"
4392 : );
4393 1202 : for id in &left {
4394 1202 : st.store
4395 1200 : .delete(&tenant, Kind::Entity, id)
4396 1200 : .await
4397 1202 : .expect("clean");
4398 4 : }
4399 4 : }
4400 :
4401 : /// 5.6.21.4 under a scopeQ, which the store narrows but cannot decide:
4402 : /// the window it is walked in counts matches, so the walk steps in
4403 : /// matches too. Stepping by the store rows read skipped every match past
4404 : /// the first window and, with a full window of non-matching rows behind
4405 : /// it, never stopped.
4406 : #[tokio::test]
4407 4 : async fn purge_with_scope_q_deletes_every_match_past_the_first_window() {
4408 4 : let st = AppState::new("antares-test".into());
4409 4 : let tenant = TenantId::default();
4410 3600 : for i in 0..900usize {
4411 3600 : let id = format!("urn:ngsi-ld:Purge:{i:05}");
4412 3600 : let mut doc = json!({"id": &id,
4413 3600 : "type": ["https://uri.etsi.org/ngsi-ld/default-context/Purge"]});
4414 3600 : if i < 700 {
4415 2800 : doc["scope"] = json!(["/a"]);
4416 2800 : }
4417 3600 : st.store
4418 3600 : .upsert(&tenant, Kind::Entity, &id, doc)
4419 3600 : .await
4420 3600 : .expect("seed");
4421 : }
4422 4 : let resp = purge_inner(
4423 4 : &st,
4424 4 : &purge_params(&[("type", "Purge"), ("scopeQ", "/a")]),
4425 4 : &HeaderMap::new(),
4426 4 : )
4427 4 : .await
4428 4 : .expect("purge");
4429 4 : assert_eq!(resp.status(), StatusCode::NO_CONTENT);
4430 4 : let left = st.store.list(&tenant, Kind::Entity).await.expect("list");
4431 800 : let scoped = left.iter().filter(|d| d.get("scope").is_some()).count();
4432 4 : assert_eq!(scoped, 0, "{scoped} scoped matches survived the purge");
4433 4 : assert_eq!(left.len(), 200, "an Entity outside the scope was purged");
4434 4 : }
4435 :
4436 : /// The same walk ends when nothing matches: a full window of rows the
4437 : /// scopeQ refuses is no match to step over. Run on a thread of its own,
4438 : /// so a walk that never yields fails the test instead of hanging it.
4439 : #[test]
4440 4 : fn purge_with_scope_q_matching_nothing_terminates() {
4441 4 : let (tx, rx) = std::sync::mpsc::channel();
4442 4 : std::thread::spawn(move || {
4443 4 : let rt = tokio::runtime::Builder::new_current_thread()
4444 4 : .enable_all()
4445 4 : .build()
4446 4 : .expect("runtime");
4447 4 : let left = rt.block_on(async {
4448 4 : let st = AppState::new("antares-test".into());
4449 4 : let tenant = TenantId::default();
4450 4 : seed(&st, &tenant, PURGE_CHUNK + 1).await;
4451 4 : let params = purge_params(&[("type", "Purge"), ("scopeQ", "/none")]);
4452 4 : let resp = purge_inner(&st, ¶ms, &HeaderMap::new())
4453 4 : .await
4454 4 : .expect("purge");
4455 4 : assert_eq!(resp.status(), StatusCode::NO_CONTENT);
4456 4 : st.store
4457 4 : .list(&tenant, Kind::Entity)
4458 4 : .await
4459 4 : .expect("list")
4460 4 : .len()
4461 4 : });
4462 4 : let _ = tx.send(left);
4463 4 : });
4464 4 : let left = rx
4465 4 : .recv_timeout(std::time::Duration::from_secs(60))
4466 4 : .expect("the purge walk ended");
4467 4 : assert_eq!(left, PURGE_CHUNK + 1, "nothing matched, nothing is purged");
4468 4 : }
4469 :
4470 : /// The same walk terminates: a chunk the store did not page IS the whole
4471 : /// match set, a short chunk is the last one, and a round that deleted its
4472 : /// whole chunk re-reads the window the deletions shifted down.
4473 : #[test]
4474 4 : fn purge_stops_when_the_match_set_is_exhausted() {
4475 4 : assert_eq!(
4476 4 : next_scan_offset(0, 500, 500, false, 500),
4477 : None,
4478 : "an unpaged answer is the whole match set"
4479 : );
4480 4 : assert_eq!(
4481 4 : next_scan_offset(0, 7, 7, true, 500),
4482 : None,
4483 : "a chunk shorter than the page size is the last one"
4484 : );
4485 4 : assert_eq!(
4486 4 : next_scan_offset(0, 500, 500, true, 500),
4487 : Some(0),
4488 : "everything deleted — the rest of the match set shifted to the front"
4489 : );
4490 4 : }
4491 : }
4492 :
4493 : /// 5.6.17 Merge Entity and the Linked Entity Retrieval parameters of
4494 : /// Table 6.4.3.2-1, over the HTTP surface.
4495 : #[cfg(test)]
4496 : mod clause_5_6_17_and_6_4_3_2 {
4497 : use super::*;
4498 : use axum::body::Body;
4499 : use axum::http::Request;
4500 : use http_body_util::BodyExt;
4501 : use serde_json::json;
4502 : use tower::ServiceExt;
4503 :
4504 16 : fn app() -> axum::Router {
4505 16 : crate::router(AppState::new("antares-test".into()))
4506 16 : }
4507 :
4508 24 : async fn create(app: &axum::Router, body: Value) {
4509 24 : let payload = body.to_string();
4510 24 : let resp = app
4511 24 : .clone()
4512 24 : .oneshot(
4513 24 : Request::post("/ngsi-ld/v1/entities")
4514 24 : .header("Content-Type", "application/json")
4515 24 : .header("Content-Length", payload.len().to_string())
4516 24 : .body(Body::from(payload))
4517 24 : .expect("req"),
4518 24 : )
4519 24 : .await
4520 24 : .expect("resp");
4521 24 : assert_eq!(resp.status(), StatusCode::CREATED, "create failed");
4522 24 : }
4523 :
4524 12 : async fn patch(app: &axum::Router, uri: &str, body: Value) -> StatusCode {
4525 12 : let payload = body.to_string();
4526 12 : app.clone()
4527 12 : .oneshot(
4528 12 : Request::patch(uri)
4529 12 : .header("Content-Type", "application/json")
4530 12 : .header("Content-Length", payload.len().to_string())
4531 12 : .body(Body::from(payload))
4532 12 : .expect("req"),
4533 12 : )
4534 12 : .await
4535 12 : .expect("resp")
4536 12 : .status()
4537 12 : }
4538 :
4539 12 : async fn get(app: &axum::Router, uri: &str) -> Value {
4540 12 : let resp = app
4541 12 : .clone()
4542 12 : .oneshot(Request::get(uri).body(Body::empty()).expect("req"))
4543 12 : .await
4544 12 : .expect("resp");
4545 12 : let bytes = resp.into_body().collect().await.expect("body").to_bytes();
4546 12 : serde_json::from_slice(&bytes).expect("json")
4547 12 : }
4548 :
4549 : /// 5.6.17.3: "An optional parameter indicating a common observedAt
4550 : /// timestamp to use across merged Attributes." It is the timestamp of
4551 : /// every Attribute this merge touches; an Attribute the Fragment gives an
4552 : /// observedAt of its own keeps that one, and an Attribute the Fragment
4553 : /// does not mention is not a merged Attribute and is left alone.
4554 : #[tokio::test]
4555 4 : async fn merge_applies_the_common_observed_at() {
4556 4 : let app = app();
4557 4 : let id = "urn:ngsi-ld:Obs:1";
4558 4 : create(
4559 4 : &app,
4560 4 : json!({"id": id, "type": "T",
4561 4 : "a": {"type": "Property", "value": 1},
4562 4 : "b": {"type": "Property", "value": 1},
4563 4 : "untouched": {"type": "Property", "value": 1,
4564 4 : "observedAt": "2020-01-01T00:00:00Z"}}),
4565 4 : )
4566 4 : .await;
4567 4 : let uri = format!("/ngsi-ld/v1/entities/{id}?observedAt=2026-08-17T10:00:00Z");
4568 4 : assert_eq!(
4569 4 : patch(
4570 4 : &app,
4571 4 : &uri,
4572 4 : json!({"a": {"value": 2},
4573 4 : "b": {"value": 2, "observedAt": "2021-01-01T00:00:00Z"}}),
4574 4 : )
4575 4 : .await,
4576 : StatusCode::NO_CONTENT
4577 : );
4578 4 : let body = get(&app, &format!("/ngsi-ld/v1/entities/{id}")).await;
4579 4 : assert_eq!(
4580 4 : body["a"]["observedAt"], "2026-08-17T10:00:00Z",
4581 : "the merged Attribute takes the common timestamp: {body}"
4582 : );
4583 4 : assert_eq!(
4584 4 : body["b"]["observedAt"], "2021-01-01T00:00:00Z",
4585 : "a Fragment instance carrying its own observedAt keeps it: {body}"
4586 : );
4587 4 : assert_eq!(
4588 4 : body["untouched"]["observedAt"], "2020-01-01T00:00:00Z",
4589 4 : "an Attribute this merge does not touch is not restamped: {body}"
4590 4 : );
4591 4 : }
4592 :
4593 : /// 4.8 makes observedAt a DateTime, so a value that is not one cannot be
4594 : /// stamped across the merged Attributes.
4595 : #[tokio::test]
4596 4 : async fn merge_rejects_an_observed_at_that_is_not_a_datetime() {
4597 4 : let app = app();
4598 4 : let id = "urn:ngsi-ld:Obs:2";
4599 4 : create(&app, json!({"id": id, "type": "T"})).await;
4600 4 : assert_eq!(
4601 4 : patch(
4602 4 : &app,
4603 4 : &format!("/ngsi-ld/v1/entities/{id}?observedAt=yesterday"),
4604 4 : json!({"a": {"type": "Property", "value": 1}}),
4605 4 : )
4606 4 : .await,
4607 4 : StatusCode::BAD_REQUEST
4608 4 : );
4609 4 : }
4610 :
4611 : /// 5.6.17.4: "If a common language tag is defined and a LanguageProperty
4612 : /// Attribute to be merged is represented as a string, the pre-existing
4613 : /// languageMap JSON object shall be preserved. The string value shall
4614 : /// only replace the value associated to the language tag key found
4615 : /// within the languageMap."
4616 : #[tokio::test]
4617 4 : async fn merge_with_a_common_lang_replaces_only_that_language_key() {
4618 4 : let app = app();
4619 4 : let id = "urn:ngsi-ld:Lang:1";
4620 4 : create(
4621 4 : &app,
4622 4 : json!({"id": id, "type": "T",
4623 4 : "greeting": {"type": "LanguageProperty",
4624 4 : "languageMap": {"en": "hello", "es": "adios"}}}),
4625 4 : )
4626 4 : .await;
4627 4 : assert_eq!(
4628 4 : patch(
4629 4 : &app,
4630 4 : &format!("/ngsi-ld/v1/entities/{id}?lang=es"),
4631 4 : json!({"greeting": "hola"}),
4632 4 : )
4633 4 : .await,
4634 : StatusCode::NO_CONTENT
4635 : );
4636 4 : let body = get(&app, &format!("/ngsi-ld/v1/entities/{id}")).await;
4637 4 : assert_eq!(
4638 4 : body["greeting"]["languageMap"],
4639 4 : json!({"en": "hello", "es": "hola"}),
4640 : "the pre-existing languageMap survives, only the tagged key moves: {body}"
4641 : );
4642 4 : assert!(
4643 4 : body["greeting"].get("value").is_none(),
4644 : "the string never lands as a plain Property value: {body}"
4645 : );
4646 4 : assert_eq!(body["greeting"]["type"], "LanguageProperty", "{body}");
4647 4 : }
4648 :
4649 : /// Table 6.4.3.2-1 containedBy: "List of entity ids which have previously
4650 : /// been encountered whilst retrieving the Entity Graph" — 4.5.23.1 keeps
4651 : /// the walk from retrieving them a second time.
4652 : #[tokio::test]
4653 4 : async fn contained_by_ids_are_not_retrieved_again() {
4654 4 : let app = app();
4655 4 : let (root, held, fresh) = (
4656 4 : "urn:ngsi-ld:Graph:root",
4657 4 : "urn:ngsi-ld:Graph:held",
4658 4 : "urn:ngsi-ld:Graph:fresh",
4659 4 : );
4660 4 : create(&app, json!({"id": held, "type": "G"})).await;
4661 4 : create(&app, json!({"id": fresh, "type": "G"})).await;
4662 4 : create(
4663 4 : &app,
4664 4 : json!({"id": root, "type": "G",
4665 4 : "toHeld": {"type": "Relationship", "object": held},
4666 4 : "toFresh": {"type": "Relationship", "object": fresh}}),
4667 4 : )
4668 4 : .await;
4669 :
4670 4 : let body = get(
4671 4 : &app,
4672 4 : &format!("/ngsi-ld/v1/entities/{root}?join=flat&joinLevel=2&containedBy={held}"),
4673 4 : )
4674 4 : .await;
4675 4 : let arr = match body {
4676 4 : Value::Array(a) => a,
4677 0 : other => vec![other],
4678 : };
4679 8 : let ids: Vec<&str> = arr.iter().filter_map(|e| e["id"].as_str()).collect();
4680 4 : assert!(
4681 4 : !ids.contains(&held),
4682 : "an id the client already holds is not retrieved again: {ids:?}"
4683 : );
4684 4 : assert!(
4685 4 : ids.contains(&fresh),
4686 : "the Linked Entity it does not hold is still returned: {ids:?}"
4687 : );
4688 4 : assert!(ids.contains(&root), "the Linking Entity is there: {ids:?}");
4689 4 : }
4690 : }
4691 :
4692 : #[cfg(test)]
4693 : mod clause_4_8_system_attributes {
4694 : use super::*;
4695 : use crate::stamp::stamp_instances;
4696 : use axum::body::Body;
4697 : use axum::http::Request;
4698 : use serde_json::json;
4699 : use tower::ServiceExt;
4700 :
4701 : /// 4.8 stamps a sub-Attribute because "a sub-Property is a Property".
4702 : /// The members an Attribute instance carries under 4.5 are NOT
4703 : /// sub-Attributes, so none of them may be descended into and stamped —
4704 : /// several of them (`previousJson` and `previousVocab` hold uninterpreted
4705 : /// JSON per 4.5.20/4.5.21, `entityList` holds Linked Entities) can be an
4706 : /// array of objects, which is exactly the shape the walk mistakes for an
4707 : /// instance array. The list the expander keeps verbatim is the one list;
4708 : /// a second copy is what drifts.
4709 : #[test]
4710 4 : fn no_reserved_instance_member_is_walked_as_a_sub_attribute() {
4711 4 : let carrier = json!([{"probe": "untouched"}]);
4712 120 : for member in antares_jsonld::RESERVED_MEMBERS {
4713 : // the two the stamp itself writes; every other member is the
4714 : // instance's own and is left exactly as it was found
4715 120 : if matches!(*member, "createdAt" | "modifiedAt") {
4716 8 : continue;
4717 112 : }
4718 112 : let mut inst = json!({"type": "Property", "value": 1});
4719 112 : inst[*member] = carrier.clone();
4720 112 : let mut attr = json!([inst]);
4721 112 : stamp_instances(&mut attr, "2020-01-01T00:00:00Z");
4722 112 : let held = attr[0].get(*member).expect("the member survives");
4723 112 : assert_eq!(
4724 112 : held, &carrier,
4725 : "{member} was walked as a sub-Attribute and stamped"
4726 : );
4727 : }
4728 4 : }
4729 :
4730 : /// 4.8: createdAt is "the temporal Property at which the Entity,
4731 : /// Property or Relationship was entered into an NGSI-LD system" and
4732 : /// modifiedAt the one at which it "was last modified". Both are
4733 : /// generated by the system, at every level: an Entity, an Attribute
4734 : /// instance and a sub-Attribute (a sub-Property is a Property). A client
4735 : /// that could set them would rewrite the provenance of its own data, and
4736 : /// a subscriber filtering on modifiedAt would never see the write.
4737 : #[tokio::test]
4738 4 : async fn the_client_cannot_write_its_own_created_and_modified_stamps() {
4739 4 : let st = AppState::new("antares-sysattrs".into());
4740 4 : let forged = "1970-01-01T00:00:00Z";
4741 4 : let payload = json!({
4742 4 : "id": "urn:ngsi-ld:Vehicle:stamped",
4743 4 : "type": "Vehicle",
4744 4 : "createdAt": forged,
4745 4 : "modifiedAt": forged,
4746 4 : "speed": {
4747 4 : "type": "Property",
4748 4 : "value": 10,
4749 4 : "createdAt": forged,
4750 4 : "modifiedAt": forged,
4751 4 : "accuracy": {"type": "Property", "value": 1,
4752 4 : "createdAt": forged, "modifiedAt": forged},
4753 : },
4754 : })
4755 4 : .to_string();
4756 4 : let resp = crate::router(st.clone())
4757 4 : .oneshot(
4758 4 : Request::post("/ngsi-ld/v1/entities")
4759 4 : .header("Content-Type", "application/json")
4760 4 : .header("Content-Length", payload.len().to_string())
4761 4 : .body(Body::from(payload))
4762 4 : .expect("req"),
4763 4 : )
4764 4 : .await
4765 4 : .expect("resp");
4766 4 : assert_eq!(resp.status(), StatusCode::CREATED);
4767 :
4768 4 : let resp = crate::router(st)
4769 4 : .oneshot(
4770 4 : Request::get("/ngsi-ld/v1/entities/urn:ngsi-ld:Vehicle:stamped?options=sysAttrs")
4771 4 : .body(Body::empty())
4772 4 : .expect("req"),
4773 4 : )
4774 4 : .await
4775 4 : .expect("resp");
4776 4 : assert_eq!(resp.status(), StatusCode::OK);
4777 4 : let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
4778 4 : .await
4779 4 : .expect("body");
4780 4 : let served = String::from_utf8_lossy(&body);
4781 4 : assert!(
4782 4 : served.contains("createdAt") && served.contains("modifiedAt"),
4783 : "sysAttrs serves the stamps: {served}"
4784 : );
4785 4 : assert!(
4786 4 : !served.contains(forged),
4787 4 : "no level of the document may carry the stamp the client sent: {served}"
4788 4 : );
4789 4 : }
4790 :
4791 : /// 4.8 again, on every route that writes an Attribute. Create is the
4792 : /// obvious way in; a client that is refused there and accepted on Append,
4793 : /// Merge, Partial Update or Replace has the same forged provenance one
4794 : /// request later. Each of the four carries the stamps at Entity,
4795 : /// Attribute and sub-Attribute level.
4796 : #[tokio::test]
4797 4 : async fn no_write_route_lets_the_client_stamp_its_own_attributes() {
4798 4 : let st = AppState::new("antares-sysattrs-writes".into());
4799 4 : let forged = "1970-01-01T00:00:00Z";
4800 4 : let id = "urn:ngsi-ld:Vehicle:writes";
4801 44 : let stamped = |v: Value| {
4802 44 : let mut o = v;
4803 44 : if let Some(m) = o.as_object_mut() {
4804 44 : m.insert("createdAt".into(), json!(forged));
4805 44 : m.insert("modifiedAt".into(), json!(forged));
4806 44 : }
4807 44 : o
4808 44 : };
4809 36 : let send = |req: Request<Body>| {
4810 36 : let st = st.clone();
4811 36 : async move { crate::router(st).oneshot(req).await.expect("resp") }
4812 36 : };
4813 :
4814 4 : let seed = json!({"id": id, "type": "Vehicle"}).to_string();
4815 4 : let resp = send(
4816 4 : Request::post("/ngsi-ld/v1/entities")
4817 4 : .header("Content-Type", "application/json")
4818 4 : .header("Content-Length", seed.len().to_string())
4819 4 : .body(Body::from(seed))
4820 4 : .expect("req"),
4821 4 : )
4822 4 : .await;
4823 4 : assert_eq!(resp.status(), StatusCode::CREATED);
4824 :
4825 16 : let attr = || {
4826 16 : stamped(json!({
4827 16 : "type": "Property",
4828 16 : "value": 10,
4829 16 : "accuracy": stamped(json!({"type": "Property", "value": 1})),
4830 16 : }))
4831 16 : };
4832 : // Append (5.6.3), Partial Update (5.6.4), Merge (5.6.17), Replace
4833 : // (5.6.16) — the whole write surface that reaches expand_entity.
4834 4 : let calls: Vec<(&str, String, Value)> = vec![
4835 4 : (
4836 4 : "POST",
4837 4 : format!("/ngsi-ld/v1/entities/{id}/attrs"),
4838 4 : stamped(json!({"speed": attr()})),
4839 4 : ),
4840 4 : (
4841 4 : "PATCH",
4842 4 : format!("/ngsi-ld/v1/entities/{id}/attrs/speed"),
4843 4 : attr(),
4844 4 : ),
4845 4 : (
4846 4 : "PATCH",
4847 4 : format!("/ngsi-ld/v1/entities/{id}"),
4848 4 : stamped(json!({"speed": attr()})),
4849 4 : ),
4850 4 : (
4851 4 : "PUT",
4852 4 : format!("/ngsi-ld/v1/entities/{id}"),
4853 4 : stamped(json!({"id": id, "type": "Vehicle", "speed": attr()})),
4854 4 : ),
4855 : ];
4856 16 : for (method, path, payload) in calls {
4857 16 : let body = payload.to_string();
4858 16 : let resp = send(
4859 16 : Request::builder()
4860 16 : .method(method)
4861 16 : .uri(&path)
4862 16 : .header("Content-Type", "application/json")
4863 16 : .header("Content-Length", body.len().to_string())
4864 16 : .body(Body::from(body))
4865 16 : .expect("req"),
4866 16 : )
4867 16 : .await;
4868 16 : assert!(
4869 16 : resp.status().is_success(),
4870 4 : "{method} {path} answered {}",
4871 4 : resp.status()
4872 4 : );
4873 16 : let resp = send(
4874 16 : Request::get(format!("/ngsi-ld/v1/entities/{id}?options=sysAttrs"))
4875 16 : .body(Body::empty())
4876 16 : .expect("req"),
4877 16 : )
4878 16 : .await;
4879 16 : let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
4880 16 : .await
4881 16 : .expect("body");
4882 16 : let served = String::from_utf8_lossy(&bytes);
4883 16 : assert!(
4884 16 : !served.contains(forged),
4885 4 : "{method} {path} let the client stamp the document: {served}"
4886 4 : );
4887 4 : }
4888 4 : }
4889 : }
4890 :
4891 : #[cfg(test)]
4892 : mod judged_types_tests {
4893 : use super::judged_types;
4894 : use crate::policy::{Filter, Rule};
4895 : use std::borrow::Cow;
4896 :
4897 16 : fn rules(types: &[&[&str]]) -> Filter {
4898 : Filter {
4899 16 : rules: types
4900 16 : .iter()
4901 16 : .map(|t| Rule {
4902 28 : types: t.iter().map(|s| (*s).to_owned()).collect(),
4903 28 : ..Rule::default()
4904 28 : })
4905 16 : .collect(),
4906 16 : ..Filter::default()
4907 : }
4908 16 : }
4909 :
4910 20 : fn sel(groups: &[&[&str]]) -> Vec<Vec<String>> {
4911 20 : groups
4912 20 : .iter()
4913 64 : .map(|g| g.iter().map(|s| (*s).to_owned()).collect())
4914 20 : .collect()
4915 20 : }
4916 :
4917 : #[test]
4918 4 : fn a_selection_naming_a_rule_type_in_every_group_is_kept_as_is() {
4919 4 : let client = sel(&[&["a", "T"], &["U"]]);
4920 4 : let got = judged_types(Some(&client), &rules(&[&["T"], &["U"]]));
4921 4 : assert!(matches!(got, Some(Cow::Borrowed(g)) if g == client.as_slice()));
4922 4 : }
4923 :
4924 : #[test]
4925 4 : fn a_group_naming_no_rule_type_takes_each_in_turn() {
4926 4 : let client = sel(&[&["a", "b"], &["T"]]);
4927 4 : let got = judged_types(Some(&client), &rules(&[&["U", "T"], &["T"]]));
4928 4 : assert_eq!(
4929 4 : got.as_deref(),
4930 4 : Some(sel(&[&["a", "b", "T"], &["a", "b", "U"], &["T"]]).as_slice())
4931 : );
4932 4 : }
4933 :
4934 : #[test]
4935 4 : fn no_client_selection_selects_the_rule_types() {
4936 4 : let got = judged_types(None, &rules(&[&["T"], &["U"]]));
4937 4 : assert_eq!(got.as_deref(), Some(sel(&[&["T"], &["U"]]).as_slice()));
4938 4 : }
4939 :
4940 : #[test]
4941 4 : fn rules_without_types_judge_nothing() {
4942 4 : assert!(judged_types(None, &rules(&[&[]])).is_none());
4943 4 : assert!(judged_types(Some(&sel(&[&["a"]])), &Filter::default()).is_none());
4944 4 : }
4945 : }
|
