vacuum-wall/lib/db_sqlite.py

"""SQLite backend for Vacuum Wall database.

Concrete implementation of the Database abstract class using SQLite3.
Uses Python 3.13+ sqlite3.Statement for prepared statements.
"""

from __future__ import annotations

import contextlib
import logging
import sqlite3
from typing import Any, ClassVar

from lib.db import (
    Q_DELETE_EXPIRED_BLACKLIST,
    Q_DELETE_PERMISSIONS,
    Q_DELETE_REFRESH_TOKEN,
    Q_DELETE_USER,
    Q_DELETE_WEBAUTHN,
    Q_INSERT_BLACKLIST,
    Q_INSERT_USER,
    Q_INSERT_WEBAUTHN,
    Q_SELECT_ALL_USERS,
    Q_SELECT_BLACKLIST,
    Q_SELECT_PERMISSIONS,
    Q_SELECT_REFRESH_TOKEN,
    Q_SELECT_USER_BY_ID,
    Q_SELECT_USER_BY_NAME,
    Q_SELECT_USER_JWT_SECRET,
    Q_SELECT_USERS_WITH_PERMS,
    Q_SELECT_WEBAUTHN_COUNTS,
    Q_SELECT_WEBAUTHN_ID,
    Q_SELECT_WEBAUTHN_USER,
    Q_UPDATE_JWT_SECRET,
    Q_UPDATE_PASSWORD,
    Q_UPDATE_WEBAUTHN_SIGN_COUNT,
    Q_UPSERT_PERMISSION,
    Q_UPSERT_REFRESH_TOKEN,
    Database,
)

logger = logging.getLogger(__name__)


