confex_core/
config.rs

1use std::collections::BTreeMap;
2
3use serde::{Serialize, de::DeserializeOwned};
4
5use crate::{
6    crypto::{Encryptor, decrypt_fields},
7    deserialize::ConfigValueDeserializer,
8    error::LoadError,
9    merge::{FieldStrategies, merge},
10    migrate::MigrationChain,
11    source::Source,
12    value::ConfigValue,
13};
14
15#[cfg(feature = "async")]
16use crate::source::AsyncSource;
17
18/// Marker trait for types that can be loaded via [`ConfigLoader`].
19///
20/// Implement manually or via `#[derive(ConfigSchema)]` (Phase 1).
21pub trait ConfigSchema: DeserializeOwned + Serialize {
22    const APP_NAME: &'static str;
23    const SCHEMA_VERSION: u32;
24
25    /// Per-field merge strategies. Default: all fields use `Replace`.
26    #[must_use]
27    fn field_strategies() -> FieldStrategies {
28        &[]
29    }
30
31    /// Field names whose string values are encrypted at rest.
32    #[must_use]
33    fn encrypted_fields() -> &'static [&'static str] {
34        &[]
35    }
36}
37
38/// One source in the loader's pipeline.
39///
40/// `watch_paths` is populated only for entries added via
41/// [`ConfigLoader::with_watched_source`]; passive sources (env, CLI) leave it
42/// empty.
43pub(crate) struct LoaderEntry {
44    pub(crate) source: std::sync::Arc<dyn Source + Send + Sync>,
45    #[cfg(feature = "hot-reload")]
46    pub(crate) watch_paths: Vec<std::path::PathBuf>,
47}
48
49/// Builds and executes a layered configuration load.
50///
51/// Sources are merged in registration order (first = lowest priority).
52/// Because sources are stored as `Arc`, the loader is cheaply cloneable and
53/// its sources can be captured in hot-reload closures without duplicating state.
54pub struct ConfigLoader<S: ConfigSchema> {
55    entries: Vec<LoaderEntry>,
56    migrations: Option<std::sync::Arc<MigrationChain>>,
57    encryptor: Option<std::sync::Arc<dyn Encryptor>>,
58    _phantom: std::marker::PhantomData<S>,
59}
60
61impl<S: ConfigSchema> Clone for ConfigLoader<S> {
62    fn clone(&self) -> Self {
63        Self {
64            entries: self
65                .entries
66                .iter()
67                .map(|e| LoaderEntry {
68                    source: std::sync::Arc::clone(&e.source),
69                    #[cfg(feature = "hot-reload")]
70                    watch_paths: e.watch_paths.clone(),
71                })
72                .collect(),
73            migrations: self.migrations.clone(),
74            encryptor: self.encryptor.clone(),
75            _phantom: std::marker::PhantomData,
76        }
77    }
78}
79
80impl<S: ConfigSchema> Default for ConfigLoader<S> {
81    fn default() -> Self {
82        Self {
83            entries: Vec::new(),
84            migrations: None,
85            encryptor: None,
86            _phantom: std::marker::PhantomData,
87        }
88    }
89}
90
91impl<S: ConfigSchema> ConfigLoader<S> {
92    /// Add a passive source (env vars, CLI args, etc.).
93    ///
94    /// Passive sources are collected once at load time. In selective hot-reload
95    /// mode ([`watch_auto`](ConfigLoader::watch_auto)) their initial cache is
96    /// reused for every re-merge; they never trigger a reload themselves. Use
97    /// [`with_watched_source`](ConfigLoader::with_watched_source) for
98    /// file-backed or redb sources that should trigger selective reloads.
99    #[must_use]
100    pub fn with_source(mut self, source: impl Source + Send + Sync + 'static) -> Self {
101        let source = source.with_app_name(S::APP_NAME);
102        self.entries.push(LoaderEntry {
103            source: std::sync::Arc::new(source),
104            #[cfg(feature = "hot-reload")]
105            watch_paths: Vec::new(),
106        });
107        self
108    }
109
110    /// Add a watchable source whose filesystem changes trigger selective reloads.
111    ///
112    /// Only meaningful when combined with
113    /// [`watch_auto`](ConfigLoader::watch_auto). Using this with
114    /// [`watch`](ConfigLoader::watch) or [`load`](ConfigLoader::load) works but
115    /// provides no benefit over [`with_source`](ConfigLoader::with_source).
116    #[cfg(feature = "hot-reload")]
117    #[must_use]
118    pub fn with_watched_source(
119        mut self,
120        source: impl crate::source::WatchableSource + 'static,
121    ) -> Self {
122        let source = source.with_app_name(S::APP_NAME);
123        let watch_paths = source.watch_paths();
124        self.entries.push(LoaderEntry {
125            source: std::sync::Arc::new(source),
126            watch_paths,
127        });
128        self
129    }
130
131    #[must_use]
132    pub fn with_migrations(mut self, migrations: MigrationChain) -> Self {
133        self.migrations = Some(std::sync::Arc::new(migrations));
134        self
135    }
136
137    #[must_use]
138    pub fn with_encryptor(mut self, encryptor: impl Encryptor + 'static) -> Self {
139        self.encryptor = Some(std::sync::Arc::new(encryptor));
140        self
141    }
142
143    /// Merge all sources, run migrations if needed, decrypt fields, and deserialize.
144    ///
145    /// # Errors
146    /// Returns [`LoadError`] on source, migration, crypto, or deserialization failure.
147    pub fn load(&self) -> Result<S, LoadError> {
148        let strategies = S::field_strategies();
149
150        let mut merged = ConfigValue::Table(BTreeMap::new());
151        for entry in &self.entries {
152            let collected = entry.source.collect()?;
153            merged = merge(merged, collected, strategies);
154        }
155
156        let current_version = match &merged {
157            ConfigValue::Table(t) => match t.get("__schema_version") {
158                Some(ConfigValue::Integer(v)) => u32::try_from(*v).unwrap_or(0),
159                _ => 0,
160            },
161            _ => 0,
162        };
163
164        if current_version < S::SCHEMA_VERSION
165            && let Some(chain) = &self.migrations
166        {
167            merged = chain.run(merged, current_version, S::SCHEMA_VERSION)?;
168        }
169
170        if let Some(enc) = &self.encryptor {
171            decrypt_fields(&mut merged, S::encrypted_fields(), enc.as_ref())?;
172        }
173
174        Ok(S::deserialize(ConfigValueDeserializer(merged))?)
175    }
176}
177
178#[cfg(feature = "hot-reload")]
179impl<S: ConfigSchema + Clone + Send + Sync + 'static> ConfigLoader<S> {
180    /// Full-pipeline hot-reload: every change to `watch_path` re-collects all
181    /// sources, merges, migrates, decrypts, and deserializes.
182    ///
183    /// Prefer [`watch_auto`](ConfigLoader::watch_auto) when using
184    /// [`with_watched_source`](ConfigLoader::with_watched_source), as it
185    /// avoids re-collecting passive sources on every event.
186    ///
187    /// # Errors
188    ///
189    /// Returns [`crate::handle::HotReloadError`] if the initial load fails or
190    /// the file watcher cannot be initialised.
191    pub fn watch(
192        self,
193        watch_path: &std::path::Path,
194        options: crate::handle::WatchOptions,
195    ) -> Result<
196        (crate::handle::ConfigHandle<S>, crate::handle::WatcherHandle),
197        crate::handle::HotReloadError,
198    > {
199        use std::sync::Arc;
200
201        let initial = self.load()?;
202        let loader = Arc::new(self);
203        let reload: Arc<dyn Fn() -> Result<S, LoadError> + Send + Sync> =
204            Arc::new(move || loader.load());
205
206        crate::handle::ConfigHandle::spawn_watcher(
207            initial,
208            watch_path,
209            &reload,
210            None,
211            options.debounce,
212        )
213    }
214
215    /// Selective hot-reload: only re-collects the source whose watch path changed.
216    ///
217    /// Sources added via [`with_watched_source`](ConfigLoader::with_watched_source)
218    /// register their filesystem paths. When a change is detected on one of those
219    /// paths, only that source is re-collected; all other sources reuse their cached
220    /// values from the last collection. The results are merged in registration order.
221    ///
222    /// Sources added via [`with_source`](ConfigLoader::with_source) (env vars, CLI
223    /// args) are passive — they keep their initial cached value forever and never
224    /// trigger a reload.
225    ///
226    /// Self-write suppression: if the merged result after re-collection equals the
227    /// previous merged result, the in-memory swap is skipped. This avoids
228    /// redundant deserialization when the current process itself triggered the
229    /// filesystem event.
230    ///
231    /// # Errors
232    ///
233    /// Returns [`crate::handle::HotReloadError`] if the initial load fails or
234    /// the file watcher cannot be initialised.
235    pub fn watch_auto(
236        self,
237        options: crate::handle::WatchOptions,
238    ) -> Result<
239        (crate::handle::ConfigHandle<S>, crate::handle::WatcherHandle),
240        crate::handle::HotReloadError,
241    > {
242        use std::sync::Arc;
243
244        use crate::handle::SelectiveEntry;
245
246        let strategies = S::field_strategies();
247        let migrations = self.migrations.clone();
248        let encryptor = self.encryptor.clone();
249
250        let mut selective_entries = Vec::with_capacity(self.entries.len());
251        let mut initial_merged = ConfigValue::Table(BTreeMap::new());
252
253        for entry in &self.entries {
254            let cache = entry.source.collect().map_err(LoadError::from)?;
255            initial_merged = merge(initial_merged, cache.clone(), strategies);
256            selective_entries.push(SelectiveEntry {
257                source: Arc::clone(&entry.source),
258                cache: std::sync::Mutex::new(cache),
259                watch_paths: entry.watch_paths.clone(),
260            });
261        }
262
263        let transform: Arc<dyn Fn(ConfigValue) -> Result<S, LoadError> + Send + Sync> =
264            Arc::new(move |mut merged: ConfigValue| {
265                let current_version = match &merged {
266                    ConfigValue::Table(t) => match t.get("__schema_version") {
267                        Some(ConfigValue::Integer(v)) => u32::try_from(*v).unwrap_or(0),
268                        _ => 0,
269                    },
270                    _ => 0,
271                };
272                if current_version < S::SCHEMA_VERSION
273                    && let Some(chain) = &migrations
274                {
275                    merged = chain.run(merged, current_version, S::SCHEMA_VERSION)?;
276                }
277                if let Some(enc) = &encryptor {
278                    decrypt_fields(&mut merged, S::encrypted_fields(), enc.as_ref())?;
279                }
280                Ok(S::deserialize(ConfigValueDeserializer(merged))?)
281            });
282
283        let initial = transform(initial_merged.clone())?;
284
285        crate::handle::ConfigHandle::spawn_selective_watcher(
286            initial,
287            initial_merged,
288            selective_entries,
289            strategies,
290            transform,
291            options.debounce,
292        )
293    }
294}
295
296/// Async variant of [`ConfigLoader`]. Sources are driven with `.await`; the
297/// executor (tokio, async-std, …) is provided by the caller.
298///
299/// Sync sources automatically implement [`AsyncSource`] via the blanket impl,
300/// so they can be mixed with async sources in the same loader.
301#[cfg(feature = "async")]
302pub struct AsyncConfigLoader<S: ConfigSchema> {
303    sources: Vec<std::sync::Arc<dyn AsyncSource + Send + Sync>>,
304    migrations: Option<std::sync::Arc<MigrationChain>>,
305    encryptor: Option<std::sync::Arc<dyn Encryptor>>,
306    _phantom: std::marker::PhantomData<S>,
307}
308
309#[cfg(feature = "async")]
310impl<S: ConfigSchema> Default for AsyncConfigLoader<S> {
311    fn default() -> Self {
312        Self {
313            sources: Vec::new(),
314            migrations: None,
315            encryptor: None,
316            _phantom: std::marker::PhantomData,
317        }
318    }
319}
320
321#[cfg(feature = "async")]
322impl<S: ConfigSchema> AsyncConfigLoader<S> {
323    #[must_use]
324    pub fn with_source(mut self, source: impl AsyncSource + 'static) -> Self {
325        self.sources.push(std::sync::Arc::new(source));
326        self
327    }
328
329    #[must_use]
330    pub fn with_migrations(mut self, migrations: MigrationChain) -> Self {
331        self.migrations = Some(std::sync::Arc::new(migrations));
332        self
333    }
334
335    #[must_use]
336    pub fn with_encryptor(mut self, encryptor: impl Encryptor + 'static) -> Self {
337        self.encryptor = Some(std::sync::Arc::new(encryptor));
338        self
339    }
340
341    /// Drive all sources concurrently, merge results, run migrations, decrypt, deserialize.
342    ///
343    /// # Errors
344    /// Returns [`LoadError`] on source, migration, crypto, or deserialization failure.
345    pub async fn load(&self) -> Result<S, LoadError> {
346        let strategies = S::field_strategies();
347
348        let mut merged = ConfigValue::Table(std::collections::BTreeMap::new());
349        for source in &self.sources {
350            let collected = source.collect().await?;
351            merged = merge(merged, collected, strategies);
352        }
353
354        let current_version = match &merged {
355            ConfigValue::Table(t) => match t.get("__schema_version") {
356                Some(ConfigValue::Integer(v)) => u32::try_from(*v).unwrap_or(0),
357                _ => 0,
358            },
359            _ => 0,
360        };
361
362        if current_version < S::SCHEMA_VERSION
363            && let Some(chain) = &self.migrations
364        {
365            merged = chain.run(merged, current_version, S::SCHEMA_VERSION)?;
366        }
367
368        if let Some(enc) = &self.encryptor {
369            decrypt_fields(&mut merged, S::encrypted_fields(), enc.as_ref())?;
370        }
371
372        Ok(S::deserialize(ConfigValueDeserializer(merged))?)
373    }
374}
confex_core/config.rs

confex_core/
config.rs
