use std::fs;
use std::io;
use std::path::{Path, PathBuf};
use std::time::{Duration, SystemTime};

use anyhow::{Context, Result, anyhow};
use jwalk::{Parallelism, WalkDirGeneric};

/// The signature line every cache directory tag starts with.
///
/// See <https://bford.info/cachedir/>. Cargo writes one of these into every
/// target directory and into `registry/` and `git/` in the cargo home.
const CACHEDIR_TAG_SIGNATURE: &str = "Signature: 8a477f597d28d172789f06886806bc55";

/// The size of a tree and the most recent moment anything in it changed.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Measurement {
    pub size: u64,
    pub newest: SystemTime,
}

/// What one directory entry contributes to its tree's measurement.
type EntryFacts = Option<(u64, SystemTime)>;

/// How to spread a walk across threads, given where the caller is standing.
///
/// A walk spawns its directory reads onto rayon's global pool. If the caller is
/// *already* running on that pool — inside a `par_iter`, say — those tasks
/// queue behind the very work waiting on them, and jwalk gives up and yields an
/// empty tree rather than deadlocking. Silently: an empty tree measures zero
/// bytes and takes its age from the root directory alone, which is to say it
/// looks like a huge stale target that is safe to delete. It is not.
///
/// So a nested walk runs serially. It is slower than it could be, and it is
/// right, and nothing about the caller has to remember this.
fn parallelism() -> Parallelism {
    if rayon::current_thread_index().is_some() {
        Parallelism::Serial
    } else {
        Parallelism::RayonDefaultPool {
            busy_timeout: Duration::from_secs(30),
        }
    }
}

/// Total up a path's bytes and find its newest modification time in one pass.
///
/// Both numbers come from the same walk because callers always want both: the
/// size to report and the mtime to age against. Unreadable entries are skipped
/// rather than failing the walk — a directory we cannot read is a directory we
/// cannot delete either, and the rest of the tree is still worth measuring.
///
/// This is the whole cost of looking at a target directory, and it is one
/// `stat` per file — a large target runs to six figures of them. The work is
/// almost pure I/O wait, so the stats happen inside `process_read_dir`, which
/// jwalk runs on a thread pool as it reads each directory. That parallelises
/// the waiting rather than just the directory reads.
pub fn measure(path: &Path) -> Result<Measurement> {
    // A cutoff in the far future is one nothing can be newer than, so the walk
    // never stops early and always has a measurement to return.
    walk(path, SystemTime::now() + FAR_FUTURE)?
        .ok_or_else(|| anyhow!("{} changed while it was being measured", path.display()))
}

/// Long enough that no file's mtime plausibly reaches it, short enough not to
/// overflow anyone's clock arithmetic.
const FAR_FUTURE: Duration = Duration::from_secs(100 * 365 * 24 * 60 * 60);

/// Measure a tree, unless something in it is newer than `cutoff`.
///
/// Returns `None` as soon as it meets an entry modified after `cutoff`: the
/// tree is in use, and no one is going to need its size. Otherwise it walks to
/// the end and returns the full measurement.
///
/// This asymmetry is the point. Deciding a target is too new to touch takes one
/// recent file, and a target being actively built is full of them — so the
/// question costs a few stats instead of a walk of the whole tree. The case
/// where nothing is newer is the case where the walk runs to completion, and
/// there the size comes for free from a tree that is about to be deleted
/// anyway. Cheap when the answer is "leave it", thorough when it is "delete
/// it", which is the way round that matters.
pub fn measure_if_older_than(path: &Path, cutoff: SystemTime) -> Result<Option<Measurement>> {
    walk(path, cutoff)
}

