fv-compute 0.2.0

//! Registry + directory-per-transform loader tests. Covers both goals:
//! discovery from one directory (define-once) and pluggable roots/backends + precedence
//! (customizable). Hermetic (MemoryRoot / tests-fixtures + a mock backend), plus one integration
//! test that the real provided `transforms/` package validates.

use arrow::array::{Int64Array, RecordBatch};
use arrow::datatypes::{DataType, Field, Schema};
use fv_compute::contract::{Compute, ComputeError, TransformBackend};
use fv_compute::{Backends, DirRoot, ImplKind, MemoryRoot, Registry, RegistryError, TransformManifest};
use std::path::{Path, PathBuf};
use std::sync::Arc;

fn toml_for(id: &str, version: &str) -> String {
    format!(
        r#"
        id = "{id}"
        version = "{version}"
        impl = "builtin"
        ref = "{id}"
        [[inputs]]
        columns = [{{ name = "x", type = "int64" }}]
        [output]
        columns = [{{ name = "x", type = "int64" }}]
        "#
    )
}

fn fixtures_dir() -> PathBuf {
    PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/transforms")
}

// ---- discovery from a directory (define-once) -------------------------------

#[test]
fn dirroot_discovers_units_and_skips_non_transform_dirs() {
    let reg = Registry::builder()
        .root(DirRoot::new("fixtures", fixtures_dir()))
        .build()
        .unwrap();
    assert_eq!(reg.len(), 2, "alpha + beta, not the manifest-less dir");
    assert!(reg.get("alpha", "1.0.0").is_some());
    assert!(reg.get("beta", "0.2.0").is_some());
    // the directory is carried through for the backend to load its impl from
    assert!(reg.get("beta", "0.2.0").unwrap().dir.ends_with("beta"));
}

#[test]
fn missing_root_path_yields_no_units_not_an_error() {
    let reg = Registry::builder()
        .root(DirRoot::new("ghost", "/no/such/path/transforms"))
        .build()
        .unwrap();
    assert!(reg.is_empty());
}

// ---- pluggable roots + precedence (customizable) ----------------------------

#[test]
fn later_root_overrides_earlier_for_same_id_at_version() {
    let provided = MemoryRoot::new("provided").unit("/p/foo", toml_for("foo", "1.0.0"));
    let business = MemoryRoot::new("meridian").unit("/b/foo", toml_for("foo", "1.0.0"));
    let reg = Registry::builder().root(provided).root(business).build().unwrap();
    let foo = reg.get("foo", "1.0.0").unwrap();
    assert_eq!(foo.root, "meridian", "business root wins on precedence");
    assert_eq!(foo.dir, PathBuf::from("/b/foo"));
    assert_eq!(reg.len(), 1);
}

#[test]
fn different_versions_coexist_and_latest_wins() {
    let root = MemoryRoot::new("provided")
        .unit("/p/foo-1", toml_for("foo", "1.0.0"))
        .unit("/p/foo-2", toml_for("foo", "2.3.0"))
        .unit("/p/foo-1b", toml_for("foo", "1.5.0"));
    let reg = Registry::builder().root(root).build().unwrap();
    assert_eq!(reg.len(), 3);
    assert_eq!(reg.latest("foo").unwrap().manifest.version, "2.3.0");
    // resolve: bare id -> latest; id@version -> exact
    assert_eq!(reg.resolve("foo").unwrap().manifest.version, "2.3.0");
    assert_eq!(reg.resolve("foo@1.5.0").unwrap().manifest.version, "1.5.0");
    assert!(reg.resolve("foo@9.9.9").is_none());
}

#[test]
fn duplicate_within_a_single_root_is_an_error() {
    let root = MemoryRoot::new("provided")
        .unit("/p/a", toml_for("dup", "1.0.0"))
        .unit("/p/b", toml_for("dup", "1.0.0"));
    let err = Registry::builder().root(root).build().unwrap_err();
    assert!(matches!(err, RegistryError::DuplicateInRoot { .. }), "got {err:?}");
}

