rustango 0.60.0

//! 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.
//!    Before the row lands, so a failed `INSERT` leaves no orphan
//!    schema.
//! 4. [`RegisterOrg`][step] — the `rustango_orgs` row, written
//!    **inactive**.
//! 5. [`Migrate`][step] — this tenant's schema, via
//!    [`tenant_migrate::migrate_one_tenant`].
//! 6. [`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).
//!
//! [`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;

/// 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.
#[derive(Debug, Clone)]
pub struct ProvisionRequest {
    /// Globally unique. Also the default schema name, display name, and
    /// subdomain label.
    pub slug: String,
    pub mode: StorageMode,
    pub backend: BackendKind,
    /// Defaults to the slug.
    pub display_name: Option<String>,
    /// Required in database-mode; meaningless in schema-mode.
    pub database_url: Option<String>,
    /// Schema-mode only; defaults to the slug.
    pub schema_name: Option<String>,
    /// Defaults to `<slug>.<RUSTANGO_APEX_DOMAIN>` when that env var is
    /// set, and to nothing when it is not.
    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,
}

impl ProvisionRequest {
    /// A database-mode request with everything else defaulted.
    #[must_use]
    pub fn database(slug: impl Into<String>, database_url: impl Into<String>) -> Self {
        Self {
            slug: slug.into(),
            mode: StorageMode::Database,
            backend: BackendKind::default(),
            display_name: None,
            database_url: Some(database_url.into()),
            schema_name: None,
            host_pattern: None,
            port: None,
            path_prefix: None,
            run_migrations: true,
            preflight: preflight::Preflight::default(),
        }
    }
}

/// One stage of provisioning, reportable on its own.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
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 in database-mode.
    ProvisionStorage,
    /// 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,
    /// Set `active`, after which the tenant resolves. Last on purpose,
    /// so a half-provisioned tenant never serves a request.
    Activate,
}

/// 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,
}

/// 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,
    /// Next `seq`. Dense and per-run, because `Last-Event-ID` counts
    /// from it.
    seq: std::sync::atomic::AtomicI64,
    buffered: std::sync::Mutex<Vec<(String, String, String)>>,
}

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) -> Self {
        self.store = Some(RunStore {
            registry,
            run_id,
            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: &str) {
        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(
            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: &str, status: &str, message: &str) {
        let Some(store) = &self.store else {
            return;
        };
        if let Ok(mut buf) = store.buffered.lock() {
            buf.push((step.to_owned(), status.to_owned(), message.to_owned()));
        }
    }

    /// 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, status: StepStatus) {
        self.write(step.as_str(), status.as_str(), status.detail())
            .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> {
        self.step(at, StepStatus::Failed(e.to_string())).await;
        Err(e)
    }

    /// A migration event from the tenant's own run. Buffered rather
    /// than written — see the type docs.
    fn migration(&self, event: tenant_migrate::TenantMigrationEvent) {
        self.buffer("migration", "info", &render_migration(&event));
        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::RegisterOrg => "register_org",
            Self::Migrate => "migrate",
            Self::Activate => "activate",
        }
    }
}

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,
        }
    }
}

/// One line describing a migration event, for the stored log.
fn render_migration(event: &tenant_migrate::TenantMigrationEvent) -> String {
    use crate::migrate::{MigrationEvent as M, Outcome};
    use tenant_migrate::{Chain, TenantMigrationEvent as E};

    let chain_tag = |c: Chain| match c {
        Chain::System => "system",
        Chain::Project => "app",
    };
    match event {
        E::Planned { tenants } => format!("migrating {tenants} tenant(s)"),
        E::TenantStarted { slug, .. } => format!("tenant {slug}"),
        E::TenantFinished {
            slug,
            applied,
            error,
            ..
        } => match error {
            Some(e) => format!("tenant {slug} failed: {e}"),
            None => format!("tenant {slug}: {applied} migration(s)"),
        },
        E::Migration { chain, event, .. } => match event {
            M::Planned { total } => format!("{}: {total} pending", chain_tag(*chain)),
            M::Started { name, .. } => format!("{}/{name} started", chain_tag(*chain)),
            M::Finished {
                name,
                outcome,
                elapsed,
                ..
            } => {
                let verb = match outcome {
                    Outcome::Ran => "applied",
                    Outcome::RanPartial { .. } => "applied (partial)",
                    Outcome::Faked => "faked",
                };
                format!(
                    "{verb} {}/{name} ({:.1}s)",
                    chain_tag(*chain),
                    elapsed.as_secs_f64()
                )
            }
            M::Failed { name, error, .. } => {
                format!("{}/{name} failed: {error}", chain_tag(*chain))
            }
        },
    }
}

