Transaktionen-und-ACID-Lab

package com.example.txdeepdive.application;

import com.example.txdeepdive.adapters.*;
import com.example.txdeepdive.domain.*;
import com.example.txdeepdive.ports.*;
import com.example.txdeepdive.shared.*;
import com.example.txdeepdive.tx.*;
import java.util.List;

public final class Run4C1Demo {
    public static void main(String[] args) {
        TransactionManager tx = new TransactionManager();
        InMemoryDatabase db = new InMemoryDatabase();
        ProductId product = new ProductId("P-100");
        db.putStock(product, 10);
        OrderRepository orders = new InMemoryOrderRepository(db, tx);
        OutboxRepository outbox = new InMemoryOutboxRepository(db, tx);
        InventoryPort inventory = new TransactionalInventoryAdapter(db, tx);
        InMemoryEventBroker broker = new InMemoryEventBroker();
        SafeOrderApplicationService service = new SafeOrderApplicationService(tx, orders, inventory, outbox);
        PlaceOrderCommand command = new PlaceOrderCommand("idem-1", OrderId.of("O-100"), new CustomerId("C-100"), List.of(new OrderLine(product, 2, Money.eur("19.90"))));
        var result = service.placeOrder(command);
        new OutboxPublisher(outbox, broker).publishPending();
        System.out.println("RUN4C1_DEMO_OK result=" + result.isOk() + " orders=" + orders.count() + " outbox=" + outbox.count() + " broker=" + broker.publishedCount());
        System.out.println(String.join(" | ", tx.log()));
    }
}

package com.example.txdeepdive.application;

import com.example.txdeepdive.domain.*;
import com.example.txdeepdive.ports.*;
import com.example.txdeepdive.tx.*;

// Anti-Pattern: Dual Write - Datenbank-Transaktion und Broker-Side-Effect werden unkoordiniert gemischt.
public final class BadDualWriteOrderService {
    private final TransactionManager txManager;
    private final OrderRepository orders;
    private final EventBroker broker;
    public BadDualWriteOrderService(TransactionManager txManager, OrderRepository orders, EventBroker broker) {
        this.txManager = txManager; this.orders = orders; this.broker = broker;
    }
    public void placeOrderAndCrashBeforeCommit(PlaceOrderCommand command) {
        txManager.inTransaction("bad-place-order", () -> {
            Order order = new Order(command.orderId(), command.customerId(), command.lines());
            orders.save(order);
            broker.publish("orders", "OrderPlaced:" + order.id()); // gefaehrlich: nicht rollbackfaehig
            throw new DualWriteFailure("crash after broker publish but before DB commit");
        });
    }
}

package com.example.txdeepdive.application;

import com.example.txdeepdive.domain.*;
import com.example.txdeepdive.outbox.OutboxMessage;
import com.example.txdeepdive.ports.*;
import com.example.txdeepdive.shared.*;
import com.example.txdeepdive.tx.TransactionManager;
import java.time.Instant;

// Pattern: Application Service - orchestriert den Use Case und definiert die Transaction Boundary.
public final class SafeOrderApplicationService {
    private final TransactionManager txManager;
    private final OrderRepository orders;
    private final InventoryPort inventory;
    private final OutboxRepository outbox;
    public SafeOrderApplicationService(TransactionManager txManager, OrderRepository orders, InventoryPort inventory, OutboxRepository outbox) {
        this.txManager = txManager; this.orders = orders; this.inventory = inventory; this.outbox = outbox;
    }
    public Result<PlaceOrderResult, DomainError> placeOrder(PlaceOrderCommand command) {
        return txManager.inTransaction("place-order", () -> {
            Order order = new Order(command.orderId(), command.customerId(), command.lines());
            for (OrderLine line : command.lines()) {
                Result<Void, DomainError> reserved = inventory.reserve(line.productId(), line.quantity());
                if (!reserved.isOk()) {
                    txManager.required().setRollbackOnly();
                    return Result.err(reserved.errorOrNull());
                }
            }
            order.markReserved();
            orders.save(order);
            OrderPlacedEvent event = new OrderPlacedEvent(order.id(), order.customerId(), order.total(), Instant.now());
            outbox.save(new OutboxMessage(order.id().value(), "OrderPlaced", event.toString()));
            return Result.ok(new PlaceOrderResult(order.id(), "order reserved and outbox staged"));
        });
    }
    public void placeOrderAndCrashBeforeCommit(PlaceOrderCommand command) {
        txManager.inTransaction("safe-place-order-crash", () -> {
            Order order = new Order(command.orderId(), command.customerId(), command.lines());
            orders.save(order);
            outbox.save(new OutboxMessage(order.id().value(), "OrderPlaced", "payload"));
            throw new RuntimeException("crash before commit");
        });
    }
}

package com.example.txdeepdive.tx;

import java.util.*;
import java.util.function.Supplier;

// Pattern: Transaction Script Boundary - der Application Service fuehrt den Use Case in einer klaren Grenze aus.
public final class TransactionManager {
    private final ThreadLocal<Transaction> current = new ThreadLocal<>();
    private final List<String> log = new ArrayList<>();

    public <T> T inTransaction(String name, Supplier<T> work) {
        if (current.get() != null) throw new IllegalStateException("nested transactions are not supported in this lab");
        Transaction tx = new Transaction(name);
        current.set(tx);
        log.add("BEGIN " + name);
        try {
            T result = work.get();
            if (tx.rollbackOnly()) {
                log.add("ROLLBACK_ONLY " + name + " staged=" + tx.size());
                return result;
            }
            tx.commit();
            log.add("COMMIT " + name + " staged=" + tx.size());
            return result;
        } catch (RuntimeException ex) {
            log.add("ROLLBACK " + name + " reason=" + ex.getClass().getSimpleName() + " staged=" + tx.size());
            throw ex;
        } finally {
            current.remove();
        }
    }

