sz-orm-config 1.2.2

//! Consul 配置源：从 Consul KV 存储读取配置，支持递归查询与 blocking query watch。

use crate::{ConfigError, ConfigSource};
use async_trait::async_trait;
use serde::Deserialize;
use serde_json::Value;
use std::collections::HashMap;
use std::time::Duration;

/// Consul 配置源
///
/// 从 Consul KV 存储读取配置：
/// - **递归查询**：`GET /v1/kv/{prefix}?recurse` 获取前缀下所有 KV
/// - **键映射**：去除配置前缀和前导 `/`，剩余部分作为键名
/// - **Base64 解码**：Consul KV 的 Value 字段为 Base64 编码，自动解码
/// - **类型推断**：解码后的值自动推断为布尔/整数/浮点数/JSON/字符串
/// - **watch**：基于 Consul blocking query（`index` + `wait` 长轮询），变更时回调
pub struct ConsulConfigSource {
    /// Consul Agent URL，如 `http://localhost:8500`
    url: String,
    /// ACL Token（可选，未启用 ACL 时为 None）
    token: Option<String>,
    /// 配置键前缀，如 `myapp/`
    prefix: String,
}

/// Consul KV 响应条目（仅反序列化需要的字段）
#[derive(Deserialize)]
struct ConsulKV {
    #[serde(rename = "Key")]
    key: String,
    #[serde(rename = "Value")]
    value: Option<String>,
}

impl ConsulConfigSource {
    /// 创建 Consul 配置源
    ///
    /// - `url`: Consul Agent URL，如 `http://localhost:8500`
    /// - `token`: ACL Token，未启用 ACL 时传 `None`
    /// - `prefix`: 配置键前缀，仅加载此前缀下的 KV
    pub fn new(
        url: impl Into<String>,
        token: Option<String>,
        prefix: impl Into<String>,
    ) -> Self {
        Self {
            url: url.into(),
            token,
            prefix: prefix.into(),
        }
    }

    /// 构建 Consul KV 递归查询 URL
    ///
    /// - `index`: 传入 Some(idx) 时添加 blocking query 参数（`&index={idx}&wait=10s`）
    fn build_url(&self, index: Option<u64>) -> String {
        let base = self.url.trim_end_matches('/');
        let mut url = format!("{}/v1/kv/{}?recurse", base, self.prefix);
        if let Some(idx) = index {
            url.push_str(&format!("&index={}&wait=10s", idx));
        }
        url
    }

    /// 执行一次 KV 查询，返回配置 map 和 Consul Index
    ///
    /// - `index`: 传入 Some(idx) 时使用 blocking query（长轮询等待变更）
    /// - 返回值包含配置 map 和响应头中的 `X-Consul-Index`
    async fn fetch(
        &self,
        client: &reqwest::Client,
        index: Option<u64>,
    ) -> Result<(HashMap<String, Value>, u64), ConfigError> {
        let url = self.build_url(index);
        let mut req = client.get(&url);
        if let Some(token) = &self.token {
            req = req.header("X-Consul-Token", token);
        }
        let resp = req
            .send()
            .await
            .map_err(|e| ConfigError::Remote(format!("Consul 请求失败: {}", e)))?;

        // 从响应头提取 X-Consul-Index（blocking query 用于检测变更）
        let consul_index = resp
            .headers()
            .get("X-Consul-Index")
            .and_then(|v| v.to_str().ok())
            .and_then(|s| s.parse::<u64>().ok())
            .unwrap_or(0);

        // 404 表示前缀下无 KV，返回空 map
        if resp.status() == reqwest::StatusCode::NOT_FOUND {
            return Ok((HashMap::new(), consul_index));
        }
        if !resp.status().is_success() {
            return Err(ConfigError::Remote(format!(
                "Consul 返回错误状态: {}",
                resp.status()
            )));
        }

        let kvs: Vec<ConsulKV> = resp
            .json()
            .await
            .map_err(|e| ConfigError::Remote(format!("Consul 响应解析失败: {}", e)))?;

        let mut map = HashMap::new();
        for kv in kvs {
            // 去掉配置前缀和前导 '/'
            let key = kv
                .key
                .strip_prefix(&self.prefix)
                .unwrap_or(&kv.key)
                .trim_start_matches('/')
                .to_string();
            // Value 字段为 Base64 编码，解码后进行类型推断
            if let Some(val_b64) = kv.value {
                match crate::base64_decode(&val_b64) {
                    Ok(bytes) => {
                        let val_str = String::from_utf8_lossy(&bytes).to_string();
                        map.insert(key, crate::parse_config_value(&val_str));
                    }
                    Err(e) => {
                        tracing::warn!("Consul 值 Base64 解码失败 (key={}): {}", key, e);
                    }
                }
            }
        }
        Ok((map, consul_index))
    }
}

#[async_trait]
impl ConfigSource for ConsulConfigSource {
    /// 从 Consul KV 加载全量配置（递归查询）
    async fn load(&self) -> Result<HashMap<String, Value>, ConfigError> {
        let client = reqwest::Client::new();
        let (map, _) = self.fetch(&client, None).await?;
        Ok(map)
    }

    /// 基于 Consul blocking query 监听配置变更
    ///
    /// 实现细节：
    /// - 在独立线程中创建 tokio 运行时，避免占用调用方的运行时
    /// - 使用 `index` + `wait=10s` 参数进行长轮询，Consul 在等待期间有变更时立即返回
    /// - 每次返回后更新 `last_index` 并触发回调，然后发起新一轮长轮询
    /// - 请求失败时等待 1 秒后重试，避免密集请求
    async fn watch(
        &self,
        callback: Box<dyn Fn(HashMap<String, Value>) + Send + Sync>,
    ) -> Result<(), ConfigError> {
        let url = self.url.clone();
        let token = self.token.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!("Consul watch: 创建运行时失败: {}", e);
                    return;
                }
            };
            rt.block_on(async move {
                let src = ConsulConfigSource { url, token, prefix };
                let client = reqwest::Client::new();
                let mut last_index: u64 = 0;
                loop {
                    match src.fetch(&client, Some(last_index)).await {
                        Ok((map, idx)) => {
                            // 仅当 index 增大时才触发回调（避免重复通知）
                            if idx > last_index {
                                last_index = idx;
                                callback(map);
                            }
                        }
                        Err(e) => {
                            tracing::warn!("Consul watch 请求失败: {}", e);
                            // 错误后等待 1 秒再重试，避免密集请求
                            tokio::time::sleep(Duration::from_secs(1)).await;
                        }
                    }
                }
            });
        });
        Ok(())
    }
}

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

    #[tokio::test]
    #[ignore = "需要真实 Consul 服务，通过 CONSUL_URL 环境变量指定地址"]
    async fn test_consul_load_real() {
        let url =
            std::env::var("CONSUL_URL").unwrap_or_else(|_| "http://localhost:8500".to_string());
        let src = ConsulConfigSource::new(url, None, "sz-orm-test/");
        let data = src.load().await;
        // 仅验证不 panic，具体值取决于 Consul 中的数据
        println!("Consul load result: {:?}", data);
    }

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