my_voice_assistant_v3/app/store.py

"""Persistenzschicht: Nutzer und Sessions.

Ein abstraktes Store-Interface mit SQLite-Default (stdlib). Spaetere Backends
(Postgres/Redis) koennen dasselbe Interface implementieren, ohne die App zu aendern.
"""

from __future__ import annotations

import json
import hashlib
import secrets
import sqlite3
import uuid
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path

ANONYMOUS_USER_ID = "anonymous"


def hash_token(raw_token: str) -> str:
    return hashlib.sha256(raw_token.encode("utf-8")).hexdigest()


def _now() -> str:
    return datetime.now(timezone.utc).isoformat()


@dataclass
class User:
    id: str
    display_name: str
    prefs: dict = field(default_factory=dict)
    created_at: str = ""


@dataclass
class Session:
    id: str
    user_id: str
    data: dict = field(default_factory=dict)


class SessionOwnershipError(Exception):
    """Eine Session gehoert einem anderen Nutzer (-> HTTP 403)."""


class Store(ABC):
    @abstractmethod
    def create_user(self, display_name: str) -> tuple[User, str]:
        """Legt einen Nutzer an und liefert (User, Klartext-Token). Token nur hier sichtbar."""

    @abstractmethod
    def get_user_by_token(self, raw_token: str) -> User | None: ...

    @abstractmethod
    def get_user(self, user_id: str) -> User | None: ...

    @abstractmethod
    def set_user_prefs(self, user_id: str, prefs: dict) -> User: ...

    @abstractmethod
    def ensure_anonymous_user(self) -> User: ...

    @abstractmethod
    def get_session(self, session_id: str) -> Session | None: ...

    @abstractmethod
    def update_session(self, session_id: str, user_id: str, values: dict) -> Session:
        """Erstellt/aktualisiert eine Session des Nutzers. Fremde Session -> SessionOwnershipError."""

    @abstractmethod
    def append_message(self, session_id: str, user_id: str, role: str, content: str) -> None:
        """Haengt eine Nachricht an die Session an. Fremde Session -> SessionOwnershipError."""

    @abstractmethod
    def get_recent_messages(self, session_id: str, limit: int) -> list[dict]:
        """Liefert die letzten `limit` Nachrichten chronologisch ([{'role','content'}, ...])."""


