chenxing/UserBaseService
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

推送订阅功能

Browse Source
This commit is contained in:
chenxing
2026-08-14 09:03:17 +08:00
parent 0233ecb525
commit e55c2cb209
15 changed files with 1154 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/com/yunhai/userbase/controller/api/AppSettingsController.java
+13
View File
@@ -14,6 +14,8 @@ import com.yunhai.userbase.model.application.ApplicationInfoQuery;
import com.yunhai.userbase.model.application.ApplicationInfoRequest;
import com.yunhai.userbase.model.application.ApplicationInfoSortVo;
import com.yunhai.userbase.model.application.ApplicationInfoVo;
import com.yunhai.userbase.model.application.PushSettingRequest;
import com.yunhai.userbase.model.application.PushSettingVo;
import com.yunhai.userbase.model.appmodule.AppModuleQuery;
import com.yunhai.userbase.model.appmodule.AppModuleRequest;
import com.yunhai.userbase.model.appmodule.AppModuleVo;
@@ -162,4 +164,15 @@ public class AppSettingsController {
public List<AppModuleVo> moduleList(AppModuleQuery query) {
return appModuleService.list(query);
}
@OperationLog("保存应用推送设置")
@PostMapping("SavePushSetting/{id}")
public void savePushSetting(@PathVariable Integer id, @RequestBody PushSettingRequest request) {
applicationInfoService.savePushSetting(id, request);
}
@GetMapping("GetPushSetting/{id}")
public PushSettingVo getPushSetting(@PathVariable Integer id) {
return applicationInfoService.getPushSetting(id);
}
}
src/main/java/com/yunhai/userbase/controller/api/PushSubscriptionController.java
+60
View File
@@ -0,0 +1,60 @@
package com.yunhai.userbase.controller.api;
import com.yunhai.userbase.config.Config;
import com.yunhai.userbase.model.BasePageQuery;
import com.yunhai.userbase.model.BasePageResult;
import com.yunhai.userbase.model.push.PushRecordEntity;
import com.yunhai.userbase.model.push.PushRecordQuery;
import com.yunhai.userbase.model.push.PushTestRequest;
import com.yunhai.userbase.service.push.PushDispatcher;
import com.yunhai.userbase.service.push.PushRecordService;
import com.yunhai.userbase.utils.PageUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Sort;
import org.springframework.data.web.SortDefault;
import org.springframework.web.bind.annotation.*;
import java.util.LinkedHashMap;
import java.util.Map;
/**
* 推送订阅控制器：测试推送 + 推送日志记录 / 重试。
*
* @author chenx
* @date 2026/8/14
*/
@RestController
@RequestMapping(value = Config.API_PATH + "/push/")
public class PushSubscriptionController {
@Autowired
private PushDispatcher pushDispatcher;
@Autowired
private PushRecordService pushRecordService;
/** 测试推送：手动触发一次推送（投递 MQ），并返回请求预览（URL/头/体，密钥不外露）。 */
@PostMapping("Test")
public Map<String, Object> test(@RequestBody PushTestRequest request) {
pushDispatcher.dispatchManual(request.getAppId(), request.getInterfacePath(), request.getTargetIds());
Map<String, Object> preview = pushDispatcher.previewManual(
request.getAppId(), request.getInterfacePath(), request.getTargetIds());
Map<String, Object> result = new LinkedHashMap<>();
result.put("sent", true);
result.put("message", "已投递推送队列，可在「推送日志」查看实际响应与耗时");
result.put("preview", preview);
return result;
}
/** 推送日志列表。 */
@GetMapping("RecordList")
public BasePageResult<PushRecordEntity> recordList(PushRecordQuery query, BasePageQuery pageQuery,
@SortDefault(sort = "createdAt", direction = Sort.Direction.DESC) Sort sort) {
return PageUtils.toResult(pushRecordService.query(query, PageUtils.toPageable(pageQuery, sort)));
}
/** 手动重试一条失败的推送记录。 */
@PostMapping("Retry/{id}")
public void retry(@PathVariable Long id) {
pushRecordService.retry(id);
}
}
src/main/java/com/yunhai/userbase/entity/ApplicationInfo.java
+39
View File
@@ -104,4 +104,43 @@ public class ApplicationInfo extends BaseEntity {
@Column(name = "dingtalk_url")
private String dingtalkUrl;
// ===================== 推送订阅配置（文档《弘育基础平台数据同步接口 2》） =====================
/**
* 推送外部接口地址（不含固定后缀），如 http://host/notice/api/base/ 。
* 完整调用地址 = pushBaseUrl + 接口路径（如 group/AddGroup）。
*/
@Column(name = "push_base_url")
private String pushBaseUrl;
/**
* 推送总开关：0关 1开（默认 0，自动推送默认不触发）。
*/
@Column(name = "push_enabled", nullable = false)
private Integer pushEnabled = 0;
/**
* 启用的接口路径，逗号分隔（如 group/AddGroup,user/AddUser）；空=不推送任何接口。
*/
@Column(name = "push_event_types")
private String pushEventTypes;
/**
* 关联学校组 id，逗号分隔；空=全局（监听所有学校变化）。
*/
@Column(name = "push_school_ids")
private String pushSchoolIds;
/**
* 推送角色 key，逗号分隔；空=全部角色。
*/
@Column(name = "push_role_ids")
private String pushRoleIds;
/**
* 推送签名密钥；空则复用 appSecret。
*/
@Column(name = "push_secret")
private String pushSecret;
}
src/main/java/com/yunhai/userbase/repository/ApplicationInfoRepository.java
+3
View File
@@ -42,6 +42,9 @@ public interface ApplicationInfoRepository extends JpaRepository<ApplicationInf
*/
ApplicationInfo findByAppKey(Integer appKey);
/** 查询所有开启推送的应用（自动推送遍历订阅方用）。 */
List<ApplicationInfo> findByPushEnabled(Integer pushEnabled);
/**
* 按 id 更新排序号（同一应用类型内排序使用）。<br>
*
src/main/java/com/yunhai/userbase/service/IApplicationInfoService.java
+19
View File
@@ -5,6 +5,8 @@ import com.yunhai.userbase.model.application.ApplicationInfoQuery;
import com.yunhai.userbase.model.application.ApplicationInfoRequest;
import com.yunhai.userbase.model.application.ApplicationInfoSortVo;
import com.yunhai.userbase.model.application.ApplicationInfoVo;
import com.yunhai.userbase.model.application.PushSettingRequest;
import com.yunhai.userbase.model.application.PushSettingVo;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.Pageable;
@@ -82,4 +84,21 @@ public interface IApplicationInfoService {
* @param ids 同一应用类型下的应用ID有序列表
*/
void sort(List<Integer> ids);
/**
* 保存应用推送设置（外部接口地址 / 总开关 / 启用的接口 / 学校范围 / 角色范围 / 推送密钥）。<br>
* 仅更新推送相关字段，写后同步更新 appInfo 缓存。
*
* @param id 应用ID
* @param request 推送设置（各字段可选，null 表示不修改）
*/
ApplicationInfo savePushSetting(Integer id, PushSettingRequest request);
/**
* 获取应用推送设置回显 VO。
*
* @param id 应用ID
* @return 推送设置 VO（仅含推送相关字段）
*/
PushSettingVo getPushSetting(Integer id);
}
src/main/java/com/yunhai/userbase/service/impl/ApplicationInfoService.java
+50
View File
@@ -22,6 +22,8 @@ import com.yunhai.userbase.model.application.ApplicationInfoQuery;
import com.yunhai.userbase.model.application.ApplicationInfoRequest;
import com.yunhai.userbase.model.application.ApplicationInfoSortVo;
import com.yunhai.userbase.model.application.ApplicationInfoVo;
import com.yunhai.userbase.model.application.PushSettingRequest;
import com.yunhai.userbase.model.application.PushSettingVo;
import com.yunhai.userbase.event.ApplicationInfoCacheChangedEvent;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
@@ -266,4 +268,52 @@ public class ApplicationInfoService implements IApplicationInfoService {
public List<ApplicationInfoSortVo> findListByCategoryId(Integer categoryId) {
return applicationInfoRepository.findSortVoByCategoryId(categoryId);
}
@Override
@Transactional
@CachePut(value = "appInfo", key = "#id")
public ApplicationInfo savePushSetting(Integer id, PushSettingRequest request) {
Objects.requireNonNull(id, "应用ID不能为空");
Objects.requireNonNull(request, "推送设置入参不能为空");
ApplicationInfo entity = applicationInfoRepository.findById(id)
.orElseThrow(() -> new RuntimeException("应用不存在: " + id));
if (request.getPushBaseUrl() != null) {
entity.setPushBaseUrl(request.getPushBaseUrl());
}
if (request.getPushEnabled() != null) {
entity.setPushEnabled(request.getPushEnabled());
}
if (request.getPushEventTypes() != null) {
entity.setPushEventTypes(request.getPushEventTypes());
}
if (request.getPushSchoolIds() != null) {
entity.setPushSchoolIds(request.getPushSchoolIds());
}
if (request.getPushRoleIds() != null) {
entity.setPushRoleIds(request.getPushRoleIds());
}
// pushSecret：null=不修改；空串=清空（后续复用 appSecret）
if (request.getPushSecret() != null) {
entity.setPushSecret(request.getPushSecret());
}
return applicationInfoRepository.save(entity);
}
@Override
public PushSettingVo getPushSetting(Integer id) {
Objects.requireNonNull(id, "应用ID不能为空");
ApplicationInfo entity = applicationInfoRepository.findById(id)
.orElseThrow(() -> new RuntimeException("应用不存在: " + id));
PushSettingVo vo = new PushSettingVo();
vo.setId(entity.getId());
vo.setAppKey(entity.getAppKey());
vo.setName(entity.getName());
vo.setPushBaseUrl(entity.getPushBaseUrl());
vo.setPushEnabled(entity.getPushEnabled());
vo.setPushEventTypes(entity.getPushEventTypes());
vo.setPushSchoolIds(entity.getPushSchoolIds());
vo.setPushRoleIds(entity.getPushRoleIds());
vo.setPushSecret(entity.getPushSecret());
return vo;
}
}
src/main/java/com/yunhai/userbase/service/impl/SysGroupService.java
+33 -1
View File
@@ -5,6 +5,9 @@ import com.yunhai.platform.utils.exception.DException;
import com.yunhai.platform.utils.model.BoaGroup;
import com.yunhai.userbase.enums.GroupTypeEnum;
import com.yunhai.userbase.enums.IEnum;
import com.yunhai.userbase.event.DataSyncEvent;
import com.yunhai.userbase.model.push.PushInterfaceType;
import org.springframework.context.ApplicationEventPublisher;
import com.yunhai.userbase.model.sysgroup.SysGroupBatchRequest;
import com.yunhai.userbase.model.sysgroup.SysGroupMoveRequest;
import com.yunhai.userbase.model.sysgroup.SysGroupQuery;
@@ -23,6 +26,7 @@ import org.springframework.util.StringUtils;
import javax.transaction.Transactional;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
@@ -55,6 +59,9 @@ public class SysGroupService implements ISysGroupService {
@Autowired
private SysGroupModulePermRepository sysGroupModulePermRepository;
@Autowired
private ApplicationEventPublisher eventPublisher;
// ===================== 基础增删改（含根节点保护） =====================
@Override
@@ -75,7 +82,8 @@ public class SysGroupService implements ISysGroupService {
entity.setParentId(parentId);
entity.setType(typeEnum);
entity.setSort(nextSort(parentId));
sysGroupRepository.save(entity);
SysGroup saved = sysGroupRepository.save(entity);
publishGroupEvent(PushInterfaceType.ADD_GROUP, saved.getId());
}
@Override
@@ -96,6 +104,7 @@ public class SysGroupService implements ISysGroupService {
}
// 注：移动请走 move 接口，edit 不改 parentId，避免与移动逻辑混淆
sysGroupRepository.save(entity);
publishGroupEvent(PushInterfaceType.UPDATE_GROUP, id);
}
@Override
@@ -117,6 +126,7 @@ public class SysGroupService implements ISysGroupService {
sysGroupAppRelRepository.deleteByGroupIdBulk(id);
sysGroupModulePermRepository.deleteByGroupIdBulk(id);
sysGroupRepository.deleteById(id);
publishGroupEvent(PushInterfaceType.DELETE_GROUP, id);
}
@Override
@@ -326,6 +336,8 @@ public class SysGroupService implements ISysGroupService {
batch.add(g);
}
sysGroupRepository.saveAll(batch);
publishGroupEvent(PushInterfaceType.ADD_GROUP,
batch.stream().map(SysGroup::getId).collect(Collectors.toList()));
}
// ===================== 子组排序 =====================
@@ -392,6 +404,7 @@ public class SysGroupService implements ISysGroupService {
moveEntity.setParentId(pid);
moveEntity.setSort(nextSort(pid));
sysGroupRepository.save(moveEntity);
publishGroupEvent(PushInterfaceType.UPDATE_GROUP, moveId);
}
/**
@@ -522,4 +535,23 @@ public class SysGroupService implements ISysGroupService {
vo.setType(entity.getType() == null ? null : entity.getType().getKey());
return vo;
}
// ===================== 推送事件（AFTER_COMMIT 由 PushDispatcher 消费，push_enabled 默认 0 不触发） =====================
private void publishGroupEvent(PushInterfaceType type, Integer groupId) {
DataSyncEvent e = new DataSyncEvent();
e.setType(type);
e.setGroupIds(Collections.singletonList(groupId));
eventPublisher.publishEvent(e);
}
private void publishGroupEvent(PushInterfaceType type, List<Integer> groupIds) {
if (groupIds == null || groupIds.isEmpty()) {
return;
}
DataSyncEvent e = new DataSyncEvent();
e.setType(type);
e.setGroupIds(groupIds);
eventPublisher.publishEvent(e);
}
}
src/main/java/com/yunhai/userbase/service/impl/SysUserService.java
+38 -1
View File
@@ -18,6 +18,9 @@ import com.yunhai.userbase.repository.SysUserGroupRelRepository;
import com.yunhai.userbase.repository.SysUserRepository;
import com.yunhai.userbase.service.ISysGroupService;
import com.yunhai.userbase.service.ISysUserService;
import com.yunhai.userbase.event.DataSyncEvent;
import com.yunhai.userbase.model.push.PushInterfaceType;
import org.springframework.context.ApplicationEventPublisher;
import cn.afterturn.easypoi.excel.ExcelExportUtil;
import cn.afterturn.easypoi.excel.entity.ExportParams;
import cn.afterturn.easypoi.excel.entity.enmus.ExcelType;
@@ -79,6 +82,9 @@ public class SysUserService implements ISysUserService {
@Autowired
private SysUserAppRoleRelRepository sysUserAppRoleRelRepository;
@Autowired
private ApplicationEventPublisher eventPublisher;
@Autowired
private SysGroupRepository sysGroupRepository;
@@ -130,7 +136,8 @@ public class SysUserService implements ISysUserService {
// disabled 用枚举存储：不传时默认 NORMAL（正常/未禁用）
entity.setDisabled(request.getDisabled() != null ? resolveStatus(request.getDisabled()) : StatusEnum.NORMAL);
entity.setExpired(request.getExpired() != null ? request.getExpired() : false);
sysUserRepository.save(entity);
SysUser saved = sysUserRepository.save(entity);
publishUserEvent(PushInterfaceType.ADD_USER, saved.getId());
// 写入组用户关联
SysUserGroupRel rel = new SysUserGroupRel();
@@ -175,6 +182,7 @@ public class SysUserService implements ISysUserService {
entity.setExpired(request.getExpired());
}
sysUserRepository.save(entity);
publishUserEvent(PushInterfaceType.UPDATE_USER, id);
}
@Override
@@ -188,6 +196,7 @@ public class SysUserService implements ISysUserService {
// AI-FIX(角色授权): 删除用户时级联清理用户-业务角色绑定（逻辑外键无物理 FK，避免残留孤儿数据）
sysUserAppRoleRelRepository.deleteByUserIdBulk(id);
sysUserRepository.deleteById(id);
publishUserEvent(PushInterfaceType.DELETE_USER, id);
}
@Override
@@ -211,6 +220,7 @@ public class SysUserService implements ISysUserService {
throw new DException("存在无效用户");
}
List<Integer> deletedUserIds = new ArrayList<>();
for (Integer userId : userIds) {
// 删除当前组关联
sysUserGroupRelRepository.deleteByUserIdAndGroupId(userId, groupId);
@@ -218,8 +228,12 @@ public class SysUserService implements ISysUserService {
if (sysUserGroupRelRepository.countByUserId(userId) == 0) {
sysUserAppRoleRelRepository.deleteByUserIdBulk(userId);
sysUserRepository.deleteById(userId);
deletedUserIds.add(userId);
}
}
if (!deletedUserIds.isEmpty()) {
publishUserEvent(PushInterfaceType.DELETE_USER, deletedUserIds);
}
}
@Override
@@ -230,6 +244,7 @@ public class SysUserService implements ISysUserService {
// 实体 disabled 字段为 StatusEnum（0=正常/未禁用，1=禁用）：boolean 直接映射枚举
entity.setDisabled(disabled ? StatusEnum.DISABLED : StatusEnum.NORMAL);
sysUserRepository.save(entity);
publishUserEvent(PushInterfaceType.UPDATE_USER, id);
}
@Override
@@ -247,6 +262,8 @@ public class SysUserService implements ISysUserService {
user.setDisabled(targetStatus);
}
sysUserRepository.saveAll(users);
publishUserEvent(PushInterfaceType.UPDATE_USER,
users.stream().map(SysUser::getId).collect(Collectors.toList()));
}
@Override
@@ -362,6 +379,7 @@ public class SysUserService implements ISysUserService {
}
sysUserGroupRelRepository.saveAll(rels);
}
publishUserEvent(PushInterfaceType.UPDATE_USER, userIds);
}
@Override
@@ -936,4 +954,23 @@ public class SysUserService implements ISysUserService {
return null;
}
}
// ===================== 推送事件 =====================
private void publishUserEvent(PushInterfaceType type, Integer userId) {
DataSyncEvent e = new DataSyncEvent();
e.setType(type);
e.setUserIds(Collections.singletonList(userId));
eventPublisher.publishEvent(e);
}
private void publishUserEvent(PushInterfaceType type, List<Integer> userIds) {
if (userIds == null || userIds.isEmpty()) {
return;
}
DataSyncEvent e = new DataSyncEvent();
e.setType(type);
e.setUserIds(userIds);
eventPublisher.publishEvent(e);
}
}
src/main/java/com/yunhai/userbase/service/push/GroupScopeHelper.java
+60
View File
@@ -0,0 +1,60 @@
package com.yunhai.userbase.service.push;
import com.yunhai.userbase.enums.GroupTypeEnum;
import com.yunhai.userbase.entity.SysGroup;
import com.yunhai.userbase.repository.SysGroupRepository;
import com.yunhai.userbase.service.ISysGroupService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.Collections;
import java.util.List;
/**
* 学校范围辅助：把「学校组」语义落地为可计算的范围。<br><br>
* 语义（已与用户确认）：一个组属于学校 S ⇔ 该组是学校组 S 本身，或 S 的任意后代子节点。
*
* @author chenx
* @date 2026/8/14
*/
@Service
public class GroupScopeHelper {
private static final int ROOT_PARENT_ID = 0;
@Autowired
private SysGroupRepository sysGroupRepository;
@Autowired
private ISysGroupService sysGroupService;
/**
* 计算组所属学校范围 id：自身为学校组则取自身，否则沿 parentId 向上找最近的 SCHOOL_GROUP。
* 无学校祖先返回 null（属于教育云中心直属，全局）。
*/
public Integer resolveSchoolId(Integer groupId) {
if (groupId == null) {
return null;
}
SysGroup g = sysGroupRepository.findById(groupId).orElse(null);
while (g != null) {
if (g.getType() == GroupTypeEnum.SCHOOL_GROUP) {
return g.getId();
}
if (g.getParentId() == null || g.getParentId() == ROOT_PARENT_ID) {
break;
}
g = sysGroupRepository.findById(g.getParentId()).orElse(null);
}
return null;
}
/**
* 学校范围 = 学校组自身 + 全部子孙（复用 SysGroupService.listSelfAndDescendantIds）。
*/
public List<Integer> getSchoolScopeGroupIds(Integer schoolId) {
if (schoolId == null) {
return Collections.emptyList();
}
return sysGroupService.listSelfAndDescendantIds(schoolId);
}
}
src/main/java/com/yunhai/userbase/service/push/PushConsumer.java
+159
View File
@@ -0,0 +1,159 @@
package com.yunhai.userbase.service.push;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.rabbitmq.client.Channel;
import com.yunhai.userbase.config.PushMqConfig;
import com.yunhai.userbase.entity.ApplicationInfo;
import com.yunhai.userbase.model.push.PushRecordEntity;
import com.yunhai.userbase.model.push.PushTaskMessage;
import com.yunhai.userbase.repository.ApplicationInfoRepository;
import com.yunhai.userbase.repository.PushRecordRepository;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.http.client.SimpleClientHttpRequestFactory;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;
import java.io.IOException;
import java.util.Collections;
import java.util.Map;
/**
* 推送消费者（订阅方调用方）：<br>
* 从 push.main 串行消费，手动 ack；POST 给订阅方 Webhook，按文档契约带 t/sign 头；<br>
* 每次尝试均落 push_record；失败应用层自管重试（共 3 次），耗尽进死信队列。
*
* @author chenx
* @date 2026/8/14
*/
@Slf4j
@Service
public class PushConsumer {
@Autowired
private ApplicationInfoRepository applicationInfoRepository;
@Autowired
private PushRecordRepository pushRecordRepository;
@Autowired
@Qualifier("pushRabbitTemplate")
private RabbitTemplate pushRabbitTemplate;
@Autowired
private ObjectMapper objectMapper;
@RabbitListener(queues = PushMqConfig.QUEUE_MAIN, containerFactory = "pushListenerContainerFactory")
public void onMessage(PushTaskMessage msg, Channel channel,
@Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
long start = System.currentTimeMillis();
String url = null;
Integer httpStatus = null;
Integer respCode = null;
String respMsg = null;
boolean ok = false;
String appName = null;
try {
ApplicationInfo app = applicationInfoRepository.findByAppKey(msg.getAppKey());
if (app == null) {
respMsg = "订阅方不存在 appKey=" + msg.getAppKey();
} else {
appName = app.getName();
url = joinUrl(app.getPushBaseUrl(), msg.getInterfacePath());
HttpHeaders headers = new HttpHeaders();
headers.setContentType(MediaType.APPLICATION_JSON);
headers.set("t", msg.getT());
headers.set("sign", msg.getSign());
int readTimeout = msg.getInterfacePath().contains("Sync") ? 60000 : 10000;
SimpleClientHttpRequestFactory rf = new SimpleClientHttpRequestFactory();
rf.setConnectTimeout(5000);
rf.setReadTimeout(readTimeout);
RestTemplate rt = new RestTemplate(rf);
ResponseEntity<String> resp = rt.postForEntity(url, new HttpEntity<>(msg.getBody(), headers), String.class);
httpStatus = resp.getStatusCodeValue();
String body = resp.getBody();
try {
JsonNode node = objectMapper.readTree(body);
respCode = node.path("code").asInt();
respMsg = node.path("msg").asText();
} catch (Exception e) {
respMsg = body;
}
ok = (httpStatus != null && httpStatus == 200 && respCode != null && respCode == 0);
}
} catch (Exception e) {
respMsg = e.getMessage();
} finally {
long cost = System.currentTimeMillis() - start;
try {
PushRecordEntity rec = new PushRecordEntity();
rec.setSubscriberAppKey(msg.getAppKey());
rec.setAppName(appName);
rec.setInterfacePath(msg.getInterfacePath());
rec.setInterfaceLabel(msg.getInterfaceLabel());
rec.setRequestUrl(url == null ? "" : url);
rec.setPayload(msg.getBody());
rec.setSignT(msg.getT());
rec.setHttpStatus(httpStatus);
rec.setRespCode(respCode);
rec.setRespMsg(truncate(respMsg, 500));
rec.setSuccess(ok ? 1 : 0);
rec.setAttemptNo(msg.getAttemptNo());
rec.setCostMs((int) cost);
rec.setTriggerType(msg.getTriggerType());
pushRecordRepository.save(rec);
} catch (Exception dbErr) {
log.error("[推送] 落库失败", dbErr);
}
try {
if (ok) {
channel.basicAck(deliveryTag, false);
} else if (msg.getAttemptNo() < 3) {
PushTaskMessage retry = new PushTaskMessage(msg.getAppKey(), msg.getInterfacePath(),
msg.getInterfaceLabel(), msg.getBody(), msg.getT(), msg.getSign(),
msg.getAttemptNo() + 1, msg.getTriggerType());
pushRabbitTemplate.convertAndSend(PushMqConfig.EXCHANGE_TOPIC, msg.getInterfacePath(), retry);
channel.basicAck(deliveryTag, false);
} else {
// 重试耗尽：进死信队列归档
channel.basicNack(deliveryTag, false, false);
}
} catch (IOException io) {
log.error("[推送] ACK 失败", io);
}
}
}
private static String truncate(String s, int max) {
if (s == null) {
return null;
}
return s.length() > max ? s.substring(0, max) : s;
}
private static String joinUrl(String base, String path) {
if (base == null) {
return path;
}
String b = base.endsWith("/") ? base : base + "/";
String p = path.startsWith("/") ? path.substring(1) : path;
return b + p;
}
/** 供 {@link PushDispatcher} 共享的 URL 拼接（保持语义一致）。 */
static String joinUrlForTest(String base, String path) {
return joinUrl(base, path);
}
@SuppressWarnings("unused")
private static Map<String, Object> unused() {
return Collections.emptyMap();
}
}
src/main/java/com/yunhai/userbase/service/push/PushDispatcher.java
+328
View File
@@ -0,0 +1,328 @@
package com.yunhai.userbase.service.push;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.yunhai.platform.utils.exception.DException;
import com.yunhai.userbase.config.PushMqConfig;
import com.yunhai.userbase.entity.ApplicationInfo;
import com.yunhai.userbase.entity.SysGroup;
import com.yunhai.userbase.entity.SysUser;
import com.yunhai.userbase.entity.SysUserGroupRel;
import com.yunhai.userbase.event.DataSyncEvent;
import com.yunhai.userbase.model.push.PushInterfaceType;
import com.yunhai.userbase.model.push.PushTaskMessage;
import com.yunhai.userbase.repository.ApplicationInfoRepository;
import com.yunhai.userbase.repository.SysGroupRepository;
import com.yunhai.userbase.repository.SysUserGroupRelRepository;
import com.yunhai.userbase.repository.SysUserRepository;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;
import org.springframework.transaction.event.TransactionalEventListener;
import org.springframework.transaction.event.TransactionPhase;
import java.util.*;
import java.util.stream.Collectors;
/**
* 推送分发器（生产方）：<br>
* ① 自动触发：监听 {@link DataSyncEvent}（AFTER_COMMIT），遍历所有开启推送的订阅方，按配置过滤后投递 MQ；<br>
* ② 手动测试：{@link #dispatchManual} 供测试推送页调用。<br><br>
* 签名在此处由基础平台计算一次（HMAC-SHA256(业务JSON+t, secret)），消息携带 t/sign，消费方原样 POST。
*
* @author chenx
* @date 2026/8/14
*/
@Slf4j
@Service
public class PushDispatcher {
@Autowired
private ApplicationInfoRepository applicationInfoRepository;
@Autowired
private SysGroupRepository sysGroupRepository;
@Autowired
private SysUserRepository sysUserRepository;
@Autowired
private SysUserGroupRelRepository sysUserGroupRelRepository;
@Autowired
private GroupScopeHelper groupScopeHelper;
@Autowired
private PushPayloadAssembler pushPayloadAssembler;
@Autowired
private ObjectMapper objectMapper;
@Autowired
@Qualifier("pushRabbitTemplate")
private RabbitTemplate pushRabbitTemplate;
// ===================== 自动触发（AFTER_COMMIT，推送失败不影响主业务） =====================
@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
public void onDataSync(DataSyncEvent event) {
try {
PushInterfaceType type = event.getType();
if (type == null) {
return;
}
List<SysGroup> groups = (event.getGroupIds() != null && !event.getGroupIds().isEmpty())
? sysGroupRepository.findAllById(event.getGroupIds()) : Collections.emptyList();
List<SysUser> users = (event.getUserIds() != null && !event.getUserIds().isEmpty())
? sysUserRepository.findAllById(event.getUserIds()) : Collections.emptyList();
// 全量同步需要全部数据
if (type == PushInterfaceType.SYNC_GROUP) {
groups = sysGroupRepository.findAll();
}
if (type == PushInterfaceType.SYNC_USER) {
users = sysUserRepository.findAll();
}
for (ApplicationInfo app : applicationInfoRepository.findByPushEnabled(1)) {
dispatchToOne(app, type, groups, users, "auto");
}
} catch (Exception e) {
log.error("[推送] 分发失败（不影响主业务）", e);
}
}
// ===================== 手动测试 =====================
public void dispatchManual(Integer appId, String interfacePath, List<Integer> targetIds) {
ApplicationInfo app = applicationInfoRepository.findById(appId).orElse(null);
if (app == null) {
throw new DException("应用不存在");
}
PushInterfaceType type = PushInterfaceType.getByPath(interfacePath);
if (type == null) {
throw new DException("不支持的接口路径：" + interfacePath);
}
List<SysGroup> groups = Collections.emptyList();
List<SysUser> users = Collections.emptyList();
if (type == PushInterfaceType.SYNC_GROUP) {
groups = sysGroupRepository.findAll();
} else if (type == PushInterfaceType.SYNC_USER) {
users = sysUserRepository.findAll();
} else if (type.isGroup()) {
groups = (targetIds == null || targetIds.isEmpty())
? sysGroupRepository.findAll() : sysGroupRepository.findAllById(targetIds);
} else {
users = (targetIds == null || targetIds.isEmpty())
? sysUserRepository.findAll() : sysUserRepository.findAllById(targetIds);
}
dispatchToOne(app, type, groups, users, "manual");
}
/**
* 生成一次推送请求的预览（URL / 头 / 体），供测试推送页展示。<br>
* 签名在服务端用 secret 计算，前端不持有密钥；实际投递仍走 {@link #dispatchManual} → MQ。
*/
public Map<String, Object> previewManual(Integer appId, String interfacePath, List<Integer> targetIds) {
ApplicationInfo app = applicationInfoRepository.findById(appId).orElse(null);
if (app == null) {
throw new DException("应用不存在");
}
PushInterfaceType type = PushInterfaceType.getByPath(interfacePath);
if (type == null) {
throw new DException("不支持的接口路径：" + interfacePath);
}
List<SysGroup> groups = Collections.emptyList();
List<SysUser> users = Collections.emptyList();
if (type == PushInterfaceType.SYNC_GROUP) {
groups = sysGroupRepository.findAll();
} else if (type == PushInterfaceType.SYNC_USER) {
users = sysUserRepository.findAll();
} else if (type.isGroup()) {
groups = (targetIds == null || targetIds.isEmpty())
? sysGroupRepository.findAll() : sysGroupRepository.findAllById(targetIds);
} else {
users = (targetIds == null || targetIds.isEmpty())
? sysUserRepository.findAll() : sysUserRepository.findAllById(targetIds);
}
List<Object> bodies = buildBodies(type, groups, users);
if (bodies.isEmpty()) {
Map<String, Object> empty = new LinkedHashMap<>();
empty.put("url", joinUrl(app.getPushBaseUrl(), type.getPath()));
empty.put("method", "POST");
empty.put("headers", buildHeaders("", ""));
empty.put("body", "{}");
empty.put("batch", 0);
return empty;
}
try {
String body = objectMapper.writeValueAsString(bodies.get(0));
String t = String.valueOf(System.currentTimeMillis());
String secret = (app.getPushSecret() != null && !app.getPushSecret().isEmpty())
? app.getPushSecret() : app.getAppSecret();
String sign = PushSigner.sign(body, secret, t);
Map<String, Object> preview = new LinkedHashMap<>();
preview.put("url", joinUrl(app.getPushBaseUrl(), type.getPath()));
preview.put("method", "POST");
preview.put("headers", buildHeaders(t, sign));
preview.put("body", body);
preview.put("batch", bodies.size());
return preview;
} catch (Exception e) {
throw new DException("预览生成失败：" + e.getMessage());
}
}
private static Map<String, String> buildHeaders(String t, String sign) {
Map<String, String> headers = new LinkedHashMap<>();
headers.put("Content-Type", "application/json");
headers.put("t", t);
headers.put("sign", sign);
return headers;
}
// ===================== 核心 =====================
private void dispatchToOne(ApplicationInfo app, PushInterfaceType type,
List<SysGroup> groups, List<SysUser> users, String triggerType) {
if (app.getPushEnabled() == null || app.getPushEnabled() != 1) {
return;
}
if (!containsEventType(app.getPushEventTypes(), type.getPath())) {
return;
}
boolean isFull = (type == PushInterfaceType.SYNC_GROUP || type == PushInterfaceType.SYNC_USER);
List<SysGroup> fGroups = (type.isGroup() && !isFull) ? filterGroups(groups, app) : groups;
List<SysUser> fUsers = (type.isUser() && !isFull) ? filterUsers(users, app) : users;
// 增量事件无匹配则跳过（全量即使为空也照常投递，供订阅方清空）
if (!isFull) {
if (type.isGroup() && fGroups.isEmpty()) {
return;
}
if (type.isUser() && fUsers.isEmpty()) {
return;
}
}
for (Object bodyObj : buildBodies(type, fGroups, fUsers)) {
sendMessage(app, type, bodyObj, triggerType);
}
}
/** 按接口形状构建待发送 body（删除类接口 body 形态与增改不同）。 */
private List<Object> buildBodies(PushInterfaceType type, List<SysGroup> groups, List<SysUser> users) {
switch (type) {
case SYNC_GROUP:
return Collections.singletonList(pushPayloadAssembler.buildFullGroupTree());
case ADD_GROUP:
case UPDATE_GROUP:
return Collections.singletonList(pushPayloadAssembler.toGroupDtoList(groups));
case DELETE_GROUP:
// 文档 3.5：json 对象 {id}；每个组独立一条消息
return groups.stream()
.map(g -> Collections.singletonMap("id", g.getId()))
.collect(Collectors.toList());
case SYNC_USER:
return Collections.singletonList(pushPayloadAssembler.toUserDtoList(users));
case ADD_USER:
case UPDATE_USER:
return Collections.singletonList(pushPayloadAssembler.toUserDtoList(users));
case DELETE_USER:
// 文档 3.8：json 数组 [id,...]
return Collections.singletonList(users.stream().map(SysUser::getId).collect(Collectors.toList()));
default:
return Collections.emptyList();
}
}
private void sendMessage(ApplicationInfo app, PushInterfaceType type, Object bodyObj, String triggerType) {
try {
String body = objectMapper.writeValueAsString(bodyObj);
String t = String.valueOf(System.currentTimeMillis());
String secret = (app.getPushSecret() != null && !app.getPushSecret().isEmpty())
? app.getPushSecret() : app.getAppSecret();
String sign = PushSigner.sign(body, secret, t);
PushTaskMessage msg = new PushTaskMessage(app.getAppKey(), type.getPath(), type.getLabel(),
body, t, sign, 1, triggerType);
pushRabbitTemplate.convertAndSend(PushMqConfig.EXCHANGE_TOPIC, type.getPath(), msg);
log.info("[推送] 投递 appKey={} type={} trigger={}", app.getAppKey(), type.getPath(), triggerType);
} catch (Exception e) {
log.error("[推送] 消息投递失败 appKey={} type={}", app.getAppKey(), type.getPath(), e);
}
}
// ===================== 过滤 =====================
private List<SysGroup> filterGroups(List<SysGroup> groups, ApplicationInfo app) {
Set<Integer> schoolSet = parseIds(app.getPushSchoolIds());
if (schoolSet.isEmpty()) {
return groups;
}
return groups.stream()
.filter(g -> {
Integer sid = groupScopeHelper.resolveSchoolId(g.getId());
return sid != null && schoolSet.contains(sid);
})
.collect(Collectors.toList());
}
private List<SysUser> filterUsers(List<SysUser> users, ApplicationInfo app) {
Set<Integer> schoolSet = parseIds(app.getPushSchoolIds());
Set<Integer> roleSet = parseIds(app.getPushRoleIds());
if (schoolSet.isEmpty() && roleSet.isEmpty()) {
return users;
}
return users.stream().filter(u -> {
if (!schoolSet.isEmpty()) {
List<SysUserGroupRel> rels = sysUserGroupRelRepository.findByUserId(u.getId());
boolean schoolMatch = rels.stream().anyMatch(r -> {
Integer sid = groupScopeHelper.resolveSchoolId(r.getGroupId());
return sid != null && schoolSet.contains(sid);
});
if (!schoolMatch) {
return false;
}
}
if (!roleSet.isEmpty()) {
if (u.getRoles() == null || u.getRoles().stream().noneMatch(roleSet::contains)) {
return false;
}
}
return true;
}).collect(Collectors.toList());
}
// ===================== 工具 =====================
private boolean containsEventType(String eventTypes, String path) {
if (eventTypes == null || eventTypes.trim().isEmpty()) {
return false;
}
for (String s : eventTypes.split(",")) {
if (s.trim().equals(path)) {
return true;
}
}
return false;
}
private Set<Integer> parseIds(String csv) {
Set<Integer> set = new HashSet<>();
if (csv == null || csv.trim().isEmpty()) {
return set;
}
for (String s : csv.split(",")) {
s = s.trim();
if (!s.isEmpty()) {
try {
set.add(Integer.valueOf(s));
} catch (NumberFormatException ignored) {
// 跳过非数字片段
}
}
}
return set;
}
private static String joinUrl(String base, String path) {
if (base == null) {
return path;
}
String b = base.endsWith("/") ? base : base + "/";
String p = path.startsWith("/") ? path.substring(1) : path;
return b + p;
}
}
src/main/java/com/yunhai/userbase/service/push/PushPayloadAssembler.java
+186
View File
@@ -0,0 +1,186 @@
package com.yunhai.userbase.service.push;
import com.yunhai.userbase.entity.SysGroup;
import com.yunhai.userbase.entity.SysUser;
import com.yunhai.userbase.entity.SysUserGroupRel;
import com.yunhai.userbase.enums.GroupTypeEnum;
import com.yunhai.userbase.enums.StatusEnum;
import com.yunhai.userbase.model.push.dto.PushGroupDTO;
import com.yunhai.userbase.model.push.dto.PushUserDTO;
import com.yunhai.userbase.repository.SysGroupRepository;
import com.yunhai.userbase.repository.SysUserGroupRelRepository;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.time.LocalDate;
import java.util.*;
import java.util.stream.Collectors;
/**
* 领域模型 → 文档契约 DTO 的装配器。<br><br>
* 关键映射（方案 v5 §2 / 文档 3.x）：
* <ul>
* <li>组 type：GroupTypeEnum.key 直接透传（1-8 零映射）</li>
* <li>用户 status：NORMAL(0)→0正常 / DISABLED(1)→2禁用</li>
* <li>groups[].schoolId：组所属学校范围 id（沿 parentId 向上最近 SCHOOL_GROUP）</li>
* <li>roleList[].schId：用户主组所属学校范围 id（userbase 无角色-学校关联，统一取主校）</li>
* </ul>
*
* @author chenx
* @date 2026/8/14
*/
@Service
public class PushPayloadAssembler {
private static final int ROOT_PARENT_ID = 0;
private static final int IS_MAIN = 1;
@Autowired
private SysGroupRepository sysGroupRepository;
@Autowired
private SysUserGroupRelRepository sysUserGroupRelRepository;
@Autowired
private GroupScopeHelper groupScopeHelper;
// ===================== 组 =====================
/** 组节点（不含 children）。AddGroup / UpdateGroup 使用。 */
public PushGroupDTO toGroupDto(SysGroup g) {
if (g == null) {
return null;
}
PushGroupDTO d = new PushGroupDTO();
d.setId(g.getId());
d.setName(g.getName());
// 文档 code 为「机构代码」，userbase 的 sys_group 无此属性，按可选项留空（序列化后为 null）
d.setType(g.getType() == null ? null : g.getType().getKey());
d.setPid(g.getParentId());
d.setNo(g.getSort());
return d;
}
/** 组节点 + 递归 children（SyncGroup 使用）。 */
public PushGroupDTO toGroupDtoWithChildren(SysGroup g, Map<Integer, List<SysGroup>> childrenMap) {
PushGroupDTO d = toGroupDto(g);
List<SysGroup> children = childrenMap.get(g.getId());
if (children != null) {
d.setChildren(children.stream()
.map(c -> toGroupDtoWithChildren(c, childrenMap))
.collect(Collectors.toList()));
}
return d;
}
/** 多个组（AddGroup / UpdateGroup），无 children，带 pid。 */
public List<PushGroupDTO> toGroupDtoList(List<SysGroup> groups) {
if (groups == null) {
return Collections.emptyList();
}
return groups.stream().map(this::toGroupDto).collect(Collectors.toList());
}
// ===================== 用户 =====================
/** 单个用户 → PushUserDTO（含 groups / roleList，学校 id 已解析）。 */
public PushUserDTO toUserDto(SysUser u) {
if (u == null) {
return null;
}
PushUserDTO d = new PushUserDTO();
d.setId(u.getId());
d.setAccount(u.getAccount());
d.setName(u.getRealName());
d.setIdcard(u.getIdcard());
d.setPhone(u.getPhone());
d.setEmail(u.getEmail());
d.setGender(u.getGender());
d.setBirthday(u.getBirthday() == null ? null : u.getBirthday().toString());
// status：NORMAL(0)→0正常 / DISABLED(1)→2禁用
d.setStatus(u.getDisabled() == StatusEnum.NORMAL ? 0 : 2);
d.setProfilePicture(u.getProfilePicture());
List<SysUserGroupRel> rels = sysUserGroupRelRepository.findByUserId(u.getId());
Integer mainSchoolId = resolveMainSchoolId(rels);
// groups
List<PushUserDTO.PushUserGroupRef> groupRefs = new ArrayList<>();
for (SysUserGroupRel rel : rels) {
PushUserDTO.PushUserGroupRef ref = new PushUserDTO.PushUserGroupRef();
ref.setGroupId(rel.getGroupId());
ref.setNo(rel.getSortNo());
ref.setSchoolId(groupScopeHelper.resolveSchoolId(rel.getGroupId()));
groupRefs.add(ref);
}
d.setGroups(groupRefs);
// roleList：每个 roleId 一行，schId 取主组学校（userbase 无角色-学校关联）
List<PushUserDTO.PushUserRoleRef> roleRefs = new ArrayList<>();
if (u.getRoles() != null) {
for (Integer roleId : u.getRoles()) {
PushUserDTO.PushUserRoleRef ref = new PushUserDTO.PushUserRoleRef();
ref.setRoleId(roleId);
ref.setSchId(mainSchoolId);
roleRefs.add(ref);
}
}
d.setRoleList(roleRefs);
return d;
}
/** 多个用户（AddUser / UpdateUser / SyncUser）。 */
public List<PushUserDTO> toUserDtoList(List<SysUser> users) {
if (users == null) {
return Collections.emptyList();
}
return users.stream().map(this::toUserDto).collect(Collectors.toList());
}
/** 取用户主组（isMain=1，否则首个关联）所属学校范围 id。 */
private Integer resolveMainSchoolId(List<SysUserGroupRel> rels) {
if (rels == null || rels.isEmpty()) {
return null;
}
Integer mainGroupId = rels.stream()
.filter(r -> r.getIsMain() != null && r.getIsMain() == IS_MAIN)
.map(SysUserGroupRel::getGroupId)
.findFirst()
.orElse(rels.get(0).getGroupId());
return groupScopeHelper.resolveSchoolId(mainGroupId);
}
// ===================== 公共工具 =====================
/** 构建 parentId → 直接子节点列表（null 视为根 0），同层按 sort / id 升序。 */
public Map<Integer, List<SysGroup>> buildChildrenMap(List<SysGroup> all) {
Map<Integer, List<SysGroup>> childrenMap = new HashMap<>();
for (SysGroup g : all) {
Integer key = (g.getParentId() == null) ? ROOT_PARENT_ID : g.getParentId();
childrenMap.computeIfAbsent(key, k -> new ArrayList<>()).add(g);
}
Comparator<SysGroup> order = (a, b) -> {
int sa = a.getSort() == null ? 999 : a.getSort();
int sb = b.getSort() == null ? 999 : b.getSort();
if (sa != sb) {
return Integer.compare(sa, sb);
}
return Integer.compare(a.getId(), b.getId());
};
for (List<SysGroup> siblings : childrenMap.values()) {
siblings.sort(order);
}
return childrenMap;
}
/** 构建整棵组树（从根 pid=0 起），SyncGroup 全量同步入口。 */
public List<PushGroupDTO> buildFullGroupTree() {
List<SysGroup> all = sysGroupRepository.findAll();
if (all.isEmpty()) {
return Collections.emptyList();
}
Map<Integer, List<SysGroup>> childrenMap = buildChildrenMap(all);
List<SysGroup> roots = childrenMap.getOrDefault(ROOT_PARENT_ID, new ArrayList<>());
return roots.stream()
.map(r -> toGroupDtoWithChildren(r, childrenMap))
.collect(Collectors.toList());
}
}
src/main/java/com/yunhai/userbase/service/push/PushRecordService.java
+110
View File
@@ -0,0 +1,110 @@
package com.yunhai.userbase.service.push;
import com.yunhai.platform.utils.exception.DException;
import com.yunhai.userbase.config.PushMqConfig;
import com.yunhai.userbase.entity.ApplicationInfo;
import com.yunhai.userbase.model.push.PushRecordEntity;
import com.yunhai.userbase.model.push.PushRecordQuery;
import com.yunhai.userbase.model.push.PushTaskMessage;
import com.yunhai.userbase.repository.ApplicationInfoRepository;
import com.yunhai.userbase.repository.PushRecordRepository;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.Pageable;
import org.springframework.data.jpa.domain.Specification;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import javax.persistence.criteria.Predicate;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.List;
/**
* 推送记录服务：日志查询 / 手动重试 / 定期清理（与操作日志同范式）。
*
* @author chenx
* @date 2026/8/14
*/
@Slf4j
@Service
public class PushRecordService {
@Autowired
private PushRecordRepository pushRecordRepository;
@Autowired
private ApplicationInfoRepository applicationInfoRepository;
@Autowired
@Qualifier("pushRabbitTemplate")
private RabbitTemplate pushRabbitTemplate;
/** 分页查询（按条件过滤，默认按创建时间倒序在 Controller 层设置）。 */
public Page<PushRecordEntity> query(PushRecordQuery q, Pageable pageable) {
Specification<PushRecordEntity> spec = (root, cq, cb) -> {
List<Predicate> ps = new ArrayList<>();
if (q.getAppKey() != null) {
ps.add(cb.equal(root.get("subscriberAppKey"), q.getAppKey()));
}
if (q.getInterfacePath() != null && !q.getInterfacePath().isEmpty()) {
ps.add(cb.equal(root.get("interfacePath"), q.getInterfacePath()));
}
if (q.getSuccess() != null) {
ps.add(cb.equal(root.get("success"), q.getSuccess()));
}
if (q.getStartTime() != null && !q.getStartTime().isEmpty()) {
ps.add(cb.greaterThanOrEqualTo(root.get("createdAt"), parseStart(q.getStartTime())));
}
if (q.getEndTime() != null && !q.getEndTime().isEmpty()) {
ps.add(cb.lessThanOrEqualTo(root.get("createdAt"), parseEnd(q.getEndTime())));
}
return cb.and(ps.toArray(new Predicate[0]));
};
return pushRecordRepository.findAll(spec, pageable);
}
/** 手动重试：用原始 payload 重新签名（新时间戳）投递 MQ，由消费者再次落库。 */
@Transactional
public void retry(Long recordId) {
PushRecordEntity rec = pushRecordRepository.findById(recordId)
.orElseThrow(() -> new DException("推送记录不存在"));
if (rec.getSuccess() != null && rec.getSuccess() == 1) {
return; // 已成功无需重试
}
ApplicationInfo app = applicationInfoRepository.findByAppKey(rec.getSubscriberAppKey());
if (app == null) {
throw new DException("订阅方不存在");
}
String secret = (app.getPushSecret() != null && !app.getPushSecret().isEmpty())
? app.getPushSecret() : app.getAppSecret();
String t = String.valueOf(System.currentTimeMillis());
String sign = PushSigner.sign(rec.getPayload(), secret, t);
PushTaskMessage msg = new PushTaskMessage(app.getAppKey(), rec.getInterfacePath(),
rec.getInterfaceLabel(), rec.getPayload(), t, sign, 1, "retry");
pushRabbitTemplate.convertAndSend(PushMqConfig.EXCHANGE_TOPIC, rec.getInterfacePath(), msg);
}
/** 清理 1 年前的记录（每日 3 点由 PushRecordCleanupTask 调用）。 */
@Transactional
public long cleanupExpired() {
LocalDateTime cutoff = LocalDateTime.now().minusYears(1);
List<PushRecordEntity> expired = pushRecordRepository.findAll(
(root, cq, cb) -> cb.lessThan(root.get("createdAt"), cutoff));
long n = expired.size();
if (n > 0) {
pushRecordRepository.deleteAll(expired);
}
return n;
}
private LocalDateTime parseStart(String s) {
return LocalDateTime.parse(s + "T00:00:00", DateTimeFormatter.ISO_LOCAL_DATE_TIME);
}
private LocalDateTime parseEnd(String s) {
return LocalDateTime.parse(s + "T23:59:59", DateTimeFormatter.ISO_LOCAL_DATE_TIME);
}
}
src/main/java/com/yunhai/userbase/service/push/PushSigner.java
+50
View File
@@ -0,0 +1,50 @@
package com.yunhai.userbase.service.push;
import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;
import java.nio.charset.StandardCharsets;
/**
* 推送签名器：复刻文档 2.x 规范。<br><br>
* 签名串 DATA2 = 业务JSON(原始字符串) + 时间戳t；<br>
* sign = HMAC-SHA256(DATA2, secret)，UTF-8，hex 小写。<br><br>
* <b>关键</b>：签名与发送均基于「发送方实际写出的 body 原始字符串字节」，
* 订阅方须用接收到的原始 body 验签（不得反序列化后重新序列化，否则字节不一致导致验签失败）。
*
* @author chenx
* @date 2026/8/14
*/
public final class PushSigner {
private PushSigner() {
}
/**
* 计算签名。
*
* @param body 业务数据原始 JSON 字符串（与发送 body 完全一致）
* @param secret 签名密钥（push_secret 优先，否则 app_secret）
* @param t 时间戳（毫秒或秒，与订阅方约定一致即可；本实现直接拼接字符串）
* @return hex 小写签名
*/
public static String sign(String body, String secret, String t) {
try {
Mac mac = Mac.getInstance("HmacSHA256");
SecretKeySpec keySpec = new SecretKeySpec(secret.getBytes(StandardCharsets.UTF_8), "HmacSHA256");
mac.init(keySpec);
String data2 = body + t;
byte[] raw = mac.doFinal(data2.getBytes(StandardCharsets.UTF_8));
return toHex(raw);
} catch (Exception e) {
throw new RuntimeException("推送签名失败", e);
}
}
private static String toHex(byte[] bytes) {
StringBuilder sb = new StringBuilder(bytes.length * 2);
for (byte b : bytes) {
sb.append(String.format("%02x", b));
}
return sb.toString();
}
}