Source code for constellation_utils.notifications

"""Slack notifications via ``chat.postMessage``.

Stdlib-only. One public function :func:`notify` plus :class:`SlackNotifyError`
for the Slack-API quirk: ``chat.postMessage`` returns HTTP 200 even for
API-level failures, with ``{"ok": false, "error": "..."}`` in the body —
we parse and raise.

Auth: bot user OAuth token (``xoxb-...``) retrieved via
:func:`constellation_utils.secrets.slack_bot_token`. Profile is required;
no default at this layer (matches the secrets-module convention).
"""

from __future__ import annotations

import json
import urllib.request
from typing import Literal

from constellation_utils import secrets

_SLACK_POST_MESSAGE = "https://slack.com/api/chat.postMessage"




class SlackNotifyError(RuntimeError):
    """Raised when ``chat.postMessage`` returns ``ok=false`` or a body
    that isn't valid UTF-8 JSON (e.g. an HTML error page from a WAF or
    transparent proxy in front of ``slack.com``).

    Non-2xx HTTP responses surface as :class:`urllib.error.HTTPError`
    unchanged — they're handled separately because they represent
    transport/auth problems rather than Slack API errors.
    """





def notify(
    message: str,
    *,
    level: Literal["info", "warning", "error"],
    channel: str,
    profile: str,
) -> None:
    """POST ``f"[{level}] {message}"`` to *channel* via Slack ``chat.postMessage``.

    Blocks the caller for up to 10s on the HTTP POST. No retries and no
    exponential backoff — a 5xx surfaces as ``urllib.error.HTTPError``;
    a Slack ``ok=false`` response surfaces as :class:`SlackNotifyError`.

    *level*, *channel*, and *profile* are keyword-only with no default —
    matches the secrets-module convention. Per-package channel defaults
    (e.g. Virgo → ``#virgo-status``) are the caller's responsibility.
    """
    token = secrets.slack_bot_token(profile=profile)
    body = json.dumps({"channel": channel, "text": f"[{level}] {message}"}).encode("utf-8")
    req = urllib.request.Request(
        _SLACK_POST_MESSAGE,
        data=body,
        headers={
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json; charset=utf-8",
        },
        method="POST",
    )
    with urllib.request.urlopen(req, timeout=10) as resp:
        raw = resp.read()
    try:
        payload = json.loads(raw.decode("utf-8"))
    except (json.JSONDecodeError, UnicodeDecodeError) as exc:
        # HTTP 200 with a non-JSON body — typically an HTML error page from a
        # WAF or transparent proxy in front of slack.com. Surface as the
        # documented SlackNotifyError so callers don't see a raw JSONDecodeError.
        snippet = raw[:120].decode("utf-8", errors="replace")
        raise SlackNotifyError(
            f"chat.postMessage returned a non-JSON body (channel={channel!r}): {snippet!r}"
        ) from exc
    if not payload.get("ok"):
        raise SlackNotifyError(
            f"chat.postMessage failed: {payload.get('error', 'unknown error')!r} "
            f"(channel={channel!r})"
        )