fix: recover community access visibility

crates/buzz-core/src/kind.rs

@@ -677,12 +677,13 @@ pub const fn is_command_kind(kind: u32) -> bool {
     )
 }
 
-/// Returns `true` if `kind` is a relay-only sidecar kind.
+/// Returns `true` if `kind` may only be authored by the relay.
 /// Client submission of these kinds must be rejected.
 pub const fn is_relay_only_kind(kind: u32) -> bool {
     matches!(
         kind,
-        KIND_CHANNEL_SUMMARY
+        KIND_NIP43_MEMBERSHIP_LIST
+            | KIND_CHANNEL_SUMMARY
             | KIND_PRESENCE_SNAPSHOT
             | KIND_DM_VISIBILITY
             | KIND_THREAD_SUMMARY
@@ -755,6 +756,12 @@ mod tests {
         }
     }
 
+    #[test]
+    fn nip43_membership_snapshot_is_relay_only() {
+        assert!(is_relay_only_kind(KIND_NIP43_MEMBERSHIP_LIST));
+        assert!(!is_relay_only_kind(KIND_NIP43_LEAVE_REQUEST));
+    }
+
     #[test]
     fn parameterized_replaceable_range() {
         assert!(!is_parameterized_replaceable(29999));

crates/buzz-db/src/lib.rs

@@ -3232,6 +3232,53 @@ impl Db {
         ))
     }
 
+    /// Returns whether the relay-authored NIP-43 snapshot is absent or differs
+    /// from the canonical membership rows for `community_id`.
+    ///
+    /// Snapshot and canonical rows are compared directly rather than by
+    /// timestamp: relay membership events use whole-second Nostr timestamps,
+    /// and multiple mutations within one second must still be repaired.
+    pub async fn nip43_membership_snapshot_needs_reconciliation(
+        &self,
+        community_id: CommunityId,
+        relay_pubkey: &nostr::PublicKey,
+    ) -> Result<bool> {
+        let snapshot = self
+            .query_events(&crate::event::EventQuery {
+                kinds: Some(vec![buzz_core::kind::KIND_NIP43_MEMBERSHIP_LIST as i32]),
+                pubkey: Some(relay_pubkey.to_bytes().to_vec()),
+                global_only: true,
+                limit: Some(1),
+                ..crate::event::EventQuery::for_community(community_id)
+            })
+            .await?
+            .into_iter()
+            .next();
+        let members = self.list_relay_members(community_id).await?;
+
+        let Some(snapshot) = snapshot else {
+            return Ok(true);
+        };
+        let mut snapshot_members = snapshot
+            .event
+            .tags
+            .iter()
+            .filter_map(|tag| {
+                let parts = tag.as_slice();
+                (parts.first().map(String::as_str) == Some("member") && parts.len() >= 3)
+                    .then(|| (parts[1].to_ascii_lowercase(), parts[2].clone()))
+            })
+            .collect::<Vec<_>>();
+        let mut canonical_members = members
+            .into_iter()
+            .map(|member| (member.pubkey.to_ascii_lowercase(), member.role))
+            .collect::<Vec<_>>();
+        snapshot_members.sort_unstable();
+        canonical_members.sort_unstable();
+
+        Ok(snapshot_members != canonical_members)
+    }
+
     /// Atomically publish a NIP-43 membership snapshot under a single
     /// transaction-scoped advisory lock.
     ///

crates/buzz-relay/src/connection.rs

