backend/session_store.py

import shutil
import time
import json
from pathlib import Path
from typing import Any, Dict, List, Optional

from backend.types import SessionRecord


class SessionStore:
    def __init__(self, root: Path, ttl_seconds: int) -> None:
        self.root = Path(root)
        self.ttl_seconds = ttl_seconds
        self.root.mkdir(parents=True, exist_ok=True)
        self._sessions: Dict[str, SessionRecord] = {}

    def _now(self) -> float:
        return time.time()

    def ensure_session(self, session_id: str) -> SessionRecord:
        now = self._now()
        session_root = self.root / session_id
        session_root.mkdir(parents=True, exist_ok=True)
        for child in ("uploads", "previews", "renders", "exports"):
            (session_root / child).mkdir(exist_ok=True)
        existing = self._sessions.get(session_id)
        if existing is None:
            existing = SessionRecord(
                session_id=session_id,
                root=session_root,
                created_at=now,
                touched_at=now,
            )
            self._sessions[session_id] = existing
        else:
            existing.touched_at = now
        return existing

    def touch(self, session_id: str) -> SessionRecord:
        session = self.ensure_session(session_id)
        session.touched_at = self._now()
        return session

    def save_json(self, session_id: str, name: str, payload: Dict[str, Any]) -> Path:
        session = self.touch(session_id)
        path = session.root / name
        path.write_text(json.dumps(payload, ensure_ascii=True, indent=2), encoding="utf-8")
        return path

    def load_json(self, session_id: str, name: str) -> Dict[str, Any]:
        session = self.touch(session_id)
        path = session.root / name
        if not path.exists():
            raise FileNotFoundError(f"Session asset not found: {name}")
        return json.loads(path.read_text(encoding="utf-8"))

    def cleanup_expired(self, now: Optional[float] = None) -> List[str]:
        current = self._now() if now is None else now
        removed: List[str] = []
        for session_id, record in list(self._sessions.items()):
            if current - record.touched_at <= self.ttl_seconds:
                continue
            if record.root.exists():
                shutil.rmtree(record.root)
            removed.append(session_id)
            del self._sessions[session_id]
        return removed