stackchain-dashboard/src/following_store.py

"""Encrypted, account-scoped registry of explicitly followed Gitea issues."""

import re
import sqlite3
from pathlib import Path

from src.private_state import connect_private_sqlite
from src.state_encryption import PrivateStateCipher, PrivateStateEncryptionError, private_state_encryption_config


_REPOSITORY = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$")
_STATES = {"open", "closed"}


class FollowingStore:
    def __init__(
        self,
        path: str | Path,
        *,
        limit: int = 50,
        timeout: float = 1.0,
        encryption_key: bytes | None = None,
    ):
        self.path = Path(path)
        self.limit = limit
        self.timeout = timeout
        self._cipher = PrivateStateCipher(
            encryption_key if encryption_key is not None else private_state_encryption_config(),
            store="following",
        )
        self._initialize()

    def _connect(self) -> sqlite3.Connection:
        return connect_private_sqlite(self.path, timeout=self.timeout)

    def _initialize(self) -> None:
        with self._connect() as connection:
            connection.execute("PRAGMA journal_mode=WAL")
            connection.execute(
                """
                CREATE TABLE IF NOT EXISTS following_issues (
                    login TEXT PRIMARY KEY,
                    revision INTEGER NOT NULL,
                    items TEXT NOT NULL
                )
                """
            )

    @staticmethod
    def _login(login: str) -> str:
        normalized = str(login).strip().lower()
        if not normalized:
            raise ValueError("login is required")
        return normalized

    def _snapshot(self, row, login: str) -> tuple[dict, bool]:
        if row is None:
            return {"revision": 0, "items": []}, False
        items, legacy = self._cipher.open(row[1], binding=f"items:{login}")
        if not isinstance(items, list):
            raise PrivateStateEncryptionError("private state could not be decrypted")
        return {"revision": int(row[0]), "items": items}, legacy

    def _seal(self, login: str, items: list[dict]) -> str:
        return self._cipher.seal(items, binding=f"items:{login}")

    @staticmethod
    def _normalize_item(raw: dict) -> dict:
        if not isinstance(raw, dict):
            raise ValueError("following item must be an object")
        repository = raw.get("repository")
        if not isinstance(repository, str) or not _REPOSITORY.fullmatch(repository):
            raise ValueError("repository is invalid")
        number = raw.get("number")
        if not isinstance(number, int) or isinstance(number, bool) or number < 1:
            raise ValueError("number is invalid")
        title = raw.get("title")
        if not isinstance(title, str) or not title.strip() or len(title.strip()) > 300:
            raise ValueError("title is invalid")
        state = raw.get("state")
        if state not in _STATES:
            raise ValueError("state is invalid")
        updated_at = raw.get("updated_at")
        if not isinstance(updated_at, str) or not updated_at or len(updated_at) > 64:
            raise ValueError("updated_at is invalid")
        url = raw.get("url")
        if not isinstance(url, str) or not url.startswith(("http://", "https://")) or len(url) > 2048:
            raise ValueError("url is invalid")
        return {
            "repository": repository,
            "number": number,
            "title": title.strip(),
            "state": state,
            "updated_at": updated_at,
            "url": url,
        }

    @staticmethod
    def _identity(item: dict) -> tuple[str, int]:
        return item["repository"].lower(), item["number"]

    def get(self, login: str) -> dict:
        login = self._login(login)
        with self._connect() as connection:
            row = connection.execute(
                "SELECT revision, items FROM following_issues WHERE login = ?", (login,)
            ).fetchone()
            snapshot, legacy = self._snapshot(row, login)
            if row is not None and legacy:
                connection.execute(
                    "UPDATE following_issues SET items = ? WHERE login = ? AND items = ?",
                    (self._seal(login, snapshot["items"]), login, row[1]),
                )
        return snapshot

    def preflight(self, login: str, raw_item: dict, watching: bool) -> dict:
        """Validate a requested change and capacity without mutating the registry."""
        login = self._login(login)
        item = self._normalize_item(raw_item)
        if not isinstance(watching, bool):
            raise ValueError("watching is invalid")
        if not watching:
            return item
        identity = self._identity(item)
        with self._connect() as connection:
            row = connection.execute(
                "SELECT revision, items FROM following_issues WHERE login = ?", (login,)
            ).fetchone()
            current, _legacy = self._snapshot(row, login)
        if (
            len(current["items"]) >= self.limit
            and not any(self._identity(candidate) == identity for candidate in current["items"])
        ):
            raise ValueError(f"following is limited to {self.limit} issues")
        return item

    def set_watching(self, login: str, raw_item: dict, watching: bool) -> dict:
        login = self._login(login)
        item = self._normalize_item(raw_item)
        if not isinstance(watching, bool):
            raise ValueError("watching is invalid")
        identity = self._identity(item)
        with self._connect() as connection:
            connection.execute("BEGIN IMMEDIATE")
            row = connection.execute(
                "SELECT revision, items FROM following_issues WHERE login = ?", (login,)
            ).fetchone()
            current, _legacy = self._snapshot(row, login)
            items = list(current["items"])
            index = next(
                (position for position, candidate in enumerate(items)
                 if self._identity(candidate) == identity),
                None,
            )
            if watching:
                if index is None:
                    if len(items) >= self.limit:
                        raise ValueError(f"following is limited to {self.limit} issues")
                    items.insert(0, item)
                elif items[index] == item:
                    return current
                else:
                    items.pop(index)
                    items.insert(0, item)
            elif index is None:
                return current
            else:
                items.pop(index)
            revision = current["revision"] + 1
            connection.execute(
                "INSERT INTO following_issues(login, revision, items) VALUES (?, ?, ?) "
                "ON CONFLICT(login) DO UPDATE SET revision=excluded.revision, items=excluded.items",
                (login, revision, self._seal(login, items)),
            )
        return {"revision": revision, "items": items}