feat: sync Today plan across devices (#357)

src/main.py

@@ -2,6 +2,7 @@ import asyncio
 import hmac
 import math
 import os
+import sqlite3
 import time
 from collections.abc import Awaitable, Coroutine
 from contextlib import asynccontextmanager
@@ -34,6 +35,7 @@ from src.login_attempt_store import LoginAttemptStore, LoginAttemptStoreError, c
 from src.models import Issue, Milestone, PullRequest, Repo, User
 from src.request_boundary import RequestBodyLimitMiddleware, request_body_limit
 from src.suggestion_engine import compute
+from src.today_store import TodayPlanFull, TodayStore
 from src.views import router as frontend_router
 
 @asynccontextmanager
@@ -199,6 +201,21 @@ class NotificationReadBatch(BaseModel):
     ids: list[PositiveInt] = Field(min_length=1, max_length=50)
 
 
+class TodayOperation(BaseModel):
+    operation_id: str = Field(min_length=1, max_length=100)
+    action: Literal["add", "remove", "move"]
+    item_id: str = Field(min_length=1, max_length=500)
+    direction: Literal["up", "down"] | None = None
+
+    @model_validator(mode="after")
+    def require_move_direction(self):
+        if self.action == "move" and self.direction is None:
+            raise ValueError("move requires a direction")
+        if self.action != "move" and self.direction is not None:
+            raise ValueError("direction is only valid for move")
+        return self
+
+
 class NotificationReply(BaseModel):
     body: str = Field(min_length=1, max_length=10_000)
 
@@ -616,7 +633,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"} 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"} or path.startswith("/api/v1/work/") or (
         path.startswith("/api/v1/repos/")
         and path.endswith("/review")
     ) or path.startswith("/api/v1/notifications") or (
@@ -838,6 +855,58 @@ async def session_status(request: Request):
     }
 
 
+def _today_store() -> TodayStore:
+    return TodayStore(
+        os.getenv("STACKCHAIN_TODAY_DB", str(_state_dir / "today.sqlite3")), limit=5
+    )
+
+
+async def _confirmed_login() -> str:
+    try:
+        user = await asyncio.wait_for(
+            current_user(), timeout=ISSUE_ACTION_TIMEOUT_SECONDS
+        )
+    except Exception as exc:
+        raise HTTPException(
+            status_code=503, detail="Operator identity is unavailable"
+        ) from exc
+    login = user.get("login", "") if isinstance(user, dict) else ""
+    if not isinstance(login, str) or not login.strip():
+        raise HTTPException(status_code=503, detail="Operator identity is unavailable")
+    return login.strip().lower()
+
+
+@app.get("/api/v1/today")
+async def get_today_plan():
+    login = await _confirmed_login()
+    try:
+        return await asyncio.to_thread(_today_store().get, login)
+    except (OSError, sqlite3.Error):
+        raise HTTPException(
+            status_code=503, detail="Today synchronization is unavailable"
+        )
+
+
+@app.patch("/api/v1/today")
+async def update_today_plan(payload: TodayOperation):
+    login = await _confirmed_login()
+    try:
+        return await asyncio.to_thread(
+            _today_store().apply,
+            login,
+            payload.operation_id,
+            payload.action,
+            payload.item_id,
+            direction=payload.direction,
+        )
+    except TodayPlanFull:
+        raise HTTPException(status_code=409, detail="Today is limited to 5 items")
+    except (OSError, sqlite3.Error):
+        raise HTTPException(
+            status_code=503, detail="Today synchronization is unavailable"
+        )
+
+
 @app.delete("/api/v1/session")
 async def sign_out(request: Request, response: Response):
     session = request.state.dashboard_session

src/today_store.py

