feat: bound planning sync history (#381)

src/later_store.py

@@ -2,14 +2,26 @@
 
 import json
 import sqlite3
+import time
 from datetime import datetime
 from pathlib import Path
 
 
 class LaterStore:
-    def __init__(self, path: str | Path, *, timeout: float = 1.0):
+    def __init__(
+        self,
+        path: str | Path,
+        *,
+        timeout: float = 1.0,
+        operation_limit: int = 4096,
+        operation_retention_seconds: float = 30 * 24 * 60 * 60,
+        clock=time.time,
+    ):
         self.path = Path(path)
         self.timeout = timeout
+        self.operation_limit = operation_limit
+        self.operation_retention_seconds = operation_retention_seconds
+        self.clock = clock
 
     def _connect(self) -> sqlite3.Connection:
         self.path.parent.mkdir(parents=True, exist_ok=True)
@@ -29,6 +41,7 @@ class LaterStore:
             CREATE TABLE IF NOT EXISTS later_operations (
                 login TEXT NOT NULL,
                 operation_id TEXT NOT NULL,
+                created_at REAL NOT NULL,
                 PRIMARY KEY (login, operation_id)
             )
             """
@@ -43,8 +56,33 @@ class LaterStore:
             )
             """
         )
+        columns = {row[1] for row in connection.execute("PRAGMA table_info(later_operations)")}
+        if "created_at" not in columns:
+            connection.execute("ALTER TABLE later_operations ADD COLUMN created_at REAL")
+        connection.execute(
+            "UPDATE later_operations SET created_at = ? WHERE created_at IS NULL",
+            (self.clock(),),
+        )
+        connection.commit()
         return connection
 
+    def _record_operation(self, connection: sqlite3.Connection, login: str, operation_id: str) -> None:
+        now = self.clock()
+        connection.execute(
+            "INSERT INTO later_operations(login, operation_id, created_at) VALUES (?, ?, ?)",
+            (login, operation_id, now),
+        )
+        connection.execute(
+            "DELETE FROM later_operations WHERE login = ? AND created_at < ?",
+            (login, now - self.operation_retention_seconds),
+        )
+        connection.execute(
+            "DELETE FROM later_operations WHERE login = ? AND rowid NOT IN "
+            "(SELECT rowid FROM later_operations WHERE login = ? "
+            "ORDER BY created_at DESC, rowid DESC LIMIT ?)",
+            (login, login, self.operation_limit),
+        )
+
     @staticmethod
     def _normalize_login(login: str) -> str:
         normalized = login.strip().lower()
@@ -148,10 +186,7 @@ class LaterStore:
                 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),
-                    )
+                    self._record_operation(connection, login, operation_id)
                     rejected.append({
                         "operation_id": operation_id,
                         "reason": "stale_intent",
@@ -173,10 +208,7 @@ class LaterStore:
                         "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),
-                )
+                self._record_operation(connection, login, operation_id)
                 accepted.append(operation_id)
 
             serialized = json.dumps(records, separators=(",", ":"), sort_keys=True)

src/main.py

@@ -207,6 +207,7 @@ class TodayOperation(BaseModel):
     action: Literal["add", "remove", "move"]
     item_id: str = Field(min_length=1, max_length=500)
     direction: Literal["up", "down"] | None = None
+    base_revision: int | None = Field(default=None, ge=0)
 
     @model_validator(mode="after")
     def require_move_direction(self):

src/today_store.py

@@ -2,6 +2,7 @@
 
 import json
 import sqlite3
+import time
 from pathlib import Path
 
 
@@ -10,10 +11,22 @@ class TodayPlanFull(ValueError):
 
 
 class TodayStore:
-    def __init__(self, path: str | Path, *, limit: int = 5, timeout: float = 1.0):
+    def __init__(
+        self,
+        path: str | Path,
+        *,
+        limit: int = 5,
+        timeout: float = 1.0,
+        operation_limit: int = 4096,
+        operation_retention_seconds: float = 30 * 24 * 60 * 60,
+        clock=time.time,
+    ):
         self.path = Path(path)
         self.limit = limit
         self.timeout = timeout
+        self.operation_limit = operation_limit
+        self.operation_retention_seconds = operation_retention_seconds
+        self.clock = clock
 
     def _connect(self) -> sqlite3.Connection:
         self.path.parent.mkdir(parents=True, exist_ok=True)
@@ -33,12 +46,38 @@ class TodayStore:
             CREATE TABLE IF NOT EXISTS today_operations (
                 login TEXT NOT NULL,
                 operation_id TEXT NOT NULL,
+                created_at REAL NOT NULL,
                 PRIMARY KEY (login, operation_id)
             )
             """
         )
+        columns = {row[1] for row in connection.execute("PRAGMA table_info(today_operations)")}
+        if "created_at" not in columns:
+            connection.execute("ALTER TABLE today_operations ADD COLUMN created_at REAL")
+        connection.execute(
+            "UPDATE today_operations SET created_at = ? WHERE created_at IS NULL",
+            (self.clock(),),
+        )
+        connection.commit()
         return connection
 
+    def _record_operation(self, connection: sqlite3.Connection, login: str, operation_id: str) -> None:
+        now = self.clock()
+        connection.execute(
+            "INSERT INTO today_operations(login, operation_id, created_at) VALUES (?, ?, ?)",
+            (login, operation_id, now),
+        )
+        connection.execute(
+            "DELETE FROM today_operations WHERE login = ? AND created_at < ?",
+            (login, now - self.operation_retention_seconds),
+        )
+        connection.execute(
+            "DELETE FROM today_operations WHERE login = ? AND rowid NOT IN "
+            "(SELECT rowid FROM today_operations WHERE login = ? "
+            "ORDER BY created_at DESC, rowid DESC LIMIT ?)",
+            (login, login, self.operation_limit),
+        )
+
     @staticmethod
     def _normalize_login(login: str) -> str:
         normalized = login.strip().lower()
@@ -123,10 +162,7 @@ class TodayStore:
                     "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),
-            )
+            self._record_operation(connection, login, operation_id)
             return {"revision": revision, "ids": ids}
 
     def apply_batch(self, login: str, operations: list[dict]) -> dict:
@@ -163,6 +199,16 @@ class TodayStore:
                     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 base_revision < snapshot["revision"]:
+                    self._record_operation(connection, login, operation_id)
+                    rejected.append({"operation_id": operation_id, "reason": "stale_intent"})
+                    continue
+
                 changed = False
                 if action == "add":
                     if item_id not in ids:
@@ -185,10 +231,7 @@ class TodayStore:
                         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),
-                )
+                self._record_operation(connection, login, operation_id)
                 accepted.append(operation_id)
 
             serialized = json.dumps(ids, separators=(",", ":"))