faucet_cli/templates/

store.rs

//! Registration + materialization of pipeline templates (#444).
//!
//! Pure orchestration over [`crate::serve::history::RunHistory`]'s template
//! methods and [`crate::params`]; no HTTP, no clap, no MCP shapes — the three
//! front-ends are thin adapters over the two entry points here.

use crate::error::{CliError, CliResult};
use crate::params::{self, BindMode, SuppliedParams};
use crate::serve::config::HistoryBackendSpec;
use crate::serve::history::templates::{
    DeprecationRecord, TemplateDraft, TemplateId, TemplateRecord, TemplateState, TemplateStatus,
    TemplateSummary, VersionChannel, VersionSelector,
};
use crate::serve::history::{self, RunHistory};
use crate::serve::load::ConfigFormat;
use serde_json::Value;
use std::collections::BTreeMap;
use std::sync::Arc;
use std::time::Duration;

/// The registry handle. Any `RunHistory` backend will do — `faucet serve` passes
/// its own `--history` store so templates live beside run records; the CLI
/// connects one from `--store` / the config's `catalog:` block.
pub type TemplateStore = Arc<dyn RunHistory>;

/// A registration, before validation.
#[derive(Debug, Clone)]
pub struct RegisterRequest {
    /// Explicit id. When `None` the id is derived from the config's `name:`.
    pub id: Option<String>,
    /// The config document, stored verbatim.
    pub body: String,
    pub format: ConfigFormat,
    /// Free-text description (falls back to nothing).
    pub description: Option<String>,
    /// Named environment channels to point at the newly registered version. The
    /// version number itself always auto-increments; these are the human-facing
    /// pointers (`dev`, `pre-prod`, …) moved onto it in the same step. Derived
    /// channels are rejected.
    pub tags: Vec<VersionChannel>,
    /// Launch the newly registered version immediately, making it `stable`.
    /// Without this a register is inert — a new build never moves existing
    /// callers, which is the point of the model — so this is the explicit
    /// "register and go live" shortcut.
    pub launch: bool,
    /// Principal performing the registration, for provenance.
    pub created_by: Option<String>,
}

/// A template rendered for one trigger: a config document with every
/// `${param.*}` bound, ready to hand to the ordinary run path.
#[derive(Debug, Clone)]
pub struct MaterializedConfig {
    pub template_id: String,
    pub version: u32,
    /// The config's own `name:`, for the run record.
    pub name: Option<String>,
    /// JSON config document (params bound). JSON regardless of how the template
    /// was registered — one canonical hand-off shape for the run path.
    pub body: String,
    /// Bound param values with `secret: true` entries replaced by `"***"` — the
    /// only form safe to echo, audit, or persist.
    pub params_redacted: BTreeMap<String, Value>,
    /// True when at least one bound param was declared `secret: true`.
    pub used_secret_params: bool,
}

impl MaterializedConfig {
    /// Wire format of [`Self::body`]. Always JSON.
    pub fn format(&self) -> ConfigFormat {
        ConfigFormat::Json
    }
}

/// Parse a config document by declared format into an untyped value.
fn parse_body(body: &str, format: ConfigFormat) -> CliResult<Value> {
    match format {
        ConfigFormat::Yaml => {
            serde_yaml::from_str(body).map_err(|e| CliError::Config(format!("invalid YAML: {e}")))
        }
        ConfigFormat::Json => {
            serde_json::from_str(body).map_err(|e| CliError::Config(format!("invalid JSON: {e}")))
        }
    }
}

/// Validate a submitted config and append it as a new template version.
///
/// Validation deliberately runs against a **placeholder binding**: required
/// params have no value at registration time, so each is filled with a
/// type-shaped stand-in and the config is then taken through the real
/// `PipelineConfig` parse plus `expand` (matrix mode) or
/// [`crate::topology::validate_topology_spec`] (topology mode). That checks
/// everything structural — grammar, named templates, the matrix graph
/// (parent/`depends_on` cycles, duplicate state keys), the exactly-once and
/// write-mode gates, edge endpoints — without resolving a single secret or
/// constructing a single connector. Node arity in topology mode is validated
/// when the graph is built, i.e. at trigger time, because building it requires
/// live connectors that a placeholder-bound config must not create.
pub async fn register(store: &TemplateStore, req: RegisterRequest) -> CliResult<TemplateRecord> {
    let mut doc = parse_body(&req.body, req.format)?;
    if !doc.is_object() {
        return Err(CliError::Config(
            "a pipeline template must be a config document (a YAML/JSON mapping)".into(),
        ));
    }

    // The declared trigger surface, validated and stored alongside the body so
    // callers can discover it without re-parsing.
    let declared = params::declared(&doc)?;

    // Structural validation on a placeholder-bound copy. `${env:…}` and secret
    // directives are left untouched — registration must never read the server's
    // secrets, and the body we persist is the one that was submitted.
    let mut probe = doc.clone();
    params::bind_document(&mut probe, &SuppliedParams::new(), BindMode::Placeholder)?;
    let cfg = crate::config::PipelineConfig::from_value(probe)?;
    if crate::topology::is_topology(&cfg) {
        crate::topology::validate_topology_spec(&cfg)?;
    } else {
        // Compile each row's transform chain too — `expand` only checks an entry's
        // shape, so without this a template with a misspelled transform field
        // registers cleanly and fails at trigger time instead.
        for node in crate::expand::expand(&cfg)? {
            if node.transforms.is_empty() {
                continue;
            }
            crate::transforms::compile_transforms(&node.transforms)
                .map_err(|e| CliError::Config(format!("row '{}': {e}", node.id)))?;
        }
    }

    let id = match &req.id {
        Some(raw) => TemplateId::parse(raw)?,
        None => {
            let name = cfg.name.as_deref().ok_or_else(|| {
                CliError::Config(
                    "no template id given and the config has no `name:` to derive one from — \
                     pass an explicit id"
                        .into(),
                )
            })?;
            TemplateId::from_config_name(name)?
        }
    };

    // Keep the persisted body byte-identical to what was submitted.
    let _ = &mut doc;

    // Reject a derived channel before writing anything, so a bad request never
    // leaves a half-registered version behind.
    for tag in &req.tags {
        reject_derived(*tag)?;
    }

    // A description describes the *template*, not the build, so carry the previous
    // version's forward when the caller omits one. Without this, a deploy that
    // re-registers without `--description` blanks the listing for everybody.
    let description = match &req.description {
        Some(d) => Some(d.clone()),
        None => store
            .template_get(id.as_str(), None)
            .await
            .map_err(|e| CliError::Internal(format!("template registry read: {e}")))?
            .and_then(|prev| prev.description),
    };

    let draft = TemplateDraft {
        id,
        name: cfg.name.clone(),
        description,
        body: req.body.clone(),
        format: req.format,
        params: declared,
        created_by: req.created_by.clone(),
    };
    let record = store
        .template_register(&draft)
        .await
        .map_err(|e| CliError::Internal(format!("template registry write: {e}")))?;

    // Point the requested channels at the version just created.
    for tag in &req.tags {
        store
            .template_set_tag(&record.id, tag.as_str(), record.version)
            .await
            .map_err(|e| CliError::Internal(format!("template channel write: {e}")))?;
    }
    // `--launch` is the only way a register makes a version live.
    if req.launch {
        store
            .template_launch(&record.id, record.version, req.created_by.as_deref())
            .await
            .map_err(|e| CliError::Internal(format!("template launch write: {e}")))?;
    }
    Ok(record)
}

