dbnexus 0.3.1

// Copyright (c) 2026 Kirky.X
//
// Licensed under the MIT License
// See LICENSE file in the project root for full license information.

//! 连接池管理模块
//!
//! 提供数据库连接池的创建、管理和自动修正功能

#[cfg(feature = "permission")]
use crate::access::permission::RolePolicy;
use async_trait::async_trait;
#[cfg(feature = "permission")]
use oxcache::Cache;
use std::sync::Arc;
use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
use std::time::Duration;
#[cfg(feature = "metrics")]
use std::time::Instant;
use tokio::sync::{Mutex as AsyncMutex, Notify, Semaphore};
#[cfg(feature = "pool-health-check")]
use tokio::time::interval;
use tokio::time::timeout;

// ponytail: ConnectionLifecycle and dead consts removed; add back when health telemetry is wired

use super::Session;
#[cfg(feature = "permission")]
use crate::access::permission::PermissionConfig;
use crate::foundation::config::{ConfigError, DbConfig};
use crate::foundation::error::{DbError, DbResult};
#[cfg(feature = "metrics")]
use crate::observability::metrics::MetricsCollector;

// 导入 Sea-ORM 的连接 trait
use sea_orm::ConnectionTrait;

/// 数据库连接类型（SeaORM 原始类型别名，保留用于向后兼容）
pub type DatabaseConnection = sea_orm::DatabaseConnection;

/// 统一数据库连接枚举（0.3.0 新增）
///
/// 支持 SeaORM（SQLite/PostgreSQL/MySQL）和 DuckDB 两种后端连接。
/// `DbPool` 和 `Session` 通过此枚举统一管理不同后端的连接。
#[derive(Clone)]
pub enum DbConnection {
    /// SeaORM 连接（SQLite/PostgreSQL/MySQL）
    SeaOrm(DatabaseConnection),
    /// DuckDB 嵌入式连接（duckdb feature）
    #[cfg(feature = "duckdb")]
    DuckDb(crate::database::pool::DuckDbConnection),
}

impl DbConnection {
    /// 获取 SeaORM 连接引用，若为 DuckDB 则返回错误
    pub fn as_sea_orm(&self) -> DbResult<&DatabaseConnection> {
        match self {
            DbConnection::SeaOrm(conn) => Ok(conn),
            #[cfg(feature = "duckdb")]
            DbConnection::DuckDb(_) => Err(DbError::Connection(sea_orm::DbErr::Custom(
                "Operation requires SeaORM connection but got DuckDb".to_string(),
            ))),
        }
    }

    /// 获取 DuckDB 连接引用，若为 SeaORM 则返回错误
    #[cfg(feature = "duckdb")]
    pub fn as_duckdb(&self) -> DbResult<&crate::database::pool::DuckDbConnection> {
        match self {
            DbConnection::DuckDb(conn) => Ok(conn),
            DbConnection::SeaOrm(_) => Err(DbError::Connection(sea_orm::DbErr::Custom(
                "Operation requires DuckDb connection but got SeaOrm".to_string(),
            ))),
        }
    }

    /// 判断是否为 DuckDB 连接
    pub fn is_duckdb(&self) -> bool {
        #[cfg(feature = "duckdb")]
        {
            matches!(self, DbConnection::DuckDb(_))
        }
        #[cfg(not(feature = "duckdb"))]
        {
            false
        }
    }
}

impl std::fmt::Debug for DbConnection {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            DbConnection::SeaOrm(_) => write!(f, "DbConnection::SeaOrm(..)"),
            #[cfg(feature = "duckdb")]
            DbConnection::DuckDb(conn) => write!(f, "DbConnection::DuckDb({conn:?})"),
        }
    }
}

/// 连接池管理器
#[derive(Clone)]
pub struct DbPool {
    /// 内部连接池
    inner: Arc<DbPoolInner>,
}

pub(crate) struct DbPoolInner {
    /// 配置
    pub(crate) config: DbConfig,

    /// 信号量控制最大连接数（优化锁竞争）
    connection_semaphore: Arc<Semaphore>,

    /// 空闲连接队列
    idle_connections: AsyncMutex<Vec<DbConnection>>,

    /// 连接可用通知（替代忙等待）
    connection_available: Notify,

    /// 活跃连接数
    pub(super) active_count: AtomicU32,

    /// 总连接数
    pub(super) total_count: AtomicU32,

    /// 权限策略缓存（直接使用 oxcache）
    #[cfg(feature = "permission")]
    pub(crate) policy_cache: Arc<Cache<String, RolePolicy>>,

    /// 权限配置（懒加载，使用 tokio 异步锁）
    #[cfg(feature = "permission")]
    permission_config: Arc<AsyncMutex<Option<PermissionConfig>>>,

    /// 后台健康检查任务（用于优雅关闭）
    health_check_shutdown: Arc<Notify>,

    /// 管理员角色名称
    pub(super) admin_role: String,

    /// 指标收集器（可选，用于 metrics 特性）
    #[cfg(feature = "metrics")]
    pub(crate) metrics_collector: Option<Arc<MetricsCollector>>,

    /// 等待计数
    pub(super) wait_count: AtomicU32,

    /// 最大等待计数（历史峰值）
    pub(super) max_waiters: AtomicU32,

    /// 借用计数
    pub(super) borrow_count: AtomicU64,

    /// 最大活跃连接数
    pub(super) max_active: AtomicU32,
}

impl DbPool {
    /// 更新最大活跃连接数（使用 CAS 操作避免竞态条件）
    ///
    /// 此方法使用 compare_exchange 循环确保原子性地更新 max_active，
    /// 只有当新值大于当前值时才更新。
    fn update_max_active(&self, active: u32) {
        // 使用 Acquire 语义确保看到最新的值
        let mut current = self.inner.max_active.load(Ordering::Acquire);
        while active > current {
            match self
                .inner
                .max_active
                .compare_exchange(current, active, Ordering::SeqCst, Ordering::Acquire)
            {
                Ok(_) => return,
                Err(observed) => {
                    // CAS 失败，使用观察到的值重试
                    current = observed;
                }
            }
        }
    }

    /// 创建新的连接池
    ///
    /// # Arguments
    ///
    /// * `url` - 数据库连接 URL
    ///
    /// # Errors
    ///
    /// 如果 URL 格式无效或不支持，返回错误
    ///
    /// # Example
    ///
    /// ```ignore
    /// use dbnexus::DbPool;
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), Box<dyn std::error::Error>> {
    ///     let pool = DbPool::new("sqlite://example.db").await?;
    ///     Ok(())
    /// }
    /// ```
    pub async fn new(url: &str) -> DbResult<Self> {
        let config = DbConfig {
            url: url.to_string(),
            ..Default::default()
        };
        Self::with_config(config).await
    }

    /// 使用配置创建连接池
    pub async fn with_config(config: DbConfig) -> DbResult<Self> {
        // 创建连接（复用 create_connection 保持错误转换一致）
        let _connection = Self::create_connection(&config).await?;

        // 创建权限策略缓存并加载初始权限配置（含首次预加载）
        #[cfg(feature = "permission")]
        let (policy_cache, permission_config) = Self::setup_permission_cache(&config).await?;

        let pool = Self {
            inner: Arc::new(DbPoolInner {
                config: config.clone(),
                connection_semaphore: Arc::new(Semaphore::new(config.max_connections as usize)),
                idle_connections: AsyncMutex::new(Vec::new()),
                connection_available: Notify::new(),
                active_count: AtomicU32::new(0),
                total_count: AtomicU32::new(0),
                #[cfg(feature = "permission")]
                policy_cache,
                #[cfg(feature = "permission")]
                permission_config: Arc::new(AsyncMutex::new(permission_config)),
                health_check_shutdown: Arc::new(Notify::new()),
                admin_role: config.admin_role.clone(),
                #[cfg(feature = "metrics")]
                metrics_collector: None,
                wait_count: AtomicU32::new(0),
                max_waiters: AtomicU32::new(0),
                borrow_count: AtomicU64::new(0),
                max_active: AtomicU32::new(0),
            }),
        };

        // 启动后台健康检查任务
        #[cfg(feature = "pool-health-check")]
        pool.start_background_health_check();

        // 预创建最小连接数（并行创建以提高启动速度，带超时和重试）
        #[cfg(feature = "pool-warmup")]
        pool.warmup_connections().await?;

        // 注意：权限策略缓存的预加载已在 setup_permission_cache() 中完成（HIGH-004 修复）
        // 此处不再重复预加载，避免冗余 IO 和缓存覆盖。

        #[cfg(feature = "auto-migrate")]
        if config.auto_migrate {
            if let Some(ref migrations_dir) = config.migrations_dir {
                if migrations_dir.exists() {
                    let _applied = pool.run_migrations(migrations_dir).await?;
                } else {
                    // migrations directory does not exist, skip migration
                }
            }
        }

        Ok(pool)
    }

    /// 使用配置结构体创建连接池
    ///
    /// 此方法接受一个 [`DbConfig`] 结构体，用于配置连接池的所有参数。
    /// 与 [`Self::new`] 方法功能相同，但更适合从配置结构体直接初始化。
    ///
    /// # Example
    #[cfg_attr(
        feature = "sqlite",
        doc = r###"
    /// ```rust
    /// use dbnexus::DbPool;
    /// use dbnexus::config::DbConfig;
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), Box<dyn std::error::Error>> {
    ///     let config = DbConfig {
    ///         url: "sqlite::memory:".to_string(),
    ///         max_connections: 10,
    ///         min_connections: 2,
    ///         ..Default::default()
    ///     };
    ///
    ///     let pool = DbPool::try_from_config(config).await?;
    ///     Ok(())
    /// }
    /// ```
    "###
    )]
    #[cfg_attr(
        not(feature = "sqlite"),
        doc = r###"
    /// ```rust,ignore
    /// // 此文档测试需要 sqlite 特性
    /// // 在使用其他数据库时，请参考相应的文档和示例
    /// ```
    "###
    )]
    ///
    /// # Errors
    ///
    /// 如果连接失败或配置无效，返回错误
    pub async fn try_from_config(config: DbConfig) -> DbResult<Self> {
        Self::with_config(config).await
    }

    /// 使用配置引用同步创建连接池（简化版本）
    ///
    /// 此方法是同步的，不会创建数据库连接。
    /// 实际的连接池创建和连接验证在首次获取连接时进行。
    ///
    /// 注意：此方法不会初始化权限缓存功能（需要异步初始化）。
    /// 如果需要完整的异步权限缓存功能，请使用 `with_config()` 异步方法。
    ///
    /// # Example
    ///
    /// ```rust
    /// use dbnexus::DbPool;
    /// use dbnexus::config::DbConfig;
    ///
    /// let runtime = tokio::runtime::Runtime::new()?;
    /// let _guard = runtime.enter();
    ///
    /// let config = DbConfig {
    ///     url: "sqlite::memory:".to_string(),
    ///     max_connections: 10,
    ///     min_connections: 1,
    ///     idle_timeout: 300,
    ///     acquire_timeout: 5000,
    ///     ..Default::default()
    /// };
    ///
    /// let pool = DbPool::try_from(&config)?;
    /// # Ok::<_, Box<dyn std::error::Error>>(())
    /// ```
    ///
    /// # Errors
    ///
    /// 如果配置验证失败，返回错误
    #[cfg(not(feature = "permission"))]
    pub fn try_from(config: &DbConfig) -> Result<Self, ConfigError> {
        Ok(Self {
            inner: Arc::new(DbPoolInner {
                config: config.clone(),
                connection_semaphore: Arc::new(Semaphore::new(config.max_connections as usize)),
                idle_connections: AsyncMutex::new(Vec::new()),
                connection_available: Notify::new(),
                active_count: AtomicU32::new(0),
                total_count: AtomicU32::new(0),
                health_check_shutdown: Arc::new(Notify::new()),
                admin_role: config.admin_role.clone(),
                #[cfg(feature = "metrics")]
                metrics_collector: None,
                wait_count: AtomicU32::new(0),
                max_waiters: AtomicU32::new(0),
                borrow_count: AtomicU64::new(0),
                max_active: AtomicU32::new(0),
            }),
        })
    }

    /// 使用配置引用同步创建连接池（简化版本，带权限但不初始化缓存）
    ///
    /// 此方法是同步的，不会创建数据库连接。
    /// 实际的连接池创建和连接验证在首次获取连接时进行。
    ///
    /// 注意：此方法不会初始化权限缓存（需要异步初始化）。
    /// 如果需要完整的异步权限缓存功能，请使用 `with_config()` 异步方法。
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// # // 需要 config feature
    /// use dbnexus::DbPool;
    /// use dbnexus::config::DbConfig;
    ///
    /// let runtime = tokio::runtime::Runtime::new()?;
    /// let _guard = runtime.enter();
    ///
    /// let config = DbConfig {
    ///     url: "sqlite::memory:".to_string(),
    ///     max_connections: 10,
    ///     min_connections: 1,
    ///     idle_timeout: 300,
    ///     acquire_timeout: 5000,
    ///     ..Default::default()
    /// };
    ///
    /// let pool = DbPool::try_from(&config)?;
    /// # Ok::<_, Box<dyn std::error::Error>>(())
    /// ```
    ///
    /// # Errors
    ///
    /// 如果配置验证失败，返回错误
    #[cfg(feature = "permission")]
    pub fn try_from(config: &DbConfig) -> Result<Self, ConfigError> {
        // permission feature 启用时，需要异步创建 oxcache 缓存。
        // 同步 try_from 无法在 tokio runtime 内安全调用 block_on（会 panic）。
        // 显式失败，引导用户使用异步构造器 with_config()。
        let _ = config;
        Err(ConfigError::InvalidValue {
            key: "permission".to_string(),
            message: "DbPool::try_from cannot be used with `permission` feature enabled; \
             use `DbPool::with_config(config).await` instead (async constructor required for cache initialization)"
                .to_string(),
        })
    }

    /// 构造权限策略缓存并加载初始权限配置（含首次预加载）
    ///
    /// 完成两件事：
    /// 1. 用 `cache_config.policy_cache_capacity` 创建 oxcache `Cache`。
    /// 2. 调用 [`Self::load_permission_config`] 读取权限文件，并把 `roles` 写入缓存（首次预加载）。
    ///
    /// 返回构造好的缓存和加载到的权限配置（若未配置文件或加载失败则为 `None`）。
    #[cfg(feature = "permission")]
    async fn setup_permission_cache(
        config: &DbConfig,
    ) -> DbResult<(Arc<Cache<String, RolePolicy>>, Option<PermissionConfig>)> {
        // 创建权限策略缓存（使用 oxcache 后端）
        let policy_cache = Arc::new(
            Cache::builder()
                .capacity(config.cache_config.policy_cache_capacity)
                .build()
                .await
                .map_err(|_e| {
                    DbError::Connection(sea_orm::DbErr::ConnectionAcquire(sea_orm::ConnAcquireErr::Timeout))
                })?,
        );

        // 加载权限配置（如果指定了路径）- 仅在启用permission特性时
        let permission_config = Self::load_permission_config(config).await;

        // 预加载权限策略到缓存（如果存在权限配置）
        if let Some(ref perm_config) = permission_config {
            for (role_name, policy) in &perm_config.roles {
                let _ = policy_cache.set(role_name, policy).await;
            }
        }

        Ok((policy_cache, permission_config))
    }

    /// 加载权限配置文件
    ///
    /// 通过 `serde_yaml_ng` / `serde_json` 直接解析 YAML/JSON 文件，与项目配置管理策略一致
    ///
    /// # Returns
    ///
    /// - `Some(PermissionConfig)` - 成功加载的权限配置
    /// - `None` - 没有配置权限文件或加载失败，使用默认的 deny_all 策略
    #[cfg(feature = "permission")]
    async fn load_permission_config(config: &DbConfig) -> Option<PermissionConfig> {
        // 尝试从配置文件加载
        if let Some(ref path) = config.permissions_path {
            match tokio::fs::read_to_string(path).await {
                Ok(content) => match Self::parse_permission_yaml(&content, path) {
                    Ok(perm_config) => {
                        return Some(perm_config);
                    }
                    Err(_e) => {
                        return None;
                    }
                },
                Err(_e) => {
                    return None;
                }
            }
        }

        // 没有配置权限文件
        None
    }

    /// 使用 JSON 直接解析权限配置
    #[cfg(feature = "json")]
    fn parse_permission_json(content: &str, source: &str) -> Result<PermissionConfig, String> {
        serde_json::from_str(content).map_err(|e| format!("JSON parse error in '{}': {}", source, e))
    }

    /// 解析权限配置
    /// 优先使用 JSON 格式（YAML 是 JSON 的超集，serde_yaml_ng 也能解析 JSON）
    fn parse_permission_yaml(content: &str, source: &str) -> Result<PermissionConfig, String> {
        // 直接使用 JSON 解析
        #[cfg(feature = "json")]
        {
            Self::parse_permission_json(content, source)
        }
        #[cfg(not(feature = "json"))]
        {
            // 如果没有 json feature，使用 serde_yaml_ng 直接解析
            #[cfg(feature = "yaml")]
            {
                serde_yaml_ng::from_str(content).map_err(|e| format!("YAML parse error in '{}': {}", source, e))
            }
            #[cfg(not(feature = "yaml"))]
            {
                Err(format!(
                    "Cannot parse permission config from '{}': neither JSON nor YAML support available",
                    source
                ))
            }
        }
    }

    /// 获取指标收集器（如果已设置）
    #[cfg(feature = "metrics")]
    pub fn metrics(&self) -> Option<&Arc<MetricsCollector>> {
        self.inner.metrics_collector.as_ref()
    }

    /// 获取实际应用的配置
    ///
    /// 返回当前连接池使用的配置。
    ///
    /// # Returns
    ///
    /// 实际应用的配置
    pub fn get_actual_config(&self) -> &DbConfig {
        &self.inner.config
    }

    /// 从池中获取 Session（带 metrics 支持）
    ///
    /// # Arguments
    ///
    /// * `role` - 角色名称，必须在权限配置中定义
    ///
    /// # Errors
    ///
    /// 如果角色未在权限配置中定义，返回错误
    /// # 安全警告
    ///
    /// 此方法接受角色字符串参数，调用者应确保：
    /// - 已通过其他方式验证用户身份（如JWT Token、API Key等）
    /// - 角色字符串来自可信来源，而非直接的用户输入
    /// - 在生产环境中不要硬编码角色字符串
    /// - 建议结合 `dbnexus::authentication::AuthenticationManager` 使用
    ///
    /// # 示例
    ///
    /// ```rust,no_run,ignore
    /// use dbnexus::{DbPool, authentication::AuthenticationManager};
    ///
    /// // 安全用法：先验证Token，再从Token中提取角色
    /// let auth_manager = AuthenticationManager::new(&jwt_secret);
    /// let claims = auth_manager.verify_token(token)?;
    /// let session = pool.get_session(&claims.role).await?;
    ///
    /// // 不安全用法：直接使用用户输入
    /// // let session = pool.get_session(user_input_role).await?; // 不要这样做！
    /// ```
    ///
    pub async fn get_session(&self, role: &str) -> DbResult<Session> {
        // 验证角色名称
        #[cfg(feature = "permission")]
        self.validate_role_name(role).await?;

        let connection = self.acquire_connection().await?;
        let pool_ref = Arc::new(self.clone());
        let session = Session::new(connection, pool_ref, self.inner.clone(), role.to_string());

        Ok(session)
    }

    /// 验证角色名称是否在权限配置中定义
    ///
    /// 仅在权限配置文件存在且成功加载时验证角色。
    /// 如果没有配置权限文件，使用安全默认策略（仅允许 admin/system 角色）。
    ///
    /// **显性化（v0.3.0 修复）**：未配置权限文件时输出 warn 日志，明确说明
    /// 正在使用安全默认策略，提醒用户配置权限文件以启用完整角色验证。
    #[cfg(feature = "permission")]
    async fn validate_role_name(&self, role: &str) -> DbResult<()> {
        // 获取权限配置锁
        let permission_config = self.inner.permission_config.lock().await;

        // 检查权限配置是否存在（用户是否显式配置了权限文件）
        if permission_config.is_none() {
            // 没有配置权限文件时，使用安全默认策略
            // 只允许预定义的安全角色，防止未授权访问
            #[cfg(feature = "tracing")]
            tracing::warn!(
                role = %role,
                "dbnexus::pool: no permission config loaded, using safe default policy (admin/system only)"
            );
            #[cfg(not(feature = "tracing"))]
            eprintln!(
                "[warn] dbnexus::pool: no permission config loaded, using safe default policy (admin/system only) for role '{}'",
                role
            );
            let safe_roles = ["admin", "system"];
            if !safe_roles.contains(&role) {
                return Err(DbError::Permission(format!(
                    "Role '{}' is not allowed without explicit permission configuration. Allowed roles: {}",
                    role,
                    safe_roles.join(", ")
                )));
            }
            return Ok(());
        }

        // 检查角色是否存在
        if permission_config
            .as_ref()
            .is_some_and(|c| c.get_role_policy(role).is_none())
        {
            // 角色不存在
            return Err(DbError::Permission(format!(
                "Role '{}' is not defined in permission configuration",
                role
            )));
        }

        Ok(())
    }

    /// 创建单个数据库连接
    ///
    /// 使用配置中的 URL 建立新的数据库连接。
    /// 此方法不进行连接池管理，仅创建原始连接。
    ///
    /// # Arguments
    ///
    /// * `config` - 数据库配置，包含连接 URL
    ///
    /// # Returns
    ///
    /// 成功创建的数据库连接
    ///
    /// # Errors
    ///
    /// 如果连接失败，返回数据库错误
    async fn create_connection(config: &DbConfig) -> DbResult<DbConnection> {
        let db_type = config
            .database_type()
            .map_err(|e| DbError::Connection(sea_orm::DbErr::Custom(format!("Invalid database URL: {e}"))))?;

        match db_type {
            crate::foundation::config::DatabaseType::DuckDb => {
                #[cfg(feature = "duckdb")]
                {
                    let conn = crate::database::pool::DuckDbConnection::new(&config.url)?;
                    Ok(DbConnection::DuckDb(conn))
                }
                #[cfg(not(feature = "duckdb"))]
                {
                    Err(DbError::Connection(sea_orm::DbErr::Custom(
                        "DuckDB feature is not enabled".to_string(),
                    )))
                }
            }
            _ => {
                let conn = sea_orm::Database::connect(&config.url).await?;
                Ok(DbConnection::SeaOrm(conn))
            }
        }
    }

    /// 预创建最小连接数（并行创建以提高启动速度，带超时和重试）
    ///
    /// 并行启动 `min_connections` 个建连任务，每个任务带 `warmup_timeout` 超时
    /// 和 `warmup_retries` 次重试。
    ///
    /// **失败语义（v0.3.0 修复）**：
    /// - 全部失败：返回 `Err`（第一个错误），避免静默成功
    /// - 部分失败：返回 `Ok`，warn 日志记录失败数量
    /// - 全部成功：返回 `Ok`
    #[cfg(feature = "pool-warmup")]
    async fn warmup_connections(&self) -> DbResult<()> {
        let initial_connections = self.inner.config.min_connections;
        let warmup_timeout = Duration::from_secs(self.inner.config.warmup_timeout);
        let warmup_retries = self.inner.config.warmup_retries;

        let mut connection_tasks = Vec::new();

        for _ in 0..initial_connections {
            let config = self.inner.config.clone();
            connection_tasks.push(async move {
                let mut retries = 0;
                let mut last_error = None;

                while retries <= warmup_retries {
                    match timeout(warmup_timeout, Self::create_connection(&config)).await {
                        Ok(Ok(conn)) => return Ok(conn),
                        Ok(Err(e)) => {
                            last_error = Some(e);
                            retries += 1;
                            if retries <= warmup_retries {
                                tokio::time::sleep(Duration::from_millis(100)).await;
                            }
                        }
                        Err(_) => {
                            last_error = Some(DbError::Connection(sea_orm::DbErr::ConnectionAcquire(
                                sea_orm::ConnAcquireErr::Timeout,
                            )));
                            break;
                        }
                    }
                }

                Err(last_error.unwrap_or_else(|| {
                    DbError::Connection(sea_orm::DbErr::ConnectionAcquire(sea_orm::ConnAcquireErr::Timeout))
                }))
            });
        }

        // 并行执行所有连接创建任务
        let results = futures::future::join_all(connection_tasks).await;

        // 统计成功/失败（规则 12：失败必须显性化，不可静默丢弃）
        let mut success_count = 0usize;
        let mut errors: Vec<DbError> = Vec::new();

        for result in results {
            match result {
                Ok(conn) => {
                    self.inner.idle_connections.lock().await.push(conn);
                    self.inner.total_count.fetch_add(1, Ordering::SeqCst);
                    success_count += 1;
                }
                Err(e) => errors.push(e),
            }
        }

        if success_count == 0 && initial_connections > 0 {
            // 全部失败：返回第一个错误（显性化失败，避免静默成功）
            return Err(errors.into_iter().next().unwrap_or_else(|| {
                DbError::Connection(sea_orm::DbErr::ConnectionAcquire(sea_orm::ConnAcquireErr::Timeout))
            }));
        }

        if !errors.is_empty() {
            // 部分失败：warn 日志记录（显性化，但不阻断初始化）
            #[cfg(feature = "tracing")]
            tracing::warn!(
                success = success_count,
                total = initial_connections,
                errors = errors.len(),
                "dbnexus::pool::warmup: warmup_connections partially failed"
            );
            #[cfg(not(feature = "tracing"))]
            eprintln!(
                "[warn] dbnexus::pool::warmup: warmup_connections partially failed: {}/{} connections created, {} errors",
                success_count,
                initial_connections,
                errors.len()
            );
        }

        Ok(())
    }

    /// 检查连接健康状态
    ///
    /// 通过执行轻量级查询来验证数据库连接的有效性。
    /// 使用数据库特定的健康检查查询：
    /// - SQLite: `SELECT 1`
    /// - PostgreSQL: `SELECT 1`
    /// - MySQL: `SELECT 1`
    ///
    /// # Arguments
    ///
    /// * `conn` - 要检查的数据库连接
    ///
    /// # Returns
    ///
    /// 如果连接有效返回 `true`，否则返回 `false`
    pub async fn check_connection_health(&self, conn: &DbConnection) -> bool {
        match conn {
            DbConnection::SeaOrm(sea_conn) => {
                let backend = Self::get_database_backend(&self.inner.config.url);
                let result = timeout(
                    Duration::from_secs(5),
                    sea_conn.execute_raw(sea_orm::Statement::from_string(backend, "SELECT 1".to_string())),
                )
                .await;
                matches!(result, Ok(Ok(_)))
            }
            #[cfg(feature = "duckdb")]
            DbConnection::DuckDb(duck_conn) => {
                let result = timeout(Duration::from_secs(5), duck_conn.health_check()).await;
                matches!(result, Ok(Ok(_)))
            }
        }
    }

    /// 获取数据库类型
    ///
    /// 根据数据库 URL 的协议部分解析数据库类型。
    /// 支持的数据库类型包括 SQLite、PostgreSQL、MySQL 和 DuckDB。
    ///
    /// # Arguments
    ///
    /// * `url` - 数据库连接 URL
    ///
    /// # Returns
    ///
    /// 对应的 Sea-ORM 数据库后端类型
    ///
    /// # Note
    ///
    /// 如果 URL 无法识别，默认返回 SQLite 类型。
    /// DuckDB 连接不使用 SeaORM 后端，此方法仅用于 SeaORM 连接。
    fn get_database_backend(url: &str) -> sea_orm::DatabaseBackend {
        if url.starts_with("sqlite:") {
            sea_orm::DatabaseBackend::Sqlite
        } else if url.starts_with("postgres:") || url.starts_with("postgresql:") {
            sea_orm::DatabaseBackend::Postgres
        } else if url.starts_with("mysql:") {
            sea_orm::DatabaseBackend::MySql
        } else if url.starts_with("duckdb:") {
            // DuckDB 不使用 SeaORM 后端，映射到 Sqlite 以避免 panic
            sea_orm::DatabaseBackend::Sqlite
        } else {
            sea_orm::DatabaseBackend::Sqlite
        }
    }

    #[cfg(feature = "pool-health-check")]
    /// 验证空闲连接的有效性（并行版本）
    ///
    /// 遍历空闲连接池，对每个连接并发执行健康检查，
    /// 将连接分区为有效和无效两组。
    ///
    /// 使用 `futures::future::join_all()` 并行验证所有连接，
    /// 显著减少大量连接时的总等待时间。
    ///
    /// # Arguments
    ///
    /// * `idle` - 空闲连接队列的可变引用
    /// * `config` - 数据库配置
    ///
    /// # Returns
    ///
    /// 返回元组 (有效连接列表, 无效连接数量)
    async fn validate_idle_connections(idle: &mut Vec<DbConnection>, config: &DbConfig) -> (Vec<DbConnection>, usize) {
        let backend = Self::get_database_backend(&config.url);

        // 先将所有连接移出，避免在持有锁期间进行 I/O 操作
        let connections: Vec<DbConnection> = std::mem::take(idle);

        // 并行执行所有健康检查
        let check_futures: Vec<_> = connections
            .into_iter()
            .map(|conn| async {
                let is_valid = match &conn {
                    DbConnection::SeaOrm(sea_conn) => timeout(
                        Duration::from_secs(2),
                        sea_conn.execute_raw(sea_orm::Statement::from_string(backend, "SELECT 1".to_string())),
                    )
                    .await
                    .is_ok_and(|result| result.is_ok()),
                    #[cfg(feature = "duckdb")]
                    DbConnection::DuckDb(duck_conn) => timeout(Duration::from_secs(2), duck_conn.health_check())
                        .await
                        .is_ok_and(|result| result.is_ok()),
                };
                (conn, is_valid)
            })
            .collect();

        let results: Vec<(DbConnection, bool)> = futures::future::join_all(check_futures).await;

        // 分区为有效和无效连接（先计算 invalid_count，再转移所有权）
        let invalid_count = results.iter().filter(|(_, is_valid)| !*is_valid).count();

        let valid_connections: Vec<DbConnection> = results
            .into_iter()
            .filter_map(|(conn, is_valid)| if is_valid { Some(conn) } else { None })
            .collect();

        (valid_connections, invalid_count)
    }

    #[cfg(feature = "pool-health-check")]
    /// 清理无效连接
    ///
    /// 遍历空闲连接池，验证每个连接的有效性，
    /// 移除超时或断开连接的实例。
    ///
    /// # Returns
    ///
    /// 被移除的无效连接数量
    pub async fn clean_invalid_connections(&self) -> u32 {
        let mut idle = self.inner.idle_connections.lock().await;
        let config = &self.inner.config;

        // 使用辅助方法验证连接
        let (valid_connections, removed_count) = Self::validate_idle_connections(&mut idle, config).await;

        // 重建空闲连接队列
        idle.extend(valid_connections);

        // 更新总连接数
        if removed_count > 0 {
            self.inner.total_count.fetch_sub(removed_count as u32, Ordering::SeqCst);
        }

        removed_count as u32
    }

    #[cfg(feature = "pool-health-check")]
    /// 验证并重新创建无效连接
    ///
    /// 检查所有空闲连接的健康状态，自动替换无效连接。
    /// 此方法会确保池中至少保持配置的最小连接数。
    ///
    /// # Returns
    ///
    /// 被重新创建的连接数量，或错误
    pub async fn validate_and_recreate_connections(&self) -> Result<u32, sea_orm::DbErr> {
        let mut idle = self.inner.idle_connections.lock().await;
        let config = &self.inner.config;

        // 使用辅助方法验证连接
        let (valid_connections, invalid_count) = Self::validate_idle_connections(&mut idle, config).await;

        let mut recreated_count = 0;

        if invalid_count > 0 {
            // 更新总连接数
            self.inner.total_count.fetch_sub(invalid_count as u32, Ordering::SeqCst);

            // 重建空闲队列（只保留有效连接）
            idle.extend(valid_connections);

            // 重新创建连接以维持最小连接数
            let current_idle = idle.len();
            let needed = config.min_connections.saturating_sub(current_idle as u32) as usize;

            for _ in 0..needed {
                match Self::create_connection(config).await {
                    Ok(new_conn) => {
                        idle.push(new_conn);
                        self.inner.total_count.fetch_add(1, Ordering::SeqCst);
                        recreated_count += 1;
                    }
                    Err(e) => {
                        return Err(sea_orm::DbErr::Custom(format!("Failed to recreate connections: {}", e)));
                    }
                }
            }
        } else {
            // 没有无效连接，恢复有效连接到池中
            idle.extend(valid_connections);
        }

        Ok(recreated_count as u32)
    }

    /// 解析健康检查间隔配置
    ///
    /// 解析传入的间隔值（秒），并限制在 5-300 秒范围内。
    /// 超出范围的值会触发警告日志。
    ///
    /// # Arguments
    ///
    /// * `value` - 健康检查间隔配置值（由调用方从环境变量 `DB_HEALTH_CHECK_INTERVAL` 读取）
    ///
    /// # Returns
    ///
    /// 返回解析后的间隔秒数，默认为 30 秒。
    ///
    /// # Examples
    ///
    /// ```
    /// use dbnexus::DbPool;
    /// // 空字符串返回默认值 30
    /// assert_eq!(DbPool::parse_health_check_interval(""), 30);
    ///
    /// // 有效值返回该值
    /// assert_eq!(DbPool::parse_health_check_interval("60"), 60);
    ///
    /// // 超出范围的值返回限制后的值
    /// assert_eq!(DbPool::parse_health_check_interval("1000"), 300);
    /// ```
    #[cfg(feature = "pool-health-check")]
    pub fn parse_health_check_interval(value: &str) -> u64 {
        value.parse::<u64>().ok().map(|v| v.clamp(5, 300)).unwrap_or(30)
    }

    /// 启动后台连接健康检查任务
    ///
    /// 该任务会定期检查所有空闲连接的健康状态，
    /// 自动移除无效连接并重建新连接以维持最小连接数。
    ///
    /// 健康检查间隔默认为 30 秒，可通过环境变量 `DB_HEALTH_CHECK_INTERVAL` 配置（秒）。
    /// 间隔值会被限制在 5-300 秒范围内，超出范围的值会触发警告日志。
    #[cfg(feature = "pool-health-check")]
    fn start_background_health_check(&self) {
        let pool = self.clone();
        let shutdown = self.inner.health_check_shutdown.clone();

        // 从环境变量读取健康检查间隔配置并解析
        let env_value = std::env::var("DB_HEALTH_CHECK_INTERVAL").unwrap_or_default();
        let interval_secs = Self::parse_health_check_interval(&env_value);

        tokio::spawn(async move {
            let mut interval = interval(Duration::from_secs(interval_secs));

            loop {
                tokio::select! {
                    _ = interval.tick() => {
                        // 执行连接健康检查
                        let _ = pool.validate_and_recreate_connections().await;
                    }
                    _ = shutdown.notified() => {
                        break;
                    }
                }
            }
        });
    }

    /// 从池中获取连接
    ///
    /// 实现连接获取逻辑，包括：
    /// 1. 记录等待者计数（wait_count），追踪当前等待获取连接的协程数
    /// 2. 追踪历史最大等待者峰值（max_waiters）
    /// 3. 获取信号量许可（控制最大并发数），带超时保护
    /// 4. 尝试从空闲连接队列获取，队列为空则创建新连接
    ///
    /// ## 等待计数与告警
    ///
    /// - 每次进入获取流程时 `wait_count += 1`，无论最终成功或超时都 `wait_count -= 1`
    /// - `max_waiters` 记录历史最大并发等待者数量（CAS 更新）
    /// - 获取超时后根据等待时长触发分级告警：
    ///   - ≥3s：warn 级别
    ///   - ≥5s：error 级别
    ///   - ≥10s：critical 级别
    ///
    /// ## 锁竞争优化策略
    ///
    /// 使用信号量（Semaphore）替代部分锁逻辑，减少锁竞争：
    /// - 信号量在获取锁之前就控制并发数量，实现更公平的等待机制
    /// - 锁持有时间最小化：仅在操作空闲队列时持有锁
    /// - 创建新连接时不持有锁，避免阻塞其他操作
    ///
    /// ## 信号量许可管理
    ///
    /// - 获取连接时：`permit.forget()` 消耗许可（连接被借出）
    /// - 释放连接时：`add_permits(1)` 归还许可（连接归还池中）
    ///
    /// # Returns
    ///
    /// 成功获取的数据库连接
    ///
    /// # Errors
    ///
    /// 如果获取连接超时（池耗尽）或创建连接失败，返回错误
    async fn acquire_connection(&self) -> DbResult<DbConnection> {
        // 步骤 0: 记录等待计数
        let waiters = self.inner.wait_count.fetch_add(1, Ordering::SeqCst) + 1;
        self.update_max_waiters(waiters);

        // 步骤 1: 获取信号量许可（等待可用槽位，带超时）
        // 信号量提供公平的等待队列，避免惊群效应
        let timeout_duration = self.inner.config.acquire_timeout_duration();

        // 仅在启用 metrics 时需要记录开始时间
        #[cfg(feature = "metrics")]
        let start = Instant::now();

        let acquire_result = timeout(timeout_duration, self.inner.connection_semaphore.acquire()).await;

        // wait_count 递减（无论成功或失败）
        self.inner.wait_count.fetch_sub(1, Ordering::SeqCst);

        let permit = match acquire_result {
            Ok(Ok(p)) => {
                // 记录获取延迟和慢获取
                #[cfg(feature = "metrics")]
                if let Some(ref collector) = self.inner.metrics_collector {
                    collector.record_connection_acquire_duration(start.elapsed());
                }
                p
            }
            Ok(Err(_)) => {
                // permit error - shouldn't happen with tokio semaphore
                return Err(DbError::Connection(sea_orm::DbErr::ConnectionAcquire(
                    sea_orm::ConnAcquireErr::Timeout,
                )));
            }
            Err(_) => {
                // Timeout - 记录超时指标
                #[cfg(feature = "metrics")]
                {
                    let elapsed_ms = start.elapsed().as_millis() as u64;
                    if let Some(ref collector) = self.inner.metrics_collector {
                        collector.record_connection_timeout_level(elapsed_ms);
                    }
                }

                return Err(DbError::Connection(sea_orm::DbErr::ConnectionAcquire(
                    sea_orm::ConnAcquireErr::Timeout,
                )));
            }
        };

        // 步骤 2: 尝试从空闲队列获取（最小化锁持有时间）
        // 使用独立作用域确保锁尽快释放
        {
            let mut idle = self.inner.idle_connections.lock().await;
            if let Some(conn) = idle.pop() {
                // 更新统计计数
                let active = self.inner.active_count.fetch_add(1, Ordering::SeqCst) + 1;
                self.update_max_active(active);
                self.inner.borrow_count.fetch_add(1, Ordering::SeqCst);
                // 忘记许可，因为连接被借出（release_connection 会归还）
                permit.forget();
                return Ok(conn);
            }
            // 锁在此处释放
        }

        // 步骤 3: 创建新连接（不持有锁，避免阻塞其他操作）
        // 先更新计数，再创建连接
        self.inner.total_count.fetch_add(1, Ordering::SeqCst);
        let active = self.inner.active_count.fetch_add(1, Ordering::SeqCst) + 1;
        self.update_max_active(active);

        match Self::create_connection(&self.inner.config).await {
            Ok(conn) => {
                self.inner.borrow_count.fetch_add(1, Ordering::SeqCst);
                // 忘记许可，因为连接被借出
                permit.forget();
                Ok(conn)
            }
            Err(e) => {
                // 创建失败，回滚计数并释放许可
                self.inner.total_count.fetch_sub(1, Ordering::SeqCst);
                self.inner.active_count.fetch_sub(1, Ordering::SeqCst);
                // 释放许可（drop 会自动释放）
                drop(permit);
                Err(e)
            }
        }
    }

    /// 更新最大等待者计数（使用 CAS 避免竞态条件）
    fn update_max_waiters(&self, current_waiters: u32) {
        let mut current = self.inner.max_waiters.load(Ordering::Acquire);
        while current_waiters > current {
            match self
                .inner
                .max_waiters
                .compare_exchange(current, current_waiters, Ordering::SeqCst, Ordering::Acquire)
            {
                Ok(_) => return,
                Err(observed) => {
                    current = observed;
                }
            }
        }
    }

    /// 归还连接到池中
    ///
    /// 将使用完毕的连接归还到空闲连接队列。
    /// 如果空闲队列已满（达到最大连接数），则丢弃该连接。
    /// 归还后会通知一个等待的请求者有新连接可用。
    ///
    /// # Arguments
    ///
    /// * `conn` - 要归还的数据库连接
    ///
    /// # Note
    ///
    /// 此方法会归还信号量许可，确保连接池可以继续接受新的连接请求。
    /// 使用 tokio::spawn 在后台执行异步操作，避免阻塞调用者。
    pub(crate) fn release_connection(&self, conn: DbConnection) {
        self.inner.active_count.fetch_sub(1, Ordering::SeqCst);
        let inner = self.inner.clone();

        // 尝试快速路径：非阻塞获取锁
        if let Ok(mut idle) = inner.idle_connections.try_lock() {
            if idle.len() < inner.config.max_connections as usize {
                idle.push(conn);
                inner.connection_available.notify_one();
                // 归还信号量许可
                inner.connection_semaphore.add_permits(1);
            } else {
                // 空闲队列已满，丢弃连接
                inner.total_count.fetch_sub(1, Ordering::SeqCst);
                // 归还信号量许可
                inner.connection_semaphore.add_permits(1);
            }
            return;
        }

        // 异步路径：在 tokio 运行时中执行
        if tokio::runtime::Handle::try_current().is_ok() {
            tokio::spawn(async move {
                let mut idle = inner.idle_connections.lock().await;
                if idle.len() < inner.config.max_connections as usize {
                    idle.push(conn);
                    inner.connection_available.notify_one();
                } else {
                    // 空闲队列已满，丢弃连接
                    inner.total_count.fetch_sub(1, Ordering::SeqCst);
                }
                // 归还信号量许可
                inner.connection_semaphore.add_permits(1);
            });
        } else {
            // 没有 tokio 运行时，丢弃连接
            inner.total_count.fetch_sub(1, Ordering::SeqCst);
            // 归还信号量许可
            inner.connection_semaphore.add_permits(1);
        }
    }

    /// 获取连接池状态
    ///
    /// 返回当前连接池的统计信息，包括总连接数、活跃连接数和空闲连接数。
    ///
    /// # Returns
    ///
    /// 连接池状态信息
    ///
    /// # Example
    ///
    /// ```rust
    /// use dbnexus::DbPool;
    ///
    /// # async fn example(pool: &DbPool) {
    /// let status = pool.status();
    /// println!("Total: {}, Active: {}, Idle: {}",
    ///     status.total, status.active, status.idle);
    /// # }
    /// ```
    pub fn status(&self) -> PoolStatus {
        let total = self.inner.total_count.load(Ordering::SeqCst);
        let active = self.inner.active_count.load(Ordering::SeqCst);
        let wait_count = self.inner.wait_count.load(Ordering::SeqCst);
        let max_waiters = self.inner.max_waiters.load(Ordering::SeqCst);
        let borrow_count = self.inner.borrow_count.load(Ordering::SeqCst);
        let max_active = self.inner.max_active.load(Ordering::SeqCst);

        PoolStatus {
            total,
            active,
            idle: total.saturating_sub(active),
            wait_count,
            max_waiters,
            borrow_count,
            max_active,
        }
    }

    /// 获取连接池告警指标
    ///
    /// 从 metrics_collector（如果启用）获取告警相关指标，包括：
    /// - `slow_acquires`：获取时长超过 1s 的次数
    /// - `timeout_errors`：获取超时总次数（warn + error + critical 之和）
    /// - `critical_timeouts`：严重超时（≥10s）次数
    /// - `wait_count`：当前正在等待获取连接的协程数
    /// - `max_waiters`：历史最大并发等待者峰值
    ///
    /// # Returns
    ///
    /// 连接池告警指标（始终返回有效值，metrics 未启用时所有计数为 0）
    #[cfg(feature = "metrics")]
    pub fn pool_metrics(&self) -> PoolMetrics {
        let wait_count = self.inner.wait_count.load(Ordering::SeqCst);
        let max_waiters = self.inner.max_waiters.load(Ordering::SeqCst);
        if let Some(ref collector) = self.inner.metrics_collector {
            let stats = collector.connection_acquire_stats();
            PoolMetrics {
                slow_acquires: stats.slow_acquires,
                timeout_errors: stats.timeout_warn + stats.timeout_error + stats.timeout_critical,
                critical_timeouts: stats.timeout_critical,
                wait_count,
                max_waiters,
            }
        } else {
            PoolMetrics {
                slow_acquires: 0,
                timeout_errors: 0,
                critical_timeouts: 0,
                wait_count,
                max_waiters,
            }
        }
    }

    /// 获取配置
    ///
    /// 返回连接池的配置引用。
    ///
    /// # Returns
    ///
    /// 连接池配置的引用
    ///
    /// # Example
    ///
    /// ```rust
    /// use dbnexus::DbPool;
    ///
    /// # async fn example(pool: &DbPool) {
    /// let config = pool.config();
    /// println!("Max connections: {}", config.max_connections);
    /// # }
    /// ```
    pub fn config(&self) -> &DbConfig {
        &self.inner.config
    }

    /// 运行自动迁移
    ///
    /// 如果配置中启用了 `auto_migrate`，此方法会在连接池创建后自动执行迁移。
    /// 也可以手动调用此方法来执行迁移。
    ///
    /// # Returns
    ///
    /// 成功应用的迁移数量
    #[cfg(feature = "auto-migrate")]
    pub async fn run_auto_migrate(&self) -> Result<u32, DbError> {
        if let Some(ref migrations_dir) = self.inner.config.migrations_dir {
            self.run_migrations(migrations_dir).await
        } else {
            Ok(0)
        }
    }

    /// 手动运行迁移
    ///
    /// # Arguments
    ///
    /// * `migrations_dir` - 迁移文件目录路径
    ///
    /// # Returns
    ///
    /// 成功应用的迁移数量
    #[cfg(feature = "auto-migrate")]
    pub async fn run_migrations(&self, migrations_dir: &std::path::Path) -> Result<u32, DbError> {
        use crate::database::migration::MigrationExecutor;

        let db_type = self
            .inner
            .config
            .database_type()
            .map_err(|e| DbError::Config(e.to_string()))?;

        // 获取一个连接来执行迁移
        let connection = self.acquire_connection().await?;

        // 从 DbConnection 提取 SeaORM 连接用于迁移执行器
        let connection_for_migration = connection.as_sea_orm()?.clone();

        let mut executor = MigrationExecutor::new(connection_for_migration, db_type);

        let applied = executor.run_migrations(migrations_dir).await?;

        // 归还连接到池中
        self.release_connection(connection);

        Ok(applied)
    }
}

