generated from bisco/codex-bootstrap
314 lines
9.9 KiB
Python
314 lines
9.9 KiB
Python
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
|
|
|
|
|
|
class ReservationNotConfirmed(BookingServiceError):
|
|
pass
|
|
|
|
|
|
class ReservationNotPending(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 | None
|
|
check_in_token: ReservationToken
|
|
raw_check_in_token: str
|
|
available_seats: int
|
|
qr_code_image: str
|
|
qr_code_url: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ReservationQRResult:
|
|
reservation: Reservation
|
|
qr_code_image: str
|
|
qr_code_url: str
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ReservationCheckInAccessResult:
|
|
reservation: Reservation
|
|
check_in_token: ReservationToken
|
|
raw_check_in_token: str
|
|
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,
|
|
)
|
|
|
|
transaction.on_commit(
|
|
lambda reservation=reservation, raw_confirmation_token=raw_confirmation_token: send_confirmation_email(
|
|
reservation=reservation,
|
|
raw_confirmation_token=raw_confirmation_token,
|
|
)
|
|
)
|
|
|
|
result = PendingReservationResult(
|
|
reservation=reservation,
|
|
confirmation_token=confirmation_token,
|
|
raw_confirmation_token=raw_confirmation_token,
|
|
available_seats=available_seats,
|
|
)
|
|
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:
|
|
result = _confirm_pending_reservation(
|
|
reservation=reservation,
|
|
confirmation_token=confirmation_token,
|
|
)
|
|
|
|
if token_was_expired:
|
|
raise ExpiredToken("Confirmation token has expired.")
|
|
|
|
return result
|
|
|
|
|
|
def confirm_reservation_manually(*, reservation_id):
|
|
with transaction.atomic():
|
|
reservation = Reservation.objects.select_for_update().get(pk=reservation_id)
|
|
if reservation.status == Reservation.Status.CONFIRMED:
|
|
raise AlreadyConfirmedReservation("Reservation is already confirmed.")
|
|
if reservation.status != Reservation.Status.PENDING:
|
|
raise ReservationNotPending("Only pending reservations can be confirmed manually.")
|
|
|
|
confirmation_token = (
|
|
ReservationToken.objects.select_for_update()
|
|
.filter(
|
|
reservation=reservation,
|
|
purpose=ReservationToken.Purpose.CONFIRMATION,
|
|
)
|
|
.order_by("-created_at")
|
|
.first()
|
|
)
|
|
return _confirm_pending_reservation(
|
|
reservation=reservation,
|
|
confirmation_token=confirmation_token,
|
|
)
|
|
|
|
|
|
def issue_check_in_access_for_reservation(*, reservation_id):
|
|
with transaction.atomic():
|
|
reservation = (
|
|
Reservation.objects.select_for_update()
|
|
.select_related("performance__show", "performance__venue")
|
|
.get(pk=reservation_id)
|
|
)
|
|
if reservation.status != Reservation.Status.CONFIRMED:
|
|
raise ReservationNotConfirmed("Reservation must be confirmed before QR retrieval.")
|
|
|
|
result = _issue_check_in_access(reservation)
|
|
|
|
return result
|
|
|
|
|
|
def retrieve_reservation_qr_from_token(raw_token):
|
|
try:
|
|
check_in_token = ReservationToken.objects.select_related("reservation").get_valid_token(
|
|
raw_token,
|
|
ReservationToken.Purpose.CHECK_IN,
|
|
)
|
|
except ReservationToken.DoesNotExist as exc:
|
|
raise InvalidToken("Check-in token is invalid.") from exc
|
|
|
|
reservation = check_in_token.reservation
|
|
if reservation.status != Reservation.Status.CONFIRMED:
|
|
raise ReservationNotConfirmed("Reservation must be confirmed before QR retrieval.")
|
|
|
|
return ReservationQRResult(
|
|
reservation=reservation,
|
|
qr_code_image=generate_check_in_qr_base64(
|
|
reservation=reservation,
|
|
raw_check_in_token=raw_token,
|
|
),
|
|
qr_code_url=build_check_in_preview_url(raw_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
|
|
|
|
|
|
def _confirm_pending_reservation(*, reservation, confirmation_token=None):
|
|
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()
|
|
if confirmation_token is not None and not confirmation_token.is_used:
|
|
confirmation_token.mark_used()
|
|
|
|
check_in_access = _issue_check_in_access(reservation)
|
|
|
|
return ConfirmedReservationResult(
|
|
reservation=reservation,
|
|
confirmation_token=confirmation_token,
|
|
check_in_token=check_in_access.check_in_token,
|
|
raw_check_in_token=check_in_access.raw_check_in_token,
|
|
available_seats=available_seats - reservation.party_size,
|
|
qr_code_image=check_in_access.qr_code_image,
|
|
qr_code_url=check_in_access.qr_code_url,
|
|
)
|
|
|
|
|
|
def _issue_check_in_access(reservation):
|
|
check_in_token, raw_check_in_token = ReservationToken.create_token(
|
|
reservation=reservation,
|
|
purpose=ReservationToken.Purpose.CHECK_IN,
|
|
expires_at=reservation.performance.starts_at + CHECK_IN_TOKEN_AFTER_PERFORMANCE_TTL,
|
|
)
|
|
if reservation.qr_code_generated_at is None:
|
|
reservation.qr_code_generated_at = timezone.now()
|
|
reservation.save(update_fields=["qr_code_generated_at", "updated_at"])
|
|
|
|
return ReservationCheckInAccessResult(
|
|
reservation=reservation,
|
|
check_in_token=check_in_token,
|
|
raw_check_in_token=raw_check_in_token,
|
|
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),
|
|
)
|