//! Runs the `mst-diff` cases from DavidBuchanan314/mst-test-suite.
//!
//! Each case gives two MSTs as CAR files, the record operations between them,
//! and `inductive_proof_nodes`: "the nodes required for MST operation
//! inversion". That is exactly this crate's job, so the test is to load tree
//! *b* from nothing but those nodes, invert the operations, and land on tree
//! *a*'s root.
//!
//! Restricting the blocks to `inductive_proof_nodes` also checks that we do
//! not ask for more of the tree than a producer is obliged to send.
//!
//! A curated subset is vendored under `tests/fixtures/mst-test-suite` and runs
//! by default. The full 16 384 cases run from a checkout:
//!
//! ```text
//! git submodule update --init hubble-commit/tests/mst-test-suite
//! cargo test -p hubble-commit --test mst_diff_suite -- --ignored
//! ```

use std::collections::HashMap;
use std::path::{Path, PathBuf};

use hubble_commit::{CarReader, Cid, Key, Op, OpOrder, invert_ops};
use serde::Deserialize;

#[derive(Debug, Deserialize)]
struct TestCase {
    #[serde(rename = "$type")]
    kind: String,
    description: String,
    inputs: Inputs,
    results: Results,
}

#[derive(Debug, Deserialize)]
struct Inputs {
    mst_a: String,
    mst_b: String,
}

#[derive(Debug, Deserialize)]
struct Results {
    record_ops: Vec<RecordOp>,
    inductive_proof_nodes: Vec<String>,
}

#[derive(Debug, Deserialize)]
struct RecordOp {
    rpath: String,
    old_value: Option<String>,
    new_value: Option<String>,
}

impl RecordOp {
    /// The op that inverts this change, taking tree *b* back towards tree *a*.
    fn to_op(&self) -> Result<Op, String> {
        let key = Key::new(&self.rpath).map_err(|e| format!("bad rpath: {e}"))?;
        let old = self.old_value.as_deref().map(parse_cid).transpose()?;
        let new = self.new_value.as_deref().map(parse_cid).transpose()?;
        match (old, new) {
            (None, Some(cid)) => Ok(Op::Create { key, cid }),
            (Some(prev), Some(cid)) => Ok(Op::Update { key, cid, prev }),
            (Some(prev), None) => Ok(Op::Delete { key, prev }),
            (None, None) => Err(format!("record op on {} changes nothing", self.rpath)),
        }
    }
}

fn parse_cid(s: &str) -> Result<Cid, String> {
    s.parse().map_err(|e| format!("bad cid {s:?}: {e}"))
}

/// Every block in a CAR, plus its root.
fn read_car(path: &Path) -> Result<(Cid, HashMap<Cid, Vec<u8>>), String> {
    let bytes = std::fs::read(path).map_err(|e| format!("reading {}: {e}", path.display()))?;
    let reader = CarReader::new(&bytes).map_err(|e| format!("{}: {e}", path.display()))?;
    let root = reader.root();
    let blocks = reader
        .map(|block| {
            block
                .map(|(cid, bytes)| (cid, bytes.to_vec()))
                .map_err(|e| format!("{}: {e}", path.display()))
        })
        .collect::<Result<_, _>>()?;
    Ok((root, blocks))
}

