houseplan-card/custom_components/houseplan/frontend_registration.py

"""Register the House Plan card with the Home Assistant frontend."""
from __future__ import annotations

import asyncio
import logging
from collections.abc import Callable
from dataclasses import dataclass, field
from pathlib import Path
from typing import Literal

from homeassistant.components import frontend, persistent_notification
from homeassistant.core import HomeAssistant, callback
from homeassistant.helpers.event import async_call_later
from homeassistant.helpers.start import async_at_started
from homeassistant.helpers.translation import async_get_translations

from .const import DOMAIN, FRONTEND_URL, VERSION
from .store import HouseplanConfigEntry

_LOGGER = logging.getLogger(__name__)

FRONTEND_REGISTRATION_KEY = "frontend_registration"
FRONTEND_STATIC_REGISTERED_KEY = "static_registered"
FRONTEND_FALLBACK_URLS_KEY = "frontend_fallback_urls"
FRONTEND_RELOAD_NOTICE_DATA_KEY = "frontend_reload_notice_created"
FRONTEND_RELOAD_NOTICE_ID = "houseplan_frontend_reload_notice"
FRONTEND_RETRY_DELAY_SECONDS = 1.0

_NOTICE_TITLE_KEY = "component.houseplan.issues.frontend_reload_notice.title"
_NOTICE_DESCRIPTION_KEY = (
    "component.houseplan.issues.frontend_reload_notice.description"
)
_NOTICE_TITLE_FALLBACK = "House Plan is ready"
_NOTICE_DESCRIPTION_FALLBACK = (
    "Open House Plan from the sidebar. After an update, fully reload the page "
    "to use the latest frontend (`Ctrl+F5` on Windows/Linux or `Cmd+Shift+R` "
    "on macOS). The dashboard card remains available; if you manage its "
    "resource manually in storage mode, open Settings → Dashboards → "
    "Resources. If the sidebar item is missing, check House Plan in System Health."
)

RegistrationStatus = Literal[
    "not_attempted",
    "created",
    "updated",
    "existing",
    "registry_pending",
    "yaml_fallback",
    "transient_error",
    "error_fallback",
]
ResourceLoader = Literal[
    "none",
    "lovelace_resource",
    "extra_module_url",
    "lovelace_resource_with_session_fallback",
]
ReloadNoticeStatus = Literal[
    "created",
    "already_created",
    "pending_frontend",
]


@dataclass(frozen=True, slots=True)
class RegistrationOutcome:
    """Result of one Lovelace resource registry attempt."""

    status: RegistrationStatus
    last_error: str | None = None


@dataclass(frozen=True, slots=True)
class StaticPathRegistrationOutcome:
    """Result of registering one exact public frontend URL."""

    registered: bool
    last_error: str | None = None


@dataclass(slots=True)
class FrontendRegistrationState:
    """Observable state and lifecycle handles for the current config-entry run."""

    card_file_present: bool
    static_path_registered: bool = False
    resource_status: RegistrationStatus = "not_attempted"
    loader: ResourceLoader = "none"
    module_url: str | None = None
    retry_pending: bool = False
    retry_attempted: bool = False
    last_error: str | None = None
    first_reload_notice: ReloadNoticeStatus = "pending_frontend"
    active: bool = True
    fallback_added: bool = False
    _cancel_started: Callable[[], None] | None = field(default=None, repr=False)
    _cancel_timer: Callable[[], None] | None = field(default=None, repr=False)
    _retry_task: asyncio.Task[None] | None = field(default=None, repr=False)
    _notice_lock: asyncio.Lock = field(default_factory=asyncio.Lock, repr=False)

    @callback
    def cancel(self) -> None:
        """Prevent all delayed work for this setup generation."""
        self.active = False
        self.retry_pending = False
        if self._cancel_started is not None:
            self._cancel_started()
            self._cancel_started = None
        if self._cancel_timer is not None:
            self._cancel_timer()
            self._cancel_timer = None
        if self._retry_task is not None and not self._retry_task.done():
            self._retry_task.cancel()
        self._retry_task = None