fn walk(path: &Path, cutoff: SystemTime) -> Result<Option<Measurement>> {
    let root_meta = fs::symlink_metadata(path)
        .with_context(|| format!("reading metadata for {}", path.display()))?;

    let mut size = 0;
    let mut newest = root_meta.modified().unwrap_or(SystemTime::UNIX_EPOCH);
    if newest > cutoff {
        return Ok(None);
    }

    if root_meta.is_file() {
        return Ok(Some(Measurement {
            size: root_meta.len(),
            newest,
        }));
    }

    let walk = WalkDirGeneric::<((), EntryFacts)>::new(path)
        .skip_hidden(false)
        .follow_links(false)
        .parallelism(parallelism())
        .process_read_dir(|_depth, _path, _state, children| {
            for child in children.iter_mut().flatten() {
                // Not `metadata()`: a symlink must weigh what the link weighs,
                // not what it points at, which may be outside the tree
                // entirely — or be this tree, counted twice.
                child.client_state = fs::symlink_metadata(child.path()).ok().map(|meta| {
                    (
                        meta.len(),
                        meta.modified().unwrap_or(SystemTime::UNIX_EPOCH),
                    )
                });
            }
        });

    for entry in walk.into_iter().flatten() {
        let Some((len, modified)) = entry.client_state else {
            continue;
        };
        if modified > cutoff {
            // Dropping the iterator stops the walk; whatever jwalk has already
            // read ahead is wasted, and that is the price of reading ahead.
            return Ok(None);
        }
        size += len;
        if modified > newest {
            newest = modified;
        }
    }

    Ok(Some(Measurement { size, newest }))
}

/// Delete a file or directory tree, treating an already-absent path as success.
///
/// Reports whether anything was actually there. Callers total up what they
/// freed from what this says went, so a path that had already vanished
/// contributes nothing — which is what lets a caller skip checking in advance
/// whether its records match the disk, and still report an exact figure.
pub fn remove(path: &Path) -> Result<bool> {
    let meta = match fs::symlink_metadata(path) {
        Ok(meta) => meta,
        Err(err) if err.kind() == io::ErrorKind::NotFound => return Ok(false),
        Err(err) => {
            return Err(err).with_context(|| format!("reading metadata for {}", path.display()));
        }
    };

    let result = if meta.is_dir() {
        fs::remove_dir_all(path)
    } else {
        fs::remove_file(path)
    };

    match result {
        Ok(()) => Ok(true),
        // Something else got there first, which is still the outcome we wanted.
        Err(err) if err.kind() == io::ErrorKind::NotFound => Ok(false),
        Err(err) => Err(err).with_context(|| format!("removing {}", path.display())),
    }
}

/// Whether a directory carries a cache directory tag.
pub fn has_cachedir_tag(dir: &Path) -> bool {
    let Ok(contents) = fs::read_to_string(dir.join("CACHEDIR.TAG")) else {
        return false;
    };
    contents.starts_with(CACHEDIR_TAG_SIGNATURE)
}

/// The cache of `rustc`'s version probe cargo keeps at a target directory's root.
///
/// Cargo has written this since long before it began tagging target directories,
/// so it recognises an old build the [`CACHEDIR_TAG_SIGNATURE`] check would miss.
const RUSTC_INFO_CACHE: &str = ".rustc_info.json";

/// Whether cargo has built into `dir`, making its contents regenerable.
///
/// This is the question that decides whether a directory is grunk's to delete.
/// It is not "is this cargo's target directory" — cargo metadata answers *where*
/// build output goes, but a project that has never been built has a `target`
/// path pointing at whatever a person happened to leave there. What matters is
/// whether cargo has actually produced output, and two markers settle it, each
/// written by cargo and by nothing else that would land here: the cache tag
/// modern cargo stamps on the directory, and the rustc-info cache every cargo
/// since has left at its root. Old targets predating the tag carry only the
/// second, which is exactly why the tag alone is not enough.
pub fn is_cargo_build_output(dir: &Path) -> bool {
    has_cachedir_tag(dir) || dir.join(RUSTC_INFO_CACHE).is_file()
}

/// Whether `dir` is a cargo target directory.
///
/// Two things have to hold: cargo has built into the directory, and its parent
/// is a cargo package. Neither alone is enough — a neighbouring `Cargo.toml`
/// says nothing about what the directory is, and cargo's build markers turn up
/// under other tools too (`.pytest_cache` writes the same cache tag). Together
/// they are decisive.
pub fn is_cargo_target_dir(dir: &Path) -> bool {
    is_cargo_build_output(dir)
        && dir
            .parent()
            .is_some_and(|parent| parent.join("Cargo.toml").is_file())
}

/// Expand a leading `~` to the user's home directory.
///
/// Paths in the config file are written by hand, so they get the shell's
/// courtesy even though no shell has been near them.
pub fn expand_tilde(path: &Path) -> PathBuf {
    let Ok(rest) = path.strip_prefix("~") else {
        return path.to_path_buf();
    };
    match dirs::home_dir() {
        Some(home) => home.join(rest),
        None => path.to_path_buf(),
    }
}

