ironflow_api/routes/
events.rs

1//! SSE endpoint for real-time event streaming.
2
3use std::convert::Infallible;
4use std::str::FromStr;
5use std::time::Duration;
6
7use axum::extract::{Query, State};
8use axum::response::sse::{Event as SseEvent, KeepAlive, Sse};
9use futures_util::stream::{Stream, StreamExt};
10use serde::Deserialize;
11use serde::de::{self, Deserializer};
12use tokio_stream::wrappers::BroadcastStream;
13use uuid::Uuid;
14
15use crate::state::AppState;
16use ironflow_auth::extractor::Authenticated;
17use ironflow_engine::notify::Event;
18
19pub use ironflow_store::entities::EventKind;
20
21/// Deserialize a comma-separated string into `Option<Vec<EventKind>>`.
22fn deserialize_comma_event_kinds<'de, D>(
23    deserializer: D,
24) -> Result<Option<Vec<EventKind>>, D::Error>
25where
26    D: Deserializer<'de>,
27{
28    let opt: Option<String> = Option::deserialize(deserializer)?;
29    match opt {
30        None => Ok(None),
31        Some(raw) => {
32            let kinds: Result<Vec<EventKind>, _> = raw
33                .split(',')
34                .map(|s| s.trim())
35                .filter(|s| !s.is_empty())
36                .map(EventKind::from_str)
37                .collect();
38            kinds.map(Some).map_err(de::Error::custom)
39        }
40    }
41}
42
43/// Query parameters for the SSE events endpoint.
44///
45/// Both fields are optional. When set, only matching events are streamed.
46///
47/// # Examples
48///
49/// ```
50/// use ironflow_api::routes::events::EventsQuery;
51///
52/// let query = EventsQuery {
53///     run_id: None,
54///     types: None,
55/// };
56/// ```
57#[derive(Debug, Deserialize)]
58pub struct EventsQuery {
59    /// Only stream events related to this run.
60    pub run_id: Option<Uuid>,
61    /// Comma-separated list of event types to include (e.g. `?types=run_status_changed,step_completed`).
62    #[serde(default, deserialize_with = "deserialize_comma_event_kinds")]
63    pub types: Option<Vec<EventKind>>,
64}
65
66/// `GET /api/v1/events` -- Server-Sent Events stream.
67///
68/// Streams domain events in real time. Supports optional filtering:
69/// - `?run_id=<uuid>` -- only events for that run
70/// - `?types=run_status_changed,step_completed` -- only those event types
71///
72/// Each SSE message has:
73/// - `event:` set to the event type (e.g. `run_status_changed`)
74/// - `data:` JSON-serialized event payload
75///
76/// A keep-alive comment is sent every 30 seconds.
77///
78/// # Errors
79///
80/// Returns 401 if the request is not authenticated.
81pub async fn events(
82    auth: Authenticated,
83    State(state): State<AppState>,
84    Query(query): Query<EventsQuery>,
85) -> Sse<impl Stream<Item = Result<SseEvent, Infallible>>> {
86    let receiver = state.event_sender.subscribe();
87    // Provider Account events describe admin-only resources.
88    let is_admin = auth.is_admin();
89    let type_filter = query.types;
90
91    let stream = BroadcastStream::new(receiver).filter_map(move |result: Result<Event, _>| {
92        let type_filter = type_filter.clone();
93        let run_id_filter = query.run_id;
94        async move {
95            let event = result.ok()?;
96
97            if !is_admin
98                && matches!(
99                    event,
100                    Event::ProviderAccountUpdated(_) | Event::ProviderAccountUsageUpdated(_)
101                )
102            {
103                return None;
104            }
105
106            if let Some(ref rid) = run_id_filter
107                && event.run_id() != Some(*rid)
108            {
109                return None;
110            }
111
112            if let Some(ref kinds) = type_filter {
113                let event_type = event.event_type();
114                if !kinds.iter().any(|k| k.as_str() == event_type) {
115                    return None;
116                }
117            }
118
119            let data = serde_json::to_string(&event).ok()?;
120            let sse_event = SseEvent::default().event(event.event_type()).data(data);
121
122            Some(Ok::<_, Infallible>(sse_event))
123        }
124    });
125
126    Sse::new(stream).keep_alive(KeepAlive::new().interval(Duration::from_secs(30)))
127}
128
129#[cfg(test)]
130mod tests {
131    use std::collections::HashMap;
132    use std::sync::Arc;
133    use std::time::Duration;
134
135    use axum::Router;
136    use axum::routing::get;
137    use chrono::Utc;
138    use ironflow_core::providers::claude::ClaudeCodeProvider;
139    use ironflow_engine::engine::Engine;
140    use ironflow_engine::notify::{Event, RunStatusChangedEvent, UserSignedInEvent};
141    use ironflow_store::memory::InMemoryStore;
142    use ironflow_store::models::RunStatus;
143    use rust_decimal::Decimal;
144    use tokio::io::AsyncBufReadExt;
145    use tokio::io::BufReader;
146    use tokio::net::TcpListener;
147    use tokio::sync::broadcast;
148    use tokio::time::{sleep, timeout};
149    use uuid::Uuid;
150
151    use super::events;
152    use crate::routes::test_helpers::create_user_auth_header;
153    use crate::state::AppState;
154
155    fn test_state() -> AppState {
156        let store = Arc::new(InMemoryStore::new());
157        let provider = Arc::new(ClaudeCodeProvider::new());
158        let engine = Arc::new(Engine::new(store.clone(), provider));
159        let jwt_config = Arc::new(ironflow_auth::jwt::JwtConfig {
160            secret: "test-secret".to_string(),
161            access_token_ttl_secs: 900,
162            refresh_token_ttl_secs: 604800,
163            cookie_domain: None,
164            cookie_secure: false,
165        });
166        let (event_sender, _) = broadcast::channel::<Event>(16);
167        AppState::new(
168            store,
169            engine,
170            jwt_config,
171            "test-worker-token".to_string(),
172            event_sender,
173        )
174    }
175
176    fn sample_run_event(run_id: Uuid) -> Event {
177        Event::RunStatusChanged(RunStatusChangedEvent {
178            run_id,
179            workflow_name: "deploy".to_string(),
180            from: RunStatus::Running,
181            to: RunStatus::Completed,
182            error: None,
183            cost_usd: Decimal::ZERO,
184            duration_ms: 1000,
185            labels: HashMap::new(),
186            at: Utc::now(),
187        })
188    }
189
190    fn sample_user_event() -> Event {
191        Event::UserSignedIn(UserSignedInEvent {
192            user_id: Uuid::now_v7(),
193            username: "alice".to_string(),
194            at: Utc::now(),
195        })
196    }
197
198    /// Start a real TCP server and return (address, sender, auth header).
199    async fn start_sse_server(state: AppState) -> (String, broadcast::Sender<Event>, String) {
200        let sender = state.event_sender.clone();
201        let auth = create_user_auth_header(&state, "testuser", false).await;
202        let app = Router::new()
203            .route("/events", get(events))
204            .with_state(state);
205
206        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
207        let addr = listener.local_addr().unwrap().to_string();
208        tokio::spawn(async move {
209            axum::serve(listener, app).await.unwrap();
210        });
211        (addr, sender, auth)
212    }
213
214    /// Connect to the SSE endpoint with auth and return a line reader.
215    async fn connect_sse(addr: &str, query: &str, auth: &str) -> BufReader<tokio::net::TcpStream> {
216        let stream = tokio::net::TcpStream::connect(addr).await.unwrap();
217        let (reader, mut writer) = stream.into_split();
218
219        use tokio::io::AsyncWriteExt;
220        writer
221            .write_all(
222                format!(
223                    "GET /events{query} HTTP/1.1\r\nHost: {addr}\r\nAccept: text/event-stream\r\nAuthorization: {auth}\r\n\r\n"
224                )
225                .as_bytes(),
226            )
227            .await
228            .unwrap();
229
230        BufReader::new(reader.reunite(writer).unwrap())
231    }
232
233    /// Read all available data from the SSE stream until `needle` is found
234    /// in the accumulated text, or timeout.
235    async fn read_until_contains(
236        reader: &mut BufReader<tokio::net::TcpStream>,
237        needle: &str,
238        dur: Duration,
239    ) -> String {
240        let mut accumulated = String::new();
241        let result = timeout(dur, async {
242            loop {
243                let mut line = String::new();
244                let n = reader.read_line(&mut line).await.unwrap();
245                if n == 0 {
246                    break;
247                }
248                accumulated.push_str(&line);
249                if accumulated.contains(needle) {
250                    break;
251                }
252            }
253        })
254        .await;
255        if result.is_err() {
256            panic!("timeout waiting for '{needle}' in SSE stream. Data so far:\n{accumulated}");
257        }
258        accumulated
259    }
260
261    /// Wait until the SSE handler has subscribed to the broadcast channel.
262    ///
263    /// Sending before a receiver exists makes `broadcast::Sender::send` return
264    /// `SendError`, so a fixed sleep is racy on slow runners. Poll the receiver
265    /// count instead.
266    async fn wait_for_subscriber(sender: &broadcast::Sender<Event>) {
267        let ready = timeout(Duration::from_secs(5), async {
268            while sender.receiver_count() == 0 {
269                sleep(Duration::from_millis(5)).await;
270            }
271        })
272        .await;
273        assert!(ready.is_ok(), "SSE handler did not subscribe in time");
274    }
275
276    #[tokio::test]
277    async fn sse_stream_receives_events() {
278        let state = test_state();
279        let (addr, sender, auth) = start_sse_server(state).await;
280        let mut reader = connect_sse(&addr, "", &auth).await;
281
282        wait_for_subscriber(&sender).await;
283
284        let run_id = Uuid::now_v7();
285        sender.send(sample_run_event(run_id)).unwrap();
286
287        let text =
288            read_until_contains(&mut reader, &run_id.to_string(), Duration::from_secs(5)).await;
289
290        assert!(text.contains("run_status_changed"));
291        assert!(text.contains(&run_id.to_string()));
292    }
293
294    #[tokio::test]
295    async fn sse_filters_by_run_id() {
296        let state = test_state();
297        let (addr, sender, auth) = start_sse_server(state).await;
298
299        let target_run = Uuid::now_v7();
300        let other_run = Uuid::now_v7();
301
302        let mut reader = connect_sse(&addr, &format!("?run_id={target_run}"), &auth).await;
303        wait_for_subscriber(&sender).await;
304
305        sender.send(sample_run_event(other_run)).unwrap();
306        sender.send(sample_run_event(target_run)).unwrap();
307
308        let text =
309            read_until_contains(&mut reader, &target_run.to_string(), Duration::from_secs(5)).await;
310
311        assert!(text.contains(&target_run.to_string()));
312        assert!(!text.contains(&other_run.to_string()));
313    }
314
315    #[tokio::test]
316    async fn sse_filters_by_event_type() {
317        let state = test_state();
318        let (addr, sender, auth) = start_sse_server(state).await;
319
320        let mut reader = connect_sse(&addr, "?types=user_signed_in", &auth).await;
321        wait_for_subscriber(&sender).await;
322
323        let run_id = Uuid::now_v7();
324        sender.send(sample_run_event(run_id)).unwrap();
325        sender.send(sample_user_event()).unwrap();
326
327        let text = read_until_contains(&mut reader, "user_signed_in", Duration::from_secs(5)).await;
328
329        assert!(text.contains("user_signed_in"));
330        assert!(!text.contains("run_status_changed"));
331    }
332
333    #[tokio::test]
334    async fn sse_returns_correct_content_type() {
335        let state = test_state();
336        let (addr, _sender, auth) = start_sse_server(state).await;
337        let mut reader = connect_sse(&addr, "", &auth).await;
338
339        let text =
340            read_until_contains(&mut reader, "text/event-stream", Duration::from_secs(5)).await;
341
342        assert!(text.contains("text/event-stream"));
343    }
344
345    #[tokio::test]
346    async fn sse_rejects_unauthenticated() {
347        let state = test_state();
348        let (addr, _sender, _auth) = start_sse_server(state).await;
349        // Connect without auth header
350        let stream = tokio::net::TcpStream::connect(&addr).await.unwrap();
351        let (reader, mut writer) = stream.into_split();
352
353        use tokio::io::AsyncWriteExt;
354        writer
355            .write_all(
356                format!(
357                    "GET /events HTTP/1.1\r\nHost: {addr}\r\nAccept: text/event-stream\r\n\r\n"
358                )
359                .as_bytes(),
360            )
361            .await
362            .unwrap();
363
364        let mut buf_reader = BufReader::new(reader.reunite(writer).unwrap());
365        let text = read_until_contains(&mut buf_reader, "401", Duration::from_secs(5)).await;
366
367        assert!(text.contains("401"));
368    }
369}
ironflow_api/routes/events.rs

ironflow_api/routes/
events.rs
