fv_compute/

registry.rs

//! The registry + directory-per-transform loader.
//!
//! Two goals drive this module:
//!  1. **Define a transform in as few places as possible** — one directory (`transform.toml` +
//!     impl + co-located tests) is the whole definition. The registry discovers and validates it;
//!     nothing else declares it.
//!  2. **Registry as customizable as possible** — [`Root`] is a trait (local dir, in-memory, and
//!     later git/ or a caller's own source all plug in), roots are ordered with configurable
//!     precedence, and backends are a caller-extensible [`Backends`] map. Any service embeds its
//!     own [`Registry`] with its own roots + backends.
//!
//! Discovery + validation + resolution + a serializable [`RegistryIndex`] (so startup
//! doesn't re-scan). Backend *execution* is the backends' job; [`Registry::load`] wires the seam and is
//! testable today with any [`TransformBackend`].

use crate::contract::{Compute, ComputeError, TransformBackend};
use crate::manifest::{parse_and_validate, ImplKind, ManifestError, TransformManifest};
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;
use std::collections::HashMap;
use std::path::{Path, PathBuf};

/// A discovered-but-not-yet-validated unit: where it came from + its raw `transform.toml`.
pub struct RawUnit {
    pub root: String,
    pub dir: PathBuf,
    pub toml: String,
}

/// A source of transform directories. Implement this to add a new kind of root — a git repo
/// an HTTP endpoint, an in-browser buffer, or anything else — without touching the
/// registry core.
pub trait Root: Send + Sync {
    /// A stable name for precedence/diagnostics (e.g. "provided", "meridian").
    fn name(&self) -> &str;
    /// Enumerate the transform directories this root provides.
    fn discover(&self) -> Result<Vec<RawUnit>, RegistryError>;
}

/// A local-directory root: scans `<path>/<name>/transform.toml`. A non-existent path yields no
/// units (so optional business roots are friendly); a strict caller checks existence itself.
pub struct DirRoot {
    name: String,
    path: PathBuf,
}

impl DirRoot {
    pub fn new(name: impl Into<String>, path: impl Into<PathBuf>) -> Self {
        Self {
            name: name.into(),
            path: path.into(),
        }
    }
}

impl Root for DirRoot {
    fn name(&self) -> &str {
        &self.name
    }

    fn discover(&self) -> Result<Vec<RawUnit>, RegistryError> {
        if !self.path.exists() {
            return Ok(Vec::new());
        }
        let mut out = Vec::new();
        let mut entries: Vec<PathBuf> = std::fs::read_dir(&self.path)
            .map_err(|e| RegistryError::Io {
                path: self.path.clone(),
                msg: e.to_string(),
            })?
            .filter_map(|e| e.ok().map(|e| e.path()))
            .filter(|p| p.is_dir())
            .collect();
        entries.sort(); // deterministic discovery order
        for dir in entries {
            let manifest_path = dir.join("transform.toml");
            if !manifest_path.exists() {
                continue; // a directory without a manifest is not a transform unit
            }
            let toml = std::fs::read_to_string(&manifest_path).map_err(|e| RegistryError::Io {
                path: manifest_path.clone(),
                msg: e.to_string(),
            })?;
            out.push(RawUnit {
                root: self.name.clone(),
                dir,
                toml,
            });
        }
        Ok(out)
    }
}

/// An in-memory root — for tests, generated builtins, or the in-browser IDE.
pub struct MemoryRoot {
    name: String,
    units: Vec<(PathBuf, String)>,
}

impl MemoryRoot {
    pub fn new(name: impl Into<String>) -> Self {
        Self {
            name: name.into(),
            units: Vec::new(),
        }
    }
    /// Add a `(dir, transform.toml)` unit.
    pub fn unit(mut self, dir: impl Into<PathBuf>, toml: impl Into<String>) -> Self {
        self.units.push((dir.into(), toml.into()));
        self
    }
}