/// Render a byte count the way a human reads a disk usage report.
pub fn human_size(bytes: u64) -> String {
    const UNITS: [&str; 6] = ["B", "KiB", "MiB", "GiB", "TiB", "PiB"];

    let mut value = bytes as f64;
    let mut unit = 0;
    while value >= 1024.0 && unit < UNITS.len() - 1 {
        value /= 1024.0;
        unit += 1;
    }

    if unit == 0 {
        format!("{bytes} B")
    } else {
        format!("{value:.1} {}", UNITS[unit])
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    use std::fs::File;
    use std::io::Write;

    use tempfile::TempDir;

    fn write(path: &Path, contents: &str) {
        if let Some(parent) = path.parent() {
            fs::create_dir_all(parent).unwrap();
        }
        File::create(path)
            .unwrap()
            .write_all(contents.as_bytes())
            .unwrap();
    }

    fn cargo_tag() -> &'static str {
        "Signature: 8a477f597d28d172789f06886806bc55\n\
         # This file is a cache directory tag created by cargo.\n"
    }

    #[test]
    fn human_size_scales_and_keeps_bytes_exact() {
        assert_eq!(human_size(0), "0 B");
        assert_eq!(human_size(999), "999 B");
        assert_eq!(human_size(1024), "1.0 KiB");
        assert_eq!(human_size(1536), "1.5 KiB");
        assert_eq!(human_size(3 * 1024 * 1024 * 1024), "3.0 GiB");
    }

    #[test]
    fn measure_totals_a_tree() {
        let tmp = TempDir::new().unwrap();
        write(&tmp.path().join("a.txt"), "0123456789");
        write(&tmp.path().join("nested/b.txt"), "01234");

        let measured = measure(tmp.path()).unwrap();
        assert!(
            measured.size >= 15,
            "expected at least the file bytes, got {}",
            measured.size
        );
    }

    #[test]
    fn measure_counts_the_directory_itself() {
        // A directory occupies space of its own, and `du` counts it. Pinning it
        // against the directory's own metadata rather than a literal keeps this
        // honest across filesystems, which disagree about the number.
        let tmp = TempDir::new().unwrap();
        let empty = tmp.path().join("empty");
        fs::create_dir(&empty).unwrap();

        assert_eq!(
            measure(&empty).unwrap().size,
            fs::symlink_metadata(&empty).unwrap().len()
        );
    }

    #[test]
    fn measure_counts_every_directory_in_the_tree() {
        let tmp = TempDir::new().unwrap();
        let root = tmp.path().join("root");
        let nested = root.join("nested");
        fs::create_dir_all(&nested).unwrap();
        fs::write(nested.join("file.txt"), "0123456789").unwrap();

        let expected = fs::symlink_metadata(&root).unwrap().len()
            + fs::symlink_metadata(&nested).unwrap().len()
            + 10;
        assert_eq!(measure(&root).unwrap().size, expected);
    }

    #[test]
    fn measure_is_correct_when_called_from_inside_a_parallel_iterator() {
        // Regression: `measure` spawns its directory reads onto rayon's global
        // pool. Called from inside a `par_iter` on that same pool, those tasks
        // starved and the walk yielded nothing at all — reporting a 34 GiB
        // target as 0 bytes, aged from its own directory, i.e. stale enough to
        // delete. Enough trees to fill the pool, each deep enough to need
        // several directory reads.
        use rayon::prelude::*;

        let tmp = TempDir::new().unwrap();
        let trees: Vec<PathBuf> = (0..64)
            .map(|t| {
                let root = tmp.path().join(format!("tree{t}"));
                for d in 0..8 {
                    let dir = root.join(format!("d{d}"));
                    for f in 0..8 {
                        write(&dir.join(format!("f{f}.txt")), "0123456789");
                    }
                }
                root
            })
            .collect();

        let expected: Vec<u64> = trees.iter().map(|t| measure(t).unwrap().size).collect();
        assert!(
            expected.iter().all(|&size| size >= 640),
            "fixture should hold 64 files of 10 bytes per tree: {expected:?}"
        );

        let in_parallel: Vec<u64> = trees.par_iter().map(|t| measure(t).unwrap().size).collect();

        assert_eq!(
            in_parallel, expected,
            "measuring inside a par_iter must agree with measuring outside one"
        );
    }

    #[test]
    fn measure_reports_a_files_own_size() {
        let tmp = TempDir::new().unwrap();
        let file = tmp.path().join("a.txt");
        write(&file, "0123456789");

        assert_eq!(measure(&file).unwrap().size, 10);
    }

    #[test]
    fn target_dir_needs_both_a_tag_and_a_manifest() {
        let tmp = TempDir::new().unwrap();
        let project = tmp.path().join("proj");
        let target = project.join("target");
        fs::create_dir_all(&target).unwrap();

        // Tag but no manifest: not a target dir.
        write(&target.join("CACHEDIR.TAG"), cargo_tag());
        assert!(!is_cargo_target_dir(&target));

        // Manifest as well: now it is.
        write(&project.join("Cargo.toml"), "[package]\n");
        assert!(is_cargo_target_dir(&target));
    }

    #[test]
    fn manifest_beside_an_unbuilt_target_is_not_a_target_dir() {
        let tmp = TempDir::new().unwrap();
        let project = tmp.path().join("proj");
        write(&project.join("Cargo.toml"), "[package]\n");
        // A `target` with a manifest beside it but nothing cargo ever wrote:
        // the path where output would go, holding data cargo did not make.
        fs::create_dir_all(project.join("target")).unwrap();

        assert!(!is_cargo_target_dir(&project.join("target")));
    }

    #[test]
    fn an_old_target_is_recognised_by_its_rustc_info_alone() {
        // Cargo tagged target directories only from 1.42 on. A build older than
        // that carries `.rustc_info.json` but no `CACHEDIR.TAG`, and is every
        // bit as regenerable — so it must still read as a target directory.
        let tmp = TempDir::new().unwrap();
        let project = tmp.path().join("proj");
        let target = project.join("target");
        write(&project.join("Cargo.toml"), "[package]\n");
        write(&target.join(".rustc_info.json"), "{}");
        assert!(!has_cachedir_tag(&target));

        assert!(is_cargo_build_output(&target));
        assert!(is_cargo_target_dir(&target));
    }

    #[test]
    fn build_output_is_either_marker_and_hand_made_data_is_neither() {
        let tmp = TempDir::new().unwrap();

        let tagged = tmp.path().join("tagged");
        write(&tagged.join("CACHEDIR.TAG"), cargo_tag());
        assert!(is_cargo_build_output(&tagged));

        let old = tmp.path().join("old");
        write(&old.join(".rustc_info.json"), "{}");
        assert!(is_cargo_build_output(&old));

        // A directory a person filled themselves carries neither marker.
        let handmade = tmp.path().join("handmade");
        write(&handmade.join("precious.txt"), "do not delete");
        assert!(!is_cargo_build_output(&handmade));
    }

    #[test]
    fn a_foreign_cache_tag_is_not_a_target_dir() {
        // pytest writes the same signature into its own cache directory.
        let tmp = TempDir::new().unwrap();
        let project = tmp.path().join("proj");
        write(&project.join("Cargo.toml"), "[package]\n");
        let cache = project.join(".pytest_cache");
        write(
            &cache.join("CACHEDIR.TAG"),
            "Signature: 8a477f597d28d172789f06886806bc55\n\
             # This file is a cache directory tag created by pytest.\n",
        );

        // The tag check passes, so only the caller's target-shaped naming
        // separates these; `has_cachedir_tag` is deliberately permissive.
        assert!(has_cachedir_tag(&cache));
    }

    #[test]
    fn remove_tolerates_a_missing_path() {
        let tmp = TempDir::new().unwrap();
        assert!(remove(&tmp.path().join("never-existed")).is_ok());
    }

    #[test]
    fn remove_deletes_files_and_trees() {
        let tmp = TempDir::new().unwrap();
        let file = tmp.path().join("a.txt");
        write(&file, "x");
        remove(&file).unwrap();
        assert!(!file.exists());

        let tree = tmp.path().join("tree");
        write(&tree.join("deep/b.txt"), "x");
        remove(&tree).unwrap();
        assert!(!tree.exists());
    }
}