From 0596a6c8474c6128ab3f6d237a461fbcf3a41f41 Mon Sep 17 00:00:00 2001 From: Illia Polosukhin Date: Wed, 4 Feb 2026 22:02:13 -0800 Subject: [PATCH] Webhook and polling integrations for channels --- channels-src/telegram/src/lib.rs | 82 ++++- src/channels/wasm/loader.rs | 133 +++++--- src/channels/wasm/mod.rs | 7 +- src/channels/wasm/router.rs | 94 +++++- src/channels/wasm/schema.rs | 173 ++++++++++ src/channels/wasm/wrapper.rs | 513 +++++++++++++++++++++--------- src/config.rs | 26 -- src/main.rs | 198 +++++------- src/settings.rs | 10 +- src/setup/channels.rs | 116 ++++++- src/setup/wizard.rs | 169 ++++++++-- tests/wasm_channel_integration.rs | 10 +- 12 files changed, 1146 insertions(+), 385 deletions(-) diff --git a/channels-src/telegram/src/lib.rs b/channels-src/telegram/src/lib.rs index 013f37e7..8a828d1b 100644 --- a/channels-src/telegram/src/lib.rs +++ b/channels-src/telegram/src/lib.rs @@ -241,11 +241,22 @@ impl Guest for TelegramChannel { channel_host::LogLevel::Info, "Webhook mode enabled (polling disabled)", ); - if let Some(ref url) = config.tunnel_url { + + // Register webhook with Telegram API + if let Some(ref tunnel_url) = config.tunnel_url { channel_host::log( channel_host::LogLevel::Info, - &format!("Tunnel URL: {}", url), + &format!("Registering webhook with Telegram API: {}", tunnel_url), ); + + if let Err(e) = register_webhook(tunnel_url, config.webhook_secret.as_deref()) { + channel_host::log( + channel_host::LogLevel::Error, + &format!("Failed to register webhook: {}", e), + ); + // Continue anyway, the host will start the server + // and Telegram will eventually receive updates + } } } @@ -495,6 +506,73 @@ impl Guest for TelegramChannel { } } +// ============================================================================ +// Webhook Registration +// ============================================================================ + +/// Register webhook URL with Telegram API. +/// +/// Called during on_start() when tunnel_url is configured. +fn register_webhook(tunnel_url: &str, webhook_secret: Option<&str>) -> Result<(), String> { + let webhook_url = format!("{}/webhook/telegram", tunnel_url); + + // Build setWebhook request body + let mut body = serde_json::json!({ + "url": webhook_url, + "allowed_updates": ["message", "edited_message"] + }); + + if let Some(secret) = webhook_secret { + body["secret_token"] = serde_json::Value::String(secret.to_string()); + } + + let body_bytes = serde_json::to_vec(&body).map_err(|e| format!("Failed to serialize body: {}", e))?; + + let headers = serde_json::json!({ + "Content-Type": "application/json" + }); + + // Make HTTP request to Telegram API + // Note: {TELEGRAM_BOT_TOKEN} is replaced by host with the actual token + let result = channel_host::http_request( + "POST", + "https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/setWebhook", + &headers.to_string(), + Some(&body_bytes), + ); + + match result { + Ok(response) => { + if response.status != 200 { + let body_str = String::from_utf8_lossy(&response.body); + return Err(format!("HTTP {}: {}", response.status, body_str)); + } + + // Parse Telegram API response + let api_response: TelegramApiResponse = + serde_json::from_slice(&response.body) + .map_err(|e| format!("Failed to parse response: {}", e))?; + + if !api_response.ok { + return Err(format!( + "Telegram API error: {}", + api_response + .description + .unwrap_or_else(|| "unknown".to_string()) + )); + } + + channel_host::log( + channel_host::LogLevel::Info, + &format!("Webhook registered successfully: {}", webhook_url), + ); + + Ok(()) + } + Err(e) => Err(format!("HTTP request failed: {}", e)), + } +} + // ============================================================================ // Update Handling // ============================================================================ diff --git a/src/channels/wasm/loader.rs b/src/channels/wasm/loader.rs index fc2df29d..5e94d38c 100644 --- a/src/channels/wasm/loader.rs +++ b/src/channels/wasm/loader.rs @@ -40,7 +40,7 @@ impl WasmChannelLoader { name: &str, wasm_path: &Path, capabilities_path: Option<&Path>, - ) -> Result { + ) -> Result { // Validate name if name.is_empty() || name.contains('/') || name.contains('\\') || name.contains("..") { return Err(WasmChannelError::InvalidName(name.to_string())); @@ -53,56 +53,59 @@ impl WasmChannelLoader { let wasm_bytes = fs::read(wasm_path).await?; // Read capabilities file - let (capabilities, config_json, description) = if let Some(cap_path) = capabilities_path { - if cap_path.exists() { - let cap_bytes = fs::read(cap_path).await?; - let cap_file = ChannelCapabilitiesFile::from_bytes(&cap_bytes) - .map_err(|e| WasmChannelError::InvalidCapabilities(e.to_string()))?; + let (capabilities, config_json, description, cap_file) = + if let Some(cap_path) = capabilities_path { + if cap_path.exists() { + let cap_bytes = fs::read(cap_path).await?; + let cap_file = ChannelCapabilitiesFile::from_bytes(&cap_bytes) + .map_err(|e| WasmChannelError::InvalidCapabilities(e.to_string()))?; - // Debug: log raw capabilities - tracing::debug!( - channel = name, - raw_capabilities = ?cap_file.capabilities, - "Parsed capabilities file" - ); + // Debug: log raw capabilities + tracing::debug!( + channel = name, + raw_capabilities = ?cap_file.capabilities, + "Parsed capabilities file" + ); - let caps = cap_file.to_capabilities(); + let caps = cap_file.to_capabilities(); - // Debug: log resulting capabilities - tracing::info!( - channel = name, - http_allowed = caps.tool_capabilities.http.is_some(), - http_allowlist_count = caps - .tool_capabilities - .http - .as_ref() - .map(|h| h.allowlist.len()) - .unwrap_or(0), - "Channel capabilities loaded" - ); + // Debug: log resulting capabilities + tracing::info!( + channel = name, + http_allowed = caps.tool_capabilities.http.is_some(), + http_allowlist_count = caps + .tool_capabilities + .http + .as_ref() + .map(|h| h.allowlist.len()) + .unwrap_or(0), + "Channel capabilities loaded" + ); - let config = cap_file.config_json(); - let desc = cap_file.description.clone(); + let config = cap_file.config_json(); + let desc = cap_file.description.clone(); - (caps, config, desc) + (caps, config, desc, Some(cap_file)) + } else { + tracing::warn!( + path = %cap_path.display(), + "Capabilities file not found, using defaults" + ); + ( + ChannelCapabilities::for_channel(name), + "{}".to_string(), + None, + None, + ) + } } else { - tracing::warn!( - path = %cap_path.display(), - "Capabilities file not found, using defaults" - ); ( ChannelCapabilities::for_channel(name), "{}".to_string(), None, + None, ) - } - } else { - ( - ChannelCapabilities::for_channel(name), - "{}".to_string(), - None, - ) - }; + }; // Prepare the module let prepared = self @@ -119,7 +122,10 @@ impl WasmChannelLoader { "Loaded WASM channel from file" ); - Ok(channel) + Ok(LoadedChannel { + channel, + capabilities_file: cap_file, + }) } /// Load all WASM channels from a directory. @@ -176,8 +182,8 @@ impl WasmChannelLoader { }; match self.load_from_files(&name, &path, cap_path_option).await { - Ok(channel) => { - results.loaded.push(channel); + Ok(loaded) => { + results.loaded.push(loaded); } Err(e) => { tracing::error!( @@ -194,7 +200,7 @@ impl WasmChannelLoader { if !results.loaded.is_empty() { tracing::info!( count = results.loaded.len(), - channels = ?results.loaded.iter().map(|c| c.channel_name()).collect::>(), + channels = ?results.loaded.iter().map(|c| c.name()).collect::>(), "Loaded WASM channels from directory" ); } @@ -203,11 +209,42 @@ impl WasmChannelLoader { } } +/// A loaded WASM channel with its capabilities file. +pub struct LoadedChannel { + /// The loaded channel. + pub channel: WasmChannel, + + /// The parsed capabilities file (if present). + pub capabilities_file: Option, +} + +impl LoadedChannel { + /// Get the channel name. + pub fn name(&self) -> &str { + self.channel.channel_name() + } + + /// Get the webhook secret header name from capabilities. + pub fn webhook_secret_header(&self) -> Option<&str> { + self.capabilities_file + .as_ref() + .and_then(|f| f.webhook_secret_header()) + } + + /// Get the webhook secret name from capabilities. + pub fn webhook_secret_name(&self) -> String { + self.capabilities_file + .as_ref() + .map(|f| f.webhook_secret_name()) + .unwrap_or_else(|| format!("{}_webhook_secret", self.channel.channel_name())) + } +} + /// Results from loading multiple channels. #[derive(Default)] pub struct LoadResults { - /// Successfully loaded channels. - pub loaded: Vec, + /// Successfully loaded channels with their capabilities. + pub loaded: Vec, /// Errors encountered (path, error). pub errors: Vec<(PathBuf, WasmChannelError)>, @@ -229,9 +266,9 @@ impl LoadResults { self.errors.len() } - /// Take ownership of loaded channels. + /// Take ownership of loaded channels (extracts just the WasmChannel). pub fn take_channels(self) -> Vec { - self.loaded + self.loaded.into_iter().map(|l| l.channel).collect() } } diff --git a/src/channels/wasm/mod.rs b/src/channels/wasm/mod.rs index 55fafafd..24905fc0 100644 --- a/src/channels/wasm/mod.rs +++ b/src/channels/wasm/mod.rs @@ -92,11 +92,14 @@ pub use capabilities::{ChannelCapabilities, EmitRateLimitConfig, HttpEndpointCon pub use error::WasmChannelError; pub use host::{ChannelEmitRateLimiter, ChannelHostState, EmittedMessage}; pub use loader::{ - DiscoveredChannel, LoadResults, WasmChannelLoader, default_channels_dir, discover_channels, + DiscoveredChannel, LoadResults, LoadedChannel, WasmChannelLoader, default_channels_dir, + discover_channels, }; pub use router::{ RegisteredEndpoint, WasmChannelRouter, WasmChannelServer, create_wasm_channel_router, }; pub use runtime::{PreparedChannelModule, WasmChannelRuntime, WasmChannelRuntimeConfig}; -pub use schema::{ChannelCapabilitiesFile, ChannelConfig}; +pub use schema::{ + ChannelCapabilitiesFile, ChannelConfig, SecretSetupSchema, SetupSchema, WebhookSchema, +}; pub use wrapper::{HttpResponse, SharedWasmChannel, WasmChannel}; diff --git a/src/channels/wasm/router.rs b/src/channels/wasm/router.rs index 609ea9e9..892f9e4a 100644 --- a/src/channels/wasm/router.rs +++ b/src/channels/wasm/router.rs @@ -41,6 +41,8 @@ pub struct WasmChannelRouter { path_to_channel: RwLock>, /// Expected webhook secrets by channel name. secrets: RwLock>, + /// Webhook secret header names by channel name (e.g., "X-Telegram-Bot-Api-Secret-Token"). + secret_headers: RwLock>, } impl WasmChannelRouter { @@ -50,15 +52,24 @@ impl WasmChannelRouter { channels: RwLock::new(HashMap::new()), path_to_channel: RwLock::new(HashMap::new()), secrets: RwLock::new(HashMap::new()), + secret_headers: RwLock::new(HashMap::new()), } } /// Register a channel with its endpoints. + /// + /// # Arguments + /// * `channel` - The WASM channel to register + /// * `endpoints` - HTTP endpoints to register for this channel + /// * `secret` - Optional webhook secret for validation + /// * `secret_header` - Optional HTTP header name for secret validation + /// (e.g., "X-Telegram-Bot-Api-Secret-Token"). Defaults to "X-Webhook-Secret". pub async fn register( &self, channel: Arc, endpoints: Vec, secret: Option, + secret_header: Option, ) { let name = channel.channel_name().to_string(); @@ -79,14 +90,32 @@ impl WasmChannelRouter { // Store secret if provided if let Some(s) = secret { - self.secrets.write().await.insert(name, s); + self.secrets.write().await.insert(name.clone(), s); } + + // Store secret header if provided + if let Some(h) = secret_header { + self.secret_headers.write().await.insert(name, h); + } + } + + /// Get the secret header name for a channel. + /// + /// Returns the configured header or "X-Webhook-Secret" as default. + pub async fn get_secret_header(&self, channel_name: &str) -> String { + self.secret_headers + .read() + .await + .get(channel_name) + .cloned() + .unwrap_or_else(|| "X-Webhook-Secret".to_string()) } /// Unregister a channel and its endpoints. pub async fn unregister(&self, channel_name: &str) { self.channels.write().await.remove(channel_name); self.secrets.write().await.remove(channel_name); + self.secret_headers.write().await.remove(channel_name); // Remove all paths for this channel self.path_to_channel @@ -224,23 +253,29 @@ async fn webhook_handler( // Check if secret is required if state.router.requires_secret(channel_name).await { - // Try to get secret from query param or header - // Telegram uses X-Telegram-Bot-Api-Secret-Token header + // Get the secret header name for this channel (from capabilities or default) + let secret_header_name = state.router.get_secret_header(channel_name).await; + + // Try to get secret from query param or the channel's configured header let provided_secret = query .get("secret") .cloned() .or_else(|| { headers - .get("X-Telegram-Bot-Api-Secret-Token") + .get(&secret_header_name) .and_then(|v| v.to_str().ok()) .map(|s| s.to_string()) }) .or_else(|| { - // Fallback to generic header - headers - .get("X-Webhook-Secret") - .and_then(|v| v.to_str().ok()) - .map(|s| s.to_string()) + // Fallback to generic header if different from configured + if secret_header_name != "X-Webhook-Secret" { + headers + .get("X-Webhook-Secret") + .and_then(|v| v.to_str().ok()) + .map(|s| s.to_string()) + } else { + None + } }); tracing::debug!( @@ -447,7 +482,7 @@ mod tests { }]; router - .register(channel, endpoints, Some("secret123".to_string())) + .register(channel, endpoints, Some("secret123".to_string()), None) .await; // Should find channel by path @@ -466,7 +501,7 @@ mod tests { let channel = create_test_channel("slack"); router - .register(channel, vec![], Some("secret123".to_string())) + .register(channel, vec![], Some("secret123".to_string()), None) .await; // Correct secret @@ -477,7 +512,7 @@ mod tests { // Channel without secret always validates let channel2 = create_test_channel("telegram"); - router.register(channel2, vec![], None).await; + router.register(channel2, vec![], None, None).await; assert!(router.validate_secret("telegram", "anything").await); } @@ -493,7 +528,7 @@ mod tests { require_secret: false, }]; - router.register(channel, endpoints, None).await; + router.register(channel, endpoints, None, None).await; // Should exist assert!( @@ -522,12 +557,41 @@ mod tests { let channel1 = create_test_channel("slack"); let channel2 = create_test_channel("telegram"); - router.register(channel1, vec![], None).await; - router.register(channel2, vec![], None).await; + router.register(channel1, vec![], None, None).await; + router.register(channel2, vec![], None, None).await; let channels = router.list_channels().await; assert_eq!(channels.len(), 2); assert!(channels.contains(&"slack".to_string())); assert!(channels.contains(&"telegram".to_string())); } + + #[tokio::test] + async fn test_router_secret_header() { + let router = WasmChannelRouter::new(); + let channel = create_test_channel("telegram"); + + // Register with custom secret header + router + .register( + channel, + vec![], + Some("secret123".to_string()), + Some("X-Telegram-Bot-Api-Secret-Token".to_string()), + ) + .await; + + // Should return the custom header + assert_eq!( + router.get_secret_header("telegram").await, + "X-Telegram-Bot-Api-Secret-Token" + ); + + // Channel without custom header should use default + let channel2 = create_test_channel("slack"); + router + .register(channel2, vec![], Some("secret456".to_string()), None) + .await; + assert_eq!(router.get_secret_header("slack").await, "X-Webhook-Secret"); + } } diff --git a/src/channels/wasm/schema.rs b/src/channels/wasm/schema.rs index 06876822..f0769dee 100644 --- a/src/channels/wasm/schema.rs +++ b/src/channels/wasm/schema.rs @@ -62,6 +62,10 @@ pub struct ChannelCapabilitiesFile { #[serde(default)] pub description: Option, + /// Setup configuration for the wizard. + #[serde(default)] + pub setup: SetupSchema, + /// Capabilities (tool + channel specific). #[serde(default)] pub capabilities: ChannelCapabilitiesSchema, @@ -95,6 +99,29 @@ impl ChannelCapabilitiesFile { pub fn config_json(&self) -> String { serde_json::to_string(&self.config).unwrap_or_else(|_| "{}".to_string()) } + + /// Get the webhook secret header name for this channel. + /// + /// Returns the configured header name from capabilities, or a sensible default. + pub fn webhook_secret_header(&self) -> Option<&str> { + self.capabilities + .channel + .as_ref() + .and_then(|c| c.webhook.as_ref()) + .and_then(|w| w.secret_header.as_deref()) + } + + /// Get the webhook secret name for this channel. + /// + /// Returns the configured secret name or defaults to "{channel_name}_webhook_secret". + pub fn webhook_secret_name(&self) -> String { + self.capabilities + .channel + .as_ref() + .and_then(|c| c.webhook.as_ref()) + .and_then(|w| w.secret_name.clone()) + .unwrap_or_else(|| format!("{}_webhook_secret", self.name)) + } } /// Schema for channel capabilities. @@ -178,6 +205,80 @@ pub struct ChannelSpecificCapabilitiesSchema { /// Callback timeout in seconds. #[serde(default)] pub callback_timeout_secs: Option, + + /// Webhook configuration (secret header, etc.). + #[serde(default)] + pub webhook: Option, +} + +/// Webhook configuration schema. +/// +/// Allows channels to specify their webhook validation requirements. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct WebhookSchema { + /// HTTP header name for secret validation. + /// + /// Examples: + /// - Telegram: "X-Telegram-Bot-Api-Secret-Token" + /// - Slack: "X-Slack-Signature" + /// - GitHub: "X-Hub-Signature-256" + /// - Generic: "X-Webhook-Secret" + #[serde(default)] + pub secret_header: Option, + + /// Secret name in secrets store for webhook validation. + /// Default: "{channel_name}_webhook_secret" + #[serde(default)] + pub secret_name: Option, +} + +/// Setup configuration schema. +/// +/// Allows channels to declare their setup requirements for the wizard. +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct SetupSchema { + /// Required secrets that must be configured during setup. + #[serde(default)] + pub required_secrets: Vec, + + /// Optional validation endpoint to verify configuration. + /// Placeholders like {secret_name} are replaced with actual values. + #[serde(default)] + pub validation_endpoint: Option, +} + +/// Configuration for a secret required during setup. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct SecretSetupSchema { + /// Secret name in the secrets store (e.g., "telegram_bot_token"). + pub name: String, + + /// Prompt to show the user during setup. + pub prompt: String, + + /// Optional regex for validation. + #[serde(default)] + pub validation: Option, + + /// Whether this secret is optional. + #[serde(default)] + pub optional: bool, + + /// Auto-generate configuration if the user doesn't provide a value. + #[serde(default)] + pub auto_generate: Option, +} + +/// Configuration for auto-generating a secret value. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct AutoGenerateSchema { + /// Length of the generated value in bytes (will be hex-encoded). + #[serde(default = "default_auto_generate_length")] + pub length: usize, +} + +fn default_auto_generate_length() -> usize { + 32 } /// Schema for emit rate limiting. @@ -412,4 +513,76 @@ mod tests { assert_eq!(caps.emit_rate_limit.messages_per_minute, 50); assert_eq!(caps.emit_rate_limit.messages_per_hour, 1000); } + + #[test] + fn test_webhook_schema() { + let json = r#"{ + "name": "telegram", + "capabilities": { + "channel": { + "allowed_paths": ["/webhook/telegram"], + "webhook": { + "secret_header": "X-Telegram-Bot-Api-Secret-Token", + "secret_name": "telegram_webhook_secret" + } + } + } + }"#; + + let file = ChannelCapabilitiesFile::from_json(json).unwrap(); + assert_eq!( + file.webhook_secret_header(), + Some("X-Telegram-Bot-Api-Secret-Token") + ); + assert_eq!(file.webhook_secret_name(), "telegram_webhook_secret"); + } + + #[test] + fn test_webhook_secret_name_default() { + let json = r#"{ + "name": "mybot", + "capabilities": {} + }"#; + + let file = ChannelCapabilitiesFile::from_json(json).unwrap(); + assert_eq!(file.webhook_secret_header(), None); + assert_eq!(file.webhook_secret_name(), "mybot_webhook_secret"); + } + + #[test] + fn test_setup_schema() { + let json = r#"{ + "name": "telegram", + "setup": { + "required_secrets": [ + { + "name": "telegram_bot_token", + "prompt": "Enter your Telegram Bot Token", + "validation": "^[0-9]+:[A-Za-z0-9_-]+$" + }, + { + "name": "telegram_webhook_secret", + "prompt": "Webhook secret (leave empty to auto-generate)", + "optional": true, + "auto_generate": { "length": 64 } + } + ], + "validation_endpoint": "https://api.telegram.org/bot{telegram_bot_token}/getMe" + } + }"#; + + let file = ChannelCapabilitiesFile::from_json(json).unwrap(); + assert_eq!(file.setup.required_secrets.len(), 2); + assert_eq!(file.setup.required_secrets[0].name, "telegram_bot_token"); + assert!(!file.setup.required_secrets[0].optional); + assert!(file.setup.required_secrets[1].optional); + assert_eq!( + file.setup.required_secrets[1] + .auto_generate + .as_ref() + .unwrap() + .length, + 64 + ); + } } diff --git a/src/channels/wasm/wrapper.rs b/src/channels/wasm/wrapper.rs index 2bb1db40..30850ee5 100644 --- a/src/channels/wasm/wrapper.rs +++ b/src/channels/wasm/wrapper.rs @@ -356,23 +356,29 @@ pub struct WasmChannel { channel_config: RwLock>, /// Message sender (for emitting messages to the stream). - message_tx: RwLock>>, + /// Wrapped in Arc for sharing with the polling task. + message_tx: Arc>>>, /// Pending responses (for synchronous response handling). pending_responses: RwLock>>, /// Rate limiter for message emission. - rate_limiter: RwLock, + /// Wrapped in Arc for sharing with the polling task. + rate_limiter: Arc>, /// Shutdown signal sender. shutdown_tx: RwLock>>, + /// Polling shutdown signal sender (keeps polling alive while held). + poll_shutdown_tx: RwLock>>, + /// Registered HTTP endpoints. endpoints: RwLock>, /// Injected credentials for HTTP requests (e.g., bot tokens). /// Keys are placeholder names like "TELEGRAM_BOT_TOKEN". - credentials: RwLock>, + /// Wrapped in Arc for sharing with the polling task. + credentials: Arc>>, } impl WasmChannel { @@ -393,12 +399,13 @@ impl WasmChannel { capabilities, config_json, channel_config: RwLock::new(None), - message_tx: RwLock::new(None), + message_tx: Arc::new(RwLock::new(None)), pending_responses: RwLock::new(HashMap::new()), - rate_limiter: RwLock::new(rate_limiter), + rate_limiter: Arc::new(RwLock::new(rate_limiter)), shutdown_tx: RwLock::new(None), + poll_shutdown_tx: RwLock::new(None), endpoints: RwLock::new(Vec::new()), - credentials: RwLock::new(HashMap::new()), + credentials: Arc::new(RwLock::new(HashMap::new())), } } @@ -430,142 +437,6 @@ impl WasmChannel { self.endpoints.read().await.clone() } - /// Register a webhook URL with Telegram. - /// - /// Called during channel startup if tunnel_url is configured. - /// This enables instant message delivery instead of polling. - pub async fn register_telegram_webhook( - &self, - tunnel_url: &str, - bot_token: &str, - secret_token: Option<&str>, - ) -> Result<(), WasmChannelError> { - let webhook_url = format!("{}/webhook/telegram", tunnel_url); - - tracing::info!( - channel = %self.name, - webhook_url = %webhook_url, - "Registering Telegram webhook" - ); - - // Build form parameters - let mut form_params = vec![ - ("url", webhook_url.as_str()), - ("allowed_updates", r#"["message","edited_message"]"#), - ]; - - let secret_owned: String; - if let Some(secret) = secret_token { - secret_owned = secret.to_string(); - form_params.push(("secret_token", &secret_owned)); - } - - let client = reqwest::Client::builder() - .timeout(std::time::Duration::from_secs(30)) - .build() - .map_err(|e| WasmChannelError::HttpRequest(e.to_string()))?; - - let response = client - .post(format!( - "https://api.telegram.org/bot{}/setWebhook", - bot_token - )) - .form(&form_params) - .send() - .await - .map_err(|e| WasmChannelError::HttpRequest(e.to_string()))?; - - if !response.status().is_success() { - let status = response.status(); - let body = response.text().await.unwrap_or_default(); - return Err(WasmChannelError::WebhookRegistration { - name: self.name.clone(), - reason: format!("HTTP {}: {}", status, body), - }); - } - - // Parse Telegram API response - let result: serde_json::Value = response - .json() - .await - .map_err(|e| WasmChannelError::HttpRequest(e.to_string()))?; - - if result["ok"].as_bool() != Some(true) { - let description = result["description"] - .as_str() - .unwrap_or("unknown error") - .to_string(); - return Err(WasmChannelError::WebhookRegistration { - name: self.name.clone(), - reason: description, - }); - } - - tracing::info!( - channel = %self.name, - webhook_url = %webhook_url, - "Telegram webhook registered successfully" - ); - - Ok(()) - } - - /// Delete the webhook and switch back to polling mode. - /// - /// Called during shutdown if webhook was registered. - pub async fn delete_telegram_webhook(&self, bot_token: &str) -> Result<(), WasmChannelError> { - tracing::info!( - channel = %self.name, - "Deleting Telegram webhook" - ); - - let client = reqwest::Client::builder() - .timeout(std::time::Duration::from_secs(30)) - .build() - .map_err(|e| WasmChannelError::HttpRequest(e.to_string()))?; - - let response = client - .post(format!( - "https://api.telegram.org/bot{}/deleteWebhook", - bot_token - )) - .send() - .await - .map_err(|e| WasmChannelError::HttpRequest(e.to_string()))?; - - if !response.status().is_success() { - let status = response.status(); - let body = response.text().await.unwrap_or_default(); - return Err(WasmChannelError::WebhookRegistration { - name: self.name.clone(), - reason: format!("HTTP {} (delete): {}", status, body), - }); - } - - let result: serde_json::Value = response - .json() - .await - .map_err(|e| WasmChannelError::HttpRequest(e.to_string()))?; - - if result["ok"].as_bool() != Some(true) { - let description = result["description"] - .as_str() - .unwrap_or("unknown error") - .to_string(); - return Err(WasmChannelError::WebhookRegistration { - name: self.name.clone(), - reason: format!("delete failed: {}", description), - }); - } - - tracing::info!( - channel = %self.name, - "Telegram webhook deleted" - ); - - Ok(()) - } - /// Add channel host functions to the linker using generated bindings. /// /// Uses the wasmtime::component::bindgen! generated `add_to_linker` function @@ -1154,11 +1025,20 @@ impl WasmChannel { } /// Start the polling loop if configured. + /// + /// Since we can't hold `Arc` from `&self`, we pass all the components + /// needed for polling to a spawned task. Each poll tick creates a fresh WASM + /// instance (matching our "fresh instance per callback" pattern). fn start_polling(&self, interval: Duration, shutdown_rx: oneshot::Receiver<()>) { let channel_name = self.name.clone(); + let runtime = Arc::clone(&self.runtime); + let prepared = Arc::clone(&self.prepared); + let capabilities = self.capabilities.clone(); + let message_tx = self.message_tx.clone(); + let rate_limiter = self.rate_limiter.clone(); + let credentials = self.credentials.clone(); + let callback_timeout = self.runtime.config().callback_timeout; - // Clone self reference for the async block - // In a real implementation, we'd hold an Arc tokio::spawn(async move { let mut interval_timer = tokio::time::interval(interval); let mut shutdown = std::pin::pin!(shutdown_rx); @@ -1168,9 +1048,45 @@ impl WasmChannel { _ = interval_timer.tick() => { tracing::debug!( channel = %channel_name, - "Polling tick (stub - would call on_poll)" + "Polling tick - calling on_poll" ); - // In real implementation: self.call_on_poll().await + + // Execute on_poll with fresh WASM instance + let result = Self::execute_poll( + &channel_name, + &runtime, + &prepared, + &capabilities, + &credentials, + callback_timeout, + ).await; + + match result { + Ok(emitted_messages) => { + // Process any emitted messages + if !emitted_messages.is_empty() { + if let Err(e) = Self::dispatch_emitted_messages( + &channel_name, + emitted_messages, + &message_tx, + &rate_limiter, + ).await { + tracing::warn!( + channel = %channel_name, + error = %e, + "Failed to dispatch emitted messages from poll" + ); + } + } + } + Err(e) => { + tracing::warn!( + channel = %channel_name, + error = %e, + "Polling callback failed" + ); + } + } } _ = &mut shutdown => { tracing::info!( @@ -1183,6 +1099,156 @@ impl WasmChannel { } }); } + + /// Execute a single poll callback with a fresh WASM instance. + /// + /// Returns any emitted messages from the callback. + async fn execute_poll( + channel_name: &str, + runtime: &Arc, + prepared: &Arc, + capabilities: &ChannelCapabilities, + credentials: &RwLock>, + timeout: Duration, + ) -> Result, WasmChannelError> { + // Skip if no WASM bytes (testing mode) + if prepared.component_bytes.is_empty() { + tracing::debug!( + channel = %channel_name, + "WASM channel on_poll called (no WASM module)" + ); + return Ok(Vec::new()); + } + + let runtime = Arc::clone(runtime); + let prepared = Arc::clone(prepared); + let capabilities = capabilities.clone(); + let credentials_snapshot = credentials.read().await.clone(); + let channel_name_owned = channel_name.to_string(); + + // Execute in blocking task with timeout + let result = tokio::time::timeout(timeout, async move { + tokio::task::spawn_blocking(move || { + let mut store = + Self::create_store(&runtime, &prepared, &capabilities, credentials_snapshot)?; + let instance = Self::instantiate_component(&runtime, &prepared, &mut store)?; + + // Call on_poll using the generated typed interface + let channel_iface = instance.near_agent_channel(); + channel_iface + .call_on_poll(&mut store) + .map_err(|e| Self::map_wasm_error(e, &prepared.name, prepared.limits.fuel))?; + + let host_state = + Self::extract_host_state(&mut store, &prepared.name, &capabilities); + Ok(host_state) + }) + .await + .map_err(|e| WasmChannelError::ExecutionPanicked { + name: channel_name_owned.clone(), + reason: e.to_string(), + })? + }) + .await; + + match result { + Ok(Ok(mut host_state)) => { + let emitted = host_state.take_emitted_messages(); + tracing::debug!( + channel = %channel_name, + emitted_count = emitted.len(), + "WASM channel on_poll completed" + ); + Ok(emitted) + } + Ok(Err(e)) => Err(e), + Err(_) => Err(WasmChannelError::Timeout { + name: channel_name.to_string(), + callback: "on_poll".to_string(), + }), + } + } + + /// Dispatch emitted messages to the message channel. + /// + /// This is a static helper used by the polling loop since it doesn't have + /// access to `&self`. + async fn dispatch_emitted_messages( + channel_name: &str, + messages: Vec, + message_tx: &RwLock>>, + rate_limiter: &RwLock, + ) -> Result<(), WasmChannelError> { + tracing::info!( + channel = %channel_name, + message_count = messages.len(), + "Processing emitted messages from polling callback" + ); + + let tx_guard = message_tx.read().await; + let Some(tx) = tx_guard.as_ref() else { + tracing::error!( + channel = %channel_name, + count = messages.len(), + "Messages emitted but no sender available - channel may not be started!" + ); + return Ok(()); + }; + + let mut limiter = rate_limiter.write().await; + + for emitted in messages { + // Check rate limit + if !limiter.check_and_record() { + tracing::warn!( + channel = %channel_name, + "Message emission rate limited" + ); + return Err(WasmChannelError::EmitRateLimited { + name: channel_name.to_string(), + }); + } + + // Convert to IncomingMessage + let mut msg = IncomingMessage::new(channel_name, &emitted.user_id, &emitted.content); + + if let Some(name) = emitted.user_name { + msg = msg.with_user_name(name); + } + + if let Some(thread_id) = emitted.thread_id { + msg = msg.with_thread(thread_id); + } + + // Parse metadata JSON + if let Ok(metadata) = serde_json::from_str(&emitted.metadata_json) { + msg = msg.with_metadata(metadata); + } + + // Send to stream + tracing::info!( + channel = %channel_name, + user_id = %emitted.user_id, + content_len = emitted.content.len(), + "Sending polled message to agent" + ); + + if tx.send(msg).await.is_err() { + tracing::error!( + channel = %channel_name, + "Failed to send polled message, channel closed" + ); + break; + } + + tracing::info!( + channel = %channel_name, + "Message successfully sent to agent queue" + ); + } + + Ok(()) + } } #[async_trait] @@ -1245,8 +1311,9 @@ impl Channel for WasmChannel { reason: e, })?; - // Create a new shutdown receiver for polling - let (_poll_shutdown_tx, poll_shutdown_rx) = oneshot::channel(); + // Create shutdown channel for polling and store the sender to keep it alive + let (poll_shutdown_tx, poll_shutdown_rx) = oneshot::channel(); + *self.poll_shutdown_tx.write().await = Some(poll_shutdown_tx); self.start_polling(Duration::from_millis(interval as u64), poll_shutdown_rx); } @@ -1316,6 +1383,9 @@ impl Channel for WasmChannel { let _ = tx.send(()); } + // Stop polling by dropping the sender (receiver will complete) + let _ = self.poll_shutdown_tx.write().await.take(); + // Clear the message sender *self.message_tx.write().await = None; @@ -1558,4 +1628,151 @@ mod tests { // Health check should fail after shutdown assert!(channel.health_check().await.is_err()); } + + #[tokio::test] + async fn test_execute_poll_no_wasm_returns_empty() { + // When there's no WASM module (empty component_bytes), execute_poll + // should return an empty vector of messages + let config = WasmChannelRuntimeConfig::for_testing(); + let runtime = Arc::new(WasmChannelRuntime::new(config).unwrap()); + + let prepared = Arc::new(PreparedChannelModule { + name: "poll-test".to_string(), + description: "Test channel".to_string(), + component_bytes: Vec::new(), // No WASM bytes + limits: ResourceLimits::default(), + }); + + let capabilities = ChannelCapabilities::for_channel("poll-test").with_polling(1000); + let credentials = Arc::new(tokio::sync::RwLock::new(std::collections::HashMap::new())); + let timeout = std::time::Duration::from_secs(5); + + let result = WasmChannel::execute_poll( + "poll-test", + &runtime, + &prepared, + &capabilities, + &credentials, + timeout, + ) + .await; + + assert!(result.is_ok()); + assert!(result.unwrap().is_empty()); + } + + #[tokio::test] + async fn test_dispatch_emitted_messages_sends_to_channel() { + use crate::channels::wasm::host::EmittedMessage; + + let (tx, mut rx) = tokio::sync::mpsc::channel(10); + let message_tx = Arc::new(tokio::sync::RwLock::new(Some(tx))); + + let rate_limiter = Arc::new(tokio::sync::RwLock::new( + crate::channels::wasm::host::ChannelEmitRateLimiter::new( + crate::channels::wasm::capabilities::EmitRateLimitConfig::default(), + ), + )); + + let messages = vec![ + EmittedMessage::new("user1", "Hello from polling!"), + EmittedMessage::new("user2", "Another message"), + ]; + + let result = WasmChannel::dispatch_emitted_messages( + "test-channel", + messages, + &message_tx, + &rate_limiter, + ) + .await; + + assert!(result.is_ok()); + + // Verify messages were sent + let msg1 = rx.try_recv().expect("Should receive first message"); + assert_eq!(msg1.user_id, "user1"); + assert_eq!(msg1.content, "Hello from polling!"); + + let msg2 = rx.try_recv().expect("Should receive second message"); + assert_eq!(msg2.user_id, "user2"); + assert_eq!(msg2.content, "Another message"); + + // No more messages + assert!(rx.try_recv().is_err()); + } + + #[tokio::test] + async fn test_dispatch_emitted_messages_no_sender_returns_ok() { + use crate::channels::wasm::host::EmittedMessage; + + // No sender available (channel not started) + let message_tx = Arc::new(tokio::sync::RwLock::new(None)); + let rate_limiter = Arc::new(tokio::sync::RwLock::new( + crate::channels::wasm::host::ChannelEmitRateLimiter::new( + crate::channels::wasm::capabilities::EmitRateLimitConfig::default(), + ), + )); + + let messages = vec![EmittedMessage::new("user1", "Hello!")]; + + // Should return Ok even without a sender (logs warning but doesn't fail) + let result = WasmChannel::dispatch_emitted_messages( + "test-channel", + messages, + &message_tx, + &rate_limiter, + ) + .await; + + assert!(result.is_ok()); + } + + #[tokio::test] + async fn test_channel_with_polling_stores_shutdown_sender() { + // Create a channel with polling capabilities + let config = WasmChannelRuntimeConfig::for_testing(); + let runtime = Arc::new(WasmChannelRuntime::new(config).unwrap()); + + let prepared = Arc::new(PreparedChannelModule { + name: "poll-channel".to_string(), + description: "Polling test channel".to_string(), + component_bytes: Vec::new(), + limits: ResourceLimits::default(), + }); + + // Enable polling with a 1 second minimum interval + let capabilities = ChannelCapabilities::for_channel("poll-channel") + .with_path("/webhook/poll") + .with_polling(1000); + + let channel = WasmChannel::new(runtime, prepared, capabilities, "{}".to_string()); + + // Start the channel + let _stream = channel.start().await.expect("Channel should start"); + + // Verify poll_shutdown_tx is set (polling was started) + // Note: For testing channels without WASM, on_start returns no poll config, + // so polling won't actually be started. This verifies the basic lifecycle. + assert!(channel.health_check().await.is_ok()); + + // Shutdown should clean up properly + channel.shutdown().await.expect("Shutdown should succeed"); + assert!(channel.health_check().await.is_err()); + } + + #[tokio::test] + async fn test_call_on_poll_no_wasm_succeeds() { + // Verify call_on_poll returns Ok when there's no WASM module + let channel = create_test_channel(); + + // Start the channel first to set up message_tx + let _stream = channel.start().await.expect("Channel should start"); + + // call_on_poll should succeed (no-op for no WASM) + let result = channel.call_on_poll().await; + assert!(result.is_ok()); + + channel.shutdown().await.expect("Shutdown should succeed"); + } } diff --git a/src/config.rs b/src/config.rs index bdbd0200..e561fe73 100644 --- a/src/config.rs +++ b/src/config.rs @@ -308,37 +308,12 @@ fn default_session_path() -> PathBuf { pub struct ChannelsConfig { pub cli: CliConfig, pub http: Option, - pub telegram: TelegramChannelConfig, /// Directory containing WASM channel modules (default: ~/.near-agent/channels/). pub wasm_channels_dir: std::path::PathBuf, /// Whether WASM channels are enabled. pub wasm_channels_enabled: bool, } -/// Telegram channel configuration. -/// -/// The tunnel URL for webhook mode comes from the global `TunnelConfig`. -/// This config only contains Telegram-specific settings. -#[derive(Debug, Clone, Default)] -pub struct TelegramChannelConfig { - /// Secret token for webhook validation (optional but recommended). - /// - /// When set, Telegram will include this value in the - /// `X-Telegram-Bot-Api-Secret-Token` header of webhook requests. - /// The agent validates this header to ensure requests come from Telegram. - /// - /// Generate a secure random token (32+ characters recommended). - pub webhook_secret: Option, -} - -impl TelegramChannelConfig { - fn from_env() -> Result { - Ok(Self { - webhook_secret: optional_env("TELEGRAM_WEBHOOK_SECRET")?, - }) - } -} - #[derive(Debug, Clone)] pub struct CliConfig { pub enabled: bool, @@ -379,7 +354,6 @@ impl ChannelsConfig { enabled: cli_enabled, }, http, - telegram: TelegramChannelConfig::from_env()?, wasm_channels_dir: optional_env("WASM_CHANNELS_DIR")? .map(PathBuf::from) .unwrap_or_else(default_channels_dir), diff --git a/src/main.rs b/src/main.rs index 532d80e9..f40e39fc 100644 --- a/src/main.rs +++ b/src/main.rs @@ -364,13 +364,15 @@ async fn main() -> anyhow::Result<()> { let wasm_router = Arc::new(WasmChannelRouter::new()); let mut has_webhook_channels = false; - for channel in results.loaded { - let channel_name = channel.channel_name().to_string(); + for loaded in results.loaded { + 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 { - let secret_name = format!("{}_webhook_secret", channel_name); secrets .get_decrypted("default", &secret_name) .await @@ -380,6 +382,10 @@ 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); @@ -390,82 +396,50 @@ async fn main() -> anyhow::Result<()> { require_secret: webhook_secret.is_some(), }]; - let channel_arc = Arc::new(channel); - - // Clone webhook_secret before moving it to register() - // We need it later for Telegram API registration - let webhook_secret_for_telegram = webhook_secret.clone(); + let channel_arc = Arc::new(loaded.channel); tracing::info!( channel = %channel_name, has_webhook_secret = webhook_secret.is_some(), + secret_header = ?secret_header, "Registering channel with router" ); wasm_router - .register(Arc::clone(&channel_arc), endpoints, webhook_secret) + .register( + Arc::clone(&channel_arc), + endpoints, + webhook_secret, + secret_header, + ) .await; has_webhook_channels = true; - // Set up Telegram channel credentials and optionally register webhook - if channel_name == "telegram" { - if let Some(ref secrets) = secrets_store { - // Inject bot token for HTTP request URL substitution - // This is needed for both webhook and polling modes - match inject_telegram_credentials( - &channel_arc, - secrets.as_ref(), - ) - .await - { - Ok(()) => { - tracing::debug!("Telegram bot token injected"); - } - Err(e) => { - tracing::error!( - "Failed to inject Telegram credentials: {}", - e - ); - tracing::warn!( - "Telegram channel may not be able to send responses" + // Inject credentials for this channel (generic pattern-based injection) + if let Some(ref secrets) = secrets_store { + match inject_channel_credentials( + &channel_arc, + secrets.as_ref(), + &channel_name, + ) + .await + { + Ok(count) => { + if count > 0 { + tracing::info!( + channel = %channel_name, + credentials_injected = count, + "Channel credentials injected" ); } } - - // Register webhook if tunnel URL is configured - // Use the SAME webhook_secret that the router expects (from secrets store) - if let Some(ref tunnel_url) = config.tunnel.public_url { - match register_telegram_webhook( - &channel_arc, - tunnel_url, - webhook_secret_for_telegram.as_deref(), - ) - .await - { - Ok(()) => { - tracing::info!( - "Telegram webhook registered at {}/webhook/telegram", - tunnel_url - ); - } - Err(e) => { - tracing::error!( - "Failed to register Telegram webhook: {}", - e - ); - tracing::warn!( - "Telegram will fall back to polling mode" - ); - } - } + Err(e) => { + tracing::error!( + channel = %channel_name, + error = %e, + "Failed to inject channel credentials" + ); } - } else { - tracing::warn!( - "Telegram channel loaded but secrets store not available" - ); - tracing::warn!( - "Set SECRETS_MASTER_KEY to enable Telegram bot token injection" - ); } } @@ -559,66 +533,60 @@ async fn main() -> anyhow::Result<()> { Ok(()) } -/// Inject Telegram bot token into the channel's credentials. +/// Inject credentials for a channel based on naming convention. /// -/// This allows the WASM channel to use `{TELEGRAM_BOT_TOKEN}` in HTTP URLs -/// without ever seeing the actual token value. Required for both webhook -/// and polling modes to send responses. -async fn inject_telegram_credentials( +/// Looks for secrets matching the pattern `{channel_name}_*` and injects them +/// as credential placeholders (e.g., `telegram_bot_token` -> `{TELEGRAM_BOT_TOKEN}`). +/// +/// Returns the number of credentials injected. +async fn inject_channel_credentials( channel: &Arc, secrets: &dyn SecretsStore, -) -> anyhow::Result<()> { - tracing::info!("Injecting Telegram bot token into channel credentials"); - - // Get bot token from secrets - let decrypted = secrets - .get_decrypted("default", "telegram_bot_token") + channel_name: &str, +) -> anyhow::Result { + // List all secrets for this user and filter by channel prefix + let all_secrets = secrets + .list("default") .await - .map_err(|e| { - tracing::error!(error = %e, "Failed to get telegram_bot_token from secrets"); - anyhow::anyhow!("Failed to get Telegram bot token: {}", e) - })?; + .map_err(|e| anyhow::anyhow!("Failed to list secrets: {}", e))?; - let bot_token = decrypted.expose(); - let token_len = bot_token.len(); + let prefix = format!("{}_", channel_name); + let mut count = 0; - // Inject the token into the channel's credentials for URL substitution - channel - .set_credential("TELEGRAM_BOT_TOKEN", bot_token.to_string()) - .await; + for secret_meta in all_secrets { + // Only process secrets matching the channel prefix + if !secret_meta.name.starts_with(&prefix) { + continue; + } - // Verify injection - let creds = channel.get_credentials().await; - tracing::info!( - token_length = token_len, - has_token = creds.contains_key("TELEGRAM_BOT_TOKEN"), - credential_count = creds.len(), - "Telegram bot token injected successfully" - ); + // Get the decrypted value + let decrypted = match secrets.get_decrypted("default", &secret_meta.name).await { + Ok(d) => d, + Err(e) => { + tracing::warn!( + secret = %secret_meta.name, + error = %e, + "Failed to decrypt secret for channel credential injection" + ); + continue; + } + }; - Ok(()) -} + // Convert secret name to placeholder format (SCREAMING_SNAKE_CASE) + let placeholder = secret_meta.name.to_uppercase(); -/// Register Telegram webhook for instant message delivery. -/// -/// Calls the Telegram setWebhook API. Assumes credentials have already been -/// injected via `inject_telegram_credentials` (gets token from channel). -async fn register_telegram_webhook( - channel: &Arc, - tunnel_url: &str, - webhook_secret: Option<&str>, -) -> anyhow::Result<()> { - // Get the bot token via the public getter - let credentials = channel.get_credentials().await; - let bot_token = credentials.get("TELEGRAM_BOT_TOKEN").ok_or_else(|| { - anyhow::anyhow!("Bot token not injected - call inject_telegram_credentials first") - })?; + tracing::debug!( + channel = %channel_name, + secret = %secret_meta.name, + placeholder = %placeholder, + "Injecting credential" + ); - // Register the webhook with Telegram API - channel - .register_telegram_webhook(tunnel_url, bot_token, webhook_secret) - .await - .map_err(|e| anyhow::anyhow!("Webhook registration failed: {}", e))?; + channel + .set_credential(&placeholder, decrypted.expose().to_string()) + .await; + count += 1; + } - Ok(()) + Ok(count) } diff --git a/src/settings.rs b/src/settings.rs index c52f1095..de2aea02 100644 --- a/src/settings.rs +++ b/src/settings.rs @@ -47,13 +47,11 @@ pub struct ChannelSettings { #[serde(default)] pub http_port: Option, - /// Whether Telegram channel is enabled. + /// Enabled WASM channels by name. + /// Channels not in this list but present in the channels directory will still load. + /// This is primarily used by the setup wizard to track which channels were configured. #[serde(default)] - pub telegram_enabled: bool, - - /// Whether Slack channel is enabled. - #[serde(default)] - pub slack_enabled: bool, + pub wasm_channels: Vec, } impl Settings { diff --git a/src/setup/channels.rs b/src/setup/channels.rs index b91d582d..f99081a9 100644 --- a/src/setup/channels.rs +++ b/src/setup/channels.rs @@ -330,7 +330,7 @@ pub async fn setup_http(secrets: &SecretsContext) -> Result String { +pub fn generate_webhook_secret() -> String { use rand::RngCore; let mut rng = rand::thread_rng(); let mut bytes = [0u8; 32]; @@ -338,6 +338,120 @@ fn generate_webhook_secret() -> String { bytes.iter().map(|b| format!("{:02x}", b)).collect() } +/// Result of WASM channel setup. +#[derive(Debug, Clone)] +pub struct WasmChannelSetupResult { + pub enabled: bool, + pub channel_name: String, +} + +/// Set up a WASM channel using its capabilities file setup schema. +/// +/// Reads setup requirements from the channel's capabilities file and +/// prompts the user for each required secret. +pub async fn setup_wasm_channel( + secrets: &SecretsContext, + channel_name: &str, + setup: &crate::channels::wasm::SetupSchema, +) -> Result { + println!("{} Setup:", channel_name); + println!(); + + for secret_config in &setup.required_secrets { + // Check if this secret already exists + if secrets.secret_exists(&secret_config.name).await { + print_info(&format!( + "Existing {} found in database.", + secret_config.name + )); + if !confirm("Replace existing value?", false).map_err(|e| e.to_string())? { + continue; + } + } + + // Get the value from user or auto-generate + let value = if secret_config.optional { + let input_value = + optional_input(&secret_config.prompt, Some("leave empty to auto-generate")) + .map_err(|e| e.to_string())?; + + if let Some(v) = input_value { + if !v.is_empty() { + SecretString::from(v) + } else if let Some(ref auto_gen) = secret_config.auto_generate { + let generated = generate_secret_with_length(auto_gen.length); + print_info(&format!( + "Auto-generated {} ({} bytes)", + secret_config.name, auto_gen.length + )); + SecretString::from(generated) + } else { + continue; // Skip optional secret with no auto-generate + } + } else if let Some(ref auto_gen) = secret_config.auto_generate { + let generated = generate_secret_with_length(auto_gen.length); + print_info(&format!( + "Auto-generated {} ({} bytes)", + secret_config.name, auto_gen.length + )); + SecretString::from(generated) + } else { + continue; // Skip optional secret with no auto-generate + } + } else { + // Required secret + let input_value = secret_input(&secret_config.prompt).map_err(|e| e.to_string())?; + + // Validate if pattern is provided + if let Some(ref pattern) = secret_config.validation { + let re = regex::Regex::new(pattern) + .map_err(|e| format!("Invalid validation pattern: {}", e))?; + if !re.is_match(input_value.expose_secret()) { + print_error(&format!( + "Value does not match expected format: {}", + pattern + )); + return Err("Validation failed".to_string()); + } + } + + input_value + }; + + // Save the secret + secrets.save_secret(&secret_config.name, &value).await?; + print_success(&format!("{} saved to database", secret_config.name)); + } + + // Optionally validate the configuration + if let Some(ref validation_endpoint) = setup.validation_endpoint { + print_info("Validating configuration..."); + // The validation endpoint may contain placeholders like {telegram_bot_token} + // For now, we skip validation since we'd need to substitute secrets + // A full implementation would fetch secrets and substitute them + print_info(&format!( + "Validation endpoint configured: {} (validation skipped)", + validation_endpoint + )); + } + + print_success(&format!("{} channel configured", channel_name)); + + Ok(WasmChannelSetupResult { + enabled: true, + channel_name: channel_name.to_string(), + }) +} + +/// Generate a random secret of specified length (in bytes). +fn generate_secret_with_length(length: usize) -> String { + use rand::RngCore; + let mut rng = rand::thread_rng(); + let mut bytes = vec![0u8; length]; + rng.fill_bytes(&mut bytes); + bytes.iter().map(|b| format!("{:02x}", b)).collect() +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/setup/wizard.rs b/src/setup/wizard.rs index bfa8a409..d63cf741 100644 --- a/src/setup/wizard.rs +++ b/src/setup/wizard.rs @@ -11,10 +11,13 @@ use deadpool_postgres::{Config as PoolConfig, Runtime}; use secrecy::SecretString; use tokio_postgres::NoTls; +use crate::channels::wasm::ChannelCapabilitiesFile; use crate::llm::{SessionConfig, SessionManager}; use crate::secrets::SecretsCrypto; use crate::settings::Settings; -use crate::setup::channels::{SecretsContext, setup_http, setup_telegram, setup_tunnel}; +use crate::setup::channels::{ + SecretsContext, setup_http, setup_telegram, setup_tunnel, setup_wasm_channel, +}; use crate::setup::prompts::{ input, print_header, print_info, print_step, print_success, select_many, select_one, }; @@ -330,16 +333,36 @@ impl SetupWizard { } println!(); - let options = [ - ("CLI/TUI (always enabled)", true), - ("HTTP webhook", self.settings.channels.http_enabled), - ("Telegram", self.settings.channels.telegram_enabled), + // Discover available WASM channels + let channels_dir = dirs::home_dir() + .unwrap_or_default() + .join(".near-agent/channels"); + + let discovered_channels = discover_wasm_channels(&channels_dir).await; + + // Build options list dynamically + let mut options: Vec<(String, bool)> = vec![ + ("CLI/TUI (always enabled)".to_string(), true), + ( + "HTTP webhook".to_string(), + self.settings.channels.http_enabled, + ), ]; - let selected = select_many("Which channels do you want to enable?", &options)?; + // Add discovered WASM channels + for (name, _) in &discovered_channels { + let is_enabled = self.settings.channels.wasm_channels.contains(name); + let display_name = format!("{} (WASM)", capitalize_first(name)); + options.push((display_name, is_enabled)); + } - // Only initialize secrets context if we need it (HTTP or Telegram selected) - let needs_secrets = selected.contains(&1) || selected.contains(&2); + let options_refs: Vec<(&str, bool)> = + options.iter().map(|(s, b)| (s.as_str(), *b)).collect(); + + let selected = select_many("Which channels do you want to enable?", &options_refs)?; + + // Determine if we need secrets context + let needs_secrets = selected.iter().any(|&i| i >= 1); let secrets = if needs_secrets { Some(self.init_secrets_context().await?) } else { @@ -358,16 +381,47 @@ impl SetupWizard { self.settings.channels.http_enabled = false; } - // Telegram is index 2 - if selected.contains(&2) { - println!(); - if let Some(ref ctx) = secrets { - let result = setup_telegram(ctx).await.map_err(SetupError::Channel)?; - self.settings.channels.telegram_enabled = result.enabled; + // Process WASM channels (index 2 and above) + let mut enabled_wasm_channels = Vec::new(); + for (idx, (channel_name, cap_file)) in discovered_channels.iter().enumerate() { + let option_idx = idx + 2; // Offset for CLI and HTTP + + if selected.contains(&option_idx) { + println!(); + if let Some(ref ctx) = secrets { + // Use setup schema from capabilities if available + let result = if !cap_file.setup.required_secrets.is_empty() { + setup_wasm_channel(ctx, channel_name, &cap_file.setup) + .await + .map_err(SetupError::Channel)? + } else { + // Fall back to legacy Telegram setup for backwards compatibility + if channel_name == "telegram" { + let telegram_result = + setup_telegram(ctx).await.map_err(SetupError::Channel)?; + crate::setup::channels::WasmChannelSetupResult { + enabled: telegram_result.enabled, + channel_name: "telegram".to_string(), + } + } else { + print_info(&format!( + "No setup configuration found for {}", + channel_name + )); + crate::setup::channels::WasmChannelSetupResult { + enabled: true, + channel_name: channel_name.to_string(), + } + } + }; + + if result.enabled { + enabled_wasm_channels.push(result.channel_name); + } + } } - } else { - self.settings.channels.telegram_enabled = false; } + self.settings.channels.wasm_channels = enabled_wasm_channels; Ok(()) } @@ -407,13 +461,17 @@ impl SetupWizard { println!(" - HTTP: enabled (port {})", port); } - if self.settings.channels.telegram_enabled { + for channel_name in &self.settings.channels.wasm_channels { let mode = if self.settings.tunnel.public_url.is_some() { "webhook" } else { "polling" }; - println!(" - Telegram: enabled ({})", mode); + println!( + " - {}: enabled ({})", + capitalize_first(channel_name), + mode + ); } println!(); @@ -440,6 +498,81 @@ impl Default for SetupWizard { } } +/// Discover WASM channels in a directory. +/// +/// Returns a list of (channel_name, capabilities_file) pairs. +async fn discover_wasm_channels(dir: &std::path::Path) -> Vec<(String, ChannelCapabilitiesFile)> { + let mut channels = Vec::new(); + + if !dir.is_dir() { + return channels; + } + + let mut entries = match tokio::fs::read_dir(dir).await { + Ok(e) => e, + Err(_) => return channels, + }; + + while let Ok(Some(entry)) = entries.next_entry().await { + let path = entry.path(); + + // Look for .capabilities.json files + let extension = path.file_name().and_then(|n| n.to_str()).unwrap_or(""); + + if !extension.ends_with(".capabilities.json") { + continue; + } + + // Extract channel name + let name = extension.trim_end_matches(".capabilities.json").to_string(); + if name.is_empty() { + continue; + } + + // Check if corresponding .wasm file exists + let wasm_path = dir.join(format!("{}.wasm", name)); + if !wasm_path.exists() { + continue; + } + + // Parse capabilities file + match tokio::fs::read(&path).await { + Ok(bytes) => match ChannelCapabilitiesFile::from_bytes(&bytes) { + Ok(cap_file) => { + channels.push((name, cap_file)); + } + Err(e) => { + tracing::warn!( + path = %path.display(), + error = %e, + "Failed to parse channel capabilities file" + ); + } + }, + Err(e) => { + tracing::warn!( + path = %path.display(), + error = %e, + "Failed to read channel capabilities file" + ); + } + } + } + + // Sort by name for consistent ordering + channels.sort_by(|a, b| a.0.cmp(&b.0)); + channels +} + +/// Capitalize the first letter of a string. +fn capitalize_first(s: &str) -> String { + let mut chars = s.chars(); + match chars.next() { + None => String::new(), + Some(first) => first.to_uppercase().chain(chars).collect(), + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/tests/wasm_channel_integration.rs b/tests/wasm_channel_integration.rs index 3b2241bf..388ccc52 100644 --- a/tests/wasm_channel_integration.rs +++ b/tests/wasm_channel_integration.rs @@ -62,7 +62,9 @@ mod router_tests { require_secret: false, }]; - router.register(channel.clone(), endpoints, None).await; + router + .register(channel.clone(), endpoints, None, None) + .await; // Verify channel is found by path let found = router.get_channel_for_path("/webhook/test").await; @@ -86,7 +88,7 @@ mod router_tests { )); router - .register(channel, vec![], Some("my-secret-123".to_string())) + .register(channel, vec![], Some("my-secret-123".to_string()), None) .await; // Correct secret validates @@ -125,7 +127,7 @@ mod router_tests { require_secret: false, }]; - router.register(channel, endpoints, None).await; + router.register(channel, endpoints, None, None).await; // Channel exists assert!(router.get_channel_for_path("/webhook/temp").await.is_some()); @@ -157,7 +159,7 @@ mod router_tests { require_secret: false, }]; - router.register(channel, endpoints, None).await; + router.register(channel, endpoints, None, None).await; } // Verify all channels are registered