boatramp_node/

node.rs

//! The node-graph assembly: given a built store (blobs + KV), a configured
//! [`Auth`](boatramp_server::Auth), and resolved
//! [`ServerOptions`](boatramp_server::ServerOptions), wire the deploy store,
//! handler runtime, compute reconcile loop, and domain-verify reconcile loop
//! into a [`RunningNode`] ready to hand to a transport (`serve_with` & friends).
//!
//! This is the headline extraction of `PLAN-node-library`: the binary's
//! `serve::run` used to inline this wiring, so no embedder or in-process test
//! could exercise the same graph the `boatramp serve` binary runs. `run` now
//! resolves the *environment* (args -> backends -> store, signal handlers,
//! migration, auth) and calls [`assemble`]; the cluster path keeps its own inline
//! copy until a later step converges it here.

use std::path::Path;
use std::sync::Arc;

use boatramp_core::Storage;
use boatramp_core::deploy::DeployStore;
use boatramp_core::kv::KvStore;

use crate::config::ServerConfig;
use crate::error::{Error, Result};

/// How often the compute reconcile loop converges desired vs actual workloads.
/// Defaults to 30s; override with `BOATRAMP_COMPUTE_RECONCILE_TICK_MS` (milliseconds)
/// so compute-backed tests can converge in a fraction of a second instead of
/// waiting a full tick for the launch/scale reconcile.
pub fn compute_reconcile_tick() -> std::time::Duration {
    std::env::var("BOATRAMP_COMPUTE_RECONCILE_TICK_MS")
        .ok()
        .and_then(|s| s.parse::<u64>().ok())
        .filter(|&ms| ms > 0)
        .map(std::time::Duration::from_millis)
        .unwrap_or(std::time::Duration::from_secs(30))
}
/// How often the domain-verify reconcile loop re-checks pending challenges.
pub const DOMAIN_VERIFY_RECONCILE_TICK: std::time::Duration = std::time::Duration::from_secs(60);
/// How long a compute workload may be idle before scale-to-zero sleeps it.
pub const COMPUTE_IDLE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(300);

/// The built store + resolved config handed to [`assemble`]. Owns the blob/KV
/// backends and the auth/options the caller already resolved; borrows the parsed
/// config and data directory.
pub struct NodeInput<'a> {
    /// The full parsed server config (the handler + compute sections are read here).
    pub config: &'a ServerConfig,
    /// The node data directory (per-site SQL, handler state).
    pub data_dir: &'a Path,
    /// The object store built by [`crate::blobs::build_blobs`].
    pub storage: Arc<dyn Storage>,
    /// The metadata KV, already cache-fronted, built by [`crate::backends::build_kv`].
    pub kv: Arc<dyn KvStore>,
    /// The control-plane auth built by [`crate::auth::configure_auth`].
    pub auth: boatramp_server::Auth,
    /// Server options, already carrying the resolved posture, daemon runtime, and
    /// (post-`configure_auth`/`configure_oidc`) issuer / OIDC verifier.
    pub options: boatramp_server::ServerOptions,
    /// The public HTTP serve bind address, if known — used (under
    /// `allow_guest_self_egress`) to let a handler guest's `wasi:http` reach this
    /// instance's own front door over loopback. `None` (an in-process embedder with no
    /// listener) disables self-egress.
    pub serve_addr: Option<std::net::SocketAddr>,
    /// The cloud blob-change watch provider (FA-5b2), if the backend is a cloud one.
    pub watch_provider: Option<Arc<dyn boatramp_core::blob_provision::WatchProvider>>,
    /// The provisioning tier for the watch provider.
    pub provision_tier: boatramp_core::blob_notify::ProvisionTier,
    /// The `wasi:messaging` substrate override for the handler runtime. `None` uses
    /// the single-node default (`LogMessaging` over the same backends); the cluster
    /// path passes its Raft-backed coordinator.
    pub messaging: Option<Arc<dyn boatramp_core::messaging::Messaging>>,
    /// The single leader gate for cron firing + the compute / domain-verify reconcile
    /// loops. Single-node passes an always-true gate (there is one node); the cluster
    /// passes its Raft `is_leader` check so a single node drives each sweep.
    pub is_leader: boatramp_server::CronLeaderGate,
    /// This node's compute scheduler id (`0` single-node; the cluster node id in a
    /// fleet, so replicas are tagged to the right node).
    pub node_id: u64,
    /// The binary the re-exec'd compute workers run as — the container backend's
    /// `__sandbox` jailer and the microVM backends' `__vmm-run`/`__vz-run` VM hosts.
    /// `None` uses this process's own executable (`current_exe`), which is what
    /// `boatramp serve` wants (the child *is* boatramp). An **embedding harness**
    /// whose own binary doesn't implement those subcommands should point this at a
    /// built `boatramp` binary, so it can drive the real container/microVM backends
    /// in-process (only the per-workload worker re-execs; the serving plane stays
    /// embedded). The docker backend needs neither — it talks to a daemon.
    pub worker_exe: Option<std::path::PathBuf>,
}

/// A fully wired node: the deploy store, handler runtime, auth, and options a
/// transport consumes, plus the detached reconcile loops kept alive for the
/// node's serving life. Destructure it and hold `reconcile` across the serve
/// await so the loops outlive assembly.
pub struct RunningNode {
    /// The deploy store (blob + KV) the router serves from.
    pub deploy: DeployStore,
    /// The handler runtime for wasm handlers (a disabled build ⇒ a no-op runtime).
    pub handlers: boatramp_server::HandlerRuntime,
    /// The control-plane auth.
    pub auth: boatramp_server::Auth,
    /// The resolved server options.
    pub options: boatramp_server::ServerOptions,
    /// The detached reconcile loops (compute + domain-verify). Tokio `JoinHandle`s
    /// do not abort on drop, so the loops run for the process life regardless; the
    /// handles are retained so an embedder can join/abort them on shutdown.
    pub reconcile: Vec<tokio::task::JoinHandle<()>>,
}

