brainos_backends/
messaging.rs

1//! Message backend — webhook delivery with resilience.
2
3use std::collections::HashMap;
4use std::sync::Arc;
5
6use metrics::SubsystemMetrics;
7
8use crate::resilience::{http_breaker, resilient_send, CircuitBreaker};
9
10/// The capability this backend declares: outbound messaging through the channel
11/// dispatcher, gated on `actions.messaging.enabled`.
12pub fn capabilities(config: &brain::BrainConfig) -> Vec<intent::ToolDescriptor> {
13    use crate::capabilities::{backend, native, usage};
14    use intent::ToolAnnotations;
15    if !config.actions.messaging.enabled {
16        return Vec::new();
17    }
18    vec![native(
19        "notify",
20        "send",
21        backend("messaging"),
22        ToolAnnotations::default(),
23        usage(
24            "The user asks to send a message/notification out through a configured channel.",
25            "For replying inside the current conversation — just answer.",
26            &[
27                "actions.messaging.enabled = true",
28                "A channel transport is configured.",
29            ],
30            "network call (external delivery)",
31            "\"Ping the ops webhook when the job finishes\"",
32        ),
33    )]
34}
35
36pub const DEFAULT_MESSAGE_BODY: &str = r#"{"channel":"{{channel}}","recipient":"{{recipient}}","content":"{{content}}","namespace":"{{namespace}}","timestamp":"{{timestamp}}"}"#;
37
38/// JSON-escape a string value (without surrounding quotes).
39pub fn json_escape(s: &str) -> String {
40    let escaped = serde_json::to_string(s).unwrap_or_else(|_| format!("\"{}\"", s));
41    escaped[1..escaped.len() - 1].to_string()
42}
43
44/// Render a message template by replacing `{{placeholder}}` tokens.
45pub fn render_message_template(
46    template: &str,
47    channel: &str,
48    recipient: &str,
49    content: &str,
50    namespace: &str,
51    timestamp: &str,
52) -> String {
53    template
54        .replace("{{channel}}", &json_escape(channel))
55        .replace("{{recipient}}", &json_escape(recipient))
56        .replace("{{content}}", &json_escape(content))
57        .replace("{{namespace}}", &json_escape(namespace))
58        .replace("{{timestamp}}", &json_escape(timestamp))
59}
60
61pub struct WebhookMessageBackend {
62    channels: HashMap<String, brain::config::ChannelConfig>,
63    client: reqwest::Client,
64    circuit_breaker: Arc<CircuitBreaker>,
65    max_retries: u32,
66    retry_base_ms: u64,
67}
68
69impl WebhookMessageBackend {
70    pub fn new(
71        channels: &HashMap<String, brain::config::ChannelConfig>,
72        timeout_ms: u64,
73        resilience: &brain::config::ResilienceConfig,
74    ) -> Result<Self, crate::error::BackendInitError> {
75        Self::new_with_metrics(channels, timeout_ms, resilience, None)
76    }
77
78    pub fn new_with_metrics(
79        channels: &HashMap<String, brain::config::ChannelConfig>,
80        timeout_ms: u64,
81        resilience: &brain::config::ResilienceConfig,
82        metrics: Option<Arc<SubsystemMetrics>>,
83    ) -> Result<Self, crate::error::BackendInitError> {
84        let client = reqwest::Client::builder()
85            .timeout(std::time::Duration::from_millis(timeout_ms.max(1)))
86            .build()
87            .map_err(|e| crate::error::BackendInitError::HttpClient("message client", e))?;
88        let cb = http_breaker(
89            "webhook-message",
90            resilience.circuit_breaker_threshold,
91            resilience.circuit_breaker_cooldown_secs,
92            metrics,
93        );
94        Ok(Self {
95            channels: channels
96                .iter()
97                .map(|(k, v)| (k.to_ascii_lowercase(), v.clone()))
98                .collect(),
99            client,
100            circuit_breaker: Arc::new(cb),
101            max_retries: resilience.max_retries,
102            retry_base_ms: resilience.retry_base_ms,
103        })
104    }
105}
106
107#[async_trait::async_trait]
108impl signal::notification::WebhookSender for WebhookMessageBackend {
109    async fn send_notification(
110        &self,
111        channel: &str,
112        content: &str,
113        namespace: &str,
114    ) -> Result<(), String> {
115        let channel_cfg = self
116            .channels
117            .get(&channel.to_ascii_lowercase())
118            .ok_or_else(|| format!("No webhook mapping for channel '{channel}'"))?
119            .clone();
120
121        let client = self.client.clone();
122        let template = if channel_cfg.body.is_empty() {
123            DEFAULT_MESSAGE_BODY.to_string()
124        } else {
125            channel_cfg.body.clone()
126        };
127        let url = channel_cfg.url.clone();
128        let headers = channel_cfg.headers.clone();
129        let channel_owned = channel.to_string();
130        let content_owned = content.to_string();
131        let namespace_owned = namespace.to_string();
132
133        let response = resilient_send(
134            || {
135                let timestamp = chrono::Utc::now().to_rfc3339();
136                let rendered = render_message_template(
137                    &template,
138                    &channel_owned,
139                    "",
140                    &content_owned,
141                    &namespace_owned,
142                    &timestamp,
143                );
144                let mut req = client
145                    .post(&url)
146                    .header("Content-Type", "application/json")
147                    .body(rendered);
148                for (key, value) in &headers {
149                    req = req.header(key.as_str(), value.as_str());
150                }
151                req
152            },
153            &self.circuit_breaker,
154            self.max_retries,
155            self.retry_base_ms,
156        )
157        .await
158        .map_err(|e| format!("webhook send failed: {e}"))?;
159
160        if !response.status().is_success() {
161            return Err(format!(
162                "webhook for channel '{}' returned HTTP {}",
163                channel,
164                response.status()
165            ));
166        }
167        Ok(())
168    }
169}
170
171#[async_trait::async_trait]
172impl cortex::actions::MessageBackend for WebhookMessageBackend {
173    async fn send(
174        &self,
175        channel: &str,
176        recipient: &str,
177        content: &str,
178        namespace: &str,
179    ) -> Result<cortex::actions::MessageOutcome, cortex::actions::ActionError> {
180        let channel_cfg = self
181            .channels
182            .get(&channel.to_ascii_lowercase())
183            .ok_or_else(|| {
184                cortex::actions::ActionError::InvalidArguments(format!(
185                    "No webhook mapping for channel '{}'",
186                    channel
187                ))
188            })?
189            .clone();
190
191        let client = self.client.clone();
192        let template = if channel_cfg.body.is_empty() {
193            DEFAULT_MESSAGE_BODY.to_string()
194        } else {
195            channel_cfg.body.clone()
196        };
197        let url = channel_cfg.url.clone();
198        let headers = channel_cfg.headers.clone();
199        let channel_owned = channel.to_string();
200        let recipient_owned = recipient.to_string();
201        let content_owned = content.to_string();
202        let namespace_owned = namespace.to_string();
203
204        let response = resilient_send(
205            || {
206                let timestamp = chrono::Utc::now().to_rfc3339();
207                let rendered = render_message_template(
208                    &template,
209                    &channel_owned,
210                    &recipient_owned,
211                    &content_owned,
212                    &namespace_owned,
213                    &timestamp,
214                );
215                let mut req = client
216                    .post(&url)
217                    .header("Content-Type", "application/json")
218                    .body(rendered);
219                for (key, value) in &headers {
220                    req = req.header(key.as_str(), value.as_str());
221                }
222                req
223            },
224            &self.circuit_breaker,
225            self.max_retries,
226            self.retry_base_ms,
227        )
228        .await
229        .map_err(|e| cortex::actions::ActionError::ExecutionFailed(e.to_string()))?;
230
231        if !response.status().is_success() {
232            return Err(cortex::actions::ActionError::ExecutionFailed(format!(
233                "webhook for channel '{}' returned HTTP {}",
234                channel,
235                response.status()
236            )));
237        }
238
239        let body = response.text().await.unwrap_or_default();
240        let mut delivery_id = format!("msg-{}", chrono::Utc::now().timestamp_micros());
241        let mut status = "accepted".to_string();
242        if let Ok(value) = serde_json::from_str::<serde_json::Value>(&body) {
243            if let Some(id) = value
244                .get("id")
245                .or_else(|| value.get("delivery_id"))
246                .and_then(serde_json::Value::as_str)
247            {
248                delivery_id = id.to_string();
249            }
250            if let Some(s) = value.get("status").and_then(serde_json::Value::as_str) {
251                status = s.to_string();
252            }
253        }
254
255        Ok(cortex::actions::MessageOutcome {
256            delivery_id,
257            status,
258        })
259    }
260}
261
262#[cfg(test)]
263mod tests {
264    use super::*;
265
266    #[test]
267    fn test_render_message_template_default() {
268        let rendered = render_message_template(
269            DEFAULT_MESSAGE_BODY,
270            "alerts",
271            "alice",
272            "deploy done",
273            "work",
274            "2026-03-08T12:00:00Z",
275        );
276        let parsed: serde_json::Value =
277            serde_json::from_str(&rendered).expect("default template should produce valid JSON");
278        assert_eq!(parsed["channel"], "alerts");
279        assert_eq!(parsed["recipient"], "alice");
280        assert_eq!(parsed["content"], "deploy done");
281        assert_eq!(parsed["namespace"], "work");
282        assert_eq!(parsed["timestamp"], "2026-03-08T12:00:00Z");
283    }
284
285    #[test]
286    fn test_render_message_template_custom_template() {
287        let template = r#"{"text": "[{{channel}}] {{content}}"}"#;
288        let rendered = render_message_template(
289            template,
290            "ops",
291            "bob",
292            "server is down",
293            "personal",
294            "2026-03-08T12:00:00Z",
295        );
296        let parsed: serde_json::Value =
297            serde_json::from_str(&rendered).expect("custom template should produce valid JSON");
298        assert_eq!(parsed["text"], "[ops] server is down");
299    }
300
301    #[test]
302    fn test_render_message_template_escapes_quotes() {
303        let rendered = render_message_template(
304            DEFAULT_MESSAGE_BODY,
305            "alerts",
306            "alice",
307            r#"He said "hello""#,
308            "work",
309            "2026-03-08T12:00:00Z",
310        );
311        let parsed: serde_json::Value =
312            serde_json::from_str(&rendered).expect("escaped content should produce valid JSON");
313        assert_eq!(parsed["content"], r#"He said "hello""#);
314    }
315
316    #[test]
317    fn test_render_message_template_escapes_newlines() {
318        let rendered = render_message_template(
319            DEFAULT_MESSAGE_BODY,
320            "alerts",
321            "alice",
322            "line1\nline2",
323            "work",
324            "2026-03-08T12:00:00Z",
325        );
326        let parsed: serde_json::Value =
327            serde_json::from_str(&rendered).expect("newline content should produce valid JSON");
328        assert_eq!(parsed["content"], "line1\nline2");
329    }
330
331    #[test]
332    fn test_json_escape() {
333        assert_eq!(json_escape("hello"), "hello");
334        assert_eq!(json_escape(r#"say "hi""#), r#"say \"hi\""#);
335        assert_eq!(json_escape("a\nb"), r#"a\nb"#);
336        assert_eq!(json_escape("back\\slash"), r#"back\\slash"#);
337    }
338
339    // ─── Mock HTTP tests ────────────────────────────────────────────────────
340
341    fn fast_resilience() -> brain::config::ResilienceConfig {
342        brain::config::ResilienceConfig {
343            max_retries: 0,
344            retry_base_ms: 10,
345            circuit_breaker_threshold: 5,
346            circuit_breaker_cooldown_secs: 60,
347        }
348    }
349
350    fn build_channels(name: &str, url: &str) -> HashMap<String, brain::config::ChannelConfig> {
351        let mut map = HashMap::new();
352        map.insert(
353            name.to_string(),
354            brain::config::ChannelConfig {
355                url: url.to_string(),
356                body: String::new(),
357                headers: HashMap::new(),
358            },
359        );
360        map
361    }
362
363    #[tokio::test]
364    async fn test_webhook_send_success() {
365        use cortex::actions::MessageBackend;
366
367        let mut server = mockito::Server::new_async().await;
368        let mock = server
369            .mock("POST", "/hook")
370            .match_header("content-type", "application/json")
371            .with_status(200)
372            .with_header("content-type", "application/json")
373            .with_body(r#"{"id": "delivery-123", "status": "sent"}"#)
374            .create_async()
375            .await;
376
377        let webhook_url = format!("{}/hook", server.url());
378        let channels = build_channels("alerts", &webhook_url);
379        let backend = WebhookMessageBackend::new(&channels, 5000, &fast_resilience()).unwrap();
380        let result = backend
381            .send("alerts", "alice", "server is down", "work")
382            .await
383            .unwrap();
384        assert_eq!(result.delivery_id, "delivery-123");
385        assert_eq!(result.status, "sent");
386        mock.assert_async().await;
387    }
388
389    #[tokio::test]
390    async fn test_webhook_send_5xx_fails() {
391        use cortex::actions::MessageBackend;
392
393        let mut server = mockito::Server::new_async().await;
394        let _mock = server
395            .mock("POST", "/hook")
396            .with_status(500)
397            .with_body("server error")
398            .expect_at_least(1)
399            .create_async()
400            .await;
401
402        let webhook_url = format!("{}/hook", server.url());
403        let channels = build_channels("alerts", &webhook_url);
404        let backend = WebhookMessageBackend::new(&channels, 5000, &fast_resilience()).unwrap();
405        let result = backend.send("alerts", "alice", "boom", "work").await;
406        assert!(result.is_err(), "expected 5xx to fail");
407    }
408
409    #[tokio::test]
410    async fn test_webhook_unknown_channel_returns_invalid_args() {
411        use cortex::actions::MessageBackend;
412
413        let channels = build_channels("alerts", "http://unused");
414        let backend = WebhookMessageBackend::new(&channels, 5000, &fast_resilience()).unwrap();
415        let err = backend
416            .send("unknown-channel", "alice", "hi", "work")
417            .await
418            .unwrap_err();
419        assert!(
420            matches!(err, cortex::actions::ActionError::InvalidArguments(_)),
421            "expected InvalidArguments error, got {err:?}"
422        );
423    }
424
425    #[tokio::test]
426    async fn test_webhook_fallback_delivery_id_when_no_body_id() {
427        use cortex::actions::MessageBackend;
428
429        let mut server = mockito::Server::new_async().await;
430        let _mock = server
431            .mock("POST", "/hook")
432            .with_status(200)
433            .with_body(r#"{"other": "field"}"#)
434            .create_async()
435            .await;
436
437        let webhook_url = format!("{}/hook", server.url());
438        let channels = build_channels("alerts", &webhook_url);
439        let backend = WebhookMessageBackend::new(&channels, 5000, &fast_resilience()).unwrap();
440        let result = backend.send("alerts", "alice", "hi", "work").await.unwrap();
441        // Fallback ID is a generated timestamp-based token starting with "msg-"
442        assert!(result.delivery_id.starts_with("msg-"));
443        assert_eq!(result.status, "accepted");
444    }
445}
brainos_backends/messaging.rs

brainos_backends/
messaging.rs
