static USAGE: &str = r#"
Removes qsv-generated cache files to reduce clutter and simplify data packaging.

By default, clean removes the three auto-regenerable CACHES qsv writes next to a
source file:

  * <file>.idx                                              (index)
  * <stem>.stats.csv / .stats.weighted.csv / .stats.csv.json
    / .stats.csv.data.jsonl                                 (stats cache)
  * <stem>.freq.csv.data.jsonl                              (frequency cache)

Only files that are VERIFIABLY generated by qsv are removed, so a user file that
merely happens to be named e.g. report.stats.csv is never touched (qsv stats
caches are only removed when their companion .stats.csv.json sidecar is present).

clean is, by default, a DRY-RUN: it only lists what it WOULD remove (and the
bytes that would be reclaimed). Pass --force to actually delete.

User OUTPUTS from schema, validate and moarstats are NOT caches and are only
removed when you ask for them explicitly with --schema / --validate / --moarstats
(or --all).

Examples:
  Preview the caches qsv would remove from the current directory:
    $ qsv clean

  Actually remove them:
    $ qsv clean --force

  Remove only STALE or orphaned caches (source changed, or source is gone),
  recursively, from a specific directory:
    $ qsv clean --stale --recursive --force ./data

  Remove just the caches & outputs associated with one file:
    $ qsv clean --all --force data.csv

For the tests, see https://github.com/dathere/qsv/blob/master/tests/test_clean.rs.

Usage:
    qsv clean [options] [<input>]
    qsv clean --help

clean options:
    --stale            Only remove STALE or ORPHANED caches: a cache whose
                       source file is newer than the cache, or whose source
                       file no longer exists.
    -r, --recursive    Recurse into subdirectories when <input> is a directory.
    -n, --dry-run      Preview only; never delete. Wins over --force if both
                       are given (this is also the default when neither is set).
    -f, --force        Actually delete the files.

  Category selection. When no category flag is given, the three default caches
  (index, stats & frequency) are cleaned:
    --index            Index (.idx) caches. [default category]
    --stats            Stats caches. [default category]
    --frequency        Frequency caches. [default category]
    --schema           schema .schema.json / .pschema.json outputs.
    --validate         validate .valid / .invalid / .validation-errors.tsv outputs.
    --moarstats        moarstats .stats.bivariate[.joined].csv outputs.
    --all              Select every category above.

Common options:
    -h, --help         Display this message
    -q, --quiet        Do not print the per-file / summary report.
"#;

use std::{
    fs,
    io::{BufRead, BufReader, Read},
    path::{Path, PathBuf},
};

use serde::Deserialize;
use serde_json::Value;

use crate::{CliResult, util};

#[derive(Deserialize)]
struct Args {
    arg_input:      Option<String>,
    flag_stale:     bool,
    flag_recursive: bool,
    flag_dry_run:   bool,
    flag_force:     bool,
    flag_index:     bool,
    flag_stats:     bool,
    flag_frequency: bool,
    flag_schema:    bool,
    flag_validate:  bool,
    flag_moarstats: bool,
    flag_all:       bool,
    flag_quiet:     bool,
}

// Which artifact categories are in scope for this run.
struct Selected {
    index:     bool,
    stats:     bool,
    frequency: bool,
    schema:    bool,
    validate:  bool,
    moarstats: bool,
}

impl Args {
    fn selected(&self) -> Selected {
        // if no category flag (and not --all) is given, default to the three caches
        let any = self.flag_index
            || self.flag_stats
            || self.flag_frequency
            || self.flag_schema
            || self.flag_validate
            || self.flag_moarstats;
        let default_caches = !any && !self.flag_all;
        Selected {
            index:     self.flag_all || self.flag_index || default_caches,
            stats:     self.flag_all || self.flag_stats || default_caches,
            frequency: self.flag_all || self.flag_frequency || default_caches,
            schema:    self.flag_all || self.flag_schema,
            validate:  self.flag_all || self.flag_validate,
            moarstats: self.flag_all || self.flag_moarstats,
        }
    }
}

