boatramp-node 0.4.27

//! Managed credentials for a boatramp-run SQL database (PLAN-managed-compute-sql,
//! Phase 2). boatramp generates a strong password on first use, seals it with the
//! secrets [`KeyEnvelope`], and persists it in the control-plane KV — **stable
//! across restarts** (the DB server was initialized with it) and **never stored in
//! cleartext**. The same password configures the DB workload's server env at launch
//! and connects the handler `sql` binding, so an operator sets no DB secret at all.

use std::collections::HashMap;
use std::sync::Arc;

use async_trait::async_trait;
use boatramp_core::compute::{ManagedDbEnvResolver, PrivilegeDirective, ReplicaPhase};

use crate::config::ManagedDbPrivilege;
use boatramp_core::deploy::DeployStore;
use boatramp_core::envelope::KeyEnvelope;
use boatramp_core::kv::KvStore;
use boatramp_core::project::ProjectRef;
use boatramp_core::sql::SqlError;
#[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
use boatramp_core::sql::{
    AppliedMigration, LedgerOrigin, MigrateDdl, MigrateDdlError, MigrationError, MigrationStep,
    MigrationSubstrate, SubstrateStepOutcome,
};
use boatramp_storage::sql_compute::{ComputeEndpointResolver, ReplicaDiag};
use boatramp_storage::ExternalSqlKind;

/// The env vars a managed DB server image reads to **initialize on first boot** with
/// boatramp's managed credential — so the handler can then connect as `user`/`password`
/// to `database`. (Postgres: `POSTGRES_*`; MySQL: `MYSQL_*`, incl. a root password —
/// unused by handlers but required by the image to init.) Injected into the DB
/// workload's env at launch (P2-b); the values come from [`ManagedSqlCredentials`].
#[cfg_attr(not(feature = "handlers"), allow(dead_code))]
pub fn managed_db_server_env(
    kind: ExternalSqlKind,
    database: &str,
    user: &str,
    password: &str,
) -> Vec<(String, String)> {
    match kind {
        ExternalSqlKind::Postgres => vec![
            ("POSTGRES_USER".into(), user.into()),
            ("POSTGRES_PASSWORD".into(), password.into()),
            ("POSTGRES_DB".into(), database.into()),
        ],
        ExternalSqlKind::Mysql => vec![
            ("MYSQL_USER".into(), user.into()),
            ("MYSQL_PASSWORD".into(), password.into()),
            ("MYSQL_DATABASE".into(), database.into()),
            // The image requires a root password to initialize; reuse the managed
            // secret (root is not exposed to handlers, which connect as `user`).
            ("MYSQL_ROOT_PASSWORD".into(), password.into()),
        ],
    }
}

/// Generates + seals + persists a stable password per managed-DB workload.
#[cfg_attr(not(feature = "handlers"), allow(dead_code))]
pub struct ManagedSqlCredentials {
    kv: Arc<dyn KvStore>,
    envelope: Arc<dyn KeyEnvelope>,
}

impl ManagedSqlCredentials {
    /// Build over the control-plane KV and the secrets envelope. A managed DB
    /// requires an envelope (`[secrets]`) so the password is never stored in clear.
    #[cfg_attr(not(feature = "handlers"), allow(dead_code))]
    pub fn new(kv: Arc<dyn KvStore>, envelope: Arc<dyn KeyEnvelope>) -> Self {
        Self { kv, envelope }
    }

    /// KV key holding a workload's sealed password.
    fn key(project: &str, workload: &str) -> String {
        format!("managed-sql-cred/{project}/{workload}")
    }

    /// The stable password for managed DB `workload` in `project`: unsealed from the
    /// store if present, else generated (32 random bytes → hex), sealed, and stored.
    /// Idempotent + stable across restarts, so the DB (initialized with it on first
    /// boot) keeps accepting the same credential.
    ///
    /// Single-node correct (get-then-put). A cluster where two nodes generate
    /// concurrently would race to a mismatch; that needs a put-if-absent (tracked in
    /// the plan) — not yet implemented here.
    #[cfg_attr(not(feature = "handlers"), allow(dead_code))]
    pub async fn password(&self, project: &str, workload: &str) -> Result<String, String> {
        let key = Self::key(project, workload);
        if let Some(sealed) = self.kv.get(&key).await.map_err(|e| e.to_string())? {
            let plain = self
                .envelope
                .unwrap(&sealed)
                .await
                .map_err(|e| e.to_string())?;
            return String::from_utf8(plain).map_err(|_| {
                format!("managed sql credential for {workload:?} is not valid UTF-8")
            });
        }
        let mut bytes = [0u8; 32];
        getrandom::getrandom(&mut bytes).map_err(|e| format!("rng: {e}"))?;
        let password: String = bytes.iter().map(|b| format!("{b:02x}")).collect();
        let sealed = self
            .envelope
            .wrap(password.as_bytes())
            .await
            .map_err(|e| e.to_string())?;
        self.kv.put(&key, sealed).await.map_err(|e| e.to_string())?;
        Ok(password)
    }

    /// Delete a workload's sealed credential (a tenant deprovision hook). Idempotent:
    /// deleting an absent credential is a no-op (the underlying KV `delete` treats a
    /// missing key as success), so re-running a teardown is harmless.
    #[cfg_attr(not(feature = "handlers"), allow(dead_code))]
    pub async fn delete(&self, project: &str, workload: &str) -> Result<(), String> {
        self.kv
            .delete(&Self::key(project, workload))
            .await
            .map_err(|e| e.to_string())
    }
}

/// One managed database's non-secret connection parts, keyed in [`ManagedDbEnv`]
/// by the compute **workload** that backs it.
#[cfg_attr(not(feature = "handlers"), allow(dead_code))]
struct ManagedDbSpec {
    kind: ExternalSqlKind,
    database: String,
    user: String,
    /// The binding's isolation mechanism. A `Single` binding may also back a
    /// **per-tenant** workload named `<base>-<tenant_ident>`, so its server-init env
    /// must be resolvable under that derived name too (with a per-tenant credential
    /// keyed by the derived workload). A `Shared` binding never spawns a separate
    /// per-tenant workload (its per-tenant databases live inside the base server), so
    /// only its exact base name resolves.
    tenant: crate::config::TenantIsolation,
}

/// The node's [`ManagedDbEnvResolver`]: the set of managed databases (from the
/// handler `sql` config) keyed by backing workload, plus the sealed-credential
/// store. At launch the reconcile asks this for a workload's server-init env; a
/// non-managed workload gets nothing. Both sides (this injector and the handler's
/// [`ComputeResolvedSqlBackend`]) read the **same** sealed credential, so the DB is
/// initialized with exactly the password the handler later connects with.
#[cfg_attr(not(feature = "handlers"), allow(dead_code))]
pub struct ManagedDbEnv {
    dbs: HashMap<String, ManagedDbSpec>,
    creds: ManagedSqlCredentials,
    /// How a managed DB's stock image runs on a shared-kernel backend so it can init
    /// (`[compute].managed_db_privilege`).
    privilege: ManagedDbPrivilege,
}

/// The uid:gid a stock DB image runs its server process as — both the official
/// `postgres` and `mysql` images use `999:999`. Used for the rootless strategy so the
/// entrypoint owns its pre-chowned volume without needing any capability.
#[cfg_attr(not(feature = "handlers"), allow(dead_code))]
fn managed_db_default_ids(_kind: ExternalSqlKind) -> (u32, u32) {
    (999, 999)
}

/// The minimal capabilities a stock DB entrypoint needs when it runs as root: `chown`
/// its data dir + socket dir, then `gosu`/`su-exec` drop to the DB user.
#[cfg_attr(not(feature = "handlers"), allow(dead_code))]
fn managed_db_caps() -> Vec<String> {
    ["CHOWN", "DAC_OVERRIDE", "FOWNER", "SETUID", "SETGID"]
        .iter()
        .map(|s| (*s).to_string())
        .collect()
}

impl ManagedDbEnv {
    /// Build from the handler `sql` `databases` config + the credential store,
    /// selecting only the **managed** ones (compute-backed, no `password_env`).
    /// A database with an unparsable engine or missing parts is skipped (config
    /// validation already rejects those before serve).
    #[cfg_attr(not(feature = "handlers"), allow(dead_code))]
    pub fn from_config(
        databases: &std::collections::BTreeMap<String, crate::config::ExternalDatabaseConfig>,
        creds: ManagedSqlCredentials,
        privilege: ManagedDbPrivilege,
    ) -> Self {
        let mut dbs = HashMap::new();
        for db in databases.values() {
            if !db.is_managed_credential() {
                continue;
            }
            let (Some(workload), Some(kind), Some(database), Some(user)) = (
                db.compute.clone(),
                ExternalSqlKind::parse(&db.kind),
                db.database.clone(),
                db.user.clone(),
            ) else {
                continue;
            };
            dbs.insert(
                workload,
                ManagedDbSpec {
                    kind,
                    database,
                    user,
                    tenant: db.tenant,
                },
            );
        }
        Self {
            dbs,
            creds,
            privilege,
        }
    }

    /// No managed databases configured — the caller can skip wiring this resolver.
    #[cfg_attr(not(feature = "handlers"), allow(dead_code))]
    pub fn is_empty(&self) -> bool {
        self.dbs.is_empty()
    }

