purge: skip broadcasts of deleted authors, purge outbox log, guards, 410 inbox test

crates/adapters/activitypub/src/event_handler.rs

@@ -6,7 +6,7 @@ use domain::{
     events::DomainEvent,
     ports::{
         GoalQuery, LocalApContentQuery, MovieQuery, ReviewRepository, StatsRepository,
-        UserFederationSettingsQuery,
+        UserFederationSettingsQuery, UserRepository,
     },
     value_objects::{InstanceIdentity, MovieId, ReviewId, UserId},
 };
@@ -25,6 +25,7 @@ pub struct ActivityPubEventHandler {
     goal_repo: Arc<dyn GoalQuery>,
     stats_repo: Arc<dyn StatsRepository>,
     federation_settings: Arc<dyn UserFederationSettingsQuery>,
+    user_repo: Arc<dyn UserRepository>,
     instance: InstanceIdentity,
 }
 
@@ -38,6 +39,7 @@ impl ActivityPubEventHandler {
         goal_repo: Arc<dyn GoalQuery>,
         stats_repo: Arc<dyn StatsRepository>,
         federation_settings: Arc<dyn UserFederationSettingsQuery>,
+        user_repo: Arc<dyn UserRepository>,
         instance: InstanceIdentity,
     ) -> Self {
         Self {
@@ -48,6 +50,7 @@ impl ActivityPubEventHandler {
             goal_repo,
             stats_repo,
             federation_settings,
+            user_repo,
             instance,
         }
     }
@@ -60,6 +63,15 @@ impl EventHandler for ActivityPubEventHandler {
     }
 
     async fn handle(&self, event: &DomainEvent) -> Result<(), DomainError> {
+        // A deleted account must not speak again: its queued events would
+        // otherwise sign and send from the tombstone after the Delete. Only
+        // the Delete itself (sent by the deletion notifier) goes out.
+        if let Some(author) = self.outbound_author(event)
+            && self.is_tombstoned(&author).await
+        {
+            tracing::debug!(user = %author.value(), "skipping federation event of a deleted account");
+            return Ok(());
+        }
         match event {
             DomainEvent::ReviewLogged {
                 review_id, user_id, ..
@@ -161,6 +173,38 @@ impl EventHandler for ActivityPubEventHandler {
 }
 
 impl ActivityPubEventHandler {
+    /// The local account an outbound activity would be sent as. A queued
+    /// delivery counts unless it is the account's `Delete`.
+    fn outbound_author(&self, event: &DomainEvent) -> Option<UserId> {
+        match event {
+            DomainEvent::ReviewLogged { user_id, .. }
+            | DomainEvent::ReviewUpdated { user_id, .. }
+            | DomainEvent::ReviewDeleted { user_id, .. }
+            | DomainEvent::UserUpdated { user_id }
+            | DomainEvent::WatchlistEntryAdded { user_id, .. }
+            | DomainEvent::WatchlistEntryRemoved { user_id, .. }
+            | DomainEvent::GoalCreated { user_id, .. }
+            | DomainEvent::GoalUpdated { user_id, .. }
+            | DomainEvent::GoalDeleted { user_id, .. }
+            | DomainEvent::UserAccountMoved { user_id, .. } => Some(user_id.clone()),
+            DomainEvent::FederationDeliveryRequested {
+                activity_json,
+                signing_actor_id,
+                ..
+            } if activity_json.get("type").and_then(|t| t.as_str()) != Some("Delete") => {
+                Some(UserId::from_uuid(*signing_actor_id))
+            }
+            _ => None,
+        }
+    }
+
+    async fn is_tombstoned(&self, user: &UserId) -> bool {
+        matches!(
+            self.user_repo.find_tombstoned_by_id(user).await,
+            Ok(Some(_))
+        )
+    }
+
     async fn on_review_logged(&self, user_id: &UserId, review_id: &ReviewId) -> anyhow::Result<()> {
         let flags = self
             .federation_settings

crates/adapters/activitypub/src/lib.rs

@@ -179,6 +179,7 @@ pub async fn wire(deps: ActivityPubDeps) -> anyhow::Result<ActivityPubWire> {
         goal_repo,
         stats_repo,
         federation_settings,
+        std::sync::Arc::clone(&user_repo),
         instance,
     )) as std::sync::Arc<dyn domain::ports::EventHandler>;
 

crates/adapters/activitypub/tests/activity_ids.rs

@@ -16,7 +16,9 @@ use domain::{
     errors::DomainError,
     events::DomainEvent,
     models::{DiaryEntry, WatchlistWithMovie},
-    ports::{EventHandler, LocalApContentQuery, MovieCommand},
+    ports::{
+        EventHandler, FederationDeletionNotifier, LocalApContentQuery, MovieCommand, UserRepository,
+    },
     testing::{
         InMemoryStore, PanicGoalQuery, PanicReviewRepository, PanicStatsRepository, test_movie,
     },
@@ -71,6 +73,8 @@ struct Harness {
     delivered: mpsc::UnboundedReceiver<serde_json::Value>,
     user: UserId,
     movie_id: MovieId,
+    store: Arc<InMemoryStore>,
+    service: Arc<ActivityPubService>,
 }
 
 async fn harness() -> Harness {
@@ -119,14 +123,23 @@ async fn harness() -> Harness {
     let store = InMemoryStore::new();
     let movie = test_movie("Heat");
     store.upsert_movie(&movie, &[]).await.unwrap();
+    let service = Arc::new(service);
+    store
+        .save(&domain::testing::test_user_with_role(
+            user.clone(),
+            domain::models::UserRole::Standard,
+        ))
+        .await
+        .unwrap();
     let handler = ActivityPubEventHandler::new(
-        Arc::new(service),
+        Arc::clone(&service),
         Arc::new(NoContent),
         Arc::new(PanicReviewRepository),
         store.clone(),
         Arc::new(PanicGoalQuery),
         Arc::new(PanicStatsRepository),
-        store,
+        store.clone(),
+        store.clone(),
         InstanceIdentity::new("https://diary.example"),
     );
     Harness {
@@ -134,6 +147,8 @@ async fn harness() -> Harness {
         delivered: rx,
         user,
         movie_id: movie.id().clone(),
+        store,
+        service,
     }
 }
 
@@ -191,3 +206,61 @@ async fn redelivered_delete_gets_a_fresh_activity_id_known_k_ap_limitation() {
 
     assert_ne!(first, second);
 }
+
+/// Queued events of a deleted account must not sign and send from its
+/// tombstone; the account's own Delete still goes out.
+#[tokio::test]
+async fn a_deleted_account_sends_nothing_but_its_delete() {
+    let mut h = harness().await;
+    h.store.mark_deleted(&h.user, &[]).await.unwrap();
+    let logged = DomainEvent::ReviewLogged {
+        review_id: domain::value_objects::ReviewId::generate(),
+        user_id: h.user.clone(),
+        movie_id: h.movie_id.clone(),
+        rating: domain::value_objects::Rating::new(4).unwrap(),
+        watched_at: NaiveDate::from_ymd_opt(2026, 1, 1)
+            .unwrap()
+            .and_hms_opt(0, 0, 0)
+            .unwrap(),
+    };
+    let added = DomainEvent::WatchlistEntryAdded {
+        user_id: h.user.clone(),
+        movie_id: h.movie_id.clone(),
+        movie_title: "Heat".into(),
+        release_year: 1995,
+        external_metadata_id: None,
+        added_at: NaiveDate::from_ymd_opt(2026, 1, 1)
+            .unwrap()
+            .and_hms_opt(0, 0, 0)
+            .unwrap(),
+    };
+    let queued = DomainEvent::FederationDeliveryRequested {
+        inbox_url: "https://remote.example/inbox".into(),
+        activity_json: serde_json::json!({"type": "Create"}),
+        signing_actor_id: h.user.value(),
+    };
+
+    // The review repository panics if the handler gets past the guard.
+    for event in [&logged, &added, &queued] {
+        h.handler.handle(event).await.unwrap();
+    }
+    assert!(
+        tokio::time::timeout(Duration::from_millis(300), h.delivered.recv())
+            .await
+            .is_err(),
+        "nothing is delivered for a deleted account"
+    );
+
+    activitypub::ApDeletionNotifier::new(
+        Arc::clone(&h.service),
+        InstanceIdentity::new("https://diary.example"),
+    )
+    .notify_deleted(&h.user)
+    .await
+    .unwrap();
+    let delete = tokio::time::timeout(Duration::from_secs(5), h.delivered.recv())
+        .await
+        .expect("the Delete is delivered")
+        .unwrap();
+    assert_eq!(delete["type"], "Delete");
+}

crates/adapters/sqlite/src/lib.rs

@@ -8,15 +8,15 @@ mod shared;
 pub use event_queue::{DbEventQueueConfig, SqliteEventQueue};
 use repositories::SqliteUserSettingsRepository;
 pub use repositories::{
-    PURGED_USER_TABLES, RETAINED_USER_TABLES, SqliteAccountPurger, SqliteApContentQuery,
+    PAYLOAD_KEYED_USER_COLUMNS, PURGED_USER_TABLES, RETAINED_USER_TABLES, SqliteAccountPurger,
-    SqliteDiaryRepository, SqliteGoalRepository, SqliteImageRefAdapter,
+    SqliteApContentQuery, SqliteDiaryRepository, SqliteGoalRepository, SqliteImageRefAdapter,
     SqliteImportProfileRepository, SqliteImportSessionRepository, SqliteInsightsQuery,
     SqliteMovieDeduplicator, SqliteMovieProfileRepository, SqliteMovieRepository,
     SqlitePersonAdapter, SqliteProcessedEventLedger, SqliteProfileFieldsRepository,
     SqliteRefreshSessionAdapter, SqliteRemoteGoalRepository, SqliteReviewRepository,
     SqliteSocialRepository, SqliteStatsRepository, SqliteUserRepository,
     SqliteWatchEventRepository, SqliteWatchlistRepository, SqliteWebhookTokenRepository,
-    SqliteWrapUpRepository, SqliteWrapUpStatsQuery, URL_KEYED_USER_TABLES,
+    SqliteWrapUpRepository, SqliteWrapUpStatsQuery, URL_KEYED_USER_COLUMNS,
     create_federated_profile_query, create_image_ref, create_person_adapter, create_search_adapter,
 };
 

crates/adapters/sqlite/src/repositories/account_purge/mod.rs

@@ -34,9 +34,19 @@ pub const PURGED_USER_TABLES: &[(&str, &str)] = &[
 /// 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
+/// Columns that identify the user by actor or content URL rather than by id.
-/// URL; the purge clears them by URL.
+/// The purge matches rows on them; the guard test asserts each still exists so
-pub const URL_KEYED_USER_TABLES: &[&str] = &["ap_announces", "ap_followers", "ap_following"];
+/// 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).
+pub const PAYLOAD_KEYED_USER_COLUMNS: &[(&str, &str)] = &[("event_queue", "payload")];
 
 pub struct SqliteAccountPurger {
     pool: SqlitePool,
@@ -116,6 +126,22 @@ impl AccountPurger for SqliteAccountPurger {
         .await
         .map_err(map_sqlx_error)?;
 
+        // The outbox log: events and signed activities authored as this user.
+        // The `UserDeleted` row (which triggered this run) and the pending
+        // `Delete` deliveries must stay; both go with the tombstone.
+        sqlx::query(
+            "DELETE FROM event_queue
+             WHERE instr(payload, ?1) > 0
+               AND event_type != 'UserDeleted'
+               AND NOT (event_type = 'FederationDeliveryRequested'
+                        AND status IN ('pending', 'processing')
+                        AND instr(payload, '\"type\":\"Delete\"') > 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 = ?"))
@@ -158,6 +184,22 @@ impl AccountPurger for SqliteAccountPurger {
     ) -> Result<u64, DomainError> {
         let cutoff = cutoff.format(TIMESTAMP_FORMAT).to_string();
         let mut tx = self.pool.begin().await.map_err(map_sqlx_error)?;
+        let expiring: Vec<String> = sqlx::query_scalar(
+            "SELECT id FROM users WHERE purged_at IS NOT NULL AND purged_at < ?",
+        )
+        .bind(&cutoff)
+        .fetch_all(&mut *tx)
+        .await
+        .map_err(map_sqlx_error)?;
+        for id in &expiring {
+            // What the purge had to leave: 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 IN (

crates/adapters/sqlite/src/repositories/mod.rs

@@ -25,7 +25,8 @@ mod watchlist;
 mod wrapup;
 
 pub use account_purge::{
-    PURGED_USER_TABLES, RETAINED_USER_TABLES, SqliteAccountPurger, URL_KEYED_USER_TABLES,
+    PAYLOAD_KEYED_USER_COLUMNS, PURGED_USER_TABLES, RETAINED_USER_TABLES, SqliteAccountPurger,
+    URL_KEYED_USER_COLUMNS,
 };
 pub use ap_content::SqliteApContentQuery;
 pub use diary::SqliteDiaryRepository;

crates/adapters/sqlite/tests/account_purge_test.rs

@@ -14,9 +14,9 @@ use domain::ports::{
 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,
+    PAYLOAD_KEYED_USER_COLUMNS, PURGED_USER_TABLES, RETAINED_USER_TABLES, SqliteAccountPurger,
-    SqliteSocialRepository, SqliteWatchEventRepository, SqliteWebhookTokenRepository,
+    SqliteProfileFieldsRepository, SqliteSocialRepository, SqliteWatchEventRepository,
-    URL_KEYED_USER_TABLES,
+    SqliteWebhookTokenRepository, URL_KEYED_USER_COLUMNS,
 };
 use sqlx::SqlitePool;
 use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
@@ -266,6 +266,45 @@ async fn seed(pool: &SqlitePool, parts: &sqlite::SqliteWireOutput) -> Seeded {
         )
         .await;
     }
+    for (event_type, status, payload) in [
+        (
+            "ReviewLogged",
+            "done",
+            format!(r#"{{"user_id":"{a_str}","comment":"secret"}}"#),
+        ),
+        (
+            "FederationDeliveryRequested",
+            "done",
+            format!(
+                r#"{{"signing_actor_id":"{a_str}","activity_json":{{"type":"Create","content":"secret"}}}}"#
+            ),
+        ),
+        (
+            "FederationDeliveryRequested",
+            "pending",
+            format!(
+                r#"{{"signing_actor_id":"{a_str}","activity_json":{{"type":"Delete","id":"x"}}}}"#
+            ),
+        ),
+        (
+            "UserDeleted",
+            "processing",
+            format!(r#"{{"user_id":"{a_str}"}}"#),
+        ),
+        (
+            "ReviewLogged",
+            "done",
+            format!(r#"{{"user_id":"{b_str}"}}"#),
+        ),
+    ] {
+        sqlx::query("INSERT INTO event_queue (event_type, payload, status) VALUES (?, ?, ?)")
+            .bind(event_type)
+            .bind(payload)
+            .bind(status)
+            .execute(pool)
+            .await
+            .unwrap();
+    }
     Seeded {
         user: a,
         other: b,
@@ -332,6 +371,24 @@ async fn purge_removes_every_user_owned_row() {
         .unwrap();
     assert_eq!(announces, vec!["an3"], "only bob's announce survives");
 
+    let queue: Vec<(String, String)> =
+        sqlx::query_as("SELECT event_type, status FROM event_queue ORDER BY id")
+            .fetch_all(&pool)
+            .await
+            .unwrap();
+    assert_eq!(
+        queue,
+        vec![
+            (
+                "FederationDeliveryRequested".to_string(),
+                "pending".to_string()
+            ),
+            ("UserDeleted".to_string(), "processing".to_string()),
+            ("ReviewLogged".to_string(), "done".to_string()),
+        ],
+        "the user's authored events and activities are gone; the pending Delete, the trigger and bob's event stay"
+    );
+
     // The other account is untouched.
     for (table, column) in PURGED_USER_TABLES {
         assert!(
@@ -384,17 +441,36 @@ async fn the_purge_table_lists_cover_every_user_referencing_column_in_the_schema
             "{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.
+    // Columns the purge matches by URL or payload must still exist, so a
-    for t in URL_KEYED_USER_TABLES {
+    // rename fails here instead of silently leaving rows behind.
-        let n: i64 = count(
+    for (table, column) in URL_KEYED_USER_COLUMNS
-            &pool,
+        .iter()
-            &format!("SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name='{t}'"),
+        .chain(PAYLOAD_KEYED_USER_COLUMNS)
-        )
+    {
-        .await;
+        let n: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM pragma_table_info(?) WHERE name = ?")
-        assert_eq!(n, 1, "{t} is listed as URL-keyed but missing");
+            .bind(table)
+            .bind(column)
+            .fetch_one(&pool)
+            .await
+            .unwrap();
+        assert_eq!(n, 1, "{table}.{column} is matched by the purge but missing");
     }
 }
 
+/// `ap_activities` (k-ap's inbound dedup) holds only remote activity ids and a
+/// timestamp: nothing in it is attributable to a local user. If it ever grows a
+/// column, the purge must start clearing it.
+#[tokio::test]
+async fn ap_activities_holds_nothing_attributable_to_a_local_user() {
+    let pool = migrated_pool().await;
+    let columns: Vec<String> =
+        sqlx::query_scalar("SELECT name FROM pragma_table_info('ap_activities') ORDER BY name")
+            .fetch_all(&pool)
+            .await
+            .unwrap();
+    assert_eq!(columns, vec!["id", "processed_at"]);
+}
+
 #[tokio::test]
 async fn purge_keeps_movies_and_metadata() {
     let pool = migrated_pool().await;

crates/application/tests/users/purge_account_test.rs

@@ -90,6 +90,7 @@ impl ObjectStorage for Storage {
 
 struct Fixture {
     handler: Arc<AccountPurgeHandler>,
+    purger: Arc<Purger>,
     log: Log,
     store: Arc<InMemoryStore>,
     user: UserId,
@@ -103,21 +104,23 @@ async fn fixture(notify_fails: bool, tombstoned: bool) -> Fixture {
         store.mark_deleted(alice.id(), &[]).await.unwrap();
     }
     let log: Log = Arc::default();
+    let purger = Arc::new(Purger {
+        log: log.clone(),
+        purged: Mutex::new(false),
+        keys: vec!["avatars/a.webp".into(), "banners/a.webp".into()],
+    });
     let handler = Arc::new(AccountPurgeHandler::new(
         store.clone(),
         Arc::new(Notifier {
             log: log.clone(),
             fail: notify_fails,
         }),
-        Arc::new(Purger {
+        purger.clone(),
-            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,
+        purger,
         log,
         store,
         user: alice.id().clone(),
@@ -251,3 +254,25 @@ async fn a_redelivered_event_is_skipped_by_the_ledger() {
     let purges = entries(&f.log).iter().filter(|e| *e == "purge").count();
     assert_eq!(purges, 1);
 }
+
+/// The rows were committed but the worker died before deleting the image
+/// objects: the redelivery finds the account purged, still sees the image
+/// references, deletes the objects and clears the references.
+#[tokio::test]
+async fn a_rerun_after_a_crash_following_the_commit_deletes_the_images() {
+    let f = fixture(false, true).await;
+    *f.purger.purged.lock().unwrap() = true;
+
+    f.handler.handle(&deleted(&f.user)).await.unwrap();
+
+    assert_eq!(
+        entries(&f.log),
+        [
+            "purge",
+            "delete avatars/a.webp",
+            "delete banners/a.webp",
+            "forget"
+        ],
+        "no second federation notice, images removed and forgotten"
+    );
+}

crates/bootstrap/tests/bootstrap_test.rs

@@ -300,3 +300,32 @@ async fn actor_document_for_a_tombstone_is_gone() {
     .await;
     assert_eq!(status, 404, "the classic profile page stays 404");
 }
+
+/// Remote servers posting to a deleted account's inbox are told it is gone,
+/// before any signature check.
+#[cfg(feature = "federation")]
+#[tokio::test]
+async fn posting_to_a_tombstones_inbox_is_gone() {
+    let (router, pool) = support::server().await;
+    let alice = support::register(&router, "alice", "alice@example.com").await;
+    let post = |path: String| {
+        axum::http::Request::builder()
+            .method("POST")
+            .uri(path)
+            .header("content-type", "application/activity+json")
+            .body(axum::body::Body::from("{}"))
+            .unwrap()
+    };
+    let path = format!("/users/{}/inbox", alice.user_id);
+    let (live, _, _) = support::send(&router, post(path.clone())).await;
+    assert_ne!(live, 410, "a live user's inbox is not gone");
+
+    sqlx::query("UPDATE users SET deleted_at = '2026-01-01T00:00:00Z' WHERE id = ?")
+        .bind(&alice.user_id)
+        .execute(&pool)
+        .await
+        .unwrap();
+
+    let (status, _, body) = support::send(&router, post(path)).await;
+    assert_eq!(status, 410, "{body}");
+}