rustango 0.61.0

//! Typed Rust API for tenancy provisioning.
//!
//! Mirrors the manage-CLI verbs (`create-tenant`, `create-operator`,
//! `create-user`) but with structured argument types and direct return
//! values — for use from `Builder::seed_with` closures and any other
//! in-process caller. CLI-style verb dispatch via
//! [`run_with_writer`](crate::tenancy::manage::run_with_writer) is
//! unchanged for shell consumers.
//!
//! Every function suffixed `_if_missing` is **idempotent** — if a row
//! with the requested key already exists, the existing row is
//! returned without error. This is the canonical seed/bootstrap
//! shape: re-running the demo or a deployment hook does not error.

use std::path::Path;

use sqlx::Database;

use crate::core::Column as _;
use crate::sql::{Auto, FetcherPool};
use crate::tenancy::auth::{Operator, User};
use crate::tenancy::error::TenancyError;
use crate::tenancy::migrate as tenant_migrate;
use crate::tenancy::org::{Org, StorageMode};
use crate::tenancy::pools::TenantPools;
use crate::tenancy::provision::{self, DatabaseSource, TenantStorage};

/// Inputs for [`create_tenant`] / [`create_tenant_if_missing`].
///
/// Defaults: `mode = Schema`, all `Option`s `None`, `no_migrate =
/// false`. When `host_pattern` is `None`, the default is
/// `"{slug}.{RUSTANGO_APEX_DOMAIN}"` (read from env at provisioning
/// time) — same convention the CLI verb uses.
pub struct CreateTenantOpts {
    pub mode: StorageMode,
    /// `None` takes the registry's; anything else must equal it.
    pub backend: Option<crate::tenancy::BackendKind>,
    pub display_name: Option<String>,
    pub schema_name: Option<String>,
    /// An existing database. Database mode needs this or `database_server`.
    pub database_url: Option<String>,
    /// Create the database, named as the slug, on this saved server.
    pub database_server: Option<String>,
    pub host_pattern: Option<String>,
    pub port: Option<i32>,
    pub path_prefix: Option<String>,
    pub no_migrate: bool,
    /// A superuser created before the tenant goes live. Issue their
    /// set-password link with [`crate::tenancy::first_admin::hand_over_user`].
    pub first_admin: Option<crate::tenancy::FirstAdmin>,
}

impl Default for CreateTenantOpts {
    fn default() -> Self {
        Self {
            mode: StorageMode::Schema,
            backend: None,
            display_name: None,
            schema_name: None,
            database_url: None,
            database_server: None,
            host_pattern: None,
            port: None,
            path_prefix: None,
            no_migrate: false,
            first_admin: None,
        }
    }
}

impl CreateTenantOpts {
    /// The one storage these options name.
    fn storage(&self) -> Result<TenantStorage, TenancyError> {
        let bad = |m: &str| Err(TenancyError::Validation(format!("create_tenant: {m}")));
        match (self.mode, &self.database_url, &self.database_server) {
            (StorageMode::Schema, None, None) => {
                Ok(TenantStorage::Schema(self.schema_name.clone()))
            }
            (StorageMode::Schema, _, _) => bad("schema mode takes no database URL or server"),
            (StorageMode::Database, Some(url), None) => {
                Ok(TenantStorage::Database(DatabaseSource::Url(url.clone())))
            }
            (StorageMode::Database, None, Some(server)) => Ok(TenantStorage::Database(
                DatabaseSource::Server(server.clone()),
            )),
            (StorageMode::Database, Some(_), Some(_)) => {
                bad("give database_url or database_server, not both")
            }
            (StorageMode::Database, None, None) => {
                bad("mode=Database requires database_url or database_server")
            }
        }
    }
}

/// Look up an org by slug. Returns `Ok(None)` for "no such tenant" —
/// not an error.
///
/// # Errors
/// Driver / SQL failures during the lookup.
pub async fn find_org<DB: Database>(
    pools: &TenantPools<DB>,
    slug: &str,
) -> Result<Option<Org>, TenancyError>
where
    crate::sql::Pool: From<sqlx::Pool<DB>>,
{
    let mut rows: Vec<Org> = Org::objects()
        .where_(Org::slug.eq(slug.to_owned()))
        .fetch(&pools.registry_pool())
        .await?;
    Ok(rows.pop())
}

