"""Daily missed-sweep background scheduler for the API service. The API service has no Home Assistant, so nobody re-fires the integration's day-rollover listener for it: this module owns the API plane's own background job. One :class:`MissedSweepScheduler` per app (installed on ``app.state`` by the lifespan) sleeps until the configured ``day_rollover_time`` — read fresh from the settings store at the top of EVERY cycle, so an admin's settings change applies on the next scheduling decision without a restart — then runs the HA-free core's :func:`nestquest_core.sweep.run_missed_sweep` for the household-local date (the stored ``timezone`` setting; the API host's local date when unset) and publishes each returned ``nestquest_quest_missed`` event on the app's ONE transition publisher (the same SSE stream every route publishes through). All sweep policy lives in the core (the watermark, the ``[watermark, today)`` window, the no-completion-event rule): this scheduler only decides WHEN. A same-night rerun — e.g. two workers, or a settings change moving the rollover past itself — fires nothing: the core's watermark makes the run an empty no-op. Testability: ``clock`` (an aware-now callable) and ``sleep`` (an awaitable delay) are injectable, so tests pin the time, fast-forward through the wait, and observe the computed delays without waiting a real day. Cancellation: :meth:`stop` cancels the loop task and waits for it, so the lifespan's shutdown is clean and a mid-run sweep is interrupted before the connection closes. A failed iteration (a database hiccup, a rejected setting) is logged and retried after a fixed backoff rather than silently killing the job. """ from __future__ import annotations import asyncio import datetime import logging from api.nestquest_core import core_settings_store, core_sweep LOGGER = logging.getLogger(__name__) #: Backoff after a failed scheduling cycle (settings read or sweep #: error), so a persistent failure retries every minute instead of #: spinning or dying. _RETRY_BACKOFF_SECONDS = 60.0 def _delay_seconds(target: datetime.time, now: datetime.datetime) -> float: """Seconds from ``now`` until the next occurrence of ``target``. ``target`` is a plain ``HH:MM`` time of day: the next wall-clock occurrence strictly AFTER ``now`` (an equal time schedules for tomorrow, so the just-run sweep is never instantly re-run). """ scheduled = now.replace( hour=target.hour, minute=target.minute, second=0, microsecond=0 ) if scheduled <= now: scheduled += datetime.timedelta(days=1) return (scheduled - now).total_seconds() class MissedSweepScheduler: """Run the missed sweep once per day at ``day_rollover_time``.""" def __init__( self, database: object, *, publisher: object | None = None, clock: object | None = None, sleep: object | None = None, ) -> None: """Create the scheduler (not yet running). ``database`` is the app's live :class:`~api.database.NestQuestDatabase`. ``publisher`` is the app's ONE :class:`~api.events.TransitionPublisher` (publishing is skipped when ``None``). ``clock`` defaults to the API host's local aware now (``datetime.datetime.now().astimezone()``) — the same single-clock discipline the routes' ``_local_now`` uses; ``sleep`` defaults to :func:`asyncio.sleep`. """ self._database = database self._publisher = publisher self._clock = clock if clock is not None else ( lambda: datetime.datetime.now().astimezone() ) self._sleep = sleep if sleep is not None else asyncio.sleep self._task: asyncio.Task | None = None def start(self) -> None: """Start the background loop (idempotent: a running one stays).""" if self._task is None or self._task.done(): self._task = asyncio.create_task( self._run_forever(), name="nestquest-missed-sweep" ) async def stop(self) -> None: """Cancel the loop and wait for it to finish (safe if never started).""" if self._task is not None: self._task.cancel() try: await self._task except asyncio.CancelledError: pass self._task = None @property def running(self) -> bool: """Whether the background loop task is alive.""" return self._task is not None and not self._task.done() async def _run_forever(self) -> None: """Schedule → sweep → repeat, forever (until cancelled).""" while True: try: settings = await core_settings_store.load_settings( self._database ) rollover = datetime.time.fromisoformat( settings.day_rollover_time ) except Exception: # Not silent: the failure is logged and retried after # the backoff, so one bad settings document or a # transient database error never kills the scheduler. LOGGER.exception( "NestQuest missed-sweep scheduler failed to resolve " "the day rollover time; retrying in %gs", _RETRY_BACKOFF_SECONDS, ) await self._sleep(_RETRY_BACKOFF_SECONDS) continue # The host may run UTC: the rollover is a household wall-clock # time, so the clock read is re-expressed in the stored zone. delay = _delay_seconds( rollover, settings.household_now(self._clock()) ) LOGGER.info( "NestQuest missed sweep scheduled for %s (%.0fs from now)", settings.day_rollover_time, delay, ) await self._sleep(delay) try: await self._run_once( settings.household_now(self._clock()).date() ) except Exception: # Not silent either: the sweep error is logged and the # loop backs off before the next cycle, whose watermark # re-read makes an already-swept day a no-op. LOGGER.exception( "NestQuest scheduled missed sweep failed; retrying in " "%gs", _RETRY_BACKOFF_SECONDS, ) await self._sleep(_RETRY_BACKOFF_SECONDS) async def _run_once(self, today: datetime.date) -> None: """Run ONE sweep for ``today`` and publish what it built. A failure propagates to :meth:`_run_forever`, which logs it and backs off before the next cycle; the watermark makes a re-run of an already-swept day announce nothing twice. """ events = await core_sweep.run_missed_sweep( self._database, today=today ) for event_type, payload in events: if self._publisher is not None: self._publisher.publish(event_type, payload) LOGGER.info( "NestQuest scheduled missed sweep for %s published %d event(s)", today.isoformat(), len(events), ) __all__ = ["MissedSweepScheduler"]