Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! The distributed arm of the batch operations (5.6.7.4, 5.6.8.4, 5.6.9.4,
3 : //! 5.6.10.4, 5.6.20.4): what a forward sends, and how a peer's answer merges
4 : //! into the client's S and E arrays.
5 :
6 : use super::*;
7 :
8 : /// 5.2.17 BatchEntityError for a failed forwarded part — the remote status
9 : /// and detail travel inside the ProblemDetails (5.6.7.4/5.6.10.4: remote
10 : /// error results merge into E).
11 : /// The status a Context Source put in a batch error entry, when it is an
12 : /// HTTP error status; anything else (absent, not a number, outside 400..=599,
13 : /// past u16) is the status its response came back with. A plain `as u16`
14 : /// wrapped 65 736 to 200 and reported the failure as a success.
15 26 : pub(super) fn remote_status(entry: &Value, fallback: u16) -> u16 {
16 26 : entry
17 26 : .get("error")
18 26 : .and_then(|p| p.get("status"))
19 26 : .and_then(Value::as_u64)
20 26 : .and_then(|s| u16::try_from(s).ok())
21 26 : .filter(|s| (400..=599).contains(s))
22 26 : .unwrap_or(fallback)
23 26 : }
24 :
25 112 : pub(super) fn err_remote(id: Option<&str>, status: u16, detail: &str) -> Value {
26 112 : let etype = match status {
27 0 : 400 => "BadRequestData",
28 4 : 404 => "ResourceNotFound",
29 4 : 409 => "AlreadyExists",
30 0 : 422 => "OperationNotSupported",
31 104 : _ => "InternalError",
32 : };
33 112 : json!({
34 112 : "entityId": id.unwrap_or("unknown"),
35 112 : "error": {
36 112 : "type": format!("https://uri.etsi.org/ngsi-ld/errors/{etype}"),
37 112 : "title": "distributed operation failed",
38 112 : "status": status,
39 112 : "detail": detail,
40 : }
41 : })
42 112 : }
43 :
44 : /// Merge one forwarded BATCH operation's outcome into the remote S/E sets
45 : /// (5.6.7.4: "Merge the returned list of Entities successfully created with
46 : /// S. Merge the returned list of Entities in Error with E.").
47 204 : pub(super) fn merge_remote_batch(
48 204 : status: u16,
49 204 : body: &Value,
50 204 : sent_ids: &[String],
51 204 : created: bool,
52 204 : ok: &mut Vec<(String, bool)>,
53 204 : err: &mut Vec<Value>,
54 204 : ) {
55 : // Only the Entities this broker forwarded may appear in the client's
56 : // S/E arrays: an id the Context Source names but we never sent is not
57 : // part of this request's outcome, and the source's own error text is
58 : // rebuilt from its status rather than relayed verbatim.
59 204 : let mine = |id: &str| sent_ids.iter().any(|s| s == id);
60 204 : match (status, body) {
61 204 : (200..=206, Value::Array(a)) => {
62 18 : for id in a.iter().filter_map(Value::as_str).filter(|i| mine(i)) {
63 8 : ok.push((id.to_owned(), created));
64 8 : }
65 : }
66 : (200..=206, _) => {
67 258 : for id in sent_ids {
68 258 : ok.push((id.clone(), created));
69 258 : }
70 : }
71 6 : (207, Value::Object(o)) => {
72 6 : if let Some(Value::Array(a)) = o.get("success") {
73 10 : for id in a.iter().filter_map(Value::as_str).filter(|i| mine(i)) {
74 6 : ok.push((id.to_owned(), created));
75 6 : }
76 0 : }
77 6 : if let Some(Value::Array(a)) = o.get("errors") {
78 10 : for e in a {
79 10 : let Some(id) = e
80 10 : .get("entityId")
81 10 : .and_then(Value::as_str)
82 10 : .filter(|i| mine(i))
83 : else {
84 4 : continue;
85 : };
86 6 : let remote = remote_status(e, status);
87 6 : err.push(err_remote(
88 6 : Some(id),
89 6 : remote,
90 6 : "forwarded batch operation reported an error for this entity",
91 : ));
92 : }
93 0 : }
94 : }
95 : _ => {
96 104 : for id in sent_ids {
97 104 : err.push(err_remote(
98 104 : Some(id),
99 104 : status,
100 104 : &format!("forwarded batch operation returned {status}"),
101 104 : ));
102 104 : }
103 : }
104 : }
105 204 : }
106 :
107 : /// The registration-matching input of a batch (4.3.6.1): the ids, the
108 : /// expanded types and the expanded Attribute names the whole array carries,
109 : /// as one `CsrSpec`, together with the items in the shape a forward sends
110 : /// them. An entry whose body is not an object, or whose `@context` did not
111 : /// resolve, contributes nothing and is left to the local arm to report.
112 510 : pub(super) fn batch_spec(items: &[BatchItem]) -> (crate::registry::CsrSpec, Vec<FwdItem>) {
113 510 : let mut fwd_items = Vec::new();
114 510 : let mut spec = crate::registry::CsrSpec::default();
115 510 : let (mut types, mut ids, mut attrs) = (Vec::new(), Vec::new(), Vec::new());
116 6256 : for (item, ctx) in items {
117 6256 : let (Some(o), Ok(c)) = (item.as_object(), ctx.as_ref()) else {
118 420 : continue;
119 : };
120 5836 : if let Some(id) = o.get("id").and_then(Value::as_str) {
121 5836 : ids.push(id.to_owned());
122 5836 : }
123 5836 : match o.get("type") {
124 5816 : Some(Value::String(t)) => types.push(c.expand_key(t)),
125 16 : Some(Value::Array(a)) => {
126 32 : types.extend(a.iter().filter_map(Value::as_str).map(|t| c.expand_key(t)))
127 : }
128 4 : _ => {}
129 : }
130 20654 : for k in o.keys() {
131 20654 : if !matches!(k.as_str(), "id" | "type" | "scope" | "@context") {
132 7944 : attrs.push(c.expand_key(k));
133 13390 : }
134 : }
135 5836 : fwd_items.push((o.clone(), c.clone()));
136 : }
137 510 : if !types.is_empty() {
138 496 : spec.types = Some(types);
139 496 : }
140 510 : if !ids.is_empty() {
141 496 : spec.ids = Some(ids);
142 496 : }
143 510 : if !attrs.is_empty() {
144 426 : spec.attrs = Some(attrs);
145 426 : }
146 510 : (spec, fwd_items)
147 510 : }
148 :
149 : /// What the forwards of one batch brought back: the ids a peer applied, each
150 : /// with whether it created the Entity, and the BatchEntityErrors for E.
151 : #[derive(Default)]
152 : pub(super) struct Remote {
153 : pub(super) ok: Vec<(String, bool)>,
154 : pub(super) err: Vec<Value>,
155 : }
156 :
157 : impl Remote {
158 : /// The outcome of one forwarded single-Entity request.
159 18 : fn one(&mut self, id: &str, status: u16, created: bool) {
160 18 : if (200..300).contains(&status) && status != 207 {
161 16 : self.ok.push((id.to_owned(), created && status == 201));
162 2 : } else {
163 2 : self.err.push(err_remote(
164 2 : Some(id),
165 2 : status,
166 2 : &format!("forwarded operation returned {status}"),
167 2 : ));
168 2 : }
169 18 : }
170 : }
171 :
172 : /// The distributed arm of a batch write (5.6.7.4, 5.6.8.4, 5.6.9.4,
173 : /// 5.6.20.4): one forwarded request per matching registration, whose
174 : /// outcomes merge into the same S and E arrays the local arm fills, as
175 : /// Entity ids and BatchEntityErrors (5.2.16, 5.2.17) rather than opaque
176 : /// part descriptors.
177 106 : pub(super) async fn forward_batch(
178 106 : st: &AppState,
179 106 : params: &HashMap<String, String>,
180 106 : headers: &HeaderMap,
181 106 : tenant: &antares_model::TenantId,
182 106 : opts: WriteOpts,
183 106 : fed_regs: &[crate::federation::FedReg],
184 106 : fwd_items: &[FwdItem],
185 106 : ) -> Remote {
186 106 : let src = fwd_items
187 106 : .first()
188 106 : .map(|(_, c)| c.source.clone())
189 106 : .unwrap_or(Value::Null);
190 106 : let fwd = Fwd {
191 106 : st,
192 106 : headers,
193 106 : tenant,
194 106 : ctx_url: crate::federation::ctx_link_url(headers, &src),
195 106 : };
196 106 : let mut query: Vec<(String, String)> = Vec::new();
197 106 : if let Some(o) = params.get("options") {
198 40 : query.push(("options".into(), o.clone()));
199 66 : }
200 106 : let mut remote = Remote::default();
201 130 : for reg in fed_regs {
202 130 : let arr: Vec<Value> = fwd_items
203 130 : .iter()
204 240 : .filter_map(|(o, c)| crate::federation::reduce_to_scope(o, reg, c))
205 130 : .collect();
206 130 : if arr.is_empty() {
207 0 : continue;
208 130 : }
209 130 : let sent_ids: Vec<String> = arr
210 130 : .iter()
211 232 : .filter_map(|e| e.get("id").and_then(Value::as_str).map(str::to_owned))
212 130 : .collect();
213 130 : let op = opts.mode.batch_op();
214 130 : if reg.supports(op) {
215 108 : fwd.batch(reg, opts.mode, &query, arr, &sent_ids, &mut remote)
216 108 : .await;
217 22 : } else if !fwd.singly(reg, opts, arr, &mut remote).await && reg.is_proxy() {
218 : // 5.6.7.4/5.6.8.4/5.6.9.4/5.6.20.4 last rung: "In case CSR is
219 : // an exclusive or redirect Context Source Registration, add an
220 : // Error of type Conflict for each Entity in IN to E."
221 8 : for id in &sent_ids {
222 8 : remote.err.push(err_entry(
223 8 : Some(id),
224 8 : &NgsiError::Conflict(format!(
225 8 : "registration does not accept the operation {op}"
226 8 : )),
227 8 : ));
228 8 : }
229 16 : }
230 : }
231 106 : remote
232 106 : }
233 :
234 : /// The request context every forward of one batch shares.
235 : struct Fwd<'a> {
236 : st: &'a AppState,
237 : headers: &'a HeaderMap,
238 : tenant: &'a antares_model::TenantId,
239 : ctx_url: String,
240 : }
241 :
242 : impl Fwd<'_> {
243 : /// The whole reduced array as the peer's own batch operation.
244 108 : async fn batch(
245 108 : &self,
246 108 : reg: &crate::federation::FedReg,
247 108 : mode: BatchMode,
248 108 : query: &[(String, String)],
249 108 : arr: Vec<Value>,
250 108 : sent_ids: &[String],
251 108 : remote: &mut Remote,
252 108 : ) {
253 108 : let (status, body, _) = crate::federation::forward(
254 108 : self.st,
255 108 : reqwest::Method::POST,
256 108 : format!(
257 : "{}/ngsi-ld/v1/entityOperations/{}",
258 : reg.endpoint,
259 108 : mode.batch_path()
260 : ),
261 108 : query,
262 108 : self.headers,
263 108 : self.tenant,
264 108 : reg,
265 108 : &self.ctx_url,
266 108 : Some(Value::Array(arr)),
267 : )
268 108 : .await;
269 : // 5.6.8.5: an upsert forward only CREATED entities when the
270 : // remote said so — 201 (body lists the created ids); a 204
271 : // means every forwarded entity was updated.
272 108 : merge_remote_batch(
273 108 : status,
274 108 : &body,
275 108 : sent_ids,
276 108 : mode == BatchMode::Create || (mode == BatchMode::Upsert && status == 201),
277 108 : &mut remote.ok,
278 108 : &mut remote.err,
279 : );
280 108 : }
281 :
282 : /// 5.6.7.4/5.6.8.4/5.6.9.4/5.6.20.4 single-op fallbacks, one request per
283 : /// Entity; false when the peer supports none of them. These never
284 : /// inherit the batch `options` parameter — e.g. options=update is no
285 : /// Create Entity parameter (5.6.1.3) and would 400 the forward; the
286 : /// append fallback re-adds noOverwrite explicitly (5.6.9.4).
287 22 : async fn singly(
288 22 : &self,
289 22 : reg: &crate::federation::FedReg,
290 22 : opts: WriteOpts,
291 22 : arr: Vec<Value>,
292 22 : remote: &mut Remote,
293 22 : ) -> bool {
294 22 : if opts.mode == BatchMode::Upsert && reg.supports("createEntity") {
295 2 : for ent in arr {
296 2 : self.upsert_one(reg, opts, ent, remote).await;
297 : }
298 2 : return true;
299 20 : }
300 20 : let Some(call) = BatchFwd::single(reg, opts) else {
301 6 : return false;
302 : };
303 16 : for ent in arr {
304 16 : let (id, status) = self.one(reg, ent, &call).await;
305 16 : remote.one(&id, status, call.created);
306 : }
307 14 : true
308 22 : }
309 :
310 : /// 5.6.8.4: an upsert forwarded as Create Entity; an Entity the peer
311 : /// already holds is not a failed upsert — it falls through to the
312 : /// operation that updates the one that is there.
313 2 : async fn upsert_one(
314 2 : &self,
315 2 : reg: &crate::federation::FedReg,
316 2 : opts: WriteOpts,
317 2 : ent: Value,
318 2 : remote: &mut Remote,
319 2 : ) {
320 2 : let (id, status) = self.one(reg, ent.clone(), &BatchFwd::create()).await;
321 2 : if status != 409 {
322 0 : remote.one(&id, status, true);
323 0 : return;
324 2 : }
325 2 : let call = if !opts.update && reg.supports("replaceEntity") {
326 2 : BatchFwd::replace()
327 0 : } else if opts.update && reg.supports("updateEntity") {
328 0 : BatchFwd::update()
329 : } else {
330 : // 5.6.8.4: neither replace nor update available
331 0 : remote.err.push(err_remote(
332 0 : Some(&id),
333 : 422,
334 0 : &format!("OperationNotSupported: no upsert path for {id}"),
335 : ));
336 0 : return;
337 : };
338 2 : let (id, status) = self.one(reg, ent, &call).await;
339 2 : remote.one(&id, status, call.created);
340 2 : }
341 :
342 20 : async fn one(
343 20 : &self,
344 20 : reg: &crate::federation::FedReg,
345 20 : ent: Value,
346 20 : call: &BatchFwd,
347 20 : ) -> (String, u16) {
348 20 : forward_one(
349 20 : self.st,
350 20 : reg,
351 20 : self.headers,
352 20 : self.tenant,
353 20 : &self.ctx_url,
354 20 : ent,
355 20 : call,
356 20 : )
357 20 : .await
358 20 : }
359 : }
360 :
361 : /// Where a forwarded batch item lands on the peer.
362 : #[derive(Clone, Copy)]
363 : enum FwdPath {
364 : Collection,
365 : Entity,
366 : Attrs,
367 : }
368 :
369 : /// The single-Entity request a batch operation forwards when the peer does
370 : /// not support the batch operation itself — the 5.6.7.4 / 5.6.8.4 / 5.6.9.4 /
371 : /// 5.6.20.4 fallback ladders. 5.6.7 posts to the collection, 5.6.8 replaces
372 : /// the Entity or updates its Attributes, 5.6.9 updates them or appends with
373 : /// overwrite disabled, 5.6.20 merges the Entity; the loop around them is the
374 : /// same. None of them inherits the batch `options` parameter, which is no
375 : /// parameter of the single-Entity operation being carried — 5.6.9.4's
376 : /// noOverwrite is re-added here, explicitly, because that one is.
377 : struct BatchFwd {
378 : path: FwdPath,
379 : method: reqwest::Method,
380 : query: Vec<(String, String)>,
381 : /// whether a 201 from the peer counts as a creation in the 5.2.16 S array
382 : created: bool,
383 : }
384 :
385 : impl BatchFwd {
386 : /// The single-Entity operation a peer without the batch operation takes
387 : /// for each Entity of a create, replace-upsert, update or merge, in the
388 : /// order of the support ladders; None when it supports none of them.
389 : /// The upsert that starts with Create Entity is [`Fwd::upsert_one`].
390 20 : fn single(reg: &crate::federation::FedReg, opts: WriteOpts) -> Option<Self> {
391 12 : match opts.mode {
392 12 : BatchMode::Create if reg.supports("createEntity") => Some(Self::create()),
393 2 : BatchMode::Upsert if !opts.update && reg.supports("replaceEntity") => {
394 0 : Some(Self::replace())
395 : }
396 2 : BatchMode::Upsert if opts.update && reg.supports("updateEntity") => {
397 2 : Some(Self::update())
398 : }
399 4 : BatchMode::Update if !opts.no_overwrite && reg.supports("updateEntity") => {
400 2 : Some(Self::update())
401 : }
402 : // 5.6.20.4 support ladder: no mergeBatch -> per-entity
403 : // Merge Entity (5.6.17) forwards.
404 2 : BatchMode::Merge if reg.supports("mergeEntity") => Some(Self::merge()),
405 : // 5.6.9.4: append with Attribute overwrite disabled.
406 2 : BatchMode::Update if opts.no_overwrite && reg.supports("appendAttrs") => {
407 2 : Some(Self::append_no_overwrite())
408 : }
409 6 : _ => None,
410 : }
411 20 : }
412 :
413 : /// 5.6.1 Create Entity.
414 8 : fn create() -> Self {
415 8 : Self {
416 8 : path: FwdPath::Collection,
417 8 : method: reqwest::Method::POST,
418 8 : query: Vec::new(),
419 8 : created: true,
420 8 : }
421 8 : }
422 : /// 5.6.4 Replace Entity.
423 2 : fn replace() -> Self {
424 2 : Self {
425 2 : path: FwdPath::Entity,
426 2 : method: reqwest::Method::PUT,
427 2 : query: Vec::new(),
428 2 : created: false,
429 2 : }
430 2 : }
431 : /// 5.6.2 Update Entity Attributes.
432 4 : fn update() -> Self {
433 4 : Self {
434 4 : path: FwdPath::Attrs,
435 4 : method: reqwest::Method::PATCH,
436 4 : query: Vec::new(),
437 4 : created: false,
438 4 : }
439 4 : }
440 : /// 5.6.17 Merge Entity.
441 2 : fn merge() -> Self {
442 2 : Self {
443 2 : path: FwdPath::Entity,
444 2 : method: reqwest::Method::PATCH,
445 2 : query: Vec::new(),
446 2 : created: false,
447 2 : }
448 2 : }
449 : /// 5.6.3 Append Entity Attributes with overwrite disabled (5.6.9.4).
450 2 : fn append_no_overwrite() -> Self {
451 2 : Self {
452 2 : path: FwdPath::Attrs,
453 2 : method: reqwest::Method::POST,
454 2 : query: vec![("options".into(), "noOverwrite".into())],
455 2 : created: false,
456 2 : }
457 2 : }
458 : }
459 :
460 : /// Forward one Entity of the batch to one registration (4.3.6.3) and report
461 : /// the id it was sent under with the status the peer answered. The id is
462 : /// reported back to the client verbatim, so it is kept raw here and
463 : /// percent-encoded per RFC 3986 clause 3.3 only where it becomes a path
464 : /// segment of the forwarded URL.
465 20 : async fn forward_one(
466 20 : st: &AppState,
467 20 : reg: &crate::federation::FedReg,
468 20 : headers: &HeaderMap,
469 20 : tenant: &antares_model::TenantId,
470 20 : ctx_url: &str,
471 20 : ent: Value,
472 20 : call: &BatchFwd,
473 20 : ) -> (String, u16) {
474 20 : let id = ent
475 20 : .get("id")
476 20 : .and_then(Value::as_str)
477 20 : .unwrap_or_default()
478 20 : .to_owned();
479 20 : let seg = crate::federation::path_segment(&id);
480 20 : let base = ®.endpoint;
481 20 : let url = match call.path {
482 10 : FwdPath::Collection => format!("{base}/ngsi-ld/v1/entities"),
483 4 : FwdPath::Entity => format!("{base}/ngsi-ld/v1/entities/{seg}"),
484 6 : FwdPath::Attrs => format!("{base}/ngsi-ld/v1/entities/{seg}/attrs"),
485 : };
486 20 : let (status, _, _) = crate::federation::forward(
487 20 : st,
488 20 : call.method.clone(),
489 20 : url,
490 20 : &call.query,
491 20 : headers,
492 20 : tenant,
493 20 : reg,
494 20 : ctx_url,
495 20 : Some(ent),
496 : )
497 20 : .await;
498 20 : (id, status)
499 20 : }
|