impl Root for MemoryRoot {
    fn name(&self) -> &str {
        &self.name
    }
    fn discover(&self) -> Result<Vec<RawUnit>, RegistryError> {
        Ok(self
            .units
            .iter()
            .map(|(dir, toml)| RawUnit {
                root: self.name.clone(),
                dir: dir.clone(),
                toml: toml.clone(),
            })
            .collect())
    }
}

/// A validated, resolvable transform unit.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RegisteredTransform {
    pub manifest: TransformManifest,
    pub dir: PathBuf,
    /// Which root supplied it (after precedence resolution).
    pub root: String,
}

impl RegisteredTransform {
    /// `id@version` — the registry key.
    pub fn key(&self) -> String {
        format!("{}@{}", self.manifest.id, self.manifest.version)
    }
}

#[derive(Debug, thiserror::Error)]
pub enum RegistryError {
    #[error("I/O error at {path}: {msg}")]
    Io { path: PathBuf, msg: String },
    #[error("invalid transform in root `{root}` at {dir}: {source}")]
    Manifest {
        root: String,
        dir: PathBuf,
        source: Box<ManifestError>,
    },
    #[error("duplicate transform `{key}` within root `{root}` ({dir})")]
    DuplicateInRoot { key: String, root: String, dir: PathBuf },
    #[error("registry index contract version `{found}` != `{expected}`")]
    IndexVersion { found: String, expected: String },
}

/// Build a registry from an ordered list of roots. Later roots OVERRIDE earlier ones for the same
/// `id@version` (a business's `transforms/` shadows the provided package) — the configurable
/// precedence. A duplicate `id@version` *within a single root* is an error.
#[derive(Default)]
pub struct RegistryBuilder {
    roots: Vec<Box<dyn Root>>,
}

impl RegistryBuilder {
    pub fn new() -> Self {
        Self::default()
    }

    /// Add a root. Order matters: later = higher precedence.
    pub fn root(mut self, r: impl Root + 'static) -> Self {
        self.roots.push(Box::new(r));
        self
    }

    pub fn build(self) -> Result<Registry, RegistryError> {
        let mut by_key: BTreeMap<String, RegisteredTransform> = BTreeMap::new();
        for root in &self.roots {
            let mut seen_in_root: BTreeMap<String, PathBuf> = BTreeMap::new();
            for raw in root.discover()? {
                let manifest = parse_and_validate(&raw.toml).map_err(|source| RegistryError::Manifest {
                    root: raw.root.clone(),
                    dir: raw.dir.clone(),
                    source: Box::new(source),
                })?;
                let reg = RegisteredTransform {
                    manifest,
                    dir: raw.dir.clone(),
                    root: raw.root.clone(),
                };
                let key = reg.key();
                if let Some(prev) = seen_in_root.insert(key.clone(), raw.dir.clone()) {
                    let _ = prev;
                    return Err(RegistryError::DuplicateInRoot {
                        key,
                        root: raw.root,
                        dir: raw.dir,
                    });
                }
                by_key.insert(key, reg); // later root overrides earlier
            }
        }
        Ok(Registry { by_key })
    }
}

/// The resolved set of transforms, addressable by `id@version`.
#[derive(Debug, Clone)]
pub struct Registry {
    by_key: BTreeMap<String, RegisteredTransform>,
}

impl Registry {
    pub fn builder() -> RegistryBuilder {
        RegistryBuilder::new()
    }

    /// Exact `id@version` lookup.
    pub fn get(&self, id: &str, version: &str) -> Option<&RegisteredTransform> {
        self.by_key.get(&format!("{id}@{version}"))
    }

    /// Highest semver for an id.
    pub fn latest(&self, id: &str) -> Option<&RegisteredTransform> {
        self.by_key.values().filter(|r| r.manifest.id == id).max_by(|a, b| {
            let va = semver::Version::parse(&a.manifest.version).ok();
            let vb = semver::Version::parse(&b.manifest.version).ok();
            va.cmp(&vb)
        })
    }

