assay_workflow/api/
mod.rs

1pub mod activities;
2pub mod api_keys;
3pub mod auth;
4pub mod events;
5pub mod namespaces;
6pub mod openapi;
7pub mod public;
8pub mod queues;
9pub mod schedules;
10pub mod tasks;
11pub mod workers;
12pub mod workflow_tasks;
13pub mod workflows;
14
15use std::sync::Arc;
16
17use axum::middleware;
18use axum::Router;
19use tokio::sync::broadcast;
20use tracing::info;
21
22use crate::auth_mode::AuthMode;
23use crate::ctx::{BroadcastEvent, EngineEvent, WorkflowCtx};
24use crate::store::WorkflowStore;
25
26/// Build the full API router.
27///
28/// Three tiers:
29///   1. **Authenticated `/api/v1/*`** — workflows, schedules, namespaces,
30///      activities, tasks, workers, queues, api-keys, events, meta/version.
31///      Gated by `auth::auth_middleware` when an auth mode is enabled.
32///   2. **Public `/api/v1/*`** — health, version. Always unauthenticated so
33///      Kubernetes probes, load balancers, and third-party monitors can
34///      reach them without a bearer token.
35///   3. **Dashboard + OpenAPI** — HTML/JSON at the root. Always public.
36pub fn router<S: WorkflowStore>(state: Arc<WorkflowCtx<S>>) -> Router {
37    let needs_auth = state.auth_mode.is_enabled();
38
39    let authed_api = Router::new()
40        .nest("/api/v1", api_v1_router::<S>())
41        .nest("/api/v1", events::router::<S>());
42
43    let authed_api = if needs_auth {
44        authed_api.layer(middleware::from_fn_with_state(
45            Arc::clone(&state),
46            auth::auth_middleware::<S>,
47        ))
48    } else {
49        authed_api
50    };
51
52    // Public /api/v1/* routes — outside the auth layer by construction.
53    let public_api = Router::new().nest("/api/v1", public::router::<S>());
54
55    let app = authed_api
56        .merge(public_api)
57        .merge(openapi::router::<S>());
58
59    app.with_state(state)
60}
61
62fn api_v1_router<S: WorkflowStore>() -> Router<Arc<WorkflowCtx<S>>> {
63    Router::new()
64        .merge(workflows::router::<S>())
65        .merge(activities::router::<S>())
66        .merge(workflow_tasks::router::<S>())
67        .merge(tasks::router::<S>())
68        .merge(schedules::router::<S>())
69        .merge(workers::router::<S>())
70        .merge(namespaces::router::<S>())
71        .merge(queues::router::<S>())
72        .merge(api_keys::router::<S>())
73}
74
75/// Start the HTTP server on the given port.
76pub async fn serve(
77    store: impl WorkflowStore + 'static,
78    port: u16,
79    auth_mode: AuthMode,
80) -> anyhow::Result<()> {
81    serve_with_version(store, port, auth_mode, None).await
82}
83
84/// Like `serve`, but lets the embedder (e.g. the `assay` binary) pass
85/// its own semver so `/api/v1/version` reflects the binary users are
86/// actually running instead of the internal `assay-workflow` crate
87/// version. Without this, the dashboard would show a misleading
88/// "engine crate" version to operators.
89pub async fn serve_with_version(
90    store: impl WorkflowStore + 'static,
91    port: u16,
92    auth_mode: AuthMode,
93    binary_version: Option<&'static str>,
94) -> anyhow::Result<()> {
95    // The SSE channel that dashboard browsers subscribe to. The
96    // engine pushes EngineEvents into the bridge below; the bridge
97    // converts each one to a BroadcastEvent and forwards to every
98    // connected dashboard.
99    let (sse_tx, _) = broadcast::channel::<BroadcastEvent>(1024);
100    let (engine_tx, mut engine_rx) = broadcast::channel::<EngineEvent>(1024);
101
102    let store = Arc::new(store);
103    let mut ctx = WorkflowCtx::start(store)
104        .with_event_broadcaster(engine_tx)
105        .with_auth_mode(auth_mode.clone())
106        .with_sse_tx(sse_tx.clone());
107
108    if let Some(v) = binary_version {
109        ctx = ctx.with_binary_version(v);
110    }
111
112    {
113        let sse_tx = sse_tx.clone();
114        tokio::spawn(async move {
115            while let Ok(evt) = engine_rx.recv().await {
116                let _ = sse_tx.send(BroadcastEvent {
117                    event_type: evt.event_type,
118                    workflow_id: evt.workflow_id,
119                    payload: Some(serde_json::json!({
120                        "namespace": evt.namespace,
121                    }).to_string()),
122                });
123            }
124        });
125    }
126
127    let mode_desc = auth_mode.describe();
128    let state = Arc::new(ctx);
129
130    let app = router(state);
131    let listener = tokio::net::TcpListener::bind(format!("0.0.0.0:{port}")).await?;
132    info!("Workflow API listening on 0.0.0.0:{port} (auth: {mode_desc})");
133
134    axum::serve(listener, app).await?;
135    Ok(())
136}
assay_workflow/api/mod.rs

assay_workflow/api/
mod.rs
