stackchain-dashboard/src/recent_work_store.py

"""Encrypted, account-scoped recent work shared by signed-in devices."""

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


KINDS = frozenset({"issue", "filed", "pull", "review", "update"})


class RecentWorkStore:
    def __init__(
        self,
        path: str | Path,
        *,
        timeout: float = 1.0,
        encryption_key: bytes | None = None,
        limit: int = 5,
        pinned_limit: int = 20,
    ):
        self.path = Path(path)
        self.timeout = timeout
        self.limit = max(1, int(limit))
        self.pinned_limit = max(1, int(pinned_limit))
        self._cipher = PrivateStateCipher(
            encryption_key if encryption_key is not None else private_state_encryption_config(),
            store="recent-work",
        )
        with self._connect() as connection:
            connection.execute("PRAGMA journal_mode=WAL")
            connection.execute(
                "CREATE TABLE IF NOT EXISTS recent_work ("
                "login TEXT PRIMARY KEY, items TEXT NOT NULL)"
            )

    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 _normalize(item: dict) -> dict:
        if not isinstance(item, dict):
            raise ValueError("recent work item is invalid")
        kind = item.get("kind")
        number = item.get("number")
        title = item.get("title")
        repository = item.get("repository", "")
        if (
            kind not in KINDS
            or not isinstance(number, int)
            or isinstance(number, bool)
            or number < 1
            or not isinstance(title, str)
            or not title.strip()
        ):
            raise ValueError("recent work item is invalid")
        title = title.strip()[:180]
        if kind == "update":
            if repository:
                raise ValueError("recent work item is invalid")
            route = f"#/my-work/update/{number}"
            normalized = {"kind": kind, "number": number, "title": title, "route": route}
        else:
            if (
                not isinstance(repository, str)
                or repository.count("/") != 1
                or any(not part or not all(character.isalnum() or character in "_.-" for character in part)
                       for part in repository.split("/"))
            ):
                raise ValueError("recent work item is invalid")
            route = f"#/my-work/{kind}/{repository}/{number}"
            normalized = {
                "kind": kind,
                "repository": repository,
                "number": number,
                "title": title,
                "route": route,
            }
        if item.get("route", route) != route:
            raise ValueError("recent work item is invalid")
        return normalized

    def _state(self, row, login: str) -> tuple[dict, bool]:
        if row is None:
            return {"items": [], "pinned": []}, False
        payload, legacy = self._cipher.open(row[0], binding=f"items:{login}")
        if isinstance(payload, list):
            payload = {"items": payload, "pinned": []}
            legacy = True
        if not isinstance(payload, dict) or not isinstance(payload.get("items"), list) or not isinstance(payload.get("pinned"), list):
            raise PrivateStateEncryptionError("private state could not be decrypted")
        try:
            items = [self._normalize(item) for item in payload["items"]][: self.limit]
            pinned = [self._normalize(item) for item in payload["pinned"]][: self.pinned_limit]
            if len({item["route"] for item in pinned}) != len(pinned):
                raise ValueError("recent work item is invalid")
            return {"items": items, "pinned": pinned}, legacy
        except ValueError as error:
            raise PrivateStateEncryptionError("private state could not be decrypted") from error

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

    def _write(self, connection: sqlite3.Connection, login: str, state: dict) -> None:
        connection.execute(
            "INSERT INTO recent_work(login, items) VALUES (?, ?) "
            "ON CONFLICT(login) DO UPDATE SET items=excluded.items",
            (login, self._seal(state, login)),
        )

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

    def record(self, login: str, item: dict) -> dict:
        login = self._login(login)
        normalized = self._normalize(item)
        with self._connect() as connection:
            connection.execute("BEGIN IMMEDIATE")
            row = connection.execute(
                "SELECT items FROM recent_work WHERE login = ?", (login,)
            ).fetchone()
            state, _legacy = self._state(row, login)
            state["items"] = [normalized, *(entry for entry in state["items"] if entry["route"] != normalized["route"])][: self.limit]
            state["pinned"] = [
                normalized,
                *(entry for entry in state["pinned"] if entry["route"] != normalized["route"]),
            ] if any(entry["route"] == normalized["route"] for entry in state["pinned"]) else state["pinned"]
            self._write(connection, login, state)
        return state

    def pin(self, login: str, item: dict) -> dict:
        login = self._login(login)
        normalized = self._normalize(item)
        with self._connect() as connection:
            connection.execute("BEGIN IMMEDIATE")
            row = connection.execute(
                "SELECT items FROM recent_work WHERE login = ?", (login,)
            ).fetchone()
            state, _legacy = self._state(row, login)
            state["pinned"] = [
                normalized,
                *(entry for entry in state["pinned"] if entry["route"] != normalized["route"]),
            ][: self.pinned_limit]
            self._write(connection, login, state)
        return state

    def unpin(self, login: str, route: str) -> dict:
        login = self._login(login)
        if not isinstance(route, str) or not route:
            raise ValueError("recent work route is invalid")
        with self._connect() as connection:
            connection.execute("BEGIN IMMEDIATE")
            row = connection.execute(
                "SELECT items FROM recent_work WHERE login = ?", (login,)
            ).fetchone()
            state, _legacy = self._state(row, login)
            state["pinned"] = [entry for entry in state["pinned"] if entry["route"] != route]
            self._write(connection, login, state)
        return state