NestQuest/api/scheduler.py

"""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


#: Upper bound on the minute-by-minute search past a nonexistent wall
#: time; real DST gaps are an hour, the largest ever skipped a day.
_MAX_GAP_MINUTES = 2 * 24 * 60


def _resolve_wall_time(
    day: datetime.date, target: datetime.time, zone: datetime.tzinfo
) -> datetime.datetime:
    """The aware instant ``target`` (``HH:MM``) names on ``day`` in ``zone``.

    A nonexistent wall time (inside a spring-forward gap) resolves to
    the first valid wall minute at/after it — the gap's end, e.g. 02:30
    on a 02:00→03:00 day becomes 03:00.  An ambiguous wall time (the
    repeated fall-back hour) resolves to its FIRST occurrence
    (``fold=0``).
    """
    wall = datetime.datetime.combine(
        day, datetime.time(target.hour, target.minute)
    )
    for _ in range(_MAX_GAP_MINUTES):
        candidate = wall.replace(tzinfo=zone)
        round_trip = candidate.astimezone(datetime.timezone.utc).astimezone(
            zone
        )
        if round_trip.replace(tzinfo=None) == wall:
            return candidate
        wall += datetime.timedelta(minutes=1)
    raise ValueError(f"no valid wall time at/after {target} on {day}")


def _delay_seconds(target: datetime.time, now: datetime.datetime) -> float:
    """Real seconds from ``now`` until the next occurrence of ``target``.

    ``target`` is a plain ``HH:MM`` household time of day, resolved in
    ``now``'s zone by :func:`_resolve_wall_time` (nonexistent → first
    valid instant after the gap; ambiguous → first occurrence).  The
    result is the next such instant strictly AFTER ``now`` (an equal
    time schedules for tomorrow, so the just-run sweep is never
    instantly re-run).  Both instants are compared and subtracted in
    UTC: same-``tzinfo`` aware arithmetic is wall-clock arithmetic and
    would be an hour off across a DST transition.
    """
    now_utc = now.astimezone(datetime.timezone.utc)
    day = now.date()
    while True:
        scheduled = _resolve_wall_time(day, target, now.tzinfo)
        scheduled_utc = scheduled.astimezone(datetime.timezone.utc)
        if scheduled_utc > now_utc:
            return (scheduled_utc - now_utc).total_seconds()
        day += datetime.timedelta(days=1)


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"]