arcature 2026.2.1

//! The ordered engine startup — connect subsystems in a deterministic order
//! from a single `PgPool` (engine spec §28).
//!
//! Startup order (each step is cfg-gated; only configured subsystems connect):
//!
//! ```text
//! db (builds the one PgPool)
//!   └─ jobs.migrate() (shares the pool)
//!   └─ Worker::run (shares the pool; spawned as a task)
//! cache   (independent connection)
//! storage (independent connection)
//! mail    (synchronous: builds the SMTP transport)
//! ```
//!
//! The `Db` is built first because `jobs` shares its `PgPool`. A failure in
//! any step aborts startup and the already-started subsystems are torn down
//! (reverse order) before the error is returned — no partial startup leaks.

use crate::application::error::{EngineError, StartupError};
use crate::application::resources::Resources;
use crate::application::ty::Application;

#[cfg(feature = "jobs")]
use crate::application::resources::WorkerHandle;

/// The live subsystem handles plus the running worker task (if any).
///
/// Built by [`startup`] and consumed by
/// [`super::shutdown`]. The `Resources` are passed to the application's
/// state-building closure; the `WorkerHandle` is held by the engine for
/// coordinated shutdown.
pub(crate) struct Started {
    pub(crate) resources: Resources,
    #[cfg(feature = "jobs")]
    pub(crate) worker: Option<WorkerHandle>,
}

