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

ironflow_api/routes/
get_run_logs.rs
