fix(telegram): unify send routine across notifications and commands

- Route cache_key values that look like asset UUIDs through asset_cache
  in TelegramClient._get_cache_and_key. Single-asset sends previously
  stored file_ids in url_cache while the media-group path stored them
  in asset_cache, so repeat sends never hit.
- Extract build_asset_media_urls so the notification dispatcher
  (asset_to_media) and the bot command handlers (common._format_assets)
  share one rule for /video/playback vs thumbnail URLs.
- Add services/telegram_send.py as the single factory for constructing
  a TelegramClient. It always wires the shared aiohttp session and both
  file caches, so commands now reuse file_ids populated by notification
  dispatches (and vice versa) instead of re-uploading the same bytes.
- send_reply / send_media_group in commands/handler.py now delegate to
  the factory rather than constructing their own uncached clients.

packages/core/src/notify_bridge_core/notifications/telegram/client.py

@@ -89,6 +89,18 @@ class TelegramClient:
         self, url: str | None, cache_key: str | None = None,
     ) -> tuple[TelegramFileCache | None, str | None, str | None]:
         if cache_key:
+            # Route asset-UUID cache keys to the asset cache so single-item
+            # sends hit the same cache the media-group path uses. Without
+            # this, a command returning one photo stored file_ids in the
+            # URL cache and a command returning multiple stored them in
+            # the asset cache — repeated sends never hit.
+            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):

packages/core/src/notify_bridge_core/providers/immich/asset_utils.py

