feat: batch queued planning edits (#375)

src/later_store.py

@@ -122,3 +122,69 @@ class LaterStore:
                 (login, operation_id),
             )
             return {"revision": revision, "records": records}
+
+    def apply_batch(self, login: str, operations: list[dict]) -> dict:
+        """Apply ordered deferrals with one SQLite write transaction."""
+        login = self._normalize_login(login)
+        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)
+            records = dict(snapshot["records"])
+            revision = snapshot["revision"]
+            accepted: list[str] = []
+            duplicates: list[str] = []
+
+            for operation in operations:
+                operation_id = operation.get("operation_id", "")
+                action = operation.get("action", "")
+                item_id = operation.get("item_id", "")
+                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")
+                wake_at = operation.get("wake_at")
+                if action == "defer":
+                    wake_at = self._validate_wake_at(wake_at)
+                duplicate = connection.execute(
+                    "SELECT 1 FROM later_operations WHERE login = ? AND operation_id = ?",
+                    (login, operation_id),
+                ).fetchone()
+                if duplicate:
+                    duplicates.append(operation_id)
+                    continue
+
+                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 += 1 if changed else 0
+                connection.execute(
+                    "INSERT INTO later_operations(login, operation_id) VALUES (?, ?)",
+                    (login, operation_id),
+                )
+                accepted.append(operation_id)
+
+            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 accepted:
+                connection.execute(
+                    "UPDATE later_plans SET revision = ?, records = ? WHERE login = ?",
+                    (revision, serialized, login),
+                )
+            return {
+                "revision": revision,
+                "records": records,
+                "accepted_operation_ids": accepted,
+                "duplicate_operation_ids": duplicates,
+                "rejected_operations": [],
+            }

src/main.py

@@ -224,6 +224,14 @@ class LaterOperation(BaseModel):
     wake_at: str | None = Field(default=None, max_length=100)
 
 
+class TodayOperationBatch(BaseModel):
+    operations: list[TodayOperation] = Field(min_length=1, max_length=50)
+
+
+class LaterOperationBatch(BaseModel):
+    operations: list[LaterOperation] = Field(min_length=1, max_length=50)
+
+
 class NotificationReply(BaseModel):
     body: str = Field(min_length=1, max_length=10_000)
 
@@ -904,9 +912,15 @@ async def get_today_plan():
 
 
 @app.patch("/api/v1/today")
-async def update_today_plan(payload: TodayOperation):
+async def update_today_plan(payload: TodayOperation | TodayOperationBatch):
     login = await _confirmed_login()
     try:
+        if isinstance(payload, TodayOperationBatch):
+            return await asyncio.to_thread(
+                _today_store().apply_batch,
+                login,
+                [operation.model_dump() for operation in payload.operations],
+            )
         return await asyncio.to_thread(
             _today_store().apply,
             login,
@@ -939,9 +953,15 @@ async def get_later_plan():
 
 
 @app.patch("/api/v1/later")
-async def update_later_plan(payload: LaterOperation):
+async def update_later_plan(payload: LaterOperation | LaterOperationBatch):
     login = await _confirmed_login()
     try:
+        if isinstance(payload, LaterOperationBatch):
+            return await asyncio.to_thread(
+                _later_store().apply_batch,
+                login,
+                [operation.model_dump() for operation in payload.operations],
+            )
         return await asyncio.to_thread(
             _later_store().apply,
             login,

src/today_store.py

@@ -128,3 +128,84 @@ class TodayStore:
                 (login, operation_id),
             )
             return {"revision": revision, "ids": ids}
+
+    def apply_batch(self, login: str, operations: list[dict]) -> dict:
+        """Apply an ordered batch with one lock and receipt per operation."""
+        login = self._normalize_login(login)
+        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)
+            ids = list(snapshot["ids"])
+            revision = snapshot["revision"]
+            accepted: list[str] = []
+            duplicates: list[str] = []
+            rejected: list[dict[str, str]] = []
+
+            for operation in operations:
+                operation_id = operation.get("operation_id", "")
+                action = operation.get("action", "")
+                item_id = operation.get("item_id", "")
+                direction = operation.get("direction")
+                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")
+                duplicate = connection.execute(
+                    "SELECT 1 FROM today_operations WHERE login = ? AND operation_id = ?",
+                    (login, operation_id),
+                ).fetchone()
+                if duplicate:
+                    duplicates.append(operation_id)
+                    continue
+
+                changed = False
+                if action == "add":
+                    if item_id not in ids:
+                        if len(ids) >= self.limit:
+                            rejected.append({"operation_id": operation_id, "reason": "today_full"})
+                            continue
+                        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 += 1 if changed else 0
+                connection.execute(
+                    "INSERT INTO today_operations(login, operation_id) VALUES (?, ?)",
+                    (login, operation_id),
+                )
+                accepted.append(operation_id)
+
+            serialized = json.dumps(ids, separators=(",", ":"))
+            if row is None:
+                connection.execute(
+                    "INSERT INTO today_plans(login, revision, ids) VALUES (?, ?, ?)",
+                    (login, revision, serialized),
+                )
+            elif accepted:
+                connection.execute(
+                    "UPDATE today_plans SET revision = ?, ids = ? WHERE login = ?",
+                    (revision, serialized, login),
+                )
+            return {
+                "revision": revision,
+                "ids": ids,
+                "accepted_operation_ids": accepted,
+                "duplicate_operation_ids": duplicates,
+                "rejected_operations": rejected,
+            }