Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Query temporal evolution (5.7.4) and the batch temporal query (6.24).
3 :
4 : use super::*;
5 :
6 : // ---------- GET /temporal/entities/ (5.7.4) ----------
7 :
8 1034 : pub async fn query_temporal(
9 1034 : State(st): State<AppState>,
10 1034 : CleanParams(params): CleanParams,
11 1034 : headers: HeaderMap,
12 1034 : ) -> Response {
13 1034 : match query_temporal_outer(&st, params, &headers).await {
14 854 : Ok(r) => r,
15 180 : Err(e) => e.into_response(),
16 : }
17 1034 : }
18 :
19 : /// 5.7.4.4 EntityMap usage on the temporal query: a live map referenced by
20 : /// the NGSILD-EntityMap header fixes the result set to the map's Entities
21 : /// (5.5.14) and its location is echoed; an unknown or expired reference
22 : /// means "a new EntityMap shall be created" (the entityMap=true branch,
23 : /// answering 201 + the fresh location).
24 1034 : pub(super) async fn query_temporal_outer(
25 1034 : st: &AppState,
26 1034 : mut params: HashMap<String, String>,
27 1034 : headers: &HeaderMap,
28 1034 : ) -> ApiResult<Response> {
29 1034 : let tenant = tenant_from(headers)?;
30 1034 : let named = crate::entities::ReadMembers::of(st, headers, ¶ms).await;
31 1034 : let filter = gate!(
32 : st, &tenant, headers, "5.7.4",
33 : scope_q: params.get("scopeQ").map(String::as_str),
34 : q: named.q.as_ref(),
35 : types: &named.types,
36 : attrs: &named.attrs,
37 : geo: named.geo.as_ref(),
38 : )
39 1034 : .await?;
40 1032 : let Some(map_ref) = single_header(headers, "NGSILD-EntityMap")? else {
41 1010 : return query_temporal_inner(st, ¶ms, headers, &filter).await;
42 : };
43 20 : let map_id = map_ref.rsplit('/').next().unwrap_or(&map_ref).to_owned();
44 20 : let Some(mut map) = crate::entity_map::map_if_accessible(st, &tenant, headers, &map_id).await
45 : else {
46 6 : params.insert("entityMap".into(), "true".into());
47 6 : return query_temporal_inner(st, ¶ms, headers, &filter).await;
48 : };
49 14 : params.remove("entityMap");
50 : // 5.5.9.3: the map fixes the candidate set and the request's own filters
51 : // narrow it, so `id=` on this request selects from the map rather than
52 : // replacing it.
53 14 : let candidates = crate::entity_map::candidate_ids(&map, ¶ms);
54 : // "filters shall be rechecked before returning results" and "Entities not
55 : // or no longer fitting the query shall be removed from the Entity map
56 : // during pagination" — so the recheck asks about the map's OWN Entities,
57 : // in bounded chunks. Asking the whole Tenant instead judged, and then
58 : // deleted, entries this request never asked about, and lost every
59 : // candidate past the first page of the recheck. Pruning is judgeable only
60 : // for "@none" (local) entries: a remote-backed id may merely have an
61 : // unreachable source right now (5.5.14). Known cost: this recheck is a
62 : // second temporal query per map-using request, same shape as the entity
63 : // query's filter re-run.
64 14 : let mut matching: std::collections::HashSet<String> = std::collections::HashSet::new();
65 14 : let id_headers = ids_shown(headers);
66 14 : for chunk in candidates.chunks(st.max_limit.max(1)) {
67 14 : let mut eff = params.clone();
68 42 : for k in ["limit", "offset", "count"] {
69 42 : eff.remove(k);
70 42 : }
71 14 : eff.insert("limit".into(), st.max_limit.to_string());
72 14 : eff.insert("id".into(), chunk.join(","));
73 14 : show_ids(&mut eff);
74 14 : let resp = query_temporal_inner(st, &eff, &id_headers, &filter).await?;
75 14 : let bytes = axum::body::to_bytes(resp.into_body(), st.max_answer_bytes)
76 14 : .await
77 14 : .map_err(|_| NgsiError::TooManyResults(crate::bounds::ANSWER_TOO_LARGE.into()))?;
78 12 : matching.extend(
79 12 : serde_json::from_slice::<Value>(&bytes)
80 12 : .ok()
81 12 : .and_then(|v| v.as_array().cloned())
82 12 : .unwrap_or_default()
83 12 : .iter()
84 12 : .filter_map(|d| d.get("id").and_then(Value::as_str).map(str::to_owned)),
85 : );
86 : }
87 12 : let mut stale: Vec<String> = Vec::new();
88 12 : if let Some(emap) = map.get_mut("entityMap").and_then(Value::as_object_mut) {
89 12 : stale = candidates
90 12 : .iter()
91 12 : .filter(|eid| {
92 12 : emap.get(eid.as_str())
93 12 : .and_then(Value::as_array)
94 12 : .is_some_and(|a| matches!(a.as_slice(), [only] if only == "@none"))
95 12 : && !matching.contains(eid.as_str())
96 12 : })
97 12 : .cloned()
98 12 : .collect();
99 12 : for k in &stale {
100 0 : emap.remove(k);
101 0 : }
102 0 : }
103 12 : crate::entity_map::map_prune(st, &tenant, &map_id, &stale).await?;
104 : // fix the query to the candidates that survived the recheck (5.5.14)
105 12 : let ids: Vec<&str> = candidates
106 12 : .iter()
107 12 : .filter(|id| {
108 12 : map.get("entityMap")
109 12 : .and_then(|m| m.get(id.as_str()))
110 12 : .is_some()
111 12 : })
112 12 : .map(String::as_str)
113 12 : .collect();
114 12 : params.insert(
115 12 : "id".into(),
116 12 : if ids.is_empty() {
117 0 : "urn:ngsi-ld:entitymap:empty".to_owned()
118 : } else {
119 12 : ids.join(",")
120 : },
121 : );
122 12 : let mut resp = query_temporal_inner(st, ¶ms, headers, &filter).await?;
123 12 : if let Ok(v) = format!("/ngsi-ld/v1/entityMaps/{map_id}").parse() {
124 12 : resp.headers_mut().insert("NGSILD-EntityMap", v);
125 12 : }
126 12 : Ok(resp)
127 1034 : }
128 :
129 : /// The EntityMap walks above read the matching Entities' ids off a rendered
130 : /// answer, so it has to show them: a `pick` without `id` or an `omit=id`
131 : /// rendered Entities with none, and the walk read the map as empty. `id`
132 : /// is added to `pick` and taken out of `omit`; every other term stays,
133 : /// since `pick`'s and `omit`'s Attributes decide which Entities the window
134 : /// keeps.
135 66 : fn show_ids(params: &mut HashMap<String, String>) {
136 66 : if let Some(pick) = params.get_mut("pick") {
137 6 : if !top_level_terms(pick).contains(&"id") {
138 4 : *pick = format!("id,{pick}");
139 4 : }
140 60 : }
141 66 : if let Some(omit) = params.get("omit") {
142 4 : let rest: Vec<&str> = top_level_terms(omit)
143 4 : .into_iter()
144 6 : .filter(|t| *t != "id")
145 4 : .collect();
146 4 : if rest.is_empty() {
147 2 : params.remove("omit");
148 2 : } else {
149 2 : let rest = rest.join(",");
150 2 : params.insert("omit".into(), rest);
151 2 : }
152 62 : }
153 66 : }
154 :
155 : /// The terms of a 4.21 projection outside any `{…}`, trimmed. 4.21 writes
156 : /// the disjunction as `,` or `|` (`orOp`), and `parse_projection` splits on
157 : /// both: split here on `,` alone, `pick=id|speed` was one term, gained a
158 : /// second `id` and was refused as a duplicate member.
159 10 : fn top_level_terms(projection: &str) -> Vec<&str> {
160 10 : let (mut terms, mut depth, mut start) = (Vec::new(), 0usize, 0usize);
161 56 : for (i, c) in projection.char_indices() {
162 0 : match c {
163 0 : '{' => depth = depth.saturating_add(1),
164 0 : '}' => depth = depth.saturating_sub(1),
165 4 : ',' | '|' if depth == 0 => {
166 4 : terms.push(projection.get(start..i).unwrap_or_default().trim());
167 4 : start = i.saturating_add(1);
168 4 : }
169 52 : _ => {}
170 : }
171 : }
172 10 : terms.push(projection.get(start..).unwrap_or_default().trim());
173 10 : terms
174 10 : }
175 :
176 : /// The request's headers with the answer asked for as plain JSON, whose
177 : /// Entities carry their `id` at the top: a GeoJSON FeatureCollection is no
178 : /// array, and the walk read it as no Entity at all.
179 66 : fn ids_shown(headers: &HeaderMap) -> HeaderMap {
180 66 : let mut h = headers.clone();
181 66 : h.insert(
182 66 : axum::http::header::ACCEPT,
183 66 : axum::http::HeaderValue::from_static("application/json"),
184 : );
185 66 : h
186 66 : }
187 :
188 : /// The 5.7.4.4 preconditions of a temporal query, before any store is
189 : /// touched: the filter must qualify, Linked Entity conditions are not
190 : /// defined for it, a context source filter must parse, and ordering may
191 : /// name only `id` and only where the execution stays local (4.23.1).
192 1222 : pub(super) async fn check_temporal_query(
193 1222 : st: &AppState,
194 1222 : params: &HashMap<String, String>,
195 1222 : headers: &HeaderMap,
196 1222 : tenant: &TenantId,
197 1222 : ctx: &Context,
198 1222 : q_ast: Option<&antares_ql::QNode>,
199 1222 : ) -> ApiResult<()> {
200 : // Table 6.4.3.2-1, as on the entity query (5.7.2.4)
201 1222 : crate::entities::star_type_local(params)?;
202 1220 : if !crate::entities::qualifies_non_wide(params, q_ast) {
203 16 : return Err(NgsiError::BadRequestData(
204 16 : "temporal query needs at least one of type, attrs, q, georel (5.7.4)".into(),
205 16 : )
206 16 : .into());
207 1204 : }
208 : // 5.7.4.4: Linked Entity retrieval is not defined for temporal queries —
209 : // linked filter conditions are an unconditional BadRequestData
210 1204 : if q_ast.map(antares_ql::QNode::max_link_depth).unwrap_or(0) > 0 {
211 8 : return Err(NgsiError::BadRequestData(
212 8 : "temporal q must not reference Linked Entity attributes (5.7.4.4)".into(),
213 8 : )
214 8 : .into());
215 1196 : }
216 : // 5.7.4.4: a syntactically invalid context source filter is 400; the
217 : // filter itself gates registrations in federation::reg_matches.
218 1196 : if let Some(csf) = params.get("csf") {
219 8 : parse_q(csf)?;
220 1188 : }
221 1188 : crate::paging::check_collation(params)?;
222 : // 5.7.4.4: temporal ordering may only refer to the "id" entity member,
223 : // and only where the execution "is limited to the local scope (see
224 : // clause 5.5.13)" — 4.23.1 gives the reason: "Sort ordering is never
225 : // applied to distributed operations." The subject is the EXECUTION, so a
226 : // query nothing would federate to orders without `local=true`.
227 1186 : if let Some(spec) = params.get("orderBy") {
228 40 : if crate::federation::would_federate(st, tenant, ctx, params, headers).await? {
229 2 : return Err(NgsiError::BadRequestData(
230 2 : "orderBy requires local scope — ordering is never applied to \
231 2 : distributed operations (5.7.4.4, 4.23.1)"
232 2 : .into(),
233 2 : )
234 2 : .into());
235 38 : }
236 42 : let non_id = spec.split(',').any(|part| {
237 42 : let m = part.trim().split(';').next().unwrap_or("").trim();
238 42 : m.split('[').next().unwrap_or(m) != "id"
239 42 : });
240 38 : if non_id {
241 24 : return Err(NgsiError::BadRequestData(
242 24 : "temporal orderBy may only name \"id\" (5.7.4.4)".into(),
243 24 : )
244 24 : .into());
245 14 : }
246 1146 : }
247 1160 : Ok(())
248 1222 : }
249 :
250 : /// 4.5.19: the aggregated answer of a temporal query. The store computed
251 : /// the bucket matrix, so the page is presented with the core members the
252 : /// instance path renders and the aggregated attribute objects copied
253 : /// under their compacted names.
254 : #[allow(clippy::too_many_arguments)]
255 20 : pub(super) fn respond_aggregated(
256 20 : st: &AppState,
257 20 : params: &HashMap<String, String>,
258 20 : tenant: &TenantId,
259 20 : ctx: &Context,
260 20 : accept: Accept,
261 20 : trepr: &TRepr,
262 20 : tq: Option<&TemporalQ>,
263 20 : attrs_filter: Option<&Vec<String>>,
264 20 : outcome: antares_store::filter::TemporalOutcome,
265 20 : ) -> ApiResult<Response> {
266 20 : let total = outcome
267 20 : .total
268 20 : .map(|t| t as usize)
269 20 : .unwrap_or(outcome.rows.len());
270 20 : let (page, count_hdr, links) = crate::paging::paginate_pre(
271 20 : st,
272 20 : params,
273 20 : outcome.rows,
274 20 : "/ngsi-ld/v1/temporal/entities",
275 20 : total,
276 0 : )?;
277 20 : let timeprop = tq.map_or("observedAt", |t| t.timeproperty.as_str());
278 20 : let none = Windowed {
279 20 : attrs: Default::default(),
280 20 : max_per_attr: 0,
281 20 : ts_min: None,
282 20 : ts_max: None,
283 20 : truncated: false,
284 20 : cut_served: None,
285 20 : more: false,
286 20 : held: Default::default(),
287 20 : };
288 20 : let mut payload: Vec<Value> = Vec::new();
289 22 : for d in &page {
290 : // core members exactly as the instance path presents them; the
291 : // aggregated attribute objects are copied under compacted names
292 22 : let mut presented = present_temporal(d, &none, ctx, trepr, tq, timeprop)?;
293 22 : let Some(out) = presented.as_object_mut() else {
294 0 : continue;
295 : };
296 22 : let mut any = false;
297 132 : for (k, v) in d.as_object().into_iter().flatten() {
298 132 : if is_meta(k) || attrs_filter.is_some_and(|a| !a.contains(k)) {
299 92 : continue;
300 40 : }
301 40 : out.insert(ctx.compact_iri(k), v.clone());
302 40 : any = true;
303 : }
304 22 : if any {
305 22 : payload.push(presented);
306 22 : }
307 : }
308 20 : let mut resp = crate::negotiate::respond_list(StatusCode::OK, payload, ctx, accept, tenant);
309 20 : attach_paging(&mut resp, count_hdr, &links);
310 20 : Ok(resp)
311 20 : }
312 :
313 : /// The two Table 5.2.33-1 selectors that need no compiled pattern: the
314 : /// type set, and the `attrs` list, which excludes an entity carrying
315 : /// none of the named Attributes. Each drops an entity the store could
316 : /// not narrow away itself.
317 2322 : pub(super) fn type_and_attrs_match(
318 2322 : doc: &Value,
319 2322 : type_sel: Option<&Vec<Vec<String>>>,
320 2322 : entity_attr_filter: Option<&Vec<String>>,
321 2322 : ) -> bool {
322 2322 : if let Some(sel) = type_sel {
323 : // 4.17: one alternative whose every type the Entity has
324 2226 : let etypes = doc["type"].as_array().cloned().unwrap_or_default();
325 2314 : if !sel.iter().any(|and_group| {
326 2314 : and_group
327 2314 : .iter()
328 2336 : .all(|w| etypes.iter().any(|t| t.as_str() == Some(w.as_str())))
329 2314 : }) {
330 74 : return false;
331 2152 : }
332 96 : }
333 2248 : if let Some(want) = entity_attr_filter {
334 66 : if !want.iter().any(|a| doc.get(a).is_some()) {
335 2 : return false;
336 64 : }
337 2182 : }
338 2246 : true
339 2322 : }
340 :
341 : /// the narrowing rules, for every narrowing: a history is served while the
342 : /// narrowing's own predicates (its `q`, `scopeQ` and areas) hold for the
343 : /// Entity as it is now, so an Entity that left the narrowing takes its
344 : /// history with it. Its rules' conditions are not read here: rule 11 reads
345 : /// them per instance. An Entity held only temporally (11.2.2) has no state
346 : /// now, and its history `held` is the only state it has: the predicates
347 : /// hold when some instance satisfies them, as the query's window check
348 : /// reads them; `held` absent leaves that to the caller's own check.
349 934 : pub(super) async fn member_now(
350 934 : st: &AppState,
351 934 : tenant: &TenantId,
352 934 : ctx: &Context,
353 934 : filter: &crate::policy::Filter,
354 934 : id: &str,
355 934 : held: Option<&Value>,
356 934 : ) -> ApiResult<bool> {
357 934 : let Some(reach) = crate::repr::narrowing(filter).policy_reach else {
358 886 : return Ok(true);
359 : };
360 48 : if reach.q.is_none() && reach.scope_q.is_none() && reach.areas.is_empty() {
361 28 : return Ok(true);
362 20 : }
363 20 : let reach = crate::repr::PolicyReach {
364 20 : rules: None,
365 20 : ..(*reach).clone()
366 20 : };
367 20 : match st
368 20 : .store
369 20 : .get(tenant, antares_store::Kind::Entity, id)
370 20 : .await?
371 : {
372 8 : Some(doc) => Ok(crate::repr::policy_admits(st, tenant, ctx, Some(&reach), &doc).await),
373 12 : None => Ok(held.is_none_or(|h| {
374 4 : let areas: Vec<&antares_ql::geo::GeoQuery> = reach.areas.iter().collect();
375 4 : window_conditions_match(h, None, reach.q.as_ref(), &areas, ctx)
376 2 : && reach
377 2 : .scope_q
378 2 : .as_deref()
379 2 : .is_none_or(|sq| scope_window_retain(&mut h.clone(), sq, None, "observedAt"))
380 4 : })),
381 : }
382 934 : }
383 :
384 1094 : pub(crate) async fn query_temporal_inner(
385 1094 : st: &AppState,
386 1094 : params: &HashMap<String, String>,
387 1094 : headers: &HeaderMap,
388 1094 : filter: &crate::policy::Filter,
389 1094 : ) -> ApiResult<Response> {
390 1094 : query_temporal_collected(st, params, headers, filter, None, None).await
391 1094 : }
392 :
393 : /// The same query, with the merged documents of each page handed to `merged`
394 : /// as well as rendered into the response. 5.16.1.4 fills a Snapshot from this
395 : /// query and has to STORE what it matched: for an Evolution held by a
396 : /// registered Context Source there is nothing else in this broker to store —
397 : /// the fan-out's answer is the only copy, and the response body is a
398 : /// presentation of it that compaction does not undo. `ctx`, when given,
399 : /// stands for the Link header's: a Snapshot's own @context, which may be
400 : /// inline and so has no Link form.
401 1248 : pub(crate) async fn query_temporal_collected(
402 1248 : st: &AppState,
403 1248 : params: &HashMap<String, String>,
404 1248 : headers: &HeaderMap,
405 1248 : filter: &crate::policy::Filter,
406 1248 : merged: Option<&mut Vec<Value>>,
407 1248 : ctx: Option<std::sync::Arc<Context>>,
408 1248 : ) -> ApiResult<Response> {
409 1248 : let tenant = tenant_from(headers)?;
410 1248 : check_params(params, QUERY_PARAMS)?;
411 1234 : let accept = parse_accept(headers)?;
412 1226 : let ctx = match ctx {
413 154 : Some(ctx) => ctx,
414 1072 : None => request_context(&st.loader, headers).await?,
415 : };
416 : // 5.7.4.4 a-e: id/idPattern alone are NOT sufficient, and the attrs
417 : // list / q must include at least one non-system Attribute to qualify.
418 1226 : let q_ast = params.get("q").map(|q| parse_q(q)).transpose()?;
419 1222 : let q_ast = q_ast.map(|ast| expand_q_values(ast, params, &ctx));
420 1222 : check_temporal_query(st, params, headers, &tenant, &ctx, q_ast.as_ref()).await?;
421 : // ADR-0020: the engine's narrowing joins the request's own filters after
422 : // 5.7.4.4's "too wide" judgement, which is about what the client asked
423 : // for. Everything below reads the narrowed query, the forward included.
424 : // the rules, and the client's own q apart from the
425 : // engine's, read on each Entity as its rules show it
426 1160 : let types_rule = crate::repr::type_rules(filter);
427 1160 : let client_q = if types_rule.is_some() {
428 54 : crate::repr::client_q(params, &ctx)?
429 : } else {
430 1106 : None
431 : };
432 1160 : let narrowed = filter.narrow_window(&filter.narrow_params(params, &ctx)?, true)?;
433 1160 : let req = QueryReq {
434 1160 : st,
435 1160 : tenant: &tenant,
436 1160 : headers,
437 1160 : ctx: &ctx,
438 1160 : params: &narrowed,
439 1160 : filter,
440 1160 : };
441 1160 : let qry = Parsed::new(&req, types_rule, client_q)?;
442 1024 : let (geos, shown_geos) = qry.geos(filter);
443 1024 : let plan = Pushdown::plan(&req, &qry, geos.is_empty()).await?;
444 1024 : let outcome = fetch_temporal(&req, &qry, &geos, &plan).await?;
445 1020 : if outcome.aggregated {
446 20 : return respond_aggregated(
447 20 : st,
448 20 : req.params,
449 20 : &tenant,
450 20 : &ctx,
451 20 : accept,
452 20 : &qry.trepr,
453 20 : qry.tq.as_ref(),
454 20 : qry.attrs_filter.as_ref(),
455 20 : outcome,
456 : );
457 1000 : }
458 1000 : let (all, pre_paged, pre_total) = (outcome.rows, outcome.paged, outcome.total);
459 1000 : let mut warnings: Vec<String> = Vec::new();
460 1000 : let all = federate_temporal(&req, &qry.timeprop, all, &mut warnings).await?;
461 1000 : let mut matches = Vec::new();
462 2300 : for doc in all {
463 2300 : let Some(doc) = qualifies(doc, &qry, &geos, &shown_geos, &ctx) else {
464 1144 : continue;
465 : };
466 1156 : if plan.narrowed_now {
467 12 : let id = doc
468 12 : .get("id")
469 12 : .and_then(Value::as_str)
470 12 : .unwrap_or_default()
471 12 : .to_owned();
472 12 : if !member_now(st, &tenant, &ctx, filter, &id, None).await? {
473 2 : continue;
474 10 : }
475 1144 : }
476 1154 : matches.push(doc);
477 : }
478 1000 : if let Some(spec) = req.params.get("orderBy") {
479 14 : crate::paging::order_entities(&mut matches, spec, req.params, &ctx)?;
480 986 : }
481 1000 : let (page, count_hdr, links) = if pre_paged {
482 214 : let total = pre_total.map(|t| t as usize).unwrap_or(matches.len());
483 214 : crate::paging::paginate_pre(
484 214 : st,
485 214 : req.params,
486 214 : matches,
487 214 : "/ngsi-ld/v1/temporal/entities",
488 214 : total,
489 0 : )?
490 : } else {
491 786 : crate::paging::paginate(st, req.params, matches, "/ngsi-ld/v1/temporal/entities")?
492 : };
493 1000 : if let Some(sink) = merged {
494 22 : sink.extend(page.iter().cloned());
495 978 : }
496 1000 : let (payload, g) = assemble_page(&req, &qry, &page)?;
497 1000 : let mut resp = respond_temporal_page(&req, accept, &qry, payload, &g, &warnings);
498 1000 : attach_paging(&mut resp, count_hdr, &links);
499 1000 : attach_paging(&mut resp, None, &g.links(&req, accept, &qry));
500 : // 6.18.3.2: entityMap=true — the temporal EntityMap for this query is
501 : // (re)created; the response carries NGSILD-EntityMap and 201 Created.
502 1000 : if req.params.get("entityMap").map(String::as_str) == Some("true") {
503 24 : let map = build_temporal_map(st, &tenant, headers, &ctx, req.params, filter).await?;
504 22 : *resp.status_mut() = StatusCode::CREATED;
505 22 : if let Some(id) = map.get("id").and_then(Value::as_str) {
506 22 : if let Ok(v) = format!("/ngsi-ld/v1/entityMaps/{id}").parse() {
507 22 : resp.headers_mut().insert("NGSILD-EntityMap", v);
508 22 : }
509 0 : }
510 976 : }
511 998 : filter.mark(&st.host_alias, resp.headers_mut());
512 998 : Ok(resp)
513 1248 : }
514 :
515 : /// The parameters 5.7.4 accepts (6.18.3.2 Table 6.18.3.2-1).
516 : const QUERY_PARAMS: &[&str] = &[
517 : "id",
518 : "idPattern",
519 : "type",
520 : "attrs",
521 : "q",
522 : "georel",
523 : "geometry",
524 : "coordinates",
525 : "geoproperty",
526 : "scopeQ",
527 : "csf",
528 : "timerel",
529 : "timeAt",
530 : "endTimeAt",
531 : "timeproperty",
532 : "aggrMethods",
533 : "aggrPeriodDuration",
534 : "lastN",
535 : "firstN",
536 : "offsetN",
537 : "limit",
538 : "offset",
539 : "count",
540 : "options",
541 : "format",
542 : "lang",
543 : "local",
544 : "entityMap",
545 : "pick",
546 : "omit",
547 : "datasetId",
548 : "orderBy",
549 : "orderFrom",
550 : "orderGeometry",
551 : "collation",
552 : "entityMapLifetime",
553 : "splitEntities",
554 : "expandValues",
555 : "jsonKeys",
556 : ];
557 :
558 : /// 5.7.4.3 expandValues: the same 4.9 EXAMPLE 12 coercion as the entity
559 : /// query — term values expanded against the @context before executing,
560 : /// less the Attributes jsonKeys declares uninterpretable.
561 510 : fn expand_q_values(
562 510 : ast: antares_ql::QNode,
563 510 : params: &HashMap<String, String>,
564 510 : ctx: &Context,
565 510 : ) -> antares_ql::QNode {
566 510 : antares_ql::eval::apply_expand_values(
567 510 : ast,
568 510 : antares_ql::eval::expansion_list(
569 510 : params.get("expandValues").map(String::as_str),
570 510 : params.get("jsonKeys").map(String::as_str),
571 510 : )
572 510 : .as_deref(),
573 510 : ctx,
574 : )
575 510 : }
576 :
577 : /// One 5.7.4 request as every stage after the narrowing reads it; `params`
578 : /// is the narrowed query.
579 : struct QueryReq<'a> {
580 : st: &'a AppState,
581 : tenant: &'a TenantId,
582 : headers: &'a HeaderMap,
583 : ctx: &'a Context,
584 : params: &'a HashMap<String, String>,
585 : filter: &'a crate::policy::Filter,
586 : }
587 :
588 : /// The narrowed 5.7.4 query, parsed and validated up front.
589 : struct Parsed<'p> {
590 : /// the type rules each Entity is shown by
591 : types_rule: Option<std::sync::Arc<crate::policy::Filter>>,
592 : /// the client's own q under rules, judged on the shown history
593 : client_q: Option<antares_ql::QNode>,
594 : /// the q the store and the stored history read
595 : q_ast: Option<antares_ql::QNode>,
596 : scope_q: Option<&'p str>,
597 : tq: Option<TemporalQ>,
598 : trepr: TRepr,
599 : ids: Option<Vec<&'p str>>,
600 : id_pattern: Option<IdMatcher>,
601 : /// the 4.17 type selection: alternatives of types an Entity must all
602 : /// have; `None` for no selector or `*`
603 : type_sel: Option<Vec<Vec<String>>>,
604 : /// every type the selection names, the store's (OR) prefilter
605 : types: Option<Vec<String>>,
606 : /// what the window keeps per Entity
607 : attrs_filter: Option<Vec<String>>,
608 : /// only the attrs= param excludes entities; pick is projection-only
609 : entity_attr_filter: Option<Vec<String>>,
610 : /// under rules: the attrs= the shown history must hold
611 : shown_attrs: Option<Vec<String>>,
612 : own_geo: Option<antares_ql::geo::GeoQuery>,
613 : timeprop: String,
614 : }
615 :
616 : impl<'p> Parsed<'p> {
617 1160 : fn new(
618 1160 : req: &QueryReq<'p>,
619 1160 : types_rule: Option<std::sync::Arc<crate::policy::Filter>>,
620 1160 : client_q: Option<antares_ql::QNode>,
621 1160 : ) -> ApiResult<Self> {
622 1160 : let (params, ctx) = (req.params, req.ctx);
623 1160 : let ruled = types_rule.is_some();
624 1160 : let q_ast = params
625 1160 : .get("q")
626 1160 : .map(|q| antares_ql::parse_narrowed_q(q))
627 1160 : .transpose()?
628 1160 : .map(|ast| expand_q_values(ast, params, ctx));
629 : // under rules only the engine's own q reaches the store and the stored
630 : // history; the client's waits for the shown one
631 1160 : let q_ast = if ruled {
632 54 : req.filter.q.as_deref().cloned()
633 : } else {
634 1106 : q_ast
635 : };
636 1160 : let tq = TemporalQ::from_params(params, true)?;
637 1120 : let mut trepr = parse_trepr(params, ctx)?;
638 1044 : crate::repr::narrow_projection(&mut trepr.pick, &mut trepr.omit, req.filter, ctx)?;
639 : // 5.7.4.4: {…} projection is Linked Entity retrieval — unconditional 400
640 1044 : reject_linked_projection(&trepr, "5.7.4.4")?;
641 1032 : let ids: Option<Vec<&str>> = params.get("id").map(|s| s.split(',').collect());
642 : // 5.7.4.4: an invalid URI in the id list is BadRequestData
643 1032 : for id in ids.iter().flatten() {
644 100 : antares_model::EntityId::new(id)?;
645 : }
646 1024 : let id_pattern = id_matcher(params.get("idPattern"))?;
647 1024 : let type_sel = crate::entities::type_selection(params, ctx);
648 1024 : let types: Option<Vec<String>> = type_sel.as_ref().map(|sel| {
649 958 : let mut all: Vec<String> = sel.iter().flatten().cloned().collect();
650 958 : all.sort();
651 958 : all.dedup();
652 958 : all
653 958 : });
654 1024 : let attrs_filter = selection(&trepr);
655 1024 : let (entity_attr_filter, shown_attrs) = if ruled {
656 54 : (None, trepr.attrs.clone())
657 : } else {
658 970 : (trepr.attrs.clone(), None)
659 : };
660 1024 : let own_geo = antares_ql::geo::GeoQuery::from_params(params)?;
661 1024 : let timeprop = tq
662 1024 : .as_ref()
663 1024 : .map_or("observedAt", |t| t.timeproperty.as_str())
664 1024 : .to_owned();
665 1024 : Ok(Self {
666 1024 : types_rule,
667 1024 : client_q,
668 1024 : q_ast,
669 1024 : scope_q: params.get("scopeQ").map(String::as_str),
670 1024 : tq,
671 1024 : trepr,
672 1024 : ids,
673 1024 : id_pattern,
674 1024 : type_sel,
675 1024 : types,
676 1024 : attrs_filter,
677 1024 : entity_attr_filter,
678 1024 : shown_attrs,
679 1024 : own_geo,
680 1024 : timeprop,
681 1024 : })
682 1160 : }
683 :
684 : /// The request's own geoquery and the narrowing's area, all of which
685 : /// must hold; under rules the request's reads the
686 : /// shown history.
687 1024 : fn geos<'g>(&'g self, filter: &'g crate::policy::Filter) -> (Geos<'g>, Geos<'g>) {
688 1024 : if self.types_rule.is_some() {
689 54 : (filter.areas().collect(), self.own_geo.iter().collect())
690 : } else {
691 970 : (
692 970 : self.own_geo.iter().chain(filter.areas()).collect(),
693 970 : Vec::new(),
694 970 : )
695 : }
696 1024 : }
697 :
698 1270 : fn instance_range(&self) -> Option<antares_store::filter::InstanceRange<'_>> {
699 1270 : self.tq
700 1270 : .as_ref()
701 1270 : .map(|t| antares_store::filter::InstanceRange {
702 1270 : timerel: &t.timerel,
703 1270 : time_at: &t.time_at,
704 1270 : end_time_at: t.end_time_at.as_deref(),
705 1270 : timeproperty: &t.timeproperty,
706 1270 : })
707 1270 : }
708 : }
709 :
710 : /// The geoqueries an Evolution's window is judged by.
711 : type Geos<'g> = Vec<&'g antares_ql::geo::GeoQuery>;
712 :
713 : /// The compiled idPattern, as the one test the query asks of an id.
714 : type IdMatcher = Box<dyn Fn(&str) -> bool + Send + Sync>;
715 :
716 : /// 5.7.4.4: an idPattern that does not compile is BadRequestData.
717 1024 : fn id_matcher(pattern: Option<&String>) -> Result<Option<IdMatcher>, NgsiError> {
718 1024 : let Some(p) = pattern else {
719 1020 : return Ok(None);
720 : };
721 4 : let re = antares_ql::regex::compile(p)
722 4 : .map_err(|_| NgsiError::BadRequestData(format!("invalid idPattern {p:?}")))?;
723 8 : Ok(Some(Box::new(move |id: &str| re.is_match(id))))
724 1024 : }
725 :
726 : /// What the store decides itself for one query.
727 : struct Pushdown {
728 : /// q, geo and scopeQ absent: the store's instance cap is exact
729 : exact: bool,
730 : /// the store pages the Entities
731 : page: bool,
732 : /// the store answers the 4.5.19 aggregation
733 : agg: bool,
734 : /// a narrowing reads the state now: each Entity is checked after the store
735 : narrowed_now: bool,
736 : offset: usize,
737 : limit: usize,
738 : /// with `page` under rules: the Attributes an Entity qualifies through
739 : page_attrs: Option<Vec<String>>,
740 : }
741 :
742 : impl Pushdown {
743 : // Push entity narrowing (ids/types/attrs) and instance-window
744 : // pruning (range + RANK()-capped lastN) into the store. The loop
745 : // and window() stay the arbiters — pruning is byte-exact against
746 : // instance_matches (compile::temporal), so it cannot change an answer.
747 : // 5.7.4.4 S2/S3: q and geo are judged on the instances WITHIN the
748 : // temporal interval (the eval_doc retain in `qualifies`), so RANGE pruning is
749 : // verdict-safe to push even with q=/geo present. Only the lastN cap
750 : // (its ordering vs the values filter is unspecified) and entity paging
751 : // (q/geo still drop entities after SQL) wait for exactness.
752 : // scopeQ joins q/geo here: the 4.18 validity filter drops entities and
753 : // instances AFTER SQL, so a pushed page/lastN cap would under-return.
754 1024 : async fn plan(req: &QueryReq<'_>, qry: &Parsed<'_>, no_geo: bool) -> ApiResult<Self> {
755 1024 : let (st, params, filter) = (req.st, req.params, req.filter);
756 1024 : let exact = qry.q_ast.is_none() && no_geo && qry.scope_q.is_none();
757 : // Entity-page pushdown: a temporal query used to materialize the
758 : // tenant's ENTIRE history. Pushed only when every filter the store
759 : // cannot see is absent — same gate family as the entity-query pushdown.
760 : // A values filter no longer blocks paging when its prefilter compiles
761 : // EXACTLY (every leaf a Cmp with the byte-exact text window): the SQL
762 : // entity verdict then equals the evaluator's. datasetId/pick still
763 : // block: their entity drops happen at presentation, after the page.
764 1024 : let q_page_exact = qry.q_ast.as_ref().is_none_or(|ast| {
765 246 : let r = qry.instance_range();
766 246 : st.temporal
767 246 : .q_pushdown_exact(ast, r.as_ref(), &|t| req.ctx.expand_key(t))
768 246 : });
769 1024 : let (offset, limit, _) = crate::paging::page_params(st, params)?;
770 : // rule 8 drops Entities after the store answered, so the store may not
771 : // page under a narrowing that reads the state now (one read
772 : // per Entity; push it into the store query when it shows up)
773 1024 : let narrowed_now =
774 1024 : filter.q.is_some() || filter.scope_q.is_some() || filter.areas().next().is_some();
775 : // 5.7.4.4 + 5.5.9: pagination applies to the MERGED federated union, so
776 : // the store may only pre-page when nothing will federate — otherwise
777 : // page 1 is local-page + every remote row (matrix-9 IOP_EXT_TMP_03_04).
778 : // under rules only where their verdict on an Entity is the store's,
779 : // and the request's own q and geoquery, read on the shown history,
780 : // are absent
781 1024 : let ruled = match qry.types_rule.as_deref() {
782 970 : None => Some(None),
783 54 : Some(_) if qry.client_q.is_some() || qry.own_geo.is_some() => None,
784 48 : Some(rules) => page_attrs(
785 48 : rules,
786 48 : qry.types.as_deref(),
787 48 : qry.tq.as_ref(),
788 48 : qry.shown_attrs.as_deref(),
789 : ),
790 : };
791 1024 : let page = (exact || (no_geo && qry.scope_q.is_none() && q_page_exact))
792 712 : && ruled.is_some()
793 680 : && qry.id_pattern.is_none()
794 : // a 4.17 AND group is judged after the store's OR prefilter
795 676 : && qry
796 676 : .type_sel
797 676 : .iter()
798 676 : .flatten()
799 676 : .all(|and_group| and_group.len() == 1)
800 670 : && params.get("orderBy").is_none()
801 656 : && params.get("datasetId").is_none()
802 632 : && params.get("pick").is_none()
803 606 : && limit > 0
804 606 : && !narrowed_now
805 606 : && !crate::federation::would_federate(st, req.tenant, req.ctx, params, req.headers)
806 606 : .await?;
807 : // the store aggregates every Attribute it reads: never one the rules hide
808 1024 : let agg = exact && page && qry.types_rule.is_none() && store_aggregates(req, &qry.trepr);
809 1024 : Ok(Self {
810 1024 : exact,
811 1024 : page,
812 1024 : agg,
813 1024 : narrowed_now,
814 1024 : offset,
815 1024 : limit,
816 1024 : page_attrs: ruled.flatten(),
817 1024 : })
818 1024 : }
819 : }
820 :
821 : /// 4.5.19 computed by the store: the numeric bucket matrix per attribute
822 : /// comes back aggregated when nothing after the store call could change
823 : /// the answer — every filter exact in SQL, the page pushed, no
824 : /// projection/lastN/month periods — otherwise the instances are
825 : /// aggregated here as before. A store that cannot (memory) or a
826 : /// non-numeric value class leaves `outcome.aggregated` false.
827 512 : fn store_aggregates(req: &QueryReq<'_>, trepr: &TRepr) -> bool {
828 512 : trepr.aggregated
829 54 : && !crate::history::records_on_change(req.st, req.tenant)
830 54 : && trepr.omit.is_none()
831 54 : && trepr.page == InstancePage::default()
832 54 : && !matches!(trepr.aggr_period, AggrPeriod::Months(..))
833 54 : && req.params.get("entityMap").map(String::as_str) != Some("true")
834 54 : && trepr
835 54 : .aggr_methods
836 54 : .iter()
837 88 : .all(|m| antares_store::filter::AGGREGATE_METHODS.contains(&m.as_str()))
838 512 : }
839 :
840 : /// The store call of one query, with what `plan` pushed down.
841 1024 : async fn fetch_temporal(
842 1024 : req: &QueryReq<'_>,
843 1024 : qry: &Parsed<'_>,
844 1024 : geos: &[&antares_ql::geo::GeoQuery],
845 1024 : plan: &Pushdown,
846 1024 : ) -> ApiResult<antares_store::filter::TemporalOutcome> {
847 1024 : let (ctx, trepr, page_n) = (req.ctx, &qry.trepr, qry.trepr.page);
848 : // scoped: the &dyn expander must not live across an await (handler
849 : // futures are Send; the store call itself is synchronous)
850 1024 : let expand = |t: &str| ctx.expand_key(t);
851 1024 : let geo_pre: Vec<_> = geos.iter().map(|g| g.to_instance_spec(ctx)).collect();
852 1024 : let tf = antares_store::filter::TemporalFilter {
853 1024 : ids: qry.ids.as_deref(),
854 1024 : types: qry.types.as_deref(),
855 1024 : attrs: qry.entity_attr_filter.as_deref(),
856 : // a condition rebuilds each instance's state from the whole
857 : // history; rules without one read only the window
858 1024 : range: qry
859 1024 : .instance_range()
860 1024 : .filter(|_| !replays_state(qry.types_rule.as_deref())),
861 1024 : last_n: page_n
862 1024 : .store_cap(default_page(req.st, req.tenant, trepr))
863 1024 : .filter(|_| {
864 970 : plan.exact
865 602 : && cap_keeps_shown(
866 602 : qry.types_rule.as_deref(),
867 602 : &page_n,
868 602 : &qry.timeprop,
869 602 : range_start(qry.tq.as_ref()),
870 : )
871 970 : }),
872 1024 : oldest_first: !page_n.desc,
873 1024 : timeproperty: &qry.timeprop,
874 1024 : page: plan.page.then_some(antares_store::filter::Page {
875 1024 : offset: plan.offset as i64,
876 1024 : limit: plan.limit as i64,
877 1024 : count: true,
878 1024 : }),
879 1024 : page_attrs: plan.page_attrs.as_deref().filter(|_| plan.page),
880 1024 : q: qry.q_ast.as_ref(),
881 1024 : expand: &expand,
882 1024 : geo: &geo_pre,
883 1024 : aggregate: plan.agg.then_some(antares_store::filter::Aggregate {
884 1024 : methods: &trepr.aggr_methods,
885 1024 : period_secs: match trepr.aggr_period {
886 20 : AggrPeriod::Seconds(sc) => Some(sc),
887 1004 : _ => None,
888 : },
889 1024 : anchor: qry.tq.as_ref().map(|t| t.time_at.as_str()),
890 : }),
891 : };
892 1024 : Ok(req.st.temporal.query_temporal(req.tenant, &tf).await?)
893 1024 : }
894 :
895 : /// 5.7.4.4: fan the query out to matching queryTemporal registrations
896 : /// and merge the remote Temporal Evolutions with the local set (4.5.5;
897 : /// auxiliary data never introduces new entities).
898 1000 : async fn federate_temporal(
899 1000 : req: &QueryReq<'_>,
900 1000 : timeprop: &str,
901 1000 : all: Vec<Value>,
902 1000 : warnings: &mut Vec<String>,
903 1000 : ) -> ApiResult<Vec<Value>> {
904 1000 : let looped = crate::federation::via_loop(
905 1000 : req.headers,
906 1000 : &crate::federation::alias_for(&req.st.host_alias, req.tenant),
907 : );
908 1000 : if !crate::federation::active(req.params) || looped {
909 44 : return Ok(all);
910 956 : }
911 956 : let mut proxies = Vec::new();
912 956 : let fed = crate::federation::fed_query_temporal(
913 956 : req.st,
914 956 : req.tenant,
915 956 : req.headers,
916 956 : req.ctx,
917 956 : req.params,
918 956 : warnings,
919 956 : &mut proxies,
920 956 : )
921 956 : .await?;
922 956 : let mut all = all;
923 2272 : for doc in &mut all {
924 2272 : crate::federation::drop_proxied_local(doc, &proxies, req.ctx);
925 2272 : }
926 956 : Ok(merge_federated(all, &fed, timeprop))
927 1000 : }
928 :
929 : /// The local Evolutions in their order, each merged with its remote
930 : /// copies, then the Entities only an exclusive registration holds.
931 956 : fn merge_federated(all: Vec<Value>, fed: &[(bool, Value)], timeprop: &str) -> Vec<Value> {
932 956 : let mut order: Vec<String> = Vec::new();
933 956 : let mut by_id: std::collections::HashMap<String, Value> = Default::default();
934 2272 : for doc in all {
935 2272 : if let Some(id) = doc.get("id").and_then(Value::as_str) {
936 2272 : order.push(id.to_owned());
937 2272 : by_id.insert(id.to_owned(), doc);
938 2272 : }
939 : }
940 1912 : for aux_pass in [false, true] {
941 1912 : for (aux, d) in fed {
942 32 : if *aux != aux_pass {
943 16 : continue;
944 16 : }
945 16 : let Some(id) = d.get("id").and_then(Value::as_str) else {
946 0 : continue;
947 : };
948 16 : match by_id.get_mut(id) {
949 2 : Some(base) => merge_temporal_docs(base, d, *aux, timeprop),
950 14 : None if !aux => {
951 14 : order.push(id.to_owned());
952 14 : by_id.insert(id.to_owned(), d.clone());
953 14 : }
954 0 : None => {}
955 : }
956 : }
957 : }
958 956 : order
959 956 : .into_iter()
960 2286 : .filter_map(|id| by_id.remove(&id))
961 956 : .collect()
962 956 : }
963 :
964 : /// One candidate Evolution judged by the query: `None` drops it, `Some`
965 : /// is the Evolution as the query keeps it (q instances retained, scopeQ
966 : /// windows cut, the rules' shown history).
967 2300 : fn qualifies(
968 2300 : mut doc: Value,
969 2300 : qry: &Parsed<'_>,
970 2300 : geos: &[&antares_ql::geo::GeoQuery],
971 2300 : shown_geos: &[&antares_ql::geo::GeoQuery],
972 2300 : ctx: &Context,
973 2300 : ) -> Option<Value> {
974 2300 : let tq = qry.tq.as_ref();
975 2300 : let id = doc.get("id").and_then(Value::as_str).unwrap_or("");
976 2300 : match (&qry.ids, &qry.id_pattern) {
977 128 : (Some(ids), _) if !ids.contains(&id) => return None,
978 : // 5.2.33: "id takes precedence over idPattern"
979 8 : (None, Some(matches)) if !matches(id) => return None,
980 2264 : _ => {}
981 : }
982 2264 : if !type_and_attrs_match(&doc, qry.type_sel.as_ref(), qry.entity_attr_filter.as_ref()) {
983 74 : return None;
984 2190 : }
985 2190 : if !conditions_hold(&mut doc, tq, qry.q_ast.as_ref(), geos, ctx) {
986 734 : return None;
987 1456 : }
988 1456 : if let Some(sq) = qry.scope_q {
989 368 : if !scope_window_retain(&mut doc, sq, tq, &qry.timeprop) {
990 272 : return None;
991 96 : }
992 1088 : }
993 1184 : if let Some(rules) = qry.types_rule.as_deref() {
994 60 : doc = rules_per_instance(&doc, rules)?;
995 58 : if !type_and_attrs_match(&doc, None, qry.shown_attrs.as_ref()) {
996 2 : return None;
997 56 : }
998 56 : if !conditions_hold(&mut doc, tq, qry.client_q.as_ref(), shown_geos, ctx) {
999 4 : return None;
1000 52 : }
1001 1124 : }
1002 : // entity qualifies only if some instance falls in the window
1003 1176 : any_in_window(&doc, tq).then_some(doc)
1004 2300 : }
1005 :
1006 : /// 5.7.4.4 S2/S3: q and the geoqueries hold within the window, and the
1007 : /// instances q does not hold for leave the Evolution.
1008 2246 : fn conditions_hold(
1009 2246 : doc: &mut Value,
1010 2246 : tq: Option<&TemporalQ>,
1011 2246 : q: Option<&antares_ql::QNode>,
1012 2246 : geos: &[&antares_ql::geo::GeoQuery],
1013 2246 : ctx: &Context,
1014 2246 : ) -> bool {
1015 2246 : if (q.is_some() || !geos.is_empty()) && !window_conditions_match(doc, tq, q, geos, ctx) {
1016 738 : return false;
1017 1508 : }
1018 1508 : if let Some(ast) = q {
1019 238 : retain_q_instances(doc, tq, ast, ctx);
1020 1270 : }
1021 1508 : true
1022 2246 : }
1023 :
1024 1176 : fn any_in_window(doc: &Value, tq: Option<&TemporalQ>) -> bool {
1025 1176 : doc.as_object().is_some_and(|o| {
1026 2420 : o.iter().any(|(k, v)| {
1027 2420 : !is_meta(k)
1028 1198 : && v.as_array().is_some_and(|arr| {
1029 1198 : arr.iter()
1030 1544 : .any(|inst| tq.is_none_or(|tq| tq.instance_matches(inst)))
1031 1198 : })
1032 2420 : })
1033 1176 : })
1034 1176 : }
1035 :
1036 : /// What the page's windows add up to: the 6.3.10 Content-Range and the
1037 : /// 6.4.7.3 next-page links read it.
1038 : #[derive(Default)]
1039 : struct PageCut {
1040 : truncated: bool,
1041 : ts_min: Option<String>,
1042 : ts_max: Option<String>,
1043 : more: bool,
1044 : /// the next page's move: the fewest instances any Entity's cut Attribute
1045 : /// served, and the boundary every Entity's cut reached (`interval_links`)
1046 : cut: Option<usize>,
1047 : bound: Option<String>,
1048 : }
1049 :
1050 : impl PageCut {
1051 1128 : fn add(&mut self, w: &Windowed, desc: bool) {
1052 1128 : self.truncated |= w.truncated;
1053 1128 : self.more |= w.more;
1054 1128 : if let Some(c) = w.cut_served {
1055 112 : self.cut = Some(self.cut.map_or(c, |g| g.min(c)));
1056 1016 : }
1057 1128 : if let Some(b) = w.truncated.then(|| cut_boundary(w, desc)).flatten() {
1058 : // the earliest boundary forwards, the latest backwards: no
1059 : // Entity's instance past its own boundary is left behind
1060 112 : if self
1061 112 : .bound
1062 112 : .as_deref()
1063 112 : .is_none_or(|g| (dt_key(b) < dt_key(g)) != desc)
1064 104 : {
1065 104 : self.bound = Some(b.to_owned());
1066 104 : }
1067 1016 : }
1068 1128 : if let Some(m) = &w.ts_min {
1069 1112 : if self.ts_min.as_deref().is_none_or(|c| dt_key(m) < dt_key(c)) {
1070 807 : self.ts_min = Some(m.clone());
1071 807 : }
1072 16 : }
1073 1128 : if let Some(m) = &w.ts_max {
1074 1112 : if self.ts_max.as_deref().is_none_or(|c| dt_key(m) > dt_key(c)) {
1075 930 : self.ts_max = Some(m.clone());
1076 930 : }
1077 16 : }
1078 1128 : }
1079 :
1080 : /// aggregated responses are complete by construction — never 206 (6.3.10)
1081 1000 : fn content_range(&self, qry: &Parsed<'_>) -> Option<String> {
1082 1000 : if qry.trepr.aggregated {
1083 34 : return None;
1084 966 : }
1085 966 : content_range(
1086 966 : self.truncated,
1087 966 : self.ts_min.as_deref(),
1088 966 : self.ts_max.as_deref(),
1089 966 : qry.tq.as_ref(),
1090 966 : qry.trepr.page,
1091 : )
1092 1000 : }
1093 :
1094 1000 : fn links(&self, req: &QueryReq<'_>, accept: Accept, qry: &Parsed<'_>) -> Vec<String> {
1095 1000 : let page_n = qry.trepr.page;
1096 1000 : interval_links(
1097 1000 : req.params,
1098 1000 : "/ngsi-ld/v1/temporal/entities",
1099 1000 : accept,
1100 1000 : page_n,
1101 1000 : self.more,
1102 1000 : page_n.step(
1103 1000 : qry.trepr.aggregated,
1104 1000 : req.st.temporal_default_instances,
1105 1000 : req.st.temporal_instance_limit,
1106 : ),
1107 1000 : self.cut.zip(self.bound.as_deref()),
1108 : )
1109 1000 : }
1110 : }
1111 :
1112 : /// Each Entity of the page windowed, cut and presented. Under the narrowing rules:
1113 : /// each Entity's history was narrowed by its rules in `qualifies`, before
1114 : /// the window, so lastN counts only what may be seen (rule 8).
1115 1000 : fn assemble_page(
1116 1000 : req: &QueryReq<'_>,
1117 1000 : qry: &Parsed<'_>,
1118 1000 : page: &[Value],
1119 1000 : ) -> ApiResult<(Vec<Value>, PageCut)> {
1120 1000 : let (st, trepr, page_n) = (req.st, &qry.trepr, qry.trepr.page);
1121 1000 : let core_only_pick = qry.attrs_filter.as_ref().is_some_and(Vec::is_empty);
1122 1000 : let mut payload: Vec<Value> = Vec::new();
1123 1000 : let mut g = PageCut::default();
1124 1128 : for d in page {
1125 1128 : let mut w = window(
1126 1128 : d,
1127 1128 : qry.tq.as_ref(),
1128 1128 : page_n,
1129 1128 : qry.attrs_filter.as_ref(),
1130 1128 : trepr.omit.as_ref(),
1131 1128 : trepr.dataset_id.as_ref(),
1132 1128 : &qry.timeprop,
1133 1128 : crate::history::held_attrs(st, req.tenant, d),
1134 : );
1135 1128 : if !trepr.aggregated {
1136 1114 : truncate(
1137 1114 : &mut w,
1138 1114 : &qry.timeprop,
1139 1114 : page_n.desc,
1140 1114 : page_n.size(st.temporal_default_instances, st.temporal_instance_limit),
1141 1114 : );
1142 1114 : }
1143 1128 : g.add(&w, page_n.desc);
1144 : // no instance survived the window/dataset filters ⇒ the entity is
1145 : // not part of the temporal result (unless the projection is
1146 : // deliberately core-only, e.g. pick=id)
1147 1128 : if w.attrs.is_empty() && !core_only_pick {
1148 4 : continue;
1149 1124 : }
1150 1124 : let presented = present_temporal(d, &w, req.ctx, trepr, qry.tq.as_ref(), &qry.timeprop)?;
1151 1124 : if trepr.pick.is_some() && presented.as_object().is_some_and(|o| o.is_empty()) {
1152 0 : continue;
1153 1124 : }
1154 1124 : payload.push(presented);
1155 : }
1156 1000 : Ok((payload, g))
1157 1000 : }
1158 :
1159 : /// The page's body, 200 or 206 with its Content-Range, and the
1160 : /// federation's warnings.
1161 1000 : fn respond_temporal_page(
1162 1000 : req: &QueryReq<'_>,
1163 1000 : accept: Accept,
1164 1000 : qry: &Parsed<'_>,
1165 1000 : payload: Vec<Value>,
1166 1000 : g: &PageCut,
1167 1000 : warnings: &[String],
1168 1000 : ) -> Response {
1169 1000 : let cr = g.content_range(qry);
1170 1000 : let status = if cr.is_some() {
1171 82 : StatusCode::PARTIAL_CONTENT
1172 : } else {
1173 918 : StatusCode::OK
1174 : };
1175 1000 : let mut resp = crate::negotiate::respond_list(status, payload, req.ctx, accept, req.tenant);
1176 1000 : crate::paging::attach_warnings(&mut resp, warnings);
1177 1000 : if let Some(cr) = cr {
1178 82 : if let Ok(v) = cr.parse() {
1179 82 : resp.headers_mut().insert("Content-Range", v);
1180 82 : }
1181 918 : }
1182 1000 : resp
1183 1000 : }
1184 :
1185 : // ---------- POST /temporal/entityOperations/query (6.24) ----------
1186 :
1187 204 : pub async fn batch_temporal_query(
1188 204 : State(st): State<AppState>,
1189 204 : CleanParams(params): CleanParams,
1190 204 : headers: HeaderMap,
1191 204 : body: Bytes,
1192 204 : ) -> Response {
1193 204 : let go = async {
1194 : // 6.3.14 and 6.3.4: a Tenant outside the grammar and an Accept the
1195 : // operation cannot serve are both refused here. Neither VALUE is
1196 : // needed — the inner query reads the headers again — but the request
1197 : // must not reach it having skipped either check.
1198 204 : let tenant = tenant_from(&headers)?;
1199 204 : check_params(
1200 204 : ¶ms,
1201 204 : &["limit", "offset", "count", "options", "format", "local"],
1202 0 : )?;
1203 204 : parse_accept(&headers)?;
1204 204 : let parsed = parse_body(&st, &headers, &body, BodyKind::Standard).await?;
1205 204 : let q = parsed.object(NgsiError::BadRequestData(
1206 204 : "query body must be an object".into(),
1207 204 : ))?;
1208 204 : if q.get("type").and_then(Value::as_str) != Some("Query") {
1209 0 : return Err(NgsiError::BadRequestData("body type must be Query".into()).into());
1210 204 : }
1211 : // 5.2.23 Query (temporal reading): members flattened with their
1212 : // Table 5.2.23-1 value spaces enforced, incl. temporalQ (5.2.21)
1213 : // and aggrParams (5.2.44).
1214 204 : let mut vp: HashMap<String, String> = params.clone();
1215 204 : crate::paging::query_doc_params(q, true, &mut vp)?;
1216 : // ADR-0020: the engine is shown every member of the Query body
1217 134 : let named = crate::entities::ReadMembers::with_ctx(&st, &vp, &parsed.ctx);
1218 134 : let filter = gate!(
1219 : st, &tenant, &headers, "5.7.4",
1220 : q: named.q.as_ref(),
1221 : scope_q: vp.get("scopeQ").map(String::as_str),
1222 : types: &named.types,
1223 : attrs: &named.attrs,
1224 : geo: named.geo.as_ref(),
1225 : )
1226 134 : .await?;
1227 : // 5.5.7: the terms of an application/ld+json body resolve against
1228 : // the body's own @context, not the Link header's
1229 132 : query_temporal_collected(&st, &vp, &headers, &filter, None, Some(parsed.ctx.clone())).await
1230 204 : };
1231 204 : go.await.unwrap_or_else(|e| e.into_response())
1232 204 : }
1233 :
1234 : /// 5.14.5.4: temporal query required; the S1–S4 candidate selection is the
1235 : /// temporal query pipeline itself (5.7.4.4) run unpaged — the ids of its
1236 : /// result set form the EntityMap; the createEntityMapQueryTemporal
1237 : /// registrations are then merged like 5.14.4.
1238 : /// Known ceiling: candidate ids are read from the internal 5.7.4 response capped
1239 : /// at max_limit; raise the cap if temporal sets outgrow it.
1240 64 : pub(crate) async fn build_temporal_map(
1241 64 : st: &AppState,
1242 64 : tenant: &TenantId,
1243 64 : headers: &HeaderMap,
1244 64 : ctx: &antares_jsonld::Context,
1245 64 : params: &HashMap<String, String>,
1246 64 : filter: &crate::policy::Filter,
1247 64 : ) -> ApiResult<Value> {
1248 64 : if !params.contains_key("timerel") {
1249 12 : return Err(NgsiError::BadRequestData(
1250 12 : "a temporal query is required to create a temporal EntityMap (5.14.5.4)".into(),
1251 12 : )
1252 12 : .into());
1253 52 : }
1254 52 : let local_scope = params.get("local").map(String::as_str) == Some("true");
1255 52 : let split = params.get("splitEntities").map(String::as_str) == Some("true");
1256 52 : let mut eff: HashMap<String, String> = if split && !local_scope {
1257 0 : params
1258 0 : .iter()
1259 0 : .filter(|(k, _)| {
1260 0 : [
1261 0 : "id",
1262 0 : "idPattern",
1263 0 : "type",
1264 0 : "local",
1265 0 : "timerel",
1266 0 : "timeAt",
1267 0 : "endTimeAt",
1268 0 : "timeproperty",
1269 0 : ]
1270 0 : .contains(&k.as_str())
1271 0 : })
1272 0 : .map(|(k, v)| (k.clone(), v.clone()))
1273 0 : .collect()
1274 : } else {
1275 52 : params.clone()
1276 : };
1277 260 : for k in [
1278 52 : "entityMap",
1279 52 : "entityMapLifetime",
1280 52 : "splitEntities",
1281 52 : "offset",
1282 52 : "count",
1283 260 : ] {
1284 260 : eff.remove(k);
1285 260 : }
1286 52 : eff.insert("limit".into(), st.max_limit.to_string());
1287 52 : show_ids(&mut eff);
1288 : // Box::pin: build_temporal_map is reachable from query_temporal_inner
1289 : // (entityMap=true), so this recursive edge needs indirection.
1290 52 : let resp = Box::pin(query_temporal_inner(st, &eff, &ids_shown(headers), filter)).await?;
1291 52 : let bytes = axum::body::to_bytes(resp.into_body(), st.max_answer_bytes)
1292 52 : .await
1293 52 : .map_err(|_| NgsiError::TooManyResults(crate::bounds::ANSWER_TOO_LARGE.into()))?;
1294 : // 5.5.14: the map FIXES the Entities considered by every later request, so
1295 : // an unreadable candidate set must fail rather than become "no candidates".
1296 50 : let candidates: Value = serde_json::from_slice(&bytes)
1297 50 : .map_err(|_| NgsiError::InternalError("temporal candidate parse".into()))?;
1298 50 : let mut emap = Map::new();
1299 50 : if let Some(arr) = candidates.as_array() {
1300 50 : for d in arr {
1301 36 : if let Some(id) = d.get("id").and_then(Value::as_str) {
1302 36 : emap.insert(id.to_owned(), json!(["@none"]));
1303 36 : }
1304 : }
1305 0 : }
1306 50 : crate::entity_map::merge_and_store_map(st, tenant, headers, ctx, params, true, emap).await
1307 64 : }
|
