"""Notification dispatcher — routes events to targets via templates."""

from __future__ import annotations

import asyncio
import contextlib
import logging
import uuid
from dataclasses import dataclass, field
from typing import Any, AsyncIterator, Awaitable, Callable, Final

import aiohttp

from notify_bridge_core.log_context import bind_log_context, dispatch_id_var
from notify_bridge_core.models.events import ServiceEvent
from notify_bridge_core.templates.context import build_template_context
from notify_bridge_core.templates.renderer import render_template

from .http_base import safe_headers
from .receiver import (
    DiscordReceiver,
    EmailReceiver,
    MatrixReceiver,
    NtfyReceiver,
    Receiver,
    SlackReceiver,
    TelegramReceiver,
    WebhookReceiver,
)
from .redact import redact_exc
from .ssrf import UnsafeURLError, avalidate_outbound_url
from .telegram.cache import TelegramFileCache
from .telegram.client import TelegramClient
from .telegram.media import (
    build_telegram_asset_entry,
    extract_asset_id_from_url,
    is_asset_cache_key,
    is_asset_id,
)
from .webhook.client import WebhookClient

_LOGGER = logging.getLogger(__name__)

DEFAULT_TEMPLATE: Final = '{{ event_type }}: "{{ collection_name }}"'

_HTTP_TIMEOUT: Final = aiohttp.ClientTimeout(total=30, connect=10)

# Cap on how many asset downloads run concurrently inside
# ``_preload_asset_data``. Peak memory during a send is bounded to roughly
# ``_PRELOAD_CONCURRENCY * max_asset_size`` instead of ``max_media_to_send *
# max_asset_size``.
_PRELOAD_CONCURRENCY: Final = 6

# Cap on how many targets the dispatcher fans out to at once. With dozens
# of targets and a single hung peer, unbounded ``gather`` can pin the
# dispatch task. The cap also protects against credential-reuse rate
# limits on shared providers.
_DISPATCH_CONCURRENCY: Final = 16

# Cap on parallel per-receiver sends within a single target.
_RECEIVER_CONCURRENCY: Final = 8

# Per-target soft timeout — at the top of the dispatch tree so a single
# misbehaving target can't hold the whole batch open. Individual provider
# clients carry their own per-request timeouts on top of this.
_TARGET_TIMEOUT_S: Final = 120.0


def _new_session() -> aiohttp.ClientSession:
    return aiohttp.ClientSession(timeout=_HTTP_TIMEOUT)


@dataclass
class TargetConfig:
    """Configuration for a notification target."""

    type: str  # "telegram", "webhook", "email", "discord", "slack", "ntfy", "matrix"
    config: dict[str, Any]
    template_slots: dict[str, dict[str, str]] | None = None
    locale: str = "en"
    date_format: str = "%d.%m.%Y, %H:%M UTC"
    date_only_format: str = "%d.%m.%Y"
    provider_api_key: str | None = None
    provider_internal_url: str | None = None
    provider_external_url: str | None = None
    receivers: list[Receiver] = field(default_factory=list)


_SendMethod = Callable[
    ["NotificationDispatcher", TargetConfig, str, ServiceEvent],
    Awaitable[dict[str, Any]],
]


