"""Shared dispatch helpers used by both watcher and webhook handlers."""

from __future__ import annotations

import dataclasses
import logging
import random
from dataclasses import dataclass
from datetime import datetime, time, timedelta, timezone
from enum import Enum
from typing import Any, Callable
from zoneinfo import ZoneInfo, ZoneInfoNotFoundError

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

from notify_bridge_core.models.events import ServiceEvent
from notify_bridge_core.models.media import MediaAsset
from notify_bridge_core.notifications.receiver import Receiver, build_receiver

from ..database.models import (
    EmailBot,
    MatrixBot,
    NotificationTarget,
    NotificationTracker,
    NotificationTrackerTarget,
    TargetReceiver,
    TelegramBot,
    TelegramChat,
    TemplateConfig,
    TemplateSlot,
    TrackingConfig,
)

_LOGGER = logging.getLogger(__name__)


class GateReason(str, Enum):
    """Why ``evaluate_event_gate`` allowed or blocked a dispatch.

    String-backed so it can be persisted in ``EventLog.details`` JSON and
    round-trip cleanly.
    """

    ALLOWED = "allowed"
    EVENT_TYPE_DISABLED = "event_type_disabled"
    QUIET_HOURS = "quiet_hours"


@dataclass(frozen=True)
class GateOutcome:
    """Result of evaluating a (event, tracking_config) pair against dispatch gates.

    ``quiet_hours_end_at`` is set iff ``reason == QUIET_HOURS`` and gives the
    UTC datetime at which the current quiet window ends — used by the
    deferred-dispatch scheduler to know when to fire the held notification.
    """

    reason: GateReason
    quiet_hours_end_at: datetime | None = None

    @property
    def allowed(self) -> bool:
        return self.reason is GateReason.ALLOWED


def _resolve_zoneinfo(tz_name: str | None) -> ZoneInfo:
    """Resolve an IANA tz string to a ZoneInfo, falling back to UTC on any error."""
    if not tz_name:
        return ZoneInfo("UTC")
    try:
        return ZoneInfo(tz_name)
    except (ZoneInfoNotFoundError, ValueError):
        _LOGGER.warning("Unknown timezone %r; falling back to UTC", tz_name)
        return ZoneInfo("UTC")


def quiet_hours_status(
    start: str | None,
    end: str | None,
    tz_name: str | None = "UTC",
) -> datetime | None:
    """Return the UTC datetime when the current quiet window ends, or None.

    Returns ``None`` when:
      * either bound is missing,
      * the bounds are malformed,
      * the current local time is outside the configured window.

    Returns a UTC ``datetime`` aligned to ``HH:MM`` (seconds=0, microseconds=0)
    representing the next end-of-window moment after "now" when the current
    time IS inside the window. For overnight windows (e.g. 22:00-06:00) the
    end may be tomorrow.
    """
    if not start or not end:
        return None
    try:
        tz = _resolve_zoneinfo(tz_name)
        now_local = datetime.now(timezone.utc).astimezone(tz)
        t_start = time.fromisoformat(start)
        t_end = time.fromisoformat(end)
    except (ValueError, TypeError):
        return None

    # ``start == end`` (e.g. "00:00-00:00") has no consistent meaning: under
    # the normal-window branch the window is one instant wide; under the
    # overnight-window branch it's effectively always-on. Either is almost
    # certainly a user mistake, so treat it as "no window configured" rather
    # than silently deferring every notification all day.
    if t_start == t_end:
        return None

    now_t = now_local.time()
    if t_start <= t_end:
        in_window = t_start <= now_t <= t_end
    else:
        in_window = now_t >= t_start or now_t <= t_end
    if not in_window:
        return None

    end_today = now_local.replace(
        hour=t_end.hour, minute=t_end.minute, second=0, microsecond=0,
    )
    # If today's end already passed (overnight window, post-midnight half),
    # the actual end is tomorrow at the same wall-clock time.
    if end_today <= now_local:
        end_today = end_today + timedelta(days=1)
    return end_today.astimezone(timezone.utc)


