orion-server 1.11.1

//! Database abstraction: pool construction, backend detection, migrations,
//! row models and repositories.
//!
//! ## Why there are no foreign keys (D26)
//!
//! `grep REFERENCES migrations/` returns nothing, on any backend, and that is
//! deliberate rather than an oversight:
//!
//! - **`workflows` and `channels` are version-keyed.** Both have a composite
//!   primary key `(id, version)`, and `channels.workflow_id` references a
//!   workflow *id*, never a specific version. There is no key to point an FK
//!   at, and adding one to `(workflow_id, version)` would freeze a channel to
//!   the workflow version current when it was authored — the opposite of the
//!   rollout model.
//! - **`traces` outlive what they reference.** A trace records what ran, so it
//!   must survive the deletion of its channel or workflow. An FK would either
//!   block that delete or cascade away the audit record.
//! - **`audit_logs.resource_id` is polymorphic** — it holds a workflow,
//!   channel, connector or breaker id depending on `resource_type`, which no
//!   single FK can express.
//! - **`trace_dlq.trace_id`** is the one plausible FK. It is omitted so DLQ
//!   rows survive trace retention cleanup, which is exactly when they matter.
//!
//! Referential integrity is enforced in the repositories instead. SQLite is
//! still opened with `PRAGMA foreign_keys = ON` so a future FK takes effect
//! immediately; see `init_sqlite_pool`.

pub mod config_encryption;
pub mod content;
pub mod models;
pub mod repositories;
pub mod schema;

use std::time::Duration;

use sea_query_sqlx::SqlxBinder;
// sqlx 0.9 will not take a non-`'static` query string unwrapped, so that
// assembling SQL by concatenation has to be stated rather than assumed. Every
// `sql` reaching these helpers is rendered by a `sea-query` builder — the
// binder emits placeholders and hands the values back separately in the
// `SqlxValues` beside it — so no caller-supplied value is ever in the text.
use sqlx::AssertSqlSafe;

use crate::config::StorageConfig;
use crate::errors::OrionError;

// ============================================================
// Database backend detection — determined at runtime from URL
// ============================================================

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DbBackend {
    Sqlite,
    Postgres,
    Mysql,
}

impl std::fmt::Display for DbBackend {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Sqlite => write!(f, "sqlite"),
            Self::Postgres => write!(f, "postgres"),
            Self::Mysql => write!(f, "mysql"),
        }
    }
}

/// Detect the database backend from a connection URL.
pub fn detect_backend(url: &str) -> Result<DbBackend, OrionError> {
    if url.starts_with("sqlite:") || url.starts_with("file:") {
        Ok(DbBackend::Sqlite)
    } else if url.starts_with("postgres://") || url.starts_with("postgresql://") {
        Ok(DbBackend::Postgres)
    } else if url.starts_with("mysql://") || url.starts_with("mariadb://") {
        Ok(DbBackend::Mysql)
    } else {
        Err(OrionError::Config {
            message: format!(
                "Unsupported database URL scheme: {url}. Expected sqlite:, postgres://, or mysql://"
            ),
        })
    }
}

// ============================================================
// DbPool — enum wrapping concrete pool types
// ============================================================

/// A type [`DbPool`] and [`DbTransaction`] can decode a row into on any of the
/// three backends.
///
/// `DbPool` dispatches over three concrete sqlx pools, so every typed fetch
/// needs `FromRow` for all three row types plus `Send + Unpin`. Spelled out,
/// that is a five-line `where` clause; it was repeated in eight signatures
/// across this file and `repositories/helpers.rs` (D19). The blanket impl
/// below means nothing has to implement this explicitly — deriving
/// `sqlx::FromRow` on a struct in `models::rows` is enough.
pub trait DbRow:
    for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
    + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>
    + for<'r> sqlx::FromRow<'r, sqlx::mysql::MySqlRow>
    + Send
    + Unpin
{
}

impl<T> DbRow for T where
    T: for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>
        + for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>
        + for<'r> sqlx::FromRow<'r, sqlx::mysql::MySqlRow>
        + Send
        + Unpin
{
}

/// Database connection pool that wraps the concrete sqlx pool type,
/// selected at runtime based on the connection URL.
#[derive(Clone)]
pub enum DbPool {
    Sqlite(sqlx::SqlitePool),
    Postgres(sqlx::PgPool),
    Mysql(sqlx::MySqlPool),
}

/// Dispatch a query expression across all pool variants.
///
/// Each arm binds the inner pool to `$p` and evaluates `$body`.
/// The expression is type-checked independently per arm, so the pool type
/// is correctly inferred by sqlx in each case.
macro_rules! dispatch_pool {
    ($self:expr, $p:ident => $body:expr) => {
        match $self {
            DbPool::Sqlite($p) => $body,
            DbPool::Postgres($p) => $body,
            DbPool::Mysql($p) => $body,
        }
    };
}

impl DbPool {
    /// The backend this pool speaks.
    ///
    /// D-item: this was a process-global `OnceLock<DbBackend>` set by
    /// `init_pool_no_migrate` and read by `build_sqlx`, `write_returning_row`,
    /// `insert_if_absent`, `archive_latest_active` and `snapshot_pages`. A
    /// global made one process able to hold exactly one Orion store, which is
    /// why `schema_parity` could not go through `run_migrations`, why every
    /// repository unit test had to call `set_backend_for_test`, and why the
    /// Postgres, MySQL and cluster suites are separate test binaries. The
    /// variant already carries the answer.
    pub fn backend(&self) -> DbBackend {
        match self {
            DbPool::Sqlite(_) => DbBackend::Sqlite,
            DbPool::Postgres(_) => DbBackend::Postgres,
            DbPool::Mysql(_) => DbBackend::Mysql,
        }
    }