def _safe_error(phase: str, err: BaseException) -> str:
    """Return a support-safe failure fingerprint without message or path data."""
    return f"{phase}:{type(err).__name__}"


def _static_registered_urls(hass: HomeAssistant) -> set[str]:
    """Return the exact static URLs registered for this Core process.

    Before the panel entry existed this state was a boolean.  A legacy ``True``
    proves only that the card URL was registered; treating it as a generic
    truthy flag would incorrectly skip the panel route after an integration
    reload.
    """
    domain_data = hass.data.setdefault(DOMAIN, {})
    value = domain_data.get(FRONTEND_STATIC_REGISTERED_KEY)
    if value is True:
        urls = {FRONTEND_URL}
        domain_data[FRONTEND_STATIC_REGISTERED_KEY] = urls
        return urls
    if isinstance(value, set):
        return value
    if isinstance(value, dict):
        urls = {url for url, registered in value.items() if registered is True}
        domain_data[FRONTEND_STATIC_REGISTERED_KEY] = urls
        return urls
    return set()


async def async_register_frontend_static_path(
    hass: HomeAssistant,
    frontend_url: str,
    file_path: Path,
) -> StaticPathRegistrationOutcome:
    """Register one exact bundle entry across supported Home Assistant APIs."""
    registered_urls = _static_registered_urls(hass)
    if frontend_url in registered_urls:
        return StaticPathRegistrationOutcome(True)

    try:
        try:
            from homeassistant.components.http import StaticPathConfig
        except ImportError:  # Home Assistant versions before the async API
            hass.http.register_static_path(
                frontend_url, str(file_path), cache_headers=False
            )
        else:
            await hass.http.async_register_static_paths(
                [StaticPathConfig(frontend_url, str(file_path), cache_headers=False)]
            )
    except asyncio.CancelledError:
        raise
    except Exception as err:  # noqa: BLE001 - callers intentionally fail soft
        return StaticPathRegistrationOutcome(
            False, _safe_error("static_path", err)
        )

    registered_urls.add(frontend_url)
    hass.data.setdefault(DOMAIN, {})[
        FRONTEND_STATIC_REGISTERED_KEY
    ] = registered_urls
    return StaticPathRegistrationOutcome(True)


def _lovelace_resources(hass: HomeAssistant):
    """Return the Lovelace resource collection across supported HA versions."""
    lovelace = hass.data.get("lovelace")
    resources = getattr(lovelace, "resources", None)
    if resources is None and isinstance(lovelace, dict):
        resources = lovelace.get("resources")
    return resources


def _is_current(hass: HomeAssistant, state: FrontendRegistrationState) -> bool:
    """Return whether a delayed operation still belongs to the loaded entry."""
    return (
        state.active
        and hass.data.get(DOMAIN, {}).get(FRONTEND_REGISTRATION_KEY) is state
    )


def _owned_fallback_urls(hass: HomeAssistant) -> set[str]:
    """Return exact extra-module URLs owned for this Home Assistant run."""
    domain_data = hass.data.setdefault(DOMAIN, {})
    urls = domain_data.get(FRONTEND_FALLBACK_URLS_KEY)
    if not isinstance(urls, set):
        urls = set()
        domain_data[FRONTEND_FALLBACK_URLS_KEY] = urls
    return urls


def _active_extra_module_urls(hass: HomeAssistant) -> frozenset[str] | None:
    """Return HA's current module URLs, or None when the API is unknowable.

    ``add_extra_js_url`` is intentionally idempotent and has no ownership
    return value.  Looking before adding is therefore the only way to avoid
    claiming (and later removing) an identical URL configured by the user.
    Both the oldest supported HA and current HA expose ``UrlManager.urls``;
    unknown future/legacy shapes fail closed and are never claimed.
    """
    key = getattr(frontend, "DATA_EXTRA_MODULE_URL", None)
    if key is None:
        return None
    manager = hass.data.get(key)
    if manager is None:
        # Frontend setup normally creates the manager before this integration.
        # If it is genuinely absent, no configured URL can predate our add;
        # add_extra_js_url will either create/use its supported path or fail.
        return frozenset()
    urls = getattr(manager, "urls", None)
    if not isinstance(urls, (set, frozenset, list, tuple)):
        return None
    return frozenset(url for url in urls if isinstance(url, str))


