b. Holds & Locking

package boxoffice.inventory;

import java.time.Duration;
import java.time.OffsetDateTime;
import java.time.ZoneOffset;
import java.util.List;
import java.util.UUID;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.http.HttpStatus;
import org.springframework.jdbc.core.simple.JdbcClient;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.server.ResponseStatusException;

/**
 * Holds: "these seats are mine for 10 minutes". The one rule that must never break:
 * a seat belongs to at most ONE live hold. Three ways to (try to) enforce it are
 * selectable with INVENTORY_LOCKING, so you can measure each one.
 */
@Service
public class HoldService {

    private static final Logger log = LoggerFactory.getLogger(HoldService.class);

    private record Seat(long id, String status, int version) { }

    private final JdbcClient jdbc;
    private final String locking;
    private final Duration holdTtl;
    private final int maxSeats;

    public HoldService(JdbcClient jdbc,
                       @Value("${inventory.locking}") String locking,
                       @Value("${inventory.hold-ttl}") Duration holdTtl,
                       @Value("${inventory.max-seats-per-hold}") int maxSeats) {
        this.jdbc = jdbc;
        this.locking = locking;
        this.holdTtl = holdTtl;
        this.maxSeats = maxSeats;
        log.info("seat locking strategy: {}", locking);
    }

    @Transactional        // READ COMMITTED (PostgreSQL's default), on the primary
    public HoldOut hold(String userId, HoldIn in) {
        var seatIds = in.seatIds().stream().distinct().sorted().toList();
        if (seatIds.size() > maxSeats) {
            throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "at most " + maxSeats + " seats per hold");
        }
        var holdId = UUID.randomUUID();
        var expiresAt = OffsetDateTime.now(ZoneOffset.UTC).plus(holdTtl);

        switch (locking) {
            case "naive" -> holdNaive(in.eventId(), seatIds, holdId);
            case "optimistic" -> holdOptimistic(in.eventId(), seatIds, holdId);
            default -> holdPessimistic(in.eventId(), seatIds, holdId);
        }

