diode_base/
dynamic_config.rs

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

diode_base/
dynamic_config.rs
