diode_base/
dynamic_config.rs

1use tracing::Instrument;
2
3use std::collections::BTreeMap;
4use std::marker::PhantomData;
5use std::path::PathBuf;
6use std::sync::{
7    Arc, RwLock,
8    atomic::{AtomicBool, Ordering},
9};
10use std::time::Duration;
11
12use diode::{
13    AddServiceExt, App, AppBuilder, Dependencies, Plugin, Service, ServiceDependencyExt, StdError,
14};
15use serde::{Deserialize, Deserializer, Serialize, de::DeserializeOwned};
16use tokio_util::sync::CancellationToken;
17
18use crate::{AddDaemonExt, Config, ConfigSection, Daemon, defer};
19
20/// Configuration for dynamic config system
21#[derive(Debug, Default, Clone, Serialize, Deserialize)]
22pub struct DynamicConfigConfig {
23    /// Path to cache file for persistent storage
24    #[serde(default)]
25    pub cache_path: Option<PathBuf>,
26    /// How often to write cache to disk (default: 1 second)
27    #[serde(default, deserialize_with = "deserialize_duration_option")]
28    pub cache_period: Option<Duration>,
29    /// Path to fallback config file
30    #[serde(default)]
31    pub fallback_path: Option<PathBuf>,
32}
33
34impl ConfigSection for DynamicConfigConfig {
35    fn key() -> &'static str {
36        "dynamic_config"
37    }
38}
39
40/// Main dynamic configuration store
41pub struct DynamicConfig {
42    /// Fallback values loaded from file
43    fallback: BTreeMap<String, serde_json::Value>,
44    /// In-memory cache of configuration values
45    cache: RwLock<BTreeMap<String, serde_json::Value>>,
46    /// Flag indicating cache needs to be written to disk
47    cache_dirty: Arc<AtomicBool>,
48    /// Event subscribers for configuration changes
49    subscribers:
50        RwLock<BTreeMap<String, Vec<Box<dyn Fn(Option<&serde_json::Value>) + Send + Sync>>>>,
51}
52
53impl DynamicConfig {
54    /// Get configuration value by key
55    pub fn get<T: DeserializeOwned>(&self, key: &str) -> Option<T> {
56        tracing::trace!(key = key, "Getting dynamic config value");
57        let cache = self.cache.read().unwrap();
58        let value = cache.get(key).or_else(|| self.fallback.get(key));
59        value.and_then(|v| {
60            serde_json::from_value(v.clone())
61                .map_err(|e| {
62                    tracing::warn!(key = key, error = %e, "Failed to deserialize config value");
63                    e
64                })
65                .ok()
66        })
67    }
68
69    /// Subscribe to configuration changes for a specific key
70    pub fn subscribe<T, F>(&self, key: &str, callback: F)
71    where
72        T: DeserializeOwned + 'static,
73        F: Fn(Option<T>) + Send + Sync + 'static,
74    {
75        let key = key.to_string();
76        tracing::debug!(key = key, "Subscribing to dynamic config changes");
77        // Call callback immediately with current value
78        let cache = self.cache.read().unwrap();
79        let value = cache
80            .get(&key)
81            .or_else(|| self.fallback.get(&key))
82            .and_then(|v| {
83                serde_json::from_value(v.clone())
84                    .map_err(|e| {
85                        tracing::warn!(key = key, error = %e, "Failed to deserialize config value");
86                        e
87                    })
88                    .ok()
89            });
90        callback(value);
91        // Add to subscribers
92        let wrapper = Box::new(move |value: Option<&serde_json::Value>| {
93            let typed_value = value.and_then(|v| serde_json::from_value(v.clone()).ok());
94            callback(typed_value);
95        });
96        let mut subscribers = self.subscribers.write().unwrap();
97        subscribers.entry(key).or_default().push(wrapper);
98    }
99
100    /// Update configuration snapshot (internal method for providers)
101    pub fn update_snapshot(&self, snapshot: BTreeMap<String, serde_json::Value>) {
102        tracing::debug!("Updating dynamic config snapshot");
103        let mut cache = self.cache.write().unwrap();
104        let mut changed_keys = Vec::new();
105        // Update existing keys and add new ones
106        for (key, value) in &snapshot {
107            match cache.get(key) {
108                Some(v) if v == value => {
109                    // No change, skip
110                    continue;
111                }
112                _ => {
113                    cache.insert(key.clone(), value.clone());
114                    changed_keys.push(key.clone());
115                }
116            }
117        }
118        // Remove keys that are no longer in snapshot
119        let keys_to_remove: Vec<String> = cache
120            .keys()
121            .filter(|key| !snapshot.contains_key(*key))
122            .cloned()
123            .collect();
124        for key in keys_to_remove {
125            cache.remove(&key);
126            changed_keys.push(key);
127        }
128        drop(cache);
129        if !changed_keys.is_empty() {
130            self.cache_dirty.store(true, Ordering::Relaxed);
131            self.notify_subscribers(changed_keys);
132        }
133    }
134
135    /// Update single configuration key (internal method for providers)
136    fn update_key(&self, key: String, value: serde_json::Value) {
137        tracing::debug!(key = key, "Updating dynamic config key");
138        let mut cache = self.cache.write().unwrap();
139        let changed = match cache.get(&key) {
140            Some(existing) => existing != &value,
141            None => true,
142        };
143        if changed {
144            cache.insert(key.clone(), value);
145            drop(cache);
146            self.cache_dirty.store(true, Ordering::Relaxed);
147            self.notify_subscribers(vec![key]);
148        }
149    }
150
151    /// Remove configuration key (internal method for providers)
152    fn remove_key(&self, key: &str) {
153        tracing::debug!(key = key, "Removing dynamic config key");
154        let mut cache = self.cache.write().unwrap();
155        if cache.remove(key).is_some() {
156            drop(cache);
157            self.cache_dirty.store(true, Ordering::Relaxed);
158            self.notify_subscribers(vec![key.to_string()]);
159        }
160    }
161
162    /// Notify subscribers about configuration changes
163    fn notify_subscribers(&self, changed_keys: Vec<String>) {
164        let cache = self.cache.read().unwrap();
165        let subscribers = self.subscribers.read().unwrap();
166        for key in changed_keys {
167            if let Some(key_subscribers) = subscribers.get(&key) {
168                let value = cache.get(&key).or_else(|| self.fallback.get(&key));
169
170                for subscriber in key_subscribers {
171                    subscriber(value);
172                }
173            }
174        }
175    }
176
177    /// Save config cache to disk
178    async fn save_cache(&self, cache_path: &PathBuf) -> Result<(), StdError> {
179        let content = {
180            let cache = self.cache.read().unwrap();
181            serde_json::to_string_pretty(&*cache)?
182        };
183        tokio::fs::write(cache_path, content).await?;
184        tracing::debug!("Saved dynamic config cache to disk");
185        Ok(())
186    }
187}
188
189/// Custom deserializer for optional Duration that supports string format like "1s", "100ms", etc.
190fn deserialize_duration_option<'de, D>(deserializer: D) -> Result<Option<Duration>, D::Error>
191where
192    D: Deserializer<'de>,
193{
194    use serde::de::Error;
195
196    #[derive(Deserialize)]
197    #[serde(untagged)]
198    enum DurationValue {
199        String(String),
200        Number(u64),
201    }
202
203    let value: Option<DurationValue> = Option::deserialize(deserializer)?;
204
205    match value {
206        None => Ok(None),
207        Some(DurationValue::String(s)) => duration_str::parse(&s)
208            .map(Some)
209            .map_err(|e| D::Error::custom(format!("Invalid duration format '{s}': {e}"))),
210        Some(DurationValue::Number(n)) => Ok(Some(Duration::from_secs(n))),
211    }
212}
213
214/// Trait for dynamic configuration providers
215pub trait DynamicConfigService: Service<Handle = Arc<Self>> {
216    /// Get current snapshot of all configuration values
217    fn get_snapshot(
218        &self,
219    ) -> impl Future<Output = Result<BTreeMap<String, serde_json::Value>, StdError>> + Send;
220
221    /// Start watching for configuration changes
222    fn watch_changes(
223        &self,
224        updater: DynamicConfigUpdater,
225        shutdown: CancellationToken,
226    ) -> impl Future<Output = Result<(), StdError>> + Send {
227        // Default implementation: no watching, just wait for shutdown
228        let _ = updater;
229        async move {
230            shutdown.cancelled().await;
231            Ok(())
232        }
233    }
234}
235
236/// Updater interface for providers to update configuration
237pub struct DynamicConfigUpdater {
238    config: Arc<DynamicConfig>,
239}
240
241impl DynamicConfigUpdater {
242    /// Update entire configuration snapshot
243    pub fn set_snapshot(&self, snapshot: BTreeMap<String, serde_json::Value>) {
244        self.config.update_snapshot(snapshot);
245    }
246
247    /// Update single configuration key
248    pub fn update_key(&self, key: String, value: serde_json::Value) {
249        self.config.update_key(key, value);
250    }
251
252    /// Remove configuration key
253    pub fn remove_key(&self, key: &str) {
254        self.config.remove_key(key);
255    }
256}
257
258struct DynamicConfigProvider<T>(PhantomData<T>);
259
260impl<T> Plugin for DynamicConfigProvider<T>
261where
262    T: DynamicConfigService + 'static,
263{
264    /// Apply the dynamic config plugin to the app builder
265    async fn build(&self, app: &mut AppBuilder) -> Result<(), StdError> {
266        // Get plugin configuration
267        let config = app
268            .get_component_ref::<Config>()
269            .unwrap()
270            .get::<DynamicConfigConfig>("dynamic_config")
271            .unwrap_or_default();
272        // Get fallback config
273        let fallback = match &config.fallback_path {
274            Some(path) => load_dynamic_config(path).await?,
275            None => BTreeMap::new(),
276        };
277        // Get config service
278        let service = app.get_component::<T::Handle>();
279        // Get cache config
280        let cache = match &config.cache_path {
281            Some(path) => match load_dynamic_config(path).await {
282                Ok(v) => Some(v),
283                Err(e) => {
284                    tracing::warn!(error = %e, "Failed to load dynamic config cache");
285                    None
286                }
287            },
288            None => None,
289        };
290        let (cache, cache_dirty) = match cache {
291            Some(v) => (v, false),
292            None => (
293                match service.as_ref() {
294                    Some(v) => v.get_snapshot().await?,
295                    None => fallback.clone(),
296                },
297                true,
298            ),
299        };
300        // Create DynamicConfig instance synchronously
301        let dynamic_config = Arc::new(DynamicConfig {
302            fallback,
303            cache: RwLock::new(cache),
304            cache_dirty: Arc::new(AtomicBool::new(cache_dirty)),
305            subscribers: Default::default(),
306        });
307        app.add_component(dynamic_config.clone())
308            .add_daemon(DynamicConfigDaemon {
309                dynamic_config,
310                service,
311                config,
312            });
313        Ok(())
314    }
315
316    fn dependencies(&self) -> Dependencies {
317        Dependencies::new().service::<T>()
318    }
319}
320
321/// Daemon that manages dynamic configuration lifecycle
322struct DynamicConfigDaemon<T> {
323    dynamic_config: Arc<DynamicConfig>,
324    service: Option<Arc<T>>,
325    config: DynamicConfigConfig,
326}
327
328impl<T> Daemon for DynamicConfigDaemon<T>
329where
330    T: DynamicConfigService + 'static,
331{
332    async fn run(&self, _app: &App, shutdown: CancellationToken) -> Result<(), StdError> {
333        let dynamic_config = self.dynamic_config.clone();
334        let service = self.service.clone();
335        let config = self.config.clone();
336        let span = tracing::info_span!("dynamic_config_daemon");
337        tracing::info!(parent: &span, "Dynamic config daemon starting");
338        defer! {
339            tracing::info!(parent: &span, "Dynamic config daemon stopped");
340        }
341        // Initialize with provider snapshot if available
342        if let Some(service) = &service {
343            match service.get_snapshot().await {
344                Ok(snapshot) => {
345                    tracing::info!(parent: &span, "Loaded initial snapshot from provider");
346                    dynamic_config.update_snapshot(snapshot);
347                }
348                Err(e) => {
349                    tracing::warn!(parent: &span, error = %e, "Failed to get initial snapshot from provider");
350                }
351            }
352            // Start service watcher in background task
353            let updater = DynamicConfigUpdater {
354                config: dynamic_config.clone(),
355            };
356            let shutdown = shutdown.clone();
357            let service = service.clone();
358            tokio::spawn(async move {
359                let span = tracing::info_span!("dynamic_config_provider");
360                tracing::info!(parent: &span, "Dynamic config provider starting");
361                defer! {
362                    tracing::info!(parent: &span, "Dynamic config provider stopped");
363                }
364                if let Err(e) = service
365                    .watch_changes(updater, shutdown.clone())
366                    .instrument(span.clone())
367                    .await
368                {
369                    tracing::error!(parent: &span, error = %e, "Dynamic config provider failed");
370                    shutdown.cancel();
371                }
372            });
373        }
374        // Cache persistence loop
375        if let Some(cache_path) = &config.cache_path {
376            let cache_period = config
377                .cache_period
378                .unwrap_or_else(|| Duration::from_secs(10));
379            tracing::debug!(parent: &span, cache_period = ?cache_period, "Starting cache persistence loop");
380            loop {
381                tokio::select! {
382                    _ = tokio::time::sleep(cache_period) => {
383                        if dynamic_config.cache_dirty.compare_exchange(
384                            true,
385                            false,
386                            std::sync::atomic::Ordering::Relaxed,
387                            std::sync::atomic::Ordering::Relaxed
388                        ).is_ok()
389                            && let Err(e) = dynamic_config.save_cache(cache_path).await
390                        {
391                            tracing::warn!(parent: &span, error = %e, "Failed to save cache to disk");
392                            dynamic_config.cache_dirty.store(true, std::sync::atomic::Ordering::Relaxed);
393                        }
394                    }
395                    _ = shutdown.cancelled() => {
396                        if dynamic_config.cache_dirty.load(std::sync::atomic::Ordering::Relaxed) {
397                            if let Err(e) = dynamic_config.save_cache(cache_path).await {
398                                tracing::warn!(parent: &span, error = %e, "Failed to save dynamic config cache during shutdown");
399                            } else {
400                                tracing::debug!(parent: &span, "Saved dynamic config cache during shutdown");
401                            }
402                        }
403                        break;
404                    }
405                }
406            }
407        } else {
408            shutdown.cancelled().await;
409        }
410        Ok(())
411    }
412}
413
414pub trait AddDynamicConfigExt {
415    fn add_dynamic_config<T>(&mut self) -> &mut Self
416    where
417        T: DynamicConfigService + 'static;
418
419    fn has_dynamic_config<T>(&self) -> bool
420    where
421        T: DynamicConfigService + 'static;
422}
423
424impl AddDynamicConfigExt for AppBuilder {
425    fn add_dynamic_config<T>(&mut self) -> &mut Self
426    where
427        T: DynamicConfigService + 'static,
428    {
429        if !self.has_service::<T>() {
430            self.add_service::<T>();
431        }
432        self.add_plugin(DynamicConfigProvider::<T>(PhantomData));
433        self
434    }
435
436    fn has_dynamic_config<T>(&self) -> bool
437    where
438        T: DynamicConfigService + 'static,
439    {
440        self.has_plugin::<DynamicConfigProvider<T>>()
441    }
442}
443
444async fn load_dynamic_config(
445    path: &PathBuf,
446) -> Result<BTreeMap<String, serde_json::Value>, StdError> {
447    let content = tokio::fs::read_to_string(path).await?;
448    let config: BTreeMap<String, serde_json::Value> = serde_json::from_str(&content)?;
449    Ok(config)
450}
diode_base/dynamic_config.rs

diode_base/
dynamic_config.rs
