engine/repositories/play_session.py

"""플레이 세션 영속화 — SQLite (계획: docs/plans/persistence.md).

- 권위: CLAUDE.md "현재 유저의 서사 위치 권위: DB play_session.current_state_id"의 MVP 구현.
  payload(JSON)에 StoryEngine.snapshot() 전체를 담는다 (story 위치 + 숨은 상태)
- 콘텐츠 버전 불일치 세션은 load에서 None (무효화 — 새 세션 시작 유도)
- 포켓(채팅) 중간 상태는 저장하지 않는다 (유저 결정 — 복원 시 포켓 폐기)
"""

from __future__ import annotations

import json
import sqlite3
from contextlib import closing
from datetime import datetime, timedelta, timezone
from pathlib import Path

_SCHEMA = """
CREATE TABLE IF NOT EXISTS play_session (
    sid             TEXT PRIMARY KEY,
    payload         TEXT NOT NULL,
    content_version TEXT NOT NULL,
    updated_at      TEXT NOT NULL
)
"""


class PlaySessionRepository:
    def __init__(self, db_path: str | Path):
        self.db_path = Path(db_path)
        self.db_path.parent.mkdir(parents=True, exist_ok=True)
        with closing(self._conn()) as conn, conn:
            conn.execute(_SCHEMA)

    def _conn(self) -> sqlite3.Connection:
        conn = sqlite3.connect(self.db_path)
        conn.execute("PRAGMA journal_mode=WAL")
        return conn

    def save(self, sid: str, payload: dict, content_version: str) -> None:
        with closing(self._conn()) as conn, conn:
            conn.execute(
                "INSERT INTO play_session (sid, payload, content_version, updated_at) "
                "VALUES (?, ?, ?, ?) "
                "ON CONFLICT(sid) DO UPDATE SET payload=excluded.payload, "
                "content_version=excluded.content_version, updated_at=excluded.updated_at",
                (sid, json.dumps(payload, ensure_ascii=False), content_version,
                 datetime.now(timezone.utc).isoformat()))

    def load(self, sid: str, content_version: str) -> dict | None:
        """스냅샷 로드. 미존재 또는 콘텐츠 버전 불일치면 None."""
        with closing(self._conn()) as conn:
            row = conn.execute(
                "SELECT payload, content_version FROM play_session WHERE sid=?",
                (sid,)).fetchone()
        if row is None or row[1] != content_version:
            return None
        return json.loads(row[0])

    def purge(self, days: int = 30) -> int:
        """N일 미갱신 세션 삭제. 삭제 건수 반환."""
        cutoff = (datetime.now(timezone.utc) - timedelta(days=days)).isoformat()
        with closing(self._conn()) as conn, conn:
            cur = conn.execute(
                "DELETE FROM play_session WHERE updated_at < ?", (cutoff,))
            return cur.rowcount