fix: keep Later plans conflict-safe across devices (#377)

src/later_store.py

@@ -33,6 +33,16 @@ class LaterStore:
             )
             """
         )
+        connection.execute(
+            """
+            CREATE TABLE IF NOT EXISTS later_item_revisions (
+                login TEXT NOT NULL,
+                item_id TEXT NOT NULL,
+                revision INTEGER NOT NULL,
+                PRIMARY KEY (login, item_id)
+            )
+            """
+        )
         return connection
 
     @staticmethod
@@ -74,54 +84,16 @@ class LaterStore:
         item_id: str,
         *,
         wake_at: str | None = None,
+        base_revision: int | 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}
+        result = self.apply_batch(login, [{
+            "operation_id": operation_id,
+            "action": action,
+            "item_id": item_id,
+            "wake_at": wake_at,
+            "base_revision": base_revision,
+        }])
+        return {"revision": result["revision"], "records": result["records"]}
 
     def apply_batch(self, login: str, operations: list[dict]) -> dict:
         """Apply ordered deferrals with one SQLite write transaction."""
@@ -136,6 +108,20 @@ class LaterStore:
             revision = snapshot["revision"]
             accepted: list[str] = []
             duplicates: list[str] = []
+            rejected: list[dict[str, str]] = []
+
+            # Existing databases predate per-item revisions. Conservatively mark
+            # active deferrals as changed at the latest known plan revision.
+            for item_id in records:
+                connection.execute(
+                    "INSERT OR IGNORE INTO later_item_revisions(login, item_id, revision) VALUES (?, ?, ?)",
+                    (login, item_id, revision),
+                )
+            item_revisions = dict(connection.execute(
+                "SELECT item_id, revision FROM later_item_revisions WHERE login = ?",
+                (login,),
+            ).fetchall())
+            batch_start_item_revisions = dict(item_revisions)
 
             for operation in operations:
                 operation_id = operation.get("operation_id", "")
@@ -156,6 +142,22 @@ class LaterStore:
                     duplicates.append(operation_id)
                     continue
 
+                base_revision = operation.get("base_revision")
+                if base_revision is None:
+                    base_revision = snapshot["revision"]
+                if not isinstance(base_revision, int) or isinstance(base_revision, bool) or base_revision < 0:
+                    raise ValueError("base_revision must be a non-negative integer")
+                if batch_start_item_revisions.get(item_id, 0) > base_revision:
+                    connection.execute(
+                        "INSERT INTO later_operations(login, operation_id) VALUES (?, ?)",
+                        (login, operation_id),
+                    )
+                    rejected.append({
+                        "operation_id": operation_id,
+                        "reason": "stale_intent",
+                    })
+                    continue
+
                 before = records.get(item_id)
                 if action == "defer":
                     records[item_id] = wake_at
@@ -164,6 +166,13 @@ class LaterStore:
                     changed = item_id in records
                     records.pop(item_id, None)
                 revision += 1 if changed else 0
+                if changed:
+                    item_revisions[item_id] = revision
+                    connection.execute(
+                        "INSERT INTO later_item_revisions(login, item_id, revision) VALUES (?, ?, ?) "
+                        "ON CONFLICT(login, item_id) DO UPDATE SET revision = excluded.revision",
+                        (login, item_id, revision),
+                    )
                 connection.execute(
                     "INSERT INTO later_operations(login, operation_id) VALUES (?, ?)",
                     (login, operation_id),
@@ -186,5 +195,5 @@ class LaterStore:
                 "records": records,
                 "accepted_operation_ids": accepted,
                 "duplicate_operation_ids": duplicates,
-                "rejected_operations": [],
+                "rejected_operations": rejected,
             }

src/main.py

@@ -222,6 +222,7 @@ class LaterOperation(BaseModel):
     action: Literal["defer", "restore"]
     item_id: str = Field(min_length=1, max_length=500)
     wake_at: str | None = Field(default=None, max_length=100)
+    base_revision: int | None = Field(default=None, ge=0)
 
 
 class TodayOperationBatch(BaseModel):
@@ -969,6 +970,7 @@ async def update_later_plan(payload: LaterOperation | LaterOperationBatch):
             payload.action,
             payload.item_id,
             wake_at=payload.wake_at,
+            base_revision=payload.base_revision,
         )
     except ValueError as error:
         raise HTTPException(status_code=422, detail=str(error))