horizon-sdk 16.0.0

//! SDK configuration primitives.
//!
//! Configuration is layered: struct defaults -> env vars -> programmatic overrides.
//! The builder in `lib.rs` consumes these to construct subsystems.

use std::fmt;

use serde::Deserialize;

use super::constants;

/// Coverage configuration: the inference service that serves the `/coverage` endpoints.
#[derive(Clone, Deserialize)]
pub struct CoverageConfig {
    /// Bearer token the inference deployment expects, when it requires one.
    #[serde(default)]
    pub api_key: Option<String>,
    pub base_url: String,
    #[serde(default = "default_coverage_timeout_secs")]
    pub timeout_secs: u64,
}

impl fmt::Debug for CoverageConfig {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("CoverageConfig")
            .field("api_key", &self.api_key.as_ref().map(|_| "[REDACTED]"))
            .field("base_url", &self.base_url)
            .field("timeout_secs", &self.timeout_secs)
            .finish()
    }
}

#[derive(Clone, Deserialize)]
pub struct HorizonApiConfig {
    pub api_key: String,
    #[serde(default = "default_horizon_api_base_url")]
    pub base_url: String,
    #[serde(default = "default_horizon_api_timeout_secs")]
    pub timeout_secs: u64,
}

impl fmt::Debug for HorizonApiConfig {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("HorizonApiConfig")
            .field("api_key", &"[REDACTED]")
            .field("base_url", &self.base_url)
            .field("timeout_secs", &self.timeout_secs)
            .finish()
    }
}

/// Iceberg catalog configuration for schema discovery.
#[derive(Debug, Clone, Deserialize)]
pub struct IcebergConfig {
    #[serde(default = "default_catalog_uri")]
    pub catalog_uri: String,
    pub catalog_warehouse: Option<String>,
    /// `OAuth2` client credential (`client_id:client_secret`) for the REST catalog.
    #[serde(default)]
    pub credential: Option<String>,
    #[serde(default = "default_data_row_table")]
    pub data_row_table: String,
    #[serde(default = "default_metadata_row_table")]
    pub metadata_row_table: String,
    /// `OAuth2` token endpoint for the REST catalog.
    #[serde(default)]
    pub oauth2_server_uri: Option<String>,
    /// `OAuth2` scope requested when fetching a catalog token.
    #[serde(default)]
    pub scope: Option<String>,
}

/// Kafka producer configuration (SASL/SSL + topic routing).
#[derive(Clone, Deserialize)]
pub struct KafkaConfig {
    #[serde(default = "default_bootstrap_servers")]
    pub bootstrap_servers: String,
    #[serde(default = "default_compression_type")]
    pub compression_type: String,
    #[serde(default = "default_data_row_topic")]
    pub data_row_topic: String,
    #[serde(default = "default_metadata_row_topic")]
    pub metadata_row_topic: String,
    #[serde(default = "default_sasl_mechanism")]
    pub sasl_mechanism: String,
    pub sasl_password: Option<String>,
    pub sasl_username: Option<String>,
    #[serde(default = "default_security_protocol")]
    pub security_protocol: String,
}

impl fmt::Debug for KafkaConfig {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("KafkaConfig")
            .field("bootstrap_servers", &self.bootstrap_servers)
            .field("compression_type", &self.compression_type)
            .field("data_row_topic", &self.data_row_topic)
            .field("metadata_row_topic", &self.metadata_row_topic)
            .field("sasl_mechanism", &self.sasl_mechanism)
            .field(
                "sasl_password",
                &self.sasl_password.as_ref().map(|_| "[REDACTED]"),
            )
            .field(
                "sasl_username",
                &self.sasl_username.as_ref().map(|_| "[REDACTED]"),
            )
            .field("security_protocol", &self.security_protocol)
            .finish()
    }
}

/// `PostgreSQL` connection configuration.
#[derive(Clone, Deserialize)]
pub struct PostgresConfig {
    pub database: String,
    pub host: String,
    pub password: String,
    pub port: u16,
    #[serde(rename = "organization_id")]
    pub possible_organization_id: Option<uuid::Uuid>,
    pub role: Option<String>,
    #[serde(default = "default_sslmode")]
    pub sslmode: String,
    pub user: String,
}

