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    use super::*;
115    use crate::sse::RawSseEvent;
116
117    #[test]
118    fn parse_output_event() {
119        let event = RawSseEvent {
120            event_type: "output".to_string(),
121            data: r#"{"id":"01234567-89ab-cdef-0123-456789abcdef","text":"hello\n","stream":"stdout","timestamp":"2025-01-01T00:00:00Z","seq":0}"#.to_string(),
122            id: None,
123        };
124        let result = parse_typed_event(event).expect("should produce event");
125        let typed = result.expect("should parse");
126        assert!(matches!(typed, UpdateOutputEvent::Output(ref o) if o.text == "hello\n"));
127    }
128
129    #[test]
130    fn parse_completed_event() {
131        let event = RawSseEvent {
132            event_type: "completed".to_string(),
133            data: r#"{"status":"completed","error":null}"#.to_string(),
134            id: None,
135        };
136        let result = parse_typed_event(event).expect("should produce event");
137        let typed = result.expect("should parse");
138        assert!(matches!(typed, UpdateOutputEvent::Completed(ref c) if c.status == "completed"));
139    }
140
141    #[test]
142    fn parse_unknown_event_returns_none() {
143        let event = RawSseEvent {
144            event_type: "ping".to_string(),
145            data: "{}".to_string(),
146            id: None,
147        };
148        assert!(parse_typed_event(event).is_none());
149    }
150
151    #[test]
152    fn parse_malformed_data_returns_error() {
153        let event = RawSseEvent {
154            event_type: "output".to_string(),
155            data: "not json".to_string(),
156            id: None,
157        };
158        let result = parse_typed_event(event).expect("should produce event");
159        assert!(result.is_err());
160    }
161}
uptrakit_openapi_client/update_output_stream.rs

uptrakit_openapi_client/
update_output_stream.rs
