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

ironflow_api/routes/
list_workflows.rs