async def _async_remove_duplicate_resources(
    resources,
    existing: list[dict],
    canonical_id: str,
    *,
    is_active: Callable[[], bool],
) -> None:
    """Best-effort remove legacy duplicates after choosing one authority."""
    remover = getattr(resources, "async_delete_item", None)
    if remover is None:
        return
    for item in existing:
        if item.get("id") == canonical_id:
            continue
        if not is_active():
            return
        try:
            await remover(item["id"])
        except asyncio.CancelledError:
            raise
        except Exception:  # noqa: BLE001 - canonical registration still succeeded
            _LOGGER.debug(
                "Could not remove a duplicate House Plan Lovelace resource",
                exc_info=True,
            )


async def async_register_lovelace_resource(
    hass: HomeAssistant,
    module_url: str,
    *,
    is_active: Callable[[], bool] | None = None,
) -> RegistrationOutcome:
    """Create or update the canonical Lovelace resource entry."""
    active = is_active or (lambda: True)
    resources = _lovelace_resources(hass)
    if resources is None:
        return RegistrationOutcome("registry_pending")
    if not hasattr(resources, "async_create_item"):
        return RegistrationOutcome("yaml_fallback")

    try:
        if not active():
            return RegistrationOutcome("registry_pending")
        if hasattr(resources, "async_get_info"):
            await resources.async_get_info()
        elif hasattr(resources, "loaded") and not resources.loaded:
            await resources.async_load()
            resources.loaded = True
        if not active():
            return RegistrationOutcome("registry_pending")

        existing = [
            item
            for item in resources.async_items()
            if str(item.get("url", "")).split("?", 1)[0] == FRONTEND_URL
        ]
        if existing:
            # Prefer an already-current entry when legacy duplicates exist. This
            # preserves its stable id while the remaining same-base entries are
            # removed best effort below.
            item = next(
                (candidate for candidate in existing if candidate.get("url") == module_url),
                existing[0],
            )
            if item.get("url") == module_url and item.get("type") == "module":
                await _async_remove_duplicate_resources(
                    resources,
                    existing,
                    item["id"],
                    is_active=active,
                )
                return RegistrationOutcome("existing")
            if not hasattr(resources, "async_update_item"):
                return RegistrationOutcome(
                    "transient_error", "registry_update:UnsupportedOperation"
                )
            if not active():
                return RegistrationOutcome("registry_pending")
            await resources.async_update_item(
                item["id"], {"url": module_url, "res_type": "module"}
            )
            await _async_remove_duplicate_resources(
                resources,
                existing,
                item["id"],
                is_active=active,
            )
            return RegistrationOutcome("updated")

        if not active():
            return RegistrationOutcome("registry_pending")
        await resources.async_create_item(
            {"url": module_url, "res_type": "module"}
        )
        _LOGGER.debug("House Plan card registered as a Lovelace resource")
        return RegistrationOutcome("created")
    except asyncio.CancelledError:
        raise
    except Exception as err:  # noqa: BLE001 - typed fallback is intentional
        _LOGGER.debug("Could not register the House Plan Lovelace resource", exc_info=True)
        return RegistrationOutcome("transient_error", _safe_error("registry", err))


def _add_fallback(
    hass: HomeAssistant, state: FrontendRegistrationState
) -> None:
    """Add this setup's exact versioned extra-module URL, best effort."""
    if state.module_url is None or not _is_current(hass, state):
        return
    owned_urls = _owned_fallback_urls(hass)
    if state.module_url in owned_urls:
        state.fallback_added = True
        state.loader = "extra_module_url"
        return
    active_before = _active_extra_module_urls(hass)
    if active_before is not None and state.module_url in active_before:
        # The exact URL predates this setup. It is a working loader, but it is
        # not ours to remove after registry recovery or on uninstall.
        state.fallback_added = True
        state.loader = "extra_module_url"
        return
    try:
        frontend.add_extra_js_url(hass, state.module_url)
    except Exception as err:  # noqa: BLE001 - frontend setup must remain available
        state.loader = "none"
        state.last_error = _safe_error("fallback_add", err)
        _LOGGER.warning("Could not add the House Plan frontend fallback", exc_info=True)
        return
    if active_before is not None:
        owned_urls.add(state.module_url)
    else:
        _LOGGER.debug(
            "House Plan could not prove ownership of the frontend fallback; "
            "it will remain until Home Assistant restarts"
        )
    state.fallback_added = True
    state.loader = "extra_module_url"


