faucet_cli/serve/
load.rs

1//! Turn a submitted config body into expanded nodes, applying the workspace
2//! `--default-config` base. Mirrors `PipelineConfig::from_path_async` but merges
3//! a base `Value` and uses `from_value`. All `${env}`/`${file}`/`${secret}` and
4//! `${vault:…}`-style directives resolve against the *server's* environment and
5//! credentials (the documented privilege surface — spec §13).
6
7use crate::config::PipelineConfig;
8use crate::expand::{ExpandedNode, expand};
9use crate::serve::error::ServeError;
10use serde_json::Value;
11
12/// Wire format of a submitted config body.
13#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
14pub enum ConfigFormat {
15    #[default]
16    Yaml,
17    Json,
18}
19
20/// A loaded submission: the merged/resolved config and its expanded nodes.
21#[derive(Debug)]
22pub struct LoadedSubmission {
23    pub cfg: PipelineConfig,
24    pub nodes: Vec<ExpandedNode>,
25}
26
27/// Load + merge + expand a submitted config body.
28pub async fn load_submission(
29    body: &str,
30    format: ConfigFormat,
31    default_base: Option<&Value>,
32) -> Result<LoadedSubmission, ServeError> {
33    // 1. ${env}/${file}/${secret} interpolation against the server's env/fs.
34    let interpolated =
35        crate::interpolate::interpolate(body).map_err(|e| ServeError::BadConfig(e.to_string()))?;
36
37    // 2. Parse to a Value per the declared format.
38    let submitted: Value = match format {
39        ConfigFormat::Yaml => serde_yaml::from_str(&interpolated)
40            .map_err(|e| ServeError::BadConfig(format!("invalid YAML: {e}")))?,
41        ConfigFormat::Json => serde_json::from_str(&interpolated)
42            .map_err(|e| ServeError::BadConfig(format!("invalid JSON: {e}")))?,
43    };
44
45    // 3. Merge onto the workspace default (submitted wins; see merge.rs semantics).
46    let merged = match default_base {
47        Some(base) => {
48            let mut m = base.clone();
49            crate::merge::merge_value(&mut m, submitted);
50            m
51        }
52        None => submitted,
53    };
54
55    // 4. Version gate + structural-ref resolution.
56    let mut cfg = PipelineConfig::from_value(merged).map_err(|e| ServeError::Unprocessable {
57        message: e.to_string(),
58        details: None,
59    })?;
60
61    // serve runs once per submission; a schedule: block is a category error.
62    #[cfg(feature = "schedule")]
63    if cfg.schedule.is_some() {
64        return Err(ServeError::BadConfig(
65            "submitted config contains a `schedule:` block — serve runs once per \
66             submission; use `faucet schedule` for cron scheduling"
67                .into(),
68        ));
69    }
70
71    // 5. Secret-manager directives (${vault:…} etc.) with the server's creds.
72    crate::secrets::resolve_secrets(&mut cfg)
73        .await
74        .map_err(|e| ServeError::BadConfig(e.to_string()))?;
75
76    // 6. Expand the matrix.
77    let nodes = expand(&cfg).map_err(|e| ServeError::Unprocessable {
78        message: e.to_string(),
79        details: None,
80    })?;
81
82    Ok(LoadedSubmission { cfg, nodes })
83}
84
85#[cfg(test)]
86mod tests {
87    use super::*;
88    use serde_json::json;
89
90    fn base() -> Value {
91        json!({
92            "version": 1,
93            "pipeline": {
94                "source": { "type": "csv", "config": { "path": "DEFAULT.csv" } },
95                "sink": { "type": "jsonl", "config": { "path": "out.jsonl" } }
96            }
97        })
98    }
99
100    #[tokio::test]
101    async fn submitted_overrides_default() {
102        let body = r#"{ "pipeline": { "source": { "config": { "path": "OVERRIDE.csv" } } } }"#;
103        let loaded = load_submission(body, ConfigFormat::Json, Some(&base()))
104            .await
105            .unwrap();
106        // The override wins; the default sink survives the merge.
107        let node = &loaded.nodes[0];
108        assert_eq!(node.source.config["path"], "OVERRIDE.csv");
109        assert_eq!(node.sink.config["path"], "out.jsonl");
110    }
111
112    #[tokio::test]
113    async fn missing_version_without_base_is_unprocessable() {
114        // version defaults to 1 via serde, so this exercises the expand/validation
115        // failure path (pipeline with no source/sink). Accept either layer's error.
116        let body = r#"{ "pipeline": {} }"#;
117        let err = load_submission(body, ConfigFormat::Json, None)
118            .await
119            .unwrap_err();
120        assert!(matches!(
121            err,
122            ServeError::Unprocessable { .. } | ServeError::BadConfig(_)
123        ));
124    }
125
126    #[cfg(feature = "schedule")]
127    #[tokio::test]
128    async fn schedule_block_is_rejected() {
129        let body = r#"
130version: 1
131pipeline:
132  source: { type: csv, config: { path: x.csv } }
133  sink: { type: jsonl, config: { path: out.jsonl } }
134schedule:
135  cron: "0 * * * *"
136  timezone: UTC
137"#;
138        let err = load_submission(body, ConfigFormat::Yaml, None)
139            .await
140            .unwrap_err();
141        match err {
142            ServeError::BadConfig(m) => assert!(m.contains("schedule:")),
143            other => panic!("expected BadConfig, got {other:?}"),
144        }
145    }
146
147    #[tokio::test]
148    async fn invalid_yaml_is_bad_config() {
149        let err = load_submission("{[bad", ConfigFormat::Yaml, None)
150            .await
151            .unwrap_err();
152        assert!(matches!(err, ServeError::BadConfig(_)));
153    }
154}
faucet_cli/serve/load.rs

faucet_cli/serve/
load.rs
