//! Postgres work the gates need: the scratch database they run against, the
//! backup restore that seeds it, and the URL surgery both require.
//!
//! A peer of the gate families rather than one family's helper.
//! [`clean_stale_test_dbs`] is called from both `cargo_test` and
//! `hardening_test`, and [`pg_url_with_dbname`] from the migration check as
//! well as from code_smoke.

use super::log::GateLog;
use anyhow::{Context, Result};
use ops_exec::sh_quote;
use tokio::process::Command;

pub(crate) async fn reset_scratch(db_url: &str, owner_role: &str) -> Result<()> {
    use sqlx::Executor;
    use sqlx::postgres::PgPoolOptions;
    let pool = PgPoolOptions::new()
        .max_connections(1)
        .connect(db_url)
        .await?;
    // `owner_role` is validated `[A-Za-z0-9_]+` at config load, so interpolating
    // it into DDL is sound. It still goes through `format('%I')` inside the DO
    // block for the quoting Postgres expects on an identifier.
    let sql = format!(
        r#"
        DO $$
        DECLARE s text;
        BEGIN
            -- The dump restores objects owned by the prod role and re-grants to
            -- it (`ALTER ... OWNER TO {owner_role}`), which errors if the role
            -- is absent — superuser does not imply the role exists. Create it
            -- NOLOGIN: the scratch DB needs the role as an *owner* only, never
            -- as a connecting identity. Idempotent, so a re-reset is a no-op.
            IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = '{owner_role}') THEN
                EXECUTE format('CREATE ROLE %I NOLOGIN', '{owner_role}');
            END IF;

            -- Drop every non-system schema, not just public — migrations create
            -- custom schemas (e.g. tower_sessions) that survive `DROP SCHEMA
            -- public CASCADE` and then collide on the next migration run.
            FOR s IN
                SELECT nspname FROM pg_namespace
                WHERE nspname NOT LIKE 'pg_%'
                  AND nspname NOT IN ('information_schema')
            LOOP
                EXECUTE format('DROP SCHEMA IF EXISTS %I CASCADE', s);
            END LOOP;
            EXECUTE 'CREATE SCHEMA public';
            -- Restore the pre-PG15 public-schema default on the throwaway
            -- scratch DB. Without this, the freshly-created public is owned by
            -- the connecting role (sando) with no grant to anyone else, so a
            -- migration's FK/trigger check that Postgres runs as a *restored*
            -- prod-owned table's owner ({owner_role} from the backup dump)
            -- fails with "permission denied for schema public". Granting to
            -- PUBLIC is role-agnostic and safe here — this DB is disposable and
            -- exists only to dry-run migrations.
            EXECUTE 'GRANT USAGE, CREATE ON SCHEMA public TO PUBLIC';
            -- PG15+: the new owner needs CREATE on public in its own right, not
            -- only via PUBLIC, for the restore's owner-scoped DDL.
            EXECUTE format('GRANT USAGE, CREATE ON SCHEMA public TO %I', '{owner_role}');
        END $$;
        "#
    );
    pool.execute(sqlx::raw_sql(sqlx::AssertSqlSafe(sql)))
        .await?;
    pool.close().await;
    Ok(())
}

/// Startup assertion for the scratch cluster: the gates reset it, seed an owner
/// role into it, and drop leftover test databases in it, none of which a plain
/// unprivileged role can do. Satisfied by hand on the build host
/// (`ALTER ROLE sando SUPERUSER`, a created `makenotwork` role); unasserted, a
/// rebuild elsewhere fails one gate at a time with an opaque permissions error.
/// Assert once, at boot, loudly.
///
/// Not part of `--check-config`: that path is pure by design (no DB, no
/// network), and a green there must mean "this build understands its config",
/// not "the cluster is reachable".
pub async fn preflight_scratch_privileges(db_url: &str) -> Result<()> {
    use sqlx::postgres::PgPoolOptions;
    let pool = PgPoolOptions::new()
        .max_connections(1)
        .connect(db_url)
        .await
        .context("connecting to scratch_db_url for the startup privilege check")?;
    let (is_super, can_signal): (bool, bool) = sqlx::query_as(
        "SELECT rolsuper, pg_catalog.pg_has_role(current_user, 'pg_signal_backend', 'USAGE')
         FROM pg_roles WHERE rolname = current_user",
    )
    .fetch_one(&pool)
    .await?;
    pool.close().await;
    anyhow::ensure!(
        is_super || can_signal,
        "the scratch_db_url role has neither SUPERUSER nor pg_signal_backend; migration_dry_run \
         and cargo_test cannot reset the scratch DB or clear stale test databases. Grant one:\n  \
         ALTER ROLE <role> SUPERUSER;            -- what fw13 uses\n  \
         GRANT pg_signal_backend TO <role>;      -- narrower: terminate only, cannot drop \
         foreign-owned databases",
    );
    if !is_super {
        tracing::warn!(
            "scratch role has pg_signal_backend but not SUPERUSER: stale test databases owned by \
             another role cannot be dropped, and the scratch owner role cannot be created if absent"
        );
    }
    Ok(())
}

