feat: add check-in service layer

backend/checkins/services.py

@@ -0,0 +1,130 @@
+from dataclasses import dataclass
+
+from django.db import IntegrityError, transaction
+
+from bookings.models import Reservation, ReservationToken
+
+from .models import CheckIn
+
+
+class CheckInServiceError(Exception):
+    """Base class for check-in service domain errors."""
+
+
+class InvalidToken(CheckInServiceError):
+    pass
+
+
+class ReservationNotConfirmed(CheckInServiceError):
+    pass
+
+
+class AlreadyCheckedIn(CheckInServiceError):
+    pass
+
+
+class MissingStaffUser(CheckInServiceError):
+    pass
+
+
+@dataclass(frozen=True)
+class CheckInPreview:
+    reservation_id: int
+    performance_id: int
+    show_title: str
+    venue_name: str
+    starts_at: object
+    party_size: int
+    checked_in: bool
+
+
+@dataclass(frozen=True)
+class CheckInResult:
+    check_in: CheckIn
+    preview: CheckInPreview
+
+
+def preview_check_in_token(raw_token, *, staff_user):
+    _validate_staff_user(staff_user)
+    reservation = _get_reservation_for_check_in_token(raw_token)
+    _validate_reservation_for_check_in(reservation)
+    return _build_preview(reservation)
+
+
+def confirm_check_in_from_token(raw_token, *, staff_user, source=CheckIn.Source.QR_SCAN):
+    _validate_staff_user(staff_user)
+
+    with transaction.atomic():
+        reservation = _get_reservation_for_check_in_token(raw_token, lock_token=True)
+        reservation = (
+            Reservation.objects.select_for_update()
+            .select_related("performance__show", "performance__venue")
+            .get(pk=reservation.pk)
+        )
+        _validate_reservation_for_check_in(reservation)
+
+        try:
+            check_in = CheckIn.objects.create(
+                reservation=reservation,
+                checked_in_by=staff_user,
+                source=source,
+            )
+        except IntegrityError as exc:
+            raise AlreadyCheckedIn("Reservation has already been checked in.") from exc
+
+    return CheckInResult(check_in=check_in, preview=_build_preview(reservation))
+
+
+def _validate_staff_user(staff_user):
+    if staff_user is None:
+        raise MissingStaffUser("A staff user is required for check-in.")
+    if not getattr(staff_user, "is_authenticated", False):
+        raise MissingStaffUser("An authenticated staff user is required for check-in.")
+    if not getattr(staff_user, "is_staff", False):
+        raise MissingStaffUser("A staff user is required for check-in.")
+
+
+def _get_reservation_for_check_in_token(raw_token, *, lock_token=False):
+    if not isinstance(raw_token, str) or not raw_token:
+        raise InvalidToken("Check-in token is invalid.")
+
+    queryset = ReservationToken.objects.select_related(
+        "reservation__performance__show",
+        "reservation__performance__venue",
+    )
+    if lock_token:
+        queryset = queryset.select_for_update()
+
+    try:
+        token = queryset.get(
+            token_hash=ReservationToken.hash_token(raw_token),
+            purpose=ReservationToken.Purpose.CHECK_IN,
+            used_at__isnull=True,
+        )
+    except ReservationToken.DoesNotExist as exc:
+        raise InvalidToken("Check-in token is invalid.") from exc
+
+    if token.is_expired:
+        raise InvalidToken("Check-in token is invalid.")
+
+    return token.reservation
+
+
+def _validate_reservation_for_check_in(reservation):
+    if reservation.status != Reservation.Status.CONFIRMED:
+        raise ReservationNotConfirmed("Reservation must be confirmed before check-in.")
+    if hasattr(reservation, "check_in"):
+        raise AlreadyCheckedIn("Reservation has already been checked in.")
+
+
+def _build_preview(reservation):
+    performance = reservation.performance
+    return CheckInPreview(
+        reservation_id=reservation.id,
+        performance_id=performance.id,
+        show_title=performance.show.title,
+        venue_name=performance.venue.name,
+        starts_at=performance.starts_at,
+        party_size=reservation.party_size,
+        checked_in=hasattr(reservation, "check_in"),
+    )

backend/checkins/test_services.py

