liminal-server 0.25.4

use std::ffi::OsString;
use std::path::Path;

use crate::ServerError;

use super::env::apply_env_overrides_from;
use super::types::ServerConfig;
use super::validation::validate;

/// Loads a server configuration from a TOML file.
///
/// # Errors
///
/// Returns [`ServerError::ConfigLoad`] when the file cannot be read, the TOML is
/// malformed, or strict deserialization rejects an unknown field.
pub fn load_from_file(path: impl AsRef<Path>) -> Result<ServerConfig, ServerError> {
    let path = path.as_ref();
    let contents = std::fs::read_to_string(path).map_err(|error| ServerError::ConfigLoad {
        message: format!(
            "failed to read configuration file '{}': {error}",
            path.display()
        ),
    })?;

    toml::from_str::<ServerConfig>(&contents).map_err(|error| ServerError::ConfigLoad {
        message: format!(
            "failed to parse configuration file '{}': {error}",
            path.display()
        ),
    })
}

/// Loads and validates the configuration before any server resources are opened.
///
/// Startup and configuration checks share file parsing, environment overrides,
/// semantic validation and schema resolution. This only reads files and metadata;
/// it opens no store, starts no service and performs no network I/O.
///
/// # Errors
///
/// Returns [`ServerError::ConfigLoad`] for an empty path, unreadable file or
/// malformed configuration, and [`ServerError::ConfigValidation`] for refusals.
pub fn load_config(path: impl AsRef<Path>) -> Result<ServerConfig, ServerError> {
    if path.as_ref().as_os_str().is_empty() {
        return Err(ServerError::ConfigLoad {
            message: "configuration path is empty".to_owned(),
        });
    }
    load_config_with_env(path.as_ref(), std::env::vars_os())
}

fn load_config_with_env(
    path: &Path,
    variables: impl IntoIterator<Item = (OsString, OsString)>,
) -> Result<ServerConfig, ServerError> {
    let config = load_from_file(path)?;
    let mut config = apply_env_overrides_from(config, variables)?;
    // Channel `schema_ref` paths are resolved relative to the directory holding
    // the config file, so validation loads each schema from there.
    validate(&mut config, path.parent())?;
    Ok(config)
}

#[cfg(test)]
mod tests {
    use std::fs;
    use std::path::{Path, PathBuf};
    use std::sync::atomic::{AtomicU64, Ordering};

    use crate::ServerError;

    use super::{load_config_with_env, load_from_file};

    static NEXT_TEMP_FILE_ID: AtomicU64 = AtomicU64::new(0);

    /// Absolute path to the config example shipped in the repository.
    ///
    /// Resolved from `CARGO_MANIFEST_DIR` (`<repo>/crates/liminal-server`) rather
    /// than the process working directory, so the test finds the file identically
    /// under `cargo test`, `cargo nextest`, and any invocation directory.
    fn shipped_example_config_path() -> PathBuf {
        Path::new(env!("CARGO_MANIFEST_DIR"))
            .join("..")
            .join("..")
            .join("config")
            .join("liminal.example.toml")
    }

    /// The shipped example must boot-load through the SAME entry point the binary
    /// uses — `load_config` (file parse + environment overrides + validation), the
    /// one `server::runtime::run` calls. This is the anti-rot pin: an example that
    /// drifts from the schema, names an unknown field, or references a channel that
    /// does not exist stops the build here rather than at a newcomer's first boot.
    #[test]
    fn shipped_example_config_loads_through_the_real_loader()
    -> Result<(), Box<dyn std::error::Error>> {
        let path = shipped_example_config_path();
        let directory = tempfile::tempdir()?;
        let config = load_config_with_env(
            &path,
            [(
                "LIMINAL_PERSISTENCE_PATH".into(),
                directory.path().as_os_str().to_owned(),
            )],
        )
        .map_err(|error| {
            format!(
                "the shipped example config '{}' must load and validate through the real loader: \
                 {error}",
                path.display()
            )
        })?;

        // A newcomer's first boot needs at least one channel and the mandatory
        // routing_rules key populated, not an empty husk.
        assert!(
            !config.channels.is_empty(),
            "the example must declare at least one channel"
        );
        assert!(
            !config.routing_rules.is_empty(),
            "the example must exercise the mandatory routing_rules key"
        );
        assert_eq!(config.persistence_path.as_deref(), Some(directory.path()));
        assert!(load_from_file(&path)?.persistence_path.is_none());
        let refused = load_config_with_env(&path, []);
        assert!(
            matches!(refused, Err(ServerError::ConfigValidation { message }) if message.contains("persistence_path") && message.contains("LIMINAL_PERSISTENCE_PATH") && message.contains("no store is opened"))
        );

        Ok(())
    }