    pub fn size(&self) -> u32 {
        dispatch_pool!(self, p => p.size())
    }

    /// Database connectivity check: `SELECT 1` through the pool.
    ///
    /// D22: this lived on `WorkflowRepository` purely because the health
    /// checks needed *a* pool and that trait happened to hold one. It is a
    /// property of the pool, not of any entity.
    pub async fn ping(&self) -> Result<(), sqlx::Error> {
        crate::metrics::timed_db_op("db.ping", async {
            let (sql, values) = build_sqlx(
                self.backend(),
                sea_query::Query::select().expr(sea_query::Expr::val(1i32)),
            );
            self.fetch_scalar::<i32>(&sql, values).await?;
            Ok(())
        })
        .await
    }

    pub fn num_idle(&self) -> usize {
        dispatch_pool!(self, p => p.num_idle())
    }

    pub async fn fetch_all_as<T: DbRow>(
        &self,
        sql: &str,
        values: sea_query_sqlx::SqlxValues,
    ) -> Result<Vec<T>, sqlx::Error> {
        dispatch_pool!(self, p => sqlx::query_as_with::<_, T, _>(AssertSqlSafe(sql), values).fetch_all(p).await)
    }

    pub async fn fetch_one_as<T: DbRow>(
        &self,
        sql: &str,
        values: sea_query_sqlx::SqlxValues,
    ) -> Result<T, sqlx::Error> {
        dispatch_pool!(self, p => sqlx::query_as_with::<_, T, _>(AssertSqlSafe(sql), values).fetch_one(p).await)
    }

    pub async fn fetch_optional_as<T: DbRow>(
        &self,
        sql: &str,
        values: sea_query_sqlx::SqlxValues,
    ) -> Result<Option<T>, sqlx::Error> {
        dispatch_pool!(self, p => sqlx::query_as_with::<_, T, _>(AssertSqlSafe(sql), values).fetch_optional(p).await)
    }

    pub async fn execute_query(
        &self,
        sql: &str,
        values: sea_query_sqlx::SqlxValues,
    ) -> Result<u64, sqlx::Error> {
        dispatch_pool!(self, p => {
            let r = sqlx::query_with(AssertSqlSafe(sql), values).execute(p).await?;
            Ok(r.rows_affected())
        })
    }

    pub async fn fetch_scalar<T>(
        &self,
        sql: &str,
        values: sea_query_sqlx::SqlxValues,
    ) -> Result<T, sqlx::Error>
    where
        T: Send + Unpin + 'static,
        T: sqlx::Type<sqlx::Sqlite> + for<'r> sqlx::Decode<'r, sqlx::Sqlite>,
        T: sqlx::Type<sqlx::Postgres> + for<'r> sqlx::Decode<'r, sqlx::Postgres>,
        T: sqlx::Type<sqlx::MySql> + for<'r> sqlx::Decode<'r, sqlx::MySql>,
        (T,): for<'r> sqlx::FromRow<'r, sqlx::sqlite::SqliteRow>,
        (T,): for<'r> sqlx::FromRow<'r, sqlx::postgres::PgRow>,
        (T,): for<'r> sqlx::FromRow<'r, sqlx::mysql::MySqlRow>,
    {
        dispatch_pool!(self, p => sqlx::query_scalar_with::<_, T, _>(AssertSqlSafe(sql), values).fetch_one(p).await)
    }

    /// Begin a deferred transaction. Fine for read-only transactions and for
    /// transactions whose *first* statement is a write; a transaction that
    /// SELECTs before its first write must use [`Self::begin_write_tx`] (D30).
    pub async fn begin_tx(&self) -> Result<DbTransaction, sqlx::Error> {
        match self {
            DbPool::Sqlite(p) => Ok(DbTransaction::Sqlite(p.begin().await?)),
            DbPool::Postgres(p) => Ok(DbTransaction::Postgres(p.begin().await?)),
            DbPool::Mysql(p) => Ok(DbTransaction::Mysql(p.begin().await?)),
        }
    }

    /// Begin a transaction that will read first and write later (D30).
    ///
    /// On SQLite this issues `BEGIN IMMEDIATE`, taking the write lock at
    /// BEGIN. A plain deferred BEGIN pins a WAL read snapshot at the first
    /// SELECT; if any other connection (e.g. the async audit-log writer)
    /// commits before this transaction's first write, the read→write upgrade
    /// fails with SQLITE_BUSY_SNAPSHOT (extended code 517), which the busy
    /// handler never retries — the transaction can only be rolled back and
    /// replayed (issue #254). With the lock held from BEGIN the snapshot
    /// cannot go stale, and contention degrades to plain SQLITE_BUSY,
    /// absorbed by the connection's `busy_timeout`.
    ///
    /// Postgres and MySQL take row locks at the first write and resolve this
    /// internally; they keep a plain `begin()`.
    pub async fn begin_write_tx(&self) -> Result<DbTransaction, sqlx::Error> {
        match self {
            DbPool::Sqlite(p) => Ok(DbTransaction::Sqlite(
                p.begin_with("BEGIN IMMEDIATE").await?,
            )),
            DbPool::Postgres(p) => Ok(DbTransaction::Postgres(p.begin().await?)),
            DbPool::Mysql(p) => Ok(DbTransaction::Mysql(p.begin().await?)),
        }
    }
}

