sz-orm-config 1.2.2

//! etcd 配置源：从 etcd KV 存储读取配置，支持前缀查询与 watch 实时推送。

use crate::{ConfigError, ConfigSource};
use async_trait::async_trait;
use etcd_client::{Client, GetOptions, WatchOptions};
use futures::StreamExt;
use serde_json::Value;
use std::collections::HashMap;

/// etcd 配置源
///
/// 从 etcd KV 存储读取配置：
/// - **前缀查询**：使用 `with_prefix()` 获取指定前缀下所有 KV
/// - **键映射**：去除配置前缀和前导 `/`，剩余部分作为键名
/// - **类型推断**：值自动推断为布尔/整数/浮点数/JSON/字符串
/// - **watch**：基于 etcd watch API，变更时重载全量配置并回调
pub struct EtcdConfigSource {
    /// etcd 端点列表，如 `["localhost:2379"]`
    endpoints: Vec<String>,
    /// 配置键前缀，如 `myapp/`
    prefix: String,
}

impl EtcdConfigSource {
    /// 创建 etcd 配置源
    ///
    /// - `endpoints`: etcd 端点列表，如 `vec!["localhost:2379".to_string()]`
    /// - `prefix`: 配置键前缀，仅加载此前缀下的 KV
    pub fn new(endpoints: Vec<String>, prefix: impl Into<String>) -> Self {
        Self {
            endpoints,
            prefix: prefix.into(),
        }
    }

    /// 从 etcd 加载全量配置（前缀查询），供 load 与 watch 重载复用
    ///
    /// 注：etcd-client 0.14 起 `kv` 字段为私有，需通过 `Client::get` 直接调用（内部委托给 kv_client）
    async fn fetch(client: &mut Client, prefix: &str) -> Result<HashMap<String, Value>, ConfigError> {
        let resp = client
            .get(prefix, Some(GetOptions::new().with_prefix()))
            .await
            .map_err(|e| ConfigError::Remote(format!("etcd get 失败: {}", e)))?;
        let mut map = HashMap::new();
        for kv in resp.kvs() {
            let key = String::from_utf8_lossy(kv.key()).to_string();
            // 去掉配置前缀和可能的前导 '/'
            let key = key
                .strip_prefix(prefix)
                .unwrap_or(&key)
                .trim_start_matches('/')
                .to_string();
            let value_str = String::from_utf8_lossy(kv.value()).to_string();
            map.insert(key, crate::parse_config_value(&value_str));
        }
        Ok(map)
    }
}

#[async_trait]
impl ConfigSource for EtcdConfigSource {
    /// 连接 etcd 并加载前缀下的全量配置
    async fn load(&self) -> Result<HashMap<String, Value>, ConfigError> {
        let mut client = Client::connect(self.endpoints.clone(), None)
            .await
            .map_err(|e| ConfigError::Remote(format!("连接 etcd 失败: {}", e)))?;
        Self::fetch(&mut client, &self.prefix).await
    }

    /// 基于 etcd watch API 监听配置变更
    ///
    /// 实现细节：
    /// - 在独立线程中创建 tokio 运行时，避免占用调用方的运行时
    /// - 使用 `watch(prefix, with_prefix)` 注册前缀监听
    /// - 每收到一个变更事件，重新加载全量配置并触发回调
    /// - watch 流结束（如连接断开）时线程退出并记录日志
    async fn watch(
        &self,
        callback: Box<dyn Fn(HashMap<String, Value>) + Send + Sync>,
    ) -> Result<(), ConfigError> {
        let endpoints = self.endpoints.clone();
        let prefix = self.prefix.clone();
        std::thread::spawn(move || {
            // 使用 current_thread 运行时（仅需 `rt` feature），避免依赖 `rt-multi-thread`。
            // 所在的 std::thread 已是独立线程，单线程运行时即可驱动异步任务。
            let rt = match tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
            {
                Ok(rt) => rt,
                Err(e) => {
                    tracing::error!("etcd watch: 创建运行时失败: {}", e);
                    return;
                }
            };
            rt.block_on(async move {
                let mut client = match Client::connect(endpoints, None).await {
                    Ok(c) => c,
                    Err(e) => {
                        tracing::error!("etcd watch: 连接失败: {}", e);
                        return;
                    }
                };
                // 注册前缀 watch
                let (_watcher, mut stream) = match client
                    .watch(prefix.clone(), Some(WatchOptions::new().with_prefix()))
                    .await
                {
                    Ok(t) => t,
                    Err(e) => {
                        tracing::error!("etcd watch: 注册失败: {}", e);
                        return;
                    }
                };
                // 每收到一个变更事件，重新加载全量配置并回调
                while let Some(result) = stream.next().await {
                    match result {
                        Ok(_watch_response) => match Self::fetch(&mut client, &prefix).await {
                            Ok(map) => callback(map),
                            Err(e) => tracing::warn!("etcd watch: 重载配置失败: {}", e),
                        },
                        Err(e) => tracing::warn!("etcd watch: 事件错误: {}", e),
                    }
                }
                tracing::warn!("etcd watch: 流已结束");
            });
        });
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    #[ignore = "需要真实 etcd 服务，通过 ETCD_ENDPOINTS 环境变量指定端点"]
    async fn test_etcd_load_real() {
        let endpoints: Vec<String> = std::env::var("ETCD_ENDPOINTS")
            .unwrap_or_else(|_| "localhost:2379".to_string())
            .split(',')
            .map(|s| s.to_string())
            .collect();
        let src = EtcdConfigSource::new(endpoints, "sz-orm-test/");
        let data = src.load().await;
        // 仅验证不 panic，具体值取决于 etcd 中的数据
        println!("etcd load result: {:?}", data);
    }

    #[tokio::test]
    #[ignore = "需要真实 etcd 服务"]
    async fn test_etcd_load_unreachable() {
        // 连接不存在的 etcd 端口，load 应返回错误
        let src = EtcdConfigSource::new(vec!["localhost:19999".to_string()], "test/");
        let result = src.load().await;
        assert!(result.is_err(), "连接不存在的 etcd 应返回错误");
    }
}