"""Durable, account-scoped Later deferrals.""" import sqlite3 import time from datetime import datetime from pathlib import Path from src.private_state import connect_private_sqlite from src.state_encryption import PrivateStateCipher, PrivateStateEncryptionError, private_state_encryption_key class LaterStore: def __init__( self, path: str | Path, *, timeout: float = 1.0, operation_limit: int = 4096, operation_retention_seconds: float = 30 * 24 * 60 * 60, encryption_key: bytes | None = None, 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._cipher = PrivateStateCipher( encryption_key if encryption_key is not None else private_state_encryption_key(), store="later", ) self._initialize() def _initialize(self) -> None: connection = connect_private_sqlite(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 connect_private_sqlite(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 def _snapshot(self, row, login: str) -> tuple[dict, bool]: if row is None: return {"revision": 0, "records": {}}, False records, legacy = self._cipher.open(row[1], binding=f"plan:{login}") if not isinstance(records, dict): raise PrivateStateEncryptionError("private state could not be decrypted") return {"revision": int(row[0]), "records": records}, legacy def _sealed_records(self, login: str, records: dict) -> str: return self._cipher.seal(records, binding=f"plan:{login}") def _item_revisions( self, connection: sqlite3.Connection, login: str ) -> tuple[dict[str, int], dict[str, str]]: revisions: dict[str, int] = {} stored_ids: dict[str, str] = {} for stored_item_id, revision in connection.execute( "SELECT item_id, revision FROM later_item_revisions WHERE login = ?", (login,), ): if stored_item_id.startswith("v1:"): item_id, _legacy = self._cipher.open( stored_item_id, binding=f"item-revision:{login}" ) else: item_id = stored_item_id if not isinstance(item_id, str) or not item_id: raise PrivateStateEncryptionError("private state could not be decrypted") revisions[item_id] = int(revision) stored_ids[item_id] = stored_item_id return revisions, stored_ids def _migrate_item_ids( self, connection: sqlite3.Connection, login: str, stored_ids: dict[str, str] ) -> dict[str, str]: migrated = dict(stored_ids) for item_id, stored_item_id in stored_ids.items(): if stored_item_id.startswith("v1:"): continue sealed_item_id = self._cipher.seal( item_id, binding=f"item-revision:{login}" ) connection.execute( "UPDATE later_item_revisions SET item_id = ? " "WHERE login = ? AND item_id = ?", (sealed_item_id, login, stored_item_id), ) migrated[item_id] = sealed_item_id return migrated @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: login = self._normalize_login(login) with self._connect() as connection: row = connection.execute( "SELECT revision, records FROM later_plans WHERE login = ?", (login,), ).fetchone() snapshot, legacy_plan = self._snapshot(row, login) _revisions, stored_item_ids = self._item_revisions(connection, login) if row is not None and legacy_plan: connection.execute( "UPDATE later_plans SET records = ? WHERE login = ? AND records = ?", (self._sealed_records(login, snapshot["records"]), login, row[1]), ) self._migrate_item_ids(connection, login, stored_item_ids) return snapshot def apply( self, login: str, operation_id: str, action: str, item_id: str, *, wake_at: str | None = None, handoff: 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, "handoff": handoff, "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, legacy_plan = self._snapshot(row, login) 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. item_revisions, stored_item_ids = self._item_revisions(connection, login) stored_item_ids = self._migrate_item_ids(connection, login, stored_item_ids) for item_id in records: if item_id not in item_revisions: stored_item_id = self._cipher.seal( item_id, binding=f"item-revision:{login}" ) connection.execute( "INSERT INTO later_item_revisions(login, item_id, revision) VALUES (?, ?, ?)", (login, stored_item_id, revision), ) item_revisions[item_id] = revision stored_item_ids[item_id] = stored_item_id 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) handoff = operation.get("handoff") if handoff not in {None, "today"}: raise ValueError("unsupported Later handoff") 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": record = {"wake_at": wake_at, "handoff": handoff} if handoff else wake_at records[item_id] = record changed = before != record else: changed = item_id in records records.pop(item_id, None) revision += 1 if changed else 0 if changed: item_revisions[item_id] = revision stored_item_id = stored_item_ids.get(item_id) if stored_item_id is None: stored_item_id = self._cipher.seal( item_id, binding=f"item-revision:{login}" ) stored_item_ids[item_id] = stored_item_id connection.execute( "INSERT INTO later_item_revisions(login, item_id, revision) VALUES (?, ?, ?)", (login, stored_item_id, revision), ) else: connection.execute( "UPDATE later_item_revisions SET revision = ? " "WHERE login = ? AND item_id = ?", (revision, login, stored_item_id), ) self._record_operation(connection, login, operation_id) accepted.append(operation_id) serialized = self._sealed_records(login, records) if row is None: connection.execute( "INSERT INTO later_plans(login, revision, records) VALUES (?, ?, ?)", (login, revision, serialized), ) elif accepted or legacy_plan: 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, }