新增统一消息中心基础模块

建立统一消息请求、模板、策略、接收人解析与可靠投递机制。实现站内信、告警事件接入及短信邮件语音模拟渠道，并补充数据库迁移、测试和设计文档。
2026-08-11 00:25:58 +08:00
parent f79806499d
commit 4e1dc0e0eb
39 changed files with 1799 additions and 0 deletions
@@ -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 升级，应先由告警域更新权威告警状态并发布新事件，消息中心再以新的幂等键执行升级通知。
@@ -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<Map<String, Object>> 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<?))
+                order by e.created_at limit ?
+                """, properties.getMaxAttempts(), properties.getBatchSize());
+        for (Map<String, Object> event : events) {
+            process(event);
+        }
+    }
+
+    private void process(Map<String, Object> 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<Map<String, Object>> 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<String, Object> alarm = alarms.get(0);
+            MessagePriority priority = priority(String.valueOf(alarm.get("severity")));
+            Map<String, Object> 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);
+        };
+    }
+}
@@ -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));
+    }
+}
@@ -0,0 +1,6 @@
+/**
+ * HTTP and event-consumer adapters for the notification module.
+ *
+ * <p>No public endpoint is exposed in the initial skeleton.</p>
+ */
+package com.ai.trackwalker.notification.api;
@@ -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<MessageChannel> channels = policyResolver.resolve(command);
+        List<RecipientResolver.ResolvedRecipient> 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();
+    }
+}
@@ -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<Map<String, Object>> 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<String, Object> 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);
+    }
+}
@@ -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<String, Object> variables) {
+        List<Map<String, Object>> 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<String, Object> 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<String, Object> schema, Map<String, Object> variables) {
+        Object required = schema.get("required");
+        if (!(required instanceof List<?> names)) {
+            return;
+        }
+        List<String> 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<String, Object> variables) {
+        String result = template;
+        for (Map.Entry<String, Object> entry : variables.entrySet()) {
+            result = result.replace("{{" + entry.getKey() + "}}", String.valueOf(entry.getValue()));
+        }
+        return result.replaceAll("\\{\\{[^}]+}}", "-");
+    }
+
+    public record RenderedMessage(int version, String title, String body) {
+    }
+}
@@ -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);
+}
@@ -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<MessageChannel> resolve(SendNotificationCommand command) {
+        if (!command.requestedChannels().isEmpty()) {
+            return command.requestedChannels().stream().distinct().toList();
+        }
+        List<Map<String, Object>> 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<String, Object> 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();
+    }
+}
@@ -0,0 +1,10 @@
+package com.ai.trackwalker.notification.application;
+
+import java.time.Instant;
+
+public record NotificationSubmission(
+        String messageId,
+        String status,
+        Instant acceptedAt
+) {
+}
@@ -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<ResolvedRecipient> resolve(SendNotificationCommand.Audience audience) {
+        if (audience == null) {
+            return List.of();
+        }
+        Map<String, ResolvedRecipient> 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<Map<String, Object>> queryUsersByRoles(List<String> roleCodes, List<String> 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<Object> 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<Map<String, Object>> queryUsersByIds(List<String> 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<String, ResolvedRecipient> target, List<Map<String, Object>> rows) {
+        for (Map<String, Object> 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) {
+    }
+}
@@ -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<String, Object> variables,
+        List<MessageChannel> 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<String> userIds,
+            List<String> organizationIds,
+            List<String> 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);
+        }
+    }
+}
@@ -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<String, Object> templateParameters,
+        String title,
+        String body,
+        Map<String, Object> metadata
+) {
+    public ChannelSendCommand {
+        templateParameters = templateParameters == null ? Map.of() : Map.copyOf(templateParameters);
+        metadata = metadata == null ? Map.of() : Map.copyOf(metadata);
+    }
+}
@@ -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<String, Object> receipt
+) {
+    public ChannelSendResult {
+        receipt = receipt == null ? Map.of() : Map.copyOf(receipt);
+    }
+}
@@ -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<MessageChannel, NotificationChannelSender> senders;
+
+    public ChannelSenderRegistry(List<NotificationChannelSender> strategies) {
+        Map<MessageChannel, NotificationChannelSender> 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<NotificationChannelSender> find(MessageChannel channel) {
+        return Optional.ofNullable(senders.get(channel));
+    }
+}
@@ -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);
+}
@@ -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())
+        );
+    }
+}
@@ -0,0 +1,2 @@
+/** Local in-app inbox channel adapter. */
+package com.ai.trackwalker.notification.channel.inapp;
@@ -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);
+    }
+}
@@ -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()
+                )
+        );
+    }
+}
@@ -0,0 +1,2 @@
+/** Tencent Cloud Simple Email Service adapter. */
+package com.ai.trackwalker.notification.channel.tencent.ses;
@@ -0,0 +1,2 @@
+/** Tencent Cloud SMS adapter. */
+package com.ai.trackwalker.notification.channel.tencent.sms;
@@ -0,0 +1,2 @@
+/** Tencent Cloud Voice Message Service adapter. */
+package com.ai.trackwalker.notification.channel.tencent.vms;
@@ -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
+}
@@ -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
+}
@@ -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
+}
@@ -0,0 +1,8 @@
+package com.ai.trackwalker.notification.domain;
+
+public enum MessagePriority {
+    P1_CRITICAL,
+    P2_HIGH,
+    P3_NORMAL,
+    P4_LOW
+}
@@ -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<RecipientResolver.ResolvedRecipient> recipients,
+            List<MessageChannel> channels,
+            MessageTemplateRenderer.RenderedMessage rendered
+    ) {
+        List<Map<String, Object>> 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<String, Object> 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<String, Object> 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) {
+    }
+}
@@ -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<String, Object> delivery : claim()) {
+            send(delivery);
+        }
+    }
+
+    private void expireDeliveries() {
+        List<String> 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<Map<String, Object>> 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<String, Object> 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<String, Object> 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;
+    }
+}
@@ -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;
+    }
+}
@@ -0,0 +1,4 @@
+/**
+ * Persistence, event-bus, provider routing, scheduling, and callback infrastructure.
+ */
+package com.ai.trackwalker.notification.infrastructure;
@@ -0,0 +1,9 @@
+/**
+ * Unified notification module.
+ *
+ * <p>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.</p>
+ */
+package com.ai.trackwalker.notification;
@@ -0,0 +1,7 @@
+/**
+ * Isolated verification-code use cases.
+ *
+ * <p>This package may reuse channel clients but must not send codes through the ordinary
+ * notification facade or persist plaintext verification codes.</p>
+ */
+package com.ai.trackwalker.notification.verification;
@@ -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}
@@ -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';
@@ -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()
+        );
+    }
+}
@@ -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<MessageChannel> channels = List.of(MessageChannel.IN_APP, MessageChannel.SMS);
+        List<RecipientResolver.ResolvedRecipient> 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<MessageChannel> resolve(SendNotificationCommand ignored) {
+                return channels;
+            }
+        };
+        RecipientResolver recipients = new RecipientResolver(null) {
+            @Override
+            public List<ResolvedRecipient> resolve(SendNotificationCommand.Audience ignored) {
+                return resolved;
+            }
+        };
+        MessageTemplateRenderer templates = new MessageTemplateRenderer(null) {
+            @Override
+            public RenderedMessage render(String code, Integer version, Map<String, Object> variables) {
+                return rendered;
+            }
+        };
+        JdbcMessageStore store = new JdbcMessageStore(null) {
+            @Override
+            public StoredMessage create(
+                    SendNotificationCommand ignored,
+                    List<RecipientResolver.ResolvedRecipient> ignoredRecipients,
+                    List<MessageChannel> 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"
+        );
+    }
+}
@@ -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<Map<String, Object>> 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\"]}"
+                ));
+            }
+        };
+    }
+}
@@ -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<Map<String, Object>> 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<MessageChannel> 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"
+        );
+    }
+}