W12 코드 뒤풀이 · 같은 요청을 한 번만 바꾸는 법

package com.example.financialcore.idempotency;

import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.HexFormat;

@Component
public final class RequestHasher {
    public static final String VERSION = "v1";

    public byte[] canonicalBytes(long fromAccountId, long toAccountId, long amount) {
        if (fromAccountId <= 0 || toAccountId <= 0 || amount <= 0) {
            throw new IllegalArgumentException("hash fields must be positive");
        }
        String canonical = VERSION
            + "\nfrom=" + Long.toString(fromAccountId)
            + "\nto=" + Long.toString(toAccountId)
            + "\namount=" + Long.toString(amount)
            + "\n";
        return canonical.getBytes(StandardCharsets.UTF_8);
    }

    public String hash(long fromAccountId, long toAccountId, long amount) {
        try {
            return HexFormat.of().formatHex(
                MessageDigest.getInstance("SHA-256")
                    .digest(canonicalBytes(fromAccountId, toAccountId, amount))
            );
        } catch (NoSuchAlgorithmException impossible) {
            throw new IllegalStateException("SHA-256 is required by the JDK", impossible);
        }
    }
}

package com.example.financialcore.idempotency;

import org.junit.jupiter.api.Test;

import java.nio.charset.StandardCharsets;

import static org.assertj.core.api.Assertions.assertThat;

class RequestHasherTest {
    private final RequestHasher hasher = new RequestHasher();

    @Test
    void canonicalContractIsVersionedFixedOrderUtf8Decimal() {
        assertThat(new String(hasher.canonicalBytes(10, 20, 3000), StandardCharsets.UTF_8))
            .as("W12D1_RED_EXPECTED_VERSIONED_CANONICAL_HASH")
            .isEqualTo("v1\nfrom=10\nto=20\namount=3000\n");
        assertThat(hasher.hash(10, 20, 3000)).hasSize(64).matches("[0-9a-f]{64}");
    }

    @Test
    void eachSemanticFieldChangesTheHash() {
        String canonical = hasher.hash(10, 20, 3000);
        assertThat(hasher.hash(11, 20, 3000)).isNotEqualTo(canonical);
        assertThat(hasher.hash(10, 21, 3000)).isNotEqualTo(canonical);
        assertThat(hasher.hash(10, 20, 3001)).isNotEqualTo(canonical);
    }
}

CREATE TABLE account (
    id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    owner_id VARCHAR(64) NOT NULL,
    account_no VARCHAR(32) NOT NULL UNIQUE,
    status VARCHAR(16) NOT NULL CHECK (status IN ('ACTIVE', 'CLOSED')),
    currency VARCHAR(3) NOT NULL CHECK (currency = 'KRW'),
    balance BIGINT NOT NULL CHECK (balance >= 0),
    version BIGINT NOT NULL DEFAULT 0
);

CREATE INDEX idx_account_owner_id ON account(owner_id, id);

CREATE TABLE business_tx (
    id UUID PRIMARY KEY,
    tx_type VARCHAR(32) NOT NULL,
    status VARCHAR(16) NOT NULL,
    correlation_id VARCHAR(64) NOT NULL UNIQUE,
    requested_at TIMESTAMPTZ NOT NULL,
    completed_at TIMESTAMPTZ
);

CREATE TABLE ledger_entry (
    id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    business_tx_id UUID NOT NULL REFERENCES business_tx(id),
    account_id BIGINT NOT NULL REFERENCES account(id),
    entry_type VARCHAR(32) NOT NULL,
    amount BIGINT NOT NULL CHECK (amount > 0),
    signed_amount BIGINT NOT NULL,
    balance_after BIGINT NOT NULL CHECK (balance_after >= 0),
    reversal_of BIGINT REFERENCES ledger_entry(id),
    created_at TIMESTAMPTZ NOT NULL,
    UNIQUE (business_tx_id, account_id, entry_type)
);

CREATE INDEX idx_ledger_account_created_id
    ON ledger_entry(account_id, created_at DESC, id DESC);

CREATE TABLE idempotency_request (
    id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
    scope VARCHAR(64) NOT NULL,
    actor_id VARCHAR(64) NOT NULL,
    idempotency_key VARCHAR(128) NOT NULL,
    request_hash CHAR(64) NOT NULL,
    status VARCHAR(16) NOT NULL,
    response_status INTEGER,
    response_body TEXT,
    created_at TIMESTAMPTZ NOT NULL,
    completed_at TIMESTAMPTZ,
    UNIQUE (scope, actor_id, idempotency_key)
);

package com.example.financialcore.idempotency;

import com.example.financialcore.PostgresIntegrationTestSupport;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.dao.DataIntegrityViolationException;
import org.springframework.jdbc.core.simple.JdbcClient;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;

@SpringBootTest
class IdempotencySchemaIT extends PostgresIntegrationTestSupport {
    @Autowired JdbcClient jdbc;

    @BeforeEach void clean() { jdbc.sql("TRUNCATE idempotency_request RESTART IDENTITY").update(); }