def in_quiet_hours(
    start: str | None,
    end: str | None,
    tz_name: str | None = "UTC",
) -> bool:
    """Check if the current time (in the given timezone) is within the quiet window.

    Thin wrapper over ``quiet_hours_status`` preserved for back-compat with
    callers that only need the boolean. New code should prefer
    ``quiet_hours_status`` (or ``evaluate_event_gate``) when the window end
    time matters.
    """
    return quiet_hours_status(start, end, tz_name) is not None


async def get_app_timezone(session: AsyncSession) -> str:
    """Load the app-level timezone from AppSetting (falls back to UTC)."""
    from ..api.app_settings import get_setting
    value = await get_setting(session, "timezone")
    return value or "UTC"


def _event_type_enabled(event: ServiceEvent, tc: TrackingConfig) -> bool:
    """Return True iff the tracking config's per-event-type flag allows this event.

    Quiet hours are NOT considered here — this is the user's "do I care about
    this kind of event at all" gate. See ``evaluate_event_gate`` for the
    combined gate that also folds in quiet hours.
    """
    event_type = event.event_type.value
    flag_map = {
        # Immich events
        "assets_added": tc.track_assets_added,
        "assets_removed": tc.track_assets_removed,
        "collection_renamed": tc.track_collection_renamed,
        "collection_deleted": tc.track_collection_deleted,
        "sharing_changed": tc.track_sharing_changed,
        # Gitea events
        "push": tc.track_push,
        "issue_opened": tc.track_issue_opened,
        "issue_closed": tc.track_issue_closed,
        "issue_commented": tc.track_issue_commented,
        "pr_opened": tc.track_pr_opened,
        "pr_closed": tc.track_pr_closed,
        "pr_merged": tc.track_pr_merged,
        "pr_commented": tc.track_pr_commented,
        "release_published": tc.track_release_published,
        # Planka events
        "card_created": tc.track_card_created,
        "card_updated": tc.track_card_updated,
        "card_moved": tc.track_card_moved,
        "card_deleted": tc.track_card_deleted,
        "card_commented": tc.track_card_commented,
        "comment_updated": tc.track_comment_updated,
        "board_created": tc.track_board_created,
        "board_updated": tc.track_board_updated,
        "board_deleted": tc.track_board_deleted,
        "list_created": tc.track_list_created,
        "list_updated": tc.track_list_updated,
        "list_deleted": tc.track_list_deleted,
        "attachment_created": tc.track_attachment_created,
        "card_label_added": tc.track_card_label_added,
        "task_completed": tc.track_task_completed,
        # Scheduler events
        "scheduled_message": tc.track_scheduled_message,
        # Generic Webhook events
        "webhook_received": tc.track_webhook_received,
        # NUT (UPS) events
        "ups_online": tc.track_ups_online,
        "ups_on_battery": tc.track_ups_on_battery,
        "ups_low_battery": tc.track_ups_low_battery,
        "ups_battery_restored": tc.track_ups_battery_restored,
        "ups_comms_lost": tc.track_ups_comms_lost,
        "ups_comms_restored": tc.track_ups_comms_restored,
        "ups_replace_battery": tc.track_ups_replace_battery,
        "ups_overload": tc.track_ups_overload,
        # Home Assistant events — use getattr so legacy DB rows / test mocks
        # that pre-date the columns still pass the gate (default to tracked).
        "ha_state_changed": getattr(tc, "track_ha_state_changed", True),
        "ha_automation_triggered": getattr(tc, "track_ha_automation_triggered", False),
        "ha_service_called": getattr(tc, "track_ha_service_called", False),
        "ha_event_fired": getattr(tc, "track_ha_event_fired", False),
        # Bridge self-monitoring — defaults True so a tracker created before the
        # columns existed still surfaces the alerts. Legacy rows are extremely
        # unlikely here (the columns ship in the same release as the provider),
        # but the safer default matches the rest of this map.
        "bridge_self_poll_failures": getattr(tc, "track_bridge_self_poll_failures", True),
        "bridge_self_deferred_backlog": getattr(tc, "track_bridge_self_deferred_backlog", True),
        "bridge_self_target_failures": getattr(tc, "track_bridge_self_target_failures", True),
    }
    return flag_map.get(event_type, True)


