From 8751637b92abaddc95fa5a66f727a8319f12c21b Mon Sep 17 00:00:00 2001 From: Nils Schneider Date: Tue, 11 Nov 2025 12:17:24 +0100 Subject: [PATCH] tenant deletion/reset --- .../down.sql | 13 + .../up.sql | 14 + backend/src/bin/admin.rs | 204 ++++++++- backend/src/documents/search.rs | 14 + backend/src/jobs.rs | 1 + backend/src/models.rs | 3 +- backend/src/schema.rs | 3 +- backend/src/storage.rs | 4 + backend/src/test_support/mod.rs | 19 + backend/src/workers/analyze.rs | 13 +- backend/src/workers/mod.rs | 20 +- backend/src/workers/purge.rs | 13 +- backend/src/workers/tenants.rs | 409 +++++++++++++++++- backend/tests/tenant_deletion.rs | 272 ++++++++++++ docs/jobs.md | 40 ++ 15 files changed, 1015 insertions(+), 27 deletions(-) create mode 100644 backend/migrations/202511050000_make_jobs_tenant_nullable/down.sql create mode 100644 backend/migrations/202511050000_make_jobs_tenant_nullable/up.sql create mode 100644 backend/tests/tenant_deletion.rs create mode 100644 docs/jobs.md diff --git a/backend/migrations/202511050000_make_jobs_tenant_nullable/down.sql b/backend/migrations/202511050000_make_jobs_tenant_nullable/down.sql new file mode 100644 index 0000000..3bbbf91 --- /dev/null +++ b/backend/migrations/202511050000_make_jobs_tenant_nullable/down.sql @@ -0,0 +1,13 @@ +ALTER TABLE jobs + DROP CONSTRAINT jobs_tenant_id_fkey; + +ALTER TABLE jobs + ALTER COLUMN tenant_id SET NOT NULL; + +ALTER TABLE jobs + ADD CONSTRAINT jobs_tenant_id_fkey + FOREIGN KEY (tenant_id) + REFERENCES tenants(id); + +ALTER TABLE jobs + DROP COLUMN result; diff --git a/backend/migrations/202511050000_make_jobs_tenant_nullable/up.sql b/backend/migrations/202511050000_make_jobs_tenant_nullable/up.sql new file mode 100644 index 0000000..5032ea2 --- /dev/null +++ b/backend/migrations/202511050000_make_jobs_tenant_nullable/up.sql @@ -0,0 +1,14 @@ +ALTER TABLE jobs + ALTER COLUMN tenant_id DROP NOT NULL; + +ALTER TABLE jobs + DROP CONSTRAINT jobs_tenant_id_fkey; + +ALTER TABLE jobs + ADD CONSTRAINT jobs_tenant_id_fkey + FOREIGN KEY (tenant_id) + REFERENCES tenants(id) + ON DELETE SET NULL; + +ALTER TABLE jobs + ADD COLUMN result JSONB; diff --git a/backend/src/bin/admin.rs b/backend/src/bin/admin.rs index 9945f49..a108142 100644 --- a/backend/src/bin/admin.rs +++ b/backend/src/bin/admin.rs @@ -14,7 +14,7 @@ use papercrate::{ config::AppConfig, db::{self, PgPool}, documents::search::ensure_quickwit_index, - jobs::{enqueue_job, JOB_ANALYZE_DOCUMENT}, + jobs::{enqueue_job, JOB_ANALYZE_DOCUMENT, JOB_DELETE_TENANT}, models::{ DocumentAsset, DocumentAssetObject, MagicToken, MagicTokenKind, NewUser, NewUserMembership, Tenant, TenantStatus, User, @@ -27,6 +27,7 @@ use papercrate::{ storage::{ObjectStorage, S3Storage, TenantStorage}, tenants::TenantService, utils::{text::normalize_identifier, tracing::init_tracing}, + workers::tenants::{build_delete_proof_message, sign_delete_proof, DeleteAction}, }; #[derive(Parser)] @@ -58,6 +59,15 @@ enum Command { }, DeleteTenant { tenant_id: Uuid, + #[arg(long = "tenant-name")] + tenant_name: String, + }, + ResetTenant { + tenant_id: Uuid, + #[arg(long = "tenant-name")] + tenant_name: String, + #[arg(long = "final-status", value_enum, default_value_t = TenantFinalStatusArg::Active)] + final_status: TenantFinalStatusArg, }, AddUserToTenant { username: String, @@ -80,6 +90,15 @@ enum Command { QuickwitDelete { tenant_id: Uuid, }, + EnqueueDeleteTenant { + tenant_id: Uuid, + #[arg(long = "tenant-name")] + tenant_name: String, + #[arg(long = "remove-tenant")] + remove_tenant: bool, + #[arg(long = "final-status", value_enum)] + final_status: Option, + }, MagicToken { username: String, #[arg(long = "ttl-minutes", default_value_t = 10)] @@ -112,6 +131,21 @@ impl From for MagicTokenKind { } } +#[derive(Copy, Clone, Debug, ValueEnum)] +enum TenantFinalStatusArg { + Active, + Suspended, +} + +impl TenantFinalStatusArg { + fn as_str(&self) -> &'static str { + match self { + TenantFinalStatusArg::Active => "active", + TenantFinalStatusArg::Suspended => "suspended", + } + } +} + #[tokio::main] async fn main() -> Result<()> { init_tracing("info"); @@ -128,7 +162,15 @@ async fn main() -> Result<()> { storage_root, quickwit_index, } => create_tenant(&pool, &name, storage_root, quickwit_index)?, - Command::DeleteTenant { tenant_id } => delete_tenant(&pool, tenant_id)?, + Command::DeleteTenant { + tenant_id, + tenant_name, + } => delete_tenant(&config, &pool, tenant_id, &tenant_name)?, + Command::ResetTenant { + tenant_id, + tenant_name, + final_status, + } => reset_tenant(&config, &pool, tenant_id, &tenant_name, final_status)?, Command::AddUserToTenant { username, tenant_id, @@ -148,6 +190,19 @@ async fn main() -> Result<()> { Command::QuickwitDelete { tenant_id } => { quickwit_index(&config, &pool, tenant_id, Method::DELETE).await? } + Command::EnqueueDeleteTenant { + tenant_id, + tenant_name, + remove_tenant, + final_status, + } => enqueue_delete_tenant_job( + &config, + &pool, + tenant_id, + &tenant_name, + remove_tenant, + final_status, + )?, Command::MagicToken { username, ttl_minutes, @@ -233,8 +288,17 @@ fn delete_user(pool: &PgPool, username: &str) -> Result<()> { .optional()? .ok_or_else(|| anyhow!("user '{}' not found", username))?; - diesel::delete(user_memberships::table.filter(user_memberships::user_id.eq(user.id))) - .execute(&mut conn)?; + let has_memberships: bool = select(exists( + user_memberships::table.filter(user_memberships::user_id.eq(user.id)), + )) + .get_result(&mut conn)?; + if has_memberships { + bail!( + "user '{}' is still a member of one or more tenants; remove memberships first", + username + ); + } + diesel::delete(users::table.filter(users::id.eq(user.id))).execute(&mut conn)?; println!("deleted user '{}'", username); @@ -335,7 +399,51 @@ fn hash_token(token: &str) -> String { hex::encode(hasher.finalize()) } -fn delete_tenant(pool: &PgPool, tenant_id: Uuid) -> Result<()> { +fn delete_tenant( + config: &AppConfig, + pool: &PgPool, + tenant_id: Uuid, + tenant_name: &str, +) -> Result<()> { + enqueue_delete_tenant_job_internal(config, pool, tenant_id, tenant_name, true, None)?; + println!( + "delete job enqueued; tenant '{}' will be permanently removed", + tenant_name + ); + Ok(()) +} + +fn reset_tenant( + config: &AppConfig, + pool: &PgPool, + tenant_id: Uuid, + tenant_name: &str, + final_status: TenantFinalStatusArg, +) -> Result<()> { + enqueue_delete_tenant_job_internal( + config, + pool, + tenant_id, + tenant_name, + false, + Some(final_status), + )?; + println!( + "reset job enqueued; tenant '{}' will be wiped and set to {}", + tenant_name, + final_status.as_str() + ); + Ok(()) +} + +fn enqueue_delete_tenant_job_internal( + config: &AppConfig, + pool: &PgPool, + tenant_id: Uuid, + expected_name: &str, + remove_tenant: bool, + final_status: Option, +) -> Result<()> { let mut conn = pool.get().context("failed to get database connection")?; let tenant: Tenant = tenants::table @@ -344,19 +452,89 @@ fn delete_tenant(pool: &PgPool, tenant_id: Uuid) -> Result<()> { .optional()? .ok_or_else(|| anyhow!("tenant '{}' not found", tenant_id))?; - let member_exists: bool = select(exists( - user_memberships::table.filter(user_memberships::tenant_id.eq(tenant.id)), - )) - .get_result(&mut conn)?; - if member_exists { - bail!("tenant '{}' still has user memberships", tenant.name); + if tenant.name != expected_name { + bail!( + "tenant name mismatch: expected '{}', database has '{}'", + expected_name, + tenant.name + ); } - diesel::delete(tenants::table.filter(tenants::id.eq(tenant.id))).execute(&mut conn)?; - println!("deleted tenant '{}'", tenant.name); + diesel::update(tenants::table.find(tenant.id)) + .set(tenants::status.eq(TenantStatus::Deleting)) + .execute(&mut conn)?; + + let nonce = generate_random_token(); + let issued_at = Utc::now(); + let issued_at_str = issued_at.to_rfc3339(); + let action = if remove_tenant { + DeleteAction::Delete + } else { + DeleteAction::Reset + }; + let resolved_final_status = if remove_tenant { + None + } else { + Some( + final_status + .unwrap_or(TenantFinalStatusArg::Suspended) + .as_str(), + ) + }; + + let message = build_delete_proof_message( + tenant.id, + expected_name, + action, + &nonce, + &issued_at_str, + resolved_final_status, + ); + let signature = sign_delete_proof(&config.jwt_secret, &message).map_err(|err| anyhow!(err))?; + + let mut payload = serde_json::json!({ + "remove_tenant": remove_tenant, + "tenant_name": tenant.name.clone(), + "action": action.as_str(), + "nonce": nonce, + "issued_at": issued_at_str, + "signature": signature, + }); + if let Some(status) = resolved_final_status { + payload["final_status"] = serde_json::json!(status); + } + + enqueue_job(&mut conn, tenant.id, JOB_DELETE_TENANT, payload, None)?; + let status_label = if remove_tenant { + "deleted" + } else { + final_status.map(|s| s.as_str()).unwrap_or("suspended") + }; + println!( + "delete-tenant job enqueued for '{}' (remove_tenant={}, final_status={})", + tenant.name, remove_tenant, status_label + ); Ok(()) } +fn enqueue_delete_tenant_job( + config: &AppConfig, + pool: &PgPool, + tenant_id: Uuid, + tenant_name: &str, + remove_tenant: bool, + final_status: Option, +) -> Result<()> { + enqueue_delete_tenant_job_internal( + config, + pool, + tenant_id, + tenant_name, + remove_tenant, + final_status, + ) +} + fn add_user_to_tenant(pool: &PgPool, username: &str, tenant_id: Uuid) -> Result<()> { let mut conn = pool.get().context("failed to get database connection")?; diff --git a/backend/src/documents/search.rs b/backend/src/documents/search.rs index d1e41c9..f238af7 100644 --- a/backend/src/documents/search.rs +++ b/backend/src/documents/search.rs @@ -163,6 +163,20 @@ pub async fn ensure_quickwit_index(client: &Client, endpoint: &str, index_id: &s } } +pub async fn delete_quickwit_index(client: &Client, endpoint: &str, index_id: &str) -> Result<()> { + let base = endpoint.trim_end_matches('/'); + let url = format!("{}/api/v1/indexes/{}", base, index_id); + let response = client.delete(&url).send().await?; + match response.status() { + status if status.is_success() => Ok(()), + StatusCode::NOT_FOUND => Ok(()), + status => { + let body = response.text().await.unwrap_or_default(); + bail!("quickwit delete index failed with status {status}: {body}"); + } + } +} + pub fn extract_document_id(hit: &Value) -> Option { for key in ["_source", "source", "fields", "stored_fields"] { if let Some(value) = hit.get(key) { diff --git a/backend/src/jobs.rs b/backend/src/jobs.rs index a5eb538..9d1df2a 100644 --- a/backend/src/jobs.rs +++ b/backend/src/jobs.rs @@ -20,6 +20,7 @@ pub const JOB_GENERATE_THUMBNAILS: &str = "generate-thumbnails"; pub const JOB_GENERATE_OCR_TEXT: &str = "generate-ocr-text"; pub const JOB_PROVISION_TENANT: &str = "provision-tenant"; pub const JOB_PURGE_DOCUMENT: &str = "purge-document"; +pub const JOB_DELETE_TENANT: &str = "delete-tenant"; #[derive(Debug, Error)] pub enum JobQueueError { diff --git a/backend/src/models.rs b/backend/src/models.rs index 9b57f20..ede3a84 100644 --- a/backend/src/models.rs +++ b/backend/src/models.rs @@ -654,7 +654,8 @@ pub struct Job { pub last_error: Option, pub created_at: NaiveDateTime, pub updated_at: NaiveDateTime, - pub tenant_id: Uuid, + pub tenant_id: Option, + pub result: Option, } #[derive(Debug, Insertable)] diff --git a/backend/src/schema.rs b/backend/src/schema.rs index c25f783..4a42aeb 100644 --- a/backend/src/schema.rs +++ b/backend/src/schema.rs @@ -131,7 +131,8 @@ diesel::table! { last_error -> Nullable, created_at -> Timestamptz, updated_at -> Timestamptz, - tenant_id -> Uuid, + tenant_id -> Nullable, + result -> Nullable, } } diff --git a/backend/src/storage.rs b/backend/src/storage.rs index ff9a030..1009cbe 100644 --- a/backend/src/storage.rs +++ b/backend/src/storage.rs @@ -133,6 +133,10 @@ impl TenantStorage { format!("{}{}", self.root, key) } + pub fn root_prefix(&self) -> &str { + &self.root + } + pub async fn put_object( &self, key: &str, diff --git a/backend/src/test_support/mod.rs b/backend/src/test_support/mod.rs index a0ce6e6..1cb90d5 100644 --- a/backend/src/test_support/mod.rs +++ b/backend/src/test_support/mod.rs @@ -129,6 +129,25 @@ impl FakeStorage { let guard = self.objects.lock().await; guard.len() } + + pub async fn object_count_with_prefix(&self, prefix: &str) -> usize { + let guard = self.objects.lock().await; + guard.keys().filter(|key| key.starts_with(prefix)).count() + } + + pub async fn contains_key(&self, key: &str) -> bool { + let guard = self.objects.lock().await; + guard.contains_key(key) + } + + pub async fn keys_with_prefix(&self, prefix: &str) -> Vec { + let guard = self.objects.lock().await; + guard + .keys() + .filter(|key| key.starts_with(prefix)) + .cloned() + .collect() + } } pub struct TestApp { diff --git a/backend/src/workers/analyze.rs b/backend/src/workers/analyze.rs index 835fb72..a94c8bf 100644 --- a/backend/src/workers/analyze.rs +++ b/backend/src/workers/analyze.rs @@ -50,7 +50,16 @@ impl JobHandler for AnalyzeDocumentJob { job: crate::models::Job, storage: TenantStorage, ) -> JobExecution { - if let Err(err) = ensure_active_tenant(&state, job.tenant_id) { + let tenant_id = match job.tenant_id { + Some(id) => id, + None => { + return JobExecution::Failed { + error: "job is no longer associated with a tenant".to_string(), + } + } + }; + + if let Err(err) = ensure_active_tenant(&state, tenant_id) { return JobExecution::Failed { error: err.to_string(), }; @@ -68,7 +77,7 @@ impl JobHandler for AnalyzeDocumentJob { let mut context = DocumentVersionTaskContext::new( job.id, JOB_ANALYZE_DOCUMENT, - job.tenant_id, + tenant_id, payload.document_id, payload.document_version_id, payload.force, diff --git a/backend/src/workers/mod.rs b/backend/src/workers/mod.rs index 61c9619..e6c7447 100644 --- a/backend/src/workers/mod.rs +++ b/backend/src/workers/mod.rs @@ -23,7 +23,7 @@ pub mod taskflow; pub mod tenants; pub mod thumbnails; -use tenants::ProvisionTenantJob; +use tenants::{DeleteTenantJob, ProvisionTenantJob}; #[derive(Debug)] pub enum JobExecution { @@ -93,8 +93,23 @@ impl Worker { drop(conn); if let Some(job) = job_opt { + let tenant_id = match job.tenant_id { + Some(id) => id, + None => { + warn!(job_id = %job.id, job_type = %job.job_type, "job detached from tenant; marking failed"); + if let Ok(mut conn) = self.state.db_unscoped() { + let _ = mark_job_failed( + &mut conn, + job.id, + "job detached from tenant context", + ); + } + return Ok(true); + } + }; + if let Some(handler) = self.handlers.get(job.job_type.as_str()) { - let execution = match self.state.storage_for_tenant(job.tenant_id) { + let execution = match self.state.storage_for_tenant(tenant_id) { Ok(storage) => { handler .handle(self.state.clone(), job.clone(), storage) @@ -153,6 +168,7 @@ pub fn default_handlers() -> Vec> { Arc::new(analyze::AnalyzeDocumentJob::new()), Arc::new(purge::PurgeDocumentJob::new()), Arc::new(ProvisionTenantJob::new()), + Arc::new(DeleteTenantJob::new()), ] } diff --git a/backend/src/workers/purge.rs b/backend/src/workers/purge.rs index ad43aa8..052d3e5 100644 --- a/backend/src/workers/purge.rs +++ b/backend/src/workers/purge.rs @@ -53,7 +53,16 @@ impl JobHandler for PurgeDocumentJob { job: crate::models::Job, storage: TenantStorage, ) -> JobExecution { - if let Err(err) = ensure_active_tenant(&state, job.tenant_id) { + let tenant_id = match job.tenant_id { + Some(id) => id, + None => { + return JobExecution::Failed { + error: "job is no longer associated with a tenant".to_string(), + } + } + }; + + if let Err(err) = ensure_active_tenant(&state, tenant_id) { return JobExecution::Failed { error: err.to_string(), }; @@ -71,7 +80,7 @@ impl JobHandler for PurgeDocumentJob { let mut context = PurgeTaskContext::new( job.id, JOB_PURGE_DOCUMENT, - job.tenant_id, + tenant_id, payload.document_id, state.clone(), storage, diff --git a/backend/src/workers/tenants.rs b/backend/src/workers/tenants.rs index 6d8a8c7..b523e6c 100644 --- a/backend/src/workers/tenants.rs +++ b/backend/src/workers/tenants.rs @@ -2,21 +2,30 @@ use std::sync::Arc; use std::time::Duration; use async_trait::async_trait; -use chrono::Utc; +use chrono::{DateTime, Utc}; use diesel::prelude::*; +use diesel::sql_types::Jsonb; +use hmac::{Hmac, Mac}; use reqwest::Client; use serde::Deserialize; +use serde_json::json; use tracing::warn; use uuid::Uuid; +use sha2::Sha256; + use crate::auth::capability_sets::{ ensure_capability_set, owner_capabilities, readonly_capabilities, user_capabilities, webdav_capabilities, }; -use crate::documents::search::ensure_quickwit_index; -use crate::jobs::JOB_PROVISION_TENANT; -use crate::models::{NewUserMembership, TenantStatus}; -use crate::schema::{tenants, user_memberships}; +use crate::documents::search::{delete_quickwit_index, ensure_quickwit_index}; +use crate::jobs::{JOB_DELETE_TENANT, JOB_PROVISION_TENANT}; +use crate::models::{NewUserMembership, Tenant, TenantStatus}; +use crate::schema::{ + api_tokens, correspondents, document_asset_objects, document_assets, document_correspondents, + document_tags, document_versions, documents, folders, tags, tenants, user_memberships, + user_sessions, +}; use crate::state::AppState; use crate::tenants::TenantRepository; use crate::workers::{ @@ -25,6 +34,10 @@ use crate::workers::{ JobExecution, JobHandler, }; +type HmacSha256 = Hmac; +const DELETE_PROOF_TTL_SECONDS: i64 = 300; +const DELETE_PROOF_VERSION: &str = "v1"; + pub struct ProvisionTenantJob; impl ProvisionTenantJob { @@ -45,11 +58,20 @@ impl JobHandler for ProvisionTenantJob { job: crate::models::Job, _storage: crate::storage::TenantStorage, ) -> JobExecution { + let tenant_id = match job.tenant_id { + Some(id) => id, + None => { + return JobExecution::Failed { + error: "provision job is missing tenant context".to_string(), + } + } + }; + let members = ProvisionPayload::from_job(&job).unwrap_or_default(); let mut context = ProvisionContext::new( job.id, JOB_PROVISION_TENANT, - job.tenant_id, + tenant_id, state.clone(), members, ); @@ -233,6 +255,244 @@ impl Task for ProvisionTask { } } +async fn delete_tenant( + state: &Arc, + storage: crate::storage::TenantStorage, + tenant_id: Uuid, + current_job_id: Uuid, + payload: &DeleteTenantPayload, +) -> Result<(), String> { + let remove_tenant = payload.remove_tenant; + let tenant = { + let mut conn = state + .db_unscoped() + .map_err(|err| format!("failed to get db connection: {err:?}"))?; + TenantRepository::get_by_id(&mut conn, tenant_id) + .map_err(|err| format!("tenant lookup failed: {err:?}"))? + }; + + if tenant.name != payload.tenant_name { + return Err(format!( + "tenant name mismatch: expected '{}', got '{}'", + payload.tenant_name, tenant.name + )); + } + + if tenant.status != TenantStatus::Deleting { + return Err(format!( + "tenant status '{}' not eligible for deletion", + tenant.status.as_str() + )); + } + + match (payload_action_applicable(remove_tenant), payload.action) { + (DeleteAction::Delete, DeleteAction::Delete) + | (DeleteAction::Reset, DeleteAction::Reset) => {} + _ => { + return Err("delete payload action mismatch".into()); + } + } + + let issued_at = DateTime::parse_from_rfc3339(&payload.issued_at) + .map_err(|_| "invalid issued_at timestamp".to_string())? + .with_timezone(&Utc); + if (Utc::now() - issued_at).num_seconds().abs() > DELETE_PROOF_TTL_SECONDS { + return Err("delete confirmation expired".into()); + } + + let resolved_final_status = if remove_tenant { + None + } else { + Some(payload.final_status.unwrap_or(FinalTenantStatus::Suspended)) + }; + let final_status_str = resolved_final_status.map(|s| s.as_str()); + + let message = build_delete_proof_message( + tenant_id, + &tenant.name, + payload.action, + &payload.nonce, + &payload.issued_at, + final_status_str, + ); + + verify_delete_proof(&state.config.jwt_secret, &message, &payload.signature)?; + + let object_keys = { + let mut conn = state + .db_for_tenant(tenant_id) + .map_err(|err| format!("failed to scope tenant connection: {err:?}"))?; + collect_object_keys(&mut conn) + .map_err(|err| format!("failed to collect storage keys: {err}"))? + }; + + delete_storage_objects(&storage, &object_keys) + .await + .map_err(|err| format!("failed to delete storage objects: {err}"))?; + + reset_quickwit_index(state, &tenant, remove_tenant) + .await + .map_err(|err| format!("quickwit cleanup failed: {err}"))?; + + { + let mut conn = state + .db_for_tenant(tenant_id) + .map_err(|err| format!("failed to scope tenant connection: {err:?}"))?; + delete_tenant_rows(&mut conn, tenant_id, remove_tenant) + .map_err(|err| format!("tenant data cleanup failed: {err}"))?; + } + + { + let mut conn = state + .db_unscoped() + .map_err(|err| format!("failed to get db connection: {err:?}"))?; + + let detach_result = json!({ + "tenant": { + "id": tenant.id, + "name": tenant.name, + }, + "action": payload.action.as_str(), + "remove_tenant": remove_tenant, + "final_status": final_status_str, + "timestamp": Utc::now().to_rfc3339(), + }); + + diesel::sql_query( + "UPDATE jobs \ + SET tenant_id = NULL, \ + result = jsonb_set(COALESCE(result, '{}'::jsonb), '{detached_tenant}', $3::jsonb, true) \ + WHERE tenant_id = $1 AND id <> $2", + ) + .bind::(tenant_id) + .bind::(current_job_id) + .bind::(detach_result) + .execute(&mut conn) + .map_err(|err| format!("failed to detach tenant jobs: {err}"))?; + + if remove_tenant { + diesel::delete(tenants::table.find(tenant_id)) + .execute(&mut conn) + .map_err(|err| format!("failed to delete tenant row: {err}"))?; + } else { + let new_status = match resolved_final_status.unwrap_or(FinalTenantStatus::Suspended) { + FinalTenantStatus::Active => TenantStatus::Active, + FinalTenantStatus::Suspended => TenantStatus::Suspended, + }; + diesel::update(tenants::table.find(tenant_id)) + .set(( + tenants::status.eq(new_status), + tenants::updated_at.eq(Utc::now().naive_utc()), + )) + .execute(&mut conn) + .map_err(|err| format!("failed to update tenant status: {err}"))?; + } + } + + Ok(()) +} + +struct TenantObjectKeys { + version_keys: Vec, + asset_keys: Vec, +} + +fn collect_object_keys(conn: &mut PgConnection) -> Result { + let version_keys = document_versions::table + .select(document_versions::s3_key) + .load::(conn)?; + let asset_keys = document_asset_objects::table + .select(document_asset_objects::s3_key) + .load::(conn)?; + + Ok(TenantObjectKeys { + version_keys, + asset_keys, + }) +} + +async fn delete_storage_objects( + storage: &crate::storage::TenantStorage, + keys: &TenantObjectKeys, +) -> Result<(), String> { + for key in keys.version_keys.iter().chain(keys.asset_keys.iter()) { + storage + .delete_object(key) + .await + .map_err(|err| format!("failed to delete object '{key}': {err}"))?; + } + Ok(()) +} + +async fn reset_quickwit_index( + state: &Arc, + tenant: &Tenant, + remove_index: bool, +) -> Result<(), String> { + let endpoint = match state.config.quickwit_endpoint.as_ref() { + Some(endpoint) => endpoint, + None => return Ok(()), + }; + + let index_id = match tenant.quickwit_index.as_deref() { + Some(index) => index, + None => return Ok(()), + }; + + let client = Client::new(); + delete_quickwit_index(&client, endpoint, index_id) + .await + .map_err(|err| format!("quickwit delete failed: {err}"))?; + + if !remove_index { + ensure_quickwit_index(&client, endpoint, index_id) + .await + .map_err(|err| format!("quickwit ensure failed: {err}"))?; + } + + Ok(()) +} + +fn delete_tenant_rows( + conn: &mut PgConnection, + tenant_id: Uuid, + remove_memberships: bool, +) -> Result<(), diesel::result::Error> { + conn.transaction(|conn| { + diesel::delete( + document_asset_objects::table.filter(document_asset_objects::tenant_id.eq(tenant_id)), + ) + .execute(conn)?; + diesel::delete(document_assets::table.filter(document_assets::tenant_id.eq(tenant_id))) + .execute(conn)?; + diesel::delete( + document_correspondents::table.filter(document_correspondents::tenant_id.eq(tenant_id)), + ) + .execute(conn)?; + diesel::delete(document_tags::table.filter(document_tags::tenant_id.eq(tenant_id))) + .execute(conn)?; + diesel::delete(document_versions::table.filter(document_versions::tenant_id.eq(tenant_id))) + .execute(conn)?; + diesel::delete(documents::table.filter(documents::tenant_id.eq(tenant_id))) + .execute(conn)?; + diesel::delete(folders::table.filter(folders::tenant_id.eq(tenant_id))).execute(conn)?; + diesel::delete(correspondents::table.filter(correspondents::tenant_id.eq(tenant_id))) + .execute(conn)?; + diesel::delete(tags::table.filter(tags::tenant_id.eq(tenant_id))).execute(conn)?; + diesel::delete(user_sessions::table.filter(user_sessions::tenant_id.eq(tenant_id))) + .execute(conn)?; + diesel::delete(api_tokens::table.filter(api_tokens::tenant_id.eq(tenant_id))) + .execute(conn)?; + if remove_memberships { + diesel::delete( + user_memberships::table.filter(user_memberships::tenant_id.eq(tenant_id)), + ) + .execute(conn)?; + } + Ok(()) + }) +} + #[derive(Deserialize, Default)] struct ProvisionPayload { #[serde(default)] @@ -246,3 +506,140 @@ impl ProvisionPayload { .ok() } } +pub struct DeleteTenantJob; + +impl DeleteTenantJob { + pub fn new() -> Self { + Self + } +} + +pub fn build_delete_proof_message( + tenant_id: Uuid, + tenant_name: &str, + action: DeleteAction, + nonce: &str, + issued_at: &str, + final_status: Option<&str>, +) -> String { + let status = final_status.unwrap_or("none"); + format!( + "{}|{}|{}|{}|{}|{}|{}", + DELETE_PROOF_VERSION, + tenant_id, + tenant_name, + action.as_str(), + nonce, + issued_at, + status + ) +} + +pub fn sign_delete_proof(secret: &str, message: &str) -> Result { + let mut mac = HmacSha256::new_from_slice(secret.as_bytes()) + .map_err(|err| format!("failed to init hmac: {err}"))?; + mac.update(message.as_bytes()); + let bytes = mac.finalize().into_bytes(); + Ok(hex::encode(bytes)) +} + +fn verify_delete_proof(secret: &str, message: &str, signature: &str) -> Result<(), String> { + let signature_bytes = hex::decode(signature) + .map_err(|_| "invalid delete proof signature encoding".to_string())?; + + let mut mac = HmacSha256::new_from_slice(secret.as_bytes()) + .map_err(|err| format!("failed to init hmac: {err}"))?; + mac.update(message.as_bytes()); + mac.verify_slice(&signature_bytes) + .map_err(|_| "delete proof signature mismatch".to_string()) +} + +#[derive(Debug, Deserialize)] +struct DeleteTenantPayload { + #[serde(default)] + remove_tenant: bool, + #[serde(default)] + final_status: Option, + tenant_name: String, + action: DeleteAction, + nonce: String, + issued_at: String, + signature: String, +} + +#[derive(Debug, Deserialize, Clone, Copy)] +#[serde(rename_all = "lowercase")] +enum FinalTenantStatus { + Active, + Suspended, +} + +impl FinalTenantStatus { + fn as_str(&self) -> &'static str { + match self { + FinalTenantStatus::Active => "active", + FinalTenantStatus::Suspended => "suspended", + } + } +} + +#[derive(Debug, Deserialize, Clone, Copy, PartialEq, Eq)] +#[serde(rename_all = "lowercase")] +pub enum DeleteAction { + Delete, + Reset, +} + +impl DeleteAction { + pub fn as_str(&self) -> &'static str { + match self { + DeleteAction::Delete => "delete", + DeleteAction::Reset => "reset", + } + } +} + +fn payload_action_applicable(remove_tenant: bool) -> DeleteAction { + if remove_tenant { + DeleteAction::Delete + } else { + DeleteAction::Reset + } +} + +#[async_trait] +impl JobHandler for DeleteTenantJob { + fn job_type(&self) -> &'static str { + JOB_DELETE_TENANT + } + + async fn handle( + &self, + state: Arc, + job: crate::models::Job, + storage: crate::storage::TenantStorage, + ) -> JobExecution { + let payload: DeleteTenantPayload = match serde_json::from_value(job.payload.clone()) { + Ok(payload) => payload, + Err(err) => { + return JobExecution::Failed { + error: format!("invalid delete tenant payload: {err}"), + } + } + }; + + let tenant_id = match job.tenant_id { + Some(id) => id, + None => { + return JobExecution::Failed { + error: "delete job is missing tenant context".to_string(), + } + } + }; + + match delete_tenant(&state, storage, tenant_id, job.id, &payload).await { + Ok(()) => JobExecution::Success, + Err(err) => JobExecution::Failed { error: err }, + } + } +} diff --git a/backend/tests/tenant_deletion.rs b/backend/tests/tenant_deletion.rs new file mode 100644 index 0000000..e70aed4 --- /dev/null +++ b/backend/tests/tenant_deletion.rs @@ -0,0 +1,272 @@ +use anyhow::{anyhow, Result}; +use axum::http::StatusCode; +use diesel::prelude::*; +use papercrate::jobs::{enqueue_job, JOB_DELETE_TENANT}; +use papercrate::models::TenantStatus; +use papercrate::schema::{documents, 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 std::sync::Arc; +use uuid::Uuid; + +#[tokio::test] +async fn delete_tenant_job_keeps_tenant_when_requested() -> Result<()> { + let _lock = acquire_db_lock().await; + let app = TestApp::new().await?; + + let password = "delete-keep"; + 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?; + + let tenant_id = 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, false).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!(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?; + + app.cleanup().await?; + Ok(()) +} + +#[tokio::test] +async fn delete_tenant_job_can_reset_tenant_to_active() -> Result<()> { + let _lock = acquire_db_lock().await; + let app = TestApp::new().await?; + + let password = "delete-reset"; + app.insert_user("tenant-reset", TestUserRole::Owner).await?; + let token = app.login_token("tenant-reset", password).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; + set_tenant_status(&app, tenant_id, TenantStatus::Deleting).await?; + + let job = enqueue_delete_job_with_status(&app, tenant_id, false, Some("active")).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!(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?; + + app.cleanup().await?; + Ok(()) +} + +#[tokio::test] +async fn delete_tenant_job_removes_tenant_entirely() -> Result<()> { + let _lock = acquire_db_lock().await; + let app = TestApp::new().await?; + + let password = "delete-remove"; + app.insert_user("tenant-remove", TestUserRole::Owner) + .await?; + let token = app.login_token("tenant-remove", password).await?; + + let upload = app + .upload_document( + "/api/documents", + "remove.pdf", + "application/pdf", + b"remove", + 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; + 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 = 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(()) +} + +async fn set_tenant_status(app: &TestApp, tenant_id: Uuid, status: TenantStatus) -> Result<()> { + app.with_conn(move |conn| { + diesel::update(tenants::table.find(tenant_id)) + .set(tenants::status.eq(status)) + .execute(conn)?; + Ok(()) + }) + .await +} + +async fn enqueue_delete_job( + app: &TestApp, + tenant_id: Uuid, + remove_tenant: bool, +) -> Result { + enqueue_delete_job_with_status(app, tenant_id, remove_tenant, None).await +} + +async fn enqueue_delete_job_with_status( + app: &TestApp, + tenant_id: Uuid, + remove_tenant: bool, + final_status: Option<&'static str>, +) -> Result { + app.with_conn(move |conn| { + let tenant_name: String = tenants::table + .find(tenant_id) + .select(tenants::name) + .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); + } + + enqueue_job(conn, tenant_id, JOB_DELETE_TENANT, payload, None) + .map_err(|err| anyhow!("enqueue failed: {err}")) + }) + .await +} diff --git a/docs/jobs.md b/docs/jobs.md new file mode 100644 index 0000000..a570e0c --- /dev/null +++ b/docs/jobs.md @@ -0,0 +1,40 @@ +# Job Catalogue + +Papercrate stores asynchronous work in the shared `jobs` table. Each job carries a +`tenant_id`, a small JSON payload, and one of the statuses defined in +`backend/src/jobs.rs` (`queued`, `processing`, `succeeded`, `failed`). Workers +continuously reserve jobs by type and execute the appropriate handler. This +document lists every job type that is currently recognized by the backend and +briefly describes what it does. + +| Job type | Payload shape | When it is enqueued | Work performed | +| --- | --- | --- | --- | +| `analyze-document` | `{ "document_id": Uuid, "document_version_id": Uuid, "force": bool }` | Uploading a document, calling the re-analyze bulk action, or after a metadata edit (e.g. title change) | Runs the taskflow pipeline (`GenerateThumbnailsTask`, `GenerateOcrTask`, `DetermineIssuedAtTask`, `IndexDocumentTask`) for the specified document version. The handler refuses to run if the tenant is not `Active`. | +| `purge-document` | `{ "document_id": Uuid }` | `DELETE /api/documents/{id}` after the document has been trashed | Removes every version and asset object from tenant storage, deletes database rows (`documents`, `document_versions`, associated assets/tags/correspondents), and leaves the system ready for GC. | +| `provision-tenant` | `{ "members": [Uuid, ...] }` | When a tenant is created with status `creating` | Creates/ensures the tenant’s Quickwit index, materializes the system capability sets (`owner`, `user`, `readonly`, `webdav`), attaches the initial member list, and flips the tenant status to `active`. | +| `delete-tenant` | `{ "remove_tenant": bool, "tenant_name": string, "action": "delete"\|"reset", "nonce": string, "issued_at": RFC3339 datetime, "signature": hex(HMAC-SHA256), "final_status"?: "active"\|"suspended" }` | Administrative action after a tenant has been marked `deleting` | Deletes all tenant-scoped storage objects, wipes the tenant’s Quickwit index (and optionally deletes it entirely), truncates the tenant schemas/tables, removes queued jobs for that tenant, and either deletes the tenant row or leaves it in the requested final status (defaults to `suspended`) while recreating an empty Quickwit index. | + +## Retired job types + +`generate-thumbnails` and `generate-ocr-text` once existed as standalone jobs. +Those behaviors now run as tasks inside `analyze-document`. No worker is +registered for the legacy types; keep them out of new payloads. + +### Tenant delete/reset safety checks + +The `delete-tenant` job refuses to run without a signed payload. The admin CLI +derives a message of the form `v1|tenant_id|tenant_name|action|nonce|issued_at|final_status` +and signs it with an HMAC-SHA256 key based on the server’s JWT secret. +Workers verify the signature, ensure the payload matches the job flags, and +require the `issued_at` timestamp to be no more than five minutes old. This +protects against accidental wipes triggered by stale requests or insufficiently +scoped API calls. + +## Operational notes + +* Every job handler calls `ensure_active_tenant` (or an equivalent guard) before + touching tenant data. If a tenant is suspended or deleting, the job will fail + immediately. +* Jobs are only enqueued for the tenant they operate on. Consequently, wiping a + tenant with `delete-tenant` also removes any remaining queued jobs for that + tenant so workers do not waste effort on work that can no longer succeed.