systemprompt_users/repository/user/
purge.rs

1//! Deleting a user's rows from every table that keys on them.
2//!
3//! The schema cannot cascade a user delete: activity tables carry `user_id`
4//! without a foreign key, because the column also holds sentinel principals
5//! and the identifiers a session column means differ table to table. Each
6//! owning crate therefore declares its tables in the
7//! [`systemprompt_extension::purge`] registry and the delete runs them here,
8//! inside the same transaction that deletes the `users` row. Content
9//! the user's rows referenced but did not own — an artifact body shared by
10//! digest — is declared as an orphan sweep: the keys the user's rows reference
11//! are captured first, and once the user-keyed tables are gone only those keys
12//! that nothing references any more are cleared, in the same transaction. The
13//! same lists, counted instead of deleted, are the dry-run report: what a
14//! deletion would remove, table by table — and, run against a user who no
15//! longer exists, an orphan report.
16//!
17//! Copyright (c) systemprompt.io — Business Source License 1.1.
18//! See <https://systemprompt.io> for licensing details.
19
20use sqlx::{AssertSqlSafe, Postgres, Transaction};
21use systemprompt_database::admin::SafeIdentifier;
22use systemprompt_extension::purge::{
23    OrphanSweep, UserPurgeTable, registered_orphan_sweeps, registered_user_purge_tables,
24};
25use systemprompt_extension::{orphan_sweeps, user_purge_tables};
26use systemprompt_identifiers::UserId;
27
28use crate::error::{Result, UserError};
29use crate::repository::UserRepository;
30
31// Why: core's own user-keyed tables without a foreign key to `users`, one
32// entry per owning crate's table. `ai_requests` cascades to its children;
33// `user_sessions` is deleted by the caller first because `ai_requests` and
34// `user_contexts` point at it with SET NULL and must go before the session.
35user_purge_tables!(
36    "systemprompt-core",
37    [
38        ("ai_requests", "user_id"),
39        ("mcp_artifacts", "user_id"),
40        ("mcp_tool_executions", "user_id"),
41        ("mcp_sessions", "user_id"),
42        ("mcp_external_sessions", "user_id"),
43        ("mcp_proxy_identities", "user_id"),
44        ("governance_decisions", "user_id"),
45        ("agent_tasks", "user_id"),
46        ("task_messages", "user_id"),
47        ("files", "user_id"),
48        ("event_outbox", "user_id"),
49        ("logs", "user_id"),
50        ("user_rate_limit_buckets", "user_id"),
51        ("oauth_jti_revocations", "user_id"),
52        ("sso_login_events", "user_id"),
53    ]
54);
55
56orphan_sweeps!(
57    "systemprompt-core",
58    [{
59        table: "artifact_payloads",
60        key: "sha256",
61        referenced_by: "mcp_artifacts",
62        via: "payload_sha256",
63        user_column: "user_id",
64    }]
65);
66
67/// How many rows one purge table holds for a user.
68#[derive(Debug, Clone, Copy)]
69pub struct PurgeCount {
70    pub owner: &'static str,
71    pub table: &'static str,
72    pub rows: i64,
73}
74
75fn identifier(kind: &'static str, raw: &str) -> Result<SafeIdentifier> {
76    SafeIdentifier::parse(raw).map_err(|source| UserError::PurgeIdentifier {
77        kind,
78        name: raw.to_owned(),
79        source,
80    })
81}
82
83// Why: the predicate is a fixed fragment from a crate's registration, never
84// user input, but it is interpolated into SQL — so it is held to a character
85// set that cannot terminate the statement or open a comment.
86fn predicate(entry: &UserPurgeTable) -> Result<String> {
87    let Some(raw) = entry.predicate else {
88        return Ok(String::new());
89    };
90    let allowed = |c: char| c.is_ascii_alphanumeric() || " _'<>=!.()".contains(c);
91    if raw.is_empty() || raw.contains("''") || !raw.chars().all(allowed) {
92        return Err(UserError::Validation(format!(
93            "purge predicate for {}: not a plain comparison: {raw}",
94            entry.table
95        )));
96    }
97    Ok(format!(" AND ({raw})"))
98}
99
100fn purge_where(entry: &UserPurgeTable) -> Result<String> {
101    let table = identifier("table", entry.table)?;
102    let column = identifier("column", entry.column)?;
103    // Why: `::text` comparison — purge tables key on TEXT and VARCHAR
104    // columns, and the cast keeps one text bind valid for every one of them.
105    Ok(format!(
106        "FROM {} WHERE {}::text = $1{}",
107        table.quoted(),
108        column.quoted(),
109        predicate(entry)?
110    ))
111}
112
113struct SweepSql {
114    candidates: String,
115    delete: String,
116    preview: String,
117}
118
119fn sweep_sql(sweep: &OrphanSweep) -> Result<SweepSql> {
120    let table = identifier("sweep table", sweep.table)?.quoted();
121    let key = identifier("sweep key", sweep.key)?.quoted();
122    let referenced_by = identifier("sweep referrer", sweep.referenced_by)?.quoted();
123    let via = identifier("sweep reference", sweep.via)?.quoted();
124    let user = identifier("sweep user column", sweep.user_column)?.quoted();
125    let referrer = format!("SELECT 1 FROM {referenced_by} r WHERE r.{via} = t.{key}");
126    Ok(SweepSql {
127        candidates: format!(
128            "SELECT DISTINCT r.{via}::text FROM {referenced_by} r \
129             WHERE r.{user}::text = $1 AND r.{via} IS NOT NULL"
130        ),
131        delete: format!(
132            "DELETE FROM {table} t WHERE t.{key}::text = ANY($1) AND NOT EXISTS ({referrer})"
133        ),
134        preview: format!(
135            "SELECT COUNT(*) FROM {table} t \
136             WHERE EXISTS ({referrer} AND r.{user}::text = $1) \
137             AND NOT EXISTS ({referrer} AND r.{user}::text IS DISTINCT FROM $1)"
138        ),
139    })
140}
141
142impl UserRepository {
143    pub(super) async fn purge_user_rows(
144        tx: &mut Transaction<'_, Postgres>,
145        id: &UserId,
146    ) -> Result<Vec<PurgeCount>> {
147        let mut sweeps = Vec::new();
148        for sweep in registered_orphan_sweeps() {
149            let sql = sweep_sql(sweep)?;
150            let keys: Vec<String> = sqlx::query_scalar(AssertSqlSafe(sql.candidates))
151                .bind(id.as_str())
152                .fetch_all(&mut **tx)
153                .await?;
154            sweeps.push((sweep, sql.delete, keys));
155        }
156        let mut removed = Vec::new();
157        for entry in registered_user_purge_tables() {
158            let sql = format!("DELETE {}", purge_where(entry)?);
159            let result = sqlx::query(AssertSqlSafe(sql))
160                .bind(id.as_str())
161                .execute(&mut **tx)
162                .await?;
163            removed.push(PurgeCount {
164                owner: entry.owner,
165                table: entry.table,
166                rows: i64::try_from(result.rows_affected()).unwrap_or(i64::MAX),
167            });
168        }
169        for (sweep, delete, keys) in sweeps {
170            let result = sqlx::query(AssertSqlSafe(delete))
171                .bind(keys)
172                .execute(&mut **tx)
173                .await?;
174            removed.push(PurgeCount {
175                owner: sweep.owner,
176                table: sweep.table,
177                rows: i64::try_from(result.rows_affected()).unwrap_or(i64::MAX),
178            });
179        }
180        Ok(removed)
181    }
182
183    pub async fn purge_preview(&self, id: &UserId) -> Result<Vec<PurgeCount>> {
184        let mut counts = Vec::new();
185        let sessions: i64 = sqlx::query_scalar!(
186            r#"SELECT COUNT(*) AS "count!" FROM user_sessions WHERE user_id = $1"#,
187            id.as_str()
188        )
189        .fetch_one(&*self.pool)
190        .await?;
191        counts.push(PurgeCount {
192            owner: "systemprompt-core",
193            table: "user_sessions",
194            rows: sessions,
195        });
196        for entry in registered_user_purge_tables() {
197            let sql = format!("SELECT COUNT(*) {}", purge_where(entry)?);
198            let rows: i64 = sqlx::query_scalar(AssertSqlSafe(sql))
199                .bind(id.as_str())
200                .fetch_one(&*self.pool)
201                .await?;
202            counts.push(PurgeCount {
203                owner: entry.owner,
204                table: entry.table,
205                rows,
206            });
207        }
208        for sweep in registered_orphan_sweeps() {
209            let rows: i64 = sqlx::query_scalar(AssertSqlSafe(sweep_sql(sweep)?.preview))
210                .bind(id.as_str())
211                .fetch_one(&*self.pool)
212                .await?;
213            counts.push(PurgeCount {
214                owner: sweep.owner,
215                table: sweep.table,
216                rows,
217            });
218        }
219        Ok(counts)
220    }
221}
systemprompt_users/repository/user/purge.rs

systemprompt_users/repository/user/
purge.rs
