"""Durable, encrypted, account-scoped completed Filed review receipts.""" import re import sqlite3 from datetime import datetime 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_.-]+$") class CompletedFiledReviewStore: def __init__( self, path: str | Path, *, limit: int = 200, 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="completed-filed-reviews", ) self._initialize() def _initialize(self) -> None: with self._connect() as connection: connection.execute("PRAGMA journal_mode=WAL") connection.execute("PRAGMA secure_delete=ON") connection.execute( """ CREATE TABLE IF NOT EXISTS completed_filed_review_collections ( login TEXT PRIMARY KEY, receipts TEXT NOT NULL ) """ ) legacy = connection.execute( "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = 'completed_filed_reviews'" ).fetchone() if legacy: connection.execute("BEGIN IMMEDIATE") rows = connection.execute( "SELECT login, repository, issue_number, updated_at " "FROM completed_filed_reviews ORDER BY login, touched_at" ).fetchall() grouped: dict[str, list[dict]] = {} for login, repository, number, updated_at in rows: grouped.setdefault(login, []).append({ "repository": repository, "number": int(number), "updated_at": updated_at, }) for login, receipts in grouped.items(): normalized = [self._receipt(item) for item in receipts][-self.limit:] connection.execute( "INSERT INTO completed_filed_review_collections(login, receipts) VALUES (?, ?) " "ON CONFLICT(login) DO UPDATE SET receipts=excluded.receipts", (login, self._seal(login, normalized)), ) connection.execute("DROP TABLE completed_filed_reviews") with self._connect() as connection: connection.execute("PRAGMA wal_checkpoint(TRUNCATE)") def _connect(self) -> sqlite3.Connection: return connect_private_sqlite(self.path, timeout=self.timeout) @staticmethod def _login(login: str) -> str: normalized = login.strip().lower() if not normalized: raise ValueError("login is required") return normalized @staticmethod def _receipt(raw: dict) -> dict: if not isinstance(raw, dict): raise ValueError("receipt must be an object") repository = raw.get("repository") number = raw.get("number") updated_at = raw.get("updated_at") if not isinstance(repository, str) or not _REPOSITORY.fullmatch(repository): raise ValueError("repository is invalid") if not isinstance(number, int) or isinstance(number, bool) or number < 1: raise ValueError("number is invalid") if not isinstance(updated_at, str): raise ValueError("updated_at is invalid") try: parsed = datetime.fromisoformat(updated_at.replace("Z", "+00:00")) except ValueError as error: raise ValueError("updated_at is invalid") from error if parsed.tzinfo is None: raise ValueError("updated_at is invalid") return {"repository": repository, "number": number, "updated_at": updated_at} def _open(self, login: str, payload: str | None) -> tuple[list[dict], bool]: if payload is None: return [], False value, stale = self._cipher.open(payload, binding=f"receipts:{login}") if not isinstance(value, list): raise PrivateStateEncryptionError("private state could not be decrypted") try: return [self._receipt(item) for item in value], stale except ValueError as error: raise PrivateStateEncryptionError("private state could not be decrypted") from error def _seal(self, login: str, receipts: list[dict]) -> str: return self._cipher.seal(receipts, binding=f"receipts:{login}") @staticmethod def _snapshot(receipts: list[dict]) -> dict: return {"receipts": receipts} def get(self, login: str) -> dict: login = self._login(login) with self._connect() as connection: row = connection.execute( "SELECT receipts FROM completed_filed_review_collections WHERE login = ?", (login,), ).fetchone() receipts, stale = self._open(login, row[0] if row else None) if row is not None and stale: connection.execute( "UPDATE completed_filed_review_collections SET receipts = ? " "WHERE login = ? AND receipts = ?", (self._seal(login, receipts), login, row[0]), ) return self._snapshot(receipts) def merge(self, login: str, receipts: list[dict]) -> dict: login = self._login(login) if not isinstance(receipts, list) or len(receipts) > self.limit: raise ValueError(f"receipts are limited to {self.limit}") normalized = [self._receipt(receipt) for receipt in receipts] with self._connect() as connection: connection.execute("BEGIN IMMEDIATE") row = connection.execute( "SELECT receipts FROM completed_filed_review_collections WHERE login = ?", (login,), ).fetchone() current, _stale = self._open(login, row[0] if row else None) for incoming in normalized: match = next(( item for item in current if item["repository"] == incoming["repository"] and item["number"] == incoming["number"] ), None) if match is not None and match["updated_at"] >= incoming["updated_at"]: continue if match is not None: current.remove(match) current.append(incoming) current = current[-self.limit:] connection.execute( "INSERT INTO completed_filed_review_collections(login, receipts) VALUES (?, ?) " "ON CONFLICT(login) DO UPDATE SET receipts=excluded.receipts", (login, self._seal(login, current)), ) return self._snapshot(current)