bvandeusenandClaude Opus 4.8
7ce5bb8450
Drift-audit Group 8 (lifecycle gaps): - change_password no longer 500s for OAuth-only users: short-circuit when password_hash is None (verify_password would crash on None) so the route returns a clean 4xx instead of a 500. - register_with_invitation no longer locks the invitee out on a username collision: create the user FIRST, then mark the token used, so a failed creation (409) leaves the single-use invite valid for retry. - update_event re-arms reminder_sent_at when start_dt/reminder_minutes change, so a rescheduled event fires again instead of being permanently suppressed. - Migration 0061: uq_topic_per_rulebook / uq_rule_per_topic become PARTIAL unique indexes (WHERE deleted_at IS NULL). Trashing 'X' then recreating it no longer 500s on the dead row's title. Model __table_args__ updated to match. Deferred: per-occurrence reminders for recurring events (event_scheduler) — needs a per-occurrence reminder-state design, not a one-line gate tweak. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
478 lines
18 KiB
Python
478 lines
18 KiB
Python
"""Internal event store service with CalDAV push sync.
|
|
|
|
Storage model: an event is anchored at ``start_dt`` and has an optional
|
|
``duration_minutes``. The end of the event is *derived* via
|
|
``Event.end_dt`` (a Python property), never stored. Callers may still
|
|
pass ``end_dt`` on writes for ergonomic compatibility — the service
|
|
converts to ``duration_minutes`` internally. This rules out the entire
|
|
"end before start" bug class structurally (Fable #160 / migration
|
|
0043). Open-ended events use ``duration_minutes = None``.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import uuid
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from dateutil.rrule import rrulestr
|
|
from sqlalchemy import or_, select
|
|
|
|
from fabledassistant.models import async_session
|
|
from fabledassistant.models.event import Event
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _normalize_duration(
|
|
*,
|
|
start_dt: datetime,
|
|
end_dt: datetime | None,
|
|
duration_minutes: int | None,
|
|
) -> int | None:
|
|
"""Reduce (end_dt, duration_minutes) inputs to a single canonical
|
|
``duration_minutes`` value.
|
|
|
|
Resolution order:
|
|
1. If ``duration_minutes`` is explicit, use it (validate >= 0).
|
|
If ``end_dt`` is also given, validate the two agree.
|
|
2. Otherwise, derive from ``end_dt - start_dt``.
|
|
3. Otherwise None (point event with no end).
|
|
|
|
Raises ``ValueError`` for any invalid combination — duration < 0,
|
|
end_dt < start_dt, or end_dt and duration_minutes inconsistent.
|
|
"""
|
|
if duration_minutes is not None:
|
|
if duration_minutes < 0:
|
|
raise ValueError(
|
|
f"duration_minutes must be >= 0, got {duration_minutes}"
|
|
)
|
|
if end_dt is not None:
|
|
expected = int((end_dt - start_dt).total_seconds() // 60)
|
|
if expected != duration_minutes:
|
|
raise ValueError(
|
|
f"end_dt ({end_dt.isoformat()}) implies "
|
|
f"{expected} minutes but duration_minutes={duration_minutes} "
|
|
f"was passed; pass only one or make them agree."
|
|
)
|
|
return duration_minutes
|
|
if end_dt is not None:
|
|
delta_seconds = (end_dt - start_dt).total_seconds()
|
|
if delta_seconds < 0:
|
|
raise ValueError(
|
|
f"end_dt ({end_dt.isoformat()}) must be at or after "
|
|
f"start_dt ({start_dt.isoformat()}); pass end_dt=None "
|
|
f"or omit it for point events."
|
|
)
|
|
return int(delta_seconds // 60)
|
|
return None
|
|
|
|
|
|
async def _localize_naive(user_id: int, dt: datetime | None) -> datetime | None:
|
|
"""Anchor a naive datetime in the user's timezone; pass tz-aware through.
|
|
|
|
Naive datetimes are the user's local wall-clock time (the MCP create/update
|
|
tools combine date+time without a zone). Attaching the user's tzinfo lets
|
|
asyncpg store the correct UTC instant, matching the REST/UI path.
|
|
"""
|
|
if dt is not None and dt.tzinfo is None:
|
|
from fabledassistant.services.tz import get_user_tz # noqa: PLC0415
|
|
return dt.replace(tzinfo=await get_user_tz(user_id))
|
|
return dt
|
|
|
|
|
|
async def create_event(
|
|
user_id: int,
|
|
title: str,
|
|
start_dt: datetime,
|
|
end_dt: datetime | None = None,
|
|
duration_minutes: int | None = None,
|
|
all_day: bool = False,
|
|
description: str = "",
|
|
location: str = "",
|
|
color: str = "",
|
|
recurrence: str | None = None,
|
|
project_id: int | None = None,
|
|
reminder_minutes: int | None = None,
|
|
# ``duration`` is a legacy alias kept for the calendar tool layer
|
|
# and CalDAV pass-through callers; promotes to duration_minutes
|
|
# when duration_minutes isn't otherwise specified.
|
|
duration: int | None = None,
|
|
attendees: list[str] | None = None,
|
|
calendar_name: str | None = None,
|
|
) -> Event:
|
|
"""Create an event in the DB, then fire a CalDAV push task.
|
|
|
|
Either ``end_dt`` or ``duration_minutes`` may be supplied; the
|
|
service converts to ``duration_minutes`` internally. Raises
|
|
``ValueError`` on invalid combinations (negative duration, end
|
|
before start, end/duration disagreement).
|
|
"""
|
|
if duration is not None and duration_minutes is None:
|
|
duration_minutes = duration
|
|
# Canonical localization point: a naive datetime (e.g. from the MCP tool's
|
|
# date+time split) is the user's wall-clock time, so anchor it in their
|
|
# timezone before storage. tz-aware inputs (REST, CalDAV pass-through) are
|
|
# left untouched. Without this, MCP-created events landed at the same
|
|
# wall-clock numerals in UTC and drifted from UI-created ones by the offset.
|
|
start_dt = await _localize_naive(user_id, start_dt)
|
|
end_dt = await _localize_naive(user_id, end_dt)
|
|
duration_minutes = _normalize_duration(
|
|
start_dt=start_dt, end_dt=end_dt, duration_minutes=duration_minutes,
|
|
)
|
|
uid = str(uuid.uuid4())
|
|
async with async_session() as session:
|
|
event = Event(
|
|
user_id=user_id,
|
|
uid=uid,
|
|
title=title,
|
|
start_dt=start_dt,
|
|
duration_minutes=duration_minutes,
|
|
all_day=all_day,
|
|
description=description,
|
|
location=location,
|
|
color=color,
|
|
recurrence=recurrence,
|
|
project_id=project_id,
|
|
reminder_minutes=reminder_minutes,
|
|
)
|
|
session.add(event)
|
|
await session.commit()
|
|
await session.refresh(event)
|
|
|
|
extra_fields = {
|
|
"duration": duration_minutes,
|
|
"reminder_minutes": reminder_minutes,
|
|
"attendees": attendees,
|
|
"calendar_name": calendar_name,
|
|
}
|
|
asyncio.create_task(_push_create(event, user_id, extra_fields))
|
|
return event
|
|
|
|
|
|
async def get_event(user_id: int, event_id: int) -> Event | None:
|
|
"""Return event owned by user_id, or None."""
|
|
async with async_session() as session:
|
|
result = await session.execute(
|
|
select(Event).where(
|
|
Event.id == event_id, Event.user_id == user_id, Event.deleted_at.is_(None)
|
|
)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def list_events(
|
|
user_id: int,
|
|
date_from: datetime,
|
|
date_to: datetime,
|
|
) -> list[dict]:
|
|
"""List events for user_id that overlap [date_from, date_to].
|
|
|
|
Recurring events (with an RRULE recurrence string) are expanded into
|
|
individual occurrences within the range. Non-recurring events are
|
|
returned as-is. All results are sorted by start time and returned as
|
|
dicts (same shape as ``Event.to_dict()``).
|
|
|
|
Filtering strategy: a coarse SQL prefilter (events that start on or
|
|
before ``date_to``), then refine in Python using the event's derived
|
|
end (``start_dt + duration_minutes``). Doing the end-of-event math
|
|
in SQL would require Postgres-specific interval arithmetic; the
|
|
Python-side refinement is a few row-loops over a small per-user
|
|
result set, which is fine for personal-scale data and avoids
|
|
coupling the query to a specific dialect.
|
|
"""
|
|
async with async_session() as session:
|
|
result = await session.execute(
|
|
select(Event)
|
|
.where(
|
|
Event.user_id == user_id,
|
|
Event.deleted_at.is_(None),
|
|
or_(
|
|
Event.recurrence.isnot(None),
|
|
Event.start_dt <= date_to,
|
|
),
|
|
)
|
|
.order_by(Event.start_dt)
|
|
)
|
|
events = list(result.scalars().all())
|
|
|
|
items: list[dict] = []
|
|
for event in events:
|
|
if event.recurrence:
|
|
duration = (
|
|
timedelta(minutes=event.duration_minutes)
|
|
if event.duration_minutes is not None
|
|
else None
|
|
)
|
|
try:
|
|
rule = rrulestr(event.recurrence, dtstart=event.start_dt, ignoretz=False)
|
|
occurrences = rule.between(date_from, date_to, inc=True)
|
|
except Exception:
|
|
logger.warning("Failed to expand RRULE for event %d: %r", event.id, event.recurrence)
|
|
# Fall back to canonical event row; still apply the
|
|
# window check so a far-future canonical row doesn't
|
|
# leak into today's list.
|
|
if date_from <= event.start_dt <= date_to:
|
|
items.append(event.to_dict())
|
|
continue
|
|
|
|
base = event.to_dict()
|
|
for occ in occurrences:
|
|
if occ.tzinfo is None:
|
|
occ = occ.replace(tzinfo=timezone.utc)
|
|
occurrence_dict = dict(base)
|
|
occurrence_dict["start_dt"] = occ.isoformat()
|
|
if duration is not None:
|
|
occurrence_dict["end_dt"] = (occ + duration).isoformat()
|
|
items.append(occurrence_dict)
|
|
continue
|
|
|
|
# Non-recurring: refine the coarse prefilter in Python using the
|
|
# derived end_dt. A point event (duration None) is included when
|
|
# its start is at or after date_from. A timed event is included
|
|
# when its end is at or after date_from.
|
|
derived_end = event.end_dt
|
|
if derived_end is None:
|
|
if event.start_dt >= date_from:
|
|
items.append(event.to_dict())
|
|
else:
|
|
if derived_end >= date_from:
|
|
items.append(event.to_dict())
|
|
|
|
items.sort(key=lambda x: x["start_dt"])
|
|
return items
|
|
|
|
|
|
async def search_events(
|
|
user_id: int,
|
|
query: str,
|
|
days_ahead: int = 90,
|
|
include_past: bool = False,
|
|
) -> list[Event]:
|
|
"""Search events by keyword in title, description, or location."""
|
|
now = datetime.now(timezone.utc)
|
|
q = f"%{query}%"
|
|
async with async_session() as session:
|
|
where = [
|
|
Event.user_id == user_id,
|
|
Event.deleted_at.is_(None),
|
|
or_(
|
|
Event.title.ilike(q),
|
|
Event.description.ilike(q),
|
|
Event.location.ilike(q),
|
|
),
|
|
]
|
|
if not include_past:
|
|
date_to = now + timedelta(days=days_ahead)
|
|
where.extend([Event.start_dt >= now, Event.start_dt <= date_to])
|
|
result = await session.execute(
|
|
select(Event).where(*where).order_by(Event.start_dt)
|
|
)
|
|
return result.scalars().all()
|
|
|
|
|
|
async def update_event(user_id: int, event_id: int, **fields) -> Event | None:
|
|
"""Partial update. Returns updated event or None if not found.
|
|
|
|
Accepts ``end_dt`` or ``duration_minutes`` (or both, validated for
|
|
agreement). The service converts to ``duration_minutes`` before
|
|
persisting; ``end_dt`` is never stored. Raises ``ValueError`` for
|
|
invalid combinations against the post-update state.
|
|
"""
|
|
async with async_session() as session:
|
|
result = await session.execute(
|
|
select(Event).where(
|
|
Event.id == event_id, Event.user_id == user_id,
|
|
Event.deleted_at.is_(None),
|
|
)
|
|
)
|
|
event = result.scalar_one_or_none()
|
|
if event is None:
|
|
return None
|
|
old_title = event.title # capture before mutation for CalDAV lookup
|
|
|
|
# Localize a naive start_dt patch to the user's timezone (same canonical
|
|
# rule as create_event) before it's used or persisted.
|
|
if fields.get("start_dt") is not None:
|
|
fields["start_dt"] = await _localize_naive(user_id, fields["start_dt"])
|
|
|
|
# Resolve any end_dt/duration_minutes inputs against the
|
|
# post-update start_dt. If neither is in the patch, leave the
|
|
# existing duration_minutes alone.
|
|
post_update_start = (
|
|
fields["start_dt"]
|
|
if fields.get("start_dt") is not None
|
|
else event.start_dt
|
|
)
|
|
if "end_dt" in fields or "duration_minutes" in fields:
|
|
new_end = fields.pop("end_dt", None)
|
|
new_duration = fields.pop("duration_minutes", None)
|
|
# If end_dt is in the patch but explicitly None, that's a
|
|
# clear → duration_minutes = None. Same shape duration_minutes=None.
|
|
if new_end is None and new_duration is None:
|
|
fields["duration_minutes"] = None
|
|
else:
|
|
fields["duration_minutes"] = _normalize_duration(
|
|
start_dt=post_update_start,
|
|
end_dt=new_end,
|
|
duration_minutes=new_duration,
|
|
)
|
|
|
|
allowed = {
|
|
"title", "start_dt", "duration_minutes", "all_day",
|
|
"description", "location", "color", "recurrence",
|
|
"project_id", "reminder_minutes",
|
|
}
|
|
# Nullable fields callers can explicitly clear by passing None
|
|
nullable = {
|
|
"duration_minutes", "recurrence", "project_id",
|
|
"reminder_minutes",
|
|
}
|
|
for key, value in fields.items():
|
|
if key in allowed and (value is not None or key in nullable):
|
|
setattr(event, key, value)
|
|
# Re-arm the reminder when the timing changes, so an event moved to a
|
|
# new (future) time — or given a new lead time — fires again instead of
|
|
# being permanently suppressed by a stale reminder_sent_at.
|
|
if "start_dt" in fields or "reminder_minutes" in fields:
|
|
event.reminder_sent_at = None
|
|
await session.commit()
|
|
await session.refresh(event)
|
|
|
|
asyncio.create_task(_push_update(event, user_id, old_title=old_title))
|
|
return event
|
|
|
|
|
|
async def delete_event(user_id: int, event_id: int) -> None:
|
|
"""Delete event. Fires CalDAV delete push if caldav_uid is set."""
|
|
async with async_session() as session:
|
|
result = await session.execute(
|
|
select(Event).where(Event.id == event_id, Event.user_id == user_id)
|
|
)
|
|
event = result.scalar_one_or_none()
|
|
if event is None:
|
|
return
|
|
caldav_uid = event.caldav_uid
|
|
event_title = event.title # needed to find the event on CalDAV by title
|
|
await session.delete(event)
|
|
await session.commit()
|
|
|
|
if caldav_uid:
|
|
asyncio.create_task(_push_delete(caldav_uid, event_title, user_id))
|
|
|
|
|
|
async def find_events_by_query(user_id: int, query: str) -> list[Event]:
|
|
"""ILIKE search on title — used by AI update/delete tools.
|
|
|
|
Returns upcoming events first (start_dt >= now), falling back to
|
|
past events so the AI operates on the most relevant match.
|
|
"""
|
|
q = f"%{query}%"
|
|
now = datetime.now(timezone.utc)
|
|
async with async_session() as session:
|
|
# Prefer events at or after now; fall back to past events
|
|
upcoming = (await session.execute(
|
|
select(Event).where(
|
|
Event.user_id == user_id,
|
|
Event.deleted_at.is_(None),
|
|
Event.title.ilike(q),
|
|
Event.start_dt >= now,
|
|
).order_by(Event.start_dt)
|
|
)).scalars().all()
|
|
if upcoming:
|
|
return list(upcoming)
|
|
past = (await session.execute(
|
|
select(Event).where(
|
|
Event.user_id == user_id,
|
|
Event.deleted_at.is_(None),
|
|
Event.title.ilike(q),
|
|
Event.start_dt < now,
|
|
).order_by(Event.start_dt.desc())
|
|
)).scalars().all()
|
|
return list(past)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CalDAV push helpers (fire-and-forget)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
async def _push_create(event: Event, user_id: int, extra: dict) -> None:
|
|
try:
|
|
from fabledassistant.services.caldav import (
|
|
create_event as caldav_create,
|
|
is_caldav_configured,
|
|
)
|
|
if not await is_caldav_configured(user_id):
|
|
return
|
|
derived_end = event.end_dt # property: start + duration_minutes
|
|
await caldav_create(
|
|
user_id=user_id,
|
|
title=event.title,
|
|
start=event.start_dt.isoformat(),
|
|
end=derived_end.isoformat() if derived_end else None,
|
|
description=event.description or None,
|
|
location=event.location or None,
|
|
all_day=event.all_day,
|
|
recurrence=event.recurrence,
|
|
uid=event.uid,
|
|
duration=extra.get("duration"),
|
|
reminder_minutes=extra.get("reminder_minutes"),
|
|
attendees=extra.get("attendees"),
|
|
calendar_name=extra.get("calendar_name"),
|
|
)
|
|
# Mark as synced
|
|
async with async_session() as session:
|
|
result = await session.execute(
|
|
select(Event).where(Event.id == event.id)
|
|
)
|
|
ev = result.scalar_one_or_none()
|
|
if ev:
|
|
ev.caldav_uid = event.uid
|
|
await session.commit()
|
|
except Exception:
|
|
logger.warning("CalDAV push (create) failed for event %d", event.id, exc_info=True)
|
|
|
|
|
|
async def _push_update(event: Event, user_id: int, old_title: str = "") -> None:
|
|
"""Push an update to CalDAV. Uses old_title to locate the event by its pre-rename SUMMARY."""
|
|
if not event.caldav_uid:
|
|
return
|
|
try:
|
|
from fabledassistant.services.caldav import (
|
|
update_event as caldav_update,
|
|
is_caldav_configured,
|
|
)
|
|
if not await is_caldav_configured(user_id):
|
|
return
|
|
# Use old_title so CalDAV can find the event even if the title was changed
|
|
query_title = old_title or event.title
|
|
derived_end = event.end_dt
|
|
await caldav_update(
|
|
user_id=user_id,
|
|
query=query_title,
|
|
title=event.title,
|
|
start=event.start_dt.isoformat(),
|
|
end=derived_end.isoformat() if derived_end else None,
|
|
description=event.description or None,
|
|
location=event.location or None,
|
|
# Propagate the (possibly cleared) RRULE so a local recurrence edit
|
|
# isn't overwritten by the stale remote rule on the next pull.
|
|
recurrence=event.recurrence,
|
|
)
|
|
except Exception:
|
|
logger.warning("CalDAV push (update) failed for event %d", event.id, exc_info=True)
|
|
|
|
|
|
async def _push_delete(caldav_uid: str, event_title: str, user_id: int) -> None:
|
|
"""Push a delete to CalDAV. Uses event_title to locate the event by SUMMARY."""
|
|
try:
|
|
from fabledassistant.services.caldav import (
|
|
delete_event as caldav_delete,
|
|
is_caldav_configured,
|
|
)
|
|
if not await is_caldav_configured(user_id):
|
|
return
|
|
await caldav_delete(user_id=user_id, query=event_title)
|
|
except Exception:
|
|
logger.warning("CalDAV push (delete) failed for uid %s", caldav_uid, exc_info=True)
|