def evaluate_event_gate(
    event: ServiceEvent,
    tc: TrackingConfig,
    tz_name: str | None = "UTC",
) -> GateOutcome:
    """Decide whether an event should dispatch through the given tracking config.

    Returns a :class:`GateOutcome` carrying both the verdict and — when blocked
    by quiet hours — the UTC datetime at which the window ends so the caller
    can schedule a deferred dispatch.

    Order of checks: per-event-type flag FIRST, then quiet hours. Otherwise a
    disabled event type would get deferred during quiet hours and then
    silently dropped at drain time — wasted work and a confusing "deferred
    then dropped" trail in the dashboard. The user already said "don't tell
    me about this kind of event"; honour that immediately.
    """
    if not _event_type_enabled(event, tc):
        return GateOutcome(reason=GateReason.EVENT_TYPE_DISABLED)

    if tc.quiet_hours_enabled:
        end_at = quiet_hours_status(
            tc.quiet_hours_start, tc.quiet_hours_end, tz_name,
        )
        if end_at is not None:
            return GateOutcome(
                reason=GateReason.QUIET_HOURS,
                quiet_hours_end_at=end_at,
            )

    return GateOutcome(reason=GateReason.ALLOWED)


def event_allowed_by_config(
    event: ServiceEvent,
    tc: TrackingConfig,
    tz_name: str | None = "UTC",
) -> bool:
    """Boolean back-compat wrapper around :func:`evaluate_event_gate`.

    New call sites should use ``evaluate_event_gate`` directly so they can
    distinguish a quiet-hours suppression (deferrable) from an event-type
    disable (drop forever).
    """
    return evaluate_event_gate(event, tc, tz_name).allowed


# --- Display-time filters driven by TrackingConfig -------------------------
#
# These transform a ServiceEvent so the dispatched notification reflects the
# user's per-tracker "asset display" preferences. Event-tracking flags (which
# events fire at all) live in ``event_allowed_by_config`` above; the filters
# here only reshape an already-allowed event.

# Asset.extra keys stripped when ``include_asset_details=False``. These are
# the enrichment fields the default templates render as prose (city/country,
# ⭐ rating, ❤️ favorite). ``thumbhash``/``file_size``/``playback_size``/
# ``owner_id``/``cache_key`` stay — they are load-bearing for media send and
# caching, not user-facing prose.
_ASSET_DETAIL_KEYS: tuple[str, ...] = (
    "city", "country", "state",
    "latitude", "longitude",
    "is_favorite", "rating",
)


def _sort_key_for(order_by: str) -> Callable[[MediaAsset], Any] | None:
    if order_by == "date":
        return lambda a: a.created_at
    if order_by == "name":
        return lambda a: a.filename.lower()
    if order_by == "rating":
        # None ratings sort last regardless of direction.
        return lambda a: (
            a.extra.get("rating") is None,
            a.extra.get("rating") or 0,
        )
    return None


def _sort_assets(
    assets: list[MediaAsset],
    order_by: str,
    order: str,
) -> list[MediaAsset]:
    """Sort MediaAssets by the configured key/direction.

    ``order_by="none"`` preserves the input order (the provider's own
    ordering, usually detection order). ``"random"`` shuffles in place
    on a copy so repeated renders of the same event aren't identical.
    """
    if order_by in ("none", "") or len(assets) < 2:
        return list(assets)
    if order_by == "random":
        shuffled = list(assets)
        random.shuffle(shuffled)
        return shuffled
    key_fn = _sort_key_for(order_by)
    if key_fn is None:
        return list(assets)
    return sorted(assets, key=key_fn, reverse=(order == "descending"))