// ============================================================
// DbTransaction — enum wrapping concrete transaction types
// ============================================================

pub enum DbTransaction {
    Sqlite(sqlx::Transaction<'static, sqlx::Sqlite>),
    Postgres(sqlx::Transaction<'static, sqlx::Postgres>),
    Mysql(sqlx::Transaction<'static, sqlx::MySql>),
}

macro_rules! dispatch_tx {
    ($self:expr, $tx:ident => $body:expr) => {
        match $self {
            DbTransaction::Sqlite($tx) => $body,
            DbTransaction::Postgres($tx) => $body,
            DbTransaction::Mysql($tx) => $body,
        }
    };
}

impl DbTransaction {
    /// The backend this transaction speaks — see [`DbPool::backend`].
    pub fn backend(&self) -> DbBackend {
        match self {
            DbTransaction::Sqlite(_) => DbBackend::Sqlite,
            DbTransaction::Postgres(_) => DbBackend::Postgres,
            DbTransaction::Mysql(_) => DbBackend::Mysql,
        }
    }

    pub async fn commit(self) -> Result<(), sqlx::Error> {
        match self {
            DbTransaction::Sqlite(tx) => tx.commit().await,
            DbTransaction::Postgres(tx) => tx.commit().await,
            DbTransaction::Mysql(tx) => tx.commit().await,
        }
    }

    pub async fn fetch_all_as<T: DbRow>(
        &mut self,
        sql: &str,
        values: sea_query_sqlx::SqlxValues,
    ) -> Result<Vec<T>, sqlx::Error> {
        dispatch_tx!(self, tx => sqlx::query_as_with::<_, T, _>(AssertSqlSafe(sql), values).fetch_all(&mut **tx).await)
    }

    pub async fn fetch_optional_as<T: DbRow>(
        &mut self,
        sql: &str,
        values: sea_query_sqlx::SqlxValues,
    ) -> Result<Option<T>, sqlx::Error> {
        dispatch_tx!(self, tx => sqlx::query_as_with::<_, T, _>(AssertSqlSafe(sql), values).fetch_optional(&mut **tx).await)
    }

    pub async fn execute_query(
        &mut self,
        sql: &str,
        values: sea_query_sqlx::SqlxValues,
    ) -> Result<u64, sqlx::Error> {
        dispatch_tx!(self, tx => {
            let r = sqlx::query_with(AssertSqlSafe(sql), values).execute(&mut **tx).await?;
            Ok(r.rows_affected())
        })
    }
}

// ============================================================
// Query builder helper — builds SQL + bound values
// ============================================================

/// Build a SQL string and bound values in `backend`'s dialect.
///
/// The backend is a parameter rather than a global read: every caller holds a
/// [`DbPool`] or a [`DbTransaction`] and can answer it with `.backend()`, and
/// the pure query-builder helpers take it from their caller. See
/// [`DbPool::backend`] for what the global cost.
pub fn build_sqlx<S: SqlxBinder>(
    backend: DbBackend,
    stmt: &mut S,
) -> (String, sea_query_sqlx::SqlxValues) {
    match backend {
        DbBackend::Sqlite => stmt.build_sqlx(sea_query::SqliteQueryBuilder),
        DbBackend::Postgres => stmt.build_sqlx(sea_query::PostgresQueryBuilder),
        DbBackend::Mysql => stmt.build_sqlx(sea_query::MysqlQueryBuilder),
    }
}

// ============================================================
// Embedded migrations — all backends compiled in, selected at runtime
// ============================================================

static MIGRATOR_SQLITE: sqlx::migrate::Migrator = sqlx::migrate!("./migrations/sqlite");
static MIGRATOR_POSTGRES: sqlx::migrate::Migrator = sqlx::migrate!("./migrations/postgres");
static MIGRATOR_MYSQL: sqlx::migrate::Migrator = sqlx::migrate!("./migrations/mysql");

/// The embedded migration set for one backend.
pub fn migrator_for(backend: DbBackend) -> &'static sqlx::migrate::Migrator {
    match backend {
        DbBackend::Sqlite => &MIGRATOR_SQLITE,
        DbBackend::Postgres => &MIGRATOR_POSTGRES,
        DbBackend::Mysql => &MIGRATOR_MYSQL,
    }
}

// ============================================================
// Pool initialization
// ============================================================

/// Initialize the database connection pool and run migrations.
/// Fresh in-memory SQLite pool with migrations applied — the shared
/// constructor for in-file repository tests.
#[cfg(test)]
pub(crate) async fn test_sqlite_pool() -> DbPool {
    init_pool(&crate::config::StorageConfig {
        url: "sqlite::memory:".to_string(),
        max_connections: 1,
        ..Default::default()
    })
    .await
    .expect("test pool")
}

pub async fn init_pool(config: &StorageConfig) -> Result<DbPool, OrionError> {
    let pool = init_pool_no_migrate(config).await?;
    run_migrations(&pool).await?;
    Ok(pool)
}

/// Initialize the pool the way the server does at boot.
///
/// With `auto_migrate = false` (the multi-replica deploy shape) a stale schema
/// is a hard startup error — a replica must never serve against pending
/// migrations. `orion-server migrate` is the deploy step that clears it, and
/// this refusal is what the Helm pre-upgrade Job and the compose `migrate`
/// service rely on. Kept out of `main` so it is reachable from tests (T5).
pub async fn init_pool_for_startup(config: &StorageConfig) -> Result<DbPool, OrionError> {
    if config.auto_migrate {
        return init_pool(config).await;
    }
    let pool = init_pool_no_migrate(config).await?;
    let pending = pending_migrations(&pool).await?;
    if !pending.is_empty() {
        return Err(OrionError::Config {
            message: format!(
                "{} pending migration(s) and storage.auto_migrate = false — \
                 run `orion-server migrate` first",
                pending.len()
            ),
        });
    }
    Ok(pool)
}

/// Initialize the database connection pool without running migrations.
pub async fn init_pool_no_migrate(config: &StorageConfig) -> Result<DbPool, OrionError> {
    init_pool_no_migrate_waiting(
        config,
        ConnectWait {
            window: Duration::from_secs(config.connect_retry_secs),
            transient_only: false,
            on_retry: None,
        },
    )
    .await
}

/// How a caller wants the initial connection waited for.
pub struct ConnectWait<'a> {
    /// The total retry window. Replaces `storage.connect_retry_secs` for this
    /// call. Ignored for SQLite, whose failures do not heal on their own.
    pub window: Duration,
    /// Retry only the failures that mean "not accepting connections yet"
    /// ([`is_transient_connect_error`]); anything else — a wrong password, an
    /// unknown database, a TLS or URL problem — fails at once.
    pub transient_only: bool,
    /// Called before each sleep, e.g. to print progress. `None` keeps the
    /// tracing-only reporting.
    pub on_retry: Option<&'a (dyn Fn(&RetryReport<'_>) + Sync)>,
}