/// `latest` is computed from the version list, so promoting or deleting it makes
/// no sense — say so instead of silently no-oping.
fn reject_derived(tag: VersionChannel) -> CliResult<()> {
    if tag.is_derived() {
        let how = match tag {
            VersionChannel::Stable => " — move it with `faucet template launch` instead",
            VersionChannel::Previous => " — it is whatever was launched before the current version",
            _ => " — it is always the highest version number",
        };
        return Err(CliError::Config(format!(
            "`{tag}` is a derived channel and cannot be promoted{how}. Promotable channels: {}",
            VersionChannel::ASSIGNABLE
                .iter()
                .map(|c| c.as_str())
                .collect::<Vec<_>>()
                .join(", ")
        )));
    }
    Ok(())
}

/// Resolve a [`VersionSelector`] to the exact version to act on.
///
/// **Every** channel — derived or assigned — is looked up here; nothing falls back
/// to "the newest build". A selector that names an unset channel is an error
/// listing what *is* set, because silently substituting another version is how a
/// caller ends up running code they did not ask for.
pub async fn resolve_version(
    store: &TemplateStore,
    id: &str,
    selector: VersionSelector,
) -> CliResult<u32> {
    if let VersionSelector::Pinned(n) = selector {
        return Ok(n);
    }
    let channel = selector
        .channel()
        .expect("non-pinned selector names a channel");
    let state = template_state(store, id).await?;
    if state.versions.is_empty() {
        return Err(CliError::UnknownPipelineTemplate {
            id: id.to_string(),
            version: None,
        });
    }
    state
        .derived(channel)
        .ok_or_else(|| unresolved_channel(id, channel, &state))
}

/// The error for a selector that names a channel with nothing behind it. Phrased
/// per channel, because the fix differs: `stable` needs a *launch*, `previous`
/// needs a second launch, an environment channel needs a *promote*.
fn unresolved_channel(id: &str, channel: VersionChannel, state: &TemplateState) -> CliError {
    let newest = state
        .newest
        .map(|v| v.to_string())
        .unwrap_or_else(|| "1".into());
    match channel {
        // Phrased for both audiences — the same error surfaces on the CLI, over
        // HTTP, and in the console's versions page.
        VersionChannel::Stable => CliError::Config(format!(
            "template '{id}' has no launched version (status: {}). Launch one first \
             (`faucet template launch {id} --version {newest}`, or \
             `POST /v1/templates/{id}/launch`), or select a specific build with \
             `newest` / a version number",
            state.status
        )),
        VersionChannel::Previous => CliError::Config(format!(
            "template '{id}' has no previous version — {}. `previous` is the version launched \
             before the current one, so it only exists after a second launch",
            match state.stable {
                Some(v) => format!("v{v} is the first and only launched version"),
                None => "nothing has been launched yet".to_string(),
            }
        )),
        // `newest` is unreachable here (a template with versions always has one),
        // so this arm only guards a future channel gaining derived status.
        VersionChannel::Newest => {
            CliError::Config(format!("template '{id}' has no versions registered"))
        }
        assigned => CliError::Config(format!(
            "template '{id}' has no `{assigned}` version. Channels currently set: {}. Promote one \
             with `faucet template promote {id} --tag {assigned} --version <n>`",
            if state.tags.is_empty() {
                String::from("(none)")
            } else {
                state
                    .tags
                    .iter()
                    .map(|(t, v)| format!("{t}=v{v}"))
                    .collect::<Vec<_>>()
                    .join(", ")
            }
        )),
    }
}

/// Every registered template's latest version, each carrying its release state.
///
/// One extra read per template — the registry is a small, human-curated set, and
/// assembling the state per row keeps the list and detail views consistent by
/// construction rather than by convention.
pub async fn list_with_state(store: &TemplateStore) -> CliResult<Vec<TemplateSummary>> {
    let mut out = store
        .template_list()
        .await
        .map_err(|e| CliError::Internal(format!("template registry read: {e}")))?;
    for summary in &mut out {
        summary.state = Some(template_state(store, &summary.id).await?);
    }
    Ok(out)
}

