rustango 0.61.13

//! Standing up a tenant: the steps, in order, with no terminal in
//! them.
//!
//! The `create-tenant` verb keeps only what is a CLI's job: turning
//! `argv` into a [`ProvisionRequest`], and printing
//! [`ProvisionEvent`]s. An HTTP handler, a webhook or a job can drive
//! the same sequence.
//!
//! ## The steps
//!
//! 1. [`Validate`][step] — slug, mode and backend agree with each
//!    other and with this build.
//! 2. [`CheckConnection`][step] — reach the tenant's database and
//!    prove this role can create tables there, **before** anything is
//!    written. See [`preflight`](crate::tenancy::preflight). Schema
//!    mode skips it; those tenants live in the registry's own
//!    database.
//! 3. [`ProvisionStorage`][step] — `CREATE SCHEMA` for schema mode,
//!    refusing one that exists. A failed `INSERT` drops it again.
//! 4. [`RegisterOrg`][step] — the `rustango_orgs` row, written
//!    **inactive**.
//! 5. [`Migrate`][step] — this tenant's schema, via
//!    [`tenant_migrate::migrate_one_tenant`].
//! 6. [`CreateAdmin`][step] — the requested first admin, if any. See
//!    [`first_admin`](crate::tenancy::first_admin).
//! 7. [`Activate`][step] — set `active`, and the tenant starts
//!    resolving.
//!
//! ## What a failed run leaves behind
//!
//! The row is written inactive and activated last, so a
//! half-provisioned tenant serves nothing: the resolver filters on
//! `active`. No tenant ever resolves to a database with no schema.
//!
//! A run that fails at migrate leaves an inactive `Org` row and
//! whatever schema was created. The row stays on purpose: an operator
//! needs to see what was half-made, and rolling back is not always
//! possible once a schema exists.
//!
//! So `active = false` now means either "suspended" or "never
//! finished". The resolver treats both the same way. To tell them
//! apart, read [`provision_store`](crate::tenancy::provision_store).
//!
//! ## On a saved server
//!
//! [`DatabaseSource::Server`] makes storage three steps: create the
//! database named as the slug, create its own login, grant it that
//! database. The connection check probes the server's admin rights first.
//! The database is recorded in [`TenantDatabase`](crate::tenancy::TenantDatabase).
//! Any later failure drops what the run made and the row, so nothing is
//! left to resume. If that drop fails, the record and inactive row stay.
//!
//! [`ProvisionRequest`]: crate::tenancy::provision::ProvisionRequest
//! [`ProvisionEvent`]: crate::tenancy::provision::ProvisionEvent
//! [step]: crate::tenancy::provision::ProvisionStep
//! [`tenant_migrate::migrate_one_tenant`]: crate::tenancy::migrate::migrate_one_tenant

use std::path::Path;

use sqlx::Database;

use crate::core::Column as _;
use crate::migrate::file::Migration;
use crate::sql::{Auto, FetcherPool};

use super::error::TenancyError;
use super::migrate as tenant_migrate;
use super::org::{BackendKind, Org, StorageMode};
use super::pools::TenantPools;
use super::preflight;
use super::provision_store::RunText;
use super::secrets::{SecretsResolver, TenantSecrets};

/// Everything needed to stand up one tenant.
///
/// The CLI's `--no-migrate` flag is
/// [`run_migrations`](Self::run_migrations) here: a negative flag reads
/// well on a command line and badly on a field.
///
/// `Debug` is safe: [`DatabaseSource`] masks the URL's password.
#[derive(Debug, Clone)]
pub struct ProvisionRequest {
    /// Globally unique. Also the default schema name, display name, and
    /// subdomain label.
    pub slug: String,
    pub storage: TenantStorage,
    /// `None` takes the registry's. Anything else must equal it: tenant
    /// pools open with the registry's driver.
    pub backend: Option<BackendKind>,
    /// Defaults to the slug.
    pub display_name: Option<String>,
    /// Defaults to `<slug>.<RUSTANGO_APEX_DOMAIN>` when that env var is
    /// set and no `path_prefix` is given, and to nothing otherwise.
    pub host_pattern: Option<String>,
    pub port: Option<i32>,
    pub path_prefix: Option<String>,
    /// Run the tenant's migrations once the row is in place.
    pub run_migrations: bool,
    /// How hard to check the target database before writing anything.
    /// Defaults to a full check including the write probe — see
    /// [`preflight`] for why a bare `SELECT 1` is not enough.
    pub preflight: preflight::Preflight,
    /// A superuser to create after the migrations, before the tenant goes live.
    pub first_admin: Option<super::first_admin::FirstAdmin>,
}

/// Where a tenant's tables live.
#[derive(Debug, Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum TenantStorage {
    /// A schema in the registry's database, named by the slug when `None`.
    Schema(Option<String>),
    /// A database of its own.
    Database(DatabaseSource),
}

/// Where a database-mode tenant's database comes from.
#[derive(Clone, PartialEq, Eq)]
#[non_exhaustive]
pub enum DatabaseSource {
    /// One that exists: a connection URL or a secret reference.
    Url(String),
    /// Create one named exactly as the slug, with its own login, on the
    /// saved [`DbServer`](super::DbServer) of this name.
    Server(String),
}

impl std::fmt::Debug for DatabaseSource {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Url(u) => f
                .debug_tuple("Url")
                .field(&crate::sql::connect_diagnosis::redact(u))
                .finish(),
            Self::Server(s) => f.debug_tuple("Server").field(s).finish(),
        }
    }
}

impl TenantStorage {
    #[must_use]
    pub fn mode(&self) -> StorageMode {
        match self {
            Self::Schema(_) => StorageMode::Schema,
            Self::Database(_) => StorageMode::Database,
        }
    }
}

impl ProvisionRequest {
    /// A request with everything but the slug and storage defaulted.
    #[must_use]
    pub fn new(slug: impl Into<String>, storage: TenantStorage) -> Self {
        Self {
            slug: slug.into(),
            storage,
            backend: None,
            display_name: None,
            host_pattern: None,
            port: None,
            path_prefix: None,
            run_migrations: true,
            preflight: preflight::Preflight::default(),
            first_admin: None,
        }
    }

    /// A tenant in a database that exists.
    #[must_use]
    pub fn database(slug: impl Into<String>, database_url: impl Into<String>) -> Self {
        Self::new(
            slug,
            TenantStorage::Database(DatabaseSource::Url(database_url.into())),
        )
    }

    /// A tenant in a new database on the saved server `server`.
    #[must_use]
    pub fn on_server(slug: impl Into<String>, server: impl Into<String>) -> Self {
        Self::new(
            slug,
            TenantStorage::Database(DatabaseSource::Server(server.into())),
        )
    }

    #[must_use]
    pub fn mode(&self) -> StorageMode {
        self.storage.mode()
    }

    /// The URL or reference of an existing database, if that is the source.
    #[must_use]
    pub fn database_url(&self) -> Option<&str> {
        match &self.storage {
            TenantStorage::Database(DatabaseSource::Url(u)) => Some(u),
            _ => None,
        }
    }

    /// The saved server to create the database on, if that is the source.
    #[must_use]
    pub fn server(&self) -> Option<&str> {
        match &self.storage {
            TenantStorage::Database(DatabaseSource::Server(s)) => Some(s),
            _ => None,
        }
    }

    /// [`Self::backend`], or `registry`'s when unset.
    #[must_use]
    pub fn backend_on(&self, registry: &crate::sql::Pool) -> BackendKind {
        self.backend
            .unwrap_or_else(|| BackendKind::of_pool(registry))
    }
}

/// One stage of provisioning, reportable on its own.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum ProvisionStep {
    /// Slug is free; mode, backend and build agree.
    Validate,
    /// Reach the target database, and prove this role can create
    /// tables in it, before anything is written.
    CheckConnection,
    /// `CREATE SCHEMA` for schema-mode. Nothing to do for an existing
    /// database.
    ProvisionStorage,
    /// Create the tenant's database on a saved server.
    CreateDatabase,
    /// Create its own login, with a random password.
    CreateLogin,
    /// Grant that login its database, and nothing else.
    Grant,
    /// Sign in as the new login, before anything is registered.
    CheckLogin,
    /// Insert the `rustango_orgs` row — **inactive**, so nothing routes
    /// to the tenant until its schema is in place.
    RegisterOrg,
    /// Apply the tenant's migrations.
    Migrate,
    /// Create the requested first admin, a superuser with no usable
    /// password until they set one.
    CreateAdmin,
    /// Set `active`, after which the tenant resolves. Last on purpose,
    /// so a half-provisioned tenant never serves a request.
    Activate,
    /// Undo a failed run on a saved server: drop the database and login
    /// it made, and the tenant row.
    Rollback,
}

/// How a step went.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum StepStatus {
    Started,
    Ok,
    /// Nothing to do, and why. Separate from `Ok` so a console can grey
    /// the step out instead of claiming work that never happened.
    Skipped(String),
    /// The run stops here.
    Failed(String),
}

/// Something that happened while provisioning.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ProvisionEvent {
    Step {
        step: ProvisionStep,
        status: StepStatus,
    },
    /// The row landed and the tenant now resolves. Separate from
    /// `Step { RegisterOrg, Ok }` because it carries the id, which is
    /// the one thing a caller cannot compute for itself.
    Registered { org_id: i64 },
    /// A migration event from the tenant's own run, passed straight
    /// through, so a watcher sees what `manage migrate` prints.
    Migration(tenant_migrate::TenantMigrationEvent),
}

/// Receives [`ProvisionEvent`]s as a run progresses.
///
/// Implemented for any `Fn(ProvisionEvent)`, so a closure works.
/// **Must not block**: migration events reach it while the tenant's
/// migrate lock is held. See [`crate::migrate::progress`].
pub trait ProvisionObserver: Send + Sync {
    /// Handle one event. Must return promptly and must not panic.
    fn on_event(&self, event: ProvisionEvent);
}

impl<F> ProvisionObserver for F
where
    F: Fn(ProvisionEvent) + Send + Sync,
{
    fn on_event(&self, event: ProvisionEvent) {
        self(event);
    }
}

/// What the tenant's migration step did.
#[derive(Debug, Clone)]
pub enum MigrationsOutcome {
    /// The caller asked for no migrations.
    Skipped,
    /// The batch ran but reported nothing for this tenant, usually a
    /// migrations directory with nothing tenant-scoped in it.
    NotMatched,
    Applied(Vec<Migration>),
    /// The tenant's migrations failed. Not an `Err`, because the org
    /// row exists and the caller must hear about it.
    Failed(String),
}

/// The result of a completed provisioning run.
#[derive(Debug, Clone)]
pub struct ProvisionOutcome {
    pub org_id: i64,
    pub slug: String,
    pub mode: StorageMode,
    pub migrations: MigrationsOutcome,
    /// The first admin, when the request asked for one.
    pub first_admin: Option<super::first_admin::CreatedAdmin>,
}

/// Where a run reports: the caller's observer, and optionally the
/// durable store.
///
/// Step transitions are written as they happen, with no lock held, so
/// another pod can follow a run in progress.
///
/// Migration events are buffered instead. They arrive from inside the
/// migrate lock (see [`crate::migrate::progress`]), and awaiting a
/// registry write there would hold that lock while other pods queue
/// behind it. They are flushed when the migrate step ends.
///
/// So a watcher sees `Migrate: started`, a pause, then the whole
/// per-migration log at once.
pub(super) struct Reporter<'a> {
    observer: Option<&'a dyn ProvisionObserver>,
    store: Option<RunStore<'a>>,
    /// Pick up a tenant a failed run left inactive, instead of refusing
    /// its slug (#1883).
    resume: bool,
}

struct RunStore<'a> {
    registry: &'a crate::sql::Pool,
    run_id: i64,
    /// Names the run in the log line of a withheld failure.
    slug: String,
    log: super::provision_store::RunLog,
    /// The operator-safe text of the last failure, for the run's `error`.
    failure: std::sync::Mutex<Option<RunText>>,
    /// Next `seq`. Dense and per-run, because `Last-Event-ID` counts
    /// from it.
    seq: std::sync::atomic::AtomicI64,
    buffered: std::sync::Mutex<Vec<(&'static str, &'static str, RunText)>>,
}

/// A step transition the store may record. No `Failed`: a failure goes
/// through [`Reporter::fail`], which withholds its cause (#2212).
enum Progress {
    Started,
    Ok,
    /// `Ok`, with what was done.
    Done(RunText),
    Skipped(&'static str),
}

impl Progress {
    fn status(&self) -> StepStatus {
        match self {
            Self::Started => StepStatus::Started,
            Self::Ok | Self::Done(_) => StepStatus::Ok,
            Self::Skipped(why) => StepStatus::Skipped((*why).to_owned()),
        }
    }
}

impl<'a> Reporter<'a> {
    pub(super) fn new(observer: Option<&'a dyn ProvisionObserver>) -> Self {
        Self {
            observer,
            store: None,
            resume: false,
        }
    }

    /// Also persist to `run_id` in the registry.
    pub(super) fn persisting(
        mut self,
        registry: &'a crate::sql::Pool,
        run_id: i64,
        slug: &str,
    ) -> Self {
        self.store = Some(RunStore {
            registry,
            run_id,
            slug: slug.to_owned(),
            log: super::provision_store::RunLog::new(run_id),
            failure: std::sync::Mutex::new(None),
            seq: std::sync::atomic::AtomicI64::new(1),
            buffered: std::sync::Mutex::new(Vec::new()),
        });
        self
    }

    /// Link the new Org to this run as soon as it exists, so a retry can
    /// tell which run made it.
    async fn registered(&self, org_id: i64) {
        let Some(store) = &self.store else {
            return;
        };
        let attach = || super::provision_store::attach_org(store.registry, store.run_id, org_id);
        // One retry: this is the run's only org link (#2061).
        if let Err(e) = match attach().await {
            Err(_) => attach().await,
            ok => ok,
        } {
            tracing::warn!(target: "rustango::tenancy::provision", error = %e, "could not attach org id to run");
        }
    }

    fn notify(&self, event: ProvisionEvent) {
        if let Some(observer) = self.observer {
            observer.on_event(event);
        }
    }

    /// Write one row now. Failures are logged, never propagated: the
    /// record of a provisioning run must not be what fails it.
    async fn write(&self, step: &str, status: &str, message: &RunText) {
        let Some(store) = &self.store else {
            return;
        };
        let seq = store.seq.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        if let Err(e) = super::provision_store::append_event_text(
            store.registry,
            store.run_id,
            seq,
            step,
            status,
            message,
        )
        .await
        {
            tracing::warn!(
                target: "rustango::tenancy::provision",
                run_id = store.run_id,
                error = %e,
                "could not record a provisioning event; the run continues"
            );
        }
    }

    /// Queue a migration event for the next [`flush`](Self::flush).
    fn buffer(&self, step: &'static str, status: &'static str, message: RunText) {
        let Some(store) = &self.store else {
            return;
        };
        if let Ok(mut buf) = store.buffered.lock() {
            buf.push((step, status, message));
        }
    }

    /// Write everything buffered, in order.
    async fn flush(&self) {
        let Some(store) = &self.store else {
            return;
        };
        let pending = match store.buffered.lock() {
            Ok(mut buf) => std::mem::take(&mut *buf),
            Err(_) => return,
        };
        for (step, status, message) in pending {
            self.write(step, status, &message).await;
        }
    }
}

impl Reporter<'_> {
    /// Announce a step transition — to the observer, and to the store.
    async fn step(&self, step: ProvisionStep, progress: Progress) {
        let status = progress.status();
        let text = match progress {
            Progress::Skipped(why) => RunText::fixed(why),
            Progress::Done(text) => text,
            Progress::Started | Progress::Ok => RunText::default(),
        };
        self.write(step.as_str(), status.as_str(), &text).await;
        self.notify(ProvisionEvent::Step { step, status });
    }

    /// Report a step's failure, then hand back the error.
    ///
    /// Every failure path goes through this rather than a bare `?`, so
    /// a watcher never sees a step stuck on `Started` with no
    /// explanation — which is the state the whole event stream exists
    /// to prevent.
    async fn fail<T>(&self, at: ProvisionStep, e: TenancyError) -> Result<T, TenancyError> {
        // Refusals too, so the server log always holds the reason.
        tracing::warn!(
            target: "rustango::tenancy::provision",
            step = at.as_str(),
            reason = %crate::sql::connect_diagnosis::redact(&e.to_string()).replace('\n', " "),
            "a provisioning step failed"
        );
        let text = self
            .store
            .as_ref()
            .map(|store| store.log.failure(&store.slug, "Step failed", &e));
        self.step_failed(at, e.to_string(), text).await;
        Err(e)
    }

    /// A failed migrate step: the cause was a migration, already logged once.
    async fn migrate_failed(&self, cause: &str) {
        let text = self
            .store
            .as_ref()
            .map(|store| store.log.withheld(&store.slug, "Step failed", &cause));
        self.step_failed(ProvisionStep::Migrate, cause.to_owned(), text)
            .await;
    }

    /// The observer gets `cause`; the store gets `text` (#2198).
    async fn step_failed(&self, step: ProvisionStep, cause: String, text: Option<RunText>) {
        if let (Some(store), Some(text)) = (&self.store, text) {
            self.write(step.as_str(), "failed", &text).await;
            if let Ok(mut last) = store.failure.lock() {
                *last = Some(text);
            }
        }
        self.notify(ProvisionEvent::Step {
            step,
            status: StepStatus::Failed(cause),
        });
    }

    /// A failure whose text is operator-safe already.
    async fn failed_with(&self, step: ProvisionStep, text: RunText) {
        let stored = self.store.as_ref().map(|_| text.clone());
        self.step_failed(step, text.to_string(), stored).await;
    }

    /// The run's stored `error`: the last failed step's text, else `e`'s.
    fn run_error(&self, e: Option<&TenancyError>) -> Option<RunText> {
        let store = self.store.as_ref()?;
        let last = store.failure.lock().ok().and_then(|f| f.clone());
        match e {
            Some(e) => Some(
                RunText::user_facing(e)
                    .or(last)
                    .unwrap_or_else(|| store.log.failure(&store.slug, "Provisioning failed", e)),
            ),
            None => last,
        }
    }

    /// A migration event from the tenant's own run. Buffered rather
    /// than written — see the type docs.
    fn migration(&self, event: tenant_migrate::TenantMigrationEvent) {
        if let Some(store) = &self.store {
            let (step, status, text) = store.log.migration(&event);
            self.buffer(step, status, text);
        }
        self.notify(ProvisionEvent::Migration(event));
    }
}

impl ProvisionStep {
    /// Stable wire name, stored in `rustango_provisioning_events.step`
    /// and matched on by a console. Written out rather than derived
    /// from `Debug`, which is not a stable format.
    #[must_use]
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Validate => "validate",
            Self::CheckConnection => "check_connection",
            Self::ProvisionStorage => "provision_storage",
            Self::CreateDatabase => "create_database",
            Self::CreateLogin => "create_login",
            Self::Grant => "grant",
            Self::CheckLogin => "check_login",
            Self::RegisterOrg => "register_org",
            Self::Migrate => "migrate",
            Self::CreateAdmin => "create_admin",
            Self::Activate => "activate",
            Self::Rollback => "rollback",
        }
    }
}

impl StepStatus {
    /// Stable wire name.
    #[must_use]
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::Started => "started",
            Self::Ok => "ok",
            Self::Skipped(_) => "skipped",
            Self::Failed(_) => "failed",
        }
    }

    /// The reason carried by `Skipped` / `Failed`, or empty.
    #[must_use]
    pub fn detail(&self) -> &str {
        match self {
            Self::Started | Self::Ok => "",
            Self::Skipped(why) | Self::Failed(why) => why,
        }
    }
}

/// Stand up a tenant: validate, provision its storage, register it, and
/// migrate it.
///
/// The CLI verb is a thin wrapper over this, as is anything else that
/// creates a tenant.
///
/// # Errors
/// `Err` if the slug is taken, the request contradicts itself (database
/// mode with no URL, schema mode on a non-PG build), the schema could
/// not be created, or the row could not be inserted.
///
/// A migration failure is not an error: the tenant already exists, so
/// it comes back as [`MigrationsOutcome::Failed`] in
/// [`ProvisionOutcome::migrations`].
pub async fn provision_tenant<DB: Database>(
    pools: &TenantPools<DB>,
    registry_url: &str,
    dir: &Path,
    request: &ProvisionRequest,
    observer: Option<&dyn ProvisionObserver>,
) -> Result<ProvisionOutcome, TenancyError>
where
    crate::sql::Pool: From<sqlx::Pool<DB>>,
{
    let rep = Reporter::new(observer);
    provision_reported(pools, registry_url, dir, request, &rep).await
}

/// [`provision_tenant`], with the run persisted to
/// [`super::provision_store`] as it goes.
///
/// Step transitions are written as they happen, so another pod can
/// read a run, or replay it from the start.
///
/// # Errors
/// As [`provision_tenant`]. A failure to *record* the run is only
/// logged: bookkeeping must not fail a tenant creation.
pub async fn provision_tenant_recorded<DB: Database>(
    pools: &TenantPools<DB>,
    registry_url: &str,
    dir: &Path,
    request: &ProvisionRequest,
    observer: Option<&dyn ProvisionObserver>,
    requested_by: Option<&str>,
    idempotency_key: Option<&str>,
) -> Result<(super::provision_store::ProvisioningRun, ProvisionOutcome), TenancyError>
where
    crate::sql::Pool: From<sqlx::Pool<DB>>,
{
    let by = RunBy {
        requested_by,
        idempotency_key,
    };
    recorded(pools, registry_url, dir, request, observer, by, false).await
}

/// [`provision_tenant_recorded`] that, with `resume`, finishes a tenant
/// an earlier failed run left inactive, as `create-tenant --resume` does.
pub(crate) async fn provision_tenant_recorded_or_resume<DB: Database>(
    pools: &TenantPools<DB>,
    registry_url: &str,
    dir: &Path,
    request: &ProvisionRequest,
    observer: Option<&dyn ProvisionObserver>,
    requested_by: Option<&str>,
    resume: bool,
) -> Result<(super::provision_store::ProvisioningRun, ProvisionOutcome), TenancyError>
where
    crate::sql::Pool: From<sqlx::Pool<DB>>,
{
    let by = RunBy {
        requested_by,
        idempotency_key: None,
    };
    recorded(pools, registry_url, dir, request, observer, by, resume).await
}

/// Who asked for a recorded run.
struct RunBy<'a> {
    requested_by: Option<&'a str>,
    idempotency_key: Option<&'a str>,
}

async fn recorded<DB: Database>(
    pools: &TenantPools<DB>,
    registry_url: &str,
    dir: &Path,
    request: &ProvisionRequest,
    observer: Option<&dyn ProvisionObserver>,
    by: RunBy<'_>,
    resume: bool,
) -> Result<(super::provision_store::ProvisioningRun, ProvisionOutcome), TenancyError>
where
    crate::sql::Pool: From<sqlx::Pool<DB>>,
{
    use super::provision_store as store;
    let RunBy {
        requested_by,
        idempotency_key,
    } = by;

    let registry = pools.registry_pool();
    // A taken slug leaves no failed run behind. A lookup error falls
    // through, so Validate records it; Validate also checks again.
    // Resuming, an inactive row goes on to Validate, which decides.
    let taken = Org::objects()
        .where_(Org::slug.eq(request.slug.clone()))
        .first(&registry)
        .await;
    if matches!(&taken, Ok(Some(org)) if !resume || org.active) {
        return Err(TenancyError::Validation(format!(
            "tenant slug `{}` already exists",
            request.slug
        )));
    }
    let run = store::open_run(
        &registry,
        &request.slug,
        request.mode().as_str(),
        request.backend_on(&registry).as_str(),
        request.database_url(),
        requested_by,
        idempotency_key,
    )
    .await?;
    let run_id = run.id.get().copied().unwrap_or_default();

    let outcome = in_run(pools, registry_url, dir, request, observer, run_id, resume)
        .await
        // A pre-row failure still closed the run inside
        // `provision_tenant_in_run`; propagate the reason unchanged.
        ?;
    // Re-read so the caller sees the closed run, not the one that was
    // handed back at `open_run` time.
    let refreshed = store::run_by_id(&registry, run_id).await?.unwrap_or(run);
    Ok((refreshed, outcome))
}

/// [`provision_tenant_recorded`] against a run somebody else already
/// opened.
///
/// The inbound webhook needs this. It opens the run first, so the
/// response can carry the id and any `idempotency_key` conflict, then
/// hands the slow part to a task. Without this the task would open a
/// second run for the same tenant.
///
/// A retry resumes: if the slug's Org was made by a provision run that
/// failed and is still inactive, it is migrated and activated instead of
/// refused as taken (#1883).
///
/// # Errors
/// As [`provision_tenant`]. The run is closed either way.
pub async fn provision_tenant_in_run<DB: Database>(
    pools: &TenantPools<DB>,
    registry_url: &str,
    dir: &Path,
    request: &ProvisionRequest,
    observer: Option<&dyn ProvisionObserver>,
    run_id: i64,
) -> Result<ProvisionOutcome, TenancyError>
where
    crate::sql::Pool: From<sqlx::Pool<DB>>,
{
    in_run(pools, registry_url, dir, request, observer, run_id, true).await
}

async fn in_run<DB: Database>(
    pools: &TenantPools<DB>,
    registry_url: &str,
    dir: &Path,
    request: &ProvisionRequest,
    observer: Option<&dyn ProvisionObserver>,
    run_id: i64,
    resume: bool,
) -> Result<ProvisionOutcome, TenancyError>
where
    crate::sql::Pool: From<sqlx::Pool<DB>>,
{
    use super::provision_store::{self as store, RunState};

    let registry = pools.registry_pool();
    let mut rep = Reporter::new(observer).persisting(&registry, run_id, &request.slug);
    rep.resume = resume;
    let result = provision_reported(pools, registry_url, dir, request, &rep).await;

    // Close the run whatever happened, error paths included. A run
    // stuck at `running` is the ambiguity this table removes.
    // The stored error is operator-safe; the failed step logged the cause (#2198).
    let (state, error) = match &result {
        Ok(outcome) => match &outcome.migrations {
            MigrationsOutcome::Failed(_) => (RunState::Failed, rep.run_error(None)),
            _ => (RunState::Succeeded, None),
        },
        Err(e) => (RunState::Failed, rep.run_error(Some(e))),
    };
    // No `attach_org` here: `Reporter::registered` linked the org already.
    if let Err(e) = store::finish_run_text(&registry, run_id, state, error.as_ref()).await {
        tracing::warn!(target: "rustango::tenancy::provision", error = %e, "could not close provisioning run");
    }

    result
}

async fn provision_reported<DB: Database>(
    pools: &TenantPools<DB>,
    registry_url: &str,
    dir: &Path,
    request: &ProvisionRequest,
    rep: &Reporter<'_>,
) -> Result<ProvisionOutcome, TenancyError>
where
    crate::sql::Pool: From<sqlx::Pool<DB>>,
{
    // ---- 1. Validate ----
    rep.step(ProvisionStep::Validate, Progress::Started).await;
    let registry = pools.registry_pool();

    // Reject a duplicate slug up front — it saves a partial-state mess
    // where `CREATE SCHEMA` succeeds and the `INSERT` then fails.
    let existing: Vec<Org> = match Org::objects()
        .where_(Org::slug.eq(request.slug.clone()))
        .fetch(&registry)
        .await
    {
        Ok(rows) => rows,
        Err(e) => return rep.fail(ProvisionStep::Validate, e.into()).await,
    };
    if let Some(org) = existing.into_iter().next() {
        let resumable = rep.resume
            && !org.active
            && org.storage_mode == request.mode().as_str()
            && match super::provision_store::org_left_by_failed_run(
                &registry,
                org.id.get().copied().unwrap_or_default(),
            )
            .await
            {
                Ok(left) => left,
                Err(e) => return rep.fail(ProvisionStep::Validate, e).await,
            };
        if !resumable {
            return rep
                .fail(
                    ProvisionStep::Validate,
                    TenancyError::Validation(format!(
                        "tenant slug `{}` already exists",
                        request.slug
                    )),
                )
                .await;
        }
        let warned = resumed_on_registry(pools, registry_url, &org).await;
        let validated = warned.map_or(Progress::Ok, |w| Progress::Done(RunText::fixed(w)));
        rep.step(ProvisionStep::Validate, validated).await;
        rep.step(
            ProvisionStep::RegisterOrg,
            Progress::Skipped("resuming the tenant an earlier run left inactive"),
        )
        .await;
        let org_id = org.id.get().copied().unwrap_or_default();
        rep.registered(org_id).await;
        return migrate_and_activate(pools, registry_url, dir, request, &org, rep).await;
    }

    let checked = match checked_request(&registry, pools.secrets(), registry_url, request).await {
        Ok(r) => r,
        Err(e) => return rep.fail(ProvisionStep::Validate, e).await,
    };
    let request = &checked.request;

    rep.step(ProvisionStep::Validate, Progress::Ok).await;

    if let Some(server) = checked.server() {
        return provision_on_server(pools, registry_url, dir, &checked, server, rep).await;
    }

    // ---- 2. Check the connection ----
    //
    // Before anything is written. A database-mode URL that is wrong —
    // typo'd host, wrong password, database not created yet — used to
    // be discovered *after* the `Org` row landed, leaving a tenant the
    // resolver matches in front of a database with no schema.
    check_connection(&checked, &registry, rep).await?;

    let schema_name = schema_name_for(request);

    // ---- 3. Provision storage ----
    //
    // Before the row, not after: a failed `INSERT` must not leave an
    // orphan schema behind. An existing schema is refused (#2394).
    provision_storage(pools, schema_name.as_deref(), rep).await?;

    // ---- 4. Register the org ----
    rep.step(ProvisionStep::RegisterOrg, Progress::Started)
        .await;
    let mut org = new_org_row(&checked, schema_name, checked.stored_database_url());
    if let Err(e) = super::org_host::insert_org(&registry, &mut org).await {
        if let Some(schema) = org.schema_name.as_deref() {
            release_schema(pools, schema).await;
        }
        return rep.fail(ProvisionStep::RegisterOrg, e).await;
    }
    registered(&org, rep).await;

    migrate_and_activate(pools, registry_url, dir, request, &org, rep).await
}

/// Announce a new `Org` row.
async fn registered(org: &Org, rep: &Reporter<'_>) {
    // This pod sees the new tenant immediately; others converge on the
    // registry fingerprint (see `resolver::sync_org_generation`).
    super::invalidate_org_cache();
    let org_id = org.id.get().copied().unwrap_or_default();
    rep.registered(org_id).await;
    rep.step(ProvisionStep::RegisterOrg, Progress::Ok).await;
    rep.notify(ProvisionEvent::Registered { org_id });
}

/// Steps 2–6 when the database is created on a saved server. Any failure
/// after the create drops what this run made, row included.
async fn provision_on_server<DB: Database>(
    pools: &TenantPools<DB>,
    registry_url: &str,
    dir: &Path,
    checked: &CheckedRequest,
    server: &super::DbServer,
    rep: &Reporter<'_>,
) -> Result<ProvisionOutcome, TenancyError>
where
    crate::sql::Pool: From<sqlx::Pool<DB>>,
{
    let request = &checked.request;
    let registry = pools.registry_pool();

    rep.step(ProvisionStep::CheckConnection, Progress::Started)
        .await;
    let admin = match ServerDatabase::open(server, &registry).await {
        Ok((admin, warning)) => {
            let done = warning.map_or(Progress::Ok, |w| Progress::Done(RunText::fixed(w)));
            rep.step(ProvisionStep::CheckConnection, done).await;
            admin
        }
        Err(e) => return rep.fail(ProvisionStep::CheckConnection, e).await,
    };

    let mut db = ServerDatabase::create(admin, server, &request.slug, &registry, rep).await?;
    if let Err(e) = db.check_login(rep).await {
        let undone = db.roll_back(pools, None, rep).await;
        return Err(with_undone(e, &undone));
    }

    rep.step(ProvisionStep::RegisterOrg, Progress::Started)
        .await;
    let stored = match db
        .stored_url(&registry, pools.secrets(), registry_url)
        .await
    {
        Ok(s) => s,
        Err(e) => {
            let failed: Result<ProvisionOutcome, _> = rep.fail(ProvisionStep::RegisterOrg, e).await;
            let undone = db.roll_back(pools, None, rep).await;
            return failed.map_err(|e| with_undone(e, &undone));
        }
    };
    let mut org = new_org_row(checked, None, Some(stored));
    if let Err(e) = super::org_host::insert_org(&registry, &mut org).await {
        let failed: Result<ProvisionOutcome, _> = rep.fail(ProvisionStep::RegisterOrg, e).await;
        let undone = db.roll_back(pools, None, rep).await;
        return failed.map_err(|e| with_undone(e, &undone));
    }
    db.attach(&registry, org.id.get().copied().unwrap_or_default())
        .await;
    registered(&org, rep).await;

    match migrate_and_activate(pools, registry_url, dir, request, &org, rep).await {
        Ok(outcome) if !matches!(outcome.migrations, MigrationsOutcome::Failed(_)) => {
            db.keep().await;
            Ok(outcome)
        }
        // The tenant is gone again, so this is an error, not an outcome.
        Ok(_) => {
            let undone = db.roll_back(pools, Some(&org), rep).await;
            Err(TenancyError::Validation(format!(
                "the tenant's migrations failed; {}. The cause is in the run log",
                undone.describe()
            )))
        }
        Err(e) => {
            let undone = db.roll_back(pools, Some(&org), rep).await;
            Err(with_undone(e, &undone))
        }
    }
}

/// Steps 5 to 7, shared by a fresh tenant and a resumed one.
async fn migrate_and_activate<DB: Database>(
    pools: &TenantPools<DB>,
    registry_url: &str,
    dir: &Path,
    request: &ProvisionRequest,
    org: &Org,
    rep: &Reporter<'_>,
) -> Result<ProvisionOutcome, TenancyError>
where
    crate::sql::Pool: From<sqlx::Pool<DB>>,
{
    let registry = pools.registry_pool();
    let org_id = org.id.get().copied().unwrap_or_default();

    // ---- 5. Migrate ----
    let migrations = if request.run_migrations {
        rep.step(ProvisionStep::Migrate, Progress::Started).await;
        let outcome = migrate_new_tenant(pools, registry_url, dir, org, rep).await;
        // The migration events arrived synchronously while the migrate
        // lock was held, so they were buffered. The lock is released by
        // now — write them before reporting the step's own outcome, so
        // the stored log reads in the order things happened.
        rep.flush().await;
        match &outcome {
            MigrationsOutcome::Failed(e) => rep.migrate_failed(e).await,
            _ => rep.step(ProvisionStep::Migrate, Progress::Ok).await,
        }
        outcome
    } else {
        rep.step(
            ProvisionStep::Migrate,
            Progress::Skipped("caller asked for no migrations"),
        )
        .await;
        MigrationsOutcome::Skipped
    };

    // ---- 6. First admin ----
    //
    // Before activation, so a failure here rolls back like any other
    // step and a live tenant always has the admin it was asked for.
    let first_admin = match &request.first_admin {
        None => None,
        Some(_) if matches!(migrations, MigrationsOutcome::Failed(_)) => {
            rep.step(
                ProvisionStep::CreateAdmin,
                Progress::Skipped("migrations failed"),
            )
            .await;
            None
        }
        Some(admin) => {
            rep.step(ProvisionStep::CreateAdmin, Progress::Started)
                .await;
            match super::first_admin::create(pools, org, admin).await {
                Ok(created) => {
                    rep.step(ProvisionStep::CreateAdmin, Progress::Ok).await;
                    Some(created)
                }
                Err(e) => return rep.fail(ProvisionStep::CreateAdmin, e).await,
            }
        }
    };

    // ---- 7. Activate ----
    //
    // The row went in inactive, so until now the tenant does not
    // resolve. A migration failure leaves it that way on purpose: the
    // row stays so an operator can see what was half-made, but nothing
    // routes to it.
    if matches!(migrations, MigrationsOutcome::Failed(_)) {
        rep.step(
            ProvisionStep::Activate,
            Progress::Skipped("migrations failed; the tenant stays inactive"),
        )
        .await;
    } else {
        rep.step(ProvisionStep::Activate, Progress::Started).await;
        let activated = match injected_failure(ProvisionStep::Activate) {
            Ok(()) => activate(&registry, org_id).await,
            Err(e) => Err(e),
        };
        if let Err(e) = activated {
            return rep.fail(ProvisionStep::Activate, e).await;
        }
        super::invalidate_org_cache();
        rep.step(ProvisionStep::Activate, Progress::Ok).await;
    }

    Ok(ProvisionOutcome {
        org_id,
        slug: request.slug.clone(),
        mode: request.mode(),
        migrations,
        first_admin,
    })
}

/// Reach the tenant's own database before anything is written.
///
/// Only database-mode has a separate database to reach; a schema-mode
/// tenant lives in the registry's, which is already connected.
async fn check_connection(
    checked: &CheckedRequest,
    registry: &crate::sql::Pool,
    rep: &Reporter<'_>,
) -> Result<(), TenancyError> {
    if checked.probe_target().is_none() {
        // Schema mode: the registry's own login makes the schema.
        rep.step(ProvisionStep::CheckConnection, Progress::Started)
            .await;
        return match schema_rights_refusal(registry).await {
            None => {
                rep.step(ProvisionStep::CheckConnection, Progress::Ok).await;
                Ok(())
            }
            Some(e) => rep.fail(ProvisionStep::CheckConnection, e).await,
        };
    }

    rep.step(ProvisionStep::CheckConnection, Progress::Started)
        .await;
    match checked.preflight().await {
        Ok(()) => {
            let url = checked.probe_target().unwrap_or_default();
            let done = registry_alias(url, registry)
                .await
                .map_or(Progress::Ok, |w| Progress::Done(RunText::fixed(w)));
            rep.step(ProvisionStep::CheckConnection, done).await;
            Ok(())
        }
        Err(e) => rep.fail(ProvisionStep::CheckConnection, e).await,
    }
}

/// What [`registry_alias`] reports.
pub(crate) const REGISTRY_ALIAS: &str = "warning: this URL reaches the registry's own database \
     under another name; give the tenant a database of its own (a later release refuses it)";

/// The registry's own database under another host name, link or mount,
/// which the endpoint check cannot see: compared by server identity. Only
/// warned in this release; `None` when it differs or cannot be read.
pub(crate) async fn registry_alias(url: &str, registry: &crate::sql::Pool) -> Option<&'static str> {
    use super::db_identity::DbIdentity;
    let theirs = DbIdentity::of_pool(registry).await.ok()??;
    let ours = DbIdentity::of_url(url).await.ok()??;
    if ours != theirs {
        return None;
    }
    tracing::warn!(target: "rustango::tenancy::provision", "{REGISTRY_ALIAS}");
    Some(REGISTRY_ALIAS)
}

/// What [`resumed_on_registry`] reports.
const RESUMED_ON_REGISTRY: &str = "warning: this tenant's database URL now reaches \
     the registry's own database; its migrations run there (a later release refuses it)";

/// A resumed tenant skips [`checked_request`], and its secret may resolve
/// elsewhere since: the same registry checks, on the URL as it resolves now.
async fn resumed_on_registry<DB: Database>(
    pools: &TenantPools<DB>,
    registry_url: &str,
    org: &Org,
) -> Option<&'static str>
where
    crate::sql::Pool: From<sqlx::Pool<DB>>,
{
    let reference = org.database_url.as_deref()?;
    // An unresolvable URL fails at migrate, with its own error.
    let url = pools.resolve_secret(reference).await.ok()?;
    let registry = pools.registry_pool();
    let (kind, pooled) = Endpoint::of_pool(&registry);
    let named = Endpoint::scheme_kind(registry_url)
        .and_then(|k| Some((k, Endpoint::parse(registry_url, k).ok()?)));
    let reaches = |kind, registry: &Endpoint| {
        Endpoint::of_url(&url, kind).ok().flatten().as_ref() == Some(registry)
    };
    if reaches(kind, &pooled) || named.is_some_and(|(k, e)| reaches(k, &e)) {
        tracing::warn!(target: "rustango::tenancy::provision", org = %org.slug, "{RESUMED_ON_REGISTRY}");
        return Some(RESUMED_ON_REGISTRY);
    }
    registry_alias(&url, &registry).await
}

/// Why the registry's login may not create a tenant schema; `None` when
/// it may, or when the server would not say.
pub(crate) async fn schema_rights_refusal(registry: &crate::sql::Pool) -> Option<TenancyError> {
    let rights = super::rights::RightsReport::for_new_schema(registry).await;
    (!rights.is_ready())
        .then(|| TenancyError::Validation(rights.refusal("schema-mode tenants cannot be created")))
}

/// Make the tenant's storage, if it needs making.
///
/// Schema-mode gets a `CREATE SCHEMA`; database-mode brings its own
/// database, already reached by the connection check. Deliberately
/// before the `Org` row lands, so a failed `INSERT` leaves no orphan
/// schema.
async fn provision_storage<DB: Database>(
    pools: &TenantPools<DB>,
    schema_name: Option<&str>,
    rep: &Reporter<'_>,
) -> Result<(), TenancyError> {
    // `schema_name_for` is `Some` exactly in schema mode.
    let Some(schema) = schema_name else {
        rep.step(
            ProvisionStep::ProvisionStorage,
            Progress::Skipped("database-mode tenants bring their own database"),
        )
        .await;
        return Ok(());
    };

    rep.step(ProvisionStep::ProvisionStorage, Progress::Started)
        .await;
    if let Err(e) = provision_schema(pools, schema).await {
        return rep.fail(ProvisionStep::ProvisionStorage, e).await;
    }
    rep.step(ProvisionStep::ProvisionStorage, Progress::Ok)
        .await;
    Ok(())
}

/// A slug becomes a database name, a schema name and a hostname label
/// (`<slug>.<apex>`), so it must be legal as all three.
///
/// The hostname label is strictest: RFC 1123 allows lowercase letters,
/// digits and hyphens only, and no hyphen at either end. A slug with a
/// space would give a live tenant nobody can reach.
fn validate_slug(slug: &str) -> Result<(), String> {
    if slug.is_empty() {
        return Err("a slug is required".into());
    }
    if slug.len() > 63 {
        // The hostname-label limit; the database-name limits are
        // higher, so this is the binding one.
        return Err(format!(
            "slug `{slug}` is {} characters; a hostname label allows at most 63",
            slug.len()
        ));
    }
    let legal = |b: u8| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-';
    if let Some(bad) = slug.bytes().find(|b| !legal(*b)) {
        return Err(format!(
            "slug `{slug}` contains `{}` — only lowercase letters, digits and hyphens are \
             allowed, because the slug becomes a database name, a schema name and a \
             hostname label",
            bad as char
        ));
    }
    if slug.starts_with('-') || slug.ends_with('-') {
        return Err(format!(
            "slug `{slug}` may not start or end with a hyphen — a hostname label cannot"
        ));
    }
    Ok(())
}

/// Check every operator-supplied field and return the request with the
/// host pattern normalized.
///
/// One function, called from the `Validate` step, so the console, the
/// webhook and `manage create-tenant` agree on what a legal tenant is.
///
/// Each field feeds an exact matcher or identifier. A value the matcher
/// can never produce gives a tenant that is registered, active and
/// unreachable, so it is refused here.
fn validate_fields(request: &ProvisionRequest) -> Result<ProvisionRequest, String> {
    validate_slug(&request.slug)?;

    // The effective name, so the slug-derived default is checked by the
    // same rule as an explicit one.
    if let Some(schema) = schema_name_for(request) {
        validate_schema_name(&schema)?;
    }
    if let Some(prefix) = &request.path_prefix {
        validate_new_path_prefix(prefix)?;
    }
    if let Some(port) = request.port {
        validate_port(port)?;
    }
    if request.first_admin.is_some() && !request.run_migrations {
        return Err("a first admin needs the tenant's migrations; do not skip them".into());
    }

    let mut out = request.clone();
    // The `<slug>.<APEX>` default is filled in here so it is validated too.
    let pattern = request.host_pattern.clone().or_else(|| {
        default_host_pattern(request, crate::tenancy::server::configured_apex_domain())
    });
    out.host_pattern = pattern.as_deref().map(validate_host_pattern).transpose()?;
    Ok(out)
}

/// The host a tenant created without one answers to: `<slug>.<apex>`.
/// A path-prefix tenant lives on the apex itself, so it gets none.
fn default_host_pattern(request: &ProvisionRequest, apex: Option<String>) -> Option<String> {
    if request.path_prefix.is_some() {
        return None;
    }
    apex.map(|apex| format!("{}.{apex}", request.slug))
}

/// The schema a schema-mode tenant is created in.
///
/// The name reaches `CREATE SCHEMA` and `SET search_path` as an
/// identifier. [`crate::sql::Dialect::quote_ident`] quotes it, so odd
/// characters cannot execute, but a schema nobody can name again
/// without quoting is a tenant nobody can operate. So reject it here.
///
/// The rule is the slug's, plus underscores: a schema name is not a
/// hostname label, and Postgres is happy with `_`.
pub(crate) fn validate_schema_name(name: &str) -> Result<(), String> {
    validate_stored_schema_name(name)?;
    let legal = |b: u8| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'_' || b == b'-';
    if let Some(bad) = name.bytes().find(|b| !legal(*b)) {
        return Err(format!(
            "schema name `{name}` contains `{}` — only lowercase letters, digits, underscores \
             and hyphens are allowed",
            bad as char
        ));
    }
    // A leading digit is fine, because the default schema name is the
    // slug and a slug may start with one. Only a leading hyphen is
    // refused, since it reads as a flag.
    if name.starts_with('-') {
        return Err(format!("schema name `{name}` may not start with a hyphen"));
    }
    Ok(())
}

/// What an existing row's schema must still pass to be purged or moved:
/// older rows may hold any name, used quoted, but never a reserved one.
pub(crate) fn validate_stored_schema_name(name: &str) -> Result<(), String> {
    if name.is_empty() {
        return Err("a schema name is required".into());
    }
    if name.len() > 63 {
        // Postgres truncates identifiers at NAMEDATALEN-1 silently,
        // which would make the stored name and the real schema differ.
        return Err(format!(
            "schema name `{name}` is {} characters; Postgres allows at most 63",
            name.len()
        ));
    }
    // `pg_*` is reserved. Catch it here so it reads as a validation
    // error rather than a failed run.
    if name.starts_with("pg_") || name == "information_schema" {
        return Err(format!(
            "schema name `{name}` is reserved by Postgres — choose another"
        ));
    }
    // `public` holds the registry tables and ends every tenant's
    // `search_path`, so a tenant there would share both.
    if name == "public" {
        return Err(format!(
            "schema name `{name}` is the registry's schema — choose another"
        ));
    }
    Ok(())
}

/// The `Host` header this tenant answers to.
///
/// The resolver lowercases the header and strips the port, then
/// compares exactly. So uppercase, a `:port` suffix or a wildcard all
/// give a tenant that can never be reached.
///
/// Uppercase is lowercased here, since that is what the resolver does
/// anyway. A port or a wildcard is refused: the operator wants
/// something this matcher cannot do.
pub(crate) fn validate_host_pattern(pattern: &str) -> Result<String, String> {
    if pattern.contains(':') {
        return Err(format!(
            "host pattern `{pattern}` carries a port — the `Host` header is matched with the \
             port stripped, so this could never match. Put the port in the port field"
        ));
    }
    if pattern.contains('*') {
        return Err(format!(
            "host pattern `{pattern}` uses a wildcard — patterns are matched exactly, so this \
             could never match. Register one tenant per hostname"
        ));
    }
    if pattern.len() > 253 {
        return Err(format!(
            "host pattern `{pattern}` is {} characters; a hostname allows at most 253",
            pattern.len()
        ));
    }
    let normalized = pattern.to_ascii_lowercase();
    for label in normalized.split('.') {
        if label.is_empty() {
            return Err(format!(
                "host pattern `{pattern}` has an empty label — check for a doubled or trailing dot"
            ));
        }
        if label.len() > 63 {
            return Err(format!(
                "host pattern `{pattern}` has a label longer than the 63 characters a hostname \
                 allows"
            ));
        }
        let legal = |b: u8| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'-';
        if let Some(bad) = label.bytes().find(|b| !legal(*b)) {
            return Err(format!(
                "host pattern `{pattern}` contains `{}` — a hostname allows only letters, \
                 digits, hyphens and dots",
                bad as char
            ));
        }
        if label.starts_with('-') || label.ends_with('-') {
            return Err(format!(
                "host pattern `{pattern}` has a label starting or ending with a hyphen, which a \
                 hostname cannot"
            ));
        }
    }
    Ok(normalized)
}

/// The URL path segment this tenant answers to.
///
/// `PathPrefixResolver` takes the **first** path segment and looks up
/// `"/<segment>"`. So a stored prefix that is not exactly one
/// leading-slash segment — no slash, a second segment, a trailing slash
/// — cannot be produced by that lookup and never matches.
pub(crate) fn validate_path_prefix(prefix: &str) -> Result<(), String> {
    let Some(segment) = prefix.strip_prefix('/') else {
        return Err(format!(
            "path prefix `{prefix}` must start with `/` — the resolver looks up `/<segment>`"
        ));
    };
    if segment.is_empty() {
        return Err("path prefix `/` is the apex, which resolves to no tenant".into());
    }
    if segment.contains('/') {
        return Err(format!(
            "path prefix `{prefix}` has more than one segment — only the first segment of the \
             URL is matched, so this could never match"
        ));
    }
    if segment.bytes().all(|b| b == b'.') {
        return Err(format!(
            "path prefix `{prefix}` is a relative-path segment, not a tenant prefix"
        ));
    }
    let legal = |b: u8| b.is_ascii_alphanumeric() || matches!(b, b'-' | b'_' | b'.' | b'~');
    if let Some(bad) = segment.bytes().find(|b| !legal(*b)) {
        return Err(format!(
            "path prefix `{prefix}` contains `{}` — a path segment that needs escaping would \
             not match the decoded path",
            bad as char
        ));
    }
    Ok(())
}

/// A prefix a tenant may be given: legal, and not a segment the apex
/// serves first (operator console, health probes), which would hide it.
pub(crate) fn validate_new_path_prefix(prefix: &str) -> Result<(), String> {
    validate_path_prefix(prefix)?;
    let console = super::operator_console::CONSOLE_SEGMENTS
        .iter()
        .any(|seg| prefix.strip_prefix('/') == Some(seg));
    if console || crate::health::PROBE_PATHS.contains(&prefix) {
        return Err(format!(
            "path prefix `{prefix}` is an operator console route on the apex, which would \
             hide the tenant — choose another"
        ));
    }
    Ok(())
}

/// The TCP port this tenant answers on.
///
/// `i32` is the column's type, not the range of a port. `-1` and
/// `999999` both parsed and both stored, producing a tenant matched
/// against a port no listener can ever have.
pub(crate) fn validate_port(port: i32) -> Result<(), String> {
    if !(1..=65535).contains(&port) {
        return Err(format!(
            "port {port} is outside the 1–65535 a TCP port can be"
        ));
    }
    Ok(())
}

/// A request that passed [`checked_request`]. The only source of a new
/// tenant's stored `database_url` and of the URL its preflight probes.
pub(crate) struct CheckedRequest {
    /// `host_pattern` normalized.
    pub(crate) request: ProvisionRequest,
    /// The registry's; the request's, if it named one, equals it.
    pub(crate) backend: BackendKind,
    source: CheckedSource,
}

/// Where the database comes from, checked.
enum CheckedSource {
    Schema,
    Url(VettedUrl),
    /// Active, and on the registry's backend.
    Server(Box<super::DbServer>),
}

impl CheckedRequest {
    /// What goes in `Org.database_url` for an existing database.
    pub(crate) fn stored_database_url(&self) -> Option<String> {
        match &self.source {
            CheckedSource::Url(u) => Some(u.stored.clone()),
            CheckedSource::Schema | CheckedSource::Server(_) => None,
        }
    }

    /// The saved server to create the database on.
    pub(crate) fn server(&self) -> Option<&super::DbServer> {
        match &self.source {
            CheckedSource::Server(s) => Some(s),
            CheckedSource::Schema | CheckedSource::Url(_) => None,
        }
    }

    /// The resolved URL of an existing database, to probe.
    fn probe_target(&self) -> Option<&str> {
        match &self.source {
            CheckedSource::Url(u) => Some(&u.resolved),
            CheckedSource::Schema | CheckedSource::Server(_) => None,
        }
    }

    /// Reach the tenant's own database before anything is written. `Ok`
    /// when there is none. Shared by every create path.
    pub(crate) async fn preflight(&self) -> Result<(), TenancyError> {
        let Some(url) = self.probe_target() else {
            return Ok(());
        };
        preflight::check(url, &self.request.preflight)
            .await
            .map(|_| ())
            // `Validation`: the URL is wrong and the summary says what to
            // change. The driver's own words are logged, not shown.
            .map_err(|d| {
                tracing::warn!(
                    target: "rustango::tenancy::provision",
                    org = %self.request.slug,
                    detail = %d.detail,
                    "tenant connection check failed"
                );
                TenancyError::Validation(d.summary())
            })
    }
}

/// Every free-text field, the database URL, then the routing clash check.
/// Shared with `api::create_tenant`.
pub(crate) async fn checked_request(
    registry: &crate::sql::Pool,
    secrets: &TenantSecrets,
    registry_url: &str,
    request: &ProvisionRequest,
) -> Result<CheckedRequest, TenancyError> {
    let normalized = validate_fields(request).map_err(TenancyError::Validation)?;
    let backend = BackendKind::of_pool(registry);
    if let Some(asked) = normalized.backend.filter(|b| *b != backend) {
        return Err(TenancyError::Validation(format!(
            "the registry runs on {backend}, so a {asked} tenant cannot be opened; a tenant uses \
             the registry's backend"
        )));
    }
    backend
        .validate_storage_mode(normalized.mode())
        .map_err(|m| TenancyError::Validation(m.to_owned()))?;
    let source = match &normalized.storage {
        TenantStorage::Schema(_) => CheckedSource::Schema,
        TenantStorage::Database(DatabaseSource::Url(reference)) => CheckedSource::Url(
            vet_database_url(reference, secrets, registry, Some(registry_url)).await?,
        ),
        TenantStorage::Database(DatabaseSource::Server(name)) => CheckedSource::Server(Box::new(
            checked_server(registry, secrets, name, &normalized.slug, backend).await?,
        )),
    };
    if let Some(clash) = routing_clash(registry, &normalized).await? {
        return Err(TenancyError::Validation(format!(
            "{clash} is already used by another tenant"
        )));
    }
    Ok(CheckedRequest {
        request: normalized,
        backend,
        source,
    })
}

/// The saved server `name`, refused unless it is active, on `backend`,
/// and a URL with a password could be stored for a tenant on it.
async fn checked_server(
    registry: &crate::sql::Pool,
    secrets: &TenantSecrets,
    name: &str,
    slug: &str,
    backend: BackendKind,
) -> Result<super::DbServer, TenancyError> {
    let server = super::db_server::get(registry, name).await?;
    if !server.active {
        return Err(TenancyError::Validation(format!(
            "database server `{}` is inactive",
            server.name
        )));
    }
    if server.backend_kind()? != backend {
        return Err(TenancyError::Validation(format!(
            "database server `{}` is a {} server, but the registry runs on {backend}",
            server.name, server.backend
        )));
    }
    crate::sql::DatabaseName::new(slug).map_err(|e| TenancyError::Validation(e.to_string()))?;
    // Before any DDL: the login rustango mints is stored only sealed,
    // whatever the registry's strictness.
    if backend != BackendKind::Sqlite {
        if crate::aead::env_key().is_none() {
            return Err(TenancyError::Validation(format!(
                "{} is not set: the login rustango creates for a tenant is stored only sealed",
                crate::aead::KEY_VAR
            )));
        }
        let probe = format!("{backend}://tenant:password@localhost/{slug}");
        super::secrets::stored_form(&probe, &probe, &secrets.clone().require_sealed())
            .await
            .map_err(TenancyError::Validation)?;
    }
    Ok(server)
}

/// A tenant database URL that passed [`vet_database_url`].
pub(crate) struct VettedUrl {
    /// The literal URL, secret opened. For probing only: never stored or shown.
    pub(crate) resolved: String,
    /// What goes in `Org.database_url`: the reference, or the literal sealed.
    pub(crate) stored: String,
}

/// [`check_database_url`], then the form to store: what create, edit and
/// storage moves write.
pub(crate) async fn vet_database_url(
    reference: &str,
    secrets: &TenantSecrets,
    registry: &crate::sql::Pool,
    registry_url: Option<&str>,
) -> Result<VettedUrl, TenancyError> {
    let resolved = check_database_url(reference, secrets, registry, registry_url).await?;
    let stored = super::secrets::stored_form(reference, &resolved, secrets)
        .await
        .map_err(TenancyError::Validation)?;
    Ok(VettedUrl { resolved, stored })
}

/// Resolve `reference`, then refuse a URL this registry's pools cannot
/// open or that names the registry itself, so a reference is checked like
/// a literal. Returns the resolved URL. Errors never echo it: it may be a
/// bare secret.
pub(crate) async fn check_database_url(
    reference: &str,
    secrets: &dyn SecretsResolver,
    registry: &crate::sql::Pool,
    registry_url: Option<&str>,
) -> Result<String, TenancyError> {
    let resolved = secrets
        .resolve(reference)
        .await
        .map_err(|e| TenancyError::Validation(format!("cannot resolve the database URL: {e}")))?;
    refuse_other_backend(&resolved, reference, registry).map_err(TenancyError::Validation)?;
    refuse_registry_pool_as(&resolved, reference, registry).map_err(TenancyError::Validation)?;
    if let Some(url) = registry_url {
        refuse_registry_url_as(&resolved, reference, url).map_err(TenancyError::Validation)?;
    }
    Ok(resolved)
}

/// `TenantPools<DB>` opens every tenant with the registry's driver, which
/// reads any URL as its own: a MySQL URL on a Postgres registry passes the
/// probe and then fails at pool build. Names `reference`, never `url`.
fn refuse_other_backend(
    url: &str,
    reference: &str,
    registry: &crate::sql::Pool,
) -> Result<(), String> {
    let want = BackendKind::of_pool(registry);
    let shown = crate::sql::connect_diagnosis::redact(reference);
    match Endpoint::scheme_kind(url) {
        Some(got) if got == want => Ok(()),
        Some(got) => Err(format!(
            "{shown} is a {got} URL, but the registry runs on {want}; a tenant database must \
             use the registry's backend"
        )),
        // The text has no known scheme, so `redact` cannot mask it.
        None if reference == url => Err(format!("the database URL is not a {want} URL")),
        None => Err(format!("{shown} does not resolve to a {want} URL")),
    }
}

/// The schema, host, path prefix or port of `request` another tenant uses.
async fn routing_clash(
    registry: &crate::sql::Pool,
    request: &ProvisionRequest,
) -> Result<Option<String>, crate::sql::ExecError> {
    use super::org_host::{host_claimed, port_claimed, prefix_claimed, schema_claimed};
    if let Some(schema) = schema_name_for(request) {
        if schema_claimed(registry, &schema, None).await? {
            return Ok(Some(format!("schema `{schema}`")));
        }
    }
    if let Some(host) = &request.host_pattern {
        if host_claimed(registry, host, None).await? {
            return Ok(Some(format!("host `{host}`")));
        }
    }
    if let Some(prefix) = &request.path_prefix {
        if prefix_claimed(registry, prefix, None).await? {
            return Ok(Some(format!("path prefix `{prefix}`")));
        }
    }
    if let Some(port) = request.port {
        if port_claimed(registry, port, None).await? {
            return Ok(Some(format!("port {port}")));
        }
    }
    Ok(None)
}

/// Refuse a tenant URL that points at the registry's own database.
///
/// Otherwise provisioning runs the *tenant* migration chain into the
/// *registry*, creating tenant tables there and writing entries into
/// the registry's own ledger. Any destructive tenant migration would
/// then run against the registry.
///
/// Compares the [`Endpoint`] only, because two URLs may name the same
/// database with different credentials.
#[cfg(test)]
fn refuse_registry_url(tenant_url: &str, registry_url: &str) -> Result<(), String> {
    refuse_registry_url_as(tenant_url, tenant_url, registry_url)
}

/// [`refuse_registry_url`], naming `reference` (what the operator gave)
/// in errors instead of the resolved `tenant_url`.
fn refuse_registry_url_as(
    tenant_url: &str,
    reference: &str,
    registry_url: &str,
) -> Result<(), String> {
    let Some(kind) = Endpoint::scheme_kind(registry_url) else {
        return Ok(());
    };
    match Endpoint::parse(registry_url, kind) {
        Ok(registry) => refuse_endpoint(tenant_url, reference, kind, &registry),
        Err(_) => Ok(()),
    }
}

/// [`refuse_registry_url`] against the database the registry pool is
/// connected to (#2320).
#[cfg(test)]
fn refuse_registry_pool(tenant_url: &str, registry: &crate::sql::Pool) -> Result<(), String> {
    refuse_registry_pool_as(tenant_url, tenant_url, registry)
}

fn refuse_registry_pool_as(
    tenant_url: &str,
    reference: &str,
    registry: &crate::sql::Pool,
) -> Result<(), String> {
    let (kind, endpoint) = Endpoint::of_pool(registry);
    refuse_endpoint(tenant_url, reference, kind, &endpoint)
}

/// Errors name `reference`, password masked; the parse detail only when
/// it describes that same text.
fn refuse_endpoint(
    tenant_url: &str,
    reference: &str,
    kind: BackendKind,
    registry: &Endpoint,
) -> Result<(), String> {
    let shown = crate::sql::connect_diagnosis::redact(reference);
    let tenant = Endpoint::of_url(tenant_url, kind).map_err(|e| {
        if reference == tenant_url {
            format!("cannot read {shown} ({e}), so cannot tell it from the registry's own database")
        } else {
            format!(
                "cannot read what {shown} resolves to, so cannot tell it from the registry's \
                 own database"
            )
        }
    })?;
    if tenant.as_ref() == Some(registry) {
        return Err(format!(
            "this is the registry's own database ({shown}). A tenant needs its own — pointing one \
             here would run the tenant migrations over the registry"
        ));
    }
    Ok(())
}

/// Where a connection lands, as the backend's sqlx options read it, so
/// `?host=`, sockets, `PG*` env and default ports resolve as the pool did (#2320).
#[derive(Debug, PartialEq, Eq)]
enum Endpoint {
    #[cfg(any(feature = "postgres", feature = "mysql"))]
    Server {
        place: Place,
        port: u16,
        database: Option<String>,
    },
    #[cfg(feature = "sqlite")]
    Sqlite(std::path::PathBuf),
}

#[cfg(any(feature = "postgres", feature = "mysql"))]
#[derive(Debug, PartialEq, Eq)]
enum Place {
    /// Lowercased.
    Host(String),
    Socket(std::path::PathBuf),
}

impl Endpoint {
    fn scheme_kind(url: &str) -> Option<BackendKind> {
        BackendKind::of_url(url)
    }

    /// Read as the registry's backend connects it: sqlx ignores a PG/MySQL scheme and SQLite
    /// takes a bare path. `Ok(None)` when another scheme fails to parse, e.g. a secret reference.
    fn of_url(url: &str, kind: BackendKind) -> Result<Option<Self>, String> {
        match Self::parse(url, kind) {
            Ok(endpoint) => Ok(Some(endpoint)),
            Err(e) if Self::scheme_kind(url) == Some(kind) => Err(e),
            Err(_) => Ok(None),
        }
    }

    fn parse(url: &str, kind: BackendKind) -> Result<Self, String> {
        match kind {
            #[cfg(feature = "postgres")]
            BackendKind::Postgres => url
                .parse::<sqlx::postgres::PgConnectOptions>()
                .map(|o| Self::of_pg(&o))
                .map_err(|e| e.to_string()),
            #[cfg(feature = "mysql")]
            BackendKind::MySql => url
                .parse::<sqlx::mysql::MySqlConnectOptions>()
                .map(|o| Self::of_mysql(&o))
                .map_err(|e| e.to_string()),
            #[cfg(feature = "sqlite")]
            BackendKind::Sqlite => url
                .parse::<sqlx::sqlite::SqliteConnectOptions>()
                .map(|o| Self::Sqlite(sqlite_file(o.get_filename())))
                .map_err(|e| e.to_string()),
            #[allow(unreachable_patterns)]
            other => Err(format!("this build has no {} backend", other.as_str())),
        }
    }

    fn of_pool(pool: &crate::sql::Pool) -> (BackendKind, Self) {
        match pool {
            #[cfg(feature = "postgres")]
            crate::sql::Pool::Postgres(p) => {
                (BackendKind::Postgres, Self::of_pg(&p.connect_options()))
            }
            #[cfg(feature = "mysql")]
            crate::sql::Pool::Mysql(p) => {
                (BackendKind::MySql, Self::of_mysql(&p.connect_options()))
            }
            #[cfg(feature = "sqlite")]
            crate::sql::Pool::Sqlite(p) => (
                BackendKind::Sqlite,
                Self::Sqlite(sqlite_file(p.connect_options().get_filename())),
            ),
        }
    }

    /// No database means the server picks the user's name.
    #[cfg(feature = "postgres")]
    fn of_pg(o: &sqlx::postgres::PgConnectOptions) -> Self {
        Self::Server {
            place: Place::of(o.get_socket(), o.get_host()),
            port: o.get_port(),
            database: Some(o.get_database().unwrap_or(o.get_username()).to_owned()),
        }
    }

    /// A MySQL socket is the whole address; the port is unused.
    #[cfg(feature = "mysql")]
    fn of_mysql(o: &sqlx::mysql::MySqlConnectOptions) -> Self {
        Self::Server {
            port: if o.get_socket().is_some() {
                0
            } else {
                o.get_port()
            },
            place: Place::of(o.get_socket(), o.get_host()),
            database: o.get_database().map(str::to_owned),
        }
    }
}

#[cfg(any(feature = "postgres", feature = "mysql"))]
impl Place {
    /// sqlx keeps a default or `PGHOST` socket dir in `host`, a `?host=/dir` one in `socket`.
    fn of(socket: Option<&std::path::PathBuf>, host: &str) -> Self {
        match socket {
            Some(s) => Self::Socket(file_identity(s)),
            None if host.starts_with('/') => Self::Socket(file_identity(Path::new(host))),
            None => {
                let host = host.trim_start_matches('[').trim_end_matches(']');
                Self::Host(host.trim_end_matches('.').to_ascii_lowercase())
            }
        }
    }
}

/// The file SQLite opens: a `file:` URI names its path, percent-encoded, before `?`.
#[cfg(feature = "sqlite")]
pub(crate) fn sqlite_file(name: &Path) -> std::path::PathBuf {
    let name = name.to_string_lossy();
    let Some(uri) = name.strip_prefix("file:") else {
        return file_identity(Path::new(&*name));
    };
    let uri = uri.split(['?', '#']).next().unwrap_or_default();
    // `file://host/path`: only an empty or `localhost` authority is legal.
    let path = match uri.strip_prefix("//") {
        Some(rest) => rest.find('/').map_or("", |i| &rest[i..]),
        None => uri,
    };
    file_identity(Path::new(&crate::url_codec::percent_decode_path(path)))
}

/// The file `path` opens, walked as SQLite's unix VFS does: `..` pops by
/// text, and each symlink is spliced in where it stands, up to 200 hops.
#[cfg(unix)]
fn file_identity(path: &Path) -> std::path::PathBuf {
    let mut out = std::path::PathBuf::from("/");
    let mut hops = 0;
    if path.is_relative() {
        if let Ok(cwd) = std::env::current_dir() {
            walk_sqlite_path(&mut out, &cwd, &mut hops);
        }
    }
    walk_sqlite_path(&mut out, path, &mut hops);
    out
}

/// `appendAllPathElements` of `os_unix.c`.
#[cfg(unix)]
fn walk_sqlite_path(out: &mut std::path::PathBuf, path: &Path, hops: &mut u32) {
    use std::path::Component;
    const SQLITE_MAX_SYMLINK: u32 = 200;
    for part in path.components() {
        match part {
            Component::RootDir => *out = std::path::PathBuf::from("/"),
            Component::ParentDir => {
                out.pop();
            }
            Component::CurDir | Component::Prefix(_) => {}
            Component::Normal(name) => {
                out.push(name);
                let is_link = std::fs::symlink_metadata(&*out).is_ok_and(|m| m.is_symlink());
                if !is_link {
                    continue;
                }
                // SQLite refuses to open past the cap; the text is as good as any.
                *hops += 1;
                if *hops > SQLITE_MAX_SYMLINK {
                    return;
                }
                let Ok(target) = std::fs::read_link(&*out) else {
                    continue;
                };
                out.pop();
                walk_sqlite_path(out, &target, hops);
            }
        }
    }
}

/// `GetFullPathNameW`, as SQLite's Windows VFS: absolute, `..` popped by
/// text, links left alone.
#[cfg(not(unix))]
fn file_identity(path: &Path) -> std::path::PathBuf {
    use std::path::Component;
    let abs = std::env::current_dir().map_or_else(|_| path.to_owned(), |d| d.join(path));
    let mut out = std::path::PathBuf::new();
    for part in abs.components() {
        match part {
            Component::ParentDir => {
                out.pop();
            }
            Component::CurDir => {}
            other => out.push(other),
        }
    }
    out
}

/// Build a tenant URL on the same server as the registry, naming
/// `database`.
///
/// For the common "one Postgres, one database per tenant" setup. The
/// caller supplies a database *name*, not a URL, so the registry
/// password never reaches a page or a form post.
///
/// Returns `None` when the registry URL has no database segment to
/// replace; the operator should then supply a full URL.
#[must_use]
pub fn tenant_url_on_registry_server(registry_url: &str, database: &str) -> Option<String> {
    // sqlite is a file path, not a server: "same server, other
    // database" means a sibling file.
    if let Some(path) = registry_url.strip_prefix("sqlite://") {
        let path = path.split('?').next().unwrap_or(path);
        // An in-memory registry has no directory to put a sibling in.
        if path.is_empty() || path.starts_with(":memory:") {
            return None;
        }
        // A bare filename is relative to the working directory, which
        // is where the sibling belongs.
        let dir = path.rsplit_once('/').map_or(".", |(d, _)| d);
        // A sqlite database is a file, so `acme` becomes `acme.db` and
        // `acme.db` is left as-is. The check ignores case: `acme.DB` is
        // the same file on a case-insensitive filesystem.
        let file = if std::path::Path::new(database)
            .extension()
            .is_some_and(|e| e.eq_ignore_ascii_case("db"))
        {
            database.to_owned()
        } else {
            format!("{database}.db")
        };
        return Some(format!("sqlite://{dir}/{file}?mode=rwc"));
    }

    let (scheme, rest) = registry_url.split_once("://")?;
    // Split the query off first: it can hold a `/` (`sslrootcert=/ca.pem`)
    // and its TLS options must carry over to the tenant.
    let (rest, query) = match rest.split_once('?') {
        Some((path, q)) => (path, Some(q)),
        None => (rest, None),
    };
    // Keep userinfo and authority; replace only the path segment.
    let (authority, _old_db) = rest.rsplit_once('/')?;
    if authority.is_empty() {
        return None;
    }
    Some(format!(
        "{scheme}://{authority}/{database}{}",
        tls_query_suffix(query)
    ))
}

/// `?<tls keys>` from `query`, or empty. Only TLS keys carry over: sqlx
/// lets `dbname`/`host`/`user` in the query override the URL, so keeping
/// them could point at the registry.
pub(crate) fn tls_query_suffix(query: Option<&str>) -> String {
    let tls: Vec<&str> = query
        .unwrap_or("")
        .split('&')
        .filter(|chunk| {
            let key = chunk.split_once('=').map_or(*chunk, |(k, _)| k);
            TLS_QUERY_KEYS.contains(&crate::url_codec::url_decode(key).as_str())
        })
        .collect();
    if tls.is_empty() {
        String::new()
    } else {
        format!("?{}", tls.join("&"))
    }
}

/// The TLS query keys sqlx 0.8 reads for Postgres and MySQL.
const TLS_QUERY_KEYS: &[&str] = &[
    "sslmode",
    "ssl-mode",
    "sslrootcert",
    "ssl-root-cert",
    "sslca",
    "ssl-ca",
    "sslcert",
    "ssl-cert",
    "sslkey",
    "ssl-key",
];

/// The schema a schema-mode tenant lives in: whatever the request
/// named, or the slug. `None` in database-mode, which has no schema.
pub(crate) fn schema_name_for(request: &ProvisionRequest) -> Option<String> {
    match &request.storage {
        TenantStorage::Schema(name) => {
            Some(super::org::effective_schema(name.as_deref(), &request.slug).to_owned())
        }
        TenantStorage::Database(_) => None,
    }
}

/// The registry row for a validated request, slug-derived defaults
/// filled in.
fn new_org_row(
    checked: &CheckedRequest,
    schema_name: Option<String>,
    database_url: Option<String>,
) -> Org {
    let request = &checked.request;
    Org {
        id: Auto::default(),
        slug: request.slug.clone(),
        display_name: request
            .display_name
            .clone()
            .unwrap_or_else(|| request.slug.clone()),
        storage_mode: request.mode().as_str().into(),
        backend_kind: checked.backend.as_str().into(),
        database_url,
        schema_name,
        // `validate_fields` already filled in the `<slug>.<APEX>` default.
        host_pattern: request.host_pattern.clone(),
        port: request.port,
        path_prefix: request.path_prefix.clone(),
        // Inactive until the schema is in place. The resolver already
        // filters on this column, so it costs nothing on the hot path
        // and is the only signal that keeps a half-provisioned tenant
        // from serving. `ProvisionStep::Activate` flips it.
        active: false,
        created_at: chrono::Utc::now(),
        brand_name: None,
        brand_tagline: None,
        logo_path: None,
        favicon_path: None,
        primary_color: None,
        theme_mode: None,
    }
}

/// `CREATE SCHEMA` on the registry. An existing schema is refused, not
/// adopted: it may hold another app's data, and `purge-tenant` drops it (#2394).
/// A resumed run never gets here; its schema is already in place.
///
/// Schema mode is Postgres-only: `CREATE SCHEMA` and `SET search_path`
/// do not exist on SQLite or MySQL. There are two separate refusals:
/// the build has no `postgres` feature, or it does but these pools are
/// not Postgres, which only a runtime downcast can tell.
pub(crate) async fn provision_schema<DB: Database>(
    pools: &TenantPools<DB>,
    schema: &str,
) -> Result<(), TenancyError> {
    #[cfg(feature = "postgres")]
    {
        use crate::sql::Dialect as _;

        let pg_pools = (pools as &dyn std::any::Any)
            .downcast_ref::<TenantPools<sqlx::Postgres>>()
            .ok_or_else(|| {
                TenancyError::Validation(
                    "schema mode needs a Postgres registry — choose the database storage mode"
                        .into(),
                )
            })?;
        // The default `search_path` starts with "$user": a schema named
        // after a role would lead that role's sessions, the registry's included.
        if is_role_name(pg_pools.registry(), schema).await? {
            return Err(TenancyError::Validation(format!(
                "schema name `{schema}` is the name of a database role — choose another"
            )));
        }
        // Use the dialect's quoter, not a local copy: it doubles any
        // embedded `"`.
        let sql = format!("CREATE SCHEMA {}", crate::sql::Postgres.quote_ident(schema));
        match rustango::sql::sqlx::query(&sql)
            .execute(pg_pools.registry())
            .await
        {
            Ok(_) => Ok(()),
            // duplicate_schema; or unique_violation when a concurrent create
            // inserted the same name first.
            Err(sqlx::Error::Database(e))
                if e.code().as_deref() == Some("42P06")
                    || (e.code().as_deref() == Some("23505")
                        && e.constraint() == Some("pg_namespace_nspname_index")) =>
            {
                Err(TenancyError::Validation(format!(
                    "schema `{schema}` already exists — choose another schema name, or, if a \
                     failed create left it empty, run `DROP SCHEMA {}`",
                    crate::sql::Postgres.quote_ident(schema)
                )))
            }
            Err(e) => Err(e.into()),
        }
    }
    #[cfg(not(feature = "postgres"))]
    {
        let _ = (pools, schema);
        Err(TenancyError::Validation(
            "schema mode is not available on this server — choose the database storage mode".into(),
        ))
    }
}

/// Whether a database role is called `name`. Catalog SQL: schema mode is Postgres-only.
#[cfg(feature = "postgres")]
pub(crate) async fn is_role_name(pg: &sqlx::PgPool, name: &str) -> Result<bool, sqlx::Error> {
    rustango::sql::sqlx::query_scalar("SELECT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = $1)")
        .bind(name)
        .fetch_one(pg)
        .await
}

/// Drop the schema [`provision_schema`] just made when its `Org` row failed,
/// so a retry is not refused as taken. `RESTRICT`: an empty schema only.
pub(crate) async fn release_schema<DB: Database>(pools: &TenantPools<DB>, schema: &str) {
    #[cfg(feature = "postgres")]
    if let Some(pg_pools) =
        (pools as &dyn std::any::Any).downcast_ref::<TenantPools<sqlx::Postgres>>()
    {
        use crate::sql::Dialect as _;
        let sql = format!(
            "DROP SCHEMA {} RESTRICT",
            crate::sql::Postgres.quote_ident(schema)
        );
        if let Err(e) = rustango::sql::sqlx::query(&sql)
            .execute(pg_pools.registry())
            .await
        {
            tracing::warn!(target: "rustango::tenancy::provision", schema, error = %e, "could not drop the new schema");
        }
    }
    #[cfg(not(feature = "postgres"))]
    let _ = (pools, schema);
}

/// Flip the tenant live.
pub(crate) async fn activate(registry: &crate::sql::Pool, org_id: i64) -> Result<(), TenancyError> {
    use crate::sql::UpdaterPool as _;
    let updated = Org::objects()
        .where_(Org::id.eq(org_id))
        .update()
        .set("active", true)
        .execute_pool(registry)
        .await?;
    if updated == 0 {
        return Err(TenancyError::Validation(format!(
            "activate: no row updated for org {org_id} — was it deleted mid-provision?"
        )));
    }
    Ok(())
}

/// How long a saved server's admin login gets to connect.
const ADMIN_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);

/// A record with no tenant this young may belong to a run in progress, so
/// it is neither reclaimed nor ignored by `used_by`.
pub(crate) const RECORD_IN_PROGRESS: std::time::Duration = std::time::Duration::from_secs(600);

/// A database this run created on a saved server, and its record. Dropped
/// again by [`Self::roll_back`] unless [`Self::keep`] is called.
pub(super) struct ServerDatabase {
    admin: super::server_admin::ServerAdmin,
    created: super::server_admin::Created,
    record: super::TenantDatabase,
}

/// Fail the named step on purpose, so tests reach rollback paths no live
/// server can be made to hit (`RUSTANGO_PROVISION_FAIL_AT=<step>`).
#[cfg(feature = "testkit")]
fn injected_failure(step: ProvisionStep) -> Result<(), TenancyError> {
    if std::env::var("RUSTANGO_PROVISION_FAIL_AT").is_ok_and(|s| s == step.as_str()) {
        return Err(TenancyError::Validation(format!(
            "injected failure at {}",
            step.as_str()
        )));
    }
    Ok(())
}

#[cfg(not(feature = "testkit"))]
fn injected_failure(_: ProvisionStep) -> Result<(), TenancyError> {
    Ok(())
}

/// What [`ServerDatabase::roll_back`] managed.
pub(super) struct RolledBack {
    database: crate::sql::DatabaseName,
    login: Option<crate::sql::LoginName>,
    dropped: bool,
    row_removed: bool,
}

impl RolledBack {
    /// For an error message. Names only, never the URL.
    pub(super) fn describe(&self) -> String {
        let what = match &self.login {
            Some(l) => format!("database `{}` and login `{l}`", self.database),
            None => format!("database `{}`", self.database),
        };
        match (self.dropped, self.row_removed) {
            (true, true) => format!("the new {what} were dropped and the tenant removed"),
            (true, false) => format!(
                "the new {what} were dropped, but the tenant row could not be removed; it \
                 was left inactive without a database URL"
            ),
            (false, _) => format!(
                "the new {what} could not be dropped and remain, recorded in \
                 rustango_tenant_databases; a retry reclaims them"
            ),
        }
    }
}

/// The checks a create on a saved server runs before any DDL, and no DDL:
/// slug free, fields and routing, server rights, database and login free.
pub(crate) async fn dry_run_on_server(
    registry: &crate::sql::Pool,
    secrets: &TenantSecrets,
    registry_url: &str,
    request: &ProvisionRequest,
) -> Result<(), TenancyError> {
    let taken = Org::objects()
        .where_(Org::slug.eq(request.slug.clone()))
        .first(registry)
        .await?;
    if taken.is_some() {
        return Err(TenancyError::Validation(format!(
            "tenant slug `{}` already exists",
            request.slug
        )));
    }
    let checked = checked_request(registry, secrets, registry_url, request).await?;
    let Some(server) = checked.server() else {
        return Ok(());
    };
    let (admin, _) = ServerDatabase::open(server, registry).await?;
    let name = crate::sql::DatabaseName::new(request.slug.as_str())
        .map_err(|e| TenancyError::Validation(e.to_string()))?;
    let login = crate::sql::Login::new(super::server_admin::tenant_login_name(&name));
    let free = admin.names_free(&name, &login).await;
    admin.close().await;
    Ok(free?)
}

/// The registry database's utf8mb4 collation, which tenant databases copy:
/// the app was built against it. `Err` when it cannot be read.
async fn registry_collation(
    registry: &crate::sql::Pool,
) -> Result<Option<crate::sql::Collation>, crate::sql::ExecError> {
    let Some(sql) = registry.dialect().default_collation_sql() else {
        return Ok(None);
    };
    let rows: Vec<(String,)> = crate::sql::raw_query_pool(sql, Vec::new(), registry).await?;
    Ok(rows
        .into_iter()
        .next()
        .and_then(|(c,)| crate::sql::Collation::new(c).ok()))
}

/// The collation `admin` should create databases with, and a warning when
/// it falls back to the server's own utf8mb4 default.
async fn collation_for(
    admin: &super::server_admin::ServerAdmin,
    registry: &crate::sql::Pool,
    server: &str,
) -> (Option<crate::sql::Collation>, Option<&'static str>) {
    match registry_collation(registry).await {
        Ok(Some(c)) if admin.supports_collation(&c).await == Some(false) => {
            tracing::warn!(target: "rustango::tenancy::provision", server, collation = c.as_str(), "the server lacks the registry's collation; using its utf8mb4 default");
            (
                None,
                Some("the server lacks the registry's collation, so its utf8mb4 default is used"),
            )
        }
        Ok(c) => (c, None),
        Err(e) => {
            tracing::warn!(target: "rustango::tenancy::provision", server, error = %e, "could not read the registry's collation");
            (None, Some("could not read the registry's collation, so the server's utf8mb4 default is used"))
        }
    }
}

/// `e`, with what the rollback did added. A cause that is not
/// operator-facing is logged and replaced, so the leftovers always show.
pub(super) fn with_undone(e: TenancyError, undone: &RolledBack) -> TenancyError {
    match e {
        TenancyError::Validation(m) => {
            TenancyError::Validation(format!("{m} So {}.", undone.describe()))
        }
        other => {
            tracing::error!(target: "rustango::tenancy::provision", error = %other, "provisioning on a saved server failed");
            TenancyError::Validation(format!(
                "provisioning failed (the cause is in the server log), so {}.",
                undone.describe()
            ))
        }
    }
}

/// Empty `org`'s database URL and keep it inactive, best effort.
async fn forget_database(registry: &crate::sql::Pool, org: &Org) {
    use crate::sql::UpdaterPool as _;
    let res = Org::objects()
        .where_(Org::id.eq(org.id.get().copied().unwrap_or_default()))
        .update()
        .set("database_url", None::<String>)
        .set("active", false)
        .execute_pool(registry)
        .await;
    if let Err(e) = res {
        tracing::error!(target: "rustango::tenancy::provision", org = %org.slug, error = %e, "could not empty the database URL of a rolled-back tenant");
    }
}

/// Note that the database now exists. A failure leaves `created = false`,
/// which a retry treats with care.
async fn mark_created(record: &mut super::TenantDatabase, registry: &crate::sql::Pool) {
    use crate::sql::UpdaterPool as _;
    let id = record.id.get().copied().unwrap_or_default();
    let marked = super::TenantDatabase::objects()
        .where_(super::TenantDatabase::id.eq(id))
        .update()
        .set("created", true)
        .execute_pool(registry)
        .await;
    match marked {
        Ok(_) => record.created = true,
        Err(e) => {
            tracing::warn!(target: "rustango::tenancy::provision", error = %e, "could not mark the tenant database created");
        }
    }
}

/// Deal with a record an interrupted run left for `name` on `server`
/// before creating it again. A database the record says rustango created
/// is dropped; one it may not have created is refused, never dropped.
async fn reclaim(
    admin: &super::server_admin::ServerAdmin,
    server: &super::DbServer,
    name: &crate::sql::DatabaseName,
    registry: &crate::sql::Pool,
    rep: &Reporter<'_>,
) -> Result<(), TenancyError> {
    let Some(record) = super::TenantDatabase::objects()
        .where_(super::TenantDatabase::server_name.eq(server.name.clone()))
        .where_(super::TenantDatabase::database_name.eq(name.to_string()))
        .first(registry)
        .await?
    else {
        return Ok(());
    };
    let tenant = match record.org_id {
        Some(id) => {
            Org::objects()
                .where_(Org::id.eq(id))
                .first(registry)
                .await?
        }
        None => None,
    };
    if let Some(org) = tenant {
        return Err(TenancyError::Validation(format!(
            "database `{name}` on server `{}` belongs to tenant `{}`",
            server.name, org.slug
        )));
    }
    let cutoff =
        chrono::Utc::now() - chrono::Duration::from_std(RECORD_IN_PROGRESS).unwrap_or_default();
    if record.org_id.is_none() && record.created_at.get().is_some_and(|t| *t > cutoff) {
        return Err(TenancyError::Validation(format!(
            "another run is creating database `{name}` on server `{}`; try again in a few minutes",
            server.name
        )));
    }
    let login = crate::sql::Login::new(super::server_admin::tenant_login_name(name));
    if record.created {
        rep.step(ProvisionStep::Rollback, Progress::Started).await;
        if let Err(e) = admin.drop_tenant_database(name, &login).await {
            tracing::error!(target: "rustango::tenancy::provision", database = %name, error = %e, "could not drop what an interrupted run left");
            return Err(TenancyError::Validation(format!(
                "an interrupted run left database `{name}` and login `{}`, and they could not be \
                 dropped; drop them on the server, then retry",
                login.name()
            )));
        }
        record.delete_pool(registry).await?;
        let text = RunText::about_database(
            "dropped what an interrupted run left",
            name,
            Some(login.name()),
        );
        rep.step(ProvisionStep::Rollback, Progress::Done(text))
            .await;
        return Ok(());
    }
    // The run stopped around its `CREATE`: the database may be someone else's.
    let free = admin.names_free(name, &login).await;
    if let Err(super::server_admin::TenantDbError::DatabaseExists(_)) = free {
        return Err(TenancyError::Validation(format!(
            "a run was interrupted while creating database `{name}`, and a database of that \
             name exists that may not be rustango's; choose another slug, or drop it on the \
             server and retry"
        )));
    }
    record.delete_pool(registry).await?;
    Ok(())
}

/// Why the new login could not sign in, and where to look.
fn login_refused(
    login: &crate::sql::LoginName,
    db: &crate::sql::DatabaseName,
    d: &crate::sql::connect_diagnosis::ConnectDiagnosis,
) -> String {
    use crate::sql::connect_diagnosis::ConnectFault;
    let hint = match d.fault {
        ConnectFault::AuthFailed => {
            " Check that pg_hba.conf admits it (rules can name users), and on MySQL or \
             MariaDB that no more specific account of that name, or an anonymous \
             ''@'localhost' account, shadows it."
        }
        _ => "",
    };
    format!(
        "the new login `{login}` cannot sign in to database `{db}`: {}.{hint}",
        d.summary()
    )
}

impl ServerDatabase {
    /// Log in to `server` and refuse it unless its admin may provision.
    /// Nothing is created yet. Also returns a collation warning, if any.
    pub(super) async fn open(
        server: &super::DbServer,
        registry: &crate::sql::Pool,
    ) -> Result<(super::server_admin::ServerAdmin, Option<&'static str>), TenancyError> {
        let admin = server.connect(ADMIN_TIMEOUT).await?;
        let (collation, warning) = collation_for(&admin, registry, &server.name).await;
        let admin = admin.with_collation(collation);
        let report = match admin.probe().await {
            Ok(r) => r,
            Err(e) => {
                admin.close().await;
                return Err(e.into());
            }
        };
        let rights = report.rights();
        if !rights.is_ready() {
            admin.close().await;
            return Err(TenancyError::Validation(rights.refusal(&format!(
                "database server `{}` cannot create tenant databases",
                server.name
            ))));
        }
        Ok((admin, warning))
    }

    /// Create a database named `slug` and its own login with a random
    /// password, reporting each stage. The record is written first, so an
    /// interrupted run leaves a database that is always findable. An
    /// existing database or login is refused; a failed create leaves
    /// nothing behind.
    pub(super) async fn create(
        admin: super::server_admin::ServerAdmin,
        server: &super::DbServer,
        slug: &str,
        registry: &crate::sql::Pool,
        rep: &Reporter<'_>,
    ) -> Result<Self, TenancyError> {
        match Self::create_inner(&admin, server, slug, registry, rep).await {
            Ok((created, record)) => Ok(Self {
                admin,
                created,
                record,
            }),
            Err(e) => {
                admin.close().await;
                Err(e)
            }
        }
    }

    async fn create_inner(
        admin: &super::server_admin::ServerAdmin,
        server: &super::DbServer,
        slug: &str,
        registry: &crate::sql::Pool,
        rep: &Reporter<'_>,
    ) -> Result<(super::server_admin::Created, super::TenantDatabase), TenancyError> {
        use super::server_admin::CreateStage;
        let stage_step = |s: CreateStage| match s {
            CreateStage::Login => ProvisionStep::CreateLogin,
            CreateStage::Grant => ProvisionStep::Grant,
            _ => ProvisionStep::CreateDatabase,
        };
        let name = match crate::sql::DatabaseName::new(slug) {
            Ok(n) => n,
            Err(e) => {
                return rep
                    .fail(
                        ProvisionStep::CreateDatabase,
                        TenancyError::Validation(e.to_string()),
                    )
                    .await;
            }
        };
        let login = crate::sql::Login::new(super::server_admin::tenant_login_name(&name));
        if let Err(e) = reclaim(admin, server, &name, registry, rep).await {
            return rep.fail(ProvisionStep::CreateDatabase, e).await;
        }
        let mut record = super::TenantDatabase {
            id: Auto::default(),
            org_slug: slug.to_owned(),
            org_id: None,
            server_name: server.name.clone(),
            database_name: name.to_string(),
            login_name: (BackendKind::of_pool(registry) != BackendKind::Sqlite)
                .then(|| login.name().to_string()),
            created: false,
            created_at: Auto::default(),
        };
        if let Err(e) = record.insert_pool(registry).await {
            let e = if e.is_unique_violation() {
                TenancyError::Validation(format!(
                    "another run is creating database `{name}` on server `{}`; try again in a few \
                     minutes",
                    server.name
                ))
            } else {
                e.into()
            };
            return rep.fail(ProvisionStep::CreateDatabase, e).await;
        }
        let password = crate::sql::LoginPassword::generate();
        let current = std::sync::Mutex::new(None::<ProvisionStep>);
        let result = admin
            .create_tenant_database_with(&name, &login, &password, |stage| {
                let step = stage_step(stage);
                let prev = current
                    .lock()
                    .unwrap_or_else(std::sync::PoisonError::into_inner)
                    .replace(step);
                async move {
                    if let Some(prev) = prev {
                        rep.step(prev, Progress::Ok).await;
                    }
                    rep.step(step, Progress::Started).await;
                }
            })
            .await;
        let last = current
            .into_inner()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .unwrap_or(ProvisionStep::CreateDatabase);
        let created = match result {
            Ok(c) => c,
            // The create made a database it could not undo: ours, so a
            // retry may reclaim it.
            Err(e @ super::server_admin::TenantDbError::RollbackFailed { .. }) => {
                mark_created(&mut record, registry).await;
                return rep.fail(last, e.into()).await;
            }
            Err(e) => {
                // Nothing of ours exists: the create made nothing or undid it,
                // or found someone else's database.
                if let Err(d) = record.clone().delete_pool(registry).await {
                    tracing::warn!(target: "rustango::tenancy::provision", error = %d, "could not delete the record of a failed create");
                }
                return rep.fail(last, e.into()).await;
            }
        };
        mark_created(&mut record, registry).await;
        rep.step(last, Progress::Ok).await;
        if created.login.is_none() {
            for step in [
                ProvisionStep::CreateLogin,
                ProvisionStep::Grant,
                ProvisionStep::CheckLogin,
            ] {
                rep.step(step, Progress::Skipped("a SQLite tenant has no login"))
                    .await;
            }
        }
        Ok((created, record))
    }

    /// Sign in as the new login. A server rule can refuse a login the
    /// create accepted, and the migrate step would then fail obscurely.
    pub(super) async fn check_login(&self, rep: &Reporter<'_>) -> Result<(), TenancyError> {
        let Some(login) = &self.created.login else {
            return Ok(());
        };
        rep.step(ProvisionStep::CheckLogin, Progress::Started).await;
        if let Err(e) = injected_failure(ProvisionStep::CheckLogin) {
            return rep.fail(ProvisionStep::CheckLogin, e).await;
        }
        match crate::sql::Pool::connect_diagnosed(self.created.url(), ADMIN_TIMEOUT).await {
            Ok(pool) => {
                pool.close().await;
                rep.step(ProvisionStep::CheckLogin, Progress::Ok).await;
                Ok(())
            }
            Err(d) => {
                tracing::warn!(target: "rustango::tenancy::provision", detail = %d.detail, "the new tenant login cannot sign in");
                let e = TenancyError::Validation(login_refused(
                    login.name(),
                    &self.created.database,
                    &d,
                ));
                rep.fail(ProvisionStep::CheckLogin, e).await
            }
        }
    }

    /// What goes in `Org.database_url`: the tenant URL, always sealed. A
    /// login rustango minted is never stored in plain text.
    pub(super) async fn stored_url(
        &self,
        registry: &crate::sql::Pool,
        secrets: &TenantSecrets,
        registry_url: &str,
    ) -> Result<String, TenancyError> {
        let strict = secrets.clone().require_sealed();
        let vetted =
            vet_database_url(self.created.url(), &strict, registry, Some(registry_url)).await?;
        Ok(vetted.stored)
    }

    /// Link the record to the new tenant row. A failure only loses the
    /// link; the slug still names it.
    pub(super) async fn attach(&mut self, registry: &crate::sql::Pool, org_id: i64) {
        use crate::sql::UpdaterPool as _;
        let Some(id) = self.record.id.get().copied() else {
            return;
        };
        self.record.org_id = Some(org_id);
        let res = super::TenantDatabase::objects()
            .where_(super::TenantDatabase::id.eq(id))
            .update()
            .set("org_id", Some(org_id))
            .execute_pool(registry)
            .await;
        if let Err(e) = res {
            tracing::warn!(target: "rustango::tenancy::provision", error = %e, "could not link the tenant database record");
        }
    }

    /// The run succeeded: keep everything.
    pub(super) async fn keep(self) {
        self.admin.close().await;
    }

    /// Drop the database and login, then the record and `org` if given.
    /// When the drop fails both rows stay, so a resuming retry or a purge
    /// still finds the database.
    pub(super) async fn roll_back<DB: Database>(
        self,
        pools: &TenantPools<DB>,
        org: Option<&Org>,
        rep: &Reporter<'_>,
    ) -> RolledBack
    where
        crate::sql::Pool: From<sqlx::Pool<DB>>,
    {
        rep.step(ProvisionStep::Rollback, Progress::Started).await;
        let registry = pools.registry_pool();
        let mut undone = RolledBack {
            database: self.created.database.clone(),
            login: self.created.login.as_ref().map(|l| l.name().clone()),
            dropped: false,
            row_removed: false,
        };
        if let Some(org) = org {
            // Its pool holds sessions in the database about to go.
            pools.invalidate(&org.slug).await;
        }
        let dropped = self.admin.roll_back(self.created).await;
        self.admin.close().await;
        if let Err(e) = dropped {
            tracing::error!(
                target: "rustango::tenancy::provision",
                database = %undone.database,
                error = %e,
                "could not drop the tenant database a failed run created"
            );
            let text = RunText::about_database(
                "could not drop what this run created; it remains",
                &undone.database,
                undone.login.as_ref(),
            );
            rep.failed_with(ProvisionStep::Rollback, text).await;
            return undone;
        }
        undone.dropped = true;
        let tenant_row = match org {
            Some(org) => match super::decommission::remove_org_row(&registry, org).await {
                Ok(()) => Ok(()),
                Err(e) => {
                    // Never leave a resumable row pointing at a dropped database.
                    forget_database(&registry, org).await;
                    Err(e)
                }
            },
            None => Ok(()),
        };
        // The database is gone either way. A record left behind still says
        // `created`, and a retry reclaims it with a no-op drop.
        let record = self.record.delete_pool(&registry).await.map(drop);
        let rows = tenant_row.and(record.map_err(TenancyError::from));
        match rows {
            Ok(()) => {
                undone.row_removed = true;
                let text = RunText::about_database(
                    "dropped and the tenant removed",
                    &undone.database,
                    undone.login.as_ref(),
                );
                rep.step(ProvisionStep::Rollback, Progress::Done(text))
                    .await;
            }
            Err(e) => {
                tracing::error!(target: "rustango::tenancy::provision", error = %e, "could not remove the rows of a rolled-back tenant");
                let text = RunText::about_database(
                    "dropped, but the registry rows could not be removed",
                    &undone.database,
                    undone.login.as_ref(),
                );
                rep.failed_with(ProvisionStep::Rollback, text).await;
            }
        }
        undone
    }
}

/// Migrate the tenant that was just created, and only that one.
///
/// Uses [`tenant_migrate::migrate_one_tenant`], not the batch. The new
/// tenant is still inactive, so the batch would skip it; and the batch
/// would also migrate every other tenant, dragging an unrelated broken
/// chain into this run.
///
/// Never returns `Err`: the `Org` row already exists, and an error
/// return would hide that. A failure comes back as
/// [`MigrationsOutcome::Failed`].
async fn migrate_new_tenant<DB: Database>(
    pools: &TenantPools<DB>,
    registry_url: &str,
    dir: &Path,
    org: &Org,
    rep: &Reporter<'_>,
) -> MigrationsOutcome
where
    crate::sql::Pool: From<sqlx::Pool<DB>>,
{
    // Forward every migration event to the reporter, so a caller
    // watching a provisioning run sees the same per-migration detail
    // `manage migrate` prints (#1320). The reporter decides what to do
    // with them — the observer gets them now, the store when the lock
    // is released.
    let forward = move |event: tenant_migrate::TenantMigrationEvent| rep.migration(event);
    let forward: Option<&dyn tenant_migrate::TenantMigrationObserver> = Some(&forward);

    // No backend branch: `migrate_one_tenant` owns the
    // schema-mode-is-PG-only dispatch that used to be duplicated here,
    // and walks the app folders.
    match tenant_migrate::migrate_one_tenant(pools, org, dir, registry_url, forward).await {
        Ok(applied) => MigrationsOutcome::Applied(applied),
        Err(e) => MigrationsOutcome::Failed(e.to_string()),
    }
}

/// Provisioning for surfaces that have erased the backend type.
///
/// The operator console holds its pools as
/// `Arc<dyn TenantPoolInvalidator>` — deliberately, so `<DB>` does not
/// cascade through every handler — while [`provision_tenant`] is
/// generic over `DB`. This is the bridge, following the same
/// boxed-future shape [`super::TenantPoolInvalidator`] already uses.
///
/// It also closes over the registry URL and migrations directory, which
/// a request handler has no business carrying around.
///
/// Note what is **not** here: an observer. A caller watching a run
/// reads it back from [`super::provision_store`] instead, which is what
/// makes a console work when the pod serving the stream is not the pod
/// doing the work.
pub trait TenantProvisioner: Send + Sync {
    /// Stand up a tenant, recording the run.
    fn provision<'a>(
        &'a self,
        request: &'a ProvisionRequest,
        requested_by: Option<&'a str>,
        idempotency_key: Option<&'a str>,
    ) -> BoxFuture<'a, (super::provision_store::ProvisioningRun, ProvisionOutcome)>;

    /// Stand up a tenant into a run the caller already opened.
    ///
    /// The inbound webhook opens its run synchronously — so the
    /// response can carry an id, and so the `idempotency_key` unique
    /// constraint fires where an HTTP status can report it — and then
    /// hands the slow part to a task. Without this the task would open
    /// a *second* run for the same tenant.
    fn provision_in_run<'a>(
        &'a self,
        run_id: i64,
        request: &'a ProvisionRequest,
    ) -> BoxFuture<'a, ProvisionOutcome>;

    /// Migrate one tenant, or every active one when `slug` is `None`,
    /// recording into an already-open run.
    ///
    /// It lives here, not on its own trait, for the same reason as the
    /// rest: the console holds `Arc<dyn TenantProvisioner>` and has no
    /// `DB` to name.
    fn migrate_in_run<'a>(&'a self, run_id: i64, slug: Option<&'a str>) -> BoxFuture<'a, ()>;

    /// Purge a tenant into an open run of kind `purge`. The default
    /// refuses and closes the run, so older implementors still build.
    fn purge_in_run<'a>(
        &'a self,
        run_id: i64,
        slug: &'a str,
        purge_database: bool,
    ) -> BoxFuture<'a, super::decommission::Report> {
        let _ = (slug, purge_database);
        Box::pin(async move { Err(unsupported_run(&self.registry(), run_id).await) })
    }

    /// Rotate a server-made tenant's login password into an open run of
    /// kind `rotate_password`. The default refuses and closes the run.
    fn rotate_password_in_run<'a>(&'a self, run_id: i64, slug: &'a str) -> BoxFuture<'a, ()> {
        let _ = slug;
        Box::pin(async move { Err(unsupported_run(&self.registry(), run_id).await) })
    }

    /// The registry's own connection URL, for deriving a tenant URL on
    /// the same server. **Never render it**: it carries credentials.
    /// Derive first, then redact for display.
    fn registry_url(&self) -> String;

    /// The registry pool, so a caller can read runs and events back.
    fn registry(&self) -> crate::sql::Pool;

    /// How a first admin is handed over; `None` refuses first admins.
    fn first_admin_handover(&self) -> Option<&super::first_admin::FirstAdminHandover> {
        None
    }
}

/// Close `run_id` as failed: this provisioner cannot do that work.
async fn unsupported_run(registry: &crate::sql::Pool, run_id: i64) -> TenancyError {
    const WHY: &str = "this provisioner cannot run that operation";
    let text = RunText::fixed(WHY);
    let state = super::provision_store::RunState::Failed;
    if let Err(e) =
        super::provision_store::finish_run_text(registry, run_id, state, Some(&text)).await
    {
        tracing::warn!(target: "rustango::tenancy::provision", error = %e, "could not close provisioning run");
    }
    TenancyError::Validation(WHY.to_owned())
}

/// The boxed-future shape an object-safe async method has to return.
///
/// Spelled once: written out inline it is four lines of angle brackets
/// per method, which buries what each one actually does.
pub type BoxFuture<'a, T> =
    std::pin::Pin<Box<dyn std::future::Future<Output = Result<T, TenancyError>> + Send + 'a>>;

/// A [`TenantProvisioner`] over concrete pools.
pub struct Provisioner<DB: Database> {
    pools: std::sync::Arc<TenantPools<DB>>,
    registry_url: String,
    migrations_dir: std::path::PathBuf,
    first_admin: Option<super::first_admin::FirstAdminHandover>,
    /// `run-server --apex`: a blank host becomes `<slug>.<apex>`.
    apex: Option<String>,
}

impl<DB: Database> Provisioner<DB> {
    pub fn new(
        pools: std::sync::Arc<TenantPools<DB>>,
        registry_url: impl Into<String>,
        migrations_dir: impl Into<std::path::PathBuf>,
    ) -> Self {
        Self {
            pools,
            registry_url: registry_url.into(),
            migrations_dir: migrations_dir.into(),
            first_admin: None,
            apex: None,
        }
    }

    /// The apex a blank host defaults under, over the env var and settings.
    #[must_use]
    pub(crate) fn with_default_apex(mut self, apex: Option<String>) -> Self {
        self.apex = apex;
        self
    }

    fn with_default_host<'r>(
        &self,
        request: &'r ProvisionRequest,
    ) -> std::borrow::Cow<'r, ProvisionRequest> {
        match (
            &request.host_pattern,
            default_host_pattern(request, self.apex.clone()),
        ) {
            (None, Some(host)) => std::borrow::Cow::Owned(ProvisionRequest {
                host_pattern: Some(host),
                ..request.clone()
            }),
            _ => std::borrow::Cow::Borrowed(request),
        }
    }

    /// Let callers ask for a first admin, handed over by `handover`.
    #[must_use]
    pub fn with_first_admin_handover(
        mut self,
        handover: super::first_admin::FirstAdminHandover,
    ) -> Self {
        self.first_admin = Some(handover);
        self
    }

    /// Type-erase for the console and anything else that does not want
    /// `<DB>` in its state.
    #[must_use]
    pub fn erased(self) -> std::sync::Arc<dyn TenantProvisioner>
    where
        crate::sql::Pool: From<sqlx::Pool<DB>>,
    {
        std::sync::Arc::new(self)
    }
}

impl<DB: Database> TenantProvisioner for Provisioner<DB>
where
    crate::sql::Pool: From<sqlx::Pool<DB>>,
{
    fn provision<'a>(
        &'a self,
        request: &'a ProvisionRequest,
        requested_by: Option<&'a str>,
        idempotency_key: Option<&'a str>,
    ) -> BoxFuture<'a, (super::provision_store::ProvisioningRun, ProvisionOutcome)> {
        // Detached: a dropped request must not stop a run between its DDL
        // and the rows that record it. The caller still waits for it.
        let pools = self.pools.clone();
        let (url, dir) = (self.registry_url.clone(), self.migrations_dir.clone());
        let request = self.with_default_host(request).into_owned();
        let requested_by = requested_by.map(ToOwned::to_owned);
        let idempotency_key = idempotency_key.map(ToOwned::to_owned);
        Box::pin(async move {
            tokio::spawn(async move {
                provision_tenant_recorded(
                    pools.as_ref(),
                    &url,
                    &dir,
                    &request,
                    None,
                    requested_by.as_deref(),
                    idempotency_key.as_deref(),
                )
                .await
            })
            .await
            .map_err(|e| TenancyError::Validation(format!("the provisioning task stopped: {e}")))?
        })
    }

    fn provision_in_run<'a>(
        &'a self,
        run_id: i64,
        request: &'a ProvisionRequest,
    ) -> BoxFuture<'a, ProvisionOutcome> {
        Box::pin(async move {
            provision_tenant_in_run(
                self.pools.as_ref(),
                &self.registry_url,
                &self.migrations_dir,
                &self.with_default_host(request),
                None,
                run_id,
            )
            .await
        })
    }

    fn migrate_in_run<'a>(&'a self, run_id: i64, slug: Option<&'a str>) -> BoxFuture<'a, ()> {
        Box::pin(async move {
            super::migrate_run::migrate_in_run(
                self.pools.as_ref(),
                &self.migrations_dir,
                &self.registry_url,
                run_id,
                slug,
            )
            .await
        })
    }

    fn purge_in_run<'a>(
        &'a self,
        run_id: i64,
        slug: &'a str,
        purge_database: bool,
    ) -> BoxFuture<'a, super::decommission::Report> {
        Box::pin(async move {
            super::decommission::purge_in_run(
                self.pools.as_ref(),
                slug,
                purge_database,
                None,
                run_id,
            )
            .await
        })
    }

    fn rotate_password_in_run<'a>(&'a self, run_id: i64, slug: &'a str) -> BoxFuture<'a, ()> {
        Box::pin(async move {
            super::server_tenant::rotate_password_in_run(self.pools.as_ref(), slug, None, run_id)
                .await
        })
    }

    fn registry_url(&self) -> String {
        self.registry_url.clone()
    }

    fn registry(&self) -> crate::sql::Pool {
        self.pools.registry_pool()
    }

    fn first_admin_handover(&self) -> Option<&super::first_admin::FirstAdminHandover> {
        self.first_admin.as_ref()
    }
}

#[cfg(test)]
mod validation_tests {
    use super::*;

    #[test]
    fn request_debug_masks_the_database_password() {
        let r = ProvisionRequest::database("acme", "postgres://u:db-pw-123@h/acme");
        let s = format!("{r:?}");
        assert!(s.contains("acme") && !s.contains("db-pw-123"), "{s}");
    }

    /// The exact input that created a live, unreachable tenant during
    /// QA: a space in the slug, accepted by the console, producing
    /// `host_pattern = "tennant 1.localhost"`.
    #[test]
    fn a_slug_with_a_space_is_refused() {
        let err = validate_slug("tennant 1").expect_err("a space is not a hostname character");
        assert!(err.contains("tennant 1"), "{err}");
        assert!(err.contains("hostname"), "should say why: {err}");
    }

    #[test]
    fn a_slug_is_restricted_to_what_is_legal_in_all_three_uses() {
        for bad in [
            "",          // empty
            "Acme",      // uppercase — not a hostname label
            "ac me",     // space
            "acme_corp", // underscore is legal in a DB name, not a hostname
            "acme.corp", // dot would make it two labels
            "acme;DROP", // punctuation
            "../etc",    // traversal
            "-acme",     // leading hyphen
            "acme-",     // trailing hyphen
        ] {
            assert!(
                validate_slug(bad).is_err(),
                "slug `{bad}` should have been refused"
            );
        }
        for good in ["acme", "acme-2", "a", "tenant-42"] {
            assert!(
                validate_slug(good).is_ok(),
                "slug `{good}` should be allowed"
            );
        }
    }

    /// The two schema names that were actually posted through the
    /// console during QA. Both were accepted, and both became real
    /// Postgres schemas behind live, active tenants.
    ///
    /// `quote_ident` did hold — no `pwned_a` table was created and
    /// `rustango_operators` was still there — so this is not a fix for
    /// an injection that worked. It is a fix for the schema name being
    /// unchecked, which is how an operator ends up with a tenant whose
    /// identifier they cannot type again.
    #[test]
    fn the_schema_names_that_got_through_qa_are_refused() {
        for injected in [
            r#"x"; CREATE TABLE public.pwned_a(i int); --"#,
            r#"y"; DROP TABLE public.rustango_operators; --"#,
        ] {
            let err =
                validate_schema_name(injected).expect_err("an SQL fragment is not a schema name");
            assert!(err.contains("only lowercase"), "{err}");
        }
    }

    #[test]
    fn a_schema_name_is_a_postgres_identifier() {
        for bad in [
            "",                   // empty
            "Acme",               // uppercase would be a *different* schema
            "ac me",              // space
            "acme;DROP",          // punctuation
            "acme.other",         // a dot is schema-qualification
            "-acme",              // leading hyphen
            "pg_toast",           // reserved
            "pg_anything",        // the whole `pg_` namespace is reserved
            "information_schema", // reserved
            "public",             // the registry's schema (#1868)
        ] {
            assert!(
                validate_schema_name(bad).is_err(),
                "schema name `{bad}` should have been refused"
            );
        }
        // Underscores are the difference from the slug rule: legal in
        // an identifier, illegal in a hostname label.
        for good in ["acme", "acme_corp", "tenant-42", "a", "2acme"] {
            assert!(
                validate_schema_name(good).is_ok(),
                "schema name `{good}` should be allowed"
            );
        }
        assert!(validate_schema_name(&"a".repeat(63)).is_ok());
        assert!(
            validate_schema_name(&"a".repeat(64)).is_err(),
            "Postgres would silently truncate it"
        );
    }

    /// The resolver lowercases the `Host` header and strips `:port`
    /// before comparing. Anything that cannot come out of that
    /// normalization can never match.
    #[test]
    fn a_host_pattern_that_could_never_match_is_refused() {
        let err = validate_host_pattern("acme.example.com:8080")
            .expect_err("the port is stripped before matching");
        assert!(
            err.contains("port field"),
            "should say where it goes: {err}"
        );

        let err = validate_host_pattern("*.example.com").expect_err("patterns are matched exactly");
        assert!(err.contains("wildcard"), "{err}");

        for bad in [
            "not a hostname",  // spaces
            "acme..com",       // empty label
            "acme.com.",       // trailing dot leaves an empty label
            "-acme.com",       // leading hyphen in a label
            "acme-.com",       // trailing hyphen in a label
            r#"acme";DROP--"#, // punctuation
        ] {
            assert!(
                validate_host_pattern(bad).is_err(),
                "host pattern `{bad}` should have been refused"
            );
        }
    }

    /// Uppercase is the one case that is normalized instead: the
    /// resolver lowercases the header anyway, so the operator's intent
    /// is not in doubt and refusing would be pedantry.
    #[test]
    fn an_uppercase_host_pattern_is_lowercased_not_refused() {
        assert_eq!(
            validate_host_pattern("Acme.Example.COM").as_deref(),
            Ok("acme.example.com")
        );
    }

    /// `PathPrefixResolver` looks up `"/<first segment>"`. Nothing else
    /// is ever the lookup key.
    #[test]
    fn a_path_prefix_must_be_one_leading_slash_segment() {
        for bad in [
            "acme",             // no leading slash
            "/",                // the apex resolves to no tenant
            "/acme/",           // trailing slash is a second, empty segment
            "/acme/dashboard",  // only the first segment is matched
            "../../etc/passwd", // the traversal string tried in QA
            "/..",              // a relative segment
            "/ac me",           // a space would arrive percent-encoded
        ] {
            assert!(
                validate_path_prefix(bad).is_err(),
                "path prefix `{bad}` should have been refused"
            );
        }
        for good in ["/acme", "/tenant-42", "/a_b", "/v1.0"] {
            assert!(
                validate_path_prefix(good).is_ok(),
                "path prefix `{good}` should be allowed"
            );
        }
    }

    /// `i32` is the column's type, not a port's range. Both of these
    /// parsed and were stored during QA.
    #[test]
    fn a_port_outside_the_tcp_range_is_refused() {
        assert!(validate_port(-1).is_err());
        assert!(validate_port(0).is_err());
        assert!(validate_port(999_999).is_err());
        assert!(validate_port(1).is_ok());
        assert!(validate_port(8080).is_ok());
        assert!(validate_port(65_535).is_ok());
    }