/// The instance's own serve socket(s) a guest self-call may reach, given the bind `addr` and
/// whether the posture (`allow_guest_self_egress`) permits it. A wildcard bind
/// (`0.0.0.0`/`::`) is reachable over loopback, so it normalizes to `127.0.0.1` **and** `::1`
/// on the serve port; a specific bind is reachable at itself. Empty when disabled or no
/// listener.
fn self_egress_addrs(
    addr: Option<std::net::SocketAddr>,
    enabled: bool,
) -> Vec<std::net::SocketAddr> {
    use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
    let Some(addr) = addr.filter(|_| enabled) else {
        return Vec::new();
    };
    if addr.ip().is_unspecified() {
        let port = addr.port();
        vec![
            SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), port),
            SocketAddr::new(IpAddr::V6(Ipv6Addr::LOCALHOST), port),
        ]
    } else {
        vec![addr]
    }
}

/// Whether a node with **no configured control-plane issuer** should auto-provision an ephemeral
/// in-memory fleet signer (for host session cookies + delegable capabilities) from its serve bind.
///
/// The fleet signer is a distinct trust domain from control-plane admin auth, so a DEV / loopback
/// node with auth disabled should still be able to sign/verify session cookies and capabilities. It
/// is gated **strictly to a loopback bind** (`127.0.0.1`/`::1`) or an in-process embedder with no
/// bind address (`None`): NEVER a public or wildcard (`0.0.0.0`/`::`, reachable off-host) bind,
/// where an ephemeral key would silently invalidate live capabilities across a restart — a
/// public node that wants guest capabilities without control-plane auth must supply a persistent
/// key. (`is_loopback()` is already `false` for a wildcard/unspecified address, so a `0.0.0.0` bind
/// correctly does NOT auto-provision.)
#[cfg(feature = "handlers")]
fn should_autoprovision_fleet_signer(serve_addr: Option<std::net::SocketAddr>) -> bool {
    serve_addr.is_none_or(|a| a.ip().is_loopback())
}

