Design a Movie Ticket Booking System.

# booking.py, part 1: the catalogue (things that rarely change)
from __future__ import annotations

import itertools
import threading
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from enum import Enum


class SeatType(Enum):
    REGULAR = "regular"
    PREMIUM = "premium"


@dataclass(frozen=True)
class Seat:
    seat_id: str  # "A1"
    type: SeatType


@dataclass(frozen=True)
class Show:
    show_id: str
    movie: str
    screen: str
    starts_at: float
    seats: dict[str, Seat]
    prices: dict[SeatType, int]  # rupees


class BookingStatus(Enum):
    HELD = "held"
    CONFIRMED = "confirmed"
    EXPIRED = "expired"
    CANCELLED = "cancelled"


@dataclass
class Booking:
    booking_id: str
    user: str
    show_id: str
    seat_ids: tuple[str, ...]
    amount: int
    expires_at: float
    status: BookingStatus = BookingStatus.HELD
    payment_ref: str | None = None

# part 2: the seam that decides who owns a seat right now
class SeatLockProvider(ABC):
    @abstractmethod
    def lock(self, show_id: str, seat_ids: list[str], owner: str, ttl: float, now: float) -> bool: ...

    @abstractmethod
    def holds(self, show_id: str, seat_ids: list[str], owner: str, now: float) -> bool: ...

    @abstractmethod
    def unlock(self, show_id: str, seat_ids: list[str], owner: str) -> None: ...

    @abstractmethod
    def is_held(self, show_id: str, seat_id: str, now: float) -> bool: ...


@dataclass
class _Hold:
    owner: str
    expires_at: float


class InMemorySeatLocks(SeatLockProvider):
    """All-or-nothing holds with a TTL; an expired hold is treated as free (lazy expiry)."""

    def __init__(self) -> None:
        self._holds: dict[tuple[str, str], _Hold] = {}
        self._mu = threading.Lock()

    def lock(self, show_id: str, seat_ids: list[str], owner: str, ttl: float, now: float) -> bool:
        with self._mu:
            for s in seat_ids:
                h = self._holds.get((show_id, s))
                if h and h.expires_at > now and h.owner != owner:
                    return False  # one seat taken: take none
            for s in seat_ids:
                self._holds[(show_id, s)] = _Hold(owner, now + ttl)
            return True

    def holds(self, show_id: str, seat_ids: list[str], owner: str, now: float) -> bool:
        with self._mu:
            return all((h := self._holds.get((show_id, s))) is not None and h.owner == owner and h.expires_at > now for s in seat_ids)

    def unlock(self, show_id: str, seat_ids: list[str], owner: str) -> None:
        with self._mu:
            for s in seat_ids:
                if (h := self._holds.get((show_id, s))) and h.owner == owner:
                    del self._holds[(show_id, s)]

    def is_held(self, show_id: str, seat_id: str, now: float) -> bool:
        with self._mu:
            return (h := self._holds.get((show_id, seat_id))) is not None and h.expires_at > now


class PaymentGateway(ABC):
    @abstractmethod
    def refund(self, payment_ref: str, amount: int) -> None: ...


class FakeGateway(PaymentGateway):
    def __init__(self) -> None:
        self.refunds: list[tuple[str, int]] = []

    def refund(self, payment_ref: str, amount: int) -> None:
        self.refunds.append((payment_ref, amount))

# part 3: the service: hold, confirm, cancel
class SeatUnavailable(Exception): ...


class HoldExpired(Exception): ...