def _remove_owned_fallback_url(
    hass: HomeAssistant,
    state: FrontendRegistrationState,
    module_url: str,
) -> bool:
    """Remove one exact integration-owned fallback when HA supports it."""
    owned_urls = _owned_fallback_urls(hass)
    if module_url not in owned_urls:
        return True
    remover = getattr(frontend, "remove_extra_js_url", None)
    if remover is None:
        return False
    try:
        remover(hass, module_url)
    except KeyError:
        # Another lifecycle path already removed the exact URL.
        owned_urls.discard(module_url)
        return True
    except Exception as err:  # noqa: BLE001 - registry remains authoritative
        state.last_error = _safe_error("fallback_remove", err)
        _LOGGER.debug("Could not remove the House Plan frontend fallback", exc_info=True)
        return False
    owned_urls.discard(module_url)
    return True


def _remove_fallback(
    hass: HomeAssistant, state: FrontendRegistrationState
) -> bool:
    """Remove this setup's exact fallback when the running HA supports it."""
    if state.module_url is None:
        return True
    owned_urls = _owned_fallback_urls(hass)
    if state.module_url in owned_urls:
        removed = _remove_owned_fallback_url(hass, state, state.module_url)
        state.fallback_added = not removed
        return removed

    # A pre-existing/user-owned or uninspectable exact URL must never be
    # removed. Report the combined loader while it is known (or conservatively
    # assumed) to remain active in this HA document lifecycle.
    active = _active_extra_module_urls(hass)
    state.fallback_added = active is None or state.module_url in active
    return not state.fallback_added


async def _async_create_reload_notice(
    hass: HomeAssistant,
    entry: HouseplanConfigEntry,
    state: FrontendRegistrationState,
) -> None:
    """Create and persist the one-time localized hard-reload instruction."""
    # Initial fallback setup and the one-second retry may overlap when loading a
    # translation is slow. Serialize the persisted one-shot contract rather
    # than relying only on the stable notification ID (which prevents two UI
    # rows but would still perform and persist two creates).
    async with state._notice_lock:
        if entry.data.get(FRONTEND_RELOAD_NOTICE_DATA_KEY) is True:
            if state.first_reload_notice != "created":
                state.first_reload_notice = "already_created"
            return
        if (
            not _is_current(hass, state)
            or not state.card_file_present
            or not state.static_path_registered
            or state.loader == "none"
        ):
            state.first_reload_notice = "pending_frontend"
            return

        title = _NOTICE_TITLE_FALLBACK
        description = _NOTICE_DESCRIPTION_FALLBACK
        try:
            translations = await async_get_translations(
                hass,
                hass.config.language,
                "issues",
                integrations={DOMAIN},
            )
            title = translations.get(_NOTICE_TITLE_KEY, title)
            description = translations.get(_NOTICE_DESCRIPTION_KEY, description)
        except asyncio.CancelledError:
            raise
        except Exception:  # noqa: BLE001 - built-in English fallback is intentional
            _LOGGER.debug(
                "Could not load the House Plan reload notice translation",
                exc_info=True,
            )

        if not _is_current(hass, state):
            return
        try:
            persistent_notification.async_create(
                hass,
                description,
                title=title,
                notification_id=FRONTEND_RELOAD_NOTICE_ID,
            )
        except Exception:  # noqa: BLE001 - do not consume flag on failure
            state.first_reload_notice = "pending_frontend"
            _LOGGER.warning(
                "Could not create the House Plan reload notice", exc_info=True
            )
            return
        try:
            hass.config_entries.async_update_entry(
                entry,
                data={**entry.data, FRONTEND_RELOAD_NOTICE_DATA_KEY: True},
            )
        except Exception:  # noqa: BLE001 - roll back the unpersisted notice
            # A stable ID prevents duplicate rows, but dismissing the notice is
            # still important: without a durable flag, displaying it would make
            # System Health claim that the one-shot contract is pending while a
            # user-visible notice already exists. A later setup/retry can now
            # attempt the complete create+persist operation honestly.
            try:
                persistent_notification.async_dismiss(
                    hass, FRONTEND_RELOAD_NOTICE_ID
                )
            except Exception:  # noqa: BLE001 - stable ID remains a safe fallback
                _LOGGER.debug(
                    "Could not roll back an unpersisted House Plan reload notice",
                    exc_info=True,
                )
            state.first_reload_notice = "pending_frontend"
            _LOGGER.warning(
                "Could not persist the House Plan reload notice state",
                exc_info=True,
            )
            return
        state.first_reload_notice = "created"


