sz-orm-config 3.5.0

//! 真实 Consul HTTP API 客户端
//!
//! 基于 Consul KV Store API v1：https://developer.hashicorp.com/consul/api-docs/kv
//!
//! 支持：
//! - `get_config` / `set_config` / `delete_config` — KV 读写
//! - `watch` — 长轮询监听配置变更（blocking query via `X-Consul-Index`）
//! - `register_service` — 服务注册（Catalog API）
//! - ACL Token 认证（通过 `X-Consul-Token` header）

#![cfg(feature = "real-consul")]

use serde::{Deserialize, Serialize};
use std::collections::HashMap;

/// Consul 客户端错误
#[derive(Debug, thiserror::Error)]
pub enum ConsulError {
    #[error("HTTP error: {0}")]
    Http(#[from] reqwest::Error),
    #[error("Consul API error: status={status}, body={body}")]
    Api { status: u16, body: String },
    #[error("Config not found: {0}")]
    NotFound(String),
    #[error("Invalid response: {0}")]
    InvalidResponse(String),
}

/// Consul 客户端配置
#[derive(Debug, Clone)]
pub struct ConsulConfig {
    pub endpoint: String,
    pub acl_token: Option<String>,
    pub datacenter: Option<String>,
    pub timeout_secs: u64,
}

impl Default for ConsulConfig {
    fn default() -> Self {
        Self {
            endpoint: "http://127.0.0.1:8500".to_string(),
            acl_token: None,
            datacenter: None,
            timeout_secs: 10,
        }
    }
}

impl ConsulConfig {
    pub fn new(endpoint: impl Into<String>) -> Self {
        Self {
            endpoint: endpoint.into(),
            ..Default::default()
        }
    }

    pub fn with_acl_token(mut self, token: impl Into<String>) -> Self {
        self.acl_token = Some(token.into());
        self
    }

    pub fn with_datacenter(mut self, dc: impl Into<String>) -> Self {
        self.datacenter = Some(dc.into());
        self
    }
}

/// Consul KV 条目
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ConsulKvEntry {
    #[serde(rename = "Key")]
    pub key: String,
    #[serde(rename = "Value")]
    pub value: Option<String>,
    #[serde(rename = "CreateIndex")]
    pub create_index: Option<u64>,
    #[serde(rename = "ModifyIndex")]
    pub modify_index: Option<u64>,
}

/// Consul 服务注册请求
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ConsulServiceRegistration {
    #[serde(rename = "ID", skip_serializing_if = "Option::is_none")]
    pub id: Option<String>,
    #[serde(rename = "Name")]
    pub name: String,
    #[serde(rename = "Address", skip_serializing_if = "Option::is_none")]
    pub address: Option<String>,
    #[serde(rename = "Port", skip_serializing_if = "Option::is_none")]
    pub port: Option<u16>,
    #[serde(rename = "Tags", skip_serializing_if = "Option::is_none")]
    pub tags: Option<Vec<String>>,
    #[serde(rename = "Meta", skip_serializing_if = "Option::is_none")]
    pub meta: Option<HashMap<String, String>>,
}

/// 真实 Consul HTTP API 客户端
pub struct ConsulClient {
    config: ConsulConfig,
    http: reqwest::Client,
}

impl ConsulClient {
    pub fn new(config: ConsulConfig) -> Result<Self, ConsulError> {
        let http = reqwest::Client::builder()
            .timeout(std::time::Duration::from_secs(config.timeout_secs))
            .build()?;
        Ok(Self { config, http })
    }

    fn build_url(&self, path: &str) -> String {
        format!("{}{}", self.config.endpoint, path)
    }

    fn add_auth(&self, req: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
        if let Some(token) = &self.config.acl_token {
            req.header("X-Consul-Token", token)
        } else {
            req
        }
    }

    fn add_dc(&self, req: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
        if let Some(dc) = &self.config.datacenter {
            req.query(&[("dc", dc)])
        } else {
            req
        }
    }

    /// 读取 KV 配置值
    pub async fn get_config(&self, key: &str) -> Result<String, ConsulError> {
        let url = self.build_url(&format!("/v1/kv/{}", key));
        let req = self.http.get(&url);
        let req = self.add_auth(req);
        let req = self.add_dc(req);
        let resp = req.send().await?;

        if resp.status() == reqwest::StatusCode::NOT_FOUND {
            return Err(ConsulError::NotFound(key.to_string()));
        }
        if !resp.status().is_success() {
            let status = resp.status().as_u16();
            let body = resp.text().await.unwrap_or_default();
            return Err(ConsulError::Api { status, body });
        }

        let entries: Vec<ConsulKvEntry> = resp.json().await?;
        if entries.is_empty() {
            return Err(ConsulError::NotFound(key.to_string()));
        }
        let encoded = entries[0]
            .value
            .as_ref()
            .ok_or_else(|| ConsulError::InvalidResponse("missing Value field".into()))?;
        let decoded = base64_decode(encoded)?;
        Ok(decoded)
    }