    @Test
    void uniqueScopeActorAndKeyAreEnforcedByV001() {
        insert("a".repeat(64));
        assertThatThrownBy(() -> insert("b".repeat(64)))
            .isInstanceOf(DataIntegrityViolationException.class);
        assertThat(jdbc.sql("SELECT COUNT(*) FROM idempotency_request")
            .query(Long.class).single()).isEqualTo(1);
    }

    private void insert(String hash) {
        jdbc.sql("""
                INSERT INTO idempotency_request(
                  scope,actor_id,idempotency_key,request_hash,status,created_at
                ) VALUES ('TRANSFER','customer-1','same-key',:hash,'PROCESSING',now())
                """).param("hash", hash).update();
    }
}

package com.example.financialcore.idempotency;

import jakarta.persistence.EntityManager;
import jakarta.persistence.PersistenceContext;
import org.springframework.stereotype.Repository;
import org.springframework.transaction.annotation.Transactional;

import java.time.Instant;
import java.util.List;
import java.util.Optional;

@Repository
public class IdempotencyStore {
    public record Existing(String requestHash, String status, Integer responseStatus, String responseBody) {}

    @PersistenceContext
    private EntityManager entityManager;

    @Transactional
    public Optional<Long> claim(String scope, String actorId, String key, String requestHash) {
        List<?> rows = entityManager.createNativeQuery("""
                INSERT INTO idempotency_request
                    (scope, actor_id, idempotency_key, request_hash, status, created_at)
                VALUES (:scope, :actorId, :key, :requestHash, 'PROCESSING', :now)
                ON CONFLICT (scope, actor_id, idempotency_key) DO NOTHING
                RETURNING id
                """)
            .setParameter("scope", scope)
            .setParameter("actorId", actorId)
            .setParameter("key", key)
            .setParameter("requestHash", requestHash)
            .setParameter("now", Instant.now())
            .getResultList();
        if (rows.isEmpty()) return Optional.empty();
        return Optional.of(((Number) rows.getFirst()).longValue());
    }

    @Transactional(readOnly = true)
    public Existing find(String scope, String actorId, String key) {
        Object[] row = (Object[]) entityManager.createNativeQuery("""
                SELECT request_hash, status, response_status, response_body
                FROM idempotency_request
                WHERE scope=:scope AND actor_id=:actorId AND idempotency_key=:key
                """)
            .setParameter("scope", scope)
            .setParameter("actorId", actorId)
            .setParameter("key", key)
            .getSingleResult();
        return new Existing(
            (String) row[0],
            (String) row[1],
            row[2] == null ? null : ((Number) row[2]).intValue(),
            (String) row[3]
        );
    }

    @Transactional
    public void complete(long id, int responseStatus, String responseBody) {
        int updated = entityManager.createNativeQuery("""
                UPDATE idempotency_request
                SET status='COMPLETED', response_status=:responseStatus,
                    response_body=:responseBody, completed_at=:now
                WHERE id=:id AND status='PROCESSING'
                """)
            .setParameter("responseStatus", responseStatus)
            .setParameter("responseBody", responseBody)
            .setParameter("now", Instant.now())
            .setParameter("id", id)
            .executeUpdate();
        if (updated != 1) throw new IllegalStateException("idempotency completion failed");
    }
}

package com.example.financialcore.idempotency;

import com.example.financialcore.PostgresIntegrationTestSupport;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.jdbc.core.simple.JdbcClient;

import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

import static org.assertj.core.api.Assertions.assertThat;

@SpringBootTest
class AtomicClaim50IT extends PostgresIntegrationTestSupport {
    @Autowired IdempotencyStore store;
    @Autowired JdbcClient jdbc;

    @BeforeEach
    void clean() {
        jdbc.sql("TRUNCATE idempotency_request RESTART IDENTITY").update();
    }

    @Test
    void fifty_concurrent_claims_have_one_owner_and_forty_nine_existing_results() throws Exception {
        int tasks = 50;
        var ready = new CountDownLatch(tasks);
        var start = new CountDownLatch(1);
        var pool = Executors.newFixedThreadPool(tasks);
        try {
            List<Future<Optional<Long>>> futures = new ArrayList<>();
            for (int i = 0; i < tasks; i++) {
                futures.add(pool.submit(() -> {
                    ready.countDown();
                    start.await();
                    return store.claim("TRANSFER", "customer-50", "same-key", "a".repeat(64));
                }));
            }
            assertThat(ready.await(10, TimeUnit.SECONDS)).isTrue();
            start.countDown();
            int owners = 0;
            int existing = 0;
            for (Future<Optional<Long>> future : futures) {
                if (future.get(30, TimeUnit.SECONDS).isPresent()) owners++; else existing++;
            }
            assertThat(owners).as("W12D3_RED_EXPECTED_ATOMIC_CLAIM")
                .isEqualTo(1);
            assertThat(existing).isEqualTo(49);
            assertThat(jdbc.sql("SELECT COUNT(*) FROM idempotency_request")
                .query(Long.class).single()).isEqualTo(1);
        } finally {
            start.countDown();
            pool.shutdownNow();
        }
    }

