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/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,
     )