//! Standby replication: pushes every repo and a database snapshot to a passive SSH host (see
//! `docs/backup.md`, "Standby replication"). The standby is a dedicated unix account that holds
//! bare repos and snapshots. It never runs forge.
//!
//! Everything that reaches the standby goes through [`StandbyTransport`]. [`SshTransport`] runs
//! the system `ssh`; [`LocalTransport`] (tests only) runs the same shell commands with `sh -c` on
//! this machine, where the remote base is a temporary directory. Remote paths are validated and
//! quoted before they are placed in any command.

pub mod worker;

use std::collections::HashMap;
use std::future::Future;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Mutex, PoisonError};
use std::time::{Duration, Instant};

use forge_git::GitEnv;
use forge_git::command::GitCommand;
use tokio::process::Command;
use tokio::sync::Notify;

use crate::state::AppState;

/// How long a repo waits after its last ref update before it is pushed.
pub const PUSH_DEBOUNCE: Duration = Duration::from_secs(5);
/// Newest database snapshots kept on the standby.
pub const KEEP_SNAPSHOTS: usize = 24;
/// Name of the snapshot directory on the standby, under the base.
const DB_DIR: &str = "db";
/// Name of the repo directory on the standby, under the base.
const REPOS_DIR: &str = "repos";
/// Suffix of the replication status subject that names a repo's wiki.
const WIKI_SUBJECT_SUFFIX: &str = ".wiki";
/// Prefix and suffix of a snapshot file name: `forge-<UTC stamp>.db`.
const SNAPSHOT_PREFIX: &str = "forge-";
const SNAPSHOT_SUFFIX: &str = ".db";
/// Stamp layout in snapshot names: `YYYYMMDDTHHMMSSZ`.
const STAMP_LEN: usize = 16;
const REFS_FORMAT: &str = "--format=%(objectname) %(refname)";

/// Checks a remote base or repo path: an absolute path of `[A-Za-z0-9._/-]`, with no `..`, no
/// empty components and no trailing slash.
pub fn validate_remote_path(path: &str) -> Result<(), String> {
    let well_formed = path.starts_with('/')
        && path.len() > 1
        && !path.ends_with('/')
        && !path.contains("//")
        && !path.contains("..")
        && path
            .bytes()
            .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'.' | b'_' | b'/' | b'-'));
    if well_formed {
        Ok(())
    } else {
        Err(format!(
            "remote path must be absolute, use only [A-Za-z0-9._/-], and have no '..': {path:?}"
        ))
    }
}

/// The replication status subject of a repo's wiki: `<id>.wiki`.
pub fn wiki_subject(repo_id: &str) -> String {
    format!("{repo_id}{WIKI_SUBJECT_SUFFIX}")
}

/// The repo id a wiki subject names, or `None` for any other subject.
pub fn wiki_subject_repo(subject: &str) -> Option<&str> {
    subject.strip_suffix(WIKI_SUBJECT_SUFFIX)
}

/// Repo ids are ULIDs: 26 ASCII alphanumerics. Checked before an id is used in a remote path.
pub fn validate_repo_id(id: &str) -> Result<(), String> {
    if id.len() == 26 && id.bytes().all(|b| b.is_ascii_alphanumeric()) {
        Ok(())
    } else {
        Err(format!("not a repo id: {id:?}"))
    }
}

/// Quotes one argument for a POSIX shell: wraps it in single quotes and writes each embedded
/// single quote as `'\''`. The result is one word whatever the input holds.
pub fn sh_quote(arg: &str) -> String {
    format!("'{}'", arg.replace('\'', "'\\''"))
}

/// Where the standby is and how to reach it: `[user@]host:/absolute/base` plus the SSH port.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Target {
    pub user: Option<String>,
    pub host: String,
    pub port: u16,
    /// Validated absolute base directory on the standby.
    pub base: String,
}