pub fn run(argv: &[&str]) -> CliResult<()> {
    let args: Args = util::get_args(USAGE, argv)?;
    let sel = args.selected();
    let stale = args.flag_stale;
    // a dry-run unless the user explicitly forces deletion; --dry-run always wins
    let dry_run = args.flag_dry_run || !args.flag_force;

    let input = args.arg_input.clone().unwrap_or_else(|| ".".to_string());
    let input_path = Path::new(&input);

    let mut victims: Vec<(PathBuf, u64)> = Vec::new();

    if input_path.is_file() {
        // single-file mode: <input> is a SOURCE file; clean its associated artifacts
        for anchor in candidate_anchors(input_path) {
            if anchor.is_file() {
                consider_anchor(&anchor, &sel, stale, &mut victims);
            }
        }
    } else if input_path.is_dir() {
        let mut files = Vec::new();
        collect_files(input_path, args.flag_recursive, &mut files)?;
        for f in &files {
            consider_anchor(f, &sel, stale, &mut victims);
        }
    } else {
        return fail_clierror!("Input \"{input}\" is not an existing file or directory.");
    }

    // a family member can be reached only once, but dedupe defensively
    victims.sort();
    victims.dedup();

    let total_bytes: u64 = victims.iter().map(|(_, s)| s).sum();

    if victims.is_empty() {
        if !args.flag_quiet {
            wout!("No qsv cache files found to clean.");
        }
        return Ok(());
    }

    let mut removed_count = 0_usize;
    let mut removed_bytes = 0_u64;
    let mut failed = 0_usize;

    for (path, size) in &victims {
        if dry_run {
            if !args.flag_quiet {
                wout!("would remove: {} ({size} bytes)", path.display());
            }
        } else {
            match fs::remove_file(path) {
                Ok(()) => {
                    removed_count += 1;
                    removed_bytes += size;
                    if !args.flag_quiet {
                        wout!("removed: {} ({size} bytes)", path.display());
                    }
                },
                Err(e) => {
                    failed += 1;
                    wwarn!("could not remove {}: {e}", path.display());
                },
            }
        }
    }

    if !args.flag_quiet {
        if dry_run {
            wout!(
                "Dry-run: would remove {} file(s), {total_bytes} bytes. Pass --force to delete.",
                victims.len()
            );
        } else {
            wout!("Removed {removed_count} file(s), {removed_bytes} bytes.");
        }
    }

    if failed > 0 {
        return fail_clierror!("Failed to remove {failed} file(s).");
    }

    Ok(())
}

/// Recursively (or one-level) collect regular files under `dir`.
fn collect_files(dir: &Path, recursive: bool, out: &mut Vec<PathBuf>) -> std::io::Result<()> {
    for entry in fs::read_dir(dir)? {
        let entry = entry?;
        let ft = entry.file_type()?;
        let path = entry.path();
        if ft.is_dir() {
            if recursive {
                collect_files(&path, recursive, out)?;
            }
        } else if ft.is_file() {
            out.push(path);
        }
    }
    Ok(())
}

