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 : mod aggregate;
24 : mod filter;
25 : mod query;
26 : mod render;
27 : mod retrieve;
28 : #[cfg(test)]
29 : mod tests;
30 : mod write;
31 :
32 : use aggregate::*;
33 : pub(crate) use filter::*;
34 : pub use query::*;
35 : pub(crate) use render::*;
36 : pub use retrieve::*;
37 : pub use write::*;
38 :
39 : // ---------- temporal query params (4.11) ----------
40 :
41 : /// Windowed per-entity temporal data: filtered+ordered instances per attr.
42 : struct Windowed {
43 : attrs: std::collections::BTreeMap<String, Vec<Value>>,
44 : max_per_attr: usize,
45 : ts_min: Option<String>,
46 : ts_max: Option<String>,
47 : truncated: bool,
48 : /// some Attribute holds instances past this page in its order: the
49 : /// response links to the next page (6.4.7.3)
50 : more: bool,
51 : /// the fewest instances an Attribute the ceiling cut shortened still
52 : /// served (`truncate`): how far the next offsetN page may move
53 : cut_served: Option<usize>,
54 : /// Attributes their type records `onChange`: each value
55 : /// holds until the next row, so a window starts at the level carried in
56 : /// and an aggregation weighs a value by the time it held
57 : held: std::collections::HashSet<String>,
58 : }
59 :
60 : /// TS 104 176 6.4.7.3: which instances of each Attribute one response
61 : /// carries. firstN pages oldest first, lastN newest first, neither oldest
62 : /// first; offsetN skips that many in the chosen order.
63 : #[derive(Default, Clone, Copy, Debug, PartialEq)]
64 : pub(crate) struct InstancePage {
65 : /// lastN: newest first
66 : desc: bool,
67 : /// the firstN or lastN asked for
68 : n: Option<usize>,
69 : /// offsetN
70 : offset: usize,
71 : }
72 :
73 : impl InstancePage {
74 : /// firstN, lastN and offsetN, validated: N positive (5.2.21), offsetN
75 : /// not negative (Table 6.4.7.3-2), both at most i64::MAX (bound as a
76 : /// `bigint` they would wrap negative and silently select nothing).
77 : /// firstN asks for ascending order and lastN for descending; one page
78 : /// cannot be both, so the pair is BadRequestData.
79 2128 : fn parse(params: &HashMap<String, String>) -> Result<Self, NgsiError> {
80 6362 : let count = |name: &str, min: usize| -> Result<Option<usize>, NgsiError> {
81 6362 : params
82 6362 : .get(name)
83 6362 : .map(|v| {
84 448 : v.parse::<usize>()
85 448 : .ok()
86 448 : .filter(|n| *n >= min && *n <= i64::MAX as usize)
87 448 : .ok_or_else(|| NgsiError::BadRequestData(format!("invalid {name} {v:?}")))
88 448 : })
89 6362 : .transpose()
90 6362 : };
91 2128 : let (first, last) = (count("firstN", 1)?, count("lastN", 1)?);
92 2116 : if first.is_some() && last.is_some() {
93 6 : return Err(NgsiError::BadRequestData(
94 6 : "firstN pages ascending and lastN descending: give one of them (6.4.7.3)".into(),
95 6 : ));
96 2110 : }
97 : Ok(Self {
98 2110 : desc: last.is_some(),
99 2110 : n: first.or(last),
100 2110 : offset: count("offsetN", 0)?.unwrap_or(0),
101 : })
102 2128 : }
103 :
104 : /// The most instances of one Attribute a page holds: N, or the broker's
105 : /// ceiling when N is above it (6.4.7.3 "whichever is smaller"); with no
106 : /// N, the broker's default, at most the ceiling.
107 3872 : fn size(&self, default: usize, limit: usize) -> usize {
108 3872 : self.n.map_or(default.min(limit), |n| n.min(limit))
109 3872 : }
110 :
111 : /// How far one page moves offsetN: its size, or N itself for an
112 : /// aggregated representation, which the ceiling never cuts.
113 1890 : fn step(&self, aggregated: bool, default: usize, limit: usize) -> usize {
114 1890 : match (aggregated, self.n) {
115 0 : (true, Some(n)) => n,
116 1890 : _ => self.size(default, limit),
117 : }
118 1890 : }
119 :
120 : /// The per-Attribute cap a store may apply before the window: every
121 : /// instance up to the end of this page in its order, and one more, so
122 : /// the window still sees whether a next page exists. Without N the page
123 : /// is `default` instances, the broker's default page, when the caller
124 : /// pages instances at all (`None` for an aggregation, which reads the
125 : /// whole window): a chart's plain window read paged by offsetN used to
126 : /// rebuild every instance of the window per page.
127 2020 : fn store_cap(&self, default: Option<usize>) -> Option<i64> {
128 2020 : self.n.or(default).map(|n| {
129 1878 : n.saturating_add(self.offset)
130 1878 : .saturating_add(1)
131 1878 : .min(i64::MAX as usize) as i64
132 1878 : })
133 2020 : }
134 : }
135 :
136 : /// The page a read without firstN/lastN serves per Attribute, for the
137 : /// store's cap: none for an aggregation (it reads the whole window) or a
138 : /// tenant recording `onChange`, whose window starts at the level carried in.
139 2008 : fn default_page(st: &AppState, tenant: &TenantId, trepr: &TRepr) -> Option<usize> {
140 2008 : (!trepr.aggregated && !crate::history::records_on_change(st, tenant)).then(|| {
141 1782 : st.temporal_default_instances
142 1782 : .min(st.temporal_instance_limit)
143 1782 : })
144 2008 : }
145 :
146 : /// Parsed temporal representation params (options/format/lastN/pick/omit/
147 : /// datasetId/aggregation), fully validated up front.
148 : #[derive(Default, Clone)]
149 : pub(crate) struct TRepr {
150 : temporal_values: bool,
151 : aggregated: bool,
152 : sys: bool,
153 : page: InstancePage,
154 : pick: Option<Vec<crate::repr::ProjNode>>,
155 : omit: Option<Vec<crate::repr::ProjNode>>,
156 : dataset_id: Option<Vec<String>>,
157 : attrs: Option<Vec<String>>,
158 : aggr_methods: Vec<String>,
159 : aggr_period: AggrPeriod,
160 : lang: Option<String>,
161 : }
162 :
163 : #[derive(Clone, Copy, Debug, Default, PartialEq)]
164 : pub(crate) enum AggrPeriod {
165 : /// PT0S / absent: one bucket over the whole range
166 : #[default]
167 : Whole,
168 : Seconds(i64),
169 : /// 4.5.19.1: a period may mix date and time elements
170 : /// ("P3Y6M4DT12H30M5S"), so a month step carries the leftover seconds —
171 : /// months are not a fixed number of seconds and cannot be folded in.
172 : Months(u32, i64),
173 : }
174 :
175 : /// 5.6.13.4-5.6.15.4 shared forwarding: proxy registrations without the
176 : /// operation's support are an error of type Conflict and are never
177 : /// contacted; supporting registrations receive the forwarded request. None
178 : /// = no matching registrations (the operation stays purely local).
179 : /// `path_suffix` arrives with its client-controlled segments already
180 : /// percent-encoded (RFC 3986 clause 3.3); the entity id is encoded here.
181 : /// What the local `mutate` did, in the words the answer needs. 5.6.13,
182 : /// 5.6.14, 5.6.15 and 5.6.16 all answer the same three ways: 204 when the
183 : /// target was there, ResourceNotFound naming the Entity when the Entity was
184 : /// not, and ResourceNotFound naming the Attribute or the instance when the
185 : /// Entity was there but the target inside it was not.
186 : struct LocalWrite {
187 : /// `None` when the Temporal Evolution itself is absent.
188 : res: Option<Result<(), NgsiError>>,
189 : /// whether the operation's target inside it was there.
190 : found: bool,
191 : /// what a 404 names once the Entity itself was found.
192 : missing: String,
193 : /// what the 204 Part reports was done here.
194 : applied: &'static str,
195 : }
|