/// Wire [`NodeInput`] into a [`RunningNode`]: build the handler runtime, the
/// deploy store (materializing the reserved `default` project), the compute
/// backends + reconcile loop, and the domain-verify reconcile loop.
///
/// The caller has already built the store and configured auth/OIDC on `options`;
/// this is the pure node-graph wiring, identical to what `boatramp serve` runs.
pub async fn assemble(input: NodeInput<'_>) -> Result<RunningNode> {
    let NodeInput {
        config,
        data_dir,
        storage,
        kv,
        auth,
        options,
        serve_addr,
        watch_provider,
        provision_tier,
        messaging,
        is_leader,
        node_id,
        worker_exe,
    } = input;
    // Copy out the posture scalars up front so `options` can be moved into the
    // returned `RunningNode` without a lingering borrow.
    let max_handler_blob_bytes = options.posture.max_handler_blob_bytes;
    let max_component_bytes = options.posture.max_component_bytes;
    let allow_guest_private_egress = options.posture.allow_guest_private_egress;
    let allow_env_secret_refs = options.posture.allow_env_secret_refs;
    let allow_guest_email = options.posture.allow_guest_email;
    // The instance's own serve socket(s) a guest self-call may reach, when the posture allows
    // it: a wildcard bind (`0.0.0.0`/`::`) is reachable on loopback, so normalize to
    // `127.0.0.1`/`::1`; a specific bind is itself.
    let self_egress_addrs = self_egress_addrs(serve_addr, options.posture.allow_guest_self_egress);
    let allow_shared_kernel = options.posture.allow_shared_kernel_compute;
    let domain_verify_allow_private = options.posture.domain_verify_allow_private;

    // The deploy store the router serves from — built up front so the handler
    // runtime's managed compute-backed `sql` binding can resolve DB endpoints from
    // the same store the reconcile writes.
    let compute_storage = storage.clone();
    let deploy = DeployStore::new(storage, kv.clone());
    // The `[secrets]` envelope (local KEK / Vault) that seals a managed SQL
    // credential at rest. `None` ⇒ no wrapping (a managed DB then fails closed).
    let secrets_envelope = build_secrets_envelope(config.secrets.as_ref(), data_dir)?;
    // The project-scoped internal secret store, built from the same KV + `[secrets]`
    // envelope that seal managed-DB credentials. Backs both the `boatramp:<name>`
    // resolver (wired into the handler runtime below, when that feature is present)
    // and the admin secrets API (threaded into `ServerOptions` unconditionally, so it
    // works on a lean node too). `None` when no envelope is configured — the admin
    // endpoints then fail closed with a clear 501, never a panic.
    let secret_store = secrets_envelope.clone().map(|envelope| {
        Arc::new(boatramp_core::secret_store::SecretStore::new(
            kv.clone(),
            envelope,
        ))
    });
    // The per-TENANT sealed secret store (task #493), sealed with the SAME `[secrets]` envelope.
    // Backs both the control-plane CRUD (threaded into `ServerOptions` below) and — when the
    // `tenant-secrets` feature is compiled — the runtime guest binding (the SAME `Arc` handed to
    // `set_tenant_secret_store`). `None` when no envelope is configured, so the endpoints fail
    // closed with a clear 501 and the guest binding is not built.
    let tenant_secret_store = secrets_envelope.clone().map(|envelope| {
        Arc::new(boatramp_core::secret_store::TenantSecretStore::new(
            kv.clone(),
            envelope,
        ))
    });
    // The project-scoped SMTP email-profile store, built from the same KV + envelope
    // (the password is sealed at rest). Backs the admin API (`options` below,
    // unconditionally, so it works on a lean node) and — when the `email` feature +
    // `allow_guest_email` posture permit — the runtime's host-side profile
    // resolution (wired inside `build_handler_runtime`). `None` with no envelope, so
    // the admin email endpoints fail closed with a clear 501.
    let email_profile_store = secrets_envelope.clone().map(|envelope| {
        Arc::new(boatramp_core::email_config::EmailProfileStore::new(
            kv.clone(),
            envelope,
        ))
    });

    // Dev-posture guest-egress extra CA(s): when the posture permits (off/refused under
    // multi-tenant) AND the operator pointed `BOATRAMP_GUEST_EGRESS_EXTRA_CA_FILE` at a PEM, parse
    // it into trust anchors the guest's outbound `wasi:http` TLS client trusts on TOP of the webpki
    // roots (for a hermetic HTTPS test double). Empty otherwise. A configured-but-unreadable/
    // unparsable file is a hard config error (fail closed), never a silent no-trust.
    let guest_egress_extra_roots =
        load_guest_egress_extra_roots(options.posture.allow_guest_egress_extra_ca)?;

    // The handler runtime reuses the same blob/KV backends (per-site prefixed)
    // for its wasi:blobstore/keyvalue bindings; the sql binding is selected by
    // `[handlers.bindings.sql]` (default: per-site libsql files under <data-dir>).
    let handlers = crate::handlers::build_handler_runtime(
        kv.clone(),
        compute_storage.clone(),
        data_dir,
        config.handlers.as_ref(),
        messaging,
        max_handler_blob_bytes,
        max_component_bytes,
        allow_guest_private_egress,
        self_egress_addrs,
        guest_egress_extra_roots,
        allow_env_secret_refs,
        allow_guest_email,
        options.posture.require_tenancy_declaration,
        options.posture.allow_cross_tenant_db,
        &deploy,
        secrets_envelope.clone(),
    )
    .await?;
    // Hand the runtime the SAME per-tenant secret store `Arc` the control-plane routes hold (task
    // #493), so a guest `tenant-secrets` `get` and a control-plane `PUT` seal/unseal against ONE
    // store. Unset when no `[secrets]` envelope, so the guest binding is not built (fail-closed).
    // Handlers-gated: `set_tenant_secret_store` lives on the handler runtime, so a lean (no-handlers)
    // node has no guest binding to wire (the control-plane routes still work via `ServerOptions`).
    #[cfg(feature = "handlers")]
    if let Some(store) = tenant_secret_store.clone() {
        handlers.set_tenant_secret_store(store);
    }
    // Wire the fleet session-cookie signer (R3, PLAN-tenancy-principal): the same issuer that mints
    // control-plane tokens signs + verifies the host-issued anonymous session cookie AND the
    // delegable capabilities (PLAN-delegable-capabilities). Handlers-gated: the session-cookie
    // machinery lives on the handler runtime, so a lean (no-handlers) build has nothing to wire.
    //
    // The fleet signer is a DIFFERENT trust domain from control-plane admin auth (signing a
    // customer's session cookie / an embed capability is not the authority to admit an operator to
    // the control plane), but production derives it from the control-plane issuer for convenience.
    // For a DEV / loopback node with control-plane auth disabled (`options.issuer` is `None`),
    // auto-provision an EPHEMERAL in-memory Ed25519 fleet key so the guest-facing signer just works
    // — session cookies + capability mint/verify — WITHOUT turning on control-plane auth. Strictly
    // gated to a loopback bind (or an in-process embedder with no bind address): never on a public
    // bind, where an ephemeral key would silently invalidate live capabilities across a restart (a
    // public node that wants guest capabilities without control-plane auth must supply a persistent
    // key). Ephemeral = issue + verify within one process run; nothing persisted, no cross-process
    // or cross-deploy trust. Production is byte-identical: a real deploy supplies a control-plane key
    // ⇒ `issuer` is `Some` ⇒ this fallback is never taken.
    #[cfg(feature = "handlers")]
    {
        let fleet_signer = options.issuer.clone().or_else(|| {
            should_autoprovision_fleet_signer(serve_addr).then(|| {
                tracing::warn!(
                    "control-plane auth is disabled and no signer is configured; auto-provisioning \
                     an EPHEMERAL in-memory fleet signer (Ed25519) for host session cookies + \
                     delegable capabilities on this loopback/dev node — regenerated each start, \
                     never persisted. Configure a control-plane key (or a dedicated signer) for \
                     production."
                );
                Arc::new(boatramp_core::cose::LocalSigner::generate(
                    boatramp_core::cose::TokenAlg::Ed25519,
                )) as Arc<dyn boatramp_core::cose::Signer>
            })
        });
        if let Some(issuer) = fleet_signer {
            handlers.set_session_signer(issuer);
        }
    }
    // Enable guest capability minting (`boatramp:handlers/capability`, PLAN-delegable-capabilities)
    // when the operator posture allows it. A minted capability is verified against the same fleet
    // signer as the session cookie (wired just above), so this only enables the mint path + the TTL
    // ceiling; posture-off (or a zero ceiling) ⇒ not offered (a guest `mint` is access-denied).
    #[cfg(feature = "capability")]
    if options.posture.allow_guest_mint_capability {
        handlers.set_capability_minting(options.posture.max_guest_capability_ttl_secs);
    }
    // Per-project tenancy/capability posture overrides (Gap 4a): resolve each
    // `[security.projects.<p>]` override against the fleet base so one serve process can run a
    // strict-isolation project beside a looser one on a shared, multi-project machine. Empty ⇒
    // every project uses the node base wired just above. Only these four in-project knobs are
    // per-project; cross-project isolation stays structural (project = database).
    #[cfg(feature = "handlers")]
    {
        let base = &options.posture;
        let overrides: std::collections::BTreeMap<
            String,
            boatramp_core::security::ResolvedProjectTenancy,
        > = config
            .security
            .as_ref()
            .map(|s| {
                s.projects
                    .iter()
                    .map(|(project, ovr)| (project.clone(), base.project_tenancy(ovr)))
                    .collect()
            })
            .unwrap_or_default();
        handlers.set_project_tenancy_overrides(overrides);
    }
    // Wire the guest project self-config capability (`boatramp:handlers/admin`) when the
    // operator posture enables at least one surface. The controller reuses the same in-process
    // domain-verify / email-profile / secret / site-config subsystems + the real domain probe;
    // it's project-scoped per grant and rate-limited + audited. Posture-off ⇒ not offered.
    #[cfg(feature = "admin")]
    {
        use boatramp_handlers::AdminSurface;
        let p = &options.posture;
        let mut surfaces = std::collections::BTreeSet::new();
        if p.allow_guest_admin_domains {
            surfaces.insert(AdminSurface::Domains);
        }
        if p.allow_guest_admin_email {
            surfaces.insert(AdminSurface::Email);
        }
        if p.allow_guest_admin_site {
            surfaces.insert(AdminSurface::Site);
        }
        if p.allow_guest_admin_secrets {
            surfaces.insert(AdminSurface::Secrets);
        }
        if !surfaces.is_empty() {
            let controller = Arc::new(boatramp_server::ServerAdminController::with_server_probe(
                deploy.clone(),
                email_profile_store.clone(),
                secret_store.clone(),
                p.domain_verify_allow_private,
            ));
            handlers.set_admin(controller, surfaces);
        }
    }
    // Leader-gate cron firing (cluster: only the Raft leader fires; single-node: an
    // always-true gate, equivalent to the unset default). The same gate drives the
    // reconcile loops below, so all three converge on one leader per fleet. Only the
    // handler runtime has a scheduler, so this is a no-op without the `handlers` feature.
    #[cfg(feature = "handlers")]
    handlers.set_cron_leader_gate(is_leader.clone());
    // FA-5b2: on a cloud backend, wire the blob-change notification provisioner +
    // its tier so adding a `blob` trigger provisions (and removing it retracts).
    #[cfg(feature = "handlers")]
    if let Some(provider) = watch_provider {
        handlers.set_watch_provider(provider);
        handlers.set_provision_tier(provision_tier);
    }
    #[cfg(not(feature = "handlers"))]
    let _ = (watch_provider, provision_tier);

    // Materialize the reserved `default` project so `project ls` / `project show
    // default` reflect it on a fresh install, not only after a migration. Best
    // effort: the reader backstop keeps listings correct even if this write can't
    // land, so a transient failure must never block serving.
    match deploy.ensure_default_project().await {
        Ok(true) => tracing::info!("materialized the reserved `default` project record"),
        Ok(false) => {}
        Err(e) => tracing::warn!(
            error = %e,
            "could not materialize the `default` project record; readers use the synthesized default"
        ),
    }
    // Wire the function-to-function invoke resolver now the deploy store exists,
    // so a function granted `invoke` can call a sibling in-process (FI).
    #[cfg(feature = "handlers")]
    handlers.set_invoker(deploy.clone());

    // Compute reconcile loop. Single-node is always the "leader". Backends are
    // built from the `[compute]` config + capability detection; a no-op when none
    // are registered. Detached for the server's life.
    let (compute_backends, compute_node) = crate::compute::build_compute(
        config.compute.as_ref(),
        compute_storage,
        data_dir,
        node_id,
        !allow_shared_kernel,
        options.daemon_runtime.clone(),
        worker_exe.as_deref(),
    )
    .await;
    // Adopt the IPs of already-running replicas into each backend's fresh-on-boot
    // IP pool BEFORE the reconcile loop starts allocating. A backend with a per-node
    // pool (the native container backend) rebuilds it empty each process start; without
    // this the boot reconcile could re-hand a live address to a different workload —
    // the container-IP collision — or move a replica's endpoint on relaunch. Feeds
    // every persisted replica's `(workload, replica, endpoint-ip)`; each backend keeps
    // only the IPs in its own subnet (a cheap no-op for docker/cloudflare/VMM).
    crate::compute::adopt_running_replica_ips(&deploy, &compute_backends).await;
    // Per-project internal DNS (service discovery): start the resolver on the bridge
    // gateway so a guest resolves peers by name within its project. On by default;
    // starts only when the container backend + bridge are up (Linux). Detached for
    // the node's serving life (pushed into `reconcile` below). Started before the
    // reconcile loop consumes `compute_backends` — it borrows the registry to check
    // the container backend is present.
    let internal_dns =
        crate::compute::spawn_internal_dns(config.compute.as_ref(), &compute_backends, &deploy);
    // Activate the compute sql-shim (PLAN-compute-bindings): bind its listener +
    // build the resolver when a sql provider and `compute.sql_shim_url` are both present.
    #[cfg(feature = "handlers")]
    let sql_resolver = boatramp_server::sql_shim::spawn_sql_shim(
        handlers.sql_backends(),
        config.compute.as_ref().and_then(|c| c.sql_shim_url.clone()),
    )
    .await;
    #[cfg(not(feature = "handlers"))]
    let sql_resolver: Option<Arc<dyn boatramp_core::compute::ComputeBindingResolver>> = None;

    // Managed compute-backed SQL (PLAN-managed-compute-sql P2-b): if the handler
    // `sql` config declares any managed database, inject its `POSTGRES_*`/`MYSQL_*`
    // server-init env into the DB workload at launch from the sealed credential.
    // Reaching here with a managed DB implies an envelope (build_handler_runtime
    // fails closed otherwise), so the credential store always has one to seal with.
    // Keep a clone of the secrets envelope for the operator-SQL capability below
    // (the managed_db_resolver match moves the original).
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
    let operator_envelope = secrets_envelope.clone();
    // …and a second clone for the tenant-deprovision capability (drops a deleted
    // tenant's managed DB/role/credential on project/site delete). It needs a real
    // envelope to seal/unseal + delete per-tenant credentials, so it is wired only
    // when one is present (same fail-closed gating as the managed-DB paths).
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
    let deprovision_envelope = secrets_envelope.clone();
    // …and a clone for the provisioning drift-repair capability (owner-model retrofit /
    // reconcile). Like the migrate path it connects as the sealed owner role and re-seals
    // credentials, so it needs a real envelope; wired only when one is present.
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
    let repair_envelope = secrets_envelope.clone();
    // …and a third clone for the soft-delete tombstone reaper (the leader-gated task
    // that hard-drops a Shared-Postgres tenant once its grace window elapses). It, too,
    // needs a real envelope to unseal the superuser credential + delete the per-tenant
    // one on hard-drop.
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
    let reaper_envelope = secrets_envelope.clone();
    // …and a clone for the declarative managed-database capability (#501 Stage B): it
    // provisions the declared DB + seals its credential server-side, so it needs a real
    // envelope; wired only when one is present (fail-closed like every managed-DB path).
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
    let declare_envelope = secrets_envelope.clone();
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
    let managed_db_resolver: Option<Arc<dyn boatramp_core::compute::ManagedDbEnvResolver>> = match (
        config
            .handlers
            .as_ref()
            .and_then(|h| h.bindings.sql.as_ref()),
        secrets_envelope,
    ) {
        (Some(sql), Some(envelope)) if !sql.databases.is_empty() => {
            let creds = crate::managed_sql::ManagedSqlCredentials::new(kv.clone(), envelope);
            let privilege = config
                .compute
                .as_ref()
                .map(|c| c.managed_db_privilege)
                .unwrap_or_default();
            let env =
                crate::managed_sql::ManagedDbEnv::from_config(&sql.databases, creds, privilege);
            (!env.is_empty()).then(|| Arc::new(env) as Arc<_>)
        }
        _ => None,
    };
    #[cfg(not(any(feature = "sql-postgres", feature = "sql-mysql")))]
    let managed_db_resolver: Option<Arc<dyn boatramp_core::compute::ManagedDbEnvResolver>> = None;

    // Turnkey managed DB: auto-register the compute workload(s) backing each managed
    // co-located database that has none yet, so declaring the `databases` binding is
    // enough to boot the DB (no separate `compute set` / apply). Tenant-aware — a
    // `Shared` binding registers its one shared server; a `Single` binding registers
    // nothing at boot (its per-tenant `<compute>-<ident>` is created durably by the lazy
    // resolve on first `sql` use and relaunched by the reconcile, so a project that never
    // uses `sql` — e.g. a static-only site — never gets a spurious DB). Non-clobbering +
    // idempotent; runs before the reconcile loop so its first tick can launch what it
    // registered.
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
    if let Some(sql) = config
        .handlers
        .as_ref()
        .and_then(|h| h.bindings.sql.as_ref())
        .filter(|sql| !sql.databases.is_empty())
    {
        crate::managed_sql::auto_register_managed_db_workloads(&deploy, &sql.databases).await;
    }

    // Operator SQL capability (managed-DB migrations/queries via the sealed credential, resolved
    // server-side) — backs `POST /api/sql/{db}/{exec,query}`. The SAME concrete NodeOperatorSql
    // also backs the owner-gated schema-migration runner (which reuses its owner + superuser
    // backends), so build it ONCE and share it.
    // Build the SAME concrete NodeOperatorSql once (when a managed DB is configured) and share it:
    // it backs both `operator_sql` (the sql exec/query cap) and the migration runner (which reuses
    // its owner + superuser backends). Two separate bindings so neither annotation is a complex type.
    // The migration substrate is a single dispatcher (crate::managed_sql::DispatchMigrationRunner)
    // routing each `(project, db)` to its engine's substrate: the sqlx NodeMigrationRunner for a
    // Postgres/MySQL binding, the LibsqlMigrationRunner for a `libsql` binding. It compiles whenever a
    // sqlx engine OR `migrate` (⇒ libsql) is on, so the embedded-libsql default can migrate even on a
    // node with no external sqlx engine. `operator_sql` (the `POST /api/sql/{db}/{exec,query}` cap)
    // stays sqlx-only — a libsql file has no operator-SQL/credential seam.
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
    let operator_sql: Option<Arc<dyn boatramp_core::sql::OperatorSql>>;
    // Late-init: the match arms assign it (and, under sqlx, `operator_sql` in the same block), and the
    // arms differ by feature-cfg — so a direct `let … = match {…}` would need cfg'd arm bodies. The
    // late-init keeps that readable; the value is always assigned before use.
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql", feature = "migrate"))]
    #[allow(clippy::needless_late_init)]
    let migration_substrate: Option<Arc<dyn boatramp_core::sql::MigrationSubstrate>>;
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql", feature = "migrate"))]
    match config
        .handlers
        .as_ref()
        .and_then(|h| h.bindings.sql.as_ref())
        .filter(|sql| !sql.databases.is_empty())
    {
        Some(sql) => {
            // The sqlx (Postgres/MySQL) arm — the shared NodeOperatorSql backs both the operator-SQL
            // cap and the sqlx migration runner. Only built when a sqlx engine is compiled in.
            #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
            let node_op = Arc::new(crate::managed_sql::NodeOperatorSql::new(
                sql.databases.clone(),
                kv.clone(),
                operator_envelope,
                deploy.clone(),
            ));
            // The operator's trusted-extension allowlist — the only extensions a migration may
            // enable (empty ⇒ none). See ExternalSqlConfig::migrate_trusted_extensions.
            #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
            let trusted: std::collections::BTreeSet<String> = sql
                .migrate_trusted_extensions
                .clone()
                .unwrap_or_default()
                .into_iter()
                .collect();
            migration_substrate = Some(Arc::new(crate::managed_sql::DispatchMigrationRunner::new(
                #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
                node_op.clone(),
                #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
                trusted,
                #[cfg(feature = "migrate")]
                sql.databases.clone(),
            ))
                as Arc<dyn boatramp_core::sql::MigrationSubstrate>);
            #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
            {
                operator_sql = Some(node_op as Arc<dyn boatramp_core::sql::OperatorSql>);
            }
        }
        None => {
            #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
            {
                operator_sql = None;
            }
            migration_substrate = None;
        }
    }
    // When only `migrate` (no sqlx) is compiled, the operator-SQL cap does not exist.
    #[cfg(all(
        not(any(feature = "sql-postgres", feature = "sql-mysql")),
        feature = "migrate"
    ))]
    let operator_sql: Option<Arc<dyn boatramp_core::sql::OperatorSql>> = None;
    #[cfg(not(any(feature = "sql-postgres", feature = "sql-mysql", feature = "migrate")))]
    let migration_substrate: Option<Arc<dyn boatramp_core::sql::MigrationSubstrate>> = None;
    #[cfg(not(any(feature = "sql-postgres", feature = "sql-mysql", feature = "migrate")))]
    let operator_sql: Option<Arc<dyn boatramp_core::sql::OperatorSql>> = None;

    // Tenant-deprovision capability (drop a deleted tenant's managed DB/role/sealed
    // credential on project/site delete). Wired only when a compute-backed managed
    // database + a secrets envelope are both present — same gating as operator_sql,
    // plus the envelope requirement (it must seal/unseal per-tenant credentials).
    // The soft-delete grace window for a Shared-Postgres managed tenant
    // (`handlers.bindings.sql.deprovision_grace_secs`, env-settable). Default 7 days;
    // `0` disables the soft path (immediate hard drop). Threaded to the deprovisioner
    // (which soft-deletes) and implicitly honored by the reaper (which only ever finds
    // tombstones a >0 grace produced).
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
    let deprovision_grace_secs = config
        .handlers
        .as_ref()
        .and_then(|h| h.bindings.sql.as_ref())
        .and_then(|sql| sql.deprovision_grace_secs)
        .unwrap_or(crate::tenant_sql::DEFAULT_DEPROVISION_GRACE_SECS);
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
    let tenant_deprovisioner: Option<Arc<dyn boatramp_core::sql::TenantDeprovisioner>> = config
        .handlers
        .as_ref()
        .and_then(|h| h.bindings.sql.as_ref())
        .filter(|sql| !sql.databases.is_empty())
        .zip(deprovision_envelope)
        .map(|(sql, envelope)| {
            Arc::new(crate::tenant_sql::NodeTenantDeprovisioner::new(
                deploy.clone(),
                kv.clone(),
                envelope,
                sql.databases.clone(),
                deprovision_grace_secs,
            )) as Arc<_>
        });
    #[cfg(not(any(feature = "sql-postgres", feature = "sql-mysql")))]
    let tenant_deprovisioner: Option<Arc<dyn boatramp_core::sql::TenantDeprovisioner>> = None;

    // Provisioning drift-repair capability (owner-model retrofit / reconcile) — backs the
    // `Project·Admin`-gated `/api/repair/{db}` + `/dry-run`. Same gating as operator_sql plus
    // the envelope requirement (it re-seals the owner credential + connects as it). The
    // envelope is threaded as `Some(_)` so a lean-but-managed node still gets a clear
    // per-check error rather than a panic if none is configured.
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
    let tenant_repair: Option<Arc<dyn boatramp_core::sql::TenantRepair>> = config
        .handlers
        .as_ref()
        .and_then(|h| h.bindings.sql.as_ref())
        .filter(|sql| !sql.databases.is_empty())
        .map(|sql| {
            Arc::new(crate::repair::NodeTenantRepair::new(
                sql.databases.clone(),
                deploy.clone(),
                kv.clone(),
                repair_envelope.clone(),
            )) as Arc<_>
        });
    #[cfg(not(any(feature = "sql-postgres", feature = "sql-mysql")))]
    let tenant_repair: Option<Arc<dyn boatramp_core::sql::TenantRepair>> = None;

    // Declarative managed-database capability (#501 Stage B) — backs the `Project·Admin`-gated
    // `PUT /api/projects/{proj}/databases/{name}` + `POST …/ensure`. It persists a manifest
    // `databases:` entry to `project/{project}/database/{name}`, enforces daemon-config-wins at
    // the merge point (against the node-static `sql.databases`), refuses an identity change,
    // binds provisioning to the caller's project, and eagerly provisions via `provision_tenant`.
    // Requires a sqlx engine (it provisions a Postgres/MySQL server) AND a `[secrets]` envelope
    // (a managed DB seals its credential). UNLIKE repair/operator_sql it is NOT gated on
    // `!sql.databases.is_empty()` — a node with NO node-static databases can still accept a
    // project declaration (the whole point of the declarative front door). The node-static map
    // (possibly empty) is threaded so the daemon-wins conflict check has both sources.
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
    let managed_db_declare: Option<Arc<dyn boatramp_core::sql::ManagedDbDeclare>> =
        declare_envelope.map(|envelope| {
            let sql_cfg = config
                .handlers
                .as_ref()
                .and_then(|h| h.bindings.sql.as_ref());
            let static_dbs = sql_cfg.map(|sql| sql.databases.clone()).unwrap_or_default();
            // Per-project declaration ceilings (#501 Stage B MEDIUM-1 — the disk-exhaustion
            // guard): the operator's `max_declared_databases`/`max_declared_volume_mib`, or
            // the built-in `apply_db_caps` defaults (16 DBs / 512 GiB) when unset.
            let quota = crate::managed_db_declare::DeclareQuota::from_config(
                sql_cfg.and_then(|sql| sql.max_declared_databases),
                sql_cfg.and_then(|sql| sql.max_declared_volume_mib),
            );
            Arc::new(crate::managed_db_declare::NodeManagedDbDeclare::new(
                static_dbs,
                deploy.clone(),
                kv.clone(),
                envelope,
                quota,
            )) as Arc<_>
        });
    #[cfg(not(any(feature = "sql-postgres", feature = "sql-mysql")))]
    let managed_db_declare: Option<Arc<dyn boatramp_core::sql::ManagedDbDeclare>> = None;

    // Operator compute-exec capability (run a command inside a running workload) —
    // backs `POST /api/compute/{name}/exec`, gated by the `allow_compute_exec`
    // posture. Clone the backend registry before the reconcile loop consumes it.
    let compute_exec: Option<Arc<dyn boatramp_core::compute::ComputeExec>> = Some(Arc::new(
        crate::compute::NodeComputeExec::new(compute_backends.clone(), deploy.clone()),
    ) as Arc<_>);

    // Operator volume-reclamation capability (list + remove persistent volumes) —
    // backs `GET /api/compute/volumes` + `DELETE /api/compute/volumes/{name}`.
    // Same admin-scoped `/api/compute/*` gate; clone the registry before the
    // reconcile loop consumes the original below.
    let compute_volumes: Option<Arc<dyn boatramp_core::compute::ComputeVolumes>> = Some(Arc::new(
        crate::compute::NodeComputeVolumes::new(compute_backends.clone(), deploy.clone()),
    )
        as Arc<_>);

    // Operator reconcile-plane control capability (restart a replica) — backs
    // `POST /api/compute/maintenance/restart` (admin-scoped). Clone the registry
    // before the reconcile loop consumes the original below.
    let compute_control: Option<Arc<dyn boatramp_core::compute::ComputeControl>> = Some(Arc::new(
        crate::compute::NodeComputeControl::new(compute_backends.clone(), deploy.clone()),
    )
        as Arc<_>);

    let compute_reconcile = boatramp_server::spawn_compute_reconcile(
        deploy.clone(),
        compute_backends,
        vec![compute_node],
        boatramp_core::compute::BackendPolicy::from_shared_kernel_allowed(allow_shared_kernel),
        is_leader.clone(),
        compute_reconcile_tick(),
        COMPUTE_IDLE_TIMEOUT,
        sql_resolver,
        managed_db_resolver,
    );

    // Tenant tombstone reaper: leader-gated hard-drop of soft-deleted Shared-Postgres
    // tenants past their grace window (safe deprovision — see `tenant_sql`). Wired only
    // when a compute-backed managed database + a secrets envelope are both present
    // (same gating as the deprovisioner); each tombstone carries its own server +
    // superuser, so the reaper needs no per-binding config. A `0` grace never writes a
    // tombstone, so the sweep is simply inert then.
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
    let tombstone_reaper: Option<tokio::task::JoinHandle<()>> = config
        .handlers
        .as_ref()
        .and_then(|h| h.bindings.sql.as_ref())
        .filter(|sql| !sql.databases.is_empty())
        .zip(reaper_envelope)
        .map(|(_sql, envelope)| {
            crate::tenant_sql::spawn_tenant_tombstone_reaper(
                deploy.clone(),
                kv.clone(),
                envelope,
                is_leader.clone(),
                crate::tenant_sql::TOMBSTONE_REAPER_TICK,
            )
        });
    #[cfg(not(any(feature = "sql-postgres", feature = "sql-mysql")))]
    let tombstone_reaper: Option<tokio::task::JoinHandle<()>> = None;

    // Domain-verify auto-complete: periodically re-check every site's pending
    // ownership challenges and attach any that now pass — a published token (e.g.
    // via `domain add --provider`) converges without a manual `domain verify`.
    let dv_reconcile = boatramp_server::spawn_domain_verify_reconcile(
        deploy.clone(),
        domain_verify_allow_private,
        is_leader,
        DOMAIN_VERIFY_RECONCILE_TICK,
    );

    // Wire the operator capabilities onto the options the router is built from.
    let mut options = options;
    options.operator_sql = operator_sql;
    options.migration_substrate = migration_substrate;
    options.tenant_repair = tenant_repair;
    options.managed_db_declare = managed_db_declare;
    options.tenant_deprovisioner = tenant_deprovisioner;
    options.compute_exec = compute_exec;
    options.compute_volumes = compute_volumes;
    options.compute_control = compute_control;
    // The internal secret store backs the admin secrets API (set/list/delete). Not
    // handlers-gated — it must be reachable even on a lean node.
    options.secret_store = secret_store;
    // The per-tenant sealed secret store backs the control-plane CRUD (task #493). Like the secret
    // store it is not handlers-gated, so the endpoints work on a lean node.
    options.tenant_secret_store = tenant_secret_store;
    // The email-profile store backs the admin API (`/api/email/profiles`); like the
    // secret store it is not handlers-gated, so it works on a lean node.
    options.email_profile_store = email_profile_store;

    // The detached reconcile loops: the always-present compute + domain-verify ones,
    // plus the optional tenant-tombstone reaper (only when a managed DB is configured).
    let mut reconcile = vec![compute_reconcile, dv_reconcile];
    if let Some(reaper) = tombstone_reaper {
        reconcile.push(reaper);
    }
    if let Some(dns) = internal_dns {
        reconcile.push(dns);
    }

    Ok(RunningNode {
        deploy,
        handlers,
        auth,
        options,
        reconcile,
    })
}

