feat: new config system

Author: raphael-goetz

src/configuration/config.rs

@@ -1,113 +1,278 @@
-use code0_flow::flow_config::{env_with_default, environment::Environment, mode::Mode};
+use std::{fmt, path::Path};
 
-/// Struct for all relevant `Aquila` startup configurations
-pub struct Config {
-    /// Aquila environment
-    ///
-    /// Options:
-    /// `development` (default)
-    /// `staging`
-    /// `production`
-    pub environment: Environment,
+use config::{Config as ConfigLoader, ConfigError, File};
+use serde::{Deserialize, Serialize};
 
-    /// Aquila mode
-    ///
-    /// Options:
-    /// `static` (default)
-    /// `dynamic`
-    pub mode: Mode,
+use super::{env::Environment, mode::Mode};
 
-    /// URL to the NATS Server.
-    pub nats_url: String,
+const CONFIG_FILE: &str = "aquila";
+const BACKEND_TOKEN_ENV: &str = "AQUILA_BACKEND_TOKEN";
 
-    /// Name of the NATS Bucket.
-    pub nats_bucket: String,
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(default)]
+pub struct Config {
+    pub environment: Environment,
+    pub mode: Mode,
+    pub log_level: String,
+    pub nats: Nats,
+    pub static_config: StaticConfig,
+    pub dynamic_config: DynamicConfig,
+    pub grpc: Grpc,
+    pub runtime_status: RuntimeStatus,
+}
 
-    /// Fallback file to load flows if gRPC & scheduling is disabled.
-    pub flow_fallback_path: String,
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(default)]
+pub struct Nats {
+    pub url: String,
+    pub bucket: String,
+}
 
-    /// Verification Token required for internal communication
-    pub runtime_token: String,
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(default)]
+pub struct StaticConfig {
+    pub flow_path: String,
+}
 
-    /// URL to the `Sagittarius` Server.
+#[derive(Clone, Deserialize, Serialize)]
+#[serde(default)]
+pub struct DynamicConfig {
     pub backend_url: String,
+    pub backend_token: String,
+    pub backend_unary_timeout_secs: u64,
+}
 
-    // Port of the `Aquila` Server
-    pub grpc_port: u16,
+impl std::fmt::Debug for DynamicConfig {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        formatter
+            .debug_struct("DynamicConfig")
+            .field("backend_url", &self.backend_url)
+            .field("backend_token", &"[FILTERED]")
+            .field(
+                "backend_unary_timeout_secs",
+                &self.backend_unary_timeout_secs,
+            )
+            .finish()
+    }
+}
 
-    // Host of the `Aquila` Server
-    pub grpc_host: String,
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(default)]
+pub struct Grpc {
+    pub host: String,
+    pub port: u16,
+    pub health_service: bool,
+}
 
-    pub with_health_service: bool,
+#[derive(Clone, Debug, Deserialize, Serialize)]
+#[serde(default)]
+pub struct RuntimeStatus {
+    pub not_responding_after_secs: u64,
+    pub stopped_after_not_responding_secs: u64,
+    pub monitor_interval_secs: u64,
+}
 
-    pub service_config_path: String,
+impl Default for Config {
+    fn default() -> Self {
+        Self {
+            environment: Environment::Development,
+            mode: Mode::Static,
+            log_level: "debug".into(),
+            nats: Nats::default(),
+            static_config: StaticConfig::default(),
+            dynamic_config: DynamicConfig::default(),
+            grpc: Grpc::default(),
+            runtime_status: RuntimeStatus::default(),
+        }
+    }
+}
 
-    /// Runtime heartbeat timeout in seconds before a service is marked as NOT_RESPONDING.
-    pub runtime_status_not_responding_after_secs: u64,
+impl Default for Nats {
+    fn default() -> Self {
+        Self {
+            url: "nats://localhost:4222".into(),
+            bucket: "flow_store".into(),
+        }
+    }
+}
 
