use crate::cli::{
TemplateArgs, TemplateCommand, TemplateDeleteArgs, TemplateDeprecateArgs, TemplateLaunchArgs,
TemplateListArgs, TemplatePromoteArgs, TemplateRegisterArgs, TemplateRollbackArgs,
TemplateRunArgs, TemplateShowArgs, TemplateStoreArgs,
};
use crate::error::{CliError, CliResult};
use crate::serve::history::templates::{
TemplateRecord, TemplateSummary, VersionChannel, VersionSelector,
};
use crate::serve::load::ConfigFormat;
use crate::templates::{RegisterRequest, TemplateStore};
pub async fn run(args: TemplateArgs) -> CliResult<()> {
match args.command {
TemplateCommand::Register(a) => register(a).await,
TemplateCommand::List(a) => list(a).await,
TemplateCommand::Show(a) => show(a).await,
TemplateCommand::Launch(a) => launch(a).await,
TemplateCommand::Rollback(a) => rollback(a).await,
TemplateCommand::Deprecate(a) => deprecate(a).await,
TemplateCommand::Promote(a) => promote(a).await,
TemplateCommand::Delete(a) => delete(a).await,
TemplateCommand::Run(a) => run_template(a).await,
}
}
async fn connect(common: &TemplateStoreArgs) -> CliResult<TemplateStore> {
let cwd = std::env::current_dir()?;
let env_path =
crate::env_loader::resolve_env_file(common.env_file.as_deref(), common.no_env_file, &cwd)?;
crate::env_loader::load_env_file_if_present(env_path.as_deref())?;
crate::templates::resolve_store_url(&common.store).await
}
fn to_pretty<T: serde::Serialize>(value: &T) -> CliResult<String> {
serde_json::to_string_pretty(value)
.map_err(|e| CliError::Internal(format!("rendering template JSON: {e}")))
}
fn format_of(path: &std::path::Path) -> CliResult<ConfigFormat> {
match path
.extension()
.and_then(|e| e.to_str())
.map(str::to_ascii_lowercase)
.as_deref()
{
Some("yaml" | "yml") => Ok(ConfigFormat::Yaml),
Some("json") => Ok(ConfigFormat::Json),
_ => Err(CliError::UnknownExtension {
path: path.to_path_buf(),
}),
}
}
async fn register(args: TemplateRegisterArgs) -> CliResult<()> {
let store = connect(&args.common).await?;
let format = format_of(&args.config)?;
let body = std::fs::read_to_string(&args.config).map_err(|e| {
CliError::Config(format!(
"reading template config '{}': {e}",
args.config.display()
))
})?;
let tags = args
.tag
.iter()
.map(|t| VersionChannel::parse(t))
.collect::<CliResult<Vec<_>>>()?;
let record = crate::templates::register(
&store,
RegisterRequest {
id: args.id.clone(),
body,
format,
description: args.description.clone(),
tags: tags.clone(),
launch: args.launch,
created_by: None,
},
)
.await?;
if args.common.json {
println!("{}", to_pretty(&record.summary())?);
return Ok(());
}
println!(
"registered template '{}' version {}{}",
record.id,
record.version,
if tags.is_empty() {
String::new()
} else {
format!(
" (channels: {})",
tags.iter()
.map(|c| c.as_str())
.collect::<Vec<_>>()
.join(", ")
)
}
);
print_params(&record.summary());
println!(
"\ntrigger it with:\n faucet template run {} --store {}{}",
record.id,
args.common.store,
required_param_hint(&record.summary())
);
Ok(())
}
fn required_param_hint(summary: &TemplateSummary) -> String {
summary
.params
.iter()
.filter(|(_, p)| p.required)
.map(|(name, p)| format!(" --param {name}=<{}>", p.kind.as_str()))
.collect()
}
fn print_params(summary: &TemplateSummary) {
if summary.params.is_empty() {
println!("params: (none — this template takes no overrides)");
return;
}
println!("\nparams:");
for (name, p) in &summary.params {
let requirement = if p.required {
"required".to_string()
} else {
match &p.default {
Some(d) => format!("default {d}"),
None => "optional".to_string(),
}
};
println!(
" {:<20} {:<7} {}{}{}",
name,
p.kind.as_str(),
requirement,
if p.secret { " [secret]" } else { "" },
match &p.description {
Some(d) => format!(" — {d}"),
None => String::new(),
}
);
}
}
async fn list(args: TemplateListArgs) -> CliResult<()> {
let store = connect(&args.common).await?;
let templates = crate::templates::list_with_state(&store).await?;
if args.common.json {
println!(
"{}",
to_pretty(&serde_json::json!({ "templates": templates }))?
);
return Ok(());
}
if templates.is_empty() {
println!("no templates registered in this store — add one with `faucet template register`");
return Ok(());
}
println!(
"{:<26} {:<11} {:<6} {:<7} {:>6} DESCRIPTION",
"ID", "STATUS", "LIVE", "NEWEST", "PARAMS"
);
for t in &templates {
let (status, live, newest) = match &t.state {
Some(st) => (
st.status.to_string(),
st.stable.map(|v| format!("v{v}")).unwrap_or("—".into()),
st.newest.map(|v| format!("v{v}")).unwrap_or("—".into()),
),
None => ("?".into(), "?".into(), format!("v{}", t.version)),
};
println!(
"{:<26} {:<11} {:<6} {:<7} {:>6} {}",
t.id,
status,
live,
newest,
t.params.len(),
t.description.as_deref().unwrap_or("")
);
}
Ok(())
}
async fn fetch(store: &TemplateStore, id: &str, version: Option<u32>) -> CliResult<TemplateRecord> {
store
.template_get(id, version)
.await
.map_err(|e| CliError::Internal(format!("template registry read: {e}")))?
.ok_or_else(|| CliError::UnknownPipelineTemplate {
id: id.to_string(),
version,
})
}
async fn show(args: TemplateShowArgs) -> CliResult<()> {
let store = connect(&args.common).await?;
let selector = VersionSelector::parse(&args.version)?;
let want = crate::templates::resolve_version(&store, &args.id, selector).await?;
let record = fetch(&store, &args.id, Some(want)).await?;
if args.clean && !args.common.json {
print!("{}", crate::templates::clean_config_yaml(&record.body)?);
return Ok(());
}
let state = crate::templates::template_state(&store, &args.id).await?;
let launches = store
.template_launches(&args.id)
.await
.map_err(|e| CliError::Internal(format!("template launch read: {e}")))?;
if args.common.json {
println!(
"{}",
to_pretty(&serde_json::json!({
"template": record,
"state": state,
"is_stable": state.stable == Some(record.version),
"launches": launches,
}))?
);
return Ok(());
}
println!("template {} [{}]", record.id, state.status);
if let Some(name) = &record.name {
println!("name {name}");
}
if let Some(d) = &record.description {
println!("about {d}");
}
println!(
"created {}{}",
record.created_at.format("%Y-%m-%dT%H:%M:%SZ"),
match &record.created_by {
Some(p) => format!(" by {p}"),
None => String::new(),
}
);
println!(
"showing v{}{}",
record.version,
if state.stable == Some(record.version) {
" (live)"
} else {
""
}
);
println!("\nversions:");
for v in &state.versions {
let mut marks: Vec<String> = Vec::new();
if state.stable == Some(*v) {
marks.push("live".into());
}
if state.previous == Some(*v) {
marks.push("previous".into());
}
if state.newest == Some(*v) {
marks.push("newest".into());
}
marks.extend(
state
.tags
.iter()
.filter(|(_, pointed)| *pointed == v)
.map(|(t, _)| t.clone()),
);
println!(
" v{:<4} {}",
v,
if marks.is_empty() {
String::from("—")
} else {
marks.join(", ")
}
);
}
if let Some(d) = &state.deprecation {
println!(
"\ndeprecated {}{}",
d.deprecated_at.format("%Y-%m-%dT%H:%M:%SZ"),
match &d.reason {
Some(r) => format!(" — {r}"),
None => String::new(),
}
);
}
if !launches.is_empty() {
println!("\nlaunch history (newest first):");
for l in launches.iter().take(10) {
println!(
" v{:<4} {}{}",
l.version,
l.launched_at.format("%Y-%m-%dT%H:%M:%SZ"),
match &l.launched_by {
Some(by) => format!(" by {by}"),
None => String::new(),
}
);
}
}
print_params(&record.summary());
println!("\nconfig ({:?}, stored verbatim):", record.format);
for line in record.body.lines() {
println!(" {line}");
}
Ok(())
}
async fn delete(args: TemplateDeleteArgs) -> CliResult<()> {
let store = connect(&args.common).await?;
let pinned = match args.version.as_deref() {
None => None,
Some(raw) => Some(
crate::templates::resolve_version(&store, &args.id, VersionSelector::parse(raw)?)
.await?,
),
};
let removed = store
.template_delete(&args.id, pinned)
.await
.map_err(|e| CliError::Internal(format!("template registry write: {e}")))?;
if removed == 0 {
return Err(CliError::UnknownPipelineTemplate {
id: args.id.clone(),
version: pinned,
});
}
if args.common.json {
println!(
"{}",
to_pretty(&serde_json::json!({ "id": args.id, "deleted_versions": removed }))?
);
return Ok(());
}
println!("deleted {removed} version(s) of template '{}'", args.id);
Ok(())
}
fn report_launch(
id: &str,
outcome: &crate::templates::LaunchOutcome,
json: bool,
verb: &str,
) -> CliResult<()> {
if json {
println!(
"{}",
to_pretty(&serde_json::json!({
"id": id,
"version": outcome.version,
"replaced": outcome.replaced,
"already_launched": outcome.already_launched,
"first_launch": outcome.first_launch,
}))?
);
return Ok(());
}
if outcome.already_launched {
println!(
"template '{id}': v{} was already live — nothing changed",
outcome.version
);
return Ok(());
}
println!(
"template '{id}': {verb} v{}{}",
outcome.version,
match outcome.replaced {
Some(prev) => format!(" (was v{prev}; previous → v{prev})"),
None => String::from(" — first launch, template is now `launched`"),
}
);
Ok(())
}
async fn launch(args: TemplateLaunchArgs) -> CliResult<()> {
let store = connect(&args.common).await?;
let target = VersionSelector::parse(&args.version)?;
let outcome = crate::templates::launch(&store, &args.id, target, None).await?;
report_launch(&args.id, &outcome, args.common.json, "launched")
}
async fn rollback(args: TemplateRollbackArgs) -> CliResult<()> {
let store = connect(&args.common).await?;
let outcome = crate::templates::rollback(&store, &args.id, None).await?;
report_launch(&args.id, &outcome, args.common.json, "rolled back to")
}
async fn deprecate(args: TemplateDeprecateArgs) -> CliResult<()> {
let store = connect(&args.common).await?;
let status =
crate::templates::set_deprecated(&store, &args.id, args.reason.clone(), None, !args.undo)
.await?;
if args.common.json {
println!(
"{}",
to_pretty(&serde_json::json!({ "id": args.id, "status": status.as_str() }))?
);
return Ok(());
}
println!("template '{}' is now {status}", args.id);
if !args.undo {
println!(
" existing callers keep working (pinned runs and `stable` still resolve) but every \
trigger warns — use `faucet template delete` for a hard stop"
);
}
Ok(())
}
async fn promote(args: TemplatePromoteArgs) -> CliResult<()> {
let store = connect(&args.common).await?;
let tag = VersionChannel::parse(&args.tag)?;
let target = VersionSelector::parse(&args.version)?;
let version = crate::templates::promote(&store, &args.id, tag, target).await?;
if args.common.json {
println!(
"{}",
to_pretty(&serde_json::json!({
"id": args.id, "tag": tag.as_str(), "version": version,
}))?
);
return Ok(());
}
println!("template '{}': {tag} → v{version}", args.id);
Ok(())
}
async fn run_template(args: TemplateRunArgs) -> CliResult<()> {
let store = connect(&args.common).await?;
let supplied = crate::params::collect_cli_params(&args.param)?;
let env = crate::params::collect_env_overrides(&args.param_env)?;
let selector = VersionSelector::parse(&args.version)?;
let want = crate::templates::resolve_version(&store, &args.id, selector).await?;
let materialized = crate::templates::materialize(
&store,
&args.id,
want,
&supplied,
&env,
crate::templates::Materialize::Local,
)
.await?;
tracing::info!(
template = %materialized.template_id,
version = materialized.version,
"materialized pipeline template"
);
let doc: serde_json::Value = serde_json::from_str(&materialized.body)
.map_err(|e| CliError::Internal(format!("re-parsing materialized template: {e}")))?;
let mut cfg = crate::config::PipelineConfig::from_value(doc)?;
crate::secrets::resolve_secrets(&mut cfg).await?;
if args.dry_run && args.common.json {
println!("{}", to_pretty(&cfg)?);
return Ok(());
}
let run_args = crate::cli::RunArgs {
dry_run: args.dry_run,
limit: args.limit,
no_env_file: true,
..Default::default()
};
crate::commands::run::execute(cfg, run_args, None).await
}
#[cfg(test)]
mod tests {
use super::*;
use crate::cli::TemplateStoreArgs;
#[test]
fn clean_config_strips_comments_and_preserves_params() {
let body = "\
version: 1 # trailing comment
# a top-level comment
name: orders
pipeline:
source: { type: csv, config: { path: \"${param.p}\" } } # inline
sink: { type: jsonl, config: { path: ./out.jsonl } }
";
let out = crate::templates::clean_config_yaml(body).unwrap();
assert!(!out.contains('#'), "comments must be stripped: {out}");
assert!(
out.contains("${param.p}"),
"param token must survive: {out}"
);
let before: serde_yaml::Value = serde_yaml::from_str(body).unwrap();
let after: serde_yaml::Value = serde_yaml::from_str(&out).unwrap();
assert_eq!(before, after, "clean output must parse to the same config");
}
#[test]
fn clean_config_normalizes_json_body_to_yaml() {
let body = r#"{"version":1,"name":"j","pipeline":{"source":{"type":"csv"}}}"#;
let out = crate::templates::clean_config_yaml(body).unwrap();
assert!(out.contains("version: 1"), "should be YAML now: {out}");
assert!(
!out.contains('{'),
"no JSON braces in canonical YAML: {out}"
);
}
fn common(store: &str, json: bool) -> TemplateStoreArgs {
TemplateStoreArgs {
store: store.to_string(),
env_file: None,
no_env_file: true,
json,
}
}
const BODY: &str = "\
version: 1
name: cli-tpl
params:
tag: { required: true, description: Output tag }
page: { type: int, default: 5 }
pipeline:
source:
type: csv
config:
path: IN_PATH
sink:
type: jsonl
config:
path: OUT_PATH
";
#[cfg(feature = "serve-history-sqlite")]
#[tokio::test]
async fn register_launch_promote_run_round_trip() {
let dir = tempfile::tempdir().unwrap();
let input = dir.path().join("in.csv");
std::fs::write(&input, "id,name\n1,alice\n2,bob\n").unwrap();
let output = dir.path().join("out.jsonl");
let cfg_path = dir.path().join("tpl.yaml");
std::fs::write(
&cfg_path,
BODY.replace("IN_PATH", &input.display().to_string())
.replace("OUT_PATH", &output.display().to_string()),
)
.unwrap();
let store = format!("sqlite:{}", dir.path().join("registry.db").display());
let reg = |launch: bool, tag: Vec<String>| TemplateRegisterArgs {
config: cfg_path.clone(),
id: None,
description: Some("round trip".into()),
tag,
launch,
common: common(&store, false),
};
register(reg(false, vec!["dev".into()]))
.await
.expect("register v1");
list(TemplateListArgs {
common: common(&store, true),
})
.await
.expect("list");
let err = run_template(TemplateRunArgs {
id: "cli-tpl".into(),
version: "stable".into(),
param: vec!["tag=alpha".into()],
param_env: vec![],
dry_run: true,
limit: None,
common: common(&store, false),
})
.await
.unwrap_err()
.to_string();
assert!(err.contains("no launched version"), "{err}");
launch(TemplateLaunchArgs {
id: "cli-tpl".into(),
version: "newest".into(),
common: common(&store, false),
})
.await
.expect("launch");
run_template(TemplateRunArgs {
id: "cli-tpl".into(),
version: "stable".into(),
param: vec!["tag=alpha".into()],
param_env: vec![],
dry_run: false,
limit: None,
common: common(&store, false),
})
.await
.expect("run");
assert_eq!(
std::fs::read_to_string(&output).unwrap().lines().count(),
2,
"the launched version's pipeline wrote both records"
);
register(reg(false, vec![])).await.expect("register v2");
show(TemplateShowArgs {
id: "cli-tpl".into(),
version: "stable".into(),
clean: false,
common: common(&store, false),
})
.await
.expect("show");
promote(TemplatePromoteArgs {
id: "cli-tpl".into(),
tag: "pre-prod".into(),
version: "newest".into(),
common: common(&store, false),
})
.await
.expect("promote");
launch(TemplateLaunchArgs {
id: "cli-tpl".into(),
version: "pre-prod".into(),
common: common(&store, true),
})
.await
.expect("launch from channel");
rollback(TemplateRollbackArgs {
id: "cli-tpl".into(),
common: common(&store, false),
})
.await
.expect("rollback");
for tag in ["stable", "previous", "newest", "prd", "latest"] {
assert!(
promote(TemplatePromoteArgs {
id: "cli-tpl".into(),
tag: tag.into(),
version: "1".into(),
common: common(&store, false),
})
.await
.is_err(),
"`{tag}` must not be promotable"
);
}
deprecate(TemplateDeprecateArgs {
id: "cli-tpl".into(),
reason: Some("superseded".into()),
undo: false,
common: common(&store, false),
})
.await
.expect("deprecate");
deprecate(TemplateDeprecateArgs {
id: "cli-tpl".into(),
reason: None,
undo: true,
common: common(&store, true),
})
.await
.expect("undeprecate");
delete(TemplateDeleteArgs {
id: "cli-tpl".into(),
version: Some("newest".into()),
common: common(&store, false),
})
.await
.expect("delete newest");
delete(TemplateDeleteArgs {
id: "cli-tpl".into(),
version: None,
common: common(&store, true),
})
.await
.expect("delete all");
let err = delete(TemplateDeleteArgs {
id: "cli-tpl".into(),
version: None,
common: common(&store, false),
})
.await
.unwrap_err();
assert!(
matches!(err, CliError::UnknownPipelineTemplate { .. }),
"{err:?}"
);
}
#[tokio::test]
async fn show_and_run_report_an_unknown_template() {
let c = common("memory", false);
let store = connect(&c).await.unwrap();
let err = fetch(&store, "nope", None).await.unwrap_err();
assert!(
matches!(err, CliError::UnknownPipelineTemplate { ref id, .. } if id == "nope"),
"{err:?}"
);
let err = promote(TemplatePromoteArgs {
id: "nope".into(),
tag: "prod".into(),
version: "newest".into(),
common: common("memory", false),
})
.await
.unwrap_err();
assert!(
matches!(err, CliError::UnknownPipelineTemplate { .. }),
"{err:?}"
);
}
#[test]
fn format_is_taken_from_the_extension() {
assert_eq!(
format_of(std::path::Path::new("a.yaml")).unwrap(),
ConfigFormat::Yaml
);
assert_eq!(
format_of(std::path::Path::new("a.YML")).unwrap(),
ConfigFormat::Yaml
);
assert_eq!(
format_of(std::path::Path::new("a.json")).unwrap(),
ConfigFormat::Json
);
assert!(format_of(std::path::Path::new("a.toml")).is_err());
assert!(format_of(std::path::Path::new("a")).is_err());
}
#[test]
fn required_param_hint_lists_only_required_params() {
let mut params = crate::params::ParamsSpec::new();
params.insert(
"tag".into(),
crate::params::ParamSpec {
kind: crate::params::ParamType::String,
required: true,
default: None,
secret: false,
description: None,
computed: None,
},
);
params.insert(
"page".into(),
crate::params::ParamSpec {
kind: crate::params::ParamType::Int,
required: false,
default: Some(serde_json::json!(5)),
secret: false,
computed: None,
description: None,
},
);
let summary = TemplateSummary {
state: None,
id: "t".into(),
version: 1,
name: None,
description: None,
params,
created_at: chrono::Utc::now(),
created_by: None,
};
let hint = required_param_hint(&summary);
assert_eq!(hint, " --param tag=<string>");
print_params(&summary);
}
#[tokio::test]
async fn register_rejects_a_bad_extension_and_a_missing_file() {
let dir = tempfile::tempdir().unwrap();
let bad = dir.path().join("cfg.toml");
std::fs::write(&bad, "x = 1").unwrap();
let err = register(TemplateRegisterArgs {
config: bad,
id: None,
description: None,
tag: vec![],
launch: false,
common: common("memory", false),
})
.await
.unwrap_err();
assert!(matches!(err, CliError::UnknownExtension { .. }), "{err:?}");
let err = register(TemplateRegisterArgs {
config: dir.path().join("nope.yaml"),
id: None,
description: None,
tag: vec![],
launch: false,
common: common("memory", false),
})
.await
.unwrap_err()
.to_string();
assert!(err.contains("reading template config"), "{err}");
}
#[tokio::test]
async fn list_reports_an_empty_store() {
list(TemplateListArgs {
common: common("memory", false),
})
.await
.expect("empty list is not an error");
}
}