package com.baiyue.eventbridge.service; import com.baiyue.common.event.EventTypes; import com.baiyue.eventbridge.config.EventDispatcherProperties; import com.baiyue.eventbridge.domain.BridgeDeliveryAttempt; import com.baiyue.eventbridge.domain.BridgeInbox; import com.baiyue.eventbridge.mapper.BridgeDeliveryAttemptMapper; import com.baiyue.eventbridge.mapper.BridgeInboxMapper; import com.baiyue.eventbridge.metrics.EventBridgeMetrics; import com.baiyue.common.web.CorrelationContext; import io.micrometer.core.instrument.simple.SimpleMeterRegistry; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.mockito.ArgumentCaptor; import org.springframework.test.util.ReflectionTestUtils; import org.slf4j.MDC; import java.net.URI; import java.time.Duration; import java.time.LocalDateTime; import java.util.List; import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.ArgumentMatchers.*; import static org.mockito.Mockito.*; class OverseasEventDispatcherTest { private final BridgeInboxMapper inboxMapper = mock(BridgeInboxMapper.class); private final BridgeDeliveryAttemptMapper attemptMapper = mock(BridgeDeliveryAttemptMapper.class); private final com.baiyue.eventbridge.service.LastlegEventClient eventClient = mock(com.baiyue.eventbridge.service.LastlegEventClient.class); private final com.baiyue.eventbridge.service.FirstlegEventClient firstlegEventClient = mock(com.baiyue.eventbridge.service.FirstlegEventClient.class); private final FinanceEventClient financeEventClient = mock(FinanceEventClient.class); private final SecurityProjectionEventClient securityProjectionEventClient = mock(SecurityProjectionEventClient.class); private OverseasEventDispatcher dispatcher; private BridgeInbox event; private SimpleMeterRegistry registry; @BeforeEach void setUp() { EventDispatcherProperties properties = new EventDispatcherProperties( URI.create("https://lastleg.internal/events"), 50, 3, Duration.ofSeconds(30), Duration.ofSeconds(5), Duration.ofMinutes(1), Duration.ofSeconds(3), Duration.ofSeconds(10)); dispatcher = new OverseasEventDispatcher(); registry = new SimpleMeterRegistry(); ReflectionTestUtils.setField(dispatcher, "inboxMapper", inboxMapper); ReflectionTestUtils.setField(dispatcher, "attemptMapper", attemptMapper); ReflectionTestUtils.setField(dispatcher, "eventClient", eventClient); ReflectionTestUtils.setField(dispatcher, "firstlegEventClient", firstlegEventClient); ReflectionTestUtils.setField(dispatcher, "financeEventClient", financeEventClient); ReflectionTestUtils.setField(dispatcher, "securityProjectionEventClient", securityProjectionEventClient); ReflectionTestUtils.setField(dispatcher, "properties", properties); ReflectionTestUtils.setField(dispatcher, "metrics", new EventBridgeMetrics(registry)); event = new BridgeInbox(); event.setEventId("event-1"); event.setEventType(EventTypes.FIRST_LEG_LOADING_PLAN_COMPLETED); event.setSourceRegion("CN"); event.setTargetRegion("OVERSEAS"); event.setRetryCount(0); event.setReceivedAt(LocalDateTime.now().minusSeconds(5)); when(inboxMapper.claim(eq("event-1"), anyString(), anyLong())).thenReturn(1); when(inboxMapper.selectById("event-1")).thenReturn(event); when(inboxMapper.nextAttemptNo("event-1")).thenReturn(1); } @Test void successfulDeliveryIsConfirmedAndAudited() throws Exception { event.setTraceId("550e8400-e29b-41d4-a716-446655440000"); when(eventClient.deliver(event)).thenAnswer(invocation -> { assertThat(MDC.get(CorrelationContext.TRACE_ID)) .isEqualTo("550e8400e29b41d4a716446655440000"); return "200"; }); when(inboxMapper.markProcessed(eq("event-1"), anyString())).thenReturn(1); dispatcher.dispatchOne("event-1"); verify(inboxMapper).markProcessed(eq("event-1"), anyString()); verify(eventClient).deliver(event); verifyNoInteractions(securityProjectionEventClient); assertThat(updatedAttempt().getStatus()).isEqualTo("DELIVERED"); assertThat(transportCount("DELIVERED")).isEqualTo(1.0); assertThat(registry.get(EventBridgeMetrics.DELIVERY_LATENCY_METRIC).timer().count()).isEqualTo(1); assertThat(MDC.get(CorrelationContext.TRACE_ID)).isNull(); } @Test void failedDeliveryIsScheduledForRetry() throws Exception { when(eventClient.deliver(event)).thenThrow(new IllegalStateException("network down")); when(inboxMapper.markFailed(eq("event-1"), anyString(), eq(3), anyLong(), eq("network down"))) .thenReturn(1); dispatcher.dispatchOne("event-1"); verify(inboxMapper).markFailed(eq("event-1"), anyString(), eq(3), eq(5L), eq("network down")); assertThat(updatedAttempt().getStatus()).isEqualTo("RETRY"); assertThat(transportCount("RETRY")).isEqualTo(1.0); } @Test void finalFailureIsMovedToDeadState() throws Exception { event.setRetryCount(2); when(eventClient.deliver(event)).thenThrow(new IllegalStateException("still down")); when(inboxMapper.markFailed(eq("event-1"), anyString(), eq(3), anyLong(), eq("still down"))) .thenReturn(1); dispatcher.dispatchOne("event-1"); verify(inboxMapper).markFailed(eq("event-1"), anyString(), eq(3), eq(20L), eq("still down")); assertThat(updatedAttempt().getStatus()).isEqualTo("DEAD"); assertThat(transportCount("DEAD")).isEqualTo(1.0); } @Test void exponentialBackoffIsCappedAtConfiguredMaximum() throws Exception { event.setRetryCount(4); when(eventClient.deliver(event)).thenThrow(new IllegalStateException("still down")); when(inboxMapper.markFailed(eq("event-1"), anyString(), eq(3), anyLong(), eq("still down"))) .thenReturn(1); dispatcher.dispatchOne("event-1"); verify(inboxMapper).markFailed(eq("event-1"), anyString(), eq(3), eq(60L), eq("still down")); } @Test void eventClaimedByAnotherInstanceIsSkipped() { when(inboxMapper.claim(eq("event-1"), anyString(), anyLong())).thenReturn(0); dispatcher.dispatchOne("event-1"); verifyNoInteractions(eventClient); verifyNoInteractions(firstlegEventClient); verify(attemptMapper, never()).insert(any(BridgeDeliveryAttempt.class)); } @Test void cnTargetIsDeliveredToFirstleg() throws Exception { event.setEventType(EventTypes.OVERSEAS_CONTAINER_UNLOADED); event.setSourceRegion("OVERSEAS"); event.setTargetRegion("CN"); when(firstlegEventClient.deliver(event)).thenReturn("200"); when(inboxMapper.markProcessed(eq("event-1"), anyString())).thenReturn(1); dispatcher.dispatchOne("event-1"); verify(firstlegEventClient).deliver(event); verifyNoInteractions(eventClient); verifyNoInteractions(securityProjectionEventClient); assertThat(updatedAttempt().getStatus()).isEqualTo("DELIVERED"); } @Test void overseasFeeSourceIsDeliveredToFinanceAndNeverFirstleg() throws Exception { event.setEventType(EventTypes.BUSINESS_FEE_SOURCE_CHANGED); event.setSourceRegion("OVERSEAS"); event.setTargetRegion("CN"); when(financeEventClient.deliver(event)).thenReturn("200"); when(inboxMapper.markProcessed(eq("event-1"), anyString())).thenReturn(1); dispatcher.dispatchOne("event-1"); verify(financeEventClient).deliver(event); verifyNoInteractions(firstlegEventClient); verifyNoInteractions(eventClient); assertThat(updatedAttempt().getStatus()).isEqualTo("DELIVERED"); } @Test void iamSecurityEventsAreDeliveredToSecurityProjectionAndNeverLastleg() throws Exception { for (String eventType : List.of(EventTypes.IAM_USER_STATUS_CHANGED, EventTypes.IAM_TENANT_STATUS_CHANGED, EventTypes.IAM_PERMISSION_VERSION_CHANGED)) { reset(inboxMapper, attemptMapper, eventClient, firstlegEventClient, securityProjectionEventClient); event.setEventType(eventType); when(inboxMapper.claim(eq("event-1"), anyString(), anyLong())).thenReturn(1); when(inboxMapper.selectById("event-1")).thenReturn(event); when(inboxMapper.nextAttemptNo("event-1")).thenReturn(1); when(inboxMapper.markProcessed(eq("event-1"), anyString())).thenReturn(1); when(securityProjectionEventClient.deliver(event)).thenReturn("200"); dispatcher.dispatchOne("event-1"); verify(securityProjectionEventClient).deliver(event); verifyNoInteractions(eventClient); verifyNoInteractions(firstlegEventClient); } } @Test void unsupportedTargetRegionIsNeverMisroutedToLastleg() { event.setTargetRegion("UNKNOWN"); when(inboxMapper.markFailed(eq("event-1"), anyString(), eq(3), anyLong(), contains("不支持"))) .thenReturn(1); dispatcher.dispatchOne("event-1"); verifyNoInteractions(eventClient); verifyNoInteractions(firstlegEventClient); verifyNoInteractions(securityProjectionEventClient); assertThat(updatedAttempt().getStatus()).isEqualTo("RETRY"); } @Test void missingEventTypeFailsWithExplicitDiagnosticAndIsNeverDelivered() { event.setEventType(null); when(inboxMapper.markFailed(eq("event-1"), anyString(), eq(3), anyLong(), eq("事件类型不能为空,无法选择下游消费者"))).thenReturn(1); dispatcher.dispatchOne("event-1"); verifyNoInteractions(eventClient); verifyNoInteractions(firstlegEventClient); verifyNoInteractions(securityProjectionEventClient); BridgeDeliveryAttempt attempt = updatedAttempt(); assertThat(attempt.getStatus()).isEqualTo("RETRY"); assertThat(attempt.getErrorMessage()).isEqualTo("事件类型不能为空,无法选择下游消费者"); } @Test void acknowledgedDeliveryWithLostLeaseIsNotRecordedAsRetryOrDead() throws Exception { when(eventClient.deliver(event)).thenReturn("200"); when(inboxMapper.markProcessed(eq("event-1"), anyString())).thenReturn(0); dispatcher.dispatchOne("event-1"); verify(inboxMapper, never()).markFailed(anyString(), anyString(), anyInt(), anyLong(), anyString()); BridgeDeliveryAttempt attempt = updatedAttempt(); assertThat(attempt.getStatus()).isEqualTo("ACKNOWLEDGED_LEASE_LOST"); assertThat(attempt.getResponseCode()).isEqualTo("200"); assertThat(attempt.getErrorMessage()).contains("租约已失效"); assertThat(transportCount("ACKNOWLEDGED_LEASE_LOST")).isEqualTo(1.0); } @Test void failedDeliveryAfterLeaseTakeoverDoesNotOverwriteNewOwnerState() throws Exception { when(eventClient.deliver(event)).thenThrow(new IllegalStateException("network down")); when(inboxMapper.markFailed(eq("event-1"), anyString(), eq(3), anyLong(), eq("network down"))) .thenReturn(0); dispatcher.dispatchOne("event-1"); assertThat(updatedAttempt().getStatus()).isEqualTo("FAILED_LEASE_LOST"); assertThat(transportCount("FAILED_LEASE_LOST")).isEqualTo(1.0); } @Test void downstream503RetriesThenDeliversWithDiagnosticAttemptsInBothDirections() throws Exception { for (String target : List.of("OVERSEAS", "CN")) { reset(inboxMapper, attemptMapper, eventClient, firstlegEventClient); event.setSourceRegion("CN".equals(target) ? "OVERSEAS" : "CN"); event.setTargetRegion(target); event.setRetryCount(0); when(inboxMapper.claim(eq("event-1"), anyString(), anyLong())).thenReturn(1); when(inboxMapper.selectById("event-1")).thenReturn(event); when(inboxMapper.nextAttemptNo("event-1")).thenReturn(1, 2); doAnswer(invocation -> { event.setRetryCount(1); event.setStatus("RETRY"); event.setLastError(invocation.getArgument(4)); event.setNextRetryAt(java.time.LocalDateTime.now().plusSeconds( ((Number) invocation.getArgument(3)).longValue())); return 1; }).when(inboxMapper).markFailed(eq("event-1"), anyString(), eq(3), anyLong(), anyString()); when(inboxMapper.markProcessed(eq("event-1"), anyString())).thenAnswer(invocation -> { event.setStatus("PROCESSED"); event.setLastError(null); event.setNextRetryAt(null); return 1; }); if ("CN".equals(target)) { when(firstlegEventClient.deliver(event)) .thenThrow(new IllegalStateException("HTTP 503 code=S0503 requestId=req-cn traceId=trace-cn")) .thenReturn("200"); } else { when(eventClient.deliver(event)) .thenThrow(new IllegalStateException("HTTP 503 code=S0503 requestId=req-overseas traceId=trace-overseas")) .thenReturn("200"); } dispatcher.dispatchOne("event-1"); assertThat(event.getStatus()).isEqualTo("RETRY"); assertThat(event.getNextRetryAt()).isNotNull(); assertThat(event.getLastError()).contains("HTTP 503", "requestId=", "traceId="); dispatcher.dispatchOne("event-1"); ArgumentCaptor attempts = ArgumentCaptor.forClass(BridgeDeliveryAttempt.class); verify(attemptMapper, times(2)).updateById(attempts.capture()); assertThat(attempts.getAllValues()).extracting(BridgeDeliveryAttempt::getAttemptNo) .containsExactly(1, 2); assertThat(attempts.getAllValues()).extracting(BridgeDeliveryAttempt::getStatus) .containsExactly("RETRY", "DELIVERED"); assertThat(attempts.getAllValues().get(0).getErrorMessage()) .contains("HTTP 503", "requestId=", "traceId="); assertThat(event.getStatus()).isEqualTo("PROCESSED"); assertThat(event.getLastError()).isNull(); } } private BridgeDeliveryAttempt updatedAttempt() { ArgumentCaptor captor = ArgumentCaptor.forClass(BridgeDeliveryAttempt.class); verify(attemptMapper).updateById(captor.capture()); return captor.getValue(); } private double transportCount(String result) { return registry.get(EventBridgeMetrics.TRANSPORT_METRIC) .tags("target_region", event.getTargetRegion(), "event_type", event.getEventType(), "result", result) .counter().count(); } }