This commit is contained in:
@@ -3,17 +3,25 @@ package com.yunhai.userbase.controller.api;
|
|||||||
import com.yunhai.userbase.config.Config;
|
import com.yunhai.userbase.config.Config;
|
||||||
import com.yunhai.userbase.model.BasePageQuery;
|
import com.yunhai.userbase.model.BasePageQuery;
|
||||||
import com.yunhai.userbase.model.BasePageResult;
|
import com.yunhai.userbase.model.BasePageResult;
|
||||||
import com.yunhai.userbase.model.push.PushRecordEntity;
|
import com.yunhai.userbase.model.push.PushRecordListVo;
|
||||||
import com.yunhai.userbase.model.push.PushRecordQuery;
|
import com.yunhai.userbase.model.push.PushRecordQuery;
|
||||||
import com.yunhai.userbase.model.push.PushTestRequest;
|
import com.yunhai.userbase.model.push.PushTestRequest;
|
||||||
import com.yunhai.userbase.service.push.PushDispatcher;
|
import com.yunhai.userbase.service.push.PushDispatcher;
|
||||||
import com.yunhai.userbase.service.push.PushRecordService;
|
import com.yunhai.userbase.service.push.PushRecordService;
|
||||||
import com.yunhai.userbase.utils.PageUtils;
|
import com.yunhai.userbase.utils.PageUtils;
|
||||||
|
import lombok.extern.slf4j.Slf4j;
|
||||||
import org.springframework.beans.factory.annotation.Autowired;
|
import org.springframework.beans.factory.annotation.Autowired;
|
||||||
|
import org.springframework.core.io.InputStreamResource;
|
||||||
import org.springframework.data.domain.Sort;
|
import org.springframework.data.domain.Sort;
|
||||||
import org.springframework.data.web.SortDefault;
|
import org.springframework.data.web.SortDefault;
|
||||||
|
import org.springframework.http.HttpHeaders;
|
||||||
|
import org.springframework.http.HttpStatus;
|
||||||
|
import org.springframework.http.MediaType;
|
||||||
|
import org.springframework.http.ResponseEntity;
|
||||||
import org.springframework.web.bind.annotation.*;
|
import org.springframework.web.bind.annotation.*;
|
||||||
|
|
||||||
|
import java.io.ByteArrayInputStream;
|
||||||
|
import java.nio.charset.StandardCharsets;
|
||||||
import java.util.LinkedHashMap;
|
import java.util.LinkedHashMap;
|
||||||
import java.util.Map;
|
import java.util.Map;
|
||||||
|
|
||||||
@@ -24,6 +32,7 @@ import java.util.Map;
|
|||||||
* @date 2026/8/14
|
* @date 2026/8/14
|
||||||
*/
|
*/
|
||||||
@RestController
|
@RestController
|
||||||
|
@Slf4j
|
||||||
@RequestMapping(value = Config.API_PATH + "/push/")
|
@RequestMapping(value = Config.API_PATH + "/push/")
|
||||||
public class PushSubscriptionController {
|
public class PushSubscriptionController {
|
||||||
|
|
||||||
@@ -45,9 +54,9 @@ public class PushSubscriptionController {
|
|||||||
return result;
|
return result;
|
||||||
}
|
}
|
||||||
|
|
||||||
/** 推送日志列表。 */
|
/** 推送日志列表(不含 payload,避免全量推送数据量过大;payload 经 Download 接口单独获取)。 */
|
||||||
@GetMapping("RecordList")
|
@GetMapping("RecordList")
|
||||||
public BasePageResult<PushRecordEntity> recordList(PushRecordQuery query, BasePageQuery pageQuery,
|
public BasePageResult<PushRecordListVo> recordList(PushRecordQuery query, BasePageQuery pageQuery,
|
||||||
@SortDefault(sort = "createdAt", direction = Sort.Direction.DESC) Sort sort) {
|
@SortDefault(sort = "createdAt", direction = Sort.Direction.DESC) Sort sort) {
|
||||||
return PageUtils.toResult(pushRecordService.query(query, PageUtils.toPageable(pageQuery, sort)));
|
return PageUtils.toResult(pushRecordService.query(query, PageUtils.toPageable(pageQuery, sort)));
|
||||||
}
|
}
|
||||||
@@ -57,4 +66,32 @@ public class PushSubscriptionController {
|
|||||||
public void retry(@PathVariable Long id) {
|
public void retry(@PathVariable Long id) {
|
||||||
pushRecordService.retry(id);
|
pushRecordService.retry(id);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 下载某条推送记录的请求体原文(payload 文本文件)。<br>
|
||||||
|
* 采用与项目 DownloadFileUtil 一致的 octet-stream + attachment 写法,确保浏览器稳定触发下载。<br>
|
||||||
|
* 若该记录未记录 payload,返回一段说明文本文件而非 404,便于排查。
|
||||||
|
*/
|
||||||
|
@GetMapping("/Download/{id}")
|
||||||
|
public ResponseEntity<InputStreamResource> download(@PathVariable Long id) {
|
||||||
|
String data = pushRecordService.getPayload(id);
|
||||||
|
if (data == null) {
|
||||||
|
data = "该条推送记录未记录请求体(payload 为空),无法下载原始内容。";
|
||||||
|
}
|
||||||
|
try {
|
||||||
|
ByteArrayInputStream bis = new ByteArrayInputStream(data.getBytes(StandardCharsets.UTF_8));
|
||||||
|
HttpHeaders headers = new HttpHeaders();
|
||||||
|
headers.add("Cache-Control", "no-cache, no-store, must-revalidate");
|
||||||
|
headers.add(HttpHeaders.CONTENT_DISPOSITION,
|
||||||
|
"attachment; filename=" + new String(("payload-" + id + ".txt").getBytes("gbk"), "iso8859-1"));
|
||||||
|
headers.add("Pragma", "no-cache");
|
||||||
|
headers.add("Expires", "0");
|
||||||
|
return ResponseEntity.ok().headers(headers)
|
||||||
|
.contentType(MediaType.parseMediaType("application/octet-stream"))
|
||||||
|
.body(new InputStreamResource(bis));
|
||||||
|
} catch (Exception e) {
|
||||||
|
log.error("推送日志请求体下载失败, id={}", id, e);
|
||||||
|
return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).build();
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,74 @@
|
|||||||
|
package com.yunhai.userbase.model.push;
|
||||||
|
|
||||||
|
import com.fasterxml.jackson.annotation.JsonFormat;
|
||||||
|
import lombok.Data;
|
||||||
|
|
||||||
|
import java.time.LocalDateTime;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 推送记录列表项视图。<br>
|
||||||
|
* 不含 payload(请求体原文),列表不展示,需通过下载接口单独获取,避免全量推送时数据量过大。<br>
|
||||||
|
*
|
||||||
|
* @author chenx
|
||||||
|
* @date 2026/8/17
|
||||||
|
*/
|
||||||
|
@Data
|
||||||
|
public class PushRecordListVo {
|
||||||
|
|
||||||
|
/** 主键 */
|
||||||
|
private Long id;
|
||||||
|
/** 订阅方 appKey */
|
||||||
|
private Integer subscriberAppKey;
|
||||||
|
/** 应用名称 */
|
||||||
|
private String appName;
|
||||||
|
/** 接口路径,如 group/SyncGroup */
|
||||||
|
private String interfacePath;
|
||||||
|
/** 接口中文标签 */
|
||||||
|
private String interfaceLabel;
|
||||||
|
/** 请求地址 */
|
||||||
|
private String requestUrl;
|
||||||
|
/** 签名时间 t */
|
||||||
|
private String signT;
|
||||||
|
/** HTTP 状态码 */
|
||||||
|
private Integer httpStatus;
|
||||||
|
/** 业务码 */
|
||||||
|
private Integer respCode;
|
||||||
|
/** 业务消息 */
|
||||||
|
private String respMsg;
|
||||||
|
/** 是否成功:0失败 1成功 */
|
||||||
|
private Integer success;
|
||||||
|
/** 重试次数 */
|
||||||
|
private Integer attemptNo;
|
||||||
|
/** 耗时(ms) */
|
||||||
|
private Integer costMs;
|
||||||
|
/** 触发类型:manual / auto / retry */
|
||||||
|
private String triggerType;
|
||||||
|
/** 创建时间 */
|
||||||
|
@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
|
||||||
|
private LocalDateTime createdAt;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 全参构造,顺序需与 {@code PushRecordEntity} 的列表投影字段一致。<br>
|
||||||
|
* 用于 JPA Criteria {@code cb.construct} 投影查询。
|
||||||
|
*/
|
||||||
|
public PushRecordListVo(Long id, Integer subscriberAppKey, String appName, String interfacePath,
|
||||||
|
String interfaceLabel, String requestUrl, String signT, Integer httpStatus,
|
||||||
|
Integer respCode, String respMsg, Integer success, Integer attemptNo,
|
||||||
|
Integer costMs, String triggerType, LocalDateTime createdAt) {
|
||||||
|
this.id = id;
|
||||||
|
this.subscriberAppKey = subscriberAppKey;
|
||||||
|
this.appName = appName;
|
||||||
|
this.interfacePath = interfacePath;
|
||||||
|
this.interfaceLabel = interfaceLabel;
|
||||||
|
this.requestUrl = requestUrl;
|
||||||
|
this.signT = signT;
|
||||||
|
this.httpStatus = httpStatus;
|
||||||
|
this.respCode = respCode;
|
||||||
|
this.respMsg = respMsg;
|
||||||
|
this.success = success;
|
||||||
|
this.attemptNo = attemptNo;
|
||||||
|
this.costMs = costMs;
|
||||||
|
this.triggerType = triggerType;
|
||||||
|
this.createdAt = createdAt;
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -4,6 +4,7 @@ import com.yunhai.platform.utils.exception.DException;
|
|||||||
import com.yunhai.userbase.config.PushMqConfig;
|
import com.yunhai.userbase.config.PushMqConfig;
|
||||||
import com.yunhai.userbase.entity.ApplicationInfo;
|
import com.yunhai.userbase.entity.ApplicationInfo;
|
||||||
import com.yunhai.userbase.model.push.PushRecordEntity;
|
import com.yunhai.userbase.model.push.PushRecordEntity;
|
||||||
|
import com.yunhai.userbase.model.push.PushRecordListVo;
|
||||||
import com.yunhai.userbase.model.push.PushRecordQuery;
|
import com.yunhai.userbase.model.push.PushRecordQuery;
|
||||||
import com.yunhai.userbase.model.push.PushTaskMessage;
|
import com.yunhai.userbase.model.push.PushTaskMessage;
|
||||||
import com.yunhai.userbase.repository.ApplicationInfoRepository;
|
import com.yunhai.userbase.repository.ApplicationInfoRepository;
|
||||||
@@ -14,11 +15,18 @@ import org.springframework.beans.factory.annotation.Autowired;
|
|||||||
import org.springframework.beans.factory.annotation.Qualifier;
|
import org.springframework.beans.factory.annotation.Qualifier;
|
||||||
import org.springframework.data.domain.Page;
|
import org.springframework.data.domain.Page;
|
||||||
import org.springframework.data.domain.Pageable;
|
import org.springframework.data.domain.Pageable;
|
||||||
import org.springframework.data.jpa.domain.Specification;
|
import org.springframework.data.domain.Sort;
|
||||||
import org.springframework.stereotype.Service;
|
import org.springframework.stereotype.Service;
|
||||||
import org.springframework.transaction.annotation.Transactional;
|
import org.springframework.transaction.annotation.Transactional;
|
||||||
|
|
||||||
|
import javax.persistence.EntityManager;
|
||||||
|
import javax.persistence.PersistenceContext;
|
||||||
|
import javax.persistence.TypedQuery;
|
||||||
|
import javax.persistence.criteria.CriteriaBuilder;
|
||||||
|
import javax.persistence.criteria.CriteriaQuery;
|
||||||
|
import javax.persistence.criteria.Order;
|
||||||
import javax.persistence.criteria.Predicate;
|
import javax.persistence.criteria.Predicate;
|
||||||
|
import javax.persistence.criteria.Root;
|
||||||
import java.time.LocalDateTime;
|
import java.time.LocalDateTime;
|
||||||
import java.time.format.DateTimeFormatter;
|
import java.time.format.DateTimeFormatter;
|
||||||
import java.util.ArrayList;
|
import java.util.ArrayList;
|
||||||
@@ -42,28 +50,46 @@ public class PushRecordService {
|
|||||||
@Qualifier("pushRabbitTemplate")
|
@Qualifier("pushRabbitTemplate")
|
||||||
private RabbitTemplate pushRabbitTemplate;
|
private RabbitTemplate pushRabbitTemplate;
|
||||||
|
|
||||||
/** 分页查询(按条件过滤,默认按创建时间倒序在 Controller 层设置)。 */
|
@PersistenceContext
|
||||||
public Page<PushRecordEntity> query(PushRecordQuery q, Pageable pageable) {
|
private EntityManager em;
|
||||||
Specification<PushRecordEntity> spec = (root, cq, cb) -> {
|
|
||||||
List<Predicate> ps = new ArrayList<>();
|
/** 分页查询(列表投影排除 payload,避免全量推送时数据量过大;默认按创建时间倒序)。 */
|
||||||
if (q.getAppKey() != null) {
|
public Page<PushRecordListVo> query(PushRecordQuery q, Pageable pageable) {
|
||||||
ps.add(cb.equal(root.get("subscriberAppKey"), q.getAppKey()));
|
CriteriaBuilder cb = em.getCriteriaBuilder();
|
||||||
}
|
// 列表投影:显式选择字段,不包含 payload(MEDIUMTEXT,全量同步可能非常大)
|
||||||
if (q.getInterfacePath() != null && !q.getInterfacePath().isEmpty()) {
|
CriteriaQuery<PushRecordListVo> cq = cb.createQuery(PushRecordListVo.class);
|
||||||
ps.add(cb.equal(root.get("interfacePath"), q.getInterfacePath()));
|
Root<PushRecordEntity> root = cq.from(PushRecordEntity.class);
|
||||||
}
|
cq.select(cb.construct(PushRecordListVo.class,
|
||||||
if (q.getSuccess() != null) {
|
root.get("id"), root.get("subscriberAppKey"), root.get("appName"),
|
||||||
ps.add(cb.equal(root.get("success"), q.getSuccess()));
|
root.get("interfacePath"), root.get("interfaceLabel"), root.get("requestUrl"),
|
||||||
}
|
root.get("signT"), root.get("httpStatus"), root.get("respCode"), root.get("respMsg"),
|
||||||
if (q.getStartTime() != null && !q.getStartTime().isEmpty()) {
|
root.get("success"), root.get("attemptNo"), root.get("costMs"),
|
||||||
ps.add(cb.greaterThanOrEqualTo(root.get("createdAt"), parseStart(q.getStartTime())));
|
root.get("triggerType"), root.get("createdAt")));
|
||||||
}
|
List<Predicate> predicates = buildPredicates(cb, root, q);
|
||||||
if (q.getEndTime() != null && !q.getEndTime().isEmpty()) {
|
cq.where(predicates.toArray(new Predicate[0]));
|
||||||
ps.add(cb.lessThanOrEqualTo(root.get("createdAt"), parseEnd(q.getEndTime())));
|
cq.orderBy(buildOrders(cb, root, pageable.getSort()));
|
||||||
}
|
|
||||||
return cb.and(ps.toArray(new Predicate[0]));
|
// 总数查询
|
||||||
};
|
CriteriaQuery<Long> countCq = cb.createQuery(Long.class);
|
||||||
return pushRecordRepository.findAll(spec, pageable);
|
Root<PushRecordEntity> countRoot = countCq.from(PushRecordEntity.class);
|
||||||
|
countCq.select(cb.count(countRoot));
|
||||||
|
countCq.where(buildPredicates(cb, countRoot, q).toArray(new Predicate[0]));
|
||||||
|
long total = em.createQuery(countCq).getSingleResult();
|
||||||
|
|
||||||
|
TypedQuery<PushRecordListVo> typed = em.createQuery(cq);
|
||||||
|
int pageIndex = pageable.getPageNumber();
|
||||||
|
int pageSize = pageable.getPageSize();
|
||||||
|
typed.setFirstResult(pageIndex * pageSize);
|
||||||
|
typed.setMaxResults(pageSize);
|
||||||
|
List<PushRecordListVo> content = typed.getResultList();
|
||||||
|
|
||||||
|
return new org.springframework.data.domain.PageImpl<>(content, pageable, total);
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 获取某条推送记录的请求体原文(payload)。无记录返回 null。 */
|
||||||
|
public String getPayload(Long id) {
|
||||||
|
PushRecordEntity rec = pushRecordRepository.findById(id).orElse(null);
|
||||||
|
return rec != null ? rec.getPayload() : null;
|
||||||
}
|
}
|
||||||
|
|
||||||
/** 手动重试:用原始 payload 重新签名(新时间戳)投递 MQ,由消费者再次落库。 */
|
/** 手动重试:用原始 payload 重新签名(新时间戳)投递 MQ,由消费者再次落库。 */
|
||||||
@@ -87,6 +113,43 @@ public class PushRecordService {
|
|||||||
pushRabbitTemplate.convertAndSend(PushMqConfig.EXCHANGE_TOPIC, rec.getInterfacePath(), msg);
|
pushRabbitTemplate.convertAndSend(PushMqConfig.EXCHANGE_TOPIC, rec.getInterfacePath(), msg);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private List<Predicate> buildPredicates(CriteriaBuilder cb, Root<PushRecordEntity> root, PushRecordQuery q) {
|
||||||
|
List<Predicate> ps = new ArrayList<>();
|
||||||
|
if (q == null) {
|
||||||
|
return ps;
|
||||||
|
}
|
||||||
|
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 ps;
|
||||||
|
}
|
||||||
|
|
||||||
|
private List<Order> buildOrders(CriteriaBuilder cb, Root<PushRecordEntity> root, Sort sort) {
|
||||||
|
List<Order> orders = new ArrayList<>();
|
||||||
|
if (sort != null && sort.isSorted()) {
|
||||||
|
for (Sort.Order o : sort) {
|
||||||
|
javax.persistence.criteria.Path<?> path = root.get(o.getProperty());
|
||||||
|
orders.add(o.isAscending() ? cb.asc(path) : cb.desc(path));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (orders.isEmpty()) {
|
||||||
|
orders.add(cb.desc(root.get("createdAt")));
|
||||||
|
}
|
||||||
|
return orders;
|
||||||
|
}
|
||||||
|
|
||||||
/** 清理 1 年前的记录(每日 3 点由 PushRecordCleanupTask 调用)。 */
|
/** 清理 1 年前的记录(每日 3 点由 PushRecordCleanupTask 调用)。 */
|
||||||
@Transactional
|
@Transactional
|
||||||
public long cleanupExpired() {
|
public long cleanupExpired() {
|
||||||
|
|||||||
Reference in New Issue
Block a user