package com.yunhai.svchall.job; import com.yunhai.svchall.entity.ServiceApply; import com.yunhai.svchall.enums.ApplyFlowStatusEnum; import com.yunhai.svchall.repository.ServiceApplyRepository; import com.yunhai.svchall.service.FlowClient; import com.yunhai.svchall.service.SvcBizTypes; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; import java.time.LocalDateTime; import java.util.List; import java.util.Map; /** * 对账兜底 Job(降级为应急闸,V2.0 Part B.4#5 → 契约冻结 v1.4): * 升级后 flow-center 回调为持续重试(1m→5m→30m→2h→6h→60m 封顶)+ retryCount=6 挂起并 * UserMsg 告警恰一次 + 死信管理端手工处理——消费方免每日对账(本应用 5xx 语义已对齐), * 故本 Job 默认停用(svchall.flow.reconcile-enabled=false);仅在大面积漂移事故应急时置 true * 手工开闸。开启时语义不变:拉在途办件(审批中)→ 按 biz 键查 flow-center 真实状态 → * 漂移以 flow-center 为准修正(对账直改状态,不写 service_flow_event 去重表与结果明细—— * 与回调链路的差异,仅应急用不混淆镜像口径)。 */ @Slf4j @Component public class FlowReconcileJob { private final ServiceApplyRepository applyRepository; private final FlowClient flowClient; private final boolean reconcileEnabled; public FlowReconcileJob(ServiceApplyRepository applyRepository, FlowClient flowClient, @Value("${svchall.flow.reconcile-enabled:false}") boolean reconcileEnabled) { this.applyRepository = applyRepository; this.flowClient = flowClient; this.reconcileEnabled = reconcileEnabled; } @Scheduled(cron = "0 0 4 * * ?") public void reconcile() { if (!reconcileEnabled) { return; // 默认停用:免对账契约下推拉主链路由回调持续重试承担 } if (!flowClient.enabled()) { return; } // 全租户口径:审批中的在途办件(分页上限 500,量级评估:全校审批 << 千/周) List pending = applyRepository.findByFlowStatus(ApplyFlowStatusEnum.APPROVING.getCode(), org.springframework.data.domain.PageRequest.of(0, 500)); int fixed = 0; for (ServiceApply apply : pending) { if (apply.getFlowInstanceId() == null) { continue; } try { // 双值双查(P3.2 方案B):先终值后存量,首个命中实例为准——命中判据用内容 // (id/status 非空)而非"非 null",对 miss 形态(运行时 data=null / 空载体)都稳 Map remote = null; for (String bizType : SvcBizTypes.queryOrder()) { Map candidate = flowClient.getInstanceByBiz( apply.getSchId(), bizType, String.valueOf(apply.getId())); if (candidate != null && (candidate.get("id") != null || candidate.get("status") != null)) { remote = candidate; break; } } if (remote == null) { continue; } Integer remoteStatus = toInt(remote.get("status")); Integer remoteVersion = toInt(remote.get("version")); if (remoteStatus != null && !remoteStatus.equals(apply.getFlowStatus()) && (remoteVersion == null || remoteVersion > apply.getLastInstanceVersion())) { apply.setFlowStatus(remoteStatus); if (remoteVersion != null) { apply.setLastInstanceVersion(remoteVersion); } if (remoteStatus.equals(ApplyFlowStatusEnum.COMPLETED.getCode()) || remoteStatus.equals(ApplyFlowStatusEnum.REJECTED.getCode()) || remoteStatus.equals(ApplyFlowStatusEnum.WITHDRAWN.getCode())) { apply.setFinishTime(LocalDateTime.now()); } applyRepository.save(apply); fixed++; } } catch (Exception e) { log.warn("[svc-hall] 对账跳过 applyId={}:{}", apply.getId(), e.toString()); } } if (fixed > 0 || !pending.isEmpty()) { log.info("[svc-hall] 对账完成(应急闸开启中):在途 {} 条,修正 {} 条", pending.size(), fixed); } } private static Integer toInt(Object value) { if (value == null) { return null; } if (value instanceof Number) { return ((Number) value).intValue(); } try { return Integer.parseInt(String.valueOf(value)); } catch (NumberFormatException e) { return null; } } }