2. Transactional Outbox

package boxoffice.order;

import jakarta.validation.constraints.Email;
import jakarta.validation.constraints.NotBlank;
import jakarta.validation.constraints.NotNull;
import jakarta.validation.constraints.Positive;
import java.math.BigDecimal;

public record OrderRequest(
        @NotBlank String showId,
        @NotBlank String holdId,
        @NotBlank @Email String customerEmail,
        @NotNull @Positive BigDecimal amount) {
}

package boxoffice.order;

import java.math.BigDecimal;
import java.time.OffsetDateTime;
import java.util.UUID;

public record OrderView(
        UUID id,
        String showId,
        String holdId,
        String customerEmail,
        BigDecimal amount,
        String status,
        OffsetDateTime createdAt) {
}

package boxoffice.order;

import java.math.BigDecimal;
import java.time.Instant;
import java.util.UUID;

/** The fact published on order-events. eventId is unique per event: consumers dedupe by it. */
public record OrderEvent(
        UUID eventId,
        String eventType,
        UUID orderId,
        String showId,
        String holdId,
        String customerEmail,
        BigDecimal amount,
        Instant occurredAt) {

    public static OrderEvent placed(OrderView order) {
        return new OrderEvent(UUID.randomUUID(), "OrderPlaced", order.id(), order.showId(),
                order.holdId(), order.customerEmail(), order.amount(), Instant.now());
    }
}

package boxoffice.order;

import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.context.properties.bind.DefaultValue;

@ConfigurationProperties(prefix = "boxoffice.order")
public record OrderProperties(
        @DefaultValue("outbox") NotifyMode notifyMode,
        @DefaultValue("http://notification:8086") String notificationUrl,
        @DefaultValue("order-events") String topic,
        @DefaultValue("200") long relayIntervalMs) {

    public enum NotifyMode { SYNC, OUTBOX }
}

package boxoffice.order;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.context.properties.ConfigurationPropertiesScan;
import org.springframework.scheduling.annotation.EnableScheduling;

@SpringBootApplication
@EnableScheduling
@ConfigurationPropertiesScan
public class OrderApplication {

    public static void main(String[] args) {
        SpringApplication.run(OrderApplication.class, args);
    }
}

package boxoffice.order;

import boxoffice.order.OrderProperties.NotifyMode;
import boxoffice.order.outbox.Outbox;
import java.util.Optional;
import java.util.UUID;
import org.springframework.jdbc.core.simple.JdbcClient;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

@Service
public class OrderService {

    public record PlaceResult(OrderView order, boolean created) {
    }

    private static final String SELECT = """
            SELECT id, show_id, hold_id, customer_email, amount, status, created_at
            FROM orders
            """;

    private final JdbcClient jdbc;
    private final Outbox outbox;
    private final NotificationClient notifications;
    private final OrderProperties props;

    public OrderService(JdbcClient jdbc, Outbox outbox, NotificationClient notifications,
                        OrderProperties props) {
        this.jdbc = jdbc;
        this.outbox = outbox;
        this.notifications = notifications;
        this.props = props;
    }

    @Transactional
    public PlaceResult place(OrderRequest req) {
        int inserted = jdbc.sql("""
                INSERT INTO orders (id, show_id, hold_id, customer_email, amount, status)
                VALUES (?, ?, ?, ?, ?, 'PENDING_PAYMENT')
                ON CONFLICT (hold_id) DO NOTHING
                """)
                .params(UUID.randomUUID(), req.showId(), req.holdId(), req.customerEmail(), req.amount())
                .update();

        OrderView order = jdbc.sql(SELECT + " WHERE hold_id = ?")
                .param(req.holdId())
                .query(OrderView.class)
                .single();

        if (inserted == 0) {
            // same hold already ordered: idempotent replay, no second event
            return new PlaceResult(order, false);
        }

        OrderEvent event = OrderEvent.placed(order);
        if (props.notifyMode() == NotifyMode.SYNC) {
            notifications.send(event); // BEFORE: the fan waits for the e-mail
        } else {
            outbox.append(event.eventId(), "order", order.id(), event.eventType(), event);
        }
        return new PlaceResult(order, true);
    }

    public Optional<OrderView> find(UUID id) {
        return jdbc.sql(SELECT + " WHERE id = ?").param(id).query(OrderView.class).optional();
    }
}

package boxoffice.order;

import org.springframework.stereotype.Component;
import org.springframework.web.client.RestClient;

/** Only used in NOTIFY_MODE=sync: the "before" architecture. */
@Component
public class NotificationClient {

    private final RestClient http;

    public NotificationClient(RestClient.Builder builder, OrderProperties props) {
        this.http = builder.baseUrl(props.notificationUrl()).build();
    }

