rustango 0.60.7

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

/// 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,
    /// Backend driver (v0.33+). Defaults to `Postgres` for
    /// back-compat. `validate_storage_mode` is called against the
    /// `(mode, backend)` pair before the row is inserted — see
    /// [`crate::tenancy::BackendKind::validate_storage_mode`].
    pub backend: crate::tenancy::BackendKind,
    pub display_name: Option<String>,
    pub schema_name: Option<String>,
    pub database_url: Option<String>,
    pub host_pattern: Option<String>,
    pub port: Option<i32>,
    pub path_prefix: Option<String>,
    pub no_migrate: bool,
}

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

/// 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.
///
/// # 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>>,
{
    if find_org(pools, slug).await?.is_some() {
        return Err(TenancyError::Validation(format!(
            "tenant slug `{slug}` already exists"
        )));
    }

    if opts.mode == StorageMode::Database && opts.database_url.is_none() {
        return Err(TenancyError::Validation(
            "create_tenant: mode=Database requires database_url".into(),
        ));
    }
    // v0.33 — reject schema-mode for non-Postgres backends. MySQL +
    // SQLite must use database-mode.
    opts.backend
        .validate_storage_mode(opts.mode)
        .map_err(|msg| TenancyError::Validation(msg.to_owned()))?;

    // The provisioner's own checks: slug, schema, host, prefix, port, and
    // that no other tenant routes on them (#2097).
    let checked = crate::tenancy::provision::checked_request(
        &pools.registry_pool(),
        registry_url,
        &crate::tenancy::provision::ProvisionRequest {
            slug: slug.to_owned(),
            mode: opts.mode,
            backend: opts.backend,
            display_name: opts.display_name.clone(),
            database_url: opts.database_url.clone(),
            schema_name: opts.schema_name.clone(),
            host_pattern: opts.host_pattern.clone(),
            port: opts.port,
            path_prefix: opts.path_prefix.clone(),
            run_migrations: !opts.no_migrate,
            preflight: crate::tenancy::preflight::Preflight::default(),
        },
    )
    .await?;
    let schema_name = crate::tenancy::provision::schema_name_for(&checked);
    let host_pattern = checked.host_pattern;
    let display_name = opts.display_name.clone().unwrap_or_else(|| slug.to_owned());

    if let Some(schema) = schema_name.as_deref() {
        crate::tenancy::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 registry = pools.registry_pool();
    let run_id = match crate::tenancy::provision_store::open_run(
        &registry,
        slug,
        opts.mode.as_str(),
        opts.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: opts.backend.as_str().into(),
        database_url: opts.database_url.clone(),
        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() {
            crate::tenancy::provision::release_schema(pools, schema).await;
        }
        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(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 result =
        migrate_and_activate(pools, registry_url, migrations_dir, &org, !opts.no_migrate).await;
    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,
    )
    .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, then activate it (#2392).
async fn migrate_and_activate<DB: Database>(
    pools: &TenantPools<DB>,
    registry_url: &str,
    migrations_dir: &Path,
    org: &Org,
    migrate: bool,
) -> 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?;
        }
    }
    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);
    }
    let mut user = User {
        id: Auto::default(),
        username: username.to_owned(),
        password_hash: crate::tenancy::password::hash_async(password).await?,
        #[cfg(feature = "sso")]
        email: None,
        is_superuser: superuser,
        active: true,
        created_at: chrono::Utc::now(),
        data: serde_json::json!({}),
        password_changed_at: None,
        sessions_revoked_at: None,
    };
    user.save_pool(&scoped).await?;
    Ok(user)
}

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