rustango 0.60.0

//! Where a provisioning run lives while it happens, and after.
//!
//! Slices [#1318] and [#1320] produce a stream of events. Holding them
//! only in memory fails two ways that matter:
//!
//! **Replay.** [`crate::sse::EventBus`] is a `tokio::sync::broadcast`
//! wrapper, so a subscriber sees only what is sent *after* it connects.
//! Provisioning starts on a POST and the browser connects on the *next*
//! request — the opening steps are gone before anyone is listening, and
//! a mid-run reload loses the rest.
//!
//! **Two pods.** Behind a load balancer the operator watching the
//! stream may not be on the pod doing the work. In-process state means
//! they watch an empty page while provisioning runs fine elsewhere.
//! This is the same problem the resolver solved with a registry
//! fingerprint rather than a shared cache: put it in the database both
//! pods already have.
//!
//! [#1318]: https://github.com/ujeenet/rustango/issues/1318
//! [#1320]: https://github.com/ujeenet/rustango/issues/1320
//!
//! ## Credentials are never stored
//!
//! A run records the URL it was asked to provision, and a connection
//! URL has a password in it. [`crate::sql::connect_diagnosis::redact`]
//! is applied on the way in — not on the way out, so there is no
//! reading path that can forget it. The same goes for event messages,
//! which carry connection diagnoses.

use serde::Serialize;

use crate::core::Column as _;
use crate::sql::{Auto, FetcherPool, Pool, UpdaterPool};

use super::error::TenancyError;

/// Where a provisioning run has got to.
///
/// Stored as a short string rather than an enum column so a state can
/// be added without a migration, and so a human reading the table with
/// `psql` can see what it says.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RunState {
    /// Accepted, not started. The state a webhook returns on.
    Pending,
    Running,
    Succeeded,
    /// Stopped at some step. `ProvisioningRun::error` says which.
    Failed,
}

impl RunState {
    #[must_use]
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Pending => "pending",
            Self::Running => "running",
            Self::Succeeded => "succeeded",
            Self::Failed => "failed",
        }
    }

    /// Parse a stored value. Unknown strings read as `Failed` rather
    /// than panicking: a row written by a newer build should not take
    /// down an older one, and "something went wrong" is the safe
    /// reading of a state this binary does not know.
    #[must_use]
    pub fn parse(raw: &str) -> Self {
        match raw {
            "pending" => Self::Pending,
            "running" => Self::Running,
            "succeeded" => Self::Succeeded,
            _ => Self::Failed,
        }
    }

    /// Nothing more will happen to this run.
    #[must_use]
    pub fn is_terminal(self) -> bool {
        matches!(self, Self::Succeeded | Self::Failed)
    }
}

/// One attempt to stand up a tenant.
/// What a run was for.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RunKind {
    Provision,
    Migrate,
}

impl RunKind {
    #[must_use]
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Provision => "provision",
            Self::Migrate => "migrate",
        }
    }

    /// Unknown values read as `Provision`: every row written before the
    /// column existed was one.
    #[must_use]
    pub fn parse(raw: &str) -> Self {
        match raw {
            "migrate" => Self::Migrate,
            _ => Self::Provision,
        }
    }
}

#[derive(crate::Model, Debug, Clone, Serialize)]
#[rustango(
    table = "rustango_provisioning_runs",
    scope = "registry",
    display = "slug",
    admin(
        list_display = "slug, state, started_at, finished_at",
        ordering = "-started_at",
    )
)]
pub struct ProvisioningRun {
    #[rustango(primary_key)]
    pub id: crate::sql::Auto<i64>,

    /// What the run was: standing a tenant up, or migrating one.
    ///
    /// Both produce the same stream of steps and both are worth keeping,
    /// so they share a table; without this a migrate run reads as a
    /// provisioning run that skipped most of its steps.
    #[rustango(max_length = 16, index, default = "'provision'")]
    pub kind: String,

    /// The tenant this run is for. Not a foreign key: the row is
    /// written *before* the `Org` exists, and must survive the tenant
    /// later being purged — a run is a record of what happened, and an
    /// `ON DELETE CASCADE` would erase exactly the history someone
    /// comes looking for.
    #[rustango(max_length = 100, index)]
    pub slug: String,

