1use axum::Router;
31use axum::extract::{Path, Query, State};
32use axum::http::{HeaderMap, StatusCode, header};
33use axum::response::{IntoResponse, Json, Response};
34use axum::routing::{get, post};
35use serde::{Deserialize, Serialize};
36use serde_json::{Value, json};
37
38use assay_workflow::WorkflowStore;
39
40use crate::config::{BackendConfig, EngineConfig};
41use crate::state::EngineState;
42
43pub fn router<S>() -> Router<EngineState<S>>
47where
48 S: WorkflowStore + Clone + 'static,
49{
50 Router::new()
51 .route("/api/v1/engine/core/info", get(engine_info::<S>))
52 .route("/api/v1/engine/core/health", get(engine_health::<S>))
53 .route(
54 "/api/v1/engine/core/active-modules",
55 get(active_modules::<S>),
56 )
57 .route("/api/v1/engine/core/modules", get(list_modules::<S>))
58 .route(
59 "/api/v1/engine/core/modules/{name}/toggle",
60 post(toggle_module::<S>),
61 )
62 .route("/api/v1/engine/core/instances", get(list_instances::<S>))
63 .route("/api/v1/engine/core/audit", get(list_audit::<S>))
64 .route("/api/v1/engine/core/config", get(get_config::<S>))
65}
66
67async fn engine_health<S: WorkflowStore + Clone + 'static>(
76 State(s): State<EngineState<S>>,
77) -> Json<Value> {
78 Json(json!({
79 "status": "ok",
80 "engine_version": s.engine_version,
81 "instance_id": s.instance_id.to_string(),
82 "modules": &*s.modules,
83 "leader": true,
88 }))
89}
90
91async fn active_modules<S: WorkflowStore + Clone + 'static>(
99 State(s): State<EngineState<S>>,
100) -> Json<Value> {
101 Json(json!({
102 "modules": &*s.modules,
103 }))
104}
105
106#[derive(Debug, Clone, Serialize)]
111pub struct EngineInfo {
112 pub version: &'static str,
113 pub instance_id: String,
114 pub started_at: f64,
115 pub leader: bool,
116 pub modules: Vec<String>,
117 pub backend_kind: &'static str,
118 pub backend_data_dir: Option<String>,
120 pub backend_url_redacted: Option<String>,
123 pub bind_addr: String,
124 pub public_url: String,
125}
126
127async fn engine_info<S: WorkflowStore + Clone + 'static>(
128 State(s): State<EngineState<S>>,
129) -> Json<EngineInfo> {
130 let cfg: &EngineConfig = &s.engine_config;
131 let (kind, data_dir, url_redacted) = match &cfg.backend {
132 BackendConfig::Sqlite { data_dir, .. } => ("sqlite", Some(data_dir.clone()), None),
133 BackendConfig::Postgres { url } => ("postgres", None, Some(redact_pg_url(url))),
134 };
135 Json(EngineInfo {
136 version: s.engine_version,
137 instance_id: s.instance_id.to_string(),
138 started_at: s.started_at,
139 leader: true,
140 modules: (*s.modules).clone(),
141 backend_kind: kind,
142 backend_data_dir: data_dir,
143 backend_url_redacted: url_redacted,
144 bind_addr: cfg.server.bind_addr.clone(),
145 public_url: cfg.server.public_url.clone(),
146 })
147}
148
149#[derive(Debug, Clone, Serialize)]
154pub struct ModuleEntry {
155 pub name: String,
156 pub enabled: bool,
157 pub enabled_at: Option<f64>,
158 pub enabled_by: Option<String>,
159 pub version: Option<String>,
160 pub config: Value,
161}
162
163#[derive(Debug, Clone, Serialize)]
164pub struct ListModulesResponse {
165 pub items: Vec<ModuleEntry>,
166}
167
168async fn list_modules<S: WorkflowStore + Clone + 'static>(
169 State(s): State<EngineState<S>>,
170 headers: HeaderMap,
171) -> Response {
172 if let Err(r) = require_admin(&headers, &s).await {
173 return *r;
174 }
175 let items = match list_module_records(&s.engine_config).await {
176 Ok(v) => v,
177 Err(e) => return server_error(&format!("list modules: {e}")),
178 };
179 let entries = items
180 .into_iter()
181 .map(|m| ModuleEntry {
182 name: m.name,
183 enabled: m.enabled,
184 enabled_at: m.enabled_at,
185 enabled_by: m.enabled_by,
186 version: m.version,
187 config: m.config,
188 })
189 .collect();
190 (StatusCode::OK, Json(ListModulesResponse { items: entries })).into_response()
191}
192
193#[derive(Debug, Clone, Deserialize)]
194pub struct ToggleBody {
195 pub enabled: Option<bool>,
198}
199
200#[derive(Debug, Clone, Serialize)]
201pub struct ToggleResponse {
202 pub enabled: bool,
203 pub restart_required: bool,
204 pub message: String,
205}
206
207async fn toggle_module<S: WorkflowStore + Clone + 'static>(
208 State(s): State<EngineState<S>>,
209 headers: HeaderMap,
210 Path(name): Path<String>,
211 body: Option<Json<ToggleBody>>,
212) -> Response {
213 if let Err(r) = require_admin(&headers, &s).await {
214 return *r;
215 }
216 let modules = match list_module_records(&s.engine_config).await {
218 Ok(v) => v,
219 Err(e) => return server_error(&format!("list modules: {e}")),
220 };
221 let Some(current) = modules.iter().find(|m| m.name == name) else {
222 return (
223 StatusCode::NOT_FOUND,
224 Json(json!({"error": "unknown module name"})),
225 )
226 .into_response();
227 };
228 let target = body
229 .and_then(|Json(b)| b.enabled)
230 .unwrap_or(!current.enabled);
231 let actor = bearer_token(&headers).map(short_actor);
232 if let Err(e) = set_module_enabled(&s.engine_config, &name, target, actor.as_deref()).await {
233 return server_error(&format!("set enabled: {e}"));
234 }
235 if let Err(e) = audit_module_toggle(&s.engine_config, &name, target, actor.as_deref()).await {
236 tracing::warn!(?e, "engine.audit insert failed for module toggle");
238 }
239 let msg = if target {
240 format!("module {name} marked enabled — restart engine to load")
241 } else {
242 format!("module {name} marked disabled — restart engine to unload")
243 };
244 (
245 StatusCode::OK,
246 Json(ToggleResponse {
247 enabled: target,
248 restart_required: true,
249 message: msg,
250 }),
251 )
252 .into_response()
253}
254
255#[derive(Debug, Clone, Serialize)]
260pub struct InstanceEntry {
261 pub id: String,
262 pub started_at: f64,
263 pub last_heartbeat: f64,
264 pub namespaces: Vec<String>,
265 pub version: Option<String>,
266}
267
268#[derive(Debug, Clone, Serialize)]
269pub struct ListInstancesResponse {
270 pub items: Vec<InstanceEntry>,
271}
272
273#[derive(Debug, Clone, Default, Deserialize)]
274pub struct PageQuery {
275 #[serde(default)]
276 pub limit: Option<i64>,
277 #[serde(default)]
278 pub offset: Option<i64>,
279}
280
281async fn list_instances<S: WorkflowStore + Clone + 'static>(
282 State(s): State<EngineState<S>>,
283 headers: HeaderMap,
284 Query(_q): Query<PageQuery>,
285) -> Response {
286 if let Err(r) = require_admin(&headers, &s).await {
287 return *r;
288 }
289 let items = match list_instance_records(&s.engine_config).await {
290 Ok(v) => v,
291 Err(e) => return server_error(&format!("list instances: {e}")),
292 };
293 let entries = items
294 .into_iter()
295 .map(|i| InstanceEntry {
296 id: i.id,
297 started_at: i.started_at,
298 last_heartbeat: i.last_heartbeat,
299 namespaces: i.namespaces,
300 version: i.version,
301 })
302 .collect();
303 (
304 StatusCode::OK,
305 Json(ListInstancesResponse { items: entries }),
306 )
307 .into_response()
308}
309
310#[derive(Debug, Clone, Default, Deserialize)]
315pub struct AuditQuery {
316 #[serde(default)]
317 pub limit: Option<i64>,
318 #[serde(default)]
319 pub offset: Option<i64>,
320 #[serde(default)]
321 pub actor: Option<String>,
322 #[serde(default)]
323 pub action: Option<String>,
324 #[serde(default)]
325 pub since: Option<f64>,
326 #[serde(default)]
327 pub until: Option<f64>,
328}
329
330#[derive(Debug, Clone, Serialize)]
331pub struct AuditEntry {
332 pub id: String,
333 pub ts: f64,
334 pub actor: Option<String>,
335 pub action: String,
336 pub details: Value,
337}
338
339#[derive(Debug, Clone, Serialize)]
340pub struct ListAuditResponse {
341 pub items: Vec<AuditEntry>,
342 pub total: i64,
343 pub limit: i64,
344 pub offset: i64,
345}
346
347async fn list_audit<S: WorkflowStore + Clone + 'static>(
348 State(s): State<EngineState<S>>,
349 headers: HeaderMap,
350 Query(q): Query<AuditQuery>,
351) -> Response {
352 if let Err(r) = require_admin(&headers, &s).await {
353 return *r;
354 }
355 let limit = q.limit.unwrap_or(50).clamp(1, 500);
356 let offset = q.offset.unwrap_or(0).max(0);
357 let (rows, total) = match list_audit_records(
358 &s.engine_config,
359 limit,
360 offset,
361 q.actor.as_deref(),
362 q.action.as_deref(),
363 q.since,
364 q.until,
365 )
366 .await
367 {
368 Ok(v) => v,
369 Err(e) => return server_error(&format!("list audit: {e}")),
370 };
371 let items = rows
372 .into_iter()
373 .map(|a| AuditEntry {
374 id: a.id,
375 ts: a.ts,
376 actor: a.actor,
377 action: a.action,
378 details: a.details,
379 })
380 .collect();
381 (
382 StatusCode::OK,
383 Json(ListAuditResponse {
384 items,
385 total,
386 limit,
387 offset,
388 }),
389 )
390 .into_response()
391}
392
393async fn get_config<S: WorkflowStore + Clone + 'static>(
398 State(s): State<EngineState<S>>,
399 headers: HeaderMap,
400) -> Response {
401 if let Err(r) = require_admin(&headers, &s).await {
402 return *r;
403 }
404 let mut value = match serde_json::to_value(&*s.engine_config) {
405 Ok(v) => v,
406 Err(e) => return server_error(&format!("serialise config: {e}")),
407 };
408 redact_secrets(&mut value);
409 (StatusCode::OK, Json(value)).into_response()
410}
411
412fn redact_secrets(v: &mut Value) {
418 let placeholder = Value::String("[REDACTED]".to_string());
419 match v {
420 Value::Object(map) => {
421 let keys: Vec<String> = map.keys().cloned().collect();
424 for k in keys {
425 let lk = k.to_lowercase();
426 if k == "admin_api_keys" {
427 if let Some(Value::Array(arr)) = map.get_mut(&k) {
428 for entry in arr {
429 *entry = placeholder.clone();
430 }
431 }
432 continue;
433 }
434 if (lk.contains("password")
435 || lk.contains("secret")
436 || lk.contains("api_key")
437 || lk.contains("api-key"))
438 && let Some(slot) = map.get_mut(&k)
439 {
440 match slot {
441 Value::String(_) => *slot = placeholder.clone(),
442 Value::Array(arr) => {
443 for entry in arr {
444 if let Value::String(_) = entry {
445 *entry = placeholder.clone();
446 } else {
447 redact_secrets(entry);
448 }
449 }
450 }
451 other => redact_secrets(other),
452 }
453 continue;
454 }
455 if let Some(slot) = map.get_mut(&k) {
456 redact_secrets(slot);
457 }
458 }
459 }
460 Value::Array(arr) => {
461 for entry in arr {
462 redact_secrets(entry);
463 }
464 }
465 _ => {}
466 }
467}
468
469async fn require_admin<S: WorkflowStore + Clone + 'static>(
477 headers: &HeaderMap,
478 state: &EngineState<S>,
479) -> Result<(), Box<Response>> {
480 let keys = crate::state::AdminApiKeys(std::sync::Arc::clone(&state.admin_api_keys));
481 assay_auth::gate::require_admin_bearer(headers, &keys)
482}
483
484fn bearer_token(headers: &HeaderMap) -> Option<String> {
485 let raw = headers
486 .get(header::AUTHORIZATION)
487 .and_then(|v| v.to_str().ok())?;
488 raw.strip_prefix("Bearer ")
489 .or_else(|| raw.strip_prefix("bearer "))
490 .map(|s| s.trim().to_string())
491}
492
493fn short_actor(token: String) -> String {
498 let t = token.trim();
499 if t.len() <= 6 {
500 return format!("admin:****{t}");
501 }
502 let tail = &t[t.len() - 6..];
503 format!("admin:****{tail}")
504}
505
506fn server_error(msg: &str) -> Response {
507 (
508 StatusCode::INTERNAL_SERVER_ERROR,
509 Json(json!({"error": "server_error", "error_description": msg})),
510 )
511 .into_response()
512}
513
514async fn list_module_records(
523 cfg: &EngineConfig,
524) -> anyhow::Result<Vec<assay_domain::engine::ModuleRecord>> {
525 match &cfg.backend {
526 #[cfg(feature = "backend-postgres")]
527 BackendConfig::Postgres { url } => {
528 let pool = sqlx::PgPool::connect(url)
529 .await
530 .map_err(|e| anyhow::anyhow!("connect pg: {e}"))?;
531 let schema = assay_domain::engine::PgEngineSchema::new(pool);
532 let rows = schema
533 .list_modules()
534 .await
535 .map_err(|e| anyhow::anyhow!("list modules (pg): {e}"))?;
536 Ok(rows)
537 }
538 #[cfg(feature = "backend-sqlite")]
539 BackendConfig::Sqlite { .. } => {
540 let pool = open_sqlite_engine_pool(cfg).await?;
541 let schema = assay_domain::engine::SqliteEngineSchema::new(pool);
542 let rows = schema
543 .list_modules()
544 .await
545 .map_err(|e| anyhow::anyhow!("list modules (sqlite): {e}"))?;
546 Ok(rows)
547 }
548 #[allow(unreachable_patterns)]
549 _ => anyhow::bail!("backend not enabled at compile time"),
550 }
551}
552
553async fn set_module_enabled(
554 cfg: &EngineConfig,
555 name: &str,
556 enabled: bool,
557 actor: Option<&str>,
558) -> anyhow::Result<bool> {
559 match &cfg.backend {
560 #[cfg(feature = "backend-postgres")]
561 BackendConfig::Postgres { url } => {
562 let pool = sqlx::PgPool::connect(url).await?;
563 let schema = assay_domain::engine::PgEngineSchema::new(pool);
564 schema.set_module_enabled(name, enabled, actor).await
565 }
566 #[cfg(feature = "backend-sqlite")]
567 BackendConfig::Sqlite { .. } => {
568 let pool = open_sqlite_engine_pool(cfg).await?;
569 let schema = assay_domain::engine::SqliteEngineSchema::new(pool);
570 schema.set_module_enabled(name, enabled, actor).await
571 }
572 #[allow(unreachable_patterns)]
573 _ => anyhow::bail!("backend not enabled at compile time"),
574 }
575}
576
577async fn audit_module_toggle(
578 cfg: &EngineConfig,
579 name: &str,
580 enabled: bool,
581 actor: Option<&str>,
582) -> anyhow::Result<()> {
583 let details = json!({"module": name, "enabled": enabled});
584 match &cfg.backend {
585 #[cfg(feature = "backend-postgres")]
586 BackendConfig::Postgres { url } => {
587 let pool = sqlx::PgPool::connect(url).await?;
588 let schema = assay_domain::engine::PgEngineSchema::new(pool);
589 schema.audit(actor, "engine.module.toggle", &details).await
590 }
591 #[cfg(feature = "backend-sqlite")]
592 BackendConfig::Sqlite { .. } => {
593 let pool = open_sqlite_engine_pool(cfg).await?;
594 let schema = assay_domain::engine::SqliteEngineSchema::new(pool);
595 schema.audit(actor, "engine.module.toggle", &details).await
596 }
597 #[allow(unreachable_patterns)]
598 _ => anyhow::bail!("backend not enabled at compile time"),
599 }
600}
601
602async fn list_instance_records(
603 cfg: &EngineConfig,
604) -> anyhow::Result<Vec<assay_domain::engine::InstanceRecord>> {
605 match &cfg.backend {
606 #[cfg(feature = "backend-postgres")]
607 BackendConfig::Postgres { url } => {
608 let pool = sqlx::PgPool::connect(url).await?;
609 let schema = assay_domain::engine::PgEngineSchema::new(pool);
610 schema.list_instances().await
611 }
612 #[cfg(feature = "backend-sqlite")]
613 BackendConfig::Sqlite { .. } => {
614 let pool = open_sqlite_engine_pool(cfg).await?;
615 let schema = assay_domain::engine::SqliteEngineSchema::new(pool);
616 schema.list_instances().await
617 }
618 #[allow(unreachable_patterns)]
619 _ => anyhow::bail!("backend not enabled at compile time"),
620 }
621}
622
623async fn list_audit_records(
624 cfg: &EngineConfig,
625 limit: i64,
626 offset: i64,
627 actor: Option<&str>,
628 action: Option<&str>,
629 since: Option<f64>,
630 until: Option<f64>,
631) -> anyhow::Result<(Vec<assay_domain::engine::AuditRecord>, i64)> {
632 match &cfg.backend {
633 #[cfg(feature = "backend-postgres")]
634 BackendConfig::Postgres { url } => {
635 let pool = sqlx::PgPool::connect(url).await?;
636 let schema = assay_domain::engine::PgEngineSchema::new(pool);
637 schema
638 .list_audit(limit, offset, actor, action, since, until)
639 .await
640 }
641 #[cfg(feature = "backend-sqlite")]
642 BackendConfig::Sqlite { .. } => {
643 let pool = open_sqlite_engine_pool(cfg).await?;
644 let schema = assay_domain::engine::SqliteEngineSchema::new(pool);
645 schema
646 .list_audit(limit, offset, actor, action, since, until)
647 .await
648 }
649 #[allow(unreachable_patterns)]
650 _ => anyhow::bail!("backend not enabled at compile time"),
651 }
652}
653
654#[cfg(feature = "backend-sqlite")]
660async fn open_sqlite_engine_pool(cfg: &EngineConfig) -> anyhow::Result<sqlx::SqlitePool> {
661 use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
662 use std::str::FromStr;
663
664 let data_dir = cfg
665 .backend
666 .sqlite_data_dir()
667 .ok_or_else(|| anyhow::anyhow!("sqlite_data_dir missing on non-sqlite backend"))?;
668 let path = format!("file:{data_dir}/engine.db?mode=rw");
669 let opts = SqliteConnectOptions::from_str("sqlite::memory:")?.create_if_missing(true);
670 let pool = SqlitePoolOptions::new()
671 .max_connections(1)
672 .after_connect(move |conn, _meta| {
673 let path = path.clone();
674 Box::pin(async move {
675 use sqlx::Executor;
676 conn.execute(format!("ATTACH DATABASE '{path}' AS engine").as_str())
677 .await?;
678 Ok(())
679 })
680 })
681 .connect_with(opts)
682 .await
683 .map_err(|e| anyhow::anyhow!("connect engine.db: {e}"))?;
684 Ok(pool)
685}
686
687fn redact_pg_url(raw: &str) -> String {
694 let Ok(u) = url::Url::parse(raw) else {
695 return "[REDACTED]".to_string();
696 };
697 let scheme = u.scheme();
698 let host = u.host_str().unwrap_or("");
699 let port = u.port().map(|p| format!(":{p}")).unwrap_or_default();
700 let path = u.path();
701 let query = match u.query() {
702 Some(q) => format!("?{q}"),
703 None => String::new(),
704 };
705 let userinfo = if !u.username().is_empty() || u.password().is_some() {
706 "[REDACTED]@"
707 } else {
708 ""
709 };
710 format!("{scheme}://{userinfo}{host}{port}{path}{query}")
711}
712
713#[cfg(test)]
714mod tests {
715 use super::*;
716
717 #[test]
718 fn redact_secrets_replaces_admin_api_keys_array() {
719 let mut v = json!({
720 "auth": {
721 "admin_api_keys": ["secret1", "secret2"],
722 "issuer": "https://issuer.example",
723 }
724 });
725 redact_secrets(&mut v);
726 let arr = v["auth"]["admin_api_keys"].as_array().unwrap();
727 assert!(arr.iter().all(|x| x.as_str() == Some("[REDACTED]")));
728 assert_eq!(v["auth"]["issuer"], json!("https://issuer.example"));
730 }
731
732 #[test]
733 fn redact_secrets_replaces_password_fields() {
734 let mut v = json!({"backend": {"password": "hunter2"}});
735 redact_secrets(&mut v);
736 assert_eq!(v["backend"]["password"], json!("[REDACTED]"));
737 }
738
739 #[test]
740 fn redact_pg_url_strips_userinfo() {
741 let raw = "postgres://alice:hunter2@db.example.com:5432/assay";
742 let red = redact_pg_url(raw);
743 assert!(!red.contains("hunter2"), "password leak: {red}");
744 assert!(red.contains("[REDACTED]"));
745 assert!(red.contains("db.example.com:5432"));
746 assert!(red.contains("/assay"));
747 }
748
749 #[test]
754 fn short_actor_safely_truncates() {
755 let s = short_actor("abcdef0123456789".to_string());
756 assert_eq!(s, "admin:****456789");
757 let s = short_actor("xyz".to_string());
759 assert!(s.starts_with("admin:****"));
760 }
761}