Outbox-und-Idempotenz-Lab

package com.example.outboxdeepdive.demo;
import com.example.outboxdeepdive.billing.*;
import com.example.outboxdeepdive.broker.*;
import com.example.outboxdeepdive.order.*;
import com.example.outboxdeepdive.outbox.*;
import com.example.outboxdeepdive.shared.*;
import java.time.Instant;

public final class Run4C3Demo {
    public static void main(String[] args) {
        var orders = new InMemoryOrderRepository();
        var outbox = new InMemoryOutboxRepository();
        var broker = new FakeBroker();
        var service = new PlaceOrderService(orders, outbox);
        var publisher = new OutboxPublisher(outbox, broker, 10, 3);
        Instant now = Instant.parse("2026-01-01T10:00:00Z");
        service.placeOrder("O-100", Money.eur("129.90"), now);
        broker.failNextPublishes(1);
        publisher.publishDue(now); // failed, retry scheduled
        publisher.publishDue(now.plusSeconds(5)); // still not due
        publisher.publishDue(now.plusSeconds(20)); // sent
        var idem = new InMemoryIdempotencyStore();
        var invoices = new InvoiceRepository();
        var consumer = new BillingConsumer(idem, invoices);
        var event = broker.published().getFirst();
        consumer.onMessage(event);
        consumer.onMessage(event); // duplicate delivery must not create second invoice
        System.out.println("RUN4C3_DEMO_OK orders=" + orders.size() + " outbox=" + outbox.all().size() + " broker=" + broker.published().size() + " invoices=" + invoices.invoices().size());
    }
}

package com.example.outboxdeepdive.order;
import com.example.outboxdeepdive.outbox.*;
import com.example.outboxdeepdive.shared.*;
import java.time.Instant;

// Pattern: Application Service - definiert Use-Case- und Transaktionsgrenze.
public final class PlaceOrderService {
    private final OrderRepository orders;
    private final OutboxRepository outbox;
    public PlaceOrderService(OrderRepository orders, OutboxRepository outbox) { this.orders = orders; this.outbox = outbox; }

    public OrderId placeOrder(String id, Money total, Instant now) {
        Order order = Order.accept(OrderId.of(id), total, now);
        // In einer echten Datenbank waere dies eine Transaktion:
        // 1) Order speichern
        // 2) OutboxMessage speichern
        // 3) Commit
        orders.save(order);
        outbox.save(OutboxMessage.pending("Order", order.id().value(), "OrderAccepted", 1, payload(order), now));
        return order.id();
    }
    private String payload(Order order) {
        return "{"orderId":"" + order.id().value() + "","amount":"" + order.total().amount() + "","currency":"" + order.total().currency() + ""}";
    }
}

package com.example.outboxdeepdive.outbox;
import java.time.Instant;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

public final class InMemoryOutboxRepository implements OutboxRepository {
    private final Map<UUID, OutboxMessage> messages = new ConcurrentHashMap<>();
    public void save(OutboxMessage message) { messages.put(message.id(), message); }
    public List<OutboxMessage> findDue(Instant now, int limit) {
        return messages.values().stream()
                .filter(m -> m.dueAt(now))
                .sorted(Comparator.comparing(OutboxMessage::createdAt))
                .limit(limit)
                .toList();
    }
    public Optional<OutboxMessage> findById(UUID id) { return Optional.ofNullable(messages.get(id)); }
    public List<OutboxMessage> all() { return messages.values().stream().sorted(Comparator.comparing(OutboxMessage::createdAt)).toList(); }
}

package com.example.outboxdeepdive.outbox;
import com.example.outboxdeepdive.broker.*;
import java.time.*;

// Pattern: Publisher - entkoppelt DB-Transaktion und Broker-Versand.
public final class OutboxPublisher {
    private final OutboxRepository repository;
    private final BrokerPort broker;
    private final int batchSize;
    private final int maxAttempts;

    public OutboxPublisher(OutboxRepository repository, BrokerPort broker, int batchSize, int maxAttempts) {
        this.repository = repository; this.broker = broker; this.batchSize = batchSize; this.maxAttempts = maxAttempts;
    }
    public int publishDue(Instant now) {
        int sent = 0;
        for (OutboxMessage message : repository.findDue(now, batchSize)) {
            message.claim();
            try {
                broker.publish(new EventEnvelope(message.id(), message.eventType(), message.eventVersion(), message.payload()));
                message.markSent();
                sent++;
            } catch (RuntimeException ex) {
                message.markFailed(ex.getClass().getSimpleName() + ": " + ex.getMessage(), nextAttempt(now, message.attempts() + 1), maxAttempts);
            }
        }
        return sent;
    }
    private Instant nextAttempt(Instant now, int attempt) {
        long seconds = Math.min(300, (long)Math.pow(2, Math.max(1, attempt)));
        return now.plusSeconds(seconds);
    }
    public void recoverInFlightAfterCrash() {
        repository.all().forEach(OutboxMessage::requeueAfterPublisherCrash);
    }
}

package com.example.outboxdeepdive.broker;
import java.util.*;

