From 9d156411fc3d3d038d972760dc7e3780a510397b Mon Sep 17 00:00:00 2001 From: Illia Polosukhin Date: Fri, 6 Feb 2026 11:12:17 -0800 Subject: [PATCH] Unify webhook servers into single WebhookServer Replace the dual-server architecture (HttpChannel + WasmChannelServer both competing for port 8080) with a single WebhookServer that composes route fragments from all sources. Channels define routes but never spawn servers. - Add WebhookServer struct that collects Router fragments and binds one listener - Extract routes() from HttpChannel, remove server-spawning from start/shutdown - Delete WasmChannelServer (keep WasmChannelRouter and route builder) - Rewire main.rs to compose all webhook routes into one server Co-Authored-By: Claude Opus 4.6 --- CLAUDE.md | 4 ++ src/channels/http.rs | 82 ++++++++------------------ src/channels/mod.rs | 2 + src/channels/wasm/mod.rs | 4 +- src/channels/wasm/router.rs | 51 ---------------- src/channels/webhook_server.rs | 92 +++++++++++++++++++++++++++++ src/main.rs | 104 ++++++++++++++++----------------- 7 files changed, 174 insertions(+), 165 deletions(-) create mode 100644 src/channels/webhook_server.rs diff --git a/CLAUDE.md b/CLAUDE.md index e55bcc24..c26c7a14 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -151,6 +151,10 @@ src/ ## Key Patterns +### Architecture + +When designing new features or systems, always prefer generic/extensible architectures over hardcoding specific integrations. Ask clarifying questions about the desired abstraction level before implementing. + ### Error Handling - Use `thiserror` for error types in `error.rs` - Never use `.unwrap()` in production code (tests are fine) diff --git a/src/channels/http.rs b/src/channels/http.rs index 986ebf7c..77576a46 100644 --- a/src/channels/http.rs +++ b/src/channels/http.rs @@ -1,6 +1,5 @@ //! HTTP webhook channel for receiving messages via HTTP POST. -use std::net::SocketAddr; use std::sync::Arc; use async_trait::async_trait; @@ -32,8 +31,6 @@ struct HttpChannelState { tx: RwLock>>, /// Pending responses keyed by message ID. pending_responses: RwLock>>, - /// Server shutdown signal. - shutdown_tx: RwLock>>, /// Expected webhook secret for authentication (if configured). webhook_secret: Option, /// Fixed user ID for this HTTP channel. @@ -74,7 +71,6 @@ impl HttpChannel { state: Arc::new(HttpChannelState { tx: RwLock::new(None), pending_responses: RwLock::new(std::collections::HashMap::new()), - shutdown_tx: RwLock::new(None), webhook_secret, user_id, rate_limit: tokio::sync::Mutex::new(RateLimitState { @@ -84,6 +80,24 @@ impl HttpChannel { }), } } + + /// Return the channel's axum routes with state applied. + /// + /// The returned `Router` shares the same `Arc` that + /// `start()` later populates. Before `start()` is called the webhook + /// handler returns 503 ("Channel not started"). + pub fn routes(&self) -> Router { + Router::new() + .route("/health", get(health_handler)) + .route("/webhook", post(webhook_handler)) + .layer(DefaultBodyLimit::max(MAX_BODY_BYTES)) + .with_state(self.state.clone()) + } + + /// Return the configured host and port for this channel. + pub fn addr(&self) -> (&str, u16) { + (&self.config.host, self.config.port) + } } #[derive(Debug, Deserialize)] @@ -303,53 +317,11 @@ impl Channel for HttpChannel { let (tx, rx) = mpsc::channel(256); *self.state.tx.write().await = Some(tx); - let state = self.state.clone(); - let host = self.config.host.clone(); - let port = self.config.port; - - // Parse address before spawning so we can return errors - let addr: SocketAddr = - format!("{}:{}", host, port) - .parse() - .map_err(|e| ChannelError::StartupFailed { - name: "http".to_string(), - reason: format!("Invalid address '{}:{}': {}", host, port, e), - })?; - - // Bind listener before spawning so we can return errors - let listener = - tokio::net::TcpListener::bind(addr) - .await - .map_err(|e| ChannelError::StartupFailed { - name: "http".to_string(), - reason: format!("Failed to bind to {}: {}", addr, e), - })?; - - tracing::info!("HTTP channel listening on {}", addr); - - // Create router - let app = Router::new() - .route("/health", get(health_handler)) - .route("/webhook", post(webhook_handler)) - .layer(DefaultBodyLimit::max(MAX_BODY_BYTES)) - .with_state(state.clone()); - - // Create shutdown channel - let (shutdown_tx, shutdown_rx) = oneshot::channel(); - *self.state.shutdown_tx.write().await = Some(shutdown_tx); - - // Spawn server (listener is already bound, serve errors are logged) - tokio::spawn(async move { - if let Err(e) = axum::serve(listener, app) - .with_graceful_shutdown(async { - let _ = shutdown_rx.await; - tracing::info!("HTTP channel shutting down"); - }) - .await - { - tracing::error!("HTTP server error: {}", e); - } - }); + tracing::info!( + "HTTP channel ready ({}:{})", + self.config.host, + self.config.port + ); Ok(Box::pin(ReceiverStream::new(rx))) } @@ -363,13 +335,10 @@ impl Channel for HttpChannel { if let Some(tx) = self.state.pending_responses.write().await.remove(&msg.id) { let _ = tx.send(response.content); } - // For async webhooks, we'd need to make an HTTP callback here - // but that requires the caller to provide a callback URL Ok(()) } async fn health_check(&self) -> Result<(), ChannelError> { - // Check if we have an active sender if self.state.tx.read().await.is_some() { Ok(()) } else { @@ -380,11 +349,6 @@ impl Channel for HttpChannel { } async fn shutdown(&self) -> Result<(), ChannelError> { - // Send shutdown signal - if let Some(tx) = self.state.shutdown_tx.write().await.take() { - let _ = tx.send(()); - } - // Clear the message sender *self.state.tx.write().await = None; Ok(()) } diff --git a/src/channels/mod.rs b/src/channels/mod.rs index fc85436e..3796cc20 100644 --- a/src/channels/mod.rs +++ b/src/channels/mod.rs @@ -32,8 +32,10 @@ mod http; mod manager; mod repl; pub mod wasm; +mod webhook_server; pub use channel::{Channel, IncomingMessage, MessageStream, OutgoingResponse, StatusUpdate}; pub use http::HttpChannel; pub use manager::ChannelManager; pub use repl::ReplChannel; +pub use webhook_server::{WebhookServer, WebhookServerConfig}; diff --git a/src/channels/wasm/mod.rs b/src/channels/wasm/mod.rs index 7535d033..e6f4f0d8 100644 --- a/src/channels/wasm/mod.rs +++ b/src/channels/wasm/mod.rs @@ -95,9 +95,7 @@ pub use loader::{ DiscoveredChannel, LoadResults, LoadedChannel, WasmChannelLoader, default_channels_dir, discover_channels, }; -pub use router::{ - RegisteredEndpoint, WasmChannelRouter, WasmChannelServer, create_wasm_channel_router, -}; +pub use router::{RegisteredEndpoint, WasmChannelRouter, create_wasm_channel_router}; pub use runtime::{PreparedChannelModule, WasmChannelRuntime, WasmChannelRuntimeConfig}; pub use schema::{ ChannelCapabilitiesFile, ChannelConfig, SecretSetupSchema, SetupSchema, WebhookSchema, diff --git a/src/channels/wasm/router.rs b/src/channels/wasm/router.rs index 44f81f99..0bd3182f 100644 --- a/src/channels/wasm/router.rs +++ b/src/channels/wasm/router.rs @@ -4,7 +4,6 @@ //! registered paths. Handles secret validation at the host level. use std::collections::HashMap; -use std::net::SocketAddr; use std::sync::Arc; use axum::{ @@ -469,56 +468,6 @@ pub fn create_wasm_channel_router( .with_state(state) } -/// HTTP server for WASM channel webhooks. -pub struct WasmChannelServer { - router: Arc, - extension_manager: Option>, -} - -impl WasmChannelServer { - /// Create a new server. - pub fn new(router: Arc) -> Self { - Self { - router, - extension_manager: None, - } - } - - /// Set the extension manager for OAuth callback handling. - pub fn with_extension_manager( - mut self, - manager: Arc, - ) -> Self { - self.extension_manager = Some(manager); - self - } - - /// Start the HTTP server. - /// - /// Returns a handle that can be used to shut down the server. - pub async fn start( - &self, - addr: SocketAddr, - ) -> Result, std::io::Error> { - let app = create_wasm_channel_router(self.router.clone(), self.extension_manager.clone()); - - let listener = tokio::net::TcpListener::bind(addr).await?; - - tracing::info!( - addr = %addr, - "WASM channel HTTP server started" - ); - - let handle = tokio::spawn(async move { - if let Err(e) = axum::serve(listener, app).await { - tracing::error!("WASM channel HTTP server error: {}", e); - } - }); - - Ok(handle) - } -} - #[cfg(test)] mod tests { use std::sync::Arc; diff --git a/src/channels/webhook_server.rs b/src/channels/webhook_server.rs new file mode 100644 index 00000000..e38341f6 --- /dev/null +++ b/src/channels/webhook_server.rs @@ -0,0 +1,92 @@ +//! Unified HTTP server for all webhook routes. +//! +//! Composes route fragments from HttpChannel, WASM channel router, etc. +//! into a single axum server. Channels define routes but never spawn servers. + +use std::net::SocketAddr; + +use axum::Router; +use tokio::sync::oneshot; +use tokio::task::JoinHandle; + +use crate::error::ChannelError; + +/// Configuration for the unified webhook server. +pub struct WebhookServerConfig { + /// Address to bind the server to. + pub addr: SocketAddr, +} + +/// A single HTTP server that hosts all webhook routes. +/// +/// Channels contribute route fragments via `add_routes()`, then a single +/// `start()` call binds the listener and spawns the server task. +pub struct WebhookServer { + config: WebhookServerConfig, + routes: Vec, + shutdown_tx: Option>, + handle: Option>, +} + +impl WebhookServer { + /// Create a new webhook server with the given bind address. + pub fn new(config: WebhookServerConfig) -> Self { + Self { + config, + routes: Vec::new(), + shutdown_tx: None, + handle: None, + } + } + + /// Accumulate a route fragment. Each fragment should already have its + /// state applied via `.with_state()`. + pub fn add_routes(&mut self, router: Router) { + self.routes.push(router); + } + + /// Bind the listener, merge all route fragments, and spawn the server. + pub async fn start(&mut self) -> Result<(), ChannelError> { + let mut app = Router::new(); + for fragment in self.routes.drain(..) { + app = app.merge(fragment); + } + + let listener = tokio::net::TcpListener::bind(self.config.addr) + .await + .map_err(|e| ChannelError::StartupFailed { + name: "webhook_server".to_string(), + reason: format!("Failed to bind to {}: {}", self.config.addr, e), + })?; + + tracing::info!("Webhook server listening on {}", self.config.addr); + + let (shutdown_tx, shutdown_rx) = oneshot::channel(); + self.shutdown_tx = Some(shutdown_tx); + + let handle = tokio::spawn(async move { + if let Err(e) = axum::serve(listener, app) + .with_graceful_shutdown(async { + let _ = shutdown_rx.await; + tracing::info!("Webhook server shutting down"); + }) + .await + { + tracing::error!("Webhook server error: {}", e); + } + }); + + self.handle = Some(handle); + Ok(()) + } + + /// Signal graceful shutdown and wait for the server task to finish. + pub async fn shutdown(&mut self) { + if let Some(tx) = self.shutdown_tx.take() { + let _ = tx.send(()); + } + if let Some(handle) = self.handle.take() { + let _ = handle.await; + } + } +} diff --git a/src/main.rs b/src/main.rs index cb0b25b0..7314aee6 100644 --- a/src/main.rs +++ b/src/main.rs @@ -8,10 +8,10 @@ use tracing_subscriber::{EnvFilter, layer::SubscriberExt, util::SubscriberInitEx use ironclaw::{ agent::{Agent, AgentDeps}, channels::{ - ChannelManager, HttpChannel, ReplChannel, + ChannelManager, HttpChannel, ReplChannel, WebhookServer, WebhookServerConfig, wasm::{ RegisteredEndpoint, SharedWasmChannel, WasmChannelLoader, WasmChannelRouter, - WasmChannelRuntime, WasmChannelRuntimeConfig, WasmChannelServer, + WasmChannelRuntime, WasmChannelRuntimeConfig, create_wasm_channel_router, }, }, cli::{ @@ -476,19 +476,10 @@ async fn main() -> anyhow::Result<()> { } } - // Add HTTP channel if configured and not CLI-only mode - if !cli.cli_only { - if let Some(ref http_config) = config.channels.http { - channels.add(Box::new(HttpChannel::new(http_config.clone()))); - tracing::info!( - "HTTP channel enabled on {}:{}", - http_config.host, - http_config.port - ); - } - } + // Collect webhook route fragments; a single WebhookServer hosts them all. + let mut webhook_routes: Vec = Vec::new(); - // Load WASM channels if enabled + // Load WASM channels and register their webhook routes. if config.channels.wasm_channels_enabled && config.channels.wasm_channels_dir.exists() { match WasmChannelRuntime::new(WasmChannelRuntimeConfig::default()) { Ok(runtime) => { @@ -500,7 +491,6 @@ async fn main() -> anyhow::Result<()> { .await { Ok(results) => { - // Create router for WASM channel webhooks let wasm_router = Arc::new(WasmChannelRouter::new()); let mut has_webhook_channels = false; @@ -508,10 +498,8 @@ async fn main() -> anyhow::Result<()> { let channel_name = loaded.name().to_string(); tracing::info!("Loaded WASM channel: {}", channel_name); - // Get webhook secret name from capabilities (generic) let secret_name = loaded.webhook_secret_name(); - // Get webhook secret for this channel from secrets store let webhook_secret = if let Some(ref secrets) = secrets_store { secrets .get_decrypted("default", &secret_name) @@ -522,12 +510,9 @@ async fn main() -> anyhow::Result<()> { None }; - // Get the secret header name from capabilities let secret_header = loaded.webhook_secret_header().map(|s| s.to_string()); - // Register channel with router for webhook handling - // Use known webhook path based on channel name let webhook_path = format!("/webhook/{}", channel_name); let endpoints = vec![RegisteredEndpoint { channel_name: channel_name.clone(), @@ -538,8 +523,6 @@ async fn main() -> anyhow::Result<()> { let channel_arc = Arc::new(loaded.channel); - // Inject runtime config into the channel (tunnel_url, webhook_secret) - // This must be done before start() is called { let mut config_updates = std::collections::HashMap::new(); @@ -585,7 +568,6 @@ async fn main() -> anyhow::Result<()> { .await; has_webhook_channels = true; - // Inject credentials for this channel (generic pattern-based injection) if let Some(ref secrets) = secrets_store { match inject_channel_credentials( &channel_arc, @@ -613,38 +595,14 @@ async fn main() -> anyhow::Result<()> { } } - // Wrap in SharedWasmChannel for ChannelManager - // Both the router and ChannelManager share the same underlying channel channels.add(Box::new(SharedWasmChannel::new(channel_arc))); } - // Start WASM channel webhook server if we have channels with webhooks. - // Skip when the HTTP channel already occupies port 8080. - let http_uses_port = - config.channels.http.as_ref().map(|h| h.port) == Some(8080); - if has_webhook_channels - && config.tunnel.public_url.is_some() - && !http_uses_port - { - let mut server = WasmChannelServer::new(wasm_router); - if let Some(ref ext_mgr) = extension_manager { - server = server.with_extension_manager(Arc::clone(ext_mgr)); - } - let addr = std::net::SocketAddr::from(([0, 0, 0, 0], 8080)); - match server.start(addr).await { - Ok(_handle) => { - tracing::info!( - "WASM channel webhook server started on {}", - addr - ); - } - Err(e) => { - tracing::error!( - "Failed to start WASM channel webhook server: {}", - e - ); - } - } + if has_webhook_channels && config.tunnel.public_url.is_some() { + webhook_routes.push(create_wasm_channel_router( + wasm_router, + extension_manager.as_ref().map(Arc::clone), + )); } for (path, err) in &results.errors { @@ -666,6 +624,43 @@ async fn main() -> anyhow::Result<()> { } } + // Add HTTP channel if configured and not CLI-only mode. + // Extract its routes for the unified server; the channel itself just + // provides the mpsc stream. + let mut webhook_server_addr: Option = None; + if !cli.cli_only { + if let Some(ref http_config) = config.channels.http { + let http_channel = HttpChannel::new(http_config.clone()); + webhook_routes.push(http_channel.routes()); + let (host, port) = http_channel.addr(); + webhook_server_addr = Some( + format!("{}:{}", host, port) + .parse() + .expect("HttpConfig host:port must be a valid SocketAddr"), + ); + channels.add(Box::new(http_channel)); + tracing::info!( + "HTTP channel enabled on {}:{}", + http_config.host, + http_config.port + ); + } + } + + // Start the unified webhook server if any routes were registered. + let mut webhook_server = if !webhook_routes.is_empty() { + let addr = + webhook_server_addr.unwrap_or_else(|| std::net::SocketAddr::from(([0, 0, 0, 0], 8080))); + let mut server = WebhookServer::new(WebhookServerConfig { addr }); + for routes in webhook_routes { + server.add_routes(routes); + } + server.start().await?; + Some(server) + } else { + None + }; + // Create workspace for agent (shared with memory tools) let workspace = store.as_ref().map(|s| { let mut ws = Workspace::new("default", s.pool()); @@ -715,6 +710,11 @@ async fn main() -> anyhow::Result<()> { // Run the agent (blocks until shutdown) agent.run().await?; + // Shut down the webhook server if one was started + if let Some(ref mut server) = webhook_server { + server.shutdown().await; + } + tracing::info!("Agent shutdown complete"); Ok(()) }