From ab244a17deaf1ffa954c23ee65e32e407607c893 Mon Sep 17 00:00:00 2001 From: ScaleLeanChris Date: Wed, 29 Jul 2026 23:49:41 -0600 Subject: [PATCH] fix(relay): emit participant-addressed DM discovery Signed-off-by: ScaleLeanChris --- .../src/handlers/command_executor.rs | 40 +++- .../buzz-relay/src/handlers/side_effects.rs | 60 +++++- .../tests/e2e_nostr_interop.rs | 179 +++++++++++++++++- 3 files changed, 271 insertions(+), 8 deletions(-) diff --git a/crates/buzz-relay/src/handlers/command_executor.rs b/crates/buzz-relay/src/handlers/command_executor.rs index 2d82736807..861cb0fa12 100644 --- a/crates/buzz-relay/src/handlers/command_executor.rs +++ b/crates/buzz-relay/src/handlers/command_executor.rs @@ -28,8 +28,8 @@ use crate::webhook_secret; use super::ingest::{extract_channel_id, IngestAuth, IngestError, IngestResult}; use super::side_effects::{ - emit_group_discovery_events, emit_membership_notification, emit_system_message, - publish_dm_visibility_snapshot, + emit_dm_created, emit_group_discovery_events, emit_membership_notification, + emit_system_message, publish_dm_visibility_snapshot, }; /// Route a command-kind event to the appropriate handler. @@ -307,6 +307,35 @@ fn compute_definition_hash(json_str: &str) -> Vec { Sha256::digest(json_str.as_bytes()).to_vec() } +async fn publish_dm_discovery( + tenant: &TenantContext, + state: &Arc, + channel: &buzz_db::channel::ChannelRecord, + participant_pubkeys: &[Vec], + operation: &'static str, +) { + match u64::try_from(channel.created_at.timestamp()) { + Ok(created_at) => { + if let Err(e) = + emit_dm_created(tenant, state, channel.id, created_at, participant_pubkeys).await + { + warn!( + channel = %channel.id, + operation, + "DM discovery emission failed: {e}" + ); + } + } + Err(_) => { + warn!( + channel = %channel.id, + operation, + "DM discovery has an invalid channel creation timestamp" + ); + } + } +} + async fn handle_dm_open( tenant: &TenantContext, state: &Arc, @@ -426,6 +455,11 @@ async fn handle_dm_open( } } + // Use the channel creation timestamp so retries produce the same event id. + // This also backfills discovery for DMs created before kind:41001 emission + // was implemented without adding duplicate rows on each re-open. + publish_dm_discovery(tenant, state, &channel, &all_bytes, "open").await; + // 6. Return response Ok(IngestResult { event_id: event.id.to_hex(), @@ -564,6 +598,8 @@ async fn handle_dm_add_member( } } + publish_dm_discovery(tenant, state, &new_channel, &all_bytes, "add_member").await; + // 8. Return response Ok(IngestResult { event_id: event.id.to_hex(), diff --git a/crates/buzz-relay/src/handlers/side_effects.rs b/crates/buzz-relay/src/handlers/side_effects.rs index 65d04ef0ba..51a07a88ac 100644 --- a/crates/buzz-relay/src/handlers/side_effects.rs +++ b/crates/buzz-relay/src/handlers/side_effects.rs @@ -7,11 +7,11 @@ use tracing::{info, warn}; use uuid::Uuid; use buzz_core::kind::{ - event_kind_u32, is_parameterized_replaceable, KIND_AGENT_PROFILE, KIND_DM_VISIBILITY, - KIND_GIT_REPO_ANNOUNCEMENT, KIND_IA_ARCHIVED, KIND_IA_ARCHIVED_LIST, KIND_IA_UNARCHIVED, - KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION, KIND_NIP29_GROUP_ADMINS, - KIND_NIP29_GROUP_MEMBERS, KIND_NIP29_GROUP_METADATA, KIND_NIP43_MEMBERSHIP_LIST, KIND_REACTION, - KIND_THREAD_SUMMARY, + event_kind_u32, is_parameterized_replaceable, KIND_AGENT_PROFILE, KIND_DM_CREATED, + KIND_DM_VISIBILITY, KIND_GIT_REPO_ANNOUNCEMENT, KIND_IA_ARCHIVED, KIND_IA_ARCHIVED_LIST, + KIND_IA_UNARCHIVED, KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION, + KIND_NIP29_GROUP_ADMINS, KIND_NIP29_GROUP_MEMBERS, KIND_NIP29_GROUP_METADATA, + KIND_NIP43_MEMBERSHIP_LIST, KIND_REACTION, KIND_THREAD_SUMMARY, }; use buzz_core::StoredEvent; use buzz_db::channel::{MemberRecord, MemberRole}; @@ -790,6 +790,56 @@ pub async fn emit_system_message( Ok(()) } +/// Emit the relay-confirmed DM discovery event consumed by `buzz dms list`. +/// +/// The event is channel-scoped so the normal private-channel read gate applies. +/// Every participant receives a `p` tag, which lets each identity discover the +/// same conversation with a `#p` query. +pub async fn emit_dm_created( + tenant: &TenantContext, + state: &Arc, + channel_id: Uuid, + created_at: u64, + participant_pubkeys: &[Vec], +) -> anyhow::Result<()> { + let channel_id_str = channel_id.to_string(); + let mut participant_hexes: Vec = participant_pubkeys.iter().map(hex::encode).collect(); + participant_hexes.sort_unstable(); + participant_hexes.dedup(); + + let mut tags = Vec::with_capacity(participant_hexes.len() + 2); + tags.push(Tag::parse(["d", &channel_id_str])?); + tags.push(Tag::parse(["h", &channel_id_str])?); + for pubkey in &participant_hexes { + tags.push(Tag::parse(["p", pubkey])?); + } + + let event = EventBuilder::new(Kind::Custom(KIND_DM_CREATED as u16), "") + .tags(tags) + .custom_created_at(nostr::Timestamp::from(created_at)) + .sign_with_keys(&state.relay_keypair) + .map_err(|e| anyhow::anyhow!("failed to sign DM discovery event: {e}"))?; + + let (stored, was_inserted) = state + .db + .insert_event(tenant.community(), &event, Some(channel_id)) + .await?; + if was_inserted { + let relay_pubkey_hex = state.relay_keypair.public_key().to_hex(); + dispatch_persistent_event( + tenant, + state, + &stored, + KIND_DM_CREATED, + &relay_pubkey_hex, + None, + ) + .await; + } + + Ok(()) +} + /// Sign and fan out a fresh relay-signed `kind:39005` thread-summary overlay /// for `root_id` after a thread mutation (reply insert or threaded delete). /// diff --git a/crates/buzz-test-client/tests/e2e_nostr_interop.rs b/crates/buzz-test-client/tests/e2e_nostr_interop.rs index fce7877676..0a218582fd 100644 --- a/crates/buzz-test-client/tests/e2e_nostr_interop.rs +++ b/crates/buzz-test-client/tests/e2e_nostr_interop.rs @@ -146,6 +146,44 @@ async fn create_dm(requester_keys: &Keys, other_pubkey_hex: &str) -> String { .to_string() } +/// Add a participant to an existing DM and return the newly created immutable +/// group-DM channel id from the relay response. +async fn add_dm_member(requester_keys: &Keys, channel_id: &str, new_pubkey_hex: &str) -> String { + let client = reqwest::Client::new(); + let event = EventBuilder::new(Kind::Custom(41011), "") + .tags(vec![ + Tag::parse(["h", channel_id]).unwrap(), + Tag::parse(["p", new_pubkey_hex]).unwrap(), + ]) + .sign_with_keys(requester_keys) + .unwrap(); + let resp = client + .post(format!("{}/events", relay_http_url())) + .header("X-Pubkey", requester_keys.public_key().to_hex()) + .header("Content-Type", "application/json") + .body(serde_json::to_string(&event).unwrap()) + .send() + .await + .expect("add DM member request"); + assert!( + resp.status().is_success(), + "add DM member failed: {}", + resp.status() + ); + let body: serde_json::Value = resp.json().await.expect("parse add-member response"); + assert!( + body["accepted"].as_bool().unwrap_or(false), + "DM add-member not accepted: {body}" + ); + let msg = body["message"].as_str().expect("message"); + let payload = msg.strip_prefix("response:").expect("response: prefix"); + let parsed: serde_json::Value = serde_json::from_str(payload).expect("response JSON"); + parsed["channel_id"] + .as_str() + .expect("channel_id") + .to_string() +} + /// Submit a signed command event via REST and assert it was accepted. async fn post_signed_event(keys: &Keys, kind: u16, tags: Vec) { let client = reqwest::Client::new(); @@ -752,8 +790,9 @@ async fn test_nip17_gift_wrap_recipient_receives() { client_b.disconnect().await.expect("disconnect B"); } -/// Create a DM via REST, then subscribe as a participant to verify discovery events. +/// Create a DM via REST, then subscribe as both participants to verify discovery events. /// Verify: kind:39000 event received with `hidden` and `private` tags. +/// Verify: kind:41001 is queryable by every participant and identifies the DM. /// Verify: kind:44100 membership notification received. #[tokio::test] #[ignore] @@ -761,14 +800,26 @@ async fn test_dm_discovery_events_emitted() { let url = relay_url(); let keys_a = Keys::generate(); let keys_b = Keys::generate(); + let keys_c = Keys::generate(); let a_pubkey_hex = keys_a.public_key().to_hex(); let b_pubkey_hex = keys_b.public_key().to_hex(); + let c_pubkey_hex = keys_c.public_key().to_hex(); // Create the DM via REST (A creates DM with B). This persists the relay's // kind:39000 discovery event and the kind:44100 membership notification // (stored globally, channel_id = None), then fans both out live. let channel_id = create_dm(&keys_a, &b_pubkey_hex).await; + // Re-open from the recipient before querying. The relay should self-heal a + // missing confirmation for older DMs and must not create a duplicate for a + // DM that already has one. + post_signed_event( + &keys_b, + 41010, + vec![Tag::parse(["p", &a_pubkey_hex]).unwrap()], + ) + .await; + // Connect A and subscribe AFTER create_dm. Both events are now in history, // so each subscription replays its event before EOSE — no dependency on // catching a live fan-out. (The previous ordering subscribed first, then let @@ -857,6 +908,132 @@ async fn test_dm_discovery_events_emitted() { "kind:39000 missing 'private' tag. tags: {tags:?}" ); + for (label, keys, pubkey_hex) in [ + ("creator", &keys_a, &a_pubkey_hex), + ("recipient", &keys_b, &b_pubkey_hex), + ] { + let mut client = BuzzTestClient::connect(&url, keys) + .await + .unwrap_or_else(|e| panic!("{label} connect: {e}")); + let sid = sub_id(&format!("dm-discovery-41001-{label}")); + let filter = Filter::new() + .kind(Kind::Custom(41001)) + .custom_tag(SingleLetterTag::lowercase(Alphabet::P), pubkey_hex.as_str()); + client + .subscribe(&sid, vec![filter]) + .await + .unwrap_or_else(|e| panic!("{label} subscribe: {e}")); + let events = client + .collect_until_eose(&sid, Duration::from_secs(10)) + .await + .unwrap_or_else(|e| panic!("{label} discovery EOSE: {e}")); + assert_eq!( + events.len(), + 1, + "create plus re-open must leave one kind:41001 event for {label}; events: {events:?}" + ); + + let event = events + .iter() + .find(|event| { + event.kind == Kind::Custom(41001) + && event.tags.iter().any(|tag| { + let values = tag.as_slice(); + values.len() >= 2 && values[0] == "d" && values[1] == channel_id + }) + }) + .unwrap_or_else(|| { + panic!( + "kind:41001 for DM {channel_id} must be discoverable by {label}; events: {events:?}" + ) + }); + let event_tags: Vec> = event + .tags + .iter() + .map(|tag| { + tag.as_slice() + .iter() + .map(|value| value.to_string()) + .collect() + }) + .collect(); + for participant in [&a_pubkey_hex, &b_pubkey_hex] { + assert!( + event_tags + .iter() + .any(|tag| tag.len() >= 2 && tag[0] == "p" && tag[1] == *participant), + "kind:41001 missing participant {participant}; tags: {event_tags:?}" + ); + } + assert!( + event_tags + .iter() + .any(|tag| tag.len() >= 2 && tag[0] == "h" && tag[1] == channel_id), + "kind:41001 missing h tag = DM channel id; tags: {event_tags:?}" + ); + + client.disconnect().await.expect("disconnect"); + } + + // Adding a participant creates a new immutable group DM. That producer + // path must emit the same participant-addressed discovery event. + let group_channel_id = add_dm_member(&keys_a, &channel_id, &c_pubkey_hex).await; + assert_ne!( + group_channel_id, channel_id, + "adding a participant must create a new group DM" + ); + let mut client_c = BuzzTestClient::connect(&url, &keys_c) + .await + .expect("new group-DM participant connect"); + let sid_group = sub_id("dm-discovery-41001-group-recipient"); + let group_filter = Filter::new().kind(Kind::Custom(41001)).custom_tag( + SingleLetterTag::lowercase(Alphabet::P), + c_pubkey_hex.as_str(), + ); + client_c + .subscribe(&sid_group, vec![group_filter]) + .await + .expect("subscribe group-DM discovery"); + let group_events = client_c + .collect_until_eose(&sid_group, Duration::from_secs(10)) + .await + .expect("group-DM discovery EOSE"); + let group_event = group_events + .iter() + .find(|event| { + event.kind == Kind::Custom(41001) + && event.tags.iter().any(|tag| { + let values = tag.as_slice(); + values.len() >= 2 + && values[0] == "d" + && values[1] == group_channel_id + }) + }) + .unwrap_or_else(|| { + panic!( + "kind:41001 for group DM {group_channel_id} must be discoverable by its new participant; events: {group_events:?}" + ) + }); + let group_tags: Vec> = group_event + .tags + .iter() + .map(|tag| { + tag.as_slice() + .iter() + .map(|value| value.to_string()) + .collect() + }) + .collect(); + for participant in [&a_pubkey_hex, &b_pubkey_hex, &c_pubkey_hex] { + assert!( + group_tags + .iter() + .any(|tag| tag.len() >= 2 && tag[0] == "p" && tag[1] == *participant), + "group kind:41001 missing participant {participant}; tags: {group_tags:?}" + ); + } + client_c.disconnect().await.expect("disconnect"); + client_a.disconnect().await.expect("disconnect"); }