    /// Set once the `Org` row lands. `None` while a run is still before
    /// that step, or if it failed before reaching it.
    pub org_id: Option<i64>,

    /// `pending` / `running` / `succeeded` / `failed`. See [`RunState`].
    #[rustango(max_length = 16, index)]
    pub state: String,

    #[rustango(max_length = 16)]
    pub storage_mode: String,

    #[rustango(max_length = 16)]
    pub backend_kind: String,

    /// **Redacted.** The password is replaced before this is stored;
    /// see the module docs.
    #[rustango(max_length = 500)]
    pub database_url: Option<String>,

    /// Who asked for this run — an operator username, a webhook's
    /// caller id, or `cli`. Free text on purpose: the sources are not
    /// one namespace.
    #[rustango(max_length = 150)]
    pub requested_by: Option<String>,

    /// The caller-supplied key that makes a retried request return this
    /// run instead of creating a second tenant. Unique so the database,
    /// not the handler, is what wins a race between two simultaneous
    /// deliveries of one event.
    #[rustango(max_length = 200, unique)]
    pub idempotency_key: Option<String>,

    /// Why it failed, if it did. Already rendered for display.
    pub error: Option<String>,

    #[rustango(auto_now_add)]
    pub started_at: crate::sql::Auto<chrono::DateTime<chrono::Utc>>,

    pub finished_at: Option<chrono::DateTime<chrono::Utc>>,
}

/// One thing that happened during a run.
///
/// The stream a console replays. Deliberately append-only: an event is
/// a fact about the past, and the value of the log is that it is what
/// actually happened rather than a summary someone kept up to date.
#[derive(crate::Model, Debug, Clone, Serialize)]
#[rustango(
    table = "rustango_provisioning_events",
    scope = "registry",
    display = "message",
    admin(list_display = "run_id, seq, step, status", ordering = "run_id, seq")
)]
pub struct ProvisioningEvent {
    #[rustango(primary_key)]
    pub id: crate::sql::Auto<i64>,

    #[rustango(fk = "rustango_provisioning_runs", on = "id", index)]
    pub run_id: i64,

    /// Monotonic within a run, from 1. **This is what `Last-Event-ID`
    /// resumes from**, so it has to be dense and ordered — an
    /// autoincrement `id` would be globally ordered but not per-run,
    /// and a reconnecting client would have nothing to count from.
    pub seq: i64,

    /// Which provisioning step, or `migration` for a forwarded
    /// migration event.
    #[rustango(max_length = 32)]
    pub step: String,

    /// `started` / `ok` / `skipped` / `failed` / `info`.
    #[rustango(max_length = 16)]
    pub status: String,

    /// Human-readable detail. Empty for a bare transition.
    pub message: String,

    #[rustango(auto_now_add)]
    pub at: crate::sql::Auto<chrono::DateTime<chrono::Utc>>,
}

/// Open a run, before any work starts.
///
/// `database_url` is redacted here rather than by the caller: this is
/// the only way into the table, so redacting on the way in means there
/// is no path that can forget.
///
/// # Errors
/// A registry write failure — or a unique-constraint violation on
/// `idempotency_key`, which is the intended way two simultaneous
/// deliveries of one event resolve to one run.
pub async fn open_run(
    registry: &Pool,
    slug: &str,
    storage_mode: &str,
    backend_kind: &str,
    database_url: Option<&str>,
    requested_by: Option<&str>,
    idempotency_key: Option<&str>,
) -> Result<ProvisioningRun, TenancyError> {
    let mut run = ProvisioningRun {
        id: Auto::default(),
        kind: RunKind::Provision.as_str().to_owned(),
        slug: slug.to_owned(),
        org_id: None,
        state: RunState::Running.as_str().to_owned(),
        storage_mode: storage_mode.to_owned(),
        backend_kind: backend_kind.to_owned(),
        database_url: database_url.map(crate::sql::connect_diagnosis::redact),
        requested_by: requested_by.map(ToOwned::to_owned),
        idempotency_key: idempotency_key.map(ToOwned::to_owned),
        error: None,
        started_at: Auto::default(),
        finished_at: None,
    };
    run.insert_pool(registry).await?;
    Ok(run)
}

