datatools-dev/server/app/adapters/gumroad.py

"""Gumroad adapter.

Receives "Ping" notifications from Gumroad — form-encoded POSTs sent
when a sale occurs. Gumroad's Ping URL is configured in the seller
dashboard (Settings → Advanced → Ping URL).

Authentication
--------------

Gumroad does not HMAC-sign the body. Their recommended pattern is
to put a secret in the URL itself::

    https://licenses.datatools.unalogix.com/webhooks/gumroad?secret=...

The webhook receiver pulls the secret from the query string and
:meth:`GumroadAdapter.verify_webhook` constant-time-compares it
against the configured value. If they don't match, the request is
dropped with 404 (so a probing attacker can't tell whether the
endpoint exists, much less that it's the wrong secret).

The "test" field
----------------

Gumroad sends ``test=true`` on test pings fired from the dashboard.
We treat test pings as real sales (they create licenses just like
production sales), but tag them with ``notes='gumroad test ping'``
so the operator can filter / delete them later. Refusing test pings
would block the standard "Send Test Ping" verification flow.

Refunds, disputes, cancellations
--------------------------------

Stubbed for now (``parse_refund`` returns None). Gumroad doesn't
include refund signals in the standard sale Ping — refunds arrive
via the separate "Resource subscriptions" mechanism. Wiring that
in is PR 2.1; until then, refunds are handled by the operator
running ``datatools-admin revoke``.
"""

from __future__ import annotations

import hmac
from decimal import Decimal
from typing import Any, Optional

from app.adapters.base import RefundEvent, SaleEvent
from app.products import lookup as product_lookup


class GumroadAdapter:
    source_name = "gumroad"

    def __init__(self, secret: Optional[str]) -> None:
        self._secret = secret

    # --- Auth ----------------------------------------------------------------

    def verify_webhook(self, *, body: bytes, headers: dict[str, str]) -> bool:
        """Not used — Gumroad authentication is via URL query param,
        which only the route handler has direct access to. Call
        :meth:`verify_secret` instead."""
        return False

    def verify_secret(self, presented: Optional[str]) -> bool:
        """Constant-time compare against the configured secret.

        Returns False (not an exception) so the route handler can
        decide the response code — we return 404 to avoid signaling
        endpoint existence to an unauthenticated prober.
        """
        if not self._secret or not presented:
            return False
        return hmac.compare_digest(presented, self._secret)

    # --- Parsing -------------------------------------------------------------

    def parse_sale(self, payload: dict[str, Any]) -> Optional[SaleEvent]:
        """Parse a Gumroad Ping form-encoded payload into a SaleEvent.

        Returns None if the payload isn't a sale (e.g. some future
        event type we don't yet handle). Returns None *with no row
        side-effect* if the product_id is unmapped — the caller
        should treat that as an error and record it in the audit
        row, not silently drop.
        """
        # Sale pings always include sale_id (the order ID) and email.
        sale_id = payload.get("sale_id")
        email = payload.get("email")
        product_id = (
            payload.get("product_id")
            or payload.get("product_permalink")
            or payload.get("permalink")
        )
        if not (sale_id and email and product_id):
            return None

        mapping = product_lookup(self.source_name, str(product_id))
        if mapping is None:
            # Unmapped — surface to caller as a SaleEvent with no tier.
            # We deliberately don't raise here so the caller can
            # log it to gumroad_events with error info and still
            # return 200 (no Gumroad retry storm).
            raise UnmappedProductError(
                f"Gumroad product_id {product_id!r} has no entry in "
                "config/products.yaml. Add a mapping and replay this "
                f"sale (sale_id={sale_id})."
            )

        name = (payload.get("full_name") or "").strip() or _email_local(email)

        price_cents = _to_int(payload.get("price"))
        amount_paid = Decimal(price_cents) / Decimal(100) if price_cents is not None else None
        currency = (payload.get("currency") or "USD").upper()
        promotion = (payload.get("offer_code") or "").strip() or None

        notes = None
        if _is_truthy(payload.get("test")):
            notes = "gumroad test ping"

        return SaleEvent(
            source=self.source_name,
            source_order_id=str(sale_id),
            buyer_name=name,
            buyer_email=email.strip(),
            tier=mapping.tier,
            years=mapping.years,
            promotion=promotion,
            amount_paid=amount_paid,
            currency=currency,
            notes=notes,
            raw_payload=dict(payload),
        )

    def parse_refund(self, payload: dict[str, Any]) -> Optional[RefundEvent]:
        # PR 2.1.
        return None


class UnmappedProductError(ValueError):
    """Raised when a sale arrives for a product not in products.yaml.

    Caller catches and logs into ``gumroad_events.error`` so the
    operator can fix the mapping and replay.
    """


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def _email_local(email: str) -> str:
    """Fallback display name when ``full_name`` is missing — the part
    of the email before the ``@``, capitalized. Better than 'Unknown'
    for support tickets and the buyer's own delivery email."""
    local = email.split("@", 1)[0]
    return local.replace(".", " ").title()


def _to_int(v: Any) -> Optional[int]:
    if v is None or v == "":
        return None
    try:
        return int(v)
    except (TypeError, ValueError):
        return None


def _is_truthy(v: Any) -> bool:
    if isinstance(v, bool):
        return v
    if v is None:
        return False
    return str(v).strip().lower() in {"1", "true", "yes", "on"}