feat(relay): add durable community archival

crates/buzz-db/src/channel.rs

@@ -1345,6 +1345,7 @@ pub async fn reap_expired_ephemeral_channels(pool: &PgPool) -> Result<Vec<Reaped
            AND ch.ttl_deadline < NOW() \
            AND ch.archived_at IS NULL \
            AND ch.deleted_at IS NULL \
+           AND c.archived_at IS NULL \
          RETURNING ch.community_id, c.host, ch.id",
     )
     .fetch_all(pool)

crates/buzz-db/src/event.rs

@@ -1306,6 +1306,7 @@ pub async fn query_due_reminders(
           AND e.not_before <= $2
           AND e.deleted_at IS NULL
           AND e.delivered_at IS NULL
+          AND c.archived_at IS NULL
         ORDER BY e.community_id, e.pubkey, e.d_tag, e.created_at DESC, e.id ASC
         LIMIT $3
         "#,

crates/buzz-db/src/lib.rs

@@ -261,6 +261,19 @@ pub struct OwnedCommunityRecord {
     pub host: String,
     /// When the community row was created.
     pub created_at: DateTime<Utc>,
+    /// When the community was archived; absent while active.
+    pub archived_at: Option<DateTime<Utc>>,
+}
+
+/// Community row returned by an owner-authorized archive operation.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ArchivedCommunityRecord {
+    /// Stable server-resolved community id.
+    pub id: CommunityId,
+    /// Reserved canonical host.
+    pub host: String,
+    /// Durable first-archive timestamp.
+    pub archived_at: DateTime<Utc>,
 }
 
 /// Token summary returned by [`Db::list_active_tokens`].
@@ -407,6 +420,7 @@ impl Db {
             SELECT id, host
             FROM communities
             WHERE lower(host) = lower($1)
+              AND archived_at IS NULL
             "#,
         )
         .bind(normalized_host)
@@ -425,6 +439,35 @@ impl Db {
         .transpose()
     }
 
+    /// Returns whether a community id still exists in the active lifecycle state.
+    pub async fn is_community_active(&self, community_id: CommunityId) -> Result<bool> {
+        let active = sqlx::query_scalar::<_, bool>(
+            "SELECT EXISTS(SELECT 1 FROM communities WHERE id = $1 AND archived_at IS NULL)",
+        )
+        .bind(community_id.as_uuid())
+        .fetch_one(&self.pool)
+        .await?;
+        Ok(active)
+    }
+
+    /// Returns a community by host regardless of lifecycle state. Operator-plane only.
+    pub async fn lookup_community_by_host_for_management(
+        &self,
+        normalized_host: &str,
+    ) -> Result<Option<CommunityRecord>> {
+        let row = sqlx::query("SELECT id, host FROM communities WHERE lower(host) = lower($1)")
+            .bind(normalized_host)
+            .fetch_optional(&self.pool)
+            .await?;
+        row.map(|row| {
+            Ok(CommunityRecord {
+                id: CommunityId::from_uuid(row.try_get("id")?),
+                host: row.try_get("host")?,
+            })
+        })
+        .transpose()
+    }
+
     /// Lists communities where `owner_pubkey` currently holds the `owner` role.
     ///
     /// This is an operator-plane helper, not a tenant-scoped data-plane read:
@@ -436,7 +479,7 @@ impl Db {
         let owner_pubkey = owner_pubkey.to_ascii_lowercase();
         let rows = sqlx::query(
             r#"
-            SELECT c.id, c.host, c.created_at
+            SELECT c.id, c.host, c.created_at, c.archived_at
             FROM communities c
             JOIN relay_members rm ON rm.community_id = c.id
             WHERE rm.pubkey = $1
@@ -453,10 +496,12 @@ impl Db {
                 let id: Uuid = row.try_get("id")?;
                 let host: String = row.try_get("host")?;
                 let created_at: DateTime<Utc> = row.try_get("created_at")?;
+                let archived_at: Option<DateTime<Utc>> = row.try_get("archived_at")?;
                 Ok(OwnedCommunityRecord {
                     id: CommunityId::from_uuid(id),
                     host,
                     created_at,
+                    archived_at,
                 })
             })
             .collect()
@@ -478,6 +523,7 @@ impl Db {
             SELECT host
             FROM communities
             WHERE id = $1
+              AND archived_at IS NULL
             "#,
         )
         .bind(community_id.as_uuid())
@@ -632,6 +678,7 @@ impl Db {
                 WHERE lower(c.host) = lower($1)
                   AND lower(rm.pubkey) = lower($2)
                   AND rm.role = 'owner'
+                  AND c.archived_at IS NULL
                 "#,
             )
             .bind(normalized_host)
@@ -654,6 +701,39 @@ impl Db {
         ))
     }
 