/// The template's full release state (status, `stable` / `previous` / `newest`,
/// channel pointers). Errors only if the registry itself is unreadable.
pub async fn template_state(store: &TemplateStore, id: &str) -> CliResult<TemplateState> {
    store
        .template_state(id)
        .await
        .map_err(|e| CliError::Internal(format!("template registry read: {e}")))
}

/// Confirm a version exists, returning a typed error naming it if not.
async fn require_version(store: &TemplateStore, id: &str, version: u32) -> CliResult<()> {
    if store
        .template_get(id, Some(version))
        .await
        .map_err(|e| CliError::Internal(format!("template registry read: {e}")))?
        .is_none()
    {
        return Err(CliError::UnknownPipelineTemplate {
            id: id.to_string(),
            version: Some(version),
        });
    }
    Ok(())
}

/// Point a named environment channel at a version, moving it if already set.
///
/// The target is itself a selector, so `--tag prod --version stable` promotes
/// whatever is currently launched — the "use the version I already blessed" case —
/// and `--version 3` pins an exact build. Derived channels (`stable`, `previous`,
/// `newest`) are not valid *targets*: `stable` moves via [`launch`], and the other
/// two are computed.
pub async fn promote(
    store: &TemplateStore,
    id: &str,
    tag: VersionChannel,
    target: VersionSelector,
) -> CliResult<u32> {
    reject_derived(tag)?;
    let version = resolve_version(store, id, target).await?;
    require_version(store, id, version).await?;
    store
        .template_set_tag(id, tag.as_str(), version)
        .await
        .map_err(|e| CliError::Internal(format!("template channel write: {e}")))?;
    Ok(version)
}

/// The outcome of a [`launch`]: which version is now live, and what it replaced.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LaunchOutcome {
    /// The version now launched (`stable`).
    pub version: u32,
    /// The version it replaced — the new `previous`. `None` on a first launch.
    pub replaced: Option<u32>,
    /// True when the requested version was already launched, so nothing changed.
    pub already_launched: bool,
    /// Whether this launch flipped the template out of `draft`.
    pub first_launch: bool,
}

/// **Launch** a version: make it `stable`, so unpinned callers start using it.
///
/// This is the one deliberate act that moves consumers. Registering a build never
/// does — that is the whole point of the model, so a nightly can land without
/// dragging anyone along.
///
/// Refuses to launch while the template is deprecated: reviving a retired template
/// by moving its live pointer is almost certainly a mistake, and `--undo` makes the
/// intent explicit.
pub async fn launch(
    store: &TemplateStore,
    id: &str,
    target: VersionSelector,
    launched_by: Option<&str>,
) -> CliResult<LaunchOutcome> {
    let version = resolve_version(store, id, target).await?;
    require_version(store, id, version).await?;
    let before = template_state(store, id).await?;
    if before.status == TemplateStatus::Deprecated {
        return Err(CliError::Config(format!(
            "template '{id}' is deprecated — un-deprecate it first with \
             `faucet template deprecate {id} --undo`, then launch"
        )));
    }
    let seq = store
        .template_launch(id, version, launched_by)
        .await
        .map_err(|e| CliError::Internal(format!("template launch write: {e}")))?;
    Ok(LaunchOutcome {
        version,
        replaced: before.stable,
        already_launched: seq.is_none(),
        first_launch: before.stable.is_none(),
    })
}

/// Roll back to the previously launched version — `launch` of `previous`, named
/// for the thing you actually want to find under pressure.
pub async fn rollback(
    store: &TemplateStore,
    id: &str,
    launched_by: Option<&str>,
) -> CliResult<LaunchOutcome> {
    launch(
        store,
        id,
        VersionSelector::Channel(VersionChannel::Previous),
        launched_by,
    )
    .await
}

/// Retire (`Some`) or revive (`None`) a template.
///
/// Deprecation is **template-wide**, not per version: a build that should not be
/// used simply never gets launched (or gets deleted). A deprecated template keeps
/// serving callers who pin or ride `stable` — retiring must not hard-break
/// them — but every trigger warns and listings mark it. Returns the resulting
/// status.
pub async fn set_deprecated(
    store: &TemplateStore,
    id: &str,
    reason: Option<String>,
    by: Option<&str>,
    deprecated: bool,
) -> CliResult<TemplateStatus> {
    let state = template_state(store, id).await?;
    if state.versions.is_empty() {
        return Err(CliError::UnknownPipelineTemplate {
            id: id.to_string(),
            version: None,
        });
    }
    let record = deprecated.then(|| DeprecationRecord {
        deprecated_at: chrono::Utc::now(),
        deprecated_by: by.map(str::to_string),
        reason,
    });
    store
        .template_set_deprecation(id, record.as_ref())
        .await
        .map_err(|e| CliError::Internal(format!("template deprecation write: {e}")))?;
    Ok(TemplateStatus::derive(state.stable.is_some(), deprecated))
}

/// Where the materialized config is going — which decides whether load-time
/// directives may be resolved here.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Materialize {
    /// The body is executed by *this* process and never stored. Load-time
    /// directives (`${env:}` / `${file:}` / `${secret:}`) are resolved here, so a
    /// caller-supplied `env` overlay takes effect.
    Local,
    /// The body will be **persisted** for another instance to execute (a
    /// clustered submit). Load-time directives are left as tokens for the
    /// executing instance to resolve, so a resolved credential is never written
    /// to the shared run-history database (#456 C5).
    Persisted,
}

