//! `File::open_storage` over a read_at-only storage reads every file as
//! `File::open` does (range reads, milestone M2 in
//! `docs/design/range-reads.md`).
//!
//! Each file is read end to end twice — through `File::open` (the mmap
//! fast path) and through `File::open_storage` over a
//! [`CountingStorage`], which serves the file through `read_at` only
//! (`as_contiguous()` is `None`, so no reader can fall back to a slice of
//! the whole file) — and the two transcripts must be identical: the tree
//! (every group's entries, followed by address), every object's attributes
//! (the whole map and each one by name), and every dataset's shape, types
//! and values (all bytes, as `f64`, a hyperslab of them, strings and
//! variable-length sequences).
//!
//! The storage also counts its `read_at` calls and bytes: what a remote
//! backend without a cache would be asked for. The totals and the files
//! that cost most are printed.
//!
//! - `CLAWHDF5_STORAGE_CORPUS=dir[:dir...]` adds every HDF5 file under those
//!   directories (the conformance corpus is `conformance/.cache/corpus`);
//!   `CLAWHDF5_STORAGE_REPORT=1` prints every file's counts.

use std::collections::{BTreeMap, HashSet, VecDeque};
use std::fmt::Write as _;
use std::path::{Path, PathBuf};
use std::sync::Arc;

use clawhdf5::{DType, File, Selection};
use clawhdf5_format::error::FormatError;
use clawhdf5_format::storage::CountingStorage;

/// Objects visited per file.
const MAX_OBJECTS: usize = 2000;
/// Datasets with more bytes than this are not read (their metadata is).
const MAX_DATA_BYTES: u64 = 64 << 20;

/// A short, stable digest of a value's `Debug` form.
fn digest<T: std::fmt::Debug>(v: &T) -> String {
    let s = format!("{v:?}");
    if s.len() <= 200 {
        return s;
    }
    let mut h = 0xcbf2_9ce4_8422_2325u64;
    for b in s.bytes() {
        h = (h ^ u64::from(b)).wrapping_mul(0x100_0000_01b3);
    }
    format!("{}…[{} bytes, fnv {h:016x}]", &s[..80], s.len())
}

/// A data read's result: its value, or its error in full (the two paths
/// must fail the same way, not just both fail). One case is known to vary
/// between two `File`s and is allowed for in [`check`]: a full read goes
/// through the file's chunk cache, which lists a damaged dataset's chunks
/// in hash-map order, so which failing chunk it reports varies.
fn value<T: std::fmt::Debug, E: std::fmt::Debug>(r: &Result<T, E>) -> String {
    match r {
        Ok(v) => digest(v),
        Err(e) => format!("Err({e:?})"),
    }
}

fn transcript(file: &File) -> String {
    let mut out = String::new();
    let mut seen = HashSet::new();
    let mut queue = VecDeque::from([(String::from("/"), file.superblock().root_group_address)]);
    while let Some((path, addr)) = queue.pop_front() {
        if seen.len() >= MAX_OBJECTS || !seen.insert(addr) {
            continue;
        }
        let group = file.group_at(addr);
        let entries = group.entries();
        writeln!(out, "{path} @{addr} entries {}", digest(&entries)).unwrap();
        match file.dataset_at(addr) {
            Ok(ds) => dataset(&mut out, &path, &ds),
            Err(e) => writeln!(out, "{path} dataset_at {e:?}").unwrap(),
        }
        let attrs = group.attrs_with_errors().map(|(a, e)| (sorted(a), e));
        writeln!(out, "{path} attrs {}", digest(&attrs)).unwrap();
        if let Ok((attrs, _)) = &attrs {
            for name in attrs.keys().take(50) {
                writeln!(out, "{path} attr {name:?} {}", digest(&group.attr(name))).unwrap();
            }
        }
        if let Ok(entries) = entries {
            for (name, child) in entries {
                queue.push_back((format!("{}/{name}", path.trim_end_matches('/')), child));
                // Name lookups (through the name index of a dense group).
                if queue.len() < 64 {
                    writeln!(
                        out,
                        "{path} group({name:?}) {}",
                        digest(&group.group(&name).map(|_| ()))
                    )
                    .unwrap();
                }
            }
        }
    }
    out
}

