ironflow_api/routes/
list_workflows.rs

1//! `GET /api/v1/workflows` — List registered workflows.
2
3use std::collections::HashMap;
4
5use axum::extract::{Query, State};
6use axum::response::IntoResponse;
7use chrono::{DateTime, Utc};
8use ironflow_auth::extractor::Authenticated;
9use ironflow_store::entities::ApiKeyScope;
10use serde::{Deserialize, Serialize};
11
12use crate::authorization::AuthorizeExt;
13use crate::error::ApiError;
14use crate::response::ok;
15use crate::state::AppState;
16
17/// Query parameters for listing workflows.
18#[cfg_attr(feature = "openapi", derive(utoipa::IntoParams, utoipa::ToSchema))]
19#[derive(Debug, Deserialize)]
20pub struct ListWorkflowsQuery {
21    /// Optional case-insensitive partial match on workflow name.
22    pub name: Option<String>,
23    /// Optional case-insensitive partial match on the category path
24    /// (e.g. `etl` matches `Data/ETL` and `data/etl/nightly`).
25    ///
26    /// Pass `__uncategorized__` to list only workflows without any
27    /// category.
28    pub category: Option<String>,
29}
30
31/// Summary entry returned by `GET /api/v1/workflows`.
32#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
33#[derive(Debug, Serialize, Deserialize)]
34pub struct WorkflowSummary {
35    /// Workflow name (unique identifier).
36    pub name: String,
37    /// Optional `/`-separated category path.
38    pub category: Option<String>,
39    /// Current handler version.
40    pub version: Option<String>,
41    /// Optional 6-field cron expression for automatic execution.
42    #[serde(skip_serializing_if = "Option::is_none")]
43    pub schedule: Option<String>,
44    /// When the workflow was paused: its queued runs are not picked up until
45    /// it is resumed. Omitted when the workflow is not paused.
46    #[serde(default, skip_serializing_if = "Option::is_none")]
47    pub paused_at: Option<DateTime<Utc>>,
48}
49
50/// Sentinel value for the `category` query parameter that selects only
51/// uncategorized workflows.
52pub const UNCATEGORIZED_FILTER: &str = "__uncategorized__";
53
54/// List registered workflows, optionally filtered by name and category.
55///
56/// # Query Parameters
57///
58/// - `name` — Case-insensitive partial match on workflow name (optional)
59/// - `category` — Case-insensitive partial match on category path, or
60///   [`UNCATEGORIZED_FILTER`] to filter only uncategorized workflows
61///   (optional)
62#[cfg_attr(
63    feature = "openapi",
64    utoipa::path(
65        get,
66        path = "/api/v1/workflows",
67        tags = ["workflows"],
68        params(ListWorkflowsQuery),
69        responses(
70            (status = 200, description = "List of workflow summaries", body = [WorkflowSummary]),
71            (status = 401, description = "Unauthorized")
72        ),
73        security(("Bearer" = []))
74    )
75)]
76pub async fn list_workflows(
77    auth: Authenticated,
78    State(state): State<AppState>,
79    Query(params): Query<ListWorkflowsQuery>,
80) -> Result<impl IntoResponse, ApiError> {
81    auth.require(ApiKeyScope::WorkflowsRead, false)?;
82    let paused_at: HashMap<String, DateTime<Utc>> = state
83        .store
84        .list_workflow_pauses()
85        .await?
86        .into_iter()
87        .map(|pause| (pause.workflow_name, pause.paused_at))
88        .collect();
89
90    let mut summaries: Vec<WorkflowSummary> = state
91        .engine
92        .handler_names()
93        .into_iter()
94        .map(|name| {
95            let info = state.engine.handler_info(name);
96            let category = info.as_ref().and_then(|i| i.category.clone());
97            let version = info.as_ref().and_then(|i| i.version.clone());
98            let schedule = info.and_then(|i| i.schedule.map(|s| s.as_str().to_string()));
99            WorkflowSummary {
100                name: name.to_string(),
101                category,
102                version,
103                schedule,
104                paused_at: paused_at.get(name).copied(),
105            }
106        })
107        .collect();
108
109    if let Some(ref filter) = params.name {
110        let lower = filter.to_lowercase();
111        summaries.retain(|s| s.name.to_lowercase().contains(&lower));
112    }
113
114    if let Some(ref cat_filter) = params.category {
115        if cat_filter == UNCATEGORIZED_FILTER {
116            summaries.retain(|s| s.category.is_none());
117        } else {
118            let needle = cat_filter.to_lowercase();
119            summaries.retain(|s| {
120                s.category
121                    .as_deref()
122                    .is_some_and(|c| c.to_lowercase().contains(&needle))
123            });
124        }
125    }
126
127    summaries.sort_by(|a, b| {
128        a.category
129            .as_deref()
130            .unwrap_or("")
131            .cmp(b.category.as_deref().unwrap_or(""))
132            .then_with(|| a.name.cmp(&b.name))
133    });
134
135    Ok(ok(summaries))
136}
137
138#[cfg(test)]
139mod tests {
140    use axum::Router;
141    use axum::body::Body;
142    use axum::http::{Request, StatusCode};
143    use axum::routing::get;
144    use http_body_util::BodyExt;
145    use ironflow_core::providers::claude::ClaudeCodeProvider;
146    use ironflow_engine::context::WorkflowContext;
147    use ironflow_engine::engine::Engine;
148    use ironflow_engine::handler::{HandlerFuture, WorkflowHandler};
149    use ironflow_engine::notify::Event;
150    use ironflow_store::memory::InMemoryStore;
151    use serde_json::{Value as JsonValue, from_slice, from_value};
152    use std::sync::Arc;
153    use tokio::sync::broadcast;
154    use tower::ServiceExt;
155
156    use super::*;
157    use crate::routes::test_helpers::create_user_auth_header;
158
159    struct TestWorkflow;
160
161    impl WorkflowHandler for TestWorkflow {
162        fn name(&self) -> &str {
163            "test-workflow"
164        }
165
166        fn execute<'a>(&'a self, _ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
167            Box::pin(async move { Ok(()) })
168        }
169    }
170
171    struct AnotherWorkflow;
172
173    impl WorkflowHandler for AnotherWorkflow {
174        fn name(&self) -> &str {
175            "another-workflow"
176        }
177
178        fn execute<'a>(&'a self, _ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
179            Box::pin(async move { Ok(()) })
180        }
181    }
182
183    struct EtlNightlyWorkflow;
184
185    impl WorkflowHandler for EtlNightlyWorkflow {
186        fn name(&self) -> &str {
187            "etl-nightly"
188        }
189        fn category(&self) -> Option<&str> {
190            Some("data/etl")
191        }
192        fn execute<'a>(&'a self, _ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
193            Box::pin(async move { Ok(()) })
194        }
195    }
196
197    struct ReportsDailyWorkflow;
198
199    impl WorkflowHandler for ReportsDailyWorkflow {
200        fn name(&self) -> &str {
201            "reports-daily"
202        }
203        fn category(&self) -> Option<&str> {
204            Some("data/reports")
205        }
206        fn execute<'a>(&'a self, _ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
207            Box::pin(async move { Ok(()) })
208        }
209    }
210
211    fn base_state(engine: Engine) -> AppState {
212        let store = Arc::new(InMemoryStore::new());
213        Arc::new(InMemoryStore::new());
214        let jwt_config = Arc::new(ironflow_auth::jwt::JwtConfig {
215            secret: "test-secret".to_string(),
216            access_token_ttl_secs: 900,
217            refresh_token_ttl_secs: 604800,
218            cookie_domain: None,
219            cookie_secure: false,
220        });
221        let (event_sender, _) = broadcast::channel::<Event>(1);
222        AppState::new(
223            store,
224            Arc::new(engine),
225            jwt_config,
226            "test-worker-token".to_string(),
227            event_sender,
228        )
229    }
230
231    fn test_state() -> AppState {
232        let store = Arc::new(InMemoryStore::new());
233        let provider = Arc::new(ClaudeCodeProvider::new());
234        let mut engine = Engine::new(store.clone(), provider);
235        engine.register(TestWorkflow).unwrap();
236        engine.register(AnotherWorkflow).unwrap();
237        base_state(engine)
238    }
239
240    fn test_state_with_categories() -> AppState {
241        let store = Arc::new(InMemoryStore::new());
242        let provider = Arc::new(ClaudeCodeProvider::new());
243        let mut engine = Engine::new(store.clone(), provider);
244        engine.register(TestWorkflow).unwrap();
245        engine.register(EtlNightlyWorkflow).unwrap();
246        engine.register(ReportsDailyWorkflow).unwrap();
247        base_state(engine)
248    }
249
250    async fn run_request(state: AppState, uri: &str) -> (StatusCode, Vec<WorkflowSummary>) {
251        let auth_header = create_user_auth_header(&state, "testuser", false).await;
252        let app = Router::new()
253            .route("/", get(list_workflows))
254            .with_state(state);
255
256        let req = Request::builder()
257            .uri(uri)
258            .header("authorization", auth_header)
259            .body(Body::empty())
260            .unwrap();
261
262        let resp = app.oneshot(req).await.unwrap();
263        let status = resp.status();
264        let body = resp.into_body().collect().await.unwrap().to_bytes();
265        let json_val: JsonValue = from_slice(&body).unwrap();
266        let summaries: Vec<WorkflowSummary> = from_value(json_val["data"].clone()).unwrap();
267        (status, summaries)
268    }
269
270    #[tokio::test]
271    async fn list_workflows_empty() {
272        let store = Arc::new(InMemoryStore::new());
273        let provider = Arc::new(ClaudeCodeProvider::new());
274        let engine = Engine::new(store.clone(), provider);
275        let state = base_state(engine);
276
277        let (status, summaries) = run_request(state, "/").await;
278        assert_eq!(status, StatusCode::OK);
279        assert!(summaries.is_empty());
280    }
281
282    #[tokio::test]
283    async fn list_workflows_multiple_returns_summaries() {
284        let state = test_state();
285        let (status, summaries) = run_request(state, "/").await;
286        assert_eq!(status, StatusCode::OK);
287        assert_eq!(summaries.len(), 2);
288        assert!(summaries.iter().any(|s| s.name == "test-workflow"));
289        assert!(summaries.iter().any(|s| s.name == "another-workflow"));
290        assert!(summaries.iter().all(|s| s.category.is_none()));
291    }
292
293    #[tokio::test]
294    async fn list_workflows_filtered_by_name() {
295        let state = test_state();
296        let (_, summaries) = run_request(state, "/?name=test").await;
297        assert_eq!(summaries.len(), 1);
298        assert_eq!(summaries[0].name, "test-workflow");
299    }
300
301    #[tokio::test]
302    async fn list_workflows_filter_name_case_insensitive() {
303        let state = test_state();
304        let (_, summaries) = run_request(state, "/?name=TEST").await;
305        assert_eq!(summaries.len(), 1);
306        assert_eq!(summaries[0].name, "test-workflow");
307    }
308
309    #[tokio::test]
310    async fn list_workflows_filter_no_match() {
311        let state = test_state();
312        let (_, summaries) = run_request(state, "/?name=nonexistent").await;
313        assert!(summaries.is_empty());
314    }
315
316    #[tokio::test]
317    async fn list_workflows_returns_category_when_present() {
318        let state = test_state_with_categories();
319        let (_, summaries) = run_request(state, "/").await;
320        let etl = summaries.iter().find(|s| s.name == "etl-nightly").unwrap();
321        assert_eq!(etl.category.as_deref(), Some("data/etl"));
322        let test = summaries
323            .iter()
324            .find(|s| s.name == "test-workflow")
325            .unwrap();
326        assert!(test.category.is_none());
327    }
328
329    #[tokio::test]
330    async fn list_workflows_filter_by_category_partial() {
331        let state = test_state_with_categories();
332        let (_, summaries) = run_request(state, "/?category=data").await;
333        assert_eq!(summaries.len(), 2);
334        assert!(
335            summaries
336                .iter()
337                .all(|s| { s.category.as_deref().is_some_and(|c| c.contains("data")) })
338        );
339    }
340
341    #[tokio::test]
342    async fn list_workflows_filter_by_category_nested_substring() {
343        let state = test_state_with_categories();
344        let (_, summaries) = run_request(state, "/?category=etl").await;
345        assert_eq!(summaries.len(), 1);
346        assert_eq!(summaries[0].name, "etl-nightly");
347    }
348
349    #[tokio::test]
350    async fn list_workflows_filter_by_category_case_insensitive() {
351        let state = test_state_with_categories();
352        let (_, summaries) = run_request(state, "/?category=DATA").await;
353        assert_eq!(summaries.len(), 2);
354        let (_, summaries) = run_request(test_state_with_categories(), "/?category=Etl").await;
355        assert_eq!(summaries.len(), 1);
356        assert_eq!(summaries[0].name, "etl-nightly");
357    }
358
359    #[tokio::test]
360    async fn list_workflows_filter_by_category_no_match_excludes_uncategorized() {
361        let state = test_state_with_categories();
362        let (_, summaries) = run_request(state, "/?category=nonexistent").await;
363        assert!(summaries.is_empty());
364    }
365
366    #[tokio::test]
367    async fn list_workflows_filter_uncategorized_sentinel() {
368        let state = test_state_with_categories();
369        let (_, summaries) = run_request(state, "/?category=__uncategorized__").await;
370        assert_eq!(summaries.len(), 1);
371        assert_eq!(summaries[0].name, "test-workflow");
372    }
373
374    struct ScheduledWorkflow {
375        schedule: ironflow_engine::prelude::CronSchedule,
376    }
377    impl ScheduledWorkflow {
378        fn new() -> Self {
379            Self {
380                schedule: ironflow_engine::prelude::CronSchedule::new("0 30 9 * * MON-FRI")
381                    .unwrap(),
382            }
383        }
384    }
385
386    impl WorkflowHandler for ScheduledWorkflow {
387        fn name(&self) -> &str {
388            "scheduled-workflow"
389        }
390        fn schedule(&self) -> Option<&ironflow_engine::prelude::CronSchedule> {
391            Some(&self.schedule)
392        }
393        fn execute<'a>(&'a self, _ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
394            Box::pin(async move { Ok(()) })
395        }
396    }
397
398    #[tokio::test]
399    async fn list_workflows_returns_schedule_when_present() {
400        let store = Arc::new(InMemoryStore::new());
401        let provider = Arc::new(ClaudeCodeProvider::new());
402        let mut engine = Engine::new(store.clone(), provider);
403        engine.register(TestWorkflow).unwrap();
404        engine.register(ScheduledWorkflow::new()).unwrap();
405        let state = base_state(engine);
406
407        let (_, summaries) = run_request(state, "/").await;
408        let scheduled = summaries
409            .iter()
410            .find(|s| s.name == "scheduled-workflow")
411            .unwrap();
412        assert_eq!(scheduled.schedule.as_deref(), Some("0 30 9 * * MON-FRI"));
413
414        let unscheduled = summaries
415            .iter()
416            .find(|s| s.name == "test-workflow")
417            .unwrap();
418        assert!(unscheduled.schedule.is_none());
419    }
420
421    #[tokio::test]
422    async fn list_workflows_returns_paused_at_for_paused_workflows() {
423        let state = test_state();
424        let pause = state
425            .store
426            .pause_workflow("test-workflow", None)
427            .await
428            .unwrap();
429
430        let (_, summaries) = run_request(state, "/").await;
431        let paused = summaries
432            .iter()
433            .find(|s| s.name == "test-workflow")
434            .unwrap();
435        assert_eq!(paused.paused_at, Some(pause.paused_at));
436        let active = summaries
437            .iter()
438            .find(|s| s.name == "another-workflow")
439            .unwrap();
440        assert!(active.paused_at.is_none());
441    }
442}
ironflow_api/routes/list_workflows.rs

ironflow_api/routes/
list_workflows.rs
