feat: sync Later deferrals across devices (#363)

src/later_store.py

@@ -0,0 +1,124 @@
+"""Durable, account-scoped Later deferrals."""
+
+import json
+import sqlite3
+from datetime import datetime
+from pathlib import Path
+
+
+class LaterStore:
+    def __init__(self, path: str | Path, *, timeout: float = 1.0):
+        self.path = Path(path)
+        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 later_plans (
+                login TEXT PRIMARY KEY,
+                revision INTEGER NOT NULL,
+                records TEXT NOT NULL
+            )
+            """
+        )
+        connection.execute(
+            """
+            CREATE TABLE IF NOT EXISTS later_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, "records": {}}
+        return {"revision": int(row[0]), "records": json.loads(row[1])}
+
+    @staticmethod
+    def _validate_wake_at(wake_at: str | None) -> str:
+        if not wake_at:
+            raise ValueError("wake_at is required for defer")
+        try:
+            datetime.fromisoformat(wake_at.replace("Z", "+00:00"))
+        except ValueError as error:
+            raise ValueError("wake_at must be an ISO timestamp") from error
+        return wake_at
+
+    def get(self, login: str) -> dict:
+        with self._connect() as connection:
+            row = connection.execute(
+                "SELECT revision, records FROM later_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,
+        *,
+        wake_at: 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 {"defer", "restore"}:
+            raise ValueError("unsupported Later action")
+        if action == "defer":
+            wake_at = self._validate_wake_at(wake_at)
+
+        with self._connect() as connection:
+            connection.execute("BEGIN IMMEDIATE")
+            row = connection.execute(
+                "SELECT revision, records FROM later_plans WHERE login = ?", (login,)
+            ).fetchone()
+            snapshot = self._snapshot(row)
+            duplicate = connection.execute(
+                "SELECT 1 FROM later_operations WHERE login = ? AND operation_id = ?",
+                (login, operation_id),
+            ).fetchone()
+            if duplicate:
+                return snapshot
+
+            records = dict(snapshot["records"])
+            before = records.get(item_id)
+            if action == "defer":
+                records[item_id] = wake_at
+                changed = before != wake_at
+            else:
+                changed = item_id in records
+                records.pop(item_id, None)
+
+            revision = snapshot["revision"] + (1 if changed else 0)
+            serialized = json.dumps(records, separators=(",", ":"), sort_keys=True)
+            if row is None:
+                connection.execute(
+                    "INSERT INTO later_plans(login, revision, records) VALUES (?, ?, ?)",
+                    (login, revision, serialized),
+                )
+            elif changed:
+                connection.execute(
+                    "UPDATE later_plans SET revision = ?, records = ? WHERE login = ?",
+                    (revision, serialized, login),
+                )
+            connection.execute(
+                "INSERT INTO later_operations(login, operation_id) VALUES (?, ?)",
+                (login, operation_id),
+            )
+            return {"revision": revision, "records": records}

src/main.py

@@ -35,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.later_store import LaterStore
 from src.today_store import TodayPlanFull, TodayStore
 from src.views import router as frontend_router
 
@@ -216,6 +217,13 @@ class TodayOperation(BaseModel):
         return self
 
 
+class LaterOperation(BaseModel):
+    operation_id: str = Field(min_length=1, max_length=100)
+    action: Literal["defer", "restore"]
+    item_id: str = Field(min_length=1, max_length=500)
+    wake_at: str | None = Field(default=None, max_length=100)
+
+
 class NotificationReply(BaseModel):
     body: str = Field(min_length=1, max_length=10_000)
 
@@ -633,7 +641,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"} 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/later"} or path.startswith("/api/v1/work/") or (
         path.startswith("/api/v1/repos/")
         and path.endswith("/review")
     ) or path.startswith("/api/v1/notifications") or (
@@ -861,6 +869,12 @@ def _today_store() -> TodayStore:
     )
 
 
+def _later_store() -> LaterStore:
+    return LaterStore(
+        os.getenv("STACKCHAIN_LATER_DB", str(_state_dir / "later.sqlite3"))
+    )
+
+
 async def _confirmed_login() -> str:
     try:
         user = await asyncio.wait_for(
@@ -907,6 +921,37 @@ async def update_today_plan(payload: TodayOperation):
         )
 
 
+@app.get("/api/v1/later")
+async def get_later_plan():
+    login = await _confirmed_login()
+    try:
+        return await asyncio.to_thread(_later_store().get, login)
+    except (OSError, sqlite3.Error):
+        raise HTTPException(
+            status_code=503, detail="Later synchronization is unavailable"
+        )
+
+
+@app.patch("/api/v1/later")
+async def update_later_plan(payload: LaterOperation):
+    login = await _confirmed_login()
+    try:
+        return await asyncio.to_thread(
+            _later_store().apply,
+            login,
+            payload.operation_id,
+            payload.action,
+            payload.item_id,
+            wake_at=payload.wake_at,
+        )
+    except ValueError as error:
+        raise HTTPException(status_code=422, detail=str(error))
+    except (OSError, sqlite3.Error):
+        raise HTTPException(
+            status_code=503, detail="Later synchronization is unavailable"
+        )
+
+
 @app.delete("/api/v1/session")
 async def sign_out(request: Request, response: Response):
     session = request.state.dashboard_session