mud/src/mudlib/store/__init__.py

"""SQLite persistence for player accounts and state."""

import hashlib
import hmac
import os
import sqlite3
from pathlib import Path
from typing import TypedDict

from mudlib.player import Player


class PlayerData(TypedDict):
    """Shape of persisted player data from the database."""

    x: int
    y: int
    pl: float
    stamina: float
    max_stamina: float
    flying: bool


# Module-level database path
_db_path: str | None = None


def init_db(db_path: str | Path) -> None:
    """Initialize the database and create tables if needed.

    Args:
        db_path: Path to the SQLite database file
    """
    global _db_path
    _db_path = str(db_path)

    # Ensure parent directory exists
    Path(_db_path).parent.mkdir(parents=True, exist_ok=True)

    conn = sqlite3.connect(_db_path)
    cursor = conn.cursor()

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS accounts (
            name TEXT PRIMARY KEY COLLATE NOCASE,
            password_hash TEXT NOT NULL,
            salt TEXT NOT NULL,
            x INTEGER NOT NULL DEFAULT 0,
            y INTEGER NOT NULL DEFAULT 0,
            pl REAL NOT NULL DEFAULT 100.0,
            stamina REAL NOT NULL DEFAULT 100.0,
            max_stamina REAL NOT NULL DEFAULT 100.0,
            flying INTEGER NOT NULL DEFAULT 0,
            created_at TEXT NOT NULL DEFAULT (datetime('now')),
            last_login TEXT
        )
    """)

    conn.commit()
    conn.close()


def _get_connection() -> sqlite3.Connection:
    """Get a connection to the database.

    Returns:
        sqlite3.Connection

    Raises:
        RuntimeError: If init_db has not been called
    """
    if _db_path is None:
        raise RuntimeError("Database not initialized. Call init_db() first.")
    return sqlite3.connect(_db_path)


def _hash_password(password: str, salt: bytes) -> str:
    """Hash a password with the given salt.

    Args:
        password: The password to hash
        salt: Salt bytes

    Returns:
        Hex-encoded hash
    """
    return hashlib.pbkdf2_hmac(
        "sha256", password.encode("utf-8"), salt, iterations=100000
    ).hex()


def create_account(name: str, password: str) -> bool:
    """Create a new account with the given name and password.

    Args:
        name: Account name (case-insensitive)
        password: Password to hash and store

    Returns:
        True if account was created, False if name is already taken
    """
    if account_exists(name):
        return False

    # Generate random salt
    salt = os.urandom(32)
    password_hash = _hash_password(password, salt)

    conn = _get_connection()
    cursor = conn.cursor()

    try:
        cursor.execute(
            "INSERT INTO accounts (name, password_hash, salt) VALUES (?, ?, ?)",
            (name, password_hash, salt.hex()),
        )
        conn.commit()
        return True
    except sqlite3.IntegrityError:
        # Race condition: someone created the account between our check and insert
        return False
    finally:
        conn.close()


def account_exists(name: str) -> bool:
    """Check if an account exists.

    Args:
        name: Account name (case-insensitive)

    Returns:
        True if account exists, False otherwise
    """
    conn = _get_connection()
    cursor = conn.cursor()

    cursor.execute("SELECT 1 FROM accounts WHERE name = ? LIMIT 1", (name,))
    result = cursor.fetchone()

    conn.close()
    return result is not None


def authenticate(name: str, password: str) -> bool:
    """Verify a password for the given account.

    Args:
        name: Account name (case-insensitive)
        password: Password to verify

    Returns:
        True if password is correct, False otherwise
    """
    conn = _get_connection()
    cursor = conn.cursor()

    cursor.execute("SELECT password_hash, salt FROM accounts WHERE name = ?", (name,))
    result = cursor.fetchone()
    conn.close()

    if result is None:
        return False

    stored_hash, salt_hex = result
    salt = bytes.fromhex(salt_hex)
    password_hash = _hash_password(password, salt)

    return hmac.compare_digest(password_hash, stored_hash)


def save_player(player: Player) -> None:
    """Save player state to the database.

    The account must already exist. This updates game state only.

    Args:
        player: Player instance to save
    """
    conn = _get_connection()
    cursor = conn.cursor()

    cursor.execute(
        """
        UPDATE accounts
        SET x = ?, y = ?, pl = ?, stamina = ?, max_stamina = ?, flying = ?
        WHERE name = ?
        """,
        (
            player.x,
            player.y,
            player.pl,
            player.stamina,
            player.max_stamina,
            1 if player.flying else 0,
            player.name,
        ),
    )

    conn.commit()
    conn.close()


def load_player_data(name: str) -> PlayerData | None:
    """Load player data from the database.

    Args:
        name: Account name (case-insensitive)

    Returns:
        Dictionary of persisted fields, or None if account not found
    """
    conn = _get_connection()
    cursor = conn.cursor()

    cursor.execute(
        """
        SELECT x, y, pl, stamina, max_stamina, flying
        FROM accounts
        WHERE name = ?
        """,
        (name,),
    )
    result = cursor.fetchone()
    conn.close()

    if result is None:
        return None

    x, y, pl, stamina, max_stamina, flying_int = result
    return {
        "x": x,
        "y": y,
        "pl": pl,
        "stamina": stamina,
        "max_stamina": max_stamina,
        "flying": bool(flying_int),
    }


def update_last_login(name: str) -> None:
    """Update the last_login timestamp for an account.

    Args:
        name: Account name (case-insensitive)
    """
    conn = _get_connection()
    cursor = conn.cursor()

    cursor.execute(
        "UPDATE accounts SET last_login = datetime('now') WHERE name = ?", (name,)
    )

    conn.commit()
    conn.close()