优化

iam/src/main/java/com/baiyue/iam/notification/dto/NotificationDtos.java

@@ -1,5 +1,6 @@
 package com.baiyue.iam.notification.dto;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import jakarta.validation.constraints.Future;
 import jakarta.validation.constraints.NotBlank;
 import jakarta.validation.constraints.NotNull;
@@ -24,7 +25,8 @@ public final class NotificationDtos {
             @NotNull @Size(max = 32) Map<String, String> variables,
             @Future LocalDateTime scheduledAt,
             @NotBlank @Size(min = 8, max = 128) String idempotencyKey,
-            @Size(max = 128) String sourceEventId) { }
+            @Size(max = 128) String sourceEventId,
+            @JsonIgnore @Size(max = 128) String sourceEventType) { }
 
     public record NotificationSummary(Long id, String category, String title, String status,
                                       LocalDateTime sentAt, LocalDateTime readAt) { }

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

@@ -1,5 +1,16 @@
 package com.baiyue.iam.notification.mapper;
 import com.baiyue.iam.notification.entity.NotificationEventInbox;
 import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import org.apache.ibatis.annotations.Insert;
 import org.apache.ibatis.annotations.Mapper;
-@Mapper public interface NotificationEventInboxMapper extends BaseMapper<NotificationEventInbox> { }
+@Mapper public interface NotificationEventInboxMapper extends BaseMapper<NotificationEventInbox> {
+    @Insert("""
+            INSERT IGNORE INTO iam_notification_event_inbox
+              (id, source_event_id, event_type, payload_hash, task_id, status, last_error,
+               processed_at, created_at, updated_at)
+            VALUES
+              (#{id}, #{sourceEventId}, #{eventType}, #{payloadHash}, #{taskId}, #{status}, #{lastError},
+               #{processedAt}, #{createdAt}, #{updatedAt})
+            """)
+    int insertIfAbsent(NotificationEventInbox value);
+}

iam/src/main/java/com/baiyue/iam/notification/mq/NotificationBusinessEventConsumer.java

@@ -3,6 +3,8 @@ package com.baiyue.iam.notification.mq;
 import com.baiyue.iam.notification.dto.NotificationDtos.CreateNotificationRequest;
 import com.baiyue.iam.notification.dto.NotificationDtos.RecipientType;
 import com.baiyue.iam.notification.service.NotificationService;
+import com.baiyue.iam.notification.service.NotificationService.EventClaim;
+import com.baiyue.iam.notification.service.NotificationService.EventClaimStatus;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -12,7 +14,11 @@ import org.apache.rocketmq.spring.core.RocketMQListener;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.stereotype.Component;
 import java.util.List;
+import java.util.HexFormat;
 import java.util.Map;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import com.fasterxml.jackson.databind.SerializationFeature;
 
 @Component
 @ConditionalOnProperty(name="baiyue.notification.mq-consumer-enabled",havingValue="true")
