diff --git a/pom.xml b/pom.xml index 83cce41..2c11623 100644 --- a/pom.xml +++ b/pom.xml @@ -93,6 +93,12 @@ spring-boot-starter-data-redis + + + org.springframework.boot + spring-boot-starter-amqp + + org.springframework.boot diff --git a/src/main/java/com/yunhai/userbase/controller/api/AppSettingsController.java b/src/main/java/com/yunhai/userbase/controller/api/AppSettingsController.java index b693763..cc3e9d6 100644 --- a/src/main/java/com/yunhai/userbase/controller/api/AppSettingsController.java +++ b/src/main/java/com/yunhai/userbase/controller/api/AppSettingsController.java @@ -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 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); + } } diff --git a/src/main/java/com/yunhai/userbase/controller/api/PushSubscriptionController.java b/src/main/java/com/yunhai/userbase/controller/api/PushSubscriptionController.java new file mode 100644 index 0000000..3999d75 --- /dev/null +++ b/src/main/java/com/yunhai/userbase/controller/api/PushSubscriptionController.java @@ -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 test(@RequestBody PushTestRequest request) { + pushDispatcher.dispatchManual(request.getAppId(), request.getInterfacePath(), request.getTargetIds()); + Map preview = pushDispatcher.previewManual( + request.getAppId(), request.getInterfacePath(), request.getTargetIds()); + Map result = new LinkedHashMap<>(); + result.put("sent", true); + result.put("message", "已投递推送队列，可在「推送日志」查看实际响应与耗时"); + result.put("preview", preview); + return result; + } + + /** 推送日志列表。 */ + @GetMapping("RecordList") + public BasePageResult 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); + } +} diff --git a/src/main/java/com/yunhai/userbase/entity/ApplicationInfo.java b/src/main/java/com/yunhai/userbase/entity/ApplicationInfo.java index da2fa88..6d17c06 100644 --- a/src/main/java/com/yunhai/userbase/entity/ApplicationInfo.java +++ b/src/main/java/com/yunhai/userbase/entity/ApplicationInfo.java @@ -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; + } \ No newline at end of file diff --git a/src/main/java/com/yunhai/userbase/repository/ApplicationInfoRepository.java b/src/main/java/com/yunhai/userbase/repository/ApplicationInfoRepository.java index 8bde07a..a34acb3 100644 --- a/src/main/java/com/yunhai/userbase/repository/ApplicationInfoRepository.java +++ b/src/main/java/com/yunhai/userbase/repository/ApplicationInfoRepository.java @@ -42,6 +42,9 @@ public interface ApplicationInfoRepository extends JpaRepository findByPushEnabled(Integer pushEnabled); + /** * 按 id 更新排序号（同一应用类型内排序使用）。
* diff --git a/src/main/java/com/yunhai/userbase/service/IApplicationInfoService.java b/src/main/java/com/yunhai/userbase/service/IApplicationInfoService.java index b91c55f..08d31c0 100644 --- a/src/main/java/com/yunhai/userbase/service/IApplicationInfoService.java +++ b/src/main/java/com/yunhai/userbase/service/IApplicationInfoService.java @@ -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 ids); + + /** + * 保存应用推送设置（外部接口地址 / 总开关 / 启用的接口 / 学校范围 / 角色范围 / 推送密钥）。
+ * 仅更新推送相关字段，写后同步更新 appInfo 缓存。 + * + * @param id 应用ID + * @param request 推送设置（各字段可选，null 表示不修改） + */ + ApplicationInfo savePushSetting(Integer id, PushSettingRequest request); + + /** + * 获取应用推送设置回显 VO。 + * + * @param id 应用ID + * @return 推送设置 VO（仅含推送相关字段） + */ + PushSettingVo getPushSetting(Integer id); } diff --git a/src/main/java/com/yunhai/userbase/service/impl/ApplicationInfoService.java b/src/main/java/com/yunhai/userbase/service/impl/ApplicationInfoService.java index 0bedc80..a24150a 100644 --- a/src/main/java/com/yunhai/userbase/service/impl/ApplicationInfoService.java +++ b/src/main/java/com/yunhai/userbase/service/impl/ApplicationInfoService.java @@ -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 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; + } } \ No newline at end of file diff --git a/src/main/java/com/yunhai/userbase/service/impl/SysGroupService.java b/src/main/java/com/yunhai/userbase/service/impl/SysGroupService.java index f357502..33044a0 100644 --- a/src/main/java/com/yunhai/userbase/service/impl/SysGroupService.java +++ b/src/main/java/com/yunhai/userbase/service/impl/SysGroupService.java @@ -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 groupIds) { + if (groupIds == null || groupIds.isEmpty()) { + return; + } + DataSyncEvent e = new DataSyncEvent(); + e.setType(type); + e.setGroupIds(groupIds); + eventPublisher.publishEvent(e); + } } diff --git a/src/main/java/com/yunhai/userbase/service/impl/SysUserService.java b/src/main/java/com/yunhai/userbase/service/impl/SysUserService.java index a3ef3f5..638ebdd 100644 --- a/src/main/java/com/yunhai/userbase/service/impl/SysUserService.java +++ b/src/main/java/com/yunhai/userbase/service/impl/SysUserService.java @@ -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 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 userIds) { + if (userIds == null || userIds.isEmpty()) { + return; + } + DataSyncEvent e = new DataSyncEvent(); + e.setType(type); + e.setUserIds(userIds); + eventPublisher.publishEvent(e); + } } diff --git a/src/main/java/com/yunhai/userbase/service/push/GroupScopeHelper.java b/src/main/java/com/yunhai/userbase/service/push/GroupScopeHelper.java new file mode 100644 index 0000000..a708f08 --- /dev/null +++ b/src/main/java/com/yunhai/userbase/service/push/GroupScopeHelper.java @@ -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; + +/** + * 学校范围辅助：把「学校组」语义落地为可计算的范围。

+ * 语义（已与用户确认）：一个组属于学校 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 getSchoolScopeGroupIds(Integer schoolId) { + if (schoolId == null) { + return Collections.emptyList(); + } + return sysGroupService.listSelfAndDescendantIds(schoolId); + } +} diff --git a/src/main/java/com/yunhai/userbase/service/push/PushConsumer.java b/src/main/java/com/yunhai/userbase/service/push/PushConsumer.java new file mode 100644 index 0000000..bb84f7f --- /dev/null +++ b/src/main/java/com/yunhai/userbase/service/push/PushConsumer.java @@ -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; + +/** + * 推送消费者（订阅方调用方）：
+ * 从 push.main 串行消费，手动 ack；POST 给订阅方 Webhook，按文档契约带 t/sign 头；
+ * 每次尝试均落 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 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 unused() { + return Collections.emptyMap(); + } +} diff --git a/src/main/java/com/yunhai/userbase/service/push/PushDispatcher.java b/src/main/java/com/yunhai/userbase/service/push/PushDispatcher.java new file mode 100644 index 0000000..2e4bc9a --- /dev/null +++ b/src/main/java/com/yunhai/userbase/service/push/PushDispatcher.java @@ -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; + +/** + * 推送分发器（生产方）：
+ * ① 自动触发：监听 {@link DataSyncEvent}（AFTER_COMMIT），遍历所有开启推送的订阅方，按配置过滤后投递 MQ；
+ * ② 手动测试：{@link #dispatchManual} 供测试推送页调用。

+ * 签名在此处由基础平台计算一次（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 groups = (event.getGroupIds() != null && !event.getGroupIds().isEmpty()) + ? sysGroupRepository.findAllById(event.getGroupIds()) : Collections.emptyList(); + List 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 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 groups = Collections.emptyList(); + List 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 / 头 / 体），供测试推送页展示。
+ * 签名在服务端用 secret 计算，前端不持有密钥；实际投递仍走 {@link #dispatchManual} → MQ。 + */ + public Map previewManual(Integer appId, String interfacePath, List 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 groups = Collections.emptyList(); + List 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 bodies = buildBodies(type, groups, users); + if (bodies.isEmpty()) { + Map 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 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 buildHeaders(String t, String sign) { + Map 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 groups, List 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 fGroups = (type.isGroup() && !isFull) ? filterGroups(groups, app) : groups; + List 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 buildBodies(PushInterfaceType type, List groups, List 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 filterGroups(List groups, ApplicationInfo app) { + Set 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 filterUsers(List users, ApplicationInfo app) { + Set schoolSet = parseIds(app.getPushSchoolIds()); + Set roleSet = parseIds(app.getPushRoleIds()); + if (schoolSet.isEmpty() && roleSet.isEmpty()) { + return users; + } + return users.stream().filter(u -> { + if (!schoolSet.isEmpty()) { + List 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 parseIds(String csv) { + Set 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; + } +} diff --git a/src/main/java/com/yunhai/userbase/service/push/PushPayloadAssembler.java b/src/main/java/com/yunhai/userbase/service/push/PushPayloadAssembler.java new file mode 100644 index 0000000..4596df5 --- /dev/null +++ b/src/main/java/com/yunhai/userbase/service/push/PushPayloadAssembler.java @@ -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 的装配器。

+ * 关键映射（方案 v5 §2 / 文档 3.x）： + *
    + *
  • 组 type：GroupTypeEnum.key 直接透传（1-8 零映射）
    • + *
  • 用户 status：NORMAL(0)→0正常 / DISABLED(1)→2禁用
    • + *
  • groups[].schoolId：组所属学校范围 id（沿 parentId 向上最近 SCHOOL_GROUP）
    • + *
  • roleList[].schId：用户主组所属学校范围 id（userbase 无角色-学校关联，统一取主校）
    • + *
+ * + * @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> childrenMap) { + PushGroupDTO d = toGroupDto(g); + List 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 toGroupDtoList(List 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 rels = sysUserGroupRelRepository.findByUserId(u.getId()); + Integer mainSchoolId = resolveMainSchoolId(rels); + + // groups + List 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 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 toUserDtoList(List users) { + if (users == null) { + return Collections.emptyList(); + } + return users.stream().map(this::toUserDto).collect(Collectors.toList()); + } + + /** 取用户主组（isMain=1，否则首个关联）所属学校范围 id。 */ + private Integer resolveMainSchoolId(List 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> buildChildrenMap(List all) { + Map> 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 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 siblings : childrenMap.values()) { + siblings.sort(order); + } + return childrenMap; + } + + /** 构建整棵组树（从根 pid=0 起），SyncGroup 全量同步入口。 */ + public List buildFullGroupTree() { + List all = sysGroupRepository.findAll(); + if (all.isEmpty()) { + return Collections.emptyList(); + } + Map> childrenMap = buildChildrenMap(all); + List roots = childrenMap.getOrDefault(ROOT_PARENT_ID, new ArrayList<>()); + return roots.stream() + .map(r -> toGroupDtoWithChildren(r, childrenMap)) + .collect(Collectors.toList()); + } +} diff --git a/src/main/java/com/yunhai/userbase/service/push/PushRecordService.java b/src/main/java/com/yunhai/userbase/service/push/PushRecordService.java new file mode 100644 index 0000000..6388aec --- /dev/null +++ b/src/main/java/com/yunhai/userbase/service/push/PushRecordService.java @@ -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 query(PushRecordQuery q, Pageable pageable) { + Specification spec = (root, cq, cb) -> { + List 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 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); + } +} diff --git a/src/main/java/com/yunhai/userbase/service/push/PushSigner.java b/src/main/java/com/yunhai/userbase/service/push/PushSigner.java new file mode 100644 index 0000000..338956c --- /dev/null +++ b/src/main/java/com/yunhai/userbase/service/push/PushSigner.java @@ -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 规范。

+ * 签名串 DATA2 = 业务JSON(原始字符串) + 时间戳t；
+ * sign = HMAC-SHA256(DATA2, secret)，UTF-8，hex 小写。

+ * 关键：签名与发送均基于「发送方实际写出的 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(); + } +}