/// One failed attempt that is about to be retried.
pub struct RetryReport<'a> {
    /// The attempt that just failed, from 1.
    pub attempt: u32,
    /// Time since the first attempt started.
    pub elapsed: Duration,
    /// How long until the next attempt.
    pub retry_in: Duration,
    /// [`connect_failure_reason`] of the failure.
    pub reason: &'a str,
}

/// [`init_pool_no_migrate`] with the caller's own wait policy — what
/// `migrate --wait` and `test-connectivity --wait` open the pool with.
pub async fn init_pool_no_migrate_waiting(
    config: &StorageConfig,
    wait: ConnectWait<'_>,
) -> Result<DbPool, OrionError> {
    let backend = detect_backend(&config.url)?;
    connect_with_retry(config, backend, wait).await
}

/// Whether a pool-open failure means the database is not reachable *yet* —
/// starting, failing over, or not yet resolvable by name — rather than
/// misconfigured.
///
/// sqlx already retries a refused connection and a server that is starting
/// up *inside* one attempt, until `acquire_timeout_secs`; what reaches this
/// level for those is `PoolTimedOut`. Any other I/O error — reset, aborted,
/// unreachable, a DNS name a compose or Kubernetes service has not
/// registered yet — is matched on the variant, since a lookup failure has no
/// `ErrorKind` of its own. A database error is transient only for SQLSTATE
/// class `08` (connection exception, which carries MySQL's 1040 and 1053)
/// and PostgreSQL's shutdown, startup and too-many-connections codes; an
/// authentication or unknown-database error will not heal by waiting.
pub fn is_transient_connect_error(err: &OrionError) -> bool {
    match connect_source(err) {
        Some(sqlx::Error::PoolTimedOut | sqlx::Error::Io(_)) => true,
        Some(sqlx::Error::Database(db)) => db.code().is_some_and(|code| {
            code.starts_with("08") || matches!(&*code, "57P01" | "57P02" | "57P03" | "53300")
        }),
        _ => false,
    }
}

/// A short reason for a failed connect, fit for a progress line. Never
/// includes the URL.
pub fn connect_failure_reason(err: &OrionError, acquire_timeout_secs: u64) -> String {
    match connect_source(err) {
        Some(sqlx::Error::PoolTimedOut) => {
            format!("not accepting connections within {acquire_timeout_secs}s")
        }
        Some(sqlx::Error::Io(io)) => io.to_string(),
        Some(sqlx::Error::Database(db)) => db.message().to_string(),
        Some(other) => other.to_string(),
        None => err.to_string(),
    }
}

/// The sqlx error a pool-open failure wraps, if it is one.
fn connect_source(err: &OrionError) -> Option<&sqlx::Error> {
    match err {
        OrionError::Internal {
            source: Some(source),
            ..
        } => source.downcast_ref::<sqlx::Error>(),
        _ => None,
    }
}

// ============================================================
// Startup connect retry (D14)
// ============================================================

/// Backoff between startup connect attempts: 250 ms doubling to a 5 s ceiling.
///
/// Deliberately a constant rather than four more config knobs — the only
/// number an operator has a reason to tune is the total window
/// (`storage.connect_retry_secs`), which is what bounds the whole loop.
const CONNECT_RETRY_BACKOFF_MAX: Duration = Duration::from_secs(5);

/// Backoff before the attempt following `failures` failed attempts:
/// 250 ms, 500 ms, 1 s, 2 s, 4 s, then 5 s forever.
fn connect_backoff(failures: u32) -> Duration {
    let doublings = failures.saturating_sub(1).min(20);
    Duration::from_millis(250u64 << doublings).min(CONNECT_RETRY_BACKOFF_MAX)
}

