Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Retrieve temporal evolution of one Entity (5.7.3).
3 :
4 : use super::*;
5 :
6 : // ---------- GET /temporal/entities/{id} (5.7.3) ----------
7 :
8 1040 : pub async fn retrieve_temporal(
9 1040 : State(st): State<AppState>,
10 1040 : Path(id): Path<String>,
11 1040 : CleanParams(params): CleanParams,
12 1040 : headers: HeaderMap,
13 1040 : ) -> Response {
14 1040 : match retrieve_temporal_outer(&st, &id, ¶ms, &headers).await {
15 890 : Ok(r) => r,
16 150 : Err(e) => e.into_response(),
17 : }
18 1040 : }
19 :
20 : /// 5.7.3.4 Retrieve Temporal Evolution: the EntityMap half of the clause is
21 : /// the shared rule (`entity_maps::retrieve_with_map`); this is the retrieve
22 : /// it wraps.
23 1040 : pub(super) async fn retrieve_temporal_outer(
24 1040 : st: &AppState,
25 1040 : id: &str,
26 1040 : params: &HashMap<String, String>,
27 1040 : headers: &HeaderMap,
28 1040 : ) -> ApiResult<Response> {
29 1040 : crate::entity_map::retrieve_with_map(st, id, params, headers, true, |map| async move {
30 1040 : retrieve_temporal_inner(st, id, params, headers, map.as_ref()).await
31 2080 : })
32 1040 : .await
33 1040 : }
34 :
35 : /// Vanilla: `check_params` already refused `versions` as unknown.
36 1040 : pub(super) fn versions_asked(
37 1040 : _st: &AppState,
38 1040 : _params: &HashMap<String, String>,
39 1040 : _headers: &HeaderMap,
40 1040 : ) -> ApiResult<bool> {
41 1040 : Ok(false)
42 1040 : }
43 :
44 1040 : pub(super) async fn retrieve_temporal_inner(
45 1040 : st: &AppState,
46 1040 : id: &str,
47 1040 : params: &HashMap<String, String>,
48 1040 : headers: &HeaderMap,
49 1040 : map: Option<&Value>,
50 1040 : ) -> ApiResult<Response> {
51 1040 : let tenant = tenant_from(headers)?;
52 1040 : check_params(params, RETRIEVE_PARAMS)?;
53 1040 : let versions = versions_asked(st, params, headers)?;
54 1040 : let accept = parse_accept(headers)?;
55 1040 : let ctx = request_context(&st.loader, headers).await?;
56 1040 : let filter = gate!(st, &tenant, headers, "5.7.3", ids: &[id]).await?;
57 : // Every reader below takes the window clamped to the grant
58 1040 : let windowed = filter.narrow_window(params, false)?;
59 1040 : let params = &windowed;
60 1040 : let tq = TemporalQ::from_params(params, false)?;
61 1028 : let mut trepr = parse_trepr(params, &ctx)?;
62 1000 : crate::repr::narrow_projection(&mut trepr.pick, &mut trepr.omit, &filter, &ctx)?;
63 : // 5.7.3.4: "If projection attributes are present and indicate the
64 : // use of Linked Entity retrieval, an error of type BadRequestData
65 : // shall be raised" — unconditional, temporal defines no join.
66 1000 : reject_linked_projection(&trepr, "5.7.3.4")?;
67 988 : antares_model::EntityId::new(id)?;
68 984 : let types_rule = crate::repr::type_rules(&filter);
69 984 : let timeprop = tq
70 984 : .as_ref()
71 984 : .map_or("observedAt", |t| t.timeproperty.as_str())
72 984 : .to_owned();
73 984 : let req = RetrieveReq {
74 984 : st,
75 984 : tenant: &tenant,
76 984 : headers,
77 984 : ctx: &ctx,
78 984 : id,
79 984 : params,
80 984 : map,
81 984 : };
82 984 : let tf = retrieve_filter(
83 984 : tq.as_ref(),
84 984 : trepr.page,
85 984 : default_page(st, &tenant, &trepr),
86 984 : types_rule.as_deref(),
87 984 : &timeprop,
88 : );
89 984 : let mut warnings: Vec<String> = Vec::new();
90 984 : let doc = load_evolution(&req, &tf, &timeprop, &mut warnings).await?;
91 922 : let not_found = || NgsiError::ResourceNotFound(format!("temporal entity {id} not found"));
92 : // ADR-0020: an Entity outside the
93 : // narrowing now is the 404 of an absent one
94 922 : if !member_now(st, &tenant, &ctx, &filter, id, Some(&doc)).await? {
95 4 : return Err(not_found().into());
96 918 : }
97 : // the history its rules show, before anything
98 : // reads it; no rule for its types is the 404 of an absent Entity
99 918 : let doc = match types_rule.as_deref() {
100 28 : Some(r) => rules_per_instance(&doc, r).ok_or_else(not_found)?,
101 890 : None => doc,
102 : };
103 916 : let attrs_filter = selection(&trepr);
104 916 : check_selection(&doc, &trepr, attrs_filter.as_ref(), id)?;
105 906 : let mut w = window(
106 906 : &doc,
107 906 : tq.as_ref(),
108 906 : trepr.page,
109 906 : attrs_filter.as_ref(),
110 906 : trepr.omit.as_ref(),
111 906 : trepr.dataset_id.as_ref(),
112 906 : &timeprop,
113 906 : crate::history::held_attrs(st, &tenant, &doc),
114 : );
115 906 : if !trepr.aggregated {
116 852 : let page_n = trepr.page;
117 852 : truncate(
118 852 : &mut w,
119 852 : &timeprop,
120 852 : page_n.desc,
121 852 : page_n.size(st.temporal_default_instances, st.temporal_instance_limit),
122 : );
123 : // after the cut: lastN counts measurements, never versions
124 852 : if versions {
125 0 : with_versions(&mut w, &st.temporal.versions(&tenant, id).await?);
126 852 : }
127 54 : }
128 906 : let payload = present_temporal(&doc, &w, &ctx, &trepr, tq.as_ref(), &timeprop)?;
129 894 : if (trepr.pick.is_some() || trepr.omit.is_some())
130 34 : && payload.as_object().is_some_and(|o| o.is_empty())
131 : {
132 4 : return Err(NgsiError::ResourceNotFound(format!(
133 4 : "projection matches nothing on temporal entity {id}"
134 4 : ))
135 4 : .into());
136 890 : }
137 890 : let mut resp = respond_retrieved(&req, accept, &w, &trepr, tq.as_ref(), payload);
138 890 : crate::paging::attach_warnings(&mut resp, &warnings);
139 890 : filter.mark(&st.host_alias, resp.headers_mut());
140 890 : Ok(resp)
141 1040 : }
142 :
143 : /// The parameters 5.7.3 accepts (6.19.3.1 Table 6.19.3.1-1).
144 : const RETRIEVE_PARAMS: &[&str] = &[
145 : "attrs",
146 : "timerel",
147 : "timeAt",
148 : "endTimeAt",
149 : "timeproperty",
150 : "lastN",
151 : "firstN",
152 : "offsetN",
153 : "aggrMethods",
154 : "aggrPeriodDuration",
155 : "options",
156 : "format",
157 : "lang",
158 : "local",
159 : "pick",
160 : "omit",
161 : "datasetId",
162 : "entityMap",
163 : "entityMapLifetime",
164 : ];
165 :
166 : /// One 5.7.3 request as the stages after the gate read it; `params` is
167 : /// the window clamped to the grant.
168 : struct RetrieveReq<'a> {
169 : st: &'a AppState,
170 : tenant: &'a TenantId,
171 : headers: &'a HeaderMap,
172 : ctx: &'a Context,
173 : id: &'a str,
174 : params: &'a HashMap<String, String>,
175 : map: Option<&'a Value>,
176 : }
177 :
178 : /// Instance pruning pushed into the store (no q=/geo on retrieve, so it
179 : /// is always safe here); window() stays the arbiter. Under the narrowing rules, a
180 : /// condition rebuilds each instance's state from the whole history, so
181 : /// the store prunes no instance under one; rules without one read only
182 : /// the window.
183 984 : fn retrieve_filter<'a>(
184 984 : tq: Option<&'a TemporalQ>,
185 984 : page_n: InstancePage,
186 984 : default: Option<usize>,
187 984 : types_rule: Option<&crate::policy::Filter>,
188 984 : timeprop: &'a str,
189 984 : ) -> antares_store::filter::TemporalFilter<'a> {
190 984 : let replays = replays_state(types_rule);
191 : antares_store::filter::TemporalFilter {
192 984 : range: tq
193 984 : .filter(|_| !replays)
194 984 : .map(|t| antares_store::filter::InstanceRange {
195 422 : timerel: &t.timerel,
196 422 : time_at: &t.time_at,
197 422 : end_time_at: t.end_time_at.as_deref(),
198 422 : timeproperty: &t.timeproperty,
199 422 : }),
200 984 : last_n: page_n
201 984 : .store_cap(default)
202 984 : .filter(|_| cap_keeps_shown(types_rule, &page_n, timeprop, range_start(tq))),
203 984 : oldest_first: !page_n.desc,
204 984 : timeproperty: timeprop,
205 984 : ..Default::default()
206 : }
207 984 : }
208 :
209 : /// The Entity's Temporal Evolution: the local one, and under 5.7.3.4 the
210 : /// matching retrieveTemporal registrations' merged into it (4.5.5;
211 : /// auxiliary instances only fill timestamps absent elsewhere).
212 984 : async fn load_evolution(
213 984 : req: &RetrieveReq<'_>,
214 984 : tf: &antares_store::filter::TemporalFilter<'_>,
215 984 : timeprop: &str,
216 984 : warnings: &mut Vec<String>,
217 984 : ) -> ApiResult<Value> {
218 984 : let (st, tenant, id) = (req.st, req.tenant, req.id);
219 984 : let local = match st.temporal.get_temporal(tenant, id, tf).await? {
220 918 : Some(doc) => Some(doc),
221 62 : None => kept_no_history(st, tenant, id).await?,
222 : };
223 980 : let looped = crate::federation::via_loop(
224 980 : req.headers,
225 980 : &crate::federation::alias_for(&st.host_alias, tenant),
226 : );
227 980 : if !crate::federation::active(req.params) || looped {
228 0 : return Ok(local.ok_or_else(|| {
229 0 : NgsiError::ResourceNotFound(format!("temporal entity {id} not found"))
230 0 : })?);
231 980 : }
232 980 : let mut proxies = Vec::new();
233 980 : let fed = crate::federation::fed_retrieve_temporal(
234 980 : st,
235 980 : tenant,
236 980 : req.headers,
237 980 : req.ctx,
238 980 : id,
239 980 : req.params,
240 980 : req.map,
241 980 : warnings,
242 980 : &mut proxies,
243 980 : )
244 980 : .await?;
245 980 : let mut local = local;
246 980 : if let Some(b) = local.as_mut() {
247 920 : crate::federation::drop_proxied_local(b, &proxies, req.ctx);
248 920 : }
249 980 : Ok(merge_retrieved(local, &fed, id, timeprop)?)
250 984 : }
251 :
252 : /// The local Evolution, or the first exclusive remote one when there is
253 : /// none (any remote one when all are auxiliary), with every other remote
254 : /// copy merged in, the exclusive ones first.
255 980 : fn merge_retrieved(
256 980 : local: Option<Value>,
257 980 : fed: &[(bool, Value)],
258 980 : id: &str,
259 980 : timeprop: &str,
260 980 : ) -> Result<Value, NgsiError> {
261 980 : let (mut base, skip) = match local {
262 920 : Some(b) => (b, None),
263 : None => {
264 60 : let idx = fed.iter().position(|(aux, _)| !aux).or(if fed.is_empty() {
265 58 : None
266 : } else {
267 2 : Some(0)
268 : });
269 60 : match idx.and_then(|i| fed.get(i).map(|(_, d)| (d.clone(), i))) {
270 2 : Some((d, i)) => (d, Some(i)),
271 : None => {
272 58 : return Err(NgsiError::ResourceNotFound(format!(
273 58 : "temporal entity {id} not found"
274 58 : )))
275 : }
276 : }
277 : }
278 : };
279 1844 : for aux_pass in [false, true] {
280 1844 : for (i, (aux, d)) in fed.iter().enumerate() {
281 8 : if Some(i) == skip {
282 4 : continue;
283 4 : }
284 4 : if *aux == aux_pass {
285 2 : merge_temporal_docs(&mut base, d, *aux, timeprop);
286 2 : }
287 : }
288 : }
289 922 : Ok(base)
290 980 : }
291 :
292 : /// 5.7.3: attrs matching nothing ⇒ 404. A `pick` may name core members
293 : /// instead — 5.7.3.3 admits "id", "type", "scope" or an Attribute name
294 : /// — and then it selects no Attribute at all. 5.7.3.5 still reduces the
295 : /// Entity to the members it names, so that empty selection is the
296 : /// answer rather than a Temporal Evolution holding none of it.
297 916 : fn check_selection(
298 916 : doc: &Value,
299 916 : trepr: &TRepr,
300 916 : attrs_filter: Option<&Vec<String>>,
301 916 : id: &str,
302 916 : ) -> Result<(), NgsiError> {
303 916 : let core_only_pick =
304 916 : trepr.attrs.is_none() && trepr.pick.is_some() && attrs_filter.is_some_and(Vec::is_empty);
305 150 : match attrs_filter {
306 150 : Some(want) if !core_only_pick && !want.iter().any(|a| doc.get(a).is_some()) => {
307 10 : Err(NgsiError::ResourceNotFound(format!(
308 10 : "temporal entity {id} has none of the requested attributes"
309 10 : )))
310 : }
311 906 : _ => Ok(()),
312 : }
313 916 : }
314 :
315 : /// The Evolution's body, 200 or 206 with its Content-Range, and the
316 : /// 6.4.7.3 links to the next page.
317 890 : fn respond_retrieved(
318 890 : req: &RetrieveReq<'_>,
319 890 : accept: Accept,
320 890 : w: &Windowed,
321 890 : trepr: &TRepr,
322 890 : tq: Option<&TemporalQ>,
323 890 : payload: Value,
324 890 : ) -> Response {
325 890 : let (st, page_n) = (req.st, trepr.page);
326 : // aggregated responses are complete by construction — never 206 (6.3.10)
327 890 : let cr = if trepr.aggregated {
328 42 : None
329 : } else {
330 848 : content_range(
331 848 : w.truncated,
332 848 : w.ts_min.as_deref(),
333 848 : w.ts_max.as_deref(),
334 848 : tq,
335 848 : page_n,
336 : )
337 : };
338 890 : let status = if cr.is_some() {
339 120 : StatusCode::PARTIAL_CONTENT
340 : } else {
341 770 : StatusCode::OK
342 : };
343 890 : let mut resp = respond(status, payload, req.ctx, accept, req.tenant);
344 890 : attach_paging(
345 890 : &mut resp,
346 890 : None,
347 890 : &interval_links(
348 890 : req.params,
349 890 : &format!(
350 890 : "/ngsi-ld/v1/temporal/entities/{}",
351 890 : crate::federation::path_segment(req.id)
352 890 : ),
353 890 : accept,
354 890 : page_n,
355 890 : w.more,
356 890 : page_n.step(
357 890 : trepr.aggregated,
358 890 : st.temporal_default_instances,
359 890 : st.temporal_instance_limit,
360 890 : ),
361 890 : w.cut_served.zip(cut_boundary(w, page_n.desc)),
362 890 : ),
363 : );
364 890 : if let Some(cr) = cr {
365 120 : if let Ok(v) = cr.parse() {
366 120 : resp.headers_mut().insert("Content-Range", v);
367 120 : }
368 770 : }
369 890 : resp
370 890 : }
371 :
372 : /// 5.7.3.4 / 4.5.5: merge one remote Temporal Evolution into `base` by
373 : /// appending instances per Attribute; auxiliary data contributes an
374 : /// instance only when no instance with the same timeproperty value was
375 : /// received from elsewhere.
376 24 : pub(super) fn merge_temporal_docs(base: &mut Value, add: &Value, aux: bool, timeprop: &str) {
377 24 : let (Some(target), Some(source)) = (base.as_object_mut(), add.as_object()) else {
378 0 : return;
379 : };
380 72 : for (k, v) in source {
381 72 : if [
382 72 : "id",
383 72 : "type",
384 72 : "scope",
385 72 : "createdAt",
386 72 : "modifiedAt",
387 72 : "deletedAt",
388 72 : "expiresAt",
389 72 : ]
390 72 : .contains(&k.as_str())
391 : {
392 48 : target.entry(k.clone()).or_insert_with(|| v.clone());
393 48 : continue;
394 24 : }
395 24 : let incoming: Vec<Value> = match v {
396 24 : Value::Array(x) => x.clone(),
397 0 : other => vec![other.clone()],
398 : };
399 24 : match target.get_mut(k).and_then(Value::as_array_mut) {
400 4 : None => {
401 4 : target.insert(k.clone(), Value::Array(incoming));
402 4 : }
403 20 : Some(cur) => {
404 32 : for ni in incoming {
405 32 : if aux {
406 12 : let ts = ni.get(timeprop).and_then(Value::as_str).map(dt_key);
407 12 : if ts.is_some()
408 12 : && cur.iter().any(|ci| {
409 12 : ci.get(timeprop).and_then(Value::as_str).map(dt_key) == ts
410 12 : })
411 : {
412 8 : continue;
413 4 : }
414 20 : }
415 : // 4.5.5.3: an instance with the same datasetId (or both
416 : // default) AND the same timeproperty value is a
417 : // CONFLICTING instance of one slot — resolve to one, the
418 : // most recent modifiedAt winning.
419 : // Two instances only share a slot when they carry the SAME
420 : // timeproperty value. Without the is_some() guard a pair of
421 : // instances that both lack it would compare None == None,
422 : // letting a remote instance replace unrelated local history.
423 : // 4.6.3 leaves the seconds fraction optional, so two
424 : // Context Sources may spell one instant differently: the
425 : // slot and its winner are decided on the canonical key,
426 : // or the same instance comes back twice.
427 24 : let ni_ts = ni.get(timeprop).and_then(Value::as_str).map(dt_key);
428 24 : let slot = ni_ts.and_then(|ts| {
429 28 : cur.iter_mut().find(|ci| {
430 28 : ci.get(timeprop).and_then(Value::as_str).map(dt_key) == Some(ts.clone())
431 16 : && ci.get("datasetId") == ni.get("datasetId")
432 28 : })
433 24 : });
434 24 : if let Some(existing) = slot {
435 24 : let stamp = |i: &Value| {
436 24 : dt_key(i.get("modifiedAt").and_then(Value::as_str).unwrap_or(""))
437 24 : };
438 12 : let newer = stamp(&ni) > stamp(existing);
439 12 : if newer {
440 12 : *existing = ni;
441 12 : }
442 12 : continue;
443 12 : }
444 12 : cur.push(ni);
445 : }
446 : }
447 : }
448 : }
449 24 : }
450 :
451 : /// The Entity `id` whose retention keeps no history (a type no
452 : /// rule opts in, or `off`), as a Temporal Evolution of no Attribute: it
453 : /// exists and recorded nothing, which is no 404. The member checks after
454 : /// the read hold for it as for any other.
455 62 : pub(super) async fn kept_no_history(
456 62 : st: &AppState,
457 62 : tenant: &TenantId,
458 62 : id: &str,
459 62 : ) -> Result<Option<Value>, ApiError> {
460 62 : let Some(doc) = st
461 62 : .store
462 62 : .get(tenant, antares_store::Kind::Entity, id)
463 62 : .await?
464 : else {
465 46 : return Ok(None);
466 : };
467 16 : if crate::history::records_history(st, tenant, &doc) {
468 14 : return Ok(None);
469 2 : }
470 2 : let mut shell = serde_json::Map::new();
471 6 : for k in ["id", "type", "scope"] {
472 6 : if let Some(v) = doc.get(k) {
473 4 : shell.insert(k.to_owned(), v.clone());
474 4 : }
475 : }
476 2 : Ok(Some(Value::Object(shell)))
477 62 : }
|