impl Target {
    /// Parses the `target` setting. Host and user may use `[A-Za-z0-9.-]` and `[A-Za-z0-9._-]`;
    /// neither may start with `-`, so neither can be read by ssh as an option.
    pub fn parse(spec: &str, port: u16) -> Result<Target, String> {
        let (dest, base) = spec
            .split_once(':')
            .ok_or_else(|| "replication target must be [user@]host:/absolute/path".to_string())?;
        let (user, host) = match dest.split_once('@') {
            Some((user, host)) => (Some(user), host),
            None => (None, dest),
        };
        if let Some(user) = user {
            let ok = !user.is_empty()
                && !user.starts_with('-')
                && user
                    .bytes()
                    .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'.' | b'_' | b'-'));
            if !ok {
                return Err(format!("invalid user in replication target: {user:?}"));
            }
        }
        let host_ok = !host.is_empty()
            && !host.starts_with('-')
            && host
                .bytes()
                .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'.' | b'-'));
        if !host_ok {
            return Err(format!("invalid host in replication target: {host:?}"));
        }
        validate_remote_path(base)?;
        Ok(Target {
            user: user.map(str::to_owned),
            host: host.to_owned(),
            port,
            base: base.to_owned(),
        })
    }

    /// `user@host`, or `host` without a user.
    pub fn destination(&self) -> String {
        match &self.user {
            Some(user) => format!("{user}@{}", self.host),
            None => self.host.clone(),
        }
    }

    /// The `ssh://` URL git pushes a repo to.
    pub fn repo_url(&self, remote_repo_dir: &str) -> String {
        format!(
            "ssh://{}:{}{remote_repo_dir}",
            self.destination(),
            self.port
        )
    }
}

/// Remote layout under the base. `repos/<id>.git` holds bare repos, `db/` holds snapshots and the
/// `latest` link. `keys/` belongs to the operator and is never written here.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Layout {
    pub base: String,
}

impl Layout {
    pub fn new(base: &str) -> Result<Layout, String> {
        validate_remote_path(base)?;
        Ok(Layout {
            base: base.to_owned(),
        })
    }

    pub fn repos_dir(&self) -> String {
        format!("{}/{REPOS_DIR}", self.base)
    }

    pub fn db_dir(&self) -> String {
        format!("{}/{DB_DIR}", self.base)
    }

    /// `<base>/repos/<id>.git`. Fails for an id that is not a ULID.
    pub fn repo_dir(&self, repo_id: &str) -> Result<String, String> {
        validate_repo_id(repo_id)?;
        Ok(format!("{}/{repo_id}.git", self.repos_dir()))
    }

    /// `<base>/repos/<id>.wiki.git`: the standby copy of the repo's wiki.
    pub fn wiki_dir(&self, repo_id: &str) -> Result<String, String> {
        validate_repo_id(repo_id)?;
        Ok(format!("{}/{repo_id}.wiki.git", self.repos_dir()))
    }

    /// `<base>/db/forge-<stamp>.db`.
    pub fn snapshot_file(&self, stamp: &str) -> String {
        format!(
            "{}/{SNAPSHOT_PREFIX}{stamp}{SNAPSHOT_SUFFIX}",
            self.db_dir()
        )
    }

    pub fn latest_link(&self) -> String {
        format!("{}/latest", self.db_dir())
    }
}

/// A snapshot name `forge-<16-char stamp>.db` with a stamp of digits, `T` and `Z`.
pub fn is_snapshot_name(name: &str) -> bool {
    let Some(stamp) = name
        .strip_prefix(SNAPSHOT_PREFIX)
        .and_then(|rest| rest.strip_suffix(SNAPSHOT_SUFFIX))
    else {
        return false;
    };
    stamp.len() == STAMP_LEN
        && stamp.bytes().enumerate().all(|(i, b)| match i {
            8 => b == b'T',
            15 => b == b'Z',
            _ => b.is_ascii_digit(),
        })
}

