ctms/backend/app/core/permission_cache.py

"""权限缓存管理器

实现权限矩阵和成员身份的缓存，以提高权限检查性能。
采用内存缓存 + TTL 的方式，避免权限检查的 N+1 查询问题。
"""

from __future__ import annotations

import time
import uuid
from typing import Any

from sqlalchemy.ext.asyncio import AsyncSession

from app.core.project_permissions import (
    get_project_role_permissions,
    get_member_role,
)


class PermissionCache:
    """权限缓存管理器

    使用内存缓存存储权限矩阵和成员身份信息，减少数据库查询。
    每个缓存项都有 TTL（生存时间），过期后自动失效。
    """

    def __init__(self, default_ttl: int = 300):
        """初始化缓存管理器

        Args:
            default_ttl: 默认缓存生存时间（秒），默认5分钟
        """
        self.default_ttl = default_ttl
        self._project_permissions_cache: dict[str, tuple[Any, float]] = {}
        self._member_role_cache: dict[str, tuple[str | None, float]] = {}

    def _is_expired(self, timestamp: float, ttl: int) -> bool:
        """检查缓存是否已过期"""
        return time.time() - timestamp > ttl

    def _make_project_cache_key(self, study_id: uuid.UUID) -> str:
        """生成项目权限缓存键"""
        return f"project_permissions:{study_id}"

    def _make_member_cache_key(self, study_id: uuid.UUID, user_id: uuid.UUID) -> str:
        """生成成员角色缓存键"""
        return f"member_role:{study_id}:{user_id}"

    async def get_project_role_permissions(
        self,
        db: AsyncSession,
        study_id: uuid.UUID,
        ttl: int | None = None,
    ) -> dict[str, dict[str, dict[str, bool]]]:
        """获取项目权限矩阵（带缓存）

        Args:
            db: 数据库会话
            study_id: 项目ID
            ttl: 缓存生存时间（秒），默认使用 default_ttl

        Returns:
            权限矩阵：{role: {module: {action: bool}}}
        """
        if ttl is None:
            ttl = self.default_ttl

        cache_key = self._make_project_cache_key(study_id)

        # 检查缓存
        if cache_key in self._project_permissions_cache:
            cached_data, timestamp = self._project_permissions_cache[cache_key]
            if not self._is_expired(timestamp, ttl):
                return cached_data

        # 缓存未命中，从数据库查询
        permissions = await get_project_role_permissions(db, study_id)

        # 存储到缓存
        self._project_permissions_cache[cache_key] = (permissions, time.time())

        return permissions

    async def get_member_role(
        self,
        db: AsyncSession,
        study_id: uuid.UUID,
        user_id: uuid.UUID,
        ttl: int | None = None,
    ) -> str | None:
        """获取成员角色（带缓存）

        Args:
            db: 数据库会话
            study_id: 项目ID
            user_id: 用户ID
            ttl: 缓存生存时间（秒），默认使用 default_ttl

        Returns:
            成员在项目中的角色，如果不是成员则返回 None
        """
        if ttl is None:
            ttl = self.default_ttl

        cache_key = self._make_member_cache_key(study_id, user_id)

        # 检查缓存
        if cache_key in self._member_role_cache:
            cached_role, timestamp = self._member_role_cache[cache_key]
            if not self._is_expired(timestamp, ttl):
                return cached_role

        # 缓存未命中，从数据库查询
        role = await get_member_role(db, study_id, user_id)

        # 存储到缓存
        self._member_role_cache[cache_key] = (role, time.time())

        return role

    def invalidate_project_permissions(self, study_id: uuid.UUID) -> None:
        """失效项目权限缓存

        当项目权限被修改时调用此方法，清除相关的缓存。

        Args:
            study_id: 项目ID
        """
        cache_key = self._make_project_cache_key(study_id)
        if cache_key in self._project_permissions_cache:
            del self._project_permissions_cache[cache_key]

    def invalidate_member_role(self, study_id: uuid.UUID, user_id: uuid.UUID) -> None:
        """失效成员角色缓存

        当成员角色被修改时调用此方法，清除相关的缓存。

        Args:
            study_id: 项目ID
            user_id: 用户ID
        """
        cache_key = self._make_member_cache_key(study_id, user_id)
        if cache_key in self._member_role_cache:
            del self._member_role_cache[cache_key]

    def invalidate_all_member_roles(self, study_id: uuid.UUID) -> None:
        """失效项目中所有成员的角色缓存

        当项目权限矩阵被修改时调用此方法，清除项目中所有成员的缓存。

        Args:
            study_id: 项目ID
        """
        # 清除所有包含该项目ID的成员角色缓存
        keys_to_delete = [
            key for key in self._member_role_cache.keys()
            if key.startswith(f"member_role:{study_id}:")
        ]
        for key in keys_to_delete:
            del self._member_role_cache[key]

    def clear_all(self) -> None:
        """清除所有缓存

        用于测试或系统重启时清除所有缓存。
        """
        self._project_permissions_cache.clear()
        self._member_role_cache.clear()

    def get_cache_stats(self) -> dict[str, Any]:
        """获取缓存统计信息

        Returns:
            缓存统计信息：{
                'project_permissions_count': 项目权限缓存数,
                'member_role_count': 成员角色缓存数,
                'total_count': 总缓存数,
            }
        """
        return {
            "project_permissions_count": len(self._project_permissions_cache),
            "member_role_count": len(self._member_role_cache),
            "total_count": len(self._project_permissions_cache) + len(self._member_role_cache),
        }


# 全局缓存实例
_permission_cache: PermissionCache | None = None


def get_permission_cache() -> PermissionCache:
    """获取全局权限缓存实例"""
    global _permission_cache
    if _permission_cache is None:
        _permission_cache = PermissionCache()
    return _permission_cache


def set_permission_cache(cache: PermissionCache) -> None:
    """设置全局权限缓存实例（用于测试）"""
    global _permission_cache
    _permission_cache = cache