def _transform_asset(
    asset: MediaAsset,
    *,
    strip_details: bool,
    strip_tags: bool,
) -> MediaAsset:
    """Return a copy of ``asset`` with details and/or tags removed."""
    new_extra = asset.extra
    new_description = asset.description
    new_tags = asset.tags
    if strip_details:
        new_extra = {k: v for k, v in asset.extra.items() if k not in _ASSET_DETAIL_KEYS}
        new_description = None
    if strip_tags:
        new_tags = []
    return dataclasses.replace(
        asset,
        description=new_description,
        tags=list(new_tags) if new_tags is not asset.tags else asset.tags,
        extra=new_extra,
    )


def apply_tracking_display_filters(
    event: ServiceEvent,
    tc: TrackingConfig | None,
) -> ServiceEvent | None:
    """Apply per-tracker display preferences to an already-allowed event.

    Semantics:
      * ``notify_favorites_only`` + ``assets_order_by`` + ``max_assets_to_show``
        only apply to ``ASSETS_ADDED`` events — the album-change path. Scheduled
        / periodic / memory events have their own limits and ordering
        (``scheduled_limit``, ``scheduled_order_by``, etc.), so reapplying the
        album-change cap would wrongly truncate them.
      * ``include_tags`` and ``include_asset_details`` apply to every event
        that carries assets, since they control rendering irrespective of
        how the assets were selected.

    Returns:
        A new ``ServiceEvent`` with filters applied, or ``None`` if the event
        should be dropped entirely (``notify_favorites_only=True`` and none of
        the added assets are favorites).
    """
    if tc is None:
        return event

    assets = list(event.added_assets)
    new_added_count = event.added_count
    is_change_event = event.event_type.value == "assets_added"

    if is_change_event:
        if tc.notify_favorites_only:
            assets = [a for a in assets if a.extra.get("is_favorite")]
            new_added_count = len(assets)
            if not assets:
                return None
        assets = _sort_assets(assets, tc.assets_order_by, tc.assets_order)
        if tc.max_assets_to_show >= 0:
            assets = assets[: tc.max_assets_to_show]

    strip_details = not tc.include_asset_details
    strip_tags = not tc.include_tags
    if (strip_details or strip_tags) and assets:
        assets = [
            _transform_asset(a, strip_details=strip_details, strip_tags=strip_tags)
            for a in assets
        ]

    new_extra = event.extra
    if strip_tags and "people" in event.extra:
        new_extra = {k: v for k, v in event.extra.items() if k != "people"}

    return dataclasses.replace(
        event,
        added_assets=assets,
        added_count=new_added_count,
        extra=new_extra,
    )


def resolve_provider_credential(cfg: dict[str, Any] | None) -> str | None:
    """Pick the first non-empty provider credential field.

    Provider configs use different field names (Immich → ``api_key``,
    Gitea → ``api_token``, HA → ``access_token``). All four dispatch
    sites used to pick one field by hand; centralising here keeps the
    fallback order consistent so a config edit on one provider type can't
    silently break dispatch for another.
    """
    if not cfg:
        return None
    return cfg.get("api_key") or cfg.get("api_token") or cfg.get("access_token")