/// Classify an on-disk file by its qsv artifact suffix and, if its category is
/// selected, verify provenance and (optionally) staleness before collecting it.
fn consider_anchor(path: &Path, sel: &Selected, stale: bool, victims: &mut Vec<(PathBuf, u64)>) {
    let full = path.to_string_lossy().to_string();

    // each artifact suffix maps to exactly one category, so folding the category
    // check into the condition is safe: a non-selected match simply falls through
    // (no other branch's suffix can also match the same file).
    if sel.index && full.ends_with(".idx") {
        handle_index(path, &full, stale, victims);
    } else if sel.stats
        && (full.ends_with(".stats.csv.json") || full.ends_with(".stats.weighted.csv.json"))
    {
        handle_stats(path, &full, stale, victims);
    } else if sel.frequency && full.ends_with(".freq.csv.data.jsonl") {
        handle_frequency(path, stale, victims);
    } else if sel.schema && full.ends_with(".pschema.json") {
        handle_pschema(path, &full, stale, victims);
    } else if sel.schema && full.ends_with(".schema.json") {
        handle_schema(path, &full, stale, victims);
    } else if sel.moarstats
        && (full.ends_with(".stats.bivariate.joined.csv") || full.ends_with(".stats.bivariate.csv"))
    {
        handle_moarstats(path, &full, stale, victims);
    } else if sel.validate
        && (full.ends_with(".validation-errors.tsv")
            || full.ends_with(".valid")
            || full.ends_with(".invalid"))
    {
        handle_validate(path, &full, stale, victims);
    }
}

// ---- per-category handlers ---------------------------------------------------

fn handle_index(path: &Path, full: &str, stale: bool, victims: &mut Vec<(PathBuf, u64)>) {
    let Ok(md) = fs::metadata(path) else {
        return;
    };
    let size = md.len();
    let Some(base) = full.strip_suffix(".idx") else {
        return;
    };
    let src = PathBuf::from(base);
    // require the source to exist so the index can be FULLY validated against it
    // (offset ordering AND bounds). we never delete an orphaned .idx, whose
    // provenance can't be fully verified without its source.
    if !src.is_file() {
        return;
    }
    if !looks_like_csv_index(path, size, &src) {
        return;
    }
    if !stale || stale_or_orphaned(path, Some(&src)) {
        victims.push((path.to_path_buf(), size));
    }
}

fn handle_stats(path: &Path, full: &str, stale: bool, victims: &mut Vec<(PathBuf, u64)>) {
    // the .stats.csv.json (args) sidecar is the provenance anchor for the whole
    // stats cache family. csv_base strips the trailing ".json".
    let Some(csv_base) = full.strip_suffix(".json") else {
        return;
    };
    let Ok(text) = fs::read_to_string(path) else {
        return;
    };
    let Ok(v) = serde_json::from_str::<Value>(&text) else {
        return;
    };
    if json_str(&v, "qsv_version").is_none() {
        return;
    }
    // resolve the recorded source as a sibling of the cache (caches are always
    // written next to their source) so staleness is correct regardless of the
    // cwd clean is run from, or the directory having been moved since generation.
    let source = json_str(&v, "canonical_input_path")
        .or_else(|| json_str(&v, "arg_input"))
        .and_then(|recorded| sibling_source(path, recorded));

    if stale && !stale_or_orphaned(path, source.as_deref()) {
        return;
    }

    for member in [
        csv_base.to_string(),
        format!("{csv_base}.json"),
        format!("{csv_base}.data.jsonl"),
    ] {
        push_if_exists(victims, PathBuf::from(member));
    }
}

fn handle_frequency(path: &Path, stale: bool, victims: &mut Vec<(PathBuf, u64)>) {
    let Some(line) = read_first_line(path) else {
        return;
    };
    let Ok(v) = serde_json::from_str::<Value>(&line) else {
        return;
    };
    // provenance: a qsv frequency cache carries its version & input in metadata
    if json_str(&v, "qsv_version").is_none() || json_str(&v, "arg_input").is_none() {
        return;
    }
    if stale {
        // resolve the EXACT recorded source (symlink-resolved, with its real
        // extension) as a sibling of the cache — robust to cwd, symlinks and
        // directory moves. legacy caches predating canonical_input_path can't be
        // resolved reliably, so we KEEP them (conservative) rather than risk
        // deleting a fresh cache as falsely "orphaned".
        let Some(canonical) = json_str(&v, "canonical_input_path") else {
            return;
        };
        let source = sibling_source(path, canonical);
        if !stale_or_orphaned(path, source.as_deref()) {
            return;
        }
    }
    push_if_exists(victims, path.to_path_buf());
}

