use crate::{Mecha10Error, Result};
use anyhow::Context as _;
use notify::{Config, Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher as NotifyWatcher};
use serde::de::DeserializeOwned;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::RwLock;
use tokio::time::sleep;
use tracing::{debug, error, info, warn};
pub struct ConfigWatcher<T>
where
T: DeserializeOwned + Clone + Send + Sync + 'static,
{
config: Arc<RwLock<T>>,
path: PathBuf,
_watcher: RecommendedWatcher,
}
impl<T> ConfigWatcher<T>
where
T: DeserializeOwned + Clone + Send + Sync + 'static,
{
pub fn new<P, F>(path: P, on_change: F) -> Result<Self>
where
P: AsRef<Path>,
F: Fn(&T) + Send + Sync + 'static,
{
let path = path.as_ref().to_path_buf();
let initial_config = Self::load_config(&path)?;
let config = Arc::new(RwLock::new(initial_config));
let config_clone = Arc::clone(&config);
let path_clone = path.clone();
let on_change = Arc::new(on_change);
let (tx, mut rx) = tokio::sync::mpsc::channel(100);
let mut watcher = RecommendedWatcher::new(
move |res: notify::Result<Event>| {
if let Ok(event) = res {
if matches!(event.kind, EventKind::Modify(_) | EventKind::Create(_)) {
let _ = tx.blocking_send(event);
}
}
},
Config::default().with_poll_interval(Duration::from_millis(500)),
)
.map_err(|e| Mecha10Error::Configuration(format!("Failed to create file watcher: {}", e)))?;
watcher
.watch(&path, RecursiveMode::NonRecursive)
.map_err(|e| Mecha10Error::Configuration(format!("Failed to watch file {}: {}", path.display(), e)))?;
info!("Started watching config file: {}", path.display());
tokio::spawn(async move {
let mut last_reload = tokio::time::Instant::now();
while let Some(_event) = rx.recv().await {
if last_reload.elapsed() < Duration::from_secs(1) {
continue;
}
debug!("Config file changed, debouncing...");
sleep(Duration::from_millis(100)).await;
match Self::load_config(&path_clone) {
Ok(new_config) => {
info!("Successfully reloaded config from: {}", path_clone.display());
{
let mut config = config_clone.write().await;
*config = new_config.clone();
}
on_change(&new_config);
last_reload = tokio::time::Instant::now();
}
Err(e) => {
error!(
"Failed to reload config from {}: {}. Keeping old config.",
path_clone.display(),
e
);
warn!("Fix the configuration errors to apply changes.");
}
}
}
});
Ok(Self {
config,
path,
_watcher: watcher,
})
}
pub fn get(&self) -> T {
let config = self.config.blocking_read();
config.clone()
}
pub async fn get_async(&self) -> T {
let config = self.config.read().await;
config.clone()
}
pub async fn reload(&self) -> Result<()> {
let new_config = Self::load_config(&self.path)?;
let mut config = self.config.write().await;
*config = new_config;
info!("Manually reloaded config from: {}", self.path.display());
Ok(())
}
pub fn path(&self) -> &Path {
&self.path
}
fn load_config(path: &Path) -> Result<T> {
use crate::config::load_config;
load_config(path)
.with_context(|| format!("Failed to load configuration from {}", path.display()))
.map_err(|e| Mecha10Error::Configuration(format!("{:#}", e)))
}
}
pub struct ValidatedConfigWatcher<T>
where
T: DeserializeOwned + Clone + Send + Sync + crate::config::Validate + 'static,
{
inner: ConfigWatcher<T>,
}
impl<T> ValidatedConfigWatcher<T>
where
T: DeserializeOwned + Clone + Send + Sync + crate::config::Validate + 'static,
{
pub fn new<P, F>(path: P, on_change: F) -> Result<Self>
where
P: AsRef<Path>,
F: Fn(&T) + Send + Sync + 'static,
{
let inner = ConfigWatcher::new(path, move |config: &T| {
if let Err(e) = config.validate() {
error!("Configuration validation failed: {}", e);
warn!("Keeping previous configuration.");
return;
}
on_change(config);
})?;
Ok(Self { inner })
}
pub fn get(&self) -> T {
self.inner.get()
}
pub async fn get_async(&self) -> T {
self.inner.get_async().await
}
pub async fn reload(&self) -> Result<()> {
self.inner.reload().await?;
self.inner.get().validate()?;
Ok(())
}
pub fn path(&self) -> &Path {
self.inner.path()
}
}