//! HTTP range reads against a local server (no internet): every value read
//! over HTTP equals `File::open`'s, and the server's misbehaviour — no
//! range support, a file replaced mid-read, cut-off bodies, errors, slow
//! answers under concurrent readers — is an error or the right data, never
//! wrong data.
//!
//! `CLAWHDF5_REMOTE_CORPUS=dir[:dir...]` adds every HDF5 file under those
//! directories (the conformance corpus is `conformance/.cache/corpus`), and
//! `CLAWHDF5_REMOTE_REPORT=1` prints the per-file request counts.

#![cfg(feature = "http")]

mod common;

use std::sync::Arc;
use std::sync::atomic::Ordering;
use std::time::Duration;

use clawhdf5::File;
use clawhdf5_remote::{
    CacheConfig, Error, HttpOptions, HttpStorage, Options, RemoteError, open_url, open_url_with,
    storage_for_url,
};
use common::server::Server;
use common::{list_and_read_one, multi_block_file, transcript};

/// Options for tests: fast retries.
fn quick() -> Options {
    Options {
        http: HttpOptions {
            backoff: Duration::from_millis(5),
            ..HttpOptions::default()
        },
        ..Options::default()
    }
}

fn changed(e: &str) -> bool {
    e.contains("changed while open")
}

/// Serve `files`, open each by URL and through `File::open`, and require
/// identical transcripts. Returns (files compared, files both refused).
fn compare(files: &[std::path::PathBuf], report: bool) -> (usize, usize) {
    let served: Vec<(String, Vec<u8>)> = files
        .iter()
        .enumerate()
        .filter_map(|(i, p)| {
            let bytes = std::fs::read(p).ok()?;
            let name = p.file_name()?.to_str()?.replace(' ', "_");
            Some((format!("/f{i}/{name}"), bytes))
        })
        .collect();
    let server = Server::start(served.clone());
    let (mut same, mut refused) = (0, 0);
    let mut totals = [0u64; 7];
    for (i, p) in files.iter().enumerate() {
        let Some((url_path, bytes)) = served
            .iter()
            .find(|(u, _)| u.starts_with(&format!("/f{i}/")))
        else {
            continue;
        };
        let url = server.url(url_path);
        let local = File::open(p);
        let remote = storage_for_url(&url, &quick())
            .map_err(|e| e.to_string())
            .and_then(|s| {
                File::open_storage(s.clone())
                    .map(|mut f| {
                        f.set_vds_resolver(common::sibling_resolver(p.parent().unwrap().into()));
                        (f, s)
                    })
                    .map_err(|e| e.to_string())
            });
        let (local, (remote, storage)) = match (local, remote) {
            (Ok(l), Ok(r)) => (l, r),
            (Err(l), Err(r)) => {
                assert!(
                    !r.contains("HTTP") && !r.contains("network"),
                    "{url}: {r} ({l})"
                );
                refused += 1;
                continue;
            }
            (l, r) => panic!(
                "{}: File::open {:?}, open_url {:?}",
                p.display(),
                l.map(|_| ()),
                r.map(|_| ())
            ),
        };
        let want = transcript(&local);
        let got = transcript(&remote);
        if want != got {
            let first = want
                .lines()
                .zip(got.lines())
                .find(|(w, g)| w != g)
                .map(|(w, g)| format!("\n local:  {w}\n remote: {g}"))
                .unwrap_or_default();
            panic!("{}: open_url differs from File::open{first}", p.display());
        }
        assert!(storage.stats().reads > 0, "read through the cache");
        same += 1;

        // Cost of "open + list" (a tree view) and of then reading the
        // largest dataset (a plot): with the block cache, and with none
        // (every read a request). Requests and bytes as the server saw
        // them, the open included.
        server.reset();
        let with = storage_for_url(&url, &quick()).unwrap();
        let f = File::open_storage(with.clone()).unwrap();
        let pick = common::list(&f);
        let (list_requests, list_bytes) = (server.requests(), server.bytes());
        if let Some((addr, _)) = pick {
            common::read_one(&f, addr);
        }
        let (cached_requests, cached_bytes) = (server.requests(), server.bytes());
        server.reset();
        let (bare, _) = HttpStorage::open(&url, quick().http).unwrap();
        if let Ok(f) = File::open_storage(Arc::new(bare)) {
            list_and_read_one(&f);
        }
        let uncached_requests = server.requests();
        for (t, v) in totals.iter_mut().zip([
            list_requests,
            list_bytes,
            cached_requests,
            cached_bytes,
            uncached_requests,
            bytes.len() as u64,
            1,
        ]) {
            *t += v;
        }
        if report {
            eprintln!(
                "open+list {list_requests:>4} req {list_bytes:>10} B | +read {:>5} req \
                 {cached_bytes:>10} B | uncached {uncached_requests:>7} req | file {:>10} B  {}",
                cached_requests,
                bytes.len(),
                p.display()
            );
        }
        // Files within one block: opening fetched everything.
        if bytes.len() as u64 <= with.config().block_size {
            assert_eq!(cached_requests, 1, "{}", p.display());
        }
        // The budget of docs/design/range-reads.md (Testing): listing the
        // IMERG file (file A of section 2) takes at most 3 requests.
        if p.ends_with("xarray-data/imerghh_730.hdf5") {
            assert!(list_requests <= 3, "{}: {list_requests}", p.display());
        }
    }
    eprintln!(
        "{} files ({} bytes): open + list {} requests, {} bytes; + read the largest \
         dataset {} requests, {} bytes (1 MiB block cache); without a cache {} requests",
        totals[6], totals[5], totals[0], totals[1], totals[2], totals[3], totals[4]
    );
    (same, refused)
}

