diff --git a/Cargo.lock b/Cargo.lock index 214965503..8804b5546 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2942,6 +2942,7 @@ dependencies = [ "cron", "dashmap", "evalexpr", + "hex", "nostr", "reqwest", "serde", diff --git a/README.md b/README.md index 7c2deb8cd..e001191ef 100644 --- a/README.md +++ b/README.md @@ -42,7 +42,7 @@ append-only and audited. │ Clients │ │ │ │ Human client AI agent Third-party Nostr client │ -│ (any Nostr app) (goose, etc.) (Coracle, nak, Amethyst) │ +│ (Sprout desktop) (goose, etc.) (Coracle, nak, Amethyst) │ │ │ ┌──────────────┐ │ │ │ │ │ sprout-acp │ │ │ │ │ │ (ACP ↔ MCP) │ │ │ diff --git a/crates/sprout-relay/src/handlers/event.rs b/crates/sprout-relay/src/handlers/event.rs index 1b57eba70..3e20a93b0 100644 --- a/crates/sprout-relay/src/handlers/event.rs +++ b/crates/sprout-relay/src/handlers/event.rs @@ -94,7 +94,18 @@ pub(crate) async fn dispatch_persistent_event( } }); - if !is_workflow_execution_kind(kind_u32) { + // Skip workflow triggering for workflow-execution kinds (46001+) and for + // relay-signed workflow-generated messages (sprout:workflow tag). The tag + // check is combined with a pubkey check so that user-submitted events + // carrying the same tag cannot bypass workflow triggers. + let is_relay_workflow_msg = stored_event.event.pubkey == state.relay_keypair.public_key() + && stored_event + .event + .tags + .iter() + .any(|t| t.as_slice().first().map(|s| s.as_str()) == Some("sprout:workflow")); + + if !is_workflow_execution_kind(kind_u32) && !is_relay_workflow_msg { let workflow_engine = Arc::clone(&state.workflow_engine); let workflow_event = stored_event.clone(); tokio::spawn(async move { diff --git a/crates/sprout-relay/src/lib.rs b/crates/sprout-relay/src/lib.rs index 4797963f9..1868ac7d9 100644 --- a/crates/sprout-relay/src/lib.rs +++ b/crates/sprout-relay/src/lib.rs @@ -24,6 +24,8 @@ pub mod state; pub mod subscription; /// Webhook secret generation and constant-time comparison. pub mod webhook_secret; +/// Workflow action sink — relay-side implementation of [`sprout_workflow::ActionSink`]. +pub mod workflow_sink; pub use config::Config; pub use error::{RelayError, Result}; diff --git a/crates/sprout-relay/src/main.rs b/crates/sprout-relay/src/main.rs index b598331c5..37e274650 100644 --- a/crates/sprout-relay/src/main.rs +++ b/crates/sprout-relay/src/main.rs @@ -89,9 +89,6 @@ async fn main() -> anyhow::Result<()> { let workflow_config = sprout_workflow::WorkflowConfig::default(); let workflow_engine = Arc::new(WorkflowEngine::new(db.clone(), workflow_config)); - let wf_cron = Arc::clone(&workflow_engine); - tokio::spawn(async move { wf_cron.run().await }); - let relay_keypair = if let Some(hex) = &config.relay_private_key { nostr::Keys::parse(hex) .map_err(|e| anyhow::anyhow!("invalid SPROUT_RELAY_PRIVATE_KEY: {e}"))? @@ -108,9 +105,19 @@ async fn main() -> anyhow::Result<()> { pubsub, auth, search, - workflow_engine, + Arc::clone(&workflow_engine), relay_keypair, )); + + // Wire the action sink — must happen after AppState (which creates + // sub_registry, conn_manager) and before the cron loop starts. + let action_sink = Arc::new(sprout_relay::workflow_sink::RelayActionSink::new(&state)); + workflow_engine.set_action_sink(action_sink); + + // Start the cron loop AFTER the action sink is wired. + let wf_cron = Arc::clone(&workflow_engine); + tokio::spawn(async move { wf_cron.run().await }); + // Multi-node fan-out consumer: receive events from Redis pub/sub // (published by other relay instances) and fan out to local WS subscribers. { diff --git a/crates/sprout-relay/src/workflow_sink.rs b/crates/sprout-relay/src/workflow_sink.rs new file mode 100644 index 000000000..ac8cc0081 --- /dev/null +++ b/crates/sprout-relay/src/workflow_sink.rs @@ -0,0 +1,152 @@ +//! Relay-side implementation of [`ActionSink`] for workflow actions. +//! +//! Builds Nostr events, persists them, and delegates post-persist side effects +//! (WebSocket fan-out, Redis pub/sub, search indexing, audit logging) to the +//! existing [`dispatch_persistent_event`] helper. + +use std::future::Future; +use std::pin::Pin; +use std::sync::{Arc, Weak}; + +use chrono::Utc; +use nostr::{EventBuilder, Kind, Tag}; +use sprout_core::kind::KIND_STREAM_MESSAGE; +use sprout_workflow::action_sink::{ActionSink, ActionSinkError}; +use tracing::info; +use uuid::Uuid; + +use crate::handlers::event::dispatch_persistent_event; +use crate::state::AppState; + +/// Relay-side action sink — executes workflow side-effects directly. +/// +/// Holds a **weak** reference to `AppState` to avoid an `Arc` reference cycle: +/// `AppState` → `WorkflowEngine` → `ActionSink` → `AppState`. Using `Weak` +/// breaks the cycle so all structs can be dropped on shutdown. +/// +/// Post-persist side effects are delegated to [`dispatch_persistent_event`] +/// for consistency with the REST/WebSocket paths. +pub struct RelayActionSink { + state: Weak, +} + +impl RelayActionSink { + /// Create a new `RelayActionSink` from the shared application state. + pub fn new(state: &Arc) -> Self { + Self { + state: Arc::downgrade(state), + } + } +} + +impl ActionSink for RelayActionSink { + fn send_message( + &self, + channel_id: &str, + text: &str, + author_pubkey: &str, + ) -> Pin> + Send + '_>> { + let channel_id = channel_id.to_owned(); + let text = text.to_owned(); + let author_pubkey = author_pubkey.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. Validate content is not empty/whitespace-only + if text.trim().is_empty() { + return Err(ActionSinkError::EmptyContent); + } + + // 2. Parse and validate channel — canonicalize UUID immediately + let channel_uuid = Uuid::parse_str(&channel_id) + .map_err(|e| ActionSinkError::InvalidInput(format!("invalid UUID: {e}")))?; + let channel_id_canonical = channel_uuid.to_string(); + + let channel = state + .db + .get_channel(channel_uuid) + .await + .map_err(|e| match &e { + sprout_db::DbError::ChannelNotFound(_) | sprout_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 kind:40001 Nostr event + // - Signed by relay keypair (event.pubkey = relay pubkey) + // - `p` tag attributes the message to the workflow owner + // - `h` tag scopes to the channel (NIP-29, canonical UUID) + // - `sprout:workflow` tag prevents recursive workflow triggering + let tags = vec![ + Tag::parse(&["p", &author_pubkey]) + .map_err(|e| ActionSinkError::EventBuild(format!("p tag: {e}")))?, + Tag::parse(&["h", &channel_id_canonical]) + .map_err(|e| ActionSinkError::EventBuild(format!("h tag: {e}")))?, + Tag::parse(&["sprout:workflow", "true"]) + .map_err(|e| ActionSinkError::EventBuild(format!("workflow tag: {e}")))?, + ]; + + let kind = Kind::from(KIND_STREAM_MESSAGE as u16); + let event = EventBuilder::new(kind, &text, tags) + .sign_with_keys(&state.relay_keypair) + .map_err(|e| ActionSinkError::EventBuild(format!("signing: {e}")))?; + + let event_id_hex = event.id.to_hex(); + let event_id_bytes = event.id.as_bytes().to_vec(); + let kind_u32 = KIND_STREAM_MESSAGE; + + let event_created_at = { + let ts = event.created_at.as_u64() as i64; + chrono::DateTime::from_timestamp(ts, 0).unwrap_or_else(Utc::now) + }; + + info!( + event_id = %event_id_hex, + channel_id = %channel_id_canonical, + author = %author_pubkey, + "Workflow SendMessage: posting kind {kind_u32} event" + ); + + // 4. Persist event with thread metadata (matches REST handler path). + // Workflow messages are always top-level: depth=0, no parent/root. + let thread_meta = Some(sprout_db::event::ThreadMetadataParams { + event_id: &event_id_bytes, + event_created_at, + channel_id: channel_uuid, + parent_event_id: None, + parent_event_created_at: None, + root_event_id: None, + root_event_created_at: None, + depth: 0, + broadcast: false, + }); + + let (stored_event, was_inserted) = state + .db + .insert_event_with_thread_metadata(&event, Some(channel_uuid), thread_meta) + .await + .map_err(|e| ActionSinkError::Database(e.to_string()))?; + + // 5. Post-persist side effects (fan-out, search, audit) + // Only if actually inserted (idempotency guard). + if was_inserted { + let _ = dispatch_persistent_event(&state, &stored_event, kind_u32, &author_pubkey) + .await; + } + + Ok(event_id_hex) + }) + } +} diff --git a/crates/sprout-workflow/Cargo.toml b/crates/sprout-workflow/Cargo.toml index 0d9af31ac..7fdf22403 100644 --- a/crates/sprout-workflow/Cargo.toml +++ b/crates/sprout-workflow/Cargo.toml @@ -10,6 +10,7 @@ description = "YAML-as-code workflow engine for Sprout" [dependencies] sprout-core = { workspace = true } sprout-db = { workspace = true } +hex = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } serde_yaml = { workspace = true } diff --git a/crates/sprout-workflow/src/action_sink.rs b/crates/sprout-workflow/src/action_sink.rs new file mode 100644 index 000000000..a940d56e3 --- /dev/null +++ b/crates/sprout-workflow/src/action_sink.rs @@ -0,0 +1,61 @@ +//! Action sink trait — interface for workflow side-effects. +//! +//! The relay implements [`ActionSink`] to provide direct DB access to the +//! executor, replacing the HTTP loopback pattern. + +use std::future::Future; +use std::pin::Pin; + +/// Errors from action sink operations. +#[derive(Debug, thiserror::Error)] +pub enum ActionSinkError { + /// An input parameter is malformed (e.g. invalid UUID). + #[error("invalid input: {0}")] + InvalidInput(String), + /// The target channel does not exist. + #[error("channel not found: {0}")] + ChannelNotFound(String), + /// The target channel is archived. + #[error("channel is archived: {0}")] + ChannelArchived(String), + /// Nostr event construction or signing failed. + #[error("event construction failed: {0}")] + EventBuild(String), + /// A database operation failed. + #[error("database error: {0}")] + Database(String), + /// Message content is empty or whitespace-only. + #[error("empty message content")] + EmptyContent, +} + +impl From for crate::WorkflowError { + fn from(e: ActionSinkError) -> Self { + crate::WorkflowError::WebhookError(e.to_string()) + } +} + +/// Interface for workflow actions that produce side effects. +/// +/// Implemented by the relay to provide direct DB/event access to the executor. +/// This replaces the HTTP loopback where the executor POSTed to the relay's +/// REST API (which failed with 401 auth errors). +/// +/// Returns `Pin>` for dyn-compatibility — required because +/// `WorkflowEngine` stores `Arc`. +pub trait ActionSink: Send + Sync { + /// Post a message to a channel on behalf of a workflow owner. + /// + /// - `channel_id`: UUID string of the target channel + /// - `text`: message body (must not be empty/whitespace-only) + /// - `author_pubkey`: hex-encoded pubkey of the workflow owner (used for + /// the `p` attribution tag; the relay keypair signs the event) + /// + /// Returns the event ID hex string on success. + fn send_message( + &self, + channel_id: &str, + text: &str, + author_pubkey: &str, + ) -> Pin> + Send + '_>>; +} diff --git a/crates/sprout-workflow/src/executor.rs b/crates/sprout-workflow/src/executor.rs index b27f65488..8f31a32d2 100644 --- a/crates/sprout-workflow/src/executor.rs +++ b/crates/sprout-workflow/src/executor.rs @@ -505,7 +505,7 @@ pub enum StepResult { pub async fn dispatch_action( step_id: &str, action: &ActionDef, - _engine: &WorkflowEngine, + engine: &WorkflowEngine, run_id: Uuid, trigger_ctx: &TriggerContext, ) -> Result { @@ -535,23 +535,34 @@ pub async fn dispatch_action( )); } - #[cfg(feature = "reqwest")] - { - let result = send_message_impl(channel_id, text).await?; - Ok(StepResult::Completed(result)) - } - - #[cfg(not(feature = "reqwest"))] - { - warn!( - run_id = %run_id, - step = step_id, - "SendMessage: reqwest feature not enabled, skipping HTTP call" - ); - Ok(StepResult::Completed( - serde_json::json!({ "sent": false, "skipped": true }), + // Look up workflow owner for message attribution. + let wf_run = engine.db.get_workflow_run(run_id).await.map_err(|e| { + WorkflowError::WebhookError(format!( + "SendMessage: failed to load workflow run {run_id}: {e}" )) - } + })?; + let workflow = engine + .db + .get_workflow(wf_run.workflow_id) + .await + .map_err(|e| { + WorkflowError::WebhookError(format!( + "SendMessage: failed to load workflow {}: {e}", + wf_run.workflow_id + )) + })?; + let owner_pubkey_hex = hex::encode(&workflow.owner_pubkey); + + let event_id = engine + .action_sink()? + .send_message(channel_id, text, &owner_pubkey_hex) + .await + .map_err(WorkflowError::from)?; + + Ok(StepResult::Completed(serde_json::json!({ + "sent": true, + "event_id": event_id, + }))) } SendDm { to, text: _ } => { @@ -848,26 +859,8 @@ async fn call_webhook_impl( })) } -// ── send_message implementation (feature-gated) ────────────────────────────── +// ── HTTP helpers for actions that still use the loopback (AddReaction) ──────── -/// POST `{"content": text}` to `POST /api/channels/{channel_id}/messages`. -/// -/// Relay base URL is read from `SPROUT_RELAY_BASE_URL` (default: `http://localhost:3000`). -/// -/// Auth strategy (in priority order): -/// 1. `SPROUT_API_TOKEN` env var → `Authorization: Bearer ` (production) -/// 2. `SPROUT_RELAY_PUBKEY` env var → `X-Pubkey: ` (dev mode only, -/// requires `SPROUT_REQUIRE_AUTH_TOKEN=false`) -/// -/// NOTE: This implementation uses X-Pubkey auth which only works in dev mode -/// (SPROUT_REQUIRE_AUTH_TOKEN=false). For production, the executor needs to -/// either: (a) use an API token (SPROUT_API_TOKEN env var), or (b) sign -/// events directly and submit via WebSocket. See WF-07. -/// TODO(WF-07): Support production auth for workflow-generated messages. -/// -/// This is an internal call — the workflow engine runs inside the relay process, -/// so `localhost:3000` is always reachable without SSRF concerns. -/// /// Returns a shared `reqwest::Client` reused across all workflow HTTP calls. /// Sharing a single client reuses the underlying connection pool. #[cfg(feature = "reqwest")] @@ -883,57 +876,6 @@ fn shared_http_client() -> &'static reqwest::Client { &CLIENT } -#[cfg(feature = "reqwest")] -async fn send_message_impl(channel_id: &str, text: &str) -> Result { - let base_url = std::env::var("SPROUT_RELAY_BASE_URL") - .unwrap_or_else(|_| "http://localhost:3000".to_owned()); - - let url = format!("{base_url}/api/channels/{channel_id}/messages"); - - let client = shared_http_client(); - - let mut req = client - .post(&url) - .header("Content-Type", "application/json") - .json(&serde_json::json!({ "content": text })); - - // Attach auth header: prefer API token (production), fall back to X-Pubkey (dev mode). - if let Ok(token) = std::env::var("SPROUT_API_TOKEN") { - req = req.header("Authorization", format!("Bearer {token}")); - } else if let Ok(pubkey) = std::env::var("SPROUT_RELAY_PUBKEY") { - req = req.header("X-Pubkey", pubkey); - } - - let resp = req - .send() - .await - .map_err(|e| WorkflowError::WebhookError(format!("SendMessage 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!( - "SendMessage: relay returned {status} for channel {channel_id}: {body}" - ))); - } - - let body_text = resp.text().await.unwrap_or_else(|_| String::new()); - - // Try to parse the response as JSON; fall back to wrapping the raw text. - let body_json: JsonValue = serde_json::from_str(&body_text) - .unwrap_or_else(|_| serde_json::json!({ "raw": body_text })); - - Ok(serde_json::json!({ - "sent": true, - "status": status.as_u16(), - "response": body_json, - })) -} - /// POST `{"emoji": emoji}` to `POST /api/messages/{message_id}/reactions`. #[cfg(feature = "reqwest")] async fn add_reaction_impl(message_id: &str, emoji: &str) -> Result { @@ -982,7 +924,6 @@ async fn add_reaction_impl(message_id: &str, emoji: &str) -> Result>, + /// Action sink for executing side-effects (SendMessage, etc.). + /// Late-initialized via [`set_action_sink`] after `AppState` construction. + pub(crate) action_sink: OnceLock>, } impl WorkflowEngine { @@ -90,9 +96,34 @@ impl WorkflowEngine { config, run_semaphore, last_fired: DashMap::new(), + action_sink: OnceLock::new(), } } + /// Set the action sink. Called once after `AppState` construction. + /// + /// # Panics + /// Panics if called more than once. + pub fn set_action_sink(&self, sink: Arc) { + if self.action_sink.set(sink).is_err() { + panic!("action_sink already initialized"); + } + } + + /// Get the action sink reference. + /// + /// Returns `Err(WorkflowError)` if the sink has not been initialized via + /// [`set_action_sink`]. This avoids a panic if the engine is used before + /// wiring is complete. + pub(crate) fn action_sink(&self) -> Result<&dyn ActionSink, WorkflowError> { + self.action_sink.get().map(|s| s.as_ref()).ok_or_else(|| { + WorkflowError::InvalidDefinition( + "action_sink not initialized — call set_action_sink() before executing workflows" + .into(), + ) + }) + } + /// Parse and validate a YAML workflow definition. /// /// Returns `(WorkflowDef, canonical_json)` on success. The canonical JSON diff --git a/crates/sprout-workflow/src/schema.rs b/crates/sprout-workflow/src/schema.rs index 27aafe638..e6b4642ee 100644 --- a/crates/sprout-workflow/src/schema.rs +++ b/crates/sprout-workflow/src/schema.rs @@ -102,7 +102,7 @@ pub enum ActionDef { SendMessage { /// Message text (supports template variables). text: String, - /// Optional channel override (e.g. `"#engineering-oncall"`). + /// Optional channel UUID override. Must be a valid UUID string. #[serde(default)] channel: Option, },