    public void send(OrderEvent event) {
        http.post().uri("/notifications").body(event).retrieve().toBodilessEntity();
    }
}

package boxoffice.order;

import jakarta.validation.Valid;
import java.net.URI;
import java.util.UUID;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
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.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.server.ResponseStatusException;

@RestController
@RequestMapping("/orders")
public class OrderController {

    private final OrderService orders;

    public OrderController(OrderService orders) {
        this.orders = orders;
    }

    @PostMapping
    public ResponseEntity<OrderView> place(@Valid @RequestBody OrderRequest req) {
        OrderService.PlaceResult result = orders.place(req);
        if (!result.created()) {
            return ResponseEntity.ok(result.order());
        }
        return ResponseEntity.created(URI.create("/orders/" + result.order().id())).body(result.order());
    }

    @GetMapping("/{id}")
    public OrderView get(@PathVariable UUID id) {
        return orders.find(id).orElseThrow(() -> new ResponseStatusException(HttpStatus.NOT_FOUND));
    }
}

package boxoffice.order.outbox;

import tools.jackson.databind.ObjectMapper;
import java.util.UUID;
import org.springframework.jdbc.core.simple.JdbcClient;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

@Component
public class Outbox {

    private final JdbcClient jdbc;
    private final ObjectMapper mapper;

    public Outbox(JdbcClient jdbc, ObjectMapper mapper) {
        this.jdbc = jdbc;
        this.mapper = mapper;
    }

    /** MANDATORY: fails fast if someone calls it outside the business transaction. */
    @Transactional(propagation = Propagation.MANDATORY)
    public void append(UUID eventId, String aggregateType, UUID aggregateId, String eventType, Object payload) {
        String json = mapper.writeValueAsString(payload); // Jackson 3: exceptions are unchecked
        jdbc.sql("""
                INSERT INTO outbox (id, aggregate_type, aggregate_id, event_type, payload)
                VALUES (?, ?, ?, ?, ?::jsonb)
                """)
                .params(eventId, aggregateType, aggregateId, eventType, json)
                .update();
    }
}

package boxoffice.order.outbox;

import boxoffice.order.OrderProperties;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.simple.JdbcClient;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

@Component
public class OutboxRelay {

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

    record Pending(UUID id, UUID aggregateId, String eventType, String payload) {
    }

    private final JdbcClient jdbc;
    private final JdbcTemplate jdbcTemplate;
    private final KafkaTemplate<String, String> kafka;
    private final String topic;

    public OutboxRelay(JdbcClient jdbc, JdbcTemplate jdbcTemplate,
                       KafkaTemplate<String, String> kafka, OrderProperties props) {
        this.jdbc = jdbc;
        this.jdbcTemplate = jdbcTemplate;
        this.kafka = kafka;
        this.topic = props.topic();
    }

    @Scheduled(fixedDelayString = "${boxoffice.order.relay-interval-ms:200}")
    @Transactional
    public void relay() throws Exception {
        // SKIP LOCKED: several order replicas can relay in parallel without taking the same rows
        List<Pending> batch = jdbc.sql("""
                SELECT id, aggregate_id, event_type, payload::text AS payload
                FROM outbox
                WHERE published_at IS NULL
                ORDER BY created_at
                LIMIT 100
                FOR UPDATE SKIP LOCKED
                """)
                .query(Pending.class)
                .list();
        if (batch.isEmpty()) {
            return;
        }

        List<CompletableFuture<SendResult<String, String>>> acks = new ArrayList<>();
        for (Pending p : batch) {
            ProducerRecord<String, String> record =
                    new ProducerRecord<>(topic, p.aggregateId().toString(), p.payload());
            record.headers().add("eventId", p.id().toString().getBytes(StandardCharsets.UTF_8));
            record.headers().add("eventType", p.eventType().getBytes(StandardCharsets.UTF_8));
            acks.add(kafka.send(record));
        }
        // wait for every ack; on failure the transaction rolls back and the rows are retried
        CompletableFuture.allOf(acks.toArray(new CompletableFuture<?>[0])).get(10, TimeUnit.SECONDS);

        jdbcTemplate.batchUpdate("UPDATE outbox SET published_at = clock_timestamp() WHERE id = ?",
                batch.stream().map(p -> new Object[] {p.id()}).toList());
        log.info("relayed {} event(s) to {}", batch.size(), topic);
    }

    @Scheduled(cron = "0 0 * * * *")
    @Transactional
    public void purge() {
        int n = jdbcTemplate.update(
                "DELETE FROM outbox WHERE published_at < now() - interval '1 day'");
        log.info("purged {} published outbox row(s)", n);
    }
}