113 lines
4.5 KiB
Python
113 lines
4.5 KiB
Python
"""Process due announcement calls from the notification worker."""
|
|
import logging
|
|
from datetime import timedelta
|
|
|
|
import requests
|
|
from django.db import transaction
|
|
from django.db.models import Case, IntegerField, Value, When
|
|
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):
|
|
if job.source == AnnouncementJob.Source.MANUAL:
|
|
if not job.ticket_id:
|
|
return "message_unavailable"
|
|
customer = job.ticket.customer
|
|
else:
|
|
if job.message is None:
|
|
return "message_unavailable"
|
|
customer = job.message.ticket.customer
|
|
reason = _automatic_cancellation_reason(job.message, customer)
|
|
if reason:
|
|
return reason
|
|
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 _automatic_cancellation_reason(message, 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"
|
|
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.ticket.customer if job.source == AnnouncementJob.Source.MANUAL else 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(*, job_id=None):
|
|
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():
|
|
queryset = (
|
|
# Nullable message/ticket joins must not be included in FOR UPDATE on PostgreSQL.
|
|
AnnouncementJob.objects.select_for_update(skip_locked=True, of=("self",))
|
|
.filter(status=AnnouncementJob.Status.PENDING, next_attempt_at__lte=now)
|
|
.select_related("message__ticket__customer", "ticket__customer")
|
|
.annotate(priority=Case(When(source=AnnouncementJob.Source.MANUAL, then=Value(0)), default=Value(1), output_field=IntegerField()))
|
|
)
|
|
if job_id is not None:
|
|
queryset = queryset.filter(pk=job_id)
|
|
jobs = list(queryset.order_by("priority", "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"])
|