Blame for app/core/cache.py

"""In-memory user snapshots (so most updates need ZERO database round-trips)
plus per-user locks (same user's updates run one at a time)."""
import asyncio
import time
import weakref
from typing import Optional

_TTL = 600  # seconds
_MAX = 20_000
_snapshots: dict[int, tuple[float, dict]] = {}
_locks: "weakref.WeakValueDictionary[int, asyncio.Lock]" = weakref.WeakValueDictionary()


def get(user_id: int) -> Optional[dict]:
    item = _snapshots.get(user_id)
    if item is None:
        return None
    expires, data = item
    if expires < time.monotonic():
        _snapshots.pop(user_id, None)
        return None
    return dict(data)


def put(user_id: int, data: dict) -> None:
    if len(_snapshots) >= _MAX:
        _snapshots.clear()
    _snapshots[user_id] = (time.monotonic() + _TTL, dict(data))


def invalidate(user_id: int) -> None:
    _snapshots.pop(user_id, None)


def lock_for(user_id: int) -> asyncio.Lock:
    lock = _locks.get(user_id)
    if lock is None:
        lock = asyncio.Lock()
        _locks[user_id] = lock
    return lock