feat(flow): 回调信封升级——事件携带操作人身份(operatorUserId/operatorName)

在飞工作合入:FlowEvent 增操作人字段(可空),引擎/推进监听/回调派发全程透传,
对齐 SDK FlowCallbackEvent 的 taskId/operator 字段(规范 §6.4)。
This commit is contained in:
chenxing
2026-09-22 16:54:02 +08:00
parent 695e20ab52
commit d2395c20ce
9 changed files with 102 additions and 32 deletions
@@ -111,6 +111,11 @@ public class CallbackDispatcher {
payload.put("bizId", instance.getBizId()); payload.put("bizId", instance.getBizId());
payload.put("bizTitle", instance.getBizTitle()); payload.put("bizTitle", instance.getBizTitle());
payload.put("occurredTime", LocalDateTime.now().format(TIME)); payload.put("occurredTime", LocalDateTime.now().format(TIME));
// 2026-09-20 信封升级:任务定位与操作人(均可空,办件级事件/历史重放体为 null——
// 契约见《子应用接入流程中心规范.md》§6.3/§6.4)
payload.put("taskId", event.getTaskId());
payload.put("operatorUserId", event.getOperatorId());
payload.put("operatorName", event.getOperatorName());
Map<String, Object> data = new HashMap<>(); Map<String, Object> data = new HashMap<>();
if (event.getData() != null) { if (event.getData() != null) {
data.putAll(event.getData()); data.putAll(event.getData());
@@ -33,10 +33,12 @@ public class FlowAdvanceService {
* @param edgeName 多出口节点所选出口(办理事件携带);可空——兜底从该节点 * @param edgeName 多出口节点所选出口(办理事件携带);可空——兜底从该节点
* 最近一条已同意任务的 edge_name 列重放(推进在 AFTER_COMMIT 新事务重查, * 最近一条已同意任务的 edge_name 列重放(推进在 AFTER_COMMIT 新事务重查,
* 事件传参只是快路径,落库的值才是权威) * 事件传参只是快路径,落库的值才是权威)
* @param operator 本节点办理人(TASK_HANDLED 事件 data 穿透;推进产生的
* TASK_CREATED/COMPLETED 操作人信封字段来源),可空
* @return 是否实际推进了 * @return 是否实际推进了
*/ */
@Transactional(propagation = Propagation.REQUIRES_NEW) @Transactional(propagation = Propagation.REQUIRES_NEW)
public boolean tryAdvance(Long instanceId, String nodeId, String edgeName) { public boolean tryAdvance(Long instanceId, String nodeId, String edgeName, FlowOperator operator) {
FlwInstance instance = instanceRepository.findById(instanceId).orElse(null); FlwInstance instance = instanceRepository.findById(instanceId).orElse(null);
if (instance == null if (instance == null
|| instance.getStatus() != FlwEnums.InstanceStatus.APPROVING.getCode()) { || instance.getStatus() != FlwEnums.InstanceStatus.APPROVING.getCode()) {
@@ -72,7 +74,7 @@ public class FlowAdvanceService {
.map(FlwTask::getEdgeName) .map(FlwTask::getEdgeName)
.orElse(null); .orElse(null);
} }
flowEngine.continueAdvance(instance, def, nodeId, chosen); flowEngine.continueAdvance(instance, def, nodeId, chosen, operator);
instanceRepository.save(instance); instanceRepository.save(instance);
return true; return true;
} }
@@ -38,11 +38,11 @@ public class FlowEngine {
private final AssigneeResolver assigneeResolver; private final AssigneeResolver assigneeResolver;
private final ApplicationEventPublisher eventPublisher; private final ApplicationEventPublisher eventPublisher;
/** 发起后首推:从 START 激活(instance 已建好、status=10) */ /** 发起后首推:从 START 激活(instance 已建好、status=10)。operator=null:TASK_CREATED/COMPLETED 的操作人回退发起人 */
public void startAdvance(FlwInstance instance, FlwDef def) { public void startAdvance(FlwInstance instance, FlwDef def) {
FlowDefModel model = parseModel(def); FlowDefModel model = parseModel(def);
setCurrentNodes(instance, null); setCurrentNodes(instance, null);
advance(instance, model, model.start, null); advance(instance, model, model.start, null, null);
} }
/** /**
@@ -50,8 +50,10 @@ public class FlowEngine {
* 单出边走唯一出边(旧定义逐字节兼容);多出边按办理人选的出口名(edgeName,即办理按钮文案) * 单出边走唯一出边(旧定义逐字节兼容);多出边按办理人选的出口名(edgeName,即办理按钮文案)
* 选路——未命中时记录错误并保持节点在途,绝不静默走第一条边。 * 选路——未命中时记录错误并保持节点在途,绝不静默走第一条边。
* 分支上下文由 {@link BranchScope#keyOf} 从图上回溯推导(分支内串行链的中间节点无任务级分支标记)。 * 分支上下文由 {@link BranchScope#keyOf} 从图上回溯推导(分支内串行链的中间节点无任务级分支标记)。
* operator=本节点办理人(TASK_CREATED/COMPLETED 操作人信封字段来源),由推进监听器从事件 data 穿透传入。
*/ */
public void continueAdvance(FlwInstance instance, FlwDef def, String fromNodeId, String edgeName) { public void continueAdvance(FlwInstance instance, FlwDef def, String fromNodeId, String edgeName,
FlowOperator operator) {
FlowDefModel model = parseModel(def); FlowDefModel model = parseModel(def);
FlowDefModel.Node node = model.nodeById(fromNodeId); FlowDefModel.Node node = model.nodeById(fromNodeId);
List<FlowDefModel.Edge> outs = model.outEdges(fromNodeId); List<FlowDefModel.Edge> outs = model.outEdges(fromNodeId);
@@ -73,26 +75,29 @@ public class FlowEngine {
FlowDefModel.Node nextNode = model.nodeById(next); FlowDefModel.Node nextNode = model.nodeById(next);
if (FlowDefModel.TYPE_PARALLEL_JOIN.equals(nextNode.type)) { if (FlowDefModel.TYPE_PARALLEL_JOIN.equals(nextNode.type)) {
// 会签合并守门:标记本分支到达、全部到达才放行(直通无分支键时 JOIN 直通放行) // 会签合并守门:标记本分支到达、全部到达才放行(直通无分支键时 JOIN 直通放行)
advance(instance, model, next, branchKey); advance(instance, model, next, branchKey, operator);
} else { } else {
activate(instance, model, nextNode, branchKey); activate(instance, model, nextNode, branchKey, operator);
} }
} }
/** /**
* 退回目标激活(FlowTaskService.returnBack 调用):从 currentNodes 摘除来源节点, * 退回目标激活(FlowTaskService.returnBack 调用):从 currentNodes 摘除来源节点,
* 重新激活目标审批节点建新任务。沿用统一激活入口,split 幂等化后重入并行段不撞分支唯一约束。 * 重新激活目标审批节点建新任务。沿用统一激活入口,split 幂等化后重入并行段不撞分支唯一约束。
* operator=退回人(重件 TASK_CREATED 操作人信封字段来源)。
*/ */
public void returnTo(FlwInstance instance, FlwDef def, String fromNodeId, String targetNodeId) { public void returnTo(FlwInstance instance, FlwDef def, String fromNodeId, String targetNodeId,
FlowOperator operator) {
FlowDefModel model = parseModel(def); FlowDefModel model = parseModel(def);
removeCurrentNode(instance, fromNodeId); removeCurrentNode(instance, fromNodeId);
FlowDefModel.Node target = model.nodeById(targetNodeId); FlowDefModel.Node target = model.nodeById(targetNodeId);
String branchKey = BranchScope.keyOf(model, targetNodeId); String branchKey = BranchScope.keyOf(model, targetNodeId);
advance(instance, model, target.id, branchKey); advance(instance, model, target.id, branchKey, operator);
} }
/** 统一激活入口:含 JOIN 的到达标记与放行判断 */ /** 统一激活入口:含 JOIN 的到达标记与放行判断 */
private void advance(FlwInstance instance, FlowDefModel model, String nodeId, String branchKey) { private void advance(FlwInstance instance, FlowDefModel model, String nodeId, String branchKey,
FlowOperator operator) {
FlowDefModel.Node node = model.nodeById(nodeId); FlowDefModel.Node node = model.nodeById(nodeId);
if (FlowDefModel.TYPE_PARALLEL_JOIN.equals(node.type)) { if (FlowDefModel.TYPE_PARALLEL_JOIN.equals(node.type)) {
if (branchKey != null) { if (branchKey != null) {
@@ -114,24 +119,25 @@ public class FlowEngine {
if (outs.isEmpty()) { if (outs.isEmpty()) {
return; return;
} }
activate(instance, model, model.nodeById(outs.get(0).to), null); advance(instance, model, outs.get(0).to, null, operator);
return; return;
} }
activate(instance, model, node, branchKey); activate(instance, model, node, branchKey, operator);
} }
/** 非 JOIN 节点激活 */ /** 非 JOIN 节点激活 */
private void activate(FlwInstance instance, FlowDefModel model, FlowDefModel.Node node, String branchKey) { private void activate(FlwInstance instance, FlowDefModel model, FlowDefModel.Node node, String branchKey,
FlowOperator operator) {
switch (node.type) { switch (node.type) {
case FlowDefModel.TYPE_START: { case FlowDefModel.TYPE_START: {
List<FlowDefModel.Edge> outs = model.outEdges(node.id); List<FlowDefModel.Edge> outs = model.outEdges(node.id);
if (!outs.isEmpty()) { if (!outs.isEmpty()) {
advance(instance, model, outs.get(0).to, branchKey); advance(instance, model, outs.get(0).to, branchKey, operator);
} }
break; break;
} }
case FlowDefModel.TYPE_APPROVAL: case FlowDefModel.TYPE_APPROVAL:
createTasks(instance, model, node); createTasks(instance, model, node, operator);
addCurrentNode(instance, node.id); addCurrentNode(instance, node.id);
break; break;
case FlowDefModel.TYPE_PARALLEL_SPLIT: { case FlowDefModel.TYPE_PARALLEL_SPLIT: {
@@ -148,23 +154,26 @@ public class FlowEngine {
branch.setStatus(FlwEnums.BranchStatus.RUNNING.getCode()); branch.setStatus(FlwEnums.BranchStatus.RUNNING.getCode());
} }
branchRepository.save(branch); branchRepository.save(branch);
advance(instance, model, edge.to, edge.to); advance(instance, model, edge.to, edge.to, operator);
} }
break; break;
} }
case FlowDefModel.TYPE_END: case FlowDefModel.TYPE_END:
complete(instance); complete(instance, operator);
break; break;
default: default:
log.warn("[flow-center] 未知节点类型 {}(节点 {}),推进中止", node.type, node.id); log.warn("[flow-center] 未知节点类型 {}(节点 {}),推进中止", node.type, node.id);
} }
} }
/** 建任务(ALL=每人一条;ANY=每人一条、任一人办理即放行)+ 到达事件 */ /** 建任务(ALL=每人一条;ANY=每人一条、任一人办理即放行)+ 到达事件。operator=null 回退发起人 */
private void createTasks(FlwInstance instance, FlowDefModel model, FlowDefModel.Node node) { private void createTasks(FlwInstance instance, FlowDefModel model, FlowDefModel.Node node,
FlowOperator operator) {
List<AssigneeRef> assignees = assigneeResolver.resolve(instance.getSchId(), node); List<AssigneeRef> assignees = assigneeResolver.resolve(instance.getSchId(), node);
LocalDateTime now = LocalDateTime.now(); LocalDateTime now = LocalDateTime.now();
LocalDateTime deadline = node.deadlineHours == null ? null : now.plusHours(node.deadlineHours); LocalDateTime deadline = node.deadlineHours == null ? null : now.plusHours(node.deadlineHours);
FlowOperator op = operator != null ? operator
: FlowOperator.of(instance.getSubmitterId(), instance.getSubmitterName());
for (AssigneeRef ref : assignees) { for (AssigneeRef ref : assignees) {
FlwTask task = new FlwTask(); FlwTask task = new FlwTask();
task.setInstanceId(instance.getId()); task.setInstanceId(instance.getId());
@@ -185,20 +194,23 @@ public class FlowEngine {
data.put("totalTasks", assignees.size()); data.put("totalTasks", assignees.size());
data.put("doneTasks", 0); data.put("doneTasks", 0);
eventPublisher.publishEvent(FlowEvent.of(FlowEvent.Type.TASK_CREATED, eventPublisher.publishEvent(FlowEvent.of(FlowEvent.Type.TASK_CREATED,
instance.getId(), task.getId(), data)); instance.getId(), task.getId(), op.getUserId(), op.getUserName(), data));
} }
} }
/** END 激活:办件完成、残留任务/分支失效、完成事件 */ /** END 激活:办件完成、残留任务/分支失效、完成事件。operator=null 回退发起人 */
private void complete(FlwInstance instance) { private void complete(FlwInstance instance, FlowOperator operator) {
instance.setStatus(FlwEnums.InstanceStatus.COMPLETED.getCode()); instance.setStatus(FlwEnums.InstanceStatus.COMPLETED.getCode());
instance.setFinishTime(LocalDateTime.now()); instance.setFinishTime(LocalDateTime.now());
setCurrentNodes(instance, null); setCurrentNodes(instance, null);
taskRepository.invalidatePendingByInstance(instance.getId()); taskRepository.invalidatePendingByInstance(instance.getId());
branchRepository.invalidateRunningByInstance(instance.getId()); branchRepository.invalidateRunningByInstance(instance.getId());
FlowOperator op = operator != null ? operator
: FlowOperator.of(instance.getSubmitterId(), instance.getSubmitterName());
Map<String, Object> data = new HashMap<>(); Map<String, Object> data = new HashMap<>();
data.put("result", "APPROVED"); data.put("result", "APPROVED");
eventPublisher.publishEvent(FlowEvent.of(FlowEvent.Type.COMPLETED, instance.getId(), null, data)); eventPublisher.publishEvent(FlowEvent.of(FlowEvent.Type.COMPLETED, instance.getId(), null,
op.getUserId(), op.getUserName(), data));
} }
public FlowDefModel parseModel(FlwDef def) { public FlowDefModel parseModel(FlwDef def) {
@@ -59,14 +59,24 @@ public class FlowEvent {
private final Long instanceId; private final Long instanceId;
/** 任务事件带 taskId(SUBMITTED 等办件级事件为 null) */ /** 任务事件带 taskId(SUBMITTED 等办件级事件为 null) */
private final Long taskId; private final Long taskId;
/** 操作人 id(谁导致本事件;回调信封提升为 operatorUserId)。可空:语义见《子应用接入流程中心规范.md》§6.4 */
private final Integer operatorId;
/** 操作人姓名快照。可空 */
private final String operatorName;
/** 事件明细(result/opinion/rejectNode/task.created 的 n/m 等),可为空 */ /** 事件明细(result/opinion/rejectNode/task.created 的 n/m 等),可为空 */
private final Map<String, Object> data; private final Map<String, Object> data;
public static FlowEvent of(Type type, Long instanceId) { public static FlowEvent of(Type type, Long instanceId) {
return new FlowEvent(type, instanceId, null, new HashMap<>()); return new FlowEvent(type, instanceId, null, null, null, new HashMap<>());
} }
public static FlowEvent of(Type type, Long instanceId, Long taskId, Map<String, Object> data) { public static FlowEvent of(Type type, Long instanceId, Long taskId, Map<String, Object> data) {
return new FlowEvent(type, instanceId, taskId, data == null ? new HashMap<>() : data); return new FlowEvent(type, instanceId, taskId, null, null, data == null ? new HashMap<>() : data);
}
public static FlowEvent of(Type type, Long instanceId, Long taskId,
Integer operatorId, String operatorName, Map<String, Object> data) {
return new FlowEvent(type, instanceId, taskId, operatorId, operatorName,
data == null ? new HashMap<>() : data);
} }
} }
@@ -31,9 +31,13 @@ public class FlowNodeAdvanceListener {
return; return;
} }
String edgeName = (String) event.getData().get("edgeName"); String edgeName = (String) event.getData().get("edgeName");
// 办理人随事件 data 穿透进推进链:新 TASK_CREATED/COMPLETED 的操作人信封字段(2026-09-20)
FlowOperator operator = FlowOperator.of(
(Integer) event.getData().get("assigneeId"),
(String) event.getData().get("assigneeName"));
for (int attempt = 1; attempt <= 3; attempt++) { for (int attempt = 1; attempt <= 3; attempt++) {
try { try {
advanceService.tryAdvance(event.getInstanceId(), nodeId, edgeName); advanceService.tryAdvance(event.getInstanceId(), nodeId, edgeName, operator);
return; return;
} catch (ObjectOptimisticLockingFailureException e) { } catch (ObjectOptimisticLockingFailureException e) {
log.warn("[flow-center] 推进冲突重试 {}/3:instance={} node={}", log.warn("[flow-center] 推进冲突重试 {}/3:instance={} node={}",
@@ -0,0 +1,23 @@
package com.yunhai.flowcenter.engine;
import lombok.Getter;
import lombok.RequiredArgsConstructor;
/**
* 推进链携带的操作人上下文(2026-09-20 信封升级):任务办理后的放行推进在
* AFTER_COMMIT 新事务里由 FlowNodeAdvanceListener 重查执行,"是谁办的"只存在于
* TASK_HANDLED 事件 data 里,必须显式穿透 listener → tryAdvance → continueAdvance
* → activate → createTasks/complete,否则 TASK_CREATED/COMPLETED 事件拿不到操作人。
* null 语义=未穿透(发起直达路径),事件构建点回退办件发起人。
*/
@Getter
@RequiredArgsConstructor
public class FlowOperator {
private final Integer userId;
private final String userName;
public static FlowOperator of(Integer userId, String userName) {
return new FlowOperator(userId, userName);
}
}
@@ -142,7 +142,8 @@ public class FlowInstanceService {
FlwEnums.TaskStatus.WITHDRAWN, instance.getSubmitterId(), instance.getSubmitterName()); FlwEnums.TaskStatus.WITHDRAWN, instance.getSubmitterId(), instance.getSubmitterName());
Map<String, Object> data = new HashMap<>(); Map<String, Object> data = new HashMap<>();
data.put("result", "WITHDRAWN"); data.put("result", "WITHDRAWN");
eventPublisher.publishEvent(FlowEvent.of(FlowEvent.Type.WITHDRAWN, instanceId, null, data)); eventPublisher.publishEvent(FlowEvent.of(FlowEvent.Type.WITHDRAWN, instanceId, null,
instance.getSubmitterId(), instance.getSubmitterName(), data));
} }
/** /**
@@ -230,7 +231,8 @@ public class FlowInstanceService {
FlowDefModel model = flowEngine.parseModel(requireDef(instance.getDefId())); FlowDefModel model = flowEngine.parseModel(requireDef(instance.getDefId()));
Map<String, Object> data = new HashMap<>(); Map<String, Object> data = new HashMap<>();
data.put("nodeSummary", FlowEngine.nodeSummary(model)); data.put("nodeSummary", FlowEngine.nodeSummary(model));
eventPublisher.publishEvent(FlowEvent.of(FlowEvent.Type.SUBMITTED, instance.getId(), null, data)); eventPublisher.publishEvent(FlowEvent.of(FlowEvent.Type.SUBMITTED, instance.getId(), null,
instance.getSubmitterId(), instance.getSubmitterName(), data));
} }
/** 代发起结果值对象 */ /** 代发起结果值对象 */
@@ -6,6 +6,7 @@ import com.yunhai.flowcenter.engine.BranchScope;
import com.yunhai.flowcenter.engine.FlowDefModel; import com.yunhai.flowcenter.engine.FlowDefModel;
import com.yunhai.flowcenter.engine.FlowEvent; import com.yunhai.flowcenter.engine.FlowEvent;
import com.yunhai.flowcenter.engine.FlowEngine; import com.yunhai.flowcenter.engine.FlowEngine;
import com.yunhai.flowcenter.engine.FlowOperator;
import com.yunhai.flowcenter.entity.FlwDef; import com.yunhai.flowcenter.entity.FlwDef;
import com.yunhai.flowcenter.entity.FlwInstance; import com.yunhai.flowcenter.entity.FlwInstance;
import com.yunhai.flowcenter.entity.FlwTask; import com.yunhai.flowcenter.entity.FlwTask;
@@ -97,7 +98,7 @@ public class FlowTaskService {
data.put("edgeName", chosenEdge); data.put("edgeName", chosenEdge);
} }
eventPublisher.publishEvent(FlowEvent.of(FlowEvent.Type.TASK_HANDLED, eventPublisher.publishEvent(FlowEvent.of(FlowEvent.Type.TASK_HANDLED,
instance.getId(), taskId, data)); instance.getId(), taskId, task.getAssigneeId(), task.getAssigneeName(), data));
if (!approved) { if (!approved) {
rejectFlow(instance, task, opinion); rejectFlow(instance, task, opinion);
} else if (FlowDefModel.COUNTERSIGN_ANY.equals(countersignOf(task))) { } else if (FlowDefModel.COUNTERSIGN_ANY.equals(countersignOf(task))) {
@@ -163,7 +164,7 @@ public class FlowTaskService {
data.put("rejectTaskId", task.getId()); data.put("rejectTaskId", task.getId());
data.put("opinion", opinion); data.put("opinion", opinion);
eventPublisher.publishEvent(FlowEvent.of(FlowEvent.Type.REJECTED, eventPublisher.publishEvent(FlowEvent.of(FlowEvent.Type.REJECTED,
instance.getId(), task.getId(), data)); instance.getId(), task.getId(), task.getAssigneeId(), task.getAssigneeName(), data));
} }
/** /**
@@ -199,7 +200,8 @@ public class FlowTaskService {
throw new BizException(ErrorCode.PARAM_ERROR, "任务已被办理或已失效,请刷新"); throw new BizException(ErrorCode.PARAM_ERROR, "任务已被办理或已失效,请刷新");
} }
taskRepository.invalidatePendingByNode(instance.getId(), task.getNodeId()); taskRepository.invalidatePendingByNode(instance.getId(), task.getNodeId());
flowEngine.returnTo(instance, def, task.getNodeId(), target); flowEngine.returnTo(instance, def, task.getNodeId(), target,
FlowOperator.of(task.getAssigneeId(), task.getAssigneeName()));
instanceRepository.save(instance); instanceRepository.save(instance);
Map<String, Object> data = new HashMap<>(); Map<String, Object> data = new HashMap<>();
@@ -291,7 +293,7 @@ public class FlowTaskService {
data.put("opinion", reason); data.put("opinion", reason);
} }
eventPublisher.publishEvent(FlowEvent.of(FlowEvent.Type.TRANSFERRED, eventPublisher.publishEvent(FlowEvent.of(FlowEvent.Type.TRANSFERRED,
instance.getId(), newTask.getId(), data)); instance.getId(), newTask.getId(), task.getAssigneeId(), task.getAssigneeName(), data));
} }
/** 待办任务守卫:存在、本校、本人办理、仍待办 */ /** 待办任务守卫:存在、本校、本人办理、仍待办 */
@@ -197,6 +197,16 @@ class CallbackDispatchIT extends FlowTestSupport {
assertTrue(payload.get("instanceVersion").asInt() >= 1, "instanceVersion 应为办件乐观锁版本"); assertTrue(payload.get("instanceVersion").asInt() >= 1, "instanceVersion 应为办件乐观锁版本");
assertNotNull(payload.get("eventId")); assertNotNull(payload.get("eventId"));
assertNotNull(payload.get("occurredTime")); assertNotNull(payload.get("occurredTime"));
// 2026-09-20 信封升级:taskId + 操作人(发起直达链路 → 操作人=发起人快照)
assertEquals("201", payload.get("operatorUserId").asText(), "operatorUserId 应为发起人");
assertEquals("张三", payload.get("operatorName").asText(), "operatorName 应为发起人");
if ("flow.task.created".equals(event)) {
assertTrue(payload.get("taskId") != null && payload.get("taskId").isNumber(),
"task.created 必带顶层 taskId");
} else if ("flow.submitted".equals(event)) {
assertTrue(payload.get("taskId") == null || payload.get("taskId").isNull(),
"办件级事件 taskId 应为 null");
}
// 签名断言:Base64(HmacSHA256(secret, ts.nonce.body)) // 签名断言:Base64(HmacSHA256(secret, ts.nonce.body))
String expected = hmac(reg.getSecret(), r.timestamp, r.nonce, r.body); String expected = hmac(reg.getSecret(), r.timestamp, r.nonce, r.body);
assertEquals(expected, r.signature, "推送必须带合法 X-Flw-Signature"); assertEquals(expected, r.signature, "推送必须带合法 X-Flw-Signature");