/// DbPool 的优雅关闭
impl Drop for DbPool {
    fn drop(&mut self) {
        // 通知后台健康检查任务关闭
        self.inner.health_check_shutdown.notify_one();
    }
}

/// 连接池告警指标（用于分级告警和监控）
///
/// 包含所有与连接池告警相关的指标，用于告警规则配置和监控告警。
#[cfg(feature = "metrics")]
#[derive(Debug, Clone)]
pub struct PoolMetrics {
    /// 慢获取次数（>3s）
    pub slow_acquires: u64,
    /// 超时总次数
    pub timeout_errors: u64,
    /// 严重级超时次数（>=10s）
    pub critical_timeouts: u64,
    /// 当前等待者数量
    pub wait_count: u32,
    /// 最大等待者数量（历史峰值）
    pub max_waiters: u32,
}

/// 连接池状态
#[derive(Debug, Clone)]
pub struct PoolStatus {
    /// 总连接数
    pub total: u32,

    /// 活跃连接数
    pub active: u32,

    /// 空闲连接数
    pub idle: u32,

    /// 当前等待连接的请求数
    pub wait_count: u32,

    /// 最大等待计数（历史峰值）
    pub max_waiters: u32,

    /// 借用次数
    pub borrow_count: u64,

    /// 最大活跃连接数（历史峰值）
    pub max_active: u32,
}

// 实现 ConnectionPool trait
#[async_trait]
impl super::ConnectionPool for DbPool {
    async fn get_session(&self, role: &str) -> DbResult<Session> {
        self.get_session(role).await
    }

    fn status(&self) -> PoolStatus {
        self.status()
    }

    fn config(&self) -> &DbConfig {
        self.config()
    }
}