/// Provision a new tenant and run tenant-scoped migrations against it.
/// Errors if a tenant with this slug already exists. On a saved server,
/// a failure after the create drops the new database, login and row.
///
/// # Errors
/// * `TenancyError::Validation` for duplicate slug or bad
///   schema/database options.
/// * Driver / SQL failures during schema creation, the Org INSERT, or
///   the tenant migration pass. A failed migration leaves the Org inactive.
pub async fn create_tenant<DB: Database>(
    pools: &TenantPools<DB>,
    registry_url: &str,
    migrations_dir: &Path,
    slug: &str,
    opts: CreateTenantOpts,
) -> Result<Org, TenancyError>
where
    crate::sql::Pool: From<sqlx::Pool<DB>>,
{
    // Detached: a dropped caller must not stop a create between its DDL
    // and the rows that record it.
    let pools = pools.fresh();
    let (url, dir, slug) = (
        registry_url.to_owned(),
        migrations_dir.to_owned(),
        slug.to_owned(),
    );
    tokio::spawn(async move { create_tenant_inner(&pools, &url, &dir, &slug, opts).await })
        .await
        .map_err(|e| TenancyError::Validation(format!("the create task stopped: {e}")))?
}

async fn create_tenant_inner<DB: Database>(
    pools: &TenantPools<DB>,
    registry_url: &str,
    migrations_dir: &Path,
    slug: &str,
    opts: CreateTenantOpts,
) -> Result<Org, TenancyError>
where
    crate::sql::Pool: From<sqlx::Pool<DB>>,
{
    if find_org(pools, slug).await?.is_some() {
        return Err(TenancyError::Validation(format!(
            "tenant slug `{slug}` already exists"
        )));
    }

    let storage = opts.storage()?;

    // The provisioner's own checks: fields, backend, the database URL or
    // server, routing clashes, then the same preflight, before anything
    // is written.
    let registry = pools.registry_pool();
    let checked = provision::checked_request(
        &registry,
        pools.secrets(),
        registry_url,
        &provision::ProvisionRequest {
            backend: opts.backend,
            display_name: opts.display_name.clone(),
            host_pattern: opts.host_pattern.clone(),
            port: opts.port,
            path_prefix: opts.path_prefix.clone(),
            run_migrations: !opts.no_migrate,
            first_admin: opts.first_admin.clone(),
            ..provision::ProvisionRequest::new(slug, storage)
        },
    )
    .await?;
    // No observer: steps go nowhere, failures come back as `Err`.
    let rep = provision::Reporter::new(None);
    let mut created = match checked.server() {
        Some(server) => {
            let (admin, _) = provision::ServerDatabase::open(server, &registry).await?;
            let db =
                provision::ServerDatabase::create(admin, server, slug, &registry, &rep).await?;
            if let Err(e) = db.check_login(&rep).await {
                let undone = db.roll_back(pools, None, &rep).await;
                return Err(provision::with_undone(e, &undone));
            }
            Some(db)
        }
        None => {
            checked.preflight().await?;
            None
        }
    };
    let database_url = match created.take() {
        Some(db) => match db
            .stored_url(&registry, pools.secrets(), registry_url)
            .await
        {
            Ok(stored) => {
                created = Some(db);
                Some(stored)
            }
            Err(e) => {
                let undone = db.roll_back(pools, None, &rep).await;
                return Err(provision::with_undone(e, &undone));
            }
        },
        None => checked.stored_database_url(),
    };
    let schema_name = provision::schema_name_for(&checked.request);
    let host_pattern = checked.request.host_pattern.clone();
    let display_name = opts.display_name.clone().unwrap_or_else(|| slug.to_owned());

    if let Some(schema) = schema_name.as_deref() {
        provision::provision_schema(pools, schema).await?;
    }

    // A run record lets `create_tenant_if_missing` tell a failed create
    // from a suspended tenant. Best effort: bookkeeping must not block a create.
    let run_id = match crate::tenancy::provision_store::open_run(
        &registry,
        slug,
        opts.mode.as_str(),
        checked.backend.as_str(),
        opts.database_url.as_deref(),
        None,
        None,
    )
    .await
    {
        Ok(run) => run.id.get().copied(),
        Err(e) => {
            tracing::warn!(target: "rustango::tenancy::provision", error = %e, "could not record the create run");
            None
        }
    };

    let mut org = Org {
        id: Auto::default(),
        slug: slug.to_owned(),
        display_name,
        storage_mode: opts.mode.as_str().into(),
        backend_kind: checked.backend.as_str().into(),
        database_url,
        schema_name,
        host_pattern,
        port: opts.port,
        path_prefix: opts.path_prefix.clone(),
        // Activated only once its migrations applied (#2392).
        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,
    };
    use crate::tenancy::provision_store::{self as store, RunState, RunText};
    if let Err(e) = crate::tenancy::org_host::insert_org(&registry, &mut org).await {
        if let Some(schema) = org.schema_name.as_deref() {
            provision::release_schema(pools, schema).await;
        }
        let e = match created {
            Some(db) => provision::with_undone(e, &db.roll_back(pools, None, &rep).await),
            None => e,
        };
        if let Some(run) = run_id {
            let text =
                RunText::user_facing(&e).unwrap_or(RunText::fixed("registering the tenant failed"));
            let _ = store::finish_run_text(&registry, run, RunState::Failed, Some(&text)).await;
        }
        return Err(e);
    }
    let org_id = org.id.get().copied().unwrap_or_default();
    if let Some(db) = created.as_mut() {
        db.attach(&registry, org_id).await;
    }
    if let Some(run) = run_id {
        if let Err(e) = store::attach_org(&registry, run, org_id).await {
            tracing::warn!(target: "rustango::tenancy::provision", error = %e, "could not attach org id to run");
        }
    }

    let mut result = migrate_and_activate(
        pools,
        registry_url,
        migrations_dir,
        &org,
        !opts.no_migrate,
        opts.first_admin.as_ref(),
    )
    .await;
    if let Some(db) = created {
        match result {
            Ok(()) => db.keep().await,
            Err(e) => {
                let undone = db.roll_back(pools, Some(&org), &rep).await;
                result = Err(provision::with_undone(e, &undone));
            }
        }
    }
    if let Some(run) = run_id {
        let (state, text) = match &result {
            Ok(()) => (RunState::Succeeded, None),
            Err(e) => (
                RunState::Failed,
                Some(RunText::user_facing(e).unwrap_or(RunText::fixed("tenant migrations failed"))),
            ),
        };
        if let Err(e) = store::finish_run_text(&registry, run, state, text.as_ref()).await {
            tracing::warn!(target: "rustango::tenancy::provision", error = %e, "could not close the create run");
        }
    }
    result?;

    // Re-fetch so the returned Org has whatever the runner / triggers
    // wrote back; cheap, and lets callers chain `pools.acquire(&org)`
    // immediately.
    find_org(pools, slug).await?.ok_or_else(|| {
        TenancyError::Validation(format!(
            "create_tenant: tenant `{slug}` vanished after INSERT"
        ))
    })
}

