"""Telegram Bot API client for sending notifications with media."""

from __future__ import annotations

import asyncio
import json
import logging
import mimetypes
import re
from dataclasses import dataclass, field
from typing import Any, Callable, Final

import aiohttp
from aiohttp import FormData

from ..redact import redact, redact_exc
from ..ssrf import UnsafeURLError, avalidate_outbound_url
from .cache import TelegramFileCache
from .media import (
    TELEGRAM_API_BASE_URL,
    TELEGRAM_MAX_CAPTION_LENGTH,
    TELEGRAM_MAX_PHOTO_SIZE,
    TELEGRAM_MAX_TEXT_LENGTH,
    TELEGRAM_MAX_VIDEO_SIZE,
    asset_id_from_cache_key,
    check_photo_limits,
    extract_asset_id_from_url,
    is_asset_cache_key,
    is_asset_id,
    split_media_by_upload_size,
)

_LOGGER = logging.getLogger(__name__)

NotificationResult = dict[str, Any]

# Telegram 429 / flood-control retry settings.
_TG_429_MAX_ATTEMPTS: Final = 4
_TG_429_MAX_WAIT_S: Final = 60
_RETRY_AFTER_RE: Final = re.compile(r"retry after (\d+)", re.IGNORECASE)

# Media-group fan-out cap. With 10 large videos, unbounded gather can
# spike memory to ``N * max_asset_data_size`` and saturate the upstream
# provider. Three concurrent fetches still saturates a typical home LAN
# while keeping peak RAM bounded to ~3 * max_asset_data_size.
_MEDIA_FETCH_CONCURRENCY: Final = 3

# Telegram chat actions expire after ~5s, so we refresh slightly faster.
_KEEPALIVE_REFRESH_S: Final = 4

# Per-call timeouts. The session may already carry a default but we set
# explicit values here so a future caller without a session-default
# can't hang the dispatcher.
_API_TIMEOUT: Final = aiohttp.ClientTimeout(total=30, connect=10)
_UPLOAD_TIMEOUT: Final = aiohttp.ClientTimeout(total=120, connect=10)
_DOWNLOAD_TIMEOUT: Final = aiohttp.ClientTimeout(total=120, connect=10)


def _extract_retry_after(result: dict[str, Any]) -> int | None:
    """Return the retry_after seconds from a Telegram error response.

    Prefers the structured ``parameters.retry_after`` field; falls back to
    parsing the human-readable description (``"Too Many Requests: retry
    after N"``) which Telegram has been known to return without the
    structured field on some endpoints.
    """
    params = result.get("parameters") or {}
    ra = params.get("retry_after")
    if isinstance(ra, (int, float)) and ra > 0:
        return int(ra)
    desc = str(result.get("description", ""))
    m = _RETRY_AFTER_RE.search(desc)
    if m:
        try:
            return int(m.group(1))
        except ValueError:
            return None
    return None


def _is_rate_limited(status: int, result: dict[str, Any]) -> bool:
    return status == 429 or result.get("error_code") == 429


def _is_parse_error(status: int, result: dict[str, Any]) -> bool:
    """True iff the response signals a Bot-API parse-mode failure.

    Gates on the canonical 400 status AND a parse-related description so a
    5xx that incidentally mentions "parse" doesn't trigger a retry.
    """
    if status != 400 and result.get("error_code") != 400:
        return False
    desc = str(result.get("description", "")).lower()
    return "can't parse" in desc or "parse_mode" in desc or "parse entities" in desc


@dataclass(frozen=True)
class _MediaKind:
    """Describes one Telegram media kind (photo / video / document)."""
    api_method: str
    form_field: str
    cache_type: str
    default_filename: str
    default_content_type: str

    def file_id_from_result(self, result: dict[str, Any]) -> str | None:
        obj = result.get(self.form_field)
        if isinstance(obj, list) and obj:
            last = obj[-1]
            return last.get("file_id") if isinstance(last, dict) else None
        if isinstance(obj, dict):
            return obj.get("file_id")
        return None


_PHOTO_KIND: Final = _MediaKind("sendPhoto", "photo", "photo", "photo.jpg", "image/jpeg")
_VIDEO_KIND: Final = _MediaKind("sendVideo", "video", "video", "video.mp4", "video/mp4")
_DOCUMENT_KIND: Final = _MediaKind(
    "sendDocument", "document", "document", "file", "application/octet-stream",
)


@dataclass
class _KeepaliveHandle:
    """Carries a chat-action keepalive task and its stop event."""
    task: asyncio.Task | None = None
    stop_event: asyncio.Event = field(default_factory=asyncio.Event)


@dataclass
class _MediaItem:
    """One element of a sendMediaGroup chunk after fetch+filter.

    Pairs a media JSON entry with the optional cache-update tuple
    keyed by position. Bundling these together prevents the
    ``media_json`` and ``cache_info`` lists from drifting out of
    alignment under future edits.
    """
    media_json: dict[str, Any]
    cache_info: tuple[str, str, str | None, int] | None
    attachment: tuple[str, bytes, str, str] | None  # (name, data, filename, content_type)


def _truncate(text: str, limit: int, *, marker: str = "…") -> str:
    """Soft-truncate ``text`` to ``limit`` chars with a trailing marker."""
    if len(text) <= limit:
        return text
    cutoff = max(0, limit - len(marker))
    return text[:cutoff] + marker