/// Build the `[secrets]` envelope (secrets-at-rest wrapping) from `boatramp.cfg`'s
/// `[secrets]` section: `local` (a machine-local AES-256-GCM KEK) or `vault` (Vault
/// Env var an operator points at a PEM file of extra CA(s) the guest's outbound `wasi:http` TLS
/// client should trust on top of the webpki roots — honored only under the
/// `allow_guest_egress_extra_ca` posture (a hermetic HTTPS test double lever).
const GUEST_EGRESS_EXTRA_CA_ENV: &str = "BOATRAMP_GUEST_EGRESS_EXTRA_CA_FILE";

/// Parse the operator's guest-egress extra-CA PEM ([`GUEST_EGRESS_EXTRA_CA_ENV`]) into rustls trust
/// anchors, gated by the `allow_guest_egress_extra_ca` posture (`allow`). No env set ⇒ empty (the
/// default). `allow == false` (e.g. multi-tenant) with a file set ⇒ empty + a warning (the posture
/// refuses it). Set + readable + ≥1 cert ⇒ those certs. Set-but-unreadable / no valid cert ⇒ a hard
/// error (fail closed — a configured-but-broken CA must not silently degrade to no-trust).
fn load_guest_egress_extra_roots(
    allow: bool,
) -> Result<Vec<rustls::pki_types::CertificateDer<'static>>> {
    let path = match std::env::var(GUEST_EGRESS_EXTRA_CA_ENV) {
        Ok(p) if !p.is_empty() => p,
        _ => return Ok(Vec::new()),
    };
    if !allow {
        tracing::warn!(
            env = GUEST_EGRESS_EXTRA_CA_ENV,
            "ignoring a guest-egress extra CA: the security posture forbids it \
             (allow_guest_egress_extra_ca is off — e.g. under multi-tenant)"
        );
        return Ok(Vec::new());
    }
    let pem =
        std::fs::read(&path).map_err(|e| Error::GuestEgressCa(format!("reading {path:?}: {e}")))?;
    let certs = rustls_pemfile::certs(&mut std::io::BufReader::new(&pem[..]))
        .collect::<std::result::Result<Vec<_>, _>>()
        .map_err(|e| Error::GuestEgressCa(format!("parsing {path:?}: {e}")))?;
    if certs.is_empty() {
        return Err(Error::GuestEgressCa(format!(
            "{path:?} contained no PEM certificate"
        )));
    }
    tracing::info!(
        env = GUEST_EGRESS_EXTRA_CA_ENV,
        count = certs.len(),
        path = %path,
        "guest egress trusts operator extra CA(s) (dev-posture; webpki roots still apply)"
    );
    Ok(certs)
}

