//! End-to-end: a real server on a loopback port, clients over HTTP.

use oxy_core::repo::{Repo, SignatureStatus};
use std::fs;
use std::net::TcpListener;
use std::path::Path;

fn write(root: &Path, rel: &str, content: &[u8]) {
    let p = root.join(rel);
    fs::create_dir_all(p.parent().unwrap()).unwrap();
    fs::write(p, content).unwrap();
}

/// Point `$OXY_HOME` at a throwaway directory once per test process so no
/// test reads or writes the developer's real user configuration or keys.
fn hermetic() {
    static ONCE: std::sync::Once = std::sync::Once::new();
    static DIR: std::sync::OnceLock<tempfile::TempDir> = std::sync::OnceLock::new();
    ONCE.call_once(|| {
        let d = tempfile::tempdir().unwrap();
        std::env::set_var("OXY_HOME", d.path());
        DIR.set(d).ok();
    });
}

fn start(root: &Path, tokens: Vec<String>) -> String {
    start_with(root, tokens, None)
}

fn start_with(root: &Path, tokens: Vec<String>, tls: Option<oxy_server::TlsFiles>) -> String {
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    let port = listener.local_addr().unwrap().port();
    let scheme = if tls.is_some() { "https" } else { "http" };
    let cfg = oxy_server::ServerConfig {
        root: root.to_path_buf(),
        tokens,
        tls,
    };
    std::thread::spawn(move || oxy_server::run_blocking(listener, cfg).unwrap());
    // Give the acceptor a moment; the first request otherwise races the bind.
    std::thread::sleep(std::time::Duration::from_millis(50));
    format!("{scheme}://127.0.0.1:{port}")
}

/// A self-signed certificate for 127.0.0.1 written to `dir`: (cert, key).
fn self_signed(dir: &Path) -> (std::path::PathBuf, std::path::PathBuf) {
    fs::create_dir_all(dir).unwrap();
    let ck =
        rcgen::generate_simple_self_signed(vec!["127.0.0.1".into(), "localhost".into()]).unwrap();
    let cert = dir.join("cert.pem");
    let key = dir.join("key.pem");
    fs::write(&cert, ck.cert.pem()).unwrap();
    fs::write(&key, ck.signing_key.serialize_pem()).unwrap();
    (cert, key)
}

#[test]
fn clone_push_pull_over_http() {
    hermetic();
    let dir = tempfile::tempdir().unwrap();
    let root = dir.path();
    let served = root.join("served");
    Repo::init(&served.join("origin")).unwrap();
    let base = start(&served, vec![]);
    let url = format!("{base}/origin");

    let mut a = Repo::clone(&url, &root.join("a"))
        .unwrap()
        .with_home(root.join("home"));
    let (key, _) = a
        .create_identity(
            "Ada",
            "ada@example.org",
            oxy_core::repo::IdentityScope::Repo,
        )
        .unwrap();
    write(&root.join("a"), "f.txt", b"hello\n");
    let h1 = a.commit("hello", &[]).unwrap();
    let r = a.push("origin", false).unwrap();
    assert!(r[0].ok, "{}", r[0].reason);

    let origin = Repo::open(&served.join("origin")).unwrap();
    assert_eq!(origin.bookmark_heads().unwrap(), vec![h1]);
    assert!(
        matches!(origin.authenticate_change(&h1).unwrap(), SignatureStatus::Valid { key: k, sealed: true, .. } if k == key)
    );

    let b = Repo::clone(&url, &root.join("b")).unwrap();
    assert_eq!(fs::read(root.join("b/f.txt")).unwrap(), b"hello\n");
    write(&root.join("b"), "g.txt", b"g\n");
    let h2 = b.commit("g", &[]).unwrap();
    assert!(b.push("origin", false).unwrap()[0].ok);

    let pr = a.pull("origin").unwrap();
    assert!(pr.changed && pr.checked_out);
    assert_eq!(fs::read(root.join("a/g.txt")).unwrap(), b"g\n");
    let mut heads = a.bookmark_heads().unwrap();
    heads.sort();
    let mut want = vec![h1, h2];
    want.sort();
    assert_eq!(heads, want);

    // Unknown repositories and names the server refuses are surfaced as errors.
    let err = Repo::clone(&format!("{base}/nope"), &root.join("c"))
        .err()
        .unwrap()
        .to_string();
    assert!(err.contains("404"), "{err}");
    fs::create_dir_all(served.join(".hidden/.oxy")).unwrap();
    assert!(Repo::clone(&format!("{base}/.hidden"), &root.join("d"))
        .err()
        .unwrap()
        .to_string()
        .contains("404"));
}

