Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Attribute-level operations (5.6.2–5.6.5, 5.6.19; resources 6.6/6.7).
3 :
4 : use crate::federation::path_segment;
5 : use crate::negotiate::*;
6 : use crate::state::{now_iso, AppState};
7 : use antares_jsonld::{expand_entity, ExpandOpts};
8 : use antares_model::{check_attr_name, NgsiError};
9 : use antares_store::CurrentStateDriverExt;
10 : use antares_store::Kind;
11 : use axum::body::Bytes;
12 : use axum::extract::{Path, State};
13 : use axum::http::HeaderMap;
14 : use axum::response::{IntoResponse, Response};
15 : use serde_json::{Map, Value};
16 : use std::collections::HashMap;
17 :
18 : use crate::negotiate::CleanParams;
19 :
20 : /// The fully qualified name of the Entity `scope` member (core @context).
21 : /// The stored document keeps it under the short name, so the attribute
22 : /// operations that address it (5.6.4.4, 5.6.5.4, 5.6.19.4) compare against
23 : /// both spellings.
24 : const SCOPE_IRI: &str = "https://uri.etsi.org/ngsi-ld/scope";
25 :
26 : /// Outcome of a multi-attribute write: 204 when everything applied, else 207
27 : /// with an UpdateResult (5.2.18).
28 136 : fn update_result(
29 136 : tenant: &antares_model::TenantId,
30 136 : updated: Vec<String>,
31 136 : not_updated: Vec<(String, String)>,
32 136 : ) -> Response {
33 136 : if not_updated.is_empty() {
34 128 : return no_content(tenant);
35 8 : }
36 : // Attribute names are the expanded IRIs the write worked on (5.5.7);
37 : // the Entity core members keep their reserved names `type` and `scope`.
38 8 : let payload = serde_json::json!({
39 8 : "updated": updated,
40 8 : "notUpdated": not_updated
41 8 : .iter()
42 8 : .map(|(a, r)| serde_json::json!({
43 8 : "attributeName": a,
44 8 : "reason": r,
45 : }))
46 8 : .collect::<Vec<_>>(),
47 : });
48 8 : multi_status(payload, tenant)
49 136 : }
50 :
51 : // ---------- POST /entities/{id}/attrs/ — Append (5.6.3) ----------
52 :
53 118 : pub async fn append_attrs(
54 118 : State(st): State<AppState>,
55 118 : Path(id): Path<String>,
56 118 : CleanParams(params): CleanParams,
57 118 : headers: HeaderMap,
58 118 : body: Bytes,
59 118 : ) -> Response {
60 118 : match append_attrs_inner(&st, &id, ¶ms, &headers, &body).await {
61 90 : Ok(r) => r,
62 28 : Err(e) => e.into_response(),
63 : }
64 118 : }
65 :
66 : /// What separates 5.6.3 Append Attributes from 5.6.2 Update Attributes: the
67 : /// body kind each accepts, whether NGSI-LD Null may appear in it (5.5.4:
68 : /// only the update form carries deletions), and the operation name and
69 : /// method a forward to a registered Context Source carries. Everything else
70 : /// -- id and parameter validation, expansion, the registration plan, the
71 : /// loop guard, which half is served locally, the multi-status assembly --
72 : /// is one operation, written once in `write_attrs`.
73 : struct AttrWrite {
74 : body_kind: BodyKind,
75 : allow_null: bool,
76 : op: &'static str,
77 : method: reqwest::Method,
78 : /// The clause the policy seam is asked about, since one function serves
79 : /// both of these operations.
80 : clause: &'static str,
81 : }
82 :
83 : const APPEND: AttrWrite = AttrWrite {
84 : body_kind: BodyKind::Standard,
85 : allow_null: false,
86 : op: "appendAttrs",
87 : method: reqwest::Method::POST,
88 : clause: "5.6.3",
89 : };
90 :
91 : const UPDATE: AttrWrite = AttrWrite {
92 : body_kind: BodyKind::MergePatch,
93 : allow_null: true,
94 : op: "updateEntity",
95 : method: reqwest::Method::PATCH,
96 : clause: "5.6.2",
97 : };
98 :
99 : /// Why an attribute write never reached the store. `Untouched` is not a
100 : /// failure: 5.6.3.4 leaves an Attribute that may not be overwritten
101 : /// "untouched", and a write that applied nothing modified nothing -- there is
102 : /// no `modifiedAt` to move (4.8) and no change to hand to the store.
103 : enum Unwritten {
104 : Failed(NgsiError),
105 : Untouched,
106 : }
107 :
108 : impl From<NgsiError> for Unwritten {
109 16 : fn from(e: NgsiError) -> Self {
110 16 : Self::Failed(e)
111 16 : }
112 : }
113 :
114 : /// One entity-level attribute write. `merge` is the clause's own algorithm
115 : /// for the Entity Fragment's `scope` and its Attributes; the Entity Type
116 : /// union above it is the same rule in both clauses ("added to the list of
117 : /// Entity Type names of the target Entity").
118 218 : async fn write_attrs(
119 218 : st: &AppState,
120 218 : id: &str,
121 218 : params: &HashMap<String, String>,
122 218 : headers: &HeaderMap,
123 218 : body: &[u8],
124 218 : mode: &AttrWrite,
125 218 : merge: impl FnOnce(
126 218 : &mut Map<String, Value>,
127 218 : &Map<String, Value>,
128 218 : &str,
129 218 : &mut Vec<String>,
130 218 : &mut Vec<(String, String)>,
131 218 : ) + Send,
132 218 : ) -> ApiResult<Response> {
133 218 : let tenant = tenant_from(headers)?;
134 218 : antares_model::EntityId::new(id)?;
135 218 : check_params(params, &["options", "local", "type"])?;
136 218 : let parsed = parse_body(&st.loader, headers, body, mode.body_kind).await?;
137 202 : let obj = parsed.object(NgsiError::BadRequestData(
138 202 : "fragment must be a JSON object".into(),
139 202 : ))?;
140 202 : gate!(st, &tenant, headers, mode.clause, ids: &[id]).await?;
141 190 : let fragment = expand_entity(
142 190 : obj,
143 190 : &parsed.ctx,
144 190 : ExpandOpts {
145 190 : fragment: true,
146 190 : allow_null: mode.allow_null,
147 190 : temporal: false,
148 190 : ..Default::default()
149 190 : },
150 0 : )?;
151 190 : let (plan, all_attr_iris) =
152 190 : attr_fed_plan(st, &tenant, id, &fragment, &parsed.ctx, params, headers).await?;
153 190 : let regs = match plan {
154 0 : crate::federation::WritePlan::Answered(r) => return Ok(*r),
155 190 : crate::federation::WritePlan::Forward(regs) => regs,
156 : };
157 190 : let local_covered = proxies_cover_all(®s, &all_attr_iris);
158 190 : let fragment = crate::federation::strip_covered_expanded(&fragment, ®s);
159 190 : let local_iris = attr_iris_of(&fragment);
160 190 : let ts = now_iso();
161 190 : let mut updated = Vec::new();
162 190 : let mut not_updated = Vec::new();
163 190 : let local_resp: Option<ApiResult<Response>> = if local_covered {
164 18 : None
165 : } else {
166 172 : let res = st
167 172 : .store
168 172 : .mutate(&tenant, Kind::Entity, id, |doc| {
169 : // 5.6.2.4 / 5.6.3.4: the ?type selector narrows the target — a
170 : // mismatch means the entity is not known for this operation.
171 152 : if !matches_type_param(doc, params, &parsed.ctx) {
172 16 : return Err(NgsiError::ResourceNotFound(format!(
173 16 : "entity {id} does not match the type selector"
174 16 : ))
175 16 : .into());
176 136 : }
177 136 : let target = antares_store::stored_object(doc)?;
178 136 : let frag = antares_jsonld::expanded_object(&fragment)?;
179 : // 5.6.2.4 / 5.6.3.4: Entity Type names not yet in the target are
180 : // added to its list
181 136 : if let Some(new_types) = frag.get("type").and_then(Value::as_array) {
182 2 : let mut cur: Vec<Value> = target
183 2 : .get("type")
184 2 : .and_then(Value::as_array)
185 2 : .cloned()
186 2 : .unwrap_or_default();
187 2 : let known = cur.len();
188 2 : for t in new_types {
189 2 : if !cur.contains(t) {
190 0 : cur.push(t.clone());
191 2 : }
192 : }
193 2 : if cur.len() > known {
194 0 : target.insert("type".into(), Value::Array(cur));
195 0 : updated.push("type".into());
196 2 : }
197 134 : }
198 136 : merge(target, frag, &ts, &mut updated, &mut not_updated);
199 136 : if updated.is_empty() {
200 6 : return Err(Unwritten::Untouched);
201 130 : }
202 130 : target.insert("modifiedAt".into(), Value::String(ts.clone()));
203 130 : Ok::<(), Unwritten>(())
204 152 : })
205 172 : .await?;
206 22 : Some(match res {
207 20 : None => Err(NgsiError::ResourceNotFound(format!("entity {id} not found")).into()),
208 16 : Some(Err(Unwritten::Failed(e))) => Err(e.into()),
209 : Some(Ok(())) | Some(Err(Unwritten::Untouched)) => {
210 136 : Ok(update_result(&tenant, updated.clone(), not_updated.clone()))
211 : }
212 : })
213 : };
214 190 : if regs.is_empty() {
215 : // the local half is skipped only while registrations cover the
216 : // attributes, so an empty list always leaves a local response
217 170 : return local_resp
218 170 : .unwrap_or_else(|| Err(NgsiError::InternalError("no local result".into()).into()));
219 20 : }
220 20 : let local_outcome = classify_local(&local_resp);
221 20 : let query = fwd_query(params, &["options", "type"]);
222 20 : let fed_parts = crate::federation::fed_attr_parts(
223 20 : st,
224 20 : headers,
225 20 : &tenant,
226 20 : &parsed.ctx.source,
227 20 : ®s,
228 20 : mode.op,
229 20 : mode.method.clone(),
230 20 : &format!("/entities/{}/attrs/", path_segment(id)),
231 20 : &query,
232 20 : Some(Value::Object(without_context_map(obj))),
233 20 : )
234 20 : .await;
235 20 : Ok(combine_attr_parts(
236 20 : &tenant,
237 20 : &all_attr_iris,
238 20 : &local_iris,
239 20 : local_outcome,
240 20 : updated,
241 20 : not_updated.into_iter().map(|(a, r)| (a, r, None)).collect(),
242 20 : ®s,
243 20 : &fed_parts,
244 : ))
245 218 : }
246 :
247 : /// 5.6.3.4 Append Attributes: an Attribute the target does not have is
248 : /// appended; one it has is replaced datasetId-wise unless `noOverwrite` was
249 : /// asked for, in which case it is reported in `notUpdated`.
250 118 : async fn append_attrs_inner(
251 118 : st: &AppState,
252 118 : id: &str,
253 118 : params: &HashMap<String, String>,
254 118 : headers: &HeaderMap,
255 118 : body: &[u8],
256 118 : ) -> ApiResult<Response> {
257 118 : let no_overwrite = params
258 118 : .get("options")
259 118 : .is_some_and(|o| o.split(',').any(|s| s.trim() == "noOverwrite"));
260 118 : write_attrs(
261 118 : st,
262 118 : id,
263 118 : params,
264 118 : headers,
265 118 : body,
266 118 : &APPEND,
267 82 : |target, frag, ts, updated, not_updated| {
268 : // appended scope: overwrite replaces; noOverwrite unions (010_07)
269 82 : if let Some(new_scope) = frag.get("scope") {
270 0 : if target.contains_key("scope") && no_overwrite {
271 0 : let mut cur: Vec<Value> = target
272 0 : .get("scope")
273 0 : .and_then(Value::as_array)
274 0 : .cloned()
275 0 : .unwrap_or_default();
276 0 : let known = cur.len();
277 0 : for sc in new_scope.as_array().cloned().unwrap_or_default() {
278 0 : if !cur.contains(&sc) {
279 0 : cur.push(sc);
280 0 : }
281 : }
282 0 : if cur.len() > known {
283 0 : target.insert("scope".into(), Value::Array(cur));
284 0 : updated.push("scope".into());
285 0 : }
286 0 : } else {
287 0 : target.insert("scope".into(), new_scope.clone());
288 0 : updated.push("scope".into());
289 0 : }
290 82 : }
291 86 : for (k, v) in frag {
292 86 : if is_fragment_meta(k) {
293 2 : continue;
294 84 : }
295 84 : let mut incoming = v.clone();
296 84 : crate::stamp::stamp_instances(&mut incoming, ts);
297 84 : match target.get_mut(k) {
298 72 : None => {
299 72 : target.insert(k.clone(), incoming);
300 72 : updated.push(k.clone());
301 72 : }
302 12 : Some(existing) => {
303 12 : let merged = merge_instance_sets(existing, &incoming, no_overwrite);
304 12 : if merged {
305 4 : updated.push(k.clone());
306 8 : } else {
307 8 : not_updated
308 8 : .push((k.clone(), "attribute already exists (noOverwrite)".into()));
309 8 : }
310 : }
311 : }
312 : }
313 82 : },
314 : )
315 118 : .await
316 118 : }
317 :
318 : /// 5.6.2.4 Update Attributes: every Attribute of the Fragment is applied —
319 : /// an unknown one is appended silently (011_01_03), an NGSI-LD Null deletes
320 : /// the instance it matches (5.5.8), and an Attribute left with no instances
321 : /// leaves the Entity.
322 100 : async fn update_attrs_inner(
323 100 : st: &AppState,
324 100 : id: &str,
325 100 : params: &HashMap<String, String>,
326 100 : headers: &HeaderMap,
327 100 : body: &[u8],
328 100 : ) -> ApiResult<Response> {
329 100 : write_attrs(
330 100 : st,
331 100 : id,
332 100 : params,
333 100 : headers,
334 100 : body,
335 100 : &UPDATE,
336 54 : |target, frag, ts, updated, not_updated| {
337 : // 5.6.2: scope updates only when the entity already has one
338 54 : if let Some(new_scope) = frag.get("scope") {
339 0 : if target.contains_key("scope") {
340 0 : target.insert("scope".into(), new_scope.clone());
341 0 : updated.push("scope".into());
342 0 : } else {
343 0 : not_updated.push(("scope".into(), "entity has no scope".into()));
344 0 : }
345 54 : }
346 54 : for (k, v) in frag {
347 54 : if is_fragment_meta(k) {
348 0 : continue;
349 54 : }
350 54 : let mut incoming = v.clone();
351 54 : crate::stamp::stamp_instances(&mut incoming, ts);
352 54 : match target.get_mut(k) {
353 : // 5.6.2 + 011_01_03: unknown attributes are appended silently
354 : None => {
355 0 : let live: Vec<Value> = incoming
356 0 : .as_array()
357 0 : .cloned()
358 0 : .unwrap_or_default()
359 0 : .into_iter()
360 0 : .filter(|i| !antares_jsonld::is_deletion_instance(i))
361 0 : .collect();
362 0 : if !live.is_empty() {
363 0 : target.insert(k.clone(), Value::Array(live));
364 0 : }
365 0 : updated.push(k.clone());
366 : }
367 54 : Some(existing) => {
368 54 : merge_instance_sets(existing, &incoming, false);
369 54 : if existing.as_array().is_some_and(Vec::is_empty) {
370 4 : target.remove(k);
371 50 : }
372 54 : updated.push(k.clone());
373 : }
374 : }
375 : }
376 54 : },
377 : )
378 100 : .await
379 100 : }
380 :
381 : /// The Entity Fragment members that are not Attributes: handled by the
382 : /// write itself (`id`, `type`, `scope`) or server-generated (4.8).
383 140 : fn is_fragment_meta(k: &str) -> bool {
384 140 : matches!(k, "id" | "type" | "scope" | "createdAt" | "modifiedAt")
385 140 : }
386 :
387 : /// Shared federation plan for attribute writes: the matching non-aux
388 : /// registrations, plus the touched attribute IRIs they were matched on.
389 190 : async fn attr_fed_plan(
390 190 : st: &AppState,
391 190 : tenant: &antares_model::TenantId,
392 190 : id: &str,
393 190 : fragment: &Value,
394 190 : ctx: &antares_jsonld::Context,
395 190 : params: &HashMap<String, String>,
396 190 : headers: &axum::http::HeaderMap,
397 190 : ) -> Result<(crate::federation::WritePlan, Vec<String>), NgsiError> {
398 190 : let attr_iris = attr_iris_of(fragment);
399 190 : let plan = attr_fed_plan_iris(st, tenant, id, &attr_iris, ctx, params, headers).await?;
400 190 : Ok((plan, attr_iris))
401 190 : }
402 :
403 400 : async fn attr_fed_plan_iris(
404 400 : st: &AppState,
405 400 : tenant: &antares_model::TenantId,
406 400 : id: &str,
407 400 : attr_iris: &[String],
408 400 : ctx: &antares_jsonld::Context,
409 400 : params: &HashMap<String, String>,
410 400 : headers: &axum::http::HeaderMap,
411 400 : ) -> Result<crate::federation::WritePlan, NgsiError> {
412 400 : let spec = crate::registry::CsrSpec {
413 400 : ids: Some(vec![id.to_owned()]),
414 400 : attrs: (!attr_iris.is_empty()).then(|| attr_iris.to_vec()),
415 400 : ..Default::default()
416 : };
417 400 : crate::federation::write_plan(st, tenant, &spec, ctx, params, headers).await
418 400 : }
419 :
420 : /// The registrations an attribute-level operation has to consider, with the
421 : /// 6.3.18 loop guard already applied. A returned response is the loop
422 : /// chain's own answer (Table 6.3.18-2) and ends the operation before
423 : /// anything local happens.
424 210 : async fn attr_regs_or_loop(
425 210 : st: &AppState,
426 210 : tenant: &antares_model::TenantId,
427 210 : id: &str,
428 210 : attr_iri: &str,
429 210 : ctx: &antares_jsonld::Context,
430 210 : params: &HashMap<String, String>,
431 210 : headers: &HeaderMap,
432 210 : ) -> Result<(Vec<crate::federation::FedReg>, Option<Response>), NgsiError> {
433 : Ok(
434 210 : match attr_fed_plan_iris(st, tenant, id, &[attr_iri.to_owned()], ctx, params, headers)
435 210 : .await?
436 : {
437 0 : crate::federation::WritePlan::Answered(r) => (Vec::new(), Some(*r)),
438 210 : crate::federation::WritePlan::Forward(regs) => (regs, None),
439 : },
440 : )
441 210 : }
442 :
443 : /// The local half of a distributed attribute write is skipped only when every
444 : /// touched attribute is held by an exclusive/redirect registration (5.6.2.4
445 : /// and siblings). The decision is taken AFTER 6.3.18 loop handling: a `Via`
446 : /// chain naming this broker removes registrations from matching (Table
447 : /// 6.3.18-2), and the operation then has to be served locally.
448 400 : fn proxies_cover_all(regs: &[crate::federation::FedReg], attr_iris: &[String]) -> bool {
449 400 : !regs.is_empty()
450 54 : && !attr_iris.is_empty()
451 54 : && attr_iris
452 54 : .iter()
453 54 : .all(|a| regs.iter().any(|r| r.is_proxy() && r.covers_attr(a)))
454 400 : }
455 :
456 : /// The URL parameters that travel with a forwarded attribute operation.
457 : /// `type` narrows the target Entity for the registered source exactly as it
458 : /// does locally — 5.6.5.4 identifies the target by its "id (URI), and where
459 : /// specified type", and the parameter is shall-support on each of these
460 : /// resources (Tables 6.6.3.1-1, 6.6.3.2-1, 6.7.3.1-1, 6.7.3.2-1, 6.7.3.3-1).
461 : /// `local` never travels: Table 6.3.18-1 scopes it to the receiving broker.
462 54 : fn fwd_query(params: &HashMap<String, String>, keys: &[&str]) -> Vec<(String, String)> {
463 54 : keys.iter()
464 106 : .filter_map(|k| params.get(*k).map(|v| ((*k).to_owned(), v.clone())))
465 54 : .collect()
466 54 : }
467 :
468 : /// Attribute IRIs of an expanded fragment (entity meta members excluded).
469 380 : fn attr_iris_of(fragment: &Value) -> Vec<String> {
470 380 : fragment
471 380 : .as_object()
472 380 : .map(|o| {
473 380 : o.keys()
474 380 : .filter(|k| {
475 4 : !matches!(
476 370 : k.as_str(),
477 370 : "id" | "type" | "scope" | "createdAt" | "modifiedAt"
478 : )
479 370 : })
480 380 : .cloned()
481 380 : .collect()
482 380 : })
483 380 : .unwrap_or_default()
484 380 : }
485 :
486 : /// Fold the buffered local response into a `LocalOutcome` without losing the
487 : /// ProblemDetails reason.
488 54 : fn classify_local(resp: &Option<ApiResult<Response>>) -> LocalOutcome {
489 0 : match resp {
490 52 : None => LocalOutcome::Skipped,
491 2 : Some(Ok(_)) => LocalOutcome::Ok,
492 0 : Some(Err(ApiError::Ngsi(e))) => {
493 0 : let pd = e.to_problem_details();
494 0 : if pd.status == 404 {
495 0 : LocalOutcome::NotFound(pd.detail)
496 : } else {
497 0 : LocalOutcome::Failed(pd.detail)
498 : }
499 : }
500 0 : Some(Err(_)) => LocalOutcome::Failed("local operation failed".into()),
501 : }
502 54 : }
503 :
504 : /// How the LOCAL half of a distributed /attrs operation ended.
505 : enum LocalOutcome {
506 : /// proxies cover every touched attribute — no local write attempted
507 : Skipped,
508 : /// entity (or the addressed attribute) unknown locally
509 : NotFound(String),
510 : /// local write failed for another reason
511 : Failed(String),
512 : Ok,
513 : }
514 :
515 : /// 6.3.17: a distributed /attrs operation answers 204, 404, or **207 with an
516 : /// UpdateResult** (Tables 6.6.3.1-2, 6.6.3.2-2, 6.7.3.1-2, 6.7.3.2-2,
517 : /// 6.7.3.3-2; 5.2.18 applies "regardless of whether local or distributed")
518 : /// — never the batch {success, errors} shape. Per-registration failures are
519 : /// listed per covered attribute with `registrationId` (5.2.19).
520 : #[allow(clippy::too_many_arguments)] // one param per input of the 6.3.17 answer
521 94 : fn combine_attr_parts(
522 94 : tenant: &antares_model::TenantId,
523 94 : attr_iris: &[String],
524 94 : local_iris: &[String],
525 94 : local: LocalOutcome,
526 94 : mut updated: Vec<String>,
527 94 : mut not_updated: Vec<(String, String, Option<String>)>,
528 94 : regs: &[crate::federation::FedReg],
529 94 : fed_parts: &[crate::federation::Part],
530 94 : ) -> Response {
531 94 : match &local {
532 4 : LocalOutcome::NotFound(d) | LocalOutcome::Failed(d) => {
533 4 : for a in local_iris {
534 4 : not_updated.push((a.clone(), d.clone(), None));
535 4 : }
536 : }
537 90 : _ => {}
538 : }
539 94 : let mut any_fed_ok = false;
540 98 : for (reg, part) in regs.iter().zip(fed_parts) {
541 : // status 0: inclusive registration skipped for lack of operation
542 : // support (5.6.2.4) — neither a success nor a failure branch.
543 98 : if part.status == 0 {
544 2 : continue;
545 96 : }
546 100 : let covered: Vec<&String> = attr_iris.iter().filter(|a| reg.covers_attr(a)).collect();
547 96 : if part.ok() {
548 58 : any_fed_ok = true;
549 58 : for a in covered {
550 58 : if !updated.iter().any(|u| u == a.as_str()) {
551 54 : updated.push(a.clone());
552 54 : }
553 : }
554 : } else {
555 38 : for a in covered {
556 38 : not_updated.push((a.clone(), part.detail.clone(), Some(reg.reg_id.clone())));
557 38 : }
558 : }
559 : }
560 : // 6.3.17: "In the case of an exclusive or redirect registration, where
561 : // all of the data is held outside of the Context Broker and held in a
562 : // single registered source ... 508 Loop Detected" — a proxy part's loop
563 : // verdict passes through instead of dissolving into 207/404.
564 94 : if updated.is_empty()
565 30 : && !any_fed_ok
566 30 : && !matches!(local, LocalOutcome::Ok)
567 30 : && regs
568 30 : .iter()
569 30 : .zip(fed_parts)
570 30 : .any(|(r, p)| r.is_proxy() && p.status == 508)
571 : {
572 0 : return crate::federation::loop_508(tenant);
573 94 : }
574 : // nothing was found anywhere → 404 ProblemDetails (6.6/6.7 tables)
575 94 : if matches!(&local, LocalOutcome::NotFound(_)) && !any_fed_ok && updated.is_empty() {
576 4 : if let LocalOutcome::NotFound(d) = local {
577 4 : return ApiError::from(NgsiError::ResourceNotFound(d)).into_response();
578 0 : }
579 90 : }
580 : // 5.6.2.4: unsupported operation on a proxy registration is "an error of
581 : // type Conflict if the complete update failed" — nothing updated
582 : // anywhere and at least one registration refused the operation.
583 90 : if updated.is_empty()
584 26 : && !matches!(local, LocalOutcome::Ok)
585 26 : && !any_fed_ok
586 26 : && fed_parts
587 26 : .iter()
588 26 : .any(|p| p.detail.contains("does not accept"))
589 : {
590 2 : let mut resp = (
591 2 : axum::http::StatusCode::CONFLICT,
592 2 : [(axum::http::header::CONTENT_TYPE, "application/json")],
593 2 : axum::Json(serde_json::json!({
594 2 : "type": "https://uri.etsi.org/ngsi-ld/errors/Conflict",
595 2 : "title": "Conflict",
596 2 : "detail": "registration does not accept the operation",
597 2 : "status": 409,
598 2 : })),
599 2 : )
600 2 : .into_response();
601 2 : crate::negotiate::echo_tenant(tenant, &mut resp);
602 2 : return resp;
603 88 : }
604 : // 6.3.17: "In the case of an exclusive or redirect registration, where all
605 : // of the data is held outside of the Context Broker and held in a single
606 : // registered source, the following errors shall be returned: 508 Loop
607 : // Detected … 504 Gateway Timeout … 404 Not Found … 502 Bad Gateway."
608 : // 207 Multi Status is the answer for an entity distributed over multiple
609 : // endpoints, so a lone proxied source's failure passes through as itself.
610 88 : if let ([reg], [part]) = (regs, fed_parts) {
611 84 : if reg.is_proxy() && !part.ok() && matches!(local, LocalOutcome::Skipped) {
612 24 : return crate::federation::combine(
613 24 : vec![crate::federation::Part {
614 24 : status: part.status,
615 24 : detail: part.detail.clone(),
616 24 : }],
617 24 : no_content(tenant),
618 24 : tenant,
619 : );
620 60 : }
621 4 : }
622 64 : if not_updated.is_empty() {
623 56 : return no_content(tenant);
624 8 : }
625 8 : let nu: Vec<Value> = not_updated
626 8 : .into_iter()
627 8 : .map(|(a, r, reg_id)| {
628 8 : let mut m = Map::new();
629 8 : m.insert("attributeName".into(), Value::String(a));
630 8 : m.insert("reason".into(), Value::String(r));
631 8 : if let Some(rid) = reg_id.filter(|r| !r.is_empty()) {
632 8 : m.insert("registrationId".into(), Value::String(rid));
633 8 : }
634 8 : Value::Object(m)
635 8 : })
636 8 : .collect();
637 8 : multi_status(
638 8 : serde_json::json!({"updated": updated, "notUpdated": nu}),
639 8 : tenant,
640 : )
641 94 : }
642 :
643 : /// The local half's answer for an operation that addresses exactly ONE
644 : /// Attribute (5.6.4 Partial Update, 5.6.5 Delete, 5.6.19 Replace): per
645 : /// Table 6.3.2-1 the Entity may be absent (404 naming the Entity), present
646 : /// without the addressed Attribute (404 naming the Attribute), or edited
647 : /// (204). `found` is what the edit itself reports.
648 176 : fn single_attr_local(
649 176 : res: Option<Result<(), NgsiError>>,
650 176 : found: bool,
651 176 : id: &str,
652 176 : attr: &str,
653 176 : tenant: &antares_model::TenantId,
654 176 : ) -> ApiResult<Response> {
655 124 : match res {
656 28 : None => Err(NgsiError::ResourceNotFound(format!("entity {id} not found")).into()),
657 24 : Some(Err(e)) => Err(e.into()),
658 118 : Some(Ok(())) if found => Ok(no_content(tenant)),
659 : Some(Ok(())) => {
660 6 : Err(NgsiError::ResourceNotFound(format!("attribute {attr} not found")).into())
661 : }
662 : }
663 176 : }
664 :
665 : /// The multi-status assembly for the same three operations: one Attribute is
666 : /// addressed, so the local and distributed halves are combined over a
667 : /// one-element attribute list, and the Attribute counts as updated only when
668 : /// the local half actually applied it. `local_covered` means an
669 : /// exclusive/redirect registration holds it and there was no local half.
670 34 : fn combine_single_attr(
671 34 : tenant: &antares_model::TenantId,
672 34 : attr_iri: &str,
673 34 : local_covered: bool,
674 34 : local_resp: &Option<ApiResult<Response>>,
675 34 : regs: &[crate::federation::FedReg],
676 34 : fed_parts: &[crate::federation::Part],
677 34 : ) -> Response {
678 34 : let local_outcome = classify_local(local_resp);
679 34 : let all_attr_iris = vec![attr_iri.to_owned()];
680 34 : let local_iris = if local_covered {
681 34 : Vec::new()
682 : } else {
683 0 : all_attr_iris.clone()
684 : };
685 34 : let updated = if matches!(local_outcome, LocalOutcome::Ok) {
686 0 : all_attr_iris.clone()
687 : } else {
688 34 : Vec::new()
689 : };
690 34 : combine_attr_parts(
691 34 : tenant,
692 34 : &all_attr_iris,
693 34 : &local_iris,
694 34 : local_outcome,
695 34 : updated,
696 34 : Vec::new(),
697 34 : regs,
698 34 : fed_parts,
699 : )
700 34 : }
701 :
702 : /// Merge incoming instances into an existing instance array by datasetId.
703 : /// Deletion-marker instances (urn:ngsi-ld:null) remove the matched instance.
704 : /// Returns false when nothing was applied (noOverwrite and all existed).
705 78 : fn merge_instance_sets(existing: &mut Value, incoming: &Value, no_overwrite: bool) -> bool {
706 78 : let (Some(cur), Some(inc)) = (existing.as_array_mut(), incoming.as_array()) else {
707 0 : return false;
708 : };
709 78 : let mut any = false;
710 86 : for ni in inc {
711 86 : let ds = ni.get("datasetId").and_then(Value::as_str);
712 86 : let pos = cur
713 86 : .iter()
714 94 : .position(|ci| ci.get("datasetId").and_then(Value::as_str) == ds);
715 86 : if antares_jsonld::is_deletion_instance(ni) {
716 8 : if let Some(p) = pos {
717 8 : cur.remove(p);
718 8 : any = true;
719 8 : }
720 8 : continue;
721 78 : }
722 78 : match pos {
723 66 : Some(p) => {
724 66 : if !no_overwrite {
725 : // keep original createdAt
726 54 : let created = cur[p].get("createdAt").cloned();
727 54 : cur[p] = ni.clone();
728 54 : if let (Some(o), Some(c)) = (cur[p].as_object_mut(), created) {
729 54 : o.insert("createdAt".into(), c);
730 54 : }
731 54 : any = true;
732 12 : }
733 : }
734 12 : None => {
735 12 : cur.push(ni.clone());
736 12 : any = true;
737 12 : }
738 : }
739 : }
740 78 : any
741 78 : }
742 :
743 : // ---------- PATCH /entities/{id}/attrs/ — Update (5.6.2) ----------
744 :
745 100 : pub async fn update_attrs(
746 100 : State(st): State<AppState>,
747 100 : Path(id): Path<String>,
748 100 : CleanParams(params): CleanParams,
749 100 : headers: HeaderMap,
750 100 : body: Bytes,
751 100 : ) -> Response {
752 100 : match update_attrs_inner(&st, &id, ¶ms, &headers, &body).await {
753 64 : Ok(r) => r,
754 36 : Err(e) => e.into_response(),
755 : }
756 100 : }
757 :
758 : // ---------- PATCH /entities/{id}/attrs/{attrId} — Partial update (5.6.4) ----------
759 :
760 102 : pub async fn partial_update_attr(
761 102 : State(st): State<AppState>,
762 102 : Path((id, attr)): Path<(String, String)>,
763 102 : CleanParams(params): CleanParams,
764 102 : headers: HeaderMap,
765 102 : body: Bytes,
766 102 : ) -> Response {
767 102 : match partial_update_inner(&st, &id, &attr, ¶ms, &headers, &body).await {
768 46 : Ok(r) => r,
769 56 : Err(e) => e.into_response(),
770 : }
771 102 : }
772 :
773 102 : async fn partial_update_inner(
774 102 : st: &AppState,
775 102 : id: &str,
776 102 : attr: &str,
777 102 : params: &HashMap<String, String>,
778 102 : headers: &HeaderMap,
779 102 : body: &[u8],
780 102 : ) -> ApiResult<Response> {
781 102 : let tenant = tenant_from(headers)?;
782 102 : antares_model::EntityId::new(id)?;
783 102 : check_attr_name(attr)?;
784 86 : check_params(params, &["local", "type"])?;
785 86 : let parsed = parse_body(&st.loader, headers, body, BodyKind::MergePatch).await?;
786 82 : let obj = parsed.object(NgsiError::BadRequestData(
787 82 : "fragment must be a JSON object".into(),
788 82 : ))?;
789 82 : gate!(st, &tenant, headers, "5.6.4", ids: &[id]).await?;
790 82 : let frag_inst = antares_jsonld::expand_attr_fragment(obj, &parsed.ctx)?;
791 : // 5.6.4.4: "Apply term expansion as mandated by clause 5.5.7, so that
792 : // the fully qualified name (URI) associated to the target Attribute is
793 : // properly obtained" — a path name with no fully qualified name is not a
794 : // valid Attribute name, and lands on a member of the stored document
795 : // that is not an Attribute.
796 78 : let attr_iri = antares_jsonld::expand_attr_name(attr, &parsed.ctx)?;
797 : // 5.6.4.4: "If the target Attribute is scope, then an error of type
798 : // BadRequestData shall be raised."
799 74 : if attr == "scope" || attr_iri == SCOPE_IRI {
800 8 : return Err(NgsiError::BadRequestData(
801 8 : "scope cannot be the target of a partial attribute update (5.6.4)".into(),
802 8 : )
803 8 : .into());
804 66 : }
805 66 : let (regs, loop_answer) =
806 66 : attr_regs_or_loop(st, &tenant, id, &attr_iri, &parsed.ctx, params, headers).await?;
807 66 : if let Some(r) = loop_answer {
808 0 : return Ok(r);
809 66 : }
810 66 : let local_covered = proxies_cover_all(®s, std::slice::from_ref(&attr_iri));
811 66 : let want_ds = frag_inst
812 66 : .get("datasetId")
813 66 : .and_then(Value::as_str)
814 66 : .map(String::from);
815 66 : let is_deletion = antares_jsonld::is_deletion_instance(&frag_inst);
816 66 : let ts = now_iso();
817 66 : let mut found = false;
818 66 : let local_resp: Option<ApiResult<Response>> = if local_covered {
819 10 : None
820 : } else {
821 56 : let res = st
822 56 : .store
823 56 : .mutate(&tenant, Kind::Entity, id, |doc| {
824 : // 5.6.4.4: the ?type selector narrows the target entity
825 46 : if !matches_type_param(doc, params, &parsed.ctx) {
826 8 : return Err(NgsiError::ResourceNotFound(format!(
827 8 : "entity {id} does not match the type selector"
828 8 : )));
829 38 : }
830 38 : let target = antares_store::stored_object(doc)?;
831 38 : if let Some(existing) = target.get_mut(&attr_iri).and_then(Value::as_array_mut) {
832 36 : let pos = existing.iter().position(|ci| {
833 36 : ci.get("datasetId").and_then(Value::as_str) == want_ds.as_deref()
834 36 : });
835 36 : if let Some(p) = pos {
836 36 : found = true;
837 36 : if is_deletion {
838 0 : existing.remove(p);
839 0 : } else {
840 : // 5.6.4.4: the fragment may not change the Attribute type
841 28 : if let (Some(ft), Some(et)) = (
842 36 : frag_inst.get("type").and_then(Value::as_str),
843 36 : existing[p].get("type").and_then(Value::as_str),
844 : ) {
845 28 : if ft != et {
846 0 : return Err(NgsiError::BadRequestData(format!(
847 0 : "attribute type mismatch: {ft} != {et} (5.6.4)"
848 0 : )));
849 28 : }
850 8 : }
851 36 : let t = antares_store::stored_object(&mut existing[p])?;
852 82 : for (k, v) in antares_jsonld::expanded_object(&frag_inst)? {
853 82 : if matches!(k.as_str(), "createdAt" | "modifiedAt") {
854 0 : continue;
855 82 : }
856 82 : if v.is_null() || antares_jsonld::is_ngsi_null(v) {
857 0 : t.remove(k);
858 82 : } else {
859 82 : t.insert(k.clone(), v.clone());
860 82 : }
861 : }
862 36 : t.insert("modifiedAt".into(), Value::String(ts.clone()));
863 : }
864 0 : }
865 36 : if existing.is_empty() {
866 0 : target.remove(&attr_iri);
867 36 : }
868 2 : }
869 38 : if found {
870 36 : target.insert("modifiedAt".into(), Value::String(ts.clone()));
871 36 : }
872 38 : Ok::<(), NgsiError>(())
873 46 : })
874 56 : .await?;
875 56 : Some(single_attr_local(res, found, id, attr, &tenant))
876 : };
877 66 : if regs.is_empty() {
878 : // the local half is skipped only while registrations cover the
879 : // attributes, so an empty list always leaves a local response
880 56 : return local_resp
881 56 : .unwrap_or_else(|| Err(NgsiError::InternalError("no local result".into()).into()));
882 10 : }
883 10 : let fed_parts = crate::federation::fed_attr_parts(
884 10 : st,
885 10 : headers,
886 10 : &tenant,
887 10 : &parsed.ctx.source,
888 10 : ®s,
889 10 : "updateAttrs",
890 10 : reqwest::Method::PATCH,
891 10 : &format!(
892 10 : "/entities/{}/attrs/{}",
893 10 : path_segment(id),
894 10 : path_segment(attr)
895 10 : ),
896 10 : &fwd_query(params, &["type"]),
897 10 : Some(Value::Object(without_context_map(obj))),
898 10 : )
899 10 : .await;
900 10 : Ok(combine_single_attr(
901 10 : &tenant,
902 10 : &attr_iri,
903 10 : local_covered,
904 10 : &local_resp,
905 10 : ®s,
906 10 : &fed_parts,
907 10 : ))
908 102 : }
909 :
910 102 : fn without_context_map(o: &Map<String, Value>) -> Map<String, Value> {
911 102 : let mut o = o.clone();
912 102 : o.remove("@context");
913 102 : o
914 102 : }
915 :
916 : // ---------- PUT /entities/{id}/attrs/{attrId} — Replace attribute (5.6.19) ----------
917 :
918 108 : pub async fn replace_attr(
919 108 : State(st): State<AppState>,
920 108 : Path((id, attr)): Path<(String, String)>,
921 108 : CleanParams(params): CleanParams,
922 108 : headers: HeaderMap,
923 108 : body: Bytes,
924 108 : ) -> Response {
925 108 : let go = async {
926 108 : let tenant = tenant_from(&headers)?;
927 108 : antares_model::EntityId::new(&id)?;
928 108 : check_attr_name(&attr)?;
929 92 : check_params(¶ms, &["local", "type"])?;
930 92 : let parsed = parse_body(&st.loader, &headers, &body, BodyKind::Standard).await?;
931 : // 5.5.7 term expansion first, then 5.6.19.4: "If the target Attribute
932 : // is scope, then an error of type BadRequestData shall be raised" —
933 : // the target is the expanded name, so the IRI spelling counts too.
934 92 : gate!(st, &tenant, &headers, "5.6.19", ids: &[&id]).await?;
935 92 : let attr_iri = antares_jsonld::expand_attr_name(&attr, &parsed.ctx)?;
936 76 : if attr == "scope" || attr_iri == SCOPE_IRI {
937 12 : return Err(NgsiError::BadRequestData(
938 12 : "scope cannot be the target of a replace attribute (5.6.19)".into(),
939 12 : )
940 12 : .into());
941 64 : }
942 64 : let obj = parsed.object(NgsiError::BadRequestData(
943 64 : "fragment must be a JSON object".into(),
944 64 : ))?;
945 64 : let mut wrapper = Map::new();
946 64 : wrapper.insert(attr.clone(), Value::Object(without_context_map(obj)));
947 64 : let fragment = expand_entity(
948 64 : &wrapper,
949 64 : &parsed.ctx,
950 64 : ExpandOpts {
951 64 : fragment: true,
952 64 : allow_null: false,
953 64 : temporal: false,
954 64 : ..Default::default()
955 64 : },
956 0 : )?;
957 64 : let incoming_arr = fragment
958 64 : .get(&attr_iri)
959 64 : .cloned()
960 64 : .ok_or_else(|| NgsiError::BadRequestData("invalid attribute fragment".into()))?;
961 64 : let new_inst = incoming_arr
962 64 : .as_array()
963 64 : .and_then(|a| a.first())
964 64 : .cloned()
965 64 : .ok_or_else(|| NgsiError::BadRequestData("invalid attribute fragment".into()))?;
966 64 : let want_ds = new_inst
967 64 : .get("datasetId")
968 64 : .and_then(Value::as_str)
969 64 : .map(String::from);
970 64 : let (regs, loop_answer) =
971 64 : attr_regs_or_loop(&st, &tenant, &id, &attr_iri, &parsed.ctx, ¶ms, &headers).await?;
972 64 : if let Some(r) = loop_answer {
973 0 : return Ok(r);
974 64 : }
975 64 : let local_covered = proxies_cover_all(®s, std::slice::from_ref(&attr_iri));
976 64 : let ts = now_iso();
977 64 : let mut found = false;
978 64 : let local_resp: Option<ApiResult<Response>> = if local_covered {
979 8 : None
980 : } else {
981 56 : let res = st
982 56 : .store
983 56 : .mutate(&tenant, Kind::Entity, &id, |doc| {
984 : // 5.6.19.4: the ?type selector narrows the target entity
985 48 : if !matches_type_param(doc, ¶ms, &parsed.ctx) {
986 4 : return Err(NgsiError::ResourceNotFound(format!(
987 4 : "entity {id} does not match the type selector"
988 4 : )));
989 44 : }
990 44 : let target = antares_store::stored_object(doc)?;
991 44 : if let Some(existing) = target.get_mut(&attr_iri).and_then(Value::as_array_mut)
992 : {
993 : // 5.6.19: only the instance with the matching datasetId is
994 : // replaced; its createdAt survives (055_01/055_02)
995 42 : if let Some(p) = existing.iter().position(|ci| {
996 42 : ci.get("datasetId").and_then(Value::as_str) == want_ds.as_deref()
997 42 : }) {
998 42 : found = true;
999 42 : let created = existing[p].get("createdAt").cloned();
1000 42 : let mut ni = new_inst.clone();
1001 42 : if let Some(o) = ni.as_object_mut() {
1002 42 : if let Some(c) = created {
1003 42 : o.insert("createdAt".into(), c);
1004 42 : } else {
1005 0 : o.insert("createdAt".into(), Value::String(ts.clone()));
1006 0 : }
1007 42 : o.insert("modifiedAt".into(), Value::String(ts.clone()));
1008 0 : }
1009 42 : existing[p] = ni;
1010 0 : }
1011 2 : }
1012 44 : if found {
1013 42 : target.insert("modifiedAt".into(), Value::String(ts.clone()));
1014 42 : }
1015 44 : Ok::<(), NgsiError>(())
1016 48 : })
1017 56 : .await?;
1018 56 : Some(single_attr_local(res, found, &id, &attr, &tenant))
1019 : };
1020 64 : if regs.is_empty() {
1021 : // the local half is skipped only while registrations cover the
1022 : // attributes, so an empty list always leaves a local response
1023 56 : return local_resp
1024 56 : .unwrap_or_else(|| Err(NgsiError::InternalError("no local result".into()).into()));
1025 8 : }
1026 8 : let fed_parts = crate::federation::fed_attr_parts(
1027 8 : &st,
1028 8 : &headers,
1029 8 : &tenant,
1030 8 : &parsed.ctx.source,
1031 8 : ®s,
1032 8 : "replaceAttrs",
1033 8 : reqwest::Method::PUT,
1034 8 : &format!(
1035 8 : "/entities/{}/attrs/{}",
1036 8 : path_segment(&id),
1037 8 : path_segment(&attr)
1038 8 : ),
1039 8 : &fwd_query(¶ms, &["type"]),
1040 8 : Some(Value::Object(without_context_map(obj))),
1041 8 : )
1042 8 : .await;
1043 8 : Ok(combine_single_attr(
1044 8 : &tenant,
1045 8 : &attr_iri,
1046 8 : local_covered,
1047 8 : &local_resp,
1048 8 : ®s,
1049 8 : &fed_parts,
1050 8 : ))
1051 108 : };
1052 108 : go.await.unwrap_or_else(|e: ApiError| e.into_response())
1053 108 : }
1054 :
1055 : // ---------- DELETE /entities/{id}/attrs/{attrId} (5.6.5) ----------
1056 :
1057 100 : pub async fn delete_attr(
1058 100 : State(st): State<AppState>,
1059 100 : Path((id, attr)): Path<(String, String)>,
1060 100 : CleanParams(params): CleanParams,
1061 100 : headers: HeaderMap,
1062 100 : ) -> Response {
1063 100 : match delete_attr_inner(&st, &id, &attr, ¶ms, &headers).await {
1064 56 : Ok(r) => r,
1065 44 : Err(e) => e.into_response(),
1066 : }
1067 100 : }
1068 :
1069 100 : async fn delete_attr_inner(
1070 100 : st: &AppState,
1071 100 : id: &str,
1072 100 : attr: &str,
1073 100 : params: &HashMap<String, String>,
1074 100 : headers: &HeaderMap,
1075 100 : ) -> ApiResult<Response> {
1076 100 : let tenant = tenant_from(headers)?;
1077 100 : antares_model::EntityId::new(id)?;
1078 100 : check_attr_name(attr)?;
1079 84 : check_params(params, &["datasetId", "deleteAll", "local", "type"])?;
1080 84 : let ctx = request_context(&st.loader, headers).await?;
1081 84 : gate!(st, &tenant, headers, "5.6.5", ids: &[id]).await?;
1082 : // 5.6.5.4 expands the path name the same way 5.6.4.4 does, and then
1083 : // addresses `scope` as itself rather than as an Attribute. The target is
1084 : // what the name expands to, so both spellings — the reserved short name
1085 : // and the fully qualified one a client @context can map a term to —
1086 : // address the member the document stores under its short name.
1087 84 : let attr_iri = if attr == "scope" {
1088 10 : "scope".to_owned()
1089 : } else {
1090 74 : match antares_jsonld::expand_attr_name(attr, &ctx)? {
1091 70 : iri if iri == SCOPE_IRI => "scope".to_owned(),
1092 68 : iri => iri,
1093 : }
1094 : };
1095 80 : let delete_all = params.get("deleteAll").map(String::as_str) == Some("true");
1096 80 : let want_ds = crate::repr::target_dataset_id(params).map(String::from);
1097 80 : let (regs, loop_answer) =
1098 80 : attr_regs_or_loop(st, &tenant, id, &attr_iri, &ctx, params, headers).await?;
1099 80 : if let Some(r) = loop_answer {
1100 0 : return Ok(r);
1101 80 : }
1102 80 : let local_covered = proxies_cover_all(®s, std::slice::from_ref(&attr_iri));
1103 80 : let ts = now_iso();
1104 80 : let mut found = false;
1105 80 : let local_resp: Option<ApiResult<Response>> = if local_covered {
1106 16 : None
1107 : } else {
1108 64 : let res = st
1109 64 : .store
1110 64 : .mutate(&tenant, Kind::Entity, id, |doc| {
1111 : // 5.6.5.4: the ?type selector narrows the target entity
1112 54 : if !matches_type_param(doc, params, &ctx) {
1113 12 : return Err(NgsiError::ResourceNotFound(format!(
1114 12 : "entity {id} does not match the type selector"
1115 12 : )));
1116 42 : }
1117 42 : if attr_iri == "scope" {
1118 12 : let target = antares_store::stored_object(doc)?;
1119 12 : found = target.remove("scope").is_some();
1120 12 : if found {
1121 12 : target.insert("modifiedAt".into(), Value::String(ts.clone()));
1122 12 : }
1123 12 : return Ok(());
1124 30 : }
1125 30 : let target = antares_store::stored_object(doc)?;
1126 30 : if let Some(existing) = target.get_mut(&attr_iri).and_then(Value::as_array_mut) {
1127 28 : if delete_all {
1128 0 : found = !existing.is_empty();
1129 0 : existing.clear();
1130 0 : } else {
1131 30 : let pos = existing.iter().position(|ci| {
1132 30 : ci.get("datasetId").and_then(Value::as_str) == want_ds.as_deref()
1133 30 : });
1134 28 : if let Some(p) = pos {
1135 28 : existing.remove(p);
1136 28 : found = true;
1137 28 : }
1138 : }
1139 28 : if existing.is_empty() {
1140 24 : target.remove(&attr_iri);
1141 24 : }
1142 2 : }
1143 30 : if found {
1144 28 : target.insert("modifiedAt".into(), Value::String(ts.clone()));
1145 28 : }
1146 30 : Ok::<(), NgsiError>(())
1147 54 : })
1148 64 : .await?;
1149 : // The temporal representation records the deletion (4.8 deletedAt) — and
1150 : // the entity may exist ONLY temporally (created via 5.6.11), so a
1151 : // missing current-state entity still records. A REFUSED delete (the
1152 : // 5.6.5.4 type selector did not match) deleted nothing and must leave
1153 : // the attribute history untouched.
1154 64 : let temporal_had = !matches!(res, Some(Err(_)))
1155 52 : && crate::history::mirror_delete_attr(
1156 52 : st,
1157 52 : &tenant,
1158 52 : id,
1159 52 : &attr_iri,
1160 52 : want_ds.as_deref(),
1161 52 : &ts,
1162 52 : )
1163 52 : .await;
1164 : // An entity that exists only temporally still answers 204: the
1165 : // deletion was recorded even though there was no current state.
1166 64 : Some(if res.is_none() && temporal_had {
1167 0 : Ok(no_content(&tenant))
1168 : } else {
1169 64 : single_attr_local(res, found || temporal_had, id, attr, &tenant)
1170 : })
1171 : };
1172 80 : if regs.is_empty() {
1173 : // the local half is skipped only while registrations cover the
1174 : // attributes, so an empty list always leaves a local response
1175 64 : return local_resp
1176 64 : .unwrap_or_else(|| Err(NgsiError::InternalError("no local result".into()).into()));
1177 16 : }
1178 16 : let query = fwd_query(params, &["datasetId", "deleteAll", "type"]);
1179 16 : let fed_parts = crate::federation::fed_attr_parts(
1180 16 : st,
1181 16 : headers,
1182 16 : &tenant,
1183 16 : &ctx.source,
1184 16 : ®s,
1185 16 : "deleteAttrs",
1186 16 : reqwest::Method::DELETE,
1187 16 : &format!(
1188 16 : "/entities/{}/attrs/{}",
1189 16 : path_segment(id),
1190 16 : path_segment(attr)
1191 16 : ),
1192 16 : &query,
1193 16 : None,
1194 16 : )
1195 16 : .await;
1196 16 : Ok(combine_single_attr(
1197 16 : &tenant,
1198 16 : &attr_iri,
1199 16 : local_covered,
1200 16 : &local_resp,
1201 16 : ®s,
1202 16 : &fed_parts,
1203 16 : ))
1204 100 : }
1205 :
1206 : #[cfg(test)]
1207 : mod attr_name_and_via_paths {
1208 : use crate::AppState;
1209 : use axum::body::Body;
1210 : use axum::http::{Request, StatusCode};
1211 : use std::io::{Read, Write};
1212 : use std::sync::atomic::{AtomicUsize, Ordering};
1213 : use std::sync::{Arc, Mutex};
1214 : use tower::ServiceExt;
1215 :
1216 : const ALIAS: &str = "antares1";
1217 :
1218 : /// A Context Source answering 204 to everything, counting the hits and
1219 : /// recording the request-target of each one (the wire truth a forwarded
1220 : /// operation is judged on).
1221 16 : fn mock_source() -> (u16, Arc<AtomicUsize>, Arc<Mutex<Vec<String>>>) {
1222 16 : let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
1223 16 : let port = listener.local_addr().expect("addr").port();
1224 16 : let hits: Arc<AtomicUsize> = Arc::default();
1225 16 : let seen = hits.clone();
1226 16 : let targets: Arc<Mutex<Vec<String>>> = Arc::default();
1227 16 : let seen_targets = targets.clone();
1228 16 : std::thread::spawn(move || {
1229 44 : for stream in listener.incoming() {
1230 44 : let Ok(mut s) = stream else { continue };
1231 44 : seen.fetch_add(1, Ordering::SeqCst);
1232 44 : let mut buf = [0u8; 8192];
1233 44 : let n = s.read(&mut buf).unwrap_or(0);
1234 44 : let head = String::from_utf8_lossy(&buf[..n]).into_owned();
1235 44 : let line = head.lines().next().unwrap_or_default().to_owned();
1236 44 : if let Some(t) = line.split_whitespace().nth(1) {
1237 44 : seen_targets.lock().expect("targets").push(t.to_owned());
1238 44 : }
1239 44 : let _ = s.write_all(
1240 44 : b"HTTP/1.1 204 No Content\r\nConnection: close\r\nContent-Length: 0\r\n\r\n",
1241 44 : );
1242 : }
1243 16 : });
1244 16 : (port, hits, targets)
1245 16 : }
1246 :
1247 : /// The request-targets the mock has seen so far, in arrival order.
1248 44 : fn seen(targets: &Arc<Mutex<Vec<String>>>) -> Vec<String> {
1249 44 : targets.lock().expect("targets").clone()
1250 44 : }
1251 :
1252 24 : fn state() -> AppState {
1253 : // the mock source is loopback, denied by the egress policy by default
1254 24 : crate::allow_private();
1255 24 : AppState::new(ALIAS.into())
1256 24 : }
1257 :
1258 148 : async fn send(st: &AppState, req: Request<Body>) -> axum::http::Response<Body> {
1259 148 : crate::router(st.clone())
1260 148 : .oneshot(req)
1261 148 : .await
1262 148 : .expect("response")
1263 148 : }
1264 :
1265 36 : async fn post(st: &AppState, uri: &str, body: String) -> StatusCode {
1266 36 : let req = Request::builder()
1267 36 : .method("POST")
1268 36 : .uri(uri)
1269 36 : .header("Content-Type", "application/json")
1270 36 : .header("Content-Length", body.len())
1271 36 : .body(Body::from(body))
1272 36 : .expect("request");
1273 36 : send(st, req).await.status()
1274 36 : }
1275 :
1276 : /// One registration for `entity` covering every attribute. The store is
1277 : /// process-wide, so every test uses its own ids. `mode` is a parameter
1278 : /// because 5.9.2 forbids two proxied (exclusive or redirect)
1279 : /// registrations from overlapping — a test that needs two matching
1280 : /// registrations has to register them inclusive.
1281 20 : async fn register(st: &AppState, port: u16, id: &str, entity: &str, mode: &str) {
1282 20 : let doc = serde_json::json!({
1283 20 : "id": format!("urn:ngsi-ld:ContextSourceRegistration:{id}"),
1284 20 : "type": "ContextSourceRegistration",
1285 20 : "mode": mode,
1286 20 : "operations": ["updateEntity", "updateAttrs", "replaceAttrs", "deleteAttrs", "appendAttrs"],
1287 20 : "information": [{"entities": [{"type": "Vehicle", "id": entity}]}],
1288 20 : "endpoint": format!("http://127.0.0.1:{port}"),
1289 : });
1290 20 : assert_eq!(
1291 20 : post(st, "/ngsi-ld/v1/csourceRegistrations", doc.to_string()).await,
1292 : StatusCode::CREATED,
1293 : "registration create"
1294 : );
1295 20 : }
1296 :
1297 12 : async fn create_entity(st: &AppState, entity: &str) {
1298 12 : let doc = serde_json::json!({
1299 12 : "id": entity, "type": "Vehicle",
1300 12 : "speed": {"type": "Property", "value": 1},
1301 : });
1302 12 : assert_eq!(
1303 12 : post(st, "/ngsi-ld/v1/entities", doc.to_string()).await,
1304 : StatusCode::CREATED,
1305 : "entity create"
1306 : );
1307 12 : }
1308 :
1309 : /// 4.6.2: a name starts with a letter, so no valid Attribute name is a
1310 : /// relative-path dot segment (RFC 3986 clause 5.2.4). Such a name reaching
1311 : /// a forwarded request URL re-targets the registration endpoint —
1312 : /// `DELETE /entities/{id}/attrs/..` becomes a Delete Entity on the peer —
1313 : /// so it is refused with BadRequestData before anything is forwarded.
1314 : #[tokio::test(flavor = "multi_thread")]
1315 4 : async fn dot_segment_attribute_name_is_refused_and_never_forwarded() {
1316 : const ENTITY: &str = "urn:ngsi-ld:Vehicle:attrs-traversal";
1317 4 : let st = state();
1318 4 : let (port, hits, targets) = mock_source();
1319 4 : register(&st, port, "csr-traversal", ENTITY, "redirect").await;
1320 4 : let frag = r#"{"type":"Property","value":1}"#;
1321 : // raw, decoded once by this broker, and decoded once more by the peer
1322 16 : for attr in ["..", "%2e%2e", "%252e%252e", "."] {
1323 48 : for (method, body) in [("DELETE", None), ("PATCH", Some(frag)), ("PUT", Some(frag))] {
1324 48 : let mut req = Request::builder()
1325 48 : .method(method)
1326 48 : .uri(format!("/ngsi-ld/v1/entities/{ENTITY}/attrs/{attr}"));
1327 48 : if let Some(b) = body {
1328 32 : req = req
1329 32 : .header("Content-Type", "application/json")
1330 32 : .header("Content-Length", b.len());
1331 32 : }
1332 48 : let req = req
1333 48 : .body(body.map_or_else(Body::empty, |b| Body::from(b.to_owned())))
1334 48 : .expect("request");
1335 48 : assert_eq!(
1336 48 : send(&st, req).await.status(),
1337 : StatusCode::BAD_REQUEST,
1338 : "{method} attribute name {attr:?}"
1339 : );
1340 : }
1341 : }
1342 4 : assert_eq!(
1343 4 : hits.load(Ordering::SeqCst),
1344 : 0,
1345 : "a rejected attribute name must never reach a registration endpoint"
1346 : );
1347 : // an absolute IRI carries '/' and '.' legitimately: it passes the check
1348 : // and is forwarded, so the guard rejects the shape, not the characters
1349 4 : let req = Request::builder()
1350 4 : .method("DELETE")
1351 4 : .uri(format!(
1352 : "/ngsi-ld/v1/entities/{ENTITY}/attrs/https%3A%2F%2Fexample.org%2Fv1.0%2Fspeed"
1353 : ))
1354 4 : .body(Body::empty())
1355 4 : .expect("request");
1356 4 : assert_eq!(
1357 4 : send(&st, req).await.status(),
1358 : StatusCode::NO_CONTENT,
1359 : "an absolute IRI is a valid attribute name"
1360 : );
1361 : // ':' is a legal path character (RFC 3986 clause 3.3 pchar), '/' is not
1362 4 : assert_eq!(
1363 4 : seen(&targets),
1364 4 : vec![format!(
1365 : "/ngsi-ld/v1/entities/{ENTITY}/attrs/https:%2F%2Fexample.org%2Fv1.0%2Fspeed"
1366 : )],
1367 : "the peer is asked for that attribute of that entity, nothing else"
1368 : );
1369 4 : assert_eq!(
1370 4 : hits.load(Ordering::SeqCst),
1371 4 : 1,
1372 4 : "a valid attribute name is still forwarded"
1373 4 : );
1374 4 : }
1375 :
1376 : /// Tables 6.6.3.1-1/6.6.3.2-1/6.7.3.1-1 make the entity id and the
1377 : /// attribute name single URI path variables, so a forwarded attribute
1378 : /// operation must address the SAME resource at the registered source.
1379 : /// RFC 3986 clause 3.3 ends a path segment at `#`, `?` or `/`: spliced
1380 : /// raw, an id carrying `#` truncates the forwarded path and Delete
1381 : /// Attribute (5.6.5) reaches the peer as Delete Entity (5.6.6).
1382 : #[tokio::test(flavor = "multi_thread")]
1383 4 : async fn forwarded_attribute_path_keeps_the_whole_id_and_name() {
1384 : // five forwards through a blocking mock, each under the 8 s forward
1385 : // cap; a sanitizer's slowdown turns them into 504s
1386 4 : if std::env::var_os("ANTARES_TEST_SANITIZER").is_some() {
1387 0 : return;
1388 4 : }
1389 : // both delimiters an id may legally carry (RFC 3986 clause 3.3)
1390 : const ENTITY: &str = "urn:ngsi-ld:Vehicle:attrs-frag#1?x";
1391 : const ID: &str = "urn:ngsi-ld:Vehicle:attrs-frag%231%3Fx";
1392 : // an absolute IRI is a legal Attribute name (4.6.2) and carries '/'
1393 : const ATTR: &str = "https%3A%2F%2Fexample.org%2Fv1.0%2Fspeed";
1394 : // ':' is a legal path character (RFC 3986 clause 3.3 pchar), '/' is not
1395 : const SEG: &str = "https:%2F%2Fexample.org%2Fv1.0%2Fspeed";
1396 4 : let st = state();
1397 4 : let (port, hits, targets) = mock_source();
1398 4 : register(&st, port, "csr-frag", ENTITY, "redirect").await;
1399 4 : let multi = r#"{"speed":{"type":"Property","value":1}}"#;
1400 4 : let single = r#"{"type":"Property","value":2}"#;
1401 4 : let single_tail = format!("/attrs/{ATTR}");
1402 4 : let cases = [
1403 4 : ("POST", "/attrs".to_owned(), Some(multi), "/attrs/"),
1404 4 : ("PATCH", "/attrs".to_owned(), Some(multi), "/attrs/"),
1405 4 : ("PATCH", single_tail.clone(), Some(single), "single"),
1406 4 : ("PUT", single_tail.clone(), Some(single), "single"),
1407 4 : ("DELETE", single_tail.clone(), None, "single"),
1408 4 : ];
1409 20 : for (method, tail, body, want_tail) in cases {
1410 20 : let mut req = Request::builder()
1411 20 : .method(method)
1412 20 : .uri(format!("/ngsi-ld/v1/entities/{ID}{tail}"));
1413 20 : if let Some(b) = body {
1414 16 : req = req
1415 16 : .header("Content-Type", "application/json")
1416 16 : .header("Content-Length", b.len());
1417 16 : }
1418 20 : let req = req
1419 20 : .body(body.map_or_else(Body::empty, |b| Body::from(b.to_owned())))
1420 20 : .expect("request");
1421 20 : assert_eq!(
1422 20 : send(&st, req).await.status(),
1423 4 : StatusCode::NO_CONTENT,
1424 4 : "{method} {tail}"
1425 4 : );
1426 20 : let want = if want_tail == "single" {
1427 12 : format!("/ngsi-ld/v1/entities/{ID}/attrs/{SEG}")
1428 4 : } else {
1429 8 : format!("/ngsi-ld/v1/entities/{ID}{want_tail}")
1430 4 : };
1431 20 : let got = seen(&targets).pop().unwrap_or_default();
1432 20 : assert_eq!(got, want, "{method} {tail} forwarded target");
1433 4 : // the id must not have been cut short at its '#' or '?', and no
1434 4 : // delimiter may survive undecoded inside a segment
1435 20 : assert!(
1436 20 : !got.contains('#') && !got.contains('?'),
1437 4 : "{method} {tail}: raw delimiter in the forwarded path {got}"
1438 4 : );
1439 4 : }
1440 4 : assert_eq!(hits.load(Ordering::SeqCst), 5, "one forward per operation");
1441 4 : }
1442 :
1443 : /// `type` is a shall-support URL parameter of every attribute write
1444 : /// (Tables 6.6.3.1-1, 6.6.3.2-1, 6.7.3.1-1, 6.7.3.2-1, 6.7.3.3-1) and
1445 : /// 5.6.5.4 identifies the target by its "id (URI), and where specified
1446 : /// type" — so the selector has to travel with the forwarded operation,
1447 : /// or the registered source writes on entities the client scoped out.
1448 : #[tokio::test(flavor = "multi_thread")]
1449 4 : async fn type_selector_travels_with_the_forwarded_attribute_write() {
1450 : // five forwards through a blocking mock, each under the 8 s forward
1451 : // cap; a sanitizer's slowdown turns them into 504s
1452 4 : if std::env::var_os("ANTARES_TEST_SANITIZER").is_some() {
1453 0 : return;
1454 4 : }
1455 : const ENTITY: &str = "urn:ngsi-ld:Vehicle:attrs-typesel";
1456 4 : let st = state();
1457 4 : let (port, hits, targets) = mock_source();
1458 4 : register(&st, port, "csr-typesel", ENTITY, "redirect").await;
1459 4 : let multi = r#"{"speed":{"type":"Property","value":1}}"#;
1460 4 : let single = r#"{"type":"Property","value":2}"#;
1461 4 : let cases = [
1462 4 : ("POST", "/attrs?type=Vehicle", Some(multi)),
1463 4 : ("PATCH", "/attrs?type=Vehicle", Some(multi)),
1464 4 : ("PATCH", "/attrs/speed?type=Vehicle", Some(single)),
1465 4 : ("PUT", "/attrs/speed?type=Vehicle", Some(single)),
1466 4 : ("DELETE", "/attrs/speed?type=Vehicle", None),
1467 4 : ];
1468 20 : for (method, tail, body) in cases {
1469 20 : let mut req = Request::builder()
1470 20 : .method(method)
1471 20 : .uri(format!("/ngsi-ld/v1/entities/{ENTITY}{tail}"));
1472 20 : if let Some(b) = body {
1473 16 : req = req
1474 16 : .header("Content-Type", "application/json")
1475 16 : .header("Content-Length", b.len());
1476 16 : }
1477 20 : let req = req
1478 20 : .body(body.map_or_else(Body::empty, |b| Body::from(b.to_owned())))
1479 20 : .expect("request");
1480 20 : assert_eq!(
1481 20 : send(&st, req).await.status(),
1482 4 : StatusCode::NO_CONTENT,
1483 4 : "{method} {tail}"
1484 4 : );
1485 20 : let got = seen(&targets).pop().unwrap_or_default();
1486 20 : assert!(
1487 20 : got.contains("type=Vehicle"),
1488 4 : "{method} {tail} forwarded without the type selector: {got}"
1489 4 : );
1490 20 : assert!(
1491 20 : !got.contains("local="),
1492 4 : "{method} {tail}: local is never forwarded"
1493 4 : );
1494 4 : }
1495 4 : assert_eq!(hits.load(Ordering::SeqCst), 5, "one forward per operation");
1496 4 : }
1497 :
1498 : /// 5.6.5.4: with a `?type` selector that the target Entity does not
1499 : /// match, the entity is "not known" for this operation and
1500 : /// ResourceNotFound is raised — nothing is deleted, so the temporal
1501 : /// representation must not gain a 4.8 `deletedAt` instance either.
1502 : #[tokio::test(flavor = "multi_thread")]
1503 4 : async fn refused_delete_writes_no_temporal_deletion() {
1504 : const ENTITY: &str = "urn:ngsi-ld:Vehicle:attrs-refused";
1505 4 : let st = state();
1506 4 : create_entity(&st, ENTITY).await;
1507 : // 5.6.11: the temporal representation of the same entity
1508 4 : let history = serde_json::json!({
1509 4 : "id": ENTITY, "type": "Vehicle",
1510 4 : "speed": [{"type": "Property", "value": 1,
1511 4 : "observedAt": "2026-01-01T00:00:00Z"}],
1512 : });
1513 4 : assert_eq!(
1514 4 : post(&st, "/ngsi-ld/v1/temporal/entities", history.to_string()).await,
1515 : StatusCode::CREATED,
1516 : "temporal create"
1517 : );
1518 4 : let tenant = antares_model::TenantId::new("default").expect("tenant");
1519 4 : let req = Request::builder()
1520 4 : .method("DELETE")
1521 4 : .uri(format!(
1522 : "/ngsi-ld/v1/entities/{ENTITY}/attrs/speed?type=Building"
1523 : ))
1524 4 : .body(Body::empty())
1525 4 : .expect("request");
1526 4 : assert_eq!(send(&st, req).await.status(), StatusCode::NOT_FOUND);
1527 4 : let doc = st
1528 4 : .store
1529 4 : .get(&tenant, antares_store::Kind::Temporal, ENTITY)
1530 4 : .await
1531 4 : .expect("store read")
1532 4 : .expect("temporal doc");
1533 4 : assert!(
1534 4 : !doc.to_string().contains("deletedAt"),
1535 : "a refused delete must not tombstone the attribute history: {doc}"
1536 : );
1537 4 : assert!(
1538 4 : !doc.to_string().contains("urn:ngsi-ld:null"),
1539 4 : "nor write a deletion instance"
1540 4 : );
1541 4 : }
1542 :
1543 : /// 5.6.19.4: "If the target Attribute is scope, then an error of type
1544 : /// BadRequestData shall be raised" — the target is the expanded name
1545 : /// (5.5.7), so the IRI spelling of scope is refused just like the term.
1546 : #[tokio::test(flavor = "multi_thread")]
1547 4 : async fn replace_attr_refuses_the_scope_iri_spelling() {
1548 : const ENTITY: &str = "urn:ngsi-ld:Vehicle:attrs-scope-iri";
1549 4 : let st = state();
1550 4 : create_entity(&st, ENTITY).await;
1551 4 : let frag = r#"{"type":"Property","value":"/Madrid"}"#;
1552 8 : for attr in ["scope", "https%3A%2F%2Furi.etsi.org%2Fngsi-ld%2Fscope"] {
1553 8 : let req = Request::builder()
1554 8 : .method("PUT")
1555 8 : .uri(format!("/ngsi-ld/v1/entities/{ENTITY}/attrs/{attr}"))
1556 8 : .header("Content-Type", "application/json")
1557 8 : .header("Content-Length", frag.len())
1558 8 : .body(Body::from(frag))
1559 8 : .expect("request");
1560 8 : let resp = send(&st, req).await;
1561 8 : assert_eq!(
1562 8 : resp.status(),
1563 4 : StatusCode::BAD_REQUEST,
1564 4 : "scope target {attr}"
1565 4 : );
1566 8 : let bytes = http_body_util::BodyExt::collect(resp.into_body())
1567 8 : .await
1568 8 : .expect("body")
1569 8 : .to_bytes();
1570 8 : let body: serde_json::Value = serde_json::from_slice(&bytes).expect("json");
1571 8 : assert_eq!(
1572 8 : body["type"], "https://uri.etsi.org/ngsi-ld/errors/BadRequestData",
1573 4 : "scope target {attr}"
1574 4 : );
1575 4 : // the 5.6.19.4 guard, not an incidental fragment-shape rejection
1576 8 : assert!(
1577 8 : body["detail"]
1578 8 : .as_str()
1579 8 : .is_some_and(|d| d.contains("scope cannot be the target")),
1580 4 : "scope target {attr}: {body}"
1581 4 : );
1582 4 : }
1583 4 : }
1584 :
1585 : /// 6.3.17/6.3.18 (Table 6.3.18-2): a Via chain naming this broker leaves
1586 : /// the matching registrations out of the operation, which then runs
1587 : /// locally. 508 is reserved for a SINGLE exclusive/redirect source, so two
1588 : /// registrations must produce an answer — never a 500.
1589 : #[tokio::test(flavor = "multi_thread")]
1590 4 : async fn self_via_with_two_registrations_answers_locally() {
1591 : const ENTITY: &str = "urn:ngsi-ld:Vehicle:attrs-via";
1592 4 : let st = state();
1593 4 : create_entity(&st, ENTITY).await;
1594 4 : let (port, hits, _targets) = mock_source();
1595 4 : register(&st, port, "csr-via-a", ENTITY, "inclusive").await;
1596 4 : register(&st, port, "csr-via-b", ENTITY, "inclusive").await;
1597 4 : let via = format!("1.1 {ALIAS}");
1598 :
1599 : // 5.6.2 Update Attributes — the multi-attribute fragment plan
1600 4 : let body = r#"{"speed":{"type":"Property","value":9}}"#;
1601 4 : let req = Request::builder()
1602 4 : .method("PATCH")
1603 4 : .uri(format!("/ngsi-ld/v1/entities/{ENTITY}/attrs"))
1604 4 : .header("Content-Type", "application/json")
1605 4 : .header("Content-Length", body.len())
1606 4 : .header("Via", via.as_str())
1607 4 : .body(Body::from(body))
1608 4 : .expect("request");
1609 4 : assert_eq!(send(&st, req).await.status(), StatusCode::NO_CONTENT);
1610 :
1611 : // 5.6.5 Delete Attribute — the single-attribute plan
1612 4 : let req = Request::builder()
1613 4 : .method("DELETE")
1614 4 : .uri(format!("/ngsi-ld/v1/entities/{ENTITY}/attrs/speed"))
1615 4 : .header("Via", via.as_str())
1616 4 : .body(Body::empty())
1617 4 : .expect("request");
1618 4 : assert_eq!(send(&st, req).await.status(), StatusCode::NO_CONTENT);
1619 :
1620 4 : assert_eq!(
1621 4 : hits.load(Ordering::SeqCst),
1622 4 : 0,
1623 4 : "registrations dropped by the Via chain must not be contacted"
1624 4 : );
1625 4 : }
1626 : }
1627 :
1628 : #[cfg(test)]
1629 : mod clause_4_8 {
1630 : use serde_json::{json, Value};
1631 :
1632 : const TS: &str = "2026-08-30T10:00:00.000Z";
1633 :
1634 : /// 4.8: createdAt and modifiedAt are the times at which "the Entity,
1635 : /// Property or Relationship" entered and was last modified in an NGSI-LD
1636 : /// system. A sub-Property is a Property, so an Attribute arriving through
1637 : /// 5.6.2/5.6.3 is stamped to the same depth as one arriving through
1638 : /// 5.6.1 — the served representation cannot depend on which operation
1639 : /// wrote the Attribute.
1640 : #[test]
1641 4 : fn an_appended_attribute_is_stamped_to_the_same_depth_as_a_created_one() {
1642 4 : let attr = json!([{
1643 4 : "type": "Property",
1644 4 : "value": 21,
1645 4 : "unitCode": "CEL",
1646 4 : "https://example.org/accuracy": [{
1647 4 : "type": "Property",
1648 4 : "value": 0.5,
1649 4 : "https://example.org/basis": [{"type": "Property", "value": "spec"}]
1650 : }],
1651 4 : "https://example.org/provider": [{
1652 4 : "type": "Relationship",
1653 4 : "object": "urn:ngsi-ld:Sensor:1"
1654 : }]
1655 : }]);
1656 4 : let mut appended = attr.clone();
1657 4 : crate::stamp::stamp_instances(&mut appended, TS);
1658 :
1659 : // what 5.6.1 Create Entity produces for the same attribute
1660 4 : let mut created = json!({
1661 4 : "id": "urn:ngsi-ld:Room:1",
1662 4 : "type": "Room",
1663 4 : "https://example.org/temperature": attr
1664 : });
1665 4 : crate::stamp::stamp_new(&mut created, TS);
1666 4 : assert_eq!(
1667 4 : appended, created["https://example.org/temperature"],
1668 : "an appended attribute must carry the timestamps a created one carries"
1669 : );
1670 :
1671 : // and spelled out, so the shape is not merely equal to a wrong shape
1672 4 : let inst = &appended[0];
1673 4 : assert_eq!(inst["createdAt"], TS);
1674 4 : assert_eq!(inst["modifiedAt"], TS);
1675 4 : let acc = &inst["https://example.org/accuracy"][0];
1676 4 : assert_eq!(acc["createdAt"], TS, "sub-Property createdAt");
1677 4 : assert_eq!(acc["modifiedAt"], TS, "sub-Property modifiedAt");
1678 4 : assert_eq!(
1679 4 : acc["https://example.org/basis"][0]["createdAt"], TS,
1680 : "sub-Property of a sub-Property"
1681 : );
1682 4 : assert_eq!(
1683 4 : appended[0]["https://example.org/provider"][0]["createdAt"], TS,
1684 : "sub-Relationship createdAt"
1685 : );
1686 : // 4.8 NOTE 1: a TemporalProperty is not reified — the members that
1687 : // carry the Attribute's own value are left exactly as they arrived
1688 4 : assert_eq!(inst["value"], 21);
1689 4 : assert_eq!(inst["unitCode"], "CEL");
1690 4 : assert!(
1691 4 : !matches!(inst["unitCode"], Value::Array(_)),
1692 : "a reserved member is never walked as an instance array"
1693 : );
1694 4 : }
1695 : }
1696 :
1697 : #[cfg(test)]
1698 : mod clause_5_5_8 {
1699 : use super::merge_instance_sets;
1700 : use serde_json::{json, Value};
1701 :
1702 : /// 5.5.8 update algorithm: a Fragment member replaces the WHOLE matching
1703 : /// instance (unmapped sub-members are dropped — EXAMPLE 2), a different
1704 : /// datasetId is added as a new instance, an NGSI-LD Null deletes the
1705 : /// matching instance (EXAMPLE 3), and createdAt survives the replace.
1706 : #[test]
1707 4 : fn update_replaces_whole_instances_dataset_id_wise() {
1708 4 : let mut existing = json!([
1709 4 : {"type": "Property", "value": 25, "unitCode": "CEL",
1710 4 : "observedAt": "2022-03-14T01:59:26.535Z",
1711 4 : "createdAt": "2022-01-01T00:00:00Z"},
1712 4 : {"type": "Property", "value": 7, "datasetId": "urn:ngsi-ld:Dataset:a",
1713 4 : "createdAt": "2022-01-01T00:00:00Z"}
1714 : ]);
1715 : // default instance replaced wholesale; new datasetId appended
1716 4 : let incoming = json!([
1717 4 : {"type": "Property", "value": 100,
1718 4 : "observedAt": "2022-03-14T13:00:00.000Z"},
1719 4 : {"type": "Property", "value": 8, "datasetId": "urn:ngsi-ld:Dataset:b"}
1720 : ]);
1721 4 : assert!(merge_instance_sets(&mut existing, &incoming, false));
1722 4 : let arr = existing.as_array().unwrap();
1723 4 : assert_eq!(arr.len(), 3, "default replaced + a kept + b added");
1724 4 : let default = arr
1725 4 : .iter()
1726 4 : .find(|i| i.get("datasetId").is_none())
1727 4 : .expect("default instance");
1728 4 : assert_eq!(default["value"], 100);
1729 4 : assert!(
1730 4 : default.get("unitCode").is_none(),
1731 : "EXAMPLE 2: whole-attribute replace drops unitCode"
1732 : );
1733 4 : assert_eq!(
1734 4 : default["createdAt"], "2022-01-01T00:00:00Z",
1735 : "createdAt survives the replace"
1736 : );
1737 : // NGSI-LD Null deletes exactly the matching datasetId instance
1738 4 : let deletion = json!([
1739 4 : {"type": "Property", "value": "urn:ngsi-ld:null",
1740 4 : "datasetId": "urn:ngsi-ld:Dataset:a"}
1741 : ]);
1742 4 : assert!(merge_instance_sets(&mut existing, &deletion, false));
1743 4 : let arr = existing.as_array().unwrap();
1744 4 : assert_eq!(arr.len(), 2);
1745 4 : assert!(
1746 8 : !arr.iter().any(|i| {
1747 8 : i.get("datasetId").and_then(Value::as_str) == Some("urn:ngsi-ld:Dataset:a")
1748 8 : }),
1749 : "EXAMPLE 3: null deletes the instance"
1750 : );
1751 4 : assert!(
1752 4 : !existing.to_string().contains("urn:ngsi-ld:null"),
1753 : "the sentinel never persists"
1754 : );
1755 4 : }
1756 :
1757 : /// 5.5.8 (noOverwrite append, 5.6.3): an existing instance is NOT
1758 : /// replaced, an absent one is still added.
1759 : #[test]
1760 4 : fn no_overwrite_keeps_existing_instances() {
1761 4 : let mut existing = json!([{"type": "Property", "value": 1}]);
1762 4 : let incoming = json!([
1763 4 : {"type": "Property", "value": 2},
1764 4 : {"type": "Property", "value": 3, "datasetId": "urn:ngsi-ld:Dataset:n"}
1765 : ]);
1766 4 : assert!(merge_instance_sets(&mut existing, &incoming, true));
1767 4 : let arr = existing.as_array().unwrap();
1768 4 : assert_eq!(arr.len(), 2);
1769 4 : assert_eq!(
1770 4 : arr[0]["value"], 1,
1771 : "noOverwrite leaves the existing default instance"
1772 : );
1773 4 : assert_eq!(arr[1]["value"], 3);
1774 4 : }
1775 : }
1776 :
1777 : #[cfg(test)]
1778 : mod update_result_tests {
1779 : use super::*;
1780 : use crate::federation::{FedReg, Part};
1781 : use http_body_util::BodyExt;
1782 :
1783 40 : fn reg(reg_id: &str, attrs: Option<Vec<String>>) -> FedReg {
1784 40 : FedReg {
1785 40 : reg_id: reg_id.into(),
1786 40 : endpoint: "http://peer:9090".into(),
1787 40 : mode: "inclusive".into(),
1788 40 : attrs,
1789 40 : ..Default::default()
1790 40 : }
1791 40 : }
1792 :
1793 28 : async fn body_of(resp: Response) -> Value {
1794 28 : let bytes = resp.into_body().collect().await.expect("body").to_bytes();
1795 28 : serde_json::from_slice(&bytes).expect("json")
1796 28 : }
1797 :
1798 : /// Tables 6.6.3.1-2 / 6.7.3.1-2 — the /attrs 207 body is an
1799 : /// UpdateResult (updated: String[], notUpdated: NotUpdatedDetails[] with
1800 : /// mandatory attributeName+reason, optional registrationId), never the
1801 : /// batch {success, errors} shape.
1802 : #[tokio::test]
1803 4 : async fn distributed_attr_207_is_an_update_result() {
1804 4 : let tenant = antares_model::TenantId::new("default").expect("tenant");
1805 4 : let speed = "https://uri.etsi.org/ngsi-ld/default-context/speed".to_owned();
1806 4 : let brand = "https://uri.etsi.org/ngsi-ld/default-context/brandName".to_owned();
1807 4 : let all = vec![speed.clone(), brand.clone()];
1808 4 : let regs = vec![reg(
1809 4 : "urn:ngsi-ld:ContextSourceRegistration:csr1",
1810 4 : Some(vec![brand.clone()]),
1811 : )];
1812 4 : let parts = vec![Part {
1813 4 : status: 504,
1814 4 : detail: "distributed operation timed out".into(),
1815 4 : }];
1816 4 : let resp = combine_attr_parts(
1817 4 : &tenant,
1818 4 : &all,
1819 4 : std::slice::from_ref(&speed),
1820 4 : LocalOutcome::Ok,
1821 4 : vec![speed.clone()],
1822 4 : Vec::new(),
1823 4 : ®s,
1824 4 : &parts,
1825 : );
1826 4 : assert_eq!(resp.status().as_u16(), 207);
1827 4 : let body = body_of(resp).await;
1828 4 : assert_eq!(body["updated"], serde_json::json!([speed]));
1829 4 : assert_eq!(body["notUpdated"][0]["attributeName"], brand);
1830 4 : assert_eq!(
1831 4 : body["notUpdated"][0]["registrationId"],
1832 : "urn:ngsi-ld:ContextSourceRegistration:csr1"
1833 : );
1834 4 : assert!(body["notUpdated"][0]["reason"].is_string());
1835 4 : assert!(body.get("success").is_none(), "not the batch shape");
1836 4 : assert!(body.get("errors").is_none(), "not the batch shape");
1837 4 : }
1838 :
1839 : /// 6.3.17: "In the case of an exclusive or redirect registration, where
1840 : /// all of the data is held outside of the Context Broker and held in a
1841 : /// single registered source, the following errors shall be returned: 508
1842 : /// Loop Detected … 504 Gateway Timeout … 404 Not Found … 502 Bad
1843 : /// Gateway." 207 Multi-Status is reserved for an entity distributed over
1844 : /// multiple endpoints, so a single proxied source's failure passes
1845 : /// through as itself.
1846 : #[tokio::test]
1847 4 : async fn single_proxied_source_error_passes_through_instead_of_207() {
1848 4 : let tenant = antares_model::TenantId::new("default").expect("tenant");
1849 4 : let speed = "https://uri.etsi.org/ngsi-ld/default-context/speed".to_owned();
1850 8 : for mode in ["exclusive", "redirect"] {
1851 24 : for (status, etype) in [
1852 8 : (404u16, "ResourceNotFound"),
1853 8 : (504, "InternalError"),
1854 8 : (502, "InternalError"),
1855 8 : ] {
1856 24 : let regs = vec![FedReg {
1857 24 : mode: mode.into(),
1858 24 : ..reg("urn:ngsi-ld:ContextSourceRegistration:csr1", None)
1859 24 : }];
1860 24 : let parts = vec![Part {
1861 24 : status,
1862 24 : detail: "distributed operation to registration csr1 failed".into(),
1863 24 : }];
1864 24 : let resp = combine_attr_parts(
1865 24 : &tenant,
1866 24 : std::slice::from_ref(&speed),
1867 24 : &[],
1868 4 : // every attribute is held by the proxy: no local half ran
1869 24 : LocalOutcome::Skipped,
1870 24 : Vec::new(),
1871 24 : Vec::new(),
1872 24 : ®s,
1873 24 : &parts,
1874 4 : );
1875 24 : assert_eq!(resp.status().as_u16(), status, "{mode} part {status}");
1876 24 : let body = body_of(resp).await;
1877 24 : assert_eq!(
1878 24 : body["type"],
1879 24 : format!("https://uri.etsi.org/ngsi-ld/errors/{etype}"),
1880 4 : "{mode} part {status}"
1881 4 : );
1882 24 : assert_eq!(body["status"], status);
1883 24 : assert!(
1884 24 : body.get("updated").is_none() && body.get("notUpdated").is_none(),
1885 4 : "a single-source failure is ProblemDetails, not an UpdateResult"
1886 4 : );
1887 4 : }
1888 4 : }
1889 4 : }
1890 :
1891 : /// The same single-source rule must not swallow a MULTI-endpoint failure:
1892 : /// two inclusive registrations over one entity stay 207 (6.3.17).
1893 : #[tokio::test]
1894 4 : async fn two_registrations_still_answer_207() {
1895 4 : let tenant = antares_model::TenantId::new("default").expect("tenant");
1896 4 : let speed = "https://uri.etsi.org/ngsi-ld/default-context/speed".to_owned();
1897 4 : let regs = vec![
1898 4 : reg("urn:ngsi-ld:ContextSourceRegistration:csr1", None),
1899 4 : reg("urn:ngsi-ld:ContextSourceRegistration:csr2", None),
1900 : ];
1901 4 : let parts = vec![
1902 4 : Part {
1903 4 : status: 204,
1904 4 : detail: "ok".into(),
1905 4 : },
1906 4 : Part {
1907 4 : status: 404,
1908 4 : detail: "not found".into(),
1909 4 : },
1910 : ];
1911 4 : let resp = combine_attr_parts(
1912 4 : &tenant,
1913 4 : std::slice::from_ref(&speed),
1914 4 : &[],
1915 4 : LocalOutcome::Skipped,
1916 4 : Vec::new(),
1917 4 : Vec::new(),
1918 4 : ®s,
1919 4 : &parts,
1920 : );
1921 4 : assert_eq!(resp.status().as_u16(), 207);
1922 4 : }
1923 :
1924 : /// All halves succeeded → 204; everything missing → 404.
1925 : #[tokio::test]
1926 4 : async fn distributed_attr_success_and_not_found_edges() {
1927 4 : let tenant = antares_model::TenantId::new("default").expect("tenant");
1928 4 : let speed = "https://uri.etsi.org/ngsi-ld/default-context/speed".to_owned();
1929 4 : let regs = vec![reg("urn:ngsi-ld:ContextSourceRegistration:csr1", None)];
1930 4 : let ok_parts = vec![Part {
1931 4 : status: 204,
1932 4 : detail: "ok".into(),
1933 4 : }];
1934 4 : let resp = combine_attr_parts(
1935 4 : &tenant,
1936 4 : std::slice::from_ref(&speed),
1937 4 : std::slice::from_ref(&speed),
1938 4 : LocalOutcome::Ok,
1939 4 : vec![speed.clone()],
1940 4 : Vec::new(),
1941 4 : ®s,
1942 4 : &ok_parts,
1943 : );
1944 4 : assert_eq!(resp.status().as_u16(), 204);
1945 : // entity unknown locally AND every forward failed → 404 ProblemDetails
1946 4 : let bad_parts = vec![Part {
1947 4 : status: 404,
1948 4 : detail: "not found".into(),
1949 4 : }];
1950 4 : let resp = combine_attr_parts(
1951 4 : &tenant,
1952 4 : std::slice::from_ref(&speed),
1953 4 : std::slice::from_ref(&speed),
1954 4 : LocalOutcome::NotFound("entity urn:x not found".into()),
1955 4 : Vec::new(),
1956 4 : Vec::new(),
1957 4 : ®s,
1958 4 : &bad_parts,
1959 : );
1960 4 : assert_eq!(resp.status().as_u16(), 404);
1961 4 : }
1962 : }
|