    @Test
    void named_parameters_do_not_turn_actor_text_into_sql() {
        String probe = "owner' OR '1'='1";
        assertThat(store.claim("TRANSFER", probe, "probe-key", "b".repeat(64))).isPresent();
        assertThat(jdbc.sql("SELECT COUNT(*) FROM idempotency_request WHERE actor_id <> :probe")
            .param("probe", probe).query(Long.class).single()).isZero();
    }
}

package com.example.financialcore.api;

public enum ErrorCode {
    INVALID_REQUEST,
    ACCOUNT_NOT_FOUND,
    DUPLICATE_ACCOUNT,
    INSUFFICIENT_BALANCE,
    IDEMPOTENCY_CONFLICT,
    IDEMPOTENCY_IN_PROGRESS,
    ACCESS_DENIED,
    INTERNAL_ERROR
}

package com.example.financialcore.api;

import jakarta.servlet.http.HttpServletRequest;
import org.springframework.dao.DataIntegrityViolationException;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.security.access.AccessDeniedException;
import org.springframework.web.bind.MethodArgumentNotValidException;
import org.springframework.web.bind.annotation.ExceptionHandler;
import org.springframework.web.bind.annotation.RestControllerAdvice;

@RestControllerAdvice
public final class ApiExceptionHandler {
    @ExceptionHandler(BusinessException.class)
    ResponseEntity<ApiError> business(BusinessException failure, HttpServletRequest request) {
        HttpStatus status = switch (failure.code()) {
            case INVALID_REQUEST -> HttpStatus.BAD_REQUEST;
            case ACCOUNT_NOT_FOUND -> HttpStatus.NOT_FOUND;
            case ACCESS_DENIED -> HttpStatus.FORBIDDEN;
            case DUPLICATE_ACCOUNT, INSUFFICIENT_BALANCE,
                 IDEMPOTENCY_CONFLICT, IDEMPOTENCY_IN_PROGRESS -> HttpStatus.CONFLICT;
            case INTERNAL_ERROR -> HttpStatus.INTERNAL_SERVER_ERROR;
        };
        return response(status, failure.code(), failure.getMessage(), request);
    }

    @ExceptionHandler(MethodArgumentNotValidException.class)
    ResponseEntity<ApiError> validation(MethodArgumentNotValidException ignored, HttpServletRequest request) {
        return response(HttpStatus.BAD_REQUEST, ErrorCode.INVALID_REQUEST, "request validation failed", request);
    }

    @ExceptionHandler(IllegalArgumentException.class)
    ResponseEntity<ApiError> illegalArgument(IllegalArgumentException ignored, HttpServletRequest request) {
        return response(HttpStatus.BAD_REQUEST, ErrorCode.INVALID_REQUEST, "request is invalid", request);
    }

    @ExceptionHandler(DataIntegrityViolationException.class)
    ResponseEntity<ApiError> constraint(DataIntegrityViolationException ignored, HttpServletRequest request) {
        return response(HttpStatus.CONFLICT, ErrorCode.DUPLICATE_ACCOUNT, "resource already exists", request);
    }

    @ExceptionHandler(AccessDeniedException.class)
    ResponseEntity<ApiError> denied(AccessDeniedException ignored, HttpServletRequest request) {
        return response(HttpStatus.FORBIDDEN, ErrorCode.ACCESS_DENIED, "access denied", request);
    }

    @ExceptionHandler(RuntimeException.class)
    ResponseEntity<ApiError> unexpected(RuntimeException ignored, HttpServletRequest request) {
        return response(HttpStatus.INTERNAL_SERVER_ERROR, ErrorCode.INTERNAL_ERROR,
            "unexpected server error", request);
    }

    private static ResponseEntity<ApiError> response(
        HttpStatus status, ErrorCode code, String message, HttpServletRequest request
    ) {
        return ResponseEntity.status(status)
            .body(new ApiError(code.name(), message, RequestIdFilter.current(request)));
    }
}

package com.example.financialcore.transfer;

/** Optional test hook. Production has no bean and therefore uses the no-op instance. */
public interface TransferFailureHook {
    TransferFailureHook NONE = new TransferFailureHook() {};

    default void afterClaim() {}

    default void afterBusinessMutation() {}
}

package com.example.financialcore.transfer;

import com.example.financialcore.account.Account;
import com.example.financialcore.account.AccountRepository;
import com.example.financialcore.api.BusinessException;
import com.example.financialcore.api.ErrorCode;
import com.example.financialcore.idempotency.IdempotencyStore;
import com.example.financialcore.idempotency.RequestHasher;
import com.example.financialcore.ledger.BusinessTransaction;
import com.example.financialcore.ledger.BusinessTransactionRepository;
import com.example.financialcore.ledger.LedgerEntry;
import com.example.financialcore.ledger.LedgerEntryRepository;
import org.springframework.stereotype.Service;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.transaction.annotation.Transactional;

import java.time.Instant;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Service
public class TransferService {
    public record Command(
        String actorId, String idempotencyKey,
        long fromAccountId, long toAccountId, long amount,
        String requestId
    ) {
        public Command(
            String actorId, String idempotencyKey,
            long fromAccountId, long toAccountId, long amount
        ) {
            this(actorId, idempotencyKey, fromAccountId, toAccountId, amount,
                "internal:" + idempotencyKey);
        }
    }

