205 lines
11 KiB
Java
205 lines
11 KiB
Java
package com.baiyue.eventbridge.service;
|
|
|
|
import com.baiyue.common.event.EventEnvelope;
|
|
import com.baiyue.common.event.EventPayloadCanonicalizer;
|
|
import com.baiyue.common.event.EventPayloadHash;
|
|
import com.baiyue.common.event.RegionCode;
|
|
import com.baiyue.eventbridge.domain.BridgeInbox;
|
|
import com.baiyue.eventbridge.metrics.EventBridgeMetrics;
|
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
|
import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
|
|
import org.junit.jupiter.api.Test;
|
|
import org.springframework.dao.DuplicateKeyException;
|
|
import org.springframework.test.util.ReflectionTestUtils;
|
|
|
|
import java.time.Instant;
|
|
import java.time.LocalDateTime;
|
|
|
|
import static org.assertj.core.api.Assertions.assertThat;
|
|
import static org.assertj.core.api.Assertions.assertThatThrownBy;
|
|
import static org.mockito.ArgumentMatchers.any;
|
|
import static org.mockito.Mockito.doThrow;
|
|
import static org.mockito.Mockito.mock;
|
|
import static org.mockito.Mockito.verify;
|
|
import static org.mockito.Mockito.when;
|
|
|
|
class EventInboxServiceTest {
|
|
private final ObjectMapper objectMapper = new ObjectMapper();
|
|
private final InboxStore store = mock(InboxStore.class);
|
|
private final SimpleMeterRegistry registry = new SimpleMeterRegistry();
|
|
private final EventBridgeMetrics metrics = new EventBridgeMetrics(registry);
|
|
private final EventInboxService service = service("OVERSEAS");
|
|
|
|
@Test
|
|
void 首次事件落库并返回已接收() {
|
|
EventEnvelope envelope = envelope("sha256:placeholder");
|
|
envelope = withHash(envelope);
|
|
|
|
assertThat(service.receive(envelope).result()).isEqualTo("已接收");
|
|
verify(store).insert(any(BridgeInbox.class));
|
|
assertThat(receiveCount("RECEIVED")).isEqualTo(1.0);
|
|
}
|
|
|
|
@Test
|
|
void 拒绝旧数字规范化摘要() {
|
|
EventEnvelope source = envelope("sha256:placeholder");
|
|
source.payload().withObject("/snapshot").put("weight", 30.0).put("volume", 0.3150);
|
|
String legacyHash = EventPayloadHash.sha256(EventPayloadCanonicalizer.canonicalizeLegacy(source.payload()));
|
|
EventEnvelope envelope = new EventEnvelope(source.eventId(), source.eventType(), source.eventVersion(),
|
|
source.occurredAt(), source.sourceRegion(), source.targetRegion(), source.tenantId(),
|
|
source.aggregateType(), source.aggregateId(), source.aggregateVersion(), source.handoverVersion(),
|
|
source.traceId(), source.correlationId(), source.causationId(), source.payload(), legacyHash);
|
|
|
|
assertThatThrownBy(() -> service.receive(envelope))
|
|
.isInstanceOf(IllegalArgumentException.class)
|
|
.hasMessageContaining("payloadHash");
|
|
}
|
|
|
|
@Test
|
|
void 相同事件和摘要重复接收不再次写入() {
|
|
EventEnvelope envelope = withHash(envelope("sha256:placeholder"));
|
|
BridgeInbox existing = existing(envelope, envelope.payloadHash());
|
|
when(store.findByEventId(envelope.eventId())).thenReturn(existing);
|
|
|
|
assertThat(service.receive(envelope).result()).isEqualTo("重复");
|
|
assertThat(receiveCount("DUPLICATE")).isEqualTo(1.0);
|
|
}
|
|
|
|
@Test
|
|
void 数据库毫秒精度截断后相同事件仍判定为重复() {
|
|
EventEnvelope source = envelope("sha256:placeholder");
|
|
EventEnvelope precise = new EventEnvelope(source.eventId(), source.eventType(), source.eventVersion(),
|
|
Instant.parse("2026-07-29T08:30:00.123456789Z"), source.sourceRegion(), source.targetRegion(),
|
|
source.tenantId(), source.aggregateType(), source.aggregateId(), source.aggregateVersion(),
|
|
source.handoverVersion(), source.traceId(), source.correlationId(), source.causationId(),
|
|
source.payload(), source.payloadHash());
|
|
EventEnvelope envelope = withHash(precise);
|
|
BridgeInbox existing = existing(envelope, envelope.payloadHash());
|
|
existing.setOccurredAt(Instant.parse("2026-07-29T08:30:00.123Z"));
|
|
when(store.findByEventId(envelope.eventId())).thenReturn(existing);
|
|
|
|
assertThat(service.receive(envelope).result()).isEqualTo("重复");
|
|
}
|
|
|
|
@Test
|
|
void 相同事件但摘要不同时拒绝覆盖() {
|
|
EventEnvelope envelope = withHash(envelope("sha256:placeholder"));
|
|
when(store.findByEventId(envelope.eventId())).thenReturn(existing(envelope, "sha256:" + "0".repeat(64)));
|
|
|
|
assertThatThrownBy(() -> service.receive(envelope))
|
|
.isInstanceOf(IllegalArgumentException.class)
|
|
.hasMessageContaining("拒绝覆盖");
|
|
assertThat(receiveCount("CONFLICT")).isEqualTo(1.0);
|
|
}
|
|
|
|
@Test
|
|
void 相同事件和摘要但聚合标识不同时拒绝覆盖() {
|
|
EventEnvelope envelope = withHash(envelope("sha256:placeholder"));
|
|
EventEnvelope conflicting = new EventEnvelope(envelope.eventId(), envelope.eventType(), envelope.eventVersion(),
|
|
envelope.occurredAt(), envelope.sourceRegion(), envelope.targetRegion(), envelope.tenantId(),
|
|
envelope.aggregateType(), "DIFFERENT", envelope.aggregateVersion(), envelope.handoverVersion(),
|
|
envelope.traceId(), envelope.correlationId(), envelope.causationId(), envelope.payload(), envelope.payloadHash());
|
|
when(store.findByEventId(conflicting.eventId())).thenReturn(existing(envelope, envelope.payloadHash()));
|
|
|
|
assertThatThrownBy(() -> service.receive(conflicting))
|
|
.isInstanceOf(IllegalArgumentException.class)
|
|
.hasMessageContaining("事件信封");
|
|
}
|
|
|
|
@Test
|
|
void 相同业务eventKey的新eventId只允许同负载重放并拒绝柜号漂移() {
|
|
EventEnvelope envelope = withHash(envelope("sha256:placeholder"));
|
|
BridgeInbox existing = existing(envelope, envelope.payloadHash());
|
|
when(store.findByEventId(envelope.eventId())).thenReturn(null);
|
|
when(store.findByIdempotencyKey("CN:flow-1")).thenReturn(existing);
|
|
EventEnvelope replay = new EventEnvelope("550e8400-e29b-41d4-a716-446655440001",
|
|
envelope.eventType(), envelope.eventVersion(), Instant.parse("2026-07-29T09:00:00Z"),
|
|
envelope.sourceRegion(), envelope.targetRegion(), envelope.tenantId(), envelope.aggregateType(),
|
|
envelope.aggregateId(), envelope.aggregateVersion(), envelope.handoverVersion(), "trace-2",
|
|
envelope.correlationId(), null, envelope.payload(), envelope.payloadHash());
|
|
assertThat(service.receive(replay).result()).isEqualTo("重复");
|
|
|
|
com.fasterxml.jackson.databind.node.ObjectNode driftPayload =
|
|
(com.fasterxml.jackson.databind.node.ObjectNode) envelope.payload().deepCopy();
|
|
driftPayload.put("containerNo", "DIFFERENT");
|
|
String driftHash = EventPayloadHash.sha256(EventPayloadCanonicalizer.canonicalize(driftPayload));
|
|
EventEnvelope drift = new EventEnvelope("550e8400-e29b-41d4-a716-446655440002",
|
|
envelope.eventType(), envelope.eventVersion(), envelope.occurredAt(), envelope.sourceRegion(),
|
|
envelope.targetRegion(), envelope.tenantId(), envelope.aggregateType(), envelope.aggregateId(),
|
|
envelope.aggregateVersion(), envelope.handoverVersion(), "trace-3", envelope.correlationId(),
|
|
null, driftPayload, driftHash);
|
|
assertThatThrownBy(() -> service.receive(drift))
|
|
.hasMessageContaining("eventKey").hasMessageContaining("漂移");
|
|
}
|
|
|
|
@Test
|
|
void 并发插入冲突后按既有记录判定为重复() {
|
|
EventEnvelope envelope = withHash(envelope("sha256:placeholder"));
|
|
when(store.findByEventId(envelope.eventId())).thenReturn(null, existing(envelope, envelope.payloadHash()));
|
|
doThrow(new DuplicateKeyException("并发重复")).when(store).insert(any());
|
|
|
|
assertThat(service.receive(envelope).result()).isEqualTo("重复");
|
|
}
|
|
|
|
@Test
|
|
void 目标区域与当前部署区域不一致时拒绝接收() {
|
|
EventInboxService cnService = service("CN");
|
|
EventEnvelope envelope = withHash(envelope("sha256:placeholder"));
|
|
|
|
assertThatThrownBy(() -> cnService.receive(envelope))
|
|
.isInstanceOf(IllegalArgumentException.class)
|
|
.hasMessageContaining("目标区域");
|
|
}
|
|
|
|
private EventEnvelope envelope(String hash) {
|
|
return new EventEnvelope("550e8400-e29b-41d4-a716-446655440000", "OrderApproved", 1,
|
|
Instant.parse("2026-07-29T08:30:00Z"), RegionCode.CN, RegionCode.OVERSEAS,
|
|
"20001", "ORDER", "90001", 1, null, "trace-1", "flow-1", null,
|
|
objectMapper.createObjectNode().put("orderId", "90001"), hash);
|
|
}
|
|
|
|
private EventInboxService service(String region) {
|
|
EventInboxService service = new EventInboxService();
|
|
ReflectionTestUtils.setField(service, "inboxStore", store);
|
|
ReflectionTestUtils.setField(service, "objectMapper", objectMapper);
|
|
ReflectionTestUtils.setField(service, "metrics", metrics);
|
|
ReflectionTestUtils.setField(service, "localRegion", region);
|
|
return service;
|
|
}
|
|
|
|
private double receiveCount(String result) {
|
|
return registry.get(EventBridgeMetrics.RECEIVE_METRIC)
|
|
.tags("target_region", "OVERSEAS", "event_type", "OrderApproved", "result", result)
|
|
.counter().count();
|
|
}
|
|
|
|
private EventEnvelope withHash(EventEnvelope source) {
|
|
String hash = EventPayloadHash.sha256(EventPayloadCanonicalizer.canonicalize(source.payload()));
|
|
return new EventEnvelope(source.eventId(), source.eventType(), source.eventVersion(), source.occurredAt(),
|
|
source.sourceRegion(), source.targetRegion(), source.tenantId(), source.aggregateType(),
|
|
source.aggregateId(), source.aggregateVersion(), source.handoverVersion(), source.traceId(),
|
|
source.correlationId(), source.causationId(), source.payload(), hash);
|
|
}
|
|
|
|
private BridgeInbox existing(EventEnvelope envelope, String hash) {
|
|
BridgeInbox inbox = new BridgeInbox();
|
|
inbox.setEventId(envelope.eventId());
|
|
inbox.setEventType(envelope.eventType());
|
|
inbox.setEventVersion(envelope.eventVersion());
|
|
inbox.setSourceRegion(envelope.sourceRegion().name());
|
|
inbox.setTargetRegion(envelope.targetRegion().name());
|
|
inbox.setCustomerId(envelope.tenantId());
|
|
inbox.setAggregateType(envelope.aggregateType());
|
|
inbox.setAggregateId(envelope.aggregateId());
|
|
inbox.setAggregateVersion(envelope.aggregateVersion());
|
|
inbox.setHandoverVersion(envelope.handoverVersion());
|
|
inbox.setOccurredAt(envelope.occurredAt());
|
|
inbox.setTraceId(envelope.traceId());
|
|
inbox.setCorrelationId(envelope.correlationId());
|
|
inbox.setCausationId(envelope.causationId());
|
|
inbox.setPayloadHash(hash);
|
|
inbox.setReceivedAt(LocalDateTime.of(2026, 7, 29, 18, 0));
|
|
return inbox;
|
|
}
|
|
}
|