@@ -541,7 +541,8 @@ async fn handle_text_message(text: String, conn: Arc<ConnectionState>, state: Ar
             let permit = match state.handler_semaphore.clone().try_acquire_owned() {
                 Ok(p) => p,
                 Err(_) => {
-                    conn.send(RelayMessage::notice(
+                    conn.send(request_rejection_message(
+                        Some(&sub_id),
                         "rate-limited: too many concurrent requests",
                     ));
                     return;
@@ -583,6 +584,13 @@ async fn handle_text_message(text: String, conn: Arc<ConnectionState>, state: Ar
     }
 }
 
+fn request_rejection_message(sub_id: Option<&str>, reason: &str) -> String {
+    match sub_id {
+        Some(sub_id) => RelayMessage::closed(sub_id, reason),
+        None => RelayMessage::notice(reason),
+    }
+}
+
 async fn enforce_ws_admission(
     msg: &ClientMessage,
     conn: &ConnectionState,
@@ -613,7 +621,11 @@ async fn enforce_ws_admission(
         ws_limit,
     )
     .await;
-    if !send_admission_result(conn, ws_result) {
+    let sub_id = match msg {
+        ClientMessage::Req { sub_id, .. } => Some(sub_id.as_str()),
+        _ => None,
+    };
+    if !send_admission_result(conn, ws_result, sub_id) {
         return false;
     }
 
@@ -632,7 +644,7 @@ async fn enforce_ws_admission(
             message_limit,
         )
         .await;
-        if !send_admission_result(conn, message_result) {
+        if !send_admission_result(conn, message_result, None) {
             return false;
         }
     }
@@ -643,19 +655,22 @@ async fn enforce_ws_admission(
 fn send_admission_result(
     conn: &ConnectionState,
     result: Result<(), crate::admission::AdmissionError>,
+    sub_id: Option<&str>,
 ) -> bool {
     match result {
         Ok(()) => true,
         Err(crate::admission::AdmissionError::Exceeded { reset_in_secs }) => {
             metrics::counter!("buzz_admission_rejections_total", "transport" => "websocket", "reason" => "quota").increment(1);
-            conn.send(RelayMessage::notice(&format!(
-                "rate-limited: quota exceeded; retry in {reset_in_secs}s"
-            )));
+            conn.send(request_rejection_message(
+                sub_id,
+                &format!("rate-limited: quota exceeded; retry in {reset_in_secs}s"),
+            ));
             false
         }
         Err(crate::admission::AdmissionError::Unavailable) => {
             metrics::counter!("buzz_admission_rejections_total", "transport" => "websocket", "reason" => "unavailable").increment(1);
-            conn.send(RelayMessage::notice(
+            conn.send(request_rejection_message(
+                sub_id,
                 "rate-limited: shared admission unavailable",
             ));
             false
@@ -757,6 +772,19 @@ mod tests {
             .collect()
     }
 
+    #[test]
+    fn req_rejections_are_subscription_scoped() {
+        let reason = "rate-limited: too many concurrent requests";
+        let closed: serde_json::Value =
+            serde_json::from_str(&request_rejection_message(Some("history-123"), reason))
+                .expect("parse CLOSED");
+        assert_eq!(closed, serde_json::json!(["CLOSED", "history-123", reason]));
+
+        let notice: serde_json::Value =
+            serde_json::from_str(&request_rejection_message(None, reason)).expect("parse NOTICE");
+        assert_eq!(notice, serde_json::json!(["NOTICE", reason]));
+    }
+
     #[tokio::test]
     async fn send_loop_batches_queued_data_frames_into_one_flush() {
         let (data_tx, data_rx) = mpsc::channel(MAX_WS_SEND_BATCH);

crates/buzz-relay/src/handlers/req.rs

@@ -86,15 +86,21 @@ pub async fn handle_req(
         }
     };
 
-    let mut accessible_channels = match state
-        .get_accessible_channel_ids_cached(conn.tenant.community(), &pubkey_bytes)
-        .await
-    {
-        Ok(ids) => ids,
-        Err(e) => {
-            warn!(conn_id = %conn_id, "Failed to get accessible channels: {e}");
-            conn.send(RelayMessage::closed(&sub_id, "error: database error"));
-            return;
+    let mut accessible_channels = if filters_are_nip43_membership_only(&filters) {
+        metrics::counter!("buzz_req_global_access_resolution_skips_total", "kind" => "13534")
+            .increment(1);
+        Vec::new()
+    } else {
+        match state
+            .get_accessible_channel_ids_cached(conn.tenant.community(), &pubkey_bytes)
+            .await
+        {
+            Ok(ids) => ids,
+            Err(e) => {
+                warn!(conn_id = %conn_id, "Failed to get accessible channels: {e}");
+                conn.send(RelayMessage::closed(&sub_id, "error: database error"));
+                return;
+            }
         }
     };
     if let Some(allowed) = token_channel_ids.as_deref() {
@@ -816,6 +822,21 @@ pub fn filter_fully_pushable(filter: &Filter) -> bool {
     true
 }
 
+/// Return whether every filter exclusively targets the globally stored NIP-43
+/// membership snapshot. Such requests cannot return channel-scoped rows, so
+/// resolving the caller's complete accessible-channel set is wasted I/O.
+fn filters_are_nip43_membership_only(filters: &[Filter]) -> bool {
+    !filters.is_empty()
+        && filters.iter().all(|filter| {
+            filter.kinds.as_ref().is_some_and(|kinds| {
+                !kinds.is_empty()
+                    && kinds.iter().all(|kind| {
+                        kind.as_u16() as u32 == buzz_core::kind::KIND_NIP43_MEMBERSHIP_LIST
+                    })
+            })
+        })
+}
+
 /// Extract a channel UUID from a single filter's `#h` tag.
 fn extract_channel_id_from_filter(filter: &Filter) -> Option<uuid::Uuid> {
     for (tag_key, tag_values) in filter.generic_tags.iter() {
@@ -1323,6 +1344,20 @@ mod tests {
         ));
     }
 
+    #[test]
+    fn nip43_only_filters_skip_channel_access_resolution() {
+        let membership = Filter::new().kind(nostr::Kind::Custom(13_534));
+        assert!(filters_are_nip43_membership_only(&[
+            membership.clone(),
+            membership,
+        ]));
+        assert!(!filters_are_nip43_membership_only(&[]));
+        assert!(!filters_are_nip43_membership_only(&[Filter::new()]));
+        assert!(!filters_are_nip43_membership_only(&[
+            Filter::new().kinds([nostr::Kind::Custom(13_534), nostr::Kind::TextNote]),
+        ]));
+    }
+
     #[test]
     fn test_extract_channel_id_single_channel() {
         let channel_id = uuid::Uuid::new_v4();

crates/buzz-relay/src/handlers/side_effects.rs

@@ -2750,6 +2750,38 @@ async fn emit_initial_ref_state(
     Ok(())
 }
 
+/// Reconcile every community's event-backed NIP-43 membership view.
+///
+/// `relay_members` is canonical. A snapshot is rebuilt only when it is absent
+/// or older than the newest canonical membership mutation. This makes the
+/// sweep safe to run at startup and periodically without producing an event
+/// stream when nothing changed.
+pub async fn reconcile_nip43_membership_snapshots(state: &Arc<AppState>) -> anyhow::Result<usize> {
+    let communities = state.db.usage_community_hosts().await?;
+    let mut reconciled = 0usize;
+
+    for community in communities {
+        let community_id = buzz_core::CommunityId::from_uuid(community.id);
+        if !state
+            .db
+            .nip43_membership_snapshot_needs_reconciliation(
+                community_id,
+                &state.relay_keypair.public_key(),
+            )
+            .await?
+        {
+            continue;
+        }
+
+        let tenant = TenantContext::resolved(community_id, community.host);
+        publish_nip43_membership_list(&tenant, state).await?;
+        reconciled += 1;
+    }
+
+    metrics::counter!("buzz_nip43_membership_reconciliations_total").increment(reconciled as u64);
+    Ok(reconciled)
+}
+
 /// Publish a kind:13534 relay membership list event (NIP-43).
 ///
 /// Queries all current relay members and emits a relay-signed, NIP-70-protected
@@ -2763,6 +2795,24 @@ async fn emit_initial_ref_state(
 pub async fn publish_nip43_membership_list(
     tenant: &TenantContext,
     state: &Arc<AppState>,
+) -> anyhow::Result<()> {
+    let started_at = std::time::Instant::now();
+    metrics::counter!("buzz_nip43_membership_publications_total", "result" => "attempted")
+        .increment(1);
+    let result = publish_nip43_membership_list_inner(tenant, state).await;
+    metrics::histogram!("buzz_nip43_membership_publication_seconds")
+        .record(started_at.elapsed().as_secs_f64());
+    metrics::counter!(
+        "buzz_nip43_membership_publications_total",
+        "result" => if result.is_ok() { "succeeded" } else { "failed" }
+    )
+    .increment(1);
+    result
+}
+
+async fn publish_nip43_membership_list_inner(
+    tenant: &TenantContext,
+    state: &Arc<AppState>,
 ) -> anyhow::Result<()> {
     let relay_pubkey_hex = state.relay_keypair.public_key().to_hex();
     let community = tenant.community();

crates/buzz-relay/src/main.rs

@@ -461,34 +461,46 @@ async fn main() -> anyhow::Result<()> {
         );
     }
 
-    // NIP-43: publish the initial membership list on startup so clients can
-    // REQ kind:13534 immediately without waiting for the next membership change.
+    // NIP-43: reconcile the event-backed roster for every provisioned
+    // community before opening the listener. `relay_members` is canonical;
+    // this repairs pre-snapshot communities and any publication that failed
+    // after a membership transaction committed.
     if config.require_relay_membership {
-        // Resolve the deployment's community from the configured relay URL
-        // host (single-community per deployment), failing closed if the host
-        // isn't mapped. Await publication before opening the listener so the
-        // first client query observes the roster.
-        match buzz_relay::tenant::bind_deployment_community(&state.db, &state.config.relay_url)
-            .await
+        match buzz_relay::handlers::side_effects::reconcile_nip43_membership_snapshots(&state).await
         {
-            Ok(tenant) => {
-                if let Err(e) = buzz_relay::handlers::side_effects::publish_nip43_membership_list(
-                    &tenant, &state,
+            Ok(count) => info!(count, "NIP-43 membership snapshots reconciled on startup"),
+            Err(error) => {
+                tracing::warn!(%error, "NIP-43 membership snapshot startup reconciliation failed")
+            }
+        }
+
+        let reconcile_state = Arc::clone(&state);
+        let interval_secs = std::env::var("BUZZ_NIP43_RECONCILE_INTERVAL_SECS")
+            .ok()
+            .and_then(|value| value.parse::<u64>().ok())
+            .unwrap_or(60)
+            .max(1);
+        tokio::spawn(async move {
+            let mut interval = tokio::time::interval(std::time::Duration::from_secs(interval_secs));
+            interval.tick().await;
+            loop {
+                interval.tick().await;
+                match buzz_relay::handlers::side_effects::reconcile_nip43_membership_snapshots(
+                    &reconcile_state,
                 )
                 .await
                 {
-                    tracing::warn!(error = %e, "failed to publish initial NIP-43 membership list on startup");
-                } else {
-                    tracing::info!("NIP-43 membership list published on startup");
+                    Ok(count) if count > 0 => {
+                        info!(count, "NIP-43 membership snapshots repaired")
+                    }
+                    Ok(_) => {}
+                    Err(error) => tracing::warn!(
+                        %error,
+                        "periodic NIP-43 membership snapshot reconciliation failed"
+                    ),
                 }
             }
-            Err(e) => {
-                tracing::warn!(
-                    error = ?e,
-                    "initial NIP-43 membership list skipped: relay host is not mapped to a community"
-                );
-            }
-        }
+        });
     }
 
     // Emit kind:39000/39002 discovery events for channels that exist in the DB