sz-orm-config 1.2.2

//! 文件配置源：从 JSON/TOML 配置文件读取，并使用 notify 监听文件变化实现热更新。

use crate::{ConfigError, ConfigSource};
use async_trait::async_trait;
use notify::{EventKind, RecommendedWatcher, RecursiveMode, Watcher};
use serde_json::Value;
use std::collections::HashMap;
use std::path::PathBuf;
use std::sync::Mutex;

/// 文件配置源
///
/// 支持从 JSON (`.json`) 和 TOML (`.toml`) 配置文件读取配置。
/// 嵌套对象会被递归展平为点分键，例如 `{"db":{"host":"x"}}` -> `db.host`，
/// 叶节点保留原始 JSON 类型（字符串/数字/布尔/数组/null）。
///
/// 通过 [`notify`] crate 监听文件所在目录的变化，检测到目标文件被修改时
/// 重新读取并触发回调，实现配置热更新。
pub struct FileConfigSource {
    /// 配置文件路径
    path: PathBuf,
    /// 文件监听器（持有以保持监听活跃，drop 时自动停止监听）
    watcher: Mutex<Option<RecommendedWatcher>>,
}

impl FileConfigSource {
    /// 创建文件配置源
    ///
    /// - `path`: 配置文件路径，扩展名决定解析方式（`.json` / `.toml`）
    pub fn new(path: impl Into<PathBuf>) -> Self {
        Self {
            path: path.into(),
            watcher: Mutex::new(None),
        }
    }

    /// 读取并解析配置文件
    ///
    /// 根据文件扩展名选择解析器：`.json` 使用 `serde_json`，`.toml` 使用 `toml` crate。
    /// 解析后递归展平嵌套对象为点分键 map。
    fn read_file(&self) -> Result<HashMap<String, Value>, ConfigError> {
        let content = std::fs::read_to_string(&self.path)
            .map_err(|e| ConfigError::Source(format!("读取文件失败 {}: {}", self.path.display(), e)))?;
        let ext = self
            .path
            .extension()
            .and_then(|s| s.to_str())
            .unwrap_or("");
        let value = match ext {
            "json" => serde_json::from_str::<Value>(&content)
                .map_err(|e| ConfigError::Parse(format!("解析 JSON 失败: {}", e)))?,
            "toml" => {
                let toml_value: toml::Value = toml::from_str(&content)
                    .map_err(|e| ConfigError::Parse(format!("解析 TOML 失败: {}", e)))?;
                serde_json::to_value(toml_value)
                    .map_err(|e| ConfigError::Parse(format!("TOML 转 JSON 失败: {}", e)))?
            }
            other => {
                return Err(ConfigError::Source(format!(
                    "不支持的配置文件格式: .{}（仅支持 .json / .toml）",
                    other
                )));
            }
        };
        let mut map = HashMap::new();
        flatten_json("", value, &mut map);
        Ok(map)
    }
}#[async_trait]
impl ConfigSource for FileConfigSource {
    /// 加载配置文件内容，返回点分键的配置 map
    async fn load(&self) -> Result<HashMap<String, Value>, ConfigError> {
        self.read_file()
    }

    /// 监听文件变化，文件被修改/创建/删除时重新读取并回调
    ///
    /// 实现细节：
    /// - 使用 `notify` 的推荐监听器监听文件所在目录（非递归），监听目录比监听
    ///   单文件在跨平台更可靠（部分编辑器会原子替换文件）。
    /// - 在独立线程中接收事件，过滤出目标文件且为修改/创建/删除的事件，
    ///   重新读取配置后通过回调返回最新全量配置。
    /// - `FileConfigSource` 被 drop 时 `watcher` 随之 drop，事件通道关闭，线程自动退出。
    async fn watch(
        &self,
        callback: Box<dyn Fn(HashMap<String, Value>) + Send + Sync>,
    ) -> Result<(), ConfigError> {
        use std::sync::mpsc;

        // 创建事件通道：notify 通过闭包投递事件，主线程接收
        let (tx, rx) = mpsc::channel();
        let mut watcher = notify::recommended_watcher(move |res: notify::Result<notify::Event>| {
            let _ = tx.send(res);
        })
        .map_err(|e| ConfigError::Watch(format!("创建文件监听器失败: {}", e)))?;

        // 监听文件所在目录（非递归）；若无法取得父目录则监听当前目录
        let watch_dir = self
            .path
            .parent()
            .map(|p| p.to_path_buf())
            .unwrap_or_else(|| std::path::PathBuf::from("."));
        watcher
            .watch(&watch_dir, RecursiveMode::NonRecursive)
            .map_err(|e| ConfigError::Watch(format!("注册监听失败: {}", e)))?;

        // 保存 watcher 以保持监听活跃
        if let Ok(mut guard) = self.watcher.lock() {
            *guard = Some(watcher);
        }

        // 启动接收线程：过滤目标文件事件，变更时重读并回调
        let path = self.path.clone();
        std::thread::spawn(move || {
            for res in rx {
                let Ok(event) = res else { continue };
                // 仅处理修改/创建/删除类事件
                let relevant = matches!(
                    event.kind,
                    EventKind::Create(_) | EventKind::Modify(_) | EventKind::Remove(_)
                );
                if !relevant {
                    continue;
                }
                // 仅响应目标文件本身的事件
                if !event.paths.iter().any(|p| *p == path) {
                    continue;
                }
                // 重新读取文件；读取失败时跳过本次回调（例如文件被临时清空）
                let tmp = FileConfigSource::new(path.clone());
                if let Ok(data) = tmp.read_file() {
                    callback(data);
                }
            }
        });
        Ok(())
    }
}