#[test]
fn objects_endpoint_serves_lazy_blobs() {
    hermetic();
    let dir = tempfile::tempdir().unwrap();
    let root = dir.path();
    let served = root.join("served");
    let origin = Repo::init(&served.join("origin")).unwrap();
    let big: Vec<u8> = (0..300_000u32).map(|i| (i % 253) as u8).collect(); // has NUL bytes: opaque
    write(&served.join("origin"), "big.bin", &big);
    write(&served.join("origin"), "t.txt", b"t\n");
    origin.commit("big", &[]).unwrap();
    let base = start(&served, vec![]);
    let url = format!("{base}/origin");
    let a = Repo::clone_with(&url, &root.join("a"), Some(1000), &[]).unwrap();
    assert_eq!(
        fs::read(root.join("a/big.bin")).unwrap(),
        big,
        "fetched through the objects endpoint during checkout"
    );
    assert!(!a.snapshot("status").unwrap().changed);
}

#[test]
fn push_requires_a_token_when_configured() {
    hermetic();
    let dir = tempfile::tempdir().unwrap();
    let root = dir.path();
    let served = root.join("served");
    Repo::init(&served.join("origin")).unwrap();
    let base = start(&served, vec!["s3cret".into()]);
    let url = format!("{base}/origin");

    // Reads are open.
    let mut a = Repo::clone(&url, &root.join("a")).unwrap();
    write(&root.join("a"), "f.txt", b"x\n");
    a.commit("x", &[]).unwrap();
    let err = a.push("origin", false).err().unwrap().to_string();
    assert!(err.contains("401"), "{err}");

    a.config.set("remote.origin.token", "wrong").unwrap();
    a.save_config().unwrap();
    assert!(a
        .push("origin", false)
        .err()
        .unwrap()
        .to_string()
        .contains("401"));

    a.config.set("remote.origin.token", "s3cret").unwrap();
    a.save_config().unwrap();
    assert!(a.push("origin", false).unwrap()[0].ok);
    let origin = Repo::open(&served.join("origin")).unwrap();
    assert_eq!(origin.bookmark_heads().unwrap().len(), 1);
}

#[test]
fn https_with_a_self_signed_certificate() {
    hermetic();
    let dir = tempfile::tempdir().unwrap();
    let root = dir.path();
    let served = root.join("served");
    Repo::init(&served.join("origin")).unwrap();
    let (cert, key) = self_signed(root);
    let base = start_with(
        &served,
        vec![],
        Some(oxy_server::TlsFiles {
            cert: cert.clone(),
            key,
        }),
    );
    let url = format!("{base}/origin");

    // Without trusting the certificate the clone fails; with `OXY_CA` it works.
    let err = Repo::clone(&url, &root.join("untrusted"))
        .err()
        .unwrap()
        .to_string();
    assert!(
        err.contains("certificate")
            || err.contains("tls")
            || err.contains("TLS")
            || err.contains("handshake"),
        "{err}"
    );
    std::env::set_var("OXY_CA", &cert);
    let mut a = Repo::clone(&url, &root.join("a")).unwrap();
    std::env::remove_var("OXY_CA");
    // Persist the trust for this remote so later commands need no env var.
    a.config
        .set("remote.origin.ca", cert.to_str().unwrap())
        .unwrap();
    a.save_config().unwrap();
    write(&root.join("a"), "f.txt", b"tls\n");
    let h = a.commit("over tls", &[]).unwrap();
    assert!(a.push("origin", false).unwrap()[0].ok);
    let origin = Repo::open(&served.join("origin")).unwrap();
    assert_eq!(origin.bookmark_heads().unwrap(), vec![h]);
    // A wrong CA is refused.
    let (other_cert, _) = self_signed(&root.join("other"));
    a.config
        .set("remote.origin.ca", other_cert.to_str().unwrap())
        .unwrap();
    a.save_config().unwrap();
    assert!(a.fetch("origin").is_err());
}