async def _resolve_target(
    session: AsyncSession,
    target: NotificationTarget,
    *,
    receivers_by_target: dict[int, list[TargetReceiver]] | None = None,
    telegram_chats_by_bot: dict[int, dict[str, TelegramChat]] | None = None,
    email_bots_by_id: dict[int, EmailBot] | None = None,
    matrix_bots_by_id: dict[int, MatrixBot] | None = None,
) -> dict[str, Any]:
    """Resolve a single target into dispatch-ready data (config + receivers + credentials).

    Returns a dict with target_type, target_config, and receivers.
    Does NOT include tracking_config or template_slots — those come from the tracker link.

    Optional ``*_by_*`` maps short-circuit per-target DB queries when the
    caller has batch-prefetched the data. When omitted, we fall back to the
    original single-query path so direct callers (manual_dispatch) still work.
    """
    # Receivers — prefer pre-fetched map.
    if receivers_by_target is not None:
        recv_rows = [r for r in receivers_by_target.get(target.id, []) if r.enabled]
    else:
        recv_result = await session.exec(
            select(TargetReceiver).where(
                TargetReceiver.target_id == target.id,
                TargetReceiver.enabled == True,
            )
        )
        recv_rows = recv_result.all()

    # For Telegram targets, resolve locale from TelegramChat
    chat_locale_map: dict[str, str] = {}
    if target.type == "telegram":
        bot_id = target.config.get("bot_id")
        if bot_id:
            chat_ids = [str(r.config.get("chat_id", "")) for r in recv_rows if r.config.get("chat_id")]
            if chat_ids:
                chats_for_bot = (
                    telegram_chats_by_bot.get(bot_id, {})
                    if telegram_chats_by_bot is not None else None
                )
                if chats_for_bot is not None:
                    rows = [
                        chats_for_bot[cid] for cid in chat_ids
                        if cid in chats_for_bot
                    ]
                else:
                    chat_result = await session.exec(
                        select(TelegramChat).where(
                            TelegramChat.bot_id == bot_id,
                            TelegramChat.chat_id.in_(chat_ids),
                        )
                    )
                    rows = chat_result.all()
                for chat in rows:
                    resolved = (
                        getattr(chat, 'language_override', '') or
                        getattr(chat, 'language_code', '') or ''
                    )
                    if resolved:
                        chat_locale_map[chat.chat_id] = resolved[:2].lower()

    receivers: list[Receiver] = []
    for r in recv_rows:
        explicit_locale = getattr(r, 'locale', '') or ''
        locale = explicit_locale
        if not locale and target.type == "telegram":
            chat_id = str(r.config.get("chat_id", ""))
            locale = chat_locale_map.get(chat_id, "")
        receivers.append(build_receiver(target.type, dict(r.config), locale))

    target_config = dict(target.config)
    # chat_action lives on the model column — single source of truth.
    # Strip any legacy/stale value from config so an old config-stored value
    # can't shadow the user's UI choice. When the column is unset, leave the
    # key absent so the dispatcher's "typing" fallback applies.
    target_config.pop("chat_action", None)
    if hasattr(target, 'chat_action') and target.chat_action:
        target_config["chat_action"] = target.chat_action
    # Inject bot credentials for bot-backed target types
    if target.type == "email":
        email_bot_id = target.config.get("email_bot_id")
        if email_bot_id:
            if email_bots_by_id is not None:
                email_bot = email_bots_by_id.get(email_bot_id)
            else:
                email_bot = await session.get(EmailBot, email_bot_id)
            if email_bot:
                target_config["smtp"] = {
                    "host": email_bot.smtp_host,
                    "port": email_bot.smtp_port,
                    "username": email_bot.smtp_username,
                    "password": email_bot.smtp_password,
                    "from_address": email_bot.email,
                    "from_name": email_bot.name,
                    "use_tls": email_bot.smtp_use_tls,
                }
    elif target.type == "matrix":
        matrix_bot_id = target.config.get("matrix_bot_id")
        if matrix_bot_id:
            if matrix_bots_by_id is not None:
                matrix_bot = matrix_bots_by_id.get(matrix_bot_id)
            else:
                matrix_bot = await session.get(MatrixBot, matrix_bot_id)
            if matrix_bot:
                target_config["homeserver_url"] = matrix_bot.homeserver_url
                target_config["access_token"] = matrix_bot.access_token

    return {
        "target_id": target.id,
        "target_name": target.name,
        "target_type": target.type,
        "target_config": target_config,
        "receivers": receivers,
    }