/// Best-effort cleanup of stale per-test database clones (`mnw_test_<uuid>`)
/// left behind by a killed `cargo_test` run.
///
/// Drops **foreign-owned leftovers too**, which is why the daemon asserts
/// SUPERUSER at startup (`preflight_scratch_privileges`): `DROP DATABASE`
/// requires ownership or superuser, and the `WITH (FORCE)` terminate requires
/// superuser or `pg_signal_backend`. Without both, orphans from a run under a
/// different role accumulate and degrade the gate — the failure this cleanup
/// exists to prevent.
///
/// OPERATIONAL HAZARD: fw13 runs one Postgres cluster shared with local `cargo
/// test` as `max`, so a gate firing mid-local-test will force-drop that run's
/// databases out from under it. That collision is known and tracked separately
/// (give the gate its own cluster); until then, do not run local tests on fw13
/// while a Sando gate is live.
///
/// Deliberately **excludes the template** (`mnw_test_template_*`): the harness
/// reuses it across runs when it's migration-current (skipping a full
/// drop+migrate), so dropping it here would force a needless rebuild every
/// gate run. Templates are bounded (one per role) and never accumulate, so
/// leaving them is free. Never returns an error: a cleanup miss must not turn a
/// deploy red.
pub(super) async fn clean_stale_test_dbs(db_url: &str) {
    use sqlx::Executor;
    use sqlx::postgres::PgPoolOptions;
    let pool = match PgPoolOptions::new()
        .max_connections(1)
        .connect(db_url)
        .await
    {
        Ok(p) => p,
        Err(e) => {
            tracing::warn!(error = %e, "stale test-db cleanup: could not connect; skipping");
            return;
        }
    };
    // Every per-test clone, whoever owns it. The ownership filter this used to
    // carry is what let foreign-owned orphans pile up; superuser (asserted at
    // startup) makes them droppable.
    let names: Vec<(String,)> = sqlx::query_as(
        "SELECT datname FROM pg_database
         WHERE datname LIKE 'mnw_test_%'
           AND datname NOT LIKE '%template%'",
    )
    .fetch_all(&pool)
    .await
    .unwrap_or_default();
    let count = names.len();
    for (name,) in names {
        // `name` comes straight from pg_database; quoting it is sufficient.
        if let Err(e) = pool
            .execute(sqlx::raw_sql(sqlx::AssertSqlSafe(format!(
                "DROP DATABASE IF EXISTS \"{name}\" WITH (FORCE)"
            ))))
            .await
        {
            tracing::warn!(error = %e, db = %name, "stale test-db cleanup: drop failed");
        }
    }
    if count > 0 {
        tracing::info!(
            count,
            "stale test-db cleanup: dropped leftover mnw_test_* databases"
        );
    }
    pool.close().await;
}

/// Build the restore shell line. Two pipelines we accept:
///   *.sql      -> psql -v ON_ERROR_STOP=1 $url < dump
///   *.sql.gz   -> set -o pipefail; gunzip -c dump | psql -v ON_ERROR_STOP=1 $url
///
/// Two safety flags are load-bearing (CF4):
/// - `ON_ERROR_STOP=1`: without it, psql exits 0 even when individual statements
///   error, so a partial/corrupt restore would *pass* the gate.
/// - `set -o pipefail`: without it a shell pipeline reports only the last
///   command's status, so a `gunzip` failure on a truncated archive is masked by
///   psql's exit. pipefail is a bash builtin (not POSIX sh), so the runner uses
///   `bash -c`.
pub(super) fn restore_shell(db_url: &str, dump: &str) -> String {
    if std::path::Path::new(dump)
        .extension()
        .is_some_and(|ext| ext.eq_ignore_ascii_case("gz"))
    {
        format!(
            "set -o pipefail; gunzip -c {q} | psql -v ON_ERROR_STOP=1 {url}",
            q = sh_quote(dump),
            url = sh_quote(db_url),
        )
    } else {
        format!(
            "psql -v ON_ERROR_STOP=1 {url} < {q}",
            url = sh_quote(db_url),
            q = sh_quote(dump),
        )
    }
}