    public record Result(String businessTransactionId, long fromBalance, long toBalance, boolean replayed) {}

    private final AccountRepository accounts;
    private final BusinessTransactionRepository transactions;
    private final LedgerEntryRepository ledger;
    private final IdempotencyStore idempotency;
    private final RequestHasher requestHasher;
    private final TransferFailureHook failureHook;

    public TransferService(
        AccountRepository accounts,
        BusinessTransactionRepository transactions,
        LedgerEntryRepository ledger,
        IdempotencyStore idempotency,
        RequestHasher requestHasher,
        ObjectProvider<TransferFailureHook> failureHooks
    ) {
        this.accounts = accounts;
        this.transactions = transactions;
        this.ledger = ledger;
        this.idempotency = idempotency;
        this.requestHasher = requestHasher;
        this.failureHook = failureHooks.getIfAvailable(() -> TransferFailureHook.NONE);
    }

    @Transactional
    public Result transfer(Command command) {
        validate(command);
        String requestHash = requestHasher.hash(
            command.fromAccountId(), command.toAccountId(), command.amount()
        );
        String ownerId = accounts.findOwnerId(command.fromAccountId())
            .orElseThrow(() -> new BusinessException(ErrorCode.ACCOUNT_NOT_FOUND, "account not found"));
        if (!ownerId.equals(command.actorId())) {
            throw new BusinessException(ErrorCode.ACCESS_DENIED, "access denied");
        }
        var claim = idempotency.claim("TRANSFER", command.actorId(), command.idempotencyKey(), requestHash);
        if (claim.isEmpty()) return replay(command, requestHash);
        failureHook.afterClaim();

        List<Long> ids = List.of(command.fromAccountId(), command.toAccountId()).stream().sorted().toList();
        List<Account> locked = accounts.findAllForUpdateOrderById(ids);
        if (locked.size() != 2) throw new BusinessException(ErrorCode.ACCOUNT_NOT_FOUND, "account not found");
        Map<Long, Account> byId = new HashMap<>();
        locked.forEach(a -> byId.put(a.getId(), a));
        Account from = byId.get(command.fromAccountId());
        Account to = byId.get(command.toAccountId());
        if (from == null || to == null) throw new BusinessException(ErrorCode.ACCOUNT_NOT_FOUND, "account not found");

        from.withdraw(command.amount());
        to.deposit(command.amount());
        Instant now = Instant.now();
        String correlationId = command.actorId() + ":" + command.idempotencyKey();
        BusinessTransaction tx = transactions.save(BusinessTransaction.completedTransfer(correlationId, now));
        ledger.save(LedgerEntry.transferOut(tx, from, command.amount(), now));
        ledger.save(LedgerEntry.transferIn(tx, to, command.amount(), now));
        failureHook.afterBusinessMutation();

        String body = tx.getId() + "," + from.getBalance() + "," + to.getBalance();
        idempotency.complete(claim.orElseThrow(), 201, body);
        return new Result(tx.getId().toString(), from.getBalance(), to.getBalance(), false);
    }

    private Result replay(Command command, String requestHash) {
        IdempotencyStore.Existing existing = idempotency.find("TRANSFER", command.actorId(), command.idempotencyKey());
        if (!existing.requestHash().equals(requestHash)) {
            throw new BusinessException(ErrorCode.IDEMPOTENCY_CONFLICT, "idempotency key conflicts with another request");
        }
        if (!"COMPLETED".equals(existing.status()) || existing.responseBody() == null) {
            throw new BusinessException(ErrorCode.IDEMPOTENCY_IN_PROGRESS, "idempotent request is still processing");
        }
        String[] values = existing.responseBody().split(",");
        return new Result(values[0], Long.parseLong(values[1]), Long.parseLong(values[2]), true);
    }

    private static void validate(Command command) {
        if (command.actorId() == null || command.actorId().isBlank()) throw new IllegalArgumentException("actorId");
        if (command.idempotencyKey() == null || command.idempotencyKey().isBlank()) throw new IllegalArgumentException("key");
        if (command.fromAccountId() <= 0 || command.toAccountId() <= 0) throw new IllegalArgumentException("account id");
        if (command.fromAccountId() == command.toAccountId()) throw new IllegalArgumentException("same account");
        if (command.amount() <= 0) throw new IllegalArgumentException("amount");
    }
}

package com.example.financialcore.transfer;

import com.example.financialcore.PostgresIntegrationTestSupport;
import com.example.financialcore.account.Account;
import com.example.financialcore.account.AccountOpeningService;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.context.TestConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Import;
import org.springframework.jdbc.core.simple.JdbcClient;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;

@SpringBootTest
@Import(TransferFailurePointIT.FailureConfiguration.class)
class TransferFailurePointIT extends PostgresIntegrationTestSupport {
    @Autowired JdbcClient jdbc;
    @Autowired AccountOpeningService openings;
    @Autowired TransferService transfers;
    @Autowired ControlledFailureHook hook;
    Account from;
    Account to;

    @BeforeEach void clean() {
        hook.point = ControlledFailureHook.Point.NONE;
        jdbc.sql("TRUNCATE idempotency_request,ledger_entry,business_tx,account RESTART IDENTITY CASCADE").update();
        from = openings.open("customer-1", "ROLL-FROM", 10_000);
        to = openings.open("customer-2", "ROLL-TO", 5_000);
    }

