ref: major codebase refactoring

Author: raphael-goetz

src/authorization/mod.rs

@@ -1,3 +1,8 @@
+//! Bearer-token helpers shared by every gRPC client and server in Aquila:
+//! [`authorization::get_authorization_metadata`] to attach a token to an
+//! outgoing request, [`authorization::extract_token`] to read one back off
+//! an incoming request.
+
 pub mod authorization {
     use std::str::FromStr;
     use tonic::{
@@ -32,6 +37,9 @@ pub mod authorization {
         map
     }
 
+    /// Reads and validates the bearer token off an incoming request's
+    /// `authorization` header. Generic over the request body type so it
+    /// works for both unary and streaming gRPC requests.
     pub fn extract_token<T>(request: &Request<T>) -> Result<&str, Status> {
         let header = request.metadata().get("authorization").ok_or_else(|| {
             log::warn!("Missing authorization header");

src/configuration/config/display.rs

@@ -0,0 +1,111 @@
+//! Human-readable rendering of [`Config`] for the startup log line, kept
+//! separate from the struct definitions so the config *shape* isn't buried
+//! under formatting code. Must stay in sync with the redactions in
+//! [`super::DynamicConfig`]'s `Debug` impl — anything secret here needs the
+//! same `[FILTERED]` treatment.
+
+use std::fmt;
+
+use super::Config;
+
+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, "  OpenTelemetry")?;
+        writeln!(formatter, "    Enabled:   {}", self.opentelemetry.enabled)?;
+        writeln!(
+            formatter,
+            "    Service:   {}",
+            self.opentelemetry.service_name
+        )?;
+        writeln!(
+            formatter,
+            "    Logs:      {}",
+            display_optional_url(&self.opentelemetry.logs_endpoint)
+        )?;
+        writeln!(
+            formatter,
+            "    Metrics:   {}",
+            display_optional_url(&self.opentelemetry.metrics_endpoint)
+        )?;
+        writeln!(
+            formatter,
+            "    Traces:    {}",
+            display_optional_url(&self.opentelemetry.traces_endpoint)
+        )?;
+        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
+        )
+    }
+}
+
+fn display_optional_url(url: &Option<String>) -> &str {
+    url.as_deref()
+        .filter(|value| !value.trim().is_empty())
+        .unwrap_or("<disabled>")
+}
+
+#[cfg(test)]
+mod tests {
+    use super::Config;
+
+    #[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 {"));
+    }
+}

src/configuration/config.rs src/configuration/config/mod.rssrc/configuration/config.rs renamed to src/configuration/config/mod.rs

@@ -1,4 +1,12 @@
-use std::{fmt, path::Path};
+//! Aquila's top-level configuration: loaded from `aquila.{yaml,toml,json,...}`
+//! (via the `config` crate), overridable by environment variables, and
+//! falling back to the [`Default`] impls below when no file is present.
+//!
+//! See [`display`] for how this renders in the startup log line.
+
+mod display;
+
+use std::path::Path;
 
 use code0_flow::flow_telemetry::OpenTelemetry;
 use config::{Config as ConfigLoader, ConfigError, File};
@@ -181,87 +189,6 @@ impl Config {
     }
 }
 