        jdbc.sql("""
                INSERT INTO inventory.holds (id, event_id, user_id, status, expires_at)
                VALUES (:id, :eventId, :userId, 'ACTIVE', :expiresAt)""")
                .param("id", holdId).param("eventId", in.eventId())
                .param("userId", userId).param("expiresAt", expiresAt)
                .update();
        for (var seatId : seatIds) {
            jdbc.sql("INSERT INTO inventory.hold_seats (hold_id, seat_id) VALUES (:holdId, :seatId)")
                    .param("holdId", holdId).param("seatId", seatId)
                    .update();
        }
        return new HoldOut(holdId, in.eventId(), "ACTIVE", expiresAt, seatIds);
    }

    // ---- strategy 1: check-then-act, no lock (WRONG on purpose) ---------------
    private void holdNaive(long eventId, List<Long> seatIds, UUID holdId) {
        var seats = read(eventId, seatIds, "");
        requireAvailable(seats, seatIds);
        // race window: another transaction may have read the same AVAILABLE seats and
        // be about to run this same UPDATE — both will "win"
        jdbc.sql("UPDATE inventory.seats SET status = 'HELD', hold_id = :holdId WHERE id IN (:ids)")
                .param("holdId", holdId).param("ids", seatIds)
                .update();
    }

    // ---- strategy 2: pessimistic, SELECT ... FOR UPDATE -----------------------
    private void holdPessimistic(long eventId, List<Long> seatIds, UUID holdId) {
        // row locks until commit; a second transaction on the same seats WAITS here.
        // Locking in id order (ORDER BY id) prevents deadlocks between overlapping holds.
        var seats = read(eventId, seatIds, "ORDER BY id FOR UPDATE");
        requireAvailable(seats, seatIds);
        jdbc.sql("UPDATE inventory.seats SET status = 'HELD', hold_id = :holdId, version = version + 1 WHERE id IN (:ids)")
                .param("holdId", holdId).param("ids", seatIds)
                .update();
    }

    // ---- strategy 3: optimistic, compare-and-set on a version column ---------------
    private void holdOptimistic(long eventId, List<Long> seatIds, UUID holdId) {
        var seats = read(eventId, seatIds, "");                      // no lock
        requireAvailable(seats, seatIds);
        for (var seat : seats) {
            int changed = jdbc.sql("""
                    UPDATE inventory.seats
                    SET status = 'HELD', hold_id = :holdId, version = version + 1
                    WHERE id = :id AND version = :version AND status = 'AVAILABLE'""")
                    .param("holdId", holdId).param("id", seat.id()).param("version", seat.version())
                    .update();
            if (changed == 0) {
                // someone changed this seat after we read it: give up (the exception rolls back)
                throw new ResponseStatusException(HttpStatus.CONFLICT, "seat " + seat.id() + " was just taken");
            }
        }
    }

    private List<Seat> read(long eventId, List<Long> seatIds, String suffix) {
        return jdbc.sql("SELECT id, status, version FROM inventory.seats "
                        + "WHERE event_id = :eventId AND id IN (:ids) " + suffix)
                .param("eventId", eventId).param("ids", seatIds)
                .query((rs, n) -> new Seat(rs.getLong("id"), rs.getString("status"), rs.getInt("version")))
                .list();
    }

    private static void requireAvailable(List<Seat> seats, List<Long> seatIds) {
        if (seats.size() != seatIds.size()) {
            throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "some seats do not belong to this event");
        }
        if (seats.stream().anyMatch(s -> !"AVAILABLE".equals(s.status()))) {
            throw new ResponseStatusException(HttpStatus.CONFLICT, "some seats are not available");
        }
    }

    // ---- reading, confirming and releasing a hold (always on the primary) -----

    @Transactional        // NOT readOnly: read-your-writes, the user just created it
    public HoldOut find(String userId, UUID holdId) {
        var hold = jdbc.sql("""
                SELECT id, event_id, status, expires_at FROM inventory.holds
                WHERE id = :id AND user_id = :userId""")
                .param("id", holdId).param("userId", userId)
                .query((rs, n) -> new HoldOut(rs.getObject("id", UUID.class), rs.getLong("event_id"),
                        rs.getString("status"), rs.getObject("expires_at", OffsetDateTime.class), List.of()))
                .optional()
                .orElseThrow(() -> new ResponseStatusException(HttpStatus.NOT_FOUND, "hold not found"));
        var seatIds = jdbc.sql("SELECT seat_id FROM inventory.hold_seats WHERE hold_id = :id ORDER BY seat_id")
                .param("id", holdId).query(Long.class).list();
        return new HoldOut(hold.id(), hold.eventId(), hold.status(), hold.expiresAt(), seatIds);
    }

    /** Payment succeeded (lab 7 calls this): seats become SOLD. */
    @Transactional
    public HoldOut confirm(String userId, UUID holdId) {
        int changed = jdbc.sql("""
                UPDATE inventory.holds SET status = 'CONFIRMED'
                WHERE id = :id AND user_id = :userId AND status = 'ACTIVE' AND expires_at > now()""")
                .param("id", holdId).param("userId", userId)
                .update();
        if (changed == 0) {
            throw new ResponseStatusException(HttpStatus.CONFLICT, "hold is not active (expired, released or unknown)");
        }
        jdbc.sql("UPDATE inventory.seats SET status = 'SOLD', version = version + 1 WHERE hold_id = :id")
                .param("id", holdId).update();
        return find(userId, holdId);
    }

    @Transactional
    public void release(String userId, UUID holdId) {
        int changed = jdbc.sql("""
                UPDATE inventory.holds SET status = 'RELEASED'
                WHERE id = :id AND user_id = :userId AND status = 'ACTIVE'""")
                .param("id", holdId).param("userId", userId)
                .update();
        if (changed == 0) {
            throw new ResponseStatusException(HttpStatus.NOT_FOUND, "no active hold " + holdId);
        }
        freeSeatsOf(holdId);
    }

    private void freeSeatsOf(UUID holdId) {
        jdbc.sql("""
                UPDATE inventory.seats SET status = 'AVAILABLE', hold_id = NULL, version = version + 1
                WHERE hold_id = :id AND status = 'HELD'""")
                .param("id", holdId).update();
    }

    /**
     * Every 5 s, expire old holds and free their seats. FOR UPDATE SKIP LOCKED lets several
     * inventory replicas run this job at the same time without processing the same hold twice.
     */
    @Scheduled(fixedDelay = 5000)
    @Transactional
    public void expireHolds() {
        int freed = jdbc.sql("""
                WITH expired AS (
                    UPDATE inventory.holds SET status = 'EXPIRED'
                    WHERE id IN (SELECT id FROM inventory.holds
                                 WHERE status = 'ACTIVE' AND expires_at < now()
                                 ORDER BY expires_at
                                 LIMIT 500
                                 FOR UPDATE SKIP LOCKED)
                    RETURNING id)
                UPDATE inventory.seats
                SET status = 'AVAILABLE', hold_id = NULL, version = version + 1
                WHERE status = 'HELD' AND hold_id IN (SELECT id FROM expired)""")
                .update();
        if (freed > 0) {
            log.info("expired holds freed {} seats", freed);
        }
    }
}

