feat(flow): 新增流程回调投递、草稿存储及流程可靠性服务

- 新增 FlowCallbackDelivery 类，实现 HTTP 回调的至少一次投递机制 - 新增 FlowDraftService 类，支持个人流程草稿的保存、读取和删除 - 新增 FlowReliabilityService 类，处理流程通知、催办与回调的持久化 - 删除 FlowWorkspaceTests 测试类，整合测试逻辑至其他模块 (cherry picked from commit 07c8ccc0e845d30dcfaeeb6d01c999460093c8ac)
2026-11-08 04:39:00 +08:00 · 2026-11-03 14:55:33 +08:00
1 parent d0aa3a7284
commit cf917b5988
@@ -1,113 +0,0 @@
-package com.pig4cloud.pigx.flow.constant;
-
-import cn.hutool.core.util.ObjectUtil;
-
-import java.util.EnumSet;
-import java.util.Objects;
-
-/**
- * 枚举通用接口
- * <p>
- * 为所有业务枚举提供统一的接口规范，实现枚举值和标签的标准化管理。
- * 提供了根据值获取枚举、根据值获取标签、根据标签获取值等通用方法。
- * 所有业务枚举类都应实现此接口，以保证枚举使用的一致性。
- * </p>
- *
- * @param <T> 枚举值的类型，通常为Integer或String
- * @author haoxr
- * @since 2022/3/27 12:06
- */
-public interface IBaseEnum<T> {
-
-	/**
-	 * 获取枚举值
-	 * <p>
-	 * 返回枚举对应的实际值，通常用于数据库存储或接口传输
-	 * </p>
-	 *
-	 * @return 枚举对应的值
-	 */
-	T getValue();
-
-	/**
-	 * 获取枚举标签
-	 * <p>
-	 * 返回枚举对应的显示标签，通常用于界面展示
-	 * </p>
-	 *
-	 * @return 枚举对应的中文标签
-	 */
-	String getLabel();
-
-	/**
-	 * 根据值获取对应的枚举实例
-	 * <p>
-	 * 通过枚举值查找对应的枚举实例，常用于数据库查询结果的转换。
-	 * 如果找不到匹配的枚举，返回null。
-	 * </p>
-	 *
-	 * @param value 枚举值
-	 * @param clazz 枚举类的Class对象
-	 * @param <E>   枚举类型，必须实现IBaseEnum接口
-	 * @return 匹配的枚举实例，如果没有找到返回null
-	 */
-	static <E extends Enum<E> & IBaseEnum> E getEnumByValue(Object value, Class<E> clazz) {
-		Objects.requireNonNull(value);
-		EnumSet<E> allEnums = EnumSet.allOf(clazz); // 获取类型下的所有枚举
-		E matchEnum = allEnums.stream().filter(e -> ObjectUtil.equal(e.getValue(), value)).findFirst().orElse(null);
-		return matchEnum;
-	}
-
-	/**
-	 * 根据值获取对应的标签
-	 * <p>
-	 * 通过枚举值直接获取对应的显示标签，常用于列表展示时的值转换。
-	 * 如果找不到匹配的枚举，返回null。
-	 * </p>
-	 *
-	 * @param value 枚举值
-	 * @param clazz 枚举类的Class对象
-	 * @param <E>   枚举类型，必须实现IBaseEnum接口
-	 * @return 枚举对应的标签，如果没有找到返回null
-	 */
-	static <E extends Enum<E> & IBaseEnum> String getLabelByValue(Object value, Class<E> clazz) {
-		Objects.requireNonNull(value);
-		EnumSet<E> allEnums = EnumSet.allOf(clazz); // 获取类型下的所有枚举
-		E matchEnum = allEnums.stream().filter(e -> ObjectUtil.equal(e.getValue(), value)).findFirst().orElse(null);
-
-		String label = null;
-		if (matchEnum != null) {
-			label = matchEnum.getLabel();
-		}
-		return label;
-	}
-
-	/**
-	 * 根据标签获取对应的值
-	 * <p>
-	 * 通过枚举标签反向查找对应的枚举值，常用于表单提交时的数据转换。
-	 * 如果找不到匹配的枚举，返回null。
-	 * </p>
-	 *
-	 * @param label 枚举标签
-	 * @param clazz 枚举类的Class对象
-	 * @param <E>   枚举类型，必须实现IBaseEnum接口
-	 * @return 枚举对应的值，如果没有找到返回null
-	 */
-	static <E extends Enum<E> & IBaseEnum> Object getValueByLabel(String label, Class<E> clazz) {
-		Objects.requireNonNull(label);
-		EnumSet<E> allEnums = EnumSet.allOf(clazz); // 获取类型下的所有枚举
-		String finalLabel = label;
-		E matchEnum = allEnums.stream()
-			.filter(e -> ObjectUtil.equal(e.getLabel(), finalLabel))
-			.findFirst()
-			.orElse(null);
-
-		Object value = null;
-		if (matchEnum != null) {
-			value = matchEnum.getValue();
-		}
-		return value;
-	}
-
-}
@@ -1,40 +0,0 @@
-package com.pig4cloud.pigx.flow.dto;
-
-import lombok.Data;
-
-import java.util.Map;
-
-/**
- * 节点HTTP请求结果数据传输对象
- * <p>
- * 该DTO用于封装流程节点执行HTTP请求后的返回结果。
- * 在流程执行过程中，某些节点可能需要调用外部系统的接口（如获取审批人、
- * 验证业务数据等），此对象用于统一封装这些HTTP请求的响应结果。
- * </p>
- * 
- * @author pigx
- */
-@Data
-public class NodeHttpResultVO {
-
-	/**
-	 * 请求是否成功
-	 * <p>
-	 * 标识HTTP请求是否执行成功：
-	 * - true: 请求成功，data字段包含返回的业务数据
-	 * - false: 请求失败，data字段可能包含错误信息
-	 * </p>
-	 */
-	private Boolean ok;
-
-	/**
-	 * 返回的数据内容
-	 * <p>
-	 * HTTP请求返回的业务数据，以Map形式存储，便于灵活处理各种数据结构。
-	 * 成功时包含业务数据，失败时可能包含错误码、错误信息等。
-	 * 具体内容取决于调用的接口定义。
-	 * </p>
-	 */
-	private Map<String, Object> data;
-
-}
@@ -1,8 +1,8 @@
 package com.pig4cloud.pigx.flow.config;

-import com.pig4cloud.pigx.flow.service.FlowCallbackDelivery;
+import com.pig4cloud.pigx.flow.support.workspace.FlowCallbackDelivery;
 import com.pig4cloud.pigx.flow.constant.FlowNoticeStatusEnum;
-import com.pig4cloud.pigx.flow.service.FlowReliabilityService;
+import com.pig4cloud.pigx.flow.support.workspace.FlowReliabilityService;
 import com.pig4cloud.pigx.flow.support.listeners.FlowDurableEventListener;
 import com.pig4cloud.pigx.flow.constant.FlowCallbackStatusEnum;
 import lombok.RequiredArgsConstructor;
@@ -3,8 +3,8 @@ package com.pig4cloud.pigx.flow.controller;
 import com.pig4cloud.pigx.common.core.util.R;
 import com.pig4cloud.pigx.common.data.tenant.TenantContextHolder;
 import com.pig4cloud.pigx.common.security.util.SecurityUtils;
-import com.pig4cloud.pigx.flow.service.FlowDraftService;
-import com.pig4cloud.pigx.flow.service.FlowReliabilityService;
+import com.pig4cloud.pigx.flow.support.workspace.FlowDraftService;
+import com.pig4cloud.pigx.flow.support.workspace.FlowReliabilityService;
 import com.pig4cloud.pigx.common.log.annotation.SysLog;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.tags.Tag;
@@ -1,6 +1,6 @@
 package com.pig4cloud.pigx.flow.support.listeners;

-import com.pig4cloud.pigx.flow.service.FlowReliabilityService;
+import com.pig4cloud.pigx.flow.support.workspace.FlowReliabilityService;
 import lombok.RequiredArgsConstructor;
 import org.flowable.common.engine.api.delegate.event.FlowableEntityEvent;
 import org.flowable.common.engine.api.delegate.event.FlowableEvent;
