"""ShieldConnectionManager — entry-scoped owner of the ADB connection (Phase 1).

The ADB socket, its watchdog and all ADB-derived state live here instead of on
the media-player entity (plan blockers B1/B2/B5):

- one persistent socket; a watchdog task reconnects with exponential backoff
  (5 s doubling to a 300 s cap, ±25 % jitter) when it drops, and pulls the
  heartbeat data (wake + media sessions + running apps) while connected;
- ADB-derived values live in a
  :class:`StateStore` with per-field timestamps and TTL expiry — stale values
  read as expired (None/empty) instead of clinging on;
- entities are thin callback-views: they ``register_listener()`` and write HA
  state when the store changes (the androidtv core ``entity.py`` pattern);
- ADB is optional: when no poller factory is provided the manager is *passive*
  and the entry still sets up — a failed ADB connect is a background retry,
  never a ``ConfigEntryNotReady``.

This module deliberately stays Home-Assistant-free at import time so it is
unit-testable with plain pytest; every hass capability is injected (the entry
builds the concrete poller factory). Task creation additionally works without
hass (plain ``asyncio.create_task``) for the unit tests.
"""

import asyncio
import contextlib
import logging
import random
import time
from collections.abc import Callable
from typing import Any

from .adb_session import AdbMediaSessionPoller
from .const import (
    ADB_POLL_INTERVAL,
    ADB_RECONNECT_BACKOFF_CAP,
    ADB_RECONNECT_BACKOFF_START,
    ADB_RECONNECT_JITTER,
    STATE_STORE_TTL,
)
from .ha_adb_common import MediaSessionInfo, parse_audio_playback

_LOGGER = logging.getLogger(__name__)

# StateStore field names
FIELD_AWAKE = "awake"
FIELD_SESSIONS = "sessions"
FIELD_AUDIO_PLAYBACK = "audio_playback"
FIELD_RUNNING_APPS = "running_apps"

ALL_FIELDS = (FIELD_AWAKE, FIELD_SESSIONS, FIELD_AUDIO_PLAYBACK, FIELD_RUNNING_APPS)


class StateStore:
    """Timed per-field state cache shared by all entities of one entry.

    Reading an expired (or missing) field returns the caller-supplied
    default, so a stale "playing" session can never outlive the TTL (plan
    blocker B5). Values are written only via :meth:`set`, which re-stamps the
    field's freshness timestamp.
    """

    def __init__(self, ttl: float = STATE_STORE_TTL, clock: Callable[[], float] = time.monotonic):
        self._ttl = ttl
        self._clock = clock
        self._values: dict[str, tuple[float, Any]] = {}

    def set(self, name: str, value: Any) -> None:
        """Store a value, re-stamping its freshness timestamp."""
        self._values[name] = (self._clock(), value)

    def get(self, name: str, default: Any = None) -> Any:
        """Return the field's value, or ``default`` when it is missing/expired."""
        stamped = self._values.get(name)
        if stamped is None:
            return default
        at, value = stamped
        if self._clock() - at > self._ttl:
            return default
        return value

    def clear(self, *names: str) -> None:
        """Drop fields immediately so they read as expired on the next get."""
        for name in names:
            self._values.pop(name, None)

    @property
    def ttl(self) -> float:
        return self._ttl