class TelegramClient:
    """Async Telegram Bot API client for sending notifications with media."""

    def __init__(
        self,
        session: aiohttp.ClientSession,
        bot_token: str,
        *,
        url_cache: TelegramFileCache | None = None,
        asset_cache: TelegramFileCache | None = None,
        url_resolver: Callable[[str], str] | None = None,
        thumbhash_resolver: Callable[[str], str | None] | None = None,
    ) -> None:
        self._session = session
        self._token = bot_token
        self._url_cache = url_cache
        self._asset_cache = asset_cache
        self._url_resolver = url_resolver
        self._thumbhash_resolver = thumbhash_resolver

    # ------------------------------------------------------------------
    # URL / cache helpers
    # ------------------------------------------------------------------

    def _api_url(self, method: str) -> str:
        return f"{TELEGRAM_API_BASE_URL}{self._token}/{method}"

    def _resolve_url(self, url: str) -> str:
        if self._url_resolver:
            return self._url_resolver(url)
        return url

    def _get_cache_and_key(
        self, url: str | None, cache_key: str | None = None,
    ) -> tuple[TelegramFileCache | None, str | None, str | None]:
        if cache_key:
            if is_asset_cache_key(cache_key):
                bare_id = asset_id_from_cache_key(cache_key)
                thumbhash = (
                    self._thumbhash_resolver(bare_id)
                    if self._thumbhash_resolver else None
                )
                return self._asset_cache, cache_key, thumbhash
            return self._url_cache, cache_key, None
        if url:
            if is_asset_id(url):
                thumbhash = self._thumbhash_resolver(url) if self._thumbhash_resolver else None
                return self._asset_cache, url, thumbhash
            asset_cache_key = extract_asset_id_from_url(url)
            if asset_cache_key:
                bare_id = asset_id_from_cache_key(asset_cache_key)
                thumbhash = self._thumbhash_resolver(bare_id) if self._thumbhash_resolver else None
                return self._asset_cache, asset_cache_key, thumbhash
            return self._url_cache, url, None
        return None, None, None

    def _get_cache_for_key(self, key: str, is_asset: bool | None = None) -> TelegramFileCache | None:
        if is_asset is None:
            is_asset = is_asset_cache_key(key)
        return self._asset_cache if is_asset else self._url_cache

    # ------------------------------------------------------------------
    # Asset download with SSRF guard
    # ------------------------------------------------------------------

    async def _safe_get(
        self,
        url: str,
        headers: dict[str, str] | None,
    ) -> tuple[bytes | None, str | None]:
        """SSRF-guarded GET that returns ``(data, error)``.

        Validates the URL via ``avalidate_outbound_url`` before any HTTP
        traffic. Redirects are walked manually so each ``Location`` is
        re-validated — without this an attacker-controlled origin could
        302 to a private-IP target after the initial guard passed.
        Errors are returned (not raised) and stripped of any embedded
        secrets before they propagate to the operator-visible result
        dict.
        """
        max_redirects = 3
        current_url = url
        try:
            await avalidate_outbound_url(current_url)
        except UnsafeURLError as err:
            return None, f"Unsafe URL: {redact_exc(err)}"
        try:
            for _ in range(max_redirects + 1):
                async with self._session.get(
                    current_url,
                    headers=headers or {},
                    timeout=_DOWNLOAD_TIMEOUT,
                    allow_redirects=False,
                ) as resp:
                    if resp.status in (301, 302, 303, 307, 308):
                        loc = resp.headers.get("Location")
                        if not loc:
                            return None, f"HTTP {resp.status} without Location header"
                        # ``resp.url`` is a yarl.URL; ``.join`` resolves
                        # relative redirects (``/foo/bar``) against it.
                        from yarl import URL as _URL
                        try:
                            next_url = str(resp.url.join(_URL(loc)))
                        except (ValueError, TypeError):
                            return None, "Malformed redirect Location"
                        try:
                            await avalidate_outbound_url(next_url)
                        except UnsafeURLError as err:
                            return None, f"Unsafe redirect: {redact_exc(err)}"
                        current_url = next_url
                        continue
                    if resp.status != 200:
                        return None, f"HTTP {resp.status}"
                    return await resp.read(), None
            return None, f"Too many redirects (>{max_redirects})"
        except (aiohttp.ClientError, asyncio.TimeoutError, OSError) as err:
            return None, redact_exc(err)

    async def _fetch_bytes(
        self,
        url: str,
        headers: dict[str, str] | None,
        preloaded: bytes | None,
    ) -> tuple[bytes | None, str | None]:
        """Return ``(data, error_msg)``. Uses ``preloaded`` bytes if provided."""
        if preloaded is not None:
            return preloaded, None
        return await self._safe_get(self._resolve_url(url), headers)

    # ------------------------------------------------------------------
    # Cached / fresh single-media send paths
    # ------------------------------------------------------------------

    async def _send_from_cache(
        self,
        kind: _MediaKind,
        chat_id: str,
        file_id: str,
        caption: str | None,
        reply_to_message_id: int | None,
        parse_mode: str,
    ) -> NotificationResult | None:
        """POST a file_id reference. Return None on transient error so the
        caller can fall through to a fresh upload."""
        payload: dict[str, Any] = {
            "chat_id": chat_id,
            kind.form_field: file_id,
            "parse_mode": parse_mode,
        }
        if caption:
            payload["caption"] = _truncate(caption, TELEGRAM_MAX_CAPTION_LENGTH)
        if reply_to_message_id is not None:
            payload["reply_parameters"] = {"message_id": reply_to_message_id}
        try:
            async with self._session.post(
                self._api_url(kind.api_method), json=payload, timeout=_API_TIMEOUT,
            ) as response:
                result = await response.json()
                if response.status == 200 and result.get("ok"):
                    return {
                        "success": True,
                        "message_id": result.get("result", {}).get("message_id"),
                        "cached": True,
                    }
                _LOGGER.debug(
                    "Telegram %s (cached) returned non-ok: status=%s code=%s desc=%r — falling back to fresh upload",
                    kind.api_method, response.status, result.get("error_code"),
                    redact(str(result.get("description", ""))),
                )
        except (aiohttp.ClientError, asyncio.TimeoutError, OSError) as err:
            _LOGGER.debug(
                "Telegram %s (cached) transport error — falling back to fresh upload: %s",
                kind.api_method, redact_exc(err),
            )
        return None

    async def _upload_media(
        self,
        kind: _MediaKind,
        chat_id: str,
        data: bytes,
        filename: str,
        content_type: str,
        caption: str | None,
        reply_to_message_id: int | None,
        parse_mode: str,
        cache: TelegramFileCache | None,
        cache_key: str | None,
        thumbhash: str | None,
    ) -> NotificationResult:
        """Multipart-upload ``data`` to Telegram and cache the returned file_id."""
        capped_caption = _truncate(caption, TELEGRAM_MAX_CAPTION_LENGTH) if caption else None

        def _build_form() -> FormData:
            f = FormData()
            f.add_field("chat_id", chat_id)
            f.add_field(kind.form_field, data, filename=filename, content_type=content_type)
            f.add_field("parse_mode", parse_mode)
            if capped_caption:
                f.add_field("caption", capped_caption)
            if reply_to_message_id is not None:
                f.add_field("reply_parameters", json.dumps({"message_id": reply_to_message_id}))
            return f

        for attempt in range(1, _TG_429_MAX_ATTEMPTS + 1):
            try:
                async with self._session.post(
                    self._api_url(kind.api_method),
                    data=_build_form(),
                    timeout=_UPLOAD_TIMEOUT,
                ) as response:
                    result = await response.json()
                    if response.status == 200 and result.get("ok"):
                        res = result.get("result", {})
                        file_id = kind.file_id_from_result(res)
                        if file_id and cache and cache_key:
                            await cache.async_set(
                                cache_key, file_id, kind.cache_type,
                                thumbhash=thumbhash, size=len(data),
                            )
                        return {"success": True, "message_id": res.get("message_id")}

                    if _is_rate_limited(response.status, result) and attempt < _TG_429_MAX_ATTEMPTS:
                        retry_after = _extract_retry_after(result) or 1
                        wait_s = min(retry_after + 1, _TG_429_MAX_WAIT_S)
                        _LOGGER.warning(
                            "Telegram %s 429 (retry_after=%ds, attempt %d/%d) bytes=%d — sleeping %ds",
                            kind.api_method, retry_after, attempt, _TG_429_MAX_ATTEMPTS,
                            len(data), wait_s,
                        )
                        await asyncio.sleep(wait_s)
                        continue

                    _LOGGER.error(
                        "Telegram %s failed: status=%s code=%s desc=%r bytes=%d",
                        kind.api_method, response.status, result.get("error_code"),
                        redact(str(result.get("description", "Unknown"))), len(data),
                    )
                    return {
                        "success": False,
                        "error": redact(str(result.get("description", "Unknown Telegram error"))),
                    }
            except (aiohttp.ClientError, asyncio.TimeoutError, OSError) as err:
                _LOGGER.error(
                    "Telegram %s transport error (bytes=%d): %s",
                    kind.api_method, len(data), redact_exc(err),
                )
                return {"success": False, "error": redact_exc(err)}
        return {"success": False, "error": "Telegram rate limit: max retries exhausted"}

    # ------------------------------------------------------------------
    # Public send entry points
    # ------------------------------------------------------------------

    async def send_notification(
        self,
        chat_id: str,
        assets: list[dict[str, Any]] | None = None,
        caption: str | None = None,
        reply_to_message_id: int | None = None,
        disable_web_page_preview: bool | None = None,
        parse_mode: str = "HTML",
        max_group_size: int = 10,
        chunk_delay: int = 0,
        max_asset_data_size: int | None = None,
        send_large_photos_as_documents: bool = False,
        chat_action: str | None = "typing",
    ) -> NotificationResult:
        if not assets:
            return await self.send_message(
                chat_id, caption or "", reply_to_message_id,
                disable_web_page_preview, parse_mode,
            )

        keepalive: _KeepaliveHandle | None = None
        if chat_action:
            keepalive = self.start_chat_action_keepalive(chat_id, chat_action)

        try:
            if len(assets) == 1 and assets[0].get("type") == "photo":
                return await self._send_photo(
                    chat_id, assets[0].get("url"), caption, reply_to_message_id,
                    parse_mode, max_asset_data_size, send_large_photos_as_documents,
                    assets[0].get("content_type"), assets[0].get("cache_key"),
                    download_headers=assets[0].get("headers"),
                    preloaded_data=assets[0].get("data"),
                )
            if len(assets) == 1 and assets[0].get("type") == "video":
                return await self._send_video(
                    chat_id, assets[0].get("url"), caption, reply_to_message_id,
                    parse_mode, max_asset_data_size,
                    assets[0].get("content_type"), assets[0].get("cache_key"),
                    download_headers=assets[0].get("headers"),
                    preloaded_data=assets[0].get("data"),
                )
            if len(assets) == 1 and assets[0].get("type", "document") == "document":
                url = assets[0].get("url")
                if not url:
                    return {"success": False, "error": "Missing 'url' for document"}
                data, err = await self._fetch_bytes(
                    url, assets[0].get("headers"), assets[0].get("data"),
                )
                if data is None:
                    return {"success": False, "error": f"Failed to download media: {err}"}
                if max_asset_data_size is not None and len(data) > max_asset_data_size:
                    return {"success": False, "error": "Media size exceeds limit"}
                filename = url.split("/")[-1].split("?")[0] or "file"
                return await self._send_document(
                    chat_id, data, filename, caption, reply_to_message_id,
                    parse_mode, url, assets[0].get("content_type"),
                    assets[0].get("cache_key"),
                )

            return await self._send_media_group(
                chat_id, assets, caption, reply_to_message_id, max_group_size,
                chunk_delay, parse_mode, max_asset_data_size,
                send_large_photos_as_documents,
            )
        finally:
            await self.stop_keepalive(keepalive)

    async def send_message(
        self,
        chat_id: str,
        text: str,
        reply_to_message_id: int | None = None,
        disable_web_page_preview: bool | None = None,
        parse_mode: str = "HTML",
    ) -> NotificationResult:
        if not text:
            _LOGGER.warning("send_message called with empty text — using placeholder")
        body = _truncate(text or "Notification", TELEGRAM_MAX_TEXT_LENGTH)
        payload: dict[str, Any] = {
            "chat_id": chat_id,
            "text": body,
            "parse_mode": parse_mode,
        }
        if reply_to_message_id is not None:
            payload["reply_parameters"] = {"message_id": reply_to_message_id}
        if disable_web_page_preview:
            payload["link_preview_options"] = {"is_disabled": True}

        url = self._api_url("sendMessage")
        try:
            async with self._session.post(url, json=payload, timeout=_API_TIMEOUT) as response:
                result = await response.json()
                if response.status == 200 and result.get("ok"):
                    return {"success": True, "message_id": result.get("result", {}).get("message_id")}
                if _is_parse_error(response.status, result):
                    _LOGGER.warning(
                        "Telegram rejected parse_mode=%s (%r); retrying as plain text. "
                        "Check template output for unescaped characters.",
                        payload.get("parse_mode"),
                        redact(str(result.get("description", ""))),
                    )
                    payload.pop("parse_mode", None)
                    # Brief backoff so retry doesn't dogpile on a 4xx.
                    await asyncio.sleep(0.1)
                    async with self._session.post(url, json=payload, timeout=_API_TIMEOUT) as retry_resp:
                        retry_result = await retry_resp.json()
                        if retry_resp.status == 200 and retry_result.get("ok"):
                            return {
                                "success": True,
                                "message_id": retry_result.get("result", {}).get("message_id"),
                            }
                _LOGGER.error(
                    "Telegram sendMessage failed: status=%s code=%s desc=%r",
                    response.status, result.get("error_code"),
                    redact(str(result.get("description", "Unknown"))),
                )
                return {
                    "success": False,
                    "error": redact(str(result.get("description", "Unknown Telegram error"))),
                    "error_code": result.get("error_code"),
                }
        except (aiohttp.ClientError, asyncio.TimeoutError, OSError) as err:
            _LOGGER.error("Telegram sendMessage transport error: %s", redact_exc(err))
            return {"success": False, "error": redact_exc(err)}

    async def send_chat_action(self, chat_id: str, action: str = "typing") -> bool:
        try:
            async with self._session.post(
                self._api_url("sendChatAction"),
                json={"chat_id": chat_id, "action": action},
                timeout=_API_TIMEOUT,
            ) as response:
                result = await response.json()
                return response.status == 200 and result.get("ok", False)
        except (aiohttp.ClientError, asyncio.TimeoutError, OSError):
            return False

    def start_chat_action_keepalive(
        self, chat_id: str, action: str = "typing",
    ) -> _KeepaliveHandle:
        """Repeatedly post ``action`` until stopped.

        Returns a :class:`_KeepaliveHandle` that bundles the asyncio.Task
        and its stop event. Stop cleanly via :meth:`stop_keepalive` —
        setting the event before cancellation prevents the loop from
        firing one last ``sendChatAction`` after the caller's final
        user-visible message, which would otherwise leave a phantom
        indicator hanging.
        """
        handle = _KeepaliveHandle()

        async def action_loop() -> None:
            try:
                while not handle.stop_event.is_set():
                    await self.send_chat_action(chat_id, action)
                    try:
                        await asyncio.wait_for(
                            handle.stop_event.wait(), timeout=_KEEPALIVE_REFRESH_S,
                        )
                    except asyncio.TimeoutError:
                        pass  # refresh window elapsed
            except asyncio.CancelledError:
                pass

        handle.task = asyncio.create_task(action_loop())
        return handle

    @staticmethod
    async def stop_keepalive(handle: _KeepaliveHandle | asyncio.Task | None) -> None:
        """Stop a keepalive started by :meth:`start_chat_action_keepalive`.

        Accepts the new ``_KeepaliveHandle`` form as well as a bare
        ``asyncio.Task`` (back-compat for callers that may have stored
        one before the refactor).
        """
        if handle is None:
            return
        if isinstance(handle, _KeepaliveHandle):
            handle.stop_event.set()
            task = handle.task
        else:
            task = handle
            stop_event = getattr(task, "stop_event", None)
            if stop_event is not None:
                stop_event.set()
        if task is None:
            return
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass

    # ------------------------------------------------------------------
    # Single-media helpers
    # ------------------------------------------------------------------

    async def _send_photo(
        self, chat_id: str, url: str | None, caption: str | None = None,
        reply_to_message_id: int | None = None, parse_mode: str = "HTML",
        max_asset_data_size: int | None = None, send_large_photos_as_documents: bool = False,
        content_type: str | None = None, cache_key: str | None = None,
        download_headers: dict[str, str] | None = None,
        preloaded_data: bytes | None = None,
    ) -> NotificationResult:
        if not url:
            return {"success": False, "error": "Missing 'url' for photo"}

        cache, key, thumbhash = self._get_cache_and_key(url, cache_key)
        cached = cache.get(key, thumbhash=thumbhash) if cache and key else None
        if cached and cached.get("file_id"):
            cached_result = await self._send_from_cache(
                _PHOTO_KIND, chat_id, cached["file_id"],
                caption, reply_to_message_id, parse_mode,
            )
            if cached_result is not None:
                return cached_result

        data, err = await self._fetch_bytes(url, download_headers, preloaded_data)
        if data is None:
            return {"success": False, "error": f"Failed to download photo: {err}"}

        if max_asset_data_size is not None and len(data) > max_asset_data_size:
            return {"success": False, "error": "Photo exceeds size limit", "skipped": True}

        exceeds_limits, reason, _, _ = check_photo_limits(data)
        if exceeds_limits:
            if send_large_photos_as_documents:
                return await self._send_document(
                    chat_id, data, "photo.jpg", caption, reply_to_message_id,
                    parse_mode, url, None, cache_key,
                )
            return {"success": False, "error": f"Photo {reason}", "skipped": True}

        return await self._upload_media(
            _PHOTO_KIND, chat_id, data,
            _PHOTO_KIND.default_filename,
            content_type or _PHOTO_KIND.default_content_type,
            caption, reply_to_message_id, parse_mode,
            cache, key, thumbhash,
        )

    async def _send_video(
        self, chat_id: str, url: str | None, caption: str | None = None,
        reply_to_message_id: int | None = None, parse_mode: str = "HTML",
        max_asset_data_size: int | None = None, content_type: str | None = None,
        cache_key: str | None = None, download_headers: dict[str, str] | None = None,
        preloaded_data: bytes | None = None,
    ) -> NotificationResult:
        if not url:
            return {"success": False, "error": "Missing 'url' for video"}

        cache, key, thumbhash = self._get_cache_and_key(url, cache_key)
        cached = cache.get(key, thumbhash=thumbhash) if cache and key else None
        if cached and cached.get("file_id"):
            cached_result = await self._send_from_cache(
                _VIDEO_KIND, chat_id, cached["file_id"],
                caption, reply_to_message_id, parse_mode,
            )
            if cached_result is not None:
                return cached_result

        data, err = await self._fetch_bytes(url, download_headers, preloaded_data)
        if data is None:
            return {"success": False, "error": f"Failed to download video: {err}"}

        if max_asset_data_size is not None and len(data) > max_asset_data_size:
            return {"success": False, "error": "Video exceeds size limit", "skipped": True}
        if len(data) > TELEGRAM_MAX_VIDEO_SIZE:
            return {
                "success": False,
                "error": f"Video exceeds Telegram's {TELEGRAM_MAX_VIDEO_SIZE // (1024*1024)} MB limit",
                "skipped": True,
            }

        return await self._upload_media(
            _VIDEO_KIND, chat_id, data,
            _VIDEO_KIND.default_filename,
            content_type or _VIDEO_KIND.default_content_type,
            caption, reply_to_message_id, parse_mode,
            cache, key, thumbhash,
        )

    async def _send_document(
        self, chat_id: str, data: bytes, filename: str = "file",
        caption: str | None = None, reply_to_message_id: int | None = None,
        parse_mode: str = "HTML", source_url: str | None = None,
        content_type: str | None = None, cache_key: str | None = None,
    ) -> NotificationResult:
        if not content_type:
            content_type, _ = mimetypes.guess_type(filename)
            if not content_type:
                content_type = _DOCUMENT_KIND.default_content_type

        cache, key, thumbhash = self._get_cache_and_key(source_url, cache_key)
        if cache and key:
            cached = cache.get(key, thumbhash=thumbhash)
            if cached and cached.get("file_id") and cached.get("type") == _DOCUMENT_KIND.cache_type:
                cached_result = await self._send_from_cache(
                    _DOCUMENT_KIND, chat_id, cached["file_id"],
                    caption, reply_to_message_id, parse_mode,
                )
                if cached_result is not None:
                    return cached_result

        return await self._upload_media(
            _DOCUMENT_KIND, chat_id, data, filename, content_type,
            caption, reply_to_message_id, parse_mode,
            cache, key, thumbhash,
        )

    # ------------------------------------------------------------------
    # Media group
    # ------------------------------------------------------------------

    async def _send_media_group(
        self, chat_id: str, assets: list[dict[str, Any]],
        caption: str | None = None, reply_to_message_id: int | None = None,
        max_group_size: int = 10, chunk_delay: int = 0, parse_mode: str = "HTML",
        max_asset_data_size: int | None = None, send_large_photos_as_documents: bool = False,
    ) -> NotificationResult:
        # Telegram rejects mixed photo/video + document in a single
        # sendMediaGroup. Split before chunking so a malformed input
        # batch can't poison the entire send.
        partitions = self._partition_media_by_kind(assets)

        all_message_ids: list[int] = []
        first_chunk_overall = True
        for partition in partitions:
            chunks = [
                partition[i:i + max_group_size]
                for i in range(0, len(partition), max_group_size)
            ]
            for chunk_idx, chunk in enumerate(chunks):
                if not first_chunk_overall and chunk_delay > 0:
                    await asyncio.sleep(chunk_delay / 1000)

                # Single-item chunk → use the simpler send_photo/video path.
                if len(chunk) == 1:
                    item = chunk[0]
                    chunk_caption = caption if first_chunk_overall else None
                    chunk_reply = reply_to_message_id if first_chunk_overall else None
                    if item.get("type") == "photo":
                        result = await self._send_photo(
                            chat_id, item.get("url"), chunk_caption, chunk_reply, parse_mode,
                            max_asset_data_size, send_large_photos_as_documents,
                            item.get("content_type"), item.get("cache_key"),
                            download_headers=item.get("headers"),
                            preloaded_data=item.get("data"),
                        )
                    elif item.get("type") == "video":
                        result = await self._send_video(
                            chat_id, item.get("url"), chunk_caption, chunk_reply, parse_mode,
                            max_asset_data_size,
                            item.get("content_type"), item.get("cache_key"),
                            download_headers=item.get("headers"),
                            preloaded_data=item.get("data"),
                        )
                    else:
                        first_chunk_overall = False
                        continue
                    first_chunk_overall = False
                    if not result.get("success"):
                        result["failed_at_chunk"] = chunk_idx + 1
                        return result
                    if result.get("message_id") is not None:
                        all_message_ids.append(result["message_id"])
                    continue

                items = await self._build_media_items(
                    chunk, max_asset_data_size, caption if first_chunk_overall else None,
                    parse_mode,
                )
                if not items:
                    _LOGGER.warning(
                        "sendMediaGroup skipped — chunk %d/%d had %d input items but 0 usable (all filtered/failed)",
                        chunk_idx + 1, len(chunks), len(chunk),
                    )
                    first_chunk_overall = False
                    continue

                chunk_msg_ids, chunk_err = await self._post_media_group(
                    chat_id, items, reply_to_message_id if first_chunk_overall else None,
                    chunk_idx, len(chunks),
                )
                first_chunk_overall = False
                if chunk_err is not None:
                    return chunk_err
                all_message_ids.extend(chunk_msg_ids)

        if not all_message_ids:
            _LOGGER.warning(
                "sendMediaGroup completed with 0 message_ids — nothing was delivered",
            )
            return {"success": False, "error": "no_items_delivered"}
        return {"success": True, "message_ids": all_message_ids}

    @staticmethod
    def _partition_media_by_kind(
        assets: list[dict[str, Any]],
    ) -> list[list[dict[str, Any]]]:
        """Split assets into runs of compatible kinds.

        Telegram allows photos+videos in the same media group but rejects
        mixing those with documents. We preserve user-provided order
        within each partition so the visual sequence the operator
        configured is honored.
        """
        partitions: list[list[dict[str, Any]]] = []
        current: list[dict[str, Any]] = []
        current_doc = False
        for asset in assets:
            kind = asset.get("type") or "photo"
            is_doc = kind == "document"
            if current and current_doc != is_doc:
                partitions.append(current)
                current = []
            current.append(asset)
            current_doc = is_doc
        if current:
            partitions.append(current)
        return partitions

    async def _build_media_items(
        self,
        chunk: list[dict[str, Any]],
        max_asset_data_size: int | None,
        first_caption: str | None,
        parse_mode: str,
    ) -> list[_MediaItem]:
        """Fetch + filter a chunk and return aligned media-group items.

        Concurrency is bounded by ``_MEDIA_FETCH_CONCURRENCY`` so peak
        memory stays predictable. Per-fetch exceptions are isolated via
        ``return_exceptions=True`` so a single failed download cannot
        cancel its peers.
        """
        sem = asyncio.Semaphore(_MEDIA_FETCH_CONCURRENCY)

        async def fetch(idx: int, item: dict[str, Any]) -> tuple[int, dict | None, bytes | None]:
            url = item.get("url")
            if not url:
                _LOGGER.warning("Media skipped: missing url (idx=%d type=%s)", idx, item.get("type"))
                return idx, None, None
            media_type = item.get("type", "photo")
            custom_cache_key = item.get("cache_key")

            ck = custom_cache_key or extract_asset_id_from_url(url) or url
            ck_is_asset = is_asset_cache_key(ck)
            item_cache = self._get_cache_for_key(ck, ck_is_asset)
            bare_ck = asset_id_from_cache_key(ck) if ck_is_asset else ck
            item_thumbhash = (
                self._thumbhash_resolver(bare_ck)
                if ck_is_asset and self._thumbhash_resolver else None
            )
            cached = item_cache.get(ck, thumbhash=item_thumbhash) if item_cache else None
            if cached and cached.get("file_id"):
                return idx, cached, None

            preloaded = item.get("data")
            data: bytes | None
            if preloaded is not None:
                data = preloaded
            else:
                async with sem:
                    data, err = await self._safe_get(self._resolve_url(url), item.get("headers"))
                if data is None:
                    _LOGGER.warning(
                        "Media skipped: download failed (idx=%d type=%s): %s",
                        idx, media_type, err,
                    )
                    return idx, None, None

            if max_asset_data_size and len(data) > max_asset_data_size:
                _LOGGER.warning(
                    "Media skipped: size %d exceeds max_asset_data_size %d (idx=%d type=%s)",
                    len(data), max_asset_data_size, idx, media_type,
                )
                return idx, None, None
            if media_type == "video" and len(data) > TELEGRAM_MAX_VIDEO_SIZE:
                _LOGGER.warning(
                    "Media skipped: video %d bytes exceeds Telegram limit %d (idx=%d)",
                    len(data), TELEGRAM_MAX_VIDEO_SIZE, idx,
                )
                return idx, None, None
            if media_type == "photo":
                exceeds, reason, _, _ = check_photo_limits(data)
                if exceeds:
                    _LOGGER.warning(
                        "Media skipped: photo %s (idx=%d)", reason, idx,
                    )
                    return idx, None, None
            return idx, None, data

        raw = await asyncio.gather(
            *(fetch(i, item) for i, item in enumerate(chunk)),
            return_exceptions=True,
        )
        results: list[tuple[int, dict | None, bytes | None]] = []
        for entry in raw:
            if isinstance(entry, Exception):
                _LOGGER.warning("Media fetch raised: %s", redact_exc(entry))
                continue
            results.append(entry)

        items: list[_MediaItem] = []
        upload_idx = 0
        for idx, cached_entry, data in results:
            item = chunk[idx]
            url = item.get("url")
            if not url:
                continue
            media_type = item.get("type") or "photo"
            custom_cache_key = item.get("cache_key")

            if cached_entry and cached_entry.get("file_id"):
                mij: dict[str, Any] = {"type": media_type, "media": cached_entry["file_id"]}
                cache_info: tuple[str, str, str | None, int] | None = None
                attachment: tuple[str, bytes, str, str] | None = None
            elif data is not None:
                attach_name = f"file{upload_idx}"
                ct = item.get("content_type") or ("image/jpeg" if media_type == "photo" else "video/mp4")
                ext = "jpg" if media_type == "photo" else "mp4"
                attachment = (attach_name, data, f"media_{idx}.{ext}", ct)
                mij = {"type": media_type, "media": f"attach://{attach_name}"}
                upload_idx += 1
                ck = custom_cache_key or extract_asset_id_from_url(url) or url
                ck_is_asset = is_asset_cache_key(ck)
                bare_ck = asset_id_from_cache_key(ck) if ck_is_asset else ck
                th = (
                    self._thumbhash_resolver(bare_ck)
                    if ck_is_asset and self._thumbhash_resolver else None
                )
                cache_info = (ck, media_type, th, len(data))
            else:
                continue

            if first_caption and not items:
                # Only the first usable item in the first chunk receives
                # the caption, per Telegram's media-group semantics.
                mij["caption"] = _truncate(first_caption, TELEGRAM_MAX_CAPTION_LENGTH)
                mij["parse_mode"] = parse_mode

            items.append(_MediaItem(media_json=mij, cache_info=cache_info, attachment=attachment))
        return items

    async def _post_media_group(
        self,
        chat_id: str,
        items: list[_MediaItem],
        reply_to_message_id: int | None,
        chunk_idx: int,
        total_chunks: int,
    ) -> tuple[list[int], dict[str, Any] | None]:
        """POST the media group with bounded 429 retry.

        Returns ``(message_ids, None)`` on success or ``([], err_dict)``.
        """
        media_json = [it.media_json for it in items]
        attachments = [it.attachment for it in items if it.attachment]
        cache_infos = [it.cache_info for it in items]

        def _build_form() -> FormData:
            f = FormData()
            f.add_field("chat_id", chat_id)
            if reply_to_message_id is not None:
                f.add_field("reply_parameters", json.dumps({"message_id": reply_to_message_id}))
            for name, payload, filename, ct in attachments:
                f.add_field(name, payload, filename=filename, content_type=ct)
            f.add_field("media", json.dumps(media_json))
            return f

        for attempt in range(1, _TG_429_MAX_ATTEMPTS + 1):
            try:
                async with self._session.post(
                    self._api_url("sendMediaGroup"),
                    data=_build_form(),
                    timeout=_UPLOAD_TIMEOUT,
                ) as response:
                    result = await response.json()
                    if response.status == 200 and result.get("ok"):
                        result_msgs = result.get("result", [])
                        msg_ids = [
                            msg.get("message_id") for msg in result_msgs
                            if msg.get("message_id") is not None
                        ]
                        await self._cache_media_group_response(result_msgs, cache_infos)
                        return msg_ids, None

                    if _is_rate_limited(response.status, result) and attempt < _TG_429_MAX_ATTEMPTS:
                        retry_after = _extract_retry_after(result) or 1
                        wait_s = min(retry_after + 1, _TG_429_MAX_WAIT_S)
                        _LOGGER.warning(
                            "Telegram sendMediaGroup 429 (retry_after=%ds, attempt %d/%d) chunk=%d/%d items=%d — sleeping %ds",
                            retry_after, attempt, _TG_429_MAX_ATTEMPTS,
                            chunk_idx + 1, total_chunks, len(media_json), wait_s,
                        )
                        await asyncio.sleep(wait_s)
                        continue

                    _LOGGER.error(
                        "Telegram sendMediaGroup failed: status=%s code=%s desc=%r chunk=%d/%d items=%d",
                        response.status, result.get("error_code"),
                        redact(str(result.get("description", "Unknown"))),
                        chunk_idx + 1, total_chunks, len(media_json),
                    )
                    return [], {
                        "success": False,
                        "error": redact(str(result.get("description", "Unknown"))),
                        "error_code": result.get("error_code"),
                        "failed_at_chunk": chunk_idx + 1,
                    }
            except (aiohttp.ClientError, asyncio.TimeoutError, OSError) as err:
                _LOGGER.error(
                    "Telegram sendMediaGroup transport error on chunk %d/%d (%d items): %s",
                    chunk_idx + 1, total_chunks, len(media_json), redact_exc(err),
                )
                return [], {
                    "success": False,
                    "error": redact_exc(err),
                    "failed_at_chunk": chunk_idx + 1,
                }

        return [], {
            "success": False,
            "error": "Telegram rate limit: max retries exhausted",
            "failed_at_chunk": chunk_idx + 1,
        }

    async def _cache_media_group_response(
        self,
        result_msgs: list[dict[str, Any]],
        cache_infos: list[tuple[str, str, str | None, int] | None],
    ) -> None:
        """Persist file_ids returned by sendMediaGroup.

        Entries align by position: ``cache_infos[i]`` corresponds to
        ``result_msgs[i]``. ``None`` means the item was a cache hit and
        does not need re-storing.
        """
        cache_entries: list[tuple[str, str, str, str | None, int | None]] = []
        for i, msg in enumerate(result_msgs):
            if i >= len(cache_infos):
                break
            info = cache_infos[i]
            if info is None:
                continue
            ck, mt, th, sz = info
            file_id: str | None = None
            if msg.get("photo"):
                file_id = msg["photo"][-1].get("file_id")
            elif msg.get("video"):
                file_id = msg["video"].get("file_id")
            elif msg.get("document"):
                file_id = msg["document"].get("file_id")
            if file_id:
                cache_entries.append((ck, file_id, mt, th, sz))
        if cache_entries:
            eff_cache = self._get_cache_for_key(
                cache_entries[0][0], is_asset_cache_key(cache_entries[0][0]),
            )
            if eff_cache:
                await eff_cache.async_set_many(cache_entries)

    # ------------------------------------------------------------------
    # Bot management methods
    # ------------------------------------------------------------------

    async def _api_call(
        self,
        method: str,
        *,
        json_payload: dict[str, Any] | None = None,
        params: dict[str, Any] | None = None,
        timeout: aiohttp.ClientTimeout | None = None,
        http_method: str = "POST",
    ) -> dict[str, Any]:
        """Generic Bot-API call with redacted errors and explicit timeout."""
        url = self._api_url(method)
        try:
            async with self._session.request(
                http_method, url, json=json_payload, params=params,
                timeout=timeout or _API_TIMEOUT,
            ) as resp:
                data = await resp.json()
                if data.get("ok"):
                    return {"success": True, "result": data.get("result", {})}
                return {
                    "success": False,
                    "error": redact(str(data.get("description", "Unknown error"))),
                }
        except (aiohttp.ClientError, asyncio.TimeoutError, OSError) as err:
            return {"success": False, "error": redact_exc(err)}

    async def get_me(self) -> dict[str, Any]:
        """Call getMe to verify the bot token and get bot info."""
        return await self._api_call("getMe", http_method="GET")

    async def get_chat(self, chat_id: str) -> dict[str, Any]:
        """Call getChat to fetch up-to-date chat metadata."""
        return await self._api_call("getChat", json_payload={"chat_id": chat_id})

    async def get_webhook_info(self) -> dict[str, Any]:
        """Call getWebhookInfo to check current webhook status."""
        return await self._api_call("getWebhookInfo", http_method="GET")

    async def set_webhook(self, webhook_url: str, secret: str | None = None) -> dict[str, Any]:
        """Register a webhook URL with Telegram."""
        payload: dict[str, Any] = {"url": webhook_url}
        if secret:
            payload["secret_token"] = secret
        result = await self._api_call("setWebhook", json_payload=payload)
        # Caller previously got plain ``{"success": True}``; preserve.
        if result.get("success"):
            return {"success": True}
        return result

    async def delete_webhook(self) -> dict[str, Any]:
        """Remove the webhook from Telegram."""
        result = await self._api_call("deleteWebhook")
        if result.get("success"):
            return {"success": True}
        return result

    async def get_updates(
        self, offset: int | None = None, limit: int = 50, timeout: int = 0,
    ) -> dict[str, Any]:
        """Long-poll for updates via getUpdates."""
        url = self._api_url("getUpdates")
        params: dict[str, Any] = {
            "timeout": timeout,
            "limit": limit,
            "allowed_updates": json.dumps(["message"]),
        }
        if offset is not None:
            params["offset"] = offset
        try:
            async with self._session.get(
                url, params=params,
                timeout=aiohttp.ClientTimeout(total=max(10, timeout + 5)),
            ) as resp:
                data = await resp.json()
                if data.get("ok"):
                    return {"success": True, "result": data.get("result", [])}
                return {
                    "success": False,
                    "error": redact(str(data.get("description", "Unknown error"))),
                }
        except (aiohttp.ClientError, asyncio.TimeoutError, OSError) as err:
            return {"success": False, "error": redact_exc(err)}

    async def set_my_commands(
        self, commands: list[dict[str, str]], language_code: str | None = None,
        scope: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        """Register bot commands with the BotFather API."""
        payload: dict[str, Any] = {"commands": commands}
        if language_code:
            payload["language_code"] = language_code
        if scope:
            payload["scope"] = scope
        result = await self._api_call("setMyCommands", json_payload=payload)
        if result.get("success"):
            return {"success": True}
        return result

    async def delete_my_commands(
        self, language_code: str | None = None,
        scope: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        """Clear bot commands for the given scope/language."""
        payload: dict[str, Any] = {}
        if language_code:
            payload["language_code"] = language_code
        if scope:
            payload["scope"] = scope
        result = await self._api_call("deleteMyCommands", json_payload=payload)
        if result.get("success"):
            return {"success": True}
        return result