#[test]
fn fixtures_read_identically_over_http() {
    let files = common::fixtures();
    assert!(files.len() >= 45, "{} fixtures", files.len());
    let report = std::env::var("CLAWHDF5_REMOTE_REPORT").is_ok_and(|v| v == "1");
    let (same, _) = compare(&files, report);
    assert!(same >= 40, "{same}");
}

#[test]
fn corpus_reads_identically_over_http() {
    let Some(files) = common::corpus() else {
        eprintln!("CLAWHDF5_REMOTE_CORPUS not set; skipping the corpus");
        return;
    };
    let report = std::env::var("CLAWHDF5_REMOTE_REPORT").is_ok_and(|v| v == "1");
    let (same, refused) = compare(&files, report);
    eprintln!("corpus: {same} files identical, {refused} refused by both");
    assert!(same > 0);
}

#[test]
fn multi_block_file_values_and_request_budget() {
    let bytes = multi_block_file();
    assert!(bytes.len() > 3 << 20, "{}", bytes.len());
    let server = Server::start(vec![("/m.h5".into(), bytes.clone())]);
    let url = server.url("/m.h5");
    let local = File::from_bytes(bytes.clone()).unwrap();
    let storage = storage_for_url(&url, &quick()).unwrap();
    assert_eq!(server.requests(), 1, "opening is one request");
    let remote = File::open_storage(storage.clone()).unwrap();
    assert!(remote.contiguous_bytes().is_none());
    let mut names = remote.root().datasets().unwrap();
    names.sort();
    assert_eq!(names, ["big", "flat"]);
    assert_eq!(remote.root().groups().unwrap(), ["grp"]);
    assert_eq!(
        remote.dataset("grp/small").unwrap().read_f64().unwrap(),
        [1.0, 2.0, 3.0]
    );
    let big = remote.dataset("big").unwrap().read_f64().unwrap();
    assert_eq!(big, local.dataset("big").unwrap().read_f64().unwrap());
    let flat = remote.dataset("flat").unwrap().read_f64().unwrap();
    assert_eq!(flat.len(), 300_000);
    assert_eq!(flat[299_999], 299_999.0);
    // Every byte was fetched at most once, in whole 1 MiB blocks.
    let log = server.log();
    let mut blocks: Vec<u64> = Vec::new();
    for (_, r) in &log {
        let (a, b) = r.expect("every request is ranged");
        assert_eq!(a % (1 << 20), 0, "block-aligned");
        blocks.extend(a >> 20..=b >> 20);
    }
    let n = blocks.len();
    blocks.sort_unstable();
    blocks.dedup();
    assert_eq!(n, blocks.len(), "a block was fetched twice: {log:?}");
    assert!(
        server.requests() <= (bytes.len() as u64 >> 20) + 2,
        "{} requests for a {} byte file",
        server.requests(),
        bytes.len()
    );
    // Reading again costs nothing.
    let before = server.requests();
    assert_eq!(transcript(&remote), transcript(&local));
    assert_eq!(server.requests(), before);
}