/// 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>>,
{
    use super::provision_store as store;

    let registry = pools.registry_pool();
    let run = store::open_run(
        &registry,
        &request.slug,
        request.mode.as_str(),
        request.backend.as_str(),
        request.database_url.as_deref(),
        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, false)
        .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);
    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.
    let (state, error) = match &result {
        Ok(outcome) => match &outcome.migrations {
            MigrationsOutcome::Failed(e) => (RunState::Failed, Some(e.clone())),
            _ => (RunState::Succeeded, None),
        },
        Err(e) => (RunState::Failed, Some(e.to_string())),
    };
    // No `attach_org` here: `Reporter::registered` linked the org already.
    if let Err(e) = store::finish_run(&registry, run_id, state, error.as_deref()).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, StepStatus::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;
        }
        rep.step(ProvisionStep::Validate, StepStatus::Ok).await;
        rep.step(
            ProvisionStep::RegisterOrg,
            StepStatus::Skipped("resuming the tenant an earlier run left inactive".into()),
        )
        .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 normalized = match checked_request(&registry, request).await {
        Ok(r) => r,
        Err(e) => return rep.fail(ProvisionStep::Validate, e).await,
    };
    let request = &normalized;

    if request.mode == StorageMode::Database && request.database_url.is_none() {
        return rep
            .fail(
                ProvisionStep::Validate,
                TenancyError::Validation("database mode needs a database URL".into()),
            )
            .await;
    }

    if let Some(url) = &request.database_url {
        if let Err(msg) = refuse_registry_url(url, registry_url) {
            return rep
                .fail(ProvisionStep::Validate, TenancyError::Validation(msg))
                .await;
        }
    }
    rep.step(ProvisionStep::Validate, StepStatus::Ok).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(request, 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. Idempotent via `IF NOT EXISTS`.
    provision_storage(pools, request, schema_name.as_deref(), rep).await?;

    // ---- 4. Register the org ----
    rep.step(ProvisionStep::RegisterOrg, StepStatus::Started)
        .await;
    let mut org = new_org_row(request, schema_name);
    if let Err(e) = org.insert_pool(&registry).await {
        return rep.fail(ProvisionStep::RegisterOrg, e.into()).await;
    }
    // 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, StepStatus::Ok).await;
    rep.notify(ProvisionEvent::Registered { org_id });

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

/// Steps 5 and 6, 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, StepStatus::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.step(ProvisionStep::Migrate, StepStatus::Failed(e.clone()))
                    .await;
            }
            _ => rep.step(ProvisionStep::Migrate, StepStatus::Ok).await,
        }
        outcome
    } else {
        rep.step(
            ProvisionStep::Migrate,
            StepStatus::Skipped("caller asked for no migrations".into()),
        )
        .await;
        MigrationsOutcome::Skipped
    };

    // ---- 6. 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,
            StepStatus::Skipped("migrations failed; the tenant stays inactive".into()),
        )
        .await;
    } else {
        rep.step(ProvisionStep::Activate, StepStatus::Started).await;
        if let Err(e) = activate(&registry, org_id).await {
            return rep.fail(ProvisionStep::Activate, e).await;
        }
        super::invalidate_org_cache();
        rep.step(ProvisionStep::Activate, StepStatus::Ok).await;
    }

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

