Line data Source code
1 : // SPDX-License-Identifier: EUPL-1.2
2 : //! Batch Entity Delete (5.6.10): the per-member decision, one multi-row
3 : //! DELETE, the 5.6.10.4 distributed ladder and the S/E outcome.
4 :
5 : use super::*;
6 :
7 : // ---------- POST /entityOperations/delete (5.6.10) ----------
8 :
9 326 : pub async fn batch_delete(
10 326 : State(st): State<AppState>,
11 326 : CleanParams(params): CleanParams,
12 326 : headers: HeaderMap,
13 326 : body: Bytes,
14 326 : ) -> Response {
15 326 : batch_delete_inner(&st, ¶ms, &headers, &body)
16 326 : .await
17 326 : .unwrap_or_else(|e| e.into_response())
18 326 : }
19 :
20 326 : async fn batch_delete_inner(
21 326 : st: &AppState,
22 326 : params: &HashMap<String, String>,
23 326 : headers: &HeaderMap,
24 326 : body: &[u8],
25 326 : ) -> ApiResult<Response> {
26 326 : let tenant = tenant_from(headers)?;
27 326 : check_params(params, &["local"])?;
28 : // A delete batch holds a connection as long as a write batch does.
29 326 : let _admitted = crate::bounds::admit_batch(body.len())
30 326 : .await
31 326 : .ok_or(ApiError::Overloaded(crate::negotiate::RETRY_AFTER_SECONDS))?;
32 326 : let ids = delete_ids(headers, body)?;
33 312 : let mut out = BatchOutcome::new();
34 : // ADR-0020 per member, as for the write batches: each id
35 : // is decided as the single-Entity delete decides it, with the types
36 : // stored under it. The body has to be read to know them, so the gate
37 : // sits behind the parse the way the single-Entity delete sits behind
38 : // its path parameter. A refused id lands in E, the rest are deleted.
39 312 : let mut ids: Vec<&Value> = ids.iter().collect();
40 312 : let mut conds = Conditions::default();
41 312 : decide_deletes(st, &tenant, headers, &mut ids, &mut out, &mut conds).await?;
42 308 : let spec = crate::registry::CsrSpec {
43 : ids: Some(
44 308 : ids.iter()
45 560 : .filter_map(|v| v.as_str())
46 308 : .map(str::to_owned)
47 308 : .collect(),
48 : ),
49 308 : ..Default::default()
50 : };
51 308 : let regs =
52 308 : match crate::federation::write_plan(st, &tenant, &spec, &st.loader.core(), params, headers)
53 308 : .await?
54 : {
55 0 : crate::federation::WritePlan::Answered(r) => return Ok(*r),
56 308 : crate::federation::WritePlan::Forward(regs) => regs,
57 : };
58 308 : conds.unforwarded(!regs.is_empty())?;
59 : // One multi-row DELETE for the whole batch; flags in input order.
60 308 : let id_strs: Vec<String> = ids
61 308 : .iter()
62 308 : .copied()
63 308 : .filter_map(Value::as_str)
64 308 : .map(str::to_owned)
65 308 : .collect();
66 308 : let (flags, failed) = delete_local(st, &tenant, &id_strs, &conds).await?;
67 : // a store error after a committed delete: the ids from the failing one
68 : // on were not attempted, and go to E with that error (5.6.10.4 reports
69 : // each Entity) rather than a 5xx that hides the deletes that landed
70 306 : let attempted = flags.len();
71 306 : let mut flags = flags.into_iter();
72 306 : let mut local_ok: std::collections::HashSet<String> = Default::default();
73 306 : let mut local_miss: Vec<String> = Vec::new();
74 306 : let mut seen = 0usize;
75 558 : for id in ids {
76 558 : let Some(id) = id.as_str() else {
77 6 : out.errors.push(err_entry(
78 6 : None,
79 6 : &NgsiError::BadRequestData("entity id must be a string".into()),
80 : ));
81 6 : continue;
82 : };
83 552 : seen = seen.saturating_add(1);
84 552 : if let Some(e) = failed.as_ref().filter(|_| seen > attempted) {
85 2 : out.errors.push(err_entry(Some(id), e));
86 2 : continue;
87 550 : }
88 550 : if flags.next().unwrap_or(false) {
89 : // 5.6.10 deletes carry the same temporal-deletion semantics
90 : // as 5.6.6 — without this, batch-deleted entities live on in
91 : // the temporal store (the reset's batch delete leaked
92 : // every prior suite's Buildings into the orderBy queries).
93 482 : crate::history::mirror_delete_entity(st, &tenant, id).await;
94 482 : local_ok.insert(id.to_owned());
95 68 : } else {
96 68 : // proxied entities may live remotely only (4.3.6.3) — a
97 68 : // local miss becomes an error only if no forward covers it.
98 68 : local_miss.push(id.to_owned());
99 68 : }
100 : }
101 306 : let id_strs = if failed.is_some() {
102 2 : id_strs.get(..attempted).unwrap_or(&id_strs)
103 : } else {
104 304 : &id_strs
105 : };
106 306 : let remote = forward_deletes(st, &tenant, headers, ®s, id_strs).await;
107 306 : delete_outcome(&mut out, id_strs, &local_ok, local_miss, remote);
108 306 : Ok(out.respond(&tenant, StatusCode::NO_CONTENT, false))
109 326 : }
110 :
111 : /// The Batch Entity Delete body: a non-empty JSON array without null items,
112 : /// inside the batch ceiling.
113 326 : fn delete_ids(headers: &HeaderMap, body: &[u8]) -> ApiResult<Vec<Value>> {
114 326 : let ct = content_type(headers)?;
115 326 : if ct != "application/json" && ct != "application/ld+json" {
116 0 : return Err(ApiError::Bare(StatusCode::UNSUPPORTED_MEDIA_TYPE));
117 326 : }
118 326 : let value: Value = serde_json::from_slice(body)
119 326 : .map_err(|e| NgsiError::InvalidRequest(format!("body is not valid JSON: {e}")))?;
120 322 : let ids = match value {
121 322 : Value::Array(a) if !a.is_empty() => a,
122 : _ => {
123 4 : return Err(NgsiError::BadRequestData(
124 4 : "batch delete body must be a non-empty array".into(),
125 4 : )
126 4 : .into())
127 : }
128 : };
129 : // 5.6.10.4: a null item fails the whole request
130 318 : if ids.iter().any(Value::is_null) {
131 2 : return Err(
132 2 : NgsiError::BadRequestData("batch array must not contain null items".into()).into(),
133 2 : );
134 316 : }
135 : // Batch entity count cap — the same ceiling the write batches carry,
136 : // and the one that bounds the per-id forwarded DELETE fan-out.
137 316 : if ids.len() > *crate::bounds::MAX_BATCH_ITEMS {
138 4 : return Err(NgsiError::BadRequestData(format!(
139 4 : "batch of {} exceeds the {}-entity limit",
140 4 : ids.len(),
141 4 : *crate::bounds::MAX_BATCH_ITEMS
142 4 : ))
143 4 : .into());
144 312 : }
145 312 : Ok(ids)
146 326 : }
147 :
148 : /// The engine's verdict per id: a refused id leaves `ids` for E, an allowed
149 : /// one leaves its condition in `conds`; every id refused is the 403.
150 312 : async fn decide_deletes(
151 312 : st: &AppState,
152 312 : tenant: &antares_model::TenantId,
153 312 : headers: &HeaderMap,
154 312 : ids: &mut Vec<&Value>,
155 312 : out: &mut BatchOutcome,
156 312 : conds: &mut Conditions,
157 312 : ) -> ApiResult<()> {
158 312 : let Some(engine) = &st.policy else {
159 294 : return Ok(());
160 : };
161 : // One read for every id, failing closed as `pre_image`.
162 26 : let wanted: Vec<&str> = ids.iter().filter_map(|v| v.as_str()).collect();
163 18 : let stored = if wanted.is_empty() {
164 4 : HashMap::new()
165 : } else {
166 14 : st.store.get_entities(tenant, &wanted).await?
167 : };
168 16 : let pres: Vec<crate::policy::PreImage> = ids
169 16 : .iter()
170 24 : .map(|v| match v.as_str() {
171 20 : Some(id) => crate::policy::image(stored.get(id).cloned(), None),
172 4 : None => crate::policy::PreImage::default(),
173 24 : })
174 16 : .collect();
175 16 : let one: Vec<Vec<&str>> = ids
176 16 : .iter()
177 24 : .map(|v| v.as_str().into_iter().collect())
178 16 : .collect();
179 16 : let ops: Vec<crate::policy::Operation<'_>> = one
180 16 : .iter()
181 16 : .zip(&pres)
182 16 : .map(|(ids, pre)| crate::policy::Operation {
183 24 : ids,
184 24 : types: &pre.types,
185 24 : stored: pre.stored.as_ref(),
186 24 : ..crate::policy::Operation::new("5.6.10")
187 24 : })
188 16 : .collect();
189 16 : let asking = crate::snapshots::asking_tenant(st, tenant).await;
190 16 : let refused = crate::policy::gate_each(engine.as_ref(), &asking, headers, &ops).await;
191 16 : if let Some(why) = whole_refusal(&refused) {
192 2 : return Err(crate::policy::Denied(why).into());
193 14 : }
194 14 : let mut verdicts = refused.into_iter();
195 22 : ids.retain(|id| match verdicts.next() {
196 2 : Some(Err(why)) => {
197 2 : out.errors.push(denied_entry(id.as_str(), &why));
198 2 : false
199 : }
200 20 : Some(Ok(granted)) => {
201 20 : conds.add(id.as_str(), &granted);
202 20 : true
203 : }
204 0 : None => true,
205 22 : });
206 14 : Ok(())
207 312 : }
208 :
209 : /// The local deletes, a flag per id in input order. Under a policy view
210 : /// or an engine's condition each delete reads its Entity under the lock and
211 : /// leaves one out of the view, or one the condition does not select, which
212 : /// then answers as absent.
213 : ///
214 : /// Those deletes commit one by one. A store error before any of them
215 : /// deleted anything is the request's answer; one after is returned beside
216 : /// the flags of the ids attempted before it, and the rest are not tried.
217 308 : async fn delete_local(
218 308 : st: &AppState,
219 308 : tenant: &antares_model::TenantId,
220 308 : id_strs: &[String],
221 308 : conds: &Conditions,
222 308 : ) -> Result<(Vec<bool>, Option<NgsiError>), NgsiError> {
223 308 : let judged = !conds.0.is_empty();
224 308 : if !judged {
225 302 : return Ok((st.store.batch_delete(tenant, id_strs).await?, None));
226 6 : }
227 6 : let mut flags = Vec::with_capacity(id_strs.len());
228 10 : for id in id_strs {
229 10 : let keep = |d: &Value| conds.holds(id, d);
230 10 : match st.store.delete_entity_if(tenant, id, &keep).await {
231 6 : Ok(deleted) => flags.push(deleted),
232 4 : Err(e) if !flags.contains(&true) => return Err(e),
233 2 : Err(e) => return Ok((flags, Some(e))),
234 : }
235 : }
236 2 : Ok((flags, None))
237 308 : }
238 :
239 : /// 5.6.10.4 support ladder: deleteBatch -> one batch forward; else
240 : /// per-entity Delete Entity forwards; else proxy modes get Conflict
241 : /// per entity. Remote outcomes merge into S/E (never opaque parts).
242 306 : async fn forward_deletes(
243 306 : st: &AppState,
244 306 : tenant: &antares_model::TenantId,
245 306 : headers: &HeaderMap,
246 306 : regs: &[crate::federation::FedReg],
247 306 : id_strs: &[String],
248 306 : ) -> Remote {
249 306 : let mut remote = Remote::default();
250 306 : if regs.is_empty() {
251 226 : return remote;
252 80 : }
253 80 : let ctx_url = crate::federation::ctx_link_url(headers, &st.loader.core().source);
254 92 : for reg in regs {
255 : // "Remove from IN all Entities not matched by CSR" — an
256 : // id-scoped registration (exact ids OR idPattern, 5.12)
257 : // only receives its own ids (4.3.6.1).
258 92 : let sent_ids: Vec<String> = id_strs
259 92 : .iter()
260 176 : .filter(|i| reg.can_match_id(i))
261 92 : .cloned()
262 92 : .collect();
263 92 : if sent_ids.is_empty() {
264 0 : continue;
265 92 : }
266 92 : if reg.supports("deleteBatch") {
267 88 : let sent_vals: Vec<Value> = sent_ids.iter().cloned().map(Value::String).collect();
268 88 : let (status, body, _) = crate::federation::forward(
269 88 : st,
270 88 : reqwest::Method::POST,
271 88 : format!("{}/ngsi-ld/v1/entityOperations/delete", reg.endpoint),
272 88 : &[],
273 88 : headers,
274 88 : tenant,
275 88 : reg,
276 88 : &ctx_url,
277 88 : Some(Value::Array(sent_vals)),
278 : )
279 88 : .await;
280 88 : merge_remote_batch(
281 88 : status,
282 88 : &body,
283 88 : &sent_ids,
284 : false,
285 88 : &mut remote.ok,
286 88 : &mut remote.err,
287 : );
288 4 : } else if reg.supports("deleteEntity") {
289 2 : for id in &sent_ids {
290 2 : let status = forward_delete(st, tenant, headers, reg, &ctx_url, id).await;
291 2 : if (200..300).contains(&status) && status != 207 {
292 2 : remote.ok.push((id.clone(), false));
293 2 : } else {
294 0 : remote.err.push(err_remote(
295 0 : Some(id),
296 0 : status,
297 0 : &format!("forwarded delete returned {status}"),
298 0 : ));
299 0 : }
300 : }
301 2 : } else if reg.is_proxy() {
302 : // 5.6.10.4 last rung: an exclusive or redirect CSR that
303 : // supports neither delete operation contributes an Error
304 : // of type Conflict for each Entity in IN.
305 2 : for id in &sent_ids {
306 2 : remote.err.push(err_entry(
307 2 : Some(id),
308 2 : &NgsiError::Conflict(
309 2 : "registration does not accept the operation deleteBatch".into(),
310 2 : ),
311 2 : ));
312 2 : }
313 0 : }
314 : }
315 80 : remote
316 306 : }
317 :
318 : /// One Delete Entity (5.6.6) forwarded to `reg`; the status it answered.
319 2 : async fn forward_delete(
320 2 : st: &AppState,
321 2 : tenant: &antares_model::TenantId,
322 2 : headers: &HeaderMap,
323 2 : reg: &crate::federation::FedReg,
324 2 : ctx_url: &str,
325 2 : id: &str,
326 2 : ) -> u16 {
327 2 : let (status, _, _) = crate::federation::forward(
328 2 : st,
329 2 : reqwest::Method::DELETE,
330 2 : format!(
331 : "{}/ngsi-ld/v1/entities/{}",
332 : reg.endpoint,
333 2 : crate::federation::path_segment(id)
334 : ),
335 2 : &[],
336 2 : headers,
337 2 : tenant,
338 2 : reg,
339 2 : ctx_url,
340 2 : None,
341 : )
342 2 : .await;
343 2 : status
344 2 : }
345 :
346 : /// S in input order (local first occurrence), then remote-only; E gets the
347 : /// remote errors, then each local miss no forward resolved.
348 306 : fn delete_outcome(
349 306 : out: &mut BatchOutcome,
350 306 : id_strs: &[String],
351 306 : local_ok: &std::collections::HashSet<String>,
352 306 : local_miss: Vec<String>,
353 306 : remote: Remote,
354 306 : ) {
355 306 : let remote_success: std::collections::HashSet<String> =
356 306 : remote.ok.iter().map(|(id, _)| id.clone()).collect();
357 550 : for id in id_strs.iter().filter(|id| local_ok.contains(*id)) {
358 490 : out.succeed(id);
359 490 : }
360 306 : for (id, _) in &remote.ok {
361 66 : out.succeed(id);
362 66 : }
363 306 : let erred: Vec<String> = remote
364 306 : .err
365 306 : .iter()
366 306 : .filter_map(|e| e.get("entityId").and_then(Value::as_str).map(str::to_owned))
367 306 : .collect();
368 306 : out.errors.extend(remote.err);
369 : // 5.6.10.4 local step (5.6.6 limited to local): a missed occurrence
370 : // is ResourceNotFound unless a FORWARD resolved that id — a local
371 : // success does not excuse it (5.5.11.4 duplicate-id semantics: the
372 : // second occurrence of the same id errors).
373 306 : for id in local_miss {
374 68 : if !remote_success.contains(&id) && !erred.contains(&id) {
375 40 : out.errors.push(err_entry(
376 40 : Some(&id),
377 40 : &NgsiError::ResourceNotFound(format!("entity {id} not found")),
378 40 : ));
379 40 : }
380 : }
381 306 : }
|
