fix

backend/tests/auth_flow.rs

@@ -13,13 +13,8 @@ use papercrate::models::{
 };
 use papercrate::openapi::schemas::PasskeySummary;
 use papercrate::schema::{
-    capability_sets,
-    magic_tokens::dsl as magic_dsl,
-    tenants,
-    tenants::dsl as tenant_dsl,
-    user_memberships,
-    user_sessions,
-    users,
+    capability_sets, magic_tokens::dsl as magic_dsl, tenants, tenants::dsl as tenant_dsl,
+    user_memberships, user_sessions, users,
 };
 use papercrate::test_support::{acquire_db_lock, body_to_vec, TestApp, TestUserRole};
 use rand::{rngs::OsRng, TryRngCore};

backend/tests/documents_flow.rs

@@ -519,6 +519,29 @@ async fn duplicate_and_restore_document() -> Result<()> {
         .await?;
     assert_eq!(delete.status(), StatusCode::NO_CONTENT);
 
+    let trashed_conflict = app
+        .upload_document(
+            "/api/documents",
+            "dup.bin",
+            "application/octet-stream",
+            &payload,
+            None,
+            &token,
+        )
+        .await?;
+    assert_eq!(trashed_conflict.status(), StatusCode::CONFLICT);
+    let trashed_body = body_to_vec(trashed_conflict.into_body()).await?;
+    let trashed_error: ApiErrorResponse = serde_json::from_slice(&trashed_body)?;
+    assert_eq!(trashed_error.code.as_deref(), Some("duplicate_document"));
+    assert!(trashed_error.error.contains("trash"));
+    let trashed_details = trashed_error.details.as_ref().expect("details present");
+    assert_eq!(
+        trashed_details
+            .get("conflict_document_in_trash")
+            .and_then(|value| value.as_bool()),
+        Some(true)
+    );
+
     let third = app
         .upload_document_with_extras(
             "/api/documents",
@@ -1837,7 +1860,7 @@ async fn purge_document_removes_data() -> Result<()> {
     let state = Arc::new(app.state.clone());
     let storage = app
         .state
-        .storage_for_tenant(job.tenant_id)
+        .storage_for_tenant(job.tenant_id.expect("job should have tenant"))
         .map_err(|err| anyhow!("tenant storage unavailable: {err:?}"))?;
     let execution = handler.handle(state, job.clone(), storage).await;
     assert!(matches!(execution, JobExecution::Success));

backend/tests/tenant_deletion.rs

@@ -1,13 +1,20 @@
-use anyhow::{anyhow, Result};
+use anyhow::{anyhow, bail, Result};
 use axum::http::StatusCode;
+use chrono::{Duration as ChronoDuration, Utc};
+use diesel::dsl::{count_star, exists, select};
 use diesel::prelude::*;
-use papercrate::jobs::{enqueue_job, JOB_DELETE_TENANT};
+use papercrate::jobs::{
+    enqueue_job, mark_job_failed, mark_job_succeeded, JOB_DELETE_TENANT, STATUS_FAILED,
+    STATUS_SUCCEEDED,
+};
 use papercrate::models::TenantStatus;
-use papercrate::schema::{documents, tenants, user_memberships};
+use papercrate::schema::{documents, jobs, tenants, user_memberships};
 use papercrate::test_support::{acquire_db_lock, body_to_vec, TestApp, TestUserRole};
-use papercrate::workers::tenants::DeleteTenantJob;
-use papercrate::workers::JobHandler;
-use serde_json::json;
+use papercrate::workers::tenants::{
+    build_delete_proof_message, sign_delete_proof, DeleteAction, DeleteTenantJob,
+};
+use papercrate::workers::{JobExecution, JobHandler};
+use serde_json::{json, Value};
 use std::sync::Arc;
 use uuid::Uuid;
 
@@ -20,63 +27,39 @@ async fn delete_tenant_job_keeps_tenant_when_requested() -> Result<()> {
     app.insert_user("tenant-keep", TestUserRole::Owner).await?;
     let token = app.login_token("tenant-keep", password).await?;
 
-    let upload = app
-        .upload_document(
-            "/api/documents",
-            "keep.pdf",
-            "application/pdf",
-            b"keep",
-            None,
-            &token,
-        )
-        .await?;
-    assert_eq!(upload.status(), StatusCode::CREATED);
-    body_to_vec(upload.into_body()).await?;
+    upload_fixture(&app, &token, "keep.pdf", b"keep").await?;
 
-    let tenant_id = app
-        .state
-        .tenants
-        .get_by_name("test_tenant")
-        .map_err(|err| anyhow!("tenant lookup failed: {err:?}"))?
-        .id;
+    let tenant_id = default_tenant_id(&app)?;
     set_tenant_status(&app, tenant_id, TenantStatus::Deleting).await?;
+    assert_tenant_status(&app, tenant_id, TenantStatus::Deleting).await?;
+
+    let storage = tenant_storage(&app, tenant_id)?;
+    let storage_prefix = storage.root_prefix().to_string();
+    let before = tenant_snapshot(&app, tenant_id, &storage_prefix).await?;
+    assert_eq!(before.doc_count, 1);
+    assert!(before.membership_count > 0);
+    assert!(!before.storage_keys.is_empty());
+    assert_storage_keys_present(&app, &before.storage_keys).await?;
 
     let job = enqueue_delete_job(&app, tenant_id, false).await?;
+    let job_id = job.id;
     let handler = DeleteTenantJob::new();
     let state = Arc::new(app.state.clone());
-    let storage = app
-        .state
-        .storage_for_tenant(tenant_id)
-        .map_err(|err| anyhow!("storage unavailable: {err:?}"))?;
-
     let execution = handler.handle(state, job, storage).await;
-    assert!(matches!(
-        execution,
-        papercrate::workers::JobExecution::Success
-    ));
+    assert_job_success(&execution);
+    record_job_outcome(&app, job_id, &execution).await?;
+    assert_eq!(fetch_job_status(&app, job_id).await?, STATUS_SUCCEEDED);
 
-    assert_eq!(app.storage().object_count().await, 0);
-
-    let tenant_id_for_check = tenant_id;
-    app.with_conn(move |conn| {
-        use diesel::dsl::count_star;
-
-        let doc_count: i64 = documents::table.select(count_star()).get_result(conn)?;
-        assert_eq!(doc_count, 0);
-
-        let membership_count: i64 = user_memberships::table
-            .select(count_star())
-            .get_result(conn)?;
-        assert!(membership_count > 0, "memberships should remain");
-
-        let status: TenantStatus = tenants::table
-            .find(tenant_id_for_check)
-            .select(tenants::status)
-            .first(conn)?;
-        assert_eq!(status, TenantStatus::Suspended);
-        Ok(())
-    })
-    .await?;
+    let after = tenant_snapshot(&app, tenant_id, &storage_prefix).await?;
+    assert_eq!(after.doc_count, 0);
+    assert!(after.membership_count > 0);
+    assert_storage_keys_absent(&app, &before.storage_keys).await?;
+    assert_eq!(storage_object_count(&app, &storage_prefix).await?, 0);
+    assert_eq!(
+        fetch_tenant_status(&app, tenant_id).await?,
+        TenantStatus::Suspended
+    );
+    assert!(tenant_exists(&app, tenant_id).await?);
 
     app.cleanup().await?;
     Ok(())
@@ -90,64 +73,38 @@ async fn delete_tenant_job_can_reset_tenant_to_active() -> Result<()> {
     let password = "delete-reset";
     app.insert_user("tenant-reset", TestUserRole::Owner).await?;
     let token = app.login_token("tenant-reset", password).await?;
+    upload_fixture(&app, &token, "reset.pdf", b"reset").await?;
 
-    let upload = app
-        .upload_document(
-            "/api/documents",
-            "reset.pdf",
-            "application/pdf",
-            b"reset",
-            None,
-            &token,
-        )
-        .await?;
-    assert_eq!(upload.status(), StatusCode::CREATED);
-    body_to_vec(upload.into_body()).await?;
-
-    let tenant_id = app
-        .state
-        .tenants
-        .get_by_name("test_tenant")
-        .map_err(|err| anyhow!("tenant lookup failed: {err:?}"))?
-        .id;
+    let tenant_id = default_tenant_id(&app)?;
     set_tenant_status(&app, tenant_id, TenantStatus::Deleting).await?;
+    assert_tenant_status(&app, tenant_id, TenantStatus::Deleting).await?;
+
+    let storage = tenant_storage(&app, tenant_id)?;
+    let storage_prefix = storage.root_prefix().to_string();
+    let before = tenant_snapshot(&app, tenant_id, &storage_prefix).await?;
+    assert_eq!(before.doc_count, 1);
+    assert!(before.membership_count > 0);
+    assert!(!before.storage_keys.is_empty());
+    assert_storage_keys_present(&app, &before.storage_keys).await?;
 
     let job = enqueue_delete_job_with_status(&app, tenant_id, false, Some("active")).await?;
+    let job_id = job.id;
     let handler = DeleteTenantJob::new();
     let state = Arc::new(app.state.clone());
-    let storage = app
-        .state
-        .storage_for_tenant(tenant_id)
-        .map_err(|err| anyhow!("storage unavailable: {err:?}"))?;
-
     let execution = handler.handle(state, job, storage).await;
-    assert!(matches!(
-        execution,
-        papercrate::workers::JobExecution::Success
-    ));
+    assert_job_success(&execution);
+    record_job_outcome(&app, job_id, &execution).await?;
+    assert_eq!(fetch_job_status(&app, job_id).await?, STATUS_SUCCEEDED);
 
-    assert_eq!(app.storage().object_count().await, 0);
-
-    let tenant_id_for_check = tenant_id;
-    app.with_conn(move |conn| {
-        use diesel::dsl::count_star;
-
-        let doc_count: i64 = documents::table.select(count_star()).get_result(conn)?;
-        assert_eq!(doc_count, 0);
-
-        let membership_count: i64 = user_memberships::table
-            .select(count_star())
-            .get_result(conn)?;
-        assert!(membership_count > 0, "memberships should remain");
-
-        let status: TenantStatus = tenants::table
-            .find(tenant_id_for_check)
-            .select(tenants::status)
-            .first(conn)?;
-        assert_eq!(status, TenantStatus::Active);
-        Ok(())
-    })
-    .await?;
+    let after = tenant_snapshot(&app, tenant_id, &storage_prefix).await?;
+    assert_eq!(after.doc_count, 0);
+    assert!(after.membership_count > 0);
+    assert_storage_keys_absent(&app, &before.storage_keys).await?;
+    assert_eq!(storage_object_count(&app, &storage_prefix).await?, 0);
+    assert_eq!(
+        fetch_tenant_status(&app, tenant_id).await?,
+        TenantStatus::Active
+    );
 
     app.cleanup().await?;
     Ok(())
@@ -163,67 +120,269 @@ async fn delete_tenant_job_removes_tenant_entirely() -> Result<()> {
         .await?;
     let token = app.login_token("tenant-remove", password).await?;
 
-    let upload = app
+    upload_fixture(&app, &token, "remove-1.pdf", b"remove-1").await?;
+    upload_fixture(&app, &token, "remove-2.pdf", b"remove-2").await?;
+
+    let tenant_id = default_tenant_id(&app)?;
+    set_tenant_status(&app, tenant_id, TenantStatus::Deleting).await?;
+    assert_tenant_status(&app, tenant_id, TenantStatus::Deleting).await?;
+
+    let storage = tenant_storage(&app, tenant_id)?;
+    let storage_prefix = storage.root_prefix().to_string();
+    let before = tenant_snapshot(&app, tenant_id, &storage_prefix).await?;
+    assert!(before.doc_count >= 2);
+    assert!(before.membership_count > 0);
+    assert!(before.storage_keys.len() >= 2);
+    assert_storage_keys_present(&app, &before.storage_keys).await?;
+
+    let job = enqueue_delete_job(&app, tenant_id, true).await?;
+    let job_id = job.id;
+    let handler = DeleteTenantJob::new();
+    let state = Arc::new(app.state.clone());
+    let execution = handler.handle(state, job, storage).await;
+    assert_job_success(&execution);
+    record_job_outcome(&app, job_id, &execution).await?;
+    assert_eq!(fetch_job_status(&app, job_id).await?, STATUS_SUCCEEDED);
+
+    let after = tenant_snapshot(&app, tenant_id, &storage_prefix).await?;
+    assert_eq!(after.doc_count, 0);
+    assert_eq!(after.membership_count, 0);
+    assert!(after.storage_keys.is_empty());
+    assert_storage_keys_absent(&app, &before.storage_keys).await?;
+    assert_eq!(storage_object_count(&app, &storage_prefix).await?, 0);
+    assert!(!tenant_exists(&app, tenant_id).await?);
+
+    app.cleanup().await?;
+    Ok(())
+}
+
+#[tokio::test]
+async fn delete_tenant_job_rejects_invalid_signature() -> Result<()> {
+    let _lock = acquire_db_lock().await;
+    let app = TestApp::new().await?;
+
+    let password = "delete-invalid";
+    app.insert_user("tenant-invalid", TestUserRole::Owner)
+        .await?;
+    let token = app.login_token("tenant-invalid", password).await?;
+
+    upload_fixture(&app, &token, "invalid.pdf", b"invalid").await?;
+
+    let tenant_id = default_tenant_id(&app)?;
+    set_tenant_status(&app, tenant_id, TenantStatus::Deleting).await?;
+    assert_tenant_status(&app, tenant_id, TenantStatus::Deleting).await?;
+
+    let storage = tenant_storage(&app, tenant_id)?;
+    let storage_prefix = storage.root_prefix().to_string();
+    let before = tenant_snapshot(&app, tenant_id, &storage_prefix).await?;
+    assert_eq!(before.doc_count, 1);
+    assert!(before.membership_count > 0);
+    assert!(!before.storage_keys.is_empty());
+    assert_storage_keys_present(&app, &before.storage_keys).await?;
+
+    let job = enqueue_delete_job_with_invalid_signature(&app, tenant_id, false).await?;
+    let job_id = job.id;
+    let handler = DeleteTenantJob::new();
+    let state = Arc::new(app.state.clone());
+    let execution = handler.handle(state, job, storage).await;
+    match execution {
+        JobExecution::Failed { ref error } => {
+            assert!(error.contains("signature"), "unexpected error: {error}");
+        }
+        _ => bail!("delete job should fail when signature is invalid"),
+    }
+    record_job_outcome(&app, job_id, &execution).await?;
+    assert_eq!(fetch_job_status(&app, job_id).await?, STATUS_FAILED);
+
+    let after = tenant_snapshot(&app, tenant_id, &storage_prefix).await?;
+    assert_eq!(after.doc_count, before.doc_count);
+    assert_eq!(after.membership_count, before.membership_count);
+    assert_storage_keys_present(&app, &before.storage_keys).await?;
+    assert_eq!(
+        storage_object_count(&app, &storage_prefix).await?,
+        before.storage_keys.len()
+    );
+    assert_eq!(
+        fetch_tenant_status(&app, tenant_id).await?,
+        TenantStatus::Deleting
+    );
+
+    app.cleanup().await?;
+    Ok(())
+}
+
+#[tokio::test]
+async fn delete_tenant_job_rejects_invalid_final_status() -> Result<()> {
+    let _lock = acquire_db_lock().await;
+    let app = TestApp::new().await?;
+
+    let password = "delete-invalid-status";
+    app.insert_user("tenant-invalid-status", TestUserRole::Owner)
+        .await?;
+    let token = app.login_token("tenant-invalid-status", password).await?;
+
+    upload_fixture(&app, &token, "invalid-status.pdf", b"payload").await?;
+
+    let tenant_id = default_tenant_id(&app)?;
+    set_tenant_status(&app, tenant_id, TenantStatus::Deleting).await?;
+    assert_tenant_status(&app, tenant_id, TenantStatus::Deleting).await?;
+
+    let storage = tenant_storage(&app, tenant_id)?;
+    let storage_prefix = storage.root_prefix().to_string();
+    let before = tenant_snapshot(&app, tenant_id, &storage_prefix).await?;
+    assert_eq!(before.doc_count, 1);
+    assert!(before.membership_count > 0);
+
+    let job = enqueue_delete_job_with_overrides(
+        &app,
+        tenant_id,
+        false,
+        None,
+        PayloadOverrides {
+            final_status: Some("weird"),
+            ..PayloadOverrides::default()
+        },
+    )
+    .await?;
+    let job_id = job.id;
+    let handler = DeleteTenantJob::new();
+    let state = Arc::new(app.state.clone());
+    let execution = handler.handle(state, job, storage).await;
+    match execution {
+        JobExecution::Failed { ref error } => {
+            assert!(error.contains("payload"), "unexpected error: {error}");
+        }
+        _ => bail!("delete job should fail when final_status is invalid"),
+    }
+    record_job_outcome(&app, job_id, &execution).await?;
+    assert_eq!(fetch_job_status(&app, job_id).await?, STATUS_FAILED);
+
+    let after = tenant_snapshot(&app, tenant_id, &storage_prefix).await?;
+    assert_eq!(after.doc_count, before.doc_count);
+    assert_eq!(after.membership_count, before.membership_count);
+    assert_storage_keys_present(&app, &before.storage_keys).await?;
+    assert_eq!(
+        fetch_tenant_status(&app, tenant_id).await?,
+        TenantStatus::Deleting
+    );
+
+    app.cleanup().await?;
+    Ok(())
+}
+
+#[tokio::test]
+async fn delete_tenant_job_rejects_malformed_issued_at() -> Result<()> {
+    let _lock = acquire_db_lock().await;
+    let app = TestApp::new().await?;
+
+    let password = "delete-bad-issued-at";
+    app.insert_user("tenant-issued", TestUserRole::Owner)
+        .await?;
+    let token = app.login_token("tenant-issued", password).await?;
+
+    upload_fixture(&app, &token, "issued.pdf", b"issued").await?;
+
+    let tenant_id = default_tenant_id(&app)?;
+    set_tenant_status(&app, tenant_id, TenantStatus::Deleting).await?;
+    assert_tenant_status(&app, tenant_id, TenantStatus::Deleting).await?;
+
+    let storage = tenant_storage(&app, tenant_id)?;
+    let job = enqueue_delete_job_with_overrides(
+        &app,
+        tenant_id,
+        false,
+        None,
+        PayloadOverrides {
+            issued_at: Some("definitely-not-time"),
+            ..PayloadOverrides::default()
+        },
+    )
+    .await?;
+    let job_id = job.id;
+    let handler = DeleteTenantJob::new();
+    let state = Arc::new(app.state.clone());
+    let execution = handler.handle(state, job, storage).await;
+    match execution {
+        JobExecution::Failed { ref error } => {
+            assert!(error.contains("issued_at"), "unexpected error: {error}");
+        }
+        _ => bail!("delete job should fail when issued_at is malformed"),
+    }
+    record_job_outcome(&app, job_id, &execution).await?;
+    assert_eq!(fetch_job_status(&app, job_id).await?, STATUS_FAILED);
+
+    app.cleanup().await?;
+    Ok(())
+}
+
+#[tokio::test]
+async fn delete_tenant_job_rejects_stale_confirmation() -> Result<()> {
+    let _lock = acquire_db_lock().await;
+    let app = TestApp::new().await?;
+
+    let password = "delete-stale";
+    app.insert_user("tenant-stale", TestUserRole::Owner).await?;
+    let token = app.login_token("tenant-stale", password).await?;
+
+    upload_fixture(&app, &token, "stale.pdf", b"stale").await?;
+
+    let tenant_id = default_tenant_id(&app)?;
+    set_tenant_status(&app, tenant_id, TenantStatus::Deleting).await?;
+    assert_tenant_status(&app, tenant_id, TenantStatus::Deleting).await?;
+
+    let storage = tenant_storage(&app, tenant_id)?;
+    let stale_time = (Utc::now() - ChronoDuration::minutes(10)).to_rfc3339();
+    let job = enqueue_delete_job_with_overrides(
+        &app,
+        tenant_id,
+        false,
+        None,
+        PayloadOverrides {
+            issued_at: Some(&stale_time),
+            ..PayloadOverrides::default()
+        },
+    )
+    .await?;
+    let job_id = job.id;
+    let handler = DeleteTenantJob::new();
+    let state = Arc::new(app.state.clone());
+    let execution = handler.handle(state, job, storage).await;
+    match execution {
+        JobExecution::Failed { ref error } => {
+            assert!(error.contains("expired"), "unexpected error: {error}");
+        }
+        _ => bail!("delete job should fail when confirmation is stale"),
+    }
+    record_job_outcome(&app, job_id, &execution).await?;
+    assert_eq!(fetch_job_status(&app, job_id).await?, STATUS_FAILED);
+
+    app.cleanup().await?;
+    Ok(())
+}
+
+async fn upload_fixture(app: &TestApp, token: &str, filename: &str, contents: &[u8]) -> Result<()> {
+    let response = app
         .upload_document(
             "/api/documents",
-            "remove.pdf",
+            filename,
             "application/pdf",
-            b"remove",
+            contents,
             None,
-            &token,
+            token,
         )
         .await?;
-    assert_eq!(upload.status(), StatusCode::CREATED);
-    body_to_vec(upload.into_body()).await?;
+    assert_eq!(response.status(), StatusCode::CREATED);
+    body_to_vec(response.into_body()).await?;
+    Ok(())
+}
 
-    let tenant_id = app
+fn default_tenant_id(app: &TestApp) -> Result<Uuid> {
+    Ok(app
         .state
         .tenants
         .get_by_name("test_tenant")
         .map_err(|err| anyhow!("tenant lookup failed: {err:?}"))?
-        .id;
-    set_tenant_status(&app, tenant_id, TenantStatus::Deleting).await?;
-
-    let job = enqueue_delete_job(&app, tenant_id, true).await?;
-    let handler = DeleteTenantJob::new();
-    let state = Arc::new(app.state.clone());
-    let storage = app
-        .state
-        .storage_for_tenant(tenant_id)
-        .map_err(|err| anyhow!("storage unavailable: {err:?}"))?;
-
-    let execution = handler.handle(state, job, storage).await;
-    assert!(matches!(
-        execution,
-        papercrate::workers::JobExecution::Success
-    ));
-
-    assert_eq!(app.storage().object_count().await, 0);
-
-    let tenant_id_for_check = tenant_id;
-    app.with_conn(move |conn| {
-        use diesel::dsl::count_star;
-
-        let doc_count: i64 = documents::table.select(count_star()).get_result(conn)?;
-        assert_eq!(doc_count, 0);
-
-        let membership_count: i64 = user_memberships::table
-            .select(count_star())
-            .get_result(conn)?;
-        assert_eq!(membership_count, 0, "memberships should be removed");
-
-        let exists: Option<TenantStatus> = tenants::table
-            .find(tenant_id_for_check)
-            .select(tenants::status)
-            .first(conn)
-            .optional()?;
-        assert!(exists.is_none(), "tenant row should be deleted");
-        Ok(())
-    })
-    .await?;
-
-    app.cleanup().await?;
-    Ok(())
+        .id)
 }
 
 async fn set_tenant_status(app: &TestApp, tenant_id: Uuid, status: TenantStatus) -> Result<()> {
@@ -236,6 +395,131 @@ async fn set_tenant_status(app: &TestApp, tenant_id: Uuid, status: TenantStatus)
     .await
 }
 
+async fn assert_tenant_status(
+    app: &TestApp,
+    tenant_id: Uuid,
+    expected: TenantStatus,
+) -> Result<()> {
+    let status = fetch_tenant_status(app, tenant_id).await?;
+    assert_eq!(status, expected);
+    Ok(())
+}
+
+async fn tenant_snapshot(
+    app: &TestApp,
+    tenant_id: Uuid,
+    storage_prefix: &str,
+) -> Result<TenantSnapshot> {
+    let (doc_count, membership_count) = app
+        .with_conn(move |conn| {
+            let doc_count: i64 = documents::table
+                .filter(documents::tenant_id.eq(tenant_id))
+                .select(count_star())
+                .get_result(conn)?;
+            let membership_count: i64 = user_memberships::table
+                .filter(user_memberships::tenant_id.eq(tenant_id))
+                .select(count_star())
+                .get_result(conn)?;
+            Ok::<_, anyhow::Error>((doc_count, membership_count))
+        })
+        .await?;
+
+    let storage_keys = app.storage().keys_with_prefix(storage_prefix).await;
+
+    Ok(TenantSnapshot {
+        doc_count,
+        membership_count,
+        storage_keys,
+    })
+}
+
+async fn tenant_exists(app: &TestApp, tenant_id: Uuid) -> Result<bool> {
+    app.with_conn(move |conn| {
+        let exists_value: bool =
+            select(exists(tenants::table.filter(tenants::id.eq(tenant_id)))).get_result(conn)?;
+        Ok(exists_value)
+    })
+    .await
+}
+
+async fn fetch_tenant_status(app: &TestApp, tenant_id: Uuid) -> Result<TenantStatus> {
+    app.with_conn(move |conn| {
+        tenants::table
+            .find(tenant_id)
+            .select(tenants::status)
+            .first(conn)
+            .map_err(Into::into)
+    })
+    .await
+}
+
+async fn record_job_outcome(app: &TestApp, job_id: Uuid, execution: &JobExecution) -> Result<()> {
+    match execution {
+        JobExecution::Success => {
+            app.with_conn(move |conn| {
+                mark_job_succeeded(conn, job_id)
+                    .map_err(|err| anyhow!("mark succeeded failed: {err}"))
+            })
+            .await?
+        }
+        JobExecution::Failed { error } => {
+            let error = error.clone();
+            app.with_conn(move |conn| {
+                mark_job_failed(conn, job_id, &error)
+                    .map_err(|err| anyhow!("mark failed failed: {err}"))
+            })
+            .await?
+        }
+        JobExecution::Retry { .. } => bail!("retry outcome not expected in tenant deletion tests"),
+    }
+    Ok(())
+}
+
+async fn fetch_job_status(app: &TestApp, job_id: Uuid) -> Result<String> {
+    app.with_conn(move |conn| {
+        jobs::table
+            .find(job_id)
+            .select(jobs::status)
+            .first::<String>(conn)
+            .map_err(Into::into)
+    })
+    .await
+}
+
+fn tenant_storage(app: &TestApp, tenant_id: Uuid) -> Result<papercrate::storage::TenantStorage> {
+    app.state
+        .storage_for_tenant(tenant_id)
+        .map_err(|err| anyhow!("storage unavailable: {err:?}"))
+}
+
+async fn storage_object_count(app: &TestApp, prefix: &str) -> Result<usize> {
+    Ok(app.storage().object_count_with_prefix(prefix).await)
+}
+
+async fn assert_storage_keys_present(app: &TestApp, keys: &[String]) -> Result<()> {
+    let storage = app.storage();
+    for key in keys {
+        assert!(
+            storage.contains_key(key).await,
+            "expected storage object '{}' to exist",
+            key
+        );
+    }
+    Ok(())
+}
+
+async fn assert_storage_keys_absent(app: &TestApp, keys: &[String]) -> Result<()> {
+    let storage = app.storage();
+    for key in keys {
+        assert!(
+            !storage.contains_key(key).await,
+            "expected storage object '{}' to be deleted",
+            key
+        );
+    }
+    Ok(())
+}
+
 async fn enqueue_delete_job(
     app: &TestApp,
     tenant_id: Uuid,
@@ -250,6 +534,58 @@ async fn enqueue_delete_job_with_status(
     remove_tenant: bool,
     final_status: Option<&'static str>,
 ) -> Result<papercrate::models::Job> {
+    enqueue_delete_job_with_overrides(
+        app,
+        tenant_id,
+        remove_tenant,
+        final_status,
+        PayloadOverrides::default(),
+    )
+    .await
+}
+
+async fn enqueue_delete_job_with_invalid_signature(
+    app: &TestApp,
+    tenant_id: Uuid,
+    remove_tenant: bool,
+) -> Result<papercrate::models::Job> {
+    let mut job = enqueue_delete_job_with_overrides(
+        app,
+        tenant_id,
+        remove_tenant,
+        None,
+        PayloadOverrides::default(),
+    )
+    .await?;
+
+    let job_id = job.id;
+    let mut payload = job.payload.clone();
+    payload["signature"] = json!("deadbeefdeadbeefdeadbeefdeadbeef");
+    let payload_for_db = payload.clone();
+
+    app.with_conn(move |conn| {
+        diesel::update(jobs::table.find(job_id))
+            .set(jobs::payload.eq(payload_for_db))
+            .execute(conn)?;
+        Ok(())
+    })
+    .await?;
+
+    job.payload = payload;
+    Ok(job)
+}
+
+async fn enqueue_delete_job_with_overrides(
+    app: &TestApp,
+    tenant_id: Uuid,
+    remove_tenant: bool,
+    final_status: Option<&str>,
+    overrides: PayloadOverrides<'_>,
+) -> Result<papercrate::models::Job> {
+    let secret = app.state.config.jwt_secret.clone();
+    let overrides_owned = PayloadOverridesOwned::from(overrides);
+    let requested_final_status = final_status.map(|value| value.to_string());
+
     app.with_conn(move |conn| {
         let tenant_name: String = tenants::table
             .find(tenant_id)
@@ -257,16 +593,123 @@ async fn enqueue_delete_job_with_status(
             .first(conn)
             .map_err(|err| anyhow!("tenant lookup failed: {err}"))?;
 
-        let mut payload = json!({
-            "remove_tenant": remove_tenant,
-            "tenant_name": tenant_name,
-        });
-        if let Some(status) = final_status {
-            payload["final_status"] = json!(status);
-        }
+        let payload = build_signed_delete_payload(
+            tenant_id,
+            &tenant_name,
+            remove_tenant,
+            requested_final_status.as_deref(),
+            &secret,
+            overrides_owned.as_borrowed(),
+        )?;
 
         enqueue_job(conn, tenant_id, JOB_DELETE_TENANT, payload, None)
             .map_err(|err| anyhow!("enqueue failed: {err}"))
     })
     .await
 }
+
+fn build_signed_delete_payload(
+    tenant_id: Uuid,
+    tenant_name: &str,
+    remove_tenant: bool,
+    requested_final_status: Option<&str>,
+    secret: &str,
+    overrides: PayloadOverrides<'_>,
+) -> Result<Value> {
+    let action = if remove_tenant {
+        DeleteAction::Delete
+    } else {
+        DeleteAction::Reset
+    };
+
+    let nonce = overrides
+        .nonce
+        .map(|value| value.to_string())
+        .unwrap_or_else(|| format!("test-delete-nonce-{}", Uuid::new_v4()));
+    let issued_at = overrides
+        .issued_at
+        .map(|value| value.to_string())
+        .unwrap_or_else(|| Utc::now().to_rfc3339());
+    let payload_final_status = if remove_tenant {
+        None
+    } else if let Some(value) = overrides.final_status {
+        Some(value.to_string())
+    } else {
+        Some(requested_final_status.unwrap_or("suspended").to_string())
+    };
+
+    let message = build_delete_proof_message(
+        tenant_id,
+        tenant_name,
+        action,
+        &nonce,
+        &issued_at,
+        payload_final_status.as_deref(),
+    );
+    let signature = sign_delete_proof(secret, &message)
+        .map_err(|err| anyhow!("failed to sign delete proof: {err}"))?;
+
+    let mut payload = json!({
+        "remove_tenant": remove_tenant,
+        "tenant_name": tenant_name,
+        "action": action.as_str(),
+        "nonce": nonce,
+        "issued_at": issued_at,
+        "signature": signature,
+    });
+    if let Some(status) = payload_final_status {
+        payload["final_status"] = json!(status);
+    }
+
+    Ok(payload)
+}
+
+struct TenantSnapshot {
+    doc_count: i64,
+    membership_count: i64,
+    storage_keys: Vec<String>,
+}
+
+#[derive(Default)]
+struct PayloadOverrides<'a> {
+    final_status: Option<&'a str>,
+    issued_at: Option<&'a str>,
+    nonce: Option<&'a str>,
+}
+
+fn assert_job_success(execution: &JobExecution) {
+    match execution {
+        JobExecution::Success => {}
+        JobExecution::Failed { error } => panic!("delete job failed unexpectedly: {error}"),
+        JobExecution::Retry { error, .. } => {
+            panic!("delete job asked for retry unexpectedly: {error}")
+        }
+    }
+}
+
+#[derive(Default, Clone)]
+struct PayloadOverridesOwned {
+    final_status: Option<String>,
+    issued_at: Option<String>,
+    nonce: Option<String>,
+}
+
+impl<'a> From<PayloadOverrides<'a>> for PayloadOverridesOwned {
+    fn from(value: PayloadOverrides<'a>) -> Self {
+        Self {
+            final_status: value.final_status.map(|s| s.to_string()),
+            issued_at: value.issued_at.map(|s| s.to_string()),
+            nonce: value.nonce.map(|s| s.to_string()),
+        }
+    }
+}
+
+impl PayloadOverridesOwned {
+    fn as_borrowed(&self) -> PayloadOverrides<'_> {
+        PayloadOverrides {
+            final_status: self.final_status.as_deref(),
+            issued_at: self.issued_at.as_deref(),
+            nonce: self.nonce.as_deref(),
+        }
+    }
+}