/// Open the pool, retrying a failure until `storage.connect_retry_secs`
/// elapses (D14).
///
/// A hard exit on an unreachable database means every replica crash-loops for
/// the duration of a Postgres/MySQL failover, and the container restart
/// backoff outlives the outage; the readiness probe already keeps traffic off
/// a pod that has not finished booting, so failing fast bought nothing.
///
/// Two things stay fail-fast on purpose: SQLite (its failures — bad path, bad
/// permissions, corrupt file — do not heal on their own), and the *migration*
/// check in [`init_pool_for_startup`], which is about schema state rather than
/// reachability.
async fn connect_with_retry(
    config: &StorageConfig,
    backend: DbBackend,
    wait: ConnectWait<'_>,
) -> Result<DbPool, OrionError> {
    let window = match backend {
        DbBackend::Sqlite => Duration::ZERO,
        _ => wait.window,
    };
    let transient_only = wait.transient_only;
    let report = wait.on_retry.map(|on_retry| {
        move |attempt: u32, elapsed: Duration, retry_in: Duration, err: &OrionError| {
            let reason = connect_failure_reason(err, config.acquire_timeout_secs);
            on_retry(&RetryReport {
                attempt,
                elapsed,
                retry_in,
                reason: &reason,
            });
        }
    });
    retry_within(
        window,
        |err| !transient_only || is_transient_connect_error(err),
        report.as_ref(),
        |attempt| async move {
            if attempt > 1 {
                tracing::info!(attempt, backend = %backend, "Retrying database connection");
            }
            match backend {
                DbBackend::Sqlite => init_sqlite_pool(config).await,
                DbBackend::Postgres => init_postgres_pool(config).await,
                DbBackend::Mysql => init_mysql_pool(config).await,
            }
        },
    )
    .await
}

/// Run `attempt` until it succeeds, fails with an error `should_retry`
/// refuses, or the next backoff would land past `window`, returning the last
/// error. `on_retry` hears about each failure that is about to be retried,
/// with the elapsed time and the backoff ahead.
///
/// Generic over the attempt so the retry *policy* is unit-testable without a
/// database outage — see `retry_within_*` in this module's tests.
async fn retry_within<T, F, Fut, R>(
    window: Duration,
    should_retry: impl Fn(&OrionError) -> bool,
    on_retry: Option<&R>,
    mut attempt: F,
) -> Result<T, OrionError>
where
    F: FnMut(u32) -> Fut,
    Fut: std::future::Future<Output = Result<T, OrionError>>,
    R: Fn(u32, Duration, Duration, &OrionError),
{
    let started = tokio::time::Instant::now();
    let deadline = started + window;
    let mut failures = 0u32;
    loop {
        match attempt(failures + 1).await {
            Ok(value) => {
                if failures > 0 {
                    tracing::info!(
                        attempts = failures + 1,
                        elapsed_ms = started.elapsed().as_millis() as u64,
                        "Database connection established after retrying"
                    );
                }
                return Ok(value);
            }
            Err(err) => {
                failures += 1;
                if !should_retry(&err) {
                    return Err(err);
                }
                let backoff = connect_backoff(failures);
                if tokio::time::Instant::now() + backoff > deadline {
                    if failures > 1 {
                        tracing::error!(
                            attempts = failures,
                            elapsed_ms = started.elapsed().as_millis() as u64,
                            "Giving up on the database connection — retry window exhausted"
                        );
                    }
                    return Err(err);
                }
                tracing::warn!(
                    attempt = failures,
                    retry_in_ms = backoff.as_millis() as u64,
                    error = %err,
                    "Database unavailable at startup; retrying \
                     (bounded by storage.connect_retry_secs)"
                );
                if let Some(on_retry) = on_retry {
                    on_retry(failures, started.elapsed(), backoff, &err);
                }
                tokio::time::sleep(backoff).await;
            }
        }
    }
}

/// Run pending database migrations.
pub async fn run_migrations(pool: &DbPool) -> Result<(), OrionError> {
    let m = migrator_for(pool.backend());
    match pool {
        DbPool::Sqlite(p) => m.run(p).await,
        DbPool::Postgres(p) => m.run(p).await,
        DbPool::Mysql(p) => m.run(p).await,
    }
    .map_err(|e| OrionError::Internal {
        context: "Failed to run migrations".to_string(),
        source: Some(Box::new(e)),
    })
}

/// List pending migrations that have not yet been applied.
pub async fn pending_migrations(pool: &DbPool) -> Result<Vec<(i64, String)>, OrionError> {
    let applied: std::collections::HashSet<i64> = {
        let sql = "SELECT version FROM _sqlx_migrations ORDER BY version";
        let result: Result<Vec<i64>, _> = match pool {
            DbPool::Sqlite(p) => sqlx::query_scalar::<_, i64>(sql).fetch_all(p).await,
            DbPool::Postgres(p) => sqlx::query_scalar::<_, i64>(sql).fetch_all(p).await,
            DbPool::Mysql(p) => sqlx::query_scalar::<_, i64>(sql).fetch_all(p).await,
        };
        match result {
            Ok(versions) => versions.into_iter().collect(),
            Err(_) => std::collections::HashSet::new(),
        }
    };

    let pending: Vec<(i64, String)> = migrator_for(pool.backend())
        .iter()
        .filter(|m| !applied.contains(&m.version))
        .map(|m| (m.version, m.description.to_string()))
        .collect();

    Ok(pending)
}

// ============================================================
// Backend-specific pool initialization
// ============================================================