-    /// Additional timeout in seconds after NOT_RESPONDING before a service is marked as STOPPED.
-    pub runtime_status_stopped_after_not_responding_secs: u64,
+impl Default for StaticConfig {
+    fn default() -> Self {
+        Self {
+            flow_path: "./flowExport.json".into(),
+        }
+    }
+}
 
-    /// Interval in seconds for the runtime status timeout monitor loop.
-    pub runtime_status_monitor_interval_secs: u64,
+impl Default for DynamicConfig {
+    fn default() -> Self {
+        Self {
+            backend_url: "http://localhost:50051".into(),
+            backend_token: "default_session_token".into(),
+            backend_unary_timeout_secs: 5,
+        }
+    }
+}
 
-    /// Timeout in seconds for unary RPC calls from Aquila to Sagittarius.
-    pub sagittarius_unary_rpc_timeout_secs: u64,
+impl Default for Grpc {
+    fn default() -> Self {
+        Self {
+            host: "127.0.0.1".into(),
+            port: 8081,
+            health_service: false,
+        }
+    }
 }
 
-/// Implementation for all relevant `Aquila` startup configurations
-///
-/// Behavior:
-/// Searches for the env. file at root level. Filename: `.env`
-impl Default for Config {
+impl Default for RuntimeStatus {
     fn default() -> Self {
-        Self::new()
+        Self {
+            not_responding_after_secs: 90,
+            stopped_after_not_responding_secs: 180,
+            monitor_interval_secs: 30,
+        }
     }
 }
 
 impl Config {
     pub fn new() -> Self {
-        Config {
-            environment: env_with_default("ENVIRONMENT", Environment::Development),
-            mode: env_with_default("MODE", Mode::STATIC),
-            nats_url: env_with_default("NATS_URL", String::from("nats://localhost:4222")),
-            nats_bucket: env_with_default("NATS_BUCKET", String::from("flow_store")),
-            flow_fallback_path: env_with_default(
-                "FLOW_FALLBACK_PATH",
-                String::from("./flowExport.json"),
-            ),
-            grpc_port: env_with_default("GRPC_PORT", 8081),
-            grpc_host: env_with_default("GRPC_HOST", String::from("127.0.0.1")),
-            with_health_service: env_with_default("WITH_HEALTH_SERVICE", false),
-            runtime_token: env_with_default("RUNTIME_TOKEN", String::from("default_session_token")),
-            backend_url: env_with_default(
-                "SAGITTARIUS_URL",
-                String::from("http://localhost:50051"),
-            ),
-            service_config_path: env_with_default(
-                "SERVICE_CONFIG_PATH",
-                String::from("./service.configuration.json"),
-            ),
-            runtime_status_not_responding_after_secs: env_with_default(
-                "RUNTIME_STATUS_NOT_RESPONDING_AFTER_SECS",
-                90_u64,
-            ),
-            runtime_status_stopped_after_not_responding_secs: env_with_default(
-                "RUNTIME_STATUS_STOPPED_AFTER_NOT_RESPONDING_SECS",
-                180_u64,
-            ),
-            runtime_status_monitor_interval_secs: env_with_default(
-                "RUNTIME_STATUS_MONITOR_INTERVAL_SECS",
-                30_u64,
-            ),
-            sagittarius_unary_rpc_timeout_secs: env_with_default(
-                "SAGITTARIUS_UNARY_RPC_TIMEOUT_SECS",
-                5_u64,
-            ),
+        Self::try_new()
+            .unwrap_or_else(|error| panic!("failed to load Aquila configuration: {error}"))
+    }
+
+    pub fn try_new() -> Result<Self, ConfigError> {
+        Self::try_from_optional_path(None)
+    }
+
+    pub fn try_from_path(path: impl AsRef<Path>) -> Result<Self, ConfigError> {
+        Self::try_from_optional_path(Some(path.as_ref()))
+    }
+
+    fn try_from_optional_path(path: Option<&Path>) -> Result<Self, ConfigError> {
+        let mut builder =
+            ConfigLoader::builder().add_source(ConfigLoader::try_from(&Self::default())?);
+
+        builder = match path {
+            Some(path) => builder.add_source(File::from(path).required(true)),
+            None => builder.add_source(File::with_name(CONFIG_FILE).required(false)),
+        };
+
+        if let Ok(token) = std::env::var(BACKEND_TOKEN_ENV) {
+            builder = builder.set_override("dynamic_config.backend_token", token)?;
         }
+
+        builder.build()?.try_deserialize()
     }
 
     pub fn is_static(&self) -> bool {
-        self.mode == Mode::STATIC
+        self.mode == Mode::Static
+    }
+}
+
+impl fmt::Display for Config {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        writeln!(formatter, "Aquila configuration")?;
+        writeln!(formatter, "  Environment: {}", self.environment)?;
+        writeln!(formatter, "  Mode:        {}", self.mode)?;
+        writeln!(formatter, "  Log level:   {}", self.log_level)?;
+        writeln!(formatter, "  NATS")?;
+        writeln!(formatter, "    URL:       {}", self.nats.url)?;
+        writeln!(formatter, "    Bucket:    {}", self.nats.bucket)?;
+        writeln!(formatter, "  gRPC")?;
+        writeln!(
+            formatter,
+            "    Address:   {}:{}",
+            self.grpc.host, self.grpc.port
+        )?;
+        writeln!(
+            formatter,
+            "    Health service: {}",
+            self.grpc.health_service
+        )?;
+        writeln!(formatter, "  Static mode")?;
+        writeln!(formatter, "    Flow path: {}", self.static_config.flow_path)?;
+        writeln!(formatter, "  Dynamic mode")?;
+        writeln!(
+            formatter,
+            "    Backend URL:     {}",
+            self.dynamic_config.backend_url
+        )?;
+        writeln!(formatter, "    Backend token:   [FILTERED]")?;
+        writeln!(
+            formatter,
+            "    Request timeout: {}s",
+            self.dynamic_config.backend_unary_timeout_secs
+        )?;
+        writeln!(formatter, "  Runtime status")?;
+        writeln!(
+            formatter,
+            "    Not responding after: {}s",
+            self.runtime_status.not_responding_after_secs
+        )?;
+        writeln!(
+            formatter,
+            "    Stopped after:        {}s",
+            self.runtime_status.stopped_after_not_responding_secs
+        )?;
+        write!(
+            formatter,
+            "    Monitor interval:     {}s",
+            self.runtime_status.monitor_interval_secs
+        )
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Mutex;
+
+    use super::Config;
+
+    static ENV_LOCK: Mutex<()> = Mutex::new(());
+
+    #[test]
+    fn environment_overrides_backend_token() {
+        let _guard = ENV_LOCK.lock().expect("environment test lock poisoned");
+
+        // SAFETY: access to these process-wide variables is serialized for this test.
+        unsafe {
+            std::env::set_var("AQUILA_BACKEND_TOKEN", "environment-token");
+        }
+
+        let config = Config::try_new().expect("configuration should load");
+
+        // SAFETY: access to these process-wide variables is serialized for this test.
+        unsafe {
+            std::env::remove_var("AQUILA_BACKEND_TOKEN");
+        }
+
+        assert_eq!(config.dynamic_config.backend_token, "environment-token");
+    }
+
+    #[test]
+    fn debug_output_filters_backend_token() {
+        let mut config = Config::default();
+        config.dynamic_config.backend_token = "super-secret".into();
+
+        let output = format!("{config:#?}");
+
+        assert!(output.contains("[FILTERED]"));
+        assert!(!output.contains("super-secret"));
+    }
+
+    #[test]
+    fn display_output_is_readable_and_filters_backend_token() {
+        let mut config = Config::default();
+        config.dynamic_config.backend_token = "super-secret".into();
+
+        let output = config.to_string();
+
+        assert!(output.starts_with("Aquila configuration\n"));
+        assert!(output.contains("  Environment: development"));
+        assert!(output.contains("    Address:   127.0.0.1:8081"));
+        assert!(output.contains("    Request timeout: 5s"));
+        assert!(output.contains("    Backend token:   [FILTERED]"));
+        assert!(!output.contains("super-secret"));
+        assert!(!output.contains("Config {"));
     }
 }