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::{Event, WorkflowEvent, WorkflowEventBus};
163    use ironflow_store::memory::InMemoryStore;
164    use ironflow_store::models::{NewRun, TriggerKind};
165    use serde_json::json;
166    use tokio::io::AsyncBufReadExt;
167    use tokio::io::BufReader;
168    use tokio::net::TcpListener;
169    use tokio::sync::broadcast;
170    use tokio::time::{sleep, timeout};
171    use uuid::Uuid;
172
173    use super::run_events;
174    use crate::state::AppState;
175
176    fn test_state_with_bus() -> (AppState, WorkflowEventBus) {
177        let store = Arc::new(InMemoryStore::new());
178        let provider = Arc::new(ClaudeCodeProvider::new());
179        let engine = Arc::new(Engine::new(store.clone(), provider));
180        let jwt_config = Arc::new(ironflow_auth::jwt::JwtConfig {
181            secret: "test-secret".to_string(),
182            access_token_ttl_secs: 900,
183            refresh_token_ttl_secs: 604800,
184            cookie_domain: None,
185            cookie_secure: false,
186        });
187        let (event_sender, _) = broadcast::channel::<Event>(16);
188        let bus = WorkflowEventBus::new();
189        let state = AppState::new(
190            store,
191            engine,
192            jwt_config,
193            "test-worker-token".to_string(),
194            event_sender,
195        )
196        .with_event_bus(bus.clone());
197        (state, bus)
198    }
199
200    fn make_auth_token(state: &AppState) -> String {
201        let user_id = Uuid::now_v7();
202        let token = AccessToken::for_user(user_id, "testuser", false, &state.jwt_config).unwrap();
203        format!("Bearer {}", token.0)
204    }
205
206    async fn create_run(state: &AppState) -> Uuid {
207        state
208            .store
209            .create_run(NewRun {
210                created_by: None,
211                workflow_name: "test".to_string(),
212                trigger: TriggerKind::Manual,
213                payload: json!({}),
214                max_retries: 0,
215                handler_version: None,
216                labels: HashMap::new(),
217                scheduled_at: None,
218                idempotency_key: None,
219                max_cost_usd: None,
220            })
221            .await
222            .unwrap()
223            .into_run()
224            .id
225    }
226
227    async fn start_sse_server(state: AppState) -> (String, String) {
228        let auth = make_auth_token(&state);
229        let app = Router::new()
230            .route("/{id}/events", get(run_events))
231            .with_state(state);
232
233        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
234        let addr = listener.local_addr().unwrap().to_string();
235        tokio::spawn(async move {
236            axum::serve(listener, app).await.unwrap();
237        });
238        (addr, auth)
239    }
240
241    async fn connect_sse(addr: &str, path: &str, auth: &str) -> BufReader<tokio::net::TcpStream> {
242        let stream = tokio::net::TcpStream::connect(addr).await.unwrap();
243        let (reader, mut writer) = stream.into_split();
244
245        use tokio::io::AsyncWriteExt;
246        writer
247            .write_all(
248                format!(
249                    "GET {path} HTTP/1.1\r\nHost: {addr}\r\nAccept: text/event-stream\r\nAuthorization: {auth}\r\n\r\n"
250                )
251                .as_bytes(),
252            )
253            .await
254            .unwrap();
255
256        BufReader::new(reader.reunite(writer).unwrap())
257    }
258
259    async fn read_until_contains(
260        reader: &mut BufReader<tokio::net::TcpStream>,
261        needle: &str,
262        dur: Duration,
263    ) -> String {
264        let mut accumulated = String::new();
265        let result = timeout(dur, async {
266            loop {
267                let mut line = String::new();
268                let n = reader.read_line(&mut line).await.unwrap();
269                if n == 0 {
270                    break;
271                }
272                accumulated.push_str(&line);
273                if accumulated.contains(needle) {
274                    break;
275                }
276            }
277        })
278        .await;
279        if result.is_err() {
280            panic!("timeout waiting for '{needle}' in SSE stream. Data so far:\n{accumulated}");
281        }
282        accumulated
283    }
284
285    #[tokio::test]
286    async fn sse_stream_receives_workflow_events() {
287        let (state, bus) = test_state_with_bus();
288        let run_id = create_run(&state).await;
289        let (addr, auth) = start_sse_server(state).await;
290
291        let mut reader = connect_sse(&addr, &format!("/{run_id}/events"), &auth).await;
292        sleep(Duration::from_millis(50)).await;
293
294        bus.publish(
295            run_id,
296            WorkflowEvent::StepStarted {
297                step_name: "build".to_string(),
298                step_index: 0,
299                timestamp: Utc::now(),
300            },
301        );
302
303        let text = read_until_contains(&mut reader, "build", Duration::from_secs(5)).await;
304
305        assert!(text.contains("event: step_started"));
306        assert!(text.contains("build"));
307    }
308
309    #[tokio::test]
310    async fn returns_404_for_unknown_run() {
311        let (state, _bus) = test_state_with_bus();
312        let (addr, auth) = start_sse_server(state).await;
313
314        let unknown = Uuid::nil();
315        let mut reader = connect_sse(&addr, &format!("/{unknown}/events"), &auth).await;
316
317        let text = read_until_contains(&mut reader, "404", Duration::from_secs(5)).await;
318        assert!(text.contains("404"));
319    }
320
321    #[tokio::test]
322    async fn rejects_unauthenticated() {
323        let (state, _bus) = test_state_with_bus();
324        let run_id = create_run(&state).await;
325        let (addr, _auth) = start_sse_server(state).await;
326
327        let stream = tokio::net::TcpStream::connect(&addr).await.unwrap();
328        let (reader, mut writer) = stream.into_split();
329
330        use tokio::io::AsyncWriteExt;
331        writer
332            .write_all(
333                format!(
334                    "GET /{run_id}/events HTTP/1.1\r\nHost: {addr}\r\nAccept: text/event-stream\r\n\r\n"
335                )
336                .as_bytes(),
337            )
338            .await
339            .unwrap();
340
341        let mut buf_reader = BufReader::new(reader.reunite(writer).unwrap());
342        let text = read_until_contains(&mut buf_reader, "401", Duration::from_secs(5)).await;
343        assert!(text.contains("401"));
344    }
345
346    #[tokio::test]
347    async fn filters_by_event_type() {
348        let (state, bus) = test_state_with_bus();
349        let run_id = create_run(&state).await;
350        let (addr, auth) = start_sse_server(state).await;
351
352        let mut reader = connect_sse(
353            &addr,
354            &format!("/{run_id}/events?types=step_completed"),
355            &auth,
356        )
357        .await;
358        sleep(Duration::from_millis(50)).await;
359
360        bus.publish(
361            run_id,
362            WorkflowEvent::StepStarted {
363                step_name: "build".to_string(),
364                step_index: 0,
365                timestamp: Utc::now(),
366            },
367        );
368        bus.publish(
369            run_id,
370            WorkflowEvent::StepCompleted {
371                step_name: "build".to_string(),
372                step_index: 0,
373                duration_ms: 1234,
374                output_summary: None,
375            },
376        );
377
378        let text = read_until_contains(&mut reader, "step_completed", Duration::from_secs(5)).await;
379
380        assert!(text.contains("step_completed"));
381        assert!(!text.contains("event: step_started"));
382    }
383
384    #[tokio::test]
385    async fn events_isolated_between_runs() {
386        let (state, bus) = test_state_with_bus();
387        let run_a = create_run(&state).await;
388        let run_b = create_run(&state).await;
389        let (addr, auth) = start_sse_server(state).await;
390
391        let mut reader_a = connect_sse(&addr, &format!("/{run_a}/events"), &auth).await;
392        sleep(Duration::from_millis(50)).await;
393
394        bus.publish(
395            run_b,
396            WorkflowEvent::StepStarted {
397                step_name: "only-for-b".to_string(),
398                step_index: 0,
399                timestamp: Utc::now(),
400            },
401        );
402        bus.publish(
403            run_a,
404            WorkflowEvent::StepStarted {
405                step_name: "only-for-a".to_string(),
406                step_index: 0,
407                timestamp: Utc::now(),
408            },
409        );
410
411        let text = read_until_contains(&mut reader_a, "only-for-a", Duration::from_secs(5)).await;
412
413        assert!(text.contains("only-for-a"));
414        assert!(!text.contains("only-for-b"));
415    }
416}
ironflow_api/routes/run_events.rs

ironflow_api/routes/
run_events.rs