/// Fetch a template version and bind the supplied params into a runnable config
/// document.
///
/// Ordering mirrors the file-load path: `${env:}` / `${file:}` / `${secret:}`
/// resolve **first** (with `env_overrides` taking precedence over the process
/// environment), then `${param.*}` binds. A supplied param value is therefore
/// never itself scanned for directives, so a caller cannot use a param to read
/// the server's environment or secret store.
///
/// Under [`Materialize::Persisted`] the first step is **skipped**: the directives
/// stay as tokens and are resolved later by `load_submission` on whichever
/// instance runs the job. Resolving them here would serialise the *values* into
/// the body that gets stored in the shared database — which is how a
/// `${env:DB_PASSWORD}` in a template body ended up in plaintext there. The
/// trade-off is that a typed (`int`/`float`/`bool`) param whose `default` is
/// itself a directive cannot be coerced in this mode; it fails loudly at trigger
/// time naming the param, rather than silently.
pub async fn materialize(
    store: &TemplateStore,
    id: &str,
    version: u32,
    supplied: &SuppliedParams,
    env_overrides: &BTreeMap<String, String>,
    mode: Materialize,
) -> CliResult<MaterializedConfig> {
    // Takes a concrete version, never an `Option`: "no version given" is resolved
    // by `resolve_version` against the registry, so there is no code path where a
    // `None` here could quietly mean "the newest build".
    let record = store
        .template_get(id, Some(version))
        .await
        .map_err(|e| CliError::Internal(format!("template registry read: {e}")))?
        .ok_or_else(|| CliError::UnknownPipelineTemplate {
            id: id.to_string(),
            version: Some(version),
        })?;

    let mut doc = parse_body(&record.body, record.format)?;
    if mode == Materialize::Local {
        let overlay: crate::interpolate::EnvOverlay = env_overrides
            .iter()
            .map(|(k, v)| (k.clone(), v.clone()))
            .collect();
        crate::interpolate::interpolate_value_with_env(&mut doc, &overlay)?;
    }
    let bound = params::bind_document(&mut doc, supplied, BindMode::Strict)?;
    // Drop the declaration block: materialization is the moment params cease to
    // exist. Leaving it would make any later load re-run the bind pass with no
    // supplied values and reject the config for a "missing" required param —
    // and the param surface is already recorded on the template and echoed to
    // the caller, so nothing is lost.
    if let Some(map) = doc.as_object_mut() {
        map.remove(params::PARAMS_KEY);
    }

    let body = serde_json::to_string(&doc)
        .map_err(|e| CliError::Internal(format!("re-serializing template body: {e}")))?;
    Ok(MaterializedConfig {
        template_id: record.id.clone(),
        version: record.version,
        name: record.name.clone(),
        body,
        params_redacted: bound.redacted(),
        used_secret_params: bound.has_secrets(),
    })
}

