//! Fetch the prod backup that `migration_dry_run` runs against.
//!
//! Sources supported:
//!   - `file:///abs/path/to/dump.sql.gz`        — local copy (dev).
//!   - `rsync://host/module/path`               — rsync daemon protocol.
//!   - `ssh://user@host[:port]/path/file.sql.gz` — rsync-over-ssh. Used to pull
//!     prod backups from `backup-puller@alpha-west-1`.
//!
//! The fetch is command-driven: the operator triggers it via /backup/fetch, it
//! is not implicit in promote. That keeps the slowest, most failure-prone step
//! visible in the TUI rather than buried inside a deploy.

use crate::config::AppConfig;
use crate::topology::Topology;
use anyhow::{Context, Result, bail};
use chrono::Utc;
use ops_exec::{CapabilitySet, Executor, SshExec, SyncOpts};
use sqlx::SqlitePool;
use std::path::Path;
use std::sync::Arc;
use tokio::process::Command;

#[derive(Debug, Clone)]
pub struct FetchedBackup {
    /// Which configured dump this is (`BackupConfig::name`), so the caller can
    /// tell the server's from multithreaded's in one response.
    pub name: String,
    pub source: String,
    pub local_path: String,
    pub byte_size: Option<i64>,
}

/// Parsed `backup.source` URL. Owned strings so the parsed form outlives the
/// (possibly transient) URL we read from config.
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum BackupSource {
    /// Local file copy. Path follows the `file://` prefix.
    File { path: String },
    /// rsync daemon protocol. Full URL stays intact (rsync handles it).
    RsyncDaemon { url: String },
    /// rsync-over-ssh. Port is optional.
    Ssh {
        user_host: String,
        port: Option<u16>,
        path: String,
    },
}

/// Parse a `backup.source` URL into a `BackupSource`. Rejects unsupported
/// schemes and malformed `ssh://` URLs (no path part).
pub(crate) fn parse_source(s: &str) -> Result<BackupSource> {
    if let Some(rest) = s.strip_prefix("file://") {
        if rest.is_empty() {
            bail!("file:// URL is missing a path: {s}");
        }
        return Ok(BackupSource::File { path: rest.into() });
    }
    if s.starts_with("rsync://") {
        return Ok(BackupSource::RsyncDaemon { url: s.into() });
    }
    if let Some(rest) = s.strip_prefix("ssh://") {
        let (user_host_port, path_rest) = rest
            .split_once('/')
            .with_context(|| format!("ssh:// URL missing path: {s}"))?;
        if user_host_port.is_empty() {
            bail!("ssh:// URL missing user@host: {s}");
        }
        let path = format!("/{path_rest}");
        let (user_host, port) = match user_host_port.rsplit_once(':') {
            Some((uh, p)) => {
                // Heuristic: trailing `:digits` after the final `:` is the port.
                // Anything else (IPv6 literal, etc.) gets left alone.
                match p.parse::<u16>() {
                    Ok(n) => (uh.to_string(), Some(n)),
                    Err(_) => (user_host_port.to_string(), None),
                }
            }
            None => (user_host_port.to_string(), None),
        };
        if user_host.is_empty() {
            bail!("ssh:// URL has empty host (port {port:?})");
        }
        return Ok(BackupSource::Ssh {
            user_host,
            port,
            path,
        });
    }
    bail!("unsupported backup source scheme: {s}");
}

/// Absolute floor: backups smaller than this are an empty or header-only file
/// (an rsync that wrote zero bytes, an empty source). This is the fallback when
/// there is no prior backup to compare against (the first-ever fetch).
const MIN_BACKUP_BYTES: u64 = 64;

/// Plausibility floor as a fraction of the last verified backup's size. A real
/// dump never abruptly halves; a source-side truncation that still closed a
/// valid gzip would pass `gzip -t` and the absolute floor, but not this. Every
/// row in `backups` is a previously-verified dump, so the last one is a sound
/// reference. Integer-halved at the call site.
///
/// A dump *can* legitimately halve, though — a retention prune landing, a bloated
/// table finally being swept. When that happens the floor is self-sealing: no new
/// row is written unless a fetch clears it, so the reference can never advance and
/// every later fetch fails on the same stale number. That is not hypothetical;
/// Sando sat wedged from 2026-06-12 to 2026-07-27 after MNW started deleting
/// expired `tower_sessions` rows and the dump dropped 43 MB -> 6 MB overnight.
/// `force` (operator-supplied, via `POST /backup/fetch {"force":true}`) is the way
/// out: it drops to the absolute floor for one fetch, so the accepted dump becomes
/// the new reference. It never skips `gzip -t` — a truncated file is still refused.
const MIN_BACKUP_FRACTION_DENOM: i64 = 2;

