use async_trait::async_trait; use domain::{ errors::DomainError, ports::AccountPurger, value_objects::InstanceIdentity, value_objects::UserId, }; use sqlx::SqlitePool; use crate::shared::map_sqlx_error; const TIMESTAMP_FORMAT: &str = "%Y-%m-%dT%H:%M:%SZ"; /// Every table the purge empties for the account, with the column that holds /// the user id. `reviews` is further restricted to the user's own (non-remote) /// rows. A guard test compares this list with the schema, so a new /// user-owned table cannot be forgotten. pub const PURGED_USER_TABLES: &[(&str, &str)] = &[ ("reviews", "user_id"), ("watchlist_entries", "user_id"), ("goals", "user_id"), ("user_settings", "user_id"), ("user_profile_fields", "user_id"), ("webhook_tokens", "user_id"), ("watch_events", "user_id"), ("wrap_up_records", "user_id"), ("import_sessions", "user_id"), ("import_profiles", "user_id"), ("refresh_sessions", "user_id"), ("ap_followers", "local_user_id"), ("ap_following", "local_user_id"), ("blocked_actors", "local_user_id"), ]; /// Tables that reference a user but are kept until the tombstone cleanup: /// the keypair is needed to sign queued federation deliveries. pub const RETAINED_USER_TABLES: &[&str] = &["ap_local_actors"]; /// Columns that identify the user by actor or content URL rather than by id. /// The purge matches rows on them; the guard test asserts each still exists so /// a rename cannot silently stop the purge from clearing them. pub const URL_KEYED_USER_COLUMNS: &[(&str, &str)] = &[ ("ap_announces", "object_url"), ("ap_announces", "actor_url"), ("ap_followers", "remote_actor_url"), ("ap_following", "remote_actor_url"), ]; /// Columns holding serialized events and activities that carry the user's id /// (the outbox log keeps finished rows, including authored activity JSON) and /// the global wrap-up reports, whose `most_active_user` names a member. pub const PAYLOAD_KEYED_USER_COLUMNS: &[(&str, &str)] = &[ ("event_queue", "payload"), ("wrap_up_records", "report_json"), ]; pub struct SqliteAccountPurger { pool: SqlitePool, instance: InstanceIdentity, } impl SqliteAccountPurger { pub fn new(pool: SqlitePool, instance: InstanceIdentity) -> Self { Self { pool, instance } } } #[async_trait] impl AccountPurger for SqliteAccountPurger { async fn is_purged(&self, user: &UserId) -> Result { sqlx::query_scalar::<_, i64>( "SELECT EXISTS(SELECT 1 FROM users WHERE id = ? AND purged_at IS NOT NULL)", ) .bind(user.value().to_string()) .fetch_one(&self.pool) .await .map(|n| n != 0) .map_err(map_sqlx_error) } async fn owned_image_keys(&self, user: &UserId) -> Result, DomainError> { let row: Option<(Option, Option)> = sqlx::query_as( "SELECT avatar_path, banner_path FROM users WHERE id = ? AND deleted_at IS NOT NULL", ) .bind(user.value().to_string()) .fetch_optional(&self.pool) .await .map_err(map_sqlx_error)?; Ok(row .map(|(avatar, banner)| avatar.into_iter().chain(banner).collect()) .unwrap_or_default()) } async fn purge(&self, user: &UserId) -> Result<(), DomainError> { let id = user.value().to_string(); let actor_url = self.instance.actor_url_for(user); let base = self.instance.base_url().to_string(); let mut tx = self.pool.begin().await.map_err(map_sqlx_error)?; // Strip first: it claims the purge. Only a tombstoned, not yet purged // account matches, so a live account (or a re-run) changes nothing. let claimed = sqlx::query( "UPDATE users SET email = 'deleted-' || id || '@deleted.invalid', password_hash = '!purged', display_name = NULL, bio = NULL, also_known_as = NULL, purged_at = ? WHERE id = ? AND deleted_at IS NOT NULL AND purged_at IS NULL", ) .bind(chrono::Utc::now().format(TIMESTAMP_FORMAT).to_string()) .bind(&id) .execute(&mut *tx) .await .map_err(map_sqlx_error)? .rows_affected(); if claimed == 0 { return Ok(()); } // Announces of the user's content, matched before the reviews go. sqlx::query( "DELETE FROM ap_announces WHERE object_url IN ( SELECT ?1 || '/reviews/' || id FROM reviews WHERE user_id = ?2 AND remote_actor_url IS NULL) OR substr(object_url, 1, length(?3)) = ?3 OR actor_url = ?4", ) .bind(&base) .bind(&id) .bind(format!("{base}/users/{id}/")) .bind(&actor_url) .execute(&mut *tx) .await .map_err(map_sqlx_error)?; // The outbox log: finished events and activities authored as, or // mentioning, this user. Unfinished rows are never touched (they // belong to other users' work too and finish or dead-letter on their // own); the tombstone cleanup removes their history later. The // `UserDeleted` row that triggered this run stays until then as well. sqlx::query( "DELETE FROM event_queue WHERE instr(payload, ?1) > 0 AND event_type != 'UserDeleted' AND status IN ('done', 'dead_lettered')", ) .bind(&id) .execute(&mut *tx) .await .map_err(map_sqlx_error)?; // Global wrap-up reports name their most active member; redact it // (the type allows none) and keep the rest of the report. sqlx::query( "UPDATE wrap_up_records SET report_json = json_set(report_json, '$.most_active_user', NULL) WHERE user_id IS NULL AND report_json IS NOT NULL AND instr(report_json, ?) > 0", ) .bind(&id) .execute(&mut *tx) .await .map_err(map_sqlx_error)?; // Other accounts' follows of this one. for table in ["ap_followers", "ap_following"] { sqlx::query(&format!("DELETE FROM {table} WHERE remote_actor_url = ?")) .bind(&actor_url) .execute(&mut *tx) .await .map_err(map_sqlx_error)?; } for (table, column) in PURGED_USER_TABLES { let extra = if *table == "reviews" { " AND remote_actor_url IS NULL" } else { "" }; sqlx::query(&format!("DELETE FROM {table} WHERE {column} = ?{extra}")) .bind(&id) .execute(&mut *tx) .await .map_err(map_sqlx_error)?; } tx.commit().await.map_err(map_sqlx_error) } async fn forget_images(&self, user: &UserId) -> Result<(), DomainError> { sqlx::query( "UPDATE users SET avatar_path = NULL, banner_path = NULL WHERE id = ? AND purged_at IS NOT NULL", ) .bind(user.value().to_string()) .execute(&self.pool) .await .map(|_| ()) .map_err(map_sqlx_error) } async fn purge_expired_tombstones( &self, cutoff: chrono::NaiveDateTime, ) -> Result { let cutoff = cutoff.format(TIMESTAMP_FORMAT).to_string(); let mut tx = self.pool.begin().await.map_err(map_sqlx_error)?; // A tombstone whose id is still named by an unfinished event waits for // the next run: that event may yet be handled, and its history row // must be erased with the rest. let expiring: Vec = sqlx::query_scalar( "SELECT id FROM users u WHERE purged_at IS NOT NULL AND purged_at < ? AND NOT EXISTS ( SELECT 1 FROM event_queue q WHERE q.status NOT IN ('done', 'dead_lettered') AND instr(q.payload, u.id) > 0)", ) .bind(&cutoff) .fetch_all(&mut *tx) .await .map_err(map_sqlx_error)?; for id in &expiring { // Includes the `UserDeleted` row and the `Delete` deliveries. sqlx::query("DELETE FROM event_queue WHERE instr(payload, ?) > 0") .bind(id) .execute(&mut *tx) .await .map_err(map_sqlx_error)?; for table in RETAINED_USER_TABLES { sqlx::query(&format!("DELETE FROM {table} WHERE user_id = ?")) .bind(id) .execute(&mut *tx) .await .map_err(map_sqlx_error)?; } sqlx::query("DELETE FROM users WHERE id = ?") .bind(id) .execute(&mut *tx) .await .map_err(map_sqlx_error)?; } tx.commit().await.map_err(map_sqlx_error)?; Ok(expiring.len() as u64) } }