feat: sync mobile queue priority across devices (Closes #1464)

src/main.py

@@ -71,6 +71,7 @@ from src.push_endpoint_policy import UnsafePushEndpoint, validate_public_push_en
 from src.push_subscription_store import build_push_subscription_store
 from src.request_boundary import RequestBodyLimitMiddleware, request_body_limit
 from src.saved_search_store import SavedSearchConflict, SavedSearchStore
+from src.queue_priority_store import QueuePriorityConflict, QueuePriorityStore
 from src.following_store import FollowingStore
 from src.unfiled_draft_store import (
     UnfiledDraftConflict,
@@ -965,6 +966,11 @@ class SavedSearchCollection(BaseModel):
     views: list[SavedSearchView] = Field(max_length=20)
 
 
+class QueuePriorityCollection(BaseModel):
+    revision: int = Field(ge=0)
+    order: list[str] = Field(min_length=9, max_length=9)
+
+
 class CompletedFiledReviewReceipt(BaseModel):
     repository: str = Field(
         min_length=3,
@@ -1784,7 +1790,7 @@ async def require_operator_session(request: Request, call_next):
 async def prevent_live_api_caching(request, call_next):
     response = await call_next(request)
     path = dashboard_auth.application_path(request)
-    if path in {"/api/v1/context", "/api/v1/background-identity", "/api/v1/events", "/api/v1/live", "/api/v1/available-issues", "/api/v1/search", "/api/v1/work-route", "/api/v1/today", "/api/v1/tomorrow", "/api/v1/tomorrow/promote", "/api/v1/week", "/api/v1/week/promote", "/api/v1/week/start-early", "/api/v1/week/reconcile", "/api/v1/week/reschedule", "/api/v1/week/pull-item", "/api/v1/today/session", "/api/v1/later", "/api/v1/saved-searches", "/api/v1/completed-filed-reviews", "/api/v1/security-events", "/api/v1/push-subscription"} or path.startswith("/api/v1/human-gate") or path.startswith("/api/v1/work/") or (
+    if path in {"/api/v1/context", "/api/v1/background-identity", "/api/v1/events", "/api/v1/live", "/api/v1/available-issues", "/api/v1/search", "/api/v1/work-route", "/api/v1/today", "/api/v1/tomorrow", "/api/v1/tomorrow/promote", "/api/v1/week", "/api/v1/week/promote", "/api/v1/week/start-early", "/api/v1/week/reconcile", "/api/v1/week/reschedule", "/api/v1/week/pull-item", "/api/v1/today/session", "/api/v1/later", "/api/v1/saved-searches", "/api/v1/queue-priority", "/api/v1/completed-filed-reviews", "/api/v1/security-events", "/api/v1/push-subscription"} or path.startswith("/api/v1/human-gate") or path.startswith("/api/v1/work/") or (
         path.startswith("/api/v1/repos/")
         and path.endswith("/review")
     ) or path.startswith("/api/v1/notifications") or (
@@ -3102,6 +3108,12 @@ def _saved_search_store() -> SavedSearchStore:
     )
 
 
+def _queue_priority_store() -> QueuePriorityStore:
+    return QueuePriorityStore(
+        os.getenv("STACKCHAIN_QUEUE_PRIORITY_DB", str(_state_dir / "queue-priority.sqlite3"))
+    )
+
+
 def _following_store() -> FollowingStore:
     return FollowingStore(
         os.getenv("STACKCHAIN_FOLLOWING_DB", str(_state_dir / "following.sqlite3"))
@@ -3369,6 +3381,46 @@ async def replace_saved_searches(payload: SavedSearchCollection):
         )
 
 
+@app.get("/api/v1/queue-priority")
+async def get_queue_priority(response: Response):
+    login = await _confirmed_login()
+    try:
+        snapshot = await asyncio.to_thread(_queue_priority_store().get, login)
+    except (OSError, sqlite3.Error, PrivateStateEncryptionError):
+        raise HTTPException(
+            status_code=503,
+            detail="Queue priority synchronization is unavailable",
+            headers={"Retry-After": "1"},
+        )
+    response.headers["Cache-Control"] = "no-store"
+    return snapshot
+
+
+@app.put("/api/v1/queue-priority")
+async def replace_queue_priority(payload: QueuePriorityCollection):
+    login = await _confirmed_login()
+    try:
+        return await asyncio.to_thread(
+            _queue_priority_store().replace, login, payload.revision, payload.order
+        )
+    except QueuePriorityConflict as exc:
+        raise HTTPException(
+            status_code=409,
+            detail={
+                "message": "Queue priority changed on another device.",
+                "snapshot": exc.snapshot,
+            },
+        )
+    except ValueError as exc:
+        raise HTTPException(status_code=422, detail=str(exc))
+    except (OSError, sqlite3.Error, PrivateStateEncryptionError):
+        raise HTTPException(
+            status_code=503,
+            detail="Queue priority synchronization is unavailable",
+            headers={"Retry-After": "1"},
+        )
+
+
 @app.get("/api/v1/unfiled-drafts")
 async def get_unfiled_drafts(response: Response):
     login = await _confirmed_login()

src/queue_priority_store.py

@@ -0,0 +1,100 @@
+"""Encrypted, revisioned mobile routine queue priority."""
+
+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
+
+
+DEFAULT_QUEUE_ORDER = (
+    "attention", "today", "update", "agenda", "following", "authored", "filed", "later", "draft",
+)
+
+
+class QueuePriorityConflict(ValueError):
+    """Raised when a stale client attempts to replace the queue order."""
+
+    def __init__(self, snapshot: dict):
+        super().__init__("queue priority changed on another device")
+        self.snapshot = snapshot
+
+
+class QueuePriorityStore:
+    def __init__(self, path: str | Path, *, timeout: float = 1.0, encryption_key: bytes | None = None):
+        self.path = Path(path)
+        self.timeout = timeout
+        self._cipher = PrivateStateCipher(
+            encryption_key if encryption_key is not None else private_state_encryption_config(),
+            store="queue-priority",
+        )
+        with self._connect() as connection:
+            connection.execute("PRAGMA journal_mode=WAL")
+            connection.execute(
+                "CREATE TABLE IF NOT EXISTS queue_priorities ("
+                "login TEXT PRIMARY KEY, revision INTEGER NOT NULL, queue_order 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(order: list[str] | tuple[str, ...]) -> list[str]:
+        if not isinstance(order, (list, tuple)) or len(order) != len(DEFAULT_QUEUE_ORDER):
+            raise ValueError("order must contain every routine queue")
+        if any(not isinstance(name, str) for name in order) or set(order) != set(DEFAULT_QUEUE_ORDER):
+            raise ValueError("order must contain every routine queue exactly once")
+        return list(order)
+
+    def _snapshot(self, row, login: str) -> tuple[dict, bool]:
+        if row is None:
+            return {"revision": 0, "order": list(DEFAULT_QUEUE_ORDER)}, False
+        order, legacy = self._cipher.open(row[1], binding=f"order:{login}")
+        try:
+            normalized = self._normalize(order)
+        except ValueError as error:
+            raise PrivateStateEncryptionError("private state could not be decrypted") from error
+        return {"revision": int(row[0]), "order": normalized}, legacy
+
+    def get(self, login: str) -> dict:
+        login = self._login(login)
+        with self._connect() as connection:
+            row = connection.execute(
+                "SELECT revision, queue_order FROM queue_priorities WHERE login = ?", (login,)
+            ).fetchone()
+            snapshot, legacy = self._snapshot(row, login)
+            if row is not None and legacy:
+                connection.execute(
+                    "UPDATE queue_priorities SET queue_order = ? WHERE login = ? AND queue_order = ?",
+                    (self._cipher.seal(snapshot["order"], binding=f"order:{login}"), login, row[1]),
+                )
+        return snapshot
+
+    def replace(self, login: str, expected_revision: int, order: list[str]) -> dict:
+        login = self._login(login)
+        if not isinstance(expected_revision, int) or isinstance(expected_revision, bool) or expected_revision < 0:
+            raise ValueError("revision is invalid")
+        normalized = self._normalize(order)
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            row = connection.execute(
+                "SELECT revision, queue_order FROM queue_priorities WHERE login = ?", (login,)
+            ).fetchone()
+            current, _legacy = self._snapshot(row, login)
+            if current["revision"] != expected_revision:
+                raise QueuePriorityConflict(current)
+            revision = expected_revision + 1
+            sealed = self._cipher.seal(normalized, binding=f"order:{login}")
+            connection.execute(
+                "INSERT INTO queue_priorities(login, revision, queue_order) VALUES (?, ?, ?) "
+                "ON CONFLICT(login) DO UPDATE SET revision=excluded.revision, queue_order=excluded.queue_order",
+                (login, revision, sealed),
+            )
+        return {"revision": revision, "order": normalized}