#[test]
fn invalid_manifest_fails_the_build_with_context() {
    let root = MemoryRoot::new("provided").unit(
        "/p/bad",
        r#"id = "x"
           version = "not-semver"
           impl = "builtin"
           [output]
           columns = [{ name = "y", type = "int64" }]"#,
    );
    let err = Registry::builder().root(root).build().unwrap_err();
    match err {
        RegistryError::Manifest { root, .. } => assert_eq!(root, "provided"),
        other => panic!("expected Manifest error, got {other:?}"),
    }
}

// ---- derived index (no re-scan at startup) ----------------------------------

#[test]
fn index_roundtrips_and_resolves_identically() {
    let reg = Registry::builder()
        .root(DirRoot::new("fixtures", fixtures_dir()))
        .build()
        .unwrap();
    let index = reg.to_index();
    let json = serde_json::to_string(&index).unwrap();
    let back: fv_compute::RegistryIndex = serde_json::from_str(&json).unwrap();
    let reg2 = Registry::from_index(back).unwrap();
    assert_eq!(reg2.len(), reg.len());
    assert!(reg2.get("alpha", "1.0.0").is_some());
    assert_eq!(
        reg2.get("beta", "0.2.0").unwrap().manifest.id,
        reg.get("beta", "0.2.0").unwrap().manifest.id
    );
}

#[test]
fn index_from_wrong_contract_version_is_rejected() {
    let mut index = Registry::builder()
        .root(MemoryRoot::new("p").unit("/p/foo", toml_for("foo", "1.0.0")))
        .build()
        .unwrap()
        .to_index();
    index.contract_version = "999.0.0".into();
    assert!(matches!(
        Registry::from_index(index),
        Err(RegistryError::IndexVersion { .. })
    ));
}

// ---- pluggable backends + load() wiring (customizable) ----------------------

/// A mock backend proving the seam: any caller registers backends per impl kind. Slices 3–4 land
/// the real wasm/container/expression/builtin backends behind this exact trait.
struct MockBuiltin;
struct IdentityCompute {
    manifest: TransformManifest,
}
impl Compute for IdentityCompute {
    fn manifest(&self) -> &TransformManifest {
        &self.manifest
    }
    fn run(&self, inputs: &[RecordBatch]) -> Result<RecordBatch, ComputeError> {
        inputs.first().cloned().ok_or(ComputeError::Run {
            id: self.manifest.id.clone(),
            msg: "no input".into(),
        })
    }
}
impl TransformBackend for MockBuiltin {
    fn kind(&self) -> ImplKind {
        ImplKind::Builtin
    }
    fn load(&self, manifest: &TransformManifest, _root: &Path) -> Result<Box<dyn Compute>, ComputeError> {
        Ok(Box::new(IdentityCompute {
            manifest: manifest.clone(),
        }))
    }
}

#[test]
fn load_dispatches_to_registered_backend() {
    let reg = Registry::builder()
        .root(MemoryRoot::new("p").unit("/p/alpha", toml_for("alpha", "1.0.0")))
        .build()
        .unwrap();
    let backends = Backends::new().register(MockBuiltin);

    let compute = reg.load("alpha", &backends).unwrap();
    assert_eq!(compute.manifest().id, "alpha");

    // it actually runs through the seam
    let schema = Arc::new(Schema::new(vec![Field::new("x", DataType::Int64, false)]));
    let batch = RecordBatch::try_new(schema, vec![Arc::new(Int64Array::from(vec![1, 2, 3]))]).unwrap();
    let out = compute.run(std::slice::from_ref(&batch)).unwrap();
    assert_eq!(out, batch);
}

#[test]
fn load_without_a_backend_reports_no_backend() {
    let reg = Registry::builder()
        .root(MemoryRoot::new("p").unit("/p/alpha", toml_for("alpha", "1.0.0")))
        .build()
        .unwrap();
    let backends = Backends::new(); // none registered
    let err = reg
        .load("alpha", &backends)
        .err()
        .expect("should fail without a backend");
    match err {
        ComputeError::NoBackend { kind } => assert_eq!(kind, ImplKind::Builtin),
        other => panic!("expected NoBackend, got {other:?}"),
    }
}