channel-exit/app/providers.py

"""Talking to SendGrid and Twilio.

Kept behind one tiny HTTP seam (`post`) for two reasons. It is the only thing
tests need to replace, so a test can prove the right thing was sent without a
network or a live credential — and nothing in this estate should be able to
send a real message because someone ran the suite.

Nothing here raises for a provider refusal. A refusal is data: the caller has
to log it and tell its consumer, and an exception carrying a provider payload
is exactly the kind of thing that ends up in a log with a key in it.
"""

from __future__ import annotations

import base64
import json
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass
from typing import Callable

SENDGRID_URL = "https://api.sendgrid.com/v3/mail/send"
TWILIO_URL = "https://api.twilio.com/2010-04-01/Accounts/{sid}/Messages.json"

# An SMS is billed per 160-character segment, so a caller with a runaway body
# is a bill, not just a long message. Truncating is the kinder failure: the
# person still gets the reminder, and nobody pays for a novel.
SMS_LIMIT = 1200


@dataclass(frozen=True)
class Response:
    status: int
    body: str
    headers: dict[str, str]


@dataclass(frozen=True)
class Outcome:
    """What happened, in terms safe to log."""

    ok: bool
    status: int
    detail: str
    message_id: str = ""


Post = Callable[..., Response]


def post(url: str, *, data: bytes, headers: dict[str, str], timeout: float) -> Response:
    request = urllib.request.Request(url, method="POST", data=data, headers=headers)
    try:
        with urllib.request.urlopen(request, timeout=timeout) as reply:
            return Response(reply.status, reply.read().decode("utf-8", "replace"), dict(reply.headers))
    except urllib.error.HTTPError as error:
        # A 4xx from a provider is an answer, not a crash: it carries the
        # reason the message was rejected, which is what we want to report.
        return Response(error.code, error.read().decode("utf-8", "replace"), dict(error.headers or {}))


def send_email(
    *,
    to: str,
    subject: str,
    body: str,
    api_key: str,
    sender: str,
    timeout: float,
    post: Post = post,
) -> Outcome:
    payload = {
        "personalizations": [{"to": [{"email": to}]}],
        "from": {"email": sender},
        # SendGrid rejects an empty subject outright. A reminder with no
        # subject is still worth delivering, so it gets a neutral one rather
        # than a 400 the caller cannot act on.
        "subject": subject or "Notification",
        "content": [{"type": "text/plain", "value": body}],
    }
    reply = post(
        SENDGRID_URL,
        data=json.dumps(payload).encode(),
        headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
        timeout=timeout,
    )
    ok = 200 <= reply.status < 300
    return Outcome(
        ok=ok,
        status=reply.status,
        detail="accepted" if ok else _reason(reply.body),
        message_id=reply.headers.get("X-Message-Id", ""),
    )


def send_sms(
    *,
    to: str,
    subject: str,
    body: str,
    account_sid: str,
    auth_token: str,
    sender: str,
    timeout: float,
    post: Post = post,
) -> Outcome:
    # One channel contract, two shapes of message: callers compose a subject
    # and a body because email has both. An SMS has neither, so they are joined
    # rather than one of them being silently dropped.
    text = f"{subject}\n\n{body}" if subject and body else (subject or body)
    text = text[:SMS_LIMIT]

    reply = post(
        TWILIO_URL.format(sid=urllib.parse.quote(account_sid, safe="")),
        data=urllib.parse.urlencode({"To": to, "From": sender, "Body": text}).encode(),
        headers={
            "Authorization": "Basic " + base64.b64encode(f"{account_sid}:{auth_token}".encode()).decode(),
            "Content-Type": "application/x-www-form-urlencoded",
        },
        timeout=timeout,
    )
    ok = 200 <= reply.status < 300
    message_id = ""
    if ok:
        try:
            message_id = json.loads(reply.body).get("sid", "") or ""
        except (ValueError, AttributeError):
            pass
    return Outcome(ok=ok, status=reply.status, detail="accepted" if ok else _reason(reply.body), message_id=message_id)


def _reason(body: str) -> str:
    """The provider's own words, trimmed — never the body we tried to send.

    Both providers answer a rejection in JSON, and both put something usable in
    `message`. Anything else is truncated rather than logged whole, because an
    error body is the one place a provider might echo back what we sent it.
    """
    try:
        parsed = json.loads(body)
    except ValueError:
        return body.strip()[:200]
    if isinstance(parsed, dict):
        if parsed.get("message"):
            code = parsed.get("code")
            return f"{parsed['message']}" + (f" ({code})" if code else "")
        errors = parsed.get("errors")
        if isinstance(errors, list) and errors:
            first = errors[0]
            if isinstance(first, dict) and first.get("message"):
                return str(first["message"])[:200]
    return str(parsed)[:200]