package boxoffice.inventory;

import java.util.List;
import org.springframework.jdbc.core.simple.JdbcClient;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

/** Read-only queries: they run on the REPLICA (readOnly = true), so they may be slightly stale. */
@Service
public class SeatQueries {

    private final JdbcClient jdbc;

    public SeatQueries(JdbcClient jdbc) {
        this.jdbc = jdbc;
    }

    @Transactional(readOnly = true)
    public List<SectionAvailability> availability(long eventId) {
        return jdbc.sql("""
                SELECT section,
                       count(*) FILTER (WHERE status = 'AVAILABLE') AS available,
                       count(*) FILTER (WHERE status = 'HELD')      AS held,
                       count(*) FILTER (WHERE status = 'SOLD')      AS sold
                FROM inventory.seats
                WHERE event_id = :eventId
                GROUP BY section
                ORDER BY section""")
                .param("eventId", eventId)
                .query((rs, n) -> new SectionAvailability(rs.getString("section"),
                        rs.getLong("available"), rs.getLong("held"), rs.getLong("sold")))
                .list();
    }

    @Transactional(readOnly = true)
    public List<SeatOut> seats(long eventId, String section) {
        return jdbc.sql("""
                SELECT id, section, seat_row, seat_number, price, status
                FROM inventory.seats
                WHERE event_id = :eventId AND section = :section
                ORDER BY seat_row, seat_number""")
                .param("eventId", eventId)
                .param("section", section)
                .query((rs, n) -> new SeatOut(rs.getLong("id"), rs.getString("section"),
                        rs.getInt("seat_row"), rs.getInt("seat_number"),
                        rs.getBigDecimal("price"), rs.getString("status")))
                .list();
    }

    /** Seats that belong to more than one live hold: must always be 0. Reads the PRIMARY. */
    @Transactional
    public long doubleBookedSeats() {
        return jdbc.sql("""
                SELECT count(*) FROM (
                    SELECT hs.seat_id
                    FROM inventory.hold_seats hs
                    JOIN inventory.holds h ON h.id = hs.hold_id
                    WHERE h.status IN ('ACTIVE', 'CONFIRMED')
                    GROUP BY hs.seat_id
                    HAVING count(*) > 1) AS doubles""")
                .query(Long.class)
                .single();
    }
}

package boxoffice.inventory;

import jakarta.validation.Valid;
import java.net.URI;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.DeleteMapping;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestHeader;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.ResponseStatus;
import org.springframework.web.bind.annotation.RestController;

@RestController
@RequestMapping("/inventory")
public class InventoryController {

    private final SeatQueries queries;
    private final HoldService holds;

    public InventoryController(SeatQueries queries, HoldService holds) {
        this.queries = queries;
        this.holds = holds;
    }

    // ---- open (GET, through the gateway without a token), served by the replica ----

    @GetMapping("/events/{eventId}/availability")
    public List<SectionAvailability> availability(@PathVariable long eventId) {
        return queries.availability(eventId);
    }

    @GetMapping("/events/{eventId}/seats")
    public List<SeatOut> seats(@PathVariable long eventId, @RequestParam String section) {
        return queries.seats(eventId, section);
    }

    // ---- protected: the gateway sets X-User-Id from the JWT ----