pub(super) async fn restore_dump(db_url: &str, dump: &str, log: &GateLog) -> Result<()> {
    // Split the password out of the URL and hand it to psql via PGPASSWORD, so it
    // never lands in argv (visible in /proc/<pid>/cmdline to any local user).
    // The sanitized URL — user/host/db, no secret — goes on the command line.
    let (sanitized, password) = split_pg_password(db_url);
    let shell = restore_shell(&sanitized, dump);
    // `bash` (not `sh`): `set -o pipefail` is a bash builtin. The restore runs
    // locally on the Sando host (fw13), which has bash.
    let mut cmd = Command::new("bash");
    cmd.arg("-c").arg(&shell);
    // kill_on_drop so the gate's wall-clock ceiling (dispatcher-level timeout on
    // migration_dry_run) can't orphan a wedged psql restore.
    cmd.kill_on_drop(true);
    if let Some(pw) = password {
        cmd.env("PGPASSWORD", pw);
    }
    // Streamed, not `.output()`: a prod-sized restore runs for minutes, and
    // psql's progress is the only thing an operator has to watch during it.
    let (_stdout, stderr, status) = log.run(&mut cmd).await?;
    anyhow::ensure!(
        status.success(),
        "restore failed: {}",
        String::from_utf8_lossy(&stderr),
    );
    Ok(())
}

/// Split a `postgres://user:password@host/db` URL into its password-free form and
/// the (percent-decoded) password. Returns the URL unchanged with `None` when
/// there is no userinfo password. psql reads the password from `PGPASSWORD`, so
/// keeping it off the command line removes the /proc exposure.
pub(super) fn split_pg_password(db_url: &str) -> (String, Option<String>) {
    let Some(after) = db_url.find("://").map(|i| i + 3) else {
        return (db_url.to_string(), None);
    };
    // The authority ends at the first '/', '?' or '#'; the password (if any) is
    // between the first ':' and the '@' within the userinfo of that authority.
    let authority_end = db_url[after..]
        .find(['/', '?', '#'])
        .map_or(db_url.len(), |i| after + i);
    let Some(at) = db_url[after..authority_end].find('@').map(|i| after + i) else {
        return (db_url.to_string(), None);
    };
    let userinfo = &db_url[after..at];
    let Some(colon) = userinfo.find(':') else {
        return (db_url.to_string(), None);
    };
    let password = percent_decode(&userinfo[colon + 1..]);
    let sanitized = format!(
        "{}{}{}",
        &db_url[..after],
        &userinfo[..colon],
        &db_url[at..]
    );
    (sanitized, Some(password))
}

/// Minimal `%XX` percent-decode for a URL userinfo component. Non-escape bytes
/// pass through; a malformed escape is left literal.
pub(super) fn percent_decode(s: &str) -> String {
    let b = s.as_bytes();
    let mut out = Vec::with_capacity(b.len());
    let mut i = 0;
    while i < b.len() {
        if b[i] == b'%'
            && i + 2 < b.len()
            && let (Some(h), Some(l)) = (hex_val(b[i + 1]), hex_val(b[i + 2]))
        {
            out.push((h << 4) | l);
            i += 3;
        } else {
            out.push(b[i]);
            i += 1;
        }
    }
    String::from_utf8_lossy(&out).into_owned()
}

pub(super) fn hex_val(c: u8) -> Option<u8> {
    match c {
        b'0'..=b'9' => Some(c - b'0'),
        b'a'..=b'f' => Some(c - b'a' + 10),
        b'A'..=b'F' => Some(c - b'A' + 10),
        _ => None,
    }
}

pub(crate) async fn run_migrator(db_url: &str, dir: &std::path::Path) -> Result<()> {
    use sqlx::postgres::PgPoolOptions;
    let pool = PgPoolOptions::new()
        .max_connections(1)
        .connect(db_url)
        .await?;
    let migrator = sqlx::migrate::Migrator::new(dir).await?;
    migrator.run(&pool).await?;
    pool.close().await;
    Ok(())
}

