faucet_cli/commands/
init.rs

1//! `faucet init` — scaffold a starter `pipeline.yaml` from each connector's
2//! JSON Schema. Defaults to a `rest` → `jsonl` pipeline so `faucet init` with
3//! no flags continues to produce the same shape it did before this command
4//! grew schema-driven scaffolding.
5
6use std::collections::HashMap;
7
8use crate::cli::InitArgs;
9use crate::error::{CliError, CliResult};
10#[cfg(feature = "cli-interactive")]
11use crate::init_template::discover_tagged_enum_fields;
12use crate::init_template::schema_to_yaml_template_with_choices;
13use crate::registry;
14
15const DEFAULT_SOURCE: &str = "rest";
16const DEFAULT_SINK: &str = "jsonl";
17const DEFAULT_NAME: &str = "my-pipeline";
18const CONFIG_INDENT: usize = 8;
19
20/// Execute the `init` subcommand.
21pub async fn run(args: InitArgs) -> CliResult<()> {
22    if args.output.exists() && !args.force {
23        return Err(CliError::ScaffoldExists {
24            path: args.output.clone(),
25        });
26    }
27
28    let (source_kind, sink_kind) = resolve_kinds(&args)?;
29    let name = args.name.as_deref().unwrap_or(DEFAULT_NAME);
30    let template = &args.template;
31
32    let source_schema = registry::source_schema(&source_kind)?;
33    let sink_schema = registry::sink_schema(&sink_kind)?;
34    let (source_choices, sink_choices) = if args.interactive {
35        interactive_variant_choices(&source_schema, &sink_schema)?
36    } else {
37        (HashMap::new(), HashMap::new())
38    };
39
40    let body = render_pipeline(
41        name,
42        template,
43        &source_kind,
44        &source_schema,
45        &source_choices,
46        &sink_kind,
47        &sink_schema,
48        &sink_choices,
49    );
50    std::fs::write(&args.output, body)?;
51    println!("wrote {}", args.output.display());
52    Ok(())
53}
54
55fn resolve_kinds(args: &InitArgs) -> CliResult<(String, String)> {
56    let source = args.source.clone();
57    let sink = args.sink.clone();
58
59    let (source, sink) = if args.interactive {
60        interactive_prompt(source, sink)?
61    } else {
62        (
63            source.unwrap_or_else(|| DEFAULT_SOURCE.to_string()),
64            sink.unwrap_or_else(|| DEFAULT_SINK.to_string()),
65        )
66    };
67
68    if !registry::source_exists(&source) {
69        return Err(unknown_kind_err("source", &source));
70    }
71    if !registry::sink_exists(&sink) {
72        return Err(unknown_kind_err("sink", &sink));
73    }
74    Ok((source, sink))
75}
76
77#[cfg(feature = "cli-interactive")]
78fn interactive_prompt(source: Option<String>, sink: Option<String>) -> CliResult<(String, String)> {
79    use std::io::IsTerminal;
80    if !std::io::stdin().is_terminal() {
81        return fallback_kinds(source, sink, "stdin is not a TTY");
82    }
83    let sources = registry::source_kinds();
84    let sinks = registry::sink_kinds();
85    let s = if let Some(s) = source {
86        s
87    } else {
88        prompt_select("source", &sources)?
89    };
90    let k = if let Some(k) = sink {
91        k
92    } else {
93        prompt_select("sink", &sinks)?
94    };
95    Ok((s, k))
96}
97
98#[cfg(not(feature = "cli-interactive"))]
99fn interactive_prompt(source: Option<String>, sink: Option<String>) -> CliResult<(String, String)> {
100    fallback_kinds(
101        source,
102        sink,
103        "the `cli-interactive` build feature is not enabled",
104    )
105}
106
107fn fallback_kinds(
108    source: Option<String>,
109    sink: Option<String>,
110    reason: &str,
111) -> CliResult<(String, String)> {
112    match (source, sink) {
113        (Some(s), Some(k)) => Ok((s, k)),
114        (s, k) => {
115            tracing::warn!(
116                "ignoring --interactive: {reason}; falling back to --source/--sink (or defaults)"
117            );
118            Ok((
119                s.unwrap_or_else(|| DEFAULT_SOURCE.to_string()),
120                k.unwrap_or_else(|| DEFAULT_SINK.to_string()),
121            ))
122        }
123    }
124}
125
126#[cfg(feature = "cli-interactive")]
127fn prompt_select(kind: &str, options: &[&'static str]) -> CliResult<String> {
128    let choice = inquire::Select::new(&format!("Pick a {kind} connector"), options.to_vec())
129        .prompt()
130        .map_err(|e| CliError::Io(std::io::Error::other(e.to_string())))?;
131    Ok(choice.to_string())
132}
133
134#[cfg(feature = "cli-interactive")]
135fn interactive_variant_choices(
136    source_schema: &serde_json::Value,
137    sink_schema: &serde_json::Value,
138) -> CliResult<(HashMap<String, String>, HashMap<String, String>)> {
139    use std::io::IsTerminal;
140    if !std::io::stdin().is_terminal() {
141        return Ok((HashMap::new(), HashMap::new()));
142    }
143    let source = prompt_variants_for("source", source_schema)?;
144    let sink = prompt_variants_for("sink", sink_schema)?;
145    Ok((source, sink))
146}
147
148#[cfg(not(feature = "cli-interactive"))]
149fn interactive_variant_choices(
150    _source_schema: &serde_json::Value,
151    _sink_schema: &serde_json::Value,
152) -> CliResult<(HashMap<String, String>, HashMap<String, String>)> {
153    tracing::warn!("ignoring --interactive: the `cli-interactive` build feature is not enabled");
154    Ok((HashMap::new(), HashMap::new()))
155}
156
157#[cfg(feature = "cli-interactive")]
158fn prompt_variants_for(
159    side: &str,
160    schema: &serde_json::Value,
161) -> CliResult<HashMap<String, String>> {
162    let fields = discover_tagged_enum_fields(schema);
163    let mut choices = HashMap::new();
164    for field in fields {
165        let label = format!("Pick a variant for {side}.{}", field.path);
166        let opts: Vec<String> = field.variants.clone();
167        let chosen = inquire::Select::new(&label, opts)
168            .prompt()
169            .map_err(|e| CliError::Io(std::io::Error::other(e.to_string())))?;
170        choices.insert(field.path, chosen);
171    }
172    Ok(choices)
173}
174
175fn unknown_kind_err(kind: &'static str, name: &str) -> CliError {
176    let available = if kind == "source" {
177        registry::source_kinds()
178    } else {
179        registry::sink_kinds()
180    };
181    CliError::UnknownConnector {
182        kind,
183        name: name.to_owned(),
184        available: if available.is_empty() {
185            "(none — rebuild faucet-cli with the relevant feature enabled)".to_owned()
186        } else {
187            available.join(", ")
188        },
189    }
190}
191
192#[allow(clippy::too_many_arguments)]
193fn render_pipeline(
194    name: &str,
195    template: &str,
196    source_kind: &str,
197    source_schema: &serde_json::Value,
198    source_choices: &HashMap<String, String>,
199    sink_kind: &str,
200    sink_schema: &serde_json::Value,
201    sink_choices: &HashMap<String, String>,
202) -> String {
203    let source_yaml =
204        schema_to_yaml_template_with_choices(source_schema, CONFIG_INDENT, source_choices);
205    let sink_yaml = schema_to_yaml_template_with_choices(sink_schema, CONFIG_INDENT, sink_choices);
206
207    let mut body = String::new();
208    body.push_str("version: 1\n");
209    body.push_str(&format!("name: {name}\n\n"));
210    body.push_str("# Optional shared constants. Reference these anywhere via ${vars.KEY}.\n");
211    body.push_str("# vars:\n");
212    body.push_str("#   api_base: https://api.example.com\n\n");
213    body.push_str("# Named source/sink templates. Matrix rows pick from these via ref:.\n");
214    body.push_str("# A matrix row that omits ref: inherits the `default` template,\n");
215    body.push_str("# which keeps backwards-compat with the legacy singular shape.\n");
216    body.push_str("pipeline:\n");
217    body.push_str("  sources:\n");
218    body.push_str(&format!("    {template}:\n"));
219    body.push_str(&format!("      type: {source_kind}\n"));
220    body.push_str("      config:\n");
221    body.push_str(&source_yaml);
222    body.push('\n');
223    body.push_str("  # transforms:\n");
224    body.push_str("  #   - { type: keys_case, config: { mode: snake } }\n\n");
225    body.push_str("  sinks:\n");
226    body.push_str(&format!("    {template}:\n"));
227    body.push_str(&format!("      type: {sink_kind}\n"));
228    body.push_str("      config:\n");
229    body.push_str(&sink_yaml);
230    body.push('\n');
231    body.push_str("  # Optional state store (required by CDC sources and resumable runs).\n");
232    body.push_str("  # state:\n");
233    body.push_str("  #   type: file\n");
234    body.push_str("  #   config: { path: ./.faucet-state }\n\n");
235    body.push_str("  # Optional Dead Letter Queue.\n");
236    body.push_str("  # dlq:\n");
237    body.push_str("  #   sink:\n");
238    body.push_str("  #     type: jsonl\n");
239    body.push_str("  #     config: { path: ./dlq.jsonl }\n");
240    body.push_str("  #   on_batch_error: propagate   # or dlq_all\n\n");
241    body.push_str("# Optional matrix block. Each row picks a template via ref:\n");
242    body.push_str("# (omit ref: to inherit the `default` template above) and may\n");
243    body.push_str("# override `type:` / `config:` per row.\n");
244    body.push_str("# matrix:\n");
245    body.push_str("#   - id: users\n");
246    body.push_str(&format!(
247        "#     source: {{ ref: {template}, config: {{ path: /v1/users }} }}\n"
248    ));
249    body
250}
faucet_cli/commands/init.rs

faucet_cli/commands/
init.rs
