"""Internal event store service with CalDAV push sync.""" from __future__ import annotations import asyncio import logging import uuid from datetime import datetime, timedelta, timezone from dateutil.rrule import rrulestr from sqlalchemy import and_, or_, select from fabledassistant.models import async_session from fabledassistant.models.event import Event logger = logging.getLogger(__name__) async def create_event( user_id: int, title: str, start_dt: datetime, end_dt: datetime | 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, # CalDAV-only fields (not stored in DB, forwarded to push) 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.""" uid = str(uuid.uuid4()) async with async_session() as session: event = Event( user_id=user_id, uid=uid, title=title, start_dt=start_dt, end_dt=end_dt, 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, "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) ) 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()). """ async with async_session() as session: # Match strategy: # - Recurring events: fetch all, expand via rrule below. # - Non-recurring with an end_dt: standard overlap — starts before # date_to and ends after date_from. # - Non-recurring with no end_dt: treat as a point event at # start_dt, include only if start_dt falls within the window. # (Previously this branch matched any event with a null end_dt, # returning all past events as "happening today".) result = await session.execute( select(Event).where( Event.user_id == user_id, or_( Event.recurrence.isnot(None), and_( Event.recurrence.is_(None), Event.start_dt <= date_to, or_( Event.end_dt >= date_from, and_( Event.end_dt.is_(None), Event.start_dt >= date_from, ), ), ), ), ).order_by(Event.start_dt) ) events = list(result.scalars().all()) items: list[dict] = [] for event in events: if not event.recurrence: items.append(event.to_dict()) continue # Expand recurring event occurrences within [date_from, date_to] duration = (event.end_dt - event.start_dt) if event.end_dt 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) items.append(event.to_dict()) continue base = event.to_dict() for occ in occurrences: # Ensure occurrence is UTC-aware 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) 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, 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.""" 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 None old_title = event.title # capture before mutation for CalDAV lookup allowed = {"title", "start_dt", "end_dt", "all_day", "description", "location", "color", "recurrence", "project_id", "reminder_minutes"} # Nullable fields that callers can explicitly set to None to clear nullable = {"end_dt", "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) 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.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.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 await caldav_create( user_id=user_id, title=event.title, start=event.start_dt.isoformat(), end=event.end_dt.isoformat() if event.end_dt 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 await caldav_update( user_id=user_id, query=query_title, title=event.title, start=event.start_dt.isoformat(), end=event.end_dt.isoformat() if event.end_dt else None, description=event.description or None, location=event.location or None, ) 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)