-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, "  OpenTelemetry")?;
-        writeln!(formatter, "    Enabled:   {}", self.opentelemetry.enabled)?;
-        writeln!(
-            formatter,
-            "    Service:   {}",
-            self.opentelemetry.service_name
-        )?;
-        writeln!(
-            formatter,
-            "    Logs:      {}",
-            display_optional_url(&self.opentelemetry.logs_endpoint)
-        )?;
-        writeln!(
-            formatter,
-            "    Metrics:   {}",
-            display_optional_url(&self.opentelemetry.metrics_endpoint)
-        )?;
-        writeln!(
-            formatter,
-            "    Traces:    {}",
-            display_optional_url(&self.opentelemetry.traces_endpoint)
-        )?;
-        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
-        )
-    }
-}
-
-fn display_optional_url(url: &Option<String>) -> &str {
-    url.as_deref()
-        .filter(|value| !value.trim().is_empty())
-        .unwrap_or("<disabled>")
-}
-
 #[cfg(test)]
 mod tests {
     use std::sync::Mutex;
@@ -304,22 +231,6 @@ mod tests {
         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 {"));
-    }
-
     #[test]
     fn opentelemetry_endpoints_are_enabled_by_presence() {
         let config: OpenTelemetry = ConfigLoader::builder()

src/configuration/env.rs

@@ -1,7 +1,12 @@
+//! The deployment environment Aquila believes it's running in, used to gate
+//! environment-specific behavior (like the dev-only flow JSON export in
+//! [`crate::sagittarius::flow_service_client_impl`]).
+
 use std::fmt;
 
 use serde::{Deserialize, Serialize};
 
+/// Which deployment environment Aquila is running in.
 #[derive(Clone, Copy, Debug, Default, Deserialize, Serialize, PartialEq, Eq)]
 #[serde(rename_all = "lowercase")]
 pub enum Environment {

src/configuration/mod.rs

@@ -1,3 +1,8 @@
+//! Everything Aquila loads before it can start serving: the process-wide
+//! [`config::Config`] (file + env), the static [`service::ServiceConfiguration`]
+//! allowlist, and small shared value types ([`env::Environment`], [`mode::Mode`],
+//! [`state::AppReadiness`]) used across both.
+
 pub mod config;
 pub mod env;
 pub mod mode;

src/configuration/mode.rs

@@ -1,3 +1,8 @@
+//! Aquila's two run modes: [`Mode::Static`] serves a fixed local flow export
+//! with no Sagittarius dependency, [`Mode::Dynamic`] syncs flows and module
+//! state live over gRPC. See [`crate::startup::static_mode`] and
+//! [`crate::startup::dynamic_mode`] for what each mode actually wires up.
+
 use std::fmt;
 
 use serde::{Deserialize, Serialize};

src/configuration/service/dto.rs

@@ -0,0 +1,95 @@
+//! Wire format for the service configuration file (`AQUILA_SERVICE_CONFIG_PATH`)
+//! and its conversion into the domain types in the parent module.
+//!
+//! The file format is intentionally a bit friendlier than the domain model:
+//! actions are declared with a plain `identifier` and per-project `configs`,
+//! then expanded here into the [`tucana::shared::ModuleConfigurations`]
+//! shape the rest of Aquila works with. Runtimes need no such expansion, so
+//! [`RuntimeServiceConfiguration`] doubles as both the wire format and the
+//! domain type.
+
+use serde::{Deserialize, Serialize};
+use tucana::shared::{ModuleConfigurations, helper::value::from_json_value};
+
+use super::{ActionServiceConfiguration, ServiceConfiguration};
+
+#[derive(Serialize, Deserialize, Clone)]
+pub(super) struct SerializableModuleConfiguration {
+    pub(super) identifier: String,
+    pub(super) value: serde_json::Value,
+}
+
+#[derive(Serialize, Deserialize, Clone)]
+pub(super) struct SerializableModuleProjectConfiguration {
+    pub(super) project_id: i64,
+    #[serde(default)]
+    pub(super) configs: Vec<SerializableModuleConfiguration>,
+}
+
+#[derive(Serialize, Deserialize, Clone)]
+pub(super) struct SerializableActionServiceConfiguration {
+    pub(super) token: String,
+    pub(super) identifier: String,
+    #[serde(default)]
+    pub(super) configs: Vec<SerializableModuleProjectConfiguration>,
+}
+
+#[derive(Serialize, Deserialize, Clone, Default)]
+pub(super) struct SerializableServiceConfiguration {
+    #[serde(default)]
+    pub(super) actions: Vec<SerializableActionServiceConfiguration>,
+    #[serde(default)]
+    pub(super) runtimes: Vec<RuntimeServiceConfiguration>,
+}
+
+/// A pre-provisioned runtime's token and identity. Used directly as both the
+/// file format and the domain type since, unlike actions, no expansion is needed.
+#[derive(Serialize, Deserialize, Clone)]
+pub struct RuntimeServiceConfiguration {
+    pub(super) token: String,
+    pub(super) identifier: String,
+    #[serde(default)]
+    pub(super) resolved_modules: Vec<String>,
+}
+
+impl From<SerializableModuleConfiguration> for tucana::shared::ModuleConfiguration {
+    fn from(value: SerializableModuleConfiguration) -> Self {
+        Self {
+            identifier: value.identifier,
+            value: Some(from_json_value(value.value)),
+        }
+    }
+}
+
+impl From<SerializableModuleProjectConfiguration> for tucana::shared::ModuleProjectConfigurations {
+    fn from(value: SerializableModuleProjectConfiguration) -> Self {
+        Self {
+            project_id: value.project_id,
+            module_configurations: value.configs.into_iter().map(Into::into).collect(),
+        }
+    }
+}
+
+impl From<SerializableActionServiceConfiguration> for ActionServiceConfiguration {
+    fn from(value: SerializableActionServiceConfiguration) -> Self {
+        let module_identifier = value.identifier.clone();
+
+        Self {
+            token: value.token,
+            service_name: value.identifier,
+            config: vec![ModuleConfigurations {
+                module_identifier,
+                module_configurations: value.configs.into_iter().map(Into::into).collect(),
+            }],
+        }
+    }
+}
+
+impl From<SerializableServiceConfiguration> for ServiceConfiguration {
+    fn from(value: SerializableServiceConfiguration) -> Self {
+        Self {
+            actions: value.actions.into_iter().map(Into::into).collect(),
+            runtimes: value.runtimes.into_iter().collect(),
+        }
+    }
+}