fn run_case(case_path: &Path, suite_root: &Path) -> Result<(), String> {
    let json = std::fs::read_to_string(case_path)
        .map_err(|e| format!("reading {}: {e}", case_path.display()))?;
    let case: TestCase = serde_json::from_str(&json).map_err(|e| format!("parsing case: {e}"))?;
    if case.kind != "mst-diff" {
        return Ok(()); // some other kind of case, not ours to run
    }

    let (root_a, blocks_a) = read_car(&resolve(suite_root, &case.inputs.mst_a))?;
    let (root_b, blocks_b) = read_car(&resolve(suite_root, &case.inputs.mst_b))?;

    let ops: Result<Vec<Op>, String> = case
        .results
        .record_ops
        .iter()
        .map(RecordOp::to_op)
        .collect();
    // deliberately left in the suite's own order
    let ops = ops?;

    // correctness, with the whole of tree b available: any op order must land
    // on tree a's root
    let mut all = blocks_b.clone();
    all.extend(blocks_a);
    let got = invert_ops(&all, root_b, &ops, OpOrder::AsGiven).map_err(|e| {
        format!(
            "{}: inverting {} ops failed: {e}",
            case.description,
            ops.len()
        )
    })?;
    if got != root_a {
        return Err(format!(
            "{}: inverted to {got}, expected {root_a}",
            case.description
        ));
    }

    // minimality, with nothing but the blocks the suite says inversion needs.
    // that set is narrower than sync §4.1.2 obliges producers to send, and it
    // is computed for highest-key-first application, so this only holds for
    // that order — which makes it a tripwire for reaching past what we need.
    let allowed: Result<Vec<Cid>, String> = case
        .results
        .inductive_proof_nodes
        .iter()
        .map(|s| parse_cid(s))
        .collect();
    let proof: HashMap<Cid, Vec<u8>> = allowed?
        .into_iter()
        .filter_map(|cid| all.get(&cid).map(|block| (cid, block.clone())))
        .collect();

    let got = invert_ops(&proof, root_b, &ops, OpOrder::HighestKeyFirst).map_err(|e| {
        format!(
            "{}: inverting {} ops from inductive_proof_nodes failed: {e}",
            case.description,
            ops.len()
        )
    })?;

    if got != root_a {
        return Err(format!(
            "{}: inverted to {got}, expected {root_a}",
            case.description
        ));
    }
    Ok(())
}

/// `MST_SUITE_FAILURES=all` to see every failure rather than a sample.
fn failure_limit() -> usize {
    match std::env::var("MST_SUITE_FAILURES").as_deref() {
        Ok("all") => usize::MAX,
        _ => 20,
    }
}

fn resolve(suite_root: &Path, relative: &str) -> PathBuf {
    suite_root.join(relative.trim_start_matches("./"))
}

fn case_files(dir: &Path) -> Vec<PathBuf> {
    let mut found = Vec::new();
    let Ok(entries) = std::fs::read_dir(dir) else {
        return found;
    };
    for entry in entries.flatten() {
        let path = entry.path();
        if path.is_dir() {
            found.extend(case_files(&path));
        } else if path.extension().is_some_and(|e| e == "json") {
            found.push(path);
        }
    }
    found.sort();
    found
}

/// Run every case under `suite_root/tests`, reporting all failures at once.
fn run_suite(suite_root: &Path) {
    let cases = case_files(&suite_root.join("tests"));
    assert!(
        !cases.is_empty(),
        "no test cases under {}",
        suite_root.join("tests").display()
    );

    let failures: Vec<String> = cases
        .iter()
        .filter_map(|case| run_case(case, suite_root).err())
        .collect();

    assert!(
        failures.is_empty(),
        "{} of {} cases failed:\n{}",
        failures.len(),
        cases.len(),
        failures
            .iter()
            .take(failure_limit())
            .map(String::as_str)
            .collect::<Vec<_>>()
            .join("\n")
    );
    eprintln!("{} mst-diff cases passed", cases.len());
}

#[test]
fn vendored_subset() {
    run_suite(&PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/mst-test-suite"));
}

#[test]
#[ignore = "needs the mst-test-suite submodule: git submodule update --init"]
fn exhaustive() {
    let root = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("tests/mst-test-suite");
    assert!(
        root.join("tests").is_dir(),
        "{} is empty — run `git submodule update --init {}`",
        root.display(),
        root.strip_prefix(env!("CARGO_MANIFEST_DIR"))
            .unwrap_or(&root)
            .display()
    );
    run_suite(&root);
}