Skip to content

Migrate channel scoping to h tags #22

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
wesbillman merged 2 commits into from
Mar 10, 2026
Merged

Migrate channel scoping to h tags #22

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
299aa3b
Migrate channel tags to h and fix REST fan-out
wesbillman Mar 10, 2026
193223f
Reject channel content events without h tags
wesbillman Mar 10, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 9 additions & 20 deletions crates/sprout-mcp/src/server.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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!(
Expand Down Expand Up @@ -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()],
)
Comment on lines +551 to +554
Copy link

@chatgpt-codex-connector chatgpt-codex-connector bot Mar 10, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Preserve MCP canvas reads for legacy-tagged canvas events

get_canvas now queries only by #h, but older canvas events were written with the previous channel tag format. Since matching is tag-literal, pre-migration canvas documents will no longer be returned, causing MCP to report "No canvas set" even when a canvas already exists. Add a fallback path for legacy tags (or migrate existing canvas events) before removing old-read compatibility.

Useful? React with 👍 / 👎.

.limit(50);

let sub_id = format!("canvas-{}", uuid::Uuid::new_v4());
Expand All @@ -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()
Expand All @@ -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)
{
Expand Down
13 changes: 10 additions & 3 deletions crates/sprout-relay/src/api/messages.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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::{
Expand Down Expand Up @@ -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).
Expand Down Expand Up @@ -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!({
Expand Down
201 changes: 141 additions & 60 deletions crates/sprout-relay/src/handlers/event.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand All @@ -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();
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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"
);
}
Expand Down Expand Up @@ -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" {
Copy link

@chatgpt-codex-connector chatgpt-codex-connector bot Mar 10, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Reject legacy channel-tagged events instead of treating them global

This change makes extract_channel_id accept only h, but handle_event enforces membership and archived-channel checks only when a channel_id is found. Any client still emitting the previous channel tag format (used before this migration) will now be stored as an unscoped global event, which bypasses private-channel authorization and archived-channel guards rather than failing closed. Please keep legacy parsing during rollout or explicitly reject channel message kinds that lack h.

Useful? React with 👍 / 👎.

if let Some(val) = tag.content() {
if let Ok(id) = val.parse::<uuid::Uuid>() {
return Some(id);
Expand All @@ -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"
);
}
}
6 changes: 3 additions & 3 deletions crates/sprout-relay/src/handlers/req.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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;
Expand Down Expand Up @@ -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()],
)
}
Expand Down
2 changes: 1 addition & 1 deletion crates/sprout-relay/src/handlers/side_effects.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
Loading