Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Batch Entity Create, Upsert, Update and Merge (5.6.7, 5.6.8, 5.6.9,
3 : //! 5.6.20): one [`BatchRun`] takes the array through its phases in order,
4 : //! the per-member decision, the expansion of each member, the chronological
5 : //! rounds, the one multi-row create, the distributed arm and the S/E outcome.
6 :
7 : use super::*;
8 :
9 : // ---------- POST /entityOperations/create (5.6.7) ----------
10 :
11 210 : pub async fn batch_create(
12 210 : State(st): State<AppState>,
13 210 : CleanParams(params): CleanParams,
14 210 : headers: HeaderMap,
15 210 : body: Bytes,
16 210 : ) -> Response {
17 210 : match batch_write(&st, ¶ms, &headers, &body, BatchMode::Create).await {
18 170 : Ok(r) => r,
19 40 : Err(e) => e.into_response(),
20 : }
21 210 : }
22 :
23 202 : pub async fn batch_upsert(
24 202 : State(st): State<AppState>,
25 202 : CleanParams(params): CleanParams,
26 202 : headers: HeaderMap,
27 202 : body: Bytes,
28 202 : ) -> Response {
29 202 : match batch_write(&st, ¶ms, &headers, &body, BatchMode::Upsert).await {
30 188 : Ok(r) => r,
31 14 : Err(e) => e.into_response(),
32 : }
33 202 : }
34 :
35 96 : pub async fn batch_update(
36 96 : State(st): State<AppState>,
37 96 : CleanParams(params): CleanParams,
38 96 : headers: HeaderMap,
39 96 : body: Bytes,
40 96 : ) -> Response {
41 96 : match batch_write(&st, ¶ms, &headers, &body, BatchMode::Update).await {
42 88 : Ok(r) => r,
43 8 : Err(e) => e.into_response(),
44 : }
45 96 : }
46 :
47 : /// 5.6.20 Batch Entity Merge: each entity merged per 5.6.17 locally;
48 : /// 204 when all succeed, 207 with S/E arrays otherwise (5.6.20.5).
49 74 : pub async fn batch_merge(
50 74 : State(st): State<AppState>,
51 74 : CleanParams(params): CleanParams,
52 74 : headers: HeaderMap,
53 74 : body: Bytes,
54 74 : ) -> Response {
55 74 : match batch_write(&st, ¶ms, &headers, &body, BatchMode::Merge).await {
56 58 : Ok(r) => r,
57 16 : Err(e) => e.into_response(),
58 : }
59 74 : }
60 :
61 : /// The 6.3.7 `options` of a batch write: a comma separated list of strings,
62 : /// so a mode is selected when it appears as ONE MEMBER of the list, never by
63 : /// whole-string equality. 6.15.3.1 replace (default) | update for upsert;
64 : /// 6.16.3.1 noOverwrite for update.
65 582 : fn write_opts(params: &HashMap<String, String>, mode: BatchMode) -> WriteOpts {
66 1164 : let has_option = |name: &str| {
67 1164 : params
68 1164 : .get("options")
69 1164 : .is_some_and(|o| o.split(',').any(|s| s.trim() == name))
70 1164 : };
71 582 : WriteOpts {
72 582 : mode,
73 582 : update: has_option("update"),
74 582 : no_overwrite: has_option("noOverwrite"),
75 582 : }
76 582 : }
77 :
78 582 : async fn batch_write(
79 582 : st: &AppState,
80 582 : params: &HashMap<String, String>,
81 582 : headers: &HeaderMap,
82 582 : body: &[u8],
83 582 : mode: BatchMode,
84 582 : ) -> ApiResult<Response> {
85 582 : let tenant = tenant_from(headers)?;
86 582 : check_params(params, &["options", "local"])?;
87 : // a batch runs only inside the in-flight budget, held until it answers.
88 582 : let _admitted = crate::bounds::admit_batch(body.len())
89 582 : .await
90 582 : .ok_or(ApiError::Overloaded(crate::negotiate::RETRY_AFTER_SECONDS))?;
91 582 : let opts = write_opts(params, mode);
92 582 : let mut items = parse_batch(st, &tenant, headers, body).await?;
93 522 : let mut run = BatchRun {
94 522 : st,
95 522 : tenant,
96 522 : opts,
97 522 : out: BatchOutcome::new(),
98 522 : conds: Conditions::default(),
99 522 : created_ids: Vec::new(),
100 522 : any_created: false,
101 522 : };
102 522 : run.decide(headers, &mut items).await?;
103 510 : let (spec, fwd_items) = batch_spec(&items);
104 510 : let fed_regs = match crate::federation::write_plan(
105 510 : st,
106 510 : &run.tenant,
107 510 : &spec,
108 510 : &st.loader.core(),
109 510 : params,
110 510 : headers,
111 510 : )
112 510 : .await?
113 : {
114 0 : crate::federation::WritePlan::Answered(r) => return Ok(*r),
115 508 : crate::federation::WritePlan::Forward(regs) => regs,
116 : };
117 508 : run.conds.unforwarded(!fed_regs.is_empty())?;
118 506 : let proxies: Vec<&crate::federation::FedReg> =
119 506 : fed_regs.iter().filter(|r| r.is_proxy()).collect();
120 506 : let (creates, prepped) = run.prep(items, &proxies);
121 506 : let prepped = run.new_comma_ids(prepped).await?;
122 506 : for round in rounds(prepped) {
123 300 : run.round(round).await?;
124 : }
125 504 : run.create(creates).await?;
126 : // Distributed arm (5.6.7.4/5.6.8.4/5.6.9.4/5.6.20.4): remote outcomes
127 : // merge into the same S/E arrays as local ones — the response body
128 : // carries Entity IDs and BatchEntityErrors (5.2.16/5.2.17), never
129 : // opaque part descriptors.
130 504 : if !fed_regs.is_empty() {
131 106 : let remote = forward_batch(
132 106 : st,
133 106 : params,
134 106 : headers,
135 106 : &run.tenant,
136 106 : opts,
137 106 : &fed_regs,
138 106 : &fwd_items,
139 106 : )
140 106 : .await;
141 106 : run.merge_remote(remote);
142 398 : }
143 504 : Ok(run.finish())
144 582 : }
145 :
146 : /// A member as the engine is shown it: its id, its pre-image and its
147 : /// expanded body.
148 : type Shown = (Option<String>, crate::policy::PreImage, Option<Value>);
149 :
150 : /// One member of a batch, as the store writes it: the id and the expanded
151 : /// document.
152 : type Member = (String, Value);
153 :
154 : /// One batch write in flight: what its phases read and fill.
155 : struct BatchRun<'a> {
156 : st: &'a AppState,
157 : tenant: antares_model::TenantId,
158 : opts: WriteOpts,
159 : out: BatchOutcome,
160 : /// The conditions the engine set on the allowed members.
161 : conds: Conditions,
162 : /// The ids the batch created, local and forwarded, each once (5.6.8.5).
163 : created_ids: Vec<String>,
164 : any_created: bool,
165 : }
166 :
167 : impl BatchRun<'_> {
168 : /// ADR-0020 per member: before any item is written and before
169 : /// any forward, the engine decides each member as the single-Entity write
170 : /// would — its id, its types (stored and written), its Attributes and its
171 : /// expanded body. A refused member lands in E with the AccessDenied
172 : /// problem and the rest are applied (207); a batch refused whole is the
173 : /// 403 it always was. An entry that does not expand is shown by its id
174 : /// alone and left to the local arm to report.
175 522 : async fn decide(&mut self, headers: &HeaderMap, items: &mut Vec<BatchItem>) -> ApiResult<()> {
176 522 : let st = self.st;
177 522 : let Some(engine) = &st.policy else {
178 462 : return Ok(());
179 : };
180 60 : let shown = self.pre_images(members(items)).await?;
181 56 : let attrs: Vec<Vec<String>> = shown
182 56 : .iter()
183 248 : .map(|(_, _, b)| written_attrs(b.as_ref()))
184 56 : .collect();
185 56 : let ids: Vec<Vec<&str>> = shown
186 56 : .iter()
187 248 : .map(|(id, _, _)| id.as_deref().into_iter().collect())
188 56 : .collect();
189 56 : let ops: Vec<crate::policy::Operation<'_>> = shown
190 56 : .iter()
191 56 : .zip(&ids)
192 56 : .zip(&attrs)
193 56 : .map(|(((_, pre, body), ids), attrs)| crate::policy::Operation {
194 248 : ids,
195 248 : types: &pre.types,
196 248 : attrs,
197 248 : body: body.as_ref(),
198 248 : stored: pre.stored.as_ref(),
199 248 : ..crate::policy::Operation::new(self.opts.mode.clause())
200 248 : })
201 56 : .collect();
202 56 : let asking = crate::snapshots::asking_tenant(st, &self.tenant).await;
203 56 : let refused = crate::policy::gate_each(engine.as_ref(), &asking, headers, &ops).await;
204 56 : drop(ops);
205 56 : if let Some(why) = whole_refusal(&refused) {
206 8 : return Err(crate::policy::Denied(why).into());
207 48 : }
208 48 : let mut verdicts = refused.into_iter().zip(shown);
209 48 : let (out, conds) = (&mut self.out, &mut self.conds);
210 238 : items.retain(|_| match verdicts.next() {
211 10 : Some((Err(why), (id, _, _))) => {
212 10 : out.errors.push(denied_entry(id.as_deref(), &why));
213 10 : false
214 : }
215 228 : Some((Ok(granted), (id, _, _))) => {
216 228 : conds.add(id.as_deref(), &granted);
217 228 : true
218 : }
219 0 : None => true,
220 238 : });
221 48 : Ok(())
222 522 : }
223 :
224 : /// Each member with its pre-image, from one read for every member's
225 : /// stored Entity and, under a view, every Relationship target the batch
226 : /// names; a store failure fails the batch (fail-closed,
227 : /// F-ANT2-009), never reads as absent.
228 60 : async fn pre_images(
229 60 : &mut self,
230 60 : members: Vec<(Option<String>, Option<Value>)>,
231 60 : ) -> Result<Vec<Shown>, NgsiError> {
232 252 : let mut wanted: Vec<&str> = members.iter().filter_map(|(id, _)| id.as_deref()).collect();
233 60 : wanted.sort_unstable();
234 60 : wanted.dedup();
235 60 : let stored = if wanted.is_empty() {
236 0 : HashMap::new()
237 : } else {
238 60 : self.st.store.get_entities(&self.tenant, &wanted).await?
239 : };
240 56 : let shown: Vec<Shown> = members
241 56 : .into_iter()
242 248 : .map(|(id, body)| {
243 248 : let pre = match &id {
244 248 : Some(id) => crate::policy::image(stored.get(id).cloned(), body.as_ref()),
245 0 : None => crate::policy::PreImage::default(),
246 : };
247 248 : (id, pre, body)
248 248 : })
249 56 : .collect();
250 56 : Ok(shown)
251 60 : }
252 :
253 : /// Expansion and validation per member, ahead of every store operation:
254 : /// the creates, and the members the rounds write. A member that fails
255 : /// lands in E; one wholly proxied has no local part.
256 506 : fn prep(
257 506 : &mut self,
258 506 : items: Vec<BatchItem>,
259 506 : proxies: &[&crate::federation::FedReg],
260 506 : ) -> (Vec<Member>, Vec<Member>) {
261 : // Creates are collected and written as ONE multi-row statement;
262 : // upsert/update/merge run batched per round.
263 : // 5.5.11.0: "All Entities and Attributes in the batch will get the same
264 : // modifiedAt timestamp" — the clock is read once for the whole array, not
265 : // per document. Stamping per document spreads a large create over several
266 : // milliseconds, and a Context Consumer filtering or paging on modifiedAt
267 : // then sees one batch as two. The per-round stamp is the separate
268 : // case: repeated instances of one id are sequential operations (5.5.11.2,
269 : // 5.5.11.5) and do carry their own instants.
270 506 : let create_ts = now_iso();
271 506 : let (mut creates, mut prepped) = (Vec::new(), Vec::new());
272 6252 : for (item, ctx) in items {
273 6252 : let id_hint = item.get("id").and_then(Value::as_str).map(str::to_owned);
274 6252 : let ctx = match ctx {
275 5836 : Ok(c) => c,
276 416 : Err(e) => {
277 416 : self.out
278 416 : .errors
279 416 : .push(err_entry(id_hint.as_deref(), &ngsi_of(e)));
280 416 : continue;
281 : }
282 : };
283 5836 : let Some(item) = local_part(item, proxies, &ctx) else {
284 92 : continue; // wholly proxied: no local part for this item
285 : };
286 5744 : let (prep, into) = if self.opts.mode == BatchMode::Create {
287 5212 : (prep_create(&item, &ctx, &create_ts), &mut creates)
288 : } else {
289 532 : (prep_write(&item, &ctx, self.opts.mode), &mut prepped)
290 : };
291 5744 : match prep {
292 5716 : Ok(pair) => into.push(pair),
293 28 : Err(e) => self.out.errors.push(err_entry(id_hint.as_deref(), &e)),
294 : }
295 : }
296 506 : (creates, prepped)
297 506 : }
298 :
299 : /// An upsert creates what is missing, so the create rule on a comma id
300 : /// holds for a member whose Entity is not stored; one
301 : /// already stored under such an id is still updated, so a source that
302 : /// keeps upserting a legacy comma id does not turn red.
303 506 : async fn new_comma_ids(&mut self, prepped: Vec<Member>) -> Result<Vec<Member>, NgsiError> {
304 506 : let comma: Vec<&str> = prepped
305 506 : .iter()
306 526 : .map(|(id, _)| id.as_str())
307 526 : .filter(|id| id.contains(','))
308 506 : .collect();
309 506 : if self.opts.mode != BatchMode::Upsert || comma.is_empty() {
310 498 : return Ok(prepped);
311 8 : }
312 8 : let stored = self.st.store.get_entities(&self.tenant, &comma).await?;
313 8 : let mut kept = Vec::with_capacity(prepped.len());
314 8 : for (id, doc) in prepped {
315 8 : match antares_jsonld::new_entity_id(&doc) {
316 8 : Err(e) if !stored.contains_key(&id) => {
317 4 : self.out.errors.push(err_entry(Some(&id), &e));
318 4 : }
319 4 : _ => kept.push((id, doc)),
320 : }
321 : }
322 8 : Ok(kept)
323 506 : }
324 :
325 : /// One round of an upsert, update or merge, every id in it unique, its
326 : /// store operations batched: one transaction per round instead of one
327 : /// per item.
328 300 : async fn round(&mut self, round: Vec<Member>) -> Result<(), NgsiError> {
329 300 : let ts = now_iso();
330 300 : match self.opts.mode {
331 : // Creates are collected apart and never reach a round; a request
332 : // handler answers, it does not panic.
333 : BatchMode::Create => {
334 0 : for (id, _) in round {
335 0 : self.out.errors.push(err_entry(
336 0 : Some(&id),
337 0 : &NgsiError::InternalError(
338 0 : "batch create is handled before the rounds".into(),
339 0 : ),
340 0 : ));
341 0 : }
342 0 : Ok(())
343 : }
344 180 : BatchMode::Upsert => self.upsert(round, &ts).await,
345 120 : BatchMode::Update | BatchMode::Merge => self.update(round, &ts).await,
346 : }
347 300 : }
348 :
349 180 : async fn upsert(&mut self, round: Vec<Member>, ts: &str) -> Result<(), NgsiError> {
350 180 : let replaces = if self.opts.update {
351 48 : self.upsert_merge(round, ts).await?
352 : } else {
353 132 : self.replace_conditioned(round, ts).await?
354 : };
355 180 : if !replaces.is_empty() {
356 146 : self.replace(replaces, ts).await?;
357 34 : }
358 178 : Ok(())
359 180 : }
360 :
361 : /// options=update: update existing rows first (Update Attributes); ids that turn out
362 : /// absent (or vanish mid-flight) fall through to the replace batch —
363 : /// never a silent success (TOCTOU fix). The members left to replace.
364 48 : async fn upsert_merge(
365 48 : &mut self,
366 48 : round: Vec<Member>,
367 48 : ts: &str,
368 48 : ) -> Result<Vec<Member>, NgsiError> {
369 74 : let ids: Vec<String> = round.iter().map(|(id, _)| id.clone()).collect();
370 74 : let docs: HashMap<&str, &Value> = round.iter().map(|(id, d)| (id.as_str(), d)).collect();
371 48 : let conds = &self.conds;
372 48 : let res = self
373 48 : .st
374 48 : .store
375 54 : .batch_mutate(&self.tenant, &ids, |id, doc| {
376 54 : crate::history::restage(id);
377 54 : conds.pre(id, doc)?;
378 50 : let src = member(&docs, id)?;
379 50 : let before = doc.clone();
380 : // 10.3.3.4: update mode runs Update Attributes (10.2.3.4),
381 : // which replaces each instance it names whole
382 50 : crate::entities::update_into(doc, src, ts);
383 : // a member that only restates the stored state
384 : // (a parked vehicle, a poll that saw nothing
385 : // new) is no change (4.8, 5.8.3): handing the
386 : // store the document unchanged writes no row,
387 : // version, history point or notification
388 50 : if same_but_stamps(&before, doc) {
389 6 : *doc = before;
390 44 : }
391 50 : Ok::<(), NgsiError>(())
392 54 : })
393 48 : .await;
394 48 : let res = match res {
395 48 : Ok(res) => res,
396 0 : Err(e) => return self.step_failed(&ids, e).map(|()| Vec::new()),
397 : };
398 48 : crate::history::settle_all(&ids, &res);
399 48 : let mut replaces = Vec::new();
400 74 : for ((id, expanded), r) in round.iter().zip(res) {
401 20 : match r {
402 4 : Some(Err(e)) => self.out.errors.push(err_entry(Some(id), &e)),
403 50 : Some(Ok(())) => self.out.succeed(id),
404 20 : None if self.conds.has(id) => self
405 2 : .out
406 2 : .errors
407 2 : .push(err_entry(Some(id), &Conditions::not_found(id))),
408 18 : None => replaces.push((id.clone(), expanded.clone())),
409 : }
410 : }
411 48 : Ok(replaces)
412 48 : }
413 :
414 : /// A replace-upsert: a conditioned member is replaced only inside the
415 : /// store's read of it, where its condition is judged. The free members,
416 : /// left to replace.
417 132 : async fn replace_conditioned(
418 132 : &mut self,
419 132 : round: Vec<Member>,
420 132 : ts: &str,
421 132 : ) -> Result<Vec<Member>, NgsiError> {
422 132 : let (conditioned, free): (Vec<_>, Vec<_>) =
423 224 : round.into_iter().partition(|(id, _)| self.conds.has(id));
424 132 : if conditioned.is_empty() {
425 128 : return Ok(free);
426 4 : }
427 8 : let ids: Vec<String> = conditioned.iter().map(|(id, _)| id.clone()).collect();
428 4 : let docs: HashMap<&str, &Value> =
429 8 : conditioned.iter().map(|(id, d)| (id.as_str(), d)).collect();
430 4 : let conds = &self.conds;
431 4 : let res = self
432 4 : .st
433 4 : .store
434 6 : .batch_mutate(&self.tenant, &ids, |id, doc| {
435 6 : conds.pre(id, doc)?;
436 2 : let src = member(&docs, id)?;
437 2 : *doc = replacement(src, doc, ts);
438 2 : Ok::<(), NgsiError>(())
439 6 : })
440 4 : .await;
441 4 : let res = match res {
442 4 : Ok(res) => res,
443 0 : Err(e) => return self.step_failed(&ids, e).map(|()| free),
444 : };
445 8 : for (id, r) in ids.iter().zip(res) {
446 6 : match r {
447 2 : Some(Ok(())) => self.out.succeed(id),
448 4 : Some(Err(e)) => self.out.errors.push(err_entry(Some(id), &e)),
449 2 : None => self
450 2 : .out
451 2 : .errors
452 2 : .push(err_entry(Some(id), &Conditions::not_found(id))),
453 : }
454 : }
455 4 : Ok(free)
456 132 : }
457 :
458 : /// The replace batch of an upsert round: absent ids are created.
459 146 : async fn replace(&mut self, mut replaces: Vec<Member>, ts: &str) -> Result<(), NgsiError> {
460 234 : for (_, doc) in replaces.iter_mut() {
461 234 : stamp_new(doc, ts);
462 234 : }
463 : // ids first, then MOVE the documents into the store —
464 : // the loop below only needs the ids, so the batch payload
465 : // is never deep-cloned on the ingest hot path.
466 234 : let ids: Vec<String> = replaces.iter().map(|(id, _)| id.clone()).collect();
467 146 : let flags = match self.st.store.batch_upsert(&self.tenant, replaces).await {
468 142 : Ok(flags) => flags,
469 4 : Err(e) => return self.step_failed(&ids, e),
470 : };
471 230 : for (id, created) in ids.into_iter().zip(flags) {
472 230 : self.out.succeed(&id);
473 230 : if created {
474 112 : self.created(id);
475 124 : }
476 : }
477 142 : Ok(())
478 146 : }
479 :
480 : /// One round of a Batch Update or Merge.
481 120 : async fn update(&mut self, round: Vec<Member>, ts: &str) -> Result<(), NgsiError> {
482 210 : let ids: Vec<String> = round.iter().map(|(id, _)| id.clone()).collect();
483 210 : let docs: HashMap<&str, &Value> = round.iter().map(|(id, d)| (id.as_str(), d)).collect();
484 : // batch update with noOverwrite: existing attribute instances
485 : // are left alone; if any existed, the entity is a partial
486 : // failure (005_02 ⇒ 207)
487 120 : let mut skipped: std::collections::HashSet<String> = Default::default();
488 120 : let (conds, opts) = (&self.conds, self.opts);
489 120 : let res = self
490 120 : .st
491 120 : .store
492 180 : .batch_mutate(&self.tenant, &ids, |id, doc| {
493 180 : crate::history::restage(id);
494 180 : conds.pre(id, doc)?;
495 176 : let expanded = member(&docs, id)?;
496 176 : if opts.mode == BatchMode::Update && opts.no_overwrite {
497 46 : if append_new_instances(doc, expanded, ts)? {
498 20 : skipped.insert(id.to_owned());
499 26 : }
500 130 : } else if opts.mode == BatchMode::Update {
501 62 : crate::entities::update_into(doc, expanded, ts);
502 68 : } else {
503 68 : merge_into(doc, expanded, ts);
504 68 : }
505 176 : Ok::<(), NgsiError>(())
506 180 : })
507 120 : .await;
508 120 : let res = match res {
509 120 : Ok(res) => res,
510 0 : Err(e) => return self.step_failed(&ids, e),
511 : };
512 120 : crate::history::settle_all(&ids, &res);
513 210 : for ((id, _), r) in round.iter().zip(res) {
514 176 : match r {
515 30 : None => self.out.errors.push(err_entry(
516 30 : Some(id),
517 30 : &NgsiError::ResourceNotFound(format!("entity {id} not found")),
518 : )),
519 4 : Some(Err(e)) => self.out.errors.push(err_entry(Some(id), &e)),
520 176 : Some(Ok(())) if skipped.contains(id) => self.out.errors.push(err_entry(
521 20 : Some(id),
522 20 : &NgsiError::BadRequestData(
523 20 : "some attributes already existed (noOverwrite)".into(),
524 20 : ),
525 : )),
526 156 : Some(Ok(())) => self.out.succeed(id),
527 : }
528 : }
529 120 : Ok(())
530 120 : }
531 :
532 : /// The collected creates, one multi-row statement, one transaction.
533 504 : async fn create(&mut self, creates: Vec<Member>) -> Result<(), NgsiError> {
534 504 : if creates.is_empty() {
535 372 : return Ok(());
536 132 : }
537 5204 : let ids: Vec<String> = creates.iter().map(|(id, _)| id.clone()).collect();
538 132 : let flags = match self.st.store.batch_create(&self.tenant, creates).await {
539 132 : Ok(flags) => flags,
540 0 : Err(e) => return self.step_failed(&ids, e),
541 : };
542 5204 : for (id, created) in ids.into_iter().zip(flags) {
543 5204 : if created {
544 5190 : self.out.succeed(&id);
545 5190 : self.created(id);
546 5190 : } else {
547 14 : self.out.errors.push(err_entry(
548 14 : Some(&id),
549 14 : &NgsiError::AlreadyExists(format!("entity {id} already exists")),
550 14 : ));
551 14 : }
552 : }
553 132 : Ok(())
554 504 : }
555 :
556 : /// A store error in one step (one transaction) of the batch. Before any
557 : /// member succeeded it is the request's answer, as it always was; after
558 : /// one did, that member's write stands, so the members of this step go to
559 : /// E with the error and the batch goes on to its 207 (5.6.7.4, 5.6.8.4:
560 : /// a BatchEntityError per Entity whose write failed). A whole-request
561 : /// error there told the client nothing was applied, and its retry
562 : /// replayed what was.
563 4 : fn step_failed(&mut self, ids: &[String], e: NgsiError) -> Result<(), NgsiError> {
564 4 : if self.out.success.is_empty() {
565 2 : return Err(e);
566 2 : }
567 2 : for id in ids {
568 2 : self.out.errors.push(err_entry(Some(id), &e));
569 2 : }
570 2 : Ok(())
571 4 : }
572 :
573 : /// What the forwards brought back joins S and E after the local outcome.
574 106 : fn merge_remote(&mut self, remote: Remote) {
575 216 : for (id, was_created) in remote.ok {
576 216 : self.out.succeed(&id);
577 216 : if was_created {
578 42 : // 5.6.8.4: one Entity ID per Entity, the local and the forwarded
579 42 : // part of one Entity included
580 42 : self.created(id);
581 174 : }
582 : }
583 106 : self.out.errors.extend(remote.err);
584 106 : }
585 :
586 5344 : fn created(&mut self, id: String) {
587 5344 : self.any_created = true;
588 5344 : if !self.created_ids.contains(&id) {
589 5326 : self.created_ids.push(id);
590 5326 : }
591 5344 : }
592 :
593 : /// 5.6.7.5, 5.6.8.5, 5.6.9.5, 5.6.20.5: the all-ok status, else 207.
594 504 : fn finish(mut self) -> Response {
595 : // 5.6.8.5: the created-only S array is the ALL-SUCCEEDED reading ("if all
596 : // Entities not existing prior to this request have been successfully
597 : // created and the others have been successfully updated"). Once E is
598 : // non-empty the third bullet applies and S is "the list of Entities
599 : // successfully created or updated".
600 504 : if self.opts.mode == BatchMode::Upsert && self.any_created && self.out.errors.is_empty() {
601 78 : self.out.success = self.created_ids.into_iter().map(Value::String).collect();
602 426 : }
603 504 : let (status, body_on_ok) = match self.opts.mode {
604 170 : BatchMode::Create => (StatusCode::CREATED, true),
605 88 : BatchMode::Upsert if self.any_created => (StatusCode::CREATED, true),
606 : BatchMode::Upsert | BatchMode::Update | BatchMode::Merge => {
607 246 : (StatusCode::NO_CONTENT, false)
608 : }
609 : };
610 504 : self.out.respond(&self.tenant, status, body_on_ok)
611 504 : }
612 : }
613 :
614 : /// Each entry's id and, when it expands, its expanded body: what the engine
615 : /// is shown of it.
616 60 : fn members(items: &[BatchItem]) -> Vec<(Option<String>, Option<Value>)> {
617 60 : items
618 60 : .iter()
619 252 : .map(|(item, ctx)| {
620 252 : let id = item.get("id").and_then(Value::as_str).map(str::to_owned);
621 252 : let body = match (item.as_object(), ctx.as_ref()) {
622 252 : (Some(o), Ok(c)) => expand_entity(o, c, ExpandOpts::default()).ok(),
623 0 : _ => None,
624 : };
625 252 : (id, body)
626 252 : })
627 60 : .collect()
628 60 : }
629 :
630 : /// The Attribute names an expanded member writes.
631 248 : fn written_attrs(body: Option<&Value>) -> Vec<String> {
632 248 : body.and_then(Value::as_object)
633 248 : .map(|o| {
634 248 : o.keys()
635 738 : .filter(|k| !matches!(k.as_str(), "id" | "type" | "scope" | "@context"))
636 248 : .cloned()
637 248 : .collect()
638 248 : })
639 248 : .unwrap_or_default()
640 248 : }
641 :
642 : /// The part of a member stored here: proxied (exclusive/redirect)
643 : /// attributes are never stored locally. None when every attribute is.
644 5836 : fn local_part(
645 5836 : item: Value,
646 5836 : proxies: &[&crate::federation::FedReg],
647 5836 : ctx: &antares_jsonld::Context,
648 5836 : ) -> Option<Value> {
649 5836 : if proxies.is_empty() {
650 5696 : return Some(item);
651 140 : }
652 140 : let Some(o) = item.as_object() else {
653 0 : return Some(item);
654 : };
655 140 : let (rest, has_attrs) = crate::federation::strip_proxied(o, proxies, ctx);
656 140 : has_attrs.then_some(Value::Object(rest))
657 5836 : }
658 :
659 : /// A create member, expanded and stamped with the batch's one instant.
660 5212 : fn prep_create(item: &Value, ctx: &antares_jsonld::Context, ts: &str) -> Result<Member, NgsiError> {
661 5212 : let obj = item
662 5212 : .as_object()
663 5212 : .ok_or_else(|| NgsiError::BadRequestData("entity must be an object".into()))?;
664 5212 : let mut expanded = expand_entity(obj, ctx, ExpandOpts::default())?;
665 5208 : let id = antares_jsonld::new_entity_id(&expanded)?.to_owned();
666 5204 : stamp_new(&mut expanded, ts);
667 5204 : Ok((id, expanded))
668 5212 : }
669 :
670 : /// An upsert, update or merge member, expanded; a merge allows the null
671 : /// that deletes an Attribute.
672 532 : fn prep_write(
673 532 : item: &Value,
674 532 : ctx: &antares_jsonld::Context,
675 532 : mode: BatchMode,
676 532 : ) -> Result<Member, NgsiError> {
677 532 : let obj = item
678 532 : .as_object()
679 532 : .ok_or_else(|| NgsiError::BadRequestData("entity must be an object".into()))?;
680 528 : let fragment_ok = mode == BatchMode::Merge;
681 528 : let expanded = expand_entity(
682 528 : obj,
683 528 : ctx,
684 528 : ExpandOpts {
685 528 : fragment: false,
686 528 : allow_null: fragment_ok,
687 528 : merge: fragment_ok,
688 528 : temporal: false,
689 528 : ..Default::default()
690 528 : },
691 16 : )?;
692 512 : let id = antares_jsonld::expanded_id(&expanded)?.to_owned();
693 512 : Ok((id, expanded))
694 532 : }
695 :
696 : /// 4.6.6: duplicate instances of one Entity in a batch array "shall come
697 : /// in chronological order" — first oldest. Sequential semantics are kept
698 : /// by splitting into rounds: the Nth occurrence of an id lands in round
699 : /// N, rounds execute in order, and within a round every id is unique.
700 506 : fn rounds(prepped: Vec<Member>) -> Vec<Vec<Member>> {
701 506 : let mut rounds: Vec<Vec<Member>> = Vec::new();
702 506 : let mut occurrence: HashMap<String, usize> = HashMap::new();
703 524 : for (id, doc) in prepped {
704 508 : let n = occurrence.entry(id.clone()).or_insert(0);
705 : // `n` never passes `rounds.len()`: each id's count grows by one
706 : // and a round is added the first time a count reaches it
707 508 : match rounds.get_mut(*n) {
708 208 : Some(round) => round.push((id, doc)),
709 300 : None => rounds.push(vec![(id, doc)]),
710 : }
711 508 : *n = n.saturating_add(1);
712 : }
713 506 : rounds
714 506 : }
715 :
716 : /// The document a store closure writes for `id`; every id the store hands
717 : /// it is one the round names.
718 228 : fn member<'d>(docs: &HashMap<&str, &'d Value>, id: &str) -> Result<&'d Value, NgsiError> {
719 228 : docs.get(id)
720 228 : .copied()
721 228 : .ok_or_else(|| NgsiError::InternalError(format!("batch member {id} lost its document")))
722 228 : }
723 :
724 : /// `src` stamped as the new state of the stored `doc`. 4.8: a replaced
725 : /// Entity keeps the instant it was created.
726 2 : fn replacement(src: &Value, doc: &Value, ts: &str) -> Value {
727 2 : let mut new = src.clone();
728 2 : stamp_new(&mut new, ts);
729 2 : if let (Some(o), Some(created)) = (new.as_object_mut(), doc.get("createdAt")) {
730 2 : o.insert("createdAt".into(), created.clone());
731 2 : }
732 2 : new
733 2 : }
734 :
735 : /// 5.6.9 with noOverwrite, which is instance-level: an incoming instance
736 : /// whose datasetId the stored Attribute already holds is skipped, every
737 : /// other one appended. Whether any instance was skipped.
738 46 : fn append_new_instances(doc: &mut Value, expanded: &Value, ts: &str) -> Result<bool, NgsiError> {
739 46 : let mut skipped = false;
740 46 : let target = antares_store::stored_object(doc)?;
741 206 : for (k, v) in antares_jsonld::expanded_object(expanded)? {
742 108 : if matches!(
743 206 : k.as_str(),
744 206 : "id" | "type" | "scope" | "createdAt" | "modifiedAt"
745 : ) {
746 108 : continue;
747 98 : }
748 : // stamped as every other write path stamps a new instance
749 98 : let mut incoming = v.clone();
750 98 : crate::stamp::stamp_instances(&mut incoming, ts);
751 98 : let incoming: Vec<Value> = match incoming {
752 98 : Value::Array(a) => a,
753 0 : _ => Vec::new(),
754 : };
755 98 : match target.get_mut(k).and_then(Value::as_array_mut) {
756 38 : None => {
757 38 : target.insert(k.clone(), Value::Array(incoming));
758 38 : }
759 60 : Some(cur) => {
760 68 : for ni in incoming {
761 68 : let ds = ni.get("datasetId").and_then(Value::as_str);
762 68 : if cur
763 68 : .iter()
764 76 : .any(|ci| ci.get("datasetId").and_then(Value::as_str) == ds)
765 52 : {
766 52 : skipped = true;
767 52 : } else {
768 16 : cur.push(ni);
769 16 : }
770 : }
771 : }
772 : }
773 : }
774 46 : target.insert("modifiedAt".into(), Value::String(ts.to_owned()));
775 46 : Ok(skipped)
776 46 : }
777 :
778 416 : fn ngsi_of(e: ApiError) -> NgsiError {
779 416 : match e {
780 416 : ApiError::Ngsi(n) => n,
781 0 : ApiError::Bare(code) => NgsiError::BadRequestData(format!("HTTP {code}")),
782 0 : ApiError::NotAcceptable(_) => NgsiError::BadRequestData("HTTP 406".into()),
783 : // One item of a batch cannot answer 503: the array's other entries
784 : // may have succeeded, and 5.6.7.4 gives each failure its own
785 : // ProblemDetails. The item carries the same detail the whole-request
786 : // answer would, so the cause is not lost in the collapse.
787 : ApiError::Overloaded(_) => {
788 0 : NgsiError::InternalError(antares_model::error::DB_OVERLOADED.into())
789 : }
790 : // The policy gate runs once per request, before the array is
791 : // touched, so a refusal is the whole array's 403 and never one
792 : // entry's — this arm is exhaustiveness, not a path. If a per-item
793 : // gate is ever added, 422 is the nearest thing Table 6.3.2-1 has:
794 : // the entry was refused, and the reason travels with it.
795 0 : ApiError::Denied(why) => NgsiError::OperationNotSupported(why),
796 : }
797 416 : }
798 :
799 : /// `a` and `b` hold the same Entity apart from the `modifiedAt` stamps a
800 : /// merge leaves on the Entity and on each Attribute instance it touched.
801 : /// Only those two levels are skipped: a `modifiedAt` key inside a value
802 : /// is data and compares like any other.
803 66 : pub(super) fn same_but_stamps(a: &Value, b: &Value) -> bool {
804 316 : fn without(o: &serde_json::Map<String, Value>) -> impl Iterator<Item = (&String, &Value)> {
805 1436 : o.iter().filter(|(k, _)| k.as_str() != "modifiedAt")
806 316 : }
807 116 : fn same_obj(
808 116 : a: &serde_json::Map<String, Value>,
809 116 : b: &serde_json::Map<String, Value>,
810 116 : inner: fn(&Value, &Value) -> bool,
811 116 : ) -> bool {
812 116 : without(a).count() == without(b).count()
813 236 : && without(a).all(|(k, v)| b.get(k).is_some_and(|w| inner(v, w)))
814 116 : }
815 104 : fn instances(a: &Value, b: &Value) -> bool {
816 104 : match (a.as_array(), b.as_array()) {
817 60 : (Some(x), Some(y)) => {
818 60 : x.len() == y.len()
819 60 : && x.iter()
820 60 : .zip(y)
821 60 : .all(|(i, j)| match (i.as_object(), j.as_object()) {
822 132 : (Some(i), Some(j)) => same_obj(i, j, |v, w| v == w),
823 10 : _ => i == j,
824 60 : })
825 : }
826 44 : _ => a == b,
827 : }
828 104 : }
829 66 : match (a.as_object(), b.as_object()) {
830 66 : (Some(a), Some(b)) => same_obj(a, b, instances),
831 0 : _ => a == b,
832 : }
833 66 : }
|
