feat: add mobile Following queue for watched issues (Closes #1293)

src/following_store.py

@@ -0,0 +1,154 @@
+"""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")
+        return {
+            "repository": repository,
+            "number": number,
+            "title": title.strip(),
+            "state": state,
+            "updated_at": updated_at,
+            "url": url,
+        }
+
+    @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)
+            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 snapshot
+
+    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 = list(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)
+                elif items[index] == item:
+                    return current
+                else:
+                    items.pop(index)
+                    items.insert(0, item)
+            elif index is None:
+                return current
+            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 {"revision": revision, "items": items}

src/frontend_bundle.py

@@ -39,7 +39,7 @@ FEATURE_SOURCES = {
         "static/tomorrow-plan.js", "static/week-calendar.js", "static/week-calendar-import.js", "static/week-plan.js", "static/today-week-reschedule.js", "static/search-week-plan.js", "static/search-batch-plan.js", "static/agenda-session-launcher.js", "static/mobile-plan-today-nav.js",
     ),
     "today-timer": (
-        "static/mobile-app-badge.js", "static/conversation.js", "static/widgets.js", "static/voice-transcript-store.js", "static/voice-conversation-capture.js", "static/mobile-launch.js", "static/mobile-insights.js", "static/mobile-app-shortcuts.js", "static/mobile-find-work-nav.js", "static/mobile-pull-refresh.js", "static/live-data-status.js",
+        "static/mobile-app-badge.js", "static/conversation.js", "static/widgets.js", "static/voice-transcript-store.js", "static/voice-conversation-capture.js", "static/mobile-launch.js", "static/mobile-insights.js", "static/mobile-app-shortcuts.js", "static/mobile-find-work-nav.js", "static/mobile-pull-refresh.js", "static/live-data-status.js", "static/following.js", "static/mobile-composer-viewport.js",
         "static/today-completion.js", "static/card-planning.js", "static/work-detail-position.js", "static/work-route.js", "static/commands.js", "static/saved-searches.js", "static/task-overlay-history.js", "static/search-preview.js", "static/mobile-search-preview-nav.js", "static/search-reply-draft-store.js", "static/conversation-reply-draft-store.js", "static/conversation-photo-drafts.js", "static/search-defer.js", "static/mobile-search-viewport.js", "static/agenda-replan.js", "static/agenda-calendar.js", "static/my-work.js", "static/protect-today.js", "static/mobile-today-command-bar.js", "static/mobile-task-dock.js", "static/mobile-first-task.js", "static/mobile-work-entry.js", "static/mobile-queue-launcher.js", "static/mobile-delivery-recovery.js", "static/mobile-start-day.js", "static/update-triage-session.js", "static/update-review-handoff.js", "static/update-triage-launcher.js", "static/update-triage-gesture.js", "static/notification-undo.js", "static/today-timer.js", "static/today-break.js", "static/today-progress.js", "static/today-lock-screen.js", "static/today-session-sync.js", "static/today-recap.js", "static/today-wrap-up.js", "static/today-summary.js", "static/today-handoff.js",
         "static/later-work.js", "static/detail-defer.js", "static/later-picker.js", "static/drafts.js", "static/unfiled-captures.js", "static/unfiled-draft-sync.js",
         "static/assign-and-start.js", "static/filed-claim.js", "static/queue-today.js", "static/create-and-start.js",

src/main.py

@@ -63,6 +63,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.following_store import FollowingStore
 from src.unfiled_draft_store import (
     UnfiledDraftConflict,
     UnfiledDraftEncryptionError,
@@ -2557,6 +2558,12 @@ def _saved_search_store() -> SavedSearchStore:
     )
 
 
+def _following_store() -> FollowingStore:
+    return FollowingStore(
+        os.getenv("STACKCHAIN_FOLLOWING_DB", str(_state_dir / "following.sqlite3"))
+    )
+
+
 def _completed_filed_review_store() -> CompletedFiledReviewStore:
     return CompletedFiledReviewStore(
         os.getenv(
@@ -2601,6 +2608,21 @@ async def _confirmed_login() -> str:
     return login.strip().lower()
 
 
+@app.get("/api/v1/following")
+async def get_following(response: Response):
+    login = await _confirmed_login()
+    try:
+        snapshot = await asyncio.to_thread(_following_store().get, login)
+    except (OSError, sqlite3.Error, PrivateStateEncryptionError):
+        raise HTTPException(
+            status_code=503,
+            detail="Following synchronization is unavailable",
+            headers={"Retry-After": "1"},
+        )
+    response.headers["Cache-Control"] = "no-store"
+    return snapshot
+
+
 @app.get("/api/v1/completed-filed-reviews")
 async def get_completed_filed_reviews(response: Response):
     login = await _confirmed_login()
@@ -3662,7 +3684,7 @@ async def global_search_preview(
 
 async def _search_preview_subscription_target(
     owner: str, repo: str, number: int, kind: str
-) -> str:
+) -> tuple[str, dict]:
     repository = f"{owner}/{repo}"
     preview = await gitea_proxy.work_preview(repository, kind, number)
     if (
@@ -3673,7 +3695,7 @@ async def _search_preview_subscription_target(
         or preview.get("state") != "open"
     ):
         raise HTTPException(status_code=404, detail="Watchable search result not found")
-    return repository
+    return repository, preview
 
 
 @app.get("/api/v1/repos/{owner}/{repo}/issues/{number}/preview/subscription")
@@ -3684,7 +3706,7 @@ async def global_search_preview_subscription(
     kind: Literal["issue", "pull"] = Query(),
 ) -> JSONResponse:
     try:
-        repository = await _search_preview_subscription_target(owner, repo, number, kind)
+        repository, _preview = await _search_preview_subscription_target(owner, repo, number, kind)
         result = await asyncio.wait_for(
             gitea_proxy.issue_subscription(repository, number),
             timeout=GLOBAL_SEARCH_TIMEOUT_SECONDS,
@@ -3713,11 +3735,29 @@ async def mutate_global_search_preview_subscription(
 ) -> JSONResponse:
     watching = request.method == "PUT"
     try:
-        repository = await _search_preview_subscription_target(owner, repo, number, kind)
+        repository, preview = await _search_preview_subscription_target(owner, repo, number, kind)
+        login = await _confirmed_login()
+        store = _following_store()
+        await asyncio.to_thread(store.get, login)
         result = await asyncio.wait_for(
             gitea_proxy.set_issue_subscription(repository, number, watching),
             timeout=ISSUE_ACTION_TIMEOUT_SECONDS,
         )
+        if result.get("watching") is not watching:
+            raise RuntimeError("Gitea did not confirm subscription state")
+        following = await asyncio.to_thread(
+            store.set_watching,
+            login,
+            {
+                "repository": repository,
+                "number": number,
+                "title": preview.get("title", ""),
+                "state": preview.get("state", ""),
+                "updated_at": preview.get("updated_at", ""),
+                "url": preview.get("url", ""),
+            },
+            watching,
+        )
     except HTTPException:
         raise
     except Exception:
@@ -3726,7 +3766,11 @@ async def mutate_global_search_preview_subscription(
             status_code=503,
             headers={"Retry-After": "1"},
         )
-    return JSONResponse(result)
+    return JSONResponse({
+        **result,
+        "following_revision": following["revision"],
+        "following_count": len(following["items"]),
+    })
 
 
 @app.get("/api/v1/repos/{owner}/{repo}/issues/{number}/preview/conversation")