use std::sync::Arc;
use assay_dashboard::{DashboardCtx, WhitelabelConfig};
use assay_domain::events::EngineEventBus;
use assay_workflow::{WorkflowCtx, WorkflowStore};
use crate::config::EngineConfig;
use crate::init::EngineBoot;
use crate::state::EngineState;
#[non_exhaustive]
pub struct EmbeddedEngine {
pub router: axum::Router,
pub pool: EmbeddedPool,
pub instance_id: uuid::Uuid,
pub modules: Vec<String>,
pub engine_version: &'static str,
}
#[non_exhaustive]
pub enum EmbeddedPool {
#[cfg(feature = "backend-postgres")]
Postgres(sqlx::PgPool),
#[cfg(feature = "backend-sqlite")]
Sqlite(sqlx::SqlitePool),
}
pub async fn build(cfg: EngineConfig) -> anyhow::Result<EmbeddedEngine> {
let boot = EngineBoot::run(&cfg).await?;
match boot {
#[cfg(feature = "backend-postgres")]
EngineBoot::Postgres(b) => build_pg(cfg, b).await,
#[cfg(feature = "backend-sqlite")]
EngineBoot::Sqlite(b) => build_sqlite(cfg, b).await,
}
}
#[cfg(feature = "backend-postgres")]
async fn build_pg(cfg: EngineConfig, b: crate::init::PgBoot) -> anyhow::Result<EmbeddedEngine> {
let store = assay_workflow::PostgresStore::from_pool(b.pool.clone())
.await
.map_err(|e| anyhow::anyhow!("workflow store (pg): {e}"))?;
let auth_ctx = crate::build_auth_ctx_pg(&cfg, &b.pool).await?;
#[cfg(feature = "vault")]
let vault_ctx = crate::build_vault_ctx_pg(&b.modules, &b.pool).await?;
#[cfg(not(feature = "vault"))]
let vault_ctx: Option<()> = None;
compose(
cfg,
store,
b.bus,
b.modules,
b.instance_id,
Some(auth_ctx),
vault_ctx,
EmbeddedPool::Postgres(b.pool),
)
.await
}
#[cfg(feature = "backend-sqlite")]
async fn build_sqlite(
cfg: EngineConfig,
b: crate::init::SqliteBoot,
) -> anyhow::Result<EmbeddedEngine> {
let store = assay_workflow::SqliteStore::from_attached_pool(b.pool.clone())
.await
.map_err(|e| anyhow::anyhow!("workflow store (sqlite): {e}"))?;
let auth_ctx = crate::build_auth_ctx_sqlite(&cfg, &b.pool).await?;
#[cfg(feature = "vault")]
let vault_ctx = crate::build_vault_ctx_sqlite(&b.modules, &b.pool).await?;
#[cfg(not(feature = "vault"))]
let vault_ctx: Option<()> = None;
compose(
cfg,
store,
b.bus,
b.modules,
b.instance_id,
Some(auth_ctx),
vault_ctx,
EmbeddedPool::Sqlite(b.pool),
)
.await
}
#[allow(clippy::too_many_arguments)]
async fn compose<S: WorkflowStore + Clone + 'static>(
cfg: EngineConfig,
store: S,
bus: Arc<dyn EngineEventBus>,
modules: Vec<String>,
instance_id: uuid::Uuid,
auth_ctx: Option<assay_auth::AuthCtx>,
#[cfg(feature = "vault")] vault_ctx: Option<assay_vault::VaultCtx>,
#[cfg(not(feature = "vault"))] _vault_ctx: Option<()>,
pool: EmbeddedPool,
) -> anyhow::Result<EmbeddedEngine> {
if let Some(auth) = auth_ctx.as_ref() {
let user_count = auth
.users
.count_users(None)
.await
.map_err(|e| anyhow::anyhow!("count auth.users: {e}"))?;
if user_count == 0
&& cfg.auth.admin_api_keys.is_empty()
&& cfg.auth.external_issuers().is_empty()
{
anyhow::bail!(
"engine refuses to start: no operator users exist, \
`auth.admin_api_keys` is empty, and no external issuers \
are configured. Either run `assay-engine bootstrap-admin \
--email <e> --password <p>` to seed the first user, add \
at least one entry to `auth.admin_api_keys` in \
engine.toml as a break-glass, or configure \
`[[auth.external_issuers]]` with an upstream OIDC \
provider (e.g. Hydra) that mints the JWTs your callers \
forward."
);
}
}
let workflow_ctx: Arc<WorkflowCtx<S>> =
crate::server::build_workflow_ctx_with_bus(store, Arc::clone(&bus));
tokio::spawn(assay_workflow::events_cleanup::run_events_cleanup(
Arc::clone(&bus),
std::time::Duration::from_secs(3600),
cfg.engine_events_ttl_secs,
));
let whitelabel = Arc::new(WhitelabelConfig::from_env());
let asset_version = env!("CARGO_PKG_VERSION").to_string();
let dashboard_ctx = Arc::new(DashboardCtx::new(whitelabel, asset_version));
let admin_api_keys = Arc::new(cfg.auth.admin_api_keys.clone());
let started_at = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_secs_f64();
let engine_config = Arc::new(cfg);
let state = EngineState {
workflow: workflow_ctx,
dashboard: dashboard_ctx,
auth: auth_ctx,
#[cfg(feature = "vault")]
vault: vault_ctx,
admin_api_keys,
modules: Arc::new(modules.clone()),
instance_id,
engine_version: env!("CARGO_PKG_VERSION"),
started_at,
engine_config,
};
let router = crate::server::build_app(state);
Ok(EmbeddedEngine {
router,
pool,
instance_id,
modules,
engine_version: env!("CARGO_PKG_VERSION"),
})
}
pub async fn migrate(cfg: &EngineConfig) -> anyhow::Result<()> {
let _boot = EngineBoot::run(cfg).await?;
Ok(())
}