diff --git a/Cargo.lock b/Cargo.lock index 34c2b47361..4a82cf4c9b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -235,7 +235,7 @@ version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" dependencies = [ - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -246,7 +246,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -1815,7 +1815,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "117725a109d387c937a1533ce01b450cbde6b88abceea8473c4d7a85853cda3c" dependencies = [ "lazy_static", - "windows-sys 0.48.0", + "windows-sys 0.59.0", ] [[package]] @@ -1824,7 +1824,7 @@ version = "3.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "faf9468729b8cbcea668e36183cb69d317348c2e08e994829fb56ebfdfbaac34" dependencies = [ - "windows-sys 0.48.0", + "windows-sys 0.61.2", ] [[package]] @@ -2814,7 +2814,7 @@ dependencies = [ "libc", "option-ext", "redox_users 0.5.2", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -3144,7 +3144,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -5892,7 +5892,7 @@ dependencies = [ "portable-atomic", "portable-atomic-util", "serde_core", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -7994,6 +7994,7 @@ dependencies = [ "moltis-common", "moltis-config", "moltis-metrics", + "nostr-relay-builder", "nostr-sdk", "secrecy 0.8.0", "serde", @@ -8967,6 +8968,23 @@ dependencies = [ "nostr", ] +[[package]] +name = "nostr-relay-builder" +version = "0.44.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "08ed09e8363503c6f464087b530a1f43cf68be6292bd4928b5ea1141b03dc11c" +dependencies = [ + "async-utility", + "async-wsocket", + "atomic-destructor", + "hex", + "negentropy", + "nostr", + "nostr-database", + "tokio", + "tracing", +] + [[package]] name = "nostr-relay-pool" version = "0.44.0" @@ -9055,7 +9073,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -9191,7 +9209,7 @@ version = "5.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "51e219e79014df21a225b1860a479e2dcd7cbd9130f4defd4bd0e191ea31d67d" dependencies = [ - "base64 0.21.7", + "base64 0.22.1", "chrono", "getrandom 0.2.17", "http 1.4.2", @@ -10000,8 +10018,8 @@ version = "0.14.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "03da047801ff44bb6a4d407d4860c05fd70bb81714e6b2f3812603d5b145b042" dependencies = [ - "heck 0.4.1", - "itertools 0.10.5", + "heck 0.5.0", + "itertools 0.14.0", "log", "multimap", "petgraph", @@ -10018,7 +10036,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8a56d757972c98b346a9b766e3f02746cde6dd1cd1d1d563472929fdd74bec4d" dependencies = [ "anyhow", - "itertools 0.10.5", + "itertools 0.14.0", "proc-macro2", "quote", "syn 2.0.117", @@ -10031,7 +10049,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf" dependencies = [ "anyhow", - "itertools 0.10.5", + "itertools 0.14.0", "proc-macro2", "quote", "syn 2.0.117", @@ -11024,7 +11042,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -11129,7 +11147,7 @@ dependencies = [ "security-framework 3.7.0", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -12039,7 +12057,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3a766e1110788c36f4fa1c2b71b387a7815aa65f88ce0229841826633d93723e" dependencies = [ "libc", - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -12282,7 +12300,7 @@ dependencies = [ "cfg-if", "libc", "psm", - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -12662,7 +12680,7 @@ dependencies = [ "getrandom 0.4.2", "once_cell", "rustix 1.1.4", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -12702,7 +12720,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "230a1b821ccbd75b185820a1f1ff7b14d21da1e442e22c0863ea5f08771a8874" dependencies = [ "rustix 1.1.4", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -15042,7 +15060,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.48.0", + "windows-sys 0.61.2", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index aeb9e23b25..1684aaaf69 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -250,6 +250,7 @@ jsonwebtoken = { features = ["aws_lc_rs"], version = "10" } llama-cpp-2 = "0.1" matrix-sdk = { default-features = false, features = ["e2e-encryption", "markdown", "native-tls", "sqlite"], version = "0.16" } mockito = "1.7" +nostr-relay-builder = { version = "0.44" } nostr-sdk = { features = ["nip04", "nip44", "nip59"], version = "0.44" } open = "5.3" rand = "0.10" diff --git a/README.md b/README.md index 605105c8a7..41e4e2c4ba 100644 --- a/README.md +++ b/README.md @@ -121,7 +121,7 @@ Verify releases with `gh attestation verify -R moltis-org/moltis` or ## Features - **AI Gateway** — Multi-provider LLM support (OpenAI Codex, GitHub Copilot, Local), streaming responses, agent loop with sub-agent delegation, session modes, parallel tool execution -- **Communication** — Web UI, Telegram, Signal, Microsoft Teams, Discord, API access, voice I/O (8 TTS + 7 STT providers), mobile PWA with push notifications +- **Communication** — Web UI, Telegram, Signal, Microsoft Teams, Discord, Nostr DMs + NIP-29 group chat (incl. Block's Buzz channels), API access, voice I/O (8 TTS + 7 STT providers), mobile PWA with push notifications - **Memory & Recall** — Per-agent memory workspaces, embeddings-powered long-term memory, hybrid vector + full-text search, session persistence with auto-compaction, cross-session recall, Cursor-compatible project context, context-file safety scanning - **Safer Agent Editing** — Automatic checkpoints before built-in skill and memory mutations, restore tooling, session branching - **Extensibility** — MCP servers (stdio + HTTP/SSE), skill system, 15 lifecycle hook events with circuit breaker, destructive command guard diff --git a/crates/config/src/template.rs b/crates/config/src/template.rs index 93e3f46ec7..622884d0cc 100644 --- a/crates/config/src/template.rs +++ b/crates/config/src/template.rs @@ -830,6 +830,23 @@ port = {port} # Port number (auto-generated for this i # otp_self_approval = true # otp_cooldown_secs = 300 +# Example Nostr account. Handles NIP-04/NIP-59 encrypted DMs and, when `groups` +# is set, NIP-29 group chat on Buzz-style relays (https://github.com/block/buzz). +# [channels.nostr.my-bot] +# secret_key = "nsec1..." +# relays = ["wss://relay.damus.io", "wss://relay.nostr.band", "wss://nos.lol"] +# dm_policy = "allowlist" +# allowed_pubkeys = ["npub1..."] +# # Buzz / NIP-29 group chat: the `h`-tag group ids the bot joins. Requires a +# # relay that supports NIP-29 + NIP-42 (Buzz relays do). Empty = DM-only. +# # This list is also the allowlist — messages for any other group are dropped. +# groups = ["buzz-general"] +# group_mention_mode = "mention" # mention (p-tagged only) | always | none +# # Dialect for bot-initiated group messages. Both kinds are always read and +# # replies mirror the message they answer; set buzz_v2 on a Buzz relay. +# group_message_kind = "nip29" # nip29 (kind:9) | buzz_v2 (kind:40002) +# group_ack_reactions = true # 👀 on receipt, ✅/❌ on completion (NIP-25) + # See docs or defaults.toml for full channel configuration examples # (WhatsApp, Telegram, Teams, Discord, Slack, Matrix, Nostr, Signal). diff --git a/crates/nostr/Cargo.toml b/crates/nostr/Cargo.toml index bae21817a0..b1ff8fcd2c 100644 --- a/crates/nostr/Cargo.toml +++ b/crates/nostr/Cargo.toml @@ -24,7 +24,8 @@ default = [] metrics = ["dep:moltis-metrics"] [dev-dependencies] -tokio = { workspace = true } +nostr-relay-builder = { workspace = true } +tokio = { workspace = true } [lints] workspace = true diff --git a/crates/nostr/src/bus.rs b/crates/nostr/src/bus.rs index 3e8891663f..f8860eb63e 100644 --- a/crates/nostr/src/bus.rs +++ b/crates/nostr/src/bus.rs @@ -1,9 +1,11 @@ -//! Nostr relay subscription loop — inbound DM pipeline. +//! Nostr relay subscription loop — inbound DM and group-chat pipeline. //! //! Subscribes to kind:4 (NIP-04 encrypted DMs) and kind:1059 (NIP-59 gift -//! wraps) addressed to the bot's pubkey. Events flow through dedup, -//! self-message filtering, access control, and decryption/unwrapping before -//! being dispatched to the gateway via `ChannelEventSink`. +//! wraps) addressed to the bot's pubkey, plus group chat (kind:9 NIP-29 and +//! kind:40002 Buzz) scoped to the configured groups via the `h` tag. Events flow +//! through dedup, self-message filtering, access control, and (for DMs) +//! decryption/unwrapping before being dispatched to the gateway via +//! `ChannelEventSink`. use std::sync::{Arc, RwLock}; @@ -19,7 +21,7 @@ use crate::{ access::{self, AccessDenied}, config::NostrAccountConfig, seen::SeenTracker, - state::SharedOtp, + state::{SharedOtp, SharedReplyContexts}, }; #[cfg(feature = "metrics")] @@ -43,6 +45,7 @@ pub async fn run_subscription_loop( config: Arc>, cached_allowlist: Arc>>, otp: SharedOtp, + reply_ctx: SharedReplyContexts, account_id: String, event_sink: Arc, cancel: CancellationToken, @@ -78,7 +81,31 @@ pub async fn run_subscription_loop( let mut seen = SeenTracker::new(); - for filter in [nip04_filter, gift_wrap_filter] { + let mut filters = vec![nip04_filter, gift_wrap_filter]; + + // NIP-29 group chat (Buzz channels) — subscribe to kind:9 scoped to the + // configured groups via the `h` tag. Relay subscriptions are fixed at + // connect, so group membership changes take effect on account restart. + let group_ids: Vec = { + let cfg = config.read().unwrap_or_else(|e| e.into_inner()); + cfg.groups.clone() + }; + if !group_ids.is_empty() { + // Both dialects: standard NIP-29 kind:9 and Buzz kind:40002. A client + // filtering only one never sees the other, so read both. + let group_filter = Filter::new() + .kinds(crate::groups::group_message_kinds()) + .custom_tags(crate::groups::h_tag(), group_ids.clone()) + .since(since); + filters.push(group_filter); + tracing::info!( + account_id, + groups = ?group_ids, + "subscribing to group chat (kind:9 + kind:40002)" + ); + } + + for filter in filters { if let Err(e) = client.subscribe(filter, None).await { tracing::error!(account_id, "failed to subscribe: {e}"); return; @@ -106,6 +133,7 @@ pub async fn run_subscription_loop( &config, &cached_allowlist, &otp, + &reply_ctx, &account_id, &event_sink, ).await; @@ -141,9 +169,28 @@ async fn handle_event( config: &Arc>, cached_allowlist: &Arc>>, otp: &SharedOtp, + reply_ctx: &SharedReplyContexts, account_id: &str, event_sink: &Arc, ) { + // 0. Route group chat (kind:9 / kind:40002) separately from encrypted DMs. + if crate::groups::is_group_message_kind(event.kind) { + handle_group_event( + event, + client, + bot_pubkey, + since, + seen, + config, + cached_allowlist, + reply_ctx, + account_id, + event_sink, + ) + .await; + return; + } + // 1. Kind gate — accept kind:4 (legacy NIP-04) and kind:1059 (NIP-59 gift wrap) let is_gift_wrap = event.kind == Kind::GiftWrap; if event.kind != Kind::EncryptedDirectMessage && !is_gift_wrap { @@ -357,6 +404,218 @@ async fn handle_event( event_sink.dispatch_to_chat(text, reply_to, meta).await; } +/// Process an inbound group chat message (`kind:9` NIP-29 or `kind:40002` +/// Buzz, e.g. a Buzz channel). +/// +/// Group messages are plaintext (the relay enforces membership), so there is +/// no decryption or OTP flow. The event is deduped, filtered for self-messages +/// and staleness, gated by group membership and mention mode, then dispatched +/// to the gateway with the group id as the session/chat id so replies route +/// back to the group. +#[allow(clippy::too_many_arguments)] +async fn handle_group_event( + event: &Event, + client: &Client, + bot_pubkey: &PublicKey, + since: Timestamp, + seen: &mut SeenTracker, + config: &Arc>, + cached_allowlist: &Arc>>, + reply_ctx: &SharedReplyContexts, + account_id: &str, + event_sink: &Arc, +) { + // Dedup, skip self-messages, skip stale events. + if seen.check_and_insert(&event.id) { + return; + } + if event.pubkey == *bot_pubkey { + return; + } + if event.created_at < since { + return; + } + + let Some(group_id) = crate::groups::extract_group_id(event) else { + tracing::debug!( + account_id, + event_id = %event.id, + "group message without `h` tag, ignoring" + ); + return; + }; + + // Read group config fields (drop the guard before any .await). + let (groups, mention_mode, ack_reactions) = { + let cfg = config.read().unwrap_or_else(|e| e.into_inner()); + ( + cfg.groups.clone(), + cfg.group_mention_mode.clone(), + cfg.group_ack_reactions, + ) + }; + + // Re-check membership against the configured join list. The relay is + // untrusted and `nostr-sdk` does not verify that delivered events match our + // subscription filters, so a `kind:9` for a group we never joined can still + // arrive on this socket. + if let Err(denied) = crate::groups::check_group_access(&group_id, &groups) { + #[cfg(feature = "metrics")] + counter!(nostr_metrics::ACCESS_CONTROL_DENIALS_TOTAL, "reason" => "group").increment(1); + tracing::debug!( + account_id, + group = group_id, + "group message rejected: {denied}" + ); + return; + } + + if !crate::groups::should_respond(&mention_mode, event, bot_pubkey) { + tracing::trace!( + account_id, + group = group_id, + "group message ignored by mention mode" + ); + return; + } + + // Size validation — truncate at a safe UTF-8 boundary. + let text = if event.content.len() > MAX_MESSAGE_BYTES { + tracing::warn!( + account_id, + len = event.content.len(), + "group message exceeds size limit" + ); + &event.content[..event.content.floor_char_boundary(MAX_MESSAGE_BYTES)] + } else { + event.content.as_str() + }; + if text.trim().is_empty() { + return; + } + + let sender_hex = event.pubkey.to_hex(); + let sender_npub = event + .pubkey + .to_bech32() + .unwrap_or_else(|_| sender_hex.clone()); + + // Remember who wrote this and in which dialect, so the reply can `p`-tag + // them and answer with the same kind. Outbound only receives the chat id + // and this event id, so it cannot recover either on its own. + { + let mut ctxs = reply_ctx.lock().unwrap_or_else(|e| e.into_inner()); + ctxs.record(event.id, &group_id, event.pubkey, event.kind); + } + + #[cfg(feature = "metrics")] + counter!(nostr_metrics::MESSAGES_RECEIVED_TOTAL).increment(1); + + event_sink + .emit(moltis_channels::ChannelEvent::InboundMessage { + channel_type: moltis_channels::ChannelType::Nostr, + account_id: account_id.to_string(), + peer_id: group_id.clone(), + username: Some(sender_npub.clone()), + sender_name: None, + message_count: None, + access_granted: true, + }) + .await; + + // Reply routes back to the group (chat_id = group id), threaded to this + // message so NIP-29 clients render it as a reply. + // Setting `ack_message_id` opts this message into the gateway's + // acknowledgement reactions (👀 on receipt, ✅/❌ on completion). + let reply_to = moltis_channels::ChannelReplyTarget { + ack_message_id: ack_reactions.then(|| event.id.to_hex()), + channel_type: moltis_channels::ChannelType::Nostr, + account_id: account_id.to_string(), + // Prefixed so the target carries its own kind: a group id can look + // exactly like a pubkey, and this target outlives the config. + chat_id: crate::groups::group_target(&group_id), + message_id: Some(event.id.to_hex()), + thread_id: None, + }; + + // Intercept slash commands before the LLM, matching DM behaviour so + // `/help` and friends work the same way in a Buzz channel. + if let Some(cmd_text) = text.strip_prefix('/') { + let cmd_name = cmd_text.split_whitespace().next().unwrap_or(""); + if moltis_channels::commands::is_channel_command(cmd_name, cmd_text) { + // `/sh` and `/sandbox` control code execution. Group membership is + // the relay's call, not ours, so restrict them to keys the owner + // nominated — and say so rather than ignoring the message. + if crate::groups::is_operator_only_command(cmd_name) { + let is_operator = { + let allowed = cached_allowlist.read().unwrap_or_else(|e| e.into_inner()); + crate::groups::is_group_operator(&event.pubkey, &allowed) + }; + if !is_operator { + tracing::warn!( + account_id, + group = group_id, + sender = sender_hex, + command = cmd_name, + "refused operator-only command from non-allowlisted group sender" + ); + if let Err(e) = crate::groups::send_group_message( + client, + event.kind, + &group_id, + &format!("`/{cmd_name}` is restricted to this bot's operators."), + Some(event.id), + Some(event.pubkey), + ) + .await + { + tracing::warn!(account_id, "failed to send command refusal: {e}"); + } + return; + } + } + let response = if cmd_name == "help" { + Ok(moltis_channels::commands::help_text()) + } else { + event_sink + .dispatch_command(cmd_text, reply_to, Some(&sender_hex)) + .await + }; + let reply_text = match response { + Ok(msg) => msg, + Err(e) => format!("Error: {e}"), + }; + if let Err(e) = crate::groups::send_group_message( + client, + event.kind, + &group_id, + &reply_text, + Some(event.id), + Some(event.pubkey), + ) + .await + { + tracing::warn!(account_id, "failed to send group command response: {e}"); + } + return; + } + } + + let meta = moltis_channels::ChannelMessageMeta { + channel_type: moltis_channels::ChannelType::Nostr, + sender_name: None, + username: Some(sender_npub), + sender_id: Some(sender_hex), + message_kind: Some(moltis_channels::ChannelMessageKind::Text), + model: None, + agent_id: None, + audio_filename: None, + documents: None, + }; + + event_sink.dispatch_to_chat(text, reply_to, meta).await; +} + /// Try to decrypt a DM (NIP-04, then NIP-44). Returns `None` on failure. fn try_decrypt(keys: &Keys, sender: &PublicKey, content: &str) -> Option { nip04::decrypt(keys.secret_key(), sender, content) @@ -505,11 +764,363 @@ async fn handle_otp_challenge( } #[cfg(test)] +#[allow(clippy::expect_used)] mod tests { - use super::*; + use std::sync::Mutex; + + use { + async_trait::async_trait, + moltis_channels::{ + ChannelEvent, ChannelEventSink, ChannelMessageMeta, ChannelReplyTarget, + Result as ChannelResult, + }, + nostr_sdk::prelude::{EventBuilder, Keys}, + }; + + use { + super::*, + crate::{groups, reply_ctx::ReplyContexts}, + }; + + /// What a dispatched message looked like, for assertions. + #[derive(Debug, Clone, PartialEq, Eq)] + struct Dispatched { + text: String, + chat_id: String, + message_id: Option, + } + + /// Recording `ChannelEventSink` — captures dispatches instead of running + /// the agent, so the inbound gate can be exercised without a gateway. + #[derive(Default)] + struct RecordingSink { + dispatched: Mutex>, + } + + impl RecordingSink { + fn dispatched(&self) -> Vec { + self.dispatched + .lock() + .unwrap_or_else(|e| e.into_inner()) + .clone() + } + } + + #[async_trait] + impl ChannelEventSink for RecordingSink { + async fn emit(&self, _event: ChannelEvent) {} + + async fn dispatch_to_chat( + &self, + text: &str, + reply_to: ChannelReplyTarget, + _meta: ChannelMessageMeta, + ) { + self.dispatched + .lock() + .unwrap_or_else(|e| e.into_inner()) + .push(Dispatched { + text: text.to_string(), + chat_id: reply_to.chat_id, + message_id: reply_to.message_id, + }); + } + + async fn dispatch_command( + &self, + _command: &str, + _reply_to: ChannelReplyTarget, + _sender_id: Option<&str>, + ) -> ChannelResult { + Ok(String::new()) + } + + async fn request_disable_account( + &self, + _channel_type: &str, + _account_id: &str, + _reason: &str, + ) { + } + } + + /// Fixtures for driving `handle_group_event` without any network. + struct Harness { + client: Client, + bot: Keys, + seen: SeenTracker, + config: Arc>, + allowlist: Arc>>, + reply_ctx: SharedReplyContexts, + sink: Arc, + since: Timestamp, + } + + impl Harness { + fn new(config: NostrAccountConfig) -> Self { + let bot = Keys::generate(); + Self { + client: Client::new(bot.clone()), + bot, + seen: SeenTracker::new(), + config: Arc::new(RwLock::new(config)), + allowlist: Arc::new(RwLock::new(Vec::new())), + reply_ctx: Arc::new(Mutex::new(ReplyContexts::new())), + sink: Arc::new(RecordingSink::default()), + since: Timestamp::from(0), + } + } + + fn with_groups(groups: Vec<&str>, mention_mode: MentionModeAlias) -> Self { + Self::new(NostrAccountConfig { + groups: groups.into_iter().map(String::from).collect(), + group_mention_mode: mention_mode, + ..Default::default() + }) + } + + /// Build a signed group message from a third party. + fn incoming(&self, group: &str, text: &str, kind: Kind, mention_bot: bool) -> Event { + let author = Keys::generate(); + let mention = mention_bot.then(|| self.bot.public_key()); + EventBuilder::new(kind, text) + .tags(groups::build_group_message_tags(group, None, mention)) + .sign_with_keys(&author) + .expect("sign group event") + } + + async fn handle(&mut self, event: &Event) { + let sink: Arc = + Arc::clone(&self.sink) as Arc; + handle_group_event( + event, + &self.client, + &self.bot.public_key(), + self.since, + &mut self.seen, + &self.config, + &self.allowlist, + &self.reply_ctx, + "acct", + &sink, + ) + .await; + } + } + + type MentionModeAlias = moltis_channels::gating::MentionMode; #[test] fn max_message_size_is_reasonable() { assert_eq!(MAX_MESSAGE_BYTES, 64 * 1024); } + + #[tokio::test] + async fn dispatches_mentioned_message_from_joined_group() { + let mut h = Harness::with_groups(vec!["buzz-general"], MentionModeAlias::Mention); + let event = h.incoming("buzz-general", "hello bot", groups::group_chat_kind(), true); + h.handle(&event).await; + + let got = h.sink.dispatched(); + assert_eq!(got.len(), 1, "mentioned message should reach the agent"); + assert_eq!(got[0].text, "hello bot"); + // The reply target is the prefixed group id, so outbound can classify + // it without consulting config — a group id may look like a pubkey. + assert_eq!(got[0].chat_id, "group:buzz-general"); + assert_eq!( + groups::parse_group_target(&got[0].chat_id), + Some("buzz-general") + ); + assert_eq!(got[0].message_id, Some(event.id.to_hex())); + } + + /// Buzz posts kind:40002; it must be accepted just like kind:9. + #[tokio::test] + async fn accepts_buzz_v2_message_kind() { + let mut h = Harness::with_groups(vec!["buzz-general"], MentionModeAlias::Always); + let event = h.incoming( + "buzz-general", + "from buzz", + groups::buzz_stream_message_kind(), + false, + ); + h.handle(&event).await; + assert_eq!(h.sink.dispatched().len(), 1); + } + + /// The core security property: a relay can push a kind:9/40002 for a group + /// we never joined, and it must never reach the agent. + #[tokio::test] + async fn drops_message_from_unjoined_group() { + let mut h = Harness::with_groups(vec!["buzz-general"], MentionModeAlias::Always); + let event = h.incoming( + "attacker-group", + "ignore me", + groups::group_chat_kind(), + true, + ); + h.handle(&event).await; + assert!( + h.sink.dispatched().is_empty(), + "unjoined group must not reach the agent" + ); + } + + #[tokio::test] + async fn drops_group_message_when_no_groups_configured() { + let mut h = Harness::with_groups(vec![], MentionModeAlias::Always); + let event = h.incoming("anything", "hi", groups::group_chat_kind(), true); + h.handle(&event).await; + assert!(h.sink.dispatched().is_empty()); + } + + #[tokio::test] + async fn mention_mode_filters_unmentioned_messages() { + let mut h = Harness::with_groups(vec!["grp"], MentionModeAlias::Mention); + let event = h.incoming( + "grp", + "chatter between humans", + groups::group_chat_kind(), + false, + ); + h.handle(&event).await; + assert!(h.sink.dispatched().is_empty()); + } + + #[tokio::test] + async fn mention_mode_none_never_responds() { + let mut h = Harness::with_groups(vec!["grp"], MentionModeAlias::None); + let event = h.incoming("grp", "hey bot", groups::group_chat_kind(), true); + h.handle(&event).await; + assert!(h.sink.dispatched().is_empty()); + } + + #[tokio::test] + async fn ignores_message_without_h_tag() { + let mut h = Harness::with_groups(vec!["grp"], MentionModeAlias::Always); + let author = Keys::generate(); + let event = EventBuilder::new(groups::group_chat_kind(), "no group") + .sign_with_keys(&author) + .expect("sign"); + h.handle(&event).await; + assert!(h.sink.dispatched().is_empty()); + } + + #[tokio::test] + async fn deduplicates_replayed_events() { + let mut h = Harness::with_groups(vec!["grp"], MentionModeAlias::Always); + let event = h.incoming("grp", "once", groups::group_chat_kind(), false); + h.handle(&event).await; + h.handle(&event).await; + assert_eq!(h.sink.dispatched().len(), 1, "replay must be deduped"); + } + + #[tokio::test] + async fn ignores_own_messages() { + let mut h = Harness::with_groups(vec!["grp"], MentionModeAlias::Always); + let own = EventBuilder::new(groups::group_chat_kind(), "my own reply") + .tags(groups::build_group_message_tags("grp", None, None)) + .sign_with_keys(&h.bot) + .expect("sign"); + h.handle(&own).await; + assert!( + h.sink.dispatched().is_empty(), + "bot must not answer itself and loop" + ); + } + + #[tokio::test] + async fn skips_blank_messages() { + let mut h = Harness::with_groups(vec!["grp"], MentionModeAlias::Always); + let event = h.incoming("grp", " ", groups::group_chat_kind(), false); + h.handle(&event).await; + assert!(h.sink.dispatched().is_empty()); + } + + /// A channel command is answered by the command handler, not the model — + /// the same interception DMs get. Publishing the response needs a relay, + /// which this harness has none of, so the assertion is that the text never + /// reached the agent. + #[tokio::test] + async fn slash_commands_are_intercepted_in_groups() { + let mut h = Harness::with_groups(vec!["grp"], MentionModeAlias::Always); + let event = h.incoming("grp", "/help", groups::group_chat_kind(), false); + h.handle(&event).await; + + assert!( + h.sink.dispatched().is_empty(), + "/help must be handled as a command, not sent to the model" + ); + } + + /// A slash that is not a known command is ordinary text and must still + /// reach the agent, rather than being silently swallowed. + #[tokio::test] + async fn unknown_slash_text_still_reaches_the_agent() { + let mut h = Harness::with_groups(vec!["grp"], MentionModeAlias::Always); + let event = h.incoming( + "grp", + "/notacommand hello", + groups::group_chat_kind(), + false, + ); + h.handle(&event).await; + + let got = h.sink.dispatched(); + assert_eq!(got.len(), 1, "unknown slash text is just a message"); + assert_eq!(got[0].text, "/notacommand hello"); + } + + /// `/sh` grants shell execution, and group membership is the relay's call, + /// so a member who is not on the account's allowlist must not reach it. + #[tokio::test] + async fn operator_only_commands_are_refused_from_group_strangers() { + let mut h = Harness::with_groups(vec!["grp"], MentionModeAlias::Always); + let event = h.incoming("grp", "/sh on", groups::group_chat_kind(), false); + h.handle(&event).await; + + assert!( + h.sink.dispatched().is_empty(), + "/sh from a non-operator must not reach the command handler" + ); + } + + /// The same command from an allowlisted key is intercepted as a command + /// (handled, not forwarded to the model). + #[tokio::test] + async fn operator_only_commands_are_allowed_from_operators() { + let author = Keys::generate(); + let mut h = Harness::with_groups(vec!["grp"], MentionModeAlias::Always); + { + let mut allowed = h.allowlist.write().unwrap_or_else(|e| e.into_inner()); + allowed.push(author.public_key()); + } + let event = EventBuilder::new(groups::group_chat_kind(), "/sh on") + .tags(groups::build_group_message_tags("grp", None, None)) + .sign_with_keys(&author) + .expect("sign"); + h.handle(&event).await; + + assert!( + h.sink.dispatched().is_empty(), + "commands are handled by the command path, not dispatched to the model" + ); + } + + /// Accepted messages record who to `p`-tag and which dialect to answer in. + #[tokio::test] + async fn records_reply_context_for_accepted_message() { + let mut h = Harness::with_groups(vec!["grp"], MentionModeAlias::Always); + let event = h.incoming("grp", "hi", groups::buzz_stream_message_kind(), false); + h.handle(&event).await; + + let ctxs = h.reply_ctx.lock().unwrap_or_else(|e| e.into_inner()); + let ctx = ctxs.get(&event.id).expect("context recorded"); + assert_eq!(ctx.author, event.pubkey); + assert_eq!(ctx.kind, groups::buzz_stream_message_kind()); + assert_eq!( + ctxs.kind_for_group("grp"), + Some(groups::buzz_stream_message_kind()) + ); + } } diff --git a/crates/nostr/src/config.rs b/crates/nostr/src/config.rs index f26f9e62d2..fd27ac82bf 100644 --- a/crates/nostr/src/config.rs +++ b/crates/nostr/src/config.rs @@ -3,13 +3,15 @@ use { moltis_channels::{ config_view::ChannelConfigView, - gating::{DmPolicy, GroupPolicy}, + gating::{DmPolicy, GroupPolicy, MentionMode}, }, moltis_common::secret_serde, secrecy::Secret, serde::{Deserialize, Serialize, ser::SerializeStruct}, }; +use crate::groups::GroupMessageKind; + /// NIP-01 profile metadata to publish on connect. #[derive(Debug, Clone, Default, Serialize, Deserialize)] #[serde(default)] @@ -48,6 +50,30 @@ pub struct NostrAccountConfig { /// Public keys allowed to send DMs (npub1/hex). pub allowed_pubkeys: Vec, + /// NIP-29 group ids (the `h` tag values) to join — e.g. Buzz channels. + /// + /// This list is authoritative for group access: it is both the set we + /// subscribe to and the set we accept messages from. Empty (the default) + /// keeps the account in DM-only mode. Changing it takes effect on the next + /// account restart (relay subscriptions are fixed at connect). + pub groups: Vec, + + /// When the bot should respond in group chats: `mention` (only when its + /// pubkey is `p`-tagged — the default), `always`, or `none`. + pub group_mention_mode: MentionMode, + + /// Acknowledge inbound group messages with reactions: 👀 on receipt, then + /// ✅ or ❌ when the turn finishes (NIP-25 `kind:7`, retracted via NIP-09). + pub group_ack_reactions: bool, + + /// Dialect for group messages that are not replies: `nip29` (`kind:9`) or + /// `buzz_v2` (`kind:40002`). + /// + /// Both kinds are always *read*, and replies mirror whatever kind they + /// answer, so this only affects cold-start sends into a group we have not + /// yet received a message from. Set it to `buzz_v2` for a Buzz workspace. + pub group_message_kind: GroupMessageKind, + /// Whether this account is enabled. pub enabled: bool, @@ -81,6 +107,10 @@ impl Default for NostrAccountConfig { relays: default_relays(), dm_policy: DmPolicy::Allowlist, allowed_pubkeys: Vec::new(), + groups: Vec::new(), + group_mention_mode: MentionMode::Mention, + group_ack_reactions: true, + group_message_kind: GroupMessageKind::default(), enabled: true, profile: None, model: None, @@ -107,6 +137,10 @@ impl std::fmt::Debug for NostrAccountConfig { .field("relays", &self.relays) .field("dm_policy", &self.dm_policy) .field("allowed_pubkeys", &self.allowed_pubkeys) + .field("groups", &self.groups) + .field("group_mention_mode", &self.group_mention_mode) + .field("group_ack_reactions", &self.group_ack_reactions) + .field("group_message_kind", &self.group_message_kind) .field("enabled", &self.enabled) .field("profile", &self.profile) .field("model", &self.model) @@ -124,13 +158,17 @@ pub struct RedactedConfig<'a>(pub &'a NostrAccountConfig); impl Serialize for RedactedConfig<'_> { fn serialize(&self, serializer: S) -> Result { let c = self.0; - let mut count = 10; + let mut count = 15; count += c.agent_id.is_some() as usize; let mut s = serializer.serialize_struct("NostrAccountConfig", count)?; s.serialize_field("secret_key", "[REDACTED]")?; s.serialize_field("relays", &c.relays)?; s.serialize_field("dm_policy", &c.dm_policy)?; s.serialize_field("allowed_pubkeys", &c.allowed_pubkeys)?; + s.serialize_field("groups", &c.groups)?; + s.serialize_field("group_mention_mode", &c.group_mention_mode)?; + s.serialize_field("group_ack_reactions", &c.group_ack_reactions)?; + s.serialize_field("group_message_kind", &c.group_message_kind)?; s.serialize_field("enabled", &c.enabled)?; s.serialize_field("profile", &c.profile)?; s.serialize_field("model", &c.model)?; @@ -150,8 +188,8 @@ impl ChannelConfigView for NostrAccountConfig { } fn group_allowlist(&self) -> &[String] { - // Nostr DMs are always 1:1, no group concept - &[] + // NIP-29 group ids the bot joins (e.g. Buzz channels). + &self.groups } fn dm_policy(&self) -> DmPolicy { @@ -159,7 +197,13 @@ impl ChannelConfigView for NostrAccountConfig { } fn group_policy(&self) -> GroupPolicy { - GroupPolicy::Disabled + // `groups` is the join list *and* the allowlist — there is no "open" + // mode, because we only ever subscribe to the groups named here. + if self.groups.is_empty() { + GroupPolicy::Disabled + } else { + GroupPolicy::Allowlist + } } fn model(&self) -> Option<&str> { @@ -235,4 +279,51 @@ mod tests { assert_eq!(cfg.model(), Some("test-model")); assert_eq!(cfg.model_provider(), Some("test-provider")); } + + #[test] + fn default_group_config_is_dm_only() { + let cfg = NostrAccountConfig::default(); + assert!(cfg.groups.is_empty()); + assert_eq!(cfg.group_mention_mode, MentionMode::Mention); + // Group policy reports Disabled until a group is configured. + assert_eq!(cfg.group_policy(), GroupPolicy::Disabled); + } + + /// The join list doubles as the allowlist — there is no "open" group mode. + #[test] + fn configured_groups_expose_policy_and_allowlist() { + let cfg = NostrAccountConfig { + groups: vec!["buzz-general".into(), "buzz-dev".into()], + group_mention_mode: MentionMode::Always, + ..Default::default() + }; + assert_eq!(cfg.group_policy(), GroupPolicy::Allowlist); + assert_eq!(cfg.group_allowlist(), &["buzz-general", "buzz-dev"]); + } + + #[test] + fn groups_round_trip_json() { + let cfg = NostrAccountConfig { + secret_key: Secret::new("deadbeef".repeat(8)), + groups: vec!["buzz-general".into()], + group_mention_mode: MentionMode::Always, + ..Default::default() + }; + let json = serde_json::to_value(&cfg).unwrap_or_default(); + let parsed: NostrAccountConfig = serde_json::from_value(json).unwrap_or_default(); + assert_eq!(parsed.groups, vec!["buzz-general"]); + assert_eq!(parsed.group_mention_mode, MentionMode::Always); + } + + #[test] + fn redacted_config_includes_group_fields() { + let cfg = NostrAccountConfig { + groups: vec!["buzz-general".into()], + ..Default::default() + }; + let json = serde_json::to_value(RedactedConfig(&cfg)).unwrap_or_default(); + assert_eq!(json["secret_key"], "[REDACTED]"); + assert!(json["groups"].is_array()); + assert!(json.get("group_mention_mode").is_some()); + } } diff --git a/crates/nostr/src/groups.rs b/crates/nostr/src/groups.rs new file mode 100644 index 0000000000..c3b1c7c9b1 --- /dev/null +++ b/crates/nostr/src/groups.rs @@ -0,0 +1,657 @@ +//! NIP-29 relay-based group chat — support for [Buzz](https://github.com/block/buzz). +//! +//! Block's Buzz is a Nostr relay where AI agents and humans collaborate in +//! group "channels". Its primary API is +//! [NIP-29](https://github.com/nostr-protocol/nips/blob/master/29.md) group +//! chat: `kind:9` chat messages scoped to a group via an `h` tag, delivered +//! over a relay connection authenticated with +//! [NIP-42](https://github.com/nostr-protocol/nips/blob/master/42.md). +//! +//! This module lets the Moltis Nostr channel join those groups, decide when to +//! respond, and post threaded replies. It speaks plain NIP-29 — the same wire +//! protocol any NIP-29 client uses — so it works against Buzz relays and +//! generic NIP-29 relays alike. Unlike DMs (`gift_wrap`), group messages are +//! **not** encrypted; the relay enforces membership and authorization. + +use { + moltis_channels::gating::MentionMode, + nostr_sdk::prelude::{ + Alphabet, Client, Event, EventBuilder, EventId, Kind, PublicKey, SingleLetterTag, Tag, + TagKind, + }, +}; + +use crate::error::Error; + +/// Standard NIP-29 group chat message kind (`kind:9`). +pub const GROUP_CHAT_KIND: u16 = 9; + +/// Buzz's stream message kind (`kind:40002`, `KIND_STREAM_MESSAGE_V2`). +/// +/// Buzz defines both `kind:9` and this v2 kind, and posts channel messages as +/// v2. Per its architecture notes, "clients filtering by kind 9 alone would not +/// receive kind 40002 messages, and vice versa" — so a bot that only speaks +/// `kind:9` is invisible in a Buzz channel, and vice versa. We therefore read +/// both and answer in whichever dialect we were addressed in. +pub const BUZZ_STREAM_MESSAGE_V2_KIND: u16 = 40002; + +/// The standard NIP-29 group chat message [`Kind`] (`kind:9`). +#[must_use] +pub fn group_chat_kind() -> Kind { + Kind::from_u16(GROUP_CHAT_KIND) +} + +/// Buzz's stream message [`Kind`] (`kind:40002`). +#[must_use] +pub fn buzz_stream_message_kind() -> Kind { + Kind::from_u16(BUZZ_STREAM_MESSAGE_V2_KIND) +} + +/// Every message kind treated as group chat, for subscription filters. +#[must_use] +pub fn group_message_kinds() -> [Kind; 2] { + [group_chat_kind(), buzz_stream_message_kind()] +} + +/// Whether `kind` is one this module handles as a group chat message. +#[must_use] +pub fn is_group_message_kind(kind: Kind) -> bool { + group_message_kinds().contains(&kind) +} + +/// Which dialect to use for group messages that are not replies. +/// +/// Replies mirror the kind of the message they answer, so this only decides +/// cold-start sends into a group we have not heard from yet. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum GroupMessageKind { + /// Standard NIP-29 `kind:9` — the interoperable default. + #[default] + Nip29, + /// Buzz `kind:40002` — set this when the relay is a Buzz workspace. + BuzzV2, +} + +impl From for Kind { + fn from(value: GroupMessageKind) -> Self { + match value { + GroupMessageKind::Nip29 => group_chat_kind(), + GroupMessageKind::BuzzV2 => buzz_stream_message_kind(), + } + } +} + +/// The single-letter `h` tag NIP-29 uses to scope an event to a group. +#[must_use] +pub fn h_tag() -> SingleLetterTag { + SingleLetterTag::lowercase(Alphabet::H) +} + +/// Extract the NIP-29 group id (the `h` tag value) from an event. +#[must_use] +pub fn extract_group_id(event: &Event) -> Option { + event + .tags + .find(TagKind::h()) + .and_then(|t| t.content()) + .map(str::to_owned) +} + +/// Whether the bot's pubkey is tagged (`p` tag) in the event — a NIP-29 mention. +#[must_use] +pub fn is_bot_mentioned(event: &Event, bot: &PublicKey) -> bool { + event.tags.public_keys().any(|pk| pk == bot) +} + +/// Decide whether the bot should respond to a group message under the +/// configured [`MentionMode`]. +#[must_use] +pub fn should_respond(mode: &MentionMode, event: &Event, bot: &PublicKey) -> bool { + match mode { + MentionMode::Always => true, + MentionMode::None => false, + MentionMode::Mention => is_bot_mentioned(event, bot), + } +} + +/// Reason a group message was rejected by access control. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum GroupDenied { + /// No groups are configured, so group chat is off for this account. + Disabled, + /// The event's group is not one this account joined. + NotJoined, +} + +impl std::fmt::Display for GroupDenied { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Disabled => write!(f, "no groups configured"), + Self::NotJoined => write!(f, "group not joined"), + } + } +} + +/// Check that an inbound group message belongs to a group this account joined. +/// +/// The configured `joined` list is authoritative: it is the only set of groups +/// we ever subscribe to, so anything else must be rejected. This is a security +/// boundary, not a convenience filter — a relay is untrusted and can push any +/// event it likes down the socket, including a `kind:9` carrying an arbitrary +/// `h` tag. The `h` tag is an unauthenticated label (the signature proves who +/// wrote the event, not which group it belongs to), so without this check a +/// hostile or buggy relay could inject text straight into the agent. +/// +/// An empty `joined` list therefore denies everything rather than allowing +/// everything — the opposite of the `moltis_channels::gating::is_allowed` +/// convention used for optional allowlists, because here the list defines +/// membership itself rather than filtering within it. +pub fn check_group_access(group_id: &str, joined: &[String]) -> Result<(), GroupDenied> { + if joined.is_empty() { + return Err(GroupDenied::Disabled); + } + if joined.iter().any(|g| g == group_id) { + Ok(()) + } else { + Err(GroupDenied::NotJoined) + } +} + +/// Marks a reply target as a group rather than a DM recipient. +/// +/// Outbound only receives a `chat_id`, and a NIP-29 group id is an opaque +/// string that can perfectly well be 64 hex characters — which is also exactly +/// what a Nostr public key looks like. Classifying by "is this id currently in +/// `groups`?" therefore breaks the moment the group is removed: the reply +/// target persists in the session (`_channel_reply_target`), so a queued or +/// resumed turn would fall through to the DM path, parse the former group id as +/// a pubkey, and gift-wrap the channel's reply to whoever owns that key. +/// +/// Prefixing removes the ambiguity: the target carries its own kind, so +/// classification never depends on mutable config, and a stale group target +/// fails closed instead of being misdelivered. +pub const GROUP_TARGET_PREFIX: &str = "group:"; + +/// Build the reply-target chat id for a group. +#[must_use] +pub fn group_target(group_id: &str) -> String { + format!("{GROUP_TARGET_PREFIX}{group_id}") +} + +/// Recover the group id from a reply target, if it is one. +#[must_use] +pub fn parse_group_target(target: &str) -> Option<&str> { + target.strip_prefix(GROUP_TARGET_PREFIX) +} + +/// Channel commands that must not be reachable by an arbitrary group member. +/// +/// Group chat has no per-sender allowlist by design — NIP-29 makes the relay +/// the authority on membership. That is fine for conversation, but these two +/// commands control *code execution*: `/sh` turns subsequent messages in the +/// session into shell commands, and `/sandbox` can move execution off the +/// sandbox onto the host. Anyone the relay admits to a joined channel could +/// otherwise reach them. +/// +/// The gateway already gates `/approve`, `/deny` and `/update` centrally, so +/// they are deliberately not repeated here. +pub const OPERATOR_ONLY_COMMANDS: &[&str] = &["sh", "sandbox"]; + +/// Whether `command` may only be run by an operator (see +/// [`OPERATOR_ONLY_COMMANDS`]). +#[must_use] +pub fn is_operator_only_command(command: &str) -> bool { + OPERATOR_ONLY_COMMANDS.contains(&command) +} + +/// Whether a group sender counts as an operator of this account. +/// +/// Reuses the account's DM allowlist (`allowed_pubkeys`) as the operator list +/// rather than inventing a second one: those are the keys the owner already +/// nominated as trusted. An empty allowlist means nobody is an operator, so +/// execution commands fail closed in groups. +#[must_use] +pub fn is_group_operator(sender: &PublicKey, allowed: &[PublicKey]) -> bool { + allowed.contains(sender) +} + +/// Build the tags for an outbound group message: the required `h` tag scoping +/// it to the group, plus an optional NIP-10 reply `e` tag and an author `p` tag +/// mentioning the person being replied to. +/// +/// The reply tag carries the NIP-10 `"reply"` marker (`["e", , "", "reply"]`) +/// rather than a bare `["e", ]`, matching how Buzz's own SDK builds threaded +/// messages — an unmarked `e` tag is not recognised as a threaded reply. +#[must_use] +pub fn build_group_message_tags( + group_id: &str, + reply_to: Option, + mention: Option, +) -> Vec { + let mut tags = vec![Tag::custom(TagKind::h(), [group_id.to_string()])]; + if let Some(event_id) = reply_to { + tags.push(Tag::custom(TagKind::e(), [ + event_id.to_hex(), + String::new(), + "reply".to_string(), + ])); + } + if let Some(pubkey) = mention { + tags.push(Tag::public_key(pubkey)); + } + tags +} + +/// Publish a plaintext group chat message to the relay. +/// +/// `kind` selects the dialect — `kind:9` for standard NIP-29 or `kind:40002` +/// for Buzz. The event is signed by the client's configured signer (the bot +/// keys) and carries the `h` tag routing it to `group_id`. When `reply_to` is +/// set, a NIP-10 `e` tag threads the reply to the originating message, and +/// `mention` `p`-tags its author so they are notified. +pub async fn send_group_message( + client: &Client, + kind: Kind, + group_id: &str, + text: &str, + reply_to: Option, + mention: Option, +) -> Result { + let tags = build_group_message_tags(group_id, reply_to, mention); + let builder = EventBuilder::new(kind, text).tags(tags); + let output = client.send_event_builder(builder).await?; + require_accepted(output) +} + +/// Treat an event that no relay accepted as a failure. +/// +/// `RelayPool::send_event_to` returns `Ok` even when every relay answered +/// `OK false` — the rejections are collected into `output.failed` and the call +/// still succeeds. That default is dangerous for group chat, where rejection is +/// the *expected* response to a routine misconfiguration: a NIP-29 relay +/// refuses posts from a pubkey it has not admitted to the group +/// (`"restricted: ..."`). Without this check the bot would report success, +/// acknowledge the user with a ✅, and say nothing in the channel. +fn require_accepted(output: nostr_sdk::prelude::Output) -> Result { + if output.success.is_empty() { + let reasons = output + .failed + .iter() + .map(|(url, err)| format!("{url}: {err}")) + .collect::>() + .join("; "); + return Err(Error::Relay(if reasons.is_empty() { + "no relay accepted the event".to_string() + } else { + format!("all relays rejected the event ({reasons})") + })); + } + Ok(*output.id()) +} + +/// Buzz's stream-message edit kind (`kind:40003`, `KIND_STREAM_MESSAGE_EDIT`). +pub const BUZZ_STREAM_MESSAGE_EDIT_KIND: u16 = 40_003; + +/// The edit [`Kind`] (`kind:40003`). +#[must_use] +pub fn buzz_edit_kind() -> Kind { + Kind::from_u16(BUZZ_STREAM_MESSAGE_EDIT_KIND) +} + +/// Whether messages in this dialect can be edited in place. +/// +/// Editing is a Buzz extension (`kind:40003`); a plain NIP-29 relay has no +/// equivalent, so callers must fall back to posting a single final message. +#[must_use] +pub fn supports_edit(kind: Kind) -> bool { + kind == buzz_stream_message_kind() +} + +/// Replace the text of a previously published group message (Buzz `kind:40003`). +/// +/// Tags mirror Buzz's own builder exactly: `["h", ]` and +/// `["e", ]`, with the full replacement text as the content. +pub async fn edit_group_message( + client: &Client, + group_id: &str, + target: EventId, + text: &str, +) -> Result<(), Error> { + let builder = EventBuilder::new(buzz_edit_kind(), text).tags([ + Tag::custom(TagKind::h(), [group_id.to_string()]), + Tag::custom(TagKind::e(), [target.to_hex()]), + ]); + let output = client.send_event_builder(builder).await?; + require_accepted(output)?; + Ok(()) +} + +/// React to a message with an emoji (NIP-25 `kind:7`). +/// +/// The emoji goes in the content and `["e", ]` names the message, as in +/// Buzz's own builder. The `["h", ]` tag is additional: NIP-29 scopes +/// group events by `h`, and a strict relay will refuse an unscoped write, so +/// omitting it would make acknowledgements fail everywhere except Buzz. Buzz +/// itself does not require the tag, and an extra tag is inert there. +/// +/// Returns the reaction's own event id, which is what a later NIP-09 deletion +/// has to reference. +pub async fn send_reaction( + client: &Client, + group_id: &str, + target: EventId, + emoji: &str, +) -> Result { + let builder = EventBuilder::new(Kind::Reaction, emoji).tags([ + Tag::custom(TagKind::h(), [group_id.to_string()]), + Tag::custom(TagKind::e(), [target.to_hex()]), + ]); + let output = client.send_event_builder(builder).await?; + require_accepted(output) +} + +/// Request deletion of an event we published (NIP-09 `kind:5`). +/// +/// Used to retract an acknowledgement reaction once the turn finishes. Nostr +/// deletion is a *request* — relays honour it at their discretion. Carries the +/// `h` tag for the same NIP-29 scoping reason as [`send_reaction`]. +pub async fn delete_event(client: &Client, group_id: &str, target: EventId) -> Result<(), Error> { + let builder = EventBuilder::new(Kind::EventDeletion, "").tags([ + Tag::custom(TagKind::h(), [group_id.to_string()]), + Tag::custom(TagKind::e(), [target.to_hex()]), + ]); + let output = client.send_event_builder(builder).await?; + require_accepted(output)?; + Ok(()) +} + +/// Translate the gateway's Slack-style ack shortcodes into emoji glyphs. +/// +/// The gateway emits shortcodes (`eyes`, `white_check_mark`, `x`) because Slack's +/// Web API wants them, but NIP-25 expects the literal glyph (or `+`/`-`). +/// Anything unrecognised is passed through, so callers can supply a glyph +/// directly. +#[must_use] +pub fn ack_emoji_glyph(shortcode: &str) -> &str { + match shortcode { + "eyes" => "👀", + "white_check_mark" | "heavy_check_mark" => "✅", + "x" | "negative_squared_cross_mark" => "❌", + other => other, + } +} + +#[cfg(test)] +#[allow(clippy::expect_used)] +mod tests { + use nostr_sdk::prelude::{EventBuilder, Keys}; + + use super::*; + + /// Build a signed `kind:9` group event with the given `h` tag and optional + /// `p` mention, for exercising the pure extraction/gating helpers. + fn group_event(group_id: &str, mention: Option) -> Event { + let keys = Keys::generate(); + let tags = build_group_message_tags(group_id, None, mention); + EventBuilder::new(group_chat_kind(), "hello") + .tags(tags) + .sign_with_keys(&keys) + .expect("sign group event") + } + + #[test] + fn group_chat_kind_is_nine() { + assert_eq!(group_chat_kind().as_u16(), 9); + } + + /// `RelayPool::send_event_to` returns `Ok` even when every relay answered + /// `OK false`, so a NIP-29 relay refusing a post from a pubkey it has not + /// admitted would otherwise look like a successful send — the bot would ✅ + /// the user and say nothing in the channel. + #[test] + fn rejection_by_every_relay_is_an_error() { + use std::collections::{HashMap, HashSet}; + + use nostr_sdk::prelude::{Output, RelayUrl}; + + let url = RelayUrl::parse("wss://relay.example").expect("parse relay url"); + let rejected: Output = Output { + val: EventId::all_zeros(), + success: HashSet::new(), + failed: HashMap::from([(url, "restricted: not a member of the group".to_string())]), + }; + + let err = require_accepted(rejected).expect_err("all-rejected must fail"); + let msg = err.to_string(); + assert!(msg.contains("rejected"), "unexpected message: {msg}"); + // The relay's reason has to survive, or the operator cannot diagnose it. + assert!(msg.contains("not a member"), "reason lost: {msg}"); + } + + /// One accepting relay is enough — partial failure is normal on Nostr. + #[test] + fn acceptance_by_any_relay_succeeds() { + use std::collections::{HashMap, HashSet}; + + use nostr_sdk::prelude::{Output, RelayUrl}; + + let good = RelayUrl::parse("wss://good.example").expect("parse"); + let bad = RelayUrl::parse("wss://bad.example").expect("parse"); + let id = EventId::all_zeros(); + let mixed: Output = Output { + val: id, + success: HashSet::from([good]), + failed: HashMap::from([(bad, "rate-limited".to_string())]), + }; + + assert_eq!(require_accepted(mixed).ok(), Some(id)); + } + + #[test] + fn buzz_stream_message_kind_is_40002() { + assert_eq!(buzz_stream_message_kind().as_u16(), 40_002); + } + + /// Buzz clients filtering kind:40002 never see kind:9 and vice versa, so + /// both must be accepted — anything else is not group chat. + #[test] + fn both_dialects_are_group_messages() { + assert!(is_group_message_kind(group_chat_kind())); + assert!(is_group_message_kind(buzz_stream_message_kind())); + assert!(!is_group_message_kind(Kind::from_u16(1))); + assert!(!is_group_message_kind(Kind::from_u16(40_003))); + assert_eq!(group_message_kinds().len(), 2); + } + + #[test] + fn message_kind_config_maps_to_wire_kind() { + assert_eq!(Kind::from(GroupMessageKind::Nip29), group_chat_kind()); + assert_eq!( + Kind::from(GroupMessageKind::BuzzV2), + buzz_stream_message_kind() + ); + // Default stays on the interoperable standard kind. + assert_eq!(GroupMessageKind::default(), GroupMessageKind::Nip29); + } + + #[test] + fn message_kind_serde_round_trip() { + for value in [GroupMessageKind::Nip29, GroupMessageKind::BuzzV2] { + let json = serde_json::to_value(value).unwrap_or_default(); + let parsed: GroupMessageKind = serde_json::from_value(json).unwrap_or_default(); + assert_eq!(parsed, value); + } + let from_str: GroupMessageKind = + serde_json::from_value(serde_json::json!("buzz_v2")).unwrap_or_default(); + assert_eq!(from_str, GroupMessageKind::BuzzV2); + } + + #[test] + fn extracts_group_id_from_h_tag() { + let event = group_event("buzz-general", None); + assert_eq!(extract_group_id(&event).as_deref(), Some("buzz-general")); + } + + #[test] + fn no_group_id_when_h_tag_absent() { + let keys = Keys::generate(); + let event = EventBuilder::new(group_chat_kind(), "hi") + .sign_with_keys(&keys) + .expect("sign"); + assert_eq!(extract_group_id(&event), None); + } + + #[test] + fn detects_mention_via_p_tag() { + let bot = Keys::generate().public_key(); + let mentioned = group_event("g1", Some(bot)); + let not_mentioned = group_event("g1", None); + assert!(is_bot_mentioned(&mentioned, &bot)); + assert!(!is_bot_mentioned(¬_mentioned, &bot)); + } + + #[test] + fn should_respond_honours_mention_mode() { + let bot = Keys::generate().public_key(); + let mentioned = group_event("g1", Some(bot)); + let plain = group_event("g1", None); + + assert!(should_respond(&MentionMode::Always, &plain, &bot)); + assert!(!should_respond(&MentionMode::None, &mentioned, &bot)); + assert!(should_respond(&MentionMode::Mention, &mentioned, &bot)); + assert!(!should_respond(&MentionMode::Mention, &plain, &bot)); + } + + #[test] + fn group_access_allows_joined_group() { + let joined = vec!["buzz-general".to_string(), "buzz-dev".to_string()]; + assert!(check_group_access("buzz-general", &joined).is_ok()); + assert!(check_group_access("buzz-dev", &joined).is_ok()); + } + + /// An empty join list denies everything — group chat is simply off. + #[test] + fn group_access_empty_join_list_denies_all() { + assert_eq!( + check_group_access("anything", &[]), + Err(GroupDenied::Disabled) + ); + } + + /// A relay can push any event down the socket, so a `kind:9` for a group + /// we never joined must be rejected rather than fed to the agent. + #[test] + fn group_access_rejects_unjoined_group_from_hostile_relay() { + let joined = vec!["buzz-general".to_string()]; + assert_eq!( + check_group_access("attacker-controlled-group", &joined), + Err(GroupDenied::NotJoined) + ); + } + + /// Group ids are matched exactly — no glob or case folding, since an `h` + /// tag is an opaque identifier rather than a user-facing handle. + #[test] + fn group_access_matches_exactly() { + let joined = vec!["buzz-general".to_string()]; + assert_eq!( + check_group_access("buzz-general-2", &joined), + Err(GroupDenied::NotJoined) + ); + assert_eq!( + check_group_access("BUZZ-GENERAL", &joined), + Err(GroupDenied::NotJoined) + ); + assert_eq!( + check_group_access("*", &joined), + Err(GroupDenied::NotJoined) + ); + } + + #[test] + fn build_tags_includes_h_and_reply() { + let author = Keys::generate().public_key(); + let event_id = EventId::all_zeros(); + let tags = build_group_message_tags("grp", Some(event_id), Some(author)); + // h tag is always first. + assert_eq!(tags[0].content(), Some("grp")); + // e tag and p tag present. + assert!(tags.iter().any(|t| t.kind() == TagKind::e())); + assert!(tags.iter().any(|t| t.kind() == TagKind::p())); + } + + /// Buzz threads on the NIP-10 `"reply"` marker; a bare `["e", id]` is not + /// recognised as a threaded reply. + #[test] + fn reply_tag_carries_nip10_reply_marker() { + let event_id = EventId::all_zeros(); + let tags = build_group_message_tags("grp", Some(event_id), None); + let reply = tags + .iter() + .find(|t| t.kind() == TagKind::e()) + .expect("e tag present"); + assert_eq!(reply.as_slice(), &[ + "e".to_string(), + event_id.to_hex(), + String::new(), + "reply".to_string() + ]); + } + + #[test] + fn edit_kind_is_40003_and_only_buzz_supports_it() { + assert_eq!(buzz_edit_kind().as_u16(), 40_003); + assert!(supports_edit(buzz_stream_message_kind())); + // Plain NIP-29 has no edit kind, so streaming must not try to edit. + assert!(!supports_edit(group_chat_kind())); + } + + /// Only the execution-controlling commands are operator-gated; ordinary + /// ones must stay usable by anyone in the channel. + #[test] + fn only_execution_commands_are_operator_only() { + assert!(is_operator_only_command("sh")); + assert!(is_operator_only_command("sandbox")); + for open_cmd in ["help", "new", "clear", "context", "model", "compact"] { + assert!( + !is_operator_only_command(open_cmd), + "/{open_cmd} should not be operator-gated" + ); + } + } + + /// The DM allowlist doubles as the operator list; empty means nobody, so + /// execution commands fail closed rather than open. + #[test] + fn group_operator_is_drawn_from_the_allowlist() { + let operator = Keys::generate().public_key(); + let stranger = Keys::generate().public_key(); + let allowed = vec![operator]; + + assert!(is_group_operator(&operator, &allowed)); + assert!(!is_group_operator(&stranger, &allowed)); + assert!( + !is_group_operator(&operator, &[]), + "an empty allowlist must not make everyone an operator" + ); + } + + #[test] + fn ack_shortcodes_map_to_nip25_glyphs() { + // The gateway speaks Slack shortcodes; NIP-25 wants the glyph. + assert_eq!(ack_emoji_glyph("eyes"), "👀"); + assert_eq!(ack_emoji_glyph("white_check_mark"), "✅"); + assert_eq!(ack_emoji_glyph("x"), "❌"); + // Already a glyph, or unknown: pass through untouched. + assert_eq!(ack_emoji_glyph("🎉"), "🎉"); + assert_eq!(ack_emoji_glyph("+"), "+"); + } + + #[test] + fn build_tags_h_only_when_no_reply() { + let tags = build_group_message_tags("grp", None, None); + assert_eq!(tags.len(), 1); + assert_eq!(tags[0].content(), Some("grp")); + } +} diff --git a/crates/nostr/src/lib.rs b/crates/nostr/src/lib.rs index 5ac82ec7d8..1e109fa66f 100644 --- a/crates/nostr/src/lib.rs +++ b/crates/nostr/src/lib.rs @@ -3,10 +3,12 @@ pub mod bus; pub mod config; pub mod error; pub mod gift_wrap; +pub mod groups; pub mod keys; pub mod outbound; pub mod plugin; pub mod profile; +pub mod reply_ctx; pub mod seen; pub mod state; diff --git a/crates/nostr/src/outbound.rs b/crates/nostr/src/outbound.rs index 90474e6308..3ad42c491a 100644 --- a/crates/nostr/src/outbound.rs +++ b/crates/nostr/src/outbound.rs @@ -1,7 +1,10 @@ //! Outbound message sending for Nostr channels. //! -//! Implements `ChannelOutbound` and `ChannelStreamOutbound` — sends text as -//! NIP-59 gift-wrapped DMs (kind:1059) to connected relays. +//! Implements `ChannelOutbound` and `ChannelStreamOutbound`. A send target is +//! either a configured NIP-29 group id — published as a plaintext kind:9 group +//! chat message (e.g. a Buzz channel reply) — or a pubkey, sent as a NIP-59 +//! gift-wrapped DM (kind:1059). Routing is decided by whether the target +//! matches one of the account's configured `groups`. use std::{ collections::HashMap, @@ -35,21 +38,155 @@ pub struct NostrOutbound { pub accounts: AccountStateMap, } +/// Where an outbound message should go: a NIP-29 group or a DM recipient. +enum SendTarget { + /// A configured group id (`h` tag value). + Group(String), + /// A DM recipient pubkey. + Dm(PublicKey), +} + +/// How to publish a group reply: which dialect, and whom to notify. +struct GroupSendPlan { + kind: Kind, + mention: Option, +} + +impl NostrOutbound { + /// Decide the dialect and `p`-tag for a group send. + /// + /// Prefer the exact message being replied to (mirrors its kind and + /// notifies its author), then the last kind seen in that group, then the + /// account's configured default. Buzz uses `kind:40002` while plain NIP-29 + /// uses `kind:9`, and a client filtering one never sees the other — so + /// answering in the wrong dialect makes the bot invisible. + fn plan_group_send( + &self, + account_id: &str, + group_id: &str, + reply_to: Option<&str>, + ) -> GroupSendPlan { + let accounts = self.accounts.read().unwrap_or_else(|e| e.into_inner()); + let Some(state) = accounts.get(account_id) else { + return GroupSendPlan { + kind: crate::groups::group_chat_kind(), + mention: None, + }; + }; + + let reply_event = reply_to.and_then(|id| EventId::parse(id).ok()); + let ctx = reply_event.and_then(|id| { + let ctxs = state.reply_ctx.lock().unwrap_or_else(|e| e.into_inner()); + ctxs.get(&id) + }); + if let Some(ctx) = ctx { + return GroupSendPlan { + kind: ctx.kind, + mention: Some(ctx.author), + }; + } + + let learned = { + let ctxs = state.reply_ctx.lock().unwrap_or_else(|e| e.into_inner()); + ctxs.kind_for_group(group_id) + }; + let kind = learned.unwrap_or_else(|| { + let cfg = state.config.read().unwrap_or_else(|e| e.into_inner()); + Kind::from(cfg.group_message_kind) + }); + GroupSendPlan { + kind, + mention: None, + } + } +} + +/// Minimum gap between edit events while streaming a reply. +/// +/// Every token would otherwise be a signed event on the relay; this keeps the +/// reply visibly live without flooding the channel or the audit log. +const STREAM_EDIT_INTERVAL: std::time::Duration = std::time::Duration::from_millis(900); + impl NostrOutbound { - /// Look up account state and resolve the recipient pubkey. + /// Drain a stream and send the result as one message. + /// + /// Used for DMs and plain NIP-29 groups, neither of which can edit in place. + async fn collect_then_send( + &self, + account_id: &str, + to: &str, + reply_to: Option<&str>, + stream: StreamReceiver, + ) -> ChannelResult<()> { + self.collect_remaining_then_send(account_id, to, reply_to, stream, String::new()) + .await + } + + /// Same as [`Self::collect_then_send`], but keeping text already received. + async fn collect_remaining_then_send( + &self, + account_id: &str, + to: &str, + reply_to: Option<&str>, + mut stream: StreamReceiver, + mut buffer: String, + ) -> ChannelResult<()> { + while let Some(event) = stream.recv().await { + match event { + StreamEvent::Delta(chunk) | StreamEvent::ProgressDelta(chunk) => { + buffer.push_str(&chunk); + }, + StreamEvent::Done => break, + StreamEvent::Error(e) => { + tracing::warn!(account_id, "stream error: {e}"); + if buffer.is_empty() { + buffer.push_str("[Error generating response]"); + } + break; + }, + } + } + + if !buffer.is_empty() { + self.send_text(account_id, to, &buffer, reply_to).await?; + } + Ok(()) + } + + /// Look up account state and resolve the target: a configured group id is + /// treated as a NIP-29 group send; anything else is parsed as a DM pubkey. async fn resolve( &self, account_id: &str, to: &str, - ) -> ChannelResult<(Client, Keys, PublicKey)> { + ) -> ChannelResult<(Client, Keys, SendTarget)> { let accounts = self.accounts.read().unwrap_or_else(|e| e.into_inner()); let state = accounts.get(account_id).ok_or_else(|| { moltis_channels::Error::unavailable(format!("nostr account not found: {account_id}")) })?; + let client = state.client.clone(); + let keys = state.keys.clone(); + + // A prefixed target is a group, full stop — never reinterpret it as a + // pubkey. Membership is then re-checked so a group removed from the + // config fails closed rather than being delivered anywhere. + if let Some(group_id) = crate::groups::parse_group_target(to) { + let still_joined = { + let cfg = state.config.read().unwrap_or_else(|e| e.into_inner()); + cfg.groups.iter().any(|g| g == group_id) + }; + if !still_joined { + return Err(moltis_channels::Error::unavailable(format!( + "nostr group no longer configured: {group_id}" + ))); + } + return Ok((client, keys, SendTarget::Group(group_id.to_string()))); + } + let recipient = PublicKey::parse(to).map_err(|e| { moltis_channels::Error::invalid_input(format!("invalid recipient pubkey: {e}")) })?; - Ok((state.client.clone(), state.keys.clone(), recipient)) + Ok((client, keys, SendTarget::Dm(recipient))) } } @@ -60,22 +197,57 @@ impl ChannelOutbound for NostrOutbound { account_id: &str, to: &str, text: &str, - _reply_to: Option<&str>, + reply_to: Option<&str>, ) -> ChannelResult<()> { - let (client, keys, recipient) = self.resolve(account_id, to).await?; + let (client, keys, target) = self.resolve(account_id, to).await?; #[cfg(feature = "metrics")] let start = tokio::time::Instant::now(); - // Send as NIP-59 gift-wrapped DM (kind:1059) - crate::gift_wrap::send_gift_wrapped_dm(&client, &keys, &recipient, text) - .await - .map_err(|e| { - #[cfg(feature = "metrics")] - counter!(nostr_metrics::MESSAGE_SEND_ERRORS_TOTAL, "reason" => "gift_wrap") - .increment(1); - moltis_channels::Error::external("nostr", e) - })?; + match target { + SendTarget::Group(group_id) => { + // Group chat reply. `reply_to` is the inbound event id, + // threaded via a NIP-10 `e` tag; the plan mirrors that + // message's kind and `p`-tags its author so they are notified. + let reply_event = reply_to.and_then(|id| EventId::parse(id).ok()); + let plan = self.plan_group_send(account_id, &group_id, reply_to); + crate::groups::send_group_message( + &client, + plan.kind, + &group_id, + text, + reply_event, + plan.mention, + ) + .await + .map_err(|e| { + #[cfg(feature = "metrics")] + counter!(nostr_metrics::MESSAGE_SEND_ERRORS_TOTAL, "reason" => "group") + .increment(1); + moltis_channels::Error::external("nostr", e) + })?; + tracing::debug!( + account_id, + group = %group_id, + kind = plan.kind.as_u16(), + len = text.len(), + "sent group message" + ); + }, + SendTarget::Dm(recipient) => { + // NIP-59 gift-wrapped DM (kind:1059). + crate::gift_wrap::send_gift_wrapped_dm(&client, &keys, &recipient, text) + .await + .map_err(|e| { + #[cfg(feature = "metrics")] + counter!(nostr_metrics::MESSAGE_SEND_ERRORS_TOTAL, "reason" => "gift_wrap") + .increment(1); + moltis_channels::Error::external("nostr", e) + })?; + let npub = recipient.to_bech32().unwrap_or_else(|_| recipient.to_hex()); + tracing::debug!(account_id, to = %npub, len = text.len(), "sent gift-wrapped DM"); + }, + } #[cfg(feature = "metrics")] { @@ -84,9 +256,78 @@ impl ChannelOutbound for NostrOutbound { .record(start.elapsed().as_secs_f64()); } - let npub = recipient.to_bech32().unwrap_or_else(|_| recipient.to_hex()); - tracing::debug!(account_id, to = %npub, len = text.len(), "sent gift-wrapped DM"); + Ok(()) + } + + /// React to a group message (NIP-25 `kind:7`), used for ack reactions. + /// + /// The gateway passes Slack-style shortcodes, which are translated to the + /// emoji glyph NIP-25 expects. Reactions only make sense on group messages; + /// DMs are gift-wrapped, so reacting would leak the conversation. + async fn add_reaction( + &self, + account_id: &str, + channel_id: &str, + message_id: &str, + emoji: &str, + ) -> ChannelResult<()> { + let Ok((client, _, SendTarget::Group(_))) = self.resolve(account_id, channel_id).await + else { + return Ok(()); + }; + let Ok(target) = EventId::parse(message_id) else { + return Ok(()); + }; + + let glyph = crate::groups::ack_emoji_glyph(emoji); + let reaction = crate::groups::send_reaction(&client, channel_id, target, glyph) + .await + .map_err(|e| moltis_channels::Error::external("nostr", e))?; + // Remember it so `remove_reaction` can retract it via NIP-09. + let accounts = self.accounts.read().unwrap_or_else(|e| e.into_inner()); + if let Some(state) = accounts.get(account_id) { + let mut ctxs = state.reply_ctx.lock().unwrap_or_else(|e| e.into_inner()); + ctxs.record_reaction(target, glyph, reaction); + } + Ok(()) + } + + /// Retract a reaction by publishing a NIP-09 deletion (`kind:5`) for it. + /// + /// Nostr has no "unreact", and deletion is only a request the relay may + /// honour — so this is best-effort and silently no-ops when the reaction + /// id is no longer known. + async fn remove_reaction( + &self, + account_id: &str, + channel_id: &str, + message_id: &str, + emoji: &str, + ) -> ChannelResult<()> { + let Ok((client, _, SendTarget::Group(_))) = self.resolve(account_id, channel_id).await + else { + return Ok(()); + }; + let Ok(target) = EventId::parse(message_id) else { + return Ok(()); + }; + let glyph = crate::groups::ack_emoji_glyph(emoji); + + let reaction = { + let accounts = self.accounts.read().unwrap_or_else(|e| e.into_inner()); + accounts.get(account_id).and_then(|state| { + let mut ctxs = state.reply_ctx.lock().unwrap_or_else(|e| e.into_inner()); + ctxs.take_reaction(target, glyph) + }) + }; + let Some(reaction) = reaction else { + return Ok(()); + }; + + if let Err(e) = crate::groups::delete_event(&client, channel_id, reaction).await { + tracing::debug!(account_id, "failed to retract reaction: {e}"); + } Ok(()) } @@ -112,14 +353,33 @@ impl ChannelStreamOutbound for NostrOutbound { reply_to: Option<&str>, mut stream: StreamReceiver, ) -> ChannelResult<()> { - // Nostr doesn't support edit-in-place streaming. - // Collect all chunks and send as a single message. + // Buzz groups support edit-in-place (kind:40003), so the reply can be + // published early and revised as tokens arrive. Plain NIP-29 has no + // edit kind, and DMs are gift-wrapped, so both collect and send once. + let live = match self.resolve(account_id, to).await { + Ok((client, _, SendTarget::Group(group_id))) => { + let plan = self.plan_group_send(account_id, &group_id, reply_to); + crate::groups::supports_edit(plan.kind).then_some((client, group_id, plan)) + }, + _ => None, + }; + + let Some((client, group_id, plan)) = live else { + return self + .collect_then_send(account_id, to, reply_to, stream) + .await; + }; + + let reply_event = reply_to.and_then(|id| EventId::parse(id).ok()); let mut buffer = String::new(); + let mut published: Option = None; + let mut last_edit = tokio::time::Instant::now(); + let mut pending_edit = false; while let Some(event) = stream.recv().await { match event { StreamEvent::Delta(chunk) | StreamEvent::ProgressDelta(chunk) => { - buffer.push_str(&chunk) + buffer.push_str(&chunk); }, StreamEvent::Done => break, StreamEvent::Error(e) => { @@ -130,10 +390,83 @@ impl ChannelStreamOutbound for NostrOutbound { break; }, } + + if buffer.trim().is_empty() { + continue; + } + + match published { + // First non-empty chunk: publish so the channel sees a reply + // forming instead of waiting for the whole turn. + None => { + match crate::groups::send_group_message( + &client, + plan.kind, + &group_id, + &buffer, + reply_event, + plan.mention, + ) + .await + { + Ok(id) => { + published = Some(id); + last_edit = tokio::time::Instant::now(); + }, + Err(e) => { + tracing::warn!(account_id, "failed to publish streamed reply: {e}"); + return self + .collect_remaining_then_send( + account_id, to, reply_to, stream, buffer, + ) + .await; + }, + } + }, + // Throttle edits: every token would be an event on the relay. + Some(target) => { + if last_edit.elapsed() >= STREAM_EDIT_INTERVAL { + let result = + crate::groups::edit_group_message(&client, &group_id, target, &buffer) + .await; + if let Err(ref e) = result { + tracing::warn!(account_id, "streamed edit failed: {e}"); + } + last_edit = tokio::time::Instant::now(); + // Keep the flag set when the edit failed: it means + // "the buffer has moved on since what the relay holds", + // so clearing it here would skip the final edit and + // leave the reply permanently truncated. + pending_edit = result.is_err(); + } else { + pending_edit = true; + } + }, + } } - if !buffer.is_empty() { - self.send_text(account_id, to, &buffer, reply_to).await?; + match published { + // Always land the final text, even if the last chunks were + // throttled — otherwise the message stays truncated. + Some(target) => { + if pending_edit + && let Err(e) = + crate::groups::edit_group_message(&client, &group_id, target, &buffer).await + { + tracing::warn!(account_id, "final streamed edit failed: {e}"); + } + // Counted here rather than after the match: the initial publish + // went through `send_group_message`, which records no metrics, + // and the edits that follow are all the same logical message. + #[cfg(feature = "metrics")] + counter!(nostr_metrics::MESSAGES_SENT_TOTAL).increment(1); + }, + // `send_text` records its own metrics, so counting again here + // would double it — and an empty stream sent nothing at all. + None if !buffer.trim().is_empty() => { + self.send_text(account_id, to, &buffer, reply_to).await?; + }, + None => {}, } Ok(()) @@ -149,3 +482,283 @@ impl std::fmt::Debug for NostrOutbound { f.debug_struct("NostrOutbound").finish_non_exhaustive() } } + +#[cfg(test)] +mod tests { + use { + crate::{ + config::NostrAccountConfig, + groups::{GroupMessageKind, buzz_stream_message_kind, group_chat_kind}, + reply_ctx::ReplyContexts, + state::AccountState, + }, + tokio_util::sync::CancellationToken, + }; + + use super::*; + + /// Build an outbound adapter with one account whose config names `groups`. + /// No relay connection is made — `resolve` is pure routing logic. + fn outbound_with_groups(groups: Vec) -> NostrOutbound { + outbound_with_config(NostrAccountConfig { + groups, + ..Default::default() + }) + } + + fn outbound_with_config(config: NostrAccountConfig) -> NostrOutbound { + let keys = Keys::generate(); + let client = Client::new(keys.clone()); + let state = AccountState { + client, + keys, + config: Arc::new(RwLock::new(config)), + cached_allowlist: Arc::new(RwLock::new(Vec::new())), + cancel: CancellationToken::new(), + otp: Arc::new(std::sync::Mutex::new(moltis_channels::otp::OtpState::new( + 300, + ))), + reply_ctx: Arc::new(std::sync::Mutex::new(ReplyContexts::new())), + }; + let accounts: AccountStateMap = Arc::new(RwLock::new(HashMap::new())); + accounts + .write() + .unwrap_or_else(|e| e.into_inner()) + .insert("acct".to_string(), state); + NostrOutbound { accounts } + } + + #[tokio::test] + async fn resolves_configured_group_as_group_send() { + let outbound = outbound_with_groups(vec!["buzz-general".into()]); + let resolved = outbound + .resolve("acct", &crate::groups::group_target("buzz-general")) + .await; + assert!(matches!(resolved, Ok((_, _, SendTarget::Group(ref g))) if g == "buzz-general")); + } + + #[tokio::test] + async fn resolves_pubkey_as_dm_send() { + let outbound = outbound_with_groups(vec!["buzz-general".into()]); + let peer = Keys::generate().public_key().to_hex(); + let resolved = outbound.resolve("acct", &peer).await; + assert!(matches!(resolved, Ok((_, _, SendTarget::Dm(_))))); + } + + /// Turning group chat off means clearing `groups`, and outbound reads that + /// same list — so a queued reply cannot keep publishing to a group the + /// operator has since removed. + #[tokio::test] + async fn refuses_group_send_once_groups_cleared() { + let outbound = outbound_with_groups(Vec::new()); + let resolved = outbound + .resolve("acct", &crate::groups::group_target("buzz-general")) + .await; + assert!( + resolved.is_err(), + "group send must fail once the group is no longer configured" + ); + } + + /// A group id may be 64 hex characters, which is also exactly what a public + /// key looks like. If routing fell back to pubkey parsing for a removed + /// group, the channel's reply would be gift-wrapped to whoever owns that + /// key — so a stale group target must fail, never become a DM. + #[tokio::test] + async fn removed_group_whose_id_is_a_valid_pubkey_never_becomes_a_dm() { + // A real, parseable public key used as the group's `h` tag. + let pubkey_shaped_id = Keys::generate().public_key().to_hex(); + assert!( + PublicKey::parse(&pubkey_shaped_id).is_ok(), + "precondition: the group id really is a valid pubkey" + ); + + // Group removed from config while a reply target still points at it. + let outbound = outbound_with_groups(Vec::new()); + let resolved = outbound + .resolve("acct", &crate::groups::group_target(&pubkey_shaped_id)) + .await; + + assert!( + !matches!(resolved, Ok((_, _, SendTarget::Dm(_)))), + "a removed group must never be reinterpreted as a DM recipient" + ); + assert!(resolved.is_err(), "it must fail closed"); + } + + /// The same id, still configured, routes as a group — proving the guard + /// above rejects because of removal, not because of the id's shape. + #[tokio::test] + async fn pubkey_shaped_group_id_still_routes_as_a_group_when_configured() { + let pubkey_shaped_id = Keys::generate().public_key().to_hex(); + let outbound = outbound_with_groups(vec![pubkey_shaped_id.clone()]); + let resolved = outbound + .resolve("acct", &crate::groups::group_target(&pubkey_shaped_id)) + .await; + assert!( + matches!(resolved, Ok((_, _, SendTarget::Group(ref g))) if *g == pubkey_shaped_id), + "a configured group keeps routing as a group even when its id parses as a pubkey" + ); + } + + /// Record an inbound message so the reply can mirror it. + fn record_inbound(outbound: &NostrOutbound, group: &str, kind: Kind) -> (EventId, PublicKey) { + let event_id = EventId::from_byte_array(Keys::generate().public_key().to_bytes()); + let author = Keys::generate().public_key(); + let accounts = outbound.accounts.read().unwrap_or_else(|e| e.into_inner()); + if let Some(state) = accounts.get("acct") { + let mut ctxs = state.reply_ctx.lock().unwrap_or_else(|e| e.into_inner()); + ctxs.record(event_id, group, author, kind); + } + (event_id, author) + } + + /// A reply to a Buzz kind:40002 message must go back as kind:40002 — a + /// Buzz client filtering v2 would never see a kind:9 answer. + #[test] + fn reply_mirrors_buzz_dialect_and_tags_author() { + let outbound = outbound_with_groups(vec!["buzz-general".into()]); + let (event_id, author) = + record_inbound(&outbound, "buzz-general", buzz_stream_message_kind()); + + let plan = outbound.plan_group_send("acct", "buzz-general", Some(&event_id.to_hex())); + assert_eq!(plan.kind, buzz_stream_message_kind()); + assert_eq!(plan.mention, Some(author)); + } + + #[test] + fn reply_mirrors_nip29_dialect() { + let outbound = outbound_with_groups(vec!["grp".into()]); + let (event_id, author) = record_inbound(&outbound, "grp", group_chat_kind()); + + let plan = outbound.plan_group_send("acct", "grp", Some(&event_id.to_hex())); + assert_eq!(plan.kind, group_chat_kind()); + assert_eq!(plan.mention, Some(author)); + } + + /// With no specific message to answer, fall back to the dialect last seen + /// in that group rather than the configured default. + #[test] + fn non_reply_uses_dialect_learned_from_group() { + let outbound = outbound_with_groups(vec!["buzz-general".into()]); + record_inbound(&outbound, "buzz-general", buzz_stream_message_kind()); + + let plan = outbound.plan_group_send("acct", "buzz-general", None); + assert_eq!(plan.kind, buzz_stream_message_kind()); + // Nobody specific to notify on a proactive send. + assert_eq!(plan.mention, None); + } + + /// Cold start with nothing learned falls back to the configured default. + #[test] + fn cold_start_uses_configured_dialect() { + let nip29 = outbound_with_groups(vec!["grp".into()]); + assert_eq!( + nip29.plan_group_send("acct", "grp", None).kind, + group_chat_kind(), + "default config is the interoperable kind:9" + ); + + let buzz = outbound_with_config(NostrAccountConfig { + groups: vec!["grp".into()], + group_message_kind: GroupMessageKind::BuzzV2, + ..Default::default() + }); + assert_eq!( + buzz.plan_group_send("acct", "grp", None).kind, + buzz_stream_message_kind(), + "buzz_v2 config must post kind:40002" + ); + } + + /// An unparseable or unknown reply id must not panic or mis-tag. + #[test] + fn unknown_reply_id_falls_back_without_mention() { + let outbound = outbound_with_groups(vec!["grp".into()]); + let plan = outbound.plan_group_send("acct", "grp", Some("not-an-event-id")); + assert_eq!(plan.kind, group_chat_kind()); + assert_eq!(plan.mention, None); + } + + /// Streaming edits are a Buzz extension; a plain NIP-29 group must fall + /// back to collecting and sending once. + #[test] + fn only_buzz_dialect_streams_via_edits() { + let outbound = outbound_with_groups(vec!["grp".into()]); + record_inbound(&outbound, "grp", buzz_stream_message_kind()); + assert!(crate::groups::supports_edit( + outbound.plan_group_send("acct", "grp", None).kind + )); + + let plain = outbound_with_groups(vec!["grp".into()]); + record_inbound(&plain, "grp", group_chat_kind()); + assert!(!crate::groups::supports_edit( + plain.plan_group_send("acct", "grp", None).kind + )); + } + + /// Reacting to a DM would leak a gift-wrapped conversation, so it no-ops. + #[tokio::test] + async fn reactions_are_ignored_for_dm_targets() { + let outbound = outbound_with_groups(vec!["grp".into()]); + let peer = Keys::generate().public_key().to_hex(); + let event_id = EventId::from_byte_array(Keys::generate().public_key().to_bytes()); + let result = outbound + .add_reaction("acct", &peer, &event_id.to_hex(), "eyes") + .await; + assert!(result.is_ok(), "DM reaction must be a silent no-op"); + } + + /// A malformed message id must not panic or publish anything. + #[tokio::test] + async fn reactions_ignore_unparseable_message_id() { + let outbound = outbound_with_groups(vec!["grp".into()]); + assert!( + outbound + .add_reaction( + "acct", + &crate::groups::group_target("grp"), + "not-an-event-id", + "eyes" + ) + .await + .is_ok() + ); + assert!( + outbound + .remove_reaction( + "acct", + &crate::groups::group_target("grp"), + "not-an-event-id", + "eyes" + ) + .await + .is_ok() + ); + } + + /// Retracting a reaction we never recorded is a no-op, not an error. + #[tokio::test] + async fn removing_unknown_reaction_is_noop() { + let outbound = outbound_with_groups(vec!["grp".into()]); + let event_id = EventId::from_byte_array(Keys::generate().public_key().to_bytes()); + let result = outbound + .remove_reaction( + "acct", + &crate::groups::group_target("grp"), + &event_id.to_hex(), + "eyes", + ) + .await; + assert!(result.is_ok()); + } + + #[tokio::test] + async fn unknown_account_is_unavailable() { + let outbound = outbound_with_groups(vec!["buzz-general".into()]); + let resolved = outbound + .resolve("missing", &crate::groups::group_target("buzz-general")) + .await; + assert!(resolved.is_err(), "unknown account must not resolve"); + } +} diff --git a/crates/nostr/src/plugin.rs b/crates/nostr/src/plugin.rs index e9bedcd76e..791e335f8a 100644 --- a/crates/nostr/src/plugin.rs +++ b/crates/nostr/src/plugin.rs @@ -117,8 +117,24 @@ impl ChannelPlugin for NostrPlugin { "starting Nostr account" ); - // Build nostr-sdk client - let client = Client::new(bot_keys.clone()); + // Build the nostr-sdk client with both trust-relevant options pinned + // explicitly rather than inherited from upstream defaults: + // + // * `verify_subscriptions` (off by default) makes the SDK drop events + // that do not match the filters we asked for. Relays are untrusted + // and can push anything down the socket, and an inbound event is + // agent input — so this is a prompt-injection boundary, not a tidiness + // knob. Group access is re-checked in `bus` regardless. + // * NIP-42 auto-authentication is already the upstream default, but + // Buzz relays challenge every connection before granting NIP-29 group + // scopes, so state it here to keep it from silently regressing. + let opts = ClientOptions::new() + .verify_subscriptions(true) + .automatic_authentication(true); + let client = Client::builder() + .signer(bot_keys.clone()) + .opts(opts) + .build(); // Add relays for relay_url in &nostr_config.relays { @@ -156,12 +172,15 @@ impl ChannelPlugin for NostrPlugin { let shared_otp = Arc::new(std::sync::Mutex::new(moltis_channels::otp::OtpState::new( otp_cooldown, ))); + let shared_reply_ctx = + Arc::new(std::sync::Mutex::new(crate::reply_ctx::ReplyContexts::new())); let loop_client = client.clone(); let loop_keys = bot_keys.clone(); let loop_config = Arc::clone(&shared_config); let loop_allowlist = Arc::clone(&cached_allowlist); let loop_otp = Arc::clone(&shared_otp); + let loop_reply_ctx = Arc::clone(&shared_reply_ctx); let loop_account_id = account_id.to_string(); let loop_cancel = cancel.clone(); let loop_sink = Arc::clone(&event_sink); @@ -173,6 +192,7 @@ impl ChannelPlugin for NostrPlugin { loop_config, loop_allowlist, loop_otp, + loop_reply_ctx, loop_account_id, loop_sink, loop_cancel, @@ -188,6 +208,7 @@ impl ChannelPlugin for NostrPlugin { cached_allowlist, cancel, otp: shared_otp, + reply_ctx: shared_reply_ctx, }; self.accounts @@ -395,6 +416,7 @@ mod tests { cached_allowlist, cancel, otp, + reply_ctx: Arc::new(std::sync::Mutex::new(crate::reply_ctx::ReplyContexts::new())), }; let plugin = NostrPlugin::new(); diff --git a/crates/nostr/src/reply_ctx.rs b/crates/nostr/src/reply_ctx.rs new file mode 100644 index 0000000000..000157d641 --- /dev/null +++ b/crates/nostr/src/reply_ctx.rs @@ -0,0 +1,285 @@ +//! Remembers how to answer an inbound group message. +//! +//! Two things about a group message are needed to reply correctly but are not +//! recoverable from the outbound side, where all the gateway hands back is a +//! chat id and the event id being replied to: +//! +//! * **Who wrote it** — so the reply can `p`-tag them, which is what turns a +//! reply into a notification for that person on NIP-29 clients. +//! * **Which message kind it used** — Buzz posts `kind:40002` +//! (`KIND_STREAM_MESSAGE_V2`) while plain NIP-29 relays use `kind:9`, and a +//! client filtering for one does not see the other. Answering in the dialect +//! we were addressed in keeps the bot visible on both without configuration. +//! +//! Entries are bounded and evicted oldest-first; losing one only costs the +//! `p` tag and falls back to the configured default kind. + +use std::collections::HashMap; + +use nostr_sdk::prelude::{EventId, Kind, PublicKey}; + +/// Maximum number of remembered messages. +const DEFAULT_CAPACITY: usize = 10_000; + +/// How to answer a specific inbound group message. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct ReplyContext { + /// Author of the message, `p`-tagged so the reply notifies them. + pub author: PublicKey, + /// Kind the message used, mirrored so the reply lands in the same view. + pub kind: Kind, +} + +/// Bounded store of per-event reply context, plus the last kind seen per group. +pub struct ReplyContexts { + /// event id -> (context, insertion sequence) + entries: HashMap, + /// group id -> most recent message kind observed in that group + group_kinds: HashMap, + /// (reacted-to event, emoji) -> our reaction's own event id. + /// + /// NIP-25 has no "unreact": retracting means publishing a NIP-09 deletion + /// that references the reaction event, so its id has to be kept around + /// between adding 👀 and replacing it with ✅/❌. + reactions: HashMap<(EventId, String), (EventId, u64)>, + capacity: usize, + seq: u64, +} + +impl ReplyContexts { + #[must_use] + pub fn new() -> Self { + Self::with_capacity(DEFAULT_CAPACITY) + } + + #[must_use] + pub fn with_capacity(capacity: usize) -> Self { + Self { + entries: HashMap::new(), + group_kinds: HashMap::new(), + reactions: HashMap::new(), + capacity: capacity.max(1), + seq: 0, + } + } + + /// Record an inbound group message. + pub fn record(&mut self, event_id: EventId, group_id: &str, author: PublicKey, kind: Kind) { + if self.entries.len() >= self.capacity { + self.evict_oldest(); + } + self.seq = self.seq.saturating_add(1); + self.entries + .insert(event_id, (ReplyContext { author, kind }, self.seq)); + self.group_kinds.insert(group_id.to_string(), kind); + } + + /// Look up how to answer a specific message. + #[must_use] + pub fn get(&self, event_id: &EventId) -> Option { + self.entries.get(event_id).map(|(ctx, _)| *ctx) + } + + /// The kind last seen in a group, for sends that are not replies. + #[must_use] + pub fn kind_for_group(&self, group_id: &str) -> Option { + self.group_kinds.get(group_id).copied() + } + + /// Remember the reaction event we published, so it can be retracted later. + /// + /// Bounded like `entries`: the gateway retracts the 👀 it adds on receipt, + /// but the terminal ✅/❌ is never retracted, so without a ceiling this map + /// would grow by one entry per completed group turn for the life of the + /// process — driven entirely by remote traffic. Losing an old id only costs + /// a best-effort retraction that nothing is waiting on. + pub fn record_reaction(&mut self, target: EventId, emoji: &str, reaction: EventId) { + if self.reactions.len() >= self.capacity { + self.evict_oldest_reactions(); + } + self.seq = self.seq.saturating_add(1); + self.reactions + .insert((target, emoji.to_string()), (reaction, self.seq)); + } + + /// Drop the oldest ~10% of remembered reactions (at least one). + fn evict_oldest_reactions(&mut self) { + let to_remove = (self.capacity / 10).max(1); + let mut by_age: Vec<((EventId, String), u64)> = self + .reactions + .iter() + .map(|(key, (_, seq))| (key.clone(), *seq)) + .collect(); + by_age.sort_by_key(|(_, seq)| *seq); + for (key, _) in by_age.into_iter().take(to_remove) { + self.reactions.remove(&key); + } + } + + /// Take back the id of a reaction we published, if we still know it. + pub fn take_reaction(&mut self, target: EventId, emoji: &str) -> Option { + self.reactions + .remove(&(target, emoji.to_string())) + .map(|(reaction, _)| reaction) + } + + /// Drop the oldest ~10% of entries (at least one) to stay under capacity. + fn evict_oldest(&mut self) { + let to_remove = (self.capacity / 10).max(1); + let mut by_age: Vec<(EventId, u64)> = self + .entries + .iter() + .map(|(id, (_, seq))| (*id, *seq)) + .collect(); + by_age.sort_by_key(|(_, seq)| *seq); + for (id, _) in by_age.into_iter().take(to_remove) { + self.entries.remove(&id); + } + } + + /// Number of remembered messages. + #[cfg(test)] + pub fn len(&self) -> usize { + self.entries.len() + } + + /// Whether anything is remembered yet. + #[cfg(test)] + pub fn is_empty(&self) -> bool { + self.entries.is_empty() + } +} + +impl Default for ReplyContexts { + fn default() -> Self { + Self::new() + } +} + +#[cfg(test)] +mod tests { + use nostr_sdk::prelude::Keys; + + use super::*; + + fn event_id() -> EventId { + EventId::from_byte_array(Keys::generate().public_key().to_bytes()) + } + + #[test] + fn records_and_returns_author_and_kind() { + let mut ctxs = ReplyContexts::new(); + let id = event_id(); + let author = Keys::generate().public_key(); + ctxs.record(id, "grp", author, Kind::from_u16(40002)); + + let got = ctxs.get(&id); + assert_eq!(got.map(|c| c.author), Some(author)); + assert_eq!(got.map(|c| c.kind), Some(Kind::from_u16(40002))); + } + + #[test] + fn unknown_event_has_no_context() { + let ctxs = ReplyContexts::new(); + assert!(ctxs.is_empty()); + assert!(ctxs.get(&event_id()).is_none()); + } + + /// A group's dialect is learned from traffic so non-reply sends still land + /// in the right view. + #[test] + fn tracks_last_kind_per_group() { + let mut ctxs = ReplyContexts::new(); + ctxs.record( + event_id(), + "buzz", + Keys::generate().public_key(), + Kind::from_u16(40002), + ); + ctxs.record( + event_id(), + "nip29", + Keys::generate().public_key(), + Kind::from_u16(9), + ); + + assert_eq!(ctxs.kind_for_group("buzz"), Some(Kind::from_u16(40002))); + assert_eq!(ctxs.kind_for_group("nip29"), Some(Kind::from_u16(9))); + assert_eq!(ctxs.kind_for_group("unseen"), None); + } + + #[test] + fn newer_kind_wins_for_group() { + let mut ctxs = ReplyContexts::new(); + let author = Keys::generate().public_key(); + ctxs.record(event_id(), "grp", author, Kind::from_u16(9)); + ctxs.record(event_id(), "grp", author, Kind::from_u16(40002)); + assert_eq!(ctxs.kind_for_group("grp"), Some(Kind::from_u16(40002))); + } + + #[test] + fn reaction_ids_round_trip_for_retraction() { + let mut ctxs = ReplyContexts::new(); + let target = event_id(); + let reaction = event_id(); + ctxs.record_reaction(target, "\u{1F440}", reaction); + + // Taking it returns the id once, then forgets it. + assert_eq!(ctxs.take_reaction(target, "\u{1F440}"), Some(reaction)); + assert_eq!(ctxs.take_reaction(target, "\u{1F440}"), None); + } + + #[test] + fn reactions_are_keyed_by_target_and_emoji() { + let mut ctxs = ReplyContexts::new(); + let target = event_id(); + let eyes = event_id(); + let check = event_id(); + ctxs.record_reaction(target, "\u{1F440}", eyes); + ctxs.record_reaction(target, "\u{2705}", check); + + assert_eq!(ctxs.take_reaction(target, "\u{2705}"), Some(check)); + // Removing one must not disturb the other. + assert_eq!(ctxs.take_reaction(target, "\u{1F440}"), Some(eyes)); + assert_eq!(ctxs.take_reaction(event_id(), "\u{1F440}"), None); + } + + /// The gateway never retracts the terminal ✅/❌, so without a ceiling this + /// map would grow once per completed group turn, forever. + #[test] + fn reactions_are_bounded_too() { + let mut ctxs = ReplyContexts::with_capacity(10); + let author = Keys::generate().public_key(); + for _ in 0..200 { + ctxs.record_reaction(event_id(), "\u{2705}", event_id()); + } + assert!( + ctxs.reactions.len() <= 10, + "reaction ids must not grow without bound, got {}", + ctxs.reactions.len() + ); + let _ = author; + } + + #[test] + fn evicts_to_stay_within_capacity() { + let mut ctxs = ReplyContexts::with_capacity(10); + let author = Keys::generate().public_key(); + for _ in 0..50 { + ctxs.record(event_id(), "grp", author, Kind::from_u16(9)); + } + assert!(ctxs.len() <= 10, "capacity must be respected"); + } + + /// Eviction must not lose the group dialect, which outlives individual + /// messages and is what non-reply sends depend on. + #[test] + fn eviction_preserves_group_kind() { + let mut ctxs = ReplyContexts::with_capacity(4); + let author = Keys::generate().public_key(); + for _ in 0..20 { + ctxs.record(event_id(), "grp", author, Kind::from_u16(40002)); + } + assert_eq!(ctxs.kind_for_group("grp"), Some(Kind::from_u16(40002))); + } +} diff --git a/crates/nostr/src/state.rs b/crates/nostr/src/state.rs index 12d86d14c8..0a4841a9dc 100644 --- a/crates/nostr/src/state.rs +++ b/crates/nostr/src/state.rs @@ -8,7 +8,7 @@ use { tokio_util::sync::CancellationToken, }; -use crate::config::NostrAccountConfig; +use crate::{config::NostrAccountConfig, reply_ctx::ReplyContexts}; /// Shared config reference — the bus loop and plugin both read/write through /// this same `Arc` so runtime config updates (DM policy, allowlist) take @@ -22,6 +22,11 @@ pub type SharedConfig = Arc>; /// Shared OTP state — bus loop initiates challenges, plugin reads pending list. pub type SharedOtp = Arc>; +/// Shared group reply context — the bus loop records inbound group messages, +/// the outbound adapter reads them to mirror the message kind and `p`-tag the +/// author it is replying to. +pub type SharedReplyContexts = Arc>; + /// Runtime state for a single active Nostr account. pub struct AccountState { /// The nostr-sdk client connected to relays. @@ -36,6 +41,8 @@ pub struct AccountState { pub cancel: CancellationToken, /// OTP self-approval state — shared with bus loop. pub otp: SharedOtp, + /// Group reply context — shared with bus loop. + pub reply_ctx: SharedReplyContexts, } impl std::fmt::Debug for AccountState { diff --git a/crates/nostr/tests/buzz_wire_format.rs b/crates/nostr/tests/buzz_wire_format.rs new file mode 100644 index 0000000000..63c3ad5ffa --- /dev/null +++ b/crates/nostr/tests/buzz_wire_format.rs @@ -0,0 +1,534 @@ +//! Wire-format round trips for Buzz group chat, against a real relay. +//! +//! The kinds and tag shapes here were taken from Buzz's own source +//! (`crates/buzz-core/src/kind.rs` and `crates/buzz-sdk/src/builders.rs`). +//! Reading them is not the same as proving we emit them, so these tests publish +//! through the actual `nostr-sdk` client into an in-process relay and assert on +//! what comes back off the wire — kind numbers, tag names, tag ordering and the +//! NIP-10 marker. A mistake in any of those is invisible in unit tests but makes +//! the bot silently malformed in a Buzz channel. +//! +//! No network access is involved: `MockRelay` binds to loopback. + +#![allow(clippy::expect_used)] + +use std::{ + collections::HashMap, + sync::{Arc, Mutex, RwLock}, + time::Duration, +}; + +use { + moltis_channels::{ChannelStreamOutbound, StreamReceiver, otp::OtpState, plugin::StreamEvent}, + moltis_nostr::{ + config::NostrAccountConfig, + groups::{self, GroupMessageKind}, + outbound::{AccountStateMap, NostrOutbound}, + reply_ctx::ReplyContexts, + state::AccountState, + }, + nostr_relay_builder::MockRelay, + nostr_sdk::prelude::{ + Client, Event, EventBuilder, EventId, Filter, Keys, Kind, PublicKey, Tag, TagKind, + }, + tokio::sync::mpsc, + tokio_util::sync::CancellationToken, +}; + +const FETCH_TIMEOUT: Duration = Duration::from_secs(10); + +/// Connect a client to a fresh in-process relay. +async fn connected_client() -> (MockRelay, Client, Keys) { + let relay = MockRelay::run().await.expect("start mock relay"); + let keys = Keys::generate(); + let client = Client::new(keys.clone()); + client + .add_relay(relay.url().await) + .await + .expect("add relay"); + client.connect().await; + (relay, client, keys) +} + +/// Read back every stored event of `kind`. +async fn fetch_kind(client: &Client, kind: Kind) -> Vec { + client + .fetch_events(Filter::new().kind(kind), FETCH_TIMEOUT) + .await + .expect("fetch events") + .into_iter() + .collect() +} + +/// All values of a single-letter tag, in document order. +fn tag_values<'a>(event: &'a Event, kind: TagKind<'a>) -> Vec<&'a [String]> { + event.tags.filter(kind).map(Tag::as_slice).collect() +} + +/// A Buzz reply must land as kind:40002 carrying the channel `h` tag, an `e` +/// tag with the NIP-10 `"reply"` marker, and a `p` tag notifying the author. +#[tokio::test] +async fn buzz_reply_round_trips_with_expected_kind_and_tags() { + let (_relay, client, _keys) = connected_client().await; + + let parent = EventId::all_zeros(); + let author = Keys::generate().public_key(); + let published = groups::send_group_message( + &client, + groups::buzz_stream_message_kind(), + "buzz-general", + "hello from moltis", + Some(parent), + Some(author), + ) + .await + .expect("publish group message"); + + let events = fetch_kind(&client, groups::buzz_stream_message_kind()).await; + let event = events + .iter() + .find(|e| e.id == published) + .expect("published message stored by relay"); + + assert_eq!(event.kind.as_u16(), 40_002, "Buzz stream message v2"); + assert_eq!(event.content, "hello from moltis"); + + // Channel scoping: NIP-29 `h` tag, not an `e` tag. + let h = tag_values(event, TagKind::h()); + assert_eq!(h.len(), 1, "exactly one channel tag"); + assert_eq!(h[0], ["h".to_string(), "buzz-general".to_string()]); + + // Threading: NIP-10 marker, matching Buzz's build_message. + let e = tag_values(event, TagKind::e()); + assert_eq!(e.len(), 1); + assert_eq!( + e[0], + [ + "e".to_string(), + parent.to_hex(), + String::new(), + "reply".to_string() + ], + "reply must carry the NIP-10 \"reply\" marker" + ); + + // Mention: the author being replied to is notified. + let p = tag_values(event, TagKind::p()); + assert_eq!(p.len(), 1); + assert_eq!(p[0][1], author.to_hex()); +} + +/// A standard NIP-29 relay reply must be kind:9, with no Buzz-only kinds. +#[tokio::test] +async fn nip29_reply_round_trips_as_kind_9() { + let (_relay, client, _keys) = connected_client().await; + + let published = groups::send_group_message( + &client, + groups::group_chat_kind(), + "grp", + "plain nip29", + None, + None, + ) + .await + .expect("publish group message"); + + let events = fetch_kind(&client, groups::group_chat_kind()).await; + let event = events + .iter() + .find(|e| e.id == published) + .expect("stored by relay"); + + assert_eq!(event.kind.as_u16(), 9); + // Only the channel tag when there is nothing to thread or notify. + assert_eq!(tag_values(event, TagKind::h()).len(), 1); + assert!(tag_values(event, TagKind::e()).is_empty()); + assert!(tag_values(event, TagKind::p()).is_empty()); +} + +/// Streaming edits publish kind:40003 with `["h", channel]` and `["e", target]`, +/// exactly as Buzz's build_edit does, and carry the replacement text. +#[tokio::test] +async fn edit_round_trips_as_kind_40003() { + let (_relay, client, _keys) = connected_client().await; + + let target = groups::send_group_message( + &client, + groups::buzz_stream_message_kind(), + "buzz-general", + "partial", + None, + None, + ) + .await + .expect("publish original"); + + groups::edit_group_message( + &client, + "buzz-general", + target, + "partial answer, now complete", + ) + .await + .expect("publish edit"); + + let events = fetch_kind(&client, groups::buzz_edit_kind()).await; + let edit = events.first().expect("edit stored by relay"); + + assert_eq!(edit.kind.as_u16(), 40_003); + assert_eq!(edit.content, "partial answer, now complete"); + assert_eq!(tag_values(edit, TagKind::h())[0], [ + "h".to_string(), + "buzz-general".to_string() + ]); + // Buzz's build_edit uses a bare `e` tag here — no NIP-10 marker. + assert_eq!(tag_values(edit, TagKind::e())[0], [ + "e".to_string(), + target.to_hex() + ]); +} + +/// Acknowledgement reactions are NIP-25 kind:7 with the emoji as content, an +/// `e` tag naming the message (as in Buzz's build_reaction) and an `h` tag +/// scoping them to the group, which NIP-29 relays require. +#[tokio::test] +async fn reaction_round_trips_as_nip25_kind_7() { + let (_relay, client, _keys) = connected_client().await; + + let target = groups::send_group_message( + &client, + groups::buzz_stream_message_kind(), + "buzz-general", + "work please", + None, + None, + ) + .await + .expect("publish target"); + + let reaction_id = groups::send_reaction( + &client, + "buzz-general", + target, + groups::ack_emoji_glyph("eyes"), + ) + .await + .expect("publish reaction"); + + let events = fetch_kind(&client, Kind::Reaction).await; + let reaction = events + .iter() + .find(|e| e.id == reaction_id) + .expect("reaction stored by relay"); + + assert_eq!(reaction.kind.as_u16(), 7); + assert_eq!(reaction.content, "👀", "NIP-25 wants the glyph"); + assert_eq!(tag_values(reaction, TagKind::e())[0], [ + "e".to_string(), + target.to_hex() + ]); + // NIP-29 scopes group events by `h`; a strict relay refuses unscoped writes. + assert_eq!(tag_values(reaction, TagKind::h())[0], [ + "h".to_string(), + "buzz-general".to_string() + ]); +} + +/// Retracting an acknowledgement publishes a NIP-09 deletion referencing the +/// reaction event, not the message that was reacted to. +#[tokio::test] +async fn retracting_a_reaction_publishes_nip09_deletion() { + let (_relay, client, _keys) = connected_client().await; + + let target = EventId::all_zeros(); + let reaction_id = groups::send_reaction(&client, "grp", target, "👀") + .await + .expect("publish reaction"); + + groups::delete_event(&client, "grp", reaction_id) + .await + .expect("publish deletion"); + + let events = fetch_kind(&client, Kind::EventDeletion).await; + let deletion = events.first().expect("deletion stored by relay"); + + assert_eq!(deletion.kind.as_u16(), 5); + assert_eq!( + tag_values(deletion, TagKind::e())[0], + ["e".to_string(), reaction_id.to_hex()], + "deletion must reference the reaction, not the reacted-to message" + ); + assert_eq!(tag_values(deletion, TagKind::h())[0], [ + "h".to_string(), + "grp".to_string() + ]); +} + +/// An `h`-tag subscription must deliver both dialects, since a Buzz channel can +/// legitimately carry either. +#[tokio::test] +async fn group_filter_matches_both_dialects() { + let (_relay, client, _keys) = connected_client().await; + + for kind in groups::group_message_kinds() { + groups::send_group_message(&client, kind, "buzz-general", "hi", None, None) + .await + .expect("publish"); + } + // A message in a different channel must not match the filter. + groups::send_group_message( + &client, + groups::buzz_stream_message_kind(), + "other-channel", + "elsewhere", + None, + None, + ) + .await + .expect("publish"); + + let filter = Filter::new() + .kinds(groups::group_message_kinds()) + .custom_tags(groups::h_tag(), ["buzz-general".to_string()]); + let events: Vec = client + .fetch_events(filter, FETCH_TIMEOUT) + .await + .expect("fetch") + .into_iter() + .collect(); + + assert_eq!(events.len(), 2, "both kind:9 and kind:40002 in the channel"); + let mut kinds: Vec = events.iter().map(|e| e.kind.as_u16()).collect(); + kinds.sort_unstable(); + assert_eq!(kinds, vec![9, 40_002]); +} + +/// The `h` tag survives the round trip well enough for the inbound gate to read +/// it back — the value the access check keys on. +#[tokio::test] +async fn published_group_id_is_readable_by_the_inbound_gate() { + let (_relay, client, _keys) = connected_client().await; + + groups::send_group_message( + &client, + groups::buzz_stream_message_kind(), + "buzz-general", + "hi", + None, + None, + ) + .await + .expect("publish"); + + let events = fetch_kind(&client, groups::buzz_stream_message_kind()).await; + let event = events.first().expect("stored"); + + assert_eq!( + groups::extract_group_id(event).as_deref(), + Some("buzz-general") + ); + assert!(groups::is_group_message_kind(event.kind)); +} + +/// A `p`-tagged message is seen as a mention by the responder gate; an +/// untagged one is not. +#[tokio::test] +async fn mention_detection_works_on_round_tripped_events() { + let (_relay, client, _keys) = connected_client().await; + let bot: PublicKey = Keys::generate().public_key(); + + let mentioned = groups::send_group_message( + &client, + groups::buzz_stream_message_kind(), + "buzz-general", + "hey bot", + None, + Some(bot), + ) + .await + .expect("publish mention"); + + let events = fetch_kind(&client, groups::buzz_stream_message_kind()).await; + let event = events + .iter() + .find(|e| e.id == mentioned) + .expect("stored by relay"); + + assert!(groups::is_bot_mentioned(event, &bot)); + assert!(!groups::is_bot_mentioned( + event, + &Keys::generate().public_key() + )); +} + +/// Guard against silently dropping the `h` tag: an event built without it is +/// rejected by the gate even after a relay round trip. +#[tokio::test] +async fn message_without_channel_tag_has_no_group_id() { + let (_relay, client, keys) = connected_client().await; + + let builder = EventBuilder::new(groups::buzz_stream_message_kind(), "oops"); + let event = builder.sign_with_keys(&keys).expect("sign"); + client.send_event(&event).await.expect("publish"); + + let events = fetch_kind(&client, groups::buzz_stream_message_kind()).await; + let stored = events + .iter() + .find(|e| e.id == event.id) + .expect("stored by relay"); + + assert_eq!(groups::extract_group_id(stored), None); +} + +// ── Streaming replies ──────────────────────────────────────────────────── + +/// Build an outbound adapter whose single account talks to `relay`. +async fn outbound_for(relay: &MockRelay, config: NostrAccountConfig) -> NostrOutbound { + let keys = Keys::generate(); + let client = Client::new(keys.clone()); + client + .add_relay(relay.url().await) + .await + .expect("add relay"); + client.connect().await; + + let state = AccountState { + client, + keys, + config: Arc::new(RwLock::new(config)), + cached_allowlist: Arc::new(RwLock::new(Vec::new())), + cancel: CancellationToken::new(), + otp: Arc::new(Mutex::new(OtpState::new(300))), + reply_ctx: Arc::new(Mutex::new(ReplyContexts::new())), + }; + let accounts: AccountStateMap = Arc::new(RwLock::new(HashMap::new())); + accounts + .write() + .expect("lock") + .insert("acct".to_string(), state); + NostrOutbound { accounts } +} + +/// Feed a finished stream of chunks. +fn stream_of(chunks: &[&str]) -> StreamReceiver { + let (tx, rx) = mpsc::channel(16); + for chunk in chunks { + tx.try_send(StreamEvent::Delta((*chunk).to_string())) + .expect("queue chunk"); + } + tx.try_send(StreamEvent::Done).expect("queue done"); + rx +} + +fn group_config(group: &str, kind: GroupMessageKind) -> NostrAccountConfig { + NostrAccountConfig { + groups: vec![group.to_string()], + group_message_kind: kind, + ..Default::default() + } +} + +/// In a Buzz channel the reply is published early and then revised, so the +/// channel watches the answer form. Chunks arrive faster than the throttle, so +/// the intermediate edits collapse into a single final edit carrying the whole +/// text — the message must never be left truncated. +#[tokio::test] +async fn buzz_stream_publishes_then_edits_to_final_text() { + let relay = MockRelay::run().await.expect("relay"); + let outbound = outbound_for( + &relay, + group_config("buzz-general", GroupMessageKind::BuzzV2), + ) + .await; + + outbound + .send_stream( + "acct", + &groups::group_target("buzz-general"), + None, + stream_of(&["Hello", ", ", "world"]), + ) + .await + .expect("stream"); + + let client = { + let accounts = outbound.accounts.read().expect("lock"); + accounts.get("acct").expect("account").client.clone() + }; + + let messages = fetch_kind(&client, groups::buzz_stream_message_kind()).await; + assert_eq!(messages.len(), 1, "exactly one message, then edits"); + assert_eq!(messages[0].content, "Hello", "published on the first chunk"); + + let edits = fetch_kind(&client, groups::buzz_edit_kind()).await; + assert_eq!(edits.len(), 1, "throttled into one final edit"); + assert_eq!( + edits[0].content, "Hello, world", + "final edit must carry the complete answer" + ); + assert_eq!( + tag_values(&edits[0], TagKind::e())[0], + ["e".to_string(), messages[0].id.to_hex()], + "edit must target the published message" + ); +} + +/// A plain NIP-29 relay has no edit kind, so the reply is collected and posted +/// once — publishing a partial message there would strand it truncated. +#[tokio::test] +async fn nip29_stream_sends_one_complete_message_without_edits() { + let relay = MockRelay::run().await.expect("relay"); + let outbound = outbound_for(&relay, group_config("grp", GroupMessageKind::Nip29)).await; + + outbound + .send_stream( + "acct", + &groups::group_target("grp"), + None, + stream_of(&["Hello", ", ", "world"]), + ) + .await + .expect("stream"); + + let client = { + let accounts = outbound.accounts.read().expect("lock"); + accounts.get("acct").expect("account").client.clone() + }; + + let messages = fetch_kind(&client, groups::group_chat_kind()).await; + assert_eq!(messages.len(), 1); + assert_eq!(messages[0].content, "Hello, world", "sent once, complete"); + assert!( + fetch_kind(&client, groups::buzz_edit_kind()) + .await + .is_empty(), + "must not emit Buzz edit kinds on a plain NIP-29 relay" + ); +} + +/// A stream that errors before producing text still says something rather than +/// leaving the channel silent. +#[tokio::test] +async fn stream_error_before_any_text_still_replies() { + let relay = MockRelay::run().await.expect("relay"); + let outbound = outbound_for(&relay, group_config("grp", GroupMessageKind::Nip29)).await; + + let (tx, rx) = mpsc::channel(4); + tx.try_send(StreamEvent::Error("boom".into())) + .expect("queue"); + drop(tx); + + outbound + .send_stream("acct", &groups::group_target("grp"), None, rx) + .await + .expect("stream"); + + let client = { + let accounts = outbound.accounts.read().expect("lock"); + accounts.get("acct").expect("account").client.clone() + }; + let messages = fetch_kind(&client, groups::group_chat_kind()).await; + assert_eq!(messages.len(), 1, "no silent failure"); + assert!(messages[0].content.contains("Error")); +} diff --git a/crates/web/ui/e2e/specs/channels-nostr.spec.js b/crates/web/ui/e2e/specs/channels-nostr.spec.js index ef06aaa3ca..54f2fa454d 100644 --- a/crates/web/ui/e2e/specs/channels-nostr.spec.js +++ b/crates/web/ui/e2e/specs/channels-nostr.spec.js @@ -74,6 +74,46 @@ test.describe("Nostr channel", () => { expect(pageErrors).toEqual([]); }); + test("add modal has Buzz / NIP-29 group fields", async ({ page }) => { + const pageErrors = watchPageErrors(page); + await navigateAndWait(page, "/settings/channels"); + await waitForWsConnected(page); + + await page.getByRole("button", { name: "Connect Nostr", exact: true }).click(); + await expect(page.getByRole("heading", { name: "Connect Nostr", exact: true })).toBeVisible(); + + // Groups input is present and empty by default (DM-only). + const groupsInput = page.locator('input[data-field="groups"]'); + await expect(groupsInput).toBeVisible(); + await expect(groupsInput).toHaveValue(""); + await expect(groupsInput).toHaveAttribute("placeholder", "buzz-general, buzz-dev"); + + // Mention mode select defaults to "mention". + const mentionModeSelect = page.locator('select[data-field="groupMentionMode"]'); + await expect(mentionModeSelect).toBeVisible(); + await expect(mentionModeSelect).toHaveValue("mention"); + + // It offers always / none as well. + await mentionModeSelect.selectOption("always"); + await expect(mentionModeSelect).toHaveValue("always"); + + // Message kind defaults to the interoperable NIP-29 kind:9 and offers Buzz's kind:40002. + const messageKindSelect = page.locator('select[data-field="groupMessageKind"]'); + await expect(messageKindSelect).toBeVisible(); + await expect(messageKindSelect).toHaveValue("nip29"); + await messageKindSelect.selectOption("buzz_v2"); + await expect(messageKindSelect).toHaveValue("buzz_v2"); + + // Ack reactions are on by default and can be turned off. + const ackToggle = page.locator('input[data-field="groupAckReactions"]'); + await expect(ackToggle).toBeVisible(); + await expect(ackToggle).toBeChecked(); + await ackToggle.uncheck(); + await expect(ackToggle).not.toBeChecked(); + + expect(pageErrors).toEqual([]); + }); + test("add modal validates required fields", async ({ page }) => { const pageErrors = watchPageErrors(page); await navigateAndWait(page, "/settings/channels"); diff --git a/crates/web/ui/src/pages/ChannelsPage.tsx b/crates/web/ui/src/pages/ChannelsPage.tsx index 2b244600c6..79c62bf959 100644 --- a/crates/web/ui/src/pages/ChannelsPage.tsx +++ b/crates/web/ui/src/pages/ChannelsPage.tsx @@ -109,6 +109,10 @@ export interface ChannelConfig { secret_key?: string; relays?: string[]; allowed_pubkeys?: string[]; + groups?: string[]; + group_mention_mode?: string; + group_message_kind?: string; + group_ack_reactions?: boolean; // Signal account?: string; account_uuid?: string; diff --git a/crates/web/ui/src/pages/channels/modals/AddNostrModal.tsx b/crates/web/ui/src/pages/channels/modals/AddNostrModal.tsx index 8e0e31dc9f..e8349d43e6 100644 --- a/crates/web/ui/src/pages/channels/modals/AddNostrModal.tsx +++ b/crates/web/ui/src/pages/channels/modals/AddNostrModal.tsx @@ -5,7 +5,7 @@ import type { VNode } from "preact"; import { addChannel, parseChannelConfigPatch } from "../../../channel-utils"; import { models as modelsSig } from "../../../stores/model-store"; -import { targetValue } from "../../../typed-events"; +import { targetChecked, targetValue } from "../../../typed-events"; import { ChannelType } from "../../../types"; import { Modal, ModelSelect } from "../../../ui"; import { type ChannelConfig, ConnectionModeHint, loadChannels, showAddNostr } from "../../ChannelsPage"; @@ -19,6 +19,10 @@ export function AddNostrModal(): VNode { const accountDraft = useSignal(""); const secretKeyDraft = useSignal(""); const relaysDraft = useSignal("wss://relay.damus.io, wss://relay.nostr.band, wss://nos.lol"); + const groupsDraft = useSignal(""); + const groupMentionMode = useSignal("mention"); + const groupMessageKind = useSignal("nip29"); + const groupAckReactions = useSignal(true); const advancedConfigPatch = useSignal(""); function onSubmit(e: Event): void { @@ -51,6 +55,19 @@ export function AddNostrModal(): VNode { dm_policy: (form.querySelector("[data-field=dmPolicy]") as HTMLSelectElement).value, allowed_pubkeys: allowlistItems.value, }; + const groups = groupsDraft.value + .split(",") + .map((g) => g.trim()) + .filter(Boolean); + if (groups.length > 0) { + // NIP-29 group chat (Buzz channels). Left empty keeps DM-only mode. + // `groups` is both the join list and the allowlist, so there is no + // separate group policy to send. + addConfig.groups = groups; + addConfig.group_mention_mode = groupMentionMode.value; + addConfig.group_message_kind = groupMessageKind.value; + addConfig.group_ack_reactions = groupAckReactions.value; + } if (addModel.value) { addConfig.model = addModel.value; const found = modelsSig.value.find((x) => x.id === addModel.value); @@ -67,6 +84,10 @@ export function AddNostrModal(): VNode { accountDraft.value = ""; secretKeyDraft.value = ""; relaysDraft.value = "wss://relay.damus.io, wss://relay.nostr.band, wss://nos.lol"; + groupsDraft.value = ""; + groupMentionMode.value = "mention"; + groupMessageKind.value = "nip29"; + groupAckReactions.value = true; advancedConfigPatch.value = ""; loadChannels(); } else { @@ -159,6 +180,60 @@ export function AddNostrModal(): VNode { allowlistItems.value = v; }} /> + + { + groupsDraft.value = targetValue(e); + }} + className="channel-input" + /> +
+ NIP-29 group ids to join on a Buzz-style relay. Leave empty for DM-only. +
+ + + + +
+ Both kinds are always read and replies match the message they answer. This only sets the kind for messages the + bot starts itself. +
+ { diff --git a/docs/src/nostr.md b/docs/src/nostr.md index 81e0dc1475..da7793cec0 100644 --- a/docs/src/nostr.md +++ b/docs/src/nostr.md @@ -4,7 +4,10 @@ Moltis can receive and send encrypted direct messages over [Nostr](https://nostr.com), the decentralized social protocol. The integration uses [NIP-04](https://github.com/nostr-protocol/nips/blob/master/04.md) encrypted DMs (kind:4) and connects to relays via `nostr-sdk` — no public URL -or server infrastructure is required. +or server infrastructure is required. It can also join +[NIP-29](https://github.com/nostr-protocol/nips/blob/master/29.md) group chats, +including Block's [Buzz](https://github.com/block/buzz) channels — see +[Buzz & NIP-29 Groups](#buzz--nip-29-groups). ## How It Works @@ -85,6 +88,10 @@ offered = ["telegram", "discord", "slack", "matrix", "nostr"] | `model_provider` | no | — | Provider for the overridden model | | `otp_self_approval` | no | `true` | Allow non-allowlisted senders to self-approve via OTP code | | `otp_cooldown_secs` | no | `300` | Cooldown after 3 failed OTP attempts | +| `groups` | no | `[]` | NIP-29 group ids (`h` tags) to join — see [Buzz & NIP-29 Groups](#buzz--nip-29-groups). Empty = DM-only | +| `group_mention_mode` | no | `"mention"` | When to respond in groups: `"mention"` (p-tagged only), `"always"`, or `"none"` | +| `group_message_kind` | no | `"nip29"` | Dialect for bot-initiated group messages: `"nip29"` (kind:9) or `"buzz_v2"` (kind:40002) | +| `group_ack_reactions` | no | `true` | React 👀 on receipt and ✅/❌ on completion in groups (NIP-25) | | `profile.name` | no | — | NIP-01 profile display name | | `profile.display_name` | no | — | NIP-01 longer display name | | `profile.about` | no | — | NIP-01 bio / about text | @@ -132,6 +139,134 @@ When `otp_self_approval` is enabled and a non-allowlisted sender messages the bot, the sender appears in the Senders tab of the web UI where they can be approved or denied. This works the same as OTP for Telegram and Matrix. +## Buzz & NIP-29 Groups + +[Buzz](https://github.com/block/buzz) is Block's open-source team-chat and +agent-collaboration platform built on Nostr — a Slack/GitHub rival where AI +agents and humans are first-class members of shared "channels". Its primary API +is [NIP-29](https://github.com/nostr-protocol/nips/blob/master/29.md) +relay-based group chat over a connection authenticated with +[NIP-42](https://github.com/nostr-protocol/nips/blob/master/42.md). + +Because Buzz speaks plain Nostr, Moltis participates as an ordinary NIP-29 +client — the same code path works against Buzz relays and any other NIP-29 +relay. A Buzz channel is a group identified by an `h` tag; messages are +`kind:9` (standard NIP-29) or `kind:40002` (Buzz) chat events — both are read, +see [Message Kinds](#message-kinds-kind9-vs-kind40002). Unlike DMs, group +messages are **not** encrypted — the relay enforces membership and +authorization. + +### Configuration + +```toml +[channels.nostr.my-bot] +secret_key = "nsec1..." +# Point at the Buzz relay (self-hosted or hosted). NIP-42 auth is automatic. +relays = ["wss://relay.example-buzz.dev"] +# The NIP-29 group ids (h tags) the bot joins. Empty keeps DM-only mode. +groups = ["buzz-general", "buzz-dev"] +group_mention_mode = "mention" # mention | always | none +group_message_kind = "buzz_v2" # nip29 (kind:9) | buzz_v2 (kind:40002) +group_ack_reactions = true # 👀 on receipt, ✅/❌ on completion +``` + +### Live-Streaming Replies (Buzz only) + +Buzz supports editing a published message (`kind:40003`, +`KIND_STREAM_MESSAGE_EDIT`). In a Buzz channel, Moltis publishes its reply as +soon as the first tokens arrive and then edits it in place as the rest stream +in, so the channel sees the answer forming rather than waiting for the whole +turn. Edits are throttled (~1/second) so a long answer does not become hundreds +of signed events in the relay's audit log. + +Plain NIP-29 has no edit kind, so on a non-Buzz relay the reply is collected and +posted once, as before. DMs are unchanged (gift wrap has no edit path). + +### Acknowledgement Reactions + +With `group_ack_reactions` enabled (the default), the bot reacts 👀 to a message +it has picked up, then replaces it with ✅ or ❌ when the turn finishes — the +same acknowledgement pattern Moltis uses on Slack. + +Reactions are NIP-25 `kind:7` events. Retracting the 👀 means publishing a NIP-09 +deletion (`kind:5`) for the reaction, which relays honour at their discretion, +so the 👀 may briefly linger. Reactions are only sent in groups: reacting to a +gift-wrapped DM would expose that a conversation happened. + +### Message Kinds (kind:9 vs kind:40002) + +Buzz defines two chat kinds and posts channel messages as **`kind:40002`** +(`KIND_STREAM_MESSAGE_V2`), while plain NIP-29 relays use **`kind:9`**. Per +Buzz's own architecture notes, a client filtering `kind:9` alone never receives +`kind:40002` messages and vice versa — so speaking only one dialect makes the +bot invisible in the other. + +Moltis therefore **always reads both**, and **replies mirror the kind of the +message they answer**, which makes Buzz channels work without any configuration. +`group_message_kind` only decides bot-*initiated* messages in a group nothing +has been received from yet; once a message arrives, the group's observed dialect +is used. Set it to `buzz_v2` on a Buzz relay so even the first proactive message +lands correctly. + +### How Responses Are Triggered + +`group_mention_mode` controls when the bot replies in a group: + +- **`mention`** (default) — respond only when the bot's pubkey is `p`-tagged + (an `@`-mention). Best for busy channels. +- **`always`** — respond to every message in joined groups. Use for dedicated + agent channels. +- **`none`** — never respond in groups (receive-only). + +Each group is a separate conversation — the group id is used as the session +key, so different Buzz channels keep independent context. A reply carries the +group's `h` tag, a NIP-10 `e` tag threading it to the message being answered, +and a `p` tag notifying that message's author. It is published with the same +kind as the message it answers. + +### Access Control + +`groups` is the whole access model: it is both the set of groups the bot +subscribes to *and* the set it accepts messages from. There is deliberately no +"respond in any group" mode. + +```admonish warning +A group's `h` tag is an unauthenticated label. An event's signature proves who +wrote it, not which group it belongs to, and a relay can push any event down +the socket — `nostr-sdk` does not verify that delivered events match the filters +you subscribed with. Moltis therefore re-checks every inbound group message +(`kind:9` and `kind:40002` alike) against `groups` and drops anything else, so +a hostile or buggy relay cannot inject text into the agent by inventing an `h` +tag. Keep `groups` to channels you actually intend the bot to work in. +``` + +Note that group chat has no sender allowlist: in NIP-29 the relay owns +membership, so anyone the relay admits to a joined channel can address the bot +there. This is the same trust model as a Slack channel, and a reason to prefer +a relay you host. + +The exception is commands that control code execution. `/sh` (which turns +subsequent messages into shell commands) and `/sandbox` (which can move +execution off the sandbox onto the host) are refused in groups unless the +sender's key is in `allowed_pubkeys` — the same list that gates DMs, reused +here as the operator list. An empty `allowed_pubkeys` means nobody may run +them, so they fail closed. The refusal is posted in the channel rather than +ignored. + +### Setup + +1. Generate the bot's key pair (see [Prerequisites](#prerequisites)) and share + its `npub` with the Buzz workspace admin so they can invite/approve it. +2. Set `relays` to the Buzz relay URL and list the channel ids in `groups`. +3. Restart the account. Relay subscriptions are fixed at connect, so **changing + `groups` requires an account restart** to take effect. + +```admonish note +NIP-42 authentication is enabled for every Nostr account, so the bot +transparently authenticates to relays that require it (Buzz relays challenge +every connection before granting group read/write scopes). +``` + ## NIP-04 Encryption All messages between the bot and users are encrypted using