    @PostMapping("/holds")
    public ResponseEntity<HoldOut> hold(@RequestHeader("X-User-Id") String userId, @Valid @RequestBody HoldIn in) {
        var out = holds.hold(userId, in);
        return ResponseEntity.created(URI.create("/inventory/holds/" + out.id())).body(out);
    }

    @GetMapping("/holds/{id}")
    public HoldOut find(@RequestHeader("X-User-Id") String userId, @PathVariable UUID id) {
        return holds.find(userId, id);
    }

    @PostMapping("/holds/{id}/confirm")
    public HoldOut confirm(@RequestHeader("X-User-Id") String userId, @PathVariable UUID id) {
        return holds.confirm(userId, id);
    }

    @DeleteMapping("/holds/{id}")
    @ResponseStatus(HttpStatus.NO_CONTENT)
    public void release(@RequestHeader("X-User-Id") String userId, @PathVariable UUID id) {
        holds.release(userId, id);
    }

    /** The invariant check used after a flash-sale test: must be {"doubleBookedSeats": 0}. */
    @GetMapping("/audit/double-bookings")
    public Map<String, Long> audit() {
        return Map.of("doubleBookedSeats", queries.doubleBookedSeats());
    }
}

package boxoffice.gateway;

import java.util.List;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.http.HttpMethod;
import org.springframework.security.config.Customizer;
import org.springframework.security.config.annotation.web.reactive.EnableWebFluxSecurity;
import org.springframework.security.config.web.server.ServerHttpSecurity;
import org.springframework.security.oauth2.core.DelegatingOAuth2TokenValidator;
import org.springframework.security.oauth2.jwt.JwtClaimValidator;
import org.springframework.security.oauth2.jwt.JwtValidators;
import org.springframework.security.oauth2.jwt.NimbusReactiveJwtDecoder;
import org.springframework.security.oauth2.jwt.ReactiveJwtDecoder;
import org.springframework.security.web.server.SecurityWebFilterChain;

@Configuration
@EnableWebFluxSecurity
public class SecurityConfig {

    /** Which routes are open, which need a valid token, which need the admin scope. */
    @Bean
    SecurityWebFilterChain security(ServerHttpSecurity http) {
        return http
                .csrf(ServerHttpSecurity.CsrfSpec::disable)        // stateless API, no cookies
                .authorizeExchange(ex -> ex
                        // open: getting a token, and the public keys to check one
                        .pathMatchers("/auth/register", "/auth/login", "/auth/.well-known/**").permitAll()
                        // open: browsing the catalog (HEAD is a GET without the body)
                        .pathMatchers(HttpMethod.GET, "/catalog/**").permitAll()
                        .pathMatchers(HttpMethod.HEAD, "/catalog/**").permitAll()
                        // admin only: changing the catalog (scope claim "user admin")
                        .pathMatchers("/catalog/**").hasAuthority("SCOPE_admin")
                        // open: seat maps and availability (lab 6); holds need a token
                        .pathMatchers(HttpMethod.GET, "/inventory/events/**").permitAll()
                        .pathMatchers(HttpMethod.HEAD, "/inventory/events/**").permitAll()
                        // health, metrics, routes: the edge never forwards /actuator from outside
                        .pathMatchers("/actuator/**").permitAll()
                        // everything else (and every future route): any valid token
                        .anyExchange().authenticated())
                .oauth2ResourceServer(oauth -> oauth.jwt(Customizer.withDefaults()))
                .build();
    }

    /** Signature from the JWKS of auth, plus expiry, issuer and audience checks. */
    @Bean
    ReactiveJwtDecoder jwtDecoder(@Value("${spring.security.oauth2.resourceserver.jwt.jwk-set-uri}") String jwksUri,
                                  @Value("${jwt.issuer}") String issuer,
                                  @Value("${jwt.audience}") String audience) {
        var decoder = NimbusReactiveJwtDecoder.withJwkSetUri(jwksUri).build();
        decoder.setJwtValidator(new DelegatingOAuth2TokenValidator<>(
                JwtValidators.createDefaultWithIssuer(issuer),     // exp, nbf, iss
                new JwtClaimValidator<List<String>>("aud", aud -> aud != null && aud.contains(audience))));
        return decoder;
    }
}