uptrakit_openapi_client/
batch_progress_stream.rs

1//! Typed SSE streaming method for batch progress.
2//!
3//! Provides [`UptrakitClient::stream_batch_progress`] which connects to the
4//! `GET /api/v1/update-batches/{id}/stream` endpoint and returns a typed
5//! stream of batch progress events.
6
7use crate::sse::{self, RawSseEvent, SseError};
8use crate::{ClientError, Result, UptrakitClient};
9use rootcause::prelude::*;
10use serde::Deserialize;
11use uuid::Uuid;
12
13/// A typed SSE event from the batch progress stream.
14#[derive(Debug, Clone)]
15pub enum BatchProgressEvent {
16    /// An individual update within the batch changed status.
17    Update(BatchUpdateEventData),
18    /// Overall batch progress summary.
19    Progress(BatchProgressData),
20    /// The batch has reached a terminal status.
21    BatchCompleted(BatchCompletedData),
22}
23
24/// Data for an individual update event within a batch.
25#[derive(Debug, Clone, Deserialize)]
26pub struct BatchUpdateEventData {
27    pub event: String,
28    pub update_history_id: Uuid,
29    pub software_item_name: String,
30    pub host_name: String,
31    #[serde(default)]
32    pub error: Option<String>,
33}
34
35/// Data for a batch progress summary event.
36#[derive(Debug, Clone, Deserialize)]
37pub struct BatchProgressData {
38    pub completed: i64,
39    pub failed: i64,
40    pub pending: i64,
41    pub total: i32,
42}
43
44/// Data for a batch completion event.
45#[derive(Debug, Clone, Deserialize)]
46pub struct BatchCompletedData {
47    pub status: String,
48}
49
50/// Errors specific to batch progress streaming.
51#[derive(Debug, thiserror::Error)]
52pub enum StreamError {
53    #[error("SSE transport error: {0}")]
54    Sse(#[from] SseError),
55
56    #[error("failed to parse SSE event data: {0}")]
57    Parse(#[from] serde_json::Error),
58}
59
60impl UptrakitClient {
61    /// Connect to the batch progress SSE stream and return a stream of typed events.
62    ///
63    /// The returned stream yields [`BatchProgressEvent`] values until the batch
64    /// completes (indicated by a `BatchCompleted` event) or the connection closes.
65    ///
66    /// This method uses no request timeout since SSE connections are long-lived.
67    pub async fn stream_batch_progress(
68        &self,
69        id: &Uuid,
70    ) -> Result<
71        impl futures_util::Stream<Item = std::result::Result<BatchProgressEvent, StreamError>>,
72    > {
73        let url = format!(
74            "{}{}",
75            self.base_url,
76            crate::paths::update_batches::stream(id)
77        );
78
79        let mut req = self
80            .http
81            .get(&url)
82            .header("Accept", "text/event-stream")
83            .timeout(std::time::Duration::from_secs(86400));
84
85        if let Some(token) = &self.token {
86            req = req.bearer_auth(token);
87        }
88
89        let resp = req.send().await.context_to()?;
90
91        let status = resp.status();
92        if status == reqwest::StatusCode::UNAUTHORIZED {
93            bail!(ClientError::NotAuthenticated);
94        }
95        if status == reqwest::StatusCode::NOT_FOUND {
96            let text = resp.text().await.context_to()?;
97            let message = crate::extract_error_message(&text);
98            bail!(ClientError::NotFound(message));
99        }
100        if status.is_client_error() || status.is_server_error() {
101            let text = resp.text().await.context_to()?;
102            let message = crate::extract_error_message(&text);
103            bail!(ClientError::Api { status, message });
104        }
105
106        let raw_stream = sse::parse_sse_stream(resp);
107
108        let typed_stream = futures_util::StreamExt::filter_map(raw_stream, |result| async move {
109            match result {
110                Ok(event) => parse_typed_event(event),
111                Err(e) => Some(Err(StreamError::Sse(e))),
112            }
113        });
114
115        Ok(typed_stream)
116    }
117}
118
119/// Parse a raw SSE event into a typed [`BatchProgressEvent`].
120fn parse_typed_event(
121    event: RawSseEvent,
122) -> Option<std::result::Result<BatchProgressEvent, StreamError>> {
123    match event.event_type.as_str() {
124        "update" => {
125            let parsed: std::result::Result<BatchUpdateEventData, _> =
126                serde_json::from_str(&event.data);
127            Some(parsed.map(BatchProgressEvent::Update).map_err(Into::into))
128        }
129        "progress" => {
130            let parsed: std::result::Result<BatchProgressData, _> =
131                serde_json::from_str(&event.data);
132            Some(parsed.map(BatchProgressEvent::Progress).map_err(Into::into))
133        }
134        "batch_completed" => {
135            let parsed: std::result::Result<BatchCompletedData, _> =
136                serde_json::from_str(&event.data);
137            Some(
138                parsed
139                    .map(BatchProgressEvent::BatchCompleted)
140                    .map_err(Into::into),
141            )
142        }
143        _ => None,
144    }
145}
146
147#[cfg(test)]
148mod tests {
149    use super::*;
150
151    #[test]
152    fn parse_update_event() {
153        let event = RawSseEvent {
154            event_type: "update".to_string(),
155            data: r#"{"event":"update_completed","update_history_id":"01234567-89ab-cdef-0123-456789abcdef","software_item_name":"nginx","host_name":"web-01"}"#.to_string(),
156            id: None,
157        };
158        let result = parse_typed_event(event).expect("should produce event");
159        let typed = result.expect("should parse");
160        assert!(
161            matches!(typed, BatchProgressEvent::Update(ref u) if u.software_item_name == "nginx")
162        );
163    }
164
165    #[test]
166    fn parse_progress_event() {
167        let event = RawSseEvent {
168            event_type: "progress".to_string(),
169            data: r#"{"event":"progress","completed":2,"failed":0,"pending":3,"total":5}"#
170                .to_string(),
171            id: None,
172        };
173        let result = parse_typed_event(event).expect("should produce event");
174        let typed = result.expect("should parse");
175        assert!(matches!(typed, BatchProgressEvent::Progress(ref p) if p.total == 5));
176    }
177
178    #[test]
179    fn parse_batch_completed_event() {
180        let event = RawSseEvent {
181            event_type: "batch_completed".to_string(),
182            data: r#"{"event":"batch_completed","status":"completed"}"#.to_string(),
183            id: None,
184        };
185        let result = parse_typed_event(event).expect("should produce event");
186        let typed = result.expect("should parse");
187        assert!(
188            matches!(typed, BatchProgressEvent::BatchCompleted(ref c) if c.status == "completed")
189        );
190    }
191
192    #[test]
193    fn parse_unknown_event_returns_none() {
194        let event = RawSseEvent {
195            event_type: "ping".to_string(),
196            data: "{}".to_string(),
197            id: None,
198        };
199        assert!(parse_typed_event(event).is_none());
200    }
201}
uptrakit_openapi_client/batch_progress_stream.rs

uptrakit_openapi_client/
batch_progress_stream.rs