/// Idempotent variant of [`create_tenant`]: returns the existing org
/// if one already has this slug, otherwise creates + returns the new
/// one. The seed/bootstrap default. An org whose create failed at
/// migrate is migrated and activated instead.
///
/// # Errors
/// Driver / SQL failures, or [`create_tenant`]'s validation errors
/// (mode/url mismatch).
pub async fn create_tenant_if_missing<DB: Database>(
    pools: &TenantPools<DB>,
    registry_url: &str,
    migrations_dir: &Path,
    slug: &str,
    opts: CreateTenantOpts,
) -> Result<Org, TenancyError>
where
    crate::sql::Pool: From<sqlx::Pool<DB>>,
{
    let Some(existing) = find_org(pools, slug).await? else {
        return create_tenant(pools, registry_url, migrations_dir, slug, opts).await;
    };
    // Finish a create whose migrations failed; a suspended tenant stays as is.
    let registry = pools.registry_pool();
    let org_id = existing.id.get().copied().unwrap_or_default();
    if existing.active
        || !crate::tenancy::provision_store::org_left_by_failed_run(&registry, org_id).await?
    {
        return Ok(existing);
    }
    migrate_and_activate(
        pools,
        registry_url,
        migrations_dir,
        &existing,
        !opts.no_migrate,
        opts.first_admin.as_ref(),
    )
    .await?;
    let forget = crate::tenancy::provision_store::forget_unsucceeded_runs(org_id)?;
    crate::sql::update_pool(&registry, &forget).await?;
    find_org(pools, slug)
        .await?
        .ok_or_else(|| TenancyError::Validation(format!("create_tenant: tenant `{slug}` vanished")))
}

