"""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, capacity_minutes INTEGER, estimates TEXT NOT NULL DEFAULT '{}' ) """ ) plan_columns = {row[1] for row in connection.execute("PRAGMA table_info(today_plans)")} if "capacity_minutes" not in plan_columns: connection.execute("ALTER TABLE today_plans ADD COLUMN capacity_minutes INTEGER") if "estimates" not in plan_columns: connection.execute("ALTER TABLE today_plans ADD COLUMN estimates TEXT NOT NULL DEFAULT '{}'") 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": [], "capacity_minutes": None, "estimates": {}} ids = json.loads(row[1]) estimates = json.loads(row[3] or "{}") return { "revision": int(row[0]), "ids": ids, "capacity_minutes": row[2], "estimates": {item_id: minutes for item_id, minutes in estimates.items() if item_id in ids}, } def get(self, login: str) -> dict: with self._connect() as connection: row = connection.execute( "SELECT revision, ids, capacity_minutes, estimates 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, capacity_minutes, estimates 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"]) estimates = dict(snapshot["estimates"]) 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) estimates.pop(item_id, None) 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) serialized_ids = json.dumps(ids, separators=(",", ":")) serialized_estimates = json.dumps(estimates, separators=(",", ":")) if row is None: connection.execute( "INSERT INTO today_plans(login, revision, ids, capacity_minutes, estimates) VALUES (?, ?, ?, ?, ?)", (login, revision, serialized_ids, snapshot["capacity_minutes"], serialized_estimates), ) elif changed: connection.execute( "UPDATE today_plans SET revision = ?, ids = ?, estimates = ? WHERE login = ?", (revision, serialized_ids, serialized_estimates, login), ) self._record_operation(connection, login, operation_id) return { "revision": revision, "ids": ids, "capacity_minutes": snapshot["capacity_minutes"], "estimates": estimates, } 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, capacity_minutes, estimates FROM today_plans WHERE login = ?", (login,) ).fetchone() snapshot = self._snapshot(row) ids = list(snapshot["ids"]) capacity_minutes = snapshot["capacity_minutes"] estimates = dict(snapshot["estimates"]) 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", "configure"}: 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) estimates.pop(item_id, None) changed = True elif action == "move": 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 else: proposed_capacity = operation.get("capacity_minutes") if proposed_capacity is not None and ( not isinstance(proposed_capacity, int) or isinstance(proposed_capacity, bool) or proposed_capacity < 15 or proposed_capacity > 1440 ): raise ValueError("capacity_minutes must be between 15 and 1440") proposed_estimates = operation.get("estimates", {}) if not isinstance(proposed_estimates, dict): raise ValueError("estimates must be an object") normalized_estimates = {} for estimate_id, minutes in proposed_estimates.items(): if estimate_id not in ids: continue if not isinstance(minutes, int) or isinstance(minutes, bool) or minutes < 5 or minutes > 1440: raise ValueError("estimate minutes must be between 5 and 1440") normalized_estimates[estimate_id] = minutes changed = capacity_minutes != proposed_capacity or estimates != normalized_estimates capacity_minutes = proposed_capacity estimates = normalized_estimates revision += 1 if changed else 0 self._record_operation(connection, login, operation_id) accepted.append(operation_id) serialized = json.dumps(ids, separators=(",", ":")) serialized_estimates = json.dumps(estimates, separators=(",", ":")) if row is None: connection.execute( "INSERT INTO today_plans(login, revision, ids, capacity_minutes, estimates) VALUES (?, ?, ?, ?, ?)", (login, revision, serialized, capacity_minutes, serialized_estimates), ) elif accepted: connection.execute( "UPDATE today_plans SET revision = ?, ids = ?, capacity_minutes = ?, estimates = ? WHERE login = ?", (revision, serialized, capacity_minutes, serialized_estimates, login), ) return { "revision": revision, "ids": ids, "capacity_minutes": capacity_minutes, "estimates": estimates, "accepted_operation_ids": accepted, "duplicate_operation_ids": duplicates, "rejected_operations": rejected, }