"""Durable, account-scoped Later deferrals.""" 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, 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 self._initialize() def _initialize(self) -> None: self.path.parent.mkdir(parents=True, exist_ok=True) connection = sqlite3.connect(self.path, timeout=self.timeout) if connection.execute("PRAGMA user_version").fetchone()[0] >= 1: connection.close() return connection.execute("PRAGMA journal_mode=WAL") connection.execute( """ CREATE TABLE IF NOT EXISTS later_plans ( login TEXT PRIMARY KEY, revision INTEGER NOT NULL, records TEXT NOT NULL ) """ ) connection.execute( """ 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) ) """ ) 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) ) """ ) 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.execute("PRAGMA user_version = 1") connection.commit() connection.close() def _connect(self) -> sqlite3.Connection: return sqlite3.connect(self.path, timeout=self.timeout) 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() if not normalized: raise ValueError("login is required") return normalized @staticmethod def _snapshot(row) -> dict: if row is None: return {"revision": 0, "records": {}} return {"revision": int(row[0]), "records": json.loads(row[1])} @staticmethod def _validate_wake_at(wake_at: str | None) -> str: if not wake_at: raise ValueError("wake_at is required for defer") try: datetime.fromisoformat(wake_at.replace("Z", "+00:00")) except ValueError as error: raise ValueError("wake_at must be an ISO timestamp") from error return wake_at def get(self, login: str) -> dict: with self._connect() as connection: row = connection.execute( "SELECT revision, records FROM later_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, *, wake_at: str | None = None, base_revision: int | None = None, ) -> dict: 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.""" 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] = [] 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", "") 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 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: self._record_operation(connection, 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 changed = before != wake_at else: 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), ) self._record_operation(connection, 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": rejected, }