"""Scheduler jobs for background maintenance tasks. - Reminder notifications: checks every 5 minutes for due event reminders and delivers them to the in-app notification feed. - CalDAV pull sync: runs every hour for all users with CalDAV configured. - Recurring-task spawn: every 15 minutes, creates the next occurrence of any recurring task whose spawn time has arrived. Uses the BackgroundScheduler pattern shared with the other *_scheduler modules. """ from __future__ import annotations import asyncio import logging from datetime import datetime, timedelta, timezone from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.interval import IntervalTrigger 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__) _scheduler: BackgroundScheduler | None = None _loop: asyncio.AbstractEventLoop | None = None # --------------------------------------------------------------------------- # Reminder job # --------------------------------------------------------------------------- async def _fire_reminders() -> None: """Fire in-app reminders for events whose reminder time has arrived. One-shot events fire once (gated on reminder_sent_at IS NULL). Recurring events fire once PER OCCURRENCE: reminder_sent_at stores the start of the occurrence we last reminded about, so each new occurrence re-arms the reminder instead of the whole series firing only once. """ now = datetime.now(timezone.utc) window_end = now + timedelta(minutes=5) async with async_session() as session: result = await session.execute( select(Event).where( Event.reminder_minutes.isnot(None), Event.deleted_at.is_(None), or_( # Recurring events are evaluated every sweep against their # next occurrence (the base start_dt is long past). Event.recurrence.isnot(None), # One-shot events: classic gate. and_(Event.reminder_sent_at.is_(None), Event.start_dt > now), ), ) ) candidates = list(result.scalars().all()) # (event_id, occurrence_start) — occurrence_start is also the dedup marker # written to reminder_sent_at, so a given occurrence reminds exactly once. to_notify: list[tuple[int, datetime]] = [] for event in candidates: if event.recurrence: try: rule = rrulestr(event.recurrence, dtstart=event.start_dt, ignoretz=False) occ = rule.after(now, inc=True) except Exception: logger.warning("Failed to expand RRULE for event %d reminder", event.id, exc_info=True) continue if occ is None: continue reminder_dt = occ - timedelta(minutes=event.reminder_minutes) if reminder_dt <= window_end and event.reminder_sent_at != occ: to_notify.append((event.id, occ)) else: reminder_dt = event.start_dt - timedelta(minutes=event.reminder_minutes) if reminder_dt <= window_end: to_notify.append((event.id, event.start_dt)) if not to_notify: return # Deliver via the in-app notification feed (push was removed in Phase 8). from fabledassistant.services.notifications import create_in_app_notification async with async_session() as session: for event_id, occurrence_start in to_notify: ev = (await session.execute( select(Event).where(Event.id == event_id) )).scalar_one_or_none() # Skip if this exact occurrence was already reminded (covers a # concurrent sweep and the one-shot already-sent case). if ev is None or ev.reminder_sent_at == occurrence_start: continue await create_in_app_notification(ev.user_id, "event_reminder", { "event_id": ev.id, "title": ev.title, "start_dt": occurrence_start.isoformat(), "url": "/calendar", }) # Stamp the occurrence marker only after the notification is # created, so a delivery failure leaves it eligible to retry. ev.reminder_sent_at = occurrence_start await session.commit() def _run_reminders(loop: asyncio.AbstractEventLoop) -> None: asyncio.run_coroutine_threadsafe(_fire_reminders(), loop) # --------------------------------------------------------------------------- # CalDAV pull sync job # --------------------------------------------------------------------------- async def _run_caldav_sync() -> None: from fabledassistant.services.caldav_sync import sync_all_users # noqa: PLC0415 try: await sync_all_users() except Exception: logger.warning("CalDAV pull sync job failed", exc_info=True) def _run_caldav_sync_threadsafe(loop: asyncio.AbstractEventLoop) -> None: asyncio.run_coroutine_threadsafe(_run_caldav_sync(), loop) # --------------------------------------------------------------------------- # Recurring-task spawn job # --------------------------------------------------------------------------- async def _run_recurrence_spawn() -> None: from fabledassistant.services.recurrence import spawn_recurring_tasks # noqa: PLC0415 try: await spawn_recurring_tasks() except Exception: logger.warning("Recurring-task spawn job failed", exc_info=True) def _run_recurrence_spawn_threadsafe(loop: asyncio.AbstractEventLoop) -> None: asyncio.run_coroutine_threadsafe(_run_recurrence_spawn(), loop) # --------------------------------------------------------------------------- # Lifecycle # --------------------------------------------------------------------------- def start_event_scheduler(loop: asyncio.AbstractEventLoop) -> None: global _scheduler, _loop if _scheduler is not None: return _loop = loop _scheduler = BackgroundScheduler() # Check reminders every 5 minutes _scheduler.add_job( _run_reminders, trigger=IntervalTrigger(minutes=5), args=[loop], id="event_reminders", replace_existing=True, ) # CalDAV pull sync every hour _scheduler.add_job( _run_caldav_sync_threadsafe, trigger=IntervalTrigger(hours=1), args=[loop], id="caldav_pull_sync", replace_existing=True, ) # Spawn the next occurrence of due recurring tasks every 15 minutes. # Without this job, recurrence_next_spawn_at is armed on completion but # never drained, so recurring tasks never recur. _scheduler.add_job( _run_recurrence_spawn_threadsafe, trigger=IntervalTrigger(minutes=15), args=[loop], id="recurrence_spawn", replace_existing=True, ) _scheduler.start() logger.info( "Event scheduler started (reminders every 5m, CalDAV sync every 1h, " "recurring-task spawn every 15m)" ) def stop_event_scheduler() -> None: global _scheduler if _scheduler is not None: _scheduler.shutdown(wait=False) _scheduler = None logger.info("Event scheduler stopped")