uptrakit_openapi_client/
device_auth_stream.rs

1//! Typed SSE streaming method for device authorization.
2//!
3//! Provides [`UptrakitClient::stream_device_auth`] which connects to the
4//! `GET /api/v1/auth/device/stream` endpoint and returns a typed stream of
5//! device authorization events.
6
7use crate::sse::{self, RawSseEvent, SseError};
8use crate::types_impl::device_auth::DeviceAuthAuthorizedSse;
9use crate::{ClientError, Result, UptrakitClient};
10use rootcause::prelude::*;
11
12/// A typed SSE event from the device auth stream.
13#[derive(Debug, Clone)]
14pub enum DeviceAuthSseEvent {
15    /// The device flow was approved; contains the API token and token name.
16    Authorized { token: String, token_name: String },
17    /// The device flow expired before approval.
18    Expired,
19}
20
21/// Errors specific to device auth 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 device auth SSE stream and return a stream of typed events.
33    ///
34    /// The returned stream yields [`DeviceAuthSseEvent`] values until the device
35    /// flow is authorized or expires, then the stream closes.
36    ///
37    /// This is an unauthenticated endpoint (same as device auth poll).
38    /// Uses a 700s timeout (slightly beyond the 600s flow TTL).
39    #[cfg_attr(feature = "tracing", tracing::instrument(skip_all))]
40    pub async fn stream_device_auth(
41        &self,
42        device_code: &str,
43    ) -> Result<
44        impl futures_util::Stream<Item = std::result::Result<DeviceAuthSseEvent, StreamError>>,
45    > {
46        let url = format!("{}{}", self.base_url, crate::paths::auth::DEVICE_STREAM);
47
48        let req = self
49            .http
50            .get(&url)
51            .query(&[("device_code", device_code)])
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(700));
56
57        let resp = req.send().await.context_to()?;
58
59        let status = resp.status();
60        if status == reqwest::StatusCode::NOT_FOUND {
61            let text = resp.text().await.context_to()?;
62            let message = crate::extract_error_message(&text);
63            bail!(ClientError::NotFound(message));
64        }
65        if status.is_client_error() || status.is_server_error() {
66            let text = resp.text().await.context_to()?;
67            let message = crate::extract_error_message(&text);
68            bail!(ClientError::Api { status, message });
69        }
70
71        let raw_stream = sse::parse_sse_stream(resp);
72
73        let typed_stream = futures_util::StreamExt::filter_map(raw_stream, |result| async move {
74            match result {
75                Ok(event) => parse_typed_event(event),
76                Err(e) => Some(Err(StreamError::Sse(e))),
77            }
78        });
79
80        Ok(typed_stream)
81    }
82}
83
84/// Parse a raw SSE event into a typed [`DeviceAuthSseEvent`].
85fn parse_typed_event(
86    event: RawSseEvent,
87) -> Option<std::result::Result<DeviceAuthSseEvent, StreamError>> {
88    match event.event_type.as_str() {
89        "authorized" => {
90            let parsed: std::result::Result<DeviceAuthAuthorizedSse, _> =
91                serde_json::from_str(&event.data);
92            Some(
93                parsed
94                    .map(|a| DeviceAuthSseEvent::Authorized {
95                        token: a.token.expose_secret().to_string(),
96                        token_name: a.token_name,
97                    })
98                    .map_err(Into::into),
99            )
100        }
101        "expired" => Some(Ok(DeviceAuthSseEvent::Expired)),
102        _ => {
103            // Unknown event types (e.g. keep-alive comments) are silently skipped.
104            None
105        }
106    }
107}
108
109#[cfg(test)]
110mod tests {
111    use super::*;
112    use crate::sse::RawSseEvent;
113
114    #[test]
115    fn parse_authorized_event() {
116        let event = RawSseEvent {
117            event_type: "authorized".to_string(),
118            data: r#"{"token":"upt_secret123","token_name":"cli-host-2026"}"#.to_string(),
119            id: None,
120        };
121        let result = parse_typed_event(event).expect("should produce event");
122        let typed = result.expect("should parse");
123        match typed {
124            DeviceAuthSseEvent::Authorized { token, token_name } => {
125                assert_eq!(token, "upt_secret123");
126                assert_eq!(token_name, "cli-host-2026");
127            }
128            _ => panic!("expected Authorized"),
129        }
130    }
131
132    #[test]
133    fn parse_expired_event() {
134        let event = RawSseEvent {
135            event_type: "expired".to_string(),
136            data: r#"{"message":"Device flow expired"}"#.to_string(),
137            id: None,
138        };
139        let result = parse_typed_event(event).expect("should produce event");
140        let typed = result.expect("should parse");
141        assert!(matches!(typed, DeviceAuthSseEvent::Expired));
142    }
143
144    #[test]
145    fn parse_unknown_event_returns_none() {
146        let event = RawSseEvent {
147            event_type: "ping".to_string(),
148            data: "{}".to_string(),
149            id: None,
150        };
151        assert!(parse_typed_event(event).is_none());
152    }
153
154    #[test]
155    fn parse_malformed_authorized_returns_error() {
156        let event = RawSseEvent {
157            event_type: "authorized".to_string(),
158            data: "not json".to_string(),
159            id: None,
160        };
161        let result = parse_typed_event(event).expect("should produce event");
162        assert!(result.is_err());
163    }
164}
uptrakit_openapi_client/device_auth_stream.rs

uptrakit_openapi_client/
device_auth_stream.rs
