rustango 0.60.7

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

/// Text an operator may read in a run log: the `message` of an event or
/// a run's `error` (#2212).
///
/// Built only from fixed text, an operator-facing [`TenancyError`], or a
/// cause [`RunLog`] logged and withheld, so driver text cannot be stored.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct RunText(String);

impl RunText {
    /// Fixed text, such as a step's reason for skipping.
    #[must_use]
    pub fn fixed(text: &'static str) -> Self {
        Self(text.to_owned())
    }

    /// A validation (or busy) error's own text; `None` for a cause to withhold.
    #[must_use]
    pub fn user_facing(e: &TenancyError) -> Option<Self> {
        e.user_facing().map(Self)
    }

    #[must_use]
    pub fn as_str(&self) -> &str {
        &self.0
    }
}

impl std::fmt::Display for RunText {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(&self.0)
    }
}

/// Renders a run's log lines and withholds failure causes. A cause is
/// logged once per tenant, under a reference the stored text repeats.
pub(crate) struct RunLog {
    run_id: i64,
    next_ref: std::sync::atomic::AtomicU32,
    /// Tenant slug to the reference its first failure was logged under.
    refs: std::sync::Mutex<std::collections::HashMap<String, String>>,
}

impl RunLog {
    pub(crate) fn new(run_id: i64) -> Self {
        Self {
            run_id,
            next_ref: std::sync::atomic::AtomicU32::new(1),
            refs: std::sync::Mutex::new(std::collections::HashMap::new()),
        }
    }

    /// `e`'s own text if the operator can act on it, else `what` with the cause withheld.
    pub(crate) fn failure(&self, slug: &str, what: &'static str, e: &TenancyError) -> RunText {
        RunText::user_facing(e).unwrap_or_else(|| self.withhold(slug, what.to_owned(), e))
    }

    /// `what` and a reference; `cause` is logged under it, once per tenant.
    pub(crate) fn withheld(
        &self,
        slug: &str,
        what: &'static str,
        cause: &dyn std::fmt::Display,
    ) -> RunText {
        self.withhold(slug, what.to_owned(), cause)
    }

    fn withhold(&self, slug: &str, what: String, cause: &dyn std::fmt::Display) -> RunText {
        let mut refs = self
            .refs
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        if let Some(reference) = refs.get(slug) {
            return RunText(format!("{what} (ref {reference})"));
        }
        let n = self
            .next_ref
            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        let reference = format!("{}.{n}", self.run_id);
        let body = tracing::error_span!(
            "provisioning_run",
            org = slug,
            run_id = self.run_id,
            reference = %reference
        )
        .in_scope(|| crate::error::server_error_body("tenancy::provisioning_run", cause));
        refs.insert(slug.to_owned(), reference.clone());
        RunText(format!("{what} ({body}; ref {reference})"))
    }

    /// The one renderer of a migration event: `(step, status, text)`.
    /// Error text never reaches `text`; it goes through [`Self::withhold`].
    pub(crate) fn migration(
        &self,
        event: &super::migrate::TenantMigrationEvent,
    ) -> (&'static str, &'static str, RunText) {
        use super::migrate::{Chain, TenantMigrationEvent as T};
        use crate::migrate::{MigrationEvent as M, Outcome};
        let chain_name = |c: Chain| match c {
            Chain::System => "system",
            Chain::Project => "app",
        };
        match event {
            T::Planned { tenants } => (
                "plan",
                "info",
                RunText(format!("{tenants} tenant(s) to migrate")),
            ),
            T::TenantStarted { slug, index, total } => (
                "tenant",
                "started",
                RunText(format!("{slug} ({index}/{total})")),
            ),
            T::Migration { slug, chain, event } => {
                let chain = chain_name(*chain);
                match event {
                    M::Planned { total } => (
                        "migration",
                        "info",
                        RunText(format!("{slug} {chain}: {total} pending")),
                    ),
                    M::Started { name, .. } => (
                        "migration",
                        "info",
                        RunText(format!("{slug} {chain}: {name} started")),
                    ),
                    M::Finished {
                        name,
                        outcome,
                        elapsed,
                        ..
                    } => {
                        let verb = match outcome {
                            Outcome::Ran => "applied",
                            Outcome::RanPartial { .. } => "applied (partial)",
                            Outcome::Faked => "faked",
                        };
                        (
                            "migration",
                            "info",
                            RunText(format!(
                                "{slug} {chain}: {verb} {name} ({:.1}s)",
                                elapsed.as_secs_f64()
                            )),
                        )
                    }
                    M::Failed { name, error, .. } => (
                        "migration",
                        "failed",
                        self.withhold(slug, format!("{slug} {chain}: {name} failed"), error),
                    ),
                }
            }
            T::TenantFinished {
                slug,
                applied,
                error,
                ..
            } => match error {
                None => (
                    "tenant",
                    "ok",
                    RunText(format!("{slug}: {applied} applied")),
                ),
                Some(e) => (
                    "tenant",
                    "failed",
                    self.withhold(slug, format!("{slug} failed"), e),
                ),
            },
        }
    }
}

/// 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_text(
    registry: &Pool,
    run_id: i64,
    seq: i64,
    step: &str,
    status: &str,
    message: &RunText,
) -> Result<(), TenancyError> {
    let mut event = ProvisioningEvent {
        id: Auto::default(),
        run_id,
        seq,
        step: step.to_owned(),
        status: status.to_owned(),
        message: message.as_str().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(())
}

/// [`append_event_text`] with untyped text, stored as given.
///
/// # Errors
/// A registry write failure.
#[deprecated(note = "use append_event_text / finish_run_text")]
pub async fn append_event(
    registry: &Pool,
    run_id: i64,
    seq: i64,
    step: &str,
    status: &str,
    message: &str,
) -> Result<(), TenancyError> {
    append_event_text(
        registry,
        run_id,
        seq,
        step,
        status,
        &RunText(message.to_owned()),
    )
    .await
}

/// [`finish_run_text`] with untyped text, stored as given.
///
/// # Errors
/// A registry write failure.
#[deprecated(note = "use append_event_text / finish_run_text")]
pub async fn finish_run(
    registry: &Pool,
    run_id: i64,
    state: RunState,
    error: Option<&str>,
) -> Result<(), TenancyError> {
    let error = error.map(|e| RunText(e.to_owned()));
    finish_run_text(registry, run_id, state, error.as_ref()).await
}

/// 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_text(
    registry: &Pool,
    run_id: i64,
    state: RunState,
    error: Option<&RunText>,
) -> 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.map(RunText::as_str))
        .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. An operator edit of an active org,
/// an activation or a deactivation unlinks the run.
///
/// # 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))
}

/// Unlink `org_id` from every provision run that did not succeed: a running
/// one can still end failed. Called once the org is activated, so a retry
/// never resumes it (#2292).
pub(crate) fn forget_unsucceeded_runs(
    org_id: i64,
) -> Result<crate::core::UpdateQuery, crate::sql::ExecError> {
    ProvisioningRun::objects()
        .where_(ProvisioningRun::org_id.eq(Some(org_id)))
        .where_(ProvisioningRun::kind.eq(RunKind::Provision.as_str().to_owned()))
        .where_(ProvisioningRun::state.ne(RunState::Succeeded.as_str().to_owned()))
        .update()
        .set("org_id", None::<i64>)
        .compile()
        .map_err(crate::sql::ExecError::from)
}

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