/// Transit). `None`/empty ⇒ no wrapping. The Vault token is read from the
/// environment (`token_env`), never a file. This seals a managed SQL credential at
/// rest; a managed database fails closed without it.
fn build_secrets_envelope(
    secrets: Option<&crate::config::SecretsConfig>,
    data_dir: &Path,
) -> Result<Option<Arc<dyn boatramp_core::envelope::KeyEnvelope>>> {
    use boatramp_server::envelope::{EnvelopeSpec, build_envelope};
    let Some(cfg) = secrets else {
        return Ok(None);
    };
    let spec = match cfg.envelope.as_str() {
        "" => EnvelopeSpec::None,
        "local" => EnvelopeSpec::Local {
            kek_file: cfg
                .kek_file
                .clone()
                .unwrap_or_else(|| data_dir.join("secrets/kek")),
        },
        "vault" => {
            let v = cfg.vault.as_ref().ok_or_else(|| {
                Error::Envelope(
                    "secrets.envelope = \"vault\" needs a [secrets.vault] section".into(),
                )
            })?;
            let token = std::env::var(&v.token_env).map_err(|_| {
                Error::Envelope(format!("Vault token env `{}` is not set", v.token_env))
            })?;
            EnvelopeSpec::Vault {
                addr: v.addr.clone(),
                key: v.key.clone(),
                token,
            }
        }
        other => {
            return Err(Error::Envelope(format!(
                "unknown secrets.envelope {other:?} (want \"local\" or \"vault\")"
            )));
        }
    };
    build_envelope(spec).map_err(|e| Error::Envelope(e.to_string()))
}

