reliar-store-postgres 0.9.0

use sqlx::PgPool;
use sqlx::postgres::PgConnectOptions;
use sqlx::{Connection, Executor};
use testcontainers::ContainerAsync;
use testcontainers::ImageExt;
use testcontainers::core::Mount;
use testcontainers::runners::AsyncRunner;
use testcontainers_modules::postgres::Postgres;
use tokio::sync::OnceCell;

static ADMIN_URL: OnceCell<String> = OnceCell::const_new();

/// Starts the shared Postgres container this whole binary's scenarios use (unless `DATABASE_URL`
/// is set, e.g. CI's service container), and records its admin connection URL for every
/// [`fresh_db`]/[`fresh_unmigrated_db`] call. Returns the container itself — **the caller
/// (`main`) owns it as a local for the rest of the process's life and must drop it explicitly
/// before exiting** (RELIAR-27); this function never stashes it anywhere longer-lived than that.
///
/// Must run to completion exactly once, before any trial touches Postgres.
pub(crate) async fn start_shared_container() -> Option<ContainerAsync<Postgres>> {
    if let Ok(url) = std::env::var("DATABASE_URL") {
        ADMIN_URL
            .set(url)
            .expect("start_shared_container must run exactly once");

        return None;
    }

    // `reliar-` name prefix + `reliar.test=true` label (RELIAR-27): the
    // built-in `org.testcontainers.managed-by=testcontainers` label alone is too broad for
    // the manual sweep in `CONTRIBUTING.md` to key on — it would also match a different project's
    // testcontainers-managed containers on the same Docker host. Both together are what let the
    // sweep (and a human skimming `docker ps`) tell "this crate's leftovers" apart from
    // anything else testcontainers-rs is managing.
    //
    // `postgres:18-alpine` declares `VOLUME /var/lib/postgresql` (its `PGDATA`,
    // `/var/lib/postgresql/18/docker`, is a subdirectory of it), so Docker creates an anonymous
    // volume there on every `start()` unless something else already occupies that exact path.
    // Mounting a `tmpfs` there (RELIAR-94) does: it satisfies the image's `VOLUME` declaration —
    // so no anonymous volume is created at all — and is removed with the container automatically,
    // which is also materially faster than a bind-mounted filesystem for a throwaway database
    // this process only reads back the schema of, never durability across a restart. Bounded to
    // 6 GiB (RELIAR-94 M3): an unbounded `tmpfs` defaults to 50% of host RAM, and this binary's
    // per-test databases (`fresh_db`/`fresh_unmigrated_db`) are never dropped, only ever added —
    // measured at ~3 GB for a full local run of this, by far the largest, test binary; 2 GiB was
    // tried first and genuinely ran out mid-suite (`could not extend file … No space left on
    // device`), so the bound leaves headroom above the measured figure rather than pinning to it.
    let container = Postgres::default()
        .with_tag("18-alpine")
        .with_container_name(format!("reliar-pg-{}", uuid::Uuid::now_v7().simple()))
        .with_label("reliar.test", "true")
        .with_mount(
            Mount::tmpfs_mount("/var/lib/postgresql").with_size_bytes(6 * 1024 * 1024 * 1024),
        )
        .start()
        .await
        .expect("start postgres container");
    let port = container
        .get_host_port_ipv4(5432)
        .await
        .expect("mapped port");
    let url = format!("postgres://postgres:postgres@127.0.0.1:{port}/postgres");
    ADMIN_URL
        .set(url)
        .expect("start_shared_container must run exactly once");

    Some(container)
}

fn admin_url() -> &'static str {
    ADMIN_URL
        .get()
        .expect("start_shared_container must run before any scenario touches Postgres")
}

