Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! /temporal/entities (5.6.11–5.6.16, 5.7.3/5.7.4; resources 6.18–6.22).
3 :
4 : use crate::negotiate::*;
5 : use crate::state::{now_iso, AppState};
6 : use crate::temporalq::TemporalQ;
7 : use antares_jsonld::compact::compact_instance;
8 : use antares_jsonld::{expand_entity, Context, ExpandOpts};
9 : use antares_model::{dt_key, NgsiError, TenantId};
10 : use antares_ql::parse_q;
11 : use antares_store::TemporalDriverExt as _;
12 : use axum::body::Bytes;
13 : use axum::extract::{Path, State};
14 : use axum::http::{HeaderMap, StatusCode};
15 : use axum::response::{IntoResponse, Response};
16 : use serde_json::{json, Map, Value};
17 : use std::collections::HashMap;
18 :
19 : use crate::negotiate::CleanParams;
20 :
21 : use antares_model::is_meta;
22 :
23 : /// 5.6.11 input: the pushed Temporal Evolution may carry the 4.5.7
24 : /// deleted-instance representation (value = NGSI-LD Null), which 5.5.4
25 : /// explicitly excepts for "the temporal evolution" — hence allow_null.
26 : const TEMPORAL_OPTS: ExpandOpts = ExpandOpts {
27 : fragment: false,
28 : allow_null: true,
29 : merge: false,
30 : temporal: true,
31 : sys: false,
32 : };
33 :
34 : /// 4.5.7 and 4.5.8: each recorded instance of a Property or Relationship
35 : /// carries an instanceId and the 4.8 timestamps. The id is minted here and
36 : /// only when the pushed instance brought none, because the temporal API is
37 : /// add-only (5.6.11.4, 5.6.12.1) and 5.6.14/5.6.15 address an instance by
38 : /// that id. Sub-Attributes are part of the instance and are not stamped
39 : /// separately -- the current-state path stamps them one level deeper
40 : /// (`crate::stamp::stamp_instances`).
41 316 : fn stamp_temporal_instances(doc: &mut Value, ts: &str) {
42 316 : if let Some(obj) = doc.as_object_mut() {
43 1072 : for (k, v) in obj.iter_mut() {
44 1072 : if is_meta(k) {
45 664 : continue;
46 408 : }
47 408 : if let Some(arr) = v.as_array_mut() {
48 586 : for inst in arr {
49 586 : if let Some(o) = inst.as_object_mut() {
50 586 : o.entry("instanceId".to_owned()).or_insert_with(|| {
51 586 : Value::String(format!("urn:ngsi-ld:Instance:{}", uuid::Uuid::new_v4()))
52 586 : });
53 586 : o.insert("createdAt".into(), Value::String(ts.to_owned()));
54 586 : o.insert("modifiedAt".into(), Value::String(ts.to_owned()));
55 0 : }
56 : }
57 0 : }
58 : }
59 0 : }
60 316 : }
61 :
62 : // ---------- POST /temporal/entities/ — Upsert temporal (5.6.11) ----------
63 :
64 816 : pub async fn upsert_temporal(
65 816 : State(st): State<AppState>,
66 816 : CleanParams(params): CleanParams,
67 816 : headers: HeaderMap,
68 816 : body: Bytes,
69 816 : ) -> Response {
70 816 : let go = async {
71 816 : let tenant = tenant_from(&headers)?;
72 816 : check_params(¶ms, &["options", "local"])?;
73 812 : let parsed = parse_body(&st.loader, &headers, &body, BodyKind::Standard).await?;
74 324 : let obj = parsed.object(NgsiError::BadRequestData(
75 324 : "temporal entity must be a JSON object".into(),
76 324 : ))?;
77 324 : let expanded = expand_entity(obj, &parsed.ctx, TEMPORAL_OPTS)?;
78 312 : let id = antares_jsonld::expanded_id(&expanded)?.to_owned();
79 312 : gate!(st, &tenant, &headers, "5.6.11", ids: &[&id]).await?;
80 : // 5.6.11.4: exclusive/redirect registrations matching the input are
81 : // forwarded when "Create or Update Temporal" is supported; proxy
82 : // modes without it are an error of type Conflict; inclusive ones
83 : // forward when supported. Matching attributes are removed from the
84 : // local fragment.
85 312 : let spec = crate::registry::CsrSpec {
86 312 : ids: Some(vec![id.clone()]),
87 312 : ..Default::default()
88 312 : };
89 312 : let regs = match crate::federation::write_plan(
90 312 : &st,
91 312 : &tenant,
92 312 : &spec,
93 312 : &parsed.ctx,
94 312 : ¶ms,
95 312 : &headers,
96 312 : )
97 312 : .await?
98 : {
99 0 : crate::federation::WritePlan::Answered(r) => return Ok(*r),
100 312 : crate::federation::WritePlan::Forward(regs) => regs,
101 : };
102 312 : if !regs.is_empty() {
103 4 : let mut parts = Vec::new();
104 4 : let mut fwd = Vec::new();
105 4 : for reg in ®s {
106 4 : if !reg.supports("upsertTemporal") {
107 2 : if reg.is_proxy() {
108 2 : parts.push(crate::federation::conflict_part("upsertTemporal"));
109 2 : }
110 2 : continue;
111 2 : }
112 2 : if let Some(frag) = crate::federation::reduce_to_scope(obj, reg, &parsed.ctx) {
113 2 : fwd.push((reg.clone(), frag));
114 2 : }
115 : }
116 4 : let proxies: Vec<&crate::federation::FedReg> =
117 4 : regs.iter().filter(|r| r.is_proxy()).collect();
118 4 : let (rest, has_attrs) = crate::federation::strip_proxied(obj, &proxies, &parsed.ctx);
119 4 : if has_attrs || proxies.is_empty() {
120 0 : let local = expand_entity(&rest, &parsed.ctx, TEMPORAL_OPTS)?;
121 0 : let status = upsert_temporal_local(&st, &tenant, &id, local).await?;
122 0 : parts.push(crate::federation::Part {
123 0 : status: status.as_u16(),
124 0 : detail: "local temporal upsert".into(),
125 0 : });
126 4 : }
127 4 : let ctx_url = crate::federation::ctx_link_url(&headers, &parsed.ctx.source);
128 4 : for (reg, frag) in fwd {
129 2 : parts.push(
130 2 : crate::federation::forward_part(
131 2 : &st,
132 2 : reqwest::Method::POST,
133 2 : format!("{}/ngsi-ld/v1/temporal/entities", reg.endpoint),
134 2 : &[],
135 2 : &headers,
136 2 : &tenant,
137 2 : ®,
138 2 : &ctx_url,
139 2 : Some(frag),
140 2 : )
141 2 : .await,
142 : );
143 : }
144 4 : return Ok(crate::federation::combine(
145 4 : parts,
146 4 : created(
147 4 : format!(
148 4 : "/ngsi-ld/v1/temporal/entities/{}",
149 4 : crate::federation::path_segment(&id)
150 4 : ),
151 4 : &tenant,
152 4 : ),
153 4 : &tenant,
154 4 : ));
155 308 : }
156 308 : let status = upsert_temporal_local(&st, &tenant, &id, expanded).await?;
157 308 : Ok::<_, ApiError>(if status == StatusCode::CREATED {
158 270 : created(
159 270 : format!(
160 : "/ngsi-ld/v1/temporal/entities/{}",
161 270 : crate::federation::path_segment(&id)
162 : ),
163 270 : &tenant,
164 : )
165 : } else {
166 38 : no_content(&tenant)
167 : })
168 816 : };
169 816 : go.await.unwrap_or_else(|e| e.into_response())
170 816 : }
171 :
172 : /// 5.6.11.4 local half: create the Temporal Evolution, or add the provided
173 : /// instances to the existing one per 5.6.12 (merge key = datasetId +
174 : /// observedAt) with Entity Type names unioned. Returns 201 vs 204.
175 308 : async fn upsert_temporal_local(
176 308 : st: &AppState,
177 308 : tenant: &antares_model::TenantId,
178 308 : id: &str,
179 308 : mut expanded: Value,
180 308 : ) -> ApiResult<StatusCode> {
181 308 : let ts = now_iso();
182 308 : stamp_temporal_instances(&mut expanded, &ts);
183 : // get->create/mutate is a TOCTOU pair: two concurrent first-upserts
184 : // both see "absent", and the loser's create must NOT be silently
185 : // dropped (201 with a discarded payload). Loop: a lost create retries
186 : // as a merge, a mutate on a just-deleted doc retries as a create.
187 308 : let mut attempts = 0;
188 : loop {
189 308 : attempts += 1;
190 308 : if attempts > 16 {
191 0 : return Err(NgsiError::InternalError("upsert retry storm".into()).into());
192 308 : }
193 308 : let existed = st.temporal.get(tenant, id).await?.is_some();
194 308 : if existed {
195 38 : let res = st
196 38 : .temporal
197 38 : .mutate(tenant, id, |doc| {
198 38 : let target = antares_store::stored_object(doc)?;
199 : // 5.6.11.4: new Entity Type names are added to the target
200 38 : if let Some(new_types) = expanded.get("type").and_then(Value::as_array) {
201 38 : let mut cur: Vec<Value> = target
202 38 : .get("type")
203 38 : .and_then(Value::as_array)
204 38 : .cloned()
205 38 : .unwrap_or_default();
206 46 : for t in new_types {
207 46 : if !cur.contains(t) {
208 8 : cur.push(t.clone());
209 38 : }
210 : }
211 38 : target.insert("type".into(), Value::Array(cur));
212 0 : }
213 118 : for (k, v) in antares_jsonld::expanded_object(&expanded)? {
214 118 : if is_meta(k) {
215 76 : continue;
216 42 : }
217 42 : let incoming = v.as_array().cloned().unwrap_or_default();
218 42 : match target.get_mut(k).and_then(Value::as_array_mut) {
219 42 : Some(cur) => {
220 : // 5.6.11: instances merge by (datasetId, observedAt)
221 42 : for ni in incoming {
222 42 : let key = (
223 42 : ni.get("datasetId")
224 42 : .and_then(Value::as_str)
225 42 : .map(String::from),
226 42 : ni.get("observedAt")
227 42 : .and_then(Value::as_str)
228 42 : .map(String::from),
229 42 : );
230 42 : let pos = cur.iter().position(|ci| {
231 42 : (
232 42 : ci.get("datasetId")
233 42 : .and_then(Value::as_str)
234 42 : .map(String::from),
235 42 : ci.get("observedAt")
236 42 : .and_then(Value::as_str)
237 42 : .map(String::from),
238 42 : ) == key
239 2 : && key.1.is_some()
240 42 : });
241 42 : match pos {
242 : // A correction, not a new instance: it
243 : // keeps the instanceId its client was
244 : // handed and the createdAt it was created
245 : // at, which is 5.6.14.4's rule for the
246 : // same kind of in-place change ("The
247 : // createdAt property of the concerned
248 : // instance shall remain unchanged").
249 : // `stamp_temporal_instances` has already put a
250 : // fresh pair on the incoming instance.
251 2 : Some(p) => {
252 2 : let mut ni = ni;
253 4 : for keep in ["instanceId", "createdAt"] {
254 4 : let Some(had) = cur[p].get(keep).cloned() else {
255 0 : continue;
256 : };
257 4 : if let Some(o) = ni.as_object_mut() {
258 4 : o.insert(keep.to_owned(), had);
259 4 : }
260 : }
261 2 : cur[p] = ni;
262 : }
263 40 : None => cur.push(ni),
264 : }
265 : }
266 : }
267 0 : None => {
268 0 : target.insert(k.clone(), Value::Array(incoming));
269 0 : }
270 : }
271 : }
272 38 : target.insert("modifiedAt".into(), Value::String(ts.clone()));
273 38 : Ok::<(), NgsiError>(())
274 38 : })
275 38 : .await?;
276 38 : match res {
277 0 : Some(Err(e)) => return Err(ApiError::from(e)),
278 38 : Some(Ok(())) => return Ok(StatusCode::NO_CONTENT),
279 0 : None => continue, // deleted between get and mutate - retry as create
280 : }
281 : } else {
282 270 : let mut doc = expanded.clone();
283 270 : if let Some(o) = doc.as_object_mut() {
284 270 : o.insert("createdAt".into(), Value::String(ts.clone()));
285 270 : o.insert("modifiedAt".into(), Value::String(ts.clone()));
286 270 : }
287 270 : if st.temporal.create(tenant, id, doc).await? {
288 270 : return Ok(StatusCode::CREATED);
289 0 : }
290 : // lost the create race - the doc exists now; retry as a merge
291 : }
292 : }
293 308 : }
294 :
295 : // ---------- temporal query params (4.11) ----------
296 :
297 : /// Windowed per-entity temporal data: filtered+ordered instances per attr.
298 : struct Windowed {
299 : attrs: std::collections::BTreeMap<String, Vec<Value>>,
300 : max_per_attr: usize,
301 : ts_min: Option<String>,
302 : ts_max: Option<String>,
303 : truncated: bool,
304 : }
305 :
306 : /// NGSI-LD 6.3.10: the most instances of one Attribute the broker serves in
307 : /// one response — beyond it the representation is cut and answered "206" with
308 : /// a Content-Range. The ETSI suite triggers 206 at 20 instances and expects
309 : /// 200 at <=5, so any limit in (5,20) is spec-valid; 9 keeps margin.
310 : const TEMPORAL_INSTANCE_LIMIT: usize = 9;
311 :
312 : /// 6.3.10: the ceiling is a CUT, not a label. An Attribute holding more
313 : /// instances than the broker serves at once is truncated to the ceiling in
314 : /// the query direction, and the 206 + Content-Range then describes the
315 : /// instances actually returned. This is what caps a lastN above the ceiling
316 : /// and what caps a request naming no temporal window at all. Aggregated
317 : /// representations (5.7.4.4) are computed over the whole evolution and are
318 : /// complete by construction, so they are never cut.
319 : ///
320 : /// The cut is ONE time boundary for the whole entity, not a per-attribute
321 : /// count: the partial content "shall" be the representation the
322 : /// Content-Range describes, so every attribute is trimmed to the tightest
323 : /// ceiling instant among the over-full ones (ties at that instant kept),
324 : /// and a client continuing from the advertised range-end misses no instance
325 : /// of any attribute. An attribute lying entirely beyond the boundary comes
326 : /// back empty on this page.
327 678 : fn truncate(w: &mut Windowed, timeprop: &str, descending: bool) {
328 678 : if w.max_per_attr <= TEMPORAL_INSTANCE_LIMIT {
329 654 : return;
330 24 : }
331 24 : w.truncated = true;
332 24 : w.max_per_attr = TEMPORAL_INSTANCE_LIMIT;
333 660 : let key = |inst: &Value| inst.get(timeprop).and_then(Value::as_str).map(dt_key);
334 : // the tightest ceiling instant: earliest forwards, latest backwards
335 24 : let boundary = w
336 24 : .attrs
337 24 : .values()
338 32 : .filter(|insts| insts.len() > TEMPORAL_INSTANCE_LIMIT)
339 32 : .filter_map(|insts| key(&insts[TEMPORAL_INSTANCE_LIMIT - 1]))
340 24 : .reduce(|a, b| if (b < a) != descending { b } else { a });
341 24 : let (mut ts_min, mut ts_max) = (None::<String>, None::<String>);
342 32 : for instances in w.attrs.values_mut() {
343 32 : match &boundary {
344 628 : Some(bd) => instances.retain(|inst| {
345 628 : key(inst).is_none_or(|k| if descending { k >= *bd } else { k <= *bd })
346 628 : }),
347 0 : None => instances.truncate(TEMPORAL_INSTANCE_LIMIT),
348 : }
349 232 : for inst in instances.iter() {
350 232 : if let Some(t) = inst.get(timeprop).and_then(Value::as_str) {
351 232 : if ts_min.as_deref().is_none_or(|m| dt_key(t) < dt_key(m)) {
352 92 : ts_min = Some(t.to_owned());
353 140 : }
354 232 : if ts_max.as_deref().is_none_or(|m| dt_key(t) > dt_key(m)) {
355 132 : ts_max = Some(t.to_owned());
356 134 : }
357 0 : }
358 : }
359 : }
360 24 : (w.ts_min, w.ts_max) = (ts_min, ts_max);
361 678 : }
362 :
363 : /// 5.7.4.4 S4/S7: does a scope VALUE (string or array of strings) match the
364 : /// 4.19 Scope query? The 4.5.7 deletion sentinel never matches.
365 356 : fn scope_value_matches(sq: &str, v: &Value) -> bool {
366 356 : if v.is_null() || v.as_str() == Some("urn:ngsi-ld:null") {
367 0 : return false;
368 356 : }
369 356 : crate::scope_matches(sq, &serde_json::json!({ "scope": v }))
370 356 : }
371 :
372 : /// 4.18 over a Temporal Evolution: "a given Scope is considered valid from
373 : /// the time it has been set until the time it has been explicitly removed by
374 : /// an update or delete operation" (example: annex C.5.16). Instance-shaped
375 : /// scope arrays become [set-time, next-set-time) validity intervals
376 : /// (set-time = observedAt‖modifiedAt‖createdAt — 4.5.6 mirrors them from the
377 : /// Core API change); a plain string/array scope is valid for all time.
378 : /// Returns only the intervals whose value matches `sq`; "" start = -inf,
379 : /// None end = +inf.
380 368 : fn scope_match_intervals(doc: &Value, sq: &str) -> Vec<(String, Option<String>)> {
381 368 : match doc.get("scope") {
382 308 : Some(Value::Array(a)) if a.first().is_some_and(Value::is_object) => {
383 304 : let mut states: Vec<(&str, &Value)> = a
384 304 : .iter()
385 356 : .filter_map(|i| {
386 356 : scope_set_time(i).map(|t| (t, i.get("value").unwrap_or(&Value::Null)))
387 356 : })
388 304 : .collect();
389 304 : states.sort_by_key(|(t, _)| dt_key(t));
390 304 : (0..states.len())
391 356 : .filter(|&n| scope_value_matches(sq, states[n].1))
392 304 : .map(|n| {
393 : (
394 108 : states[n].0.to_owned(),
395 108 : states.get(n + 1).map(|(t, _)| (*t).to_owned()),
396 : )
397 108 : })
398 304 : .collect()
399 : }
400 4 : Some(_) if crate::scope_matches(sq, doc) => vec![(String::new(), None)],
401 62 : _ => Vec::new(),
402 : }
403 368 : }
404 :
405 : /// The time a temporal scope instance was set (4.5.6: observedAt is a copy
406 : /// of modifiedAt on Core-API changes; direct 5.6.11 input may carry any).
407 550 : fn scope_set_time(i: &Value) -> Option<&str> {
408 550 : ["observedAt", "modifiedAt", "createdAt"]
409 550 : .iter()
410 550 : .find_map(|k| i.get(*k).and_then(Value::as_str))
411 550 : }
412 :
413 696 : fn window(
414 696 : doc: &Value,
415 696 : tq: Option<&TemporalQ>,
416 696 : last_n: Option<usize>,
417 696 : attrs_filter: Option<&Vec<String>>,
418 696 : omit: Option<&Vec<crate::repr::ProjNode>>,
419 696 : dataset: Option<&Vec<String>>,
420 696 : timeprop: &str,
421 696 : ) -> Windowed {
422 696 : let mut w = Windowed {
423 696 : attrs: std::collections::BTreeMap::new(),
424 696 : max_per_attr: 0,
425 696 : ts_min: None,
426 696 : ts_max: None,
427 696 : truncated: false,
428 696 : };
429 696 : let Some(obj) = doc.as_object() else { return w };
430 3798 : for (k, v) in obj {
431 : // 4.5.6: the Scope of a Temporal Evolution is represented as the
432 : // temporal representation of a Property — instance-shaped scope
433 : // arrays window like attributes (plain-string scope stays meta).
434 3798 : let scope_instances = k == "scope"
435 102 : && v.as_array()
436 102 : .is_some_and(|a| a.first().is_some_and(Value::is_object));
437 3798 : if is_meta(k) && !scope_instances {
438 2714 : continue;
439 1084 : }
440 1084 : if let Some(want) = attrs_filter {
441 26 : if !want.contains(k) {
442 12 : continue;
443 14 : }
444 1058 : }
445 1072 : if let Some(omit) = omit {
446 20 : if omit.iter().any(|n| n.iri == *k && n.children.is_none()) {
447 10 : continue;
448 10 : }
449 1052 : }
450 1062 : let mut instances: Vec<Value> = v
451 1062 : .as_array()
452 1062 : .cloned()
453 1062 : .unwrap_or_default()
454 1062 : .into_iter()
455 1948 : .filter(|inst| tq.is_none_or(|tq| tq.instance_matches(inst)))
456 1836 : .filter(|inst| match (dataset, inst.get("datasetId")) {
457 1836 : (None, _) => true,
458 0 : (Some(want), Some(Value::String(have))) => want.iter().any(|w| w == have),
459 0 : (Some(want), None) => want.iter().any(|w| w == "@none"),
460 0 : _ => false,
461 1836 : })
462 1062 : .collect();
463 : // 4.18/C.5.16: the scope valid AT the window start was set at or
464 : // before it — carry the latest pre-window instance into the
465 : // representation (a temporal scope stays valid until replaced).
466 1062 : if scope_instances {
467 96 : let start = tq.and_then(|t| match t.timerel.as_str() {
468 94 : "after" | "between" => Some(t.time_at.as_str()),
469 0 : _ => None,
470 94 : });
471 96 : if let Some(start) = start {
472 94 : let carry = v
473 94 : .as_array()
474 94 : .into_iter()
475 94 : .flatten()
476 114 : .filter(|i| scope_set_time(i).is_some_and(|t| dt_key(t) < dt_key(start)))
477 94 : .max_by_key(|i| scope_set_time(i).map(dt_key))
478 94 : .cloned();
479 94 : if let Some(c) = carry {
480 80 : if !instances.contains(&c) {
481 80 : instances.push(c);
482 80 : }
483 14 : }
484 2 : }
485 966 : }
486 1062 : instances.sort_by(|a, b| {
487 : // Canonicalize before comparing: '.' sorts before 'Z', so a raw
488 : // string compare puts "…00.5Z" ahead of "…00Z" and lastN keeps the
489 : // wrong instant. 4.6.3 allows both fraction spellings.
490 854 : let ta = a.get(timeprop).and_then(Value::as_str).unwrap_or("");
491 854 : let tb = b.get(timeprop).and_then(Value::as_str).unwrap_or("");
492 854 : dt_key(ta).cmp(&dt_key(tb))
493 854 : });
494 1062 : if let Some(n) = last_n {
495 20 : if instances.len() > n {
496 12 : instances = instances.split_off(instances.len() - n);
497 12 : }
498 : // lastN delivers newest-first (DESC), Scorpio parity
499 20 : instances.reverse();
500 1042 : }
501 1062 : if instances.is_empty() {
502 0 : continue;
503 1062 : }
504 1062 : w.max_per_attr = w.max_per_attr.max(instances.len());
505 1884 : for inst in &instances {
506 1884 : if let Some(t) = inst.get(timeprop).and_then(Value::as_str) {
507 1846 : if w.ts_min.as_deref().is_none_or(|m| dt_key(t) < dt_key(m)) {
508 966 : w.ts_min = Some(t.to_owned());
509 988 : }
510 1846 : if w.ts_max.as_deref().is_none_or(|m| dt_key(t) > dt_key(m)) {
511 1132 : w.ts_max = Some(t.to_owned());
512 1132 : }
513 38 : }
514 : }
515 1062 : w.attrs.insert(k.clone(), instances);
516 : }
517 696 : w
518 696 : }
519 :
520 : /// `Content-Range: date-time <start>-<end>/<size>` (Scorpio-parity semantics).
521 704 : fn content_range(
522 704 : truncated: bool,
523 704 : ts_min: Option<&str>,
524 704 : ts_max: Option<&str>,
525 704 : tq: Option<&TemporalQ>,
526 704 : last_n: Option<usize>,
527 704 : ) -> Option<String> {
528 704 : if !truncated {
529 690 : return None;
530 14 : }
531 14 : let (data_min, data_max) = (ts_min?, ts_max?);
532 : // The window bound is the query's own when the query names one; matching
533 : // on the pair keeps the timerel and the query that produced it together,
534 : // so no arm can reach for a query that is not there.
535 14 : let named = tq.filter(|t| t.timerel != "any");
536 14 : let (start, end) = if last_n.is_none() {
537 10 : let start = match named.map(|t| (t.timerel.as_str(), t)) {
538 6 : Some(("after" | "between", t)) => t.time_at.clone(),
539 4 : _ => data_min.to_owned(),
540 : };
541 10 : (start, data_max.to_owned())
542 : } else {
543 4 : let start = match named.map(|t| (t.timerel.as_str(), t)) {
544 0 : Some(("before", t)) => t.time_at.clone(),
545 0 : Some(("between", t)) => t.end_time_at.clone().unwrap_or_else(|| data_max.to_owned()),
546 4 : _ => data_max.to_owned(),
547 : };
548 4 : (start, data_min.to_owned())
549 : };
550 : // start/end bound the instances actually returned; the size is the length
551 : // of the complete representation the client asked for — the requested
552 : // lastN, or "*" when the window leaves it unknown.
553 14 : let size = last_n.map_or_else(|| "*".to_owned(), |n| n.to_string());
554 14 : Some(format!("date-time {start}-{end}/{size}"))
555 704 : }
556 :
557 : /// Render one temporal entity from its windowed data.
558 676 : fn present_temporal(
559 676 : doc: &Value,
560 676 : w: &Windowed,
561 676 : ctx: &Context,
562 676 : r: &TRepr,
563 676 : tq: Option<&TemporalQ>,
564 676 : timeprop: &str,
565 676 : ) -> Result<Value, NgsiError> {
566 676 : let Some(obj) = doc.as_object() else {
567 0 : return Ok(doc.clone());
568 : };
569 676 : let mut out = Map::new();
570 3754 : for (k, v) in obj {
571 3754 : let scope_instances = k == "scope"
572 102 : && v.as_array()
573 102 : .is_some_and(|a| a.first().is_some_and(Value::is_object));
574 3754 : if is_meta(k) && !scope_instances {
575 2690 : match k.as_str() {
576 : // Table 6.3.11-1: `sysAttrs` is what admits "the system
577 : // generated temporal attributes createdAt, modifiedAt and
578 : // the system temporal attribute expiresAt … In the case of
579 : // temporal representations, also the system generated
580 : // temporal attribute deletedAt". Without it none of them is
581 : // in the payload — the same set `repr.rs` gates on the
582 : // current-state path.
583 2690 : "createdAt" | "modifiedAt" | "expiresAt" | "deletedAt" if !r.sys => continue,
584 1432 : _ => {}
585 : }
586 1432 : if !crate::repr::meta_projected(r.pick.as_deref(), r.omit.as_deref(), k) {
587 12 : continue;
588 1420 : }
589 1420 : if k == "type" {
590 670 : out.insert("type".into(), antares_jsonld::compact_types(v, ctx));
591 750 : } else {
592 750 : out.insert(k.clone(), v.clone());
593 750 : }
594 1064 : }
595 : }
596 676 : if r.aggregated {
597 26 : for (k, v) in render_aggregated(w, tq, r, ctx, timeprop)? {
598 12 : out.insert(k, v);
599 12 : }
600 20 : return Ok(Value::Object(out));
601 650 : }
602 1008 : for (k, instances) in &w.attrs {
603 : // Table 6.18.3.2-1 / 6.19.3.1 `lang`: each LanguageProperty
604 : // instance becomes a Property in the chosen language (4.15) before
605 : // either representation renders it.
606 : let reduced: Vec<Value>;
607 1008 : let instances: &[Value] = match &r.lang {
608 48 : Some(lang) => {
609 48 : reduced = instances
610 48 : .iter()
611 72 : .map(|inst| {
612 72 : let mut inst = inst.clone();
613 72 : if let Some(o) = inst.as_object_mut() {
614 72 : crate::repr::apply_lang(o, lang);
615 72 : }
616 72 : inst
617 72 : })
618 48 : .collect();
619 48 : &reduced
620 : }
621 960 : None => instances,
622 : };
623 1008 : if instances.is_empty() {
624 : // gap-cut leftovers render as empty arrays
625 0 : out.insert(ctx.compact_iri(k), Value::Array(vec![]));
626 0 : continue;
627 1008 : }
628 1008 : if r.temporal_values {
629 : // group instances by datasetId (4.5.9)
630 44 : let mut groups: Vec<(Option<String>, Vec<&Value>)> = Vec::new();
631 56 : for inst in instances {
632 56 : let ds = inst
633 56 : .get("datasetId")
634 56 : .and_then(Value::as_str)
635 56 : .map(String::from);
636 56 : match groups.iter_mut().find(|(g, _)| *g == ds) {
637 12 : Some((_, list)) => list.push(inst),
638 44 : None => groups.push((ds, vec![inst])),
639 : }
640 : }
641 44 : let mut rendered: Vec<Value> = groups
642 44 : .iter()
643 44 : .map(|(ds, list)| {
644 44 : let atype = list
645 44 : .first()
646 44 : .and_then(|i| i.get("type"))
647 44 : .cloned()
648 44 : .unwrap_or_else(|| Value::String("Property".into()));
649 44 : let values: Vec<Value> = list
650 44 : .iter()
651 56 : .map(|inst| {
652 : // 4.5.9: Property/Relationship pairs carry the bare
653 : // value/object; other attribute kinds wrap it under
654 : // their member name.
655 56 : let v = if let Some(v) = inst.get("value") {
656 : // 4.5.9: "the first element shall be a
657 : // Property value" — the one the instance
658 : // holds. A f64 round trip would retype an
659 : // integer and drop a digit past 2^53.
660 36 : v.clone()
661 20 : } else if let Some(o) = inst.get("object") {
662 2 : o.clone()
663 18 : } else if let Some(lm) = inst.get("languageMap") {
664 2 : serde_json::json!({"languageMap": lm})
665 16 : } else if let Some(j) = inst.get("json") {
666 0 : serde_json::json!({"json": j})
667 16 : } else if let Some(vv) = inst.get("vocab") {
668 0 : let compacted = match vv {
669 0 : Value::String(iri) => Value::String(ctx.compact_iri(iri)),
670 0 : Value::Array(a) => Value::Array(
671 0 : a.iter()
672 0 : .map(|s| match s {
673 0 : Value::String(iri) => {
674 0 : Value::String(ctx.compact_iri(iri))
675 : }
676 0 : o => o.clone(),
677 0 : })
678 0 : .collect(),
679 : ),
680 0 : o => o.clone(),
681 : };
682 0 : serde_json::json!({"vocab": compacted})
683 16 : } else if let Some(l) = inst.get("valueList") {
684 : // 4.5.9 p.63 EXAMPLE 3: the pair's first element
685 : // is the BARE ordered array, not a {"valueList"}
686 : // wrapper — unlike languageMap/
687 : // json/vocab, which the clause does wrap
688 10 : l.clone()
689 6 : } else if let Some(l) = inst.get("objectList") {
690 : // 4.5.9 p.65: same bare form for ListRelationship
691 6 : l.clone()
692 : } else {
693 0 : Value::Null
694 : };
695 56 : let t = inst.get(timeprop).cloned().unwrap_or(Value::Null);
696 56 : Value::Array(vec![v, t])
697 56 : })
698 44 : .collect();
699 44 : let mut o = Map::new();
700 : // 4.5.9: the simplified member name follows the attribute type
701 44 : let member = match atype.as_str() {
702 44 : Some("Relationship") => "objects",
703 42 : Some("LanguageProperty") => "languageMaps",
704 40 : Some("VocabProperty") => "vocabs",
705 40 : Some("JsonProperty") => "jsons",
706 40 : Some("ListProperty") => "valueLists",
707 34 : Some("ListRelationship") => "objectLists",
708 28 : _ => "values",
709 : };
710 44 : o.insert("type".into(), atype);
711 44 : if let Some(ds) = ds {
712 0 : o.insert("datasetId".into(), Value::String(ds.clone()));
713 44 : }
714 44 : o.insert(member.into(), Value::Array(values));
715 44 : Value::Object(o)
716 44 : })
717 44 : .collect();
718 44 : let rendered = if rendered.len() == 1 {
719 44 : rendered.remove(0)
720 : } else {
721 0 : Value::Array(rendered)
722 : };
723 44 : out.insert(ctx.compact_iri(k), rendered);
724 : } else {
725 964 : let presented: Vec<Value> = instances
726 964 : .iter()
727 1144 : .map(|inst| {
728 1144 : let mut ci = inst.clone();
729 1144 : if !r.sys {
730 1106 : if let Some(o) = ci.as_object_mut() {
731 1106 : o.remove("createdAt");
732 1106 : o.remove("modifiedAt");
733 1106 : // 6.3.11: expiresAt is sysAttrs-gated too
734 1106 : o.remove("expiresAt");
735 1106 : }
736 38 : }
737 1144 : compact_instance(&ci, ctx)
738 1144 : })
739 964 : .collect();
740 964 : out.insert(ctx.compact_iri(k), Value::Array(presented));
741 : }
742 : }
743 650 : Ok(Value::Object(out))
744 676 : }
745 :
746 : /// Parsed temporal representation params (options/format/lastN/pick/omit/
747 : /// datasetId/aggregation), fully validated up front.
748 : #[derive(Default, Clone)]
749 : struct TRepr {
750 : temporal_values: bool,
751 : aggregated: bool,
752 : sys: bool,
753 : last_n: Option<usize>,
754 : pick: Option<Vec<crate::repr::ProjNode>>,
755 : omit: Option<Vec<crate::repr::ProjNode>>,
756 : dataset_id: Option<Vec<String>>,
757 : attrs: Option<Vec<String>>,
758 : aggr_methods: Vec<String>,
759 : aggr_period: AggrPeriod,
760 : lang: Option<String>,
761 : }
762 :
763 : #[derive(Clone, Copy, Debug, Default, PartialEq)]
764 : enum AggrPeriod {
765 : /// PT0S / absent: one bucket over the whole range
766 : #[default]
767 : Whole,
768 : Seconds(i64),
769 : /// 4.5.19.1: a period may mix date and time elements
770 : /// ("P3Y6M4DT12H30M5S"), so a month step carries the leftover seconds —
771 : /// months are not a fixed number of seconds and cannot be folded in.
772 : Months(u32, i64),
773 : }
774 :
775 92 : fn parse_iso_duration(s: &str) -> Option<AggrPeriod> {
776 92 : let d = antares_model::parse_iso_duration(s)?;
777 : // saturating: an absurd magnitude must not panic (debug) or wrap
778 : // (release) — f64→int `as` casts already saturate, guard the ops.
779 72 : let months = (d.years as u32)
780 72 : .saturating_mul(12)
781 72 : .saturating_add(d.months as u32);
782 72 : let secs = [
783 72 : (d.weeks, 604_800.0),
784 72 : (d.days, 86_400.0),
785 72 : (d.hours, 3_600.0),
786 72 : (d.minutes, 60.0),
787 72 : (d.seconds, 1.0),
788 72 : ]
789 72 : .into_iter()
790 360 : .fold(0i64, |acc, (n, per)| acc.saturating_add((n * per) as i64));
791 72 : Some(match (months, secs) {
792 30 : (0, 0) => AggrPeriod::Whole,
793 18 : (0, sc) => AggrPeriod::Seconds(sc),
794 24 : (m, sc) => AggrPeriod::Months(m, sc),
795 : })
796 92 : }
797 :
798 : const AGGR_METHODS: &[&str] = &[
799 : "totalCount",
800 : "distinctCount",
801 : "sum",
802 : "avg",
803 : "min",
804 : "max",
805 : "stddev",
806 : "sumsq",
807 : ];
808 :
809 : /// 5.7.3.4 / 5.7.4.4: "If projection attributes are present and indicate the
810 : /// use of Linked Entity retrieval, an error of type BadRequestData shall be
811 : /// raised." Unconditional on both temporal consumption operations, because
812 : /// neither defines a join; only the clause number in the message differs.
813 808 : fn reject_linked_projection(trepr: &TRepr, clause: &str) -> Result<(), NgsiError> {
814 1596 : let depth = |p: &Option<Vec<crate::repr::ProjNode>>| {
815 1596 : p.as_deref().map(crate::repr::proj_depth).unwrap_or(0)
816 1596 : };
817 808 : if depth(&trepr.pick) > 0 || depth(&trepr.omit) > 0 {
818 24 : return Err(NgsiError::BadRequestData(format!(
819 24 : "temporal projection must not use Linked Entity selection ({clause})"
820 24 : )));
821 784 : }
822 784 : Ok(())
823 808 : }
824 :
825 872 : fn parse_trepr(params: &HashMap<String, String>, ctx: &Context) -> Result<TRepr, NgsiError> {
826 872 : let mut r = TRepr {
827 872 : lang: params.get("lang").cloned(),
828 872 : ..TRepr::default()
829 872 : };
830 872 : if let Some(opts) = params.get("options") {
831 78 : for o in opts.split(',') {
832 78 : match o.trim() {
833 78 : "sysAttrs" => r.sys = true,
834 42 : "temporalValues" => r.temporal_values = true,
835 36 : "aggregatedValues" => r.aggregated = true,
836 0 : "normalized" => {}
837 0 : other => {
838 0 : return Err(NgsiError::InvalidRequest(format!(
839 0 : "unsupported options value {other:?}"
840 0 : )))
841 : }
842 : }
843 : }
844 794 : }
845 : // format wins over options on conflict (6.3.7)
846 872 : if let Some(f) = params.get("format") {
847 42 : match f.as_str() {
848 42 : "temporalValues" => {
849 10 : r.temporal_values = true;
850 10 : r.aggregated = false;
851 10 : }
852 32 : "aggregatedValues" => {
853 32 : r.aggregated = true;
854 32 : r.temporal_values = false;
855 32 : }
856 0 : "normalized" => {
857 0 : r.temporal_values = false;
858 0 : r.aggregated = false;
859 0 : }
860 0 : other => {
861 0 : return Err(NgsiError::InvalidRequest(format!(
862 0 : "unsupported format value {other:?}"
863 0 : )))
864 : }
865 : }
866 830 : }
867 872 : crate::repr::check_projection_exclusive(params)?;
868 856 : if let Some(pck) = params.get("pick") {
869 36 : r.pick = Some(crate::repr::parse_projection(pck, ctx)?);
870 820 : }
871 856 : if let Some(o) = params.get("omit") {
872 8 : r.omit = Some(crate::repr::parse_projection(o, ctx)?);
873 848 : }
874 856 : if let Some(a) = params.get("attrs") {
875 20 : r.attrs = Some(a.split(',').map(|t| ctx.expand_key(t.trim())).collect());
876 836 : }
877 856 : r.dataset_id = params
878 856 : .get("datasetId")
879 856 : .map(|s| s.split(',').map(|d| d.trim().to_owned()).collect());
880 856 : r.last_n = match params.get("lastN") {
881 20 : Some(n) => {
882 : // 5.2.21: lastN is a POSITIVE integer — 0 is outside the value
883 : // space.
884 20 : let v = n
885 20 : .parse::<usize>()
886 20 : .ok()
887 20 : .filter(|v| *v >= 1)
888 20 : .ok_or_else(|| NgsiError::BadRequestData(format!("invalid lastN {n:?}")))?;
889 : // Above i64::MAX it wraps negative when bound as the RANK cap
890 : // (`rk <= $n::bigint`), silently returning an empty set.
891 12 : if v > i64::MAX as usize {
892 0 : return Err(NgsiError::BadRequestData(format!(
893 0 : "lastN {v} is out of range"
894 0 : )));
895 12 : }
896 12 : Some(v)
897 : }
898 836 : None => None,
899 : };
900 848 : if let Some(m) = params.get("aggrMethods") {
901 100 : for method in m.split(',') {
902 100 : let method = method.trim();
903 100 : if !AGGR_METHODS.contains(&method) {
904 12 : return Err(NgsiError::BadRequestData(format!(
905 12 : "invalid aggrMethods value {method:?} (4.5.19)"
906 12 : )));
907 88 : }
908 88 : r.aggr_methods.push(method.to_owned());
909 : }
910 : // aggrMethods implies aggregation UNLESS an explicit format says otherwise
911 56 : if !params.contains_key("format") {
912 32 : r.aggregated = true;
913 38 : }
914 780 : }
915 836 : if r.aggregated && r.aggr_methods.is_empty() {
916 4 : return Err(NgsiError::BadRequestData(
917 4 : "aggregatedValues requires aggrMethods (4.5.19)".into(),
918 4 : ));
919 832 : }
920 832 : if let Some(d) = params.get("aggrPeriodDuration") {
921 28 : let p = parse_iso_duration(d).ok_or_else(|| {
922 8 : NgsiError::BadRequestData(format!("invalid aggrPeriodDuration {d:?}"))
923 8 : })?;
924 : // 4.11: the value space ends where duration arithmetic does —
925 : // beyond ~100 years chrono::Duration::seconds is out of bounds
926 : // (a panic, i.e. a remote 500), so such periods are rejected.
927 : // Both components are bounded at ~100 years, the months in their own
928 : // unit since a month is not a fixed number of seconds.
929 20 : let (months, secs) = match p {
930 10 : AggrPeriod::Whole => (0, 0),
931 10 : AggrPeriod::Seconds(sc) => (0, sc),
932 0 : AggrPeriod::Months(m, sc) => (m, sc),
933 : };
934 20 : if months > 1200 || secs > 86_400 * 366 * 100 {
935 4 : return Err(NgsiError::BadRequestData(format!(
936 4 : "aggrPeriodDuration {d:?} is out of range"
937 4 : )));
938 16 : }
939 16 : r.aggr_period = p;
940 804 : }
941 820 : Ok(r)
942 872 : }
943 :
944 : /// The attribute-selection set for windowing: attrs= or pick=.
945 736 : fn selection(r: &TRepr) -> Option<Vec<String>> {
946 736 : if let Some(a) = &r.attrs {
947 12 : return Some(a.clone());
948 724 : }
949 : // core-member picks (id/type/…) are presentation-only, not attr selection
950 724 : r.pick.as_ref().map(|p| {
951 12 : p.iter()
952 14 : .filter(|n| !is_meta(&n.raw))
953 12 : .map(|n| n.iri.clone())
954 12 : .collect()
955 12 : })
956 736 : }
957 :
958 : /// Aggregated representation (4.5.19): attr → `{type, <method>: [[v,start,end]]}`.
959 : /// Aggregation datatype class per 4.5.19.1 (Tables -1, -2, -3). Booleans
960 : /// count as numbers (1/0, table NOTE); a JSON String, a DateTime and a Date
961 : /// share the ordered min/max column; a Time additionally supports avg.
962 : #[derive(Clone, Copy, PartialEq, Debug)]
963 : enum AggrClass {
964 : Number,
965 : Text,
966 : /// 4.6.3 DateTime or Date: ordered, and Table 4.5.19.1-2 gives it no
967 : /// arithmetic.
968 : Instant,
969 : TimeOfDay,
970 : List,
971 : Opaque,
972 : Relationship,
973 : }
974 :
975 : /// The 4.6.3 datatype a Property instance's value carries, in either
976 : /// representation C.6 gives for one: a JSON-LD typed value
977 : /// (`{"@type": "DateTime", "@value": …}`), or a string whose `valueType`
978 : /// names the datatype and is coerced to its URI on the way in (4.5.2.2).
979 : /// Table 4.5.19.1-2 applies to the datatype, not to the spelling.
980 98 : fn value_datatype(inst: &Value) -> Option<&str> {
981 56 : fn term(s: &str) -> &str {
982 56 : s.strip_prefix(antares_jsonld::NGSI_LD_BASE).unwrap_or(s)
983 56 : }
984 98 : if let Some(vt) = inst.get("valueType").and_then(Value::as_str) {
985 4 : if inst.get("value").is_some_and(Value::is_string) {
986 4 : return Some(term(vt));
987 0 : }
988 94 : }
989 94 : inst.get("value")?
990 94 : .get("@type")
991 94 : .and_then(Value::as_str)
992 94 : .map(term)
993 98 : }
994 :
995 : /// The lexical form of a value: the string itself, or the `@value` of a
996 : /// JSON-LD typed value.
997 152 : fn lexical_of(v: &Value) -> Option<&str> {
998 152 : v.as_str()
999 152 : .or_else(|| v.get("@value").and_then(Value::as_str))
1000 152 : }
1001 :
1002 : /// The key the ordered classes compare by. A JSON String and a Date are
1003 : /// compared as written — 4.6.3 fixes the width of every component of a Date,
1004 : /// so lexicographical order is chronological — a DateTime by its canonical
1005 : /// instant, since an optional seconds fraction is written before the `Z` it
1006 : /// follows and sorts ahead of it, and a Time by its second of the day, at a
1007 : /// fixed width so one string comparison serves all four.
1008 112 : fn order_key(class: AggrClass, v: &Value) -> Option<String> {
1009 112 : let s = lexical_of(v)?;
1010 112 : Some(match class {
1011 56 : AggrClass::Instant => antares_model::dt_key(s),
1012 16 : AggrClass::TimeOfDay => format!("{:013.6}", seconds_of_day(s)?),
1013 40 : _ => s.to_owned(),
1014 : })
1015 112 : }
1016 :
1017 110 : fn classify_instance(inst: &Value) -> AggrClass {
1018 110 : if inst.get("object").is_some() {
1019 8 : return AggrClass::Relationship;
1020 102 : }
1021 102 : if inst.get("valueList").is_some() || inst.get("objectList").is_some() {
1022 0 : return AggrClass::List;
1023 102 : }
1024 102 : if inst.get("vocab").is_some()
1025 98 : || inst.get("languageMap").is_some()
1026 98 : || inst.get("json").is_some()
1027 : {
1028 : // URI / JSON-object valued kinds: only counting methods apply
1029 4 : return AggrClass::Opaque;
1030 98 : }
1031 98 : match value_datatype(inst) {
1032 56 : Some("DateTime" | "Date") => return AggrClass::Instant,
1033 20 : Some("Time") => return AggrClass::TimeOfDay,
1034 42 : _ => {}
1035 : }
1036 42 : match inst.get("value") {
1037 26 : Some(Value::Number(_)) | Some(Value::Bool(_)) => AggrClass::Number,
1038 16 : Some(Value::String(_)) => AggrClass::Text,
1039 0 : Some(Value::Array(_)) => AggrClass::List,
1040 0 : _ => AggrClass::Opaque,
1041 : }
1042 110 : }
1043 :
1044 : /// Table 4.5.19.1 eligibility: which methods apply to which datatype class.
1045 164 : fn aggr_eligible(class: AggrClass, method: &str) -> bool {
1046 164 : match method {
1047 164 : "totalCount" | "distinctCount" => true,
1048 98 : "min" | "max" => !matches!(class, AggrClass::Opaque | AggrClass::Relationship),
1049 50 : "avg" => matches!(
1050 16 : class,
1051 : AggrClass::Number | AggrClass::List | AggrClass::TimeOfDay
1052 : ),
1053 34 : "sum" => matches!(class, AggrClass::Number | AggrClass::List),
1054 16 : "stddev" | "sumsq" => matches!(class, AggrClass::Number),
1055 0 : _ => false,
1056 : }
1057 164 : }
1058 :
1059 : /// `HH:MM:SS[.f]` → seconds of day (4.6.3 Time is UTC with optional `Z`).
1060 56 : fn seconds_of_day(s: &str) -> Option<f64> {
1061 56 : let t = s.strip_suffix('Z').unwrap_or(s);
1062 56 : let b = t.as_bytes();
1063 56 : if b.len() < 8 || b[2] != b':' || b[5] != b':' {
1064 0 : return None;
1065 56 : }
1066 56 : let h: f64 = t.get(0..2)?.parse().ok()?;
1067 56 : let m: f64 = t.get(3..5)?.parse().ok()?;
1068 56 : let sec: f64 = t.get(6..)?.parse().ok()?;
1069 56 : (h < 24.0 && m < 60.0 && sec < 62.0).then_some(h * 3600.0 + m * 60.0 + sec)
1070 56 : }
1071 :
1072 : /// The raw member an instance carries its data under.
1073 212 : fn raw_of(inst: &Value) -> Option<&Value> {
1074 260 : for k in [
1075 212 : "value",
1076 212 : "object",
1077 212 : "valueList",
1078 212 : "objectList",
1079 212 : "vocab",
1080 212 : "languageMap",
1081 212 : "json",
1082 212 : ] {
1083 260 : if let Some(v) = inst.get(k) {
1084 212 : return Some(v);
1085 48 : }
1086 : }
1087 0 : None
1088 212 : }
1089 :
1090 : /// Numeric view of one raw value for the class (None ⇒ excluded from
1091 : /// numeric methods; List aggregates SIZES per Table 4.5.19.1-1).
1092 282 : fn numeric_of(class: AggrClass, v: &Value) -> Option<f64> {
1093 282 : match class {
1094 78 : AggrClass::Number => match v {
1095 0 : Value::Bool(b) => Some(if *b { 1.0 } else { 0.0 }),
1096 78 : _ => v.as_f64(),
1097 : },
1098 0 : AggrClass::List => v.as_array().map(|a| a.len() as f64),
1099 40 : AggrClass::TimeOfDay => lexical_of(v).and_then(seconds_of_day),
1100 164 : _ => None,
1101 : }
1102 282 : }
1103 :
1104 118 : fn render_aggregated(
1105 118 : w: &Windowed,
1106 118 : tq: Option<&TemporalQ>,
1107 118 : r: &TRepr,
1108 118 : ctx: &Context,
1109 118 : timeprop: &str,
1110 118 : ) -> Result<Map<String, Value>, NgsiError> {
1111 : use chrono::{DateTime, Datelike, FixedOffset};
1112 276 : let fmt = |d: DateTime<FixedOffset>| d.format("%Y-%m-%dT%H:%M:%SZ").to_string();
1113 118 : let mut out = Map::new();
1114 118 : for (k, instances) in &w.attrs {
1115 110 : let mut times: Vec<(DateTime<FixedOffset>, &Value)> = Vec::new();
1116 110 : let mut class: Option<AggrClass> = None;
1117 212 : for inst in instances {
1118 212 : let Some(t) = inst
1119 212 : .get(timeprop)
1120 212 : .and_then(Value::as_str)
1121 212 : .and_then(|t| DateTime::parse_from_rfc3339(t).ok())
1122 : else {
1123 0 : continue;
1124 : };
1125 212 : if class.is_none() {
1126 110 : class = Some(classify_instance(inst));
1127 110 : }
1128 212 : let Some(raw) = raw_of(inst) else { continue };
1129 212 : times.push((t, raw));
1130 : }
1131 110 : if times.is_empty() {
1132 0 : continue;
1133 110 : }
1134 : // Set alongside the first instance, and `times` is non-empty here.
1135 110 : let Some(class) = class else { continue };
1136 : // 5.7.4.4 p.211: "If an aggregated temporal representation is
1137 : // requested and any of the requested Attributes is not eligible for
1138 : // at least one of the aggregation methods specified in the request
1139 : // parameters, then an error of type InvalidRequest shall be raised."
1140 164 : for method in &r.aggr_methods {
1141 164 : if !aggr_eligible(class, method) {
1142 38 : return Err(NgsiError::InvalidRequest(format!(
1143 38 : "attribute {} ({class:?}-valued) is not eligible for \
1144 38 : aggregation method {method} (4.5.19.1, 5.7.4.4)",
1145 38 : ctx.compact_iri(k)
1146 38 : )));
1147 126 : }
1148 : }
1149 72 : times.sort_by_key(|(t, _)| *t);
1150 72 : let anchor = tq
1151 72 : .and_then(|tq| DateTime::parse_from_rfc3339(&tq.time_at).ok())
1152 72 : .unwrap_or(times[0].0);
1153 : // 4.5.19.1: "A duration of 0 second (e.g. expressed as "PT0S" or
1154 : // "P0D") is valid and is interpreted as a duration spanning the whole
1155 : // time range specified by the temporal query." The query names one
1156 : // edge of that range; 4.11 leaves the other open for `before` and
1157 : // `after`, so the data closes the open one.
1158 72 : let whole = {
1159 72 : let at = |s: &str| DateTime::parse_from_rfc3339(s).ok();
1160 72 : let (Some(&(first, _)), Some(&(last_at, _))) = (times.first(), times.last()) else {
1161 0 : continue;
1162 : };
1163 72 : let last = last_at + chrono::Duration::seconds(1);
1164 20 : match tq {
1165 32 : Some(q) if q.timerel == "before" => (first, at(&q.time_at).unwrap_or(last)),
1166 24 : Some(q) if q.timerel == "between" => (
1167 4 : at(&q.time_at).unwrap_or(first),
1168 4 : q.end_time_at.as_deref().and_then(at).unwrap_or(last),
1169 4 : ),
1170 20 : Some(q) if q.timerel == "after" => (at(&q.time_at).unwrap_or(first), last),
1171 40 : _ => (first, last),
1172 : }
1173 : };
1174 : // bucket boundaries
1175 72 : let bucket_of =
1176 160 : |t: DateTime<FixedOffset>| -> (DateTime<FixedOffset>, DateTime<FixedOffset>) {
1177 160 : match r.aggr_period {
1178 136 : AggrPeriod::Whole => whole,
1179 4 : AggrPeriod::Seconds(sc) => {
1180 : // checked throughout: an offset no representable
1181 : // date can hold puts the instant in one final
1182 : // open-ended bucket instead of panicking
1183 4 : let idx = (t - anchor).num_seconds().div_euclid(sc);
1184 4 : let bucket = idx
1185 4 : .checked_mul(sc)
1186 4 : .and_then(chrono::Duration::try_seconds)
1187 4 : .and_then(|off| anchor.checked_add_signed(off))
1188 4 : .and_then(|start| {
1189 4 : chrono::Duration::try_seconds(sc)
1190 4 : .and_then(|w| start.checked_add_signed(w))
1191 4 : .map(|end| (start, end))
1192 4 : });
1193 4 : match bucket {
1194 4 : Some(b) => b,
1195 0 : None => (anchor, chrono::DateTime::<chrono::Utc>::MAX_UTC.into()),
1196 : }
1197 : }
1198 20 : AggrPeriod::Months(m, sc) => {
1199 : // start of the k-th period, O(1) in k. Negative k are
1200 : // the periods BEFORE the anchor, which is what a
1201 : // `before` query is made of.
1202 76 : let step = |k: i64| -> Option<DateTime<FixedOffset>> {
1203 76 : let n = k.unsigned_abs().checked_mul(u64::from(m))?;
1204 76 : let n = chrono::Months::new(u32::try_from(n).ok()?);
1205 76 : let base = if k < 0 {
1206 40 : anchor.checked_sub_months(n)?
1207 : } else {
1208 36 : anchor.checked_add_months(n)?
1209 : };
1210 76 : let off = chrono::Duration::try_seconds(k.checked_mul(sc)?)?;
1211 76 : base.checked_add_signed(off)
1212 76 : };
1213 : // The whole-month distance ignores the day and the time
1214 : // of day, and one period is at least one month, so it
1215 : // brackets the exact index within one step: binary-search
1216 : // between it and the anchor instead of walking there,
1217 : // which is O(log) however far the instant is.
1218 20 : let approx = (i64::from(t.year() - anchor.year()) * 12
1219 20 : + i64::from(t.month())
1220 20 : - i64::from(anchor.month()))
1221 20 : .div_euclid(i64::from(m));
1222 20 : let (mut lo, mut hi) = (approx.min(0) - 1, approx.max(0) + 1);
1223 56 : while hi - lo > 1 {
1224 36 : let mid = lo + (hi - lo) / 2;
1225 36 : if step(mid).is_some_and(|s| s <= t) {
1226 24 : lo = mid;
1227 24 : } else {
1228 12 : hi = mid;
1229 12 : }
1230 : }
1231 : // saturate instead of panic: a huge month period or a
1232 : // far-future timeAt overflows chrono's range — treat the
1233 : // remainder as one open-ended bucket
1234 20 : match (step(lo), step(hi)) {
1235 20 : (Some(start), Some(end)) if start <= t => (start, end),
1236 0 : (Some(start), None) if start <= t => {
1237 0 : (start, chrono::DateTime::<chrono::Utc>::MAX_UTC.into())
1238 : }
1239 0 : _ => (anchor, chrono::DateTime::<chrono::Utc>::MAX_UTC.into()),
1240 : }
1241 : }
1242 : }
1243 160 : };
1244 : type Bucket = (DateTime<FixedOffset>, DateTime<FixedOffset>);
1245 72 : let mut buckets: Vec<(Bucket, Vec<&Value>)> = Vec::new();
1246 160 : for (t, v) in × {
1247 160 : let b = bucket_of(*t);
1248 160 : match buckets.last_mut() {
1249 88 : Some((bb, vals)) if bb.0 == b.0 => vals.push(v),
1250 84 : _ => buckets.push((b, vec![v])),
1251 : }
1252 : }
1253 72 : let mut attr_out = Map::new();
1254 : // 4.5.19.0: the member is labelled "Property" for Properties and
1255 : // "Relationship" for Relationships.
1256 72 : let label = if class == AggrClass::Relationship {
1257 6 : "Relationship"
1258 : } else {
1259 66 : "Property"
1260 : };
1261 72 : attr_out.insert("type".into(), Value::String(label.into()));
1262 126 : for method in &r.aggr_methods {
1263 126 : let rows: Vec<Value> = buckets
1264 126 : .iter()
1265 138 : .map(|((bs, be), vals)| {
1266 138 : let val = aggregate_bucket(method, class, vals);
1267 138 : Value::Array(vec![val, Value::String(fmt(*bs)), Value::String(fmt(*be))])
1268 138 : })
1269 126 : .collect();
1270 126 : attr_out.insert(method.clone(), Value::Array(rows));
1271 : }
1272 72 : out.insert(ctx.compact_iri(k), Value::Object(attr_out));
1273 : }
1274 80 : Ok(out)
1275 118 : }
1276 :
1277 : /// One bucket, one method — per-class semantics from Tables 4.5.19.1-1/2/3.
1278 : /// Never emits an out-of-range float (the old fold seeded with
1279 : /// f64::INFINITY, which serde_json serializes as null).
1280 138 : fn aggregate_bucket(method: &str, class: AggrClass, vals: &[&Value]) -> Value {
1281 282 : let nums: Vec<f64> = vals.iter().filter_map(|v| numeric_of(class, v)).collect();
1282 138 : let finite = |x: f64| {
1283 8 : if x.is_finite() {
1284 8 : serde_json::json!(x)
1285 : } else {
1286 0 : Value::Null
1287 : }
1288 8 : };
1289 138 : match method {
1290 138 : "totalCount" => serde_json::json!(vals.len()),
1291 80 : "distinctCount" => {
1292 : // Relationship: "count of distinct relationship TARGETS" — an
1293 : // object may be a URI or an array of URIs, so flatten first.
1294 20 : let mut seen: std::collections::HashSet<String> = std::collections::HashSet::new();
1295 40 : for v in vals {
1296 40 : let items: Vec<&Value> = match (class, v) {
1297 0 : (AggrClass::Relationship, Value::Array(a)) => a.iter().collect(),
1298 40 : _ => vec![*v],
1299 : };
1300 40 : for it in items {
1301 40 : seen.insert(it.to_string());
1302 40 : }
1303 : }
1304 20 : serde_json::json!(seen.len())
1305 : }
1306 60 : "min" | "max" => match class {
1307 : // ordered classes: the first or last value in the order the
1308 : // tables give the datatype, returned as it was written
1309 : AggrClass::Text | AggrClass::Instant | AggrClass::TimeOfDay => {
1310 48 : let mut keyed: Vec<(String, &Value)> = vals
1311 48 : .iter()
1312 112 : .filter_map(|v| order_key(class, v).map(|k| (k, *v)))
1313 48 : .collect();
1314 80 : keyed.sort_by(|a, b| a.0.cmp(&b.0));
1315 48 : let pick = if method == "min" {
1316 24 : keyed.first()
1317 : } else {
1318 24 : keyed.last()
1319 : };
1320 48 : pick.map_or(Value::Null, |(_, v)| (*v).clone())
1321 : }
1322 : _ => {
1323 0 : let it = nums.iter().copied();
1324 0 : let picked = if method == "min" {
1325 0 : it.fold(None, |a: Option<f64>, v| Some(a.map_or(v, |x| x.min(v))))
1326 : } else {
1327 0 : it.fold(None, |a: Option<f64>, v| Some(a.map_or(v, |x| x.max(v))))
1328 : };
1329 0 : picked.map_or(Value::Null, &finite)
1330 : }
1331 : },
1332 12 : "sum" => finite(nums.iter().sum::<f64>()),
1333 8 : "avg" => {
1334 8 : if nums.is_empty() {
1335 0 : Value::Null
1336 8 : } else if class == AggrClass::TimeOfDay {
1337 4 : let mean = nums.iter().sum::<f64>() / nums.len() as f64;
1338 4 : let (h, m, sec) = (
1339 4 : (mean / 3600.0) as u32,
1340 4 : ((mean % 3600.0) / 60.0) as u32,
1341 4 : (mean % 60.0) as u32,
1342 4 : );
1343 : // 4.6.3: a Time is `hh:mm:ssZ`, and its JSON-LD type is what
1344 : // tells a reader it is one — written bare it reads back as a
1345 : // JSON String, which Table 4.5.19.1-1 gives no average.
1346 4 : serde_json::json!({
1347 4 : "@type": "Time",
1348 4 : "@value": format!("{h:02}:{m:02}:{sec:02}Z"),
1349 : })
1350 : } else {
1351 4 : finite(nums.iter().sum::<f64>() / nums.len() as f64)
1352 : }
1353 : }
1354 0 : "stddev" => {
1355 0 : if nums.is_empty() {
1356 0 : Value::Null
1357 : } else {
1358 0 : let n = nums.len() as f64;
1359 0 : let mean = nums.iter().sum::<f64>() / n;
1360 0 : finite((nums.iter().map(|v| (v - mean).powi(2)).sum::<f64>() / n).sqrt())
1361 : }
1362 : }
1363 0 : "sumsq" => finite(nums.iter().map(|v| v * v).sum::<f64>()),
1364 0 : _ => Value::Null,
1365 : }
1366 138 : }
1367 :
1368 : // ---------- GET /temporal/entities/ (5.7.4) ----------
1369 :
1370 610 : pub async fn query_temporal(
1371 610 : State(st): State<AppState>,
1372 610 : CleanParams(params): CleanParams,
1373 610 : headers: HeaderMap,
1374 610 : ) -> Response {
1375 610 : match query_temporal_outer(&st, params, &headers).await {
1376 446 : Ok(r) => r,
1377 164 : Err(e) => e.into_response(),
1378 : }
1379 610 : }
1380 :
1381 : /// 5.7.4.4 EntityMap usage on the temporal query: a live map referenced by
1382 : /// the NGSILD-EntityMap header fixes the result set to the map's Entities
1383 : /// (5.5.14) and its location is echoed; an unknown or expired reference
1384 : /// means "a new EntityMap shall be created" (the entityMap=true branch,
1385 : /// answering 201 + the fresh location).
1386 610 : async fn query_temporal_outer(
1387 610 : st: &AppState,
1388 610 : mut params: HashMap<String, String>,
1389 610 : headers: &HeaderMap,
1390 610 : ) -> ApiResult<Response> {
1391 610 : let tenant = tenant_from(headers)?;
1392 610 : let filter =
1393 610 : gate!(st, &tenant, headers, "5.7.4", scope_q: params.get("scopeQ").map(String::as_str))
1394 610 : .await?;
1395 610 : let Some(map_ref) = single_header(headers, "NGSILD-EntityMap")? else {
1396 594 : return query_temporal_inner(st, ¶ms, headers, &filter).await;
1397 : };
1398 14 : let map_id = map_ref.rsplit('/').next().unwrap_or(&map_ref).to_owned();
1399 14 : let Some(mut map) = crate::entity_map::map_if_accessible(st, &tenant, headers, &map_id).await
1400 : else {
1401 6 : params.insert("entityMap".into(), "true".into());
1402 6 : return query_temporal_inner(st, ¶ms, headers, &filter).await;
1403 : };
1404 8 : params.remove("entityMap");
1405 : // 5.5.9.3: the map fixes the candidate set and the request's own filters
1406 : // narrow it, so `id=` on this request selects from the map rather than
1407 : // replacing it.
1408 8 : let candidates = crate::entity_map::candidate_ids(&map, ¶ms);
1409 : // "filters shall be rechecked before returning results" and "Entities not
1410 : // or no longer fitting the query shall be removed from the Entity map
1411 : // during pagination" — so the recheck asks about the map's OWN Entities,
1412 : // in bounded chunks. Asking the whole Tenant instead judged, and then
1413 : // deleted, entries this request never asked about, and lost every
1414 : // candidate past the first page of the recheck. Pruning is judgeable only
1415 : // for "@none" (local) entries: a remote-backed id may merely have an
1416 : // unreachable source right now (5.5.14). Known cost: this recheck is a
1417 : // second temporal query per map-using request, same shape as the entity
1418 : // query's filter re-run.
1419 8 : let mut matching: std::collections::HashSet<String> = std::collections::HashSet::new();
1420 8 : for chunk in candidates.chunks(st.max_limit.max(1)) {
1421 8 : let mut eff = params.clone();
1422 24 : for k in ["limit", "offset", "count"] {
1423 24 : eff.remove(k);
1424 24 : }
1425 8 : eff.insert("limit".into(), st.max_limit.to_string());
1426 8 : eff.insert("id".into(), chunk.join(","));
1427 8 : let resp = query_temporal_inner(st, &eff, headers, &filter).await?;
1428 8 : let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
1429 8 : .await
1430 8 : .map_err(|_| NgsiError::InternalError("entityMap recheck read".into()))?;
1431 8 : matching.extend(
1432 8 : serde_json::from_slice::<Value>(&bytes)
1433 8 : .ok()
1434 8 : .and_then(|v| v.as_array().cloned())
1435 8 : .unwrap_or_default()
1436 8 : .iter()
1437 8 : .filter_map(|d| d.get("id").and_then(Value::as_str).map(str::to_owned)),
1438 : );
1439 : }
1440 8 : if let Some(emap) = map.get_mut("entityMap").and_then(Value::as_object_mut) {
1441 8 : let stale: Vec<String> = candidates
1442 8 : .iter()
1443 8 : .filter(|eid| {
1444 8 : emap.get(eid.as_str())
1445 8 : .and_then(Value::as_array)
1446 8 : .is_some_and(|a| a.len() == 1 && a[0] == "@none")
1447 8 : && !matching.contains(eid.as_str())
1448 8 : })
1449 8 : .cloned()
1450 8 : .collect();
1451 8 : for k in stale {
1452 0 : emap.remove(&k);
1453 0 : }
1454 0 : }
1455 8 : crate::entity_map::map_put(st, &tenant, map.clone()).await?;
1456 : // fix the query to the candidates that survived the recheck (5.5.14)
1457 8 : let ids: Vec<&str> = candidates
1458 8 : .iter()
1459 8 : .filter(|id| map["entityMap"].get(id.as_str()).is_some())
1460 8 : .map(String::as_str)
1461 8 : .collect();
1462 8 : params.insert(
1463 8 : "id".into(),
1464 8 : if ids.is_empty() {
1465 0 : "urn:ngsi-ld:entitymap:empty".to_owned()
1466 : } else {
1467 8 : ids.join(",")
1468 : },
1469 : );
1470 8 : let mut resp = query_temporal_inner(st, ¶ms, headers, &filter).await?;
1471 8 : if let Ok(v) = format!("/ngsi-ld/v1/entityMaps/{map_id}").parse() {
1472 8 : resp.headers_mut().insert("NGSILD-EntityMap", v);
1473 8 : }
1474 8 : Ok(resp)
1475 610 : }
1476 :
1477 : /// The 5.7.4.4 preconditions of a temporal query, before any store is
1478 : /// touched: the filter must qualify, Linked Entity conditions are not
1479 : /// defined for it, a context source filter must parse, and ordering may
1480 : /// name only `id` and only where the execution stays local (4.23.1).
1481 628 : async fn check_temporal_query(
1482 628 : st: &AppState,
1483 628 : params: &HashMap<String, String>,
1484 628 : headers: &HeaderMap,
1485 628 : tenant: &TenantId,
1486 628 : ctx: &Context,
1487 628 : q_ast: Option<&antares_ql::QNode>,
1488 628 : ) -> ApiResult<()> {
1489 628 : let attrs_qualify = params.get("attrs").is_some_and(|a| {
1490 10 : a.split(',')
1491 10 : .any(|n| antares_ql::is_non_system_attr(n.trim()))
1492 10 : });
1493 628 : let q_qualifies = q_ast.is_some_and(|ast| {
1494 220 : ast.attribute_paths()
1495 220 : .iter()
1496 220 : .any(|h| antares_ql::is_non_system_attr(h))
1497 220 : });
1498 628 : let has_filter = params.contains_key("type")
1499 44 : || attrs_qualify
1500 44 : || q_qualifies
1501 44 : || params.contains_key("georel")
1502 44 : || params.get("local").map(String::as_str) == Some("true");
1503 628 : if !has_filter {
1504 8 : return Err(NgsiError::BadRequestData(
1505 8 : "temporal query needs at least one of type, attrs, q, georel (5.7.4)".into(),
1506 8 : )
1507 8 : .into());
1508 620 : }
1509 : // 5.7.4.4: Linked Entity retrieval is not defined for temporal queries —
1510 : // linked filter conditions are an unconditional BadRequestData
1511 620 : if q_ast.map(antares_ql::QNode::max_link_depth).unwrap_or(0) > 0 {
1512 8 : return Err(NgsiError::BadRequestData(
1513 8 : "temporal q must not reference Linked Entity attributes (5.7.4.4)".into(),
1514 8 : )
1515 8 : .into());
1516 612 : }
1517 : // 5.7.4.4: a syntactically invalid context source filter is 400; the
1518 : // filter itself gates registrations in federation::reg_matches.
1519 612 : if let Some(csf) = params.get("csf") {
1520 8 : parse_q(csf)?;
1521 604 : }
1522 604 : crate::paging::check_collation(params)?;
1523 : // 5.7.4.4: temporal ordering may only refer to the "id" entity member,
1524 : // and only where the execution "is limited to the local scope (see
1525 : // clause 5.5.13)" — 4.23.1 gives the reason: "Sort ordering is never
1526 : // applied to distributed operations." The subject is the EXECUTION, so a
1527 : // query nothing would federate to orders without `local=true`.
1528 602 : if let Some(spec) = params.get("orderBy") {
1529 16 : if crate::federation::would_federate(st, tenant, ctx, params, headers).await? {
1530 2 : return Err(NgsiError::BadRequestData(
1531 2 : "orderBy requires local scope — ordering is never applied to \
1532 2 : distributed operations (5.7.4.4, 4.23.1)"
1533 2 : .into(),
1534 2 : )
1535 2 : .into());
1536 14 : }
1537 14 : let non_id = spec.split(',').any(|part| {
1538 14 : let m = part.trim().split(';').next().unwrap_or("").trim();
1539 14 : m.split('[').next().unwrap_or(m) != "id"
1540 14 : });
1541 14 : if non_id {
1542 8 : return Err(NgsiError::BadRequestData(
1543 8 : "temporal orderBy may only name \"id\" (5.7.4.4)".into(),
1544 8 : )
1545 8 : .into());
1546 6 : }
1547 586 : }
1548 592 : Ok(())
1549 628 : }
1550 :
1551 : /// 4.5.19: the aggregated answer of a temporal query. The store computed
1552 : /// the bucket matrix, so the page is presented with the core members the
1553 : /// instance path renders and the aggregated attribute objects copied
1554 : /// under their compacted names.
1555 : #[allow(clippy::too_many_arguments)]
1556 6 : fn respond_aggregated(
1557 6 : st: &AppState,
1558 6 : params: &HashMap<String, String>,
1559 6 : tenant: &TenantId,
1560 6 : ctx: &Context,
1561 6 : accept: Accept,
1562 6 : trepr: &TRepr,
1563 6 : tq: Option<&TemporalQ>,
1564 6 : attrs_filter: Option<&Vec<String>>,
1565 6 : outcome: antares_store::filter::TemporalOutcome,
1566 6 : ) -> ApiResult<Response> {
1567 6 : let total = outcome
1568 6 : .total
1569 6 : .map(|t| t as usize)
1570 6 : .unwrap_or(outcome.rows.len());
1571 6 : let (page, count_hdr, links) = crate::paging::paginate_pre(
1572 6 : st,
1573 6 : params,
1574 6 : outcome.rows,
1575 6 : "/ngsi-ld/v1/temporal/entities",
1576 6 : total,
1577 0 : )?;
1578 6 : let timeprop = tq.map_or("observedAt", |t| t.timeproperty.as_str());
1579 6 : let none = Windowed {
1580 6 : attrs: Default::default(),
1581 6 : max_per_attr: 0,
1582 6 : ts_min: None,
1583 6 : ts_max: None,
1584 6 : truncated: false,
1585 6 : };
1586 6 : let mut payload: Vec<Value> = Vec::new();
1587 8 : for d in &page {
1588 : // core members exactly as the instance path presents them; the
1589 : // aggregated attribute objects are copied under compacted names
1590 8 : let mut presented = present_temporal(d, &none, ctx, trepr, tq, timeprop)?;
1591 8 : let Some(out) = presented.as_object_mut() else {
1592 0 : continue;
1593 : };
1594 8 : let mut any = false;
1595 48 : for (k, v) in d.as_object().into_iter().flatten() {
1596 48 : if is_meta(k) || attrs_filter.is_some_and(|a| !a.contains(k)) {
1597 34 : continue;
1598 14 : }
1599 14 : out.insert(ctx.compact_iri(k), v.clone());
1600 14 : any = true;
1601 : }
1602 8 : if any {
1603 8 : payload.push(presented);
1604 8 : }
1605 : }
1606 6 : let mut resp = crate::negotiate::respond_list(StatusCode::OK, payload, ctx, accept, tenant);
1607 6 : attach_paging(&mut resp, count_hdr, &links);
1608 6 : Ok(resp)
1609 6 : }
1610 :
1611 : /// The two Table 5.2.33-1 selectors that need no compiled pattern: the
1612 : /// type set, and the `attrs` list, which excludes an entity carrying
1613 : /// none of the named Attributes. Each drops an entity the store could
1614 : /// not narrow away itself.
1615 1430 : fn type_and_attrs_match(
1616 1430 : doc: &Value,
1617 1430 : types: Option<&Vec<String>>,
1618 1430 : entity_attr_filter: Option<&Vec<String>>,
1619 1430 : ) -> bool {
1620 1430 : if let Some(types) = types {
1621 1420 : let etypes = doc["type"].as_array().cloned().unwrap_or_default();
1622 1420 : if !etypes
1623 1420 : .iter()
1624 1420 : .any(|t| types.iter().any(|w| Some(w.as_str()) == t.as_str()))
1625 : {
1626 0 : return false;
1627 1420 : }
1628 10 : }
1629 1430 : if let Some(want) = entity_attr_filter {
1630 0 : if !want.iter().any(|a| doc.get(a).is_some()) {
1631 0 : return false;
1632 0 : }
1633 1430 : }
1634 1430 : true
1635 1430 : }
1636 :
1637 : /// 5.7.4.4 S2/S3: the values filter and the geoquery are judged on the
1638 : /// Attribute instances WITHIN the temporal-query interval, so an
1639 : /// out-of-window instance must not satisfy either.
1640 910 : fn window_conditions_match(
1641 910 : doc: &Value,
1642 910 : tq: Option<&TemporalQ>,
1643 910 : q_ast: Option<&antares_ql::QNode>,
1644 910 : geo: Option<&antares_ql::geo::GeoQuery>,
1645 910 : ctx: &Context,
1646 910 : ) -> bool {
1647 : // 5.7.4.4 S2/S3: the values filter and geoquery are checked against
1648 : // the Attribute instances WITHIN the temporal-query interval — an
1649 : // out-of-window instance must not satisfy them
1650 910 : let mut eval_doc = doc.clone();
1651 910 : if let (Some(tqv), Some(o)) = (tq, eval_doc.as_object_mut()) {
1652 4960 : for (k, v) in o.iter_mut() {
1653 4960 : if is_meta(k) {
1654 3632 : continue;
1655 1328 : }
1656 1328 : if let Some(arr) = v.as_array_mut() {
1657 1470 : arr.retain(|inst| tqv.instance_matches(inst));
1658 0 : }
1659 : }
1660 0 : }
1661 910 : if let Some(ast) = q_ast {
1662 820 : if !antares_ql::eval::eval_q(ast, &eval_doc, ctx, &|_| None) {
1663 606 : return false;
1664 214 : }
1665 90 : }
1666 304 : if let Some(g) = geo {
1667 90 : if !g.matches(&eval_doc, ctx) {
1668 58 : return false;
1669 32 : }
1670 214 : }
1671 246 : true
1672 910 : }
1673 :
1674 : /// 5.7.4.4 S4 — and S7: the federation merge precedes this check, so
1675 : /// split and aggregated entities are re-filtered here too. The entity
1676 : /// qualifies if one of its scope's 4.18 validity intervals intersects
1677 : /// the query window; attribute instances outside every matching interval
1678 : /// are excluded (annex C.5.16).
1679 368 : fn scope_window_retain(doc: &mut Value, sq: &str, tq: Option<&TemporalQ>, timeprop: &str) -> bool {
1680 368 : let iv = scope_match_intervals(doc, sq);
1681 368 : let (wstart, wend) = match tq {
1682 368 : Some(t) => match t.timerel.as_str() {
1683 368 : "before" => (None, Some(t.time_at.as_str())),
1684 368 : "after" => (Some(t.time_at.as_str()), None),
1685 362 : "between" => (Some(t.time_at.as_str()), t.end_time_at.as_deref()),
1686 0 : _ => (None, None),
1687 : },
1688 0 : None => (None, None),
1689 : };
1690 368 : let intersects = iv.iter().any(|(s, e)| {
1691 106 : wend.is_none_or(|w| s.as_str() < w)
1692 96 : && e.as_deref().is_none_or(|e| wstart.is_none_or(|w| e > w))
1693 106 : });
1694 368 : if !intersects {
1695 272 : return false;
1696 96 : }
1697 96 : if let Some(o) = doc.as_object_mut() {
1698 576 : for (k, v) in o.iter_mut() {
1699 576 : if is_meta(k) {
1700 480 : continue;
1701 96 : }
1702 96 : if let Some(arr) = v.as_array_mut() {
1703 130 : arr.retain(|inst| {
1704 130 : inst.get(timeprop).and_then(Value::as_str).is_some_and(|t| {
1705 130 : iv.iter()
1706 134 : .any(|(s, e)| t >= s.as_str() && e.as_deref().is_none_or(|e| t < e))
1707 130 : })
1708 130 : });
1709 0 : }
1710 : }
1711 0 : }
1712 96 : true
1713 368 : }
1714 :
1715 700 : pub(crate) async fn query_temporal_inner(
1716 700 : st: &AppState,
1717 700 : params: &HashMap<String, String>,
1718 700 : headers: &HeaderMap,
1719 700 : filter: &crate::policy::Filter,
1720 700 : ) -> ApiResult<Response> {
1721 700 : query_temporal_collected(st, params, headers, filter, None).await
1722 700 : }
1723 :
1724 : /// The same query, with the merged documents of each page handed to `merged`
1725 : /// as well as rendered into the response. 5.16.1.4 fills a Snapshot from this
1726 : /// query and has to STORE what it matched: for an Evolution held by a
1727 : /// registered Context Source there is nothing else in this broker to store —
1728 : /// the fan-out's answer is the only copy, and the response body is a
1729 : /// presentation of it that compaction does not undo.
1730 712 : pub(crate) async fn query_temporal_collected(
1731 712 : st: &AppState,
1732 712 : params: &HashMap<String, String>,
1733 712 : headers: &HeaderMap,
1734 712 : filter: &crate::policy::Filter,
1735 712 : merged: Option<&mut Vec<Value>>,
1736 712 : ) -> ApiResult<Response> {
1737 712 : let tenant = tenant_from(headers)?;
1738 712 : check_params(
1739 712 : params,
1740 712 : &[
1741 712 : "id",
1742 712 : "idPattern",
1743 712 : "type",
1744 712 : "attrs",
1745 712 : "q",
1746 712 : "georel",
1747 712 : "geometry",
1748 712 : "coordinates",
1749 712 : "geoproperty",
1750 712 : "scopeQ",
1751 712 : "csf",
1752 712 : "timerel",
1753 712 : "timeAt",
1754 712 : "endTimeAt",
1755 712 : "timeproperty",
1756 712 : "aggrMethods",
1757 712 : "aggrPeriodDuration",
1758 712 : "lastN",
1759 712 : "limit",
1760 712 : "offset",
1761 712 : "count",
1762 712 : "options",
1763 712 : "format",
1764 712 : "lang",
1765 712 : "local",
1766 712 : "entityMap",
1767 712 : "pick",
1768 712 : "omit",
1769 712 : "datasetId",
1770 712 : "orderBy",
1771 712 : "orderFrom",
1772 712 : "orderGeometry",
1773 712 : "collation",
1774 712 : "entityMapLifetime",
1775 712 : "splitEntities",
1776 712 : "expandValues",
1777 712 : "jsonKeys",
1778 712 : ],
1779 4 : )?;
1780 708 : let accept = parse_accept(headers)?;
1781 700 : let ctx = request_context(&st.loader, headers).await?;
1782 : // 5.7.4.4 a-e: id/idPattern alone are NOT sufficient, and the attrs
1783 : // list / q must include at least one non-system Attribute to qualify.
1784 : // 5.7.4.3 expandValues: the same 4.9 EXAMPLE 12 coercion as the entity
1785 : // query — term values expanded against the @context before executing,
1786 : // less the Attributes jsonKeys declares uninterpretable.
1787 628 : let q_ast = params.get("q").map(|q| parse_q(q)).transpose()?.map(|ast| {
1788 220 : antares_ql::eval::apply_expand_values(
1789 220 : ast,
1790 220 : antares_ql::eval::expansion_list(
1791 220 : params.get("expandValues").map(String::as_str),
1792 220 : params.get("jsonKeys").map(String::as_str),
1793 220 : )
1794 220 : .as_deref(),
1795 220 : &ctx,
1796 : )
1797 220 : });
1798 628 : check_temporal_query(st, params, headers, &tenant, &ctx, q_ast.as_ref()).await?;
1799 : // ADR-0020: the engine's narrowing joins the request's own filters after
1800 : // 5.7.4.4's "too wide" judgement, which is about what the client asked
1801 : // for. Everything below reads the narrowed query, the forward included.
1802 592 : let narrowed = filter.narrow_params(params)?;
1803 592 : let params = &narrowed;
1804 592 : let q_ast = params.get("q").map(|q| parse_q(q)).transpose()?.map(|ast| {
1805 212 : antares_ql::eval::apply_expand_values(
1806 212 : ast,
1807 212 : antares_ql::eval::expansion_list(
1808 212 : params.get("expandValues").map(String::as_str),
1809 212 : params.get("jsonKeys").map(String::as_str),
1810 212 : )
1811 212 : .as_deref(),
1812 212 : &ctx,
1813 : )
1814 212 : });
1815 592 : let scope_q = params.get("scopeQ").map(String::as_str);
1816 592 : let tq = TemporalQ::from_params(params, true)?;
1817 576 : let mut trepr = parse_trepr(params, &ctx)?;
1818 544 : crate::repr::narrow_projection(&mut trepr.pick, &mut trepr.omit, filter, &ctx)?;
1819 : // 5.7.4.4: {…} projection is Linked Entity retrieval — unconditional 400
1820 544 : reject_linked_projection(&trepr, "5.7.4.4")?;
1821 532 : let last_n = trepr.last_n;
1822 :
1823 532 : let ids: Option<Vec<&str>> = params.get("id").map(|s| s.split(',').collect());
1824 : // 5.7.4.4: an invalid URI in the id list is BadRequestData
1825 532 : if let Some(ids) = &ids {
1826 56 : for id in ids {
1827 56 : antares_model::EntityId::new(id)?;
1828 : }
1829 476 : }
1830 524 : let id_pattern = match params.get("idPattern") {
1831 0 : Some(p) => Some(
1832 0 : antares_ql::regex::compile(p)
1833 0 : .map_err(|_| NgsiError::BadRequestData(format!("invalid idPattern {p:?}")))?,
1834 : ),
1835 524 : None => None,
1836 : };
1837 524 : let types: Option<Vec<String>> = params.get("type").map(|s| {
1838 488 : s.split([',', '|'])
1839 488 : .map(|t| ctx.expand_key(t.trim()))
1840 488 : .collect()
1841 488 : });
1842 524 : let attrs_filter = selection(&trepr);
1843 : // only the attrs= param excludes entities; pick is projection-only
1844 524 : let entity_attr_filter = trepr.attrs.clone();
1845 524 : let geo = antares_ql::geo::GeoQuery::from_params(params)?;
1846 :
1847 : // Push entity narrowing (ids/types/attrs) and instance-window
1848 : // pruning (range + RANK()-capped lastN) into the store. The loop below
1849 : // and window() stay the arbiters — pruning is byte-exact against
1850 : // instance_matches (compile::temporal), so it cannot change an answer.
1851 : // 5.7.4.4 S2/S3: q and geo are judged on the instances WITHIN the
1852 : // temporal interval (the eval_doc retain below), so RANGE pruning is
1853 : // verdict-safe to push even with q=/geo present. Only the lastN cap
1854 : // (its ordering vs the values filter is unspecified) and entity paging
1855 : // (q/geo still drop entities after SQL) wait for exactness.
1856 : // scopeQ joins q/geo here: the 4.18 validity filter drops entities and
1857 : // instances AFTER SQL, so a pushed page/lastN cap would under-return.
1858 524 : let exact_push = q_ast.is_none() && geo.is_none() && scope_q.is_none();
1859 : // Entity-page pushdown: a temporal query used to materialize the
1860 : // tenant's ENTIRE history. Pushed only when every filter the store
1861 : // cannot see is absent — same gate family as the entity-query pushdown.
1862 : // A values filter no longer blocks paging when its prefilter compiles
1863 : // EXACTLY (every leaf a Cmp with the byte-exact text window): the SQL
1864 : // entity verdict then equals the evaluator's. datasetId/pick still
1865 : // block: their entity drops happen at presentation, after the page.
1866 524 : let q_page_exact = q_ast.as_ref().is_none_or(|ast| {
1867 212 : let r = tq.as_ref().map(|t| antares_store::filter::InstanceRange {
1868 212 : timerel: &t.timerel,
1869 212 : time_at: &t.time_at,
1870 212 : end_time_at: t.end_time_at.as_deref(),
1871 212 : timeproperty: &t.timeproperty,
1872 212 : });
1873 212 : st.temporal
1874 212 : .q_pushdown_exact(ast, r.as_ref(), &|t| ctx.expand_key(t))
1875 212 : });
1876 524 : let (p_offset, p_limit, _) = crate::paging::page_params(st, params)?;
1877 : // 5.7.4.4 + 5.5.9: pagination applies to the MERGED federated union, so
1878 : // the store may only pre-page when nothing will federate — otherwise
1879 : // page 1 is local-page + every remote row (matrix-9 IOP_EXT_TMP_03_04).
1880 524 : let push_page = (exact_push || (geo.is_none() && scope_q.is_none() && q_page_exact))
1881 254 : && id_pattern.is_none()
1882 254 : && params.get("orderBy").is_none()
1883 248 : && params.get("datasetId").is_none()
1884 248 : && params.get("pick").is_none()
1885 246 : && p_limit > 0
1886 246 : && !crate::federation::would_federate(st, &tenant, &ctx, params, headers).await?;
1887 : // 4.5.19 computed by the store: the numeric bucket matrix per attribute
1888 : // comes back aggregated when nothing after the store call could change
1889 : // the answer — every filter exact in SQL, the page pushed, no
1890 : // projection/lastN/month periods — otherwise the instances are
1891 : // aggregated here as before. A store that cannot (memory) or a
1892 : // non-numeric value class leaves `outcome.aggregated` false.
1893 524 : let push_agg = trepr.aggregated
1894 26 : && exact_push
1895 26 : && push_page
1896 26 : && trepr.omit.is_none()
1897 26 : && last_n.is_none()
1898 26 : && !matches!(trepr.aggr_period, AggrPeriod::Months(..))
1899 26 : && params.get("entityMap").map(String::as_str) != Some("true")
1900 26 : && trepr
1901 26 : .aggr_methods
1902 26 : .iter()
1903 48 : .all(|m| antares_store::filter::AGGREGATE_METHODS.contains(&m.as_str()));
1904 : // scoped: the &dyn expander must not live across an await (handler
1905 : // futures are Send; the store call itself is synchronous)
1906 522 : let outcome = {
1907 524 : let expand = |t: &str| ctx.expand_key(t);
1908 524 : let geo_pre = geo.as_ref().map(|g| g.to_instance_spec(&ctx));
1909 524 : let tf = antares_store::filter::TemporalFilter {
1910 524 : ids: ids.as_deref(),
1911 524 : types: types.as_deref(),
1912 524 : attrs: entity_attr_filter.as_deref(),
1913 524 : range: tq.as_ref().map(|t| antares_store::filter::InstanceRange {
1914 524 : timerel: &t.timerel,
1915 524 : time_at: &t.time_at,
1916 524 : end_time_at: t.end_time_at.as_deref(),
1917 524 : timeproperty: &t.timeproperty,
1918 524 : }),
1919 524 : last_n: match (last_n, exact_push) {
1920 8 : (Some(n), true) => Some(n as i64),
1921 516 : _ => None,
1922 : },
1923 524 : timeproperty: tq
1924 524 : .as_ref()
1925 524 : .map_or("observedAt", |t| t.timeproperty.as_str()),
1926 524 : page: push_page.then_some(antares_store::filter::Page {
1927 524 : offset: p_offset as i64,
1928 524 : limit: p_limit as i64,
1929 524 : count: true,
1930 524 : }),
1931 524 : q: q_ast.as_ref(),
1932 524 : expand: &expand,
1933 524 : geo: geo_pre.as_ref().map(|(s, iri)| (s, iri.as_str())),
1934 524 : aggregate: push_agg.then_some(antares_store::filter::Aggregate {
1935 524 : methods: &trepr.aggr_methods,
1936 524 : period_secs: match trepr.aggr_period {
1937 4 : AggrPeriod::Seconds(sc) => Some(sc),
1938 520 : _ => None,
1939 : },
1940 524 : anchor: tq.as_ref().map(|t| t.time_at.as_str()),
1941 : }),
1942 : };
1943 524 : st.temporal.query_temporal(&tenant, &tf).await?
1944 : };
1945 522 : if outcome.aggregated {
1946 6 : return respond_aggregated(
1947 6 : st,
1948 6 : params,
1949 6 : &tenant,
1950 6 : &ctx,
1951 6 : accept,
1952 6 : &trepr,
1953 6 : tq.as_ref(),
1954 6 : attrs_filter.as_ref(),
1955 6 : outcome,
1956 : );
1957 516 : }
1958 516 : let (all, pre_paged, pre_total) = (outcome.rows, outcome.paged, outcome.total);
1959 : // 5.7.4.4: fan the query out to matching queryTemporal registrations
1960 : // and merge the remote Temporal Evolutions with the local set (4.5.5;
1961 : // auxiliary data never introduces new entities)
1962 516 : let mut warnings: Vec<String> = Vec::new();
1963 516 : let looped = crate::federation::via_loop(
1964 516 : headers,
1965 516 : &crate::federation::alias_for(&st.host_alias, &tenant),
1966 : );
1967 516 : let timeprop = tq
1968 516 : .as_ref()
1969 516 : .map_or("observedAt", |t| t.timeproperty.as_str())
1970 516 : .to_owned();
1971 516 : let all = if crate::federation::active(params) && !looped {
1972 478 : let fed = crate::federation::fed_query_temporal(
1973 478 : st,
1974 478 : &tenant,
1975 478 : headers,
1976 478 : &ctx,
1977 478 : params,
1978 478 : &mut warnings,
1979 478 : )
1980 478 : .await?;
1981 478 : let mut order: Vec<String> = Vec::new();
1982 478 : let mut by_id: std::collections::HashMap<String, Value> = Default::default();
1983 1442 : for doc in all {
1984 1442 : if let Some(id) = doc.get("id").and_then(Value::as_str) {
1985 1442 : order.push(id.to_owned());
1986 1442 : by_id.insert(id.to_owned(), doc);
1987 1442 : }
1988 : }
1989 956 : for aux_pass in [false, true] {
1990 956 : for (aux, d) in &fed {
1991 20 : if *aux != aux_pass {
1992 10 : continue;
1993 10 : }
1994 10 : let Some(id) = d.get("id").and_then(Value::as_str) else {
1995 0 : continue;
1996 : };
1997 10 : match by_id.get_mut(id) {
1998 0 : Some(base) => merge_temporal_docs(base, d, *aux, &timeprop),
1999 10 : None if !aux => {
2000 10 : order.push(id.to_owned());
2001 10 : by_id.insert(id.to_owned(), d.clone());
2002 10 : }
2003 0 : None => {}
2004 : }
2005 : }
2006 : }
2007 478 : order
2008 478 : .into_iter()
2009 1452 : .filter_map(|id| by_id.remove(&id))
2010 478 : .collect()
2011 : } else {
2012 38 : all
2013 : };
2014 516 : let mut matches = Vec::new();
2015 1462 : for mut doc in all {
2016 1462 : let id = doc["id"].as_str().unwrap_or("");
2017 1462 : if let Some(ids) = &ids {
2018 80 : if !ids.contains(&id) {
2019 32 : continue;
2020 48 : }
2021 1382 : }
2022 : // 5.2.33: "id takes precedence over idPattern"
2023 1430 : if ids.is_none() {
2024 1382 : if let Some(re) = &id_pattern {
2025 0 : if !re.is_match(id) {
2026 0 : continue;
2027 0 : }
2028 1382 : }
2029 48 : }
2030 1430 : if !type_and_attrs_match(&doc, types.as_ref(), entity_attr_filter.as_ref()) {
2031 0 : continue;
2032 1430 : }
2033 1430 : if (q_ast.is_some() || geo.is_some())
2034 910 : && !window_conditions_match(&doc, tq.as_ref(), q_ast.as_ref(), geo.as_ref(), &ctx)
2035 : {
2036 664 : continue;
2037 766 : }
2038 766 : if let Some(sq) = scope_q {
2039 368 : if !scope_window_retain(&mut doc, sq, tq.as_ref(), timeprop.as_str()) {
2040 272 : continue;
2041 96 : }
2042 398 : }
2043 : // entity qualifies only if some instance falls in the window
2044 494 : let any_instance = doc.as_object().is_some_and(|o| {
2045 1020 : o.iter().any(|(k, v)| {
2046 1020 : !is_meta(k)
2047 494 : && v.as_array().is_some_and(|arr| {
2048 494 : arr.iter()
2049 498 : .any(|inst| tq.as_ref().is_none_or(|tq| tq.instance_matches(inst)))
2050 494 : })
2051 1020 : })
2052 494 : });
2053 494 : if !any_instance {
2054 14 : continue;
2055 480 : }
2056 480 : matches.push(doc);
2057 : }
2058 516 : if let Some(spec) = params.get("orderBy") {
2059 6 : crate::paging::order_entities(&mut matches, spec, params, &ctx)?;
2060 510 : }
2061 516 : let (page, count_hdr, links) = if pre_paged {
2062 90 : let total = pre_total.map(|t| t as usize).unwrap_or(matches.len());
2063 90 : crate::paging::paginate_pre(st, params, matches, "/ngsi-ld/v1/temporal/entities", total)?
2064 : } else {
2065 426 : crate::paging::paginate(st, params, matches, "/ngsi-ld/v1/temporal/entities")?
2066 : };
2067 516 : if let Some(sink) = merged {
2068 12 : sink.extend(page.iter().cloned());
2069 504 : }
2070 516 : let core_only_pick = attrs_filter.as_ref().is_some_and(Vec::is_empty);
2071 516 : let mut payload: Vec<Value> = Vec::new();
2072 516 : let (mut g_trunc, mut g_min, mut g_maxts) = (false, None::<String>, None::<String>);
2073 516 : for d in &page {
2074 458 : let mut w = window(
2075 458 : d,
2076 458 : tq.as_ref(),
2077 458 : last_n,
2078 458 : attrs_filter.as_ref(),
2079 458 : trepr.omit.as_ref(),
2080 458 : trepr.dataset_id.as_ref(),
2081 458 : &timeprop,
2082 : );
2083 458 : if !trepr.aggregated {
2084 458 : truncate(&mut w, &timeprop, last_n.is_some());
2085 458 : }
2086 458 : g_trunc |= w.truncated;
2087 458 : if let Some(m) = &w.ts_min {
2088 458 : if g_min.as_deref().is_none_or(|c| dt_key(m) < dt_key(c)) {
2089 356 : g_min = Some(m.clone());
2090 356 : }
2091 0 : }
2092 458 : if let Some(m) = &w.ts_max {
2093 458 : if g_maxts.as_deref().is_none_or(|c| dt_key(m) > dt_key(c)) {
2094 376 : g_maxts = Some(m.clone());
2095 376 : }
2096 0 : }
2097 : // no instance survived the window/dataset filters ⇒ the entity is
2098 : // not part of the temporal result (unless the projection is
2099 : // deliberately core-only, e.g. pick=id)
2100 458 : if w.attrs.is_empty() && !core_only_pick {
2101 0 : continue;
2102 458 : }
2103 458 : let presented = present_temporal(d, &w, &ctx, &trepr, tq.as_ref(), &timeprop)?;
2104 458 : if trepr.pick.is_some() && presented.as_object().is_some_and(|o| o.is_empty()) {
2105 0 : continue;
2106 458 : }
2107 458 : payload.push(presented);
2108 : }
2109 : // aggregated responses are complete by construction — never 206 (6.3.10)
2110 516 : let cr = if trepr.aggregated {
2111 20 : None
2112 : } else {
2113 496 : content_range(
2114 496 : g_trunc,
2115 496 : g_min.as_deref(),
2116 496 : g_maxts.as_deref(),
2117 496 : tq.as_ref(),
2118 496 : last_n,
2119 : )
2120 : };
2121 516 : let status = if cr.is_some() {
2122 2 : StatusCode::PARTIAL_CONTENT
2123 : } else {
2124 514 : StatusCode::OK
2125 : };
2126 516 : let mut resp = crate::negotiate::respond_list(status, payload, &ctx, accept, &tenant);
2127 516 : crate::paging::attach_warnings(&mut resp, &warnings);
2128 516 : if let Some(cr) = cr {
2129 2 : if let Ok(v) = cr.parse() {
2130 2 : resp.headers_mut().insert("Content-Range", v);
2131 2 : }
2132 514 : }
2133 516 : attach_paging(&mut resp, count_hdr, &links);
2134 : // 6.18.3.2: entityMap=true — the temporal EntityMap for this query is
2135 : // (re)created; the response carries NGSILD-EntityMap and 201 Created.
2136 516 : if params.get("entityMap").map(String::as_str) == Some("true") {
2137 12 : let map = build_temporal_map(st, &tenant, headers, &ctx, params, filter).await?;
2138 12 : *resp.status_mut() = StatusCode::CREATED;
2139 12 : if let Some(id) = map.get("id").and_then(Value::as_str) {
2140 12 : if let Ok(v) = format!("/ngsi-ld/v1/entityMaps/{id}").parse() {
2141 12 : resp.headers_mut().insert("NGSILD-EntityMap", v);
2142 12 : }
2143 0 : }
2144 504 : }
2145 516 : filter.mark_restricted(resp.headers_mut());
2146 516 : Ok(resp)
2147 712 : }
2148 :
2149 : // ---------- GET /temporal/entities/{id} (5.7.3) ----------
2150 :
2151 284 : pub async fn retrieve_temporal(
2152 284 : State(st): State<AppState>,
2153 284 : Path(id): Path<String>,
2154 284 : CleanParams(params): CleanParams,
2155 284 : headers: HeaderMap,
2156 284 : ) -> Response {
2157 284 : match retrieve_temporal_outer(&st, &id, ¶ms, &headers).await {
2158 204 : Ok(r) => r,
2159 80 : Err(e) => e.into_response(),
2160 : }
2161 284 : }
2162 :
2163 : /// 5.7.3.4 Retrieve Temporal Evolution: the EntityMap half of the clause is
2164 : /// the shared rule (`entity_maps::retrieve_with_map`); this is the retrieve
2165 : /// it wraps.
2166 284 : async fn retrieve_temporal_outer(
2167 284 : st: &AppState,
2168 284 : id: &str,
2169 284 : params: &HashMap<String, String>,
2170 284 : headers: &HeaderMap,
2171 284 : ) -> ApiResult<Response> {
2172 284 : crate::entity_map::retrieve_with_map(st, id, params, headers, true, |map| async move {
2173 284 : retrieve_temporal_inner(st, id, params, headers, map.as_ref()).await
2174 568 : })
2175 284 : .await
2176 284 : }
2177 :
2178 284 : async fn retrieve_temporal_inner(
2179 284 : st: &AppState,
2180 284 : id: &str,
2181 284 : params: &HashMap<String, String>,
2182 284 : headers: &HeaderMap,
2183 284 : map: Option<&Value>,
2184 284 : ) -> ApiResult<Response> {
2185 : {
2186 284 : let tenant = tenant_from(headers)?;
2187 284 : check_params(
2188 284 : params,
2189 284 : &[
2190 284 : "attrs",
2191 284 : "timerel",
2192 284 : "timeAt",
2193 284 : "endTimeAt",
2194 284 : "timeproperty",
2195 284 : "lastN",
2196 284 : "aggrMethods",
2197 284 : "aggrPeriodDuration",
2198 284 : "options",
2199 284 : "format",
2200 284 : "lang",
2201 284 : "local",
2202 284 : "pick",
2203 284 : "omit",
2204 284 : "datasetId",
2205 284 : "entityMap",
2206 284 : "entityMapLifetime",
2207 284 : ],
2208 0 : )?;
2209 284 : let accept = parse_accept(headers)?;
2210 284 : let ctx = request_context(&st.loader, headers).await?;
2211 268 : let filter = gate!(st, &tenant, headers, "5.7.3", ids: &[id]).await?;
2212 268 : let tq = TemporalQ::from_params(params, false)?;
2213 268 : let mut trepr = parse_trepr(params, &ctx)?;
2214 264 : crate::repr::narrow_projection(&mut trepr.pick, &mut trepr.omit, &filter, &ctx)?;
2215 : // 5.7.3.4: "If projection attributes are present and indicate the
2216 : // use of Linked Entity retrieval, an error of type BadRequestData
2217 : // shall be raised" — unconditional, temporal defines no join.
2218 264 : reject_linked_projection(&trepr, "5.7.3.4")?;
2219 252 : let last_n = trepr.last_n;
2220 252 : antares_model::EntityId::new(id)?;
2221 : // Instance pruning pushed into the store (no q=/geo on retrieve,
2222 : // so it is always safe here); window() below stays the arbiter.
2223 248 : let tf = antares_store::filter::TemporalFilter {
2224 248 : range: tq.as_ref().map(|t| antares_store::filter::InstanceRange {
2225 116 : timerel: &t.timerel,
2226 116 : time_at: &t.time_at,
2227 116 : end_time_at: t.end_time_at.as_deref(),
2228 116 : timeproperty: &t.timeproperty,
2229 116 : }),
2230 248 : last_n: last_n.map(|n| n as i64),
2231 248 : timeproperty: tq
2232 248 : .as_ref()
2233 248 : .map_or("observedAt", |t| t.timeproperty.as_str()),
2234 248 : ..Default::default()
2235 : };
2236 248 : let timeprop = tq
2237 248 : .as_ref()
2238 248 : .map_or("observedAt", |t| t.timeproperty.as_str())
2239 248 : .to_owned();
2240 248 : let local = st.temporal.get_temporal(&tenant, id, &tf).await?;
2241 : // 5.7.3.4: forward to matching retrieveTemporal registrations and
2242 : // merge the remote instance data (4.5.5; auxiliary instances only
2243 : // fill timestamps absent elsewhere)
2244 246 : let mut warnings: Vec<String> = Vec::new();
2245 246 : let looped = crate::federation::via_loop(
2246 246 : headers,
2247 246 : &crate::federation::alias_for(&st.host_alias, &tenant),
2248 : );
2249 246 : let doc = if crate::federation::active(params) && !looped {
2250 246 : let fed = crate::federation::fed_retrieve_temporal(
2251 246 : st,
2252 246 : &tenant,
2253 246 : headers,
2254 246 : &ctx,
2255 246 : id,
2256 246 : params,
2257 246 : map,
2258 246 : &mut warnings,
2259 246 : )
2260 246 : .await?;
2261 246 : let (mut base, skip) = match local {
2262 210 : Some(b) => (b, None),
2263 : None => {
2264 36 : let idx = fed.iter().position(|(aux, _)| !aux).or(if fed.is_empty() {
2265 34 : None
2266 : } else {
2267 2 : Some(0)
2268 : });
2269 36 : match idx {
2270 2 : Some(i) => (fed[i].1.clone(), Some(i)),
2271 : None => {
2272 34 : return Err(NgsiError::ResourceNotFound(format!(
2273 34 : "temporal entity {id} not found"
2274 34 : ))
2275 34 : .into())
2276 : }
2277 : }
2278 : }
2279 : };
2280 424 : for aux_pass in [false, true] {
2281 424 : for (i, (aux, d)) in fed.iter().enumerate() {
2282 4 : if Some(i) == skip {
2283 4 : continue;
2284 0 : }
2285 0 : if *aux == aux_pass {
2286 0 : merge_temporal_docs(&mut base, d, *aux, &timeprop);
2287 0 : }
2288 : }
2289 : }
2290 212 : base
2291 : } else {
2292 0 : local.ok_or_else(|| {
2293 0 : NgsiError::ResourceNotFound(format!("temporal entity {id} not found"))
2294 0 : })?
2295 : };
2296 212 : let attrs_filter = selection(&trepr);
2297 : // 5.7.3: attrs matching nothing ⇒ 404. A `pick` may name core members
2298 : // instead — 5.7.3.3 admits "id", "type", "scope" or an Attribute name
2299 : // — and then it selects no Attribute at all. 5.7.3.5 still reduces the
2300 : // Entity to the members it names, so that empty selection is the
2301 : // answer rather than a Temporal Evolution holding none of it.
2302 212 : let core_only_pick = trepr.attrs.is_none()
2303 202 : && trepr.pick.is_some()
2304 10 : && attrs_filter.as_ref().is_some_and(Vec::is_empty);
2305 212 : if let Some(want) = &attrs_filter {
2306 20 : if !core_only_pick && !want.iter().any(|a| doc.get(a).is_some()) {
2307 2 : return Err(NgsiError::ResourceNotFound(format!(
2308 2 : "temporal entity {id} has none of the requested attributes"
2309 2 : ))
2310 2 : .into());
2311 18 : }
2312 192 : }
2313 210 : let mut w = window(
2314 210 : &doc,
2315 210 : tq.as_ref(),
2316 210 : last_n,
2317 210 : attrs_filter.as_ref(),
2318 210 : trepr.omit.as_ref(),
2319 210 : trepr.dataset_id.as_ref(),
2320 210 : &timeprop,
2321 : );
2322 210 : if !trepr.aggregated {
2323 192 : truncate(&mut w, &timeprop, last_n.is_some());
2324 192 : }
2325 210 : let cr = if trepr.aggregated {
2326 18 : None
2327 : } else {
2328 192 : content_range(
2329 192 : w.truncated,
2330 192 : w.ts_min.as_deref(),
2331 192 : w.ts_max.as_deref(),
2332 192 : tq.as_ref(),
2333 192 : last_n,
2334 : )
2335 : };
2336 210 : let payload = present_temporal(&doc, &w, &ctx, &trepr, tq.as_ref(), &timeprop)?;
2337 204 : if (trepr.pick.is_some() || trepr.omit.is_some())
2338 10 : && payload.as_object().is_some_and(|o| o.is_empty())
2339 : {
2340 0 : return Err(NgsiError::ResourceNotFound(format!(
2341 0 : "projection matches nothing on temporal entity {id}"
2342 0 : ))
2343 0 : .into());
2344 204 : }
2345 204 : let status = if cr.is_some() {
2346 0 : StatusCode::PARTIAL_CONTENT
2347 : } else {
2348 204 : StatusCode::OK
2349 : };
2350 204 : let mut resp = respond(status, payload, &ctx, accept, &tenant);
2351 204 : if let Some(cr) = cr {
2352 0 : if let Ok(v) = cr.parse() {
2353 0 : resp.headers_mut().insert("Content-Range", v);
2354 0 : }
2355 204 : }
2356 204 : crate::paging::attach_warnings(&mut resp, &warnings);
2357 204 : filter.mark_restricted(resp.headers_mut());
2358 204 : Ok(resp)
2359 : }
2360 284 : }
2361 :
2362 : /// 5.7.3.4 / 4.5.5: merge one remote Temporal Evolution into `base` by
2363 : /// appending instances per Attribute; auxiliary data contributes an
2364 : /// instance only when no instance with the same timeproperty value was
2365 : /// received from elsewhere.
2366 20 : fn merge_temporal_docs(base: &mut Value, add: &Value, aux: bool, timeprop: &str) {
2367 20 : let (Some(target), Some(source)) = (base.as_object_mut(), add.as_object()) else {
2368 0 : return;
2369 : };
2370 60 : for (k, v) in source {
2371 60 : if [
2372 60 : "id",
2373 60 : "type",
2374 60 : "scope",
2375 60 : "createdAt",
2376 60 : "modifiedAt",
2377 60 : "deletedAt",
2378 60 : "expiresAt",
2379 60 : ]
2380 60 : .contains(&k.as_str())
2381 : {
2382 40 : target.entry(k.clone()).or_insert_with(|| v.clone());
2383 40 : continue;
2384 20 : }
2385 20 : let incoming: Vec<Value> = match v {
2386 20 : Value::Array(x) => x.clone(),
2387 0 : other => vec![other.clone()],
2388 : };
2389 20 : match target.get_mut(k).and_then(Value::as_array_mut) {
2390 0 : None => {
2391 0 : target.insert(k.clone(), Value::Array(incoming));
2392 0 : }
2393 20 : Some(cur) => {
2394 32 : for ni in incoming {
2395 32 : if aux {
2396 12 : let ts = ni.get(timeprop).and_then(Value::as_str).map(dt_key);
2397 12 : if ts.is_some()
2398 12 : && cur.iter().any(|ci| {
2399 12 : ci.get(timeprop).and_then(Value::as_str).map(dt_key) == ts
2400 12 : })
2401 : {
2402 8 : continue;
2403 4 : }
2404 20 : }
2405 : // 4.5.5.3: an instance with the same datasetId (or both
2406 : // default) AND the same timeproperty value is a
2407 : // CONFLICTING instance of one slot — resolve to one, the
2408 : // most recent modifiedAt winning.
2409 : // Two instances only share a slot when they carry the SAME
2410 : // timeproperty value. Without the is_some() guard a pair of
2411 : // instances that both lack it would compare None == None,
2412 : // letting a remote instance replace unrelated local history.
2413 : // 4.6.3 leaves the seconds fraction optional, so two
2414 : // Context Sources may spell one instant differently: the
2415 : // slot and its winner are decided on the canonical key,
2416 : // or the same instance comes back twice.
2417 24 : let ni_ts = ni.get(timeprop).and_then(Value::as_str).map(dt_key);
2418 24 : let slot = ni_ts.and_then(|ts| {
2419 28 : cur.iter_mut().find(|ci| {
2420 28 : ci.get(timeprop).and_then(Value::as_str).map(dt_key) == Some(ts.clone())
2421 16 : && ci.get("datasetId") == ni.get("datasetId")
2422 28 : })
2423 24 : });
2424 24 : if let Some(existing) = slot {
2425 24 : let stamp = |i: &Value| {
2426 24 : dt_key(i.get("modifiedAt").and_then(Value::as_str).unwrap_or(""))
2427 24 : };
2428 12 : let newer = stamp(&ni) > stamp(existing);
2429 12 : if newer {
2430 12 : *existing = ni;
2431 12 : }
2432 12 : continue;
2433 12 : }
2434 12 : cur.push(ni);
2435 : }
2436 : }
2437 : }
2438 : }
2439 20 : }
2440 :
2441 : // ---------- DELETE /temporal/entities/{id} (5.6.16) ----------
2442 :
2443 2196 : pub async fn delete_temporal(
2444 2196 : State(st): State<AppState>,
2445 2196 : Path(id): Path<String>,
2446 2196 : CleanParams(params): CleanParams,
2447 2196 : headers: HeaderMap,
2448 2196 : ) -> Response {
2449 2196 : let go = async {
2450 2196 : let tenant = tenant_from(&headers)?;
2451 2196 : antares_model::EntityId::new(&id)?;
2452 2184 : check_params(¶ms, &["local"])?;
2453 : // 5.6.16.4: forward to registrations supporting the operation;
2454 : // unsupported proxy modes are Conflict.
2455 2184 : let ctx = st.loader.core();
2456 2184 : gate!(st, &tenant, &headers, "5.6.16", ids: &[&id]).await?;
2457 2184 : let regs = match temporal_write_regs(&st, &tenant, &headers, &ctx, ¶ms, &id).await {
2458 2182 : Ok(regs) => regs,
2459 2 : Err(refused) => return Ok(*refused),
2460 : };
2461 2182 : let deleted = st.temporal.delete(&tenant, &id).await?;
2462 2182 : answer_temporal_attr_write(
2463 2182 : &st,
2464 2182 : &tenant,
2465 2182 : &headers,
2466 2182 : &ctx,
2467 2182 : &id,
2468 2182 : "deleteTemporal",
2469 2182 : reqwest::Method::DELETE,
2470 2182 : "",
2471 2182 : None,
2472 2182 : regs,
2473 2182 : LocalWrite {
2474 2182 : res: deleted.then_some(Ok(())),
2475 2182 : found: deleted,
2476 2182 : missing: format!("temporal entity {id}"),
2477 2182 : applied: "deleted locally",
2478 2182 : },
2479 2182 : )
2480 2182 : .await
2481 2196 : };
2482 2196 : go.await.unwrap_or_else(|e| e.into_response())
2483 2196 : }
2484 :
2485 : // ---------- POST /temporal/entities/{id}/attrs/ (5.6.12) ----------
2486 :
2487 16 : pub async fn add_temporal_attrs(
2488 16 : State(st): State<AppState>,
2489 16 : Path(id): Path<String>,
2490 16 : CleanParams(params): CleanParams,
2491 16 : headers: HeaderMap,
2492 16 : body: Bytes,
2493 16 : ) -> Response {
2494 16 : let go = async {
2495 16 : let tenant = tenant_from(&headers)?;
2496 16 : antares_model::EntityId::new(&id)?;
2497 16 : check_params(¶ms, &["local"])?;
2498 16 : let parsed = parse_body(&st.loader, &headers, &body, BodyKind::Standard).await?;
2499 12 : let obj = parsed.object(NgsiError::BadRequestData(
2500 12 : "fragment must be a JSON object".into(),
2501 12 : ))?;
2502 12 : gate!(st, &tenant, &headers, "5.6.12", ids: &[&id]).await?;
2503 : // 5.6.12 input is pushed history — the 4.5.7 deleted-instance
2504 : // representation is legal (5.5.4 temporal exception), hence
2505 : // allow_null (mirrors 5.6.11).
2506 12 : let mut expanded = expand_entity(
2507 12 : obj,
2508 12 : &parsed.ctx,
2509 12 : ExpandOpts {
2510 12 : fragment: true,
2511 12 : allow_null: true,
2512 12 : temporal: true,
2513 12 : ..Default::default()
2514 12 : },
2515 0 : )?;
2516 : // 5.6.12.4: forwarding — proxy modes without appendAttrsTemporal
2517 : // are Conflict; supporting registrations receive the fragment and
2518 : // the matching attributes are stripped from the local half.
2519 12 : let spec = crate::registry::CsrSpec {
2520 12 : ids: Some(vec![id.clone()]),
2521 12 : ..Default::default()
2522 12 : };
2523 12 : let regs = match crate::federation::write_plan(
2524 12 : &st,
2525 12 : &tenant,
2526 12 : &spec,
2527 12 : &parsed.ctx,
2528 12 : ¶ms,
2529 12 : &headers,
2530 12 : )
2531 12 : .await?
2532 : {
2533 0 : crate::federation::WritePlan::Answered(r) => return Ok(*r),
2534 12 : crate::federation::WritePlan::Forward(regs) => regs,
2535 : };
2536 12 : if !regs.is_empty() {
2537 4 : let mut parts = Vec::new();
2538 4 : let mut fwd = Vec::new();
2539 4 : for reg in ®s {
2540 4 : if !reg.supports("appendAttrsTemporal") {
2541 2 : if reg.is_proxy() {
2542 2 : parts.push(crate::federation::conflict_part("appendAttrsTemporal"));
2543 2 : }
2544 2 : continue;
2545 2 : }
2546 2 : if let Some(frag) = crate::federation::reduce_to_scope(obj, reg, &parsed.ctx) {
2547 2 : fwd.push((reg.clone(), frag));
2548 2 : }
2549 : }
2550 4 : let proxies: Vec<&crate::federation::FedReg> =
2551 4 : regs.iter().filter(|r| r.is_proxy()).collect();
2552 4 : let (rest, has_attrs) = crate::federation::strip_proxied(obj, &proxies, &parsed.ctx);
2553 4 : if has_attrs || proxies.is_empty() {
2554 0 : let mut local = expand_entity(
2555 0 : &rest,
2556 0 : &parsed.ctx,
2557 0 : ExpandOpts {
2558 0 : fragment: true,
2559 0 : allow_null: true,
2560 0 : temporal: true,
2561 0 : ..Default::default()
2562 0 : },
2563 0 : )?;
2564 0 : let ts = now_iso();
2565 0 : stamp_temporal_instances(&mut local, &ts);
2566 0 : let res = st
2567 0 : .temporal
2568 0 : .mutate(&tenant, &id, |doc| {
2569 0 : add_temporal_instances(doc, &local, &ts);
2570 0 : Ok::<(), NgsiError>(())
2571 0 : })
2572 0 : .await?;
2573 0 : parts.push(match res {
2574 0 : Some(Ok(())) => crate::federation::Part {
2575 0 : status: 204,
2576 0 : detail: "added locally".into(),
2577 0 : },
2578 0 : _ => crate::federation::Part {
2579 0 : status: 404,
2580 0 : detail: format!("temporal entity {id} not found locally"),
2581 0 : },
2582 : });
2583 4 : }
2584 4 : let ctx_url = crate::federation::ctx_link_url(&headers, &parsed.ctx.source);
2585 4 : for (reg, frag) in fwd {
2586 2 : parts.push(
2587 2 : crate::federation::forward_part(
2588 2 : &st,
2589 2 : reqwest::Method::POST,
2590 2 : format!(
2591 2 : "{}/ngsi-ld/v1/temporal/entities/{}/attrs",
2592 2 : reg.endpoint,
2593 2 : crate::federation::path_segment(&id)
2594 2 : ),
2595 2 : &[],
2596 2 : &headers,
2597 2 : &tenant,
2598 2 : ®,
2599 2 : &ctx_url,
2600 2 : Some(frag),
2601 2 : )
2602 2 : .await,
2603 : );
2604 : }
2605 4 : return Ok(crate::federation::combine(
2606 4 : parts,
2607 4 : no_content(&tenant),
2608 4 : &tenant,
2609 4 : ));
2610 8 : }
2611 8 : let ts = now_iso();
2612 8 : stamp_temporal_instances(&mut expanded, &ts);
2613 8 : let res = st
2614 8 : .temporal
2615 8 : .mutate(&tenant, &id, |doc| {
2616 0 : add_temporal_instances(doc, &expanded, &ts);
2617 0 : Ok::<(), NgsiError>(())
2618 0 : })
2619 8 : .await?;
2620 0 : match res {
2621 : None => {
2622 8 : Err(NgsiError::ResourceNotFound(format!("temporal entity {id} not found")).into())
2623 : }
2624 0 : Some(Err(e)) => Err(ApiError::from(e)),
2625 0 : Some(Ok(())) => Ok(no_content(&tenant)),
2626 : }
2627 16 : };
2628 16 : go.await.unwrap_or_else(|e| e.into_response())
2629 16 : }
2630 :
2631 : /// 5.6.12.4: append the fragment's Attribute instances to the Temporal
2632 : /// Evolution (instances accumulate — history is never overwritten here).
2633 0 : fn add_temporal_instances(doc: &mut Value, expanded: &Value, ts: &str) {
2634 0 : let Some(target) = doc.as_object_mut() else {
2635 0 : return;
2636 : };
2637 0 : for (k, v) in expanded.as_object().into_iter().flatten() {
2638 0 : if is_meta(k) {
2639 0 : continue;
2640 0 : }
2641 0 : let incoming = v.as_array().cloned().unwrap_or_default();
2642 0 : match target.get_mut(k).and_then(Value::as_array_mut) {
2643 0 : Some(cur) => cur.extend(incoming),
2644 0 : None => {
2645 0 : target.insert(k.clone(), Value::Array(incoming));
2646 0 : }
2647 : }
2648 : }
2649 0 : target.insert("modifiedAt".into(), Value::String(ts.to_owned()));
2650 0 : }
2651 :
2652 : // ---------- DELETE /temporal/entities/{id}/attrs/{attrId} (5.6.13) ----------
2653 :
2654 48 : pub async fn delete_temporal_attr(
2655 48 : State(st): State<AppState>,
2656 48 : Path((id, attr)): Path<(String, String)>,
2657 48 : CleanParams(params): CleanParams,
2658 48 : headers: HeaderMap,
2659 48 : ) -> Response {
2660 48 : let go = async {
2661 48 : let tenant = tenant_from(&headers)?;
2662 48 : antares_model::EntityId::new(&id)?;
2663 : // 5.6.13.4: "If the target Attribute name is not a valid name, then an
2664 : // error of type BadRequestData shall be raised." The shared guard is
2665 : // the one that also refuses dot segments — the name is interpolated
2666 : // into the forwarded request path, where `..` addresses the peer's
2667 : // Temporal Evolution resource instead of its attribute.
2668 48 : antares_model::check_attr_name(&attr)?;
2669 32 : check_params(¶ms, &["datasetId", "deleteAll", "local"])?;
2670 32 : let ctx = request_context(&st.loader, &headers).await?;
2671 32 : gate!(st, &tenant, &headers, "5.6.13", ids: &[&id]).await?;
2672 32 : let attr_iri = antares_jsonld::expand_attr_name(&attr, &ctx)?;
2673 28 : let delete_all = params.get("deleteAll").map(String::as_str) == Some("true");
2674 28 : let want_ds = crate::repr::target_dataset_id(¶ms).map(String::from);
2675 28 : let regs = match temporal_write_regs(&st, &tenant, &headers, &ctx, ¶ms, &id).await {
2676 26 : Ok(regs) => regs,
2677 2 : Err(refused) => return Ok(*refused),
2678 : };
2679 26 : let mut found = false;
2680 26 : let ts = now_iso();
2681 26 : let res = st
2682 26 : .temporal
2683 26 : .mutate(&tenant, &id, |doc| {
2684 6 : let target = antares_store::stored_object(doc)?;
2685 6 : if delete_all
2686 6 : || (want_ds.is_none()
2687 6 : && !target
2688 6 : .get(&attr_iri)
2689 6 : .and_then(Value::as_array)
2690 6 : .is_some_and(|a| a.iter().any(|i| i.get("datasetId").is_some())))
2691 : {
2692 : // deleteAll, or single-instance-set attribute: drop it whole
2693 4 : if target.remove(&attr_iri).is_some() {
2694 2 : found = true;
2695 2 : }
2696 2 : } else if let Some(arr) = target.get_mut(&attr_iri).and_then(Value::as_array_mut) {
2697 : // 5.6.13: only the matching datasetId instance set is deleted
2698 2 : let before = arr.len();
2699 4 : arr.retain(|i| {
2700 4 : i.get("datasetId").and_then(Value::as_str) != want_ds.as_deref()
2701 4 : });
2702 2 : found = arr.len() != before;
2703 2 : if arr.is_empty() {
2704 0 : target.remove(&attr_iri);
2705 2 : }
2706 0 : }
2707 6 : if found {
2708 4 : target.insert("modifiedAt".into(), Value::String(ts.clone()));
2709 4 : }
2710 6 : Ok::<(), NgsiError>(())
2711 6 : })
2712 26 : .await?;
2713 26 : answer_temporal_attr_write(
2714 26 : &st,
2715 26 : &tenant,
2716 26 : &headers,
2717 26 : &ctx,
2718 26 : &id,
2719 26 : "deleteAttrsTemporal",
2720 26 : reqwest::Method::DELETE,
2721 26 : &format!("/attrs/{}", crate::federation::path_segment(&attr)),
2722 26 : None,
2723 26 : regs,
2724 26 : LocalWrite {
2725 26 : res,
2726 26 : found,
2727 26 : missing: format!("attribute {attr}"),
2728 26 : applied: "deleted locally",
2729 26 : },
2730 26 : )
2731 26 : .await
2732 48 : };
2733 48 : go.await.unwrap_or_else(|e| e.into_response())
2734 48 : }
2735 :
2736 : /// 5.6.13.4-5.6.15.4 shared forwarding: proxy registrations without the
2737 : /// operation's support are an error of type Conflict and are never
2738 : /// contacted; supporting registrations receive the forwarded request. None
2739 : /// = no matching registrations (the operation stays purely local).
2740 : /// `path_suffix` arrives with its client-controlled segments already
2741 : /// percent-encoded (RFC 3986 clause 3.3); the entity id is encoded here.
2742 : /// What the local `mutate` did, in the words the answer needs. 5.6.13,
2743 : /// 5.6.14, 5.6.15 and 5.6.16 all answer the same three ways: 204 when the
2744 : /// target was there, ResourceNotFound naming the Entity when the Entity was
2745 : /// not, and ResourceNotFound naming the Attribute or the instance when the
2746 : /// Entity was there but the target inside it was not.
2747 : struct LocalWrite {
2748 : /// `None` when the Temporal Evolution itself is absent.
2749 : res: Option<Result<(), NgsiError>>,
2750 : /// whether the operation's target inside it was there.
2751 : found: bool,
2752 : /// what a 404 names once the Entity itself was found.
2753 : missing: String,
2754 : /// what the 204 Part reports was done here.
2755 : applied: &'static str,
2756 : }
2757 :
2758 : /// 5.6.13 / 5.6.14 / 5.6.15 / 5.6.16 answer: the local result becomes one
2759 : /// 4.3.6 Part, every registration supporting `op` (Table 4.20-1) contributes
2760 : /// another, and 4.3.6.4 combines them into the one status the client sees.
2761 : /// With no registration to forward to — the common case — the local result
2762 : /// is the whole answer. The four operations differ in what they change and
2763 : /// in what they forward, never in how the two halves are answered together.
2764 : /// The registrations a 5.6.13/5.6.14/5.6.15/5.6.16 write forwards to, with
2765 : /// 6.3.17/6.3.18 loop handling already applied. The check belongs here,
2766 : /// ahead of the local write: 508 Loop Detected is an error status, so the
2767 : /// request it answers has to leave the Temporal Evolution as it found it.
2768 : /// The refusal travels boxed — a whole `Response` in an `Err` makes every
2769 : /// `Ok` of this function carry its width.
2770 2256 : async fn temporal_write_regs(
2771 2256 : st: &AppState,
2772 2256 : tenant: &antares_model::TenantId,
2773 2256 : headers: &HeaderMap,
2774 2256 : ctx: &antares_jsonld::Context,
2775 2256 : params: &HashMap<String, String>,
2776 2256 : id: &str,
2777 2256 : ) -> Result<Vec<crate::federation::FedReg>, Box<Response>> {
2778 2256 : let spec = crate::registry::CsrSpec {
2779 2256 : ids: Some(vec![id.to_owned()]),
2780 2256 : ..Default::default()
2781 2256 : };
2782 2256 : match crate::federation::write_plan(st, tenant, &spec, ctx, params, headers)
2783 2256 : .await
2784 2256 : .map_err(|e| Box::new(crate::negotiate::ApiError::from(e).into_response()))?
2785 : {
2786 8 : crate::federation::WritePlan::Answered(refused) => Err(refused),
2787 2248 : crate::federation::WritePlan::Forward(regs) => Ok(regs),
2788 : }
2789 2256 : }
2790 :
2791 : #[allow(clippy::too_many_arguments)] // mirrors the wire: one param per forwarded request part
2792 2248 : async fn answer_temporal_attr_write(
2793 2248 : st: &AppState,
2794 2248 : tenant: &antares_model::TenantId,
2795 2248 : headers: &HeaderMap,
2796 2248 : ctx: &antares_jsonld::Context,
2797 2248 : id: &str,
2798 2248 : op: &str,
2799 2248 : method: reqwest::Method,
2800 2248 : path_suffix: &str,
2801 2248 : body: Option<Value>,
2802 2248 : regs: Vec<crate::federation::FedReg>,
2803 2248 : local: LocalWrite,
2804 2248 : ) -> ApiResult<Response> {
2805 2248 : if regs.is_empty() {
2806 168 : return match local.res {
2807 : None => {
2808 2060 : Err(NgsiError::ResourceNotFound(format!("temporal entity {id} not found")).into())
2809 : }
2810 0 : Some(Err(e)) => Err(ApiError::from(e)),
2811 162 : Some(Ok(())) if local.found => Ok(no_content(tenant)),
2812 : Some(Ok(())) => {
2813 6 : Err(NgsiError::ResourceNotFound(format!("{} not found", local.missing)).into())
2814 : }
2815 : };
2816 20 : }
2817 20 : let local_part = match &local.res {
2818 20 : None => crate::federation::Part {
2819 20 : status: 404,
2820 20 : detail: format!("temporal entity {id} not found locally"),
2821 20 : },
2822 0 : Some(_) if local.found => crate::federation::Part {
2823 0 : status: 204,
2824 0 : detail: local.applied.into(),
2825 0 : },
2826 0 : Some(_) => crate::federation::Part {
2827 0 : status: 404,
2828 0 : detail: format!("{} not found locally", local.missing),
2829 0 : },
2830 : };
2831 20 : let mut parts = vec![local_part];
2832 20 : let ctx_url = crate::federation::ctx_link_url(headers, &ctx.source);
2833 20 : for reg in ®s {
2834 20 : if !reg.supports(op) {
2835 4 : if reg.is_proxy() {
2836 4 : parts.push(crate::federation::conflict_part(op));
2837 4 : }
2838 4 : continue;
2839 16 : }
2840 16 : parts.push(
2841 16 : crate::federation::forward_part(
2842 16 : st,
2843 16 : method.clone(),
2844 16 : format!(
2845 16 : "{}/ngsi-ld/v1/temporal/entities/{}{path_suffix}",
2846 16 : reg.endpoint,
2847 16 : crate::federation::path_segment(id)
2848 16 : ),
2849 16 : &[],
2850 16 : headers,
2851 16 : tenant,
2852 16 : reg,
2853 16 : &ctx_url,
2854 16 : body.clone(),
2855 16 : )
2856 16 : .await,
2857 : );
2858 : }
2859 20 : Ok(crate::federation::combine(
2860 20 : parts,
2861 20 : no_content(tenant),
2862 20 : tenant,
2863 20 : ))
2864 2248 : }
2865 :
2866 : // ---------- PATCH/DELETE .../attrs/{attrId}/{instanceId} (5.6.14/5.6.15) ----------
2867 :
2868 20 : pub async fn modify_temporal_instance(
2869 20 : State(st): State<AppState>,
2870 20 : Path((id, attr, instance_id)): Path<(String, String, String)>,
2871 20 : CleanParams(params): CleanParams,
2872 20 : headers: HeaderMap,
2873 20 : body: Bytes,
2874 20 : ) -> Response {
2875 20 : let go = async {
2876 20 : let tenant = tenant_from(&headers)?;
2877 : // Empty attr segment ⇒ the URI names no resource with a PATCH method
2878 : // (suite 016_02_06 asserts 405 here, vs 400 on the DELETE sibling).
2879 20 : if attr.is_empty() {
2880 0 : return Err(ApiError::Bare(StatusCode::METHOD_NOT_ALLOWED));
2881 20 : }
2882 20 : antares_model::EntityId::new(&id)?;
2883 20 : antares_model::check_attr_name(&attr)?;
2884 20 : antares_model::EntityId::new(&instance_id)
2885 20 : .map_err(|_| NgsiError::BadRequestData("invalid instance id".into()))?;
2886 20 : check_params(¶ms, &["local"])?;
2887 20 : gate!(st, &tenant, &headers, "5.6.14", ids: &[&id]).await?;
2888 20 : let parsed = parse_body(&st.loader, &headers, &body, BodyKind::MergePatch).await?;
2889 20 : let obj = parsed.object(NgsiError::BadRequestData(
2890 20 : "fragment must be a JSON object".into(),
2891 20 : ))?;
2892 20 : let mut wrapper = Map::new();
2893 20 : let mut frag = obj.clone();
2894 20 : frag.remove("@context");
2895 20 : wrapper.insert(attr.clone(), Value::Object(frag));
2896 20 : let expanded = expand_entity(
2897 20 : &wrapper,
2898 20 : &parsed.ctx,
2899 20 : ExpandOpts {
2900 20 : fragment: true,
2901 20 : allow_null: false,
2902 20 : temporal: true,
2903 20 : ..Default::default()
2904 20 : },
2905 0 : )?;
2906 20 : let attr_iri = antares_jsonld::expand_attr_name(&attr, &parsed.ctx)?;
2907 20 : let frag_inst = expanded
2908 20 : .get(&attr_iri)
2909 20 : .and_then(Value::as_array)
2910 20 : .and_then(|a| a.first())
2911 20 : .cloned()
2912 20 : .ok_or_else(|| NgsiError::BadRequestData("invalid instance fragment".into()))?;
2913 18 : let regs =
2914 20 : match temporal_write_regs(&st, &tenant, &headers, &parsed.ctx, ¶ms, &id).await {
2915 18 : Ok(regs) => regs,
2916 2 : Err(refused) => return Ok(*refused),
2917 : };
2918 18 : let ts = now_iso();
2919 18 : let mut found = false;
2920 18 : let res = st
2921 18 : .temporal
2922 18 : .mutate(&tenant, &id, |doc| {
2923 6 : let target = antares_store::stored_object(doc)?;
2924 6 : if let Some(arr) = target.get_mut(&attr_iri).and_then(Value::as_array_mut) {
2925 6 : if let Some(inst) = arr.iter_mut().find(|i| {
2926 6 : i.get("instanceId").and_then(Value::as_str) == Some(instance_id.as_str())
2927 6 : }) {
2928 4 : found = true;
2929 : // 5.6.14.4: "Replace the target Attribute instance
2930 : // identified by the instanceId with the Attribute instance
2931 : // in the EntityTemporal Fragment. The createdAt property
2932 : // of the concerned instance shall remain unchanged, but
2933 : // the modifiedAt property shall be set to the timestamp
2934 : // corresponding to this modification." A replace, so a
2935 : // member only the stored instance carries does not survive
2936 : // it; the instance keeps the identity it is addressed by.
2937 4 : let t = antares_store::stored_object(inst)?;
2938 4 : let kept: Vec<(String, Value)> = ["createdAt", "instanceId"]
2939 4 : .iter()
2940 8 : .filter_map(|k| t.get(*k).map(|v| ((*k).to_owned(), v.clone())))
2941 4 : .collect();
2942 4 : t.clear();
2943 4 : t.extend(kept);
2944 12 : for (k, v) in antares_jsonld::expanded_object(&frag_inst)? {
2945 12 : if matches!(k.as_str(), "createdAt" | "instanceId") {
2946 0 : continue;
2947 12 : }
2948 12 : t.insert(k.clone(), v.clone());
2949 : }
2950 4 : t.insert("modifiedAt".into(), Value::String(ts.clone()));
2951 2 : }
2952 0 : }
2953 6 : Ok::<(), NgsiError>(())
2954 6 : })
2955 18 : .await?;
2956 18 : answer_temporal_attr_write(
2957 18 : &st,
2958 18 : &tenant,
2959 18 : &headers,
2960 18 : &parsed.ctx,
2961 18 : &id,
2962 18 : "updateAttrInstanceTemporal",
2963 18 : reqwest::Method::PATCH,
2964 18 : &format!(
2965 18 : "/attrs/{}/{}",
2966 18 : crate::federation::path_segment(&attr),
2967 18 : crate::federation::path_segment(&instance_id)
2968 18 : ),
2969 18 : Some(parsed.value.clone()),
2970 18 : regs,
2971 18 : LocalWrite {
2972 18 : res,
2973 18 : found,
2974 18 : missing: format!("instance {instance_id}"),
2975 18 : applied: "applied locally",
2976 18 : },
2977 18 : )
2978 18 : .await
2979 20 : };
2980 20 : go.await.unwrap_or_else(|e| e.into_response())
2981 20 : }
2982 :
2983 40 : pub async fn delete_temporal_instance(
2984 40 : State(st): State<AppState>,
2985 40 : Path((id, attr, instance_id)): Path<(String, String, String)>,
2986 40 : CleanParams(params): CleanParams,
2987 40 : headers: HeaderMap,
2988 40 : ) -> Response {
2989 40 : let go = async {
2990 40 : let tenant = tenant_from(&headers)?;
2991 40 : antares_model::EntityId::new(&id)?;
2992 40 : antares_model::check_attr_name(&attr)?;
2993 24 : antares_model::EntityId::new(&instance_id)
2994 24 : .map_err(|_| NgsiError::BadRequestData("invalid instance id".into()))?;
2995 24 : check_params(¶ms, &["local"])?;
2996 24 : let ctx = request_context(&st.loader, &headers).await?;
2997 24 : gate!(st, &tenant, &headers, "5.6.15", ids: &[&id]).await?;
2998 24 : let attr_iri = antares_jsonld::expand_attr_name(&attr, &ctx)?;
2999 24 : let regs = match temporal_write_regs(&st, &tenant, &headers, &ctx, ¶ms, &id).await {
3000 22 : Ok(regs) => regs,
3001 2 : Err(refused) => return Ok(*refused),
3002 : };
3003 22 : let mut found = false;
3004 22 : let ts = now_iso();
3005 22 : let res = st
3006 22 : .temporal
3007 22 : .mutate(&tenant, &id, |doc| {
3008 6 : let target = antares_store::stored_object(doc)?;
3009 6 : if let Some(arr) = target.get_mut(&attr_iri).and_then(Value::as_array_mut) {
3010 6 : let before = arr.len();
3011 6 : arr.retain(|i| {
3012 6 : i.get("instanceId").and_then(Value::as_str) != Some(instance_id.as_str())
3013 6 : });
3014 6 : found = arr.len() != before;
3015 6 : if arr.is_empty() {
3016 4 : target.remove(&attr_iri);
3017 4 : }
3018 0 : }
3019 6 : if found {
3020 4 : target.insert("modifiedAt".into(), Value::String(ts.clone()));
3021 4 : }
3022 6 : Ok::<(), NgsiError>(())
3023 6 : })
3024 22 : .await?;
3025 22 : answer_temporal_attr_write(
3026 22 : &st,
3027 22 : &tenant,
3028 22 : &headers,
3029 22 : &ctx,
3030 22 : &id,
3031 22 : "deleteAttrInstanceTemporal",
3032 22 : reqwest::Method::DELETE,
3033 22 : &format!(
3034 22 : "/attrs/{}/{}",
3035 22 : crate::federation::path_segment(&attr),
3036 22 : crate::federation::path_segment(&instance_id)
3037 22 : ),
3038 22 : None,
3039 22 : regs,
3040 22 : LocalWrite {
3041 22 : res,
3042 22 : found,
3043 22 : missing: format!("instance {instance_id}"),
3044 22 : applied: "applied locally",
3045 22 : },
3046 22 : )
3047 22 : .await
3048 40 : };
3049 40 : go.await.unwrap_or_else(|e| e.into_response())
3050 40 : }
3051 :
3052 : // ---------- POST /temporal/entityOperations/query (6.24) ----------
3053 :
3054 98 : pub async fn batch_temporal_query(
3055 98 : State(st): State<AppState>,
3056 98 : CleanParams(params): CleanParams,
3057 98 : headers: HeaderMap,
3058 98 : body: Bytes,
3059 98 : ) -> Response {
3060 98 : let go = async {
3061 : // 6.3.14 and 6.3.4: a Tenant outside the grammar and an Accept the
3062 : // operation cannot serve are both refused here. Neither VALUE is
3063 : // needed — the inner query reads the headers again — but the request
3064 : // must not reach it having skipped either check.
3065 98 : let tenant = tenant_from(&headers)?;
3066 98 : check_params(
3067 98 : ¶ms,
3068 98 : &["limit", "offset", "count", "options", "format", "local"],
3069 0 : )?;
3070 98 : parse_accept(&headers)?;
3071 98 : let filter = gate!(st, &tenant, &headers, "5.7.4").await?;
3072 :
3073 98 : let parsed = parse_body(&st.loader, &headers, &body, BodyKind::Standard).await?;
3074 98 : let q = parsed.object(NgsiError::BadRequestData(
3075 98 : "query body must be an object".into(),
3076 98 : ))?;
3077 98 : if q.get("type").and_then(Value::as_str) != Some("Query") {
3078 0 : return Err(NgsiError::BadRequestData("body type must be Query".into()).into());
3079 98 : }
3080 : // 5.2.23 Query (temporal reading): members flattened with their
3081 : // Table 5.2.23-1 value spaces enforced, incl. temporalQ (5.2.21)
3082 : // and aggrParams (5.2.44).
3083 98 : let mut vp: HashMap<String, String> = params.clone();
3084 98 : crate::paging::query_doc_params(q, true, &mut vp)?;
3085 58 : query_temporal_inner(&st, &vp, &headers, &filter).await
3086 98 : };
3087 98 : go.await.unwrap_or_else(|e| e.into_response())
3088 98 : }
3089 :
3090 : /// 5.14.5.4: temporal query required; the S1–S4 candidate selection is the
3091 : /// temporal query pipeline itself (5.7.4.4) run unpaged — the ids of its
3092 : /// result set form the EntityMap; the createEntityMapQueryTemporal
3093 : /// registrations are then merged like 5.14.4.
3094 : /// Known ceiling: candidate ids are read from the internal 5.7.4 response capped
3095 : /// at max_limit; raise the cap if temporal sets outgrow it.
3096 38 : pub(crate) async fn build_temporal_map(
3097 38 : st: &AppState,
3098 38 : tenant: &TenantId,
3099 38 : headers: &HeaderMap,
3100 38 : ctx: &antares_jsonld::Context,
3101 38 : params: &HashMap<String, String>,
3102 38 : filter: &crate::policy::Filter,
3103 38 : ) -> ApiResult<Value> {
3104 38 : if !params.contains_key("timerel") {
3105 12 : return Err(NgsiError::BadRequestData(
3106 12 : "a temporal query is required to create a temporal EntityMap (5.14.5.4)".into(),
3107 12 : )
3108 12 : .into());
3109 26 : }
3110 26 : let local_scope = params.get("local").map(String::as_str) == Some("true");
3111 26 : let split = params.get("splitEntities").map(String::as_str) == Some("true");
3112 26 : let mut eff: HashMap<String, String> = if split && !local_scope {
3113 0 : params
3114 0 : .iter()
3115 0 : .filter(|(k, _)| {
3116 0 : [
3117 0 : "id",
3118 0 : "idPattern",
3119 0 : "type",
3120 0 : "local",
3121 0 : "timerel",
3122 0 : "timeAt",
3123 0 : "endTimeAt",
3124 0 : "timeproperty",
3125 0 : ]
3126 0 : .contains(&k.as_str())
3127 0 : })
3128 0 : .map(|(k, v)| (k.clone(), v.clone()))
3129 0 : .collect()
3130 : } else {
3131 26 : params.clone()
3132 : };
3133 130 : for k in [
3134 26 : "entityMap",
3135 26 : "entityMapLifetime",
3136 26 : "splitEntities",
3137 26 : "offset",
3138 26 : "count",
3139 130 : ] {
3140 130 : eff.remove(k);
3141 130 : }
3142 26 : eff.insert("limit".into(), st.max_limit.to_string());
3143 : // Box::pin: build_temporal_map is reachable from query_temporal_inner
3144 : // (entityMap=true), so this recursive edge needs indirection.
3145 26 : let resp = Box::pin(query_temporal_inner(st, &eff, headers, filter)).await?;
3146 26 : let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX)
3147 26 : .await
3148 26 : .map_err(|e| NgsiError::InternalError(format!("temporal candidate read: {e}")))?;
3149 : // 5.5.14: the map FIXES the Entities considered by every later request, so
3150 : // an unreadable candidate set must fail rather than become "no candidates".
3151 26 : let candidates: Value = serde_json::from_slice(&bytes)
3152 26 : .map_err(|_| NgsiError::InternalError("temporal candidate parse".into()))?;
3153 26 : let mut emap = Map::new();
3154 26 : if let Some(arr) = candidates.as_array() {
3155 26 : for d in arr {
3156 24 : if let Some(id) = d.get("id").and_then(Value::as_str) {
3157 24 : emap.insert(id.to_owned(), json!(["@none"]));
3158 24 : }
3159 : }
3160 0 : }
3161 26 : crate::entity_map::merge_and_store_map(st, tenant, headers, ctx, params, true, emap).await
3162 38 : }
3163 :
3164 : #[cfg(test)]
3165 : mod clause_4_11 {
3166 : use super::*;
3167 : use serde_json::json;
3168 :
3169 : /// 5.7.3.4 / 4.5.5.3: instances with the same datasetId (or both default)
3170 : /// AND the same timeproperty value are CONFLICTING instances of one slot
3171 : /// — the merge resolves them to one (most recent modifiedAt wins), never
3172 : /// serves both. Regression: the same instance held by two federated
3173 : /// brokers came back twice (IOP_EXT_TMP_02_05).
3174 : #[tokio::test]
3175 4 : async fn merge_resolves_same_slot_instances_to_one() {
3176 4 : let mut base = json!({"id": "urn:e", "type": "T", "speed": [
3177 4 : {"type": "Property", "value": 10, "observedAt": "2026-05-01T00:00:00Z",
3178 4 : "modifiedAt": "2026-05-01T00:00:00Z"},
3179 : ]});
3180 4 : let add = json!({"id": "urn:e", "type": "T", "speed": [
3181 4 : {"type": "Property", "value": 11, "observedAt": "2026-05-01T00:00:00Z",
3182 4 : "modifiedAt": "2026-06-01T00:00:00Z"},
3183 4 : {"type": "Property", "value": 20, "observedAt": "2026-05-02T00:00:00Z",
3184 4 : "modifiedAt": "2026-05-02T00:00:00Z"},
3185 4 : {"type": "Property", "value": 30, "observedAt": "2026-05-01T00:00:00Z",
3186 4 : "datasetId": "urn:d:1", "modifiedAt": "2026-05-01T00:00:00Z"},
3187 : ]});
3188 4 : merge_temporal_docs(&mut base, &add, false, "observedAt");
3189 4 : let speed = base["speed"].as_array().expect("array");
3190 : // default-instance duplicate collapsed (newer modifiedAt won),
3191 : // the other timestamp appended, the datasetId instance is its own slot
3192 4 : assert_eq!(speed.len(), 3, "{speed:?}");
3193 4 : let default_slot: Vec<&Value> = speed
3194 4 : .iter()
3195 12 : .filter(|i| i.get("datasetId").is_none() && i["observedAt"] == "2026-05-01T00:00:00Z")
3196 4 : .collect();
3197 4 : assert_eq!(default_slot.len(), 1);
3198 4 : assert_eq!(default_slot[0]["value"], 11, "newer modifiedAt wins");
3199 4 : }
3200 :
3201 : /// 4.3.6.2: "An auxiliary Context Source Registration never overrides
3202 : /// data held directly within a Context Broker. […] Context data from
3203 : /// auxiliary context sources is only included if it is supplementary to
3204 : /// the context data otherwise available to the Context Broker." On a
3205 : /// Temporal Evolution the unit is the instance, so an auxiliary instance
3206 : /// enters only where no other source supplied that timeproperty value.
3207 : #[tokio::test]
3208 4 : async fn an_auxiliary_instance_supplements_but_never_overrides() {
3209 4 : let mut base = json!({"id": "urn:e", "type": "T", "speed": [
3210 4 : {"type": "Property", "value": 10, "observedAt": "2026-05-01T00:00:00Z",
3211 4 : "modifiedAt": "2026-05-01T00:00:00Z"},
3212 : ]});
3213 4 : let add = json!({"id": "urn:e", "type": "T", "speed": [
3214 : // same slot as the local instance, and newer — still refused
3215 4 : {"type": "Property", "value": 99, "observedAt": "2026-05-01T00:00:00Z",
3216 4 : "modifiedAt": "2026-07-01T00:00:00Z"},
3217 : // a timestamp nobody else supplied: supplementary, so included
3218 4 : {"type": "Property", "value": 20, "observedAt": "2026-05-02T00:00:00Z"},
3219 : ]});
3220 4 : merge_temporal_docs(&mut base, &add, true, "observedAt");
3221 4 : let speed = base["speed"].as_array().expect("array");
3222 4 : assert_eq!(speed.len(), 2, "{speed:?}");
3223 4 : assert!(
3224 4 : !Value::Array(speed.clone()).to_string().contains("99"),
3225 : "an auxiliary instance may not override an occupied slot: {speed:?}"
3226 : );
3227 4 : assert_eq!(speed[0]["value"], 10);
3228 4 : assert_eq!(speed[1]["value"], 20);
3229 4 : }
3230 :
3231 : /// 4.6.3 leaves the seconds fraction optional, so two Context Sources
3232 : /// holding one instance may spell its timeproperty differently. The slot
3233 : /// of 4.5.5.3 is the INSTANT, not the spelling: a byte comparison treats
3234 : /// the two as separate slots and serves the same instance twice — the
3235 : /// IOP_EXT_TMP_02_05 duplicate, reached through a different door.
3236 : #[tokio::test]
3237 4 : async fn one_instant_spelled_two_ways_is_still_one_slot() {
3238 4 : let mut base = json!({"id": "urn:e", "type": "T", "speed": [
3239 4 : {"type": "Property", "value": 10, "observedAt": "2026-05-01T00:00:00Z",
3240 4 : "modifiedAt": "2026-05-01T00:00:00Z"},
3241 : ]});
3242 4 : let add = json!({"id": "urn:e", "type": "T", "speed": [
3243 4 : {"type": "Property", "value": 11, "observedAt": "2026-05-01T00:00:00.000Z",
3244 4 : "modifiedAt": "2026-06-01T00:00:00Z"},
3245 : ]});
3246 4 : merge_temporal_docs(&mut base, &add, false, "observedAt");
3247 4 : let speed = base["speed"].as_array().expect("array");
3248 4 : assert_eq!(speed.len(), 1, "one instant is one slot: {speed:?}");
3249 4 : assert_eq!(speed[0]["value"], 11, "newer modifiedAt wins");
3250 4 : }
3251 :
3252 : /// The winner of a conflicting slot is "the most recent modifiedAt", and
3253 : /// which of two `modifiedAt` values is more recent is a comparison of
3254 : /// instants for the same reason. A remote instance stamped
3255 : /// `…:00.500Z` is LATER than a local one stamped `…:00Z`, though its
3256 : /// bytes sort earlier.
3257 : #[tokio::test]
3258 4 : async fn the_more_recent_modified_at_wins_across_fraction_spellings() {
3259 4 : let mut base = json!({"id": "urn:e", "type": "T", "speed": [
3260 4 : {"type": "Property", "value": 10, "observedAt": "2026-05-01T00:00:00Z",
3261 4 : "modifiedAt": "2026-05-01T09:00:00Z"},
3262 : ]});
3263 4 : let add = json!({"id": "urn:e", "type": "T", "speed": [
3264 4 : {"type": "Property", "value": 11, "observedAt": "2026-05-01T00:00:00Z",
3265 4 : "modifiedAt": "2026-05-01T09:00:00.500Z"},
3266 : ]});
3267 4 : merge_temporal_docs(&mut base, &add, false, "observedAt");
3268 4 : let speed = base["speed"].as_array().expect("array");
3269 4 : assert_eq!(speed.len(), 1, "{speed:?}");
3270 4 : assert_eq!(speed[0]["value"], 11, "the later instant wins: {speed:?}");
3271 4 : }
3272 :
3273 : /// 4.3.6.2 auxiliary supplementation is decided on the same slot, so an
3274 : /// auxiliary instance that respells an occupied instant is still refused.
3275 : #[tokio::test]
3276 4 : async fn an_auxiliary_respelling_of_an_occupied_slot_is_refused() {
3277 4 : let mut base = json!({"id": "urn:e", "type": "T", "speed": [
3278 4 : {"type": "Property", "value": 10, "observedAt": "2026-05-01T00:00:00Z"},
3279 : ]});
3280 4 : let add = json!({"id": "urn:e", "type": "T", "speed": [
3281 4 : {"type": "Property", "value": 99, "observedAt": "2026-05-01T00:00:00.000Z"},
3282 : ]});
3283 4 : merge_temporal_docs(&mut base, &add, true, "observedAt");
3284 4 : let speed = base["speed"].as_array().expect("array");
3285 4 : assert_eq!(speed.len(), 1, "{speed:?}");
3286 4 : assert_eq!(speed[0]["value"], 10);
3287 4 : }
3288 : }
3289 :
3290 : #[cfg(test)]
3291 : mod clause_6_3_10 {
3292 : use super::*;
3293 : use serde_json::json;
3294 :
3295 : /// One attribute with `n` instances, one per minute from 00:00.
3296 24 : fn evolution(n: usize) -> Value {
3297 24 : let speed: Vec<Value> = (0..n)
3298 444 : .map(|i| json!({"type": "Property", "value": i, "observedAt": at(i)}))
3299 24 : .collect();
3300 24 : json!({"id": "urn:ngsi-ld:Vehicle:1", "type": "Vehicle", "speed": speed})
3301 24 : }
3302 :
3303 708 : fn at(i: usize) -> String {
3304 708 : format!("2020-01-01T00:{i:02}:00Z")
3305 708 : }
3306 :
3307 4 : fn tq(timerel: &str, time_at: &str) -> TemporalQ {
3308 4 : let mut p = HashMap::new();
3309 4 : p.insert("timerel".to_owned(), timerel.to_owned());
3310 4 : p.insert("timeAt".to_owned(), time_at.to_owned());
3311 4 : TemporalQ::from_params(&p, true).unwrap().unwrap()
3312 4 : }
3313 :
3314 28 : async fn windowed(doc: &Value, tq: Option<&TemporalQ>, last_n: Option<usize>) -> Windowed {
3315 28 : let mut w = window(doc, tq, last_n, None, None, None, "observedAt");
3316 28 : truncate(&mut w, "observedAt", last_n.is_some());
3317 28 : w
3318 28 : }
3319 :
3320 12 : fn observed(w: &Windowed) -> Vec<&str> {
3321 12 : w.attrs["speed"]
3322 12 : .iter()
3323 80 : .map(|i| i["observedAt"].as_str().unwrap())
3324 12 : .collect()
3325 12 : }
3326 :
3327 : /// 6.3.10: a temporal retrieval the broker cannot serve in full is
3328 : /// answered with 206 and a Content-Range. The body must then BE the
3329 : /// partial representation the header describes — a window wide enough to
3330 : /// select more instances than the broker serves at once is cut to the
3331 : /// ceiling, oldest first, and the advertised range ends at the last
3332 : /// instance returned.
3333 : #[tokio::test]
3334 4 : async fn wide_window_is_cut_to_the_ceiling_and_the_range_matches_the_body() {
3335 4 : let doc = evolution(20);
3336 4 : let q = tq("after", "2019-01-01T00:00:00Z");
3337 4 : let w = windowed(&doc, Some(&q), None).await;
3338 4 : assert_eq!(w.attrs["speed"].len(), TEMPORAL_INSTANCE_LIMIT);
3339 4 : assert!(w.truncated);
3340 4 : let got = observed(&w);
3341 4 : assert_eq!(got[0], at(0));
3342 4 : assert_eq!(
3343 4 : got[TEMPORAL_INSTANCE_LIMIT - 1],
3344 4 : at(TEMPORAL_INSTANCE_LIMIT - 1)
3345 : );
3346 4 : assert!(
3347 4 : !got.contains(&at(19).as_str()),
3348 : "instances beyond the ceiling must not be served: {got:?}"
3349 : );
3350 : // 5.7.3.4 + 6.3.10: start is the requested lower bound, end the last
3351 : // instance in the body, so the header cannot promise more than it sent
3352 4 : assert_eq!(
3353 4 : content_range(
3354 4 : w.truncated,
3355 4 : w.ts_min.as_deref(),
3356 4 : w.ts_max.as_deref(),
3357 4 : Some(&q),
3358 4 : None
3359 4 : ),
3360 4 : Some(format!(
3361 4 : "date-time 2019-01-01T00:00:00Z-{}/*",
3362 4 : at(TEMPORAL_INSTANCE_LIMIT - 1)
3363 4 : ))
3364 4 : );
3365 4 : }
3366 :
3367 : /// Two attributes, `speed` minutes 0..n and `heading` minutes 5..m.
3368 8 : fn evolution2(n: usize, m: usize) -> Value {
3369 8 : let mut doc = evolution(n);
3370 8 : doc["heading"] = (5..m)
3371 208 : .map(|i| json!({"type": "Property", "value": i, "observedAt": at(i)}))
3372 8 : .collect();
3373 8 : doc
3374 8 : }
3375 :
3376 8 : fn last_observed(w: &Windowed, attr: &str) -> Option<String> {
3377 8 : w.attrs[attr]
3378 8 : .iter()
3379 52 : .filter_map(|i| i["observedAt"].as_str().map(str::to_owned))
3380 8 : .max()
3381 8 : }
3382 :
3383 : /// 6.3.10: the partial content IS the representation the Content-Range
3384 : /// describes — so the cut is one time boundary for the whole entity.
3385 : /// With `speed` over-full first, `heading` is trimmed to the same last
3386 : /// instant, and nothing of either attribute lies past the advertised
3387 : /// range-end (a client continuing from it misses no instance).
3388 : #[tokio::test]
3389 4 : async fn the_cut_is_one_time_boundary_across_attributes() {
3390 4 : let w = windowed(&evolution2(21, 31), None, None).await;
3391 4 : assert!(w.truncated);
3392 4 : let end = at(TEMPORAL_INSTANCE_LIMIT - 1);
3393 4 : assert_eq!(w.attrs["speed"].len(), TEMPORAL_INSTANCE_LIMIT);
3394 4 : assert_eq!(last_observed(&w, "speed").as_deref(), Some(end.as_str()));
3395 4 : assert_eq!(
3396 4 : last_observed(&w, "heading").as_deref(),
3397 4 : Some(end.as_str()),
3398 : "heading must stop at speed's boundary, not at its own ninth instance"
3399 : );
3400 4 : assert_eq!(w.attrs["heading"].len(), TEMPORAL_INSTANCE_LIMIT - 5);
3401 4 : assert_eq!(w.ts_max.as_deref(), Some(end.as_str()));
3402 52 : for inst in w.attrs["heading"].iter().chain(w.attrs["speed"].iter()) {
3403 52 : assert!(
3404 52 : inst["observedAt"].as_str().expect("t") <= end.as_str(),
3405 : "no instance may lie past the advertised range-end: {inst}"
3406 : );
3407 : }
3408 : // backwards (lastN): the boundary is the LATEST ninth instant, so the
3409 : // page covers [heading's ninth-newest, newest] and `speed`, which ends
3410 : // before that, is empty on this page rather than incoherently present
3411 4 : let w = windowed(&evolution2(21, 31), None, Some(20)).await;
3412 4 : assert!(w.truncated);
3413 4 : assert_eq!(w.attrs["heading"].len(), TEMPORAL_INSTANCE_LIMIT);
3414 4 : assert_eq!(w.attrs["heading"][0]["observedAt"], json!(at(30)));
3415 4 : assert_eq!(
3416 4 : w.ts_min.as_deref(),
3417 4 : Some(at(30 - TEMPORAL_INSTANCE_LIMIT + 1).as_str())
3418 : );
3419 4 : assert!(
3420 4 : w.attrs["speed"].is_empty(),
3421 4 : "speed lies entirely before the page boundary: {:?}",
3422 4 : w.attrs["speed"]
3423 4 : );
3424 4 : }
3425 :
3426 : /// 5.7.3.4/5.7.4.4 lastN "shall be limited to the specified number of
3427 : /// instances" — an upper limit, not an entitlement: a lastN above the
3428 : /// broker ceiling is served up to the ceiling (newest first) and the
3429 : /// answer is the partial one. The Content-Range size stays the requested
3430 : /// lastN, its start-end pair the instants actually returned.
3431 : #[tokio::test]
3432 4 : async fn last_n_above_the_ceiling_is_clamped_to_it() {
3433 4 : let doc = evolution(20);
3434 4 : let w = windowed(&doc, None, Some(20)).await;
3435 4 : assert_eq!(w.attrs["speed"].len(), TEMPORAL_INSTANCE_LIMIT);
3436 4 : assert!(w.truncated);
3437 4 : let got = observed(&w);
3438 4 : assert_eq!(got[0], at(19), "lastN delivers newest first");
3439 4 : assert_eq!(
3440 4 : got[TEMPORAL_INSTANCE_LIMIT - 1],
3441 4 : at(20 - TEMPORAL_INSTANCE_LIMIT)
3442 : );
3443 4 : assert_eq!(
3444 4 : content_range(
3445 4 : w.truncated,
3446 4 : w.ts_min.as_deref(),
3447 4 : w.ts_max.as_deref(),
3448 4 : None,
3449 4 : Some(20)
3450 4 : ),
3451 4 : Some(format!(
3452 4 : "date-time {}-{}/20",
3453 4 : at(19),
3454 4 : at(20 - TEMPORAL_INSTANCE_LIMIT)
3455 4 : ))
3456 4 : );
3457 4 : }
3458 :
3459 : /// 5.7.3.4: temporalQ is optional on retrieval, so a request naming no
3460 : /// window at all asks for the whole Temporal Evolution. It is still
3461 : /// bounded by the same ceiling, and still answered as partial.
3462 : #[tokio::test]
3463 4 : async fn a_request_with_no_window_is_capped_by_default() {
3464 4 : let doc = evolution(20);
3465 4 : let w = windowed(&doc, None, None).await;
3466 4 : assert_eq!(w.attrs["speed"].len(), TEMPORAL_INSTANCE_LIMIT);
3467 4 : assert!(w.truncated);
3468 4 : assert_eq!(
3469 4 : w.ts_max.as_deref(),
3470 4 : Some(at(TEMPORAL_INSTANCE_LIMIT - 1).as_str())
3471 : );
3472 4 : assert_eq!(
3473 4 : content_range(
3474 4 : w.truncated,
3475 4 : w.ts_min.as_deref(),
3476 4 : w.ts_max.as_deref(),
3477 4 : None,
3478 4 : None
3479 4 : ),
3480 4 : Some(format!(
3481 4 : "date-time {}-{}/*",
3482 4 : at(0),
3483 4 : at(TEMPORAL_INSTANCE_LIMIT - 1)
3484 4 : ))
3485 4 : );
3486 4 : }
3487 :
3488 : /// 6.3.10: partial content is conditional on truncation — a result the
3489 : /// broker serves in full is a plain 200 with no Content-Range.
3490 : #[tokio::test]
3491 4 : async fn a_complete_result_is_not_partial_content() {
3492 4 : let doc = evolution(TEMPORAL_INSTANCE_LIMIT);
3493 4 : let w = windowed(&doc, None, None).await;
3494 4 : assert_eq!(w.attrs["speed"].len(), TEMPORAL_INSTANCE_LIMIT);
3495 4 : assert!(!w.truncated);
3496 4 : assert_eq!(
3497 4 : content_range(
3498 4 : w.truncated,
3499 4 : w.ts_min.as_deref(),
3500 4 : w.ts_max.as_deref(),
3501 4 : None,
3502 4 : None
3503 4 : ),
3504 4 : None
3505 4 : );
3506 4 : }
3507 :
3508 : /// 4.6.3 allows a DateTime to carry a seconds fraction or leave it out,
3509 : /// and both spellings of one instant are the same instant. The window's
3510 : /// own bounds are compared as raw strings nowhere: `.` sorts before `Z`,
3511 : /// so `…:00.500Z` reads as EARLIER than `…:00Z` on a byte compare, and a
3512 : /// Content-Range built from those bounds would name a range the body
3513 : /// contradicts. The instances are sorted on the canonical key already
3514 : /// (`dt_key`); the bounds are on the same key or they disagree with the
3515 : /// order they summarize.
3516 : #[tokio::test]
3517 4 : async fn the_window_bounds_are_the_true_extremes_across_fraction_spellings() {
3518 4 : let doc = json!({
3519 4 : "id": "urn:ngsi-ld:Vehicle:1",
3520 4 : "type": "Vehicle",
3521 4 : "speed": [
3522 4 : {"type": "Property", "value": 1, "observedAt": "2020-01-01T00:09:00Z"},
3523 4 : {"type": "Property", "value": 2, "observedAt": "2020-01-01T00:09:00.500Z"},
3524 : ],
3525 : });
3526 4 : let w = windowed(&doc, None, None).await;
3527 4 : assert_eq!(w.attrs["speed"].len(), 2);
3528 4 : assert_eq!(
3529 4 : observed(&w),
3530 4 : vec!["2020-01-01T00:09:00Z", "2020-01-01T00:09:00.500Z"],
3531 : "the instances themselves sort on the canonical key"
3532 : );
3533 4 : assert_eq!(w.ts_min.as_deref(), Some("2020-01-01T00:09:00Z"));
3534 4 : assert_eq!(w.ts_max.as_deref(), Some("2020-01-01T00:09:00.500Z"));
3535 4 : }
3536 : }
3537 :
3538 : #[cfg(test)]
3539 : mod forwarded_path_encoding {
3540 : use crate::AppState;
3541 : use axum::body::Body;
3542 : use axum::http::{Request, StatusCode};
3543 : use std::io::{Read, Write};
3544 : use std::sync::{Arc, Mutex};
3545 : use tower::ServiceExt;
3546 :
3547 : /// An entity id is a URI (4.6.2), and `#` is legal in one. It also ends a
3548 : /// path in RFC 3986 clause 3.3, so the id has to be percent-encoded
3549 : /// wherever it becomes a path segment.
3550 : const ENTITY: &str = "urn:ngsi-ld:Vehicle:temporal-enc#frag";
3551 : const ENCODED: &str = "urn:ngsi-ld:Vehicle:temporal-enc%23frag";
3552 :
3553 : /// A Context Source answering 204 to everything, recording request lines.
3554 8 : fn mock_source() -> (u16, Arc<Mutex<Vec<String>>>) {
3555 8 : let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
3556 8 : let port = listener.local_addr().expect("addr").port();
3557 8 : let seen: Arc<Mutex<Vec<String>>> = Arc::default();
3558 8 : let log = seen.clone();
3559 8 : std::thread::spawn(move || {
3560 8 : for stream in listener.incoming() {
3561 8 : let Ok(mut s) = stream else { continue };
3562 8 : let mut buf = [0u8; 8192];
3563 8 : let n = s.read(&mut buf).unwrap_or(0);
3564 8 : if let Some(line) = String::from_utf8_lossy(&buf[..n]).lines().next() {
3565 8 : log.lock().expect("lock").push(line.to_owned());
3566 8 : }
3567 8 : let _ = s.write_all(
3568 8 : b"HTTP/1.1 204 No Content\r\nConnection: close\r\nContent-Length: 0\r\n\r\n",
3569 8 : );
3570 : }
3571 8 : });
3572 8 : (port, seen)
3573 8 : }
3574 :
3575 12 : fn state() -> AppState {
3576 : // the mock source is loopback, denied by the egress policy by default
3577 12 : crate::allow_private();
3578 12 : AppState::new("antares-temporal-enc".into())
3579 12 : }
3580 :
3581 52 : async fn send(st: &AppState, req: Request<Body>) -> axum::http::Response<Body> {
3582 52 : crate::router(st.clone())
3583 52 : .oneshot(req)
3584 52 : .await
3585 52 : .expect("response")
3586 52 : }
3587 :
3588 12 : async fn post(st: &AppState, uri: &str, body: String) -> axum::http::Response<Body> {
3589 12 : let req = Request::builder()
3590 12 : .method("POST")
3591 12 : .uri(uri)
3592 12 : .header("Content-Type", "application/json")
3593 12 : .header("Content-Length", body.len())
3594 12 : .body(Body::from(body))
3595 12 : .expect("request");
3596 12 : send(st, req).await
3597 12 : }
3598 :
3599 8 : async fn register(st: &AppState, port: u16, id: &str, entity: &str) {
3600 8 : let doc = serde_json::json!({
3601 8 : "id": format!("urn:ngsi-ld:ContextSourceRegistration:{id}"),
3602 8 : "type": "ContextSourceRegistration",
3603 8 : "mode": "redirect",
3604 8 : "operations": ["deleteAttrsTemporal", "deleteAttrInstanceTemporal"],
3605 8 : "information": [{"entities": [{"type": "Vehicle", "id": entity}]}],
3606 8 : "endpoint": format!("http://127.0.0.1:{port}"),
3607 : });
3608 8 : assert_eq!(
3609 8 : post(st, "/ngsi-ld/v1/csourceRegistrations", doc.to_string())
3610 8 : .await
3611 8 : .status(),
3612 : StatusCode::CREATED,
3613 : "registration create"
3614 : );
3615 8 : }
3616 :
3617 : /// 5.6.13.4/5.6.15.4: the operation is forwarded to the registration
3618 : /// endpoint with the target resource named in the request path. The id,
3619 : /// the Attribute name and the instanceId arrive percent-decoded from this
3620 : /// broker's own path, so splicing them raw would let a `#` end the
3621 : /// forwarded path (RFC 3986 clause 3.3) and turn Delete Attribute into
3622 : /// Delete Temporal Evolution of an Entity (5.6.16) on the peer.
3623 : #[tokio::test(flavor = "multi_thread")]
3624 4 : async fn a_hash_in_the_id_reaches_the_peer_encoded_not_truncated() {
3625 4 : let st = state();
3626 4 : let (port, seen) = mock_source();
3627 4 : register(&st, port, "csr-temporal-enc", ENTITY).await;
3628 :
3629 : // each suffix is already in its encoded spelling, so the forwarded
3630 : // path must repeat it verbatim
3631 8 : for suffix in ["/attrs/speed", "/attrs/speed/urn:ngsi-ld:Instance:1%23x"] {
3632 8 : let req = Request::builder()
3633 8 : .method("DELETE")
3634 8 : .uri(format!("/ngsi-ld/v1/temporal/entities/{ENCODED}{suffix}"))
3635 8 : .body(Body::empty())
3636 8 : .expect("request");
3637 8 : let status = send(&st, req).await.status();
3638 8 : assert_ne!(status, StatusCode::BAD_REQUEST, "{suffix}");
3639 8 : let lines = seen.lock().expect("lock").clone();
3640 8 : let last = lines.last().cloned().unwrap_or_default();
3641 8 : assert!(
3642 8 : last.contains(&format!("/ngsi-ld/v1/temporal/entities/{ENCODED}{suffix}")),
3643 4 : "forwarded request line {last:?} for suffix {suffix}"
3644 4 : );
3645 4 : // the negative assertion: the peer must never see a path that
3646 4 : // stops at the entity resource
3647 8 : assert!(
3648 8 : !last.contains("temporal-enc HTTP/"),
3649 4 : "forwarded path truncated at the `#`: {last:?}"
3650 4 : );
3651 4 : }
3652 4 : }
3653 :
3654 : /// 5.6.13.4: "If the target Attribute name is not a valid name, then an
3655 : /// error of type BadRequestData shall be raised." A name begins with a
3656 : /// letter (4.6.2), so no valid name is a relative-path dot segment (RFC
3657 : /// 3986 clause 5.2.4) — and such a name in a forwarded path would address
3658 : /// the peer's Temporal Evolution resource instead of its Attribute, so it
3659 : /// is refused before anything leaves this broker.
3660 : #[tokio::test(flavor = "multi_thread")]
3661 4 : async fn a_dot_segment_attribute_name_is_refused_and_never_forwarded() {
3662 : const TARGET: &str = "urn:ngsi-ld:Vehicle:temporal-dots";
3663 4 : let st = state();
3664 4 : let (port, seen) = mock_source();
3665 4 : register(&st, port, "csr-temporal-dots", TARGET).await;
3666 : // raw, decoded once by this broker, and decoded once more by the peer
3667 16 : for attr in ["..", "%2e%2e", "%252e%252e", "."] {
3668 32 : for suffix in ["", "/urn:ngsi-ld:Instance:1"] {
3669 32 : let req = Request::builder()
3670 32 : .method("DELETE")
3671 32 : .uri(format!(
3672 4 : "/ngsi-ld/v1/temporal/entities/{TARGET}/attrs/{attr}{suffix}"
3673 4 : ))
3674 32 : .body(Body::empty())
3675 32 : .expect("request");
3676 32 : assert_eq!(
3677 32 : send(&st, req).await.status(),
3678 4 : StatusCode::BAD_REQUEST,
3679 4 : "attribute name {attr:?} with suffix {suffix:?}"
3680 4 : );
3681 4 : }
3682 4 : }
3683 4 : assert!(
3684 4 : seen.lock().expect("lock").is_empty(),
3685 4 : "a rejected attribute name must never reach a registration endpoint"
3686 4 : );
3687 4 : }
3688 :
3689 : /// 5.6.11.4: on creation the response carries a Location header holding
3690 : /// the resource URI of the created Temporal Representation. A URI has its
3691 : /// reserved characters percent-encoded (RFC 3986 clause 3.3), so a `#` in
3692 : /// the id may not be spliced raw — there it would read as the start of a
3693 : /// fragment identifier and address the entity collection.
3694 : #[tokio::test(flavor = "multi_thread")]
3695 4 : async fn the_location_header_percent_encodes_the_id() {
3696 4 : let st = state();
3697 4 : let doc = serde_json::json!({
3698 4 : "id": ENTITY, "type": "Vehicle",
3699 4 : "speed": [{"type": "Property", "value": 1,
3700 4 : "observedAt": "2026-03-01T12:05:00Z"}],
3701 : });
3702 4 : let res = post(&st, "/ngsi-ld/v1/temporal/entities", doc.to_string()).await;
3703 4 : assert_eq!(res.status(), StatusCode::CREATED);
3704 4 : assert_eq!(
3705 4 : res.headers()
3706 4 : .get("Location")
3707 4 : .and_then(|v| v.to_str().ok())
3708 4 : .unwrap_or_default(),
3709 4 : format!("/ngsi-ld/v1/temporal/entities/{ENCODED}")
3710 4 : );
3711 4 : }
3712 : }
3713 :
3714 : #[cfg(test)]
3715 : mod clause_4_5_19 {
3716 : use super::*;
3717 : use serde_json::json;
3718 :
3719 : /// 4.5.19.1: "The duration shall be a string in the format
3720 : /// `P[n]Y[n]M[n]DT[n]H[n]M[n]S` or `P[n]W` … For example,
3721 : /// `"P3Y6M4DT12H30M5S"` represents a duration of "three years, six
3722 : /// months, four days, twelve hours, thirty minutes, and five seconds"."
3723 : /// A period mixing date and time elements is therefore valid, and
3724 : /// "PT0S" spans the whole time range of the query.
3725 : #[test]
3726 4 : fn a_mixed_date_and_time_duration_is_a_valid_period() {
3727 : const DAY: i64 = 86_400;
3728 4 : assert_eq!(
3729 4 : parse_iso_duration("P3Y6M4DT12H30M5S"),
3730 4 : Some(AggrPeriod::Months(42, 4 * DAY + 12 * 3600 + 30 * 60 + 5))
3731 : );
3732 4 : assert_eq!(
3733 4 : parse_iso_duration("P1Y1D"),
3734 : Some(AggrPeriod::Months(12, DAY))
3735 : );
3736 4 : assert_eq!(
3737 4 : parse_iso_duration("P1MT1H"),
3738 : Some(AggrPeriod::Months(1, 3600))
3739 : );
3740 : // the pure forms are unchanged
3741 4 : assert_eq!(parse_iso_duration("PT0S"), Some(AggrPeriod::Whole));
3742 4 : assert_eq!(parse_iso_duration("P0D"), Some(AggrPeriod::Whole));
3743 4 : assert_eq!(parse_iso_duration("P1M"), Some(AggrPeriod::Months(1, 0)));
3744 4 : assert_eq!(parse_iso_duration("PT90M"), Some(AggrPeriod::Seconds(5400)));
3745 4 : assert_eq!(
3746 4 : parse_iso_duration("P1W"),
3747 4 : Some(AggrPeriod::Seconds(7 * DAY))
3748 : );
3749 : // and the grammar still rejects what is not a duration
3750 4 : assert_eq!(parse_iso_duration("P1X"), None);
3751 4 : assert_eq!(parse_iso_duration("1Y"), None);
3752 4 : assert_eq!(parse_iso_duration("P1"), None);
3753 4 : }
3754 :
3755 12 : fn windowed(times: &[&str]) -> Windowed {
3756 12 : let instances: Vec<Value> = times
3757 12 : .iter()
3758 28 : .map(|t| json!({"type": "Property", "value": 1, "observedAt": t}))
3759 12 : .collect();
3760 12 : let mut attrs = std::collections::BTreeMap::new();
3761 12 : attrs.insert("speed".to_owned(), instances);
3762 : Windowed {
3763 12 : attrs,
3764 12 : max_per_attr: times.len(),
3765 12 : ts_min: times.first().map(|s| (*s).to_owned()),
3766 12 : ts_max: times.last().map(|s| (*s).to_owned()),
3767 : truncated: false,
3768 : }
3769 12 : }
3770 :
3771 20 : fn repr(duration: &str) -> TRepr {
3772 20 : TRepr {
3773 20 : aggregated: true,
3774 20 : aggr_methods: vec!["totalCount".to_owned()],
3775 20 : aggr_period: parse_iso_duration(duration).expect("duration"),
3776 20 : ..Default::default()
3777 20 : }
3778 20 : }
3779 :
3780 : /// The periods of an aggregated response are the periods "in the time
3781 : /// range of the query" (4.5.19.0), so with `timerel=before` they run
3782 : /// backwards from `timeAt` and every returned period contains the
3783 : /// instances aggregated into it.
3784 : #[tokio::test]
3785 4 : async fn month_periods_before_the_anchor_contain_their_instances() {
3786 4 : let w = windowed(&[
3787 4 : "2020-01-15T00:00:00Z",
3788 4 : "2020-02-15T00:00:00Z",
3789 4 : "2020-03-15T00:00:00Z",
3790 4 : ]);
3791 4 : let tq = TemporalQ {
3792 4 : timerel: "before".to_owned(),
3793 4 : time_at: "2020-04-01T00:00:00Z".to_owned(),
3794 4 : end_time_at: None,
3795 4 : timeproperty: "observedAt".to_owned(),
3796 4 : };
3797 4 : let out = render_aggregated(
3798 4 : &w,
3799 4 : Some(&tq),
3800 4 : &repr("P1M"),
3801 4 : &antares_jsonld::Context::default(),
3802 4 : "observedAt",
3803 : )
3804 4 : .expect("aggregated");
3805 4 : let rows = out["speed"]["totalCount"].as_array().expect("rows").clone();
3806 4 : assert_eq!(
3807 : rows,
3808 4 : vec![
3809 4 : json!([1, "2020-01-01T00:00:00Z", "2020-02-01T00:00:00Z"]),
3810 4 : json!([1, "2020-02-01T00:00:00Z", "2020-03-01T00:00:00Z"]),
3811 4 : json!([1, "2020-03-01T00:00:00Z", "2020-04-01T00:00:00Z"]),
3812 : ]
3813 : );
3814 : // the negative assertion: no period may start at the anchor, since
3815 : // such a period holds none of the instances of a `before` query
3816 4 : assert!(
3817 4 : !Value::Array(rows)
3818 4 : .to_string()
3819 4 : .contains("2020-04-01T00:00:00Z\",\""),
3820 4 : "a period starting at timeAt contains no instance of a before query"
3821 4 : );
3822 4 : }
3823 :
3824 : /// 4.5.19.1: "A duration of 0 second (e.g. expressed as "PT0S" or
3825 : /// "P0D") is valid and is interpreted as a duration spanning the whole
3826 : /// time range specified by the temporal query." The query names one
3827 : /// edge of that range and 4.11 leaves the other open, so the period
3828 : /// runs from `timeAt` only when `timeAt` is where the range starts.
3829 : #[tokio::test]
3830 4 : async fn the_zero_duration_period_spans_the_time_range_the_query_asked_for() {
3831 4 : let w = windowed(&["2020-09-01T12:03:00Z", "2020-09-01T12:05:00Z"]);
3832 12 : let one = |tq: &TemporalQ| {
3833 12 : let out = render_aggregated(
3834 12 : &w,
3835 12 : Some(tq),
3836 12 : &repr("PT0S"),
3837 12 : &antares_jsonld::Context::default(),
3838 12 : "observedAt",
3839 : )
3840 12 : .expect("aggregated");
3841 12 : out["speed"]["totalCount"].as_array().expect("rows").clone()
3842 12 : };
3843 :
3844 : // before: timeAt ENDS the range (4.11 makes the start open, so the
3845 : // data supplies it). The period must not run backwards.
3846 4 : assert_eq!(
3847 4 : one(&TemporalQ {
3848 4 : timerel: "before".to_owned(),
3849 4 : time_at: "2030-01-01T00:00:00Z".to_owned(),
3850 4 : end_time_at: None,
3851 4 : timeproperty: "observedAt".to_owned(),
3852 4 : }),
3853 4 : vec![json!([2, "2020-09-01T12:03:00Z", "2030-01-01T00:00:00Z"])]
3854 : );
3855 :
3856 : // between: both edges are named, and the period is exactly them —
3857 : // not the last instant the data happens to hold.
3858 4 : assert_eq!(
3859 4 : one(&TemporalQ {
3860 4 : timerel: "between".to_owned(),
3861 4 : time_at: "2020-09-01T12:00:00Z".to_owned(),
3862 4 : end_time_at: Some("2020-09-01T13:00:00Z".to_owned()),
3863 4 : timeproperty: "observedAt".to_owned(),
3864 4 : }),
3865 4 : vec![json!([2, "2020-09-01T12:00:00Z", "2020-09-01T13:00:00Z"])]
3866 : );
3867 :
3868 : // after: timeAt STARTS the range, the data closes it.
3869 4 : assert_eq!(
3870 4 : one(&TemporalQ {
3871 4 : timerel: "after".to_owned(),
3872 4 : time_at: "2020-01-01T00:00:00Z".to_owned(),
3873 4 : end_time_at: None,
3874 4 : timeproperty: "observedAt".to_owned(),
3875 4 : }),
3876 4 : vec![json!([2, "2020-01-01T00:00:00Z", "2020-09-01T12:05:01Z"])]
3877 4 : );
3878 4 : }
3879 :
3880 : /// A mixed period steps by its months AND its seconds: "P1MT12H" from
3881 : /// the anchor ends one month and twelve hours later (4.5.19.1).
3882 : #[tokio::test]
3883 4 : async fn a_mixed_period_steps_by_both_components() {
3884 4 : let w = windowed(&["2020-01-01T06:00:00Z", "2020-02-02T06:00:00Z"]);
3885 4 : let tq = TemporalQ {
3886 4 : timerel: "after".to_owned(),
3887 4 : time_at: "2020-01-01T00:00:00Z".to_owned(),
3888 4 : end_time_at: None,
3889 4 : timeproperty: "observedAt".to_owned(),
3890 4 : };
3891 4 : let out = render_aggregated(
3892 4 : &w,
3893 4 : Some(&tq),
3894 4 : &repr("P1MT12H"),
3895 4 : &antares_jsonld::Context::default(),
3896 4 : "observedAt",
3897 : )
3898 4 : .expect("aggregated");
3899 4 : assert_eq!(
3900 4 : out["speed"]["totalCount"],
3901 4 : json!([
3902 4 : [1, "2020-01-01T00:00:00Z", "2020-02-01T12:00:00Z"],
3903 4 : [1, "2020-02-01T12:00:00Z", "2020-03-02T00:00:00Z"],
3904 4 : ])
3905 4 : );
3906 4 : }
3907 :
3908 : /// One attribute whose instances carry the given members, an hour apart.
3909 48 : fn windowed_props(members: &[Value]) -> Windowed {
3910 48 : let instances: Vec<Value> = members
3911 48 : .iter()
3912 48 : .enumerate()
3913 92 : .map(|(i, m)| {
3914 92 : let mut inst = json!({
3915 92 : "type": "Property",
3916 92 : "observedAt": format!("2020-01-01T{i:02}:00:00Z"),
3917 : });
3918 100 : for (k, v) in m.as_object().expect("instance members") {
3919 100 : inst[k] = v.clone();
3920 100 : }
3921 92 : inst
3922 92 : })
3923 48 : .collect();
3924 48 : let times: Vec<String> = instances
3925 48 : .iter()
3926 92 : .map(|i| i["observedAt"].as_str().unwrap_or_default().to_owned())
3927 48 : .collect();
3928 48 : let mut attrs = std::collections::BTreeMap::new();
3929 48 : attrs.insert("speed".to_owned(), instances);
3930 48 : Windowed {
3931 48 : attrs,
3932 48 : max_per_attr: members.len(),
3933 48 : ts_min: times.first().cloned(),
3934 48 : ts_max: times.last().cloned(),
3935 48 : truncated: false,
3936 48 : }
3937 48 : }
3938 :
3939 72 : fn aggregate(w: &Windowed, methods: &[&str]) -> Result<Map<String, Value>, NgsiError> {
3940 72 : let r = TRepr {
3941 : aggregated: true,
3942 116 : aggr_methods: methods.iter().map(|m| (*m).to_string()).collect(),
3943 72 : ..Default::default()
3944 : };
3945 72 : render_aggregated(
3946 72 : w,
3947 72 : None,
3948 72 : &r,
3949 72 : &antares_jsonld::Context::default(),
3950 72 : "observedAt",
3951 : )
3952 72 : }
3953 :
3954 : /// Table 4.5.19.1-2: on a DateTime and on a Date, `min` "calculates the
3955 : /// minimum value inside the period" and `max` the maximum. Both
3956 : /// datatypes reach the broker as a JSON-LD typed value — C.6's
3957 : /// `{"@type": "DateTime", "@value": "2018-12-04T12:00:00Z"}` — so the
3958 : /// aggregation reads the value through its type instead of treating the
3959 : /// wrapper as an opaque object.
3960 : #[test]
3961 4 : fn a_date_time_and_a_date_have_a_minimum_and_a_maximum() {
3962 12 : let dt = |v: &str| json!({"value": {"@type": "DateTime", "@value": v}});
3963 4 : let out = aggregate(
3964 4 : &windowed_props(&[
3965 4 : dt("2020-03-01T00:00:00Z"),
3966 4 : dt("2020-01-01T00:00:00Z"),
3967 4 : dt("2020-02-01T00:00:00Z"),
3968 4 : ]),
3969 4 : &["min", "max"],
3970 : )
3971 4 : .expect("DateTime is eligible for min and max");
3972 4 : assert_eq!(
3973 4 : out["speed"]["min"][0][0],
3974 4 : json!({"@type": "DateTime", "@value": "2020-01-01T00:00:00Z"})
3975 : );
3976 4 : assert_eq!(
3977 4 : out["speed"]["max"][0][0],
3978 4 : json!({"@type": "DateTime", "@value": "2020-03-01T00:00:00Z"})
3979 : );
3980 :
3981 8 : let d = |v: &str| json!({"value": {"@type": "Date", "@value": v}});
3982 4 : let out = aggregate(
3983 4 : &windowed_props(&[d("2020-03-01"), d("2020-01-01")]),
3984 4 : &["min", "max"],
3985 : )
3986 4 : .expect("Date is eligible for min and max");
3987 4 : assert_eq!(
3988 4 : out["speed"]["min"][0][0],
3989 4 : json!({"@type": "Date", "@value": "2020-01-01"})
3990 : );
3991 4 : assert_eq!(
3992 4 : out["speed"]["max"][0][0],
3993 4 : json!({"@type": "Date", "@value": "2020-03-01"})
3994 : );
3995 4 : }
3996 :
3997 : /// C.6 gives a second representation of the same datatypes: the value
3998 : /// stays a string and `valueType` (4.5.2.2) carries the type, coerced to
3999 : /// its datatype URI on the way in. Table 4.5.19.1-2 applies to the
4000 : /// datatype, not to the spelling, so this form aggregates identically.
4001 : #[test]
4002 4 : fn a_value_type_carries_the_datatype_as_far_as_the_typed_value_does() {
4003 4 : let dt =
4004 8 : |v: &str| json!({"value": v, "valueType": "https://uri.etsi.org/ngsi-ld/DateTime"});
4005 4 : let out = aggregate(
4006 4 : &windowed_props(&[dt("2020-03-01T00:00:00Z"), dt("2020-01-01T00:00:00Z")]),
4007 4 : &["min", "max"],
4008 : )
4009 4 : .expect("a valueType-coerced DateTime is eligible for min and max");
4010 4 : assert_eq!(out["speed"]["min"][0][0], json!("2020-01-01T00:00:00Z"));
4011 4 : assert_eq!(out["speed"]["max"][0][0], json!("2020-03-01T00:00:00Z"));
4012 4 : }
4013 :
4014 : /// Table 4.5.19.1-2, Time column: `avg` "calculates the average time
4015 : /// inside the period", and min/max apply as well. 4.6.3 mandates
4016 : /// `hh:mm:ssZ` for a Time, so the computed average is one — carrying its
4017 : /// type, since a bare string would read back as a JSON String, whose own
4018 : /// column in Table 4.5.19.1-1 has no average at all.
4019 : #[test]
4020 4 : fn a_time_has_an_average_a_minimum_and_a_maximum() {
4021 8 : let t = |v: &str| json!({"value": {"@type": "Time", "@value": v}});
4022 4 : let out = aggregate(
4023 4 : &windowed_props(&[t("09:30:00Z"), t("08:30:00Z")]),
4024 4 : &["avg", "min", "max"],
4025 : )
4026 4 : .expect("Time is eligible for avg, min and max");
4027 4 : assert_eq!(
4028 4 : out["speed"]["avg"][0][0],
4029 4 : json!({"@type": "Time", "@value": "09:00:00Z"})
4030 : );
4031 4 : assert_eq!(
4032 4 : out["speed"]["min"][0][0],
4033 4 : json!({"@type": "Time", "@value": "08:30:00Z"})
4034 : );
4035 4 : assert_eq!(
4036 4 : out["speed"]["max"][0][0],
4037 4 : json!({"@type": "Time", "@value": "09:30:00Z"})
4038 : );
4039 4 : }
4040 :
4041 : /// The N/A cells of Table 4.5.19.1-2 are refused, not computed: 5.7.4.4
4042 : /// p.211 raises InvalidRequest when an Attribute "is not eligible for at
4043 : /// least one of the aggregation methods specified in the request".
4044 : /// DateTime and Date have no avg, sum, stddev or sumsq; Time has no sum,
4045 : /// stddev or sumsq.
4046 : #[test]
4047 4 : fn the_methods_a_temporal_datatype_does_not_support_are_refused() {
4048 4 : let w = windowed_props(&[
4049 4 : json!({"value": {"@type": "DateTime", "@value": "2020-01-01T00:00:00Z"}}),
4050 4 : ]);
4051 16 : for method in ["avg", "sum", "stddev", "sumsq"] {
4052 16 : assert!(
4053 16 : matches!(aggregate(&w, &[method]), Err(NgsiError::InvalidRequest(_))),
4054 : "DateTime must not be eligible for {method}"
4055 : );
4056 : }
4057 4 : let w = windowed_props(&[json!({"value": {"@type": "Time", "@value": "08:30:00Z"}})]);
4058 12 : for method in ["sum", "stddev", "sumsq"] {
4059 12 : assert!(
4060 12 : matches!(aggregate(&w, &[method]), Err(NgsiError::InvalidRequest(_))),
4061 : "Time must not be eligible for {method}"
4062 : );
4063 : }
4064 4 : }
4065 :
4066 : /// Table 4.5.19.1-1, JSON String column: `avg` is N/A. A string is a
4067 : /// JSON String whatever it spells, so a value that reads like a
4068 : /// time-of-day is averaged only when its datatype says it is a Time.
4069 : #[test]
4070 4 : fn a_json_string_has_no_average_however_it_reads() {
4071 4 : let w = windowed_props(&[json!({"value": "08:30:00Z"}), json!({"value": "09:30:00Z"})]);
4072 4 : assert!(matches!(
4073 4 : aggregate(&w, &["avg"]),
4074 : Err(NgsiError::InvalidRequest(_))
4075 : ));
4076 : // its own row of the table is unchanged: lexicographic min and max
4077 4 : let out = aggregate(&w, &["min", "max"]).expect("a string has min and max");
4078 4 : assert_eq!(out["speed"]["min"][0][0], json!("08:30:00Z"));
4079 4 : assert_eq!(out["speed"]["max"][0][0], json!("09:30:00Z"));
4080 4 : }
4081 :
4082 : /// The counting methods have no N/A cell in any of the three tables:
4083 : /// `totalCount` "the number of times the value has been updated" and
4084 : /// `distinctCount` "the count of distinct values", for every datatype
4085 : /// including the ones with no other method at all.
4086 : #[test]
4087 4 : fn every_datatype_is_counted() {
4088 20 : for members in [
4089 4 : json!({"value": {"@type": "DateTime", "@value": "2020-01-01T00:00:00Z"}}),
4090 4 : json!({"value": {"@type": "Date", "@value": "2020-01-01"}}),
4091 4 : json!({"value": {"@type": "Time", "@value": "08:30:00Z"}}),
4092 4 : json!({"vocab": "urn:ngsi-ld:Colour:red"}),
4093 4 : json!({"object": "urn:ngsi-ld:Car:1"}),
4094 4 : ] {
4095 20 : let w = windowed_props(&[members.clone(), members.clone()]);
4096 20 : let out = aggregate(&w, &["totalCount", "distinctCount"])
4097 20 : .unwrap_or_else(|e| panic!("{members} must be counted: {e:?}"));
4098 20 : assert_eq!(out["speed"]["totalCount"][0][0], json!(2));
4099 20 : assert_eq!(out["speed"]["distinctCount"][0][0], json!(1));
4100 : }
4101 4 : }
4102 : }
4103 :
4104 : #[cfg(test)]
4105 : mod clause_4_21 {
4106 : use super::*;
4107 : use antares_jsonld::Loader;
4108 :
4109 28 : fn params(kv: &[(&str, &str)]) -> HashMap<String, String> {
4110 28 : kv.iter()
4111 48 : .map(|(k, v)| ((*k).to_owned(), (*v).to_owned()))
4112 28 : .collect()
4113 28 : }
4114 :
4115 : /// 4.21 Projections: "pick, omit and attrs are mutually exclusive" holds
4116 : /// on the temporal representation exactly as it does on the current-state
4117 : /// one — the temporal operations define no exception to it, so any pair
4118 : /// is BadRequestData and each one alone parses.
4119 : #[test]
4120 4 : fn pick_omit_and_attrs_cannot_be_combined_on_a_temporal_query() {
4121 4 : let ctx = Loader::new().core();
4122 16 : for p in [
4123 4 : params(&[("pick", "a"), ("omit", "b")]),
4124 4 : params(&[("pick", "a"), ("attrs", "b")]),
4125 4 : params(&[("omit", "a"), ("attrs", "b")]),
4126 4 : params(&[("pick", "a"), ("omit", "b"), ("attrs", "c")]),
4127 4 : ] {
4128 16 : match parse_trepr(&p, &ctx) {
4129 16 : Err(NgsiError::BadRequestData(_)) => {}
4130 0 : other => panic!("must be BadRequestData, got {:?}", other.err()),
4131 : }
4132 : }
4133 12 : for p in [
4134 4 : params(&[("pick", "a")]),
4135 4 : params(&[("omit", "a")]),
4136 4 : params(&[("attrs", "a")]),
4137 4 : ] {
4138 12 : assert!(parse_trepr(&p, &ctx).is_ok());
4139 : }
4140 4 : }
4141 :
4142 : /// 4.21 on the core members of a temporal Entity: `pick` constrains them
4143 : /// strictly (only what is named survives) and `omit` drops a named member
4144 : /// only when the node carries no children — the same reading the
4145 : /// current-state representation applies, so the two never disagree about
4146 : /// whether `id` or `type` is in the answer.
4147 : #[test]
4148 4 : fn core_members_follow_the_same_projection_rule() {
4149 4 : let pick = crate::repr::parse_projection("id", &Loader::new().core()).expect("pick");
4150 4 : assert!(crate::repr::meta_projected(Some(&pick), None, "id"));
4151 4 : assert!(!crate::repr::meta_projected(Some(&pick), None, "type"));
4152 4 : let omit = crate::repr::parse_projection("type", &Loader::new().core()).expect("omit");
4153 4 : assert!(!crate::repr::meta_projected(None, Some(&omit), "type"));
4154 4 : assert!(crate::repr::meta_projected(None, Some(&omit), "id"));
4155 4 : assert!(crate::repr::meta_projected(None, None, "type"));
4156 4 : }
4157 : }
|
