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 assay_domain::events::EngineEventBus;
18use axum::Router;
19use axum::middleware;
20use tracing::info;
21
22use crate::auth_mode::AuthMode;
23use crate::ctx::WorkflowCtx;
24use crate::events::WorkflowEventBus;
25use crate::store::WorkflowStore;
26
27/// Build the full API router.
28///
29/// Three tiers:
30///   1. **Authenticated `/api/v1/*`** — workflows, schedules, namespaces,
31///      activities, tasks, workers, queues, api-keys, events, meta/version.
32///      Gated by `auth::auth_middleware` when an auth mode is enabled.
33///   2. **Public `/api/v1/*`** — health, version. Always unauthenticated so
34///      Kubernetes probes, load balancers, and third-party monitors can
35///      reach them without a bearer token.
36///   3. **Dashboard + OpenAPI** — HTML/JSON at the root. Always public.
37pub fn router<S: WorkflowStore>(state: Arc<WorkflowCtx<S>>) -> Router {
38    let needs_auth = state.auth_mode.is_enabled();
39
40    let authed_api = Router::new()
41        .nest("/api/v1", api_v1_router::<S>())
42        .nest("/api/v1", events::router::<S>());
43
44    let authed_api = if needs_auth {
45        authed_api.layer(middleware::from_fn_with_state(
46            Arc::clone(&state),
47            auth::auth_middleware::<S>,
48        ))
49    } else {
50        authed_api
51    };
52
53    // Public /api/v1/* routes — outside the auth layer by construction.
54    let public_api = Router::new().nest("/api/v1", public::router::<S>());
55
56    let app = authed_api.merge(public_api).merge(openapi::router::<S>());
57
58    app.with_state(state)
59}
60
61fn api_v1_router<S: WorkflowStore>() -> Router<Arc<WorkflowCtx<S>>> {
62    Router::new()
63        .merge(workflows::router::<S>())
64        .merge(activities::router::<S>())
65        .merge(workflow_tasks::router::<S>())
66        .merge(tasks::router::<S>())
67        .merge(schedules::router::<S>())
68        .merge(workers::router::<S>())
69        .merge(namespaces::router::<S>())
70        .merge(queues::router::<S>())
71        .merge(api_keys::router::<S>())
72}
73
74/// Start the HTTP server on the given port.
75///
76/// Legacy entry point without event-bus wiring. Kept so existing embedders
77/// (e.g. tests / the `assay-lua` runtime harness) keep compiling; the engine
78/// binary goes through `serve_with_bus` so dashboard SSE + dispatch-wakeup
79/// loop have a live bus.
80pub async fn serve(
81    store: impl WorkflowStore + 'static,
82    port: u16,
83    auth_mode: AuthMode,
84) -> anyhow::Result<()> {
85    serve_inner(store, port, auth_mode, None, None).await
86}
87
88/// Like `serve`, but lets the embedder pass its own semver so
89/// `/api/v1/version` reflects the binary users are actually running.
90pub async fn serve_with_version(
91    store: impl WorkflowStore + 'static,
92    port: u16,
93    auth_mode: AuthMode,
94    binary_version: Option<&'static str>,
95) -> anyhow::Result<()> {
96    serve_inner(store, port, auth_mode, binary_version, None).await
97}
98
99/// Preferred entry point for the `assay-engine` binary. Wires the
100/// engine-wide `EngineEventBus` into the workflow context so emits
101/// from state-mutating methods reach the SSE stream + dispatch-wakeup
102/// loop.
103pub async fn serve_with_bus(
104    store: impl WorkflowStore + 'static,
105    bus: Arc<dyn EngineEventBus>,
106    port: u16,
107    auth_mode: AuthMode,
108    binary_version: Option<&'static str>,
109) -> anyhow::Result<()> {
110    serve_inner(store, port, auth_mode, binary_version, Some(bus)).await
111}
112
113async fn serve_inner(
114    store: impl WorkflowStore + 'static,
115    port: u16,
116    auth_mode: AuthMode,
117    binary_version: Option<&'static str>,
118    bus: Option<Arc<dyn EngineEventBus>>,
119) -> anyhow::Result<()> {
120    let store = Arc::new(store);
121    let mut ctx = WorkflowCtx::start(store).with_auth_mode(auth_mode.clone());
122    if let Some(v) = binary_version {
123        ctx = ctx.with_binary_version(v);
124    }
125    if let Some(b) = bus {
126        ctx = ctx.with_event_bus(WorkflowEventBus::new(b));
127    }
128
129    let mode_desc = auth_mode.describe();
130    let state = Arc::new(ctx);
131
132    let app = router(state);
133    let listener = tokio::net::TcpListener::bind(format!("0.0.0.0:{port}")).await?;
134    info!("Workflow API listening on 0.0.0.0:{port} (auth: {mode_desc})");
135
136    axum::serve(listener, app).await?;
137    Ok(())
138}
assay_workflow/api/mod.rs

assay_workflow/api/
mod.rs