class SQLiteBackend(Database):
    """SQLite implementation of the Database interface.

    Uses ``sqlite3.Connection.execute()`` for statement execution and
    ``sqlite3.Row`` for row-factory dict access.
    """

    def __init__(self, connection_string: str) -> None:
        super().__init__(connection_string)
        self._last_rowid: int = 0

    QUERY_MAP: ClassVar[dict[str, str]] = {
        # Schema init is handled by direct execution, not prepared statements
        # init_tables is called as _execute_direct(INIT_SQL)
        # Users
        Q_INSERT_USER: (
            "INSERT INTO users (username, password_hash, jwt_secret) VALUES (?, ?, ?)"
        ),
        Q_SELECT_USER_BY_NAME: (
            "SELECT id, username, password_hash, jwt_secret, created_at FROM users WHERE username = ?"
        ),
        Q_SELECT_USER_BY_ID: (
            "SELECT id, username, password_hash, jwt_secret, created_at FROM users WHERE id = ?"
        ),
        Q_UPDATE_PASSWORD: "UPDATE users SET password_hash = ? WHERE username = ?",
        Q_DELETE_USER: "DELETE FROM users WHERE username = ?",
        Q_SELECT_ALL_USERS: (
            "SELECT id, username, created_at FROM users ORDER BY username"
        ),
        Q_SELECT_USERS_WITH_PERMS: (
            "SELECT u.id, u.username, u.created_at, p.subsystem, p.level "
            "FROM users u LEFT JOIN permissions p ON u.username = p.username "
            "ORDER BY u.username"
        ),
        Q_SELECT_USER_JWT_SECRET: ("SELECT jwt_secret FROM users WHERE username = ?"),
        Q_UPDATE_JWT_SECRET: ("UPDATE users SET jwt_secret = ? WHERE username = ?"),
        # Permissions
        Q_UPSERT_PERMISSION: (
            "INSERT INTO permissions (username, subsystem, level) "
            "VALUES (?, ?, ?) "
            "ON CONFLICT(username, subsystem) DO UPDATE SET level = excluded.level"
        ),
        Q_SELECT_PERMISSIONS: (
            "SELECT subsystem, level FROM permissions WHERE username = ?"
        ),
        Q_DELETE_PERMISSIONS: ("DELETE FROM permissions WHERE username = ?"),
        # Token blacklist
        Q_INSERT_BLACKLIST: (
            "INSERT OR IGNORE INTO token_blacklist (jti, token_type, expires) VALUES (?, ?, ?)"
        ),
        Q_SELECT_BLACKLIST: "SELECT jti FROM token_blacklist WHERE jti = ?",
        Q_DELETE_EXPIRED_BLACKLIST: "DELETE FROM token_blacklist WHERE expires < ?",
        # Refresh tokens
        Q_UPSERT_REFRESH_TOKEN: (
            "INSERT INTO refresh_tokens (username, jti, issued_at) "
            "VALUES (?, ?, ?) "
            "ON CONFLICT(username) DO UPDATE SET jti = excluded.jti, issued_at = excluded.issued_at"
        ),
        Q_SELECT_REFRESH_TOKEN: "SELECT username, jti, issued_at FROM refresh_tokens WHERE username = ?",
        Q_DELETE_REFRESH_TOKEN: "DELETE FROM refresh_tokens WHERE username = ?",
        # WebAuthn
        Q_INSERT_WEBAUTHN: (
            "INSERT INTO webauthn_creds "
            "(username, credential_id, public_key, sign_count, name, transports) "
            "VALUES (?, ?, ?, ?, ?, ?)"
        ),
        Q_SELECT_WEBAUTHN_USER: (
            "SELECT id, credential_id, public_key, sign_count, name, transports "
            "FROM webauthn_creds WHERE username = ?"
        ),
        Q_SELECT_WEBAUTHN_ID: (
            "SELECT id, username, credential_id, public_key, sign_count, name, transports "
            "FROM webauthn_creds WHERE credential_id = ?"
        ),
        Q_SELECT_WEBAUTHN_COUNTS: (
            "SELECT username, COUNT(*) as cred_count FROM webauthn_creds "
            "GROUP BY username"
        ),
        Q_UPDATE_WEBAUTHN_SIGN_COUNT: (
            "UPDATE webauthn_creds SET sign_count = ? WHERE credential_id = ?"
        ),
        Q_DELETE_WEBAUTHN: "DELETE FROM webauthn_creds WHERE credential_id = ?",
    }

    def _connect(self, cs: str) -> Any:
        """Create a SQLite connection with WAL mode and row factory."""
        conn = sqlite3.connect(cs, isolation_level=None)
        conn.execute("PRAGMA journal_mode=WAL")
        conn.execute("PRAGMA foreign_keys=ON")
        conn.row_factory = sqlite3.Row
        return conn

    def _prepare(self, sql: str) -> Any:
        """Store the SQL string for later execution.

        SQLite in-memory DB doesn't support the Python 3.13 conn.prepare()
        API, so we store the raw SQL and execute via conn.execute().
        """
        return sql

    def _execute(self, stmt: Any, params: tuple) -> tuple[list[dict], int]:
        """Execute a prepared statement, returning (rows, rowcount)."""
        cursor = self.conn.execute(stmt, params)
        self._last_rowid = cursor.lastrowid
        result = cursor.fetchall()
        rows: list[dict] = []
        for row in result:
            rows.append(dict(row))
        return rows, cursor.rowcount

    def _last_insert_id(self, stmt: Any) -> int | dict:
        """Return the last insert row ID from the most recent execute."""
        return self._last_rowid

    def _begin(self) -> None:
        self.conn.execute("BEGIN")

    def _commit(self) -> None:
        self.conn.execute("COMMIT")

    def _rollback(self) -> None:
        with contextlib.suppress(sqlite3.Error):
            self.conn.execute("ROLLBACK")

    def _suppress_auto_commit(self) -> None:
        self._in_transaction = True

    def _restore_auto_commit(self) -> None:
        self._in_transaction = False

    def _execute_direct(self, sql: str) -> None:
        """Execute raw SQL without prepared statements (for DDL)."""
        self.conn.executescript(sql)