@@ -0,0 +1,134 @@
+from datetime import timedelta
+
+from django.contrib.auth import get_user_model
+from django.test import TestCase
+from django.utils import timezone
+
+from bookings.models import Reservation, ReservationToken
+from checkins.models import CheckIn
+from checkins.services import (
+    AlreadyCheckedIn,
+    InvalidToken,
+    MissingStaffUser,
+    ReservationNotConfirmed,
+    confirm_check_in_from_token,
+    preview_check_in_token,
+)
+from shows.models import Performance, Show, Venue
+
+
+class CheckInServiceTests(TestCase):
+    def setUp(self):
+        self.show = Show.objects.create(
+            title="Open Stage",
+            slug="open-stage-checkins",
+            is_published=True,
+        )
+        self.venue = Venue.objects.create(
+            name="AzioneLab Theatre",
+            slug="azionelab-theatre-checkins",
+            address="Via Example 1",
+            city="Rome",
+        )
+        self.performance = Performance.objects.create(
+            show=self.show,
+            venue=self.venue,
+            starts_at=timezone.now() + timedelta(days=7),
+            room_capacity=20,
+        )
+        self.staff_user = get_user_model().objects.create_user(
+            username="staff",
+            password="test",
+            is_staff=True,
+        )
+
+    def test_successful_preview_returns_minimum_admission_data(self):
+        reservation = self.create_reservation()
+        _, raw_token = self.create_check_in_token(reservation)
+
+        preview = preview_check_in_token(raw_token, staff_user=self.staff_user)
+
+        self.assertEqual(preview.reservation_id, reservation.id)
+        self.assertEqual(preview.performance_id, self.performance.id)
+        self.assertEqual(preview.show_title, self.show.title)
+        self.assertEqual(preview.venue_name, self.venue.name)
+        self.assertEqual(preview.party_size, reservation.party_size)
+        self.assertFalse(preview.checked_in)
+        self.assertFalse(hasattr(preview, "name"))
+        self.assertFalse(hasattr(preview, "email"))
+        self.assertFalse(hasattr(preview, "phone"))
+
+    def test_preview_fails_for_invalid_token(self):
+        with self.assertRaises(InvalidToken):
+            preview_check_in_token("invalid-token", staff_user=self.staff_user)
+
+    def test_check_in_succeeds_for_confirmed_reservation(self):
+        reservation = self.create_reservation()
+        _, raw_token = self.create_check_in_token(reservation)
+
+        result = confirm_check_in_from_token(raw_token, staff_user=self.staff_user)
+
+        self.assertEqual(result.check_in.reservation, reservation)
+        self.assertEqual(result.check_in.checked_in_by, self.staff_user)
+        self.assertEqual(result.check_in.source, CheckIn.Source.QR_SCAN)
+        self.assertTrue(result.preview.checked_in)
+
+    def test_check_in_fails_for_pending_reservation(self):
+        reservation = self.create_reservation(status=Reservation.Status.PENDING, confirmed_at=None)
+        _, raw_token = self.create_check_in_token(reservation)
+
+        with self.assertRaises(ReservationNotConfirmed):
+            confirm_check_in_from_token(raw_token, staff_user=self.staff_user)
+
+        self.assertFalse(CheckIn.objects.filter(reservation=reservation).exists())
+
+    def test_duplicate_check_in_fails(self):
+        reservation = self.create_reservation()
+        _, raw_token = self.create_check_in_token(reservation)
+        confirm_check_in_from_token(raw_token, staff_user=self.staff_user)
+
+        with self.assertRaises(AlreadyCheckedIn):
+            confirm_check_in_from_token(raw_token, staff_user=self.staff_user)
+
+        self.assertEqual(CheckIn.objects.filter(reservation=reservation).count(), 1)
+
+    def test_check_in_stores_timestamp_and_staff_user(self):
+        reservation = self.create_reservation()
+        _, raw_token = self.create_check_in_token(reservation)
+
+        before_check_in = timezone.now()
+        result = confirm_check_in_from_token(
+            raw_token,
+            staff_user=self.staff_user,
+            source=CheckIn.Source.MANUAL,
+        )
+
+        self.assertGreaterEqual(result.check_in.checked_in_at, before_check_in)
+        self.assertEqual(result.check_in.checked_in_by, self.staff_user)
+        self.assertEqual(result.check_in.source, CheckIn.Source.MANUAL)
+
+    def test_check_in_requires_staff_user(self):
+        reservation = self.create_reservation()
+        _, raw_token = self.create_check_in_token(reservation)
+
+        with self.assertRaises(MissingStaffUser):
+            confirm_check_in_from_token(raw_token, staff_user=None)
+
+    def create_reservation(self, **overrides):
+        data = {
+            "performance": self.performance,
+            "name": "Maria Rossi",
+            "email": "maria@example.com",
+            "party_size": 2,
+            "status": Reservation.Status.CONFIRMED,
+            "confirmed_at": timezone.now(),
+        }
+        data.update(overrides)
+        return Reservation.objects.create(**data)
+
+    def create_check_in_token(self, reservation):
+        return ReservationToken.create_token(
+            reservation=reservation,
+            purpose=ReservationToken.Purpose.CHECK_IN,
+            expires_at=self.performance.starts_at + timedelta(days=1),
+        )