"""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, 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 logger = logging.getLogger(__name__) @shared_task(name="apps.events.expire_stale_registration_checkouts") def expire_stale_registration_checkouts_task(): """Release seats held by new checkout attempts that reached their deadline.""" return expire_stale_registration_checkouts() @shared_task(bind=True, max_retries=3) def send_registration_confirmation_notification(self, registration_pk: str, notification_id: str | None = None): try: 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, notification_id: str | None = None): try: 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