fn sorted<V: std::fmt::Debug>(m: std::collections::HashMap<String, V>) -> BTreeMap<String, V> {
    m.into_iter().collect()
}

fn dataset(out: &mut String, path: &str, ds: &clawhdf5::Dataset<'_>) {
    let shape = ds.shape();
    let dtype = ds.dtype();
    writeln!(
        out,
        "{path} shape {} max {} dtype {} raw {}",
        digest(&shape),
        digest(&ds.max_dimensions()),
        digest(&dtype),
        digest(&ds.raw_datatype())
    )
    .unwrap();
    let attrs = ds.attrs_with_errors().map(|(a, e)| (sorted(a), e));
    writeln!(out, "{path} dataset attrs {}", digest(&attrs)).unwrap();
    let (Ok(shape), Ok(dtype), Ok(raw_dt)) = (shape, dtype, ds.raw_datatype()) else {
        return;
    };
    let elements = shape.iter().try_fold(1u64, |a, &d| a.checked_mul(d));
    let bytes = elements.and_then(|n| n.checked_mul(u64::from(raw_dt.type_size())));
    if bytes.is_none_or(|b| b > MAX_DATA_BYTES) {
        writeln!(out, "{path} too large to read").unwrap();
        return;
    }
    writeln!(
        out,
        "{path} all {}",
        value(&ds.read_selection(&Selection::All))
    )
    .unwrap();
    let numeric = matches!(
        dtype,
        DType::F32
            | DType::F64
            | DType::I8
            | DType::I16
            | DType::I32
            | DType::I64
            | DType::U8
            | DType::U16
            | DType::U32
            | DType::U64
    );
    if numeric {
        writeln!(out, "{path} f64 {}", value(&ds.read_f64())).unwrap();
        writeln!(out, "{path} f32 {}", value(&ds.read_f32())).unwrap();
        writeln!(out, "{path} i64 {}", value(&ds.read_i64())).unwrap();
        if let Some(&d0) = shape.first() {
            let rank = shape.len();
            let sel = Selection::Hyperslab {
                start: std::iter::once(d0 / 3)
                    .chain(std::iter::repeat_n(0, rank - 1))
                    .collect(),
                stride: vec![1; rank],
                count: std::iter::once(d0.div_ceil(3))
                    .chain(shape[1..].iter().copied())
                    .collect(),
                block: vec![1; rank],
            };
            writeln!(
                out,
                "{path} f64 third {}",
                value(&ds.read_f64_selection(&sel))
            )
            .unwrap();
            writeln!(
                out,
                "{path} bytes third {}",
                value(&ds.read_selection(&sel))
            )
            .unwrap();
            // Every third row (a strided hyperslab), and a few points out
            // of order: the last element, the first, one in the middle.
            let strided = Selection::Hyperslab {
                start: vec![0; rank],
                stride: std::iter::once(3)
                    .chain(std::iter::repeat_n(1, rank - 1))
                    .collect(),
                count: std::iter::once(d0.div_ceil(3))
                    .chain(shape[1..].iter().copied())
                    .collect(),
                block: vec![1; rank],
            };
            writeln!(
                out,
                "{path} f64 strided {}",
                value(&ds.read_f64_selection(&strided))
            )
            .unwrap();
            if shape.iter().all(|&d| d > 0) {
                let points = Selection::Points(vec![
                    shape.iter().map(|&d| d - 1).collect(),
                    vec![0; rank],
                    shape.iter().map(|&d| d / 2).collect(),
                ]);
                writeln!(
                    out,
                    "{path} bytes points {}",
                    value(&ds.read_selection(&points))
                )
                .unwrap();
                writeln!(
                    out,
                    "{path} i64 points {}",
                    value(&ds.read_i64_selection(&points))
                )
                .unwrap();
            }
        }
    }
    match &raw_dt {
        clawhdf5_format::datatype::Datatype::String { .. }
        | clawhdf5_format::datatype::Datatype::VariableLength {
            is_string: true, ..
        } => {
            writeln!(out, "{path} strings {}", value(&ds.read_string_bytes())).unwrap();
            writeln!(out, "{path} string {}", value(&ds.read_string())).unwrap();
        }
        clawhdf5_format::datatype::Datatype::VariableLength { .. } => {
            writeln!(out, "{path} vlen {}", value(&ds.read_vlen::<f64>())).unwrap();
        }
        _ => {}
    }
}

