This commit is contained in:
@@ -0,0 +1,101 @@
|
||||
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.impl.FlowCallbackService;
|
||||
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 回调为<b>持续重试</b>(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<ServiceApply> 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 {
|
||||
Map<String, Object> remote = flowClient.getInstanceByBiz(
|
||||
apply.getSchId(), FlowCallbackService.BIZ_TYPE, String.valueOf(apply.getId()));
|
||||
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;
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user