baiyue-firstleg/event-bridge-service/src/test/java/com/baiyue/eventbridge/service/OverseasEventDispatcherTest.java

package com.baiyue.eventbridge.service;

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 org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.springframework.test.util.ReflectionTestUtils;

import java.net.URI;
import java.time.Duration;
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 LastlegEventClient eventClient = mock(LastlegEventClient.class);
    private final FirstlegEventClient firstlegEventClient = mock(FirstlegEventClient.class);
    private OverseasEventDispatcher dispatcher;
    private BridgeInbox event;

    @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();
        ReflectionTestUtils.setField(dispatcher, "inboxMapper", inboxMapper);
        ReflectionTestUtils.setField(dispatcher, "attemptMapper", attemptMapper);
        ReflectionTestUtils.setField(dispatcher, "eventClient", eventClient);
        ReflectionTestUtils.setField(dispatcher, "firstlegEventClient", firstlegEventClient);
        ReflectionTestUtils.setField(dispatcher, "properties", properties);
        event = new BridgeInbox();
        event.setEventId("event-1");
        event.setSourceRegion("CN");
        event.setTargetRegion("OVERSEAS");
        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);
    }

    @Test
    void successfulDeliveryIsConfirmedAndAudited() throws Exception {
        when(eventClient.deliver(event)).thenReturn("200");
        when(inboxMapper.markProcessed(eq("event-1"), anyString())).thenReturn(1);

        dispatcher.dispatchOne("event-1");

        verify(inboxMapper).markProcessed(eq("event-1"), anyString());
        assertThat(updatedAttempt().getStatus()).isEqualTo("DELIVERED");
    }

    @Test
    void failedDeliveryIsScheduledForRetry() throws Exception {
        when(eventClient.deliver(event)).thenThrow(new IllegalStateException("network down"));

        dispatcher.dispatchOne("event-1");

        verify(inboxMapper).markFailed(eq("event-1"), anyString(), eq(3), anyLong(), eq("network down"));
        assertThat(updatedAttempt().getStatus()).isEqualTo("RETRY");
    }

    @Test
    void finalFailureIsMovedToDeadState() throws Exception {
        event.setRetryCount(2);
        when(eventClient.deliver(event)).thenThrow(new IllegalStateException("still down"));

        dispatcher.dispatchOne("event-1");

        assertThat(updatedAttempt().getStatus()).isEqualTo("DEAD");
    }

    @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.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);
        assertThat(updatedAttempt().getStatus()).isEqualTo("DELIVERED");
    }

    @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<BridgeDeliveryAttempt> 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<BridgeDeliveryAttempt> captor = ArgumentCaptor.forClass(BridgeDeliveryAttempt.class);
        verify(attemptMapper).updateById(captor.capture());
        return captor.getValue();
    }
}