@dataclass
class BookingService:
    shows: dict[str, Show]
    locks: SeatLockProvider
    payments: PaymentGateway
    hold_ttl: float = 600.0  # 10 minutes to pay
    bookings: dict[str, Booking] = field(default_factory=dict)
    _sold: dict[str, set[str]] = field(default_factory=dict)  # show_id -> seats sold for good
    _ids: itertools.count = field(default_factory=lambda: itertools.count(1))
    _mu: threading.Lock = field(default_factory=threading.Lock)

    def hold(self, user: str, show_id: str, seat_ids: list[str], now: float) -> Booking:
        show = self.shows[show_id]
        seats = sorted(set(seat_ids))
        if not seats or any(s not in show.seats for s in seats):
            raise ValueError("unknown seat")
        with self._mu:  # the sold check and the lock must not interleave with a confirm
            if self._sold.get(show_id, set()) & set(seats):
                raise SeatUnavailable("already sold")
            booking_id = f"BK{next(self._ids):05d}"
            if not self.locks.lock(show_id, seats, booking_id, self.hold_ttl, now):
                raise SeatUnavailable("held by someone else")
        amount = sum(show.prices[show.seats[s].type] for s in seats)
        b = Booking(booking_id, user, show_id, tuple(seats), amount, now + self.hold_ttl)
        self.bookings[booking_id] = b
        return b

    def confirm(self, booking_id: str, payment_ref: str, now: float) -> Booking:
        """Called when the gateway reports a successful payment for this booking."""
        b = self.bookings[booking_id]
        with self._mu:  # two callbacks for one booking must not both pass the checks
            if b.status is BookingStatus.CONFIRMED and b.payment_ref == payment_ref:
                return b  # the gateway retried its callback: same booking, nothing new
            if b.status is BookingStatus.HELD and self.locks.holds(b.show_id, list(b.seat_ids), booking_id, now):
                self._sold.setdefault(b.show_id, set()).update(b.seat_ids)
                b.status, b.payment_ref = BookingStatus.CONFIRMED, payment_ref
                self.locks.unlock(b.show_id, list(b.seat_ids), booking_id)  # sold now; the hold is done
                return b
            if b.status is BookingStatus.HELD:
                b.status = BookingStatus.EXPIRED
        self.payments.refund(payment_ref, b.amount)  # money arrived for seats we no longer hold
        raise HoldExpired(f"{b.status.value}: refunded, pick seats again")

    def cancel_hold(self, booking_id: str) -> None:
        b = self.bookings[booking_id]
        if b.status is BookingStatus.HELD:
            b.status = BookingStatus.CANCELLED
            self.locks.unlock(b.show_id, list(b.seat_ids), booking_id)

    def available(self, show_id: str, now: float) -> list[str]:
        sold = self._sold.get(show_id, set())
        return [s for s in self.shows[show_id].seats if s not in sold and not self.locks.is_held(show_id, s, now)]

# part 4: a race for the same two seats, then an expired hold
if __name__ == "__main__":
    seats = {f"{r}{n}": Seat(f"{r}{n}", SeatType.PREMIUM if r == "A" else SeatType.REGULAR) for r in "AB" for n in range(1, 5)}
    show = Show("S1", "Sholay", "Screen 3", 0.0, seats, {SeatType.REGULAR: 200, SeatType.PREMIUM: 350})
    gateway = FakeGateway()
    svc = BookingService({"S1": show}, InMemorySeatLocks(), gateway)

    wins: list[Booking] = []
    losers: list[str] = []

    def try_book(user: str) -> None:
        try:
            wins.append(svc.hold(user, "S1", ["A2", "A3"], now=0.0))
        except SeatUnavailable:
            losers.append(user)

    ts = [threading.Thread(target=try_book, args=(f"user{i}",)) for i in range(50)]
    for t in ts:
        t.start()
    for t in ts:
        t.join()
    print(len(wins), "hold,", len(losers), "told no")  # 1 hold, 49 told no
    print("free while held:", svc.available("S1", now=60.0))  # A2, A3 are held
    b = svc.confirm(wins[0].booking_id, "pay_101", now=120.0)
    print(b.status.value, b.amount, b.payment_ref)  # confirmed 700 pay_101
    print(svc.confirm(b.booking_id, "pay_101", now=130.0).status.value)  # confirmed: the retry is a no-op

    late = svc.hold("riya", "S1", ["B1"], now=200.0)
    try:
        svc.confirm(late.booking_id, "pay_102", now=200.0 + 601)  # paid 1 s after the 10-minute hold
    except HoldExpired as e:
        print("riya:", e, gateway.refunds)  # expired: refunded ... [('pay_102', 200)]
    print("free:", svc.available("S1", now=900.0))  # B1 is free again; A2, A3 are sold

1 hold, 49 told no
free while held: ['A1', 'A4', 'B1', 'B2', 'B3', 'B4']
confirmed 700 pay_101
confirmed
riya: expired: refunded, pick seats again [('pay_102', 200)]
free: ['A1', 'A4', 'B1', 'B2', 'B3', 'B4']