    @Test void afterClaimRuntimeExceptionRollsBackClaim() {
        hook.point = ControlledFailureHook.Point.AFTER_CLAIM;
        assertFailureAndNoEffect("after-claim");
    }

    @Test void afterBusinessRuntimeExceptionRollsBackEveryEffect() {
        hook.point = ControlledFailureHook.Point.AFTER_BUSINESS;
        assertFailureAndNoEffect("after-business");
    }

    private void assertFailureAndNoEffect(String key) {
        assertThatThrownBy(() -> transfers.transfer(new TransferService.Command(
            "customer-1", key, from.getId(), to.getId(), 1_000)))
            .isInstanceOf(RuntimeException.class).hasMessageContaining("injected");
        assertThat(jdbc.sql("SELECT COUNT(*) FROM idempotency_request").query(Long.class).single()).isZero();
        assertThat(jdbc.sql("SELECT COUNT(*) FROM business_tx WHERE tx_type='TRANSFER'")
            .query(Long.class).single()).isZero();
        assertThat(jdbc.sql("SELECT COUNT(*) FROM ledger_entry WHERE entry_type LIKE 'TRANSFER_%'")
            .query(Long.class).single()).isZero();
    }

    @TestConfiguration(proxyBeanMethods = false)
    static class FailureConfiguration {
        @Bean ControlledFailureHook controlledFailureHook() { return new ControlledFailureHook(); }
    }

    static final class ControlledFailureHook implements TransferFailureHook {
        enum Point { NONE, AFTER_CLAIM, AFTER_BUSINESS }
        volatile Point point = Point.NONE;
        @Override public void afterClaim() {
            if (point == Point.AFTER_CLAIM) throw new RuntimeException("injected after claim");
        }
        @Override public void afterBusinessMutation() {
            if (point == Point.AFTER_BUSINESS) throw new RuntimeException("injected after business");
        }
    }
}

package com.example.financialcore.transfer;

import com.example.financialcore.PostgresIntegrationTestSupport;
import com.example.financialcore.account.Account;
import com.example.financialcore.account.AccountOpeningService;
import com.example.financialcore.api.BusinessException;
import com.example.financialcore.api.ErrorCode;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.jdbc.core.simple.JdbcClient;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.catchThrowable;

@SpringBootTest
class TransferIdempotencyIT extends PostgresIntegrationTestSupport {
    @Autowired JdbcClient jdbc;
    @Autowired AccountOpeningService openings;
    @Autowired TransferService transfers;
    Account from;
    Account to;

    @BeforeEach void clean() {
        jdbc.sql("TRUNCATE idempotency_request,ledger_entry,business_tx,account RESTART IDENTITY CASCADE").update();
        from = openings.open("customer-1", "IDEM-FROM", 10_000);
        to = openings.open("customer-2", "IDEM-TO", 5_000);
    }

    @Test
    void samePayloadReplaysAndChangedPayloadConflictsWithoutExtraEffect() {
        var first = new TransferService.Command("customer-1", "same-key", from.getId(), to.getId(), 1_000);
        var changed = new TransferService.Command("customer-1", "same-key", from.getId(), to.getId(), 2_000);
        assertThat(transfers.transfer(first).replayed()).isFalse();
        assertThat(transfers.transfer(first).replayed()).isTrue();

        Throwable failure = catchThrowable(() -> transfers.transfer(changed));
        assertThat(failure)
            .as("W12D4_RED_EXPECTED_SEMANTIC_CONFLICT")
            .isInstanceOf(BusinessException.class);
        assertThat(((BusinessException) failure).code()).isEqualTo(ErrorCode.IDEMPOTENCY_CONFLICT);
        assertThat(jdbc.sql("SELECT COUNT(*) FROM business_tx WHERE tx_type='TRANSFER'")
            .query(Long.class).single()).isEqualTo(1);
        assertThat(jdbc.sql("SELECT COUNT(*) FROM ledger_entry WHERE entry_type LIKE 'TRANSFER_%'")
            .query(Long.class).single()).isEqualTo(2);
    }
}

package com.example.financialcore.transfer;

import com.example.financialcore.PostgresIntegrationTestSupport;
import com.example.financialcore.account.Account;
import com.example.financialcore.account.AccountOpeningService;
import com.example.financialcore.account.AccountRepository;
import com.example.financialcore.api.BusinessException;
import com.example.financialcore.api.ErrorCode;
import com.example.financialcore.ledger.LedgerEntryRepository;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.jdbc.core.simple.JdbcClient;
import org.springframework.boot.test.context.TestConfiguration;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Import;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;

@SpringBootTest
@Import(TransferIntegrationTest.FailureHookConfiguration.class)
class TransferIntegrationTest extends PostgresIntegrationTestSupport {

    @TestConfiguration
    static class FailureHookConfiguration {
        @Bean
        ControlledFailureHook controlledFailureHook() {
            return new ControlledFailureHook();
        }
    }

    static final class ControlledFailureHook implements TransferFailureHook {
        enum Point { NONE, AFTER_CLAIM, AFTER_BUSINESS_MUTATION }