/// External virtual-dataset sources as `File::open` finds them: files in
/// the same directory.
fn sibling_resolver(dir: Option<PathBuf>) -> clawhdf5::VdsResolver {
    Arc::new(move |name: &str| {
        let Some(dir) = dir.as_ref() else {
            return Err(FormatError::ChunkedReadError(format!(
                "virtual dataset source file {name:?} cannot be resolved for an in-memory file"
            )));
        };
        let p = Path::new(name);
        if name.is_empty()
            || !p.components().all(|c| {
                matches!(
                    c,
                    std::path::Component::Normal(_) | std::path::Component::CurDir
                )
            })
        {
            return Err(FormatError::ChunkedReadError(format!(
                "virtual dataset source file {name:?} is outside the virtual file's \
                 directory and is not followed"
            )));
        }
        match std::fs::read(dir.join(p)) {
            Ok(bytes) => Ok(Some(bytes)),
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
            Err(e) => Err(FormatError::ChunkedReadError(format!(
                "cannot read virtual dataset source file {name:?}: {e}"
            ))),
        }
    })
}

#[derive(Default)]
struct Totals {
    files: usize,
    opened: usize,
    /// The comparison's reads (each dataset read several ways).
    reads: u64,
    bytes: u64,
    /// One pass: open, list every group, read every attribute and every
    /// dataset once (`read_selection(All)`).
    pass_reads: u64,
    pass_bytes: u64,
    file_bytes: u64,
    /// (one-pass reads, one-pass bytes, file size, name) per file.
    per_file: Vec<(u64, u64, u64, String)>,
}

/// Open the file and read everything once, as a tree viewer that then
/// shows every value would.
fn one_pass(file: &File) {
    let mut seen = HashSet::new();
    let mut queue = VecDeque::from([file.superblock().root_group_address]);
    while let Some(addr) = queue.pop_front() {
        if seen.len() >= MAX_OBJECTS || !seen.insert(addr) {
            continue;
        }
        let group = file.group_at(addr);
        let _ = group.attrs();
        if let Ok(ds) = file.dataset_at(addr) {
            let small = ds.shape().ok().and_then(|s| {
                let n = s.iter().try_fold(1u64, |a, &d| a.checked_mul(d))?;
                let size = u64::from(ds.raw_datatype().ok()?.type_size());
                n.checked_mul(size).filter(|&b| b <= MAX_DATA_BYTES)
            });
            if small.is_some() {
                let _ = ds.read_selection(&Selection::All);
            }
        }
        if let Ok(entries) = group.entries() {
            queue.extend(entries.into_iter().map(|(_, a)| a));
        }
    }
}