class NotificationDispatcher:
    """Dispatches ServiceEvent notifications to configured targets."""

    def __init__(
        self,
        *,
        url_cache: TelegramFileCache | None = None,
        asset_cache: TelegramFileCache | None = None,
        session: aiohttp.ClientSession | None = None,
    ) -> None:
        self._url_cache = url_cache
        self._asset_cache = asset_cache
        # Optional shared session owned by the caller; when supplied we reuse
        # its connection pool instead of opening a fresh per-dispatch session.
        self._shared_session = session
        # Per-dispatch render cache, keyed by locale. Populated by
        # ``_send_to_target`` and consumed inside ``_message_for_receiver``
        # so a 100-receiver fan-out renders each unique locale once.
        # Initialized to empty so handlers called outside the normal
        # dispatch path (tests) still see a valid dict.
        self._render_cache: dict[str, str] = {}

    @contextlib.asynccontextmanager
    async def _session_ctx(self) -> AsyncIterator[aiohttp.ClientSession]:
        if self._shared_session is not None and not self._shared_session.closed:
            yield self._shared_session
            return
        async with _new_session() as session:
            yield session

    async def dispatch(
        self,
        event: ServiceEvent,
        targets: list[TargetConfig],
    ) -> list[dict[str, Any]]:
        """Send event notification to all targets.

        Returns one result per target. Per-target failures are isolated;
        a single bad target cannot poison the batch.
        """
        new_id = dispatch_id_var.get() or f"disp:{uuid.uuid4().hex[:12]}"

        with bind_log_context(dispatch_id=new_id):
            _LOGGER.info(
                "Dispatching event %s (collection=%r) to %d target(s)",
                event.event_type.value if hasattr(event.event_type, "value") else event.event_type,
                getattr(event, "collection_name", None), len(targets),
            )

            sem = asyncio.Semaphore(_DISPATCH_CONCURRENCY)

            async def run_one(t: TargetConfig) -> dict[str, Any]:
                async with sem:
                    try:
                        return await asyncio.wait_for(
                            self._send_to_target(event, t),
                            timeout=_TARGET_TIMEOUT_S,
                        )
                    except asyncio.TimeoutError:
                        return {
                            "success": False,
                            "error": f"Target dispatch timed out after {_TARGET_TIMEOUT_S}s",
                        }

            raw_results = await asyncio.gather(
                *[run_one(t) for t in targets],
                return_exceptions=True,
            )
            results: list[dict[str, Any]] = []
            failures = 0
            for target, raw in zip(targets, raw_results):
                if isinstance(raw, Exception):
                    failures += 1
                    _LOGGER.error(
                        "Dispatch to target type=%s failed: %s",
                        target.type, redact_exc(raw),
                    )
                    results.append({"success": False, "error": redact_exc(raw)})
                else:
                    if isinstance(raw, dict) and not raw.get("success"):
                        failures += 1
                    results.append(raw)
            _LOGGER.info(
                "Dispatch finished: %d target(s), %d failure(s)",
                len(targets), failures,
            )
            return results

    def _resolve_template(
        self, event: ServiceEvent, target: TargetConfig, locale: str,
    ) -> str:
        template_str = DEFAULT_TEMPLATE
        if target.template_slots:
            locale_map = target.template_slots.get(event.event_type.value)
            if locale_map:
                template_str = locale_map.get(locale) or locale_map.get("en") or template_str
        return template_str

    def _render_message(
        self, event: ServiceEvent, target: TargetConfig, locale: str,
    ) -> str:
        template_str = self._resolve_template(event, target, locale)
        ctx = build_template_context(
            event, target_type=target.type,
            date_format=target.date_format,
            date_only_format=target.date_only_format,
        )
        return render_template(template_str, ctx)

    def _message_for_receiver(
        self, receiver: Receiver, default_message: str,
        event: ServiceEvent, target: TargetConfig,
        cache: dict[str, str] | None = None,
    ) -> str:
        """Render message respecting receiver locale, with optional cache.

        The ``cache`` dict (typically created in ``_send_to_target`` and
        threaded through the per-channel ``_send_*`` handlers) memoizes
        per-locale renders so a 100-receiver fan-out with two locales
        renders twice instead of one hundred times.
        """
        loc = receiver.locale or target.locale
        if loc == target.locale:
            return default_message
        if cache is not None:
            cached = cache.get(loc)
            if cached is not None:
                return cached
            rendered = self._render_message(event, target, loc)
            cache[loc] = rendered
            return rendered
        return self._render_message(event, target, loc)

    async def _send_to_target(
        self, event: ServiceEvent, target: TargetConfig
    ) -> dict[str, Any]:
        """Dispatch to a single target via the registered handler.

        Builds a per-locale render cache once and threads it through the
        send handler. The cache is keyed by receiver locale; the default
        locale's render lives in ``default_message`` and is short-circuited
        before any cache lookup.
        """
        default_message = self._render_message(event, target, target.locale)
        send_method = _PROVIDER_HANDLERS.get(target.type)
        if send_method is None:
            return {"success": False, "error": f"Unknown target type: {target.type}"}
        # Stash the cache on the dispatcher instance for the duration of
        # this dispatch — handlers pick it up via _message_for_receiver.
        # Avoids changing every _send_* signature.
        self._render_cache: dict[str, str] = {}
        try:
            return await send_method(self, target, default_message, event)
        finally:
            self._render_cache = {}

    # ------------------------------------------------------------------
    # Asset preload (Telegram-specific)
    # ------------------------------------------------------------------

    async def _preload_asset_data(
        self,
        assets: list[dict[str, Any]],
        media_assets: list[Any],
        session: aiohttp.ClientSession,
        max_size: int | None,
    ) -> None:
        """Download each non-cached asset's bytes once, with SSRF guard."""
        if not assets:
            return

        sem = asyncio.Semaphore(_PRELOAD_CONCURRENCY)

        async def fetch(entry: dict[str, Any], media: Any) -> None:
            cache, key = self._cache_for_entry(entry)
            if cache and key:
                cached = cache.get(key)
                if cached and cached.get("file_id"):
                    stored_size = cached.get("size")
                    if stored_size is not None:
                        media.extra["playback_size"] = stored_size
                    return

            url = entry["url"]
            headers = entry.get("headers") or {}
            try:
                # Defense-in-depth: validate even though TelegramClient
                # also validates. The dispatcher is what triggers the
                # download, so the guard belongs here too.
                await avalidate_outbound_url(url)
            except UnsafeURLError as err:
                _LOGGER.warning(
                    "Asset preload skipped: unsafe URL (%s)", redact_exc(err),
                )
                return
            async with sem:
                try:
                    async with session.get(url, headers=headers) as resp:
                        if resp.status != 200:
                            return
                        data = await resp.read()
                except (aiohttp.ClientError, asyncio.TimeoutError, OSError):
                    return
                if max_size is not None and len(data) > max_size:
                    return
                entry["data"] = data
                media.extra["playback_size"] = len(data)

        raw = await asyncio.gather(
            *(fetch(e, m) for e, m in zip(assets, media_assets)),
            return_exceptions=True,
        )
        for r in raw:
            if isinstance(r, Exception):
                _LOGGER.warning("Asset preload raised: %s", redact_exc(r))

    def _cache_for_entry(
        self, entry: dict[str, Any],
    ) -> tuple[TelegramFileCache | None, str | None]:
        cache_key = entry.get("cache_key")
        if cache_key:
            cache = self._asset_cache if is_asset_cache_key(cache_key) else self._url_cache
            return cache, cache_key
        url = entry.get("url")
        if url:
            if is_asset_id(url):
                return self._asset_cache, url
            extracted = extract_asset_id_from_url(url)
            if extracted:
                return self._asset_cache, extracted
            return self._url_cache, url
        return None, None

    # ------------------------------------------------------------------
    # Per-provider handlers
    # ------------------------------------------------------------------

    async def _send_telegram(
        self, target: TargetConfig, default_message: str, event: ServiceEvent
    ) -> dict[str, Any]:
        bot_token = target.config.get("bot_token")
        disable_preview = target.config.get("disable_url_preview", False)
        chat_action = target.config.get("chat_action", "typing")
        max_media = target.config.get("max_media_to_send", 50)
        max_group = target.config.get("max_media_per_group", 10)
        chunk_delay = target.config.get("media_delay", 500)
        max_size_mb = target.config.get("max_asset_size")
        max_size_bytes = max_size_mb * 1024 * 1024 if max_size_mb else None
        send_large_as_docs = target.config.get("send_large_photos_as_documents", False)

        if not bot_token:
            return {"success": False, "error": "Missing bot_token"}
        if not target.receivers:
            return {"success": False, "error": "No receivers configured"}

        internal_url = (target.provider_internal_url or "").rstrip("/")
        external_url = (target.provider_external_url or "").rstrip("/")
        assets: list[dict[str, Any]] = []
        media_assets: list[Any] = []
        for asset in event.added_assets[:max_media]:
            url = asset.preview_url or asset.thumbnail_url or asset.full_url
            if not url:
                continue
            asset_entry = build_telegram_asset_entry(
                url=url,
                media_type=asset.type.value,
                api_key=target.provider_api_key,
                internal_url=internal_url,
                external_url=external_url,
                cache_key=asset.extra.get("cache_key"),
            )
            if asset_entry is not None:
                assets.append(asset_entry)
                media_assets.append(asset)

        async with self._session_ctx() as session:
            await self._preload_asset_data(assets, media_assets, session, max_size_bytes)

            thumbhash_map = {
                asset.id: asset.extra.get("thumbhash")
                for asset in event.added_assets
                if asset.extra.get("thumbhash")
            }
            thumbhash_resolver = thumbhash_map.get if thumbhash_map else None

            client = TelegramClient(
                session, bot_token,
                url_cache=self._url_cache,
                asset_cache=self._asset_cache,
                thumbhash_resolver=thumbhash_resolver,
            )

            async def send_one(receiver: Receiver) -> dict[str, Any]:
                if not isinstance(receiver, TelegramReceiver) or not receiver.chat_id:
                    return {"success": False, "error": "Invalid telegram receiver"}
                message = self._message_for_receiver(receiver, default_message, event, target, cache=self._render_cache)
                text_result = await client.send_message(
                    chat_id=receiver.chat_id,
                    text=message,
                    disable_web_page_preview=bool(disable_preview),
                )
                if not text_result.get("success"):
                    _LOGGER.warning(
                        "Failed to send to chat %s: %s",
                        receiver.chat_id, text_result.get("error"),
                    )
                    return text_result

                if assets:
                    media_result = await client.send_notification(
                        chat_id=receiver.chat_id,
                        assets=assets,
                        reply_to_message_id=text_result.get("message_id"),
                        max_group_size=max_group,
                        chunk_delay=chunk_delay,
                        max_asset_data_size=max_size_bytes,
                        send_large_photos_as_documents=send_large_as_docs,
                        chat_action=chat_action or None,
                    )
                    if not media_result.get("success"):
                        _LOGGER.warning(
                            "Text sent OK but media failed for chat %s: %s",
                            receiver.chat_id, media_result.get("error"),
                        )
                        # Preserve both outcomes — text succeeded, media
                        # didn't. Operators losing media-failure detail
                        # in the result dict made root-cause analysis
                        # impossible.
                        return {
                            "success": True,
                            "message_id": text_result.get("message_id"),
                            "media_error": media_result.get("error"),
                            "media_failed_at_chunk": media_result.get("failed_at_chunk"),
                        }
                return text_result

            results = await self._fan_out(target.receivers, send_one)

        return self._aggregate_results(results)

    async def _send_webhook(
        self, target: TargetConfig, default_message: str, event: ServiceEvent
    ) -> dict[str, Any]:
        if not target.receivers:
            return {"success": False, "error": "No receivers configured"}

        async with self._session_ctx() as session:
            async def send_one(receiver: Receiver) -> dict[str, Any]:
                if not isinstance(receiver, WebhookReceiver) or not receiver.url:
                    return {"success": False, "error": "Invalid webhook receiver"}
                message = self._message_for_receiver(receiver, default_message, event, target, cache=self._render_cache)
                payload = {
                    "message": message,
                    "event_type": event.event_type.value,
                    "provider_type": event.provider_type.value,
                    "collection_name": event.collection_name,
                    "collection_id": event.collection_id,
                    "timestamp": event.timestamp.isoformat(),
                }
                client = WebhookClient(session, receiver.url, safe_headers(receiver.headers))
                return await client.send(payload)

            results = await self._fan_out(target.receivers, send_one)

        return self._aggregate_results(results)

    async def _send_email(
        self, target: TargetConfig, default_message: str, event: ServiceEvent
    ) -> dict[str, Any]:
        from .email.client import EmailClient, SmtpConfig

        smtp_cfg = target.config.get("smtp", {})
        if not smtp_cfg.get("host"):
            return {"success": False, "error": "SMTP not configured"}

        email_client = EmailClient(SmtpConfig(
            host=smtp_cfg["host"],
            port=int(smtp_cfg.get("port", 587)),
            username=smtp_cfg.get("username", ""),
            password=smtp_cfg.get("password", ""),
            from_address=smtp_cfg.get("from_address", ""),
            from_name=smtp_cfg.get("from_name", "Notify Bridge"),
            use_tls=smtp_cfg.get("use_tls", True),
            tls_mode=smtp_cfg.get("tls_mode", "auto"),
        ))

        if not target.receivers:
            return {"success": False, "error": "No receivers configured"}
        subject = f"[Notify Bridge] {event.event_type.value}: {event.collection_name}"

        async def send_one(receiver: Receiver) -> dict[str, Any]:
            if not isinstance(receiver, EmailReceiver) or not receiver.email:
                return {"success": False, "error": "Invalid email receiver"}
            message = self._message_for_receiver(receiver, default_message, event, target, cache=self._render_cache)
            # body_html=None lets EmailClient build a safely-escaped HTML
            # alternative from body_text instead of trusting user content.
            return await email_client.send(
                to_email=receiver.email,
                subject=subject,
                body_text=message,
                body_html=None,
                to_name=receiver.name,
            )

        results = await self._fan_out(target.receivers, send_one)
        return self._aggregate_results(results)

    async def _send_discord(
        self, target: TargetConfig, default_message: str, event: ServiceEvent
    ) -> dict[str, Any]:
        from .discord.client import DiscordClient

        if not target.receivers:
            return {"success": False, "error": "No receivers configured"}
        username = target.config.get("username")

        async with self._session_ctx() as session:
            client = DiscordClient(session)

            async def send_one(receiver: Receiver) -> dict[str, Any]:
                if not isinstance(receiver, DiscordReceiver) or not receiver.webhook_url:
                    return {"success": False, "error": "Invalid discord receiver"}
                message = self._message_for_receiver(receiver, default_message, event, target, cache=self._render_cache)
                return await client.send(receiver.webhook_url, message, username=username)

            results = await self._fan_out(target.receivers, send_one)

        return self._aggregate_results(results)

    async def _send_slack(
        self, target: TargetConfig, default_message: str, event: ServiceEvent
    ) -> dict[str, Any]:
        from .slack.client import SlackClient

        if not target.receivers:
            return {"success": False, "error": "No receivers configured"}
        username = target.config.get("username")

        async with self._session_ctx() as session:
            client = SlackClient(session)

            async def send_one(receiver: Receiver) -> dict[str, Any]:
                if not isinstance(receiver, SlackReceiver) or not receiver.webhook_url:
                    return {"success": False, "error": "Invalid slack receiver"}
                message = self._message_for_receiver(receiver, default_message, event, target, cache=self._render_cache)
                return await client.send(receiver.webhook_url, message, username=username)

            results = await self._fan_out(target.receivers, send_one)

        return self._aggregate_results(results)

    async def _send_ntfy(
        self, target: TargetConfig, default_message: str, event: ServiceEvent
    ) -> dict[str, Any]:
        from .ntfy.client import NtfyClient

        server_url = target.config.get("server_url", "https://ntfy.sh")
        auth_token = target.config.get("auth_token")
        if not target.receivers:
            return {"success": False, "error": "No receivers configured"}
        try:
            await avalidate_outbound_url(server_url)
        except UnsafeURLError as err:
            return {"success": False, "error": f"Unsafe ntfy server_url: {redact_exc(err)}"}

        title = f"{event.event_type.value}: {event.collection_name}"

        async with self._session_ctx() as session:
            client = NtfyClient(session)

            async def send_one(receiver: Receiver) -> dict[str, Any]:
                if not isinstance(receiver, NtfyReceiver) or not receiver.topic:
                    return {"success": False, "error": "Invalid ntfy receiver"}
                message = self._message_for_receiver(receiver, default_message, event, target, cache=self._render_cache)
                return await client.send(
                    server_url, receiver.topic, message,
                    title=title, priority=receiver.priority, auth_token=auth_token,
                )

            results = await self._fan_out(target.receivers, send_one)

        return self._aggregate_results(results)

    async def _send_matrix(
        self, target: TargetConfig, default_message: str, event: ServiceEvent
    ) -> dict[str, Any]:
        from .matrix.client import MatrixClient

        homeserver = target.config.get("homeserver_url")
        access_token = target.config.get("access_token")
        if not homeserver or not access_token:
            return {"success": False, "error": "Missing Matrix homeserver_url or access_token"}
        try:
            await avalidate_outbound_url(homeserver)
        except UnsafeURLError as err:
            return {"success": False, "error": f"Unsafe matrix homeserver_url: {redact_exc(err)}"}

        if not target.receivers:
            return {"success": False, "error": "No receivers configured"}

        async with self._session_ctx() as session:
            client = MatrixClient(session, homeserver, access_token)

            async def send_one(receiver: Receiver) -> dict[str, Any]:
                if not isinstance(receiver, MatrixReceiver) or not receiver.room_id:
                    return {"success": False, "error": "Invalid matrix receiver"}
                message = self._message_for_receiver(receiver, default_message, event, target, cache=self._render_cache)
                # body_html is the same plain text — Matrix accepts the
                # raw message as both ``body`` and ``formatted_body``.
                # If templates emit HTML in the future, generate a
                # separate HTML body upstream rather than aliasing here.
                return await client.send_message(
                    receiver.room_id, message, html_message=None,
                )

            results = await self._fan_out(target.receivers, send_one)

        return self._aggregate_results(results)

    # ------------------------------------------------------------------
    # Aggregation
    # ------------------------------------------------------------------

    @staticmethod
    async def _fan_out(
        receivers: list[Receiver],
        send_one: Callable[[Receiver], Awaitable[dict[str, Any]]],
    ) -> list[dict[str, Any]]:
        """Run ``send_one`` per receiver with bounded concurrency.

        Per-receiver exceptions are converted to failure dicts so a single
        bad receiver can't cancel its peers.
        """
        sem = asyncio.Semaphore(_RECEIVER_CONCURRENCY)

        async def guarded(receiver: Receiver) -> dict[str, Any]:
            async with sem:
                try:
                    return await send_one(receiver)
                except Exception as exc:  # noqa: BLE001
                    _LOGGER.error("Receiver send raised: %s", redact_exc(exc))
                    return {"success": False, "error": redact_exc(exc)}

        return await asyncio.gather(*(guarded(r) for r in receivers))

    @staticmethod
    def _aggregate_results(results: list[dict[str, Any]]) -> dict[str, Any]:
        """Aggregate per-receiver results into a single target-level result.

        Preserves the per-receiver detail under ``receivers`` so a caller
        can see exactly which receivers failed, instead of getting only
        the first error.
        """
        if not results:
            return {"success": False, "error": "No receivers configured"}

        successes = sum(1 for r in results if r.get("success"))
        failures = len(results) - successes
        out: dict[str, Any] = {
            "success": successes > 0,
            "receivers": len(results),
            "successes": successes,
            "failures": failures,
            "results": results,
        }
        if failures:
            out["errors"] = [
                r.get("error") for r in results if not r.get("success")
            ]
            if successes == 0:
                # Surface the first error at the top level for back-compat
                # with callers that only check ``error``.
                out["error"] = results[0].get("error", "All receivers failed")
        return out


# ----------------------------------------------------------------------
# Provider registry — replaces the if/elif chain so adding a provider
# means just registering it here, not editing dispatch logic.
# ----------------------------------------------------------------------

_PROVIDER_HANDLERS: dict[str, _SendMethod] = {
    "telegram": NotificationDispatcher._send_telegram,
    "webhook": NotificationDispatcher._send_webhook,
    "email": NotificationDispatcher._send_email,
    "discord": NotificationDispatcher._send_discord,
    "slack": NotificationDispatcher._send_slack,
    "ntfy": NotificationDispatcher._send_ntfy,
    "matrix": NotificationDispatcher._send_matrix,
}


def register_provider(name: str, handler: _SendMethod) -> None:
    """Register a new dispatcher provider at runtime.

    Allows out-of-tree providers to extend the dispatcher without
    forking. The handler must follow the
    ``async (dispatcher, target, default_message, event) -> dict`` shape.
    """
    _PROVIDER_HANDLERS[name] = handler
