uptrakit_openapi_client/
update_output_stream.rs

1//! Typed SSE streaming method for update output.
2//!
3//! Provides [`UptrakitClient::stream_update_output`] which connects to the
4//! `GET /api/v1/update-history/{id}/output/stream` endpoint and returns a
5//! typed stream of update output events.
6
7use crate::sse::{self, RawSseEvent, SseError};
8use crate::types_impl::update_history::{OutputLineSSE, UpdateCompletedSSE};
9use crate::{ClientError, Result, UptrakitClient};
10use rootcause::prelude::*;
11
12/// A typed SSE event from the update output stream.
13#[derive(Debug, Clone)]
14pub enum UpdateOutputEvent {
15    /// A line of output from the update process.
16    Output(OutputLineSSE),
17    /// The update has completed (or failed).
18    Completed(UpdateCompletedSSE),
19}
20
21/// Errors specific to update output streaming.
22#[derive(Debug, thiserror::Error)]
23pub enum StreamError {
24    #[error("SSE transport error: {0}")]
25    Sse(#[from] SseError),
26
27    #[error("failed to parse SSE event data: {0}")]
28    Parse(#[from] serde_json::Error),
29}
30
31impl UptrakitClient {
32    /// Connect to the update output SSE stream and return a stream of typed events.
33    ///
34    /// The returned stream yields [`UpdateOutputEvent`] values until the update
35    /// completes (indicated by a `Completed` event) or the connection closes.
36    ///
37    /// This method uses no request timeout since SSE connections are long-lived.
38    pub async fn stream_update_output(
39        &self,
40        id: &uuid::Uuid,
41    ) -> Result<impl futures_util::Stream<Item = std::result::Result<UpdateOutputEvent, StreamError>>>
42    {
43        let url = format!(
44            "{}{}",
45            self.base_url,
46            crate::paths::update_history::output_stream(id)
47        );
48
49        let mut req = self
50            .http
51            .get(&url)
52            .header("Accept", "text/event-stream")
53            // Override the client's default request timeout — SSE connections
54            // are long-lived and should not be timed out by the HTTP client.
55            .timeout(std::time::Duration::from_secs(86400));
56
57        if let Some(token) = &self.token {
58            req = req.bearer_auth(token);
59        }
60
61        let resp = req.send().await.context_to()?;
62
63        let status = resp.status();
64        if status == reqwest::StatusCode::UNAUTHORIZED {
65            bail!(ClientError::NotAuthenticated);
66        }
67        if status == reqwest::StatusCode::NOT_FOUND {
68            let text = resp.text().await.context_to()?;
69            let message = crate::extract_error_message(&text);
70            bail!(ClientError::NotFound(message));
71        }
72        if status.is_client_error() || status.is_server_error() {
73            let text = resp.text().await.context_to()?;
74            let message = crate::extract_error_message(&text);
75            bail!(ClientError::Api { status, message });
76        }
77
78        let raw_stream = sse::parse_sse_stream(resp);
79
80        let typed_stream = futures_util::StreamExt::filter_map(raw_stream, |result| async move {
81            match result {
82                Ok(event) => parse_typed_event(event),
83                Err(e) => Some(Err(StreamError::Sse(e))),
84            }
85        });
86
87        Ok(typed_stream)
88    }
89}
90
91/// Parse a raw SSE event into a typed [`UpdateOutputEvent`].
92fn parse_typed_event(
93    event: RawSseEvent,
94) -> Option<std::result::Result<UpdateOutputEvent, StreamError>> {
95    match event.event_type.as_str() {
96        "output" => {
97            let parsed: std::result::Result<OutputLineSSE, _> = serde_json::from_str(&event.data);
98            Some(parsed.map(UpdateOutputEvent::Output).map_err(Into::into))
99        }
100        "completed" => {
101            let parsed: std::result::Result<UpdateCompletedSSE, _> =
102                serde_json::from_str(&event.data);
103            Some(parsed.map(UpdateOutputEvent::Completed).map_err(Into::into))
104        }
105        _ => {
106            // Unknown event types (e.g. keep-alive comments) are silently skipped.
107            None
108        }
109    }
110}
111
112#[cfg(test)]
113mod tests {
114    #![expect(
115        clippy::assertions_on_result_states,
116        reason = "test assertions — assert!(result.is_err()) is idiomatic in tests"
117    )]
118
119    use super::*;
120    use crate::sse::RawSseEvent;
121
122    #[test]
123    fn parse_output_event() {
124        let event = RawSseEvent {
125            event_type: "output".to_string(),
126            data: r#"{"id":"01234567-89ab-cdef-0123-456789abcdef","text":"hello\n","stream":"stdout","timestamp":"2025-01-01T00:00:00Z","seq":0}"#.to_string(),
127            id: None,
128        };
129        let result = parse_typed_event(event).expect("should produce event");
130        let typed = result.expect("should parse");
131        assert!(matches!(typed, UpdateOutputEvent::Output(ref o) if o.text == "hello\n"));
132    }
133
134    #[test]
135    fn parse_completed_event() {
136        let event = RawSseEvent {
137            event_type: "completed".to_string(),
138            data: r#"{"status":"completed","error":null}"#.to_string(),
139            id: None,
140        };
141        let result = parse_typed_event(event).expect("should produce event");
142        let typed = result.expect("should parse");
143        assert!(matches!(typed, UpdateOutputEvent::Completed(ref c) if c.status == "completed"));
144    }
145
146    #[test]
147    fn parse_unknown_event_returns_none() {
148        let event = RawSseEvent {
149            event_type: "ping".to_string(),
150            data: "{}".to_string(),
151            id: None,
152        };
153        assert!(parse_typed_event(event).is_none());
154    }
155
156    #[test]
157    fn parse_malformed_data_returns_error() {
158        let event = RawSseEvent {
159            event_type: "output".to_string(),
160            data: "not json".to_string(),
161            id: None,
162        };
163        let result = parse_typed_event(event).expect("should produce event");
164        assert!(result.is_err());
165    }
166}
uptrakit_openapi_client/update_output_stream.rs

uptrakit_openapi_client/
update_output_stream.rs