#[test]
fn h5py_written_file_over_http() {
    if !common::have_h5py() {
        return;
    }
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("py.h5");
    common::run_python(
        r#"
import sys, h5py, numpy as np
with h5py.File(sys.argv[1], "w", libver="latest") as f:
    f.attrs["title"] = "h5py remote test"
    g = f.create_group("sensors")
    for i in range(40):
        d = g.create_dataset(f"s{i:02d}", data=np.arange(1000, dtype="<i4") * (i + 1))
        d.attrs["index"] = i
    t = f.create_dataset("temps", data=np.sin(np.arange(2_000_000) / 1000.0),
                         chunks=(50_000,), compression="gzip", shuffle=True)
    t.attrs["units"] = "K"
    f.create_dataset("names", data=[b"alpha", b"beta", b"gamma"])
    f.create_dataset("vl", data=["one", "two", "three"], dtype=h5py.string_dtype())
    f["link"] = h5py.SoftLink("/sensors/s03")
"#,
        &[path.to_str().unwrap()],
    );
    // What libhdf5 reads, for the values the test checks.
    let sums = common::run_python(
        r#"
import sys, h5py, numpy as np
with h5py.File(sys.argv[1], "r") as f:
    print(repr(float(f["temps"][:].sum())), int(f["sensors/s39"][:].sum()), f["vl"].asstr()[1])
"#,
        &[path.to_str().unwrap()],
    );
    let bytes = std::fs::read(&path).unwrap();
    let server = Server::start(vec![("/py.h5".into(), bytes.clone())]);
    let remote = open_url_with(&server.url("/py.h5"), &quick()).unwrap();
    let local = File::open(&path).unwrap();
    assert_eq!(transcript(&remote), transcript(&local));
    let temps = remote.dataset("temps").unwrap().read_f64().unwrap();
    let s39: i64 = remote
        .dataset("sensors/s39")
        .unwrap()
        .read_i64()
        .unwrap()
        .iter()
        .sum();
    let vl = remote.dataset("vl").unwrap().read_string().unwrap();
    let mut it = sums.split_whitespace();
    let want_sum: f64 = it.next().unwrap().parse().unwrap();
    let got_sum: f64 = temps.iter().sum();
    assert!((got_sum - want_sum).abs() <= 1e-6 * want_sum.abs().max(1.0));
    assert_eq!(s39, it.next().unwrap().parse::<i64>().unwrap());
    assert_eq!(vl[1], it.next().unwrap());
    assert_eq!(
        remote.dataset("link").unwrap().read_i32().unwrap(),
        local.dataset("sensors/s03").unwrap().read_i32().unwrap()
    );
}

#[test]
fn a_server_that_ignores_range_is_refused_or_downloaded_when_allowed() {
    let bytes = multi_block_file();
    let server = Server::start(vec![("/m.h5".into(), bytes.clone())]);
    server.shared.ignore_range.store(true, Ordering::SeqCst);
    let url = server.url("/m.h5");
    let err = open_url_with(&url, &quick()).unwrap_err();
    assert!(
        matches!(err, Error::Remote(RemoteError::RangeNotSupported(_))),
        "{err}"
    );
    assert_eq!(server.requests(), 1, "refused at the first response");
    let mut opts = quick();
    opts.http.allow_full_download = true;
    let storage = storage_for_url(&url, &opts).unwrap();
    let f = File::open_storage(storage.clone()).unwrap();
    assert_eq!(f.contiguous_bytes(), Some(&bytes[..]), "read from memory");
    assert_eq!(server.requests(), 2);
    let local = File::from_bytes(bytes).unwrap();
    assert_eq!(transcript(&f), transcript(&local));
    assert_eq!(server.requests(), 2, "no further requests");
    // Too large for the download limit.
    opts.http.max_full_download = 1000;
    assert!(open_url_with(&url, &opts).is_err());
}

