channel-exit/app/reconcile.py

"""Asking the carrier what actually happened.

A provider's 2xx means it accepted the message. It does not mean anyone
received it. Twilio answers `201 Created` and then, seconds later, the carrier
may refuse the message outright — and nothing in an estate that only watches
HTTP status codes will ever notice.

That is not hypothetical here. On 2026-08-19 the ledger showed **48 consecutive
messages to one number, every one undelivered, going back to January**,
including a daily send for seven weeks. Each was recorded upstream as a
success, and each was billed.

Two decisions worth keeping:

**We poll; we are not called back.** The obvious design is Twilio's
`StatusCallback` webhook, but this relay binds to loopback on purpose — a
publicly reachable relay is a spam relay — so no carrier can reach it. Polling
costs one API call per run and keeps that property.

**We reconcile against the provider's ledger, not our own record.** Our record
is exactly the thing that was wrong, and it only knows about messages we sent.
Twilio's ledger knows about every message on the account, including the ones a
different service sent — which is how those 48 would have been caught even
though nothing here sent them.
"""

from __future__ import annotations

import base64
import json
import urllib.parse
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Any, Callable

from app.providers import Post, Response, post as default_post

LIST_URL = "https://api.twilio.com/2010-04-01/Accounts/{sid}/Messages.json"

# Twilio's terminal states. `sent` means it left Twilio and the carrier has not
# reported back — for a US handset that is usually followed by `delivered` or
# `undelivered`, so it is treated as "still in flight" rather than as success.
DELIVERED = frozenset({"delivered", "received"})
FAILED = frozenset({"undelivered", "failed"})
IN_FLIGHT = frozenset({"queued", "accepted", "sending", "sent", "scheduled"})


@dataclass(frozen=True)
class Message:
    sid: str
    to: str
    status: str
    error_code: str | None
    sent_at: str | None

    @property
    def failed(self) -> bool:
        return self.status in FAILED


@dataclass(frozen=True)
class Report:
    delivered: int
    failed: list[Message]
    in_flight: int
    unknown: list[str]

    @property
    def ok(self) -> bool:
        return not self.failed

    def summary(self) -> str:
        if not (self.delivered or self.failed or self.in_flight):
            return "no messages in the window"
        parts = [f"{self.delivered} delivered", f"{len(self.failed)} failed", f"{self.in_flight} in flight"]
        return ", ".join(parts)


def _mask(number: str) -> str:
    """Enough to recognise a number, not enough to publish one."""
    return f"***{number[-4:]}" if number and len(number) > 4 else "***"


def fetch(
    *,
    account_sid: str,
    auth_token: str,
    since: datetime,
    timeout: float = 15.0,
    page_size: int = 200,
    post: Post | None = None,
    get: Callable[..., Response] | None = None,
) -> list[Message]:
    """Every message Twilio has recorded since `since`, whoever sent it."""
    fetcher = get or _get
    query = urllib.parse.urlencode({
        # Twilio filters on the date a message was sent, at day granularity for
        # the inclusive form; asking from the start of the day is deliberate,
        # since a run just after midnight must still see last night's failures.
        "DateSent>": since.strftime("%Y-%m-%d"),
        "PageSize": str(page_size),
    })
    reply = fetcher(
        LIST_URL.format(sid=urllib.parse.quote(account_sid, safe="")) + "?" + query,
        headers={"Authorization": "Basic " + base64.b64encode(
            f"{account_sid}:{auth_token}".encode()).decode()},
        timeout=timeout,
    )
    if reply.status >= 400:
        raise RuntimeError(f"twilio refused the ledger query with {reply.status}")

    payload: dict[str, Any] = json.loads(reply.body or "{}")
    out: list[Message] = []
    for row in payload.get("messages") or []:
        sent = row.get("date_sent") or row.get("date_created")
        out.append(Message(
            sid=str(row.get("sid") or ""),
            to=str(row.get("to") or ""),
            status=str(row.get("status") or "").lower(),
            error_code=str(row["error_code"]) if row.get("error_code") else None,
            sent_at=str(sent) if sent else None,
        ))
    return out


def reconcile(messages: list[Message]) -> Report:
    delivered = sum(1 for m in messages if m.status in DELIVERED)
    failed = [m for m in messages if m.failed]
    in_flight = sum(1 for m in messages if m.status in IN_FLIGHT)
    unknown = sorted({m.status for m in messages
                      if m.status and m.status not in DELIVERED | FAILED | IN_FLIGHT})
    return Report(delivered=delivered, failed=failed, in_flight=in_flight, unknown=unknown)


