优化

iam/src/main/java/com/baiyue/iam/notification/mapper/NotificationTaskMapper.java

@@ -2,4 +2,48 @@ package com.baiyue.iam.notification.mapper;
 import com.baiyue.iam.notification.entity.NotificationTask;
 import com.baomidou.mybatisplus.core.mapper.BaseMapper;
 import org.apache.ibatis.annotations.Mapper;
-@Mapper public interface NotificationTaskMapper extends BaseMapper<NotificationTask> { }
+import org.apache.ibatis.annotations.Update;
+import org.apache.ibatis.annotations.Param;
+import java.time.LocalDateTime;
+@Mapper public interface NotificationTaskMapper extends BaseMapper<NotificationTask> {
+    @Update("""
+            UPDATE iam_notification_task task
+               SET task.status = 'DEAD',
+                   task.last_error = 'DISPATCH_TIMEOUT_WITH_DELIVERY_EVIDENCE',
+                   task.updated_at = #{now}
+             WHERE task.status = 'DISPATCHING'
+               AND task.updated_at <= #{staleBefore}
+               AND task.sent_at IS NULL
+               AND (
+                   EXISTS (
+                       SELECT 1 FROM iam_notification_recipient recipient
+                        WHERE recipient.task_id = task.id AND recipient.status = 'DELIVERED'
+                   )
+                   OR EXISTS (
+                       SELECT 1 FROM iam_notification_inbox inbox WHERE inbox.task_id = task.id
+                   )
+               )
+            """)
+    int deadTimedOutDispatchingWithDeliveryEvidence(@Param("staleBefore") LocalDateTime staleBefore,
+                                                     @Param("now") LocalDateTime now);
+
+    @Update("""
+            UPDATE iam_notification_task task
+               SET task.status = 'RETRYING',
+                   task.attempt_count = task.attempt_count + 1,
+                   task.last_error = 'DISPATCH_TIMEOUT_RECOVERED',
+                   task.updated_at = #{now}
+             WHERE task.status = 'DISPATCHING'
+               AND task.updated_at <= #{staleBefore}
+               AND task.sent_at IS NULL
+               AND NOT EXISTS (
+                   SELECT 1 FROM iam_notification_recipient recipient
+                    WHERE recipient.task_id = task.id AND recipient.status = 'DELIVERED'
+               )
+               AND NOT EXISTS (
+                   SELECT 1 FROM iam_notification_inbox inbox WHERE inbox.task_id = task.id
+               )
+            """)
+    int retryTimedOutDispatchingWithoutDeliveryEvidence(@Param("staleBefore") LocalDateTime staleBefore,
+                                                         @Param("now") LocalDateTime now);
+}

iam/src/main/java/com/baiyue/iam/notification/repository/NotificationRepository.java