async fn init_sqlite_pool(config: &StorageConfig) -> Result<DbPool, OrionError> {
    use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
    use std::str::FromStr;

    let busy_timeout = config.busy_timeout_ms.to_string();
    let options = SqliteConnectOptions::from_str(&config.url)
        .map_err(|e| OrionError::Internal {
            context: "Invalid DB path".to_string(),
            source: Some(Box::new(e)),
        })?
        .create_if_missing(true)
        .journal_mode(sqlx::sqlite::SqliteJournalMode::Wal)
        // D26: currently a no-op — the schema declares no foreign keys at all
        // (`grep REFERENCES migrations/` is empty), by design. See the module
        // doc comment. Kept on so that any FK added later is enforced from the
        // first connection rather than depending on someone remembering this.
        .pragma("foreign_keys", "ON")
        .pragma("busy_timeout", busy_timeout)
        .pragma("synchronous", "NORMAL")
        .pragma("cache_size", "-20000");

    let mut pool_opts = SqlitePoolOptions::new()
        .max_connections(config.max_connections)
        .min_connections(config.min_connections)
        .acquire_timeout(Duration::from_secs(config.acquire_timeout_secs));
    if config.idle_timeout_secs > 0 {
        pool_opts = pool_opts.idle_timeout(Duration::from_secs(config.idle_timeout_secs));
    }
    let pool = pool_opts
        .connect_with(options)
        .await
        .map_err(|e| OrionError::Internal {
            context: "Failed to connect to database".to_string(),
            source: Some(Box::new(e)),
        })?;
    Ok(DbPool::Sqlite(pool))
}

async fn init_postgres_pool(config: &StorageConfig) -> Result<DbPool, OrionError> {
    use sqlx::postgres::PgPoolOptions;

    let mut pool_opts = PgPoolOptions::new()
        .max_connections(config.max_connections)
        .min_connections(config.min_connections)
        .acquire_timeout(Duration::from_secs(config.acquire_timeout_secs));
    if config.idle_timeout_secs > 0 {
        pool_opts = pool_opts.idle_timeout(Duration::from_secs(config.idle_timeout_secs));
    }
    let pool = pool_opts
        .connect(&config.url)
        .await
        .map_err(|e| OrionError::Internal {
            context: "Failed to connect to database".to_string(),
            source: Some(Box::new(e)),
        })?;
    Ok(DbPool::Postgres(pool))
}