async def _async_retry_registration(
    hass: HomeAssistant,
    entry: HouseplanConfigEntry,
    state: FrontendRegistrationState,
) -> None:
    """Perform the single delayed resource-registry retry."""
    if not _is_current(hass, state) or state.module_url is None:
        return
    state.retry_pending = False
    state.retry_attempted = True
    outcome = await async_register_lovelace_resource(
        hass,
        state.module_url,
        is_active=lambda: _is_current(hass, state),
    )
    if not _is_current(hass, state):
        return

    if outcome.status in {"created", "updated", "existing"}:
        state.resource_status = outcome.status
        state.last_error = outcome.last_error
        state.loader = (
            "lovelace_resource"
            if _remove_fallback(hass, state)
            else "lovelace_resource_with_session_fallback"
        )
    elif outcome.status == "yaml_fallback":
        state.resource_status = "yaml_fallback"
        state.last_error = outcome.last_error
        _add_fallback(hass, state)
    else:
        state.resource_status = "error_fallback"
        state.last_error = outcome.last_error or "registry_retry:Unavailable"
        _add_fallback(hass, state)

    await _async_create_reload_notice(hass, entry, state)


def _schedule_retry(
    hass: HomeAssistant,
    entry: HouseplanConfigEntry,
    state: FrontendRegistrationState,
) -> None:
    """Schedule exactly one lifecycle-bound retry after HA start plus one second."""
    state.retry_pending = True

    @callback
    def _after_started(_event=None) -> None:
        if not _is_current(hass, state):
            return
        state._cancel_started = None

        @callback
        def _after_delay(_now=None) -> None:
            if not _is_current(hass, state):
                return
            state._cancel_timer = None
            state._retry_task = entry.async_create_background_task(
                hass,
                _async_retry_registration(hass, entry, state),
                "houseplan frontend resource retry",
            )

        state._cancel_timer = async_call_later(
            hass, FRONTEND_RETRY_DELAY_SECONDS, _after_delay
        )

    state._cancel_started = async_at_started(hass, _after_started)