        private volatile Point point = Point.NONE;

        void failAt(Point point) { this.point = point; }
        void reset() { this.point = Point.NONE; }

        @Override
        public void afterClaim() {
            if (point == Point.AFTER_CLAIM) throw new RuntimeException("injected after claim");
        }

        @Override
        public void afterBusinessMutation() {
            if (point == Point.AFTER_BUSINESS_MUTATION) {
                throw new RuntimeException("injected after business mutation");
            }
        }
    }

    @Autowired AccountRepository accounts;
    @Autowired AccountOpeningService openings;
    @Autowired LedgerEntryRepository ledger;
    @Autowired TransferService transfers;
    @Autowired JdbcClient jdbc;
    @Autowired ControlledFailureHook failureHook;

    private Account from;
    private Account to;

    @BeforeEach
    void setUp() {
        jdbc.sql("TRUNCATE idempotency_request, ledger_entry, business_tx, account RESTART IDENTITY CASCADE").update();
        failureHook.reset();
        from = openings.open("customer-1", "A", 10_000);
        to = openings.open("customer-1", "B", 10_000);
    }

    @Test
    void transfer_and_same_key_replay_once() {
        var command = new TransferService.Command("customer-1", "key-1", from.getId(), to.getId(), 3_000);
        var first = transfers.transfer(command);
        var replay = transfers.transfer(command);

        assertThat(first.replayed()).isFalse();
        assertThat(replay.replayed()).isTrue();
        assertThat(accounts.findById(from.getId()).orElseThrow().getBalance()).isEqualTo(7_000);
        assertThat(accounts.findById(to.getId()).orElseThrow().getBalance()).isEqualTo(13_000);
        assertThat(ledger.count()).isEqualTo(4);
        assertThat(jdbc.sql("SELECT COALESCE(SUM(signed_amount),0) FROM ledger_entry WHERE entry_type LIKE 'TRANSFER_%'")
            .query(Long.class).single()).isZero();
    }

    @Test
    void json_field_order_and_whitespace_variant_replays_without_extra_effect() {
        String firstJson = "{\"from\":" + from.getId() + ",\"to\":" + to.getId() + ",\"amount\":1000}";
        String variantJson = "{  \"amount\" : 1000, \n \"to\" : " + to.getId() + ", \"from\" : " + from.getId() + " }";

        var first = transfers.transfer(commandFromJson("representation-key", firstJson));
        var replay = transfers.transfer(commandFromJson("representation-key", variantJson));

        assertThat(first.replayed()).isFalse();
        assertThat(replay.replayed()).isTrue();
        assertThat(accounts.findById(from.getId()).orElseThrow().getBalance()).isEqualTo(9_000);
        assertThat(accounts.findById(to.getId()).orElseThrow().getBalance()).isEqualTo(11_000);
        assertOneTransferEffect();
    }

    @Test
    void same_key_concurrent_requests_change_business_once() throws Exception {
        int n = 20;
        var ready = new CountDownLatch(n);
        var start = new CountDownLatch(1);
        var pool = Executors.newFixedThreadPool(n);
        try {
            var command = new TransferService.Command(
                "customer-1", "burst-key", from.getId(), to.getId(), 1_000);
            List<Future<TransferService.Result>> futures = new ArrayList<>();
            for (int i = 0; i < n; i++) {
                futures.add(pool.submit(() -> {
                    ready.countDown();
                    start.await();
                    return transfers.transfer(command);
                }));
            }
            assertThat(ready.await(5, TimeUnit.SECONDS)).isTrue();
            start.countDown();
            List<TransferService.Result> results = new ArrayList<>();
            for (var future : futures) results.add(future.get(15, TimeUnit.SECONDS));

            assertThat(results.stream().filter(r -> !r.replayed()).count()).isEqualTo(1);
            assertThat(accounts.findById(from.getId()).orElseThrow().getBalance()).isEqualTo(9_000);
            assertThat(accounts.findById(to.getId()).orElseThrow().getBalance()).isEqualTo(11_000);
            assertThat(ledger.count()).isEqualTo(4);
            assertOneTransferEffect();
        } finally {
            pool.shutdownNow();
        }
    }

    @Test
    void opposite_direction_transfers_preserve_total_balance() throws Exception {
        int perDirection = 10;
        int n = perDirection * 2;
        var ready = new CountDownLatch(n);
        var start = new CountDownLatch(1);
        var pool = Executors.newFixedThreadPool(n);
        try {
            List<Future<TransferService.Result>> futures = new ArrayList<>();
            for (int i = 0; i < perDirection; i++) {
                int seq = i;
                futures.add(pool.submit(() -> invokeAfterBarrier(
                    ready, start, new TransferService.Command("customer-1", "ab-" + seq, from.getId(), to.getId(), 100))));
                futures.add(pool.submit(() -> invokeAfterBarrier(
                    ready, start, new TransferService.Command("customer-1", "ba-" + seq, to.getId(), from.getId(), 100))));
            }
            assertThat(ready.await(5, TimeUnit.SECONDS)).isTrue();
            start.countDown();
            for (var future : futures) future.get(20, TimeUnit.SECONDS);

            long fromBalance = accounts.findById(from.getId()).orElseThrow().getBalance();
            long toBalance = accounts.findById(to.getId()).orElseThrow().getBalance();
            assertThat(fromBalance).isEqualTo(10_000);
            assertThat(toBalance).isEqualTo(10_000);
            assertThat(fromBalance + toBalance).isEqualTo(20_000);
            assertThat(ledger.count()).isEqualTo(42);
        } finally {
            pool.shutdownNow();
        }
    }

