k1nq
130 lines
4.3 KiB
Python
130 lines
4.3 KiB
Python
import asyncio
|
|
import logging
|
|
import random
|
|
from collections.abc import Awaitable, Callable
|
|
from datetime import UTC, datetime
|
|
|
|
import httpx2
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker
|
|
|
|
from app.config import Settings
|
|
from app.models import Payment, PaymentStatus
|
|
from app.schemas import PaymentCreatedEvent, PaymentWebhook
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class PaymentNotFoundError(Exception):
|
|
pass
|
|
|
|
|
|
class PaymentGateway:
|
|
def __init__(self, settings: Settings) -> None:
|
|
self._min_seconds = settings.processing_min_seconds
|
|
self._max_seconds = settings.processing_max_seconds
|
|
self._success_rate = settings.processing_success_rate
|
|
|
|
async def charge(self) -> PaymentStatus:
|
|
await asyncio.sleep(random.uniform(self._min_seconds, self._max_seconds))
|
|
if random.random() < self._success_rate:
|
|
return PaymentStatus.SUCCEEDED
|
|
return PaymentStatus.FAILED
|
|
|
|
|
|
class WebhookClient:
|
|
def __init__(self, timeout: float) -> None:
|
|
self._timeout = timeout
|
|
|
|
async def send(self, url: str, payload: PaymentWebhook) -> None:
|
|
async with httpx2.AsyncClient(timeout=self._timeout) as client:
|
|
response = await client.post(url, json=payload.model_dump(mode="json"))
|
|
response.raise_for_status()
|
|
|
|
|
|
async def retry_async[T](
|
|
operation: Callable[[], Awaitable[T]],
|
|
*,
|
|
attempts: int,
|
|
base_delay: float,
|
|
sleep: Callable[[float], Awaitable[None]] = asyncio.sleep,
|
|
) -> T:
|
|
for attempt in range(attempts):
|
|
try:
|
|
return await operation()
|
|
except Exception:
|
|
if attempt == attempts - 1:
|
|
raise
|
|
delay = base_delay * (2**attempt)
|
|
logger.warning(
|
|
"Payment handling attempt %d/%d failed; retrying in %.2fs",
|
|
attempt + 1,
|
|
attempts,
|
|
delay,
|
|
exc_info=True,
|
|
)
|
|
await sleep(delay)
|
|
raise RuntimeError("Retry loop completed without a result")
|
|
|
|
|
|
class PaymentProcessor:
|
|
def __init__(
|
|
self,
|
|
session_factory: async_sessionmaker,
|
|
gateway: PaymentGateway,
|
|
webhook_client: WebhookClient,
|
|
settings: Settings,
|
|
) -> None:
|
|
self._session_factory = session_factory
|
|
self._gateway = gateway
|
|
self._webhook_client = webhook_client
|
|
self._attempts = settings.processing_max_attempts
|
|
self._base_delay = settings.retry_base_delay
|
|
|
|
async def process(self, event: PaymentCreatedEvent) -> None:
|
|
await retry_async(
|
|
lambda: self._process_once(event),
|
|
attempts=self._attempts,
|
|
base_delay=self._base_delay,
|
|
)
|
|
|
|
async def _process_once(self, event: PaymentCreatedEvent) -> None:
|
|
async with self._session_factory() as session:
|
|
payment = await session.scalar(
|
|
select(Payment).where(Payment.id == event.payment_id).with_for_update()
|
|
)
|
|
if payment is None:
|
|
raise PaymentNotFoundError(str(event.payment_id))
|
|
|
|
if payment.status is PaymentStatus.PENDING:
|
|
payment.status = await self._gateway.charge()
|
|
payment.processed_at = datetime.now(UTC)
|
|
await session.commit()
|
|
|
|
if payment.webhook_delivered_at is not None:
|
|
return
|
|
if payment.processed_at is None:
|
|
raise RuntimeError("A terminal payment has no processing timestamp")
|
|
|
|
webhook_url = payment.webhook_url
|
|
webhook = PaymentWebhook(
|
|
payment_id=payment.id,
|
|
amount=payment.amount,
|
|
currency=payment.currency,
|
|
status=payment.status,
|
|
processed_at=payment.processed_at,
|
|
metadata=payment.metadata_,
|
|
)
|
|
|
|
await self._webhook_client.send(webhook_url, webhook)
|
|
|
|
async with self._session_factory() as session:
|
|
payment = await session.scalar(
|
|
select(Payment).where(Payment.id == event.payment_id).with_for_update()
|
|
)
|
|
if payment is None:
|
|
raise PaymentNotFoundError(str(event.payment_id))
|
|
if payment.webhook_delivered_at is None:
|
|
payment.webhook_delivered_at = datetime.now(UTC)
|
|
await session.commit()
|