    /// Resolve the launched `workload` to the [`ManagedDbSpec`] whose server-init env
    /// it needs. An exact match wins (the base workload — the shared server, or a
    /// single-tenant install). Otherwise a **`Single` per-tenant** workload
    /// `<base>-<tenant_ident>` maps back to its `Single` base spec, so a dedicated
    /// per-tenant container is initialized from the same binding — with its OWN
    /// per-tenant credential, keyed by its OWN `(project, workload)` (the reconcile
    /// passes the derived workload name, so `password(project, workload)` already
    /// keys per tenant). A `Shared` base never matches a `-suffixed` name (it spawns
    /// no per-tenant workload), so a stray derived name can never smuggle a Shared
    /// server's credential.
    fn resolve_spec(&self, workload: &str) -> Option<&ManagedDbSpec> {
        if let Some(spec) = self.dbs.get(workload) {
            return Some(spec);
        }
        // A `Single` per-tenant workload `<base>-<ident>`: match the `Single` base whose
        // name is a `-`-separated prefix of `workload`. `find_map` over a HashMap is
        // non-deterministic, and if one base is itself a `-`-prefix of another (e.g.
        // `pg` and `pg-metrics`, so `pg-metrics-<ident>` matches both) iteration order
        // would decide which spec's database/user fills the server-init env. Resolve to
        // the **longest** matching base instead — `pg-metrics` wins over `pg` — so the
        // choice is deterministic and unambiguous. (The credential key is exact, so this
        // is a robustness/correctness fix, not a cross-tenant reach.)
        self.dbs
            .iter()
            .filter(|(base, spec)| {
                matches!(spec.tenant, crate::config::TenantIsolation::Single)
                    && workload
                        .strip_prefix(base.as_str())
                        .is_some_and(|rest| rest.starts_with('-') && rest.len() > 1)
            })
            .max_by_key(|(base, _)| base.len())
            .map(|(_, spec)| spec)
    }
}

#[async_trait]
impl ManagedDbEnvResolver for ManagedDbEnv {
    async fn managed_db_env(&self, project: &str, workload: &str) -> Vec<(String, String)> {
        let Some(db) = self.resolve_spec(workload) else {
            return Vec::new();
        };
        // The credential key is the launched `(project, workload)` itself — the bare
        // base for the shared server / single-tenant install, or the derived
        // `<base>-<ident>` for a Single per-tenant container. So the init password and
        // the handler's connection password agree by construction (the resolver keys
        // it identically).
        match self.creds.password(project, workload).await {
            Ok(password) => managed_db_server_env(db.kind, &db.database, &db.user, &password),
            Err(e) => {
                // Fail closed on the env: without the sealed credential we must not
                // launch the DB with a blank/default password. An empty env means
                // the image refuses to initialize, which surfaces the misconfig.
                tracing::error!(
                    %workload,
                    error = %e,
                    "managed sql: could not resolve the sealed credential; DB launched without managed env"
                );
                Vec::new()
            }
        }
    }

    fn managed_db_privilege(&self, _project: &str, workload: &str) -> Option<PrivilegeDirective> {
        let db = self.resolve_spec(workload)?;
        Some(match self.privilege {
            ManagedDbPrivilege::Rootless => {
                let (uid, gid) = managed_db_default_ids(db.kind);
                PrivilegeDirective::Rootless { uid, gid }
            }
            ManagedDbPrivilege::Caps => PrivilegeDirective::Caps(managed_db_caps()),
        })
    }
}

/// Auto-register the compute workload(s) backing each **managed co-located** database
/// (compute-backed, no `password_env`), so declaring the `databases` binding is enough
/// to boot the DB at serve time — no separate `compute set` / apply step (turnkey from
/// a stock image on a bare host). Idempotent and non-clobbering: a workload the
/// operator declared explicitly (apply / admin API) always wins; this only fills an
/// absent one, and re-running is a no-op. Best-effort — a failure is logged, never
/// fatal (serving proceeds; the reconcile simply has nothing to launch for that DB
/// until its workload exists).
///
/// **Every compute-backed managed binding is per-tenant**, so this is tenant-aware —
/// it must never register a tenant-blind bare `<compute>`/`default` workload that would
/// collide with the tenant-aware `<compute>-<ident>` the resolver/provisioner produce
/// (two servers, two `initdb` passwords, auth chaos). Per isolation:
///
/// - **`Shared`** — one shared server hosts every tenant's per-tenant database + role,
///   so there IS exactly one server workload: the bare `<compute>` under the reserved
///   default project, initialized from the binding's own env. Register it (non-clobbering);
///   per-tenant DDL stays lazy (no boot-time connection).
/// - **`Single`** — each tenant gets a *dedicated* container `<compute>-<ident>`, created
///   durably by [`provision_single`](crate::tenant_sql::provision_single) /
///   [`provision_tenant`](crate::tenant_sql::provision_tenant) on the tenant's first `sql`
///   resolve (the lazy path), and relaunched by the reconcile on every boot. So there is
///   **nothing to warm at boot**: a tenant that has ever used the DB is already registered
///   (and thus relaunched), and a project that has never used `sql` — e.g. a static-only
///   `default` — must NOT get a spurious `pg`/`pg-<ident>`. The old boot-warm enumerated
///   any project with a *site or function* (not one that uses `sql`), which over-warmed
///   static-only projects into a running DB; dropped entirely here. (Result: a Single
///   binding registers no workload at boot; the first `sql` resolve provisions it.)
///
/// The synthesized spec comes from [`managed_db_spec`](boatramp_core::compute::managed_db_spec)
/// — the same builder the container capability gate exercises — so the shipped
/// managed-DB workload never diverges from the tested one.
#[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
pub async fn auto_register_managed_db_workloads(
    deploy: &DeployStore,
    databases: &std::collections::BTreeMap<String, crate::config::ExternalDatabaseConfig>,
) {
    use crate::config::TenantIsolation;

    for db in databases.values() {
        if !db.is_managed_credential() {
            continue;
        }
        let Some(workload) = db.compute.as_deref().filter(|c| !c.is_empty()) else {
            continue;
        };
        // Config validation rejects an unparsable engine before serve; skip defensively.
        if ExternalSqlKind::parse(&db.kind).is_none() {
            continue;
        }
        match db.tenant {
            // One shared server for all tenants: register the bare `<compute>` under the
            // reserved default project (per-tenant databases live inside it, provisioned
            // lazily — no boot-time connection).
            TenantIsolation::Shared => {
                register_shared_server(deploy, db, workload).await;
            }
            // A dedicated container per tenant, created durably by the lazy resolve
            // (`provision_single`) on first `sql` use and relaunched by the reconcile on
            // each boot. Nothing to warm at boot — warming any project with a site/function
            // (not one that uses `sql`) over-warmed static-only projects into a running DB.
            TenantIsolation::Single => {}
        }
    }
}

/// Register the ONE shared server workload for a `Shared` binding: the bare `<compute>`
/// under the reserved default project, non-clobbering (an operator-declared workload
/// wins; a re-run is a no-op). The synthesized spec is the tested `managed_db_spec` with
/// the historical `"data"` volume (a shared server is never per-tenant-volume-isolated).
#[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
async fn register_shared_server(
    deploy: &DeployStore,
    db: &crate::config::ExternalDatabaseConfig,
    workload: &str,
) {
    use boatramp_core::compute::{
        managed_db_spec, ComputeWorkload, ManagedDbEngine, PlacementConstraints,
    };

    /// 10 GiB — the default managed data-volume size when the config sets none.
    const DEFAULT_VOLUME_MIB: u32 = 10 * 1024;

    let engine = match ExternalSqlKind::parse(&db.kind) {
        Some(ExternalSqlKind::Postgres) => ManagedDbEngine::Postgres,
        Some(ExternalSqlKind::Mysql) => ManagedDbEngine::Mysql,
        None => return,
    };
    // Non-clobbering: only fill an absent workload — an operator-declared one
    // (apply / admin API) always wins, which also makes re-runs idempotent.
    match deploy
        .get_compute_workload(ProjectRef::DEFAULT, workload)
        .await
    {
        Ok(Some(_)) => return,
        Ok(None) => {}
        Err(e) => {
            tracing::warn!(%workload, error = %e, "managed sql: could not check for an existing compute workload; skipping auto-register");
            return;
        }
    }
    let image = db.image.as_deref();
    let mut spec = managed_db_spec(
        engine,
        image,
        db.volume_size_mib.unwrap_or(DEFAULT_VOLUME_MIB),
    );
    // An operator-set startup grace overrides the engine default the synthesizer picked.
    // Applied identically here and in `provision_single` so the two managed-registration
    // paths build the byte-identical (content-addressed) spec.
    if let Some(grace) = db.startup_grace_secs {
        spec.startup_grace_secs = grace;
    }
    let spec_id = match deploy.put_compute_spec(&spec).await {
        Ok(id) => id,
        Err(e) => {
            tracing::warn!(%workload, error = %e, "managed sql: could not store the auto-registered compute spec");
            return;
        }
    };
    let wl = ComputeWorkload {
        version: 1,
        name: workload.to_string(),
        active: spec_id,
        replicas: 1,
        placement: PlacementConstraints::default(),
    };
    match deploy.set_compute_workload(ProjectRef::DEFAULT, &wl).await {
        Ok(()) => tracing::info!(
            %workload,
            image = %image.unwrap_or_else(|| engine.default_image()),
            "managed sql: auto-registered the shared co-located database compute workload"
        ),
        Err(e) => {
            tracing::warn!(%workload, error = %e, "managed sql: could not register the auto-registered compute workload")
        }
    }
}

/// A [`ComputeEndpointResolver`] backed by the control-plane replica state: it
/// lists a workload's **healthy, running** replicas (primary-first by replica
/// index) as `(host, port)`, scoped to a fixed project. Backs the handler's
/// [`ComputeResolvedSqlBackend`](boatramp_storage::sql_compute::ComputeResolvedSqlBackend)
/// so a managed `sql` binding follows its DB workload across restarts.
#[cfg_attr(not(feature = "handlers"), allow(dead_code))]
pub struct DeployEndpointResolver {
    deploy: DeployStore,
    project: String,
}

impl DeployEndpointResolver {
    #[cfg_attr(not(feature = "handlers"), allow(dead_code))]
    pub fn new(deploy: DeployStore, project: impl Into<String>) -> Self {
        Self {
            deploy,
            project: project.into(),
        }
    }
}

#[async_trait]
impl ComputeEndpointResolver for DeployEndpointResolver {
    async fn endpoints(&self, workload: &str) -> Result<Vec<(String, u16)>, SqlError> {
        let states = self
            .deploy
            .list_replica_states(ProjectRef::new(&self.project), workload)
            .await
            .map_err(SqlError::other)?;
        Ok(states
            .into_iter()
            .filter(|s| s.phase == ReplicaPhase::Running && s.healthy)
            .map(|s| (s.endpoint.host, s.endpoint.port))
            .collect())
    }

    /// Every replica state the control plane holds for `workload` (healthy or not), so
    /// a "no healthy replica" error can honestly say whether replicas exist but none
    /// passed the readiness probe — the reachability/health vs missing-workload split.
    /// Off the hot path (only when `endpoints` came back empty); a store error yields no
    /// diagnostics (the caller then falls back to the plainer message).
    async fn replica_diagnostics(&self, workload: &str) -> Vec<ReplicaDiag> {
        let states = match self
            .deploy
            .list_replica_states(ProjectRef::new(&self.project), workload)
            .await
        {
            Ok(s) => s,
            Err(_) => return Vec::new(),
        };
        states
            .into_iter()
            .map(|s| ReplicaDiag {
                endpoint: format!("{}:{}", s.endpoint.host, s.endpoint.port),
                healthy: s.healthy,
                phase: format!("{:?}", s.phase),
            })
            .collect()
    }
}

/// The node's [`OperatorSql`](boatramp_core::sql::OperatorSql): the operator-facing
/// migration/query capability over the handler `sql` `databases`. For a requested
/// database name it (re)builds the same connection the handler runtime uses — a
/// managed credential resolved + unsealed, or a bring-your-own URL from the
/// environment — and runs the script/query server-side (the credential never leaves
/// the node). Backs `POST /api/sql/{db}/{exec,query}`; admin-gated at the API.
#[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
pub struct NodeOperatorSql {
    databases: std::collections::BTreeMap<String, crate::config::ExternalDatabaseConfig>,
    kv: Arc<dyn KvStore>,
    envelope: Option<Arc<dyn KeyEnvelope>>,
    deploy: DeployStore,
}

#[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
impl NodeOperatorSql {
    /// Build over the handler `sql` `databases` config + the credential store.
    pub fn new(
        databases: std::collections::BTreeMap<String, crate::config::ExternalDatabaseConfig>,
        kv: Arc<dyn KvStore>,
        envelope: Option<Arc<dyn KeyEnvelope>>,
        deploy: DeployStore,
    ) -> Self {
        Self {
            databases,
            kv,
            envelope,
            deploy,
        }
    }

    /// Resolve + connect the SQL backend for database `db` in `project` (managed or
    /// bring-your-own), mirroring the handler runtime's per-database construction. The
    /// operator path (`sql exec/query/ping`) — connects as the configured identity (the
    /// superuser on a Shared server).
    pub(crate) async fn backend_for(
        &self,
        project: &str,
        db: &str,
    ) -> Result<Arc<dyn boatramp_core::sql::SqlBackend>, SqlError> {
        self.connect_for(project, db, false).await
    }

    /// Resolve + connect as the project **owner** role (the schema-migration path). Identical
    /// to [`backend_for`](Self::backend_for) except on a Shared server it connects as the
    /// per-project non-superuser owner role with the sealed owner credential — never the
    /// cluster superuser (see [`owner_target`]).
    pub(crate) async fn owner_backend_for(
        &self,
        project: &str,
        db: &str,
    ) -> Result<Arc<dyn boatramp_core::sql::SqlBackend>, SqlError> {
        self.connect_for(project, db, true).await
    }

    /// The SQL engine of managed database `db`, if configured (for the migration runner's
    /// engine gate — schema migrations are Postgres-only in this release).
    pub(crate) fn engine_kind(&self, db: &str) -> Option<ExternalSqlKind> {
        self.databases
            .get(db)
            .and_then(|cfg| ExternalSqlKind::parse(&cfg.kind))
    }

    /// Shared body of [`backend_for`](Self::backend_for) / [`owner_backend_for`](Self::owner_backend_for):
    /// `owner` selects [`owner_target`] (the non-superuser owner role) over [`operator_target`]
    /// (the configured/superuser identity) for the compute-backed case.
    async fn connect_for(
        &self,
        project: &str,
        db: &str,
        owner: bool,
    ) -> Result<Arc<dyn boatramp_core::sql::SqlBackend>, SqlError> {
        use boatramp_storage::sql_compute::ComputeResolvedSqlBackend;
        use boatramp_storage::sql_sqlx::{connect, ExternalSqlOptions};
        let cfg = self
            .databases
            .get(db)
            .ok_or_else(|| SqlError::other(format!("no database named {db:?}")))?;
        let kind = ExternalSqlKind::parse(&cfg.kind).ok_or_else(|| {
            SqlError::other(format!("database {db:?}: unknown engine {:?}", cfg.kind))
        })?;
        let timeout = cfg.connect_timeout_secs.map(std::time::Duration::from_secs);
        if cfg.compute.as_deref().is_some_and(|c| !c.is_empty()) {
            // Managed or brought-credential compute-backed database. Every compute-backed
            // binding is **per-tenant**, so derive the tenant the SAME way the resolver
            // does — otherwise operator `sql exec/query` would target the tenant-blind
            // bare `<compute>`/`default` (Bug 2's operator arm) and reach the wrong DB.
            let target = if owner {
                owner_target(cfg, project, db)?
            } else {
                operator_target(cfg, project, db)?
            };

            // The password source: an operator-supplied `password_env` (brought
            // credential) reads the env var as before; a managed credential is unsealed
            // under EXACTLY the key the provisioner/resolver used for this tenant.
            let password = match cfg.password_env.as_deref().filter(|v| !v.is_empty()) {
                Some(var) => std::env::var(var)
                    .map_err(|_| SqlError::other(format!("env var {var} (password) is unset")))?,
                None => {
                    let envelope = self.envelope.clone().ok_or_else(|| {
                        SqlError::other(format!(
                            "managed database {db:?} needs a [secrets] envelope to unseal its credential"
                        ))
                    })?;
                    ManagedSqlCredentials::new(self.kv.clone(), envelope)
                        .password(&target.cred_project, &target.cred_workload)
                        .await
                        .map_err(SqlError::other)?
                }
            };

            let resolver = Arc::new(DeployEndpointResolver::new(
                self.deploy.clone(),
                target.endpoint_project,
            ));
            Ok(Arc::new(ComputeResolvedSqlBackend::new(
                resolver,
                target.workload,
                kind,
                target.database,
                target.user,
                password,
                cfg.pool_max,
                cfg.read_only,
                timeout,
            )))
        } else {
            // Bring-your-own URL (a secret named indirectly by an env var).
            let url = std::env::var(&cfg.url_env)
                .map_err(|_| SqlError::other(format!("env var {} (url) is unset", cfg.url_env)))?;
            let read_url =
                match &cfg.read_url_env {
                    Some(var) => Some(std::env::var(var).map_err(|_| {
                        SqlError::other(format!("env var {var} (read url) is unset"))
                    })?),
                    None => None,
                };
            let opts = ExternalSqlOptions::new(url)
                .with_read_url(read_url)
                .with_max_connections(cfg.pool_max)
                .read_only(cfg.read_only)
                .with_connect_timeout(timeout);
            connect(kind, &opts)
        }
    }
}

/// The tenant-resolved connection target for operator SQL against a compute-backed
/// managed binding — the SAME derivation the per-tenant resolver
/// (`NodeTenantSqlResolver::build_backend`) uses, so `sql exec/query` reaches the
/// tenant's OWN database + credential rather than the tenant-blind bare
/// `<compute>`/`default`. Pure (no IO), so the derivation is unit-testable.
#[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
#[derive(Debug)]
pub(crate) struct OperatorTarget {
    /// The compute workload backing the connection (per-tenant for Single).
    pub workload: String,
    /// The physical database to connect to (the tenant's DB).
    pub database: String,
    /// The role/user to connect as.
    pub user: String,
    /// The project scope the workload's replica endpoints live under.
    pub endpoint_project: String,
    /// The `<project>` segment of the sealed credential's KV key.
    pub cred_project: String,
    /// The `<workload>` segment of the sealed credential's KV key.
    pub cred_workload: String,
}

/// Derive the operator-SQL connection target for a compute-backed managed binding.
///
/// Operator SQL is **project-level** (there is no site in a `POST /api/sql/{db}/...`
/// request), so the tenant is the project: `tenant_key(scope, project, "")`. A
/// **site-scoped** managed DB has no single database at the project level, so this
/// fails with a clear error rather than silently targeting the wrong (e.g. default) DB.
///
/// - **Single** — target the per-tenant workload `<compute>-<ident>` (bare `<compute>`
///   for the default tenant), connect as the configured `user` to `names.database`,
///   credential keyed by the workload's own `(single_credential_project, workload)`.
/// - **Shared** — target the shared `<compute>` server, connect as the configured
///   `user` (the server superuser, so an operator migration can touch any tenant's DB)
///   to `names.database`, credential = the superuser's under `(DEFAULT_PROJECT,
///   <compute>)`, never a per-tenant key.
#[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
pub(crate) fn operator_target(
    cfg: &crate::config::ExternalDatabaseConfig,
    project: &str,
    db: &str,
) -> Result<OperatorTarget, SqlError> {
    use crate::config::{TenantIsolation, TenantScope};
    use crate::tenant_sql::{single_credential_project, tenant_key, tenant_names};
    use boatramp_core::project::DEFAULT_PROJECT;

    let compute = cfg
        .compute
        .as_deref()
        .filter(|c| !c.is_empty())
        .ok_or_else(|| SqlError::other(format!("database {db:?}: not compute-backed")))?;
    let database = cfg.database.as_deref().unwrap_or_default();
    let user = cfg.user.as_deref().unwrap_or_default();

    if matches!(cfg.tenant_scope, TenantScope::Site) {
        return Err(SqlError::other(format!(
            "database {db:?} is a site-scoped managed database; operator sql exec/query is \
             project-level and cannot target a specific site's database"
        )));
    }

    let (tenant_ident_raw, is_default) = tenant_key(cfg.tenant_scope, project, "");
    let names = tenant_names(cfg.tenant, compute, database, &tenant_ident_raw, is_default);

    let (cred_project, cred_workload) = match cfg.tenant {
        TenantIsolation::Single => (
            single_credential_project(project, is_default),
            names.workload.clone(),
        ),
        // Shared: the superuser credential, under the reserved default project + the
        // bare `<compute>` (exactly the server-init key), never a per-tenant key.
        TenantIsolation::Shared => (DEFAULT_PROJECT.to_string(), compute.to_string()),
    };
    let endpoint_project = match cfg.tenant {
        TenantIsolation::Single if !is_default => project.to_string(),
        _ => DEFAULT_PROJECT.to_string(),
    };

    Ok(OperatorTarget {
        workload: names.workload,
        database: names.database,
        user: user.to_string(),
        endpoint_project,
        cred_project,
        cred_workload,
    })
}

/// Derive the connection target for a **schema migration** against a compute-backed managed
/// binding — identical to [`operator_target`] EXCEPT, on a multi-tenant **Shared** server, it
/// connects as the per-project non-superuser **owner role** with the sealed owner credential,
/// NOT the cluster superuser. This is the whole point of the owner-gated migration surface: DDL
/// runs bounded to project-owner authority (Postgres denies the owner role cross-database /
/// role-escalation / `COPY … TO PROGRAM` by privilege), never as a cluster superuser.
///
/// - **Shared, real tenant** — user = `names.owner_role`, credential keyed by
///   `(project, owner_credential_workload_key(<compute>, <ident>))` (the sealed owner password
///   minted at provision). This is the security-critical deviation from `operator_target`.
/// - **Shared, default tenant** / **Single** / **bring-your-own** — identical to `operator_target`
///   (the configured user is already the DB owner and is `<= project-owner`: a single-tenant
///   install's own user, a per-workload Single user, or the operator's own external credential).
#[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
pub(crate) fn owner_target(
    cfg: &crate::config::ExternalDatabaseConfig,
    project: &str,
    db: &str,
) -> Result<OperatorTarget, SqlError> {
    use crate::config::{TenantIsolation, TenantScope};
    use crate::tenant_sql::{owner_credential_workload_key, tenant_key, tenant_names};

    // Start from the operator derivation, then override the identity/credential for the one case
    // that must NOT be the superuser: a real tenant on a Shared server.
    let mut target = operator_target(cfg, project, db)?;

    if matches!(cfg.tenant, TenantIsolation::Shared)
        && !matches!(cfg.tenant_scope, TenantScope::Site)
    {
        let compute = cfg.compute.as_deref().unwrap_or_default();
        let database = cfg.database.as_deref().unwrap_or_default();
        let (tenant_ident_raw, is_default) = tenant_key(cfg.tenant_scope, project, "");
        if !is_default {
            let names = tenant_names(cfg.tenant, compute, database, &tenant_ident_raw, is_default);
            let ident = boatramp_storage::tenant_provision::sanitize_ident(&tenant_ident_raw);
            target.user = names.owner_role;
            target.cred_project = project.to_string();
            target.cred_workload = owner_credential_workload_key(compute, &ident);
        }
    }
    Ok(target)
}

#[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
#[async_trait]
impl boatramp_core::sql::OperatorSql for NodeOperatorSql {
    async fn exec_script(&self, project: &str, db: &str, script: &str) -> Result<(), SqlError> {
        self.backend_for(project, db)
            .await?
            .run_script(script)
            .await
    }

    async fn query(
        &self,
        project: &str,
        db: &str,
        sql: &str,
    ) -> Result<boatramp_core::sql::SqlRows, SqlError> {
        self.backend_for(project, db).await?.run_query(sql).await
    }

    async fn ping(
        &self,
        project: &str,
        db: &str,
    ) -> Result<Vec<boatramp_core::sql::SqlPingReplica>, SqlError> {
        use boatramp_core::sql::SqlPingReplica;
        use std::time::Duration;
        let cfg = self
            .databases
            .get(db)
            .ok_or_else(|| SqlError::other(format!("no database named {db:?}")))?;
        // A bring-your-own-URL binding isn't compute-backed, so there is no replica
        // fleet to probe — ping is for managed co-located databases.
        if cfg.compute.as_deref().is_none_or(|c| c.is_empty()) {
            return Err(SqlError::other(format!(
                "database {db:?} is not compute-backed; `sql ping` probes managed co-located \
                 replicas only"
            )));
        }
        // Derive the tenant's workload + endpoint project EXACTLY as `backend_for` does,
        // then read ALL replicas (healthy or not) from the resolver — bypassing the
        // healthy filter that `query` would hit — and actively TCP-probe each.
        let target = operator_target(cfg, project, db)?;
        let resolver = DeployEndpointResolver::new(self.deploy.clone(), target.endpoint_project);
        let diags = resolver.replica_diagnostics(&target.workload).await;
        let mut out = Vec::with_capacity(diags.len());
        for d in diags {
            let reachable = match d.endpoint.parse::<std::net::SocketAddr>() {
                Ok(addr) => matches!(
                    tokio::time::timeout(
                        Duration::from_secs(2),
                        tokio::net::TcpStream::connect(addr),
                    )
                    .await,
                    Ok(Ok(_))
                ),
                // An unparsable endpoint can't be probed — report it as unreachable.
                Err(_) => false,
            };
            out.push(SqlPingReplica {
                endpoint: d.endpoint,
                healthy: d.healthy,
                phase: d.phase,
                tcp_reachable: reachable,
            });
        }
        Ok(out)
    }
}

/// The schema-migration runner (Postgres). Owns the ordered `schema_migrations` ledger and applies
/// the pending suffix of a step set, once each, transactionally-where-possible, connecting as the
/// per-project non-superuser **owner** role (via [`NodeOperatorSql::owner_backend_for`]) for the
/// ledger + `sql` steps, and as the superuser (via [`NodeOperatorSql::backend_for`]) ONLY for the
/// allowlist-gated, host-templated `CREATE EXTENSION`. boatramp owns ordering/idempotency/atomicity;
/// the caller supplies the ordered steps. `Project·Admin`-gated at the API.
#[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
pub struct NodeMigrationRunner {
    op: Arc<NodeOperatorSql>,
    /// The operator's trusted-extension allowlist — the ONLY extensions an `Extension` step may
    /// enable (a name not here is refused fail-closed).
    trusted_extensions: std::collections::BTreeSet<String>,
}

/// The host-owned schema holding the migration ledger — owner-role-owned, and NOT in `public`, so
/// the runtime tenant role (which only ever gets DML on `public` via `grant_app_role_ddl`) has no
/// access to it: the ledger is append-only from the app's perspective by schema isolation.
#[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
const LEDGER_SCHEMA: &str = "boatramp_migrations";
#[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
const LEDGER_TABLE: &str = "schema_migrations";

/// Quote a SQL string literal (single-quoted, doubling embedded `'`). The ledger id/hash reaching
/// this are already validated ([`valid_migration_id`]) / hex, but we quote defensively so no value
/// can break out of its literal in the host-built ledger INSERT.
#[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
fn sql_quote_literal(s: &str) -> String {
    format!("'{}'", s.replace('\'', "''"))
}

/// A migration id is restricted to `[A-Za-z0-9._-]+` (non-empty) — defensive belt beside the
/// literal-quoting, and it keeps ledger ids clean/greppable. Anything else is refused fail-closed.
#[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
fn valid_migration_id(id: &str) -> bool {
    !id.is_empty()
        && id
            .chars()
            .all(|c| c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-'))
}

/// Whether `script` contains a `CREATE EXTENSION` statement. Raw `sql` steps are refused if so —
/// extensions must go through an allowlist-gated `Extension` step. (A UX/allowlist-consistency guard;
/// the security boundary is that the owner role is non-superuser and so cannot create an UNtrusted
/// extension regardless.) Delegates to the comment-/casing-immune tokenizer ([Security review
/// HIGH-1]: a byte scan is evadable via `CREATE/**/EXTENSION`; the tokenizer strips comments and
/// fails closed on an unlexable script).
#[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
fn mentions_create_extension(script: &str) -> bool {
    boatramp_core::sql::script_has_create_extension(script)
}

/// Whether `script` issues its own transaction control (`BEGIN`/`START`/`COMMIT`/`END`/`ROLLBACK`/
/// `ABORT`/`SAVEPOINT`/`RELEASE`). A transactional (`!no_transaction`) step is wrapped by the runner
/// in `BEGIN;…;<ledger-insert>;COMMIT;`, so an author's own `COMMIT`/`ROLLBACK` would desync that
/// wrapper (a `ROLLBACK` reverts the DDL but the ledger INSERT then auto-commits, recording a step
/// that didn't apply). Reject it fail-closed so the ledger can never diverge from applied state; an
/// author that genuinely needs its own transaction control uses a `no_transaction` step.
///
/// Delegates to the SQL **tokenizer** ([Security review HIGH-1/HIGH-2]): the earlier byte scan was
/// comment-evadable (`COMMIT-- x`, `/*c*/BEGIN`), missed `ABORT`/`SAVEPOINT`/`RELEASE`, and
/// false-positived on `CASE … END` and PL/pgSQL `$$ BEGIN … END $$` bodies. The tokenizer is
/// comment-/casing-/string-immune, CASE-aware, and fails closed on an unlexable script.
#[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
fn mentions_txn_control(script: &str) -> bool {
    boatramp_core::sql::script_has_txn_control(script)
}

/// Whether `script` references the host-owned migration-ledger schema (`boatramp_migrations`). The
/// owner role OWNS that schema, so an unguarded `migrate::exec` from a function step could rewrite
/// prefix-consistency/immutability history. Refused fail-closed (Security S3). Uses the tokenizer so
/// a reference can't be hidden in a comment ([Security review HIGH-1]) and a string literal
/// mentioning the name is not a false positive; matches the schema as a whole word token (quoted or
/// not). Belt-and-braces beside the schema isolation (the runtime tenant role has no grant on it at
/// all); this stops the OWNER-role path a function step runs on.
#[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
fn mentions_ledger_schema(script: &str) -> bool {
    boatramp_core::sql::script_references_word(script, LEDGER_SCHEMA)
}

/// The host-mediated owner-role DDL seam backing the guest `migrate-ddl` capability of a `function`
/// step (Security S5). Holds the orchestrator-owned OWNER-role backend for one `(project, db)`; the
/// guest never holds the credential. Each call enforces the ledger-schema (S3) + transaction-control
/// (S4) guards host-side before touching the wire, then auto-commits via `run_script`/`run_query`.
#[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
pub(crate) struct OwnerDdl {
    owner: Arc<dyn boatramp_core::sql::SqlBackend>,
}

#[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
impl OwnerDdl {
    /// Guard a guest-supplied script/query: refuse a ledger-schema reference (S3) or its own
    /// transaction control (S4) before it reaches the owner connection.
    fn guard(script: &str) -> Result<(), MigrateDdlError> {
        if mentions_ledger_schema(script) {
            return Err(MigrateDdlError::LedgerProtected);
        }
        if mentions_txn_control(script) {
            return Err(MigrateDdlError::TxnControl);
        }
        Ok(())
    }
}

#[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
#[async_trait]
impl MigrateDdl for OwnerDdl {
    async fn exec(&self, script: &str) -> Result<(), MigrateDdlError> {
        Self::guard(script)?;
        self.owner
            .run_script(script)
            .await
            .map_err(|e| MigrateDdlError::Sql(sanitize_migration_error(&e.to_string())))
    }

    async fn exec_batch(&self, scripts: Vec<String>) -> Result<(), MigrateDdlError> {
        for script in &scripts {
            self.exec(script).await?;
        }
        Ok(())
    }

    async fn query(&self, sql: &str) -> Result<boatramp_core::sql::SqlRows, MigrateDdlError> {
        Self::guard(sql)?;
        let rows = self
            .owner
            .run_query(sql)
            .await
            .map_err(|e| MigrateDdlError::Sql(sanitize_migration_error(&e.to_string())))?;
        // Bound the result handed to the guest ([Security review MEDIUM-2]): a verification query
        // is expected to return a small, checkable set; refuse an unbounded read rather than encode
        // a huge JSON blob into host memory + across the boundary. The author adds a `LIMIT`.
        if rows.rows.len() > MIGRATE_QUERY_MAX_ROWS {
            return Err(MigrateDdlError::Sql(format!(
                "query returned {} rows (cap {MIGRATE_QUERY_MAX_ROWS}); add a LIMIT — a migration \
                 verification query should read a bounded set",
                rows.rows.len()
            )));
        }
        Ok(rows)
    }
}

/// The row cap on a `migrate-ddl` `query` result (Security MEDIUM-2). A verification query reads a
/// bounded, checkable set; beyond this it is refused so a migration function can't drive an unbounded
/// owner-visibility read into a huge host-side JSON string.
#[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
const MIGRATE_QUERY_MAX_ROWS: usize = 100_000;

#[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
impl NodeMigrationRunner {
    /// Build over a [`NodeOperatorSql`] (for the owner + superuser backends) and the operator's
    /// trusted-extension allowlist.
    pub fn new(
        op: Arc<NodeOperatorSql>,
        trusted_extensions: std::collections::BTreeSet<String>,
    ) -> Self {
        Self {
            op,
            trusted_extensions,
        }
    }

    /// The fully-qualified, quoted ledger table name.
    fn ledger() -> String {
        use boatramp_storage::tenant_provision::quote_ident;
        format!(
            "{}.{}",
            quote_ident(ExternalSqlKind::Postgres, LEDGER_SCHEMA),
            quote_ident(ExternalSqlKind::Postgres, LEDGER_TABLE)
        )
    }

    /// The host-built ledger INSERT for one recorded step (all values host-controlled + quoted). The
    /// orchestrator supplies the **effective** hash (== the intrinsic content hash for sql/extension;
    /// blob-bound for a function step) and the `origin` (`apply` vs `baseline`, recorded in
    /// `applied_by` for the U6 marker).
    fn ledger_insert(
        step: &MigrationStep,
        ordinal: usize,
        effective_hash: &str,
        origin: LedgerOrigin,
    ) -> String {
        format!(
            "INSERT INTO {ledger} (id, ordinal, content_hash, kind, applied_by) \
             VALUES ({id}, {ord}, {hash}, {kind}, {origin});",
            ledger = Self::ledger(),
            id = sql_quote_literal(&step.id),
            ord = ordinal,
            hash = sql_quote_literal(effective_hash),
            kind = sql_quote_literal(step.kind()),
            origin = sql_quote_literal(origin.as_str()),
        )
    }

    /// Ensure the ledger schema + table exist (idempotent), as the owner role.
    async fn ensure_ledger(
        &self,
        owner: &Arc<dyn boatramp_core::sql::SqlBackend>,
    ) -> Result<(), SqlError> {
        use boatramp_storage::tenant_provision::quote_ident;
        let schema = quote_ident(ExternalSqlKind::Postgres, LEDGER_SCHEMA);
        owner
            .run_script(&format!("CREATE SCHEMA IF NOT EXISTS {schema};"))
            .await?;
        owner
            .run_script(&format!(
                "CREATE TABLE IF NOT EXISTS {ledger} (\
                 id text PRIMARY KEY, \
                 ordinal integer NOT NULL, \
                 content_hash text NOT NULL, \
                 kind text NOT NULL, \
                 applied_at timestamptz NOT NULL DEFAULT now(), \
                 applied_by text);",
                ledger = Self::ledger()
            ))
            .await?;
        Ok(())
    }

    /// Read the applied ledger rows, ordered by `ordinal`. `applied_by` maps to the row `origin`
    /// (`baseline` when it was baselined, else `apply` — covering NULL/legacy rows via the default).
    async fn read_applied(
        &self,
        owner: &Arc<dyn boatramp_core::sql::SqlBackend>,
    ) -> Result<Vec<AppliedMigration>, SqlError> {
        use boatramp_core::sql::SqlValue;
        let rows = owner
            .run_query(&format!(
                "SELECT id, ordinal, content_hash, kind, applied_at::text, applied_by \
                 FROM {ledger} ORDER BY ordinal;",
                ledger = Self::ledger()
            ))
            .await?;
        let text = |v: &SqlValue| match v {
            SqlValue::Text(s) => s.clone(),
            other => format!("{other:?}"),
        };
        let int = |v: &SqlValue| match v {
            SqlValue::Integer(n) => *n,
            _ => 0,
        };
        Ok(rows
            .rows
            .iter()
            .map(|r| {
                let origin = match r.get(5) {
                    Some(SqlValue::Text(s)) if s == LedgerOrigin::Baseline.as_str() => {
                        LedgerOrigin::Baseline.as_str().to_string()
                    }
                    _ => LedgerOrigin::Apply.as_str().to_string(),
                };
                AppliedMigration {
                    id: r.first().map(&text).unwrap_or_default(),
                    ordinal: r.get(1).map(&int).unwrap_or_default(),
                    content_hash: r.get(2).map(&text).unwrap_or_default(),
                    kind: r.get(3).map(&text).unwrap_or_default(),
                    applied_at: r.get(4).map(&text).unwrap_or_default(),
                    origin,
                }
            })
            .collect())
    }
}

#[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
impl NodeMigrationRunner {
    /// Engine gate: Postgres only in this release (the owner-role/RLS model + transactional DDL are
    /// Postgres semantics; MySQL Shared has no owner/runtime split). Fail closed + clear.
    fn engine_gate(&self, db: &str) -> Result<(), MigrationError> {
        match self.op.engine_kind(db) {
            Some(ExternalSqlKind::Postgres) => Ok(()),
            Some(_) => Err(MigrationError::Other(format!(
                "database {db:?}: schema migrations are supported on Postgres only in this release"
            ))),
            None => Err(MigrationError::NotConfigured),
        }
    }

    /// Connect as the OWNER role (never the superuser). A managed DB still starting surfaces as
    /// `Unavailable` → a retryable 503 at the API, not a permanent failure.
    async fn connect_owner(
        &self,
        project: &str,
        db: &str,
    ) -> Result<Arc<dyn boatramp_core::sql::SqlBackend>, MigrationError> {
        self.op
            .owner_backend_for(project, db)
            .await
            .map_err(|e| match e {
                SqlError::Unavailable(m) => MigrationError::Unavailable(m),
                other => MigrationError::Sql(other),
            })
    }
}

#[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
#[async_trait]
impl MigrationSubstrate for NodeMigrationRunner {
    async fn preflight(
        &self,
        project: &str,
        db: &str,
    ) -> Result<Vec<AppliedMigration>, MigrationError> {
        self.engine_gate(db)?;
        let owner = self.connect_owner(project, db).await?;
        self.ensure_ledger(&owner).await?;
        Ok(self.read_applied(&owner).await?)
    }

    async fn apply_substrate_step(
        &self,
        project: &str,
        db: &str,
        step: &MigrationStep,
        ordinal: usize,
        effective_hash: &str,
    ) -> Result<SubstrateStepOutcome, MigrationError> {
        use boatramp_core::sql::MigrationAction;

        // Defensive: the orchestrator already validated ids, but this layer builds the ledger SQL, so
        // re-check (a bad id is a per-step failure, not an infra error).
        if !valid_migration_id(&step.id) {
            return Ok(SubstrateStepOutcome::Failed(
                "invalid migration id (allowed: A-Za-z0-9._-)".to_string(),
            ));
        }
        let owner = self.connect_owner(project, db).await?;
        let ledger_insert = Self::ledger_insert(step, ordinal, effective_hash, LedgerOrigin::Apply);
        let outcome: Result<(), String> = match &step.action {
            MigrationAction::Sql {
                script,
                no_transaction,
            } => {
                if mentions_create_extension(script) {
                    Err("a sql step may not CREATE EXTENSION — use an extension step".to_string())
                } else if mentions_ledger_schema(script) {
                    Err(
                        "a sql step may not reference the host-owned migration-ledger schema"
                            .to_string(),
                    )
                } else if !*no_transaction && mentions_txn_control(script) {
                    Err("a transactional sql step may not contain its own \
                         BEGIN/COMMIT/ROLLBACK (it would desync the atomic wrapper) — use a \
                         no_transaction step to manage the transaction yourself"
                        .to_string())
                } else if *no_transaction {
                    // Non-transactional DDL: run the script, then record the ledger row.
                    match owner.run_script(script).await {
                        Ok(()) => owner
                            .run_script(&ledger_insert)
                            .await
                            .map_err(|e| e.to_string()),
                        Err(e) => Err(e.to_string()),
                    }
                } else {
                    // Atomic: DDL + ledger insert commit together (or roll back together). A trailing
                    // `;` after the script guards a script that omits its own final semicolon (else it
                    // would merge with the ledger INSERT into one invalid statement); a doubled `;;`
                    // is just an empty statement, harmless.
                    let batch = format!("BEGIN;\n{script};\n{ledger_insert}\nCOMMIT;");
                    owner.run_script(&batch).await.map_err(|e| e.to_string())
                }
            }
            MigrationAction::Extension { name } => {
                if !self.trusted_extensions.contains(name) {
                    Err(format!(
                        "extension {name:?} is not on the operator trusted-extension allowlist"
                    ))
                } else {
                    use boatramp_storage::tenant_provision::quote_ident;
                    // Host-templated + allowlist-bounded; run via the superuser so an allowlisted
                    // superuser-only extension also works. IF NOT EXISTS keeps it idempotent, so a
                    // crash before the ledger insert re-runs harmlessly.
                    let create = format!(
                        "CREATE EXTENSION IF NOT EXISTS {};",
                        quote_ident(ExternalSqlKind::Postgres, name)
                    );
                    match self.op.backend_for(project, db).await {
                        Ok(su) => match su.run_script(&create).await {
                            Ok(()) => owner
                                .run_script(&ledger_insert)
                                .await
                                .map_err(|e| e.to_string()),
                            Err(e) => Err(e.to_string()),
                        },
                        Err(e) => Err(e.to_string()),
                    }
                }
            }
            // A `function` step is invoked by the server-side orchestrator (it needs the invoke
            // kernel) and recorded via `record` — it must never reach the substrate executor.
            MigrationAction::Function { .. } => {
                return Err(MigrationError::Other(
                    "internal: a function step must be invoked by the orchestrator, not the \
                     substrate"
                        .to_string(),
                ))
            }
        };
        Ok(match outcome {
            Ok(()) => SubstrateStepOutcome::Applied,
            Err(error) => SubstrateStepOutcome::Failed(sanitize_migration_error(&error)),
        })
    }

    async fn record(
        &self,
        project: &str,
        db: &str,
        step: &MigrationStep,
        ordinal: usize,
        effective_hash: &str,
        origin: LedgerOrigin,
    ) -> Result<(), MigrationError> {
        if !valid_migration_id(&step.id) {
            return Err(MigrationError::Other(format!(
                "invalid migration id {:?} (allowed: A-Za-z0-9._-)",
                step.id
            )));
        }
        let owner = self.connect_owner(project, db).await?;
        self.ensure_ledger(&owner).await?;
        owner
            .run_script(&Self::ledger_insert(step, ordinal, effective_hash, origin))
            .await?;
        Ok(())
    }

    async fn owner_ddl(
        &self,
        project: &str,
        db: &str,
    ) -> Result<Arc<dyn MigrateDdl>, MigrationError> {
        self.engine_gate(db)?;
        let owner = self.connect_owner(project, db).await?;
        Ok(Arc::new(OwnerDdl { owner }))
    }
}

/// Trim a driver error string to a first line and a bounded length so a migration failure returned
/// to the client can't carry a wall of internal DSN/schema detail (defense against info leakage).
#[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
fn sanitize_migration_error(e: &str) -> String {
    let first = e.lines().next().unwrap_or(e);
    // Truncate on a CHAR boundary, not a byte index ([Security review MEDIUM-3]): a driver error can
    // carry multi-byte UTF-8 (non-ASCII table/collation names, a localized Postgres message) that a
    // migration author influences, so `&first[..300]` could panic mid-codepoint.
    if first.chars().count() > 300 {
        let truncated: String = first.chars().take(300).collect();
        format!("{truncated}…")
    } else {
        first.to_string()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use async_trait::async_trait;
    use boatramp_core::envelope::EnvelopeError;
    use boatramp_core::kv::MemoryKv;

    /// A trivial reversible "envelope" for tests — NOT encryption; it just proves the
    /// stored blob is transformed (sealed) and round-trips (cf. cert.rs's test double).
    struct ReverseEnvelope;
    #[async_trait]
    impl KeyEnvelope for ReverseEnvelope {
        async fn wrap(&self, plaintext: &[u8]) -> Result<Vec<u8>, EnvelopeError> {
            Ok(plaintext.iter().rev().copied().collect())
        }
        async fn unwrap(&self, wrapped: &[u8]) -> Result<Vec<u8>, EnvelopeError> {
            Ok(wrapped.iter().rev().copied().collect())
        }
    }

    #[tokio::test]
    async fn password_is_generated_once_sealed_and_stable() {
        let kv: Arc<dyn KvStore> = Arc::new(MemoryKv::new());
        let creds = ManagedSqlCredentials::new(kv.clone(), Arc::new(ReverseEnvelope));

        let pw = creds.password("default", "pg").await.unwrap();
        assert_eq!(pw.len(), 64, "32 random bytes, hex-encoded");

        // Stable: a second call unseals the stored value, it is not regenerated.
        assert_eq!(creds.password("default", "pg").await.unwrap(), pw);

        // Stored SEALED, never in cleartext.
        let raw = kv
            .get("managed-sql-cred/default/pg")
            .await
            .unwrap()
            .unwrap();
        assert_ne!(
            raw,
            pw.as_bytes(),
            "the stored blob is sealed, not the password"
        );
        assert_eq!(
            raw.iter().rev().copied().collect::<Vec<u8>>(),
            pw.as_bytes()
        );

        // A fresh store instance (a restart) unseals the SAME password.
        let after_restart = ManagedSqlCredentials::new(kv, Arc::new(ReverseEnvelope));
        assert_eq!(after_restart.password("default", "pg").await.unwrap(), pw);

        // A different workload gets a different password.
        assert_ne!(creds.password("default", "other").await.unwrap(), pw);
    }

    #[test]
    fn server_env_recipe_per_engine() {
        let pg = managed_db_server_env(ExternalSqlKind::Postgres, "analytics", "app", "pw");
        assert_eq!(
            pg,
            vec![
                ("POSTGRES_USER".into(), "app".into()),
                ("POSTGRES_PASSWORD".into(), "pw".into()),
                ("POSTGRES_DB".into(), "analytics".into()),
            ]
        );
        let my = managed_db_server_env(ExternalSqlKind::Mysql, "shop", "app", "pw");
        // MySQL needs a root password to initialize, plus the app user/db.
        assert!(my.contains(&("MYSQL_USER".into(), "app".into())));
        assert!(my.contains(&("MYSQL_DATABASE".into(), "shop".into())));
        assert!(my.contains(&("MYSQL_ROOT_PASSWORD".into(), "pw".into())));
    }

    use crate::config::ExternalDatabaseConfig;
    use std::collections::BTreeMap;

    fn db(
        kind: &str,
        compute: Option<&str>,
        url_env: &str,
        pw_env: Option<&str>,
    ) -> ExternalDatabaseConfig {
        ExternalDatabaseConfig {
            kind: kind.into(),
            url_env: url_env.into(),
            compute: compute.map(Into::into),
            database: compute.map(|_| "analytics".into()),
            user: compute.map(|_| "app".into()),
            password_env: pw_env.map(Into::into),
            ..Default::default()
        }
    }

    #[tokio::test]
    async fn managed_db_env_only_covers_managed_workloads() {
        let mut dbs = BTreeMap::new();
        // Managed: compute-backed, no password_env.
        dbs.insert(
            "analytics".to_string(),
            db("postgres", Some("pg"), "", None),
        );
        // Bring-your-own credential: compute-backed WITH password_env → not managed.
        dbs.insert(
            "byo".to_string(),
            db("postgres", Some("pg2"), "", Some("PG2_PW")),
        );
        // Bring-your-own URL: not compute-backed → not managed.
        dbs.insert("ext".to_string(), db("mysql", None, "MYSQL_URL", None));

        let kv: Arc<dyn KvStore> = Arc::new(MemoryKv::new());
        let creds = ManagedSqlCredentials::new(kv, Arc::new(ReverseEnvelope));
        let env = ManagedDbEnv::from_config(&dbs, creds, ManagedDbPrivilege::default());
        assert!(!env.is_empty());

        // The managed workload gets a rootless privilege directive by default; the
        // non-managed ones get none.
        assert_eq!(
            env.managed_db_privilege("default", "pg"),
            Some(PrivilegeDirective::Rootless { uid: 999, gid: 999 })
        );
        assert_eq!(env.managed_db_privilege("default", "nope"), None);

        // The managed workload gets its server-init env, sealed-password-derived.
        let pg = env.managed_db_env("default", "pg").await;
        assert!(pg.contains(&("POSTGRES_USER".into(), "app".into())));
        assert!(pg.contains(&("POSTGRES_DB".into(), "analytics".into())));
        let password = pg
            .iter()
            .find(|(k, _)| k == "POSTGRES_PASSWORD")
            .map(|(_, v)| v.clone())
            .expect("password present");
        assert_eq!(password.len(), 64, "managed 32-byte hex password");
        // Idempotent: the same sealed credential each call.
        let pg2 = env.managed_db_env("default", "pg").await;
        assert_eq!(pg, pg2);

        // The BYO-credential + BYO-URL workloads are NOT managed here.
        assert!(env.managed_db_env("default", "pg2").await.is_empty());
        assert!(env.managed_db_env("default", "nope").await.is_empty());
    }

    /// L3: when one `Single` compute base (`pg`) is a `-`-prefix of another
    /// (`pg-metrics`), a per-tenant workload `pg-metrics-<ident>` is a valid derived
    /// name for BOTH. `resolve_spec` must pick the **longest** matching base
    /// deterministically (`pg-metrics`), not whichever the HashMap iterates first, so
    /// the server-init env is filled from the right binding's database/user.
    #[tokio::test]
    async fn resolve_spec_prefers_the_longest_matching_single_base() {
        use crate::config::TenantIsolation;

        // Two Single bindings whose compute names are prefix-related. Give them
        // distinct databases so the resolved spec is observable.
        let mut pg = db("postgres", Some("pg"), "", None);
        pg.tenant = TenantIsolation::Single;
        pg.database = Some("appdb".into());
        pg.user = Some("app".into());

        let mut pg_metrics = db("postgres", Some("pg-metrics"), "", None);
        pg_metrics.tenant = TenantIsolation::Single;
        pg_metrics.database = Some("metricsdb".into());
        pg_metrics.user = Some("metrics".into());

        let mut dbs = BTreeMap::new();
        dbs.insert("analytics".to_string(), pg);
        dbs.insert("metrics".to_string(), pg_metrics);

        let kv: Arc<dyn KvStore> = Arc::new(MemoryKv::new());
        let creds = ManagedSqlCredentials::new(kv, Arc::new(ReverseEnvelope));
        let env = ManagedDbEnv::from_config(&dbs, creds, ManagedDbPrivilege::default());

        // A per-tenant workload of `pg-metrics` matches both `pg` and `pg-metrics`;
        // the longest base (`pg-metrics`) must win → the metrics database/user.
        let e = env.managed_db_env("acme", "pg-metrics-acme").await;
        assert!(
            e.contains(&("POSTGRES_DB".into(), "metricsdb".into())),
            "longest base (`pg-metrics`) must win over `pg`: {e:?}"
        );
        assert!(e.contains(&("POSTGRES_USER".into(), "metrics".into())));

        // A per-tenant workload of the shorter base still resolves to `pg`.
        let e = env.managed_db_env("acme", "pg-acme").await;
        assert!(e.contains(&("POSTGRES_DB".into(), "appdb".into())));
        assert!(e.contains(&("POSTGRES_USER".into(), "app".into())));

        // The privilege lookup uses the same resolver, so it is unambiguous too.
        assert_eq!(
            env.managed_db_privilege("acme", "pg-metrics-acme"),
            Some(PrivilegeDirective::Rootless { uid: 999, gid: 999 })
        );
    }

    // A no-op object store so a `DeployStore` can be built for the KV-only replica
    // state the endpoint resolver reads.
    use boatramp_core::{ByteStream, GetObject, ObjectMeta, PutMeta, Storage, StorageError};
    struct NullStorage;
    #[async_trait]
    impl Storage for NullStorage {
        async fn get(&self, _: &str) -> Result<GetObject, StorageError> {
            Err(StorageError::NotFound(String::new()))
        }
        async fn get_range(
            &self,
            _: &str,
            _: u64,
            _: Option<u64>,
        ) -> Result<GetObject, StorageError> {
            Err(StorageError::NotFound(String::new()))
        }
        async fn put(
            &self,
            _: &str,
            _: ByteStream,
            _: PutMeta,
        ) -> Result<ObjectMeta, StorageError> {
            Err(StorageError::unsupported("null"))
        }
        async fn head(&self, _: &str) -> Result<ObjectMeta, StorageError> {
            Err(StorageError::NotFound(String::new()))
        }
        async fn delete(&self, _: &str) -> Result<(), StorageError> {
            Ok(())
        }
        async fn list(&self, _: &str) -> Result<Vec<ObjectMeta>, StorageError> {
            Ok(Vec::new())
        }
    }

    fn replica(
        workload: &str,
        replica: u32,
        host: &str,
        port: u16,
        healthy: bool,
        phase: ReplicaPhase,
    ) -> boatramp_core::compute::ObservedInstance {
        use boatramp_core::compute::{Endpoint, InstanceHandle, Scheme};
        boatramp_core::compute::ObservedInstance {
            handle: InstanceHandle {
                project: "default".into(),
                workload: workload.into(),
                replica,
                backend_ref: String::new(),
            },
            node: 0,
            backend: "fake".into(),
            endpoint: Endpoint {
                scheme: Scheme::Http,
                host: host.into(),
                port,
            },
            region: None,
            healthy,
            started_at: None,
            phase,
            snapshot: None,
        }
    }

    #[tokio::test]
    async fn endpoint_resolver_returns_only_healthy_running_replicas() {
        let deploy = DeployStore::new(Arc::new(NullStorage), Arc::new(MemoryKv::new()));
        let p = ProjectRef::DEFAULT;
        // Two healthy running replicas, one unhealthy, one parked (Zero).
        deploy
            .set_replica_state(
                p,
                &replica("pg", 0, "10.0.0.1", 5432, true, ReplicaPhase::Running),
            )
            .await
            .unwrap();
        deploy
            .set_replica_state(
                p,
                &replica("pg", 1, "10.0.0.2", 5432, true, ReplicaPhase::Running),
            )
            .await
            .unwrap();
        deploy
            .set_replica_state(
                p,
                &replica("pg", 2, "10.0.0.3", 5432, false, ReplicaPhase::Running),
            )
            .await
            .unwrap();
        deploy
            .set_replica_state(
                p,
                &replica("pg", 3, "10.0.0.4", 5432, false, ReplicaPhase::Zero),
            )
            .await
            .unwrap();

        let resolver = DeployEndpointResolver::new(deploy, "default");
        let mut eps = resolver.endpoints("pg").await.unwrap();
        eps.sort();
        assert_eq!(
            eps,
            vec![
                ("10.0.0.1".to_string(), 5432),
                ("10.0.0.2".to_string(), 5432)
            ],
            "only the healthy running replicas, unhealthy + Zero filtered out"
        );
        // A workload with no replicas resolves to nothing (a clear no-endpoint state).
        assert!(resolver.endpoints("absent").await.unwrap().is_empty());
    }

    /// A `Shared` binding registers exactly ONE shared server (bare `<compute>` under
    /// the reserved default project), idempotently + non-clobbering. Per-tenant DDL is
    /// lazy, so no envelope is needed for Shared boot-warm.
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
    #[tokio::test]
    async fn auto_register_shared_registers_one_server_idempotently_and_never_clobbers() {
        use crate::config::TenantIsolation;
        use boatramp_core::compute::{ComputeWorkload, PlacementConstraints};

        let deploy = DeployStore::new(Arc::new(NullStorage), Arc::new(MemoryKv::new()));
        let p = ProjectRef::DEFAULT;

        let mut shared = db("postgres", Some("pg"), "", None);
        shared.tenant = TenantIsolation::Shared;

        let mut dbs = BTreeMap::new();
        dbs.insert("analytics".to_string(), shared);
        // BYO credential (compute-backed WITH password_env) → NOT managed, NOT registered.
        dbs.insert(
            "byo".to_string(),
            db("postgres", Some("byopg"), "", Some("PW")),
        );
        // BYO URL (not compute-backed) → NOT registered.
        dbs.insert("ext".to_string(), db("mysql", None, "MYSQL_URL", None));

        auto_register_managed_db_workloads(&deploy, &dbs).await;

        // The shared server workload was registered, desired 1 replica, spec stored.
        let wl = deploy
            .get_compute_workload(p, "pg")
            .await
            .unwrap()
            .expect("shared server workload `pg` auto-registered");
        assert_eq!(wl.replicas, 1);
        assert!(!wl.active.is_empty(), "an active spec hash was stored");
        assert!(
            deploy
                .get_compute_workload(p, "byopg")
                .await
                .unwrap()
                .is_none(),
            "a BYO-credential DB is not auto-registered"
        );

        // Idempotent: a second pass leaves the same active spec (no churn).
        auto_register_managed_db_workloads(&deploy, &dbs).await;
        let wl2 = deploy.get_compute_workload(p, "pg").await.unwrap().unwrap();
        assert_eq!(wl2.active, wl.active, "re-run is a no-op");

        // Non-clobbering: an operator-declared workload (apply / admin API) wins.
        let operator = ComputeWorkload {
            version: 1,
            name: "pg".to_string(),
            active: "operatorspec".to_string(),
            replicas: 3,
            placement: PlacementConstraints::default(),
        };
        deploy.set_compute_workload(p, &operator).await.unwrap();
        auto_register_managed_db_workloads(&deploy, &dbs).await;
        let after = deploy.get_compute_workload(p, "pg").await.unwrap().unwrap();
        assert_eq!(
            after.replicas, 3,
            "auto-register must not overwrite the operator's workload"
        );
        assert_eq!(after.active, "operatorspec");
    }

    /// Seed a `project` that exists (so `list_projects` returns it) and has a deployed
    /// `site` (so `list_sites` returns it — the "has resources" signal). The project
    /// pointer goes through `put_project`; the site's current-deployment pointer is
    /// written directly (`project/<proj>/current/<site>`, per `deploy::keys::current`) —
    /// exactly what `activate` leaves behind, without needing a real blob backend.
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
    async fn seed_project_with_site(
        deploy: &DeployStore,
        kv: &Arc<dyn KvStore>,
        project: &str,
        site: &str,
    ) {
        deploy
            .put_project(&boatramp_core::project::Project {
                version: 1,
                name: project.to_string(),
                created_at: 0,
                meta: Default::default(),
                config: Default::default(),
                secrets_ref: None,
            })
            .await
            .expect("seed the project pointer");
        let key = format!("project/{project}/current/{site}");
        kv.put(&key, b"deadbeef".to_vec())
            .await
            .expect("seed a current site deployment pointer");
    }

    /// Fix 1: a `Single` binding registers **nothing at boot** — even for a project with
    /// deployed resources. The per-tenant `pg-<ident>` is created durably by the lazy
    /// resolve (`provision_single`/`provision_tenant`) on the first `sql` use and
    /// relaunched by the reconcile; the old boot-warm enumerated projects by
    /// site/function (not `sql` use) and so over-warmed static-only projects into a
    /// spurious DB. There must be no bare `pg`/`default` and no `pg-<ident>` at boot.
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
    #[tokio::test]
    async fn auto_register_single_registers_nothing_at_boot() {
        use crate::config::{TenantIsolation, TenantScope};
        use crate::tenant_sql::tenant_key;
        use boatramp_storage::tenant_provision::sanitize_ident;

        let store_kv: Arc<dyn KvStore> = Arc::new(MemoryKv::new());
        let deploy = DeployStore::new(Arc::new(NullStorage), store_kv.clone());

        // A `construens` project that HAS a site — under the old boot-warm this would have
        // been enumerated + provisioned into a `pg-<ident>`. It must NOT be now.
        seed_project_with_site(&deploy, &store_kv, "construens", "app").await;

        let mut single = db("postgres", Some("pg"), "", None);
        single.tenant = TenantIsolation::Single; // (the default, made explicit)
        single.tenant_scope = TenantScope::Project;

        let mut dbs = BTreeMap::new();
        dbs.insert("main".to_string(), single);

        auto_register_managed_db_workloads(&deploy, &dbs).await;

        // No per-tenant `pg-<ident>` under the resourced project.
        let (raw, _is_default) = tenant_key(TenantScope::Project, "construens", "");
        let derived = format!("pg-{}", sanitize_ident(&raw));
        assert!(
            deploy
                .get_compute_workload(ProjectRef::new("construens"), &derived)
                .await
                .unwrap()
                .is_none(),
            "a Single binding must NOT boot-warm a per-tenant `pg-<ident>` (that is the lazy \
             resolve's job on first `sql` use)"
        );
        // No tenant-blind bare `pg`/`default` either.
        assert!(
            deploy
                .get_compute_workload(ProjectRef::DEFAULT, "pg")
                .await
                .unwrap()
                .is_none(),
            "a Single binding must NOT register a tenant-blind bare `pg`/`default`"
        );
        // Nothing anywhere — a Single binding is a complete no-op at boot.
        assert!(
            deploy
                .list_compute_workloads_all()
                .await
                .unwrap()
                .is_empty(),
            "a Single binding registers no managed workload at boot"
        );
    }

    /// Fix 1 (the over-warming bug proper): a `Single` binding + a project that owns ONLY
    /// a static site (never used `sql`) must register **no** managed workload at boot —
    /// no spurious `pg`/`pg-<ident>`. This is the exact construens repro (a static-only
    /// `default` was getting a bogus `pg`).
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
    #[tokio::test]
    async fn auto_register_single_static_only_project_gets_no_db_at_boot() {
        use crate::config::{TenantIsolation, TenantScope};

        let store_kv: Arc<dyn KvStore> = Arc::new(MemoryKv::new());
        let deploy = DeployStore::new(Arc::new(NullStorage), store_kv.clone());

        // The reserved `default` project owns only a static site — it has never resolved
        // `sql`, so no managed DB should ever be provisioned for it.
        seed_project_with_site(&deploy, &store_kv, "default", "www").await;

        let mut single = db("postgres", Some("pg"), "", None);
        single.tenant = TenantIsolation::Single;
        single.tenant_scope = TenantScope::Project;

        let mut dbs = BTreeMap::new();
        dbs.insert("main".to_string(), single);

        auto_register_managed_db_workloads(&deploy, &dbs).await;

        assert!(
            deploy
                .list_compute_workloads_all()
                .await
                .unwrap()
                .is_empty(),
            "a static-only project must not get a spurious managed `pg` at boot"
        );
    }

    /// Fix 2 (Bug 2, operator arm): for a `Single` project-scoped binding, operator
    /// `sql exec/query` must target the per-tenant workload `pg-<ident>` under the
    /// tenant's project and the per-tenant credential key — NOT the tenant-blind bare
    /// `pg`/`default`. Asserted on the pure `operator_target` derivation.
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
    #[test]
    fn operator_target_single_targets_per_tenant_workload_and_cred() {
        use crate::config::{TenantIsolation, TenantScope};
        use crate::tenant_sql::tenant_key;
        use boatramp_storage::tenant_provision::sanitize_ident;

        let mut single = db("postgres", Some("pg"), "", None);
        single.tenant = TenantIsolation::Single;
        single.tenant_scope = TenantScope::Project;
        single.database = Some("appdb".into());
        single.user = Some("app".into());

        // Non-default project tenant → derived per-tenant workload under its project.
        let (raw, is_default) = tenant_key(TenantScope::Project, "construens", "");
        assert!(!is_default);
        let ident = sanitize_ident(&raw);
        let derived = format!("pg-{ident}");

        let t = operator_target(&single, "construens", "main").unwrap();
        assert_eq!(
            t.workload, derived,
            "targets the per-tenant workload, not bare `pg`"
        );
        assert_eq!(t.database, "appdb");
        assert_eq!(t.user, "app");
        assert_eq!(
            t.endpoint_project, "construens",
            "a Single per-tenant workload's replicas live under its project"
        );
        // The credential key is the workload's OWN `(project, workload)` — matching
        // provision_single + the server-init env injector, so operator SQL unseals the
        // SAME password the container was initialized with.
        assert_eq!(t.cred_project, "construens");
        assert_eq!(t.cred_workload, derived);
        assert_ne!(
            t.cred_workload, "pg",
            "never the bare tenant-blind workload"
        );

        // The reserved default project keeps the plain names (single-tenant install).
        let d = operator_target(&single, "default", "main").unwrap();
        assert_eq!(d.workload, "pg");
        assert_eq!(d.cred_project, "default");
        assert_eq!(d.cred_workload, "pg");
        assert_eq!(d.endpoint_project, "default");
    }

    /// Fix 2: a `Shared` binding's operator SQL targets the shared server as the
    /// superuser (credential under the reserved default project + bare `<compute>`)
    /// against the tenant's per-tenant database.
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
    #[test]
    fn operator_target_shared_uses_superuser_cred_against_tenant_db() {
        use crate::config::{TenantIsolation, TenantScope};
        use crate::tenant_sql::tenant_key;
        use boatramp_core::project::DEFAULT_PROJECT;
        use boatramp_storage::tenant_provision::{sanitize_ident, tenant_db_name};

        let mut shared = db("postgres", Some("pg"), "", None);
        shared.tenant = TenantIsolation::Shared;
        shared.tenant_scope = TenantScope::Project;
        shared.database = Some("appdb".into());
        shared.user = Some("postgres".into());

        let (raw, _) = tenant_key(TenantScope::Project, "construens", "");
        let ident = sanitize_ident(&raw);

        let t = operator_target(&shared, "construens", "main").unwrap();
        // The shared server workload, the tenant's per-tenant database, superuser user.
        assert_eq!(t.workload, "pg");
        assert_eq!(t.database, tenant_db_name("appdb", &ident));
        assert_eq!(t.user, "postgres");
        // The superuser credential key — reserved default project + bare `<compute>`.
        assert_eq!(t.cred_project, DEFAULT_PROJECT);
        assert_eq!(t.cred_workload, "pg");
        assert_eq!(t.endpoint_project, DEFAULT_PROJECT);
    }

    /// Fix 2: a **site-scoped** managed DB has no single project-level database, so
    /// operator SQL fails with a clear error rather than hitting the wrong DB.
    #[cfg(any(feature = "sql-postgres", feature = "sql-mysql"))]
    #[test]
    fn operator_target_site_scoped_errors_clearly() {
        use crate::config::{TenantIsolation, TenantScope};

        let mut site = db("postgres", Some("pg"), "", None);
        site.tenant = TenantIsolation::Single;
        site.tenant_scope = TenantScope::Site;

        let err = operator_target(&site, "construens", "main").unwrap_err();
        let msg = err.to_string();
        assert!(
            msg.contains("site-scoped"),
            "the error explains a site-scoped DB needs a site: {msg}"
        );
    }
}