stackchain-dashboard/src/completed_filed_review_store.py

"""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)