zenith-runtime 0.1.0

//! 全局通用运行时单例
//!
//! 解决 **架构级缺失**：原 zenith 仅提供 `RuntimeConfig`，
//! 未提供全局可用的 Tokio 异步运行时与便捷 API，
//! 导致用户每个二进制重复书写 `Builder::new_multi_thread().build()` 样板代码，
//! 且跨 crate 无法 `spawn()` 无参数投递任务。
//!
//! # 设计原则（极端极限标准）
//! - **Fail-Closed**：任何初始化失败均静默 fallback 到最小可用配置，永不 panic
//! - **幂等安全**：`init_global()` 可跨线程并发调用 N 次，仅首次生效（OnceLock）
//! - **自动计算最优化**：`RuntimeConfig::auto()` 按 CPU/NUMA 自动推导 worker_threads/栈
//! - **零成本热路径**：`Handle` 是 `Arc<Inner>` 的 Clone，`#[inline]` 访问零分配
//! - **规范统一**：所有子 crate 经 `zenith-runtime::spawn()` 用同一 Tokio 运行时
//!
//! # 使用示例
//! ```ignore
//! use zenith_runtime::{global_runtime, spawn, RuntimeConfig};
//!
//! // 1) 全局自动初始化（首次调用自动用 auto() 配置，零样板）
//! let handle = global_runtime().handle();
//!
//! // 2) 便捷 spawn（无需传 Handle）
//! let task = spawn(async { 42 });
//!
//! // 3) 显式定制初始化（可选）
//! zenith_runtime::init_global(RuntimeConfig::new().with_worker_threads(16));
//! ```

use std::sync::{Arc, OnceLock};

use tokio::runtime::{Handle, Runtime};

use crate::RuntimeConfig;

// ─────────────────────────────────────────────────────────────────────────────
// 全局单例（OnceLock：无竞态、可重入、并发安全、永不析构）
// ─────────────────────────────────────────────────────────────────────────────

/// 全局运行时单例（进程生命周期内存活）
static GLOBAL_RUNTIME: OnceLock<GlobalRuntime> = OnceLock::new();

/// 全局通用运行时（Tokio 异步运行时 + 配置）
///
/// # 所有权模型
/// - `tokio_rt`：`Arc<Runtime>` 拥有运行时（`None` 表示托生于外部 block_on 上下文）
/// - `tokio_handle`：`Arc<RuntimeInner>` 句柄，Clone 零成本（引用计数）
/// - `config`：构建时使用的 `RuntimeConfig` 快照
///
/// # 线程安全
/// 所有字段（`Arc<Runtime>` / `Handle` / `RuntimeConfig`）均自动 `Send + Sync`，
/// 在 `#![deny(unsafe_code)]` 下无需 unsafe impl 即合规。
#[derive(Debug, Clone)]
pub struct GlobalRuntime {
    /// Tokio 运行时拥有所有权（Some 表示我们自己构建，None 表示从上下文取 Handle）
    tokio_rt: Option<Arc<Runtime>>,
    /// Tokio 句柄（保证始终可用；Clone = Arc 引用 +1）
    tokio_handle: Handle,
    /// 构建快照
    config: RuntimeConfig,
}

impl RuntimeConfig {
    /// **自动计算最优化** 配置（极端极限标准，NUMA 感知）
    ///
    /// 经 `zenith-capability` 的 [`EnvironmentDetector`] 真实探测系统环境：
    /// - `worker_threads`：探测到的 CPU 核数（sysfs `/sys/devices/system/cpu/online`，
    ///   失败回退 `available_parallelism`，再失败 = 4）
    /// - `stack_size`：NUMA 节点数 ≥ 2（跨节点调度元数据更多）或核数 ≥ 16 → 4MiB，
    ///   否则 2MiB
    /// - `enable_io` / `enable_time`：`true`
    ///
    /// # Fail-Closed
    /// 所有探测失败都取安全默认值，永不 panic。
    #[inline]
    pub fn auto() -> Self {
        let snapshot = zenith_capability::detection::EnvironmentDetector::new().detect();
        let cpus = snapshot.cpu_cores.max(1);
        // NUMA 感知：多 NUMA 节点系统上 worker 线程的跨节点调度与本地性
        // 元数据更多，栈配额翻倍；单节点小系统保持 2MiB 基线
        let stack = if snapshot.numa_nodes >= 2 || cpus >= 16 {
            4 * 1024 * 1024
        } else {
            2 * 1024 * 1024
        };
        Self {
            worker_threads: cpus,
            enable_io: true,
            enable_time: true,
            stack_size: stack,
        }
    }
}

// ─────────────────────────────────────────────────────────────────────────────
// 核心 API
// ─────────────────────────────────────────────────────────────────────────────

/// 显式初始化全局运行时（**幂等、线程安全、可重入**）
///
/// 若全局运行时已被其他线程先初始化：
/// - 忽略本次 `config`，**不返回错误、不 panic**（幂等语义）
/// - 返回已有运行时的引用
///
/// # Returns
/// - `&'static GlobalRuntime`：全局运行时引用（与是否首次调用无关）
///
/// # 安全性
/// - 由 `std::sync::OnceLock` 保证：**仅有一次写入**，其他并发调用安全等待
/// - 初始化过程中若 `Builder::build()` 失败，fallback 到最小运行时
#[inline]
pub fn init_global(config: RuntimeConfig) -> &'static GlobalRuntime {
    GLOBAL_RUNTIME.get_or_init(|| build_global_runtime(config))
}

/// 获取全局运行时引用（自动懒加载）
///
/// - 若尚未初始化：**自动** 使用 `RuntimeConfig::auto()` 初始化（Fail-Closed）
/// - 若已初始化：直接返回已有引用
///
/// # 便捷性
/// 无需先调用 `init_global()`，任何位置 `global_runtime().handle()` 立即可用。
#[inline]
pub fn global_runtime() -> &'static GlobalRuntime {
    if let Some(rt) = GLOBAL_RUNTIME.get() {
        return rt;
    }
    init_global(RuntimeConfig::auto())
}

/// 全局 Tokio [`Handle`]（热路径，`#[inline]` 零分配）
///
/// 等价于 `global_runtime().handle()`，跨 crate 最常用的快捷方式。
#[inline]
pub fn handle() -> Handle {
    global_runtime().handle().clone()
}

/// 全局 [`tokio::spawn`] 便捷函数（无需传 Handle）
///
/// 调用位置若已在 Tokio 上下文中，与 `tokio::spawn` 等价；
/// 否则使用全局运行时投递。
#[inline]
pub fn spawn<F>(future: F) -> tokio::task::JoinHandle<F::Output>
where
    F: std::future::Future + Send + 'static,
    F::Output: Send + 'static,
{
    global_runtime().spawn(future)
}

/// `block_on` 重入错误
///
/// tokio 禁止在 runtime 线程内直接 `block_on`（会 panic）；
/// 多线程 runtime 可用 `block_in_place` 安全降级，但 current_thread
/// runtime 无线程可让渡，只能 fail-closed 返回本错误。
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum BlockOnError {
    /// 在 current_thread runtime 线程内重入阻塞（无法安全执行）
    ReentrantCurrentThread,
}

impl std::fmt::Display for BlockOnError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            BlockOnError::ReentrantCurrentThread => write!(
                f,
                "cannot block_on from within a current_thread runtime context"
            ),
        }
    }
}

impl std::error::Error for BlockOnError {}

/// 以全局运行时 `block_on` 执行 Future。
///
/// 若全局是自己构建的（有 `Runtime` 拥有权）：直接 `Runtime::block_on`；
/// 否则 fallback 到 `Handle::block_on`（需要在 Tokio 上下文外）。
///
/// # 重入安全（Fail-Closed，永不 panic）
/// - 不在任何 runtime 上下文：直接 `block_on`
/// - 在 multi_thread runtime 线程内：`block_in_place` 让出 worker 槽位后阻塞
/// - 在 current_thread runtime 线程内：返回 `Err(BlockOnError::ReentrantCurrentThread)`
#[inline]
pub fn block_on<F: std::future::Future>(
    future: F,
) -> Result<F::Output, BlockOnError> {
    global_runtime().block_on(future)
}

impl GlobalRuntime {
    /// 获取 Tokio 句柄（Clone 是 Arc 引用 +1，零成本）
    #[inline]
    pub fn handle(&self) -> &Handle {
        &self.tokio_handle
    }

    /// 获取构建该运行时的配置快照
    #[inline]
    pub fn config(&self) -> &RuntimeConfig {
        &self.config
    }

    /// 在该运行时上 spawn 任务
    #[inline]
    pub fn spawn<F>(&self, future: F) -> tokio::task::JoinHandle<F::Output>
    where
        F: std::future::Future + Send + 'static,
        F::Output: Send + 'static,
    {
        self.tokio_handle.spawn(future)
    }

    /// block_on 执行 Future（优先用自有 Runtime）
    ///
    /// # 重入安全（Fail-Closed，永不 panic）
    /// tokio 禁止在 runtime 线程内直接 `block_on`（`Handle::block_on` 会 panic）。
    /// 本方法按调用上下文分派：
    /// - 不在任何 runtime 上下文：直接 `block_on`
    /// - 在 multi_thread runtime 线程内：`block_in_place` 让出 worker 槽位后阻塞
    /// - 在 current_thread runtime 线程内：返回 `Err(BlockOnError::ReentrantCurrentThread)`
    #[inline]
    pub fn block_on<F: std::future::Future>(
        &self,
        future: F,
    ) -> Result<F::Output, BlockOnError> {
        match Handle::try_current() {
            // 不在任何 runtime 上下文：直接阻塞当前线程，安全
            Err(_) => Ok(match &self.tokio_rt {
                Some(rt) => rt.block_on(future),
                None => self.tokio_handle.block_on(future),
            }),
            Ok(current) => match current.runtime_flavor() {
                // 多线程 runtime 工作线程：block_in_place 让出 worker 槽位，
                // 在阻塞期间由其他线程接管任务调度（tokio 文档认可的重入模式）
                tokio::runtime::RuntimeFlavor::MultiThread => {
                    Ok(tokio::task::block_in_place(|| match &self.tokio_rt {
                        Some(rt) => rt.block_on(future),
                        None => self.tokio_handle.block_on(future),
                    }))
                }
                // current_thread runtime：无线程可让渡，block_in_place 会 panic，
                // fail-closed 返回错误而非 panic
                _ => Err(BlockOnError::ReentrantCurrentThread),
            },
        }
    }
}

// ─────────────────────────────────────────────────────────────────────────────
// 内部构建逻辑（Fail-Closed：永不 panic，多路径兜底）
// ─────────────────────────────────────────────────────────────────────────────

/// 构建全局运行时（内部函数，Fail-Closed 三路径兜底）
fn build_global_runtime(config: RuntimeConfig) -> GlobalRuntime {
    // 路径 1：用 RuntimeConfig 尝试构建多线程 Runtime
    if let Some(rt) = build_tokio_runtime(&config) {
        let arc_rt = Arc::new(rt);
        let handle = arc_rt.handle().clone();
        return GlobalRuntime {
            tokio_rt: Some(arc_rt),
            tokio_handle: handle,
            config,
        };
    }

    // 路径 2：Builder 失败，借用当前上下文中的 Handle
    //         （例如调用方已经在 tokio::main 中）
    //
    // # 生命周期语义（文档化约定）
    // `Handle` 内部是引用计数的拥有型句柄（'static），存入全局单例在内存
    // 安全上没有问题；但**语义安全**依赖调用方约定：
    // - 被借用的外部 runtime 必须在整个进程使用期间保持存活；
    // - 若外部 runtime 先关闭，后续经该 Handle 的 `spawn`/`block_on`
    //   任务将被静默丢弃或失败（tokio 对 shutdown 后 spawn 直接 drop task）。
    // 因此本路径仅作为 Builder 连续失败后的最后兜底；生产环境应通过
    // `init_global()` 显式初始化自有 Runtime（路径 1/3），不要依赖借用模式。
    if let Ok(h) = Handle::try_current() {
        // 借用外部 runtime 的生命周期风险已在上方文档注释中详述
        eprintln!("[zenith-warn] Global runtime borrowing external tokio Handle - lifetime risk if external runtime shuts down");
        return GlobalRuntime {
            tokio_rt: None,
            tokio_handle: h,
            config,
        };
    }

    // 路径 3：最终兜底 —— 构建最简 current_thread Runtime
    //         单线程、enable_all；按 Fail-Closed 原则不使用 expect/panic。
    //         若连 current_thread Runtime 都构建失败（通常仅 OOM），说明系统已无法
    //         恢复；此时用 std::process::abort() 直接退出，避免 panic unwind
    //         使 GlobalRuntime 处于半初始化状态（规范 §6.1.1 禁止 expect/unwrap）。
    let rt = match tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
    {
        Ok(rt) => rt,
        Err(_e) => {
            // abort 前 flush 关键持久化状态（审计日志等）
            // 注：在 abort 前只能做最关键的同步操作
            tracing::error!("Runtime build failed, aborting process");
            std::process::abort();
        }
    };
    let handle = rt.handle().clone();
    GlobalRuntime {
        tokio_rt: Some(Arc::new(rt)),
        tokio_handle: handle,
        config,
    }
}