fn handle_schema(path: &Path, full: &str, stale: bool, victims: &mut Vec<(PathBuf, u64)>) {
    let Ok(text) = fs::read_to_string(path) else {
        return;
    };
    let Ok(v) = serde_json::from_str::<Value>(&text) else {
        return;
    };
    let is_schema = v
        .get("$schema")
        .and_then(Value::as_str)
        .is_some_and(|s| s.contains("json-schema.org"));
    if !is_schema {
        return;
    }
    let Some(base) = full.strip_suffix(".schema.json") else {
        return;
    };
    let source = PathBuf::from(base);
    if stale && !stale_or_orphaned(path, Some(&source)) {
        return;
    }
    push_if_exists(victims, path.to_path_buf());
}

fn handle_pschema(path: &Path, full: &str, stale: bool, victims: &mut Vec<(PathBuf, u64)>) {
    let Ok(text) = fs::read_to_string(path) else {
        return;
    };
    // a polars schema is JSON; require it to parse as such
    if serde_json::from_str::<Value>(&text).is_err() {
        return;
    }
    let Some(base) = full.strip_suffix(".pschema.json") else {
        return;
    };
    let source = PathBuf::from(base);
    if stale && !stale_or_orphaned(path, Some(&source)) {
        return;
    }
    push_if_exists(victims, path.to_path_buf());
}

fn handle_moarstats(path: &Path, full: &str, stale: bool, victims: &mut Vec<(PathBuf, u64)>) {
    let Some(base) = full
        .strip_suffix(".stats.bivariate.joined.csv")
        .or_else(|| full.strip_suffix(".stats.bivariate.csv"))
    else {
        return;
    };
    // provenance is weak (plain CSV), so require a sibling source file
    let source = find_tabular_sibling(Path::new(base));
    if stale {
        if stale_or_orphaned(path, source.as_deref()) {
            push_if_exists(victims, path.to_path_buf());
        }
    } else if source.is_some() {
        push_if_exists(victims, path.to_path_buf());
    }
}

fn handle_validate(path: &Path, full: &str, stale: bool, victims: &mut Vec<(PathBuf, u64)>) {
    let Some(base) = full
        .strip_suffix(".validation-errors.tsv")
        .or_else(|| full.strip_suffix(".invalid"))
        .or_else(|| full.strip_suffix(".valid"))
    else {
        return;
    };
    // provenance is weak (the splits are plain CSV/TSV), so require sibling source
    let source = PathBuf::from(base);
    if stale {
        if stale_or_orphaned(path, Some(&source)) {
            push_if_exists(victims, path.to_path_buf());
        }
    } else if source.is_file() {
        push_if_exists(victims, path.to_path_buf());
    }
}

// ---- helpers -----------------------------------------------------------------

/// Structurally verify `path` is a qsv csv-index (`csv_index::RandomAccessSimple`)
/// for `source`, not just any nonzero 8-byte-aligned file. The format is a run of
/// big-endian u64 record offsets followed by a trailing u64 record count, so:
///   * size is a non-zero multiple of 8 with at least two entries (16 bytes),
///   * the first offset is 0 (the first record starts at the top of the CSV),
///   * EVERY offset is monotonically non-decreasing and within the source CSV, and
///   * the trailing count equals the number of offsets (`entries - 1`).
///
/// All offsets are streamed through a buffered reader in a single pass, so an index
/// with a malformed intermediate offset is rejected while staying cheap for large
/// indexes.
fn looks_like_csv_index(path: &Path, size: u64, source: &Path) -> bool {
    if size < 16 || !size.is_multiple_of(8) {
        return false;
    }
    let entries = size / 8;
    let n_offsets = entries - 1; // last entry is the trailing record count
    let Ok(source_len) = fs::metadata(source).map(|m| m.len()) else {
        return false;
    };
    let Ok(f) = fs::File::open(path) else {
        return false;
    };
    let mut rdr = BufReader::new(f);
    let mut buf = [0u8; 8];
    let mut prev = 0_u64;
    for i in 0..n_offsets {
        if rdr.read_exact(&mut buf).is_err() {
            return false;
        }
        let offset = u64::from_be_bytes(buf);
        // first offset is 0; offsets are non-decreasing and within the source CSV
        if (i == 0 && offset != 0) || offset < prev || offset > source_len {
            return false;
        }
        prev = offset;
    }
    // trailing u64 is the record count, which must equal the number of offsets
    if rdr.read_exact(&mut buf).is_err() {
        return false;
    }
    u64::from_be_bytes(buf) == n_offsets
}

