package com.yunhai.flowcenter.service; import com.yunhai.flowcenter.callback.CallbackProperties; import com.yunhai.flowcenter.callback.CallbackPushService; import com.yunhai.flowcenter.entity.FlwCallbackLog; import com.yunhai.flowcenter.entity.FlwCallbackReg; import com.yunhai.flowcenter.enums.FlwEnums; import com.yunhai.flowcenter.repository.FlwCallbackLogRepository; import com.yunhai.flowcenter.repository.FlwCallbackRegRepository; import com.yunhai.platform.apistd.BizException; import com.yunhai.platform.utils.exception.ErrorCode; import com.yunhai.platform.utils.model.BasePageResult; import lombok.RequiredArgsConstructor; import org.springframework.data.domain.Page; import org.springframework.data.domain.PageRequest; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import java.time.LocalDateTime; import java.util.List; import java.util.UUID; /** * 回调注册/推送日志/死信管理(概览 §9.2): * 启停、secret 重置(重置后旧签名立即失效)、推送日志查询、死信列表与手工重推。 */ @Service @RequiredArgsConstructor public class FlowCallbackAdminService { private final FlwCallbackRegRepository regRepository; private final FlwCallbackLogRepository logRepository; private final CallbackPushService pushService; private final CallbackProperties callbackProperties; public List listRegs(Long schId) { return regRepository.findBySchIdOrderByUpdateTimeDesc(schId); } /** 启用/停用注册(停用后该注册不再收到推送,日志保留) */ @Transactional public void setRegStatus(Long schId, Long regId, boolean enabled) { FlwCallbackReg reg = requireReg(schId, regId); reg.setStatus(enabled ? 0 : 1); regRepository.save(reg); } /** secret 重置:新 secret 下发(管理端展示一次,不回显存量) */ @Transactional public String resetSecret(Long schId, Long regId) { FlwCallbackReg reg = requireReg(schId, regId); String secret = UUID.randomUUID().toString().replace("-", "").toLowerCase(); reg.setSecret(secret); regRepository.save(reg); return secret; } /** 推送日志分页(状态/事件/办件筛选) */ public BasePageResult pageLogs(Long schId, Integer status, String event, Long instanceId, Integer page, Integer size) { Page p = logRepository.pageLogs(schId, status, event, instanceId, PageRequest.of(page == null || page < 1 ? 0 : page - 1, size == null || size < 1 || size > 100 ? 10 : size)); return new BasePageResult<>(p.getContent(), (int) p.getTotalElements(), page == null || page < 1 ? 1 : page, size == null || size < 1 || size > 100 ? 10 : size, p.getTotalPages(), p.hasNext()); } /** 死信列表 */ public BasePageResult pageDead(Long schId, Integer page, Integer size) { Page p = logRepository.findBySchIdAndStatusOrderByIdDesc(schId, FlwEnums.CallbackLogStatus.DEAD.getCode(), PageRequest.of(page == null || page < 1 ? 0 : page - 1, size == null || size < 1 || size > 100 ? 10 : size)); return new BasePageResult<>(p.getContent(), (int) p.getTotalElements(), page == null || page < 1 ? 1 : page, size == null || size < 1 || size > 100 ? 10 : size, p.getTotalPages(), p.hasNext()); } /** 挂起列表:待推且 retryCount≥挂起阈值(持续重试中疑似消费方停机的可视入口) */ public BasePageResult pageHung(Long schId, Integer page, Integer size) { Page p = logRepository.pageHung(schId, callbackProperties.getHangAlertRetries(), PageRequest.of(page == null || page < 1 ? 0 : page - 1, size == null || size < 1 || size > 100 ? 10 : size)); return new BasePageResult<>(p.getContent(), (int) p.getTotalElements(), page == null || page < 1 ? 1 : page, size == null || size < 1 || size > 100 ? 10 : size, p.getTotalPages(), p.hasNext()); } /** * 管理端手工标记死信(持续重试语义下死信的唯一人工入口:人工判定事件无意义时止血)。 * 标记后可经既有 rePush 复活。 */ @Transactional public void markDead(Long schId, Long logId) { FlwCallbackLog callbackLog = logRepository.findById(logId) .orElseThrow(() -> new BizException(ErrorCode.PARAM_ERROR, "推送日志不存在:" + logId)); if (!schId.equals(callbackLog.getSchId())) { throw new BizException(ErrorCode.FORBIDDEN, "推送日志不属于该学校"); } if (callbackLog.getStatus() != null && callbackLog.getStatus() == FlwEnums.CallbackLogStatus.SUCCESS.getCode()) { throw new BizException(ErrorCode.PARAM_ERROR, "已成功日志无需标记死信"); } pushService.markDead(callbackLog, "管理端手工标记死信"); } /** 手工重推(死信/失败日志):置回待推并立即触发一次推送 */ @Transactional public void rePush(Long schId, List logIds) { if (logIds == null || logIds.isEmpty()) { throw new BizException(ErrorCode.PARAM_ERROR, "请选择要重推的日志"); } List logs = logRepository.findByIdInAndSchId(logIds, schId); if (logs.size() != logIds.size()) { throw new BizException(ErrorCode.PARAM_ERROR, "存在不属于该校或不存在 dead/pending 日志"); } for (FlwCallbackLog callbackLog : logs) { if (callbackLog.getStatus() == FlwEnums.CallbackLogStatus.SUCCESS.getCode()) { continue; } callbackLog.setStatus(FlwEnums.CallbackLogStatus.PENDING.getCode()); callbackLog.setRetryCount(0); callbackLog.setNextRetryTime(LocalDateTime.now()); logRepository.save(callbackLog); } logRepository.flush(); for (FlwCallbackLog callbackLog : logs) { if (callbackLog.getStatus() != null) { pushService.pushAsync(callbackLog.getId()); } } } private FlwCallbackReg requireReg(Long schId, Long regId) { FlwCallbackReg reg = regRepository.findById(regId) .orElseThrow(() -> new BizException(ErrorCode.PARAM_ERROR, "回调注册不存在:" + regId)); if (!reg.getSchId().equals(schId)) { throw new BizException(ErrorCode.FORBIDDEN, "回调注册不属于该学校"); } return reg; } }