diff --git a/apps/events/api/views.py b/apps/events/api/views.py index 62dfa96..7459135 100644 --- a/apps/events/api/views.py +++ b/apps/events/api/views.py @@ -39,7 +39,7 @@ from apps.events.models import Event, Registration from apps.events.checkout import ( CHECKOUT_TIMEOUT, cancel_event_checkouts, - expire_stale_registration_checkouts, + expire_user_registration_checkouts, mark_checkout_cancelled, queue_checkout_reconciliation, ) @@ -599,8 +599,13 @@ def register_for_event( """Register current user for an event""" user = request.auth now = timezone.now() - expire_stale_registration_checkouts(now=now, event_id=event_id) - + with transaction.atomic(): + event = get_object_or_404( + Event.objects.select_for_update(), + id=event_id, + is_deleted=False, + ) + expire_user_registration_checkouts(event=event, user=user, now=now) with transaction.atomic(): event = get_object_or_404( Event.objects.select_for_update(), @@ -937,10 +942,6 @@ def is_registered(request, event_id: int): .order_by("-registered_at", "-id") .first() ) - if registration and registration.status == Registration.StatusChoices.PENDING and registration.checkout_expires_at and registration.checkout_expires_at <= now: - expire_stale_registration_checkouts(now=now, event_id=event_id) - registration.refresh_from_db() - payment = None if registration: payment = registration.payments.order_by("-created_at", "-id").first() @@ -1154,7 +1155,6 @@ def registration_hub_status( lookup = {"pk": int(slug)} if slug.isdecimal() else {"slug": slug} event = get_object_or_404(Event, is_deleted=False, **lookup) now = timezone.now() - expire_stale_registration_checkouts(now=now, event_id=event.id) registration, payment = _registration_hub_records( event, request.auth, diff --git a/apps/events/checkout.py b/apps/events/checkout.py index f1d9859..3a1e8e2 100644 --- a/apps/events/checkout.py +++ b/apps/events/checkout.py @@ -1,6 +1,5 @@ """Registration checkout lifecycle rules shared by API views and workers.""" -import logging from datetime import timedelta from django.db import transaction @@ -10,16 +9,6 @@ from apps.events.models import Event, Registration CHECKOUT_TIMEOUT = timedelta(minutes=15) -logger = logging.getLogger(__name__) - - -def _dispatch_reconciliation(payment_id): - from apps.payments.tasks import reconcile_payment - - try: - reconcile_payment.apply_async(args=(payment_id,), countdown=0) - except Exception: - logger.exception("Unable to dispatch checkout payment %s for reconciliation", payment_id) def mark_checkout_cancelled(registration, reason): @@ -38,7 +27,7 @@ def mark_checkout_cancelled(registration, reason): def queue_checkout_reconciliation(registration): - """Reconcile each unresolved payment after a checkout is cancelled.""" + """Mark no-authority attempts for review; the recovery sweep dispatches the rest.""" from apps.payments.models import Payment payments = list( @@ -51,11 +40,7 @@ def queue_checkout_reconciliation(registration): ).only("id", "authority") ) for payment in payments: - if payment.authority: - transaction.on_commit( - lambda payment_id=payment.id: _dispatch_reconciliation(payment_id) - ) - else: + if not payment.authority: # The payment request may have reached the gateway before its # authority was persisted. Keep it visible for review until its # request response arrives or an administrator resolves it. @@ -69,7 +54,6 @@ def queue_checkout_reconciliation(registration): ).update(verification_manual_review_at=timezone.now()) -@transaction.atomic def expire_stale_registration_checkouts(*, now=None, event_id=None): """Release seats for new-format pending checkouts whose deadline passed. @@ -85,27 +69,48 @@ def expire_stale_registration_checkouts(*, now=None, event_id=None): if event_id is not None: candidates = candidates.filter(event_id=event_id) - expired_count = 0 - for registration_id in candidates.values_list("id", flat=True).iterator(): - registration = ( - Registration.objects.select_for_update() - .filter(pk=registration_id) - .first() + with transaction.atomic(): + batch = list( + candidates.select_for_update(skip_locked=True) + .order_by("checkout_expires_at", "pk")[:50] ) - if registration is None or registration.status != Registration.StatusChoices.PENDING: - continue + expired_count = 0 + for registration in batch: + if registration.status != Registration.StatusChoices.PENDING or registration.checkout_expires_at > now: + continue + reason = ( + Registration.CancellationReasonChoices.EVENT_CANCELLED + if registration.event.status == Event.StatusChoices.CANCELLED + else Registration.CancellationReasonChoices.CHECKOUT_EXPIRED + ) + if mark_checkout_cancelled(registration, reason): + expired_count += 1 + queue_checkout_reconciliation(registration) + return expired_count + +def expire_user_registration_checkouts(*, event, user, now): + """Expire only this user's rows while the caller holds the event lock.""" + registrations = ( + Registration.objects.select_for_update() + .filter( + event=event, + user=user, + is_deleted=False, + status=Registration.StatusChoices.PENDING, + checkout_expires_at__lte=now, + ) + .order_by("pk") + ) + for registration in registrations: reason = ( Registration.CancellationReasonChoices.EVENT_CANCELLED - if registration.event.status == Event.StatusChoices.CANCELLED + if event.status == Event.StatusChoices.CANCELLED else Registration.CancellationReasonChoices.CHECKOUT_EXPIRED ) if mark_checkout_cancelled(registration, reason): - expired_count += 1 queue_checkout_reconciliation(registration) - return expired_count - @transaction.atomic def cancel_event_checkouts(event_id): diff --git a/apps/events/migrations/0017_registrationnotification.py b/apps/events/migrations/0017_registrationnotification.py new file mode 100644 index 0000000..cedf0cc --- /dev/null +++ b/apps/events/migrations/0017_registrationnotification.py @@ -0,0 +1,37 @@ +from django.db import migrations, models +import django.db.models.deletion + + +class Migration(migrations.Migration): + dependencies = [ + ("events", "0016_registration_cancellation_reason_and_more"), + ] + + operations = [ + migrations.CreateModel( + name="RegistrationNotification", + fields=[ + ("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")), + ("kind", models.CharField(choices=[("confirmed", "Confirmed"), ("cancelled", "Cancelled")], max_length=10)), + ("created_at", models.DateTimeField(auto_now_add=True)), + ("delivered_at", models.DateTimeField(blank=True, null=True)), + ("lease_until", models.DateTimeField(blank=True, null=True)), + ( + "registration", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name="notification_outbox", + to="events.registration", + ), + ), + ], + options={ + "constraints": [ + models.UniqueConstraint( + fields=("registration", "kind"), + name="unique_registration_notification", + ), + ], + }, + ), + ] diff --git a/apps/events/migrations/0018_registrationnotification_reg_notif_pending_idx.py b/apps/events/migrations/0018_registrationnotification_reg_notif_pending_idx.py new file mode 100644 index 0000000..82dde3d --- /dev/null +++ b/apps/events/migrations/0018_registrationnotification_reg_notif_pending_idx.py @@ -0,0 +1,17 @@ +# Generated by Django 5.2.5 on 2026-10-01 09:49 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('events', '0017_registrationnotification'), + ] + + operations = [ + migrations.AddIndex( + model_name='registrationnotification', + index=models.Index(condition=models.Q(('delivered_at__isnull', True)), fields=['lease_until', 'created_at'], name='reg_notif_pending_idx'), + ), + ] diff --git a/apps/events/models.py b/apps/events/models.py index 2005143..74ecbca 100644 --- a/apps/events/models.py +++ b/apps/events/models.py @@ -1,4 +1,4 @@ -from django.db import models +from django.db import models, transaction from django.conf import settings from django.utils import timezone from django.utils.text import slugify @@ -201,39 +201,59 @@ class Registration(BaseModel): """Human-readable label for the current registration status.""" return self.get_status_display() + @transaction.atomic def save(self, *args, **kwargs): - # detect create vs update - is_create = self._state.adding - old_status = None + is_create = self._state.adding + old_status = None - if not is_create and self.pk: - old_status = ( - self.__class__.objects.only("status").get(pk=self.pk).status - ) + if not is_create and self.pk: + old_status = self.__class__.objects.only("status").get(pk=self.pk).status - # save first (so we have a pk + final values) - super().save(*args, **kwargs) + super().save(*args, **kwargs) - # 1) on create -> send confirmation if pending/confirmed (and not sent before) - if is_create and self.status == self.StatusChoices.CONFIRMED and not self.confirmation_notification_sent_at: - # lazy import to avoid circular import - from apps.events.tasks import send_registration_confirmation_notification - send_registration_confirmation_notification.delay(str(self.pk)) - self.confirmation_notification_sent_at = timezone.now() - super().save(update_fields=["confirmation_notification_sent_at"]) - # 2) status changed -> cancelled - if (not is_create) and (old_status != self.StatusChoices.CANCELLED) and (self.status == self.StatusChoices.CANCELLED) and (not self.cancellation_notification_sent_at): - from apps.events.tasks import send_registration_cancellation_notification - send_registration_cancellation_notification.delay(str(self.pk)) - self.cancellation_notification_sent_at = timezone.now() - super().save(update_fields=["cancellation_notification_sent_at"]) + if ( + self.status == self.StatusChoices.CONFIRMED + and (is_create or old_status != self.StatusChoices.CONFIRMED) + and not self.confirmation_notification_sent_at + ): + RegistrationNotification.objects.get_or_create( + registration=self, + kind=RegistrationNotification.Kind.CONFIRMED, + ) + if ( + not is_create + and old_status != self.StatusChoices.CANCELLED + and self.status == self.StatusChoices.CANCELLED + and not self.cancellation_notification_sent_at + ): + RegistrationNotification.objects.get_or_create( + registration=self, + kind=RegistrationNotification.Kind.CANCELLED, + ) - # 3) status changed -> confirmed (if not sent before) - if (not is_create) and (old_status != self.StatusChoices.CONFIRMED) and (self.status == self.StatusChoices.CONFIRMED) and (not self.confirmation_notification_sent_at): - from apps.events.tasks import send_registration_confirmation_notification - send_registration_confirmation_notification.delay(str(self.pk)) - self.confirmation_notification_sent_at = timezone.now() - super().save(update_fields=["confirmation_notification_sent_at"]) + from apps.gamification.services import sync_registration_attendance + sync_registration_attendance(self) - from apps.gamification.services import sync_registration_attendance - sync_registration_attendance(self) + +class RegistrationNotification(models.Model): + """Durable intent to notify after a registration status transition commits.""" + + class Kind(models.TextChoices): + CONFIRMED = "confirmed", "Confirmed" + CANCELLED = "cancelled", "Cancelled" + + registration = models.ForeignKey(Registration, on_delete=models.CASCADE, related_name="notification_outbox") + kind = models.CharField(max_length=10, choices=Kind.choices) + created_at = models.DateTimeField(auto_now_add=True) + delivered_at = models.DateTimeField(null=True, blank=True) + lease_until = models.DateTimeField(null=True, blank=True) + + class Meta: + constraints = [models.UniqueConstraint(fields=["registration", "kind"], name="unique_registration_notification")] + indexes = [ + models.Index( + fields=["lease_until", "created_at"], + name="reg_notif_pending_idx", + condition=models.Q(delivered_at__isnull=True), + ), + ] diff --git a/apps/events/tasks.py b/apps/events/tasks.py index bdabfa2..c05c809 100644 --- a/apps/events/tasks.py +++ b/apps/events/tasks.py @@ -1,10 +1,14 @@ """Automatic event registration notifications.""" import logging +from datetime import timedelta from celery import shared_task +from django.db import transaction +from django.db.models import Q +from django.utils import timezone -from apps.events.models import Registration +from apps.events.models import Registration, RegistrationNotification from apps.events.checkout import expire_stale_registration_checkouts from apps.notifications.services import notify_user from apps.users.tasks import send_critical_sms @@ -19,41 +23,121 @@ def expire_stale_registration_checkouts_task(): @shared_task(bind=True, max_retries=3) -def send_registration_confirmation_notification(self, registration_pk: str): +def send_registration_confirmation_notification(self, registration_pk: str, notification_id: str | None = None): try: - registration = Registration.objects.select_related("event", "user").get(pk=registration_pk) - notify_user(registration.user_id, { - "type": "event_registration", - "title": f"ثبت‌نام شما برای «{registration.event.title}» تأیید شد", - "message": "ثبت‌نام شما در این رویداد تأیید شده است.", - "level": "success", - "action_url": f"/events/{registration.event.slug}", - "entity_type": "event", - "entity_id": registration.event_id, - "meta": {"ticket_id": str(registration.ticket_id)}, - }) - return {"sent": True} + result = _send_confirmation(registration_pk, notification_id) + _mark_notification_delivered(notification_id) + return result except Exception as exc: logger.error("Failed to notify registration confirmation: %s", exc) raise self.retry(exc=exc, countdown=60) @shared_task(bind=True, max_retries=3) -def send_registration_cancellation_notification(self, registration_pk: str): +def send_registration_cancellation_notification(self, registration_pk: str, notification_id: str | None = None): try: - registration = Registration.objects.select_related("event", "user").get(pk=registration_pk) - notify_user(registration.user_id, { - "type": "event_cancellation", - "title": f"ثبت‌نام شما برای «{registration.event.title}» لغو شد", - "message": "ثبت‌نام شما در این رویداد لغو شد.", - "level": "warning", - "action_url": f"/events/{registration.event.slug}", - "entity_type": "event", - "entity_id": registration.event_id, - }) - if registration.user.mobile and registration.user.is_mobile_verified: - send_critical_sms.delay(registration.user.mobile, "event_cancellation", registration.event.title) - return {"sent": True} + result = _send_cancellation(registration_pk, notification_id) + _mark_notification_delivered(notification_id) + return result except Exception as exc: logger.error("Failed to notify registration cancellation: %s", exc) raise self.retry(exc=exc, countdown=60) + + +def _send_confirmation(registration_pk, notification_id=None): + registration = Registration.objects.select_related("event", "user").get(pk=registration_pk) + notify_user(registration.user_id, { + **({"id": notification_id} if notification_id else {}), + "type": "event_registration", + "title": f"ثبت‌نام شما برای «{registration.event.title}» تأیید شد", + "message": "ثبت‌نام شما در این رویداد تأیید شده است.", + "level": "success", + "action_url": f"/events/{registration.event.slug}", + "entity_type": "event", + "entity_id": registration.event_id, + "meta": {"ticket_id": str(registration.ticket_id)}, + }) + return {"sent": True} + + +def _send_cancellation(registration_pk, notification_id=None): + registration = Registration.objects.select_related("event", "user").get(pk=registration_pk) + notify_user(registration.user_id, { + **({"id": notification_id} if notification_id else {}), + "type": "event_cancellation", + "title": f"ثبت‌نام شما برای «{registration.event.title}» لغو شد", + "message": "ثبت‌نام شما در این رویداد لغو شد.", + "level": "warning", + "action_url": f"/events/{registration.event.slug}", + "entity_type": "event", + "entity_id": registration.event_id, + }) + if registration.user.mobile and registration.user.is_mobile_verified: + send_critical_sms.delay(registration.user.mobile, "event_cancellation", registration.event.title) + return {"sent": True} + + +def _mark_notification_delivered(notification_id): + if not notification_id or not notification_id.startswith("registration-"): + return + try: + entry_id = int(notification_id.split("-", 2)[1]) + except (IndexError, ValueError): + return + delivered_at = timezone.now() + entry = RegistrationNotification.objects.filter(pk=entry_id, delivered_at__isnull=True).first() + if entry is None: + return + entry.delivered_at = delivered_at + entry.lease_until = None + entry.save(update_fields=["delivered_at", "lease_until"]) + field = ( + "confirmation_notification_sent_at" + if entry.kind == RegistrationNotification.Kind.CONFIRMED + else "cancellation_notification_sent_at" + ) + Registration.objects.filter(pk=entry.registration_id, **{f"{field}__isnull": True}).update( + **{field: delivered_at} + ) + + +@shared_task(name="apps.events.deliver_pending_registration_notifications") +def deliver_pending_registration_notifications(): + """Recover notification delivery without involving registration write requests.""" + now = timezone.now() + entry_ids = list( + RegistrationNotification.objects.filter(delivered_at__isnull=True) + .filter(Q(lease_until__isnull=True) | Q(lease_until__lte=now)) + .order_by("created_at", "pk") + .values_list("pk", flat=True)[:100] + ) + delivered = 0 + for entry_id in entry_ids: + with transaction.atomic(): + entry = RegistrationNotification.objects.select_for_update(skip_locked=True).filter(pk=entry_id).first() + if entry is None or entry.delivered_at or (entry.lease_until and entry.lease_until > now): + continue + entry.lease_until = now + timedelta(minutes=2) + entry.save(update_fields=["lease_until"]) + try: + notification_id = f"registration-{entry.pk}-{entry.kind}" + if entry.kind == RegistrationNotification.Kind.CONFIRMED: + _send_confirmation(str(entry.registration_id), notification_id) + sent_field = "confirmation_notification_sent_at" + else: + _send_cancellation(str(entry.registration_id), notification_id) + sent_field = "cancellation_notification_sent_at" + except Exception: + logger.exception("Unable to deliver registration notification %s", entry_id) + RegistrationNotification.objects.filter(pk=entry_id, delivered_at__isnull=True).update(lease_until=None) + continue + delivered_at = timezone.now() + with transaction.atomic(): + RegistrationNotification.objects.filter(pk=entry_id, delivered_at__isnull=True).update( + delivered_at=delivered_at, lease_until=None, + ) + Registration.objects.filter(pk=entry.registration_id, **{f"{sent_field}__isnull": True}).update( + **{sent_field: delivered_at}, + ) + delivered += 1 + return delivered diff --git a/apps/events/tests/integration/test_checkout_lock_contention.py b/apps/events/tests/integration/test_checkout_lock_contention.py new file mode 100644 index 0000000..20ad41c --- /dev/null +++ b/apps/events/tests/integration/test_checkout_lock_contention.py @@ -0,0 +1,78 @@ +"""PostgreSQL contention checks for the checkout expiry worker and status API.""" + +import threading +import time +from datetime import timedelta + +import pytest +from django.db import connection, connections, transaction +from django.test import Client +from django.utils import timezone + +from apps.events.checkout import expire_stale_registration_checkouts +from apps.events.models import Event, Registration +from apps.users.models import User +from core.authentication import create_jwt_token + + +@pytest.mark.django_db(transaction=True) +@pytest.mark.integration +def test_status_reads_and_expiry_worker_progress_around_a_locked_checkout(): + if connection.vendor != "postgresql": + pytest.skip("PostgreSQL row locks are required") + + now = timezone.now() + event = Event.objects.create( + title="Checkout lock test", slug="checkout-lock-test", description="Lock test", + start_time=now + timedelta(days=2), end_time=now + timedelta(days=2, hours=1), + registration_start_date=now - timedelta(days=1), + registration_end_date=now + timedelta(days=1), + status=Event.StatusChoices.PUBLISHED, price=1000, capacity=2, + ) + user = User.objects.create_user(username="checkout-lock-owner", mobile="09191112233", password="TestPass123!") + other = User.objects.create_user(username="checkout-lock-other", mobile="09191112234", password="TestPass123!") + locked = Registration.objects.create( + event=event, user=user, status=Registration.StatusChoices.PENDING, + checkout_expires_at=now - timedelta(minutes=2), + ) + free = Registration.objects.create( + event=event, user=other, status=Registration.StatusChoices.PENDING, + checkout_expires_at=now - timedelta(minutes=1), + ) + acquired = threading.Event() + release = threading.Event() + + def hold_row_lock(): + try: + with transaction.atomic(): + Registration.objects.select_for_update().get(pk=locked.pk) + acquired.set() + release.wait(10) + finally: + connections.close_all() + + holder = threading.Thread(target=hold_row_lock) + holder.start() + try: + assert acquired.wait(5) + start = time.monotonic() + response = Client().get( + f"/api/events/{event.slug}/registration/status", + HTTP_AUTHORIZATION=f"Bearer {create_jwt_token(user)}", + ) + assert response.status_code == 200 + assert time.monotonic() - start < 2 + + assert expire_stale_registration_checkouts(event_id=event.id) == 1 + locked.refresh_from_db() + free.refresh_from_db() + assert locked.status == Registration.StatusChoices.PENDING + assert free.status == Registration.StatusChoices.CANCELLED + finally: + release.set() + holder.join(timeout=5) + + assert not holder.is_alive() + assert expire_stale_registration_checkouts(event_id=event.id) == 1 + locked.refresh_from_db() + assert locked.status == Registration.StatusChoices.CANCELLED diff --git a/apps/events/tests/integration/test_event_models.py b/apps/events/tests/integration/test_event_models.py index 63b01f1..b7885fd 100644 --- a/apps/events/tests/integration/test_event_models.py +++ b/apps/events/tests/integration/test_event_models.py @@ -4,8 +4,9 @@ from datetime import timedelta from django.contrib import admin from django.utils import timezone from apps.events.admin import EventAdmin, RegistrationAdmin -from apps.events.models import Event, Registration +from apps.events.models import Event, Registration, RegistrationNotification from apps.events.tasks import send_registration_cancellation_notification, send_registration_confirmation_notification +from apps.events.tasks import deliver_pending_registration_notifications from apps.users.models import User @pytest.mark.django_db @@ -25,10 +26,33 @@ class TestEventModelTests: def test_capacity_counts_confirmed_registration(self, mocker): delay = mocker.patch('apps.events.tasks.send_registration_confirmation_notification.delay') - Registration.objects.create(event=self.event, user=self.user, status=Registration.StatusChoices.CONFIRMED) + registration = Registration.objects.create(event=self.event, user=self.user, status=Registration.StatusChoices.CONFIRMED) assert self.event.current_attendees_count == 1 assert not self.event.has_available_slots - delay.assert_called_once() + assert RegistrationNotification.objects.filter( + registration=registration, kind=RegistrationNotification.Kind.CONFIRMED, + ).exists() + delay.assert_not_called() + + def test_notification_worker_retries_after_redis_failure(self, mocker): + registration = Registration.objects.create( + event=self.event, user=self.user, status=Registration.StatusChoices.CONFIRMED, + ) + entry = RegistrationNotification.objects.get(registration=registration) + notify = mocker.patch("apps.events.tasks.notify_user", side_effect=ConnectionError("redis unavailable")) + + assert deliver_pending_registration_notifications() == 0 + entry.refresh_from_db() + assert entry.delivered_at is None + + notify.side_effect = None + assert deliver_pending_registration_notifications() == 1 + entry.refresh_from_db() + registration.refresh_from_db() + assert entry.delivered_at is not None + assert registration.confirmation_notification_sent_at is not None + assert notify.call_args.args[1]["id"] == f"registration-{entry.pk}-confirmed" + assert deliver_pending_registration_notifications() == 0 def test_unlimited_capacity_allows_registration(self): self.event.capacity = None diff --git a/apps/events/tests/integration/test_registration_checkout.py b/apps/events/tests/integration/test_registration_checkout.py index f9c6fc5..ba081b7 100644 --- a/apps/events/tests/integration/test_registration_checkout.py +++ b/apps/events/tests/integration/test_registration_checkout.py @@ -6,7 +6,7 @@ import pytest from django.test import Client from django.utils import timezone -from apps.events.models import Event, Registration +from apps.events.models import Event, Registration, RegistrationNotification from apps.events.checkout import cancel_event_checkouts, expire_stale_registration_checkouts from apps.payments.models import Payment from apps.users.models import User @@ -142,6 +142,23 @@ class TestRegistrationCheckout: assert payload["payment"] is None assert payload["can_register"] is True + def test_status_reads_do_not_cancel_an_expired_checkout_when_notifications_are_unavailable(self, mocker): + registration, _ = self.create_pending_checkout(expires_at=timezone.now() - timedelta(seconds=1)) + mocker.patch( + "apps.events.tasks.send_registration_cancellation_notification.delay", + side_effect=RuntimeError("broker unavailable"), + ) + + hub = self.client.get(f"/api/events/{self.event.slug}/registration/status", **self.auth_headers()) + legacy = self.client.get(f"/api/events/{self.event.id}/is-registered", **self.auth_headers()) + + registration.refresh_from_db() + assert hub.status_code == 200 + assert hub.json()["state"] == "verification_pending" + assert legacy.status_code == 200 + assert registration.status == Registration.StatusChoices.PENDING + assert not RegistrationNotification.objects.filter(registration=registration).exists() + def test_registration_hub_requires_authentication(self): response = self.client.get(f"/api/events/{self.event.slug}/registration/status") assert response.status_code == 401 @@ -376,6 +393,30 @@ class TestRegistrationCheckout: assert payment.status == Payment.OrderStatusChoices.PENDING assert Payment.objects.filter(event=self.event, user=self.user).count() == 1 + def test_payment_creation_rechecks_window_after_expiring_checkout_without_payment(self): + registration = Registration.objects.create( + event=self.event, + user=self.user, + status=Registration.StatusChoices.PENDING, + checkout_expires_at=timezone.now() - timedelta(seconds=1), + final_price=self.event.price, + ) + self.event.registration_end_date = timezone.now() - timedelta(minutes=1) + self.event.save(update_fields=["registration_end_date"]) + + response = self.client.post( + "/api/payments/create", + data=json.dumps({"event_id": self.event.id, "description": "Expired checkout", + "idempotency_key": str(uuid.uuid4())}), + content_type="application/json", + **self.auth_headers(), + ) + + registration.refresh_from_db() + assert response.status_code == 400 + assert registration.status == Registration.StatusChoices.CANCELLED + assert not Payment.objects.filter(registration=registration).exists() + def test_expiry_sweep_cancels_only_new_expired_checkouts_and_releases_the_seat(self): expired_at = timezone.now() - timedelta(seconds=1) registration, payment = self.create_pending_checkout(expires_at=expired_at) @@ -463,10 +504,11 @@ class TestRegistrationCheckout: assert payment_response.status_code == 409 assert Registration.objects.filter(event=self.event, user=self.user).count() == 1 - def test_event_cancellation_ends_checkouts_and_queues_payment_reconciliation(self, mocker): + def test_event_cancellation_ends_checkouts_and_recovers_payment_reconciliation(self, mocker): registration, payment = self.create_pending_checkout() + from apps.payments.tasks import sweep_stale_payment_verifications + enqueue = mocker.patch("apps.payments.tasks.reconcile_payment.apply_async") - mocker.patch("apps.events.checkout.transaction.on_commit", side_effect=lambda callback: callback()) self.event.status = Event.StatusChoices.CANCELLED self.event.save(update_fields=["status"]) @@ -477,4 +519,9 @@ class TestRegistrationCheckout: assert registration.status == Registration.StatusChoices.CANCELLED assert registration.cancellation_reason == Registration.CancellationReasonChoices.EVENT_CANCELLED assert Event.objects.get(pk=self.event.pk).current_attendees_count == 0 - enqueue.assert_called_once_with(args=(payment.id,), countdown=0) + assert RegistrationNotification.objects.filter( + registration=registration, + kind=RegistrationNotification.Kind.CANCELLED, + ).exists() + assert sweep_stale_payment_verifications() == 1 + enqueue.assert_called_once_with(args=(payment.id,)) diff --git a/apps/notifications/services.py b/apps/notifications/services.py index 01fa743..c3f1318 100644 --- a/apps/notifications/services.py +++ b/apps/notifications/services.py @@ -88,6 +88,10 @@ class RedisNotificationStore: @classmethod def add(cls, user_id: str, payload: dict) -> dict: + if payload.get("id"): + existing = cls.get(user_id, str(payload["id"])) + if existing: + return existing data = cls._normalize_notification(payload) created_at = parse_datetime(data["created_at"]) or timezone.now() created_at_ts = created_at.timestamp() diff --git a/apps/payments/api/views.py b/apps/payments/api/views.py index efde831..687c7d5 100644 --- a/apps/payments/api/views.py +++ b/apps/payments/api/views.py @@ -18,7 +18,7 @@ from apps.payments.models import Payment, DiscountCode from apps.events.models import Event, Registration from apps.events.checkout import ( CHECKOUT_TIMEOUT, - expire_stale_registration_checkouts, + expire_user_registration_checkouts, queue_checkout_reconciliation, ) from apps.notifications.services import notify_user @@ -316,8 +316,13 @@ def _ensure_checkout_attempt_can_be_reused(payment: Payment, *, now, allow_revie def create_payment(request, payload: CreatePaymentIn): idempotency_key = payload.idempotency_key or uuid4() now = timezone.now() - expire_stale_registration_checkouts(now=now, event_id=payload.event_id) - + with transaction.atomic(): + event = get_object_or_404( + Event.objects.select_for_update(), + pk=payload.event_id, + is_deleted=False, + ) + expire_user_registration_checkouts(event=event, user=request.auth, now=now) # Serialize checkout creation for this event so concurrent taps cannot # both create a new active attempt for the same user/event pair. with transaction.atomic(): diff --git a/config/api_errors.py b/config/api_errors.py new file mode 100644 index 0000000..7ae7061 --- /dev/null +++ b/config/api_errors.py @@ -0,0 +1,18 @@ +"""HTTP handling for recoverable database errors.""" + +from django.http import JsonResponse + + +def handle_database_operational_error(request, error): + """Return a retryable response when PostgreSQL rejects a lock wait.""" + cause = error.__cause__ + while cause is not None: + if getattr(cause, "sqlstate", None) == "55P03" or getattr(cause, "pgcode", None) == "55P03": + response = JsonResponse( + {"detail": "درخواست به‌دلیل شلوغی موقت پایگاه داده انجام نشد؛ کمی بعد دوباره تلاش کنید."}, + status=503, + ) + response["Retry-After"] = "2" + return response + cause = cause.__cause__ + raise error diff --git a/config/services/celery.py b/config/services/celery.py index 8a875f0..79b1c9c 100644 --- a/config/services/celery.py +++ b/config/services/celery.py @@ -48,6 +48,11 @@ app.conf.beat_schedule = { 'schedule': crontab(minute='*'), 'description': 'Releases seats from paid checkouts after their 15-minute window.', }, + 'deliver-registration-notifications': { + 'task': 'apps.events.deliver_pending_registration_notifications', + 'schedule': crontab(minute='*'), + 'description': 'Retries committed registration notifications.', + }, 'sweep-pending-refunds': { 'task': 'apps.payments.sweep_pending_refunds', 'schedule': crontab(minute='*/5'), diff --git a/config/settings/base.py b/config/settings/base.py index 07c593b..fa99dd7 100644 --- a/config/settings/base.py +++ b/config/settings/base.py @@ -249,6 +249,15 @@ BLOG_ASSET_MAX_SIZE_MB = config('BLOG_ASSET_MAX_SIZE_MB', default=50, cast=int) BLOG_IMAGE_ASSET_MAX_SIZE_MB = config('BLOG_IMAGE_ASSET_MAX_SIZE_MB', default=10, cast=int) if DATABASES["default"]["ENGINE"] == "django.db.backends.postgresql": + lock_timeout_ms = config("DB_LOCK_TIMEOUT_MS", default=5000, cast=int) + idle_transaction_timeout_ms = config("DB_IDLE_TRANSACTION_TIMEOUT_MS", default=60000, cast=int) + database_options = DATABASES["default"].setdefault("OPTIONS", {}) + existing_options = database_options.get("options") or "" + timeout_options = ( + f"-c lock_timeout={lock_timeout_ms}ms " + f"-c idle_in_transaction_session_timeout={idle_transaction_timeout_ms}ms" + ) + database_options["options"] = f"{existing_options} {timeout_options}".strip() DATABASES["default"]["ENGINE"] = "django_prometheus.db.backends.postgresql" from config.services.unfold import * diff --git a/config/tests/unit/test_api_errors.py b/config/tests/unit/test_api_errors.py new file mode 100644 index 0000000..95777e6 --- /dev/null +++ b/config/tests/unit/test_api_errors.py @@ -0,0 +1,35 @@ +import json + +import pytest +from django.db import OperationalError + +from config.api_errors import handle_database_operational_error + +pytestmark = pytest.mark.unit + + +class PostgresError(Exception): + def __init__(self, sqlstate): + super().__init__(sqlstate) + self.sqlstate = sqlstate + + +def test_database_lock_timeout_returns_retryable_service_unavailable(): + error = OperationalError("canceling statement due to lock timeout") + error.__cause__ = PostgresError("55P03") + + response = handle_database_operational_error(None, error) + + assert response.status_code == 503 + assert response["Retry-After"] == "2" + assert json.loads(response.content) == { + "detail": "درخواست به‌دلیل شلوغی موقت پایگاه داده انجام نشد؛ کمی بعد دوباره تلاش کنید." + } + + +def test_unrelated_database_operational_error_is_not_hidden(): + error = OperationalError("database unavailable") + error.__cause__ = PostgresError("08006") + + with pytest.raises(OperationalError, match="database unavailable"): + handle_database_operational_error(None, error) diff --git a/config/urls.py b/config/urls.py index 60ea067..c58d77b 100644 --- a/config/urls.py +++ b/config/urls.py @@ -1,16 +1,19 @@ from django.contrib import admin +from django.db import OperationalError from django.urls import path, include from django.conf import settings from django.conf.urls.static import static from ninja import NinjaAPI from apps.notifications.api.views import NotificationStreamView from config.api import router as api_router +from config.api_errors import handle_database_operational_error api = NinjaAPI( title="CS Association API", version="1.0.0", description="API for University Computer Science Association", ) +api.add_exception_handler(OperationalError, handle_database_operational_error) api.add_router("", api_router)