fn check(path: &Path, totals: &mut Totals) {
    let Ok(bytes) = std::fs::read(path) else {
        return;
    };
    let name = path.display().to_string();
    let local = File::open(path);
    let storage = Arc::new(CountingStorage::new(bytes.clone()));
    let resolver = sibling_resolver(path.parent().map(Path::to_path_buf));
    let remote = File::open_storage(storage.clone()).map(|mut f| {
        f.set_vds_resolver(resolver.clone());
        f
    });
    totals.files += 1;
    let (local, remote) = match (local, remote) {
        (Ok(l), Ok(r)) => (l, r),
        (l, r) => {
            // Both refuse the file, with the same error.
            assert_eq!(
                format!("{:?}", l.map(|_| ())),
                format!("{:?}", r.map(|_| ())),
                "{name}: open"
            );
            return;
        }
    };
    totals.opened += 1;
    assert!(remote.contiguous_bytes().is_none(), "{name}");
    assert_eq!(local.user_block_size(), remote.user_block_size(), "{name}");
    let want = transcript(&local);
    let got = transcript(&remote);
    // Every read path works over a storage without the file in memory.
    assert!(
        !got.contains("ContiguousStorageRequired"),
        "{name}: a read needed the file in memory"
    );
    if want != got {
        let first = unexplained_difference(path, &want, &got);
        if let Some(first) = first {
            panic!("{name}: File::open_storage differs from File::open{first}");
        }
    }
    totals.reads += storage.reads();
    totals.bytes += storage.bytes_read();
    totals.file_bytes += bytes.len() as u64;

    let pass = Arc::new(CountingStorage::new(bytes.clone()));
    if let Ok(mut f) = File::open_storage(pass.clone()) {
        f.set_vds_resolver(resolver);
        one_pass(&f);
    }
    let (reads, read_bytes) = (pass.reads(), pass.bytes_read());
    totals.pass_reads += reads;
    totals.pass_bytes += read_bytes;
    if std::env::var("CLAWHDF5_STORAGE_REPORT").is_ok_and(|v| v == "1") {
        eprintln!(
            "{reads:>9} reads {read_bytes:>12} bytes {:>12} file  {name}",
            bytes.len()
        );
    }
    totals
        .per_file
        .push((reads, read_bytes, bytes.len() as u64, name));
}

/// Why the storage transcript `got` differs from the `File::open` one
/// `want`, or `None` when every line that differs is one `File::open` can
/// give too: a line that varies between two `File`s (the chunk cache's
/// hash-map order picks which failing chunk a damaged dataset's full read
/// reports) and whose storage value some fresh `File::open` reproduces.
/// Nothing else is allowed to differ.
fn unexplained_difference(path: &Path, want: &str, got: &str) -> Option<String> {
    let (want, got): (Vec<&str>, Vec<&str>) = (want.lines().collect(), got.lines().collect());
    if want.len() != got.len() {
        return Some(format!("\n {} vs {} lines", want.len(), got.len()));
    }
    let mut open: Vec<usize> = (0..want.len()).filter(|&i| want[i] != got[i]).collect();
    // Only reads that fail on both sides may vary.
    if let Some(&i) = open
        .iter()
        .find(|&&i| !(want[i].contains(" Err(") && got[i].contains(" Err(")))
    {
        return Some(format!("\n local:   {}\n storage: {}", want[i], got[i]));
    }
    for _ in 0..64 {
        let again = transcript(&File::open(path).unwrap());
        let again: Vec<&str> = again.lines().collect();
        open.retain(|&i| again.get(i) != Some(&got[i]));
        if open.is_empty() {
            return None;
        }
    }
    let i = open[0];
    Some(format!(
        "\n local:   {}\n storage: {}\n (no File::open of 64 gave the storage's result)",
        want[i], got[i]
    ))
}

