baiyue-firstleg/iam/src/test/java/com/baiyue/iam/notification/NotificationSchedulerRecoveryTest.java

package com.baiyue.iam.notification;

import com.baiyue.iam.notification.mapper.NotificationTaskMapper;
import com.baiyue.iam.notification.service.NotificationScheduler;
import com.baiyue.iam.notification.service.NotificationService;
import org.apache.ibatis.annotations.Update;
import org.junit.jupiter.api.Test;
import org.springframework.test.util.ReflectionTestUtils;

import java.util.List;

import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.*;

class NotificationSchedulerRecoveryTest {
    @Test void claimSqlAtomicallyIncrementsAttemptsAndEnforcesDueTimeAndLimit() throws Exception {
        Update scheduled = NotificationTaskMapper.class
                .getMethod("claimScheduled", Long.class, int.class, java.time.LocalDateTime.class)
                .getAnnotation(Update.class);
        String scheduledSql = String.join(" ", scheduled.value()).replaceAll("\\s+", " ").toUpperCase();
        assertThat(scheduledSql).contains("STATUS = 'DISPATCHING'", "ATTEMPT_COUNT = ATTEMPT_COUNT + 1",
                "STATUS = 'SCHEDULED'", "SCHEDULED_AT <= #{NOW}", "ATTEMPT_COUNT < #{MAXATTEMPTS}");

        Update retrying = NotificationTaskMapper.class
                .getMethod("claimRetrying", Long.class, int.class, java.time.LocalDateTime.class,
                        java.time.LocalDateTime.class)
                .getAnnotation(Update.class);
        String retryingSql = String.join(" ", retrying.value()).replaceAll("\\s+", " ").toUpperCase();
        assertThat(retryingSql).contains("STATUS = 'DISPATCHING'", "ATTEMPT_COUNT = ATTEMPT_COUNT + 1",
                "STATUS = 'RETRYING'", "UPDATED_AT <= #{DUEBEFORE}", "ATTEMPT_COUNT < #{MAXATTEMPTS}");
    }

    @Test void recoverySqlRequiresTimeoutAndNoSuccessfulDeliveryEvidence() throws Exception {
        Update retry = NotificationTaskMapper.class
                .getMethod("retryTimedOutDispatchingWithoutDeliveryEvidence", java.time.LocalDateTime.class,
                        java.time.LocalDateTime.class, int.class)
                .getAnnotation(Update.class);
        String retrySql = String.join(" ", retry.value()).replaceAll("\\s+", " ").toUpperCase();

        assertThat(retrySql).contains("TASK.STATUS = 'DISPATCHING'", "TASK.UPDATED_AT <= #{STALEBEFORE}",
                "TASK.SENT_AT IS NULL", "RECIPIENT.STATUS = 'DELIVERED'", "IAM_NOTIFICATION_INBOX")
                .contains("SET TASK.STATUS = 'RETRYING'", "TASK.ATTEMPT_COUNT < #{MAXATTEMPTS}")
                .doesNotContain("ATTEMPT_COUNT = TASK.ATTEMPT_COUNT + 1");

        Update dead = NotificationTaskMapper.class
                .getMethod("deadTimedOutDispatchingWithDeliveryEvidence", java.time.LocalDateTime.class, java.time.LocalDateTime.class)
                .getAnnotation(Update.class);
        String deadSql = String.join(" ", dead.value()).replaceAll("\\s+", " ").toUpperCase();
        assertThat(deadSql).contains("SET TASK.STATUS = 'DEAD'", "TASK.STATUS = 'DISPATCHING'",
                "TASK.UPDATED_AT <= #{STALEBEFORE}", "RECIPIENT.STATUS = 'DELIVERED'", "IAM_NOTIFICATION_INBOX",
                "DISPATCH_TIMEOUT_WITH_DELIVERY_EVIDENCE");

        Update attemptLimit = NotificationTaskMapper.class
                .getMethod("deadTimedOutDispatchingAtAttemptLimit", java.time.LocalDateTime.class,
                        java.time.LocalDateTime.class, int.class)
                .getAnnotation(Update.class);
        String attemptLimitSql = String.join(" ", attemptLimit.value()).replaceAll("\\s+", " ").toUpperCase();
        assertThat(attemptLimitSql).contains("SET TASK.STATUS = 'DEAD'",
                "TASK.ATTEMPT_COUNT >= #{MAXATTEMPTS}", "DISPATCH_TIMEOUT_ATTEMPT_LIMIT");

        Update failed = NotificationTaskMapper.class
                .getMethod("failDispatch", Long.class, int.class, String.class, java.time.LocalDateTime.class)
                .getAnnotation(Update.class);
        String failedSql = String.join(" ", failed.value()).replaceAll("\\s+", " ").toUpperCase();
        assertThat(failedSql).contains("STATUS = CASE WHEN ATTEMPT_COUNT >= #{MAXATTEMPTS} THEN 'DEAD' ELSE 'RETRYING' END",
                "LAST_ERROR = #{ERROR}", "STATUS = 'DISPATCHING'");
    }

    @Test void schedulerRecoversBeforeClaimingScheduledAndRetryingTasks() {
        NotificationService service = mock(NotificationService.class);
        NotificationScheduler scheduler = new NotificationScheduler();
        ReflectionTestUtils.setField(scheduler, "service", service);
        ReflectionTestUtils.setField(scheduler, "dispatchTimeoutSeconds", 300L);
        when(service.dueScheduledTaskIds()).thenReturn(List.of(11L));
        when(service.retryingTaskIds()).thenReturn(List.of(12L));

        scheduler.dispatch();

        var ordered = inOrder(service);
        ordered.verify(service).recoverTimedOutDispatching(300L);
        ordered.verify(service).dueScheduledTaskIds();
        ordered.verify(service).dispatchScheduled(11L);
        ordered.verify(service).retryingTaskIds();
        ordered.verify(service).dispatchRetrying(12L);
    }

    @Test void oneTaskFailureDoesNotBlockRemainingTasksOrRetryQueue() {
        NotificationService service = mock(NotificationService.class);
        NotificationScheduler scheduler = new NotificationScheduler();
        ReflectionTestUtils.setField(scheduler, "service", service);
        ReflectionTestUtils.setField(scheduler, "dispatchTimeoutSeconds", 300L);
        when(service.dueScheduledTaskIds()).thenReturn(List.of(11L, 12L));
        when(service.dispatchScheduled(11L)).thenThrow(new IllegalStateException("failed"));
        when(service.retryingTaskIds()).thenReturn(List.of(13L));

        scheduler.dispatch();

        verify(service).dispatchScheduled(12L);
        verify(service).dispatchRetrying(13L);
    }
}