"""Durable, account-scoped ordered Today plans.""" import json import sqlite3 import time 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, 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) 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, 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() 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), ) self._record_operation(connection, 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 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: 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 self._record_operation(connection, 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, }