From 834dca26bf0613581fc8199810f4693392aada3b Mon Sep 17 00:00:00 2001 From: ravarora2 <130506156+ravarora2@users.noreply.github.com> Date: Mon, 26 Oct 2026 15:44:12 -0400 Subject: [PATCH] Generate user community inventory for the dummy deletion workflow (#8605) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Summary User deletion can now discover current communities from a hex public key or npub, save that inventory, and run it through the existing approval and dummy workflow. `--pubkey demo` remains the default. Execution only prints messages; it does not fence users or delete data. Discovery combines explicit relay memberships with active channel memberships, including users who have no relay membership row. It deduplicates communities and preserves owner relationships. The saved manifest contains only community IDs, labels and relationships. Historical discovery and owned-community execution remain later work. Submission rejects owner inventories atomically with a specific unsupported-execution error; preview still includes owners. This is an expand-schema change after #8520. Migrations 0072–0073 replace the demo-only target check with a bounded identifier check, with 5-second lock and 30-second statement timeouts. The database keeps the user-only entity-type check while allowing different identity formats in the target field. Workflow-specific commands reject entity types they cannot execute. The application currently resolves demo and hex/npub inputs; account-ID lookup is future work. The desired-state constraint matches. No backfill, trigger, function, or foreign key is added. Existing demo rows remain valid. Old binaries fail closed on new manifests; use the new CLI for their approval and execution. Older auto-migrating binaries need the documented rollback procedure. Identity parsing and demo fixture selection stay in intake. Approval and resume use the saved target without reparsing its format. The user workflow defines its stage names and handlers; CLI stage validation reads those rules instead of a separate three-name list. Shared persistence accepts caller-defined stages. Permanent target uniqueness is unchanged. Dummy submissions and execution belong in isolated, disposable development databases; production preview uses read-only access. CLI operator names remain self-asserted. ### Related issue Follows #8520. Searched open PRs; no duplicate inventory change found. #8489 is earlier workflow scaffolding. ### Testing Built `buzz-admin` locally and applied all migrations to a fresh disposable PostgreSQL database. Ran the actual CLI through preview → submit → inspect → approve → `run --next-step` → resume → completed for both demo and a public key with a channel-only membership fixture. Demo completed five stages; the public-key run completed three. Injected `--fail-stage purge_db`, waited for the retry deadline, and resumed the same failed stage successfully. Membership data remained intact. Repeated the CLI workflow on October 26 for demo and an npub backed by a fresh production inventory. Production access used read-only credentials and the real `inventory --pubkey "$SUBJECT"` handler. Copied only the relevant discovery fields into an isolated local PostgreSQL database; local channel parent rows supplied placeholder fields for foreign keys. The inventory saved by local submission exactly matched the production result. All workflow writes stayed local. For each input, ran `submit --pubkey "$SUBJECT" --requested-by local-e2e-submitter`, inspected the saved inventory, and confirmed execution before approval was rejected. Approved its digest, then ran each handler with `run "$REQUEST_ID" --next-step --stage-delay-ms 2500` in a new process. Captured all three workflow tables during and after every stage. Demo completed five stages across two communities; the npub completed three stages for its discovered community. Each handler saved `dummy: true`. Both requests finished with no current stage, no worker lease, and no retries. Copied memberships remained intact. This additional run exercised orderly resume; it did not inject a failure. The reproducible command sequence and fixture setup are in [the inventory walkthrough](docs/operator-user-deletion-inventory.md). Production was not changed. The earlier full-workflow runs used synthetic local fixtures; the additional npub run used captured production discovery fields. Repository-wide `just ci` was not run; validation focused on the affected packages and PostgreSQL workflow. The pre-push desktop build failed with “Too many open files in system.” Stopped the broader hook run and bypassed it to publish the earlier commits; the affected-package Clippy and formatting checks had passed. Generated with Codex --------- Signed-off-by: Ravneet Arora --- crates/buzz-admin/src/main.rs | 2 +- crates/buzz-admin/src/user_deletions.rs | 17 +- .../tests/postgres_user_inventory.rs | 394 ++++++++++++++++++ crates/buzz-db/src/runtime/migration.rs | 4 +- .../src/store/entity_deletion/approvals.rs | 7 +- .../src/store/entity_deletion/leases.rs | 4 +- .../buzz-db/src/store/entity_deletion/mod.rs | 2 +- .../src/store/entity_deletion/requests.rs | 8 +- .../entity_deletion/stages_postgres_tests.rs | 3 +- .../store/entity_deletion_postgres_tests.rs | 2 +- .../src/store/user_deletion/discovery.rs | 165 ++++++++ .../user_deletion/discovery_postgres_tests.rs | 256 ++++++++++++ .../src/store/user_deletion/inventory.rs | 70 +--- .../user_deletion/inventory_postgres_tests.rs | 126 +++++- crates/buzz-db/src/store/user_deletion/mod.rs | 7 + .../src/store/user_deletion/requests.rs | 29 +- crates/buzz-deletion/src/user.rs | 60 ++- crates/buzz-deletion/src/user/inventory.rs | 13 + crates/buzz-deletion/src/user/worker.rs | 8 + crates/buzz-deletion/src/user/workflow.rs | 20 + docs/operator-user-deletion-intake.md | 25 +- docs/operator-user-deletion-inventory.md | 269 ++++++++++++ migrations/0072_user_deletion_target.sql | 10 + .../0073_validate_user_deletion_target.sql | 6 + .../public/entity_deletion_requests.sql | 4 +- 25 files changed, 1405 insertions(+), 106 deletions(-) create mode 100644 crates/buzz-admin/tests/postgres_user_inventory.rs create mode 100644 crates/buzz-db/src/store/user_deletion/discovery.rs create mode 100644 crates/buzz-db/src/store/user_deletion/discovery_postgres_tests.rs create mode 100644 crates/buzz-deletion/src/user/inventory.rs create mode 100644 docs/operator-user-deletion-inventory.md create mode 100644 migrations/0072_user_deletion_target.sql create mode 100644 migrations/0073_validate_user_deletion_target.sql 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: