"""Shared dispatch helper for push-style providers.

Push-style providers (webhook receivers in ``api/webhooks.py`` and the
Home Assistant subscription manager in ``services/ha_subscription.py``)
share the same downstream pipeline: write an :class:`EventLog`, evaluate
quiet hours / event-type gates, defer if needed, otherwise hand off to the
:class:`NotificationDispatcher`.

This module extracts that pipeline so both callers can reuse it without
either side importing from the other (which would create a server/api ->
services -> api cycle).
"""

from __future__ import annotations

import logging
import time
from typing import Any, Awaitable, Callable

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

from notify_bridge_core.models.events import ServiceEvent
from notify_bridge_core.notifications.dispatcher import (
    NotificationDispatcher,
    TargetConfig,
)

from ..database.models import EventLog, NotificationTracker
from .deferred_dispatch import defer_event, is_deferrable
from .dispatch_helpers import (
    GateReason,
    apply_tracking_display_filters,
    evaluate_event_gate,
    get_app_timezone,
    load_link_data,
    resolve_provider_credential,
)

_LOGGER = logging.getLogger(__name__)


# Filter signature: ``(event, tracker.filters dict) -> bool``. Returning False
# drops the event for that tracker before any DB writes happen. Callers pass
# provider-specific logic (Gitea sender allowlist, HA entity glob, etc.).
FilterFn = Callable[[ServiceEvent, dict[str, Any]], bool]


# ---------------------------------------------------------------------------
# Tracker cache (per-provider, TTL-bounded)
# ---------------------------------------------------------------------------
#
# HA's chat-bus emits dozens of events per second; the per-event SELECT for
# enabled trackers becomes the bottleneck on busy installs. This 5-second
# TTL cache short-circuits the lookup when the same provider is hot. Cache
# entries are invalidated explicitly by tracker CRUD endpoints; the TTL is
# the safety net for missed invalidations.

_TRACKER_CACHE_TTL_SECONDS = 5.0
_trackers_cache: dict[int, tuple[float, list[NotificationTracker]]] = {}


def invalidate_tracker_cache(provider_id: int | None = None) -> None:
    """Drop cached trackers for one provider (or all if ``provider_id`` is None).

    Call from tracker create / update / delete endpoints so the next
    inbound event sees the change without waiting out the TTL.
    """
    if provider_id is None:
        _trackers_cache.clear()
    else:
        _trackers_cache.pop(provider_id, None)


async def _load_trackers_cached(
    session: AsyncSession, provider_id: int,
) -> list[NotificationTracker]:
    """Return enabled trackers for ``provider_id``, with a short TTL cache.

    Caches the ``NotificationTracker`` rows themselves — NOT the per-tracker
    ``NotificationTrackerTarget`` link rows. ``load_link_data`` always re-reads
    links from the DB on every dispatch, so adding/removing/toggling a link
    does NOT require invalidating this cache. Only call ``invalidate_tracker_cache``
    when a tracker row is created/updated/deleted.
    """
    now = time.monotonic()
    cached = _trackers_cache.get(provider_id)
    if cached is not None:
        ts, trackers = cached
        if now - ts < _TRACKER_CACHE_TTL_SECONDS:
            return trackers
    result = await session.exec(
        select(NotificationTracker).where(
            NotificationTracker.provider_id == provider_id,
            NotificationTracker.enabled == True,  # noqa: E712
        )
    )
    trackers = list(result.all())
    # Detach cached instances so consumers don't accidentally use a stale
    # session — re-fetch by id when mutating.
    for t in trackers:
        try:
            session.expunge(t)
        except Exception:  # noqa: BLE001
            pass
    _trackers_cache[provider_id] = (now, trackers)
    return trackers


async def dispatch_provider_event(
    engine: Any,
    provider_id: int,
    provider_name: str,
    provider_config: dict[str, Any],
    event: ServiceEvent,
    detail_keys: tuple[str, ...],
    filter_fn: FilterFn,
) -> int:
    """Load matching trackers, log, gate, defer, and dispatch one event.

    Parameters
    ----------
    engine:
        SQLAlchemy async engine.
    provider_id:
        ID of the :class:`ServiceProvider` the event came from.
    provider_name:
        Human-readable name (for logging only).
    provider_config:
        ``ServiceProvider.config`` dict; flowed into :class:`TargetConfig`.
    event:
        Parsed :class:`ServiceEvent` to dispatch.
    detail_keys:
        Keys from ``event.extra`` to copy into ``EventLog.details``.
    filter_fn:
        Per-event tracker-level filter. Returning False drops the event for
        that tracker before any DB writes.

    Returns
    -------
    int
        Number of successfully dispatched notifications across all trackers.
    """
    dispatched = 0
    # Drain-scheduling is best-effort: a scheduling failure must not roll
    # back the persisted defer rows (startup catch-up re-establishes them).
    defers_to_schedule: set[Any] = set()
    # Build the dispatcher once per inbound event — its only state is the
    # shared aiohttp session and Telegram caches, both of which are reused
    # across all trackers. Re-creating per-tracker meant a fresh dispatcher
    # for every notification, paying the construction cost on every HA
    # state_changed (ha provider can fire dozens per second).
    from .http_session import get_http_session
    from .watcher import _get_telegram_caches
    url_cache, asset_cache = await _get_telegram_caches()
    dispatcher = NotificationDispatcher(
        url_cache=url_cache,
        asset_cache=asset_cache,
        session=await get_http_session(),
    )
    async with AsyncSession(engine) as session:
        # App timezone is identical across trackers in one inbound event;
        # pull it once.
        app_tz = await get_app_timezone(session)

        trackers = await _load_trackers_cached(session, provider_id)

        for tracker in trackers:
            filters = tracker.filters or {}
            if not filter_fn(event, filters):
                _LOGGER.debug(
                    "Event filtered out for tracker %d (%s)", tracker.id, tracker.name
                )
                continue

            link_data = await load_link_data(session, tracker.id)
            if not link_data:
                continue

            extra_details = {k: v for k, v in event.extra.items() if k in detail_keys}
            event_log_row = EventLog(
                user_id=tracker.user_id,
                tracker_id=tracker.id,
                tracker_name=tracker.name,
                provider_id=provider_id,
                provider_name=provider_name,
                event_type=event.event_type.value,
                collection_id=event.collection_id,
                collection_name=event.collection_name,
                assets_count=0,
                details={
                    "provider_type": event.provider_type.value,
                    **extra_details,
                },
            )
            session.add(event_log_row)
            await session.flush()
            event_log_id = event_log_row.id

            # Dedupe defers by parent link_id: broadcast links emit one
            # link_data entry per child, sharing the same parent id — the
            # deferred row is one-per-link, so we call defer_event only
            # once per distinct id (earliest fire_at wins on ties).
            groups: dict[int, tuple[Any, list[TargetConfig]]] = {}
            defers_for_event: dict[int, Any] = {}
            for ld in link_data:
                tc = ld["tracking_config"]
                if tc is not None:
                    outcome = evaluate_event_gate(event, tc, app_tz)
                    if outcome.reason is GateReason.QUIET_HOURS:
                        if (
                            is_deferrable(event.event_type.value)
                            and outcome.quiet_hours_end_at is not None
                        ):
                            link_id = ld.get("link_id")
                            if link_id is not None:
                                prior = defers_for_event.get(link_id)
                                if prior is None or outcome.quiet_hours_end_at < prior:
                                    defers_for_event[link_id] = outcome.quiet_hours_end_at
                        else:
                            # Non-deferrable event hit quiet hours — stamp
                            # the event_log so the dashboard surfaces *why*
                            # the notification never went out.
                            details = dict(event_log_row.details or {})
                            if not details.get("dispatch_status"):
                                details["dispatch_status"] = (
                                    "dropped_quiet_hours_nondeferrable"
                                )
                                event_log_row.details = details
                                session.add(event_log_row)
                        continue
                    if outcome.reason is GateReason.EVENT_TYPE_DISABLED:
                        continue

                tmpl = ld["template_config"]
                target_cfg = TargetConfig(
                    type=ld["target_type"],
                    config=ld["target_config"],
                    template_slots=ld["template_slots"],
                    date_format=tmpl.date_format if tmpl else "%d.%m.%Y, %H:%M UTC",
                    date_only_format=(
                        tmpl.date_only_format
                        if tmpl and tmpl.date_only_format
                        else "%d.%m.%Y"
                    ),
                    provider_api_key=resolve_provider_credential(provider_config),
                    provider_internal_url=provider_config.get("url", ""),
                    provider_external_url=provider_config.get("url", ""),
                    receivers=ld["receivers"],
                )
                key = id(tc) if tc is not None else 0
                if key not in groups:
                    groups[key] = (tc, [])
                # Thread per-target metadata alongside the TargetConfig so the
                # bridge_self failure counters can attribute results to a
                # specific target_id after dispatch.
                groups[key][1].append((target_cfg, ld.get("target_id"), ld.get("target_name", "")))

            # Persist defers + stamp event_log dispatch_status in the same
            # session that holds the EventLog row, so the "deferred" badge
            # only appears if the underlying queue rows actually exist.
            if defers_for_event:
                earliest = min(defers_for_event.values())
                for link_id, fire_at in defers_for_event.items():
                    await defer_event(
                        session,
                        event=event,
                        user_id=tracker.user_id,
                        tracker_id=tracker.id,
                        link_id=link_id,
                        event_log_id=event_log_id,
                        fire_at=fire_at,
                    )
                details = dict(event_log_row.details or {})
                if not details.get("dispatch_status"):
                    details["dispatch_status"] = "deferred"
                    details["deferred_until"] = earliest.isoformat()
                    event_log_row.details = details
                    session.add(event_log_row)
                defers_to_schedule.update(defers_for_event.values())

            # Dispatch to targets. Isolate dispatcher exceptions per group so
            # a failed remote call doesn't bubble out, abort the surrounding
            # transaction, and roll back the just-written defers / event_log.
            from .bridge_self import (
                maybe_emit_target_failure,
                record_target_success,
            )

            # Skip target-failure tracking when we're already dispatching a
            # bridge_self event — otherwise a failing alert target would
            # endlessly re-emit alerts about itself.
            track_target_failures = (
                event.provider_type.value != "bridge_self"
            )

            for tc, target_entries in groups.values():
                if not target_entries:
                    continue
                shaped_event = apply_tracking_display_filters(event, tc)
                if shaped_event is None:
                    continue
                target_configs = [entry[0] for entry in target_entries]
                try:
                    results = await dispatcher.dispatch(shaped_event, target_configs)
                except Exception as err:  # noqa: BLE001
                    _LOGGER.exception(
                        "Dispatcher raised for tracker %d: %s", tracker.id, err,
                    )
                    continue
                for entry, r in zip(target_entries, results):
                    _, target_id, target_name = entry
                    if r.get("success"):
                        dispatched += 1
                        if track_target_failures and target_id is not None:
                            record_target_success(int(target_id))
                    else:
                        _LOGGER.error(
                            "Notification failed for tracker %d: %s",
                            tracker.id, r.get("error", "unknown"),
                        )
                        if track_target_failures and target_id is not None:
                            try:
                                await maybe_emit_target_failure(
                                    target_id=int(target_id),
                                    target_name=target_name or "",
                                    target_type=entry[0].type,
                                    error=str(r.get("error") or ""),
                                )
                            except Exception:  # noqa: BLE001
                                _LOGGER.exception(
                                    "bridge_self target-failure emission failed",
                                )

        await session.commit()

    # Schedule drain jobs OUTSIDE the DB session so an APScheduler hiccup
    # can't roll back the persisted defer rows.
    if defers_to_schedule:
        from .scheduler import schedule_deferred_drain
        for fire_at in defers_to_schedule:
            try:
                schedule_deferred_drain(fire_at)
            except Exception:  # noqa: BLE001
                _LOGGER.exception(
                    "Failed to schedule deferred drain for %s", fire_at,
                )

    return dispatched