/// Resolve a source path recorded in cache metadata as a sibling of the cache.
/// Caches are always written next to their source, so the source basename joined
/// to the cache's directory is correct regardless of the cwd `clean` runs from or
/// whether the directory was moved since the cache was generated.
fn sibling_source(anchor: &Path, recorded: &str) -> Option<PathBuf> {
    let name = Path::new(recorded).file_name()?;
    Some(anchor.parent().unwrap_or_else(|| Path::new(".")).join(name))
}

/// The candidate artifact anchor paths for a given source file (single-file mode).
fn candidate_anchors(src: &Path) -> Vec<PathBuf> {
    let mut v = vec![util::idx_path(src)];
    let parent = src.parent().unwrap_or_else(|| Path::new("."));
    if let Some(stem) = src.file_stem() {
        let stem = stem.to_string_lossy();
        v.push(parent.join(format!("{stem}.stats.csv.json")));
        v.push(parent.join(format!("{stem}.stats.weighted.csv.json")));
        v.push(parent.join(format!("{stem}.freq.csv.data.jsonl")));
        v.push(parent.join(format!("{stem}.stats.bivariate.csv")));
        v.push(parent.join(format!("{stem}.stats.bivariate.joined.csv")));
    }
    let full = src.to_string_lossy();
    v.push(PathBuf::from(format!("{full}.schema.json")));
    v.push(PathBuf::from(format!("{full}.pschema.json")));
    v.push(PathBuf::from(format!("{full}.valid")));
    v.push(PathBuf::from(format!("{full}.invalid")));
    v.push(PathBuf::from(format!("{full}.validation-errors.tsv")));
    v
}

/// A verified cache is removable under --stale when its source is gone (orphaned)
/// or the source was modified after the cache (`source mtime > cache mtime`).
fn stale_or_orphaned(anchor: &Path, source: Option<&Path>) -> bool {
    let Some(anchor_mtime) = file_mtime(anchor) else {
        return false;
    };
    match source {
        None => true,
        Some(src) => match file_mtime(src) {
            None => true,
            Some(src_mtime) => src_mtime > anchor_mtime,
        },
    }
}

fn file_mtime(p: &Path) -> Option<u64> {
    fs::metadata(p).ok().map(|md| util::file_metadata(&md).0)
}

fn find_tabular_sibling(stem: &Path) -> Option<PathBuf> {
    for ext in ["csv", "tsv", "ssv", "tab"] {
        let cand = stem.with_extension(ext);
        if cand.is_file() {
            return Some(cand);
        }
    }
    None
}

fn push_if_exists(victims: &mut Vec<(PathBuf, u64)>, p: PathBuf) {
    if let Ok(md) = fs::metadata(&p)
        && md.is_file()
    {
        victims.push((p, md.len()));
    }
}

fn read_first_line(p: &Path) -> Option<String> {
    let f = fs::File::open(p).ok()?;
    let mut line = String::new();
    BufReader::new(f).read_line(&mut line).ok()?;
    Some(line)
}

fn json_str<'a>(v: &'a Value, key: &str) -> Option<&'a str> {
    v.get(key).and_then(Value::as_str).filter(|s| !s.is_empty())
}