async def load_link_data(
    session: AsyncSession,
    tracker_id: int,
    *,
    check_quiet_hours: bool = False,
) -> list[dict[str, Any]]:
    """Load tracker-target link data for dispatch.

    Args:
        session: Active database session.
        tracker_id: ID of the tracker whose links to load.
        check_quiet_hours: If True, skip links currently in quiet hours.
    """
    # Load the tracker itself for default config IDs
    tracker = await session.get(NotificationTracker, tracker_id)
    default_tc_id = getattr(tracker, "default_tracking_config_id", None) if tracker else None
    default_tmpl_id = getattr(tracker, "default_template_config_id", None) if tracker else None

    tt_result = await session.exec(
        select(NotificationTrackerTarget).where(
            NotificationTrackerTarget.tracker_id == tracker_id
        )
    )
    tracker_targets = tt_result.all()

    # Filter enabled links and quiet hours upfront
    active_links = [
        tt for tt in tracker_targets
        if tt.enabled and not (check_quiet_hours and in_quiet_hours(tt.quiet_hours_start, tt.quiet_hours_end))
    ]
    if not active_links:
        return []

    # Batch-load targets
    target_ids = list({tt.target_id for tt in active_links})
    target_result = await session.exec(
        select(NotificationTarget).where(NotificationTarget.id.in_(target_ids))
    )
    target_map = {t.id: t for t in target_result.all()}

    # Batch-load tracking configs (per-link + tracker default)
    tc_ids = list({tid for tid in
        [tt.tracking_config_id for tt in active_links] + [default_tc_id]
        if tid})
    tc_map: dict[int, TrackingConfig] = {}
    if tc_ids:
        tc_result = await session.exec(select(TrackingConfig).where(TrackingConfig.id.in_(tc_ids)))
        tc_map = {tc.id: tc for tc in tc_result.all()}

    # Batch-load template configs (per-link + tracker default)
    tmpl_ids = list({tid for tid in
        [tt.template_config_id for tt in active_links] + [default_tmpl_id]
        if tid})
    tmpl_map: dict[int, TemplateConfig] = {}
    if tmpl_ids:
        tmpl_result = await session.exec(select(TemplateConfig).where(TemplateConfig.id.in_(tmpl_ids)))
        tmpl_map = {tc.id: tc for tc in tmpl_result.all()}

    # Batch-load template slots for all template configs
    slots_by_config: dict[int, dict[str, dict[str, str]]] = {}
    if tmpl_ids:
        slot_result = await session.exec(
            select(TemplateSlot).where(TemplateSlot.config_id.in_(tmpl_ids))
        )
        for s in slot_result.all():
            event_key = s.slot_name.removeprefix("message_") if s.slot_name.startswith("message_") else s.slot_name
            slots_by_config.setdefault(s.config_id, {}).setdefault(event_key, {})[s.locale] = s.template

    # Pre-resolve broadcast children in one query to avoid N+1 per-child fetches
    broadcast_child_ids: set[int] = set()
    for tt in active_links:
        target = target_map.get(tt.target_id)
        if target and target.type == "broadcast":
            disabled_ids = set(target.config.get("disabled_child_ids", []))
            for cid in target.config.get("child_target_ids", []):
                if cid not in disabled_ids:
                    broadcast_child_ids.add(cid)
    child_target_map: dict[int, NotificationTarget] = {}
    if broadcast_child_ids:
        child_rows = await session.exec(
            select(NotificationTarget).where(NotificationTarget.id.in_(broadcast_child_ids))
        )
        child_target_map = {t.id: t for t in child_rows.all()}

    # ---- Batch pre-fetch for _resolve_target ----
    # Build the universe of target IDs (regular + expanded broadcast children)
    # so a single query per relation type covers every call to _resolve_target
    # below — no per-target follow-up SELECTs.
    all_target_ids: set[int] = set(target_map.keys()) | set(child_target_map.keys())

    receivers_by_target: dict[int, list[TargetReceiver]] = {}
    if all_target_ids:
        recv_result = await session.exec(
            select(TargetReceiver).where(TargetReceiver.target_id.in_(all_target_ids))
        )
        for r in recv_result.all():
            receivers_by_target.setdefault(r.target_id, []).append(r)

    # Telegram chats keyed by (bot_id, chat_id) — collect all (bot_id, chat_id)
    # pairs referenced by enabled telegram-target receivers, then one query.
    tg_pairs: dict[int, set[str]] = {}  # bot_id -> {chat_id}
    for tid in all_target_ids:
        tgt = target_map.get(tid) or child_target_map.get(tid)
        if not tgt or tgt.type != "telegram":
            continue
        bot_id = tgt.config.get("bot_id")
        if not bot_id:
            continue
        for r in receivers_by_target.get(tid, []):
            cid = str(r.config.get("chat_id", ""))
            if cid:
                tg_pairs.setdefault(bot_id, set()).add(cid)
    telegram_chats_by_bot: dict[int, dict[str, TelegramChat]] = {}
    for bot_id, chat_ids in tg_pairs.items():
        if not chat_ids:
            continue
        chat_rows = await session.exec(
            select(TelegramChat).where(
                TelegramChat.bot_id == bot_id,
                TelegramChat.chat_id.in_(chat_ids),
            )
        )
        telegram_chats_by_bot[bot_id] = {c.chat_id: c for c in chat_rows.all()}

    # Email + Matrix bots
    email_bot_ids: set[int] = set()
    matrix_bot_ids: set[int] = set()
    for tid in all_target_ids:
        tgt = target_map.get(tid) or child_target_map.get(tid)
        if not tgt:
            continue
        if tgt.type == "email":
            bid = tgt.config.get("email_bot_id")
            if bid:
                email_bot_ids.add(bid)
        elif tgt.type == "matrix":
            bid = tgt.config.get("matrix_bot_id")
            if bid:
                matrix_bot_ids.add(bid)

    email_bots_by_id: dict[int, EmailBot] = {}
    if email_bot_ids:
        rows = await session.exec(
            select(EmailBot).where(EmailBot.id.in_(email_bot_ids))
        )
        email_bots_by_id = {b.id: b for b in rows.all()}

    matrix_bots_by_id: dict[int, MatrixBot] = {}
    if matrix_bot_ids:
        rows = await session.exec(
            select(MatrixBot).where(MatrixBot.id.in_(matrix_bot_ids))
        )
        matrix_bots_by_id = {b.id: b for b in rows.all()}

    link_data: list[dict[str, Any]] = []
    for tt in active_links:
        target = target_map.get(tt.target_id)
        if not target:
            continue

        # Per-link config overrides tracker defaults
        tc_id = tt.tracking_config_id or default_tc_id
        tmpl_id = tt.template_config_id or default_tmpl_id
        tracking_config = tc_map.get(tc_id) if tc_id else None
        template_config = tmpl_map.get(tmpl_id) if tmpl_id else None
        template_slots = slots_by_config.get(template_config.id) if template_config else None

        # Broadcast target: expand into child targets (pre-loaded above)
        if target.type == "broadcast":
            disabled_ids = set(target.config.get("disabled_child_ids", []))
            for child_id in target.config.get("child_target_ids", []):
                if child_id in disabled_ids:
                    continue
                child_target = child_target_map.get(child_id)
                if not child_target or child_target.type == "broadcast":
                    continue
                resolved = await _resolve_target(
                    session, child_target,
                    receivers_by_target=receivers_by_target,
                    telegram_chats_by_bot=telegram_chats_by_bot,
                    email_bots_by_id=email_bots_by_id,
                    matrix_bots_by_id=matrix_bots_by_id,
                )
                link_data.append({
                    **resolved,
                    "link_id": tt.id,
                    "tracking_config": tracking_config,
                    "template_config": template_config,
                    "template_slots": template_slots,
                })
            continue

        # Regular target
        resolved = await _resolve_target(
            session, target,
            receivers_by_target=receivers_by_target,
            telegram_chats_by_bot=telegram_chats_by_bot,
            email_bots_by_id=email_bots_by_id,
            matrix_bots_by_id=matrix_bots_by_id,
        )
        link_data.append({
            **resolved,
            "link_id": tt.id,
            "tracking_config": tracking_config,
            "template_config": template_config,
            "template_slots": template_slots,
        })

    return link_data
