fv-compute 0.2.0

//! The `transform.toml` manifest — the single self-describing unit the registry
//! discovers. This crate defines + validates it (the publish gate); the registry discovers directories
//! of them. Every IMPL (builtin/expression/wasm/container/llm) uses this one shape.

use crate::capability::CapabilityEnvelope;
use crate::types::SchemaSpec;
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;

/// The IMPL axis: which backend runs the transform.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ImplKind {
    /// A native primitive (fv-value / a structural step-kind); `ref` points at it. NOT re-authored
    /// as a slow plugin — the manifest + golden vectors are co-located, the code stays native.
    Builtin,
    /// A value-dialect payload (`entry` → `expr.fvx`).
    Expression,
    /// A WebAssembly component (`entry` → `main.wasm`) — the flagship polyglot backend.
    Wasm,
    /// A container image / process speaking the stdio protocol (`entry` → image ref/Dockerfile).
    Container,
    /// An ONNX model artifact (`entry` → `model.onnx`), run by the native `fv-infer` backend
    /// A model is just a transform: inputs = feature columns, output = the
    /// prediction column(s). Pure/deterministic/no-io by default, so it is a legal derived/stream
    /// compute — the "model as a pipeline node" shape, pluggable via this same registry.
    Onnx,
    /// An LLM-backed compute (nondeterministic/effectful — Action bindings only).
    Llm,
}

/// The wire protocol a `container` transform speaks over stdio (ignored by other impls).
/// `arrow-ipc` is the fast Arrow handoff; `json` is the stdlib-friendly row-JSON contract
/// (no Arrow dependency in the transform — works under the stripped-env isolation).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
#[serde(rename_all = "kebab-case")]
pub enum ContainerProtocol {
    #[default]
    ArrowIpc,
    Json,
}

/// How output-dataset labels are derived (label propagation, fail-closed).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
#[serde(rename_all = "snake_case")]
pub enum LabelsRule {
    /// Output inherits the union of the inputs' labels regardless of what the transform does.
    #[default]
    Propagate,
}

#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct Labels {
    #[serde(default)]
    pub rule: LabelsRule,
}

/// A parsed, not-yet-validated `transform.toml`. Call [`TransformManifest::validate`] (or
/// [`parse_and_validate`]) before trusting it — an unvalidated manifest never executes.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TransformManifest {
    /// Stable identifier (camelCase), unique within a registry root.
    pub id: String,
    /// Semver — enables `id@version` selection + hot-swap.
    pub version: String,
    #[serde(rename = "impl")]
    pub impl_kind: ImplKind,
    /// For `builtin`: the native impl this points at (e.g. "haversineKm").
    #[serde(default, rename = "ref")]
    pub reference: Option<String>,
    /// For `wasm`/`container`/`expression`: the built artifact / payload path, relative to the dir.
    #[serde(default)]
    pub entry: Option<String>,
    /// Input signature(s). Empty for a pure generator; >1 for a JOIN.
    #[serde(default)]
    pub inputs: Vec<SchemaSpec>,
    /// The single output signature.
    pub output: SchemaSpec,
    #[serde(default)]
    pub capabilities: CapabilityEnvelope,
    /// output column -> source column(s). Bare `col` for single-input; qualified `input.col` when
    /// there is >1 input. Undeclared = opaque (dataset-level edge), like a raw sql step.
    #[serde(default, rename = "columnLineage")]
    pub column_lineage: BTreeMap<String, Vec<String>>,
    #[serde(default)]
    pub labels: Labels,
    /// For `impl = "container"`: the stdio wire protocol. Default `arrow-ipc`.
    #[serde(default)]
    pub protocol: ContainerProtocol,
}

#[derive(Debug, thiserror::Error)]
pub enum ManifestError {
    // Boxed: toml::de::Error is large, and this is a cold error path — keeps the Ok path small
    // (clippy::result_large_err).
    #[error("TOML parse error: {0}")]
    Toml(Box<toml::de::Error>),
    #[error("invalid manifest `{id}`: {msg}")]
    Invalid { id: String, msg: String },
}

impl TransformManifest {
    /// Parse from a `transform.toml` string (does NOT validate).
    pub fn from_toml_str(s: &str) -> Result<Self, ManifestError> {
        toml::from_str(s).map_err(|e| ManifestError::Toml(Box::new(e)))
    }

    fn invalid(&self, msg: impl Into<String>) -> ManifestError {
        ManifestError::Invalid {
            id: self.id.clone(),
            msg: msg.into(),
        }
    }

