Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Batch operations /entityOperations/* (5.6.7–5.6.10, 5.6.20, 5.7.2-POST;
3 : //! resources 6.14–6.17, 6.23, 6.31).
4 :
5 : use crate::entities::merge_into;
6 : use crate::negotiate::*;
7 : use crate::paging::{paginate, query_doc_params};
8 : use crate::repr::{apply, parse_repr};
9 : use crate::stamp::stamp_new;
10 : use crate::state::{now_iso, AppState};
11 : use antares_jsonld::{expand_entity, ExpandOpts};
12 : use antares_model::NgsiError;
13 : use antares_store::CurrentStateDriverExt;
14 : use axum::body::Bytes;
15 : use axum::extract::State;
16 : use axum::http::{HeaderMap, StatusCode};
17 : use axum::response::{IntoResponse, Response};
18 : use serde_json::{json, Value};
19 : use std::collections::HashMap;
20 :
21 : use crate::negotiate::CleanParams;
22 :
23 : /// Parse a batch body: JSON array of entity documents; per-document context
24 : /// resolution (ld+json ⇒ each doc's own @context; json ⇒ Link header).
25 : /// Batch bodies are the ingest hot path — sonic-rs (3–4× parse) behind
26 : /// the `sonic` feature, serde_json always compiled as the fallback.
27 : #[cfg(feature = "sonic")]
28 : fn parse_batch_body(body: &[u8]) -> Result<Value, String> {
29 : sonic_rs::from_slice(body).map_err(|e| e.to_string())
30 : }
31 : #[cfg(not(feature = "sonic"))]
32 338 : fn parse_batch_body(body: &[u8]) -> Result<Value, String> {
33 338 : serde_json::from_slice(body).map_err(|e| e.to_string())
34 338 : }
35 :
36 : /// One entry of a batch array as it arrives: the item and the result of
37 : /// resolving the `@context` that governs it. A failed resolution is carried
38 : /// per item, because 5.6.7 answers the array with one error per entity.
39 : type BatchItem = (Value, ApiResult<std::sync::Arc<antares_jsonld::Context>>);
40 :
41 : /// One entry of a batch as a forward sends it: the object and the `@context`
42 : /// that expanded it.
43 : type FwdItem = (
44 : serde_json::Map<String, Value>,
45 : std::sync::Arc<antares_jsonld::Context>,
46 : );
47 :
48 : /// One batch body into (item, its resolved @context) pairs. Each item's
49 : /// `@context` — or the Link header standing in for it — resolves within the
50 : /// requesting Tenant (5.5.10): a Hosted @context belongs to the Tenant that
51 : /// stored it (5.13.1), so for any other Tenant the URL is unresolvable rather
52 : /// than a set of term mappings the Entity would be expanded through.
53 342 : async fn parse_batch(
54 342 : st: &AppState,
55 342 : tenant: &antares_model::TenantId,
56 342 : headers: &HeaderMap,
57 342 : body: &[u8],
58 342 : ) -> ApiResult<Vec<BatchItem>> {
59 342 : let ct = content_type(headers)?;
60 342 : let ld = match ct.as_str() {
61 342 : "application/json" => false,
62 158 : "application/ld+json" => true,
63 4 : _ => return Err(ApiError::Bare(StatusCode::UNSUPPORTED_MEDIA_TYPE)),
64 : };
65 338 : let value: Value = parse_batch_body(body)
66 338 : .map_err(|e| NgsiError::InvalidRequest(format!("body is not valid JSON: {e}")))?;
67 322 : let items = value.as_array().filter(|a| !a.is_empty()).ok_or_else(|| {
68 18 : NgsiError::BadRequestData("batch body must be a non-empty JSON array".into())
69 18 : })?;
70 : // 5.6.7.4 (and siblings): a null value in ANY item fails the whole
71 : // request with BadRequestData — not a per-item 207 error.
72 304 : if items.iter().any(Value::is_null) {
73 14 : return Err(
74 14 : NgsiError::BadRequestData("batch array must not contain null items".into()).into(),
75 14 : );
76 290 : }
77 : // Batch entity count cap
78 290 : if items.len() > *crate::bounds::MAX_BATCH_ITEMS {
79 4 : return Err(NgsiError::BadRequestData(format!(
80 4 : "batch of {} exceeds the {}-entity limit",
81 4 : items.len(),
82 4 : *crate::bounds::MAX_BATCH_ITEMS
83 4 : ))
84 4 : .into());
85 286 : }
86 286 : let link = link_context(headers)?;
87 286 : if ld && link.is_some() {
88 4 : return Err(NgsiError::BadRequestData(
89 4 : "application/ld+json batch must not also carry a Link @context (6.3.5)".into(),
90 4 : )
91 4 : .into());
92 282 : }
93 282 : let mut out = Vec::new();
94 : // The loader caps ONE @context resolution at MAX_CONTEXT_FETCHES fetched
95 : // documents. A batch resolves once per item, so without a ceiling here
96 : // the item count multiplies that cap: a body inside the 4 MiB limit can
97 : // name a thousand different @contexts and buy a thousand crawls of a
98 : // chosen host from one request. Distinct values are what cost a crawl —
99 : // repeats hit the loader's merged cache, and an inline object fetches
100 : // nothing — so only those are counted.
101 282 : let mut named_contexts: std::collections::HashSet<String> = std::collections::HashSet::new();
102 632 : let mut count_context = |c: &Value| -> Result<(), ApiError> {
103 632 : if matches!(c, Value::Object(_) | Value::Null) {
104 0 : return Ok(());
105 632 : }
106 632 : let key = c.to_string();
107 632 : if named_contexts.contains(&key) {
108 116 : return Ok(());
109 516 : }
110 : // At the ceiling nothing more is remembered either: the set is keyed
111 : // by client-supplied text, so it may not grow past the cap it guards.
112 516 : if named_contexts.len() >= crate::bounds::MAX_CONTEXT_FETCHES {
113 336 : return Err(NgsiError::BadRequestData(format!(
114 336 : "batch names more than {} distinct @contexts",
115 336 : crate::bounds::MAX_CONTEXT_FETCHES
116 336 : ))
117 336 : .into());
118 180 : }
119 180 : named_contexts.insert(key);
120 180 : Ok(())
121 632 : };
122 5640 : for item in items {
123 5640 : let ctx = if ld {
124 636 : match item.get("@context") {
125 632 : Some(c) => match count_context(c) {
126 336 : Err(e) => Err(e),
127 296 : Ok(()) => st
128 296 : .loader
129 296 : .resolve_for(tenant, c)
130 296 : .await
131 296 : .map_err(ApiError::from),
132 : },
133 4 : None => Err(NgsiError::BadRequestData(
134 4 : "ld+json batch entity without @context".into(),
135 4 : )
136 4 : .into()),
137 : }
138 5004 : } else if item.get("@context").is_some() {
139 4 : Err(NgsiError::BadRequestData("application/json entity carries @context".into()).into())
140 : } else {
141 5000 : match &link {
142 12 : Some(url) => st
143 12 : .loader
144 12 : .resolve_for(tenant, &Value::String(url.clone()))
145 12 : .await
146 12 : .map_err(ApiError::from),
147 4988 : None => Ok(st.loader.core()),
148 : }
149 : };
150 5640 : out.push((item.clone(), ctx));
151 : }
152 282 : Ok(out)
153 342 : }
154 :
155 : struct BatchOutcome {
156 : success: Vec<Value>,
157 : errors: Vec<Value>,
158 : }
159 :
160 : impl BatchOutcome {
161 : /// Batch output data (5.6.7.5 / 5.6.8.5 / 5.6.9.5 / 5.6.10.5): if every
162 : /// entity succeeded, the operation's all-ok status (201 + id array for
163 : /// create, else 204 with no body); otherwise 207 with the S array
164 : /// ("success") and the E array of BatchEntityError 5.2.17 ("errors").
165 454 : fn respond(
166 454 : self,
167 454 : tenant: &antares_model::TenantId,
168 454 : all_ok_status: StatusCode,
169 454 : body_on_ok: bool,
170 454 : ) -> Response {
171 454 : if self.errors.is_empty() {
172 104 : if body_on_ok {
173 74 : let mut resp = (
174 74 : all_ok_status,
175 74 : [(axum::http::header::CONTENT_TYPE, "application/json")],
176 74 : axum::Json(Value::Array(self.success)),
177 74 : )
178 74 : .into_response();
179 74 : echo_tenant(tenant, &mut resp);
180 74 : resp
181 : } else {
182 30 : let mut resp = all_ok_status.into_response();
183 30 : echo_tenant(tenant, &mut resp);
184 30 : resp
185 : }
186 : } else {
187 350 : multi_status(
188 350 : json!({"success": self.success, "errors": self.errors}),
189 350 : tenant,
190 : )
191 : }
192 454 : }
193 : }
194 :
195 948 : fn err_entry(id: Option<&str>, e: &NgsiError) -> Value {
196 948 : json!({
197 948 : "entityId": id.unwrap_or("unknown"),
198 948 : "error": problem_value(e),
199 : })
200 948 : }
201 :
202 : /// 5.2.17 BatchEntityError for a failed forwarded part — the remote status
203 : /// and detail travel inside the ProblemDetails (5.6.7.4/5.6.10.4: remote
204 : /// error results merge into E).
205 6 : fn err_remote(id: Option<&str>, status: u16, detail: &str) -> Value {
206 6 : let etype = match status {
207 0 : 400 => "BadRequestData",
208 4 : 404 => "ResourceNotFound",
209 2 : 409 => "AlreadyExists",
210 0 : 422 => "OperationNotSupported",
211 0 : _ => "InternalError",
212 : };
213 6 : json!({
214 6 : "entityId": id.unwrap_or("unknown"),
215 6 : "error": {
216 6 : "type": format!("https://uri.etsi.org/ngsi-ld/errors/{etype}"),
217 6 : "title": "distributed operation failed",
218 6 : "status": status,
219 6 : "detail": detail,
220 : }
221 : })
222 6 : }
223 :
224 : /// Merge one forwarded BATCH operation's outcome into the remote S/E sets
225 : /// (5.6.7.4: "Merge the returned list of Entities successfully created with
226 : /// S. Merge the returned list of Entities in Error with E.").
227 12 : fn merge_remote_batch(
228 12 : status: u16,
229 12 : body: &Value,
230 12 : sent_ids: &[String],
231 12 : created: bool,
232 12 : ok: &mut Vec<(String, bool)>,
233 12 : err: &mut Vec<Value>,
234 12 : ) {
235 : // Only the Entities this broker forwarded may appear in the client's
236 : // S/E arrays: an id the Context Source names but we never sent is not
237 : // part of this request's outcome, and the source's own error text is
238 : // rebuilt from its status rather than relayed verbatim.
239 26 : let mine = |id: &str| sent_ids.iter().any(|s| s == id);
240 12 : match (status, body) {
241 12 : (200..=206, Value::Array(a)) => {
242 10 : for id in a.iter().filter_map(Value::as_str).filter(|i| mine(i)) {
243 6 : ok.push((id.to_owned(), created));
244 6 : }
245 : }
246 : (200..=206, _) => {
247 2 : for id in sent_ids {
248 2 : ok.push((id.clone(), created));
249 2 : }
250 : }
251 4 : (207, Value::Object(o)) => {
252 4 : if let Some(Value::Array(a)) = o.get("success") {
253 8 : for id in a.iter().filter_map(Value::as_str).filter(|i| mine(i)) {
254 4 : ok.push((id.to_owned(), created));
255 4 : }
256 0 : }
257 4 : if let Some(Value::Array(a)) = o.get("errors") {
258 8 : for e in a {
259 8 : let Some(id) = e
260 8 : .get("entityId")
261 8 : .and_then(Value::as_str)
262 8 : .filter(|i| mine(i))
263 : else {
264 4 : continue;
265 : };
266 4 : let remote = e
267 4 : .get("error")
268 4 : .and_then(|p| p.get("status"))
269 4 : .and_then(Value::as_u64)
270 4 : .unwrap_or(u64::from(status)) as u16;
271 4 : err.push(err_remote(
272 4 : Some(id),
273 4 : remote,
274 4 : "forwarded batch operation reported an error for this entity",
275 : ));
276 : }
277 0 : }
278 : }
279 : _ => {
280 0 : for id in sent_ids {
281 0 : err.push(err_remote(
282 0 : Some(id),
283 0 : status,
284 0 : &format!("forwarded batch operation returned {status}"),
285 0 : ));
286 0 : }
287 : }
288 : }
289 12 : }
290 :
291 586 : fn ngsi_of(e: ApiError) -> NgsiError {
292 586 : match e {
293 586 : ApiError::Ngsi(n) => n,
294 0 : ApiError::Bare(code) => NgsiError::BadRequestData(format!("HTTP {code}")),
295 0 : ApiError::NotAcceptable(_) => NgsiError::BadRequestData("HTTP 406".into()),
296 : // One item of a batch cannot answer 503: the array's other entries
297 : // may have succeeded, and 5.6.7.4 gives each failure its own
298 : // ProblemDetails. The item carries the same detail the whole-request
299 : // answer would, so the cause is not lost in the collapse.
300 : ApiError::Overloaded(_) => {
301 0 : NgsiError::InternalError(antares_model::error::DB_OVERLOADED.into())
302 : }
303 : // The policy gate runs once per request, before the array is
304 : // touched, so a refusal is the whole array's 403 and never one
305 : // entry's — this arm is exhaustiveness, not a path. If a per-item
306 : // gate is ever added, 422 is the nearest thing Table 6.3.2-1 has:
307 : // the entry was refused, and the reason travels with it.
308 0 : ApiError::Denied(why) => NgsiError::OperationNotSupported(why),
309 : }
310 586 : }
311 :
312 : // ---------- POST /entityOperations/create (5.6.7) ----------
313 :
314 162 : pub async fn batch_create(
315 162 : State(st): State<AppState>,
316 162 : CleanParams(params): CleanParams,
317 162 : headers: HeaderMap,
318 162 : body: Bytes,
319 162 : ) -> Response {
320 162 : match batch_write(&st, ¶ms, &headers, &body, BatchMode::Create).await {
321 124 : Ok(r) => r,
322 38 : Err(e) => e.into_response(),
323 : }
324 162 : }
325 :
326 94 : pub async fn batch_upsert(
327 94 : State(st): State<AppState>,
328 94 : CleanParams(params): CleanParams,
329 94 : headers: HeaderMap,
330 94 : body: Bytes,
331 94 : ) -> Response {
332 94 : match batch_write(&st, ¶ms, &headers, &body, BatchMode::Upsert).await {
333 84 : Ok(r) => r,
334 10 : Err(e) => e.into_response(),
335 : }
336 94 : }
337 :
338 32 : pub async fn batch_update(
339 32 : State(st): State<AppState>,
340 32 : CleanParams(params): CleanParams,
341 32 : headers: HeaderMap,
342 32 : body: Bytes,
343 32 : ) -> Response {
344 32 : match batch_write(&st, ¶ms, &headers, &body, BatchMode::Update).await {
345 28 : Ok(r) => r,
346 4 : Err(e) => e.into_response(),
347 : }
348 32 : }
349 :
350 : /// 5.6.20 Batch Entity Merge: each entity merged per 5.6.17 locally;
351 : /// 204 when all succeed, 207 with S/E arrays otherwise (5.6.20.5).
352 54 : pub async fn batch_merge(
353 54 : State(st): State<AppState>,
354 54 : CleanParams(params): CleanParams,
355 54 : headers: HeaderMap,
356 54 : body: Bytes,
357 54 : ) -> Response {
358 54 : match batch_write(&st, ¶ms, &headers, &body, BatchMode::Merge).await {
359 38 : Ok(r) => r,
360 16 : Err(e) => e.into_response(),
361 : }
362 54 : }
363 :
364 : #[derive(Clone, Copy, PartialEq)]
365 : enum BatchMode {
366 : Create,
367 : Upsert,
368 : Update,
369 : Merge,
370 : }
371 :
372 : impl BatchMode {
373 : /// The clause the policy seam is asked about: one function serves the
374 : /// four batch write operations, and they are four clauses.
375 24 : const fn clause(self) -> &'static str {
376 24 : match self {
377 4 : Self::Create => "5.6.7",
378 12 : Self::Upsert => "5.6.8",
379 4 : Self::Update => "5.6.9",
380 4 : Self::Merge => "5.6.20",
381 : }
382 24 : }
383 : }
384 :
385 : /// The registration-matching input of a batch (4.3.6.1): the ids, the
386 : /// expanded types and the expanded Attribute names the whole array carries,
387 : /// as one `CsrSpec`, together with the items in the shape a forward sends
388 : /// them. An entry whose body is not an object, or whose `@context` did not
389 : /// resolve, contributes nothing and is left to the local arm to report.
390 282 : fn batch_spec(items: &[BatchItem]) -> (crate::registry::CsrSpec, Vec<FwdItem>) {
391 282 : let mut fwd_items = Vec::new();
392 282 : let mut spec = crate::registry::CsrSpec::default();
393 282 : let (mut types, mut ids, mut attrs) = (Vec::new(), Vec::new(), Vec::new());
394 5640 : for (item, ctx) in items {
395 5640 : let (Some(o), Ok(c)) = (item.as_object(), ctx.as_ref()) else {
396 590 : continue;
397 : };
398 5050 : if let Some(id) = o.get("id").and_then(Value::as_str) {
399 5050 : ids.push(id.to_owned());
400 5050 : }
401 5050 : match o.get("type") {
402 5050 : Some(Value::String(t)) => types.push(c.expand_key(t)),
403 0 : Some(Value::Array(a)) => {
404 0 : types.extend(a.iter().filter_map(Value::as_str).map(|t| c.expand_key(t)))
405 : }
406 0 : _ => {}
407 : }
408 17196 : for k in o.keys() {
409 17196 : if !matches!(k.as_str(), "id" | "type" | "scope" | "@context") {
410 6620 : attrs.push(c.expand_key(k));
411 11540 : }
412 : }
413 5050 : fwd_items.push((o.clone(), c.clone()));
414 : }
415 282 : if !types.is_empty() {
416 148 : spec.types = Some(types);
417 180 : }
418 282 : if !ids.is_empty() {
419 148 : spec.ids = Some(ids);
420 180 : }
421 282 : if !attrs.is_empty() {
422 128 : spec.attrs = Some(attrs);
423 172 : }
424 282 : (spec, fwd_items)
425 282 : }
426 :
427 : /// The distributed arm of a batch write (5.6.7.4, 5.6.8.4, 5.6.9.4,
428 : /// 5.6.20.4): one forwarded request per matching registration, whose
429 : /// outcomes merge into the same S and E arrays the local arm fills, as
430 : /// Entity ids and BatchEntityErrors (5.2.16, 5.2.17) rather than opaque
431 : /// part descriptors.
432 : #[allow(clippy::too_many_arguments)]
433 22 : async fn forward_batch(
434 22 : st: &AppState,
435 22 : params: &HashMap<String, String>,
436 22 : headers: &HeaderMap,
437 22 : mode: BatchMode,
438 22 : tenant: &antares_model::TenantId,
439 22 : update_mode: bool,
440 22 : no_overwrite: bool,
441 22 : fed_regs: &[crate::federation::FedReg],
442 22 : fwd_items: &[FwdItem],
443 22 : out: &mut BatchOutcome,
444 22 : created_ids: &mut Vec<String>,
445 22 : any_created: &mut bool,
446 22 : ) {
447 14 : fn one_outcome(
448 14 : id: &str,
449 14 : status: u16,
450 14 : created: bool,
451 14 : ok: &mut Vec<(String, bool)>,
452 14 : err: &mut Vec<Value>,
453 14 : ) {
454 14 : if (200..300).contains(&status) && status != 207 {
455 12 : ok.push((id.to_owned(), created && status == 201));
456 2 : } else {
457 2 : err.push(err_remote(
458 2 : Some(id),
459 2 : status,
460 2 : &format!("forwarded operation returned {status}"),
461 2 : ));
462 2 : }
463 14 : }
464 22 : let (op, res_path) = match mode {
465 8 : BatchMode::Create => ("createBatch", "create"),
466 8 : BatchMode::Upsert => ("upsertBatch", "upsert"),
467 4 : BatchMode::Update => ("updateBatch", "update"),
468 2 : BatchMode::Merge => ("mergeBatch", "merge"),
469 : };
470 22 : let src = fwd_items
471 22 : .first()
472 22 : .map(|(_, c)| c.source.clone())
473 22 : .unwrap_or(Value::Null);
474 22 : let ctx_url = crate::federation::ctx_link_url(headers, &src);
475 22 : let mut query: Vec<(String, String)> = Vec::new();
476 22 : if let Some(o) = params.get("options") {
477 4 : query.push(("options".into(), o.clone()));
478 18 : }
479 22 : let replace_mode = !update_mode;
480 22 : let mut remote_ok: Vec<(String, bool)> = Vec::new();
481 22 : let mut remote_err: Vec<Value> = Vec::new();
482 22 : for reg in fed_regs {
483 22 : let arr: Vec<Value> = fwd_items
484 22 : .iter()
485 30 : .filter_map(|(o, c)| crate::federation::reduce_to_scope(o, reg, c))
486 22 : .collect();
487 22 : if arr.is_empty() {
488 0 : continue;
489 22 : }
490 22 : let sent_ids: Vec<String> = arr
491 22 : .iter()
492 26 : .filter_map(|e| e.get("id").and_then(Value::as_str).map(str::to_owned))
493 22 : .collect();
494 22 : if reg.supports(op) {
495 4 : let (status, body, _) = crate::federation::forward(
496 4 : st,
497 4 : reqwest::Method::POST,
498 4 : format!("{}/ngsi-ld/v1/entityOperations/{res_path}", reg.endpoint),
499 4 : &query,
500 4 : headers,
501 4 : tenant,
502 4 : reg,
503 4 : &ctx_url,
504 4 : Some(Value::Array(arr)),
505 : )
506 4 : .await;
507 : // 5.6.8.5: an upsert forward only CREATED entities when the
508 : // remote said so — 201 (body lists the created ids); a 204
509 : // means every forwarded entity was updated.
510 4 : merge_remote_batch(
511 4 : status,
512 4 : &body,
513 4 : &sent_ids,
514 4 : mode == BatchMode::Create || (mode == BatchMode::Upsert && status == 201),
515 4 : &mut remote_ok,
516 4 : &mut remote_err,
517 : );
518 4 : continue;
519 18 : }
520 : // 5.6.7.4/5.6.8.4/5.6.9.4/5.6.20.4 single-op fallbacks. These
521 : // never inherit the batch `options` parameter — e.g.
522 : // options=update is no Create Entity parameter (5.6.1.3) and
523 : // would 400 the forward; the append fallback re-adds
524 : // noOverwrite explicitly (5.6.9.4).
525 18 : let mut handled = true;
526 8 : match mode {
527 8 : BatchMode::Create if reg.supports("createEntity") => {
528 2 : let call = BatchFwd::create();
529 4 : for ent in arr.clone() {
530 4 : let (id, status) =
531 4 : forward_one(st, reg, headers, tenant, &ctx_url, ent, &call).await;
532 4 : one_outcome(&id, status, call.created, &mut remote_ok, &mut remote_err);
533 : }
534 : }
535 4 : BatchMode::Upsert if reg.supports("createEntity") => {
536 2 : let create = BatchFwd::create();
537 2 : for ent in arr.clone() {
538 2 : let (id, status) =
539 2 : forward_one(st, reg, headers, tenant, &ctx_url, ent.clone(), &create).await;
540 : // 5.6.8.4: an Entity the peer already holds is not a
541 : // failed upsert — it falls through to the operation
542 : // that updates the one that is there.
543 2 : let fallback = match () {
544 2 : _ if status != 409 => None,
545 2 : _ if replace_mode && reg.supports("replaceEntity") => {
546 2 : Some(BatchFwd::replace())
547 : }
548 0 : _ if !replace_mode && reg.supports("updateEntity") => {
549 0 : Some(BatchFwd::update())
550 : }
551 : _ => {
552 : // 5.6.8.4: neither replace nor update available
553 0 : remote_err.push(err_remote(
554 0 : Some(&id),
555 : 422,
556 0 : &format!("OperationNotSupported: no upsert path for {id}"),
557 : ));
558 0 : continue;
559 : }
560 : };
561 2 : match fallback {
562 0 : None => one_outcome(&id, status, true, &mut remote_ok, &mut remote_err),
563 2 : Some(call) => {
564 2 : let (id, status) =
565 2 : forward_one(st, reg, headers, tenant, &ctx_url, ent, &call).await;
566 2 : one_outcome(&id, status, call.created, &mut remote_ok, &mut remote_err);
567 : }
568 : }
569 : }
570 : }
571 2 : BatchMode::Upsert if replace_mode && reg.supports("replaceEntity") => {
572 0 : let call = BatchFwd::replace();
573 0 : for ent in arr.clone() {
574 0 : let (id, status) =
575 0 : forward_one(st, reg, headers, tenant, &ctx_url, ent, &call).await;
576 0 : one_outcome(&id, status, call.created, &mut remote_ok, &mut remote_err);
577 : }
578 : }
579 2 : BatchMode::Upsert if !replace_mode && reg.supports("updateEntity") => {
580 2 : let call = BatchFwd::update();
581 2 : for ent in arr.clone() {
582 2 : let (id, status) =
583 2 : forward_one(st, reg, headers, tenant, &ctx_url, ent, &call).await;
584 2 : one_outcome(&id, status, call.created, &mut remote_ok, &mut remote_err);
585 : }
586 : }
587 4 : BatchMode::Update if !no_overwrite && reg.supports("updateEntity") => {
588 2 : let call = BatchFwd::update();
589 2 : for ent in arr.clone() {
590 2 : let (id, status) =
591 2 : forward_one(st, reg, headers, tenant, &ctx_url, ent, &call).await;
592 2 : one_outcome(&id, status, call.created, &mut remote_ok, &mut remote_err);
593 : }
594 : }
595 2 : BatchMode::Merge if reg.supports("mergeEntity") => {
596 : // 5.6.20.4 support ladder: no mergeBatch -> per-entity
597 : // Merge Entity (5.6.17) forwards.
598 2 : let call = BatchFwd::merge();
599 2 : for ent in arr.clone() {
600 2 : let (id, status) =
601 2 : forward_one(st, reg, headers, tenant, &ctx_url, ent, &call).await;
602 2 : one_outcome(&id, status, call.created, &mut remote_ok, &mut remote_err);
603 : }
604 : }
605 2 : BatchMode::Update if no_overwrite && reg.supports("appendAttrs") => {
606 : // 5.6.9.4: append with Attribute overwrite disabled.
607 2 : let call = BatchFwd::append_no_overwrite();
608 2 : for ent in arr.clone() {
609 2 : let (id, status) =
610 2 : forward_one(st, reg, headers, tenant, &ctx_url, ent, &call).await;
611 2 : one_outcome(&id, status, call.created, &mut remote_ok, &mut remote_err);
612 : }
613 : }
614 6 : _ => handled = false,
615 : }
616 18 : if !handled && reg.is_proxy() {
617 : // 5.6.7.4/5.6.8.4/5.6.9.4/5.6.20.4 last rung: "In case CSR is
618 : // an exclusive or redirect Context Source Registration, add an
619 : // Error of type Conflict for each Entity in IN to E."
620 8 : for id in &sent_ids {
621 8 : remote_err.push(err_entry(
622 8 : Some(id),
623 8 : &NgsiError::Conflict(format!(
624 8 : "registration does not accept the operation {op}"
625 8 : )),
626 8 : ));
627 8 : }
628 12 : }
629 : }
630 22 : for (id, was_created) in remote_ok {
631 16 : if was_created {
632 6 : *any_created = true;
633 6 : created_ids.push(id.clone());
634 10 : }
635 16 : if !out.success.iter().any(|v| v.as_str() == Some(id.as_str())) {
636 14 : out.success.push(Value::String(id));
637 14 : }
638 : }
639 22 : out.errors.extend(remote_err);
640 22 : }
641 :
642 342 : async fn batch_write(
643 342 : st: &AppState,
644 342 : params: &HashMap<String, String>,
645 342 : headers: &HeaderMap,
646 342 : body: &[u8],
647 342 : mode: BatchMode,
648 342 : ) -> ApiResult<Response> {
649 342 : let tenant = tenant_from(headers)?;
650 342 : check_params(params, &["options", "local"])?;
651 : // 6.3.7: `options` is a comma separated list of strings, so a mode is
652 : // selected when it appears as ONE MEMBER of the list — never by
653 : // whole-string equality. 6.15.3.1 replace (default) | update for
654 : // upsert; 6.16.3.1 noOverwrite for update.
655 684 : let has_option = |name: &str| {
656 684 : params
657 684 : .get("options")
658 684 : .is_some_and(|o| o.split(',').any(|s| s.trim() == name))
659 684 : };
660 342 : let update_mode = has_option("update");
661 342 : let no_overwrite = has_option("noOverwrite");
662 342 : let items = parse_batch(st, &tenant, headers, body).await?;
663 282 : let (spec, fwd_items) = batch_spec(&items);
664 : // ADR-0020: one verdict for the whole array, before any item is written
665 : // and before any forward — the batch is one operation (5.6.7 to 5.6.20)
666 : // and a per-item gate would let half of it land on a refusal. The
667 : // engine is given what the array names, the same members the
668 : // single-Entity write hands it: an engine that decides by Entity id,
669 : // type or Attribute has to reach the same verdict whether the request
670 : // came one Entity at a time or as one array.
671 282 : let gate_ids: Vec<&str> = spec.ids.iter().flatten().map(String::as_str).collect();
672 282 : gate!(
673 : st, &tenant, headers, mode.clause(),
674 : ids: &gate_ids,
675 : types: spec.types.as_deref().unwrap_or(&[]),
676 : attrs: spec.attrs.as_deref().unwrap_or(&[]),
677 : )
678 282 : .await?;
679 274 : let fed_regs =
680 276 : match crate::federation::write_plan(st, &tenant, &spec, &st.loader.core(), params, headers)
681 276 : .await?
682 : {
683 0 : crate::federation::WritePlan::Answered(r) => return Ok(*r),
684 274 : crate::federation::WritePlan::Forward(regs) => regs,
685 : };
686 274 : let mut out = BatchOutcome {
687 274 : success: vec![],
688 274 : errors: vec![],
689 274 : };
690 274 : let mut created_ids: Vec<String> = vec![];
691 274 : let mut any_created = false;
692 274 : let proxies: Vec<&crate::federation::FedReg> =
693 274 : fed_regs.iter().filter(|r| r.is_proxy()).collect();
694 : // Creates are collected and written as ONE multi-row statement;
695 : // upsert/update/merge run batched per round below.
696 : // 5.5.11.0: "All Entities and Attributes in the batch will get the same
697 : // modifiedAt timestamp" — the clock is read once for the whole array, not
698 : // per document. Stamping per document spreads a large create over several
699 : // milliseconds, and a Context Consumer filtering or paging on modifiedAt
700 : // then sees one batch as two. The per-round stamp below is the separate
701 : // case: repeated instances of one id are sequential operations (5.5.11.2,
702 : // 5.5.11.5) and do carry their own instants.
703 274 : let create_ts = now_iso();
704 274 : let mut pending_creates: Vec<(String, Value)> = Vec::new();
705 274 : let mut prepped: Vec<(String, Value)> = Vec::new();
706 5632 : for (item, ctx) in items {
707 5632 : let id_hint = item.get("id").and_then(Value::as_str).map(str::to_owned);
708 5632 : let ctx = match ctx {
709 5046 : Ok(c) => c,
710 586 : Err(e) => {
711 586 : out.errors.push(err_entry(id_hint.as_deref(), &ngsi_of(e)));
712 586 : continue;
713 : }
714 : };
715 : // proxied (exclusive/redirect) attributes are never stored locally
716 5046 : let item = if proxies.is_empty() {
717 5018 : item
718 28 : } else if let Some(o) = item.as_object() {
719 28 : let (rest, has_attrs) = crate::federation::strip_proxied(o, &proxies, &ctx);
720 28 : if !has_attrs {
721 24 : continue; // wholly proxied: no local part for this item
722 4 : }
723 4 : Value::Object(rest)
724 : } else {
725 0 : item
726 : };
727 5022 : if mode == BatchMode::Create {
728 4944 : let prep = || -> Result<(String, Value), NgsiError> {
729 4944 : let obj = item
730 4944 : .as_object()
731 4944 : .ok_or_else(|| NgsiError::BadRequestData("entity must be an object".into()))?;
732 4944 : let mut expanded = expand_entity(obj, &ctx, ExpandOpts::default())?;
733 4940 : let id = antares_jsonld::expanded_id(&expanded)?.to_owned();
734 4940 : stamp_new(&mut expanded, &create_ts);
735 4940 : Ok((id, expanded))
736 4944 : };
737 4944 : match prep() {
738 4940 : Ok(pair) => pending_creates.push(pair),
739 4 : Err(e) => out.errors.push(err_entry(id_hint.as_deref(), &e)),
740 : }
741 4944 : continue;
742 78 : }
743 : // Expansion/validation per item only;
744 : // the store operations run BATCHED below — one transaction per round
745 : // instead of one per item.
746 78 : let prep = || -> Result<(String, Value), NgsiError> {
747 78 : let obj = item
748 78 : .as_object()
749 78 : .ok_or_else(|| NgsiError::BadRequestData("entity must be an object".into()))?;
750 74 : let fragment_ok = mode == BatchMode::Merge;
751 74 : let expanded = expand_entity(
752 74 : obj,
753 74 : &ctx,
754 74 : ExpandOpts {
755 74 : fragment: false,
756 74 : allow_null: fragment_ok,
757 74 : merge: fragment_ok,
758 74 : temporal: false,
759 74 : ..Default::default()
760 74 : },
761 0 : )?;
762 74 : let id = antares_jsonld::expanded_id(&expanded)?.to_owned();
763 74 : Ok((id, expanded))
764 78 : };
765 78 : match prep() {
766 74 : Ok(pair) => prepped.push(pair),
767 4 : Err(e) => out.errors.push(err_entry(id_hint.as_deref(), &e)),
768 : }
769 : }
770 : // 4.6.6: duplicate instances of one Entity in a batch array "shall come
771 : // in chronological order" — first oldest. Sequential semantics are kept
772 : // by splitting into rounds: the Nth occurrence of an id lands in round
773 : // N, rounds execute in order, and within a round every id is unique.
774 274 : let mut rounds: Vec<Vec<(String, Value)>> = Vec::new();
775 : {
776 274 : let mut occurrence: HashMap<String, usize> = HashMap::new();
777 274 : for (id, doc) in prepped {
778 74 : let n = occurrence.entry(id.clone()).or_insert(0);
779 74 : if rounds.len() <= *n {
780 62 : rounds.push(Vec::new());
781 62 : }
782 74 : rounds[*n].push((id, doc));
783 74 : *n += 1;
784 : }
785 : }
786 274 : for round in rounds {
787 62 : let ts = now_iso();
788 62 : match mode {
789 : // Creates are collected into pending_creates above and never
790 : // reach a round; a request handler answers, it does not panic.
791 : BatchMode::Create => {
792 0 : for (id, _) in round {
793 0 : out.errors.push(err_entry(
794 0 : Some(&id),
795 0 : &NgsiError::InternalError(
796 0 : "batch create is handled before the rounds".into(),
797 0 : ),
798 0 : ));
799 0 : }
800 : }
801 : BatchMode::Upsert => {
802 : // options=update: merge into existing rows first; ids that
803 : // turn out absent (or vanish mid-flight) fall through to the
804 : // replace batch — never a silent success (TOCTOU fix).
805 34 : let mut replaces: Vec<(String, Value)> = Vec::new();
806 34 : if update_mode {
807 4 : let ids: Vec<String> = round.iter().map(|(id, _)| id.clone()).collect();
808 4 : let docs: HashMap<&str, &Value> =
809 4 : round.iter().map(|(id, d)| (id.as_str(), d)).collect();
810 4 : let res = st
811 4 : .store
812 4 : .batch_mutate(&tenant, &ids, |id, doc| {
813 4 : merge_into(doc, docs[id], &ts);
814 4 : Ok::<(), NgsiError>(())
815 4 : })
816 4 : .await?;
817 4 : for ((id, expanded), r) in round.iter().zip(res) {
818 4 : match r {
819 0 : Some(Err(e)) => out.errors.push(err_entry(Some(id), &e)),
820 : Some(Ok(())) => {
821 4 : if !out.success.contains(&Value::String(id.clone())) {
822 4 : out.success.push(Value::String(id.clone()));
823 4 : }
824 : }
825 0 : None => replaces.push((id.clone(), expanded.clone())),
826 : }
827 : }
828 30 : } else {
829 30 : replaces = round;
830 30 : }
831 34 : if !replaces.is_empty() {
832 40 : for (_, doc) in replaces.iter_mut() {
833 40 : stamp_new(doc, &ts);
834 40 : }
835 : // ids first, then MOVE the documents into the store —
836 : // the loop below only needs the ids, so the batch payload
837 : // is never deep-cloned on the ingest hot path.
838 40 : let ids: Vec<String> = replaces.iter().map(|(id, _)| id.clone()).collect();
839 30 : let flags = st.store.batch_upsert(&tenant, replaces).await?;
840 40 : for (id, created) in ids.iter().zip(flags) {
841 40 : if created {
842 28 : any_created = true;
843 28 : if !created_ids.contains(id) {
844 28 : created_ids.push(id.clone());
845 28 : }
846 12 : }
847 40 : if !out.success.contains(&Value::String(id.clone())) {
848 36 : out.success.push(Value::String(id.clone()));
849 36 : }
850 : }
851 4 : }
852 : }
853 : BatchMode::Update | BatchMode::Merge => {
854 30 : let ids: Vec<String> = round.iter().map(|(id, _)| id.clone()).collect();
855 28 : let docs: HashMap<&str, &Value> =
856 30 : round.iter().map(|(id, d)| (id.as_str(), d)).collect();
857 : // batch update with noOverwrite: existing attribute instances
858 : // are left alone; if any existed, the entity is a partial
859 : // failure (005_02 ⇒ 207)
860 28 : let mut skipped: HashMap<String, bool> = HashMap::new();
861 28 : let res = st
862 28 : .store
863 28 : .batch_mutate(&tenant, &ids, |id, doc| {
864 8 : let expanded = docs[id];
865 8 : if mode == BatchMode::Update && no_overwrite {
866 : // noOverwrite is instance-level: only instances
867 : // whose datasetId already exists are skipped
868 4 : let target = antares_store::stored_object(doc)?;
869 16 : for (k, v) in antares_jsonld::expanded_object(expanded)? {
870 8 : if matches!(
871 16 : k.as_str(),
872 16 : "id" | "type" | "scope" | "createdAt" | "modifiedAt"
873 : ) {
874 8 : continue;
875 8 : }
876 8 : let incoming: Vec<Value> =
877 8 : v.as_array().cloned().unwrap_or_default();
878 8 : match target.get_mut(k).and_then(Value::as_array_mut) {
879 4 : None => {
880 4 : target.insert(k.clone(), Value::Array(incoming));
881 4 : }
882 4 : Some(cur) => {
883 4 : for ni in incoming {
884 4 : let ds = ni.get("datasetId").and_then(Value::as_str);
885 4 : if cur.iter().any(|ci| {
886 4 : ci.get("datasetId").and_then(Value::as_str) == ds
887 4 : }) {
888 4 : skipped.insert(id.to_owned(), true);
889 4 : } else {
890 0 : cur.push(ni);
891 0 : }
892 : }
893 : }
894 : }
895 : }
896 4 : target.insert("modifiedAt".into(), Value::String(ts.clone()));
897 4 : } else {
898 4 : merge_into(doc, expanded, &ts);
899 4 : }
900 8 : Ok::<(), NgsiError>(())
901 8 : })
902 28 : .await?;
903 30 : for ((id, _), r) in round.iter().zip(res) {
904 8 : match r {
905 22 : None => out.errors.push(err_entry(
906 22 : Some(id),
907 22 : &NgsiError::ResourceNotFound(format!("entity {id} not found")),
908 : )),
909 0 : Some(Err(e)) => out.errors.push(err_entry(Some(id), &e)),
910 : Some(Ok(())) => {
911 8 : if skipped.get(id).copied().unwrap_or(false) {
912 4 : out.errors.push(err_entry(
913 4 : Some(id),
914 4 : &NgsiError::BadRequestData(
915 4 : "some attributes already existed (noOverwrite)".into(),
916 4 : ),
917 4 : ));
918 6 : } else if !out.success.contains(&Value::String(id.clone())) {
919 4 : out.success.push(Value::String(id.clone()));
920 4 : }
921 : }
922 : }
923 : }
924 : }
925 : }
926 : }
927 : // The collected creates, one multi-row statement, one transaction.
928 274 : if !pending_creates.is_empty() {
929 4940 : let ids: Vec<String> = pending_creates.iter().map(|(id, _)| id.clone()).collect();
930 66 : let flags = st.store.batch_create(&tenant, pending_creates).await?;
931 4940 : for (id, created) in ids.iter().zip(flags) {
932 4940 : if created {
933 4936 : any_created = true;
934 4936 : if !created_ids.contains(id) {
935 4936 : created_ids.push(id.clone());
936 4936 : }
937 4936 : if !out.success.contains(&Value::String(id.clone())) {
938 4936 : out.success.push(Value::String(id.clone()));
939 4936 : }
940 4 : } else {
941 4 : out.errors.push(err_entry(
942 4 : Some(id),
943 4 : &NgsiError::AlreadyExists(format!("entity {id} already exists")),
944 4 : ));
945 4 : }
946 : }
947 208 : }
948 : // Distributed arm (5.6.7.4/5.6.8.4/5.6.9.4/5.6.20.4): remote outcomes
949 : // merge into the same S/E arrays as local ones — the response body
950 : // carries Entity IDs and BatchEntityErrors (5.2.16/5.2.17), never
951 : // opaque part descriptors.
952 274 : if !fed_regs.is_empty() {
953 22 : forward_batch(
954 22 : st,
955 22 : params,
956 22 : headers,
957 22 : mode,
958 22 : &tenant,
959 22 : update_mode,
960 22 : no_overwrite,
961 22 : &fed_regs,
962 22 : &fwd_items,
963 22 : &mut out,
964 22 : &mut created_ids,
965 22 : &mut any_created,
966 22 : )
967 22 : .await;
968 252 : }
969 : // 5.6.8.5: the created-only S array is the ALL-SUCCEEDED reading ("if all
970 : // Entities not existing prior to this request have been successfully
971 : // created and the others have been successfully updated"). Once E is
972 : // non-empty the third bullet applies and S is "the list of Entities
973 : // successfully created or updated".
974 274 : if mode == BatchMode::Upsert && any_created && out.errors.is_empty() {
975 20 : out.success = created_ids.into_iter().map(Value::String).collect();
976 254 : }
977 274 : let (status, body_on_ok) = match mode {
978 124 : BatchMode::Create => (StatusCode::CREATED, true),
979 : BatchMode::Upsert => {
980 84 : if any_created {
981 24 : (StatusCode::CREATED, true)
982 : } else {
983 60 : (StatusCode::NO_CONTENT, false)
984 : }
985 : }
986 66 : BatchMode::Update | BatchMode::Merge => (StatusCode::NO_CONTENT, false),
987 : };
988 274 : Ok(out.respond(&tenant, status, body_on_ok))
989 342 : }
990 :
991 : /// Where a forwarded batch item lands on the peer.
992 : #[derive(Clone, Copy)]
993 : enum FwdPath {
994 : Collection,
995 : Entity,
996 : Attrs,
997 : }
998 :
999 : /// The single-Entity request a batch operation forwards when the peer does
1000 : /// not support the batch operation itself — the 5.6.7.4 / 5.6.8.4 / 5.6.9.4 /
1001 : /// 5.6.20.4 fallback ladders. 5.6.7 posts to the collection, 5.6.8 replaces
1002 : /// the Entity or updates its Attributes, 5.6.9 updates them or appends with
1003 : /// overwrite disabled, 5.6.20 merges the Entity; the loop around them is the
1004 : /// same. None of them inherits the batch `options` parameter, which is no
1005 : /// parameter of the single-Entity operation being carried — 5.6.9.4's
1006 : /// noOverwrite is re-added here, explicitly, because that one is.
1007 : struct BatchFwd {
1008 : path: FwdPath,
1009 : method: reqwest::Method,
1010 : query: Vec<(String, String)>,
1011 : /// whether a 201 from the peer counts as a creation in the 5.2.16 S array
1012 : created: bool,
1013 : }
1014 :
1015 : impl BatchFwd {
1016 : /// 5.6.1 Create Entity.
1017 4 : fn create() -> Self {
1018 4 : Self {
1019 4 : path: FwdPath::Collection,
1020 4 : method: reqwest::Method::POST,
1021 4 : query: Vec::new(),
1022 4 : created: true,
1023 4 : }
1024 4 : }
1025 : /// 5.6.4 Replace Entity.
1026 2 : fn replace() -> Self {
1027 2 : Self {
1028 2 : path: FwdPath::Entity,
1029 2 : method: reqwest::Method::PUT,
1030 2 : query: Vec::new(),
1031 2 : created: false,
1032 2 : }
1033 2 : }
1034 : /// 5.6.2 Update Entity Attributes.
1035 4 : fn update() -> Self {
1036 4 : Self {
1037 4 : path: FwdPath::Attrs,
1038 4 : method: reqwest::Method::PATCH,
1039 4 : query: Vec::new(),
1040 4 : created: false,
1041 4 : }
1042 4 : }
1043 : /// 5.6.17 Merge Entity.
1044 2 : fn merge() -> Self {
1045 2 : Self {
1046 2 : path: FwdPath::Entity,
1047 2 : method: reqwest::Method::PATCH,
1048 2 : query: Vec::new(),
1049 2 : created: false,
1050 2 : }
1051 2 : }
1052 : /// 5.6.3 Append Entity Attributes with overwrite disabled (5.6.9.4).
1053 2 : fn append_no_overwrite() -> Self {
1054 2 : Self {
1055 2 : path: FwdPath::Attrs,
1056 2 : method: reqwest::Method::POST,
1057 2 : query: vec![("options".into(), "noOverwrite".into())],
1058 2 : created: false,
1059 2 : }
1060 2 : }
1061 : }
1062 :
1063 : /// Forward one Entity of the batch to one registration (4.3.6.3) and report
1064 : /// the id it was sent under with the status the peer answered. The id is
1065 : /// reported back to the client verbatim, so it is kept raw here and
1066 : /// percent-encoded per RFC 3986 clause 3.3 only where it becomes a path
1067 : /// segment of the forwarded URL.
1068 16 : async fn forward_one(
1069 16 : st: &AppState,
1070 16 : reg: &crate::federation::FedReg,
1071 16 : headers: &HeaderMap,
1072 16 : tenant: &antares_model::TenantId,
1073 16 : ctx_url: &str,
1074 16 : ent: Value,
1075 16 : call: &BatchFwd,
1076 16 : ) -> (String, u16) {
1077 16 : let id = ent
1078 16 : .get("id")
1079 16 : .and_then(Value::as_str)
1080 16 : .unwrap_or_default()
1081 16 : .to_owned();
1082 16 : let seg = crate::federation::path_segment(&id);
1083 16 : let base = ®.endpoint;
1084 16 : let url = match call.path {
1085 6 : FwdPath::Collection => format!("{base}/ngsi-ld/v1/entities"),
1086 4 : FwdPath::Entity => format!("{base}/ngsi-ld/v1/entities/{seg}"),
1087 6 : FwdPath::Attrs => format!("{base}/ngsi-ld/v1/entities/{seg}/attrs"),
1088 : };
1089 16 : let (status, _, _) = crate::federation::forward(
1090 16 : st,
1091 16 : call.method.clone(),
1092 16 : url,
1093 16 : &call.query,
1094 16 : headers,
1095 16 : tenant,
1096 16 : reg,
1097 16 : ctx_url,
1098 16 : Some(ent),
1099 : )
1100 16 : .await;
1101 16 : (id, status)
1102 16 : }
1103 :
1104 : // ---------- POST /entityOperations/delete (5.6.10) ----------
1105 :
1106 266 : pub async fn batch_delete(
1107 266 : State(st): State<AppState>,
1108 266 : CleanParams(params): CleanParams,
1109 266 : headers: HeaderMap,
1110 266 : body: Bytes,
1111 266 : ) -> Response {
1112 266 : let go = async {
1113 266 : let tenant = tenant_from(&headers)?;
1114 266 : check_params(¶ms, &["local"])?;
1115 266 : let ct = content_type(&headers)?;
1116 266 : if ct != "application/json" && ct != "application/ld+json" {
1117 0 : return Err(ApiError::Bare(StatusCode::UNSUPPORTED_MEDIA_TYPE));
1118 266 : }
1119 266 : let value: Value = serde_json::from_slice(&body)
1120 266 : .map_err(|e| NgsiError::InvalidRequest(format!("body is not valid JSON: {e}")))?;
1121 262 : let ids = value.as_array().filter(|a| !a.is_empty()).ok_or_else(|| {
1122 76 : NgsiError::BadRequestData("batch delete body must be a non-empty array".into())
1123 76 : })?;
1124 : // 5.6.10.4: a null item fails the whole request
1125 186 : if ids.iter().any(Value::is_null) {
1126 2 : return Err(NgsiError::BadRequestData(
1127 2 : "batch array must not contain null items".into(),
1128 2 : )
1129 2 : .into());
1130 184 : }
1131 : // Batch entity count cap — the same ceiling the write batches carry,
1132 : // and the one that bounds the per-id forwarded DELETE fan-out below.
1133 184 : if ids.len() > *crate::bounds::MAX_BATCH_ITEMS {
1134 4 : return Err(NgsiError::BadRequestData(format!(
1135 4 : "batch of {} exceeds the {}-entity limit",
1136 4 : ids.len(),
1137 4 : *crate::bounds::MAX_BATCH_ITEMS
1138 4 : ))
1139 4 : .into());
1140 180 : }
1141 : // ADR-0020, as for the write batches: one verdict for the whole
1142 : // array, carrying the Entity ids it names. The body has to be read
1143 : // to know them, so the gate sits behind the parse the way the
1144 : // single-Entity delete sits behind its path parameter.
1145 180 : let gate_ids: Vec<&str> = ids.iter().filter_map(Value::as_str).collect();
1146 180 : gate!(st, &tenant, &headers, "5.6.10", ids: &gate_ids).await?;
1147 180 : let spec = crate::registry::CsrSpec {
1148 328 : ids: Some(gate_ids.iter().map(|id| (*id).to_owned()).collect()),
1149 180 : ..Default::default()
1150 : };
1151 180 : let regs = match crate::federation::write_plan(
1152 180 : &st,
1153 180 : &tenant,
1154 180 : &spec,
1155 180 : &st.loader.core(),
1156 180 : ¶ms,
1157 180 : &headers,
1158 180 : )
1159 180 : .await?
1160 : {
1161 0 : crate::federation::WritePlan::Answered(r) => return Ok(*r),
1162 180 : crate::federation::WritePlan::Forward(regs) => regs,
1163 : };
1164 180 : let mut out = BatchOutcome {
1165 180 : success: vec![],
1166 180 : errors: vec![],
1167 180 : };
1168 : // One multi-row DELETE for the whole batch; flags in input order.
1169 180 : let id_strs: Vec<String> = ids
1170 180 : .iter()
1171 180 : .filter_map(Value::as_str)
1172 180 : .map(str::to_owned)
1173 180 : .collect();
1174 180 : let mut flags = st.store.batch_delete(&tenant, &id_strs).await?.into_iter();
1175 180 : let mut local_ok: std::collections::HashSet<String> = Default::default();
1176 180 : let mut local_miss: Vec<String> = Vec::new();
1177 334 : for id in ids {
1178 334 : let Some(id) = id.as_str() else {
1179 6 : out.errors.push(err_entry(
1180 6 : None,
1181 6 : &NgsiError::BadRequestData("entity id must be a string".into()),
1182 : ));
1183 6 : continue;
1184 : };
1185 328 : if flags.next().unwrap_or(false) {
1186 : // 5.6.10 deletes carry the same temporal-deletion semantics
1187 : // as 5.6.6 — without this, batch-deleted entities live on in
1188 : // the temporal store (the reset's batch delete leaked
1189 : // every prior suite's Buildings into the orderBy queries).
1190 16 : crate::history::mirror_delete_entity(&st, &tenant, id).await;
1191 16 : local_ok.insert(id.to_owned());
1192 312 : } else {
1193 312 : // proxied entities may live remotely only (4.3.6.3) — a
1194 312 : // local miss becomes an error only if no forward covers it.
1195 312 : local_miss.push(id.to_owned());
1196 312 : }
1197 : }
1198 : // 5.6.10.4 support ladder: deleteBatch -> one batch forward; else
1199 : // per-entity Delete Entity forwards; else proxy modes get Conflict
1200 : // per entity. Remote outcomes merge into S/E (never opaque parts).
1201 180 : let mut remote_ok: Vec<(String, bool)> = Vec::new();
1202 180 : let mut remote_err: Vec<Value> = Vec::new();
1203 180 : if !regs.is_empty() {
1204 4 : let ctx_url = crate::federation::ctx_link_url(&headers, &st.loader.core().source);
1205 4 : for reg in ®s {
1206 : // "Remove from IN all Entities not matched by CSR" — an
1207 : // id-scoped registration (exact ids OR idPattern, 5.12)
1208 : // only receives its own ids (4.3.6.1).
1209 4 : let sent_ids: Vec<String> = id_strs
1210 4 : .iter()
1211 4 : .filter(|i| reg.can_match_id(i))
1212 4 : .cloned()
1213 4 : .collect();
1214 4 : if sent_ids.is_empty() {
1215 0 : continue;
1216 4 : }
1217 4 : if reg.supports("deleteBatch") {
1218 0 : let sent_vals: Vec<Value> =
1219 0 : sent_ids.iter().cloned().map(Value::String).collect();
1220 0 : let (status, body, _) = crate::federation::forward(
1221 0 : &st,
1222 0 : reqwest::Method::POST,
1223 0 : format!("{}/ngsi-ld/v1/entityOperations/delete", reg.endpoint),
1224 0 : &[],
1225 0 : &headers,
1226 0 : &tenant,
1227 0 : reg,
1228 0 : &ctx_url,
1229 0 : Some(Value::Array(sent_vals)),
1230 : )
1231 0 : .await;
1232 0 : merge_remote_batch(
1233 0 : status,
1234 0 : &body,
1235 0 : &sent_ids,
1236 : false,
1237 0 : &mut remote_ok,
1238 0 : &mut remote_err,
1239 : );
1240 4 : } else if reg.supports("deleteEntity") {
1241 2 : for id in &sent_ids {
1242 2 : let (status, _, _) = crate::federation::forward(
1243 2 : &st,
1244 2 : reqwest::Method::DELETE,
1245 2 : format!(
1246 : "{}/ngsi-ld/v1/entities/{}",
1247 : reg.endpoint,
1248 2 : crate::federation::path_segment(id.as_str())
1249 : ),
1250 2 : &[],
1251 2 : &headers,
1252 2 : &tenant,
1253 2 : reg,
1254 2 : &ctx_url,
1255 2 : None,
1256 : )
1257 2 : .await;
1258 2 : if (200..300).contains(&status) && status != 207 {
1259 2 : remote_ok.push((id.clone(), false));
1260 2 : } else {
1261 0 : remote_err.push(err_remote(
1262 0 : Some(id),
1263 0 : status,
1264 0 : &format!("forwarded delete returned {status}"),
1265 0 : ));
1266 0 : }
1267 : }
1268 2 : } else if reg.is_proxy() {
1269 : // 5.6.10.4 last rung: an exclusive or redirect CSR that
1270 : // supports neither delete operation contributes an Error
1271 : // of type Conflict for each Entity in IN.
1272 2 : for id in &sent_ids {
1273 2 : remote_err.push(err_entry(
1274 2 : Some(id),
1275 2 : &NgsiError::Conflict(
1276 2 : "registration does not accept the operation deleteBatch".into(),
1277 2 : ),
1278 2 : ));
1279 2 : }
1280 0 : }
1281 : }
1282 176 : }
1283 180 : let remote_success: std::collections::HashSet<String> =
1284 180 : remote_ok.iter().map(|(id, _)| id.clone()).collect();
1285 : // success in input order (local first occurrence), then remote-only
1286 328 : for id in &id_strs {
1287 328 : if local_ok.contains(id) && !out.success.iter().any(|v| v.as_str() == Some(id.as_str()))
1288 16 : {
1289 16 : out.success.push(Value::String(id.clone()));
1290 312 : }
1291 : }
1292 180 : for (id, _) in remote_ok {
1293 2 : if !out.success.iter().any(|v| v.as_str() == Some(id.as_str())) {
1294 2 : out.success.push(Value::String(id));
1295 2 : }
1296 : }
1297 180 : let erred: Vec<String> = remote_err
1298 180 : .iter()
1299 180 : .filter_map(|e| e.get("entityId").and_then(Value::as_str).map(str::to_owned))
1300 180 : .collect();
1301 180 : out.errors.extend(remote_err);
1302 : // 5.6.10.4 local step (5.6.6 limited to local): a missed occurrence
1303 : // is ResourceNotFound unless a FORWARD resolved that id — a local
1304 : // success does not excuse it (5.5.11.4 duplicate-id semantics: the
1305 : // second occurrence of the same id errors).
1306 312 : for id in local_miss {
1307 312 : if !remote_success.contains(&id) && !erred.contains(&id) {
1308 308 : out.errors.push(err_entry(
1309 308 : Some(&id),
1310 308 : &NgsiError::ResourceNotFound(format!("entity {id} not found")),
1311 308 : ));
1312 308 : }
1313 : }
1314 180 : Ok::<_, ApiError>(out.respond(&tenant, StatusCode::NO_CONTENT, false))
1315 266 : };
1316 266 : go.await.unwrap_or_else(|e| e.into_response())
1317 266 : }
1318 :
1319 : // ---------- POST /entityOperations/query (6.23) ----------
1320 :
1321 334 : pub async fn batch_query(
1322 334 : State(st): State<AppState>,
1323 334 : CleanParams(params): CleanParams,
1324 334 : headers: HeaderMap,
1325 334 : body: Bytes,
1326 334 : ) -> Response {
1327 334 : match batch_query_inner(&st, ¶ms, &headers, &body).await {
1328 114 : Ok(r) => r,
1329 220 : Err(e) => e.into_response(),
1330 : }
1331 334 : }
1332 :
1333 334 : async fn batch_query_inner(
1334 334 : st: &AppState,
1335 334 : params: &HashMap<String, String>,
1336 334 : headers: &HeaderMap,
1337 334 : body: &[u8],
1338 334 : ) -> ApiResult<Response> {
1339 334 : let tenant = tenant_from(headers)?;
1340 334 : check_params(
1341 334 : params,
1342 334 : &["limit", "offset", "count", "options", "format", "local"],
1343 0 : )?;
1344 : // POST query IS Query Entities: geo+json is a valid Accept here (6.3.15)
1345 334 : let accept = parse_accept_geo(headers)?;
1346 334 : let filter = gate!(st, &tenant, headers, "5.7.2").await?;
1347 334 : let parsed = parse_body(&st.loader, headers, body, BodyKind::Standard).await?;
1348 334 : let q = parsed.object(NgsiError::BadRequestData(
1349 334 : "query body must be an object".into(),
1350 334 : ))?;
1351 334 : if q.get("type").and_then(Value::as_str) != Some("Query") {
1352 8 : return Err(NgsiError::BadRequestData("body type must be Query (5.2.23)".into()).into());
1353 326 : }
1354 : // Convert Query members into virtual params reusing the GET filter path.
1355 326 : let mut vp: HashMap<String, String> = HashMap::new();
1356 326 : query_doc_params(q, false, &mut vp)?;
1357 152 : if let Some(l) = params.get("local") {
1358 0 : vp.insert("local".into(), l.clone());
1359 152 : }
1360 : // 5.7.2.4 (p. 201): "At least one of the following input data shall be
1361 : // provided: a) selector of Entity Types; b) list of Attribute names,
1362 : // including at least one non-system Attribute; c) NGSI-LD Query,
1363 : // including at least one non-system Attribute; d) NGSI-LD GeoQuery;
1364 : // e) local scope. If none of the above is provided, then an error of type
1365 : // BadRequestData shall be raised (too wide query)." Query Entities is ONE
1366 : // operation: the resource that carries the Query in a body answers to the
1367 : // same behaviour clause as the one that carries it in the URI, and
1368 : // without this a bare `{"type":"Query"}` reads the whole tenant and fans
1369 : // the filterless query out to every matching registration.
1370 152 : let q_ast = vp.get("q").map(|q| antares_ql::parse_q(q)).transpose()?;
1371 : // 5.7.2.4: "if ... the query, geoquery or context source filter are not
1372 : // syntactically valid (as per the referred clauses 4.9 and 4.10) an error
1373 : // of type BadRequestData shall be raised." Registration matching parses
1374 : // the csf again and drops what it cannot parse, which widens the fan-out
1375 : // to the Context Sources the filter was there to exclude — so it is
1376 : // refused here, the way the twin carrying the Query in the URI refuses it.
1377 152 : if let Some(csf) = vp.get("csf") {
1378 4 : antares_ql::parse_q(csf)?;
1379 148 : }
1380 148 : if !crate::entities::qualifies_non_wide(&vp, q_ast.as_ref()) {
1381 12 : return Err(NgsiError::BadRequestData(
1382 12 : "query needs at least one of type, attrs, q, georel (5.7.2.4)".into(),
1383 12 : )
1384 12 : .into());
1385 136 : }
1386 : // ADR-0020: the engine's narrowing joins the Query's own filters here,
1387 : // after the "too wide" judgement — which is about what the client asked
1388 : // for. Everything below reads the narrowed query, the forwarded one
1389 : // included.
1390 136 : let vp = filter.narrow_params(&vp)?;
1391 136 : let fed = if crate::federation::active(&vp)
1392 136 : && !crate::federation::via_loop(
1393 136 : headers,
1394 136 : &crate::federation::alias_for(&st.host_alias, &tenant),
1395 136 : ) {
1396 : // 6.3.17 scopes NGSILD-Warning to GET /entities(/{id}) — collected
1397 : // here for the log only, never emitted on entityOperations/query
1398 136 : let mut warnings = Vec::new();
1399 136 : let fed =
1400 136 : crate::federation::fed_query(st, &tenant, headers, &parsed.ctx, &vp, &mut warnings)
1401 136 : .await?;
1402 136 : for w in &warnings {
1403 0 : tracing::debug!("distributed query warning (batch query): {w}");
1404 : }
1405 136 : fed
1406 : } else {
1407 0 : Vec::new()
1408 : };
1409 132 : let mut matches =
1410 136 : crate::entities::filter_entities_fed(st, &tenant, &vp, &parsed.ctx, fed).await?;
1411 132 : let mut page_params = params.clone();
1412 132 : page_params.extend(vp.clone());
1413 : // 5.2.43 ordering: same 4.23 keys as the GET twin, applied pre-pagination
1414 132 : if let Some(spec) = page_params.get("orderBy") {
1415 34 : crate::paging::order_entities(&mut matches, spec, &page_params, &parsed.ctx)?;
1416 98 : }
1417 120 : let (page, count_hdr, _links) = paginate(
1418 120 : st,
1419 120 : &page_params,
1420 120 : matches,
1421 120 : "/ngsi-ld/v1/entityOperations/query",
1422 0 : )?;
1423 : // body members (pick/omit/attrs/lang/datasetId) shape the representation
1424 : // exactly like their 6.3.7 query-parameter twins
1425 120 : let mut repr = parse_repr(&page_params, &parsed.ctx)?;
1426 116 : crate::repr::narrow_repr(&mut repr, &filter);
1427 116 : let join = crate::entities::parse_join(&vp)?;
1428 114 : crate::entities::check_linked_projection(&repr, &join)?;
1429 114 : let mut payload: Vec<Value> = page
1430 114 : .iter()
1431 116 : .filter_map(|doc| {
1432 112 : let shaped = apply(doc, &repr);
1433 112 : if repr.pick.is_some() && shaped.as_object().is_some_and(|o| o.is_empty()) {
1434 0 : return None;
1435 112 : }
1436 112 : Some(crate::repr::compact_for(&repr, &shaped, &parsed.ctx))
1437 112 : })
1438 114 : .collect();
1439 114 : if let Some((mode, level)) = &join {
1440 : // 4.5.23.1 bounds the WIDTH of the retrieval per REQUEST, so one
1441 : // allowance is spent across the whole page, exactly as the GET twin
1442 : // spends it. Minting a fresh allowance per payload Entity multiplies
1443 : // the ceiling by the page size, which is the request the ceiling
1444 : // exists to bound: a page of densely linked Entities.
1445 2 : let held = crate::entities::contained_by(&page_params);
1446 2 : let mut budget = crate::repr::MAX_JOIN_LOOKUPS;
1447 2 : match mode.as_str() {
1448 2 : "inline" => {
1449 0 : for p in &mut payload {
1450 0 : crate::repr::inline_join_beyond(
1451 0 : st,
1452 0 : &tenant,
1453 0 : &parsed.ctx,
1454 0 : &repr,
1455 0 : p,
1456 0 : *level,
1457 0 : &held,
1458 0 : &mut budget,
1459 0 : )
1460 0 : .await;
1461 : }
1462 : }
1463 2 : "flat" => {
1464 2 : let mut linked = std::collections::BTreeMap::new();
1465 4 : for doc in &page {
1466 4 : crate::repr::collect_flat_beyond(
1467 4 : st,
1468 4 : &tenant,
1469 4 : &repr,
1470 4 : doc,
1471 4 : *level,
1472 4 : &mut linked,
1473 4 : &held,
1474 4 : &mut budget,
1475 4 : )
1476 4 : .await;
1477 : }
1478 4 : let page_ids: Vec<&str> = page.iter().filter_map(|d| d["id"].as_str()).collect();
1479 2000 : for (id, (ldoc, lrepr)) in linked {
1480 2000 : if !page_ids.contains(&id.as_str()) {
1481 2000 : payload.push(crate::repr::compact_for(
1482 2000 : &lrepr,
1483 2000 : &apply(&ldoc, &lrepr),
1484 2000 : &parsed.ctx,
1485 2000 : ));
1486 2000 : }
1487 : }
1488 : }
1489 0 : _ => {}
1490 : }
1491 112 : }
1492 114 : let out = if accept == Accept::GeoJson {
1493 0 : crate::repr::to_geojson_collection(payload, None)
1494 : } else {
1495 114 : Value::Array(payload)
1496 : };
1497 114 : let mut resp = respond_prefer(StatusCode::OK, out, &parsed.ctx, accept, &tenant, headers);
1498 114 : if let Some(total) = count_hdr {
1499 0 : if let Ok(v) = total.to_string().parse() {
1500 0 : resp.headers_mut().insert("NGSILD-Results-Count", v);
1501 0 : }
1502 114 : }
1503 114 : filter.mark_restricted(resp.headers_mut());
1504 114 : Ok(resp)
1505 334 : }
1506 :
1507 : #[cfg(test)]
1508 : mod tests {
1509 : use super::*;
1510 : use axum::body::Body;
1511 : use axum::http::Request;
1512 : use axum::Router;
1513 : use http_body_util::BodyExt;
1514 : use tower::ServiceExt;
1515 :
1516 36 : fn app() -> Router {
1517 36 : crate::router(AppState::new("antares-test".into()))
1518 36 : }
1519 :
1520 124 : async fn post(app: &Router, uri: &str, body: Value) -> Response {
1521 124 : let s = body.to_string();
1522 124 : app.clone()
1523 124 : .oneshot(
1524 124 : Request::post(uri)
1525 124 : .header("Content-Type", "application/json")
1526 124 : .header("Content-Length", s.len())
1527 124 : .body(Body::from(s))
1528 124 : .expect("req"),
1529 124 : )
1530 124 : .await
1531 124 : .expect("resp")
1532 124 : }
1533 :
1534 8 : async fn get_status(app: &Router, uri: &str) -> StatusCode {
1535 8 : app.clone()
1536 8 : .oneshot(Request::get(uri).body(Body::empty()).expect("req"))
1537 8 : .await
1538 8 : .expect("resp")
1539 8 : .status()
1540 8 : }
1541 :
1542 40 : async fn body_json(resp: Response) -> Value {
1543 40 : let bytes = resp.into_body().collect().await.expect("body").to_bytes();
1544 40 : serde_json::from_slice(&bytes).expect("json body")
1545 40 : }
1546 :
1547 : /// 5.7.2.4 (p. 201): "At least one of the following input data shall be
1548 : /// provided: a) selector of Entity Types; b) list of Attribute names,
1549 : /// including at least one non-system Attribute; c) NGSI-LD Query,
1550 : /// including at least one non-system Attribute; d) NGSI-LD GeoQuery;
1551 : /// e) local scope. If none of the above is provided, then an error of
1552 : /// type BadRequestData shall be raised (too wide query)." Query Entities
1553 : /// is one operation, so the resource that carries the Query in a body is
1554 : /// bound by it exactly as the one that carries it in the URI.
1555 : #[tokio::test]
1556 4 : async fn a_too_wide_query_body_is_refused_like_its_uri_twin() {
1557 4 : let app = app();
1558 20 : for wide in [
1559 4 : json!({"type": "Query"}),
1560 4 : // ids and an id pattern alone are the case the clause names as
1561 4 : // insufficient ("it is not possible to retrieve a set of entities
1562 4 : // by only specifying desired Entity identifiers")
1563 4 : json!({"type": "Query", "entities": [{"id": "urn:ngsi-ld:Vehicle:1"}]}),
1564 4 : json!({"type": "Query", "entities": [{"idPattern": ".*"}]}),
1565 4 : // a system Attribute qualifies neither as an attrs list nor as q
1566 4 : json!({"type": "Query", "attrs": ["createdAt"]}),
1567 4 : json!({"type": "Query", "q": "createdAt>\"2020-01-01T00:00:00Z\""}),
1568 4 : ] {
1569 20 : let resp = post(&app, "/ngsi-ld/v1/entityOperations/query", wide.clone()).await;
1570 20 : assert_eq!(
1571 20 : resp.status(),
1572 4 : StatusCode::BAD_REQUEST,
1573 4 : "too wide query accepted: {wide}"
1574 4 : );
1575 20 : let body = body_json(resp).await;
1576 20 : assert_eq!(
1577 20 : body["type"], "https://uri.etsi.org/ngsi-ld/errors/BadRequestData",
1578 4 : "{body}"
1579 4 : );
1580 4 : }
1581 4 :
1582 4 : // Each of the five qualifying inputs on its own is still served.
1583 16 : for ok in [
1584 4 : json!({"type": "Query", "entities": [{"type": "Vehicle"}]}),
1585 4 : json!({"type": "Query", "attrs": ["speed"]}),
1586 4 : json!({"type": "Query", "q": "speed>100"}),
1587 4 : json!({"type": "Query", "geoQ": {"georel": "near;maxDistance==2000",
1588 4 : "geometry": "Point", "coordinates": "[1,2]"}}),
1589 4 : ] {
1590 16 : let resp = post(&app, "/ngsi-ld/v1/entityOperations/query", ok.clone()).await;
1591 16 : assert_eq!(
1592 16 : resp.status(),
1593 4 : StatusCode::OK,
1594 4 : "qualifying query refused: {ok}"
1595 4 : );
1596 4 : }
1597 4 : }
1598 :
1599 : /// A Query member lifted out of the body becomes the parameter its GET
1600 : /// twin carries in the URI, where 6.3.4 caps it — the POST form must not
1601 : /// be the cheap way past that cap.
1602 : #[tokio::test]
1603 4 : async fn query_body_members_are_capped_like_the_uri() {
1604 4 : let app = app();
1605 4 : let huge = "a".repeat(crate::bounds::MAX_URI_BYTES + 1);
1606 4 : let resp = post(
1607 4 : &app,
1608 4 : "/ngsi-ld/v1/entityOperations/query",
1609 4 : json!({"type": "Query", "entities": [{"type": "Vehicle"}], "q": format!("name==\"{huge}\"")}),
1610 4 : ).await
1611 : ;
1612 4 : assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
1613 4 : let body = body_json(resp).await;
1614 4 : assert!(
1615 4 : body["title"].as_str().is_some_and(|t| t.contains("Bad")),
1616 : "{body}"
1617 : );
1618 :
1619 : // the same member just inside the cap is still served
1620 4 : let ok = "a".repeat(64);
1621 4 : let resp = post(
1622 4 : &app,
1623 4 : "/ngsi-ld/v1/entityOperations/query",
1624 4 : json!({"type": "Query", "entities": [{"type": "Vehicle"}], "q": format!("name==\"{ok}\"")}),
1625 4 : ).await
1626 : ;
1627 4 : assert_eq!(resp.status(), StatusCode::OK, "a normal query still works");
1628 :
1629 : // and the array members are assembled under the same cap
1630 4 : let many: Vec<Value> = (0..600)
1631 2400 : .map(|i| Value::String(format!("attribute-with-a-long-name-{i:04}")))
1632 4 : .collect();
1633 4 : let resp = post(
1634 4 : &app,
1635 4 : "/ngsi-ld/v1/entityOperations/query",
1636 4 : json!({"type": "Query", "entities": [{"type": "Vehicle"}], "attrs": many}),
1637 4 : )
1638 4 : .await;
1639 4 : assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
1640 4 : }
1641 :
1642 : /// The same cap again, for the members `ordering` and `aggrParams` are
1643 : /// lifted into. 5.2.43 makes `orderBy` a String[] and 5.2.44 makes
1644 : /// `aggrMethods` one, and both are joined into the flat parameter the GET
1645 : /// twin carries in its URI — where 6.3.4's bare 414 caps them. They went
1646 : /// into the map uncapped, so the POST form handed the 4.23 ordering and
1647 : /// the 4.5.19 aggregation parsers a string the GET form cannot express:
1648 : /// one body inside `MAX_BODY_BYTES` holds ~150 000 order keys, each
1649 : /// expanded once per Entity comparison.
1650 : #[tokio::test]
1651 4 : async fn ordering_and_aggregation_members_are_capped_like_the_uri() {
1652 4 : let app = app();
1653 4 : let many: Vec<Value> = (0..600)
1654 2400 : .map(|i| Value::String(format!("attributeWithALongOrderKeyName{i:04}")))
1655 4 : .collect();
1656 4 : let resp = post(
1657 4 : &app,
1658 4 : "/ngsi-ld/v1/entityOperations/query",
1659 4 : json!({"type": "Query", "entities": [{"type": "Vehicle"}],
1660 4 : "ordering": {"orderBy": many}}),
1661 4 : )
1662 4 : .await;
1663 4 : assert_eq!(
1664 4 : resp.status(),
1665 : StatusCode::BAD_REQUEST,
1666 : "an over-cap orderBy is refused"
1667 : );
1668 :
1669 : // the temporal twin carries aggrParams; its aggrMethods is joined the
1670 : // same way and takes the same cap
1671 4 : let methods: Vec<Value> = (0..600)
1672 2400 : .map(|i| Value::String(format!("totallyUnknownAggregationMethod{i:04}")))
1673 4 : .collect();
1674 4 : let resp = post(
1675 4 : &app,
1676 4 : "/ngsi-ld/v1/temporal/entityOperations/query",
1677 4 : json!({"type": "Query", "entities": [{"type": "Vehicle"}],
1678 4 : "temporalQ": {"timerel": "before", "timeAt": "2026-01-01T00:00:00Z"},
1679 4 : "aggrParams": {"aggrMethods": methods}}),
1680 4 : )
1681 4 : .await;
1682 4 : assert_eq!(
1683 4 : resp.status(),
1684 : StatusCode::BAD_REQUEST,
1685 : "an over-cap aggrMethods is refused"
1686 : );
1687 :
1688 : // an ordinary ordering is untouched
1689 4 : let resp = post(
1690 4 : &app,
1691 4 : "/ngsi-ld/v1/entityOperations/query",
1692 4 : json!({"type": "Query", "entities": [{"type": "Vehicle"}],
1693 4 : "ordering": {"orderBy": ["name", "!speed"], "collation": "sk"}}),
1694 4 : )
1695 4 : .await;
1696 4 : assert_eq!(
1697 4 : resp.status(),
1698 4 : StatusCode::OK,
1699 4 : "a normal ordering still works"
1700 4 : );
1701 4 : }
1702 :
1703 : /// The same cap, for the three parameters the `entities` selectors are
1704 : /// lifted into. 5.2.33 makes `entities` an array of EntitySelectors, and
1705 : /// `query_doc_params` folds their `type`, `id` and `idPattern` members
1706 : /// into the flat parameters the GET twin carries in its URI — where
1707 : /// 6.3.4's bare 414 caps them at `MAX_URI_BYTES`. They went into the map
1708 : /// before the cap existed in the function, so the POST form handed the
1709 : /// expansion and store layers a string the GET form cannot express:
1710 : /// ~320 000 selectors fit inside `MAX_BODY_BYTES`, one expanded IRI and
1711 : /// one bind parameter each, past PostgreSQL's 65 535-parameter limit.
1712 : #[tokio::test]
1713 4 : async fn entity_selector_members_are_capped_like_the_uri() {
1714 4 : let app = app();
1715 4 : let many_types: Vec<Value> = (0..600)
1716 2400 : .map(|i| json!({"type": format!("VeryLongTypeNameForTheCap{i:04}")}))
1717 4 : .collect();
1718 4 : let resp = post(
1719 4 : &app,
1720 4 : "/ngsi-ld/v1/entityOperations/query",
1721 4 : json!({"type": "Query", "entities": many_types}),
1722 4 : )
1723 4 : .await;
1724 4 : assert_eq!(
1725 4 : resp.status(),
1726 : StatusCode::BAD_REQUEST,
1727 : "the joined type parameter is capped like its GET twin"
1728 : );
1729 :
1730 4 : let many_ids: Vec<Value> = (0..400)
1731 1600 : .map(|i| json!({"id": format!("urn:ngsi-ld:Vehicle:a-fairly-long-identifier-{i:06}")}))
1732 4 : .collect();
1733 4 : let resp = post(
1734 4 : &app,
1735 4 : "/ngsi-ld/v1/entityOperations/query",
1736 4 : json!({"type": "Query", "entities": many_ids}),
1737 4 : )
1738 4 : .await;
1739 4 : assert_eq!(
1740 4 : resp.status(),
1741 : StatusCode::BAD_REQUEST,
1742 : "the joined id parameter is capped like its GET twin"
1743 : );
1744 :
1745 4 : let many_pats: Vec<Value> = (0..400)
1746 1600 : .map(|i| json!({"idPattern": format!("^urn:ngsi-ld:Vehicle:pattern-{i:06}.*$")}))
1747 4 : .collect();
1748 4 : let resp = post(
1749 4 : &app,
1750 4 : "/ngsi-ld/v1/entityOperations/query",
1751 4 : json!({"type": "Query", "entities": many_pats}),
1752 4 : )
1753 4 : .await;
1754 4 : assert_eq!(
1755 4 : resp.status(),
1756 : StatusCode::BAD_REQUEST,
1757 : "the joined idPattern parameter is capped like its GET twin"
1758 : );
1759 :
1760 : // The GeoQuery strings are lifted the same way (5.2.13).
1761 4 : let resp = post(
1762 4 : &app,
1763 4 : "/ngsi-ld/v1/entityOperations/query",
1764 4 : json!({"type": "Query", "entities": [{"type": "Vehicle"}],
1765 4 : "geoQ": {"georel": format!("near;maxDistance=={}", "9".repeat(
1766 4 : crate::bounds::MAX_URI_BYTES + 1)),
1767 4 : "geometry": "Point", "coordinates": [0, 0]}}),
1768 4 : )
1769 4 : .await;
1770 4 : assert_eq!(
1771 4 : resp.status(),
1772 : StatusCode::BAD_REQUEST,
1773 : "geoQ georel is capped like its GET twin"
1774 : );
1775 :
1776 : // …but `coordinates` keeps its own ceiling: MAX_GEO_VERTICES, which a
1777 : // legal polygon can spend more than MAX_URI_BYTES on. Capping it in
1778 : // bytes too would refuse a geometry the broker advertises support for.
1779 4 : let ring: Vec<Value> = (0..600)
1780 2400 : .map(|i| json!([f64::from(i) / 10_000.0, f64::from(i) / 10_000.0]))
1781 4 : .chain(std::iter::once(json!([0.0, 0.0])))
1782 4 : .collect();
1783 4 : let coords = json!([ring]);
1784 4 : assert!(
1785 4 : coords.to_string().len() > crate::bounds::MAX_URI_BYTES,
1786 : "the polygon has to exceed the byte cap for this to prove anything"
1787 : );
1788 4 : let resp = post(
1789 4 : &app,
1790 4 : "/ngsi-ld/v1/entityOperations/query",
1791 4 : json!({"type": "Query", "entities": [{"type": "Vehicle"}],
1792 4 : "geoQ": {"georel": "within", "geometry": "Polygon",
1793 4 : "coordinates": coords}}),
1794 4 : )
1795 4 : .await;
1796 4 : assert_eq!(
1797 4 : resp.status(),
1798 : StatusCode::OK,
1799 : "a polygon under MAX_GEO_VERTICES is served whatever it weighs"
1800 : );
1801 :
1802 : // The cap may not cost an ordinary multi-selector query: 5.2.33's
1803 : // union of a handful of selectors stays well inside it.
1804 4 : let resp = post(
1805 4 : &app,
1806 4 : "/ngsi-ld/v1/entityOperations/query",
1807 4 : json!({"type": "Query", "entities": [
1808 4 : {"type": "Vehicle", "id": "urn:ngsi-ld:Vehicle:1"},
1809 4 : {"type": "Vehicle", "id": "urn:ngsi-ld:Vehicle:2"}
1810 4 : ]}),
1811 4 : )
1812 4 : .await;
1813 4 : assert_eq!(resp.status(), StatusCode::OK, "a normal union still works");
1814 4 : }
1815 :
1816 12 : async fn get_entity(app: &Router, id: &str) -> Value {
1817 12 : let resp = app
1818 12 : .clone()
1819 12 : .oneshot(
1820 12 : Request::get(format!("/ngsi-ld/v1/entities/{id}"))
1821 12 : .body(Body::empty())
1822 12 : .expect("req"),
1823 12 : )
1824 12 : .await
1825 12 : .expect("resp");
1826 12 : assert_eq!(resp.status(), StatusCode::OK, "entity {id} readable");
1827 12 : let bytes = resp.into_body().collect().await.expect("body").to_bytes();
1828 12 : serde_json::from_slice(&bytes).expect("json body")
1829 12 : }
1830 :
1831 : /// 6.3.7: `options` is a comma separated list of strings, so the
1832 : /// 6.15.3.1 "update" upsert mode applies whenever it is one member of
1833 : /// the list — existing Entity content is updated, not replaced.
1834 : #[tokio::test]
1835 4 : async fn upsert_update_mode_in_option_list_merges() {
1836 4 : let app = app();
1837 4 : let id = "urn:ngsi-ld:Building:optlist-upsert";
1838 4 : let resp = post(
1839 4 : &app,
1840 4 : "/ngsi-ld/v1/entityOperations/create",
1841 4 : json!([{"id": id, "type": "Building",
1842 4 : "speed": {"type": "Property", "value": 1},
1843 4 : "brand": {"type": "Property", "value": "acme"}}]),
1844 4 : )
1845 4 : .await;
1846 4 : assert_eq!(resp.status(), StatusCode::CREATED);
1847 4 : let resp = post(
1848 4 : &app,
1849 4 : "/ngsi-ld/v1/entityOperations/upsert?options=update,sysAttrs",
1850 4 : json!([{"id": id, "type": "Building",
1851 4 : "speed": {"type": "Property", "value": 2}}]),
1852 4 : )
1853 4 : .await;
1854 4 : assert_eq!(resp.status(), StatusCode::NO_CONTENT);
1855 4 : let doc = get_entity(&app, id).await;
1856 4 : assert_eq!(doc["speed"]["value"], 2, "the payload attribute is applied");
1857 : // update mode must NOT destroy attributes absent from the payload
1858 4 : assert_eq!(
1859 4 : doc["brand"]["value"], "acme",
1860 4 : "update mode keeps attributes not in the payload: {doc}"
1861 4 : );
1862 4 : }
1863 :
1864 : /// 4.22 + 5.6.9.4 end to end. 5.6.9.4 (PDF p.175): "For each of the
1865 : /// NGSI-LD Entities included in the input Array execute the behaviour
1866 : /// defined by clause 5.6.3, but limited to a local operation… If the
1867 : /// Entity update failed, then a new BatchEntityError shall be added to E
1868 : /// containing the failed Entity ID and the ProblemDetails associated."
1869 : /// 4.22 makes an entity past its `expiresAt` invalid, so 5.6.3 answers
1870 : /// ResourceNotFound and the batch has to report it in E — not in S, and
1871 : /// not by writing to it.
1872 : #[tokio::test]
1873 4 : async fn batch_update_of_an_expired_entity_is_an_error_not_a_success() {
1874 4 : let app = app();
1875 4 : let id = "urn:ngsi-ld:Building:expired-batch-update";
1876 4 : let resp = post(
1877 4 : &app,
1878 4 : "/ngsi-ld/v1/entityOperations/create",
1879 4 : json!([{"id": id, "type": "Building", "expiresAt": "2020-01-01T00:00:00Z",
1880 4 : "speed": {"type": "Property", "value": 1}}]),
1881 4 : )
1882 4 : .await;
1883 4 : assert_eq!(resp.status(), StatusCode::CREATED);
1884 :
1885 : // The premise the client can see: it is already absent to a read.
1886 4 : assert_eq!(
1887 4 : get_status(&app, &format!("/ngsi-ld/v1/entities/{id}")).await,
1888 : StatusCode::NOT_FOUND,
1889 : "4.22: an expired entity does not exist"
1890 : );
1891 :
1892 4 : let resp = post(
1893 4 : &app,
1894 4 : "/ngsi-ld/v1/entityOperations/update",
1895 4 : json!([{"id": id, "type": "Building",
1896 4 : "speed": {"type": "Property", "value": 2}}]),
1897 4 : )
1898 4 : .await;
1899 4 : assert_eq!(
1900 4 : resp.status(),
1901 : StatusCode::MULTI_STATUS,
1902 : "5.6.9.5: none updated, so S and E are reported"
1903 : );
1904 4 : let outcome = body_json(resp).await;
1905 4 : assert!(
1906 4 : outcome["success"].as_array().is_none_or(|a| a.is_empty()),
1907 : "an entity every read refuses may not be reported as updated: {outcome}"
1908 : );
1909 4 : let errors = outcome["errors"].as_array().expect("E array");
1910 4 : assert_eq!(errors.len(), 1, "{outcome}");
1911 4 : assert_eq!(errors[0]["entityId"], id, "{outcome}");
1912 4 : assert!(
1913 4 : errors[0]["error"]["type"]
1914 4 : .as_str()
1915 4 : .is_some_and(|t| t.ends_with("ResourceNotFound")),
1916 : "5.6.3 on an absent entity is ResourceNotFound: {outcome}"
1917 : );
1918 :
1919 : // and the write did not happen behind the error
1920 4 : assert_eq!(
1921 4 : get_status(&app, &format!("/ngsi-ld/v1/entities/{id}")).await,
1922 4 : StatusCode::NOT_FOUND,
1923 4 : "still absent"
1924 4 : );
1925 4 : }
1926 :
1927 : /// 6.3.7 + 6.16.3.1: "noOverwrite" as one member of the `options` list
1928 : /// disables Attribute overwrite for Batch Entity Update (5.6.9).
1929 : #[tokio::test]
1930 4 : async fn batch_update_no_overwrite_in_option_list_is_honoured() {
1931 4 : let app = app();
1932 4 : let id = "urn:ngsi-ld:Building:optlist-update";
1933 4 : let resp = post(
1934 4 : &app,
1935 4 : "/ngsi-ld/v1/entityOperations/create",
1936 4 : json!([{"id": id, "type": "Building",
1937 4 : "speed": {"type": "Property", "value": 1}}]),
1938 4 : )
1939 4 : .await;
1940 4 : assert_eq!(resp.status(), StatusCode::CREATED);
1941 4 : let resp = post(
1942 4 : &app,
1943 4 : "/ngsi-ld/v1/entityOperations/update?options=noOverwrite,sysAttrs",
1944 4 : json!([{"id": id, "type": "Building",
1945 4 : "speed": {"type": "Property", "value": 2},
1946 4 : "brand": {"type": "Property", "value": "acme"}}]),
1947 4 : )
1948 4 : .await;
1949 : // `speed` already existed, so the skip is a partial failure: 5.6.9.5
1950 : // makes 207 with the S and E arrays the only correct answer.
1951 4 : assert_eq!(
1952 4 : resp.status(),
1953 : StatusCode::MULTI_STATUS,
1954 : "the skipped instance must be reported"
1955 : );
1956 4 : let outcome = body_json(resp).await;
1957 4 : let errs = outcome["errors"].as_array().expect("E array");
1958 4 : assert_eq!(errs.len(), 1, "{outcome}");
1959 4 : assert_eq!(errs[0]["entityId"], id, "E names the entity: {outcome}");
1960 4 : assert!(
1961 4 : !outcome["success"]
1962 4 : .as_array()
1963 4 : .expect("S array")
1964 4 : .iter()
1965 4 : .any(|v| v.as_str() == Some(id)),
1966 : "a partially skipped entity must not also be in S: {outcome}"
1967 : );
1968 4 : let doc = get_entity(&app, id).await;
1969 4 : assert_eq!(
1970 4 : doc["speed"]["value"], 1,
1971 : "noOverwrite leaves the existing instance alone: {doc}"
1972 : );
1973 4 : assert_ne!(doc["speed"]["value"], 2, "the payload must not overwrite");
1974 4 : assert_eq!(doc["brand"]["value"], "acme", "new attributes are appended");
1975 4 : }
1976 :
1977 : /// 5.6.7.4: what merges into the client's S and E arrays is the outcome
1978 : /// of the Entities this broker forwarded. Ids a Context Source invents
1979 : /// are dropped, and its error text is never relayed verbatim.
1980 : #[tokio::test]
1981 4 : async fn remote_batch_results_are_confined_to_forwarded_ids() {
1982 4 : let sent = vec!["urn:ngsi-ld:Building:mine".to_owned()];
1983 4 : let (mut ok, mut err) = (Vec::new(), Vec::new());
1984 4 : merge_remote_batch(
1985 : 207,
1986 4 : &json!({
1987 4 : "success": ["urn:ngsi-ld:Building:mine", "urn:ngsi-ld:Secret:peer"],
1988 4 : "errors": [
1989 4 : {"entityId": "urn:ngsi-ld:Building:mine",
1990 4 : "error": {"status": 404, "detail": "row 42 of table peer_secrets"}},
1991 4 : {"entityId": "urn:ngsi-ld:Secret:other",
1992 4 : "error": {"status": 409, "detail": "peer internals"}}
1993 4 : ]
1994 4 : }),
1995 4 : &sent,
1996 : false,
1997 4 : &mut ok,
1998 4 : &mut err,
1999 : );
2000 4 : assert_eq!(
2001 : ok,
2002 4 : vec![("urn:ngsi-ld:Building:mine".to_owned(), false)],
2003 : "only forwarded ids reach S"
2004 : );
2005 4 : assert_eq!(err.len(), 1, "only forwarded ids reach E: {err:?}");
2006 4 : assert_eq!(err[0]["entityId"], "urn:ngsi-ld:Building:mine");
2007 4 : let dump = Value::Array(err.clone()).to_string();
2008 4 : assert!(
2009 4 : !dump.contains("Secret") && !dump.contains("peer"),
2010 : "peer ids and error text must not be relayed: {dump}"
2011 : );
2012 4 : assert_eq!(err[0]["error"]["status"], 404, "the remote status travels");
2013 : // a 2xx id list is confined the same way
2014 4 : let (mut ok, mut err) = (Vec::new(), Vec::new());
2015 4 : merge_remote_batch(
2016 : 201,
2017 4 : &json!(["urn:ngsi-ld:Building:mine", "urn:ngsi-ld:Secret:peer"]),
2018 4 : &sent,
2019 : true,
2020 4 : &mut ok,
2021 4 : &mut err,
2022 : );
2023 4 : assert_eq!(ok, vec![("urn:ngsi-ld:Building:mine".to_owned(), true)]);
2024 4 : assert!(err.is_empty());
2025 4 : }
2026 :
2027 : /// 5.6.10.3: the Batch Entity Delete input is "an array of Entity IDs"
2028 : /// with no per-operation exemption from this broker's batch ceiling —
2029 : /// an over-cap array is rejected whole and deletes nothing.
2030 : #[tokio::test]
2031 4 : async fn batch_delete_over_the_item_cap_deletes_nothing() {
2032 4 : let app = app();
2033 4 : let id = "urn:ngsi-ld:Building:delete-cap";
2034 4 : let resp = post(
2035 4 : &app,
2036 4 : "/ngsi-ld/v1/entityOperations/create",
2037 4 : json!([{"id": id, "type": "Building"}]),
2038 4 : )
2039 4 : .await;
2040 4 : assert_eq!(resp.status(), StatusCode::CREATED);
2041 4 : let mut ids: Vec<Value> = (0..*crate::bounds::MAX_BATCH_ITEMS)
2042 4000 : .map(|i| Value::String(format!("urn:ngsi-ld:Building:cap-{i}")))
2043 4 : .collect();
2044 4 : ids.push(Value::String(id.to_owned()));
2045 4 : let resp = post(
2046 4 : &app,
2047 4 : "/ngsi-ld/v1/entityOperations/delete",
2048 4 : Value::Array(ids),
2049 4 : )
2050 4 : .await;
2051 4 : assert_eq!(resp.status(), StatusCode::BAD_REQUEST, "over-cap delete");
2052 4 : let doc = body_json(resp).await;
2053 4 : assert!(
2054 4 : doc["type"]
2055 4 : .as_str()
2056 4 : .expect("problem type")
2057 4 : .ends_with("BadRequestData"),
2058 : "{doc}"
2059 : );
2060 : // rejected whole: the in-range entity of the same array survives
2061 4 : get_entity(&app, id).await;
2062 4 : }
2063 :
2064 : /// 5.6.8.5 third bullet: when only some Entities succeeded, S is "the
2065 : /// list of Entities successfully created **or updated**" — the
2066 : /// created-only list of the second bullet is the all-succeeded case.
2067 : #[tokio::test]
2068 4 : async fn upsert_207_success_carries_updated_entities_too() {
2069 4 : let app = app();
2070 4 : let existing = "urn:ngsi-ld:Building:upsert-207-old";
2071 4 : let fresh = "urn:ngsi-ld:Building:upsert-207-new";
2072 4 : let resp = post(
2073 4 : &app,
2074 4 : "/ngsi-ld/v1/entityOperations/create",
2075 4 : json!([{"id": existing, "type": "Building"}]),
2076 4 : )
2077 4 : .await;
2078 4 : assert_eq!(resp.status(), StatusCode::CREATED);
2079 4 : let resp = post(
2080 4 : &app,
2081 4 : "/ngsi-ld/v1/entityOperations/upsert",
2082 4 : json!([
2083 4 : {"id": existing, "type": "Building",
2084 4 : "speed": {"type": "Property", "value": 1}},
2085 4 : {"id": fresh, "type": "Building"},
2086 4 : "not an entity"
2087 4 : ]),
2088 4 : )
2089 4 : .await;
2090 4 : assert_eq!(resp.status(), StatusCode::MULTI_STATUS);
2091 4 : let doc = body_json(resp).await;
2092 4 : let s: Vec<&str> = doc["success"]
2093 4 : .as_array()
2094 4 : .expect("S array")
2095 4 : .iter()
2096 4 : .filter_map(Value::as_str)
2097 4 : .collect();
2098 4 : assert!(s.contains(&existing), "the updated Entity is in S: {doc}");
2099 4 : assert!(s.contains(&fresh), "the created Entity is in S: {doc}");
2100 4 : assert_eq!(doc["errors"].as_array().expect("E array").len(), 1, "{doc}");
2101 4 : }
2102 : }
|