@@ -54,6 +54,15 @@ public class NotificationRepository {
     public boolean claimScheduled(Long taskId) { return tasks.update(null,new LambdaUpdateWrapper<NotificationTask>()
             .eq(NotificationTask::getId,taskId).eq(NotificationTask::getStatus,"SCHEDULED")
             .set(NotificationTask::getStatus,"DISPATCHING").set(NotificationTask::getUpdatedAt,java.time.LocalDateTime.now()))==1; }
+    public int recoverTimedOutDispatching(java.time.LocalDateTime staleBefore,java.time.LocalDateTime now) {
+        int dead = tasks.deadTimedOutDispatchingWithDeliveryEvidence(staleBefore,now);
+        return dead + tasks.retryTimedOutDispatchingWithoutDeliveryEvidence(staleBefore,now);
+    }
+    public List<NotificationTask> retryingTasks() { return tasks.selectList(new LambdaQueryWrapper<NotificationTask>()
+            .eq(NotificationTask::getStatus,"RETRYING").orderByAsc(NotificationTask::getUpdatedAt).last("LIMIT 100")); }
+    public boolean claimRetrying(Long taskId) { return tasks.update(null,new LambdaUpdateWrapper<NotificationTask>()
+            .eq(NotificationTask::getId,taskId).eq(NotificationTask::getStatus,"RETRYING")
+            .set(NotificationTask::getStatus,"DISPATCHING").set(NotificationTask::getUpdatedAt,java.time.LocalDateTime.now()))==1; }
     public void insertInbox(NotificationInbox value) { inbox.insert(value); }
     public PageResult<NotificationInbox> pageInbox(long tenantId,long membershipId,String app,long page,long size,String status,String category) {
         return PageResult.from(inbox.selectPage(new Page<>(page,size), new LambdaQueryWrapper<NotificationInbox>()

iam/src/main/java/com/baiyue/iam/notification/service/NotificationScheduler.java

@@ -1,10 +1,29 @@
 package com.baiyue.iam.notification.service;
 import jakarta.annotation.Resource;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.List;
+import java.util.function.Function;
 @Component @ConditionalOnProperty(name="baiyue.notification.scheduler-enabled",havingValue="true")
 public class NotificationScheduler {
+    private static final Logger log = LoggerFactory.getLogger(NotificationScheduler.class);
     @Resource private NotificationService service;
-    @Scheduled(fixedDelayString="${baiyue.notification.scheduler-delay-ms:5000}") public void dispatch(){service.dispatchDue();}
+    @Value("${baiyue.notification.dispatch-timeout-seconds:300}") private long dispatchTimeoutSeconds;
+    @Scheduled(fixedDelayString="${baiyue.notification.scheduler-delay-ms:5000}") public void dispatch(){
+        service.recoverTimedOutDispatching(dispatchTimeoutSeconds);
+        dispatchTasks("scheduled", service.dueScheduledTaskIds(), service::dispatchScheduled);
+        dispatchTasks("retrying", service.retryingTaskIds(), service::dispatchRetrying);
+    }
+    private void dispatchTasks(String source, List<Long> taskIds, Function<Long, Boolean> dispatcher) {
+        for (Long taskId : taskIds) {
+            try { dispatcher.apply(taskId); }
+            catch (RuntimeException failure) {
+                log.error("Notification {} task dispatch failed; taskId={}", source, taskId, failure);
+            }
+        }
+    }
 }

iam/src/main/java/com/baiyue/iam/notification/service/NotificationService.java

@@ -180,11 +180,38 @@ public class NotificationService {
     }
 
     @Transactional
-    public void dispatchDue() {
-        for (NotificationTask task : repository.dueTasks()) {
-            if (!repository.claimScheduled(task.getId())) continue;
-            task.setStatus("DISPATCHING"); deliver(task, LocalDateTime.now());
-        }
+    public int recoverTimedOutDispatching(long timeoutSeconds) {
+        if (timeoutSeconds < 1) throw new IllegalArgumentException("Dispatch timeout must be positive");
+        LocalDateTime now = eventNow();
+        return repository.recoverTimedOutDispatching(now.minusSeconds(timeoutSeconds), now);
+    }
+
+    @Transactional(readOnly = true)
+    public List<Long> dueScheduledTaskIds() {
+        return repository.dueTasks().stream().map(NotificationTask::getId).toList();
+    }
+
+    @Transactional(readOnly = true)
+    public List<Long> retryingTaskIds() {
+        return repository.retryingTasks().stream().map(NotificationTask::getId).toList();
+    }
+
+    @Transactional
+    public boolean dispatchScheduled(Long taskId) {
+        if (!repository.claimScheduled(taskId)) return false;
+        NotificationTask task = requiredTask(taskId);
+        task.setStatus("DISPATCHING");
+        deliver(task, LocalDateTime.now());
+        return true;
+    }
+
+    @Transactional
+    public boolean dispatchRetrying(Long taskId) {
+        if (!repository.claimRetrying(taskId)) return false;
+        NotificationTask task = requiredTask(taskId);
+        task.setStatus("DISPATCHING");
+        deliver(task, LocalDateTime.now());
+        return true;
     }
 
     @Transactional(readOnly = true)

iam/src/main/resources/application.yml

@@ -25,6 +25,7 @@ spring:
 baiyue:
   notification:
     scheduler-enabled: ${NOTIFICATION_SCHEDULER_ENABLED:false}
+    dispatch-timeout-seconds: ${NOTIFICATION_DISPATCH_TIMEOUT_SECONDS:300}
     mq-consumer-enabled: ${NOTIFICATION_MQ_CONSUMER_ENABLED:false}
   clients:
     auth:

iam/src/test/java/com/baiyue/iam/notification/NotificationConfigurationContractTest.java

@@ -0,0 +1,32 @@
+package com.baiyue.iam.notification;
+
+import org.junit.jupiter.api.Test;
+
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class NotificationConfigurationContractTest {
+    @Test void applicationAndNacosKeepConsumersDisabledAndExposeRecoveryTimeout() throws Exception {
+        assertSafeDefaults(read(Path.of("iam", "src", "main", "resources", "application.yml"),
+                Path.of("src", "main", "resources", "application.yml")));
+        assertSafeDefaults(read(Path.of("config", "nacos", "local", "iam-service.yml"),
+                Path.of("..", "config", "nacos", "local", "iam-service.yml")));
+        assertSafeDefaults(read(Path.of("config", "nacos", "cn", "iam-service.yml"),
+                Path.of("..", "config", "nacos", "cn", "iam-service.yml")));
+    }
+
+    private void assertSafeDefaults(String yaml) {
+        assertThat(yaml).contains(
+                "scheduler-enabled: ${NOTIFICATION_SCHEDULER_ENABLED:false}",
+                "dispatch-timeout-seconds: ${NOTIFICATION_DISPATCH_TIMEOUT_SECONDS:300}",
+                "mq-consumer-enabled: ${NOTIFICATION_MQ_CONSUMER_ENABLED:false}");
+    }
+
+    private String read(Path reactorPath, Path modulePath) throws Exception {
+        Path path = Files.exists(reactorPath) ? reactorPath : modulePath;
+        return Files.readString(path.normalize(), StandardCharsets.UTF_8);
+    }
+}

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

@@ -0,0 +1,67 @@
+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 recoverySqlRequiresTimeoutAndNoSuccessfulDeliveryEvidence() throws Exception {
+        Update retry = NotificationTaskMapper.class
+                .getMethod("retryTimedOutDispatchingWithoutDeliveryEvidence", java.time.LocalDateTime.class, java.time.LocalDateTime.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'");
+
+        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");
+    }
+
+    @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);
+    }
+}

iam/src/test/java/com/baiyue/iam/notification/NotificationServiceTest.java

@@ -150,6 +150,46 @@ class NotificationServiceTest {
         assertThat(error.getValue()).hasSize(1000);
     }
 
+    @Test void recoversOnlyTasksOlderThanConfiguredDispatchTimeout() {
+        when(repository.recoverTimedOutDispatching(any(), any())).thenReturn(2);
+
+        assertThat(service.recoverTimedOutDispatching(300)).isEqualTo(2);
+
+        ArgumentCaptor<LocalDateTime> staleBefore = ArgumentCaptor.forClass(LocalDateTime.class);
+        ArgumentCaptor<LocalDateTime> now = ArgumentCaptor.forClass(LocalDateTime.class);
+        verify(repository).recoverTimedOutDispatching(staleBefore.capture(), now.capture());
+        assertThat(staleBefore.getValue()).isEqualTo(now.getValue().minusSeconds(300));
+    }
+
+    @Test void onlyOneInstanceCanClaimRecoveredTask() {
+        when(repository.claimRetrying(41L)).thenReturn(true, false);
+        NotificationTask task = task(41L, "RETRYING");
+        when(repository.task(41L)).thenReturn(task);
+        when(repository.recipients(41L)).thenReturn(List.of());
+
+        assertThat(service.dispatchRetrying(41L)).isTrue();
+        assertThat(service.dispatchRetrying(41L)).isFalse();
+
+        verify(repository, times(2)).claimRetrying(41L);
+        verify(repository, times(1)).saveTask(task);
+    }
+
+    @Test void sentTaskCannotBeClaimedForRecoveryDelivery() {
+        when(repository.claimRetrying(42L)).thenReturn(false);
+
+        assertThat(service.dispatchRetrying(42L)).isFalse();
+
+        verify(repository, never()).task(42L);
+        verify(repository, never()).recipients(42L);
+    }
+
+    private NotificationTask task(long id, String status) {
+        NotificationTask value = new NotificationTask();
+        value.setId(id); value.setStatus(status); value.setCategory("ORDER"); value.setTitle("A");
+        value.setRecipientCount(0); value.setAttemptCount(1);
+        return value;
+    }
+
     private NotificationEventInbox event(String status, String hash) {
         NotificationEventInbox value = new NotificationEventInbox();
         value.setId(7L); value.setSourceEventId("event-1"); value.setEventType("order.status.changed");

iam/src/test/java/com/baiyue/iam/notification/NotificationSqlContractTest.java

@@ -1,7 +1,23 @@
 package com.baiyue.iam.notification;
 import org.junit.jupiter.api.Test;
 import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
 import static org.assertj.core.api.Assertions.assertThat;
 class NotificationSqlContractTest {
  @Test void definesTablesIsolationIdempotencyAndPermissions() throws Exception {String sql;try(var in=getClass().getResourceAsStream("/db/notification-center.sql")){assertThat(in).isNotNull();sql=new String(in.readAllBytes(),StandardCharsets.UTF_8);}assertThat(sql).contains("iam_notification_template","iam_notification_task","iam_notification_recipient","iam_notification_inbox","iam_notification_event_inbox","uk_notification_task_no","uk_notification_task_idempotency","uk_notification_source_event","uk_notification_recipient","idx_notification_inbox_list","platform:notification:template:status","platform:notification:task:retry","merchant:notification:view");}
+
+ @Test void unifiedMigrationContainsNotificationBaselineAndSevenIndexAssertion() throws Exception {
+  Path direct=Path.of("deploy","sql","头程数据库统一增量-MySQL8.sql");
+  Path migration=Files.exists(direct)?direct:Path.of("..","deploy","sql","头程数据库统一增量-MySQL8.sql").normalize();
+  String sql=Files.readString(migration,StandardCharsets.UTF_8);
+  assertThat(sql).contains("iam_notification_template","iam_notification_task","iam_notification_recipient",
+          "iam_notification_inbox","iam_notification_event_inbox","uk_notification_template_code",
+          "uk_notification_task_no","uk_notification_task_idempotency","uk_notification_source_event",
+          "uk_notification_recipient","uk_notification_inbox_recipient","uk_notification_event",
+          "assert_notification_center_schema","permission_code='merchant:notification:view'",
+          "permission_code LIKE 'platform:notification%'");
+  assertThat(sql).contains("COUNT(DISTINCT index_name) FROM information_schema.statistics WHERE table_schema='baiyue_iam'");
+  assertThat(sql).contains(")) <> 7 THEN");
+ }
 }