Add owner deletion admission control plane

ARCHITECTURE.md

@@ -14,6 +14,43 @@ EVENT, REQ, REST, media, git, search, workflow, or pub/sub handling. Unknown
 hosts fail closed, and NIP-98/API-token stamps must agree with the host-derived
 community rather than overriding it.
 
+Deployment-root community management uses operator-signed NIP-98 HTTP requests.
+`POST /operator/communities/delete` accepts only an exact normalized, archived
+community whose asserted pubkey is still its owner. The caller supplies the
+request UUID as the stable correlation/idempotency identity; the durable row
+records owner intent, mediating operator, and acknowledgement version. Admission
+returns `202` at the `submitted` stage and performs no inventory, approval,
+quiescing, object-store access, or deletion execution synchronously. While that
+non-aborted request exists, unarchive and ownership transfer conflict and owner
+management lists suppress the archived row. Replaying the same UUID converges
+to its current stage; a different UUID conflicts with the existing one-active-
+request invariant until that request is aborted.
+
+Owner consent on this path is asserted, not proven. The mediating operator
+authenticates the owner and collects the deletion acknowledgement out of band,
+upstream of the relay; the request itself carries only the operator's NIP-98
+signature. The relay verifies operator authority and that the asserted pubkey
+is still the community's owner, then records the owner pubkey, mediating
+operator pubkey, and acknowledgement version as durable provenance for that
+upstream ceremony. No owner-signed attestation is required or checked, and
+owners have no self-service cancellation. Recovery is a privileged abort,
+which stays open across the reversible `submitted`, `inventoried`, `approved`,
+and `fenced` stages — releasing the request fence while leaving the community
+archived — and closes from `drained` onward, when tenant-state destruction may
+have begun.
+
+Manual operator handoff converges on an admitted owner request only when
+`buzz-admin deletions submit --requested-by` repeats the owner pubkey recorded
+on the row. Owner provenance pins `requested_by` to `owner_pubkey`, so passing
+the operator's own pubkey does not converge — it conflicts with the existing
+one-active-request invariant instead.
+
+Ownership is mutable only while a community is active. Archiving freezes the
+current owner. Normal transfer and deployment-root legacy convergence take the
+same community-row lock as owner-deletion admission, then reject archived,
+quiescing, deleted, or deletion-pending rotation without changing membership.
+Initial owner bootstrap for a newly created community remains supported.
+
 Buzz is a Rust monorepo, licensed Apache 2.0 under Block, Inc.
 
 ---

crates/buzz-db/src/lib.rs

@@ -73,7 +73,7 @@ pub use allowlist::AllowlistEntry;
 pub use api_token::{ApiTokenRecord, TokenSummary};
 pub use community::{
     ArchivedCommunityRecord, CommunityRecord, CreateCommunityWithOwnerResult,
-    CreatedCommunityRecord, EnsuredCommunityRecord, OwnedCommunityRecord,
+    CreatedCommunityRecord, EnsuredCommunityRecord, OwnedCommunityRecord, UnarchiveCommunityResult,
     UnarchivedCommunityRecord,
 };
 pub use error::{DbError, Result};

crates/buzz-db/src/runtime/migration.rs

