backend/backend/app/core/cache.py

"""缓存服务：Redis → 内存回退（Redis 不可用时自动降级，无连接等待）"""

import json
import logging
from collections import OrderedDict

logger = logging.getLogger(__name__)


class CacheService:
    """内存缓存实现（生产环境可替换为 Redis），内存模式最多保留 1000 条"""

    MAX_MEMORY_ITEMS = 1000

    def __init__(self):
        self._store: OrderedDict[str, dict] = OrderedDict()
        self._redis_failed = False
        self._redis = None

    async def _get_redis(self):
        """尝试获取 Redis 连接，失败即降级（每 60 秒重试一次）"""
        if self._redis_failed:
            import time as _time
            if getattr(self, '_redis_retry_at', 0) > _time.monotonic():
                return None
            self._redis_failed = False
        if self._redis is not None:
            return self._redis
        try:
            from redis.asyncio import Redis
            from app.config import settings
            r = Redis.from_url(settings.REDIS_URL, decode_responses=True, socket_connect_timeout=1)
            await r.ping()
            self._redis = r
            return r
        except Exception:
            self._redis_failed = True
            self._redis = None
            import time as _time
            self._redis_retry_at = _time.monotonic() + 60
            logger.info("Redis 不可用，使用内存缓存（60 秒后重试）")
            return None

    async def get(self, key: str) -> dict | None:
        r = await self._get_redis()
        if r:
            try:
                val = await r.get(key)
                return json.loads(val) if val else None
            except Exception:
                logger.exception("Redis GET failed: %s", key)
                return None
        val = self._store.get(key)
        if val is not None:
            self._store.move_to_end(key)
        return val

    async def set(self, key: str, value: dict, ttl: int = 300) -> bool:
        r = await self._get_redis()
        if r:
            try:
                await r.set(key, json.dumps(value, default=str), ex=ttl)
                return True
            except Exception:
                logger.exception("Redis SET failed: %s", key)
                return False
        # 内存模式：自动淘汰最旧条目，避免无限增长
        # 注：OrderedDict.__setitem__ 在 Python 3.9+ 已自动移到末尾，无需 move_to_end
        self._store[key] = value
        if len(self._store) > self.MAX_MEMORY_ITEMS:
            self._store.popitem(last=False)
        return True

    async def mget(self, keys: list[str]) -> list[dict | None]:
        """批量获取，顺序对应 keys 列表。Redis 不可用时降级到内存模式。"""
        r = await self._get_redis()
        if r:
            try:
                vals = await r.mget(*keys)
                return [json.loads(v) if v else None for v in vals]
            except Exception:
                logger.exception("Redis MGET failed")
                return [None] * len(keys)
        return [self._store.get(k) for k in keys]

    async def delete(self, key: str):
        r = await self._get_redis()
        if r:
            import contextlib
            with contextlib.suppress(Exception):
                await r.delete(key)
        else:
            self._store.pop(key, None)

    async def pop(self, key: str) -> dict | None:
        r = await self._get_redis()
        if r:
            try:
                val = await r.get(key)
                if val is not None:
                    await r.delete(key)
                return json.loads(val) if val else None
            except Exception:
                logger.exception("Redis POP failed: %s", key)
                return None
        return self._store.pop(key, None)

    async def get_or_set(self, key: str, loader, ttl: int = 300) -> dict | None:
        """原子化 get-then-set-or-fallback：先查缓存，命中即返回；未命中调用 loader() 加载、写缓存后返回。

        防缓存雪崩（cache stampede）：Redis 模式用 SETNX 互斥锁，内存模式用 asyncio.Lock。
        未抢到锁的请求会短暂自旋等待，避免重复计算。
        """
        cached = await self.get(key)
        if cached is not None:
            return cached

        lock_key = f"lock:{key}"
        if r := await self._get_redis():
            # Redis 模式：SETNX 互斥锁
            try:
                locked = await r.set(lock_key, "1", nx=True, ex=10)
            except Exception:
                locked = True  # 锁操作失败时直接执行 loader，不做等待
            if locked:
                try:
                    value = await loader()
                    if value is not None:
                        await r.set(key, json.dumps(value, default=str), ex=ttl)
                    return value
                finally:
                    await r.delete(lock_key)
            # 未抢到锁：自旋等待（最多 3 次，每次 100ms）
            import asyncio
            for _ in range(3):
                await asyncio.sleep(0.1)
                cached = await self.get(key)
                if cached is not None:
                    return cached
            # 超时仍未获缓存：直接执行（保险）
            value = await loader()
            if value is not None:
                await self.set(key, value, ttl=ttl)
            return value
        else:
            # 内存模式：asyncio.Lock 互斥
            import asyncio
            if not hasattr(self, '_mem_locks'):
                self._mem_locks: dict[str, asyncio.Lock] = {}
            if key not in self._mem_locks:
                self._mem_locks[key] = asyncio.Lock()
            async with self._mem_locks[key]:
                # 双检锁（double-check）：获取锁后可能已被其他协程写入
                double_check = await self.get(key)
                if double_check is not None:
                    return double_check
                value = await loader()
                if value is not None:
                    self._store[key] = value
                    self._store.move_to_end(key)
                return value

    async def invalidate_user(self, user_id: str):
        await self.delete(f"user:{user_id}:profile")

    async def invalidate_search_cache(self):
        """管道运行后失效所有搜索相关缓存。"""
        await self.delete_pattern("search:advanced:*")
        await self.delete_pattern("search:year_counts:*")
        await self.delete("search:year_counts:all")
        await self.delete("journals:map")
        await self.delete_pattern("atm:*")
        await self.delete("filter-options")

    async def invalidate_tenant(self, tenant_id: str):
        await self.delete(f"tenant:{tenant_id}:plan")
        await self.delete(f"tenant:{tenant_id}:settings")

    async def delete_pattern(self, pattern: str):
        """Delete all keys matching a glob pattern.

        Redis 模式用 SCAN 0 MATCH pattern 迭代删除。
        内存模式用 OrderedDict key 前缀匹配删除。
        """
        r = await self._get_redis()
        if r:
            try:
                cursor = 0
                while True:
                    cursor, keys = await r.scan(cursor, match=pattern, count=100)
                    if keys:
                        await r.delete(*keys)
                    if cursor == 0:
                        break
            except Exception:
                logger.exception("Redis SCAN/DEL failed for pattern: %s", pattern)
        else:
            import fnmatch
            to_delete = [k for k in self._store if fnmatch.fnmatch(k, pattern)]
            for k in to_delete:
                self._store.pop(k, None)


cache = CacheService()