async fn init_mysql_pool(config: &StorageConfig) -> Result<DbPool, OrionError> {
    use sqlx::mysql::MySqlPoolOptions;

    let mut pool_opts = MySqlPoolOptions::new()
        .max_connections(config.max_connections)
        .min_connections(config.min_connections)
        .acquire_timeout(Duration::from_secs(config.acquire_timeout_secs));
    if config.idle_timeout_secs > 0 {
        pool_opts = pool_opts.idle_timeout(Duration::from_secs(config.idle_timeout_secs));
    }
    let pool = pool_opts
        .connect(&config.url)
        .await
        .map_err(|e| OrionError::Internal {
            context: "Failed to connect to database".to_string(),
            source: Some(Box::new(e)),
        })?;
    Ok(DbPool::Mysql(pool))
}

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

    #[test]
    fn test_detect_backend_sqlite() {
        assert_eq!(
            detect_backend("sqlite:orion.db").expect("test"),
            DbBackend::Sqlite
        );
        assert_eq!(
            detect_backend("sqlite::memory:").expect("test"),
            DbBackend::Sqlite
        );
    }

    #[test]
    fn test_detect_backend_postgres() {
        assert_eq!(
            detect_backend("postgres://user:pass@localhost/db").expect("test"),
            DbBackend::Postgres
        );
        assert_eq!(
            detect_backend("postgresql://user:pass@localhost/db").expect("test"),
            DbBackend::Postgres
        );
    }

    #[test]
    fn test_detect_backend_mysql() {
        assert_eq!(
            detect_backend("mysql://user:pass@localhost/db").expect("test"),
            DbBackend::Mysql
        );
    }

    #[test]
    fn test_detect_backend_unsupported() {
        assert!(detect_backend("mssql://localhost").is_err());
    }

    // ------------------------------------------------------------------
    // D14: startup connect retry. The policy is tested through
    // `retry_within` rather than a real outage — a paused clock makes the
    // backoff schedule observable without sleeping through it.
    // ------------------------------------------------------------------

    fn unavailable() -> OrionError {
        OrionError::Config {
            message: "connection refused".to_string(),
        }
    }

    fn always(_: &OrionError) -> bool {
        true
    }

    type Report = fn(u32, Duration, Duration, &OrionError);
    const NO_REPORT: Option<&Report> = None;

    /// The wait `init_pool_no_migrate` uses: the config's window, every
    /// failure retried, nothing printed.
    fn config_wait(config: &StorageConfig) -> ConnectWait<'static> {
        ConnectWait {
            window: Duration::from_secs(config.connect_retry_secs),
            transient_only: false,
            on_retry: None,
        }
    }

    /// A pool-open failure, wrapped exactly as `init_*_pool` wraps it.
    fn wrapped(err: sqlx::Error) -> OrionError {
        OrionError::Internal {
            context: "Failed to connect to database".to_string(),
            source: Some(Box::new(err)),
        }
    }

    /// A database error carrying only a SQLSTATE, standing in for the
    /// Postgres and MySQL error types.
    #[derive(Debug)]
    struct Coded(&'static str);

    impl std::fmt::Display for Coded {
        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
            write!(f, "server said {}", self.0)
        }
    }

    impl std::error::Error for Coded {}

    impl sqlx::error::DatabaseError for Coded {
        fn message(&self) -> &str {
            "server said no"
        }
        fn code(&self) -> Option<std::borrow::Cow<'_, str>> {
            Some(std::borrow::Cow::Borrowed(self.0))
        }
        fn as_error(&self) -> &(dyn std::error::Error + Send + Sync + 'static) {
            self
        }
        fn as_error_mut(&mut self) -> &mut (dyn std::error::Error + Send + Sync + 'static) {
            self
        }
        fn into_error(self: Box<Self>) -> Box<dyn std::error::Error + Send + Sync + 'static> {
            self
        }
        fn kind(&self) -> sqlx::error::ErrorKind {
            sqlx::error::ErrorKind::Other
        }
    }

    #[test]
    fn transient_classification_table() {
        use std::io::{Error as IoError, ErrorKind};
        let transient = [
            wrapped(sqlx::Error::PoolTimedOut),
            wrapped(sqlx::Error::Io(IoError::from(ErrorKind::ConnectionRefused))),
            wrapped(sqlx::Error::Io(IoError::other(
                "failed to lookup address information",
            ))),
            wrapped(sqlx::Error::Database(Box::new(Coded("57P03")))),
            wrapped(sqlx::Error::Database(Box::new(Coded("08006")))),
            wrapped(sqlx::Error::Database(Box::new(Coded("08004")))),
            wrapped(sqlx::Error::Database(Box::new(Coded("53300")))),
        ];
        for err in &transient {
            assert!(is_transient_connect_error(err), "{err:?}");
        }
        let permanent = [
            wrapped(sqlx::Error::Database(Box::new(Coded("28P01")))),
            wrapped(sqlx::Error::Database(Box::new(Coded("3D000")))),
            wrapped(sqlx::Error::Configuration("bad url".into())),
            wrapped(sqlx::Error::Tls("bad certificate".into())),
            wrapped(sqlx::Error::Protocol("unexpected".into())),
            unavailable(),
        ];
        for err in &permanent {
            assert!(!is_transient_connect_error(err), "{err:?}");
        }
        assert_eq!(
            connect_failure_reason(&wrapped(sqlx::Error::PoolTimedOut), 3),
            "not accepting connections within 3s"
        );
        assert_eq!(
            connect_failure_reason(&wrapped(sqlx::Error::Io(IoError::other("no such host"))), 3),
            "no such host"
        );
    }

    #[tokio::test(start_paused = true)]
    async fn retry_within_stops_at_a_non_transient_error() {
        let mut attempts = 0u32;
        let started = tokio::time::Instant::now();
        let result: Result<&str, OrionError> = retry_within(
            Duration::from_secs(60),
            |_| false,
            NO_REPORT,
            |_| {
                attempts += 1;
                async { Err(unavailable()) }
            },
        )
        .await;
        let err = result.expect_err("refused");
        assert_eq!(attempts, 1, "a refused error must not be retried");
        assert_eq!(started.elapsed(), Duration::ZERO);
        assert_eq!(err.to_string(), unavailable().to_string());
    }

    #[tokio::test(start_paused = true)]
    async fn retry_within_reports_each_retry() {
        let reports = std::sync::Mutex::new(Vec::new());
        let report = |attempt: u32, elapsed: Duration, retry_in: Duration, _: &OrionError| {
            reports
                .lock()
                .expect("lock")
                .push((attempt, elapsed, retry_in));
        };
        let result: Result<&str, OrionError> = retry_within(
            Duration::from_secs(60),
            always,
            Some(&report),
            |attempt| async move {
                if attempt < 4 {
                    Err(unavailable())
                } else {
                    Ok("pool")
                }
            },
        )
        .await;
        assert_eq!(result.expect("recovers"), "pool");
        let reports = reports.into_inner().expect("lock");
        assert_eq!(
            reports,
            vec![
                (1, Duration::ZERO, Duration::from_millis(250)),
                (2, Duration::from_millis(250), Duration::from_millis(500)),
                (3, Duration::from_millis(750), Duration::from_millis(1000)),
            ]
        );
    }

    #[test]
    fn connect_backoff_doubles_then_caps() {
        assert_eq!(connect_backoff(1), Duration::from_millis(250));
        assert_eq!(connect_backoff(2), Duration::from_millis(500));
        assert_eq!(connect_backoff(3), Duration::from_millis(1000));
        assert_eq!(connect_backoff(4), Duration::from_millis(2000));
        assert_eq!(connect_backoff(5), Duration::from_millis(4000));
        assert_eq!(connect_backoff(6), CONNECT_RETRY_BACKOFF_MAX);
        assert_eq!(connect_backoff(60), CONNECT_RETRY_BACKOFF_MAX);
    }

    #[tokio::test(start_paused = true)]
    async fn retry_within_recovers_when_the_database_comes_back() {
        let mut seen = Vec::new();
        let started = tokio::time::Instant::now();
        let pool: Result<&str, OrionError> =
            retry_within(Duration::from_secs(60), always, NO_REPORT, |attempt| {
                seen.push(attempt);
                async move {
                    if attempt < 4 {
                        Err(unavailable())
                    } else {
                        Ok("pool")
                    }
                }
            })
            .await;

        assert_eq!(pool.expect("test"), "pool");
        assert_eq!(seen, vec![1, 2, 3, 4], "every attempt must be numbered");
        // 250 + 500 + 1000 ms of backoff between the four attempts.
        assert_eq!(started.elapsed(), Duration::from_millis(1750));
    }

    #[tokio::test(start_paused = true)]
    async fn retry_within_gives_up_at_the_window_and_returns_the_last_error() {
        let mut attempts = 0u32;
        let started = tokio::time::Instant::now();
        let pool: Result<&str, OrionError> =
            retry_within(Duration::from_secs(2), always, NO_REPORT, |_| {
                attempts += 1;
                async { Err(unavailable()) }
            })
            .await;

        assert!(pool.is_err(), "an unreachable database must still fail");
        // 250 + 500 + 1000 ms fits in the 2 s window; the fourth backoff
        // (2000 ms) would land past it, so attempt 4 is where it stops.
        assert_eq!(attempts, 4);
        assert!(
            started.elapsed() <= Duration::from_secs(2),
            "the retry window is a hard bound, elapsed {:?}",
            started.elapsed()
        );
    }

    #[tokio::test(start_paused = true)]
    async fn retry_within_zero_window_is_fail_fast() {
        let mut attempts = 0u32;
        let pool: Result<&str, OrionError> =
            retry_within(Duration::ZERO, always, NO_REPORT, |_| {
                attempts += 1;
                async { Err(unavailable()) }
            })
            .await;

        assert!(pool.is_err());
        assert_eq!(attempts, 1, "connect_retry_secs = 0 must not retry");
    }

    /// The wiring, not just the policy: a Postgres URL pointing at a closed
    /// port must be retried (SQLite must not). Uses a port that nothing is
    /// listening on, so no database — or outage — is needed.
    #[tokio::test]
    async fn postgres_connect_retries_but_sqlite_fails_fast() {
        // Bind and immediately drop, so the port is free and refuses.
        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("test");
        let port = listener.local_addr().expect("test").port();
        drop(listener);

        // One attempt costs ~acquire_timeout_secs (sqlx retries a refused
        // connection internally until that deadline), so a run longer than
        // two of them is only reachable by retrying.
        let config = StorageConfig {
            url: format!("postgres://orion:orion@127.0.0.1:{port}/orion"),
            acquire_timeout_secs: 1,
            connect_retry_secs: 2,
            min_connections: 0,
            max_connections: 1,
            ..Default::default()
        };
        let started = std::time::Instant::now();
        let err = connect_with_retry(&config, DbBackend::Postgres, config_wait(&config)).await;
        assert!(err.is_err(), "a closed port cannot yield a pool");
        assert!(
            started.elapsed() >= Duration::from_secs(2),
            "the first failure must be retried, not propagated immediately \
             (elapsed {:?})",
            started.elapsed()
        );

        // `--wait`'s policy retries only transient failures — and a closed
        // port is one: sqlx retries the refusal inside the attempt and
        // surfaces it as `PoolTimedOut`.
        let started = std::time::Instant::now();
        let reports = std::sync::atomic::AtomicU32::new(0);
        let on_retry = |_: &RetryReport<'_>| {
            reports.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        };
        let wait = ConnectWait {
            window: Duration::from_secs(2),
            transient_only: true,
            on_retry: Some(&on_retry),
        };
        assert!(
            connect_with_retry(&config, DbBackend::Postgres, wait)
                .await
                .is_err()
        );
        assert!(
            started.elapsed() >= Duration::from_secs(2),
            "a refused connection is transient (elapsed {:?})",
            started.elapsed()
        );
        assert!(reports.load(std::sync::atomic::Ordering::Relaxed) >= 1);

        // Same window, SQLite: not retried, because its failures are not
        // transient. A missing parent directory is never created for you.
        let missing = std::env::temp_dir().join("orion-d14-missing/nested/orion.db");
        let sqlite = StorageConfig {
            url: format!("sqlite:{}", missing.display()),
            connect_retry_secs: 30,
            min_connections: 0,
            max_connections: 1,
            ..Default::default()
        };
        let started = std::time::Instant::now();
        assert!(
            connect_with_retry(&sqlite, DbBackend::Sqlite, config_wait(&sqlite))
                .await
                .is_err(),
            "an unwritable SQLite path must fail"
        );
        assert!(
            started.elapsed() < Duration::from_secs(5),
            "SQLite must fail fast regardless of connect_retry_secs (elapsed {:?})",
            started.elapsed()
        );
    }

    /// One process, three dialects. This is what the `OnceLock<DbBackend>`
    /// made impossible: the first `init_pool` latched the backend for the
    /// process, so `build_sqlx` answered the same way for every caller
    /// forever after — which is why `schema_parity` could not go through
    /// `run_migrations`, why the Postgres, MySQL and cluster suites needed
    /// their own test binaries, and why an in-process migration between two
    /// stores could never be written.
    ///
    /// Placeholder syntax is the visible difference: `$1` on Postgres, `?`
    /// on SQLite and MySQL.
    #[test]
    fn three_backends_render_in_one_process() {
        let render = |backend| {
            let (sql, _) = build_sqlx(
                backend,
                sea_query::Query::select()
                    .expr(sea_query::Expr::val(1i32))
                    .and_where(sea_query::ExprTrait::eq(
                        sea_query::Expr::cust("x"),
                        "seven",
                    )),
            );
            sql
        };

        assert!(render(DbBackend::Postgres).contains("$1"));
        assert!(render(DbBackend::Sqlite).contains('?'));
        assert!(render(DbBackend::Mysql).contains('?'));
    }

    /// A pool answers for its own backend rather than for the process.
    #[tokio::test]
    async fn a_pool_knows_its_own_backend() {
        let pool = test_sqlite_pool().await;
        assert_eq!(pool.backend(), DbBackend::Sqlite);

        let tx = pool.begin_tx().await.expect("tx");
        assert_eq!(tx.backend(), DbBackend::Sqlite);
        drop(tx);
    }
}