fix(secret): openbao implementation wrong fs write call on windows

Reviewed-on: #251

.gitea/workflows/check.yml

@@ -15,4 +15,4 @@ jobs:
         uses: actions/checkout@v4
 
       - name: Run check script
-        run: bash check.sh
+        run: bash build/check.sh

check.sh

@@ -1 +0,0 @@
-build/check.sh

harmony-k8s/src/resources.rs

@@ -2,13 +2,14 @@ use std::collections::HashMap;
 
 use k8s_openapi::api::{
     apps::v1::Deployment,
-    core::v1::{Node, ServiceAccount},
+    core::v1::{Namespace, Node, ServiceAccount},
 };
 use k8s_openapi::apiextensions_apiserver::pkg::apis::apiextensions::v1::CustomResourceDefinition;
 use kube::api::ApiResource;
 use kube::{
     Error, Resource,
     api::{Api, DynamicObject, GroupVersionKind, ListParams, ObjectList},
+    core::ErrorResponse,
     runtime::conditions,
     runtime::wait::await_condition,
 };
@@ -313,4 +314,65 @@ impl K8sClient {
     ) -> Result<ObjectList<Node>, Error> {
         self.list_resources(None, list_params).await
     }
+
+    pub async fn namespace_exists(&self, name: &str) -> Result<bool, Error> {
+        let api: Api<Namespace> = Api::all(self.client.clone());
+        match api.get_opt(name).await? {
+            Some(_) => Ok(true),
+            None => Ok(false),
+        }
+    }
+
+    pub async fn create_namespace(&self, name: &str) -> Result<Namespace, Error> {
+        let namespace = Namespace {
+            metadata: k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta {
+                name: Some(name.to_string()),
+                ..Default::default()
+            },
+            ..Default::default()
+        };
+        let api: Api<Namespace> = Api::all(self.client.clone());
+        api.create(&kube::api::PostParams::default(), &namespace)
+            .await
+    }
+
+    pub async fn wait_for_namespace(
+        &self,
+        name: &str,
+        timeout: Option<Duration>,
+    ) -> Result<(), Error> {
+        let api: Api<Namespace> = Api::all(self.client.clone());
+        let timeout = timeout.unwrap_or(Duration::from_secs(60));
+        let start = std::time::Instant::now();
+
+        loop {
+            if start.elapsed() > timeout {
+                return Err(Error::Api(ErrorResponse {
+                    status: "Timeout".to_string(),
+                    message: format!("Namespace '{}' not ready within timeout", name),
+                    reason: "Timeout".to_string(),
+                    code: 408,
+                }));
+            }
+
+            match api.get_opt(name).await? {
+                Some(ns) => {
+                    if let Some(status) = ns.status {
+                        if status.phase == Some("Active".to_string()) {
+                            return Ok(());
+                        }
+                    }
+                }
+                None => {
+                    return Err(Error::Api(ErrorResponse {
+                        status: "NotFound".to_string(),
+                        message: format!("Namespace '{}' not found", name),
+                        reason: "NotFound".to_string(),
+                        code: 404,
+                    }));
+                }
+            }
+            tokio::time::sleep(Duration::from_millis(500)).await;
+        }
+    }
 }

harmony/src/infra/opnsense/load_balancer.rs

@@ -267,10 +267,16 @@ pub(crate) fn harmony_load_balancer_service_to_haproxy_xml(
                     SSL::Default => "".into(),
                     SSL::Other(other) => other.as_str().into(),
                 };