class SQLiteStore(Store):
    def __init__(self, db_path: str):
        self.db_path = db_path
        Path(db_path).parent.mkdir(parents=True, exist_ok=True)
        self._init_schema()

    def _connect(self) -> sqlite3.Connection:
        conn = sqlite3.connect(self.db_path)
        conn.row_factory = sqlite3.Row
        conn.execute("PRAGMA journal_mode=WAL")
        conn.execute("PRAGMA foreign_keys=ON")
        return conn

    def _init_schema(self) -> None:
        with self._connect() as conn:
            conn.executescript(
                """
                CREATE TABLE IF NOT EXISTS users (
                    id TEXT PRIMARY KEY,
                    display_name TEXT NOT NULL,
                    token_hash TEXT NOT NULL UNIQUE,
                    prefs_json TEXT NOT NULL DEFAULT '{}',
                    created_at TEXT NOT NULL
                );
                CREATE TABLE IF NOT EXISTS sessions (
                    id TEXT PRIMARY KEY,
                    user_id TEXT NOT NULL,
                    data_json TEXT NOT NULL DEFAULT '{}',
                    created_at TEXT NOT NULL,
                    updated_at TEXT NOT NULL
                );
                CREATE TABLE IF NOT EXISTS messages (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    session_id TEXT NOT NULL,
                    role TEXT NOT NULL,
                    content TEXT NOT NULL,
                    created_at TEXT NOT NULL
                );
                CREATE INDEX IF NOT EXISTS idx_messages_session
                    ON messages(session_id, id);
                """
            )

    # ----- Nutzer -----------------------------------------------------------
    def _row_to_user(self, row: sqlite3.Row) -> User:
        return User(
            id=row["id"],
            display_name=row["display_name"],
            prefs=json.loads(row["prefs_json"] or "{}"),
            created_at=row["created_at"],
        )

    def create_user(self, display_name: str) -> tuple[User, str]:
        raw_token = secrets.token_urlsafe(32)
        user = User(id=uuid.uuid4().hex, display_name=display_name, prefs={}, created_at=_now())
        with self._connect() as conn:
            conn.execute(
                "INSERT INTO users (id, display_name, token_hash, prefs_json, created_at)"
                " VALUES (?, ?, ?, ?, ?)",
                (user.id, user.display_name, hash_token(raw_token), "{}", user.created_at),
            )
        return user, raw_token

    def get_user_by_token(self, raw_token: str) -> User | None:
        with self._connect() as conn:
            row = conn.execute(
                "SELECT * FROM users WHERE token_hash = ?", (hash_token(raw_token),)
            ).fetchone()
        return self._row_to_user(row) if row else None

    def get_user(self, user_id: str) -> User | None:
        with self._connect() as conn:
            row = conn.execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone()
        return self._row_to_user(row) if row else None

    def set_user_prefs(self, user_id: str, prefs: dict) -> User:
        with self._connect() as conn:
            conn.execute(
                "UPDATE users SET prefs_json = ? WHERE id = ?",
                (json.dumps(prefs), user_id),
            )
            row = conn.execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone()
        if row is None:
            raise KeyError(f"Unbekannter Nutzer: {user_id}")
        return self._row_to_user(row)

    def ensure_anonymous_user(self) -> User:
        existing = self.get_user(ANONYMOUS_USER_ID)
        if existing:
            return existing
        with self._connect() as conn:
            conn.execute(
                "INSERT OR IGNORE INTO users (id, display_name, token_hash, prefs_json, created_at)"
                " VALUES (?, ?, ?, ?, ?)",
                (ANONYMOUS_USER_ID, "Anonymous", f"anon-{ANONYMOUS_USER_ID}", "{}", _now()),
            )
        return self.get_user(ANONYMOUS_USER_ID)

    # ----- Sessions ---------------------------------------------------------
    def get_session(self, session_id: str) -> Session | None:
        with self._connect() as conn:
            row = conn.execute(
                "SELECT * FROM sessions WHERE id = ?", (session_id,)
            ).fetchone()
        if row is None:
            return None
        return Session(id=row["id"], user_id=row["user_id"], data=json.loads(row["data_json"] or "{}"))

    def update_session(self, session_id: str, user_id: str, values: dict) -> Session:
        existing = self.get_session(session_id)
        if existing and existing.user_id != user_id:
            raise SessionOwnershipError(
                f"Session {session_id!r} gehoert einem anderen Nutzer"
            )

        data = dict(existing.data) if existing else {}
        data.update({k: v for k, v in values.items() if v is not None})
        payload = json.dumps(data)
        now = _now()

        with self._connect() as conn:
            if existing:
                conn.execute(
                    "UPDATE sessions SET data_json = ?, updated_at = ? WHERE id = ?",
                    (payload, now, session_id),
                )
            else:
                conn.execute(
                    "INSERT INTO sessions (id, user_id, data_json, created_at, updated_at)"
                    " VALUES (?, ?, ?, ?, ?)",
                    (session_id, user_id, payload, now, now),
                )
        return Session(id=session_id, user_id=user_id, data=data)

    # ----- Nachrichten / Gespraechsverlauf ----------------------------------
    def append_message(self, session_id: str, user_id: str, role: str, content: str) -> None:
        existing = self.get_session(session_id)
        if existing and existing.user_id != user_id:
            raise SessionOwnershipError(
                f"Session {session_id!r} gehoert einem anderen Nutzer"
            )
        now = _now()
        with self._connect() as conn:
            if not existing:
                conn.execute(
                    "INSERT INTO sessions (id, user_id, data_json, created_at, updated_at)"
                    " VALUES (?, ?, ?, ?, ?)",
                    (session_id, user_id, "{}", now, now),
                )
            conn.execute(
                "INSERT INTO messages (session_id, role, content, created_at)"
                " VALUES (?, ?, ?, ?)",
                (session_id, role, content, now),
            )

    def get_recent_messages(self, session_id: str, limit: int) -> list[dict]:
        if limit <= 0:
            return []
        with self._connect() as conn:
            rows = conn.execute(
                "SELECT role, content FROM messages WHERE session_id = ?"
                " ORDER BY id DESC LIMIT ?",
                (session_id, limit),
            ).fetchall()
        return [{"role": row["role"], "content": row["content"]} for row in reversed(rows)]