/// Eight clones push independent changes at the same moment. The server
/// serialises pushes per repository, so every push either lands or is
/// refused because the bookmark moved, and a pull-then-push retry lands it.
#[test]
fn concurrent_pushes_are_serialised() {
    hermetic();
    let dir = tempfile::tempdir().unwrap();
    let root = dir.path();
    let served = root.join("served");
    Repo::init(&served.join("origin")).unwrap();
    let base = start(&served, vec![]);
    let url = format!("{base}/origin");
    const N: usize = 8;
    let mut clones = Vec::new();
    for i in 0..N {
        let path = root.join(format!("c{i}"));
        Repo::clone(&url, &path).unwrap();
        clones.push(path);
    }
    let barrier = std::sync::Arc::new(std::sync::Barrier::new(N));
    let mut handles = Vec::new();
    for (i, path) in clones.into_iter().enumerate() {
        let barrier = barrier.clone();
        handles.push(std::thread::spawn(move || {
            let repo = Repo::open(&path).unwrap();
            write(
                &path,
                &format!("file-{i}.txt"),
                format!("from clone {i}\n").as_bytes(),
            );
            let h = repo.commit(&format!("change {i}"), &[]).unwrap();
            barrier.wait();
            // A push is refused either up front (the remote moved since the
            // clone's last fetch) or by the server's compare-and-swap; both
            // are answered by pulling and pushing again.
            for attempt in 0..(N * 4) {
                match repo.push("origin", false) {
                    Ok(r) if r.iter().all(|x| x.ok) => return (h, attempt),
                    Ok(_) => {}
                    Err(e)
                        if e.to_string().contains("pull first")
                            || e.to_string().contains("moved") => {}
                    Err(e) => panic!("clone {i}: {e}"),
                }
                repo.pull("origin")
                    .unwrap_or_else(|e| panic!("clone {i} pull: {e}"));
            }
            panic!("clone {i} never managed to push");
        }));
    }
    let mut hashes = Vec::new();
    let mut total_rejected = 0;
    for h in handles {
        let (hash, attempts) = h.join().unwrap();
        hashes.push(hash);
        total_rejected += attempts;
    }
    let origin = Repo::open(&served.join("origin")).unwrap();
    let mut heads = origin.bookmark_heads().unwrap();
    heads.sort();
    hashes.sort();
    assert_eq!(heads, hashes, "every independent change is a head");
    let snap = origin.load_state(&heads).unwrap().materialize();
    for i in 0..N {
        assert!(snap.files.contains_key(&format!("file-{i}.txt")));
    }
    assert!(!snap.is_conflicted());
    eprintln!("{total_rejected} pushes were refused for a moved bookmark and retried");
    // The operation log is a single chain: each accepted push builds on the last.
    let ops = origin.op_log().unwrap();
    assert!(ops.len() > N);
}

#[test]
fn a_bare_repository_is_served_and_browsed_without_a_working_copy() {
    hermetic();
    let dir = tempfile::tempdir().unwrap();
    let root = dir.path();
    let served = root.join("served");
    let hosted = Repo::init_bare(&served.join("lib")).unwrap();
    assert!(hosted.is_bare());
    assert!(
        hosted.working_copy().is_err(),
        "a bare repository has no working copy"
    );
    assert!(!served.join("lib").join(".oxy").join("workspace").exists());
    let base = start(&served, vec![]);
    let url = format!("{base}/lib");

    // A client clones the empty repository, commits twice, and pushes.
    let work = root.join("work");
    let client = Repo::clone(&url, &work).unwrap();
    write(&work, "src/lib.rs", b"pub fn one() {}\n");
    client.snapshot("test").unwrap();
    let c1 = client.commit("one", &[]).unwrap();
    write(&work, "src/lib.rs", b"pub fn one() {}\npub fn two() {}\n");
    client.snapshot("test").unwrap();
    let c2 = client.commit("two", &[]).unwrap();
    let results = client.push("origin", false).unwrap();
    assert!(results.iter().all(|r| r.ok), "{results:?}");

    // The host reads history, files and blame from the bookmark's state alone.
    let hosted = Repo::open(&served.join("lib")).unwrap();
    assert!(hosted.is_bare());
    let heads = hosted.bookmark_heads().unwrap();
    assert_eq!(heads, vec![c2]);
    let log = hosted.log().unwrap();
    assert_eq!(log.iter().map(|e| e.hash).collect::<Vec<_>>(), vec![c2, c1]);
    assert!(log.iter().all(|e| !e.is_working_copy));
    assert_eq!(hosted.log_state(&[c1]).unwrap().len(), 1);
    assert_eq!(
        hosted.file_at("src/lib.rs", Some(&heads)).unwrap(),
        b"pub fn one() {}\npub fn two() {}\n"
    );
    let blame = hosted.blame_in("src/lib.rs", &heads).unwrap();
    assert_eq!(
        blame.iter().map(|l| l.hash).collect::<Vec<_>>(),
        vec![c1, c2]
    );
    assert_eq!(hosted.log_for_path("src/lib.rs").unwrap().len(), 2);
    assert_eq!(hosted.diff_files(Some(&c2)).unwrap().len(), 1);
    assert!(matches!(
        hosted.verify_change(&c2).unwrap(),
        SignatureStatus::Unsigned | SignatureStatus::Valid { .. }
    ));

    // Snapshotting a bare repository is refused rather than inventing a working copy.
    assert!(hosted.snapshot("test").is_err());
}