    /// Resolve a selector: `"id"` → latest; `"id@version"` → exact.
    pub fn resolve(&self, selector: &str) -> Option<&RegisteredTransform> {
        match selector.split_once('@') {
            Some((id, version)) => self.get(id, version),
            None => self.latest(selector),
        }
    }

    pub fn iter(&self) -> impl Iterator<Item = &RegisteredTransform> {
        self.by_key.values()
    }

    pub fn len(&self) -> usize {
        self.by_key.len()
    }

    pub fn is_empty(&self) -> bool {
        self.by_key.is_empty()
    }

    /// The derived, serializable index — write it at build/publish time so runtime
    /// startup loads this instead of re-scanning + re-validating the tree.
    pub fn to_index(&self) -> RegistryIndex {
        RegistryIndex {
            contract_version: crate::CONTRACT_VERSION.to_string(),
            transforms: self.by_key.values().cloned().collect(),
        }
    }

    /// Reconstruct from a derived index (no directory scan). Fails if the index was produced
    /// against a different contract version.
    pub fn from_index(index: RegistryIndex) -> Result<Self, RegistryError> {
        if index.contract_version != crate::CONTRACT_VERSION {
            return Err(RegistryError::IndexVersion {
                found: index.contract_version,
                expected: crate::CONTRACT_VERSION.to_string(),
            });
        }
        let by_key = index.transforms.into_iter().map(|r| (r.key(), r)).collect();
        Ok(Registry { by_key })
    }

    /// Resolve a selector and load it via the matching backend — "run transform X@v" as a library
    /// call. Backend execution is the backend's job; this is the wiring + dispatch.
    pub fn load(&self, selector: &str, backends: &Backends) -> Result<Box<dyn Compute>, ComputeError> {
        let reg = self.resolve(selector).ok_or_else(|| ComputeError::Load {
            id: selector.to_string(),
            msg: "no such transform in registry".into(),
        })?;
        let backend = backends.get(reg.manifest.impl_kind).ok_or(ComputeError::NoBackend {
            kind: reg.manifest.impl_kind,
        })?;
        backend.load(&reg.manifest, &reg.dir)
    }
}

/// The serializable, re-scan-free index.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RegistryIndex {
    #[serde(rename = "contractVersion")]
    pub contract_version: String,
    pub transforms: Vec<RegisteredTransform>,
}

/// The caller-extensible backend set: `ImplKind → TransformBackend`. A deployment registers its
/// own backends (even for a new impl kind) without touching the registry core.
#[derive(Default)]
pub struct Backends {
    map: HashMap<ImplKind, Box<dyn TransformBackend>>,
}

impl Backends {
    pub fn new() -> Self {
        Self::default()
    }

    pub fn register(mut self, backend: impl TransformBackend + 'static) -> Self {
        self.map.insert(backend.kind(), Box::new(backend));
        self
    }

    pub fn get(&self, kind: ImplKind) -> Option<&dyn TransformBackend> {
        self.map.get(&kind).map(|b| b.as_ref())
    }

    pub fn kinds(&self) -> impl Iterator<Item = &ImplKind> {
        self.map.keys()
    }
}

/// Convenience: build a registry from the standard three-root layout — the provided package
/// (`<repo>/transforms`), then each business bundle dir (`<bundle>/transforms`), business winning
/// on precedence. Missing dirs are simply skipped.
pub fn standard_registry(provided: impl AsRef<Path>, business_bundles: &[PathBuf]) -> Result<Registry, RegistryError> {
    let mut b = Registry::builder().root(DirRoot::new("provided", provided.as_ref().to_path_buf()));
    for bundle in business_bundles {
        let name = bundle
            .file_name()
            .map(|s| s.to_string_lossy().to_string())
            .unwrap_or_else(|| "business".into());
        b = b.root(DirRoot::new(name, bundle.join("transforms")));
    }
    b.build()
}