/// Connect a template store from a URL: `memory`, `sqlite:<path>`, or a
/// `postgres://…` URL. Same grammar (and same build-feature requirements) as
/// `catalog.url` and `faucet serve --history`, so one store can hold run
/// history, the dataset catalog, and the template registry together.
pub async fn resolve_store_url(url: &str) -> CliResult<TemplateStore> {
    let backend = match url {
        "memory" => HistoryBackendSpec::Memory,
        u if u.starts_with("postgres://") || u.starts_with("postgresql://") => {
            HistoryBackendSpec::Postgres(u.to_string())
        }
        u if u.starts_with("sqlite:") => HistoryBackendSpec::Sqlite(u.to_string()),
        other => {
            return Err(CliError::Config(format!(
                "template store '{other}' is not recognised — expected 'memory', \
                 'sqlite:<path>', or a 'postgres://…' URL"
            )));
        }
    };
    history::connect(
        &backend,
        // Idempotency claims and run leases are run-history concerns; a
        // template-only connection never uses them.
        Duration::from_secs(3600),
        Duration::from_secs(30),
        &uuid::Uuid::now_v7().to_string(),
    )
    .await
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::serve::history::memory::MemoryHistory;
    use serde_json::json;

    fn store() -> TemplateStore {
        Arc::new(MemoryHistory::new(Duration::from_secs(60))) as TemplateStore
    }

    const PARAMETERIZED: &str = "\
version: 1
name: tenant-sync
params:
  tenant_id: { required: true, description: Tenant to sync }
  since: { default: \"1970-01-01\" }
  page: { type: int, default: 100 }
pipeline:
  source:
    type: rest
    config:
      url: \"https://api.example.com/${param.tenant_id}/events?since=${param.since}\"
  sink:
    type: jsonl
    config:
      path: ./out.jsonl
";

    fn req(body: &str) -> RegisterRequest {
        RegisterRequest {
            id: None,
            body: body.to_string(),
            format: ConfigFormat::Yaml,
            description: Some("test".into()),
            tags: Vec::new(),
            launch: false,
            created_by: Some("tester".into()),
        }
    }

    /// Register + launch in one step, for tests that only care about the result.
    fn req_launched(body: &str) -> RegisterRequest {
        RegisterRequest {
            launch: true,
            ..req(body)
        }
    }

    #[tokio::test]
    async fn registers_and_versions() {
        let s = store();
        let first = register(&s, req(PARAMETERIZED)).await.unwrap();
        assert_eq!(first.id, "tenant-sync");
        assert_eq!(first.version, 1);
        assert_eq!(first.created_by.as_deref(), Some("tester"));
        assert!(first.params["tenant_id"].required);
        assert_eq!(first.params["page"].default, Some(json!(100)));
        assert_eq!(first.body, PARAMETERIZED, "body stored verbatim");

        let second = register(&s, req(PARAMETERIZED)).await.unwrap();
        assert_eq!(second.version, 2);
        assert_eq!(
            s.template_versions("tenant-sync").await.unwrap(),
            vec![2, 1]
        );
        let listed = list_with_state(&s).await.unwrap();
        assert_eq!(listed.len(), 1, "list folds to one row per id");
        // A register is inert: two versions exist, nothing is live.
        let st = listed[0].state.as_ref().unwrap();
        assert_eq!(st.status, TemplateStatus::Draft);
        assert_eq!(st.newest, Some(2));
        assert_eq!(st.stable, None);
    }

    #[tokio::test]
    async fn register_compiles_transforms_not_just_their_shape() {
        let s = store();
        // `set` takes `values:`; `fields:` is a plausible typo that used to
        // register cleanly and then fail at trigger time.
        let mut bad = req(r#"
version: 1
name: tenant-sync
pipeline:
  source: { type: rest, config: {} }
  transforms:
    - type: set
      config: { fields: { a: 1 } }
  sink: { type: jsonl, config: { path: ./o.jsonl } }
"#);
        bad.description = None;
        let err = register(&s, bad).await.unwrap_err().to_string();
        assert!(err.contains("values"), "names the missing field: {err}");
        assert!(
            s.template_versions("tenant-sync").await.unwrap().is_empty(),
            "nothing is persisted when validation fails"
        );
    }

    #[tokio::test]
    async fn a_description_carries_forward_across_registers() {
        let s = store();
        let first = register(&s, req(PARAMETERIZED)).await.unwrap();
        assert_eq!(first.description.as_deref(), Some("test"));

        // A deploy that re-registers without `--description` must not blank it.
        let mut bare = req(PARAMETERIZED);
        bare.description = None;
        let second = register(&s, bare).await.unwrap();
        assert_eq!(second.description.as_deref(), Some("test"));

        // An explicit description still wins.
        let mut changed = req(PARAMETERIZED);
        changed.description = Some("now something else".into());
        let third = register(&s, changed).await.unwrap();
        assert_eq!(third.description.as_deref(), Some("now something else"));

        // …and the new one is what the next bare register inherits.
        let mut bare2 = req(PARAMETERIZED);
        bare2.description = None;
        let fourth = register(&s, bare2).await.unwrap();
        assert_eq!(fourth.description.as_deref(), Some("now something else"));
    }

    #[tokio::test]
    async fn a_register_never_moves_existing_callers() {
        let s = store();
        register(&s, req_launched(PARAMETERIZED)).await.unwrap(); // v1, launched
        assert_eq!(
            resolve_version(&s, "tenant-sync", VersionSelector::stable())
                .await
                .unwrap(),
            1
        );

        // A nightly lands as v2 — `stable` must not budge. This is the property
        // the whole model exists for.
        register(&s, req(PARAMETERIZED)).await.unwrap();
        assert_eq!(
            resolve_version(&s, "tenant-sync", VersionSelector::stable())
                .await
                .unwrap(),
            1,
            "registering a build must not move the launched version"
        );
        assert_eq!(
            resolve_version(&s, "tenant-sync", VersionSelector::newest())
                .await
                .unwrap(),
            2,
            "`newest` is how you reach the un-launched build"
        );

        // Launching is the deliberate act that moves them.
        let out = launch(&s, "tenant-sync", VersionSelector::newest(), Some("alice"))
            .await
            .unwrap();
        assert_eq!((out.version, out.replaced), (2, Some(1)));
        assert!(!out.first_launch);
        assert_eq!(
            resolve_version(&s, "tenant-sync", VersionSelector::stable())
                .await
                .unwrap(),
            2
        );
        // `previous` is now the version launched before it.
        assert_eq!(
            resolve_version(
                &s,
                "tenant-sync",
                VersionSelector::Channel(VersionChannel::Previous)
            )
            .await
            .unwrap(),
            1
        );
    }

    #[tokio::test]
    async fn draft_template_has_no_stable_and_says_how_to_fix_it() {
        let s = store();
        register(&s, req(PARAMETERIZED)).await.unwrap();
        let state = template_state(&s, "tenant-sync").await.unwrap();
        assert_eq!(state.status, TemplateStatus::Draft);

        // Unpinned resolution fails with the exact command to run — never a
        // silent fallback to the newest build.
        let err = resolve_version(&s, "tenant-sync", VersionSelector::stable())
            .await
            .unwrap_err()
            .to_string();
        assert!(err.contains("no launched version"), "{err}");
        assert!(err.contains("faucet template launch"), "{err}");
        // But explicit selectors work, so a draft is fully testable.
        assert_eq!(
            resolve_version(&s, "tenant-sync", VersionSelector::newest())
                .await
                .unwrap(),
            1
        );
        assert_eq!(
            resolve_version(&s, "tenant-sync", VersionSelector::Pinned(1))
                .await
                .unwrap(),
            1
        );
    }

    #[tokio::test]
    async fn first_launch_flips_status_and_relaunch_is_a_noop() {
        let s = store();
        register(&s, req(PARAMETERIZED)).await.unwrap();
        let out = launch(&s, "tenant-sync", VersionSelector::Pinned(1), None)
            .await
            .unwrap();
        assert!(out.first_launch);
        assert_eq!(out.replaced, None);
        assert_eq!(
            template_state(&s, "tenant-sync").await.unwrap().status,
            TemplateStatus::Launched
        );

        // Re-launching what is already live changes nothing — and crucially does
        // not append, which would make `previous` a duplicate of `stable`.
        let again = launch(&s, "tenant-sync", VersionSelector::Pinned(1), None)
            .await
            .unwrap();
        assert!(again.already_launched);
        assert_eq!(s.template_launches("tenant-sync").await.unwrap().len(), 1);
        let err = resolve_version(
            &s,
            "tenant-sync",
            VersionSelector::Channel(VersionChannel::Previous),
        )
        .await
        .unwrap_err()
        .to_string();
        assert!(err.contains("no previous version"), "{err}");
    }

    #[tokio::test]
    async fn rollback_returns_to_the_prior_launch() {
        let s = store();
        for _ in 0..3 {
            register(&s, req(PARAMETERIZED)).await.unwrap();
        }
        launch(&s, "tenant-sync", VersionSelector::Pinned(1), None)
            .await
            .unwrap();
        launch(&s, "tenant-sync", VersionSelector::Pinned(3), None)
            .await
            .unwrap();

        let out = rollback(&s, "tenant-sync", Some("oncall")).await.unwrap();
        assert_eq!(out.version, 1, "rollback re-launches `previous`");
        assert_eq!(out.replaced, Some(3));
        let state = template_state(&s, "tenant-sync").await.unwrap();
        assert_eq!(state.stable, Some(1));
        assert_eq!(
            state.previous,
            Some(3),
            "previous now points at what we left"
        );

        // The launch log is the audit trail: v1, v3, v1, newest first.
        let log = s.template_launches("tenant-sync").await.unwrap();
        assert_eq!(
            log.iter().map(|l| l.version).collect::<Vec<_>>(),
            vec![1, 3, 1]
        );
        assert_eq!(log[0].launched_by.as_deref(), Some("oncall"));
    }

    #[tokio::test]
    async fn deprecation_is_template_wide_and_reversible() {
        let s = store();
        register(&s, req_launched(PARAMETERIZED)).await.unwrap();

        let status = set_deprecated(
            &s,
            "tenant-sync",
            Some("superseded".into()),
            Some("bob"),
            true,
        )
        .await
        .unwrap();
        assert_eq!(status, TemplateStatus::Deprecated);
        let state = template_state(&s, "tenant-sync").await.unwrap();
        assert_eq!(state.status, TemplateStatus::Deprecated);
        assert_eq!(
            state.deprecation.as_ref().unwrap().reason.as_deref(),
            Some("superseded")
        );
        // Retiring must not break existing callers: `stable` still resolves.
        assert_eq!(
            resolve_version(&s, "tenant-sync", VersionSelector::stable())
                .await
                .unwrap(),
            1
        );
        // But launching into a retired template is refused — reviving it that way
        // is almost certainly a mistake.
        let err = launch(&s, "tenant-sync", VersionSelector::Pinned(1), None)
            .await
            .unwrap_err()
            .to_string();
        assert!(err.contains("deprecated"), "{err}");

        // `--undo` restores the prior status, derived rather than remembered.
        let status = set_deprecated(&s, "tenant-sync", None, None, false)
            .await
            .unwrap();
        assert_eq!(status, TemplateStatus::Launched);
        assert!(
            template_state(&s, "tenant-sync")
                .await
                .unwrap()
                .deprecation
                .is_none()
        );
        // Deprecating a template that does not exist is a typed error.
        assert!(matches!(
            set_deprecated(&s, "nope", None, None, true)
                .await
                .unwrap_err(),
            CliError::UnknownPipelineTemplate { .. }
        ));
    }

    #[tokio::test]
    async fn explicit_id_wins_and_is_validated() {
        let s = store();
        let mut r = req(PARAMETERIZED);
        r.id = Some("my-template".into());
        assert_eq!(register(&s, r).await.unwrap().id, "my-template");

        let mut bad = req(PARAMETERIZED);
        bad.id = Some("Bad Id".into());
        assert!(register(&s, bad).await.is_err());
    }

    #[tokio::test]
    async fn register_requires_an_id_source() {
        let s = store();
        let body = "version: 1\npipeline:\n  source: { type: csv, config: { path: a.csv } }\n  sink: { type: jsonl, config: { path: o.jsonl } }\n";
        let err = register(&s, req(body)).await.unwrap_err().to_string();
        assert!(err.contains("no template id"), "{err}");
    }

    #[tokio::test]
    async fn register_rejects_a_structurally_invalid_config() {
        let s = store();
        let err = register(&s, req("version: 1\nname: x\nnope: 1\npipeline: {}\n"))
            .await
            .unwrap_err()
            .to_string();
        assert!(err.contains("nope") || err.contains("pipeline"), "{err}");
    }

    #[tokio::test]
    async fn register_rejects_an_invalid_params_block() {
        let s = store();
        let body = "version: 1\nname: x\nparams:\n  a: { required: true, default: 1 }\npipeline:\n  source: { type: csv, config: { path: a.csv } }\n  sink: { type: jsonl, config: { path: o.jsonl } }\n";
        let err = register(&s, req(body)).await.unwrap_err().to_string();
        assert!(err.contains("required"), "{err}");
    }

    #[tokio::test]
    async fn register_rejects_a_non_mapping_body() {
        let s = store();
        let err = register(&s, req("- a\n- b\n"))
            .await
            .unwrap_err()
            .to_string();
        assert!(err.contains("mapping"), "{err}");
        let err = register(&s, req(": :\n")).await.unwrap_err().to_string();
        assert!(err.contains("YAML"), "{err}");
    }

    #[tokio::test]
    async fn materialize_binds_params_and_defaults() {
        let s = store();
        register(&s, req_launched(PARAMETERIZED)).await.unwrap();
        let supplied: SuppliedParams = [("tenant_id".to_string(), json!("acme"))].into();
        let want = resolve_version(&s, "tenant-sync", VersionSelector::stable())
            .await
            .unwrap();
        let out = materialize(
            &s,
            "tenant-sync",
            want,
            &supplied,
            &BTreeMap::new(),
            Materialize::Local,
        )
        .await
        .unwrap();
        assert_eq!(out.version, 1);
        assert_eq!(out.name.as_deref(), Some("tenant-sync"));
        assert_eq!(out.format(), ConfigFormat::Json);
        let doc: Value = serde_json::from_str(&out.body).unwrap();
        assert_eq!(
            doc["pipeline"]["source"]["config"]["url"],
            "https://api.example.com/acme/events?since=1970-01-01"
        );
        assert_eq!(out.params_redacted["tenant_id"], json!("acme"));
        assert_eq!(out.params_redacted["page"], json!(100));
        assert!(!out.used_secret_params);
    }

    #[tokio::test]
    async fn materialize_reports_missing_and_unknown_params() {
        let s = store();
        register(&s, req_launched(PARAMETERIZED)).await.unwrap();
        let err = materialize(
            &s,
            "tenant-sync",
            1,
            &SuppliedParams::new(),
            &BTreeMap::new(),
            Materialize::Local,
        )
        .await
        .unwrap_err();
        assert!(matches!(err, CliError::MissingParam { .. }), "{err:?}");

        let supplied: SuppliedParams = [
            ("tenant_id".to_string(), json!("a")),
            ("bogus".to_string(), json!("b")),
        ]
        .into();
        let err = materialize(
            &s,
            "tenant-sync",
            1,
            &supplied,
            &BTreeMap::new(),
            Materialize::Local,
        )
        .await
        .unwrap_err();
        assert!(matches!(err, CliError::UnknownParam { .. }), "{err:?}");
    }

    #[tokio::test]
    async fn unknown_template_and_version_are_typed_errors() {
        let s = store();
        register(&s, req_launched(PARAMETERIZED)).await.unwrap();
        let err = resolve_version(&s, "nope", VersionSelector::stable())
            .await
            .unwrap_err();
        assert!(
            matches!(err, CliError::UnknownPipelineTemplate { ref id, .. } if id == "nope"),
            "{err:?}"
        );
        let supplied: SuppliedParams = [("tenant_id".to_string(), json!("a"))].into();
        let err = materialize(
            &s,
            "tenant-sync",
            9,
            &supplied,
            &BTreeMap::new(),
            Materialize::Local,
        )
        .await
        .unwrap_err();
        assert!(
            matches!(
                err,
                CliError::UnknownPipelineTemplate {
                    version: Some(9),
                    ..
                }
            ),
            "{err:?}"
        );
    }

    #[tokio::test]
    async fn env_overrides_win_over_the_process_environment() {
        let s = store();
        let body = "\
version: 1
name: env-template
pipeline:
  source: { type: rest, config: { url: \"https://x/${env:FAUCET_TPL_REGION}\" } }
  sink: { type: jsonl, config: { path: ./o.jsonl } }
";
        unsafe { std::env::set_var("FAUCET_TPL_REGION", "from-process") };
        register(&s, req_launched(body)).await.unwrap();

        let out = materialize(
            &s,
            "env-template",
            1,
            &SuppliedParams::new(),
            &BTreeMap::new(),
            Materialize::Local,
        )
        .await
        .unwrap();
        let doc: Value = serde_json::from_str(&out.body).unwrap();
        assert_eq!(
            doc["pipeline"]["source"]["config"]["url"],
            "https://x/from-process"
        );

        let overrides: BTreeMap<String, String> =
            [("FAUCET_TPL_REGION".to_string(), "from-request".to_string())].into();
        let out = materialize(
            &s,
            "env-template",
            1,
            &SuppliedParams::new(),
            &overrides,
            Materialize::Local,
        )
        .await
        .unwrap();
        let doc: Value = serde_json::from_str(&out.body).unwrap();
        assert_eq!(
            doc["pipeline"]["source"]["config"]["url"],
            "https://x/from-request"
        );
        assert_eq!(std::env::var("FAUCET_TPL_REGION").unwrap(), "from-process");
        unsafe { std::env::remove_var("FAUCET_TPL_REGION") };
    }

    #[tokio::test]
    async fn secret_params_are_flagged_and_redacted() {
        let s = store();
        let body = "\
version: 1
name: secret-template
params:
  api_token: { required: true, secret: true }
pipeline:
  source:
    type: rest
    config:
      url: https://api.example.com/events
      auth: { type: bearer, config: { token: \"${param.api_token}\" } }
  sink: { type: jsonl, config: { path: ./o.jsonl } }
";
        register(&s, req_launched(body)).await.unwrap();
        let supplied: SuppliedParams =
            [("api_token".to_string(), json!("tok-abcdefghijklmnop"))].into();
        let out = materialize(
            &s,
            "secret-template",
            1,
            &supplied,
            &BTreeMap::new(),
            Materialize::Local,
        )
        .await
        .unwrap();
        assert!(out.used_secret_params);
        assert_eq!(out.params_redacted["api_token"], json!("***"));
        assert!(out.body.contains("tok-abcdefghijklmnop"));
        assert_eq!(
            crate::secrets::registry::redact("token=tok-abcdefghijklmnop"),
            "token=***"
        );
    }

    #[tokio::test]
    async fn channels_are_promoted_independently_of_launching() {
        let s = store();
        register(&s, req_launched(PARAMETERIZED)).await.unwrap(); // v1 live
        register(&s, req(PARAMETERIZED)).await.unwrap(); // v2 draft build
        let mut tagged = req(PARAMETERIZED);
        tagged.tags = vec![VersionChannel::Dev];
        register(&s, tagged).await.unwrap(); // v3, dev=v3

        assert_eq!(
            resolve_version(
                &s,
                "tenant-sync",
                VersionSelector::Channel(VersionChannel::Dev)
            )
            .await
            .unwrap(),
            3
        );
        // Promoting an environment channel never touches what is live.
        assert_eq!(
            promote(
                &s,
                "tenant-sync",
                VersionChannel::PreProd,
                VersionSelector::Channel(VersionChannel::Dev)
            )
            .await
            .unwrap(),
            3
        );
        let state = template_state(&s, "tenant-sync").await.unwrap();
        assert_eq!(state.stable, Some(1), "promote must not move `stable`");
        assert_eq!(state.tags["dev"], 3);
        assert_eq!(state.tags["pre-prod"], 3);
        assert!(!state.tags.contains_key("stable"), "derived, never stored");

        // Launching *from* a channel is the promotion pipeline's last step.
        let out = launch(
            &s,
            "tenant-sync",
            VersionSelector::Channel(VersionChannel::PreProd),
            None,
        )
        .await
        .unwrap();
        assert_eq!((out.version, out.replaced), (3, Some(1)));
    }

    #[tokio::test]
    async fn derived_channels_cannot_be_promoted() {
        let s = store();
        register(&s, req_launched(PARAMETERIZED)).await.unwrap();
        for (tag, needle) in [
            (VersionChannel::Stable, "launch"),
            (VersionChannel::Previous, "launched before"),
            (VersionChannel::Newest, "highest version"),
        ] {
            let err = promote(&s, "tenant-sync", tag, VersionSelector::Pinned(1))
                .await
                .unwrap_err()
                .to_string();
            assert!(err.contains("derived"), "{tag}: {err}");
            assert!(err.contains(needle), "{tag}: {err}");
        }
        // Same guard on the register path, before anything is written.
        let mut bad = req(PARAMETERIZED);
        bad.tags = vec![VersionChannel::Stable];
        assert!(register(&s, bad).await.is_err());
        assert_eq!(
            s.template_versions("tenant-sync").await.unwrap(),
            vec![1],
            "the rejected register must not have appended a version"
        );
    }

    #[tokio::test]
    async fn promoting_to_a_missing_version_is_rejected() {
        let s = store();
        register(&s, req(PARAMETERIZED)).await.unwrap();
        let err = promote(
            &s,
            "tenant-sync",
            VersionChannel::Prod,
            VersionSelector::Pinned(9),
        )
        .await
        .unwrap_err();
        assert!(
            matches!(
                err,
                CliError::UnknownPipelineTemplate {
                    version: Some(9),
                    ..
                }
            ),
            "{err:?}"
        );
        assert!(matches!(
            promote(&s, "nope", VersionChannel::Prod, VersionSelector::Pinned(1))
                .await
                .unwrap_err(),
            CliError::UnknownPipelineTemplate { .. }
        ));
    }

    #[tokio::test]
    async fn deleting_a_version_drops_pointers_aimed_at_it() {
        let s = store();
        register(&s, req(PARAMETERIZED)).await.unwrap();
        register(&s, req(PARAMETERIZED)).await.unwrap();
        launch(&s, "tenant-sync", VersionSelector::Pinned(1), None)
            .await
            .unwrap();
        launch(&s, "tenant-sync", VersionSelector::Pinned(2), None)
            .await
            .unwrap();
        promote(
            &s,
            "tenant-sync",
            VersionChannel::Prod,
            VersionSelector::Pinned(1),
        )
        .await
        .unwrap();

        // Deleting v1 must leave neither a channel nor a launch entry pointing at
        // it — otherwise `previous` or `prod` would resolve to a missing version.
        assert_eq!(s.template_delete("tenant-sync", Some(1)).await.unwrap(), 1);
        let state = template_state(&s, "tenant-sync").await.unwrap();
        assert!(!state.tags.contains_key("prod"), "{:?}", state.tags);
        assert_eq!(state.stable, Some(2));
        assert_eq!(state.previous, None, "v1's launch entry went with it");

        s.template_delete("tenant-sync", None).await.unwrap();
        assert!(s.template_launches("tenant-sync").await.unwrap().is_empty());
        assert!(s.template_tags("tenant-sync").await.unwrap().is_empty());
    }

    #[tokio::test]
    async fn delete_removes_one_version_or_all() {
        let s = store();
        register(&s, req(PARAMETERIZED)).await.unwrap();
        register(&s, req(PARAMETERIZED)).await.unwrap();
        assert_eq!(s.template_delete("tenant-sync", Some(1)).await.unwrap(), 1);
        assert_eq!(s.template_versions("tenant-sync").await.unwrap(), vec![2]);
        assert_eq!(s.template_delete("tenant-sync", None).await.unwrap(), 1);
        assert!(s.template_list().await.unwrap().is_empty());
        assert_eq!(s.template_delete("tenant-sync", None).await.unwrap(), 0);
        assert_eq!(s.template_delete("tenant-sync", Some(3)).await.unwrap(), 0);
    }

    #[tokio::test]
    async fn registers_a_topology_config() {
        let s = store();
        let body = "\
version: 1
name: topo-template
params:
  path: { default: ./in.csv }
pipeline:
  sources:
    s: { type: csv, config: { path: \"${param.path}\" } }
  sinks:
    o: { type: jsonl, config: { path: ./out.jsonl } }
  nodes:
    src: { kind: source, ref: s }
    w: { kind: sink, ref: o }
  edges:
    - { from: src, to: w }
";
        let rec = register(&s, req_launched(body)).await.unwrap();
        assert_eq!(rec.id, "topo-template");
        let out = materialize(
            &s,
            "topo-template",
            1,
            &SuppliedParams::new(),
            &BTreeMap::new(),
            Materialize::Local,
        )
        .await
        .unwrap();
        let doc: Value = serde_json::from_str(&out.body).unwrap();
        assert_eq!(
            doc["pipeline"]["sources"]["s"]["config"]["path"],
            "./in.csv"
        );
    }

    #[tokio::test]
    async fn version_history_is_bounded() {
        use crate::serve::history::templates::VERSION_RETAIN;
        let s = store();
        for _ in 0..(VERSION_RETAIN + 3) {
            register(&s, req(PARAMETERIZED)).await.unwrap();
        }
        let versions = s.template_versions("tenant-sync").await.unwrap();
        assert_eq!(versions.len(), VERSION_RETAIN);
        assert_eq!(versions[0], (VERSION_RETAIN + 3) as u32, "newest kept");
        assert!(!versions.contains(&1), "oldest pruned");
    }

    #[tokio::test]
    async fn store_url_grammar() {
        assert!(resolve_store_url("memory").await.is_ok());
        // `RunHistory` is not Debug, so match rather than `unwrap_err`.
        match resolve_store_url("mysql://nope").await {
            Ok(_) => panic!("an unrecognised scheme must be rejected"),
            Err(e) => assert!(e.to_string().contains("template store"), "{e}"),
        }
        // SQL schemes are recognised even without the build feature — the error
        // then names the missing feature rather than the URL grammar.
        let dir = tempfile::tempdir().unwrap();
        let url = format!("sqlite:{}", dir.path().join("t.db").display());
        if let Err(e) = resolve_store_url(&url).await {
            assert!(e.to_string().contains("serve-history-sqlite"), "{e}");
        }
    }
}