azionelab/backend/bookings/services.py

from dataclasses import dataclass
from datetime import timedelta

from django.db import transaction
from django.db.models import Sum
from django.utils import timezone

from shows.models import Performance

from .emailing import send_confirmation_email
from .models import Reservation, ReservationToken
from .qr import build_check_in_preview_url, generate_check_in_qr_base64


CONFIRMATION_TOKEN_TTL = timedelta(hours=48)
CHECK_IN_TOKEN_AFTER_PERFORMANCE_TTL = timedelta(days=1)


class BookingServiceError(Exception):
    """Base class for booking service domain errors."""


class PerformanceNotAvailable(BookingServiceError):
    pass


class NotEnoughSeats(BookingServiceError):
    pass


class InvalidToken(BookingServiceError):
    pass


class ExpiredToken(BookingServiceError):
    pass


class AlreadyConfirmedReservation(BookingServiceError):
    pass


@dataclass(frozen=True)
class PendingReservationResult:
    reservation: Reservation
    confirmation_token: ReservationToken
    raw_confirmation_token: str
    available_seats: int


@dataclass(frozen=True)
class ConfirmedReservationResult:
    reservation: Reservation
    confirmation_token: ReservationToken
    check_in_token: ReservationToken
    raw_check_in_token: str
    available_seats: int
    qr_code_image: str
    qr_code_url: str


def calculate_available_seats(performance):
    confirmed_seats = (
        Reservation.objects.filter(
            performance=performance,
            status=Reservation.Status.CONFIRMED,
        ).aggregate(total=Sum("party_size"))["total"]
        or 0
    )
    return performance.configured_capacity - confirmed_seats


def create_pending_reservation(
    *,
    performance_id,
    name,
    email,
    party_size,
    phone="",
    notes="",
    confirmation_expires_at=None,
):
    with transaction.atomic():
        performance = _get_locked_bookable_performance(performance_id)
        available_seats = calculate_available_seats(performance)
        if party_size > available_seats:
            raise NotEnoughSeats("Not enough seats are available for this performance.")

        reservation = Reservation(
            performance=performance,
            name=name,
            email=email,
            phone=phone,
            party_size=party_size,
            notes=notes,
        )
        reservation.full_clean()
        reservation.save()

        confirmation_token, raw_confirmation_token = generate_confirmation_token(
            reservation,
            expires_at=confirmation_expires_at,
        )

    result = PendingReservationResult(
        reservation=reservation,
        confirmation_token=confirmation_token,
        raw_confirmation_token=raw_confirmation_token,
        available_seats=available_seats,
    )
    send_confirmation_email(
        reservation=result.reservation,
        raw_confirmation_token=result.raw_confirmation_token,
    )
    return result


def generate_confirmation_token(reservation, *, expires_at=None):
    return ReservationToken.create_token(
        reservation=reservation,
        purpose=ReservationToken.Purpose.CONFIRMATION,
        expires_at=expires_at or timezone.now() + CONFIRMATION_TOKEN_TTL,
    )


def confirm_reservation_from_token(raw_token):
    token_hash = ReservationToken.hash_token(raw_token)
    token_was_expired = False

    with transaction.atomic():
        try:
            confirmation_token = (
                ReservationToken.objects.select_for_update()
                .select_related("reservation")
                .get(token_hash=token_hash, purpose=ReservationToken.Purpose.CONFIRMATION)
            )
        except ReservationToken.DoesNotExist as exc:
            raise InvalidToken("Confirmation token is invalid.") from exc

        reservation = Reservation.objects.select_for_update().get(pk=confirmation_token.reservation_id)

        if confirmation_token.is_used:
            if reservation.status == Reservation.Status.CONFIRMED:
                raise AlreadyConfirmedReservation("Reservation is already confirmed.")
            raise InvalidToken("Confirmation token has already been used.")

        if confirmation_token.is_expired:
            if reservation.status == Reservation.Status.PENDING:
                reservation.status = Reservation.Status.EXPIRED
                reservation.save(update_fields=["status", "updated_at"])
            token_was_expired = True

        if not token_was_expired and reservation.status == Reservation.Status.CONFIRMED:
            raise AlreadyConfirmedReservation("Reservation is already confirmed.")

        if not token_was_expired and reservation.status != Reservation.Status.PENDING:
            raise InvalidToken("Confirmation token is not valid for this reservation.")

        if not token_was_expired:
            performance = _get_locked_bookable_performance(reservation.performance_id)
            available_seats = calculate_available_seats(performance)
            if reservation.party_size > available_seats:
                raise NotEnoughSeats("Not enough seats are available for this performance.")

            reservation.confirm()
            confirmation_token.mark_used()

            check_in_token, raw_check_in_token = ReservationToken.create_token(
                reservation=reservation,
                purpose=ReservationToken.Purpose.CHECK_IN,
                expires_at=performance.starts_at + CHECK_IN_TOKEN_AFTER_PERFORMANCE_TTL,
            )
            reservation.qr_code_generated_at = timezone.now()
            reservation.save(update_fields=["qr_code_generated_at", "updated_at"])

    if token_was_expired:
        raise ExpiredToken("Confirmation token has expired.")

    return ConfirmedReservationResult(
        reservation=reservation,
        confirmation_token=confirmation_token,
        check_in_token=check_in_token,
        raw_check_in_token=raw_check_in_token,
        available_seats=available_seats - reservation.party_size,
        qr_code_image=generate_check_in_qr_base64(
            reservation=reservation,
            raw_check_in_token=raw_check_in_token,
        ),
        qr_code_url=build_check_in_preview_url(raw_check_in_token),
    )


def _get_locked_bookable_performance(performance_id):
    try:
        performance = Performance.objects.select_for_update().get(pk=performance_id)
    except Performance.DoesNotExist as exc:
        raise PerformanceNotAvailable("Performance is not available for booking.") from exc

    if not performance.is_booking_enabled:
        raise PerformanceNotAvailable("Performance is not available for booking.")

    return performance