diode_base/
dynamic_config_file.rs

1use std::collections::BTreeMap;
2use std::path::PathBuf;
3
4use diode::{Service, StdError};
5use notify::{RecommendedWatcher, RecursiveMode, Watcher};
6use serde::{Deserialize, Serialize};
7use tokio::sync::mpsc;
8use tokio_util::sync::CancellationToken;
9
10use crate::{Config, DynamicConfigService};
11
12use super::DynamicConfigUpdater;
13
14#[derive(Debug, Clone, Serialize, Deserialize)]
15pub struct DynamicConfigFileConfig {
16    pub path: PathBuf,
17}
18
19impl crate::ConfigSection for DynamicConfigFileConfig {
20    fn key() -> &'static str {
21        "dynamic_config_file"
22    }
23}
24
25/// File-based dynamic configuration provider
26#[derive(Service)]
27pub struct DynamicConfigFile {
28    #[inject(Config)]
29    config: DynamicConfigFileConfig,
30}
31
32impl DynamicConfigService for DynamicConfigFile {
33    async fn get_snapshot(&self) -> Result<BTreeMap<String, serde_json::Value>, StdError> {
34        let path = &self.config.path;
35        tracing::debug!(path = ?path, "Reading config file");
36        let content = tokio::fs::read_to_string(&path).await.map_err(|e| {
37            tracing::warn!(path = ?path, error = %e, "Failed to read config file");
38            e
39        })?;
40        let config: BTreeMap<String, serde_json::Value> =
41            serde_json::from_str(&content).map_err(|e| {
42                tracing::warn!(path = ?path, error = %e, "Failed to parse config file");
43                e
44            })?;
45        tracing::debug!(path = ?path, keys = config.len(), "Successfully loaded config file");
46        Ok(config)
47    }
48
49    async fn watch_changes(
50        &self,
51        updater: DynamicConfigUpdater,
52        shutdown: CancellationToken,
53    ) -> Result<(), StdError> {
54        let path = &self.config.path;
55        tracing::info!(path = ?path, "Starting file watcher for dynamic config");
56        let (tx, mut rx) = mpsc::channel(1);
57        let mut watcher = RecommendedWatcher::new(
58            move |res: Result<notify::Event, notify::Error>| {
59                if let Err(e) = tx.try_send(res) {
60                    tracing::warn!(error = %e, "Failed to send file watch event");
61                }
62            },
63            notify::Config::default(),
64        )
65        .map_err(|e| {
66            tracing::error!(error = %e, "Failed to create file watcher");
67            e
68        })?;
69        watcher
70            .watch(path, RecursiveMode::NonRecursive)
71            .map_err(|e| {
72                tracing::error!(path = ?path, error = %e, "Failed to start watching file");
73                e
74            })?;
75        match self.get_snapshot().await {
76            Ok(snapshot) => {
77                tracing::info!(path = ?path, "Loaded initial config snapshot");
78                updater.set_snapshot(snapshot);
79            }
80            Err(e) => {
81                tracing::error!(path = ?path, error = %e, "Failed to load initial config snapshot");
82                return Err(e);
83            }
84        }
85        loop {
86            tokio::select! {
87                event = rx.recv() => {
88                    match event {
89                        Some(Ok(event)) => {
90                            tracing::debug!(path = ?path, event = ?event, "File watch event received");
91                            if event.kind.is_modify() {
92                                match self.get_snapshot().await {
93                                    Ok(snapshot) => {
94                                        tracing::info!(path = ?path, "Config file updated, reloading");
95                                        updater.set_snapshot(snapshot);
96                                    }
97                                    Err(e) => {
98                                        tracing::error!(path = ?path, error = %e, "Failed to reload config after file change");
99                                    }
100                                }
101                            }
102                        }
103                        Some(Err(e)) => {
104                            tracing::warn!(path = ?path, error = %e, "File watch error");
105                        }
106                        None => {
107                            tracing::debug!("File watch channel closed");
108                            break;
109                        }
110                    }
111                }
112                _ = shutdown.cancelled() => {
113                    tracing::debug!(path = ?path, "File watcher shutting down");
114                    break;
115                }
116            }
117        }
118        Ok(())
119    }
120}
diode_base/dynamic_config_file.rs

diode_base/
dynamic_config_file.rs
