From 9ca60737eb8f17ee45076782cf524a9e2c2a73e2 Mon Sep 17 00:00:00 2001 From: tlongwell-block <109685178+tlongwell-block@users.noreply.github.com> Date: Sat, 14 Mar 2026 23:40:06 -0400 Subject: [PATCH] feat: event-driven channel discovery via membership notifications (kind:44100/44101) (#65) --- crates/sprout-acp/src/config.rs | 91 ++- crates/sprout-acp/src/main.rs | 66 +- crates/sprout-acp/src/relay.rs | 155 ++++- crates/sprout-core/src/kind.rs | 10 + crates/sprout-relay/src/api/channels.rs | 11 + crates/sprout-relay/src/api/members.rs | 57 ++ crates/sprout-relay/src/config.rs | 2 +- crates/sprout-relay/src/handlers/event.rs | 13 +- crates/sprout-relay/src/handlers/req.rs | 42 +- .../sprout-relay/src/handlers/side_effects.rs | 129 +++- crates/sprout-relay/src/nip11.rs | 2 +- crates/sprout-relay/src/subscription.rs | 21 +- crates/sprout-test-client/tests/e2e_relay.rs | 574 ++++++++++++++++++ 13 files changed, 1157 insertions(+), 16 deletions(-) diff --git a/crates/sprout-acp/src/config.rs b/crates/sprout-acp/src/config.rs index 004691785..7b984f03a 100644 --- a/crates/sprout-acp/src/config.rs +++ b/crates/sprout-acp/src/config.rs @@ -29,7 +29,7 @@ pub enum ConfigError { // ── Enums ───────────────────────────────────────────────────────────────────── -#[derive(Debug, Clone, clap::ValueEnum)] +#[derive(Debug, Clone, PartialEq, clap::ValueEnum)] pub enum SubscribeMode { Mentions, All, @@ -443,6 +443,95 @@ pub fn resolve_channel_filters( result } +/// Resolve the subscription filter for a single dynamically-discovered channel. +/// +/// In Mentions/All mode, `channels_override` (--channels) is enforced — the agent +/// won't subscribe to channels outside the operator's allowlist. In Config mode, +/// `--channels` is ignored (per CLI contract) and rule-matching determines scope. +/// +/// Returns `None` when the channel is outside the agent's configured scope: +/// - Mentions/All: channel not in `channels_override` (if set) +/// - Config: no subscription rules match the channel +pub fn resolve_dynamic_channel_filter( + config: &Config, + channel_id: Uuid, + rules: &[crate::filter::SubscriptionRule], +) -> Option { + use sprout_core::kind::{ + KIND_STREAM_MESSAGE, KIND_STREAM_REMINDER, KIND_WORKFLOW_APPROVAL_REQUESTED, + }; + + // In Mentions/All mode, if the operator explicitly constrained channels + // with --channels, only allow dynamic subscription to channels in that + // allowlist. Config mode ignores --channels (per CLI contract) and uses + // rule-matching instead. + if config.subscribe_mode != SubscribeMode::Config { + if let Some(ref overrides) = config.channels_override { + let allowed = overrides + .iter() + .any(|s| s.parse::().ok() == Some(channel_id)); + if !allowed { + return None; + } + } + } + + match config.subscribe_mode { + SubscribeMode::Mentions => Some(ChannelFilter { + kinds: Some(config.kinds_override.clone().unwrap_or_else(|| { + vec![ + KIND_STREAM_MESSAGE, + KIND_WORKFLOW_APPROVAL_REQUESTED, + KIND_STREAM_REMINDER, + ] + })), + require_mention: !config.no_mention_filter, + }), + SubscribeMode::All => Some(ChannelFilter { + kinds: config.kinds_override.clone(), + require_mention: false, + }), + SubscribeMode::Config => { + // Same merge logic as resolve_channel_filters() Config branch: + // evaluate ALL rules against this specific channel (including + // channel-specific rules, not just ChannelScope::All). + let mut merged_kinds: Option> = Some(vec![]); + let mut require_mention = true; + let mut has_rule = false; + + for rule in rules { + if !rule_applies_to_channel(rule, channel_id) { + continue; + } + has_rule = true; + if rule.kinds.is_empty() { + merged_kinds = None; + } else if let Some(ref mut kinds) = merged_kinds { + for k in &rule.kinds { + if !kinds.contains(k) { + kinds.push(*k); + } + } + } + if !rule.require_mention { + require_mention = false; + } + } + + if !has_rule { + // No rules match — don't subscribe. Consistent with + // resolve_channel_filters() which omits unmatched channels. + return None; + } + + Some(ChannelFilter { + kinds: merged_kinds, + require_mention, + }) + } + } +} + fn rule_applies_to_channel(rule: &SubscriptionRule, channel_id: Uuid) -> bool { use crate::filter::ChannelScope; match &rule.channels { diff --git a/crates/sprout-acp/src/main.rs b/crates/sprout-acp/src/main.rs index 672adf6ce..1124c0be7 100644 --- a/crates/sprout-acp/src/main.rs +++ b/crates/sprout-acp/src/main.rs @@ -21,10 +21,12 @@ use pool::{AgentPool, OwnedAgent, PromptContext, PromptOutcome, PromptResult, Pr use queue::{EventQueue, QueuedEvent}; use relay::HarnessRelay; use sprout_core::kind::{ - KIND_STREAM_MESSAGE, KIND_STREAM_REMINDER, KIND_WORKFLOW_APPROVAL_REQUESTED, + KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION, KIND_STREAM_MESSAGE, + KIND_STREAM_REMINDER, KIND_WORKFLOW_APPROVAL_REQUESTED, }; use tokio::sync::watch; use tracing_subscriber::EnvFilter; +use uuid::Uuid; #[tokio::main] async fn main() -> Result<()> { @@ -65,6 +67,13 @@ async fn main() -> Result<()> { tracing::info!("connected to relay at {}", config.relay_url); + // ── Step 2b: Subscribe to membership notifications ──────────────────────── + relay + .subscribe_membership_notifications() + .await + .map_err(|e| anyhow::anyhow!("membership notification subscribe error: {e}"))?; + tracing::info!("subscribed to membership notifications"); + // ── Step 3: Discover channels and build subscription rules ──────────────── let channels = relay .discover_channels() @@ -174,6 +183,12 @@ async fn main() -> Result<()> { }); } + // Track the newest membership notification timestamp per channel so that + // replayed events (returned in DESC order on reconnect) don't override the + // correct final state. The first event seen per channel is the newest; any + // older duplicate for the same channel is skipped. + let mut membership_newest_ts: HashMap = HashMap::new(); + // ── Step 8: Main orchestration loop ────────────────────────────────────── // // Branches 1 & 2 both need to borrow `pool`, but they access different @@ -202,6 +217,55 @@ async fn main() -> Result<()> { let _ = result_rx; // end split borrow before relay handling match sprout_event { Some(sprout_event) => { + let kind_u32 = sprout_event.event.kind.as_u16() as u32; + + // ── Membership notification handling ────────────── + if kind_u32 == KIND_MEMBER_ADDED_NOTIFICATION + || kind_u32 == KIND_MEMBER_REMOVED_NOTIFICATION + { + let ch = sprout_event.channel_id; + let ts = sprout_event.event.created_at.as_u64(); + + // Skip stale membership events: on reconnect the relay + // replays events newest-first, so the first event per + // channel is authoritative. Any later (older) event for + // the same channel is outdated and must be ignored. + let dominated = membership_newest_ts + .get(&ch) + .is_some_and(|&newest| ts < newest); + if dominated { + tracing::debug!( + channel_id = %ch, + kind = kind_u32, + ts, + "skipping stale membership notification (newer already processed)" + ); + continue; + } + membership_newest_ts + .entry(ch) + .and_modify(|v| *v = (*v).max(ts)) + .or_insert(ts); + + if kind_u32 == KIND_MEMBER_ADDED_NOTIFICATION { + if let Some(filter) = config::resolve_dynamic_channel_filter(&config, ch, &rules) { + tracing::info!(channel_id = %ch, "membership notification: subscribing to new channel"); + if let Err(e) = relay.subscribe_channel(ch, filter).await { + tracing::warn!("failed to subscribe to new channel {ch}: {e}"); + } + } else { + tracing::debug!(channel_id = %ch, "membership notification: no matching rules — skipping"); + } + } else { + tracing::info!(channel_id = %ch, "membership notification: unsubscribing from channel"); + if let Err(e) = relay.unsubscribe_channel(ch).await { + tracing::warn!("failed to unsubscribe from channel {ch}: {e}"); + } + } + continue; + } + // ── End membership notification handling ────────── + if config.ignore_self && sprout_event.event.pubkey.to_hex() == pubkey_hex { tracing::debug!(channel_id = %sprout_event.channel_id, "dropping self-authored event"); continue; diff --git a/crates/sprout-acp/src/relay.rs b/crates/sprout-acp/src/relay.rs index 15ee9ead6..67be64276 100644 --- a/crates/sprout-acp/src/relay.rs +++ b/crates/sprout-acp/src/relay.rs @@ -38,6 +38,7 @@ const AUTH_TIMEOUT: Duration = Duration::from_secs(5); use futures_util::{SinkExt, StreamExt}; use nostr::{Event, EventBuilder, Keys, Kind, Tag, Url as NostrUrl}; use serde_json::{json, Value}; +use sprout_core::kind::{KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION}; use tokio::sync::mpsc; use tokio::time::timeout; use tokio_tungstenite::{connect_async, tungstenite::Message, MaybeTlsStream, WebSocketStream}; @@ -122,6 +123,9 @@ enum RelayMessage { // ── Commands sent from HarnessRelay to the background task ─────────────────── +/// Subscription ID for the global membership notification subscription. +const MEMBERSHIP_NOTIF_SUB_ID: &str = "membership-notif"; + /// Commands sent from `HarnessRelay` to the background WebSocket task. enum RelayCommand { /// Subscribe to a channel (sends a NIP-01 REQ) with the given filter. @@ -136,6 +140,8 @@ enum RelayCommand { Reconnect, /// Shut down the background task. Shutdown, + /// Subscribe to global membership notifications. + SubscribeMembership, } // ── WebSocket stream type alias ─────────────────────────────────────────────── @@ -283,6 +289,15 @@ impl HarnessRelay { Ok(()) } + /// Subscribe to membership notifications for this agent. + pub async fn subscribe_membership_notifications(&mut self) -> Result<(), RelayError> { + self.cmd_tx + .send(RelayCommand::SubscribeMembership) + .await + .map_err(|_| RelayError::ConnectionClosed)?; + Ok(()) + } + /// Unsubscribe from a channel. #[allow(dead_code)] pub async fn unsubscribe_channel(&mut self, channel_id: Uuid) -> Result<(), RelayError> { @@ -335,6 +350,15 @@ struct BgState { seen_ids: HashSet, /// Per-channel filter used on subscribe (for resubscribe after reconnect). active_filters: HashMap, + /// Oldest timestamp of a membership notification that was dropped due to + /// backpressure. If set, reconnect replay must start from this timestamp + /// (minus skew) to re-deliver the lost event. Reset on successful reconnect. + membership_dropped_since: Option, + /// Newest successfully-enqueued membership notification timestamp. + /// Used as the `since` for reconnect replay when no events were dropped. + membership_last_seen: Option, + /// Whether the membership notification subscription is active. + membership_sub_active: bool, } impl BgState { @@ -344,6 +368,9 @@ impl BgState { last_seen: HashMap::new(), seen_ids: HashSet::new(), active_filters: HashMap::new(), + membership_dropped_since: None, + membership_last_seen: None, + membership_sub_active: false, } } @@ -475,6 +502,22 @@ async fn run_background_task( debug!("unsubscribed from channel {channel_id}"); } } + Some(RelayCommand::SubscribeMembership) => { + let _ = + send_membership_subscribe(&mut ws, &agent_pubkey_hex, None).await; + state.membership_sub_active = true; + // Seed the watermark so reconnect replays from this point + // rather than falling back to since=now (which could miss + // notifications during the reconnect gap). + if state.membership_last_seen.is_none() { + state.membership_last_seen = Some( + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs(), + ); + } + } Some(RelayCommand::Reconnect) => { // Reconnect command already consumed — skip the drain loop. wait_for_reconnect( @@ -527,7 +570,39 @@ async fn handle_ws_message( subscription_id, event, } => { - if let Some(channel_id) = channel_id_from_sub_id(&subscription_id) { + if subscription_id == MEMBERSHIP_NOTIF_SUB_ID { + // Membership notification — extract channel UUID from h tag. + let channel_uuid = match extract_h_tag_uuid(&event) { + Some(uuid) => uuid, + None => { + warn!("membership notification missing h tag — dropping"); + return true; + } + }; + let ts = event.created_at.as_u64(); + let sprout_event = SproutEvent { + channel_id: channel_uuid, + event: *event, + }; + match event_tx.try_send(Some(sprout_event)) { + Ok(()) => { + state.membership_last_seen = + Some(state.membership_last_seen.unwrap_or(0).max(ts)); + } + Err(mpsc::error::TrySendError::Full(_)) => { + // Track the oldest dropped timestamp so reconnect + // replay starts early enough to re-deliver it. + state.membership_dropped_since = + Some(state.membership_dropped_since.map_or(ts, |d| d.min(ts))); + warn!( + channel_id = %channel_uuid, + ts, + "membership notification dropped (backpressure) — will replay from {ts} on reconnect" + ); + } + Err(mpsc::error::TrySendError::Closed(_)) => return false, + } + } else if let Some(channel_id) = channel_id_from_sub_id(&subscription_id) { if state.record_event(channel_id, &event) { let sprout_event = SproutEvent { channel_id, @@ -635,6 +710,9 @@ async fn wait_for_reconnect( state.active_subscriptions.remove(&channel_id); state.active_filters.remove(&channel_id); } + Some(RelayCommand::SubscribeMembership) => { + state.membership_sub_active = true; + } } } } @@ -665,6 +743,24 @@ async fn wait_for_reconnect( } } + // Resubscribe to membership notifications if active. + // Use the oldest of (dropped, last_seen) so dropped events are replayed. + if state.membership_sub_active { + let replay_since = + match (state.membership_dropped_since, state.membership_last_seen) { + (Some(d), Some(l)) => Some(d.min(l)), + (Some(d), None) => Some(d), + (None, Some(l)) => Some(l), + (None, None) => None, + }; + let sent = send_membership_subscribe(ws, agent_pubkey_hex, replay_since).await; + if sent { + // Only clear drop tracker if the REQ was actually sent — + // if send failed, retain it so the next reconnect retries. + state.membership_dropped_since = None; + } + } + return; } Err(e) => { @@ -745,6 +841,63 @@ async fn send_subscribe( } } +/// Send a NIP-01 REQ for membership notifications (kind:44100+44101, global, #p=[agent_pubkey]). +/// Returns `true` if the REQ was successfully written to the WebSocket. +async fn send_membership_subscribe( + ws: &mut WsStream, + agent_pubkey_hex: &str, + since: Option, +) -> bool { + let mut req_filter = serde_json::Map::new(); + req_filter.insert( + "kinds".into(), + json!([ + KIND_MEMBER_ADDED_NOTIFICATION, + KIND_MEMBER_REMOVED_NOTIFICATION + ]), + ); + req_filter.insert("#p".into(), json!([agent_pubkey_hex])); + + let since_ts = match since { + Some(ts) => ts.saturating_sub(SINCE_SKEW_SECS), + None => std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs(), + }; + req_filter.insert("since".into(), json!(since_ts)); + + let req = json!(["REQ", MEMBERSHIP_NOTIF_SUB_ID, Value::Object(req_filter)]); + match serde_json::to_string(&req) { + Ok(text) => match ws.send(Message::Text(text.into())).await { + Ok(()) => { + debug!("subscribed to membership notifications (since={since_ts})"); + true + } + Err(e) => { + warn!("failed to send membership notification REQ: {e}"); + false + } + }, + Err(e) => { + warn!("failed to serialize membership notification REQ: {e}"); + false + } + } +} + +/// Extract a channel UUID from the h tag of a Nostr event. +fn extract_h_tag_uuid(event: &nostr::Event) -> Option { + event.tags.iter().find_map(|tag| { + let tag_vec = tag.as_slice(); + if tag_vec.len() >= 2 && tag_vec[0] == "h" { + tag_vec[1].parse::().ok() + } else { + None + } + }) +} + /// Build and send a NIP-42 AUTH response event. async fn send_auth_response( ws: &mut WsStream, diff --git a/crates/sprout-core/src/kind.rs b/crates/sprout-core/src/kind.rs index d689b98a2..04077254e 100644 --- a/crates/sprout-core/src/kind.rs +++ b/crates/sprout-core/src/kind.rs @@ -132,6 +132,14 @@ pub const KIND_SUBSCRIPTION_PAUSED: u32 = 44003; /// A paused subscription was resumed. pub const KIND_SUBSCRIPTION_RESUMED: u32 = 44004; +/// Relay-signed notification: the target pubkey was added to a channel. +/// Stored globally (channel_id = None) with p-tag = target, h-tag = channel UUID. +pub const KIND_MEMBER_ADDED_NOTIFICATION: u32 = 44100; + +/// Relay-signed notification: the target pubkey was removed from a channel. +/// Stored globally (channel_id = None) with p-tag = target, h-tag = channel UUID. +pub const KIND_MEMBER_REMOVED_NOTIFICATION: u32 = 44101; + // Forum / social (45000–45999) // V1 used addressable range (30001–30003) — wrong. /// A forum post (thread root). @@ -247,6 +255,8 @@ pub const ALL_KINDS: &[u32] = &[ KIND_SUBSCRIPTION_MATCHED, KIND_SUBSCRIPTION_PAUSED, KIND_SUBSCRIPTION_RESUMED, + KIND_MEMBER_ADDED_NOTIFICATION, + KIND_MEMBER_REMOVED_NOTIFICATION, KIND_FORUM_POST, KIND_FORUM_VOTE, KIND_FORUM_COMMENT, diff --git a/crates/sprout-relay/src/api/channels.rs b/crates/sprout-relay/src/api/channels.rs index f69fb47d8..bb4a3c589 100644 --- a/crates/sprout-relay/src/api/channels.rs +++ b/crates/sprout-relay/src/api/channels.rs @@ -160,6 +160,17 @@ pub async fn create_channel( { tracing::warn!(channel_id = %channel.id, error = %e, "NIP-29 discovery emission failed"); } + if let Err(e) = crate::handlers::side_effects::emit_membership_notification( + &state, + channel.id, + &pubkey_bytes, + &pubkey_bytes, + sprout_core::kind::KIND_MEMBER_ADDED_NOTIFICATION, + ) + .await + { + tracing::warn!("membership notification for channel creator failed: {e}"); + } Ok(( StatusCode::CREATED, diff --git a/crates/sprout-relay/src/api/members.rs b/crates/sprout-relay/src/api/members.rs index f81d9aa3a..4b1bb79b0 100644 --- a/crates/sprout-relay/src/api/members.rs +++ b/crates/sprout-relay/src/api/members.rs @@ -21,8 +21,10 @@ use serde::Deserialize; use sprout_db::channel::MemberRole; use uuid::Uuid; +use crate::handlers::side_effects::emit_membership_notification; use crate::handlers::side_effects::emit_system_message; use crate::state::AppState; +use sprout_core::kind::{KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION}; use super::{ api_error, check_channel_access, check_token_channel_access, extract_auth_context, forbidden, @@ -143,6 +145,17 @@ pub async fn add_members( { tracing::warn!("Failed to emit system message: {e}"); } + if let Err(e) = emit_membership_notification( + &state, + channel_id, + &pubkey_bytes, + &actor_bytes, + KIND_MEMBER_ADDED_NOTIFICATION, + ) + .await + { + tracing::warn!("membership notification failed: {e}"); + } added.push(hex_pk.clone()); } Err(e) => { @@ -224,6 +237,17 @@ pub async fn add_members( { tracing::warn!("Failed to emit system message: {e}"); } + if let Err(e) = emit_membership_notification( + &state, + channel_id, + &pubkey_bytes, + &actor_bytes, + KIND_MEMBER_ADDED_NOTIFICATION, + ) + .await + { + tracing::warn!("membership notification failed: {e}"); + } added.push(hex_pk.clone()); } Err(e) => { @@ -329,6 +353,17 @@ pub async fn remove_member( { tracing::warn!("Failed to emit system message: {e}"); } + if let Err(e) = emit_membership_notification( + &state, + channel_id, + &target_bytes, + &actor_bytes, + KIND_MEMBER_REMOVED_NOTIFICATION, + ) + .await + { + tracing::warn!("membership notification failed: {e}"); + } Ok(Json(serde_json::json!({ "removed": true }))) } @@ -452,6 +487,17 @@ pub async fn join_channel( { tracing::warn!("Failed to emit system message: {e}"); } + if let Err(e) = emit_membership_notification( + &state, + channel_id, + &pubkey_bytes, + &pubkey_bytes, + KIND_MEMBER_ADDED_NOTIFICATION, + ) + .await + { + tracing::warn!("membership notification failed: {e}"); + } Ok(Json(serde_json::json!({ "joined": true, @@ -522,6 +568,17 @@ pub async fn leave_channel( { tracing::warn!("Failed to emit system message: {e}"); } + if let Err(e) = emit_membership_notification( + &state, + channel_id, + &pubkey_bytes, + &pubkey_bytes, + KIND_MEMBER_REMOVED_NOTIFICATION, + ) + .await + { + tracing::warn!("membership notification failed: {e}"); + } Ok(Json(serde_json::json!({ "left": true }))) } diff --git a/crates/sprout-relay/src/config.rs b/crates/sprout-relay/src/config.rs index 03c99e830..a4d6f9548 100644 --- a/crates/sprout-relay/src/config.rs +++ b/crates/sprout-relay/src/config.rs @@ -86,7 +86,7 @@ impl Config { let max_concurrent_handlers = std::env::var("SPROUT_MAX_CONCURRENT_HANDLERS") .ok() .and_then(|v| v.parse().ok()) - .unwrap_or(64); + .unwrap_or(1024); let send_buffer_size = std::env::var("SPROUT_SEND_BUFFER") .ok() diff --git a/crates/sprout-relay/src/handlers/event.rs b/crates/sprout-relay/src/handlers/event.rs index ceeb55f3a..60dc7a30b 100644 --- a/crates/sprout-relay/src/handlers/event.rs +++ b/crates/sprout-relay/src/handlers/event.rs @@ -10,7 +10,8 @@ use sprout_audit::{AuditAction, NewAuditEntry}; use sprout_core::event::StoredEvent; use sprout_core::kind::{ event_kind_u32, is_ephemeral, is_workflow_execution_kind, KIND_AUTH, KIND_CANVAS, - KIND_DELETION, KIND_FORUM_COMMENT, KIND_FORUM_POST, KIND_FORUM_VOTE, KIND_PRESENCE_UPDATE, + KIND_DELETION, KIND_FORUM_COMMENT, KIND_FORUM_POST, KIND_FORUM_VOTE, + KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION, KIND_PRESENCE_UPDATE, KIND_STREAM_MESSAGE, KIND_STREAM_MESSAGE_BOOKMARKED, KIND_STREAM_MESSAGE_DIFF, KIND_STREAM_MESSAGE_EDIT, KIND_STREAM_MESSAGE_PINNED, KIND_STREAM_MESSAGE_SCHEDULED, KIND_STREAM_MESSAGE_V2, KIND_STREAM_REMINDER, @@ -187,6 +188,16 @@ pub async fn handle_event(event: Event, conn: Arc, state: Arc, + channel_id: Uuid, + target_pubkey: &[u8], + actor_pubkey: &[u8], + notification_kind: u32, +) -> anyhow::Result<()> { + let target_hex = hex::encode(target_pubkey); + let actor_hex = hex::encode(actor_pubkey); + let channel_id_str = channel_id.to_string(); + + let p_tag = Tag::parse(&["p", &target_hex]) + .map_err(|e| anyhow::anyhow!("failed to build p tag: {e}"))?; + let h_tag = Tag::parse(&["h", &channel_id_str]) + .map_err(|e| anyhow::anyhow!("failed to build h tag: {e}"))?; + + let event_type = match notification_kind { + KIND_MEMBER_ADDED_NOTIFICATION => "member_added", + KIND_MEMBER_REMOVED_NOTIFICATION => "member_removed", + _ => { + return Err(anyhow::anyhow!( + "invalid notification kind: {notification_kind}" + )) + } + }; + + let content = serde_json::json!({ + "type": event_type, + "channel_id": channel_id_str, + "actor": actor_hex, + }) + .to_string(); + + let event = EventBuilder::new( + Kind::Custom(notification_kind as u16), + content, + [p_tag, h_tag], + ) + .sign_with_keys(&state.relay_keypair) + .map_err(|e| anyhow::anyhow!("failed to sign membership notification: {e}"))?; + + // Store with channel_id = None → globally scoped, reachable by global subscribers. + let (stored, was_inserted) = state.db.insert_event(&event, None).await?; + if !was_inserted { + return Ok(()); + } + + // Fan-out only — skip search indexing and workflow evaluation. + let matches = state.sub_registry.fan_out(&stored); + if !matches.is_empty() { + let event_json = match serde_json::to_string(&stored.event) { + Ok(json) => json, + Err(e) => { + warn!("failed to serialize membership notification for fan-out: {e}"); + return Ok(()); + } + }; + for (target_conn_id, sub_id) in &matches { + let msg = format!(r#"["EVENT","{}",{}]"#, sub_id, event_json); + state.conn_manager.send_to(*target_conn_id, msg); + } + } + + info!( + channel = %channel_id, + target = %target_hex, + kind = notification_kind, + "membership notification emitted" + ); + Ok(()) +} + /// Sign, store (replacing previous), and fan-out a single addressable discovery event. async fn emit_addressable_discovery_event( state: &Arc, @@ -538,6 +615,18 @@ async fn handle_put_user(event: &Event, state: &Arc) -> anyhow::Result warn!(channel = %channel_id, error = %e, "NIP-29 group discovery emission failed"); } + if let Err(e) = emit_membership_notification( + state, + channel_id, + &target_pubkey, + &actor_bytes, + KIND_MEMBER_ADDED_NOTIFICATION, + ) + .await + { + warn!(channel = %channel_id, error = %e, "membership notification emission failed"); + } + info!(channel = %channel_id, target = %target_hex, "NIP-29 PUT_USER processed"); Ok(()) } @@ -589,6 +678,18 @@ async fn handle_remove_user(event: &Event, state: &Arc) -> anyhow::Res warn!(channel = %channel_id, error = %e, "NIP-29 group discovery emission failed"); } + if let Err(e) = emit_membership_notification( + state, + channel_id, + &target_pubkey, + &actor_bytes, + KIND_MEMBER_REMOVED_NOTIFICATION, + ) + .await + { + warn!(channel = %channel_id, error = %e, "membership notification emission failed"); + } + Ok(()) } @@ -783,6 +884,18 @@ async fn handle_create_group(event: &Event, state: &Arc) -> anyhow::Re warn!(channel = %channel.id, error = %e, "NIP-29 group discovery emission failed"); } + if let Err(e) = emit_membership_notification( + state, + channel.id, + &actor_bytes, + &actor_bytes, // creator is both actor and target + KIND_MEMBER_ADDED_NOTIFICATION, + ) + .await + { + warn!(channel = %channel.id, error = %e, "membership notification emission failed"); + } + info!(channel_id = %channel.id, name = %name, "NIP-29 CREATE_GROUP processed"); Ok(()) } @@ -864,6 +977,18 @@ async fn handle_leave_request(event: &Event, state: &Arc) -> anyhow::R warn!(channel = %channel_id, error = %e, "NIP-29 group discovery emission failed"); } + if let Err(e) = emit_membership_notification( + state, + channel_id, + &actor_bytes, + &actor_bytes, // self-leave: actor == target + KIND_MEMBER_REMOVED_NOTIFICATION, + ) + .await + { + warn!(channel = %channel_id, error = %e, "membership notification emission failed"); + } + Ok(()) } diff --git a/crates/sprout-relay/src/nip11.rs b/crates/sprout-relay/src/nip11.rs index 68c6821d3..f34c35a9e 100644 --- a/crates/sprout-relay/src/nip11.rs +++ b/crates/sprout-relay/src/nip11.rs @@ -61,7 +61,7 @@ impl RelayInfo { version: env!("CARGO_PKG_VERSION").to_string(), limitation: Some(RelayLimitation { max_message_length: Some(MAX_FRAME_BYTES as u64), - max_subscriptions: Some(100), + max_subscriptions: Some(1024), max_filters: Some(10), max_limit: Some(500), max_subid_length: Some(256), diff --git a/crates/sprout-relay/src/subscription.rs b/crates/sprout-relay/src/subscription.rs index 3f0cfa6a9..60c50e269 100644 --- a/crates/sprout-relay/src/subscription.rs +++ b/crates/sprout-relay/src/subscription.rs @@ -139,9 +139,16 @@ impl SubscriptionRegistry { } } } else { + // Global event (channel_id = None) — only deliver to global subscriptions. + // Channel-scoped subscriptions are skipped: they target a specific channel + // and should not receive global infrastructure events (e.g. membership + // notifications) even if tag matching would succeed. for conn_entry in self.subs.iter() { let conn_id = *conn_entry.key(); - for (sub_id, (filters, _)) in conn_entry.value().iter() { + for (sub_id, (filters, sub_channel_id)) in conn_entry.value().iter() { + if sub_channel_id.is_some() { + continue; // skip channel-scoped subscriptions + } if filters_match(filters, event) { results.push((conn_id, sub_id.clone())); } @@ -149,12 +156,12 @@ impl SubscriptionRegistry { } } - // NOTE: Global subscriptions (channel_id = None) intentionally do NOT - // receive channel-scoped events. Delivering channel events to global subs - // would bypass the channel membership check performed in req.rs, leaking - // private channel content to unauthorized subscribers. Clients must - // subscribe to a specific channel to receive its events — that path goes - // through the access-control check that verifies membership. + // NOTE: The scoping invariant is symmetric: + // - Global subscriptions (channel_id = None) do NOT receive channel-scoped events. + // - Channel-scoped subscriptions do NOT receive global events. + // This prevents both directions of information leakage: channel content + // leaking to global subscribers, and global infrastructure events (like + // membership notifications) leaking to channel subscribers. results } diff --git a/crates/sprout-test-client/tests/e2e_relay.rs b/crates/sprout-test-client/tests/e2e_relay.rs index 602f516ee..627d7d622 100644 --- a/crates/sprout-test-client/tests/e2e_relay.rs +++ b/crates/sprout-test-client/tests/e2e_relay.rs @@ -1170,3 +1170,577 @@ async fn test_nip29_standard_client_flow() { client.disconnect().await.expect("clean disconnect"); } + +/// Client-submitted kind:44100 (member-added notification) must be rejected. +/// Only the relay keypair may sign these events. +#[tokio::test] +#[ignore] +async fn test_membership_notification_kind_rejected() { + let url = relay_url(); + let keys = Keys::generate(); + let channel_id = create_test_channel(&keys).await; + + let mut client = SproutTestClient::connect(&url, &keys) + .await + .expect("connect"); + + let p_tag = Tag::parse(&["p", &keys.public_key().to_hex()]).expect("p tag"); + let h_tag = Tag::parse(&["h", &channel_id]).expect("h tag"); + let event = EventBuilder::new(Kind::Custom(44100), "", [p_tag, h_tag]) + .sign_with_keys(&keys) + .expect("sign kind:44100"); + + let ok = client.send_event(event).await.expect("send"); + + assert!( + !ok.accepted, + "relay must reject client-submitted kind:44100, but accepted it" + ); + let msg_lower = ok.message.to_lowercase(); + assert!( + msg_lower.contains("relay-signed only") + || msg_lower.contains("relay signed only") + || msg_lower.contains("relay"), + "rejection message should mention relay-signed restriction, got: {}", + ok.message + ); + + client.disconnect().await.expect("disconnect"); +} + +/// When a member is added via REST, the relay must emit a kind:44100 notification +/// to any subscriber filtering on `#p` = that member's pubkey. +#[tokio::test] +#[ignore] +async fn test_membership_notification_emitted_on_add() { + let url = relay_url(); + + let owner_keys = Keys::generate(); + let agent_keys = Keys::generate(); + let agent_pubkey_hex = agent_keys.public_key().to_hex(); + + // Connect as agent — NIP-42 auth establishes the authenticated pubkey. + let mut agent_client = SproutTestClient::connect(&url, &agent_keys) + .await + .expect("connect as agent"); + + // Create a channel owned by owner (not agent). + let channel_id = create_test_channel(&owner_keys).await; + + // Subscribe to membership notifications for agent's own pubkey. + let sid = sub_id("membership-notif"); + let filter = Filter::new() + .kinds(vec![Kind::Custom(44100), Kind::Custom(44101)]) + .custom_tag( + SingleLetterTag::lowercase(Alphabet::P), + [agent_pubkey_hex.as_str()], + ) + .since(nostr::Timestamp::now() - 5u64); + + agent_client + .subscribe(&sid, vec![filter]) + .await + .expect("subscribe to membership notifications"); + + // Drain EOSE — no historical events expected. + agent_client + .collect_until_eose(&sid, Duration::from_secs(5)) + .await + .expect("EOSE for membership sub"); + + // Add agent to the channel via REST (owner's X-Pubkey header). + let http_client = reqwest::Client::new(); + let resp = http_client + .post(format!( + "{}/api/channels/{}/members", + relay_http_url(), + channel_id + )) + .header("X-Pubkey", &owner_keys.public_key().to_hex()) + .json(&serde_json::json!({ "pubkeys": [agent_pubkey_hex] })) + .send() + .await + .expect("add member request"); + assert!( + resp.status().is_success(), + "add member failed: {}", + resp.status() + ); + + // Wait for the kind:44100 notification. + let msg = agent_client + .recv_event(Duration::from_secs(5)) + .await + .expect("recv kind:44100 notification"); + + match msg { + RelayMessage::Event { event, .. } => { + assert_eq!( + event.kind, + Kind::Custom(44100), + "expected kind:44100, got {}", + event.kind.as_u16() + ); + + let tags: Vec> = event + .tags + .iter() + .map(|t| t.as_slice().iter().map(|s| s.to_string()).collect()) + .collect(); + + let has_p = tags + .iter() + .any(|t| t.len() >= 2 && t[0] == "p" && t[1] == agent_pubkey_hex); + assert!( + has_p, + "kind:44100 missing p tag = agent pubkey. tags: {tags:?}" + ); + + let has_h = tags + .iter() + .any(|t| t.len() >= 2 && t[0] == "h" && t[1] == channel_id); + assert!( + has_h, + "kind:44100 missing h tag = channel uuid. tags: {tags:?}" + ); + } + other => panic!("expected EVENT kind:44100, got {other:?}"), + } + + agent_client.disconnect().await.expect("disconnect"); +} + +/// Subscribing to kind:44100/44101 without a `#p` filter must be rejected with CLOSED. +#[tokio::test] +#[ignore] +async fn test_membership_notification_requires_p_filter() { + let url = relay_url(); + let keys = Keys::generate(); + + let mut client = SproutTestClient::connect(&url, &keys) + .await + .expect("connect"); + + let sid = sub_id("no-p-filter"); + let filter = Filter::new().kinds(vec![Kind::Custom(44100), Kind::Custom(44101)]); + + client + .subscribe(&sid, vec![filter]) + .await + .expect("send REQ"); + + // Drain until we get the CLOSED for our subscription. + let msg = loop { + let m = client + .recv_event(Duration::from_secs(5)) + .await + .expect("recv CLOSED"); + match &m { + RelayMessage::Eose { .. } => continue, + RelayMessage::Event { .. } => continue, + _ => break m, + } + }; + + match msg { + RelayMessage::Closed { + subscription_id, + message, + } => { + assert_eq!( + subscription_id, sid, + "CLOSED for wrong subscription: {subscription_id}" + ); + assert!( + message.to_lowercase().contains("restricted"), + "expected 'restricted' in CLOSED message, got: {message}" + ); + } + other => panic!("expected CLOSED, got {other:?}"), + } + + client.disconnect().await.expect("disconnect"); +} + +/// A subscription with NO kinds filter and NO #p filter (wildcard) must be rejected with CLOSED +/// because it can match kind:44100/44101. +#[tokio::test] +#[ignore] +async fn test_membership_notification_wildcard_filter_rejected() { + let url = relay_url(); + let keys = Keys::generate(); + + let mut client = SproutTestClient::connect(&url, &keys) + .await + .expect("connect"); + + let sid = sub_id("wildcard-filter"); + // Empty filter — no kinds, no #p — can match kind:44100/44101. + let filter = Filter::new(); + + client + .subscribe(&sid, vec![filter]) + .await + .expect("send REQ"); + + // Drain until we get the CLOSED for our subscription. + let msg = loop { + let m = client + .recv_event(Duration::from_secs(5)) + .await + .expect("recv CLOSED"); + match &m { + RelayMessage::Eose { .. } => continue, + RelayMessage::Event { .. } => continue, + _ => break m, + } + }; + + match msg { + RelayMessage::Closed { + subscription_id, + message, + } => { + assert_eq!( + subscription_id, sid, + "CLOSED for wrong subscription: {subscription_id}" + ); + assert!( + message.to_lowercase().contains("restricted"), + "expected 'restricted' in CLOSED message, got: {message}" + ); + } + other => panic!("expected CLOSED, got {other:?}"), + } + + client.disconnect().await.expect("disconnect"); +} + +/// Subscribing to kind:44100/44101 with someone else's `#p` must be rejected with CLOSED. +#[tokio::test] +#[ignore] +async fn test_membership_notification_requires_own_p_filter() { + let url = relay_url(); + + let keys_a = Keys::generate(); + let keys_b = Keys::generate(); + let keys_b_pubkey_hex = keys_b.public_key().to_hex(); + + // Connect as keys_a. + let mut client = SproutTestClient::connect(&url, &keys_a) + .await + .expect("connect as keys_a"); + + let sid = sub_id("wrong-p-filter"); + // Filter uses keys_b's pubkey — not the authenticated pubkey (keys_a). + let filter = Filter::new() + .kinds(vec![Kind::Custom(44100), Kind::Custom(44101)]) + .custom_tag( + SingleLetterTag::lowercase(Alphabet::P), + [keys_b_pubkey_hex.as_str()], + ); + + client + .subscribe(&sid, vec![filter]) + .await + .expect("send REQ"); + + // Drain until we get the CLOSED for our subscription. + let msg = loop { + let m = client + .recv_event(Duration::from_secs(5)) + .await + .expect("recv CLOSED"); + match &m { + RelayMessage::Eose { .. } => continue, + RelayMessage::Event { .. } => continue, + _ => break m, + } + }; + + match msg { + RelayMessage::Closed { + subscription_id, + message, + } => { + assert_eq!( + subscription_id, sid, + "CLOSED for wrong subscription: {subscription_id}" + ); + assert!( + message.to_lowercase().contains("restricted"), + "expected 'restricted' in CLOSED message, got: {message}" + ); + } + other => panic!("expected CLOSED, got {other:?}"), + } + + client.disconnect().await.expect("disconnect"); +} + +/// When a member is removed via REST, the relay must emit a kind:44101 notification +/// to any subscriber filtering on `#p` = that member's pubkey. +#[tokio::test] +#[ignore] +async fn test_membership_notification_emitted_on_remove() { + let url = relay_url(); + + let owner_keys = Keys::generate(); + let agent_keys = Keys::generate(); + let agent_pubkey_hex = agent_keys.public_key().to_hex(); + + // Connect as agent — NIP-42 auth establishes the authenticated pubkey. + let mut agent_client = SproutTestClient::connect(&url, &agent_keys) + .await + .expect("connect as agent"); + + // Create a channel owned by owner (not agent). + let channel_id = create_test_channel(&owner_keys).await; + + // Subscribe to membership notifications for agent's own pubkey. + let sid = sub_id("membership-remove-notif"); + let filter = Filter::new() + .kinds(vec![Kind::Custom(44100), Kind::Custom(44101)]) + .custom_tag( + SingleLetterTag::lowercase(Alphabet::P), + [agent_pubkey_hex.as_str()], + ) + .since(nostr::Timestamp::now() - 5u64); + + agent_client + .subscribe(&sid, vec![filter]) + .await + .expect("subscribe to membership notifications"); + + // Drain EOSE — no historical events expected. + agent_client + .collect_until_eose(&sid, Duration::from_secs(5)) + .await + .expect("EOSE for membership sub"); + + let http_client = reqwest::Client::new(); + let owner_pubkey_hex = owner_keys.public_key().to_hex(); + + // Add agent to the channel via REST (owner's X-Pubkey header). + let resp = http_client + .post(format!( + "{}/api/channels/{}/members", + relay_http_url(), + channel_id + )) + .header("X-Pubkey", &owner_pubkey_hex) + .json(&serde_json::json!({ "pubkeys": [agent_pubkey_hex] })) + .send() + .await + .expect("add member request"); + assert!( + resp.status().is_success(), + "add member failed: {}", + resp.status() + ); + + // Consume the kind:44100 add notification before waiting for the remove. + let add_msg = agent_client + .recv_event(Duration::from_secs(5)) + .await + .expect("recv kind:44100 notification"); + match add_msg { + RelayMessage::Event { ref event, .. } => { + assert_eq!( + event.kind, + Kind::Custom(44100), + "expected kind:44100 add notification, got {}", + event.kind.as_u16() + ); + } + other => panic!("expected EVENT kind:44100, got {other:?}"), + } + + // Remove agent from the channel via REST DELETE. + let resp = http_client + .delete(format!( + "{}/api/channels/{}/members/{}", + relay_http_url(), + channel_id, + agent_pubkey_hex + )) + .header("X-Pubkey", &owner_pubkey_hex) + .send() + .await + .expect("remove member request"); + assert!( + resp.status().is_success(), + "remove member failed: {}", + resp.status() + ); + + // Wait for the kind:44101 remove notification. + let msg = agent_client + .recv_event(Duration::from_secs(5)) + .await + .expect("recv kind:44101 notification"); + + match msg { + RelayMessage::Event { event, .. } => { + assert_eq!( + event.kind, + Kind::Custom(44101), + "expected kind:44101, got {}", + event.kind.as_u16() + ); + + let tags: Vec> = event + .tags + .iter() + .map(|t| t.as_slice().iter().map(|s| s.to_string()).collect()) + .collect(); + + let has_p = tags + .iter() + .any(|t| t.len() >= 2 && t[0] == "p" && t[1] == agent_pubkey_hex); + assert!( + has_p, + "kind:44101 missing p tag = agent pubkey. tags: {tags:?}" + ); + + let has_h = tags + .iter() + .any(|t| t.len() >= 2 && t[0] == "h" && t[1] == channel_id); + assert!( + has_h, + "kind:44101 missing h tag = channel uuid. tags: {tags:?}" + ); + } + other => panic!("expected EVENT kind:44101, got {other:?}"), + } + + agent_client.disconnect().await.expect("disconnect"); +} + +/// Subscribing to kind:44100/44101 with `#p` containing BOTH the client's own pubkey AND +/// a victim's pubkey must be rejected with CLOSED. All #p values must match the authenticated +/// pubkey — including the victim's key is not allowed. +#[tokio::test] +#[ignore] +async fn test_membership_notification_multi_p_rejected() { + let url = relay_url(); + + let keys_a = Keys::generate(); + let keys_b = Keys::generate(); + let keys_a_pubkey_hex = keys_a.public_key().to_hex(); + let keys_b_pubkey_hex = keys_b.public_key().to_hex(); + + // Connect as keys_a. + let mut client = SproutTestClient::connect(&url, &keys_a) + .await + .expect("connect as keys_a"); + + let sid = sub_id("multi-p-filter"); + // Filter includes keys_a's own pubkey AND keys_b's (victim) pubkey. + // The relay must reject this because not all #p values match the authenticated pubkey. + let filter = Filter::new() + .kinds(vec![Kind::Custom(44100), Kind::Custom(44101)]) + .custom_tag( + SingleLetterTag::lowercase(Alphabet::P), + [keys_a_pubkey_hex.as_str(), keys_b_pubkey_hex.as_str()], + ); + + client + .subscribe(&sid, vec![filter]) + .await + .expect("send REQ"); + + // Drain until we get the CLOSED for our subscription. + let msg = loop { + let m = client + .recv_event(Duration::from_secs(5)) + .await + .expect("recv CLOSED"); + match &m { + RelayMessage::Eose { .. } => continue, + RelayMessage::Event { .. } => continue, + _ => break m, + } + }; + + match msg { + RelayMessage::Closed { + subscription_id, + message, + } => { + assert_eq!( + subscription_id, sid, + "CLOSED for wrong subscription: {subscription_id}" + ); + assert!( + message.to_lowercase().contains("restricted"), + "expected 'restricted' in CLOSED message, got: {message}" + ); + } + other => panic!("expected CLOSED, got {other:?}"), + } + + client.disconnect().await.expect("disconnect"); +} + +/// A mixed-filter subscription where one filter has `#h` + membership kinds and another +/// filter makes the subscription globally scoped must be rejected with CLOSED. +/// This prevents bypassing the #p requirement via mixed filters. +#[tokio::test] +#[ignore] +async fn test_membership_notification_mixed_filter_rejected() { + let url = relay_url(); + let keys = Keys::generate(); + let channel_id = create_test_channel(&keys).await; + + let mut client = SproutTestClient::connect(&url, &keys) + .await + .expect("connect"); + + let sid = sub_id("mixed-filter"); + // Filter 1: has #h + membership kinds (would skip per-filter #h check) + let filter1 = Filter::new().kinds(vec![Kind::Custom(44100)]).custom_tag( + SingleLetterTag::lowercase(Alphabet::H), + [channel_id.as_str()], + ); + // Filter 2: global filter (no #h) — makes the subscription globally scoped. + // No kinds = wildcard, no #p = should trigger rejection. + let filter2 = Filter::new().authors(vec![keys.public_key()]); + + client + .subscribe(&sid, vec![filter1, filter2]) + .await + .expect("send REQ"); + + // Drain until we get the CLOSED for our subscription. + let msg = loop { + let m = client + .recv_event(Duration::from_secs(5)) + .await + .expect("recv CLOSED"); + match &m { + RelayMessage::Eose { .. } => continue, + RelayMessage::Event { .. } => continue, + _ => break m, + } + }; + + match msg { + RelayMessage::Closed { + subscription_id, + message, + } => { + assert_eq!( + subscription_id, sid, + "CLOSED for wrong subscription: {subscription_id}" + ); + assert!( + message.to_lowercase().contains("restricted"), + "expected 'restricted' in CLOSED message, got: {message}" + ); + } + other => panic!("expected CLOSED, got {other:?}"), + } + + client.disconnect().await.expect("disconnect"); +}