backend/app/api/v1/admin/alerts_webhook.py

"""V1 Admin alerts webhook — receives AlertManager notifications.

This is the HTTP bridge between Prometheus AlertManager and the RMI backend.
AlertManager POSTs JSON payloads here for two receivers:
  - /api/v1/admin/alerts/webhook       — default (all severities, from rmi-alerts receiver)
  - /api/v1/admin/alerts/critical      — critical only (from rmi-critical receiver)

Each handler:
  1. Parses the AlertManager v2 webhook payload
  2. Persists to Redis (sorted set, capped) for in-app display
  3. Logs structured record (JSON, single line)
  4. Returns 200 OK immediately so AlertManager doesn't retry

AlertManager webhook payload shape (Prometheus):
    {
      "version": "4",
      "groupKey": "<strings>",
      "status": "firing|resolved",
      "receiver": "rmi-alerts",
      "groupLabels": {"alertname": "...", ...},
      "commonLabels": {...},
      "commonAnnotations": {...},
      "externalURL": "...",
      "alerts": [
        {
          "status": "firing|resolved",
          "labels": {...},
          "annotations": {...},
          "startsAt": "RFC3339",
          "endsAt": "RFC3339",
          "generatorURL": "..."
        }
      ]
    }
"""
from __future__ import annotations

import json
import logging
import os
import time
from typing import Any

from fastapi import APIRouter, Request
from pydantic import BaseModel, Field

logger = logging.getLogger("rmi.admin.alerts_webhook")

# Cap the in-Redis alert history so it doesn't grow unbounded.
_REDIS_CAP = int(os.getenv("RMI_ALERTS_HISTORY_CAP", "500"))

router = APIRouter(prefix="/api/v1/admin/alerts", tags=["admin-alerts"])


class AlertmanagerAlert(BaseModel):
    status: str
    labels: dict[str, str] = Field(default_factory=dict)
    annotations: dict[str, str] = Field(default_factory=dict)
    startsAt: str | None = None
    endsAt: str | None = None
    generatorURL: str | None = None


class AlertmanagerPayload(BaseModel):
    version: str | None = None
    groupKey: str | None = None
    status: str
    receiver: str | None = None
    groupLabels: dict[str, str] = Field(default_factory=dict)
    commonLabels: dict[str, str] = Field(default_factory=dict)
    commonAnnotations: dict[str, str] = Field(default_factory=dict)
    externalURL: str | None = None
    alerts: list[AlertmanagerAlert] = Field(default_factory=list)


def _redis_url_from_env() -> str:
    """Build a Redis URL from discrete REDIS_HOST/PORT/DB/PASSWORD env vars.

    Falls back to REDIS_URL if set, otherwise localhost. Returns a URL with
    auth credentials embedded if REDIS_PASSWORD is set.
    """
    explicit = os.getenv("REDIS_URL")
    if explicit:
        return explicit
    host = os.getenv("REDIS_HOST", "localhost")
    port = os.getenv("REDIS_PORT", "6379")
    db = os.getenv("REDIS_DB", "0")
    password = os.getenv("REDIS_PASSWORD", "")
    if password:
        return f"redis://:{password}@{host}:{port}/{db}"
    return f"redis://{host}:{port}/{db}"


def _redis_client():
    """Lazy Redis import — avoids forcing a Redis dep at module load."""
    try:
        import redis.asyncio as redis_async  # type: ignore

        url = _redis_url_from_env()
        return redis_async.from_url(url, decode_responses=True)
    except Exception as exc:  # pragma: no cover - degraded mode
        logger.warning("redis_unavailable", extra={"err": str(exc)})
        return None


async def _persist_to_redis(payload: AlertmanagerPayload, severity: str) -> int:
    """Push payload to Redis sorted set capped at _REDIS_CAP entries.

    Returns number of alerts persisted (0 if Redis is down).
    """
    client = _redis_client()
    if client is None:
        return 0
    try:
        score = time.time()
        record = json.dumps(
            {
                "received_at": score,
                "severity_bucket": severity,
                "receiver": payload.receiver,
                "status": payload.status,
                "group_labels": payload.groupLabels,
                "common_labels": payload.commonLabels,
                "common_annotations": payload.commonAnnotations,
                "alerts": [a.model_dump() for a in payload.alerts],
            },
            default=str,
        )
        key = "rmi:alerts:webhook"
        async with client.pipeline(transaction=False) as pipe:
            pipe.zadd(key, {record: score})
            pipe.zremrangebyrank(key, 0, -(_REDIS_CAP + 1))
            pipe.expire(key, 7 * 24 * 3600)  # 7 days
            await pipe.execute()
        return len(payload.alerts)
    except Exception as exc:
        logger.warning("redis_persist_failed", extra={"err": str(exc)})
        return 0
    finally:
        try:
            await client.aclose()
        except Exception:
            pass


def _log_payload(payload: AlertmanagerPayload, severity: str, persisted: int) -> None:
    """Single-line JSON log so log aggregators can index cleanly."""
    record = {
        "ts": time.time(),
        "event": "alertmanager_webhook",
        "severity_bucket": severity,
        "receiver": payload.receiver,
        "status": payload.status,
        "alert_count": len(payload.alerts),
        "persisted": persisted,
        "alertname": payload.groupLabels.get("alertname"),
        "common_labels": payload.commonLabels,
    }
    logger.info(json.dumps(record, default=str))


@router.post("/webhook", status_code=200)
async def alerts_webhook(payload: AlertmanagerPayload, request: Request) -> dict[str, Any]:
    """Default receiver webhook — all severities."""
    severity = payload.commonLabels.get("severity", "unknown")
    persisted = await _persist_to_redis(payload, severity)
    _log_payload(payload, severity, persisted)
    return {
        "ok": True,
        "received": len(payload.alerts),
        "persisted": persisted,
        "severity": severity,
        "alertname": payload.groupLabels.get("alertname"),
    }


@router.post("/critical", status_code=200)
async def alerts_critical_webhook(
    payload: AlertmanagerPayload, request: Request
) -> dict[str, Any]:
    """Critical-only webhook — escalations only (severity=critical)."""
    # Defensive: if someone misroutes a non-critical here, log and accept
    # (we don't want to lose data). Severity bucket stays "critical" because
    # the receiver name implies it.
    severity = "critical"
    persisted = await _persist_to_redis(payload, severity)
    _log_payload(payload, severity, persisted)
    return {
        "ok": True,
        "received": len(payload.alerts),
        "persisted": persisted,
        "severity": severity,
        "alertname": payload.groupLabels.get("alertname"),
        "bucket": "critical",
    }


@router.get("/recent", status_code=200)
async def alerts_recent(limit: int = 50) -> dict[str, Any]:
    """Read recent alerts (debug endpoint — admin only in production)."""
    client = _redis_client()
    if client is None:
        return {"ok": False, "error": "redis_unavailable", "items": []}
    try:
        # Newest first
        raw = await client.zrevrange("rmi:alerts:webhook", 0, max(0, limit - 1))
        items = [json.loads(r) for r in raw]
        return {"ok": True, "count": len(items), "items": items}
    except Exception as exc:
        return {"ok": False, "error": str(exc), "items": []}
    finally:
        try:
            await client.aclose()
        except Exception:
            pass