// Pattern: Test Fake - kontrollierbarer Broker fuer Crash- und Retry-Tests.
public final class FakeBroker implements BrokerPort {
    private final List<EventEnvelope> published = new ArrayList<>();
    private int failNextPublishes;
    public void failNextPublishes(int count) { this.failNextPublishes = count; }
    public void publish(EventEnvelope envelope) {
        if (failNextPublishes > 0) { failNextPublishes--; throw new BrokerUnavailableException("broker temporarily unavailable"); }
        published.add(envelope);
    }
    public List<EventEnvelope> published() { return List.copyOf(published); }
}

package com.example.outboxdeepdive.billing;
import com.example.outboxdeepdive.broker.EventEnvelope;

// Pattern: Idempotent Consumer - verarbeitet dasselbe Event mehrfach ohne doppelte Rechnung.
public final class BillingConsumer {
    private final IdempotencyStore idempotencyStore;
    private final InvoiceRepository invoices;
    public BillingConsumer(IdempotencyStore idempotencyStore, InvoiceRepository invoices) { this.idempotencyStore = idempotencyStore; this.invoices = invoices; }

    public void onMessage(EventEnvelope envelope) {
        if (idempotencyStore.alreadyProcessed(envelope.messageId())) return;
        String orderId = extractOrderId(envelope.payload());
        invoices.createInvoice(orderId);
        idempotencyStore.markProcessed(envelope.messageId());
    }
    private String extractOrderId(String json) {
        String marker = ""orderId":"";
        int start = json.indexOf(marker);
        if (start < 0) throw new IllegalArgumentException("orderId missing");
        start += marker.length();
        int end = json.indexOf('"', start);
        return json.substring(start, end);
    }
}

package com.example.outboxdeepdive;
import com.example.outboxdeepdive.billing.*;
import com.example.outboxdeepdive.broker.*;
import com.example.outboxdeepdive.order.*;
import com.example.outboxdeepdive.outbox.*;
import com.example.outboxdeepdive.shared.*;
import java.time.Instant;

public final class Run4C3Tests {
    public static void main(String[] args) {
        orderAndOutboxAreWrittenTogether();
        brokerFailureKeepsMessageRetryable();
        duplicateDeliveryCreatesOnlyOneInvoice();
        maxAttemptsMovesToDeadLetter();
        System.out.println("RUN4C3_TESTS_OK");
    }
    static void orderAndOutboxAreWrittenTogether() {
        var orders = new InMemoryOrderRepository(); var outbox = new InMemoryOutboxRepository();
        new PlaceOrderService(orders, outbox).placeOrder("O-1", Money.eur("10.00"), Instant.parse("2026-01-01T00:00:00Z"));
        check(orders.size() == 1, "order saved"); check(outbox.all().size() == 1, "outbox saved");
        check(outbox.all().getFirst().status() == OutboxStatus.PENDING, "pending");
    }
    static void brokerFailureKeepsMessageRetryable() {
        var outbox = new InMemoryOutboxRepository(); var broker = new FakeBroker();
        outbox.save(OutboxMessage.pending("Order", "O-2", "OrderAccepted", 1, "{"orderId":"O-2"}", Instant.parse("2026-01-01T00:00:00Z")));
        broker.failNextPublishes(1);
        var publisher = new OutboxPublisher(outbox, broker, 10, 3);
        publisher.publishDue(Instant.parse("2026-01-01T00:00:00Z"));
        check(outbox.all().getFirst().status() == OutboxStatus.FAILED, "failed after broker error");
        publisher.publishDue(Instant.parse("2026-01-01T00:00:20Z"));
        check(outbox.all().getFirst().status() == OutboxStatus.SENT, "sent after retry");
    }
    static void duplicateDeliveryCreatesOnlyOneInvoice() {
        var idem = new InMemoryIdempotencyStore(); var invoices = new InvoiceRepository(); var consumer = new BillingConsumer(idem, invoices);
        var event = new EventEnvelope(java.util.UUID.randomUUID(), "OrderAccepted", 1, "{"orderId":"O-3"}");
        consumer.onMessage(event); consumer.onMessage(event);
        check(invoices.invoices().size() == 1, "only one invoice");
    }
    static void maxAttemptsMovesToDeadLetter() {
        var outbox = new InMemoryOutboxRepository(); var broker = new FakeBroker(); broker.failNextPublishes(5);
        outbox.save(OutboxMessage.pending("Order", "O-4", "OrderAccepted", 1, "{"orderId":"O-4"}", Instant.parse("2026-01-01T00:00:00Z")));
        var publisher = new OutboxPublisher(outbox, broker, 10, 2);
        publisher.publishDue(Instant.parse("2026-01-01T00:00:00Z"));
        publisher.publishDue(Instant.parse("2026-01-01T00:01:00Z"));
        check(outbox.all().getFirst().status() == OutboxStatus.DEAD_LETTER, "dead letter after max attempts");
    }
    static void check(boolean condition, String message) { if (!condition) throw new AssertionError(message); }
}