@@ -1,752 +0,0 @@
-package com.pig4cloud.pigx.flow.support.utils;
-
-import cn.hutool.core.collection.CollUtil;
-import cn.hutool.core.lang.Dict;
-import com.pig4cloud.pigx.common.security.util.SecurityUtils;
-import com.pig4cloud.pigx.flow.constant.NodeUserTypeEnum;
-import com.pig4cloud.pigx.flow.dto.BizFlowBaseDto;
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
-import org.flowable.bpmn.model.*;
-import org.flowable.engine.history.HistoricActivityInstance;
-import org.flowable.engine.impl.bpmn.behavior.ParallelMultiInstanceBehavior;
-import org.flowable.engine.impl.bpmn.behavior.SequentialMultiInstanceBehavior;
-
-import java.util.*;
-
-/**
- * Flowable流程引擎工具类
- * <p>
- * 提供流程模型遍历、节点查找、路径分析等功能的工具类。
- * 该类是流程引擎操作的核心工具，支持复杂的流程拓扑结构分析。
- * <p>
- * 主要功能：
- * 1. 流程元素的入口/出口连线获取
- * 2. 递归遍历流程定义获取所有元素
- * 3. 向前/向后查找用户任务节点
- * 4. 脏数据线路分析（用于回退场景）
- * 5. 串行/并行关系判断
- * 6. 历史数据清洗
- * <p>
- * 支持的流程结构：
- * - 顺序流程
- * - 并行网关
- * - 排他网关
- * - 子流程
- * - 多实例任务（会签）
- *
- * @author linjinp
- * @create 2019-12-24 13:51
- */
-public class FlowableUtils {
-
-    public static final Logger logger = LogManager.getLogger(FlowableUtils.class);
-
-    /**
-     * 根据节点获取入口连线集合
-     * <p>
-     * 支持多种流程元素类型，包括任务、网关、子流程、开始/结束事件等。
-     * 入口连线是指向该节点的所有连线。
-     *
-     * @param source 流程元素节点
-     * @return 入口连线集合，如果节点类型不支持则返回null
-     */
-    public static List<SequenceFlow> getElementIncomingFlows(FlowElement source) {
-        List<SequenceFlow> sequenceFlows = null;
-        if (source instanceof Task) {
-            sequenceFlows = ((Task) source).getIncomingFlows();
-        } else if (source instanceof Gateway) {
-            sequenceFlows = ((Gateway) source).getIncomingFlows();
-        } else if (source instanceof SubProcess) {
-            sequenceFlows = ((SubProcess) source).getIncomingFlows();
-        } else if (source instanceof StartEvent) {
-            sequenceFlows = ((StartEvent) source).getIncomingFlows();
-        } else if (source instanceof EndEvent) {
-            sequenceFlows = ((EndEvent) source).getIncomingFlows();
-        }
-        return sequenceFlows;
-    }
-
-    /**
-     * 根据节点获取出口连线集合
-     * <p>
-     * 支持多种流程元素类型，包括任务、网关、子流程、开始/结束事件等。
-     * 出口连线是从该节点出发的所有连线。
-     *
-     * @param source 流程元素节点
-     * @return 出口连线集合，如果节点类型不支持则返回null
-     */
-    public static List<SequenceFlow> getElementOutgoingFlows(FlowElement source) {
-        List<SequenceFlow> sequenceFlows = null;
-        if (source instanceof Task) {
-            sequenceFlows = ((Task) source).getOutgoingFlows();
-        } else if (source instanceof Gateway) {
-            sequenceFlows = ((Gateway) source).getOutgoingFlows();
-        } else if (source instanceof SubProcess) {
-            sequenceFlows = ((SubProcess) source).getOutgoingFlows();
-        } else if (source instanceof StartEvent) {
-            sequenceFlows = ((StartEvent) source).getOutgoingFlows();
-        } else if (source instanceof EndEvent) {
-            sequenceFlows = ((EndEvent) source).getOutgoingFlows();
-        }
-        return sequenceFlows;
-    }
-
-    /**
-     * 递归获取流程定义所有的流程元素
-     * <p>
-     * 深度优先遍历流程定义，包括子流程中的所有元素。
-     * 该方法常用于分析整个流程结构或查找特定元素。
-     *
-     * @param flowElements 当前级别流程元素集合
-     * @param allElements  用于存储的全部流程元素集合，可为null
-     * @return 全部流程元素集合，包含所有层级的元素
-     */
-    public static Collection<FlowElement> getAllElements(Collection<FlowElement> flowElements,
-                                                         Collection<FlowElement> allElements) {
-        allElements = allElements == null ? new ArrayList<>() : allElements;
-
-        for (FlowElement flowElement : flowElements) {
-            allElements.add(flowElement);
-            if (flowElement instanceof SubProcess) {
-                // 继续深入子流程，进一步获取子流程
-                allElements = FlowableUtils.getAllElements(((SubProcess) flowElement).getFlowElements(), allElements);
-            }
-        }
-        return allElements;
-    }
-
-    /**
-     * 从后向前递归获取父级用户任务节点
-     * <p>
-     * 从当前节点开始，沿着入口连线向前查找所有的用户任务节点。
-     * 主要用于流程回退、驳回等场景，找到可以回退的目标节点。
-     * 支持子流程的遍历。
-     *
-     * @param source          当前节点
-     * @param hasSequenceFlow 已处理过的顺序流集合，避免循环，可为null
-     * @param userTaskList    递归获取的用户任务节点集合，可为null
-     * @return 用户任务节点集合
-     */
-    public static List<UserTask> iteratorFindParentUserTasks(FlowElement source, Set<String> hasSequenceFlow,
-                                                             List<UserTask> userTaskList) {
-        userTaskList = userTaskList == null ? new ArrayList<>() : userTaskList;
-        hasSequenceFlow = hasSequenceFlow == null ? new HashSet<>() : hasSequenceFlow;
-
-        // 如果该节点为开始节点，且存在上级子节点，则顺着上级子节点继续迭代
-        if (source instanceof StartEvent && source.getSubProcess() != null) {
-            userTaskList = iteratorFindParentUserTasks(source.getSubProcess(), hasSequenceFlow, userTaskList);
-        }
-
-        // 根据类型，获取入口连线
-        List<SequenceFlow> sequenceFlows = getElementIncomingFlows(source);
-
-        if (sequenceFlows != null) {
-            // 循环找到目标元素
-            for (SequenceFlow sequenceFlow : sequenceFlows) {
-                // 如果发现连线重复，说明循环了，跳过这个循环
-                if (hasSequenceFlow.contains(sequenceFlow.getId())) {
-                    continue;
-                }
-                // 添加已经走过的连线
-                hasSequenceFlow.add(sequenceFlow.getId());
-                // 类型为用户节点，则新增父级节点
-                if (sequenceFlow.getSourceFlowElement() instanceof UserTask) {
-                    userTaskList.add((UserTask) sequenceFlow.getSourceFlowElement());
-                    continue;
-                }
-                // 类型为子流程，则添加子流程开始节点出口处相连的节点
-                if (sequenceFlow.getSourceFlowElement() instanceof SubProcess) {
-                    // 获取子流程用户任务节点
-                    List<UserTask> childUserTaskList = findChildProcessUserTasks(
-                            (StartEvent) ((SubProcess) sequenceFlow.getSourceFlowElement()).getFlowElements()
-                                    .toArray()[0],
-                            null, null);
-                    // 如果找到节点，则说明该线路找到节点，不继续向下找，反之继续
-                    if (childUserTaskList != null && childUserTaskList.size() > 0) {
-                        userTaskList.addAll(childUserTaskList);
-                        continue;
-                    }
-                }
-                // 继续迭代
-                // 注意：已经经过的节点与连线都应该用浅拷贝出来的对象
-                // 比如分支：a->b->c与a->d->c，走完a->b->c后走另一个路线是，已经经过的节点应该不包含a->b->c路线的数据
-                userTaskList = iteratorFindParentUserTasks(sequenceFlow.getSourceFlowElement(),
-                        new HashSet<>(hasSequenceFlow), userTaskList);
-            }
-        }
-        return userTaskList;
-    }
-
-    /**
-     * 根据正在运行的任务节点，迭代获取子级任务节点列表
-     * <p>
-     * 从当前节点开始，沿着出口连线向后查找正在运行的任务节点。
-     * 主要用于找到需要撤回或影响的下游节点。
-     *
-     * @param source          起始节点
-     * @param runActiveIdList 正在运行的任务Key，用于校验任务节点是否是正在运行的节点
-     * @param hasSequenceFlow 已经经过的连线的ID，用于判断线路是否重复
-     * @param flowElementList 需要撤回的用户任务列表
-     * @return 找到的流程元素列表
-     */
-    public static List<FlowElement> iteratorFindChildUserTasks(FlowElement source, List<String> runActiveIdList,
-                                                               Set<String> hasSequenceFlow, List<FlowElement> flowElementList) {
-        hasSequenceFlow = hasSequenceFlow == null ? new HashSet<>() : hasSequenceFlow;
-        flowElementList = flowElementList == null ? new ArrayList<>() : flowElementList;
-
-        // 如果该节点为开始节点，且存在上级子节点，则顺着上级子节点继续迭代
-        if (source instanceof EndEvent && source.getSubProcess() != null) {
-            flowElementList = iteratorFindChildUserTasks(source.getSubProcess(), runActiveIdList, hasSequenceFlow,
-                    flowElementList);
-        }
-
-        // 根据类型，获取出口连线
-        List<SequenceFlow> sequenceFlows = getElementOutgoingFlows(source);
-
-        if (sequenceFlows != null) {
-            // 循环找到目标元素
-            for (SequenceFlow sequenceFlow : sequenceFlows) {
-                // 如果发现连线重复，说明循环了，跳过这个循环
-                if (hasSequenceFlow.contains(sequenceFlow.getId())) {
-                    continue;
-                }
-                // 添加已经走过的连线
-                hasSequenceFlow.add(sequenceFlow.getId());
-                // 如果为用户任务类型，或者为网关
-                // 活动节点ID 在运行的任务中存在，添加
-                if ((sequenceFlow.getTargetFlowElement() instanceof UserTask
-                        || sequenceFlow.getTargetFlowElement() instanceof Gateway)
-                        && runActiveIdList.contains((sequenceFlow.getTargetFlowElement()).getId())) {
-                    flowElementList.add(sequenceFlow.getTargetFlowElement());
-                    continue;
-                }
-                // 如果节点为子流程节点情况，则从节点中的第一个节点开始获取
-                if (sequenceFlow.getTargetFlowElement() instanceof SubProcess) {
-                    List<FlowElement> childUserTaskList = iteratorFindChildUserTasks(
-                            (FlowElement) (((SubProcess) sequenceFlow.getTargetFlowElement()).getFlowElements()
-                                    .toArray()[0]),
-                            runActiveIdList, hasSequenceFlow, null);
-                    // 如果找到节点，则说明该线路找到节点，不继续向下找，反之继续
-                    if (childUserTaskList != null && childUserTaskList.size() > 0) {
-                        flowElementList.addAll(childUserTaskList);
-                        continue;
-                    }
-                }
-                // 继续迭代
-                // 注意：已经经过的节点与连线都应该用浅拷贝出来的对象
-                // 比如分支：a->b->c与a->d->c，走完a->b->c后走另一个路线是，已经经过的节点应该不包含a->b->c路线的数据
-                flowElementList = iteratorFindChildUserTasks(sequenceFlow.getTargetFlowElement(), runActiveIdList,
-                        new HashSet<>(hasSequenceFlow), flowElementList);
-            }
-        }
-        return flowElementList;
-    }
-
-    /**
-     * 迭代获取子流程用户任务节点
-     * <p>
-     * 专门用于处理子流程内部的用户任务节点查找。
-     * 从子流程的起始节点开始，找到所有的用户任务节点。
-     *
-     * @param source          起始节点
-     * @param hasSequenceFlow 已经经过的连线的ID，用于判断线路是否重复
-     * @param userTaskList    需要撤回的用户任务列表
-     * @return 用户任务节点列表
-     */
-    public static List<UserTask> findChildProcessUserTasks(FlowElement source, Set<String> hasSequenceFlow,
-                                                           List<UserTask> userTaskList) {
-        hasSequenceFlow = hasSequenceFlow == null ? new HashSet<>() : hasSequenceFlow;
-        userTaskList = userTaskList == null ? new ArrayList<>() : userTaskList;
-
-        // 根据类型，获取出口连线
-        List<SequenceFlow> sequenceFlows = getElementOutgoingFlows(source);
-
-        if (sequenceFlows != null) {
-            // 循环找到目标元素
-            for (SequenceFlow sequenceFlow : sequenceFlows) {
-                // 如果发现连线重复，说明循环了，跳过这个循环
-                if (hasSequenceFlow.contains(sequenceFlow.getId())) {
-                    continue;
-                }
-                // 添加已经走过的连线
-                hasSequenceFlow.add(sequenceFlow.getId());
-                // 如果为用户任务类型，且任务节点的 Key 正在运行的任务中存在，添加
-                if (sequenceFlow.getTargetFlowElement() instanceof UserTask) {
-                    userTaskList.add((UserTask) sequenceFlow.getTargetFlowElement());
-                    continue;
-                }
-                // 如果节点为子流程节点情况，则从节点中的第一个节点开始获取
-                if (sequenceFlow.getTargetFlowElement() instanceof SubProcess) {
-                    List<UserTask> childUserTaskList = findChildProcessUserTasks(
-                            (FlowElement) (((SubProcess) sequenceFlow.getTargetFlowElement()).getFlowElements()
-                                    .toArray()[0]),
-                            hasSequenceFlow, null);
-                    // 如果找到节点，则说明该线路找到节点，不继续向下找，反之继续
-                    if (childUserTaskList != null && childUserTaskList.size() > 0) {
-                        userTaskList.addAll(childUserTaskList);
-                        continue;
-                    }
-                }
-                // 继续迭代
-                // 注意：已经经过的节点与连线都应该用浅拷贝出来的对象
-                // 比如分支：a->b->c与a->d->c，走完a->b->c后走另一个路线是，已经经过的节点应该不包含a->b->c路线的数据
-                userTaskList = findChildProcessUserTasks(sequenceFlow.getTargetFlowElement(),
-                        new HashSet<>(hasSequenceFlow), userTaskList);
-            }
-        }
-        return userTaskList;
-    }
-
-    /**
-     * 从后向前寻路，获取所有脏线路上的点
-     * <p>
-     * 脏线路是指由于流程回退、驳回等操作产生的废弃路径。
-     * 该方法用于识别这些废弃路径上的所有节点，以便进行清理或过滤。
-     *
-     * @param source          起始节点
-     * @param passRoads       已经经过的点集合
-     * @param hasSequenceFlow 已经经过的连线的ID，用于判断线路是否重复
-     * @param targets         目标脏线路终点
-     * @param dirtyRoads      确定为脏数据的点，因为不需要重复，因此使用set存储
-     * @return 脏线路上的所有节点ID集合
-     */
-    public static Set<String> iteratorFindDirtyRoads(FlowElement source, List<String> passRoads,
-                                                     Set<String> hasSequenceFlow, List<String> targets, Set<String> dirtyRoads) {
-        passRoads = passRoads == null ? new ArrayList<>() : passRoads;
-        dirtyRoads = dirtyRoads == null ? new HashSet<>() : dirtyRoads;
-        hasSequenceFlow = hasSequenceFlow == null ? new HashSet<>() : hasSequenceFlow;
-
-        // 如果该节点为开始节点，且存在上级子节点，则顺着上级子节点继续迭代
-        if (source instanceof StartEvent && source.getSubProcess() != null) {
-            dirtyRoads = iteratorFindDirtyRoads(source.getSubProcess(), passRoads, hasSequenceFlow, targets,
-                    dirtyRoads);
-        }
-
-        // 根据类型，获取入口连线
-        List<SequenceFlow> sequenceFlows = getElementIncomingFlows(source);
-
-        if (sequenceFlows != null) {
-            // 循环找到目标元素
-            for (SequenceFlow sequenceFlow : sequenceFlows) {
-                // 如果发现连线重复，说明循环了，跳过这个循环
-                if (hasSequenceFlow.contains(sequenceFlow.getId())) {
-                    continue;
-                }
-                // 添加已经走过的连线
-                hasSequenceFlow.add(sequenceFlow.getId());
-                // 新增经过的路线
-                passRoads.add(sequenceFlow.getSourceFlowElement().getId());
-                // 如果此点为目标点，确定经过的路线为脏线路，添加点到脏线路中，然后找下个连线
-                if (targets.contains(sequenceFlow.getSourceFlowElement().getId())) {
-                    dirtyRoads.addAll(passRoads);
-                    continue;
-                }
-                // 如果该节点为开始节点，且存在上级子节点，则顺着上级子节点继续迭代
-                if (sequenceFlow.getSourceFlowElement() instanceof SubProcess) {
-                    dirtyRoads = findChildProcessAllDirtyRoad(
-                            (StartEvent) ((SubProcess) sequenceFlow.getSourceFlowElement()).getFlowElements()
-                                    .toArray()[0],
-                            null, dirtyRoads);
-                    // 是否存在子流程上，true 是，false 否
-                    Boolean isInChildProcess = dirtyTargetInChildProcess(
-                            (StartEvent) ((SubProcess) sequenceFlow.getSourceFlowElement()).getFlowElements()
-                                    .toArray()[0],
-                            null, targets, null);
-                    if (isInChildProcess) {
-                        // 已在子流程上找到，该路线结束
-                        continue;
-                    }
-                }
-                // 继续迭代
-                // 注意：已经经过的节点与连线都应该用浅拷贝出来的对象
-                // 比如分支：a->b->c与a->d->c，走完a->b->c后走另一个路线是，已经经过的节点应该不包含a->b->c路线的数据
-                dirtyRoads = iteratorFindDirtyRoads(sequenceFlow.getSourceFlowElement(), new ArrayList<>(passRoads),
-                        new HashSet<>(hasSequenceFlow), targets, dirtyRoads);
-            }
-        }
-        return dirtyRoads;
-    }
-
-    /**
-     * 迭代获取子流程脏路线
-     * <p>
-     * 当回退点是子流程时，子流程中的所有节点都被认为是脏路线。
-     * 该方法递归遍历子流程，获取其中的所有节点。
-     *
-     * @param source          起始节点
-     * @param hasSequenceFlow 已经经过的连线的ID，用于判断线路是否重复
-     * @param dirtyRoads      确定为脏数据的点，因为不需要重复，因此使用set存储
-     * @return 子流程中所有节点的ID集合
-     */
-    public static Set<String> findChildProcessAllDirtyRoad(FlowElement source, Set<String> hasSequenceFlow,
-                                                           Set<String> dirtyRoads) {
-        hasSequenceFlow = hasSequenceFlow == null ? new HashSet<>() : hasSequenceFlow;
-        dirtyRoads = dirtyRoads == null ? new HashSet<>() : dirtyRoads;
-
-        // 根据类型，获取出口连线
-        List<SequenceFlow> sequenceFlows = getElementOutgoingFlows(source);
-
-        if (sequenceFlows != null) {
-            // 循环找到目标元素
-            for (SequenceFlow sequenceFlow : sequenceFlows) {
-                // 如果发现连线重复，说明循环了，跳过这个循环
-                if (hasSequenceFlow.contains(sequenceFlow.getId())) {
-                    continue;
-                }
-                // 添加已经走过的连线
-                hasSequenceFlow.add(sequenceFlow.getId());
-                // 添加脏路线
-                dirtyRoads.add(sequenceFlow.getTargetFlowElement().getId());
-                // 如果节点为子流程节点情况，则从节点中的第一个节点开始获取
-                if (sequenceFlow.getTargetFlowElement() instanceof SubProcess) {
-                    dirtyRoads = findChildProcessAllDirtyRoad(
-                            (FlowElement) (((SubProcess) sequenceFlow.getTargetFlowElement()).getFlowElements()
-                                    .toArray()[0]),
-                            hasSequenceFlow, dirtyRoads);
-                }
-                // 继续迭代
-                // 注意：已经经过的节点与连线都应该用浅拷贝出来的对象
-                // 比如分支：a->b->c与a->d->c，走完a->b->c后走另一个路线是，已经经过的节点应该不包含a->b->c路线的数据
-                dirtyRoads = findChildProcessAllDirtyRoad(sequenceFlow.getTargetFlowElement(),
-                        new HashSet<>(hasSequenceFlow), dirtyRoads);
-            }
-        }
-        return dirtyRoads;
-    }
-
-    /**
-     * 判断脏路线结束节点是否在子流程上
-     * <p>
-     * 用于判断脏路线的范围，如果脏路线终点在子流程中，
-     * 说明脏路线只到子流程为止，不会影响父流程。
-     *
-     * @param source          起始节点
-     * @param hasSequenceFlow 已经经过的连线的ID，用于判断线路是否重复
-     * @param targets         判断脏路线节点是否存在子流程上，只要存在一个，说明脏路线只到子流程为止
-     * @param inChildProcess  是否存在子流程上，true是，false否
-     * @return 是否在子流程中
-     */
-    public static Boolean dirtyTargetInChildProcess(FlowElement source, Set<String> hasSequenceFlow,
-                                                    List<String> targets, Boolean inChildProcess) {
-        hasSequenceFlow = hasSequenceFlow == null ? new HashSet<>() : hasSequenceFlow;
-        inChildProcess = inChildProcess == null ? false : inChildProcess;
-
-        // 根据类型，获取出口连线
-        List<SequenceFlow> sequenceFlows = getElementOutgoingFlows(source);
-
-        if (sequenceFlows != null && !inChildProcess) {
-            // 循环找到目标元素
-            for (SequenceFlow sequenceFlow : sequenceFlows) {
-                // 如果发现连线重复，说明循环了，跳过这个循环
-                if (hasSequenceFlow.contains(sequenceFlow.getId())) {
-                    continue;
-                }
-                // 添加已经走过的连线
-                hasSequenceFlow.add(sequenceFlow.getId());
-                // 如果发现目标点在子流程上存在，说明只到子流程为止
-                if (targets.contains(sequenceFlow.getTargetFlowElement().getId())) {
-                    inChildProcess = true;
-                    break;
-                }
-                // 如果节点为子流程节点情况，则从节点中的第一个节点开始获取
-                if (sequenceFlow.getTargetFlowElement() instanceof SubProcess) {
-                    inChildProcess = dirtyTargetInChildProcess(
-                            (FlowElement) (((SubProcess) sequenceFlow.getTargetFlowElement()).getFlowElements()
-                                    .toArray()[0]),
-                            hasSequenceFlow, targets, inChildProcess);
-                }
-                // 继续迭代
-                // 注意：已经经过的节点与连线都应该用浅拷贝出来的对象
-                // 比如分支：a->b->c与a->d->c，走完a->b->c后走另一个路线是，已经经过的节点应该不包含a->b->c路线的数据
-                inChildProcess = dirtyTargetInChildProcess(sequenceFlow.getTargetFlowElement(),
-                        new HashSet<>(hasSequenceFlow), targets, inChildProcess);
-            }
-        }
-        return inChildProcess;
-    }
-
-    /**
-     * 迭代从后向前扫描，判断目标节点相对于当前节点是否是串行
-     * <p>
-     * 用于判断两个节点之间是否存在并行关系。
-     * 如果两个节点之间只有唯一路径，则为串行；如果存在多条路径，则为并行。
-     * 不存在直接回退到子流程中的情况，但存在从子流程出去到父流程情况。
-     *
-     * @param source          起始节点
-     * @param targetKsy       目标节点ID
-     * @param hasSequenceFlow 已经经过的连线的ID，用于判断线路是否重复
-     * @param isSequential    是否串行，默认为true
-     * @return true表示串行，false表示并行
-     */
-    public static Boolean iteratorCheckSequentialReferTarget(FlowElement source, String targetKsy,
-                                                             Set<String> hasSequenceFlow, Boolean isSequential) {
-        isSequential = isSequential == null ? true : isSequential;
-        hasSequenceFlow = hasSequenceFlow == null ? new HashSet<>() : hasSequenceFlow;
-
-        // 如果该节点为开始节点，且存在上级子节点，则顺着上级子节点继续迭代
-        if (source instanceof StartEvent && source.getSubProcess() != null) {
-            isSequential = iteratorCheckSequentialReferTarget(source.getSubProcess(), targetKsy, hasSequenceFlow,
-                    isSequential);
-        }
-
-        // 根据类型，获取入口连线
-        List<SequenceFlow> sequenceFlows = getElementIncomingFlows(source);
-
-        if (sequenceFlows != null) {
-            // 循环找到目标元素
-            for (SequenceFlow sequenceFlow : sequenceFlows) {
-                // 如果发现连线重复，说明循环了，跳过这个循环
-                if (hasSequenceFlow.contains(sequenceFlow.getId())) {
-                    continue;
-                }
-                // 添加已经走过的连线
-                hasSequenceFlow.add(sequenceFlow.getId());
-                // 如果目标节点已被判断为并行，后面都不需要执行，直接返回
-                if (isSequential == false) {
-                    break;
-                }
-                // 这条线路存在目标节点，这条线路完成，进入下个线路
-                if (targetKsy.equals(sequenceFlow.getSourceFlowElement().getId())) {
-                    continue;
-                }
-                if (sequenceFlow.getSourceFlowElement() instanceof StartEvent) {
-                    isSequential = false;
-                    break;
-                }
-                // 否则就继续迭代
-                // 注意：已经经过的节点与连线都应该用浅拷贝出来的对象
-                // 比如分支：a->b->c与a->d->c，走完a->b->c后走另一个路线是，已经经过的节点应该不包含a->b->c路线的数据
-                isSequential = iteratorCheckSequentialReferTarget(sequenceFlow.getSourceFlowElement(), targetKsy,
-                        new HashSet<>(hasSequenceFlow), isSequential);
-            }
-        }
-        return isSequential;
-    }
-
-    /**
-     * 从后向前寻路，获取到达节点的所有路线
-     * <p>
-     * 找到从当前节点到开始节点的所有可能路径。
-     * 每条路径是一个用户任务节点的有序列表。
-     * 不存在直接回退到子流程，但是存在回退到父级流程的情况。
-     *
-     * @param source          起始节点
-     * @param passRoads       已经经过的点集合
-     * @param hasSequenceFlow 已经经过的连线的ID，用于判断线路是否重复
-     * @param roads           路线集合
-     * @return 所有可能的路径列表
-     */
-    public static List<List<UserTask>> findRoad(FlowElement source, List<UserTask> passRoads,
-                                                Set<String> hasSequenceFlow, List<List<UserTask>> roads) {
-        passRoads = passRoads == null ? new ArrayList<>() : passRoads;
-        roads = roads == null ? new ArrayList<>() : roads;
-        hasSequenceFlow = hasSequenceFlow == null ? new HashSet<>() : hasSequenceFlow;
-
-        // 如果该节点为开始节点，且存在上级子节点，则顺着上级子节点继续迭代
-        if (source instanceof StartEvent && source.getSubProcess() != null) {
-            roads = findRoad(source.getSubProcess(), passRoads, hasSequenceFlow, roads);
-        }
-
-        // 根据类型，获取入口连线
-        List<SequenceFlow> sequenceFlows = getElementIncomingFlows(source);
-
-        if (sequenceFlows != null && sequenceFlows.size() != 0) {
-            for (SequenceFlow sequenceFlow : sequenceFlows) {
-                // 如果发现连线重复，说明循环了，跳过这个循环
-                if (hasSequenceFlow.contains(sequenceFlow.getId())) {
-                    continue;
-                }
-                // 添加已经走过的连线
-                hasSequenceFlow.add(sequenceFlow.getId());
-                // 添加经过路线
-                if (sequenceFlow.getSourceFlowElement() instanceof UserTask) {
-                    passRoads.add((UserTask) sequenceFlow.getSourceFlowElement());
-                }
-                // 继续迭代
-                // 注意：已经经过的节点与连线都应该用浅拷贝出来的对象
-                // 比如分支：a->b->c与a->d->c，走完a->b->c后走另一个路线是，已经经过的节点应该不包含a->b->c路线的数据
-                roads = findRoad(sequenceFlow.getSourceFlowElement(), new ArrayList<>(passRoads),
-                        new HashSet<>(hasSequenceFlow), roads);
-            }
-        } else {
-            // 添加路线
-            roads.add(passRoads);
-        }
-        return roads;
-    }
-
-    /**
-     * 历史节点数据清洗，清洗掉由于回滚导致的脏数据
-     * <p>
-     * 在流程回退、驳回等操作后，历史数据中会存在废弃的节点记录。
-     * 该方法通过分析删除原因和节点关系，识别并过滤掉这些脏数据，
-     * 返回真实有效的历史节点记录。
-     * <p>
-     * 特别处理：
-     * - 会签节点的多实例情况
-     * - 并行网关的多分支情况
-     * - 连续驳回的情况
-     *
-     * @param allElements            全部节点信息
-     * @param historicActivityIdList 历史任务实例信息，数据采用开始时间升序
-     * @return 清洗后的有效历史节点ID列表
-     */
-    public static List<String> historicTaskInstanceClean(Collection<FlowElement> allElements,
-                                                         List<HistoricActivityInstance> historicActivityIdList) {
-        // 会签节点收集
-        List<String> multiTask = new ArrayList<>();
-        allElements.forEach(flowElement -> {
-            if (flowElement instanceof UserTask) {
-                // 如果该节点的行为为会签行为，说明该节点为会签节点
-                if (((UserTask) flowElement).getBehavior() instanceof ParallelMultiInstanceBehavior
-                        || ((UserTask) flowElement).getBehavior() instanceof SequentialMultiInstanceBehavior) {
-                    multiTask.add(flowElement.getId());
-                }
-            }
-        });
-        // 循环放入栈，栈 LIFO：后进先出
-        Stack<HistoricActivityInstance> stack = new Stack<>();
-        historicActivityIdList.forEach(item -> stack.push(item));
-        // 清洗后的历史任务实例
-        List<String> lastHistoricTaskInstanceList = new ArrayList<>();
-        // 网关存在可能只走了部分分支情况，且还存在跳转废弃数据以及其他分支数据的干扰，因此需要对历史节点数据进行清洗
-        // 临时用户任务 key
-        StringBuilder userTaskKey = null;
-        // 临时被删掉的任务 key，存在并行情况
-        List<String> deleteKeyList = new ArrayList<>();
-        // 临时脏数据线路
-        List<Set<String>> dirtyDataLineList = new ArrayList<>();
-        // 由某个点跳到会签点,此时出现多个会签实例对应 1 个跳转情况，需要把这些连续脏数据都找到
-        // 会签特殊处理下标
-        int multiIndex = -1;
-        // 会签特殊处理 key
-        StringBuilder multiKey = null;
-        // 会签特殊处理操作标识
-        boolean multiOpera = false;
-        while (!stack.empty()) {
-            // 从这里开始 userTaskKey 都还是上个栈的 key
-            // 是否是脏数据线路上的点
-            final boolean[] isDirtyData = {false};
-            for (Set<String> oldDirtyDataLine : dirtyDataLineList) {
-                if (oldDirtyDataLine.contains(stack.peek().getActivityId())) {
-                    isDirtyData[0] = true;
-                }
-            }
-            // 删除原因不为空，说明从这条数据开始回跳或者回退的
-            // MI_END：会签完成后，其他未签到节点的删除原因，不在处理范围内
-            if (stack.peek().getDeleteReason() != null && !stack.peek().getDeleteReason().equals("MI_END")) {
-                // 可以理解为脏线路起点
-                String dirtyPoint = "";
-                if (stack.peek().getDeleteReason().indexOf("Change activity to ") >= 0) {
-                    dirtyPoint = stack.peek().getDeleteReason().replace("Change activity to ", "");
-                }
-                // 会签回退删除原因有点不同
-                if (stack.peek().getDeleteReason().indexOf("Change parent activity to ") >= 0) {
-                    dirtyPoint = stack.peek().getDeleteReason().replace("Change parent activity to ", "");
-                }
-                FlowElement dirtyTask = null;
-                // 获取变更节点的对应的入口处连线
-                // 如果是网关并行回退情况，会变成两条脏数据路线，效果一样
-                for (FlowElement flowElement : allElements) {
-                    if (flowElement.getId().equals(stack.peek().getActivityId())) {
-                        dirtyTask = flowElement;
-                    }
-                }
-                // 获取脏数据线路
-                Set<String> dirtyDataLine = FlowableUtils.iteratorFindDirtyRoads(dirtyTask, null, null,
-                        Arrays.asList(dirtyPoint.split(",")), null);
-                // 自己本身也是脏线路上的点，加进去
-                dirtyDataLine.add(stack.peek().getActivityId());
-                logger.info(stack.peek().getActivityId() + "点脏路线集合：" + dirtyDataLine);
-                // 是全新的需要添加的脏线路
-                boolean isNewDirtyData = true;
-                for (int i = 0; i < dirtyDataLineList.size(); i++) {
-                    // 如果发现他的上个节点在脏线路内，说明这个点可能是并行的节点，或者连续驳回
-                    // 这时，都以之前的脏线路节点为标准，只需合并脏线路即可，也就是路线补全
-                    if (dirtyDataLineList.get(i).contains(userTaskKey.toString())) {
-                        isNewDirtyData = false;
-                        dirtyDataLineList.get(i).addAll(dirtyDataLine);
-                    }
-                }
-                // 已确定时全新的脏线路
-                if (isNewDirtyData) {
-                    // deleteKey 单一路线驳回到并行，这种同时生成多个新实例记录情况，这时 deleteKey 其实是由多个值组成
-                    // 按照逻辑，回退后立刻生成的实例记录就是回退的记录
-                    // 至于驳回所生成的 Key，直接从删除原因中获取，因为存在驳回到并行的情况
-                    deleteKeyList.add(dirtyPoint + ",");
-                    dirtyDataLineList.add(dirtyDataLine);
-                }
-                // 添加后，现在这个点变成脏线路上的点了
-                isDirtyData[0] = true;
-            }
-            // 如果不是脏线路上的点，说明是有效数据，添加历史实例 Key
-            if (!isDirtyData[0]) {
-                lastHistoricTaskInstanceList.add(stack.peek().getActivityId());
-            }
-            // 校验脏线路是否结束
-            for (int i = 0; i < deleteKeyList.size(); i++) {
-                // 如果发现脏数据属于会签，记录下下标与对应 Key，以备后续比对，会签脏数据范畴开始
-                if (multiKey == null && multiTask.contains(stack.peek().getActivityId())
-                        && deleteKeyList.get(i).contains(stack.peek().getActivityId())) {
-                    multiIndex = i;
-                    multiKey = new StringBuilder(stack.peek().getActivityId());
-                }
-                // 会签脏数据处理，节点退回会签清空
-                // 如果在会签脏数据范畴中发现 Key改变，说明会签脏数据在上个节点就结束了，可以把会签脏数据删掉
-                if (multiKey != null && !multiKey.toString().equals(stack.peek().getActivityId())) {
-                    deleteKeyList.set(multiIndex,
-                            deleteKeyList.get(multiIndex).replace(stack.peek().getActivityId() + ",", ""));
-                    multiKey = null;
-                    // 结束进行下校验删除
-                    multiOpera = true;
-                }
-                // 其他脏数据处理
-                // 发现该路线最后一条脏数据，说明这条脏数据线路处理完了，删除脏数据信息
-                // 脏数据产生的新实例中是否包含这条数据
-                if (multiKey == null && deleteKeyList.get(i).contains(stack.peek().getActivityId())) {
-                    // 删除匹配到的部分
-                    deleteKeyList.set(i, deleteKeyList.get(i).replace(stack.peek().getActivityId() + ",", ""));
-                }
-                // 如果每组中的元素都以匹配过，说明脏数据结束
-                if ("".equals(deleteKeyList.get(i))) {
-                    // 同时删除脏数据
-                    deleteKeyList.remove(i);
-                    dirtyDataLineList.remove(i);
-                    break;
-                }
-            }
-            // 会签数据处理需要在循环外处理，否则可能导致溢出
-            // 会签的数据肯定是之前放进去的所以理论上不会溢出，但还是校验下
-            if (multiOpera && deleteKeyList.size() > multiIndex && "".equals(deleteKeyList.get(multiIndex))) {
-                // 同时删除脏数据
-                deleteKeyList.remove(multiIndex);
-                dirtyDataLineList.remove(multiIndex);
-                multiIndex = -1;
-                multiOpera = false;
-            }
-            // pop() 方法与 peek() 方法不同，在返回值的同时，会把值从栈中移除
-            // 保存新的 userTaskKey 在下个循环中使用
-            userTaskKey = new StringBuilder(stack.pop().getActivityId());
-        }
-        logger.info("清洗后的历史节点数据：" + lastHistoricTaskInstanceList);
-        return lastHistoricTaskInstanceList;
-    }
-
-    /**
-     * 获取流程参数Map，并设置发起人信息
-     *
-     * @param flowBaseDto 流程基础数据传输对象
-     * @return 包含发起人信息的流程参数Map
-     */
-    public static Map<String, Object> getFlowParamMap(BizFlowBaseDto flowBaseDto) {
-        Map<String, Object> paramMap = flowBaseDto.getFlowParamMap();
-        // 设置发起人 (当前登录用户)
-        Dict rootUser = Dict.create()
-                .set("id", SecurityUtils.getUser().getId())
-                .set("name", SecurityUtils.getUser().getUsername())
-                .set("type", NodeUserTypeEnum.USER.getKey());
-        paramMap.put("root", CollUtil.newArrayList(rootUser));
-        return paramMap;
-    }
-}
@@ -1,4 +1,4 @@
-package com.pig4cloud.pigx.flow.service;
+package com.pig4cloud.pigx.flow.support.workspace;

 import cn.hutool.core.util.StrUtil;
 import cn.hutool.http.HttpRequest;
