clawsync/crates/clawsync-cli/tests/serve_all.rs

//! Integration tests for `clawsync serve-all` — the universal multi-protocol server.
//!
//! Each test starts a single `serve-all` server and connects to it with
//! different client types (onion push/pull, HDF5-sync, fs-sync) verifying that
//! the correct protocol is dispatched based on the first message.
//!
//! Run with:
//!   cargo test -p clawsync-cli --test serve_all -- --nocapture

use std::fs;
use std::io::{BufRead, BufReader};
use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Stdio};
use std::time::Duration;

use clawhdf5::FileBuilder;
use clawhdf5_onion::writer::OnionFile;
use clawsync_hdf5::DatasetManifest;
use tempfile::TempDir;

const BIN: &str = env!("CARGO_BIN_EXE_clawsync");
const H5_MAGIC: &[u8] = b"\x89HDF\r\n\x1a\n";

// ─────────────────────────────────────────────────────────────────────────────
// serve-all harness
// ─────────────────────────────────────────────────────────────────────────────

struct ServeAllServer {
    child: Child,
    pub addr: SocketAddr,
    pub dir: TempDir,
}

impl ServeAllServer {
    /// Start `clawsync serve-all <dir> --bind 127.0.0.1:0 [--allow-delete]`
    /// and wait for the startup line.
    fn start(allow_delete: bool) -> Self {
        let dir = TempDir::new().expect("failed to create temp dir");
        let mut cmd = Command::new(BIN);
        cmd.arg("serve-all")
            .arg(dir.path())
            .arg("--bind")
            .arg("127.0.0.1:0");
        if allow_delete {
            cmd.arg("--allow-delete");
        }
        cmd.stdout(Stdio::piped()).stderr(Stdio::inherit());

        let mut child = cmd.spawn().expect("failed to spawn clawsync serve-all");
        let stdout = child.stdout.take().unwrap();
        let mut reader = BufReader::new(stdout);

        // Wait for: "ClawSync server listening on 127.0.0.1:PORT"
        let deadline = std::time::Instant::now() + Duration::from_secs(10);
        let mut line = String::new();
        loop {
            line.clear();
            reader
                .read_line(&mut line)
                .expect("failed to read startup line");
            if line.contains("ClawSync server listening on") {
                break;
            }
            if std::time::Instant::now() > deadline {
                panic!("serve-all did not print startup line in time; got: {line:?}");
            }
        }

        let addr_str = line
            .trim()
            .rsplit("on ")
            .next()
            .expect("unexpected startup line")
            .trim();
        let addr: SocketAddr = addr_str
            .parse()
            .unwrap_or_else(|_| panic!("could not parse bound addr from: {addr_str}"));

        // Drain remaining stdout in a background thread.
        std::thread::spawn(move || for _ in reader.lines() {});

        Self { child, addr, dir }
    }
}

impl Drop for ServeAllServer {
    fn drop(&mut self) {
        let _ = self.child.kill();
        let _ = self.child.wait();
    }
}

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

/// Create a temp dir with an HDF5 file containing `n` onion revisions.
fn make_versioned_h5(n: u8) -> (TempDir, PathBuf, OnionFile) {
    let dir = TempDir::new().unwrap();
    let h5 = dir.path().join("data.h5");
    fs::write(&h5, H5_MAGIC).unwrap();
    let mut onion = OnionFile::create(&h5, 4096).unwrap();
    for i in 0..n {
        let mut s = onion.begin_session(None).unwrap();
        s.record_page(0, &vec![i; 4096]);
        onion.commit_session(s, Some(&format!("rev {i}"))).unwrap();
    }
    onion.flush().unwrap();
    (dir, h5, onion)
}

/// Build an HDF5 file with the given f64 datasets.
fn write_h5(path: &Path, datasets: &[(&str, Vec<f64>)]) {
    let mut b = FileBuilder::new();
    for (name, data) in datasets {
        b.create_dataset(name).with_f64_data(data);
    }
    let bytes = b.finish().unwrap();
    fs::write(path, bytes).unwrap();
}

/// Run `clawsync push <local> <addr/remote>` and assert exit 0.
fn run_push(local: &Path, addr: SocketAddr, remote_name: &str) {
    let remote = format!("{addr}/{remote_name}");
    let status = Command::new(BIN)
        .args(["push", local.to_str().unwrap(), &remote])
        .status()
        .expect("failed to run push");
    assert!(status.success(), "push exited with {status}");
}

