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    let cfg = if args.no_secrets {
26        // Grammar / structure only — never touch the network.
27        PipelineConfig::from_path_tolerating_secrets(&path)?
28    } else {
29        // Real preflight: report each secret reference, then resolve.
30        let refs = crate::secrets::scan_path_refs(&path)?;
31        let cfg = PipelineConfig::from_path_async(&path).await?;
32        for (scheme, reference) in &refs {
33            println!("secret: {scheme}:{reference} → resolved");
34        }
35        cfg
36    };
37    let nodes = expand(&cfg)?;
38
39    // Validate the schedule block (cron / timezone / bounds) so `faucet validate`
40    // catches schedule misconfiguration in CI without running. Offline-safe.
41    #[cfg(feature = "schedule")]
42    if let Some(spec) = &cfg.schedule {
43        crate::schedule::compiled::CompiledSchedule::compile(spec)?;
44        println!(
45            "schedule: cron '{}' tz '{}' — valid",
46            spec.cron, spec.timezone
47        );
48    }
49
50    for node in &nodes {
51        // Verifying the schema lookup also catches unknown connector kinds.
52        source_schema(&node.source.kind)?;
53        sink_schema(&node.sink.kind)?;
54        for t in &node.transforms {
55            if !available_transforms().contains(&t.kind.as_str()) {
56                return Err(CliError::UnknownTransform {
57                    name: format!("{} (row '{}')", t.kind, node.id),
58                    available: available_transforms().join(", "),
59                });
60            }
61        }
62        if let Some(state) = &node.state
63            && !available_state_kinds().contains(&state.kind.as_str())
64        {
65            return Err(CliError::UnknownStateStore {
66                name: format!("{} (row '{}')", state.kind, node.id),
67                available: available_state_kinds().join(", "),
68            });
69        }
70    }
71
72    let roots = nodes
73        .iter()
74        .filter(|n| matches!(n.role, NodeRole::Root))
75        .count();
76    let children = nodes.len() - roots;
77    println!(
78        "ok: '{}' rows={} (roots={}, children={}) execution={}",
79        cfg.name.as_deref().unwrap_or("(unnamed)"),
80        nodes.len(),
81        roots,
82        children,
83        cfg.execution
84            .as_ref()
85            .map(|e| format!(
86                "max_concurrent={:?} on_error={:?}",
87                e.max_concurrent.unwrap_or(0),
88                e.on_error
89            ))
90            .unwrap_or_else(|| "(defaults)".to_owned()),
91    );
92    for node in &nodes {
93        let role = match &node.role {
94            NodeRole::Root => "root".to_owned(),
95            NodeRole::Child {
96                parent_id,
97                parent_key,
98            } => {
99                format!("child of '{parent_id}' (parent_key={parent_key})")
100            }
101        };
102        println!(
103            "  - {} [{}] source={} sink={}",
104            node.id, role, node.source.kind, node.sink.kind
105        );
106    }
107    Ok(())
108}
faucet_cli/commands/validate.rs

faucet_cli/commands/
validate.rs