@@ -24,16 +30,35 @@ public class NotificationBusinessEventConsumer implements RocketMQListener<Strin
     @Resource private NotificationService service;
 
     @Override public void onMessage(String message){
+        EventClaim claim = null;
         try{
             NotificationMqContract.BusinessEvent event=objectMapper.readValue(message,NotificationMqContract.BusinessEvent.class);
             if(event.sourceEventId()==null||event.sourceEventId().isBlank()||event.eventType()==null||event.eventType().isBlank()||event.tenantId()==null)throw new IllegalArgumentException("Missing event identity or customer scope");
+            if(event.sourceEventId().length()>128||event.eventType().length()>128)throw new IllegalArgumentException("Event identity exceeds 128 characters");
             Map<String,Object> payload=event.payload();
+            if(payload==null)throw new IllegalArgumentException("Missing payload");
+            claim=service.beginBusinessEvent(event.sourceEventId(),event.eventType(),payloadHash(payload));
+            if(claim.status()==EventClaimStatus.PROCESSED)return;
+            if(claim.status()==EventClaimStatus.PROCESSING)throw new IllegalStateException("NOTIFICATION_EVENT_PROCESSING");
             String template=required(payload,"templateCode");RecipientType type=RecipientType.valueOf(required(payload,"recipientType"));
             List<Long> ids=objectMapper.convertValue(payload.getOrDefault("recipientIds",List.of()),new TypeReference<>(){});
             Map<String,String> variables=objectMapper.convertValue(payload.getOrDefault("variables",Map.of()),new TypeReference<>(){});
-            service.create(new CreateNotificationRequest(event.tenantId(),template,type,ids,(String)payload.get("roleCode"),variables,null,
-                    event.sourceEventId()+":"+event.eventType(),event.sourceEventId()),0L);
-        }catch(JsonProcessingException e){throw new IllegalArgumentException("Invalid notification business event",e);}
+            var task=service.create(new CreateNotificationRequest(event.tenantId(),template,type,ids,(String)payload.get("roleCode"),variables,null,
+                    event.sourceEventId()+":"+event.eventType(),event.sourceEventId(),event.eventType()),0L);
+            service.completeBusinessEvent(claim,task.id());
+        }catch(JsonProcessingException e){
+            throw new IllegalArgumentException("Invalid notification business event",e);
+        }catch(RuntimeException e){
+            if(claim!=null&&claim.status()==EventClaimStatus.ACQUIRED){
+                try{service.failBusinessEvent(claim,e);}catch(RuntimeException auditFailure){e.addSuppressed(auditFailure);}
+            }
+            throw e;
+        }
     }
     private String required(Map<String,Object> payload,String key){Object value=payload.get(key);if(value==null||value.toString().isBlank())throw new IllegalArgumentException("Missing "+key);return value.toString();}
+    private String payloadHash(Map<String,Object> payload) throws JsonProcessingException {
+        byte[] canonical=objectMapper.writer().with(SerializationFeature.ORDER_MAP_ENTRIES_BY_KEYS).writeValueAsBytes(payload);
+        try{return HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(canonical));}
+        catch(NoSuchAlgorithmException e){throw new IllegalStateException("SHA-256 unavailable",e);}
+    }
 }

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

