diff --git a/crates/buzz-admin/src/main.rs b/crates/buzz-admin/src/main.rs index 7853a7843e..13e7f6df82 100644 --- a/crates/buzz-admin/src/main.rs +++ b/crates/buzz-admin/src/main.rs @@ -127,7 +127,7 @@ enum Command { #[command(subcommand)] command: deletions::DeletionsCommand, }, - /// Submit and inspect synthetic user-deletion requests. + /// Preview real user memberships or manage synthetic user-deletion requests. UserDeletions { #[command(subcommand)] command: user_deletions::UserDeletionsCommand, diff --git a/crates/buzz-admin/src/user_deletions.rs b/crates/buzz-admin/src/user_deletions.rs index 6508df8c6e..594f9189b3 100644 --- a/crates/buzz-admin/src/user_deletions.rs +++ b/crates/buzz-admin/src/user_deletions.rs @@ -14,6 +14,13 @@ mod tests { #[test] fn user_workflow_commands_are_registered() { for args in [ + vec![ + "buzz-admin", + "user-deletions", + "inventory", + "--pubkey", + "5e4083a7a793d02575876eaf47932b4835999f5af2653d26250a909b450da7c9", + ], vec![ "buzz-admin", "user-deletions", @@ -90,8 +97,16 @@ mod tests { "--pause-after-handler" ]) .is_err()); - for command in ["inventory", "run", "drain", "worker"] { + for command in ["run", "drain", "worker"] { assert!(crate::Cli::try_parse_from(["buzz-admin", "user-deletions", command]).is_err()); } + assert!(crate::Cli::try_parse_from([ + "buzz-admin", + "user-deletions", + "inventory", + "--pubkey", + "not-a-key", + ]) + .is_err()); } } diff --git a/crates/buzz-admin/tests/postgres_user_inventory.rs b/crates/buzz-admin/tests/postgres_user_inventory.rs new file mode 100644 index 0000000000..2c53cb053d --- /dev/null +++ b/crates/buzz-admin/tests/postgres_user_inventory.rs @@ -0,0 +1,394 @@ +//! Exercise the real CLI with SELECT-only access and no workflow tables. + +use nostr::ToBech32; +use serde_json::Value; +use sqlx::PgPool; +use tokio::process::Command; +use url::Url; +use uuid::Uuid; + +fn admin_binary() -> std::ffi::OsString { + // Nextest relocates binaries when extracting an archive. Cargo's build-time + // path remains the fallback for local cargo test runs. + std::env::var_os("NEXTEST_BIN_EXE_buzz_admin") + .unwrap_or_else(|| env!("CARGO_BIN_EXE_buzz-admin").into()) +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn cluster_global_inventory_cli_uses_only_existing_membership_tables() { + let database_url = std::env::var("BUZZ_TEST_DATABASE_URL") + .or_else(|_| std::env::var("DATABASE_URL")) + .unwrap(); + let pool = PgPool::connect(&database_url).await.unwrap(); + // All interpolated identifiers are this fixed prefix plus generated hex. + let name = format!("inventory_{}", Uuid::new_v4().simple()); + let mut tx = pool.begin().await.unwrap(); + sqlx::raw_sql(sqlx::AssertSqlSafe(format!( + "CREATE SCHEMA {name}; SET LOCAL search_path TO {name}, public;" + ))) + .execute(&mut *tx) + .await + .unwrap(); + // These are existing deployed tables. No entity_deletion_* schema is needed. + sqlx::raw_sql(include_str!( + "../../../schema/tables/public/communities.sql" + )) + .execute(&mut *tx) + .await + .unwrap(); + sqlx::raw_sql(include_str!( + "../../../schema/tables/public/relay_members.sql" + )) + .execute(&mut *tx) + .await + .unwrap(); + for schema in [ + include_str!("../../../schema/tables/public/channels.sql"), + include_str!("../../../schema/tables/public/channel_members.sql"), + ] { + sqlx::raw_sql(schema).execute(&mut *tx).await.unwrap(); + } + let pubkey = "5e4083a7a793d02575876eaf47932b4835999f5af2653d26250a909b450da7c9"; + let community = Uuid::new_v4(); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, 'inventory-cli.example')") + .bind(community) + .execute(&mut *tx) + .await + .unwrap(); + sqlx::query("INSERT INTO relay_members (community_id, pubkey, role) VALUES ($1, $2, 'member')") + .bind(community) + .bind(pubkey) + .execute(&mut *tx) + .await + .unwrap(); + let participant_community = Uuid::new_v4(); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, 'channel-only-cli.example')") + .bind(participant_community) + .execute(&mut *tx) + .await + .unwrap(); + for target in [community, participant_community] { + for _ in 0..2 { + let channel = Uuid::new_v4(); + sqlx::query("INSERT INTO channels (community_id, id, name, created_by) VALUES ($1, $2, 'inventory-test', decode($3, 'hex'))") + .bind(target).bind(channel).bind(pubkey).execute(&mut *tx).await.unwrap(); + sqlx::query("INSERT INTO channel_members (community_id, channel_id, pubkey) VALUES ($1, $2, decode($3, 'hex'))") + .bind(target).bind(channel).bind(pubkey).execute(&mut *tx).await.unwrap(); + } + } + sqlx::raw_sql(sqlx::AssertSqlSafe(format!( + "CREATE ROLE {name} LOGIN PASSWORD 'inventory-local-fixture'; \ + GRANT USAGE ON SCHEMA {name} TO {name}; \ + GRANT SELECT ON ALL TABLES IN SCHEMA {name} TO {name};" + ))) + .execute(&mut *tx) + .await + .unwrap(); + tx.commit().await.unwrap(); + + let mut reader_url = Url::parse(&database_url).unwrap(); + reader_url.set_username(&name).unwrap(); + reader_url + .set_password(Some("inventory-local-fixture")) + .unwrap(); + reader_url + .query_pairs_mut() + .append_pair("options", &format!("-csearch_path={name}")); + let npub = nostr::PublicKey::parse(pubkey) + .unwrap() + .to_bech32() + .unwrap(); + let mut outputs = Vec::new(); + for input in [pubkey, npub.as_str()] { + outputs.push( + tokio::time::timeout( + std::time::Duration::from_secs(15), + Command::new(admin_binary()) + .args(["user-deletions", "inventory", "--pubkey", input]) + .env("DATABASE_URL", reader_url.as_str()) + .kill_on_drop(true) + .output(), + ) + .await + .unwrap() + .unwrap(), + ); + } + let demo = Command::new(admin_binary()) + .args(["user-deletions", "inventory"]) + .env("DATABASE_URL", reader_url.as_str()) + .kill_on_drop(true) + .output() + .await + .unwrap(); + // Clean cluster-wide role state before asserting the subprocess results. + sqlx::raw_sql(sqlx::AssertSqlSafe(format!( + "DROP SCHEMA {name} CASCADE; DROP OWNED BY {name}; DROP ROLE {name};" + ))) + .execute(&pool) + .await + .unwrap(); + let inventories: Vec = outputs + .into_iter() + .map(|output| { + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + serde_json::from_slice(&output.stdout).unwrap() + }) + .collect(); + assert_eq!(inventories[0], inventories[1]); + assert!( + demo.status.success(), + "{}", + String::from_utf8_lossy(&demo.stderr) + ); + let demo: Value = serde_json::from_slice(&demo.stdout).unwrap(); + assert_eq!( + demo, + serde_json::json!({ + "communities": [ + {"community_id": "00000000-0000-0000-0000-00000000000b", "label": "demo:community-b", "relationship": "participant"}, + {"community_id": "00000000-0000-0000-0000-00000000000c", "label": "demo:community-c", "relationship": "former_participant"} + ] + }) + ); + let inventory = &inventories[0]; + assert_eq!(inventory.as_object().unwrap().len(), 1); + assert_eq!(demo.as_object().unwrap().len(), 1); + let communities = inventory["communities"].as_array().unwrap(); + assert_eq!(communities.len(), 2); + let explicit_member = communities + .iter() + .find(|item| item["community_id"] == community.to_string()) + .unwrap(); + assert_eq!(explicit_member["label"], "inventory-cli.example"); + assert_eq!(explicit_member["relationship"], "participant"); + let participant = communities + .iter() + .find(|item| item["community_id"] == participant_community.to_string()) + .unwrap(); + assert_eq!(participant["label"], "channel-only-cli.example"); + assert_eq!(participant["relationship"], "participant"); + for output in [&demo, inventory] { + for row in output["communities"].as_array().unwrap() { + assert_eq!(row.as_object().unwrap().len(), 3); + } + } +} + +#[test] +fn invalid_inventory_key_fails_before_database_connection() { + let output = std::process::Command::new(admin_binary()) + .args(["user-deletions", "inventory", "--pubkey", "invalid-npub"]) + .env_remove("DATABASE_URL") + .output() + .unwrap(); + assert_eq!(output.status.code(), Some(2)); + assert!(output.stdout.is_empty()); + assert!(String::from_utf8_lossy(&output.stderr) + .contains("expected a 64-character hex public key or an npub")); +} + +async fn workflow_cli(url: &str, args: &[&str], success: bool) -> Value { + let output = tokio::time::timeout( + std::time::Duration::from_secs(20), + Command::new(admin_binary()) + .arg("user-deletions") + .args(args) + .env("DATABASE_URL", url) + .env("BUZZ_DB_POOL_SIZE", "1") + .kill_on_drop(true) + .output(), + ) + .await + .unwrap() + .unwrap(); + assert_eq!( + output.status.success(), + success, + "{args:?}: {}", + String::from_utf8_lossy(&output.stderr) + ); + if success { + serde_json::from_slice(&output.stdout).unwrap() + } else { + Value::Null + } +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn inventory_submission_runs_demo_and_public_key_workflows_end_to_end() { + let url = std::env::var("BUZZ_TEST_DATABASE_URL").unwrap(); + let pool = PgPool::connect(&url).await.unwrap(); + let key = "ab".repeat(32); + let npub = nostr::PublicKey::parse(&key).unwrap().to_bech32().unwrap(); + let community = Uuid::new_v4(); + let channel = Uuid::new_v4(); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, 'workflow-fixture.example')") + .bind(community) + .execute(&pool) + .await + .unwrap(); + sqlx::query("INSERT INTO channels (community_id, id, name, created_by) VALUES ($1, $2, 'fixture', decode($3, 'hex'))") + .bind(community).bind(channel).bind(&key).execute(&pool).await.unwrap(); + sqlx::query("INSERT INTO channel_members (community_id, channel_id, pubkey) VALUES ($1, $2, decode($3, 'hex'))") + .bind(community).bind(channel).bind(&key).execute(&pool).await.unwrap(); + + for subject in ["demo", npub.as_str()] { + let preview = workflow_cli(&url, &["inventory", "--pubkey", subject], true).await; + let saved = workflow_cli( + &url, + &[ + "submit", + "--pubkey", + subject, + "--requested-by", + "operator-a", + ], + true, + ) + .await; + assert_eq!(saved["inventory_manifest"], preview); + assert_eq!( + saved["target"], + if subject == "demo" { + "demo:default" + } else { + &key + } + ); + assert_eq!(saved["status"], "inventoried"); + let id = saved["id"].as_str().unwrap(); + workflow_cli(&url, &["run", id], false).await; + workflow_cli( + &url, + &[ + "approve", + id, + "--approved-by", + "operator-b", + "--inventory-digest", + "wrong", + ], + false, + ) + .await; + let approved = workflow_cli( + &url, + &[ + "approve", + id, + "--approved-by", + "operator-b", + "--inventory-digest", + saved["inventory_digest"].as_str().unwrap(), + ], + true, + ) + .await; + assert!(approved["stage_history"].as_array().unwrap().is_empty()); + let stepped = workflow_cli(&url, &["run", id, "--next-step"], true).await; + assert_eq!(stepped["stage_history"][0]["status"], "completed"); + let first_id = stepped["stage_history"][0]["id"].clone(); + let done = workflow_cli(&url, &["run", id], true).await; + assert_eq!(done["status"], "completed"); + assert_eq!(done["stage_history"][0]["id"], first_id); + let stages = done["stage_history"].as_array().unwrap(); + assert_eq!(stages.len(), if subject == "demo" { 5 } else { 3 }); + assert!(stages + .iter() + .all(|s| s["status"] == "completed" && s["is_current"] == false)); + let inspected = workflow_cli(&url, &["inspect", id], true).await; + assert_eq!(inspected, done); + // Hex and npub converge on the same saved request without rediscovery. + let repeat = workflow_cli( + &url, + &[ + "submit", + "--pubkey", + if subject == "demo" { "demo" } else { &key }, + "--requested-by", + "operator-a", + ], + true, + ) + .await; + for (field, value) in repeat.as_object().unwrap() { + assert_eq!(value, &done[field]); + } + workflow_cli( + &url, + &[ + "submit", + "--pubkey", + subject, + "--requested-by", + "someone-else", + ], + false, + ) + .await; + } + // Real membership data remains unchanged after all dummy handlers execute. + let count: i64 = sqlx::query_scalar("SELECT count(*) FROM channel_members WHERE community_id = $1 AND pubkey = decode($2, 'hex')") + .bind(community).bind(&key).fetch_one(&pool).await.unwrap(); + assert_eq!(count, 1); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn inventory_approval_and_resume_use_saved_account_identity_without_resolution() { + let url = std::env::var("BUZZ_TEST_DATABASE_URL").unwrap(); + let pool = PgPool::connect(&url).await.unwrap(); + let saved = workflow_cli( + &url, + &[ + "submit", + "--target", + "demo:resolved-account", + "--requested-by", + "operator-a", + ], + true, + ) + .await; + let id = saved["id"].as_str().unwrap(); + // Model the output of a future account resolver. This is not CLI support + // for resolving account IDs: the inventory is already saved and complete. + let account = "builderlab:account-123"; + sqlx::query("UPDATE entity_deletion_requests SET target = $2 WHERE id = $1") + .bind(Uuid::parse_str(id).unwrap()) + .bind(account) + .execute(&pool) + .await + .unwrap(); + workflow_cli( + &url, + &[ + "approve", + id, + "--approved-by", + "operator-b", + "--inventory-digest", + saved["inventory_digest"].as_str().unwrap(), + ], + true, + ) + .await; + let stepped = workflow_cli(&url, &["run", id, "--next-step"], true).await; + let first_stage = stepped["stage_history"][0]["id"].clone(); + let completed = workflow_cli(&url, &["run", id], true).await; + assert_eq!(completed["target"], account); + assert_eq!(completed["inventory_manifest"], saved["inventory_manifest"]); + assert_eq!(completed["status"], "completed"); + assert_eq!(completed["stage_history"][0]["id"], first_stage); + for stage in completed["stage_history"].as_array().unwrap() { + assert_eq!(stage["metadata_json"]["inputs"]["target"], account); + assert_eq!(stage["status"], "completed"); + } +} diff --git a/crates/buzz-db/src/runtime/migration.rs b/crates/buzz-db/src/runtime/migration.rs index 9caa4aa7e9..ccbd8615ab 100644 --- a/crates/buzz-db/src/runtime/migration.rs +++ b/crates/buzz-db/src/runtime/migration.rs @@ -772,7 +772,9 @@ mod postgres_tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 71); + assert_eq!(migrations.len(), 73); + assert_eq!(migrations[72].version, 73); + assert_eq!(migrations[71].version, 72); assert_eq!(migrations[70].version, 71); assert!(migrations[70] .sql diff --git a/crates/buzz-db/src/store/entity_deletion/approvals.rs b/crates/buzz-db/src/store/entity_deletion/approvals.rs index a507b58648..9cd78becc5 100644 --- a/crates/buzz-db/src/store/entity_deletion/approvals.rs +++ b/crates/buzz-db/src/store/entity_deletion/approvals.rs @@ -11,7 +11,7 @@ pub struct EntityApproval { pub request_id: Uuid, /// SHA-256 of the inventory this operator approved. pub inventory_digest: String, - /// Self-asserted demo identity, not authenticated real-deletion authorization. + /// Operator identifier supplied by the caller; this store does not authenticate it. pub approved_by: String, /// Time this approval was first recorded. pub approved_at: DateTime, @@ -21,8 +21,9 @@ impl EntityDeletionStore { /// Record approval of the saved inventory without creating or executing stages. /// /// An identical retry is read-only. Another operator adds retained evidence - /// without restarting the workflow. The operator identity is self-asserted and - /// this demo-only boundary does not authorize real user deletion. + /// without restarting the workflow. This store does not authenticate the + /// operator: today's dummy CLI accepts self-asserted operator names. Callers + /// enabling real deletion must authenticate and authorize the operator. pub async fn approve( &self, id: Uuid, diff --git a/crates/buzz-db/src/store/entity_deletion/leases.rs b/crates/buzz-db/src/store/entity_deletion/leases.rs index a8b88ef9ae..92fcc0b789 100644 --- a/crates/buzz-db/src/store/entity_deletion/leases.rs +++ b/crates/buzz-db/src/store/entity_deletion/leases.rs @@ -1,4 +1,4 @@ -//! Parent-request ownership and stage attempts for the synthetic workflow. +//! Parent-request ownership and stage attempts for deletion workflows. use std::time::Duration; @@ -35,7 +35,7 @@ pub struct EntityExecution { pub next_attempt_at: Option>, /// Last recorded stage failure. pub last_error: Option, - /// Time all synthetic stages completed. + /// Time the workflow completed. pub completed_at: Option>, } diff --git a/crates/buzz-db/src/store/entity_deletion/mod.rs b/crates/buzz-db/src/store/entity_deletion/mod.rs index 2090b0de7e..e78c5d719c 100644 --- a/crates/buzz-db/src/store/entity_deletion/mod.rs +++ b/crates/buzz-db/src/store/entity_deletion/mod.rs @@ -5,7 +5,7 @@ use serde::Serialize; use sqlx::PgPool; use uuid::Uuid; -/// Writer-database access to synthetic deletion requests. +/// Writer-database access to deletion workflow requests. #[derive(Clone)] pub struct EntityDeletionStore { pool: PgPool, diff --git a/crates/buzz-db/src/store/entity_deletion/requests.rs b/crates/buzz-db/src/store/entity_deletion/requests.rs index b8a59a254b..4e976cc538 100644 --- a/crates/buzz-db/src/store/entity_deletion/requests.rs +++ b/crates/buzz-db/src/store/entity_deletion/requests.rs @@ -8,14 +8,14 @@ pub(crate) const REQUEST_COLUMNS: &str = "id, entity_type, target, status, requested_by, reason, created_at, updated_at, inventory_manifest, inventory_digest"; -/// A demonstration request; approval authorizes only the synthetic workflow. +/// A persisted request whose identity and workflow are selected by intake. #[derive(Debug, Serialize, sqlx::FromRow)] pub struct EntityRequest { /// Stable request identifier. pub id: Uuid, - /// Currently always `user`. + /// Workflow discriminator supplied by intake. pub entity_type: String, - /// Synthetic `demo:` label, not a resolved account identity. + /// Opaque identity identifier supplied by intake. pub target: String, /// Intake, approval, execution (`running`/`retry_wait`), or `completed`. pub status: String, @@ -27,7 +27,7 @@ pub struct EntityRequest { pub created_at: DateTime, /// Time of the last change; identical resubmissions do not change it. pub updated_at: DateTime, - /// Saved synthetic communities in processing order; absent on legacy intake. + /// Saved communities in processing order; absent on legacy intake. pub inventory_manifest: Option, /// SHA-256 of the saved inventory. pub inventory_digest: Option, diff --git a/crates/buzz-db/src/store/entity_deletion/stages_postgres_tests.rs b/crates/buzz-db/src/store/entity_deletion/stages_postgres_tests.rs index b2c19b7bfa..428f392963 100644 --- a/crates/buzz-db/src/store/entity_deletion/stages_postgres_tests.rs +++ b/crates/buzz-db/src/store/entity_deletion/stages_postgres_tests.rs @@ -6,7 +6,7 @@ async fn stage_storage_accepts_caller_policy_and_rolls_back_invalid_decisions() let pool = PgPool::connect(&crate::test_support::database_url()) .await .unwrap(); - let db = Db::from_pool(pool); + let db = Db::from_pool(pool.clone()); let request = db .user_deletion_store() .submit("demo:generic-storage", "operator", None) @@ -38,6 +38,7 @@ async fn stage_storage_accepts_caller_policy_and_rolls_back_invalid_decisions() let NextStage::Stage(row) = store .transition_stage(&lease, None, |request, history| { assert!(history.is_empty()); + assert_eq!(request.entity_type, "user"); assert_eq!(request.status, "approved"); Ok(StageTransition::Initialize(NewEntityStage { stage: "arbitrary_policy_stage".into(), diff --git a/crates/buzz-db/src/store/entity_deletion_postgres_tests.rs b/crates/buzz-db/src/store/entity_deletion_postgres_tests.rs index 51a98f7421..c93c460ce6 100644 --- a/crates/buzz-db/src/store/entity_deletion_postgres_tests.rs +++ b/crates/buzz-db/src/store/entity_deletion_postgres_tests.rs @@ -125,7 +125,7 @@ async fn desired_schema_has_plain_intake_without_database_guards() { .expect("community catalog"); assert!(sqlx::query( "INSERT INTO entity_deletion_requests (target, requested_by) - VALUES ('real-user', 'operator-a')", + VALUES ('', 'operator-a')", ) .execute(&store.pool) .await diff --git a/crates/buzz-db/src/store/user_deletion/discovery.rs b/crates/buzz-db/src/store/user_deletion/discovery.rs new file mode 100644 index 0000000000..27358e98ef --- /dev/null +++ b/crates/buzz-db/src/store/user_deletion/discovery.rs @@ -0,0 +1,165 @@ +//! Read-only discovery of a public key's current community memberships. + +use super::{UserDeletionStore, MAX_COMMUNITIES}; +use crate::{DbError, Result}; +use serde_json::{json, Value}; +use uuid::Uuid; + +/// Normalize a hex public key or NIP-19 npub to Buzz's lowercase hex form. +/// This does not resolve an email, account, or other keys owned by the person. +pub fn normalize_user_deletion_pubkey(input: &str) -> Result { + let is_hex = input.len() == 64 && input.bytes().all(|b| b.is_ascii_hexdigit()); + if !is_hex && !input.starts_with("npub1") { + return Err(invalid_pubkey()); + } + nostr::PublicKey::parse(input) + .map(|key| key.to_hex()) + .map_err(|_| invalid_pubkey()) +} + +fn invalid_pubkey() -> DbError { + DbError::DeletionSafety("expected a 64-character hex public key or an npub".into()) +} + +// Identity resolution belongs to intake, never approval or workflow recovery. +pub(super) fn normalize_target(input: &str) -> Result { + if let Some(slug) = input.strip_prefix("demo:") { + if slug.is_empty() + || slug.len() > 64 + || !slug.as_bytes()[0].is_ascii_alphanumeric() + || !slug + .bytes() + .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'_' || b == b'-') + { + return Err(DbError::DeletionSafety("invalid demo target label".into())); + } + return Ok(input.to_owned()); + } + normalize_user_deletion_pubkey(input) +} + +fn demo_inventory() -> serde_json::Value { + serde_json::json!({ + "communities": [ + { + "community_id": "00000000-0000-0000-0000-00000000000b", + "label": "demo:community-b", + "relationship": "participant" + }, + { + "community_id": "00000000-0000-0000-0000-00000000000c", + "label": "demo:community-c", + "relationship": "former_participant" + } + ] + }) +} + +#[derive(sqlx::FromRow)] +struct Membership { + community_id: Uuid, + host: String, + role: Option, +} + +impl UserDeletionStore { + /// Generate only community identities and relationships for demo or real input. + /// `demo` selects the existing fixture; other inputs accept hex or npub. + pub async fn generate_inventory(&self, pubkey: &str) -> Result { + let mut tx = self.pool.begin().await?; + sqlx::query("SET TRANSACTION READ ONLY") + .execute(&mut *tx) + .await?; + let inventory = generate_inventory_on(&mut tx, pubkey).await?; + tx.commit().await?; + Ok(inventory) + } +} + +// Submission uses its existing transaction, including when the pool has one connection. +pub(super) async fn generate_inventory_on( + conn: &mut sqlx::PgConnection, + pubkey: &str, +) -> Result { + if pubkey == "demo" { + return Ok(demo_inventory()); + } + let pubkey = normalize_target(pubkey)?; + if pubkey.starts_with("demo:") { + return Ok(demo_inventory()); + } + sqlx::query("SET LOCAL statement_timeout = '30s'") + .execute(&mut *conn) + .await?; + // Operator-global discovery: deliberately do not apply the product's + // active-community filters. One statement gives one membership snapshot. + let rows = sqlx::query_as::<_, Membership>( + "WITH current_channels AS ( \ + SELECT DISTINCT community_id FROM channel_members \ + WHERE pubkey = decode($1, 'hex') AND removed_at IS NULL \ + ), matched_communities AS ( \ + SELECT community_id FROM relay_members WHERE pubkey = $1 \ + UNION SELECT community_id FROM current_channels \ + ) \ + SELECT c.id AS community_id, c.host, m.role \ + FROM matched_communities matched \ + JOIN communities c ON c.id = matched.community_id \ + LEFT JOIN relay_members m ON m.community_id = c.id AND m.pubkey = $1 \ + ORDER BY c.id LIMIT $2", + ) + .bind(&pubkey) + .bind((MAX_COMMUNITIES + 1) as i64) + .fetch_all(&mut *conn) + .await?; + if rows.len() > MAX_COMMUNITIES { + return Err(DbError::DeletionSafety(format!( + "inventory exceeds {MAX_COMMUNITIES} communities; no partial inventory returned" + ))); + } + let communities = rows + .into_iter() + .map(|row| { + let relationship = match row.role.as_deref() { + Some("owner") => "owner", + Some("admin" | "member") | None => "participant", + _ => { + return Err(DbError::DeletionSafety( + "unsupported membership role".into(), + )) + } + }; + Ok(json!({ + "community_id": row.community_id, + "label": row.host, + "relationship": relationship, + })) + }) + .collect::>>()?; + Ok(json!({ + "communities": communities, + })) +} + +#[cfg(test)] +mod tests { + use super::*; + use nostr::ToBech32; + + #[test] + fn public_key_formats_identify_the_same_subject() { + let hex = "5e4083a7a793d02575876eaf47932b4835999f5af2653d26250a909b450da7c9"; + let npub = nostr::PublicKey::parse(hex).unwrap().to_bech32().unwrap(); + for input in [hex.to_owned(), hex.to_uppercase(), npub] { + assert_eq!(normalize_user_deletion_pubkey(&input).unwrap(), hex); + } + for input in [ + "", + "alice@example.com", + "demo:alice", + "npub1invalid", + "abcdef", + ] { + assert!(normalize_user_deletion_pubkey(input).is_err()); + } + } +} diff --git a/crates/buzz-db/src/store/user_deletion/discovery_postgres_tests.rs b/crates/buzz-db/src/store/user_deletion/discovery_postgres_tests.rs new file mode 100644 index 0000000000..00b9c9df9e --- /dev/null +++ b/crates/buzz-db/src/store/user_deletion/discovery_postgres_tests.rs @@ -0,0 +1,256 @@ +use super::*; +use nostr::ToBech32; +use serde_json::json; +use uuid::Uuid; + +const PUBKEY: &str = "5e4083a7a793d02575876eaf47932b4835999f5af2653d26250a909b450da7c9"; + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn owner_preview_is_available_but_submission_rolls_back_until_scope_is_supported() { + let pool = PgPool::connect(&crate::test_support::database_url()) + .await + .unwrap(); + let store = Db::from_pool(pool.clone()).user_deletion_store(); + let community = Uuid::new_v4(); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, 'owner.inventory.example')") + .bind(community) + .execute(&pool) + .await + .unwrap(); + sqlx::query("INSERT INTO relay_members (community_id, pubkey, role) VALUES ($1, $2, 'owner')") + .bind(community) + .bind(PUBKEY) + .execute(&pool) + .await + .unwrap(); + let preview = store.generate_inventory(PUBKEY).await.unwrap(); + assert_eq!(preview["communities"][0]["relationship"], "owner"); + let error = store.submit(PUBKEY, "operator", None).await.unwrap_err(); + assert!(error + .to_string() + .contains("owned-community execution is not supported yet")); + let count: i64 = + sqlx::query_scalar("SELECT count(*) FROM entity_deletion_requests WHERE target = $1") + .bind(PUBKEY) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(count, 0); + + // A refused submission must not reserve the target or retain stale inventory. + sqlx::query("UPDATE relay_members SET role = 'member' WHERE community_id = $1 AND pubkey = $2") + .bind(community) + .bind(PUBKEY) + .execute(&pool) + .await + .unwrap(); + let saved = store.submit(PUBKEY, "operator", None).await.unwrap(); + assert_eq!(saved.status, "inventoried"); + assert_eq!( + saved.inventory_manifest.as_ref().unwrap()["communities"][0]["relationship"], + "participant" + ); + store + .approve( + saved.id, + "approver", + saved.inventory_digest.as_deref().unwrap(), + ) + .await + .unwrap(); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn channel_memberships_discover_participants_deduplicate_and_preserve_owners() { + let pool = PgPool::connect(&crate::test_support::database_url()) + .await + .unwrap(); + let mut expected = Vec::new(); + for (relay_role, removed, unrelated) in [ + (None, false, false), + (Some("owner"), false, false), + (Some("admin"), false, false), + (Some("member"), false, false), + (None, true, false), + (None, false, true), + ] { + let community = Uuid::new_v4(); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(community) + .bind(format!("{community}.inventory.example")) + .execute(&pool) + .await + .unwrap(); + if let Some(role) = relay_role { + sqlx::query( + "INSERT INTO relay_members (community_id, pubkey, role) VALUES ($1, $2, $3)", + ) + .bind(community) + .bind(PUBKEY) + .bind(role) + .execute(&pool) + .await + .unwrap(); + } + let key = if unrelated { + nostr::Keys::generate().public_key().to_hex() + } else { + PUBKEY.to_owned() + }; + // Reuse channel IDs across communities to exercise community-scoped identity. + // Two hidden channel memberships still count as one participating community. + for channel in [Uuid::from_u128(1), Uuid::from_u128(2)] { + sqlx::query("INSERT INTO channels (community_id, id, name, created_by) VALUES ($1, $2, 'inventory-test', decode($3, 'hex'))") + .bind(community).bind(channel).bind(&key) + .execute(&pool).await.unwrap(); + sqlx::query("INSERT INTO channel_members (community_id, channel_id, pubkey, removed_at, hidden_at) VALUES ($1, $2, decode($3, 'hex'), CASE WHEN $4 THEN now() END, now())") + .bind(community).bind(channel).bind(&key).bind(removed) + .execute(&pool).await.unwrap(); + } + if !removed && !unrelated { + expected.push((community, relay_role)); + } + } + expected.sort_by_key(|(community, _)| *community); + let store = Db::from_pool(pool).user_deletion_store(); + let inventory = store.generate_inventory(PUBKEY).await.unwrap(); + let communities = inventory["communities"].as_array().unwrap(); + assert_eq!(communities.len(), expected.len()); + for (actual, (community, role)) in communities.iter().zip(expected) { + assert_eq!(actual["community_id"], community.to_string()); + assert_eq!( + actual["relationship"], + if role == Some("owner") { + "owner" + } else { + "participant" + } + ); + assert_eq!(actual.as_object().unwrap().len(), 3); + } + let npub = nostr::PublicKey::parse(PUBKEY) + .unwrap() + .to_bech32() + .unwrap(); + assert_eq!(store.generate_inventory(&npub).await.unwrap(), inventory); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn current_inventory_includes_all_lifecycle_states_and_excludes_removed_memberships() { + let pool = PgPool::connect(&crate::test_support::database_url()) + .await + .unwrap(); + let store = Db::from_pool(pool.clone()).user_deletion_store(); + let mut expected = Vec::new(); + for (role, state, archived) in [ + ("owner", "active", false), + ("admin", "active", false), + ("member", "active", false), + ("member", "active", true), + ("owner", "quiescing", false), + ("member", "fenced", true), + ("member", "tombstone", false), + ] { + let id = Uuid::new_v4(); + let host = format!("{id}.inventory.example"); + let mut tx = pool.begin().await.unwrap(); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(id) + .bind(&host) + .execute(&mut *tx) + .await + .unwrap(); + sqlx::query("INSERT INTO relay_members (community_id, pubkey, role) VALUES ($1, $2, $3)") + .bind(id) + .bind(PUBKEY) + .bind(role) + .execute(&mut *tx) + .await + .unwrap(); + // Lifecycle fixture changes use the existing community executor fence. + sqlx::query("SELECT set_config('buzz.deletion_executor_community', $1, true), set_config('buzz.deletion_fence_generation', '0', true)") + .bind(id.to_string()).execute(&mut *tx).await.unwrap(); + sqlx::query("UPDATE communities SET deletion_state = $2, archived_at = CASE WHEN $3 THEN '2026-01-01'::timestamptz END, deleted_at = CASE WHEN $2 = 'tombstone' THEN '2026-01-01'::timestamptz END WHERE id = $1") + .bind(id).bind(state).bind(archived).execute(&mut *tx).await.unwrap(); + tx.commit().await.unwrap(); + expected.push(json!({ + "community_id": id, "label": host, + "relationship": if role == "owner" { "owner" } else { "participant" }, + })); + } + let former = Uuid::new_v4(); + sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)") + .bind(former) + .bind(format!("{former}.former.example")) + .execute(&pool) + .await + .unwrap(); + sqlx::query("INSERT INTO relay_members (community_id, pubkey, role) VALUES ($1, $2, 'member')") + .bind(former) + .bind(PUBKEY) + .execute(&pool) + .await + .unwrap(); + sqlx::query("DELETE FROM relay_members WHERE community_id = $1 AND pubkey = $2") + .bind(former) + .bind(PUBKEY) + .execute(&pool) + .await + .unwrap(); + // An unrelated identity in this community must not count as our membership. + let other = nostr::Keys::generate().public_key().to_hex(); + sqlx::query("INSERT INTO relay_members (community_id, pubkey, role) VALUES ($1, $2, 'owner')") + .bind(former) + .bind(&other) + .execute(&pool) + .await + .unwrap(); + + expected.sort_by(|a, b| a["community_id"].as_str().cmp(&b["community_id"].as_str())); + let inventory = store.generate_inventory(PUBKEY).await.unwrap(); + assert_eq!( + inventory, + json!({ + "communities": expected, + }) + ); + let npub = nostr::PublicKey::parse(PUBKEY) + .unwrap() + .to_bech32() + .unwrap(); + assert_eq!(store.generate_inventory(&npub).await.unwrap(), inventory); + assert_eq!( + store + .generate_inventory(&nostr::Keys::generate().public_key().to_hex()) + .await + .unwrap()["communities"], + json!([]) + ); + pool.close().await; + assert!( + store.generate_inventory(PUBKEY).await.is_err(), + "a failed lookup must not become an empty inventory" + ); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn oversized_inventory_fails_instead_of_truncating() { + let pool = PgPool::connect(&crate::test_support::database_url()) + .await + .unwrap(); + let pubkey = nostr::Keys::generate().public_key().to_hex(); + sqlx::query("WITH inserted AS (INSERT INTO communities (host) SELECT gen_random_uuid()::text || '.inventory.example' FROM generate_series(1, 1001) RETURNING id) INSERT INTO relay_members (community_id, pubkey, role) SELECT id, $1, 'member' FROM inserted") + .bind(&pubkey).execute(&pool).await.unwrap(); + let result = Db::from_pool(pool) + .user_deletion_store() + .generate_inventory(&pubkey) + .await; + assert!(result + .unwrap_err() + .to_string() + .contains("no partial inventory returned")); +} diff --git a/crates/buzz-db/src/store/user_deletion/inventory.rs b/crates/buzz-db/src/store/user_deletion/inventory.rs index 3c565db326..07941e512c 100644 --- a/crates/buzz-db/src/store/user_deletion/inventory.rs +++ b/crates/buzz-db/src/store/user_deletion/inventory.rs @@ -1,6 +1,6 @@ -//! User target validation and synthetic inventory generation. +//! Validation of saved inventories, independent of the intake identity format. +use super::MAX_COMMUNITIES; use crate::entity_deletion::EntityRequest; -use crate::store::entity_deletion::requests::validate_operator; use crate::{DbError, Result}; use serde_json::{json, Value}; use std::collections::HashSet; @@ -9,51 +9,6 @@ use uuid::Uuid; fn invalid(message: &str) -> DbError { DbError::DeletionSafety(message.into()) } -pub(super) fn validate_submission( - target: &str, - requested_by: &str, - reason: Option<&str>, -) -> Result<()> { - let slug = target.strip_prefix("demo:").unwrap_or_default(); - if slug.is_empty() - || slug.len() > 64 - || !slug.as_bytes()[0].is_ascii_alphanumeric() - || !slug - .bytes() - .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'_' || b == b'-') - { - return Err(DbError::DeletionSafety( - "target must be demo:<1..64 lowercase letters, digits, underscores or hyphens>".into(), - )); - } - validate_operator(requested_by)?; - if reason.is_some_and(|r| r.len() > 4096) { - return Err(DbError::DeletionSafety("reason exceeds 4096 bytes".into())); - } - Ok(()) -} - -pub(super) fn demo_inventory(target: &str) -> serde_json::Value { - serde_json::json!({ - "version": 1, - "target": target, - "communities": [ - { - "community_id": "00000000-0000-0000-0000-00000000000b", - "label": "demo:community-b", - "relationship": "participant", - "action": "delete_subject_data" - }, - { - "community_id": "00000000-0000-0000-0000-00000000000c", - "label": "demo:community-c", - "relationship": "former_participant", - "action": "delete_subject_data" - } - ] - }) -} - /// Validated user-deletion inventory, shared by preparation and workflow policy. pub struct UserDeletionInventory { /// Communities in approved processing order, with their saved attributes. @@ -66,33 +21,36 @@ impl UserDeletionInventory { if request.entity_type != "user" { return Err(invalid("user workflow requires entity_type user")); } - validate_submission( - &request.target, - &request.requested_by, - request.reason.as_deref(), - )?; let value = request .inventory_manifest .as_ref() .ok_or_else(|| invalid("missing inventory"))?; - if value["version"] != 1 || value["target"] != request.target { + // Preserve old manifests and their exact saved handler inputs on resume. + let legacy = value.get("version").is_some() || value.get("target").is_some(); + if legacy && (value["version"] != 1 || value["target"] != request.target) { return Err(invalid("unsupported inventory version or target")); } let rows = value["communities"] .as_array() .ok_or_else(|| invalid("missing communities"))?; - if rows.len() > 1000 { - return Err(invalid("inventory exceeds 1000 communities")); + if rows.len() > MAX_COMMUNITIES { + return Err(invalid(&format!( + "inventory exceeds {MAX_COMMUNITIES} communities" + ))); } let mut seen = HashSet::new(); let mut communities = Vec::new(); for row in rows { + if row["relationship"] == "owner" { + return Err(invalid("owned-community execution is not supported yet")); + } let id = row["community_id"] .as_str() .and_then(|s| Uuid::parse_str(s).ok()) .ok_or_else(|| invalid("invalid inventory community id"))?; if !seen.insert(id) - || row["action"] != "delete_subject_data" + || (legacy && row["action"] != "delete_subject_data") + || (!legacy && row.get("action").is_some()) || !matches!( row["relationship"].as_str(), Some("participant" | "former_participant") diff --git a/crates/buzz-db/src/store/user_deletion/inventory_postgres_tests.rs b/crates/buzz-db/src/store/user_deletion/inventory_postgres_tests.rs index a3181e3cbe..90da50e0f7 100644 --- a/crates/buzz-db/src/store/user_deletion/inventory_postgres_tests.rs +++ b/crates/buzz-db/src/store/user_deletion/inventory_postgres_tests.rs @@ -20,7 +20,9 @@ async fn user_approval_validates_inventory_shape_before_saving_evidence() { assert!(users .approve(request.id, "approver", &digest) .await - .is_err()); + .unwrap_err() + .to_string() + .contains("owned-community execution is not supported yet")); let after = db .entity_deletion_store() .inspect(request.id) @@ -30,3 +32,125 @@ async fn user_approval_validates_inventory_shape_before_saving_evidence() { assert!(after.approvals.is_empty()); assert!(after.stage_history.is_empty()); } + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn inventory_failure_rolls_back_request_and_legacy_manifests_remain_readable() { + let pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(1) + .connect(&crate::test_support::database_url()) + .await + .unwrap(); + let users = Db::from_pool(pool.clone()).user_deletion_store(); + let mut request = users.submit("demo:legacy", "operator", None).await.unwrap(); + let mut manifest = request.inventory_manifest.clone().unwrap(); + manifest["version"] = 1.into(); + manifest["target"] = request.target.clone().into(); + for row in manifest["communities"].as_array_mut().unwrap() { + row["action"] = "delete_subject_data".into(); + } + request.inventory_manifest = Some(manifest.clone()); + let legacy = UserDeletionInventory::read(&request).unwrap(); + assert_eq!(legacy.communities[0].1, manifest["communities"][0]); + request.inventory_manifest.as_mut().unwrap()["target"] = "demo:wrong".into(); + assert!(UserDeletionInventory::read(&request).is_err()); + + // A real discovery error must not leave a submitted request behind. + sqlx::query("ALTER TABLE channel_members RENAME TO unavailable_channel_members") + .execute(&pool) + .await + .unwrap(); + let key = "ab".repeat(32); + assert!(users.submit(&key, "operator", None).await.is_err()); + let count: i64 = + sqlx::query_scalar("SELECT count(*) FROM entity_deletion_requests WHERE target = $1") + .bind(&key) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(count, 0); +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn migration_schema_user_target_constraint_matches_desired_state() { + let pool = sqlx::PgPool::connect(&crate::test_support::database_url()) + .await + .unwrap(); + Db::from_pool(pool.clone()).migrate().await.unwrap(); + let mut tx = pool.begin().await.unwrap(); + sqlx::raw_sql("CREATE SCHEMA target_parity; SET LOCAL search_path TO target_parity, public;") + .execute(&mut *tx) + .await + .unwrap(); + sqlx::raw_sql(include_str!( + "../../../../../schema/tables/public/entity_deletion_requests.sql" + )) + .execute(&mut *tx) + .await + .unwrap(); + let definitions: Vec<(String, bool)> = sqlx::query_as( + "SELECT pg_get_constraintdef(oid), convalidated FROM pg_constraint + WHERE conname = 'entity_deletion_requests_target_check' ORDER BY conrelid", + ) + .fetch_all(&mut *tx) + .await + .unwrap(); + assert_eq!(definitions.len(), 2); + assert_eq!(definitions[0], definitions[1]); + assert!(definitions[0].1); + let restricted_types: i64 = sqlx::query_scalar( + "SELECT count(*) FROM pg_constraint WHERE conname = 'entity_deletion_requests_entity_type_check'", + ).fetch_one(&mut *tx).await.unwrap(); + assert_eq!(restricted_types, 2); + + tx.rollback().await.unwrap(); + let error = sqlx::query( + "INSERT INTO entity_deletion_requests (entity_type, target, requested_by) + VALUES ('repository', 'repository-123', 'operator')", + ) + .execute(&pool) + .await + .unwrap_err(); + assert_eq!( + error.as_database_error().unwrap().constraint(), + Some("entity_deletion_requests_entity_type_check") + ); + let users = Db::from_pool(pool.clone()).user_deletion_store(); + users + .submit("demo:old-client", "operator", None) + .await + .unwrap(); + users + .submit(&"ab".repeat(32), "operator", None) + .await + .unwrap(); + // Storage remains identity-format agnostic. Account lookup belongs in intake. + use nostr::ToBech32; + let npub = nostr::PublicKey::parse(&"ab".repeat(32)) + .unwrap() + .to_bech32() + .unwrap(); + for target in ["builderlab:account-123", npub.as_str()] { + sqlx::query( + "INSERT INTO entity_deletion_requests (target, requested_by) VALUES ($1, 'operator')", + ) + .bind(target) + .execute(&pool) + .await + .unwrap(); + } + for target in ["", " ", " account-123", "account-123 ", &"x".repeat(257)] { + assert!(sqlx::query( + "INSERT INTO entity_deletion_requests (target, requested_by) VALUES ($1, 'operator')" + ) + .bind(target) + .execute(&pool) + .await + .is_err()); + } + assert!(users + .submit("builderlab:account-123", "operator", None) + .await + .is_err()); +} diff --git a/crates/buzz-db/src/store/user_deletion/mod.rs b/crates/buzz-db/src/store/user_deletion/mod.rs index 25ad3b32dc..1c09394694 100644 --- a/crates/buzz-db/src/store/user_deletion/mod.rs +++ b/crates/buzz-db/src/store/user_deletion/mod.rs @@ -1,10 +1,14 @@ //! User-specific submission and inventory preparation over the shared deletion tables. use crate::Db; use sqlx::PgPool; +mod discovery; mod inventory; mod requests; +pub use discovery::normalize_user_deletion_pubkey; pub use inventory::UserDeletionInventory; +const MAX_COMMUNITIES: usize = 1000; + /// Database operations that prepare user-deletion requests. #[derive(Clone)] pub struct UserDeletionStore { @@ -22,3 +26,6 @@ impl Db { #[cfg(test)] #[path = "inventory_postgres_tests.rs"] mod postgres_tests; + +#[cfg(test)] +mod discovery_postgres_tests; diff --git a/crates/buzz-db/src/store/user_deletion/requests.rs b/crates/buzz-db/src/store/user_deletion/requests.rs index 913ed1d99e..081e056957 100644 --- a/crates/buzz-db/src/store/user_deletion/requests.rs +++ b/crates/buzz-db/src/store/user_deletion/requests.rs @@ -1,30 +1,34 @@ -//! Submit a user request and its dummy inventory in one transaction. -use super::{ - inventory::{demo_inventory, validate_submission}, - UserDeletionStore, -}; +//! Submit a user request and its discovered inventory in one transaction. +use super::{discovery::normalize_target, UserDeletionStore}; use crate::entity_deletion::EntityRequest; -use crate::store::entity_deletion::requests::{inventory_digest, REQUEST_COLUMNS}; +use crate::store::entity_deletion::requests::{ + inventory_digest, validate_operator, REQUEST_COLUMNS, +}; use crate::{DbError, Result}; impl UserDeletionStore { /// Save a request or return an identical previous submission. /// /// A conflicting operator or reason returns an error without changing the /// existing request. Concurrent identical submissions converge on one row. - /// Legacy submitted rows acquire their demo inventory on matching resubmission. + /// Legacy submitted rows acquire their inventory on matching resubmission. + /// Hex and npub inputs normalize to the same target; saved inventories are not refreshed. pub async fn submit( &self, target: &str, requested_by: &str, reason: Option<&str>, ) -> Result { - validate_submission(target, requested_by, reason)?; + let target = normalize_target(target)?; + validate_operator(requested_by)?; + if reason.is_some_and(|r| r.len() > 4096) { + return Err(DbError::DeletionSafety("reason exceeds 4096 bytes".into())); + } let mut tx = self.pool.begin().await?; sqlx::query( "INSERT INTO entity_deletion_requests (target, requested_by, reason) VALUES ($1, $2, $3) ON CONFLICT (entity_type, target) DO NOTHING", ) - .bind(target) + .bind(&target) .bind(requested_by) .bind(reason) .execute(&mut *tx) @@ -33,7 +37,7 @@ impl UserDeletionStore { "SELECT {REQUEST_COLUMNS} FROM entity_deletion_requests WHERE entity_type = 'user' AND target = $1 FOR UPDATE", ))) - .bind(target) + .bind(&target) .fetch_one(&mut *tx) .await?; if request.requested_by != requested_by || request.reason.as_deref() != reason { @@ -43,7 +47,7 @@ impl UserDeletionStore { ))); } if request.status == "submitted" { - let inventory = demo_inventory(target); + let inventory = super::discovery::generate_inventory_on(&mut tx, &target).await?; let digest = inventory_digest(&inventory)?; request = sqlx::query_as(sqlx::AssertSqlSafe(format!( "UPDATE entity_deletion_requests @@ -57,6 +61,9 @@ impl UserDeletionStore { .bind(digest) .fetch_one(&mut *tx) .await?; + // Use the approval validator before committing either the request + // or inventory, so unsupported scopes leave no new request behind. + super::UserDeletionInventory::read(&request)?; } tx.commit().await?; Ok(request) diff --git a/crates/buzz-deletion/src/user.rs b/crates/buzz-deletion/src/user.rs index d63f74cc7a..e10c57febd 100644 --- a/crates/buzz-deletion/src/user.rs +++ b/crates/buzz-deletion/src/user.rs @@ -1,17 +1,26 @@ -//! PostgreSQL-only intake, approval, and execution of synthetic user-deletion requests. +//! Read-only user inventory and synthetic user-deletion workflow commands. use anyhow::Result; use clap::Subcommand; use uuid::Uuid; -/// Prepare and execute synthetic requests; these commands never delete real data. +/// Preview real memberships or prepare and execute synthetic deletion requests. #[derive(Debug, Subcommand)] pub enum Command { + /// Print real current memberships as JSON; does not save or execute a request. + Inventory { + /// Demo fixture (default), hex public key, or npub. Current memberships only. + #[arg(long, default_value = "demo", value_parser = inventory::parse_subject)] + pubkey: String, + }, /// Save a request. An identical resubmission returns the existing row. Submit { - /// Synthetic user label, for example demo:alice. - #[arg(long)] - target: String, + /// Optional demo request label, for example demo:alice. Conflicts with --pubkey. + #[arg(long, conflicts_with = "pubkey")] + target: Option, + /// Demo fixture (default), hex public key, or npub to discover and save. + #[arg(long, value_parser = inventory::parse_subject)] + pubkey: Option, /// Submitting operator identity. #[arg(long)] requested_by: String, @@ -41,7 +50,7 @@ pub enum Command { #[arg(long)] next_step: bool, /// Inject a retryable failure when this dummy stage is reached. - #[arg(long, value_parser = ["gate_user", "purge_db", "purge_agents"])] + #[arg(long, value_parser = workflow::parse_stage)] fail_stage: Option, /// Delay each dummy action for inspection and shutdown testing (up to 60s). #[arg(long, default_value_t = 0, value_parser = clap::value_parser!(u64).range(0..=60_000))] @@ -63,20 +72,44 @@ pub enum Command { }, } -/// Run synthetic workflow commands using the existing deletion database connection. +/// Run inventory discovery or synthetic workflow commands against PostgreSQL. pub async fn run(command: Command) -> Result { - let db = crate::connect_database().await?; + let db = if matches!(command, Command::Inventory { .. }) { + buzz_db::Db::connect_cold_start( + buzz_db::DbConfig { + database_url: crate::required_env("DATABASE_URL")?, + max_connections: 1, + min_connections: 0, + default_transaction_read_only: true, + ..buzz_db::DbConfig::default() + }, + "user_inventory", + ) + .await? + } else { + crate::connect_database().await? + }; let store = db.entity_deletion_store(); match command { + Command::Inventory { pubkey } => { + crate::print_json(&inventory::generate(&db, &pubkey).await?)?; + } Command::Submit { target, + pubkey, requested_by, reason, - } => crate::print_json( - &db.user_deletion_store() - .submit(&target, &requested_by, reason.as_deref()) - .await?, - )?, + } => { + let target = target.unwrap_or_else(|| match pubkey.as_deref() { + None | Some("demo") => "demo:default".to_owned(), + Some(key) => key.to_owned(), + }); + crate::print_json( + &db.user_deletion_store() + .submit(&target, &requested_by, reason.as_deref()) + .await?, + )?; + } Command::Approve { id, approved_by, @@ -118,6 +151,7 @@ pub async fn run(command: Command) -> Result { } mod handlers; +mod inventory; mod transition; mod worker; mod workflow; diff --git a/crates/buzz-deletion/src/user/inventory.rs b/crates/buzz-deletion/src/user/inventory.rs new file mode 100644 index 0000000000..a9baaa88b0 --- /dev/null +++ b/crates/buzz-deletion/src/user/inventory.rs @@ -0,0 +1,13 @@ +//! Inventory preview and shared subject parsing for preview and submission. +use anyhow::Result; + +pub(super) async fn generate(db: &buzz_db::Db, pubkey: &str) -> Result { + Ok(db.user_deletion_store().generate_inventory(pubkey).await?) +} + +pub(super) fn parse_subject(input: &str) -> std::result::Result { + if input == "demo" { + return Ok(input.to_owned()); + } + buzz_db::user_deletion::normalize_user_deletion_pubkey(input).map_err(|error| error.to_string()) +} diff --git a/crates/buzz-deletion/src/user/worker.rs b/crates/buzz-deletion/src/user/worker.rs index 7d3cb54b7b..3cf6afe650 100644 --- a/crates/buzz-deletion/src/user/worker.rs +++ b/crates/buzz-deletion/src/user/worker.rs @@ -25,6 +25,14 @@ pub(super) async fn run( options: &RunOptions, shutdown: &CancellationToken, ) -> Result { + // The command selects this workflow; shared storage accepts other entity types. + let request = store.get(id).await?; + if request.entity_type != "user" { + bail!( + "user-deletions run does not support entity type {}", + request.entity_type + ); + } let Some(lease) = store.claim(id, owner, DEFAULT_LEASE_DURATION).await? else { if store.get(id).await?.status == "completed" { return Ok(0); diff --git a/crates/buzz-deletion/src/user/workflow.rs b/crates/buzz-deletion/src/user/workflow.rs index ce8e77a35a..ceeb3cbb08 100644 --- a/crates/buzz-deletion/src/user/workflow.rs +++ b/crates/buzz-deletion/src/user/workflow.rs @@ -63,6 +63,13 @@ pub(super) fn rule(name: &str) -> Result<&'static Rule> { .ok_or_else(|| invalid("unknown workflow stage")) } +// The CLI derives stage validation from this workflow, without a second list. +pub(super) fn parse_stage(name: &str) -> std::result::Result { + rule(name) + .map(|rule| rule.name.to_owned()) + .map_err(|error| error.to_string()) +} + pub(super) fn decision(stage: &EntityStage) -> Result { let rule = rule(&stage.stage)?; let result = stage @@ -211,3 +218,16 @@ impl Progress { } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn stage_options_follow_the_workflow_definition() { + for rule in RULES { + assert_eq!(parse_stage(rule.name).unwrap(), rule.name); + } + assert!(parse_stage("unregistered_stage").is_err()); + } +} diff --git a/docs/operator-user-deletion-intake.md b/docs/operator-user-deletion-intake.md index 44801d5c75..5e381a584c 100644 --- a/docs/operator-user-deletion-intake.md +++ b/docs/operator-user-deletion-intake.md @@ -1,11 +1,20 @@ # Prepare and approve a synthetic user-deletion request -Submit saves a request and a fixed dummy inventory: community B (participant), -then community C (former participant). These are artificial UUIDs, not discovered -communities. Approve saves authorization without creating stages. The -`run` command initializes execution through `get_next_stage`, dispatches dummy -handlers, and asks the transition function for the next saved row. No user or -community data is deleted. Only `demo:` labels are accepted. +For read-only discovery of actual memberships from a hex key or npub, see +[Preview a user's current-community inventory](operator-user-deletion-inventory.md). + +Submit saves a request and its inventory together. By default (`--pubkey demo`), +this is community B (participant), then community C (former participant). +Use `--pubkey ` to discover current communities instead. The request +stores the normalized hex key as its target; the inventory contains only +community IDs, labels and relationships. `--target demo: