core/runtime/session_cache.py

"""
会话缓存：session_id 全局唯一，映射到 (proxy_key, type, account_id)。

当前架构下 session 绑定到某个 tab/account：

- tab 被关闭或切号时，需要批量失效该 tab 下的 session
- 单个 session 失效时，需要从缓存中移除，后续按完整历史重建
- 超过 TTL 的 session 在维护循环中被自动清理
"""

from dataclasses import dataclass
import time

from core.runtime.keys import ProxyKey

# Sessions older than this are eligible for eviction during maintenance.
SESSION_TTL_SECONDS = 1800.0  # 30 minutes


@dataclass
class SessionEntry:
    """单条会话：用于通过 session_id 反查 context/page 与账号。"""

    proxy_key: ProxyKey
    type_name: str
    account_id: str
    last_used_at: float


class SessionCache:
    """进程内会话缓存，不持久化、不跨进程。"""

    def __init__(self) -> None:
        self._store: dict[str, SessionEntry] = {}

    def get(self, session_id: str) -> SessionEntry | None:
        return self._store.get(session_id)

    def put(
        self,
        session_id: str,
        proxy_key: ProxyKey,
        type_name: str,
        account_id: str,
    ) -> None:
        self._store[session_id] = SessionEntry(
            proxy_key=proxy_key,
            type_name=type_name,
            account_id=account_id,
            last_used_at=time.time(),
        )

    def touch(self, session_id: str) -> None:
        entry = self._store.get(session_id)
        if entry is not None:
            entry.last_used_at = time.time()

    def delete(self, session_id: str) -> None:
        self._store.pop(session_id, None)

    def delete_many(self, session_ids: list[str] | set[str]) -> None:
        for session_id in session_ids:
            self._store.pop(session_id, None)

    def evict_stale(self, ttl: float = SESSION_TTL_SECONDS) -> list[str]:
        """Remove sessions older than *ttl* seconds. Returns evicted IDs."""
        now = time.time()
        stale = [
            sid
            for sid, entry in self._store.items()
            if (now - entry.last_used_at) > ttl
        ]
        for sid in stale:
            del self._store[sid]
        return stale

    def __contains__(self, session_id: str) -> bool:
        return session_id in self._store

    def __len__(self) -> int:
        return len(self._store)