/// 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(
    request: &ProvisionRequest,
    rep: &Reporter<'_>,
) -> Result<(), TenancyError> {
    let (Some(url), StorageMode::Database) = (&request.database_url, request.mode) else {
        rep.step(
            ProvisionStep::CheckConnection,
            StepStatus::Skipped("schema-mode tenants share the registry's database".into()),
        )
        .await;
        return Ok(());
    };

    rep.step(ProvisionStep::CheckConnection, StepStatus::Started)
        .await;
    match preflight::check(url, &request.preflight).await {
        Ok(_) => {
            rep.step(ProvisionStep::CheckConnection, StepStatus::Ok)
                .await;
            Ok(())
        }
        // `Validation`, not a driver error: the supplied URL is wrong,
        // and the diagnosis already says what to change.
        Err(d) => {
            rep.fail(
                ProvisionStep::CheckConnection,
                TenancyError::Validation(d.to_string()),
            )
            .await
        }
    }
}

/// 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>,
    request: &ProvisionRequest,
    schema_name: Option<&str>,
    rep: &Reporter<'_>,
) -> Result<(), TenancyError> {
    if request.mode != StorageMode::Schema {
        rep.step(
            ProvisionStep::ProvisionStorage,
            StepStatus::Skipped("database-mode tenants bring their own database".into()),
        )
        .await;
        return Ok(());
    }

    rep.step(ProvisionStep::ProvisionStorage, StepStatus::Started)
        .await;
    let schema = schema_name.unwrap_or(&request.slug);
    if let Err(e) = provision_schema(pools, schema).await {
        return rep.fail(ProvisionStep::ProvisionStorage, e).await;
    }
    rep.step(ProvisionStep::ProvisionStorage, StepStatus::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_path_prefix(prefix)?;
    }
    if let Some(port) = request.port {
        validate_port(port)?;
    }

    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(|| {
        std::env::var("RUSTANGO_APEX_DOMAIN")
            .ok()
            .map(|apex| format!("{}.{apex}", request.slug))
    });
    out.host_pattern = pattern.as_deref().map(validate_host_pattern).transpose()?;
    Ok(out)
}

/// 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> {
    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()
        ));
    }
    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"));
    }
    // `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(())
}

/// 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(())
}

/// Every free-text field, then the routing clash check. Returns the
/// request with `host_pattern` normalized; shared with `api::create_tenant` (#2097).
pub(crate) async fn checked_request(
    registry: &crate::sql::Pool,
    request: &ProvisionRequest,
) -> Result<ProvisionRequest, TenancyError> {
    let normalized = validate_fields(request).map_err(TenancyError::Validation)?;
    if let Some(clash) = routing_clash(registry, &normalized).await? {
        return Err(TenancyError::Validation(format!(
            "{clash} is already used by another tenant"
        )));
    }
    Ok(normalized)
}

/// The host, path prefix or port of `request` another tenant routes on.
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};
    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 — scheme, host, port, database — because
/// two URLs may name the same database with different credentials.
pub(crate) fn refuse_registry_url(tenant_url: &str, registry_url: &str) -> Result<(), String> {
    if endpoint_identity(tenant_url) == endpoint_identity(registry_url) {
        return Err(format!(
            "this is the registry's own database ({}). A tenant needs its own — pointing one \
             here would run the tenant migrations over the registry",
            crate::sql::connect_diagnosis::redact(tenant_url)
        ));
    }
    Ok(())
}