@@ -17,6 +17,7 @@ public class NotificationRepository {
     @Resource private NotificationTaskMapper tasks;
     @Resource private NotificationRecipientMapper recipients;
     @Resource private NotificationInboxMapper inbox;
+    @Resource private NotificationEventInboxMapper eventInbox;
 
     public NotificationTemplate templateByCode(long tenantId, String code) {
         return templates.selectOne(new LambdaQueryWrapper<NotificationTemplate>().eq(NotificationTemplate::getTenantId, tenantId)
@@ -67,4 +68,32 @@ public class NotificationRepository {
     public void saveInbox(NotificationInbox value) { inbox.updateById(value); }
     public List<NotificationInbox> unreadItems(long tenantId,long membershipId,String app) { return inbox.selectList(new LambdaQueryWrapper<NotificationInbox>()
             .eq(NotificationInbox::getTenantId,tenantId).eq(NotificationInbox::getMembershipId,membershipId).eq(NotificationInbox::getAppCode,app).eq(NotificationInbox::getStatus,"UNREAD")); }
+    public boolean insertEventIfAbsent(NotificationEventInbox value) { return eventInbox.insertIfAbsent(value) == 1; }
+    public NotificationEventInbox event(String sourceEventId,String eventType) { return eventInbox.selectOne(new LambdaQueryWrapper<NotificationEventInbox>()
+            .eq(NotificationEventInbox::getSourceEventId,sourceEventId).eq(NotificationEventInbox::getEventType,eventType).last("LIMIT 1")); }
+    public boolean claimEvent(NotificationEventInbox value,java.time.LocalDateTime staleBefore,java.time.LocalDateTime claimedAt) {
+        boolean claimed = eventInbox.update(null,new LambdaUpdateWrapper<NotificationEventInbox>()
+                .eq(NotificationEventInbox::getId,value.getId()).eq(NotificationEventInbox::getPayloadHash,value.getPayloadHash())
+                .and(q -> q.eq(NotificationEventInbox::getStatus,"FAILED").or(nested -> nested
+                        .eq(NotificationEventInbox::getStatus,"PROCESSING").le(NotificationEventInbox::getUpdatedAt,staleBefore)))
+                .set(NotificationEventInbox::getStatus,"PROCESSING").set(NotificationEventInbox::getLastError,null)
+                .set(NotificationEventInbox::getProcessedAt,null).set(NotificationEventInbox::getUpdatedAt,claimedAt)) == 1;
+        if (claimed) value.setUpdatedAt(claimedAt);
+        return claimed;
+    }
+    public boolean completeEvent(Long id,String payloadHash,java.time.LocalDateTime claimToken,Long taskId,java.time.LocalDateTime now) {
+        return eventInbox.update(null,new LambdaUpdateWrapper<NotificationEventInbox>()
+                .eq(NotificationEventInbox::getId,id).eq(NotificationEventInbox::getPayloadHash,payloadHash)
+                .eq(NotificationEventInbox::getStatus,"PROCESSING").eq(NotificationEventInbox::getUpdatedAt,claimToken)
+                .set(NotificationEventInbox::getTaskId,taskId).set(NotificationEventInbox::getStatus,"PROCESSED")
+                .set(NotificationEventInbox::getLastError,null).set(NotificationEventInbox::getProcessedAt,now)
+                .set(NotificationEventInbox::getUpdatedAt,now)) == 1;
+    }
+    public boolean failEvent(Long id,String payloadHash,java.time.LocalDateTime claimToken,String error,java.time.LocalDateTime now) {
+        return eventInbox.update(null,new LambdaUpdateWrapper<NotificationEventInbox>()
+                .eq(NotificationEventInbox::getId,id).eq(NotificationEventInbox::getPayloadHash,payloadHash)
+                .eq(NotificationEventInbox::getStatus,"PROCESSING").eq(NotificationEventInbox::getUpdatedAt,claimToken)
+                .set(NotificationEventInbox::getStatus,"FAILED").set(NotificationEventInbox::getLastError,error)
+                .set(NotificationEventInbox::getProcessedAt,now).set(NotificationEventInbox::getUpdatedAt,now)) == 1;
+    }
 }

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

@@ -16,6 +16,7 @@ import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
 import java.time.LocalDateTime;
+import java.time.temporal.ChronoUnit;
 import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Map;
@@ -23,6 +24,11 @@ import java.util.Set;
 
 @Service
 public class NotificationService {
+    private static final long EVENT_PROCESSING_TIMEOUT_MINUTES = 5;
+
+    public enum EventClaimStatus { ACQUIRED, PROCESSED, PROCESSING }
+    public record EventClaim(NotificationEventInbox event, EventClaimStatus status) { }
+
     @Resource private NotificationRepository repository;
     @Resource private NotificationTemplateRenderer renderer;
     @Resource private NotificationRecipientResolver resolver;
@@ -94,6 +100,7 @@ public class NotificationService {
         task.setCategory(template.getCategory()); task.setTitle(title); task.setBody(body);
         task.setRecipientType(request.recipientType().name()); task.setVariablesJson(json(request.variables()));
         task.setIdempotencyKey(request.idempotencyKey()); task.setSourceEventId(request.sourceEventId());
+        task.setSourceEventType(request.sourceEventType());
         task.setStatus(request.scheduledAt() == null ? "DISPATCHING" : "SCHEDULED");
         task.setRecipientCount(resolved.size()); task.setAttemptCount(0); task.setScheduledAt(request.scheduledAt());
         task.setCreatedBy(operator); task.setCreatedAt(now); task.setUpdatedAt(now);
@@ -103,6 +110,47 @@ public class NotificationService {
         return taskItem(task, false);
     }
 
+    @Transactional
+    public EventClaim beginBusinessEvent(String sourceEventId, String eventType, String payloadHash) {
+        LocalDateTime now = eventNow();
+        NotificationEventInbox created = new NotificationEventInbox();
+        created.setId(IdWorker.getId()); created.setSourceEventId(sourceEventId); created.setEventType(eventType);
+        created.setPayloadHash(payloadHash); created.setStatus("PROCESSING");
+        created.setCreatedAt(now); created.setUpdatedAt(now);
+        if (repository.insertEventIfAbsent(created)) return new EventClaim(created, EventClaimStatus.ACQUIRED);
+
+        NotificationEventInbox existing = repository.event(sourceEventId, eventType);
+        if (existing == null) throw new IllegalStateException("NOTIFICATION_EVENT_REGISTRATION_RACE");
+        if (!payloadHash.equals(existing.getPayloadHash()))
+            fail(ResultCode.CONFLICT, "NOTIFICATION_EVENT_PAYLOAD_CONFLICT");
+        if ("PROCESSED".equals(existing.getStatus()))
+            return new EventClaim(existing, EventClaimStatus.PROCESSED);
+        LocalDateTime staleBefore = now.minusMinutes(EVENT_PROCESSING_TIMEOUT_MINUTES);
+        if (repository.claimEvent(existing, staleBefore, now))
+            return new EventClaim(existing, EventClaimStatus.ACQUIRED);
+        return new EventClaim(existing, EventClaimStatus.PROCESSING);
+    }
+
+    @Transactional
+    public void completeBusinessEvent(EventClaim claim, Long taskId) {
+        LocalDateTime now = eventNow();
+        if (!repository.completeEvent(claim.event().getId(), claim.event().getPayloadHash(),
+                claim.event().getUpdatedAt(), taskId, now))
+            throw new IllegalStateException("NOTIFICATION_EVENT_CLAIM_LOST");
+    }
+
+    @Transactional
+    public void failBusinessEvent(EventClaim claim, Throwable failure) {
+        String error = failure.getMessage() == null ? failure.getClass().getSimpleName() : failure.getMessage();
+        if (error.length() > 1000) error = error.substring(0, 1000);
+        LocalDateTime now = eventNow();
+        if (!repository.failEvent(claim.event().getId(), claim.event().getPayloadHash(),
+                claim.event().getUpdatedAt(), error, now))
+            throw new IllegalStateException("NOTIFICATION_EVENT_CLAIM_LOST");
+    }
+
+    private LocalDateTime eventNow() { return LocalDateTime.now().truncatedTo(ChronoUnit.MILLIS); }
+
     private void snapshot(NotificationTask task, List<NotificationRecipientResolver.ResolvedRecipient> resolved,
                           LocalDateTime now) {
         for (var value : resolved) {

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

@@ -0,0 +1,109 @@
+package com.baiyue.iam.notification;
+
+import com.baiyue.iam.notification.dto.NotificationDtos.CreateNotificationRequest;
+import com.baiyue.iam.notification.dto.NotificationDtos.TaskListItem;
+import com.baiyue.iam.notification.entity.NotificationEventInbox;
+import com.baiyue.iam.notification.mq.NotificationBusinessEventConsumer;
+import com.baiyue.iam.notification.service.NotificationService;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.*;
+
+class NotificationBusinessEventConsumerTest {
+    private final NotificationService service = mock(NotificationService.class);
+    private final NotificationBusinessEventConsumer consumer = new NotificationBusinessEventConsumer();
+
+    @BeforeEach void setup() {
+        ReflectionTestUtils.setField(consumer, "service", service);
+        ReflectionTestUtils.setField(consumer, "objectMapper", new ObjectMapper());
+    }
+
+    @Test void createsTaskWithEventTypeAndCompletesAudit() {
+        NotificationEventInbox event = event();
+        var claim = new NotificationService.EventClaim(event, NotificationService.EventClaimStatus.ACQUIRED);
+        when(service.beginBusinessEvent(eq("source-1"), eq("order.status.changed"), anyString())).thenReturn(claim);
+        when(service.create(any(), eq(0L))).thenReturn(new TaskListItem(31L, "NT31", "ORDER", "A", "SENT", 1, null, null, false));
+
+        consumer.onMessage(message("READY"));
+
+        ArgumentCaptor<CreateNotificationRequest> request = ArgumentCaptor.forClass(CreateNotificationRequest.class);
+        verify(service).create(request.capture(), eq(0L));
+        assertThat(request.getValue().sourceEventId()).isEqualTo("source-1");
+        assertThat(request.getValue().sourceEventType()).isEqualTo("order.status.changed");
+        verify(service).completeBusinessEvent(claim, 31L);
+    }
+
+    @Test void processedDuplicateIsAcknowledgedWithoutCreatingTask() {
+        var claim = new NotificationService.EventClaim(event(), NotificationService.EventClaimStatus.PROCESSED);
+        when(service.beginBusinessEvent(eq("source-1"), eq("order.status.changed"), anyString())).thenReturn(claim);
+
+        consumer.onMessage(message("READY"));
+
+        verify(service, never()).create(any(), anyLong());
+        verify(service, never()).completeBusinessEvent(any(), anyLong());
+    }
+
+    @Test void processingDuplicateIsRetriedWithoutOverwritingOwnerAudit() {
+        var claim = new NotificationService.EventClaim(event(), NotificationService.EventClaimStatus.PROCESSING);
+        when(service.beginBusinessEvent(eq("source-1"), eq("order.status.changed"), anyString())).thenReturn(claim);
+
+        assertThatThrownBy(() -> consumer.onMessage(message("READY")))
+                .isInstanceOf(IllegalStateException.class).hasMessageContaining("NOTIFICATION_EVENT_PROCESSING");
+        verify(service, never()).failBusinessEvent(any(), any());
+    }
+
+    @Test void taskFailureIsRecordedAndRethrownForRocketMqRetry() {
+        var claim = new NotificationService.EventClaim(event(), NotificationService.EventClaimStatus.ACQUIRED);
+        when(service.beginBusinessEvent(eq("source-1"), eq("order.status.changed"), anyString())).thenReturn(claim);
+        when(service.create(any(), eq(0L))).thenThrow(new IllegalStateException("delivery failed"));
+
+        assertThatThrownBy(() -> consumer.onMessage(message("READY")))
+                .isInstanceOf(IllegalStateException.class).hasMessageContaining("delivery failed");
+        verify(service).failBusinessEvent(eq(claim), argThat(error -> "delivery failed".equals(error.getMessage())));
+    }
+
+    @Test void payloadHashIsStableAcrossMapKeyOrder() {
+        var processed = new NotificationService.EventClaim(event(), NotificationService.EventClaimStatus.PROCESSED);
+        when(service.beginBusinessEvent(eq("source-1"), eq("order.status.changed"), anyString())).thenReturn(processed);
+        consumer.onMessage(message("READY"));
+        String first = hashArgument();
+        reset(service);
+        when(service.beginBusinessEvent(eq("source-1"), eq("order.status.changed"), anyString())).thenReturn(processed);
+        consumer.onMessage("{\"sourceEventId\":\"source-1\",\"tenantId\":9,\"eventType\":\"order.status.changed\",\"payload\":{\"variables\":{\"status\":\"READY\"},\"recipientIds\":[11],\"recipientType\":\"MEMBERSHIP\",\"templateCode\":\"ORDER_STATUS_CHANGED\"}}");
+        String second = hashArgument();
+        assertThat(second).isEqualTo(first);
+    }
+
+    @Test void rejectsOversizedEventIdentityBeforeDatabaseRegistration() {
+        String message = "{\"sourceEventId\":\"" + "x".repeat(129)
+                + "\",\"eventType\":\"order.status.changed\",\"tenantId\":9,\"payload\":{}}";
+        assertThatThrownBy(() -> consumer.onMessage(message))
+                .isInstanceOf(IllegalArgumentException.class).hasMessageContaining("exceeds 128");
+        verifyNoInteractions(service);
+    }
+
+    private String hashArgument() {
+        ArgumentCaptor<String> hash = ArgumentCaptor.forClass(String.class);
+        verify(service).beginBusinessEvent(eq("source-1"), eq("order.status.changed"), hash.capture());
+        return hash.getValue();
+    }
+
+    private NotificationEventInbox event() {
+        NotificationEventInbox value = new NotificationEventInbox();
+        value.setId(7L); value.setPayloadHash("hash"); value.setStatus("PROCESSING");
+        return value;
+    }
+
+    private String message(String status) {
+        return "{\"sourceEventId\":\"source-1\",\"eventType\":\"order.status.changed\",\"tenantId\":9,\"payload\":{\"templateCode\":\"ORDER_STATUS_CHANGED\",\"recipientType\":\"MEMBERSHIP\",\"recipientIds\":[11],\"variables\":{\"status\":\"" + status + "\"}}}";
+    }
+}

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

@@ -3,6 +3,7 @@ package com.baiyue.iam.notification;
 import com.baiyue.common.core.exception.BizException;
 import com.baiyue.iam.notification.dto.NotificationDtos.*;
 import com.baiyue.iam.notification.entity.NotificationRecipient;
+import com.baiyue.iam.notification.entity.NotificationEventInbox;
 import com.baiyue.iam.notification.entity.NotificationTask;
 import com.baiyue.iam.notification.entity.NotificationTemplate;
 import com.baiyue.iam.notification.repository.NotificationRepository;
@@ -18,6 +19,7 @@ import org.springframework.test.util.ReflectionTestUtils;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.time.LocalDateTime;
 
 import static org.assertj.core.api.Assertions.*;
 import static org.mockito.Mockito.*;
@@ -44,7 +46,8 @@ class NotificationServiceTest {
                 new NotificationRecipientResolver.ResolvedRecipient(9, 11, 12, "User", "MERCHANT")));
 
         var request = new CreateNotificationRequest(9L, "ORDER_STATUS_CHANGED", RecipientType.MEMBERSHIP,
-                List.of(11L), null, Map.of("order_no", "A", "status", "READY"), null, "idem-key-1", null);
+                List.of(11L), null, Map.of("order_no", "A", "status", "READY"), null, "idem-key-1",
+                "source-1", "order.status.changed");
         TaskListItem result = service.create(request, 99);
 
         assertThat(result.status()).isEqualTo("SENT");
@@ -54,6 +57,8 @@ class NotificationServiceTest {
         ArgumentCaptor<NotificationTask> task = ArgumentCaptor.forClass(NotificationTask.class);
         verify(repository).saveTask(task.capture());
         assertThat(task.getValue().getTitle()).isEqualTo("A");
+        assertThat(task.getValue().getSourceEventId()).isEqualTo("source-1");
+        assertThat(task.getValue().getSourceEventType()).isEqualTo("order.status.changed");
         assertThat(stored).hasSize(1);
         assertThat(stored.get(0).getStatus()).isEqualTo("DELIVERED");
     }
@@ -63,7 +68,7 @@ class NotificationServiceTest {
         existing.setCategory("ORDER"); existing.setTitle("A"); existing.setStatus("SENT"); existing.setRecipientCount(1);
         when(repository.taskByIdempotency("idem-key-1")).thenReturn(existing);
         var request = new CreateNotificationRequest(9L, "ORDER_STATUS_CHANGED", RecipientType.MEMBERSHIP,
-                List.of(11L), null, Map.of(), null, "idem-key-1", null);
+                List.of(11L), null, Map.of(), null, "idem-key-1", null, null);
         assertThat(service.create(request, 99).duplicated()).isTrue();
         verify(repository, never()).insertTask(any());
     }
@@ -77,6 +82,81 @@ class NotificationServiceTest {
                 .isInstanceOf(BizException.class).hasMessageContaining("NOTIFICATION_INVALID_STATE");
     }
 
+    @Test void recordsAndCompletesBusinessEventClaim() {
+        when(repository.insertEventIfAbsent(any())).thenReturn(true);
+        var claim = service.beginBusinessEvent("event-1", "order.status.changed", "hash-1");
+
+        assertThat(claim.status()).isEqualTo(NotificationService.EventClaimStatus.ACQUIRED);
+        assertThat(claim.event().getStatus()).isEqualTo("PROCESSING");
+        when(repository.completeEvent(eq(claim.event().getId()), eq("hash-1"),
+                eq(claim.event().getUpdatedAt()), eq(31L), any())).thenReturn(true);
+        service.completeBusinessEvent(claim, 31L);
+
+        verify(repository).completeEvent(eq(claim.event().getId()), eq("hash-1"),
+                eq(claim.event().getUpdatedAt()), eq(31L), any());
+    }
+
+    @Test void reusesProcessedEventWithSamePayload() {
+        NotificationEventInbox existing = event("PROCESSED", "hash-1");
+        when(repository.event("event-1", "order.status.changed")).thenReturn(existing);
+
+        var claim = service.beginBusinessEvent("event-1", "order.status.changed", "hash-1");
+
+        assertThat(claim.status()).isEqualTo(NotificationService.EventClaimStatus.PROCESSED);
+        verify(repository, never()).claimEvent(any(), any(), any());
+    }
+
+    @Test void rejectsSameEventIdentityWithDifferentPayload() {
+        when(repository.event("event-1", "order.status.changed")).thenReturn(event("PROCESSED", "hash-1"));
+
+        assertThatThrownBy(() -> service.beginBusinessEvent("event-1", "order.status.changed", "hash-2"))
+                .isInstanceOf(BizException.class).hasMessageContaining("NOTIFICATION_EVENT_PAYLOAD_CONFLICT");
+    }
+
+    @Test void freshConcurrentClaimRemainsProcessing() {
+        NotificationEventInbox existing = event("PROCESSING", "hash-1");
+        when(repository.event("event-1", "order.status.changed")).thenReturn(existing);
+        when(repository.claimEvent(eq(existing), any(), any())).thenReturn(false);
+
+        var claim = service.beginBusinessEvent("event-1", "order.status.changed", "hash-1");
+
+        assertThat(claim.status()).isEqualTo(NotificationService.EventClaimStatus.PROCESSING);
+    }
+
+    @Test void failedEventCanBeClaimedForRocketMqRetry() {
+        NotificationEventInbox existing = event("FAILED", "hash-1");
+        when(repository.event("event-1", "order.status.changed")).thenReturn(existing);
+        when(repository.claimEvent(eq(existing), any(), any())).thenAnswer(call -> {
+            existing.setUpdatedAt(call.getArgument(2));
+            return true;
+        });
+
+        var claim = service.beginBusinessEvent("event-1", "order.status.changed", "hash-1");
+
+        assertThat(claim.status()).isEqualTo(NotificationService.EventClaimStatus.ACQUIRED);
+    }
+
+    @Test void recordsTruncatedFailureForClaim() {
+        NotificationEventInbox existing = event("PROCESSING", "hash-1");
+        var claim = new NotificationService.EventClaim(existing, NotificationService.EventClaimStatus.ACQUIRED);
+        when(repository.failEvent(eq(existing.getId()), eq("hash-1"), eq(existing.getUpdatedAt()), any(), any()))
+                .thenReturn(true);
+
+        service.failBusinessEvent(claim, new IllegalStateException("x".repeat(1200)));
+
+        ArgumentCaptor<String> error = ArgumentCaptor.forClass(String.class);
+        verify(repository).failEvent(eq(existing.getId()), eq("hash-1"), eq(existing.getUpdatedAt()),
+                error.capture(), any());
+        assertThat(error.getValue()).hasSize(1000);
+    }
+
+    private NotificationEventInbox event(String status, String hash) {
+        NotificationEventInbox value = new NotificationEventInbox();
+        value.setId(7L); value.setSourceEventId("event-1"); value.setEventType("order.status.changed");
+        value.setPayloadHash(hash); value.setStatus(status); value.setUpdatedAt(LocalDateTime.now().minusMinutes(1));
+        return value;
+    }
+
     private NotificationTemplate activeTemplate() {
         NotificationTemplate value = new NotificationTemplate(); value.setId(1L);
         value.setTemplateCode("ORDER_STATUS_CHANGED"); value.setCategory("ORDER");

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

@@ -3,5 +3,5 @@ import org.junit.jupiter.api.Test;
 import java.nio.charset.StandardCharsets;
 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_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 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");}
 }