    @Test
    void same_key_with_different_semantic_request_conflicts_without_extra_effect() {
        var first = new TransferService.Command("customer-1", "conflict-key", from.getId(), to.getId(), 1_000);
        var changed = new TransferService.Command("customer-1", "conflict-key", from.getId(), to.getId(), 2_000);

        transfers.transfer(first);
        assertThatThrownBy(() -> transfers.transfer(changed))
            .isInstanceOfSatisfying(BusinessException.class,
                failure -> assertThat(failure.code()).isEqualTo(ErrorCode.IDEMPOTENCY_CONFLICT));

        assertThat(accounts.findById(from.getId()).orElseThrow().getBalance()).isEqualTo(9_000);
        assertThat(accounts.findById(to.getId()).orElseThrow().getBalance()).isEqualTo(11_000);
        assertThat(jdbc.sql("SELECT COUNT(*) FROM business_tx WHERE tx_type='TRANSFER'")
            .query(Long.class).single()).isEqualTo(1);
        assertThat(jdbc.sql("SELECT COUNT(*) FROM ledger_entry WHERE entry_type LIKE 'TRANSFER_%'")
            .query(Long.class).single()).isEqualTo(2);
        assertThat(jdbc.sql("SELECT COUNT(*) FROM idempotency_request")
            .query(Long.class).single()).isEqualTo(1);
    }

    @Test
    void same_key_with_changed_from_conflicts_without_extra_effect() {
        Account other = openings.open("customer-1", "C", 5_000);
        var first = new TransferService.Command("customer-1", "from-conflict-key", from.getId(), to.getId(), 1_000);
        var changedFrom = new TransferService.Command("customer-1", "from-conflict-key", other.getId(), to.getId(), 1_000);

        transfers.transfer(first);
        assertThatThrownBy(() -> transfers.transfer(changedFrom))
            .isInstanceOfSatisfying(BusinessException.class,
                failure -> assertThat(failure.code()).isEqualTo(ErrorCode.IDEMPOTENCY_CONFLICT));

        assertThat(accounts.findById(from.getId()).orElseThrow().getBalance()).isEqualTo(9_000);
        assertThat(accounts.findById(to.getId()).orElseThrow().getBalance()).isEqualTo(11_000);
        assertThat(accounts.findById(other.getId()).orElseThrow().getBalance()).isEqualTo(5_000);
        assertOneTransferEffect();
    }

    @Test
    void same_key_with_changed_to_conflicts_without_extra_effect() {
        Account other = openings.open("customer-1", "C", 5_000);
        var first = new TransferService.Command("customer-1", "to-conflict-key", from.getId(), to.getId(), 1_000);
        var changedTo = new TransferService.Command("customer-1", "to-conflict-key", from.getId(), other.getId(), 1_000);

        transfers.transfer(first);
        assertThatThrownBy(() -> transfers.transfer(changedTo))
            .isInstanceOfSatisfying(BusinessException.class,
                failure -> assertThat(failure.code()).isEqualTo(ErrorCode.IDEMPOTENCY_CONFLICT));

        assertThat(accounts.findById(from.getId()).orElseThrow().getBalance()).isEqualTo(9_000);
        assertThat(accounts.findById(to.getId()).orElseThrow().getBalance()).isEqualTo(11_000);
        assertThat(accounts.findById(other.getId()).orElseThrow().getBalance()).isEqualTo(5_000);
        assertOneTransferEffect();
    }

    @Test
    void runtime_exception_after_claim_rolls_back_every_database_effect() {
        failureHook.failAt(ControlledFailureHook.Point.AFTER_CLAIM);
        var command = new TransferService.Command("customer-1", "fail-claim", from.getId(), to.getId(), 1_000);

        assertThatThrownBy(() -> transfers.transfer(command))
            .isInstanceOf(RuntimeException.class)
            .hasMessage("injected after claim");
        assertOnlyOpeningStateRemains();
    }

    @Test
    void runtime_exception_after_business_mutation_rolls_back_every_database_effect() {
        failureHook.failAt(ControlledFailureHook.Point.AFTER_BUSINESS_MUTATION);
        var command = new TransferService.Command("customer-1", "fail-business", from.getId(), to.getId(), 1_000);

        assertThatThrownBy(() -> transfers.transfer(command))
            .isInstanceOf(RuntimeException.class)
            .hasMessage("injected after business mutation");
        assertOnlyOpeningStateRemains();
    }

