feat(admin): add manual reservation operations

backend/bookings/services.py

@@ -44,6 +44,10 @@ class ReservationNotConfirmed(BookingServiceError):
     pass
 
 
+class ReservationNotPending(BookingServiceError):
+    pass
+
+
 @dataclass(frozen=True)
 class PendingReservationResult:
     reservation: Reservation
@@ -55,7 +59,7 @@ class PendingReservationResult:
 @dataclass(frozen=True)
 class ConfirmedReservationResult:
     reservation: Reservation
-    confirmation_token: ReservationToken
+    confirmation_token: ReservationToken | None
     check_in_token: ReservationToken
     raw_check_in_token: str
     available_seats: int
@@ -70,6 +74,15 @@ class ReservationQRResult:
     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(
@@ -171,37 +184,53 @@ def confirm_reservation_from_token(raw_token):
             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(
+            result = _confirm_pending_reservation(
                 reservation=reservation,
-                purpose=ReservationToken.Purpose.CHECK_IN,
-                expires_at=performance.starts_at + CHECK_IN_TOKEN_AFTER_PERFORMANCE_TTL,
+                confirmation_token=confirmation_token,
             )
-            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(
+    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,
-            raw_check_in_token=raw_check_in_token,
-        ),
-        qr_code_url=build_check_in_preview_url(raw_check_in_token),
-    )
+            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):
@@ -237,3 +266,48 @@ def _get_locked_bookable_performance(performance_id):
         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),
+    )