dotfiles/python/gems/persistence.py

"""SQLite persistence for rooms, memberships, snapshots, and events."""

from __future__ import annotations

import json
import sqlite3
import threading
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from typing import TYPE_CHECKING

from .domain.models import ContentPack, GameSettings, GameState

if TYPE_CHECKING:
    from pathlib import Path


@dataclass(slots=True)
class Member:
    """Represent a human or AI seat persisted for a room."""

    room_code: str
    seat: int
    name: str
    controller: str
    difficulty: str | None
    credential_hash: str | None
    is_host: bool
    ready: bool
    last_seen: str


@dataclass(slots=True)
class Room:
    """Represent persisted room metadata and its current snapshot."""

    code: str
    status: str
    settings: GameSettings
    pack: ContentPack | None
    pack_digest: str | None
    state: GameState | None
    revision: int
    created_at: str
    updated_at: str


class Repository:
    """Small transactional repository designed for one application worker."""

    def __init__(self, path: Path) -> None:
        """Open the SQLite database and initialize its schema."""
        path.parent.mkdir(parents=True, exist_ok=True)
        self._connection = sqlite3.connect(path, check_same_thread=False)
        self._connection.row_factory = sqlite3.Row
        self._lock = threading.RLock()
        with self._connection:
            self._connection.execute("PRAGMA journal_mode=WAL")
            self._connection.execute("PRAGMA foreign_keys=ON")
            self._connection.execute("PRAGMA busy_timeout=5000")
            self._connection.executescript(
                """
                CREATE TABLE IF NOT EXISTS schema_version(version INTEGER NOT NULL);
                INSERT INTO schema_version(version) SELECT 1 WHERE NOT EXISTS (SELECT 1 FROM schema_version);
                CREATE TABLE IF NOT EXISTS rooms(
                    code TEXT PRIMARY KEY, status TEXT NOT NULL, settings_json TEXT NOT NULL,
                    pack_json TEXT, pack_digest TEXT, state_json TEXT, revision INTEGER NOT NULL,
                    created_at TEXT NOT NULL, updated_at TEXT NOT NULL
                );
                CREATE TABLE IF NOT EXISTS members(
                    room_code TEXT NOT NULL REFERENCES rooms(code) ON DELETE CASCADE,
                    seat INTEGER NOT NULL, name TEXT NOT NULL, controller TEXT NOT NULL,
                    difficulty TEXT, credential_hash TEXT, is_host INTEGER NOT NULL,
                    ready INTEGER NOT NULL, last_seen TEXT NOT NULL,
                    PRIMARY KEY(room_code, seat), UNIQUE(room_code, credential_hash)
                );
                CREATE TABLE IF NOT EXISTS room_events(
                    room_code TEXT NOT NULL REFERENCES rooms(code) ON DELETE CASCADE,
                    revision INTEGER NOT NULL, command_id TEXT NOT NULL, actor_seat INTEGER,
                    event_type TEXT NOT NULL, payload_json TEXT NOT NULL, created_at TEXT NOT NULL,
                    PRIMARY KEY(room_code, command_id)
                );
                """
            )

    def room_exists(self, code: str) -> bool:
        """Report whether a room code exists."""
        return self._connection.execute("SELECT 1 FROM rooms WHERE code=?", (code,)).fetchone() is not None

    def create_room(self, room: Room, host: Member) -> None:
        """Persist a new room and its host in one transaction."""
        with self._lock, self._connection:
            self._connection.execute(
                "INSERT INTO rooms VALUES(?,?,?,?,?,?,?,?,?)",
                (
                    room.code,
                    room.status,
                    room.settings.model_dump_json(),
                    None,
                    None,
                    None,
                    room.revision,
                    room.created_at,
                    room.updated_at,
                ),
            )
            self._insert_member(host)

    def _insert_member(self, member: Member) -> None:
        self._connection.execute(
            "INSERT INTO members VALUES(?,?,?,?,?,?,?,?,?)",
            (
                member.room_code,
                member.seat,
                member.name,
                member.controller,
                member.difficulty,
                member.credential_hash,
                int(member.is_host),
                int(member.ready),
                member.last_seen,
            ),
        )

    def add_member(self, member: Member) -> None:
        """Add a member and refresh the room's activity timestamp."""
        with self._lock, self._connection:
            self._insert_member(member)
            self.touch(member.room_code)

    def update_member(self, member: Member) -> None:
        """Persist all mutable fields for an existing member."""
        with self._lock, self._connection:
            self._connection.execute(
                """UPDATE members SET name=?,controller=?,difficulty=?,credential_hash=?,is_host=?,ready=?,last_seen=?
                   WHERE room_code=? AND seat=?""",
                (
                    member.name,
                    member.controller,
                    member.difficulty,
                    member.credential_hash,
                    int(member.is_host),
                    int(member.ready),
                    member.last_seen,
                    member.room_code,
                    member.seat,
                ),
            )

    def remove_member(self, code: str, seat: int) -> None:
        """Remove a member from a room seat."""
        with self._lock, self._connection:
            self._connection.execute("DELETE FROM members WHERE room_code=? AND seat=?", (code, seat))

    def get_room(self, code: str) -> Room | None:
        """Load a room by code, including its typed JSON fields."""
        row = self._connection.execute("SELECT * FROM rooms WHERE code=?", (code.upper(),)).fetchone()
        if row is None:
            return None
        return Room(
            code=row["code"],
            status=row["status"],
            settings=GameSettings.model_validate_json(row["settings_json"]),
            pack=ContentPack.model_validate_json(row["pack_json"]) if row["pack_json"] else None,
            pack_digest=row["pack_digest"],
            state=GameState.model_validate_json(row["state_json"]) if row["state_json"] else None,
            revision=row["revision"],
            created_at=row["created_at"],
            updated_at=row["updated_at"],
        )

    def members(self, code: str) -> list[Member]:
        """Load all room members ordered by seat."""
        rows = self._connection.execute("SELECT * FROM members WHERE room_code=? ORDER BY seat", (code,)).fetchall()
        return [
            Member(
                room_code=row["room_code"],
                seat=row["seat"],
                name=row["name"],
                controller=row["controller"],
                difficulty=row["difficulty"],
                credential_hash=row["credential_hash"],
                is_host=bool(row["is_host"]),
                ready=bool(row["ready"]),
                last_seen=row["last_seen"],
            )
            for row in rows
        ]

    def member_for_credential(self, code: str, credential_hash: str) -> Member | None:
        """Find the room member associated with a credential digest."""
        return next((member for member in self.members(code) if member.credential_hash == credential_hash), None)

    def save_lobby(self, room: Room) -> None:
        """Persist mutable lobby configuration and revision data."""
        with self._lock, self._connection:
            self._connection.execute(
                """UPDATE rooms SET settings_json=?,pack_json=?,pack_digest=?,revision=?,updated_at=? WHERE code=?""",
                (
                    room.settings.model_dump_json(),
                    room.pack.model_dump_json() if room.pack else None,
                    room.pack_digest,
                    room.revision,
                    room.updated_at,
                    room.code,
                ),
            )

    def save_state(self, room: Room, command_id: str, actor_seat: int | None, payload: dict) -> bool:
        """Persist a game snapshot and its idempotent command event."""
        now = utc_now()
        with self._lock, self._connection:
            existing = self._connection.execute(
                "SELECT 1 FROM room_events WHERE room_code=? AND command_id=?", (room.code, command_id)
            ).fetchone()
            if existing:
                return False
            self._connection.execute(
                "UPDATE rooms SET status=?,state_json=?,revision=?,updated_at=? WHERE code=?",
                (room.status, room.state.model_dump_json() if room.state else None, room.revision, now, room.code),
            )
            self._connection.execute(
                "INSERT INTO room_events VALUES(?,?,?,?,?,?,?)",
                (room.code, room.revision, command_id, actor_seat, "command", json.dumps(payload), now),
            )
        return True

    def has_command(self, code: str, command_id: str) -> bool:
        """Report whether a command was already persisted for a room."""
        row = self._connection.execute(
            "SELECT 1 FROM room_events WHERE room_code=? AND command_id=?",
            (code, command_id),
        ).fetchone()
        return row is not None

    def touch(self, code: str) -> None:
        """Refresh a room's activity timestamp."""
        with self._connection:
            self._connection.execute("UPDATE rooms SET updated_at=? WHERE code=?", (utc_now(), code))

    def cleanup(self, days: int = 30) -> int:
        """Delete rooms inactive for the requested number of days."""
        cutoff = (datetime.now(UTC) - timedelta(days=days)).isoformat()
        with self._lock, self._connection:
            cursor = self._connection.execute("DELETE FROM rooms WHERE updated_at < ?", (cutoff,))
        return cursor.rowcount

    def close(self) -> None:
        """Close the underlying SQLite connection."""
        self._connection.close()


def utc_now() -> str:
    """Return the current UTC time as an ISO 8601 string."""
    return datetime.now(UTC).isoformat()