diff --git a/crates/adapters/activitypub/src/deletion_notifier.rs b/crates/adapters/activitypub/src/deletion_notifier.rs new file mode 100644 index 00000000..dcf368f9 --- /dev/null +++ b/crates/adapters/activitypub/src/deletion_notifier.rs @@ -0,0 +1,38 @@ +use std::sync::Arc; + +use async_trait::async_trait; +use domain::{ + errors::DomainError, + ports::FederationDeletionNotifier, + value_objects::{InstanceIdentity, UserId}, +}; +use k_ap::ActivityPubService; + +use crate::urls::actor_url; + +/// Broadcasts the actor `Delete` to the user's accepted followers. +/// +/// k-ap only enqueues the per-inbox deliveries; they are signed later from the +/// tombstoned user row and keypair, which the purge keeps until the tombstone +/// cleanup (see `AccountPurger`). +pub struct ApDeletionNotifier { + service: Arc, + instance: InstanceIdentity, +} + +impl ApDeletionNotifier { + pub fn new(service: Arc, instance: InstanceIdentity) -> Self { + Self { service, instance } + } +} + +#[async_trait] +impl FederationDeletionNotifier for ApDeletionNotifier { + async fn notify_deleted(&self, user: &UserId) -> Result<(), DomainError> { + let ap_id = actor_url(&self.instance, user.value()); + self.service + .broadcast_delete_to_followers(user.value(), ap_id) + .await + .map_err(|e| DomainError::InfrastructureError(e.to_string())) + } +} diff --git a/crates/adapters/activitypub/src/event_handler.rs b/crates/adapters/activitypub/src/event_handler.rs index 7ee4e60f..4ac4a962 100644 --- a/crates/adapters/activitypub/src/event_handler.rs +++ b/crates/adapters/activitypub/src/event_handler.rs @@ -143,13 +143,6 @@ impl EventHandler for ActivityPubEventHandler { .on_goal_deleted(user_id, *year) .await .map_err(|e| DomainError::InfrastructureError(e.to_string())), - DomainEvent::UserDeleted { user_id } => { - let ap_id = actor_url(&self.instance, user_id.value()); - self.ap_service - .broadcast_delete_to_followers(user_id.value(), ap_id) - .await - .map_err(|e| DomainError::InfrastructureError(e.to_string())) - } DomainEvent::UserAccountMoved { user_id, new_actor_url, diff --git a/crates/adapters/activitypub/src/federation_ports.rs b/crates/adapters/activitypub/src/federation_ports.rs index 617a498a..f9ab121c 100644 --- a/crates/adapters/activitypub/src/federation_ports.rs +++ b/crates/adapters/activitypub/src/federation_ports.rs @@ -4,18 +4,20 @@ use async_trait::async_trait; use domain::{ errors::DomainError, models::{BlockedDomainInfo, FollowedActorInfo}, - ports::{ApBackfillPort, ApDocumentPort, InstanceBlocklistPort}, + ports::{ApBackfillPort, ApDocumentPort, InstanceBlocklistPort, UserRepository}, + value_objects::UserId, }; use k_ap::ActivityPubService; use uuid::Uuid; pub struct ApServiceAdapter { service: Arc, + users: Arc, } impl ApServiceAdapter { - pub fn new(service: Arc) -> Self { - Self { service } + pub fn new(service: Arc, users: Arc) -> Self { + Self { service, users } } } @@ -26,6 +28,13 @@ fn ap_err(e: anyhow::Error) -> DomainError { #[async_trait] impl ApDocumentPort for ApServiceAdapter { + async fn actor_is_gone(&self, user_id: Uuid) -> Result { + Ok(self + .users + .find_tombstoned_by_id(&UserId::from_uuid(user_id)) + .await? + .is_some()) + } async fn actor_json(&self, user_id: &str) -> Result { self.service.actor_json(user_id).await.map_err(ap_err) } diff --git a/crates/adapters/activitypub/src/lib.rs b/crates/adapters/activitypub/src/lib.rs index 33576e4c..62b834a3 100644 --- a/crates/adapters/activitypub/src/lib.rs +++ b/crates/adapters/activitypub/src/lib.rs @@ -1,4 +1,5 @@ pub mod composite_handler; +pub mod deletion_notifier; pub mod event_handler; pub mod federation_event_bridge; pub mod federation_ports; @@ -7,6 +8,7 @@ pub mod objects; pub mod remote_review_repository; pub mod review_handler; pub mod social_adapter; +pub mod tombstone_gate; pub(crate) mod urls; pub mod user_adapter; pub mod watchlist_handler; @@ -20,6 +22,7 @@ pub use k_ap::{ FollowRepository, Follower, FollowerStatus, FollowingStatus, LocalObject, RemoteActor, }; +pub use deletion_notifier::ApDeletionNotifier; pub use event_handler::ActivityPubEventHandler; pub use federation_ports::ApServiceAdapter; pub use remote_review_repository::{NewRemoteReview, RemoteReviewRepository, RemoteReviewUpdate}; @@ -46,6 +49,7 @@ pub struct ActivityPubWire { pub service: std::sync::Arc, pub router: axum::Router, pub event_handler: std::sync::Arc, + pub deletion_notifier: std::sync::Arc, } pub struct ActivityPubDeps { @@ -139,7 +143,7 @@ pub async fn wire(deps: ActivityPubDeps) -> anyhow::Result { .actor_repo(actor_repo) .blocklist_repo(blocklist_repo) .user_repo(std::sync::Arc::new(DomainUserRepoAdapter::new( - user_repo, + std::sync::Arc::clone(&user_repo), instance.clone(), ))) .signed_fetch_actor_id(INSTANCE_ACTOR_ID) @@ -157,7 +161,16 @@ pub async fn wire(deps: ActivityPubDeps) -> anyhow::Result { .await?, ); - let router = concrete.router(); + let router = concrete + .router() + .layer(axum::middleware::from_fn_with_state( + std::sync::Arc::clone(&user_repo), + tombstone_gate::tombstone_gate, + )); + let deletion_notifier = std::sync::Arc::new(ApDeletionNotifier::new( + std::sync::Arc::clone(&concrete), + instance.clone(), + )) as std::sync::Arc; let event_handler = std::sync::Arc::new(ActivityPubEventHandler::new( std::sync::Arc::clone(&concrete), local_ap_content, @@ -171,6 +184,7 @@ pub async fn wire(deps: ActivityPubDeps) -> anyhow::Result { let ports = std::sync::Arc::new(federation_ports::ApServiceAdapter::new( std::sync::Arc::clone(&concrete), + std::sync::Arc::clone(&user_repo), )); Ok(ActivityPubWire { @@ -182,5 +196,6 @@ pub async fn wire(deps: ActivityPubDeps) -> anyhow::Result { service: concrete, router, event_handler, + deletion_notifier, }) } diff --git a/crates/adapters/activitypub/src/tombstone_gate.rs b/crates/adapters/activitypub/src/tombstone_gate.rs new file mode 100644 index 00000000..83701a93 --- /dev/null +++ b/crates/adapters/activitypub/src/tombstone_gate.rs @@ -0,0 +1,36 @@ +use std::sync::Arc; + +use axum::{ + extract::{Request, State}, + http::StatusCode, + middleware::Next, + response::{IntoResponse, Response}, +}; +use domain::{ports::UserRepository, value_objects::UserId}; + +/// Answers `410 Gone` for the per-user ActivityPub endpoints of a deleted +/// account. k-ap resolves tombstones so it can sign the account's `Delete` +/// deliveries; this keeps that from making the actor look alive. +pub async fn tombstone_gate( + State(users): State>, + request: Request, + next: Next, +) -> Response { + if let Some(id) = gated_user(request.uri().path()) + && matches!( + users.find_tombstoned_by_id(&UserId::from_uuid(id)).await, + Ok(Some(_)) + ) + { + return StatusCode::GONE.into_response(); + } + next.run(request).await +} + +/// The user id in `/users/{id}/outbox|featured|inbox`. +pub fn gated_user(path: &str) -> Option { + let rest = path.strip_prefix("/users/")?; + let (id, endpoint) = rest.split_once('/')?; + matches!(endpoint, "outbox" | "featured" | "inbox").then_some(())?; + uuid::Uuid::parse_str(id).ok() +} diff --git a/crates/adapters/activitypub/src/user_adapter.rs b/crates/adapters/activitypub/src/user_adapter.rs index 86ea3300..ff5ec6f1 100644 --- a/crates/adapters/activitypub/src/user_adapter.rs +++ b/crates/adapters/activitypub/src/user_adapter.rs @@ -64,9 +64,16 @@ impl DomainUserRepoAdapter { impl ApUserRepository for DomainUserRepoAdapter { async fn find_by_id(&self, id: uuid::Uuid) -> anyhow::Result> { let user_id = UserId::from_uuid(id); + // A deleted account still resolves here, so k-ap can sign the queued + // `Delete` deliveries until the tombstone cleanup. Username lookups + // below stay live-only, and the AP endpoints answer 410 for it + // (`tombstone_gate`, `actor_document_response`). let user = match self.repo.find_by_id(&user_id).await? { Some(u) => u, - None => return Ok(None), + None => match self.repo.find_tombstoned_by_id(&user_id).await? { + Some(u) => u, + None => return Ok(None), + }, }; Ok(Some(self.build_user(&user))) } diff --git a/crates/adapters/activitypub/tests/user_adapter.rs b/crates/adapters/activitypub/tests/user_adapter.rs new file mode 100644 index 00000000..03787492 --- /dev/null +++ b/crates/adapters/activitypub/tests/user_adapter.rs @@ -0,0 +1,59 @@ +use domain::{ + ports::UserRepository, + testing::{InMemoryStore, test_user}, + value_objects::InstanceIdentity, +}; +use k_ap::ApUserRepository; + +use activitypub::DomainUserRepoAdapter; + +async fn adapter_with_deleted_alice() -> (DomainUserRepoAdapter, uuid::Uuid) { + let store = InMemoryStore::new(); + let alice = test_user("alice"); + store.save(&alice).await.unwrap(); + store.mark_deleted(alice.id(), &[]).await.unwrap(); + let adapter = DomainUserRepoAdapter::new(store, InstanceIdentity::new("https://md.example")); + (adapter, alice.id().value()) +} + +#[tokio::test] +async fn tombstoned_actor_can_still_sign() { + let (adapter, id) = adapter_with_deleted_alice().await; + + let user = adapter.find_by_id(id).await.unwrap(); + + assert_eq!(user.expect("k-ap resolves the tombstone").username, "alice"); +} + +#[tokio::test] +async fn a_tombstone_is_not_found_by_username() { + let (adapter, _) = adapter_with_deleted_alice().await; + + assert!(adapter.find_by_username("alice").await.unwrap().is_none()); +} + +#[tokio::test] +async fn an_unknown_user_is_none() { + let (adapter, _) = adapter_with_deleted_alice().await; + + assert!( + adapter + .find_by_id(uuid::Uuid::new_v4()) + .await + .unwrap() + .is_none() + ); +} + +#[test] +fn only_the_per_user_endpoints_are_gated() { + use activitypub::tombstone_gate::gated_user; + let id = uuid::Uuid::new_v4(); + for endpoint in ["outbox", "featured", "inbox"] { + assert_eq!(gated_user(&format!("/users/{id}/{endpoint}")), Some(id)); + } + assert_eq!(gated_user(&format!("/users/{id}")), None); + assert_eq!(gated_user(&format!("/users/{id}/followers")), None); + assert_eq!(gated_user("/inbox"), None); + assert_eq!(gated_user("/users/not-a-uuid/outbox"), None); +} diff --git a/crates/adapters/http-axum/src/federation/collections.rs b/crates/adapters/http-axum/src/federation/collections.rs index 2f4cca72..65a2d2f2 100644 --- a/crates/adapters/http-axum/src/federation/collections.rs +++ b/crates/adapters/http-axum/src/federation/collections.rs @@ -17,6 +17,15 @@ pub async fn get_followers_collection( .and_then(|v| v.to_str().ok()) .unwrap_or(""); if accept.contains("application/activity+json") || accept.contains("application/ld+json") { + if state + .app_ctx + .ap_document + .actor_is_gone(user_id) + .await + .unwrap_or(false) + { + return StatusCode::GONE.into_response(); + } let page = params.get("page").and_then(|p| p.parse::().ok()); return match state .app_ctx @@ -52,6 +61,15 @@ pub async fn get_following_collection( .and_then(|v| v.to_str().ok()) .unwrap_or(""); if accept.contains("application/activity+json") || accept.contains("application/ld+json") { + if state + .app_ctx + .ap_document + .actor_is_gone(user_id) + .await + .unwrap_or(false) + { + return StatusCode::GONE.into_response(); + } let page = params.get("page").and_then(|p| p.parse::().ok()); return match state .app_ctx diff --git a/crates/adapters/http-axum/src/federation/negotiation.rs b/crates/adapters/http-axum/src/federation/negotiation.rs index 3c3a9382..831b9775 100644 --- a/crates/adapters/http-axum/src/federation/negotiation.rs +++ b/crates/adapters/http-axum/src/federation/negotiation.rs @@ -20,6 +20,15 @@ pub async fn actor_document_response( if !(accept.contains("application/activity+json") || accept.contains("application/ld+json")) { return None; } + if state + .app_ctx + .ap_document + .actor_is_gone(user_id) + .await + .unwrap_or(false) + { + return Some(StatusCode::GONE.into_response()); + } Some( match state .app_ctx diff --git a/crates/adapters/sqlite/migrations/0102_account_deletion.sql b/crates/adapters/sqlite/migrations/0102_account_deletion.sql index 7918be5c..edc2116b 100644 --- a/crates/adapters/sqlite/migrations/0102_account_deletion.sql +++ b/crates/adapters/sqlite/migrations/0102_account_deletion.sql @@ -1,4 +1,7 @@ ALTER TABLE users ADD COLUMN deleted_at TEXT; +-- Set once the account's data is purged; the stripped row (and the signing +-- keypair) stays until the tombstone cleanup removes it. +ALTER TABLE users ADD COLUMN purged_at TEXT; CREATE TABLE retired_usernames ( username TEXT PRIMARY KEY COLLATE NOCASE, diff --git a/crates/adapters/sqlite/src/lib.rs b/crates/adapters/sqlite/src/lib.rs index 0acc52e4..82bfac2e 100644 --- a/crates/adapters/sqlite/src/lib.rs +++ b/crates/adapters/sqlite/src/lib.rs @@ -8,15 +8,16 @@ mod shared; pub use event_queue::{DbEventQueueConfig, SqliteEventQueue}; use repositories::SqliteUserSettingsRepository; pub use repositories::{ - SqliteApContentQuery, SqliteDiaryRepository, SqliteGoalRepository, SqliteImageRefAdapter, + PURGED_USER_TABLES, RETAINED_USER_TABLES, SqliteAccountPurger, SqliteApContentQuery, + SqliteDiaryRepository, SqliteGoalRepository, SqliteImageRefAdapter, SqliteImportProfileRepository, SqliteImportSessionRepository, SqliteInsightsQuery, SqliteMovieDeduplicator, SqliteMovieProfileRepository, SqliteMovieRepository, SqlitePersonAdapter, SqliteProcessedEventLedger, SqliteProfileFieldsRepository, SqliteRefreshSessionAdapter, SqliteRemoteGoalRepository, SqliteReviewRepository, SqliteSocialRepository, SqliteStatsRepository, SqliteUserRepository, SqliteWatchEventRepository, SqliteWatchlistRepository, SqliteWebhookTokenRepository, - SqliteWrapUpRepository, SqliteWrapUpStatsQuery, create_federated_profile_query, - create_image_ref, create_person_adapter, create_search_adapter, + SqliteWrapUpRepository, SqliteWrapUpStatsQuery, URL_KEYED_USER_TABLES, + create_federated_profile_query, create_image_ref, create_person_adapter, create_search_adapter, }; /// Row helpers shared with the `sqlite-federation` crate (which cannot live diff --git a/crates/adapters/sqlite/src/repositories/account_purge/mod.rs b/crates/adapters/sqlite/src/repositories/account_purge/mod.rs new file mode 100644 index 00000000..ff48f97a --- /dev/null +++ b/crates/adapters/sqlite/src/repositories/account_purge/mod.rs @@ -0,0 +1,181 @@ +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"]; + +/// Tables with no user column whose rows hang off the user's content or actor +/// URL; the purge clears them by URL. +pub const URL_KEYED_USER_TABLES: &[&str] = &["ap_announces", "ap_followers", "ap_following"]; + +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)?; + + // 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)?; + for table in RETAINED_USER_TABLES { + sqlx::query(&format!( + "DELETE FROM {table} WHERE user_id IN ( + SELECT id FROM users WHERE purged_at IS NOT NULL AND purged_at < ?)" + )) + .bind(&cutoff) + .execute(&mut *tx) + .await + .map_err(map_sqlx_error)?; + } + let removed = + sqlx::query("DELETE FROM users WHERE purged_at IS NOT NULL AND purged_at < ?") + .bind(&cutoff) + .execute(&mut *tx) + .await + .map_err(map_sqlx_error)? + .rows_affected(); + tx.commit().await.map_err(map_sqlx_error)?; + Ok(removed) + } +} diff --git a/crates/adapters/sqlite/src/repositories/mod.rs b/crates/adapters/sqlite/src/repositories/mod.rs index ac467d51..5d67dadf 100644 --- a/crates/adapters/sqlite/src/repositories/mod.rs +++ b/crates/adapters/sqlite/src/repositories/mod.rs @@ -1,3 +1,4 @@ +mod account_purge; mod ap_content; mod diary; mod goals; @@ -23,6 +24,9 @@ mod watch_event; mod watchlist; mod wrapup; +pub use account_purge::{ + PURGED_USER_TABLES, RETAINED_USER_TABLES, SqliteAccountPurger, URL_KEYED_USER_TABLES, +}; pub use ap_content::SqliteApContentQuery; pub use diary::SqliteDiaryRepository; pub use goals::SqliteGoalRepository; diff --git a/crates/adapters/sqlite/tests/account_purge_test.rs b/crates/adapters/sqlite/tests/account_purge_test.rs new file mode 100644 index 00000000..df0b72d9 --- /dev/null +++ b/crates/adapters/sqlite/tests/account_purge_test.rs @@ -0,0 +1,605 @@ +//! The purge removes every row a deleted account owns, strips the user row to +//! a data-free tombstone, and the cleanup later drops the tombstone itself. + +use std::collections::BTreeSet; +use std::str::FromStr; + +use chrono::Utc; +use domain::models::{ProfileField, UserProfile, UserSettings, WatchEvent, WatchEventSource}; +use domain::models::{WatchlistEntry, WebhookToken}; +use domain::ports::{ + AccountPurger, FollowCommand, FollowWrite, UserProfileFieldsRepository, WatchEventCommand, + WebhookTokenRepository, +}; +use domain::testing::{test_goal, test_movie, test_review, test_user, test_watched_at}; +use domain::value_objects::{FollowStatus, InstanceIdentity, UserId}; +use sqlite::{ + PURGED_USER_TABLES, RETAINED_USER_TABLES, SqliteAccountPurger, SqliteProfileFieldsRepository, + SqliteSocialRepository, SqliteWatchEventRepository, SqliteWebhookTokenRepository, + URL_KEYED_USER_TABLES, +}; +use sqlx::SqlitePool; +use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions}; + +const BASE: &str = "https://md.example"; +const REMOTE_ACTOR: &str = "https://remote.example/users/ann"; + +async fn migrated_pool() -> SqlitePool { + let opts = SqliteConnectOptions::from_str("sqlite::memory:").unwrap(); + let pool = SqlitePoolOptions::new() + .max_connections(1) + .connect_with(opts) + .await + .unwrap(); + sqlite::migrate(&pool).await.unwrap(); + pool +} + +fn instance() -> InstanceIdentity { + InstanceIdentity::new(BASE) +} + +fn purger(pool: &SqlitePool) -> SqliteAccountPurger { + SqliteAccountPurger::new(pool.clone(), instance()) +} + +async fn count(pool: &SqlitePool, sql: &str) -> i64 { + sqlx::query_scalar(sql).fetch_one(pool).await.unwrap() +} + +async fn count_for(pool: &SqlitePool, table: &str, column: &str, user: &UserId) -> i64 { + sqlx::query_scalar(&format!("SELECT COUNT(*) FROM {table} WHERE {column} = ?")) + .bind(user.value().to_string()) + .fetch_one(pool) + .await + .unwrap() +} + +async fn exec(pool: &SqlitePool, sql: &str, binds: &[&str]) { + let mut q = sqlx::query(sql); + for b in binds { + q = q.bind(*b); + } + q.execute(pool).await.unwrap(); +} + +struct Seeded { + user: UserId, + other: UserId, + movie_id: String, +} + +/// Alice owns a row in every table; bob has a little of everything, including +/// a follow of alice and an announce of his own review, which must survive. +async fn seed(pool: &SqlitePool, parts: &sqlite::SqliteWireOutput) -> Seeded { + let alice = test_user("alice"); + let bob = test_user("bob"); + let a = alice.id().clone(); + let b = bob.id().clone(); + let a_str = a.value().to_string(); + let b_str = b.value().to_string(); + parts.user.save(&alice).await.unwrap(); + parts.user.save(&bob).await.unwrap(); + let movie = test_movie("Heat"); + parts.movie_command.upsert_movie(&movie, &[]).await.unwrap(); + + for u in [&a, &b] { + parts + .watchlist + .add(&WatchlistEntry::new(u.clone(), movie.id().clone()), &[]) + .await + .unwrap(); + parts + .goal_command + .save(&test_goal(u, 2024, 50), &[]) + .await + .unwrap(); + parts + .user_settings + .save(&UserSettings::new(u.clone())) + .await + .unwrap(); + SqliteProfileFieldsRepository::new(pool.clone()) + .set_fields( + u, + vec![ProfileField { + name: "Site".into(), + value: "https://example.com".into(), + }], + &[], + ) + .await + .unwrap(); + SqliteWebhookTokenRepository::new(pool.clone()) + .save(&WebhookToken::new( + u.clone(), + format!("hash-{}", u.value()), + WatchEventSource::Jellyfin, + None, + )) + .await + .unwrap(); + SqliteWatchEventRepository::new(pool.clone()) + .save( + &WatchEvent::new( + u.clone(), + "Heat".into(), + Some(1995), + None, + WatchEventSource::Jellyfin, + test_watched_at(), + None, + ), + &[], + ) + .await + .unwrap(); + parts + .review + .save_review(&test_review(movie.id(), u, 4), &[]) + .await + .unwrap(); + exec( + pool, + "INSERT INTO import_sessions (id, user_id, parsed_data, created_at, expires_at) VALUES (?, ?, '{}', 'now', 'later')", + &[&uuid::Uuid::new_v4().to_string(), &u.value().to_string()], + ) + .await; + exec( + pool, + "INSERT INTO import_profiles (id, user_id, name, field_mappings, created_at) VALUES (?, ?, 'p', '{}', 'now')", + &[&uuid::Uuid::new_v4().to_string(), &u.value().to_string()], + ) + .await; + exec( + pool, + "INSERT INTO refresh_sessions (id, user_id, token, expires_at, created_at) VALUES (?, ?, ?, 'later', 'now')", + &[ + &uuid::Uuid::new_v4().to_string(), + &u.value().to_string(), + &uuid::Uuid::new_v4().to_string(), + ], + ) + .await; + exec( + pool, + "INSERT INTO wrap_up_records (id, user_id, start_date, end_date) VALUES (?, ?, 'a', 'b')", + &[&uuid::Uuid::new_v4().to_string(), &u.value().to_string()], + ) + .await; + exec( + pool, + "INSERT INTO blocked_actors (local_user_id, remote_actor_url, blocked_at) VALUES (?, ?, 'now')", + &[&u.value().to_string(), REMOTE_ACTOR], + ) + .await; + exec( + pool, + "INSERT INTO ap_local_actors (user_id, public_key, private_key, created_at) VALUES (?, 'pub', 'priv', 'now')", + &[&u.value().to_string()], + ) + .await; + } + let social = SqliteSocialRepository::new(pool.clone(), instance()); + let alice_actor = instance().actor_url_for(&a); + social + .apply( + &[ + FollowWrite::AddFollow { + follower_id: a.value(), + target_actor_url: REMOTE_ACTOR.into(), + status: FollowStatus::Accepted, + }, + FollowWrite::AddFollower { + local_user_id: a.value(), + follower_actor_url: REMOTE_ACTOR.into(), + status: FollowStatus::Accepted, + }, + FollowWrite::AddFollow { + follower_id: b.value(), + target_actor_url: REMOTE_ACTOR.into(), + status: FollowStatus::Accepted, + }, + FollowWrite::AddFollower { + local_user_id: b.value(), + follower_actor_url: REMOTE_ACTOR.into(), + status: FollowStatus::Accepted, + }, + // bob follows alice, and alice follows bob + FollowWrite::AddFollow { + follower_id: b.value(), + target_actor_url: alice_actor.clone(), + status: FollowStatus::Accepted, + }, + FollowWrite::AddFollower { + local_user_id: b.value(), + follower_actor_url: alice_actor, + status: FollowStatus::Accepted, + }, + ], + &[], + ) + .await + .unwrap(); + // A remote review attributed to the same account id is not the user's. + exec( + pool, + "INSERT INTO reviews (id, movie_id, user_id, rating, comment, watched_at, created_at, remote_actor_url) + SELECT ?, movie_id, ?, rating, comment, watched_at, created_at, ? FROM reviews WHERE user_id = ? LIMIT 1", + &[ + &uuid::Uuid::new_v4().to_string(), + &a_str, + REMOTE_ACTOR, + &a_str, + ], + ) + .await; + // Announces of alice's review and goal, of bob's review, and one by alice. + let alice_review: String = + sqlx::query_scalar("SELECT id FROM reviews WHERE user_id = ? AND remote_actor_url IS NULL") + .bind(&a_str) + .fetch_one(pool) + .await + .unwrap(); + let bob_review: String = sqlx::query_scalar("SELECT id FROM reviews WHERE user_id = ?") + .bind(&b_str) + .fetch_one(pool) + .await + .unwrap(); + for (id, object, actor) in [ + ( + "an1", + format!("{BASE}/reviews/{alice_review}"), + REMOTE_ACTOR, + ), + ( + "an2", + format!("{BASE}/users/{a_str}/goals/2024"), + REMOTE_ACTOR, + ), + ("an3", format!("{BASE}/reviews/{bob_review}"), REMOTE_ACTOR), + ] { + exec( + pool, + "INSERT INTO ap_announces (id, object_url, actor_url, announced_at) VALUES (?, ?, ?, 'now')", + &[id, &object, actor], + ) + .await; + } + Seeded { + user: a, + other: b, + movie_id: movie.id().value().to_string(), + } +} + +async fn tombstone(pool: &SqlitePool, user: &UserId) { + exec( + pool, + "UPDATE users SET deleted_at = '2026-01-01T00:00:00Z' WHERE id = ?", + &[&user.value().to_string()], + ) + .await; +} + +#[tokio::test] +async fn purge_removes_every_user_owned_row() { + let pool = migrated_pool().await; + let parts = sqlite::wire_pool(pool.clone()).await.unwrap(); + let s = seed(&pool, &parts).await; + for (table, column) in PURGED_USER_TABLES { + assert!( + count_for(&pool, table, column, &s.user).await > 0, + "{table} should be seeded" + ); + } + tombstone(&pool, &s.user).await; + + purger(&pool).purge(&s.user).await.unwrap(); + + for (table, column) in PURGED_USER_TABLES { + // Reviews: only the remote one, asserted below, is left. + let left = if *table == "reviews" { 1 } else { 0 }; + assert_eq!( + count_for(&pool, table, column, &s.user).await, + left, + "{table} should be purged" + ); + } + let alice_actor = instance().actor_url_for(&s.user); + for table in ["ap_followers", "ap_following"] { + let pointing: i64 = sqlx::query_scalar(&format!( + "SELECT COUNT(*) FROM {table} WHERE remote_actor_url = ?" + )) + .bind(&alice_actor) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(pointing, 0, "{table} rows pointing at the user are gone"); + } + let reviews: i64 = count( + &pool, + &format!( + "SELECT COUNT(*) FROM reviews WHERE user_id = '{}'", + s.user.value() + ), + ) + .await; + assert_eq!(reviews, 1, "the remote review is not the user's to purge"); + let announces: Vec = sqlx::query_scalar("SELECT id FROM ap_announces ORDER BY id") + .fetch_all(&pool) + .await + .unwrap(); + assert_eq!(announces, vec!["an3"], "only bob's announce survives"); + + // The other account is untouched. + for (table, column) in PURGED_USER_TABLES { + assert!( + count_for(&pool, table, column, &s.other).await > 0, + "{table} of the other user must survive" + ); + } + // The keypair stays for queued deliveries. + for table in RETAINED_USER_TABLES { + assert_eq!(count_for(&pool, table, "user_id", &s.user).await, 1); + } +} + +#[tokio::test] +async fn the_purge_table_lists_cover_every_user_referencing_column_in_the_schema() { + let pool = migrated_pool().await; + let rows: Vec<(String, String)> = sqlx::query_as( + "SELECT m.name, c.name FROM sqlite_master m, pragma_table_info(m.name) c + WHERE m.type = 'table' AND m.name NOT LIKE 'sqlite_%' AND m.name != 'users' + AND (c.name LIKE '%user_id%' OR c.name LIKE 'owner%')", + ) + .fetch_all(&pool) + .await + .unwrap(); + let fk_tables: Vec = sqlx::query_scalar( + "SELECT m.name FROM sqlite_master m, pragma_foreign_key_list(m.name) f + WHERE m.type = 'table' AND f.\"table\" = 'users'", + ) + .fetch_all(&pool) + .await + .unwrap(); + + let found: BTreeSet<(String, String)> = rows.into_iter().collect(); + let mut covered: BTreeSet<(String, String)> = PURGED_USER_TABLES + .iter() + .map(|(t, c)| (t.to_string(), c.to_string())) + .collect(); + covered.extend( + RETAINED_USER_TABLES + .iter() + .map(|t| (t.to_string(), "user_id".to_string())), + ); + assert_eq!( + found, covered, + "a table references a user but the purge neither empties nor retains it" + ); + for t in fk_tables { + assert!( + covered.iter().any(|(table, _)| *table == t), + "{t} has a foreign key to users but is not in the purge lists" + ); + } + // URL-keyed tables must exist, so a rename cannot silently drop them. + for t in URL_KEYED_USER_TABLES { + let n: i64 = count( + &pool, + &format!("SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='{t}'"), + ) + .await; + assert_eq!(n, 1, "{t} is listed as URL-keyed but missing"); + } +} + +#[tokio::test] +async fn purge_keeps_movies_and_metadata() { + let pool = migrated_pool().await; + let parts = sqlite::wire_pool(pool.clone()).await.unwrap(); + let s = seed(&pool, &parts).await; + let movies = count(&pool, "SELECT COUNT(*) FROM movies").await; + tombstone(&pool, &s.user).await; + + purger(&pool).purge(&s.user).await.unwrap(); + + assert_eq!(count(&pool, "SELECT COUNT(*) FROM movies").await, movies); + assert_eq!( + count( + &pool, + &format!("SELECT COUNT(*) FROM movies WHERE id = '{}'", s.movie_id) + ) + .await, + 1 + ); +} + +#[tokio::test] +async fn purge_strips_the_user_row_but_keeps_the_username_and_state() { + let pool = migrated_pool().await; + let parts = sqlite::wire_pool(pool.clone()).await.unwrap(); + let s = seed(&pool, &parts).await; + parts + .user + .update_profile( + &s.user, + &UserProfile { + display_name: Some("Alice".into()), + bio: Some("bio".into()), + avatar_path: Some("avatars/a.webp".into()), + banner_path: Some("banners/a.webp".into()), + also_known_as: Some("https://x.example/a".into()), + ..Default::default() + }, + &[], + ) + .await + .unwrap(); + parts.user.mark_deleted(&s.user, &[]).await.unwrap(); + let p = purger(&pool); + assert!(!p.is_purged(&s.user).await.unwrap()); + + p.purge(&s.user).await.unwrap(); + + assert!(p.is_purged(&s.user).await.unwrap()); + let tomb = parts + .user + .find_tombstoned_by_id(&s.user) + .await + .unwrap() + .unwrap(); + assert_eq!(tomb.username().value(), "alice"); + assert_eq!(tomb.display_name(), None); + assert_eq!(tomb.bio(), None); + assert_eq!(tomb.also_known_as(), None); + assert_ne!(tomb.email().value(), "alice@example.com"); + assert!( + parts + .user + .is_username_retired(tomb.username()) + .await + .unwrap() + ); + // Images stay referenced until the objects are gone. + let mut keys = p.owned_image_keys(&s.user).await.unwrap(); + keys.sort(); + assert_eq!(keys, vec!["avatars/a.webp", "banners/a.webp"]); + p.forget_images(&s.user).await.unwrap(); + assert!(p.owned_image_keys(&s.user).await.unwrap().is_empty()); +} + +#[tokio::test] +async fn purging_twice_is_a_no_op() { + let pool = migrated_pool().await; + let parts = sqlite::wire_pool(pool.clone()).await.unwrap(); + let s = seed(&pool, &parts).await; + parts.user.mark_deleted(&s.user, &[]).await.unwrap(); + let p = purger(&pool); + p.purge(&s.user).await.unwrap(); + // New data cannot appear for a purged account, but a re-run must not + // touch anything else or fail. + let before: i64 = count(&pool, "SELECT COUNT(*) FROM reviews").await; + let stamp: String = sqlx::query_scalar("SELECT purged_at FROM users WHERE id = ?") + .bind(s.user.value().to_string()) + .fetch_one(&pool) + .await + .unwrap(); + + p.purge(&s.user).await.unwrap(); + + assert_eq!(count(&pool, "SELECT COUNT(*) FROM reviews").await, before); + let again: String = sqlx::query_scalar("SELECT purged_at FROM users WHERE id = ?") + .bind(s.user.value().to_string()) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(stamp, again); + // And an id that never existed is fine too. + p.purge(&UserId::generate()).await.unwrap(); +} + +#[tokio::test] +async fn purge_never_touches_a_live_account() { + let pool = migrated_pool().await; + let parts = sqlite::wire_pool(pool.clone()).await.unwrap(); + let s = seed(&pool, &parts).await; + + purger(&pool).purge(&s.user).await.unwrap(); + + for (table, column) in PURGED_USER_TABLES { + assert!( + count_for(&pool, table, column, &s.user).await > 0, + "{table}" + ); + } + assert!(parts.user.find_by_id(&s.user).await.unwrap().is_some()); +} + +#[tokio::test] +async fn purge_frees_the_email_but_the_username_stays_retired() { + let pool = migrated_pool().await; + let parts = sqlite::wire_pool(pool.clone()).await.unwrap(); + let s = seed(&pool, &parts).await; + let again = test_user("alice"); + assert!( + parts.user.save(&again).await.is_err(), + "email is held while tombstoned" + ); + parts.user.mark_deleted(&s.user, &[]).await.unwrap(); + assert!(parts.user.save(&again).await.is_err()); + + purger(&pool).purge(&s.user).await.unwrap(); + + let reuse_email = domain::models::User::new( + domain::value_objects::Email::new("alice@example.com".into()).unwrap(), + domain::value_objects::Username::new("alice2".into()).unwrap(), + domain::value_objects::PasswordHash::new("h".into()).unwrap(), + domain::models::UserRole::Standard, + ); + parts.user.save(&reuse_email).await.unwrap(); + assert!( + parts + .user + .is_username_retired(&domain::value_objects::Username::new("alice".into()).unwrap()) + .await + .unwrap() + ); + assert!( + parts.user.save(&again).await.is_err(), + "the username is never reusable" + ); +} + +#[tokio::test] +async fn tombstone_cleanup_removes_rows_older_than_retention_and_keeps_younger_ones() { + let pool = migrated_pool().await; + let parts = sqlite::wire_pool(pool.clone()).await.unwrap(); + let s = seed(&pool, &parts).await; + parts.user.mark_deleted(&s.user, &[]).await.unwrap(); + let p = purger(&pool); + p.purge(&s.user).await.unwrap(); + let now = Utc::now().naive_utc(); + let stamp = |days: i64| { + (now - chrono::Duration::days(days)) + .format("%Y-%m-%dT%H:%M:%SZ") + .to_string() + }; + exec( + &pool, + "UPDATE users SET purged_at = ? WHERE id = ?", + &[&stamp(8), &s.user.value().to_string()], + ) + .await; + + // Younger than the cutoff: kept. + let removed = p + .purge_expired_tombstones(now - chrono::Duration::days(9)) + .await + .unwrap(); + assert_eq!(removed, 0); + assert_eq!(count_for(&pool, "users", "id", &s.user).await, 1); + + let removed = p + .purge_expired_tombstones(now - chrono::Duration::days(7)) + .await + .unwrap(); + assert_eq!(removed, 1); + assert_eq!(count_for(&pool, "users", "id", &s.user).await, 0); + assert_eq!( + count_for(&pool, "ap_local_actors", "user_id", &s.user).await, + 0 + ); + assert_eq!(count_for(&pool, "users", "id", &s.other).await, 1); + assert_eq!( + count_for(&pool, "ap_local_actors", "user_id", &s.other).await, + 1 + ); + let retired: i64 = count( + &pool, + "SELECT COUNT(*) FROM retired_usernames WHERE username = 'alice'", + ) + .await; + assert_eq!(retired, 1, "the username stays retired for good"); +} diff --git a/crates/adapters/sqlite/tests/fixtures/migration_ledger.txt b/crates/adapters/sqlite/tests/fixtures/migration_ledger.txt index fedc5c9c..2a339efe 100644 --- a/crates/adapters/sqlite/tests/fixtures/migration_ledger.txt +++ b/crates/adapters/sqlite/tests/fixtures/migration_ledger.txt @@ -30,4 +30,4 @@ 30 DAAA0AF35B4ACB271A9D8068B82F02A72EEC62F3AC07ED504C7A7887976C2C8BACB35FFF477B63A931DF15F79018390D 100 310ADC9ABF0CC2F8E9EDAB33168D009D64228FF5BD8D2A14185513699EBB782606A1EEA8FEE6EC6AF65D72396153F13F 101 19A2542D4343F3921A8BF0A10E195DA1915E3EA97A2BCF2D3BF9079504DB2053E4BD2175D94C95BCA363BC2FDF14AA88 -102 A2118D59757F99A902B613046CE93C53927FE64330684A8703B6E62A8588B87D37ACD55451566F6248A67B7BD9B779EF +102 A3995D574BDDDB634F9D7FEF5FAEE658F272C3E0CABA182080425ED5522E73DBF44EDEF28F35BDCF4669F722E4E2AA4B diff --git a/crates/application/src/jobs/mod.rs b/crates/application/src/jobs/mod.rs index 58115c4a..34020b55 100644 --- a/crates/application/src/jobs/mod.rs +++ b/crates/application/src/jobs/mod.rs @@ -3,6 +3,7 @@ mod import_cleanup; mod movie_dedup; mod processed_events_cleanup; mod refresh_session_cleanup; +mod tombstone_cleanup; mod watch_event_cleanup; mod wrapup; @@ -11,5 +12,6 @@ pub use import_cleanup::ImportSessionCleanupJob; pub use movie_dedup::MovieDeduplicationJob; pub use processed_events_cleanup::{PROCESSED_EVENTS_RETENTION_DAYS, ProcessedEventsCleanupJob}; pub use refresh_session_cleanup::RefreshSessionCleanupJob; +pub use tombstone_cleanup::{TOMBSTONE_RETENTION_DAYS, TombstoneCleanupJob}; pub use watch_event_cleanup::WatchEventCleanupJob; pub use wrapup::{WrapUpAutoGenerateJob, WrapUpCleanupJob}; diff --git a/crates/application/src/jobs/tombstone_cleanup.rs b/crates/application/src/jobs/tombstone_cleanup.rs new file mode 100644 index 00000000..76952d4e --- /dev/null +++ b/crates/application/src/jobs/tombstone_cleanup.rs @@ -0,0 +1,40 @@ +use std::sync::Arc; +use std::time::Duration; + +use async_trait::async_trait; +use domain::{ + errors::DomainError, + ports::{AccountPurger, PeriodicJob}, +}; + +/// How long a purged account's stripped user row and signing keypair stay. +/// Federation deliveries of the account's `Delete` are signed from them, and +/// their retries are long over by then. +pub const TOMBSTONE_RETENTION_DAYS: i64 = 7; + +pub struct TombstoneCleanupJob { + purger: Arc, +} + +impl TombstoneCleanupJob { + pub fn new(purger: Arc) -> Self { + Self { purger } + } +} + +#[async_trait] +impl PeriodicJob for TombstoneCleanupJob { + fn interval(&self) -> Duration { + Duration::from_secs(24 * 3600) + } + + async fn run(&self) -> Result<(), DomainError> { + let cutoff = + chrono::Utc::now().naive_utc() - chrono::Duration::days(TOMBSTONE_RETENTION_DAYS); + let n = self.purger.purge_expired_tombstones(cutoff).await?; + if n > 0 { + tracing::info!("tombstone cleanup: removed {n} purged accounts"); + } + Ok(()) + } +} diff --git a/crates/application/src/users/mod.rs b/crates/application/src/users/mod.rs index d83e3799..804075eb 100644 --- a/crates/application/src/users/mod.rs +++ b/crates/application/src/users/mod.rs @@ -11,6 +11,7 @@ pub mod get_page_viewer; pub mod get_profile_settings; pub mod get_settings; pub mod get_users; +pub mod purge_account; pub mod queries; pub mod request_account_deletion; pub mod resolve_username_to_id; diff --git a/crates/application/src/users/purge_account.rs b/crates/application/src/users/purge_account.rs new file mode 100644 index 00000000..99d0ccae --- /dev/null +++ b/crates/application/src/users/purge_account.rs @@ -0,0 +1,77 @@ +use std::sync::Arc; + +use async_trait::async_trait; +use domain::{ + errors::DomainError, + events::DomainEvent, + ports::{ + AccountPurger, EventHandler, FederationDeletionNotifier, ObjectStorage, UserRepository, + }, +}; + +/// Purges a deleted account once its `UserDeleted` event arrives. +/// +/// Order matters. Federation is told first (deliveries are only enqueued, and +/// signed later from the tombstone the purge leaves behind). A notifier +/// failure never blocks the purge: unreachable servers must not keep a +/// person's data. Then the database rows go in one transaction, and the +/// image objects after the commit. Image references are dropped only once the +/// objects are gone, so a crash in between is recovered by the redelivery. +/// +/// Idempotent: every step is a no-op on a re-run, and the notification is +/// skipped once the account is purged. +pub struct AccountPurgeHandler { + user: Arc, + notifier: Arc, + purger: Arc, + object_storage: Arc, +} + +impl AccountPurgeHandler { + pub fn new( + user: Arc, + notifier: Arc, + purger: Arc, + object_storage: Arc, + ) -> Self { + Self { + user, + notifier, + purger, + object_storage, + } + } +} + +#[async_trait] +impl EventHandler for AccountPurgeHandler { + fn name(&self) -> &'static str { + "account_purge" + } + + async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError> { + let DomainEvent::UserDeleted { user_id } = event else { + return Ok(()); + }; + // Gone already (removed by the tombstone cleanup): nothing left to do. + if self.user.find_tombstoned_by_id(user_id).await?.is_none() { + return Ok(()); + } + + if !self.purger.is_purged(user_id).await? + && let Err(e) = self.notifier.notify_deleted(user_id).await + { + tracing::warn!(user = %user_id.value(), "federation delete notice failed, purging anyway: {e}"); + } + + let keys = self.purger.owned_image_keys(user_id).await?; + self.purger.purge(user_id).await?; + for key in &keys { + self.object_storage.delete(key).await?; + } + if !keys.is_empty() { + self.purger.forget_images(user_id).await?; + } + Ok(()) + } +} diff --git a/crates/application/tests/jobs.rs b/crates/application/tests/jobs.rs index e61924a4..9be4b444 100644 --- a/crates/application/tests/jobs.rs +++ b/crates/application/tests/jobs.rs @@ -1,2 +1,4 @@ #[path = "jobs/processed_events_cleanup_test.rs"] mod processed_events_cleanup_test; +#[path = "jobs/tombstone_cleanup_test.rs"] +mod tombstone_cleanup_test; diff --git a/crates/application/tests/jobs/tombstone_cleanup_test.rs b/crates/application/tests/jobs/tombstone_cleanup_test.rs new file mode 100644 index 00000000..52c7668a --- /dev/null +++ b/crates/application/tests/jobs/tombstone_cleanup_test.rs @@ -0,0 +1,59 @@ +use std::sync::{Arc, Mutex}; + +use application::jobs::{TOMBSTONE_RETENTION_DAYS, TombstoneCleanupJob}; +use async_trait::async_trait; +use domain::{ + errors::DomainError, + ports::{AccountPurger, PeriodicJob}, + value_objects::UserId, +}; + +#[derive(Default)] +struct Purger { + cutoffs: Mutex>, +} + +#[async_trait] +impl AccountPurger for Purger { + async fn is_purged(&self, _: &UserId) -> Result { + unimplemented!() + } + async fn owned_image_keys(&self, _: &UserId) -> Result, DomainError> { + unimplemented!() + } + async fn purge(&self, _: &UserId) -> Result<(), DomainError> { + unimplemented!() + } + async fn forget_images(&self, _: &UserId) -> Result<(), DomainError> { + unimplemented!() + } + async fn purge_expired_tombstones( + &self, + cutoff: chrono::NaiveDateTime, + ) -> Result { + self.cutoffs.lock().unwrap().push(cutoff); + Ok(0) + } +} + +#[tokio::test] +async fn removes_tombstones_purged_before_the_retention_window() { + assert_eq!(TOMBSTONE_RETENTION_DAYS, 7); + let purger = Arc::new(Purger::default()); + + TombstoneCleanupJob::new(purger.clone()) + .run() + .await + .unwrap(); + + let cutoffs = purger.cutoffs.lock().unwrap(); + let expected = chrono::Utc::now().naive_utc() - chrono::Duration::days(7); + assert_eq!(cutoffs.len(), 1); + assert!((cutoffs[0] - expected).num_seconds().abs() < 5); +} + +#[test] +fn runs_daily() { + let job = TombstoneCleanupJob::new(Arc::new(Purger::default())); + assert_eq!(job.interval(), std::time::Duration::from_secs(86_400)); +} diff --git a/crates/application/tests/users.rs b/crates/application/tests/users.rs index 4cd6ff02..dd227b63 100644 --- a/crates/application/tests/users.rs +++ b/crates/application/tests/users.rs @@ -20,6 +20,8 @@ mod get_profile_settings_test; mod get_settings_test; #[path = "users/get_users_test.rs"] mod get_users_test; +#[path = "users/purge_account_test.rs"] +mod purge_account_test; #[path = "users/request_account_deletion_test.rs"] mod request_account_deletion_test; #[path = "users/resolve_username_to_id_test.rs"] diff --git a/crates/application/tests/users/purge_account_test.rs b/crates/application/tests/users/purge_account_test.rs new file mode 100644 index 00000000..5c2103c2 --- /dev/null +++ b/crates/application/tests/users/purge_account_test.rs @@ -0,0 +1,253 @@ +use std::sync::{Arc, Mutex}; + +use application::users::purge_account::AccountPurgeHandler; +use async_trait::async_trait; +use domain::{ + errors::DomainError, + events::{DomainEvent, EventEnvelope, EventId}, + ports::{ + AccountPurger, EventConsumer, EventHandler, FederationDeletionNotifier, ObjectStorage, + UserRepository, + }, + testing::{InMemoryStore, NoopAckHandle, test_user}, + value_objects::UserId, +}; +use futures::stream::{self, BoxStream}; + +type Log = Arc>>; + +struct Notifier { + log: Log, + fail: bool, +} + +#[async_trait] +impl FederationDeletionNotifier for Notifier { + async fn notify_deleted(&self, _: &UserId) -> Result<(), DomainError> { + self.log.lock().unwrap().push("notify".into()); + if self.fail { + Err(DomainError::InfrastructureError("peer unreachable".into())) + } else { + Ok(()) + } + } +} + +struct Purger { + log: Log, + purged: Mutex, + keys: Vec, +} + +#[async_trait] +impl AccountPurger for Purger { + async fn is_purged(&self, _: &UserId) -> Result { + Ok(*self.purged.lock().unwrap()) + } + async fn owned_image_keys(&self, _: &UserId) -> Result, DomainError> { + Ok(self.keys.clone()) + } + async fn purge(&self, _: &UserId) -> Result<(), DomainError> { + self.log.lock().unwrap().push("purge".into()); + *self.purged.lock().unwrap() = true; + Ok(()) + } + async fn forget_images(&self, _: &UserId) -> Result<(), DomainError> { + self.log.lock().unwrap().push("forget".into()); + Ok(()) + } + async fn purge_expired_tombstones(&self, _: chrono::NaiveDateTime) -> Result { + Ok(0) + } +} + +struct Storage { + log: Log, +} + +#[async_trait] +impl ObjectStorage for Storage { + async fn store(&self, _: &str, _: &[u8]) -> Result { + unimplemented!() + } + async fn get(&self, _: &str) -> Result, DomainError> { + unimplemented!() + } + async fn get_stream( + &self, + _: &str, + ) -> Result>, DomainError> { + unimplemented!() + } + async fn delete(&self, key: &str) -> Result<(), DomainError> { + self.log.lock().unwrap().push(format!("delete {key}")); + Ok(()) + } + async fn exists(&self, _: &str) -> Result { + unimplemented!() + } +} + +struct Fixture { + handler: Arc, + log: Log, + store: Arc, + user: UserId, +} + +async fn fixture(notify_fails: bool, tombstoned: bool) -> Fixture { + let store = InMemoryStore::new(); + let alice = test_user("alice"); + store.save(&alice).await.unwrap(); + if tombstoned { + store.mark_deleted(alice.id(), &[]).await.unwrap(); + } + let log: Log = Arc::default(); + let handler = Arc::new(AccountPurgeHandler::new( + store.clone(), + Arc::new(Notifier { + log: log.clone(), + fail: notify_fails, + }), + Arc::new(Purger { + log: log.clone(), + purged: Mutex::new(false), + keys: vec!["avatars/a.webp".into(), "banners/a.webp".into()], + }), + Arc::new(Storage { log: log.clone() }), + )); + Fixture { + handler, + log, + store, + user: alice.id().clone(), + } +} + +fn deleted(user: &UserId) -> DomainEvent { + DomainEvent::UserDeleted { + user_id: user.clone(), + } +} + +fn entries(log: &Log) -> Vec { + log.lock().unwrap().clone() +} + +#[test] +fn the_handler_name_is_stable() { + // The name keys the processed-event ledger; renaming re-runs events. + let f = tokio::runtime::Runtime::new() + .unwrap() + .block_on(fixture(false, true)); + assert_eq!(f.handler.name(), "account_purge"); +} + +#[tokio::test] +async fn notifies_first_then_purges_then_removes_images_then_forgets_them() { + let f = fixture(false, true).await; + + f.handler.handle(&deleted(&f.user)).await.unwrap(); + + assert_eq!( + entries(&f.log), + [ + "notify", + "purge", + "delete avatars/a.webp", + "delete banners/a.webp", + "forget" + ] + ); +} + +#[tokio::test] +async fn purge_continues_when_federation_notify_fails() { + let f = fixture(true, true).await; + + f.handler.handle(&deleted(&f.user)).await.unwrap(); + + let log = entries(&f.log); + assert_eq!(log[0], "notify"); + assert!(log.contains(&"purge".to_string()), "{log:?}"); + assert!(log.contains(&"forget".to_string()), "{log:?}"); +} + +#[tokio::test] +async fn a_rerun_does_not_notify_again() { + let f = fixture(false, true).await; + f.handler.handle(&deleted(&f.user)).await.unwrap(); + + f.handler.handle(&deleted(&f.user)).await.unwrap(); + + let notices = entries(&f.log).iter().filter(|e| *e == "notify").count(); + assert_eq!(notices, 1); +} + +#[tokio::test] +async fn does_nothing_when_the_account_is_not_tombstoned() { + let live = fixture(false, false).await; + live.handler.handle(&deleted(&live.user)).await.unwrap(); + assert!(entries(&live.log).is_empty(), "a live user is never purged"); + + let unknown = fixture(false, true).await; + unknown + .handler + .handle(&deleted(&UserId::generate())) + .await + .unwrap(); + assert!(entries(&unknown.log).is_empty()); +} + +#[tokio::test] +async fn ignores_other_events() { + let f = fixture(false, true).await; + f.handler + .handle(&DomainEvent::UserUpdated { + user_id: f.user.clone(), + }) + .await + .unwrap(); + assert!(entries(&f.log).is_empty()); +} + +struct TwiceConsumer { + id: EventId, + event: DomainEvent, +} + +impl EventConsumer for TwiceConsumer { + fn consume(&self) -> BoxStream<'_, Result> { + let deliveries: Vec> = (0..2) + .map(|_| { + Ok(EventEnvelope::new( + Some(self.id), + self.event.clone(), + Box::new(NoopAckHandle), + )) + }) + .collect(); + Box::pin(stream::iter(deliveries)) + } +} + +#[tokio::test] +async fn a_redelivered_event_is_skipped_by_the_ledger() { + let f = fixture(false, true).await; + let consumer = TwiceConsumer { + id: EventId::new(), + event: deleted(&f.user), + }; + + application::worker::WorkerService::new( + Arc::new(consumer), + vec![f.handler.clone() as Arc], + f.store.clone(), + ) + .unwrap() + .run(tokio::sync::watch::channel(false).1) + .await; + + let purges = entries(&f.log).iter().filter(|e| *e == "purge").count(); + assert_eq!(purges, 1); +} diff --git a/crates/bootstrap/src/database.rs b/crates/bootstrap/src/database.rs index a2cc4c00..eb795d6f 100644 --- a/crates/bootstrap/src/database.rs +++ b/crates/bootstrap/src/database.rs @@ -40,6 +40,7 @@ pub struct DatabaseOutput { pub refresh_session: Arc, pub processed_events: Arc, pub deduplicator: Arc, + pub account_purger: Arc, pub image_ref_command: Arc, pub image_ref_query: Arc, pub follow_command: Arc, @@ -117,6 +118,10 @@ fn assemble( refresh_session: Arc::new(sqlite::SqliteRefreshSessionAdapter::new(w.pool.clone())) as _, processed_events: Arc::new(sqlite::SqliteProcessedEventLedger::new(w.pool.clone())) as _, deduplicator: w.deduplicator, + account_purger: Arc::new(sqlite::SqliteAccountPurger::new( + w.pool.clone(), + instance.clone(), + )) as _, image_ref_command, image_ref_query, follow_command: Arc::clone(&social) as _, diff --git a/crates/bootstrap/src/federation.rs b/crates/bootstrap/src/federation.rs index 7fbf733b..64aba88d 100644 --- a/crates/bootstrap/src/federation.rs +++ b/crates/bootstrap/src/federation.rs @@ -123,15 +123,19 @@ pub async fn wire_server_social( }) } -/// The worker's federation event handlers: the ActivityPub delivery handler and -/// the follow-accepted outbox backfill. +/// The worker's federation event handlers (the ActivityPub delivery handler and +/// the follow-accepted outbox backfill) and the notifier that tells followers +/// an account was deleted. #[cfg(feature = "federation")] pub async fn wire_worker_handlers( db: &DatabaseOutput, instance: &InstanceIdentity, config: &config::AppConfig, event_publisher: &Arc, -) -> anyhow::Result>> { +) -> anyhow::Result<( + Vec>, + Arc, +)> { let fed_repos = sqlite_federation::wire(db.db_pool.clone(), instance.clone()); let ap_wire = activitypub::wire(activitypub::ActivityPubDeps { @@ -160,10 +164,13 @@ pub async fn wire_worker_handlers( .await?; tracing::info!("federation event handler registered"); - Ok(vec![ - ap_wire.event_handler, - Arc::new(crate::follow_backfill_handler::FollowBackfillHandler { - backfill: ap_wire.backfill, - }), - ]) + Ok(( + vec![ + ap_wire.event_handler, + Arc::new(crate::follow_backfill_handler::FollowBackfillHandler { + backfill: ap_wire.backfill, + }), + ], + ap_wire.deletion_notifier, + )) } diff --git a/crates/bootstrap/src/worker.rs b/crates/bootstrap/src/worker.rs index 5826e2e0..5a89728a 100644 --- a/crates/bootstrap/src/worker.rs +++ b/crates/bootstrap/src/worker.rs @@ -133,9 +133,12 @@ async fn assemble( let image_ref_query = Arc::clone(&db.image_ref_query); #[cfg(feature = "federation")] - let federation_handlers = + let (federation_handlers, deletion_notifier) = crate::federation::wire_worker_handlers(&db, &instance, config, &event_publisher_arc) .await?; + #[cfg(not(feature = "federation"))] + let deletion_notifier: Arc = + Arc::new(domain::ports::noop::NoopFederationDeletionNotifier); let movie_command = db.movie_command; let movie_query = db.movie_query; @@ -148,6 +151,7 @@ async fn assemble( let wrapup_repo = db.wrapup_repo; let refresh_session = db.refresh_session; let processed_events = db.processed_events; + let account_purger = db.account_purger; let event_publisher = event_publisher_arc; let metadata = metadata_client; @@ -225,6 +229,9 @@ async fn assemble( Arc::new(application::jobs::ProcessedEventsCleanupJob::new( Arc::clone(&processed_events), )), + Arc::new(application::jobs::TombstoneCleanupJob::new(Arc::clone( + &account_purger, + ))), ]; if let Some(job) = enrichment_job { periodic_jobs.push(job); @@ -282,7 +289,15 @@ async fn assemble( let reindex_handler = Arc::new(SearchReindexHandler::new(worker_deps.reindex_search)) as Arc; + let account_purge = Arc::new(application::users::purge_account::AccountPurgeHandler::new( + Arc::clone(&user), + deletion_notifier, + Arc::clone(&account_purger), + Arc::clone(&object_storage), + )) as Arc; + let mut h = vec![ + account_purge, poster, cleanup, search_cleanup, diff --git a/crates/bootstrap/tests/account_purge_test.rs b/crates/bootstrap/tests/account_purge_test.rs new file mode 100644 index 00000000..ffd156c0 --- /dev/null +++ b/crates/bootstrap/tests/account_purge_test.rs @@ -0,0 +1,130 @@ +//! End to end on the real wiring: a `UserDeleted` event in the outbox makes +//! the worker tell followers, purge the account, and leave the tombstone that +//! the queued deliveries are signed from. + +mod support; + +use std::time::Duration; + +use domain::{events::DomainEvent, ports::UserRepository, value_objects::UserId}; + +async fn wait_for(pool: &sqlx::SqlitePool, sql: &str) -> bool { + for _ in 0..100 { + let n: i64 = sqlx::query_scalar(sql).fetch_one(pool).await.unwrap(); + if n > 0 { + return true; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + false +} + +#[tokio::test] +async fn the_worker_purges_a_deleted_account_and_frees_its_email() { + let (router, pool) = support::server().await; + let alice = support::register(&router, "alice", "alice@example.com").await; + let id = UserId::from_uuid(uuid::Uuid::parse_str(&alice.user_id).unwrap()); + let users = sqlite::SqliteUserRepository::new(pool.clone()); + users + .mark_deleted( + &id, + &[DomainEvent::UserDeleted { + user_id: id.clone(), + }], + ) + .await + .unwrap(); + + let worker = bootstrap::build_worker_on(&support::test_config(), pool.clone()) + .await + .unwrap(); + let (stop, stopped) = tokio::sync::watch::channel(false); + let running = tokio::spawn(worker.run(stopped)); + + let purged = wait_for( + &pool, + "SELECT COUNT(*) FROM users WHERE username = 'alice' AND purged_at IS NOT NULL", + ) + .await; + stop.send(true).unwrap(); + running.await.unwrap().unwrap(); + assert!(purged, "the worker should purge the tombstoned account"); + + let email: String = sqlx::query_scalar("SELECT email FROM users WHERE username = 'alice'") + .fetch_one(&pool) + .await + .unwrap(); + assert_ne!(email, "alice@example.com"); + let again = support::register(&router, "alice2", "alice@example.com").await; + assert!(!again.user_id.is_empty(), "the email can register again"); +} + +/// With a follower, the Delete is enqueued for delivery even though the +/// account is already tombstoned, and the keypair outlives the purge. +#[cfg(feature = "federation")] +#[tokio::test] +async fn followers_are_told_before_the_purge_and_the_keypair_survives_it() { + let (router, pool) = support::server().await; + let alice = support::register(&router, "alice", "alice@example.com").await; + let id = UserId::from_uuid(uuid::Uuid::parse_str(&alice.user_id).unwrap()); + for sql in [ + "INSERT INTO ap_remote_actors (url, handle, inbox_url, fetched_at) + VALUES ('http://127.0.0.1:1/users/ann', 'ann@127.0.0.1', 'http://127.0.0.1:1/inbox', 'now')", + ] { + sqlx::query(sql).execute(&pool).await.unwrap(); + } + sqlx::query( + "INSERT INTO ap_followers (local_user_id, remote_actor_url, status, created_at) + VALUES (?, 'http://127.0.0.1:1/users/ann', 'accepted', 'now')", + ) + .bind(&alice.user_id) + .execute(&pool) + .await + .unwrap(); + let users = sqlite::SqliteUserRepository::new(pool.clone()); + users + .mark_deleted( + &id, + &[DomainEvent::UserDeleted { + user_id: id.clone(), + }], + ) + .await + .unwrap(); + + let worker = bootstrap::build_worker_on(&support::test_config(), pool.clone()) + .await + .unwrap(); + let (stop, stopped) = tokio::sync::watch::channel(false); + let running = tokio::spawn(worker.run(stopped)); + + let purged = wait_for( + &pool, + "SELECT COUNT(*) FROM users WHERE username = 'alice' AND purged_at IS NOT NULL", + ) + .await; + // Deliveries are enqueued from a spawned task, so wait for them too. + let enqueued = wait_for( + &pool, + "SELECT COUNT(*) FROM event_queue WHERE event_type = 'FederationDeliveryRequested'", + ) + .await; + stop.send(true).unwrap(); + running.await.unwrap().unwrap(); + assert!(purged); + assert!(enqueued, "the Delete should be enqueued for the follower"); + + let deliveries: i64 = sqlx::query_scalar( + "SELECT COUNT(*) FROM event_queue WHERE event_type = 'FederationDeliveryRequested'", + ) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(deliveries, 1, "the Delete was enqueued for the follower"); + let keys: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM ap_local_actors WHERE user_id = ?") + .bind(&alice.user_id) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(keys, 1, "the keypair stays until the tombstone cleanup"); +} diff --git a/crates/bootstrap/tests/bootstrap_test.rs b/crates/bootstrap/tests/bootstrap_test.rs index 0c1fdb8d..8bcd9ef5 100644 --- a/crates/bootstrap/tests/bootstrap_test.rs +++ b/crates/bootstrap/tests/bootstrap_test.rs @@ -129,6 +129,7 @@ async fn handler_names_are_unique() { // Names key the ledger: a rename re-runs events inside the retention // window, so changing this list must be a deliberate edit. let mut expected = vec![ + "account_purge", "image_cleanup", "movie_discovery_indexer", "movie_enrichment", @@ -253,3 +254,49 @@ async fn a_deleted_users_token_is_rejected() { assert_eq!(headers.get("location"), anon_headers.get("location")); assert_eq!(headers.get("location").unwrap(), "/login"); } + +/// A deleted account stays resolvable inside k-ap (to sign its `Delete`), but +/// every ActivityPub document for it answers 410, while the classic profile +/// page stays 404. +#[cfg(feature = "federation")] +#[tokio::test] +async fn actor_document_for_a_tombstone_is_gone() { + let (router, pool) = support::server().await; + let alice = support::register(&router, "alice", "alice@example.com").await; + let ap_get = |path: String| { + axum::http::Request::builder() + .uri(path) + .header("accept", "application/activity+json") + .body(axum::body::Body::empty()) + .unwrap() + }; + let paths = |id: &str| { + [ + format!("/users/{id}"), + format!("/users/{id}/followers"), + format!("/users/{id}/following"), + format!("/users/{id}/outbox"), + format!("/users/{id}/featured"), + ] + }; + let (status, _, body) = + support::send(&router, ap_get(format!("/users/{}", alice.user_id))).await; + assert_eq!(status, 200, "live actor: {body}"); + + sqlx::query("UPDATE users SET deleted_at = '2026-01-01T00:00:00Z' WHERE id = ?") + .bind(&alice.user_id) + .execute(&pool) + .await + .unwrap(); + + for path in paths(&alice.user_id) { + let (status, _, body) = support::send(&router, ap_get(path.clone())).await; + assert_eq!(status, 410, "{path}: {body}"); + } + let (status, _, _) = support::send( + &router, + support::get(&format!("/users/{}", alice.user_id), None), + ) + .await; + assert_eq!(status, 404, "the classic profile page stays 404"); +} diff --git a/crates/domain/src/ports/account_purge.rs b/crates/domain/src/ports/account_purge.rs new file mode 100644 index 00000000..ec991468 --- /dev/null +++ b/crates/domain/src/ports/account_purge.rs @@ -0,0 +1,44 @@ +use async_trait::async_trait; + +use crate::{errors::DomainError, value_objects::UserId}; + +/// Tells the fediverse that a local account is gone (an ActivityPub `Delete` +/// to followers). Federation-off builds wire a no-op. +/// +/// Delivery is asynchronous: the implementation may only enqueue deliveries. +/// Those are signed later with the actor's keypair and the tombstoned user row, +/// which is why the purge keeps both until the tombstone cleanup. +#[async_trait] +pub trait FederationDeletionNotifier: Send + Sync { + async fn notify_deleted(&self, user: &UserId) -> Result<(), DomainError>; +} + +/// Removes everything a deleted account owns, in one transaction, and strips +/// the user row down to a data-free tombstone. +/// +/// What stays: the username (retired for good), role, `deleted_at`, the +/// `purged_at` stamp and the signing keypair, so queued federation deliveries +/// can still be signed. `purge_expired_tombstones` removes those later. +#[async_trait] +pub trait AccountPurger: Send + Sync { + /// Whether the account's data has already been purged. + async fn is_purged(&self, user: &UserId) -> Result; + + /// Object-storage keys the account owns (avatar, banner). Still readable + /// after `purge`, until `forget_images`. + async fn owned_image_keys(&self, user: &UserId) -> Result, DomainError>; + + /// One transaction; every delete is "if present", so a re-run is a no-op. + /// Frees the email. Does nothing for an account that is gone or purged. + async fn purge(&self, user: &UserId) -> Result<(), DomainError>; + + /// Drops the image references once their objects are deleted. + async fn forget_images(&self, user: &UserId) -> Result<(), DomainError>; + + /// Deletes tombstones (user row and keypair) purged before `cutoff`. + /// Returns how many were removed. + async fn purge_expired_tombstones( + &self, + cutoff: chrono::NaiveDateTime, + ) -> Result; +} diff --git a/crates/domain/src/ports/federation.rs b/crates/domain/src/ports/federation.rs index c71ff2ee..b81d9f42 100644 --- a/crates/domain/src/ports/federation.rs +++ b/crates/domain/src/ports/federation.rs @@ -8,6 +8,11 @@ use crate::{ #[async_trait] pub trait ApDocumentPort: Send + Sync { + /// Whether the actor belonged to a deleted account: its documents are + /// answered `410 Gone` instead of being served. + async fn actor_is_gone(&self, _user_id: Uuid) -> Result { + Ok(false) + } async fn actor_json(&self, user_id: &str) -> Result; async fn followers_collection_json( &self, diff --git a/crates/domain/src/ports/mod.rs b/crates/domain/src/ports/mod.rs index f25cec7d..1d29026a 100644 --- a/crates/domain/src/ports/mod.rs +++ b/crates/domain/src/ports/mod.rs @@ -1,3 +1,4 @@ +pub mod account_purge; pub mod auth; pub mod diary; pub mod events; @@ -21,6 +22,7 @@ pub mod social; pub mod watchlist; pub mod wrapup; +pub use account_purge::*; pub use auth::*; pub use diary::*; pub use events::*; diff --git a/crates/domain/src/ports/noop.rs b/crates/domain/src/ports/noop.rs index 3c8e0de8..5a1e13fb 100644 --- a/crates/domain/src/ports/noop.rs +++ b/crates/domain/src/ports/noop.rs @@ -34,6 +34,18 @@ impl super::RemoteWatchlistRepository for NoopRemoteWatchlistRepository { } } +// ── NoopFederationDeletionNotifier ─────────────────────────────────────────── + +/// Federation compiled out: nobody remote to tell. +pub struct NoopFederationDeletionNotifier; + +#[async_trait] +impl super::FederationDeletionNotifier for NoopFederationDeletionNotifier { + async fn notify_deleted(&self, _: &UserId) -> Result<(), DomainError> { + Ok(()) + } +} + // ── NoopSocialCommand ──────────────────────────────────────────────────────── pub struct NoopSocialCommand;