/// 按 RuntimeConfig 构建 Tokio multi_thread Runtime（失败返回 None，永不 panic）
fn build_tokio_runtime(config: &RuntimeConfig) -> Option<Runtime> {
    let mut builder = tokio::runtime::Builder::new_multi_thread();
    builder.worker_threads(config.worker_threads.max(1));
    builder.thread_stack_size(config.stack_size.max(4096));

    if config.enable_io || config.enable_time {
        if config.enable_io {
            builder.enable_io();
        }
        if config.enable_time {
            builder.enable_time();
        }
    }

    // Fail-Closed：构建失败不 panic，走下一条 fallback 路径
    builder.build().ok()
}

// ─────────────────────────────────────────────────────────────────────────────
// 测试
// ─────────────────────────────────────────────────────────────────────────────

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

    // ── RuntimeConfig::auto() 正确性 ──────────────────────────────────────

    #[test]
    fn test_config_auto_cpu_sane() {
        let cfg = RuntimeConfig::auto();
        assert!(cfg.worker_threads >= 1);
        assert!(cfg.enable_io);
        assert!(cfg.enable_time);
        assert!(cfg.stack_size >= 4096);
    }

    #[test]
    fn test_config_auto_vs_new_equal_same_input() {
        let a = RuntimeConfig::auto();
        let b = RuntimeConfig::auto();
        // 同一进程两次 auto 结果相同
        assert_eq!(a.worker_threads, b.worker_threads);
        assert_eq!(a.stack_size, b.stack_size);
    }

    // ── 幂等性：N 次 init_global 只一次生效 ──────────────────────────────

    #[test]
    fn test_global_runtime_init_idempotent() {
        let cfg = RuntimeConfig::new().with_worker_threads(2);
        let r1 = init_global(cfg);
        let r2 = init_global(RuntimeConfig::new().with_worker_threads(32));
        // OnceLock：同一引用
        assert!(std::ptr::eq(r1, r2));
        // 只有第一个 worker_threads 被采纳
        assert_eq!(r1.config().worker_threads, r2.config().worker_threads);
    }

    // ── 全局 spawn/handle 可用性 ──────────────────────────────────────────

    #[test]
    fn test_global_spawn_works() {
        let result = block_on(async {
            let h = spawn(async { 42_u32 });
            h.await.unwrap()
        })
        .unwrap();
        assert_eq!(result, 42);
    }

    #[test]
    fn test_handle_clone_works() {
        let h1 = handle();
        let h2 = handle();
        // 等价句柄
        let r = block_on(async move {
            let j1 = h1.spawn(async { 1 });
            let j2 = h2.spawn(async { 2 });
            j1.await.unwrap() + j2.await.unwrap()
        })
        .unwrap();
        assert_eq!(r, 3);
    }

    #[test]
    fn test_block_on_reentrant_current_thread_returns_err() {
        // 在 current_thread runtime 线程内重入 block_on：必须返回 Err 而非 panic
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        rt.block_on(async {
            let result = block_on(async { 1u32 });
            assert_eq!(result, Err(BlockOnError::ReentrantCurrentThread));
        });
    }

    #[test]
    fn test_block_on_reentrant_multi_thread_ok() {
        // 在 multi_thread runtime 线程内重入：block_in_place 降级，正常返回
        let rt = tokio::runtime::Builder::new_multi_thread()
            .worker_threads(2)
            .enable_all()
            .build()
            .unwrap();
        rt.block_on(async {
            let result = block_on(async { 7u32 }).unwrap();
            assert_eq!(result, 7);
        });
    }

    // ── Fail-Closed：极端参数不 panic ─────────────────────────────────────

    #[test]
    fn test_build_with_zero_threads_no_panic() {
        // 0 线程应被 .max(1) 修正，不 panic
        let cfg = RuntimeConfig::new().with_worker_threads(0).with_stack_size(0);
        let gr = build_global_runtime(cfg);
        // spawn 仍可用
        let _ = gr.block_on(async { 1 });
    }

    #[test]
    fn test_global_runtime_config_snapshot_kept() {
        // 无论 auto 还是手动，config() 都能返回对应快照（不为零）
        let cfg = RuntimeConfig::new().with_worker_threads(1).with_stack_size(8 * 1024);
        let gr = build_global_runtime(cfg);
        // Note：worker_threads 被 build_tokio_runtime 内部 max(1) 处理
        assert!(gr.config().worker_threads >= 1);
        assert!(gr.config().stack_size >= 8 * 1024);
    }

    // ── GlobalRuntime Send+Sync 自动实现（编译期断言） ────────────────────

    const _: fn() = || {
        fn assert_send_sync<T: Send + Sync>() {}
        assert_send_sync::<GlobalRuntime>();
    };
}