feat(buzz-relay): idempotent owner community deletion with quota reservation

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

@@ -705,12 +705,13 @@ mod postgres_tests {
         let mut migrations: Vec<_> = MIGRATOR.iter().collect();
         migrations.sort_by_key(|migration| migration.version);
 
-        assert_eq!(migrations.len(), 53);
+        assert_eq!(migrations.len(), 54);
         assert_eq!(migrations[48].version, 49);
         assert_eq!(migrations[49].version, 50);
         assert_eq!(migrations[50].version, 51);
         assert_eq!(migrations[51].version, 52);
         assert_eq!(migrations[52].version, 53);
+        assert_eq!(migrations[53].version, 54);
         assert!(migrations[48]
             .sql
             .as_str()
@@ -724,6 +725,10 @@ mod postgres_tests {
             .sql
             .as_str()
             .contains("community_deletion_requests_owner_preparable"));
+        assert!(migrations[53]
+            .sql
+            .as_str()
+            .contains("community_deletion_requests_owner_quota_reservations"));
         assert_eq!(migrations[0].version, 1);
         assert_eq!(&*migrations[0].description, "initial schema");
         assert!(migrations[0]
@@ -1758,6 +1763,32 @@ mod postgres_tests {
         assert!(migration.contains("set local lock_timeout = '5s'"));
     }
 
+    #[test]
+    fn owner_deletion_quota_reservation_index_matches_desired_schema() {
+        let migration = MIGRATOR
+            .iter()
+            .find(|migration| migration.version == 54)
+            .expect("embedded migration 0054")
+            .sql
+            .as_ref()
+            .to_ascii_lowercase();
+        let workspace_root = std::path::Path::new(env!("CARGO_MANIFEST_DIR"))
+            .parent()
+            .and_then(std::path::Path::parent)
+            .expect("workspace root");
+        let schema = std::fs::read_to_string(workspace_root.join("schema/schema.sql"))
+            .expect("read schema/schema.sql")
+            .to_ascii_lowercase();
+
+        for sql in [&migration, &schema] {
+            assert!(sql.contains("community_deletion_requests_owner_quota_reservations"));
+            assert!(sql.contains("request_origin = 'owner'"));
+            assert!(sql.contains("stage <> 'aborted'"));
+            assert!(sql.contains("include (community_id, completed_at)"));
+        }
+        assert!(migration.contains("set local lock_timeout = '5s'"));
+    }
+
     /// Structural parity between migration 0029's deletion surface and the
     /// desired-state bootstrap schema (`schema/schema.sql`).
     ///

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, 53);
+    assert_eq!(version, 54);
     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

@@ -44,7 +44,7 @@ pub enum CreateCommunityWithOwnerResult {
     Created(CreatedCommunityRecord),
     /// The host already belongs to another owner.
     HostExists,
-    /// The intended owner already owns the maximum number of communities.
+    /// The intended owner has reached the active or lifetime community limit.
     LimitReached,
 }
 
@@ -61,6 +61,19 @@ pub struct OwnedCommunityRecord {
     pub archived_at: Option<DateTime<Utc>>,
 }
 
+/// Owner-list rows plus the authoritative quota projection from one snapshot.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct OwnedCommunitiesPage {
+    /// Visible live owner memberships, excluding communities already under deletion.
+    pub communities: Vec<OwnedCommunityRecord>,
+    /// De-duplicated live memberships and incomplete owner deletion reservations.
+    pub quota_used: i64,
+    /// Configured active limit only; the lifetime cap is reflected in `can_create`.
+    pub quota_limit: i64,
+    /// Whether the snapshot leaves room under both the active and lifetime caps.
+    pub can_create: bool,
+}
+
 /// Community row returned by an owner-authorized archive operation.
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct ArchivedCommunityRecord {
@@ -206,13 +219,17 @@ impl Db {
     pub async fn list_communities_owned_by(
         &self,
         owner_pubkey: &str,
-    ) -> Result<Vec<OwnedCommunityRecord>> {
+    ) -> Result<OwnedCommunitiesPage> {
         let owner_pubkey = owner_pubkey.to_ascii_lowercase();
-        let mut connection = crate::observability::acquire_writer(
+        let connection = crate::observability::acquire_writer(
             &self.pool,
             crate::observability::WriterOperation::Authorization,
         )
         .await?;
+        let mut tx = sqlx::Transaction::begin(connection, None).await?;
+        sqlx::query("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ, READ ONLY")
+            .execute(&mut *tx)
+            .await?;
         let rows = sqlx::query(
             r#"
             SELECT c.id, c.host, c.created_at, c.archived_at
@@ -227,11 +244,11 @@ impl Db {
             ORDER BY c.created_at ASC, c.host ASC
             "#,
         )
-        .bind(owner_pubkey)
-        .fetch_all(&mut *connection)
+        .bind(&owner_pubkey)
+        .fetch_all(&mut *tx)
         .await?;
-
-        rows.into_iter()
+        let communities = rows
+            .into_iter()
             .map(|row| {
                 let id: Uuid = row.try_get("id")?;
                 let host: String = row.try_get("host")?;
@@ -244,7 +261,15 @@ impl Db {
                     archived_at,
                 })
             })
-            .collect()
+            .collect::<Result<Vec<_>>>()?;
+        let quota = relay_members::owner_quota_in_transaction(&mut tx, &owner_pubkey).await?;
+        tx.commit().await?;
+        Ok(OwnedCommunitiesPage {
+            communities,
+            quota_used: quota.active,
+            quota_limit: relay_members::max_communities_per_owner(),
+            can_create: quota.admits(),
+        })
     }
 
     /// Returns the normalized host mapped to a community id, if the community
@@ -450,14 +475,10 @@ impl Db {
             let host: String = row.try_get("host")?;
 
             // Enforce the limit before inserting the new owner row.
-            let owned_count: i64 = sqlx::query_scalar(
-                "SELECT count(*) FROM relay_members WHERE pubkey = $1 AND role = 'owner'",
-            )
-            .bind(&owner_pubkey)
-            .fetch_one(&mut *tx)
-            .await?;
-
-            if owned_count >= relay_members::max_communities_per_owner() {
+            if !relay_members::owner_quota_in_transaction(&mut tx, &owner_pubkey)
+                .await?
+                .admits()
+            {
                 tx.rollback().await?;
                 return Ok(CreateCommunityWithOwnerResult::LimitReached);
             }
@@ -793,6 +814,7 @@ mod postgres_tests {
             "EnsuredCommunityRecord",
             "CreatedCommunityRecord",
             "OwnedCommunityRecord",
+            "OwnedCommunitiesPage",
             "ArchivedCommunityRecord",
             "UnarchivedCommunityRecord",
         ];
@@ -1093,8 +1115,168 @@ mod postgres_tests {
             .await
             .expect("list owned communities");
 
-        assert_eq!(owned.len(), 1);
-        assert_eq!(owned[0].id, community_a);
+        assert_eq!(owned.communities.len(), 1);
+        assert_eq!(owned.communities[0].id, community_a);
+        assert_eq!(owned.quota_used, 1);
+        assert_eq!(
+            owned.quota_limit,
+            crate::relay_members::max_communities_per_owner()
+        );
+        assert!(owned.can_create);
+    }
+
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn owner_quota_serializes_concurrent_create_and_transfer_with_reservation() {
+        let db = setup_db().await;
+        let recipient = format!("{}{}", Uuid::new_v4().simple(), Uuid::new_v4().simple());
+        let source_owner = format!("{}{}", Uuid::new_v4().simple(), Uuid::new_v4().simple());
+
+        for index in 0..3 {
+            let host = format!("quota-live-{index}-{}.example", Uuid::new_v4().simple());
+            assert!(matches!(
+                db.create_community_with_owner(&host, &recipient)
+                    .await
+                    .expect("create live quota fixture"),
+                CreateCommunityWithOwnerResult::Created(_)
+            ));
+        }
+        let reserved_host = format!("quota-held-{}.example", Uuid::new_v4().simple());
+        let reserved = db
+            .create_community_with_owner(&reserved_host, &recipient)
+            .await
+            .expect("create reservation fixture");
+        let CreateCommunityWithOwnerResult::Created(reserved) = reserved else {
+            panic!("expected reservation community")
+        };
+        sqlx::query(
+            "INSERT INTO community_deletion_requests \
+             (id, community_id, community_host, requested_by, request_origin, owner_pubkey, \
+              mediating_operator_pubkey, acknowledgement_version) \
+             VALUES ($1, $2, $3, $4, 'owner', $4, $5, 1)",
+        )
+        .bind(Uuid::new_v4())
+        .bind(reserved.id.as_uuid())
+        .bind(&reserved_host)
+        .bind(&recipient)
+        .bind("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb")
+        .execute(&db.pool)
+        .await
+        .expect("insert reservation");
+        sqlx::query("DELETE FROM relay_members WHERE community_id = $1")
+            .bind(reserved.id.as_uuid())
+            .execute(&db.pool)
+            .await
+            .expect("simulate purged membership");
+
+        let transfer_host = format!("quota-transfer-{}.example", Uuid::new_v4().simple());
+        let transfer_target = db
+            .create_community_with_owner(&transfer_host, &source_owner)
+            .await
+            .expect("create transfer target");
+        let CreateCommunityWithOwnerResult::Created(transfer_target) = transfer_target else {
+            panic!("expected transfer target")
+        };
+        let create_host = format!("quota-race-{}.example", Uuid::new_v4().simple());
+
+        let (create, transfer) = tokio::join!(
+            db.create_community_with_owner(&create_host, &recipient),
+            db.transfer_ownership(transfer_target.id, &recipient, &source_owner),
+        );
+        let create = create.expect("concurrent create result");
+        let transfer = transfer.expect("concurrent transfer result");
+        let create_won = matches!(create, CreateCommunityWithOwnerResult::Created(_));
+        let transfer_won = matches!(
+            transfer,
+            crate::relay_members::TransferResult::Transferred { .. }
+        );
+        assert_ne!(create_won, transfer_won, "exactly one owner grant may win");
+        assert!(
+            matches!(create, CreateCommunityWithOwnerResult::LimitReached) || create_won,
+            "create loser must observe the quota"
+        );
+        assert!(
+            matches!(transfer, crate::relay_members::TransferResult::LimitReached) || transfer_won,
+            "transfer loser must observe the quota"
+        );
+        assert_eq!(
+            db.list_communities_owned_by(&recipient)
+                .await
+                .expect("post-race quota")
+                .quota_used,
+            crate::relay_members::max_communities_per_owner()
+        );
+    }
+
+    #[tokio::test]
+    #[ignore = "requires Postgres"]
+    async fn completed_owner_deletions_count_toward_lifetime_cap() {
+        let db = setup_db().await;
+        let owner = format!("{:064x}", Uuid::new_v4().as_u128());
+        let other_owner = format!("{:064x}", Uuid::new_v4().as_u128());
+        let operator = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb";
+
+        // Create and completely delete up to the lifetime cap: each tombstone
+        // frees its active slot but keeps its host and its lifetime count.
+        for i in 0..crate::relay_members::MAX_LIFETIME_COMMUNITIES_PER_OWNER {
+            let host = format!("lifetime-{i}-{}.example", Uuid::new_v4().simple());
+            let CreateCommunityWithOwnerResult::Created(record) = db
+                .create_community_with_owner(&host, &owner)
+                .await
+                .expect("create under lifetime cap")
+            else {
+                panic!("create {i} must succeed below the lifetime cap")
+            };
+            sqlx::query(
+                "INSERT INTO community_deletion_requests \
+                 (id, community_id, community_host, requested_by, request_origin, owner_pubkey, \
+                  mediating_operator_pubkey, acknowledgement_version, stage, completed_at) \
+                 VALUES ($1, $2, $3, $4, 'owner', $4, $5, 1, 'retention_pending', now())",
+            )
+            .bind(Uuid::new_v4())
+            .bind(record.id.as_uuid())
+            .bind(&host)
+            .bind(&owner)
+            .bind(operator)
+            .execute(&db.pool)
+            .await
+            .expect("insert completed owner deletion");
+            sqlx::query("DELETE FROM relay_members WHERE community_id = $1")
+                .bind(record.id.as_uuid())
+                .execute(&db.pool)
+                .await
+                .expect("simulate purged membership");
+        }
+
+        let page = db
+            .list_communities_owned_by(&owner)
+            .await
+            .expect("owner list at lifetime cap");
+        assert_eq!(page.quota_used, 0, "completed deletions free active slots");
+        assert!(!page.can_create, "the lifetime cap still blocks creation");
+
+        let host = format!("lifetime-overflow-{}.example", Uuid::new_v4().simple());
+        assert_eq!(
+            db.create_community_with_owner(&host, &owner)
+                .await
+                .expect("create past lifetime cap"),
+            CreateCommunityWithOwnerResult::LimitReached
+        );
+
+        let transfer_host = format!("lifetime-transfer-{}.example", Uuid::new_v4().simple());
+        let CreateCommunityWithOwnerResult::Created(target) = db
+            .create_community_with_owner(&transfer_host, &other_owner)
+            .await
+            .expect("create transfer target")
+        else {
+            panic!("expected transfer target")
+        };
+        assert_eq!(
+            db.transfer_ownership(target.id, &owner, &other_owner)
+                .await
+                .expect("transfer past lifetime cap"),
+            crate::relay_members::TransferResult::LimitReached
+        );
     }
 
     #[tokio::test]