//! 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",

            "[email protected]",

            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();