From 47f99befb78197bc57fbac0d32bf03185fd452a6 Mon Sep 17 00:00:00 2001 From: ouyangxq <775153912@qq.com> Date: Wed, 12 Aug 2026 19:42:08 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- AGENTS.md | 10 ++- auth/AGENTS.md | 6 +- config/nacos/cn/iam-service.yml | 5 ++ config/nacos/local/iam-service.yml | 5 ++ deploy/sql/头程数据库统一增量-MySQL8.sql | 11 +-- file/AGENTS.md | 6 +- firstleg/AGENTS.md | 8 +-- .../db/phase3-channel-container-migration.sql | 15 +++-- iam/AGENTS.md | 6 +- .../mapper/NotificationTaskMapper.java | 46 ++++++++++++- .../repository/NotificationRepository.java | 9 +++ .../service/NotificationScheduler.java | 21 +++++- .../service/NotificationService.java | 37 ++++++++-- iam/src/main/resources/application.yml | 1 + ...NotificationConfigurationContractTest.java | 32 +++++++++ .../NotificationSchedulerRecoveryTest.java | 67 +++++++++++++++++++ .../notification/NotificationServiceTest.java | 40 +++++++++++ .../NotificationSqlContractTest.java | 16 +++++ integration-service/AGENTS.md | 2 +- order/AGENTS.md | 6 +- warehouse/AGENTS.md | 6 +- 21 files changed, 307 insertions(+), 48 deletions(-) create mode 100644 iam/src/test/java/com/baiyue/iam/notification/NotificationConfigurationContractTest.java create mode 100644 iam/src/test/java/com/baiyue/iam/notification/NotificationSchedulerRecoveryTest.java diff --git a/AGENTS.md b/AGENTS.md index a49b49989..fa284389b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -31,12 +31,10 @@ | 百度/Amazon/OCR/第三方适配 | `integration-service/` | `integration-service/AGENTS.md` | | PDF 标签、审计、事件 | `print/`、`audit-service/`、`event-bridge-service/` | 各服务指引 | -## 阶段路由 +## 全量验收路由 -- 一期：`auth`、`iam`、`warehouse`、`order`、`firstleg`、`file` 完成登录、客户/RBAC、公共仓、订单闭环；先读 `../docs/一期/README.md` 及闭环设计。 -- 二期：`firstleg` 主导入库/测量/审核/计费，`warehouse` 管托盘库位，`file` 管测量图片，`integration-service` 管 OCR，`order` 仅提供货物内部查询；终点为 `WAIT_SPLIT + APPROVED + CONFIRMED`，不实现分柜、装柜、PDA 和完整财务。 -- 三期：已确认范围可按 `../docs/三期/三期开工门禁.md` 进入契约和模块实现。`firstleg` 负责履约编排、柜次、排柜、订舱、拖车、PDA 与 `LOADED` 门禁；`order`、`warehouse`、`file`、`iam`、`gateway` 和 `integration-service` 仅在各自契约需要时协作。未冻结的字段、权限、迁移或跨服务调用只阻断对应模块。 -- 财务：作为跨阶段独立业务域，范围先读 `../docs/财务管理/README.md` 中已确认的产品结论与契约；`finance` 保存财务状态，`firstleg` 只提供不可变费用来源和柜次成本快照，`file` 保存收付款/成本凭证。不要把财务范围反向并入二期计费或三期履约。 +- 当前一期、二期、三期及财务统一进入全量验收；按本次链路读取对应阶段/专题文档，不以历史阶段终点阻断跨服务联调或回归。 +- `auth`、`iam`、`warehouse`、`order`、`firstleg`、`file`、`finance`、`integration-service` 及出仓相关能力按各自模块契约协作；状态、权限、数据库和 OpenAPI 以当前实现及已确认契约为准。 ## 契约、数据库与验证入口 @@ -52,4 +50,4 @@ - 三期：`../docs/三期/README.md`、`../docs/三期/三期开工门禁.md`、`../docs/三期/三期开发规划.md`、`../docs/三期/三期需求确认表.md`、页面与原型清单 - 财务：`../docs/财务管理/README.md` 中已确认的范围、对象、状态和接口约束 -进度、排期、负责人或验收状态问题再读取阶段进度表和验收报告；普通开发定位只使用范围、契约、模块指引和当前代码。 +全量验收任务再读取对应阶段进度表、验收报告和证据矩阵；普通代码定位只读取当前链路涉及的范围、契约、模块指引和代码。 diff --git a/auth/AGENTS.md b/auth/AGENTS.md index abc8d9d4e..6da6f0c76 100644 --- a/auth/AGENTS.md +++ b/auth/AGENTS.md @@ -25,8 +25,6 @@ - 注册跨 Auth/IAM 必须保持幂等和失败补偿；内部接口仅接受服务令牌。 - 业务错误保持 HTTP 200 + body code；Security 认证失败保留真实 401/403。 -## 分期导航 +## 全量验收导航 -- 一期：登录、注册、账号状态和 Token；先读 `../../docs/一期/README.md`。 -- 二期：无独立登录流程，仅保障入库/计费沿用 JWT、应用域和客户隔离；先读 `../../docs/二期/README.md`。 -- 三期：没有独立账号或角色模型；认证与会话仅在 IAM 三期权限契约需要时保持兼容。先读 `../../docs/三期/三期开工门禁.md`，不得借三期新增租户层级或绕过服务端授权。 +一期、二期、三期及财务全量验收均复用本模块认证能力；按本次链路读取对应阶段/专题文档，重点核对 Token、应用域、客户隔离、失效和跨 Auth/IAM 补偿，不因历史阶段划分排除其他消费者。 diff --git a/config/nacos/cn/iam-service.yml b/config/nacos/cn/iam-service.yml index 43082d56b..32dcc05c7 100644 --- a/config/nacos/cn/iam-service.yml +++ b/config/nacos/cn/iam-service.yml @@ -4,3 +4,8 @@ spring: username: ${IAM_DB_USERNAME} password: ${IAM_DB_PASSWORD} driver-class-name: com.mysql.cj.jdbc.Driver +baiyue: + notification: + scheduler-enabled: ${NOTIFICATION_SCHEDULER_ENABLED:false} + dispatch-timeout-seconds: ${NOTIFICATION_DISPATCH_TIMEOUT_SECONDS:300} + mq-consumer-enabled: ${NOTIFICATION_MQ_CONSUMER_ENABLED:false} diff --git a/config/nacos/local/iam-service.yml b/config/nacos/local/iam-service.yml index fba851416..4639f7ca1 100644 --- a/config/nacos/local/iam-service.yml +++ b/config/nacos/local/iam-service.yml @@ -3,3 +3,8 @@ spring.datasource: url: ${IAM_DB_URL:jdbc:mysql://127.0.0.1:3306/baiyue_iam?useUnicode=true&characterEncoding=utf8&serverTimezone=Asia/Shanghai&rewriteBatchedStatements=true} username: ${IAM_DB_USERNAME:root} password: ${IAM_DB_PASSWORD:123456} +baiyue: + notification: + scheduler-enabled: ${NOTIFICATION_SCHEDULER_ENABLED:false} + dispatch-timeout-seconds: ${NOTIFICATION_DISPATCH_TIMEOUT_SECONDS:300} + mq-consumer-enabled: ${NOTIFICATION_MQ_CONSUMER_ENABLED:false} diff --git a/deploy/sql/头程数据库统一增量-MySQL8.sql b/deploy/sql/头程数据库统一增量-MySQL8.sql index 6bdb44cc2..53e7ca853 100644 --- a/deploy/sql/头程数据库统一增量-MySQL8.sql +++ b/deploy/sql/头程数据库统一增量-MySQL8.sql @@ -14157,9 +14157,12 @@ DELIMITER ; CALL assert_phase3_channel_container_migration(); DROP PROCEDURE IF EXISTS assert_phase3_channel_container_migration; CREATE TABLE IF NOT EXISTS baiyue_firstleg.fl_container_transport_history ( - id BIGINT UNSIGNED NOT NULL,container_booking_id BIGINT UNSIGNED NOT NULL,before_container_no VARCHAR(64) NULL,after_container_no VARCHAR(64) NULL, - before_voyage_no VARCHAR(64) NULL,after_voyage_no VARCHAR(64) NULL,reason VARCHAR(500) NULL,operator_id BIGINT UNSIGNED NOT NULL, - operator_name VARCHAR(64) NULL,created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),PRIMARY KEY(id),KEY idx_container_transport_history(container_booking_id,created_at,id) + id BIGINT UNSIGNED NOT NULL COMMENT '运输字段变更历史ID',container_booking_id BIGINT UNSIGNED NOT NULL COMMENT '稳定柜次ID', + before_container_no VARCHAR(64) NULL COMMENT '变更前真实集装箱号',after_container_no VARCHAR(64) NULL COMMENT '变更后真实集装箱号', + before_voyage_no VARCHAR(64) NULL COMMENT '变更前航次号',after_voyage_no VARCHAR(64) NULL COMMENT '变更后航次号', + reason VARCHAR(500) NULL COMMENT '运输字段变更原因',operator_id BIGINT UNSIGNED NOT NULL COMMENT '变更操作人ID', + operator_name VARCHAR(64) NULL COMMENT '变更操作人名称快照',created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '变更记录创建时间', + PRIMARY KEY(id),KEY idx_container_transport_history(container_booking_id,created_at,id) ) ENGINE=InnoDB COMMENT='柜次真实运输字段不可变变更审计'; /* IAM 消息中心基线。来源：iam/src/main/resources/db/notification-center.sql */ @@ -14275,7 +14278,7 @@ BEGIN )) <> 5 THEN 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 ( + IF (SELECT COUNT(DISTINCT index_name) FROM information_schema.statistics WHERE table_schema='baiyue_iam' AND index_name IN ( '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' )) <> 7 THEN diff --git a/file/AGENTS.md b/file/AGENTS.md index e1e5c358f..f392a6bf6 100644 --- a/file/AGENTS.md +++ b/file/AGENTS.md @@ -23,8 +23,6 @@ - 所有文件按客户、入库单、文件类型校验归属；重量图片可 OCR，第三方尺寸证据不得触发重量 OCR。 - 生产环境显式配置 `FILE_STORAGE_ROOT`；磁盘故障返回 `S0503`，不得包装成业务错误。 -## 分期导航 +## 全量验收导航 -- 一期：资质上传与读取；先读 `../../docs/一期/README.md`。 -- 二期：测量图片和 OCR 重试；先读 `../../docs/二期/README.md` 与第三方联调说明。 -- 三期：按稳定契约实现装柜五类证据和报关回执的存储、归属、鉴权与审计；先读 `../../docs/三期/三期开工门禁.md`。照片允许追加、错误记录仅作废不硬删除，客户不得读取照片或锁码。 +一期、二期、三期及财务全量验收均复用本模块文件能力；按本次链路读取对应阶段/专题文档，统一核对资质、测量图片、装柜证据、报关回执的归属、鉴权、审计和生命周期。照片允许追加、错误记录仅作废不硬删除，客户不得读取内部照片或锁码。 diff --git a/firstleg/AGENTS.md b/firstleg/AGENTS.md index 32eb94b00..96d671e1d 100644 --- a/firstleg/AGENTS.md +++ b/firstleg/AGENTS.md @@ -4,7 +4,7 @@ - 负责履约任务/分配，以及二期入库单、测量/OCR 结果、审核、FBA 计费、完成入库编排。 - 订单主数据归 `order/`，托盘/库位归 `warehouse/`，文件对象归 `file/`，OCR 供应商归 `integration-service/`；通过客户端 DTO 和内部令牌协作。 -- 入库终点为 `WAIT_SPLIT + APPROVED + CONFIRMED`；二期不实现分柜、装柜、PDA 或完整财务。 +- 入库、分柜、装柜、PDA 和出仓能力均在本模块按对应服务端契约协作；`WAIT_SPLIT + APPROVED + CONFIRMED` 是入库链路状态节点，不是全量验收的范围终点。财务状态仍归 `finance`，本模块只生产不可变费用来源和柜次成本快照。 ## 模块入口 @@ -50,8 +50,6 @@ - 部分拣货资源结算必须调用 Warehouse 的 `/internal/warehouse/v1/pallet-locations/settlements` 原子接口；禁止拆成独立释放和转移调用。全部释放但子单仍有数量时，由仓库服务重新选择同仓普通库位，不足才落 A0。 - 数据库字段变更同时维护 `src/main/resources/db/phase3-pda-release-allocation-migration.sql` 和 `../deploy/sql/头程数据库统一增量-MySQL8.sql`。 -## 分期导航 +## 全量验收导航 -- 一期：履约任务与分配；先读 `../../docs/一期/README.md`。 -- 二期：入库、测量、审核、计费闭环；先读 `../../docs/二期/README.md`、主流程和联调报告。 -- 三期：负责订单履约编排、柜次、智能排柜、订舱、拖车、PDA 作业与 `LOADED` 门禁；先读 `../../docs/三期/三期开工门禁.md`。已确认范围按 Controller、DTO、OpenAPI、迁移和调用方共同实现；PDA 按 uni-app 在线更新安排开发与验收，不另设设备型号门禁。 +本模块承载一期履约、二期入库测量计费、三期订单编排/柜次/排柜/订舱/拖车/PDA/`LOADED` 门禁等全量链路；按本次验收范围读取对应阶段文档、Controller、DTO、OpenAPI、迁移和调用方，不以历史阶段终点阻断联调。PDA 按 uni-app 在线更新安排开发与验收，不另设设备型号门禁。 diff --git a/firstleg/src/main/resources/db/phase3-channel-container-migration.sql b/firstleg/src/main/resources/db/phase3-channel-container-migration.sql index a142f7761..b89ba7da3 100644 --- a/firstleg/src/main/resources/db/phase3-channel-container-migration.sql +++ b/firstleg/src/main/resources/db/phase3-channel-container-migration.sql @@ -163,10 +163,15 @@ CALL assert_phase3_channel_container_migration(); DROP PROCEDURE IF EXISTS assert_phase3_channel_container_migration; CREATE TABLE IF NOT EXISTS fl_container_transport_history ( - id BIGINT UNSIGNED NOT NULL, container_booking_id BIGINT UNSIGNED NOT NULL, - before_container_no VARCHAR(64) NULL, after_container_no VARCHAR(64) NULL, - before_voyage_no VARCHAR(64) NULL, after_voyage_no VARCHAR(64) NULL, - reason VARCHAR(500) NULL, operator_id BIGINT UNSIGNED NOT NULL, operator_name VARCHAR(64) NULL, - created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3), PRIMARY KEY(id), + id BIGINT UNSIGNED NOT NULL COMMENT '运输字段变更历史ID', + container_booking_id BIGINT UNSIGNED NOT NULL COMMENT '稳定柜次ID', + before_container_no VARCHAR(64) NULL COMMENT '变更前真实集装箱号', + after_container_no VARCHAR(64) NULL COMMENT '变更后真实集装箱号', + before_voyage_no VARCHAR(64) NULL COMMENT '变更前航次号', + after_voyage_no VARCHAR(64) NULL COMMENT '变更后航次号', + reason VARCHAR(500) NULL COMMENT '运输字段变更原因', + operator_id BIGINT UNSIGNED NOT NULL COMMENT '变更操作人ID', + operator_name VARCHAR(64) NULL COMMENT '变更操作人名称快照', + created_at DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '变更记录创建时间', PRIMARY KEY(id), KEY idx_container_transport_history (container_booking_id,created_at,id) ) ENGINE=InnoDB COMMENT='柜次真实运输字段不可变变更审计'; diff --git a/iam/AGENTS.md b/iam/AGENTS.md index d701bab88..22d1fb7e3 100644 --- a/iam/AGENTS.md +++ b/iam/AGENTS.md @@ -27,8 +27,6 @@ - 成员/账号跨 IAM/Auth 保持幂等补偿；资质跨 IAM/File 不保存文件内容。 - 权限资源 code/type 不可变；按钮隐藏不能替代后端权限校验。 -## 分期导航 +## 全量验收导航 -- 一期：客户建档审核、平台用户与 RBAC、企业资料；先读 `../../docs/一期/README.md`。 -- 二期：仅提供客户资格/授权快照，不承载入库或计费状态；先读 `../../docs/二期/README.md`。 -- 三期：按最小权限为操作员、拣货员、理货员落地已确认的角色与权限；先读 `../../docs/三期/三期开工门禁.md`。具体权限码与授权快照须先随服务端契约冻结，不自行扩展租户层级或客户可见范围。 +本模块承载客户建档审核、平台用户/RBAC、企业资料、消息中心及三期作业角色权限；按本次验收链路读取对应阶段/专题文档和权限契约，核对授权快照、客户隔离、权限码和消息收件箱，不因历史阶段划分排除协作场景。 diff --git a/iam/src/main/java/com/baiyue/iam/notification/mapper/NotificationTaskMapper.java b/iam/src/main/java/com/baiyue/iam/notification/mapper/NotificationTaskMapper.java index 570a06f19..0d58a2fef 100644 --- a/iam/src/main/java/com/baiyue/iam/notification/mapper/NotificationTaskMapper.java +++ b/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 { } +import org.apache.ibatis.annotations.Update; +import org.apache.ibatis.annotations.Param; +import java.time.LocalDateTime; +@Mapper public interface NotificationTaskMapper extends BaseMapper { + @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); +} 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 1bc804f8f..5f5e76809 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 @@ -54,6 +54,15 @@ public class NotificationRepository { public boolean claimScheduled(Long taskId) { return tasks.update(null,new LambdaUpdateWrapper() .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 retryingTasks() { return tasks.selectList(new LambdaQueryWrapper() + .eq(NotificationTask::getStatus,"RETRYING").orderByAsc(NotificationTask::getUpdatedAt).last("LIMIT 100")); } + public boolean claimRetrying(Long taskId) { return tasks.update(null,new LambdaUpdateWrapper() + .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 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() diff --git a/iam/src/main/java/com/baiyue/iam/notification/service/NotificationScheduler.java b/iam/src/main/java/com/baiyue/iam/notification/service/NotificationScheduler.java index 768283ccf..813cd81cd 100644 --- a/iam/src/main/java/com/baiyue/iam/notification/service/NotificationScheduler.java +++ b/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 taskIds, Function dispatcher) { + for (Long taskId : taskIds) { + try { dispatcher.apply(taskId); } + catch (RuntimeException failure) { + log.error("Notification {} task dispatch failed; taskId={}", source, taskId, failure); + } + } + } } 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 dc296e533..9b1e8bc36 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 @@ -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 dueScheduledTaskIds() { + return repository.dueTasks().stream().map(NotificationTask::getId).toList(); + } + + @Transactional(readOnly = true) + public List 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) diff --git a/iam/src/main/resources/application.yml b/iam/src/main/resources/application.yml index e79f73311..e724fd6a4 100644 --- a/iam/src/main/resources/application.yml +++ b/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: diff --git a/iam/src/test/java/com/baiyue/iam/notification/NotificationConfigurationContractTest.java b/iam/src/test/java/com/baiyue/iam/notification/NotificationConfigurationContractTest.java new file mode 100644 index 000000000..a21f4e66b --- /dev/null +++ b/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); + } +} diff --git a/iam/src/test/java/com/baiyue/iam/notification/NotificationSchedulerRecoveryTest.java b/iam/src/test/java/com/baiyue/iam/notification/NotificationSchedulerRecoveryTest.java new file mode 100644 index 000000000..d199ba723 --- /dev/null +++ b/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); + } +} 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 6c26bd740..ac6b51b98 100644 --- a/iam/src/test/java/com/baiyue/iam/notification/NotificationServiceTest.java +++ b/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 staleBefore = ArgumentCaptor.forClass(LocalDateTime.class); + ArgumentCaptor 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"); 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 d22185beb..d1edd6e14 100644 --- a/iam/src/test/java/com/baiyue/iam/notification/NotificationSqlContractTest.java +++ b/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"); + } } diff --git a/integration-service/AGENTS.md b/integration-service/AGENTS.md index fb89e12d8..b7e48949b 100644 --- a/integration-service/AGENTS.md +++ b/integration-service/AGENTS.md @@ -18,4 +18,4 @@ - 密钥、原始响应和客户敏感数据不得写入日志；真实调用默认关闭，适配器负责超时、重试、限流、错误归一化。 - OCR 只能访问受控输入/输出目录，接口仅接受内部服务令牌；契约变更同步检查 File 的 `WeightOcrClient` 和测试。 -- 一期仅保留通用适配基础；二期服务 File 重量 OCR，凭证缺失时走人工修正降级；三期仅在船期来源、频率、限流、超时、最近有效结果和过期标识的契约稳定后接入外部查询，不建设报关自动对接。阶段文档：`../../docs/二期/README.md`、`../../docs/三期/三期开工门禁.md`、`../../docs/三期/三期需求确认表.md`。 +- 全量验收按业务链路核对通用适配、File 重量 OCR、船期外部查询及已确认的报关协作契约；外部调用必须保留来源、频率、限流、超时、最近有效结果和过期标识，凭证缺失按既定人工修正降级，不把未确认的自动化能力当作已实现契约。相关依据：`../../docs/一期/`、`../../docs/二期/`、`../../docs/三期/`。 diff --git a/order/AGENTS.md b/order/AGENTS.md index b4fdb321e..0e414ffb1 100644 --- a/order/AGENTS.md +++ b/order/AGENTS.md @@ -24,8 +24,6 @@ - 所有客户查询在 Repository/Mapper 显式带 `tenant_id`；状态变更保留幂等、历史、审计和并发控制。 - Warehouse 往来单位、Firstleg 入库协作只通过客户端 DTO/内部令牌；修改契约需同步消费者。 -## 分期导航 +## 全量验收导航 -- 一期：订单下单与交易状态闭环；先读 `../../docs/一期/README.md` 与闭环设计。 -- 二期：仅提供货物内部查询，不承载入库状态；先读 `../../docs/二期/README.md`。 -- 三期：在 `firstleg` 编排契约稳定后，负责父子订单、数量守恒、订单主状态、幂等和时间线所需的主数据扩展；先读 `../../docs/三期/三期开工门禁.md`。不得自行创建独立出仓单或绕过 `LOADED` 服务端门禁。 +本模块承载订单下单、交易状态、货物查询、父子订单、数量守恒、主状态、幂等和时间线等全量订单契约；按本次验收链路读取对应阶段文档和 `firstleg` 编排契约，不创建独立出仓单，也不绕过 `LOADED` 服务端门禁。 diff --git a/warehouse/AGENTS.md b/warehouse/AGENTS.md index fa7f66896..326684f92 100644 --- a/warehouse/AGENTS.md +++ b/warehouse/AGENTS.md @@ -33,8 +33,6 @@ - A0 换绑入口包括内部接口和平台接口：平台人员使用 `/api/platform/v1/locations/temporary-occupancies/{occupancyId}/rebind`，需要 `platform:location:edit`。目标必须为同仓 `AVAILABLE` 的普通库位；A0 不允许人工编辑、停用或删除。 - 资源结算、释放、转移和换绑必须保存可重放的幂等记录或具有等价的状态幂等语义，并在测试中覆盖并发状态变更。 -## 分期导航 +## 全量验收导航 -- 一期：公共仓与主数据；先读 `../../docs/一期/README.md`。 -- 二期：托盘/库位预占占用和海外仓费用；先读 `../../docs/二期/README.md` 与主流程。 -- 三期：仅为 PDA 拣货与资源结算提供实时库存、当前位置、可用量和原子结算能力；先读 `../../docs/三期/三期开工门禁.md`。不扩展完整 WMS，且设备真机事项不改变服务端资源契约。 +本模块承载公共仓与主数据、托盘/库位预占占用、海外仓费用、PDA 拣货和资源结算协作；按本次验收链路读取对应阶段文档，核对实时库存、当前位置、可用量、原子结算和客户隔离。不扩展未确认的完整 WMS，设备真机事项不改变服务端资源契约。