From d0aa3a728450b9c38a6584c5de9a902f6d458f9b Mon Sep 17 00:00:00 2001 From: PIGCLOUD Date: Wed, 14 Oct 2026 17:26:47 +0800 Subject: [PATCH] fix(flow): unify workflow notices and draft reliability (cherry picked from commit 1632fbb83d24be1523e6fa809ed2ad0ca3e6cb0b) --- db/0pigxx_boot.sql | 12 ++- db/2pigxx.sql | 6 +- db/3pigxx_flow.sql | 6 +- .../flow/constant/FlowNoticeStatusEnum.java | 8 +- .../flow/config/FlowDurableConfiguration.java | 18 +++++ .../controller/FlowWorkspaceController.java | 29 +------ .../pigx/flow/service/FlowDraftService.java | 7 +- .../flow/service/FlowReliabilityService.java | 80 +++++++++++-------- .../flow/service/impl/TaskServiceImpl.java | 45 ++++++----- .../pigx/flow/service/FlowWorkspaceTests.java | 72 +++++++++++++---- .../service/impl/TaskServiceImplTests.java | 9 +++ .../pigx/admin/api/dto/MessageNoticeDTO.java | 9 +++ .../admin/api/entity/SysMessageEntity.java | 9 +++ .../admin/api/feign/RemoteMessageService.java | 4 +- .../controller/SysMessageController.java | 12 +++ .../service/impl/SysMessageServiceImpl.java | 13 +++ .../service/impl/SysMessageNoticeTests.java | 58 ++++++++++++++ 17 files changed, 289 insertions(+), 108 deletions(-) create mode 100644 pigx-upms/pigx-upms-biz/src/test/java/com/pig4cloud/pigx/admin/service/impl/SysMessageNoticeTests.java diff --git a/db/0pigxx_boot.sql b/db/0pigxx_boot.sql index d2b53de90..019602827 100644 --- a/db/0pigxx_boot.sql +++ b/db/0pigxx_boot.sql @@ -1533,7 +1533,11 @@ CREATE TABLE `sys_message` ( `update_by` varchar(32) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL COMMENT '更新人', `del_flag` char(1) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL COMMENT '删除时间', `tenant_id` bigint DEFAULT NULL COMMENT '租户', - PRIMARY KEY (`id`) + `business_type` varchar(32) DEFAULT NULL COMMENT '业务类型', + `business_id` varchar(64) DEFAULT NULL COMMENT '业务记录ID', + `request_key` varchar(100) DEFAULT NULL COMMENT '调用方事件标识', + PRIMARY KEY (`id`), + UNIQUE KEY `uk_message_request` (`tenant_id`,`request_key`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT='站内信息'; -- ---------------------------- @@ -3243,7 +3247,7 @@ CREATE TABLE IF NOT EXISTS flow_draft ( kind varchar(16) NOT NULL, flow_id varchar(64) NOT NULL DEFAULT '', title varchar(200) NOT NULL, payload longtext NOT NULL, revision bigint NOT NULL, updated_at timestamp(3) NOT NULL, PRIMARY KEY(id), KEY idx_draft_owner(tenant_id,user_id,kind,updated_at) -) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci; CREATE TABLE IF NOT EXISTS flow_notice ( id varchar(36) NOT NULL, tenant_id bigint NOT NULL, user_id varchar(64) NOT NULL, event_key varchar(180) NOT NULL, kind varchar(16) NOT NULL, title varchar(200) NOT NULL, @@ -3253,7 +3257,7 @@ CREATE TABLE IF NOT EXISTS flow_notice ( reminder_key varchar(64) NULL, PRIMARY KEY(id), UNIQUE KEY uk_notice_event(tenant_id,user_id,event_key), KEY idx_notice_owner(tenant_id,user_id,created_at), UNIQUE KEY uk_notice_reminder(tenant_id,reminder_key), KEY idx_notice_due(status,scheduled_at) -) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci; CREATE TABLE IF NOT EXISTS flow_callback ( id varchar(36) NOT NULL, tenant_id bigint NOT NULL, task_id varchar(64) NOT NULL, instance_id varchar(64) NOT NULL, target_url varchar(2000) NOT NULL, payload longtext NOT NULL, @@ -3262,4 +3266,4 @@ CREATE TABLE IF NOT EXISTS flow_callback ( lease_token varchar(36) NULL, lease_until timestamp(3) NULL, last_error varchar(500) NULL, created_at timestamp(3) NOT NULL, updated_at timestamp(3) NOT NULL, PRIMARY KEY(id), UNIQUE KEY uk_callback_task(tenant_id,task_id), KEY idx_callback_due(status,next_at) -) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci; diff --git a/db/2pigxx.sql b/db/2pigxx.sql index 1215fd407..b696dcc9a 100644 --- a/db/2pigxx.sql +++ b/db/2pigxx.sql @@ -1484,7 +1484,11 @@ CREATE TABLE `sys_message` ( `update_by` varchar(32) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci DEFAULT NULL COMMENT '更新人', `del_flag` char(1) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL COMMENT '删除时间', `tenant_id` bigint DEFAULT NULL COMMENT '租户', - PRIMARY KEY (`id`) + `business_type` varchar(32) DEFAULT NULL COMMENT '业务类型', + `business_id` varchar(64) DEFAULT NULL COMMENT '业务记录ID', + `request_key` varchar(100) DEFAULT NULL COMMENT '调用方事件标识', + PRIMARY KEY (`id`), + UNIQUE KEY `uk_message_request` (`tenant_id`,`request_key`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT='站内信息'; -- ---------------------------- diff --git a/db/3pigxx_flow.sql b/db/3pigxx_flow.sql index 6533693c7..48cca9109 100644 --- a/db/3pigxx_flow.sql +++ b/db/3pigxx_flow.sql @@ -296,7 +296,7 @@ CREATE TABLE IF NOT EXISTS flow_draft ( kind varchar(16) NOT NULL, flow_id varchar(64) NOT NULL DEFAULT '', title varchar(200) NOT NULL, payload longtext NOT NULL, revision bigint NOT NULL, updated_at timestamp(3) NOT NULL, PRIMARY KEY(id), KEY idx_draft_owner(tenant_id,user_id,kind,updated_at) -) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci; CREATE TABLE IF NOT EXISTS flow_notice ( id varchar(36) NOT NULL, tenant_id bigint NOT NULL, user_id varchar(64) NOT NULL, event_key varchar(180) NOT NULL, kind varchar(16) NOT NULL, title varchar(200) NOT NULL, @@ -306,7 +306,7 @@ CREATE TABLE IF NOT EXISTS flow_notice ( reminder_key varchar(64) NULL, PRIMARY KEY(id), UNIQUE KEY uk_notice_event(tenant_id,user_id,event_key), KEY idx_notice_owner(tenant_id,user_id,created_at), UNIQUE KEY uk_notice_reminder(tenant_id,reminder_key), KEY idx_notice_due(status,scheduled_at) -) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci; CREATE TABLE IF NOT EXISTS flow_callback ( id varchar(36) NOT NULL, tenant_id bigint NOT NULL, task_id varchar(64) NOT NULL, instance_id varchar(64) NOT NULL, target_url varchar(2000) NOT NULL, payload longtext NOT NULL, @@ -315,4 +315,4 @@ CREATE TABLE IF NOT EXISTS flow_callback ( lease_token varchar(36) NULL, lease_until timestamp(3) NULL, last_error varchar(500) NULL, created_at timestamp(3) NOT NULL, updated_at timestamp(3) NOT NULL, PRIMARY KEY(id), UNIQUE KEY uk_callback_task(tenant_id,task_id), KEY idx_callback_due(status,next_at) -) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci; diff --git a/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/constant/FlowNoticeStatusEnum.java b/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/constant/FlowNoticeStatusEnum.java index 46c61f15e..485819e71 100644 --- a/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/constant/FlowNoticeStatusEnum.java +++ b/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/constant/FlowNoticeStatusEnum.java @@ -4,8 +4,7 @@ import lombok.Getter; import lombok.RequiredArgsConstructor; /** - * 流程通知的持久化状态；编码与 flow_notice.status 及升级 SQL 保持一致。 待触发提醒发送后转为 SENT，任务结束时转为 CANCELLED；列表仅展示 - * SENT。 + * 流程通知事件状态；READY 经统一信息推送发送后转为 SENT，待触发提醒随任务结束取消。 */ @Getter @RequiredArgsConstructor @@ -14,7 +13,10 @@ public enum FlowNoticeStatusEnum { /** 等待截止时间，到期后确定当前审批人。 */ PENDING("PENDING"), - /** 通知已生成，允许接收人查看及标记已读。 */ + /** 等待投递到统一站内信。 */ + READY("READY"), + + /** 已投递到统一站内信，已读状态由 sys_message_relation 维护。 */ SENT("SENT"), /** 任务已经结束，取消尚未发送的提醒。 */ diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/config/FlowDurableConfiguration.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/config/FlowDurableConfiguration.java index 16ebbf55b..458365a67 100644 --- a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/config/FlowDurableConfiguration.java +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/config/FlowDurableConfiguration.java @@ -97,4 +97,22 @@ public class FlowDurableConfiguration { } } + /** 复用流程事件表投递站内信；失败延迟一分钟再试，不阻塞引擎事务。 */ + @Scheduled(fixedDelayString = "${flow.delivery.scan-delay:10000}", scheduler = "flowDeliveryScheduler") + public void scanNotices() { + var notices = jdbc.queryForList( + "SELECT id FROM flow_notice WHERE status=? AND scheduled_at<=CURRENT_TIMESTAMP(3) ORDER BY scheduled_at LIMIT 100", + FlowNoticeStatusEnum.READY.getCode()); + for (var row : notices) { + try { + reliability.getObject().deliverNotice(String.valueOf(row.get("id"))); + } + catch (Exception exception) { + jdbc.update("UPDATE flow_notice SET scheduled_at=TIMESTAMPADD(SECOND,60,CURRENT_TIMESTAMP(3)) WHERE id=? AND status=?", + row.get("id"), FlowNoticeStatusEnum.READY.getCode()); + log.warn("流程站内信暂未发送，事件 {}", row.get("id"), exception); + } + } + } + } diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/controller/FlowWorkspaceController.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/controller/FlowWorkspaceController.java index 4e956915a..1e81143f6 100644 --- a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/controller/FlowWorkspaceController.java +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/controller/FlowWorkspaceController.java @@ -55,8 +55,9 @@ public class FlowWorkspaceController { */ @Operation(summary = "分页查询个人草稿") @GetMapping("/drafts") - public R>> list(@RequestParam String kind, @RequestParam(defaultValue = "1") int page) { - return R.ok(drafts.list(tenant(), user(), kind, page)); + public R>> list(@RequestParam String kind, @RequestParam(defaultValue = "1") int page, + @RequestParam(required = false) String flowId) { + return R.ok(drafts.list(tenant(), user(), kind, page, flowId)); } /** @@ -84,30 +85,6 @@ public class FlowWorkspaceController { return R.ok(drafts.delete(tenant(), user(), id, revision)); } - /** - * 查询个人站内通知。 - * @param page 页码 - * @return 通知列表 - */ - @Operation(summary = "分页查询个人通知") - @GetMapping("/notices") - public R>> notices(@RequestParam(defaultValue = "1") int page) { - return R.ok(reliability.notices(tenant(), user(), page)); - } - - /** - * 标记个人通知已读。 - * @param id 通知 ID - * @return 操作结果 - */ - @Operation(summary = "标记个人通知已读") - @SysLog("标记个人通知已读") - @PutMapping("/notices/{id}/read") - public R read(@PathVariable String id) { - reliability.read(tenant(), user(), id); - return R.ok(); - } - /** * 对自己发起的实例催办。 * @param id 实例 ID diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/FlowDraftService.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/FlowDraftService.java index 9f2068d36..c65e8a643 100644 --- a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/FlowDraftService.java +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/FlowDraftService.java @@ -77,12 +77,13 @@ public class FlowDraftService { * @param user 当前用户 * @param kind DESIGN 或 APPLICATION * @param page 从 1 开始的页码 + * @param flowId 可选的流程 ID，为空时查询全部流程 * @return 每页最多 50 个草稿 */ - public List> list(long tenant, long user, String kind, int page) { + public List> list(long tenant, long user, String kind, int page, String flowId) { return jdbc.queryForList( - "SELECT id,kind,flow_id,title,revision,updated_at FROM flow_draft WHERE tenant_id=? AND user_id=? AND kind=? ORDER BY updated_at DESC LIMIT 50 OFFSET ?", - tenant, user, kind, Math.max(0L, (long) page - 1) * 50); + "SELECT id,kind,flow_id,title,revision,updated_at FROM flow_draft WHERE tenant_id=? AND user_id=? AND kind=? AND (? IS NULL OR flow_id=?) ORDER BY updated_at DESC, id DESC LIMIT 50 OFFSET ?", + tenant, user, kind, flowId, flowId, Math.max(0L, (long) page - 1) * 50); } /** diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/FlowReliabilityService.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/FlowReliabilityService.java index e7fd59e18..de939d41d 100644 --- a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/FlowReliabilityService.java +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/FlowReliabilityService.java @@ -3,6 +3,8 @@ package com.pig4cloud.pigx.flow.service; import cn.hutool.core.convert.Convert; import cn.hutool.core.util.StrUtil; import cn.hutool.json.JSONUtil; +import com.pig4cloud.pigx.admin.api.dto.MessageNoticeDTO; +import com.pig4cloud.pigx.admin.api.feign.RemoteMessageService; import com.pig4cloud.pigx.common.data.tenant.TenantBroker; import com.pig4cloud.pigx.flow.dto.Node; import com.pig4cloud.pigx.flow.constant.NodeStatusEnum; @@ -36,6 +38,8 @@ public class FlowReliabilityService { private final TaskService taskService; + private final RemoteMessageService messageService; + /** * 在引擎事务内记录任务事件；持久化失败必须回滚引擎操作。 * @param task 引擎当前任务 @@ -108,10 +112,10 @@ public class FlowReliabilityService { private void notice(long tenant, String user, String event, String kind, String title, String task, String instance) { jdbc.update( - "INSERT INTO flow_notice(id,tenant_id,user_id,event_key,kind,title,task_id,instance_id,created_at,sent_at,status) VALUES(?,?,?,?,?,?,?,?,CURRENT_TIMESTAMP(3),CURRENT_TIMESTAMP(3),?) ON DUPLICATE KEY UPDATE event_key=event_key", + "INSERT INTO flow_notice(id,tenant_id,user_id,event_key,kind,title,task_id,instance_id,created_at,scheduled_at,status) VALUES(?,?,?,?,?,?,?,?,CURRENT_TIMESTAMP(3),CURRENT_TIMESTAMP(3),?) ON DUPLICATE KEY UPDATE event_key=event_key", UUID.randomUUID().toString(), tenant, user, event, kind, StrUtil.maxLength(StrUtil.blankToDefault(title, "流程任务"), 200), task, instance, - FlowNoticeStatusEnum.SENT.getCode()); + FlowNoticeStatusEnum.READY.getCode()); } /** @@ -124,30 +128,33 @@ public class FlowReliabilityService { */ @Transactional public int urge(long tenant, long user, String instance) { - requireStarter(tenant, user, instance, true); + String flowName = requireStarter(tenant, user, instance, true); var tasks = taskService.createTaskQuery() .processInstanceId(instance) .taskTenantId(String.valueOf(tenant)) .active() .list(); if (tasks.isEmpty()) - throw new IllegalStateException("流程当前没有可催办任务"); + throw new IllegalStateException("当前没有待处理的审批，暂时无法催办"); var recent = jdbc.queryForList( "SELECT id FROM flow_notice WHERE tenant_id=? AND instance_id=? AND kind='URGE' AND created_at>?", tenant, instance, Timestamp.from(Instant.now().minusSeconds(600))); if (!recent.isEmpty()) - throw new IllegalStateException("已发送催办，请十分钟后再试"); + throw new IllegalStateException("已提交过催办，两次催办需间隔十分钟"); int count = 0; String event = UUID.randomUUID().toString(); for (var task : tasks) { if (StrUtil.isBlank(task.getAssignee())) continue; - notice(tenant, task.getAssignee(), "URGE:" + event + StrUtil.COLON + task.getId(), "URGE", task.getName(), + // 建模时会在节点名称末尾追加五位随机数，通知展示时移除内部后缀。 + String nodeName = StrUtil.blankToDefault(task.getName(), "审批").replaceFirst("\\[\\d{5}\\]$", StrUtil.EMPTY); + notice(tenant, task.getAssignee(), "URGE:" + event + StrUtil.COLON + task.getId(), "URGE", + flowName + " · " + nodeName, task.getId(), instance); count++; } if (count == 0) - throw new IllegalStateException("流程当前没有已分配的待办任务"); + throw new IllegalStateException("当前待办尚未分配处理人，暂时无法催办"); return count; } @@ -175,35 +182,43 @@ public class FlowReliabilityService { return; } jdbc.update( - "UPDATE flow_notice SET user_id=?,title=?,status=?,sent_at=CURRENT_TIMESTAMP(3),created_at=CURRENT_TIMESTAMP(3) WHERE id=? AND tenant_id=? AND status=?", + "UPDATE flow_notice SET user_id=?,title=?,status=?,scheduled_at=CURRENT_TIMESTAMP(3),created_at=CURRENT_TIMESTAMP(3) WHERE id=? AND tenant_id=? AND status=?", task.getAssignee(), StrUtil.maxLength(StrUtil.blankToDefault(task.getName(), "流程任务"), 200), - FlowNoticeStatusEnum.SENT.getCode(), watches.get(0).get("id"), tenant, + FlowNoticeStatusEnum.READY.getCode(), watches.get(0).get("id"), tenant, FlowNoticeStatusEnum.PENDING.getCode()); } /** - * 分页读取个人通知。 - * @param tenant 当前租户 - * @param user 当前用户 - * @param page 从 1 开始的页码 - * @return 最近通知，每页最多 50 条 + * 投递已提交的流程事件到统一站内信；行锁防止扫描器并发发送同一事件。 + * @param id 流程通知事件 ID */ - public List> notices(long tenant, long user, int page) { - return jdbc.queryForList( - "SELECT * FROM flow_notice WHERE tenant_id=? AND user_id=? AND status=? ORDER BY created_at DESC LIMIT 50 OFFSET ?", - tenant, String.valueOf(user), FlowNoticeStatusEnum.SENT.getCode(), Math.max(0L, (long) page - 1) * 50); - } - - /** - * 标记个人通知已读，其他用户记录不受影响。 - * @param tenant 当前租户 - * @param user 当前用户 - * @param id 通知 ID - */ - public void read(long tenant, long user, String id) { - jdbc.update( - "UPDATE flow_notice SET read_at=COALESCE(read_at,CURRENT_TIMESTAMP(3)) WHERE id=? AND tenant_id=? AND user_id=? AND status=?", - id, tenant, String.valueOf(user), FlowNoticeStatusEnum.SENT.getCode()); + @Transactional + public void deliverNotice(String id) { + var rows = jdbc.queryForList("SELECT * FROM flow_notice WHERE id=? AND status=? FOR UPDATE", id, + FlowNoticeStatusEnum.READY.getCode()); + if (rows.isEmpty()) + return; + var row = rows.get(0); + long tenant = ((Number) row.get("tenant_id")).longValue(); + MessageNoticeDTO notice = new MessageNoticeDTO(); + notice.setCategory("1"); + notice.setAllFlag("0"); + notice.setUserIds(List.of(Long.parseLong(String.valueOf(row.get("user_id"))))); + String label = switch (String.valueOf(row.get("kind"))) { + case "URGE" -> "催办"; + case "OVERDUE" -> "超时提醒"; + default -> "待办"; + }; + notice.setTitle(label + StrUtil.COLON + row.get("title")); + notice.setContent("您有一条" + label + "消息，请打开对应审批任务处理。"); + notice.setBusinessType("FLOW_TASK"); + notice.setBusinessId(String.valueOf(row.get("task_id"))); + notice.setRequestKey("flow:" + id); + var result = TenantBroker.applyAs(tenant, t -> messageService.sendNotice(notice)); + if (result == null || result.getCode() != 0) + throw new IllegalStateException("流程站内信发送失败"); + jdbc.update("UPDATE flow_notice SET status=?,sent_at=CURRENT_TIMESTAMP(3) WHERE id=? AND tenant_id=?", + FlowNoticeStatusEnum.SENT.getCode(), id, tenant); } /** @@ -246,15 +261,16 @@ public class FlowReliabilityService { FlowNoticeStatusEnum.CANCELLED.getCode(), tenant, task, FlowNoticeStatusEnum.PENDING.getCode()); } - private void requireStarter(long tenant, long user, String instance, boolean running) { + private String requireStarter(long tenant, long user, String instance, boolean running) { var rows = jdbc.queryForList( - "SELECT status FROM process_instance_record WHERE process_instance_id=? AND user_id=? AND tenant_id=? AND del_flag='0' LIMIT 1" + "SELECT status,name FROM process_instance_record WHERE process_instance_id=? AND user_id=? AND tenant_id=? AND del_flag='0' LIMIT 1" + (running ? " FOR UPDATE" : StrUtil.EMPTY), instance, user, tenant); if (rows.isEmpty()) throw new IllegalStateException("流程不存在或无权访问"); if (running && !Objects.equals(Convert.toInt(rows.get(0).get("status")), NodeStatusEnum.JXZ.getCode())) throw new IllegalStateException("流程已结束，不能催办"); + return StrUtil.blankToDefault(Convert.toStr(rows.get(0).get("name")), "流程申请"); } } diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/TaskServiceImpl.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/TaskServiceImpl.java index 6f6098b90..53daa8f9b 100644 --- a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/TaskServiceImpl.java +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/TaskServiceImpl.java @@ -171,6 +171,9 @@ public class TaskServiceImpl implements ITaskService { .taskTenantId(TenantContextHolder.getTenantId().toString()) .taskAssignee(String.valueOf(userId)) .singleResult()); + if (!view && task.isEmpty()) { + return R.failed("任务已结束或已转交，无法办理"); + } TaskResultDto taskResultDto = new TaskResultDto(); if (task.isPresent()) { @@ -347,7 +350,7 @@ public class TaskServiceImpl implements ITaskService { ProcessNodeRecordAssignUser currentAssignment = findActiveAssignment(task.getId(), userId); ProcessNodeRecord nodeRecord = lockActiveNodeRecord(task, currentAssignment, TenantContextHolder.getTenantId()); if (nodeRecord == null) { - return R.failed("当前审批节点已结束，无法操作"); + return R.failed("当前审批环节已结束，无法继续操作"); } Map taskLocalParamMap = taskParamDto.getTaskLocalParamMap(); @@ -449,7 +452,7 @@ public class TaskServiceImpl implements ITaskService { @Transactional(rollbackFor = Exception.class) public R addSign(AddSignTaskDto addSignTaskDto) { if (addSignTaskDto == null || StrUtil.isBlank(addSignTaskDto.getTaskId())) { - return R.failed("任务ID不能为空"); + return R.failed("审批信息不完整，请重新打开"); } List requestedUserIds = addSignTaskDto.getTargetUserIdList(); if (CollUtil.isEmpty(requestedUserIds)) { @@ -481,7 +484,7 @@ public class TaskServiceImpl implements ITaskService { ProcessNodeRecordAssignUser sourceAssignment = findActiveAssignment(sourceTask.getId(), userId); ProcessNodeRecord nodeRecord = lockActiveNodeRecord(sourceTask, sourceAssignment, tenantId); if (nodeRecord == null) { - return R.failed("当前审批节点已结束，无法加签"); + return R.failed("当前审批环节已结束，无法加签"); } String flowId = NodeUtil.getFlowId(sourceTask.getProcessDefinitionId()); @@ -489,17 +492,17 @@ public class TaskServiceImpl implements ITaskService { .getPublishedNode(sourceTask.getProcessDefinitionId(), sourceTask.getTaskDefinitionKey()) .getData(); if (!isParallelCountersign(node)) { - return R.failed("仅并行会签节点支持加签"); + return R.failed("只有多人同时审批的会签环节支持加签"); } if (!Boolean.TRUE.equals(node.getAllowAddSign())) { - return R.failed("当前节点未开启加签"); + return R.failed("当前审批环节未开启加签功能"); } if (isProcessStarter(sourceTask.getProcessInstanceId(), userId)) { return R.failed("流程发起人不能发起加签"); } if (!isOriginalApprover(sourceTask, sourceAssignment, userId)) { - return R.failed("只有未完成审批的原始审批人可以加签"); + return R.failed("只有最初指定的审批人，且尚未完成审批时，才能加签"); } List participantUserIds = listParticipantUserIds(nodeRecord.getId()); @@ -577,7 +580,7 @@ public class TaskServiceImpl implements ITaskService { } catch (FlowableException exception) { log.warn("动态加签时多实例节点已不可用: taskId={}", sourceTask.getId(), exception); - return rollbackFailed("当前审批节点已结束，无法加签"); + return rollbackFailed("当前审批环节已结束，无法加签"); } LinkedHashSet participantUserIdsAfter = new LinkedHashSet<>(participantUserIds); @@ -600,11 +603,11 @@ public class TaskServiceImpl implements ITaskService { @Override public R delegateTask(TaskParamDto taskParamDto) { if (taskParamDto == null || StrUtil.isBlank(taskParamDto.getTaskId())) { - return R.failed("任务ID不能为空"); + return R.failed("审批信息不完整，请重新打开"); } if (taskParamDto.getTargetUserIdList() == null || taskParamDto.getTargetUserIdList().size() != 1 || taskParamDto.getTargetUserIdList().get(0) == null) { - return R.failed("委托仅支持选择一名用户"); + return R.failed("一次只能委托给一人"); } long userId = SecurityUtils.getUser().getId(); Long tenantId = TenantContextHolder.getTenantId(); @@ -620,10 +623,10 @@ public class TaskServiceImpl implements ITaskService { ProcessNodeRecordAssignUser sourceAssignment = findActiveAssignment(task.getId(), userId); ProcessNodeRecord nodeRecord = lockActiveNodeRecord(task, sourceAssignment, tenantId); if (nodeRecord == null) { - return R.failed("当前审批节点已结束，无法委托"); + return R.failed("当前审批环节已结束，无法委托"); } if (listParticipantUserIds(nodeRecord.getId()).contains(targetUserId)) { - return R.failed("委托目标已参与当前会签"); + return R.failed("所选人员已参与本轮会签，请选择其他人"); } if (queryValidTenantUsers(taskParamDto.getTargetUserIdList(), tenantId).size() != 1) { return R.failed("委托用户不存在、不属于当前租户或已被锁定"); @@ -678,11 +681,11 @@ public class TaskServiceImpl implements ITaskService { @Transactional(rollbackFor = Exception.class) public R setAssignee(TaskParamDto taskParamDto) { if (taskParamDto == null || StrUtil.isBlank(taskParamDto.getTaskId())) { - return R.failed("任务ID不能为空"); + return R.failed("审批信息不完整，请重新打开"); } if (taskParamDto.getTargetUserIdList() == null || taskParamDto.getTargetUserIdList().size() != 1 || taskParamDto.getTargetUserIdList().get(0) == null) { - return R.failed("转办仅支持选择一名用户"); + return R.failed("一次只能转办给一人"); } long userId = SecurityUtils.getUser().getId(); @@ -701,17 +704,17 @@ public class TaskServiceImpl implements ITaskService { ProcessNodeRecordAssignUser sourceAssignment = findActiveAssignment(task.getId(), userId); ProcessNodeRecord nodeRecord = lockActiveNodeRecord(task, sourceAssignment, tenantId); if (nodeRecord == null) { - return R.failed("当前审批节点已结束，无法转办"); + return R.failed("当前审批环节已结束，无法转办"); } if (listParticipantUserIds(nodeRecord.getId()).contains(targetUserId)) { - return R.failed("转办目标已参与当前会签"); + return R.failed("所选人员已参与本轮会签，请选择其他人"); } if (queryValidTenantUsers(List.of(targetUserId), tenantId).size() != 1) { return R.failed("转办用户不存在或不属于当前租户"); } if (sourceAssignment == null) { - return R.failed("未找到当前任务的执行人记录"); + return R.failed("未找到当前审批的处理人记录，请联系管理员"); } Map taskLocalParamMap = taskParamDto.getTaskLocalParamMap(); @@ -778,7 +781,7 @@ public class TaskServiceImpl implements ITaskService { public R withdrawProcessInstance(TaskParamDto taskParamDto) { String processInstanceId = taskParamDto.getProcessInstanceId(); if (StrUtil.isBlank(processInstanceId)) { - return R.failed("流程实例ID不能为空"); + return R.failed("申请信息不完整，请重新打开"); } long userId = SecurityUtils.getUser().getId(); @@ -798,7 +801,7 @@ public class TaskServiceImpl implements ITaskService { Object rejectToStarter = runtimeService.getVariable(processInstanceId, ProcessInstanceConstant.REJECT_TO_STARTER_VAR); if (Boolean.TRUE.equals(rejectToStarter)) { - return R.failed("驳回待提交状态不可撤回"); + return R.failed("申请已被驳回，正在等待重新提交，不能撤回"); } } catch (Exception e) { @@ -813,7 +816,7 @@ public class TaskServiceImpl implements ITaskService { .ne(ProcessNodeRecordAssignUser::getStatus, NodeStatusEnum.JXZ.getCode())) .exists(); if (hasProcessedApproval) { - return R.failed("首个审批节点已处理，无法撤回"); + return R.failed("第一步审批已处理，无法撤回"); } return terminateProcessInstance(processInstanceId, ProcessInstanceConstant.FINISH_REASON_WITHDRAW); @@ -861,7 +864,7 @@ public class TaskServiceImpl implements ITaskService { // 2. 验证是 root 任务（发起人节点） if (!StrUtil.equals(task.getTaskDefinitionKey(), "root")) { - return R.failed("非重新提交任务，无法操作"); + return R.failed("当前审批不支持重新提交"); } String processInstanceId = task.getProcessInstanceId(); @@ -1046,7 +1049,7 @@ public class TaskServiceImpl implements ITaskService { private R terminateProcessInstance(String reqProcessInstanceId, String finishReason) { if (StrUtil.isBlank(reqProcessInstanceId)) { - return R.failed("流程实例ID不能为空"); + return R.failed("申请信息不完整，请重新打开"); } List allStopProcessInstanceIdList = getAllStopProcessInstanceIdList(reqProcessInstanceId); diff --git a/pigx-flow/pigx-flow-biz/src/test/java/com/pig4cloud/pigx/flow/service/FlowWorkspaceTests.java b/pigx-flow/pigx-flow-biz/src/test/java/com/pig4cloud/pigx/flow/service/FlowWorkspaceTests.java index 188158231..6c6fc8b3d 100644 --- a/pigx-flow/pigx-flow-biz/src/test/java/com/pig4cloud/pigx/flow/service/FlowWorkspaceTests.java +++ b/pigx-flow/pigx-flow-biz/src/test/java/com/pig4cloud/pigx/flow/service/FlowWorkspaceTests.java @@ -1,6 +1,9 @@ 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; @@ -43,6 +46,10 @@ class FlowWorkspaceTests { private FlowDraftService drafts; + private RemoteMessageService messages; + + private final List sentNotices = new ArrayList<>(); + private HttpServer server; private final List eventIds = new ArrayList<>(); @@ -61,7 +68,7 @@ class FlowWorkspaceTests { 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))"); + "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(); @@ -92,7 +99,13 @@ class FlowWorkspaceTests { 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)); - var target = new FlowReliabilityService(jdbc, nodes, engine.getTaskService()); + 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(); @@ -106,7 +119,7 @@ class FlowWorkspaceTests { - + """) @@ -137,25 +150,34 @@ class FlowWorkspaceTests { assertThat(logMapper.writeValueAsString(new Object[] { input })).doesNotContain("private-title", "secret", "payload", "title"); assertThat(FlowNoticeStatusEnum.values()).extracting(FlowNoticeStatusEnum::getCode) - .containsExactly("PENDING", "SENT", "CANCELLED", "DISABLED"); + .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)).hasSize(1); + 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()); + jdbc.update("INSERT INTO process_instance_record VALUES(1,?,1,7,1,'0','差旅申请')", instance.getId()); var task = engine.getTaskService().createTaskQuery().singleResult(); - assertThat(reliability.notices(1, 1001, 1)).hasSize(1); - String notice = reliability.notices(1, 1001, 1).get(0).get("id").toString(); - reliability.read(1, 1001, notice); - assertThat(reliability.notices(1, 1001, 1).get(0).get("read_at")).isNotNull(); - assertThat(reliability.notices(1, 0, 1)).isEmpty(); + 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()); @@ -177,12 +199,19 @@ class FlowWorkspaceTests { var scheduler = new FlowDurableConfiguration(jdbc, provider, mock(ObjectProvider.class)); scheduler.scanReminders(); scheduler.scanReminders(); - assertThat(reliability.notices(1, 2002, 1)).hasSize(2); + 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(reliability.notices(1, 1001, 1)).hasSize(2); + 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(); @@ -237,11 +266,26 @@ class FlowWorkspaceTests { })).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()); + 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 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); diff --git a/pigx-flow/pigx-flow-biz/src/test/java/com/pig4cloud/pigx/flow/service/impl/TaskServiceImplTests.java b/pigx-flow/pigx-flow-biz/src/test/java/com/pig4cloud/pigx/flow/service/impl/TaskServiceImplTests.java index d147636f5..706050c12 100644 --- a/pigx-flow/pigx-flow-biz/src/test/java/com/pig4cloud/pigx/flow/service/impl/TaskServiceImplTests.java +++ b/pigx-flow/pigx-flow-biz/src/test/java/com/pig4cloud/pigx/flow/service/impl/TaskServiceImplTests.java @@ -189,6 +189,15 @@ class TaskServiceImplTests { TenantContextHolder.clear(); } + @Test + void queryTaskRejectsUnavailableTaskInHandlingMode() { + TaskQuery query = mock(TaskQuery.class, org.mockito.Answers.RETURNS_SELF); + when(taskService.createTaskQuery()).thenReturn(query); + R result = taskServiceImpl.queryTask("finished-or-transferred", false); + assertThat(result.getCode()).isEqualTo(1); + assertThat(result.getMsg()).isEqualTo("任务已结束或已转交，无法办理"); + } + @Test void stopProcessInstancePropagatesTerminateReasonBeforeMovingToEnd() { ChangeActivityStateBuilder stateChangeBuilder = prepareTerminationFlow(null); diff --git a/pigx-upms/pigx-upms-api/src/main/java/com/pig4cloud/pigx/admin/api/dto/MessageNoticeDTO.java b/pigx-upms/pigx-upms-api/src/main/java/com/pig4cloud/pigx/admin/api/dto/MessageNoticeDTO.java index 78e76577b..95d8fdeec 100644 --- a/pigx-upms/pigx-upms-api/src/main/java/com/pig4cloud/pigx/admin/api/dto/MessageNoticeDTO.java +++ b/pigx-upms/pigx-upms-api/src/main/java/com/pig4cloud/pigx/admin/api/dto/MessageNoticeDTO.java @@ -50,4 +50,13 @@ public class MessageNoticeDTO { */ private Integer sort = 0; + /** 业务类型；FLOW_TASK 表示流程审批任务，空值按普通消息打开。 */ + private String businessType; + + /** 业务记录 ID；FLOW_TASK 时为任务 ID。 */ + private String businessId; + + /** 调用方事件标识，同一租户内用于重试去重；普通消息可不传。 */ + private String requestKey; + } diff --git a/pigx-upms/pigx-upms-api/src/main/java/com/pig4cloud/pigx/admin/api/entity/SysMessageEntity.java b/pigx-upms/pigx-upms-api/src/main/java/com/pig4cloud/pigx/admin/api/entity/SysMessageEntity.java index 0a5b3b2d8..12ac7a140 100644 --- a/pigx-upms/pigx-upms-api/src/main/java/com/pig4cloud/pigx/admin/api/entity/SysMessageEntity.java +++ b/pigx-upms/pigx-upms-api/src/main/java/com/pig4cloud/pigx/admin/api/entity/SysMessageEntity.java @@ -107,4 +107,13 @@ public class SysMessageEntity extends Model { @Schema(description = "租户") private Long tenantId; + /** 业务类型；FLOW_TASK 表示流程审批任务，空值按普通消息打开。 */ + private String businessType; + + /** 业务记录 ID；FLOW_TASK 时为任务 ID。 */ + private String businessId; + + /** 调用方事件标识，同一租户内用于重试去重；普通消息可不传。 */ + private String requestKey; + } diff --git a/pigx-upms/pigx-upms-api/src/main/java/com/pig4cloud/pigx/admin/api/feign/RemoteMessageService.java b/pigx-upms/pigx-upms-api/src/main/java/com/pig4cloud/pigx/admin/api/feign/RemoteMessageService.java index 1b3688b15..7492dc9e2 100644 --- a/pigx-upms/pigx-upms-api/src/main/java/com/pig4cloud/pigx/admin/api/feign/RemoteMessageService.java +++ b/pigx-upms/pigx-upms-api/src/main/java/com/pig4cloud/pigx/admin/api/feign/RemoteMessageService.java @@ -25,6 +25,7 @@ import com.pig4cloud.pigx.admin.api.dto.MessageNoticeDTO; import com.pig4cloud.pigx.admin.api.dto.MessageSmsDTO; import com.pig4cloud.pigx.common.core.constant.ServiceNameConstants; import com.pig4cloud.pigx.common.core.util.R; +import com.pig4cloud.pigx.common.feign.annotation.NoToken; import org.springframework.cloud.openfeign.FeignClient; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestBody; @@ -71,7 +72,8 @@ public interface RemoteMessageService { * @param noticeDTO 站内消息 DTO * @return {@link R } */ - @PostMapping("/sysMessage/send/notice") + @NoToken + @PostMapping("/sysMessage/internal/notice") R sendNotice(@RequestBody MessageNoticeDTO noticeDTO); } diff --git a/pigx-upms/pigx-upms-biz/src/main/java/com/pig4cloud/pigx/admin/controller/SysMessageController.java b/pigx-upms/pigx-upms-biz/src/main/java/com/pig4cloud/pigx/admin/controller/SysMessageController.java index 3f59c6c82..edbfd2858 100644 --- a/pigx-upms/pigx-upms-biz/src/main/java/com/pig4cloud/pigx/admin/controller/SysMessageController.java +++ b/pigx-upms/pigx-upms-biz/src/main/java/com/pig4cloud/pigx/admin/controller/SysMessageController.java @@ -267,4 +267,16 @@ public class SysMessageController { return sysMessageService.sendNotice(noticeDTO); } + /** + * 业务服务内部发送站内信，复用统一消息存储和接收人关系。 + * @param noticeDTO 消息及业务跳转信息 + * @return 消息 ID + */ + @Inner + @PostMapping("/internal/notice") + @Operation(summary = "内部发送站内信") + public R sendInternalNotice(@Valid @RequestBody MessageNoticeDTO noticeDTO) { + return sysMessageService.sendNotice(noticeDTO); + } + } diff --git a/pigx-upms/pigx-upms-biz/src/main/java/com/pig4cloud/pigx/admin/service/impl/SysMessageServiceImpl.java b/pigx-upms/pigx-upms-biz/src/main/java/com/pig4cloud/pigx/admin/service/impl/SysMessageServiceImpl.java index 9e3fe80d9..af4ba0555 100644 --- a/pigx-upms/pigx-upms-biz/src/main/java/com/pig4cloud/pigx/admin/service/impl/SysMessageServiceImpl.java +++ b/pigx-upms/pigx-upms-biz/src/main/java/com/pig4cloud/pigx/admin/service/impl/SysMessageServiceImpl.java @@ -608,6 +608,16 @@ public class SysMessageServiceImpl extends ServiceImpllambdaQuery() + .eq(SysMessageEntity::getRequestKey, noticeDTO.getRequestKey())); + if (existing != null) { + return R.ok(existing.getId()); + } + } + if (!YesNoEnum.YES.getCode().equals(noticeDTO.getAllFlag()) && CollUtil.isEmpty(noticeDTO.getUserIds())) { + return R.failed("指定用户站内信必须提供接收人"); + } // 构建消息VO SysMessageVO messageVO = new SysMessageVO(); messageVO.setCategory(noticeDTO.getCategory()); @@ -615,6 +625,9 @@ public class SysMessageServiceImpl extends ServiceImpl(); + when(mapper.selectOne(any(LambdaQueryWrapper.class))).thenAnswer(invocation -> saved.get()); + when(mapper.insert(any(SysMessageEntity.class))).thenAnswer(invocation -> { + SysMessageEntity entity = invocation.getArgument(0); + entity.setId(123L); + saved.set(entity); + return 1; + }); + var notice = new MessageNoticeDTO(); + notice.setCategory("1"); + notice.setTitle("待办:财务审批"); + notice.setContent("请打开审批任务处理"); + notice.setUserIds(List.of(1001L)); + notice.setBusinessType("FLOW_TASK"); + notice.setBusinessId("task-A"); + notice.setRequestKey("flow:event-A"); + assertThat(service.sendNotice(notice).getData()).isEqualTo(123L); + assertThat(service.sendNotice(notice).getData()).isEqualTo(123L); + verify(mapper, times(1)).insert(any(SysMessageEntity.class)); + assertThat(saved.get().getBusinessType()).isEqualTo("FLOW_TASK"); + assertThat(saved.get().getBusinessId()).isEqualTo("task-A"); + assertThat(saved.get().getRequestKey()).isEqualTo("flow:event-A"); + var recipient = ArgumentCaptor.forClass(SysMessageRelationEntity.class); + verify(relations, times(1)).insert(recipient.capture()); + assertThat(recipient.getValue().getMsgId()).isEqualTo(123L); + assertThat(recipient.getValue().getUserId()).isEqualTo(1001L); + } +}