mirror of
https://github.com/block/buzz.git
synced 2026-10-27 04:30:52 +00:00
Migrate channel scoping to h tags (#22)
This commit is contained in:
@@ -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)
|
||||
{
|
||||
|
||||
@@ -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!({
|
||||
|
||||
@@ -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,62 +330,7 @@ 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, ""));
|
||||
|
||||
@@ -311,7 +338,7 @@ pub async fn handle_event(event: Event, conn: Arc<ConnectionState>, state: Arc<A
|
||||
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"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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()],
|
||||
)
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -159,9 +159,9 @@ impl SproutTestClient {
|
||||
content: &str,
|
||||
kind: u16,
|
||||
) -> Result<OkResponse, TestClientError> {
|
||||
let e_tag = Tag::parse(&["e", channel_id])
|
||||
let h_tag = Tag::parse(&["h", channel_id])
|
||||
.map_err(|e| TestClientError::EventBuilder(e.to_string()))?;
|
||||
let event = EventBuilder::new(Kind::Custom(kind), content, [e_tag]).sign_with_keys(keys)?;
|
||||
let event = EventBuilder::new(Kind::Custom(kind), content, [h_tag]).sign_with_keys(keys)?;
|
||||
self.send_event(event).await
|
||||
}
|
||||
|
||||
@@ -534,11 +534,11 @@ mod tests {
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn text_event_carries_e_tag() {
|
||||
fn text_event_carries_h_tag() {
|
||||
let keys = Keys::generate();
|
||||
let channel_id = "my-channel-123";
|
||||
let e_tag = Tag::parse(&["e", channel_id]).unwrap();
|
||||
let event = EventBuilder::new(Kind::Custom(40001), "hello", [e_tag])
|
||||
let h_tag = Tag::parse(&["h", channel_id]).unwrap();
|
||||
let event = EventBuilder::new(Kind::Custom(40001), "hello", [h_tag])
|
||||
.sign_with_keys(&keys)
|
||||
.unwrap();
|
||||
|
||||
@@ -551,8 +551,8 @@ mod tests {
|
||||
|
||||
assert!(
|
||||
tags.iter()
|
||||
.any(|t| t.len() >= 2 && t[0] == "e" && t[1] == channel_id),
|
||||
"missing e tag"
|
||||
.any(|t| t.len() >= 2 && t[0] == "h" && t[1] == channel_id),
|
||||
"missing h tag"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -65,7 +65,7 @@ async fn test_send_event_and_receive_via_subscription() {
|
||||
let sid = sub_id("send-recv");
|
||||
let filter = Filter::new()
|
||||
.kind(Kind::Custom(kind))
|
||||
.custom_tag(SingleLetterTag::lowercase(Alphabet::E), [channel.as_str()]);
|
||||
.custom_tag(SingleLetterTag::lowercase(Alphabet::H), [channel.as_str()]);
|
||||
|
||||
client_a
|
||||
.subscribe(&sid, vec![filter])
|
||||
@@ -124,7 +124,7 @@ async fn test_subscription_filters_by_kind() {
|
||||
let sid = sub_id("filter-kind");
|
||||
let filter = Filter::new()
|
||||
.kind(Kind::Custom(target_kind))
|
||||
.custom_tag(SingleLetterTag::lowercase(Alphabet::E), [channel.as_str()]);
|
||||
.custom_tag(SingleLetterTag::lowercase(Alphabet::H), [channel.as_str()]);
|
||||
|
||||
client
|
||||
.subscribe(&sid, vec![filter])
|
||||
@@ -194,7 +194,7 @@ async fn test_close_subscription_stops_delivery() {
|
||||
let sid = sub_id("close-sub");
|
||||
let filter = Filter::new()
|
||||
.kind(Kind::Custom(kind))
|
||||
.custom_tag(SingleLetterTag::lowercase(Alphabet::E), [channel.as_str()]);
|
||||
.custom_tag(SingleLetterTag::lowercase(Alphabet::H), [channel.as_str()]);
|
||||
|
||||
client
|
||||
.subscribe(&sid, vec![filter])
|
||||
@@ -287,7 +287,7 @@ async fn test_multiple_concurrent_clients() {
|
||||
|
||||
let filter = Filter::new()
|
||||
.kind(Kind::Custom(kind))
|
||||
.custom_tag(SingleLetterTag::lowercase(Alphabet::E), [channel.as_str()]);
|
||||
.custom_tag(SingleLetterTag::lowercase(Alphabet::H), [channel.as_str()]);
|
||||
|
||||
for (i, client) in clients.iter_mut().enumerate() {
|
||||
let sid = format!("multi-{i}");
|
||||
@@ -350,7 +350,7 @@ async fn test_stored_events_returned_before_eose() {
|
||||
let sid = sub_id("stored");
|
||||
let filter = Filter::new()
|
||||
.kind(Kind::Custom(kind))
|
||||
.custom_tag(SingleLetterTag::lowercase(Alphabet::E), [channel.as_str()]);
|
||||
.custom_tag(SingleLetterTag::lowercase(Alphabet::H), [channel.as_str()]);
|
||||
|
||||
client
|
||||
.subscribe(&sid, vec![filter])
|
||||
@@ -397,7 +397,7 @@ async fn test_ephemeral_event_not_stored() {
|
||||
let sid = sub_id("ephemeral");
|
||||
let filter = Filter::new()
|
||||
.kind(Kind::Custom(ephemeral_kind))
|
||||
.custom_tag(SingleLetterTag::lowercase(Alphabet::E), [channel.as_str()]);
|
||||
.custom_tag(SingleLetterTag::lowercase(Alphabet::H), [channel.as_str()]);
|
||||
|
||||
client
|
||||
.subscribe(&sid, vec![filter])
|
||||
@@ -603,7 +603,7 @@ async fn test_eose_sent_for_empty_subscription() {
|
||||
let sid = sub_id("empty-eose");
|
||||
let filter = Filter::new()
|
||||
.kind(Kind::Custom(kind))
|
||||
.custom_tag(SingleLetterTag::lowercase(Alphabet::E), [channel.as_str()])
|
||||
.custom_tag(SingleLetterTag::lowercase(Alphabet::H), [channel.as_str()])
|
||||
.since(nostr::Timestamp::now());
|
||||
|
||||
client
|
||||
|
||||
@@ -27,9 +27,9 @@
|
||||
|
||||
use std::time::Duration;
|
||||
|
||||
use nostr::{Keys, Kind, Tag};
|
||||
use nostr::{Alphabet, Filter, Keys, Kind, SingleLetterTag, Tag};
|
||||
use reqwest::Client;
|
||||
use sprout_test_client::SproutTestClient;
|
||||
use sprout_test_client::{RelayMessage, SproutTestClient};
|
||||
|
||||
// ── Helpers ───────────────────────────────────────────────────────────────────
|
||||
|
||||
@@ -225,6 +225,83 @@ async fn test_channel_visibility_open_channels_visible_to_all() {
|
||||
);
|
||||
}
|
||||
|
||||
/// REST-created channel messages must fan out to WebSocket subscribers and
|
||||
/// carry the canonical channel `h` tag.
|
||||
#[tokio::test]
|
||||
#[ignore]
|
||||
async fn test_rest_send_message_reaches_websocket_channel_subscriptions() {
|
||||
let client = http_client();
|
||||
let subscriber_keys = Keys::generate();
|
||||
let poster_keys = Keys::generate();
|
||||
let ws_url = relay_ws_url();
|
||||
|
||||
let mut subscriber = SproutTestClient::connect(&ws_url, &subscriber_keys)
|
||||
.await
|
||||
.expect("WebSocket connect failed");
|
||||
|
||||
let sid = format!("rest-live-{}", uuid::Uuid::new_v4().simple());
|
||||
let filter = Filter::new()
|
||||
.kind(Kind::Custom(40001))
|
||||
.custom_tag(SingleLetterTag::lowercase(Alphabet::H), [CHANNEL_GENERAL]);
|
||||
|
||||
subscriber
|
||||
.subscribe(&sid, vec![filter])
|
||||
.await
|
||||
.expect("subscribe failed");
|
||||
subscriber
|
||||
.collect_until_eose(&sid, Duration::from_secs(5))
|
||||
.await
|
||||
.expect("EOSE failed");
|
||||
|
||||
let content = format!("E2E REST live message: {}", uuid::Uuid::new_v4().simple());
|
||||
let url = format!(
|
||||
"{}/api/channels/{}/messages",
|
||||
relay_http_url(),
|
||||
CHANNEL_GENERAL
|
||||
);
|
||||
let resp = authed_post_json(
|
||||
&client,
|
||||
&url,
|
||||
&poster_keys.public_key().to_hex(),
|
||||
serde_json::json!({ "content": content }),
|
||||
)
|
||||
.await;
|
||||
|
||||
assert_eq!(resp.status(), 200, "expected 200 OK from REST send_message");
|
||||
|
||||
let message = subscriber
|
||||
.recv_event(Duration::from_secs(5))
|
||||
.await
|
||||
.expect("subscriber did not receive live event");
|
||||
|
||||
match message {
|
||||
RelayMessage::Event { event, .. } => {
|
||||
assert_eq!(event.content, content);
|
||||
|
||||
let tags: Vec<Vec<String>> = event
|
||||
.tags
|
||||
.iter()
|
||||
.map(|tag| tag.as_slice().iter().map(|part| part.to_string()).collect())
|
||||
.collect();
|
||||
|
||||
assert!(
|
||||
tags.iter()
|
||||
.any(|tag| tag.len() >= 2 && tag[0] == "h" && tag[1] == CHANNEL_GENERAL),
|
||||
"REST-created message is missing the channel h tag: {tags:?}"
|
||||
);
|
||||
assert!(
|
||||
tags.iter().any(|tag| {
|
||||
tag.len() >= 2 && tag[0] == "p" && tag[1] == poster_keys.public_key().to_hex()
|
||||
}),
|
||||
"REST-created message is missing the sender attribution p tag: {tags:?}"
|
||||
);
|
||||
}
|
||||
other => panic!("expected live EVENT after REST send_message, got {other:?}"),
|
||||
}
|
||||
|
||||
subscriber.disconnect().await.expect("disconnect failed");
|
||||
}
|
||||
|
||||
/// GET /api/channels requires authentication — unauthenticated requests are rejected.
|
||||
#[tokio::test]
|
||||
#[ignore]
|
||||
|
||||
@@ -32,7 +32,7 @@ function createSearchAnchorEvent(hit: SearchHit): RelayEvent {
|
||||
pubkey: hit.pubkey,
|
||||
created_at: hit.createdAt,
|
||||
kind: hit.kind,
|
||||
tags: [["e", hit.channelId]],
|
||||
tags: [["h", hit.channelId]],
|
||||
content: hit.content,
|
||||
sig: "",
|
||||
};
|
||||
|
||||
@@ -35,10 +35,7 @@ function createOptimisticMessage(
|
||||
pubkey: identity.pubkey,
|
||||
created_at: Math.floor(Date.now() / 1_000),
|
||||
kind: 4_0001,
|
||||
tags: [
|
||||
["channel", channelId],
|
||||
["e", channelId],
|
||||
],
|
||||
tags: [["h", channelId]],
|
||||
content,
|
||||
sig: "",
|
||||
pending: true,
|
||||
|
||||
@@ -9,7 +9,7 @@ import type { RelayEvent } from "@/shared/api/types";
|
||||
|
||||
type RelaySubscriptionFilter = {
|
||||
kinds: number[];
|
||||
"#e": string[];
|
||||
"#h": string[];
|
||||
limit: number;
|
||||
};
|
||||
|
||||
@@ -122,13 +122,8 @@ class RelayClient {
|
||||
const event = await signRelayEvent({
|
||||
kind: 40001,
|
||||
content: content.trim(),
|
||||
// Include both tags for now:
|
||||
// - `channel` lets the relay persist and authorize the event as channel-scoped
|
||||
// - `e` keeps the existing desktop WebSocket history/subscription filters working
|
||||
tags: [
|
||||
["channel", channelId],
|
||||
["e", channelId],
|
||||
],
|
||||
// Channel-scoped events use the NIP-29 `h` tag.
|
||||
tags: [["h", channelId]],
|
||||
});
|
||||
|
||||
return new Promise<RelayEvent>((resolve, reject) => {
|
||||
@@ -249,7 +244,7 @@ class RelayClient {
|
||||
): RelaySubscriptionFilter {
|
||||
return {
|
||||
kinds: [40001],
|
||||
"#e": [channelId],
|
||||
"#h": [channelId],
|
||||
limit,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -230,7 +230,7 @@ function sendWsClose(handler: WsHandler) {
|
||||
}
|
||||
|
||||
function getChannelIdFromTags(tags: string[][]): string | undefined {
|
||||
return tags.find((tag) => tag[0] === "e")?.[1];
|
||||
return tags.find((tag) => tag[0] === "h")?.[1];
|
||||
}
|
||||
|
||||
function getMockMessageStore(channelId: string): RelayEvent[] {
|
||||
@@ -247,7 +247,7 @@ function getMockMessageStore(channelId: string): RelayEvent[] {
|
||||
pubkey: DEFAULT_MOCK_IDENTITY.pubkey,
|
||||
created_at: Math.floor(Date.now() / 1000),
|
||||
kind: 40001,
|
||||
tags: [["e", channelId]],
|
||||
tags: [["h", channelId]],
|
||||
content: "Welcome to #general",
|
||||
sig: "mocksig".repeat(20).slice(0, 128),
|
||||
},
|
||||
@@ -744,8 +744,8 @@ function sendToMockSocket(args: {
|
||||
|
||||
if (type === "REQ") {
|
||||
const subId = rest[0] as string;
|
||||
const filter = rest[1] as { "#e"?: string[] };
|
||||
const channelId = filter["#e"]?.[0];
|
||||
const filter = rest[1] as { "#h"?: string[] };
|
||||
const channelId = filter["#h"]?.[0];
|
||||
if (!channelId) {
|
||||
sendWsText(socket.handler, ["EOSE", subId]);
|
||||
return;
|
||||
|
||||
Reference in New Issue
Block a user