ironflow_api/routes/
run_events.rs

1//! SSE endpoint for per-run workflow event streaming.
2
3use std::convert::Infallible;
4use std::pin::Pin;
5use std::time::Duration;
6
7use axum::extract::{Path, 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::error::ApiError;
16use crate::state::AppState;
17use ironflow_auth::extractor::Authenticated;
18use ironflow_engine::notify::WorkflowEvent;
19
20/// Deserialize a comma-separated string into `Option<Vec<String>>`.
21fn deserialize_comma_strings<'de, D>(deserializer: D) -> Result<Option<Vec<String>>, D::Error>
22where
23    D: Deserializer<'de>,
24{
25    let opt: Option<String> = Option::deserialize(deserializer)?;
26    match opt {
27        None => Ok(None),
28        Some(raw) => {
29            let all_types = [
30                WorkflowEvent::STEP_STARTED,
31                WorkflowEvent::STEP_COMPLETED,
32                WorkflowEvent::STEP_FAILED,
33                WorkflowEvent::APPROVAL_REQUIRED,
34                WorkflowEvent::AGENT_STEP_TOKENS_USED,
35            ];
36
37            let kinds: Vec<String> = raw
38                .split(',')
39                .map(|s| s.trim())
40                .filter(|s| !s.is_empty())
41                .map(|s| {
42                    if all_types.contains(&s) {
43                        Ok(s.to_string())
44                    } else {
45                        Err(de::Error::custom(format!(
46                            "unknown workflow event type: {s}"
47                        )))
48                    }
49                })
50                .collect::<Result<Vec<_>, _>>()?;
51
52            Ok(Some(kinds))
53        }
54    }
55}
56
57/// Query parameters for the per-run SSE events endpoint.
58///
59/// # Examples
60///
61/// ```
62/// use ironflow_api::routes::run_events::RunEventsQuery;
63///
64/// let query = RunEventsQuery { types: None };
65/// ```
66#[derive(Debug, Deserialize)]
67pub struct RunEventsQuery {
68    /// Comma-separated list of workflow event types to include
69    /// (e.g. `?types=step_started,step_completed`).
70    #[serde(default, deserialize_with = "deserialize_comma_strings")]
71    pub types: Option<Vec<String>>,
72}
73
74/// `GET /api/v1/runs/{id}/events` -- per-run Server-Sent Events stream.
75///
76/// Streams [`WorkflowEvent`]s for a specific workflow run in real time.
77/// Supports optional filtering via `?types=step_started,step_completed`.
78///
79/// Each SSE message has:
80/// - `event:` set to the event type (e.g. `step_started`)
81/// - `data:` JSON-serialized event payload
82///
83/// A keep-alive comment is sent every 30 seconds.
84///
85/// # Errors
86///
87/// Returns 401 if the request is not authenticated.
88/// Returns 404 if the run does not exist.
89#[cfg_attr(
90    feature = "openapi",
91    utoipa::path(
92        get,
93        path = "/api/v1/runs/{id}/events",
94        tags = ["runs"],
95        params(
96            ("id" = Uuid, Path, description = "Run ID"),
97            ("types" = Option<String>, Query, description = "Comma-separated workflow event types to filter (e.g. step_started,step_completed)")
98        ),
99        responses(
100            (status = 200, description = "SSE stream of workflow events"),
101            (status = 401, description = "Unauthorized"),
102            (status = 404, description = "Run not found")
103        ),
104        security(("Bearer" = []))
105    )
106)]
107pub async fn run_events(
108    _auth: Authenticated,
109    State(state): State<AppState>,
110    Path(id): Path<Uuid>,
111    Query(query): Query<RunEventsQuery>,
112) -> Result<Sse<impl Stream<Item = Result<SseEvent, Infallible>>>, ApiError> {
113    state.get_run_or_404(id).await?;
114
115    let type_filter = query.types;
116
117    let stream: Pin<Box<dyn Stream<Item = Result<SseEvent, Infallible>> + Send>> = match state
118        .event_bus
119    {
120        Some(ref bus) => {
121            let receiver = bus.subscribe(id);
122
123            Box::pin(BroadcastStream::new(receiver).filter_map(
124                move |result: Result<WorkflowEvent, _>| {
125                    let type_filter = type_filter.clone();
126                    async move {
127                        let event = result.ok()?;
128
129                        if let Some(ref kinds) = type_filter {
130                            let event_type = event.event_type();
131                            if !kinds.iter().any(|k| k == event_type) {
132                                return None;
133                            }
134                        }
135
136                        let data = serde_json::to_string(&event).ok()?;
137                        let sse_event = SseEvent::default().event(event.event_type()).data(data);
138
139                        Some(Ok::<_, Infallible>(sse_event))
140                    }
141                },
142            ))
143        }
144        None => Box::pin(futures_util::stream::empty()),
145    };
146
147    Ok(Sse::new(stream).keep_alive(KeepAlive::new().interval(Duration::from_secs(30))))
148}
149
150#[cfg(test)]
151mod tests {
152    use std::collections::HashMap;
153    use std::sync::Arc;
154    use std::time::Duration;
155
156    use axum::Router;
157    use axum::routing::get;
158    use chrono::Utc;
159    use ironflow_auth::jwt::AccessToken;
160    use ironflow_core::providers::claude::ClaudeCodeProvider;
161    use ironflow_engine::engine::Engine;
162    use ironflow_engine::notify::{
163        Event, WorkflowEvent, WorkflowEventBus, WorkflowStepCompletedEvent,
164        WorkflowStepStartedEvent,
165    };
166    use ironflow_store::memory::InMemoryStore;
167    use ironflow_store::models::{NewRun, TriggerKind};
168    use serde_json::json;
169    use tokio::io::AsyncBufReadExt;
170    use tokio::io::BufReader;
171    use tokio::net::TcpListener;
172    use tokio::sync::broadcast;
173    use tokio::time::{sleep, timeout};
174    use uuid::Uuid;
175
176    use super::run_events;
177    use crate::state::AppState;
178
179    fn test_state_with_bus() -> (AppState, WorkflowEventBus) {
180        let store = Arc::new(InMemoryStore::new());
181        let provider = Arc::new(ClaudeCodeProvider::new());
182        let engine = Arc::new(Engine::new(store.clone(), provider));
183        let jwt_config = Arc::new(ironflow_auth::jwt::JwtConfig {
184            secret: "test-secret".to_string(),
185            access_token_ttl_secs: 900,
186            refresh_token_ttl_secs: 604800,
187            cookie_domain: None,
188            cookie_secure: false,
189        });
190        let (event_sender, _) = broadcast::channel::<Event>(16);
191        let bus = WorkflowEventBus::new();
192        let state = AppState::new(
193            store,
194            engine,
195            jwt_config,
196            "test-worker-token".to_string(),
197            event_sender,
198        )
199        .with_event_bus(bus.clone());
200        (state, bus)
201    }
202
203    fn make_auth_token(state: &AppState) -> String {
204        let user_id = Uuid::now_v7();
205        let token = AccessToken::for_user(user_id, "testuser", false, &state.jwt_config).unwrap();
206        format!("Bearer {}", token.0)
207    }
208
209    async fn create_run(state: &AppState) -> Uuid {
210        state
211            .store
212            .create_run(NewRun {
213                created_by: None,
214                workflow_name: "test".to_string(),
215                trigger: TriggerKind::Manual,
216                payload: json!({}),
217                max_retries: 0,
218                handler_version: None,
219                labels: HashMap::new(),
220                scheduled_at: None,
221                idempotency_key: None,
222                max_cost_usd: None,
223            })
224            .await
225            .unwrap()
226            .into_run()
227            .id
228    }
229
230    async fn start_sse_server(state: AppState) -> (String, String) {
231        let auth = make_auth_token(&state);
232        let app = Router::new()
233            .route("/{id}/events", get(run_events))
234            .with_state(state);
235
236        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
237        let addr = listener.local_addr().unwrap().to_string();
238        tokio::spawn(async move {
239            axum::serve(listener, app).await.unwrap();
240        });
241        (addr, auth)
242    }
243
244    async fn connect_sse(addr: &str, path: &str, auth: &str) -> BufReader<tokio::net::TcpStream> {
245        let stream = tokio::net::TcpStream::connect(addr).await.unwrap();
246        let (reader, mut writer) = stream.into_split();
247
248        use tokio::io::AsyncWriteExt;
249        writer
250            .write_all(
251                format!(
252                    "GET {path} HTTP/1.1\r\nHost: {addr}\r\nAccept: text/event-stream\r\nAuthorization: {auth}\r\n\r\n"
253                )
254                .as_bytes(),
255            )
256            .await
257            .unwrap();
258
259        BufReader::new(reader.reunite(writer).unwrap())
260    }
261
262    async fn read_until_contains(
263        reader: &mut BufReader<tokio::net::TcpStream>,
264        needle: &str,
265        dur: Duration,
266    ) -> String {
267        let mut accumulated = String::new();
268        let result = timeout(dur, async {
269            loop {
270                let mut line = String::new();
271                let n = reader.read_line(&mut line).await.unwrap();
272                if n == 0 {
273                    break;
274                }
275                accumulated.push_str(&line);
276                if accumulated.contains(needle) {
277                    break;
278                }
279            }
280        })
281        .await;
282        if result.is_err() {
283            panic!("timeout waiting for '{needle}' in SSE stream. Data so far:\n{accumulated}");
284        }
285        accumulated
286    }
287
288    #[tokio::test]
289    async fn sse_stream_receives_workflow_events() {
290        let (state, bus) = test_state_with_bus();
291        let run_id = create_run(&state).await;
292        let (addr, auth) = start_sse_server(state).await;
293
294        let mut reader = connect_sse(&addr, &format!("/{run_id}/events"), &auth).await;
295        sleep(Duration::from_millis(50)).await;
296
297        bus.publish(
298            run_id,
299            WorkflowEvent::StepStarted(WorkflowStepStartedEvent {
300                step_name: "build".to_string(),
301                step_index: 0,
302                timestamp: Utc::now(),
303            }),
304        );
305
306        let text = read_until_contains(&mut reader, "build", Duration::from_secs(5)).await;
307
308        assert!(text.contains("event: step_started"));
309        assert!(text.contains("build"));
310    }
311
312    #[tokio::test]
313    async fn returns_404_for_unknown_run() {
314        let (state, _bus) = test_state_with_bus();
315        let (addr, auth) = start_sse_server(state).await;
316
317        let unknown = Uuid::nil();
318        let mut reader = connect_sse(&addr, &format!("/{unknown}/events"), &auth).await;
319
320        let text = read_until_contains(&mut reader, "404", Duration::from_secs(5)).await;
321        assert!(text.contains("404"));
322    }
323
324    #[tokio::test]
325    async fn rejects_unauthenticated() {
326        let (state, _bus) = test_state_with_bus();
327        let run_id = create_run(&state).await;
328        let (addr, _auth) = start_sse_server(state).await;
329
330        let stream = tokio::net::TcpStream::connect(&addr).await.unwrap();
331        let (reader, mut writer) = stream.into_split();
332
333        use tokio::io::AsyncWriteExt;
334        writer
335            .write_all(
336                format!(
337                    "GET /{run_id}/events HTTP/1.1\r\nHost: {addr}\r\nAccept: text/event-stream\r\n\r\n"
338                )
339                .as_bytes(),
340            )
341            .await
342            .unwrap();
343
344        let mut buf_reader = BufReader::new(reader.reunite(writer).unwrap());
345        let text = read_until_contains(&mut buf_reader, "401", Duration::from_secs(5)).await;
346        assert!(text.contains("401"));
347    }
348
349    #[tokio::test]
350    async fn filters_by_event_type() {
351        let (state, bus) = test_state_with_bus();
352        let run_id = create_run(&state).await;
353        let (addr, auth) = start_sse_server(state).await;
354
355        let mut reader = connect_sse(
356            &addr,
357            &format!("/{run_id}/events?types=step_completed"),
358            &auth,
359        )
360        .await;
361        sleep(Duration::from_millis(50)).await;
362
363        bus.publish(
364            run_id,
365            WorkflowEvent::StepStarted(WorkflowStepStartedEvent {
366                step_name: "build".to_string(),
367                step_index: 0,
368                timestamp: Utc::now(),
369            }),
370        );
371        bus.publish(
372            run_id,
373            WorkflowEvent::StepCompleted(WorkflowStepCompletedEvent {
374                step_name: "build".to_string(),
375                step_index: 0,
376                duration_ms: 1234,
377                output_summary: None,
378            }),
379        );
380
381        let text = read_until_contains(&mut reader, "step_completed", Duration::from_secs(5)).await;
382
383        assert!(text.contains("step_completed"));
384        assert!(!text.contains("event: step_started"));
385    }
386
387    #[tokio::test]
388    async fn events_isolated_between_runs() {
389        let (state, bus) = test_state_with_bus();
390        let run_a = create_run(&state).await;
391        let run_b = create_run(&state).await;
392        let (addr, auth) = start_sse_server(state).await;
393
394        let mut reader_a = connect_sse(&addr, &format!("/{run_a}/events"), &auth).await;
395        sleep(Duration::from_millis(50)).await;
396
397        bus.publish(
398            run_b,
399            WorkflowEvent::StepStarted(WorkflowStepStartedEvent {
400                step_name: "only-for-b".to_string(),
401                step_index: 0,
402                timestamp: Utc::now(),
403            }),
404        );
405        bus.publish(
406            run_a,
407            WorkflowEvent::StepStarted(WorkflowStepStartedEvent {
408                step_name: "only-for-a".to_string(),
409                step_index: 0,
410                timestamp: Utc::now(),
411            }),
412        );
413
414        let text = read_until_contains(&mut reader_a, "only-for-a", Duration::from_secs(5)).await;
415
416        assert!(text.contains("only-for-a"));
417        assert!(!text.contains("only-for-b"));
418    }
419}
ironflow_api/routes/run_events.rs

ironflow_api/routes/
run_events.rs
