ecat-data-iotdb 7.0.0

// Copyright (c) 2026 erik <erik@erik.xyz> — https://erik.xyz
use async_trait::async_trait;
use ecat_circuit_breaker::{Breaker, BreakerConfig};
use ecat_data::{
    BackendKind, DataPoint, FieldValue, TsdbClient, breaker_error_to_backend_error,
    run_with_timeout,
};
use ecat_errors::{Error, ErrorCode};
use ecat_tls::TlsClientConfig;
use serde::Deserialize;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::{Semaphore, SemaphorePermit};

#[cfg(feature = "metrics")]
mod metrics;
#[cfg(feature = "metrics")]
pub use metrics::register_outbound_metrics;

#[derive(Debug, Clone, Deserialize)]
pub struct IotdbConfig {
    pub base_url: String,
    pub username: String,
    pub password: String,
    #[serde(default)]
    pub tls: Option<TlsClientConfig>,
    /// 单次调用超时秒数。`0` = 禁用；未配置 = 30 秒。
    ///
    /// 这是**外层**预算，与 reqwest 自带的总超时（`from_config` 建的 client 有
    /// 30 秒、`new` 没有）取先到者。
    #[serde(default)]
    pub query_timeout_secs: Option<u64>,
    /// 熔断配置；省略则用保守默认（失败率 0.5、窗口 30 秒、打开 10 秒）。
    ///
    /// 熔断**默认开启** —— 保守阈值下只在持续失败时打开。**当前没有总开关**：
    /// `BreakerConfig` 只有阈值字段，没有 `enabled`（不许写 `{"enabled": false}`：
    /// 那是反序列化错误，或被 `#[serde(default)]` 静默吞掉后以为关掉了）。
    /// 真要停用，只能把阈值调到不可能触发（如 `failure_ratio: 1.1`）。
    #[serde(default)]
    pub breaker: Option<BreakerConfig>,
    /// 并发上限。`0` = **不限并发**（与 `query_timeout_secs: 0` = 禁用同构）；
    /// 未配置 = 32。
    ///
    /// reqwest **只有** `pool_max_idle_per_host`（空闲保留数），没有「最大总连接数」
    /// —— 默认无上限意味着并发无背压。上限由本 crate 的信号量实现，不是 reqwest 的旋钮。
    #[serde(default)]
    pub max_concurrency: Option<usize>,
}

pub struct IotdbClient {
    client: reqwest::Client,
    base_url: String,
    username: String,
    password: String,
    query_timeout: Option<Duration>,
    /// 逐 client 一个 —— 熔断器要挂在**后端实例**上，不是进程上。
    breaker: Arc<Breaker>,
    /// `None` = 不限并发（`max_concurrency: 0`）。
    semaphore: Option<Arc<Semaphore>>,
}

impl IotdbClient {
    pub fn new(
        base_url: impl Into<String>,
        username: impl Into<String>,
        password: impl Into<String>,
    ) -> Self {
        Self {
            client: reqwest::Client::new(),
            base_url: base_url.into(),
            username: username.into(),
            password: password.into(),
            query_timeout: query_timeout(None),
            breaker: Arc::new(Breaker::new(BreakerConfig::default())),
            semaphore: Some(Arc::new(Semaphore::new(32))),
        }
    }

    pub fn from_config(cfg: IotdbConfig) -> Result<Self, Error> {
        let client = ecat_tls::build_reqwest_client(&cfg.tls)
            .map_err(|e| Error::new(ErrorCode::Internal, "iotdb", format!("TLS: {e}")))?;
        let breaker = Arc::new(Breaker::new(cfg.breaker.unwrap_or_default()));
        // 自动接线（lead 裁决 2026-10-08）：`metrics` feature 下**构造即注册**，
        // 用户代码零变化。`ecat_metrics::register_outbound_metrics` 是幂等的
        // （同一 backend 重复注册是覆盖闭包），所以多 client 不会炸。
        // 探针：注释掉下面两行 ⇒ from_config_registers_outbound_metrics 红。
        #[cfg(feature = "metrics")]
        crate::register_outbound_metrics(Arc::clone(&breaker));
        Ok(Self {
            client,
            base_url: cfg.base_url,
            username: cfg.username,
            password: cfg.password,
            query_timeout: query_timeout(cfg.query_timeout_secs),
            breaker,
            semaphore: match cfg.max_concurrency {
                // `0` = 不限并发（与 `query_timeout_secs: 0` = 禁用同构）：
                // 不建信号量。建 `Semaphore::new(0)` 会让每次调用静默无限挂起
                // —— `guarded` 的第一句就是 `permit().await`，超时层在它里面。
                Some(0) => None,
                Some(n) => Some(Arc::new(Semaphore::new(n))),
                None => Some(Arc::new(Semaphore::new(32))),
            },
        })
    }

