Migrate channel scoping to h tags (#22)

Author: wesbillman

crates/sprout-mcp/src/server.rs

@@ -484,8 +484,8 @@ impl SproutMcpServer {
         const MAX_HISTORY_LIMIT: u32 = 200;
         let limit = p.limit.unwrap_or(50).min(MAX_HISTORY_LIMIT);
 
-        // Always use the REST endpoint — the channel tag is multi-character ("channel")
-        // and cannot be filtered via WebSocket subscription SingleLetterTag filters.
+        // Use the REST endpoint so callers get the canonical history payload,
+        // including thread metadata when requested.
         let with_threads = p.with_threads.unwrap_or(false);
         let path = if with_threads {
             format!(
@@ -546,11 +546,12 @@ impl SproutMcpServer {
             return format!("Error: {e}");
         }
 
-        // The "channel" tag is multi-character and cannot be used in WebSocket
-        // subscription filters (nostr::Filter::custom_tag only accepts SingleLetterTag).
-        // Subscribe to all KIND_CANVAS events and filter client-side by channel tag.
         let filter = nostr::Filter::new()
             .kind(nostr::Kind::Custom(KIND_CANVAS as u16))
+            .custom_tag(
+                nostr::SingleLetterTag::lowercase(nostr::Alphabet::H),
+                [p.channel_id.as_str()],
+            )
             .limit(50);
 
         let sub_id = format!("canvas-{}", uuid::Uuid::new_v4());
@@ -560,15 +561,7 @@ impl SproutMcpServer {
         };
         let _ = self.client.close_subscription(&sub_id).await;
 
-        // Filter client-side: find the most recent canvas event for this channel.
-        let canvas_event = events.iter().rev().find(|event| {
-            event
-                .tags
-                .find(nostr::TagKind::custom("channel"))
-                .and_then(|t| t.content())
-                .map(|v| v == p.channel_id.as_str())
-                .unwrap_or(false)
-        });
+        let canvas_event = events.iter().max_by_key(|event| event.created_at.as_u64());
 
         if let Some(event) = canvas_event {
             event.content.clone()
@@ -589,19 +582,15 @@ impl SproutMcpServer {
 
         let keys = self.client.keys().clone();
 
-        let channel_tag = match nostr::Tag::parse(&["channel", &p.channel_id]) {
+        let channel_tag = match nostr::Tag::parse(&["h", &p.channel_id]) {
             Ok(t) => t,
             Err(e) => return format!("Error building tag: {e}"),
         };
-        let event_ref_tag = match nostr::Tag::parse(&["e", &p.channel_id]) {
-            Ok(t) => t,
-            Err(e) => return format!("Error building event-ref tag: {e}"),
-        };
 
         let event = match nostr::EventBuilder::new(
             nostr::Kind::Custom(KIND_CANVAS as u16),
             &p.content,
-            [channel_tag, event_ref_tag],
+            [channel_tag],
         )
         .sign_with_keys(&keys)
         {

crates/sprout-relay/src/api/messages.rs

@@ -27,6 +27,7 @@ use nostr::util::hex as nostr_hex;
 use nostr::{EventBuilder, Kind, Tag};
 use serde::Deserialize;
 
+use crate::handlers::event::dispatch_persistent_event;
 use crate::state::AppState;
 
 use super::{
@@ -237,8 +238,9 @@ pub async fn send_message(
         // Attribution to the actual sender.
         Tag::parse(&["p", &user_pubkey_hex])
             .map_err(|e| internal_error(&format!("tag build error: {e}")))?,
-        // Channel tag so Nostr clients can find this event by channel.
-        Tag::custom(nostr::TagKind::custom("channel"), [channel_id.to_string()]),
+        // Channel-scoped messages use the NIP-29 `h` tag.
+        Tag::parse(&["h", &channel_id.to_string()])
+            .map_err(|e| internal_error(&format!("tag build error: {e}")))?,
     ];
 
     // Thread reply tags (NIP-10 style).
@@ -297,12 +299,17 @@ pub async fn send_message(
         broadcast: body.broadcast_to_channel,
     });
 
-    state
+    let (stored_event, was_inserted) = state
         .db
         .insert_event_with_thread_metadata(&event, Some(channel_id), thread_meta)
         .await
         .map_err(|e| internal_error(&format!("db error: {e}")))?;
 
+    if was_inserted {
+        let kind_u32 = u32::from(event.kind.as_u16());
+        let _ = dispatch_persistent_event(&state, &stored_event, kind_u32, &user_pubkey_hex).await;
+    }
+
     // ── Response ──────────────────────────────────────────────────────────────
 
     Ok(Json(serde_json::json!({

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

@@ -9,7 +9,11 @@ use nostr::Event;
 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_PRESENCE_UPDATE,
+    event_kind_u32, is_ephemeral, is_workflow_execution_kind, KIND_AUTH, KIND_CANVAS,
+    KIND_FORUM_COMMENT, KIND_FORUM_POST, KIND_FORUM_VOTE, KIND_PRESENCE_UPDATE,
+    KIND_STREAM_MESSAGE, KIND_STREAM_MESSAGE_BOOKMARKED, KIND_STREAM_MESSAGE_EDIT,
+    KIND_STREAM_MESSAGE_PINNED, KIND_STREAM_MESSAGE_SCHEDULED, KIND_STREAM_MESSAGE_V2,
+    KIND_STREAM_REMINDER,
 };
 use sprout_core::verification::verify_event;
 
@@ -19,6 +23,75 @@ use crate::connection::{AuthState, ConnectionState};
 use crate::protocol::RelayMessage;
 use crate::state::AppState;
 
+/// Publish a stored event to subscribers and kick off async side effects.
+pub(crate) async fn dispatch_persistent_event(
+    state: &Arc<AppState>,
+    stored_event: &StoredEvent,
+    kind_u32: u32,
+    actor_pubkey_hex: &str,
+) -> usize {
+    let event_id_hex = stored_event.event.id.to_hex();
+
+    if let Some(ch_id) = stored_event.channel_id {
+        if let Err(e) = state.pubsub.publish_event(ch_id, &stored_event.event).await {
+            warn!(event_id = %event_id_hex, "Redis publish failed: {e}");
+        }
+    }
+
+    let matches = state.sub_registry.fan_out(stored_event);
+    debug!(
+        event_id = %event_id_hex,
+        channel_id = ?stored_event.channel_id,
+        match_count = matches.len(),
+        "Fan-out"
+    );
+
+    let event_json = serde_json::to_string(&stored_event.event)
+        .expect("nostr::Event serialization is infallible for well-formed events");
+    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);
+    }
+
+    let search = Arc::clone(&state.search);
+    let stored_for_search = stored_event.clone();
+    tokio::spawn(async move {
+        if let Err(e) = search.index_event(&stored_for_search).await {
+            error!(event_id = %stored_for_search.event.id.to_hex(), "Search index failed: {e}");
+        }
+    });
+
+    let audit = Arc::clone(&state.audit);
+    let audit_event_id = event_id_hex.clone();
+    let audit_actor_pubkey = actor_pubkey_hex.to_string();
+    let audit_channel_id = stored_event.channel_id;
+    tokio::spawn(async move {
+        let entry = NewAuditEntry {
+            event_id: audit_event_id.clone(),
+            event_kind: kind_u32,
+            actor_pubkey: audit_actor_pubkey,
+            action: AuditAction::EventCreated,
+            channel_id: audit_channel_id,
+            metadata: serde_json::Value::Null,
+        };
+        if let Err(e) = audit.log(entry).await {
+            error!(event_id = %audit_event_id, "Audit log failed: {e}");
+        }
+    });
+
+    if !is_workflow_execution_kind(kind_u32) {
+        let workflow_engine = Arc::clone(&state.workflow_engine);
+        let workflow_event = stored_event.clone();
+        tokio::spawn(async move {
+            if let Err(e) = workflow_engine.on_event(&workflow_event).await {
+                tracing::error!(event_id = ?workflow_event.event.id, "Workflow trigger failed: {e}");
+            }
+        });
+    }
+
+    matches.len()
+}
+
 /// Handle an EVENT message: authenticate, verify, store, fan-out, index, and audit the event.
 pub async fn handle_event(event: Event, conn: Arc<ConnectionState>, state: Arc<AppState>) {
     let event_id_hex = event.id.to_hex();
@@ -169,6 +242,15 @@ pub async fn handle_event(event: Event, conn: Arc<ConnectionState>, state: Arc<A
         extract_channel_id(&event)
     };
 
+    if requires_h_channel_scope(kind_u32) && channel_id.is_none() {
+        conn.send(RelayMessage::ok(
+            &event_id_hex,
+            false,
+            "invalid: channel-scoped events must include an h tag",
+        ));
+        return;
+    }
+
     if let Some(ch_id) = channel_id {
         if let Err(msg) =
             check_channel_membership(&state, ch_id, &pubkey_bytes, conn_id, &event_id_hex).await
@@ -248,70 +330,15 @@ pub async fn handle_event(event: Event, conn: Arc<ConnectionState>, state: Arc<A
         }
     }
 
-    if let Some(ch_id) = channel_id {
-        if let Err(e) = state.pubsub.publish_event(ch_id, &event).await {
-            warn!(event_id = %event_id_hex, "Redis publish failed: {e}");
-        }
-    }
-
-    let matches = state.sub_registry.fan_out(&stored_event);
-    debug!(
-        event_id = %event_id_hex,
-        channel_id = ?stored_event.channel_id,
-        match_count = matches.len(),
-        "Fan-out"
-    );
-    let event_json = serde_json::to_string(&stored_event.event)
-        .expect("nostr::Event serialization is infallible for well-formed events");
-    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);
-    }
-
-    let search = Arc::clone(&state.search);
-    let stored_for_search = stored_event.clone();
-    tokio::spawn(async move {
-        if let Err(e) = search.index_event(&stored_for_search).await {
-            error!(event_id = %stored_for_search.event.id.to_hex(), "Search index failed: {e}");
-        }
-    });
-
-    let audit = Arc::clone(&state.audit);
-    let audit_event_id = event_id_hex.clone();
-    let audit_pubkey = pubkey_hex.clone();
-    tokio::spawn(async move {
-        let entry = NewAuditEntry {
-            event_id: audit_event_id.clone(),
-            event_kind: kind_u32,
-            actor_pubkey: audit_pubkey,
-            action: AuditAction::EventCreated,
-            channel_id,
-            metadata: serde_json::Value::Null,
-        };
-        if let Err(e) = audit.log(entry).await {
-            error!(event_id = %audit_event_id, "Audit log failed: {e}");
-        }
-    });
-
-    // Don't trigger workflows for workflow execution events (prevents infinite loops).
-    let is_workflow_event = is_workflow_execution_kind(kind_u32);
-    if !is_workflow_event {
-        let wf = Arc::clone(&state.workflow_engine);
-        let ev = stored_event.clone();
-        tokio::spawn(async move {
-            if let Err(e) = wf.on_event(&ev).await {
-                tracing::error!(event_id = ?ev.event.id, "Workflow trigger failed: {e}");
-            }
-        });
-    }
+    let fan_out = dispatch_persistent_event(&state, &stored_event, kind_u32, &pubkey_hex).await;
 
     conn.send(RelayMessage::ok(&event_id_hex, true, ""));
 
     info!(
         event_id = %event_id_hex,
         kind = kind_u32,
         conn_id = %conn_id,
-        fan_out = matches.len(),
+        fan_out,
         "Event ingested"
     );
 }
@@ -538,12 +565,12 @@ async fn derive_reaction_channel(
 
 /// Extract a channel UUID from event tags.
 ///
-/// Checks `"channel"` custom tags and `"h"` NIP-29 group tags for a channel UUID.
+/// Checks the `"h"` NIP-29 group tag for a channel UUID.
 /// The `"e"` tag is intentionally NOT checked — it is reserved for event references only.
 fn extract_channel_id(event: &Event) -> Option<uuid::Uuid> {
     for tag in event.tags.iter() {
         let key = tag.kind().to_string();
-        if key == "channel" || key == "h" {
+        if key == "h" {
             if let Some(val) = tag.content() {
                 if let Ok(id) = val.parse::<uuid::Uuid>() {
                     return Some(id);
@@ -553,3 +580,57 @@ fn extract_channel_id(event: &Event) -> Option<uuid::Uuid> {
     }
     None
 }
+
+fn requires_h_channel_scope(kind: u32) -> bool {
+    matches!(
+        kind,
+        KIND_STREAM_MESSAGE
+            | KIND_STREAM_MESSAGE_V2
+            | KIND_STREAM_MESSAGE_EDIT
+            | KIND_STREAM_MESSAGE_PINNED
+            | KIND_STREAM_MESSAGE_BOOKMARKED
+            | KIND_STREAM_MESSAGE_SCHEDULED
+            | KIND_STREAM_REMINDER
+            | KIND_CANVAS
+            | KIND_FORUM_POST
+            | KIND_FORUM_VOTE
+            | KIND_FORUM_COMMENT
+    )
+}
+
+#[cfg(test)]
+mod tests {
+    use super::requires_h_channel_scope;
+    use sprout_core::kind::{
+        KIND_CANVAS, KIND_FORUM_COMMENT, KIND_FORUM_POST, KIND_FORUM_VOTE, KIND_PRESENCE_UPDATE,
+        KIND_STREAM_MESSAGE,
+    };
+
+    #[test]
+    fn channel_scoped_content_kinds_require_h_tags() {
+        for kind in [
+            KIND_STREAM_MESSAGE,
+            KIND_CANVAS,
+            KIND_FORUM_POST,
+            KIND_FORUM_VOTE,
+            KIND_FORUM_COMMENT,
+        ] {
+            assert!(
+                requires_h_channel_scope(kind),
+                "kind {kind} should require h"
+            );
+        }
+    }
+
+    #[test]
+    fn non_channel_kinds_do_not_require_h_tags() {
+        assert!(
+            !requires_h_channel_scope(nostr::Kind::Reaction.as_u16().into()),
+            "reactions derive channel from the target event"
+        );
+        assert!(
+            !requires_h_channel_scope(KIND_PRESENCE_UPDATE),
+            "presence updates are global/ephemeral"
+        );
+    }
+}

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

@@ -195,7 +195,7 @@ fn filter_to_query_params(filter: &Filter, channel_id: Option<uuid::Uuid>) -> Ev
 /// Extract a single channel UUID from filter generic tags, or `None` if the
 /// subscription is logically global.
 ///
-/// Checks both `"channel"` and `"e"` tag keys — clients use `#e` with a UUID value.
+/// Checks the `"h"` tag key — channel-scoped subscriptions use `#h = <uuid>`.
 ///
 /// Returns `None` when:
 /// - Any filter has no channel tag (that filter matches all channels → global sub), or
@@ -208,7 +208,7 @@ fn extract_channel_id_from_filters(filters: &[Filter]) -> Option<uuid::Uuid> {
         let mut filter_has_channel = false;
         for (tag_key, tag_values) in f.generic_tags.iter() {
             let key = tag_key.to_string();
-            if key == "channel" || key == "e" {
+            if key == "h" {
                 for val in tag_values {
                     if let Ok(id) = val.parse::<uuid::Uuid>() {
                         filter_has_channel = true;
@@ -238,7 +238,7 @@ mod tests {
 
     fn filter_with_channel(channel_id: uuid::Uuid) -> Filter {
         Filter::new().custom_tag(
-            SingleLetterTag::lowercase(Alphabet::E),
+            SingleLetterTag::lowercase(Alphabet::H),
             [channel_id.to_string()],
         )
     }

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

@@ -233,7 +233,7 @@ pub async fn emit_system_message(
     channel_id: Uuid,
     content: serde_json::Value,
 ) -> anyhow::Result<()> {
-    let channel_tag = Tag::custom(nostr::TagKind::custom("channel"), [channel_id.to_string()]);
+    let channel_tag = Tag::parse(&["h", &channel_id.to_string()])?;
 
     let event = EventBuilder::new(Kind::Custom(40099), content.to_string(), [channel_tag])
         .sign_with_keys(&state.relay_keypair)