/// Connect all configured subsystems in startup order.
///
/// Takes the lifecycle config from `Application` (the cfg-gated `Option`
/// fields) and returns the live handles plus the running worker task. On
/// failure, already-started subsystems are torn down (reverse order) before
/// the error is returned.
///
/// The `Application` is passed by value so the startup can move the config
/// out; the resolved `Application<S>` (with lifecycle config set to `None`,
/// routes/state untouched) is returned alongside the handles so the caller
/// can resolve state and serve it.
pub(crate) async fn startup<S>(
    app: Application<S>,
) -> Result<(Started, Application<S>), EngineError>
where
    S: Clone + Send + Sync + 'static,
{
    let Application {
        routes,
        proxy,
        bind_address,
        port,
        #[cfg(feature = "inertia")]
        inertia_config,
        #[cfg(feature = "inertia")]
        page_contracts,
        #[cfg(feature = "pages")]
        pages,
        #[cfg(feature = "pages")]
        maintenance_guard,
        #[cfg(feature = "db")]
        database,
        #[cfg(feature = "cache")]
        cache_config,
        #[cfg(feature = "storage")]
        storage_config,
        #[cfg(feature = "mail")]
        mail_config,
        #[cfg(feature = "jobs")]
        jobs_registry,
        #[cfg(feature = "jobs")]
        worker_config,
        #[cfg(feature = "dx")]
        error_mapping,
        #[cfg(feature = "dev-proxy")]
        dev_proxy_endpoint,
    } = app;

    // The resolved application (lifecycle config consumed) — returned to the
    // caller for serving. Pipeline config (inertia, pages, maintenance,
    // error mapping) stays.
    let resolved = Application {
        routes,
        proxy,
        bind_address,
        port,
        #[cfg(feature = "inertia")]
        inertia_config,
        #[cfg(feature = "inertia")]
        page_contracts,
        #[cfg(feature = "pages")]
        pages,
        #[cfg(feature = "pages")]
        maintenance_guard,
        #[cfg(feature = "db")]
        database: None,
        #[cfg(feature = "cache")]
        cache_config: None,
        #[cfg(feature = "storage")]
        storage_config: None,
        #[cfg(feature = "mail")]
        mail_config: None,
        #[cfg(feature = "jobs")]
        jobs_registry: None,
        #[cfg(feature = "jobs")]
        worker_config: None,
        #[cfg(feature = "dx")]
        error_mapping,
        #[cfg(feature = "dev-proxy")]
        dev_proxy_endpoint,
    };

    // Each handle is a mutable local so that on any later-step failure we can
    // hand the already-started subsystems to `shutdown` (reverse order) —
    // matching the documented contract: "no partial startup leaks". Bare `?`
    // would drop the handles, and dropping `WorkerHandle` detaches the worker
    // task (the `CancellationToken` is never cancelled) leaving a zombie with a
    // live `PgPool` clone. Calling `shutdown` cancels the worker, drains it,
    // and closes the pool properly.
    #[cfg(feature = "db")]
    let mut db_handle: Option<crate::db::Db> = None;
    #[cfg(feature = "jobs")]
    let mut jobs_handle: Option<arcature_jobs::Jobs> = None;
    #[cfg(feature = "jobs")]
    let mut worker_handle: Option<WorkerHandle> = None;
    #[cfg(feature = "cache")]
    let mut cache_handle: Option<crate::cache::Cache> = None;
    #[cfg(feature = "storage")]
    let mut storage_handle: Option<crate::storage::Storage> = None;
    #[cfg(feature = "mail")]
    let mut mail_handle: Option<crate::mail::Mailer> = None;

    // Tear down everything started so far (reverse order) and return `err`.
    // The `shutdown` call is best-effort: its error is discarded because the
    // *startup* error is the root cause the operator needs to see. The
    // invariant this enforces: no zombie worker task and no leaked pool after
    // a partial startup.
    macro_rules! fail {
        ($err:expr) => {{
            let partial = Started {
                resources: Resources {
                    #[cfg(feature = "db")]
                    db: db_handle.take(),
                    #[cfg(feature = "cache")]
                    cache: cache_handle.take(),
                    #[cfg(feature = "storage")]
                    storage: storage_handle.take(),
                    #[cfg(feature = "mail")]
                    mail: mail_handle.take(),
                    #[cfg(feature = "jobs")]
                    jobs: jobs_handle.take(),
                },
                #[cfg(feature = "jobs")]
                worker: worker_handle.take(),
            };
            let _ = super::shutdown::shutdown(partial).await;
            return Err($err);
        }};
    }

    // ── db (builds the one PgPool) ──────────────────────────────────────
    // db is the first step — nothing to tear down if it fails, so the
    // `fail!` call here constructs an empty `Started` (all handles `None`)
    // and `shutdown` is a no-op. Using `fail!` (not bare `?`) keeps the
    // macro referenced under every feature combination — without this, a
    // `db`-only build (no cache/storage/mail/jobs) never invokes `fail!`
    // and `-D warnings` rejects the unused macro.
    #[cfg(feature = "db")]
    if let Some(config) = database {
        match crate::db::Db::connect(config).await {
            Ok(db) => db_handle = Some(db),
            Err(source) => fail!(EngineError::Startup {
                subsystem: "db",
                stage: "connect",
                source: StartupError::Db(source),
            }),
        }
    }

    // ── jobs.migrate() + Worker::run (shares the pool) ───────────────────
    #[cfg(feature = "jobs")]
    if let Some(jobs_registry) = jobs_registry {
        // `jobs` requires `db` as a Cargo feature; if the db was not
        // configured (no `DbConfig`), this is a configuration error. db may
        // already be live, so tear it down on this config error.
        let pool = match db_handle.as_ref().map(|db| db.sqlx().clone()) {
            Some(pool) => pool,
            None => fail!(EngineError::Startup {
                subsystem: "jobs",
                stage: "pool",
                source: StartupError::JobsMigrate(arcature_jobs::MigrateError::Database {
                    source: arcature_db::sqlx::Error::PoolClosed,
                })
            }),
        };

        let jobs = arcature_jobs::Jobs::new(pool.clone());
        if let Err(source) = jobs.migrate().await {
            fail!(EngineError::Startup {
                subsystem: "jobs",
                stage: "migrate",
                source: StartupError::JobsMigrate(source),
            });
        }

        // Resolve the registry now that the pool is live. `Static` returns
        // the pre-built registry; `WithDb` invokes the app-supplied closure
        // with the connected `Db` so handlers can capture a `Db` clone. The
        // `Db` is guaranteed present here — the pool match above succeeded, so
        // `db_handle` is `Some`. The borrow ends after the call; the registry
        // (and any handler-captured `Db` clones) own their own `Arc` handles.
        let db = db_handle
            .as_ref()
            .expect("db configured before the jobs worker (jobs requires db)");
        let registry = jobs_registry.resolve(db);

        let worker_config = worker_config.unwrap_or_default();
        let worker = arcature_jobs::Worker::builder(pool, registry)
            .config(worker_config)
            .build();
        let shutdown = tokio_util::sync::CancellationToken::new();
        let worker_shutdown = shutdown.clone();
        let join = tokio::spawn(async move { worker.run(worker_shutdown).await });
        jobs_handle = Some(jobs);
        worker_handle = Some(WorkerHandle { join, shutdown });
    }

    // ── cache (independent connection) ──────────────────────────────────
    #[cfg(feature = "cache")]
    if let Some(config) = cache_config {
        match crate::cache::Cache::connect(config).await {
            Ok(cache) => cache_handle = Some(cache),
            Err(source) => fail!(EngineError::Startup {
                subsystem: "cache",
                stage: "connect",
                source: StartupError::Cache(source),
            }),
        }
    }

    // ── storage (independent connection) ─────────────────────────────────
    #[cfg(feature = "storage")]
    if let Some(config) = storage_config {
        match crate::storage::Storage::connect(config).await {
            Ok(storage) => storage_handle = Some(storage),
            Err(source) => fail!(EngineError::Startup {
                subsystem: "storage",
                stage: "connect",
                source: StartupError::Storage(source),
            }),
        }
    }

    // ── mail (synchronous: builds the SMTP transport) ─────────────────────
    #[cfg(feature = "mail")]
    if let Some(config) = mail_config {
        match crate::mail::Mailer::smtp(config) {
            Ok(mailer) => mail_handle = Some(mailer),
            Err(source) => fail!(EngineError::Startup {
                subsystem: "mail",
                stage: "connect",
                source: StartupError::Mail(source),
            }),
        }
    }

    Ok((
        Started {
            resources: Resources {
                #[cfg(feature = "db")]
                db: db_handle,
                #[cfg(feature = "cache")]
                cache: cache_handle,
                #[cfg(feature = "storage")]
                storage: storage_handle,
                #[cfg(feature = "mail")]
                mail: mail_handle,
                #[cfg(feature = "jobs")]
                jobs: jobs_handle,
            },
            #[cfg(feature = "jobs")]
            worker: worker_handle,
        },
        resolved,
    ))
}