/// Run `clawsync pull <addr/remote> <local>` and assert exit 0.
fn run_pull(addr: SocketAddr, remote_name: &str, local: &Path) {
    let remote = format!("{addr}/{remote_name}");
    let status = Command::new(BIN)
        .args(["pull", &remote, local.to_str().unwrap()])
        .status()
        .expect("failed to run pull");
    assert!(status.success(), "pull exited with {status}");
}

/// Run `clawsync hdf5-sync <local> <addr/rel>` and assert exit 0.
fn run_hdf5_sync(local: &Path, addr: SocketAddr, rel_path: &str) {
    let remote = format!("{addr}/{rel_path}");
    let status = Command::new(BIN)
        .args(["hdf5-sync", local.to_str().unwrap(), &remote])
        .status()
        .expect("failed to run hdf5-sync");
    assert!(status.success(), "hdf5-sync exited with {status}");
}

/// Run `clawsync sync <src> <addr/>` and assert exit 0.
fn run_fs_sync(src: &Path, addr: SocketAddr, delete: bool) {
    let remote = format!("{addr}/");
    let mut cmd = Command::new(BIN);
    cmd.args(["sync", src.to_str().unwrap(), &remote]);
    if delete {
        cmd.arg("--delete");
    }
    let status = cmd.status().expect("failed to run sync");
    assert!(status.success(), "sync exited with {status}");
}

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

/// Push onion revisions to a serve-all server.
///
/// The server should detect IbltRequest as the onion protocol and serve the
/// file at `<root>/<agent-id>.claws`.
#[test]
fn serve_all_routes_onion_push() {
    let server = ServeAllServer::start(false);
    let (_src_dir, h5, _onion) = make_versioned_h5(3);

    run_push(&h5, server.addr, "onion-data");

    // Verify the serve-all root has the served file.
    let entries: Vec<_> = fs::read_dir(server.dir.path())
        .unwrap()
        .filter_map(|e| e.ok())
        .collect();
    assert!(
        !entries.is_empty(),
        "serve-all root should contain the served file"
    );
}

/// Pull revisions via a serve-all server.
#[test]
fn serve_all_routes_onion_pull() {
    let server = ServeAllServer::start(false);
    let (_src_dir, src_h5, _onion) = make_versioned_h5(4);

    // Push first so the server has revisions.
    run_push(&src_h5, server.addr, "onion-data");

    // Pull to a fresh local file.
    let dst_dir = TempDir::new().unwrap();
    let dst_h5 = dst_dir.path().join("data.h5");
    run_pull(server.addr, "onion-data", &dst_h5);

    // Destination should now have the same number of revisions.
    let dst_onion = OnionFile::open(&dst_h5).expect("failed to open pulled file");
    assert_eq!(
        dst_onion.revision_count(),
        4,
        "pulled file should have 4 revisions"
    );
}

/// HDF5 dataset-granular sync through a serve-all server.
#[test]
fn serve_all_routes_hdf5_sync() {
    let server = ServeAllServer::start(false);

    let src_dir = TempDir::new().unwrap();
    let src_h5 = src_dir.path().join("science.h5");
    write_h5(
        &src_h5,
        &[
            ("temperature", vec![1.0, 2.0, 3.0]),
            ("pressure", vec![101.0, 102.0]),
        ],
    );

    run_hdf5_sync(&src_h5, server.addr, "science.h5");

    // Verify the served file now has the same datasets.
    let served = server.dir.path().join("science.h5");
    assert!(served.exists(), "served h5 file must exist after hdf5-sync");

    let src_manifest = DatasetManifest::from_path(&src_h5).expect("failed to read src manifest");
    let srv_manifest = DatasetManifest::from_path(&served).expect("failed to read server manifest");
    assert_eq!(
        src_manifest.len(),
        srv_manifest.len(),
        "server must have same number of datasets as source"
    );
    for (path, entry) in &src_manifest.datasets {
        let srv = srv_manifest
            .get(path)
            .unwrap_or_else(|| panic!("dataset {path} missing on server"));
        assert_eq!(entry.blake3, srv.blake3, "dataset {path} hash mismatch");
    }
}

/// General file sync (any-type) through a serve-all server.
#[test]
fn serve_all_routes_fs_sync() {
    let server = ServeAllServer::start(false);

    let src_dir = TempDir::new().unwrap();
    fs::write(src_dir.path().join("hello.txt"), b"hello, world").unwrap();
    fs::write(
        src_dir.path().join("data.bin"),
        vec![0xDE, 0xAD, 0xBE, 0xEF],
    )
    .unwrap();

    run_fs_sync(src_dir.path(), server.addr, false);

    // Both files must exist in the serve-all root.
    let hello = server.dir.path().join("hello.txt");
    let data = server.dir.path().join("data.bin");
    assert!(hello.exists(), "hello.txt must be synced");
    assert!(data.exists(), "data.bin must be synced");
    assert_eq!(fs::read(&hello).unwrap(), b"hello, world");
    assert_eq!(fs::read(&data).unwrap(), [0xDE, 0xAD, 0xBE, 0xEF]);
}