#[test]
fn a_file_replaced_mid_read_is_an_error_not_mixed_data() {
    for mode in ["etag", "last-modified", "none"] {
        let bytes = multi_block_file();
        let server = Server::start(vec![("/m.h5".into(), bytes.clone())]);
        match mode {
            "last-modified" => server.shared.no_etag.store(true, Ordering::SeqCst),
            "none" => server.shared.no_validators.store(true, Ordering::SeqCst),
            _ => {}
        }
        let url = server.url("/m.h5");
        let f = open_url_with(&url, &quick()).unwrap();
        assert_eq!(f.root().groups().unwrap(), ["grp"], "{mode}");
        // Replace the file: same layout, other values (and, for "none",
        // one byte longer, which is all that can be checked).
        let mut other = bytes.clone();
        let n = other.len();
        for b in &mut other[n / 2..n / 2 + 1000] {
            *b ^= 0xff;
        }
        if mode == "none" {
            other.push(0);
        }
        server.shared.put("/m.h5", other);
        let err = f
            .dataset("big")
            .unwrap()
            .read_f64()
            .unwrap_err()
            .to_string();
        assert!(changed(&err), "{mode}: {err}");
    }
}

#[test]
fn a_weak_etag_or_no_validator_is_refused_when_required() {
    let server = Server::start(vec![("/m.h5".into(), multi_block_file())]);
    server.shared.weak_etag.store(true, Ordering::SeqCst);
    let mut opts = quick();
    opts.http.require_validator = true;
    assert!(open_url_with(&server.url("/m.h5"), &opts).is_err());
    opts.http.require_validator = false;
    assert!(open_url_with(&server.url("/m.h5"), &opts).is_ok());
}

#[test]
fn truncated_bodies_are_retried_then_an_error() {
    let bytes = multi_block_file();
    let server = Server::start(vec![("/m.h5".into(), bytes.clone())]);
    let url = server.url("/m.h5");
    let local = File::from_bytes(bytes).unwrap();
    let want = local.dataset("big").unwrap().read_f64().unwrap();

    // One cut-off body: retried, right values.
    let (http, first) = HttpStorage::open(&url, quick().http).unwrap();
    assert_eq!(first.len(), 1 << 20);
    let storage = Arc::new(clawhdf5_remote::BlockCache::new(
        http,
        CacheConfig::default(),
    ));
    let f = File::open_storage(storage.clone()).unwrap();
    server.shared.truncate_next.store(1, Ordering::SeqCst);
    assert_eq!(f.dataset("big").unwrap().read_f64().unwrap(), want);
    assert_eq!(storage.inner().stats().retries, 1);

    // Every body cut off: an error, never partial data.
    let f = open_url_with(&url, &quick()).unwrap();
    server.shared.truncate_next.store(1000, Ordering::SeqCst);
    let err = f
        .dataset("big")
        .unwrap()
        .read_f64()
        .unwrap_err()
        .to_string();
    assert!(
        err.contains("network error") || err.contains("bad response"),
        "{err}"
    );
    server.shared.truncate_next.store(0, Ordering::SeqCst);
    // The failure was not cached: the next read succeeds.
    assert_eq!(f.dataset("big").unwrap().read_f64().unwrap(), want);
}

#[test]
fn server_errors_are_retried_with_backoff() {
    let bytes = multi_block_file();
    let server = Server::start(vec![("/m.h5".into(), bytes.clone())]);
    server.shared.fail_next.store(2, Ordering::SeqCst);
    let url = server.url("/m.h5");
    let f = open_url_with(&url, &quick()).unwrap();
    let local = File::from_bytes(bytes).unwrap();
    server.shared.fail_next.store(3, Ordering::SeqCst);
    assert_eq!(
        f.dataset("flat").unwrap().read_f64().unwrap(),
        local.dataset("flat").unwrap().read_f64().unwrap()
    );
    // More failures than retries: an error.
    let f = open_url_with(&url, &quick()).unwrap();
    server.shared.fail_next.store(100, Ordering::SeqCst);
    let err = f
        .dataset("big")
        .unwrap()
        .read_f64()
        .unwrap_err()
        .to_string();
    assert!(err.contains("503"), "{err}");
}

