huskies/server/src/upgrade.rs

//! In-container binary self-update — fetch a new `huskies` binary, atomically
//! replace the on-disk executable, drain CRDT persistence, and re-exec.

use crate::slog;
use std::path::{Path, PathBuf};

// ── Binary fetch ─────────────────────────────────────────────────────────────

/// Download a binary from `source_url` and atomically replace `target_path`.
///
/// Writes to a sibling `.tmp` file first, then renames so the replacement is
/// atomic on the same filesystem.  Sets the execute bit before renaming so the
/// file is runnable the moment it appears at the target location.
pub async fn fetch_and_replace_binary(source_url: &str, target_path: &Path) -> Result<(), String> {
    slog!("[upgrade] Fetching binary from {source_url}");

    let resp = reqwest::get(source_url)
        .await
        .map_err(|e| format!("Failed to fetch binary from {source_url}: {e}"))?;

    if !resp.status().is_success() {
        return Err(format!(
            "Binary fetch returned HTTP {}: {source_url}",
            resp.status()
        ));
    }

    let bytes = resp
        .bytes()
        .await
        .map_err(|e| format!("Failed to read binary response body: {e}"))?;

    if bytes.is_empty() {
        return Err("Binary fetch returned an empty body".to_string());
    }

    // Write to a sibling temp file so the rename is atomic on the same FS.
    let tmp_path = sibling_tmp_path(target_path)?;
    std::fs::write(&tmp_path, &bytes)
        .map_err(|e| format!("Failed to write tmp binary to {}: {e}", tmp_path.display()))?;

    set_executable(&tmp_path)?;

    std::fs::rename(&tmp_path, target_path).map_err(|e| {
        format!(
            "Failed to rename {} → {}: {e}",
            tmp_path.display(),
            target_path.display()
        )
    })?;

    slog!(
        "[upgrade] Binary replaced at {} ({} bytes)",
        target_path.display(),
        bytes.len()
    );
    Ok(())
}

// ── Full server upgrade (called from the running process) ─────────────────

/// Fetch a new binary, atomically replace the current executable, drain CRDT
/// persistence, and exit so Docker restarts the container with the new binary.
///
/// The entrypoint script checks for a rebuilt binary at
/// /app/target/release/huskies and prefers it over the image-baked
/// /usr/local/bin/huskies.
///
/// On failure it returns `Err(message)` so the caller can report the error
/// while keeping the original server running.
pub async fn upgrade_and_reexec(source_url: &str, project_root: &Path) -> Result<String, String> {
    let target = resolve_target_path();

    fetch_and_replace_binary(source_url, &target).await?;

    crate::rebuild::drain_and_exit(project_root, "upgrade").await
}

// ── CLI upgrade (no re-exec) ─────────────────────────────────────────────

/// Run the `huskies upgrade` CLI subcommand: download, replace, and exit.
///
/// Unlike [`upgrade_and_reexec`], this does not flush the CRDT or re-exec
/// because the CLI subcommand is run as a standalone command (not the server).
/// After this returns the caller should exit.
pub async fn run_cli_upgrade(source_url: &str, target: &Path) -> Result<(), String> {
    fetch_and_replace_binary(source_url, target).await?;
    println!(
        "Upgrade complete. New binary installed at {}.",
        target.display()
    );
    Ok(())
}

// ── Helpers ───────────────────────────────────────────────────────────────

/// Resolve the path to replace: `current_exe()` if accessible, else
/// `/usr/local/bin/huskies`.
pub fn resolve_target_path() -> PathBuf {
    std::env::current_exe().unwrap_or_else(|_| PathBuf::from("/usr/local/bin/huskies"))
}

fn sibling_tmp_path(target: &Path) -> Result<PathBuf, String> {
    let parent = target
        .parent()
        .ok_or_else(|| format!("Cannot determine parent dir of {}", target.display()))?;
    Ok(parent.join(".huskies_upgrade.tmp"))
}

#[cfg(unix)]
fn set_executable(path: &Path) -> Result<(), String> {
    use std::os::unix::fs::PermissionsExt;
    let meta =
        std::fs::metadata(path).map_err(|e| format!("Cannot stat {}: {e}", path.display()))?;
    let mut perms = meta.permissions();
    perms.set_mode(0o755);
    std::fs::set_permissions(path, perms)
        .map_err(|e| format!("Cannot chmod {}: {e}", path.display()))
}

