feat: account purge handler, tombstone cleanup, AP tombstone signing + 410

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;

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<Vec<chrono::NaiveDateTime>>,
+}
+
+#[async_trait]
+impl AccountPurger for Purger {
+    async fn is_purged(&self, _: &UserId) -> Result<bool, DomainError> {
+        unimplemented!()
+    }
+    async fn owned_image_keys(&self, _: &UserId) -> Result<Vec<String>, 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<u64, DomainError> {
+        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));
+}

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"]

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<Mutex<Vec<String>>>;
+
+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<bool>,
+    keys: Vec<String>,
+}
+
+#[async_trait]
+impl AccountPurger for Purger {
+    async fn is_purged(&self, _: &UserId) -> Result<bool, DomainError> {
+        Ok(*self.purged.lock().unwrap())
+    }
+    async fn owned_image_keys(&self, _: &UserId) -> Result<Vec<String>, 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<u64, DomainError> {
+        Ok(0)
+    }
+}
+
+struct Storage {
+    log: Log,
+}
+
+#[async_trait]
+impl ObjectStorage for Storage {
+    async fn store(&self, _: &str, _: &[u8]) -> Result<String, DomainError> {
+        unimplemented!()
+    }
+    async fn get(&self, _: &str) -> Result<Vec<u8>, DomainError> {
+        unimplemented!()
+    }
+    async fn get_stream(
+        &self,
+        _: &str,
+    ) -> Result<BoxStream<'static, Result<bytes::Bytes, DomainError>>, 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<bool, DomainError> {
+        unimplemented!()
+    }
+}
+
+struct Fixture {
+    handler: Arc<AccountPurgeHandler>,
+    log: Log,
+    store: Arc<InMemoryStore>,
+    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<String> {
+    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<EventEnvelope, DomainError>> {
+        let deliveries: Vec<Result<EventEnvelope, DomainError>> = (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<dyn EventHandler>],
+        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);
+}