diff --git a/AGENTS.md b/AGENTS.md index 0e0fb4794..a49b49989 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -19,6 +19,7 @@ |---|---|---| | 登录、Token、密码、注册 | `auth/` | `auth/AGENTS.md` | | 客户档案、成员、审核、RBAC、授权快照 | `iam/` | `iam/AGENTS.md` | +| 消息中心、站内通知、消息模板、收件箱 | `iam/` | `iam/AGENTS.md`，再按 `src/main/java/com/baiyue/iam/notification/` 定位；业务事件生产者回到所属服务 | | 订单、货物、快照、状态、幂等、Outbox | `order/` | `order/AGENTS.md` | | 履约分配、入库、测量、审核、FBA 计费 | `firstleg/` | `firstleg/AGENTS.md` | | 分柜、柜次、智能排柜、预装/装柜清单、PDA | `firstleg/` | `firstleg/AGENTS.md`，再按 `split/`、`loading/`、`outbound/` 定位 | diff --git a/deploy/sql/头程数据库统一增量-MySQL8.sql b/deploy/sql/头程数据库统一增量-MySQL8.sql index 23b34f840..6bdb44cc2 100644 --- a/deploy/sql/头程数据库统一增量-MySQL8.sql +++ b/deploy/sql/头程数据库统一增量-MySQL8.sql @@ -2369,10 +2369,6 @@ CREATE TABLE IF NOT EXISTS fl_inbound_order ( expected_arrival_at DATETIME(3) NULL COMMENT '预计到达时间', customer_id BIGINT UNSIGNED NULL COMMENT '客户逻辑ID', customer_name VARCHAR(128) NOT NULL COMMENT '客户名称快照', - channel_code VARCHAR(64) NOT NULL COMMENT '订舱渠道编码快照', - channel_name VARCHAR(128) NOT NULL COMMENT '订舱渠道名称快照', - warehouse_id BIGINT UNSIGNED NOT NULL COMMENT '国内中转仓ID快照', - warehouse_name VARCHAR(128) NOT NULL COMMENT '国内中转仓名称快照', company_id BIGINT UNSIGNED NULL COMMENT '运营公司逻辑ID', company_name VARCHAR(128) NULL COMMENT '运营公司名称快照', warehouse_id BIGINT UNSIGNED NOT NULL COMMENT '实际作业仓库ID', @@ -4761,6 +4757,7 @@ BEGIN DECLARE target_exists INT DEFAULT 0; DECLARE source_total INT DEFAULT 0; DECLARE source_smart INT DEFAULT 0; + DECLARE source_stable_smart INT DEFAULT 0; DECLARE source_inbound INT DEFAULT 0; DECLARE target_total INT DEFAULT 0; DECLARE target_inbound INT DEFAULT 0; @@ -4797,7 +4794,22 @@ BEGIN OR (column_name = 'updated_by' AND LOWER(column_type) = 'bigint unsigned' AND is_nullable = 'NO') OR (column_name = 'updated_by_name' AND LOWER(column_type) = 'varchar(64)' AND is_nullable = 'YES') OR (column_name = 'created_at' AND data_type = 'datetime' AND datetime_precision = 3 AND is_nullable = 'NO' AND LOWER(column_default) LIKE 'current_timestamp%') - OR (column_name = 'updated_at' AND data_type = 'datetime' AND datetime_precision = 3 AND is_nullable = 'NO' AND LOWER(column_default) LIKE 'current_timestamp%' AND LOWER(extra) LIKE '%on update current_timestamp%')); + OR (column_name = 'updated_at' AND data_type = 'datetime' AND datetime_precision = 3 AND is_nullable = 'NO' AND LOWER(column_default) LIKE 'current_timestamp%' AND LOWER(extra) LIKE '%on update current_timestamp%')); + SELECT COUNT(*) INTO source_stable_smart FROM information_schema.columns + WHERE table_schema = 'baiyue_firstleg' AND table_name = 'fl_loading_pick_task' AND ( + (column_name = 'id' AND LOWER(column_type) = 'bigint unsigned' AND is_nullable = 'NO') + OR (column_name = 'loading_plan_id' AND LOWER(column_type) = 'bigint unsigned' AND is_nullable = 'NO') + OR (column_name = 'container_booking_id' AND LOWER(column_type) = 'bigint unsigned' AND is_nullable = 'NO') + OR (column_name = 'cabinet_no' AND LOWER(column_type) = 'varchar(64)' AND is_nullable = 'NO') + OR (column_name = 'container_no' AND LOWER(column_type) = 'varchar(64)' AND is_nullable = 'YES') + OR (column_name = 'task_status' AND LOWER(column_type) = 'varchar(16)' AND is_nullable = 'NO' AND column_default = 'LOADING') + OR (column_name = 'started_by' AND LOWER(column_type) = 'bigint unsigned' AND is_nullable = 'NO') + OR (column_name = 'started_by_name' AND LOWER(column_type) = 'varchar(64)' AND is_nullable = 'YES') + OR (column_name = 'started_at' AND data_type = 'datetime' AND datetime_precision = 3 AND is_nullable = 'NO' AND LOWER(column_default) LIKE 'current_timestamp%') + OR (column_name = 'completed_by' AND LOWER(column_type) = 'bigint unsigned' AND is_nullable = 'YES') + OR (column_name = 'completed_by_name' AND LOWER(column_type) = 'varchar(64)' AND is_nullable = 'YES') + OR (column_name = 'completed_at' AND data_type = 'datetime' AND datetime_precision = 3 AND is_nullable = 'YES') + OR (column_name = 'version' AND LOWER(column_type) = 'int unsigned' AND is_nullable = 'NO' AND CAST(column_default AS CHAR) = '0')); IF NOT ( (source_total = 11 AND source_smart = 11 AND (SELECT COUNT(*) FROM information_schema.statistics WHERE table_schema = 'baiyue_firstleg' AND table_name = 'fl_loading_pick_task' AND index_name = 'uk_loading_pick_task_plan' AND non_unique = 0 AND seq_in_index = 1 AND column_name = 'loading_plan_id') = 1 @@ -4808,6 +4820,12 @@ BEGIN AND (SELECT COUNT(*) FROM information_schema.statistics WHERE table_schema = 'baiyue_firstleg' AND table_name = 'fl_loading_pick_task' AND index_name = 'uk_loading_pick_tenant_inbound' AND non_unique = 0 AND ((seq_in_index = 1 AND column_name = 'tenant_id') OR (seq_in_index = 2 AND column_name = 'inbound_id'))) = 2 AND (SELECT COUNT(*) FROM information_schema.statistics WHERE table_schema = 'baiyue_firstleg' AND table_name = 'fl_loading_pick_task' AND index_name = 'idx_loading_pick_status' AND ((seq_in_index = 1 AND column_name = 'tenant_id') OR (seq_in_index = 2 AND column_name = 'status') OR (seq_in_index = 3 AND column_name = 'updated_at'))) = 3 AND (SELECT COUNT(*) FROM information_schema.key_column_usage WHERE constraint_schema = 'baiyue_firstleg' AND table_name = 'fl_loading_pick_task' AND column_name = 'inbound_id' AND referenced_table_name = 'fl_inbound_order' AND referenced_column_name = 'id') = 1) + OR + (source_total = 13 AND source_stable_smart = 13 + AND (SELECT COUNT(*) FROM information_schema.statistics WHERE table_schema = 'baiyue_firstleg' AND table_name = 'fl_loading_pick_task' AND index_name = 'uk_loading_pick_task_plan' AND non_unique = 0 AND seq_in_index = 1 AND column_name = 'loading_plan_id') = 1 + AND (SELECT COUNT(*) FROM information_schema.statistics WHERE table_schema = 'baiyue_firstleg' AND table_name = 'fl_loading_pick_task' AND index_name = 'uk_loading_pick_task_booking' AND non_unique = 0 AND seq_in_index = 1 AND column_name = 'container_booking_id') = 1 + AND (SELECT COUNT(*) FROM information_schema.statistics WHERE table_schema = 'baiyue_firstleg' AND table_name = 'fl_loading_pick_task' AND index_name = 'idx_loading_pick_container' AND ((seq_in_index = 1 AND column_name = 'container_no') OR (seq_in_index = 2 AND column_name = 'task_status') OR (seq_in_index = 3 AND column_name = 'started_at'))) = 3 + AND (SELECT COUNT(*) FROM information_schema.key_column_usage WHERE constraint_schema = 'baiyue_firstleg' AND table_name = 'fl_loading_pick_task' AND column_name = 'loading_plan_id' AND referenced_table_name = 'fl_loading_plan' AND referenced_column_name = 'id') = 1) ) THEN SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = '拣货任务迁移失败：fl_loading_pick_task 不是完整 canonical 表型'; END IF; @@ -4836,7 +4854,7 @@ BEGIN SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = '拣货任务迁移失败：fl_inbound_loading_pick_task 不是完整 canonical 表型'; END IF; END IF; - IF require_both = 1 AND (source_exists <> 1 OR target_exists <> 1 OR source_smart <> 11) THEN + IF require_both = 1 AND (source_exists <> 1 OR target_exists <> 1 OR source_stable_smart <> 13) THEN SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = '拣货任务终验失败：两张 canonical 表未同时就绪'; END IF; END$$ @@ -14258,9 +14276,9 @@ BEGIN SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='统一迁移失败：IAM 消息中心表不完整'; END IF; IF (SELECT COUNT(*) FROM information_schema.statistics WHERE table_schema='baiyue_iam' AND index_name IN ( - 'uk_notification_template_code','uk_notification_task_idempotency','uk_notification_source_event', + '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' - )) <> 6 THEN + )) <> 7 THEN SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='统一迁移失败：IAM 消息中心唯一索引不完整'; END IF; IF (SELECT COUNT(*) FROM baiyue_iam.iam_permission WHERE tenant_id=0 AND status='ACTIVE' AND deleted=0 diff --git a/firstleg/src/test/java/com/baiyue/firstleg/repository/Phase3ChannelContainerMigrationContractTest.java b/firstleg/src/test/java/com/baiyue/firstleg/repository/Phase3ChannelContainerMigrationContractTest.java index ab6db414c..9811a39df 100644 --- a/firstleg/src/test/java/com/baiyue/firstleg/repository/Phase3ChannelContainerMigrationContractTest.java +++ b/firstleg/src/test/java/com/baiyue/firstleg/repository/Phase3ChannelContainerMigrationContractTest.java @@ -18,7 +18,26 @@ class Phase3ChannelContainerMigrationContractTest { @Test void unifiedMigrationRestoresGuardsBeforeSurfacingFailures() throws Exception { - assertSafeWindow(unifiedMigration(), true); + String sql = unifiedMigration(); + assertSafeWindow(sql, true); + assertNoDuplicateInboundColumns(sql); + } + + private static void assertNoDuplicateInboundColumns(String sql) { + int tableStart = sql.indexOf("CREATE TABLE IF NOT EXISTS fl_inbound_order ("); + int tableEnd = sql.indexOf(") ENGINE=InnoDB", tableStart); + String tableDefinition = sql.substring(tableStart, tableEnd); + + assertThat(tableStart).isGreaterThan(0); + assertThat(tableEnd).isGreaterThan(tableStart); + assertThat(countOccurrences(tableDefinition, " warehouse_id BIGINT UNSIGNED")).isEqualTo(1); + assertThat(countOccurrences(tableDefinition, " warehouse_name VARCHAR(128)")).isEqualTo(1); + assertThat(countOccurrences(tableDefinition, " channel_code VARCHAR(64)")).isEqualTo(1); + assertThat(countOccurrences(tableDefinition, " channel_name VARCHAR(128)")).isEqualTo(1); + } + + private static int countOccurrences(String text, String token) { + return (text.length() - text.replace(token, "").length()) / token.length(); } private static void assertSafeWindow(String sql, boolean qualified) { diff --git a/iam/AGENTS.md b/iam/AGENTS.md index b768c95df..d701bab88 100644 --- a/iam/AGENTS.md +++ b/iam/AGENTS.md @@ -15,9 +15,10 @@ | 平台用户 | `controller/PlatformUserManagementController.java`、`service/PlatformUserManagementService.java` | | 角色/权限/RBAC | `controller/RbacController.java`、`service/RbacService.java` | | 授权快照/内部资格 | `controller/AuthorizationController.java`、`InternalAuthorizationController.java`、`service/AuthorizationSnapshotService.java` | +| 消息中心/站内通知 | `notification/controller/NotificationInboxController.java`、`notification/controller/PlatformNotificationController.java`、`notification/service/NotificationService.java`；模板、任务、收件箱及事件消费均在 `notification/` | | File 资质元数据协作 | `controller/InternalTenantController.java`、`service/ProfileResourceService.java`、`client/AuthAccountClient.java` | | DTO/持久化 | `dto/`、`entity/`、`mapper/`、`repository/` | -| SQL/OpenAPI | `src/main/resources/db/`、`src/main/resources/openapi/iam-v1.yaml` | +| SQL/OpenAPI | IAM 主契约 `src/main/resources/openapi/iam-v1.yaml`；消息契约 `src/main/resources/openapi/notification-v1.yaml`；消息脚本 `src/main/resources/db/notification-center.sql`，其余数据库脚本在同目录 | | 测试/验证 | `src/test/java/com/baiyue/iam/`；`mvn -pl iam -am test` | ## 修改约束 diff --git a/iam/src/main/java/com/baiyue/iam/notification/dto/NotificationDtos.java b/iam/src/main/java/com/baiyue/iam/notification/dto/NotificationDtos.java index fec6cf3a0..c25ad297c 100644 --- a/iam/src/main/java/com/baiyue/iam/notification/dto/NotificationDtos.java +++ b/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 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) { } diff --git a/iam/src/main/java/com/baiyue/iam/notification/mapper/NotificationEventInboxMapper.java b/iam/src/main/java/com/baiyue/iam/notification/mapper/NotificationEventInboxMapper.java index d604cf896..2e8fcffc2 100644 --- a/iam/src/main/java/com/baiyue/iam/notification/mapper/NotificationEventInboxMapper.java +++ b/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 { } +@Mapper public interface NotificationEventInboxMapper extends BaseMapper { + @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); +} diff --git a/iam/src/main/java/com/baiyue/iam/notification/mq/NotificationBusinessEventConsumer.java b/iam/src/main/java/com/baiyue/iam/notification/mq/NotificationBusinessEventConsumer.java index 0f78b981d..7ce804b91 100644 --- a/iam/src/main/java/com/baiyue/iam/notification/mq/NotificationBusinessEventConsumer.java +++ b/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 RocketMQListener128||event.eventType().length()>128)throw new IllegalArgumentException("Event identity exceeds 128 characters"); Map 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 ids=objectMapper.convertValue(payload.getOrDefault("recipientIds",List.of()),new TypeReference<>(){}); Map 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 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 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);} + } } diff --git a/iam/src/main/java/com/baiyue/iam/notification/repository/NotificationRepository.java b/iam/src/main/java/com/baiyue/iam/notification/repository/NotificationRepository.java index 37c5a3293..1bc804f8f 100644 --- a/iam/src/main/java/com/baiyue/iam/notification/repository/NotificationRepository.java +++ b/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().eq(NotificationTemplate::getTenantId, tenantId) @@ -67,4 +68,32 @@ public class NotificationRepository { public void saveInbox(NotificationInbox value) { inbox.updateById(value); } public List unreadItems(long tenantId,long membershipId,String app) { return inbox.selectList(new LambdaQueryWrapper() .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() + .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() + .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() + .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() + .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; + } } diff --git a/iam/src/main/java/com/baiyue/iam/notification/service/NotificationService.java b/iam/src/main/java/com/baiyue/iam/notification/service/NotificationService.java index 5ea740fd6..dc296e533 100644 --- a/iam/src/main/java/com/baiyue/iam/notification/service/NotificationService.java +++ b/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 resolved, LocalDateTime now) { for (var value : resolved) { diff --git a/iam/src/test/java/com/baiyue/iam/notification/NotificationBusinessEventConsumerTest.java b/iam/src/test/java/com/baiyue/iam/notification/NotificationBusinessEventConsumerTest.java new file mode 100644 index 000000000..807ced3bb --- /dev/null +++ b/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 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 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 + "\"}}}"; + } +} diff --git a/iam/src/test/java/com/baiyue/iam/notification/NotificationServiceTest.java b/iam/src/test/java/com/baiyue/iam/notification/NotificationServiceTest.java index ad6266234..6c26bd740 100644 --- a/iam/src/test/java/com/baiyue/iam/notification/NotificationServiceTest.java +++ b/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 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 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"); diff --git a/iam/src/test/java/com/baiyue/iam/notification/NotificationSqlContractTest.java b/iam/src/test/java/com/baiyue/iam/notification/NotificationSqlContractTest.java index c7f6add54..d22185beb 100644 --- a/iam/src/test/java/com/baiyue/iam/notification/NotificationSqlContractTest.java +++ b/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");} }