+    /// Idempotently archives a community when the asserted pubkey is its current owner.
+    pub async fn archive_community_owned_by(
+        &self,
+        normalized_host: &str,
+        owner_pubkey: &str,
+        protected_deployment_host: &str,
+    ) -> Result<Option<ArchivedCommunityRecord>> {
+        let row = sqlx::query(
+            r#"UPDATE communities c
+               SET archived_at = COALESCE(c.archived_at, now())
+               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 lower(c.host) <> lower($3)
+               RETURNING c.id, c.host, c.archived_at"#,
+        )
+        .bind(normalized_host)
+        .bind(owner_pubkey)
+        .bind(protected_deployment_host)
+        .fetch_optional(&self.pool)
+        .await?;
+        row.map(|row| {
+            Ok(ArchivedCommunityRecord {
+                id: CommunityId::from_uuid(row.try_get("id")?),
+                host: row.try_get("host")?,
+                archived_at: row.try_get("archived_at")?,
+            })
+        })
+        .transpose()
+    }
+
     /// Returns the community that owns a channel, if the channel exists.
     ///
     /// Internal relay producers use this to derive tenant context from the row
@@ -4128,8 +4208,13 @@ mod tests {
         let community_a = CommunityId::from_uuid(make_community(&db.pool).await);
         let community_b = CommunityId::from_uuid(make_community(&db.pool).await);
         let community_c = CommunityId::from_uuid(make_community(&db.pool).await);
-        let owner = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
-        let other = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb";
+        // Unique per run: `list_communities_owned_by` is keyed only by pubkey,
+        // so a shared fixed pubkey picks up communities leaked by sibling
+        // ignored tests running against the same database.
+        let owner = format!("{}{}", Uuid::new_v4().simple(), Uuid::new_v4().simple());
+        let owner = owner.as_str();
+        let other = format!("{}{}", Uuid::new_v4().simple(), Uuid::new_v4().simple());
+        let other = other.as_str();
 
         db.bootstrap_owner(community_a, owner)
             .await

crates/buzz-db/src/migration.rs

@@ -548,7 +548,7 @@ mod tests {
         let mut migrations: Vec<_> = MIGRATOR.iter().collect();
         migrations.sort_by_key(|migration| migration.version);
 
-        assert_eq!(migrations.len(), 15);
+        assert_eq!(migrations.len(), 16);
         assert_eq!(migrations[0].version, 1);
         assert_eq!(&*migrations[0].description, "initial schema");
         assert!(migrations[0]
@@ -1005,7 +1005,7 @@ mod tests {
         run_migrations(&pool)
             .await
             .expect("retry succeeds after operator repair");
-        assert_eq!(applied_versions(&pool).await.last().copied(), Some(15));
+        assert_eq!(applied_versions(&pool).await.last().copied(), Some(16));
     }
 
     #[tokio::test]

crates/buzz-db/src/workflow.rs

@@ -457,13 +457,15 @@ pub async fn list_enabled_channel_workflows(
 pub async fn list_all_enabled_workflows(pool: &PgPool) -> Result<Vec<WorkflowRecord>> {
     let rows = sqlx::query(
         r#"
-        SELECT id, community_id, name, owner_pubkey, channel_id, definition, definition_hash,
-               status::text AS status, enabled, created_at, updated_at
-        FROM workflows
-        WHERE status = 'active'
-          AND enabled = TRUE
-          AND definition->'trigger'->>'on' = 'schedule'
-        ORDER BY created_at ASC
+        SELECT w.id, w.community_id, w.name, w.owner_pubkey, w.channel_id, w.definition, w.definition_hash,
+               w.status::text AS status, w.enabled, w.created_at, w.updated_at
+        FROM workflows w
+        JOIN communities c ON c.id = w.community_id
+        WHERE w.status = 'active'
+          AND w.enabled = TRUE
+          AND w.definition->'trigger'->>'on' = 'schedule'
+          AND c.archived_at IS NULL
+        ORDER BY w.created_at ASC
         LIMIT $1
         "#,
     )

crates/buzz-pubsub/src/conn_control.rs

@@ -54,6 +54,8 @@ pub fn parse_conn_control_channel(channel: &str) -> Option<CommunityId> {
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
 #[serde(tag = "op")]
 pub enum ConnControl {
+    /// Disconnect every live socket bound to the carrying community.
+    DisconnectCommunity,
     /// Disconnect every live connection authenticated as `pubkey` in the
     /// carrying community — live ban enforcement. `pubkey` is 32 raw bytes.
     /// `event_id` and `reason` reproduce the same NIP-01 `OK` frame the origin
@@ -197,6 +199,23 @@ mod tests {
         assert_eq!(parse_conn_control_channel(&extended), None);
     }
 
+    #[test]
+    fn disconnect_community_command_serde_round_trips() {
+        let cmd = ConnControl::DisconnectCommunity;
+        let json = serde_json::to_string(&cmd).unwrap();
+        assert_eq!(serde_json::from_str::<ConnControl>(&json).unwrap(), cmd);
+    }
+
+    #[test]
+    fn unknown_command_is_rejected_without_affecting_later_messages() {
+        assert!(serde_json::from_str::<ConnControl>(r#"{"op":"FutureCommand"}"#).is_err());
+        let known = serde_json::to_string(&ConnControl::DisconnectCommunity).unwrap();
+        assert_eq!(
+            serde_json::from_str::<ConnControl>(&known).unwrap(),
+            ConnControl::DisconnectCommunity
+        );
+    }
+
     #[test]
     fn disconnect_command_serde_round_trips() {
         let cmd = ConnControl::DisconnectPubkey {