faucet_cli/commands/
fmt.rs

1//! `faucet fmt` — canonicalize a pipeline config (#387).
2//!
3//! A deterministic, idempotent config formatter — the config analogue of
4//! `cargo fmt` / `terraform fmt`. It parses the file, rewrites every object with
5//! a stable key order (a curated priority order for the well-known blocks, then
6//! alphabetical for everything else), and re-serializes it. Running `fmt` twice
7//! is always a no-op, so `--check` is a cheap CI gate for "is this config
8//! normalized?".
9//!
10//! Canonicalization is a **pure** `serde_json::Value → serde_json::Value`
11//! transform ([`canonicalize`]); the command layer only reads/writes files and
12//! renders the `--check` diff.
13//!
14//! **Scope:** `fmt` formats the *literal* file — it does not resolve
15//! `${env:…}` / `${file:…}` interpolation, apply `!include` / `extends`
16//! composition, or drop schema defaults. Use `faucet validate --show-composed`
17//! to see the composed result.
18//!
19//! **Comments are not preserved** — the config is parsed and re-serialized, the
20//! same limitation `faucet migrate` documents.
21
22use crate::cli::FmtArgs;
23use crate::error::{CliError, CliResult};
24use serde_json::{Map, Value};
25use std::path::{Path, PathBuf};
26
27/// Execute the `fmt` subcommand.
28pub async fn run(args: FmtArgs) -> CliResult<()> {
29    let cwd = std::env::current_dir()?;
30    let paths: Vec<PathBuf> = if args.configs.is_empty() {
31        vec![
32            crate::env_loader::discover_config_path(&cwd)
33                .ok_or_else(|| CliError::Config("no config file found to format".into()))?,
34        ]
35    } else {
36        args.configs.clone()
37    };
38
39    let mut not_canonical = 0usize;
40    for path in &paths {
41        let text = std::fs::read_to_string(path)
42            .map_err(|e| CliError::Config(format!("cannot read '{}': {e}", path.display())))?;
43        let format = ConfigFormat::from_path(path)?;
44        let value = format.parse(&text, path)?;
45        let formatted = format.render(&canonicalize(value), path)?;
46
47        if args.check {
48            if formatted != text {
49                not_canonical += 1;
50                eprintln!("{}: not canonical", path.display());
51                eprint!("{}", unified_diff(&text, &formatted));
52            }
53            continue;
54        }
55
56        if args.stdout {
57            print!("{formatted}");
58            continue;
59        }
60
61        if formatted == text {
62            println!("{}: already formatted", path.display());
63        } else {
64            std::fs::write(path, &formatted)
65                .map_err(|e| CliError::Config(format!("cannot write '{}': {e}", path.display())))?;
66            println!("{}: formatted", path.display());
67        }
68    }
69
70    if not_canonical > 0 {
71        return Err(CliError::Config(format!(
72            "{not_canonical} file{} not canonical; run `faucet fmt` to format",
73            if not_canonical == 1 { "" } else { "s" }
74        )));
75    }
76    Ok(())
77}
78
79/// The on-disk serialization of a config file.
80#[derive(Clone, Copy)]
81enum ConfigFormat {
82    Yaml,
83    Json,
84}
85
86impl ConfigFormat {
87    fn from_path(path: &Path) -> CliResult<Self> {
88        match path.extension().and_then(|e| e.to_str()) {
89            Some("yaml") | Some("yml") => Ok(ConfigFormat::Yaml),
90            Some("json") => Ok(ConfigFormat::Json),
91            _ => Err(CliError::Config(format!(
92                "unsupported config extension for '{}' (expected .yaml/.yml/.json)",
93                path.display()
94            ))),
95        }
96    }
97
98    fn parse(self, text: &str, path: &Path) -> CliResult<Value> {
99        let err = |e: String| CliError::Config(format!("cannot parse '{}': {e}", path.display()));
100        match self {
101            ConfigFormat::Yaml => serde_yaml::from_str(text).map_err(|e| err(e.to_string())),
102            ConfigFormat::Json => serde_json::from_str(text).map_err(|e| err(e.to_string())),
103        }
104    }
105
106    fn render(self, value: &Value, path: &Path) -> CliResult<String> {
107        let err =
108            |e: String| CliError::Config(format!("cannot serialize '{}': {e}", path.display()));
109        match self {
110            ConfigFormat::Yaml => serde_yaml::to_string(value).map_err(|e| err(e.to_string())),
111            ConfigFormat::Json => {
112                let mut s = serde_json::to_string_pretty(value).map_err(|e| err(e.to_string()))?;
113                s.push('\n');
114                Ok(s)
115            }
116        }
117    }
118}
119
120/// The canonical key order for the well-known config blocks. Keys appear in this
121/// order at the front of their object; every other key follows alphabetically.
122/// A single flat list works because these names are unambiguous across the nesting
123/// levels they appear at (there is no top-level `type`, no connector-level
124/// `version`, etc.).
125const KEY_ORDER: &[&str] = &[
126    // ── top level ────────────────────────────────────────────────────────────
127    "version",
128    "name",
129    "vars",
130    "auth",
131    "pipeline",
132    "matrix",
133    "execution",
134    "selection",
135    // ── pipeline children ──────────────────────────────────────────────────────
136    "sources",
137    "source",
138    "sinks",
139    "sink",
140    "transforms",
141    "state",
142    // ── connector / matrix-row block children ──────────────────────────────────
143    "id",
144    "parent",
145    "parent_key",
146    "depends_on",
147    "type",
148    "ref",
149    "status",
150    "tags",
151    "inherit_transforms",
152    "config",
153    // ── remaining top-level blocks (kept deterministic, after the canonical set) ─
154    "delivery",
155    "resilience",
156    "sla",
157    "backfill",
158    "replication",
159    "schedule",
160    "notifications",
161    "lineage",
162    "catalog",
163    "profiles",
164];
165
166/// Rank of `key` in the canonical order, or `KEY_ORDER.len()` for any key not in
167/// the list (which then sorts alphabetically after the ranked keys).
168fn key_rank(key: &str) -> usize {
169    KEY_ORDER
170        .iter()
171        .position(|k| *k == key)
172        .unwrap_or(KEY_ORDER.len())
173}
174
175/// Rewrite `value` into canonical form: every object's keys are reordered by
176/// (canonical rank, then name), recursively. Pure and idempotent — running it on
177/// its own output is a no-op. Relies on `serde_json`'s `preserve_order` feature
178/// (enabled workspace-wide) so the rebuilt insertion order survives serialization.
179pub fn canonicalize(value: Value) -> Value {
180    match value {
181        Value::Object(map) => {
182            let mut entries: Vec<(String, Value)> =
183                map.into_iter().map(|(k, v)| (k, canonicalize(v))).collect();
184            entries.sort_by(|(a, _), (b, _)| key_rank(a).cmp(&key_rank(b)).then_with(|| a.cmp(b)));
185            Value::Object(entries.into_iter().collect::<Map<String, Value>>())
186        }
187        Value::Array(items) => Value::Array(items.into_iter().map(canonicalize).collect()),
188        scalar => scalar,
189    }
190}
191
192/// A minimal LCS-based unified-ish line diff, used only to show what `--check`
193/// would change. Lines present in both are context; removed lines are `-` and
194/// added lines are `+`. Not a git-grade diff, but deterministic and enough to
195/// point a reviewer at the reordering.
196fn unified_diff(old: &str, new: &str) -> String {
197    let a: Vec<&str> = old.lines().collect();
198    let b: Vec<&str> = new.lines().collect();
199    // LCS table.
200    let (n, m) = (a.len(), b.len());
201    let mut lcs = vec![vec![0usize; m + 1]; n + 1];
202    for i in (0..n).rev() {
203        for j in (0..m).rev() {
204            lcs[i][j] = if a[i] == b[j] {
205                lcs[i + 1][j + 1] + 1
206            } else {
207                lcs[i + 1][j].max(lcs[i][j + 1])
208            };
209        }
210    }
211    let mut out = String::new();
212    let (mut i, mut j) = (0, 0);
213    while i < n && j < m {
214        if a[i] == b[j] {
215            out.push_str(&format!("  {}\n", a[i]));
216            i += 1;
217            j += 1;
218        } else if lcs[i + 1][j] >= lcs[i][j + 1] {
219            out.push_str(&format!("- {}\n", a[i]));
220            i += 1;
221        } else {
222            out.push_str(&format!("+ {}\n", b[j]));
223            j += 1;
224        }
225    }
226    while i < n {
227        out.push_str(&format!("- {}\n", a[i]));
228        i += 1;
229    }
230    while j < m {
231        out.push_str(&format!("+ {}\n", b[j]));
232        j += 1;
233    }
234    out
235}
236
237#[cfg(test)]
238mod tests {
239    use super::*;
240    use serde_json::json;
241
242    #[test]
243    fn reorders_top_level_keys_canonically() {
244        let v = json!({
245            "matrix": [],
246            "pipeline": { "sink": {}, "source": {} },
247            "name": "demo",
248            "version": 1,
249        });
250        let out = canonicalize(v);
251        let keys: Vec<&String> = out.as_object().unwrap().keys().collect();
252        assert_eq!(keys, ["version", "name", "pipeline", "matrix"]);
253        // Nested pipeline is reordered too: source before sink.
254        let pk: Vec<&String> = out["pipeline"].as_object().unwrap().keys().collect();
255        assert_eq!(pk, ["source", "sink"]);
256    }
257
258    #[test]
259    fn connector_block_puts_type_before_config_and_sorts_rest() {
260        let v = json!({
261            "source": { "config": { "url": "x", "auth": {} }, "type": "rest", "status": "active" }
262        });
263        let out = canonicalize(v);
264        let sk: Vec<&String> = out["source"].as_object().unwrap().keys().collect();
265        assert_eq!(sk, ["type", "status", "config"]);
266    }
267
268    #[test]
269    fn unknown_keys_sort_alphabetically_after_ranked_keys() {
270        let v = json!({ "config": { "zebra": 1, "alpha": 2, "mango": 3 } });
271        let out = canonicalize(v);
272        let ck: Vec<&String> = out["config"].as_object().unwrap().keys().collect();
273        assert_eq!(ck, ["alpha", "mango", "zebra"]);
274    }
275
276    #[test]
277    fn canonicalize_is_idempotent() {
278        let v = json!({
279            "version": 1,
280            "pipeline": { "sink": { "type": "jsonl", "config": { "b": 1, "a": 2 } },
281                          "source": { "config": {}, "type": "rest" } },
282            "name": "x",
283        });
284        let once = canonicalize(v);
285        let twice = canonicalize(once.clone());
286        assert_eq!(once, twice);
287    }
288
289    #[test]
290    fn config_format_from_path() {
291        assert!(matches!(
292            ConfigFormat::from_path(Path::new("a.yaml")),
293            Ok(ConfigFormat::Yaml)
294        ));
295        assert!(matches!(
296            ConfigFormat::from_path(Path::new("a.json")),
297            Ok(ConfigFormat::Json)
298        ));
299        assert!(ConfigFormat::from_path(Path::new("a.toml")).is_err());
300    }
301
302    #[test]
303    fn render_yaml_is_byte_stable_across_two_passes() {
304        let p = Path::new("f.yaml");
305        let v = ConfigFormat::Yaml
306            .parse(
307                "pipeline:\n  sink: {}\n  source: {}\nname: d\nversion: 1\n",
308                p,
309            )
310            .unwrap();
311        let once = ConfigFormat::Yaml.render(&canonicalize(v), p).unwrap();
312        let reparsed = ConfigFormat::Yaml.parse(&once, p).unwrap();
313        let twice = ConfigFormat::Yaml
314            .render(&canonicalize(reparsed), p)
315            .unwrap();
316        assert_eq!(once, twice, "fmt must be idempotent at the byte level");
317        assert!(once.starts_with("version: 1"), "{once}");
318    }
319
320    #[test]
321    fn unified_diff_marks_added_and_removed_lines() {
322        let d = unified_diff("a\nb\nc\n", "a\nx\nc\n");
323        assert!(d.contains("- b"), "{d}");
324        assert!(d.contains("+ x"), "{d}");
325        assert!(d.contains("  a"), "{d}");
326    }
327
328    fn write_tmp(name: &str, body: &str) -> (tempfile::TempDir, PathBuf) {
329        let dir = tempfile::tempdir().expect("tempdir");
330        let path = dir.path().join(name);
331        std::fs::write(&path, body).expect("write");
332        (dir, path)
333    }
334
335    const UNSORTED: &str = "name: demo\nversion: 1\npipeline:\n  sink: { type: jsonl, config: { path: o } }\n  source: { type: rest, config: {} }\n";
336
337    #[tokio::test]
338    async fn run_rewrites_file_in_place_and_is_idempotent() {
339        let (_d, path) = write_tmp("f.yaml", UNSORTED);
340        run(FmtArgs {
341            configs: vec![path.clone()],
342            check: false,
343            stdout: false,
344        })
345        .await
346        .unwrap();
347        let after = std::fs::read_to_string(&path).unwrap();
348        assert!(after.starts_with("version: 1"), "{after}");
349        // Second pass leaves it byte-identical.
350        run(FmtArgs {
351            configs: vec![path.clone()],
352            check: false,
353            stdout: false,
354        })
355        .await
356        .unwrap();
357        assert_eq!(std::fs::read_to_string(&path).unwrap(), after);
358    }
359
360    #[tokio::test]
361    async fn run_check_fails_on_unsorted_passes_on_canonical() {
362        let (_d, path) = write_tmp("f.yaml", UNSORTED);
363        assert!(
364            run(FmtArgs {
365                configs: vec![path.clone()],
366                check: true,
367                stdout: false,
368            })
369            .await
370            .is_err(),
371            "--check must fail on a non-canonical file"
372        );
373        // Format it, then --check passes.
374        run(FmtArgs {
375            configs: vec![path.clone()],
376            check: false,
377            stdout: false,
378        })
379        .await
380        .unwrap();
381        run(FmtArgs {
382            configs: vec![path],
383            check: true,
384            stdout: false,
385        })
386        .await
387        .expect("--check passes on a canonical file");
388    }
389
390    #[tokio::test]
391    async fn run_stdout_leaves_file_untouched() {
392        let (_d, path) = write_tmp("f.yaml", UNSORTED);
393        run(FmtArgs {
394            configs: vec![path.clone()],
395            check: false,
396            stdout: true,
397        })
398        .await
399        .unwrap();
400        assert_eq!(std::fs::read_to_string(&path).unwrap(), UNSORTED);
401    }
402}
faucet_cli/commands/fmt.rs

faucet_cli/commands/
fmt.rs
