rivet/config/

mod.rs

pub mod cursor;
mod destination;
mod export;
mod format;
mod lints;
mod notifications;
pub mod resolve;
pub mod schema;
mod source;

pub use cursor::IncrementalCursorMode;
pub use destination::*;
pub use export::*;
pub use format::*;
pub use notifications::*;
#[allow(unused_imports)]
pub(crate) use resolve::resolve_env_vars;
pub use resolve::{parse_file_size, resolve_vars};
pub use schema::generate_config_schema_pretty;
pub use source::*;

use schemars::JsonSchema;
use serde::Deserialize;

/// Top-level Rivet configuration root.
///
/// Operators write this struct as YAML (typically `rivet.yaml`).  The
/// `JsonSchema` derive is the source of truth for the `schemas/rivet.schema.json`
/// artifact and the `rivet schema config` command's output (v0.7.3 P0).
#[derive(Debug, Deserialize, JsonSchema, Clone)]
#[serde(deny_unknown_fields)]
pub struct Config {
    pub source: SourceConfig,
    pub exports: Vec<ExportConfig>,
    #[serde(default)]
    pub notifications: Option<NotificationsConfig>,
    #[serde(default)]
    pub parallel_exports: bool,
    #[serde(default)]
    pub parallel_export_processes: bool,
}

impl Config {
    pub fn load(path: &str) -> crate::error::Result<Self> {
        Self::load_with_params(path, None)
    }

    pub fn load_with_params(
        path: &str,
        params: Option<&std::collections::HashMap<String, String>>,
    ) -> crate::error::Result<Self> {
        // F11 (0.7.5 audit): raw `std::io::Error` lost the path on
        // not-found.  Wrap with the file path + a hint so the operator
        // can see *which* config the tool could not open.
        let contents = std::fs::read_to_string(path).map_err(|e| {
            if e.kind() == std::io::ErrorKind::NotFound {
                anyhow::anyhow!(
                    "config file '{}' not found.\n  Hint: check the path, or run `rivet init` to generate one.",
                    path
                )
            } else {
                anyhow::anyhow!("cannot read config file '{}': {}", path, e)
            }
        })?;
        // Warn about typo'd `--param` keys once per CLI invocation, using the
        // un-resolved YAML as the haystack so the placeholders are still there.
        // We pass the raw `contents` (not `resolved`) on purpose: after
        // resolution the placeholders are gone, and every key would look unused.
        resolve::warn_unused_params(&contents, params);
        let resolved = resolve_vars(&contents, params)?;
        // F12 (0.7.5 audit): YAML parse errors did not name the config
        // file.  When loading from disk we know the path — thread it
        // into the parse error.
        Self::from_yaml(&resolved).map_err(|e| anyhow::anyhow!("config file '{}': {:#}", path, e))
    }

    pub fn from_yaml(yaml: &str) -> crate::error::Result<Self> {
        Self::check_misplaced_tuning_fields(yaml)?;
        let config: Config = serde_yaml_ng::from_str(yaml).map_err(lints::enhance_parse_error)?;
        config.validate()?;
        Ok(config)
    }

    /// Detect tuning-related fields placed directly under `source:` or an
    /// `exports[]` entry instead of inside the `tuning:` sub-key. Without this
    /// check serde silently ignores unknown keys and the user gets unexpected
    /// defaults (e.g. batch_size=10 000 instead of the intended 1 000).
    fn check_misplaced_tuning_fields(yaml: &str) -> crate::error::Result<()> {
        const TUNING_FIELDS: &[&str] = &[
            "batch_size",
            "batch_size_memory_mb",
            "throttle_ms",
            "statement_timeout_s",
            "max_retries",
            "retry_backoff_ms",
            "lock_timeout_s",
            "memory_threshold_mb",
            "profile",
        ];

        let root: serde_yaml_ng::Value = serde_yaml_ng::from_str(yaml)?;

        if let Some(source) = root.get("source") {
            let misplaced: Vec<&str> = TUNING_FIELDS
                .iter()
                .copied()
                .filter(|&f| source.get(f).is_some())
                .collect();
            if !misplaced.is_empty() {
                anyhow::bail!(
                    "source: field(s) [{}] belong under 'source.tuning:', not directly under 'source:'. \
                     Example:\n  source:\n    tuning:\n      {}: <value>",
                    misplaced.join(", "),
                    misplaced[0],
                );
            }
        }

        if let Some(exports) = root.get("exports").and_then(|e| e.as_sequence()) {
            for (i, export) in exports.iter().enumerate() {
                let name = export
                    .get("name")
                    .and_then(|n| n.as_str())
                    .unwrap_or("<unnamed>");
                let misplaced: Vec<&str> = TUNING_FIELDS
                    .iter()
                    .copied()
                    .filter(|&f| export.get(f).is_some())
                    .collect();
                if !misplaced.is_empty() {
                    anyhow::bail!(
                        "export '{}' (index {}): field(s) [{}] belong under 'exports[].tuning:', \
                         not directly in the export. Example:\n  exports:\n    - name: {}\n      tuning:\n        {}: <value>",
                        name,
                        i,
                        misplaced.join(", "),
                        name,
                        misplaced[0],
                    );
                }
            }
        }

        Ok(())
    }

