brainos_backends/
messaging.rs

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

brainos_backends/
messaging.rs