    /// 写入 KV 配置值
    pub async fn set_config(&self, key: &str, value: &str) -> Result<(), ConsulError> {
        let url = self.build_url(&format!("/v1/kv/{}", key));
        let req = self.http.put(&url).body(value.to_string());
        let req = self.add_auth(req);
        let req = self.add_dc(req);
        let resp = req.send().await?;

        if !resp.status().is_success() {
            let status = resp.status().as_u16();
            let body = resp.text().await.unwrap_or_default();
            return Err(ConsulError::Api { status, body });
        }
        Ok(())
    }

    /// 删除 KV 配置
    pub async fn delete_config(&self, key: &str) -> Result<(), ConsulError> {
        let url = self.build_url(&format!("/v1/kv/{}", key));
        let req = self.http.delete(&url);
        let req = self.add_auth(req);
        let req = self.add_dc(req);
        let resp = req.send().await?;

        if !resp.status().is_success() {
            let status = resp.status().as_u16();
            let body = resp.text().await.unwrap_or_default();
            return Err(ConsulError::Api { status, body });
        }
        Ok(())
    }

    /// 列出指定前缀下所有 KV 键
    pub async fn list_keys(&self, prefix: &str) -> Result<Vec<String>, ConsulError> {
        let url = self.build_url(&format!("/v1/kv/{}?keys", prefix));
        let req = self.http.get(&url);
        let req = self.add_auth(req);
        let req = self.add_dc(req);
        let resp = req.send().await?;

        if resp.status() == reqwest::StatusCode::NOT_FOUND {
            return Ok(Vec::new());
        }
        if !resp.status().is_success() {
            let status = resp.status().as_u16();
            let body = resp.text().await.unwrap_or_default();
            return Err(ConsulError::Api { status, body });
        }
        let keys: Vec<String> = resp.json().await?;
        Ok(keys)
    }

    /// 长轮询监听配置变更
    ///
    /// 返回变更后的值和新的 index。调用方可循环调用此方法实现持续监听。
    /// `wait_secs` 为长轮询等待时间（Consul 默认最大 10 分钟）。
    pub async fn watch(
        &self,
        key: &str,
        last_index: u64,
        wait_secs: u64,
    ) -> Result<(String, u64), ConsulError> {
        let url = self.build_url(&format!("/v1/kv/{}", key));
        let req = self.http.get(&url).query(&[
            ("index", last_index.to_string()),
            ("wait", format!("{}s", wait_secs)),
        ]);
        let req = self.add_auth(req);
        let req = self.add_dc(req);
        let resp = req.send().await?;

        if !resp.status().is_success() {
            let status = resp.status().as_u16();
            let body = resp.text().await.unwrap_or_default();
            return Err(ConsulError::Api { status, body });
        }

        let new_index = resp
            .headers()
            .get("X-Consul-Index")
            .and_then(|v| v.to_str().ok())
            .and_then(|s| s.parse::<u64>().ok())
            .unwrap_or(last_index);

        let entries: Vec<ConsulKvEntry> = resp.json().await?;
        if entries.is_empty() {
            return Err(ConsulError::NotFound(key.to_string()));
        }
        let encoded = entries[0]
            .value
            .as_ref()
            .ok_or_else(|| ConsulError::InvalidResponse("missing Value field".into()))?;
        let decoded = base64_decode(encoded)?;
        Ok((decoded, new_index))
    }

    /// 注册服务到 Consul Catalog
    pub async fn register_service(
        &self,
        service: ConsulServiceRegistration,
    ) -> Result<(), ConsulError> {
        let url = self.build_url("/v1/catalog/register");
        let body = serde_json::json!({
            "Datacenter": self.config.datacenter,
            "Node": "sz-orm-auto",
            "Address": service.address.as_deref().unwrap_or("127.0.0.1"),
            "Service": {
                "ID": service.id,
                "Service": service.name,
                "Address": service.address,
                "Port": service.port,
                "Tags": service.tags,
                "Meta": service.meta,
            }
        });
        let req = self.http.put(&url).json(&body);
        let req = self.add_auth(req);
        let resp = req.send().await?;

        if !resp.status().is_success() {
            let status = resp.status().as_u16();
            let body = resp.text().await.unwrap_or_default();
            return Err(ConsulError::Api { status, body });
        }
        Ok(())
    }

    /// 注销服务
    pub async fn deregister_service(
        &self,
        service_id: &str,
        node: &str,
    ) -> Result<(), ConsulError> {
        let url = self.build_url("/v1/catalog/deregister");
        let body = serde_json::json!({
            "Datacenter": self.config.datacenter,
            "Node": node,
            "ServiceID": service_id,
        });
        let req = self.http.put(&url).json(&body);
        let req = self.add_auth(req);
        let resp = req.send().await?;

        if !resp.status().is_success() {
            let status = resp.status().as_u16();
            let body = resp.text().await.unwrap_or_default();
            return Err(ConsulError::Api { status, body });
        }
        Ok(())
    }
}

fn base64_decode(s: &str) -> Result<String, ConsulError> {
    use base64::Engine;
    let decoded = base64::engine::general_purpose::STANDARD
        .decode(s)
        .map_err(|e| ConsulError::InvalidResponse(format!("base64 decode error: {}", e)))?;
    String::from_utf8(decoded)
        .map_err(|e| ConsulError::InvalidResponse(format!("UTF-8 decode error: {}", e)))
}