    fn valid_toml() -> &'static str {
        r#"
listen_address = "127.0.0.1:8080"
health_listen_address = "127.0.0.1:8081"
drain_timeout_ms = 30000
persistence_path = "/tmp"

[[channels]]
name = "orders"
schema_ref = "schemas/orders.json"
durable = true

[[routing_rules]]
source_channel = "orders"
target_channel = "orders"
predicate = "true"

[cluster]
node_name = "node-a"
listen_address = "127.0.0.1:9000"
seed_nodes = ["127.0.0.1:9001"]
cookie = "test-cookie"
"#
    }

    fn temp_config_path(label: &str) -> PathBuf {
        let id = NEXT_TEMP_FILE_ID.fetch_add(1, Ordering::Relaxed);
        std::env::temp_dir().join(format!(
            "liminal-server-{label}-{}-{id}.toml",
            std::process::id()
        ))
    }

    fn write_temp_config(
        label: &str,
        contents: &str,
    ) -> Result<PathBuf, Box<dyn std::error::Error>> {
        let path = temp_config_path(label);
        fs::write(&path, contents)?;
        Ok(path)
    }

    fn remove_temp_file(path: &PathBuf) -> Result<(), Box<dyn std::error::Error>> {
        if path.exists() {
            fs::remove_file(path)?;
        }
        Ok(())
    }

    #[test]
    fn valid_toml_parses_into_server_config() -> Result<(), Box<dyn std::error::Error>> {
        let path = write_temp_config("valid", valid_toml())?;
        let config = load_from_file(&path)?;
        remove_temp_file(&path)?;

        assert_eq!(config.listen_address.to_string(), "127.0.0.1:8080");
        assert_eq!(config.health_listen_address.to_string(), "127.0.0.1:8081");
        assert_eq!(config.drain_timeout_ms, 30_000);
        assert_eq!(config.channels.len(), 1);
        assert_eq!(config.channels[0].name, "orders");
        assert_eq!(config.routing_rules.len(), 1);
        assert_eq!(
            config.persistence_path.as_deref(),
            Some(std::path::Path::new("/tmp"))
        );
        let cluster = config
            .cluster
            .as_ref()
            .ok_or("cluster section should be present")?;
        assert_eq!(cluster.node_name, "node-a");
        assert_eq!(cluster.listen_address.to_string(), "127.0.0.1:9000");
        assert_eq!(cluster.seed_nodes.len(), 1);
        assert_eq!(cluster.cookie, "test-cookie");

        Ok(())
    }

    #[test]
    fn websocket_section_parses_and_absent_section_stays_none()
    -> Result<(), Box<dyn std::error::Error>> {
        // Absent section: no websocket configuration exists at all.
        let absent_path = write_temp_config("ws-absent", valid_toml())?;
        let absent = load_from_file(&absent_path)?;
        remove_temp_file(&absent_path)?;
        assert!(absent.websocket.is_none());

        // Present section: every field parses, including the optional
        // keepalive interval and origin allow-list.
        let toml = format!(
            "{}\n[websocket]\nlisten_address = \"127.0.0.1:8090\"\npath = \"/liminal\"\n\
             allowed_origins = [\"https://app.example.com\"]\nping_interval_ms = 30000\n",
            valid_toml()
        );
        let path = write_temp_config("ws-present", &toml)?;
        let config = load_from_file(&path)?;
        remove_temp_file(&path)?;
        let websocket = config.websocket.ok_or("websocket section should parse")?;
        assert_eq!(websocket.listen_address.to_string(), "127.0.0.1:8090");
        assert_eq!(websocket.path, "/liminal");
        assert_eq!(
            websocket.allowed_origins,
            vec!["https://app.example.com".to_owned()]
        );
        assert_eq!(websocket.ping_interval_ms, Some(30_000));

        // Minimal section: origins default to the fail-closed empty list and
        // the keepalive stays disabled.
        let minimal = format!(
            "{}\n[websocket]\nlisten_address = \"127.0.0.1:8091\"\npath = \"/liminal\"\n",
            valid_toml()
        );
        let minimal_path = write_temp_config("ws-minimal", &minimal)?;
        let minimal_config = load_from_file(&minimal_path)?;
        remove_temp_file(&minimal_path)?;
        let websocket = minimal_config
            .websocket
            .ok_or("minimal websocket section should parse")?;
        assert!(websocket.allowed_origins.is_empty());
        assert_eq!(websocket.ping_interval_ms, None);
        Ok(())
    }

    /// A minimal config that validates from ANY directory: no `schema_ref`, so
    /// validation has no file to resolve relative to the temp dir the limits pins
    /// write into. `valid_toml` deliberately carries a schema reference, which
    /// makes it the wrong fixture for a pin about `[limits]`.
    fn schema_free_toml() -> &'static str {
        r#"