    private void assertOnlyOpeningStateRemains() {
        assertThat(accounts.findById(from.getId()).orElseThrow().getBalance()).isEqualTo(10_000);
        assertThat(accounts.findById(to.getId()).orElseThrow().getBalance()).isEqualTo(10_000);
        assertThat(jdbc.sql("SELECT COUNT(*) FROM idempotency_request").query(Long.class).single()).isZero();
        assertThat(jdbc.sql("SELECT COUNT(*) FROM business_tx WHERE tx_type='TRANSFER'")
            .query(Long.class).single()).isZero();
        assertThat(jdbc.sql("SELECT COUNT(*) FROM ledger_entry WHERE entry_type LIKE 'TRANSFER_%'")
            .query(Long.class).single()).isZero();
    }

    private TransferService.Command commandFromJson(String key, String json) {
        return new TransferService.Command(
            "customer-1",
            key,
            semanticLong(json, "from"),
            semanticLong(json, "to"),
            semanticLong(json, "amount")
        );
    }

    private long semanticLong(String json, String field) {
        var matcher = Pattern.compile("\\\"" + Pattern.quote(field) + "\\\"\\s*:\\s*(\\d+)").matcher(json);
        if (!matcher.find()) throw new IllegalArgumentException("missing semantic field: " + field);
        return Long.parseLong(matcher.group(1));
    }

    private void assertOneTransferEffect() {
        assertThat(jdbc.sql("SELECT COUNT(*) FROM business_tx WHERE tx_type='TRANSFER'")
            .query(Long.class).single()).isEqualTo(1);
        assertThat(jdbc.sql("SELECT COUNT(*) FROM ledger_entry WHERE entry_type LIKE 'TRANSFER_%'")
            .query(Long.class).single()).isEqualTo(2);
        assertThat(jdbc.sql("SELECT COUNT(*) FROM idempotency_request")
            .query(Long.class).single()).isEqualTo(1);
    }

    private TransferService.Result invokeAfterBarrier(
        CountDownLatch ready, CountDownLatch start, TransferService.Command command
    ) throws Exception {
        ready.countDown();
        start.await();
        return transfers.transfer(command);
    }
}

package com.example.financialcore.transfer.api;

public record TransferResponse(
    String transactionId,
    String businessTransactionId,
    String status,
    long fromBalance,
    long toBalance,
    boolean replayed
) {}

package com.example.financialcore.transfer.api;

import com.example.financialcore.transfer.TransferService;
import jakarta.validation.Valid;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
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 java.security.Principal;

@RestController
@RequestMapping("/api/transfers")
public class TransferController {
    private final TransferService transfers;
    public TransferController(TransferService transfers) { this.transfers = transfers; }

    @PostMapping
    ResponseEntity<TransferResponse> transfer(
        @Valid @RequestBody TransferRequest request, Principal principal
    ) {
        var result = transfers.transfer(new TransferService.Command(
            principal.getName(), request.transactionId(),
            request.fromAccountId(), request.toAccountId(), request.amount()));
        var body = new TransferResponse(
            request.transactionId(), result.businessTransactionId(), "COMPLETED",
            result.fromBalance(), result.toBalance(), result.replayed());
        return ResponseEntity.status(result.replayed() ? HttpStatus.OK : HttpStatus.CREATED).body(body);
    }
}

package com.example.financialcore.transfer.api;

import com.example.financialcore.PostgresIntegrationTestSupport;
import com.example.financialcore.account.Account;
import com.example.financialcore.account.AccountOpeningService;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.webmvc.test.autoconfigure.AutoConfigureMockMvc;
import org.springframework.jdbc.core.simple.JdbcClient;
import org.springframework.test.web.servlet.MockMvc;

import static org.assertj.core.api.Assertions.assertThat;
import static org.springframework.security.test.web.servlet.request.SecurityMockMvcRequestPostProcessors.httpBasic;
import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;

@SpringBootTest
@AutoConfigureMockMvc
class TransferControllerReplayTest extends PostgresIntegrationTestSupport {
    @Autowired MockMvc mvc;
    @Autowired JdbcClient jdbc;
    @Autowired AccountOpeningService openings;
    Account from;
    Account to;

    @BeforeEach void clean() {
        jdbc.sql("TRUNCATE idempotency_request,ledger_entry,business_tx,account RESTART IDENTITY CASCADE").update();
        from = openings.open("customer-1", "REPLAY-FROM", 10_000);
        to = openings.open("customer-2", "REPLAY-TO", 5_000);
    }

    @Test
    void firstIs201AndReplayIs200() throws Exception {
        String body = body("W12-REPLAY", 1_000);
        assertThat(performPost(body).getStatus()).isEqualTo(201);
        var replay = performPost(body);
        assertThat(replay.getStatus())
            .as("W12D6_RED_EXPECTED_REPLAY_200")
            .isEqualTo(200);
        assertThat(replay.getContentAsString()).contains("\"replayed\":true");
    }

    private org.springframework.mock.web.MockHttpServletResponse performPost(String body) throws Exception {
        return mvc.perform(post("/api/transfers")
                .with(httpBasic("customer-1", "password"))
                .contentType("application/json").content(body))
            .andReturn().getResponse();
    }

    private String body(String key, long amount) {
        return "{\"transactionId\":\"" + key + "\",\"fromAccountId\":" + from.getId()
            + ",\"toAccountId\":" + to.getId() + ",\"amount\":" + amount + "}";
    }
}