    /// The rules have to be reachable from the one place every caller
    /// goes through, or they are only the console's rules again.
    #[test]
    fn validate_fields_checks_the_slug_derived_schema_name_too() {
        // A slug legal as a hostname label is legal as a schema name,
        // so the default never trips its own rule.
        let mut req = ProvisionRequest::new("acme-2", TenantStorage::Schema(None));
        assert!(validate_fields(&req).is_ok());

        // An explicit one is checked by the same rule.
        req.storage = TenantStorage::Schema(Some(r#"x"; DROP TABLE t; --"#.into()));
        assert!(validate_fields(&req).is_err());
    }

    /// Aliases `Pool::connect` refuses name no backend, so a URL using
    /// one is refused at validation, not at the connection check.
    #[test]
    fn alias_schemes_name_no_backend() {
        for alias in ["pg://u@h/db", "sqlite3:///tmp/a.db", "mariadb://u@h/db"] {
            assert_eq!(Endpoint::scheme_kind(alias), None, "{alias}");
        }
        assert_eq!(
            Endpoint::scheme_kind("PostgreSQL://u@h/db"),
            Some(BackendKind::Postgres)
        );
        assert_eq!(
            Endpoint::scheme_kind("sqlite::memory:"),
            Some(BackendKind::Sqlite)
        );
    }

    #[test]
    fn validate_fields_hands_back_a_normalized_host_pattern() {
        let mut req = ProvisionRequest::database("acme", "postgres://h/tenant");
        req.host_pattern = Some("ACME.Example.com".into());
        let out = validate_fields(&req).expect("a legal hostname");
        assert_eq!(out.host_pattern.as_deref(), Some("acme.example.com"));
    }

    /// The default host pattern uses `[tenancy] apex_domain` too (#2225).
    #[test]
    fn the_default_host_pattern_reads_the_apex_setting() {
        // Under the lock: another test sets RUSTANGO_APEX_DOMAIN while holding it.
        let _g = crate::tenancy::server::APEX_TEST_LOCK.blocking_lock();
        assert!(
            std::env::var("RUSTANGO_APEX_DOMAIN").is_err(),
            "unset RUSTANGO_APEX_DOMAIN to run this test; env wins over the setting"
        );
        crate::tenancy::server::reset_apex_domain_setting();
        crate::tenancy::server::set_apex_domain_setting("apex.test");
        let req = ProvisionRequest::database("acme", "postgres://h/tenant");
        let out = validate_fields(&req);
        crate::tenancy::server::reset_apex_domain_setting();
        assert_eq!(
            out.expect("legal").host_pattern.as_deref(),
            Some("acme.apex.test")
        );
    }

    /// A path-prefix tenant lives on the apex, so it gets no default host.
    #[test]
    fn a_path_prefix_tenant_gets_no_default_host() {
        let mut req = ProvisionRequest::database("acme", "postgres://h/tenant");
        assert_eq!(
            default_host_pattern(&req, Some("apex.test".into())).as_deref(),
            Some("acme.apex.test")
        );
        req.path_prefix = Some("/acme".into());
        assert_eq!(default_host_pattern(&req, Some("apex.test".into())), None);
    }

    /// A prefix the console serves on the apex is refused, naming the clash.
    #[test]
    fn a_prefix_shadowed_by_a_console_route_is_refused() {
        for clash in ["/orgs", "/login", "/__static__", "/db-servers"] {
            let mut req = ProvisionRequest::database("acme", "postgres://h/tenant");
            req.path_prefix = Some(clash.into());
            let err = validate_fields(&req).expect_err(clash);
            assert!(err.contains(clash) && err.contains("console"), "{err}");
        }
        let mut req = ProvisionRequest::database("acme", "postgres://h/tenant");
        req.path_prefix = Some("/t1".into());
        validate_fields(&req).expect("an ordinary prefix");
    }

    #[test]
    fn an_over_long_slug_is_refused_at_the_hostname_limit() {
        assert!(validate_slug(&"a".repeat(63)).is_ok());
        let err = validate_slug(&"a".repeat(64)).expect_err("too long for a hostname label");
        assert!(err.contains("63"), "{err}");
    }

    /// The critical QA finding: a tenant pointed at the registry's own
    /// database ran the tenant migration chain over the registry.
    #[cfg(feature = "postgres")]
    #[test]
    fn a_tenant_url_naming_the_registry_database_is_refused() {
        let registry = "postgres://rustango:rustango@localhost:5432/orgdemo_dev";
        let err = refuse_registry_url(registry, registry).expect_err("same database");
        assert!(err.contains("registry's own database"), "{err}");
        // And it must not echo the password while saying so.
        assert!(!err.contains("rustango:rustango"), "password leaked: {err}");
    }

    /// Different credentials, same database, is still the same
    /// database — which is the case a naive string compare misses.
    #[cfg(feature = "postgres")]
    #[test]
    fn different_credentials_for_the_same_database_are_still_refused() {
        assert!(refuse_registry_url(
            "postgres://someone_else:other@localhost:5432/orgdemo_dev",
            "postgres://rustango:rustango@localhost:5432/orgdemo_dev",
        )
        .is_err());
    }

    /// And a query string does not make it a different database —
    /// sqlite's `?mode=rwc` in particular.
    #[cfg(feature = "sqlite")]
    #[test]
    fn a_query_string_does_not_disguise_the_same_database() {
        assert!(refuse_registry_url(
            "sqlite:///var/app/reg.db?mode=rwc",
            "sqlite:///var/app/reg.db",
        )
        .is_err());
    }

    #[test]
    fn a_genuinely_separate_database_is_allowed() {
        let registry = "postgres://rustango:rustango@localhost:5432/orgdemo_dev";
        for ok in [
            "postgres://rustango:rustango@localhost:5432/acme_tenant", // other db
            "postgres://rustango:rustango@otherhost:5432/orgdemo_dev", // other host
            "postgres://rustango:rustango@localhost:5433/orgdemo_dev", // other port
        ] {
            assert!(
                refuse_registry_url(ok, registry).is_ok(),
                "`{ok}` is a different database and should be allowed"
            );
        }
    }

    /// #1332 asked for the derivation to be exercised on every dialect,
    /// because "same server, another database" is a different shape in
    /// each: a path segment on a server, a sibling file on sqlite.
    #[test]
    fn a_server_url_keeps_everything_but_the_database() {
        assert_eq!(
            tenant_url_on_registry_server(
                "postgres://app:pw@db.internal:5432/orgdemo_dev",
                "tenant_acme"
            )
            .as_deref(),
            Some("postgres://app:pw@db.internal:5432/tenant_acme")
        );
        assert_eq!(
            tenant_url_on_registry_server("mysql://app:pw@db.internal:3306/orgdemo_dev", "t_acme")
                .as_deref(),
            Some("mysql://app:pw@db.internal:3306/t_acme")
        );
    }

    /// #1932: the query is split off before the path, so a `/` inside it
    /// is not the database segment and TLS options carry over.
    #[test]
    fn a_server_url_keeps_its_query_string() {
        for (registry, want) in [
            (
                "postgres://app@db:5432/reg?sslmode=verify-full&sslrootcert=/etc/ssl/ca.pem",
                "postgres://app@db:5432/tenant_acme?sslmode=verify-full&sslrootcert=/etc/ssl/ca.pem",
            ),
            (
                "postgres://app@db:5432/reg?sslmode=require",
                "postgres://app@db:5432/tenant_acme?sslmode=require",
            ),
            (
                "mysql://app@db:3306/reg?ssl-mode=REQUIRED",
                "mysql://app@db:3306/tenant_acme?ssl-mode=REQUIRED",
            ),
        ] {
            assert_eq!(
                tenant_url_on_registry_server(registry, "tenant_acme").as_deref(),
                Some(want),
                "registry `{registry}`"
            );
        }
        assert!(tenant_url_on_registry_server("postgres://db?sslrootcert=/ca.pem", "t").is_none());
    }

    /// sqlx reads `dbname`/`password`/`host` from the query and lets it
    /// win, so a kept non-TLS key would aim tenant migrations elsewhere.
    #[test]
    fn a_server_url_drops_every_non_tls_query_key() {
        for (registry, want) in [
            (
                "postgres://app@db:5432/reg?dbname=reg",
                "postgres://app@db:5432/tenant_acme",
            ),
            (
                "postgres://app@db:5432/reg?password=s3cret&sslmode=require",
                "postgres://app@db:5432/tenant_acme?sslmode=require",
            ),
            (
                "mysql://app@db:3306/reg?ssl-ca=/ca.pem&socket=/tmp/x",
                "mysql://app@db:3306/tenant_acme?ssl-ca=/ca.pem",
            ),
        ] {
            let derived = tenant_url_on_registry_server(registry, "tenant_acme");
            assert_eq!(derived.as_deref(), Some(want), "registry `{registry}`");
            assert!(refuse_registry_url(&derived.unwrap(), registry).is_ok());
        }
    }

    /// A tenant URL whose `dbname=` names the registry is the registry.
    #[cfg(feature = "postgres")]
    #[test]
    fn a_dbname_query_naming_the_registry_is_refused() {
        let registry = "postgres://app:pw@db:5432/reg";
        for tenant in [
            "postgres://other@db:5432/tenant_acme?dbname=reg",
            "postgres://db:5432?dbname=reg",
            "postgresql://db:5432/x?dbname=t&dbname=reg",
        ] {
            assert!(
                refuse_registry_url(tenant, registry).is_err(),
                "`{tenant}` names the registry"
            );
        }
        assert!(refuse_registry_url("postgres://db:5432/reg?dbname=t", registry).is_ok());
    }

    /// sqlite has no server, so the sibling is a file in the registry
    /// file's own directory — including the `./` form the scaffolder
    /// writes into every generated `.env.example`.
    #[test]
    fn a_sqlite_url_derives_a_sibling_file() {
        for (registry, want) in [
            (
                "sqlite://./demo_dev.db?mode=rwc",
                "sqlite://./tenant_acme.db?mode=rwc",
            ),
            (
                "sqlite:///var/app/reg.db",
                "sqlite:///var/app/tenant_acme.db?mode=rwc",
            ),
            // A bare filename is relative to the working directory.
            // This derived nothing at all before, so a registry
            // configured that way fell back to "type a URL yourself".
            ("sqlite://reg.db", "sqlite://./tenant_acme.db?mode=rwc"),
        ] {
            assert_eq!(
                tenant_url_on_registry_server(registry, "tenant_acme").as_deref(),
                Some(want),
                "registry `{registry}`"
            );
        }
    }

    /// An operator naming the database `acme.db` means the file
    /// `acme.db`, not `acme.db.db`.
    #[test]
    fn a_sqlite_database_name_is_not_given_two_extensions() {
        for named in ["acme.db", "acme.DB"] {
            assert_eq!(
                tenant_url_on_registry_server("sqlite://./reg.db", named).as_deref(),
                Some(format!("sqlite://./{named}?mode=rwc")).as_deref(),
                "database name `{named}`"
            );
        }
        // A dot that is not the extension still gets one.
        assert_eq!(
            tenant_url_on_registry_server("sqlite://./reg.db", "acme.v2").as_deref(),
            Some("sqlite://./acme.v2.db?mode=rwc")
        );
    }

    /// Shapes with nowhere to put a tenant. The caller renders the
    /// "supply a full URL instead" escape hatch for these rather than
    /// guessing.
    #[test]
    fn a_url_with_no_database_to_replace_derives_nothing() {
        for undrivable in [
            "sqlite://:memory:",    // no directory to be a sibling of
            "sqlite://",            // no path at all
            "postgres://localhost", // no database segment
            "not a url",            // no scheme
            "postgres:///orgdemo",  // no authority
        ] {
            assert!(
                tenant_url_on_registry_server(undrivable, "tenant_acme").is_none(),
                "`{undrivable}` should not derive a URL"
            );
        }
    }

    /// The derivation and the registry-collision guard have to agree:
    /// a derived URL must never be the one it was derived from.
    #[test]
    fn a_derived_url_is_never_the_registry_itself() {
        for registry in [
            "postgres://app:pw@db.internal:5432/orgdemo_dev",
            "sqlite://./demo_dev.db?mode=rwc",
        ] {
            let derived =
                tenant_url_on_registry_server(registry, "tenant_acme").expect("derivable registry");
            assert!(
                refuse_registry_url(&derived, registry).is_ok(),
                "derived `{derived}` collided with registry `{registry}`"
            );
        }
    }
}

/// #2320: the registry refusal reads a tenant URL the way the pool did.
/// Every URL names its port or shares the pool's default, so `PG*` env can't skew it.
#[cfg(test)]
mod registry_endpoint_tests {
    use super::*;

    /// The registry's driver reads any URL as its own, so another
    /// backend's URL (or none) is refused by scheme.
    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn a_url_of_another_backend_is_refused() {
        let pool = crate::sql::Pool::Sqlite(
            sqlx::SqlitePool::connect_lazy("sqlite::memory:").expect("lazy"),
        );
        for other in ["postgres://u:pw@h/db", "mysql://u:pw@h/db", "/tmp/bare.db"] {
            let err = refuse_other_backend(other, other, &pool).expect_err(other);
            assert!(err.contains("sqlite"), "{err}");
            assert!(!err.contains(":pw@"), "{err}");
        }
        for same in [
            "sqlite:///tmp/t.db?mode=rwc",
            "sqlite::memory:",
            "SQLite:t.db",
        ] {
            assert!(refuse_other_backend(same, same, &pool).is_ok(), "{same}");
        }
    }

    /// A reference resolving to a bare secret: `redact` cannot mask a
    /// value with no scheme, so no error may echo the resolved text.
    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn a_resolved_value_is_never_echoed() {
        let pool = crate::sql::Pool::Sqlite(
            sqlx::SqlitePool::connect_lazy("sqlite::memory:").expect("lazy"),
        );
        for (resolved, reference) in [("hunter2", "env://X"), ("hunter2", "hunter2")] {
            let err = refuse_other_backend(resolved, reference, &pool).expect_err(resolved);
            assert!(!err.contains("hunter2"), "{err}");
        }
        let err = refuse_registry_pool_as("sqlite:/x?mode=hunter2", "env://X", &pool)
            .expect_err("unreadable");
        assert!(!err.contains("hunter2") && err.contains("env://X"), "{err}");
    }

    /// An `env://` reference is checked as the URL it names.
    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn vet_checks_the_resolved_url() {
        struct To(&'static str);
        #[async_trait::async_trait]
        impl SecretsResolver for To {
            async fn resolve(&self, _: &str) -> Result<String, super::super::SecretsError> {
                Ok(self.0.into())
            }
        }
        let pool = crate::sql::Pool::Sqlite(
            sqlx::SqlitePool::connect_lazy("sqlite::memory:").expect("lazy"),
        );
        let to = |url| TenantSecrets::new(To(url));
        let err = vet_database_url("env://T", &to("postgres://u:pw@h/db"), &pool, None)
            .await
            .err()
            .expect("another backend");
        assert!(err.to_string().contains("registry runs on sqlite"), "{err}");
        let ok = vet_database_url("env://T", &to("sqlite:///tmp/t.db"), &pool, None)
            .await
            .expect("same backend");
        assert_eq!(ok.stored, "env://T");
        assert_eq!(ok.resolved, "sqlite:///tmp/t.db");
    }

    #[cfg(feature = "postgres")]
    fn pg(url: &str) -> crate::sql::Pool {
        crate::sql::Pool::Postgres(sqlx::PgPool::connect_lazy(url).expect("lazy"))
    }

    #[cfg(feature = "postgres")]
    #[tokio::test]
    async fn an_empty_host_is_the_same_default_host() {
        let pool = pg("postgres:///reg?user=app");
        assert!(refuse_registry_pool("postgres:///reg", &pool).is_err());
        assert!(refuse_registry_pool("postgres:///reg?user=other&password=pw", &pool).is_err());
        assert!(refuse_registry_pool("postgres:///tenant", &pool).is_ok());
    }

    #[cfg(feature = "postgres")]
    #[tokio::test]
    async fn query_host_port_and_socket_overrides_are_read() {
        let pool = pg("postgres://app@db.internal:5432/reg");
        for same in [
            "postgres://elsewhere:5432/x?host=db.internal&dbname=reg",
            "postgres://db.internal:6000/reg?port=5432",
            "postgres://DB.internal:5432/reg",
        ] {
            assert!(refuse_registry_pool(same, &pool).is_err(), "{same}");
        }
        for other in [
            "postgres://db.internal:5433/reg",
            "postgres://db.internal:5432/tenant",
            "postgres://db.internal:5432/reg?host=elsewhere",
            "mysql://db.internal:3306/reg",
        ] {
            assert!(refuse_registry_pool(other, &pool).is_ok(), "{other}");
        }
        // A PG tenant pool ignores the scheme, so these connect to the registry.
        for same in ["mysql://db.internal:5432/reg", "x://db.internal:5432/reg"] {
            assert!(refuse_registry_pool(same, &pool).is_err(), "{same}");
        }
        let socket = pg("postgres://app@%2Fcloudsql%2Fp:5432/reg");
        assert!(refuse_registry_pool("postgres://x:5432/reg?host=/cloudsql/p", &socket).is_err());
    }

    #[cfg(feature = "mysql")]
    #[tokio::test]
    async fn mariadb_is_mysql() {
        let pool = crate::sql::Pool::Mysql(
            sqlx::MySqlPool::connect_lazy("mysql://app:pw@db:3306/reg").expect("lazy"),
        );
        assert!(refuse_registry_pool("mariadb://x@db:3306/reg", &pool).is_err());
        assert!(refuse_registry_pool("mysql://x@db/reg", &pool).is_err());
        assert!(refuse_registry_pool("mysql://x@db:3306/tenant", &pool).is_ok());
    }

    #[cfg(feature = "postgres")]
    #[tokio::test]
    async fn an_unreadable_url_is_refused() {
        let pool = pg("postgres://app@db.internal:5432/reg");
        for bad in [
            "postgres://db.internal:notaport/reg",
            "postgres://db.internal:5432/reg?port=x",
        ] {
            let err = refuse_registry_pool(bad, &pool).expect_err(bad);
            assert!(err.contains("cannot read"), "{err}");
        }
        assert!(refuse_registry_url("postgres://h:x/reg", "postgres://h:5432/reg").is_err());
    }

    /// `vet_database_url` resolves first; an unresolved reference is not
    /// this check's to refuse.
    #[cfg(feature = "postgres")]
    #[tokio::test]
    async fn a_secret_reference_is_not_refused() {
        let pool = pg("postgres://app@db.internal:5432/reg");
        for reference in [
            "acme-db",
            "env://TENANT_DB",
            "aws-sm://arn:aws:secretsmanager:eu:1:secret:db",
        ] {
            assert!(
                refuse_registry_pool(reference, &pool).is_ok(),
                "{reference}"
            );
        }
    }

    /// A default or `PGHOST` socket dir sits in `host`; `?host=/dir` in `socket`.
    #[cfg(feature = "postgres")]
    #[tokio::test]
    async fn a_socket_dir_matches_in_either_spelling() {
        let opts = sqlx::postgres::PgConnectOptions::new_without_pgpass()
            .host("/tmp")
            .port(5432)
            .database("reg");
        let pool = crate::sql::Pool::Postgres(sqlx::PgPool::connect_lazy_with(opts));
        assert!(refuse_registry_pool("postgres://x:5432/reg?host=/tmp", &pool).is_err());
        assert!(refuse_registry_pool("postgres://x:5432/reg?host=/tmp/", &pool).is_err());
        assert!(refuse_registry_pool("postgres://x:5432/t?host=/tmp", &pool).is_ok());
    }

    #[cfg(feature = "postgres")]
    #[tokio::test]
    async fn a_trailing_dot_or_brackets_name_the_same_host() {
        let pool = pg("postgres://app@db.internal:5432/reg");
        assert!(refuse_registry_pool("postgres://db.internal.:5432/reg", &pool).is_err());
        let v6 = pg("postgres://app@[::1]:5432/reg");
        assert!(refuse_registry_pool("postgres://x:5432/reg?host=::1", &v6).is_err());
    }

    /// Not canonicalized: a Windows canonical path starts `\\?\`, which a URL reads as a query.
    #[cfg(feature = "sqlite")]
    #[tokio::test]
    async fn a_relative_sqlite_path_is_the_same_file() {
        let dir = tempfile::tempdir_in(".").expect("tempdir");
        let cwd = std::env::current_dir().unwrap();
        let abs = cwd.join(dir.path().file_name().unwrap()).join("reg.db");
        std::fs::write(&abs, b"").unwrap();
        let pool = crate::sql::Pool::Sqlite(
            sqlx::SqlitePool::connect_lazy(&format!("sqlite://{}", abs.display())).expect("lazy"),
        );
        let rel = std::path::Path::new(".").join(abs.strip_prefix(&cwd).unwrap());
        let rel = rel.display();
        for same in [format!("sqlite:{rel}"), format!("sqlite://{rel}?mode=rwc")] {
            assert!(refuse_registry_pool(&same, &pool).is_err(), "{same}");
            assert!(refuse_registry_url(&same, &format!("sqlite://{}", abs.display())).is_err());
        }
        let abs_s = abs.display();
        let mut uris = vec![
            format!("sqlite:file:{rel}"),
            format!("sqlite:file:{abs_s}?mode=rwc"),
        ];
        // A Windows path has no leading `/` to follow the authority.
        if cfg!(unix) {
            uris.push(format!("sqlite://file://{abs_s}"));
        }
        for uri in uris {
            assert!(refuse_registry_pool(&uri, &pool).is_err(), "{uri}");
        }
        let sibling = format!("sqlite:{}", dir.path().join("t.db").display());
        assert!(refuse_registry_pool(&sibling, &pool).is_ok());
        // The tenant pool is SQLite too, so a bare path or a foreign scheme opens the file.
        let parent = abs.parent().unwrap().display();
        for same in [
            format!("{abs_s}"),
            format!("{rel}"),
            format!("{parent}/nope/../reg.db"),
            format!("sqlite:{parent}/nope/./../reg.db"),
        ] {
            assert!(refuse_registry_pool(&same, &pool).is_err(), "{same}");
        }
    }

    /// SQLite pops `..` by text, then follows a symlink in any later component.
    #[cfg(all(feature = "sqlite", unix))]
    #[tokio::test]
    async fn a_symlink_after_a_missing_dir_is_followed() {
        let dir = tempfile::tempdir().expect("tempdir");
        let real = dir.path().join("real");
        std::fs::create_dir(&real).unwrap();
        std::fs::write(real.join("reg.db"), b"").unwrap();
        std::os::unix::fs::symlink(&real, dir.path().join("link")).unwrap();
        let pool = crate::sql::Pool::Sqlite(
            sqlx::SqlitePool::connect_lazy(&format!("sqlite://{}", real.join("reg.db").display()))
                .expect("lazy"),
        );
        let d = dir.path().display();
        for same in [
            format!("sqlite:{d}/nope/../link/reg.db"),
            format!("sqlite:{d}/link/nope/../reg.db"),
        ] {
            assert!(refuse_registry_pool(&same, &pool).is_err(), "{same}");
        }
        let other = format!("sqlite:{d}/nope/../link/t.db");
        assert!(refuse_registry_pool(&other, &pool).is_ok());
    }

    /// A link through a missing dir, and a chain past the kernel's limit,
    /// resolve as SQLite resolves them, not by text.
    #[cfg(all(feature = "sqlite", unix))]
    #[tokio::test]
    async fn symlinks_resolve_where_canonicalize_fails() {
        let dir = tempfile::tempdir().expect("tempdir");
        let real = dir.path().join("real");
        std::fs::create_dir(&real).unwrap();
        std::fs::write(real.join("reg.db"), b"").unwrap();
        std::os::unix::fs::symlink("nope/../real", dir.path().join("skip")).unwrap();
        let hops = 60;
        for i in 0..hops {
            let next = if i + 1 == hops {
                "real".to_owned()
            } else {
                format!("l{}", i + 1)
            };
            std::os::unix::fs::symlink(next, dir.path().join(format!("l{i}"))).unwrap();
        }
        let pool = crate::sql::Pool::Sqlite(
            sqlx::SqlitePool::connect_lazy(&format!("sqlite://{}", real.join("reg.db").display()))
                .expect("lazy"),
        );
        let d = dir.path().display();
        for same in [
            format!("sqlite:{d}/skip/reg.db"),
            format!("sqlite:{d}/l0/reg.db"),
        ] {
            assert!(refuse_registry_pool(&same, &pool).is_err(), "{same}");
        }
        assert!(refuse_registry_pool(&format!("sqlite:{d}/l0/t.db"), &pool).is_ok());
    }
}

#[cfg(all(test, feature = "sqlite"))]
mod apex_tests {
    use super::*;

    /// A blank host under `run-server --apex` is `<slug>.<apex>`, whatever
    /// the env var says; an explicit host stays.
    #[tokio::test]
    async fn the_run_server_apex_names_a_blank_host() {
        let _g = crate::tenancy::server::APEX_TEST_LOCK.lock().await;
        std::env::set_var("RUSTANGO_APEX_DOMAIN", "localhost");
        let tmp = tempfile::tempdir().unwrap();
        let url = format!("sqlite://{}?mode=rwc", tmp.path().join("reg.db").display());
        let pool = sqlx::SqlitePool::connect(&url).await.unwrap();
        let pools = std::sync::Arc::new(TenantPools::<sqlx::Sqlite>::new(pool));
        let migrations = tempfile::tempdir().unwrap();
        let mut out = Vec::new();
        super::super::manage::run_with_writer(
            pools.as_ref(),
            &url,
            migrations.path(),
            vec!["migrate-registry".to_owned()],
            &mut out,
        )
        .await
        .unwrap();
        let p = Provisioner::new(pools.clone(), url.clone(), migrations.path())
            .with_default_apex(Some("shop.test".into()))
            .erased();
        let tenant = |slug: &str| {
            let db = tmp.path().join(format!("{slug}.db"));
            ProvisionRequest::new(
                slug,
                TenantStorage::Database(DatabaseSource::Url(format!(
                    "sqlite://{}?mode=rwc",
                    db.display()
                ))),
            )
        };
        let mut own = tenant("own");
        own.host_pattern = Some("own.example.com".into());
        let mut pre = tenant("pre");
        pre.path_prefix = Some("/pre".into());
        for req in [tenant("acme"), own, pre] {
            p.provision(&req, None, None).await.unwrap();
        }
        std::env::remove_var("RUSTANGO_APEX_DOMAIN");
        let hosts: Vec<Option<String>> = crate::query::QuerySet::<super::super::Org>::new()
            .order_by(&[("slug", false)])
            .values_list_flat("host_pattern")
            .fetch(&pools.registry_pool())
            .await
            .unwrap();
        assert_eq!(
            hosts,
            [
                Some("acme.shop.test".to_owned()),
                Some("own.example.com".to_owned()),
                None
            ]
        );
    }
}