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 7670 : fn entity_location(id: &str) -> String {
48 7670 : format!(
49 : "/ngsi-ld/v1/entities/{}",
50 7670 : crate::federation::path_segment(id)
51 : )
52 7670 : }
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 40 : fn type_selector_query(params: &HashMap<String, String>) -> Vec<(String, String)> {
60 40 : params
61 40 : .get("type")
62 40 : .map(|v| ("type".to_owned(), v.clone()))
63 40 : .into_iter()
64 40 : .collect()
65 40 : }
66 :
67 9970 : pub async fn create_entity(
68 9970 : State(st): State<AppState>,
69 9970 : CleanParams(params): CleanParams,
70 9970 : headers: HeaderMap,
71 9970 : body: Bytes,
72 9970 : ) -> Response {
73 9970 : match create_entity_inner(&st, ¶ms, &headers, &body).await {
74 7670 : Ok(r) => r,
75 2300 : Err(e) => e.into_response(),
76 : }
77 9970 : }
78 :
79 9970 : async fn create_entity_inner(
80 9970 : st: &AppState,
81 9970 : params: &HashMap<String, String>,
82 9970 : headers: &HeaderMap,
83 9970 : body: &[u8],
84 9970 : ) -> ApiResult<Response> {
85 9970 : let tenant = tenant_from(headers)?;
86 9970 : check_params(params, &["local"])?;
87 9966 : let parsed = parse_body(&st.loader, headers, body, BodyKind::Standard).await?;
88 7784 : let obj = parsed.object(NgsiError::InvalidRequest(
89 7784 : "entity document must be a JSON object".into(),
90 7784 : ))?;
91 7784 : let mut expanded = expand_entity(obj, &parsed.ctx, ExpandOpts::default())?;
92 7682 : let id = antares_jsonld::expanded_id(&expanded)?.to_owned();
93 :
94 : // distributed create (4.3.6, 6.4.3.1)
95 7682 : let types: Option<Vec<String>> = expanded["type"].as_array().map(|a| {
96 7682 : a.iter()
97 7682 : .filter_map(Value::as_str)
98 7682 : .map(str::to_owned)
99 7682 : .collect()
100 7682 : });
101 7682 : let attr_iris: Vec<String> = expanded
102 7682 : .as_object()
103 26432 : .map(|o| o.keys().filter(|k| !is_meta(k)).cloned().collect())
104 7682 : .unwrap_or_default();
105 7682 : let spec = crate::registry::CsrSpec {
106 7682 : ids: Some(vec![id.clone()]),
107 7682 : types,
108 7682 : attrs: (!attr_iris.is_empty()).then_some(attr_iris),
109 7682 : ..Default::default()
110 7682 : };
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 7682 : 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 7682 : .await?;
122 7676 : let regs = match crate::federation::write_plan(st, &tenant, &spec, &parsed.ctx, params, headers)
123 7676 : .await?
124 : {
125 0 : crate::federation::WritePlan::Answered(r) => return Ok(*r),
126 7676 : crate::federation::WritePlan::Forward(regs) => regs,
127 : };
128 7676 : if !regs.is_empty() {
129 80 : let mut conflicts = Vec::new();
130 80 : let mut fwd = Vec::new();
131 80 : for reg in ®s {
132 : // 5.6.1.4: exclusive/redirect registrations not supporting the
133 : // Create Entity operation yield an error of type Conflict (and
134 : // are never contacted); an inclusive one is simply not forwarded.
135 80 : if !reg.supports("createEntity") {
136 74 : if reg.is_proxy() {
137 2 : conflicts.push(crate::federation::conflict_part("createEntity"));
138 72 : }
139 74 : continue;
140 6 : }
141 6 : if let Some(frag) = crate::federation::reduce_to_scope(obj, reg, &parsed.ctx) {
142 6 : fwd.push((reg.clone(), frag));
143 6 : }
144 : }
145 80 : let proxies: Vec<&crate::federation::FedReg> =
146 80 : regs.iter().filter(|r| r.is_proxy()).collect();
147 80 : let (rest, has_attrs) = crate::federation::strip_proxied(obj, &proxies, &parsed.ctx);
148 80 : let mut parts = Vec::new();
149 : // local part only when something non-proxied remains (4.3.6.3)
150 80 : if has_attrs || proxies.is_empty() {
151 76 : let mut local_exp = expand_entity(&rest, &parsed.ctx, ExpandOpts::default())?;
152 76 : stamp_new(&mut local_exp, &now_iso());
153 76 : if st
154 76 : .store
155 76 : .create(&tenant, Kind::Entity, &id, local_exp.clone())
156 76 : .await?
157 76 : {
158 76 : parts.push(crate::federation::Part {
159 76 : status: 201,
160 76 : detail: "created locally".into(),
161 76 : });
162 76 : } else {
163 0 : parts.push(crate::federation::Part {
164 0 : status: 409,
165 0 : detail: format!("entity {id} already exists"),
166 0 : });
167 0 : }
168 4 : }
169 80 : parts.extend(conflicts);
170 80 : let ctx_url = crate::federation::ctx_link_url(headers, &parsed.ctx.source);
171 80 : for (reg, frag) in fwd {
172 6 : parts.push(
173 6 : crate::federation::forward_part(
174 6 : st,
175 6 : reqwest::Method::POST,
176 6 : format!("{}/ngsi-ld/v1/entities", reg.endpoint),
177 6 : &[],
178 6 : headers,
179 6 : &tenant,
180 6 : ®,
181 6 : &ctx_url,
182 6 : Some(frag),
183 6 : )
184 6 : .await,
185 : );
186 : }
187 80 : return Ok(crate::federation::combine(
188 80 : parts,
189 80 : created(entity_location(&id), &tenant),
190 80 : &tenant,
191 80 : ));
192 7596 : }
193 :
194 7596 : stamp_new(&mut expanded, &now_iso());
195 7596 : if !st
196 7596 : .store
197 7596 : .create(&tenant, Kind::Entity, &id, expanded.clone())
198 7596 : .await?
199 : {
200 6 : return Err(NgsiError::AlreadyExists(format!("entity {id} already exists")).into());
201 7590 : }
202 7590 : Ok(created(entity_location(&id), &tenant))
203 9970 : }
204 :
205 : // ---------- GET /entities/{id} (5.7.1) ----------
206 :
207 598 : pub async fn retrieve_entity(
208 598 : State(st): State<AppState>,
209 598 : Path(id): Path<String>,
210 598 : CleanParams(params): CleanParams,
211 598 : headers: HeaderMap,
212 598 : ) -> Response {
213 598 : match retrieve_entity_outer(&st, &id, ¶ms, &headers).await {
214 512 : Ok(r) => r,
215 86 : Err(e) => e.into_response(),
216 : }
217 598 : }
218 :
219 : /// 5.7.1.4 Retrieve Entity: the EntityMap half of the clause is the shared
220 : /// rule (`entity_maps::retrieve_with_map`); this is the retrieve it wraps.
221 598 : async fn retrieve_entity_outer(
222 598 : st: &AppState,
223 598 : id: &str,
224 598 : params: &HashMap<String, String>,
225 598 : headers: &HeaderMap,
226 598 : ) -> ApiResult<Response> {
227 598 : crate::entity_map::retrieve_with_map(st, id, params, headers, false, |map| async move {
228 596 : retrieve_entity_inner(st, id, params, headers, map.as_ref()).await
229 1192 : })
230 598 : .await
231 598 : }
232 :
233 596 : async fn retrieve_entity_inner(
234 596 : st: &AppState,
235 596 : id: &str,
236 596 : params: &HashMap<String, String>,
237 596 : headers: &HeaderMap,
238 596 : map: Option<&Value>,
239 596 : ) -> ApiResult<Response> {
240 596 : let tenant = tenant_from(headers)?;
241 596 : check_params(
242 596 : params,
243 596 : &[
244 596 : "attrs",
245 596 : "pick",
246 596 : "omit",
247 596 : "options",
248 596 : "format",
249 596 : "lang",
250 596 : "type",
251 596 : "geometryProperty",
252 596 : "datasetId",
253 596 : "containedBy",
254 596 : "join",
255 596 : "joinLevel",
256 596 : "local",
257 596 : "entityMap",
258 596 : ],
259 0 : )?;
260 596 : let accept = parse_accept_geo(headers)?;
261 : // 5.7.1.4: geometryProperty is only meaningful for the GeoJSON
262 : // representation — any other Accept is BadRequestData
263 592 : if params.contains_key("geometryProperty") && accept != Accept::GeoJson {
264 8 : return Err(NgsiError::BadRequestData(
265 8 : "geometryProperty requires Accept: application/geo+json (5.7.1.4)".into(),
266 8 : )
267 8 : .into());
268 584 : }
269 584 : let ctx = request_context(&st.loader, headers).await?;
270 564 : let filter = gate!(st, &tenant, headers, "5.7.1", ids: &[id]).await?;
271 564 : let mut repr = parse_repr(params, &ctx)?;
272 548 : crate::repr::narrow_repr(&mut repr, &filter);
273 548 : let join = parse_join(params)?;
274 548 : check_linked_projection(&repr, &join)?;
275 536 : antares_model::EntityId::new(id)?;
276 532 : let local_doc = st.store.get(&tenant, Kind::Entity, id).await?;
277 528 : let looped = crate::federation::via_loop(
278 528 : headers,
279 528 : &crate::federation::alias_for(&st.host_alias, &tenant),
280 : );
281 528 : let fed_on = crate::federation::active(params) && !looped;
282 : // 6.3.17: abnormal distributed-GET outcomes surface as NGSILD-Warning
283 528 : let mut warnings: Vec<String> = Vec::new();
284 528 : if crate::federation::active(params) && looped {
285 2 : let spec = crate::registry::CsrSpec {
286 2 : ids: Some(vec![id.to_owned()]),
287 2 : ..Default::default()
288 2 : };
289 : // only a loop that suppressed a real forward is abnormal behaviour
290 2 : if !crate::federation::matching_regs(st, &tenant, &spec, &ctx, headers)
291 2 : .await?
292 2 : .is_empty()
293 2 : {
294 2 : warnings.push(crate::federation::warning(
295 2 : 199,
296 2 : &crate::federation::alias_for(&st.host_alias, &tenant),
297 2 : "a registration loop has been detected",
298 2 : ));
299 2 : }
300 526 : }
301 528 : let doc = if fed_on {
302 510 : let fed = crate::federation::fed_retrieve(
303 510 : st,
304 510 : &tenant,
305 510 : headers,
306 510 : &ctx,
307 510 : id,
308 510 : map,
309 510 : None,
310 510 : &mut warnings,
311 510 : )
312 510 : .await?;
313 508 : match local_doc {
314 432 : Some(mut base) => {
315 864 : for aux_pass in [false, true] {
316 864 : for (aux, d) in &fed {
317 8 : if *aux == aux_pass {
318 4 : crate::federation::merge_docs(&mut base, d, *aux);
319 4 : }
320 : }
321 : }
322 432 : base
323 : }
324 : None => {
325 76 : let first = fed
326 76 : .iter()
327 76 : .find(|(aux, _)| !aux)
328 76 : .map(|(_, d)| d.clone())
329 76 : .or_else(|| fed.first().map(|(_, d)| d.clone()));
330 76 : let Some(mut base) = first else {
331 : // 6.3.17: abnormal distributed outcomes surface as
332 : // NGSILD-Warning even when the retrieve ends 404
333 76 : let mut resp = ApiError::from(NgsiError::ResourceNotFound(format!(
334 76 : "entity {id} not found"
335 76 : )))
336 76 : .into_response();
337 76 : attach_warnings(&mut resp, &warnings);
338 76 : echo_tenant(&tenant, &mut resp);
339 76 : return Ok(resp);
340 : };
341 0 : for aux_pass in [false, true] {
342 0 : for (aux, d) in &fed {
343 0 : if *aux == aux_pass {
344 0 : crate::federation::merge_docs(&mut base, d, *aux);
345 0 : }
346 : }
347 : }
348 0 : base
349 : }
350 : }
351 : } else {
352 18 : local_doc.ok_or_else(|| NgsiError::ResourceNotFound(format!("entity {id} not found")))?
353 : };
354 : // 5.7.1.4: no entity "whose id (URI), and where specified type, is
355 : // equivalent" — the optional ?type selector (4.17) narrows the target
356 444 : if !crate::negotiate::matches_type_param(&doc, params, &ctx) {
357 4 : return Err(NgsiError::ResourceNotFound(format!(
358 4 : "entity {id} does not match the type selector"
359 4 : ))
360 4 : .into());
361 440 : }
362 : // ADR-0020: an Entity outside the engine's narrowing answers the way
363 : // 5.7.1.4 answers an absent one — "If the NGSI-LD Entity does not
364 : // exist, an error of type ResourceNotFound shall be raised" — because a
365 : // refusal here would tell the caller the Entity is there.
366 440 : if let Some(ast) = &filter.q {
367 4 : if !crate::notify::linked_eval(st, &tenant, |l| {
368 4 : antares_ql::eval::eval_q(ast, &doc, &ctx, l)
369 4 : })
370 4 : .await
371 : {
372 2 : return Err(NgsiError::ResourceNotFound(format!("entity {id} not found")).into());
373 2 : }
374 436 : }
375 438 : if let Some(scope) = &filter.scope_q {
376 0 : if !crate::scope_matches(scope, &doc) {
377 0 : return Err(NgsiError::ResourceNotFound(format!("entity {id} not found")).into());
378 0 : }
379 438 : }
380 : // 5.7.1: attrs projection with no matching attribute ⇒ 404
381 438 : if let Some(want) = &repr.attrs {
382 0 : if !want.iter().any(|a| doc.get(a).is_some()) {
383 0 : return Err(NgsiError::ResourceNotFound(format!(
384 0 : "entity {id} has none of the requested attributes"
385 0 : ))
386 0 : .into());
387 0 : }
388 438 : }
389 438 : let shaped = apply(&doc, &repr);
390 438 : if (repr.pick.is_some() || repr.omit.is_some())
391 6 : && shaped.as_object().is_some_and(|o| o.is_empty())
392 : {
393 2 : return Err(NgsiError::ResourceNotFound(format!(
394 2 : "projection matches nothing on entity {id}"
395 2 : ))
396 2 : .into());
397 436 : }
398 436 : let mut payload = compact_for(&repr, &shaped, &ctx);
399 436 : if let Some((mode, level)) = &join {
400 22 : let held = contained_by(params);
401 22 : let complete = match mode.as_str() {
402 22 : "inline" => {
403 12 : inline_join_beyond(
404 12 : st,
405 12 : &tenant,
406 12 : &ctx,
407 12 : &repr,
408 12 : &mut payload,
409 12 : *level,
410 12 : &held,
411 12 : &mut { MAX_JOIN_LOOKUPS },
412 12 : )
413 12 : .await
414 : }
415 10 : "flat" => {
416 10 : let mut linked = std::collections::BTreeMap::new();
417 10 : let complete = collect_flat_beyond(
418 10 : st,
419 10 : &tenant,
420 10 : &repr,
421 10 : &doc,
422 10 : *level,
423 10 : &mut linked,
424 10 : &held,
425 10 : &mut { MAX_JOIN_LOOKUPS },
426 10 : )
427 10 : .await;
428 10 : if !linked.is_empty() {
429 10 : let mut arr = vec![payload];
430 12 : for (_, (ldoc, lrepr)) in linked {
431 12 : arr.push(compact_for(&lrepr, &apply(&ldoc, &lrepr), &ctx));
432 12 : }
433 10 : payload = Value::Array(arr);
434 0 : }
435 10 : complete
436 : }
437 0 : _ => true,
438 : };
439 22 : if !complete {
440 4 : warnings.push(crate::federation::warning(
441 4 : 199,
442 4 : &crate::federation::alias_for(&st.host_alias, &tenant),
443 4 : "the linked entity retrieval was truncated",
444 4 : ));
445 18 : }
446 414 : }
447 436 : let payload = if accept == Accept::GeoJson {
448 6 : to_geojson_feature(payload, params.get("geometryProperty"))
449 : } else {
450 430 : payload
451 : };
452 436 : let mut resp = respond_prefer(StatusCode::OK, payload, &ctx, accept, &tenant, headers);
453 436 : attach_warnings(&mut resp, &warnings);
454 436 : filter.mark_restricted(resp.headers_mut());
455 436 : Ok(resp)
456 596 : }
457 :
458 : /// 5.7.1.4 / 5.7.2.4: a `{…}` projection selects into Linked Entities —
459 : /// it must be requested via join, and may not select deeper than joinLevel.
460 1144 : pub(crate) fn check_linked_projection(
461 1144 : repr: &crate::repr::Repr,
462 1144 : join: &Option<(String, usize)>,
463 1144 : ) -> ApiResult<()> {
464 1144 : let depth = repr
465 1144 : .pick
466 1144 : .as_deref()
467 1144 : .map(crate::repr::proj_depth)
468 1144 : .unwrap_or(0)
469 1144 : .max(
470 1144 : repr.omit
471 1144 : .as_deref()
472 1144 : .map(crate::repr::proj_depth)
473 1144 : .unwrap_or(0),
474 : );
475 1144 : if depth == 0 {
476 1124 : return Ok(());
477 20 : }
478 8 : match join {
479 8 : Some((mode, level)) if mode != "@none" => {
480 8 : if depth > *level {
481 4 : return Err(NgsiError::BadRequestData(format!(
482 4 : "projected attribute depth {depth} exceeds joinLevel {level} (5.7.1.4/5.7.2.4)"
483 4 : ))
484 4 : .into());
485 4 : }
486 4 : Ok(())
487 : }
488 12 : _ => Err(NgsiError::BadRequestData(
489 12 : "projection uses Linked Entity selection but join is not specified (5.7.1.4/5.7.2.4)"
490 12 : .into(),
491 12 : )
492 12 : .into()),
493 : }
494 1144 : }
495 :
496 : /// join/joinLevel params (4.5.23). Returns (mode, level).
497 1152 : pub fn parse_join(params: &HashMap<String, String>) -> ApiResult<Option<(String, usize)>> {
498 1152 : let Some(mode) = params.get("join") else {
499 1096 : return Ok(None);
500 : };
501 56 : if !["inline", "flat", "@none"].contains(&mode.as_str()) {
502 4 : return Err(NgsiError::BadRequestData(format!("invalid join {mode:?}")).into());
503 52 : }
504 52 : let level = match params.get("joinLevel") {
505 48 : Some(l) => l
506 48 : .parse::<usize>()
507 48 : .ok()
508 : // Bounded traversal depth
509 48 : .filter(|l| *l >= 1 && *l <= crate::bounds::MAX_JOIN_LEVEL)
510 48 : .ok_or_else(|| {
511 4 : NgsiError::BadRequestData(format!(
512 4 : "invalid joinLevel {l:?} (1..={})",
513 4 : crate::bounds::MAX_JOIN_LEVEL
514 4 : ))
515 4 : })?,
516 4 : None => 1,
517 : };
518 48 : if mode == "@none" {
519 4 : return Ok(None);
520 44 : }
521 44 : Ok(Some((mode.clone(), level)))
522 1152 : }
523 :
524 : /// Table 6.4.3.2-1 `containedBy`: "List of entity ids which have previously
525 : /// been encountered whilst retrieving the Entity Graph. Only applicable if
526 : /// joinLevel is present." They are already in the graph the client is
527 : /// assembling, so 4.5.23.1's "avoid ... duplicates or loops" counts them as
528 : /// resolved and the walk does not follow them again.
529 32 : pub fn contained_by(params: &HashMap<String, String>) -> Vec<String> {
530 32 : params
531 32 : .get("containedBy")
532 32 : .map(|s| {
533 4 : s.split(',')
534 4 : .map(str::trim)
535 4 : .filter(|t| !t.is_empty())
536 4 : .map(str::to_owned)
537 4 : .collect()
538 4 : })
539 32 : .unwrap_or_default()
540 32 : }
541 :
542 : // ---------- GET /entities/ (5.7.2) ----------
543 :
544 624 : pub async fn query_entities(
545 624 : State(st): State<AppState>,
546 624 : CleanParams(params): CleanParams,
547 624 : headers: HeaderMap,
548 624 : ) -> Response {
549 624 : match query_entities_outer(&st, params, &headers).await {
550 434 : Ok(r) => r,
551 190 : Err(e) => e.into_response(),
552 : }
553 624 : }
554 :
555 : /// 5.5.14 / 5.5.9.3: a query referencing an EntityMap (NGSILD-EntityMap
556 : /// request header, 6.4.3.2-2) is fixed to the map's Entities; the filters
557 : /// are re-checked at processing time and local entries that no longer match
558 : /// are removed from the map by its creator. An expired or unknown map means
559 : /// "no inference can be made … a new one shall be created".
560 624 : async fn query_entities_outer(
561 624 : st: &AppState,
562 624 : mut params: HashMap<String, String>,
563 624 : headers: &HeaderMap,
564 624 : ) -> ApiResult<Response> {
565 624 : let tenant = tenant_from(headers)?;
566 624 : let q_ast = params.get("q").map(|q| parse_q(q)).transpose()?;
567 624 : let filter = gate!(
568 : st, &tenant, headers, "5.7.2",
569 : q: q_ast.as_ref(),
570 : scope_q: params.get("scopeQ").map(String::as_str),
571 : )
572 624 : .await?;
573 622 : let Some(map_ref) = single_header(headers, "NGSILD-EntityMap")? else {
574 574 : return query_entities_inner(st, ¶ms, headers, &filter).await;
575 : };
576 : // 5.7.2.4: an unknown parameter and a too-wide query are BadRequestData
577 : // for this request, whether or not it carries an EntityMap reference —
578 : // and the paged fetch below walks the whole map, locally and forwarded,
579 : // before the inner call would reach these same two checks.
580 44 : check_params(¶ms, crate::negotiate::QUERY_PARAMS)?;
581 44 : if !qualifies_non_wide(¶ms, q_ast.as_ref()) {
582 0 : return Err(NgsiError::BadRequestData(
583 0 : "query needs at least one of type, attrs, q, georel (5.7.2)".into(),
584 0 : )
585 0 : .into());
586 44 : }
587 44 : let map_id = map_ref.rsplit('/').next().unwrap_or(&map_ref).to_owned();
588 44 : let Some(mut map) = crate::entity_map::map_if_accessible(st, &tenant, headers, &map_id).await
589 : else {
590 : // 5.5.14: expired or inaccessible → a new EntityMap is created
591 6 : params.insert("entityMap".into(), "true".into());
592 6 : return query_entities_inner(st, ¶ms, headers, &filter).await;
593 : };
594 38 : let ctx = request_context(&st.loader, headers).await?;
595 : // a request that references a live map does not create a new one
596 38 : params.remove("entityMap");
597 38 : let (offset, limit, count) = page_params(st, ¶ms)?;
598 : // pagination links carry the ORIGINAL query, never the page's id list
599 38 : let link_params = params.clone();
600 38 : let accept = parse_accept_geo(headers)?;
601 :
602 : // 5.5.9.3 paged fetch: the map fixes the candidate id set; candidates
603 : // are fetched (locally + forwarded) chunk by chunk, "filters shall be
604 : // rechecked before returning results" per chunk, and visited entries
605 : // that no longer match are removed from the map — "Entities not or no
606 : // longer fitting the query shall be removed from the Entity map during
607 : // pagination". Pruning is judgeable only for "@none" (local) entries: a
608 : // remote-backed id may merely have an unreachable source right now
609 : // (5.5.14). Memory per request is O(chunk), never O(map) — the reason
610 : // EntityMaps exist for the distributed case. count=true walks every
611 : // candidate (the total needs each id checked), still chunk-bounded.
612 38 : let ids: Vec<String> = crate::entity_map::candidate_ids(&map, ¶ms);
613 38 : let looped = crate::federation::via_loop(
614 38 : headers,
615 38 : &crate::federation::alias_for(&st.host_alias, &tenant),
616 : );
617 38 : let chunk_size = limit.max(20);
618 38 : let mut page_ids: Vec<String> = Vec::new();
619 38 : let (mut skipped, mut total, mut more, mut visited) = (0usize, 0usize, false, 0usize);
620 38 : for chunk in ids.chunks(chunk_size) {
621 38 : let mut p = params.clone();
622 38 : p.remove("limit");
623 38 : p.remove("offset");
624 38 : p.remove("count");
625 38 : p.insert("id".into(), chunk.join(","));
626 : // the final page fetch below re-surfaces the same peers' warnings
627 38 : let mut chunk_warnings = Vec::new();
628 38 : let fed = if crate::federation::active(&p) && !looped {
629 38 : crate::federation::fed_query(st, &tenant, headers, &ctx, &p, &mut chunk_warnings)
630 38 : .await?
631 : } else {
632 0 : Vec::new()
633 : };
634 38 : let docs = filter_entities_fed(st, &tenant, &p, &ctx, fed).await?;
635 38 : let matched: std::collections::HashSet<&str> = docs
636 38 : .iter()
637 172 : .filter_map(|d| d.get("id").and_then(Value::as_str))
638 38 : .collect();
639 38 : if let Some(emap) = map.get_mut("entityMap").and_then(Value::as_object_mut) {
640 180 : for id in chunk {
641 180 : let local_only = emap
642 180 : .get(id)
643 180 : .and_then(Value::as_array)
644 180 : .is_some_and(|a| a.len() == 1 && a[0] == "@none");
645 180 : if local_only && !matched.contains(id.as_str()) {
646 8 : emap.remove(id);
647 172 : }
648 : }
649 0 : }
650 38 : visited += chunk.len();
651 180 : for id in chunk {
652 180 : if !matched.contains(id.as_str()) {
653 8 : continue;
654 172 : }
655 172 : total += 1;
656 172 : if skipped < offset {
657 24 : skipped += 1;
658 148 : } else if page_ids.len() < limit {
659 76 : page_ids.push(id.clone());
660 80 : } else {
661 72 : more = true;
662 72 : }
663 : }
664 38 : if !count && page_ids.len() == limit {
665 : // page full — next exists if an extra match was seen or
666 : // unvisited candidates remain ("pages shall always be filled to
667 : // the maximum, as long as Entities are available")
668 14 : more = more || visited < ids.len();
669 14 : break;
670 24 : }
671 : }
672 38 : if count {
673 6 : more = total > offset + limit;
674 32 : }
675 38 : crate::entity_map::map_put(st, &tenant, map.clone()).await?;
676 : // fix the final fetch to exactly the page's survivors (5.5.14: an empty
677 : // set is fixed to nothing); one extra page-sized fetch keeps the whole
678 : // repr pipeline shared instead of forked
679 38 : params.insert(
680 38 : "id".into(),
681 38 : if page_ids.is_empty() {
682 0 : "urn:ngsi-ld:entitymap:empty".to_owned()
683 : } else {
684 38 : page_ids.join(",")
685 : },
686 : );
687 38 : params.remove("offset");
688 38 : params.remove("count");
689 38 : if limit == 0 {
690 0 : // count-only shape: the inner default limit is irrelevant against
691 0 : // the empty id sentinel
692 0 : params.remove("limit");
693 38 : }
694 : // The narrowing reaches the answer here; the map's own contents are
695 : // still the ones the query built, which is what P5's per-subject map is
696 : // for.
697 38 : let mut resp = query_entities_inner(st, ¶ms, headers, &filter).await?;
698 : // "The location of the EntityMap used in the query operation is
699 : // returned in the response" (6.4.3.2-2)
700 38 : if let Ok(v) = format!("/ngsi-ld/v1/entityMaps/{map_id}").parse() {
701 38 : resp.headers_mut().insert("NGSILD-EntityMap", v);
702 38 : }
703 38 : if count {
704 6 : if let Ok(v) = total.to_string().parse() {
705 6 : resp.headers_mut().insert("NGSILD-Results-Count", v);
706 6 : }
707 32 : }
708 : // 6.3.10 links from the original query — the inner call saw offset 0
709 : // over exactly one page, so it emitted none
710 76 : for (off, rel, cond) in [
711 38 : (offset + limit, "next", more && limit > 0),
712 38 : (offset.saturating_sub(limit.max(1)), "prev", offset > 0),
713 : ] {
714 76 : if !cond {
715 50 : continue;
716 26 : }
717 26 : let mut qp: Vec<String> = link_params
718 26 : .iter()
719 64 : .filter(|(k, _)| k.as_str() != "offset")
720 58 : .map(|(k, v)| format!("{k}={}", crate::paging::query_value(v)))
721 26 : .collect();
722 26 : qp.push(format!("offset={off}"));
723 26 : qp.sort(); // deterministic order — the suite string-compares links
724 26 : let ty = match accept {
725 0 : Accept::LdJson => ";type=\"application/ld+json\"",
726 26 : Accept::Json => ";type=\"application/json\"",
727 0 : Accept::GeoJson => ";type=\"application/geo+json\"",
728 : };
729 26 : if let Ok(v) = format!("</ngsi-ld/v1/entities?{}>; rel=\"{rel}\"{ty}", qp.join("&")).parse()
730 26 : {
731 26 : resp.headers_mut().append(axum::http::header::LINK, v);
732 26 : }
733 : }
734 38 : Ok(resp)
735 624 : }
736 :
737 618 : async fn query_entities_inner(
738 618 : st: &AppState,
739 618 : params: &HashMap<String, String>,
740 618 : headers: &HeaderMap,
741 618 : filter: &crate::policy::Filter,
742 618 : ) -> ApiResult<Response> {
743 618 : let tenant = tenant_from(headers)?;
744 618 : check_params(params, crate::negotiate::QUERY_PARAMS)?;
745 608 : let accept = parse_accept_geo(headers)?;
746 604 : let ctx = request_context(&st.loader, headers).await?;
747 :
748 : // 5.7.2.4 a-e: id/idPattern alone are NOT sufficient, and the attrs
749 : // list / q must include "at least one non-system Attribute" to qualify.
750 : // The judgement is about what the CLIENT asked for, so it reads the
751 : // request's own `q` — a policy condition is not the client's filter and
752 : // does not make a wide query narrow (ADR-0020).
753 544 : let has_filter = qualifies_non_wide(
754 544 : params,
755 544 : params.get("q").map(|q| parse_q(q)).transpose()?.as_ref(),
756 : );
757 : // Everything below reads the narrowed query: the store push-down, the
758 : // local re-check 5.7.2.4 runs over merged results, and the query the
759 : // request is forwarded with.
760 544 : let narrowed = filter.narrow_params(params)?;
761 542 : let params = &narrowed;
762 542 : let q_ast = params.get("q").map(|q| parse_q(q)).transpose()?;
763 : // 5.7.2.4 validation bullets (p.201), in the spec's own order.
764 542 : if params.get("type").map(String::as_str) == Some("*")
765 10 : && params.get("local").map(String::as_str) == Some("false")
766 : {
767 4 : return Err(NgsiError::BadRequestData(
768 4 : "type=* implies local and shall not be combined with local=false \
769 4 : (Table 6.4.3.2-1)"
770 4 : .into(),
771 4 : )
772 4 : .into());
773 538 : }
774 538 : if params.contains_key("geometryProperty") && accept != Accept::GeoJson {
775 4 : return Err(NgsiError::BadRequestData(
776 4 : "geometryProperty requires Accept: application/geo+json (5.7.2.4)".into(),
777 4 : )
778 4 : .into());
779 534 : }
780 : // "If the ordering parameter is present and the execution of the operation
781 : // is not limited to the local scope then BadRequestData" — reinforced by
782 : // 4.23.1: "Sort ordering is never applied to distributed operations."
783 : // The subject is the EXECUTION: a query nothing federates to runs locally
784 : // regardless of `local=true`, which is why this asks would_federate rather
785 : // than active (ETSI 019_19 orders without local).
786 534 : crate::paging::check_collation(params)?;
787 526 : if params.contains_key("orderBy")
788 44 : && crate::federation::would_federate(st, &tenant, &ctx, params, headers).await?
789 : {
790 4 : return Err(NgsiError::BadRequestData(
791 4 : "orderBy requires local scope — ordering is never applied to \
792 4 : distributed operations (5.7.2.4, 4.23.1)"
793 4 : .into(),
794 4 : )
795 4 : .into());
796 522 : }
797 522 : if !has_filter {
798 28 : return Err(NgsiError::BadRequestData(
799 28 : "query needs at least one of type, attrs, q, georel (5.7.2)".into(),
800 28 : )
801 28 : .into());
802 494 : }
803 :
804 494 : let mut repr = parse_repr(params, &ctx)?;
805 488 : crate::repr::narrow_repr(&mut repr, filter);
806 488 : let join = parse_join(params)?;
807 482 : check_linked_projection(&repr, &join)?;
808 : // 5.7.2.4: filter conditions using Linked Entity attributes need join,
809 : // and their hop depth may not exceed joinLevel ("too deep query")
810 478 : let link_depth = q_ast
811 478 : .as_ref()
812 478 : .map(antares_ql::QNode::max_link_depth)
813 478 : .unwrap_or(0);
814 478 : if link_depth > 0 {
815 12 : match &join {
816 12 : Some((mode, level)) if mode != "@none" => {
817 12 : if link_depth > *level {
818 8 : return Err(NgsiError::BadRequestData(format!(
819 8 : "linked attribute query depth {link_depth} exceeds joinLevel {level} \
820 8 : (5.7.2.4 — too deep query)"
821 8 : ))
822 8 : .into());
823 4 : }
824 : }
825 : _ => {
826 8 : return Err(NgsiError::BadRequestData(
827 8 : "q references Linked Entity attributes but join is not specified \
828 8 : (5.7.2.4 — too deep query)"
829 8 : .into(),
830 8 : )
831 8 : .into());
832 : }
833 : }
834 458 : }
835 : // 5.7.2.4: a syntactically invalid context source filter is 400.
836 462 : if let Some(csf) = params.get("csf") {
837 16 : parse_q(csf)?;
838 446 : }
839 : // 6.3.17: abnormal distributed-GET outcomes surface as NGSILD-Warning
840 450 : let mut warnings: Vec<String> = Vec::new();
841 450 : let looped = crate::federation::via_loop(
842 450 : headers,
843 450 : &crate::federation::alias_for(&st.host_alias, &tenant),
844 : );
845 450 : let fed = if crate::federation::active(params) && !looped {
846 428 : crate::federation::fed_query(st, &tenant, headers, &ctx, params, &mut warnings).await?
847 : } else {
848 22 : if crate::federation::active(params)
849 2 : && looped
850 2 : && crate::federation::would_federate(st, &tenant, &ctx, params, headers).await?
851 2 : {
852 2 : warnings.push(crate::federation::warning(
853 2 : 199,
854 2 : &crate::federation::alias_for(&st.host_alias, &tenant),
855 2 : "a registration loop has been detected",
856 2 : ));
857 20 : }
858 22 : Vec::new()
859 : };
860 : // Pushdown gates: pagination per page_pushdown_allowed (no federation
861 : // candidates, no idPattern, no orderBy). Projection additionally excludes
862 : // join (linked-entity walks read page docs) and GeoJSON output.
863 446 : let (p_offset, p_limit, _) = page_params(st, params)?;
864 442 : let push_page = page_pushdown_allowed(fed.is_empty(), params);
865 442 : let push_proj = join.is_none() && accept != Accept::GeoJson;
866 442 : let filtered = filter_entities_paged(
867 442 : st,
868 442 : &tenant,
869 442 : params,
870 442 : &ctx,
871 442 : fed,
872 442 : push_page.then_some((p_offset, p_limit)),
873 442 : push_proj.then_some(&repr),
874 442 : )
875 442 : .await?;
876 436 : let mut matches = filtered.docs;
877 436 : if let Some(spec) = params.get("orderBy") {
878 40 : order_entities(&mut matches, spec, params, &ctx)?;
879 396 : }
880 434 : let (page, count_hdr, links) = if filtered.paged {
881 32 : let total = filtered.total.unwrap_or(matches.len());
882 32 : paginate_pre(st, params, matches, "/ngsi-ld/v1/entities", total)?
883 : } else {
884 402 : paginate(st, params, matches, "/ngsi-ld/v1/entities")?
885 : };
886 :
887 434 : let mut payload: Vec<Value> = page
888 434 : .iter()
889 3314 : .filter_map(|doc| {
890 3314 : let shaped = apply(doc, &repr);
891 : // pick projections that match nothing drop the entity entirely
892 3314 : if repr.pick.is_some() && shaped.as_object().is_some_and(|o| o.is_empty()) {
893 0 : return None;
894 3314 : }
895 3314 : Some(compact_for(&repr, &shaped, &ctx))
896 3314 : })
897 434 : .collect();
898 434 : if let Some((mode, level)) = &join {
899 8 : let mut complete = true;
900 8 : let held = contained_by(params);
901 : // 4.5.23.1 bounds the WIDTH of the retrieval per request, so one
902 : // allowance is spent across the whole page: minting a fresh budget per
903 : // payload Entity multiplied it by the page size, and a page of
904 : // max_limit densely linked Entities bought MAX_JOIN_LOOKUPS lookups
905 : // each.
906 8 : let mut budget = MAX_JOIN_LOOKUPS;
907 8 : match mode.as_str() {
908 8 : "inline" => {
909 6 : for p in &mut payload {
910 : complete &=
911 4 : inline_join_beyond(st, &tenant, &ctx, &repr, p, *level, &held, &mut budget)
912 4 : .await;
913 : }
914 : }
915 2 : "flat" => {
916 2 : let mut linked = std::collections::BTreeMap::new();
917 2 : for doc in &page {
918 2 : complete &= collect_flat_beyond(
919 2 : st,
920 2 : &tenant,
921 2 : &repr,
922 2 : doc,
923 2 : *level,
924 2 : &mut linked,
925 2 : &held,
926 2 : &mut budget,
927 2 : )
928 2 : .await;
929 : }
930 2 : let page_ids: Vec<&str> = page.iter().filter_map(|d| d["id"].as_str()).collect();
931 2 : for (id, (ldoc, lrepr)) in linked {
932 2 : if !page_ids.contains(&id.as_str()) {
933 2 : payload.push(compact_for(&lrepr, &apply(&ldoc, &lrepr), &ctx));
934 2 : }
935 : }
936 : }
937 0 : _ => {}
938 : }
939 8 : if !complete {
940 2 : warnings.push(crate::federation::warning(
941 2 : 199,
942 2 : &crate::federation::alias_for(&st.host_alias, &tenant),
943 2 : "the linked entity retrieval was truncated",
944 2 : ));
945 6 : }
946 426 : }
947 434 : let mut resp = if accept == Accept::GeoJson {
948 12 : let fc = to_geojson_collection(payload, params.get("geometryProperty"));
949 12 : respond_prefer(StatusCode::OK, fc, &ctx, accept, &tenant, headers)
950 : } else {
951 422 : crate::negotiate::respond_list(StatusCode::OK, payload, &ctx, accept, &tenant)
952 : };
953 434 : attach_paging(&mut resp, count_hdr, &links);
954 434 : attach_warnings(&mut resp, &warnings);
955 : // 6.4.3.2: entityMap=true — the EntityMap for this query is (re)created;
956 : // the response carries NGSILD-EntityMap and 201 Created.
957 434 : if params.get("entityMap").map(String::as_str) == Some("true") {
958 28 : let map = build_query_map(st, &tenant, headers, &ctx, params, filter).await?;
959 28 : *resp.status_mut() = StatusCode::CREATED;
960 28 : if let Some(id) = map.get("id").and_then(Value::as_str) {
961 28 : if let Ok(v) = format!("/ngsi-ld/v1/entityMaps/{id}").parse() {
962 28 : resp.headers_mut().insert("NGSILD-EntityMap", v);
963 28 : }
964 0 : }
965 406 : }
966 434 : filter.mark_restricted(resp.headers_mut());
967 434 : Ok(resp)
968 618 : }
969 :
970 : /// 5.7.2.4 / 5.7.4.4 / 5.14.4.4 a-e: a query qualifies (is not "too wide")
971 : /// only with a type selector, an attrs list or q naming at least one
972 : /// non-system Attribute, a geoquery, or local scope.
973 906 : pub(crate) fn qualifies_non_wide(
974 906 : params: &HashMap<String, String>,
975 906 : q_ast: Option<&antares_ql::QNode>,
976 906 : ) -> bool {
977 906 : let attrs_qualify = params.get("attrs").is_some_and(|a| {
978 30 : a.split(',')
979 30 : .any(|n| antares_ql::is_non_system_attr(n.trim()))
980 30 : });
981 906 : let q_qualifies = q_ast.is_some_and(|ast| {
982 82 : ast.attribute_paths()
983 82 : .iter()
984 82 : .any(|h| antares_ql::is_non_system_attr(h))
985 82 : });
986 906 : params.contains_key("type")
987 70 : || attrs_qualify
988 62 : || q_qualifies
989 58 : || params.contains_key("georel")
990 54 : || params.get("local").map(String::as_str) == Some("true")
991 906 : }
992 :
993 : /// Same, with federated candidate docs merged in before filtering (4.3.6.7).
994 174 : pub async fn filter_entities_fed(
995 174 : st: &AppState,
996 174 : tenant: &TenantId,
997 174 : params: &HashMap<String, String>,
998 174 : ctx: &antares_jsonld::Context,
999 174 : fed: Vec<(bool, Value)>,
1000 174 : ) -> ApiResult<Vec<Value>> {
1001 : Ok(
1002 174 : filter_entities_paged(st, tenant, params, ctx, fed, None, None)
1003 174 : .await?
1004 : .docs,
1005 : )
1006 174 : }
1007 :
1008 : /// What the paged variant produced. `paged` = the store already applied
1009 : /// ORDER BY id + LIMIT/OFFSET (and `total` is the pre-LIMIT match count), so
1010 : /// the caller must NOT slice again.
1011 : pub struct Filtered {
1012 : pub docs: Vec<Value>,
1013 : pub paged: bool,
1014 : pub total: Option<usize>,
1015 : /// How many documents the store returned, before the evaluator below
1016 : /// dropped any. With a pushed page that is the size of the SQL page —
1017 : /// which is what a chunked walk has to step over, since `docs` undercounts
1018 : /// it whenever idPattern (invisible to the store filter) removed rows.
1019 : pub rows: usize,
1020 : }
1021 :
1022 : /// Whether the store may be asked to project members away. Only when its
1023 : /// answer is the final answer: 5.7.2.4 applies the query, geoquery, Scope
1024 : /// query and Attribute filters after remote parts have been aggregated, so
1025 : /// with federated candidates present those filters still run over documents
1026 : /// the store would have stripped; and 4.23 orders Entities by the value of
1027 : /// the ordering member, which a projection may have removed — leaving the
1028 : /// comparator nothing to compare and the client id order.
1029 434 : fn proj_pushdown_allowed(fed_is_empty: bool, params: &HashMap<String, String>) -> bool {
1030 434 : fed_is_empty && !params.contains_key("orderBy")
1031 434 : }
1032 :
1033 : /// Whether the store may be asked to cut the page (ORDER BY id +
1034 : /// LIMIT/OFFSET). Only when nothing outside it still narrows or reorders the
1035 : /// match set:
1036 : ///
1037 : /// * federated candidates are merged in afterwards, and 5.7.2.4 applies the
1038 : /// query, geoquery, Scope query and Attribute filters only after that
1039 : /// aggregation — so a SQL page would be cut from the wrong set;
1040 : /// * `idPattern` is not part of the store filter, so it drops rows the SQL
1041 : /// page already counted;
1042 : /// * `orderBy` orders the whole match set by the value of the ordering member
1043 : /// (4.23) before the page is cut, and that comparison order is the
1044 : /// evaluator's.
1045 : ///
1046 : /// `limit=0` — the count-only shape of 6.3.10, where `count=true` is
1047 : /// mandatory — IS pushed: the store returns no rows and counts the match set,
1048 : /// which is the same count the scan derives from a materialized one, without
1049 : /// materializing 100 million documents to throw them away.
1050 462 : fn page_pushdown_allowed(fed_is_empty: bool, params: &HashMap<String, String>) -> bool {
1051 462 : fed_is_empty && !params.contains_key("idPattern") && !params.contains_key("orderBy")
1052 462 : }
1053 :
1054 : /// The full filtering path (5.7.2). `page` = (offset, limit) to push into the
1055 : /// store — pass it ONLY when every filter the store cannot see is absent
1056 : /// (idPattern, federation, orderBy); the store still refuses unless its own
1057 : /// predicates compiled exactly. `proj` = the parsed representation, offered
1058 : /// for projection pushdown (pick/omit/attrs top-level heads) under the same
1059 : /// exactness gate.
1060 992 : pub async fn filter_entities_paged(
1061 992 : st: &AppState,
1062 992 : tenant: &TenantId,
1063 992 : params: &HashMap<String, String>,
1064 992 : ctx: &antares_jsonld::Context,
1065 992 : fed: Vec<(bool, Value)>,
1066 992 : page: Option<(usize, usize)>,
1067 992 : proj: Option<&crate::repr::Repr>,
1068 992 : ) -> ApiResult<Filtered> {
1069 : // a pushed page over local rows cannot be merged with federated
1070 : // candidates — refuse here so no caller can create that page
1071 992 : let page = if fed.is_empty() { page } else { None };
1072 : // and neither can a pushed projection: the store's answer is only the
1073 : // final answer when nothing downstream still needs the stripped members
1074 992 : let proj = proj.filter(|_| proj_pushdown_allowed(fed.is_empty(), params));
1075 992 : let ids: Option<Vec<&str>> = params.get("id").map(|s| s.split(',').collect());
1076 992 : if let Some(ids) = &ids {
1077 310 : for id in ids {
1078 310 : antares_model::EntityId::new(id)?;
1079 : }
1080 878 : }
1081 990 : let id_pattern = match params.get("idPattern") {
1082 56 : Some(p) => {
1083 224 : if ["**", "++", "*+", "+*"].iter().any(|q| p.contains(q)) {
1084 0 : return Err(NgsiError::BadRequestData(format!("invalid idPattern {p:?}")).into());
1085 56 : }
1086 : Some(
1087 56 : antares_ql::regex::compile(p)
1088 56 : .map_err(|_| NgsiError::BadRequestData(format!("invalid idPattern {p:?}")))?,
1089 : )
1090 : }
1091 934 : None => None,
1092 : };
1093 : // Entity Type Selection Language (4.17): `,`/`|` = OR, `(a;b)` = AND.
1094 : // Table 6.4.3.2-1: `"*"` selects every Entity Type, i.e. no type predicate
1095 : // at all. Expanding it as a term yields an IRI nothing matches, which is
1096 : // how `type=*` silently returned an empty array.
1097 986 : let type_sel: Option<Vec<Vec<String>>> = params.get("type").filter(|s| *s != "*").map(|s| {
1098 914 : s.split([',', '|'])
1099 922 : .map(|alt| {
1100 922 : alt.trim()
1101 922 : .trim_start_matches('(')
1102 922 : .trim_end_matches(')')
1103 922 : .split(';')
1104 922 : .map(|t| ctx.expand_key(t.trim()))
1105 922 : .collect()
1106 922 : })
1107 914 : .collect()
1108 914 : });
1109 986 : let attr_filter: Option<Vec<String>> = params
1110 986 : .get("attrs")
1111 986 : .map(|s| s.split(',').map(|t| ctx.expand_key(t.trim())).collect());
1112 986 : let q_ast = match params.get("q") {
1113 : // 4.9 expandValues: "attributes whose values should be expanded
1114 : // against the supplied @context using JSON-LD type coercion prior to
1115 : // executing the query" (EXAMPLE 12), less the Attributes jsonKeys
1116 : // declares uninterpretable as JSON-LD.
1117 82 : Some(q) => Some(antares_ql::eval::apply_expand_values(
1118 82 : parse_q(q)?,
1119 82 : antares_ql::eval::expansion_list(
1120 82 : params.get("expandValues").map(String::as_str),
1121 82 : params.get("jsonKeys").map(String::as_str),
1122 82 : )
1123 82 : .as_deref(),
1124 82 : ctx,
1125 : )),
1126 904 : None => None,
1127 : };
1128 986 : let scope_q = params.get("scopeQ");
1129 986 : let geo = antares_ql::geo::GeoQuery::from_params(params)?;
1130 :
1131 : // Hand the store what it can filter on. A backend that can push
1132 : // the predicate down (postgres/timescale) returns fewer rows — and says
1133 : // via `decided` whether it applied EVERY present predicate exactly, which
1134 : // is what licenses pagination/projection pushdown and lets the loop below
1135 : // skip re-deciding. A backend that cannot (memory/file) returns the
1136 : // snapshot and the loop stays the arbiter.
1137 982 : let expand = |t: &str| ctx.expand_key(t);
1138 982 : let geo_spec = geo.as_ref().and_then(|g| g.to_sql_spec(ctx));
1139 : // A geo query whose spec declined to compile (non-default geoproperty) is
1140 : // INVISIBLE to the store — the store would truthfully claim `decided`
1141 : // about what it saw, projection would strip the very member the evaluator
1142 : // still needs, and a pushed page would page over the wrong set. Forfeit
1143 : // every pushdown up front and mask `decided` after.
1144 982 : let geo_uncompiled = geo.is_some() && geo_spec.is_none();
1145 982 : let page = if geo_uncompiled { None } else { page };
1146 982 : let proj = proj.filter(|_| !geo_uncompiled);
1147 : // pick (or attrs) heads to keep / whole-attr omit heads to drop; core
1148 : // members are never SQL-dropped (only `://` IRIs qualify) — repr::apply
1149 : // stays the decider for those.
1150 982 : let keep_attrs: Option<Vec<String>> = proj.and_then(|r| {
1151 334 : r.pick
1152 334 : .as_ref()
1153 334 : .map(|nodes| nodes.iter().map(|n| n.iri.clone()).collect())
1154 334 : .or_else(|| r.attrs.clone())
1155 334 : });
1156 982 : let drop_attrs: Option<Vec<String>> = proj
1157 982 : .and_then(|r| {
1158 334 : r.omit.as_ref().map(|nodes| {
1159 0 : nodes
1160 0 : .iter()
1161 0 : .filter(|n| n.children.is_none() && n.iri.contains("://"))
1162 0 : .map(|n| n.iri.clone())
1163 0 : .collect::<Vec<_>>()
1164 0 : })
1165 334 : })
1166 982 : .filter(|v| !v.is_empty());
1167 : // 5.7.2.4 split entities (p.202): the filters (q, geoquery, Scope query,
1168 : // Attributes) apply only AFTER remote parts and local information have
1169 : // been aggregated — so with federated candidates present the store must
1170 : // not drop (or pre-project) the LOCAL half of a split entity. The
1171 : // post-merge loop below applies them instead (`decided` is already
1172 : // false whenever `fed` is non-empty).
1173 982 : let split_agg = crate::federation::split_entities(params) && !fed.is_empty();
1174 982 : let outcome = st
1175 982 : .store
1176 982 : .query_entities(
1177 982 : tenant,
1178 : &antares_store::filter::EntityFilter {
1179 982 : ids: ids.as_deref(),
1180 : // 5.2.33: id takes precedence over idPattern, so the literal
1181 : // narrows only when no id selector was given
1182 982 : id_literal: if ids.is_none() {
1183 870 : params
1184 870 : .get("idPattern")
1185 870 : .and_then(|p| antares_store::filter::id_pattern_literal(p))
1186 : } else {
1187 112 : None
1188 : },
1189 982 : types: type_sel.as_deref(),
1190 982 : attrs: if split_agg {
1191 6 : None
1192 : } else {
1193 976 : attr_filter.as_deref()
1194 : },
1195 982 : q: if split_agg { None } else { q_ast.as_ref() },
1196 982 : scope_q: if split_agg {
1197 6 : None
1198 : } else {
1199 976 : scope_q.map(String::as_str)
1200 : },
1201 982 : geo: if split_agg { None } else { geo_spec.as_ref() },
1202 982 : expand: &expand,
1203 982 : page: page.map(|(offset, limit)| antares_store::filter::Page {
1204 708 : offset: offset as i64,
1205 708 : limit: limit as i64,
1206 708 : count: params.get("count").map(String::as_str) == Some("true"),
1207 708 : }),
1208 982 : keep_attrs: keep_attrs.as_deref(),
1209 982 : drop_attrs: drop_attrs.as_deref(),
1210 : },
1211 : )
1212 982 : .await?;
1213 982 : let decided = outcome.decided && fed.is_empty() && !geo_uncompiled;
1214 982 : let paged = outcome.paged && fed.is_empty();
1215 982 : let total = outcome.total.map(|t| t as usize);
1216 982 : let rows = outcome.rows.len();
1217 982 : let all = crate::federation::merge_candidates(outcome.rows, fed);
1218 : // the id list is client-sized (a POST query body carries an array with no
1219 : // count of its own) and the candidate set is store-sized, so the two are
1220 : // never multiplied together
1221 982 : let id_set: Option<std::collections::HashSet<&str>> =
1222 982 : ids.as_ref().map(|v| v.iter().copied().collect());
1223 982 : let mut out = Vec::new();
1224 19178 : for doc in all {
1225 19178 : let id = doc["id"].as_str().unwrap_or("");
1226 19178 : if let Some(ids) = &id_set {
1227 414 : if !decided && !ids.contains(id) {
1228 142 : continue;
1229 272 : }
1230 18764 : }
1231 : // 5.2.33: "id takes precedence over idPattern" — the pattern only
1232 : // filters when no id selector was given.
1233 19036 : if ids.is_none() {
1234 18764 : if let Some(re) = &id_pattern {
1235 : // idPattern is invisible to the store — applied even when
1236 : // decided
1237 2578 : if !re.is_match(id) {
1238 1296 : continue;
1239 1282 : }
1240 16186 : }
1241 272 : }
1242 17740 : if !decided {
1243 16300 : if let Some(sel) = &type_sel {
1244 16290 : let etypes: Vec<&str> = doc["type"]
1245 16290 : .as_array()
1246 16290 : .map(|a| a.iter().filter_map(Value::as_str).collect())
1247 16290 : .unwrap_or_default();
1248 16290 : let matched = sel
1249 16290 : .iter()
1250 16290 : .any(|and_group| and_group.iter().all(|w| etypes.contains(&w.as_str())));
1251 16290 : if !matched {
1252 3766 : continue;
1253 12524 : }
1254 10 : }
1255 12534 : if let Some(attrs) = &attr_filter {
1256 2 : if !attrs.iter().any(|a| doc.get(a).is_some()) {
1257 0 : continue;
1258 2 : }
1259 12532 : }
1260 12534 : if let Some(ast) = &q_ast {
1261 : // 4.9 linked-entity subqueries (attr{path}) resolve through
1262 : // the local store, same tenant.
1263 114 : if !crate::notify::linked_eval(st, tenant, |l| eval_q(ast, &doc, ctx, l)).await {
1264 44 : continue;
1265 70 : }
1266 12420 : }
1267 12490 : if let Some(sq) = scope_q {
1268 642 : if !crate::scope_matches(sq, &doc) {
1269 28 : continue;
1270 614 : }
1271 11848 : }
1272 12462 : if let Some(g) = &geo {
1273 4 : if !g.matches(&doc, ctx) {
1274 0 : continue;
1275 4 : }
1276 12458 : }
1277 1440 : }
1278 13902 : out.push(doc);
1279 : }
1280 982 : Ok(Filtered {
1281 982 : docs: out,
1282 982 : paged,
1283 982 : total,
1284 982 : rows,
1285 982 : })
1286 992 : }
1287 :
1288 : // ---------- DELETE /entities/{id} (5.6.6) ----------
1289 :
1290 2250 : pub async fn delete_entity(
1291 2250 : State(st): State<AppState>,
1292 2250 : Path(id): Path<String>,
1293 2250 : CleanParams(params): CleanParams,
1294 2250 : headers: HeaderMap,
1295 2250 : ) -> Response {
1296 2250 : let go = async {
1297 2250 : let tenant = tenant_from(&headers)?;
1298 2250 : antares_model::EntityId::new(&id)?;
1299 2246 : check_params(¶ms, &["local", "type"])?;
1300 : // 5.5.7: `type` is a term, so the request's own @context expands it.
1301 : // Under the core context the same word names a different type than
1302 : // the client meant, and the delete then removes an Entity the
1303 : // client's selector excluded.
1304 2246 : let ctx = request_context(&st.loader, &headers).await?;
1305 2246 : gate!(st, &tenant, &headers, "5.6.6", ids: &[&id]).await?;
1306 : // 4.17/5.6.6.4: the type selector gates the target — a registration
1307 : // for a different type must not receive the forwarded delete.
1308 2240 : let spec = crate::registry::CsrSpec {
1309 2240 : ids: Some(vec![id.clone()]),
1310 2240 : types: params
1311 2240 : .get("type")
1312 2240 : .map(|s| s.split(',').map(|t| ctx.expand_key(t.trim())).collect()),
1313 2240 : ..Default::default()
1314 : };
1315 2240 : let regs = match crate::federation::write_plan(&st, &tenant, &spec, &ctx, ¶ms, &headers)
1316 2240 : .await?
1317 : {
1318 12 : crate::federation::WritePlan::Answered(r) => return Ok(*r),
1319 2228 : crate::federation::WritePlan::Forward(regs) => regs,
1320 : };
1321 : // 5.6.6.4: the ?type selector narrows the target — an entity of a
1322 : // non-matching type is "not known" for this delete. The selector is
1323 : // tested inside the delete, under the row lock: read first and delete
1324 : // after, and the document that answered the test is not necessarily
1325 : // the one the delete removes.
1326 2228 : let keep = |d: &Value| crate::negotiate::matches_type_param(d, ¶ms, &ctx);
1327 2228 : if !regs.is_empty() {
1328 24 : let proxy_match = regs.iter().any(|r| r.is_proxy());
1329 24 : let mut parts = Vec::new();
1330 24 : if st.store.delete_entity_if(&tenant, &id, &keep).await? {
1331 6 : mirror_delete_entity(&st, &tenant, &id).await;
1332 6 : parts.push(crate::federation::Part {
1333 6 : status: 204,
1334 6 : detail: "deleted locally".into(),
1335 6 : });
1336 18 : } else if !proxy_match {
1337 2 : // nothing local to delete, and no proxy that owns it: the
1338 2 : // local half of this operation is the 404
1339 2 : parts.push(crate::federation::Part {
1340 2 : status: 404,
1341 2 : detail: format!("entity {id} not found locally"),
1342 2 : });
1343 16 : }
1344 24 : let ctx_url = crate::federation::ctx_link_url(&headers, &ctx.source);
1345 24 : let seg = crate::federation::path_segment(&id);
1346 24 : let fwd_q = type_selector_query(¶ms);
1347 24 : for reg in ®s {
1348 : // 5.6.6.4: proxy modes not supporting Delete Entity are an
1349 : // error of type Conflict; inclusive ones are not forwarded.
1350 24 : if !reg.supports("deleteEntity") {
1351 4 : if reg.is_proxy() {
1352 0 : parts.push(crate::federation::conflict_part("deleteEntity"));
1353 4 : }
1354 4 : continue;
1355 20 : }
1356 20 : parts.push(
1357 20 : crate::federation::forward_part(
1358 20 : &st,
1359 20 : reqwest::Method::DELETE,
1360 20 : format!("{}/ngsi-ld/v1/entities/{seg}", reg.endpoint),
1361 20 : &fwd_q,
1362 20 : &headers,
1363 20 : &tenant,
1364 20 : reg,
1365 20 : &ctx_url,
1366 20 : None,
1367 20 : )
1368 20 : .await,
1369 : );
1370 : }
1371 24 : return Ok(crate::federation::combine(
1372 24 : parts,
1373 24 : no_content(&tenant),
1374 24 : &tenant,
1375 24 : ));
1376 2204 : }
1377 2204 : if st.store.delete_entity_if(&tenant, &id, &keep).await? {
1378 442 : mirror_delete_entity(&st, &tenant, &id).await;
1379 442 : Ok::<_, ApiError>(no_content(&tenant))
1380 : } else {
1381 1762 : Err(NgsiError::ResourceNotFound(format!("entity {id} not found")).into())
1382 : }
1383 2250 : };
1384 2250 : go.await.unwrap_or_else(|e| e.into_response())
1385 2250 : }
1386 :
1387 : // ---------- DELETE /entities/ — Purge (5.6.21) ----------
1388 :
1389 : /// 5.6.21.3 Input data of a Purge: the Entity type selector, the identifier
1390 : /// list and id pattern, the restrictive and exclusionary Attribute-name
1391 : /// lists, the NGSI-LD Query, the GeoQuery, the Scope query and the context
1392 : /// source filter. 5.6.21.4 forwards "matching input data ... to the
1393 : /// Registration endpoint", so every one of them travels: a forward that
1394 : /// carries fewer restrictions than the client issued makes the peer execute
1395 : /// a strictly wider purge than the one requested. `local` is absent by
1396 : /// design — it selects local scope, which is what stops the forward
1397 : /// happening at all (5.5.13).
1398 : const PURGE_FORWARD_PARAMS: &[&str] = &[
1399 : "type",
1400 : "id",
1401 : "idPattern",
1402 : "attrs",
1403 : "q",
1404 : "georel",
1405 : "geometry",
1406 : "coordinates",
1407 : "geoproperty",
1408 : "scopeQ",
1409 : "csf",
1410 : "keep",
1411 : "drop",
1412 : ];
1413 :
1414 12 : fn forwarded_purge_query(params: &HashMap<String, String>) -> Vec<(String, String)> {
1415 12 : PURGE_FORWARD_PARAMS
1416 12 : .iter()
1417 156 : .filter_map(|k| params.get(*k).map(|v| ((*k).to_owned(), v.clone())))
1418 12 : .collect()
1419 12 : }
1420 :
1421 : /// How many matched Entities one round of a Purge fetches and applies. The
1422 : /// match set of 5.6.21.4 is deliberately unbounded — `DELETE /entities?type=T`
1423 : /// matches every Entity of that type — so it is walked page by page and
1424 : /// applied in batches rather than materialized whole.
1425 : const PURGE_CHUNK: usize = 500;
1426 :
1427 : /// Where the next chunk of a walked match set starts, or None when the set is
1428 : /// exhausted. `rows` = documents the store returned for this chunk (not the
1429 : /// subset that survived the evaluator: idPattern is applied after the store,
1430 : /// so a full chunk can arrive narrowed or even empty and the walk must still
1431 : /// go on — 5.6.21.4 deletes ALL matched Entities, not the first page of
1432 : /// them). `left_the_set` = how many of those rows no longer match the query
1433 : /// afterwards: a Purge deletes or prunes them, a read leaves every row where
1434 : /// it was. The rows still in the set have to be stepped over, or the walk
1435 : /// re-reads them forever.
1436 : ///
1437 : /// Termination: every round either removes rows from the match set or
1438 : /// advances the offset by a full chunk, and both are bounded by the number of
1439 : /// stored Entities.
1440 394 : pub(crate) fn next_scan_offset(
1441 394 : offset: usize,
1442 394 : rows: usize,
1443 394 : left_the_set: usize,
1444 394 : paged: bool,
1445 394 : chunk: usize,
1446 394 : ) -> Option<usize> {
1447 : // an unpaged answer IS the whole match set, which this round just applied
1448 394 : if !paged || rows < chunk {
1449 362 : return None;
1450 32 : }
1451 32 : Some(offset + rows.saturating_sub(left_the_set))
1452 394 : }
1453 :
1454 : /// 5.6.21.4 "And thereafter": with no Attribute-name list, delete every
1455 : /// matched Entity found locally; with a restrictive list, delete those
1456 : /// Attributes from them; with an exclusionary list, delete all but those.
1457 80 : async fn purge_locally(
1458 80 : st: &AppState,
1459 80 : tenant: &TenantId,
1460 80 : params: &HashMap<String, String>,
1461 80 : ctx: &antares_jsonld::Context,
1462 80 : keep: &Option<Vec<String>>,
1463 80 : drop: &Option<Vec<String>>,
1464 80 : ) -> ApiResult<()> {
1465 80 : let prune = keep.is_some() || drop.is_some();
1466 80 : let mut offset = 0usize;
1467 : loop {
1468 84 : let batch = filter_entities_paged(
1469 84 : st,
1470 84 : tenant,
1471 84 : params,
1472 84 : ctx,
1473 84 : Vec::new(),
1474 84 : Some((offset, PURGE_CHUNK)),
1475 84 : None,
1476 84 : )
1477 84 : .await?;
1478 84 : let rows = batch.rows;
1479 84 : let ids: Vec<String> = batch
1480 84 : .docs
1481 84 : .iter()
1482 9300 : .filter_map(|d| d["id"].as_str().map(str::to_owned))
1483 84 : .collect();
1484 84 : let left_the_set = if ids.is_empty() {
1485 62 : 0
1486 22 : } else if prune {
1487 4 : let mut changed = 0usize;
1488 4 : st.store
1489 4028 : .batch_mutate(tenant, &ids, |_, doc| {
1490 4028 : let target = antares_store::stored_object(doc)?;
1491 4028 : let attrs: Vec<String> =
1492 16112 : target.keys().filter(|k| !is_meta(k)).cloned().collect();
1493 4028 : let before = target.len();
1494 8056 : for a in attrs {
1495 8056 : let purge = match (keep, drop) {
1496 8056 : (Some(keep), _) => !keep.contains(&a),
1497 0 : (_, Some(drop)) => drop.contains(&a),
1498 0 : _ => true,
1499 : };
1500 8056 : if purge {
1501 4028 : target.remove(&a);
1502 4028 : }
1503 : }
1504 4028 : if target.len() != before {
1505 4028 : changed += 1;
1506 4028 : }
1507 4028 : Ok::<(), NgsiError>(())
1508 4028 : })
1509 4 : .await?;
1510 : // A prune keeps the Entity, but it may have removed the very
1511 : // Attribute the query matched on (`attrs=speed&drop=speed`), which
1512 : // takes it out of the match set and shifts the rest down. Whether
1513 : // it did is only observable by reading the window again, so a round
1514 : // that changed something re-reads it; the re-read finds those
1515 : // Entities unchanged and steps over them.
1516 4 : if changed > 0 {
1517 4 : rows
1518 : } else {
1519 0 : 0
1520 : }
1521 : } else {
1522 18 : let mut gone = 0usize;
1523 5272 : for (id, deleted) in ids.iter().zip(st.store.batch_delete(tenant, &ids).await?) {
1524 5272 : if deleted {
1525 5272 : gone += 1;
1526 5272 : mirror_delete_entity(st, tenant, id).await;
1527 0 : }
1528 : }
1529 18 : gone
1530 : };
1531 84 : match next_scan_offset(offset, rows, left_the_set, batch.paged, PURGE_CHUNK) {
1532 4 : Some(next) => offset = next,
1533 80 : None => return Ok(()),
1534 : }
1535 : }
1536 80 : }
1537 :
1538 : /// 5.6.21 Purge Entities: delete (or keep=/drop=-prune) all entities matched
1539 : /// by the query; output data is none — 204 (5.6.21.5). Too-wide queries,
1540 : /// Linked Entity paths and invalid id/q/geo/csf are BadRequestData
1541 : /// (5.6.21.4); matched registrations forward only with purgeEntity support.
1542 134 : pub async fn purge_entities(
1543 134 : State(st): State<AppState>,
1544 134 : CleanParams(params): CleanParams,
1545 134 : headers: HeaderMap,
1546 134 : ) -> Response {
1547 134 : match purge_inner(&st, ¶ms, &headers).await {
1548 68 : Ok(r) => r,
1549 66 : Err(e) => e.into_response(),
1550 : }
1551 134 : }
1552 :
1553 148 : async fn purge_inner(
1554 148 : st: &AppState,
1555 148 : params: &HashMap<String, String>,
1556 148 : headers: &HeaderMap,
1557 148 : ) -> ApiResult<Response> {
1558 148 : let tenant = tenant_from(headers)?;
1559 148 : check_params(
1560 148 : params,
1561 148 : &[
1562 148 : "id",
1563 148 : "idPattern",
1564 148 : "type",
1565 148 : "attrs",
1566 148 : "q",
1567 148 : "georel",
1568 148 : "geometry",
1569 148 : "coordinates",
1570 148 : "geoproperty",
1571 148 : "scopeQ",
1572 148 : "csf",
1573 148 : "keep",
1574 148 : "drop",
1575 148 : "local",
1576 148 : ],
1577 4 : )?;
1578 144 : let ctx = request_context(&st.loader, headers).await?;
1579 116 : gate!(st, &tenant, headers, "5.6.21", scope_q: params.get("scopeQ").map(String::as_str))
1580 116 : .await?;
1581 : // 5.6.21.4: exactly five qualifying conditions —
1582 : // a) selector of Entity Types
1583 : // b) list of Attribute names, including at least one non-system Attribute
1584 : // c) NGSI-LD Query, including at least one non-system Attribute
1585 : // d) NGSI-LD GeoQuery
1586 : // e) local scope (5.5.13)
1587 : // "If none of the above is provided, then an error of type BadRequestData
1588 : // shall be raised (too wide query)."
1589 : //
1590 : // id/idPattern are legal input data (5.6.21.3) and DO filter, but they are
1591 : // never sufficient on their own: "it is not possible to purge a set of
1592 : // entities by only specifying desired Entity identifiers". Listing them
1593 : // here is how `DELETE /entities?idPattern=.*` became a tenant wipe.
1594 114 : let attrs_qualify = params.get("attrs").is_some_and(|a| {
1595 12 : a.split(',')
1596 12 : .any(|n| antares_ql::is_non_system_attr(n.trim()))
1597 12 : });
1598 114 : let q_ast = params.get("q").map(|q| parse_q(q)).transpose()?;
1599 114 : let q_qualifies = q_ast.as_ref().is_some_and(|ast| {
1600 12 : ast.attribute_paths()
1601 12 : .iter()
1602 12 : .any(|h| antares_ql::is_non_system_attr(h))
1603 12 : });
1604 : // 5.6.21.4: Linked Entity retrieval in the projection attributes, or
1605 : // Linked Entity attributes in the filter conditions → BadRequestData.
1606 114 : if q_ast
1607 114 : .as_ref()
1608 114 : .is_some_and(antares_ql::QNode::has_linked_paths)
1609 : {
1610 4 : return Err(NgsiError::BadRequestData(
1611 4 : "purge q must not reference Linked Entity attributes (5.6.21.4)".into(),
1612 4 : )
1613 4 : .into());
1614 110 : }
1615 110 : if params.get("attrs").is_some_and(|a| a.contains('{')) {
1616 4 : return Err(NgsiError::BadRequestData(
1617 4 : "purge attrs must not use Linked Entity retrieval (5.6.21.4)".into(),
1618 4 : )
1619 4 : .into());
1620 106 : }
1621 : // 5.6.21.4: a syntactically invalid context source filter is
1622 : // BadRequestData.
1623 106 : if let Some(csf) = params.get("csf") {
1624 8 : parse_q(csf)?;
1625 98 : }
1626 102 : let has_filter = params.contains_key("type")
1627 68 : || attrs_qualify
1628 64 : || q_qualifies
1629 60 : || params.contains_key("georel")
1630 60 : || params.get("local").map(String::as_str) == Some("true");
1631 102 : if !has_filter {
1632 20 : return Err(NgsiError::BadRequestData(
1633 20 : "purge needs at least one of: type, attrs or q naming a non-system \
1634 20 : Attribute, georel, or local=true (5.6.21.4 — too wide query)"
1635 20 : .into(),
1636 20 : )
1637 20 : .into());
1638 82 : }
1639 82 : if params.contains_key("keep") && params.contains_key("drop") {
1640 0 : return Err(NgsiError::BadRequestData(
1641 0 : "keep and drop are mutually exclusive (5.6.21)".into(),
1642 0 : )
1643 0 : .into());
1644 82 : }
1645 82 : let keep: Option<Vec<String>> = params
1646 82 : .get("keep")
1647 82 : .map(|s| s.split(',').map(|t| ctx.expand_key(t.trim())).collect());
1648 82 : let drop: Option<Vec<String>> = params
1649 82 : .get("drop")
1650 82 : .map(|s| s.split(',').map(|t| ctx.expand_key(t.trim())).collect());
1651 : // distributed purge (5.6.21 / 6.4.3.3). Matching and the 6.3.17/6.3.18
1652 : // loop check come first: 508 Loop Detected is an error status, so the
1653 : // request it answers must not have deleted a page of Entities already.
1654 82 : let spec = crate::registry::CsrSpec {
1655 82 : types: params
1656 82 : .get("type")
1657 82 : .map(|s| s.split(',').map(|t| ctx.expand_key(t.trim())).collect()),
1658 82 : ids: params
1659 82 : .get("id")
1660 82 : .map(|s| s.split(',').map(str::to_owned).collect()),
1661 : // 5.12: the purge's idPattern is part of the Entity specification too
1662 82 : id_pattern: params.get("idPattern").cloned(),
1663 82 : csf: params.get("csf").and_then(|c| antares_ql::parse_q(c).ok()),
1664 82 : ..Default::default()
1665 : };
1666 80 : let regs =
1667 82 : match crate::federation::write_plan(st, &tenant, &spec, &ctx, params, headers).await? {
1668 2 : crate::federation::WritePlan::Answered(r) => return Ok(*r),
1669 80 : crate::federation::WritePlan::Forward(regs) => regs,
1670 : };
1671 80 : purge_locally(st, &tenant, params, &ctx, &keep, &drop).await?;
1672 80 : if !regs.is_empty() {
1673 4 : let mut parts = vec![crate::federation::Part {
1674 4 : status: 204,
1675 4 : detail: "purged locally".into(),
1676 4 : }];
1677 4 : let ctx_url = crate::federation::ctx_link_url(headers, &ctx.source);
1678 4 : let query = forwarded_purge_query(params);
1679 4 : for reg in ®s {
1680 : // 5.6.21.4: matching input data is forwarded only when the
1681 : // registration supports Purge Entity; an unsupported matched
1682 : // registration — any mode — is an error of type Conflict
1683 : // (partial success when other parts succeeded).
1684 4 : if !reg.supports("purgeEntity") {
1685 2 : parts.push(crate::federation::conflict_part("purgeEntity"));
1686 2 : continue;
1687 2 : }
1688 2 : parts.push(
1689 2 : crate::federation::forward_part(
1690 2 : st,
1691 2 : reqwest::Method::DELETE,
1692 2 : format!("{}/ngsi-ld/v1/entities", reg.endpoint),
1693 2 : &query,
1694 2 : headers,
1695 2 : &tenant,
1696 2 : reg,
1697 2 : &ctx_url,
1698 2 : None,
1699 2 : )
1700 2 : .await,
1701 : );
1702 : }
1703 4 : return Ok(crate::federation::combine(
1704 4 : parts,
1705 4 : no_content(&tenant),
1706 4 : &tenant,
1707 4 : ));
1708 76 : }
1709 76 : Ok(no_content(&tenant))
1710 148 : }
1711 :
1712 : // ---------- PATCH /entities/{id} — Merge (5.6.17 / 5.5.12) ----------
1713 :
1714 42 : pub async fn merge_entity(
1715 42 : State(st): State<AppState>,
1716 42 : Path(id): Path<String>,
1717 42 : CleanParams(params): CleanParams,
1718 42 : headers: HeaderMap,
1719 42 : body: Bytes,
1720 42 : ) -> Response {
1721 42 : match merge_entity_inner(&st, &id, ¶ms, &headers, &body).await {
1722 30 : Ok(r) => r,
1723 12 : Err(e) => e.into_response(),
1724 : }
1725 42 : }
1726 :
1727 42 : async fn merge_entity_inner(
1728 42 : st: &AppState,
1729 42 : id: &str,
1730 42 : params: &HashMap<String, String>,
1731 42 : headers: &HeaderMap,
1732 42 : body: &[u8],
1733 42 : ) -> ApiResult<Response> {
1734 42 : let tenant = tenant_from(headers)?;
1735 42 : antares_model::EntityId::new(id)?;
1736 42 : check_params(
1737 42 : params,
1738 42 : &["options", "format", "observedAt", "lang", "local", "type"],
1739 0 : )?;
1740 42 : let parsed = parse_body(&st.loader, headers, body, BodyKind::MergePatch).await?;
1741 42 : let obj = parsed.object(NgsiError::BadRequestData(
1742 42 : "fragment must be a JSON object".into(),
1743 42 : ))?;
1744 42 : if let Some(bid) = obj.get("id").and_then(Value::as_str) {
1745 0 : if bid != id {
1746 0 : return Err(NgsiError::BadRequestData("fragment id mismatch".into()).into());
1747 0 : }
1748 42 : }
1749 42 : let mut fragment = expand_entity(
1750 42 : obj,
1751 42 : &parsed.ctx,
1752 42 : ExpandOpts {
1753 42 : fragment: true,
1754 42 : allow_null: true,
1755 42 : merge: true,
1756 42 : temporal: false,
1757 42 : ..Default::default()
1758 42 : },
1759 0 : )?;
1760 42 : gate!(st, &tenant, headers, "5.6.17", ids: &[id], body: Some(&fragment)).await?;
1761 : // 5.6.17.3: a common observedAt timestamp to use across merged
1762 : // Attributes, and a common language tag for merged LanguageMaps.
1763 42 : let observed_at = params.get("observedAt").map(String::as_str);
1764 42 : if let Some(t) = observed_at {
1765 8 : if !antares_jsonld::parse_datetime(t) {
1766 4 : return Err(
1767 4 : NgsiError::BadRequestData("observedAt must be a DateTime (4.8)".into()).into(),
1768 4 : );
1769 4 : }
1770 34 : }
1771 38 : let lang = params.get("lang").map(String::as_str);
1772 38 : apply_common_observed_at(&mut fragment, observed_at);
1773 38 : let ts = now_iso();
1774 :
1775 38 : let spec = crate::registry::CsrSpec {
1776 38 : ids: Some(vec![id.to_owned()]),
1777 38 : ..Default::default()
1778 38 : };
1779 38 : let regs = match crate::federation::write_plan(st, &tenant, &spec, &parsed.ctx, params, headers)
1780 38 : .await?
1781 : {
1782 0 : crate::federation::WritePlan::Answered(r) => return Ok(*r),
1783 38 : crate::federation::WritePlan::Forward(regs) => regs,
1784 : };
1785 38 : if !regs.is_empty() {
1786 6 : let proxies: Vec<&crate::federation::FedReg> =
1787 6 : regs.iter().filter(|r| r.is_proxy()).collect();
1788 6 : let mut parts = Vec::new();
1789 6 : let (rest, has_attrs) = crate::federation::strip_proxied(obj, &proxies, &parsed.ctx);
1790 : // 5.6.17.4: the target is "an existing Entity whose id (URI), and
1791 : // where specified type, is equivalent held locally" — the ?type
1792 : // selector narrows it on this path exactly as on the local-only one.
1793 6 : let local_exists = st
1794 6 : .store
1795 6 : .get(&tenant, Kind::Entity, id)
1796 6 : .await?
1797 6 : .is_some_and(|d| crate::negotiate::matches_type_param(&d, params, &parsed.ctx));
1798 6 : if (local_exists || proxies.is_empty()) && has_attrs {
1799 2 : let mut local_frag = expand_entity(
1800 2 : &rest,
1801 2 : &parsed.ctx,
1802 2 : ExpandOpts {
1803 2 : fragment: true,
1804 2 : allow_null: true,
1805 2 : merge: true,
1806 2 : temporal: false,
1807 2 : ..Default::default()
1808 2 : },
1809 0 : )?;
1810 2 : apply_common_observed_at(&mut local_frag, observed_at);
1811 2 : let res = st
1812 2 : .store
1813 2 : .mutate(&tenant, Kind::Entity, id, |doc| {
1814 2 : if !crate::negotiate::matches_type_param(doc, params, &parsed.ctx) {
1815 0 : return Err(NgsiError::ResourceNotFound(format!(
1816 0 : "entity {id} does not match the type selector"
1817 0 : )));
1818 2 : }
1819 2 : let mut frag = local_frag.clone();
1820 2 : apply_common_lang(doc, &mut frag, lang);
1821 2 : merge_into(doc, &frag, &ts);
1822 2 : Ok::<(), NgsiError>(())
1823 2 : })
1824 2 : .await?;
1825 2 : parts.push(match res {
1826 2 : Some(Ok(())) => crate::federation::Part {
1827 2 : status: 204,
1828 2 : detail: "merged locally".into(),
1829 2 : },
1830 0 : _ => crate::federation::Part {
1831 0 : status: 404,
1832 0 : detail: format!("entity {id} not found locally"),
1833 0 : },
1834 : });
1835 4 : }
1836 6 : let ctx_url = crate::federation::ctx_link_url(headers, &parsed.ctx.source);
1837 6 : let seg = crate::federation::path_segment(id);
1838 6 : let fwd_q = type_selector_query(params);
1839 6 : for reg in ®s {
1840 : // 5.6.17.4: proxy modes without Merge Entity support are an
1841 : // error of type Conflict; inclusive ones are not forwarded.
1842 6 : if !reg.supports("mergeEntity") {
1843 4 : if reg.is_proxy() {
1844 2 : parts.push(crate::federation::conflict_part("mergeEntity"));
1845 2 : }
1846 4 : continue;
1847 2 : }
1848 2 : let Some(frag) = crate::federation::reduce_to_scope(obj, reg, &parsed.ctx) else {
1849 0 : continue;
1850 : };
1851 2 : parts.push(
1852 2 : crate::federation::forward_part(
1853 2 : st,
1854 2 : reqwest::Method::PATCH,
1855 2 : format!("{}/ngsi-ld/v1/entities/{seg}", reg.endpoint),
1856 2 : &fwd_q,
1857 2 : headers,
1858 2 : &tenant,
1859 2 : reg,
1860 2 : &ctx_url,
1861 2 : Some(frag),
1862 2 : )
1863 2 : .await,
1864 : );
1865 : }
1866 6 : return Ok(crate::federation::combine(
1867 6 : parts,
1868 6 : no_content(&tenant),
1869 6 : &tenant,
1870 6 : ));
1871 32 : }
1872 :
1873 32 : let res = st
1874 32 : .store
1875 32 : .mutate(&tenant, Kind::Entity, id, |doc| {
1876 : // 5.6.17.4: the ?type selector narrows the merge target
1877 24 : if !crate::negotiate::matches_type_param(doc, params, &parsed.ctx) {
1878 0 : return Err(NgsiError::ResourceNotFound(format!(
1879 0 : "entity {id} does not match the type selector"
1880 0 : )));
1881 24 : }
1882 24 : let mut frag = fragment.clone();
1883 24 : apply_common_lang(doc, &mut frag, lang);
1884 24 : merge_into(doc, &frag, &ts);
1885 24 : Ok::<(), NgsiError>(())
1886 24 : })
1887 32 : .await?;
1888 24 : match res {
1889 8 : None => Err(NgsiError::ResourceNotFound(format!("entity {id} not found")).into()),
1890 0 : Some(Err(e)) => Err(e.into()),
1891 24 : Some(Ok(())) => Ok(no_content(&tenant)),
1892 : }
1893 42 : }
1894 :
1895 : /// 5.6.17.3: "An optional parameter indicating a common observedAt timestamp
1896 : /// to use across merged Attributes." It applies to the Attribute instances of
1897 : /// the Fragment that do not carry an observedAt of their own; a deletion
1898 : /// instance (5.5.12 NGSI-LD Null) removes the Attribute and takes none.
1899 40 : fn apply_common_observed_at(fragment: &mut Value, observed_at: Option<&str>) {
1900 40 : let (Some(ts), Some(obj)) = (observed_at, fragment.as_object_mut()) else {
1901 36 : return;
1902 : };
1903 8 : for (k, v) in obj.iter_mut() {
1904 8 : if is_meta(k) {
1905 0 : continue;
1906 8 : }
1907 8 : let Some(instances) = v.as_array_mut() else {
1908 0 : continue;
1909 : };
1910 8 : for inst in instances {
1911 8 : if antares_jsonld::is_deletion_instance(inst) {
1912 0 : continue;
1913 8 : }
1914 8 : if let Some(o) = inst.as_object_mut() {
1915 8 : o.entry("observedAt".to_owned())
1916 8 : .or_insert_with(|| Value::String(ts.to_owned()));
1917 0 : }
1918 : }
1919 : }
1920 40 : }
1921 :
1922 : /// 5.6.17.4: "If a common language tag is defined and a LanguageProperty
1923 : /// Attribute to be merged is represented as a string, the pre-existing
1924 : /// languageMap JSON object shall be preserved. The string value shall only
1925 : /// replace the value associated to the language tag key found within the
1926 : /// languageMap." The string instance is rewritten into a one-key languageMap
1927 : /// patch, which 5.5.12 then merges into the stored map key by key.
1928 26 : fn apply_common_lang(target: &Value, fragment: &mut Value, lang: Option<&str>) {
1929 26 : let (Some(lang), Some(frag)) = (lang, fragment.as_object_mut()) else {
1930 22 : return;
1931 : };
1932 4 : for (k, v) in frag.iter_mut() {
1933 4 : if is_meta(k) {
1934 0 : continue;
1935 4 : }
1936 4 : let pre_existing_langmap = target
1937 4 : .get(k)
1938 4 : .and_then(Value::as_array)
1939 4 : .is_some_and(|insts| insts.iter().any(|i| i.get("languageMap").is_some()));
1940 4 : if !pre_existing_langmap {
1941 0 : continue;
1942 4 : }
1943 4 : let Some(instances) = v.as_array_mut() else {
1944 0 : continue;
1945 : };
1946 4 : for inst in instances {
1947 4 : if antares_jsonld::is_deletion_instance(inst) {
1948 0 : continue;
1949 4 : }
1950 4 : let Some(o) = inst.as_object_mut() else {
1951 0 : continue;
1952 : };
1953 4 : let Some(s) = o.get("value").and_then(Value::as_str).map(str::to_owned) else {
1954 0 : continue;
1955 : };
1956 4 : o.remove("value");
1957 4 : o.insert("type".into(), Value::String("LanguageProperty".into()));
1958 4 : let mut map = Map::new();
1959 4 : map.insert(lang.to_owned(), Value::String(s));
1960 4 : o.insert("languageMap".into(), Value::Object(map));
1961 : }
1962 : }
1963 26 : }
1964 :
1965 : /// JSON Merge-Patch over internal docs (5.5.12).
1966 62 : pub fn merge_into(doc: &mut Value, fragment: &Value, ts: &str) {
1967 62 : let (Some(target), Some(frag)) = (doc.as_object_mut(), fragment.as_object()) else {
1968 0 : return;
1969 : };
1970 92 : for (k, v) in frag {
1971 92 : match k.as_str() {
1972 92 : "id" | "createdAt" | "modifiedAt" => continue,
1973 84 : "type" => {
1974 : // union of types
1975 26 : let mut cur: Vec<Value> = target
1976 26 : .get("type")
1977 26 : .and_then(Value::as_array)
1978 26 : .cloned()
1979 26 : .unwrap_or_default();
1980 30 : for t in v.as_array().cloned().unwrap_or_default() {
1981 30 : if !cur.contains(&t) {
1982 8 : cur.push(t);
1983 22 : }
1984 : }
1985 26 : target.insert("type".into(), Value::Array(cur));
1986 : }
1987 : // 5.5.12: "For each member of the Fragment, whose value is an
1988 : // NGSI-LD Null, contained by the target, the target member is
1989 : // removed." 4.18 admits the sentinel as a scope for exactly this
1990 : // reason, and expansion hands scopes over as an array, so the
1991 : // deletion arrives as its single entry. Storing it instead left
1992 : // the Entity scoped to a string no 4.18 grammar accepts.
1993 58 : "scope" => {
1994 8 : if antares_jsonld::is_ngsi_null_list(v) {
1995 4 : target.remove("scope");
1996 4 : } else {
1997 4 : target.insert("scope".into(), v.clone());
1998 4 : }
1999 : }
2000 : // 4.22: expiresAt is a settable Entity member (5.2.4, not in the
2001 : // read-only Table 5.2.2-1) — merge updates the storage expiry;
2002 : // an NGSI-LD Null removes it (5.5.12). Without this arm it fell
2003 : // through to the attribute path, where a bare string has no
2004 : // instances and the member was silently dropped.
2005 50 : "expiresAt" => {
2006 12 : if is_ngsi_null(v) {
2007 8 : target.remove("expiresAt");
2008 8 : } else {
2009 4 : target.insert("expiresAt".into(), v.clone());
2010 4 : }
2011 : }
2012 : _ => {
2013 38 : let frag_instances = v.as_array().cloned().unwrap_or_default();
2014 38 : let mut cur: Vec<Value> = target
2015 38 : .get(k)
2016 38 : .and_then(Value::as_array)
2017 38 : .cloned()
2018 38 : .unwrap_or_default();
2019 42 : for fi in frag_instances {
2020 42 : let is_delete = antares_jsonld::is_deletion_instance(&fi);
2021 42 : let want_ds = fi.get("datasetId").and_then(Value::as_str);
2022 42 : let pos = cur
2023 42 : .iter()
2024 42 : .position(|ci| ci.get("datasetId").and_then(Value::as_str) == want_ds);
2025 42 : match (is_delete, pos) {
2026 2 : (true, Some(p)) => {
2027 2 : cur.remove(p);
2028 2 : }
2029 0 : (true, None) => {}
2030 34 : (false, Some(p)) => {
2031 34 : merge_instance(&mut cur[p], &fi, ts);
2032 34 : }
2033 : (false, None) => {
2034 6 : let mut ni = fi.clone();
2035 6 : if let Some(o) = ni.as_object_mut() {
2036 6 : o.insert("createdAt".into(), Value::String(ts.to_owned()));
2037 6 : o.insert("modifiedAt".into(), Value::String(ts.to_owned()));
2038 6 : }
2039 6 : cur.push(ni);
2040 : }
2041 : }
2042 : }
2043 38 : if cur.is_empty() {
2044 2 : target.remove(k);
2045 36 : } else {
2046 36 : target.insert(k.clone(), Value::Array(cur));
2047 36 : }
2048 : }
2049 : }
2050 : }
2051 62 : target.insert("modifiedAt".into(), Value::String(ts.to_owned()));
2052 62 : }
2053 :
2054 34 : fn merge_instance(target: &mut Value, frag: &Value, ts: &str) {
2055 34 : let (Some(t), Some(f)) = (target.as_object_mut(), frag.as_object()) else {
2056 0 : return;
2057 : };
2058 84 : for (k, v) in f {
2059 84 : if k == "createdAt" || k == "modifiedAt" {
2060 0 : continue;
2061 84 : }
2062 84 : if v.is_null() || is_ngsi_null(v) {
2063 0 : t.remove(k);
2064 8 : } else if let (Some(cur), Some(patch)) =
2065 84 : (t.get_mut(k).and_then(Value::as_object_mut), v.as_object())
2066 8 : {
2067 8 : // 5.5.12: the merge goes "into JSON objects representing a
2068 8 : // Property value" — RFC 7396 with the NGSI-LD Null as removal.
2069 8 : merge_value_object(cur, patch);
2070 76 : } else {
2071 76 : t.insert(k.clone(), v.clone());
2072 76 : }
2073 : }
2074 34 : t.insert("modifiedAt".into(), Value::String(ts.to_owned()));
2075 34 : }
2076 :
2077 : /// RFC 7396 merge patch over a compound (JSON object) member value, with
2078 : /// "urn:ngsi-ld:null" / JSON null as the key-removal marker (5.5.12); the
2079 : /// sentinel itself is never stored (5.5.4).
2080 8 : fn merge_value_object(target: &mut Map<String, Value>, patch: &Map<String, Value>) {
2081 12 : for (k, v) in patch {
2082 12 : if v.is_null() || is_ngsi_null(v) {
2083 4 : target.remove(k);
2084 4 : } else if let (Some(cur), Some(po)) = (
2085 8 : target.get_mut(k).and_then(Value::as_object_mut),
2086 8 : v.as_object(),
2087 0 : ) {
2088 0 : merge_value_object(cur, po);
2089 8 : } else {
2090 8 : target.insert(k.clone(), v.clone());
2091 8 : }
2092 : }
2093 8 : }
2094 :
2095 : // ---------- PUT /entities/{id} — Replace (5.6.18) ----------
2096 :
2097 32 : pub async fn replace_entity(
2098 32 : State(st): State<AppState>,
2099 32 : Path(id): Path<String>,
2100 32 : CleanParams(params): CleanParams,
2101 32 : headers: HeaderMap,
2102 32 : body: Bytes,
2103 32 : ) -> Response {
2104 32 : let go = async {
2105 32 : let tenant = tenant_from(&headers)?;
2106 32 : antares_model::EntityId::new(&id)?;
2107 32 : check_params(¶ms, &["local", "type"])?;
2108 : // 5.5.7 again: the selector is expanded with the request's @context,
2109 : // not the core one. The target is judged before the body is parsed
2110 : // (5.6.18: an unknown target is 404 before body validation), so the
2111 : // header context is resolved on its own here.
2112 32 : let ctx0 = request_context(&st.loader, &headers).await?;
2113 32 : gate!(st, &tenant, &headers, "5.6.18", ids: &[&id]).await?;
2114 : // 5.6.18.4: the ?type selector narrows the target — a non-matching
2115 : // entity is "not known" for this replace.
2116 26 : let local_doc = st
2117 26 : .store
2118 26 : .get(&tenant, Kind::Entity, &id)
2119 26 : .await?
2120 26 : .filter(|d| crate::negotiate::matches_type_param(d, ¶ms, &ctx0));
2121 26 : let spec = crate::registry::CsrSpec {
2122 26 : ids: Some(vec![id.clone()]),
2123 26 : ..Default::default()
2124 26 : };
2125 26 : let regs =
2126 26 : match crate::federation::write_plan(&st, &tenant, &spec, &ctx0, ¶ms, &headers)
2127 26 : .await?
2128 : {
2129 0 : crate::federation::WritePlan::Answered(r) => return Ok(*r),
2130 26 : crate::federation::WritePlan::Forward(regs) => regs,
2131 : };
2132 26 : if regs.is_empty() {
2133 : // 5.6.18: an unknown target is 404 before body validation (057_03).
2134 : // The read above answers that; the write below decides again
2135 : // under the row lock, because between the two the target can be
2136 : // deleted and a replace that writes anyway puts it back.
2137 24 : if local_doc.is_none() {
2138 14 : return Err(NgsiError::ResourceNotFound(format!("entity {id} not found")).into());
2139 10 : }
2140 10 : let parsed = parse_body(&st.loader, &headers, &body, BodyKind::Standard).await?;
2141 10 : let obj = parsed.object(NgsiError::BadRequestData(
2142 10 : "entity must be a JSON object".into(),
2143 10 : ))?;
2144 10 : let mut expanded = expand_entity(obj, &parsed.ctx, ExpandOpts::default())?;
2145 10 : if expanded["id"].as_str() != Some(id.as_str()) {
2146 0 : return Err(NgsiError::BadRequestData("entity id mismatch".into()).into());
2147 10 : }
2148 10 : let ts = now_iso();
2149 10 : stamp_new(&mut expanded, &ts);
2150 10 : let res = st
2151 10 : .store
2152 10 : .mutate(&tenant, Kind::Entity, &id, |doc| {
2153 : // 5.6.18.4: the ?type selector narrows the target here too —
2154 : // the type of the row being written is the one that counts
2155 8 : if !crate::negotiate::matches_type_param(doc, ¶ms, &ctx0) {
2156 0 : return Err(NgsiError::ResourceNotFound(format!(
2157 0 : "entity {id} does not match the type selector"
2158 0 : )));
2159 8 : }
2160 : // 4.8: "createdAt ... shall be the date and time at which the
2161 : // Entity was created" — the target's own stamp, read under
2162 : // the lock rather than from a snapshot another write may
2163 : // already have replaced.
2164 8 : if let (Some(o), Some(created)) =
2165 8 : (expanded.as_object_mut(), doc.get("createdAt").cloned())
2166 8 : {
2167 8 : o.insert("createdAt".into(), created);
2168 8 : }
2169 8 : *doc = expanded.clone();
2170 8 : Ok::<(), NgsiError>(())
2171 8 : })
2172 10 : .await?;
2173 8 : return match res {
2174 2 : None => Err(NgsiError::ResourceNotFound(format!("entity {id} not found")).into()),
2175 0 : Some(Err(e)) => Err(e.into()),
2176 8 : Some(Ok(())) => Ok::<_, ApiError>(no_content(&tenant)),
2177 : };
2178 2 : }
2179 2 : let parsed = parse_body(&st.loader, &headers, &body, BodyKind::Standard).await?;
2180 2 : let obj = parsed.object(NgsiError::BadRequestData(
2181 2 : "entity must be a JSON object".into(),
2182 2 : ))?;
2183 2 : let expanded = expand_entity(obj, &parsed.ctx, ExpandOpts::default())?;
2184 2 : if expanded["id"].as_str() != Some(id.as_str()) {
2185 0 : return Err(NgsiError::BadRequestData("entity id mismatch".into()).into());
2186 2 : }
2187 2 : let mut parts = Vec::new();
2188 2 : let proxies: Vec<&crate::federation::FedReg> =
2189 2 : regs.iter().filter(|r| r.is_proxy()).collect();
2190 2 : let proxy_match = !proxies.is_empty();
2191 2 : if local_doc.is_some() || !proxy_match {
2192 0 : let gone = crate::federation::Part {
2193 0 : status: 404,
2194 0 : detail: format!("entity {id} not found locally"),
2195 0 : };
2196 0 : if local_doc.is_none() {
2197 0 : parts.push(gone);
2198 0 : } else {
2199 0 : let (rest, _) = crate::federation::strip_proxied(obj, &proxies, &parsed.ctx);
2200 0 : let mut local_exp = expand_entity(&rest, &parsed.ctx, ExpandOpts::default())?;
2201 0 : let ts = now_iso();
2202 0 : stamp_new(&mut local_exp, &ts);
2203 : // the same row lock as the local-only path above: the read
2204 : // that found the target is not the write that replaces it
2205 0 : let res = st
2206 0 : .store
2207 0 : .mutate(&tenant, Kind::Entity, &id, |doc| {
2208 0 : if !crate::negotiate::matches_type_param(doc, ¶ms, &ctx0) {
2209 0 : return Err(NgsiError::ResourceNotFound(format!(
2210 0 : "entity {id} does not match the type selector"
2211 0 : )));
2212 0 : }
2213 0 : if let (Some(o), Some(created)) =
2214 0 : (local_exp.as_object_mut(), doc.get("createdAt").cloned())
2215 0 : {
2216 0 : o.insert("createdAt".into(), created);
2217 0 : }
2218 0 : *doc = local_exp.clone();
2219 0 : Ok::<(), NgsiError>(())
2220 0 : })
2221 0 : .await?;
2222 0 : match res {
2223 0 : Some(Ok(())) => parts.push(crate::federation::Part {
2224 0 : status: 204,
2225 0 : detail: "replaced locally".into(),
2226 0 : }),
2227 0 : _ => parts.push(gone),
2228 : }
2229 : }
2230 2 : }
2231 2 : let ctx_url = crate::federation::ctx_link_url(&headers, &parsed.ctx.source);
2232 2 : let seg = crate::federation::path_segment(&id);
2233 2 : let fwd_q = type_selector_query(¶ms);
2234 2 : for reg in ®s {
2235 : // 5.6.18.4: proxy modes without Replace Entity support are an
2236 : // error of type Conflict; inclusive ones are not forwarded.
2237 2 : if !reg.supports("replaceEntity") {
2238 2 : if reg.is_proxy() {
2239 2 : parts.push(crate::federation::conflict_part("replaceEntity"));
2240 2 : }
2241 2 : continue;
2242 0 : }
2243 0 : let Some(frag) = crate::federation::reduce_to_scope(obj, reg, &parsed.ctx) else {
2244 0 : continue;
2245 : };
2246 0 : parts.push(
2247 0 : crate::federation::forward_part(
2248 0 : &st,
2249 0 : reqwest::Method::PUT,
2250 0 : format!("{}/ngsi-ld/v1/entities/{seg}", reg.endpoint),
2251 0 : &fwd_q,
2252 0 : &headers,
2253 0 : &tenant,
2254 0 : reg,
2255 0 : &ctx_url,
2256 0 : Some(frag),
2257 0 : )
2258 0 : .await,
2259 : );
2260 : }
2261 2 : Ok(crate::federation::combine(
2262 2 : parts,
2263 2 : no_content(&tenant),
2264 2 : &tenant,
2265 2 : ))
2266 32 : };
2267 32 : go.await.unwrap_or_else(|e| e.into_response())
2268 32 : }
2269 :
2270 : // ---------- Entity Ordering (4.23) ----------
2271 :
2272 : // ---------- GeoJSON output (6.3.15) ----------
2273 :
2274 : // ---------- GET /entities/{id}/attrs/{attrId} [+ /value] ----------
2275 : // NGSI-LD 2.0 pre-adoptions #14/#15: retrieve a single
2276 : // attribute of an entity, and its bare value. Additive-only: 2.0 defines the
2277 : // resources, 1.9.1 clients never see them unless asked.
2278 :
2279 26 : pub async fn retrieve_entity_attr(
2280 26 : State(st): State<AppState>,
2281 26 : Path((id, attr)): Path<(String, String)>,
2282 26 : CleanParams(params): CleanParams,
2283 26 : headers: HeaderMap,
2284 26 : ) -> Response {
2285 26 : match retrieve_attr_inner(&st, &id, &attr, false, ¶ms, &headers).await {
2286 10 : Ok(r) => r,
2287 16 : Err(e) => e.into_response(),
2288 : }
2289 26 : }
2290 :
2291 16 : pub async fn retrieve_entity_attr_value(
2292 16 : State(st): State<AppState>,
2293 16 : Path((id, attr)): Path<(String, String)>,
2294 16 : CleanParams(params): CleanParams,
2295 16 : headers: HeaderMap,
2296 16 : ) -> Response {
2297 16 : match retrieve_attr_inner(&st, &id, &attr, true, ¶ms, &headers).await {
2298 4 : Ok(r) => r,
2299 12 : Err(e) => e.into_response(),
2300 : }
2301 16 : }
2302 :
2303 42 : async fn retrieve_attr_inner(
2304 42 : st: &AppState,
2305 42 : id: &str,
2306 42 : attr: &str,
2307 42 : value_only: bool,
2308 42 : params: &HashMap<String, String>,
2309 42 : headers: &HeaderMap,
2310 42 : ) -> ApiResult<Response> {
2311 42 : let tenant = tenant_from(headers)?;
2312 42 : check_params(params, &["options", "format", "lang", "datasetId", "local"])?;
2313 42 : let ctx = request_context(&st.loader, headers).await?;
2314 42 : let filter = gate!(st, &tenant, headers, "5.7.1", ids: &[id]).await?;
2315 42 : let mut repr = parse_repr(params, &ctx)?;
2316 42 : crate::repr::narrow_repr(&mut repr, &filter);
2317 42 : antares_model::EntityId::new(id)?;
2318 42 : antares_model::check_attr_name(attr)?;
2319 42 : let doc = st
2320 42 : .store
2321 42 : .get(&tenant, Kind::Entity, id)
2322 42 : .await?
2323 42 : .ok_or_else(|| NgsiError::ResourceNotFound(format!("entity {id} not found")))?;
2324 26 : let attr_iri = antares_jsonld::expand_attr_name(attr, &ctx)?;
2325 18 : let node = doc.get(&attr_iri).ok_or_else(|| {
2326 4 : NgsiError::ResourceNotFound(format!("entity {id} has no attribute {attr}"))
2327 4 : })?;
2328 : // Compact through the entity pipeline so the attribute serializes exactly
2329 : // as it would inside a full retrieve.
2330 14 : let mini = serde_json::json!({
2331 14 : "id": doc.get("id").cloned().unwrap_or_default(),
2332 14 : "type": doc.get("type").cloned().unwrap_or_default(),
2333 14 : attr_iri.clone(): node.clone(),
2334 : });
2335 14 : let shaped = crate::repr::apply(&mini, &repr);
2336 14 : let compacted = compact_for(&repr, &shaped, &ctx);
2337 14 : let key = ctx.compact_iri(&attr_iri);
2338 14 : let member = compacted
2339 14 : .get(&key)
2340 14 : .cloned()
2341 14 : .ok_or_else(|| NgsiError::ResourceNotFound(format!("attribute {attr} not present")))?;
2342 14 : let body = if value_only {
2343 : // #15: the bare value — value / object / languageMap, whichever the
2344 : // attribute type carries; multi-instance attributes yield an array.
2345 4 : fn bare(v: &Value) -> Value {
2346 4 : match v {
2347 0 : Value::Array(a) => Value::Array(a.iter().map(bare).collect()),
2348 4 : Value::Object(o) => o
2349 4 : .get("value")
2350 4 : .or_else(|| o.get("object"))
2351 4 : .or_else(|| o.get("languageMap"))
2352 4 : .or_else(|| o.get("json"))
2353 4 : .or_else(|| o.get("vocab"))
2354 4 : .or_else(|| o.get("valueList"))
2355 4 : .or_else(|| o.get("objectList"))
2356 4 : .cloned()
2357 4 : .unwrap_or(Value::Null),
2358 0 : other => other.clone(),
2359 : }
2360 4 : }
2361 4 : bare(&member)
2362 : } else {
2363 10 : member
2364 : };
2365 14 : let accept = parse_accept(headers)?;
2366 14 : let mut resp = respond(StatusCode::OK, body, &ctx, accept, &tenant);
2367 14 : filter.mark_restricted(resp.headers_mut());
2368 14 : Ok(resp)
2369 42 : }
2370 :
2371 : /// 5.14.4.4: run the (split-reduced when applicable) local query and record
2372 : /// each matching id under the "@none" local marker; forward to matching
2373 : /// registrations supporting createEntityMapQueryEntity and merge each
2374 : /// returned EntityMap (ids → registration id, linkedMaps → remote map id);
2375 : /// store the local EntityMap and return it.
2376 : /// Known ceiling: the local candidate ids are the first max_limit matches —
2377 : /// the query is paged into the store instead of materializing every matching
2378 : /// Entity document, so one request cannot pull a whole tenant into memory.
2379 : /// Raise the cap if local candidate sets outgrow it.
2380 170 : pub(crate) async fn build_query_map(
2381 170 : st: &AppState,
2382 170 : tenant: &TenantId,
2383 170 : headers: &HeaderMap,
2384 170 : ctx: &antares_jsonld::Context,
2385 170 : params: &HashMap<String, String>,
2386 170 : filter: &crate::policy::Filter,
2387 170 : ) -> ApiResult<Value> {
2388 170 : let q_ast = params
2389 170 : .get("q")
2390 170 : .map(|q| antares_ql::parse_q(q))
2391 170 : .transpose()?;
2392 : // 5.14.4.4 a-e: too wide query
2393 170 : if !qualifies_non_wide(params, q_ast.as_ref()) {
2394 12 : return Err(NgsiError::BadRequestData(
2395 12 : "EntityMap query needs at least one of type, attrs, q, georel, or local=true \
2396 12 : (5.14.4.4 — too wide query)"
2397 12 : .into(),
2398 12 : )
2399 12 : .into());
2400 158 : }
2401 : // the candidate set is the NARROWED query's (ADR-0020)
2402 158 : let narrowed = filter.narrow_params(params)?;
2403 158 : let params = &narrowed;
2404 : // 5.14.4.4: invalid entity ids / csf are BadRequestData
2405 158 : if let Some(ids) = params.get("id") {
2406 0 : for id in ids.split(',') {
2407 0 : antares_model::EntityId::new(id.trim())?;
2408 : }
2409 158 : }
2410 158 : if let Some(csf) = params.get("csf") {
2411 0 : antares_ql::parse_q(csf)?;
2412 158 : }
2413 158 : let local_scope = params.get("local").map(String::as_str) == Some("true");
2414 158 : let split = params.get("splitEntities").map(String::as_str) == Some("true");
2415 : // Split entities: only id/type/idPattern narrow the local candidate set —
2416 : // value/geo/scope filters cannot be judged on a fragment (5.14.4.4).
2417 158 : let eff: HashMap<String, String> = if split && !local_scope {
2418 0 : params
2419 0 : .iter()
2420 0 : .filter(|(k, _)| ["id", "idPattern", "type", "local"].contains(&k.as_str()))
2421 0 : .map(|(k, v)| (k.clone(), v.clone()))
2422 0 : .collect()
2423 : } else {
2424 158 : params.clone()
2425 : };
2426 : // idPattern is invisible to the store, so a pushed page cannot BE the
2427 : // candidate set's page — it is still the window a walk steps through, and
2428 : // the literal the pattern forces has already narrowed the rows (5.2.33).
2429 : // Without the walk a tenant-wide pattern arrives as one allocation of
2430 : // every Entity the rest of the filter matched.
2431 158 : let mut emap = Map::new();
2432 158 : let mut offset = 0usize;
2433 164 : while emap.len() < st.max_limit {
2434 164 : let batch = filter_entities_paged(
2435 164 : st,
2436 164 : tenant,
2437 164 : &eff,
2438 164 : ctx,
2439 164 : Vec::new(),
2440 164 : Some((offset, st.max_limit)),
2441 164 : None,
2442 164 : )
2443 164 : .await?;
2444 318 : for d in &batch.docs {
2445 318 : if emap.len() == st.max_limit {
2446 4 : break;
2447 314 : }
2448 314 : if let Some(id) = d.get("id").and_then(Value::as_str) {
2449 314 : // "@none" refers to an Entity held locally (5.2.39)
2450 314 : emap.insert(id.to_owned(), json!(["@none"]));
2451 314 : }
2452 : }
2453 164 : match next_scan_offset(offset, batch.rows, 0, batch.paged, st.max_limit) {
2454 6 : Some(next) => offset = next,
2455 158 : None => break,
2456 : }
2457 : }
2458 158 : crate::entity_map::merge_and_store_map(st, tenant, headers, ctx, params, false, emap).await
2459 170 : }
2460 :
2461 : #[cfg(test)]
2462 : mod tests {
2463 : use super::merge_into;
2464 : use serde_json::json;
2465 :
2466 : /// 5.5.12 EXAMPLE 1 + the datasetId/type bullets: a merge updates the
2467 : /// named sub-attributes and leaves the others untouched; a fragment
2468 : /// instance with an unknown datasetId is ADDED (not replacing the
2469 : /// default); entity types are unioned.
2470 : #[test]
2471 4 : fn clause_5_5_12_merge_algorithm() {
2472 4 : let mut doc = json!({"id": "urn:x", "type": ["https://uri.etsi.org/ngsi-ld/default-context/T"],
2473 4 : "https://uri.etsi.org/ngsi-ld/default-context/temperature": [{
2474 4 : "type": "Property", "value": 25, "unitCode": "CEL",
2475 4 : "observedAt": "2022-03-14T01:59:26.535Z"}]});
2476 4 : merge_into(
2477 4 : &mut doc,
2478 4 : &json!({
2479 4 : "type": ["https://uri.etsi.org/ngsi-ld/default-context/T",
2480 4 : "https://uri.etsi.org/ngsi-ld/default-context/U"],
2481 4 : "https://uri.etsi.org/ngsi-ld/default-context/temperature": [
2482 4 : {"type": "Property", "value": 100,
2483 4 : "observedAt": "2022-03-14T13:00:00.000Z"},
2484 4 : {"type": "Property", "value": 7,
2485 4 : "datasetId": "urn:ngsi-ld:Dataset:extra"}
2486 4 : ]}),
2487 4 : "2026-08-11T00:00:00Z",
2488 : );
2489 : // EXAMPLE 1: value/observedAt updated, unitCode untouched
2490 4 : let t = &doc["https://uri.etsi.org/ngsi-ld/default-context/temperature"];
2491 4 : let default = t
2492 4 : .as_array()
2493 4 : .unwrap()
2494 4 : .iter()
2495 4 : .find(|i| i.get("datasetId").is_none())
2496 4 : .expect("default instance");
2497 4 : assert_eq!(default["value"], 100);
2498 4 : assert_eq!(default["observedAt"], "2022-03-14T13:00:00.000Z");
2499 4 : assert_eq!(
2500 4 : default["unitCode"], "CEL",
2501 : "unmentioned sub-attribute survives"
2502 : );
2503 : // unknown datasetId is added as a NEW instance
2504 4 : assert_eq!(t.as_array().unwrap().len(), 2);
2505 : // entity types are unioned, no duplicates
2506 4 : assert_eq!(
2507 4 : doc["type"],
2508 4 : json!([
2509 : "https://uri.etsi.org/ngsi-ld/default-context/T",
2510 : "https://uri.etsi.org/ngsi-ld/default-context/U"
2511 : ])
2512 : );
2513 4 : }
2514 :
2515 : /// 5.5.12: merge "merges the provided information with the existing
2516 : /// information up to an arbitrary depth, e.g. including going into JSON
2517 : /// objects representing a Property value" (RFC 7396 with the NGSI-LD
2518 : /// Null) — untouched keys survive, null-valued keys are removed, and the
2519 : /// null sentinel never lands in the stored document (5.5.4).
2520 : #[test]
2521 4 : fn merge_goes_into_compound_property_values() {
2522 4 : let mut doc = json!({"id": "urn:x", "type": ["T"],
2523 4 : "https://uri.etsi.org/ngsi-ld/default-context/address": [{
2524 4 : "type": "Property",
2525 4 : "value": {"street": "Straße des 17. Juni", "city": "Berlin",
2526 4 : "country": "Germany"}}]});
2527 4 : merge_into(
2528 4 : &mut doc,
2529 4 : &json!({"https://uri.etsi.org/ngsi-ld/default-context/address": [{
2530 4 : "type": "Property",
2531 4 : "value": {"street": "Pariser Platz",
2532 4 : "country": "urn:ngsi-ld:null"}}]}),
2533 4 : "2026-08-11T00:00:00Z",
2534 : );
2535 4 : let v = &doc["https://uri.etsi.org/ngsi-ld/default-context/address"][0]["value"];
2536 4 : assert_eq!(v["street"], "Pariser Platz");
2537 4 : assert_eq!(v["city"], "Berlin", "untouched keys survive the merge");
2538 4 : assert!(v.get("country").is_none(), "null removes the key");
2539 4 : assert!(
2540 4 : !doc.to_string().contains("urn:ngsi-ld:null"),
2541 : "the null sentinel must never be stored"
2542 : );
2543 4 : }
2544 :
2545 : #[test]
2546 4 : fn merge_sets_and_null_removes_expires_at() {
2547 4 : let mut doc = json!({"id": "urn:x", "type": ["T"]});
2548 4 : merge_into(
2549 4 : &mut doc,
2550 4 : &json!({"expiresAt": "2030-01-01T00:00:00Z"}),
2551 4 : "2026-08-08T00:00:00Z",
2552 : );
2553 4 : assert_eq!(doc["expiresAt"], "2030-01-01T00:00:00Z");
2554 4 : merge_into(
2555 4 : &mut doc,
2556 4 : &json!({"expiresAt": "urn:ngsi-ld:null"}),
2557 4 : "2026-08-08T00:00:01Z",
2558 : );
2559 4 : assert!(doc.get("expiresAt").is_none(), "NGSI-LD Null removes it");
2560 4 : }
2561 : }
2562 :
2563 : #[cfg(test)]
2564 : mod clause_4_16 {
2565 : use super::*;
2566 : use serde_json::json;
2567 :
2568 : /// 4.16: "Entity Types can be implicitly added by all operations that
2569 : /// update or append attributes. There is no operation to remove Entity
2570 : /// Types from an Entity."
2571 : #[test]
2572 4 : fn merge_unions_types_and_never_removes() {
2573 4 : let mut doc = json!({"id": "urn:x", "type": ["A"],
2574 4 : "https://ex/p": [{"type": "Property", "value": 1}]});
2575 4 : merge_into(
2576 4 : &mut doc,
2577 4 : &json!({"type": ["B"]}),
2578 4 : "2026-08-11T00:00:00.000Z",
2579 : );
2580 4 : assert_eq!(doc["type"], json!(["A", "B"]), "types union");
2581 : // a fragment naming FEWER types must not shrink the set
2582 4 : merge_into(
2583 4 : &mut doc,
2584 4 : &json!({"type": ["A"]}),
2585 4 : "2026-08-11T00:00:00.000Z",
2586 : );
2587 4 : assert_eq!(
2588 4 : doc["type"],
2589 4 : json!(["A", "B"]),
2590 : "no operation removes Entity Types"
2591 : );
2592 : // duplicates are not accumulated
2593 4 : merge_into(
2594 4 : &mut doc,
2595 4 : &json!({"type": ["B"]}),
2596 4 : "2026-08-11T00:00:00.000Z",
2597 : );
2598 4 : assert_eq!(doc["type"], json!(["A", "B"]));
2599 4 : }
2600 : }
2601 :
2602 : #[cfg(test)]
2603 : mod clause_4_17 {
2604 : use antares_jsonld::Loader;
2605 : use antares_ql::type_selection_matches;
2606 :
2607 : /// 4.17: disjunction via `|` or `,`, conjunction via `(a;b)`; short
2608 : /// names expand against the @context.
2609 : #[test]
2610 4 : fn selection_language_semantics() {
2611 4 : let ctx = Loader::new().core();
2612 : const D: &str = "https://uri.etsi.org/ngsi-ld/default-context/";
2613 4 : let home = format!("{D}Home");
2614 4 : let vehicle = format!("{D}Vehicle");
2615 4 : let motorhome = format!("{D}Motorhome");
2616 4 : let both: Vec<&str> = vec![&home, &vehicle];
2617 4 : let only_home: Vec<&str> = vec![&home];
2618 4 : let only_motor: Vec<&str> = vec![&motorhome];
2619 : // EXAMPLE 1: OR, both spellings
2620 4 : assert!(type_selection_matches("Building|Home", &only_home, &ctx));
2621 4 : assert!(type_selection_matches("Building,Home", &only_home, &ctx));
2622 4 : assert!(!type_selection_matches("Building|House", &only_home, &ctx));
2623 : // EXAMPLE 2: conjunction — ALL listed types required
2624 4 : assert!(type_selection_matches("(Home;Vehicle)", &both, &ctx));
2625 4 : assert!(
2626 4 : !type_selection_matches("(Home;Vehicle)", &only_home, &ctx),
2627 : "an entity with only Home must NOT match the conjunction"
2628 : );
2629 : // EXAMPLE 3: (Home;Vehicle)|Motorhome in both alternative spellings
2630 8 : for sel in ["(Home;Vehicle)|Motorhome", "(Home;Vehicle),Motorhome"] {
2631 8 : assert!(type_selection_matches(sel, &both, &ctx), "{sel}");
2632 8 : assert!(type_selection_matches(sel, &only_motor, &ctx), "{sel}");
2633 8 : assert!(!type_selection_matches(sel, &only_home, &ctx), "{sel}");
2634 : }
2635 4 : }
2636 : }
2637 :
2638 : #[cfg(test)]
2639 : mod clause_5_2_2 {
2640 : use super::*;
2641 : use antares_jsonld::{ExpandOpts, Loader};
2642 : use serde_json::json;
2643 :
2644 : /// 5.2.2: createdAt/modifiedAt/deletedAt "shall not be provided by
2645 : /// Context Producers. In the event that they are provided ... NGSI-LD
2646 : /// implementations shall ignore them" — server stamps win, no error.
2647 : #[test]
2648 4 : fn client_provided_system_timestamps_are_ignored() {
2649 4 : let ctx = Loader::new().core();
2650 4 : let doc = json!({"id": "urn:x", "type": "T",
2651 4 : "createdAt": "1999-01-01T00:00:00Z",
2652 4 : "modifiedAt": "1999-01-01T00:00:00Z",
2653 4 : "deletedAt": "1999-01-01T00:00:00Z",
2654 4 : "p": {"type": "Property", "value": 1,
2655 4 : "createdAt": "1999-01-01T00:00:00Z"}});
2656 4 : let mut out = antares_jsonld::expand_entity(
2657 4 : doc.as_object().expect("obj"),
2658 4 : &ctx,
2659 4 : ExpandOpts::default(),
2660 : )
2661 4 : .expect("providing common members is not an error");
2662 4 : stamp_new(&mut out, "2026-08-11T00:00:00.000Z");
2663 4 : assert_eq!(out["createdAt"], "2026-08-11T00:00:00.000Z");
2664 4 : assert_eq!(out["modifiedAt"], "2026-08-11T00:00:00.000Z");
2665 4 : assert!(out.get("deletedAt").is_none(), "deletedAt never creatable");
2666 4 : let inst = &out["https://uri.etsi.org/ngsi-ld/default-context/p"][0];
2667 4 : assert_eq!(
2668 4 : inst["createdAt"], "2026-08-11T00:00:00.000Z",
2669 : "instance-level client timestamp ignored too"
2670 : );
2671 4 : }
2672 :
2673 : /// 5.2.2: common members are only generated "when the Context Consumer
2674 : /// explicitly asks for their inclusion" (sysAttrs, 6.3.11).
2675 : #[test]
2676 4 : fn common_members_appear_only_on_request() {
2677 4 : let doc = json!({"id": "urn:x", "type": ["T"],
2678 4 : "createdAt": "2026-08-11T00:00:00.000Z",
2679 4 : "modifiedAt": "2026-08-11T00:00:00.000Z",
2680 4 : "https://uri.etsi.org/ngsi-ld/default-context/p": [
2681 4 : {"type": "Property", "value": 1,
2682 4 : "createdAt": "2026-08-11T00:00:00.000Z",
2683 4 : "modifiedAt": "2026-08-11T00:00:00.000Z"}]});
2684 4 : let plain = crate::repr::apply(&doc, &crate::repr::Repr::default());
2685 4 : assert!(plain.get("createdAt").is_none());
2686 4 : assert!(plain["https://uri.etsi.org/ngsi-ld/default-context/p"][0]
2687 4 : .get("modifiedAt")
2688 4 : .is_none());
2689 4 : let sys = crate::repr::apply(
2690 4 : &doc,
2691 4 : &crate::repr::Repr {
2692 4 : sys_attrs: true,
2693 4 : ..Default::default()
2694 4 : },
2695 : );
2696 4 : assert!(sys.get("createdAt").is_some());
2697 4 : assert!(sys["https://uri.etsi.org/ngsi-ld/default-context/p"][0]
2698 4 : .get("modifiedAt")
2699 4 : .is_some());
2700 4 : }
2701 : }
2702 :
2703 : /// 4.5.23.1: "When retrieving Linked Entities, it is necessary to limit
2704 : /// retrieval to avoid cascades of an excessive length, duplicates or loops."
2705 : #[cfg(test)]
2706 : mod clause_4_5_23_bounds {
2707 : use super::*;
2708 : use crate::repr::inline_join;
2709 : use axum::body::Body;
2710 : use axum::http::Request;
2711 : use http_body_util::BodyExt;
2712 : use serde_json::json;
2713 : use tower::ServiceExt;
2714 :
2715 8 : fn app() -> axum::Router {
2716 8 : crate::router(AppState::new("antares-test".into()))
2717 8 : }
2718 :
2719 20 : async fn create(app: &axum::Router, body: Value) {
2720 20 : let payload = body.to_string();
2721 20 : let resp = app
2722 20 : .clone()
2723 20 : .oneshot(
2724 20 : Request::post("/ngsi-ld/v1/entities")
2725 20 : .header("Content-Type", "application/json")
2726 20 : .header("Content-Length", payload.len().to_string())
2727 20 : .body(Body::from(payload))
2728 20 : .expect("req"),
2729 20 : )
2730 20 : .await
2731 20 : .expect("resp");
2732 20 : assert_eq!(resp.status(), StatusCode::CREATED, "create failed");
2733 20 : }
2734 :
2735 8 : async fn get(app: &axum::Router, uri: &str) -> (axum::http::response::Parts, Value) {
2736 8 : let resp = app
2737 8 : .clone()
2738 8 : .oneshot(Request::get(uri).body(Body::empty()).expect("req"))
2739 8 : .await
2740 8 : .expect("resp");
2741 8 : let (parts, body) = resp.into_parts();
2742 8 : let bytes = body.collect().await.expect("body").to_bytes();
2743 8 : let json: Value = serde_json::from_slice(&bytes).expect("json");
2744 8 : (parts, json)
2745 8 : }
2746 :
2747 : /// 4.5.23.1/4.5.23.3: the flattened array carries the Linking Entity and
2748 : /// its Linked Entities — a Relationship pointing back at the root is a
2749 : /// loop, so the root shall appear exactly ONCE, not once as the Linking
2750 : /// Entity and again as its own Linked Entity.
2751 : #[tokio::test]
2752 4 : async fn flat_join_never_repeats_the_root_entity() {
2753 4 : let app = app();
2754 4 : let root = "urn:ngsi-ld:Loop:root";
2755 4 : let leaf = "urn:ngsi-ld:Loop:leaf";
2756 4 : create(&app, json!({"id": leaf, "type": "Loop"})).await;
2757 4 : create(
2758 4 : &app,
2759 4 : json!({"id": root, "type": "Loop",
2760 4 : "self": {"type": "Relationship", "object": root},
2761 4 : "other": {"type": "Relationship", "object": leaf}}),
2762 4 : )
2763 4 : .await;
2764 :
2765 4 : let (_, body) = get(
2766 4 : &app,
2767 4 : &format!("/ngsi-ld/v1/entities/{root}?join=flat&joinLevel=3"),
2768 : )
2769 4 : .await;
2770 4 : let arr = match body {
2771 4 : Value::Array(a) => a,
2772 0 : other => vec![other],
2773 : };
2774 4 : let (mut roots, mut leaves) = (0usize, 0usize);
2775 8 : for e in &arr {
2776 8 : match e["id"].as_str() {
2777 8 : Some(id) if id == root => roots += 1,
2778 4 : Some(id) if id == leaf => leaves += 1,
2779 0 : _ => {}
2780 : }
2781 : }
2782 4 : assert_eq!(
2783 : roots, 1,
2784 : "the root must appear exactly once in the flattened array: {arr:?}"
2785 : );
2786 4 : assert_eq!(
2787 : leaves, 1,
2788 : "the genuine Linked Entity is still there once: {arr:?}"
2789 : );
2790 4 : assert_eq!(arr.len(), 2, "no other entity is in the array: {arr:?}");
2791 4 : }
2792 :
2793 : /// 4.5.23.1: a cyclic graph at a high joinLevel shall not cascade — the
2794 : /// walk stops at entities it already resolved and says so with an
2795 : /// NGSILD-Warning (6.3.17) instead of expanding fan-out^joinLevel.
2796 : #[tokio::test]
2797 4 : async fn cyclic_inline_join_stops_instead_of_cascading() {
2798 4 : let app = app();
2799 4 : let ids = [
2800 4 : "urn:ngsi-ld:Cyc:a",
2801 4 : "urn:ngsi-ld:Cyc:b",
2802 4 : "urn:ngsi-ld:Cyc:c",
2803 4 : ];
2804 : // complete graph: every entity links to every entity, itself included
2805 12 : for id in ids {
2806 12 : create(
2807 12 : &app,
2808 12 : json!({"id": id, "type": "Cyc",
2809 12 : "toA": {"type": "Relationship", "object": ids[0]},
2810 12 : "toB": {"type": "Relationship", "object": ids[1]},
2811 12 : "toC": {"type": "Relationship", "object": ids[2]}}),
2812 12 : )
2813 12 : .await;
2814 : }
2815 :
2816 4 : let (parts, body) = get(
2817 4 : &app,
2818 4 : &format!("/ngsi-ld/v1/entities/{}?join=inline&joinLevel=9", ids[0]),
2819 : )
2820 4 : .await;
2821 4 : assert_eq!(parts.status, StatusCode::OK);
2822 : // 3^9 embeddings if the walk is unbounded; a handful if it is not
2823 4 : let embedded = body.to_string().matches("urn:ngsi-ld:Cyc:").count();
2824 4 : assert!(
2825 4 : embedded < 64,
2826 : "the cyclic walk cascaded: {embedded} entity references embedded"
2827 : );
2828 4 : assert!(
2829 4 : parts.headers.get("NGSILD-Warning").is_some(),
2830 4 : "a truncated Linked Entity Retrieval must be reported (6.3.17)"
2831 4 : );
2832 4 : }
2833 :
2834 : /// 4.5.23.1: joinLevel bounds the depth, not the width — one retrieval
2835 : /// may only buy MAX_JOIN_LOOKUPS Linked Entity reads, and the truncation
2836 : /// is reported back to the caller.
2837 : #[tokio::test]
2838 4 : async fn wide_inline_join_stops_at_the_lookup_budget() {
2839 4 : let st = AppState::new("antares-test".into());
2840 4 : let tenant = TenantId::default();
2841 4 : let ctx = antares_jsonld::Loader::new().core();
2842 4 : let mut targets: Vec<Value> = Vec::new();
2843 4400 : for n in 0..MAX_JOIN_LOOKUPS + 100 {
2844 4400 : let id = format!("urn:ngsi-ld:Wide:{n}");
2845 4400 : let doc = json!({"id": &id, "type":
2846 : ["https://uri.etsi.org/ngsi-ld/default-context/Wide"]});
2847 4400 : st.store
2848 4400 : .upsert(&tenant, Kind::Entity, &id, doc)
2849 4400 : .await
2850 4400 : .expect("seed");
2851 4400 : targets.push(Value::String(id));
2852 : }
2853 4 : let mut compacted = json!({"id": "urn:ngsi-ld:Wide:root", "type": "Wide",
2854 4 : "links": {"type": "Relationship", "object": Value::Array(targets)}});
2855 4 : let complete = inline_join(
2856 4 : &st,
2857 4 : &tenant,
2858 4 : &ctx,
2859 4 : &crate::repr::Repr::default(),
2860 4 : &mut compacted,
2861 4 : 1,
2862 4 : )
2863 4 : .await;
2864 4 : assert!(!complete, "the budget was hit, so the walk is incomplete");
2865 4 : assert_eq!(
2866 4 : compacted["links"]["entity"]
2867 4 : .as_array()
2868 4 : .expect("entity array")
2869 4 : .len(),
2870 4 : MAX_JOIN_LOOKUPS,
2871 4 : "no more Linked Entities than the budget are resolved"
2872 4 : );
2873 4 : }
2874 : }
2875 :
2876 : /// 5.6.6.4 / 5.6.17.4 / 5.6.18.4 / 5.6.21.4: "matching input data is
2877 : /// forwarded to the Registration endpoint" — the forward may narrow what the
2878 : /// peer does, never widen it.
2879 : #[cfg(test)]
2880 : mod forwarded_input_data {
2881 : use super::*;
2882 :
2883 44 : fn params(pairs: &[(&str, &str)]) -> HashMap<String, String> {
2884 44 : pairs
2885 44 : .iter()
2886 128 : .map(|(k, v)| ((*k).to_owned(), (*v).to_owned()))
2887 44 : .collect()
2888 44 : }
2889 :
2890 : /// 5.6.21.3 lists the type selector, the id list, the id pattern, the
2891 : /// restrictive and exclusionary Attribute-name lists, the query, the
2892 : /// geoquery, the Scope query and the context source filter as Purge
2893 : /// input data. Every one of them restricts the purge, so every one of
2894 : /// them travels.
2895 : #[test]
2896 4 : fn purge_forward_carries_every_restriction_the_client_issued() {
2897 4 : let p = params(&[
2898 4 : ("type", "Vehicle"),
2899 4 : ("id", "urn:ngsi-ld:Vehicle:A1"),
2900 4 : ("idPattern", "^urn:ngsi-ld:Vehicle:"),
2901 4 : ("attrs", "speed"),
2902 4 : ("q", "speed>5"),
2903 4 : ("georel", "near;maxDistance==100"),
2904 4 : ("geometry", "Point"),
2905 4 : ("coordinates", "[0,0]"),
2906 4 : ("geoproperty", "location"),
2907 4 : ("scopeQ", "/x"),
2908 4 : ("csf", "name==p"),
2909 4 : ("keep", "name"),
2910 4 : ("local", "false"),
2911 4 : ]);
2912 4 : let q = forwarded_purge_query(&p);
2913 48 : for k in [
2914 4 : "type",
2915 4 : "id",
2916 4 : "idPattern",
2917 4 : "attrs",
2918 4 : "q",
2919 4 : "georel",
2920 4 : "geometry",
2921 4 : "coordinates",
2922 4 : "geoproperty",
2923 4 : "scopeQ",
2924 4 : "csf",
2925 4 : "keep",
2926 4 : ] {
2927 48 : assert!(
2928 312 : q.iter().any(|(a, _)| a == k),
2929 : "{k} was dropped, so the peer executes a wider purge than the \
2930 : client issued: {q:?}"
2931 : );
2932 : }
2933 4 : assert!(
2934 48 : !q.iter().any(|(k, _)| k == "local"),
2935 : "local scope is the reason a forward happens at all — it is not \
2936 : itself forwarded (5.5.13): {q:?}"
2937 : );
2938 4 : assert_eq!(q.len(), 12, "nothing else is invented: {q:?}");
2939 4 : }
2940 :
2941 : /// The exclusionary list travels on its own too — `drop=` alone must not
2942 : /// reach the peer as a bare, entity-deleting purge.
2943 : #[test]
2944 4 : fn purge_forward_carries_drop_and_omits_absent_members() {
2945 4 : let q = forwarded_purge_query(¶ms(&[("type", "Vehicle"), ("drop", "speed")]));
2946 4 : assert!(
2947 4 : q.contains(&("drop".to_owned(), "speed".to_owned())),
2948 : "{q:?}"
2949 : );
2950 4 : assert_eq!(q.len(), 2, "absent parameters are not forwarded: {q:?}");
2951 4 : }
2952 :
2953 : /// 5.6.6.3 / 5.6.17.3 / 5.6.18.3: the selector of Entity types is input
2954 : /// data of Delete, Merge and Replace Entity. A registration may cover
2955 : /// several types, so the peer needs the selector to reach the same
2956 : /// verdict this broker reached locally.
2957 : #[test]
2958 4 : fn write_forwards_carry_the_type_selector() {
2959 4 : assert_eq!(
2960 4 : type_selector_query(¶ms(&[("type", "Vehicle"), ("local", "false")])),
2961 4 : vec![("type".to_owned(), "Vehicle".to_owned())]
2962 : );
2963 4 : assert!(
2964 4 : type_selector_query(¶ms(&[("local", "false")])).is_empty(),
2965 : "no selector, nothing to forward"
2966 : );
2967 4 : }
2968 :
2969 : /// 5.7.2.4 applies q/geoquery/Scope query/Attributes only after remote
2970 : /// parts have been aggregated, and 4.23 orders by the value of a member —
2971 : /// neither survives a store-side projection of that member.
2972 : #[test]
2973 4 : fn projection_is_pushed_down_only_when_the_store_answer_is_final() {
2974 4 : let plain = params(&[("type", "T"), ("pick", "name")]);
2975 4 : assert!(
2976 4 : proj_pushdown_allowed(true, &plain),
2977 : "a purely local query keeps the pushdown"
2978 : );
2979 4 : assert!(
2980 4 : !proj_pushdown_allowed(false, &plain),
2981 : "federated candidates mean the filters run again after the merge"
2982 : );
2983 4 : assert!(
2984 4 : !proj_pushdown_allowed(true, ¶ms(&[("attrs", "name"), ("orderBy", "age")])),
2985 : "the ordering member must survive to be compared"
2986 : );
2987 4 : }
2988 :
2989 : /// 6.3.10 makes `limit=0` legal only with `count=true` — an answer that is
2990 : /// a count and no rows. That shape is pushed to the store; the filters the
2991 : /// store cannot see (idPattern, 4.23 ordering, federated candidates merged
2992 : /// per 5.7.2.4) still forfeit the page.
2993 : #[test]
2994 4 : fn the_count_only_page_is_pushed_down_but_store_blind_filters_are_not() {
2995 4 : assert!(
2996 4 : page_pushdown_allowed(
2997 : true,
2998 4 : ¶ms(&[("type", "T"), ("limit", "0"), ("count", "true")])
2999 : ),
3000 : "a count is answered by counting, not by materializing the match set"
3001 : );
3002 4 : assert!(
3003 4 : page_pushdown_allowed(true, ¶ms(&[("type", "T"), ("limit", "10")])),
3004 : "a plain local query keeps the pushdown"
3005 : );
3006 4 : assert!(
3007 4 : !page_pushdown_allowed(true, ¶ms(&[("type", "T"), ("idPattern", "^urn:")])),
3008 : "idPattern is applied after the store, so it drops rows the SQL \
3009 : page already counted"
3010 : );
3011 4 : assert!(
3012 4 : !page_pushdown_allowed(true, ¶ms(&[("type", "T"), ("orderBy", "speed")])),
3013 : "4.23 orders the whole match set before the page is cut"
3014 : );
3015 4 : assert!(
3016 4 : !page_pushdown_allowed(false, ¶ms(&[("type", "T")])),
3017 : "federated candidates are merged after the store answered"
3018 : );
3019 4 : }
3020 : }
3021 :
3022 : /// 5.6.1 Create Entity and 5.6.21 Purge Entities, end to end over the store.
3023 : #[cfg(test)]
3024 : mod clause_5_6_1_and_5_6_21 {
3025 : use super::*;
3026 : use axum::body::Body;
3027 : use axum::http::Request;
3028 : use serde_json::json;
3029 : use tower::ServiceExt;
3030 :
3031 : /// 5.6.1.5: the output is "the URI of the created Entity", returned in
3032 : /// the Location header. An id is one path segment (RFC 3986 clause 3.3),
3033 : /// so a `#` in it must not be able to end the segment.
3034 : #[tokio::test]
3035 4 : async fn location_header_percent_encodes_the_entity_id() {
3036 4 : let app = crate::router(AppState::new("antares-test".into()));
3037 4 : let id = "urn:ngsi-ld:Vehicle:A#4567";
3038 4 : let payload = json!({"id": id, "type": "Vehicle"}).to_string();
3039 4 : let resp = app
3040 4 : .oneshot(
3041 4 : Request::post("/ngsi-ld/v1/entities")
3042 4 : .header("Content-Type", "application/json")
3043 4 : .header("Content-Length", payload.len().to_string())
3044 4 : .body(Body::from(payload))
3045 4 : .expect("req"),
3046 4 : )
3047 4 : .await
3048 4 : .expect("resp");
3049 4 : assert_eq!(resp.status(), StatusCode::CREATED);
3050 4 : let loc = resp
3051 4 : .headers()
3052 4 : .get("Location")
3053 4 : .expect("Location header")
3054 4 : .to_str()
3055 4 : .expect("ascii");
3056 4 : assert_eq!(loc, "/ngsi-ld/v1/entities/urn:ngsi-ld:Vehicle:A%234567");
3057 4 : assert!(
3058 4 : !loc.contains('#'),
3059 4 : "a raw # truncates the URL at the fragment and addresses another \
3060 4 : resource: {loc}"
3061 4 : );
3062 4 : }
3063 :
3064 14 : async fn seed(st: &AppState, tenant: &TenantId, n: usize) {
3065 10576 : for i in 0..n {
3066 10576 : let id = format!("urn:ngsi-ld:Purge:{i:05}");
3067 10576 : let doc = json!({"id": &id,
3068 10576 : "type": ["https://uri.etsi.org/ngsi-ld/default-context/Purge"],
3069 10576 : "https://uri.etsi.org/ngsi-ld/default-context/name":
3070 10576 : [{"type": "Property", "value": "n"}],
3071 10576 : "https://uri.etsi.org/ngsi-ld/default-context/speed":
3072 10576 : [{"type": "Property", "value": 1}]});
3073 10576 : st.store
3074 10576 : .upsert(tenant, Kind::Entity, &id, doc)
3075 10576 : .await
3076 10576 : .expect("seed");
3077 : }
3078 14 : }
3079 :
3080 14 : fn purge_params(pairs: &[(&str, &str)]) -> HashMap<String, String> {
3081 14 : pairs
3082 14 : .iter()
3083 24 : .map(|(k, v)| ((*k).to_owned(), (*v).to_owned()))
3084 14 : .collect()
3085 14 : }
3086 :
3087 : /// 5.6.21.4: the implementation "shall delete all Entities that can be
3088 : /// found locally using retrieved list of Entity ids" — all of them, not
3089 : /// one page of them, however many rounds that takes.
3090 : #[tokio::test]
3091 4 : async fn purge_deletes_every_match_across_page_boundaries() {
3092 4 : let st = AppState::new("antares-test".into());
3093 4 : let tenant = TenantId::default();
3094 4 : let n = PURGE_CHUNK * 2 + 7;
3095 4 : seed(&st, &tenant, n).await;
3096 4 : let resp = purge_inner(&st, &purge_params(&[("type", "Purge")]), &HeaderMap::new())
3097 4 : .await
3098 4 : .expect("purge");
3099 4 : assert_eq!(resp.status(), StatusCode::NO_CONTENT);
3100 4 : assert!(
3101 4 : st.store
3102 4 : .list(&tenant, Kind::Entity)
3103 4 : .await
3104 4 : .expect("list")
3105 4 : .is_empty(),
3106 4 : "entities survived the purge"
3107 4 : );
3108 4 : }
3109 :
3110 : /// 5.6.21.4: with an exclusionary list the implementation "shall delete
3111 : /// all but the given set of Attributes" — the Entities themselves
3112 : /// survive, again for the whole match set and not just its first page.
3113 : #[tokio::test]
3114 4 : async fn purge_with_keep_prunes_every_match_and_deletes_no_entity() {
3115 4 : let st = AppState::new("antares-test".into());
3116 4 : let tenant = TenantId::default();
3117 4 : let n = PURGE_CHUNK * 2 + 7;
3118 4 : seed(&st, &tenant, n).await;
3119 4 : let resp = purge_inner(
3120 4 : &st,
3121 4 : &purge_params(&[("type", "Purge"), ("keep", "name")]),
3122 4 : &HeaderMap::new(),
3123 4 : )
3124 4 : .await
3125 4 : .expect("purge");
3126 4 : assert_eq!(resp.status(), StatusCode::NO_CONTENT);
3127 4 : let left = st.store.list(&tenant, Kind::Entity).await.expect("list");
3128 4 : assert_eq!(left.len(), n, "keep= prunes attributes, not entities");
3129 4028 : for doc in &left {
3130 4028 : assert!(
3131 4028 : doc.get("https://uri.etsi.org/ngsi-ld/default-context/name")
3132 4028 : .is_some(),
3133 4 : "the kept attribute is still there: {doc}"
3134 4 : );
3135 4028 : assert!(
3136 4028 : doc.get("https://uri.etsi.org/ngsi-ld/default-context/speed")
3137 4028 : .is_none(),
3138 4 : "every other attribute is gone: {doc}"
3139 4 : );
3140 4 : }
3141 4 : }
3142 :
3143 : /// 5.6.21.4: "id matches the id pattern passed as a parameter" — the
3144 : /// pattern narrows the match set, and only that set is deleted.
3145 : #[tokio::test]
3146 4 : async fn purge_deletes_the_id_pattern_matches_and_nothing_else() {
3147 4 : let st = AppState::new("antares-test".into());
3148 4 : let tenant = TenantId::default();
3149 4 : seed(&st, &tenant, 30).await;
3150 4 : let resp = purge_inner(
3151 4 : &st,
3152 4 : &purge_params(&[("type", "Purge"), ("idPattern", "^urn:ngsi-ld:Purge:0000")]),
3153 4 : &HeaderMap::new(),
3154 4 : )
3155 4 : .await
3156 4 : .expect("purge");
3157 4 : assert_eq!(resp.status(), StatusCode::NO_CONTENT);
3158 4 : let left: Vec<String> = st
3159 4 : .store
3160 4 : .list(&tenant, Kind::Entity)
3161 4 : .await
3162 4 : .expect("list")
3163 4 : .iter()
3164 80 : .filter_map(|d| d["id"].as_str().map(str::to_owned))
3165 4 : .collect();
3166 4 : assert_eq!(
3167 4 : left.len(),
3168 : 20,
3169 : "only the ten pattern matches went: {left:?}"
3170 : );
3171 4 : assert!(
3172 4 : !left
3173 4 : .iter()
3174 80 : .any(|id| id.starts_with("urn:ngsi-ld:Purge:0000")),
3175 : "a pattern match survived: {left:?}"
3176 : );
3177 4 : assert!(
3178 4 : left.contains(&"urn:ngsi-ld:Purge:00010".to_owned()),
3179 4 : "an Entity the pattern does not match must not be purged: {left:?}"
3180 4 : );
3181 4 : }
3182 :
3183 : /// 5.6.21.4 deletes "all Entities that can be found locally using
3184 : /// retrieved list of Entity ids". The retrieval is chunked, and idPattern
3185 : /// is applied after the store — so a chunk can come back narrowed, or
3186 : /// empty, while matches still wait behind it. The walk continues on the
3187 : /// store's row count, never on the surviving subset.
3188 : #[test]
3189 4 : fn purge_walks_on_the_rows_the_store_returned_not_the_narrowed_subset() {
3190 4 : assert_eq!(
3191 4 : next_scan_offset(0, 500, 0, true, 500),
3192 : Some(500),
3193 : "a full chunk the pattern narrowed to nothing still has a successor"
3194 : );
3195 4 : assert_eq!(
3196 4 : next_scan_offset(500, 500, 120, true, 500),
3197 : Some(880),
3198 : "the 380 rows the round did not remove are stepped over"
3199 : );
3200 4 : assert_ne!(
3201 4 : next_scan_offset(0, 500, 0, true, 500),
3202 : None,
3203 : "stopping here leaves every match behind the first chunk alive"
3204 : );
3205 4 : }
3206 :
3207 : /// 5.6.21.4 against a store that really pages (memory answers every query
3208 : /// with the whole match set, so it cannot exercise the walk): with an
3209 : /// idPattern spread across the match set, every chunk arrives narrowed —
3210 : /// and the purge still has to delete "all Entities that can be found
3211 : /// locally using retrieved list of Entity ids", not just the first chunk's
3212 : /// share. Skips without ANTARES_TEST_DATABASE_URL.
3213 : #[cfg(feature = "postgres")]
3214 : #[tokio::test(flavor = "multi_thread")]
3215 4 : async fn purge_over_a_paging_store_deletes_every_id_pattern_match() {
3216 4 : let url = match std::env::var("ANTARES_TEST_DATABASE_URL") {
3217 4 : Ok(u) => u,
3218 : Err(_) => {
3219 0 : eprintln!("SKIP: ANTARES_TEST_DATABASE_URL not set");
3220 0 : return;
3221 : }
3222 : };
3223 4 : let pool = antares_sql::store::pg::connect(&url, 5)
3224 4 : .await
3225 4 : .expect("connect");
3226 4 : let tenant = TenantId::new("purgepaging").expect("tenant");
3227 4 : antares_sql::store::pg::ensure_tenant(&pool, &tenant)
3228 4 : .await
3229 2 : .expect("tenant row");
3230 2 : let st = AppState::with_store(
3231 2 : "antares-test".into(),
3232 2 : std::sync::Arc::new(antares_sql::store::any::AnyStore::Pg(
3233 2 : antares_sql::store::any::PgBackend::new(pool),
3234 2 : )),
3235 2 : "postgres",
3236 : );
3237 2 : for doc in st.store.list(&tenant, Kind::Entity).await.expect("list") {
3238 0 : if let Some(id) = doc["id"].as_str() {
3239 0 : st.store
3240 0 : .delete(&tenant, Kind::Entity, id)
3241 0 : .await
3242 0 : .expect("clean");
3243 0 : }
3244 : }
3245 : // more than two chunks, and the pattern matches every second id — so
3246 : // no chunk the store pages is full once the pattern filtered it
3247 2 : let n = PURGE_CHUNK * 2 + 200;
3248 2 : seed(&st, &tenant, n).await;
3249 : // the purge must run AS the seeded tenant — with no NGSILD-Tenant
3250 : // header it correctly purges the default tenant's (empty) match set
3251 : // and every seeded row survives (5.5.10)
3252 2 : let mut headers = HeaderMap::new();
3253 2 : headers.insert("NGSILD-Tenant", "purgepaging".parse().expect("header"));
3254 2 : let resp = purge_inner(
3255 2 : &st,
3256 2 : &purge_params(&[("type", "Purge"), ("idPattern", "[02468]$")]),
3257 2 : &headers,
3258 2 : )
3259 2 : .await
3260 2 : .expect("purge");
3261 2 : assert_eq!(resp.status(), StatusCode::NO_CONTENT);
3262 2 : let left: Vec<String> = st
3263 2 : .store
3264 2 : .list(&tenant, Kind::Entity)
3265 2 : .await
3266 2 : .expect("list")
3267 2 : .iter()
3268 1200 : .filter_map(|d| d["id"].as_str().map(str::to_owned))
3269 2 : .collect();
3270 2 : assert!(
3271 2 : !left
3272 2 : .iter()
3273 1200 : .any(|id| id.ends_with(['0', '2', '4', '6', '8'])),
3274 : "{} of {} pattern matches survived the chunked walk",
3275 0 : left.iter()
3276 0 : .filter(|id| id.ends_with(['0', '2', '4', '6', '8']))
3277 0 : .count(),
3278 0 : n / 2
3279 : );
3280 2 : assert_eq!(
3281 2 : left.len(),
3282 2 : n / 2,
3283 : "an Entity the pattern does not match must not be purged"
3284 : );
3285 1202 : for id in &left {
3286 1202 : st.store
3287 1200 : .delete(&tenant, Kind::Entity, id)
3288 1200 : .await
3289 1202 : .expect("clean");
3290 4 : }
3291 4 : }
3292 :
3293 : /// The same walk terminates: a chunk the store did not page IS the whole
3294 : /// match set, a short chunk is the last one, and a round that deleted its
3295 : /// whole chunk re-reads the window the deletions shifted down.
3296 : #[test]
3297 4 : fn purge_stops_when_the_match_set_is_exhausted() {
3298 4 : assert_eq!(
3299 4 : next_scan_offset(0, 500, 500, false, 500),
3300 : None,
3301 : "an unpaged answer is the whole match set"
3302 : );
3303 4 : assert_eq!(
3304 4 : next_scan_offset(0, 7, 7, true, 500),
3305 : None,
3306 : "a chunk shorter than the page size is the last one"
3307 : );
3308 4 : assert_eq!(
3309 4 : next_scan_offset(0, 500, 500, true, 500),
3310 : Some(0),
3311 : "everything deleted — the rest of the match set shifted to the front"
3312 : );
3313 4 : }
3314 : }
3315 :
3316 : /// 5.6.17 Merge Entity and the Linked Entity Retrieval parameters of
3317 : /// Table 6.4.3.2-1, over the HTTP surface.
3318 : #[cfg(test)]
3319 : mod clause_5_6_17_and_6_4_3_2 {
3320 : use super::*;
3321 : use axum::body::Body;
3322 : use axum::http::Request;
3323 : use http_body_util::BodyExt;
3324 : use serde_json::json;
3325 : use tower::ServiceExt;
3326 :
3327 16 : fn app() -> axum::Router {
3328 16 : crate::router(AppState::new("antares-test".into()))
3329 16 : }
3330 :
3331 24 : async fn create(app: &axum::Router, body: Value) {
3332 24 : let payload = body.to_string();
3333 24 : let resp = app
3334 24 : .clone()
3335 24 : .oneshot(
3336 24 : Request::post("/ngsi-ld/v1/entities")
3337 24 : .header("Content-Type", "application/json")
3338 24 : .header("Content-Length", payload.len().to_string())
3339 24 : .body(Body::from(payload))
3340 24 : .expect("req"),
3341 24 : )
3342 24 : .await
3343 24 : .expect("resp");
3344 24 : assert_eq!(resp.status(), StatusCode::CREATED, "create failed");
3345 24 : }
3346 :
3347 12 : async fn patch(app: &axum::Router, uri: &str, body: Value) -> StatusCode {
3348 12 : let payload = body.to_string();
3349 12 : app.clone()
3350 12 : .oneshot(
3351 12 : Request::patch(uri)
3352 12 : .header("Content-Type", "application/json")
3353 12 : .header("Content-Length", payload.len().to_string())
3354 12 : .body(Body::from(payload))
3355 12 : .expect("req"),
3356 12 : )
3357 12 : .await
3358 12 : .expect("resp")
3359 12 : .status()
3360 12 : }
3361 :
3362 12 : async fn get(app: &axum::Router, uri: &str) -> Value {
3363 12 : let resp = app
3364 12 : .clone()
3365 12 : .oneshot(Request::get(uri).body(Body::empty()).expect("req"))
3366 12 : .await
3367 12 : .expect("resp");
3368 12 : let bytes = resp.into_body().collect().await.expect("body").to_bytes();
3369 12 : serde_json::from_slice(&bytes).expect("json")
3370 12 : }
3371 :
3372 : /// 5.6.17.3: "An optional parameter indicating a common observedAt
3373 : /// timestamp to use across merged Attributes." It is the timestamp of
3374 : /// every Attribute this merge touches; an Attribute the Fragment gives an
3375 : /// observedAt of its own keeps that one, and an Attribute the Fragment
3376 : /// does not mention is not a merged Attribute and is left alone.
3377 : #[tokio::test]
3378 4 : async fn merge_applies_the_common_observed_at() {
3379 4 : let app = app();
3380 4 : let id = "urn:ngsi-ld:Obs:1";
3381 4 : create(
3382 4 : &app,
3383 4 : json!({"id": id, "type": "T",
3384 4 : "a": {"type": "Property", "value": 1},
3385 4 : "b": {"type": "Property", "value": 1},
3386 4 : "untouched": {"type": "Property", "value": 1,
3387 4 : "observedAt": "2020-01-01T00:00:00Z"}}),
3388 4 : )
3389 4 : .await;
3390 4 : let uri = format!("/ngsi-ld/v1/entities/{id}?observedAt=2026-08-17T10:00:00Z");
3391 4 : assert_eq!(
3392 4 : patch(
3393 4 : &app,
3394 4 : &uri,
3395 4 : json!({"a": {"value": 2},
3396 4 : "b": {"value": 2, "observedAt": "2021-01-01T00:00:00Z"}}),
3397 4 : )
3398 4 : .await,
3399 : StatusCode::NO_CONTENT
3400 : );
3401 4 : let body = get(&app, &format!("/ngsi-ld/v1/entities/{id}")).await;
3402 4 : assert_eq!(
3403 4 : body["a"]["observedAt"], "2026-08-17T10:00:00Z",
3404 : "the merged Attribute takes the common timestamp: {body}"
3405 : );
3406 4 : assert_eq!(
3407 4 : body["b"]["observedAt"], "2021-01-01T00:00:00Z",
3408 : "a Fragment instance carrying its own observedAt keeps it: {body}"
3409 : );
3410 4 : assert_eq!(
3411 4 : body["untouched"]["observedAt"], "2020-01-01T00:00:00Z",
3412 4 : "an Attribute this merge does not touch is not restamped: {body}"
3413 4 : );
3414 4 : }
3415 :
3416 : /// 4.8 makes observedAt a DateTime, so a value that is not one cannot be
3417 : /// stamped across the merged Attributes.
3418 : #[tokio::test]
3419 4 : async fn merge_rejects_an_observed_at_that_is_not_a_datetime() {
3420 4 : let app = app();
3421 4 : let id = "urn:ngsi-ld:Obs:2";
3422 4 : create(&app, json!({"id": id, "type": "T"})).await;
3423 4 : assert_eq!(
3424 4 : patch(
3425 4 : &app,
3426 4 : &format!("/ngsi-ld/v1/entities/{id}?observedAt=yesterday"),
3427 4 : json!({"a": {"type": "Property", "value": 1}}),
3428 4 : )
3429 4 : .await,
3430 4 : StatusCode::BAD_REQUEST
3431 4 : );
3432 4 : }
3433 :
3434 : /// 5.6.17.4: "If a common language tag is defined and a LanguageProperty
3435 : /// Attribute to be merged is represented as a string, the pre-existing
3436 : /// languageMap JSON object shall be preserved. The string value shall
3437 : /// only replace the value associated to the language tag key found
3438 : /// within the languageMap."
3439 : #[tokio::test]
3440 4 : async fn merge_with_a_common_lang_replaces_only_that_language_key() {
3441 4 : let app = app();
3442 4 : let id = "urn:ngsi-ld:Lang:1";
3443 4 : create(
3444 4 : &app,
3445 4 : json!({"id": id, "type": "T",
3446 4 : "greeting": {"type": "LanguageProperty",
3447 4 : "languageMap": {"en": "hello", "es": "adios"}}}),
3448 4 : )
3449 4 : .await;
3450 4 : assert_eq!(
3451 4 : patch(
3452 4 : &app,
3453 4 : &format!("/ngsi-ld/v1/entities/{id}?lang=es"),
3454 4 : json!({"greeting": "hola"}),
3455 4 : )
3456 4 : .await,
3457 : StatusCode::NO_CONTENT
3458 : );
3459 4 : let body = get(&app, &format!("/ngsi-ld/v1/entities/{id}")).await;
3460 4 : assert_eq!(
3461 4 : body["greeting"]["languageMap"],
3462 4 : json!({"en": "hello", "es": "hola"}),
3463 : "the pre-existing languageMap survives, only the tagged key moves: {body}"
3464 : );
3465 4 : assert!(
3466 4 : body["greeting"].get("value").is_none(),
3467 : "the string never lands as a plain Property value: {body}"
3468 : );
3469 4 : assert_eq!(body["greeting"]["type"], "LanguageProperty", "{body}");
3470 4 : }
3471 :
3472 : /// Table 6.4.3.2-1 containedBy: "List of entity ids which have previously
3473 : /// been encountered whilst retrieving the Entity Graph" — 4.5.23.1 keeps
3474 : /// the walk from retrieving them a second time.
3475 : #[tokio::test]
3476 4 : async fn contained_by_ids_are_not_retrieved_again() {
3477 4 : let app = app();
3478 4 : let (root, held, fresh) = (
3479 4 : "urn:ngsi-ld:Graph:root",
3480 4 : "urn:ngsi-ld:Graph:held",
3481 4 : "urn:ngsi-ld:Graph:fresh",
3482 4 : );
3483 4 : create(&app, json!({"id": held, "type": "G"})).await;
3484 4 : create(&app, json!({"id": fresh, "type": "G"})).await;
3485 4 : create(
3486 4 : &app,
3487 4 : json!({"id": root, "type": "G",
3488 4 : "toHeld": {"type": "Relationship", "object": held},
3489 4 : "toFresh": {"type": "Relationship", "object": fresh}}),
3490 4 : )
3491 4 : .await;
3492 :
3493 4 : let body = get(
3494 4 : &app,
3495 4 : &format!("/ngsi-ld/v1/entities/{root}?join=flat&joinLevel=2&containedBy={held}"),
3496 4 : )
3497 4 : .await;
3498 4 : let arr = match body {
3499 4 : Value::Array(a) => a,
3500 0 : other => vec![other],
3501 : };
3502 8 : let ids: Vec<&str> = arr.iter().filter_map(|e| e["id"].as_str()).collect();
3503 4 : assert!(
3504 4 : !ids.contains(&held),
3505 : "an id the client already holds is not retrieved again: {ids:?}"
3506 : );
3507 4 : assert!(
3508 4 : ids.contains(&fresh),
3509 : "the Linked Entity it does not hold is still returned: {ids:?}"
3510 : );
3511 4 : assert!(ids.contains(&root), "the Linking Entity is there: {ids:?}");
3512 4 : }
3513 : }
3514 :
3515 : #[cfg(test)]
3516 : mod clause_4_8_system_attributes {
3517 : use super::*;
3518 : use crate::stamp::stamp_instances;
3519 : use axum::body::Body;
3520 : use axum::http::Request;
3521 : use serde_json::json;
3522 : use tower::ServiceExt;
3523 :
3524 : /// 4.8 stamps a sub-Attribute because "a sub-Property is a Property".
3525 : /// The members an Attribute instance carries under 4.5 are NOT
3526 : /// sub-Attributes, so none of them may be descended into and stamped —
3527 : /// several of them (`previousJson` and `previousVocab` hold uninterpreted
3528 : /// JSON per 4.5.20/4.5.21, `entityList` holds Linked Entities) can be an
3529 : /// array of objects, which is exactly the shape the walk mistakes for an
3530 : /// instance array. The list the expander keeps verbatim is the one list;
3531 : /// a second copy is what drifts.
3532 : #[test]
3533 4 : fn no_reserved_instance_member_is_walked_as_a_sub_attribute() {
3534 4 : let carrier = json!([{"probe": "untouched"}]);
3535 120 : for member in antares_jsonld::RESERVED_MEMBERS {
3536 : // the two the stamp itself writes; every other member is the
3537 : // instance's own and is left exactly as it was found
3538 120 : if matches!(*member, "createdAt" | "modifiedAt") {
3539 8 : continue;
3540 112 : }
3541 112 : let mut inst = json!({"type": "Property", "value": 1});
3542 112 : inst[*member] = carrier.clone();
3543 112 : let mut attr = json!([inst]);
3544 112 : stamp_instances(&mut attr, "2020-01-01T00:00:00Z");
3545 112 : let held = attr[0].get(*member).expect("the member survives");
3546 112 : assert_eq!(
3547 112 : held, &carrier,
3548 : "{member} was walked as a sub-Attribute and stamped"
3549 : );
3550 : }
3551 4 : }
3552 :
3553 : /// 4.8: createdAt is "the temporal Property at which the Entity,
3554 : /// Property or Relationship was entered into an NGSI-LD system" and
3555 : /// modifiedAt the one at which it "was last modified". Both are
3556 : /// generated by the system, at every level: an Entity, an Attribute
3557 : /// instance and a sub-Attribute (a sub-Property is a Property). A client
3558 : /// that could set them would rewrite the provenance of its own data, and
3559 : /// a subscriber filtering on modifiedAt would never see the write.
3560 : #[tokio::test]
3561 4 : async fn the_client_cannot_write_its_own_created_and_modified_stamps() {
3562 4 : let st = AppState::new("antares-sysattrs".into());
3563 4 : let forged = "1970-01-01T00:00:00Z";
3564 4 : let payload = json!({
3565 4 : "id": "urn:ngsi-ld:Vehicle:stamped",
3566 4 : "type": "Vehicle",
3567 4 : "createdAt": forged,
3568 4 : "modifiedAt": forged,
3569 4 : "speed": {
3570 4 : "type": "Property",
3571 4 : "value": 10,
3572 4 : "createdAt": forged,
3573 4 : "modifiedAt": forged,
3574 4 : "accuracy": {"type": "Property", "value": 1,
3575 4 : "createdAt": forged, "modifiedAt": forged},
3576 : },
3577 : })
3578 4 : .to_string();
3579 4 : let resp = crate::router(st.clone())
3580 4 : .oneshot(
3581 4 : Request::post("/ngsi-ld/v1/entities")
3582 4 : .header("Content-Type", "application/json")
3583 4 : .header("Content-Length", payload.len().to_string())
3584 4 : .body(Body::from(payload))
3585 4 : .expect("req"),
3586 4 : )
3587 4 : .await
3588 4 : .expect("resp");
3589 4 : assert_eq!(resp.status(), StatusCode::CREATED);
3590 :
3591 4 : let resp = crate::router(st)
3592 4 : .oneshot(
3593 4 : Request::get("/ngsi-ld/v1/entities/urn:ngsi-ld:Vehicle:stamped?options=sysAttrs")
3594 4 : .body(Body::empty())
3595 4 : .expect("req"),
3596 4 : )
3597 4 : .await
3598 4 : .expect("resp");
3599 4 : assert_eq!(resp.status(), StatusCode::OK);
3600 4 : let body = axum::body::to_bytes(resp.into_body(), usize::MAX)
3601 4 : .await
3602 4 : .expect("body");
3603 4 : let served = String::from_utf8_lossy(&body);
3604 4 : assert!(
3605 4 : served.contains("createdAt") && served.contains("modifiedAt"),
3606 : "sysAttrs serves the stamps: {served}"
3607 : );
3608 4 : assert!(
3609 4 : !served.contains(forged),
3610 4 : "no level of the document may carry the stamp the client sent: {served}"
3611 4 : );
3612 4 : }
3613 :
3614 : /// 4.8 again, on every route that writes an Attribute. Create is the
3615 : /// obvious way in; a client that is refused there and accepted on Append,
3616 : /// Merge, Partial Update or Replace has the same forged provenance one
3617 : /// request later. Each of the four carries the stamps at Entity,
3618 : /// Attribute and sub-Attribute level.
3619 : #[tokio::test]
3620 4 : async fn no_write_route_lets_the_client_stamp_its_own_attributes() {
3621 4 : let st = AppState::new("antares-sysattrs-writes".into());
3622 4 : let forged = "1970-01-01T00:00:00Z";
3623 4 : let id = "urn:ngsi-ld:Vehicle:writes";
3624 44 : let stamped = |v: Value| {
3625 44 : let mut o = v;
3626 44 : if let Some(m) = o.as_object_mut() {
3627 44 : m.insert("createdAt".into(), json!(forged));
3628 44 : m.insert("modifiedAt".into(), json!(forged));
3629 44 : }
3630 44 : o
3631 44 : };
3632 36 : let send = |req: Request<Body>| {
3633 36 : let st = st.clone();
3634 36 : async move { crate::router(st).oneshot(req).await.expect("resp") }
3635 36 : };
3636 :
3637 4 : let seed = json!({"id": id, "type": "Vehicle"}).to_string();
3638 4 : let resp = send(
3639 4 : Request::post("/ngsi-ld/v1/entities")
3640 4 : .header("Content-Type", "application/json")
3641 4 : .header("Content-Length", seed.len().to_string())
3642 4 : .body(Body::from(seed))
3643 4 : .expect("req"),
3644 4 : )
3645 4 : .await;
3646 4 : assert_eq!(resp.status(), StatusCode::CREATED);
3647 :
3648 16 : let attr = || {
3649 16 : stamped(json!({
3650 16 : "type": "Property",
3651 16 : "value": 10,
3652 16 : "accuracy": stamped(json!({"type": "Property", "value": 1})),
3653 16 : }))
3654 16 : };
3655 : // Append (5.6.3), Partial Update (5.6.4), Merge (5.6.17), Replace
3656 : // (5.6.16) — the whole write surface that reaches expand_entity.
3657 4 : let calls: Vec<(&str, String, Value)> = vec![
3658 4 : (
3659 4 : "POST",
3660 4 : format!("/ngsi-ld/v1/entities/{id}/attrs"),
3661 4 : stamped(json!({"speed": attr()})),
3662 4 : ),
3663 4 : (
3664 4 : "PATCH",
3665 4 : format!("/ngsi-ld/v1/entities/{id}/attrs/speed"),
3666 4 : attr(),
3667 4 : ),
3668 4 : (
3669 4 : "PATCH",
3670 4 : format!("/ngsi-ld/v1/entities/{id}"),
3671 4 : stamped(json!({"speed": attr()})),
3672 4 : ),
3673 4 : (
3674 4 : "PUT",
3675 4 : format!("/ngsi-ld/v1/entities/{id}"),
3676 4 : stamped(json!({"id": id, "type": "Vehicle", "speed": attr()})),
3677 4 : ),
3678 : ];
3679 16 : for (method, path, payload) in calls {
3680 16 : let body = payload.to_string();
3681 16 : let resp = send(
3682 16 : Request::builder()
3683 16 : .method(method)
3684 16 : .uri(&path)
3685 16 : .header("Content-Type", "application/json")
3686 16 : .header("Content-Length", body.len().to_string())
3687 16 : .body(Body::from(body))
3688 16 : .expect("req"),
3689 16 : )
3690 16 : .await;
3691 16 : assert!(
3692 16 : resp.status().is_success(),
3693 4 : "{method} {path} answered {}",
3694 4 : resp.status()
3695 4 : );
3696 16 : let resp = send(
3697 16 : Request::get(format!("/ngsi-ld/v1/entities/{id}?options=sysAttrs"))
3698 16 : .body(Body::empty())
3699 16 : .expect("req"),
3700 16 : )
3701 16 : .await;
3702 16 : let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
3703 16 : .await
3704 16 : .expect("body");
3705 16 : let served = String::from_utf8_lossy(&bytes);
3706 16 : assert!(
3707 16 : !served.contains(forged),
3708 4 : "{method} {path} let the client stamp the document: {served}"
3709 4 : );
3710 4 : }
3711 4 : }
3712 : }
|