/// Two sequential clients of different protocols connect to the same serve-all
/// port; both must complete successfully without the server crashing.
#[test]
fn serve_all_multi_protocol_sequential() {
    let server = ServeAllServer::start(false);

    // First: FS sync.
    let src_dir = TempDir::new().unwrap();
    fs::write(src_dir.path().join("file.txt"), b"content").unwrap();
    run_fs_sync(src_dir.path(), server.addr, false);

    // Second: HDF5 sync.
    let h5_dir = TempDir::new().unwrap();
    let h5 = h5_dir.path().join("metrics.h5");
    write_h5(&h5, &[("latency", vec![0.1, 0.2, 0.3])]);
    run_hdf5_sync(&h5, server.addr, "metrics.h5");

    // Both outputs must be present in the serve-all root.
    assert!(server.dir.path().join("file.txt").exists());
    assert!(server.dir.path().join("metrics.h5").exists());
}

/// A warm no-op FS sync (already in sync) through serve-all succeeds.
#[test]
fn serve_all_fs_warm_noop() {
    let server = ServeAllServer::start(false);

    let src_dir = TempDir::new().unwrap();
    fs::write(src_dir.path().join("stable.txt"), b"unchanged").unwrap();

    // First sync.
    run_fs_sync(src_dir.path(), server.addr, false);
    // Second sync — same content, should be a no-op.
    run_fs_sync(src_dir.path(), server.addr, false);

    assert_eq!(
        fs::read(server.dir.path().join("stable.txt")).unwrap(),
        b"unchanged"
    );
}

/// Push/pull round-trip through serve-all preserves revision data.
#[test]
fn serve_all_push_pull_round_trip() {
    let server = ServeAllServer::start(false);
    let (_src_dir, src_h5, src_onion) = make_versioned_h5(5);
    let src_rev_count = src_onion.revision_count();

    run_push(&src_h5, server.addr, "round-trip");

    let dst_dir = TempDir::new().unwrap();
    let dst_h5 = dst_dir.path().join("round_trip.h5");
    run_pull(server.addr, "round-trip", &dst_h5);

    let dst_onion = OnionFile::open(&dst_h5).unwrap();
    assert_eq!(
        dst_onion.revision_count(),
        src_rev_count,
        "round-tripped file must have same revision count"
    );
}

// ─────────────────────────────────────────────────────────────────────────────
// QUIC transport tests for serve-all
// ─────────────────────────────────────────────────────────────────────────────

/// Harness for `clawsync serve-all --quic`.
struct ServeAllServerQuic {
    child: Child,
    pub addr: SocketAddr,
    pub dir: TempDir,
}

impl ServeAllServerQuic {
    fn start(allow_delete: bool) -> Self {
        let dir = TempDir::new().expect("failed to create temp dir");
        let mut cmd = Command::new(BIN);
        cmd.arg("serve-all")
            .arg(dir.path())
            .arg("--bind")
            .arg("127.0.0.1:0")
            .arg("--quic");
        if allow_delete {
            cmd.arg("--allow-delete");
        }
        cmd.stdout(Stdio::piped()).stderr(Stdio::inherit());

        let mut child = cmd
            .spawn()
            .expect("failed to spawn clawsync serve-all --quic");
        let stdout = child.stdout.take().unwrap();
        let mut reader = BufReader::new(stdout);

        let deadline = std::time::Instant::now() + Duration::from_secs(10);
        let mut line = String::new();
        loop {
            line.clear();
            reader
                .read_line(&mut line)
                .expect("failed to read startup line");
            if line.contains("ClawSync server listening on") {
                break;
            }
            if std::time::Instant::now() > deadline {
                panic!("serve-all --quic did not print startup line in time; got: {line:?}");
            }
        }

        let addr_str = line.trim().rsplit("on ").next().unwrap().trim();
        let addr: SocketAddr = addr_str
            .parse()
            .unwrap_or_else(|_| panic!("could not parse bound addr from: {addr_str}"));

        std::thread::spawn(move || for _ in reader.lines() {});

        Self { child, addr, dir }
    }
}

impl Drop for ServeAllServerQuic {
    fn drop(&mut self) {
        let _ = self.child.kill();
        let _ = self.child.wait();
    }
}

