feat: make Following a changed-first inbox (Closes #1297)

src/following_store.py

@@ -86,6 +86,13 @@ class FollowingStore:
         url = raw.get("url")
         if not isinstance(url, str) or not url.startswith(("http://", "https://")) or len(url) > 2048:
             raise ValueError("url is invalid")
+        last_seen_updated_at = raw.get("last_seen_updated_at", updated_at)
+        if (
+            not isinstance(last_seen_updated_at, str)
+            or not last_seen_updated_at
+            or len(last_seen_updated_at) > 64
+        ):
+            raise ValueError("last seen update is invalid")
         return {
             "repository": repository,
             "number": number,
@@ -93,8 +100,22 @@ class FollowingStore:
             "state": state,
             "updated_at": updated_at,
             "url": url,
+            "last_seen_updated_at": last_seen_updated_at,
         }
 
+    @classmethod
+    def _present(cls, snapshot: dict) -> dict:
+        changed = []
+        unchanged = []
+        for raw in snapshot["items"]:
+            stored = cls._normalize_item(raw)
+            unseen = stored["updated_at"] != stored["last_seen_updated_at"]
+            item = {key: value for key, value in stored.items() if key != "last_seen_updated_at"}
+            item["has_unseen_change"] = unseen
+            (changed if unseen else unchanged).append(item)
+        changed.sort(key=lambda item: item["updated_at"], reverse=True)
+        return {"revision": snapshot["revision"], "items": changed + unchanged}
+
     @staticmethod
     def _identity(item: dict) -> tuple[str, int]:
         return item["repository"].lower(), item["number"]
@@ -106,12 +127,13 @@ class FollowingStore:
                 "SELECT revision, items FROM following_issues WHERE login = ?", (login,)
             ).fetchone()
             snapshot, legacy = self._snapshot(row, login)
+            snapshot["items"] = [self._normalize_item(item) for item in snapshot["items"]]
             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
+        return self._present(snapshot)
 
     def preflight(self, login: str, raw_item: dict, watching: bool) -> dict:
         """Validate a requested change and capacity without mutating the registry."""
@@ -146,7 +168,7 @@ class FollowingStore:
                 "SELECT revision, items FROM following_issues WHERE login = ?", (login,)
             ).fetchone()
             current, _legacy = self._snapshot(row, login)
-            items = list(current["items"])
+            items = [self._normalize_item(candidate) for candidate in current["items"]]
             index = next(
                 (position for position, candidate in enumerate(items)
                  if self._identity(candidate) == identity),
@@ -157,13 +179,14 @@ class FollowingStore:
                     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:
+                    item["last_seen_updated_at"] = items[index]["last_seen_updated_at"]
+                    if items[index] == item:
+                        return self._present({"revision": current["revision"], "items": items})
                     items.pop(index)
                     items.insert(0, item)
             elif index is None:
-                return current
+                return self._present({"revision": current["revision"], "items": items})
             else:
                 items.pop(index)
             revision = current["revision"] + 1
@@ -172,4 +195,58 @@ class FollowingStore:
                 "ON CONFLICT(login) DO UPDATE SET revision=excluded.revision, items=excluded.items",
                 (login, revision, self._seal(login, items)),
             )
