Add Telegram channel alerts for gateway errors

gateway/app.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
+import asyncio
 import logging
 import os
 from pathlib import Path
+from datetime import datetime, timezone
 from typing import Any
 
 import httpx
@@ -14,6 +16,135 @@ from services.shared.models import HealthResponse
 
 app = FastAPI(title="api-gateway", version="1.0.0")
 logger = logging.getLogger("api-gateway.proxy")
+_ERROR_ALERT_SENT_AT: dict[str, float] = {}
+
+
+def _env_flag(name: str, default: bool = False) -> bool:
+    value = os.getenv(name)
+    if value is None:
+        return default
+    return value.strip().lower() in {"1", "true", "yes", "on"}
+
+
+def _env_int(name: str, default: int) -> int:
+    value = os.getenv(name)
+    if value is None:
+        return default
+    try:
+        parsed = int(value)
+    except ValueError:
+        return default
+    return max(1, parsed)
+
+
+def _error_alert_enabled() -> bool:
+    return _env_flag("ERROR_TELEGRAM_ALERTS", False)
+
+
+def _error_alert_bot_token() -> str:
+    return os.getenv("ERROR_TELEGRAM_ALERT_BOT_TOKEN", "").strip()
+
+
+def _error_alert_chat_ids() -> list[str]:
+    raw = os.getenv("ERROR_TELEGRAM_ALERT_CHAT_IDS", "").strip()
+    if not raw:
+        raw = os.getenv("ERROR_TELEGRAM_ALERT_CHANNEL_ID", "").strip()
+        if not raw:
+            raw = os.getenv("ERROR_TELEGRAM_ALERT_CHANNEL", "").strip()
+    return [item.strip() for item in raw.split(",") if item.strip()]
+
+
+def _error_alert_min_status() -> int:
+    return max(1, _env_int("ERROR_TELEGRAM_ALERT_MIN_STATUS", 500))
+
+
+def _error_alert_cooldown_seconds() -> int:
+    return _env_int("ERROR_TELEGRAM_ALERT_COOLDOWN_SECONDS", 30)
+
+
+def _error_alert_ignored_paths() -> tuple[str, ...]:
+    raw = os.getenv("ERROR_TELEGRAM_ALERT_IGNORE_PATHS", "/health").strip()
+    paths = tuple(item.strip() for item in raw.split(",") if item.strip())
+    return paths if paths else ("/health",)
+
+
+def _error_alert_service_name() -> str:
+    return os.getenv("ERROR_TELEGRAM_ALERT_SERVICE", "api-gateway").strip() or "api-gateway"
+
+
+def _error_alert_bot_api_base() -> str:
+    return os.getenv("ERROR_TELEGRAM_ALERT_BOT_API_BASE", "https://api.telegram.org").rstrip("/")
+
+
+def _error_alert_should_alert_path(path: str) -> bool:
+    ignore = _error_alert_ignored_paths()
+    return not any(path == item or path.startswith(f"{item}/") for item in ignore)
+
+
+def _error_alert_key(service: str, status: int, path: str, method: str) -> str:
+    return f"{service}:{status}:{method}:{path}"
+
+
+def _error_alert_is_throttled(key: str) -> bool:
+    now = datetime.now(tz=timezone.utc).timestamp()
+    cooldown = _error_alert_cooldown_seconds()
+    previous = _ERROR_ALERT_SENT_AT.get(key, 0.0)
+    if now < previous:
+        return True
+    _ERROR_ALERT_SENT_AT[key] = now + cooldown
+    return False
+
+
+async def _send_telegram_alert(message: str) -> None:
+    chat_ids = _error_alert_chat_ids()
+    bot_token = _error_alert_bot_token()
+    if not bot_token or not chat_ids:
+        return
+
+    url = f"{_error_alert_bot_api_base()}/bot{bot_token}/sendMessage"
+    payload = {"text": message}
+    async with httpx.AsyncClient(timeout=3.0) as client:
+        for chat_id in chat_ids:
+            payload["chat_id"] = chat_id
+            try:
+                response = await client.post(url, json=payload)
+            except httpx.RequestError:
+                logger.exception("Unable to send Telegram error notification")
+                return
+            if response.status_code >= 300:
+                logger.warning(
+                    "Telegram error notification failed: status=%s body=%s",
+                    response.status_code,
+                    response.text,
+                )
+
+
+def _queue_telegram_error_alert(service: str, method: str, path: str, status: int, detail: str) -> None:
+    if not _error_alert_enabled():
+        return
+    if not _error_alert_should_alert_path(path):
+        return
+    if not _error_alert_bot_token() or not _error_alert_chat_ids():
+        return
+
+    if _error_alert_is_throttled(_error_alert_key(service, status, path, method)):
+        return
+
+    timestamp = datetime.now(tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+    message = (
+        f"🚨 [{service}] {status}\n"
+        f"method: {method}\n"
+        f"path: {path}\n"
+        f"time: {timestamp}\n"
+        f"detail: {detail}"
+    )
+
+    try:
+        loop = asyncio.get_running_loop()
+    except RuntimeError:
+        logger.warning("No running loop to queue telegram error notification")
+        return
+    loop.create_task(_send_telegram_alert(message))
 
 _LEGAL_HTML = {
     "privacy": """<!doctype html>
@@ -169,13 +300,6 @@ SERVICE_PATH_PREFIX_EXEMPTIONS = {
 }
 
 
-def _env_flag(name: str, default: bool = False) -> bool:
-    value = os.getenv(name)
-    if value is None:
-        return default
-    return value.strip().lower() in {"1", "true", "yes", "on"}
-
-
 def _resolve_service_path(service: str, path: str) -> str:
     normalized = path.lstrip("/")
     prefix = SERVICE_PATH_PREFIXES.get(service, "").strip("/")
@@ -188,6 +312,35 @@ def _resolve_service_path(service: str, path: str) -> str:
     return f"{prefix}/{normalized}" if normalized else prefix
 
 
+@app.middleware("http")
+async def error_notification_middleware(request: Request, call_next):
+    try:
+        response = await call_next(request)
+    except Exception as exc:
+        status_code = 500
+        if isinstance(exc, HTTPException):
+            status_code = exc.status_code
+        if status_code >= _error_alert_min_status():
+            _queue_telegram_error_alert(
+                service=_error_alert_service_name(),
+                method=request.method,
+                path=str(request.url.path),
+                status=status_code,
+                detail=f"{type(exc).__name__}: {exc}",
+            )
+        raise
+
+    if response.status_code >= _error_alert_min_status():
+        _queue_telegram_error_alert(
+            service=_error_alert_service_name(),
+            method=request.method,
+            path=str(request.url.path),
+            status=response.status_code,
+            detail=f"HTTP {response.status_code} from {request.method} {request.url.path}",
+        )
+    return response
+
+
 @app.get("/health", response_model=HealthResponse)
 def health() -> HealthResponse:
     return HealthResponse(status="ok", service="api-gateway")
@@ -414,6 +567,13 @@ async def _forward(method: str, service: str, path: str, request: Request) -> Re
                 headers=headers,
             )
     except httpx.RequestError:
+        _queue_telegram_error_alert(
+            service=_error_alert_service_name(),
+            method=method,
+            path=f"/proxy/{service}/{path}",
+            status=502,
+            detail=f"Upstream service unavailable: service={service} base_url={base}",
+        )
         logger.warning(
             "Upstream service unavailable: service=%s base_url=%s method=%s path=%s",
             service,