/// A host embeds the protocol with its own backend: names with an owner
/// segment, reads gated per repository, and a push hook.
#[test]
fn a_custom_backend_embeds_the_protocol() {
    use axum::extract::{Path as AxPath, State};
    use axum::http::{HeaderMap, Method, StatusCode};
    use axum::response::{IntoResponse, Response};
    use oxy_server::{Access, Backend, Endpoint, Failure, Server};
    use std::sync::{Arc, Mutex};

    struct Hub {
        root: std::path::PathBuf,
        pushes: Arc<Mutex<Vec<String>>>,
    }
    impl Backend for Hub {
        fn open(&self, name: &str) -> Result<Repo, Failure> {
            let Some((owner, repo)) = name.split_once('/') else {
                return Err(Failure(StatusCode::NOT_FOUND, "no such repository".into()));
            };
            let dir = self.root.join(owner).join(repo);
            if !dir.join(".oxy").is_dir() {
                return Err(Failure(StatusCode::NOT_FOUND, "no such repository".into()));
            }
            Repo::open(&dir).map_err(|e| Failure(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))
        }
        fn authorize(
            &self,
            name: &str,
            access: Access,
            headers: &HeaderMap,
        ) -> Result<(), Failure> {
            // `secret/*` is private: unknown to anyone without the token, even for reads.
            let token = oxy_server::bearer_token(headers);
            if name.starts_with("secret/") && token != Some("s3cret") {
                return Err(Failure(StatusCode::NOT_FOUND, "no such repository".into()));
            }
            if access == Access::Push && token.is_none() {
                return Err(Failure(StatusCode::UNAUTHORIZED, "sign in to push".into()));
            }
            Ok(())
        }
        fn after_push(
            &self,
            name: &str,
            report: &oxy_core::repo::PushReport,
            _headers: &HeaderMap,
        ) {
            self.pushes.lock().unwrap().push(format!(
                "{name}: {} ok, {} updated to {} head(s), {} imported",
                report.results.iter().filter(|r| r.ok).count(),
                report.updated.len(),
                report.updated.first().map(|u| u.heads.len()).unwrap_or(0),
                report.imported.len()
            ));
        }
    }
    async fn route(
        State(server): State<Arc<Server>>,
        AxPath((owner, repo, endpoint)): AxPath<(String, String, String)>,
        method: Method,
        headers: HeaderMap,
        body: axum::body::Bytes,
    ) -> Response {
        match Endpoint::parse(&endpoint) {
            Some(e) => {
                server
                    .handle(format!("{owner}/{repo}"), e, &method, &headers, body)
                    .await
            }
            None => Failure(StatusCode::NOT_FOUND, "no such endpoint".into()).into_response(),
        }
    }

    hermetic();
    let dir = tempfile::tempdir().unwrap();
    let root = dir.path();
    Repo::init_bare(&root.join("repos/andy/app")).unwrap();
    Repo::init_bare(&root.join("repos/secret/plans")).unwrap();
    let pushes = Arc::new(Mutex::new(Vec::new()));
    let server = Server::new(Hub {
        root: root.join("repos"),
        pushes: pushes.clone(),
    });
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    let port = listener.local_addr().unwrap().port();
    std::thread::spawn(move || {
        let rt = tokio::runtime::Builder::new_multi_thread()
            .enable_all()
            .build()
            .unwrap();
        rt.block_on(async move {
            listener.set_nonblocking(true).unwrap();
            let app = axum::Router::new()
                .route(
                    "/{owner}/{repo}/oxy/v1/{endpoint}",
                    axum::routing::any(route),
                )
                .with_state(server);
            axum::serve(tokio::net::TcpListener::from_std(listener).unwrap(), app)
                .await
                .unwrap();
        });
    });
    std::thread::sleep(std::time::Duration::from_millis(50));
    let base = format!("http://127.0.0.1:{port}");

    // Anonymous clone of the public repository works; the private one does not exist for outsiders.
    let work = root.join("work");
    let client = Repo::clone(&format!("{base}/andy/app"), &work).unwrap();
    assert!(Repo::clone(&format!("{base}/secret/plans"), &root.join("nope")).is_err());

    // A push needs a token; with one it lands and the hook sees it.
    write(&work, "a.txt", b"a\n");
    client.snapshot("test").unwrap();
    client.commit("a", &[]).unwrap();
    assert!(
        client.push("origin", false).is_err(),
        "anonymous push must be refused"
    );
    let mut client = client;
    client.config.set("remote.origin.token", "s3cret").unwrap();
    let results = client.push("origin", false).unwrap();
    assert!(results.iter().all(|r| r.ok), "{results:?}");
    assert_eq!(
        pushes.lock().unwrap().as_slice(),
        ["andy/app: 1 ok, 1 updated to 1 head(s), 1 imported"]
    );

    let hosted = Repo::open(&root.join("repos/andy/app")).unwrap();
    assert_eq!(hosted.bookmark_heads().unwrap().len(), 1);
}