/// Append an event to a run.
///
/// `seq` is supplied by the caller rather than computed here: the
/// caller is emitting a known sequence and a `SELECT MAX(seq)` per
/// event would be a round-trip per line of a progress stream, plus a
/// race between two writers.
///
/// # Errors
/// A registry write failure.
pub async fn append_event(
    registry: &Pool,
    run_id: i64,
    seq: i64,
    step: &str,
    status: &str,
    message: &str,
) -> Result<(), TenancyError> {
    let mut event = ProvisioningEvent {
        id: Auto::default(),
        run_id,
        seq,
        step: step.to_owned(),
        status: status.to_owned(),
        message: message.to_owned(),
        at: Auto::default(),
    };
    event.insert_pool(registry).await?;
    Ok(())
}

/// Record that the `Org` row now exists, so a run that fails later
/// still says which tenant it half-made.
///
/// # Errors
/// A registry write failure.
pub async fn attach_org(registry: &Pool, run_id: i64, org_id: i64) -> Result<(), TenancyError> {
    ProvisioningRun::objects()
        .where_(ProvisioningRun::id.eq(run_id))
        .update()
        .set("org_id", org_id)
        .execute_pool(registry)
        .await?;
    Ok(())
}

/// Close a run. Only a `running` run changes, so one already closed as
/// interrupted keeps that result (#1883).
///
/// # Errors
/// A registry write failure.
pub async fn finish_run(
    registry: &Pool,
    run_id: i64,
    state: RunState,
    error: Option<&str>,
) -> Result<(), TenancyError> {
    ProvisioningRun::objects()
        .where_(ProvisioningRun::id.eq(run_id))
        .where_(ProvisioningRun::state.eq(RunState::Running.as_str().to_owned()))
        .update()
        .set("state", state.as_str())
        .set("error", error)
        .set("finished_at", chrono::Utc::now())
        .execute_pool(registry)
        .await?;
    Ok(())
}

/// Read a run by id.
///
/// # Errors
/// A registry read failure.
pub async fn run_by_id(
    registry: &Pool,
    run_id: i64,
) -> Result<Option<ProvisioningRun>, TenancyError> {
    Ok(ProvisioningRun::objects()
        .where_(ProvisioningRun::id.eq(run_id))
        .fetch(registry)
        .await?
        .into_iter()
        .next())
}

/// Find the run a caller's idempotency key already opened.
///
/// # Errors
/// A registry read failure.
pub async fn run_by_idempotency_key(
    registry: &Pool,
    key: &str,
) -> Result<Option<ProvisioningRun>, TenancyError> {
    Ok(ProvisioningRun::objects()
        .where_(ProvisioningRun::idempotency_key.eq(Some(key.to_owned())))
        .fetch(registry)
        .await?
        .into_iter()
        .next())
}

/// Replay a run's events, in order, from `after_seq` exclusive.
///
/// `after_seq` of 0 is the whole log — what a fresh subscriber wants.
/// A reconnecting one passes the last `seq` it saw (its
/// `Last-Event-ID`) and gets only what it missed.
///
/// # Errors
/// A registry read failure.
pub async fn events_since(
    registry: &Pool,
    run_id: i64,
    after_seq: i64,
) -> Result<Vec<ProvisioningEvent>, TenancyError> {
    Ok(ProvisioningEvent::objects()
        .where_(ProvisioningEvent::run_id.eq(run_id))
        .where_(ProvisioningEvent::seq.gt(after_seq))
        // Ascending — the bool is `desc`. A replay handed back newest
        // first is not a replay.
        .order_by(&[("seq", false)])
        .fetch(registry)
        .await?)
}

/// Open a run for a migration.
///
/// `slug` names one tenant, or `None` for every active tenant — a batch
/// stores an empty slug, and the events name each tenant as they go.
/// The storage/backend columns stay empty for the same reason: a batch
/// spans tenants that do not agree on either.
///
/// # Errors
/// Driver / insert failures.
pub async fn open_migrate_run(
    registry: &Pool,
    slug: Option<&str>,
    requested_by: Option<&str>,
) -> Result<ProvisioningRun, TenancyError> {
    let mut run = ProvisioningRun {
        id: Auto::default(),
        kind: RunKind::Migrate.as_str().to_owned(),
        slug: slug.unwrap_or_default().to_owned(),
        org_id: None,
        state: RunState::Running.as_str().to_owned(),
        storage_mode: String::new(),
        backend_kind: String::new(),
        database_url: None,
        requested_by: requested_by.map(ToOwned::to_owned),
        idempotency_key: None,
        error: None,
        started_at: Auto::default(),
        finished_at: None,
    };
    run.insert_pool(registry).await?;
    Ok(run)
}