    /// Reject a config before any plan/connect step. The body is split into
    /// three cohesive validators so each can be read — and unit-tested — on its
    /// own: the export-list shape, the source connection block, and the
    /// per-export rules. The end-to-end surface (`Config::from_yaml`) is
    /// covered by `config/tests/{validation,secops}.rs`; the split additionally
    /// lets a rule be exercised directly via `validate_export`.
    fn validate(&self) -> crate::error::Result<()> {
        self.validate_exports_list()?;
        self.validate_source_connection()?;
        for export in &self.exports {
            self.validate_export(export)?;
        }
        Ok(())
    }

    /// Whole-config shape: at least one export, names unique.
    fn validate_exports_list(&self) -> crate::error::Result<()> {
        // An empty `exports:` list is almost always a typo (wrong config file,
        // dropped anchor, merged doc with the anchor section missing). Running
        // with zero exports is a silent no-op that looks like success in CI;
        // reject fast instead. See QA backlog Task 5.1.
        if self.exports.is_empty() {
            anyhow::bail!("exports: at least one export must be defined (got empty list)");
        }

        // Duplicate export names break state tracking: `export_state`,
        // `file_log`, and `chunk_run` are all keyed by `export_name`, so
        // two configs with the same name silently share cursor/file-log rows.
        // QA backlog Task 5.1.
        let mut seen: std::collections::HashSet<&str> =
            std::collections::HashSet::with_capacity(self.exports.len());
        for e in &self.exports {
            if !seen.insert(e.name.as_str()) {
                anyhow::bail!(
                    "exports: duplicate export name '{}' (each export must have a unique name; state is keyed by name)",
                    e.name
                );
            }
        }
        Ok(())
    }

    /// Source connection block: exactly one connection method, well-formed,
    /// and the source-level tuning that is shared by every export.
    fn validate_source_connection(&self) -> crate::error::Result<()> {
        if let Some(t) = &self.source.tuning
            && t.batch_size.is_some()
            && t.batch_size_memory_mb.is_some()
        {
            anyhow::bail!("tuning: batch_size and batch_size_memory_mb are mutually exclusive");
        }

        if !self.source.has_url_fields() && !self.source.has_structured_fields() {
            // First-run footgun: a config that forgot the source block
            // entirely.  Show the recommended path (`url_env`) up-front;
            // operators who actually want structured fields know to look
            // for them.
            anyhow::bail!(
                "source: no connection method configured. Add one of:\n  url_env: DATABASE_URL                          (URL from env var — recommended)\n  url: 'postgresql://user:pass@host:5432/db'      (inline — not recommended for committed configs)\n  url_file: /etc/rivet/source.url                 (URL from file — rotation-friendly)\n  host/user/database/...                          (structured fields under `source:`)"
            );
        }

        if self.source.has_url_fields() {
            let url_count = [
                &self.source.url,
                &self.source.url_env,
                &self.source.url_file,
            ]
            .iter()
            .filter(|u| u.is_some())
            .count();
            if url_count > 1 {
                anyhow::bail!(
                    "source: specify exactly one of 'url', 'url_env', or 'url_file' (got {} set).\n  Hint: pick one — `url_env` is recommended so credentials never enter the YAML.",
                    url_count
                );
            }
        }

        if self.source.has_url_fields() && self.source.has_structured_fields() {
            anyhow::bail!(
                "source: pick either URL-based config (url/url_env/url_file) OR structured fields (host/user/database/port/password_env), not both.\n  Hint: remove whichever block you don't want; mixing the two is ambiguous."
            );
        }

        if self.source.has_structured_fields() {
            if self.source.host.is_none() {
                anyhow::bail!(
                    "source: structured config is missing 'host'.\n  Hint: add `host: localhost` (or your DB host) under `source:` in rivet.yaml.\n  Or switch to URL-based config: `url_env: DATABASE_URL`."
                );
            }
            if self.source.user.is_none() {
                anyhow::bail!(
                    "source: structured config is missing 'user'.\n  Hint: add `user: <username>` under `source:` in rivet.yaml."
                );
            }
            if self.source.database.is_none() {
                anyhow::bail!(
                    "source: structured config is missing 'database'.\n  Hint: add `database: <dbname>` under `source:` in rivet.yaml."
                );
            }
            if self.source.password.is_some() && self.source.password_env.is_some() {
                anyhow::bail!(
                    "source: specify 'password' OR 'password_env', not both.\n  Hint: prefer `password_env: DB_PASSWORD` so credentials never enter the YAML."
                );
            }
        }
        Ok(())
    }

