"""Home Assistant subscription manager.

Phase 1 of the HA provider lives here. For every enabled ``home_assistant``
:class:`ServiceProvider` row in the DB, this module spawns one long-running
asyncio task that:

1. Builds an :class:`HomeAssistantServiceProvider` from the provider row.
2. Calls ``provider.subscribe(emit)`` which loops forever — connect,
   authenticate, subscribe, drain events through ``emit`` — and reconnects
   with exponential backoff on any drop.
3. Each ``emit`` call hands the parsed :class:`ServiceEvent` to
   :func:`dispatch_provider_event` (the shared dispatch helper that webhook
   providers also use), so quiet hours, deferred dispatch, and event-log
   writes all behave identically to the rest of the system.

Lifecycle is owned by ``main.py`` via :func:`start_all` and :func:`stop_all`.
Phase 1 does not reconcile against DB changes after boot — adding,
modifying, or removing a HA provider requires a server restart. Phase 1.5
will add a CRUD-triggered :func:`reload_provider` hook.
"""

from __future__ import annotations

import asyncio
import logging
from typing import Any

from sqlmodel import select
from sqlmodel.ext.asyncio.session import AsyncSession

from notify_bridge_core.models.events import ServiceEvent
from notify_bridge_core.providers.home_assistant import (
    HomeAssistantAuthError,
    HomeAssistantServiceProvider,
)

from ..database.engine import get_engine
from ..database.models import EventLog, ServiceProvider
from .event_dispatch import dispatch_provider_event
from .http_session import get_http_session

_LOGGER = logging.getLogger(__name__)


# Per-provider running task. Keyed by provider_id so reload_provider() can
# find and replace a single task without disturbing the rest.
_running_tasks: dict[int, asyncio.Task[None]] = {}


# Keys from ``event.extra`` to copy into ``EventLog.details``. Anything not
# in this list is still available to templates via the merged extras, but
# the event-log row stays slim.
_HA_DETAIL_KEYS: tuple[str, ...] = (
    "entity_id",
    "friendly_name",
    "domain",
    "old_state",
    "new_state",
    "device_class",
    "unit_of_measurement",
    "area",
    "ha_event_type",
    "automation_name",
    "service_called",
    "target_entity",
)


def _ha_passes_filters(event: ServiceEvent, filters: dict[str, Any]) -> bool:
    """HA-specific tracker filter.

    Three filter keys, all optional, evaluated as a union: if the entity
    matches any one of them, the event passes. Empty filters mean "accept
    everything" — different from the Gitea filter which is an intersection.

    Filter shape:

    * ``collections`` — list of exact ``entity_id`` matches.
    * ``entity_glob`` — list of glob patterns (``light.*``, ``*_motion``).
    * ``domain_allowlist`` — list of HA domain prefixes (``light``).
    """
    collections = filters.get("collections") or []
    entity_globs = filters.get("entity_glob") or []
    domain_allowlist = filters.get("domain_allowlist") or []

    # No filters configured = accept everything.
    if not collections and not entity_globs and not domain_allowlist:
        return True

    entity_id = event.collection_id
    domain = event.extra.get("domain") or (
        entity_id.split(".", 1)[0] if "." in entity_id else ""
    )

    if collections and entity_id in collections:
        return True
    if domain_allowlist and domain in domain_allowlist:
        return True
    if entity_globs:
        from fnmatch import fnmatchcase
        for pattern in entity_globs:
            if isinstance(pattern, str) and fnmatchcase(entity_id, pattern):
                return True
    return False


