206 lines
11 KiB
Java
206 lines
11 KiB
Java
package com.baiyue.finance;
|
|
|
|
import com.baiyue.common.event.*;
|
|
import com.baiyue.finance.dto.FinanceWalletDtos.WalletHold;
|
|
import com.baiyue.finance.service.FinanceEventException;
|
|
import com.baiyue.finance.service.FinanceWalletService;
|
|
import com.baiyue.finance.service.WalletCommandEventService;
|
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
|
import org.h2.jdbcx.JdbcDataSource;
|
|
import org.junit.jupiter.api.BeforeEach;
|
|
import org.junit.jupiter.api.Test;
|
|
import org.springframework.dao.TransientDataAccessResourceException;
|
|
import org.springframework.jdbc.core.JdbcTemplate;
|
|
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
|
|
|
|
import java.math.BigDecimal;
|
|
import java.time.Instant;
|
|
import java.util.UUID;
|
|
import java.util.concurrent.*;
|
|
import java.util.concurrent.atomic.AtomicBoolean;
|
|
|
|
import static org.assertj.core.api.Assertions.*;
|
|
import static org.mockito.ArgumentMatchers.*;
|
|
import static org.mockito.Mockito.*;
|
|
|
|
/** Real JDBC transactions and row locks; the wallet boundary is a durable idempotent test double. */
|
|
class WalletCommandTransactionTest {
|
|
private JdbcTemplate jdbc;
|
|
private FinanceWalletService wallet;
|
|
private WalletCommandEventService service;
|
|
private final ObjectMapper mapper = new ObjectMapper().findAndRegisterModules();
|
|
|
|
@BeforeEach
|
|
void setup() {
|
|
JdbcDataSource source = new JdbcDataSource();
|
|
source.setURL("jdbc:h2:mem:" + UUID.randomUUID() + ";MODE=MySQL;DB_CLOSE_DELAY=-1;LOCK_TIMEOUT=5000");
|
|
jdbc = spy(new JdbcTemplate(source));
|
|
jdbc.execute("CREATE TABLE fin_wallet_command_inbox(event_id VARCHAR(128) PRIMARY KEY,event_type VARCHAR(100),payload_hash VARCHAR(128),payload_json JSON,status VARCHAR(32),received_at TIMESTAMP,result_json JSON,processed_at TIMESTAMP)");
|
|
jdbc.execute("CREATE TABLE fin_wallet_event_outbox(event_id VARCHAR(128) PRIMARY KEY,event_key VARCHAR(200) UNIQUE,payload_json JSON,status VARCHAR(32),next_retry_at TIMESTAMP,created_at TIMESTAMP,updated_at TIMESTAMP)");
|
|
jdbc.execute("CREATE TABLE wallet_effect(idempotency_key VARCHAR(128) PRIMARY KEY,amount DECIMAL(18,2))");
|
|
wallet = mock(FinanceWalletService.class);
|
|
service = new WalletCommandEventService(jdbc, mapper, wallet, new DataSourceTransactionManager(source));
|
|
doAnswer(invocation -> {
|
|
String key = invocation.getArgument(1);
|
|
// Models the existing FinanceWalletService's durable replay behavior.
|
|
if (jdbc.queryForObject("SELECT COUNT(*) FROM wallet_effect WHERE idempotency_key=?", Integer.class, key) == 0) {
|
|
jdbc.update("INSERT INTO wallet_effect VALUES(?,?)", key, new BigDecimal("10.00"));
|
|
}
|
|
return hold();
|
|
}).when(wallet).freeze(anyLong(), anyString(), any());
|
|
}
|
|
|
|
@Test
|
|
void transientFailureRollsBackMoneyAndCanRetryTheOriginalEvent() {
|
|
EventEnvelope event = event(UUID.randomUUID().toString(), "10.00");
|
|
AtomicBoolean first = new AtomicBoolean(true);
|
|
doAnswer(invocation -> {
|
|
jdbc.update("INSERT INTO wallet_effect VALUES('freeze-1',10.00)");
|
|
if (first.getAndSet(false)) throw new TransientDataAccessResourceException("temporary outage");
|
|
return hold();
|
|
}).when(wallet).freeze(anyLong(), anyString(), any());
|
|
|
|
assertThatThrownBy(() -> service.consume(event)).isInstanceOf(TransientDataAccessResourceException.class);
|
|
assertThat(count("wallet_effect")).isZero();
|
|
assertThat(count("fin_wallet_event_outbox")).isZero();
|
|
assertThat(inboxStatus(event)).isEqualTo("PROCESSING");
|
|
assertThat(service.consume(event).eventId()).isEqualTo(event.eventId());
|
|
assertThat(count("wallet_effect")).isEqualTo(1);
|
|
assertThat(count("fin_wallet_event_outbox")).isEqualTo(1);
|
|
assertThat(inboxStatus(event)).isEqualTo("SUCCEEDED");
|
|
}
|
|
|
|
@Test
|
|
void businessRejectionRollsBackMoneyAndPublishesExactlyOneFailedResult() {
|
|
EventEnvelope event = event(UUID.randomUUID().toString(), "10.00");
|
|
doAnswer(invocation -> {
|
|
jdbc.update("INSERT INTO wallet_effect VALUES('freeze-1',10.00)");
|
|
throw new FinanceEventException("FIN-WALLET-INSUFFICIENT", "insufficient balance");
|
|
}).when(wallet).freeze(anyLong(), anyString(), any());
|
|
service.consume(event);
|
|
service.consume(event);
|
|
assertThat(count("wallet_effect")).isZero();
|
|
assertThat(count("fin_wallet_event_outbox")).isEqualTo(1);
|
|
assertThat(inboxStatus(event)).isEqualTo("FAILED");
|
|
verify(wallet, times(1)).freeze(anyLong(), anyString(), any());
|
|
}
|
|
|
|
@Test
|
|
void retryableFinanceCodeDoesNotPublishATerminalResult() {
|
|
EventEnvelope event = event(UUID.randomUUID().toString(), "10.00");
|
|
doThrow(new FinanceEventException("FIN-IDEMPOTENCY-PROCESSING", "in progress"))
|
|
.when(wallet).freeze(anyLong(), anyString(), any());
|
|
assertThatThrownBy(() -> service.consume(event)).isInstanceOf(FinanceEventException.class);
|
|
assertThat(inboxStatus(event)).isEqualTo("PROCESSING");
|
|
assertThat(count("fin_wallet_event_outbox")).isZero();
|
|
}
|
|
|
|
@Test
|
|
void failureAfterResultInsertRollsBackMoneyAndResultAndCanRetry() {
|
|
EventEnvelope event = event(UUID.randomUUID().toString(), "10.00");
|
|
AtomicBoolean first = new AtomicBoolean(true);
|
|
doAnswer(invocation -> {
|
|
if (first.getAndSet(false)) throw new TransientDataAccessResourceException("result commit failed");
|
|
return invocation.callRealMethod();
|
|
}).when(jdbc).update(startsWith("UPDATE fin_wallet_command_inbox SET status="), any(Object[].class));
|
|
|
|
assertThatThrownBy(() -> service.consume(event)).isInstanceOf(TransientDataAccessResourceException.class);
|
|
assertThat(count("wallet_effect")).isZero();
|
|
assertThat(count("fin_wallet_event_outbox")).isZero();
|
|
assertThat(inboxStatus(event)).isEqualTo("PROCESSING");
|
|
assertThat(service.consume(event).eventId()).isEqualTo(event.eventId());
|
|
assertThat(count("wallet_effect")).isEqualTo(1);
|
|
assertThat(count("fin_wallet_event_outbox")).isEqualTo(1);
|
|
assertThat(inboxStatus(event)).isEqualTo("SUCCEEDED");
|
|
verify(wallet, times(2)).freeze(eq(31L), eq("freeze-1"), any());
|
|
}
|
|
|
|
@Test
|
|
void differentPayloadCannotReplaceIdentityAfterAResultFailure() {
|
|
EventEnvelope event = event(UUID.randomUUID().toString(), "10.00");
|
|
doThrow(new TransientDataAccessResourceException("outbox unavailable"))
|
|
.when(jdbc).update(startsWith("INSERT INTO fin_wallet_event_outbox"), any(Object[].class));
|
|
assertThatThrownBy(() -> service.consume(event)).isInstanceOf(TransientDataAccessResourceException.class);
|
|
assertThatThrownBy(() -> service.consume(event(event.eventId(), "20.00")))
|
|
.isInstanceOf(FinanceEventException.class)
|
|
.extracting(e -> ((FinanceEventException) e).code()).isEqualTo("FIN-EVT-002");
|
|
assertThat(count("wallet_effect")).isZero();
|
|
verify(wallet, times(1)).freeze(anyLong(), anyString(), any());
|
|
}
|
|
|
|
@Test
|
|
void concurrentSamePayloadExecutesOnce() throws Exception {
|
|
runConcurrent(false);
|
|
}
|
|
|
|
@Test
|
|
void concurrentDifferentPayloadIsRejectedBeforeWalletExecution() throws Exception {
|
|
runConcurrent(true);
|
|
}
|
|
|
|
private void runConcurrent(boolean differentPayload) throws Exception {
|
|
EventEnvelope firstEvent = event(UUID.randomUUID().toString(), "10.00");
|
|
EventEnvelope secondEvent = differentPayload ? event(firstEvent.eventId(), "20.00") : firstEvent;
|
|
CountDownLatch enteredWallet = new CountDownLatch(1);
|
|
CountDownLatch releaseWallet = new CountDownLatch(1);
|
|
CountDownLatch secondStarted = new CountDownLatch(1);
|
|
doAnswer(invocation -> {
|
|
enteredWallet.countDown();
|
|
if (!releaseWallet.await(5, TimeUnit.SECONDS)) throw new AssertionError("test timed out");
|
|
jdbc.update("INSERT INTO wallet_effect VALUES('freeze-1',10.00)");
|
|
return hold();
|
|
}).when(wallet).freeze(anyLong(), anyString(), any());
|
|
ExecutorService executor = Executors.newFixedThreadPool(2);
|
|
try {
|
|
Future<com.baiyue.common.event.DeliveryAck> first = executor.submit(() -> service.consume(firstEvent));
|
|
assertThat(enteredWallet.await(5, TimeUnit.SECONDS)).isTrue();
|
|
Future<com.baiyue.common.event.DeliveryAck> second = executor.submit(() -> {
|
|
secondStarted.countDown();
|
|
return service.consume(secondEvent);
|
|
});
|
|
assertThat(secondStarted.await(5, TimeUnit.SECONDS)).isTrue();
|
|
assertThatThrownBy(() -> second.get(150, TimeUnit.MILLISECONDS)).isInstanceOf(TimeoutException.class);
|
|
releaseWallet.countDown();
|
|
assertThat(first.get(5, TimeUnit.SECONDS).eventId()).isEqualTo(firstEvent.eventId());
|
|
if (differentPayload) {
|
|
assertThatThrownBy(() -> second.get(5, TimeUnit.SECONDS))
|
|
.isInstanceOf(ExecutionException.class).hasCauseInstanceOf(FinanceEventException.class)
|
|
.satisfies(e -> assertThat(((FinanceEventException) e.getCause()).code()).isEqualTo("FIN-EVT-002"));
|
|
} else {
|
|
assertThat(second.get(5, TimeUnit.SECONDS).eventId()).isEqualTo(firstEvent.eventId());
|
|
}
|
|
verify(wallet, times(1)).freeze(anyLong(), anyString(), any());
|
|
assertThat(count("wallet_effect")).isEqualTo(1);
|
|
assertThat(count("fin_wallet_event_outbox")).isEqualTo(1);
|
|
assertThat(inboxStatus(firstEvent)).isEqualTo("SUCCEEDED");
|
|
} finally {
|
|
releaseWallet.countDown();
|
|
executor.shutdownNow();
|
|
assertThat(executor.awaitTermination(5, TimeUnit.SECONDS)).isTrue();
|
|
}
|
|
}
|
|
|
|
private int count(String table) {
|
|
return jdbc.queryForObject("SELECT COUNT(*) FROM " + table, Integer.class);
|
|
}
|
|
|
|
private String inboxStatus(EventEnvelope event) {
|
|
return jdbc.queryForObject("SELECT status FROM fin_wallet_command_inbox WHERE event_id=?", String.class, event.eventId());
|
|
}
|
|
|
|
private EventEnvelope event(String id, String amount) {
|
|
var command = new WalletCommandRequestedV1("FREEZE", 31L, "OVERSEAS_PROCUREMENT_FEE", "PO-1",
|
|
new BigDecimal(amount), "fee", "freeze-1", Instant.parse("2026-09-05T01:00:00Z"), null);
|
|
var payload = mapper.valueToTree(command);
|
|
return new EventEnvelope(id, EventTypes.WALLET_COMMAND_REQUESTED, WalletCommandRequestedV1.VERSION,
|
|
Instant.now(), "31", "WALLET_HOLD", "PO-1", 1,
|
|
null, UUID.randomUUID().toString(), UUID.randomUUID().toString(), null, payload,
|
|
EventPayloadHash.sha256(EventPayloadCanonicalizer.canonicalize(payload)));
|
|
}
|
|
|
|
private WalletHold hold() {
|
|
return new WalletHold(9L, 31L, "OVERSEAS_PROCUREMENT_FEE", "PO-1", new BigDecimal("10.00"),
|
|
new BigDecimal("10.00"), BigDecimal.ZERO, "FROZEN", 71L, null, null, null, null);
|
|
}
|
|
}
|