faucet_cli/catalog/
model.rs

1//! Pure helpers for building catalog observations (#279): dataset-URI
2//! canonicalization (fold `${now.*}`-derived segments back to their tokens,
3//! redact credentials) and schema inference over the run's record samples.
4
5use chrono::{DateTime, FixedOffset};
6use serde_json::Value;
7
8/// Canonicalize a concrete dataset URI for use as the catalog identity key.
9///
10/// 1. **`${now.*}` folding** — a dated path like `./out/dt=2026-07-06/x.jsonl`
11///    produced by a `${now.date}` template would otherwise mint a new catalog
12///    dataset every day (the cardinality trap called out in #279). For every
13///    `${now.*}` token that appears in the connector's *raw* (pre-resolution)
14///    config, the token's rendered value is replaced back with the token text
15///    in the URI, so all runs of the template converge on one dataset.
16/// 2. **Credential redaction** — via `faucet_core::redact_uri_credentials`,
17///    so a `postgres://user:pass@host/db` URI never lands in the store.
18pub fn canonicalize_uri(uri: &str, raw_config: &Value, clock: DateTime<FixedOffset>) -> String {
19    let mut tokens: Vec<String> = Vec::new();
20    collect_now_tokens(raw_config, &mut tokens);
21    tokens.sort();
22    tokens.dedup();
23
24    // Render each token with the same clock the run used, then substitute the
25    // rendered text back to the token — longest rendering first, so e.g.
26    // `${now.datetime}` wins over the `${now.year}` embedded within it.
27    let mut pairs: Vec<(String, String)> = tokens
28        .into_iter()
29        .filter_map(|t| {
30            crate::interpolate::resolve_now(&t, clock)
31                .ok()
32                .filter(|rendered| !rendered.is_empty() && rendered != &t)
33                .map(|rendered| (rendered, t))
34        })
35        .collect();
36    pairs.sort_by_key(|(rendered, _)| std::cmp::Reverse(rendered.len()));
37
38    let mut out = uri.to_string();
39    for (rendered, token) in pairs {
40        out = out.replace(&rendered, &token);
41    }
42    faucet_core::redact_uri_credentials(&out)
43}
44
45/// Collect every `${now.…}` token appearing in the string values of `v`.
46fn collect_now_tokens(v: &Value, out: &mut Vec<String>) {
47    match v {
48        Value::String(s) => {
49            let mut rest = s.as_str();
50            while let Some(start) = rest.find("${now.") {
51                let tail = &rest[start..];
52                match tail.find('}') {
53                    Some(end) => {
54                        out.push(tail[..=end].to_string());
55                        rest = &tail[end + 1..];
56                    }
57                    None => break,
58                }
59            }
60        }
61        Value::Array(items) => items.iter().for_each(|i| collect_now_tokens(i, out)),
62        Value::Object(map) => map.values().for_each(|i| collect_now_tokens(i, out)),
63        _ => {}
64    }
65}
66
67/// Infer an `infer_schema`-shaped record schema from a run's sample, `None`
68/// when nothing was sampled (empty run, or non-object records only).
69pub fn schema_from_samples(samples: &[Value]) -> Option<Value> {
70    if samples.is_empty() {
71        return None;
72    }
73    let schema = faucet_core::schema::infer_schema(samples);
74    // A sample of non-object records infers no properties — not a schema
75    // worth a timeline entry.
76    schema
77        .get("properties")
78        .and_then(Value::as_object)
79        .is_some_and(|p| !p.is_empty())
80        .then_some(schema)
81}
82
83#[cfg(test)]
84mod tests {
85    use super::*;
86    use serde_json::json;
87
88    fn clock() -> DateTime<FixedOffset> {
89        DateTime::parse_from_rfc3339("2026-07-06T10:30:00Z").unwrap()
90    }
91
92    #[test]
93    fn folds_now_tokens_back_into_the_uri() {
94        let raw = json!({"path": "./out/dt=${now.date}/part.jsonl"});
95        let uri = canonicalize_uri("file://./out/dt=2026-07-06/part.jsonl", &raw, clock());
96        assert_eq!(uri, "file://./out/dt=${now.date}/part.jsonl");
97    }
98
99    #[test]
100    fn folds_strftime_tokens_and_ignores_unused_ones() {
101        let raw = json!({"path": "./out/${now.strftime.%Y/%m}/x.jsonl", "other": "${now.unix}"});
102        let uri = canonicalize_uri("file://./out/2026/07/x.jsonl", &raw, clock());
103        assert_eq!(uri, "file://./out/${now.strftime.%Y/%m}/x.jsonl");
104    }
105
106    #[test]
107    fn without_now_tokens_the_uri_is_untouched_except_redaction() {
108        let raw = json!({"connection_url": "postgres://u:pw@h:5432/db"});
109        let uri = canonicalize_uri("postgres://u:pw@h:5432/db/public.users", &raw, clock());
110        assert!(!uri.contains("pw"), "credentials must be redacted: {uri}");
111        assert!(uri.contains("h:5432"));
112    }
113
114    #[test]
115    fn collect_finds_multiple_tokens_in_one_string() {
116        let mut out = Vec::new();
117        collect_now_tokens(&json!("a-${now.year}-${now.month}-b"), &mut out);
118        assert_eq!(out, vec!["${now.year}", "${now.month}"]);
119        // Unterminated token is ignored, not a panic.
120        out.clear();
121        collect_now_tokens(&json!("broken ${now.date"), &mut out);
122        assert!(out.is_empty());
123        // Tokens inside arrays (e.g. a list-valued config field) are found too.
124        out.clear();
125        collect_now_tokens(&json!({"paths": ["x", "${now.date}"]}), &mut out);
126        assert_eq!(out, vec!["${now.date}"]);
127    }
128
129    #[test]
130    fn schema_from_samples_requires_object_records() {
131        assert!(schema_from_samples(&[]).is_none());
132        assert!(schema_from_samples(&[json!("scalar")]).is_none());
133        let schema = schema_from_samples(&[json!({"id": 1, "name": "a"})]).unwrap();
134        assert_eq!(schema["type"], "object");
135        assert!(schema["properties"]["id"].is_object());
136    }
137}
faucet_cli/catalog/model.rs

faucet_cli/catalog/
model.rs