@@ -1,4 +1,4 @@
-package com.pig4cloud.pigx.flow.service;
+package com.pig4cloud.pigx.flow.support.workspace;

 import cn.hutool.core.util.StrUtil;
 import cn.hutool.json.JSONUtil;
@@ -1,4 +1,6 @@
-package com.pig4cloud.pigx.flow.service;
+package com.pig4cloud.pigx.flow.support.workspace;
+
+import com.pig4cloud.pigx.flow.service.IProcessNodeDataService;

 import cn.hutool.core.convert.Convert;
 import cn.hutool.core.util.StrUtil;
@@ -1,297 +0,0 @@
-package com.pig4cloud.pigx.flow.service;
-
-import com.pig4cloud.pigx.common.core.util.R;
-import com.pig4cloud.pigx.admin.api.dto.MessageNoticeDTO;
-import com.pig4cloud.pigx.admin.api.feign.RemoteMessageService;
-import com.pig4cloud.pigx.common.data.tenant.TenantContextHolder;
-import cn.hutool.json.JSONUtil;
-import com.pig4cloud.pigx.flow.dto.Node;
-import com.pig4cloud.pigx.flow.constant.FlowNoticeStatusEnum;
-import com.pig4cloud.pigx.flow.config.FlowDurableConfiguration;
-import com.pig4cloud.pigx.flow.support.listeners.FlowDurableEventListener;
-import com.sun.net.httpserver.HttpServer;
-import org.flowable.engine.ProcessEngine;
-import org.flowable.engine.ProcessEngineConfiguration;
-import org.flowable.spring.SpringProcessEngineConfiguration;
-import org.h2.jdbcx.JdbcDataSource;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.Test;
-import org.springframework.aop.framework.ProxyFactory;
-import org.springframework.beans.factory.ObjectProvider;
-import org.springframework.jdbc.core.JdbcTemplate;
-import org.springframework.jdbc.datasource.DataSourceTransactionManager;
-import org.springframework.transaction.annotation.AnnotationTransactionAttributeSource;
-import org.springframework.transaction.interceptor.TransactionInterceptor;
-import org.springframework.transaction.support.TransactionTemplate;
-import java.net.InetSocketAddress;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.UUID;
-import java.util.concurrent.atomic.AtomicInteger;
-import static org.assertj.core.api.Assertions.*;
-import static org.mockito.Mockito.*;
-
-class FlowWorkspaceTests {
-
-	private ProcessEngine engine;
-
-	private JdbcTemplate jdbc;
-
-	private TransactionTemplate transaction;
-
-	private FlowReliabilityService reliability;
-
-	private FlowDraftService drafts;
-
-	private RemoteMessageService messages;
-
-	private final List<MessageNoticeDTO> sentNotices = new ArrayList<>();
-
-	private HttpServer server;
-
-	private final List<String> eventIds = new ArrayList<>();
-
-	private final AtomicInteger requests = new AtomicInteger();
-
-	@BeforeEach
-	void setUp() throws Exception {
-		JdbcDataSource source = new JdbcDataSource();
-		source.setURL("jdbc:h2:mem:" + UUID.randomUUID() + ";DB_CLOSE_DELAY=-1;DATABASE_TO_LOWER=TRUE");
-		source.setUser("sa");
-		jdbc = new JdbcTemplate(source);
-		jdbc.execute("SET MODE MySQL");
-		String schema = Files.readString(Path.of("../../db/migration/20261014_flow_first_batch.sql"));
-		for (String sql : schema.split(";"))
-			if (!sql.isBlank())
-				jdbc.execute(sql);
-		jdbc.execute(
-				"CREATE TABLE process_instance_record(id bigint primary key,process_instance_id varchar(64),tenant_id bigint,user_id bigint,status int,del_flag char(1),name varchar(200))");
-		var manager = new DataSourceTransactionManager(source);
-		transaction = new TransactionTemplate(manager);
-		var configuration = new SpringProcessEngineConfiguration();
-		configuration.setDataSource(source);
-		configuration.setTransactionManager(manager);
-		configuration.setDatabaseSchemaUpdate(ProcessEngineConfiguration.DB_SCHEMA_UPDATE_TRUE);
-		configuration.setDisableIdmEngine(true);
-		configuration.setDisableEventRegistry(true);
-		configuration.setAsyncExecutorActivate(false);
-		ObjectProvider<FlowReliabilityService> provider = mock(ObjectProvider.class);
-		when(provider.getObject()).thenAnswer(invocation -> reliability);
-		configuration.setEventListeners(List.of(new FlowDurableEventListener(provider)));
-		jdbc.execute("SET MODE LEGACY");
-		engine = configuration.buildProcessEngine();
-		jdbc.execute("SET MODE MySQL");
-		server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
-		server.createContext("/callback", exchange -> {
-			synchronized (eventIds) {
-				eventIds.add(exchange.getRequestHeaders().getFirst("X-Flow-Event-Id"));
-			}
-			exchange.getRequestBody().readAllBytes();
-			exchange.sendResponseHeaders(requests.incrementAndGet() == 1 ? 503 : 204, -1);
-			exchange.close();
-		});
-		server.start();
-		Node node = new Node();
-		node.setReminderMinutes(1);
-		node.setEventConfig("http://127.0.0.1:" + server.getAddress().getPort() + "/callback");
-		var nodes = mock(IProcessNodeDataService.class);
-		when(nodes.getPublishedNode(anyString(), anyString())).thenReturn(R.ok(node));
-		messages = mock(RemoteMessageService.class);
-		when(messages.sendNotice(any())).thenAnswer(invocation -> {
-			assertThat(TenantContextHolder.getTenantId()).isEqualTo(1L);
-			sentNotices.add(invocation.getArgument(0));
-			return R.ok(123L);
-		});
-		var target = new FlowReliabilityService(jdbc, nodes, engine.getTaskService(), messages);
-		var proxy = new ProxyFactory(target);
-		proxy.addAdvice(new TransactionInterceptor(manager, new AnnotationTransactionAttributeSource()));
-		reliability = (FlowReliabilityService) proxy.getProxy();
-		drafts = new FlowDraftService(jdbc);
-		engine.getRepositoryService()
-			.createDeployment()
-			.tenantId("1")
-			.addString("test.bpmn20.xml",
-					"""
-							<?xml version="1.0" encoding="UTF-8"?>
-							<definitions xmlns="http://www.omg.org/spec/BPMN/20100524/MODEL" xmlns:flowable="http://flowable.org/bpmn" targetNamespace="test">
-							 <process id="approval" isExecutable="true">
-							  <startEvent id="start"/><sequenceFlow id="s1" sourceRef="start" targetRef="approve"/>
-							  <userTask id="approve" name="审批[51421]" flowable:assignee="1001"/><sequenceFlow id="s2" sourceRef="approve" targetRef="end"/><endEvent id="end"/>
-							 </process>
-							</definitions>
-							""")
-			.deploy();
-	}
-
-	@AfterEach
-	void tearDown() {
-		if (server != null)
-			server.stop(0);
-		if (engine != null)
-			engine.close();
-		if (jdbc != null)
-			jdbc.execute("SHUTDOWN");
-	}
-
-	@Test
-	void draftsAndTaskEventsSurviveNewServiceInstancesAndHttpRetries() throws Exception {
-		// 请求仍能读取个人表单，但操作日志使用的 Jackson 不得输出标题和载荷。
-		var logMapper = com.pig4cloud.pigx.common.log.util.JacksonSensitiveFieldUtil.getObjectMapper().copy();
-		logMapper.setFilterProvider(new com.fasterxml.jackson.databind.ser.impl.SimpleFilterProvider()
-			.setDefaultFilter(com.fasterxml.jackson.databind.ser.impl.SimpleBeanPropertyFilter.serializeAll()));
-		var input = logMapper.readValue(
-				"{\"kind\":\"APPLICATION\",\"flowId\":\"approval\",\"title\":\"private-title\",\"payload\":\"{\\\"secret\\\":1}\",\"revision\":0}",
-				FlowDraftService.Draft.class);
-		assertThat(input.title()).isEqualTo("private-title");
-		assertThat(input.payload()).contains("secret");
-		assertThat(logMapper.writeValueAsString(new Object[] { input })).doesNotContain("private-title", "secret",
-				"payload", "title");
-		assertThat(FlowNoticeStatusEnum.values()).extracting(FlowNoticeStatusEnum::getCode)
-			.containsExactly("PENDING", "READY", "SENT", "CANCELLED", "DISABLED");
-		var design = drafts.save(1, 7,
-				new FlowDraftService.Draft(null, "DESIGN", "approval", "设计", "{\"step1\":{}}", 0));
-		var application = drafts.save(1, 7,
-				new FlowDraftService.Draft(null, "APPLICATION", "approval", "申请", "{\"formData\":{\"amount\":8}}", 0));
-		assertThat(new FlowDraftService(jdbc).get(1, 7, design.get("id").toString()).get("payload"))
-			.isEqualTo("{\"step1\":{}}");
-		assertThat(drafts.list(1, 7, "APPLICATION", 1, null)).hasSize(1);
-		assertThat(drafts.list(1, 7, "APPLICATION", 1, "approval")).hasSize(1);
-		assertThat(drafts.list(1, 7, "APPLICATION", 1, "other-flow")).isEmpty();
-		assertThat(drafts.list(1, 8, "APPLICATION", 1, "approval")).isEmpty();
-		assertThat(drafts.get(1, 7, application.get("id").toString()).get("payload"))
-			.isEqualTo("{\"formData\":{\"amount\":8}}");
-		var instance = engine.getRuntimeService()
-			.startProcessInstanceByKeyAndTenantId("approval", java.util.Map.of(), "1");
-		jdbc.update("INSERT INTO process_instance_record VALUES(1,?,1,7,1,'0','差旅申请')", instance.getId());
-		var task = engine.getTaskService().createTaskQuery().singleResult();
-		verifyNoInteractions(messages);
-		String notice = jdbc.queryForObject("SELECT id FROM flow_notice WHERE status='READY'", String.class);
-		reliability.deliverNotice(notice);
-		reliability.deliverNotice(notice);
-		assertThat(sentNotices).hasSize(1);
-		assertThat(sentNotices.get(0).getUserIds()).containsExactly(1001L);
-		assertThat(sentNotices.get(0).getBusinessType()).isEqualTo("FLOW_TASK");
-		assertThat(sentNotices.get(0).getBusinessId()).isEqualTo(task.getId());
-		assertThat(sentNotices.get(0).getRequestKey()).isEqualTo("flow:" + notice);
-		assertThat(sentNotices.get(0).getCategory()).isEqualTo("1");
-		assertThat(sentNotices.get(0).getAllFlag()).isEqualTo("0");
-		assertThat(jdbc.queryForObject("SELECT count(*) FROM flow_notice WHERE status='PENDING'", Integer.class))
-			.isEqualTo(1);
-		engine.getRuntimeService().suspendProcessInstanceById(instance.getId());
-		jdbc.update(
-				"UPDATE flow_notice SET scheduled_at=DATEADD('MINUTE',-1,CURRENT_TIMESTAMP) WHERE status='PENDING'");
-		reliability.remind(1, task.getId());
-		assertThat(
-				jdbc.queryForObject("SELECT status FROM flow_notice WHERE reminder_key=?", String.class, task.getId()))
-			.isEqualTo("PENDING");
-		assertThat(jdbc.queryForObject("SELECT scheduled_at>CURRENT_TIMESTAMP(3) FROM flow_notice WHERE reminder_key=?",
-				Boolean.class, task.getId()))
-			.isTrue();
-		engine.getRuntimeService().activateProcessInstanceById(instance.getId());
-		engine.getTaskService().setAssignee(task.getId(), "2002");
-		jdbc.update(
-				"UPDATE flow_notice SET scheduled_at=DATEADD('MINUTE',-1,CURRENT_TIMESTAMP) WHERE status='PENDING'");
-		ObjectProvider<FlowReliabilityService> provider = mock(ObjectProvider.class);
-		when(provider.getObject()).thenReturn(reliability);
-		var scheduler = new FlowDurableConfiguration(jdbc, provider, mock(ObjectProvider.class));
-		scheduler.scanReminders();
-		scheduler.scanReminders();
-		assertThat(jdbc.queryForObject("SELECT count(*) FROM flow_notice WHERE tenant_id=1 AND user_id='2002'", Integer.class)).isEqualTo(2);
-		engine.getTaskService().setAssignee(task.getId(), "1001");
-		assertThat(jdbc.queryForObject("SELECT count(*) FROM flow_notice WHERE tenant_id=1 AND user_id='1001'", Integer.class)).isEqualTo(2);
-		engine.getTaskService().setAssignee(task.getId(), "2002");
-		assertThat(reliability.urge(1, 7, instance.getId())).isEqualTo(1);
-		assertThatThrownBy(() -> reliability.urge(1, 7, instance.getId())).hasMessageContaining("十分钟");
-		scheduler.scanNotices();
-		assertThat(sentNotices).extracting(MessageNoticeDTO::getTitle)
-			.anyMatch(title -> title.startsWith("待办:"))
-			.anyMatch(title -> title.equals("催办:差旅申请 · 审批"))
-			.anyMatch(title -> title.startsWith("超时提醒:"));
-		assertThat(sentNotices.stream().filter(message -> message.getTitle().startsWith("超时提醒:"))
-			.map(MessageNoticeDTO::getUserIds).toList()).containsExactly(List.of(2002L));
-		engine.getTaskService().complete(task.getId());
-		assertThat(jdbc.queryForObject("SELECT count(*) FROM flow_notice WHERE status='PENDING'", Integer.class))
-			.isZero();
-		var callback = reliability.callbacks(1, 7, instance.getId()).get(0);
-		String id = callback.get("id").toString();
-		new FlowCallbackDelivery(jdbc, transaction).deliver(id);
-		assertThat(jdbc.queryForObject("SELECT status FROM flow_callback", String.class)).isEqualTo("PENDING");
-		jdbc.update("UPDATE flow_callback SET next_at=CURRENT_TIMESTAMP");
-		new FlowCallbackDelivery(jdbc, transaction).deliver(id);
-		assertThat(jdbc.queryForObject("SELECT status FROM flow_callback", String.class)).isEqualTo("DELIVERED");
-		var history = JSONUtil.parseArray(jdbc.queryForObject("SELECT history FROM flow_callback", String.class));
-		assertThat(history).hasSize(2);
-		assertThat(history.getJSONObject(0).getStr("result")).isEqualTo("ERROR");
-		assertThat(history.getJSONObject(1).getStr("result")).isEqualTo("DELIVERED");
-		jdbc.update("UPDATE flow_callback SET status='FAILED'");
-		reliability.retry(1, 7, id);
-		history = JSONUtil.parseArray(jdbc.queryForObject("SELECT history FROM flow_callback", String.class));
-		assertThat(history).hasSize(3);
-		assertThat(history.getJSONObject(2).getStr("result")).isEqualTo("REQUEUED");
-		new FlowCallbackDelivery(jdbc, transaction).deliver(id);
-		history = JSONUtil.parseArray(jdbc.queryForObject("SELECT history FROM flow_callback", String.class));
-		assertThat(history).hasSize(4);
-		assertThat(history.getJSONObject(0).getStr("result")).isEqualTo("ERROR");
-		assertThat(history.getJSONObject(3).getStr("result")).isEqualTo("DELIVERED");
-		assertThat(eventIds).containsExactly(id, id, id);
-		engine.getRuntimeService().startProcessInstanceByKeyAndTenantId("approval", java.util.Map.of(), "1");
-		String completed = engine.getTaskService().createTaskQuery().singleResult().getId();
-		engine.getTaskService().complete(completed);
-		jdbc.update("UPDATE flow_notice SET scheduled_at=DATEADD('MINUTE',-1,CURRENT_TIMESTAMP) WHERE reminder_key=?",
-				completed);
-		reliability.remind(1, completed);
-		assertThat(jdbc.queryForObject("SELECT status FROM flow_notice WHERE reminder_key=?", String.class, completed))
-			.isEqualTo("CANCELLED");
-		assertThat(jdbc.queryForObject("SELECT count(*) FROM flow_notice WHERE reminder_key=? AND status='SENT'",
-				Integer.class, completed))
-			.isZero();
-	}
-
-	@Test
-	void ownershipConflictsAndStorageFailuresCannotLoseCommittedWork() {
-		var saved = drafts.save(1, 7, new FlowDraftService.Draft(null, "APPLICATION", "approval", "申请", "{}", 0));
-		String id = saved.get("id").toString();
-		assertThatThrownBy(() -> drafts.get(2, 7, id)).hasMessageContaining("无权");
-		assertThatThrownBy(() -> drafts.get(1, 8, id)).hasMessageContaining("无权");
-		assertThatThrownBy(
-				() -> drafts.save(1, 7, new FlowDraftService.Draft(id, "APPLICATION", "approval", "申请", "{}", 0)))
-			.hasMessageContaining("重新打开");
-		assertThat(drafts.delete(1, 8, id, 1)).isFalse();
-		assertThatThrownBy(() -> transaction.executeWithoutResult(tx -> {
-			engine.getRuntimeService().startProcessInstanceByKeyAndTenantId("approval", java.util.Map.of(), "1");
-			throw new IllegalStateException("rollback");
-		})).hasMessage("rollback");
-		assertThat(jdbc.queryForObject("SELECT count(*) FROM flow_notice", Integer.class)).isZero();
-		assertThat(engine.getTaskService().createTaskQuery().count()).isZero();
-		verifyNoInteractions(messages);
-		var instance = engine.getRuntimeService()
-			.startProcessInstanceByKeyAndTenantId("approval", java.util.Map.of(), "1");
-		jdbc.update("INSERT INTO process_instance_record VALUES(1,?,1,7,1,'0','差旅申请')", instance.getId());
-		assertThatThrownBy(() -> reliability.urge(2, 7, instance.getId())).hasMessageContaining("无权");
-		assertThatThrownBy(() -> reliability.callbacks(1, 8, instance.getId())).hasMessageContaining("无权");
-		String notice = jdbc.queryForObject("SELECT id FROM flow_notice WHERE status='READY'", String.class);
-		doReturn(R.failed("发送暂不可用"), R.ok(123L)).when(messages).sendNotice(any());
-		ObjectProvider<FlowReliabilityService> provider = mock(ObjectProvider.class);
-		when(provider.getObject()).thenReturn(reliability);
-		var scheduler = new FlowDurableConfiguration(jdbc, provider, mock(ObjectProvider.class));
-		scheduler.scanNotices();
-		assertThat(jdbc.queryForObject("SELECT status FROM flow_notice WHERE id=?", String.class, notice)).isEqualTo("READY");
-		assertThat(jdbc.queryForObject("SELECT scheduled_at>CURRENT_TIMESTAMP(3) FROM flow_notice WHERE id=?", Boolean.class, notice)).isTrue();
-		jdbc.update("UPDATE flow_notice SET scheduled_at=CURRENT_TIMESTAMP(3) WHERE id=?", notice);
-		scheduler.scanNotices();
-		assertThat(jdbc.queryForObject("SELECT status FROM flow_notice WHERE id=?", String.class, notice)).isEqualTo("SENT");
-		var captor = org.mockito.ArgumentCaptor.forClass(MessageNoticeDTO.class);
-		verify(messages, times(2)).sendNotice(captor.capture());
-		assertThat(captor.getAllValues()).extracting(MessageNoticeDTO::getRequestKey).containsExactly("flow:" + notice, "flow:" + notice);
-		String task = engine.getTaskService().createTaskQuery().singleResult().getId();
-		jdbc.execute("DROP TABLE flow_callback");
-		assertThatThrownBy(() -> engine.getTaskService().complete(task)).isInstanceOf(RuntimeException.class);
-		assertThat(engine.getTaskService().createTaskQuery().taskId(task).count()).isEqualTo(1);
-		assertThat(jdbc.queryForObject("SELECT count(*) FROM flow_notice WHERE status='PENDING'", Integer.class))
-			.isEqualTo(1);
-	}
-
-}