class ShieldConnectionManager:
    """Owner of one entry's persistent ADB connection and state.

    :param poller_factory: builds the ADB poller when the watchdog starts;
        skipped entirely for entries without ADB (passive manager) or tests
        that inject a ready-to-use ``poller`` directly.
    """

    def __init__(
        self,
        hass=None,
        entry=None,
        poller_factory: Callable[[], AdbMediaSessionPoller] | None = None,
        poller: AdbMediaSessionPoller | None = None,
        store: StateStore | None = None,
    ):
        self.hass = hass
        self.entry = entry
        self.store = store if store is not None else StateStore(ttl=STATE_STORE_TTL)
        self._poller_factory = poller_factory
        self._poller = poller
        self._listeners: list[Callable[[], None]] = []
        self._tasks: list[asyncio.Task] = []
        self._stopping = False
        self._failures = 0
        self._down_since = False  # listeners already told the connection is down

    # ── One-shot connection test (pairing UX) ──

    @staticmethod
    async def test_connect(host: str, port: int, key_path: str) -> bool:
        """Single-shot ADB connect used by the config-flow pairing step.

        Throws away the poller immediately — the config flow runs before any
        entry exists, so it must never touch an entry's persistent socket.
        """
        poller = AdbMediaSessionPoller(host, port, key_path)
        try:
            return await poller.connect()
        finally:
            await poller.disconnect()

    # ── Introspection (thin views for entities) ──

    @property
    def available(self) -> bool:
        """True when ADB is active for this entry (a poller exists)."""
        return self._poller is not None

    @property
    def connected(self) -> bool:
        return bool(self._poller and self._poller.connected)

    @property
    def awake(self) -> bool | None:
        """Device wake state (exact mWakefulness parse), None = unknown."""
        return self.store.get(FIELD_AWAKE)

    @property
    def active_sessions(self) -> list[MediaSessionInfo]:
        """Stored media sessions, sorted playing > paused > other."""
        return self.store.get(FIELD_SESSIONS, [])

    @property
    def audio_playback(self) -> list[str]:
        """Packages that recently played audio, ordered last-played-first."""
        return self.store.get(FIELD_AUDIO_PLAYBACK, [])

    @property
    def running_apps(self) -> list[dict]:
        """Visible recent-task apps, ordered by recency."""
        return self.store.get(FIELD_RUNNING_APPS, [])

    def best_session(
        self, foreground_package: str | None = None
    ) -> MediaSessionInfo | None:
        """Best media session: foreground app first, then playing > paused."""
        sessions = self.active_sessions
        if not sessions:
            return None
        if foreground_package:
            for session in sessions:
                if session.package == foreground_package:
                    return session
        return sessions[0]

    # ── Listener API (entities are thin callback-views) ──

    def register_listener(self, listener: Callable[[], None]) -> Callable[[], None]:
        """Register a refresh callback; returns an unsubscribe callable."""
        self._listeners.append(listener)

        def _unregister() -> None:
            with contextlib.suppress(ValueError):
                self._listeners.remove(listener)

        return _unregister

    def _notify(self) -> None:
        for listener in tuple(self._listeners):
            try:
                listener()
            except Exception:
                _LOGGER.exception("Shield state listener failed")

    # ── Delegation (commands over the managed socket) ──

    async def send_media_command(self, command: str) -> bool:
        """Send a media key command over the persistent socket."""
        if not self.available:
            return False
        return await self._poller.send_media_command(command)

    async def bring_to_foreground(self, package: str) -> bool:
        if not self.available:
            return False
        return await self._poller.bring_to_foreground(package)

    # ── Lifecycle ──

    async def async_start(self) -> None:
        """Start the persistent connection loop.

        Never validates the ADB connection here: setup must not fail because
        of ADB (it is optional), so all failure handling lives in the
        watchdog's background retry.
        """
        self._stopping = False
        self._tasks.append(self._create_task(self._watchdog(), "shield_adb_watchdog"))

    async def async_stop(self) -> None:
        """Cancel all manager tasks and close the socket.

        Called on entry unload and on EVENT_HOMEASSISTANT_STOP.
        """
        self._stopping = True
        tasks, self._tasks = self._tasks, []
        for task in tasks:
            task.cancel()
        self._listeners.clear()
        if self._poller is not None:
            with contextlib.suppress(Exception):
                await self._poller.disconnect()
            self._poller = None

    def _create_task(self, coro, name: str) -> asyncio.Task:
        if self.hass is not None:
            return self.hass.async_create_background_task(coro, name)
        return asyncio.create_task(coro, name=name)

    async def _watchdog(self) -> None:
        """Persistent loop: reconnect with backoff when down, refresh when up.

        A passive manager (no poller possible) exits immediately — the entry
        recreates it when options change (full reload).
        """
        try:
            if self._poller is None:
                if self._poller_factory is None:
                    return
                self._poller = self._poller_factory()
            if self._poller is None:
                return

            backoff = ADB_RECONNECT_BACKOFF_START
            while not self._stopping:
                delay = await self._watchdog_cycle(backoff)
                if self._stopping:
                    break
                if self.connected:
                    backoff = ADB_RECONNECT_BACKOFF_START
                else:
                    backoff = min(backoff * 2, ADB_RECONNECT_BACKOFF_CAP)
                await asyncio.sleep(delay)
        except asyncio.CancelledError:
            raise
        except Exception:
            _LOGGER.exception("Shield connection watchdog crashed")

    async def _watchdog_cycle(self, backoff: float) -> float:
        """Run one watchdog cycle; return seconds to sleep before the next.

        When the connection is up this is the heartbeat: wake check + (when
        awake) media sessions + running apps, all stored with timestamps.
        The current backoff sleeps after failures; success resets it."""

        try:
            if not self.connected and not await self._poller.connect():
                await self._note_failure(backoff)
                return self._jitter(backoff)
            if self._failures:
                self._failures = 0
                self._down_since = False
                _LOGGER.info("ADB connection recovered")
            await self._refresh()
            return ADB_POLL_INTERVAL
        except Exception as err:
            await self._note_failure(backoff)
            _LOGGER.debug("Watchdog cycle failed: %s", err)
            return self._jitter(backoff)

    async def _note_failure(self, backoff: float) -> None:
        """Bookkeep a failed cycle: log sparingly, drop stale store, notify."""
        self._failures += 1
        if self._failures == 1 or self._failures % 6 == 0:
            _LOGGER.warning(
                "ADB connection down (%s consecutive failures) — retrying in ~%.0fs",
                self._failures,
                backoff,
            )
        if not self._down_since:
            self._down_since = True
            # Reconnect decision (health_check's documented role): close the
            # socket so the next cycle rebuilds it from scratch instead of
            # trusting shell()'s auto-reconnect on a broken transport.
            try:
                await self._poller.disconnect()
            except Exception:
                _LOGGER.debug("disconnect during reconnect decision failed", exc_info=True)
            self.store.clear(*ALL_FIELDS)
            self._notify()

    @staticmethod
    def _jitter(delay: float) -> float:
        """Backoff with ±jitter so many clients do not reconnect in lockstep."""
        return max(
            1.0,
            delay * (1.0 + random.uniform(-ADB_RECONNECT_JITTER, ADB_RECONNECT_JITTER)),
        )

    async def _refresh(self) -> None:
        """Heartbeat: pull all ADB-derived state into the store (with stamps)."""
        poller = self._poller
        awake = await poller.is_awake()
        if awake is None:
            # Failed poll (connection dropped / device unreachable) — surface
            # as a cycle failure instead of clinging to the last values.
            raise ConnectionError("ADB wake poll failed")
        self.store.set(FIELD_AWAKE, awake)
        if awake:
            # Full dump, no truncation (B4) — low sessions must not be lost.
            await poller.poll_media_sessions(head_lines=None)
            self.store.set(FIELD_SESSIONS, list(poller.all_sessions))
            self.store.set(
                FIELD_AUDIO_PLAYBACK, parse_audio_playback(poller.last_dump or "")
            )
            self.store.set(FIELD_RUNNING_APPS, await poller.poll_running_apps())
        else:
            # Standby: playback data is not current — expire it.
            self.store.set(FIELD_SESSIONS, [])
            self.store.set(FIELD_AUDIO_PLAYBACK, [])
            self.store.set(FIELD_RUNNING_APPS, [])
        self._notify()