/// 递归展平 JSON 对象为点分键 map
///
/// - `prefix`: 当前键前缀（顶层为空字符串）
/// - `value`: 当前值
/// - `out`: 输出的点分键 map
///
/// 对象会被递归展平；其他类型（字符串/数字/布尔/数组/null）作为叶节点。
/// 若顶层本身不是对象（如顶层为数组），使用 `value` 作为占位键。
pub(crate) fn flatten_json(prefix: &str, value: Value, out: &mut HashMap<String, Value>) {
    match value {
        Value::Object(map) => {
            for (k, v) in map {
                let new_prefix = if prefix.is_empty() {
                    k
                } else {
                    format!("{}.{}", prefix, k)
                };
                flatten_json(&new_prefix, v, out);
            }
        }
        other => {
            let key = if prefix.is_empty() {
                "value".to_string()
            } else {
                prefix.to_string()
            };
            out.insert(key, other);
        }
    }
}
#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicU64, Ordering};

    /// 自增计数器，保证临时文件名唯一
    static COUNTER: AtomicU64 = AtomicU64::new(0);

    /// 生成唯一的临时文件路径
    fn tmp_path(ext: &str) -> PathBuf {
        let n = COUNTER.fetch_add(1, Ordering::SeqCst);
        std::env::temp_dir().join(format!(
            "sz-orm-config-file-{}-{}.{}",
            std::process::id(),
            n,
            ext
        ))
    }

    /// 写入文件内容
    fn write_file(path: &PathBuf, content: &str) {
        std::fs::write(path, content).unwrap();
    }

    #[tokio::test]
    async fn test_load_json_flatten() {
        let path = tmp_path("json");
        write_file(&path, r#"{"db":{"host":"localhost","port":3306},"debug":true}"#);
        let src = FileConfigSource::new(&path);
        let data = src.load().await.unwrap();
        assert_eq!(data.get("db.host"), Some(&Value::from("localhost")));
        assert_eq!(data.get("db.port"), Some(&Value::from(3306)));
        assert_eq!(data.get("debug"), Some(&Value::from(true)));
        let _ = std::fs::remove_file(&path);
    }

    #[tokio::test]
    async fn test_load_toml_flatten() {
        let path = tmp_path("toml");
        // TOML 语法：顶级键必须在任何 [table] 之前，否则会归入最近的表
        write_file(&path, "debug = true\n\n[db]\nhost = \"localhost\"\nport = 3306\n");
        let src = FileConfigSource::new(&path);
        let data = src.load().await.unwrap();
        assert_eq!(data.get("db.host"), Some(&Value::from("localhost")));
        assert_eq!(data.get("db.port"), Some(&Value::from(3306)));
        assert_eq!(data.get("debug"), Some(&Value::from(true)));
        let _ = std::fs::remove_file(&path);
    }

    #[tokio::test]
    async fn test_load_unsupported_format_errors() {
        let path = tmp_path("yaml");
        write_file(&path, "key: value\n");
        let src = FileConfigSource::new(&path);
        let err = src.load().await.unwrap_err();
        assert!(matches!(err, ConfigError::Source(_)), "got: {:?}", err);
        let _ = std::fs::remove_file(&path);
    }

    #[tokio::test]
    async fn test_load_missing_file_errors() {
        let path = tmp_path("json");
        // 故意不创建文件
        let src = FileConfigSource::new(&path);
        let err = src.load().await.unwrap_err();
        assert!(matches!(err, ConfigError::Source(_)), "got: {:?}", err);
    }

    #[tokio::test]
    async fn test_load_invalid_json_errors() {
        let path = tmp_path("json");
        write_file(&path, "{not json");
        let src = FileConfigSource::new(&path);
        let err = src.load().await.unwrap_err();
        assert!(matches!(err, ConfigError::Parse(_)), "got: {:?}", err);
        let _ = std::fs::remove_file(&path);
    }

    #[tokio::test]
    #[ignore = "依赖文件系统事件，在某些平台/编辑器下可能不稳定"]
    async fn test_watch_detects_modify() {
        let path = tmp_path("json");
        write_file(&path, r#"{"k":"v1"}"#);
        let src = FileConfigSource::new(&path);
        let received = std::sync::Arc::new(std::sync::Mutex::new(None));
        let r = received.clone();
        src.watch(Box::new(move |m| { *r.lock().unwrap() = Some(m); }))
            .await
            .unwrap();
        // 等待 watcher 就绪
        tokio::time::sleep(std::time::Duration::from_millis(500)).await;
        write_file(&path, r#"{"k":"v2"}"#);
        // 轮询等待事件到达（最多 4 秒）
        for _ in 0..20 {
            tokio::time::sleep(std::time::Duration::from_millis(200)).await;
            if received.lock().unwrap().is_some() {
                break;
            }
        }
        let guard = received.lock().unwrap();
        let data = guard.as_ref().expect("应收到变更回调");
        assert_eq!(data.get("k"), Some(&Value::from("v2")));
        let _ = std::fs::remove_file(&path);
    }
}