/// Migrate `org` alone, create its first admin, then activate it.
async fn migrate_and_activate<DB: Database>(
    pools: &TenantPools<DB>,
    registry_url: &str,
    migrations_dir: &Path,
    org: &Org,
    migrate: bool,
    first_admin: Option<&crate::tenancy::FirstAdmin>,
) -> Result<(), TenancyError>
where
    crate::sql::Pool: From<sqlx::Pool<DB>>,
{
    if migrate {
        for dir in resolve_migration_dirs(migrations_dir) {
            tenant_migrate::migrate_one_tenant(pools, org, &dir, registry_url, None).await?;
        }
    }
    if let Some(admin) = first_admin {
        crate::tenancy::first_admin::create(pools, org, admin).await?;
    }
    let org_id = org.id.get().copied().unwrap_or_default();
    crate::tenancy::provision::activate(&pools.registry_pool(), org_id).await?;
    crate::tenancy::invalidate_org_cache();
    Ok(())
}

/// Idempotent variant of `create-operator`: creates the operator if
/// no row with this username exists, else returns the existing one.
///
/// # Errors
/// Driver / SQL failures or password-hash failures.
pub async fn create_operator_if_missing<DB: Database>(
    pools: &TenantPools<DB>,
    username: &str,
    password: &str,
) -> Result<Operator, TenancyError>
where
    crate::sql::Pool: From<sqlx::Pool<DB>>,
{
    let registry = pools.registry_pool();
    let mut existing: Vec<Operator> = Operator::objects()
        .where_(Operator::username.eq(username.to_owned()))
        .fetch(&registry)
        .await?;
    if let Some(op) = existing.pop() {
        return Ok(op);
    }
    let mut op = Operator {
        id: Auto::default(),
        username: username.to_owned(),
        password_hash: crate::tenancy::password::hash_async(password).await?,
        active: true,
        created_at: chrono::Utc::now(),
        password_changed_at: None,
        sessions_revoked_at: None,
    };
    op.insert_pool(&registry).await?;
    Ok(op)
}

/// Idempotent variant of `create-user`: creates a tenant-scoped user
/// if no row with this username exists in the tenant's schema, else
/// returns the existing one.
///
/// # Errors
/// * `TenancyError::Validation` if the tenant slug is unknown.
/// * Driver / SQL failures.
pub async fn create_user_if_missing<DB: Database>(
    pools: &TenantPools<DB>,
    slug: &str,
    username: &str,
    password: &str,
    superuser: bool,
) -> Result<User, TenancyError>
where
    crate::sql::Pool: From<sqlx::Pool<DB>>,
{
    let org = find_org(pools, slug)
        .await?
        .ok_or_else(|| TenancyError::Validation(format!("no tenant with slug `{slug}`")))?;

    // v0.38 — route through the backend-agnostic
    // `TenantPools::scoped_pool_dyn` (schema-mode-aware on PG, db-mode
    // elsewhere) + ORM `_pool` family so the same code runs on every
    // dialect.
    let scoped = pools.scoped_pool_dyn(&org).await?;

    let mut existing: Vec<User> = User::objects()
        .where_(User::username.eq(username.to_owned()))
        .fetch(&scoped)
        .await?;
    if let Some(u) = existing.pop() {
        return Ok(u);
    }
    crate::tenancy::auth::NewUser {
        username,
        password_hash: crate::tenancy::password::hash_async(password).await?,
        is_superuser: superuser,
        email: None,
    }
    .insert(&scoped)
    .await
}

/// Resolve the path the caller passed to [`create_tenant`] /
/// [`create_tenant_if_missing`] into the actual list of migrations
/// directories to run against. Mirrors the auto-detect in
/// [`crate::server::Builder::migrate`] so the typed API and the
/// builder accept the same shape.
///
/// Cases:
/// * `dir` itself contains `*.json` migrations — treat it as a flat
///   migrations directory and run against it directly.
/// * `dir` contains a `migrations/` subdir or per-app `<app>/migrations/`
///   subdirs — treat `dir` as the project root and discover.
/// * Anything else — return `[dir]` and let the runner produce its
///   normal "no migrations to apply" outcome rather than silently
///   pretending the apply happened.
fn resolve_migration_dirs(dir: &Path) -> Vec<std::path::PathBuf> {
    let dirs = crate::migrate::file::discover_migration_dirs(dir);
    if !dirs.is_empty() {
        return dirs;
    }
    if dir_has_json_files(dir) {
        return vec![dir.to_path_buf()];
    }
    vec![dir.to_path_buf()]
}

fn dir_has_json_files(dir: &Path) -> bool {
    std::fs::read_dir(dir)
        .ok()
        .into_iter()
        .flatten()
        .filter_map(Result::ok)
        .any(|e| e.path().extension().and_then(|s| s.to_str()) == Some("json"))
}