319 lines
13 KiB
Python
319 lines
13 KiB
Python
"""Bounded payment verification and recovery for lost gateway callbacks."""
|
|
|
|
import logging
|
|
from datetime import timedelta
|
|
|
|
from celery import shared_task
|
|
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 (
|
|
PaymentReversalRejected,
|
|
PaymentReversalUncertain,
|
|
PaymentVerificationUncertain,
|
|
claim_payment_verification,
|
|
request_payment_verification,
|
|
request_payment_reversal,
|
|
schedule_payment_verification_retry,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _status_result(payment):
|
|
return {
|
|
Payment.OrderStatusChoices.PAID: "paid",
|
|
Payment.OrderStatusChoices.FAILED: "failed",
|
|
Payment.OrderStatusChoices.CANCELED: "canceled",
|
|
}.get(payment.status)
|
|
|
|
|
|
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,
|
|
_mark_payment_unsuccessful,
|
|
_notify_payment_status,
|
|
)
|
|
|
|
try:
|
|
code, data = request_payment_verification(payment)
|
|
except PaymentVerificationUncertain as exc:
|
|
_mark_payment_pending(payment.id)
|
|
delay = schedule_payment_verification_retry(payment.id)
|
|
if delay is None:
|
|
logger.error(
|
|
"Payment %s remains pending after exhausting verification attempts; manual review is required",
|
|
payment.id,
|
|
exc_info=exc,
|
|
)
|
|
return "pending_manual_review"
|
|
_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}» تأیید شد؛ بازگشت وجه در حال پیگیری است"
|
|
if refund_pending
|
|
else f"پرداخت رویداد «{payment.event.title}» تأیید شد"
|
|
),
|
|
message=(
|
|
"ثبت‌نام لغو شده است. بازگشت وجه به‌صورت خودکار پیگیری می‌شود؛ اگر نتیجه روشن نشود، پشتیبانی آن را بررسی می‌کند."
|
|
if refund_pending
|
|
else "پرداخت شما با موفقیت ثبت شد و ثبت‌نام رویداد تکمیل شده است."
|
|
),
|
|
level="warning" if refund_pending else "success",
|
|
)
|
|
return "paid"
|
|
|
|
payment, result, transitioned = _mark_payment_unsuccessful(
|
|
payment.id,
|
|
Payment.OrderStatusChoices.FAILED,
|
|
)
|
|
if transitioned:
|
|
_notify_payment_status(
|
|
payment,
|
|
title=f"پرداخت رویداد «{payment.event.title}» ناموفق بود",
|
|
message="تراکنش شما توسط درگاه تأیید نشد. در صورت نیاز دوباره پرداخت را انجام دهید.",
|
|
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
|