NestQuest/custom_components/nestquest/sse.py

"""HA-bus subscription over the API service's panel SSE stream.

The NestQuest API service (Feature 16) publishes every quest
transition on ``GET /api/v1/panel/events`` as Server-Sent Events
frames.  This module is the integration-side subscriber: ONE
background task consumes :meth:`~.api_client.NestQuestApiClient
.stream_events` and re-fires each frame on ``hass.bus`` with the SAME
event type and payload fields the local transition path
(:mod:`~.events`) emits — ``nestquest_quest_completed``,
``nestquest_quest_uncompleted``, ``nestquest_quest_missed`` and
``nestquest_child_day_complete`` — so the four blueprints fire
identically whether a completion happens through the integration's own
service or through the API service's panel route.

Failure handling: a failed stream (API down, network blip, non-2xx, a
dropped connection) is retried FOREVER with a bounded exponential
backoff — :data:`RECONNECT_MIN_SECONDS` after the first failure,
doubling up to :data:`RECONNECT_MAX_SECONDS` — so a long API outage
neither pins the event loop nor spams the log, and a frame received on
a healthy stream resets the backoff.  Nothing escapes the task: every
failure is logged with its reason and the loop reconnects; a
:class:`asyncio.CancelledError` (from :meth:`stop` or integration
unload) is the only thing that ends it, and it stops cleanly — no
frame fires after :meth:`stop` returns and the response is closed.
"""
from __future__ import annotations

import asyncio
from typing import Any

from .const import LOGGER

__all__ = [
    "NestQuestEventStream",
    "RECONNECT_MAX_SECONDS",
    "RECONNECT_MIN_SECONDS",
]

#: Delay before the FIRST reconnect attempt after a stream failure.
#: One failed frame must not turn into a hot reconnect loop.
RECONNECT_MIN_SECONDS = 5.0

#: Ceiling for the exponential reconnect backoff: a long API outage
#: retries at a calm fixed cadence instead of growing without bound.
RECONNECT_MAX_SECONDS = 60.0


class NestQuestEventStream:
    """Consume the panel SSE stream in the background, fire the HA bus.

    ONE instance per config entry, built in integration setup with the
    SAME API client the coordinator polls (the client is stateless —
    one shared session, one connection pool).  ``start()`` schedules
    the background task; ``stop()`` cancels and awaits it, so unload
    never leaks the task and no frame fires after teardown.  The task
    itself is failure-proof: any exception from the stream (typed
    :class:`~.api_client.NestQuestApiError` or anything else) is
    logged and retried after the current backoff, never allowed to
    escape the task.
    """

    def __init__(
        self,
        hass: Any,
        client: Any,
        *,
        reconnect_min_seconds: float = RECONNECT_MIN_SECONDS,
        reconnect_max_seconds: float = RECONNECT_MAX_SECONDS,
    ) -> None:
        """Store the hass bus target, the API client, and the backoff.

        The client is anything exposing ``stream_events()`` (the real
        :class:`~.api_client.NestQuestApiClient` in production, a stub
        in tests).  The backoff bounds must be positive and the minimum
        must not exceed the maximum — a misconfiguration should fail
        loudly at construction, not quietly reconnect at delay 0.
        """
        if (
            not isinstance(reconnect_min_seconds, (int, float))
            or isinstance(reconnect_min_seconds, bool)
            or reconnect_min_seconds <= 0
            or reconnect_min_seconds > reconnect_max_seconds
        ):
            raise ValueError(
                "reconnect_min_seconds must be a positive number no "
                f"greater than reconnect_max_seconds, got "
                f"{reconnect_min_seconds!r}"
            )
        if (
            not isinstance(reconnect_max_seconds, (int, float))
            or isinstance(reconnect_max_seconds, bool)
            or reconnect_max_seconds <= 0
        ):
            raise ValueError(
                f"reconnect_max_seconds must be a positive number, got "
                f"{reconnect_max_seconds!r}"
            )
        self._hass = hass
        self._client = client
        self._min_backoff = float(reconnect_min_seconds)
        self._max_backoff = float(reconnect_max_seconds)
        self._backoff = float(reconnect_min_seconds)
        self._task: asyncio.Task | None = None

    @property
    def task(self) -> asyncio.Task | None:
        """The background task, or ``None`` when never started/stopped."""
        return self._task

    def start(self) -> None:
        """Schedule the subscription task (idempotent while it runs)."""
        if self._task is not None and not self._task.done():
            return
        self._backoff = self._min_backoff
        self._task = asyncio.create_task(
            self._run(), name="nestquest-sse-subscription"
        )

    async def stop(self) -> None:
        """Cancel and await the subscription task.

        Returns only after the task has actually finished, so unload
        can rely on: no further frame is fired after ``stop()``, and
        the underlying stream response has been closed.  Safe to call
        when the task was never started or already stopped.
        """
        task, self._task = self._task, None
        if task is None:
            return
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            # The task's own cancellation surfaces here — expected and
            # swallowed; a cancelling the CALLER would re-raise out of
            # the surrounding await on its own.
            pass

    async def _run(self) -> None:
        """Consume stream frames forever, reconnecting on failure."""
        while True:
            try:
                async for event_type, payload in self._client.stream_events():
                    # A frame arrived on a healthy stream: the next
                    # failure reconnects quickly again.
                    self._backoff = self._min_backoff
                    try:
                        self._hass.bus.async_fire(event_type, dict(payload))
                    except Exception as err:
                        # Firing one frame must never tear down a
                        # healthy connection: log and keep consuming.
                        LOGGER.error(
                            "NestQuest panel event %s could not be fired "
                            "on the Home Assistant event bus: %s",
                            event_type,
                            err,
                        )
            except asyncio.CancelledError:
                # Unload / shutdown: stop cleanly, never converted.
                raise
            except Exception as err:
                delay = self._backoff
                self._backoff = min(
                    self._backoff * 2, self._max_backoff
                )
                LOGGER.warning(
                    "NestQuest panel event stream failed (%s); "
                    "reconnecting in %.1f s",
                    err,
                    delay,
                )
                await asyncio.sleep(delay)