Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Temporal representation: the 4.11 window, the 6.3.10 cut, paging links and the normalized/temporalValues body.
3 :
4 : use super::*;
5 :
6 : /// The most periods an aggregated `onChange` Attribute answers, rows and held gaps together.
7 : pub(super) const MAX_HELD_BUCKETS: usize = 10_000;
8 :
9 : /// 6.3.10: the ceiling is a CUT, not a label. An Attribute holding more
10 : /// instances than the broker serves at once is truncated to the ceiling in
11 : /// the query direction, and the 206 + Content-Range then describes the
12 : /// instances actually returned. This is what caps a lastN above the ceiling
13 : /// and what caps a request naming no temporal window at all. Aggregated
14 : /// representations (5.7.4.4) are computed over the whole evolution and are
15 : /// complete by construction, so they are never cut.
16 : ///
17 : /// The cut is ONE time boundary for the whole entity, not a per-attribute
18 : /// count: the partial content "shall" be the representation the
19 : /// Content-Range describes, so every attribute is trimmed to the tightest
20 : /// ceiling instant among the over-full ones (ties at that instant kept),
21 : /// and a client continuing from the advertised range-end misses no instance
22 : /// of any attribute. An attribute lying entirely beyond the boundary comes
23 : /// back empty on this page. ETSI 020_14 checks exactly that (the
24 : /// per-Attribute count failed it on every store).
25 : ///
26 : /// An Attribute the boundary cut shorter than the ceiling holds the next
27 : /// page's instances at a smaller offset than the boundary Attribute, so the
28 : /// next offsetN Link moves by the fewest instances a cut Attribute served
29 : /// (`cut_served`): it repeats, never skips. One cut to nothing leaves no
30 : /// offsetN that continues it, and the Link continues by time from the
31 : /// boundary instead (`interval_links`).
32 2018 : pub(super) fn truncate(w: &mut Windowed, timeprop: &str, descending: bool, limit: usize) {
33 2018 : if w.max_per_attr <= limit {
34 1750 : return;
35 268 : }
36 268 : w.truncated = true;
37 268 : w.more = true;
38 268 : w.max_per_attr = limit;
39 12586 : let key = |inst: &Value| inst.get(timeprop).and_then(Value::as_str).map(dt_key);
40 : // the tightest ceiling instant: earliest forwards, latest backwards
41 268 : let boundary = w
42 268 : .attrs
43 268 : .values()
44 464 : .filter(|insts| insts.len() > limit)
45 404 : .filter_map(|insts| insts.get(limit.saturating_sub(1)).and_then(key))
46 268 : .reduce(|a, b| if (b < a) != descending { b } else { a });
47 268 : let (mut ts_min, mut ts_max) = (None::<String>, None::<String>);
48 464 : for instances in w.attrs.values_mut() {
49 464 : let before = instances.len();
50 464 : match &boundary {
51 12182 : Some(bd) => instances.retain(|inst| {
52 12182 : key(inst).is_none_or(|k| if descending { k >= *bd } else { k <= *bd })
53 12182 : }),
54 0 : None => instances.truncate(limit),
55 : }
56 464 : if instances.len() < before {
57 : w.cut_served = Some(
58 436 : w.cut_served
59 436 : .map_or(instances.len(), |c| c.min(instances.len())),
60 : );
61 28 : }
62 8196 : for inst in instances.iter() {
63 8196 : if let Some(t) = inst.get(timeprop).and_then(Value::as_str) {
64 8196 : if ts_min.as_deref().is_none_or(|m| dt_key(t) < dt_key(m)) {
65 1132 : ts_min = Some(t.to_owned());
66 7064 : }
67 8196 : if ts_max.as_deref().is_none_or(|m| dt_key(t) > dt_key(m)) {
68 6276 : ts_max = Some(t.to_owned());
69 6276 : }
70 0 : }
71 : }
72 : }
73 268 : (w.ts_min, w.ts_max) = (ts_min, ts_max);
74 2018 : }
75 :
76 : /// Each instance of `w` replaced by its chain of versions in `chains` (the
77 : /// store's `versions`), the same measurement being the same Attribute,
78 : /// `observedAt` and `datasetId`. An instance the store keeps
79 : /// no version of stays as it is.
80 4 : pub(super) fn with_versions(w: &mut Windowed, chains: &Map<String, Value>) {
81 48 : let key = |i: &Value| {
82 48 : (
83 48 : i.get("observedAt").and_then(Value::as_str).map(dt_key),
84 48 : i.get("datasetId").cloned(),
85 48 : )
86 48 : };
87 8 : for (attr, instances) in &mut w.attrs {
88 8 : let Some(chain) = chains.get(attr).and_then(Value::as_array) else {
89 4 : continue;
90 : };
91 4 : *instances = instances
92 4 : .iter()
93 8 : .flat_map(|inst| {
94 8 : let own: Vec<Value> = chain
95 8 : .iter()
96 24 : .filter(|v| key(v) == key(inst))
97 8 : .cloned()
98 8 : .collect();
99 8 : if own.is_empty() {
100 4 : vec![inst.clone()]
101 : } else {
102 4 : own
103 : }
104 8 : })
105 4 : .collect();
106 : }
107 4 : }
108 :
109 : /// The instant a cut page ends at in its order: the boundary `truncate`
110 : /// kept every Attribute to.
111 1002 : pub(super) fn cut_boundary(w: &Windowed, desc: bool) -> Option<&str> {
112 1002 : if desc {
113 252 : w.ts_min.as_deref()
114 : } else {
115 750 : w.ts_max.as_deref()
116 : }
117 1002 : }
118 :
119 : /// 5.7.4.4 S4/S7: does a scope VALUE (string or array of strings) match the
120 : /// 4.19 Scope query? The 4.5.7 deletion sentinel never matches.
121 392 : pub(super) fn scope_value_matches(sq: &str, v: &Value) -> bool {
122 392 : if v.is_null() || v.as_str() == Some("urn:ngsi-ld:null") {
123 8 : return false;
124 384 : }
125 384 : crate::scope_matches(sq, &serde_json::json!({ "scope": v }))
126 392 : }
127 :
128 : /// 4.18 over a Temporal Evolution: "a given Scope is considered valid from
129 : /// the time it has been set until the time it has been explicitly removed by
130 : /// an update or delete operation" (example: annex C.5.16). Instance-shaped
131 : /// scope arrays become [set-time, next-set-time) validity intervals
132 : /// (set-time = observedAt‖modifiedAt‖createdAt — 4.5.6 mirrors them from the
133 : /// Core API change); a plain string/array scope is valid for all time.
134 : /// Returns only the intervals whose value matches `sq`; "" start = -inf,
135 : /// None end = +inf.
136 392 : pub(super) fn scope_match_intervals(doc: &Value, sq: &str) -> Vec<(String, Option<String>)> {
137 392 : match doc.get("scope") {
138 320 : Some(Value::Array(a)) if a.first().is_some_and(Value::is_object) => {
139 316 : let mut states: Vec<(&str, &Value)> = a
140 316 : .iter()
141 400 : .filter_map(|i| {
142 400 : scope_set_time(i).map(|t| (t, i.get("value").unwrap_or(&Value::Null)))
143 400 : })
144 316 : .collect();
145 316 : states.sort_by_key(|(t, _)| dt_key(t));
146 : // each state with the time the next one was set, the last open
147 316 : let next = states.iter().skip(1).map(Some).chain(std::iter::once(None));
148 316 : states
149 316 : .iter()
150 316 : .zip(next)
151 392 : .filter(|((_, value), _)| scope_value_matches(sq, value))
152 316 : .map(|((t, _), next)| ((*t).to_owned(), next.map(|(t, _)| (*t).to_owned())))
153 316 : .collect()
154 : }
155 12 : Some(_) if crate::scope_matches(sq, doc) => vec![(String::new(), None)],
156 70 : _ => Vec::new(),
157 : }
158 392 : }
159 :
160 : /// The time a temporal scope instance was set (4.5.6: observedAt is a copy
161 : /// of modifiedAt on Core-API changes; direct 5.6.11 input may carry any).
162 634 : pub(super) fn scope_set_time(i: &Value) -> Option<&str> {
163 634 : ["observedAt", "modifiedAt", "createdAt"]
164 634 : .iter()
165 674 : .find_map(|k| i.get(*k).and_then(Value::as_str))
166 634 : }
167 :
168 : // seven of the request's own plus the Attributes the type records onChange
169 : #[allow(clippy::too_many_arguments)]
170 2086 : pub(super) fn window(
171 2086 : doc: &Value,
172 2086 : tq: Option<&TemporalQ>,
173 2086 : page: InstancePage,
174 2086 : attrs_filter: Option<&Vec<String>>,
175 2086 : omit: Option<&Vec<crate::repr::ProjNode>>,
176 2086 : dataset: Option<&Vec<String>>,
177 2086 : timeprop: &str,
178 2086 : held: std::collections::HashSet<String>,
179 2086 : ) -> Windowed {
180 2086 : let mut w = Windowed {
181 2086 : attrs: std::collections::BTreeMap::new(),
182 2086 : max_per_attr: 0,
183 2086 : ts_min: None,
184 2086 : ts_max: None,
185 2086 : truncated: false,
186 2086 : cut_served: None,
187 2086 : more: false,
188 2086 : held,
189 2086 : };
190 2086 : let Some(obj) = doc.as_object() else { return w };
191 11908 : for (k, v) in obj {
192 : // 4.5.6: the Scope of a Temporal Evolution is represented as the
193 : // temporal representation of a Property — instance-shaped scope
194 : // arrays window like attributes (plain-string scope stays meta).
195 11908 : let scope_instances = k == "scope"
196 140 : && v.as_array()
197 140 : .is_some_and(|a| a.first().is_some_and(Value::is_object));
198 11908 : if (is_meta(k) && !scope_instances) || !projected(k, attrs_filter, omit) {
199 8700 : continue;
200 3208 : }
201 3208 : let mut instances = in_window(v, tq, dataset);
202 3208 : if w.held.contains(k) {
203 132 : carry_held(&mut instances, v, tq, dataset, timeprop);
204 3076 : }
205 3208 : if scope_instances {
206 122 : carry_scope(&mut instances, v, tq);
207 3086 : }
208 37064 : instances.sort_by(|a, b| {
209 : // Canonicalize before comparing: '.' sorts before 'Z', so a raw
210 : // string compare puts "…00.5Z" ahead of "…00Z" and lastN keeps the
211 : // wrong instant. 4.6.3 allows both fraction spellings.
212 37064 : let ta = a.get(timeprop).and_then(Value::as_str).unwrap_or("");
213 37064 : let tb = b.get(timeprop).and_then(Value::as_str).unwrap_or("");
214 37064 : dt_key(ta).cmp(&dt_key(tb))
215 37064 : });
216 3208 : w.more |= page_instances(&mut instances, page);
217 3208 : if instances.is_empty() {
218 52 : continue;
219 3156 : }
220 3156 : w.note(&instances, timeprop);
221 3156 : w.attrs.insert(k.clone(), instances);
222 : }
223 2086 : w
224 2086 : }
225 :
226 : impl Windowed {
227 : /// One Attribute's page joins the window: its size and time range.
228 3156 : fn note(&mut self, instances: &[Value], timeprop: &str) {
229 3156 : self.max_per_attr = self.max_per_attr.max(instances.len());
230 26616 : for inst in instances {
231 26616 : if let Some(t) = inst.get(timeprop).and_then(Value::as_str) {
232 26542 : if self.ts_min.as_deref().is_none_or(|m| dt_key(t) < dt_key(m)) {
233 8650 : self.ts_min = Some(t.to_owned());
234 17892 : }
235 26542 : if self.ts_max.as_deref().is_none_or(|m| dt_key(t) > dt_key(m)) {
236 15920 : self.ts_max = Some(t.to_owned());
237 15920 : }
238 74 : }
239 : }
240 3156 : }
241 : }
242 :
243 : /// Whether attrs and omit keep Attribute `k`.
244 3686 : fn projected(
245 3686 : k: &str,
246 3686 : attrs_filter: Option<&Vec<String>>,
247 3686 : omit: Option<&Vec<crate::repr::ProjNode>>,
248 3686 : ) -> bool {
249 3686 : attrs_filter.is_none_or(|want| want.iter().any(|w| w == k))
250 3262 : && omit.is_none_or(|omit| !omit.iter().any(|n| n.iri == k && n.children.is_none()))
251 3686 : }
252 :
253 48318 : fn of_dataset(dataset: Option<&Vec<String>>, inst: &Value) -> bool {
254 48318 : match (dataset, inst.get("datasetId")) {
255 48082 : (None, _) => true,
256 144 : (Some(want), Some(Value::String(have))) => want.iter().any(|w| w == have),
257 160 : (Some(want), None) => want.iter().any(|w| w == "@none"),
258 0 : _ => false,
259 : }
260 48318 : }
261 :
262 : /// The instances of one Attribute in the asked datasets and the window.
263 3208 : fn in_window(v: &Value, tq: Option<&TemporalQ>, dataset: Option<&Vec<String>>) -> Vec<Value> {
264 3208 : v.as_array()
265 3208 : .into_iter()
266 3208 : .flatten()
267 47076 : .filter(|inst| of_dataset(dataset, inst))
268 46940 : .filter(|inst| tq.is_none_or(|tq| tq.instance_matches(inst)))
269 3208 : .cloned()
270 3208 : .collect()
271 3208 : }
272 :
273 : /// An `onChange` Attribute held its last row
274 : /// before the window when the window opened, so that row comes along.
275 132 : fn carry_held(
276 132 : instances: &mut Vec<Value>,
277 132 : v: &Value,
278 132 : tq: Option<&TemporalQ>,
279 132 : dataset: Option<&Vec<String>>,
280 132 : timeprop: &str,
281 132 : ) {
282 132 : let Some(start) = tq.and_then(|t| match t.timerel.as_str() {
283 118 : "after" | "between" => Some(dt_key(&t.time_at)),
284 92 : _ => None,
285 118 : }) else {
286 106 : return;
287 : };
288 1252 : let key = |i: &&Value| i.get(timeprop).and_then(Value::as_str).map(dt_key);
289 26 : let carry = v
290 26 : .as_array()
291 26 : .into_iter()
292 26 : .flatten()
293 1242 : .filter(|inst| of_dataset(dataset, inst))
294 1242 : .filter(|i| key(i).is_some_and(|t| t < start))
295 26 : .max_by_key(key);
296 26 : if let Some(c) = carry {
297 6 : instances.push(c.clone());
298 20 : }
299 132 : }
300 :
301 : /// 4.18/C.5.16: the scope valid AT the window start was set at or
302 : /// before it — carry the latest pre-window instance into the
303 : /// representation (a temporal scope stays valid until replaced).
304 122 : fn carry_scope(instances: &mut Vec<Value>, v: &Value, tq: Option<&TemporalQ>) {
305 122 : let Some(start) = tq.and_then(|t| match t.timerel.as_str() {
306 116 : "after" | "between" => Some(t.time_at.as_str()),
307 14 : _ => None,
308 116 : }) else {
309 20 : return;
310 : };
311 102 : let carry = v
312 102 : .as_array()
313 102 : .into_iter()
314 102 : .flatten()
315 138 : .filter(|i| scope_set_time(i).is_some_and(|t| dt_key(t) < dt_key(start)))
316 102 : .max_by_key(|i| scope_set_time(i).map(dt_key))
317 102 : .cloned();
318 102 : if let Some(c) = carry {
319 88 : if !instances.contains(&c) {
320 88 : instances.push(c);
321 88 : }
322 14 : }
323 122 : }
324 :
325 : /// 6.4.7.3: the sorted instances cut to one page; true when the
326 : /// Attribute holds more past it.
327 3208 : fn page_instances(instances: &mut Vec<Value>, page: InstancePage) -> bool {
328 : // lastN pages newest first, firstN and neither oldest first
329 3208 : if page.desc {
330 474 : instances.reverse();
331 2734 : }
332 : // Table 6.4.7.3-2: an offsetN past the end "should be assumed to be
333 : // equal to the size of the result set, i.e. only the last element
334 : // of the result set is to be returned"
335 3208 : let skip = page.offset.min(instances.len().saturating_sub(1));
336 3208 : instances.drain(..skip);
337 514 : match page.n {
338 514 : Some(n) if instances.len() > n => {
339 190 : instances.truncate(n);
340 190 : true
341 : }
342 3018 : _ => false,
343 : }
344 3208 : }
345 :
346 : /// `Content-Range: date-time <start>-<end>/<size>` (Scorpio-parity semantics).
347 1838 : pub(super) fn content_range(
348 1838 : truncated: bool,
349 1838 : ts_min: Option<&str>,
350 1838 : ts_max: Option<&str>,
351 1838 : tq: Option<&TemporalQ>,
352 1838 : page: InstancePage,
353 1838 : ) -> Option<String> {
354 1838 : if !truncated {
355 1616 : return None;
356 222 : }
357 222 : let (data_min, data_max) = (ts_min?, ts_max?);
358 : // The window bound is the query's own when the query names one; matching
359 : // on the pair keeps the timerel and the query that produced it together,
360 : // so no arm can reach for a query that is not there.
361 222 : let named = tq.filter(|t| t.timerel != "any");
362 222 : let (start, end) = if !page.desc {
363 130 : let start = match named.map(|t| (t.timerel.as_str(), t)) {
364 90 : Some(("after" | "between", t)) => t.time_at.clone(),
365 52 : _ => data_min.to_owned(),
366 : };
367 130 : (start, data_max.to_owned())
368 : } else {
369 92 : let start = match named.map(|t| (t.timerel.as_str(), t)) {
370 72 : Some(("before", t)) => t.time_at.clone(),
371 44 : Some(("between", t)) => t.end_time_at.clone().unwrap_or_else(|| data_max.to_owned()),
372 32 : _ => data_max.to_owned(),
373 : };
374 92 : (start, data_min.to_owned())
375 : };
376 : // start/end bound the instances actually returned; the size is the length
377 : // of the complete representation the client asked for — the requested
378 : // firstN or lastN, or "*" when the window leaves it unknown.
379 222 : let size = page.n.map_or_else(|| "*".to_owned(), |n| n.to_string());
380 222 : Some(format!("date-time {start}-{end}/{size}"))
381 1838 : }
382 :
383 : /// TS 104 176 6.4.7.3: the Links to the neighbouring instance pages, each
384 : /// the request with offsetN moved by one page and typed with the response
385 : /// media type. Ascending pages run forward in time, so their next page is
386 : /// `intervalafter` and their previous one `intervalbefore`; lastN pages run
387 : /// backward and swap the two. `step` is the page size served per Attribute.
388 : ///
389 : /// A page the ceiling cut moves its next Link by `cut`, the fewest
390 : /// instances a cut Attribute served (see `truncate`). When that is none at
391 : /// all, no offsetN continues every Attribute, and the next Link asks for the
392 : /// time past `boundary` instead (the window's end the cut moved): `after`
393 : /// it ascending, which repeats the instances at the boundary instant once
394 : /// (4.11 `after` is inclusive), `before` it descending, which is exact.
395 1890 : pub(super) fn interval_links(
396 1890 : params: &HashMap<String, String>,
397 1890 : path: &str,
398 1890 : accept: Accept,
399 1890 : page: InstancePage,
400 1890 : more: bool,
401 1890 : step: usize,
402 1890 : cut: Option<(usize, &str)>,
403 1890 : ) -> Vec<String> {
404 1890 : let (forward, back) = match page.desc {
405 1626 : false => ("intervalafter", "intervalbefore"),
406 264 : true => ("intervalbefore", "intervalafter"),
407 : };
408 : // the request with `set` replacing its own values of those names
409 1890 : let link = |set: Vec<(&str, String)>, rel: &str| {
410 476 : let mut qp: Vec<String> = params
411 476 : .iter()
412 1510 : .filter(|(k, _)| k.as_str() != "offsetN" && !set.iter().any(|(s, _)| s == k))
413 1272 : .map(|(k, v)| (k.as_str(), v.as_str()))
414 516 : .chain(set.iter().map(|(k, v)| (*k, v.as_str())))
415 1788 : .map(|(k, v)| format!("{k}={}", crate::paging::query_value(v)))
416 476 : .collect();
417 476 : qp.sort();
418 476 : format!(
419 : "<{path}?{}>; rel=\"{rel}\";type=\"{}\"",
420 476 : qp.join("&"),
421 476 : crate::paging::link_media_type(accept)
422 : )
423 476 : };
424 1890 : let offset = |n: usize| vec![("offsetN", n.to_string())];
425 1890 : let mut links = Vec::new();
426 1890 : if more {
427 368 : match cut {
428 166 : None => links.push(link(offset(page.offset.saturating_add(step)), forward)),
429 36 : Some((0, boundary)) => {
430 36 : links.push(link(resume_after(params, page.desc, boundary), forward))
431 : }
432 166 : Some((served, _)) => {
433 166 : links.push(link(
434 166 : offset(page.offset.saturating_add(served.min(step))),
435 166 : forward,
436 166 : ));
437 166 : }
438 : }
439 1522 : }
440 1890 : if page.offset > 0 {
441 108 : links.push(link(offset(page.offset.saturating_sub(step)), back));
442 1782 : }
443 1890 : links
444 1890 : }
445 :
446 : /// Ascending, a page whose boundary instant holds a whole page of
447 : /// one Attribute's instances links to itself; offsetN by time would fix it.
448 : ///
449 : /// The request's time window moved past `boundary` in the page order: the
450 : /// window's other end stays (4.11: `between` is inclusive of `timeAt`,
451 : /// exclusive of `endTimeAt`).
452 68 : pub(super) fn resume_after(
453 68 : params: &HashMap<String, String>,
454 68 : desc: bool,
455 68 : boundary: &str,
456 68 : ) -> Vec<(&'static str, String)> {
457 68 : let rel = params.get("timerel").map(String::as_str);
458 68 : let (at, end) = (params.get("timeAt"), params.get("endTimeAt"));
459 68 : let b = boundary.to_owned();
460 68 : let between = |from: String, to: String| {
461 20 : vec![
462 20 : ("timerel", "between".to_owned()),
463 20 : ("timeAt", from),
464 20 : ("endTimeAt", to),
465 : ]
466 20 : };
467 68 : match (desc, rel, at, end) {
468 24 : (false, Some("between"), _, Some(e)) | (false, Some("before"), Some(e), _) => {
469 8 : between(b, e.clone())
470 : }
471 32 : (false, ..) => vec![("timerel", "after".to_owned()), ("timeAt", b)],
472 20 : (true, Some("between" | "after"), Some(a), _) => between(a.clone(), b),
473 16 : (true, ..) => vec![("timerel", "before".to_owned()), ("timeAt", b)],
474 : }
475 68 : }
476 :
477 : /// Render one temporal entity from its windowed data.
478 2052 : pub(super) fn present_temporal(
479 2052 : doc: &Value,
480 2052 : w: &Windowed,
481 2052 : ctx: &Context,
482 2052 : r: &TRepr,
483 2052 : tq: Option<&TemporalQ>,
484 2052 : timeprop: &str,
485 2052 : ) -> Result<Value, NgsiError> {
486 2052 : let Some(obj) = doc.as_object() else {
487 0 : return Ok(doc.clone());
488 : };
489 2052 : let mut out = present_meta(obj, ctx, r);
490 2052 : if r.aggregated {
491 106 : for (k, v) in render_aggregated(w, tq, r, ctx, timeprop)? {
492 82 : out.insert(k, v);
493 82 : }
494 78 : return Ok(Value::Object(out));
495 1962 : }
496 2998 : for (k, instances) in &w.attrs {
497 : // Table 6.18.3.2-1 / 6.19.3.1 `lang`: each LanguageProperty
498 : // instance becomes a Property in the chosen language (4.15) before
499 : // either representation renders it.
500 : let reduced: Vec<Value>;
501 2998 : let instances: &[Value] = match &r.lang {
502 48 : Some(lang) => {
503 48 : reduced = instances
504 48 : .iter()
505 72 : .map(|inst| {
506 72 : let mut inst = inst.clone();
507 72 : if let Some(o) = inst.as_object_mut() {
508 72 : crate::repr::apply_lang(o, lang);
509 72 : }
510 72 : inst
511 72 : })
512 48 : .collect();
513 48 : &reduced
514 : }
515 2950 : None => instances,
516 : };
517 2998 : let rendered = if instances.is_empty() {
518 : // gap-cut leftovers render as empty arrays
519 36 : Value::Array(vec![])
520 2962 : } else if r.temporal_values {
521 304 : simplified_attr(instances, ctx, timeprop)
522 : } else {
523 : Value::Array(
524 2658 : instances
525 2658 : .iter()
526 7586 : .map(|inst| normalized_instance(inst, ctx, r.sys))
527 2658 : .collect(),
528 : )
529 : };
530 2998 : out.insert(ctx.compact_iri(k), rendered);
531 : }
532 1962 : Ok(Value::Object(out))
533 2052 : }
534 :
535 : /// The Entity's own members (id, type, scope, the system attributes) as
536 : /// sysAttrs, pick and omit admit them.
537 2052 : fn present_meta(obj: &Map<String, Value>, ctx: &Context, r: &TRepr) -> Map<String, Value> {
538 2052 : let mut out = Map::new();
539 11852 : for (k, v) in obj {
540 11852 : let scope_instances = k == "scope"
541 140 : && v.as_array()
542 140 : .is_some_and(|a| a.first().is_some_and(Value::is_object));
543 11852 : if !is_meta(k) || scope_instances {
544 3662 : continue;
545 8190 : }
546 8190 : match k.as_str() {
547 : // Table 6.3.11-1: `sysAttrs` is what admits "the system
548 : // generated temporal attributes createdAt, modifiedAt and
549 : // the system temporal attribute expiresAt … In the case of
550 : // temporal representations, also the system generated
551 : // temporal attribute deletedAt". Without it none of them is
552 : // in the payload — the same set `repr.rs` gates on the
553 : // current-state path.
554 8190 : "createdAt" | "modifiedAt" | "expiresAt" | "deletedAt" if !r.sys => continue,
555 4336 : _ => {}
556 : }
557 4336 : if !crate::repr::meta_projected(r.pick.as_deref(), r.omit.as_deref(), k) {
558 128 : continue;
559 4208 : }
560 4208 : if k == "type" {
561 1958 : out.insert("type".into(), antares_jsonld::compact_types(v, ctx));
562 2250 : } else {
563 2250 : out.insert(k.clone(), v.clone());
564 2250 : }
565 : }
566 2052 : out
567 2052 : }
568 :
569 : /// 4.5.9 temporalValues: the instances grouped by datasetId, one object
570 : /// per group, the group alone when there is one.
571 304 : fn simplified_attr(instances: &[Value], ctx: &Context, timeprop: &str) -> Value {
572 304 : let mut groups: Vec<(Option<String>, Vec<&Value>)> = Vec::new();
573 840 : for inst in instances {
574 840 : let ds = inst
575 840 : .get("datasetId")
576 840 : .and_then(Value::as_str)
577 840 : .map(String::from);
578 840 : match groups.iter_mut().find(|(g, _)| *g == ds) {
579 524 : Some((_, list)) => list.push(inst),
580 316 : None => groups.push((ds, vec![inst])),
581 : }
582 : }
583 304 : let mut rendered: Vec<Value> = groups
584 304 : .iter()
585 316 : .map(|(ds, list)| simplified_group(ds.as_deref(), list, ctx, timeprop))
586 304 : .collect();
587 304 : if rendered.len() == 1 {
588 296 : rendered.remove(0)
589 : } else {
590 8 : Value::Array(rendered)
591 : }
592 304 : }
593 :
594 : /// One datasetId's instances as `{type, datasetId?, <member>: [[v, t]]}`.
595 316 : fn simplified_group(ds: Option<&str>, list: &[&Value], ctx: &Context, timeprop: &str) -> Value {
596 316 : let atype = list
597 316 : .first()
598 316 : .and_then(|i| i.get("type"))
599 316 : .cloned()
600 316 : .unwrap_or_else(|| Value::String("Property".into()));
601 316 : let values: Vec<Value> = list
602 316 : .iter()
603 840 : .map(|inst| {
604 840 : let t = inst.get(timeprop).cloned().unwrap_or(Value::Null);
605 840 : Value::Array(vec![simplified_value(inst, ctx), t])
606 840 : })
607 316 : .collect();
608 316 : let mut o = Map::new();
609 : // 4.5.9: the simplified member name follows the attribute type
610 316 : let member = match atype.as_str() {
611 316 : Some("Relationship") => "objects",
612 310 : Some("LanguageProperty") => "languageMaps",
613 300 : Some("VocabProperty") => "vocabs",
614 296 : Some("JsonProperty") => "jsons",
615 292 : Some("ListProperty") => "valueLists",
616 282 : Some("ListRelationship") => "objectLists",
617 272 : _ => "values",
618 : };
619 316 : o.insert("type".into(), atype);
620 316 : if let Some(ds) = ds {
621 32 : o.insert("datasetId".into(), Value::String(ds.to_owned()));
622 284 : }
623 316 : o.insert(member.into(), Value::Array(values));
624 316 : Value::Object(o)
625 316 : }
626 :
627 : /// 4.5.9: Property/Relationship pairs carry the bare value/object; other
628 : /// attribute kinds wrap it under their member name.
629 840 : fn simplified_value(inst: &Value, ctx: &Context) -> Value {
630 840 : if let Some(v) = inst.get("value") {
631 : // 4.5.9: "the first element shall be a
632 : // Property value" — the one the instance
633 : // holds. A f64 round trip would retype an
634 : // integer and drop a digit past 2^53.
635 768 : v.clone()
636 72 : } else if let Some(o) = inst.get("object") {
637 6 : o.clone()
638 66 : } else if let Some(lm) = inst.get("languageMap") {
639 18 : serde_json::json!({"languageMap": lm})
640 48 : } else if let Some(j) = inst.get("json") {
641 12 : serde_json::json!({"json": j})
642 36 : } else if let Some(vv) = inst.get("vocab") {
643 12 : serde_json::json!({"vocab": compact_vocab(vv, ctx)})
644 24 : } else if let Some(l) = inst.get("valueList") {
645 : // 4.5.9 p.63 EXAMPLE 3: the pair's first element
646 : // is the BARE ordered array, not a {"valueList"}
647 : // wrapper — unlike languageMap/
648 : // json/vocab, which the clause does wrap
649 14 : l.clone()
650 10 : } else if let Some(l) = inst.get("objectList") {
651 : // 4.5.9 p.65: same bare form for ListRelationship
652 10 : l.clone()
653 : } else {
654 0 : Value::Null
655 : }
656 840 : }
657 :
658 12 : fn compact_vocab(vv: &Value, ctx: &Context) -> Value {
659 12 : match vv {
660 8 : Value::String(iri) => Value::String(ctx.compact_iri(iri)),
661 4 : Value::Array(a) => Value::Array(
662 4 : a.iter()
663 8 : .map(|s| match s {
664 8 : Value::String(iri) => Value::String(ctx.compact_iri(iri)),
665 0 : o => o.clone(),
666 8 : })
667 4 : .collect(),
668 : ),
669 0 : o => o.clone(),
670 : }
671 12 : }
672 :
673 : /// One instance in the normalized representation, its system attributes
674 : /// gone unless sysAttrs asked for them.
675 7586 : fn normalized_instance(inst: &Value, ctx: &Context, sys: bool) -> Value {
676 7586 : let mut ci = inst.clone();
677 7586 : if let Some(o) = ci.as_object_mut().filter(|_| !sys) {
678 7192 : o.remove("createdAt");
679 7192 : o.remove("modifiedAt");
680 : // 6.3.11: expiresAt is sysAttrs-gated too
681 7192 : o.remove("expiresAt");
682 : // ... and so are the sub-Attributes' own (4.8)
683 28536 : for sub in o.values_mut() {
684 28536 : let subs = match sub {
685 70 : Value::Array(a) => a.iter_mut().collect(),
686 28466 : other => vec![other],
687 : };
688 28536 : for so in subs.into_iter().filter_map(Value::as_object_mut) {
689 148 : if so.contains_key("type") {
690 88 : so.remove("createdAt");
691 88 : so.remove("modifiedAt");
692 88 : }
693 : }
694 : }
695 394 : }
696 7586 : compact_instance(&ci, ctx)
697 7586 : }
698 :
699 156 : pub(crate) fn parse_aggr_period(s: &str) -> Option<AggrPeriod> {
700 156 : let d = antares_model::parse_iso_duration(s)?;
701 : // saturating: an absurd magnitude must not panic (debug) or wrap
702 : // (release) — f64→int `as` casts already saturate, guard the ops.
703 128 : let months = (d.years as u32)
704 128 : .saturating_mul(12)
705 128 : .saturating_add(d.months as u32);
706 128 : let secs = [
707 128 : (d.weeks, 604_800.0),
708 128 : (d.days, 86_400.0),
709 128 : (d.hours, 3_600.0),
710 128 : (d.minutes, 60.0),
711 128 : (d.seconds, 1.0),
712 128 : ]
713 128 : .into_iter()
714 640 : .fold(0i64, |acc, (n, per)| acc.saturating_add((n * per) as i64));
715 128 : Some(match (months, secs) {
716 40 : (0, 0) => AggrPeriod::Whole,
717 64 : (0, sc) => AggrPeriod::Seconds(sc),
718 24 : (m, sc) => AggrPeriod::Months(m, sc),
719 : })
720 156 : }
721 :
722 : pub(super) const AGGR_METHODS: &[&str] = &[
723 : "totalCount",
724 : "distinctCount",
725 : "sum",
726 : "avg",
727 : "min",
728 : "max",
729 : "stddev",
730 : "sumsq",
731 : ];
732 :
733 : /// 5.7.3.4 / 5.7.4.4: "If projection attributes are present and indicate the
734 : /// use of Linked Entity retrieval, an error of type BadRequestData shall be
735 : /// raised." Unconditional on both temporal consumption operations, because
736 : /// neither defines a join; only the clause number in the message differs.
737 2052 : pub(super) fn reject_linked_projection(trepr: &TRepr, clause: &str) -> Result<(), NgsiError> {
738 4084 : let depth = |p: &Option<Vec<crate::repr::ProjNode>>| {
739 4084 : p.as_deref().map(crate::repr::proj_depth).unwrap_or(0)
740 4084 : };
741 : // attrs=owner{object} names a Linked Entity projection too (4.21)
742 2052 : let linked_attrs = trepr
743 2052 : .attrs
744 2052 : .as_ref()
745 2052 : .is_some_and(|a| a.iter().any(|t| t.contains('{')));
746 2052 : if depth(&trepr.pick) > 0 || depth(&trepr.omit) > 0 || linked_attrs {
747 28 : return Err(NgsiError::BadRequestData(format!(
748 28 : "temporal projection must not use Linked Entity selection ({clause})"
749 28 : )));
750 2024 : }
751 2024 : Ok(())
752 2052 : }
753 :
754 2184 : pub(crate) fn parse_trepr(
755 2184 : params: &HashMap<String, String>,
756 2184 : ctx: &Context,
757 2184 : ) -> Result<TRepr, NgsiError> {
758 2184 : let mut r = TRepr {
759 2184 : lang: params.get("lang").cloned(),
760 2184 : ..TRepr::default()
761 2184 : };
762 2184 : if let Some(opts) = params.get("options") {
763 334 : apply_options(&mut r, opts)?;
764 1850 : }
765 : // format wins over options on conflict (6.3.7)
766 2172 : if let Some(f) = params.get("format") {
767 58 : apply_format(&mut r, f)?;
768 2114 : }
769 2168 : crate::repr::check_projection_exclusive(params)?;
770 2128 : if let Some(pck) = params.get("pick") {
771 72 : r.pick = Some(crate::repr::parse_projection(pck, ctx)?);
772 2056 : }
773 2128 : if let Some(o) = params.get("omit") {
774 42 : r.omit = Some(crate::repr::parse_projection(o, ctx)?);
775 2086 : }
776 2128 : if let Some(a) = params.get("attrs") {
777 218 : r.attrs = Some(a.split(',').map(|t| ctx.expand_key(t.trim())).collect());
778 1918 : }
779 2128 : r.dataset_id = params
780 2128 : .get("datasetId")
781 2128 : .map(|s| s.split(',').map(|d| d.trim().to_owned()).collect());
782 2128 : r.page = InstancePage::parse(params)?;
783 2104 : if let Some(m) = params.get("aggrMethods") {
784 148 : r.aggr_methods = parse_aggr_methods(m)?;
785 : // aggrMethods implies aggregation UNLESS an explicit format says otherwise
786 132 : if !params.contains_key("format") {
787 100 : r.aggregated = true;
788 100 : }
789 1956 : }
790 2088 : if r.aggregated && r.aggr_methods.is_empty() {
791 4 : return Err(NgsiError::BadRequestData(
792 4 : "aggregatedValues requires aggrMethods (4.5.19)".into(),
793 4 : ));
794 2084 : }
795 2084 : if let Some(d) = params.get("aggrPeriodDuration") {
796 92 : r.aggr_period = checked_aggr_period(d)?;
797 1992 : }
798 2064 : Ok(r)
799 2184 : }
800 :
801 334 : fn apply_options(r: &mut TRepr, opts: &str) -> Result<(), NgsiError> {
802 342 : for o in opts.split(',') {
803 342 : match o.trim() {
804 342 : "sysAttrs" => r.sys = true,
805 234 : "temporalValues" => r.temporal_values = true,
806 104 : "aggregatedValues" => r.aggregated = true,
807 12 : "normalized" => {}
808 12 : other => {
809 12 : return Err(NgsiError::InvalidRequest(format!(
810 12 : "unsupported options value {other:?}"
811 12 : )))
812 : }
813 : }
814 : }
815 322 : Ok(())
816 334 : }
817 :
818 58 : fn apply_format(r: &mut TRepr, f: &str) -> Result<(), NgsiError> {
819 58 : (r.temporal_values, r.aggregated) = match f {
820 58 : "temporalValues" => (true, false),
821 40 : "aggregatedValues" => (false, true),
822 4 : "normalized" => (false, false),
823 4 : other => {
824 4 : return Err(NgsiError::InvalidRequest(format!(
825 4 : "unsupported format value {other:?}"
826 4 : )))
827 : }
828 : };
829 54 : Ok(())
830 58 : }
831 :
832 148 : fn parse_aggr_methods(m: &str) -> Result<Vec<String>, NgsiError> {
833 148 : m.split(',')
834 204 : .map(|method| {
835 204 : let method = method.trim();
836 204 : if !AGGR_METHODS.contains(&method) {
837 16 : return Err(NgsiError::BadRequestData(format!(
838 16 : "invalid aggrMethods value {method:?} (4.5.19)"
839 16 : )));
840 188 : }
841 188 : Ok(method.to_owned())
842 204 : })
843 148 : .collect()
844 148 : }
845 :
846 : /// 4.11: the value space ends where duration arithmetic does —
847 : /// beyond ~100 years chrono::Duration::seconds is out of bounds
848 : /// (a panic, i.e. a remote 500), so such periods are rejected.
849 : /// Both components are bounded at ~100 years, the months in their own
850 : /// unit since a month is not a fixed number of seconds.
851 92 : fn checked_aggr_period(d: &str) -> Result<AggrPeriod, NgsiError> {
852 92 : let p = parse_aggr_period(d)
853 92 : .ok_or_else(|| NgsiError::BadRequestData(format!("invalid aggrPeriodDuration {d:?}")))?;
854 76 : let (months, secs) = match p {
855 20 : AggrPeriod::Whole => (0, 0),
856 56 : AggrPeriod::Seconds(sc) => (0, sc),
857 0 : AggrPeriod::Months(m, sc) => (m, sc),
858 : };
859 76 : if months > 1200 || secs > 86_400 * 366 * 100 {
860 4 : return Err(NgsiError::BadRequestData(format!(
861 4 : "aggrPeriodDuration {d:?} is out of range"
862 4 : )));
863 72 : }
864 72 : Ok(p)
865 92 : }
866 :
867 : /// The attribute-selection set for windowing: attrs= or pick=.
868 1940 : pub(super) fn selection(r: &TRepr) -> Option<Vec<String>> {
869 1940 : if let Some(a) = &r.attrs {
870 194 : return Some(a.clone());
871 1746 : }
872 : // core-member picks (id/type/…) are presentation-only, not attr selection
873 1746 : r.pick.as_ref().map(|p| {
874 48 : p.iter()
875 66 : .filter(|n| !is_meta(&n.raw))
876 48 : .map(|n| n.iri.clone())
877 48 : .collect()
878 48 : })
879 1940 : }
|
