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

ironflow_api/routes/
run_events.rs