#[cfg(all(test, feature = "fs"))]
mod tests {
    use super::*;
    use boatramp_core::kv::MemoryKv;
    use boatramp_core::security::SecurityProfile;

    /// The dev/loopback ephemeral fleet-signer auto-provision is gated STRICTLY to a loopback bind
    /// (or an in-process embedder with no bind): a public / wildcard / private-network bind must
    /// NOT silently provision an ephemeral signing key (it would invalidate live capabilities on a
    /// restart — such a node must supply a persistent key). This is the security-critical boundary.
    #[cfg(feature = "handlers")]
    #[test]
    fn ephemeral_fleet_signer_auto_provisions_only_on_loopback_or_in_process() {
        use std::net::SocketAddr;
        let sa = |s: &str| s.parse::<SocketAddr>().unwrap();
        // In-process (no listener) and loopback → auto-provision the dev fleet signer.
        assert!(should_autoprovision_fleet_signer(None));
        assert!(should_autoprovision_fleet_signer(Some(sa(
            "127.0.0.1:8080"
        ))));
        assert!(should_autoprovision_fleet_signer(Some(sa("[::1]:8080"))));
        // Off-host-reachable binds MUST NOT auto-provision an ephemeral key: a public IP, a
        // private-network IP, and a wildcard bind (`0.0.0.0`/`::`, reachable off-host).
        assert!(!should_autoprovision_fleet_signer(Some(sa(
            "203.0.113.5:8080"
        ))));
        assert!(!should_autoprovision_fleet_signer(Some(sa(
            "10.0.0.4:8080"
        ))));
        assert!(!should_autoprovision_fleet_signer(Some(sa("0.0.0.0:8080"))));
        assert!(!should_autoprovision_fleet_signer(Some(sa("[::]:8080"))));
    }

