dbnexus 0.6.0-rc.4

// Copyright (c) 2026 Kirky.X
// SPDX-License-Identifier: MIT
//! 权限缓存（TTL + SWR）
//!
//! 提供 `PermissionCache`：基于 `DashMap` 的线程安全权限策略缓存，支持：
//! - **TTL 过期**：每个条目在插入时记录时间戳，超过 TTL 后视为过期
//! - **后台刷新**：`get` 命中过期条目时返回 `None` 并 `tokio::spawn` 后台刷新
//! - **SWR（stale-while-revalidate）**：启用时返回旧值并后台刷新，避免缓存击穿
//!
//! 与 `PermissionContext`（基于 oxcache）的关系：
//! - `PermissionContext` 提供容量限制的缓存 + 速率限制 + 单飞行防护
//! - `PermissionCache` 提供显式 TTL + SWR 后台刷新，适合需要"过期但仍可用"语义的场景
//!
//! 两者互补共存，不强制替换。

use std::sync::Arc;
use std::time::{Duration, Instant};

use dashmap::{DashMap, Entry};

use super::provider::PermissionProvider;
use super::types::RolePolicy;

/// 默认 TTL（5 分钟）
const DEFAULT_TTL: Duration = Duration::from_secs(300);
/// 默认最小刷新间隔（60 秒，防止刷新风暴）
const DEFAULT_REFRESH_INTERVAL: Duration = Duration::from_secs(60);

/// 缓存条目
#[derive(Clone, Debug)]
struct CacheEntry {
    /// 缓存的权限策略
    value: RolePolicy,
    /// 插入时间戳
    inserted_at: Instant,
}

/// 权限缓存配置（只读快照，构建后不可变）
#[derive(Clone, Debug)]
pub struct PermissionCacheConfig {
    /// 缓存 TTL（条目过期时间）
    pub ttl: Duration,
    /// 后台刷新最小间隔（防止同一 key 短时间内重复刷新）
    pub refresh_interval: Duration,
    /// 是否启用 SWR（stale-while-revalidate）模式
    pub stale_while_revalidate: bool,
}

impl Default for PermissionCacheConfig {
    fn default() -> Self {
        Self {
            ttl: DEFAULT_TTL,
            refresh_interval: DEFAULT_REFRESH_INTERVAL,
            stale_while_revalidate: true,
        }
    }
}

/// 权限缓存（TTL + SWR）
///
/// 基于 `DashMap` 实现，线程安全。可附加 `PermissionProvider` 用于后台刷新。
///
/// # 示例
///
/// ```ignore
/// use std::sync::Arc;
/// use std::time::Duration;
/// use dbnexus::permission::{PermissionCache, PermissionProvider, YamlPermissionProvider};
///
/// # async fn example() {
/// let provider: Arc<dyn PermissionProvider> = Arc::new(YamlPermissionProvider::new());
/// let cache = PermissionCache::new()
///     .with_ttl(Duration::from_secs(60))
///     .with_refresh_interval(Duration::from_secs(10))
///     .with_stale_while_revalidate(true)
///     .with_provider(provider);
///
/// cache.insert("admin", RolePolicy::default());
/// assert!(cache.get("admin").is_some());
/// # }
/// ```
pub struct PermissionCache {
    /// 内部 DashMap 存储（用 Arc 共享，clone 是廉价的引用计数操作）
    inner: Arc<DashMap<String, CacheEntry>>,
    /// 配置
    config: PermissionCacheConfig,
    /// 权限提供者（可选，用于后台刷新）
    provider: Option<Arc<dyn PermissionProvider>>,
    /// 上次刷新时间（key -> Instant），用于 refresh_interval 节流
    last_refresh: Arc<DashMap<String, Instant>>,
}

impl std::fmt::Debug for PermissionCache {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("PermissionCache")
            .field("entry_count", &self.inner.len())
            .field("config", &self.config)
            .field("has_provider", &self.provider.is_some())
            .finish_non_exhaustive()
    }
}

impl Default for PermissionCache {
    fn default() -> Self {
        Self::new()
    }
}

/// 原子节流占位：检查 `key` 距上次刷新是否已超过 `interval`。
///
/// 通过 `DashMap::entry` 持有分片写锁完成"检查 + 占位"（通过则立即写入当前时间），
/// 消除 check-then-act 竞态——并发调用只允许一个调用方通过并执行刷新。
/// 占位时间戳即刷新发起时刻，与既有语义一致（节流窗口从刷新发起起算；
/// 刷新失败不回滚，成功后由 `insert` 更新条目时间戳）。
fn try_reserve_refresh(
    last_refresh: &DashMap<String, Instant>,
    key: &str,
    interval: Duration,
) -> bool {
    match last_refresh.entry(key.to_string()) {
        Entry::Occupied(mut occupied) => {
            if occupied.get().elapsed() < interval {
                return false;
            }
            *occupied.get_mut() = Instant::now();
            true
        }
        Entry::Vacant(vacant) => {
            vacant.insert(Instant::now());
            true
        }
    }
}

impl PermissionCache {
    /// 创建新的权限缓存（使用默认配置）
    pub fn new() -> Self {
        Self {
            inner: Arc::new(DashMap::new()),
            config: PermissionCacheConfig::default(),
            provider: None,
            last_refresh: Arc::new(DashMap::new()),
        }
    }

    /// 链式设置 TTL
    pub fn with_ttl(mut self, ttl: Duration) -> Self {
        self.config.ttl = ttl;
        self
    }

    /// 链式设置后台刷新最小间隔
    pub fn with_refresh_interval(mut self, interval: Duration) -> Self {
        self.config.refresh_interval = interval;
        self
    }

    /// 链式启用/禁用 SWR（stale-while-revalidate）模式
    pub fn with_stale_while_revalidate(mut self, enabled: bool) -> Self {
        self.config.stale_while_revalidate = enabled;
        self
    }

    /// 链式附加权限提供者（用于后台刷新）
    pub fn with_provider(mut self, provider: Arc<dyn PermissionProvider>) -> Self {
        self.provider = Some(provider);
        self
    }

    /// 获取当前配置快照
    pub fn config(&self) -> &PermissionCacheConfig {
        &self.config
    }

    /// 插入或更新缓存条目（重置时间戳）
    pub fn insert(&self, key: &str, value: RolePolicy) {
        let entry = CacheEntry {
            value,
            inserted_at: Instant::now(),
        };
        self.inner.insert(key.to_string(), entry);
    }

    /// 失效单个条目（删除）
    pub fn invalidate(&self, key: &str) {
        self.inner.remove(key);
        self.last_refresh.remove(key);
    }

    /// 清空所有缓存条目
    pub fn clear(&self) {
        self.inner.clear();
        self.last_refresh.clear();
    }

    /// 当前缓存条目数量
    pub fn len(&self) -> usize {
        self.inner.len()
    }

    /// 缓存是否为空
    pub fn is_empty(&self) -> bool {
        self.inner.is_empty()
    }

    /// 查询缓存条目
    ///
    /// - **未过期**：返回 `Some(value)`
    /// - **已过期 + SWR 启用**：返回 `Some(旧值)` 并后台刷新
    /// - **已过期 + SWR 禁用**：返回 `None` 并后台刷新
    /// - **未命中**：返回 `None`（不触发刷新，调用方应主动 `insert`）
    pub fn get(&self, key: &str) -> Option<RolePolicy> {
        if let Some(entry) = self.inner.get(key) {
            let elapsed = entry.inserted_at.elapsed();
            if elapsed < self.config.ttl {
                // 未过期：直接返回
                return Some(entry.value.clone());
            }
            // 已过期
            self.maybe_spawn_refresh(key);
            if self.config.stale_while_revalidate {
                // SWR：返回旧值
                Some(entry.value.clone())
            } else {
                // 非 SWR：返回 None
                None
            }
        } else {
            None
        }
    }

    /// 检查条目是否过期（仅查时间戳，不触发刷新）
    pub fn is_expired(&self, key: &str) -> bool {
        if let Some(entry) = self.inner.get(key) {
            entry.inserted_at.elapsed() >= self.config.ttl
        } else {
            // 不存在的条目视为"过期"
            true
        }
    }

    /// 后台刷新条目（stale-while-revalidate）
    ///
    /// 使用 `refresh_interval` 节流，防止短时间内重复刷新。
    /// 刷新失败时保留旧值并记录 warn 日志。
    pub async fn refresh(&self, key: &str) {
        // 节流：原子"检查并占位"，避免 check-then-act 竞态导致并发重复刷新
        if !try_reserve_refresh(&self.last_refresh, key, self.config.refresh_interval) {
            return;
        }

        let provider = match &self.provider {
            Some(p) => p.clone(),
            None => {
                return;
            }
        };

        let key_owned = key.to_string();
        // 同步调用 provider（trait 方法是同步的）
        match provider.get_role_policy(&key_owned) {
            Some(new_policy) => {
                self.insert(&key_owned, new_policy);
            }
            None => {
                // 角色不存在：保留旧值（stale-while-revalidate 语义）
            }
        }
    }

    /// 触发后台刷新（如果配置了 provider 且未在节流窗口内）
    fn maybe_spawn_refresh(&self, key: &str) {
        if self.provider.is_none() {
            return;
        }
        // 节流 + 占位必须在调用线程内（spawn 之前）原子完成：
        // 若 insert 延迟到 spawn 的任务内，并发的 get 在占位写入前仍会通过节流检查，
        // 造成重复派生后台任务
        if !try_reserve_refresh(&self.last_refresh, key, self.config.refresh_interval) {
            return;
        }
        let key_owned = key.to_string();
        let provider = self.provider.clone().unwrap();
        // Arc<DashMap> clone 是廉价的引用计数，spawn 任务写入会反映到原 cache
        let inner = self.inner.clone();

        tokio::spawn(async move {
            match provider.get_role_policy(&key_owned) {
                Some(new_policy) => {
                    let entry = CacheEntry {
                        value: new_policy,
                        inserted_at: Instant::now(),
                    };
                    inner.insert(key_owned, entry);
                }
                None => {
                    // 后台刷新返回 None，保留旧值
                }
            }
        });
    }
}

