diff --git a/crates/buzz-relay/src/workflow_sink.rs b/crates/buzz-relay/src/workflow_sink.rs index 97c31c2561..9379a59065 100644 --- a/crates/buzz-relay/src/workflow_sink.rs +++ b/crates/buzz-relay/src/workflow_sink.rs @@ -8,9 +8,9 @@ use std::future::Future; use std::pin::Pin; use std::sync::{Arc, Weak}; -use buzz_core::kind::KIND_STREAM_MESSAGE; -use buzz_core::tenant::CommunityId; -use buzz_workflow::action_sink::{ActionSink, ActionSinkError}; +use buzz_core::kind::{KIND_REACTION, KIND_STREAM_MESSAGE}; +use buzz_core::tenant::{CommunityId, TenantContext}; +use buzz_workflow::action_sink::{ActionSink, ActionSinkError, AddReactionOutcome}; use chrono::Utc; use nostr::{EventBuilder, Kind, Tag}; use tracing::info; @@ -169,6 +169,23 @@ impl RelayActionSink { } } +async fn resolve_workflow_tenant( + state: &AppState, + community_id: CommunityId, +) -> Result { + 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" + )) + })?; + Ok(TenantContext::resolved(community_id, host)) +} + impl ActionSink for RelayActionSink { fn send_message( &self, @@ -196,17 +213,7 @@ impl ActionSink for RelayActionSink { // form a complete TenantContext (host is for labelling only — the // community is already fixed and is never re-derived from it). Fail // closed if the community no longer maps to a host. - 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 tenant = resolve_workflow_tenant(&state, community_id).await?; // 1. Validate content is not empty/whitespace-only if text.trim().is_empty() { @@ -362,6 +369,157 @@ impl ActionSink for RelayActionSink { Ok(event_id_hex) }) } + + fn add_reaction( + &self, + community_id: CommunityId, + message_id: &str, + emoji: &str, + author_pubkey: &str, + ) -> Pin> + Send + '_>> + { + let message_id = message_id.to_owned(); + let emoji = emoji.to_owned(); + let author_pubkey = author_pubkey.to_owned(); + + Box::pin(async move { + let state = self + .state + .upgrade() + .ok_or_else(|| ActionSinkError::Database("relay is shutting down".into()))?; + let tenant = resolve_workflow_tenant(&state, community_id).await?; + + let target_event_id = nostr::EventId::from_hex(&message_id).map_err(|e| { + ActionSinkError::InvalidInput(format!("invalid target event ID: {e}")) + })?; + let target_event_id_bytes = target_event_id.as_bytes().to_vec(); + let target_event = state + .db + .get_event_by_id(tenant.community(), &target_event_id_bytes) + .await + .map_err(|e| ActionSinkError::Database(e.to_string()))? + .ok_or_else(|| { + ActionSinkError::InvalidInput(format!( + "reaction target event not found: {message_id}" + )) + })?; + let channel_uuid = target_event.channel_id.ok_or_else(|| { + ActionSinkError::InvalidInput( + "reaction target event is not scoped to a channel".into(), + ) + })?; + + 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_uuid.to_string()) + } + _ => ActionSinkError::Database(e.to_string()), + })?; + if channel.archived_at.is_some() { + return Err(ActionSinkError::ChannelArchived(channel_uuid.to_string())); + } + + let author_pubkey = nostr::PublicKey::from_hex(&author_pubkey).map_err(|e| { + ActionSinkError::InvalidInput(format!("invalid author pubkey: {e}")) + })?; + let author_pubkey_bytes = author_pubkey.to_bytes().to_vec(); + let author_pubkey_hex = author_pubkey.to_hex(); + let is_member = state + .is_member_cached(tenant.community(), channel_uuid, &author_pubkey_bytes) + .await + .map_err(|e| ActionSinkError::Database(e.to_string()))?; + if !is_member && channel.visibility != "open" { + return Err(ActionSinkError::InvalidInput( + "workflow owner does not have access to reaction target channel".into(), + )); + } + + const MAX_REACTION_EMOJI_CHARS: usize = 64; + let emoji_char_count = emoji.chars().count(); + if emoji_char_count > MAX_REACTION_EMOJI_CHARS { + return Err(ActionSinkError::InvalidInput(format!( + "reaction emoji exceeds {MAX_REACTION_EMOJI_CHARS} characters (got {emoji_char_count})" + ))); + } + let normalized_emoji = if emoji.is_empty() { "+" } else { &emoji }; + let nonce = Uuid::new_v4().to_string(); + + // The relay signs workflow events, while the actor tag preserves the + // workflow owner as the effective reaction author for deduplication, + // deletion, and reaction summaries. A nonce keeps a rapid + // delete-then-readd from reusing the same second-granularity event + // ID and colliding with the soft-deleted kind:7 row. + let event = EventBuilder::new(Kind::from(KIND_REACTION as u16), &emoji) + .tags([ + Tag::parse(["e", &target_event_id.to_hex()]) + .map_err(|e| ActionSinkError::EventBuild(format!("e tag: {e}")))?, + Tag::parse(["actor", &author_pubkey_hex]) + .map_err(|e| ActionSinkError::EventBuild(format!("actor tag: {e}")))?, + Tag::parse(["buzz:workflow", "true"]) + .map_err(|e| ActionSinkError::EventBuild(format!("workflow tag: {e}")))?, + Tag::parse(["nonce", &nonce]) + .map_err(|e| ActionSinkError::EventBuild(format!("nonce tag: {e}")))?, + ]) + .sign_with_keys(&state.relay_keypair) + .map_err(|e| ActionSinkError::EventBuild(format!("signing: {e}")))?; + let event_id_hex = event.id.to_hex(); + + let outcome = state + .db + .insert_reaction_event_with_thread_metadata( + tenant.community(), + &event, + Some(channel_uuid), + None, + &target_event_id_bytes, + &author_pubkey_bytes, + normalized_emoji, + ) + .await + .map_err(|e| ActionSinkError::Database(e.to_string()))?; + + match outcome { + buzz_db::ReactionEventInsertOutcome::TargetMissing => { + Err(ActionSinkError::InvalidInput(format!( + "reaction target event not found: {message_id}" + ))) + } + buzz_db::ReactionEventInsertOutcome::Duplicate => { + Ok(AddReactionOutcome::AlreadyPresent) + } + buzz_db::ReactionEventInsertOutcome::Inserted { + stored_event, + was_inserted, + } => { + info!( + event_id = %event_id_hex, + target_event_id = %message_id, + channel_id = %channel_uuid, + author = %author_pubkey_hex, + "Workflow AddReaction: posting kind {KIND_REACTION} event" + ); + if was_inserted { + let _ = dispatch_persistent_event( + &tenant, + &state, + &stored_event, + KIND_REACTION, + &author_pubkey_hex, + None, + ) + .await; + } + Ok(AddReactionOutcome::Added { + event_id: event_id_hex, + }) + } + } + }) + } } #[cfg(test)] @@ -708,4 +866,163 @@ mod integration_tests { "mentioned member {agent_hex} must be p-tagged so it wakes; got {p_tag_targets:?}" ); } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn workflow_add_reaction_persists_attributed_kind_7_and_dedupes() { + let state = test_state().await; + + let owner = nostr::Keys::generate(); + let owner_hex = owner.public_key().to_hex(); + let owner_bytes = owner.public_key().to_bytes().to_vec(); + let host = format!("wf-reaction-{}.example", uuid::Uuid::new_v4().simple()); + let community = match state + .db + .create_community_with_owner(&host, &owner_hex) + .await + .expect("create community") + { + CreateCommunityWithOwnerResult::Created(rec) => rec.id, + other => panic!("expected fresh community, got {other:?}"), + }; + let channel = state + .db + .create_channel( + community, + "wf-reaction", + ChannelType::Stream, + ChannelVisibility::Open, + None, + &owner_bytes, + None, + ) + .await + .expect("create channel"); + + let target = EventBuilder::new(Kind::from(KIND_STREAM_MESSAGE as u16), "target") + .tags([Tag::parse(["h", &channel.id.to_string()]).expect("h tag")]) + .sign_with_keys(&owner) + .expect("sign target"); + let target_created_at = + chrono::DateTime::from_timestamp(target.created_at.as_secs() as i64, 0) + .expect("valid target timestamp"); + state + .db + .insert_event_with_thread_metadata(community, &target, Some(channel.id), None) + .await + .expect("persist target"); + + let sink = RelayActionSink::new(&state); + let outcome = sink + .add_reaction(community, &target.id.to_hex(), "👀", &owner_hex) + .await + .expect("add reaction"); + let AddReactionOutcome::Added { event_id } = outcome else { + panic!("expected inserted reaction, got {outcome:?}"); + }; + + let reaction_id = nostr::EventId::from_hex(&event_id).expect("reaction event id"); + let stored = state + .db + .get_event_by_id(community, reaction_id.as_bytes()) + .await + .expect("query reaction") + .expect("reaction persisted"); + assert_eq!(u32::from(stored.event.kind.as_u16()), KIND_REACTION); + assert_eq!(stored.event.content, "👀"); + assert_eq!(stored.event.pubkey, state.relay_keypair.public_key()); + assert!(stored + .event + .tags + .iter() + .any(|tag| { tag.as_slice() == ["e".to_string(), target.id.to_hex()] })); + assert!(stored + .event + .tags + .iter() + .any(|tag| { tag.as_slice() == ["actor".to_string(), owner_hex.clone()] })); + assert!(stored + .event + .tags + .iter() + .any(|tag| { tag.as_slice() == ["buzz:workflow".to_string(), "true".to_string()] })); + assert!(stored + .event + .tags + .iter() + .any(|tag| tag.as_slice().first().map(String::as_str) == Some("nonce"))); + + let active_reaction = state + .db + .get_active_reaction_record( + community, + target.id.as_bytes(), + target_created_at, + &owner_bytes, + "👀", + ) + .await + .expect("query reaction row") + .expect("reaction row exists"); + assert_eq!( + active_reaction.reaction_event_id.as_deref(), + Some(reaction_id.as_bytes().as_slice()) + ); + + let duplicate = sink + .add_reaction(community, &target.id.to_hex(), "👀", &owner_hex) + .await + .expect("repeat reaction"); + assert_eq!(duplicate, AddReactionOutcome::AlreadyPresent); + + assert!(state + .db + .soft_delete_event(community, reaction_id.as_bytes()) + .await + .expect("soft-delete reaction event")); + assert!(state + .db + .remove_reaction_by_source_event_id(community, reaction_id.as_bytes()) + .await + .expect("remove reaction row")); + + let readded = sink + .add_reaction(community, &target.id.to_hex(), "👀", &owner_hex) + .await + .expect("re-add reaction"); + let AddReactionOutcome::Added { + event_id: readded_id, + } = readded + else { + panic!("expected re-added reaction, got {readded:?}"); + }; + assert_ne!( + readded_id, event_id, + "delete-then-readd must publish a fresh kind:7 event" + ); + + let readded_id = nostr::EventId::from_hex(&readded_id).expect("re-added reaction event id"); + assert!(state + .db + .get_event_by_id(community, readded_id.as_bytes()) + .await + .expect("query re-added event") + .is_some()); + let readded_reaction = state + .db + .get_active_reaction_record( + community, + target.id.as_bytes(), + target_created_at, + &owner_bytes, + "👀", + ) + .await + .expect("query re-added reaction row") + .expect("re-added reaction row exists"); + assert_eq!( + readded_reaction.reaction_event_id.as_deref(), + Some(readded_id.as_bytes().as_slice()) + ); + } } diff --git a/crates/buzz-test-client/tests/e2e_workflow_actions.rs b/crates/buzz-test-client/tests/e2e_workflow_actions.rs new file mode 100644 index 0000000000..a300db6b10 --- /dev/null +++ b/crates/buzz-test-client/tests/e2e_workflow_actions.rs @@ -0,0 +1,183 @@ +//! End-to-end regression coverage for workflow action side effects. +//! +//! These tests require a running relay and are ignored by default. +//! +//! ```text +//! cargo test -p buzz-test-client --test e2e_workflow_actions -- --ignored +//! ``` + +use std::time::Duration; + +use buzz_core::kind::{KIND_NIP29_CREATE_GROUP, KIND_REACTION, KIND_STREAM_MESSAGE}; +use buzz_sdk::build_workflow_def; +use buzz_test_client::{BuzzTestClient, RelayMessage}; +use nostr::{Alphabet, Event, EventBuilder, Filter, Keys, Kind, SingleLetterTag, Tag}; +use uuid::Uuid; + +fn relay_url() -> String { + std::env::var("RELAY_URL").unwrap_or_else(|_| "ws://localhost:3000".to_string()) +} + +fn workflow_yaml(name: &str, follow_up: &str) -> String { + format!( + "name: {name}\n\ + trigger:\n\ + \x20 on: message_posted\n\ + steps:\n\ + \x20 - id: acknowledge\n\ + \x20 action: add_reaction\n\ + \x20 emoji: \"👀\"\n\ + \x20 - id: follow_up\n\ + \x20 action: send_message\n\ + \x20 text: \"{follow_up}\"\n" + ) +} + +fn has_tag(event: &Event, name: &str, value: &str) -> bool { + event.tags.iter().any(|tag| { + tag.as_slice().first().map(String::as_str) == Some(name) + && tag.as_slice().get(1).map(String::as_str) == Some(value) + }) +} + +/// Regression for block/buzz#2395: `add_reaction` must emit kind:7 through the +/// in-process action sink, and a successful first step must not abort the next +/// workflow step. +#[tokio::test] +#[ignore = "requires a running relay"] +async fn workflow_add_reaction_emits_kind_7_and_continues() { + let keys = Keys::generate(); + let mut client = BuzzTestClient::connect(&relay_url(), &keys) + .await + .expect("connect"); + let channel_id = Uuid::new_v4(); + let workflow_id = Uuid::new_v4(); + let follow_up = format!("reaction-follow-up-{workflow_id}"); + + let create_channel = EventBuilder::new(Kind::Custom(KIND_NIP29_CREATE_GROUP as u16), "") + .tags([ + Tag::parse(["h", channel_id.to_string().as_str()]).expect("h tag"), + Tag::parse(["name", format!("workflow-reaction-{channel_id}").as_str()]) + .expect("name tag"), + Tag::parse(["channel_type", "stream"]).expect("channel type tag"), + Tag::parse(["visibility", "open"]).expect("visibility tag"), + ]) + .sign_with_keys(&keys) + .expect("sign channel creation"); + let channel_ok = client + .send_event(create_channel) + .await + .expect("send channel creation"); + assert!( + channel_ok.accepted, + "channel creation rejected: {}", + channel_ok.message + ); + + let definition = build_workflow_def( + channel_id, + workflow_id, + &workflow_yaml("reaction-action", &follow_up), + ) + .expect("build workflow definition") + .sign_with_keys(&keys) + .expect("sign workflow definition"); + let definition_ok = client + .send_event(definition) + .await + .expect("send workflow definition"); + assert!( + definition_ok.accepted, + "workflow definition rejected: {}", + definition_ok.message + ); + + let target = EventBuilder::new( + Kind::Custom(KIND_STREAM_MESSAGE as u16), + "trigger reaction workflow", + ) + .tags([Tag::parse(["h", channel_id.to_string().as_str()]).expect("h tag")]) + .sign_with_keys(&keys) + .expect("sign target message"); + + let subscription_id = format!("workflow-reaction-{workflow_id}"); + let reaction_filter = Filter::new() + .kind(Kind::Custom(KIND_REACTION as u16)) + .custom_tag(SingleLetterTag::lowercase(Alphabet::E), target.id.to_hex()) + .custom_tag( + SingleLetterTag::lowercase(Alphabet::H), + channel_id.to_string(), + ); + let message_filter = Filter::new() + .kind(Kind::Custom(KIND_STREAM_MESSAGE as u16)) + .custom_tag( + SingleLetterTag::lowercase(Alphabet::H), + channel_id.to_string(), + ); + client + .subscribe(&subscription_id, vec![reaction_filter, message_filter]) + .await + .expect("subscribe before trigger"); + let initial = client + .collect_until_eose(&subscription_id, Duration::from_secs(5)) + .await + .expect("initial EOSE"); + assert!(initial.is_empty(), "test channel should start empty"); + + let target_ok = client + .send_event(target.clone()) + .await + .expect("send target message"); + assert!( + target_ok.accepted, + "target message rejected: {}", + target_ok.message + ); + + let deadline = tokio::time::Instant::now() + Duration::from_secs(10); + let mut reaction = None; + let mut follow_up_message = None; + while reaction.is_none() || follow_up_message.is_none() { + let remaining = deadline + .checked_duration_since(tokio::time::Instant::now()) + .expect("workflow action events timed out"); + match client + .recv_event(remaining) + .await + .expect("receive workflow action event") + { + RelayMessage::Event { + subscription_id: received_subscription, + event, + } if received_subscription == subscription_id => { + if u32::from(event.kind.as_u16()) == KIND_REACTION { + reaction = Some(*event); + } else if u32::from(event.kind.as_u16()) == KIND_STREAM_MESSAGE + && event.content == follow_up + { + follow_up_message = Some(*event); + } + } + _ => {} + } + } + + let reaction = reaction.expect("reaction event"); + assert_eq!(reaction.content, "👀"); + assert!(has_tag(&reaction, "e", &target.id.to_hex())); + assert!(has_tag(&reaction, "actor", &keys.public_key().to_hex())); + assert!(has_tag(&reaction, "buzz:workflow", "true")); + + let follow_up_message = follow_up_message.expect("follow-up message"); + assert!(has_tag(&follow_up_message, "buzz:workflow", "true")); + assert!( + has_tag(&follow_up_message, "p", &keys.public_key().to_hex()), + "workflow follow-up must preserve owner attribution" + ); + + client + .close_subscription(&subscription_id) + .await + .expect("close subscription"); + client.disconnect().await.expect("disconnect"); +} diff --git a/crates/buzz-workflow/src/action_sink.rs b/crates/buzz-workflow/src/action_sink.rs index 0c6002e74e..163f60391f 100644 --- a/crates/buzz-workflow/src/action_sink.rs +++ b/crates/buzz-workflow/src/action_sink.rs @@ -37,6 +37,18 @@ impl From for crate::WorkflowError { } } +/// Result of adding a workflow reaction. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum AddReactionOutcome { + /// A new kind:7 reaction event was persisted. + Added { + /// Hex-encoded reaction event ID. + event_id: String, + }, + /// The workflow owner already had the same active reaction on the target. + AlreadyPresent, +} + /// Interface for workflow actions that produce side effects. /// /// Implemented by the relay to provide direct DB/event access to the executor. @@ -66,4 +78,19 @@ pub trait ActionSink: Send + Sync { text: &str, author_pubkey: &str, ) -> Pin> + Send + '_>>; + + /// Add a NIP-25 reaction to a message on behalf of a workflow owner. + /// + /// - `community_id`: server-resolved community that owns the workflow run + /// - `message_id`: hex-encoded target event ID + /// - `emoji`: reaction content (up to 64 Unicode characters) + /// - `author_pubkey`: hex-encoded workflow owner pubkey, used as the + /// effective reaction actor while the relay signs the event + fn add_reaction( + &self, + community_id: CommunityId, + message_id: &str, + emoji: &str, + author_pubkey: &str, + ) -> Pin> + Send + '_>>; } diff --git a/crates/buzz-workflow/src/executor.rs b/crates/buzz-workflow/src/executor.rs index a029b44622..d9bd338f4d 100644 --- a/crates/buzz-workflow/src/executor.rs +++ b/crates/buzz-workflow/src/executor.rs @@ -20,7 +20,7 @@ use uuid::Uuid; use crate::error::WorkflowError; use crate::schema::{ActionDef, Step, WorkflowDef}; -use crate::WorkflowEngine; +use crate::{AddReactionOutcome, WorkflowEngine}; /// Data extracted from the triggering event, passed to every step. #[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)] @@ -465,6 +465,33 @@ pub enum StepResult { Skipped, } +async fn load_workflow_for_run( + engine: &WorkflowEngine, + community_id: CommunityId, + run_id: Uuid, + action_name: &str, +) -> Result { + let workflow_run = engine + .db + .get_workflow_run(community_id, run_id) + .await + .map_err(|e| { + WorkflowError::WebhookError(format!( + "{action_name}: failed to load workflow run {run_id}: {e}" + )) + })?; + engine + .db + .get_workflow(community_id, workflow_run.workflow_id) + .await + .map_err(|e| { + WorkflowError::WebhookError(format!( + "{action_name}: failed to load workflow {}: {e}", + workflow_run.workflow_id + )) + }) +} + fn resolve_send_message_channel( explicit_channel: Option<&str>, trigger_channel: &str, @@ -532,25 +559,8 @@ pub async fn dispatch_action( // attribution, scoped to the run's community — the same run/workflow // UUID may exist in another community, so a bare-id lookup could // load the wrong row and drive a side effect under it. - let wf_run = engine - .db - .get_workflow_run(community_id, run_id) - .await - .map_err(|e| { - WorkflowError::WebhookError(format!( - "SendMessage: failed to load workflow run {run_id}: {e}" - )) - })?; - let workflow = engine - .db - .get_workflow(community_id, wf_run.workflow_id) - .await - .map_err(|e| { - WorkflowError::WebhookError(format!( - "SendMessage: failed to load workflow {}: {e}", - wf_run.workflow_id - )) - })?; + let workflow = + load_workflow_for_run(engine, community_id, run_id, "SendMessage").await?; let channel_id = resolve_send_message_channel( channel.as_deref(), &trigger_ctx.channel_id, @@ -597,22 +607,27 @@ pub async fn dispatch_action( )); } - #[cfg(feature = "reqwest")] - { - let result = add_reaction_impl(&trigger_ctx.message_id, emoji).await?; - Ok(StepResult::Completed(result)) - } + let workflow = + load_workflow_for_run(engine, community_id, run_id, "AddReaction").await?; + let owner_pubkey_hex = hex::encode(&workflow.owner_pubkey); + let outcome = engine + .action_sink()? + .add_reaction( + community_id, + &trigger_ctx.message_id, + emoji, + &owner_pubkey_hex, + ) + .await + .map_err(WorkflowError::from)?; - #[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 }), - )) + match outcome { + AddReactionOutcome::Added { event_id } => Ok(StepResult::Completed( + serde_json::json!({ "added": true, "event_id": event_id }), + )), + AddReactionOutcome::AlreadyPresent => Ok(StepResult::Completed( + serde_json::json!({ "added": false, "duplicate": true }), + )), } } @@ -865,70 +880,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 93581225ee..eaea10884a 100644 --- a/crates/buzz-workflow/src/lib.rs +++ b/crates/buzz-workflow/src/lib.rs @@ -35,7 +35,7 @@ pub mod error; pub mod executor; pub mod schema; -pub use action_sink::{ActionSink, ActionSinkError}; +pub use action_sink::{ActionSink, ActionSinkError, AddReactionOutcome}; pub use error::{PartialProgress, WorkflowError}; pub use executor::ExecutionResult; pub use schema::{ActionDef, Step, TriggerDef, WorkflowDef};