fn run_fs_sync_quic(src: &Path, addr: SocketAddr) {
    let remote = format!("{addr}/");
    let status = Command::new(BIN)
        .args(["sync", src.to_str().unwrap(), &remote, "--quic"])
        .status()
        .expect("failed to run sync --quic");
    assert!(status.success(), "sync --quic exited with {status}");
}

/// Cold copy via QUIC through the universal serve-all server.
#[test]
fn serve_all_quic_fs_cold_copy() {
    let server = ServeAllServerQuic::start(false);

    let src = TempDir::new().unwrap();
    fs::write(src.path().join("quic_file.txt"), b"quic cold copy test").unwrap();
    fs::write(src.path().join("binary.bin"), vec![0x42u8; 1024]).unwrap();

    run_fs_sync_quic(src.path(), server.addr);

    let dst_txt = server.dir.path().join("quic_file.txt");
    let dst_bin = server.dir.path().join("binary.bin");
    assert!(dst_txt.exists(), "quic_file.txt must be synced via QUIC");
    assert!(dst_bin.exists(), "binary.bin must be synced via QUIC");
    assert_eq!(fs::read(&dst_txt).unwrap(), b"quic cold copy test");
    assert_eq!(fs::read(&dst_bin).unwrap(), vec![0x42u8; 1024]);
}

/// Warm no-op via QUIC: re-syncing identical content transfers nothing
/// and exits 0.
#[test]
fn serve_all_quic_fs_warm_noop() {
    let server = ServeAllServerQuic::start(false);

    // Seed server directory with the same content as the source.
    let src = TempDir::new().unwrap();
    let content = b"identical content for noop";
    fs::write(src.path().join("noop.txt"), content).unwrap();
    fs::write(server.dir.path().join("noop.txt"), content).unwrap();

    // Second sync should be a warm no-op.
    run_fs_sync_quic(src.path(), server.addr);

    assert_eq!(
        fs::read(server.dir.path().join("noop.txt")).unwrap(),
        content
    );
}

/// Incremental sync via QUIC: modifying one file transfers only that file.
#[test]
fn serve_all_quic_fs_incremental() {
    let server = ServeAllServerQuic::start(false);

    let src = TempDir::new().unwrap();

    // Put both files on server already.
    fs::write(src.path().join("unchanged.bin"), vec![0x11u8; 512]).unwrap();
    fs::write(src.path().join("changed.bin"), vec![0xAAu8; 512]).unwrap();
    run_fs_sync_quic(src.path(), server.addr);

    // Modify one file; re-sync.
    fs::write(src.path().join("changed.bin"), vec![0xBBu8; 512]).unwrap();
    run_fs_sync_quic(src.path(), server.addr);

    assert_eq!(
        fs::read(server.dir.path().join("changed.bin")).unwrap(),
        vec![0xBBu8; 512]
    );
    assert_eq!(
        fs::read(server.dir.path().join("unchanged.bin")).unwrap(),
        vec![0x11u8; 512]
    );
}

/// `serve-all` dispatch: `pull-fs` — the server pushes its files to the client.
///
/// This covers the `FsDirPullRequest` branch that was missing from
/// `handle_any_client` before the fix (it would return an error with
/// "unrecognized first message").
#[test]
fn serve_all_pull_fs() {
    let server = ServeAllServer::start(false);

    // Seed server root directly with 2 files.
    fs::write(server.dir.path().join("server_a.bin"), vec![0x11u8; 1024]).unwrap();
    fs::write(server.dir.path().join("server_b.bin"), vec![0x22u8; 2048]).unwrap();

    let dst = TempDir::new().unwrap();
    let remote = format!("{}/", server.addr);

    let out = Command::new(BIN)
        .args(["pull-fs", &remote, dst.path().to_str().unwrap()])
        .output()
        .expect("failed to run pull-fs");

    assert!(
        out.status.success(),
        "pull-fs via serve-all failed:\nstdout: {}\nstderr: {}",
        String::from_utf8_lossy(&out.stdout),
        String::from_utf8_lossy(&out.stderr),
    );

    assert!(
        dst.path().join("server_a.bin").exists(),
        "server_a.bin not pulled"
    );
    assert!(
        dst.path().join("server_b.bin").exists(),
        "server_b.bin not pulled"
    );
    assert_eq!(
        fs::read(dst.path().join("server_a.bin")).unwrap(),
        vec![0x11u8; 1024]
    );
    assert_eq!(
        fs::read(dst.path().join("server_b.bin")).unwrap(),
        vec![0x22u8; 2048]
    );
}