/// The UTC stamp used in snapshot names, `YYYYMMDDTHHMMSSZ`.
pub fn snapshot_stamp() -> String {
    const FORMAT: &[time::format_description::FormatItem<'static>] =
        time::macros::format_description!("[year][month][day]T[hour][minute][second]Z");
    time::OffsetDateTime::now_utc()
        .format(FORMAT)
        .unwrap_or_else(|_| "00000000T000000Z".to_owned())
}

/// Moves data to the standby and runs commands there. Implemented by [`SshTransport`] and, in
/// tests, [`LocalTransport`]. The worker only calls these three methods.
pub trait StandbyTransport: Send + Sync {
    /// Runs one shell command (already quoted) on the standby and returns its stdout.
    fn run(&self, remote_cmd: &str) -> impl Future<Output = Result<Vec<u8>, String>> + Send;

    /// The URL git pushes the repo at `remote_repo_dir` to.
    fn repo_url(&self, remote_repo_dir: &str) -> String;

    /// Whether git must be told to allow a plain local path as a transport.
    fn local_git(&self) -> bool;

    /// The `GIT_SSH_COMMAND` git must use to reach the standby, if any.
    fn push_ssh_command(&self) -> Option<String>;

    /// Streams the local file `local` to `remote_path` on the standby. The data goes to a temporary
    /// file in the same directory first, and is renamed into place only once complete.
    fn upload(
        &self,
        local: &Path,
        remote_path: &str,
    ) -> impl Future<Output = Result<(), String>> + Send;
}

/// Builds the `cat > tmp && mv tmp final` command used by every upload.
fn upload_command(remote_path: &str) -> String {
    let (dir, name) = remote_path.rsplit_once('/').unwrap_or(("", remote_path));
    let tmp = format!("{dir}/.tmp-{name}");
    format!(
        "cat > {} && mv -f {} {}",
        sh_quote(&tmp),
        sh_quote(&tmp),
        sh_quote(remote_path)
    )
}

/// Runs `command` with its stdin fed from `file`, and fails on a non-zero exit.
async fn pipe_file(mut command: Command, file: &Path) -> Result<(), String> {
    use std::process::Stdio;
    use tokio::io::AsyncWriteExt;
    command
        .stdin(Stdio::piped())
        .stdout(Stdio::null())
        .stderr(Stdio::piped())
        .kill_on_drop(true);
    let mut child = command
        .spawn()
        .map_err(|e| format!("cannot start ssh: {e}"))?;
    let mut stdin = child
        .stdin
        .take()
        .ok_or_else(|| "ssh stdin is not available".to_string())?;
    let mut source = tokio::fs::File::open(file)
        .await
        .map_err(|e| format!("cannot read {}: {e}", file.display()))?;
    tokio::io::copy(&mut source, &mut stdin)
        .await
        .map_err(|e| format!("upload failed: {e}"))?;
    stdin
        .shutdown()
        .await
        .map_err(|e| format!("upload failed: {e}"))?;
    drop(stdin);
    let out = child
        .wait_with_output()
        .await
        .map_err(|e| format!("upload failed: {e}"))?;
    if out.status.success() {
        Ok(())
    } else {
        Err(format!(
            "upload exited with {}: {}",
            out.status,
            String::from_utf8_lossy(&out.stderr).trim()
        ))
    }
}

/// Runs `command` and returns stdout. A non-zero exit is an error carrying stderr.
async fn capture(mut command: Command) -> Result<Vec<u8>, String> {
    use std::process::Stdio;
    command.stdin(Stdio::null()).kill_on_drop(true);
    let out = command
        .output()
        .await
        .map_err(|e| format!("cannot start ssh: {e}"))?;
    if out.status.success() {
        Ok(out.stdout)
    } else {
        Err(format!(
            "remote command exited with {}: {}",
            out.status,
            String::from_utf8_lossy(&out.stderr).trim()
        ))
    }
}

/// Pushes every ref of the local bare repo `repo_dir` to `url` with `--mirror --force`.
async fn push_mirror(
    git: &GitEnv,
    repo_dir: &Path,
    url: &str,
    local_git: bool,
    ssh_command: Option<String>,
) -> Result<(), String> {
    let mut cmd = GitCommand::new(git, "push")
        .git_dir(repo_dir)
        .arg("--mirror")
        .arg("--force")
        .arg("--quiet");
    if local_git {
        cmd = cmd.allow_local_transport();
    }
    if let Some(ssh) = ssh_command {
        cmd = cmd.env("GIT_SSH_COMMAND", ssh);
    }
    cmd.arg("--end-of-options")
        .user_arg(url)
        .map_err(|e| e.to_string())?
        .status()
        .await
        .map_err(|e| e.to_string())
}

/// Replication over the system `ssh`, with the options that pin the standby's host key and use
/// only the configured identity. Argument construction is unit-tested; no test connects.
#[derive(Debug, Clone)]
pub struct SshTransport {
    target: Target,
    identity_file: PathBuf,
    known_hosts_file: PathBuf,
}

impl SshTransport {
    pub fn new(target: Target, identity_file: PathBuf, known_hosts_file: PathBuf) -> SshTransport {
        SshTransport {
            target,
            identity_file,
            known_hosts_file,
        }
    }

    /// The `-i`/`-o`/`-p` options, in order. Every one is a fixed word or a configured path.
    pub fn ssh_options(&self) -> Vec<String> {
        vec![
            "-i".to_owned(),
            self.identity_file.to_string_lossy().into_owned(),
            "-o".to_owned(),
            "IdentitiesOnly=yes".to_owned(),
            "-o".to_owned(),
            format!(
                "UserKnownHostsFile={}",
                self.known_hosts_file.to_string_lossy()
            ),
            "-o".to_owned(),
            "StrictHostKeyChecking=yes".to_owned(),
            "-o".to_owned(),
            "BatchMode=yes".to_owned(),
            "-o".to_owned(),
            "ConnectTimeout=20".to_owned(),
            "-p".to_owned(),
            self.target.port.to_string(),
        ]
    }

    /// The full argv after the program name: options, destination, then the remote command.
    pub fn ssh_argv(&self, remote_cmd: &str) -> Vec<String> {
        let mut argv = self.ssh_options();
        argv.push(self.target.destination());
        argv.push(remote_cmd.to_owned());
        argv
    }

    /// The value for `GIT_SSH_COMMAND`: the same ssh invocation, each word quoted for the shell.
    pub fn git_ssh_command(&self) -> String {
        std::iter::once("ssh".to_owned())
            .chain(self.ssh_options())
            .map(|word| sh_quote(&word))
            .collect::<Vec<_>>()
            .join(" ")
    }

    fn command(&self, remote_cmd: &str) -> Command {
        let mut cmd = Command::new("ssh");
        cmd.args(self.ssh_argv(remote_cmd));
        cmd
    }
}

impl StandbyTransport for SshTransport {
    fn run(&self, remote_cmd: &str) -> impl Future<Output = Result<Vec<u8>, String>> + Send {
        let cmd = self.command(remote_cmd);
        capture(cmd)
    }

    fn repo_url(&self, remote_repo_dir: &str) -> String {
        self.target.repo_url(remote_repo_dir)
    }

    fn local_git(&self) -> bool {
        false
    }

    fn push_ssh_command(&self) -> Option<String> {
        Some(self.git_ssh_command())
    }

    fn upload(
        &self,
        local: &Path,
        remote_path: &str,
    ) -> impl Future<Output = Result<(), String>> + Send {
        let cmd = self.command(&upload_command(remote_path));
        let local = local.to_path_buf();
        async move { pipe_file(cmd, &local).await }
    }
}

/// Runs the same shell commands as [`SshTransport`], but with `sh -c` on this machine. The remote
/// base is then a local directory. Test only.
#[cfg(test)]
#[derive(Debug, Clone, Copy, Default)]
pub struct LocalTransport;

#[cfg(test)]
impl StandbyTransport for LocalTransport {
    fn run(&self, remote_cmd: &str) -> impl Future<Output = Result<Vec<u8>, String>> + Send {
        let mut cmd = Command::new("sh");
        cmd.arg("-c").arg(remote_cmd);
        capture(cmd)
    }

    fn repo_url(&self, remote_repo_dir: &str) -> String {
        remote_repo_dir.to_owned()
    }

    fn local_git(&self) -> bool {
        true
    }

    fn push_ssh_command(&self) -> Option<String> {
        None
    }

    fn upload(
        &self,
        local: &Path,
        remote_path: &str,
    ) -> impl Future<Output = Result<(), String>> + Send {
        let mut cmd = Command::new("sh");
        cmd.arg("-c").arg(upload_command(remote_path));
        let local = local.to_path_buf();
        async move { pipe_file(cmd, &local).await }
    }
}

/// Pushes one local repo with `--mirror --force` through a transport. The remote directory must
/// already exist (see `worker::replicate_repo`).
pub async fn push_repo<T: StandbyTransport>(
    transport: &T,
    git: &GitEnv,
    repo_dir: &Path,
    remote_repo_dir: &str,
) -> Result<(), String> {
    push_mirror(
        git,
        repo_dir,
        &transport.repo_url(remote_repo_dir),
        transport.local_git(),
        transport.push_ssh_command(),
    )
    .await
}

/// The refs checksum of the bare repo at `remote_repo_dir`, read through the transport.
pub async fn remote_refs_checksum<T: StandbyTransport>(
    transport: &T,
    remote_repo_dir: &str,
) -> Result<String, String> {
    let cmd = format!(
        "git -C {} for-each-ref {}",
        sh_quote(remote_repo_dir),
        sh_quote(REFS_FORMAT)
    );
    let out = transport.run(&cmd).await?;
    Ok(forge_git::backup::refs_checksum_of_listing(&out))
}

/// The queue the web side writes and the replication worker drains. Holds repos waiting for a
/// debounced push, and the "reconcile now" request.
#[derive(Default)]
pub struct ReplicationQueue {
    notify: Notify,
    pending: Mutex<HashMap<String, Instant>>,
    reconcile: AtomicBool,
}

impl ReplicationQueue {
    /// Queues `repo_id` for a push once [`PUSH_DEBOUNCE`] has passed without another update.
    pub fn enqueue(&self, repo_id: &str) {
        let due = Instant::now() + PUSH_DEBOUNCE;
        self.pending
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .insert(repo_id.to_owned(), due);
        self.notify.notify_one();
    }

    /// Removes and returns the repos whose debounce has passed at `now`, sorted by id.
    pub fn take_due(&self, now: Instant) -> Vec<String> {
        let mut pending = self.pending.lock().unwrap_or_else(PoisonError::into_inner);
        let mut due: Vec<String> = pending
            .iter()
            .filter(|(_, at)| **at <= now)
            .map(|(id, _)| id.clone())
            .collect();
        due.sort();
        for id in &due {
            pending.remove(id);
        }
        due
    }

    /// Asks the worker for a full reconcile pass as soon as it can.
    pub fn request_reconcile(&self) {
        self.reconcile.store(true, Ordering::SeqCst);
        self.notify.notify_one();
    }

    /// Returns true once per request.
    pub fn take_reconcile_request(&self) -> bool {
        self.reconcile.swap(false, Ordering::SeqCst)
    }

    /// Resolves when something is queued or a reconcile is requested.
    pub async fn wait(&self) {
        self.notify.notified().await;
    }
}

/// Queues a repo after any ref update. Does nothing when replication is disabled.
pub fn enqueue_repo(state: &AppState, repo_id: &str) {
    if state.config.replication.enabled() {
        state.replication.enqueue(repo_id);
    }
}

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

    #[test]
    fn remote_paths_must_be_plain_absolute_paths() {
        for ok in ["/srv/forge-standby", "/srv/a.b_c-d/e", "/x"] {
            assert!(validate_remote_path(ok).is_ok(), "{ok} should pass");
        }
        for bad in [
            "",
            "/",
            "srv/forge",
            "./srv",
            "/srv/../etc",
            "/srv/..",
            "/srv/a/../b",
            "/srv//x",
            "/srv/x/",
            "/srv/with space",
            "/srv/quote'",
            "/srv/$HOME",
            "/srv/a;rm",
            "/srv/a`b`",
            "/srv/é",
            "/srv/x\n",
        ] {
            assert!(validate_remote_path(bad).is_err(), "{bad:?} should fail");
        }
    }

    #[test]
    fn repo_ids_must_be_ulid_shaped() {
        assert!(validate_repo_id("01ARZ3NDEKTSV4RRFFQ69G5FAV").is_ok());
        assert!(validate_repo_id("01ARZ3NDEKTSV4RRFFQ69G5FA").is_err());
        assert!(validate_repo_id("../../etc/passwd0000000000").is_err());
        assert!(validate_repo_id("01ARZ3NDEKTSV4RRFFQ69G5F'V").is_err());
    }

    #[test]
    fn quoting_survives_spaces_quotes_and_metacharacters() {
        assert_eq!(sh_quote("/srv/x"), "'/srv/x'");
        assert_eq!(sh_quote(""), "''");
        assert_eq!(sh_quote("a b"), "'a b'");
        assert_eq!(sh_quote("it's"), "'it'\\''s'");
        assert_eq!(sh_quote("$(rm -rf /)`x`;|&"), "'$(rm -rf /)`x`;|&'");
        assert_eq!(sh_quote("'"), "''\\'''");
        // Round trip through a real shell: each quoted word comes back unchanged.
        let words = ["plain", "it's", "a b", "$HOME", "x'y'z", "back\\slash"];
        let line = words
            .iter()
            .map(|w| sh_quote(w))
            .collect::<Vec<_>>()
            .join(" ");
        let script = format!("printf '%s\\0' {line}");
        let out = std::process::Command::new("sh")
            .arg("-c")
            .arg(&script)
            .output()
            .expect("sh runs");
        let got: Vec<String> = out
            .stdout
            .split(|b| *b == 0)
            .filter(|s| !s.is_empty())
            .map(|s| String::from_utf8_lossy(s).into_owned())
            .collect();
        assert_eq!(got, words);
    }

    #[test]
    fn target_parses_user_host_and_base() {
        let t = Target::parse("sb@standby.example:/srv/forge-standby", 2222).expect("parse");
        assert_eq!(t.user.as_deref(), Some("sb"));
        assert_eq!(t.host, "standby.example");
        assert_eq!(t.port, 2222);
        assert_eq!(t.base, "/srv/forge-standby");
        assert_eq!(
            t.repo_url("/srv/forge-standby/repos/X.git"),
            "ssh://sb@standby.example:2222/srv/forge-standby/repos/X.git"
        );
        let bare = Target::parse("standby:/srv/x", 22).expect("parse");
        assert_eq!(bare.destination(), "standby");
    }

    #[test]
    fn target_rejects_bad_specs() {
        for bad in [
            "no-colon",
            "sb@host",
            "sb@host:relative",
            "sb@host:/srv/../x",
            "-oProxyCommand=x@host:/srv",
            "@host:/srv",
            "sb@:/srv",
            "sb@-x:/srv",
            "sb x@host:/srv",
            "sb@ho st:/srv",
            "sb@host:/srv/a b",
        ] {
            assert!(Target::parse(bad, 22).is_err(), "{bad:?} should be refused");
        }
    }

    #[test]
    fn layout_paths_follow_the_contract() {
        let layout = Layout::new("/srv/sb").expect("layout");
        assert_eq!(
            layout.repo_dir("01ARZ3NDEKTSV4RRFFQ69G5FAV").expect("id"),
            "/srv/sb/repos/01ARZ3NDEKTSV4RRFFQ69G5FAV.git"
        );
        assert!(layout.repo_dir("../x").is_err());
        assert_eq!(
            layout.snapshot_file("20260101T000000Z"),
            "/srv/sb/db/forge-20260101T000000Z.db"
        );
        assert_eq!(layout.latest_link(), "/srv/sb/db/latest");
        assert!(Layout::new("relative").is_err());
    }

    #[test]
    fn snapshot_names_are_recognised_strictly() {
        assert!(is_snapshot_name("forge-20260101T120000Z.db"));
        assert!(!is_snapshot_name("forge-20260101T120000Z.db.tmp"));
        assert!(!is_snapshot_name("latest"));
        assert!(!is_snapshot_name(".tmp-forge-20260101T120000Z.db"));
        assert!(!is_snapshot_name("forge-2026-01-01T12:00:00Z.db"));
        assert!(!is_snapshot_name("forge-20260101X120000Z.db"));
    }

    #[test]
    fn snapshot_stamp_has_the_expected_shape() {
        let stamp = snapshot_stamp();
        assert!(is_snapshot_name(&format!("forge-{stamp}.db")), "{stamp}");
    }

    #[test]
    fn ssh_argv_pins_the_host_key_and_identity() {
        let target = Target::parse("sb@standby.example:/srv/sb", 2200).expect("target");
        let transport = SshTransport::new(
            target,
            PathBuf::from("/etc/forge/standby_key"),
            PathBuf::from("/etc/forge/standby_known_hosts"),
        );
        let argv = transport.ssh_argv("true");
        let expected = [
            "-i",
            "/etc/forge/standby_key",
            "-o",
            "IdentitiesOnly=yes",
            "-o",
            "UserKnownHostsFile=/etc/forge/standby_known_hosts",
            "-o",
            "StrictHostKeyChecking=yes",
            "-o",
            "BatchMode=yes",
            "-o",
            "ConnectTimeout=20",
            "-p",
            "2200",
            "sb@standby.example",
            "true",
        ];
        assert_eq!(argv, expected);
        assert!(!argv.iter().any(|a| a.contains("StrictHostKeyChecking=no")));
        assert_eq!(
            transport.git_ssh_command(),
            "'ssh' '-i' '/etc/forge/standby_key' '-o' 'IdentitiesOnly=yes' \
             '-o' 'UserKnownHostsFile=/etc/forge/standby_known_hosts' \
             '-o' 'StrictHostKeyChecking=yes' '-o' 'BatchMode=yes' \
             '-o' 'ConnectTimeout=20' '-p' '2200'"
        );
    }

    #[test]
    fn upload_command_writes_a_temp_then_renames() {
        assert_eq!(
            upload_command("/srv/sb/db/forge-20260101T000000Z.db"),
            "cat > '/srv/sb/db/.tmp-forge-20260101T000000Z.db' && mv -f \
             '/srv/sb/db/.tmp-forge-20260101T000000Z.db' '/srv/sb/db/forge-20260101T000000Z.db'"
        );
    }

    #[test]
    fn queue_debounces_and_drains_due_repos() {
        let queue = ReplicationQueue::default();
        queue.enqueue("b");
        queue.enqueue("a");
        assert!(queue.take_due(Instant::now()).is_empty(), "not due yet");
        let later = Instant::now() + PUSH_DEBOUNCE + Duration::from_millis(10);
        assert_eq!(queue.take_due(later), ["a", "b"]);
        assert!(queue.take_due(later).is_empty(), "drained");
        assert!(!queue.take_reconcile_request());
        queue.request_reconcile();
        assert!(queue.take_reconcile_request());
        assert!(!queue.take_reconcile_request(), "one request, one pass");
    }
}