/// Rewrite a `postgres://` URL to point at database `dbname`, preserving scheme,
/// userinfo, host/port, and any query (e.g. the socket `?host=/var/run/postgresql`
/// form) + fragment. Used to derive the maintenance connection (`postgres`) and
/// the throwaway smoke DB URL from the configured `scratch_db_url`.
pub(super) fn pg_url_with_dbname(url: &str, dbname: &str) -> String {
    let Some(after_scheme) = url.find("://").map(|i| i + 3) else {
        return url.to_string();
    };
    let rest = &url[after_scheme..];
    // Authority ends at the first '/', '?' or '#'; whatever follows is the
    // path (the old dbname) plus an optional query/fragment we must keep.
    let auth_end = rest.find(['/', '?', '#']).unwrap_or(rest.len());
    let authority = &rest[..auth_end];
    let tail = &rest[auth_end..];
    let query_and_frag = match tail.find(['?', '#']) {
        Some(i) => &tail[i..],
        None => "",
    };
    format!(
        "{}{}/{}{}",
        &url[..after_scheme],
        authority,
        dbname,
        query_and_frag
    )
}

/// Create the throwaway smoke DB on the cluster `maintenance_url` points at,
/// dropping any stale one first. `dbname` is sanitized to `[a-z0-9_]` by
/// `code_smoke_db_name`, so quoting it is sufficient. `CREATE DATABASE` cannot
/// run inside a transaction, so these go through the simple-query protocol (a
/// raw `&str` execute), matching `reset_scratch`.
pub(super) async fn pg_create_db(maintenance_url: &str, dbname: &str) -> Result<()> {
    use sqlx::Executor;
    use sqlx::postgres::PgPoolOptions;
    let pool = PgPoolOptions::new()
        .max_connections(1)
        .connect(maintenance_url)
        .await?;
    pool.execute(sqlx::raw_sql(sqlx::AssertSqlSafe(format!(
        "DROP DATABASE IF EXISTS \"{dbname}\" WITH (FORCE)"
    ))))
    .await?;
    pool.execute(sqlx::raw_sql(sqlx::AssertSqlSafe(format!(
        "CREATE DATABASE \"{dbname}\""
    ))))
    .await?;
    pool.close().await;
    Ok(())
}

