eryao/backend/src/v1/payments/creem_client.py

from __future__ import annotations

import hashlib
import hmac
import logging
from dataclasses import dataclass
from typing import Any

import httpx

from core.config.settings import config

logger = logging.getLogger(__name__)


@dataclass(frozen=True)
class CreemProduct:
    product_id: str
    name: str
    price_cents: int
    currency: str


@dataclass(frozen=True)
class CreemCheckout:
    checkout_id: str
    checkout_url: str


class CreemClient:
    def __init__(self) -> None:
        settings = config.creem
        self._api_key = settings.api_key.get_secret_value() if settings.api_key else None
        self._base_url = settings.base_url.rstrip("/")
        self._timeout = httpx.Timeout(30.0, connect=5.0)

    def _headers(self) -> dict[str, str]:
        if not self._api_key:
            raise RuntimeError("CREEM API key not configured")
        return {
            "x-api-key": self._api_key,
            "Content-Type": "application/json",
        }

    async def get_products(self) -> list[CreemProduct]:
        """Fetch all products from CREEM."""
        async with httpx.AsyncClient(timeout=self._timeout) as client:
            resp = await client.get(
                f"{self._base_url}/v1/products/search",
                headers=self._headers(),
            )
            resp.raise_for_status()
            data: Any = resp.json()

        products: list[CreemProduct] = []
        for item in data.get("items", []):
            product_id = item.get("id", "")
            name = item.get("name", "")
            price = item.get("price", 0)
            currency = item.get("currency", "USD")
            products.append(
                CreemProduct(
                    product_id=product_id,
                    name=name,
                    price_cents=int(price),
                    currency=currency,
                )
            )
        return products

    async def get_product(self, product_id: str) -> CreemProduct | None:
        """Fetch a single product by ID."""
        async with httpx.AsyncClient(timeout=self._timeout) as client:
            resp = await client.get(
                f"{self._base_url}/v1/products",
                params={"product_id": product_id},
                headers=self._headers(),
            )
            if resp.status_code == 404:
                return None
            resp.raise_for_status()
            data: Any = resp.json()

        return CreemProduct(
            product_id=data.get("id", ""),
            name=data.get("name", ""),
            price_cents=int(data.get("price", 0)),
            currency=data.get("currency", "USD"),
        )

    async def create_checkout(
        self,
        *,
        product_id: str,
        success_url: str,
        customer_email: str | None = None,
        metadata: dict[str, Any] | None = None,
    ) -> CreemCheckout:
        """Create a checkout session."""
        payload: dict[str, Any] = {
            "product_id": product_id,
            "success_url": success_url,
        }
        if customer_email:
            payload["customer"] = {"email": customer_email}
        if metadata:
            payload["metadata"] = metadata

        async with httpx.AsyncClient(timeout=self._timeout) as client:
            resp = await client.post(
                f"{self._base_url}/v1/checkouts",
                headers=self._headers(),
                json=payload,
            )
            resp.raise_for_status()
            data: Any = resp.json()

        return CreemCheckout(
            checkout_id=data.get("id", ""),
            checkout_url=data.get("checkout_url", ""),
        )

    @staticmethod
    def verify_webhook_signature(
        payload: bytes,
        signature: str,
        secret: str,
    ) -> bool:
        """Verify webhook signature using HMAC-SHA256."""
        expected = hmac.new(
            secret.encode("utf-8"),
            payload,
            hashlib.sha256,
        ).hexdigest()
        return hmac.compare_digest(expected, signature)