@@ -705,12 +705,17 @@ mod postgres_tests {
         let mut migrations: Vec<_> = MIGRATOR.iter().collect();
         migrations.sort_by_key(|migration| migration.version);
 
-        assert_eq!(migrations.len(), 50);
+        assert_eq!(migrations.len(), 51);
         assert_eq!(migrations[48].version, 49);
+        assert_eq!(migrations[50].version, 51);
         assert!(migrations[48]
             .sql
             .as_str()
             .contains("idx_thread_metadata_window"));
+        assert!(migrations[50]
+            .sql
+            .as_str()
+            .contains("community_deletion_owner_provenance"));
         assert_eq!(migrations[0].version, 1);
         assert_eq!(&*migrations[0].description, "initial schema");
         assert!(migrations[0]
@@ -1820,6 +1825,12 @@ mod postgres_tests {
             .expect("embedded migration 0029")
             .sql
             .as_ref();
+        let migration_0051: &str = MIGRATOR
+            .iter()
+            .find(|migration| migration.version == 51)
+            .expect("embedded migration 0051")
+            .sql
+            .as_ref();
         let workspace_root = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
             .parent()
             .and_then(std::path::Path::parent)
@@ -1828,6 +1839,7 @@ mod postgres_tests {
             .expect("read schema/schema.sql");
 
         let migration = surface(migration_0029);
+        let owner_admission_migration = surface(migration_0051);
         let schema = surface(&schema_sql);
 
         assert_eq!(
@@ -1856,13 +1868,42 @@ mod postgres_tests {
                 .functions
                 .get(function)
                 .unwrap_or_else(|| panic!("schema.sql is missing deletion function {function}"));
-            if function != "community_write_fence_excluded_table" {
+            if function != "community_write_fence_excluded_table"
+                && function != "prevent_community_deletion_request_retargeting"
+            {
                 assert_eq!(
                     in_schema, definition,
                     "schema.sql definition of {function}() drifted from migration 0029"
                 );
             }
         }
+        assert_eq!(
+            schema
+                .functions
+                .get("prevent_community_deletion_request_retargeting")
+                .expect("schema.sql deletion retargeting guard"),
+            owner_admission_migration
+                .functions
+                .get("prevent_community_deletion_request_retargeting")
+                .expect("0051 deletion retargeting guard"),
+            "schema.sql must carry the latest immutable owner-provenance guard"
+        );
+        let request_table = schema
+            .tables
+            .get("community_deletion_requests")
+            .expect("schema.sql deletion request table");
+        for owner_provenance_fragment in [
+            "request_origin text not null default 'operator'",
+            "owner_pubkey text",
+            "mediating_operator_pubkey text",
+            "acknowledgement_version integer",
+            "constraint community_deletion_owner_provenance check",
+        ] {
+            assert!(
+                request_table.contains(owner_provenance_fragment),
+                "schema.sql deletion requests are missing {owner_provenance_fragment}"
+            );
+        }
         for (trigger, definition) in &migration.triggers {
             let in_schema = schema
                 .triggers
@@ -2512,6 +2553,22 @@ mod postgres_tests {
         assert_eq!(after, vec![(1, Some(true)), (30_179, None), (30_350, None)]);
     }
 
+    /// Migration-upgrade half of the owner-provenance contract.
+    ///
+    /// The desired-state bootstrap half lives in
+    /// `store::deletion::postgres_tests` and asserts the same shared case
+    /// table, so `schema/schema.sql` cannot admit owner rows the migration
+    /// path refuses (or the reverse).
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn migrated_schema_enforces_owner_provenance_contract() {
+        let pool = connect_test_pool().await;
+        reset_public_schema(&pool).await;
+        run_migrations(&pool).await.expect("run migrations");
+
+        crate::store::deletion::owner_provenance_contract::assert_contract(&pool).await;
+    }
+
     #[tokio::test]
     #[ignore = "requires Postgres"]
     async fn run_migrations_applies_consolidated_initial_schema_on_fresh_database() {

crates/buzz-db/src/runtime/tests/thread_window_postgres_tests.rs

@@ -311,7 +311,7 @@ async fn migration_schema_thread_window_prebuild_does_not_queue_behind_writer()
         production_result.is_ok(),
         "production migrator must preserve ingestion progress: {production_result:?}"
     );
-    assert_eq!(version, 50);
+    assert_eq!(version, 51);
     assert_eq!(final_oid, oid, "prebuild must not be replaced");
     assert_eq!(count, 4, "all writer witnesses must persist");
 }

crates/buzz-db/src/store/community.rs

@@ -81,6 +81,17 @@ pub struct UnarchivedCommunityRecord {
     pub host: String,
 }
 
+/// Result of an owner-authorized unarchive attempt.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum UnarchiveCommunityResult {
+    /// The community is active, with archive state cleared idempotently.
+    Unarchived(UnarchivedCommunityRecord),
+    /// Durable deletion intent exists and wins over restoration.
+    DeletionPending,
+    /// The host is absent, unavailable, or not owned by the asserted pubkey.
+    NotFound,
+}
+
 impl Db {
     /// Returns the community mapped to a normalized request host, if one exists.
     ///
@@ -209,6 +220,10 @@ impl Db {
             JOIN relay_members rm ON rm.community_id = c.id
             WHERE rm.pubkey = $1
               AND rm.role = 'owner'
+              AND NOT EXISTS (
+                  SELECT 1 FROM community_deletion_requests request
+                  WHERE request.community_id = c.id AND request.stage <> 'aborted'
+              )
             ORDER BY c.created_at ASC, c.host ASC
             "#,
         )
@@ -531,40 +546,69 @@ impl Db {
     }
 
     /// Idempotently restores a community when the asserted pubkey is its current owner.
+    ///
+    /// Locks the community row so owner-deletion admission and restoration have
+    /// one serial order. A non-aborted deletion request returns
+    /// [`UnarchiveCommunityResult::DeletionPending`] without clearing archive state.
     #[datastore_span(name = "unarchive_community_owned_by", system = "postgresql")]
     pub async fn unarchive_community_owned_by(
         &self,
         normalized_host: &str,
         owner_pubkey: &str,
-    ) -> Result<Option<UnarchivedCommunityRecord>> {
-        let mut connection = crate::observability::acquire_writer(
+    ) -> Result<UnarchiveCommunityResult> {
+        let connection = crate::observability::acquire_writer(
             &self.pool,
             crate::observability::WriterOperation::Authorization,
         )
         .await?;
-        let row = sqlx::query(
-            r#"UPDATE communities c
-               SET archived_at = NULL
-               FROM relay_members rm
-               WHERE lower(c.host) = lower($1)
-                 AND rm.community_id = c.id
-                 AND lower(rm.pubkey) = lower($2)
-                 AND rm.role = 'owner'
-                 AND c.deletion_state = 'active'
-                 AND c.deleted_at IS NULL
-               RETURNING c.id, c.host"#,
+        let mut tx = sqlx::Transaction::begin(connection, None).await?;
+        let target = sqlx::query(
+            "SELECT id, host FROM communities \
+             WHERE lower(host) = lower($1) AND deletion_state = 'active' \
+               AND deleted_at IS NULL FOR UPDATE",
         )
         .bind(normalized_host)
+        .fetch_optional(&mut *tx)
+        .await?;
+        let Some(target) = target else {
+            tx.rollback().await?;
+            return Ok(UnarchiveCommunityResult::NotFound);
+        };
+        let community_id: Uuid = target.try_get("id")?;
+        let is_owner: bool = sqlx::query_scalar(
+            "SELECT EXISTS(SELECT 1 FROM relay_members \
+             WHERE community_id = $1 AND lower(pubkey) = lower($2) AND role = 'owner')",
+        )
+        .bind(community_id)
         .bind(owner_pubkey)
-        .fetch_optional(&mut *connection)
+        .fetch_one(&mut *tx)
         .await?;
-        row.map(|row| {
-            Ok(UnarchivedCommunityRecord {
-                id: CommunityId::from_uuid(row.try_get("id")?),
-                host: row.try_get("host")?,
-            })
-        })
-        .transpose()
+        if !is_owner {
+            tx.rollback().await?;
+            return Ok(UnarchiveCommunityResult::NotFound);
+        }
+        let deletion_pending: bool = sqlx::query_scalar(
+            "SELECT EXISTS(SELECT 1 FROM community_deletion_requests \
+             WHERE community_id = $1 AND stage <> 'aborted')",
+        )
+        .bind(community_id)
+        .fetch_one(&mut *tx)
+        .await?;
+        if deletion_pending {
+            tx.rollback().await?;
+            return Ok(UnarchiveCommunityResult::DeletionPending);
+        }
+        sqlx::query("UPDATE communities SET archived_at = NULL WHERE id = $1")
+            .bind(community_id)
+            .execute(&mut *tx)
+            .await?;
+        tx.commit().await?;
+        Ok(UnarchiveCommunityResult::Unarchived(
+            UnarchivedCommunityRecord {
+                id: CommunityId::from_uuid(community_id),
+                host: target.try_get("host")?,
+            },
+        ))
     }
 
     /// Returns the community that owns a channel, if the channel exists.
@@ -900,22 +944,26 @@ mod postgres_tests {
                 .is_none(),
             "archived communities must fail admission"
         );
-        assert!(db
-            .unarchive_community_owned_by(&host, &outsider)
-            .await
-            .expect("wrong-owner unarchive")
-            .is_none());
-        assert!(db
-            .unarchive_community_owned_by("missing.example", &owner)
-            .await
-            .expect("unknown-host unarchive")
-            .is_none());
+        assert_eq!(
+            db.unarchive_community_owned_by(&host, &outsider)
+                .await
+                .expect("wrong-owner unarchive"),
+            UnarchiveCommunityResult::NotFound
+        );
+        assert_eq!(
+            db.unarchive_community_owned_by("missing.example", &owner)
+                .await
+                .expect("unknown-host unarchive"),
+            UnarchiveCommunityResult::NotFound
+        );
 
         let restored = db
             .unarchive_community_owned_by(&host.to_ascii_uppercase(), &owner)
             .await
-            .expect("unarchive community")
-            .expect("owned community");
+            .expect("unarchive community");
+        let UnarchiveCommunityResult::Unarchived(restored) = restored else {
+            panic!("expected owned community")
+        };
         assert_eq!(restored.id, created.id);
         assert_eq!(restored.host, host);
         assert_eq!(
@@ -938,9 +986,8 @@ mod postgres_tests {
         let retry = db
             .unarchive_community_owned_by(&host, &owner)
             .await
-            .expect("idempotent retry")
-            .expect("owned community");
-        assert_eq!(retry, restored);
+            .expect("idempotent retry");
+        assert_eq!(retry, UnarchiveCommunityResult::Unarchived(restored));
     }
 
     #[tokio::test]