#[cfg(not(unix))]
fn set_executable(_path: &Path) -> Result<(), String> {
    Ok(())
}

// ── Tests ─────────────────────────────────────────────────────────────────

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

    /// Start a tiny HTTP server in the background that serves `content` at `/`.
    async fn serve_bytes(content: Vec<u8>) -> (u16, tokio::task::JoinHandle<()>) {
        use std::sync::Arc;

        let content = Arc::new(content);
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let port = listener.local_addr().unwrap().port();

        let handle = tokio::spawn(async move {
            loop {
                let Ok((mut stream, _)) = listener.accept().await else {
                    break;
                };
                let content = Arc::clone(&content);
                tokio::spawn(async move {
                    use tokio::io::AsyncWriteExt;
                    // Drain the HTTP request (ignore it).
                    let mut buf = [0u8; 4096];
                    let _ = tokio::io::AsyncReadExt::read(&mut stream, &mut buf).await;
                    // Write a minimal HTTP/1.1 200 response.
                    let header = format!(
                        "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nContent-Type: application/octet-stream\r\n\r\n",
                        content.len()
                    );
                    let _ = stream.write_all(header.as_bytes()).await;
                    let _ = stream.write_all(&content).await;
                });
            }
        });

        (port, handle)
    }

    #[tokio::test]
    async fn fetch_and_replace_binary_downloads_and_replaces() {
        let dir = tempfile::tempdir().unwrap();
        let target = dir.path().join("huskies");
        std::fs::write(&target, b"old binary").unwrap();

        let content = b"new binary content v0.99.0".to_vec();
        let (port, _srv) = serve_bytes(content.clone()).await;

        let url = format!("http://127.0.0.1:{port}/huskies");
        fetch_and_replace_binary(&url, &target).await.unwrap();

        let on_disk = std::fs::read(&target).unwrap();
        assert_eq!(
            on_disk, content,
            "target must contain the downloaded content"
        );
    }

    #[tokio::test]
    async fn fetch_and_replace_binary_sets_executable_bit() {
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;

            let dir = tempfile::tempdir().unwrap();
            let target = dir.path().join("huskies");
            std::fs::write(&target, b"old").unwrap();

            let (port, _srv) = serve_bytes(b"#!/bin/sh\nexit 0".to_vec()).await;
            let url = format!("http://127.0.0.1:{port}/huskies");
            fetch_and_replace_binary(&url, &target).await.unwrap();

            let mode = std::fs::metadata(&target).unwrap().permissions().mode();
            assert!(mode & 0o111 != 0, "binary must be executable after upgrade");
        }
    }

    #[tokio::test]
    async fn fetch_and_replace_binary_empty_body_is_error() {
        let dir = tempfile::tempdir().unwrap();
        let target = dir.path().join("huskies");
        std::fs::write(&target, b"old").unwrap();

        let (port, _srv) = serve_bytes(vec![]).await;
        let url = format!("http://127.0.0.1:{port}/huskies");
        let err = fetch_and_replace_binary(&url, &target).await.unwrap_err();
        assert!(
            err.contains("empty"),
            "expected empty-body error, got: {err}"
        );

        // Original must be untouched (rename never happened).
        let on_disk = std::fs::read(&target).unwrap();
        assert_eq!(on_disk, b"old");
    }

    #[tokio::test]
    async fn fetch_and_replace_binary_unreachable_url_is_error() {
        let dir = tempfile::tempdir().unwrap();
        let target = dir.path().join("huskies");
        let err = fetch_and_replace_binary("http://127.0.0.1:1/huskies", &target)
            .await
            .unwrap_err();
        assert!(!err.is_empty(), "expected a non-empty error");
    }

    #[tokio::test]
    async fn persisted_ops_count_does_not_decrease_after_flush() {
        // Initialise an in-process CRDT, flush it, and verify the persisted
        // count is stable (AC 5 — no ops lost across upgrade).
        crate::crdt_state::init_for_test();

        let before = crate::crdt_state::dump_crdt_state(None).persisted_ops_count;
        crate::crdt_state::flush_persistence(std::time::Duration::from_millis(200)).await;
        let after = crate::crdt_state::dump_crdt_state(None).persisted_ops_count;

        assert!(
            after >= before,
            "persisted_ops_count must not decrease after flush: before={before} after={after}"
        );
    }
}