Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Temporal writes: upsert (5.6.11), attrs (5.6.12, 5.6.13), instances (5.6.14, 5.6.15), delete (5.6.16).
3 :
4 : use super::*;
5 :
6 : /// 5.6.11 input: the pushed Temporal Evolution may carry the 4.5.7
7 : /// deleted-instance representation (value = NGSI-LD Null), which 5.5.4
8 : /// explicitly excepts for "the temporal evolution" — hence allow_null.
9 : pub(super) const TEMPORAL_OPTS: ExpandOpts = ExpandOpts {
10 : fragment: false,
11 : allow_null: true,
12 : merge: false,
13 : temporal: true,
14 : sys: false,
15 : };
16 :
17 : /// 4.5.7 and 4.5.8: each recorded instance of a Property or Relationship
18 : /// carries an instanceId and the 4.8 timestamps. The id is minted here and
19 : /// only when the pushed instance brought none, because the temporal API is
20 : /// add-only (5.6.11.4, 5.6.12.1) and 5.6.14/5.6.15 address an instance by
21 : /// that id. Sub-Attributes are part of the instance and are not stamped
22 : /// separately -- the current-state path stamps them one level deeper
23 : /// (`crate::stamp::stamp_instances`).
24 1274 : pub(super) fn stamp_temporal_instances(doc: &mut Value, ts: &str) {
25 1274 : if let Some(obj) = doc.as_object_mut() {
26 4768 : for (k, v) in obj.iter_mut() {
27 4768 : if is_meta(k) {
28 2556 : continue;
29 2212 : }
30 2212 : if let Some(arr) = v.as_array_mut() {
31 10324 : for inst in arr {
32 10324 : if let Some(o) = inst.as_object_mut() {
33 10324 : o.entry("instanceId".to_owned()).or_insert_with(|| {
34 10324 : Value::String(format!("urn:ngsi-ld:Instance:{}", uuid::Uuid::new_v4()))
35 10324 : });
36 10324 : o.insert("createdAt".into(), Value::String(ts.to_owned()));
37 10324 : o.insert("modifiedAt".into(), Value::String(ts.to_owned()));
38 0 : }
39 : }
40 0 : }
41 : }
42 0 : }
43 1274 : }
44 :
45 : // ---------- POST /temporal/entities/ — Upsert temporal (5.6.11) ----------
46 :
47 1286 : pub async fn upsert_temporal(
48 1286 : State(st): State<AppState>,
49 1286 : CleanParams(params): CleanParams,
50 1286 : headers: HeaderMap,
51 1286 : body: Bytes,
52 1286 : ) -> Response {
53 1286 : let go = async {
54 1286 : let tenant = tenant_from(&headers)?;
55 1286 : check_params(¶ms, &["options", "local"])?;
56 1282 : let parsed = parse_body(&st, &headers, &body, BodyKind::Standard).await?;
57 1266 : let obj = parsed.object(NgsiError::BadRequestData(
58 1266 : "temporal entity must be a JSON object".into(),
59 1266 : ))?;
60 1266 : let expanded = expand_entity(obj, &parsed.ctx, TEMPORAL_OPTS)?;
61 1250 : let id = antares_jsonld::expanded_id(&expanded)?.to_owned();
62 : // the engine is asked with what 5.6.1 gives it: types, Attributes, body
63 1250 : let types: Vec<String> = expanded
64 1250 : .get("type")
65 1250 : .and_then(Value::as_array)
66 1250 : .map(|a| {
67 1250 : a.iter()
68 1250 : .filter_map(Value::as_str)
69 1250 : .map(str::to_owned)
70 1250 : .collect()
71 1250 : })
72 1250 : .unwrap_or_default();
73 1250 : let attrs: Vec<String> = expanded
74 1250 : .as_object()
75 4760 : .map(|o| o.keys().filter(|k| !is_meta(k)).cloned().collect())
76 1250 : .unwrap_or_default();
77 1250 : gate!(
78 : st, &tenant, &headers, "5.6.11",
79 : ids: &[&id],
80 : types: &types,
81 : attrs: &attrs,
82 : body: Some(&expanded),
83 : )
84 1250 : .await?;
85 : // the create rule on a comma id holds for a new Entity,
86 : // judged after the gate so a refused caller learns nothing stored;
87 : // a stored legacy one is still upserted
88 1250 : if id.contains(',') && st.temporal.get(&tenant, &id).await?.is_none() {
89 4 : antares_jsonld::new_entity_id(&expanded)?;
90 1246 : }
91 : // 5.6.11.4: exclusive/redirect registrations matching the input are
92 : // forwarded when "Create or Update Temporal" is supported; proxy
93 : // modes without it are an error of type Conflict; inclusive ones
94 : // forward when supported. Matching attributes are removed from the
95 : // local fragment.
96 1246 : let regs =
97 1246 : match temporal_write_regs(&st, &tenant, &headers, &parsed.ctx, ¶ms, &id).await {
98 1246 : Ok(regs) => regs,
99 0 : Err(refused) => return Ok(*refused),
100 : };
101 1246 : if !regs.is_empty() {
102 4 : let (mut parts, fwd) =
103 4 : crate::federation::scoped_fragments(®s, "upsertTemporal", obj, &parsed.ctx);
104 4 : let proxies: Vec<&crate::federation::FedReg> =
105 4 : regs.iter().filter(|r| r.is_proxy()).collect();
106 4 : let (rest, has_attrs) = crate::federation::strip_proxied(obj, &proxies, &parsed.ctx);
107 4 : if has_attrs || proxies.is_empty() {
108 0 : let local = expand_entity(&rest, &parsed.ctx, TEMPORAL_OPTS)?;
109 0 : let status = upsert_temporal_local(&st, &tenant, &id, local).await?;
110 0 : parts.push(crate::federation::Part::local(
111 0 : status.as_u16(),
112 : "local temporal upsert",
113 : ));
114 4 : }
115 4 : let ctx_url = crate::federation::ctx_link_url(&headers, &parsed.ctx.source);
116 4 : for (reg, frag) in fwd {
117 2 : parts.push(
118 2 : crate::federation::forward_part(
119 2 : &st,
120 2 : reqwest::Method::POST,
121 2 : format!("{}/ngsi-ld/v1/temporal/entities", reg.endpoint),
122 2 : &[],
123 2 : &headers,
124 2 : &tenant,
125 2 : ®,
126 2 : &ctx_url,
127 2 : Some(frag),
128 2 : )
129 2 : .await,
130 : );
131 : }
132 4 : return Ok(crate::federation::combine(
133 4 : st.profile,
134 4 : Some(&id),
135 4 : parts,
136 4 : created(
137 4 : format!(
138 4 : "/ngsi-ld/v1/temporal/entities/{}",
139 4 : crate::federation::path_segment(&id)
140 4 : ),
141 4 : &tenant,
142 4 : ),
143 4 : &tenant,
144 4 : ));
145 1242 : }
146 1242 : let status = upsert_temporal_local(&st, &tenant, &id, expanded).await?;
147 1242 : Ok::<_, ApiError>(if status == StatusCode::CREATED {
148 1158 : created(
149 1158 : format!(
150 : "/ngsi-ld/v1/temporal/entities/{}",
151 1158 : crate::federation::path_segment(&id)
152 : ),
153 1158 : &tenant,
154 : )
155 : } else {
156 84 : no_content(&tenant)
157 : })
158 1286 : };
159 1286 : go.await.unwrap_or_else(|e| e.into_response())
160 1286 : }
161 :
162 : /// 5.6.11.4 local half: create the Temporal Evolution, or add the provided
163 : /// instances to the existing one per 5.6.12 (merge key = datasetId +
164 : /// observedAt) with Entity Type names unioned. Returns 201 vs 204.
165 1242 : pub(super) async fn upsert_temporal_local(
166 1242 : st: &AppState,
167 1242 : tenant: &antares_model::TenantId,
168 1242 : id: &str,
169 1242 : mut expanded: Value,
170 1242 : ) -> ApiResult<StatusCode> {
171 1242 : let ts = now_iso();
172 1242 : stamp_temporal_instances(&mut expanded, &ts);
173 : // get->create/mutate is a TOCTOU pair: two concurrent first-upserts
174 : // both see "absent", and the loser's create must NOT be silently
175 : // dropped (201 with a discarded payload). Loop: a lost create retries
176 : // as a merge, a mutate on a just-deleted doc retries as a create.
177 1242 : let mut attempts: u8 = 0;
178 : loop {
179 1242 : attempts = attempts.saturating_add(1);
180 1242 : if attempts > 16 {
181 0 : return Err(NgsiError::InternalError("upsert retry storm".into()).into());
182 1242 : }
183 1242 : let existed = st.temporal.get(tenant, id).await?.is_some();
184 1242 : if existed {
185 84 : let res = st
186 84 : .temporal
187 84 : .mutate(tenant, id, |doc| {
188 84 : let target = antares_store::stored_object(doc)?;
189 : // 5.6.11.4: new Entity Type names are added to the target
190 84 : if let Some(new_types) = expanded.get("type").and_then(Value::as_array) {
191 84 : let mut cur: Vec<Value> = target
192 84 : .get("type")
193 84 : .and_then(Value::as_array)
194 84 : .cloned()
195 84 : .unwrap_or_default();
196 92 : for t in new_types {
197 92 : if !cur.contains(t) {
198 8 : cur.push(t.clone());
199 84 : }
200 : }
201 84 : target.insert("type".into(), Value::Array(cur));
202 0 : }
203 : // 5.6.11.4: the instances "shall be added" as 5.6.12
204 : // mandates; identity is the instanceId, never observedAt
205 84 : antares_jsonld::expanded_object(&expanded)?;
206 84 : add_temporal_instances(doc, &expanded, &ts);
207 84 : Ok::<(), NgsiError>(())
208 84 : })
209 84 : .await?;
210 84 : match res {
211 0 : Some(Err(e)) => return Err(ApiError::from(e)),
212 84 : Some(Ok(())) => return Ok(StatusCode::NO_CONTENT),
213 0 : None => continue, // deleted between get and mutate - retry as create
214 : }
215 : } else {
216 1158 : let mut doc = expanded.clone();
217 1158 : if let Some(o) = doc.as_object_mut() {
218 1158 : o.insert("createdAt".into(), Value::String(ts.clone()));
219 1158 : o.insert("modifiedAt".into(), Value::String(ts.clone()));
220 1158 : }
221 1158 : if st.temporal.create(tenant, id, doc).await? {
222 1158 : return Ok(StatusCode::CREATED);
223 0 : }
224 : // lost the create race - the doc exists now; retry as a merge
225 : }
226 : }
227 1242 : }
228 :
229 : // ---------- DELETE /temporal/entities/{id} (5.6.16) ----------
230 :
231 3568 : pub async fn delete_temporal(
232 3568 : State(st): State<AppState>,
233 3568 : Path(id): Path<String>,
234 3568 : CleanParams(params): CleanParams,
235 3568 : headers: HeaderMap,
236 3568 : ) -> Response {
237 3568 : let go = async {
238 3568 : let tenant = tenant_from(&headers)?;
239 3568 : antares_model::EntityId::new(&id)?;
240 3556 : check_params(¶ms, &["local"])?;
241 : // 5.6.16.4: forward to registrations supporting the operation;
242 : // unsupported proxy modes are Conflict.
243 3556 : let ctx = st.loader.core();
244 3556 : temporal_write_gate(&st, &tenant, &headers, &id, "5.6.16", &[], None).await?;
245 3548 : let regs = match temporal_write_regs(&st, &tenant, &headers, &ctx, ¶ms, &id).await {
246 3546 : Ok(regs) => regs,
247 2 : Err(refused) => return Ok(*refused),
248 : };
249 3546 : let deleted = st.temporal.delete(&tenant, &id).await?;
250 3546 : answer_temporal_attr_write(
251 3546 : &st,
252 3546 : &tenant,
253 3546 : &headers,
254 3546 : &ctx,
255 3546 : &id,
256 3546 : "deleteTemporal",
257 3546 : reqwest::Method::DELETE,
258 3546 : "",
259 3546 : None,
260 3546 : regs,
261 3546 : LocalWrite {
262 3546 : res: deleted.then_some(Ok(())),
263 3546 : found: deleted,
264 3546 : missing: format!("temporal entity {id}"),
265 3546 : applied: "deleted locally",
266 3546 : },
267 3546 : )
268 3546 : .await
269 3568 : };
270 3568 : go.await.unwrap_or_else(|e| e.into_response())
271 3568 : }
272 :
273 : // ---------- POST /temporal/entities/{id}/attrs/ (5.6.12) ----------
274 :
275 54 : pub async fn add_temporal_attrs(
276 54 : State(st): State<AppState>,
277 54 : Path(id): Path<String>,
278 54 : CleanParams(params): CleanParams,
279 54 : headers: HeaderMap,
280 54 : body: Bytes,
281 54 : ) -> Response {
282 54 : let go = async {
283 54 : let tenant = tenant_from(&headers)?;
284 54 : antares_model::EntityId::new(&id)?;
285 46 : check_params(¶ms, &["local"])?;
286 46 : let parsed = parse_body(&st, &headers, &body, BodyKind::Standard).await?;
287 38 : let obj = parsed.object(NgsiError::BadRequestData(
288 38 : "fragment must be a JSON object".into(),
289 38 : ))?;
290 : // 5.6.12 input is pushed history — the 4.5.7 deleted-instance
291 : // representation is legal (5.5.4 temporal exception), hence
292 : // allow_null (mirrors 5.6.11).
293 38 : let mut expanded = expand_entity(
294 38 : obj,
295 38 : &parsed.ctx,
296 38 : ExpandOpts {
297 38 : fragment: true,
298 38 : allow_null: true,
299 38 : temporal: true,
300 38 : ..Default::default()
301 38 : },
302 0 : )?;
303 : // the engine is asked with what 5.6.3 gives it: the Attributes, the body
304 38 : let cond = temporal_write_gate(
305 38 : &st,
306 38 : &tenant,
307 38 : &headers,
308 38 : &id,
309 38 : "5.6.12",
310 38 : &crate::attrs::attr_iris_of(&expanded),
311 38 : Some(&expanded),
312 38 : )
313 38 : .await?;
314 : // 5.6.12.4: forwarding — proxy modes without appendAttrsTemporal
315 : // are Conflict; supporting registrations receive the fragment and
316 : // the matching attributes are stripped from the local half.
317 36 : let regs =
318 36 : match temporal_write_regs(&st, &tenant, &headers, &parsed.ctx, ¶ms, &id).await {
319 36 : Ok(regs) => regs,
320 0 : Err(refused) => return Ok(*refused),
321 : };
322 36 : crate::policy::Condition::unforwarded(cond.as_ref(), !regs.is_empty())?;
323 36 : if !regs.is_empty() {
324 4 : let (mut parts, fwd) =
325 4 : crate::federation::scoped_fragments(®s, "appendAttrsTemporal", obj, &parsed.ctx);
326 4 : let proxies: Vec<&crate::federation::FedReg> =
327 4 : regs.iter().filter(|r| r.is_proxy()).collect();
328 4 : let (rest, has_attrs) = crate::federation::strip_proxied(obj, &proxies, &parsed.ctx);
329 4 : if has_attrs || proxies.is_empty() {
330 0 : let mut local = expand_entity(
331 0 : &rest,
332 0 : &parsed.ctx,
333 0 : ExpandOpts {
334 0 : fragment: true,
335 0 : allow_null: true,
336 0 : temporal: true,
337 0 : ..Default::default()
338 0 : },
339 0 : )?;
340 0 : let ts = now_iso();
341 0 : stamp_temporal_instances(&mut local, &ts);
342 0 : let res = st
343 0 : .temporal
344 0 : .mutate(&tenant, &id, |doc| {
345 0 : add_temporal_instances(doc, &local, &ts);
346 0 : Ok::<(), NgsiError>(())
347 0 : })
348 0 : .await?;
349 0 : parts.push(match res {
350 0 : Some(Ok(())) => crate::federation::Part::local(204, "added locally"),
351 0 : _ => crate::federation::Part::local(
352 : 404,
353 0 : format!("temporal entity {id} not found locally"),
354 : ),
355 : });
356 4 : }
357 4 : let ctx_url = crate::federation::ctx_link_url(&headers, &parsed.ctx.source);
358 4 : for (reg, frag) in fwd {
359 2 : parts.push(
360 2 : crate::federation::forward_part(
361 2 : &st,
362 2 : reqwest::Method::POST,
363 2 : format!(
364 2 : "{}/ngsi-ld/v1/temporal/entities/{}/attrs",
365 2 : reg.endpoint,
366 2 : crate::federation::path_segment(&id)
367 2 : ),
368 2 : &[],
369 2 : &headers,
370 2 : &tenant,
371 2 : ®,
372 2 : &ctx_url,
373 2 : Some(frag),
374 2 : )
375 2 : .await,
376 : );
377 : }
378 4 : return Ok(crate::federation::combine(
379 4 : st.profile,
380 4 : Some(&id),
381 4 : parts,
382 4 : no_content(&tenant),
383 4 : &tenant,
384 4 : ));
385 32 : }
386 32 : let ts = now_iso();
387 32 : stamp_temporal_instances(&mut expanded, &ts);
388 32 : let res = st
389 32 : .temporal
390 32 : .mutate(&tenant, &id, |doc| {
391 18 : crate::policy::Condition::pre_evolution(cond.as_ref(), &id, doc)?;
392 14 : add_temporal_instances(doc, &expanded, &ts);
393 14 : Ok::<(), NgsiError>(())
394 18 : })
395 32 : .await?;
396 16 : match res {
397 : None => {
398 14 : Err(NgsiError::ResourceNotFound(format!("temporal entity {id} not found")).into())
399 : }
400 4 : Some(Err(e)) => Err(ApiError::from(e)),
401 12 : Some(Ok(())) => Ok(no_content(&tenant)),
402 : }
403 54 : };
404 54 : go.await.unwrap_or_else(|e| e.into_response())
405 54 : }
406 :
407 : /// 5.6.12.4: append the fragment's Attribute instances to the Temporal
408 : /// Evolution (instances accumulate — history is never overwritten here).
409 106 : pub(super) fn add_temporal_instances(doc: &mut Value, expanded: &Value, ts: &str) {
410 106 : let Some(target) = doc.as_object_mut() else {
411 0 : return;
412 : };
413 : // 5.6.12.4: "If type is included in the EntityTemporal Fragment and it
414 : // includes Entity Type names that are not yet in the target ..., add
415 : // them to the list of Entity Type names"
416 106 : if let Some(Value::Array(add)) = expanded.get("type") {
417 88 : let mut cur = match target.remove("type") {
418 88 : Some(Value::Array(a)) => a,
419 0 : Some(t) => vec![t],
420 0 : None => Vec::new(),
421 : };
422 100 : for t in add {
423 100 : if !cur.contains(t) {
424 4 : cur.push(t.clone());
425 96 : }
426 : }
427 88 : target.insert("type".into(), Value::Array(cur));
428 18 : }
429 366 : for (k, v) in expanded.as_object().into_iter().flatten() {
430 366 : if is_meta(k) {
431 192 : continue;
432 174 : }
433 174 : let incoming = v.as_array().cloned().unwrap_or_default();
434 174 : match target.get_mut(k).and_then(Value::as_array_mut) {
435 158 : Some(cur) => cur.extend(incoming),
436 16 : None => {
437 16 : target.insert(k.clone(), Value::Array(incoming));
438 16 : }
439 : }
440 : }
441 106 : target.insert("modifiedAt".into(), Value::String(ts.to_owned()));
442 106 : }
443 :
444 : // ---------- DELETE /temporal/entities/{id}/attrs/{attrId} (5.6.13) ----------
445 :
446 104 : pub async fn delete_temporal_attr(
447 104 : State(st): State<AppState>,
448 104 : Path((id, attr)): Path<(String, String)>,
449 104 : CleanParams(params): CleanParams,
450 104 : headers: HeaderMap,
451 104 : ) -> Response {
452 104 : let go = async {
453 104 : let tenant = tenant_from(&headers)?;
454 104 : antares_model::EntityId::new(&id)?;
455 : // 5.6.13.4: "If the target Attribute name is not a valid name, then an
456 : // error of type BadRequestData shall be raised." The shared guard is
457 : // the one that also refuses dot segments — the name is interpolated
458 : // into the forwarded request path, where `..` addresses the peer's
459 : // Temporal Evolution resource instead of its attribute.
460 96 : antares_model::check_attr_name(&attr)?;
461 64 : check_params(¶ms, &["datasetId", "deleteAll", "local"])?;
462 64 : let ctx = request_context(&st.loader, &headers).await?;
463 64 : let attr_iri = antares_jsonld::expand_attr_name(&attr, &ctx)?;
464 62 : let cond = temporal_write_gate(
465 62 : &st,
466 62 : &tenant,
467 62 : &headers,
468 62 : &id,
469 62 : "5.6.13",
470 62 : std::slice::from_ref(&attr_iri),
471 62 : None,
472 62 : )
473 62 : .await?;
474 62 : let delete_all = params.get("deleteAll").map(String::as_str) == Some("true");
475 62 : let want_ds = crate::repr::target_dataset_id(¶ms).map(String::from);
476 62 : let regs = match temporal_write_regs(&st, &tenant, &headers, &ctx, ¶ms, &id).await {
477 60 : Ok(regs) => regs,
478 2 : Err(refused) => return Ok(*refused),
479 : };
480 60 : crate::policy::Condition::unforwarded(cond.as_ref(), !regs.is_empty())?;
481 58 : let mut found = false;
482 58 : let ts = now_iso();
483 58 : let res = st
484 58 : .temporal
485 58 : .mutate(&tenant, &id, |doc| {
486 32 : crate::policy::Condition::pre_evolution(cond.as_ref(), &id, doc)?;
487 28 : let target = antares_store::stored_object(doc)?;
488 28 : if delete_all
489 24 : || (want_ds.is_none()
490 20 : && !target
491 20 : .get(&attr_iri)
492 20 : .and_then(Value::as_array)
493 26 : .is_some_and(|a| a.iter().any(|i| i.get("datasetId").is_some())))
494 : {
495 : // deleteAll, or single-instance-set attribute: drop it whole
496 18 : if target.remove(&attr_iri).is_some() {
497 12 : found = true;
498 12 : }
499 : } else {
500 : // 5.6.13: only the matching datasetId instance set is deleted
501 28 : found = remove_instances(target, &attr_iri, |i| {
502 28 : i.get("datasetId").and_then(Value::as_str) == want_ds.as_deref()
503 28 : });
504 : }
505 28 : if found {
506 22 : target.insert("modifiedAt".into(), Value::String(ts.clone()));
507 22 : }
508 28 : Ok::<(), NgsiError>(())
509 32 : })
510 58 : .await?;
511 58 : answer_temporal_attr_write(
512 58 : &st,
513 58 : &tenant,
514 58 : &headers,
515 58 : &ctx,
516 58 : &id,
517 58 : "deleteAttrsTemporal",
518 58 : reqwest::Method::DELETE,
519 58 : &format!("/attrs/{}", crate::federation::path_segment(&attr)),
520 58 : None,
521 58 : regs,
522 58 : LocalWrite {
523 58 : res,
524 58 : found,
525 58 : missing: format!("attribute {attr}"),
526 58 : applied: "deleted locally",
527 58 : },
528 58 : )
529 58 : .await
530 104 : };
531 104 : go.await.unwrap_or_else(|e| e.into_response())
532 104 : }
533 :
534 : /// Removes the instances of `attr_iri` that `doomed` picks, and the
535 : /// Attribute with its last instance; whether any instance went.
536 34 : fn remove_instances(
537 34 : target: &mut serde_json::Map<String, Value>,
538 34 : attr_iri: &str,
539 34 : doomed: impl Fn(&Value) -> bool,
540 34 : ) -> bool {
541 34 : let Some(arr) = target.get_mut(attr_iri).and_then(Value::as_array_mut) else {
542 4 : return false;
543 : };
544 30 : let before = arr.len();
545 62 : arr.retain(|i| !doomed(i));
546 30 : let found = arr.len() != before;
547 30 : if arr.is_empty() {
548 6 : target.remove(attr_iri);
549 24 : }
550 30 : found
551 34 : }
552 :
553 : /// The policy gate of a write to one Temporal Evolution under `clause`,
554 : /// judged on its stored image, the expanded Attributes it writes and its
555 : /// expanded body, as the Core API twin of the write is: the condition a
556 : /// narrowing engine answered.
557 3760 : async fn temporal_write_gate(
558 3760 : st: &AppState,
559 3760 : tenant: &antares_model::TenantId,
560 3760 : headers: &HeaderMap,
561 3760 : id: &str,
562 3760 : clause: &'static str,
563 3760 : written: &[String],
564 3760 : fragment: Option<&Value>,
565 3760 : ) -> Result<Option<crate::policy::Condition>, ApiError> {
566 3760 : let pre = crate::policy::temporal_pre_image(st, tenant, id).await?;
567 3758 : let granted = gate!(
568 : st, tenant, headers, clause,
569 : ids: &[id],
570 : types: &pre.types,
571 : stored: pre.stored.as_ref(),
572 : attrs: written,
573 : body: fragment,
574 : )
575 3758 : .await?;
576 3750 : Ok(crate::policy::Condition::of(&granted))
577 3760 : }
578 :
579 : /// 5.6.13 / 5.6.14 / 5.6.15 / 5.6.16 answer: the local result becomes one
580 : /// 4.3.6 Part, every registration supporting `op` (Table 4.20-1) contributes
581 : /// another, and 4.3.6.4 combines them into the one status the client sees.
582 : /// With no registration to forward to — the common case — the local result
583 : /// is the whole answer. The four operations differ in what they change and
584 : /// in what they forward, never in how the two halves are answered together.
585 : /// The registrations a 5.6.13/5.6.14/5.6.15/5.6.16 write forwards to, with
586 : /// 6.3.17/6.3.18 loop handling already applied. The check belongs here,
587 : /// ahead of the local write: 508 Loop Detected is an error status, so the
588 : /// request it answers has to leave the Temporal Evolution as it found it.
589 : /// The refusal travels boxed — a whole `Response` in an `Err` makes every
590 : /// `Ok` of this function carry its width.
591 4996 : pub(super) async fn temporal_write_regs(
592 4996 : st: &AppState,
593 4996 : tenant: &antares_model::TenantId,
594 4996 : headers: &HeaderMap,
595 4996 : ctx: &antares_jsonld::Context,
596 4996 : params: &HashMap<String, String>,
597 4996 : id: &str,
598 4996 : ) -> Result<Vec<crate::federation::FedReg>, Box<Response>> {
599 4996 : match crate::federation::write_plan_for_id(st, tenant, id, ctx, params, headers)
600 4996 : .await
601 4996 : .map_err(|e| Box::new(crate::negotiate::ApiError::from(e).into_response()))?
602 : {
603 8 : crate::federation::WritePlan::Answered(refused) => Err(refused),
604 4988 : crate::federation::WritePlan::Forward(regs) => Ok(regs),
605 : }
606 4996 : }
607 :
608 : #[allow(clippy::too_many_arguments)] // mirrors the wire: one param per forwarded request part
609 3704 : pub(super) async fn answer_temporal_attr_write(
610 3704 : st: &AppState,
611 3704 : tenant: &antares_model::TenantId,
612 3704 : headers: &HeaderMap,
613 3704 : ctx: &antares_jsonld::Context,
614 3704 : id: &str,
615 3704 : op: &str,
616 3704 : method: reqwest::Method,
617 3704 : path_suffix: &str,
618 3704 : body: Option<Value>,
619 3704 : regs: Vec<crate::federation::FedReg>,
620 3704 : local: LocalWrite,
621 3704 : ) -> ApiResult<Response> {
622 3704 : if regs.is_empty() {
623 898 : return match local.res {
624 : None => {
625 2674 : Err(NgsiError::ResourceNotFound(format!("temporal entity {id} not found")).into())
626 : }
627 12 : Some(Err(e)) => Err(ApiError::from(e)),
628 872 : Some(Ok(())) if local.found => Ok(no_content(tenant)),
629 : Some(Ok(())) => {
630 26 : Err(NgsiError::ResourceNotFound(format!("{} not found", local.missing)).into())
631 : }
632 : };
633 120 : }
634 120 : let local_part = match &local.res {
635 : None => {
636 120 : crate::federation::Part::local(404, format!("temporal entity {id} not found locally"))
637 : }
638 0 : Some(_) if local.found => crate::federation::Part::local(204, local.applied),
639 : Some(_) => {
640 0 : crate::federation::Part::local(404, format!("{} not found locally", local.missing))
641 : }
642 : };
643 120 : let mut parts = vec![local_part];
644 120 : let ctx_url = crate::federation::ctx_link_url(headers, &ctx.source);
645 120 : for reg in ®s {
646 120 : if !reg.supports(op) {
647 104 : if reg.is_proxy() {
648 32 : parts.push(crate::federation::conflict_part(op, reg));
649 72 : }
650 104 : continue;
651 16 : }
652 16 : parts.push(
653 16 : crate::federation::forward_part(
654 16 : st,
655 16 : method.clone(),
656 16 : format!(
657 16 : "{}/ngsi-ld/v1/temporal/entities/{}{path_suffix}",
658 16 : reg.endpoint,
659 16 : crate::federation::path_segment(id)
660 16 : ),
661 16 : &[],
662 16 : headers,
663 16 : tenant,
664 16 : reg,
665 16 : &ctx_url,
666 16 : body.clone(),
667 16 : )
668 16 : .await,
669 : );
670 : }
671 120 : Ok(crate::federation::combine(
672 120 : st.profile,
673 120 : Some(id),
674 120 : parts,
675 120 : no_content(tenant),
676 120 : tenant,
677 120 : ))
678 3704 : }
679 :
680 : // ---------- PATCH/DELETE .../attrs/{attrId}/{instanceId} (5.6.14/5.6.15) ----------
681 :
682 74 : pub async fn modify_temporal_instance(
683 74 : State(st): State<AppState>,
684 74 : Path((id, attr, instance_id)): Path<(String, String, String)>,
685 74 : CleanParams(params): CleanParams,
686 74 : headers: HeaderMap,
687 74 : body: Bytes,
688 74 : ) -> Response {
689 74 : let go = async {
690 74 : let tenant = tenant_from(&headers)?;
691 : // Empty attr segment ⇒ the URI names no resource with a PATCH method
692 : // (suite 016_02_06 asserts 405 here, vs 400 on the DELETE sibling).
693 74 : if attr.is_empty() {
694 4 : return Err(ApiError::Bare(StatusCode::METHOD_NOT_ALLOWED));
695 70 : }
696 70 : antares_model::EntityId::new(&id)?;
697 62 : antares_model::check_attr_name(&attr)?;
698 58 : antares_model::EntityId::new(&instance_id)
699 58 : .map_err(|_| NgsiError::BadRequestData("invalid instance id".into()))?;
700 54 : check_params(¶ms, &["local"])?;
701 54 : let parsed = parse_body(&st, &headers, &body, BodyKind::MergePatch).await?;
702 54 : let obj = parsed.object(NgsiError::BadRequestData(
703 54 : "fragment must be a JSON object".into(),
704 54 : ))?;
705 54 : let attr_iri = antares_jsonld::expand_attr_name(&attr, &parsed.ctx)?;
706 54 : let frag_inst = instance_fragment(obj, &attr, &attr_iri, &parsed.ctx)?;
707 : // as 5.6.4 is asked: the Attribute and the instance written
708 52 : let cond = temporal_write_gate(
709 52 : &st,
710 52 : &tenant,
711 52 : &headers,
712 52 : &id,
713 52 : "5.6.14",
714 52 : std::slice::from_ref(&attr_iri),
715 52 : Some(&frag_inst),
716 52 : )
717 52 : .await?;
718 50 : let regs =
719 52 : match temporal_write_regs(&st, &tenant, &headers, &parsed.ctx, ¶ms, &id).await {
720 50 : Ok(regs) => regs,
721 2 : Err(refused) => return Ok(*refused),
722 : };
723 50 : crate::policy::Condition::unforwarded(cond.as_ref(), !regs.is_empty())?;
724 50 : let ts = now_iso();
725 50 : let mut found = false;
726 50 : let res = st
727 50 : .temporal
728 50 : .mutate(&tenant, &id, |doc| {
729 32 : crate::policy::Condition::pre_evolution(cond.as_ref(), &id, doc)?;
730 28 : let target = antares_store::stored_object(doc)?;
731 28 : if let Some(arr) = target.get_mut(&attr_iri).and_then(Value::as_array_mut) {
732 32 : if let Some(inst) = arr.iter_mut().find(|i| {
733 32 : i.get("instanceId").and_then(Value::as_str) == Some(instance_id.as_str())
734 32 : }) {
735 18 : found = true;
736 18 : replace_instance(inst, &frag_inst, &ts)?;
737 6 : }
738 4 : }
739 28 : Ok::<(), NgsiError>(())
740 32 : })
741 50 : .await?;
742 50 : answer_temporal_attr_write(
743 50 : &st,
744 50 : &tenant,
745 50 : &headers,
746 50 : &parsed.ctx,
747 50 : &id,
748 50 : "updateAttrInstanceTemporal",
749 50 : reqwest::Method::PATCH,
750 50 : &format!(
751 50 : "/attrs/{}/{}",
752 50 : crate::federation::path_segment(&attr),
753 50 : crate::federation::path_segment(&instance_id)
754 50 : ),
755 50 : Some(parsed.value.clone()),
756 50 : regs,
757 50 : LocalWrite {
758 50 : res,
759 50 : found,
760 50 : missing: format!("instance {instance_id}"),
761 50 : applied: "applied locally",
762 50 : },
763 50 : )
764 50 : .await
765 74 : };
766 74 : go.await.unwrap_or_else(|e| e.into_response())
767 74 : }
768 :
769 : /// 5.6.14.4 takes "the Attribute instance in the EntityTemporal
770 : /// Fragment": one keyed by the Attribute name. A bare instance (no
771 : /// such key, ETSI 016_01) is taken as that instance itself. The instance
772 : /// comes back expanded.
773 54 : fn instance_fragment(
774 54 : frag: &Map<String, Value>,
775 54 : attr: &str,
776 54 : attr_iri: &str,
777 54 : ctx: &Context,
778 54 : ) -> ApiResult<Value> {
779 54 : let mut frag = frag.clone();
780 54 : frag.remove("@context");
781 54 : let named = (!frag.contains_key("type")).then_some(()).and_then(|()| {
782 18 : frag.iter().find_map(|(k, v)| {
783 18 : let is_attr = k == attr
784 2 : || antares_jsonld::expand_attr_name(k, ctx).ok().as_deref() == Some(attr_iri);
785 18 : is_attr.then(|| v.clone())
786 18 : })
787 18 : });
788 52 : let frag = match named {
789 6 : Some(Value::Array(mut one)) if one.len() == 1 => one.remove(0),
790 10 : Some(Value::Object(o)) => Value::Object(o),
791 : Some(_) => {
792 2 : return Err(NgsiError::BadRequestData(
793 2 : "the EntityTemporal Fragment holds one instance of the Attribute (5.6.14.4)".into(),
794 2 : )
795 2 : .into())
796 : }
797 38 : None => Value::Object(frag),
798 : };
799 52 : let mut wrapper = Map::new();
800 52 : wrapper.insert(attr.to_owned(), frag);
801 52 : let expanded = expand_entity(
802 52 : &wrapper,
803 52 : ctx,
804 52 : ExpandOpts {
805 52 : fragment: true,
806 52 : allow_null: false,
807 52 : temporal: true,
808 52 : ..Default::default()
809 52 : },
810 0 : )?;
811 52 : expanded
812 52 : .get(attr_iri)
813 52 : .and_then(Value::as_array)
814 52 : .and_then(|a| a.first())
815 52 : .cloned()
816 52 : .ok_or_else(|| NgsiError::BadRequestData("invalid instance fragment".into()).into())
817 54 : }
818 :
819 : /// 5.6.14.4: "Replace the target Attribute instance identified by the
820 : /// instanceId with the Attribute instance in the EntityTemporal Fragment.
821 : /// The createdAt property of the concerned instance shall remain
822 : /// unchanged, but the modifiedAt property shall be set to the timestamp
823 : /// corresponding to this modification." A replace, so a member only the
824 : /// stored instance carries does not survive it; the instance keeps the
825 : /// identity it is addressed by.
826 18 : fn replace_instance(inst: &mut Value, frag_inst: &Value, ts: &str) -> Result<(), NgsiError> {
827 18 : let t = antares_store::stored_object(inst)?;
828 18 : let kept: Vec<(String, Value)> = ["createdAt", "instanceId"]
829 18 : .iter()
830 36 : .filter_map(|k| t.get(*k).map(|v| ((*k).to_owned(), v.clone())))
831 18 : .collect();
832 18 : t.clear();
833 18 : t.extend(kept);
834 54 : for (k, v) in antares_jsonld::expanded_object(frag_inst)? {
835 54 : if matches!(k.as_str(), "createdAt" | "instanceId") {
836 0 : continue;
837 54 : }
838 54 : t.insert(k.clone(), v.clone());
839 : }
840 18 : t.insert("modifiedAt".into(), Value::String(ts.to_owned()));
841 18 : Ok(())
842 18 : }
843 :
844 88 : pub async fn delete_temporal_instance(
845 88 : State(st): State<AppState>,
846 88 : Path((id, attr, instance_id)): Path<(String, String, String)>,
847 88 : CleanParams(params): CleanParams,
848 88 : headers: HeaderMap,
849 88 : ) -> Response {
850 88 : let go = async {
851 88 : let tenant = tenant_from(&headers)?;
852 88 : antares_model::EntityId::new(&id)?;
853 80 : antares_model::check_attr_name(&attr)?;
854 56 : antares_model::EntityId::new(&instance_id)
855 56 : .map_err(|_| NgsiError::BadRequestData("invalid instance id".into()))?;
856 52 : check_params(¶ms, &["local"])?;
857 52 : let ctx = request_context(&st.loader, &headers).await?;
858 52 : let attr_iri = antares_jsonld::expand_attr_name(&attr, &ctx)?;
859 52 : let cond = temporal_write_gate(
860 52 : &st,
861 52 : &tenant,
862 52 : &headers,
863 52 : &id,
864 52 : "5.6.15",
865 52 : std::slice::from_ref(&attr_iri),
866 52 : None,
867 52 : )
868 52 : .await?;
869 52 : let regs = match temporal_write_regs(&st, &tenant, &headers, &ctx, ¶ms, &id).await {
870 50 : Ok(regs) => regs,
871 2 : Err(refused) => return Ok(*refused),
872 : };
873 50 : crate::policy::Condition::unforwarded(cond.as_ref(), !regs.is_empty())?;
874 50 : let mut found = false;
875 50 : let ts = now_iso();
876 50 : let res = st
877 50 : .temporal
878 50 : .mutate(&tenant, &id, |doc| {
879 28 : crate::policy::Condition::pre_evolution(cond.as_ref(), &id, doc)?;
880 24 : let target = antares_store::stored_object(doc)?;
881 34 : found = remove_instances(target, &attr_iri, |i| {
882 34 : i.get("instanceId").and_then(Value::as_str) == Some(instance_id.as_str())
883 34 : });
884 24 : if found {
885 14 : target.insert("modifiedAt".into(), Value::String(ts.clone()));
886 16 : }
887 24 : Ok::<(), NgsiError>(())
888 28 : })
889 50 : .await?;
890 50 : answer_temporal_attr_write(
891 50 : &st,
892 50 : &tenant,
893 50 : &headers,
894 50 : &ctx,
895 50 : &id,
896 50 : "deleteAttrInstanceTemporal",
897 50 : reqwest::Method::DELETE,
898 50 : &format!(
899 50 : "/attrs/{}/{}",
900 50 : crate::federation::path_segment(&attr),
901 50 : crate::federation::path_segment(&instance_id)
902 50 : ),
903 50 : None,
904 50 : regs,
905 50 : LocalWrite {
906 50 : res,
907 50 : found,
908 50 : missing: format!("instance {instance_id}"),
909 50 : applied: "applied locally",
910 50 : },
911 50 : )
912 50 : .await
913 88 : };
914 88 : go.await.unwrap_or_else(|e| e.into_response())
915 88 : }
|
