xmrcheckout/api/app/btcpay_webhooks.py

from __future__ import annotations

import hashlib
import hmac
import json
import logging
import time
from datetime import datetime, timezone
from urllib.parse import urljoin

import requests
from requests import RequestException
from sqlalchemy.orm import Session

from .models import BtcpayWebhook, Invoice
from .security import decrypt_secret

logger = logging.getLogger(__name__)

_REDIRECT_STATUSES = {301, 302, 303, 307, 308}


def dispatch_btcpay_webhooks(
    db: Session,
    user_id: str,
    event_type: str,
    invoice: Invoice,
    *,
    manually_marked: bool = False,
) -> None:
    hooks = (
        db.query(BtcpayWebhook)
        .filter(
            BtcpayWebhook.user_id == user_id,
            BtcpayWebhook.enabled.is_(True),
        )
        .all()
    )
    if not hooks:
        return
    payload = _build_payload(
        event_type=event_type,
        user_id=user_id,
        invoice=invoice,
        manually_marked=manually_marked,
    )
    body = json.dumps(payload, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
    for hook in hooks:
        if not _event_allowed(hook.authorized_events, event_type):
            continue
        try:
            secret = decrypt_secret(hook.secret_encrypted)
            signature = _sign_payload(body, secret)
            headers = {
                "BTCPay-Sig": f"sha256={signature}",
                "Content-Type": "application/json",
                "User-Agent": "xmrcheckout-btcpay-webhook/1.0",
            }
            # Retry up to 3 times with backoff on transient failures
            last_response = None
            for attempt in range(3):
                try:
                    last_response = _post_with_redirects(
                        hook.url,
                        data=body,
                        headers=headers,
                        timeout=10,
                    )
                    if last_response is not None and last_response.status_code < 500:
                        break
                except RequestException:
                    if attempt == 2:
                        raise
                if attempt < 2:
                    time.sleep(1.0 * (attempt + 1))
            if last_response is None:
                logger.warning(
                    "BTCPay webhook delivery failed after retries (redirect loop)",
                    extra={"webhook_id": str(hook.id), "event": event_type, "url": hook.url},
                )
                continue
            if last_response.status_code >= 400:
                logger.warning(
                    "BTCPay webhook delivered non-success status",
                    extra={
                        "webhook_id": str(hook.id),
                        "event": event_type,
                        "http_status": last_response.status_code,
                    },
                )
        except RequestException as exc:
            logger.error(
                "BTCPay webhook delivery failed after 3 attempts: %s",
                exc,
                extra={"webhook_id": str(hook.id), "event": event_type, "url": hook.url},
            )
        except Exception as exc:
            logger.error(
                "BTCPay webhook dispatch failed: %s",
                exc,
                extra={"webhook_id": str(hook.id), "event": event_type},
            )


def _event_allowed(authorized_events: object, event_type: str) -> bool:
    if not isinstance(authorized_events, dict):
        return False
    if authorized_events.get("everything") is True:
        return True
    specific = authorized_events.get("specificEvents") or []
    return event_type in specific


def _build_payload(
    *,
    event_type: str,
    user_id: str,
    invoice: Invoice,
    manually_marked: bool,
) -> dict[str, object]:
    metadata = invoice.metadata_json or {}
    quote = metadata.get("quote") if isinstance(metadata, dict) else None
    fiat_amount = quote.get("fiat_amount") if isinstance(quote, dict) else None
    fiat_currency = quote.get("fiat_currency") if isinstance(quote, dict) else None
    return {
        "type": event_type,
        "timestamp": int(time.time()),
        "storeId": user_id,
        "invoiceId": str(invoice.id),
        "manuallyMarked": manually_marked,
        "overPaid": False,
        "partiallyPaid": False,
        "afterExpiration": _after_expiration(invoice),
        "metadata": metadata,
        # Fiat amount/currency so downstream processors (e.g. Medusa) can
        # capture at the right number. Omitting this defaults Medusa's capture
        # to 0 which leaves captured_at NULL and payment_collection stuck at
        # authorized forever.
        "amount": str(fiat_amount) if fiat_amount is not None else None,
        "currency": fiat_currency,
        # Crypto amount is also useful for reconciliation in some downstreams.
        "amount_crypto": (
            str(invoice.amount_xmr)
            if getattr(invoice, "amount_xmr", None) is not None
            else None
        ),
    }


def _after_expiration(invoice: Invoice) -> bool:
    expires_at = invoice.expires_at
    detected_at = invoice.detected_at
    if expires_at is None or detected_at is None:
        return False
    if expires_at.tzinfo is None:
        expires_at = expires_at.replace(tzinfo=timezone.utc)
    if detected_at.tzinfo is None:
        detected_at = detected_at.replace(tzinfo=timezone.utc)
    return detected_at > expires_at


def _sign_payload(body: bytes, secret: str) -> str:
    return hmac.new(secret.encode("utf-8"), body, hashlib.sha256).hexdigest()


def _post_with_redirects(
    url: str,
    *,
    data: bytes,
    headers: dict[str, str],
    timeout: int,
    max_redirects: int = 3,
) -> requests.Response | None:
    current_url = url
    for _ in range(max_redirects + 1):
        response = requests.post(
            current_url,
            data=data,
            headers=headers,
            timeout=timeout,
            allow_redirects=False,
        )
        if response.status_code not in _REDIRECT_STATUSES:
            return response
        location = response.headers.get("Location")
        if not location:
            return response
        current_url = urljoin(current_url, location)
    return None