def describe(report: Report) -> str:
    """A message an operator can act on, with no numbers in full."""
    lines = [f"SMS reconciliation: {report.summary()}."]
    if report.failed:
        lines.append("")
        lines.append("These were accepted by Twilio and then not delivered:")
        by_reason: dict[str, list[Message]] = {}
        for m in report.failed:
            by_reason.setdefault(m.error_code or "no error code", []).append(m)
        for reason, group in sorted(by_reason.items(), key=lambda kv: -len(kv[1])):
            recipients = ", ".join(sorted({_mask(m.to) for m in group})[:6])
            lines.append(f"  {len(group):>3}x error {reason} -> {recipients}")
        lines.append("")
        lines.append("Error 30032 means the sending number is an unverified toll-free "
                     "number; the carrier blocks it and no retry will help.")
    if report.unknown:
        lines.append(f"Statuses this checker does not classify: {', '.join(report.unknown)}")
    return "\n".join(lines)


def _get(url: str, *, headers: dict[str, str], timeout: float) -> Response:
    import urllib.error
    import urllib.request

    request = urllib.request.Request(url, method="GET", headers={
        **headers, "User-Agent": "channel-exit-reconcile/1.0"})
    try:
        with urllib.request.urlopen(request, timeout=timeout) as reply:
            return Response(status=reply.status, body=reply.read().decode(errors="replace"),
                            headers=dict(reply.headers))
    except urllib.error.HTTPError as error:
        return Response(status=error.code, body=error.read().decode(errors="replace"),
                        headers=dict(error.headers or {}))


def window_start(hours: int) -> datetime:
    return datetime.now(timezone.utc) - timedelta(hours=hours)


# --------------------------------------------------------------- email

SUPPRESSION_URL = "https://api.sendgrid.com/v3/suppression/bounces"

# The scope a key needs before any of this can be answered. Named here so the
# daily report can say exactly what to change rather than "not available".
EMAIL_READ_SCOPE = "suppression.read"


@dataclass(frozen=True)
class EmailReport:
    checked: bool
    bounces: list[dict[str, Any]]
    reason: str = ""

    def summary(self) -> str:
        if not self.checked:
            return f"email delivery NOT VERIFIED — {self.reason}"
        if not self.bounces:
            return "email: no bounces in the window"
        return f"email: {len(self.bounces)} bounced"


def email_bounces(
    *,
    api_key: str,
    since: datetime,
    timeout: float = 15.0,
    get: Callable[..., Response] | None = None,
) -> EmailReport:
    """Bounces SendGrid recorded, if this key is allowed to ask.

    The estate's key is send-only, so this normally reports that it could not
    check. That is the point: an unanswerable question should appear in the
    daily report as unanswered, not be quietly skipped. A silent skip is how
    "we have monitoring" turns into seven months of unnoticed failures.
    """
    if not api_key:
        return EmailReport(checked=False, bounces=[], reason="no SendGrid key configured")

    fetcher = get or _get
    reply = fetcher(
        f"{SUPPRESSION_URL}?start_time={int(since.timestamp())}",
        headers={"Authorization": f"Bearer {api_key}"},
        timeout=timeout,
    )
    if reply.status in (401, 403):
        return EmailReport(
            checked=False, bounces=[],
            reason=(f"the SendGrid key cannot read bounces (needs {EMAIL_READ_SCOPE}); "
                    "it is send-only, so email failures are invisible here"))
    if reply.status >= 400:
        return EmailReport(checked=False, bounces=[], reason=f"SendGrid answered {reply.status}")

    try:
        rows = json.loads(reply.body or "[]")
    except json.JSONDecodeError:
        return EmailReport(checked=False, bounces=[], reason="SendGrid did not return JSON")
    return EmailReport(checked=True, bounces=rows if isinstance(rows, list) else [])


def describe_email(report: EmailReport) -> str:
    lines = [report.summary()]
    for row in report.bounces[:10]:
        address = str(row.get("email") or "")
        masked = ("***@" + address.split("@")[-1]) if "@" in address else "***"
        lines.append(f"  {masked}: {str(row.get('reason') or '')[:90]}")
    return "\n".join(lines)