svc-hall-service/src/main/java/com/yunhai/svchall/service/impl/SvcFlowCallbackHandler.java

package com.yunhai.svchall.service.impl;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.yunhai.platform.flowcallback.FlowCallbackHandler;
import com.yunhai.platform.openapi.flow.model.FlowCallbackEvent;
import com.yunhai.svchall.entity.ServiceApply;
import com.yunhai.svchall.enums.ApplyFlowStatusEnum;
import com.yunhai.svchall.repository.ServiceApplyRepository;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import java.time.LocalDateTime;
import java.util.HashMap;
import java.util.Map;

/**
 * 流程中心回调业务映射（2026-09-20 迁 yh-flow-callback-spring-boot-starter）：
 * X-Flw 验签（starter 可选过滤器）、ping、必填校验、bizType 过滤、eventId 幂等、
 * instanceVersion 乱序门由 starter 接收管线承担；本类只做办件镜像状态机推进——
 * 按 bizId 行锁定位 + 单据级乱序闸门 + 事件映射（submitted/task.*→10；completed→20；
 * rejected→30；withdrawn→40）+ 终态不回退守卫 + 结果明细快照 + 终态 finishTime。
 * 应答语义不变：本方法抛任何异常 → starter 回 5xx（flow-center 持续重试追平，免对账）；
 * 正常返回（含办件不存在/未知事件等不可重试场景）→ 200。
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class SvcFlowCallbackHandler implements FlowCallbackHandler {

    /** 本应用业务类型（flow-center apply/bizType 固定值，uk(sch_id, biz_type, biz_id)） */
    public static final String BIZ_TYPE = "svc_item";

    private final ServiceApplyRepository applyRepository;
    private final ObjectMapper objectMapper;

    @Override
    public void onEvent(FlowCallbackEvent event) {
        Integer applyId;
        try {
            applyId = Integer.valueOf(event.getBizId().trim());
        } catch (NumberFormatException e) {
            log.warn("[svc-hall] 回调 bizId 非法（丢弃）：{}", event.getBizId());
            return;
        }
        // 行锁串行化：并发推送（flow-center 推送池）对同一办件的处理排队，防丢更新
        ServiceApply apply = applyRepository.findByIdForUpdate(applyId).orElse(null);
        if (apply == null) {
            log.warn("[svc-hall] 回调找不到对应办件 bizId={} eventId={}（忽略）",
                    event.getBizId(), event.getEventId());
            return;
        }
        Integer version = event.getInstanceVersion();
        // 单据级乱序闸门（starter 表级闸门之外的第二道：以行锁后读到的办件版本为准）——
        // 严格小于已消费版本才丢弃（同版本多事件合法：submitted+task.created 同 v1）
        if (version != null && apply.getLastInstanceVersion() != null
                && version < apply.getLastInstanceVersion()) {
            log.info("[svc-hall] 回调乱序丢弃 instanceId={} incoming={} last={}",
                    event.getInstanceId(), version, apply.getLastInstanceVersion());
            return;
        }
        advance(apply, event.getEvent(), version, event);
        log.info("[svc-hall] 办件镜像状态推进 applyId={} instanceId={} event={} → status={}",
                applyId, event.getInstanceId(), event.getEvent(), apply.getFlowStatus());
    }

    /** 状态推进（幂等：同一事件重放由 eventId 去重拦截，此处只需正确覆盖目标态） */
    private void advance(ServiceApply apply, String event, Integer version, FlowCallbackEvent payload) {
        Integer status = mapStatus(event);
        if (status == null) {
            log.warn("[svc-hall] 未知回调事件类型：{}（忽略）", event);
            return;
        }
        // P1-3 终态不回退守卫：终态事件与最后一个 task.handled 同 instanceVersion，若 handled
        // 在 completed/rejected 之后落库（flow-center 推送池并发不保序），无守卫会把 20/30 打回 10
        // 并用 handled 明细覆盖已落定的 flow_result_json。重提不冲突：resubmit 成功后先以
        // 窄列原子置 10，新一轮事件到达时行已非终态。
        if (ApplyFlowStatusEnum.byCode(apply.getFlowStatus()).terminal()) {
            if (!ApplyFlowStatusEnum.byCode(status).terminal()) {
                log.warn("[svc-hall] 终态不回退：applyId={} flowStatus={} 拒绝非终态事件 {}(v{})",
                        apply.getId(), apply.getFlowStatus(), event, version);
                return;
            }
            if (apply.getFlowResultJson() != null && !apply.getFlowResultJson().isEmpty()) {
                log.warn("[svc-hall] 终态结果已落定：applyId={} flowStatus={} 拒绝覆盖 flow_result_json（事件 {}）",
                        apply.getId(), apply.getFlowStatus(), event);
                return;
            }
        }
        Map<String, Object> result = new HashMap<>();
        result.put("event", event);
        Map<String, Object> data = payload.getData();
        if (data != null) {
            copyIfPresent(data, result, "result");
            copyIfPresent(data, result, "opinion");
            copyIfPresent(data, result, "rejectNode");
        }
        try {
            apply.setFlowResultJson(objectMapper.writeValueAsString(result));
        } catch (Exception ignored) {
            // 明细序列化失败不影响状态推进
        }
        apply.setFlowStatus(status);
        if (version != null) {
            apply.setLastInstanceVersion(version);
        }
        if (status.equals(ApplyFlowStatusEnum.COMPLETED.getCode())
                || status.equals(ApplyFlowStatusEnum.REJECTED.getCode())
                || status.equals(ApplyFlowStatusEnum.WITHDRAWN.getCode())) {
            apply.setFinishTime(LocalDateTime.now());
        }
        applyRepository.save(apply);
    }

    private static void copyIfPresent(Map<String, Object> data, Map<String, Object> target, String key) {
        Object v = data.get(key);
        if (v != null) {
            target.put(key, String.valueOf(v));
        }
    }

    private static Integer mapStatus(String event) {
        switch (event) {
            case "flow.submitted":
            case "flow.task.created":
            case "flow.task.handled":
                return ApplyFlowStatusEnum.APPROVING.getCode();
            case "flow.completed":
                return ApplyFlowStatusEnum.COMPLETED.getCode();
            case "flow.rejected":
                return ApplyFlowStatusEnum.REJECTED.getCode();
            case "flow.withdrawn":
                return ApplyFlowStatusEnum.WITHDRAWN.getCode();
            default:
                return null;
        }
    }
}