fn report(what: &str, totals: &mut Totals) {
    eprintln!(
        "{what}: {} files ({} open, {} bytes); comparison: {} read_at calls, {} bytes; \
         one pass (list, attributes, every dataset once): {} read_at calls, {} bytes",
        totals.files,
        totals.opened,
        totals.file_bytes,
        totals.reads,
        totals.bytes,
        totals.pass_reads,
        totals.pass_bytes
    );
    totals.per_file.sort_by_key(|a| std::cmp::Reverse(a.0));
    for (reads, bytes, size, name) in totals.per_file.iter().take(10) {
        eprintln!("  {reads:>9} reads {bytes:>12} bytes (file {size:>11})  {name}");
    }
}
fn hdf5_files(dir: &Path, out: &mut Vec<PathBuf>) {
    let Ok(entries) = std::fs::read_dir(dir) else {
        return;
    };
    for e in entries.flatten() {
        let p = e.path();
        if p.is_dir() {
            hdf5_files(&p, out);
        } else if p
            .extension()
            .and_then(|x| x.to_str())
            .is_some_and(|x| matches!(x, "h5" | "hdf5" | "he5" | "nc" | "h5ad" | "hdf"))
        {
            out.push(p);
        }
    }
}

#[test]
fn fixtures_read_identically_through_open_storage() {
    let root = Path::new(env!("CARGO_MANIFEST_DIR"));
    let mut files = Vec::new();
    hdf5_files(&root.join("tests/fixtures"), &mut files);
    hdf5_files(&root.join("../clawhdf5-format/tests/fixtures"), &mut files);
    files.sort();
    assert!(files.len() >= 45, "{} fixtures", files.len());
    let mut totals = Totals::default();
    for f in &files {
        check(f, &mut totals);
    }
    report("fixtures", &mut totals);
    assert!(totals.opened >= 40, "{}", totals.opened);
    assert!(totals.reads > 0);
}

#[test]
fn corpus_reads_identically_through_open_storage() {
    let Ok(dirs) = std::env::var("CLAWHDF5_STORAGE_CORPUS") else {
        eprintln!("CLAWHDF5_STORAGE_CORPUS not set; skipping the corpus");
        return;
    };
    let mut files = Vec::new();
    for d in std::env::split_paths(&dirs) {
        hdf5_files(&d, &mut files);
    }
    files.sort();
    let mut totals = Totals::default();
    for f in &files {
        check(f, &mut totals);
    }
    report("corpus", &mut totals);
    assert!(totals.files > 0);
}

/// A user block, a metadata cache image and a Storage that is itself in
/// memory: the in-memory view of a storage that has one is used as is.
#[test]
fn storage_backed_files_keep_their_zero_copy_views_only_in_memory() {
    let root = Path::new(env!("CARGO_MANIFEST_DIR"));
    let path = root.join("tests/fixtures/h5clear_mdc_image.h5");
    let bytes = std::fs::read(&path).unwrap();
    let local = File::open(&path).unwrap();
    // A Vec<u8> is a Storage with a contiguous view; the image still has
    // to be laid over it, so the view is not used.
    let in_memory = File::open_storage(Arc::new(bytes.clone())).unwrap();
    assert!(in_memory.contiguous_bytes().is_none());
    assert_eq!(transcript(&local), transcript(&in_memory));

    let plain = root.join("../clawhdf5-format/tests/fixtures/chunked_2d.h5");
    let bytes = std::fs::read(&plain).unwrap();
    let in_memory = File::open_storage(Arc::new(bytes.clone())).unwrap();
    assert_eq!(in_memory.contiguous_bytes(), Some(&bytes[..]));
    let counting = File::open_storage(Arc::new(CountingStorage::new(bytes))).unwrap();
    assert!(counting.contiguous_bytes().is_none());
    let result =
        std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| counting.as_bytes().len()));
    assert!(
        result.is_err(),
        "as_bytes over a range storage must not answer"
    );
}

/// A storage that returns 37 junk bytes more than every read asked for.
struct Overlong(Vec<u8>);

impl clawhdf5::Storage for Overlong {
    fn read_at(&self, offset: u64, len: usize) -> Result<std::borrow::Cow<'_, [u8]>, FormatError> {
        let mut v = clawhdf5::Storage::read_at(self.0.as_slice(), offset, len)?.into_owned();
        v.extend(std::iter::repeat_n(0xa5, 37));
        Ok(std::borrow::Cow::Owned(v))
    }

    fn len(&self) -> u64 {
        self.0.len() as u64
    }
}