/// A hosted repository created unbound adopts the id of the first
/// repository pushed to it; before that it cannot be cloned or fetched.
#[test]
fn an_unbound_repository_adopts_the_first_pushers_id() {
    hermetic();
    let dir = tempfile::tempdir().unwrap();
    let root = dir.path();
    let served = root.join("served");
    let hosted = Repo::init_bare_unbound(&served.join("lib")).unwrap();
    assert!(hosted.is_unbound());
    assert_eq!(hosted.id().unwrap(), [0; 16]);
    let base = start(&served, vec![]);
    let url = format!("{base}/lib");
    assert!(Repo::clone(&url, &root.join("early"))
        .err()
        .unwrap()
        .to_string()
        .contains("unbound"));

    // An existing repository with its own id and history.
    let work = root.join("work");
    let mut local = Repo::init(&work).unwrap();
    let my_id = local.id().unwrap();
    write(&work, "a.txt", b"a\n");
    local.snapshot("test").unwrap();
    local.commit("a", &[]).unwrap();
    local.config.set("remote.origin.url", &url).unwrap();
    local.save_config().unwrap();
    assert!(
        local
            .fetch("origin")
            .err()
            .unwrap()
            .to_string()
            .contains("unbound"),
        "fetching an unbound remote is refused"
    );
    let results = local.push("origin", false).unwrap();
    assert!(results.iter().all(|r| r.ok), "{results:?}");

    let hosted = Repo::open(&served.join("lib")).unwrap();
    assert!(!hosted.is_unbound());
    assert_eq!(hosted.id().unwrap(), my_id, "bound to the pusher's id");
    assert_eq!(hosted.bookmark_heads().unwrap().len(), 1);
    // From now on it is an ordinary remote: clones share the id, other repositories are refused.
    let cloned = Repo::clone(&url, &root.join("late")).unwrap();
    assert_eq!(cloned.id().unwrap(), my_id);
    let mut other = Repo::init(&root.join("other")).unwrap();
    other.config.set("remote.origin.url", &url).unwrap();
    other.save_config().unwrap();
    write(&root.join("other"), "b.txt", b"b\n");
    other.snapshot("test").unwrap();
    other.commit("b", &[]).unwrap();
    assert!(other
        .push("origin", false)
        .err()
        .unwrap()
        .to_string()
        .contains("different repository"));
}

