feat(events): harden registration checkout lifecycle
This commit is contained in:
@@ -1,32 +1,63 @@
|
||||
"""Bounded retries for payment verification after an inconclusive callback."""
|
||||
"""Bounded payment verification and recovery for lost gateway callbacks."""
|
||||
|
||||
import logging
|
||||
from datetime import timedelta
|
||||
|
||||
from celery import shared_task
|
||||
from django.shortcuts import get_object_or_404
|
||||
from django.conf import settings
|
||||
from django.db import transaction
|
||||
from django.db.models import Q
|
||||
from django.utils import timezone
|
||||
|
||||
from apps.events.models import Registration
|
||||
from apps.payments.models import Payment
|
||||
from apps.payments.services import PaymentVerificationUncertain, request_payment_verification
|
||||
from apps.payments.services import (
|
||||
PaymentReversalRejected,
|
||||
PaymentReversalUncertain,
|
||||
PaymentVerificationUncertain,
|
||||
claim_payment_verification,
|
||||
request_payment_verification,
|
||||
request_payment_reversal,
|
||||
schedule_payment_verification_retry,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@shared_task(bind=True, max_retries=5, name="apps.payments.reconcile_payment")
|
||||
def reconcile_payment(self, payment_id: int):
|
||||
payment = get_object_or_404(
|
||||
Payment.objects.select_related("event", "user"),
|
||||
pk=payment_id,
|
||||
)
|
||||
if payment.status in (Payment.OrderStatusChoices.PAID, Payment.OrderStatusChoices.FAILED):
|
||||
return payment.get_status_display().lower()
|
||||
if not payment.authority:
|
||||
logger.error(
|
||||
"Payment %s remains pending without a gateway authority; manual review is required",
|
||||
payment.id,
|
||||
)
|
||||
return "pending_manual_review"
|
||||
def _status_result(payment):
|
||||
return {
|
||||
Payment.OrderStatusChoices.PAID: "paid",
|
||||
Payment.OrderStatusChoices.FAILED: "failed",
|
||||
Payment.OrderStatusChoices.CANCELED: "canceled",
|
||||
}.get(payment.status)
|
||||
|
||||
# Imported at task runtime to avoid a view/task import cycle.
|
||||
|
||||
def _queue_retry(payment_id: int, countdown: int):
|
||||
try:
|
||||
reconcile_payment.apply_async(args=(payment_id,), countdown=countdown)
|
||||
except Exception:
|
||||
logger.exception(
|
||||
"Unable to queue payment verification for payment %s; the recovery sweep will retry it",
|
||||
payment_id,
|
||||
)
|
||||
|
||||
|
||||
@shared_task(name="apps.payments.reconcile_payment")
|
||||
def reconcile_payment(payment_id: int):
|
||||
payment, claimed = claim_payment_verification(payment_id)
|
||||
if payment is None:
|
||||
return "not_found"
|
||||
if not claimed:
|
||||
terminal_result = _status_result(payment)
|
||||
if terminal_result:
|
||||
return terminal_result
|
||||
if payment.verification_manual_review_at is not None or not payment.authority:
|
||||
return "pending_manual_review"
|
||||
if payment.verification_lease_until and payment.verification_lease_until > timezone.now():
|
||||
return "verification_in_progress"
|
||||
return "not_due"
|
||||
|
||||
# Imported here to avoid a view/task import cycle.
|
||||
from apps.payments.api.views import (
|
||||
_mark_payment_paid,
|
||||
_mark_payment_pending,
|
||||
@@ -38,24 +69,37 @@ def reconcile_payment(self, payment_id: int):
|
||||
code, data = request_payment_verification(payment)
|
||||
except PaymentVerificationUncertain as exc:
|
||||
_mark_payment_pending(payment.id)
|
||||
if self.request.retries >= self.max_retries:
|
||||
delay = schedule_payment_verification_retry(payment.id)
|
||||
if delay is None:
|
||||
logger.error(
|
||||
"Payment %s verification remains uncertain after bounded retries; manual review is required",
|
||||
"Payment %s remains pending after exhausting verification attempts; manual review is required",
|
||||
payment.id,
|
||||
exc_info=exc,
|
||||
)
|
||||
return "pending_manual_review"
|
||||
countdown = min(60 * (2 ** self.request.retries), 15 * 60)
|
||||
raise self.retry(exc=exc, countdown=countdown)
|
||||
_queue_retry(payment.id, delay)
|
||||
return "retry_scheduled"
|
||||
|
||||
if code in (100, 101):
|
||||
payment, newly_verified = _mark_payment_paid(payment.id, data)
|
||||
if newly_verified:
|
||||
refund_pending = payment.refund_status in (
|
||||
Payment.RefundStatusChoices.PENDING,
|
||||
Payment.RefundStatusChoices.MANUAL_REVIEW,
|
||||
)
|
||||
_notify_payment_status(
|
||||
payment,
|
||||
title=f"پرداخت رویداد «{payment.event.title}» تأیید شد",
|
||||
message="پرداخت شما با موفقیت ثبت شد و ثبت‌نام رویداد تکمیل شده است.",
|
||||
level="success",
|
||||
title=(
|
||||
f"پرداخت رویداد «{payment.event.title}» تأیید شد؛ بازگشت وجه در حال پیگیری است"
|
||||
if refund_pending
|
||||
else f"پرداخت رویداد «{payment.event.title}» تأیید شد"
|
||||
),
|
||||
message=(
|
||||
"ثبت‌نام لغو شده است. بازگشت وجه به‌صورت خودکار پیگیری می‌شود؛ اگر نتیجه روشن نشود، پشتیبانی آن را بررسی می‌کند."
|
||||
if refund_pending
|
||||
else "پرداخت شما با موفقیت ثبت شد و ثبت‌نام رویداد تکمیل شده است."
|
||||
),
|
||||
level="warning" if refund_pending else "success",
|
||||
)
|
||||
return "paid"
|
||||
|
||||
@@ -71,3 +115,204 @@ def reconcile_payment(self, payment_id: int):
|
||||
level="error",
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
@shared_task(name="apps.payments.sweep_stale_payment_verifications")
|
||||
def sweep_stale_payment_verifications():
|
||||
"""Queue due old attempts so lost callbacks and broker failures recover safely."""
|
||||
now = timezone.now()
|
||||
grace_seconds = max(
|
||||
0,
|
||||
getattr(settings, "PAYMENT_VERIFICATION_GRACE_SECONDS", 1800),
|
||||
)
|
||||
cutoff = now - timedelta(seconds=grace_seconds)
|
||||
max_attempts = getattr(settings, "PAYMENT_VERIFICATION_MAX_ATTEMPTS", 6)
|
||||
dispatch_lease_seconds = max(
|
||||
60,
|
||||
getattr(settings, "PAYMENT_VERIFICATION_DISPATCH_LEASE_SECONDS", 900),
|
||||
)
|
||||
|
||||
candidates = (
|
||||
Payment.objects.filter(
|
||||
status__in=[
|
||||
Payment.OrderStatusChoices.INIT,
|
||||
Payment.OrderStatusChoices.PENDING,
|
||||
],
|
||||
authority__isnull=False,
|
||||
verification_attempts__lt=max_attempts,
|
||||
verification_manual_review_at__isnull=True,
|
||||
)
|
||||
.filter(
|
||||
Q(created_at__lte=cutoff)
|
||||
| Q(registration__cancellation_reason__in=[
|
||||
"checkout_expired", "user_cancelled", "event_cancelled",
|
||||
])
|
||||
)
|
||||
.filter(Q(verification_next_attempt_at__isnull=True) | Q(verification_next_attempt_at__lte=now))
|
||||
.filter(Q(verification_lease_until__isnull=True) | Q(verification_lease_until__lte=now))
|
||||
.filter(
|
||||
Q(verification_dispatch_expires_at__isnull=True)
|
||||
| Q(verification_dispatch_expires_at__lte=now)
|
||||
)
|
||||
.order_by("created_at", "pk")
|
||||
.values_list("pk", flat=True)[:100]
|
||||
)
|
||||
|
||||
scheduled = 0
|
||||
for payment_id in candidates:
|
||||
with transaction.atomic():
|
||||
payment = Payment.objects.select_for_update().filter(pk=payment_id).first()
|
||||
if payment is None:
|
||||
continue
|
||||
canceled_checkout = bool(
|
||||
payment.registration_id
|
||||
and Registration.objects.filter(
|
||||
pk=payment.registration_id,
|
||||
cancellation_reason__in=[
|
||||
Registration.CancellationReasonChoices.CHECKOUT_EXPIRED,
|
||||
Registration.CancellationReasonChoices.USER_CANCELLED,
|
||||
Registration.CancellationReasonChoices.EVENT_CANCELLED,
|
||||
],
|
||||
).exists()
|
||||
)
|
||||
if (
|
||||
payment.status not in (
|
||||
Payment.OrderStatusChoices.INIT,
|
||||
Payment.OrderStatusChoices.PENDING,
|
||||
)
|
||||
or not payment.authority
|
||||
or (payment.created_at > cutoff and not canceled_checkout)
|
||||
or payment.verification_attempts >= max_attempts
|
||||
or payment.verification_manual_review_at is not None
|
||||
or (payment.verification_lease_until and payment.verification_lease_until > now)
|
||||
or (
|
||||
payment.verification_next_attempt_at
|
||||
and payment.verification_next_attempt_at > now
|
||||
)
|
||||
or (
|
||||
payment.verification_dispatch_expires_at
|
||||
and payment.verification_dispatch_expires_at > now
|
||||
)
|
||||
):
|
||||
continue
|
||||
|
||||
dispatch_expires_at = now + timedelta(seconds=dispatch_lease_seconds)
|
||||
payment.verification_dispatch_expires_at = dispatch_expires_at
|
||||
payment.save(update_fields=["verification_dispatch_expires_at"])
|
||||
try:
|
||||
reconcile_payment.apply_async(args=(payment.id,))
|
||||
except Exception:
|
||||
Payment.objects.filter(
|
||||
pk=payment.id,
|
||||
verification_dispatch_expires_at=dispatch_expires_at,
|
||||
).update(verification_dispatch_expires_at=None)
|
||||
logger.exception(
|
||||
"Unable to dispatch stale payment %s for verification",
|
||||
payment.id,
|
||||
)
|
||||
continue
|
||||
scheduled += 1
|
||||
|
||||
return scheduled
|
||||
|
||||
|
||||
@shared_task(name="apps.payments.refund_payment")
|
||||
def refund_payment(payment_id: int):
|
||||
"""Attempt a bounded ZarinPal reversal; surface unresolved refunds to admins."""
|
||||
now = timezone.now()
|
||||
lease = timedelta(seconds=max(60, getattr(settings, "PAYMENT_REFUND_LEASE_SECONDS", 120)))
|
||||
with transaction.atomic():
|
||||
payment = Payment.objects.select_for_update().filter(pk=payment_id).first()
|
||||
if payment is None:
|
||||
return "not_found"
|
||||
if (
|
||||
payment.status != Payment.OrderStatusChoices.PAID
|
||||
or payment.refund_status != Payment.RefundStatusChoices.PENDING
|
||||
):
|
||||
return payment.refund_status
|
||||
if payment.refund_next_attempt_at and payment.refund_next_attempt_at > now:
|
||||
return "not_due"
|
||||
if not payment.authority:
|
||||
payment.refund_status = Payment.RefundStatusChoices.MANUAL_REVIEW
|
||||
payment.refund_last_error = "No gateway authority is available for reversal."
|
||||
payment.refund_next_attempt_at = None
|
||||
payment.save(update_fields=[
|
||||
"refund_status", "refund_last_error", "refund_next_attempt_at",
|
||||
])
|
||||
return "manual_review"
|
||||
|
||||
max_attempts = max(1, getattr(settings, "PAYMENT_REFUND_MAX_ATTEMPTS", 3))
|
||||
if payment.refund_attempts >= max_attempts:
|
||||
payment.refund_status = Payment.RefundStatusChoices.MANUAL_REVIEW
|
||||
payment.refund_next_attempt_at = None
|
||||
payment.save(update_fields=["refund_status", "refund_next_attempt_at"])
|
||||
return "manual_review"
|
||||
|
||||
payment.refund_attempts += 1
|
||||
payment.refund_last_attempt_at = now
|
||||
# Duplicate task deliveries cannot issue concurrent reversals. If a
|
||||
# worker dies after the external request, the sweep retries at lease end.
|
||||
payment.refund_next_attempt_at = now + lease
|
||||
payment.save(update_fields=[
|
||||
"refund_attempts", "refund_last_attempt_at", "refund_next_attempt_at",
|
||||
])
|
||||
|
||||
try:
|
||||
data = request_payment_reversal(payment)
|
||||
except PaymentReversalRejected as exc:
|
||||
Payment.objects.filter(pk=payment_id).update(
|
||||
refund_status=Payment.RefundStatusChoices.MANUAL_REVIEW,
|
||||
refund_last_error=str(exc)[:255],
|
||||
refund_next_attempt_at=None,
|
||||
)
|
||||
return "manual_review"
|
||||
except PaymentReversalUncertain as exc:
|
||||
with transaction.atomic():
|
||||
current = Payment.objects.select_for_update().get(pk=payment_id)
|
||||
current.refund_last_error = str(exc)[:255]
|
||||
if current.refund_attempts >= max(1, getattr(settings, "PAYMENT_REFUND_MAX_ATTEMPTS", 3)):
|
||||
current.refund_status = Payment.RefundStatusChoices.MANUAL_REVIEW
|
||||
current.refund_next_attempt_at = None
|
||||
result = "manual_review"
|
||||
else:
|
||||
delay = min(60 * (2 ** (current.refund_attempts - 1)), 15 * 60)
|
||||
current.refund_next_attempt_at = timezone.now() + timedelta(seconds=delay)
|
||||
result = "retry_scheduled"
|
||||
current.save(update_fields=[
|
||||
"refund_status", "refund_last_error", "refund_next_attempt_at",
|
||||
])
|
||||
if result == "retry_scheduled":
|
||||
try:
|
||||
refund_payment.apply_async(args=(payment_id,), countdown=delay)
|
||||
except Exception:
|
||||
logger.exception("Unable to queue refund retry for payment %s", payment_id)
|
||||
return result
|
||||
|
||||
Payment.objects.filter(pk=payment_id).update(
|
||||
refund_status=Payment.RefundStatusChoices.SUCCEEDED,
|
||||
refund_reference=str(data.get("ref_id") or data.get("reference") or "")[:64] or None,
|
||||
refund_last_error="",
|
||||
refund_next_attempt_at=None,
|
||||
)
|
||||
return "refunded"
|
||||
|
||||
|
||||
@shared_task(name="apps.payments.sweep_pending_refunds")
|
||||
def sweep_pending_refunds():
|
||||
"""Recover refund work when an immediate Celery dispatch is lost."""
|
||||
now = timezone.now()
|
||||
due_ids = Payment.objects.filter(
|
||||
status=Payment.OrderStatusChoices.PAID,
|
||||
refund_status=Payment.RefundStatusChoices.PENDING,
|
||||
).filter(
|
||||
Q(refund_next_attempt_at__isnull=True) | Q(refund_next_attempt_at__lte=now)
|
||||
).order_by("refund_last_attempt_at", "pk").values_list("pk", flat=True)[:100]
|
||||
|
||||
scheduled = 0
|
||||
for payment_id in due_ids:
|
||||
try:
|
||||
refund_payment.apply_async(args=(payment_id,))
|
||||
scheduled += 1
|
||||
except Exception:
|
||||
logger.exception("Unable to dispatch pending refund for payment %s", payment_id)
|
||||
return scheduled
|
||||
|
||||
Reference in New Issue
Block a user