faucet_cli/commands/
validate.rs

1//! `faucet validate` — parse + expand a pipeline config without running.
2//!
3//! Surfaces every per-row error with the row id, so a config with multiple
4//! issues reports them together instead of failing at the first one.
5
6use crate::cli::ValidateArgs;
7use crate::config::PipelineConfig;
8use crate::error::{CliError, CliResult};
9use crate::expand::{NodeRole, expand};
10use crate::registry::{sink_schema, source_schema};
11use crate::state::available_state_kinds;
12use crate::transforms::available_transforms;
13
14/// Execute the `validate` subcommand.
15pub async fn run(args: ValidateArgs) -> CliResult<()> {
16    let cwd = std::env::current_dir()?;
17    let env_path =
18        crate::env_loader::resolve_env_file(args.env_file.as_deref(), args.no_env_file, &cwd)?;
19    crate::env_loader::load_env_file_if_present(env_path.as_deref())?;
20
21    let path = match args.config {
22        Some(p) => p,
23        None => crate::env_loader::discover_config_path(&cwd).ok_or(CliError::NoConfigOrFromEnv)?,
24    };
25
26    if args.show_composed {
27        let composed = crate::compose::compose(&path, args.profile.as_deref())?;
28        // Normalize to exactly one trailing newline: the YAML serializer appends
29        // one but `serde_json::to_string_pretty` (JSON-format configs) does not,
30        // and the fast path echoes the file verbatim. A single `\n` keeps
31        // `faucet validate … --show-composed > out.{yaml,json}` well-formed.
32        println!("{}", composed.trim_end_matches('\n'));
33        return Ok(());
34    }
35
36    let cfg = if args.no_secrets {
37        // Grammar / structure only — never touch the network.
38        PipelineConfig::from_path_tolerating_secrets(&path, args.profile.as_deref())?
39    } else {
40        // Real preflight: report each secret reference, then resolve.
41        let refs = crate::secrets::scan_path_refs(&path, args.profile.as_deref())?;
42        let cfg = PipelineConfig::from_path_async(&path, args.profile.as_deref()).await?;
43        for (scheme, reference) in &refs {
44            println!("secret: {scheme}:{reference} → resolved");
45        }
46        cfg
47    };
48    let nodes = expand(&cfg)?;
49
50    // Validate the schedule block (cron / timezone / bounds) so `faucet validate`
51    // catches schedule misconfiguration in CI without running. Offline-safe.
52    #[cfg(feature = "schedule")]
53    if let Some(spec) = &cfg.schedule {
54        crate::schedule::compiled::CompiledSchedule::compile(spec)?;
55        println!(
56            "schedule: cron '{}' tz '{}' — valid",
57            spec.cron, spec.timezone
58        );
59    }
60
61    // Lineage transport reachability — best-effort. A failure here is only a
62    // warning: lineage emission never blocks a pipeline run.
63    #[cfg(feature = "lineage")]
64    if let Some(lc) = cfg.lineage.as_ref() {
65        match crate::lineage_glue::check_transport(lc).await {
66            Ok(msg) => println!("lineage: {msg}"),
67            Err(msg) => println!("lineage: WARNING — {msg} (lineage never blocks a run)"),
68        }
69    }
70
71    for node in &nodes {
72        // Verifying the schema lookup also catches unknown connector kinds.
73        source_schema(&node.source.kind)?;
74        sink_schema(&node.sink.kind)?;
75        for t in &node.transforms {
76            if !available_transforms().contains(&t.kind.as_str()) {
77                return Err(CliError::UnknownTransform {
78                    name: format!("{} (row '{}')", t.kind, node.id),
79                    available: available_transforms().join(", "),
80                });
81            }
82        }
83        if let Some(state) = &node.state
84            && !available_state_kinds().contains(&state.kind.as_str())
85        {
86            return Err(CliError::UnknownStateStore {
87                name: format!("{} (row '{}')", state.kind, node.id),
88                available: available_state_kinds().join(", "),
89            });
90        }
91    }
92
93    let roots = nodes
94        .iter()
95        .filter(|n| matches!(n.role, NodeRole::Root))
96        .count();
97    let children = nodes.len() - roots;
98    println!(
99        "ok: '{}' rows={} (roots={}, children={}) execution={}",
100        cfg.name.as_deref().unwrap_or("(unnamed)"),
101        nodes.len(),
102        roots,
103        children,
104        cfg.execution
105            .as_ref()
106            .map(|e| format!(
107                "max_concurrent={:?} on_error={:?}",
108                e.max_concurrent.unwrap_or(0),
109                e.on_error
110            ))
111            .unwrap_or_else(|| "(defaults)".to_owned()),
112    );
113    for node in &nodes {
114        let role = match &node.role {
115            NodeRole::Root => "root".to_owned(),
116            NodeRole::Child {
117                parent_id,
118                parent_key,
119            } => {
120                format!("child of '{parent_id}' (parent_key={parent_key})")
121            }
122        };
123        println!(
124            "  - {} [{}] source={} sink={}",
125            node.id, role, node.source.kind, node.sink.kind
126        );
127    }
128    Ok(())
129}
faucet_cli/commands/validate.rs

faucet_cli/commands/
validate.rs
