ironflow_api/routes/
get_run_logs.rs

1//! `GET /api/v1/runs/:id/logs` -- Retrieve persisted log lines for a run.
2
3use std::collections::HashMap;
4
5use axum::Json;
6use axum::extract::{Path, Query, State};
7use axum::response::IntoResponse;
8use serde::{Deserialize, Serialize};
9use serde_json::Value;
10use uuid::Uuid;
11
12use ironflow_auth::extractor::Authenticated;
13#[cfg(feature = "openapi")]
14use ironflow_store::entities::LogEntry;
15use ironflow_store::entities::{LogFilter, LogStream};
16use ironflow_types::{ApiMeta, ApiResponse};
17
18use crate::error::ApiError;
19use crate::state::AppState;
20
21const MAX_LIMIT: u32 = 1000;
22const DEFAULT_LIMIT: u32 = 100;
23
24/// Query parameters for listing run logs.
25#[derive(Debug, Deserialize)]
26#[cfg_attr(feature = "openapi", derive(utoipa::IntoParams, utoipa::ToSchema))]
27pub struct GetRunLogsQuery {
28    /// Filter by step ID.
29    pub step_id: Option<Uuid>,
30    /// Filter by output stream (`stdout`, `stderr`, `system`).
31    pub stream: Option<LogStream>,
32    /// Cursor for pagination (last entry ID from previous page).
33    pub cursor: Option<Uuid>,
34    /// Number of entries to return (default: 100, max: 1000).
35    pub limit: Option<u32>,
36}
37
38/// Cursor-based pagination metadata for log entries.
39#[derive(Debug, Serialize)]
40#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
41pub struct LogCursorMeta {
42    /// Cursor to pass for the next page. `None` when there are no more entries.
43    pub next_cursor: Option<Uuid>,
44    /// Whether more entries exist after this page.
45    pub has_more: bool,
46}
47
48/// Retrieve persisted log lines for a run with cursor-based pagination.
49///
50/// Returns log entries ordered by time (UUID v7 ascending). Use the
51/// `cursor` query parameter with the last entry's `id` to fetch the
52/// next page.
53#[cfg_attr(
54    feature = "openapi",
55    utoipa::path(
56        get,
57        path = "/api/v1/runs/{id}/logs",
58        tags = ["runs"],
59        params(
60            ("id" = Uuid, Path, description = "Run ID"),
61            GetRunLogsQuery,
62        ),
63        responses(
64            (status = 200, description = "Log entries with cursor-based pagination", body = Vec<LogEntry>),
65            (status = 401, description = "Unauthorized"),
66            (status = 404, description = "Run not found")
67        ),
68        security(("Bearer" = []))
69    )
70)]
71pub async fn get_run_logs(
72    _auth: Authenticated,
73    State(state): State<AppState>,
74    Path(run_id): Path<Uuid>,
75    Query(params): Query<GetRunLogsQuery>,
76) -> Result<impl IntoResponse, ApiError> {
77    state.get_run_or_404(run_id).await?;
78
79    let limit = match params.limit {
80        Some(0) | None => DEFAULT_LIMIT,
81        Some(l) => l.min(MAX_LIMIT),
82    };
83
84    let filter = LogFilter {
85        step_id: params.step_id,
86        stream: params.stream,
87    };
88
89    let entries = state
90        .store
91        .get_logs(run_id, filter, params.cursor, limit + 1)
92        .await?;
93
94    let has_more = entries.len() > limit as usize;
95    let entries: Vec<_> = entries.into_iter().take(limit as usize).collect();
96    let next_cursor = if has_more {
97        entries.last().map(|e| e.id)
98    } else {
99        None
100    };
101
102    let cursor_meta = LogCursorMeta {
103        next_cursor,
104        has_more,
105    };
106    let extra = match serde_json::to_value(cursor_meta) {
107        Ok(Value::Object(map)) => map.into_iter().collect(),
108        _ => HashMap::new(),
109    };
110
111    Ok(Json(ApiResponse {
112        data: entries,
113        meta: Some(ApiMeta {
114            page: None,
115            per_page: None,
116            total: None,
117            extra,
118        }),
119    }))
120}
121
122#[cfg(test)]
123mod tests {
124    use std::collections::HashMap;
125    use std::sync::Arc;
126
127    use axum::Router;
128    use axum::body::Body;
129    use axum::http::{Request, StatusCode};
130    use axum::routing::get;
131    use http_body_util::BodyExt;
132    use serde_json::{Value as JsonValue, from_slice, json};
133    use tokio::sync::broadcast;
134    use tower::ServiceExt;
135    use uuid::Uuid;
136
137    use ironflow_auth::jwt::JwtConfig;
138    use ironflow_core::providers::claude::ClaudeCodeProvider;
139    use ironflow_engine::engine::Engine;
140    use ironflow_engine::notify::Event;
141    use ironflow_store::entities::{LogStream, NewLogEntries, NewRun, TriggerKind};
142    use ironflow_store::memory::InMemoryStore;
143
144    use super::*;
145    use crate::routes::test_helpers::create_user_auth_header;
146
147    fn test_state() -> AppState {
148        let store = Arc::new(InMemoryStore::new());
149        let provider = Arc::new(ClaudeCodeProvider::new());
150        let engine = Arc::new(Engine::new(store.clone(), provider));
151        let jwt_config = Arc::new(JwtConfig {
152            secret: "test-secret".to_string(),
153            access_token_ttl_secs: 900,
154            refresh_token_ttl_secs: 604800,
155            cookie_domain: None,
156            cookie_secure: false,
157        });
158        let (event_sender, _) = broadcast::channel::<Event>(1);
159        AppState::new(
160            store,
161            engine,
162            jwt_config,
163            "test-worker-token".to_string(),
164            event_sender,
165        )
166    }
167
168    async fn create_run(state: &AppState) -> Uuid {
169        state
170            .store
171            .create_run(NewRun {
172                created_by: None,
173                workflow_name: "test".to_string(),
174                trigger: TriggerKind::Manual,
175                payload: json!({}),
176                max_retries: 0,
177                handler_version: None,
178                labels: HashMap::new(),
179                scheduled_at: None,
180                idempotency_key: None,
181                concurrency_key: None,
182                concurrency_limits: Vec::new(),
183                max_cost_usd: None,
184                worker_tags: Vec::new(),
185            })
186            .await
187            .unwrap()
188            .into_run()
189            .id
190    }
191
192    async fn push_logs(state: &AppState, run_id: Uuid, step_id: Uuid, stream: LogStream, n: usize) {
193        state
194            .store
195            .append_logs(NewLogEntries {
196                ids: (0..n).map(|_| Uuid::now_v7()).collect(),
197                run_id,
198                step_id,
199                step_name: "build".to_string(),
200                stream,
201                lines: (0..n).map(|i| format!("line {i}")).collect(),
202            })
203            .await
204            .unwrap();
205    }
206
207    #[tokio::test]
208    async fn returns_persisted_logs() {
209        let state = test_state();
210        let auth_header = create_user_auth_header(&state, "testuser", false).await;
211        let run_id = create_run(&state).await;
212        let step_id = Uuid::now_v7();
213
214        push_logs(&state, run_id, step_id, LogStream::Stdout, 3).await;
215
216        let app = Router::new()
217            .route("/runs/{id}/logs", get(get_run_logs))
218            .with_state(state);
219
220        let req = Request::builder()
221            .uri(format!("/runs/{run_id}/logs"))
222            .header("authorization", auth_header)
223            .body(Body::empty())
224            .unwrap();
225
226        let resp = app.oneshot(req).await.unwrap();
227        assert_eq!(resp.status(), StatusCode::OK);
228
229        let body = resp.into_body().collect().await.unwrap().to_bytes();
230        let json_val: JsonValue = from_slice(&body).unwrap();
231        assert_eq!(json_val["data"].as_array().unwrap().len(), 3);
232        assert_eq!(json_val["data"][0]["line"], "line 0");
233        assert_eq!(json_val["data"][0]["stream"], "stdout");
234        assert_eq!(json_val["meta"]["has_more"], false);
235        assert!(json_val["meta"]["next_cursor"].is_null());
236    }
237
238    #[tokio::test]
239    async fn filters_by_step_id() {
240        let state = test_state();
241        let auth_header = create_user_auth_header(&state, "testuser", false).await;
242        let run_id = create_run(&state).await;
243        let step_a = Uuid::now_v7();
244        let step_b = Uuid::now_v7();
245
246        push_logs(&state, run_id, step_a, LogStream::Stdout, 2).await;
247        push_logs(&state, run_id, step_b, LogStream::Stdout, 3).await;
248
249        let app = Router::new()
250            .route("/runs/{id}/logs", get(get_run_logs))
251            .with_state(state);
252
253        let req = Request::builder()
254            .uri(format!("/runs/{run_id}/logs?step_id={step_a}"))
255            .header("authorization", auth_header)
256            .body(Body::empty())
257            .unwrap();
258
259        let resp = app.oneshot(req).await.unwrap();
260        assert_eq!(resp.status(), StatusCode::OK);
261
262        let body = resp.into_body().collect().await.unwrap().to_bytes();
263        let json_val: JsonValue = from_slice(&body).unwrap();
264        assert_eq!(json_val["data"].as_array().unwrap().len(), 2);
265    }
266
267    #[tokio::test]
268    async fn filters_by_stream() {
269        let state = test_state();
270        let auth_header = create_user_auth_header(&state, "testuser", false).await;
271        let run_id = create_run(&state).await;
272        let step_id = Uuid::now_v7();
273
274        push_logs(&state, run_id, step_id, LogStream::Stdout, 2).await;
275        push_logs(&state, run_id, step_id, LogStream::Stderr, 1).await;
276
277        let app = Router::new()
278            .route("/runs/{id}/logs", get(get_run_logs))
279            .with_state(state);
280
281        let req = Request::builder()
282            .uri(format!("/runs/{run_id}/logs?stream=stderr"))
283            .header("authorization", auth_header)
284            .body(Body::empty())
285            .unwrap();
286
287        let resp = app.oneshot(req).await.unwrap();
288        let body = resp.into_body().collect().await.unwrap().to_bytes();
289        let json_val: JsonValue = from_slice(&body).unwrap();
290        assert_eq!(json_val["data"].as_array().unwrap().len(), 1);
291        assert_eq!(json_val["data"][0]["stream"], "stderr");
292    }
293
294    #[tokio::test]
295    async fn cursor_based_pagination() {
296        let state = test_state();
297        let auth_header = create_user_auth_header(&state, "testuser", false).await;
298        let run_id = create_run(&state).await;
299        let step_id = Uuid::now_v7();
300
301        push_logs(&state, run_id, step_id, LogStream::Stdout, 5).await;
302
303        let app = Router::new()
304            .route("/runs/{id}/logs", get(get_run_logs))
305            .with_state(state);
306
307        let req = Request::builder()
308            .uri(format!("/runs/{run_id}/logs?limit=2"))
309            .header("authorization", &auth_header)
310            .body(Body::empty())
311            .unwrap();
312
313        let resp = app.clone().oneshot(req).await.unwrap();
314        let body = resp.into_body().collect().await.unwrap().to_bytes();
315        let page1: JsonValue = from_slice(&body).unwrap();
316        assert_eq!(page1["data"].as_array().unwrap().len(), 2);
317        assert_eq!(page1["meta"]["has_more"], true);
318
319        let cursor = page1["meta"]["next_cursor"].as_str().unwrap();
320
321        let req = Request::builder()
322            .uri(format!("/runs/{run_id}/logs?limit=2&cursor={cursor}"))
323            .header("authorization", &auth_header)
324            .body(Body::empty())
325            .unwrap();
326
327        let resp = app.clone().oneshot(req).await.unwrap();
328        let body = resp.into_body().collect().await.unwrap().to_bytes();
329        let page2: JsonValue = from_slice(&body).unwrap();
330        assert_eq!(page2["data"].as_array().unwrap().len(), 2);
331        assert_eq!(page2["data"][0]["line"], "line 2");
332        assert_eq!(page2["meta"]["has_more"], true);
333
334        let cursor = page2["meta"]["next_cursor"].as_str().unwrap();
335
336        let req = Request::builder()
337            .uri(format!("/runs/{run_id}/logs?limit=2&cursor={cursor}"))
338            .header("authorization", &auth_header)
339            .body(Body::empty())
340            .unwrap();
341
342        let resp = app.oneshot(req).await.unwrap();
343        let body = resp.into_body().collect().await.unwrap().to_bytes();
344        let page3: JsonValue = from_slice(&body).unwrap();
345        assert_eq!(page3["data"].as_array().unwrap().len(), 1);
346        assert_eq!(page3["meta"]["has_more"], false);
347        assert!(page3["meta"]["next_cursor"].is_null());
348    }
349
350    #[tokio::test]
351    async fn run_not_found_returns_404() {
352        let state = test_state();
353        let auth_header = create_user_auth_header(&state, "testuser", false).await;
354        let app = Router::new()
355            .route("/runs/{id}/logs", get(get_run_logs))
356            .with_state(state);
357
358        let req = Request::builder()
359            .uri(format!("/runs/{}/logs", Uuid::now_v7()))
360            .header("authorization", auth_header)
361            .body(Body::empty())
362            .unwrap();
363
364        let resp = app.oneshot(req).await.unwrap();
365        assert_eq!(resp.status(), StatusCode::NOT_FOUND);
366    }
367
368    #[tokio::test]
369    async fn unauthenticated_returns_401() {
370        let state = test_state();
371        let run_id = create_run(&state).await;
372        let app = Router::new()
373            .route("/runs/{id}/logs", get(get_run_logs))
374            .with_state(state);
375
376        let req = Request::builder()
377            .uri(format!("/runs/{run_id}/logs"))
378            .body(Body::empty())
379            .unwrap();
380
381        let resp = app.oneshot(req).await.unwrap();
382        assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
383    }
384
385    #[tokio::test]
386    async fn limit_capped_at_1000() {
387        let state = test_state();
388        let auth_header = create_user_auth_header(&state, "testuser", false).await;
389        let run_id = create_run(&state).await;
390        let step_id = Uuid::now_v7();
391
392        push_logs(&state, run_id, step_id, LogStream::Stdout, 2).await;
393
394        let app = Router::new()
395            .route("/runs/{id}/logs", get(get_run_logs))
396            .with_state(state);
397
398        let req = Request::builder()
399            .uri(format!("/runs/{run_id}/logs?limit=5000"))
400            .header("authorization", auth_header)
401            .body(Body::empty())
402            .unwrap();
403
404        let resp = app.oneshot(req).await.unwrap();
405        assert_eq!(resp.status(), StatusCode::OK);
406
407        let body = resp.into_body().collect().await.unwrap().to_bytes();
408        let json_val: JsonValue = from_slice(&body).unwrap();
409        assert_eq!(json_val["data"].as_array().unwrap().len(), 2);
410    }
411
412    #[tokio::test]
413    async fn empty_logs_returns_empty_array() {
414        let state = test_state();
415        let auth_header = create_user_auth_header(&state, "testuser", false).await;
416        let run_id = create_run(&state).await;
417
418        let app = Router::new()
419            .route("/runs/{id}/logs", get(get_run_logs))
420            .with_state(state);
421
422        let req = Request::builder()
423            .uri(format!("/runs/{run_id}/logs"))
424            .header("authorization", auth_header)
425            .body(Body::empty())
426            .unwrap();
427
428        let resp = app.oneshot(req).await.unwrap();
429        assert_eq!(resp.status(), StatusCode::OK);
430
431        let body = resp.into_body().collect().await.unwrap().to_bytes();
432        let json_val: JsonValue = from_slice(&body).unwrap();
433        assert_eq!(json_val["data"].as_array().unwrap().len(), 0);
434        assert_eq!(json_val["meta"]["has_more"], false);
435    }
436}
ironflow_api/routes/get_run_logs.rs

ironflow_api/routes/
get_run_logs.rs
