diff --git a/docs/铁路无人机智能巡检平台统一消息中心设计方案_V1.0.md b/docs/铁路无人机智能巡检平台统一消息中心设计方案_V1.0.md new file mode 100644 index 0000000..2e9b6b9 --- /dev/null +++ b/docs/铁路无人机智能巡检平台统一消息中心设计方案_V1.0.md @@ -0,0 +1,142 @@ +# 铁路无人机智能巡检平台统一消息中心设计方案 + +> 文档版本:V1.0 +> 编制日期:2026-08-10 +> 适用基线:AItrackwalker 当前平台主后端及现有告警、通知实现 +> 文档状态:第一阶段基础链路已实施,腾讯云真实适配器待资质和配置 +> 部署决策:第一阶段采用模块化单体,保留未来拆分为独立服务的边界 + +## 1. 建设目标 + +统一消息中心负责接收业务消息意图,并完成接收人解析、策略选择、模板渲染、渠道投递、失败重试、供应商回执、站内信已读和发送审计。首期核心渠道是站内信、腾讯云短信、腾讯云邮件 SES 和腾讯云语音消息 VMS;为匹配需求清单,模型同时预留大屏弹窗、APP 推送和微信渠道。 + +告警识别、告警等级判定、业务告警聚合、确认、抑制和关闭继续由告警域负责。消息中心只处理通知侧幂等、限流、静默、摘要与升级,不成为第二套告警系统。 + +## 2. 第一阶段模块位置 + +第一阶段代码位于: + +```text +platform/backend/src/main/java/com/ai/trackwalker/notification +``` + +模块内部按照 API、应用、领域、渠道和基础设施分层。其他业务模块只能通过 `NotificationFacade` 或领域事件提交消息意图,不得直接调用腾讯云 SDK、消息内部 Repository 或消息数据表。 + +若未来满足多系统复用、独立扩缩容、凭据强隔离或独立团队维护等条件,可整体迁移到 `platform/notification-center`,业务侧契约保持不变。 + +## 3. 模块边界 + +消息中心负责: + +- 统一消息请求和业务幂等; +- 场景策略、渠道路由和渠道发送策略; +- 逻辑模板、模板版本及供应商模板映射; +- 用户、角色、组织接收人解析和发送快照; +- 站内信、短信、邮件、语音投递; +- 重试、过期、死信、回执和发送审计; +- 通知级去重、限流、静默时段、摘要和升级; +- 验证码专用流程使用的底层通道能力。 + +消息中心不负责: + +- AI 识别和告警是否成立; +- 告警严重程度的业务判定; +- 相同线路、里程和场景是否合并为同一业务告警; +- 告警确认、抑制、关闭和工单状态流转; +- 用户手机号和邮箱主数据维护。 + +当接收条件同时包含“组织 + 角色”时,两者按交集解释,即仅选择该组织树内拥有目标角色的用户;指定用户与上述结果按并集合并。这样可以避免把某个局段的告警发给其他组织的同角色人员。 + +## 4. 分层策略 + +消息编排分为三层策略: + +1. 场景策略:根据场景、等级、时间和接收人决定渠道、发送时间、升级和降噪规则。 +2. 渠道路由策略:为逻辑渠道选择供应商、账号和主备路由。 +3. 渠道发送策略:将已解析、已渲染的投递命令转换为供应商请求。 + +逻辑渠道为 `IN_APP`、`WEB_POPUP`、`APP_PUSH`、`WECHAT`、`SMS`、`EMAIL`、`VOICE`。供应商信息不进入业务消息契约。没有启用发送器的渠道会明确记录为 `SKIPPED_CHANNEL_DISABLED`,不会静默丢弃,也不会伪装成发送成功。 + +首期默认告警矩阵依据需求清单配置为: + +| 告警优先级 | 默认渠道 | +| --- | --- | +| P1 / 一级重大 | 站内信、大屏、APP、微信、短信、邮件 | +| P2 / 二级较大 | 站内信、APP、微信、短信 | +| P3 / 三级一般 | 站内信、APP、微信 | +| P4 / 四级提示 | 站内信、大屏 | + +语音不默认绑定告警等级。待业务确认升级规则后,可配置为“一级告警超过指定时间未确认再电话通知”,避免识别瞬间直接呼叫。 + +## 5. 可靠性原则 + +- 告警等业务域使用本地 Outbox 发布领域事件,消息模块通过 Inbox 幂等消费。 +- 同一租户下 `idempotencyKey` 唯一,消息和“接收人 × 渠道”投递分别去重。 +- 供应商受理和最终送达是两个状态,不将 API 成功等同于用户收到。 +- 网络超时且供应商结果不确定时进入 `UNKNOWN`,优先等待回执或对账,避免立即重发。 +- 配置错误、非法号码和未审核模板属于永久失败;限流、网络错误和供应商临时错误进入退避重试。 +- 业务严重告警的再次呼叫属于升级策略,不属于技术失败重试。 +- 调度实例异常退出时,超过领取超时仍处于 `SENDING` 的投递会自动回收到重试队列。 +- 消息过期后同步刷新消息汇总状态,避免请求长期停留在处理中。 + +## 6. 聚合与降噪 + +当前第一阶段实施业务幂等和消息过期,不实施告警内容聚合。短周期内容去重、每用户每渠道限频、静默时段和复杂摘要窗口在有真实发送量后进入第二阶段,避免在业务规则尚未稳定时把重要告警误合并。 + +告警业务聚合保留在告警域。消息中心未来只增加通知侧聚合,例如普通告警五分钟摘要、静默时段延迟以及未确认后的渠道升级。 + +## 7. 验证码 + +验证码可以复用腾讯云短信和语音客户端,但必须使用独立入口、模板、队列、限流和数据模型。普通 `NotificationFacade` 不接受验证码消息。 + +验证码必须使用密码学安全随机数,只保存哈希,并绑定用途、目标、挑战 ID 和有效期;验证成功后一次性消费,同时限制手机号、账号、IP、设备和租户维度的申请与错误次数。 + +## 8. 实施顺序 + +1. 已完成:冻结统一消息契约、模块依赖规则和渠道策略 SPI。 +2. 已完成:建立消息、接收人、投递、尝试、模板、策略、站内信和事件 Inbox 数据模型。 +3. 已完成:接入 `alarm.detected`,以告警表中的权威等级选择渠道;旧历史事件不回放。 +4. 已完成:站内信发送、本人收件箱查询、未读数和已读回写。 +5. 待实施:接入腾讯云短信及状态回执。 +6. 待实施:接入腾讯云邮件 SES 及投递事件。 +7. 待实施:确认电话升级规则后接入腾讯云语音 VMS、未接听和按键回执。 +8. 待实施:建立验证码独立流程。 +9. 待实施:逐步迁移现有确认告警和工单通知 Outbox,避免一次性切换风险。 + +## 9. 当前实现范围 + +当前代码实现包括: + +- `NotificationFacade` 统一业务入口和租户级幂等; +- 数据库策略选择渠道,调用方显式渠道可覆盖策略; +- 用户、组织树、角色及“组织 + 角色”范围化接收人解析; +- 版本化模板、必填变量校验和消息快照; +- “接收人 × 渠道”投递拆分、并发领取、退避重试、过期及卡单恢复; +- 策略模式渠道注册表和站内信真实发送器; +- 短信、邮件、语音 Mock 发送器,仅在 `RAIL_NOTIFICATION_MOCK_EXTERNAL_CHANNELS=true` 时启用; +- 告警事件 Inbox 投影和历史事件防回放; +- 本人站内信查询、未读计数和已读 API。 + +腾讯云真实发送器尚未启用,原因不是代码边界不完整,而是短信资质、签名与模板,邮件发信域名,语音模板及腾讯云密钥尚未齐备。默认配置关闭外部 Mock;未启用的外部渠道会留下可审计的跳过记录,不会产生真实费用。 + +## 10. 已提供接口与配置 + +站内信接口: + +- `GET /api/v1/message-center/inbox` +- `GET /api/v1/message-center/inbox/unread-count` +- `POST /api/v1/message-center/inbox/{notificationId}/read` + +主要开关: + +- `RAIL_NOTIFICATION_CENTER_ENABLED`:消息中心总开关,默认 `true`; +- `RAIL_NOTIFICATION_MOCK_EXTERNAL_CHANNELS`:外部渠道 Mock,默认 `false`; +- `RAIL_NOTIFICATION_BATCH_SIZE`:单批领取数量,默认 `50`; +- `RAIL_NOTIFICATION_MAX_ATTEMPTS`:最大技术重试次数,默认 `5`; +- `RAIL_NOTIFICATION_CLAIM_TIMEOUT_SECONDS`:发送任务卡单回收时间,默认 `300` 秒。 + +## 11. 与告警等级的职责关系 + +检测与规则链路写入告警事实及初始严重度,告警域保存经过规则映射后的权威等级;消息中心读取这个权威等级并套用渠道矩阵。消息中心不自行根据置信度、线路或场景重新判级,也不负责把多个检测结果合并成一个告警。 + +后续若新增人工改级、持续时长升级或 SLA 升级,应先由告警域更新权威告警状态并发布新事件,消息中心再以新的幂等键执行升级通知。 diff --git a/platform/backend/src/main/java/com/ai/trackwalker/notification/api/AlarmNotificationProjector.java b/platform/backend/src/main/java/com/ai/trackwalker/notification/api/AlarmNotificationProjector.java new file mode 100644 index 0000000..2a5d031 --- /dev/null +++ b/platform/backend/src/main/java/com/ai/trackwalker/notification/api/AlarmNotificationProjector.java @@ -0,0 +1,148 @@ +package com.ai.trackwalker.notification.api; + +import com.ai.trackwalker.common.Jsonb; +import com.ai.trackwalker.notification.application.NotificationFacade; +import com.ai.trackwalker.notification.application.SendNotificationCommand; +import com.ai.trackwalker.notification.domain.MessageKind; +import com.ai.trackwalker.notification.domain.MessagePriority; +import com.ai.trackwalker.notification.infrastructure.NotificationProperties; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; + +import java.time.Duration; +import java.time.Instant; +import java.util.List; +import java.util.Map; + +@Component +public class AlarmNotificationProjector { + private final JdbcTemplate jdbc; + private final NotificationFacade notifications; + private final NotificationProperties properties; + + public AlarmNotificationProjector( + JdbcTemplate jdbc, + NotificationFacade notifications, + NotificationProperties properties + ) { + this.jdbc = jdbc; + this.notifications = notifications; + this.properties = properties; + } + + @Scheduled(initialDelay = 7000, fixedDelayString = "${rail.notification.event-projector-delay-ms:3000}") + public void project() { + if (!properties.isEnabled()) { + return; + } + List> events = jdbc.queryForList(""" + select e.id,e.event_type,e.created_at + from platform_events e + left join notification_center.event_inbox i on i.source_event_id=e.id + where e.event_type='alarm.detected' + and (i.source_event_id is null or (i.status='FAILED' and i.attempt_count event : events) { + process(event); + } + } + + private void process(Map event) { + String eventId = String.valueOf(event.get("id")); + try { + jdbc.update(""" + insert into notification_center.event_inbox( + source_event_id,event_type,status,attempt_count,created_at,updated_at + ) values (?,?,'PROCESSING',1,now(),now()) + on conflict(source_event_id) do update set + status='PROCESSING',attempt_count=notification_center.event_inbox.attempt_count+1, + last_error=null,updated_at=now() + """, eventId, event.get("event_type")); + List> alarms = jdbc.queryForList(""" + select a.id,a.scene,a.category,a.severity,a.owner_org_id,a.location::text as location, + t.line_id + from alarms a join inspection_tasks t on t.id=a.task_id + where a.id=(select payload->>'alarm_id' from platform_events where id=?) + and a.suppressed=false + """, eventId); + if (alarms.isEmpty()) { + markProcessed(eventId); + return; + } + Map alarm = alarms.get(0); + MessagePriority priority = priority(String.valueOf(alarm.get("severity"))); + Map location = Jsonb.map(String.valueOf(alarm.get("location"))); + String ownerOrgId = alarm.get("owner_org_id") == null + ? "org-railway" : String.valueOf(alarm.get("owner_org_id")); + String alarmId = String.valueOf(alarm.get("id")); + notifications.submit(new SendNotificationCommand( + "alarm-detected:" + alarmId + ":initial", + ownerOrgId, + "ALARM_DETECTED", + MessageKind.ALERT, + priority, + new SendNotificationCommand.BusinessReference("ALARM", alarmId), + new SendNotificationCommand.Audience( + List.of(), + List.of(ownerOrgId), + List.of("TASK_DISPATCHER", "WORKORDER_REVIEWER") + ), + "alarm-detected", + null, + Map.of( + "level_label", levelLabel(priority), + "scene", String.valueOf(alarm.get("scene")), + "category", String.valueOf(alarm.get("category")), + "line_name", String.valueOf(alarm.get("line_id")), + "mileage", String.valueOf(location.getOrDefault("mileage", "线路邻近")) + ), + List.of(), + Instant.now().plus(expiry(priority)), + eventId + )); + markProcessed(eventId); + } catch (Exception exception) { + String message = exception.getMessage() == null ? exception.getClass().getSimpleName() : exception.getMessage(); + jdbc.update(""" + update notification_center.event_inbox + set status='FAILED',last_error=?,updated_at=now() where source_event_id=? + """, message.length() > 1000 ? message.substring(0, 1000) : message, eventId); + } + } + + private void markProcessed(String eventId) { + jdbc.update(""" + update notification_center.event_inbox + set status='PROCESSED',processed_at=now(),updated_at=now() where source_event_id=? + """, eventId); + } + + private MessagePriority priority(String severity) { + return switch (severity.toUpperCase()) { + case "LEVEL_1", "CRITICAL" -> MessagePriority.P1_CRITICAL; + case "LEVEL_2", "HIGH" -> MessagePriority.P2_HIGH; + case "LEVEL_3", "MEDIUM" -> MessagePriority.P3_NORMAL; + default -> MessagePriority.P4_LOW; + }; + } + + private String levelLabel(MessagePriority priority) { + return switch (priority) { + case P1_CRITICAL -> "一级重大"; + case P2_HIGH -> "二级较大"; + case P3_NORMAL -> "三级一般"; + case P4_LOW -> "四级提示"; + }; + } + + private Duration expiry(MessagePriority priority) { + return switch (priority) { + case P1_CRITICAL -> Duration.ofHours(1); + case P2_HIGH -> Duration.ofHours(4); + case P3_NORMAL -> Duration.ofHours(24); + case P4_LOW -> Duration.ofDays(3); + }; + } +} diff --git a/platform/backend/src/main/java/com/ai/trackwalker/notification/api/MessageCenterController.java b/platform/backend/src/main/java/com/ai/trackwalker/notification/api/MessageCenterController.java new file mode 100644 index 0000000..47a9ba5 --- /dev/null +++ b/platform/backend/src/main/java/com/ai/trackwalker/notification/api/MessageCenterController.java @@ -0,0 +1,40 @@ +package com.ai.trackwalker.notification.api; + +import com.ai.trackwalker.api.ApiResponse; +import com.ai.trackwalker.notification.application.InAppMessageService; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RequestParam; +import org.springframework.web.bind.annotation.RestController; + +import java.util.Map; + +@RestController +@RequestMapping("/api/v1/message-center") +public class MessageCenterController { + private final InAppMessageService inAppMessages; + + public MessageCenterController(InAppMessageService inAppMessages) { + this.inAppMessages = inAppMessages; + } + + @GetMapping("/inbox") + public ApiResponse inbox( + @RequestParam(required = false) String status, + @RequestParam(defaultValue = "100") int limit + ) { + return ApiResponse.ok(Map.of("messages", inAppMessages.list(status, limit))); + } + + @GetMapping("/inbox/unread-count") + public ApiResponse unreadCount() { + return ApiResponse.ok(Map.of("unread_count", inAppMessages.unreadCount())); + } + + @PostMapping("/inbox/{notificationId}/read") + public ApiResponse markRead(@PathVariable String notificationId) { + return ApiResponse.ok(inAppMessages.markRead(notificationId)); + } +} diff --git a/platform/backend/src/main/java/com/ai/trackwalker/notification/api/package-info.java b/platform/backend/src/main/java/com/ai/trackwalker/notification/api/package-info.java new file mode 100644 index 0000000..81dbab3 --- /dev/null +++ b/platform/backend/src/main/java/com/ai/trackwalker/notification/api/package-info.java @@ -0,0 +1,6 @@ +/** + * HTTP and event-consumer adapters for the notification module. + * + *