impl fmt::Debug for PostgresConfig {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("PostgresConfig")
            .field("database", &self.database)
            .field("host", &self.host)
            .field("password", &"[REDACTED]")
            .field("port", &self.port)
            .field("possible_organization_id", &self.possible_organization_id)
            .field("role", &self.role)
            .field("sslmode", &self.sslmode)
            .field("user", &"[REDACTED]")
            .finish()
    }
}

/// Top-level SDK configuration.
#[derive(Debug, Clone, Deserialize)]
pub struct SdkConfig {
    pub coverage: Option<CoverageConfig>,
    pub horizon_api: Option<HorizonApiConfig>,
    pub iceberg: Option<IcebergConfig>,
    pub kafka: Option<KafkaConfig>,
    pub postgres: PostgresConfig,
}

/// Serde default for `KafkaConfig::bootstrap_servers`.
fn default_bootstrap_servers() -> String {
    constants::DEFAULT_BOOTSTRAP_SERVERS.to_owned()
}

/// Serde default for `IcebergConfig::catalog_uri`.
fn default_catalog_uri() -> String {
    constants::DEFAULT_CATALOG_URI.to_owned()
}

/// Serde default for `KafkaConfig::compression_type`.
fn default_compression_type() -> String {
    constants::DEFAULT_COMPRESSION_TYPE.to_owned()
}

/// Serde default for `CoverageConfig::timeout_secs`.
const fn default_coverage_timeout_secs() -> u64 {
    constants::DEFAULT_COVERAGE_TIMEOUT_SECS
}

/// Serde default for `HorizonApiConfig::base_url`.
fn default_horizon_api_base_url() -> String {
    constants::DEFAULT_HORIZON_API_BASE_URL.to_owned()
}

/// Serde default for `HorizonApiConfig::timeout_secs`.
const fn default_horizon_api_timeout_secs() -> u64 {
    constants::DEFAULT_HORIZON_API_TIMEOUT_SECS
}

/// Serde default for `IcebergConfig::data_row_table`.
fn default_data_row_table() -> String {
    constants::DEFAULT_DATA_ROW_TABLE.to_owned()
}

/// Serde default for `KafkaConfig::data_row_topic`.
fn default_data_row_topic() -> String {
    constants::DEFAULT_DATA_ROW_TOPIC.to_owned()
}

/// Serde default for `IcebergConfig::metadata_row_table`.
fn default_metadata_row_table() -> String {
    constants::DEFAULT_METADATA_ROW_TABLE.to_owned()
}

/// Serde default for `KafkaConfig::metadata_row_topic`.
fn default_metadata_row_topic() -> String {
    constants::DEFAULT_METADATA_ROW_TOPIC.to_owned()
}

/// Serde default for `KafkaConfig::sasl_mechanism`.
fn default_sasl_mechanism() -> String {
    constants::DEFAULT_SASL_MECHANISM.to_owned()
}

/// Serde default for `KafkaConfig::security_protocol`.
fn default_security_protocol() -> String {
    constants::DEFAULT_SECURITY_PROTOCOL.to_owned()
}

/// Serde default for `PostgresConfig::sslmode`.
fn default_sslmode() -> String {
    constants::DEFAULT_SSLMODE.to_owned()
}

#[cfg(test)]
mod tests {
    use std::env;
    use std::error::Error;
    use std::fmt::Debug;
    use std::fs;
    use std::io;
    use std::path::PathBuf;
    use std::sync::{Mutex, MutexGuard, PoisonError};

    use config::{Config, File, FileFormat};
    use uuid::Uuid;

    use super::*;
    use crate::load_config;

    const ENV_API_KEY: &str = "HORIZON__HORIZON_API__API_KEY";
    const ENV_BASE_URL: &str = "HORIZON__HORIZON_API__BASE_URL";
    const ENV_TIMEOUT_SECS: &str = "HORIZON__HORIZON_API__TIMEOUT_SECS";