/// Verify a freshly-downloaded backup before it is allowed to become the live
/// dump: reject anything below `min_bytes` (the plausibility floor derived from
/// history), and for a gzip require a complete, valid stream (`gzip -t` fails on
/// truncation/corruption). `is_gz` is taken from the *destination* name, not the
/// temp path (which carries a `.partial` suffix).
async fn verify_backup(tmp_path: &str, is_gz: bool, min_bytes: u64) -> Result<()> {
    let meta = tokio::fs::metadata(tmp_path)
        .await
        .with_context(|| format!("stat fetched backup {tmp_path}"))?;
    anyhow::ensure!(
        meta.len() >= min_bytes,
        "fetched backup {tmp_path} is implausibly small ({} bytes, floor {min_bytes}); \
         treating as a failed/truncated transfer",
        meta.len(),
    );
    if is_gz {
        let out = Command::new("gzip")
            .arg("-t")
            .arg(tmp_path)
            .output()
            .await
            .with_context(|| format!("spawning gzip -t {tmp_path}"))?;
        anyhow::ensure!(
            out.status.success(),
            "fetched backup {tmp_path} failed gzip integrity check (truncated/corrupt): {}",
            String::from_utf8_lossy(&out.stderr),
        );
    }
    Ok(())
}

/// Pull every configured prod dump, or just `only` when named.
///
/// Each dump is fetched independently: one source being down must not leave the
/// others un-refreshed, because a stale dump is a *blocked* gate and the whole
/// point of having more than one is that each database gets its own. So every
/// entry is attempted, and the errors are aggregated at the end — a caller
/// (`/backup/fetch`, and the daily timer through it) still sees a failure, it
/// just sees it after the work that could succeed did.
///
/// `force` re-baselines the plausibility floor: see `MIN_BACKUP_FRACTION_DENOM`.
/// Pass `false` for anything automated — it is an operator escape hatch, not a
/// retry strategy.
pub async fn fetch(
    pool: &SqlitePool,
    cfg: &Arc<AppConfig>,
    topo: &Arc<Topology>,
    force: bool,
    only: Option<&str>,
) -> Result<Vec<FetchedBackup>> {
    let selected: Vec<&crate::topology::BackupConfig> = match only {
        Some(name) => vec![topo.backup_named(name).with_context(|| {
            format!(
                "no backup named {name:?} in the topology (have: {})",
                topo.backup
                    .iter()
                    .map(|b| b.name.as_str())
                    .collect::<Vec<_>>()
                    .join(", ")
            )
        })?],
        None => topo.backup.iter().collect(),
    };

    let mut fetched = Vec::new();
    let mut failures = Vec::new();
    for backup in selected {
        match fetch_one(pool, cfg, backup, force).await {
            Ok(fb) => fetched.push(fb),
            Err(e) => {
                tracing::error!(backup = %backup.name, error = %e, "backup fetch failed");
                failures.push(format!("{}: {e:#}", backup.name));
            }
        }
    }
    anyhow::ensure!(
        failures.is_empty(),
        "{} of {} backup fetch(es) failed: {}",
        failures.len(),
        failures.len() + fetched.len(),
        failures.join("; "),
    );
    Ok(fetched)
}