/// The most recent runs, newest first.
///
/// A run is reachable by id and nothing enumerated them, so a console
/// could show a run only while its redirect was still in the address
/// bar — navigate away and the record survived in the table but not in
/// anybody's reach. That is the opposite of why the table is persisted.
///
/// `limit` is a page size rather than a promise to return everything:
/// the table grows with provisioning activity and a console asking for
/// "recent" wants a screenful.
///
/// # Errors
/// Driver / query failures.
pub async fn recent_runs(
    registry: &Pool,
    limit: i64,
    offset: i64,
) -> Result<Vec<ProvisioningRun>, TenancyError> {
    recent_runs_filtered(registry, None, None, limit, offset).await
}

/// As [`recent_runs`], narrowed to one `kind` and/or `state`.
///
/// The filter belongs in the query, not in a `.filter()` on the result:
/// filtering a page of the newest N reports "none" whenever the newest N
/// happen to be a different kind, which is exactly the answer nobody wants
/// while looking for the migration run that failed last week.
///
/// # Errors
/// Driver / query failures.
pub async fn recent_runs_filtered(
    registry: &Pool,
    kind: Option<&str>,
    state: Option<&str>,
    limit: i64,
    offset: i64,
) -> Result<Vec<ProvisioningRun>, TenancyError> {
    let mut q = ProvisioningRun::objects();
    if let Some(k) = kind {
        q = q.where_(ProvisioningRun::kind.eq(k.to_owned()));
    }
    if let Some(s) = state {
        q = q.where_(ProvisioningRun::state.eq(s.to_owned()));
    }
    Ok(q
        // Descending — the bool is `desc`, and "recent" means the run
        // somebody just started is the one at the top.
        .order_by(&[("id", true)])
        .limit(limit)
        .offset(offset)
        .fetch(registry)
        .await?)
}

/// A `running` run with no event for this long is taken to be dead:
/// runs are detached tasks, so a deploy kills them mid-way (#1883).
pub const STALE_RUN_AFTER: std::time::Duration = std::time::Duration::from_secs(3600);

/// The run is `running` and nothing has happened to it for `after`.
async fn is_stale(
    registry: &Pool,
    run: &ProvisioningRun,
    after: std::time::Duration,
) -> Result<bool, TenancyError> {
    if RunState::parse(&run.state) != RunState::Running {
        return Ok(false);
    }
    let Some(id) = run.id.get().copied() else {
        return Ok(false);
    };
    let last_event = ProvisioningEvent::objects()
        .where_(ProvisioningEvent::run_id.eq(id))
        .order_by(&[("seq", true)])
        .limit(1)
        .fetch(registry)
        .await?
        .into_iter()
        .next()
        .and_then(|e| e.at.get().copied());
    let Some(last) = last_event.or_else(|| run.started_at.get().copied()) else {
        return Ok(false);
    };
    let after = chrono::Duration::from_std(after).unwrap_or(chrono::Duration::MAX);
    Ok(chrono::Utc::now() - last > after)
}

/// Close `run` as failed if it is still `running`, and free its
/// idempotency key so a retry can open a new run. Guarded on the state,
/// so a run that finished meanwhile keeps its result.
async fn close_interrupted(registry: &Pool, id: i64) -> Result<bool, TenancyError> {
    let n = ProvisioningRun::objects()
        .where_(ProvisioningRun::id.eq(id))
        .where_(ProvisioningRun::state.eq(RunState::Running.as_str().to_owned()))
        .update()
        .set("state", RunState::Failed.as_str())
        .set(
            "error",
            Some("interrupted: the process running it stopped before it finished"),
        )
        .set("finished_at", chrono::Utc::now())
        .set("idempotency_key", None::<String>)
        .execute_pool(registry)
        .await?;
    Ok(n > 0)
}

