"""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 sqlalchemy import 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: """Find events with reminders due in the next 5 minutes and fire push notifications.""" 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.reminder_sent_at.is_(None), Event.start_dt > now, # event hasn't started yet # reminder fires when now >= start_dt - reminder_minutes # i.e. start_dt <= now + reminder_minutes (approximated by window_end check) ) ) candidates = list(result.scalars().all()) to_notify: list[Event] = [] for event in candidates: reminder_dt = event.start_dt - timedelta(minutes=event.reminder_minutes) if reminder_dt <= window_end: to_notify.append(event) 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 in to_notify: result = await session.execute( select(Event).where(Event.id == event.id) ) ev = result.scalar_one_or_none() if ev is None or ev.reminder_sent_at is not None: continue await create_in_app_notification(ev.user_id, "event_reminder", { "event_id": ev.id, "title": ev.title, "start_dt": ev.start_dt.isoformat() if ev.start_dt else None, "url": "/calendar", }) # Stamp only after the notification is created, so a delivery # failure leaves the reminder eligible to retry next sweep. ev.reminder_sent_at = datetime.now(timezone.utc) 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")