    static ENV_MUTEX: Mutex<()> = Mutex::new(());

    struct EnvOverride {
        _lock: MutexGuard<'static, ()>,
        saved: Vec<(String, Option<String>)>,
    }

    impl EnvOverride {
        fn apply(pairs: &[(&str, Option<&str>)]) -> Self {
            let lock = ENV_MUTEX.lock().unwrap_or_else(PoisonError::into_inner);
            let mut saved = Vec::with_capacity(pairs.len());
            for (key, next_value) in pairs {
                saved.push(((*key).to_owned(), env::var(key).ok()));
                match *next_value {
                    Some(value) => {
                        // SAFETY: tests hold `ENV_MUTEX` while mutating the process environment.
                        unsafe {
                            env::set_var(key, value);
                        }
                    }
                    None => {
                        // SAFETY: tests hold `ENV_MUTEX` while mutating the process environment.
                        unsafe {
                            env::remove_var(key);
                        }
                    }
                }
            }
            Self { _lock: lock, saved }
        }
    }

    impl Drop for EnvOverride {
        fn drop(&mut self) {
            for (key, saved_value) in &self.saved {
                match saved_value.as_deref() {
                    Some(value) => {
                        // SAFETY: tests hold `ENV_MUTEX` while mutating the process environment.
                        unsafe {
                            env::set_var(key, value);
                        }
                    }
                    None => {
                        // SAFETY: tests hold `ENV_MUTEX` while mutating the process environment.
                        unsafe {
                            env::remove_var(key);
                        }
                    }
                }
            }
        }
    }

    fn check_eq<T>(left: &T, right: &T) -> Result<(), String>
    where
        T: Debug + PartialEq + ?Sized,
    {
        if left == right {
            Ok(())
        } else {
            Err(format!("{left:?} != {right:?}"))
        }
    }

    fn minimal_sdk_config_toml(horizon_api_toml: Option<&str>) -> String {
        let mut toml = String::from(
            "[postgres]\n\
             host = \"localhost\"\n\
             port = 5432\n\
             user = \"test\"\n\
             password = \"test\"\n\
             database = \"horizon\"\n",
        );
        if let Some(section) = horizon_api_toml {
            toml.push_str("\n[horizon_api]\n");
            toml.push_str(section);
        }
        toml
    }

    fn write_temp_config(contents: &str) -> io::Result<PathBuf> {
        let path = env::temp_dir().join(format!("horizon-sdk-config-test-{}.toml", Uuid::new_v4()));
        fs::write(&path, contents)?;
        Ok(path)
    }

    fn remove_temp_config(path: PathBuf) -> io::Result<()> {
        fs::remove_file(path)
    }

    #[test]
    fn horizon_api_config_debug_redacts_api_key() {
        const SENTINEL_API_KEY: &str = "sentinel-api-key-do-not-leak";

        let config = HorizonApiConfig {
            api_key: SENTINEL_API_KEY.to_owned(),
            base_url: default_horizon_api_base_url(),
            timeout_secs: default_horizon_api_timeout_secs(),
        };

        let debug = format!("{config:?}");

        assert!(!debug.contains(SENTINEL_API_KEY));
        assert!(debug.contains("[REDACTED]"));
    }

    #[test]
    fn horizon_api_config_serde_defaults_missing_optional_fields() -> Result<(), Box<dyn Error>> {
        let config: HorizonApiConfig = Config::builder()
            .add_source(File::from_str(
                r#"api_key = "toml-api-key""#,
                FileFormat::Toml,
            ))
            .build()?
            .try_deserialize()?;

        check_eq(config.api_key.as_str(), "toml-api-key")?;
        check_eq(
            config.base_url.as_str(),
            constants::DEFAULT_HORIZON_API_BASE_URL,
        )?;
        check_eq(
            &config.timeout_secs,
            &constants::DEFAULT_HORIZON_API_TIMEOUT_SECS,
        )?;
        Ok(())
    }