/// Creates a fresh, empty, uniquely named database on the admin connection and returns
/// [`PgConnectOptions`] pointing at it (no `search_path` set yet).
async fn create_fresh_database() -> PgConnectOptions {
    let admin = PgPool::connect(admin_url())
        .await
        .expect("connect to admin database");
    let name = format!("t_{}", uuid::Uuid::now_v7().simple());

    // `CREATE DATABASE` cannot take a bind parameter; `name` is a freshly generated UUIDv7, never
    // user input, so this is the one sanctioned exception to "macros only" (test code, not the
    // crate) — asserted safe explicitly via `AssertSqlSafe` (sqlx 0.9's SQL-injection audit gate).
    sqlx::query(sqlx::AssertSqlSafe(format!(r#"CREATE DATABASE "{name}""#)))
        .execute(&admin)
        .await
        .expect("create test database");

    let options: PgConnectOptions = admin_url()
        .parse()
        .expect("admin url parses as PgConnectOptions");

    options.database(&name)
}

/// A fresh, empty database — **not yet migrated**. For tests that exercise
/// construction/migration failure paths before `migrate()` has run.
pub(crate) async fn fresh_unmigrated_db() -> PgPool {
    PgPool::connect_with(create_fresh_database().await)
        .await
        .expect("connect to fresh database")
}

/// Migrated once per process; every [`fresh_db`] call clones it via `CREATE DATABASE …
/// TEMPLATE …` instead of re-running `migrate()` — a storage-level file copy is materially
/// faster than replaying DDL for every test (§8.2, RELIAR-16).
static TEMPLATE_NAME: OnceCell<String> = OnceCell::const_new();

async fn template_name() -> &'static str {
    TEMPLATE_NAME
        .get_or_init(|| async {
            let options = create_fresh_database().await;
            let name = options.get_database().unwrap().to_owned();
            let pool = PgPool::connect_with(options)
                .await
                .expect("connect to template database");

            reliar_store_postgres::migrate(&pool, reliar_store_postgres::MigrateOptions::default())
                .await
                .expect("migrate the template database");
            // `CREATE DATABASE … TEMPLATE` refuses a source with open connections; closing the
            // pool here is what makes every later clone safe.
            pool.close().await;

            name
        })
        .await
}

/// A fresh database, cloned from the migrated [`template_name`], with `search_path` set on the
/// returned pool itself (§24) so the store's startup verification and every query resolve
/// `outbox` in the default `reliar` schema with **no** URL/role configuration — the equivalent
/// of a host putting `reliar` first on its connection URL.
pub(crate) async fn fresh_db() -> PgPool {
    let admin = PgPool::connect(admin_url())
        .await
        .expect("connect to admin database");
    let name = format!("t_{}", uuid::Uuid::now_v7().simple());
    let template = template_name().await;

    // Same sanctioned `AssertSqlSafe` exception as `create_fresh_database`: both names are
    // freshly generated UUIDv7s, never user input.
    sqlx::query(sqlx::AssertSqlSafe(format!(
        r#"CREATE DATABASE "{name}" TEMPLATE "{template}""#
    )))
    .execute(&admin)
    .await
    .expect("clone the migrated template database");

    let options: PgConnectOptions = admin_url()
        .parse()
        .expect("admin url parses as PgConnectOptions");

    PgPool::connect_with(
        options
            .database(&name)
            .options([("search_path", "reliar,public")]),
    )
    .await
    .expect("connect with search_path set")
}

/// The crate's own migration set, embedded a second time from test code (test code cannot see
/// `src/migrate.rs`'s private `MIGRATOR`) — the same technique `migrate.rs`'s own
/// `ALL_MIGRATIONS` uses.
static ALL_MIGRATIONS: sqlx::migrate::Migrator = sqlx::migrate!("./migrations");

/// Applies only the first `count` migrations against a fresh database, on a dedicated connection
/// with `search_path` set the way [`reliar_store_postgres::migrate`] itself sets it — every
/// migration through `0006` is unqualified SQL. Shared by any scenario that needs to observe an
/// intermediate migration state: the upgrade trial (P-1/P-2/P-3/P-4) and the `SchemaOutOfDate`
/// completion-marker trials (P-13 and its `0005`-`0009` sibling, ADR 0044 Amendment A.5).
pub(crate) async fn apply_migration_prefix(pool: &PgPool, count: usize) {
    let mut conn = sqlx::postgres::PgConnection::connect_with(&pool.connect_options())
        .await
        .expect("dedicated connection for the partial migration");

    conn.execute("SET search_path = reliar, public")
        .await
        .expect("set search_path for the unqualified migrations");

    let mut prefix = sqlx::migrate::Migrator {
        migrations: std::borrow::Cow::Owned(ALL_MIGRATIONS.migrations[..count].to_vec()),
        ignore_missing: ALL_MIGRATIONS.ignore_missing,
        locking: ALL_MIGRATIONS.locking,
        no_tx: ALL_MIGRATIONS.no_tx,
        table_name: ALL_MIGRATIONS.table_name.clone(),
        create_schemas: ALL_MIGRATIONS.create_schemas.clone(),
    };
    prefix.create_schema("reliar".to_owned());
    prefix.dangerous_set_table_name("reliar._migrations");
    prefix.set_locking(false);
    prefix
        .run(&mut conn)
        .await
        .unwrap_or_else(|err| panic!("apply the first {count} migrations: {err}"));

    conn.close().await.expect("close the dedicated connection");
}