/// Bytes a misbehaving storage returns past the range asked for are never
/// read as the file's: every fixture reads through it as through
/// `File::open`.
#[test]
fn overlong_storage_reads_identically() {
    let root = Path::new(env!("CARGO_MANIFEST_DIR"));
    let mut files = Vec::new();
    hdf5_files(&root.join("tests/fixtures"), &mut files);
    hdf5_files(&root.join("../clawhdf5-format/tests/fixtures"), &mut files);
    files.sort();
    let mut compared = 0;
    for path in &files {
        let Ok(bytes) = std::fs::read(path) else {
            continue;
        };
        let Ok(local) = File::open(path) else {
            continue;
        };
        let mut remote = File::open_storage(Arc::new(Overlong(bytes)))
            .unwrap_or_else(|e| panic!("{}: {e}", path.display()));
        remote.set_vds_resolver(sibling_resolver(path.parent().map(Path::to_path_buf)));
        assert_eq!(
            transcript(&local),
            transcript(&remote),
            "{}",
            path.display()
        );
        compared += 1;
    }
    assert!(compared >= 40, "{compared}");
}

/// The harness tells failures apart: two different errors are two
/// different transcripts, and only a difference `File::open` itself
/// produces between two opens is let through.
#[test]
fn harness_compares_errors_not_just_failures() {
    let a: Result<(), FormatError> = Err(FormatError::ContiguousStorageRequired("x"));
    let b: Result<(), FormatError> = Err(FormatError::Storage("x".into()));
    assert_ne!(value(&a), value(&b));
    let path = Path::new(env!("CARGO_MANIFEST_DIR"))
        .join("../clawhdf5-format/tests/fixtures/chunked_2d.h5");
    let want = transcript(&File::open(&path).unwrap());
    assert_eq!(unexplained_difference(&path, &want, &want), None);
    // A read that fails differently through the storage.
    let line = want.lines().position(|l| l.contains(" all ")).unwrap();
    let (mut w, mut g): (Vec<String>, Vec<String>) = (
        want.lines().map(String::from).collect(),
        want.lines().map(String::from).collect(),
    );
    w[line] = format!(
        "{} Err(Format(DataSizeMismatch))",
        &w[line][..w[line].find(" all ").unwrap() + 4]
    );
    g[line] = format!(
        "{} Err(Format(Storage(\"injected\")))",
        &g[line][..g[line].find(" all ").unwrap() + 4]
    );
    assert!(unexplained_difference(&path, &w.join("\n"), &g.join("\n")).is_some());
    // A value against an error is never let through.
    assert!(unexplained_difference(&path, &want, &g.join("\n")).is_some());
}

/// `File::storage` is the view `as_bytes` gives, for every backend: the
/// user block skipped, bounded by the end of file, a cache image laid over.
#[test]
fn file_storage_is_the_as_bytes_view_for_every_backend() {
    let root = Path::new(env!("CARGO_MANIFEST_DIR"));
    let mut files = Vec::new();
    hdf5_files(&root.join("tests/fixtures"), &mut files);
    hdf5_files(&root.join("../clawhdf5-format/tests/fixtures"), &mut files);
    let mut compared = 0;
    for p in files {
        let Ok(local) = File::open(&p) else { continue };
        let bytes = std::fs::read(&p).unwrap();
        let remote = File::open_storage(Arc::new(CountingStorage::new(bytes))).unwrap();
        let want = local.as_bytes();
        let view = local.storage();
        assert_eq!(view.as_contiguous(), Some(want), "{}", p.display());
        let got = remote.storage();
        assert!(got.as_contiguous().is_none());
        assert_eq!(got.len(), want.len() as u64, "{}", p.display());
        assert_eq!(
            &*got.read_at(0, want.len()).unwrap(),
            want,
            "{}",
            p.display()
        );
        compared += 1;
    }
    assert!(compared >= 40, "{compared}");
}