ironflow_engine/notify/
betterstack.rs

1//! [`BetterStackSubscriber`] -- forwards error events to BetterStack Logs.
2
3use reqwest::Client;
4use serde::Serialize;
5
6use super::retry::{RetryConfig, deliver_with_retry, is_accepted_202};
7use super::{Event, EventSubscriber, SubscriberFuture};
8
9/// Default BetterStack Logs ingestion endpoint.
10const DEFAULT_INGEST_URL: &str = "https://in.logs.betterstack.com";
11
12/// Payload sent to BetterStack Logs API.
13#[derive(Debug, Serialize)]
14struct LogPayload {
15    /// ISO-8601 timestamp.
16    dt: String,
17    /// Log level (always "error" for this subscriber).
18    level: &'static str,
19    /// Human-readable message.
20    message: String,
21    /// Structured event data.
22    event: serde_json::Value,
23}
24
25/// Subscriber that forwards error events to [BetterStack Logs](https://betterstack.com/docs/logs/).
26///
27/// Only acts on events that represent failures:
28/// - [`Event::StepFailed`]
29/// - [`Event::RunFailed`]
30///
31/// All other events are silently ignored (filtering by event type
32/// should already be done at subscription time, but this subscriber
33/// adds an extra safety check).
34///
35/// Retries failed deliveries with exponential backoff (up to 3 attempts,
36/// 5 s timeout per attempt).
37///
38/// # Examples
39///
40/// ```no_run
41/// use ironflow_engine::notify::{BetterStackSubscriber, Event, EventPublisher};
42///
43/// let mut publisher = EventPublisher::new();
44/// publisher.subscribe(
45///     BetterStackSubscriber::new("my-source-token"),
46///     &[Event::STEP_FAILED, Event::RUN_FAILED],
47/// );
48/// ```
49pub struct BetterStackSubscriber {
50    source_token: String,
51    authorization_header: String,
52    ingest_url: String,
53    client: Client,
54    retry_config: RetryConfig,
55}
56
57impl BetterStackSubscriber {
58    /// Create a new subscriber with the given BetterStack source token.
59    ///
60    /// Uses the default ingestion endpoint (`https://in.logs.betterstack.com`)
61    /// and default [`RetryConfig`].
62    ///
63    /// # Panics
64    ///
65    /// Panics if the HTTP client cannot be built (TLS backend unavailable).
66    ///
67    /// # Examples
68    ///
69    /// ```
70    /// use ironflow_engine::notify::BetterStackSubscriber;
71    ///
72    /// let subscriber = BetterStackSubscriber::new("my-source-token");
73    /// assert_eq!(subscriber.source_token(), "my-source-token");
74    /// ```
75    pub fn new(source_token: &str) -> Self {
76        Self::with_url(source_token, DEFAULT_INGEST_URL)
77    }
78
79    /// Create a subscriber with a custom ingestion URL.
80    ///
81    /// Useful for testing or self-hosted BetterStack instances.
82    ///
83    /// # Panics
84    ///
85    /// Panics if the HTTP client cannot be built (TLS backend unavailable).
86    ///
87    /// # Examples
88    ///
89    /// ```
90    /// use ironflow_engine::notify::BetterStackSubscriber;
91    ///
92    /// let subscriber = BetterStackSubscriber::with_url(
93    ///     "my-source-token",
94    ///     "https://custom.logs.example.com",
95    /// );
96    /// assert_eq!(subscriber.ingest_url(), "https://custom.logs.example.com");
97    /// ```
98    pub fn with_url(source_token: &str, ingest_url: &str) -> Self {
99        Self::with_url_and_retry(source_token, ingest_url, RetryConfig::default())
100    }
101
102    /// Create a subscriber with a custom ingestion URL and retry configuration.
103    ///
104    /// # Panics
105    ///
106    /// Panics if the HTTP client cannot be built (TLS backend unavailable).
107    ///
108    /// # Examples
109    ///
110    /// ```
111    /// use ironflow_engine::notify::{BetterStackSubscriber, RetryConfig};
112    ///
113    /// let config = RetryConfig::new(
114    ///     5,
115    ///     std::time::Duration::from_secs(10),
116    ///     std::time::Duration::from_secs(1),
117    /// );
118    /// let subscriber = BetterStackSubscriber::with_url_and_retry(
119    ///     "my-source-token",
120    ///     "https://custom.logs.example.com",
121    ///     config,
122    /// );
123    /// ```
124    pub fn with_url_and_retry(
125        source_token: &str,
126        ingest_url: &str,
127        retry_config: RetryConfig,
128    ) -> Self {
129        let client = retry_config.build_client();
130        Self {
131            authorization_header: format!("Bearer {}", source_token),
132            source_token: source_token.to_string(),
133            ingest_url: ingest_url.to_string(),
134            client,
135            retry_config,
136        }
137    }
138
139    /// Returns the source token.
140    pub fn source_token(&self) -> &str {
141        &self.source_token
142    }
143
144    /// Returns the ingestion URL.
145    pub fn ingest_url(&self) -> &str {
146        &self.ingest_url
147    }
148
149    /// Build a log payload from an error event. Returns `None` for non-error events.
150    fn build_payload(event: &Event) -> Option<LogPayload> {
151        match event {
152            Event::StepFailed {
153                run_id,
154                step_id,
155                step_name,
156                kind,
157                error,
158                at,
159            } => {
160                let message = format!(
161                    "Step '{}' ({}) failed on run {}: {}",
162                    step_name, kind, run_id, error
163                );
164                let event_json = serde_json::json!({
165                    "type": "step_failed",
166                    "run_id": run_id.to_string(),
167                    "step_id": step_id.to_string(),
168                    "step_name": step_name,
169                    "kind": kind.to_string(),
170                    "error": error,
171                });
172                Some(LogPayload {
173                    dt: at.to_rfc3339(),
174                    level: "error",
175                    message,
176                    event: event_json,
177                })
178            }
179            Event::RunFailed {
180                run_id,
181                workflow_name,
182                error,
183                cost_usd,
184                duration_ms,
185                at,
186                ..
187            } => {
188                let error_detail = error.as_deref().unwrap_or("unknown error");
189                let message = format!(
190                    "Run {} (workflow '{}') failed: {}",
191                    run_id, workflow_name, error_detail
192                );
193                let event_json = serde_json::json!({
194                    "type": "run_failed",
195                    "run_id": run_id.to_string(),
196                    "workflow_name": workflow_name,
197                    "error": error_detail,
198                    "cost_usd": cost_usd.to_string(),
199                    "duration_ms": duration_ms,
200                });
201                Some(LogPayload {
202                    dt: at.to_rfc3339(),
203                    level: "error",
204                    message,
205                    event: event_json,
206                })
207            }
208            _ => None,
209        }
210    }
211}
212
213impl EventSubscriber for BetterStackSubscriber {
214    fn name(&self) -> &str {
215        "betterstack"
216    }
217
218    fn handle<'a>(&'a self, event: &'a Event) -> SubscriberFuture<'a> {
219        Box::pin(async move {
220            if let Some(payload) = Self::build_payload(event) {
221                deliver_with_retry(
222                    &self.retry_config,
223                    || {
224                        self.client
225                            .post(&self.ingest_url)
226                            .header("Authorization", &self.authorization_header)
227                            .json(&payload)
228                    },
229                    is_accepted_202,
230                    "betterstack",
231                    &payload.message,
232                )
233                .await;
234            }
235        })
236    }
237}
238
239#[cfg(test)]
240mod tests {
241    use std::collections::HashMap;
242
243    use super::*;
244    use chrono::Utc;
245    use ironflow_store::models::{RunStatus, StepKind};
246    use rust_decimal::Decimal;
247    use uuid::Uuid;
248
249    #[test]
250    fn new_sets_default_ingest_url() {
251        let sub = BetterStackSubscriber::new("token-123");
252        assert_eq!(sub.source_token(), "token-123");
253        assert_eq!(sub.ingest_url(), DEFAULT_INGEST_URL);
254    }
255
256    #[test]
257    fn with_url_sets_custom_ingest_url() {
258        let sub = BetterStackSubscriber::with_url("token-123", "https://custom.example.com");
259        assert_eq!(sub.source_token(), "token-123");
260        assert_eq!(sub.ingest_url(), "https://custom.example.com");
261    }
262
263    #[test]
264    fn name_is_betterstack() {
265        let sub = BetterStackSubscriber::new("token");
266        assert_eq!(sub.name(), "betterstack");
267    }
268
269    #[test]
270    fn build_payload_step_failed() {
271        let event = Event::StepFailed {
272            run_id: Uuid::now_v7(),
273            step_id: Uuid::now_v7(),
274            step_name: "build".to_string(),
275            kind: StepKind::Shell,
276            error: "exit code 1".to_string(),
277            at: Utc::now(),
278        };
279
280        let payload = BetterStackSubscriber::build_payload(&event);
281        assert!(payload.is_some());
282        let payload = payload.unwrap();
283        assert_eq!(payload.level, "error");
284        assert!(payload.message.contains("build"));
285        assert!(payload.message.contains("exit code 1"));
286        assert_eq!(payload.event["type"], "step_failed");
287        assert_eq!(payload.event["error"], "exit code 1");
288    }
289
290    #[test]
291    fn build_payload_run_failed() {
292        let event = Event::RunFailed {
293            run_id: Uuid::now_v7(),
294            workflow_name: "deploy".to_string(),
295            error: Some("step 'build' failed".to_string()),
296            cost_usd: Decimal::new(42, 2),
297            duration_ms: 5000,
298            labels: HashMap::new(),
299            at: Utc::now(),
300        };
301
302        let payload = BetterStackSubscriber::build_payload(&event);
303        assert!(payload.is_some());
304        let payload = payload.unwrap();
305        assert_eq!(payload.level, "error");
306        assert!(payload.message.contains("deploy"));
307        assert!(payload.message.contains("step 'build' failed"));
308        assert_eq!(payload.event["type"], "run_failed");
309        assert_eq!(payload.event["workflow_name"], "deploy");
310    }
311
312    #[test]
313    fn build_payload_run_failed_without_error_message() {
314        let event = Event::RunFailed {
315            run_id: Uuid::now_v7(),
316            workflow_name: "deploy".to_string(),
317            error: None,
318            cost_usd: Decimal::ZERO,
319            duration_ms: 1000,
320            labels: HashMap::new(),
321            at: Utc::now(),
322        };
323
324        let payload = BetterStackSubscriber::build_payload(&event).unwrap();
325        assert!(payload.message.contains("unknown error"));
326        assert_eq!(payload.event["error"], "unknown error");
327    }
328
329    #[test]
330    fn build_payload_run_completed_returns_none() {
331        let event = Event::RunStatusChanged {
332            run_id: Uuid::now_v7(),
333            workflow_name: "deploy".to_string(),
334            from: RunStatus::Running,
335            to: RunStatus::Completed,
336            error: None,
337            cost_usd: Decimal::ZERO,
338            duration_ms: 1000,
339            labels: HashMap::new(),
340            at: Utc::now(),
341        };
342
343        assert!(BetterStackSubscriber::build_payload(&event).is_none());
344    }
345
346    #[test]
347    fn build_payload_run_created_returns_none() {
348        let event = Event::RunCreated {
349            run_id: Uuid::now_v7(),
350            workflow_name: "deploy".to_string(),
351            at: Utc::now(),
352        };
353
354        assert!(BetterStackSubscriber::build_payload(&event).is_none());
355    }
356
357    #[test]
358    fn build_payload_step_completed_returns_none() {
359        let event = Event::StepCompleted {
360            run_id: Uuid::now_v7(),
361            step_id: Uuid::now_v7(),
362            step_name: "build".to_string(),
363            kind: StepKind::Shell,
364            duration_ms: 500,
365            cost_usd: Decimal::ZERO,
366            at: Utc::now(),
367        };
368
369        assert!(BetterStackSubscriber::build_payload(&event).is_none());
370    }
371
372    #[test]
373    fn build_payload_approval_requested_returns_none() {
374        let event = Event::ApprovalRequested {
375            run_id: Uuid::now_v7(),
376            step_id: Uuid::now_v7(),
377            message: "Deploy to prod?".to_string(),
378            at: Utc::now(),
379        };
380
381        assert!(BetterStackSubscriber::build_payload(&event).is_none());
382    }
383
384    #[test]
385    fn build_payload_user_signed_in_returns_none() {
386        let event = Event::UserSignedIn {
387            user_id: Uuid::now_v7(),
388            username: "alice".to_string(),
389            at: Utc::now(),
390        };
391
392        assert!(BetterStackSubscriber::build_payload(&event).is_none());
393    }
394
395    #[tokio::test]
396    async fn handle_ignores_non_error_events() {
397        let sub = BetterStackSubscriber::with_url("token", "http://127.0.0.1:1");
398        let event = Event::RunCreated {
399            run_id: Uuid::now_v7(),
400            workflow_name: "deploy".to_string(),
401            at: Utc::now(),
402        };
403        // Should return immediately without attempting HTTP
404        sub.handle(&event).await;
405    }
406
407    #[tokio::test]
408    async fn deliver_to_real_endpoint_returns_202() {
409        use axum::Router;
410        use axum::http::StatusCode;
411        use axum::routing::post;
412        use tokio::net::TcpListener;
413
414        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
415        let addr = listener.local_addr().unwrap();
416
417        let app = Router::new().route("/", post(|| async { StatusCode::ACCEPTED }));
418
419        tokio::spawn(async move {
420            axum::serve(listener, app).await.unwrap();
421        });
422
423        let sub = BetterStackSubscriber::with_url("test-token", &format!("http://{}", addr));
424        let event = Event::StepFailed {
425            run_id: Uuid::now_v7(),
426            step_id: Uuid::now_v7(),
427            step_name: "build".to_string(),
428            kind: StepKind::Shell,
429            error: "exit code 1".to_string(),
430            at: Utc::now(),
431        };
432
433        sub.handle(&event).await;
434    }
435}
ironflow_engine/notify/betterstack.rs

ironflow_engine/notify/
betterstack.rs