#[test]
fn slow_server_concurrent_readers_fetch_each_block_once() {
    let bytes = multi_block_file();
    let server = Server::start(vec![("/m.h5".into(), bytes.clone())]);
    let url = server.url("/m.h5");
    let opts = Options {
        cache: CacheConfig {
            block_size: 256 << 10,
            max_request: 256 << 10,
            ..CacheConfig::default()
        },
        ..quick()
    };
    let storage = storage_for_url(&url, &opts).unwrap();
    let file = File::open_storage(storage.clone()).unwrap();
    let local = File::from_bytes(bytes).unwrap();
    let want_big = local.dataset("big").unwrap().read_f64().unwrap();
    let want_flat = local.dataset("flat").unwrap().read_f64().unwrap();
    server.shared.delay_ms.store(40, Ordering::SeqCst);
    server.reset();
    std::thread::scope(|s| {
        for t in 0..8 {
            let (file, want_big, want_flat) = (&file, &want_big, &want_flat);
            s.spawn(move || {
                if t % 2 == 0 {
                    assert_eq!(&file.dataset("big").unwrap().read_f64().unwrap(), want_big);
                } else {
                    assert_eq!(
                        &file.dataset("flat").unwrap().read_f64().unwrap(),
                        want_flat
                    );
                }
            });
        }
    });
    let log = server.log();
    let mut starts: Vec<u64> = log.iter().map(|(_, r)| r.unwrap().0).collect();
    let n = starts.len();
    starts.sort_unstable();
    starts.dedup();
    assert_eq!(n, starts.len(), "a block was fetched twice");
    assert!(storage.stats().waits > 0, "readers shared fetches");
}

#[test]
fn bad_urls_and_statuses_are_clean_errors() {
    let server = Server::start(vec![("/m.h5".into(), multi_block_file())]);
    let err = open_url_with(&server.url("/missing.h5"), &quick()).unwrap_err();
    assert!(
        matches!(err, Error::Remote(RemoteError::Status { code: 404, .. })),
        "{err}"
    );
    assert!(matches!(
        open_url("ftp://example.com/a.h5"),
        Err(Error::Remote(RemoteError::UnsupportedScheme(_)))
    ));
    assert!(matches!(
        open_url("no-scheme"),
        Err(Error::Remote(RemoteError::InvalidUrl(_)))
    ));
    #[cfg(not(feature = "s3"))]
    {
        let e = open_url("s3://bucket/key.h5").unwrap_err().to_string();
        assert!(e.contains("`s3` feature"), "{e}");
    }
    #[cfg(not(feature = "https"))]
    {
        let e = open_url("https://example.com/a.h5")
            .unwrap_err()
            .to_string();
        assert!(e.contains("`https` feature"), "{e}");
    }
    // A content-encoded body is not a byte range.
    let server = Server::start(vec![("/m.h5".into(), multi_block_file())]);
    server.shared.gzip_label.store(true, Ordering::SeqCst);
    let e = open_url(&server.url("/m.h5")).unwrap_err().to_string();
    assert!(e.contains("gzip-encoded"), "{e}");
    // Not HDF5.
    let server = Server::start(vec![("/x.h5".into(), vec![7u8; 5000])]);
    assert!(matches!(
        open_url(&server.url("/x.h5")),
        Err(Error::Hdf5(_))
    ));
    // A server that closes every connection unanswered: a network error
    // after the retries. (Not a closed port: another test's server could
    // take it meanwhile.)
    let port = {
        let l = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let port = l.local_addr().unwrap().port();
        std::thread::spawn(move || {
            for c in l.incoming() {
                drop(c);
            }
        });
        port
    };
    let err = open_url_with(&format!("http://127.0.0.1:{port}/a.h5"), &quick()).unwrap_err();
    assert!(
        matches!(err, Error::Remote(RemoteError::Transport(_))),
        "{err}"
    );
}