/// History made before an identity existed (zero author) is adopted by
/// the pusher's seal, so it passes a server that enforces a policy.
#[test]
fn anonymous_history_is_adopted_by_the_pushers_seal() {
    hermetic();
    let dir = tempfile::tempdir().unwrap();
    let root = dir.path();
    let served = root.join("served");
    Repo::init_bare_unbound(&served.join("app")).unwrap();
    let base = start(&served, vec![]);
    let url = format!("{base}/app");

    // Two anonymous changes, then an identity and a policy naming it.
    let work = root.join("work");
    let mut local = Repo::init(&work).unwrap();
    write(&work, "a.txt", b"a\n");
    local.snapshot("test").unwrap();
    let anon = local.commit("before identity", &[]).unwrap();
    local
        .create_identity(
            "Andy",
            "andy@example.com",
            oxy_core::repo::IdentityScope::Repo,
        )
        .unwrap();
    local.policy_init().unwrap();
    write(&work, "b.txt", b"b\n");
    local.snapshot("test").unwrap();
    let mine = local.commit("after identity", &[]).unwrap();
    assert!(matches!(
        local.authenticate_change(&anon).unwrap(),
        SignatureStatus::Unsigned
    ));
    local.config.set("remote.origin.url", &url).unwrap();
    local.save_config().unwrap();
    let results = local.push("origin", false).unwrap();
    assert!(results.iter().all(|r| r.ok), "{results:?}");

    // The server holds the policy and both changes verify: sealed by the adopter.
    let hosted = Repo::open(&served.join("app")).unwrap();
    assert!(hosted.current_policy().unwrap().is_some());
    for h in [anon, mine] {
        assert!(
            matches!(hosted.authenticate_change(&h).unwrap(), SignatureStatus::Valid { sealed: true, name, .. } if name == "Andy"),
            "{h}"
        );
    }
    // A change that names another author still cannot be sealed by someone else.
    let stranger = root.join("stranger");
    let mut other = Repo::init(&stranger).unwrap();
    other
        .create_identity(
            "Mallory",
            "m@example.com",
            oxy_core::repo::IdentityScope::Repo,
        )
        .unwrap();
    write(&stranger, "c.txt", b"c\n");
    other.snapshot("test").unwrap();
    let theirs = other.commit("mallory's", &[]).unwrap();
    assert!(matches!(
        other.authenticate_change(&theirs).unwrap(),
        SignatureStatus::Unsigned
    ));
    let bundle = other.bundle_for(&[theirs]).unwrap();
    local.import_bundle(&bundle, false).unwrap();
    local.seal(&[theirs]).unwrap();
    assert!(
        matches!(local.authenticate_change(&theirs).unwrap(), SignatureStatus::Invalid(why) if why.contains("neither the author nor a delegate"))
    );
}

/// `oxy push --delete` over HTTP: a bare repository's bookmark goes, and
/// its default (the served repository's own bookmark) is refused by the
/// backend (spec §16.4).
#[test]
fn bookmarks_are_deleted_over_http_but_never_the_default() {
    hermetic();
    let dir = tempfile::tempdir().unwrap();
    let root = dir.path();
    let served = root.join("served");
    let hosted = Repo::init_bare(&served.join("lib")).unwrap();
    let base = start(&served, vec![]);
    let work = root.join("work");
    let client = Repo::clone(&format!("{base}/lib"), &work).unwrap();
    write(&work, "a.txt", b"a\n");
    client.commit("a", &[]).unwrap();
    assert!(client.push("origin", false).unwrap()[0].ok);
    client.bookmark_create("feature/x", None).unwrap();
    assert!(client.push_bookmark("origin", "feature/x", false).unwrap()[0].ok);
    let r = client.push_delete("origin", "feature/x").unwrap();
    assert!(r[0].ok, "{r:?}");
    assert!(!hosted.view().unwrap().bookmarks.contains_key("feature/x"));
    let r = client.push_delete("origin", "main").unwrap();
    assert!(
        !r[0].ok && r[0].reason.contains("default bookmark"),
        "{r:?}"
    );
    assert!(hosted.view().unwrap().bookmarks.contains_key("main"));
}