listen_address = "127.0.0.1:8080"
health_listen_address = "127.0.0.1:8081"
drain_timeout_ms = 30000

[[channels]]
name = "orders"
durable = false

[[routing_rules]]
source_channel = "orders"
target_channel = "orders"
"#
    }

    fn cluster_toml(cookie: Option<&str>) -> String {
        let cookie = cookie.map_or_else(String::new, |value| format!("cookie = {value:?}\n"));
        format!(
            "{}\n[cluster]\nnode_name = \"node-a\"\nlisten_address = \"127.0.0.1:9000\"\n\
             seed_nodes = []\n{cookie}",
            schema_free_toml()
        )
    }

    #[test]
    fn the_default_cookie_is_refused_at_validation_by_name()
    -> Result<(), Box<dyn std::error::Error>> {
        let directory = tempfile::tempdir()?;
        let path = directory.path().join("liminal.toml");
        fs::write(
            &path,
            cluster_toml(Some(crate::config::types::DEFAULT_COOKIE)),
        )?;
        let result = load_config_with_env(
            &path,
            [(
                "LIMINAL_PERSISTENCE_PATH".into(),
                directory.path().as_os_str().to_owned(),
            )],
        );
        let Err(ServerError::ConfigValidation { message }) = result else {
            return Err("cluster.cookie: expected validation to require a secret cookie".into());
        };
        assert!(message.contains("cluster.cookie"));
        assert!(message.contains("secret cookie must be set"));
        assert!(!message.contains(crate::config::types::DEFAULT_COOKIE));
        Ok(())
    }

    #[test]
    fn an_omitted_cookie_is_refused() -> Result<(), Box<dyn std::error::Error>> {
        let directory = tempfile::tempdir()?;
        let path = directory.path().join("liminal.toml");
        fs::write(&path, cluster_toml(None))?;
        let result = load_from_file(&path);
        let Err(ServerError::ConfigLoad { message }) = result else {
            return Err("cluster.cookie: expected a missing cookie to fail deserialization".into());
        };
        assert!(message.contains("missing field `cookie`"));
        Ok(())
    }

    #[test]
    fn an_explicit_cookie_is_accepted() -> Result<(), Box<dyn std::error::Error>> {
        let directory = tempfile::tempdir()?;
        let path = directory.path().join("liminal.toml");
        fs::write(&path, cluster_toml(Some("explicit-test-cookie")))?;
        let config = load_config_with_env(
            &path,
            [(
                "LIMINAL_PERSISTENCE_PATH".into(),
                directory.path().as_os_str().to_owned(),
            )],
        )?;
        let cluster = config
            .cluster
            .ok_or("cluster section must remain present")?;
        assert_eq!(cluster.cookie, "explicit-test-cookie");
        Ok(())
    }

    /// P0 #55 part 2: the two delivery caps are DEFAULTS, not constants, and the
    /// operator's number must survive the WHOLE pipeline — file parse, environment
    /// overrides, validation — not just `serde`.
    ///
    /// `load_config` is deliberately the entry point here rather than
    /// `load_from_file`: the environment-override pass runs between parse and
    /// validation, and a pin that stopped at the parse would go green on a loader
    /// that silently reset limits afterwards.
    ///
    /// The section sets only TWO of the nine caps on purpose. A partial `[limits]`
    /// table is the shape an operator actually writes, and it is the shape that
    /// catches a `#[serde(default)]` regression on the whole struct: if the caps
    /// ever stop defaulting per FIELD, the seven untouched ones collapse to zero
    /// and validation refuses the file.
    #[test]
    fn operator_set_delivery_caps_survive_the_whole_load_pipeline()
    -> Result<(), Box<dyn std::error::Error>> {
        let operator_toml = format!(
            "{}\n[limits]\nmax_subscription_inbox_depth = 9001\ndelivery_slice_budget = 7\n",
            schema_free_toml()
        );
        let path = write_temp_config("operator-limits", &operator_toml)?;

        let directory = tempfile::tempdir()?;
        let config = load_config_with_env(
            &path,
            [(
                "LIMINAL_PERSISTENCE_PATH".into(),
                directory.path().as_os_str().to_owned(),
            )],
        )?;

        assert_eq!(config.limits.max_subscription_inbox_depth, 9001);
        assert_eq!(config.limits.delivery_slice_budget, 7);
        // The eight caps the operator did NOT name still carry their defaults —
        // the per-field default survived a partially-populated section.
        assert_eq!(config.limits.max_connections, 256);
        assert_eq!(config.limits.max_connection_inbox_bytes, 4 * 1024 * 1024);
        assert_eq!(config.limits.max_connection_outbound_bytes, 4 * 1024 * 1024);

        remove_temp_file(&path)?;
        Ok(())
    }

    /// The other half of the pin above: an ABSENT `[limits]` section resolves both
    /// delivery caps to the shipped defaults through the same pipeline. Without
    /// this, the override pin alone could not tell a working default from a
    /// coincidence — it only ever observes numbers the file supplied.
    #[test]
    fn absent_limits_section_resolves_the_shipped_delivery_defaults()
    -> Result<(), Box<dyn std::error::Error>> {
        let path = write_temp_config("absent-limits", schema_free_toml())?;

        let directory = tempfile::tempdir()?;
        let config = load_config_with_env(
            &path,
            [(
                "LIMINAL_PERSISTENCE_PATH".into(),
                directory.path().as_os_str().to_owned(),
            )],
        )?;

        assert_eq!(config.limits.max_subscription_inbox_depth, 4096);
        assert_eq!(config.limits.delivery_slice_budget, 32);

        remove_temp_file(&path)?;
        Ok(())
    }

    #[test]
    fn missing_file_returns_config_load() {
        let path = temp_config_path("missing");
        let result = load_from_file(&path);

        assert!(matches!(result, Err(ServerError::ConfigLoad { .. })));
    }

    #[test]
    fn malformed_toml_returns_config_load_with_parse_details()
    -> Result<(), Box<dyn std::error::Error>> {
        let path = write_temp_config("malformed", "listen_address =")?;
        let result = load_from_file(&path);
        remove_temp_file(&path)?;

        assert!(matches!(result, Err(ServerError::ConfigLoad { .. })));
        let Err(ServerError::ConfigLoad { message }) = result else {
            return Ok(());
        };
        assert!(message.contains("parse"));

        Ok(())
    }

    #[test]
    fn unknown_fields_return_config_load() -> Result<(), Box<dyn std::error::Error>> {
        let toml = format!("{}\nunknown_field = true\n", valid_toml());
        let path = write_temp_config("unknown", &toml)?;
        let result = load_from_file(&path);
        remove_temp_file(&path)?;

        assert!(matches!(result, Err(ServerError::ConfigLoad { .. })));
        let Err(ServerError::ConfigLoad { message }) = result else {
            return Ok(());
        };
        assert!(message.contains("unknown") || message.contains("unexpected"));

        Ok(())
    }
}