/// Close every `running` run with no progress for `after` (see
/// [`STALE_RUN_AFTER`]). Called at boot. Returns how many were closed.
///
/// # Errors
/// A registry read or write failure.
pub async fn reap_stale_runs(
    registry: &Pool,
    after: std::time::Duration,
) -> Result<usize, TenancyError> {
    let running: Vec<ProvisioningRun> = ProvisioningRun::objects()
        .where_(ProvisioningRun::state.eq(RunState::Running.as_str().to_owned()))
        .fetch(registry)
        .await?;
    let mut closed = 0;
    for run in running {
        if is_stale(registry, &run, after).await?
            && close_interrupted(registry, run.id.get().copied().unwrap_or_default()).await?
        {
            closed += 1;
        }
    }
    Ok(closed)
}

/// Whether a delivery carrying `run`'s idempotency key may start over:
/// the run failed, or has been `running` with no event for `stale_after`
/// (closed here). Frees the key.
///
/// # Errors
/// A registry read or write failure.
pub async fn release_for_retry(
    registry: &Pool,
    run: &ProvisioningRun,
    stale_after: std::time::Duration,
) -> Result<bool, TenancyError> {
    let Some(id) = run.id.get().copied() else {
        return Ok(false);
    };
    if is_stale(registry, run, stale_after).await? {
        return close_interrupted(registry, id).await;
    }
    if RunState::parse(&run.state) != RunState::Failed {
        return Ok(false);
    }
    let n = ProvisioningRun::objects()
        .where_(ProvisioningRun::id.eq(id))
        .where_(ProvisioningRun::state.eq(RunState::Failed.as_str().to_owned()))
        .update()
        .set("idempotency_key", None::<String>)
        .execute_pool(registry)
        .await?;
    Ok(n > 0)
}

/// The provision run that created `org_id` did not finish, so a retry
/// may pick the tenant up where it stopped.
///
/// # Errors
/// A registry read failure.
pub async fn org_left_by_failed_run(registry: &Pool, org_id: i64) -> Result<bool, TenancyError> {
    let creator = ProvisioningRun::objects()
        .where_(ProvisioningRun::org_id.eq(Some(org_id)))
        .where_(ProvisioningRun::kind.eq(RunKind::Provision.as_str().to_owned()))
        .order_by(&[("id", false)])
        .limit(1)
        .fetch(registry)
        .await?
        .into_iter()
        .next();
    Ok(creator.is_some_and(|r| RunState::parse(&r.state) == RunState::Failed))
}

/// Delete finished runs older than `cutoff`, and their events.
///
/// Runs are append-only and one per tenant creation, so the table grows
/// with provisioning activity and nothing removes it. Unfinished runs
/// are never pruned regardless of age — one stuck in `running` is
/// evidence of a pod that died mid-provision, which is exactly what
/// someone will come looking for.
///
/// Returns how many runs were removed.
///
/// # Errors
/// A registry write failure.
pub async fn prune_runs(
    registry: &Pool,
    cutoff: chrono::DateTime<chrono::Utc>,
) -> Result<usize, TenancyError> {
    let stale: Vec<ProvisioningRun> = ProvisioningRun::objects()
        .where_(ProvisioningRun::finished_at.lt(Some(cutoff)))
        .fetch(registry)
        .await?;

    let mut removed = 0;
    for run in stale {
        let Some(id) = run.id.get().copied() else {
            continue;
        };
        // Children first: the FK points this way, and a database with
        // it enforced would refuse the parent delete otherwise.
        //
        // Row at a time because the ORM has no queryset-level delete
        // through a `Pool`. Acceptable here — this is a maintenance
        // verb over a couple of dozen events per run, not a request
        // path — but it is why `prune_runs` is not something to call
        // in a loop.
        let events: Vec<ProvisioningEvent> = ProvisioningEvent::objects()
            .where_(ProvisioningEvent::run_id.eq(id))
            .fetch(registry)
            .await?;
        for event in events {
            event.delete_pool(registry).await?;
        }
        run.delete_pool(registry).await?;
        removed += 1;
    }
    Ok(removed)
}