No public endpoint is exposed in the initial skeleton.

+ */ +package com.ai.trackwalker.notification.api; diff --git a/platform/backend/src/main/java/com/ai/trackwalker/notification/application/DefaultNotificationFacade.java b/platform/backend/src/main/java/com/ai/trackwalker/notification/application/DefaultNotificationFacade.java new file mode 100644 index 0000000..800d273 --- /dev/null +++ b/platform/backend/src/main/java/com/ai/trackwalker/notification/application/DefaultNotificationFacade.java @@ -0,0 +1,61 @@ +package com.ai.trackwalker.notification.application; + +import com.ai.trackwalker.notification.domain.MessageChannel; +import com.ai.trackwalker.notification.infrastructure.JdbcMessageStore; +import org.springframework.http.HttpStatus; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; +import org.springframework.web.server.ResponseStatusException; + +import java.time.Instant; +import java.util.List; + +@Service +public class DefaultNotificationFacade implements NotificationFacade { + private final NotificationPolicyResolver policyResolver; + private final RecipientResolver recipientResolver; + private final MessageTemplateRenderer templateRenderer; + private final JdbcMessageStore store; + + public DefaultNotificationFacade( + NotificationPolicyResolver policyResolver, + RecipientResolver recipientResolver, + MessageTemplateRenderer templateRenderer, + JdbcMessageStore store + ) { + this.policyResolver = policyResolver; + this.recipientResolver = recipientResolver; + this.templateRenderer = templateRenderer; + this.store = store; + } + + @Override + @Transactional + public NotificationSubmission submit(SendNotificationCommand command) { + validate(command); + List channels = policyResolver.resolve(command); + List recipients = recipientResolver.resolve(command.audience()); + if (recipients.isEmpty()) { + throw new ResponseStatusException(HttpStatus.UNPROCESSABLE_ENTITY, "消息没有可用接收人"); + } + MessageTemplateRenderer.RenderedMessage rendered = templateRenderer.render( + command.templateCode(), command.templateVersion(), command.variables()); + JdbcMessageStore.StoredMessage stored = store.create(command, recipients, channels, rendered); + return new NotificationSubmission(stored.id(), stored.status(), stored.createdAt()); + } + + private void validate(SendNotificationCommand command) { + if (command == null || blank(command.idempotencyKey()) || blank(command.tenantId()) + || blank(command.scenarioCode()) || command.kind() == null || command.priority() == null + || blank(command.templateCode())) { + throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "消息请求缺少必要字段"); + } + if (command.expiresAt() != null && !command.expiresAt().isAfter(Instant.now())) { + throw new ResponseStatusException(HttpStatus.BAD_REQUEST, "消息已过期"); + } + } + + private boolean blank(String value) { + return value == null || value.isBlank(); + } +} diff --git a/platform/backend/src/main/java/com/ai/trackwalker/notification/application/InAppMessageService.java b/platform/backend/src/main/java/com/ai/trackwalker/notification/application/InAppMessageService.java new file mode 100644 index 0000000..d525115 --- /dev/null +++ b/platform/backend/src/main/java/com/ai/trackwalker/notification/application/InAppMessageService.java @@ -0,0 +1,76 @@ +package com.ai.trackwalker.notification.application; + +import com.ai.trackwalker.security.SecurityActor; +import org.springframework.http.HttpStatus; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; +import org.springframework.web.server.ResponseStatusException; + +import java.util.List; +import java.util.Locale; +import java.util.Map; + +@Service +public class InAppMessageService { + private final JdbcTemplate jdbc; + private final SecurityActor actor; + + public InAppMessageService(JdbcTemplate jdbc, SecurityActor actor) { + this.jdbc = jdbc; + this.actor = actor; + } + + public List> list(String status, int requestedLimit) { + actor.require("notification:read"); + int limit = Math.max(1, Math.min(requestedLimit, 200)); + String normalized = status == null || status.isBlank() ? null : status.toUpperCase(Locale.ROOT); + if (normalized == null) { + return jdbc.queryForList(""" + select i.id as notification_id,i.message_id,i.title,i.body,i.priority,i.status, + i.business_type,i.business_id,i.read_at,i.clicked_at,i.archived_at,i.created_at + from notification_center.in_app_messages i + where i.recipient_user_id=? and i.archived_at is null + order by i.created_at desc limit ? + """, actor.subject(), limit); + } + return jdbc.queryForList(""" + select i.id as notification_id,i.message_id,i.title,i.body,i.priority,i.status, + i.business_type,i.business_id,i.read_at,i.clicked_at,i.archived_at,i.created_at + from notification_center.in_app_messages i + where i.recipient_user_id=? and i.status=? and i.archived_at is null + order by i.created_at desc limit ? + """, actor.subject(), normalized, limit); + } + + public int unreadCount() { + actor.require("notification:read"); + Integer count = jdbc.queryForObject(""" + select count(*) from notification_center.in_app_messages + where recipient_user_id=? and status='UNREAD' and archived_at is null + """, Integer.class, actor.subject()); + return count == null ? 0 : count; + } + + @Transactional + public Map markRead(String notificationId) { + actor.require("notification:read"); + int updated = jdbc.update(""" + update notification_center.in_app_messages + set status='READ',read_at=coalesce(read_at,now()),updated_at=now() + where id=? and recipient_user_id=? + """, notificationId, actor.subject()); + if (updated == 0) { + throw new ResponseStatusException(HttpStatus.NOT_FOUND, "站内信不存在"); + } + jdbc.update(""" + update notification_center.message_deliveries d + set status='READ',read_at=coalesce(read_at,now()),updated_at=now() + from notification_center.in_app_messages i + where i.id=? and d.id=i.delivery_id + """, notificationId); + return jdbc.queryForMap(""" + select id as notification_id,status,read_at from notification_center.in_app_messages where id=? + """, notificationId); + } +} diff --git a/platform/backend/src/main/java/com/ai/trackwalker/notification/application/MessageTemplateRenderer.java b/platform/backend/src/main/java/com/ai/trackwalker/notification/application/MessageTemplateRenderer.java new file mode 100644 index 0000000..5455073 --- /dev/null +++ b/platform/backend/src/main/java/com/ai/trackwalker/notification/application/MessageTemplateRenderer.java @@ -0,0 +1,70 @@ +package com.ai.trackwalker.notification.application; + +import com.ai.trackwalker.common.Jsonb; +import org.springframework.http.HttpStatus; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.stereotype.Component; +import org.springframework.web.server.ResponseStatusException; + +import java.util.List; +import java.util.Map; + +@Component +public class MessageTemplateRenderer { + private final JdbcTemplate jdbc; + + public MessageTemplateRenderer(JdbcTemplate jdbc) { + this.jdbc = jdbc; + } + + public RenderedMessage render(String code, Integer requestedVersion, Map variables) { + List> rows = requestedVersion == null + ? jdbc.queryForList(""" + select version_no,title_template,body_template,variable_schema::text as variable_schema + from notification_center.message_templates + where code=? and status='PUBLISHED' + order by version_no desc limit 1 + """, code) + : jdbc.queryForList(""" + select version_no,title_template,body_template,variable_schema::text as variable_schema + from notification_center.message_templates + where code=? and version_no=? and status='PUBLISHED' + """, code, requestedVersion); + if (rows.isEmpty()) { + throw new ResponseStatusException(HttpStatus.UNPROCESSABLE_ENTITY, "消息模板不存在或未发布:" + code); + } + Map row = rows.get(0); + validateVariables(code, Jsonb.map(String.valueOf(row.get("variable_schema"))), variables); + return new RenderedMessage( + ((Number) row.get("version_no")).intValue(), + replace(String.valueOf(row.get("title_template")), variables), + replace(String.valueOf(row.get("body_template")), variables) + ); + } + + private void validateVariables(String code, Map schema, Map variables) { + Object required = schema.get("required"); + if (!(required instanceof List names)) { + return; + } + List missing = names.stream() + .map(String::valueOf) + .filter(name -> !variables.containsKey(name) || variables.get(name) == null) + .toList(); + if (!missing.isEmpty()) { + throw new ResponseStatusException(HttpStatus.UNPROCESSABLE_ENTITY, + "消息模板变量缺失:" + code + " " + missing); + } + } + + private String replace(String template, Map variables) { + String result = template; + for (Map.Entry entry : variables.entrySet()) { + result = result.replace("{{" + entry.getKey() + "}}", String.valueOf(entry.getValue())); + } + return result.replaceAll("\\{\\{[^}]+}}", "-"); + } + + public record RenderedMessage(int version, String title, String body) { + } +} diff --git a/platform/backend/src/main/java/com/ai/trackwalker/notification/application/NotificationFacade.java b/platform/backend/src/main/java/com/ai/trackwalker/notification/application/NotificationFacade.java new file mode 100644 index 0000000..352f187 --- /dev/null +++ b/platform/backend/src/main/java/com/ai/trackwalker/notification/application/NotificationFacade.java @@ -0,0 +1,8 @@ +package com.ai.trackwalker.notification.application; + +/** + * The only business-facing entry point for submitting ordinary notifications and alerts. + */ +public interface NotificationFacade { + NotificationSubmission submit(SendNotificationCommand command); +} diff --git a/platform/backend/src/main/java/com/ai/trackwalker/notification/application/NotificationPolicyResolver.java b/platform/backend/src/main/java/com/ai/trackwalker/notification/application/NotificationPolicyResolver.java new file mode 100644 index 0000000..0669783 --- /dev/null +++ b/platform/backend/src/main/java/com/ai/trackwalker/notification/application/NotificationPolicyResolver.java @@ -0,0 +1,43 @@ +package com.ai.trackwalker.notification.application; + +import com.ai.trackwalker.common.Jsonb; +import com.ai.trackwalker.notification.domain.MessageChannel; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.stereotype.Component; + +import java.util.List; +import java.util.Map; + +@Component +public class NotificationPolicyResolver { + private final JdbcTemplate jdbc; + + public NotificationPolicyResolver(JdbcTemplate jdbc) { + this.jdbc = jdbc; + } + + public List resolve(SendNotificationCommand command) { + if (!command.requestedChannels().isEmpty()) { + return command.requestedChannels().stream().distinct().toList(); + } + List> rows = jdbc.queryForList(""" + select channel_matrix::text as channel_matrix + from notification_center.message_policies + where status='ACTIVE' and ? like scenario_pattern + order by version_no desc limit 1 + """, command.scenarioCode()); + if (rows.isEmpty()) { + return List.of(MessageChannel.IN_APP); + } + Map matrix = Jsonb.map(String.valueOf(rows.get(0).get("channel_matrix"))); + Object configured = matrix.get(command.priority().name()); + if (!(configured instanceof List values)) { + return List.of(MessageChannel.IN_APP); + } + return values.stream() + .map(String::valueOf) + .map(MessageChannel::valueOf) + .distinct() + .toList(); + } +} diff --git a/platform/backend/src/main/java/com/ai/trackwalker/notification/application/NotificationSubmission.java b/platform/backend/src/main/java/com/ai/trackwalker/notification/application/NotificationSubmission.java new file mode 100644 index 0000000..d24c5cd --- /dev/null +++ b/platform/backend/src/main/java/com/ai/trackwalker/notification/application/NotificationSubmission.java @@ -0,0 +1,10 @@ +package com.ai.trackwalker.notification.application; + +import java.time.Instant; + +public record NotificationSubmission( + String messageId, + String status, + Instant acceptedAt +) { +} diff --git a/platform/backend/src/main/java/com/ai/trackwalker/notification/application/RecipientResolver.java b/platform/backend/src/main/java/com/ai/trackwalker/notification/application/RecipientResolver.java new file mode 100644 index 0000000..e965f72 --- /dev/null +++ b/platform/backend/src/main/java/com/ai/trackwalker/notification/application/RecipientResolver.java @@ -0,0 +1,100 @@ +package com.ai.trackwalker.notification.application; + +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.stereotype.Component; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +@Component +public class RecipientResolver { + private final JdbcTemplate jdbc; + + public RecipientResolver(JdbcTemplate jdbc) { + this.jdbc = jdbc; + } + + public List resolve(SendNotificationCommand.Audience audience) { + if (audience == null) { + return List.of(); + } + Map recipients = new LinkedHashMap<>(); + addUsers(recipients, queryUsersByIds(audience.userIds())); + if (!audience.roleCodes().isEmpty()) { + addUsers(recipients, queryUsersByRoles(audience.roleCodes(), audience.organizationIds())); + } else { + for (String organizationId : audience.organizationIds()) { + addUsers(recipients, jdbc.queryForList(""" + with recursive org_tree as ( + select id from organizations where id=? + union all + select o.id from organizations o join org_tree p on o.parent_id=p.id + ) + select id,org_id,display_name from platform_users + where status='ACTIVE' and org_id in (select id from org_tree) + """, organizationId)); + } + } + return List.copyOf(recipients.values()); + } + + private List> queryUsersByRoles(List roleCodes, List organizationIds) { + String rolePlaceholders = placeholders(roleCodes.size()); + if (organizationIds.isEmpty()) { + return jdbc.queryForList(""" + select distinct u.id,u.org_id,u.display_name + from platform_users u + join user_roles ur on ur.user_id=u.id + join roles r on r.id=ur.role_id + where u.status='ACTIVE' and r.code in (%s) + """.formatted(rolePlaceholders), roleCodes.toArray()); + } + String organizationPlaceholders = placeholders(organizationIds.size()); + List arguments = new ArrayList<>(organizationIds); + arguments.addAll(roleCodes); + return jdbc.queryForList(""" + with recursive org_tree as ( + select id from organizations where id in (%s) + union all + select o.id from organizations o join org_tree p on o.parent_id=p.id + ) + select distinct u.id,u.org_id,u.display_name + from platform_users u + join user_roles ur on ur.user_id=u.id + join roles r on r.id=ur.role_id + where u.status='ACTIVE' and u.org_id in (select id from org_tree) + and r.code in (%s) + """.formatted(organizationPlaceholders, rolePlaceholders), arguments.toArray()); + } + + private List> queryUsersByIds(List userIds) { + if (userIds.isEmpty()) { + return List.of(); + } + String placeholders = placeholders(userIds.size()); + return jdbc.queryForList(""" + select id,org_id,display_name from platform_users + where status='ACTIVE' and id in (%s) + """.formatted(placeholders), userIds.toArray()); + } + + private String placeholders(int size) { + return String.join(",", java.util.Collections.nCopies(size, "?")); + } + + private void addUsers(Map target, List> rows) { + for (Map row : rows) { + String userId = String.valueOf(row.get("id")); + target.putIfAbsent(userId, new ResolvedRecipient( + userId, + row.get("org_id") == null ? null : String.valueOf(row.get("org_id")), + row.get("display_name") == null ? userId : String.valueOf(row.get("display_name")) + )); + } + } + + public record ResolvedRecipient(String userId, String organizationId, String displayName) { + } +} diff --git a/platform/backend/src/main/java/com/ai/trackwalker/notification/application/SendNotificationCommand.java b/platform/backend/src/main/java/com/ai/trackwalker/notification/application/SendNotificationCommand.java new file mode 100644 index 0000000..04201ef --- /dev/null +++ b/platform/backend/src/main/java/com/ai/trackwalker/notification/application/SendNotificationCommand.java @@ -0,0 +1,48 @@ +package com.ai.trackwalker.notification.application; + +import com.ai.trackwalker.notification.domain.MessageChannel; +import com.ai.trackwalker.notification.domain.MessageKind; +import com.ai.trackwalker.notification.domain.MessagePriority; + +import java.time.Instant; +import java.util.List; +import java.util.Map; + +/** + * Provider-neutral message intent submitted by a business module. + */ +public record SendNotificationCommand( + String idempotencyKey, + String tenantId, + String scenarioCode, + MessageKind kind, + MessagePriority priority, + BusinessReference businessReference, + Audience audience, + String templateCode, + Integer templateVersion, + Map variables, + List requestedChannels, + Instant expiresAt, + String traceId +) { + public SendNotificationCommand { + variables = variables == null ? Map.of() : Map.copyOf(variables); + requestedChannels = requestedChannels == null ? List.of() : List.copyOf(requestedChannels); + } + + public record BusinessReference(String type, String id) { + } + + public record Audience( + List userIds, + List organizationIds, + List roleCodes + ) { + public Audience { + userIds = userIds == null ? List.of() : List.copyOf(userIds); + organizationIds = organizationIds == null ? List.of() : List.copyOf(organizationIds); + roleCodes = roleCodes == null ? List.of() : List.copyOf(roleCodes); + } + } +} diff --git a/platform/backend/src/main/java/com/ai/trackwalker/notification/channel/ChannelSendCommand.java b/platform/backend/src/main/java/com/ai/trackwalker/notification/channel/ChannelSendCommand.java new file mode 100644 index 0000000..073b33f --- /dev/null +++ b/platform/backend/src/main/java/com/ai/trackwalker/notification/channel/ChannelSendCommand.java @@ -0,0 +1,22 @@ +package com.ai.trackwalker.notification.channel; + +import java.util.Map; + +/** + * A fully resolved and rendered delivery. Provider adapters must not query business data. + */ +public record ChannelSendCommand( + String deliveryId, + String messageId, + String recipient, + String templateCode, + Map templateParameters, + String title, + String body, + Map metadata +) { + public ChannelSendCommand { + templateParameters = templateParameters == null ? Map.of() : Map.copyOf(templateParameters); + metadata = metadata == null ? Map.of() : Map.copyOf(metadata); + } +} diff --git a/platform/backend/src/main/java/com/ai/trackwalker/notification/channel/ChannelSendResult.java b/platform/backend/src/main/java/com/ai/trackwalker/notification/channel/ChannelSendResult.java new file mode 100644 index 0000000..ad5189b --- /dev/null +++ b/platform/backend/src/main/java/com/ai/trackwalker/notification/channel/ChannelSendResult.java @@ -0,0 +1,17 @@ +package com.ai.trackwalker.notification.channel; + +import com.ai.trackwalker.notification.domain.DeliveryStatus; + +import java.util.Map; + +public record ChannelSendResult( + DeliveryStatus status, + String providerMessageId, + String providerCode, + String providerMessage, + Map receipt +) { + public ChannelSendResult { + receipt = receipt == null ? Map.of() : Map.copyOf(receipt); + } +} diff --git a/platform/backend/src/main/java/com/ai/trackwalker/notification/channel/ChannelSenderRegistry.java b/platform/backend/src/main/java/com/ai/trackwalker/notification/channel/ChannelSenderRegistry.java new file mode 100644 index 0000000..de3534e --- /dev/null +++ b/platform/backend/src/main/java/com/ai/trackwalker/notification/channel/ChannelSenderRegistry.java @@ -0,0 +1,29 @@ +package com.ai.trackwalker.notification.channel; + +import com.ai.trackwalker.notification.domain.MessageChannel; +import org.springframework.stereotype.Component; + +import java.util.EnumMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; + +@Component +public class ChannelSenderRegistry { + private final Map senders; + + public ChannelSenderRegistry(List strategies) { + Map configured = new EnumMap<>(MessageChannel.class); + for (NotificationChannelSender strategy : strategies) { + NotificationChannelSender previous = configured.put(strategy.channel(), strategy); + if (previous != null) { + throw new IllegalStateException("Duplicate notification channel sender: " + strategy.channel()); + } + } + this.senders = Map.copyOf(configured); + } + + public Optional find(MessageChannel channel) { + return Optional.ofNullable(senders.get(channel)); + } +} diff --git a/platform/backend/src/main/java/com/ai/trackwalker/notification/channel/NotificationChannelSender.java b/platform/backend/src/main/java/com/ai/trackwalker/notification/channel/NotificationChannelSender.java new file mode 100644 index 0000000..3725c66 --- /dev/null +++ b/platform/backend/src/main/java/com/ai/trackwalker/notification/channel/NotificationChannelSender.java @@ -0,0 +1,12 @@ +package com.ai.trackwalker.notification.channel; + +import com.ai.trackwalker.notification.domain.MessageChannel; + +/** + * Strategy SPI implemented once per logical channel/provider combination. + */ +public interface NotificationChannelSender { + MessageChannel channel(); + + ChannelSendResult send(ChannelSendCommand command); +} diff --git a/platform/backend/src/main/java/com/ai/trackwalker/notification/channel/inapp/InAppChannelSender.java b/platform/backend/src/main/java/com/ai/trackwalker/notification/channel/inapp/InAppChannelSender.java new file mode 100644 index 0000000..f3efa0b --- /dev/null +++ b/platform/backend/src/main/java/com/ai/trackwalker/notification/channel/inapp/InAppChannelSender.java @@ -0,0 +1,47 @@ +package com.ai.trackwalker.notification.channel.inapp; + +import com.ai.trackwalker.common.Ids; +import com.ai.trackwalker.notification.channel.ChannelSendCommand; +import com.ai.trackwalker.notification.channel.ChannelSendResult; +import com.ai.trackwalker.notification.channel.NotificationChannelSender; +import com.ai.trackwalker.notification.domain.DeliveryStatus; +import com.ai.trackwalker.notification.domain.MessageChannel; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.stereotype.Component; + +import java.util.Map; + +@Component +public class InAppChannelSender implements NotificationChannelSender { + private final JdbcTemplate jdbc; + + public InAppChannelSender(JdbcTemplate jdbc) { + this.jdbc = jdbc; + } + + @Override + public MessageChannel channel() { + return MessageChannel.IN_APP; + } + + @Override + public ChannelSendResult send(ChannelSendCommand command) { + jdbc.update(""" + insert into notification_center.in_app_messages( + id,message_id,delivery_id,recipient_user_id,title,body, + business_type,business_id,priority,status,created_at,updated_at + ) + select ?,m.id,?, ?,?,?,m.business_type,m.business_id,m.priority,'UNREAD',now(),now() + from notification_center.message_requests m where m.id=? + on conflict(delivery_id) do nothing + """, Ids.next("in-app"), command.deliveryId(), command.recipient(), command.title(), + command.body(), command.messageId()); + return new ChannelSendResult( + DeliveryStatus.DELIVERED, + command.deliveryId(), + "LOCAL_INBOX", + "Stored in recipient inbox", + Map.of("recipient_user_id", command.recipient()) + ); + } +} diff --git a/platform/backend/src/main/java/com/ai/trackwalker/notification/channel/inapp/package-info.java b/platform/backend/src/main/java/com/ai/trackwalker/notification/channel/inapp/package-info.java new file mode 100644 index 0000000..c1774af --- /dev/null +++ b/platform/backend/src/main/java/com/ai/trackwalker/notification/channel/inapp/package-info.java @@ -0,0 +1,2 @@ +/** Local in-app inbox channel adapter. */ +package com.ai.trackwalker.notification.channel.inapp; diff --git a/platform/backend/src/main/java/com/ai/trackwalker/notification/channel/mock/MockChannelConfiguration.java b/platform/backend/src/main/java/com/ai/trackwalker/notification/channel/mock/MockChannelConfiguration.java new file mode 100644 index 0000000..49d37f7 --- /dev/null +++ b/platform/backend/src/main/java/com/ai/trackwalker/notification/channel/mock/MockChannelConfiguration.java @@ -0,0 +1,31 @@ +package com.ai.trackwalker.notification.channel.mock; + +import com.ai.trackwalker.notification.channel.NotificationChannelSender; +import com.ai.trackwalker.notification.domain.MessageChannel; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +@Configuration(proxyBeanMethods = false) +@ConditionalOnProperty( + prefix = "rail.notification", + name = "mock-external-channels", + havingValue = "true", + matchIfMissing = false +) +public class MockChannelConfiguration { + @Bean + NotificationChannelSender mockSmsChannelSender() { + return new MockChannelSender(MessageChannel.SMS); + } + + @Bean + NotificationChannelSender mockEmailChannelSender() { + return new MockChannelSender(MessageChannel.EMAIL); + } + + @Bean + NotificationChannelSender mockVoiceChannelSender() { + return new MockChannelSender(MessageChannel.VOICE); + } +} diff --git a/platform/backend/src/main/java/com/ai/trackwalker/notification/channel/mock/MockChannelSender.java b/platform/backend/src/main/java/com/ai/trackwalker/notification/channel/mock/MockChannelSender.java new file mode 100644 index 0000000..3027beb --- /dev/null +++ b/platform/backend/src/main/java/com/ai/trackwalker/notification/channel/mock/MockChannelSender.java @@ -0,0 +1,40 @@ +package com.ai.trackwalker.notification.channel.mock; + +import com.ai.trackwalker.common.Ids; +import com.ai.trackwalker.notification.channel.ChannelSendCommand; +import com.ai.trackwalker.notification.channel.ChannelSendResult; +import com.ai.trackwalker.notification.channel.NotificationChannelSender; +import com.ai.trackwalker.notification.domain.DeliveryStatus; +import com.ai.trackwalker.notification.domain.MessageChannel; + +import java.time.Instant; +import java.util.Map; + +public class MockChannelSender implements NotificationChannelSender { + private final MessageChannel channel; + + public MockChannelSender(MessageChannel channel) { + this.channel = channel; + } + + @Override + public MessageChannel channel() { + return channel; + } + + @Override + public ChannelSendResult send(ChannelSendCommand command) { + String providerMessageId = Ids.next("mock-" + channel.name().toLowerCase()); + return new ChannelSendResult( + DeliveryStatus.PROVIDER_ACCEPTED, + providerMessageId, + "MOCK_ACCEPTED", + "Mock provider accepted delivery", + Map.of( + "mock", true, + "channel", channel.name(), + "accepted_at", Instant.now().toString() + ) + ); + } +} diff --git a/platform/backend/src/main/java/com/ai/trackwalker/notification/channel/tencent/ses/package-info.java b/platform/backend/src/main/java/com/ai/trackwalker/notification/channel/tencent/ses/package-info.java new file mode 100644 index 0000000..3a2b15e --- /dev/null +++ b/platform/backend/src/main/java/com/ai/trackwalker/notification/channel/tencent/ses/package-info.java @@ -0,0 +1,2 @@ +/** Tencent Cloud Simple Email Service adapter. */ +package com.ai.trackwalker.notification.channel.tencent.ses; diff --git a/platform/backend/src/main/java/com/ai/trackwalker/notification/channel/tencent/sms/package-info.java b/platform/backend/src/main/java/com/ai/trackwalker/notification/channel/tencent/sms/package-info.java new file mode 100644 index 0000000..836ad6a --- /dev/null +++ b/platform/backend/src/main/java/com/ai/trackwalker/notification/channel/tencent/sms/package-info.java @@ -0,0 +1,2 @@ +/** Tencent Cloud SMS adapter. */ +package com.ai.trackwalker.notification.channel.tencent.sms; diff --git a/platform/backend/src/main/java/com/ai/trackwalker/notification/channel/tencent/vms/package-info.java b/platform/backend/src/main/java/com/ai/trackwalker/notification/channel/tencent/vms/package-info.java new file mode 100644 index 0000000..49ca6e5 --- /dev/null +++ b/platform/backend/src/main/java/com/ai/trackwalker/notification/channel/tencent/vms/package-info.java @@ -0,0 +1,2 @@ +/** Tencent Cloud Voice Message Service adapter. */ +package com.ai.trackwalker.notification.channel.tencent.vms; diff --git a/platform/backend/src/main/java/com/ai/trackwalker/notification/domain/DeliveryStatus.java b/platform/backend/src/main/java/com/ai/trackwalker/notification/domain/DeliveryStatus.java new file mode 100644 index 0000000..9ca1263 --- /dev/null +++ b/platform/backend/src/main/java/com/ai/trackwalker/notification/domain/DeliveryStatus.java @@ -0,0 +1,18 @@ +package com.ai.trackwalker.notification.domain; + +/** + * Provider acceptance and final delivery are intentionally represented as different states. + */ +public enum DeliveryStatus { + PENDING, + SENDING, + PROVIDER_ACCEPTED, + DELIVERED, + READ, + RETRY_WAIT, + UNKNOWN, + FAILED_PERMANENT, + SKIPPED_CHANNEL_DISABLED, + EXPIRED, + CANCELLED +} diff --git a/platform/backend/src/main/java/com/ai/trackwalker/notification/domain/MessageChannel.java b/platform/backend/src/main/java/com/ai/trackwalker/notification/domain/MessageChannel.java new file mode 100644 index 0000000..3bfe993 --- /dev/null +++ b/platform/backend/src/main/java/com/ai/trackwalker/notification/domain/MessageChannel.java @@ -0,0 +1,14 @@ +package com.ai.trackwalker.notification.domain; + +/** + * Logical delivery channels. Provider selection belongs to routing policy, not this enum. + */ +public enum MessageChannel { + IN_APP, + WEB_POPUP, + APP_PUSH, + WECHAT, + SMS, + EMAIL, + VOICE +} diff --git a/platform/backend/src/main/java/com/ai/trackwalker/notification/domain/MessageKind.java b/platform/backend/src/main/java/com/ai/trackwalker/notification/domain/MessageKind.java new file mode 100644 index 0000000..bfecd9f --- /dev/null +++ b/platform/backend/src/main/java/com/ai/trackwalker/notification/domain/MessageKind.java @@ -0,0 +1,9 @@ +package com.ai.trackwalker.notification.domain; + +/** + * Verification codes deliberately do not appear here; they use the isolated verification flow. + */ +public enum MessageKind { + NOTIFICATION, + ALERT +} diff --git a/platform/backend/src/main/java/com/ai/trackwalker/notification/domain/MessagePriority.java b/platform/backend/src/main/java/com/ai/trackwalker/notification/domain/MessagePriority.java new file mode 100644 index 0000000..10acec3 --- /dev/null +++ b/platform/backend/src/main/java/com/ai/trackwalker/notification/domain/MessagePriority.java @@ -0,0 +1,8 @@ +package com.ai.trackwalker.notification.domain; + +public enum MessagePriority { + P1_CRITICAL, + P2_HIGH, + P3_NORMAL, + P4_LOW +} diff --git a/platform/backend/src/main/java/com/ai/trackwalker/notification/infrastructure/JdbcMessageStore.java b/platform/backend/src/main/java/com/ai/trackwalker/notification/infrastructure/JdbcMessageStore.java new file mode 100644 index 0000000..725f83d --- /dev/null +++ b/platform/backend/src/main/java/com/ai/trackwalker/notification/infrastructure/JdbcMessageStore.java @@ -0,0 +1,98 @@ +package com.ai.trackwalker.notification.infrastructure; + +import com.ai.trackwalker.common.Ids; +import com.ai.trackwalker.common.Jsonb; +import com.ai.trackwalker.notification.application.MessageTemplateRenderer; +import com.ai.trackwalker.notification.application.RecipientResolver; +import com.ai.trackwalker.notification.application.SendNotificationCommand; +import com.ai.trackwalker.notification.domain.MessageChannel; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.stereotype.Repository; + +import java.sql.Timestamp; +import java.time.Instant; +import java.util.List; +import java.util.Map; + +@Repository +public class JdbcMessageStore { + private final JdbcTemplate jdbc; + + public JdbcMessageStore(JdbcTemplate jdbc) { + this.jdbc = jdbc; + } + + public StoredMessage create( + SendNotificationCommand command, + List recipients, + List channels, + MessageTemplateRenderer.RenderedMessage rendered + ) { + List> existing = jdbc.queryForList(""" + select id,status,created_at from notification_center.message_requests + where tenant_id=? and idempotency_key=? + """, command.tenantId(), command.idempotencyKey()); + if (!existing.isEmpty()) { + Map row = existing.get(0); + return new StoredMessage(String.valueOf(row.get("id")), String.valueOf(row.get("status")), + ((Timestamp) row.get("created_at")).toInstant(), true); + } + + Instant now = Instant.now(); + String messageId = Ids.next("message"); + int inserted = jdbc.update(""" + insert into notification_center.message_requests( + id,tenant_id,idempotency_key,scenario_code,message_kind,priority, + business_type,business_id,audience,template_code,template_version,variables, + requested_channels,status,expires_at,trace_id,created_at,updated_at + ) values (?,?,?,?,?,?,?,?,?::jsonb,?,?,?::jsonb,?::jsonb,'ACCEPTED',?,?,?,?) + on conflict(tenant_id,idempotency_key) do nothing + """, + messageId, command.tenantId(), command.idempotencyKey(), command.scenarioCode(), + command.kind().name(), command.priority().name(), + command.businessReference() == null ? null : command.businessReference().type(), + command.businessReference() == null ? null : command.businessReference().id(), + Jsonb.write(command.audience()), command.templateCode(), rendered.version(), + Jsonb.write(command.variables()), Jsonb.write(channels.stream().map(Enum::name).toList()), + command.expiresAt() == null ? null : Timestamp.from(command.expiresAt()), command.traceId(), + Timestamp.from(now), Timestamp.from(now)); + if (inserted == 0) { + Map row = jdbc.queryForMap(""" + select id,status,created_at from notification_center.message_requests + where tenant_id=? and idempotency_key=? + """, command.tenantId(), command.idempotencyKey()); + return new StoredMessage(String.valueOf(row.get("id")), String.valueOf(row.get("status")), + ((Timestamp) row.get("created_at")).toInstant(), true); + } + + for (RecipientResolver.ResolvedRecipient recipient : recipients) { + String recipientId = Ids.next("message-recipient"); + jdbc.update(""" + insert into notification_center.message_recipients( + id,message_id,recipient_user_id,organization_id,display_name,recipient_snapshot,created_at + ) values (?,?,?,?,?,?::jsonb,?) + """, recipientId, messageId, recipient.userId(), recipient.organizationId(), recipient.displayName(), + Jsonb.write(Map.of("user_id", recipient.userId(), "display_name", recipient.displayName())), + Timestamp.from(now)); + for (MessageChannel channel : channels) { + jdbc.update(""" + insert into notification_center.message_deliveries( + id,message_id,recipient_id,channel,provider,status,title,body, + next_attempt_at,expires_at,created_at,updated_at + ) values (?,?,?,?,?,'PENDING',?,?,?,?,?,?) + """, Ids.next("delivery"), messageId, recipientId, channel.name(), provider(channel), + rendered.title(), rendered.body(), Timestamp.from(now), + command.expiresAt() == null ? null : Timestamp.from(command.expiresAt()), + Timestamp.from(now), Timestamp.from(now)); + } + } + return new StoredMessage(messageId, "ACCEPTED", now, false); + } + + private String provider(MessageChannel channel) { + return channel == MessageChannel.IN_APP ? "LOCAL" : "MOCK_OR_DISABLED"; + } + + public record StoredMessage(String id, String status, Instant createdAt, boolean idempotent) { + } +} diff --git a/platform/backend/src/main/java/com/ai/trackwalker/notification/infrastructure/NotificationDeliveryWorker.java b/platform/backend/src/main/java/com/ai/trackwalker/notification/infrastructure/NotificationDeliveryWorker.java new file mode 100644 index 0000000..aa532d3 --- /dev/null +++ b/platform/backend/src/main/java/com/ai/trackwalker/notification/infrastructure/NotificationDeliveryWorker.java @@ -0,0 +1,218 @@ +package com.ai.trackwalker.notification.infrastructure; + +import com.ai.trackwalker.common.Ids; +import com.ai.trackwalker.common.Jsonb; +import com.ai.trackwalker.notification.channel.ChannelSendCommand; +import com.ai.trackwalker.notification.channel.ChannelSendResult; +import com.ai.trackwalker.notification.channel.ChannelSenderRegistry; +import com.ai.trackwalker.notification.domain.DeliveryStatus; +import com.ai.trackwalker.notification.domain.MessageChannel; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; + +import java.sql.Timestamp; +import java.time.Instant; +import java.util.List; +import java.util.Map; + +@Component +public class NotificationDeliveryWorker { + private static final long[] RETRY_DELAYS_SECONDS = {10, 30, 120, 600, 1800}; + + private final JdbcTemplate jdbc; + private final ChannelSenderRegistry senders; + private final NotificationProperties properties; + + public NotificationDeliveryWorker( + JdbcTemplate jdbc, + ChannelSenderRegistry senders, + NotificationProperties properties + ) { + this.jdbc = jdbc; + this.senders = senders; + this.properties = properties; + } + + @Scheduled(initialDelay = 5000, fixedDelayString = "${rail.notification.dispatcher-delay-ms:3000}") + public void dispatch() { + if (!properties.isEnabled()) { + return; + } + recoverStaleClaims(); + expireDeliveries(); + for (Map delivery : claim()) { + send(delivery); + } + } + + private void expireDeliveries() { + List messageIds = jdbc.queryForList(""" + select distinct message_id from notification_center.message_deliveries + where status in ('PENDING','RETRY_WAIT') and expires_at is not null and expires_at<=now() + """, String.class); + jdbc.update(""" + update notification_center.message_deliveries + set status='EXPIRED',updated_at=now(),last_error_code='MESSAGE_EXPIRED' + where status in ('PENDING','RETRY_WAIT') and expires_at is not null and expires_at<=now() + """); + messageIds.forEach(this::refreshMessage); + } + + private void recoverStaleClaims() { + jdbc.update(""" + update notification_center.message_deliveries + set status='RETRY_WAIT',next_attempt_at=now(),last_error_code='STALE_CLAIM_RECOVERED', + last_error_message='Previous dispatcher stopped before recording a result',updated_at=now() + where status='SENDING' and updated_at < now() - (? * interval '1 second') + """, Math.max(30, properties.getClaimTimeoutSeconds())); + } + + private List> claim() { + return jdbc.queryForList(""" + with candidates as ( + select id from notification_center.message_deliveries + where status in ('PENDING','RETRY_WAIT') and next_attempt_at<=now() + and (expires_at is null or expires_at>now()) + order by created_at + limit ? for update skip locked + ), claimed as ( + update notification_center.message_deliveries d + set status='SENDING',attempt_count=d.attempt_count+1,updated_at=now() + from candidates c where d.id=c.id + returning d.* + ) + select c.id,c.message_id,c.channel,c.attempt_count,c.title,c.body,c.expires_at, + r.recipient_user_id,m.template_code,m.variables::text as variables + from claimed c + join notification_center.message_recipients r on r.id=c.recipient_id + join notification_center.message_requests m on m.id=c.message_id + order by c.created_at + """, properties.getBatchSize()); + } + + private void send(Map delivery) { + String deliveryId = String.valueOf(delivery.get("id")); + String messageId = String.valueOf(delivery.get("message_id")); + int attempt = ((Number) delivery.get("attempt_count")).intValue(); + MessageChannel channel = MessageChannel.valueOf(String.valueOf(delivery.get("channel"))); + Instant startedAt = Instant.now(); + try { + var sender = senders.find(channel); + if (sender.isEmpty()) { + ChannelSendResult skipped = new ChannelSendResult( + DeliveryStatus.SKIPPED_CHANNEL_DISABLED, + null, + "CHANNEL_DISABLED", + "No sender is enabled for channel " + channel, + Map.of("channel", channel.name()) + ); + complete(deliveryId, messageId, attempt, startedAt, skipped); + return; + } + ChannelSendCommand command = new ChannelSendCommand( + deliveryId, + messageId, + String.valueOf(delivery.get("recipient_user_id")), + String.valueOf(delivery.get("template_code")), + Jsonb.map(String.valueOf(delivery.get("variables"))), + String.valueOf(delivery.get("title")), + String.valueOf(delivery.get("body")), + Map.of("channel", channel.name()) + ); + complete(deliveryId, messageId, attempt, startedAt, sender.get().send(command)); + } catch (Exception exception) { + fail(deliveryId, messageId, attempt, startedAt, exception); + } + } + + private void complete( + String deliveryId, + String messageId, + int attempt, + Instant startedAt, + ChannelSendResult result + ) { + Instant now = Instant.now(); + boolean accepted = List.of(DeliveryStatus.PROVIDER_ACCEPTED, DeliveryStatus.DELIVERED, DeliveryStatus.READ) + .contains(result.status()); + boolean delivered = List.of(DeliveryStatus.DELIVERED, DeliveryStatus.READ).contains(result.status()); + jdbc.update(""" + update notification_center.message_deliveries + set status=?,provider_message_id=?,last_error_code=?,last_error_message=?, + sent_at=case when ? then coalesce(sent_at,?) else sent_at end, + delivered_at=case when ? then coalesce(delivered_at,?) else delivered_at end, + updated_at=? where id=? + """, result.status().name(), result.providerMessageId(), + accepted ? null : result.providerCode(), accepted ? null : result.providerMessage(), + accepted, Timestamp.from(now), delivered, Timestamp.from(now), Timestamp.from(now), deliveryId); + recordAttempt(deliveryId, attempt, startedAt, now, result); + refreshMessage(messageId); + } + + private void fail(String deliveryId, String messageId, int attempt, Instant startedAt, Exception exception) { + Instant now = Instant.now(); + boolean exhausted = attempt >= properties.getMaxAttempts(); + DeliveryStatus status = exhausted ? DeliveryStatus.FAILED_PERMANENT : DeliveryStatus.RETRY_WAIT; + long delay = RETRY_DELAYS_SECONDS[Math.min(Math.max(attempt - 1, 0), RETRY_DELAYS_SECONDS.length - 1)]; + String message = safeMessage(exception); + jdbc.update(""" + update notification_center.message_deliveries + set status=?,next_attempt_at=?,last_error_code='CHANNEL_SEND_ERROR',last_error_message=?,updated_at=? + where id=? + """, status.name(), Timestamp.from(now.plusSeconds(delay)), message, Timestamp.from(now), deliveryId); + recordAttempt(deliveryId, attempt, startedAt, now, new ChannelSendResult( + status, null, "CHANNEL_SEND_ERROR", message, Map.of())); + refreshMessage(messageId); + } + + private void recordAttempt( + String deliveryId, + int attempt, + Instant startedAt, + Instant completedAt, + ChannelSendResult result + ) { + jdbc.update(""" + insert into notification_center.delivery_attempts( + id,delivery_id,attempt_no,status,provider_message_id,provider_code, + provider_message,receipt,started_at,completed_at + ) values (?,?,?,?,?,?,?,?::jsonb,?,?) + on conflict(delivery_id,attempt_no) do nothing + """, Ids.next("delivery-attempt"), deliveryId, attempt, result.status().name(), + result.providerMessageId(), result.providerCode(), result.providerMessage(), + Jsonb.write(result.receipt()), Timestamp.from(startedAt), Timestamp.from(completedAt)); + } + + private void refreshMessage(String messageId) { + Map counts = jdbc.queryForMap(""" + select + count(*) filter(where status in ('PENDING','SENDING','RETRY_WAIT')) as pending, + count(*) filter(where status in ('PROVIDER_ACCEPTED','DELIVERED','READ')) as succeeded, + count(*) filter(where status in ('FAILED_PERMANENT','EXPIRED')) as failed, + count(*) filter(where status='SKIPPED_CHANNEL_DISABLED') as skipped + from notification_center.message_deliveries where message_id=? + """, messageId); + long pending = ((Number) counts.get("pending")).longValue(); + long succeeded = ((Number) counts.get("succeeded")).longValue(); + long failed = ((Number) counts.get("failed")).longValue(); + long skipped = ((Number) counts.get("skipped")).longValue(); + String status; + if (pending > 0) { + status = "PROCESSING"; + } else if ((failed > 0 || skipped > 0) && succeeded > 0) { + status = "PARTIAL_SUCCESS"; + } else if (failed > 0 || (skipped > 0 && succeeded == 0)) { + status = "FAILED"; + } else { + status = "SUCCEEDED"; + } + jdbc.update("update notification_center.message_requests set status=?,updated_at=now() where id=?", + status, messageId); + } + + private String safeMessage(Exception exception) { + String value = exception.getMessage() == null ? exception.getClass().getSimpleName() : exception.getMessage(); + return value.length() > 1000 ? value.substring(0, 1000) : value; + } +} diff --git a/platform/backend/src/main/java/com/ai/trackwalker/notification/infrastructure/NotificationProperties.java b/platform/backend/src/main/java/com/ai/trackwalker/notification/infrastructure/NotificationProperties.java new file mode 100644 index 0000000..ed74360 --- /dev/null +++ b/platform/backend/src/main/java/com/ai/trackwalker/notification/infrastructure/NotificationProperties.java @@ -0,0 +1,54 @@ +package com.ai.trackwalker.notification.infrastructure; + +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.stereotype.Component; + +@Component +@ConfigurationProperties(prefix = "rail.notification") +public class NotificationProperties { + private boolean enabled = true; + private boolean mockExternalChannels = false; + private int batchSize = 50; + private int maxAttempts = 5; + private int claimTimeoutSeconds = 300; + + public boolean isEnabled() { + return enabled; + } + + public void setEnabled(boolean enabled) { + this.enabled = enabled; + } + + public boolean isMockExternalChannels() { + return mockExternalChannels; + } + + public void setMockExternalChannels(boolean mockExternalChannels) { + this.mockExternalChannels = mockExternalChannels; + } + + public int getBatchSize() { + return batchSize; + } + + public void setBatchSize(int batchSize) { + this.batchSize = batchSize; + } + + public int getMaxAttempts() { + return maxAttempts; + } + + public void setMaxAttempts(int maxAttempts) { + this.maxAttempts = maxAttempts; + } + + public int getClaimTimeoutSeconds() { + return claimTimeoutSeconds; + } + + public void setClaimTimeoutSeconds(int claimTimeoutSeconds) { + this.claimTimeoutSeconds = claimTimeoutSeconds; + } +} diff --git a/platform/backend/src/main/java/com/ai/trackwalker/notification/infrastructure/package-info.java b/platform/backend/src/main/java/com/ai/trackwalker/notification/infrastructure/package-info.java new file mode 100644 index 0000000..70e37d2 --- /dev/null +++ b/platform/backend/src/main/java/com/ai/trackwalker/notification/infrastructure/package-info.java @@ -0,0 +1,4 @@ +/** + * Persistence, event-bus, provider routing, scheduling, and callback infrastructure. + */ +package com.ai.trackwalker.notification.infrastructure; diff --git a/platform/backend/src/main/java/com/ai/trackwalker/notification/package-info.java b/platform/backend/src/main/java/com/ai/trackwalker/notification/package-info.java new file mode 100644 index 0000000..6507f1d --- /dev/null +++ b/platform/backend/src/main/java/com/ai/trackwalker/notification/package-info.java @@ -0,0 +1,9 @@ +/** + * Unified notification module. + * + *

This package is a modular-monolith boundary. Business modules submit message intent through + * the application facade and must not call provider SDKs or access notification persistence + * directly. The package is intentionally structured so it can be extracted into an independent + * service later without changing the business-facing contract.

+ */ +package com.ai.trackwalker.notification; diff --git a/platform/backend/src/main/java/com/ai/trackwalker/notification/verification/package-info.java b/platform/backend/src/main/java/com/ai/trackwalker/notification/verification/package-info.java new file mode 100644 index 0000000..df19a56 --- /dev/null +++ b/platform/backend/src/main/java/com/ai/trackwalker/notification/verification/package-info.java @@ -0,0 +1,7 @@ +/** + * Isolated verification-code use cases. + * + *

This package may reuse channel clients but must not send codes through the ordinary + * notification facade or persist plaintext verification codes.

+ */ +package com.ai.trackwalker.notification.verification; diff --git a/platform/backend/src/main/resources/application.yml b/platform/backend/src/main/resources/application.yml index 509b7cc..60e24a6 100644 --- a/platform/backend/src/main/resources/application.yml +++ b/platform/backend/src/main/resources/application.yml @@ -73,6 +73,14 @@ rail: timeout-seconds: ${UAV_ACCESS_TIMEOUT_SECONDS:60} event-topic: ${UAV_ACCESS_EVENT_TOPIC:uav.access.events.v1} event-consumer-group: ${UAV_ACCESS_EVENT_CONSUMER_GROUP:rail-platform-uav-access-v1} + notification: + enabled: ${RAIL_NOTIFICATION_CENTER_ENABLED:true} + mock-external-channels: ${RAIL_NOTIFICATION_MOCK_EXTERNAL_CHANNELS:false} + batch-size: ${RAIL_NOTIFICATION_BATCH_SIZE:50} + max-attempts: ${RAIL_NOTIFICATION_MAX_ATTEMPTS:5} + claim-timeout-seconds: ${RAIL_NOTIFICATION_CLAIM_TIMEOUT_SECONDS:300} + dispatcher-delay-ms: ${RAIL_NOTIFICATION_DISPATCHER_DELAY_MS:3000} + event-projector-delay-ms: ${RAIL_NOTIFICATION_EVENT_PROJECTOR_DELAY_MS:3000} model-test: core-pool-size: ${RAIL_MODEL_TEST_CORE_POOL_SIZE:2} max-pool-size: ${RAIL_MODEL_TEST_MAX_POOL_SIZE:4} diff --git a/platform/backend/src/main/resources/db/migration/V25__notification_center_foundation.sql b/platform/backend/src/main/resources/db/migration/V25__notification_center_foundation.sql new file mode 100644 index 0000000..f9004e9 --- /dev/null +++ b/platform/backend/src/main/resources/db/migration/V25__notification_center_foundation.sql @@ -0,0 +1,170 @@ +create schema if not exists notification_center; + +create table if not exists notification_center.message_templates ( + id varchar(64) primary key, + code varchar(128) not null, + version_no integer not null, + title_template text not null, + body_template text not null, + variable_schema jsonb not null default '{}'::jsonb, + status varchar(32) not null, + created_at timestamptz not null, + updated_at timestamptz not null, + unique(code, version_no) +); + +create table if not exists notification_center.message_policies ( + id varchar(64) primary key, + code varchar(128) not null unique, + scenario_pattern varchar(128) not null, + channel_matrix jsonb not null, + status varchar(32) not null, + version_no integer not null default 1, + created_at timestamptz not null, + updated_at timestamptz not null +); + +create table if not exists notification_center.message_requests ( + id varchar(64) primary key, + tenant_id varchar(64) not null, + idempotency_key varchar(256) not null, + scenario_code varchar(128) not null, + message_kind varchar(32) not null, + priority varchar(32) not null, + business_type varchar(64), + business_id varchar(128), + audience jsonb not null, + template_code varchar(128) not null, + template_version integer, + variables jsonb not null, + requested_channels jsonb not null default '[]'::jsonb, + status varchar(32) not null, + expires_at timestamptz, + trace_id varchar(128), + created_at timestamptz not null, + updated_at timestamptz not null, + unique(tenant_id, idempotency_key) +); + +create table if not exists notification_center.message_recipients ( + id varchar(64) primary key, + message_id varchar(64) not null references notification_center.message_requests(id), + recipient_user_id varchar(64) not null, + organization_id varchar(64), + display_name varchar(128), + recipient_snapshot jsonb not null default '{}'::jsonb, + created_at timestamptz not null, + unique(message_id, recipient_user_id) +); + +create table if not exists notification_center.message_deliveries ( + id varchar(64) primary key, + message_id varchar(64) not null references notification_center.message_requests(id), + recipient_id varchar(64) not null references notification_center.message_recipients(id), + channel varchar(32) not null, + provider varchar(64) not null, + status varchar(32) not null, + title text, + body text, + provider_message_id varchar(256), + attempt_count integer not null default 0, + next_attempt_at timestamptz not null, + last_error_code varchar(128), + last_error_message text, + sent_at timestamptz, + delivered_at timestamptz, + read_at timestamptz, + expires_at timestamptz, + created_at timestamptz not null, + updated_at timestamptz not null, + unique(message_id, recipient_id, channel) +); + +create index if not exists idx_message_delivery_dispatch + on notification_center.message_deliveries(status, next_attempt_at, created_at); +create index if not exists idx_message_delivery_recipient + on notification_center.message_deliveries(recipient_id, created_at desc); + +create table if not exists notification_center.delivery_attempts ( + id varchar(64) primary key, + delivery_id varchar(64) not null references notification_center.message_deliveries(id), + attempt_no integer not null, + status varchar(32) not null, + provider_message_id varchar(256), + provider_code varchar(128), + provider_message text, + receipt jsonb not null default '{}'::jsonb, + started_at timestamptz not null, + completed_at timestamptz not null, + unique(delivery_id, attempt_no) +); + +create table if not exists notification_center.in_app_messages ( + id varchar(64) primary key, + message_id varchar(64) not null references notification_center.message_requests(id), + delivery_id varchar(64) not null unique references notification_center.message_deliveries(id), + recipient_user_id varchar(64) not null, + title text not null, + body text not null, + business_type varchar(64), + business_id varchar(128), + priority varchar(32) not null, + status varchar(32) not null, + read_at timestamptz, + clicked_at timestamptz, + archived_at timestamptz, + created_at timestamptz not null, + updated_at timestamptz not null +); + +create index if not exists idx_in_app_recipient_status + on notification_center.in_app_messages(recipient_user_id, status, created_at desc); + +create table if not exists notification_center.event_inbox ( + source_event_id varchar(64) primary key, + event_type varchar(128) not null, + status varchar(32) not null, + attempt_count integer not null default 0, + last_error text, + processed_at timestamptz, + created_at timestamptz not null, + updated_at timestamptz not null +); + +insert into notification_center.message_templates( + id,code,version_no,title_template,body_template,variable_schema,status,created_at,updated_at +) values ( + 'message-template-alarm-detected-v1','alarm-detected',1, + '{{level_label}}告警:{{scene}}', + '{{line_name}} {{mileage}} 发现{{category}},请及时进入告警中心研判。', + '{"required":["level_label","scene","category"]}'::jsonb, + 'PUBLISHED',now(),now() +) on conflict (id) do nothing; + +insert into notification_center.message_policies( + id,code,scenario_pattern,channel_matrix,status,version_no,created_at,updated_at +) values ( + 'message-policy-railway-alarm','RAILWAY_ALARM_DEFAULT','ALARM_%', + '{ + "P1_CRITICAL":["IN_APP","WEB_POPUP","APP_PUSH","WECHAT","SMS","EMAIL"], + "P2_HIGH":["IN_APP","APP_PUSH","WECHAT","SMS"], + "P3_NORMAL":["IN_APP","APP_PUSH","WECHAT"], + "P4_LOW":["IN_APP","WEB_POPUP"] + }'::jsonb, + 'ACTIVE',1,now(),now() +) on conflict (id) do nothing; + +-- Do not replay historical alarm events when this module is first deployed. +insert into notification_center.event_inbox( + source_event_id,event_type,status,attempt_count,processed_at,created_at,updated_at +) +select id,event_type,'SKIPPED_LEGACY',0,now(),now(),now() +from platform_events +where event_type='alarm.detected' +on conflict (source_event_id) do nothing; + +-- The first alarm audiences use these roles, so their members must be able to read their own inbox. +update roles +set permissions = permissions || '["notification:read"]'::jsonb +where code in ('TASK_DISPATCHER','WORKORDER_REVIEWER') + and not permissions ? 'notification:read'; diff --git a/platform/backend/src/test/java/com/ai/trackwalker/notification/ChannelSenderStrategyTest.java b/platform/backend/src/test/java/com/ai/trackwalker/notification/ChannelSenderStrategyTest.java new file mode 100644 index 0000000..36028b1 --- /dev/null +++ b/platform/backend/src/test/java/com/ai/trackwalker/notification/ChannelSenderStrategyTest.java @@ -0,0 +1,47 @@ +package com.ai.trackwalker.notification; + +import com.ai.trackwalker.notification.channel.ChannelSendCommand; +import com.ai.trackwalker.notification.channel.ChannelSendResult; +import com.ai.trackwalker.notification.channel.ChannelSenderRegistry; +import com.ai.trackwalker.notification.channel.NotificationChannelSender; +import com.ai.trackwalker.notification.channel.mock.MockChannelSender; +import com.ai.trackwalker.notification.domain.DeliveryStatus; +import com.ai.trackwalker.notification.domain.MessageChannel; +import org.junit.jupiter.api.Test; + +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class ChannelSenderStrategyTest { + @Test + void registersAndExecutesMockStrategy() { + MockChannelSender sms = new MockChannelSender(MessageChannel.SMS); + ChannelSenderRegistry registry = new ChannelSenderRegistry(List.of(sms)); + + ChannelSendResult result = registry.find(MessageChannel.SMS).orElseThrow().send(command()); + + assertThat(result.status()).isEqualTo(DeliveryStatus.PROVIDER_ACCEPTED); + assertThat(result.providerCode()).isEqualTo("MOCK_ACCEPTED"); + assertThat(result.receipt()).containsEntry("mock", true).containsEntry("channel", "SMS"); + } + + @Test + void rejectsDuplicateChannelStrategies() { + NotificationChannelSender first = new MockChannelSender(MessageChannel.SMS); + NotificationChannelSender second = new MockChannelSender(MessageChannel.SMS); + + assertThatThrownBy(() -> new ChannelSenderRegistry(List.of(first, second))) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("SMS"); + } + + private ChannelSendCommand command() { + return new ChannelSendCommand( + "delivery-1", "message-1", "user-1", "alarm-detected", + Map.of(), "告警", "请处理", Map.of() + ); + } +} diff --git a/platform/backend/src/test/java/com/ai/trackwalker/notification/DefaultNotificationFacadeTest.java b/platform/backend/src/test/java/com/ai/trackwalker/notification/DefaultNotificationFacadeTest.java new file mode 100644 index 0000000..1c73f04 --- /dev/null +++ b/platform/backend/src/test/java/com/ai/trackwalker/notification/DefaultNotificationFacadeTest.java @@ -0,0 +1,79 @@ +package com.ai.trackwalker.notification; + +import com.ai.trackwalker.notification.application.DefaultNotificationFacade; +import com.ai.trackwalker.notification.application.MessageTemplateRenderer; +import com.ai.trackwalker.notification.application.NotificationPolicyResolver; +import com.ai.trackwalker.notification.application.NotificationSubmission; +import com.ai.trackwalker.notification.application.RecipientResolver; +import com.ai.trackwalker.notification.application.SendNotificationCommand; +import com.ai.trackwalker.notification.domain.MessageChannel; +import com.ai.trackwalker.notification.domain.MessageKind; +import com.ai.trackwalker.notification.domain.MessagePriority; +import com.ai.trackwalker.notification.infrastructure.JdbcMessageStore; +import org.junit.jupiter.api.Test; + +import java.time.Instant; +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +class DefaultNotificationFacadeTest { + @Test + void resolvesPolicyRecipientsAndTemplateBeforePersisting() { + SendNotificationCommand command = command(); + List channels = List.of(MessageChannel.IN_APP, MessageChannel.SMS); + List resolved = List.of( + new RecipientResolver.ResolvedRecipient("user-1", "org-1", "调度员")); + MessageTemplateRenderer.RenderedMessage rendered = + new MessageTemplateRenderer.RenderedMessage(1, "一级告警", "请处理"); + Instant acceptedAt = Instant.now(); + NotificationPolicyResolver policies = new NotificationPolicyResolver(null) { + @Override + public List resolve(SendNotificationCommand ignored) { + return channels; + } + }; + RecipientResolver recipients = new RecipientResolver(null) { + @Override + public List resolve(SendNotificationCommand.Audience ignored) { + return resolved; + } + }; + MessageTemplateRenderer templates = new MessageTemplateRenderer(null) { + @Override + public RenderedMessage render(String code, Integer version, Map variables) { + return rendered; + } + }; + JdbcMessageStore store = new JdbcMessageStore(null) { + @Override + public StoredMessage create( + SendNotificationCommand ignored, + List ignoredRecipients, + List ignoredChannels, + MessageTemplateRenderer.RenderedMessage ignoredMessage + ) { + return new StoredMessage("message-1", "ACCEPTED", acceptedAt, false); + } + }; + DefaultNotificationFacade facade = new DefaultNotificationFacade(policies, recipients, templates, store); + + NotificationSubmission result = facade.submit(command); + + assertThat(result.messageId()).isEqualTo("message-1"); + assertThat(result.status()).isEqualTo("ACCEPTED"); + assertThat(result.acceptedAt()).isEqualTo(acceptedAt); + } + + private SendNotificationCommand command() { + return new SendNotificationCommand( + "alarm-detected:1", "org-1", "ALARM_DETECTED", MessageKind.ALERT, + MessagePriority.P1_CRITICAL, + new SendNotificationCommand.BusinessReference("ALARM", "alarm-1"), + new SendNotificationCommand.Audience(List.of("user-1"), List.of(), List.of()), + "alarm-detected", null, + Map.of("level_label", "一级", "scene", "异物", "category", "异物"), + List.of(), Instant.now().plusSeconds(60), "trace-1" + ); + } +} diff --git a/platform/backend/src/test/java/com/ai/trackwalker/notification/MessageTemplateRendererTest.java b/platform/backend/src/test/java/com/ai/trackwalker/notification/MessageTemplateRendererTest.java new file mode 100644 index 0000000..cffb46a --- /dev/null +++ b/platform/backend/src/test/java/com/ai/trackwalker/notification/MessageTemplateRendererTest.java @@ -0,0 +1,49 @@ +package com.ai.trackwalker.notification; + +import com.ai.trackwalker.notification.application.MessageTemplateRenderer; +import org.junit.jupiter.api.Test; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.web.server.ResponseStatusException; + +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class MessageTemplateRendererTest { + @Test + void validatesAndRendersTemplateVariables() { + MessageTemplateRenderer renderer = new MessageTemplateRenderer(templateJdbc()); + + MessageTemplateRenderer.RenderedMessage result = renderer.render( + "alarm-detected", null, Map.of("scene", "接触网异物", "level", "一级")); + + assertThat(result.version()).isEqualTo(2); + assertThat(result.title()).isEqualTo("一级告警:接触网异物"); + assertThat(result.body()).isEqualTo("请处理接触网异物"); + } + + @Test + void rejectsMissingRequiredVariables() { + MessageTemplateRenderer renderer = new MessageTemplateRenderer(templateJdbc()); + + assertThatThrownBy(() -> renderer.render("alarm-detected", null, Map.of("scene", "异物"))) + .isInstanceOf(ResponseStatusException.class) + .hasMessageContaining("level"); + } + + private JdbcTemplate templateJdbc() { + return new JdbcTemplate() { + @Override + public List> queryForList(String sql, Object... args) { + return List.of(Map.of( + "version_no", 2, + "title_template", "{{level}}告警:{{scene}}", + "body_template", "请处理{{scene}}", + "variable_schema", "{\"required\":[\"level\",\"scene\"]}" + )); + } + }; + } +} diff --git a/platform/backend/src/test/java/com/ai/trackwalker/notification/NotificationPolicyResolverTest.java b/platform/backend/src/test/java/com/ai/trackwalker/notification/NotificationPolicyResolverTest.java new file mode 100644 index 0000000..f5ba71c --- /dev/null +++ b/platform/backend/src/test/java/com/ai/trackwalker/notification/NotificationPolicyResolverTest.java @@ -0,0 +1,49 @@ +package com.ai.trackwalker.notification; + +import com.ai.trackwalker.notification.application.NotificationPolicyResolver; +import com.ai.trackwalker.notification.application.SendNotificationCommand; +import com.ai.trackwalker.notification.domain.MessageChannel; +import com.ai.trackwalker.notification.domain.MessageKind; +import com.ai.trackwalker.notification.domain.MessagePriority; +import org.junit.jupiter.api.Test; +import org.springframework.jdbc.core.JdbcTemplate; + +import java.util.List; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +class NotificationPolicyResolverTest { + @Test + void resolvesConfiguredChannelsByPriority() { + JdbcTemplate jdbc = new JdbcTemplate() { + @Override + public List> queryForList(String sql, Object... args) { + return List.of(Map.of("channel_matrix", """ + {"P1_CRITICAL":["IN_APP","SMS","EMAIL"],"P2_HIGH":["IN_APP","SMS"]} + """)); + } + }; + NotificationPolicyResolver resolver = new NotificationPolicyResolver(jdbc); + + assertThat(resolver.resolve(command(MessagePriority.P1_CRITICAL, List.of()))) + .containsExactly(MessageChannel.IN_APP, MessageChannel.SMS, MessageChannel.EMAIL); + } + + @Test + void explicitChannelsOverridePolicyLookup() { + NotificationPolicyResolver resolver = new NotificationPolicyResolver(new JdbcTemplate()); + + assertThat(resolver.resolve(command(MessagePriority.P2_HIGH, List.of(MessageChannel.VOICE)))) + .containsExactly(MessageChannel.VOICE); + } + + private SendNotificationCommand command(MessagePriority priority, List channels) { + return new SendNotificationCommand( + "key", "tenant", "ALARM_DETECTED", MessageKind.ALERT, priority, + new SendNotificationCommand.BusinessReference("ALARM", "alarm-1"), + new SendNotificationCommand.Audience(List.of("user-1"), List.of(), List.of()), + "alarm-detected", null, Map.of(), channels, null, "trace" + ); + } +}