Tech Stack Advisor - Code Viewer

"""Session management for conversation state (short-term memory)."""
import time
import uuid
from typing import Any
from ..core.logging import get_logger

logger = get_logger(__name__)

# In-memory session store (for demonstration)
# In production, use Redis or a database
_sessions: dict[str, dict[str, Any]] = {}

# Session timeout (30 minutes)
SESSION_TIMEOUT = 1800


class SessionStore:
    """Manages conversation sessions (short-term memory)."""

    @staticmethod
    def create_session(user_id: str | None = None) -> str:
        """Create a new session.

        Args:
            user_id: Optional user ID for long-term memory association

        Returns:
            Session ID
        """
        session_id = str(uuid.uuid4())
        _sessions[session_id] = {
            "session_id": session_id,
            "user_id": user_id,
            "created_at": time.time(),
            "last_accessed": time.time(),
            "conversation_history": [],
            "extracted_context": {},
            "completion_percentage": 0,
            "ready_for_recommendation": False,
        }
        logger.info("session_created", session_id=session_id, user_id=user_id)
        return session_id

    @staticmethod
    def get_session(session_id: str) -> dict[str, Any] | None:
        """Get session data.

        Args:
            session_id: Session ID

        Returns:
            Session data or None if not found/expired
        """
        session = _sessions.get(session_id)
        if not session:
            return None

        # Check if session expired
        if time.time() - session["last_accessed"] > SESSION_TIMEOUT:
            logger.info("session_expired", session_id=session_id)
            del _sessions[session_id]
            return None

        # Update last accessed time
        session["last_accessed"] = time.time()
        return session

    @staticmethod
    def update_session(session_id: str, updates: dict[str, Any]) -> bool:
        """Update session data.

        Args:
            session_id: Session ID
            updates: Dictionary of fields to update

        Returns:
            True if successful, False if session not found
        """
        session = SessionStore.get_session(session_id)
        if not session:
            return False

        session.update(updates)
        session["last_accessed"] = time.time()
        logger.info("session_updated", session_id=session_id)
        return True

    @staticmethod
    def add_message(session_id: str, role: str, content: str) -> bool:
        """Add a message to conversation history.

        Args:
            session_id: Session ID
            role: Message role (assistant/user)
            content: Message content

        Returns:
            True if successful
        """
        session = SessionStore.get_session(session_id)
        if not session:
            return False

        session["conversation_history"].append({
            "role": role,
            "content": content,
            "timestamp": time.time()
        })
        session["last_accessed"] = time.time()
        return True

    @staticmethod
    def delete_session(session_id: str) -> bool:
        """Delete a session.

        Args:
            session_id: Session ID

        Returns:
            True if deleted, False if not found
        """
        if session_id in _sessions:
            del _sessions[session_id]
            logger.info("session_deleted", session_id=session_id)
            return True
        return False

    @staticmethod
    def cleanup_expired_sessions() -> int:
        """Remove expired sessions.

        Returns:
            Number of sessions cleaned up
        """
        current_time = time.time()
        expired = [
            sid for sid, session in _sessions.items()
            if current_time - session["last_accessed"] > SESSION_TIMEOUT
        ]

        for sid in expired:
            del _sessions[sid]

        if expired:
            logger.info("sessions_cleaned_up", count=len(expired))

        return len(expired)

    @staticmethod
    def get_active_session_count() -> int:
        """Get number of active sessions.

        Returns:
            Number of active sessions
        """
        return len(_sessions)