ironflow_api/routes/
get_workflow.rs

1//! `GET /api/v1/workflows/:name` — Get workflow details.
2
3use std::collections::{HashMap, HashSet};
4
5use axum::extract::{Path, State};
6use axum::response::IntoResponse;
7use chrono::{DateTime, Utc};
8use ironflow_auth::extractor::Authenticated;
9use ironflow_store::entities::ApiKeyScope;
10use ironflow_store::entities::{MAX_PRIORITY, MIN_PRIORITY};
11use rust_decimal::Decimal;
12use serde::Serialize;
13use serde_json::Value;
14
15use crate::authorization::AuthorizeExt;
16use crate::error::ApiError;
17use crate::response::ok;
18use crate::state::AppState;
19
20/// Sub-workflow detail included in the workflow response.
21#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
22#[derive(Debug, Serialize)]
23pub struct SubWorkflowDetail {
24    /// Sub-workflow name.
25    pub name: String,
26    /// Human-readable description.
27    pub description: String,
28    /// Optional Rust source code of the handler.
29    pub source_code: Option<String>,
30}
31
32/// Workflow detail response.
33#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
34#[derive(Debug, Serialize)]
35pub struct WorkflowDetailResponse {
36    /// Workflow name.
37    pub name: String,
38    /// Human-readable description.
39    pub description: String,
40    /// Optional Rust source code of the handler.
41    pub source_code: Option<String>,
42    /// Sub-workflows invoked by this handler (recursive, depth-limited).
43    pub sub_workflows: Vec<SubWorkflowDetail>,
44    /// Optional `/`-separated category path used to group workflows.
45    pub category: Option<String>,
46    /// Current handler version.
47    pub version: Option<String>,
48    /// Versions accepted for replay without `force`.
49    #[serde(default, skip_serializing_if = "Vec::is_empty")]
50    pub compatible_versions: Vec<String>,
51    /// JSON Schema describing the expected input payload.
52    #[serde(skip_serializing_if = "Option::is_none")]
53    pub input_schema: Option<Value>,
54    /// Labels automatically applied to every run of this workflow.
55    #[serde(default, skip_serializing_if = "HashMap::is_empty")]
56    pub default_labels: HashMap<String, String>,
57    /// Optional 6-field cron expression for automatic execution.
58    #[serde(skip_serializing_if = "Option::is_none")]
59    pub schedule: Option<String>,
60    /// Default cumulative cost cap applied to runs of this workflow, in USD.
61    ///
62    /// Overridden by a cap supplied at run creation. `None` means the workflow
63    /// declares no default and falls back to the server default (if any).
64    #[cfg_attr(feature = "openapi", schema(value_type = Option<f64>))]
65    #[serde(skip_serializing_if = "Option::is_none")]
66    pub default_max_cost_usd: Option<Decimal>,
67    /// When the workflow was paused: its queued runs are not picked up until
68    /// it is resumed. Omitted when the workflow is not paused.
69    #[serde(skip_serializing_if = "Option::is_none")]
70    pub paused_at: Option<DateTime<Utc>>,
71    /// Default queue priority of the runs of this workflow, from -100 to 100.
72    /// Clamped like the run creation path. Always present, 0 when the handler
73    /// declares none.
74    pub priority: i16,
75}
76
77/// Get details about a registered workflow.
78///
79/// # Errors
80///
81/// - 404 if the workflow is not registered
82#[cfg_attr(
83    feature = "openapi",
84    utoipa::path(
85        get,
86        path = "/api/v1/workflows/{name}",
87        tags = ["workflows"],
88        params(("name" = String, Path, description = "Workflow name")),
89        responses(
90            (status = 200, description = "Workflow details", body = WorkflowDetailResponse),
91            (status = 401, description = "Unauthorized"),
92            (status = 404, description = "Workflow not found")
93        ),
94        security(("Bearer" = []))
95    )
96)]
97pub async fn get_workflow(
98    auth: Authenticated,
99    State(state): State<AppState>,
100    Path(name): Path<String>,
101) -> Result<impl IntoResponse, ApiError> {
102    auth.require(ApiKeyScope::WorkflowsRead, false)?;
103    let info = state
104        .engine
105        .handler_info(&name)
106        .ok_or_else(|| ApiError::WorkflowNotFound(name.clone()))?;
107
108    let mut sub_workflows = Vec::new();
109    let mut visited = HashSet::new();
110    visited.insert(name.clone());
111    collect_sub_workflows(
112        &state,
113        &info.sub_workflows,
114        &mut sub_workflows,
115        &mut visited,
116        5,
117    );
118
119    let paused_at = state
120        .store
121        .list_workflow_pauses()
122        .await?
123        .into_iter()
124        .find(|pause| pause.workflow_name == name)
125        .map(|pause| pause.paused_at);
126
127    Ok(ok(WorkflowDetailResponse {
128        name,
129        description: info.description,
130        source_code: info.source_code,
131        sub_workflows,
132        category: info.category,
133        version: info.version,
134        compatible_versions: info.compatible_versions,
135        input_schema: info.input_schema,
136        default_labels: info.default_labels,
137        schedule: info.schedule.map(|s| s.as_str().to_string()),
138        default_max_cost_usd: info.default_max_cost_usd,
139        paused_at,
140        priority: info.priority.clamp(MIN_PRIORITY, MAX_PRIORITY),
141    }))
142}
143
144fn collect_sub_workflows(
145    state: &AppState,
146    names: &[String],
147    result: &mut Vec<SubWorkflowDetail>,
148    visited: &mut HashSet<String>,
149    depth: usize,
150) {
151    if depth == 0 {
152        return;
153    }
154    for sub_name in names {
155        if !visited.insert(sub_name.clone()) {
156            continue;
157        }
158        if let Some(sub_info) = state.engine.handler_info(sub_name) {
159            collect_sub_workflows(state, &sub_info.sub_workflows, result, visited, depth - 1);
160            result.push(SubWorkflowDetail {
161                name: sub_name.clone(),
162                description: sub_info.description,
163                source_code: sub_info.source_code,
164            });
165        }
166    }
167}
168
169#[cfg(test)]
170mod tests {
171    use axum::Router;
172    use axum::body::Body;
173    use axum::http::{Request, StatusCode};
174    use axum::routing::get;
175    use http_body_util::BodyExt;
176    use ironflow_core::providers::claude::ClaudeCodeProvider;
177    use ironflow_engine::context::WorkflowContext;
178    use ironflow_engine::engine::Engine;
179    use ironflow_engine::handler::{HandlerFuture, WorkflowHandler};
180    use ironflow_engine::notify::Event;
181    use ironflow_store::memory::InMemoryStore;
182    use serde_json::Value as JsonValue;
183    use std::sync::Arc;
184    use tokio::sync::broadcast;
185    use tower::ServiceExt;
186
187    use super::*;
188    use crate::routes::test_helpers::create_user_auth_header;
189
190    struct DescribedWorkflow;
191    impl WorkflowHandler for DescribedWorkflow {
192        fn name(&self) -> &str {
193            "my-workflow"
194        }
195        fn execute<'a>(&'a self, _ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
196            Box::pin(async move { Ok(()) })
197        }
198    }
199
200    struct CategorizedWorkflow;
201    impl WorkflowHandler for CategorizedWorkflow {
202        fn name(&self) -> &str {
203            "cat-workflow"
204        }
205        fn category(&self) -> Option<&str> {
206            Some("data/etl")
207        }
208        fn execute<'a>(&'a self, _ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
209            Box::pin(async move { Ok(()) })
210        }
211    }
212
213    fn test_state() -> AppState {
214        let store = Arc::new(InMemoryStore::new());
215        Arc::new(InMemoryStore::new());
216        let provider = Arc::new(ClaudeCodeProvider::new());
217        let mut engine = Engine::new(store.clone(), provider);
218        engine.register(DescribedWorkflow).unwrap();
219        engine.register(CategorizedWorkflow).unwrap();
220        let jwt_config = Arc::new(ironflow_auth::jwt::JwtConfig {
221            secret: "test-secret".to_string(),
222            access_token_ttl_secs: 900,
223            refresh_token_ttl_secs: 604800,
224            cookie_domain: None,
225            cookie_secure: false,
226        });
227        let (event_sender, _) = broadcast::channel::<Event>(1);
228        AppState::new(
229            store,
230            Arc::new(engine),
231            jwt_config,
232            "test-worker-token".to_string(),
233            event_sender,
234        )
235    }
236
237    #[tokio::test]
238    async fn get_workflow_found() {
239        let state = test_state();
240        let auth_header = create_user_auth_header(&state, "testuser", false).await;
241        let app = Router::new()
242            .route("/{name}", get(get_workflow))
243            .with_state(state);
244
245        let req = Request::builder()
246            .uri("/my-workflow")
247            .header("authorization", auth_header)
248            .body(Body::empty())
249            .unwrap();
250
251        let resp = app.oneshot(req).await.unwrap();
252        assert_eq!(resp.status(), StatusCode::OK);
253
254        let body = resp.into_body().collect().await.unwrap().to_bytes();
255        let json_val: JsonValue = serde_json::from_slice(&body).unwrap();
256        assert_eq!(json_val["data"]["name"], "my-workflow");
257    }
258
259    #[tokio::test]
260    async fn get_workflow_returns_category_when_set() {
261        let state = test_state();
262        let auth_header = create_user_auth_header(&state, "testuser", false).await;
263        let app = Router::new()
264            .route("/{name}", get(get_workflow))
265            .with_state(state);
266
267        let req = Request::builder()
268            .uri("/cat-workflow")
269            .header("authorization", auth_header)
270            .body(Body::empty())
271            .unwrap();
272
273        let resp = app.oneshot(req).await.unwrap();
274        assert_eq!(resp.status(), StatusCode::OK);
275        let body = resp.into_body().collect().await.unwrap().to_bytes();
276        let json_val: JsonValue = serde_json::from_slice(&body).unwrap();
277        assert_eq!(json_val["data"]["category"], "data/etl");
278    }
279
280    #[tokio::test]
281    async fn get_workflow_category_null_when_uncategorized() {
282        let state = test_state();
283        let auth_header = create_user_auth_header(&state, "testuser", false).await;
284        let app = Router::new()
285            .route("/{name}", get(get_workflow))
286            .with_state(state);
287
288        let req = Request::builder()
289            .uri("/my-workflow")
290            .header("authorization", auth_header)
291            .body(Body::empty())
292            .unwrap();
293
294        let resp = app.oneshot(req).await.unwrap();
295        let body = resp.into_body().collect().await.unwrap().to_bytes();
296        let json_val: JsonValue = serde_json::from_slice(&body).unwrap();
297        assert!(json_val["data"]["category"].is_null());
298    }
299
300    struct CappedWorkflow;
301
302    impl WorkflowHandler for CappedWorkflow {
303        fn name(&self) -> &str {
304            "capped-workflow"
305        }
306        fn default_max_cost_usd(&self) -> Option<Decimal> {
307            Some(Decimal::new(325, 2))
308        }
309        fn execute<'a>(&'a self, _ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
310            Box::pin(async move { Ok(()) })
311        }
312    }
313
314    #[tokio::test]
315    async fn get_workflow_returns_handler_default_max_cost() {
316        let state = {
317            let store = Arc::new(InMemoryStore::new());
318            let provider = Arc::new(ClaudeCodeProvider::new());
319            let mut engine = Engine::new(store.clone(), provider);
320            engine.register(DescribedWorkflow).unwrap();
321            engine.register(CappedWorkflow).unwrap();
322            let jwt_config = Arc::new(ironflow_auth::jwt::JwtConfig {
323                secret: "test-secret".to_string(),
324                access_token_ttl_secs: 900,
325                refresh_token_ttl_secs: 604800,
326                cookie_domain: None,
327                cookie_secure: false,
328            });
329            let (event_sender, _) = broadcast::channel::<Event>(1);
330            AppState::new(
331                store,
332                Arc::new(engine),
333                jwt_config,
334                "test-worker-token".to_string(),
335                event_sender,
336            )
337        };
338        let auth_header = create_user_auth_header(&state, "testuser", false).await;
339        let app = Router::new()
340            .route("/{name}", get(get_workflow))
341            .with_state(state);
342
343        let req = Request::builder()
344            .uri("/capped-workflow")
345            .header("authorization", auth_header.clone())
346            .body(Body::empty())
347            .unwrap();
348
349        let resp = app.clone().oneshot(req).await.unwrap();
350        assert_eq!(resp.status(), StatusCode::OK);
351        let body = resp.into_body().collect().await.unwrap().to_bytes();
352        let json_val: JsonValue = serde_json::from_slice(&body).unwrap();
353        assert_eq!(json_val["data"]["default_max_cost_usd"], 3.25);
354
355        // A handler without a declared cap omits the field entirely.
356        let req = Request::builder()
357            .uri("/my-workflow")
358            .header("authorization", auth_header)
359            .body(Body::empty())
360            .unwrap();
361
362        let resp = app.oneshot(req).await.unwrap();
363        let body = resp.into_body().collect().await.unwrap().to_bytes();
364        let json_val: JsonValue = serde_json::from_slice(&body).unwrap();
365        assert!(json_val["data"].get("default_max_cost_usd").is_none());
366    }
367
368    struct ScheduledWorkflow {
369        schedule: ironflow_engine::prelude::CronSchedule,
370    }
371    impl ScheduledWorkflow {
372        fn new() -> Self {
373            Self {
374                schedule: ironflow_engine::prelude::CronSchedule::new("0 0 * * * *").unwrap(),
375            }
376        }
377    }
378    impl WorkflowHandler for ScheduledWorkflow {
379        fn name(&self) -> &str {
380            "sched-workflow"
381        }
382        fn schedule(&self) -> Option<&ironflow_engine::prelude::CronSchedule> {
383            Some(&self.schedule)
384        }
385        fn execute<'a>(&'a self, _ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
386            Box::pin(async move { Ok(()) })
387        }
388    }
389
390    fn test_state_with_schedule() -> AppState {
391        let store = Arc::new(InMemoryStore::new());
392        let provider = Arc::new(ClaudeCodeProvider::new());
393        let mut engine = Engine::new(store.clone(), provider);
394        engine.register(DescribedWorkflow).unwrap();
395        engine.register(ScheduledWorkflow::new()).unwrap();
396        let jwt_config = Arc::new(ironflow_auth::jwt::JwtConfig {
397            secret: "test-secret".to_string(),
398            access_token_ttl_secs: 900,
399            refresh_token_ttl_secs: 604800,
400            cookie_domain: None,
401            cookie_secure: false,
402        });
403        let (event_sender, _) = broadcast::channel::<Event>(1);
404        AppState::new(
405            store,
406            Arc::new(engine),
407            jwt_config,
408            "test-worker-token".to_string(),
409            event_sender,
410        )
411    }
412
413    #[tokio::test]
414    async fn get_workflow_returns_schedule_when_set() {
415        let state = test_state_with_schedule();
416        let auth_header = create_user_auth_header(&state, "testuser", false).await;
417        let app = Router::new()
418            .route("/{name}", get(get_workflow))
419            .with_state(state);
420
421        let req = Request::builder()
422            .uri("/sched-workflow")
423            .header("authorization", auth_header)
424            .body(Body::empty())
425            .unwrap();
426
427        let resp = app.oneshot(req).await.unwrap();
428        assert_eq!(resp.status(), StatusCode::OK);
429        let body = resp.into_body().collect().await.unwrap().to_bytes();
430        let json_val: JsonValue = serde_json::from_slice(&body).unwrap();
431        assert_eq!(json_val["data"]["schedule"], "0 0 * * * *");
432    }
433
434    #[tokio::test]
435    async fn get_workflow_schedule_null_when_unscheduled() {
436        let state = test_state_with_schedule();
437        let auth_header = create_user_auth_header(&state, "testuser", false).await;
438        let app = Router::new()
439            .route("/{name}", get(get_workflow))
440            .with_state(state);
441
442        let req = Request::builder()
443            .uri("/my-workflow")
444            .header("authorization", auth_header)
445            .body(Body::empty())
446            .unwrap();
447
448        let resp = app.oneshot(req).await.unwrap();
449        let body = resp.into_body().collect().await.unwrap().to_bytes();
450        let json_val: JsonValue = serde_json::from_slice(&body).unwrap();
451        assert!(json_val["data"]["schedule"].is_null());
452    }
453
454    struct PrioritizedWorkflow {
455        name: &'static str,
456        priority: i16,
457    }
458    impl WorkflowHandler for PrioritizedWorkflow {
459        fn name(&self) -> &str {
460            self.name
461        }
462        fn priority(&self) -> i16 {
463            self.priority
464        }
465        fn execute<'a>(&'a self, _ctx: &'a mut WorkflowContext) -> HandlerFuture<'a> {
466            Box::pin(async move { Ok(()) })
467        }
468    }
469
470    fn test_state_with_priorities() -> AppState {
471        let store = Arc::new(InMemoryStore::new());
472        let provider = Arc::new(ClaudeCodeProvider::new());
473        let mut engine = Engine::new(store.clone(), provider);
474        engine.register(DescribedWorkflow).unwrap();
475        for (name, priority) in [
476            ("prio-workflow", 50),
477            ("extreme-workflow", 500),
478            ("abyss-workflow", -500),
479        ] {
480            engine
481                .register(PrioritizedWorkflow { name, priority })
482                .unwrap();
483        }
484        let jwt_config = Arc::new(ironflow_auth::jwt::JwtConfig {
485            secret: "test-secret".to_string(),
486            access_token_ttl_secs: 900,
487            refresh_token_ttl_secs: 604800,
488            cookie_domain: None,
489            cookie_secure: false,
490        });
491        let (event_sender, _) = broadcast::channel::<Event>(1);
492        AppState::new(
493            store,
494            Arc::new(engine),
495            jwt_config,
496            "test-worker-token".to_string(),
497            event_sender,
498        )
499    }
500
501    async fn get_workflow_json(uri: &str) -> JsonValue {
502        let state = test_state_with_priorities();
503        let auth_header = create_user_auth_header(&state, "testuser", false).await;
504        let app = Router::new()
505            .route("/{name}", get(get_workflow))
506            .with_state(state);
507
508        let req = Request::builder()
509            .uri(uri)
510            .header("authorization", auth_header)
511            .body(Body::empty())
512            .unwrap();
513
514        let resp = app.oneshot(req).await.unwrap();
515        assert_eq!(resp.status(), StatusCode::OK);
516        let body = resp.into_body().collect().await.unwrap().to_bytes();
517        serde_json::from_slice(&body).unwrap()
518    }
519
520    #[tokio::test]
521    async fn get_workflow_returns_handler_priority() {
522        let json_val = get_workflow_json("/prio-workflow").await;
523        assert_eq!(json_val["data"]["priority"], 50);
524    }
525
526    #[tokio::test]
527    async fn get_workflow_priority_defaults_to_zero() {
528        let json_val = get_workflow_json("/my-workflow").await;
529        assert!(json_val["data"].get("priority").is_some());
530        assert_eq!(json_val["data"]["priority"], 0);
531    }
532
533    #[tokio::test]
534    async fn get_workflow_priority_is_clamped() {
535        let json_val = get_workflow_json("/extreme-workflow").await;
536        assert_eq!(json_val["data"]["priority"], 100);
537
538        let json_val = get_workflow_json("/abyss-workflow").await;
539        assert_eq!(json_val["data"]["priority"], -100);
540    }
541
542    #[tokio::test]
543    async fn get_workflow_not_found() {
544        let state = test_state();
545        let auth_header = create_user_auth_header(&state, "testuser", false).await;
546        let app = Router::new()
547            .route("/{name}", get(get_workflow))
548            .with_state(state);
549
550        let req = Request::builder()
551            .uri("/nonexistent")
552            .header("authorization", auth_header)
553            .body(Body::empty())
554            .unwrap();
555
556        let resp = app.oneshot(req).await.unwrap();
557        assert_eq!(resp.status(), StatusCode::NOT_FOUND);
558    }
559}
ironflow_api/routes/get_workflow.rs

ironflow_api/routes/
get_workflow.rs
