4. Event-Driven

package boxoffice.order;

import com.fasterxml.jackson.databind.ObjectMapper;
import java.time.Instant;
import java.util.Map;
import java.util.UUID;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

@Service
public class OrderService {

    private final JdbcTemplate jdbc;
    private final ObjectMapper json;

    public OrderService(JdbcTemplate jdbc, ObjectMapper json) {
        this.jdbc = jdbc;
        this.json = json;
    }

    @Transactional
    public UUID confirm(NewOrder cmd) throws Exception {
        UUID orderId = UUID.randomUUID();
        Instant now = Instant.now();
        jdbc.update("""
            INSERT INTO orders (id, event_id, seats, customer_email, status, confirmed_at)
            VALUES (?, ?, ?, ?, 'CONFIRMED', ?)
            """, orderId, cmd.eventId(), cmd.seats(), cmd.customerEmail(), java.sql.Timestamp.from(now));

        UUID eventId = UUID.randomUUID();
        Map<String, Object> envelope = Map.of(
            "eventId", eventId.toString(),
            "type", "OrderConfirmed",
            "version", 1,
            "occurredAt", now.toString(),
            "orderId", orderId.toString(),
            "customerEmail", cmd.customerEmail(),
            "seats", cmd.seats());
        jdbc.update("""
            INSERT INTO outbox (id, aggregate_type, aggregate_id, event_type, payload)
            VALUES (?, 'order', ?, 'OrderConfirmed', ?::jsonb)
            """, eventId, orderId.toString(), json.writeValueAsString(envelope));
        return orderId; // both rows commit together, or neither does
    }
}

package boxoffice.order;

import java.util.List;
import java.util.Map;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

@Component
public class OutboxRelay {

    private final JdbcTemplate jdbc;
    private final KafkaTemplate<String, String> kafka;

    public OutboxRelay(JdbcTemplate jdbc, KafkaTemplate<String, String> kafka) {
        this.jdbc = jdbc;
        this.kafka = kafka;
    }

    @Scheduled(fixedDelay = 200)   // requires @EnableScheduling on a configuration class
    @Transactional
    public void publishBatch() {
        List<Map<String, Object>> rows = jdbc.queryForList("""
            SELECT id, aggregate_id, payload::text AS payload
              FROM outbox
             WHERE published_at IS NULL
             ORDER BY created_at
             LIMIT 100
             FOR UPDATE SKIP LOCKED
            """);
        for (Map<String, Object> row : rows) {
            String key = (String) row.get("aggregate_id");
            String payload = (String) row.get("payload");
            kafka.send("order-events", key, payload).join();   // wait for the ack (acks=all)
            jdbc.update("UPDATE outbox SET published_at = now() WHERE id = ?", row.get("id"));
        }
    }
}

package boxoffice.notification;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.util.UUID;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

@Service
public class TicketEmailService {

    private final JdbcTemplate jdbc;
    private final ObjectMapper json;

    public TicketEmailService(JdbcTemplate jdbc, ObjectMapper json) {
        this.jdbc = jdbc;
        this.json = json;
    }

    @Transactional
    public void handle(String message) {
        JsonNode event = parse(message);
        UUID eventId = UUID.fromString(event.path("eventId").asText());
        int inserted = jdbc.update(
            "INSERT INTO processed_events (event_id) VALUES (?) ON CONFLICT DO NOTHING", eventId);
        if (inserted == 0) {
            return; // duplicate delivery: already processed, skip the side effect
        }
        jdbc.update("INSERT INTO ticket_emails (order_id, event_id, recipient) VALUES (?, ?, ?)",
            UUID.fromString(event.path("orderId").asText()), eventId, event.path("customerEmail").asText());
        // The real e-mail is sent from ticket_emails by a separate sender,
        // using event_id as the idempotency key with the e-mail provider.
    }

    private JsonNode parse(String message) {
        try {
            JsonNode event = json.readTree(message);
            if (!"OrderConfirmed".equals(event.path("type").asText()) || event.path("eventId").isMissingNode()) {
                throw new InvalidEventException("unexpected event: " + message);
            }
            return event;
        } catch (InvalidEventException e) {
            throw e;
        } catch (Exception e) {
            throw new InvalidEventException("malformed event: " + e.getMessage());
        }
    }
}

package boxoffice.notification;

import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.TopicPartition;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.TopicBuilder;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.kafka.support.ExponentialBackOffWithMaxRetries;
import org.springframework.stereotype.Component;

@Configuration
public class KafkaConfig {

    @Bean
    NewTopic orderEvents() {
        return TopicBuilder.name("order-events").partitions(3).replicas(1).build();
    }

    @Bean
    NewTopic orderEventsDlt() {
        return TopicBuilder.name("order-events.DLT").partitions(3).replicas(1).build();
    }

    // Picked up by Spring Boot's listener container factory.
    @Bean
    DefaultErrorHandler errorHandler(KafkaTemplate<String, String> template) {
        DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(template,
            (record, ex) -> new TopicPartition(record.topic() + ".DLT", record.partition()));
        ExponentialBackOffWithMaxRetries backOff = new ExponentialBackOffWithMaxRetries(3);
        backOff.setInitialInterval(500);
        backOff.setMultiplier(2.0);
        DefaultErrorHandler handler = new DefaultErrorHandler(recoverer, backOff);
        handler.addNotRetryableExceptions(InvalidEventException.class); // poison: straight to the DLT
        return handler;
    }

    @Component
    static class OrderEventsListener {
        private final TicketEmailService service;

        OrderEventsListener(TicketEmailService service) {
            this.service = service;
        }

        @KafkaListener(topics = "order-events", groupId = "notification")
        void onMessage(ConsumerRecord<String, String> record) {
            service.handle(record.value());
        }
    }
}