+                let path_without_query = path.split_once('?').map_or(path.as_str(), |(p, _)| p);
+                let (port, port_name) = match port {
+                    Some(port) => (Some(port.to_string()), port.to_string()),
+                    None => (None, "serverport".to_string()),
+                };
+
                 let haproxy_check = HAProxyHealthCheck {
-                    name: format!("HTTP_{http_method}_{path}"),
+                    name: format!("HTTP_{http_method}_{path_without_query}_{port_name}"),
                     uuid: Uuid::new_v4().to_string(),
-                    http_method: http_method.to_string().into(),
+                    http_method: http_method.to_string().to_lowercase().into(),
                     health_check_type: "http".to_string(),
                     http_uri: path.clone().into(),
                     interval: "2s".to_string(),
@@ -314,7 +320,10 @@ pub(crate) fn harmony_load_balancer_service_to_haproxy_xml(
     let mut backend = HAProxyBackend {
         uuid: Uuid::new_v4().to_string(),
         enabled: 1,
-        name: format!("backend_{}", service.listening_port),
+        name: format!(
+            "backend_{}",
+            service.listening_port.to_string().replace(':', "_")
+        ),
         algorithm: "roundrobin".to_string(),
         random_draws: Some(2),
         stickiness_expire: "30m".to_string(),
@@ -346,10 +355,22 @@ pub(crate) fn harmony_load_balancer_service_to_haproxy_xml(
     let frontend = Frontend {
         uuid: uuid::Uuid::new_v4().to_string(),
         enabled: 1,
-        name: format!("frontend_{}", service.listening_port),
+        name: format!(
+            "frontend_{}",
+            service.listening_port.to_string().replace(':', "_")
+        ),
         bind: service.listening_port.to_string(),
         mode: "tcp".to_string(), // TODO do not depend on health check here
         default_backend: Some(backend.uuid.clone()),
+        stickiness_expire: "30m".to_string().into(),
+        stickiness_size: "50k".to_string().into(),
+        stickiness_conn_rate_period: "10s".to_string().into(),
+        stickiness_sess_rate_period: "10s".to_string().into(),
+        stickiness_http_req_rate_period: "10s".to_string().into(),
+        stickiness_http_err_rate_period: "10s".to_string().into(),
+        stickiness_bytes_in_rate_period: "1m".to_string().into(),
+        stickiness_bytes_out_rate_period: "1m".to_string().into(),
+        ssl_hsts_max_age: 15768000,
         ..Default::default()
     };
     info!("HAPRoxy frontend and backend mode currently hardcoded to tcp");

harmony/src/modules/postgresql/operator.rs

@@ -1,11 +1,15 @@
 use k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta;
 use serde::Serialize;
+use std::str::FromStr;
+
+use non_blank_string_rs::NonBlankString;
 
 use crate::interpret::Interpret;
+use crate::modules::helm::chart::HelmChartScore;
 use crate::modules::k8s::apps::crd::{Subscription, SubscriptionSpec};
 use crate::modules::k8s::resource::K8sResourceScore;
 use crate::score::Score;
-use crate::topology::{K8sclient, Topology};
+use crate::topology::{HelmCommand, K8sclient, Topology};
 
 /// Install the CloudNativePg (CNPG) Operator via an OperatorHub `Subscription`.
 ///
@@ -100,3 +104,41 @@ impl<T: Topology + K8sclient> Score<T> for CloudNativePgOperatorScore {
         format!("CloudNativePgOperatorScore({})", self.namespace)
     }
 }
+
+#[derive(Debug, Clone, Serialize)]
+pub struct CloudNativePgOperatorHelmScore {
+    pub namespace: String,
+}
+
+impl Default for CloudNativePgOperatorHelmScore {
+    fn default() -> Self {
+        Self {
+            namespace: "cnpg-system".to_string(),
+        }
+    }
+}
+
+impl<T: Topology + K8sclient + HelmCommand + 'static> Score<T> for CloudNativePgOperatorHelmScore {
+    fn name(&self) -> String {
+        format!("CloudNativePgOperatorHelmScore({})", self.namespace)
+    }
+
+    fn create_interpret(&self) -> Box<dyn Interpret<T>> {
+        let cnpg_helm_score = HelmChartScore {
+            namespace: Some(NonBlankString::from_str(&self.namespace).unwrap()),
+            release_name: NonBlankString::from_str("cloudnative-pg").unwrap(),
+            chart_name: NonBlankString::from_str(
+                "oci://ghcr.io/cloudnative-pg/charts/cloudnative-pg",
+            )
+            .unwrap(),
+            chart_version: None,
+            values_overrides: None,
+            values_yaml: None,
+            create_namespace: true,
+            install_only: true,
+            repository: None,
+        };
+
+        cnpg_helm_score.create_interpret()
+    }
+}

harmony/src/modules/postgresql/score_k8s.rs

@@ -1,24 +1,35 @@
 use crate::data::Version;
 use crate::interpret::{Interpret, InterpretError, InterpretName, InterpretStatus, Outcome};
 use crate::inventory::Inventory;
-
 use crate::modules::k8s::resource::K8sResourceScore;
 use crate::modules::postgresql::capability::PostgreSQLConfig;
 use crate::modules::postgresql::cnpg::{
     Bootstrap, Cluster, ClusterSpec, ExternalCluster, Initdb, PgBaseBackup, ReplicaSpec,
     SecretKeySelector, Storage,
 };
+use crate::modules::postgresql::operator::{
+    CloudNativePgOperatorHelmScore, CloudNativePgOperatorScore,
+};
 use crate::score::Score;
-use crate::topology::{K8sclient, Topology};
+use crate::topology::{HelmCommand, K8sclient, Topology};
 use async_trait::async_trait;
+use harmony_k8s::KubernetesDistribution;
 use harmony_types::id::Id;
 use k8s_openapi::ByteString;
-use k8s_openapi::api::core::v1::Secret;
+use k8s_openapi::api::core::v1::{Pod, Secret};
 use k8s_openapi::apimachinery::pkg::apis::meta::v1::ObjectMeta;
+use log::{info, warn};
 use serde::Serialize;
 
 /// Deploys an opinionated, highly available PostgreSQL cluster managed by CNPG.
 ///
+/// This score automatically ensures the CloudNativePG (CNPG) operator is installed
+/// before creating the Cluster CRD. The installation method depends on the Kubernetes
+/// distribution:
+///
+/// - **OpenShift/OKD**: Uses OperatorHub Subscription via `CloudNativePgOperatorScore`
+/// - **K3s/Other**: Uses Helm chart via `CloudNativePgOperatorHelmScore`
+///
 /// # Usage
 /// ```
 /// use harmony::modules::postgresql::PostgreSQLScore;
@@ -26,12 +37,7 @@ use serde::Serialize;
 /// ```
 ///
 /// # Limitations (Happy Path)
-/// - Requires CNPG operator installed (use CloudNativePgOperatorScore).
 /// - No backups, monitoring, extensions configured.
-///
-/// TODO : refactor this to declare a clean dependency on cnpg operator. Then cnpg operator will
-/// self-deploy either using operatorhub or helm chart depending on k8s flavor. This is cnpg
-/// specific behavior
 #[derive(Debug, Clone, Serialize)]
 pub struct K8sPostgreSQLScore {
     pub config: PostgreSQLConfig,
@@ -56,7 +62,7 @@ impl K8sPostgreSQLScore {
     }
 }
 
-impl<T: Topology + K8sclient> Score<T> for K8sPostgreSQLScore {
+impl<T: Topology + K8sclient + HelmCommand + 'static> Score<T> for K8sPostgreSQLScore {
     fn create_interpret(&self) -> Box<dyn Interpret<T>> {
         Box::new(K8sPostgreSQLInterpret {
             config: self.config.clone(),
@@ -73,13 +79,127 @@ pub struct K8sPostgreSQLInterpret {
     config: PostgreSQLConfig,
 }
 
+impl K8sPostgreSQLInterpret {
+    async fn ensure_namespace<T: Topology + K8sclient>(
+        &self,
+        topology: &T,
+    ) -> Result<(), InterpretError> {
+        let k8s_client = topology
+            .k8s_client()
+            .await
+            .map_err(|e| InterpretError::new(format!("Failed to get k8s client: {}", e)))?;
+
+        let namespace_name = &self.config.namespace;
+
+        if k8s_client
+            .namespace_exists(namespace_name)
+            .await
+            .map_err(|e| {
+                InterpretError::new(format!(
+                    "Failed to check namespace '{}': {}",
+                    namespace_name, e
+                ))
+            })?
+        {
+            info!("Namespace '{}' already exists", namespace_name);
+            return Ok(());
+        }
+
+        info!("Creating namespace '{}'", namespace_name);
+        k8s_client
+            .create_namespace(namespace_name)
+            .await
+            .map_err(|e| {
+                InterpretError::new(format!(
+                    "Failed to create namespace '{}': {}",
+                    namespace_name, e
+                ))
+            })?;
+
+        k8s_client
+            .wait_for_namespace(namespace_name, Some(std::time::Duration::from_secs(30)))
+            .await
+            .map_err(|e| {
+                InterpretError::new(format!("Namespace '{}' not ready: {}", namespace_name, e))
+            })?;
+
+        info!("Namespace '{}' is ready", namespace_name);
+        Ok(())
+    }
+
+    async fn ensure_cnpg_operator<T: Topology + K8sclient + HelmCommand + 'static>(
+        &self,
+        topology: &T,
+    ) -> Result<(), InterpretError> {
+        let k8s_client = topology
+            .k8s_client()
+            .await
+            .map_err(|e| InterpretError::new(format!("Failed to get k8s client: {}", e)))?;
+
+        let pods = k8s_client
+            .list_all_resources_with_labels::<Pod>("app.kubernetes.io/name=cloudnative-pg")
+            .await
+            .map_err(|e| {
+                InterpretError::new(format!("Failed to list CNPG operator pods: {}", e))
+            })?;
+
+        if !pods.is_empty() {
+            info!("CNPG operator is already installed");
+            return Ok(());
+        }
+
+        warn!("CNPG operator not found, installing...");
+        let distro = k8s_client.get_k8s_distribution().await.map_err(|e| {
+            InterpretError::new(format!("Failed to detect k8s distribution: {}", e))
+        })?;
+
+        match distro {
+            KubernetesDistribution::OpenshiftFamily => {
+                info!("Installing CNPG operator via OperatorHub Subscription");
+                let score = CloudNativePgOperatorScore::default_openshift();
+                score
+                    .interpret(&Inventory::empty(), topology)
+                    .await
+                    .map_err(|e| {
+                        InterpretError::new(format!("Failed to install CNPG operator: {}", e))
+                    })?;
+            }
+            KubernetesDistribution::K3sFamily | KubernetesDistribution::Default => {
+                info!("Installing CNPG operator via Helm chart");
+                let score = CloudNativePgOperatorHelmScore::default();
+                score
+                    .interpret(&Inventory::empty(), topology)
+                    .await
+                    .map_err(|e| {
+                        InterpretError::new(format!("Failed to install CNPG operator: {}", e))
+                    })?;
+            }
+        }
+
+        k8s_client
+            .wait_until_deployment_ready(
+                "cloudnative-pg",
+                Some("cnpg-system"),
+                Some(std::time::Duration::from_secs(120)),
+            )
+            .await
+            .map_err(|e| InterpretError::new(format!("CNPG operator not ready: {}", e)))?;
+
+        info!("CNPG operator is ready");
+        Ok(())
+    }
+}
+
 #[async_trait]
-impl<T: Topology + K8sclient> Interpret<T> for K8sPostgreSQLInterpret {
+impl<T: Topology + K8sclient + HelmCommand + 'static> Interpret<T> for K8sPostgreSQLInterpret {
     async fn execute(
         &self,
         inventory: &Inventory,
         topology: &T,
     ) -> Result<Outcome, InterpretError> {
+        self.ensure_cnpg_operator(topology).await?;
+        self.ensure_namespace(topology).await?;
+
         match &self.config.role {
             super::capability::PostgreSQLClusterRole::Primary => {
                 let metadata = ObjectMeta {

harmony_config/src/lib.rs

@@ -5,7 +5,7 @@ use directories::ProjectDirs;
 use interactive_parse::InteractiveParseObj;
 use log::debug;
 use schemars::JsonSchema;
-use serde::{de::DeserializeOwned, Serialize};
+use serde::{Serialize, de::DeserializeOwned};
 use std::path::PathBuf;
 use std::sync::Arc;
 use thiserror::Error;
@@ -76,12 +76,11 @@ impl ConfigManager {
     pub async fn get<T: Config>(&self) -> Result<T, ConfigError> {
         for source in &self.sources {
             if let Some(value) = source.get(T::KEY).await? {
-                let config: T = serde_json::from_value(value).map_err(|e| {
-                    ConfigError::Deserialization {
+                let config: T =
+                    serde_json::from_value(value).map_err(|e| ConfigError::Deserialization {
                         key: T::KEY.to_string(),
                         source: e,
-                    }
-                })?;
+                    })?;
                 debug!("Retrieved config for key {} from source", T::KEY);
                 return Ok(config);
             }
@@ -95,17 +94,20 @@ impl ConfigManager {
         match self.get::<T>().await {
             Ok(config) => Ok(config),
             Err(ConfigError::NotFound { .. }) => {
-                let config = T::parse_to_obj()
-                    .map_err(|e| ConfigError::PromptError(e.to_string()))?;
+                let config =
+                    T::parse_to_obj().map_err(|e| ConfigError::PromptError(e.to_string()))?;
 
                 for source in &self.sources {
                     if let Err(e) = source
-                        .set(T::KEY, &serde_json::to_value(&config).map_err(|e| {
-                            ConfigError::Serialization {
-                                key: T::KEY.to_string(),
-                                source: e,
-                            }
-                        })?)
+                        .set(
+                            T::KEY,
+                            &serde_json::to_value(&config).map_err(|e| {
+                                ConfigError::Serialization {
+                                    key: T::KEY.to_string(),
+                                    source: e,
+                                }
+                            })?,
+                        )
                         .await
                     {
                         debug!("Failed to save config to source: {e}");
@@ -175,8 +177,35 @@ mod tests {
     use super::*;
     use pretty_assertions::assert_eq;
     use serde::{Deserialize, Serialize};
+    use std::sync::Mutex;
     use std::sync::atomic::{AtomicUsize, Ordering};
 
+    static TEST_COUNTER: AtomicUsize = AtomicUsize::new(0);
+    static ENV_LOCK: Mutex<()> = Mutex::new(());
+
+    fn setup_env_vars(key: &str, value: Option<&str>) -> String {
+        let id = TEST_COUNTER.fetch_add(1, Ordering::SeqCst);
+        let env_var = format!("HARMONY_CONFIG_{}_{}", key, id);
+
+        unsafe {
+            if let Some(v) = value {
+                std::env::set_var(&env_var, v);
+            } else {
+                std::env::remove_var(&env_var);
+            }
+        }
+
+        env_var
+    }
+
+    fn run_in_isolated_env<F>(f: F)
+    where
+        F: FnOnce() + Send + 'static,
+    {
+        let handle = std::thread::spawn(f);
+        handle.join().expect("Test thread panicked");
+    }
+
     #[derive(Debug, Clone, Serialize, Deserialize, JsonSchema, PartialEq)]
     struct TestConfig {
         name: String,
@@ -339,18 +368,14 @@ mod tests {
 
     #[tokio::test]
     async fn test_env_source_reads_from_environment() {
-        unsafe {
-            std::env::set_var(
-                "HARMONY_CONFIG_TestConfig",
-                r#"{"name":"from_env","count":7}"#,
-            );
-        }
+        let _lock = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+        let env_var = setup_env_vars("TestConfig", Some(r#"{"name":"from_env","count":7}"#));
 
         let source = EnvSource;
-        let result = source.get("TestConfig").await;
+        let result = source.get(&env_var.replace("HARMONY_CONFIG_", "")).await;
 
         unsafe {
-            std::env::remove_var("HARMONY_CONFIG_TestConfig");
+            std::env::remove_var(&env_var);
         }
 
         let value = result.unwrap().unwrap();
@@ -361,26 +386,32 @@ mod tests {
 
     #[tokio::test]
     async fn test_env_source_returns_none_when_not_set() {
-        unsafe {
-            std::env::remove_var("HARMONY_CONFIG_TestConfig");
-        }
+        let _lock = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+        run_in_isolated_env(|| {
+            let env_var = setup_env_vars("TestConfig", None);
 
-        let source = EnvSource;
-        let result = source.get("TestConfig").await.unwrap();
-        assert!(result.is_none());
+            let rt = tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()
+                .unwrap();
+            rt.block_on(async {
+                let source = EnvSource;
+                let result = source.get(&env_var.replace("HARMONY_CONFIG_", "")).await;
+                assert!(result.unwrap().is_none());
+            });
+        });
     }
 
     #[tokio::test]
     async fn test_env_source_returns_error_for_invalid_json() {
-        unsafe {
-            std::env::set_var("HARMONY_CONFIG_TestConfig", "not valid json");
-        }
+        let _lock = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+        let env_var = setup_env_vars("TestConfig", Some("not valid json"));
 
         let source = EnvSource;
-        let result = source.get("TestConfig").await;
+        let result = source.get(&env_var.replace("HARMONY_CONFIG_", "")).await;
 
         unsafe {
-            std::env::remove_var("HARMONY_CONFIG_TestConfig");
+            std::env::remove_var(&env_var);
         }
 
         assert!(result.is_err());
@@ -438,4 +469,4 @@ mod tests {
 
         assert_eq!(parsed, config);
     }
-}
+}

harmony_config/src/source/env.rs

@@ -1,5 +1,5 @@
-use async_trait::async_trait;
 use crate::{ConfigError, ConfigSource};
+use async_trait::async_trait;
 
 pub struct EnvSource;
 
@@ -11,16 +11,14 @@ fn env_key_for(config_key: &str) -> String {
 impl ConfigSource for EnvSource {
     async fn get(&self, key: &str) -> Result<Option<serde_json::Value>, ConfigError> {
         let env_key = env_key_for(key);
-        
+
         match std::env::var(&env_key) {
-            Ok(value) => {
-                serde_json::from_str(&value)
-                    .map(Some)
-                    .map_err(|e| ConfigError::EnvError(format!(
-                        "Invalid JSON in environment variable {}: {}",
-                        env_key, e
-                    )))
-            }
+            Ok(value) => serde_json::from_str(&value).map(Some).map_err(|e| {
+                ConfigError::EnvError(format!(
+                    "Invalid JSON in environment variable {}: {}",
+                    env_key, e
+                ))
+            }),
             Err(std::env::VarError::NotPresent) => Ok(None),
             Err(e) => Err(ConfigError::EnvError(format!(
                 "Failed to read environment variable {}: {}",
@@ -31,12 +29,11 @@ impl ConfigSource for EnvSource {
 
     async fn set(&self, key: &str, value: &serde_json::Value) -> Result<(), ConfigError> {
         let env_key = env_key_for(key);
-        let json_string = serde_json::to_string(value)
-            .map_err(|e| ConfigError::Serialization {
-                key: key.to_string(),
-                source: e,
-            })?;
-        
+        let json_string = serde_json::to_string(value).map_err(|e| ConfigError::Serialization {
+            key: key.to_string(),
+            source: e,
+        })?;
+
         // SAFETY: Setting environment variables is generally safe in single-threaded contexts.
         // In multi-threaded contexts, this could cause races, but is acceptable for this use case
         // as config is typically set once at startup.
@@ -45,4 +42,4 @@ impl ConfigSource for EnvSource {
         }
         Ok(())
     }
-}
+}

harmony_config/src/source/local_file.rs

@@ -26,14 +26,15 @@ impl LocalFileSource {
 impl ConfigSource for LocalFileSource {
     async fn get(&self, key: &str) -> Result<Option<serde_json::Value>, ConfigError> {
         let path = self.file_path_for(key);
-        
+
         match fs::read(&path).await {
             Ok(contents) => {
-                let value: serde_json::Value = serde_json::from_slice(&contents)
-                    .map_err(|e| ConfigError::Deserialization {
+                let value: serde_json::Value = serde_json::from_slice(&contents).map_err(|e| {
+                    ConfigError::Deserialization {
                         key: key.to_string(),
                         source: e,
-                    })?;
+                    }
+                })?;
                 Ok(Some(value))
             }
             Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
@@ -46,16 +47,16 @@ impl ConfigSource for LocalFileSource {
 
     async fn set(&self, key: &str, value: &serde_json::Value) -> Result<(), ConfigError> {
         fs::create_dir_all(&self.base_path).await?;
-        
+
         let path = self.file_path_for(key);
-        let contents = serde_json::to_string_pretty(value)
-            .map_err(|e| ConfigError::Serialization {
+        let contents =
+            serde_json::to_string_pretty(value).map_err(|e| ConfigError::Serialization {
                 key: key.to_string(),
                 source: e,
             })?;
-        
+
         fs::write(&path, contents).await?;
-        
+
         Ok(())
     }
-}
+}

harmony_config/src/source/mod.rs

@@ -1,4 +1,4 @@
 pub mod env;
 pub mod local_file;
 pub mod prompt;
-pub mod store;
+pub mod store;

harmony_config/src/source/prompt.rs

@@ -18,7 +18,9 @@ impl PromptSource {
 
     #[allow(dead_code)]
     pub fn with_writer(writer: Arc<dyn std::io::Write + Send + Sync>) -> Self {
-        Self { writer: Some(writer) }
+        Self {
+            writer: Some(writer),
+        }
     }
 }
 
@@ -45,4 +47,4 @@ where
 {
     let _guard = PROMPT_MUTEX.lock().await;
     f.await
-}
+}

harmony_config/src/source/store.rs

@@ -19,8 +19,8 @@ impl<S: SecretStore + 'static> ConfigSource for StoreSource<S> {
     async fn get(&self, key: &str) -> Result<Option<serde_json::Value>, ConfigError> {
         match self.store.get_raw(&self.namespace, key).await {
             Ok(bytes) => {
-                let value: serde_json::Value = serde_json::from_slice(&bytes)
-                    .map_err(|e| ConfigError::Deserialization {
+                let value: serde_json::Value =
+                    serde_json::from_slice(&bytes).map_err(|e| ConfigError::Deserialization {
                         key: key.to_string(),
                         source: e,
                     })?;
@@ -36,10 +36,10 @@ impl<S: SecretStore + 'static> ConfigSource for StoreSource<S> {
             key: key.to_string(),
             source: e,
         })?;
-        
+
         self.store
             .set_raw(&self.namespace, key, &bytes)
             .await
             .map_err(ConfigError::StoreError)
     }
-}
+}

harmony_config_derive/src/lib.rs

@@ -1,7 +1,7 @@
 use proc_macro::TokenStream;
-use proc_macro_crate::{crate_name, FoundCrate};
+use proc_macro_crate::{FoundCrate, crate_name};
 use quote::quote;
-use syn::{parse_macro_input, DeriveInput, Ident};
+use syn::{DeriveInput, Ident, parse_macro_input};
 
 #[proc_macro_derive(Config)]
 pub fn derive_config(input: TokenStream) -> TokenStream {

harmony_secret/src/store/openbao.rs

@@ -183,7 +183,7 @@ impl OpenbaoSecretStore {
         }
         #[cfg(not(unix))]
         {
-            fs::write(path, token)?;
+            fs::write(path, serde_json::to_string(token)?.as_bytes())?;
         }
         Ok(())
     }