-        return {"revision": revision, "items": items}
+        return self._present({"revision": revision, "items": items})
+
+    def refresh(self, login: str, raw_items: list[dict]) -> dict:
+        """Merge successful upstream snapshots while preserving seen revisions."""
+        login = self._login(login)
+        fresh = {self._identity(item): self._normalize_item(item) for item in raw_items}
+        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 = [self._normalize_item(candidate) for candidate in current["items"]]
+            changed = False
+            for index, item in enumerate(items):
+                update = fresh.get(self._identity(item))
+                if update is None:
+                    continue
+                update["last_seen_updated_at"] = item["last_seen_updated_at"]
+                if update != item:
+                    items[index] = update
+                    changed = True
+            revision = current["revision"]
+            if changed:
+                revision += 1
+                connection.execute(
+                    "UPDATE following_issues SET revision = ?, items = ? WHERE login = ?",
+                    (revision, self._seal(login, items), login),
+                )
+        return self._present({"revision": revision, "items": items})
+
+    def acknowledge(self, login: str, repository: str, number: int, updated_at: str) -> dict:
+        """Mark only the exact upstream revision successfully opened by the operator."""
+        login = self._login(login)
+        identity = (str(repository).lower(), number)
+        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 = [self._normalize_item(candidate) for candidate in current["items"]]
+            revision = current["revision"]
+            for item in items:
+                if self._identity(item) != identity or item["updated_at"] != updated_at:
+                    continue
+                if item["last_seen_updated_at"] != updated_at:
+                    item["last_seen_updated_at"] = updated_at
+                    revision += 1
+                    connection.execute(
+                        "UPDATE following_issues SET revision = ?, items = ? WHERE login = ?",
+                        (revision, self._seal(login, items), login),
+                    )
+                break
+        return self._present({"revision": revision, "items": items})

src/main.py

@@ -2564,6 +2564,10 @@ def _following_store() -> FollowingStore:
     )
 
 
+class FollowingSeenRevision(BaseModel):
+    updated_at: str = Field(min_length=1, max_length=64)
+
+
 def _completed_filed_review_store() -> CompletedFiledReviewStore:
     return CompletedFiledReviewStore(
         os.getenv(
@@ -2611,8 +2615,33 @@ async def _confirmed_login() -> str:
 @app.get("/api/v1/following")
 async def get_following(response: Response):
     login = await _confirmed_login()
+    store = _following_store()
     try:
-        snapshot = await asyncio.to_thread(_following_store().get, login)
+        snapshot = await asyncio.to_thread(store.get, login)
+        semaphore = asyncio.Semaphore(5)
+
+        async def refresh_item(item: dict) -> dict:
+            async with semaphore:
+                preview = await asyncio.wait_for(
+                    gitea_proxy.work_preview(item["repository"], "issue", item["number"]),
+                    timeout=GLOBAL_SEARCH_TIMEOUT_SECONDS,
+                )
+            return store._normalize_item({
+                "repository": item["repository"],
+                "number": item["number"],
+                "title": preview.get("title", ""),
+                "state": preview.get("state", ""),
+                "updated_at": preview.get("updated_at", ""),
+                "url": preview.get("url", ""),
+            })
+
+        results = await asyncio.gather(
+            *(refresh_item(item) for item in snapshot["items"]),
+            return_exceptions=True,
+        )
+        refreshed = [result for result in results if isinstance(result, dict)]
+        failures = len(results) - len(refreshed)
+        snapshot = await asyncio.to_thread(store.refresh, login, refreshed)
     except (OSError, sqlite3.Error, PrivateStateEncryptionError):
         raise HTTPException(
             status_code=503,
@@ -2620,7 +2649,33 @@ async def get_following(response: Response):
             headers={"Retry-After": "1"},
         )
     response.headers["Cache-Control"] = "no-store"
-    return snapshot
+    return {**snapshot, "degraded": failures > 0, "refresh_failures": failures}
+
+
+@app.put("/api/v1/following/{owner}/{repo}/issues/{number}/seen")
+async def acknowledge_following_revision(
+    payload: FollowingSeenRevision,
+    owner: str,
+    repo: str,
+    number: int = PathParam(gt=0),
+):
+    login = await _confirmed_login()
+    try:
+        return await asyncio.to_thread(
+            _following_store().acknowledge,
+            login,
+            f"{owner}/{repo}",
+            number,
+            payload.updated_at,
+        )
+    except ValueError as exc:
+        raise HTTPException(status_code=422, detail=str(exc)) from exc
+    except (OSError, sqlite3.Error, PrivateStateEncryptionError):
+        raise HTTPException(
+            status_code=503,
+            detail="Following synchronization is unavailable",
+            headers={"Retry-After": "1"},
+        )
 
 
 @app.get("/api/v1/completed-filed-reviews")