    /// 本 client 的熔断器。`metrics` feature 注册指标时要读它的状态与打开次数。
    pub fn breaker(&self) -> Arc<Breaker> {
        Arc::clone(&self.breaker)
    }

    /// 取一个并发许可；不限并发（`max_concurrency: 0`）时返回 `None`。
    /// 信号量从不 `close()`，`AcquireError` 不可达。
    async fn permit(&self) -> Option<SemaphorePermit<'_>> {
        match &self.semaphore {
            Some(sem) => Some(sem.acquire().await.expect("semaphore is never closed")),
            None => None,
        }
    }

    /// 一次出站调用的公共外壳：**许可 → 熔断 → 超时**。
    ///
    /// 这个顺序不能改：**超时若在外层，熔断器会对卡死的后端永久失明** ——
    /// 超时触发时 `tokio::time::timeout` 会 drop 内层 future，而熔断器记失败的那句
    /// 在 `f().await` **之后**，于是每次都只留下一次 drop、窗口里什么都不记，
    /// 熔断器永远不打开（这正是本设计要防的头号场景）。详见批次 5a 计划的「出入 4」。
    ///
    /// 许可在最外：还在排队的请求**还没碰后端**，不该计入熔断失败、也不该被超时掐断。
    ///
    /// `kind` **写死**不收参数：本 crate 的两个 I/O 方法同属 `TsdbClient` 一个家族
    /// （ClickHouse 收参数是因为它有 `SqlExecutor` / `TsdbClient` 两条**不同家族**的路径
    /// 共用外壳）。多一个永不变化的入参就多一个填错的机会，
    /// 而填错只是静默少数（`ecat-data/src/timeout.rs:15-35`），没有编译期保护。
    ///
    /// **壳的边界 = 公开方法**：本 crate 的 `write` **循环里每次迭代都发 HTTP**，
    /// 所以整个循环在这一个壳内 —— 谁把这段挪到循环里面（每个点一个预算），
    /// `whole_call_budget_covers_every_request_in_write` 会红。判据：问「这一步发 HTTP 吗？」
    async fn guarded<F, T: 'static>(&self, fut: F) -> Result<T, Error>
    where
        F: std::future::Future<Output = Result<T, Error>> + Send,
    {
        let _permit = self.permit().await;
        self.breaker
            .call(|| run_with_timeout(BackendKind::Tsdb, self.query_timeout, fut))
            .await
            .map_err(|e| breaker_error_to_backend_error(e, "iotdb"))
    }
}

/// `0` 表示显式禁用超时；未配置时为 30 秒。
fn query_timeout(secs: Option<u64>) -> Option<Duration> {
    match secs {
        None => Some(Duration::from_secs(30)),
        Some(0) => None,
        Some(s) => Some(Duration::from_secs(s)),
    }
}

#[async_trait]
impl TsdbClient for IotdbClient {
    async fn write(&self, points: &[DataPoint]) -> Result<(), Error> {
        // 整个循环**一个预算**：本方法每个点发一次 POST（`rest/v2/insertTablet`），
        // 预算必须罩住整次调用 —— 包在循环里面就会变成「每个点一个预算」，
        // 一次 write 的墙钟上限随点数线性放大，熔断窗口也被拆成 N 份记账
        // （测试 whole_call_budget_covers_every_request_in_write 盯着这条）。
        self.guarded(async {
            for p in points {
                // Apache IoTDB REST v2 insertTablet body:
                // {"device": "...", "is_aligned": false, "timestamps": [...],
                //  "measurements": [...], "data_types": [...], "values": [[...]]}
                // `device` = measurement; tags are not representable in this API.
                let mut measurements = Vec::with_capacity(p.fields.len());
                let mut data_types = Vec::with_capacity(p.fields.len());
                let mut values: Vec<serde_json::Value> = Vec::with_capacity(p.fields.len());
                for (k, v) in &p.fields {
                    measurements.push(k.clone());
                    let (dt, val) = match v {
                        FieldValue::Float(f) => (
                            "DOUBLE",
                            serde_json::Value::Number(
                                serde_json::Number::from_f64(*f).unwrap_or(0.into()),
                            ),
                        ),
                        FieldValue::Int(i) => ("INT64", serde_json::Value::Number((*i).into())),
                        FieldValue::String(s) => ("TEXT", serde_json::Value::String(s.clone())),
                        FieldValue::Bool(b) => ("BOOLEAN", serde_json::Value::Bool(*b)),
                    };
                    data_types.push(dt);
                    values.push(val);
                }
                let body = serde_json::json!({
                    "device": p.measurement,
                    "is_aligned": false,
                    "timestamps": [p.timestamp.unwrap_or(0)],
                    "measurements": measurements,
                    "data_types": data_types,
                    "values": [values],
                });
                let resp = self
                    .client
                    .post(format!("{}/rest/v2/insertTablet", self.base_url))
                    .basic_auth(&self.username, Some(&self.password))
                    .header("Content-Type", "application/json")
                    .json(&body)
                    .send()
                    .await
                    .map_err(|e| {
                        Error::new(ErrorCode::Internal, "iotdb", format!("iotdb write: {e}"))
                    })?;
                if !resp.status().is_success() {
                    return Err(Error::new(
                        ErrorCode::Internal,
                        "iotdb",
                        resp.text().await.unwrap_or_default(),
                    ));
                }
                // IoTDB REST v2 may return HTTP 200 with a body `code` != 200 on
                // some failures; surface those too.
                if let Ok(v) = resp.json::<serde_json::Value>().await
                    && let Some(code) = v.get("code").and_then(|c| c.as_i64())
                    && code != 200
                {
                    return Err(Error::new(
                        ErrorCode::Internal,
                        "iotdb",
                        format!(
                            "iotdb write failed: code {code}: {}",
                            v.get("message")
                                .and_then(|m| m.as_str())
                                .unwrap_or("no message")
                        ),
                    ));
                }
            }
            Ok(())
        })
        .await
    }

    async fn query(&self, sql: &str) -> Result<serde_json::Value, Error> {
        self.guarded(async {
            let resp = self
                .client
                .post(format!("{}/rest/v2/query", self.base_url))
                .basic_auth(&self.username, Some(&self.password))
                .header("Content-Type", "text/plain; charset=utf-8")
                .body(sql.to_string())
                .send()
                .await
                .map_err(|e| {
                    Error::new(ErrorCode::Internal, "iotdb", format!("iotdb query: {e}"))
                })?;
            if !resp.status().is_success() {
                return Err(Error::new(
                    ErrorCode::Internal,
                    "iotdb",
                    resp.text().await.unwrap_or_default(),
                ));
            }
            resp.json()
                .await
                .map_err(|e| Error::new(ErrorCode::Internal, "iotdb", format!("iotdb parse: {e}")))
        })
        .await
    }

    // `delete` 走 `TsdbClient` 的 trait 默认实现（`ecat-data/src/tsdb.rs:55`），
    // **不包 `guarded`**：默认实现的「不支持」是**调用方的用法错**，不是后端故障。
    // 包了之后 8 次「不支持」就会打开熔断器，之后**正常写入/查询全被拒绝**。守测试见
    // `tests/resilience.rs::delete_default_does_not_trip_the_breaker`。
}

#[cfg(test)]
mod tests;