@@ -0,0 +1,130 @@
+"""Durable, account-scoped ordered Today plans."""
+
+import json
+import sqlite3
+from pathlib import Path
+
+
+class TodayPlanFull(ValueError):
+    """Raised when an add would exceed the bounded Today plan."""
+
+
+class TodayStore:
+    def __init__(self, path: str | Path, *, limit: int = 5, timeout: float = 1.0):
+        self.path = Path(path)
+        self.limit = limit
+        self.timeout = timeout
+
+    def _connect(self) -> sqlite3.Connection:
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        connection = sqlite3.connect(self.path, timeout=self.timeout)
+        connection.execute("PRAGMA journal_mode=WAL")
+        connection.execute(
+            """
+            CREATE TABLE IF NOT EXISTS today_plans (
+                login TEXT PRIMARY KEY,
+                revision INTEGER NOT NULL,
+                ids TEXT NOT NULL
+            )
+            """
+        )
+        connection.execute(
+            """
+            CREATE TABLE IF NOT EXISTS today_operations (
+                login TEXT NOT NULL,
+                operation_id TEXT NOT NULL,
+                PRIMARY KEY (login, operation_id)
+            )
+            """
+        )
+        return connection
+
+    @staticmethod
+    def _normalize_login(login: str) -> str:
+        normalized = login.strip().lower()
+        if not normalized:
+            raise ValueError("login is required")
+        return normalized
+
+    @staticmethod
+    def _snapshot(row) -> dict:
+        if row is None:
+            return {"revision": 0, "ids": []}
+        return {"revision": int(row[0]), "ids": json.loads(row[1])}
+
+    def get(self, login: str) -> dict:
+        with self._connect() as connection:
+            row = connection.execute(
+                "SELECT revision, ids FROM today_plans WHERE login = ?",
+                (self._normalize_login(login),),
+            ).fetchone()
+        return self._snapshot(row)
+
+    def apply(
+        self,
+        login: str,
+        operation_id: str,
+        action: str,
+        item_id: str,
+        *,
+        direction: str | None = None,
+    ) -> dict:
+        login = self._normalize_login(login)
+        if not operation_id or not item_id:
+            raise ValueError("operation_id and item_id are required")
+        if action not in {"add", "remove", "move"}:
+            raise ValueError("unsupported Today action")
+        if action == "move" and direction not in {"up", "down"}:
+            raise ValueError("move direction must be up or down")
+
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            row = connection.execute(
+                "SELECT revision, ids FROM today_plans WHERE login = ?", (login,)
+            ).fetchone()
+            snapshot = self._snapshot(row)
+            duplicate = connection.execute(
+                "SELECT 1 FROM today_operations WHERE login = ? AND operation_id = ?",
+                (login, operation_id),
+            ).fetchone()
+            if duplicate:
+                return snapshot
+
+            ids = list(snapshot["ids"])
+            changed = False
+            if action == "add":
+                if item_id not in ids:
+                    if len(ids) >= self.limit:
+                        raise TodayPlanFull("Today is limited to five items")
+                    ids.append(item_id)
+                    changed = True
+            elif action == "remove":
+                if item_id in ids:
+                    ids.remove(item_id)
+                    changed = True
+            else:
+                try:
+                    index = ids.index(item_id)
+                except ValueError:
+                    index = -1
+                target = index - 1 if direction == "up" else index + 1
+                if index >= 0 and 0 <= target < len(ids):
+                    ids[index], ids[target] = ids[target], ids[index]
+                    changed = True
+
+            revision = snapshot["revision"] + (1 if changed else 0)
+            if row is None:
+                connection.execute(
+                    "INSERT INTO today_plans(login, revision, ids) VALUES (?, ?, ?)",
+                    (login, revision, json.dumps(ids, separators=(",", ":"))),
+                )
+            elif changed:
+                connection.execute(
+                    "UPDATE today_plans SET revision = ?, ids = ? WHERE login = ?",
+                    (revision, json.dumps(ids, separators=(",", ":")), login),
+                )
+            connection.execute(
+                "INSERT INTO today_operations(login, operation_id) VALUES (?, ?)",
+                (login, operation_id),
+            )
+            return {"revision": revision, "ids": ids}