"""Process due announcement calls from the notification worker.""" import logging from datetime import timedelta import requests from django.db import transaction from django.utils import timezone from apps.core.models import VoipConfiguration from apps.integrations.announcements import AnnouncementRejected, place_announcement_call from apps.messaging.models import MessageReadReceipt from .models import AnnouncementJob logger = logging.getLogger(__name__) MAX_ATTEMPTS = 3 def _failure_code(exc): if isinstance(exc, requests.Timeout): return "service_timeout" if isinstance(exc, requests.ConnectionError): return "service_unreachable" if isinstance(exc, requests.HTTPError): return "service_http_error" if isinstance(exc, AnnouncementRejected): return "service_rejected" if isinstance(exc, ValueError): return "invalid_service_response" return "unexpected_error" def _cancellation_reason(job, configuration): message = job.message customer = message.ticket.customer if message.is_internal or message.deleted_at or message.author.account_type != "STAFF": return "message_unavailable" if MessageReadReceipt.objects.filter(message=message, user=customer).exists(): return "customer_read" if message.ticket.ticket_messages.filter(author=customer, created_at__gt=message.created_at, is_internal=False).exists(): return "customer_replied" if not customer.is_active or not customer.local_mobile: return "customer_mobile_unavailable" if not configuration.announcement_url or not configuration.announcement_id: return "announcement_not_configured" return "" def _deliver(job): try: configuration = VoipConfiguration.load() reason = _cancellation_reason(job, configuration) if reason: job.status = AnnouncementJob.Status.CANCELLED job.last_error = reason return place_announcement_call( configuration, mobile=job.message.ticket.customer.local_mobile, idempotency_key=job.idempotency_key, ) except Exception as exc: job.attempts += 1 job.last_error = _failure_code(exc) job.status = AnnouncementJob.Status.FAILED if job.attempts >= MAX_ATTEMPTS else AnnouncementJob.Status.PENDING job.next_attempt_at = timezone.now() + timedelta(minutes=3) logger.exception("Announcement call failed for job %s", job.pk) else: job.attempts += 1 job.status = AnnouncementJob.Status.SENT job.sent_at = timezone.now() job.last_error = "" def process_due_announcements(): now = timezone.now() # A crashed worker can safely retry with the same provider idempotency key. AnnouncementJob.objects.filter( status=AnnouncementJob.Status.PROCESSING, next_attempt_at__lt=now - timedelta(minutes=10), ).update(status=AnnouncementJob.Status.PENDING, next_attempt_at=now) with transaction.atomic(): jobs = list( AnnouncementJob.objects.select_for_update(skip_locked=True) .filter(status=AnnouncementJob.Status.PENDING, next_attempt_at__lte=now) .select_related("message__ticket__customer") .order_by("next_attempt_at", "pk")[:20] ) AnnouncementJob.objects.filter(pk__in=[job.pk for job in jobs]).update( status=AnnouncementJob.Status.PROCESSING, next_attempt_at=now, ) for job in jobs: _deliver(job) job.save(update_fields=["status", "attempts", "next_attempt_at", "sent_at", "last_error"])