    /// Per-export rules: effective tuning, query source, `query_file` SecOps,
    /// destination auth, compression, and the mode/chunk matrix. Takes `&self`
    /// because effective tuning merges the source-level block.
    fn validate_export(&self, export: &ExportConfig) -> crate::error::Result<()> {
        let merged =
            crate::tuning::merge_tuning_config(self.source.tuning.as_ref(), export.tuning.as_ref());
        if let Some(t) = merged
            && t.batch_size.is_some()
            && t.batch_size_memory_mb.is_some()
        {
            anyhow::bail!(
                "export '{}': effective tuning has both batch_size and batch_size_memory_mb (mutually exclusive)",
                export.name
            );
        }
        if let Some(et) = &export.tuning
            && et.batch_size.is_some()
            && et.batch_size_memory_mb.is_some()
        {
            anyhow::bail!(
                "export '{}': tuning.batch_size and tuning.batch_size_memory_mb are mutually exclusive",
                export.name
            );
        }

        let set_count = [
            export.query.is_some(),
            export.query_file.is_some(),
            export.table.is_some(),
        ]
        .iter()
        .filter(|b| **b)
        .count();
        if set_count == 0 {
            anyhow::bail!(
                "export '{}': must specify exactly one of 'query', 'query_file', or 'table'",
                export.name
            );
        }
        if set_count > 1 {
            anyhow::bail!(
                "export '{}': specify exactly one of 'query', 'query_file', or 'table' (got {} set)",
                export.name,
                set_count
            );
        }
        // SecOps: syntactic `query_file` checks must run at config-validate
        // time so `rivet check` / `rivet doctor` catch them before any
        // plan step. The same checks repeat (with a canonicalize-based
        // symlink probe) in `ExportConfig::resolve_query` because the
        // file may have been swapped between validation and read.
        if let Some(file) = &export.query_file {
            let p = std::path::Path::new(file);
            if p.is_absolute() {
                anyhow::bail!(
                    "export '{}': query_file must be a relative path: '{}'",
                    export.name,
                    file
                );
            }
            if p.components().any(|c| c == std::path::Component::ParentDir) {
                anyhow::bail!(
                    "export '{}': query_file path must not contain '..': '{}'",
                    export.name,
                    file
                );
            }
        }
        if export.destination.destination_type == DestinationType::S3 {
            let ak = export.destination.access_key_env.is_some();
            let sk = export.destination.secret_key_env.is_some();
            if ak != sk {
                anyhow::bail!(
                    "export '{}': S3 requires both access_key_env and secret_key_env, or neither (use default AWS credential chain)",
                    export.name
                );
            }
        }

        if export.destination.destination_type == DestinationType::Gcs
            && export.destination.allow_anonymous
            && export.destination.credentials_file.is_some()
        {
            anyhow::bail!(
                "export '{}': GCS allow_anonymous cannot be used together with credentials_file",
                export.name
            );
        }

        if export.destination.destination_type == DestinationType::Azure {
            let has_name = export.destination.account_name.is_some();
            let has_key = export.destination.account_key_env.is_some();
            let has_sas = export.destination.sas_token_env.is_some();
            if export.destination.allow_anonymous {
                if has_name || has_key || has_sas {
                    anyhow::bail!(
                        "export '{}': Azure allow_anonymous cannot be combined with account_name/account_key_env/sas_token_env",
                        export.name
                    );
                }
            } else if has_key && has_sas {
                anyhow::bail!(
                    "export '{}': Azure account_key_env and sas_token_env are mutually exclusive — pick one auth mode",
                    export.name
                );
            } else if !has_name {
                anyhow::bail!(
                    "export '{}': Azure requires account_name (plus account_key_env or sas_token_env), or allow_anonymous: true for Azurite",
                    export.name
                );
            } else if !has_key && !has_sas {
                anyhow::bail!(
                    "export '{}': Azure requires account_key_env or sas_token_env (or allow_anonymous: true for Azurite)",
                    export.name
                );
            }
        }

        if let Some(cred_path) = &export.destination.credentials_file
            && !std::path::Path::new(cred_path).exists()
        {
            anyhow::bail!(
                "export '{}': credentials_file '{}' does not exist",
                export.name,
                cred_path
            );
        }

        if let Some(ref size_str) = export.max_file_size {
            parse_file_size(size_str).map_err(|_| {
                anyhow::anyhow!(
                    "export '{}': invalid max_file_size '{}'",
                    export.name,
                    size_str
                )
            })?;
        }

        if let Some(level) = export.compression_level {
            match export.compression {
                CompressionType::Zstd => {
                    if !(1..=22).contains(&level) {
                        anyhow::bail!(
                            "export '{}': zstd compression_level must be 1..22, got {}",
                            export.name,
                            level
                        );
                    }
                }
                CompressionType::Gzip => {
                    if level > 10 {
                        anyhow::bail!(
                            "export '{}': gzip compression_level must be 0..10, got {}",
                            export.name,
                            level
                        );
                    }
                }
                _ => {
                    anyhow::bail!(
                        "export '{}': compression_level is only supported for zstd and gzip",
                        export.name
                    );
                }
            }
        }

        match export.mode {
            ExportMode::Incremental => {
                if export.cursor_column.is_none() {
                    anyhow::bail!(
                        "export '{}': incremental mode requires cursor_column",
                        export.name
                    );
                }
                match export.incremental_cursor_mode {
                    IncrementalCursorMode::Coalesce => {
                        if export.cursor_fallback_column.is_none() {
                            anyhow::bail!(
                                "export '{}': incremental_cursor_mode: coalesce requires cursor_fallback_column",
                                export.name
                            );
                        }
                    }
                    IncrementalCursorMode::SingleColumn => {
                        if export.cursor_fallback_column.is_some() {
                            anyhow::bail!(
                                "export '{}': cursor_fallback_column is only valid with incremental_cursor_mode: coalesce",
                                export.name
                            );
                        }
                    }
                }
            }
            ExportMode::Chunked => {
                // `chunk_column` is mandatory unless the user used the `table:`
                // shortcut on a Postgres source — in that case it is auto-resolved
                // from the table's single-integer PK at plan-build time (see
                // `crate::plan::build::resolve_chunk_column`).
                if export.chunk_column.is_none() && export.table.is_none() {
                    anyhow::bail!(
                        "export '{}': chunked mode requires chunk_column \
                         (or use `table:` shortcut on a Postgres source to auto-resolve from PK)",
                        export.name
                    );
                }
                // chunk_size == 0 would divide the range into zero-width
                // slices and (before the saturating fix in generate_chunks)
                // either infinite-loop or produce no progress. QA backlog
                // Task 5.1.
                if export.chunk_size == 0 {
                    anyhow::bail!(
                        "export '{}': chunked mode requires chunk_size >= 1 (got 0)",
                        export.name
                    );
                }
                // parallel == 0 means "spawn zero workers". Claiming tasks
                // with no workers stalls the pipeline. QA backlog Task 5.1.
                if export.parallel == 0 {
                    anyhow::bail!(
                        "export '{}': chunked mode requires parallel >= 1 (got 0)",
                        export.name
                    );
                }
                if let Some(0) = export.chunk_count {
                    anyhow::bail!("export '{}': chunk_count must be >= 1", export.name);
                }
                if export.chunk_count.is_some() && export.chunk_dense {
                    anyhow::bail!(
                        "export '{}': chunk_count and chunk_dense are mutually exclusive",
                        export.name
                    );
                }
                if export.chunk_count.is_some() && export.chunk_by_days.is_some() {
                    anyhow::bail!(
                        "export '{}': chunk_count and chunk_by_days are mutually exclusive",
                        export.name
                    );
                }
            }
            ExportMode::TimeWindow => {
                if export.time_column.is_none() {
                    anyhow::bail!(
                        "export '{}': time_window mode requires time_column",
                        export.name
                    );
                }
                if export.days_window.is_none() {
                    anyhow::bail!(
                        "export '{}': time_window mode requires days_window",
                        export.name
                    );
                }
            }
            ExportMode::Full => {}
        }

        if export.chunk_dense && export.mode != ExportMode::Chunked {
            anyhow::bail!(
                "export '{}': chunk_dense is only valid with mode: chunked",
                export.name
            );
        }

        if let Some(days) = export.chunk_by_days {
            if export.mode != ExportMode::Chunked {
                anyhow::bail!(
                    "export '{}': chunk_by_days requires mode: chunked",
                    export.name
                );
            }
            if export.chunk_dense {
                anyhow::bail!(
                    "export '{}': chunk_by_days cannot be combined with chunk_dense",
                    export.name
                );
            }
            if days == 0 {
                anyhow::bail!("export '{}': chunk_by_days must be at least 1", export.name);
            }
        }
        Ok(())
    }
}

#[cfg(test)]
mod tests;