|
|
""" |
|
|
Session Store Service |
|
|
|
|
|
Thread-safe session-scoped storage for user layers and context. |
|
|
Replaces global SESSION_LAYERS with per-session isolation. |
|
|
""" |
|
|
|
|
|
import logging |
|
|
import threading |
|
|
from datetime import datetime, timedelta |
|
|
from typing import Dict, List, Optional, Any |
|
|
|
|
|
|
|
|
logger = logging.getLogger(__name__) |
|
|
|
|
|
|
|
|
class SessionStore: |
|
|
""" |
|
|
Thread-safe session-scoped storage with TTL expiration. |
|
|
|
|
|
Each session maintains its own: |
|
|
- layers: Map layers created by the user |
|
|
- context: Optional conversation context |
|
|
|
|
|
Sessions expire after configurable TTL (default 2 hours). |
|
|
""" |
|
|
|
|
|
_instance = None |
|
|
|
|
|
def __new__(cls): |
|
|
if cls._instance is None: |
|
|
cls._instance = super(SessionStore, cls).__new__(cls) |
|
|
cls._instance.initialized = False |
|
|
return cls._instance |
|
|
|
|
|
def __init__(self, ttl_hours: int = 2, max_layers_per_session: int = 15): |
|
|
if self.initialized: |
|
|
return |
|
|
|
|
|
self._sessions: Dict[str, dict] = {} |
|
|
self._lock = threading.Lock() |
|
|
self.ttl = timedelta(hours=ttl_hours) |
|
|
self.max_layers = max_layers_per_session |
|
|
self.initialized = True |
|
|
|
|
|
logger.info(f"SessionStore initialized with TTL={ttl_hours}h, max_layers={max_layers_per_session}") |
|
|
|
|
|
def _get_or_create_session(self, session_id: str) -> dict: |
|
|
"""Get existing session or create new one.""" |
|
|
if session_id not in self._sessions: |
|
|
self._sessions[session_id] = { |
|
|
"layers": [], |
|
|
"created": datetime.now(), |
|
|
"accessed": datetime.now() |
|
|
} |
|
|
return self._sessions[session_id] |
|
|
|
|
|
def get_layers(self, session_id: str) -> List[dict]: |
|
|
"""Get all layers for a session.""" |
|
|
with self._lock: |
|
|
session = self._get_or_create_session(session_id) |
|
|
session["accessed"] = datetime.now() |
|
|
return session["layers"].copy() |
|
|
|
|
|
def add_layer(self, session_id: str, layer: dict) -> None: |
|
|
""" |
|
|
Add a layer to a session. |
|
|
|
|
|
Enforces max_layers limit by removing oldest layers. |
|
|
""" |
|
|
with self._lock: |
|
|
session = self._get_or_create_session(session_id) |
|
|
session["layers"].append(layer) |
|
|
session["accessed"] = datetime.now() |
|
|
|
|
|
|
|
|
while len(session["layers"]) > self.max_layers: |
|
|
removed = session["layers"].pop(0) |
|
|
logger.debug(f"Session {session_id[:8]}: removed oldest layer {removed.get('name', 'unknown')}") |
|
|
|
|
|
def update_layer(self, session_id: str, layer_id: str, updates: dict) -> bool: |
|
|
""" |
|
|
Update an existing layer in a session. |
|
|
|
|
|
Returns True if layer was found and updated. |
|
|
""" |
|
|
with self._lock: |
|
|
session = self._sessions.get(session_id) |
|
|
if not session: |
|
|
return False |
|
|
|
|
|
for layer in session["layers"]: |
|
|
if layer.get("id") == layer_id: |
|
|
layer.update(updates) |
|
|
session["accessed"] = datetime.now() |
|
|
return True |
|
|
|
|
|
return False |
|
|
|
|
|
def remove_layer(self, session_id: str, layer_id: str) -> bool: |
|
|
""" |
|
|
Remove a layer from a session. |
|
|
|
|
|
Returns True if layer was found and removed. |
|
|
""" |
|
|
with self._lock: |
|
|
session = self._sessions.get(session_id) |
|
|
if not session: |
|
|
return False |
|
|
|
|
|
original_len = len(session["layers"]) |
|
|
session["layers"] = [l for l in session["layers"] if l.get("id") != layer_id] |
|
|
session["accessed"] = datetime.now() |
|
|
|
|
|
return len(session["layers"]) < original_len |
|
|
|
|
|
def clear_session(self, session_id: str) -> None: |
|
|
"""Clear all data for a session.""" |
|
|
with self._lock: |
|
|
if session_id in self._sessions: |
|
|
del self._sessions[session_id] |
|
|
|
|
|
def get_layer_by_index(self, session_id: str, index: int) -> Optional[dict]: |
|
|
"""Get a specific layer by 1-based index (for user references like 'Layer 1').""" |
|
|
with self._lock: |
|
|
session = self._sessions.get(session_id) |
|
|
if not session: |
|
|
return None |
|
|
|
|
|
layers = session["layers"] |
|
|
if 1 <= index <= len(layers): |
|
|
return layers[index - 1].copy() |
|
|
|
|
|
return None |
|
|
|
|
|
def cleanup_expired(self) -> int: |
|
|
""" |
|
|
Remove sessions older than TTL. |
|
|
|
|
|
Returns number of expired sessions removed. |
|
|
""" |
|
|
with self._lock: |
|
|
now = datetime.now() |
|
|
expired = [ |
|
|
sid for sid, data in self._sessions.items() |
|
|
if now - data.get("accessed", data["created"]) > self.ttl |
|
|
] |
|
|
|
|
|
for sid in expired: |
|
|
del self._sessions[sid] |
|
|
|
|
|
if expired: |
|
|
logger.info(f"Cleaned up {len(expired)} expired sessions.") |
|
|
|
|
|
return len(expired) |
|
|
|
|
|
def get_stats(self) -> dict: |
|
|
"""Return statistics about active sessions.""" |
|
|
with self._lock: |
|
|
total_layers = sum(len(s["layers"]) for s in self._sessions.values()) |
|
|
|
|
|
return { |
|
|
"active_sessions": len(self._sessions), |
|
|
"total_layers": total_layers, |
|
|
"ttl_hours": self.ttl.total_seconds() / 3600, |
|
|
"max_layers_per_session": self.max_layers |
|
|
} |
|
|
|
|
|
|
|
|
|
|
|
_session_store: Optional[SessionStore] = None |
|
|
|
|
|
|
|
|
def get_session_store() -> SessionStore: |
|
|
"""Get the singleton session store instance.""" |
|
|
global _session_store |
|
|
if _session_store is None: |
|
|
_session_store = SessionStore() |
|
|
return _session_store |
|
|
|