async def _record_ha_status(
    *,
    provider_id: int,
    provider_name: str,
    state: str,
    detail: str | None,
) -> None:
    """Persist an HA connection-status transition as an EventLog row.

    Used by the supervisor's ``on_status_change`` callback so the
    dashboard surfaces "HA disconnected" / "HA reconnected" events
    alongside normal HA state changes. Best-effort: any DB failure is
    logged and swallowed so the WS reader path remains untouched.
    """
    engine = get_engine()
    try:
        async with AsyncSession(engine) as session:
            session.add(EventLog(
                user_id=None,  # provider-level event, no per-tracker owner
                tracker_id=None,
                tracker_name="",
                provider_id=provider_id,
                provider_name=provider_name,
                event_type=f"ha_status_{state}",
                collection_id="",
                collection_name="",
                assets_count=0,
                details={
                    "provider_type": "home_assistant",
                    "ha_status": state,
                    "ha_status_detail": detail or "",
                },
            ))
            await session.commit()
    except Exception:  # noqa: BLE001
        _LOGGER.exception(
            "Failed to persist HA status row for provider %s", provider_id,
        )


async def _run_provider(provider_id: int) -> None:
    """One per-provider supervisor loop.

    Reloads the provider row each iteration so config changes (URL, token,
    event types) take effect on the next reconnect cycle — no need for a
    full restart in the simple case where only credentials changed.

    The ``_emit`` closure is rebuilt every iteration. Its lifetime equals
    one ``subscribe()`` call: the callback only runs while the HA client's
    drain task is alive. ``provider_name`` is snapshotted at the start of
    each (re)connect cycle, so renames take effect on the next reconnect —
    chatty enough for v1; revisit if longer-lived WS sessions need fresher
    names mid-stream.
    """
    assert provider_id is not None, "_run_provider requires a real provider id"
    engine = get_engine()
    while True:
        try:
            async with AsyncSession(engine) as session:
                row = await session.get(ServiceProvider, provider_id)
                if row is None or row.type != "home_assistant":
                    _LOGGER.info(
                        "HA provider %s removed or retyped, stopping supervisor",
                        provider_id,
                    )
                    return
                config = dict(row.config or {})
                provider_name = row.name

            url = config.get("url", "")
            access_token = config.get("access_token", "")
            verify_tls = bool(config.get("verify_tls", True))
            event_types = config.get("event_types") or None

            if not url or not access_token:
                _LOGGER.warning(
                    "HA provider %s missing url or access_token; retrying in 60s",
                    provider_id,
                )
                await asyncio.sleep(60)
                continue

            session_http = await get_http_session()
            ha_provider = HomeAssistantServiceProvider(
                session=session_http,
                url=url,
                access_token=access_token,
                verify_tls=verify_tls,
                event_types=event_types,
                name=provider_name,
            )

            async def _emit(event: ServiceEvent) -> None:
                # Shield the DB-writing dispatch from external cancellation
                # (shutdown, supervisor restart). Without an inner Task,
                # ``asyncio.shield(coro)`` cancels the underlying coroutine
                # when the outer awaiter is cancelled — defeating the point
                # of the shield. We wrap explicitly and *drain* the inner
                # task on cancellation so the transaction completes.
                #
                # Perf note (Phase 2 follow-up): dispatch_provider_event
                # opens a fresh AsyncSession per call. For HA's chatty
                # state_changed bus this hammers the pool; batch in a
                # follow-up.
                inner = asyncio.create_task(dispatch_provider_event(
                    engine=engine,
                    provider_id=provider_id,
                    provider_name=provider_name,
                    provider_config=config,
                    event=event,
                    detail_keys=_HA_DETAIL_KEYS,
                    filter_fn=_ha_passes_filters,
                ))
                try:
                    await asyncio.shield(inner)
                except asyncio.CancelledError:
                    # Drain the in-flight write before re-raising so the DB
                    # row commits cleanly (or fails cleanly) instead of
                    # being torn down at an arbitrary await point.
                    try:
                        await inner
                    except Exception:  # noqa: BLE001
                        _LOGGER.exception(
                            "HA dispatch raised while draining shielded "
                            "task for provider %s", provider_id,
                        )
                    raise
                except Exception:  # noqa: BLE001
                    _LOGGER.exception(
                        "Failed to dispatch HA event for provider %s",
                        provider_id,
                    )

            def _on_status_change(state: str, detail: str | None) -> None:
                """Persist HA WS connect/disconnect transitions as event_log rows.

                The client invokes this synchronously from inside the WS
                run loop, so we can't ``await`` here. Schedule a fire-and-
                forget task on the same loop instead — log failures, never
                propagate them back into the WS reader.
                """
                try:
                    asyncio.create_task(_record_ha_status(
                        provider_id=provider_id,
                        provider_name=provider_name,
                        state=state,
                        detail=detail,
                    ))
                except RuntimeError:
                    # No running loop (shouldn't happen in normal operation).
                    _LOGGER.debug(
                        "Skipped HA status row for provider %s: no event loop",
                        provider_id,
                    )

            _LOGGER.info(
                "Starting HA subscription for provider %s (%s)",
                provider_id, provider_name,
            )
            await ha_provider.subscribe(_emit, on_status_change=_on_status_change)
        except asyncio.CancelledError:
            raise
        except HomeAssistantAuthError as err:
            # Fatal at the provider level — bad token. Sleep long and retry
            # so the user has time to fix the token without us hammering HA.
            # Error string already redacted by the client before re-raise.
            _LOGGER.error(
                "HA provider %s auth failed: %s — retrying in 5 minutes",
                provider_id, err,
            )
            await asyncio.sleep(300)
        except Exception:  # noqa: BLE001
            _LOGGER.exception(
                "HA supervisor for provider %s crashed; restarting in 30s",
                provider_id,
            )
            await asyncio.sleep(30)