@@ -193,6 +193,27 @@ def get_asset_video_url(
     return None
 
 
+def build_asset_media_urls(
+    external_url: str, asset_id: str, asset_type: str,
+) -> tuple[str, str]:
+    """Return ``(preview_url, full_url)`` for an Immich asset.
+
+    Single source of truth for the photo-vs-video endpoint rule. Used by
+    ``asset_to_media`` (notification path) and the bot command handlers
+    (command path) so both always pick the transcoded ``/video/playback``
+    for videos and the preview-sized thumbnail for photos — if they
+    diverge, Telegram ends up delivering a still JPEG for videos in a
+    media group.
+    """
+    is_video = asset_type == ASSET_TYPE_VIDEO
+    if is_video:
+        preview_url = f"{external_url}/api/assets/{asset_id}/video/playback"
+    else:
+        preview_url = f"{external_url}/api/assets/{asset_id}/thumbnail?size=preview"
+    full_url = f"{external_url}/api/assets/{asset_id}/original"
+    return preview_url, full_url
+
+
 def build_asset_detail(
     asset: ImmichAssetInfo,
     external_url: str,
@@ -246,12 +267,7 @@ def asset_to_media(asset: ImmichAssetInfo, external_url: str) -> MediaAsset:
     # preview_url is what the notification dispatcher feeds to Telegram as the
     # actual media bytes — for videos it must be the transcoded playback (mp4),
     # not the JPEG thumbnail, or Telegram receives a JPEG labeled as video/mp4.
-    if asset.type == ASSET_TYPE_VIDEO:
-        preview_url = f"{external_url}/api/assets/{asset.id}/video/playback"
-        full_url = f"{external_url}/api/assets/{asset.id}/original"
-    else:
-        preview_url = f"{external_url}/api/assets/{asset.id}/thumbnail?size=preview"
-        full_url = f"{external_url}/api/assets/{asset.id}/original"
+    preview_url, full_url = build_asset_media_urls(external_url, asset.id, asset.type)
 
     return MediaAsset(
         id=asset.id,

packages/server/src/notify_bridge_server/commands/handler.py

@@ -367,20 +367,23 @@ async def send_reply(
     bot_token: str, chat_id: str, text: str, reply_to_message_id: int | None = None,
     session: aiohttp.ClientSession | None = None,
 ) -> None:
-    """Send a text reply via TelegramClient.
+    """Send a text reply to a chat.
 
-    Command responses are listings (albums, people, events, ...) that embed
-    multiple links; Telegram's default behavior of rendering a preview of
-    the first URL is almost never what the user wants and clashes with the
-    "Disable link previews" toggle operators set on their Telegram target.
-    We always pass ``disable_web_page_preview=True`` here.
+    Thin wrapper that goes through the single ``services.telegram_send``
+    entry point so commands and notifications share one routine — same
+    HTTP session pool, same file_id caches.
+
+    Command responses are listings (albums, people, events, ...) that
+    embed multiple links; Telegram's default behavior of rendering a
+    preview of the first URL is almost never what the user wants and
+    clashes with the "Disable link previews" toggle operators set on
+    their Telegram target. We always pass
+    ``disable_web_page_preview=True`` here.
     """
-    if session is None:
-        from ..services.http_session import get_http_session
-        session = await get_http_session()
-    client = TelegramClient(session, bot_token)
-    result = await client.send_message(
-        chat_id, text,
+    from ..services.telegram_send import send_telegram_message
+
+    result = await send_telegram_message(
+        bot_token, chat_id, text,
         reply_to_message_id=reply_to_message_id,
         disable_web_page_preview=True,
     )
@@ -393,38 +396,28 @@ async def send_media_group(
     reply_to_message_id: int | None = None,
     session: aiohttp.ClientSession | None = None,
 ) -> None:
-    """Send media items via TelegramClient.send_notification.
+    """Send media items via the shared Telegram routine.
 
     ``media_items`` must already be in TelegramClient asset format — each
     entry contains ``type`` (``"photo"``/``"video"``/``"document"``),
     ``url``, optional ``cache_key``, and optional ``headers``. Provider
-    command handlers build this format directly (via
-    ``build_telegram_asset_entry``) so videos keep their ``"video"`` type
-    and point at a real video URL instead of a still thumbnail.
+    command handlers build this format via
+    ``build_telegram_asset_entry`` — the same helper the notification
+    dispatcher uses — so videos keep their ``"video"`` type and point at
+    a real video URL instead of a still thumbnail.
 
-    Reuses the same Telegram file_id caches as the notification dispatcher
-    so repeated ``/latest`` / ``/random`` commands don't re-upload bytes
-    for assets Telegram has already seen. If the cache hasn't been
-    initialized (no data dir configured) we fall through to a plain
-    upload — identical behavior to the notification path.
+    Uses ``services.telegram_send.send_telegram_media`` so the URL cache
+    and asset cache are wired in exactly like the notification path.
+    Repeated ``/latest`` / ``/random`` commands that match previously-sent
+    assets hit the cache and skip the re-upload.
     """
     if not media_items:
         return
 
-    if session is None:
-        from ..services.http_session import get_http_session
-        session = await get_http_session()
+    from ..services.telegram_send import send_telegram_media
 
-    from ..services.watcher import _get_telegram_caches
-    url_cache, asset_cache = await _get_telegram_caches()
-
-    client = TelegramClient(
-        session, bot_token,
-        url_cache=url_cache,
-        asset_cache=asset_cache,
-    )
-    result = await client.send_notification(
-        chat_id, assets=media_items,
+    result = await send_telegram_media(
+        bot_token, chat_id, media_items,
         reply_to_message_id=reply_to_message_id,
         chat_action=None,
     )

packages/server/src/notify_bridge_server/commands/immich/common.py

@@ -7,7 +7,10 @@ import logging
 from typing import Any
 
 from notify_bridge_core.notifications.telegram.media import build_telegram_asset_entry
-from notify_bridge_core.providers.immich.asset_utils import get_public_url
+from notify_bridge_core.providers.immich.asset_utils import (
+    build_asset_media_urls,
+    get_public_url,
+)
 
 from ..handler import _render_cmd_template
 
@@ -127,21 +130,19 @@ def _format_assets(
     })
 
     if response_mode == "media":
-        # Reuse the same entry-building helper as the notification dispatcher
-        # so videos keep their "video" type and point at /video/playback —
-        # typing them as "photo" made Telegram render the still poster
-        # thumbnail in media groups instead of the real clip.
+        # Reuse the same URL rule (build_asset_media_urls) and entry builder
+        # (build_telegram_asset_entry) as the notification dispatcher so both
+        # paths agree on video → /video/playback and photo → thumbnail. When
+        # these diverged, Telegram rendered a still JPEG for each video in
+        # the media group instead of the real clip.
         media_items: list[dict[str, Any]] = []
         for asset in assets:
             asset_id = asset.get("id", "")
-            is_video = (asset.get("type") or "").upper() == "VIDEO"
-            if is_video:
-                url = f"{client.url}/api/assets/{asset_id}/video/playback"
-            else:
-                url = f"{client.url}/api/assets/{asset_id}/thumbnail?size=preview"
+            asset_type = (asset.get("type") or "").upper()
+            preview_url, _ = build_asset_media_urls(client.url, asset_id, asset_type)
             entry = build_telegram_asset_entry(
-                url=url,
-                media_type="video" if is_video else "image",
+                url=preview_url,
+                media_type="video" if asset_type == "VIDEO" else "image",
                 api_key=client.api_key,
                 internal_url=client.url,
                 cache_key=asset_id,

packages/server/src/notify_bridge_server/services/telegram_send.py

@@ -0,0 +1,119 @@
+"""Single entry point for all Telegram send operations.
+
+Both the notification dispatcher (event-driven) and the bot command
+handlers (user-driven) funnel their Telegram API calls through this
+module. Keeping construction in one place means:
+
+* The shared aiohttp session is always reused (one TCP pool for the
+  whole process).
+* The Telegram file_id caches (URL cache + asset cache) are always
+  wired in, so repeated sends — whether from a scheduled tracker or
+  a ``/latest`` command — reuse cached file_ids instead of re-uploading
+  the same bytes.
+* Future cross-cutting concerns (rate limiting, telemetry, retries)
+  have exactly one place to live.
+
+The actual Telegram API routine is still ``TelegramClient`` in core —
+this module just guarantees every caller gets a properly-wired client.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Callable
+
+import aiohttp
+
+from notify_bridge_core.notifications.telegram.client import (
+    NotificationResult,
+    TelegramClient,
+)
+
+from .http_session import get_http_session
+from .watcher import _get_telegram_caches
+
+
+async def get_telegram_client(
+    bot_token: str,
+    *,
+    session: aiohttp.ClientSession | None = None,
+    thumbhash_resolver: Callable[[str], str | None] | None = None,
+) -> TelegramClient:
+    """Return a ``TelegramClient`` wired to shared session + shared caches.
+
+    Every Telegram send in the process should acquire its client from
+    here — constructing ``TelegramClient`` directly skips the caches and
+    silently halves cache hit rate.
+
+    Args:
+        bot_token: The bot's API token.
+        session: Optional explicit aiohttp session. Defaults to the
+            process-wide shared session.
+        thumbhash_resolver: Optional asset-id → thumbhash lookup. The
+            notification dispatcher passes one so asset-cache entries
+            invalidate on visual change; the command path doesn't need it
+            (commands always ask for a fresh result).
+    """
+    if session is None:
+        session = await get_http_session()
+    url_cache, asset_cache = await _get_telegram_caches()
+    return TelegramClient(
+        session, bot_token,
+        url_cache=url_cache,
+        asset_cache=asset_cache,
+        thumbhash_resolver=thumbhash_resolver,
+    )
+
+
+async def send_telegram_message(
+    bot_token: str,
+    chat_id: str,
+    text: str,
+    *,
+    reply_to_message_id: int | None = None,
+    disable_web_page_preview: bool = True,
+    parse_mode: str = "HTML",
+) -> NotificationResult:
+    """Send a plain-text Telegram message with caches wired in."""
+    client = await get_telegram_client(bot_token)
+    return await client.send_message(
+        chat_id, text,
+        reply_to_message_id=reply_to_message_id,
+        disable_web_page_preview=disable_web_page_preview,
+        parse_mode=parse_mode,
+    )
+
+
+async def send_telegram_media(
+    bot_token: str,
+    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,
+    max_asset_data_size: int | None = None,
+    send_large_photos_as_documents: bool = False,
+    chat_action: str | None = "typing",
+    thumbhash_resolver: Callable[[str], str | None] | None = None,
+) -> NotificationResult:
+    """Send a Telegram media group (or single asset) with caches wired in.
+
+    ``assets`` must be in ``TelegramClient`` format — see
+    ``notify_bridge_core.notifications.telegram.media.build_telegram_asset_entry``
+    for the canonical builder.
+    """
+    client = await get_telegram_client(
+        bot_token, thumbhash_resolver=thumbhash_resolver,
+    )
+    return await client.send_notification(
+        chat_id,
+        assets=assets,
+        caption=caption,
+        reply_to_message_id=reply_to_message_id,
+        max_group_size=max_group_size,
+        chunk_delay=chunk_delay,
+        max_asset_data_size=max_asset_data_size,
+        send_large_photos_as_documents=send_large_photos_as_documents,
+        chat_action=chat_action,
+    )