1579 lines
78 KiB
Python
1579 lines
78 KiB
Python
"""Durable, account-scoped ordered Today plans."""
|
|
|
|
import json
|
|
import sqlite3
|
|
import time
|
|
from datetime import date, datetime
|
|
from pathlib import Path
|
|
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
|
|
|
|
from src.private_state import connect_private_sqlite
|
|
from src.state_encryption import PrivateStateCipher, PrivateStateEncryptionError, private_state_encryption_config
|
|
|
|
|
|
class TodayPlanFull(ValueError):
|
|
"""Raised when an add would exceed the bounded Today plan."""
|
|
|
|
|
|
class TodaySessionConflict(ValueError):
|
|
"""Raised when a device updates an obsolete active-session revision."""
|
|
|
|
def __init__(self, session: dict):
|
|
super().__init__("active Today session changed on another device")
|
|
self.session = session
|
|
|
|
|
|
class TomorrowPlanConflict(ValueError):
|
|
"""Raised when Tomorrow was edited from an obsolete revision."""
|
|
|
|
def __init__(self, snapshot: dict):
|
|
super().__init__("Tomorrow plan changed on another device")
|
|
self.snapshot = snapshot
|
|
|
|
|
|
class TomorrowPlanNotDue(ValueError):
|
|
"""Raised when Tomorrow is promoted before its saved local date."""
|
|
|
|
def __init__(self, plan_date: str):
|
|
super().__init__("Tomorrow plan is not due")
|
|
self.plan_date = plan_date
|
|
|
|
|
|
class WeekPlanConflict(ValueError):
|
|
"""Raised when Week Ahead was edited from an obsolete revision."""
|
|
|
|
def __init__(self, snapshot: dict):
|
|
super().__init__("Week Ahead changed on another device")
|
|
self.snapshot = snapshot
|
|
|
|
|
|
class TodayPromotionConflict(ValueError):
|
|
"""Raised when promotion cannot safely replace the current Today plan."""
|
|
|
|
def __init__(self, today: dict, week: dict | None = None):
|
|
super().__init__("Today must be reviewed before a saved plan can be promoted")
|
|
self.today = today
|
|
self.week = week
|
|
|
|
|
|
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,
|
|
recap_limit: int = 100,
|
|
encryption_key: bytes | None = None,
|
|
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.recap_limit = recap_limit
|
|
self.clock = clock
|
|
self._cipher = PrivateStateCipher(
|
|
encryption_key if encryption_key is not None else private_state_encryption_config(),
|
|
store="today",
|
|
)
|
|
self._initialize()
|
|
|
|
def _initialize(self) -> None:
|
|
connection = connect_private_sqlite(self.path, timeout=self.timeout)
|
|
if connection.execute("PRAGMA user_version").fetchone()[0] >= 7:
|
|
connection.close()
|
|
return
|
|
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_date TEXT,
|
|
timezone TEXT
|
|
)
|
|
"""
|
|
)
|
|
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 '{}'")
|
|
if "plan_date" not in plan_columns:
|
|
connection.execute("ALTER TABLE today_plans ADD COLUMN plan_date TEXT")
|
|
if "timezone" not in plan_columns:
|
|
connection.execute("ALTER TABLE today_plans ADD COLUMN timezone TEXT")
|
|
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.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS today_recaps (
|
|
login TEXT NOT NULL,
|
|
session_id TEXT NOT NULL,
|
|
created_at REAL NOT NULL,
|
|
items TEXT NOT NULL,
|
|
PRIMARY KEY (login, session_id)
|
|
)
|
|
"""
|
|
)
|
|
connection.execute(
|
|
"CREATE INDEX IF NOT EXISTS today_recaps_recent "
|
|
"ON today_recaps(login, created_at DESC)"
|
|
)
|
|
connection.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS today_time_logs (
|
|
login TEXT NOT NULL,
|
|
session_id TEXT NOT NULL,
|
|
identity TEXT NOT NULL,
|
|
actual_minutes INTEGER NOT NULL,
|
|
status TEXT NOT NULL,
|
|
PRIMARY KEY (login, session_id, identity)
|
|
)
|
|
"""
|
|
)
|
|
connection.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS today_sessions (
|
|
login TEXT PRIMARY KEY,
|
|
revision INTEGER NOT NULL,
|
|
device_id TEXT NOT NULL,
|
|
identity TEXT NOT NULL,
|
|
elapsed_ms INTEGER NOT NULL,
|
|
running INTEGER NOT NULL,
|
|
break_deadline_at INTEGER,
|
|
updated_at REAL NOT NULL
|
|
)
|
|
"""
|
|
)
|
|
session_columns = {row[1] for row in connection.execute("PRAGMA table_info(today_sessions)")}
|
|
if "break_deadline_at" not in session_columns:
|
|
connection.execute("ALTER TABLE today_sessions ADD COLUMN break_deadline_at INTEGER")
|
|
connection.execute(
|
|
"CREATE TABLE IF NOT EXISTS tomorrow_plans ("
|
|
"login TEXT PRIMARY KEY, revision INTEGER NOT NULL, payload TEXT NOT NULL)"
|
|
)
|
|
connection.execute(
|
|
"CREATE TABLE IF NOT EXISTS tomorrow_promotions ("
|
|
"login TEXT NOT NULL, promotion_id TEXT NOT NULL, result TEXT NOT NULL, "
|
|
"created_at REAL NOT NULL, PRIMARY KEY (login, promotion_id))"
|
|
)
|
|
connection.execute(
|
|
"CREATE TABLE IF NOT EXISTS week_plans ("
|
|
"login TEXT PRIMARY KEY, revision INTEGER NOT NULL, payload TEXT NOT NULL)"
|
|
)
|
|
connection.execute(
|
|
"CREATE TABLE IF NOT EXISTS week_promotions ("
|
|
"login TEXT NOT NULL, promotion_id TEXT NOT NULL, result TEXT NOT NULL, "
|
|
"created_at REAL NOT NULL, PRIMARY KEY (login, promotion_id))"
|
|
)
|
|
connection.execute(
|
|
"CREATE TABLE IF NOT EXISTS week_reschedules ("
|
|
"login TEXT NOT NULL, operation_id TEXT NOT NULL, result TEXT NOT NULL, "
|
|
"created_at REAL NOT NULL, PRIMARY KEY (login, operation_id))"
|
|
)
|
|
connection.execute(
|
|
"CREATE TABLE IF NOT EXISTS week_item_pulls ("
|
|
"login TEXT NOT NULL, operation_id TEXT NOT NULL, result TEXT NOT NULL, "
|
|
"created_at REAL NOT NULL, PRIMARY KEY (login, operation_id))"
|
|
)
|
|
connection.execute("PRAGMA user_version = 7")
|
|
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 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
|
|
|
|
def _snapshot(self, row, login: str) -> tuple[dict, bool]:
|
|
if row is None:
|
|
return {"revision": 0, "ids": [], "capacity_minutes": None, "estimates": {}}, False
|
|
if row[1].startswith(("v1:", "v2:")):
|
|
payload, legacy = self._cipher.open(row[1], binding=f"plan:{login}")
|
|
if not isinstance(payload, dict):
|
|
raise PrivateStateEncryptionError("private state could not be decrypted")
|
|
ids = payload.get("ids")
|
|
estimates = payload.get("estimates")
|
|
capacity_minutes = payload.get("capacity_minutes")
|
|
plan_date = payload.get("plan_date")
|
|
timezone = payload.get("timezone")
|
|
first_task_state = payload.get("first_task_state", "")
|
|
else:
|
|
legacy = True
|
|
ids = json.loads(row[1])
|
|
estimates = json.loads(row[3] or "{}")
|
|
capacity_minutes = row[2]
|
|
plan_date = row[4] if len(row) > 4 else None
|
|
timezone = row[5] if len(row) > 5 else None
|
|
first_task_state = ""
|
|
if not isinstance(ids, list) or not isinstance(estimates, dict):
|
|
raise PrivateStateEncryptionError("private state could not be decrypted")
|
|
snapshot = {
|
|
"revision": int(row[0]),
|
|
"ids": ids,
|
|
"capacity_minutes": capacity_minutes,
|
|
"estimates": {item_id: minutes for item_id, minutes in estimates.items() if item_id in ids},
|
|
}
|
|
if plan_date:
|
|
snapshot["plan_date"] = plan_date
|
|
snapshot["timezone"] = timezone
|
|
if first_task_state in {"coaching", "complete"}:
|
|
snapshot["first_task_state"] = first_task_state
|
|
return snapshot, legacy
|
|
|
|
def _sealed_plan(self, login: str, snapshot: dict) -> str:
|
|
return self._cipher.seal({
|
|
"ids": snapshot["ids"],
|
|
"capacity_minutes": snapshot["capacity_minutes"],
|
|
"estimates": snapshot["estimates"],
|
|
"plan_date": snapshot.get("plan_date"),
|
|
"timezone": snapshot.get("timezone"),
|
|
"first_task_state": snapshot.get("first_task_state", ""),
|
|
}, binding=f"plan:{login}")
|
|
|
|
def get(self, login: str) -> dict:
|
|
login = self._normalize_login(login)
|
|
with self._connect() as connection:
|
|
row = connection.execute(
|
|
"SELECT revision, ids, capacity_minutes, estimates, plan_date, timezone "
|
|
"FROM today_plans WHERE login = ?",
|
|
(login,),
|
|
).fetchone()
|
|
snapshot, legacy = self._snapshot(row, login)
|
|
if row is not None and legacy:
|
|
connection.execute(
|
|
"UPDATE today_plans SET ids = ?, capacity_minutes = NULL, estimates = '{}', "
|
|
"plan_date = NULL, timezone = NULL WHERE login = ? AND ids = ?",
|
|
(self._sealed_plan(login, snapshot), login, row[1]),
|
|
)
|
|
return snapshot
|
|
|
|
@staticmethod
|
|
def _empty_tomorrow(revision: int = 0) -> dict:
|
|
return {"revision": revision, "ids": [], "capacity_minutes": None, "estimates": {}}
|
|
|
|
def _tomorrow_snapshot(self, row, login: str) -> dict:
|
|
if row is None:
|
|
return self._empty_tomorrow()
|
|
payload, _legacy = self._cipher.open(row[1], binding=f"tomorrow:{login}")
|
|
if not isinstance(payload, dict):
|
|
raise PrivateStateEncryptionError("private state could not be decrypted")
|
|
snapshot = {"revision": int(row[0]), **payload}
|
|
if not isinstance(snapshot.get("ids"), list) or not isinstance(snapshot.get("estimates"), dict):
|
|
raise PrivateStateEncryptionError("private state could not be decrypted")
|
|
return snapshot
|
|
|
|
def get_tomorrow(self, login: str) -> dict:
|
|
login = self._normalize_login(login)
|
|
with self._connect() as connection:
|
|
row = connection.execute(
|
|
"SELECT revision, payload FROM tomorrow_plans WHERE login = ?", (login,)
|
|
).fetchone()
|
|
return self._tomorrow_snapshot(row, login)
|
|
|
|
def _normalize_tomorrow(
|
|
self, *, ids: list[str], capacity_minutes: int | None,
|
|
estimates: dict[str, int], plan_date: str, timezone: str,
|
|
) -> dict:
|
|
try:
|
|
if date.fromisoformat(plan_date or "").isoformat() != plan_date:
|
|
raise ValueError
|
|
except (TypeError, ValueError):
|
|
raise ValueError("plan_date must be an ISO calendar date") from None
|
|
if not isinstance(timezone, str) or not timezone.strip() or len(timezone) > 100:
|
|
raise ValueError("timezone is required and bounded")
|
|
timezone = timezone.strip()
|
|
try:
|
|
ZoneInfo(timezone)
|
|
except (ZoneInfoNotFoundError, ValueError):
|
|
raise ValueError("timezone must be a valid IANA timezone") from None
|
|
if not isinstance(ids, list) or len(ids) > self.limit or any(
|
|
not isinstance(item_id, str) or not item_id or len(item_id) > 500 for item_id in ids
|
|
):
|
|
raise ValueError("Tomorrow IDs are invalid or exceed the plan limit")
|
|
if len(set(ids)) != len(ids):
|
|
raise ValueError("Tomorrow IDs must be unique")
|
|
if capacity_minutes is not None and (
|
|
not isinstance(capacity_minutes, int) or isinstance(capacity_minutes, bool)
|
|
or capacity_minutes < 15 or capacity_minutes > 1440
|
|
):
|
|
raise ValueError("capacity_minutes must be between 15 and 1440")
|
|
if not isinstance(estimates, dict):
|
|
raise ValueError("estimates must be an object")
|
|
normalized_estimates = {}
|
|
for item_id, minutes in estimates.items():
|
|
if item_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[item_id] = minutes
|
|
return {
|
|
"ids": list(ids), "capacity_minutes": capacity_minutes,
|
|
"estimates": normalized_estimates, "plan_date": plan_date,
|
|
"timezone": timezone.strip(),
|
|
}
|
|
|
|
def replace_tomorrow(self, login: str, *, base_revision: int, **plan) -> dict:
|
|
login = self._normalize_login(login)
|
|
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")
|
|
normalized = self._normalize_tomorrow(**plan)
|
|
with self._connect() as connection:
|
|
connection.execute("BEGIN IMMEDIATE")
|
|
row = connection.execute(
|
|
"SELECT revision, payload FROM tomorrow_plans WHERE login = ?", (login,)
|
|
).fetchone()
|
|
current = self._tomorrow_snapshot(row, login)
|
|
if current["revision"] != base_revision:
|
|
raise TomorrowPlanConflict(current)
|
|
snapshot = {"revision": base_revision + 1, **normalized}
|
|
sealed = self._cipher.seal(normalized, binding=f"tomorrow:{login}")
|
|
connection.execute(
|
|
"INSERT INTO tomorrow_plans(login, revision, payload) VALUES (?, ?, ?) "
|
|
"ON CONFLICT(login) DO UPDATE SET revision=excluded.revision, payload=excluded.payload",
|
|
(login, snapshot["revision"], sealed),
|
|
)
|
|
return snapshot
|
|
|
|
def promote_tomorrow(
|
|
self, login: str, *, promotion_id: str, tomorrow_revision: int, today_revision: int,
|
|
) -> dict:
|
|
login = self._normalize_login(login)
|
|
if not isinstance(promotion_id, str) or not promotion_id.strip() or len(promotion_id) > 100:
|
|
raise ValueError("promotion_id is required and bounded")
|
|
promotion_id = promotion_id.strip()
|
|
with self._connect() as connection:
|
|
connection.execute("BEGIN IMMEDIATE")
|
|
receipt = connection.execute(
|
|
"SELECT result FROM tomorrow_promotions WHERE login = ? AND promotion_id = ?",
|
|
(login, promotion_id),
|
|
).fetchone()
|
|
if receipt:
|
|
result, _legacy = self._cipher.open(
|
|
receipt[0], binding=f"tomorrow-promotion:{login}:{promotion_id}"
|
|
)
|
|
if not isinstance(result, dict):
|
|
raise PrivateStateEncryptionError("private state could not be decrypted")
|
|
return result
|
|
tomorrow_row = connection.execute(
|
|
"SELECT revision, payload FROM tomorrow_plans WHERE login = ?", (login,)
|
|
).fetchone()
|
|
tomorrow = self._tomorrow_snapshot(tomorrow_row, login)
|
|
if tomorrow["revision"] != tomorrow_revision or not tomorrow.get("plan_date"):
|
|
raise TomorrowPlanConflict(tomorrow)
|
|
local_date = datetime.fromtimestamp(
|
|
self.clock(), ZoneInfo(tomorrow["timezone"])
|
|
).date().isoformat()
|
|
if local_date < tomorrow["plan_date"]:
|
|
raise TomorrowPlanNotDue(tomorrow["plan_date"])
|
|
today_row = connection.execute(
|
|
"SELECT revision, ids, capacity_minutes, estimates, plan_date, timezone "
|
|
"FROM today_plans WHERE login = ?", (login,)
|
|
).fetchone()
|
|
today, _legacy = self._snapshot(today_row, login)
|
|
if today["revision"] != today_revision:
|
|
raise TodayPromotionConflict(today)
|
|
result = {"revision": today_revision + 1, **{
|
|
key: tomorrow[key] for key in
|
|
("ids", "capacity_minutes", "estimates", "plan_date", "timezone")
|
|
}}
|
|
sealed_today = self._sealed_plan(login, result)
|
|
connection.execute(
|
|
"INSERT INTO today_plans(login, revision, ids, capacity_minutes, estimates, plan_date, timezone) "
|
|
"VALUES (?, ?, ?, NULL, '{}', NULL, NULL) ON CONFLICT(login) DO UPDATE SET "
|
|
"revision=excluded.revision, ids=excluded.ids, capacity_minutes=NULL, estimates='{}', "
|
|
"plan_date=NULL, timezone=NULL",
|
|
(login, result["revision"], sealed_today),
|
|
)
|
|
empty = self._empty_tomorrow(tomorrow_revision + 1)
|
|
sealed_empty = self._cipher.seal(
|
|
{key: empty[key] for key in ("ids", "capacity_minutes", "estimates")},
|
|
binding=f"tomorrow:{login}",
|
|
)
|
|
connection.execute(
|
|
"UPDATE tomorrow_plans SET revision = ?, payload = ? WHERE login = ?",
|
|
(empty["revision"], sealed_empty, login),
|
|
)
|
|
sealed_result = self._cipher.seal(
|
|
result, binding=f"tomorrow-promotion:{login}:{promotion_id}"
|
|
)
|
|
connection.execute(
|
|
"INSERT INTO tomorrow_promotions(login, promotion_id, result, created_at) VALUES (?, ?, ?, ?)",
|
|
(login, promotion_id, sealed_result, self.clock()),
|
|
)
|
|
return result
|
|
|
|
@staticmethod
|
|
def _empty_week(revision: int = 0) -> dict:
|
|
return {"revision": revision, "timezone": None, "days": []}
|
|
|
|
def _week_snapshot(self, row, login: str) -> dict:
|
|
if row is None:
|
|
return self._empty_week()
|
|
payload, _legacy = self._cipher.open(row[1], binding=f"week:{login}")
|
|
if not isinstance(payload, dict) or not isinstance(payload.get("days"), list):
|
|
raise PrivateStateEncryptionError("private state could not be decrypted")
|
|
return {"revision": int(row[0]), **payload}
|
|
|
|
def get_start_day_plan(self, login: str) -> dict:
|
|
"""Return the next private planned day without changing either plan."""
|
|
login = self._normalize_login(login)
|
|
with self._connect() as connection:
|
|
week_row = connection.execute(
|
|
"SELECT revision, payload FROM week_plans WHERE login = ?", (login,)
|
|
).fetchone()
|
|
if week_row is not None:
|
|
week = self._week_snapshot(week_row, login)
|
|
local_date = datetime.fromtimestamp(
|
|
self.clock(), ZoneInfo(week["timezone"])
|
|
).date().isoformat()
|
|
for day in week["days"]:
|
|
if (
|
|
isinstance(day, dict)
|
|
and isinstance(day.get("ids"), list)
|
|
and day["ids"]
|
|
and day.get("plan_date", "") <= local_date
|
|
):
|
|
return {**day, "timezone": week.get("timezone")}
|
|
tomorrow_row = connection.execute(
|
|
"SELECT revision, payload FROM tomorrow_plans WHERE login = ?", (login,)
|
|
).fetchone()
|
|
tomorrow = self._tomorrow_snapshot(tomorrow_row, login)
|
|
return tomorrow if tomorrow.get("ids") and tomorrow.get("plan_date") else {}
|
|
|
|
def get_week(self, login: str) -> dict:
|
|
login = self._normalize_login(login)
|
|
with self._connect() as connection:
|
|
connection.execute("BEGIN IMMEDIATE")
|
|
row = connection.execute(
|
|
"SELECT revision, payload FROM week_plans WHERE login = ?", (login,)
|
|
).fetchone()
|
|
if row is not None:
|
|
return self._week_snapshot(row, login)
|
|
tomorrow_row = connection.execute(
|
|
"SELECT revision, payload FROM tomorrow_plans WHERE login = ?", (login,)
|
|
).fetchone()
|
|
tomorrow = self._tomorrow_snapshot(tomorrow_row, login)
|
|
if not tomorrow.get("plan_date"):
|
|
return self._empty_week()
|
|
day = {key: tomorrow[key] for key in (
|
|
"plan_date", "ids", "capacity_minutes", "estimates"
|
|
)}
|
|
payload = {"timezone": tomorrow["timezone"], "days": [day]}
|
|
connection.execute(
|
|
"INSERT INTO week_plans(login, revision, payload) VALUES (?, 1, ?)",
|
|
(login, self._cipher.seal(payload, binding=f"week:{login}")),
|
|
)
|
|
empty = self._empty_tomorrow(tomorrow["revision"] + 1)
|
|
connection.execute(
|
|
"UPDATE tomorrow_plans SET revision = ?, payload = ? WHERE login = ?",
|
|
(empty["revision"], self._cipher.seal(
|
|
{key: empty[key] for key in ("ids", "capacity_minutes", "estimates")},
|
|
binding=f"tomorrow:{login}",
|
|
), login),
|
|
)
|
|
return {"revision": 1, **payload}
|
|
|
|
def _normalize_week(
|
|
self, *, days: list[dict], timezone: str,
|
|
availability_defaults: list[int] | None = None,
|
|
) -> dict:
|
|
if not isinstance(days, list) or len(days) > 7:
|
|
raise ValueError("Week Ahead is limited to seven dates")
|
|
if not isinstance(timezone, str) or not timezone.strip() or len(timezone) > 100:
|
|
raise ValueError("timezone is required and bounded")
|
|
timezone = timezone.strip()
|
|
try:
|
|
ZoneInfo(timezone)
|
|
except (ZoneInfoNotFoundError, ValueError):
|
|
raise ValueError("timezone must be a valid IANA timezone") from None
|
|
normalized = []
|
|
seen_dates = set()
|
|
seen_ids = set()
|
|
for day in days:
|
|
if not isinstance(day, dict):
|
|
raise ValueError("each Week Ahead date must be an object")
|
|
plan_date = day.get("plan_date")
|
|
try:
|
|
if date.fromisoformat(plan_date or "").isoformat() != plan_date:
|
|
raise ValueError
|
|
except (TypeError, ValueError):
|
|
raise ValueError("plan_date must be an ISO calendar date") from None
|
|
if plan_date in seen_dates:
|
|
raise ValueError("Week Ahead dates must be unique")
|
|
seen_dates.add(plan_date)
|
|
plan = self._normalize_tomorrow(
|
|
ids=day.get("ids", []), capacity_minutes=day.get("capacity_minutes"),
|
|
estimates=day.get("estimates", {}), plan_date=plan_date, timezone=timezone,
|
|
)
|
|
if "free_windows" in day:
|
|
windows = day["free_windows"]
|
|
if not isinstance(windows, list) or len(windows) > 16:
|
|
raise ValueError("free windows must be a bounded list")
|
|
normalized_windows = []
|
|
previous_end = -1
|
|
for window in windows:
|
|
if not isinstance(window, dict) or set(window) != {"start_time", "end_time"}:
|
|
raise ValueError("free windows must contain only start and end times")
|
|
clock_values = []
|
|
for key in ("start_time", "end_time"):
|
|
value = window[key]
|
|
if (
|
|
not isinstance(value, str) or len(value) != 5 or value[2] != ":"
|
|
or not value[:2].isdigit() or not value[3:].isdigit()
|
|
or int(value[:2]) > 23 or int(value[3:]) > 59
|
|
):
|
|
raise ValueError("free windows must use HH:mm times")
|
|
clock_values.append(int(value[:2]) * 60 + int(value[3:]))
|
|
start, end = clock_values
|
|
if start < previous_end or end <= start:
|
|
raise ValueError("free windows must be ordered and non-overlapping")
|
|
normalized_windows.append({
|
|
"start_time": window["start_time"], "end_time": window["end_time"],
|
|
})
|
|
previous_end = end
|
|
plan["free_windows"] = normalized_windows
|
|
if "start_times" in day:
|
|
start_times = day["start_times"]
|
|
if not isinstance(start_times, dict) or len(start_times) > 5:
|
|
raise ValueError("task start times must be a bounded object")
|
|
if not set(start_times).issubset(plan["ids"]):
|
|
raise ValueError("task start times must belong to work on the same date")
|
|
occupied = []
|
|
for identity in plan["ids"]:
|
|
if identity not in start_times:
|
|
continue
|
|
value = start_times[identity]
|
|
if (
|
|
not isinstance(value, str) or len(value) != 5 or value[2] != ":"
|
|
or not value[:2].isdigit() or not value[3:].isdigit()
|
|
or int(value[:2]) > 23 or int(value[3:]) > 59
|
|
):
|
|
raise ValueError("task start times must use HH:mm times")
|
|
start = int(value[:2]) * 60 + int(value[3:])
|
|
minutes = plan["estimates"].get(identity)
|
|
if not isinstance(minutes, int) or start + minutes > 1440:
|
|
raise ValueError("task start times must fit their estimates within the date")
|
|
end = start + minutes
|
|
if any(start < other_end and end > other_start for other_start, other_end in occupied):
|
|
raise ValueError("task start times must not overlap")
|
|
windows = plan.get("free_windows")
|
|
if windows and not any(
|
|
int(window["start_time"][:2]) * 60 + int(window["start_time"][3:]) <= start
|
|
and end <= int(window["end_time"][:2]) * 60 + int(window["end_time"][3:])
|
|
for window in windows
|
|
):
|
|
raise ValueError("task start times must fit retained free windows")
|
|
occupied.append((start, end))
|
|
plan["start_times"] = {identity: start_times[identity] for identity in plan["ids"] if identity in start_times}
|
|
duplicate_ids = seen_ids.intersection(plan["ids"])
|
|
if duplicate_ids:
|
|
raise ValueError("work must be assigned to only one Week Ahead date")
|
|
seen_ids.update(plan["ids"])
|
|
plan.pop("timezone")
|
|
normalized.append(plan)
|
|
normalized.sort(key=lambda item: item["plan_date"])
|
|
result = {"timezone": timezone, "days": normalized}
|
|
if availability_defaults is not None:
|
|
if (
|
|
not isinstance(availability_defaults, list)
|
|
or len(availability_defaults) != 7
|
|
or any(
|
|
not isinstance(value, int) or isinstance(value, bool)
|
|
or value < 0 or value > 1440
|
|
for value in availability_defaults
|
|
)
|
|
):
|
|
raise ValueError("availability defaults must contain seven weekday capacities")
|
|
result["availability_defaults"] = list(availability_defaults)
|
|
return result
|
|
|
|
def replace_week(
|
|
self, login: str, *, base_revision: int, days: list[dict], timezone: str,
|
|
availability_defaults: list[int] | None = None,
|
|
) -> dict:
|
|
login = self._normalize_login(login)
|
|
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")
|
|
normalized = self._normalize_week(
|
|
days=days, timezone=timezone,
|
|
availability_defaults=availability_defaults,
|
|
)
|
|
with self._connect() as connection:
|
|
connection.execute("BEGIN IMMEDIATE")
|
|
row = connection.execute(
|
|
"SELECT revision, payload FROM week_plans WHERE login = ?", (login,)
|
|
).fetchone()
|
|
current = self._week_snapshot(row, login)
|
|
if current["revision"] != base_revision:
|
|
raise WeekPlanConflict(current)
|
|
snapshot = {"revision": base_revision + 1, **normalized}
|
|
connection.execute(
|
|
"INSERT INTO week_plans(login, revision, payload) VALUES (?, ?, ?) "
|
|
"ON CONFLICT(login) DO UPDATE SET revision=excluded.revision, payload=excluded.payload",
|
|
(login, snapshot["revision"], self._cipher.seal(normalized, binding=f"week:{login}")),
|
|
)
|
|
return snapshot
|
|
|
|
def reschedule_today_to_week(
|
|
self, login: str, *, operation_id: str, identity: str, estimate_minutes: int,
|
|
plan_date: str, today_revision: int, week_revision: int,
|
|
allow_over_capacity: bool = False,
|
|
) -> dict:
|
|
"""Atomically move one current Today item to one Week Ahead date."""
|
|
login = self._normalize_login(login)
|
|
if not isinstance(operation_id, str) or not operation_id.strip() or len(operation_id) > 100:
|
|
raise ValueError("operation_id is required and bounded")
|
|
operation_id = operation_id.strip()
|
|
if not isinstance(identity, str) or not identity or len(identity) > 500:
|
|
raise ValueError("identity is required and bounded")
|
|
if (
|
|
not isinstance(estimate_minutes, int) or isinstance(estimate_minutes, bool)
|
|
or estimate_minutes < 5 or estimate_minutes > 1440
|
|
):
|
|
raise ValueError("estimate_minutes must be between 5 and 1440")
|
|
with self._connect() as connection:
|
|
connection.execute("BEGIN IMMEDIATE")
|
|
receipt = connection.execute(
|
|
"SELECT result FROM week_reschedules WHERE login = ? AND operation_id = ?",
|
|
(login, operation_id),
|
|
).fetchone()
|
|
if receipt:
|
|
result, _legacy = self._cipher.open(
|
|
receipt[0], binding=f"week-reschedule:{login}:{operation_id}"
|
|
)
|
|
if not isinstance(result, dict):
|
|
raise PrivateStateEncryptionError("private state could not be decrypted")
|
|
return result
|
|
today_row = connection.execute(
|
|
"SELECT revision, ids, capacity_minutes, estimates, plan_date, timezone "
|
|
"FROM today_plans WHERE login = ?", (login,),
|
|
).fetchone()
|
|
today, _legacy = self._snapshot(today_row, login)
|
|
week_row = connection.execute(
|
|
"SELECT revision, payload FROM week_plans WHERE login = ?", (login,)
|
|
).fetchone()
|
|
week = self._week_snapshot(week_row, login)
|
|
if today["revision"] != today_revision or identity not in today["ids"]:
|
|
raise TodayPromotionConflict(today, week)
|
|
if week["revision"] != week_revision:
|
|
raise WeekPlanConflict(week)
|
|
|
|
days = []
|
|
for source in week["days"]:
|
|
estimates = dict(source.get("estimates", {}))
|
|
estimates.pop(identity, None)
|
|
days.append({
|
|
**source,
|
|
"ids": [item_id for item_id in source.get("ids", []) if item_id != identity],
|
|
"estimates": estimates,
|
|
})
|
|
destination = next((day for day in days if day["plan_date"] == plan_date), None)
|
|
if destination is None:
|
|
destination = {
|
|
"plan_date": plan_date, "ids": [],
|
|
"capacity_minutes": None, "estimates": {},
|
|
}
|
|
days.append(destination)
|
|
destination["ids"].append(identity)
|
|
destination["estimates"][identity] = estimate_minutes
|
|
planned_minutes = sum(
|
|
int(destination["estimates"].get(item_id, 0)) for item_id in destination["ids"]
|
|
)
|
|
if (
|
|
destination.get("capacity_minutes") is not None
|
|
and planned_minutes > destination["capacity_minutes"]
|
|
and not allow_over_capacity
|
|
):
|
|
raise ValueError("explicit overload confirmation is required")
|
|
normalized_week = self._normalize_week(
|
|
days=days, timezone=week["timezone"],
|
|
availability_defaults=week.get("availability_defaults"),
|
|
)
|
|
|
|
today_ids = [item_id for item_id in today["ids"] if item_id != identity]
|
|
today_estimates = {
|
|
item_id: minutes for item_id, minutes in today["estimates"].items()
|
|
if item_id != identity
|
|
}
|
|
today_result = {
|
|
**today, "revision": today_revision + 1,
|
|
"ids": today_ids, "estimates": today_estimates,
|
|
}
|
|
week_result = {"revision": week_revision + 1, **normalized_week}
|
|
connection.execute(
|
|
"UPDATE today_plans SET revision = ?, ids = ?, capacity_minutes = NULL, "
|
|
"estimates = '{}', plan_date = NULL, timezone = NULL WHERE login = ?",
|
|
(today_result["revision"], self._sealed_plan(login, today_result), login),
|
|
)
|
|
connection.execute(
|
|
"UPDATE week_plans SET revision = ?, payload = ? WHERE login = ?",
|
|
(week_result["revision"], self._cipher.seal(
|
|
normalized_week, binding=f"week:{login}"
|
|
), login),
|
|
)
|
|
result = {"today": today_result, "week": week_result}
|
|
connection.execute(
|
|
"INSERT INTO week_reschedules(login, operation_id, result, created_at) "
|
|
"VALUES (?, ?, ?, ?)",
|
|
(login, operation_id, self._cipher.seal(
|
|
result, binding=f"week-reschedule:{login}:{operation_id}"
|
|
), self.clock()),
|
|
)
|
|
return result
|
|
|
|
def pull_week_item(
|
|
self, login: str, *, operation_id: str, identity: str,
|
|
today_revision: int, week_revision: int, allow_over_capacity: bool = False,
|
|
) -> dict:
|
|
"""Atomically append one Week Ahead item to Today and remove it from Week."""
|
|
login = self._normalize_login(login)
|
|
if not isinstance(operation_id, str) or not operation_id.strip() or len(operation_id) > 100:
|
|
raise ValueError("operation_id is required and bounded")
|
|
if not isinstance(identity, str) or not identity or len(identity) > 500:
|
|
raise ValueError("identity is required and bounded")
|
|
operation_id = operation_id.strip()
|
|
with self._connect() as connection:
|
|
connection.execute("BEGIN IMMEDIATE")
|
|
receipt = connection.execute(
|
|
"SELECT result FROM week_item_pulls WHERE login = ? AND operation_id = ?",
|
|
(login, operation_id),
|
|
).fetchone()
|
|
if receipt:
|
|
result, _legacy = self._cipher.open(
|
|
receipt[0], binding=f"week-item-pull:{login}:{operation_id}"
|
|
)
|
|
if not isinstance(result, dict):
|
|
raise PrivateStateEncryptionError("private state could not be decrypted")
|
|
return result
|
|
today_row = connection.execute(
|
|
"SELECT revision, ids, capacity_minutes, estimates, plan_date, timezone "
|
|
"FROM today_plans WHERE login = ?", (login,),
|
|
).fetchone()
|
|
today, _legacy = self._snapshot(today_row, login)
|
|
week_row = connection.execute(
|
|
"SELECT revision, payload FROM week_plans WHERE login = ?", (login,),
|
|
).fetchone()
|
|
week = self._week_snapshot(week_row, login)
|
|
if today["revision"] != today_revision:
|
|
raise TodayPromotionConflict(today, week)
|
|
if week["revision"] != week_revision:
|
|
raise WeekPlanConflict(week)
|
|
if identity in today["ids"]:
|
|
raise ValueError("work is already in Today")
|
|
if len(today["ids"]) >= self.limit:
|
|
raise TodayPlanFull("Today plan is full")
|
|
source = next((day for day in week["days"] if identity in day["ids"]), None)
|
|
if source is None:
|
|
raise WeekPlanConflict(week)
|
|
estimate = source.get("estimates", {}).get(identity)
|
|
planned_minutes = sum(
|
|
int(today["estimates"].get(item_id, 0)) for item_id in today["ids"]
|
|
) + (int(estimate) if estimate else 0)
|
|
if (
|
|
today.get("capacity_minutes") is not None
|
|
and planned_minutes > today["capacity_minutes"]
|
|
and not allow_over_capacity
|
|
):
|
|
raise ValueError("moving work over Today capacity requires explicit overload confirmation")
|
|
today_result = {
|
|
**today, "revision": today_revision + 1,
|
|
"ids": [*today["ids"], identity],
|
|
"estimates": {**today["estimates"], **({identity: estimate} if estimate else {})},
|
|
}
|
|
remaining_days = []
|
|
for day in week["days"]:
|
|
estimates = dict(day.get("estimates", {}))
|
|
estimates.pop(identity, None)
|
|
remaining_days.append({
|
|
**day, "ids": [item_id for item_id in day["ids"] if item_id != identity],
|
|
"estimates": estimates,
|
|
})
|
|
week_payload = {"timezone": week["timezone"], "days": remaining_days}
|
|
week_result = {"revision": week_revision + 1, **week_payload}
|
|
connection.execute(
|
|
"INSERT INTO today_plans(login, revision, ids, capacity_minutes, estimates, plan_date, timezone) "
|
|
"VALUES (?, ?, ?, NULL, '{}', NULL, NULL) ON CONFLICT(login) DO UPDATE SET "
|
|
"revision=excluded.revision, ids=excluded.ids, capacity_minutes=NULL, estimates='{}', "
|
|
"plan_date=NULL, timezone=NULL",
|
|
(login, today_result["revision"], self._sealed_plan(login, today_result)),
|
|
)
|
|
connection.execute(
|
|
"UPDATE week_plans SET revision = ?, payload = ? WHERE login = ?",
|
|
(week_result["revision"], self._cipher.seal(week_payload, binding=f"week:{login}"), login),
|
|
)
|
|
result = {"today": today_result, "week": week_result}
|
|
connection.execute(
|
|
"INSERT INTO week_item_pulls(login, operation_id, result, created_at) VALUES (?, ?, ?, ?)",
|
|
(login, operation_id, self._cipher.seal(
|
|
result, binding=f"week-item-pull:{login}:{operation_id}"
|
|
), self.clock()),
|
|
)
|
|
return result
|
|
|
|
def promote_week(
|
|
self, login: str, *, promotion_id: str, week_revision: int,
|
|
plan_date: str, today_revision: int, allow_future: bool = False,
|
|
) -> dict:
|
|
login = self._normalize_login(login)
|
|
if not isinstance(promotion_id, str) or not promotion_id.strip() or len(promotion_id) > 100:
|
|
raise ValueError("promotion_id is required and bounded")
|
|
promotion_id = promotion_id.strip()
|
|
with self._connect() as connection:
|
|
connection.execute("BEGIN IMMEDIATE")
|
|
receipt = connection.execute(
|
|
"SELECT result FROM week_promotions WHERE login = ? AND promotion_id = ?",
|
|
(login, promotion_id),
|
|
).fetchone()
|
|
if receipt:
|
|
result, _legacy = self._cipher.open(
|
|
receipt[0], binding=f"week-promotion:{login}:{promotion_id}"
|
|
)
|
|
if not isinstance(result, dict):
|
|
raise PrivateStateEncryptionError("private state could not be decrypted")
|
|
return result
|
|
row = connection.execute(
|
|
"SELECT revision, payload FROM week_plans WHERE login = ?", (login,)
|
|
).fetchone()
|
|
week = self._week_snapshot(row, login)
|
|
if week["revision"] != week_revision:
|
|
raise WeekPlanConflict(week)
|
|
day = next((item for item in week["days"] if item["plan_date"] == plan_date), None)
|
|
if day is None:
|
|
raise WeekPlanConflict(week)
|
|
local_date = datetime.fromtimestamp(self.clock(), ZoneInfo(week["timezone"])).date().isoformat()
|
|
if local_date < plan_date and not allow_future:
|
|
raise TomorrowPlanNotDue(plan_date)
|
|
today_row = connection.execute(
|
|
"SELECT revision, ids, capacity_minutes, estimates, plan_date, timezone "
|
|
"FROM today_plans WHERE login = ?", (login,)
|
|
).fetchone()
|
|
today, _legacy = self._snapshot(today_row, login)
|
|
if today["revision"] != today_revision:
|
|
raise TodayPromotionConflict(today)
|
|
if today["ids"]:
|
|
raise TodayPromotionConflict(today, week)
|
|
result = {"revision": today_revision + 1, **day, "timezone": week["timezone"]}
|
|
connection.execute(
|
|
"INSERT INTO today_plans(login, revision, ids, capacity_minutes, estimates, plan_date, timezone) "
|
|
"VALUES (?, ?, ?, NULL, '{}', NULL, NULL) ON CONFLICT(login) DO UPDATE SET "
|
|
"revision=excluded.revision, ids=excluded.ids, capacity_minutes=NULL, estimates='{}', "
|
|
"plan_date=NULL, timezone=NULL",
|
|
(login, result["revision"], self._sealed_plan(login, result)),
|
|
)
|
|
remaining = {"timezone": week["timezone"], "days": [
|
|
item for item in week["days"] if item["plan_date"] != plan_date
|
|
]}
|
|
connection.execute(
|
|
"UPDATE week_plans SET revision = ?, payload = ? WHERE login = ?",
|
|
(week_revision + 1, self._cipher.seal(remaining, binding=f"week:{login}"), login),
|
|
)
|
|
connection.execute(
|
|
"INSERT INTO week_promotions(login, promotion_id, result, created_at) VALUES (?, ?, ?, ?)",
|
|
(login, promotion_id, self._cipher.seal(
|
|
result, binding=f"week-promotion:{login}:{promotion_id}"
|
|
), self.clock()),
|
|
)
|
|
return result
|
|
|
|
def reconcile_week(
|
|
self, login: str, *, promotion_id: str, week_revision: int,
|
|
plan_date: str, today_revision: int, ids: list[str],
|
|
capacity_minutes: int | None, estimates: dict[str, int],
|
|
) -> dict:
|
|
"""Atomically replace Today from preserved Today/due-week work and consume that day."""
|
|
login = self._normalize_login(login)
|
|
if not isinstance(promotion_id, str) or not promotion_id.strip() or len(promotion_id) > 100:
|
|
raise ValueError("promotion_id is required and bounded")
|
|
promotion_id = promotion_id.strip()
|
|
with self._connect() as connection:
|
|
connection.execute("BEGIN IMMEDIATE")
|
|
receipt = connection.execute(
|
|
"SELECT result FROM week_promotions WHERE login = ? AND promotion_id = ?",
|
|
(login, promotion_id),
|
|
).fetchone()
|
|
if receipt:
|
|
result, _legacy = self._cipher.open(
|
|
receipt[0], binding=f"week-promotion:{login}:{promotion_id}"
|
|
)
|
|
if not isinstance(result, dict):
|
|
raise PrivateStateEncryptionError("private state could not be decrypted")
|
|
return result
|
|
row = connection.execute(
|
|
"SELECT revision, payload FROM week_plans WHERE login = ?", (login,)
|
|
).fetchone()
|
|
week = self._week_snapshot(row, login)
|
|
if week["revision"] != week_revision:
|
|
raise WeekPlanConflict(week)
|
|
day = next((item for item in week["days"] if item["plan_date"] == plan_date), None)
|
|
if day is None:
|
|
raise WeekPlanConflict(week)
|
|
local_date = datetime.fromtimestamp(self.clock(), ZoneInfo(week["timezone"])).date().isoformat()
|
|
if local_date < plan_date:
|
|
raise TomorrowPlanNotDue(plan_date)
|
|
today_row = connection.execute(
|
|
"SELECT revision, ids, capacity_minutes, estimates, plan_date, timezone "
|
|
"FROM today_plans WHERE login = ?", (login,)
|
|
).fetchone()
|
|
today, _legacy = self._snapshot(today_row, login)
|
|
if today["revision"] != today_revision:
|
|
raise TodayPromotionConflict(today, week)
|
|
allowed = set(today["ids"]) | set(day["ids"])
|
|
if any(item_id not in allowed for item_id in ids):
|
|
raise ValueError("selected work must come from Today or the due Week Ahead day")
|
|
normalized = self._normalize_tomorrow(
|
|
ids=ids, capacity_minutes=capacity_minutes, estimates=estimates,
|
|
plan_date=plan_date, timezone=week["timezone"],
|
|
)
|
|
today_result = {"revision": today_revision + 1, **normalized}
|
|
connection.execute(
|
|
"INSERT INTO today_plans(login, revision, ids, capacity_minutes, estimates, plan_date, timezone) "
|
|
"VALUES (?, ?, ?, NULL, '{}', NULL, NULL) ON CONFLICT(login) DO UPDATE SET "
|
|
"revision=excluded.revision, ids=excluded.ids, capacity_minutes=NULL, estimates='{}', "
|
|
"plan_date=NULL, timezone=NULL",
|
|
(login, today_result["revision"], self._sealed_plan(login, today_result)),
|
|
)
|
|
remaining_payload = {"timezone": week["timezone"], "days": [
|
|
item for item in week["days"] if item["plan_date"] != plan_date
|
|
]}
|
|
week_result = {"revision": week_revision + 1, **remaining_payload}
|
|
connection.execute(
|
|
"UPDATE week_plans SET revision = ?, payload = ? WHERE login = ?",
|
|
(week_result["revision"], self._cipher.seal(
|
|
remaining_payload, binding=f"week:{login}"
|
|
), login),
|
|
)
|
|
result = {"today": today_result, "week": week_result}
|
|
connection.execute(
|
|
"INSERT INTO week_promotions(login, promotion_id, result, created_at) VALUES (?, ?, ?, ?)",
|
|
(login, promotion_id, self._cipher.seal(
|
|
result, binding=f"week-promotion:{login}:{promotion_id}"
|
|
), self.clock()),
|
|
)
|
|
return result
|
|
|
|
@staticmethod
|
|
def _empty_session() -> dict:
|
|
return {
|
|
"revision": 0, "device_id": "", "identity": "",
|
|
"elapsed_ms": 0, "running": False, "break_deadline_at": None, "updated_at": None,
|
|
}
|
|
|
|
def _session_snapshot(self, row, login: str) -> tuple[dict, bool]:
|
|
if row is None:
|
|
return self._empty_session(), False
|
|
if row[1].startswith(("v1:", "v2:")):
|
|
payload, legacy = self._cipher.open(row[1], binding=f"session:{login}")
|
|
if not isinstance(payload, dict):
|
|
raise PrivateStateEncryptionError("private state could not be decrypted")
|
|
return {"revision": int(row[0]), **payload, "updated_at": row[6]}, legacy
|
|
return {
|
|
"revision": int(row[0]), "device_id": row[1], "identity": row[2],
|
|
"elapsed_ms": int(row[3]), "running": bool(row[4]),
|
|
"break_deadline_at": row[5], "updated_at": row[6],
|
|
}, True
|
|
|
|
def _sealed_session(self, login: str, session: dict) -> str:
|
|
return self._cipher.seal({
|
|
"device_id": session["device_id"], "identity": session["identity"],
|
|
"elapsed_ms": session["elapsed_ms"], "running": session["running"],
|
|
"break_deadline_at": session["break_deadline_at"],
|
|
}, binding=f"session:{login}")
|
|
|
|
def get_session(self, login: str) -> dict:
|
|
login = self._normalize_login(login)
|
|
with self._connect() as connection:
|
|
row = connection.execute(
|
|
"SELECT revision, device_id, identity, elapsed_ms, running, break_deadline_at, updated_at "
|
|
"FROM today_sessions WHERE login = ?", (login,),
|
|
).fetchone()
|
|
session, legacy = self._session_snapshot(row, login)
|
|
if row is not None and legacy:
|
|
connection.execute(
|
|
"UPDATE today_sessions SET device_id = ?, identity = '', elapsed_ms = 0, "
|
|
"running = 0, break_deadline_at = NULL WHERE login = ? AND device_id = ?",
|
|
(self._sealed_session(login, session), login, row[1]),
|
|
)
|
|
return session
|
|
|
|
def update_session(
|
|
self, login: str, *, base_revision: int, device_id: str,
|
|
identity: str, elapsed_ms: int, running: bool, break_deadline_at: int | None = None,
|
|
) -> dict:
|
|
login = self._normalize_login(login)
|
|
updated_at = self.clock()
|
|
with self._connect() as connection:
|
|
connection.execute("BEGIN IMMEDIATE")
|
|
current = connection.execute(
|
|
"SELECT revision, device_id, identity, elapsed_ms, running, break_deadline_at, updated_at "
|
|
"FROM today_sessions WHERE login = ?", (login,)
|
|
).fetchone()
|
|
current_session, _legacy = self._session_snapshot(current, login)
|
|
current_revision = current_session["revision"]
|
|
if base_revision != current_revision:
|
|
raise TodaySessionConflict(current_session)
|
|
revision = current_revision + 1
|
|
session = {
|
|
"revision": revision, "device_id": device_id, "identity": identity,
|
|
"elapsed_ms": elapsed_ms, "running": bool(running),
|
|
"break_deadline_at": break_deadline_at, "updated_at": updated_at,
|
|
}
|
|
connection.execute(
|
|
"INSERT INTO today_sessions(login, revision, device_id, identity, elapsed_ms, running, break_deadline_at, updated_at) "
|
|
"VALUES (?, ?, ?, ?, ?, ?, ?, ?) "
|
|
"ON CONFLICT(login) DO UPDATE SET revision=excluded.revision, device_id=excluded.device_id, "
|
|
"identity=excluded.identity, elapsed_ms=excluded.elapsed_ms, running=excluded.running, "
|
|
"break_deadline_at=excluded.break_deadline_at, updated_at=excluded.updated_at",
|
|
(login, revision, self._sealed_session(login, session), "", 0, 0, None, updated_at),
|
|
)
|
|
return session
|
|
|
|
@staticmethod
|
|
def _normalize_recap_items(items: list[dict]) -> list[dict]:
|
|
if not isinstance(items, list) or not items:
|
|
raise ValueError("recap requires at least one item")
|
|
if len(items) > 20:
|
|
raise ValueError("recap is limited to 20 items")
|
|
normalized = []
|
|
seen = set()
|
|
for item in items:
|
|
identity = item.get("identity", "") if isinstance(item, dict) else ""
|
|
if not isinstance(identity, str) or not identity.strip() or len(identity) > 500:
|
|
raise ValueError("recap item identity is required and bounded")
|
|
identity = identity.strip()
|
|
if identity in seen:
|
|
raise ValueError("recap item identities must be unique")
|
|
actual = item.get("actual_minutes")
|
|
estimate = item.get("estimate_minutes")
|
|
if not isinstance(actual, int) or isinstance(actual, bool) or actual < 0 or actual > 1440:
|
|
raise ValueError("actual minutes must be between 0 and 1440")
|
|
if estimate is not None and (
|
|
not isinstance(estimate, int) or isinstance(estimate, bool)
|
|
or estimate < 5 or estimate > 1440
|
|
):
|
|
raise ValueError("estimate minutes must be between 5 and 1440")
|
|
seen.add(identity)
|
|
normalized.append({
|
|
"identity": identity,
|
|
"estimate_minutes": estimate,
|
|
"actual_minutes": actual,
|
|
})
|
|
return normalized
|
|
|
|
def _recap_snapshot(
|
|
self, login: str, encrypted_session_id: str, created_at: float, serialized: str
|
|
) -> tuple[dict, bool]:
|
|
if encrypted_session_id.startswith(("v1:", "v2:")):
|
|
session_id, legacy_session = self._cipher.open(
|
|
encrypted_session_id, binding=f"recap-id:{login}"
|
|
)
|
|
items, legacy_items = self._cipher.open(
|
|
serialized, binding=f"recap-items:{login}:{session_id}"
|
|
)
|
|
else:
|
|
session_id, legacy_session = encrypted_session_id, True
|
|
try:
|
|
items, legacy_items = json.loads(serialized), True
|
|
except json.JSONDecodeError as error:
|
|
raise PrivateStateEncryptionError("private state could not be decrypted") from error
|
|
if not isinstance(session_id, str) or not isinstance(items, list):
|
|
raise PrivateStateEncryptionError("private state could not be decrypted")
|
|
estimated = sum(item["estimate_minutes"] for item in items if item["estimate_minutes"] is not None)
|
|
actual = sum(item["actual_minutes"] for item in items)
|
|
return {
|
|
"session_id": session_id,
|
|
"created_at": created_at,
|
|
"items": items,
|
|
"estimated_minutes": estimated,
|
|
"actual_minutes": actual,
|
|
"variance_minutes": actual - estimated,
|
|
}, legacy_session or legacy_items
|
|
|
|
def _migrate_recap(self, connection, login: str, stored_session: str, snapshot: dict) -> None:
|
|
encrypted_session = self._cipher.seal(
|
|
snapshot["session_id"], binding=f"recap-id:{login}"
|
|
)
|
|
encrypted_items = self._cipher.seal(
|
|
snapshot["items"], binding=f"recap-items:{login}:{snapshot['session_id']}"
|
|
)
|
|
connection.execute(
|
|
"UPDATE today_recaps SET session_id = ?, items = ? "
|
|
"WHERE login = ? AND session_id = ?",
|
|
(encrypted_session, encrypted_items, login, stored_session),
|
|
)
|
|
|
|
def save_recap(self, login: str, session_id: str, items: list[dict]) -> dict:
|
|
login = self._normalize_login(login)
|
|
if not isinstance(session_id, str) or not session_id.strip() or len(session_id) > 100:
|
|
raise ValueError("session_id is required and bounded")
|
|
session_id = session_id.strip()
|
|
normalized = self._normalize_recap_items(items)
|
|
with self._connect() as connection:
|
|
connection.execute("BEGIN IMMEDIATE")
|
|
existing_rows = connection.execute(
|
|
"SELECT session_id, created_at, items FROM today_recaps WHERE login = ?", (login,)
|
|
).fetchall()
|
|
for existing in existing_rows:
|
|
snapshot, legacy = self._recap_snapshot(login, *existing)
|
|
if snapshot["session_id"] == session_id:
|
|
if legacy:
|
|
self._migrate_recap(connection, login, existing[0], snapshot)
|
|
return snapshot
|
|
created_at = self.clock()
|
|
encrypted_session_id = self._cipher.seal(session_id, binding=f"recap-id:{login}")
|
|
serialized = self._cipher.seal(
|
|
normalized, binding=f"recap-items:{login}:{session_id}"
|
|
)
|
|
connection.execute(
|
|
"INSERT INTO today_recaps(login, session_id, created_at, items) VALUES (?, ?, ?, ?)",
|
|
(login, encrypted_session_id, created_at, serialized),
|
|
)
|
|
connection.execute(
|
|
"DELETE FROM today_recaps WHERE login = ? AND rowid NOT IN "
|
|
"(SELECT rowid FROM today_recaps WHERE login = ? "
|
|
"ORDER BY created_at DESC, rowid DESC LIMIT ?)",
|
|
(login, login, self.recap_limit),
|
|
)
|
|
return self._recap_snapshot(login, encrypted_session_id, created_at, serialized)[0]
|
|
|
|
def list_recaps(self, login: str, *, limit: int = 30) -> list[dict]:
|
|
bounded_limit = max(1, min(int(limit), self.recap_limit, 100))
|
|
login = self._normalize_login(login)
|
|
with self._connect() as connection:
|
|
rows = connection.execute(
|
|
"SELECT session_id, created_at, items FROM today_recaps WHERE login = ? "
|
|
"ORDER BY created_at DESC, rowid DESC LIMIT ?",
|
|
(login, bounded_limit),
|
|
).fetchall()
|
|
snapshots = []
|
|
for row in rows:
|
|
snapshot, legacy = self._recap_snapshot(login, *row)
|
|
if legacy:
|
|
self._migrate_recap(connection, login, row[0], snapshot)
|
|
snapshots.append(snapshot)
|
|
return snapshots
|
|
|
|
def _time_log_snapshot(self, login: str, row) -> tuple[dict, bool]:
|
|
encrypted_session, encrypted_identity, payload_value, status_value = row
|
|
if encrypted_session.startswith(("v1:", "v2:")):
|
|
session_id = self._cipher.open(
|
|
encrypted_session, binding=f"time-log-session:{login}"
|
|
)[0]
|
|
identity = self._cipher.open(
|
|
encrypted_identity, binding=f"time-log-identity:{login}:{session_id}"
|
|
)[0]
|
|
payload, legacy = self._cipher.open(
|
|
str(payload_value), binding=f"time-log-payload:{login}:{session_id}:{identity}"
|
|
)
|
|
if not isinstance(session_id, str) or not isinstance(identity, str) or not isinstance(payload, dict):
|
|
raise PrivateStateEncryptionError("private state could not be decrypted")
|
|
return {
|
|
"session_id": session_id, "identity": identity,
|
|
"actual_minutes": payload.get("actual_minutes"), "status": payload.get("status"),
|
|
"stored_session": encrypted_session, "stored_identity": encrypted_identity,
|
|
}, legacy
|
|
return {
|
|
"session_id": encrypted_session, "identity": encrypted_identity,
|
|
"actual_minutes": payload_value, "status": status_value,
|
|
"stored_session": encrypted_session, "stored_identity": encrypted_identity,
|
|
}, True
|
|
|
|
def _sealed_time_log(self, login: str, session_id: str, identity: str, actual: int, status: str) -> tuple[str, str, str]:
|
|
return (
|
|
self._cipher.seal(session_id, binding=f"time-log-session:{login}"),
|
|
self._cipher.seal(identity, binding=f"time-log-identity:{login}:{session_id}"),
|
|
self._cipher.seal(
|
|
{"actual_minutes": actual, "status": status},
|
|
binding=f"time-log-payload:{login}:{session_id}:{identity}",
|
|
),
|
|
)
|
|
|
|
def _migrate_time_log(self, connection, login: str, snapshot: dict) -> dict:
|
|
sealed_session, sealed_identity, sealed_payload = self._sealed_time_log(
|
|
login, snapshot["session_id"], snapshot["identity"],
|
|
snapshot["actual_minutes"], snapshot["status"],
|
|
)
|
|
connection.execute(
|
|
"UPDATE today_time_logs SET session_id = ?, identity = ?, actual_minutes = ?, status = 'sealed' "
|
|
"WHERE login = ? AND session_id = ? AND identity = ?",
|
|
(sealed_session, sealed_identity, sealed_payload, login,
|
|
snapshot["stored_session"], snapshot["stored_identity"]),
|
|
)
|
|
return {**snapshot, "stored_session": sealed_session, "stored_identity": sealed_identity}
|
|
|
|
def begin_time_log(self, login: str, session_id: str, identity: str, actual_minutes: int) -> str:
|
|
"""Claim one recap item for upstream logging, returning its current state."""
|
|
login = self._normalize_login(login)
|
|
with self._connect() as connection:
|
|
connection.execute("BEGIN IMMEDIATE")
|
|
rows = connection.execute(
|
|
"SELECT session_id, identity, actual_minutes, status FROM today_time_logs WHERE login = ?",
|
|
(login,),
|
|
).fetchall()
|
|
match = None
|
|
for row in rows:
|
|
candidate, legacy = self._time_log_snapshot(login, row)
|
|
if candidate["session_id"] == session_id and candidate["identity"] == identity:
|
|
match = self._migrate_time_log(connection, login, candidate) if legacy else candidate
|
|
break
|
|
if match is None:
|
|
sealed_session, sealed_identity, sealed_payload = self._sealed_time_log(
|
|
login, session_id, identity, actual_minutes, "pending"
|
|
)
|
|
connection.execute(
|
|
"INSERT INTO today_time_logs(login, session_id, identity, actual_minutes, status) "
|
|
"VALUES (?, ?, ?, ?, 'sealed')",
|
|
(login, sealed_session, sealed_identity, sealed_payload),
|
|
)
|
|
return "claimed"
|
|
if match["actual_minutes"] != actual_minutes:
|
|
raise ValueError("logged recap time cannot be changed")
|
|
if match["status"] == "failed":
|
|
sealed_payload = self._sealed_time_log(
|
|
login, session_id, identity, actual_minutes, "pending"
|
|
)[2]
|
|
connection.execute(
|
|
"UPDATE today_time_logs SET actual_minutes = ?, status = 'sealed' "
|
|
"WHERE login = ? AND session_id = ? AND identity = ?",
|
|
(sealed_payload, login, match["stored_session"], match["stored_identity"]),
|
|
)
|
|
return "claimed"
|
|
return match["status"]
|
|
|
|
def finish_time_log(self, login: str, session_id: str, identity: str, *, succeeded: bool) -> None:
|
|
login = self._normalize_login(login)
|
|
with self._connect() as connection:
|
|
rows = connection.execute(
|
|
"SELECT session_id, identity, actual_minutes, status FROM today_time_logs WHERE login = ?",
|
|
(login,),
|
|
).fetchall()
|
|
for row in rows:
|
|
match, legacy = self._time_log_snapshot(login, row)
|
|
if match["session_id"] != session_id or match["identity"] != identity or match["status"] != "pending":
|
|
continue
|
|
if legacy:
|
|
match = self._migrate_time_log(connection, login, match)
|
|
sealed_payload = self._sealed_time_log(
|
|
login, session_id, identity, match["actual_minutes"],
|
|
"logged" if succeeded else "failed",
|
|
)[2]
|
|
connection.execute(
|
|
"UPDATE today_time_logs SET actual_minutes = ?, status = 'sealed' "
|
|
"WHERE login = ? AND session_id = ? AND identity = ?",
|
|
(sealed_payload, login, match["stored_session"], match["stored_identity"]),
|
|
)
|
|
return
|
|
|
|
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, plan_date, timezone "
|
|
"FROM today_plans WHERE login = ?", (login,)
|
|
).fetchone()
|
|
snapshot, _legacy = self._snapshot(row, login)
|
|
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)
|
|
sealed = self._sealed_plan(login, {
|
|
"ids": ids, "capacity_minutes": snapshot["capacity_minutes"],
|
|
"estimates": estimates, "plan_date": snapshot.get("plan_date"),
|
|
"timezone": snapshot.get("timezone"),
|
|
"first_task_state": snapshot.get("first_task_state", ""),
|
|
})
|
|
if row is None:
|
|
connection.execute(
|
|
"INSERT INTO today_plans(login, revision, ids, capacity_minutes, estimates, plan_date, timezone) "
|
|
"VALUES (?, ?, ?, ?, ?, ?, ?)",
|
|
(login, revision, sealed, None, "{}", None, None),
|
|
)
|
|
elif changed:
|
|
connection.execute(
|
|
"UPDATE today_plans SET revision = ?, ids = ?, capacity_minutes = NULL, "
|
|
"estimates = '{}', plan_date = NULL, timezone = NULL WHERE login = ?",
|
|
(revision, sealed, login),
|
|
)
|
|
self._record_operation(connection, login, operation_id)
|
|
result = {
|
|
"revision": revision,
|
|
"ids": ids,
|
|
"capacity_minutes": snapshot["capacity_minutes"],
|
|
"estimates": estimates,
|
|
}
|
|
if snapshot.get("plan_date"):
|
|
result["plan_date"] = snapshot["plan_date"]
|
|
result["timezone"] = snapshot["timezone"]
|
|
if snapshot.get("first_task_state"):
|
|
result["first_task_state"] = snapshot["first_task_state"]
|
|
return result
|
|
|
|
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, plan_date, timezone "
|
|
"FROM today_plans WHERE login = ?", (login,)
|
|
).fetchone()
|
|
snapshot, _legacy = self._snapshot(row, login)
|
|
ids = list(snapshot["ids"])
|
|
capacity_minutes = snapshot["capacity_minutes"]
|
|
estimates = dict(snapshot["estimates"])
|
|
plan_date = snapshot.get("plan_date")
|
|
timezone = snapshot.get("timezone")
|
|
first_task_state = snapshot.get("first_task_state", "")
|
|
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", "rollover", "activate"}:
|
|
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 action != "activate" and 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
|
|
elif action == "configure":
|
|
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
|
|
elif action == "activate":
|
|
proposed_state = operation.get("activation_state")
|
|
if item_id != "first-task" or proposed_state not in {"coaching", "complete"}:
|
|
raise ValueError("first-task activation state is invalid")
|
|
rank = {"": 0, "coaching": 1, "complete": 2}
|
|
changed = rank[proposed_state] > rank[first_task_state]
|
|
if changed:
|
|
first_task_state = proposed_state
|
|
else:
|
|
proposed_date = operation.get("plan_date")
|
|
proposed_timezone = operation.get("timezone")
|
|
try:
|
|
if date.fromisoformat(proposed_date or "").isoformat() != proposed_date:
|
|
raise ValueError
|
|
except (TypeError, ValueError):
|
|
raise ValueError("plan_date must be an ISO calendar date") from None
|
|
if not isinstance(proposed_timezone, str) or not proposed_timezone.strip() or len(proposed_timezone) > 100:
|
|
raise ValueError("timezone is required and bounded")
|
|
proposed_ids = operation.get("ids")
|
|
if not isinstance(proposed_ids, list) or len(proposed_ids) > self.limit or any(
|
|
not isinstance(candidate, str) or not candidate or len(candidate) > 500
|
|
for candidate in proposed_ids
|
|
):
|
|
raise ValueError("rollover IDs are invalid or exceed the Today limit")
|
|
if len(set(proposed_ids)) != len(proposed_ids):
|
|
raise ValueError("rollover IDs must be unique")
|
|
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 proposed_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 = (ids != proposed_ids or capacity_minutes != proposed_capacity or
|
|
estimates != normalized_estimates or plan_date != proposed_date or
|
|
timezone != proposed_timezone.strip())
|
|
ids = list(proposed_ids)
|
|
capacity_minutes = proposed_capacity
|
|
estimates = normalized_estimates
|
|
plan_date = proposed_date
|
|
timezone = proposed_timezone.strip()
|
|
|
|
revision += 1 if changed else 0
|
|
self._record_operation(connection, login, operation_id)
|
|
accepted.append(operation_id)
|
|
|
|
sealed = self._sealed_plan(login, {
|
|
"ids": ids, "capacity_minutes": capacity_minutes, "estimates": estimates,
|
|
"plan_date": plan_date, "timezone": timezone,
|
|
"first_task_state": first_task_state,
|
|
})
|
|
if row is None:
|
|
connection.execute(
|
|
"INSERT INTO today_plans(login, revision, ids, capacity_minutes, estimates, plan_date, timezone) "
|
|
"VALUES (?, ?, ?, ?, ?, ?, ?)",
|
|
(login, revision, sealed, None, "{}", None, None),
|
|
)
|
|
elif accepted:
|
|
connection.execute(
|
|
"UPDATE today_plans SET revision = ?, ids = ?, capacity_minutes = ?, estimates = ?, "
|
|
"plan_date = ?, timezone = ? WHERE login = ?",
|
|
(revision, sealed, None, "{}", None, None, login),
|
|
)
|
|
result = {
|
|
"revision": revision,
|
|
"ids": ids,
|
|
"capacity_minutes": capacity_minutes,
|
|
"estimates": estimates,
|
|
"accepted_operation_ids": accepted,
|
|
"duplicate_operation_ids": duplicates,
|
|
"rejected_operations": rejected,
|
|
}
|
|
if plan_date:
|
|
result["plan_date"] = plan_date
|
|
result["timezone"] = timezone
|
|
if first_task_state:
|
|
result["first_task_state"] = first_task_state
|
|
return result
|