def _make_done_callback(provider_id: int):
    """Return a done-callback that prunes the task from ``_running_tasks``.

    Without this, finished supervisors (provider deleted, fatal auth error
    after long sleep) leave stale entries in the dict — across many
    reload cycles the dict would grow unboundedly. The callback is
    registered on every task spawned via ``start_all`` / ``reload_provider``.
    """
    def _cb(task: asyncio.Task[None]) -> None:
        current = _running_tasks.get(provider_id)
        if current is task:
            _running_tasks.pop(provider_id, None)
    return _cb


async def start_all() -> None:
    """Start a supervisor task for every enabled HA provider."""
    engine = get_engine()
    async with AsyncSession(engine) as session:
        result = await session.exec(
            select(ServiceProvider).where(
                ServiceProvider.type == "home_assistant",
            )
        )
        providers = result.all()

    for prov in providers:
        if prov.id in _running_tasks and not _running_tasks[prov.id].done():
            continue
        task = asyncio.create_task(
            _run_provider(prov.id),
            name=f"ha-subscription-{prov.id}",
        )
        task.add_done_callback(_make_done_callback(prov.id))
        _running_tasks[prov.id] = task
    if providers:
        _LOGGER.info(
            "Started HA subscription manager: %d provider(s)", len(providers),
        )


async def stop_all() -> None:
    """Cancel every HA supervisor task and wait for clean shutdown."""
    if not _running_tasks:
        return
    for task in _running_tasks.values():
        task.cancel()
    # Wait for all to drain; swallow cancellation errors.
    await asyncio.gather(*_running_tasks.values(), return_exceptions=True)
    _running_tasks.clear()
    _LOGGER.info("Stopped all HA subscription supervisors")


async def reload_provider(provider_id: int) -> None:
    """Stop and restart the supervisor for a single provider id.

    Hook for the provider CRUD routes — Phase 1.5 will wire it in. For Phase
    1, configure-then-restart-backend is the supported flow.
    """
    task = _running_tasks.pop(provider_id, None)
    if task is not None:
        task.cancel()
        try:
            await task
        except (asyncio.CancelledError, Exception):  # noqa: BLE001
            pass

    engine = get_engine()
    async with AsyncSession(engine) as session:
        prov = await session.get(ServiceProvider, provider_id)
        if prov is None or prov.type != "home_assistant":
            return

    new_task = asyncio.create_task(
        _run_provider(provider_id),
        name=f"ha-subscription-{provider_id}",
    )
    new_task.add_done_callback(_make_done_callback(provider_id))
    _running_tasks[provider_id] = new_task
