External/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-08-18 06:50:31 +02:00
Code Issues Packages Projects Releases Wiki Activity

feat: event-driven channel discovery via membership notifications (kind:44100/44101) (#65)

Browse Source
This commit is contained in:
tlongwell-block
2026-03-14 23:40:06 -04:00
committed by GitHub
parent 3c68325c67
commit 9ca60737eb
13 changed files with 1157 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/sprout-acp/src/config.rs
+90 -1
View File
@@ -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<ChannelFilter> {
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::<Uuid>().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<Vec<u32>> = 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 {
crates/sprout-acp/src/main.rs
+65 -1
View File
@@ -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<Uuid, u64> = 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;
crates/sprout-acp/src/relay.rs
+154 -1
View File
@@ -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<String>,
/// Per-channel filter used on subscribe (for resubscribe after reconnect).
active_filters: HashMap<Uuid, ChannelFilter>,
/// 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<u64>,
/// Newest successfully-enqueued membership notification timestamp.
/// Used as the `since` for reconnect replay when no events were dropped.
membership_last_seen: Option<u64>,
/// 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<u64>,
) -> 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<Uuid> {
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::<Uuid>().ok()
} else {
None
}
})
}
/// Build and send a NIP-42 AUTH response event.
async fn send_auth_response(
ws: &mut WsStream,
crates/sprout-core/src/kind.rs
+10
View File
@@ -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,
crates/sprout-relay/src/api/channels.rs
+11
View File
@@ -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,
crates/sprout-relay/src/api/members.rs
+57
View File
@@ -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 })))
}
crates/sprout-relay/src/config.rs
+1 -1
View File
@@ -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()
crates/sprout-relay/src/handlers/event.rs
+12 -1
View File
@@ -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<ConnectionState>, state: Arc<A
return;
}
// Membership notification events are relay-signed only — reject client submissions.
if kind_u32 == KIND_MEMBER_ADDED_NOTIFICATION || kind_u32 == KIND_MEMBER_REMOVED_NOTIFICATION {
conn.send(RelayMessage::ok(
&event_id_hex,
false,
"invalid: membership notifications are relay-signed only",
));
return;
}
if is_ephemeral(kind_u32) {
handle_ephemeral_event(
event,
crates/sprout-relay/src/handlers/req.rs
+41 -1
View File
@@ -5,8 +5,10 @@ use std::sync::Arc;
use tracing::{debug, warn};
use hex;
use nostr::Filter;
use sprout_core::filter::filters_match;
use sprout_core::kind::{KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION};
use sprout_db::EventQuery;
use sprout_auth::Scope;
@@ -16,7 +18,7 @@ use crate::protocol::RelayMessage;
use crate::state::AppState;
const MAX_HISTORICAL_LIMIT: i64 = 500;
const MAX_SUBSCRIPTIONS: usize = 100;
const MAX_SUBSCRIPTIONS: usize = 1024;
/// Handle a REQ message: register the subscription, deliver historical events, then send EOSE.
pub async fn handle_req(
@@ -75,6 +77,44 @@ pub async fn handle_req(
let channel_id = extract_channel_id_from_filters(&filters);
// Enforce #p filter for membership notification subscriptions.
//
// Only applies to GLOBAL subscriptions (channel_id = None). Channel-scoped
// subscriptions can never receive globally-stored membership events — the
// fan_out() invariant in subscription.rs prevents it.
//
// We use the resolved subscription scope (channel_id) rather than per-filter
// #h presence to prevent mixed-filter bypass: a client could send
// [{#h:..., kinds:[44100]}, {authors:[...]}] which resolves to global scope
// but would skip the #p check if we only looked at per-filter #h tags.
if channel_id.is_none() {
let authed_pubkey_hex = hex::encode(&pubkey_bytes);
for filter in &filters {
let can_match_membership = filter.kinds.as_ref().is_none_or(|ks| {
ks.iter().any(|k| {
let ku = k.as_u16() as u32;
ku == KIND_MEMBER_ADDED_NOTIFICATION || ku == KIND_MEMBER_REMOVED_NOTIFICATION
})
});
if can_match_membership {
let p_tag = nostr::SingleLetterTag::lowercase(nostr::Alphabet::P);
// ALL #p values must match the authenticated pubkey — prevents
// a client from sneaking in a victim's pubkey alongside their own
// (e.g. #p:[my_key, victim_key]) to receive the victim's notifications.
let has_matching_p = filter.generic_tags.get(&p_tag).is_some_and(|values| {
!values.is_empty() && values.iter().all(|v| *v == authed_pubkey_hex)
});
if !has_matching_p {
conn.send(RelayMessage::closed(
&sub_id,
"restricted: membership notifications require #p matching your pubkey",
));
return;
}
}
}
}
// Check channel access BEFORE registering the subscription.
// Registering first would allow non-members to receive live fan-out events
// from private channels before the access check fires.
crates/sprout-relay/src/handlers/side_effects.rs
+127 -2
View File
@@ -7,8 +7,8 @@ use tracing::{info, warn};
use uuid::Uuid;
use sprout_core::kind::{
event_kind_u32, KIND_NIP29_GROUP_ADMINS, KIND_NIP29_GROUP_MEMBERS, KIND_NIP29_GROUP_METADATA,
KIND_REACTION,
event_kind_u32, KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION,
KIND_NIP29_GROUP_ADMINS, KIND_NIP29_GROUP_MEMBERS, KIND_NIP29_GROUP_METADATA, KIND_REACTION,
};
use sprout_db::channel::MemberRole;
@@ -324,6 +324,83 @@ pub async fn emit_system_message(
Ok(())
}
/// Emit a relay-signed membership notification event stored globally (channel_id = None).
///
/// kind:44100 = member added, kind:44101 = member removed.
/// The p tag addresses the target pubkey; the h tag carries the channel UUID as metadata.
/// Stored with channel_id = None so global subscribers receive it via slow-path fan-out.
pub async fn emit_membership_notification(
state: &Arc<AppState>,
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<AppState>,
@@ -538,6 +615,18 @@ async fn handle_put_user(event: &Event, state: &Arc<AppState>) -> 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<AppState>) -> 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<AppState>) -> 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<AppState>) -> 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(())
}
crates/sprout-relay/src/nip11.rs
+1 -1
View File
@@ -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),
crates/sprout-relay/src/subscription.rs
+14 -7
View File
@@ -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
}
crates/sprout-test-client/tests/e2e_relay.rs
+574
View File
@@ -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<Vec<String>> = 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<Vec<String>> = 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");
}