/// 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;
    }
    // Only TLS keys carry over: sqlx lets `dbname`/`host`/`user` in the
    // query override the URL, so keeping them could point at the registry.
    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();
    Some(if tls.is_empty() {
        format!("{scheme}://{authority}/{database}")
    } else {
        format!("{scheme}://{authority}/{database}?{}", 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",
];

/// Scheme + host + port + database, lowercased, credentials and query
/// dropped. Two URLs naming the same database compare equal even when
/// they authenticate differently.
fn endpoint_identity(url: &str) -> String {
    let (scheme, rest) = url.split_once("://").unwrap_or_else(|| {
        // `sqlite:path` has no authority — the whole tail is the file.
        url.split_once(':').unwrap_or(("", url))
    });
    // Drop userinfo (everything before the last `@` of the authority).
    let rest = rest.rsplit_once('@').map_or(rest, |(_, after)| after);
    // Drop the query string: `?mode=rwc` does not change which database
    // this is. A Postgres `dbname=` does, and sqlx lets it win over the path.
    let (rest, query) = rest.split_once('?').unwrap_or((rest, ""));
    // sqlx reads `postgresql://` as `postgres://`.
    let scheme = match scheme.to_ascii_lowercase().as_str() {
        "postgresql" => "postgres".to_owned(),
        other => other.to_owned(),
    };
    let dbname = if scheme == "postgres" {
        crate::urls::parse_query_pairs(query)
            .into_iter()
            .rev()
            .find(|(k, _)| k == "dbname")
            .map(|(_, v)| v)
    } else {
        None
    };
    let rest = match dbname {
        Some(db) => {
            let authority = rest.split_once('/').map_or(rest, |(a, _)| a);
            format!("{authority}/{db}")
        }
        None => rest.to_owned(),
    };
    format!("{scheme}://{}", rest.to_ascii_lowercase())
}

/// The schema a schema-mode tenant lives in: whatever the request
/// named, or the slug. `None` in database-mode, which has no schema.
fn schema_name_for(request: &ProvisionRequest) -> Option<String> {
    match request.mode {
        StorageMode::Schema => Some(
            request
                .schema_name
                .clone()
                .unwrap_or_else(|| request.slug.clone()),
        ),
        StorageMode::Database => None,
    }
}

/// The registry row for a validated request, slug-derived defaults
/// filled in.
fn new_org_row(request: &ProvisionRequest, schema_name: Option<String>) -> Org {
    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: request.backend.as_str().into(),
        database_url: request.database_url.clone(),
        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 IF NOT EXISTS` on the registry.
///
/// 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.
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 tenants require a Postgres registry — pass --mode database \
                     on sqlite/mysql"
                        .into(),
                )
            })?;
        // Use the dialect's quoter, not a local copy: it doubles any
        // embedded `"`.
        let sql = format!(
            "CREATE SCHEMA IF NOT EXISTS {}",
            crate::sql::Postgres.quote_ident(schema)
        );
        rustango::sql::sqlx::query(&sql)
            .execute(pg_pools.registry())
            .await?;
        Ok(())
    }
    #[cfg(not(feature = "postgres"))]
    {
        let _ = (pools, schema);
        Err(TenancyError::Validation(
            "schema-mode tenants require the `postgres` feature — pass --mode database \
             on sqlite/mysql builds"
                .into(),
        ))
    }
}

/// Flip the tenant live.
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(())
}

/// 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);

    // One call, no backend branch: `migrate_one_tenant` owns the
    // schema-mode-is-PG-only dispatch that used to be duplicated here.
    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, ()>;

    /// 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;
}

/// 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,
}

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(),
        }
    }

    /// 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)> {
        Box::pin(async move {
            provision_tenant_recorded(
                self.pools.as_ref(),
                &self.registry_url,
                &self.migrations_dir,
                request,
                None,
                requested_by,
                idempotency_key,
            )
            .await
        })
    }

    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,
                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 registry_url(&self) -> String {
        self.registry_url.clone()
    }

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

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

    /// 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::database("acme-2", "postgres://h/tenant");
        req.mode = StorageMode::Schema;
        req.database_url = None;
        assert!(validate_fields(&req).is_ok());

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

    #[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"));
    }

    #[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.
    #[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.
    #[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.
    #[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.
    #[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}`"
            );
        }
    }
}