diff --git a/crates/buzz-core/src/tenant.rs b/crates/buzz-core/src/tenant.rs index f7894a69991..2da7a4cefc1 100644 --- a/crates/buzz-core/src/tenant.rs +++ b/crates/buzz-core/src/tenant.rs @@ -109,6 +109,11 @@ impl TenantContext { /// /// Rules (host only — the caller has already split off any path/scheme): /// - ASCII-lowercase (hosts are case-insensitive per RFC 3986); +/// - loopback aliases collapse: `127.0.0.1` and `[::1]` (with or without a +/// port) canonicalize to `localhost` — they name the same machine, and a +/// deployment seeded under one spelling must also answer the others (the +/// classic local failure: community seeded as `localhost:3999`, browser +/// sends `Host: 127.0.0.1:3999`, resolution fails closed with a 404); /// - strip a single trailing dot (the FQDN root label); /// - strip a default port suffix (`:80`, `:443`) — non-default ports are kept, /// since a deployment may legitimately serve different communities on @@ -134,6 +139,20 @@ pub fn normalize_host(host: &str) -> String { if let Some(stripped) = host.strip_suffix('.') { host = stripped.to_string(); } + // Collapse loopback aliases. `127.0.0.1`, `[::1]`, and `localhost` name + // the same machine, so they must resolve to the same tenant — a community + // seeded under one spelling has to answer requests arriving with another. + // Applied after default-port stripping so `[::1]:443` → `[::1]` → + // `localhost` too; a non-default port (`127.0.0.1:3999`) is preserved. + // The alias must be the whole host or end at a port boundary — a domain + // like `127.0.0.1.example` is NOT loopback and stays untouched. + let loopback_port = ["127.0.0.1", "[::1]"].iter().find_map(|alias| { + host.strip_prefix(alias) + .filter(|rest| rest.is_empty() || rest.starts_with(':')) + }); + if let Some(port) = loopback_port { + host = format!("localhost{port}"); + } host } @@ -219,10 +238,13 @@ mod tests { } #[test] - fn normalize_host_leaves_ipv6_literal_intact() { - // IPv6 literals contain colons but no trailing default-port suffix. - assert_eq!(normalize_host("[::1]"), "[::1]"); - assert_eq!(normalize_host("[::1]:443"), "[::1]"); + fn normalize_host_collapses_ipv6_loopback_literal() { + // The IPv6 loopback literal is a loopback alias like 127.0.0.1 and + // collapses to `localhost`; non-loopback IPv6 literals stay intact. + assert_eq!(normalize_host("[::1]"), "localhost"); + assert_eq!(normalize_host("[::1]:3000"), "localhost:3000"); + assert_eq!(normalize_host("[2001:db8::1]"), "[2001:db8::1]"); + assert_eq!(normalize_host("[2001:db8::1]:443"), "[2001:db8::1]"); } #[test] @@ -232,6 +254,23 @@ mod tests { assert_eq!(normalize_host(" "), ""); } + #[test] + fn normalize_host_collapses_loopback_aliases() { + // All loopback spellings name the same machine and must normalize to + // the `localhost` form, so a community seeded as `localhost:3999` + // also answers `Host: 127.0.0.1:3999` (and the IPv6 literal). + assert_eq!(normalize_host("127.0.0.1:3999"), "localhost:3999"); + assert_eq!(normalize_host("127.0.0.1"), "localhost"); + assert_eq!(normalize_host("127.0.0.1:443"), "localhost"); + assert_eq!(normalize_host("[::1]:3999"), "localhost:3999"); + assert_eq!(normalize_host("[::1]"), "localhost"); + assert_eq!(normalize_host("[::1]:443"), "localhost"); + assert_eq!(normalize_host("LOCALHOST:3999"), "localhost:3999"); + // Non-loopback hosts are untouched. + assert_eq!(normalize_host("127.0.0.1.example"), "127.0.0.1.example"); + assert_eq!(normalize_host("example.com"), "example.com"); + } + #[test] fn relay_url_authority_keeps_explicit_nondefault_port() { // The default dev seed: startup, bind_deployment_community, and @@ -259,11 +298,23 @@ mod tests { assert_eq!(relay_url_authority("wss://relay.example"), "relay.example"); } + #[test] + fn relay_url_authority_collapses_loopback_aliases() { + // A relay URL spelled with a loopback IP must derive the same + // authority as the `localhost` spelling — otherwise startup seeds the + // community under one host and requests arriving with the other 404. + assert_eq!(relay_url_authority("ws://127.0.0.1:3999"), "localhost:3999"); + assert_eq!(relay_url_authority("ws://[::1]:3000"), "localhost:3000"); + assert_eq!(relay_url_authority("http://127.0.0.1"), "localhost"); + } + #[test] fn relay_url_authority_preserves_ipv6_brackets() { - // `host_str()` strips IPv6 brackets and the port; `relay_url_authority` - // must keep both so the authority matches `communities.host`. - assert_eq!(relay_url_authority("ws://[::1]:3000"), "[::1]:3000"); + // Non-loopback IPv6 literals keep brackets and port. + assert_eq!( + relay_url_authority("ws://[2001:db8::1]:3000"), + "[2001:db8::1]:3000" + ); } #[test] diff --git a/crates/buzz-relay/src/handlers/event.rs b/crates/buzz-relay/src/handlers/event.rs index ccba40f3282..a5f0fdefe28 100644 --- a/crates/buzz-relay/src/handlers/event.rs +++ b/crates/buzz-relay/src/handlers/event.rs @@ -346,7 +346,7 @@ pub async fn fan_out_pubsub_event(state: &Arc, channel_event: buzz_pub /// audit DB is overloaded; the spawned task still runs the same guarded fan-out /// path, Redis publish, `mark_local_event` echo dedupe, and delivery metrics as /// the former inline path. -pub(crate) async fn dispatch_persistent_event( +pub async fn dispatch_persistent_event( tenant: &TenantContext, state: &Arc, stored_event: &StoredEvent, diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index 566b684f830..5b3edff019b 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -657,6 +657,77 @@ async fn main() -> anyhow::Result<()> { let wf_cron = Arc::clone(&workflow_engine); tokio::spawn(async move { wf_cron.run().await }); + // Relay self-identity profile — publish a kind:0 profile for the relay's + // own pubkey so clients render relay-authored posts (workflow notifications, + // moderation notices) with a recognizable name instead of a raw pubkey. + // Replaceable (NIP-01), so re-publishing at every startup is idempotent and + // lets operators update the name via BUZZ_RELAY_PROFILE_NAME without DB edits. + { + let profile_state = Arc::clone(&state); + tokio::spawn(async move { + let name = std::env::var("BUZZ_RELAY_PROFILE_NAME") + .unwrap_or_else(|_| "buzz-relay".to_string()); + let about = std::env::var("BUZZ_RELAY_PROFILE_ABOUT").unwrap_or_else(|_| { + "Relay infrastructure: workflow notifications, reminders, \ + and moderation notices are authored by this key." + .to_string() + }); + let metadata = serde_json::json!({ + "name": name, + "display_name": name, + "about": about, + }); + let event = match nostr::EventBuilder::new(nostr::Kind::Metadata, metadata.to_string()) + .sign_with_keys(&profile_state.relay_keypair) + { + Ok(e) => e, + Err(e) => { + warn!(error = %e, "relay profile signing failed"); + return; + } + }; + let tenant = match buzz_relay::tenant::bind_deployment_community( + &profile_state.db, + &profile_state.config.relay_url, + ) + .await + { + Ok(ctx) => ctx, + Err(e) => { + warn!( + error = ?e, + "relay profile publish skipped: relay host is not mapped to a community" + ); + return; + } + }; + let relay_pubkey_hex = profile_state.relay_keypair.public_key().to_hex(); + match profile_state + .db + .replace_addressable_event(tenant.community(), &event, None) + .await + { + Ok((stored, was_inserted)) => { + if was_inserted { + let kind_u32 = + buzz_core::kind::event_kind_u32(&stored.event); + buzz_relay::handlers::event::dispatch_persistent_event( + &tenant, + &profile_state, + &stored, + kind_u32, + &relay_pubkey_hex, + None, + ) + .await; + } + info!(name = %name, "relay identity profile published"); + } + Err(e) => warn!(error = %e, "relay profile publish failed"), + } + }); + } + // Ephemeral channel reaper — archives channels whose TTL deadline has passed. // Runs every 60s, matching the workflow cron loop pattern. The SQL UPDATE // uses `archived_at IS NULL` as a guard, so concurrent runs from multiple diff --git a/crates/buzz-relay/src/tenant.rs b/crates/buzz-relay/src/tenant.rs index 88b75f7d6ee..aef0977e725 100644 --- a/crates/buzz-relay/src/tenant.rs +++ b/crates/buzz-relay/src/tenant.rs @@ -236,8 +236,13 @@ mod tests { #[test] fn relay_url_authority_preserves_ipv6_brackets() { - assert_eq!(relay_url_authority("ws://[::1]:3000"), "[::1]:3000"); - assert_eq!(relay_url_authority("wss://[::1]:443"), "[::1]"); + // Non-loopback IPv6 literals keep brackets and port; the loopback + // literal collapses to `localhost` like 127.0.0.1 does. + assert_eq!( + relay_url_authority("ws://[2001:db8::1]:3000"), + "[2001:db8::1]:3000" + ); + assert_eq!(relay_url_authority("ws://[::1]:3000"), "localhost:3000"); } #[tokio::test] diff --git a/crates/buzz-relay/src/workflow_sink.rs b/crates/buzz-relay/src/workflow_sink.rs index 8ce23a2e8ea..7477a885964 100644 --- a/crates/buzz-relay/src/workflow_sink.rs +++ b/crates/buzz-relay/src/workflow_sink.rs @@ -429,6 +429,127 @@ impl ActionSink for RelayActionSink { Ok(event_id_hex) }) } + + fn add_reaction( + &self, + community_id: CommunityId, + target_event_id: &str, + emoji: &str, + ) -> Pin> + Send + '_>> { + let target_event_id = target_event_id.to_owned(); + let emoji = emoji.to_owned(); + + Box::pin(async move { + // 0. Upgrade weak reference — fails only during shutdown. + let state = self + .state + .upgrade() + .ok_or_else(|| ActionSinkError::Database("relay is shutting down".into()))?; + + // 1. Parse the target event ID. + let target_eid = nostr::EventId::parse(&target_event_id) + .map_err(|e| ActionSinkError::InvalidInput(format!("invalid event id: {e}")))?; + let target_bytes = target_eid.as_bytes().to_vec(); + + // 2. Resolve the target's channel (same derivation as the ingest + // path's `derive_reaction_channel`): the target event's own + // channel_id, failing closed when the target is unknown. + let stored_target = state + .db + .get_event_by_id(community_id, &target_bytes) + .await + .map_err(|e| ActionSinkError::Database(e.to_string()))? + .ok_or_else(|| { + ActionSinkError::InvalidInput(format!( + "reaction target event not found: {target_event_id}" + )) + })?; + let channel_uuid = stored_target.channel_id.ok_or_else(|| { + ActionSinkError::InvalidInput(format!( + "reaction target {target_event_id} is not scoped to a channel" + )) + })?; + let channel_id_canonical = channel_uuid.to_string(); + + let host = state + .db + .lookup_community_host(community_id) + .await + .map_err(|e| ActionSinkError::Database(e.to_string()))? + .ok_or_else(|| { + ActionSinkError::Database(format!( + "workflow run community {community_id} is not mapped to a host" + )) + })?; + let tenant = buzz_core::tenant::TenantContext::resolved(community_id, host); + + let channel = state + .db + .get_channel(tenant.community(), channel_uuid) + .await + .map_err(|e| match &e { + buzz_db::DbError::ChannelNotFound(_) | buzz_db::DbError::NotFound(_) => { + ActionSinkError::ChannelNotFound(channel_id_canonical.clone()) + } + _ => ActionSinkError::Database(e.to_string()), + })?; + if channel.archived_at.is_some() { + return Err(ActionSinkError::ChannelArchived( + channel_id_canonical.clone(), + )); + } + + // 3. Build the NIP-25 kind:7 reaction — signed by the relay + // keypair, `e` tag on the target, `h` tag scoping the channel, + // `buzz:workflow` tag preventing recursive workflow triggering. + let tags = vec![ + Tag::parse(["e", &target_eid.to_hex()]) + .map_err(|e| ActionSinkError::EventBuild(format!("e tag: {e}")))?, + Tag::parse(["h", &channel_id_canonical]) + .map_err(|e| ActionSinkError::EventBuild(format!("h tag: {e}")))?, + Tag::parse(["buzz:workflow", "true"]) + .map_err(|e| ActionSinkError::EventBuild(format!("workflow tag: {e}")))?, + ]; + let kind = Kind::from(buzz_core::kind::KIND_REACTION as u16); + let event = EventBuilder::new(kind, &emoji) + .tags(tags) + .sign_with_keys(&state.relay_keypair) + .map_err(|e| ActionSinkError::EventBuild(format!("signing: {e}")))?; + + let event_id_hex = event.id.to_hex(); + let kind_u32 = buzz_core::kind::KIND_REACTION; + + info!( + event_id = %event_id_hex, + target = %target_event_id, + channel_id = %channel_id_canonical, + "Workflow AddReaction: posting kind {kind_u32} event" + ); + + // 4. Persist. Reactions carry no thread metadata of their own. + let (stored_event, was_inserted) = state + .db + .insert_event(tenant.community(), &event, Some(channel_uuid)) + .await + .map_err(|e| ActionSinkError::Database(e.to_string()))?; + + // 5. Post-persist side effects (fan-out) — only if actually + // inserted (idempotency guard). + if was_inserted { + let _ = dispatch_persistent_event( + &tenant, + &state, + &stored_event, + kind_u32, + &stored_event.event.pubkey.to_hex(), + None, + ) + .await; + } + + Ok(event_id_hex) + }) + } } #[cfg(test)] diff --git a/crates/buzz-workflow/src/action_sink.rs b/crates/buzz-workflow/src/action_sink.rs index 079c27a913d..7175a225ab8 100644 --- a/crates/buzz-workflow/src/action_sink.rs +++ b/crates/buzz-workflow/src/action_sink.rs @@ -70,4 +70,22 @@ pub trait ActionSink: Send + Sync { author_pubkey: &str, reply_to: Option<&str>, ) -> Pin> + Send + '_>>; + + /// Add a NIP-25 reaction (kind:7) to a target event on behalf of the relay. + /// + /// - `community_id`: the server-resolved community that owns the workflow + /// run driving this side effect. The reaction is published under *this* + /// community. + /// - `target_event_id`: hex-encoded 32-byte ID of the event to react to. + /// The relay resolves the target's channel itself (same derivation the + /// ingest path uses), so reactions work on any channel-scoped event. + /// - `emoji`: the reaction content (emoji character or `:shortcode:`). + /// + /// Returns the reaction event ID hex string on success. + fn add_reaction( + &self, + community_id: CommunityId, + target_event_id: &str, + emoji: &str, + ) -> Pin> + Send + '_>>; } diff --git a/crates/buzz-workflow/src/executor.rs b/crates/buzz-workflow/src/executor.rs index 5c712dcff7c..77f81aae309 100644 --- a/crates/buzz-workflow/src/executor.rs +++ b/crates/buzz-workflow/src/executor.rs @@ -659,23 +659,15 @@ pub async fn dispatch_action( "AddReaction: no trigger.message_id available".into(), )) } else { - #[cfg(feature = "reqwest")] - { - let result = add_reaction_impl(&trigger_ctx.message_id, emoji).await?; - Ok(StepResult::Completed(result)) - } - - #[cfg(not(feature = "reqwest"))] - { - warn!( - run_id = %run_id, - step = step_id, - "AddReaction: reqwest feature not enabled, skipping HTTP call" - ); - Ok(StepResult::Completed( - serde_json::json!({ "added": false, "skipped": true }), - )) - } + let reaction_event_id = engine + .action_sink()? + .add_reaction(community_id, &trigger_ctx.message_id, emoji) + .await + .map_err(WorkflowError::from)?; + Ok(StepResult::Completed(serde_json::json!({ + "added": true, + "event_id": reaction_event_id, + }))) } } @@ -949,70 +941,6 @@ async fn call_webhook_impl( })) } -/// Returns a shared `reqwest::Client` reused across all workflow HTTP calls. -/// Sharing a single client reuses the underlying connection pool. -#[cfg(feature = "reqwest")] -fn shared_http_client() -> &'static reqwest::Client { - use std::sync::LazyLock; - use std::time::Duration; - static CLIENT: LazyLock = LazyLock::new(|| { - reqwest::Client::builder() - .timeout(Duration::from_secs(10)) - .build() - .expect("HTTP client build must succeed") - }); - &CLIENT -} - -/// POST `{"emoji": emoji}` to `POST /api/messages/{message_id}/reactions`. -#[cfg(feature = "reqwest")] -async fn add_reaction_impl(message_id: &str, emoji: &str) -> Result { - let base_url = - std::env::var("BUZZ_RELAY_BASE_URL").unwrap_or_else(|_| "http://localhost:3000".to_owned()); - - let url = format!("{base_url}/api/messages/{message_id}/reactions"); - - let client = shared_http_client(); - - let mut req = client - .post(&url) - .header("Content-Type", "application/json") - .json(&serde_json::json!({ "emoji": emoji })); - - if let Ok(token) = std::env::var("BUZZ_API_TOKEN") { - req = req.header("Authorization", format!("Bearer {token}")); - } else if let Ok(pubkey) = std::env::var("BUZZ_RELAY_PUBKEY") { - req = req.header("X-Pubkey", pubkey); - } - - let resp = req - .send() - .await - .map_err(|e| WorkflowError::WebhookError(format!("AddReaction HTTP error: {e}")))?; - - let status = resp.status(); - - if !status.is_success() { - let body = resp - .text() - .await - .unwrap_or_else(|_| "".to_owned()); - return Err(WorkflowError::WebhookError(format!( - "AddReaction: relay returned {status} for message {message_id}: {body}" - ))); - } - - let body_text = resp.text().await.unwrap_or_else(|_| String::new()); - let body_json: JsonValue = serde_json::from_str(&body_text) - .unwrap_or_else(|_| serde_json::json!({ "raw": body_text })); - - Ok(serde_json::json!({ - "added": true, - "status": status.as_u16(), - "response": body_json, - })) -} - /// Rich return type from `execute_run` / `execute_from_step`. /// /// Carries enough information for the caller to: diff --git a/crates/buzz-workflow/src/lib.rs b/crates/buzz-workflow/src/lib.rs index bceb6d8bd8d..c4635f2ad11 100644 --- a/crates/buzz-workflow/src/lib.rs +++ b/crates/buzz-workflow/src/lib.rs @@ -357,7 +357,49 @@ impl WorkflowEngine { return Ok(()); } - let trigger_ctx = build_trigger_context(event); + let mut trigger_ctx = build_trigger_context(event); + + // Reaction triggers: `{{ trigger.text }}` / `{{ trigger.author }}` + // must describe the *reacted-to message*, not the reaction event + // (whose content is just the emoji). Resolve the target via the + // reaction's `e` tag and overwrite the context fields; on lookup + // failure keep the raw reaction fields so the run still fires. + if kind_u32 == KIND_REACTION && !trigger_ctx.message_id.is_empty() { + let target_hex = trigger_ctx.message_id.clone(); + if let Ok(target_bytes) = hex::decode(&target_hex) { + match self.db.get_event_by_id(community_id, &target_bytes).await { + Ok(Some(target)) => { + let target_kind = event_kind_u32(&target.event); + // Only unwrap one level: reacting to a message yields + // the message; reacting to a reaction keeps the emoji. + if target_kind != KIND_REACTION { + trigger_ctx.text = target.event.content.clone(); + trigger_ctx.author = target.event.pubkey.to_hex(); + trigger_ctx.timestamp = + target.event.created_at.as_secs().to_string(); + if trigger_ctx.channel_id.is_empty() { + trigger_ctx.channel_id = target + .channel_id + .map(|id| id.to_string()) + .unwrap_or_default(); + } + } + } + Ok(None) => { + tracing::warn!( + target = %target_hex, + "Reaction trigger: target event not found; keeping reaction fields" + ); + } + Err(e) => { + tracing::warn!( + target = %target_hex, + "Reaction trigger: target lookup failed ({e}); keeping reaction fields" + ); + } + } + } + } let trigger_ctx_json: serde_json::Value = match serde_json::to_value(&trigger_ctx) { Ok(v) => v,