feat(msg): 接入消息中心——NoticeUserMsgNotifier(发布/转发/催办→UserMsg WS 实时推送,afterCommit 容错+分片+幂等)+ SDK 门面客户端 + 用户目录归一(09-19 二期验收 9/9)
build / build-push (push) Successful in 57s

This commit is contained in:
chenxing
2026-09-19 18:55:17 +08:00
parent a5204680e6
commit 5db629cfdb
12 changed files with 474 additions and 9 deletions
@@ -89,6 +89,9 @@ public class NoticeService implements INoticeService {
@Autowired
private com.yunhai.notice.service.flow.NoticeFlowGuard noticeFlowGuard;
@Autowired
private com.yunhai.notice.service.msg.NoticeUserMsgNotifier noticeUserMsgNotifier;
private static final int PUBLISHED = NoticeStatusEnum.PUBLISHED.getCode();
private static final int UNPUBLISHED = NoticeStatusEnum.UNPUBLISHED.getCode();
private static final int AUDIT_APPROVING = com.yunhai.notice.enums.FlowAuditStatusEnum.APPROVING.getCode();
@@ -161,6 +164,9 @@ public class NoticeService implements INoticeService {
} else {
// 立即发布且开启短信时预留短信发送(真实通道后续接入)
trySendSms(entity);
// 站内消息推送(afterCommit,指定人员;ALL 全员无法枚举一期不推,见 NoticeUserMsgNotifier)
noticeUserMsgNotifier.notifyPublishedAfterCommit(entity,
parseReceiveAccounts(entity.getReceiveList()));
}
return entity.getId();
}
@@ -191,6 +197,7 @@ public class NoticeService implements INoticeService {
Objects.requireNonNull(id, "通知ID不得为空");
Notice entity = noticeRepository.findById(id)
.orElseThrow(() -> new BizException(ErrorCode.FORBIDDEN, "通知不存在: " + id));
boolean wasPublished = entity.getStatus() == PUBLISHED;
// 审批中锁定:防审批做一半被抽走(删除需先撤回);驳回/撤回态允许编辑后重提
if (entity.getAuditStatus() != null && entity.getAuditStatus() == AUDIT_APPROVING) {
throw new BizException(ErrorCode.FORBIDDEN, "审批中不可编辑,请先撤回");
@@ -247,9 +254,16 @@ public class NoticeService implements INoticeService {
if (needAudit) {
// 编辑重提:复用同一 bizId(notice.id)→ flow-center R4 语义重启推进,轨迹连续
noticeFlowService.submitAfterCommit(entity.getId());
} else if (wasFlowBound) {
// 由审核轨道转直发:审核期从未发布过,此刻才真正可见——补发布后置(sms_sent_at 原子认领防重发)
trySendSms(entity);
} else {
if (wasFlowBound && !wasPublished) {
// 由审核轨道转直发:审核期从未发布过,此刻才真正可见——补发布后置(sms_sent_at 原子认领防重发)
trySendSms(entity);
}
if (!wasPublished) {
// 首次从非发布态转直发:推送站内消息(已发布通知的普通编辑不重复打扰)
noticeUserMsgNotifier.notifyPublishedAfterCommit(entity,
parseReceiveAccounts(entity.getReceiveList()));
}
}
}
@@ -364,6 +378,9 @@ public class NoticeService implements INoticeService {
noticeFlowService.submitAfterCommit(entity.getId());
} else {
trySendSms(entity);
// 转发直发后推送站内消息(与 publish 同口径)
noticeUserMsgNotifier.notifyPublishedAfterCommit(entity,
parseReceiveAccounts(entity.getReceiveList()));
}
return entity.getId();
}
@@ -783,7 +800,8 @@ public class NoticeService implements INoticeService {
.filter(a -> !readAccounts.contains(a))
.collect(Collectors.toList());
log.info("[expedite] 通知 {} 催收未读 {} 人: {}", serid, unreadAccounts.size(), unreadAccounts);
// TODO: 接入 yh-platform 消息推送后,对 unreadAccounts 发送催办消息
// 接入消息中心:对未读接收人推送催办站内消息(推送失败仅记日志,不阻断催收)
noticeUserMsgNotifier.notifyExpedited(notice, unreadAccounts);
}
// ============================ 统计 ============================
@@ -0,0 +1,141 @@
package com.yunhai.notice.service.msg;
import com.yunhai.notice.entity.Notice;
import com.yunhai.platform.openapi.msg.model.SendMsgRequest;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.util.StringUtils;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;
/**
* 通知公告 → 消息中心推送器(方案《消息中心建设方案》§6.1 试点生产方)。<br>
* 接入点:publish/forward 直发分支、edit 转直发(从未发布到发布)、expedite 催办未读。
* 先审后发(ALL + 绑定流程)审核通过才可见,但该路径接收人恒为 ALL 无法枚举——一期不推
* (催办入口可对已发布通知补推未读提醒)。<br>
* 容错语义(对齐 PartyLearn 范式):发送失败仅记日志,绝不阻断通知主流程;
* msgNo 幂等键 "ntc-{action}-{id}-{yyMMddHHmmss}"——同一秒内重试安全,跨秒重复动作视为再次提醒。
*/
@Slf4j
@Component
public class NoticeUserMsgNotifier {
/** SDK 单次接收人上限(oapi 契约:去重后 ≤1000,超出报错) */
public static final int MAX_RECV_PER_SEND = 1000;
private static final DateTimeFormatter MSG_NO_TS = DateTimeFormatter.ofPattern("yyMMddHHmmss");
private final UserMsgClient userMsgClient;
private final UserDirectory userDirectory;
private final NoticeUserMsgProperties properties;
public NoticeUserMsgNotifier(UserMsgClient userMsgClient, UserDirectory userDirectory,
NoticeUserMsgProperties properties) {
this.userMsgClient = userMsgClient;
this.userDirectory = userDirectory;
this.properties = properties;
}
/**
* 发布/转发/编辑转直发后推送(在发送事务内调用:注册 afterCommit,
* 避免通知主体尚未提交时收件端先收到消息)。
*
* @param accounts 接收人账号串列表(数字ID或登录名;ALL 通知为空列表 → 不推)
*/
public void notifyPublishedAfterCommit(Notice notice, List<String> accounts) {
if (!properties.isPushEnabled()) {
return;
}
if (TransactionSynchronizationManager.isSynchronizationActive()) {
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
@Override
public void afterCommit() {
doNotify(notice, accounts, "pub", "notice");
}
});
} else {
doNotify(notice, accounts, "pub", "notice");
}
}
/** 催办即时推送(expedite 无事务上下文,读侧已查完库) */
public void notifyExpedited(Notice notice, List<String> unreadAccounts) {
if (!properties.isPushEnabled()) {
return;
}
doNotify(notice, unreadAccounts, "exp", "notice.expedite");
}
private void doNotify(Notice notice, List<String> accounts, String action, String bizType) {
if (accounts == null || accounts.isEmpty()) {
log.info("[usermsg] 通知 {} 无可枚举接收人(ALL 全员),跳过站内消息", notice.getId());
return;
}
try {
Set<Integer> userIds = new LinkedHashSet<>();
for (String account : accounts) {
Integer userId = userDirectory.toUserId(account);
if (userId == null) {
log.warn("[usermsg] 接收人无法解析为用户ID,跳过 notice={} account={}", notice.getId(), account);
} else {
userIds.add(userId);
}
}
if (userIds.isEmpty()) {
log.warn("[usermsg] 通知 {} 接收人全部无法解析,未推送", notice.getId());
return;
}
List<Integer> all = new ArrayList<>(userIds);
int chunks = (all.size() + MAX_RECV_PER_SEND - 1) / MAX_RECV_PER_SEND;
for (int i = 0; i < chunks; i++) {
List<Integer> part = all.subList(i * MAX_RECV_PER_SEND, Math.min((i + 1) * MAX_RECV_PER_SEND, all.size()));
try {
userMsgClient.sendUserMsg(buildRequest(notice, action, bizType, part, chunks > 1 ? i + 1 : 0));
} catch (Exception e) {
log.warn("[usermsg] 站内消息发送失败(不影响通知主流程) notice={} part={}/{}:{}",
notice.getId(), i + 1, chunks, e.toString());
}
}
log.info("[usermsg] 通知 {} 站内消息已推送 {} 人(action={})", notice.getId(), all.size(), action);
} catch (Exception e) {
log.warn("[usermsg] 站内消息推送异常(不影响通知主流程) notice={}:{}", notice.getId(), e.toString());
}
}
private SendMsgRequest buildRequest(Notice notice, String action, String bizType,
List<Integer> userIds, int chunkIndex) {
SendMsgRequest request = new SendMsgRequest();
request.setUserIds(userIds);
request.setTitle("通知公告");
request.setContent(buildContent(notice));
String msgNo = "ntc-" + action + "-" + notice.getId() + "-"
+ MSG_NO_TS.format(LocalDateTime.now())
+ (chunkIndex > 0 ? "-" + chunkIndex : "");
request.setMsgNo(msgNo);
request.setBizType(bizType);
request.setBizId(String.valueOf(notice.getId()));
// 0系统通知;跳转 Notice 前端详情页(type=1 打开即标读),UserHome 侧按 bizType=notice 基址拼绝对地址
request.setMsgType(0);
request.setJumpType(1);
request.setJumpPath("/list/detail?serid=" + notice.getId() + "&type=1");
request.setSchId(notice.getSchId());
return request;
}
/** 消息文案:对齐旧版"您有一条来自xx老师的xx通知"口吻 */
private String buildContent(Notice notice) {
String publisher = StringUtils.hasText(notice.getPublisherName()) ? notice.getPublisherName() : "管理员";
String title = StringUtils.hasText(notice.getTitle()) ? notice.getTitle() : "";
if (title.length() > 120) {
title = title.substring(0, 120) + "…";
}
return "您有一条来自「" + publisher + "」的通知《" + title + "》,请及时查阅。";
}
}
@@ -0,0 +1,19 @@
package com.yunhai.notice.service.msg;
import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;
/**
* 通知公告 → 消息中心(UserMsg)推送配置。<br>
* push-enabled 默认关闭:UserMsg 未部署的环境(本地无配置/生产上线前)不产生空转告警——
* 上线 UserMsg 后在环境配置显式打开(对齐 flow-center.usermsg 空转的反面教材)。
*/
@Data
@Component
@ConfigurationProperties(prefix = "notice.usermsg")
public class NoticeUserMsgProperties {
/** 是否向 UserMsg 推送站内消息(发布/转发/编辑转直发/催办) */
private boolean pushEnabled = false;
}
@@ -0,0 +1,20 @@
package com.yunhai.notice.service.msg;
import com.yunhai.platform.openapi.msg.YschMsgOpenApi;
import com.yunhai.platform.openapi.msg.model.SendMsgRequest;
import org.springframework.stereotype.Component;
/**
* SDK 实现:委托 yh-openapi-client 的 msg 域门面(服务目录 key={@code usermsg},
* 地址解析 targets &gt; yh.open.services.usermsg &gt; 枚举 base-host 兜底;签名/信封全由 SDK 承载)。
* 门面由 yh-openapi-spring-boot-starter 自动装配;未装配时调用抛 IllegalStateException
* (推送方按容错语义捕获记日志,不阻断主流程)。
*/
@Component
public class SdkUserMsgClient implements UserMsgClient {
@Override
public void sendUserMsg(SendMsgRequest request) {
YschMsgOpenApi.sendUserMsg(request);
}
}
@@ -0,0 +1,38 @@
package com.yunhai.notice.service.msg;
import com.yunhai.platform.openapi.base.YschBaseOpenApi;
import com.yunhai.platform.utils.model.BoaUser;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;
/**
* 用户中心实现:纯数字视为 userId 直接收敛;其余按登录名查用户中心取 id
* (对齐 NoticeService.getReadDetailGrouped 的既有归一化语义)。
*/
@Slf4j
@Component
public class UserCenterUserDirectory implements UserDirectory {
@Override
public Integer toUserId(String account) {
if (!StringUtils.hasText(account)) {
return null;
}
String trimmed = account.trim();
if (trimmed.chars().allMatch(Character::isDigit)) {
try {
return Integer.valueOf(trimmed);
} catch (NumberFormatException e) {
return null;
}
}
try {
BoaUser user = YschBaseOpenApi.getUserInfo(trimmed);
return user == null ? null : user.getId();
} catch (Exception e) {
log.warn("[usermsg] 登录名换 userId 失败 account={}:{}", trimmed, e.toString());
return null;
}
}
}
@@ -0,0 +1,16 @@
package com.yunhai.notice.service.msg;
/**
* 接收人账号 → 数字用户ID 归一接口(测试缝)。<br>
* 新数据 receiveList 存数字用户ID(组织树叶 id,UserBase userId);
* 历史数据可能存登录名——登录名经用户中心换 userId。
*/
public interface UserDirectory {
/**
* 归一为数字用户ID:纯数字直接解析;否则按登录名查用户中心。
*
* @return 数字用户ID;解析失败返回 null(调用方跳过该人)
*/
Integer toUserId(String account);
}
@@ -0,0 +1,13 @@
package com.yunhai.notice.service.msg;
import com.yunhai.platform.openapi.msg.model.SendMsgRequest;
/**
* 消息中心发送客户端接口(测试缝):生产实现委托 SDK 门面 YschMsgOpenApi,
* 单测可 mock 以验证发送参数与容错语义。
*/
public interface UserMsgClient {
/** POST /oapi/msg/sendUserMsg(应用签名;同 appKey+msgNo 幂等返回原消息 id) */
void sendUserMsg(SendMsgRequest request);
}