impl Clone for PermissionCache {
    /// 克隆缓存（共享内部 DashMap）
    ///
    /// 内部 DashMap 用 `Arc` 包装，clone 是廉价的引用计数操作，
    /// 写入会反映到所有 clone 副本。
    fn clone(&self) -> Self {
        Self {
            inner: self.inner.clone(),
            config: self.config.clone(),
            provider: self.provider.clone(),
            last_refresh: self.last_refresh.clone(),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicUsize, Ordering};

    use crate::access::permission::PermissionProviderError;
    use crate::access::{PermissionAction, TablePermission};

    fn sample_policy(table: &str) -> RolePolicy {
        RolePolicy {
            tables: vec![TablePermission {
                name: table.to_string(),
                operations: vec![PermissionAction::Select],
            }],
        }
    }

    #[tokio::test]
    async fn test_insert_and_get_fresh() {
        let cache = PermissionCache::new().with_ttl(Duration::from_secs(60));
        cache.insert("admin", sample_policy("users"));
        let got = cache.get("admin");
        assert!(got.is_some());
        assert_eq!(got.unwrap().tables.len(), 1);
    }

    #[tokio::test]
    async fn test_get_missing_returns_none() {
        let cache = PermissionCache::new();
        assert!(cache.get("ghost").is_none());
    }

    #[tokio::test]
    async fn test_expired_without_swr_returns_none() {
        let cache = PermissionCache::new()
            .with_ttl(Duration::from_millis(1))
            .with_stale_while_revalidate(false);
        cache.insert("admin", sample_policy("users"));
        // 等待过期
        tokio::time::sleep(Duration::from_millis(20)).await;
        assert!(cache.get("admin").is_none());
    }

    #[tokio::test]
    async fn test_expired_with_swr_returns_stale() {
        let cache = PermissionCache::new()
            .with_ttl(Duration::from_millis(1))
            .with_stale_while_revalidate(true);
        cache.insert("admin", sample_policy("users"));
        tokio::time::sleep(Duration::from_millis(20)).await;
        let got = cache.get("admin");
        assert!(got.is_some(), "SWR should return stale value");
    }

    #[tokio::test]
    async fn test_is_expired() {
        let cache = PermissionCache::new().with_ttl(Duration::from_millis(10));
        cache.insert("admin", sample_policy("users"));
        assert!(!cache.is_expired("admin"));
        tokio::time::sleep(Duration::from_millis(20)).await;
        assert!(cache.is_expired("admin"));
        assert!(cache.is_expired("ghost"));
    }

    #[tokio::test]
    async fn test_invalidate_and_clear() {
        let cache = PermissionCache::new();
        cache.insert("a", sample_policy("t1"));
        cache.insert("b", sample_policy("t2"));
        assert_eq!(cache.len(), 2);
        cache.invalidate("a");
        assert_eq!(cache.len(), 1);
        assert!(cache.get("a").is_none());
        cache.clear();
        assert!(cache.is_empty());
    }

    /// 计数 provider：统计 get_role_policy 调用次数，并阻塞一段时间
    /// 放大并发窗口（模拟慢 provider）
    struct CountingProvider {
        calls: AtomicUsize,
        /// provider 内部阻塞时长（放大竞态窗口 / 模拟慢源）
        delay: Duration,
    }

    impl CountingProvider {
        fn new() -> Self {
            Self::with_delay(Duration::from_millis(50))
        }

        fn with_delay(delay: Duration) -> Self {
            Self {
                calls: AtomicUsize::new(0),
                delay,
            }
        }
    }

    impl PermissionProvider for CountingProvider {
        fn get_role_policy(&self, _role: &str) -> Option<RolePolicy> {
            self.calls.fetch_add(1, Ordering::SeqCst);
            std::thread::sleep(self.delay);
            Some(sample_policy("users"))
        }

        fn check_access(
            &self,
            role: &str,
            table: &str,
            operation: PermissionAction,
        ) -> Result<bool, PermissionProviderError> {
            Ok(self.get_role_policy(role).is_some_and(|p| {
                p.tables
                    .iter()
                    .any(|t| t.name == table && t.operations.contains(&operation))
            }))
        }

        fn get_roles(&self) -> Vec<String> {
            vec!["admin".to_string()]
        }
    }

    /// 节流占位必须原子：并发 refresh 同一 key 时 provider 只应被调用 1 次
    /// （未修复的 check-then-act 竞态下 8 个并发都会通过节流检查）
    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn test_concurrent_refresh_single_provider_call() {
        let provider = Arc::new(CountingProvider::new());
        let cache = PermissionCache::new()
            .with_ttl(Duration::from_millis(1))
            .with_refresh_interval(Duration::from_secs(60))
            .with_provider(provider.clone());
        cache.insert("admin", sample_policy("users"));
        // 等待条目过期，使 refresh 走完整刷新路径
        tokio::time::sleep(Duration::from_millis(20)).await;

        let mut handles = Vec::new();
        for _ in 0..8 {
            let cache = cache.clone();
            handles.push(tokio::spawn(async move {
                cache.refresh("admin").await;
            }));
        }
        for h in handles {
            h.await.unwrap();
        }

        assert_eq!(provider.calls.load(Ordering::SeqCst), 1);
    }

    /// maybe_spawn_refresh 的占位必须在 spawn 前由调用线程写入：
    /// 并发 get 过期 key 只应派生一次后台刷新
    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn test_concurrent_get_single_background_refresh() {
        let provider = Arc::new(CountingProvider::new());
        let cache = PermissionCache::new()
            .with_ttl(Duration::from_millis(1))
            .with_refresh_interval(Duration::from_secs(60))
            .with_stale_while_revalidate(true)
            .with_provider(provider.clone());
        cache.insert("admin", sample_policy("users"));
        tokio::time::sleep(Duration::from_millis(20)).await;

        let mut handles = Vec::new();
        for _ in 0..8 {
            let cache = cache.clone();
            handles.push(tokio::spawn(async move {
                let _ = cache.get("admin");
            }));
        }
        for h in handles {
            h.await.unwrap();
        }
        // 等待后台刷新任务执行完毕
        tokio::time::sleep(Duration::from_millis(100)).await;

        assert_eq!(provider.calls.load(Ordering::SeqCst), 1);
    }

    // =========================================================================
    // 强同步并发回归测试（Barrier 多轮竞速版）
    //
    // 上面两个普通并发测试的任务各自独立 spawn，抵达节流检查的时间点分散，
    // 竞态窗口（纳秒到微秒级）难以命中——盲审实证：把实现改回 check-then-act
    // 后二者依然通过，回归保护力不足。以下测试改用 `tokio::sync::Barrier`
    // 强同步 + 多轮独立竞速：
    //
    // - Barrier + 倒计数锁步放行：16 个任务先在 Barrier 汇合，再经原子倒
    //   计数（race_start）近似同时起跑，把"检查 → 占位"窗口重叠到同一时刻；
    //   worker_threads = 16 保证竞速时每个任务独占线程（真实并行）；
    // - 多轮：每轮开始前 `invalidate` 清空节流时间戳，重开一个互相独立的
    //   竞态窗口。单轮能否命中取决于线程唤醒抖动（概率事件），多轮累积后
    //   回退到非原子 check-then-act 的实现会被稳定捕获（自证实验：回退后
    //   refresh 版测试连续 8/8 FAILED，每轮多穿透 1-8 次 provider 调用）；
    // - provider 内部 sleep 数十毫秒放大窗口：一旦多个竞速者同时穿透节流，
    //   重复的 provider 调用会被 AtomicUsize 计数捕获。
    //
    // 断言：provider 恰好每轮被调用 1 次（原子占位下每轮确定恰好 1 次穿透，
    // 连续 10 次运行无假阳性）。
    //
    // 原有两个普通并发测试保留：它们覆盖常规 spawn 并发下的端到端行为，
    // 与 Barrier 版的"竞态探测器"定位互补。
    // =========================================================================

    /// 竞速任务数（与 worker_threads 相同，保证放行时全员并行）
    const RACE_TASKS: usize = 16;
    /// 独立竞速轮数（单轮命中是概率事件，多轮累积保证稳定检出）
    const RACE_ROUNDS: usize = 24;

    /// 竞速起跑线：Barrier 负责全员汇合，但 futex 唤醒有微秒级抖动，
    /// 不足以命中纳秒级竞态窗口。汇合后各任务对原子倒计数做 fetch_sub
    /// 并等待归零——最后一个减到 0 的任务立即起跑，其余任务在一个缓存行
    /// 传播周期内观察到 0 并同时起跑，实现近似锁步放行。
    ///
    /// 等待采用"短自旋 + yield"混合：纯自旋会阻塞 worker 线程，若同一线程
    /// 的任务队列中还有未 poll 的竞速者，会使其永远无法运行（倒计数无法
    /// 归零）导致死锁；短自旋保证计数临近归零时纳秒级起跑，兜底 yield
    /// 保证饿死不可能发生。
    async fn race_start(barrier: &tokio::sync::Barrier, latch: &AtomicUsize) {
        barrier.wait().await;
        latch.fetch_sub(1, Ordering::AcqRel);
        while latch.load(Ordering::Acquire) != 0 {
            for _ in 0..4096 {
                std::hint::spin_loop();
                if latch.load(Ordering::Acquire) == 0 {
                    return;
                }
            }
            tokio::task::yield_now().await;
        }
    }

    /// 强同步并发 refresh：每轮 Barrier 汇合 + 原子倒计数锁步放行后，
    /// RACE_TASKS 个任务同时调用 refresh，节流占位必须原子——
    /// 每轮 provider 只应被调用 1 次
    #[tokio::test(flavor = "multi_thread", worker_threads = 16)]
    async fn test_barrier_refresh_racers_single_provider_call() {
        // 20ms/轮 × 16 轮：控制测试耗时的同时保留"数十毫秒级"窗口放大
        let provider = Arc::new(CountingProvider::with_delay(Duration::from_millis(20)));
        let cache = PermissionCache::new()
            .with_refresh_interval(Duration::from_secs(60))
            .with_provider(provider.clone());

        // refresh 不依赖缓存条目，每轮清空节流表即可重开竞态窗口
        let barrier = Arc::new(tokio::sync::Barrier::new(RACE_TASKS));
        for _ in 0..RACE_ROUNDS {
            cache.invalidate("admin");
            let latch = Arc::new(AtomicUsize::new(RACE_TASKS));
            let mut handles = Vec::with_capacity(RACE_TASKS);
            for _ in 0..RACE_TASKS {
                let cache = cache.clone();
                let barrier = barrier.clone();
                let latch = latch.clone();
                handles.push(tokio::spawn(async move {
                    race_start(&barrier, &latch).await;
                    cache.refresh("admin").await;
                }));
            }
            for h in handles {
                h.await.unwrap();
            }
        }

        assert_eq!(
            provider.calls.load(Ordering::SeqCst),
            RACE_ROUNDS,
            "锁步放行后每轮并发 refresh 仍只允许 1 次 provider 调用"
        );
    }

    /// 强同步 refresh/get 混合路径：两条刷新路径共享同一 last_refresh 节流表，
    /// 每轮锁步放行后同时触发，总 provider 调用次数（含后台刷新）每轮仍应为 1
    #[tokio::test(flavor = "multi_thread", worker_threads = 16)]
    async fn test_barrier_mixed_refresh_and_get_single_provider_call() {
        let provider = Arc::new(CountingProvider::with_delay(Duration::from_millis(20)));
        let cache = PermissionCache::new()
            .with_ttl(Duration::from_millis(1))
            .with_refresh_interval(Duration::from_secs(60))
            .with_stale_while_revalidate(true)
            .with_provider(provider.clone());

        let barrier = Arc::new(tokio::sync::Barrier::new(RACE_TASKS));
        for _ in 0..RACE_ROUNDS {
            // invalidate 同时清空条目与节流表，隔离上一轮（含后台任务）的写入
            cache.invalidate("admin");
            cache.insert("admin", sample_policy("users"));
            // 等待条目过期（ttl = 1ms），使 get 走 maybe_spawn_refresh 后台路径
            tokio::time::sleep(Duration::from_millis(3)).await;

            let latch = Arc::new(AtomicUsize::new(RACE_TASKS));
            let mut handles = Vec::with_capacity(RACE_TASKS);
            for i in 0..RACE_TASKS {
                let cache = cache.clone();
                let barrier = barrier.clone();
                let latch = latch.clone();
                handles.push(tokio::spawn(async move {
                    race_start(&barrier, &latch).await;
                    if i % 2 == 0 {
                        // 偶数任务走显式 refresh 路径
                        cache.refresh("admin").await;
                    } else {
                        // 奇数任务走 get 过期 key → maybe_spawn_refresh 后台路径
                        let _ = cache.get("admin");
                    }
                }));
            }
            for h in handles {
                h.await.unwrap();
            }
        }
        // 等待最后一轮 get 派生的后台刷新任务执行完毕（provider 内部 sleep 20ms）
        tokio::time::sleep(Duration::from_millis(100)).await;

        assert_eq!(
            provider.calls.load(Ordering::SeqCst),
            RACE_ROUNDS,
            "refresh 与 get 两条路径并发时每轮仍只允许 1 次 provider 调用"
        );
    }
}