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 : mod delete;
6 : mod forward;
7 : mod query;
8 : #[cfg(test)]
9 : mod tests;
10 : mod write;
11 :
12 : pub use delete::batch_delete;
13 : use forward::*;
14 : pub use query::batch_query;
15 : pub use write::{batch_create, batch_merge, batch_update, batch_upsert};
16 :
17 : use crate::entities::merge_into;
18 : use crate::negotiate::*;
19 : use crate::paging::{paginate, query_doc_params};
20 : use crate::repr::{apply, parse_repr};
21 : use crate::stamp::stamp_new;
22 : use crate::state::{now_iso, AppState};
23 : use antares_jsonld::{expand_entity, ExpandOpts};
24 : use antares_model::NgsiError;
25 : use antares_store::CurrentStateDriverExt;
26 : use axum::body::Bytes;
27 : use axum::extract::State;
28 : use axum::http::{HeaderMap, StatusCode};
29 : use axum::response::{IntoResponse, Response};
30 : use serde_json::{json, Value};
31 : use std::collections::HashMap;
32 :
33 : use crate::negotiate::CleanParams;
34 :
35 : /// Parse a batch body: JSON array of entity documents; per-document context
36 : /// resolution (ld+json ⇒ each doc's own @context; json ⇒ Link header).
37 : /// Batch bodies are the ingest hot path — sonic-rs (3–4× parse) behind
38 : /// the `sonic` feature, serde_json always compiled as the fallback.
39 : #[cfg(feature = "sonic")]
40 : fn parse_batch_body(body: &[u8]) -> Result<Value, String> {
41 : sonic_rs::from_slice(body).map_err(|e| e.to_string())
42 : }
43 : #[cfg(not(feature = "sonic"))]
44 578 : fn parse_batch_body(body: &[u8]) -> Result<Value, String> {
45 578 : serde_json::from_slice(body).map_err(|e| e.to_string())
46 578 : }
47 :
48 : /// One entry of a batch array as it arrives: the item and the result of
49 : /// resolving the `@context` that governs it. A failed resolution is carried
50 : /// per item, because 5.6.7 answers the array with one error per entity.
51 : type BatchItem = (Value, ApiResult<std::sync::Arc<antares_jsonld::Context>>);
52 :
53 : /// One entry of a batch as a forward sends it: the object and the `@context`
54 : /// that expanded it.
55 : type FwdItem = (
56 : serde_json::Map<String, Value>,
57 : std::sync::Arc<antares_jsonld::Context>,
58 : );
59 :
60 : /// One batch body into (item, its resolved @context) pairs. Each item's
61 : /// `@context` — or the Link header standing in for it — resolves within the
62 : /// requesting Tenant (5.5.10): a Hosted @context belongs to the Tenant that
63 : /// stored it (5.13.1), so for any other Tenant the URL is unresolvable rather
64 : /// than a set of term mappings the Entity would be expanded through.
65 582 : async fn parse_batch(
66 582 : st: &AppState,
67 582 : tenant: &antares_model::TenantId,
68 582 : headers: &HeaderMap,
69 582 : body: &[u8],
70 582 : ) -> ApiResult<Vec<BatchItem>> {
71 582 : let ct = content_type(headers)?;
72 582 : let ld = match ct.as_str() {
73 582 : "application/json" => false,
74 300 : "application/ld+json" => true,
75 4 : _ => return Err(ApiError::Bare(StatusCode::UNSUPPORTED_MEDIA_TYPE)),
76 : };
77 578 : let value: Value = parse_batch_body(body)
78 578 : .map_err(|e| NgsiError::InvalidRequest(format!("body is not valid JSON: {e}")))?;
79 562 : let items = value.as_array().filter(|a| !a.is_empty()).ok_or_else(|| {
80 18 : NgsiError::BadRequestData("batch body must be a non-empty JSON array".into())
81 18 : })?;
82 : // 5.6.7.4 (and siblings): a null value in ANY item fails the whole
83 : // request with BadRequestData — not a per-item 207 error.
84 544 : if items.iter().any(Value::is_null) {
85 14 : return Err(
86 14 : NgsiError::BadRequestData("batch array must not contain null items".into()).into(),
87 14 : );
88 530 : }
89 : // Batch entity count cap
90 530 : if items.len() > *crate::bounds::MAX_BATCH_ITEMS {
91 4 : return Err(NgsiError::BadRequestData(format!(
92 4 : "batch of {} exceeds the {}-entity limit",
93 4 : items.len(),
94 4 : *crate::bounds::MAX_BATCH_ITEMS
95 4 : ))
96 4 : .into());
97 526 : }
98 526 : let link = link_context(headers)?;
99 526 : if ld && link.is_some() {
100 4 : return Err(NgsiError::BadRequestData(
101 4 : "application/ld+json batch must not also carry a Link @context (6.3.5)".into(),
102 4 : )
103 4 : .into());
104 522 : }
105 522 : let mut out = Vec::new();
106 : // The loader caps ONE @context resolution at MAX_CONTEXT_FETCHES fetched
107 : // documents. A batch resolves once per item, so without a ceiling here
108 : // the item count multiplies that cap: a body inside the 4 MiB limit can
109 : // name a thousand different @contexts and buy a thousand crawls of a
110 : // chosen host from one request. Distinct values are what cost a crawl —
111 : // repeats hit the loader's merged cache, and an inline object fetches
112 : // nothing — so only those are counted.
113 522 : let mut named_contexts: std::collections::HashSet<String> = std::collections::HashSet::new();
114 980 : let mut count_context = |c: &Value| -> Result<(), ApiError> {
115 : // an array names each of its URLs: count them one by one, inline
116 : // objects fetch nothing
117 980 : let named: std::collections::HashSet<&str> = match c {
118 416 : Value::String(u) => std::iter::once(u.as_str()).collect(),
119 564 : Value::Array(a) => a.iter().filter_map(Value::as_str).collect(),
120 0 : _ => std::collections::HashSet::new(),
121 : };
122 980 : let fresh: Vec<&str> = named
123 980 : .into_iter()
124 2976 : .filter(|u| !named_contexts.contains(*u))
125 980 : .collect();
126 : // Past the ceiling nothing more is remembered either: the set is
127 : // keyed by client-supplied text, so it may not grow past the cap it
128 : // guards.
129 980 : if named_contexts.len().saturating_add(fresh.len()) > crate::bounds::MAX_CONTEXT_FETCHES {
130 398 : return Err(NgsiError::BadRequestData(format!(
131 398 : "batch names more than {} distinct @contexts",
132 398 : crate::bounds::MAX_CONTEXT_FETCHES
133 398 : ))
134 398 : .into());
135 582 : }
136 582 : named_contexts.extend(fresh.into_iter().map(str::to_owned));
137 582 : Ok(())
138 980 : };
139 6280 : for item in items {
140 6280 : let ctx = if ld {
141 984 : match item.get("@context") {
142 980 : Some(c) => match count_context(c) {
143 398 : Err(e) => Err(e),
144 582 : Ok(()) => st
145 582 : .loader
146 582 : .resolve_for(tenant, c)
147 582 : .await
148 582 : .map_err(ApiError::from),
149 : },
150 4 : None => Err(NgsiError::BadRequestData(
151 4 : "ld+json batch entity without @context".into(),
152 4 : )
153 4 : .into()),
154 : }
155 5296 : } else if item.get("@context").is_some() {
156 4 : Err(NgsiError::BadRequestData("application/json entity carries @context".into()).into())
157 : } else {
158 5292 : match &link {
159 12 : Some(url) => st
160 12 : .loader
161 12 : .resolve_for(tenant, &Value::String(url.clone()))
162 12 : .await
163 12 : .map_err(ApiError::from),
164 5280 : None => Ok(st.loader.core()),
165 : }
166 : };
167 6280 : out.push((item.clone(), ctx));
168 : }
169 522 : Ok(out)
170 582 : }
171 :
172 : struct BatchOutcome {
173 : success: Vec<Value>,
174 : errors: Vec<Value>,
175 : }
176 :
177 : impl BatchOutcome {
178 834 : const fn new() -> Self {
179 834 : Self {
180 834 : success: Vec::new(),
181 834 : errors: Vec::new(),
182 834 : }
183 834 : }
184 :
185 : /// `id` joins S once, however many members or forwards succeeded for it.
186 6400 : fn succeed(&mut self, id: &str) {
187 1246932 : if !self.success.iter().any(|v| v.as_str() == Some(id)) {
188 6184 : self.success.push(Value::String(id.to_owned()));
189 6184 : }
190 6400 : }
191 :
192 : /// Batch output data (5.6.7.5 / 5.6.8.5 / 5.6.9.5 / 5.6.10.5): if every
193 : /// entity succeeded, the operation's all-ok status (201 + id array for
194 : /// create, else 204 with no body); otherwise 207 with the S array
195 : /// ("success") and the E array of BatchEntityError 5.2.17 ("errors").
196 810 : fn respond(
197 810 : self,
198 810 : tenant: &antares_model::TenantId,
199 810 : all_ok_status: StatusCode,
200 810 : body_on_ok: bool,
201 810 : ) -> Response {
202 810 : if self.errors.is_empty() {
203 586 : if body_on_ok {
204 200 : let mut resp = (
205 200 : all_ok_status,
206 200 : [(axum::http::header::CONTENT_TYPE, "application/json")],
207 200 : axum::Json(Value::Array(self.success)),
208 200 : )
209 200 : .into_response();
210 200 : echo_tenant(tenant, &mut resp);
211 200 : resp
212 : } else {
213 386 : let mut resp = all_ok_status.into_response();
214 386 : echo_tenant(tenant, &mut resp);
215 386 : resp
216 : }
217 : } else {
218 224 : multi_status(
219 224 : json!({"success": self.success, "errors": self.errors}),
220 224 : tenant,
221 : )
222 : }
223 810 : }
224 : }
225 :
226 : /// 5.2.17 BatchEntityError for a member the policy engine refused: the
227 : /// AccessDenied problem a refused single-Entity write answers with (ADR-0020).
228 12 : fn denied_entry(id: Option<&str>, why: &str) -> Value {
229 12 : json!({
230 12 : "entityId": id.unwrap_or("unknown"),
231 12 : "error": {
232 12 : "type": crate::policy::ACCESS_DENIED_TYPE,
233 12 : "title": crate::policy::ACCESS_DENIED_TITLE,
234 12 : "status": StatusCode::FORBIDDEN.as_u16(),
235 12 : "detail": why,
236 : },
237 : })
238 12 : }
239 :
240 : /// The reason a batch is refused whole: every member was refused, so the
241 : /// answer is the 403 a single refusal gives, carrying the first reason.
242 72 : fn whole_refusal(refused: &[Result<crate::policy::Filter, String>]) -> Option<String> {
243 72 : if refused.iter().all(Result::is_err) {
244 10 : refused.first().and_then(|r| r.clone().err())
245 : } else {
246 62 : None
247 : }
248 72 : }
249 :
250 : /// The conditions the engine set on the allowed members, by id.
251 : /// One id named by several members holds to every condition any of them
252 : /// carries: a round writes each id once, so the strictest reading is the
253 : /// one that never applies a member's write past its own condition.
254 : #[derive(Default)]
255 : struct Conditions(HashMap<String, Vec<crate::policy::Condition>>);
256 :
257 : impl Conditions {
258 248 : fn add(&mut self, id: Option<&str>, granted: &crate::policy::Filter) {
259 248 : if let (Some(id), Some(c)) = (id, crate::policy::Condition::of(granted)) {
260 36 : self.0.entry(id.to_owned()).or_default().push(c);
261 212 : }
262 248 : }
263 :
264 246 : fn holds(&self, id: &str, doc: &Value) -> bool {
265 246 : self.0
266 246 : .get(id)
267 246 : .is_none_or(|cs| cs.iter().all(|c| c.holds(doc)))
268 246 : }
269 :
270 : /// Inside the store closure: a member whose stored Entity a condition
271 : /// does not select is not found, as an absent one is.
272 240 : fn pre(&self, id: &str, doc: &Value) -> Result<(), NgsiError> {
273 240 : if self.holds(id, doc) {
274 228 : Ok(())
275 : } else {
276 12 : Err(Self::not_found(id))
277 : }
278 240 : }
279 :
280 : /// Whether a condition holds `id`: an upsert of it may update an Entity
281 : /// the condition selects and never creates one (5.6.8).
282 244 : fn has(&self, id: &str) -> bool {
283 244 : self.0.contains_key(id)
284 244 : }
285 :
286 : /// The one answer for a conditioned member that is absent or not
287 : /// selected, so the two cannot be told apart.
288 16 : fn not_found(id: &str) -> NgsiError {
289 16 : NgsiError::ResourceNotFound(format!("entity {id} not found"))
290 16 : }
291 :
292 816 : fn unforwarded(&self, forwards: bool) -> Result<(), crate::policy::Denied> {
293 816 : if forwards && !self.0.is_empty() {
294 2 : Err(crate::policy::Denied(
295 2 : crate::policy::CONDITION_NOT_FORWARDED.to_owned(),
296 2 : ))
297 : } else {
298 814 : Ok(())
299 : }
300 816 : }
301 : }
302 :
303 588 : fn err_entry(id: Option<&str>, e: &NgsiError) -> Value {
304 588 : json!({
305 588 : "entityId": id.unwrap_or("unknown"),
306 588 : "error": problem_value(e),
307 : })
308 588 : }
309 :
310 : #[derive(Clone, Copy, PartialEq)]
311 : enum BatchMode {
312 : Create,
313 : Upsert,
314 : Update,
315 : Merge,
316 : }
317 :
318 : impl BatchMode {
319 : /// The clause the policy seam is asked about: one function serves the
320 : /// four batch write operations, and they are four clauses.
321 248 : const fn clause(self) -> &'static str {
322 248 : match self {
323 186 : Self::Create => "5.6.7",
324 36 : Self::Upsert => "5.6.8",
325 14 : Self::Update => "5.6.9",
326 12 : Self::Merge => "5.6.20",
327 : }
328 248 : }
329 :
330 : /// The operation a registration names to take the whole array (4.3.6.1).
331 130 : const fn batch_op(self) -> &'static str {
332 130 : match self {
333 34 : Self::Create => "createBatch",
334 42 : Self::Upsert => "upsertBatch",
335 36 : Self::Update => "updateBatch",
336 18 : Self::Merge => "mergeBatch",
337 : }
338 130 : }
339 :
340 : /// The `/entityOperations/{path}` resource of the operation.
341 108 : const fn batch_path(self) -> &'static str {
342 108 : match self {
343 22 : Self::Create => "create",
344 38 : Self::Upsert => "upsert",
345 32 : Self::Update => "update",
346 16 : Self::Merge => "merge",
347 : }
348 108 : }
349 : }
350 :
351 : /// The write a batch asks for: the operation and the 6.3.7 `options` that
352 : /// select its mode, 6.15.3.1 update (else replace) for upsert and 6.16.3.1
353 : /// noOverwrite for update.
354 : #[derive(Clone, Copy)]
355 : struct WriteOpts {
356 : mode: BatchMode,
357 : update: bool,
358 : no_overwrite: bool,
359 : }
|