    #[test]
    fn load_config_loads_horizon_api_from_toml() -> Result<(), Box<dyn Error>> {
        let _env = EnvOverride::apply(&[
            (ENV_API_KEY, None),
            (ENV_BASE_URL, None),
            (ENV_TIMEOUT_SECS, None),
        ]);

        let toml = minimal_sdk_config_toml(Some(
            r#"api_key = "toml-api-key"
base_url = "https://toml.example.com"
timeout_secs = 45
"#,
        ));
        let path = write_temp_config(&toml)?;
        let config_path = path.to_string_lossy().into_owned();

        let config = load_config(Some(&config_path))?;
        let horizon_api = config.horizon_api.ok_or("expected horizon_api")?;

        check_eq(horizon_api.api_key.as_str(), "toml-api-key")?;
        check_eq(horizon_api.base_url.as_str(), "https://toml.example.com")?;
        check_eq(&horizon_api.timeout_secs, &45)?;

        remove_temp_config(path)?;
        Ok(())
    }

    #[test]
    fn load_config_applies_horizon_api_serde_defaults_from_toml() -> Result<(), Box<dyn Error>> {
        let _env = EnvOverride::apply(&[
            (ENV_API_KEY, None),
            (ENV_BASE_URL, None),
            (ENV_TIMEOUT_SECS, None),
        ]);

        let toml = minimal_sdk_config_toml(Some(r#"api_key = "toml-api-key""#));
        let path = write_temp_config(&toml)?;
        let config_path = path.to_string_lossy().into_owned();

        let config = load_config(Some(&config_path))?;
        let horizon_api = config.horizon_api.ok_or("expected horizon_api")?;

        check_eq(horizon_api.api_key.as_str(), "toml-api-key")?;
        check_eq(
            horizon_api.base_url.as_str(),
            constants::DEFAULT_HORIZON_API_BASE_URL,
        )?;
        check_eq(
            &horizon_api.timeout_secs,
            &constants::DEFAULT_HORIZON_API_TIMEOUT_SECS,
        )?;

        remove_temp_config(path)?;
        Ok(())
    }

    #[test]
    fn load_config_env_overrides_toml_for_horizon_api() -> Result<(), Box<dyn Error>> {
        let _env = EnvOverride::apply(&[
            (ENV_API_KEY, Some("env-api-key")),
            (ENV_BASE_URL, Some("https://env.example.com")),
            (ENV_TIMEOUT_SECS, Some("60")),
        ]);

        let toml = minimal_sdk_config_toml(Some(
            r#"api_key = "toml-api-key"
base_url = "https://toml.example.com"
timeout_secs = 45
"#,
        ));
        let path = write_temp_config(&toml)?;
        let config_path = path.to_string_lossy().into_owned();

        let config = load_config(Some(&config_path))?;
        let horizon_api = config.horizon_api.ok_or("expected horizon_api")?;

        check_eq(horizon_api.api_key.as_str(), "env-api-key")?;
        check_eq(horizon_api.base_url.as_str(), "https://env.example.com")?;
        check_eq(&horizon_api.timeout_secs, &60)?;

        remove_temp_config(path)?;
        Ok(())
    }

    #[test]
    fn load_config_loads_horizon_api_from_environment_without_toml_section()
    -> Result<(), Box<dyn Error>> {
        let _env = EnvOverride::apply(&[
            (ENV_API_KEY, Some("env-api-key")),
            (ENV_BASE_URL, Some("https://env.example.com")),
            (ENV_TIMEOUT_SECS, Some("60")),
        ]);

        let toml = minimal_sdk_config_toml(None);
        let path = write_temp_config(&toml)?;
        let config_path = path.to_string_lossy().into_owned();

        let config = load_config(Some(&config_path))?;
        let horizon_api = config.horizon_api.ok_or("expected horizon_api")?;

        check_eq(horizon_api.api_key.as_str(), "env-api-key")?;
        check_eq(horizon_api.base_url.as_str(), "https://env.example.com")?;
        check_eq(&horizon_api.timeout_secs, &60)?;

        remove_temp_config(path)?;
        Ok(())
    }
}