/// Pull one configured dump into its `local_path`.
async fn fetch_one(
    pool: &SqlitePool,
    cfg: &Arc<AppConfig>,
    backup: &crate::topology::BackupConfig,
    force: bool,
) -> Result<FetchedBackup> {
    let name = backup.name.clone();
    let source = backup.source.clone();
    let local_path = backup.local_path.clone();

    if let Some(parent) = Path::new(&local_path).parent() {
        tokio::fs::create_dir_all(parent).await?;
    }

    // Download to a sibling temp path, verify integrity, then atomically rename
    // into place. The live `local_path` is never the write target, so a partial
    // or corrupt transfer can never become the backup `migration_dry_run`
    // restores (CF4). `--inplace`/`--partial` are deliberately NOT used — those
    // keep a truncated file on failure, the opposite of what we want here.
    let tmp_path = format!("{local_path}.partial");
    let is_gz = std::path::Path::new(&local_path)
        .extension()
        .is_some_and(|ext| ext.eq_ignore_ascii_case("gz"));

    // Plausibility floor: half the last verified backup's size, never below the
    // absolute floor. Scoped to this dump's name — the server's dump is two
    // orders of magnitude larger than multithreaded's, so a shared floor would
    // reject every mt fetch as implausibly small and, on the other side, let a
    // truncated server dump through. The first-ever fetch of a name (no prior
    // row) falls back to the absolute floor.
    let last_size: Option<i64> = sqlx::query_scalar(
        "SELECT byte_size FROM backups
         WHERE app = ? AND name = ? ORDER BY fetched_at DESC LIMIT 1",
    )
    .bind(&cfg.id)
    .bind(&name)
    .fetch_optional(pool)
    .await?;
    let min_bytes = if force {
        tracing::warn!(
            last_verified_bytes = last_size,
            "force: re-baselining the backup plausibility floor to the absolute minimum; \
             this fetch's size becomes the new reference"
        );
        MIN_BACKUP_BYTES
    } else {
        last_size.map_or(MIN_BACKUP_BYTES, |s| {
            ((s / MIN_BACKUP_FRACTION_DENOM) as u64).max(MIN_BACKUP_BYTES)
        })
    };

    let parsed = parse_source(&source)?;
    let downloaded: Result<()> = async {
        match parsed {
            BackupSource::File { path } => {
                tokio::fs::copy(&path, &tmp_path)
                    .await
                    .with_context(|| format!("copy {path} -> {tmp_path}"))?;
            }
            BackupSource::RsyncDaemon { url } => {
                let out = Command::new("rsync")
                    .args(["-az", &url, &tmp_path])
                    .output()
                    .await
                    .context("spawning rsync")?;
                anyhow::ensure!(
                    out.status.success(),
                    "rsync (daemon) failed: {}",
                    String::from_utf8_lossy(&out.stderr),
                );
            }
            BackupSource::Ssh {
                user_host,
                path,
                port,
            } => {
                // Through the executor rather than a hand-rolled `rsync -e ssh`:
                // one transport, one set of SSH flags. The sync plane is now
                // gated (ops_exec::gate_pull): grant `observe:artifact` and
                // confine to the dump's own directory. `path` is operator config
                // (the BackupSource in sando.toml), not attacker input, so the
                // parent-dir root is a formality here — but it keeps this pull
                // fail-closed like every other, rather than an open read of the
                // remote host.
                let caps = CapabilitySet::from_tokens(Vec::<&str>::new(), ["artifact"]);
                let exec = SshExec::new(user_host, caps).with_port(port);
                let exec = match Path::new(&path).parent() {
                    Some(dir) if !dir.as_os_str().is_empty() => exec.with_pull_root(dir),
                    // A bare filename with no directory: confine to the path
                    // itself (starts_with is reflexive), still fail-closed.
                    _ => exec.with_pull_root(&path),
                };
                // NOT `--partial`: a truncated leftover here is dangerous, not
                // useful — a resumed fetch could splice two different dumps into
                // one plausible-looking file (CF4). NOT `-z` either: the dump is
                // already compressed.
                let opts = SyncOpts {
                    compress: false,
                    partial: false,
                    ..SyncOpts::default()
                };
                exec.pull_file(Path::new(&path), Path::new(&tmp_path), &opts)
                    .await
                    .context("rsync (ssh) failed")?;
            }
        }
        verify_backup(&tmp_path, is_gz, min_bytes).await
    }
    .await;

    // On any download/verify failure, remove the temp file so a corrupt
    // `.partial` never lingers, and leave the existing live backup untouched.
    if let Err(e) = downloaded {
        let _ = tokio::fs::remove_file(&tmp_path).await;
        return Err(e);
    }

    tokio::fs::rename(&tmp_path, &local_path)
        .await
        .with_context(|| format!("atomic rename {tmp_path} -> {local_path}"))?;

    let meta = tokio::fs::metadata(&local_path).await?;
    let size = meta.len() as i64;

    // Record the new backup and prune stale rows in one transaction, so a crash
    // between the two can't leave the insert without the prune (or, worse, lose
    // the insert while keeping a half-applied delete). The on-disk file is
    // overwritten each fetch (single `local_path`), so rows older than 30 days
    // reference a path that no longer exists — keep the table from growing.
    let mut tx = pool.begin().await?;
    sqlx::query(
        "INSERT INTO backups (app, name, fetched_at, source, local_path, byte_size) \
         VALUES (?, ?, ?, ?, ?, ?)",
    )
    .bind(&cfg.id)
    .bind(&name)
    .bind(Utc::now().to_rfc3339())
    .bind(&source)
    .bind(&local_path)
    .bind(size)
    .execute(&mut *tx)
    .await?;
    // Scoped: one product's fetch is not a licence to prune another's history,
    // and the 30-day window is about this product's own overwritten dumps.
    sqlx::query("DELETE FROM backups WHERE app = ? AND fetched_at < datetime('now', '-30 days')")
        .bind(&cfg.id)
        .execute(&mut *tx)
        .await?;
    tx.commit().await?;

    Ok(FetchedBackup {
        name,
        source,
        local_path,
        byte_size: Some(size),
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::topology::{BackupConfig, RepoConfig};

    // ---- CF4: atomic write + integrity ----

    async fn mem_pool() -> SqlitePool {
        let pool = sqlx::sqlite::SqlitePoolOptions::new()
            .max_connections(1)
            .connect("sqlite::memory:")
            .await
            .unwrap();
        crate::db::migrate(&pool).await.unwrap();
        pool
    }

    fn topo_with_backup(source: String, local_path: String) -> Topology {
        Topology {
            repo: Some(RepoConfig {
                bare_path: "/tmp/x.git".into(),
                branch: "main".into(),
                upstream: None,
            }),
            backup: vec![BackupConfig {
                name: "server".into(),
                source,
                local_path,
            }],
            tiers: vec![],
            aux_repos: Vec::new(),
        }
    }

    /// ~4 KB of poorly-compressible bytes so a gzip of it stays well above the
    /// size floor and a half-truncation lands mid-stream (failing `gzip -t`).
    fn incompressible(n: usize) -> Vec<u8> {
        (0..n)
            .map(|i| (i.wrapping_mul(2_654_435_761) >> 13) as u8)
            .collect()
    }

    async fn write_valid_gz(path: &Path) {
        let plain = path.with_extension("plain");
        tokio::fs::write(&plain, incompressible(4096))
            .await
            .unwrap();
        let out = Command::new("sh")
            .arg("-c")
            .arg(format!("gzip -c {} > {}", plain.display(), path.display()))
            .output()
            .await
            .unwrap();
        assert!(out.status.success(), "gzip shim failed");
    }

    #[tokio::test]
    async fn fetch_file_source_writes_atomically_and_records_row() {
        let tmp = tempfile::tempdir().unwrap();
        let src = tmp.path().join("src.sql.gz");
        write_valid_gz(&src).await;
        let dest = tmp.path().join("backups/latest.sql.gz");
        let topo = Arc::new(topo_with_backup(
            format!("file://{}", src.display()),
            dest.to_string_lossy().into_owned(),
        ));
        let pool = mem_pool().await;
        let cfg = Arc::new(AppConfig::for_tests());

        let fb = fetch(&pool, &cfg, &topo, false, None)
            .await
            .unwrap()
            .remove(0);
        assert!(dest.exists(), "live backup written");
        assert!(
            !dest.with_file_name("latest.sql.gz.partial").exists(),
            "temp file consumed by the atomic rename",
        );
        assert!(fb.byte_size.unwrap() > 0);
        let count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM backups")
            .fetch_one(&pool)
            .await
            .unwrap();
        assert_eq!(count.0, 1, "a row is recorded for a successful fetch");
    }

    #[tokio::test]
    async fn fetch_rejects_a_dump_far_below_the_last_backup_size() {
        // Plausibility floor: a complete, valid gzip that is far smaller than the
        // last verified backup is a likely source-side truncation and is rejected
        // even though `gzip -t` passes — the gap the 64-byte absolute floor missed.
        let tmp = tempfile::tempdir().unwrap();
        let src = tmp.path().join("src.sql.gz");
        write_valid_gz(&src).await; // ~4 KB valid gzip
        let dest = tmp.path().join("backups/latest.sql.gz");
        let topo = Arc::new(topo_with_backup(
            format!("file://{}", src.display()),
            dest.to_string_lossy().into_owned(),
        ));
        let pool = mem_pool().await;
        let cfg = Arc::new(AppConfig::for_tests());

        // Seed a prior verified backup far larger than the incoming one; the floor
        // becomes 500_000, well above the ~4 KB dump.
        sqlx::query("INSERT INTO backups (fetched_at, source, local_path, byte_size) VALUES (?, 'x', ?, 1000000)")
            .bind(Utc::now().to_rfc3339())
            .bind(dest.to_string_lossy().into_owned())
            .execute(&pool).await.unwrap();

        let err = fetch(&pool, &cfg, &topo, false, None)
            .await
            .unwrap_err()
            .to_string();
        assert!(err.contains("implausibly small"), "{err}");
        assert!(
            !dest.exists(),
            "a rejected dump never becomes the live backup"
        );
        let count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM backups")
            .fetch_one(&pool)
            .await
            .unwrap();
        assert_eq!(count.0, 1, "the rejected fetch records no new row");
    }

    #[tokio::test]
    async fn the_plausibility_floor_is_scoped_to_one_dump() {
        // The server's dump is two orders of magnitude larger than
        // multithreaded's. A shared floor would reject every mt fetch as
        // implausibly small (and, the other way round, let a badly truncated
        // server dump through on mt's reference). Seed a large `server` row and
        // fetch a small `multithreaded` one: it must be accepted.
        let tmp = tempfile::tempdir().unwrap();
        let src = tmp.path().join("mt.sql.gz");
        write_valid_gz(&src).await; // ~4 KB valid gzip
        let dest = tmp.path().join("backups/mt-latest.sql.gz");
        let mut topo = topo_with_backup(
            format!("file://{}", src.display()),
            dest.to_string_lossy().into_owned(),
        );
        topo.backup[0].name = "multithreaded".into();
        let topo = Arc::new(topo);
        let pool = mem_pool().await;
        let cfg = Arc::new(AppConfig::for_tests());

        sqlx::query(
            "INSERT INTO backups (name, fetched_at, source, local_path, byte_size) \
             VALUES ('server', ?, 'x', '/tmp/server.sql.gz', 1000000)",
        )
        .bind(Utc::now().to_rfc3339())
        .execute(&pool)
        .await
        .unwrap();

        let fetched = fetch(&pool, &cfg, &topo, false, None)
            .await
            .expect("the server's size must not set multithreaded's floor");
        assert_eq!(fetched.len(), 1);
        assert_eq!(fetched[0].name, "multithreaded");
        let recorded: (String,) =
            sqlx::query_as("SELECT name FROM backups ORDER BY id DESC LIMIT 1")
                .fetch_one(&pool)
                .await
                .unwrap();
        assert_eq!(recorded.0, "multithreaded", "the row is recorded by name");
    }

    #[tokio::test]
    async fn fetching_an_unknown_name_is_an_error_not_a_silent_no_op() {
        // A typo'd `{"name":"mt"}` must not report success having fetched
        // nothing — the operator would read that as a refreshed dump.
        let tmp = tempfile::tempdir().unwrap();
        let topo = Arc::new(topo_with_backup(
            "file:///nope".into(),
            tmp.path().join("x.sql.gz").to_string_lossy().into_owned(),
        ));
        let pool = mem_pool().await;
        let cfg = Arc::new(AppConfig::for_tests());

        let err = fetch(&pool, &cfg, &topo, false, Some("mt"))
            .await
            .unwrap_err()
            .to_string();
        assert!(err.contains("no backup named"), "{err}");
    }

    #[tokio::test]
    async fn one_failing_source_does_not_skip_the_others() {
        // Each dump gates a different database, and a stale dump is a blocked
        // gate — so a broken source must not cost the working one its refresh.
        let tmp = tempfile::tempdir().unwrap();
        let good_src = tmp.path().join("good.sql.gz");
        write_valid_gz(&good_src).await;
        let good_dest = tmp.path().join("backups/good.sql.gz");
        let mut topo = topo_with_backup(
            "file:///nonexistent/sando-test-missing.sql.gz".into(),
            tmp.path()
                .join("backups/bad.sql.gz")
                .to_string_lossy()
                .into_owned(),
        );
        topo.backup.push(crate::topology::BackupConfig {
            name: "multithreaded".into(),
            source: format!("file://{}", good_src.display()),
            local_path: good_dest.to_string_lossy().into_owned(),
        });
        let topo = Arc::new(topo);
        let pool = mem_pool().await;
        let cfg = Arc::new(AppConfig::for_tests());

        let err = fetch(&pool, &cfg, &topo, false, None)
            .await
            .unwrap_err()
            .to_string();
        assert!(err.contains("server:"), "the failure names its dump: {err}");
        assert!(
            good_dest.exists(),
            "the reachable dump is still fetched after the unreachable one fails"
        );
    }

    #[tokio::test]
    async fn force_rebaselines_the_floor_after_a_legitimate_shrink() {
        // The wedge this exists for: the floor is derived from a row that only a
        // passing fetch can replace, so a dump that legitimately halves locks the
        // fetch out permanently. `force` accepts one undersized dump and makes it
        // the new reference, unwedging the next ordinary fetch.
        let tmp = tempfile::tempdir().unwrap();
        let src = tmp.path().join("src.sql.gz");
        write_valid_gz(&src).await;
        let dest = tmp.path().join("backups/latest.sql.gz");
        let topo = Arc::new(topo_with_backup(
            format!("file://{}", src.display()),
            dest.to_string_lossy().into_owned(),
        ));
        let pool = mem_pool().await;
        let cfg = Arc::new(AppConfig::for_tests());
        sqlx::query("INSERT INTO backups (fetched_at, source, local_path, byte_size) VALUES (?, 'x', ?, 1000000)")
            .bind(Utc::now().to_rfc3339())
            .bind(dest.to_string_lossy().into_owned())
            .execute(&pool).await.unwrap();

        // Same dump the un-forced fetch rejects above.
        let fb = fetch(&pool, &cfg, &topo, true, None)
            .await
            .unwrap()
            .remove(0);
        assert!(dest.exists(), "the forced dump becomes the live backup");
        let recorded = fb.byte_size.unwrap();
        assert!(
            recorded < 1_000_000,
            "the accepted dump really is the small one"
        );

        // The new row is now the reference, so the next fetch passes unforced.
        fetch(&pool, &cfg, &topo, false, None)
            .await
            .expect("floor re-baselined to the forced fetch's size");
    }

    #[tokio::test]
    async fn force_still_rejects_a_corrupt_gzip() {
        // `force` relaxes the size floor only. A truncated dump is refused either
        // way — otherwise the escape hatch would be a way to install a broken
        // backup as the thing migration_dry_run restores.
        let tmp = tempfile::tempdir().unwrap();
        let src = tmp.path().join("src.sql.gz");
        write_valid_gz(&src).await;
        let whole = tokio::fs::read(&src).await.unwrap();
        tokio::fs::write(&src, &whole[..whole.len() / 2])
            .await
            .unwrap();
        let dest = tmp.path().join("backups/latest.sql.gz");
        let topo = Arc::new(topo_with_backup(
            format!("file://{}", src.display()),
            dest.to_string_lossy().into_owned(),
        ));
        let pool = mem_pool().await;
        let cfg = Arc::new(AppConfig::for_tests());

        let err = fetch(&pool, &cfg, &topo, true, None)
            .await
            .unwrap_err()
            .to_string();
        assert!(err.contains("gzip integrity check"), "{err}");
        assert!(
            !dest.exists(),
            "a corrupt dump never becomes the live backup"
        );
    }

    #[tokio::test]
    async fn fetch_rejects_truncated_gz_and_leaves_no_live_file() {
        let tmp = tempfile::tempdir().unwrap();
        let src = tmp.path().join("src.sql.gz");
        write_valid_gz(&src).await;
        // Truncate to half: a valid gzip prefix that fails `gzip -t` mid-stream.
        let full = tokio::fs::read(&src).await.unwrap();
        assert!(
            full.len() / 2 > MIN_BACKUP_BYTES as usize,
            "half must clear the size floor to exercise gzip -t"
        );
        tokio::fs::write(&src, &full[..full.len() / 2])
            .await
            .unwrap();

        let dest = tmp.path().join("backups/latest.sql.gz");
        let topo = Arc::new(topo_with_backup(
            format!("file://{}", src.display()),
            dest.to_string_lossy().into_owned(),
        ));
        let pool = mem_pool().await;
        let cfg = Arc::new(AppConfig::for_tests());

        let res = fetch(&pool, &cfg, &topo, false, None).await;
        assert!(res.is_err(), "a truncated gzip must fail the fetch");
        assert!(
            !dest.exists(),
            "no live backup file results from a failed fetch"
        );
        assert!(
            !dest.with_file_name("latest.sql.gz.partial").exists(),
            "the corrupt temp file is cleaned up",
        );
        let count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM backups")
            .fetch_one(&pool)
            .await
            .unwrap();
        assert_eq!(count.0, 0, "no row recorded for a failed fetch");
    }

    #[test]
    fn parses_file_url() {
        let s = parse_source("file:///opt/backups/latest.sql.gz").unwrap();
        assert_eq!(
            s,
            BackupSource::File {
                path: "/opt/backups/latest.sql.gz".into()
            }
        );
    }

    #[test]
    fn file_url_without_path_errors() {
        assert!(parse_source("file://").is_err());
    }

    #[test]
    fn parses_rsync_daemon_url() {
        let s = parse_source("rsync://astra/mnw/latest.sql.gz").unwrap();
        assert_eq!(
            s,
            BackupSource::RsyncDaemon {
                url: "rsync://astra/mnw/latest.sql.gz".into()
            }
        );
    }

    #[test]
    fn parses_ssh_url_with_port() {
        let s = parse_source("ssh://backup-puller@alpha-west-1:2200/latest.sql.gz").unwrap();
        assert_eq!(
            s,
            BackupSource::Ssh {
                user_host: "backup-puller@alpha-west-1".into(),
                port: Some(2200),
                path: "/latest.sql.gz".into(),
            }
        );
    }

    #[test]
    fn parses_ssh_url_without_port() {
        let s = parse_source("ssh://max@astra/opt/backups/mnw/latest.sql.gz").unwrap();
        assert_eq!(
            s,
            BackupSource::Ssh {
                user_host: "max@astra".into(),
                port: None,
                path: "/opt/backups/mnw/latest.sql.gz".into(),
            }
        );
    }

    #[test]
    fn ssh_url_without_path_errors() {
        // `split_once('/')` — `ssh://user@host` has no `/` after the scheme.
        assert!(parse_source("ssh://backup-puller@alpha-west-1").is_err());
    }

    #[test]
    fn ssh_url_without_user_host_errors() {
        // Empty user@host: `ssh:///foo`. Caught by the empty-prefix check.
        assert!(parse_source("ssh:///latest.sql.gz").is_err());
    }

    #[test]
    fn ssh_url_with_non_numeric_after_colon_treats_as_part_of_host() {
        // `host:notaport` should NOT parse `notaport` as a port. Leave the
        // colon part of user_host; libssh/rsync will reject if truly wrong.
        let s = parse_source("ssh://user@host:notaport/path").unwrap();
        assert_eq!(
            s,
            BackupSource::Ssh {
                user_host: "user@host:notaport".into(),
                port: None,
                path: "/path".into(),
            }
        );
    }

    #[test]
    fn rejects_unknown_scheme() {
        assert!(parse_source("ftp://example.com/file").is_err());
        assert!(parse_source("just-a-path.sql.gz").is_err());
        assert!(parse_source("").is_err());
    }

    #[test]
    fn ssh_url_preserves_multi_segment_path() {
        let s = parse_source("ssh://a@b:22/opt/foo/bar/baz.sql.gz").unwrap();
        match s {
            BackupSource::Ssh { path, .. } => assert_eq!(path, "/opt/foo/bar/baz.sql.gz"),
            _ => panic!("wrong variant"),
        }
    }
}