/// Drop the throwaway smoke DB, forcing off any lingering connection (the killed
/// server's pool). Best-effort at the call site — a failure is logged, not fatal.
pub(super) async fn pg_drop_db(maintenance_url: &str, dbname: &str) -> Result<()> {
    use sqlx::Executor;
    use sqlx::postgres::PgPoolOptions;
    let pool = PgPoolOptions::new()
        .max_connections(1)
        .connect(maintenance_url)
        .await?;
    pool.execute(sqlx::raw_sql(sqlx::AssertSqlSafe(format!(
        "DROP DATABASE IF EXISTS \"{dbname}\" WITH (FORCE)"
    ))))
    .await?;
    pool.close().await;
    Ok(())
}

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

    /// reset_scratch must drop every non-system schema, not just `public` —
    /// otherwise migrations that create custom schemas (e.g. tower_sessions)
    /// collide on the next run. This regressed once (Phase 0) and the fix is
    /// load-bearing for migration_dry_run.
    ///
    /// Gated on `SANDO_TEST_PG_URL` so it only runs where postgres is
    /// available. Set `SANDO_TEST_PG_URL=postgres:///sando_scratch?host=/var/run/postgresql`
    /// (or similar) before `cargo test`.
    #[tokio::test]
    async fn reset_scratch_drops_all_non_system_schemas() {
        let Ok(url) = std::env::var("SANDO_TEST_PG_URL") else {
            eprintln!("skipping: SANDO_TEST_PG_URL not set");
            return;
        };
        use sqlx::Executor;
        use sqlx::postgres::PgPoolOptions;

        let pool = PgPoolOptions::new()
            .max_connections(1)
            .connect(&url)
            .await
            .unwrap();
        // Plant two non-system schemas + a table in each.
        pool.execute(
            "DROP SCHEMA IF EXISTS foo CASCADE; CREATE SCHEMA foo; CREATE TABLE foo.t (i int);",
        )
        .await
        .unwrap();
        pool.execute("DROP SCHEMA IF EXISTS tower_sessions CASCADE; CREATE SCHEMA tower_sessions; CREATE TABLE tower_sessions.session (id text);")
            .await.unwrap();
        pool.close().await;

        reset_scratch(&url, "makenotwork")
            .await
            .expect("reset_scratch");

        let pool = PgPoolOptions::new()
            .max_connections(1)
            .connect(&url)
            .await
            .unwrap();
        let rows: Vec<(String,)> = sqlx::query_as(
            "SELECT nspname FROM pg_namespace WHERE nspname NOT LIKE 'pg_%' AND nspname <> 'information_schema'",
        )
        .fetch_all(&pool)
        .await
        .unwrap();
        let names: Vec<String> = rows.into_iter().map(|(s,)| s).collect();
        // After reset, only `public` should remain among non-system schemas.
        assert_eq!(names, vec!["public".to_string()], "got: {names:?}");
        pool.close().await;
    }

    /// reset_scratch must leave the dump's owner role existing and able to
    /// create in `public`, because a prod `pg_dump` carries `ALTER ... OWNER TO
    /// <role>` for every object. This was satisfied by a hand-created NOLOGIN
    /// role on fw13; nothing recorded it, so any other box failed
    /// migration_dry_run at the restore with "role does not exist".
    ///
    /// Uses a throwaway role name so it can prove the *creation* path rather
    /// than passing on fw13's pre-existing `makenotwork`. Same
    /// `SANDO_TEST_PG_URL` gate as above; needs a superuser connection.
    #[tokio::test]
    async fn reset_scratch_seeds_the_dump_owner_role_when_absent() {
        let Ok(url) = std::env::var("SANDO_TEST_PG_URL") else {
            eprintln!("skipping: SANDO_TEST_PG_URL not set");
            return;
        };
        use sqlx::Executor;
        use sqlx::postgres::PgPoolOptions;

        let role = "sando_test_owner_probe";
        // `DROP ROLE` refuses while the role still holds the grants reset_scratch
        // gave it, so drop what it owns first. Idempotent, and a no-op when the
        // role is absent (the usual case on a first run).
        let drop_role = format!(
            "DO $$ BEGIN
               IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = '{role}') THEN
                 EXECUTE 'DROP OWNED BY {role}';
                 EXECUTE 'DROP ROLE {role}';
               END IF;
             END $$;"
        );

        let pool = PgPoolOptions::new()
            .max_connections(1)
            .connect(&url)
            .await
            .unwrap();
        pool.execute(sqlx::raw_sql(sqlx::AssertSqlSafe(drop_role.clone())))
            .await
            .unwrap();
        pool.close().await;

        reset_scratch(&url, role)
            .await
            .expect("reset_scratch creates the owner role");

        let pool = PgPoolOptions::new()
            .max_connections(1)
            .connect(&url)
            .await
            .unwrap();
        let (exists, can_login): (bool, bool) =
            sqlx::query_as("SELECT true, rolcanlogin FROM pg_roles WHERE rolname = $1")
                .bind(role)
                .fetch_one(&pool)
                .await
                .expect("owner role exists after reset");
        assert!(exists);
        assert!(
            !can_login,
            "the owner role is an owner only, never a login identity"
        );

        // The restore's owner-scoped DDL needs CREATE on public in the role's
        // own right (PG15+ dropped the implicit grant).
        let (has_create,): (bool,) =
            sqlx::query_as("SELECT pg_catalog.has_schema_privilege($1, 'public', 'CREATE')")
                .bind(role)
                .fetch_one(&pool)
                .await
                .unwrap();
        assert!(has_create, "owner role must be able to create in public");

        // Idempotent: a second reset must not error on the now-existing role.
        pool.close().await;
        reset_scratch(&url, role)
            .await
            .expect("reset_scratch is idempotent");

        let pool = PgPoolOptions::new()
            .max_connections(1)
            .connect(&url)
            .await
            .unwrap();
        pool.execute(sqlx::raw_sql(sqlx::AssertSqlSafe(drop_role)))
            .await
            .unwrap();
        pool.close().await;
    }

    /// The preflight must pass against a privileged scratch connection. Guards
    /// the catalog query itself: a wrong column or a `current_user` that matches
    /// no `pg_roles` row would make `fetch_one` error (or, worse, a silently
    /// swapped pair would invert the check) and brick startup for everyone.
    /// `SANDO_TEST_PG_URL` is expected to be a superuser connection, as the
    /// gates require.
    #[tokio::test]
    async fn preflight_passes_on_a_privileged_scratch_connection() {
        let Ok(url) = std::env::var("SANDO_TEST_PG_URL") else {
            eprintln!("skipping: SANDO_TEST_PG_URL not set");
            return;
        };
        preflight_scratch_privileges(&url)
            .await
            .expect("a superuser scratch connection must satisfy the preflight");
    }

    /// CF4: the restore pipeline must carry `ON_ERROR_STOP=1` (so psql fails on
    /// a bad statement instead of exiting 0 on a partial restore) and, for a
    /// gzip source, `set -o pipefail` (so a `gunzip` failure on a truncated
    /// archive isn't masked by psql's exit). Pure string check — no postgres.
    #[test]
    fn restore_shell_has_error_stop_and_pipefail() {
        let gz = restore_shell("postgres:///scratch", "/srv/sando/backups/latest.sql.gz");
        assert!(gz.contains("ON_ERROR_STOP=1"), "gz: {gz}");
        assert!(gz.contains("set -o pipefail"), "gz: {gz}");
        assert!(gz.contains("gunzip -c"), "gz: {gz}");

        let plain = restore_shell("postgres:///scratch", "/srv/sando/backups/dump.sql");
        assert!(plain.contains("ON_ERROR_STOP=1"), "plain: {plain}");
        // No pipeline for a plain .sql, so pipefail is unnecessary there.
        assert!(!plain.contains("gunzip"), "plain: {plain}");
        // The db url is single-quote escaped in both forms.
        assert!(plain.contains("'postgres:///scratch'"), "plain: {plain}");
    }

    #[test]
    fn split_pg_password_extracts_and_sanitizes() {
        // Password lifted out of the URL; the sanitized form keeps user/host/db.
        let (url, pw) = split_pg_password("postgres://sando:s3cret@db.host:5432/scratch");
        assert_eq!(url, "postgres://sando@db.host:5432/scratch");
        assert_eq!(pw.as_deref(), Some("s3cret"));
        // Percent-encoded password is decoded for PGPASSWORD.
        let (url, pw) = split_pg_password("postgresql://u:p%40ss%2Fword@h/d");
        assert_eq!(url, "postgresql://u@h/d");
        assert_eq!(pw.as_deref(), Some("p@ss/word"));
    }

    #[test]
    fn split_pg_password_noop_without_password() {
        // No userinfo password -> unchanged, None. (A ':' after the '@', e.g. a
        // port, must not be mistaken for the password delimiter.)
        assert_eq!(
            split_pg_password("postgres:///scratch"),
            ("postgres:///scratch".to_string(), None),
        );
        assert_eq!(
            split_pg_password("postgres://sando@db.host:5432/scratch"),
            ("postgres://sando@db.host:5432/scratch".to_string(), None),
        );
    }

    #[test]
    fn percent_decode_handles_escapes_and_malformed() {
        assert_eq!(percent_decode("plain"), "plain");
        assert_eq!(percent_decode("a%2Fb"), "a/b");
        // A malformed trailing escape is left literal, not dropped.
        assert_eq!(percent_decode("ab%2"), "ab%2");
        assert_eq!(percent_decode("ab%zz"), "ab%zz");
    }

    #[test]
    fn pg_url_with_dbname_rewrites_the_database() {
        // user:pass@host:port/db?query — swap db, keep everything else.
        assert_eq!(
            pg_url_with_dbname(
                "postgres://sando:pw@db.host:5432/sando_scratch?sslmode=require",
                "postgres"
            ),
            "postgres://sando:pw@db.host:5432/postgres?sslmode=require",
        );
        // Socket form: the query carries `host=/var/run/postgresql` and must survive.
        assert_eq!(
            pg_url_with_dbname(
                "postgres:///sando_scratch?host=/var/run/postgresql",
                "sando_code_smoke_0_9_6"
            ),
            "postgres:///sando_code_smoke_0_9_6?host=/var/run/postgresql",
        );
        // Plain host/db, no query.
        assert_eq!(
            pg_url_with_dbname("postgres://localhost/scratch", "postgres"),
            "postgres://localhost/scratch".replace("scratch", "postgres"),
        );
        // No authority, no query (loopback socket, default db path).
        assert_eq!(
            pg_url_with_dbname("postgres:///scratch", "postgres"),
            "postgres:///postgres",
        );
    }

    /// Sanity: applying MNW migrations from a *non-existent* dir errors,
    /// rather than silently no-op'ing. Cheap pure check, no postgres needed
    /// (the sqlx::Migrator::new constructor itself reads the dir).
    #[tokio::test]
    async fn run_migrator_errors_on_missing_dir() {
        // The first thing run_migrator does is `Migrator::new(dir)`, which
        // needs a real dir to read migration files from.
        let res = run_migrator(
            "postgres:///does-not-matter",
            std::path::Path::new("/nonexistent/sando-test-migrations"),
        )
        .await;
        assert!(res.is_err());
    }
}