    /// The headline in-process fidelity check (PLAN-node-library N2b.3): `assemble`
    /// over a temp `FsStorage` + `MemoryKv` produces a `RunningNode` whose deploy
    /// store is live (the reserved `default` project was materialized during
    /// assembly) and whose router — the exact one `boatramp serve` builds — answers
    /// `/healthz`. No listener is bound: the request is driven through the router
    /// via `tower::oneshot`, so the whole assembly runs in-process.
    #[tokio::test]
    async fn assemble_produces_a_serving_node_over_a_temp_store() {
        use axum::body::Body;
        use axum::http::{Request, StatusCode};
        use tower::ServiceExt;

        let tmp = tempfile::tempdir().unwrap();
        let storage: Arc<dyn Storage> = Arc::new(boatramp_storage::FsStorage::new(tmp.path()));
        let kv: Arc<dyn KvStore> = Arc::new(MemoryKv::new());
        let config = ServerConfig::default();
        let options = boatramp_server::ServerOptions {
            // The strict `multi-tenant` posture, as an unconfigured `serve` resolves.
            posture: SecurityProfile::MultiTenant.preset(),
            ..Default::default()
        };

        let node = assemble(NodeInput {
            config: &config,
            data_dir: tmp.path(),
            storage,
            kv,
            auth: boatramp_server::Auth::disabled(),
            options,
            serve_addr: None,
            watch_provider: None,
            provision_tier: boatramp_core::blob_notify::ProvisionTier::default(),
            messaging: None,
            is_leader: Arc::new(|| true),
            node_id: 0,
            worker_exe: None,
        })
        .await
        .expect("assemble a node over a temp store");

        // The deploy store is live: `assemble` already materialized the reserved
        // `default` project, so a second ensure reports "already present" (`false`).
        assert!(
            !node
                .deploy
                .ensure_default_project()
                .await
                .expect("read the default project"),
            "assemble should have materialized the default project"
        );

        // The assembled router (the same wiring `serve` binds) answers /healthz.
        let router =
            boatramp_server::router_with(node.deploy, node.auth, node.handlers, node.options);
        let response = router
            .oneshot(
                Request::builder()
                    .uri("/healthz")
                    .body(Body::empty())
                    .unwrap(),
            )
            .await
            .expect("route /healthz");
        assert_eq!(response.status(), StatusCode::OK);
    }
}