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    "params",
131    "auth",
132    "pipeline",
133    "matrix",
134    "execution",
135    "selection",
136    // ── pipeline children ──────────────────────────────────────────────────────
137    "sources",
138    "source",
139    "sinks",
140    "sink",
141    "transforms",
142    "state",
143    // ── connector / matrix-row block children ──────────────────────────────────
144    "id",
145    "parent",
146    "parent_key",
147    "depends_on",
148    "type",
149    "ref",
150    "status",
151    "tags",
152    "inherit_transforms",
153    "config",
154    // ── remaining top-level blocks (kept deterministic, after the canonical set) ─
155    "delivery",
156    "resilience",
157    "sla",
158    "backfill",
159    "replication",
160    "schedule",
161    "notifications",
162    "lineage",
163    "catalog",
164    "profiles",
165];
166
167/// Rank of `key` in the canonical order, or `KEY_ORDER.len()` for any key not in
168/// the list (which then sorts alphabetically after the ranked keys).
169fn key_rank(key: &str) -> usize {
170    KEY_ORDER
171        .iter()
172        .position(|k| *k == key)
173        .unwrap_or(KEY_ORDER.len())
174}
175
176/// Rewrite `value` into canonical form: every object's keys are reordered by
177/// (canonical rank, then name), recursively. Pure and idempotent — running it on
178/// its own output is a no-op. Relies on `serde_json`'s `preserve_order` feature
179/// (enabled workspace-wide) so the rebuilt insertion order survives serialization.
180pub fn canonicalize(value: Value) -> Value {
181    match value {
182        Value::Object(map) => {
183            let mut entries: Vec<(String, Value)> =
184                map.into_iter().map(|(k, v)| (k, canonicalize(v))).collect();
185            entries.sort_by(|(a, _), (b, _)| key_rank(a).cmp(&key_rank(b)).then_with(|| a.cmp(b)));
186            Value::Object(entries.into_iter().collect::<Map<String, Value>>())
187        }
188        Value::Array(items) => Value::Array(items.into_iter().map(canonicalize).collect()),
189        scalar => scalar,
190    }
191}
192
193/// A minimal LCS-based unified-ish line diff, used only to show what `--check`
194/// would change. Lines present in both are context; removed lines are `-` and
195/// added lines are `+`. Not a git-grade diff, but deterministic and enough to
196/// point a reviewer at the reordering.
197fn unified_diff(old: &str, new: &str) -> String {
198    let a: Vec<&str> = old.lines().collect();
199    let b: Vec<&str> = new.lines().collect();
200    // LCS table.
201    let (n, m) = (a.len(), b.len());
202    let mut lcs = vec![vec![0usize; m + 1]; n + 1];
203    for i in (0..n).rev() {
204        for j in (0..m).rev() {
205            lcs[i][j] = if a[i] == b[j] {
206                lcs[i + 1][j + 1] + 1
207            } else {
208                lcs[i + 1][j].max(lcs[i][j + 1])
209            };
210        }
211    }
212    let mut out = String::new();
213    let (mut i, mut j) = (0, 0);
214    while i < n && j < m {
215        if a[i] == b[j] {
216            out.push_str(&format!("  {}\n", a[i]));
217            i += 1;
218            j += 1;
219        } else if lcs[i + 1][j] >= lcs[i][j + 1] {
220            out.push_str(&format!("- {}\n", a[i]));
221            i += 1;
222        } else {
223            out.push_str(&format!("+ {}\n", b[j]));
224            j += 1;
225        }
226    }
227    while i < n {
228        out.push_str(&format!("- {}\n", a[i]));
229        i += 1;
230    }
231    while j < m {
232        out.push_str(&format!("+ {}\n", b[j]));
233        j += 1;
234    }
235    out
236}
237
238#[cfg(test)]
239mod tests {
240    use super::*;
241    use serde_json::json;
242
243    #[test]
244    fn reorders_top_level_keys_canonically() {
245        let v = json!({
246            "matrix": [],
247            "pipeline": { "sink": {}, "source": {} },
248            "name": "demo",
249            "version": 1,
250        });
251        let out = canonicalize(v);
252        let keys: Vec<&String> = out.as_object().unwrap().keys().collect();
253        assert_eq!(keys, ["version", "name", "pipeline", "matrix"]);
254        // Nested pipeline is reordered too: source before sink.
255        let pk: Vec<&String> = out["pipeline"].as_object().unwrap().keys().collect();
256        assert_eq!(pk, ["source", "sink"]);
257    }
258
259    #[test]
260    fn connector_block_puts_type_before_config_and_sorts_rest() {
261        let v = json!({
262            "source": { "config": { "url": "x", "auth": {} }, "type": "rest", "status": "active" }
263        });
264        let out = canonicalize(v);
265        let sk: Vec<&String> = out["source"].as_object().unwrap().keys().collect();
266        assert_eq!(sk, ["type", "status", "config"]);
267    }
268
269    #[test]
270    fn unknown_keys_sort_alphabetically_after_ranked_keys() {
271        let v = json!({ "config": { "zebra": 1, "alpha": 2, "mango": 3 } });
272        let out = canonicalize(v);
273        let ck: Vec<&String> = out["config"].as_object().unwrap().keys().collect();
274        assert_eq!(ck, ["alpha", "mango", "zebra"]);
275    }
276
277    #[test]
278    fn canonicalize_is_idempotent() {
279        let v = json!({
280            "version": 1,
281            "pipeline": { "sink": { "type": "jsonl", "config": { "b": 1, "a": 2 } },
282                          "source": { "config": {}, "type": "rest" } },
283            "name": "x",
284        });
285        let once = canonicalize(v);
286        let twice = canonicalize(once.clone());
287        assert_eq!(once, twice);
288    }
289
290    #[test]
291    fn config_format_from_path() {
292        assert!(matches!(
293            ConfigFormat::from_path(Path::new("a.yaml")),
294            Ok(ConfigFormat::Yaml)
295        ));
296        assert!(matches!(
297            ConfigFormat::from_path(Path::new("a.json")),
298            Ok(ConfigFormat::Json)
299        ));
300        assert!(ConfigFormat::from_path(Path::new("a.toml")).is_err());
301    }
302
303    #[test]
304    fn render_yaml_is_byte_stable_across_two_passes() {
305        let p = Path::new("f.yaml");
306        let v = ConfigFormat::Yaml
307            .parse(
308                "pipeline:\n  sink: {}\n  source: {}\nname: d\nversion: 1\n",
309                p,
310            )
311            .unwrap();
312        let once = ConfigFormat::Yaml.render(&canonicalize(v), p).unwrap();
313        let reparsed = ConfigFormat::Yaml.parse(&once, p).unwrap();
314        let twice = ConfigFormat::Yaml
315            .render(&canonicalize(reparsed), p)
316            .unwrap();
317        assert_eq!(once, twice, "fmt must be idempotent at the byte level");
318        assert!(once.starts_with("version: 1"), "{once}");
319    }
320
321    #[test]
322    fn unified_diff_marks_added_and_removed_lines() {
323        let d = unified_diff("a\nb\nc\n", "a\nx\nc\n");
324        assert!(d.contains("- b"), "{d}");
325        assert!(d.contains("+ x"), "{d}");
326        assert!(d.contains("  a"), "{d}");
327    }
328
329    fn write_tmp(name: &str, body: &str) -> (tempfile::TempDir, PathBuf) {
330        let dir = tempfile::tempdir().expect("tempdir");
331        let path = dir.path().join(name);
332        std::fs::write(&path, body).expect("write");
333        (dir, path)
334    }
335
336    const UNSORTED: &str = "name: demo\nversion: 1\npipeline:\n  sink: { type: jsonl, config: { path: o } }\n  source: { type: rest, config: {} }\n";
337
338    #[tokio::test]
339    async fn run_rewrites_file_in_place_and_is_idempotent() {
340        let (_d, path) = write_tmp("f.yaml", UNSORTED);
341        run(FmtArgs {
342            configs: vec![path.clone()],
343            check: false,
344            stdout: false,
345        })
346        .await
347        .unwrap();
348        let after = std::fs::read_to_string(&path).unwrap();
349        assert!(after.starts_with("version: 1"), "{after}");
350        // Second pass leaves it byte-identical.
351        run(FmtArgs {
352            configs: vec![path.clone()],
353            check: false,
354            stdout: false,
355        })
356        .await
357        .unwrap();
358        assert_eq!(std::fs::read_to_string(&path).unwrap(), after);
359    }
360
361    #[tokio::test]
362    async fn run_check_fails_on_unsorted_passes_on_canonical() {
363        let (_d, path) = write_tmp("f.yaml", UNSORTED);
364        assert!(
365            run(FmtArgs {
366                configs: vec![path.clone()],
367                check: true,
368                stdout: false,
369            })
370            .await
371            .is_err(),
372            "--check must fail on a non-canonical file"
373        );
374        // Format it, then --check passes.
375        run(FmtArgs {
376            configs: vec![path.clone()],
377            check: false,
378            stdout: false,
379        })
380        .await
381        .unwrap();
382        run(FmtArgs {
383            configs: vec![path],
384            check: true,
385            stdout: false,
386        })
387        .await
388        .expect("--check passes on a canonical file");
389    }
390
391    #[tokio::test]
392    async fn run_stdout_leaves_file_untouched() {
393        let (_d, path) = write_tmp("f.yaml", UNSORTED);
394        run(FmtArgs {
395            configs: vec![path.clone()],
396            check: false,
397            stdout: true,
398        })
399        .await
400        .unwrap();
401        assert_eq!(std::fs::read_to_string(&path).unwrap(), UNSORTED);
402    }
403}
faucet_cli/commands/fmt.rs

faucet_cli/commands/
fmt.rs
