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::authorization::AuthorizeExt;
16use crate::error::ApiError;
17use crate::state::AppState;
18use ironflow_auth::extractor::Authenticated;
19use ironflow_engine::notify::Event;
20use ironflow_store::entities::ApiKeyScope;
21
22pub use ironflow_store::entities::EventKind;
23
24/// Deserialize a comma-separated string into `Option<Vec<EventKind>>`.
25fn deserialize_comma_event_kinds<'de, D>(
26    deserializer: D,
27) -> Result<Option<Vec<EventKind>>, D::Error>
28where
29    D: Deserializer<'de>,
30{
31    let opt: Option<String> = Option::deserialize(deserializer)?;
32    match opt {
33        None => Ok(None),
34        Some(raw) => {
35            let kinds: Result<Vec<EventKind>, _> = raw
36                .split(',')
37                .map(|s| s.trim())
38                .filter(|s| !s.is_empty())
39                .map(EventKind::from_str)
40                .collect();
41            kinds.map(Some).map_err(de::Error::custom)
42        }
43    }
44}
45
46/// Query parameters for the SSE events endpoint.
47///
48/// Both fields are optional. When set, only matching events are streamed.
49///
50/// # Examples
51///
52/// ```
53/// use ironflow_api::routes::events::EventsQuery;
54///
55/// let query = EventsQuery {
56///     run_id: None,
57///     types: None,
58/// };
59/// ```
60#[derive(Debug, Deserialize)]
61pub struct EventsQuery {
62    /// Only stream events related to this run.
63    pub run_id: Option<Uuid>,
64    /// Comma-separated list of event types to include (e.g. `?types=run_status_changed,step_completed`).
65    #[serde(default, deserialize_with = "deserialize_comma_event_kinds")]
66    pub types: Option<Vec<EventKind>>,
67}
68
69/// `GET /api/v1/events` -- Server-Sent Events stream.
70///
71/// Streams domain events in real time. Supports optional filtering:
72/// - `?run_id=<uuid>` -- only events for that run
73/// - `?types=run_status_changed,step_completed` -- only those event types
74///
75/// Each SSE message has:
76/// - `event:` set to the event type (e.g. `run_status_changed`)
77/// - `data:` JSON-serialized event payload
78///
79/// A keep-alive comment is sent every 30 seconds.
80///
81/// Provider Account events reach an admin session, or an API key holding the
82/// `admin` scope whose owner is an admin.
83///
84/// # Errors
85///
86/// - 401 if the request is not authenticated
87/// - 403 `INSUFFICIENT_SCOPE` if the API key lacks `runs_read`
88pub async fn events(
89    auth: Authenticated,
90    State(state): State<AppState>,
91    Query(query): Query<EventsQuery>,
92) -> Result<Sse<impl Stream<Item = Result<SseEvent, Infallible>>>, ApiError> {
93    auth.require(ApiKeyScope::RunsRead, false)?;
94    let receiver = state.event_sender.subscribe();
95    // Provider Account events describe admin-only resources, which an API
96    // key reaches only through the `admin` scope.
97    let sees_provider_account_events = auth.require(ApiKeyScope::Admin, true).is_ok();
98    let type_filter = query.types;
99
100    let stream = BroadcastStream::new(receiver).filter_map(move |result: Result<Event, _>| {
101        let type_filter = type_filter.clone();
102        let run_id_filter = query.run_id;
103        async move {
104            let event = result.ok()?;
105
106            if !sees_provider_account_events
107                && matches!(
108                    event,
109                    Event::ProviderAccountUpdated(_) | Event::ProviderAccountUsageUpdated(_)
110                )
111            {
112                return None;
113            }
114
115            if let Some(ref rid) = run_id_filter
116                && event.run_id() != Some(*rid)
117            {
118                return None;
119            }
120
121            if let Some(ref kinds) = type_filter {
122                let event_type = event.event_type();
123                if !kinds.iter().any(|k| k.as_str() == event_type) {
124                    return None;
125                }
126            }
127
128            let data = serde_json::to_string(&event).ok()?;
129            let sse_event = SseEvent::default().event(event.event_type()).data(data);
130
131            Some(Ok::<_, Infallible>(sse_event))
132        }
133    });
134
135    let sse = Sse::new(stream).keep_alive(KeepAlive::new().interval(Duration::from_secs(30)));
136    Ok(sse)
137}
138
139#[cfg(test)]
140mod tests {
141    use std::collections::HashMap;
142    use std::sync::Arc;
143    use std::time::Duration;
144
145    use axum::Router;
146    use axum::routing::get;
147    use chrono::Utc;
148    use ironflow_auth::extractor::{API_KEY_PREFIX, API_KEY_SUFFIX_LEN};
149    use ironflow_auth::password;
150    use ironflow_core::providers::claude::ClaudeCodeProvider;
151    use ironflow_engine::engine::Engine;
152    use ironflow_engine::notify::{
153        Event, ProviderAccountChange, ProviderAccountUpdatedEvent, RunStatusChangedEvent,
154        UserSignedInEvent,
155    };
156    use ironflow_store::entities::{ApiKeyScope, NewApiKey, NewUser};
157    use ironflow_store::memory::InMemoryStore;
158    use ironflow_store::models::RunStatus;
159    use rust_decimal::Decimal;
160    use tokio::io::AsyncBufReadExt;
161    use tokio::io::BufReader;
162    use tokio::net::TcpListener;
163    use tokio::sync::broadcast;
164    use tokio::time::{sleep, timeout};
165    use uuid::Uuid;
166
167    use super::events;
168    use crate::routes::test_helpers::create_user_auth_header;
169    use crate::state::AppState;
170
171    fn test_state() -> AppState {
172        let store = Arc::new(InMemoryStore::new());
173        let provider = Arc::new(ClaudeCodeProvider::new());
174        let engine = Arc::new(Engine::new(store.clone(), provider));
175        let jwt_config = Arc::new(ironflow_auth::jwt::JwtConfig {
176            secret: "test-secret".to_string(),
177            access_token_ttl_secs: 900,
178            refresh_token_ttl_secs: 604800,
179            cookie_domain: None,
180            cookie_secure: false,
181        });
182        let (event_sender, _) = broadcast::channel::<Event>(16);
183        AppState::new(
184            store,
185            engine,
186            jwt_config,
187            "test-worker-token".to_string(),
188            event_sender,
189        )
190    }
191
192    fn sample_run_event(run_id: Uuid) -> Event {
193        Event::RunStatusChanged(RunStatusChangedEvent {
194            run_id,
195            workflow_name: "deploy".to_string(),
196            from: RunStatus::Running,
197            to: RunStatus::Completed,
198            error: None,
199            cost_usd: Decimal::ZERO,
200            duration_ms: 1000,
201            labels: HashMap::new(),
202            at: Utc::now(),
203        })
204    }
205
206    fn sample_user_event() -> Event {
207        Event::UserSignedIn(UserSignedInEvent {
208            user_id: Uuid::now_v7(),
209            username: "alice".to_string(),
210            at: Utc::now(),
211        })
212    }
213
214    fn sample_provider_account_event() -> Event {
215        Event::ProviderAccountUpdated(ProviderAccountUpdatedEvent {
216            account_id: Uuid::now_v7(),
217            name: "acct".into(),
218            change: ProviderAccountChange::Updated,
219            at: Utc::now(),
220        })
221    }
222
223    /// A `Bearer` header for an API key holding `scopes`, owned by a fresh
224    /// user who is an admin when `owner_is_admin` is set.
225    ///
226    /// The key is written to the store directly, so a member may hold scopes
227    /// the `/api-keys` route would refuse them.
228    async fn api_key_header(
229        state: &AppState,
230        owner_is_admin: bool,
231        scopes: Vec<ApiKeyScope>,
232    ) -> String {
233        let suffix = Uuid::now_v7().simple().to_string();
234        let owner = state
235            .store
236            .create_user(NewUser {
237                email: format!("owner-{suffix}@example.com"),
238                username: format!("owner-{suffix}"),
239                password_hash: "argon2hash".to_string(),
240                is_admin: Some(owner_is_admin),
241            })
242            .await
243            .unwrap();
244        let raw_key = format!("irfl_{}rest-of-secret-key", &suffix[24..]);
245        let prefix = &raw_key[..API_KEY_PREFIX.len() + API_KEY_SUFFIX_LEN];
246        state
247            .store
248            .create_api_key(NewApiKey {
249                user_id: owner.id,
250                name: format!("key-{suffix}"),
251                key_hash: password::hash(&raw_key).unwrap(),
252                key_prefix: prefix.to_string(),
253                scopes,
254                expires_at: None,
255                rate_limit_override: None,
256            })
257            .await
258            .unwrap();
259        format!("Bearer {raw_key}")
260    }
261
262    /// Start a real TCP server and return (address, sender, auth header).
263    async fn start_sse_server(state: AppState) -> (String, broadcast::Sender<Event>, String) {
264        let sender = state.event_sender.clone();
265        let auth = create_user_auth_header(&state, "testuser", false).await;
266        let app = Router::new()
267            .route("/events", get(events))
268            .with_state(state);
269
270        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
271        let addr = listener.local_addr().unwrap().to_string();
272        tokio::spawn(async move {
273            axum::serve(listener, app).await.unwrap();
274        });
275        (addr, sender, auth)
276    }
277
278    /// Connect to the SSE endpoint with auth and return a line reader.
279    async fn connect_sse(addr: &str, query: &str, auth: &str) -> BufReader<tokio::net::TcpStream> {
280        let stream = tokio::net::TcpStream::connect(addr).await.unwrap();
281        let (reader, mut writer) = stream.into_split();
282
283        use tokio::io::AsyncWriteExt;
284        writer
285            .write_all(
286                format!(
287                    "GET /events{query} HTTP/1.1\r\nHost: {addr}\r\nAccept: text/event-stream\r\nAuthorization: {auth}\r\n\r\n"
288                )
289                .as_bytes(),
290            )
291            .await
292            .unwrap();
293
294        BufReader::new(reader.reunite(writer).unwrap())
295    }
296
297    /// Read all available data from the SSE stream until `needle` is found
298    /// in the accumulated text, or timeout.
299    async fn read_until_contains(
300        reader: &mut BufReader<tokio::net::TcpStream>,
301        needle: &str,
302        dur: Duration,
303    ) -> String {
304        let mut accumulated = String::new();
305        let result = timeout(dur, async {
306            loop {
307                let mut line = String::new();
308                let n = reader.read_line(&mut line).await.unwrap();
309                if n == 0 {
310                    break;
311                }
312                accumulated.push_str(&line);
313                if accumulated.contains(needle) {
314                    break;
315                }
316            }
317        })
318        .await;
319        if result.is_err() {
320            panic!("timeout waiting for '{needle}' in SSE stream. Data so far:\n{accumulated}");
321        }
322        accumulated
323    }
324
325    /// Wait until the SSE handler has subscribed to the broadcast channel.
326    ///
327    /// Sending before a receiver exists makes `broadcast::Sender::send` return
328    /// `SendError`, so a fixed sleep is racy on slow runners. Poll the receiver
329    /// count instead.
330    async fn wait_for_subscriber(sender: &broadcast::Sender<Event>) {
331        let ready = timeout(Duration::from_secs(5), async {
332            while sender.receiver_count() == 0 {
333                sleep(Duration::from_millis(5)).await;
334            }
335        })
336        .await;
337        assert!(ready.is_ok(), "SSE handler did not subscribe in time");
338    }
339
340    #[tokio::test]
341    async fn sse_stream_receives_events() {
342        let state = test_state();
343        let (addr, sender, auth) = start_sse_server(state).await;
344        let mut reader = connect_sse(&addr, "", &auth).await;
345
346        wait_for_subscriber(&sender).await;
347
348        let run_id = Uuid::now_v7();
349        sender.send(sample_run_event(run_id)).unwrap();
350
351        let text =
352            read_until_contains(&mut reader, &run_id.to_string(), Duration::from_secs(5)).await;
353
354        assert!(text.contains("run_status_changed"));
355        assert!(text.contains(&run_id.to_string()));
356    }
357
358    #[tokio::test]
359    async fn sse_filters_by_run_id() {
360        let state = test_state();
361        let (addr, sender, auth) = start_sse_server(state).await;
362
363        let target_run = Uuid::now_v7();
364        let other_run = Uuid::now_v7();
365
366        let mut reader = connect_sse(&addr, &format!("?run_id={target_run}"), &auth).await;
367        wait_for_subscriber(&sender).await;
368
369        sender.send(sample_run_event(other_run)).unwrap();
370        sender.send(sample_run_event(target_run)).unwrap();
371
372        let text =
373            read_until_contains(&mut reader, &target_run.to_string(), Duration::from_secs(5)).await;
374
375        assert!(text.contains(&target_run.to_string()));
376        assert!(!text.contains(&other_run.to_string()));
377    }
378
379    #[tokio::test]
380    async fn sse_filters_by_event_type() {
381        let state = test_state();
382        let (addr, sender, auth) = start_sse_server(state).await;
383
384        let mut reader = connect_sse(&addr, "?types=user_signed_in", &auth).await;
385        wait_for_subscriber(&sender).await;
386
387        let run_id = Uuid::now_v7();
388        sender.send(sample_run_event(run_id)).unwrap();
389        sender.send(sample_user_event()).unwrap();
390
391        let text = read_until_contains(&mut reader, "user_signed_in", Duration::from_secs(5)).await;
392
393        assert!(text.contains("user_signed_in"));
394        assert!(!text.contains("run_status_changed"));
395    }
396
397    #[tokio::test]
398    async fn sse_returns_correct_content_type() {
399        let state = test_state();
400        let (addr, _sender, auth) = start_sse_server(state).await;
401        let mut reader = connect_sse(&addr, "", &auth).await;
402
403        let text =
404            read_until_contains(&mut reader, "text/event-stream", Duration::from_secs(5)).await;
405
406        assert!(text.contains("text/event-stream"));
407    }
408
409    #[tokio::test]
410    async fn sse_rejects_unauthenticated() {
411        let state = test_state();
412        let (addr, _sender, _auth) = start_sse_server(state).await;
413        // Connect without auth header
414        let stream = tokio::net::TcpStream::connect(&addr).await.unwrap();
415        let (reader, mut writer) = stream.into_split();
416
417        use tokio::io::AsyncWriteExt;
418        writer
419            .write_all(
420                format!(
421                    "GET /events HTTP/1.1\r\nHost: {addr}\r\nAccept: text/event-stream\r\n\r\n"
422                )
423                .as_bytes(),
424            )
425            .await
426            .unwrap();
427
428        let mut buf_reader = BufReader::new(reader.reunite(writer).unwrap());
429        let text = read_until_contains(&mut buf_reader, "401", Duration::from_secs(5)).await;
430
431        assert!(text.contains("401"));
432    }
433
434    /// Stream a provider account event, then a run event, to a client
435    /// authenticated with `auth`, and assert only the run event arrives.
436    ///
437    /// Both events go through the same channel in order: had the provider
438    /// account event not been filtered out, it would arrive first.
439    async fn assert_provider_account_events_hidden(state: AppState, auth: &str) {
440        let (addr, sender, _jwt) = start_sse_server(state).await;
441        let mut reader = connect_sse(&addr, "", auth).await;
442        wait_for_subscriber(&sender).await;
443
444        let run_id = Uuid::now_v7();
445        sender.send(sample_provider_account_event()).unwrap();
446        sender.send(sample_run_event(run_id)).unwrap();
447
448        let text =
449            read_until_contains(&mut reader, &run_id.to_string(), Duration::from_secs(5)).await;
450
451        assert!(!text.contains("provider_account"));
452    }
453
454    #[tokio::test]
455    async fn sse_scope_hides_provider_account_events_from_a_runs_read_key() {
456        let state = test_state();
457        let auth = api_key_header(&state, true, vec![ApiKeyScope::RunsRead]).await;
458        assert_provider_account_events_hidden(state, &auth).await;
459    }
460
461    #[tokio::test]
462    async fn sse_scope_hides_provider_account_events_from_an_admin_scope_key_of_a_member() {
463        let state = test_state();
464        let auth = api_key_header(&state, false, vec![ApiKeyScope::Admin]).await;
465        assert_provider_account_events_hidden(state, &auth).await;
466    }
467
468    #[tokio::test]
469    async fn sse_scope_streams_provider_account_events_to_an_admin_scope_key() {
470        let state = test_state();
471        let auth = api_key_header(&state, true, vec![ApiKeyScope::Admin]).await;
472        let (addr, sender, _jwt) = start_sse_server(state).await;
473        let mut reader = connect_sse(&addr, "", &auth).await;
474        wait_for_subscriber(&sender).await;
475
476        sender.send(sample_provider_account_event()).unwrap();
477
478        let text =
479            read_until_contains(&mut reader, "provider_account", Duration::from_secs(5)).await;
480
481        assert!(text.contains("provider_account"));
482    }
483
484    #[tokio::test]
485    async fn sse_scope_rejects_a_key_without_runs_read() {
486        let state = test_state();
487        let auth = api_key_header(&state, true, vec![ApiKeyScope::StatsRead]).await;
488        let (addr, _sender, _jwt) = start_sse_server(state).await;
489        let mut reader = connect_sse(&addr, "", &auth).await;
490
491        let text = read_until_contains(&mut reader, "\r\n", Duration::from_secs(5)).await;
492
493        assert!(text.starts_with("HTTP/1.1 403"), "status line: {text}");
494    }
495}
ironflow_api/routes/events.rs

ironflow_api/routes/
events.rs
