"""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") last_seen_updated_at = raw.get("last_seen_updated_at", updated_at) if ( not isinstance(last_seen_updated_at, str) or not last_seen_updated_at or len(last_seen_updated_at) > 64 ): raise ValueError("last seen update is invalid") return { "repository": repository, "number": number, "title": title.strip(), "state": state, "updated_at": updated_at, "url": url, "last_seen_updated_at": last_seen_updated_at, } @classmethod def _present(cls, snapshot: dict) -> dict: changed = [] unchanged = [] for raw in snapshot["items"]: stored = cls._normalize_item(raw) unseen = stored["updated_at"] != stored["last_seen_updated_at"] item = {key: value for key, value in stored.items() if key != "last_seen_updated_at"} item["has_unseen_change"] = unseen (changed if unseen else unchanged).append(item) changed.sort(key=lambda item: item["updated_at"], reverse=True) return {"revision": snapshot["revision"], "items": changed + unchanged} @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) snapshot["items"] = [self._normalize_item(item) for item in snapshot["items"]] 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 self._present(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 = [self._normalize_item(candidate) for candidate in 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) else: item["last_seen_updated_at"] = items[index]["last_seen_updated_at"] if items[index] == item: return self._present({"revision": current["revision"], "items": items}) items.pop(index) items.insert(0, item) elif index is None: return self._present({"revision": current["revision"], "items": items}) 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 self._present({"revision": revision, "items": items}) def refresh(self, login: str, raw_items: list[dict]) -> dict: """Merge successful upstream snapshots while preserving seen revisions.""" login = self._login(login) fresh = {self._identity(item): self._normalize_item(item) for item in raw_items} 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 = [self._normalize_item(candidate) for candidate in current["items"]] changed = False for index, item in enumerate(items): update = fresh.get(self._identity(item)) if update is None: continue update["last_seen_updated_at"] = item["last_seen_updated_at"] if update != item: items[index] = update changed = True revision = current["revision"] if changed: revision += 1 connection.execute( "UPDATE following_issues SET revision = ?, items = ? WHERE login = ?", (revision, self._seal(login, items), login), ) return self._present({"revision": revision, "items": items}) def acknowledge(self, login: str, repository: str, number: int, updated_at: str) -> dict: """Mark only the exact upstream revision successfully opened by the operator.""" login = self._login(login) identity = (str(repository).lower(), number) 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 = [self._normalize_item(candidate) for candidate in current["items"]] revision = current["revision"] for item in items: if self._identity(item) != identity or item["updated_at"] != updated_at: continue if item["last_seen_updated_at"] != updated_at: item["last_seen_updated_at"] = updated_at revision += 1 connection.execute( "UPDATE following_issues SET revision = ?, items = ? WHERE login = ?", (revision, self._seal(login, items), login), ) break return self._present({"revision": revision, "items": items})