Fix cross-pod membership notification fanout (#1291)

Signed-off-by: npub1mprnacetjua2xx3p5eddmhxyk6wv929ymm5py8kd2xfxurxahspqqlgyta <d8473ee32b973aa31a21a65adddcc4b69cc2a8a4dee8121ecd51926e0cddbc02@sprout-oss.stage.blox.sqprod.co>
Co-authored-by: npub1mprnacetjua2xx3p5eddmhxyk6wv929ymm5py8kd2xfxurxahspqqlgyta <d8473ee32b973aa31a21a65adddcc4b69cc2a8a4dee8121ecd51926e0cddbc02@sprout-oss.stage.blox.sqprod.co>

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

@@ -1021,7 +1021,7 @@ mod tests {
         use std::sync::Arc;
 
         use axum::extract::ws::Message;
-        use buzz_core::kind::KIND_PRESENCE_UPDATE;
+        use buzz_core::kind::{KIND_MEMBER_ADDED_NOTIFICATION, KIND_PRESENCE_UPDATE};
         use buzz_pubsub::ChannelEvent;
         use nostr::{EventBuilder, Filter, Keys, Kind};
         use tokio::sync::{mpsc, Mutex};
@@ -1035,9 +1035,11 @@ mod tests {
             super::fanout_access::test_state().await
         }
 
-        fn register_presence_sub(
+        fn register_global_sub(
             state: &AppState,
             sub_id: &str,
+            filter: Filter,
+            pubkey: Option<Vec<u8>>,
         ) -> (Uuid, mpsc::Receiver<Message>) {
             let conn_id = Uuid::new_v4();
             let (tx, rx) = mpsc::channel(10);
@@ -1049,21 +1051,58 @@ mod tests {
                 Arc::new(Mutex::new(HashMap::new())),
                 3,
             );
-            state.sub_registry.register(
-                conn_id,
-                sub_id.to_string(),
-                vec![Filter::new().kind(Kind::Custom(KIND_PRESENCE_UPDATE as u16))],
-                None,
-            );
+            if let Some(pubkey) = pubkey {
+                state.conn_manager.set_authenticated_pubkey(conn_id, pubkey);
+            }
+            state
+                .sub_registry
+                .register(conn_id, sub_id.to_string(), vec![filter], None);
             (conn_id, rx)
         }
 
+        fn register_presence_sub(
+            state: &AppState,
+            sub_id: &str,
+        ) -> (Uuid, mpsc::Receiver<Message>) {
+            register_global_sub(
+                state,
+                sub_id,
+                Filter::new().kind(Kind::Custom(KIND_PRESENCE_UPDATE as u16)),
+                None,
+            )
+        }
+
+        fn register_membership_sub(
+            state: &AppState,
+            sub_id: &str,
+            target: &Keys,
+        ) -> (Uuid, mpsc::Receiver<Message>) {
+            register_global_sub(
+                state,
+                sub_id,
+                Filter::new()
+                    .kind(Kind::Custom(KIND_MEMBER_ADDED_NOTIFICATION as u16))
+                    .pubkey(target.public_key()),
+                Some(target.public_key().to_bytes().to_vec()),
+            )
+        }
+
         fn presence_event(status: &str) -> nostr::Event {
             EventBuilder::new(Kind::Custom(KIND_PRESENCE_UPDATE as u16), status)
                 .sign_with_keys(&Keys::generate())
                 .expect("sign presence")
         }
 
+        fn membership_event(target: &Keys, channel_id: Uuid) -> nostr::Event {
+            EventBuilder::new(Kind::Custom(KIND_MEMBER_ADDED_NOTIFICATION as u16), "{}")
+                .tags([
+                    nostr::Tag::parse(["p", &target.public_key().to_hex()]).expect("p tag"),
+                    nostr::Tag::parse(["h", &channel_id.to_string()]).expect("h tag"),
+                ])
+                .sign_with_keys(&Keys::generate())
+                .expect("sign membership notification")
+        }
+
         fn event_from_ws_message(msg: Message) -> nostr::Event {
             let Message::Text(text) = msg else {
                 panic!("expected text ws message");
@@ -1116,6 +1155,39 @@ mod tests {
             );
         }
 
+        #[tokio::test]
+        async fn global_membership_pubsub_event_fans_out_by_p_tag() {
+            let state = test_state().await;
+            let target = Keys::generate();
+            let other = Keys::generate();
+            let (_target_conn, mut target_rx) =
+                register_membership_sub(&state, "membership-target", &target);
+            let (_other_conn, mut other_rx) =
+                register_membership_sub(&state, "membership-other", &other);
+            let event = membership_event(&target, Uuid::new_v4());
+            let event_id = event.id;
+
+            fan_out_pubsub_event(
+                &state,
+                ChannelEvent {
+                    channel_id: Uuid::nil(),
+                    event,
+                },
+            )
+            .await;
+
+            let delivered = event_from_ws_message(
+                target_rx
+                    .try_recv()
+                    .expect("target receives membership notification"),
+            );
+            assert_eq!(delivered.id, event_id);
+            assert!(
+                other_rx.try_recv().is_err(),
+                "membership notification should only reach matching #p subscribers"
+            );
+        }
+
         async fn redis_url_if_available() -> Option<String> {
             let redis_url =
                 std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379".to_string());

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

@@ -605,9 +605,26 @@ pub async fn emit_membership_notification(
         return Ok(());
     }
 
-    // Fan-out only — skip search indexing and workflow evaluation. Routed
-    // through the guarded send path for uniformity; the access gate no-ops for
-    // these globally-scoped (channel_id = None) events.
+    // Fan-out only — skip search indexing and workflow evaluation. Publish through
+    // Redis before local fan-out so agents connected to other relay pods receive
+    // the global membership notification and can subscribe to the new channel.
+    // Use the nil UUID sentinel for globally-scoped events, matching
+    // `dispatch_persistent_event` and `fan_out_pubsub_event`.
+    state.mark_local_event(&stored.event.id);
+    if let Err(e) = state.pubsub.publish_event(Uuid::nil(), &stored.event).await {
+        state
+            .local_event_ids
+            .invalidate(&stored.event.id.to_bytes());
+        warn!(
+            channel = %channel_id,
+            target = %target_hex,
+            kind = notification_kind,
+            "membership notification Redis publish failed: {e}"
+        );
+    }
+
+    // Routed through the guarded send path for uniformity; the access gate no-ops
+    // for these globally-scoped (channel_id = None) events.
     crate::handlers::event::fan_out_event_to_local_subscribers(state, &stored).await;
 
     info!(