valence_core/
admin_entity_delete.rs

1//! Table-keyed queued delete for admin tooling.
2
3use crate::deletion::dag::table_skips_pending_deletion_filter;
4use crate::deletion::dag::DeletionDag;
5use crate::deletion::{dispatch, DeletionRequest, DeletionService};
6use crate::error::{Error, Result};
7use crate::ownership;
8use crate::privacy::{PrivacyEvaluator, PrivacyOperation};
9use crate::query::QueryCore;
10use crate::runtime::Valence;
11use crate::schema::SchemaRegistry;
12
13/// Queue a privacy-checked deletion run for `table`/`id`.
14///
15/// # Errors
16///
17/// Returns an error when the requested operation cannot be completed.
18pub async fn queue_delete_entity(table: &str, id: &str, v: &Valence) -> Result<()> {
19    let _ = queue_delete_entity_returning_run_id(table, id, v).await?;
20    Ok(())
21}
22
23/// Like [`queue_delete_entity`], but returns the new `valence_deletion_run` id when a run
24/// was created (`None` when the row was already missing or already `pending_deletion`).
25///
26/// # Errors
27///
28/// Returns an error when the requested operation cannot be completed.
29#[allow(
30    clippy::cast_possible_truncation,
31    clippy::cast_sign_loss,
32    reason = "deletion metrics require a usize count after clamping negative values"
33)]
34pub async fn queue_delete_entity_returning_run_id(
35    table: &str,
36    id: &str,
37    v: &Valence,
38) -> Result<Option<String>> {
39    if table_skips_pending_deletion_filter(table) {
40        return Err(Error::Validation(format!(
41            "queued delete is not supported for table {table:?}"
42        )));
43    }
44
45    let registry = SchemaRegistry::global();
46    let schema = registry
47        .get_schema(table)
48        .ok_or_else(|| Error::NotFound(format!("unknown table {table}")))?;
49
50    let Some(existing) = QueryCore::get_record_json(table, id, v).await? else {
51        return Ok(None);
52    };
53
54    PrivacyEvaluator::check_entity_access(schema, PrivacyOperation::Delete, &existing, v).await?;
55
56    let bare = ownership::normalize_record_id_for_ownership(id);
57    if let Ok(Some(ownership)) =
58        ownership::OwnershipService::get_ownership_json(table, &bare, v).await
59    {
60        if ownership.get("status").and_then(|s| s.as_str()) == Some("pending_deletion") {
61            return Ok(None);
62        }
63    }
64
65    let dag = DeletionDag::compute(table, &bare, v).await?;
66    if !dag.restrict_violations.is_empty() {
67        #[cfg(feature = "instrumentation")]
68        for v in &dag.restrict_violations {
69            crate::instrumentation::record_restrict_blocked(
70                table,
71                &bare,
72                &v.connection_name,
73                v.blocking_record_count.max(0) as usize,
74            );
75        }
76        return Err(Error::Validation(format!(
77            "delete restricted: {:?}",
78            dag.restrict_violations
79        )));
80    }
81    crate::deletion::check_dag_delete_privacy(&dag, v).await?;
82
83    ownership::OwnershipService::mark_pending_deletion(table, &bare, v).await?;
84
85    let actor_json = serde_json::to_value(v.actor()).unwrap_or(serde_json::Value::Null);
86    let run_id = DeletionService::create_run(table, &bare, actor_json.clone(), v).await?;
87    #[cfg(feature = "instrumentation")]
88    {
89        let max_depth = dag.nodes.iter().map(|n| n.depth).max().unwrap_or(0) as usize;
90        crate::instrumentation::record_run_queued(table, &bare, dag.nodes.len(), max_depth);
91    }
92    dispatch(DeletionRequest {
93        run_id: run_id.clone(),
94        root_table: table.to_string(),
95        root_record_id: bare,
96        actor_json,
97    })
98    .await?;
99
100    Ok(Some(run_id))
101}
valence_core/admin_entity_delete.rs

valence_core/
admin_entity_delete.rs