    public Transaction required() {
        Transaction tx = current.get();
        if (tx == null) throw new IllegalStateException("no active transaction");
        return tx;
    }
    public boolean active() { return current.get() != null; }
    public List<String> log() { return List.copyOf(log); }
}

package com.example.txdeepdive.adapters;

import com.example.txdeepdive.domain.Order;
import com.example.txdeepdive.outbox.OutboxMessage;
import com.example.txdeepdive.shared.*;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

public final class InMemoryDatabase {
    final Map<OrderId, Order> orders = new ConcurrentHashMap<>();
    final List<OutboxMessage> outbox = Collections.synchronizedList(new ArrayList<>());
    final Map<ProductId, Integer> stock = new ConcurrentHashMap<>();
    public void putStock(ProductId id, int quantity) { stock.put(id, quantity); }
    public int orderCount() { return orders.size(); }
    public int outboxCount() { return outbox.size(); }
}

package com.example.txdeepdive.application;

import com.example.txdeepdive.outbox.OutboxMessage;
import com.example.txdeepdive.ports.*;

// Pattern: Outbox Publisher - Broker-Versand ist nachgelagert, wiederholbar und vom DB-Commit getrennt.
public final class OutboxPublisher {
    private final OutboxRepository outbox;
    private final EventBroker broker;
    public OutboxPublisher(OutboxRepository outbox, EventBroker broker) { this.outbox = outbox; this.broker = broker; }
    public int publishPending() {
        int sent = 0;
        for (OutboxMessage message : outbox.pending()) {
            broker.publish(message.eventType(), message.payload());
            message.markSent();
            sent++;
        }
        return sent;
    }
}

package com.example.txdeepdive;

import com.example.txdeepdive.adapters.*;
import com.example.txdeepdive.application.*;
import com.example.txdeepdive.domain.*;
import com.example.txdeepdive.ports.*;
import com.example.txdeepdive.shared.*;
import com.example.txdeepdive.tx.*;
import java.util.List;

public final class Run4C1TestRunner {
    public static void main(String[] args) {
        badDualWritePublishesEventAlthoughOrderRolledBack();
        safeOutboxRollsBackOrderAndOutboxTogether();
        successfulSafeUseCaseCommitsOrderInventoryAndOutbox();
        System.out.println("RUN4C1_TESTS_OK");
    }

    static Fixtures fixtures() {
        TransactionManager tx = new TransactionManager();
        InMemoryDatabase db = new InMemoryDatabase();
        ProductId product = new ProductId("P-42");
        db.putStock(product, 5);
        OrderRepository orders = new InMemoryOrderRepository(db, tx);
        OutboxRepository outbox = new InMemoryOutboxRepository(db, tx);
        InventoryPort inventory = new TransactionalInventoryAdapter(db, tx);
        InMemoryEventBroker broker = new InMemoryEventBroker();
        return new Fixtures(tx, db, product, orders, outbox, inventory, broker);
    }

    static PlaceOrderCommand command(ProductId product, String orderId) {
        return new PlaceOrderCommand("idem-" + orderId, OrderId.of(orderId), new CustomerId("C-42"), List.of(new OrderLine(product, 2, Money.eur("12.50"))));
    }

    static void badDualWritePublishesEventAlthoughOrderRolledBack() {
        Fixtures f = fixtures();
        BadDualWriteOrderService bad = new BadDualWriteOrderService(f.tx, f.orders, f.broker);
        try { bad.placeOrderAndCrashBeforeCommit(command(f.product, "O-BAD")); throw new AssertionError("expected failure"); }
        catch (DualWriteFailure expected) { }
        assertEquals(0, f.orders.count(), "order must be rolled back");
        assertEquals(1, f.broker.publishedCount(), "event was already published and cannot be rolled back");
    }

    static void safeOutboxRollsBackOrderAndOutboxTogether() {
        Fixtures f = fixtures();
        SafeOrderApplicationService safe = new SafeOrderApplicationService(f.tx, f.orders, f.inventory, f.outbox);
        try { safe.placeOrderAndCrashBeforeCommit(command(f.product, "O-SAFE-FAIL")); throw new AssertionError("expected failure"); }
        catch (RuntimeException expected) { }
        assertEquals(0, f.orders.count(), "order rolled back");
        assertEquals(0, f.outbox.count(), "outbox rolled back with order");
        assertEquals(0, f.broker.publishedCount(), "broker did not see uncommitted data");
    }

    static void successfulSafeUseCaseCommitsOrderInventoryAndOutbox() {
        Fixtures f = fixtures();
        SafeOrderApplicationService safe = new SafeOrderApplicationService(f.tx, f.orders, f.inventory, f.outbox);
        Result<PlaceOrderResult, DomainError> result = safe.placeOrder(command(f.product, "O-OK"));
        assertTrue(result.isOk(), "result should be ok");
        assertEquals(1, f.orders.count(), "order committed");
        assertEquals(1, f.outbox.count(), "outbox committed");
        assertEquals(3, f.inventory.available(f.product), "inventory decremented after commit");
        new OutboxPublisher(f.outbox, f.broker).publishPending();
        assertEquals(1, f.broker.publishedCount(), "publisher sends after commit");
    }

    static void assertTrue(boolean value, String msg) { if (!value) throw new AssertionError(msg); }
    static void assertEquals(int expected, int actual, String msg) { if (expected != actual) throw new AssertionError(msg + " expected=" + expected + " actual=" + actual); }
    record Fixtures(TransactionManager tx, InMemoryDatabase db, ProductId product, OrderRepository orders, OutboxRepository outbox, InventoryPort inventory, InMemoryEventBroker broker) { }
}