    /// The publish gate: reject anything malformed before it can execute.
    pub fn validate(&self) -> Result<(), ManifestError> {
        // id
        if self.id.is_empty() || !is_ident(&self.id) {
            return Err(self.invalid("id must be a non-empty identifier (alnum, starting with a letter)"));
        }
        // version
        if semver::Version::parse(&self.version).is_err() {
            return Err(self.invalid(format!("version `{}` is not valid semver", self.version)));
        }
        // impl-specific artifact presence
        match self.impl_kind {
            ImplKind::Builtin => {
                if self.entry.is_some() {
                    return Err(self.invalid("builtin must not set `entry` (it points at native code via `ref`)"));
                }
                // `ref` defaults to `id` if omitted.
            }
            ImplKind::Expression | ImplKind::Wasm | ImplKind::Container | ImplKind::Onnx => {
                if self.entry.as_deref().unwrap_or("").is_empty() {
                    return Err(self.invalid("this impl requires a non-empty `entry` (the artifact/payload path)"));
                }
                if self.reference.is_some() {
                    return Err(self.invalid("`ref` is only for builtin impls"));
                }
            }
            ImplKind::Llm => { /* entry/config are impl-defined; nothing structural to enforce yet */ }
        }
        // output must declare at least one column
        if self.output.columns.is_empty() {
            return Err(self.invalid("output must declare at least one column"));
        }
        // multi-input datasets must be named + distinct (so qualified lineage resolves)
        if self.inputs.len() > 1 {
            let mut seen = std::collections::HashSet::new();
            for inp in &self.inputs {
                match &inp.name {
                    None => return Err(self.invalid("every input must be named when there is >1 input")),
                    Some(n) if !seen.insert(n.clone()) => {
                        return Err(self.invalid(format!("duplicate input name `{n}`")));
                    }
                    _ => {}
                }
            }
        }
        // column lineage integrity
        self.validate_lineage()?;
        Ok(())
    }

    fn validate_lineage(&self) -> Result<(), ManifestError> {
        let multi = self.inputs.len() > 1;
        for (out_col, sources) in &self.column_lineage {
            if !self.output.has_column(out_col) {
                return Err(self.invalid(format!("columnLineage references unknown output column `{out_col}`")));
            }
            if sources.is_empty() {
                return Err(self.invalid(format!("columnLineage for `{out_col}` has no sources")));
            }
            for src in sources {
                if multi {
                    // qualified `input.col` — the input must exist and carry the column
                    let (inp, col) = src.split_once('.').ok_or_else(|| {
                        self.invalid(format!(
                            "multi-input lineage source `{src}` must be qualified as `input.column`"
                        ))
                    })?;
                    let found = self
                        .inputs
                        .iter()
                        .find(|i| i.name.as_deref() == Some(inp))
                        .ok_or_else(|| self.invalid(format!("lineage source references unknown input `{inp}`")))?;
                    if !found.has_column(col) {
                        return Err(
                            self.invalid(format!("input `{inp}` has no column `{col}` (lineage for `{out_col}`)"))
                        );
                    }
                } else {
                    // single input: bare column name must exist in the (sole) input, if declared
                    if let Some(inp) = self.inputs.first() {
                        if !inp.has_column(src) {
                            return Err(self.invalid(format!(
                                "lineage source `{src}` is not a column of the input (for `{out_col}`)"
                            )));
                        }
                    }
                }
            }
        }
        Ok(())
    }

    /// The `ref` a builtin resolves to (defaults to `id`).
    pub fn builtin_ref(&self) -> &str {
        self.reference.as_deref().unwrap_or(&self.id)
    }
}

/// Parse + validate in one step — the normal entry point.
pub fn parse_and_validate(s: &str) -> Result<TransformManifest, ManifestError> {
    let m = TransformManifest::from_toml_str(s)?;
    m.validate()?;
    Ok(m)
}

/// An identifier: starts with an ASCII letter, then alnum. (camelCase ids like `haversineKm`.)
fn is_ident(s: &str) -> bool {
    let mut chars = s.chars();
    match chars.next() {
        Some(c) if c.is_ascii_alphabetic() => {}
        _ => return false,
    }
    s.chars().all(|c| c.is_ascii_alphanumeric())
}

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

    #[test]
    fn is_ident_rules() {
        assert!(is_ident("haversineKm"));
        assert!(is_ident("rename"));
        assert!(!is_ident("2cool"));
        assert!(!is_ident("has space"));
        assert!(!is_ident("snake_case"));
        assert!(!is_ident(""));
    }

    #[test]
    fn builtin_ref_defaults_to_id() {
        let m = parse_and_validate(
            r#"
            id = "haversineKm"
            version = "0.1.0"
            impl = "builtin"
            [output]
            columns = [{ name = "km", type = "float64" }]
            "#,
        )
        .unwrap();
        assert_eq!(m.builtin_ref(), "haversineKm");
    }
}