aion_server/api/http/
router.rs

1//! Public HTTP router construction.
2
3use axum::{
4    Router,
5    extract::DefaultBodyLimit,
6    http::StatusCode,
7    routing::{any, get, post},
8};
9
10use super::deploy::{list_versions, route_version, unload_version, upload_package};
11use super::events::subscribe_events_socket;
12use super::schedules::{
13    create_schedule, delete_schedule, describe_schedule, list_schedules, pause_schedule,
14    resume_schedule, update_schedule,
15};
16use super::workflows::{
17    cancel_workflow, count_workflows, describe_workflow, get_workflows, post_list_workflows,
18    query_workflow, signal_workflow, start_workflow,
19};
20use crate::{ServerError, ServerState, dashboard::assets, observability};
21
22/// Build the public HTTP application: workflow-management routes first, then
23/// the dashboard static asset fallback. The dashboard adds no data API.
24///
25/// # Errors
26///
27/// Returns [`ServerError::Config`] when dashboard assets are misconfigured.
28pub fn http_router(state: ServerState) -> Result<Router, ServerError> {
29    let dashboard = assets::dashboard_router(&state.runtime_config().dashboard)?;
30    let metrics = state.metrics().cloned();
31    let health = state.health().cloned();
32    let mut router = workflow_router(state);
33    if let Some(metrics) = metrics {
34        router = router.merge(Router::new().route(
35            "/metrics",
36            get(observability::metrics::metrics_handler).with_state(metrics),
37        ));
38    }
39    if let Some(health) = health {
40        router = router.merge(
41            Router::new()
42                .route("/health/live", get(observability::health::live))
43                .route(
44                    "/health/ready",
45                    get(observability::health::ready).with_state(health),
46                ),
47        );
48    }
49    Ok(router.merge(dashboard))
50}
51
52/// Disabled deploy surface: a plain 404 with no body, indistinguishable
53/// from an unmounted route family.
54async fn deploy_disabled() -> StatusCode {
55    StatusCode::NOT_FOUND
56}
57
58/// Build the public workflow-management HTTP router.
59pub fn workflow_router(state: ServerState) -> Router {
60    // The deploy surface is dark by default: when `[deploy].enabled` is
61    // false the routes are not mounted and every `/deploy/*` path is a
62    // plain 404 (the explicit catch-all keeps the dashboard SPA fallback
63    // from answering for the deploy namespace). The archive upload route
64    // disables the default body limit because `read_archive_body` enforces
65    // the operator-configured `deploy.max_archive_bytes` ceiling while
66    // streaming.
67    let deploy = if state.runtime_config().deploy.enabled {
68        Router::new()
69            .route(
70                "/deploy/packages",
71                post(upload_package).layer(DefaultBodyLimit::disable()),
72            )
73            .route("/deploy/versions", get(list_versions))
74            .route("/deploy/route", post(route_version))
75            .route("/deploy/unload", post(unload_version))
76    } else {
77        Router::new().route("/deploy/{*rest}", any(deploy_disabled))
78    };
79    deploy
80        .route("/workflows", get(get_workflows))
81        .route("/workflows/count", get(count_workflows))
82        .route("/workflows/start", post(start_workflow))
83        .route("/workflows/signal", post(signal_workflow))
84        .route("/workflows/query", post(query_workflow))
85        .route("/workflows/cancel", post(cancel_workflow))
86        .route("/workflows/list", post(post_list_workflows))
87        .route("/workflows/describe", post(describe_workflow))
88        .route("/events/stream", get(subscribe_events_socket))
89        .route("/schedules", post(create_schedule).get(list_schedules))
90        .route(
91            "/schedules/{id}",
92            get(describe_schedule)
93                .put(update_schedule)
94                .delete(delete_schedule),
95        )
96        .route("/schedules/{id}/pause", post(pause_schedule))
97        .route("/schedules/{id}/resume", post(resume_schedule))
98        .with_state(state)
99}
100
101#[cfg(test)]
102mod tests {
103    use std::{fs, sync::Arc};
104
105    use aion::EngineBuilder;
106    use aion_proto::{ProtoListWorkflowsRequest, ProtoListWorkflowsResponse};
107    use aion_store::{EventStore, InMemoryStore, visibility::ListWorkflowsFilter};
108    use axum::{body, http::Request, http::StatusCode};
109    use tower::ServiceExt;
110
111    use super::super::test_support::{
112        NAMESPACE, json_request, read_json, read_text, runtime_config, server_state,
113    };
114    use super::*;
115    use crate::{
116        NamespaceResolver, StaticScheduleNamespaces, StaticWorkflowNamespaces,
117        config::{DashboardAssetSource, DashboardConfig, NamespaceMode},
118    };
119
120    #[tokio::test]
121    async fn dashboard_assets_serve_index_asset_and_do_not_shadow_public_api()
122    -> Result<(), Box<dyn std::error::Error>> {
123        let bundle = tempfile::tempdir()?;
124        fs::write(
125            bundle.path().join("index.html"),
126            "<!doctype html><title>Aion</title><script src=\"/app.js\"></script>",
127        )?;
128        fs::write(bundle.path().join("app.js"), "window.AION = true;")?;
129
130        let store: Arc<dyn EventStore> = Arc::new(InMemoryStore::default());
131        let engine = Arc::new(
132            EngineBuilder::new()
133                .store_arc(Arc::clone(&store))
134                .in_memory_visibility()
135                .scheduler_threads(1)
136                .build()
137                .await?,
138        );
139        let resolver = NamespaceResolver::from_parts(
140            NamespaceMode::SharedEngine,
141            Some(engine),
142            Arc::new(StaticWorkflowNamespaces::default()),
143            Arc::new(StaticScheduleNamespaces::default()),
144        );
145        let mut config = runtime_config();
146        config.dashboard = DashboardConfig {
147            source: DashboardAssetSource::FileSystem {
148                asset_path: bundle.path().to_path_buf(),
149            },
150        };
151        let router = http_router(server_state(resolver, config).await?)?;
152
153        let root = router
154            .clone()
155            .oneshot(Request::builder().uri("/").body(body::Body::empty())?)
156            .await?;
157        assert_eq!(root.status(), StatusCode::OK);
158        assert!(read_text(root).await?.contains("<title>Aion</title>"));
159
160        let asset = router
161            .clone()
162            .oneshot(
163                Request::builder()
164                    .uri("/app.js")
165                    .body(body::Body::empty())?,
166            )
167            .await?;
168        assert_eq!(asset.status(), StatusCode::OK);
169        assert_eq!(read_text(asset).await?, "window.AION = true;");
170
171        let spa = router
172            .clone()
173            .oneshot(
174                Request::builder()
175                    .uri("/dashboard/workflows/demo")
176                    .body(body::Body::empty())?,
177            )
178            .await?;
179        assert_eq!(spa.status(), StatusCode::OK);
180        assert!(read_text(spa).await?.contains("<title>Aion</title>"));
181
182        let list = ProtoListWorkflowsRequest {
183            namespace: NAMESPACE.to_owned(),
184            filter: Some(aion_proto::encode_core_value(
185                NAMESPACE,
186                None,
187                &ListWorkflowsFilter {
188                    workflow_type: Some(String::from("nonexistent")),
189                    ..ListWorkflowsFilter::default()
190                },
191            )?),
192        };
193        let list_response = router
194            .oneshot(json_request("/workflows/list", &list)?)
195            .await?;
196        assert_eq!(list_response.status(), StatusCode::OK);
197        let list_body: ProtoListWorkflowsResponse = read_json(list_response).await?;
198        assert!(list_body.summaries.is_empty());
199        Ok(())
200    }
201
202    #[tokio::test]
203    async fn observability_routes_are_public_and_expose_expected_payloads()
204    -> Result<(), Box<dyn std::error::Error>> {
205        // This test rides the production `build_with_store` startup path, so
206        // under `feature = "auth"` the configured jwks_url must be a live
207        // endpoint for the initial JWKS fetch.
208        #[cfg(feature = "auth")]
209        let config = {
210            let mut config = runtime_config();
211            config.auth.jwks_url = Some(crate::auth::test_support::serve_jwks()?);
212            config
213        };
214        #[cfg(not(feature = "auth"))]
215        let config = runtime_config();
216        let router = http_router(
217            crate::ServerState::build_with_store(InMemoryStore::default(), config).await?,
218        )?;
219
220        let metrics_response = router
221            .clone()
222            .oneshot(
223                Request::builder()
224                    .uri("/metrics")
225                    .body(body::Body::empty())?,
226            )
227            .await?;
228        assert_eq!(metrics_response.status(), StatusCode::OK);
229        assert_eq!(
230            metrics_response
231                .headers()
232                .get(axum::http::header::CONTENT_TYPE)
233                .and_then(|value| value.to_str().ok()),
234            Some("text/plain; version=0.0.4; charset=utf-8")
235        );
236        let metrics_body = read_text(metrics_response).await?;
237        assert!(metrics_body.contains("# HELP aion_workflows_started_total"));
238        assert!(metrics_body.contains("# TYPE aion_workflows_started_total counter"));
239        assert!(metrics_body.contains("# HELP aion_activity_duration_seconds"));
240        assert!(metrics_body.contains("# TYPE aion_activity_duration_seconds histogram"));
241        assert!(metrics_body.contains("aion_activity_duration_seconds_bucket"));
242        assert!(metrics_body.contains("aion_store_operation_duration_seconds_bucket"));
243
244        let live_response = router
245            .clone()
246            .oneshot(
247                Request::builder()
248                    .uri("/health/live")
249                    .body(body::Body::empty())?,
250            )
251            .await?;
252        assert_eq!(live_response.status(), StatusCode::OK);
253
254        let ready_response = router
255            .oneshot(
256                Request::builder()
257                    .uri("/health/ready")
258                    .body(body::Body::empty())?,
259            )
260            .await?;
261        assert_eq!(ready_response.status(), StatusCode::OK);
262        Ok(())
263    }
264}
aion_server/api/http/router.rs

aion_server/api/http/
router.rs