async def async_setup_frontend_registration(
    hass: HomeAssistant,
    entry: HouseplanConfigEntry,
    card_path: Path,
) -> FrontendRegistrationState:
    """Register static/card resources without blocking the rest of entry setup."""
    domain_data = hass.data.setdefault(DOMAIN, {})
    previous = domain_data.get(FRONTEND_REGISTRATION_KEY)
    if isinstance(previous, FrontendRegistrationState):
        previous.cancel()

    card_file_present = card_path.is_file()
    module_url = f"{FRONTEND_URL}?v={VERSION}" if card_file_present else None
    active_fallbacks = _active_extra_module_urls(hass)
    fallback_added = module_url is not None and (
        module_url in _owned_fallback_urls(hass)
        or (active_fallbacks is not None and module_url in active_fallbacks)
    )
    state = FrontendRegistrationState(
        card_file_present=card_file_present,
        loader="extra_module_url" if fallback_added else "none",
        module_url=module_url,
        first_reload_notice=(
            "already_created"
            if entry.data.get(FRONTEND_RELOAD_NOTICE_DATA_KEY) is True
            else "pending_frontend"
        ),
        fallback_added=fallback_added,
    )
    domain_data[FRONTEND_REGISTRATION_KEY] = state
    entry.async_on_unload(state.cancel)

    if card_file_present:
        static_outcome = await async_register_frontend_static_path(
            hass, FRONTEND_URL, card_path
        )
        state.static_path_registered = static_outcome.registered
        if not static_outcome.registered:
            state.last_error = static_outcome.last_error
            _LOGGER.warning(
                "Could not register the House Plan card static path (%s)",
                static_outcome.last_error or "static_path:UnknownError",
            )

    if not card_file_present:
        _LOGGER.warning("houseplan-card.js was not found next to the integration")
        return state
    if not state.static_path_registered or state.module_url is None:
        return state

    outcome = await async_register_lovelace_resource(
        hass,
        state.module_url,
        is_active=lambda: _is_current(hass, state),
    )
    if not _is_current(hass, state):
        return state
    state.resource_status = outcome.status
    state.last_error = outcome.last_error

    if outcome.status in {"created", "updated", "existing"}:
        state.loader = (
            "lovelace_resource"
            if _remove_fallback(hass, state)
            else "lovelace_resource_with_session_fallback"
        )
        _LOGGER.info(
            "House Plan card auto-registered as a Lovelace resource: %s",
            state.module_url,
        )
    else:
        _add_fallback(hass, state)
        if outcome.status in {"registry_pending", "transient_error"}:
            _schedule_retry(hass, entry, state)
        else:
            _LOGGER.info(
                "House Plan card uses the frontend extra-module fallback: %s",
                state.module_url,
            )

    await _async_create_reload_notice(hass, entry, state)
    return state


@callback
def get_frontend_registration_state(
    hass: HomeAssistant,
) -> FrontendRegistrationState | None:
    """Return current frontend registration state for System Health."""
    state = hass.data.get(DOMAIN, {}).get(FRONTEND_REGISTRATION_KEY)
    return state if isinstance(state, FrontendRegistrationState) else None


async def async_remove_frontend_registration(
    hass: HomeAssistant, entry: HouseplanConfigEntry
) -> None:
    """Remove all House Plan resource entries and its one-time notice."""
    state = get_frontend_registration_state(hass)
    if state is not None:
        state.cancel()
        domain_data = hass.data.get(DOMAIN, {})
        if domain_data.get(FRONTEND_REGISTRATION_KEY) is state:
            domain_data.pop(FRONTEND_REGISTRATION_KEY, None)

    # Fallback ownership belongs to this HA run, not to the latest config-entry
    # generation. A bundle-missing reload can therefore have no current URL
    # while an earlier exact URL is still registered. Uninstall cleans every
    # URL that this integration itself added, and never pattern-matches foreign
    # extra-module entries.
    cleanup_state = state or FrontendRegistrationState(card_file_present=False)
    for module_url in tuple(_owned_fallback_urls(hass)):
        _remove_owned_fallback_url(hass, cleanup_state, module_url)

    try:
        persistent_notification.async_dismiss(hass, FRONTEND_RELOAD_NOTICE_ID)
    except Exception:  # noqa: BLE001 - uninstall cleanup is best effort
        _LOGGER.debug("Could not dismiss the House Plan reload notice", exc_info=True)

    try:
        resources = _lovelace_resources(hass)
        if resources is None or not hasattr(resources, "async_delete_item"):
            return
        if hasattr(resources, "async_get_info"):
            await resources.async_get_info()
        for item in list(resources.async_items()):
            if str(item.get("url", "")).split("?", 1)[0] != FRONTEND_URL:
                continue
            try:
                await resources.async_delete_item(item["id"])
            except asyncio.CancelledError:
                raise
            except Exception:  # noqa: BLE001 - continue best-effort cleanup
                _LOGGER.debug(
                    "Could not remove one House Plan Lovelace resource",
                    exc_info=True,
                )
                continue
            _LOGGER.debug("House Plan Lovelace resource removed")
    except asyncio.CancelledError:
        raise
    except Exception:  # noqa: BLE001 - uninstall cleanup is best effort
        _LOGGER.debug("Could not remove the House Plan Lovelace resource", exc_info=True)