diff --git a/db/0pigxx_boot.sql b/db/0pigxx_boot.sql index 1152a1e7b..2b0d0930d 100644 --- a/db/0pigxx_boot.sql +++ b/db/0pigxx_boot.sql @@ -1568,7 +1568,8 @@ CREATE TABLE `sys_message_relation` ( `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`) + PRIMARY KEY (`id`), + KEY `idx_relation_msg_user` (`msg_id`,`user_id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT='系统消息推送记录'; @@ -1852,7 +1853,7 @@ CREATE TABLE `flow_notice` ( `task_id` varchar(64) COLLATE utf8mb4_general_ci NOT NULL COMMENT '引擎任务ID', `instance_id` varchar(64) COLLATE utf8mb4_general_ci NOT NULL COMMENT '引擎流程实例ID', `created_at` timestamp(3) NOT NULL COMMENT '创建时间', - `status` varchar(16) COLLATE utf8mb4_general_ci NOT NULL DEFAULT 'SENT' COMMENT '状态：PENDING 待到期，READY 待投递，SENT 已投递，CANCELLED 已取消，DISABLED 已停用', + `status` varchar(16) COLLATE utf8mb4_general_ci NOT NULL DEFAULT 'SENT' COMMENT '状态：PENDING 待到期，READY 待投递，SENT 已投递，REMOVING 待撤销，CANCELLED 已取消，DISABLED 已停用', `scheduled_at` timestamp(3) NULL DEFAULT NULL COMMENT '计划投递或到期时间', `sent_at` timestamp(3) NULL DEFAULT NULL COMMENT '投递到统一站内信的时间', `reminder_key` varchar(64) COLLATE utf8mb4_general_ci DEFAULT NULL COMMENT '超时提醒任务ID，每个任务最多一条', @@ -1860,7 +1861,8 @@ CREATE TABLE `flow_notice` ( UNIQUE KEY `uk_notice_event` (`tenant_id`,`user_id`,`event_key`), UNIQUE KEY `uk_notice_reminder` (`tenant_id`,`reminder_key`), KEY `idx_notice_owner` (`tenant_id`,`user_id`,`created_at`), - KEY `idx_notice_due` (`status`,`scheduled_at`) + KEY `idx_notice_due` (`status`,`scheduled_at`), + KEY `idx_notice_task` (`tenant_id`,`task_id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT='流程通知事件'; -- ---------------------------- diff --git a/db/2pigxx.sql b/db/2pigxx.sql index 3f8246729..7f18d9779 100644 --- a/db/2pigxx.sql +++ b/db/2pigxx.sql @@ -1519,7 +1519,8 @@ CREATE TABLE `sys_message_relation` ( `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`) + PRIMARY KEY (`id`), + KEY `idx_relation_msg_user` (`msg_id`,`user_id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT='系统消息推送记录'; DROP TABLE IF EXISTS `sys_system_config`; diff --git a/db/3pigxx_flow.sql b/db/3pigxx_flow.sql index 67c4e344d..f92358de1 100644 --- a/db/3pigxx_flow.sql +++ b/db/3pigxx_flow.sql @@ -319,7 +319,7 @@ CREATE TABLE `flow_notice` ( `task_id` varchar(64) COLLATE utf8mb4_general_ci NOT NULL COMMENT '引擎任务ID', `instance_id` varchar(64) COLLATE utf8mb4_general_ci NOT NULL COMMENT '引擎流程实例ID', `created_at` timestamp(3) NOT NULL COMMENT '创建时间', - `status` varchar(16) COLLATE utf8mb4_general_ci NOT NULL DEFAULT 'SENT' COMMENT '状态：PENDING 待到期，READY 待投递，SENT 已投递，CANCELLED 已取消，DISABLED 已停用', + `status` varchar(16) COLLATE utf8mb4_general_ci NOT NULL DEFAULT 'SENT' COMMENT '状态：PENDING 待到期，READY 待投递，SENT 已投递，REMOVING 待撤销，CANCELLED 已取消，DISABLED 已停用', `scheduled_at` timestamp(3) NULL DEFAULT NULL COMMENT '计划投递或到期时间', `sent_at` timestamp(3) NULL DEFAULT NULL COMMENT '投递到统一站内信的时间', `reminder_key` varchar(64) COLLATE utf8mb4_general_ci DEFAULT NULL COMMENT '超时提醒任务ID，每个任务最多一条', @@ -327,7 +327,8 @@ CREATE TABLE `flow_notice` ( UNIQUE KEY `uk_notice_event` (`tenant_id`,`user_id`,`event_key`), UNIQUE KEY `uk_notice_reminder` (`tenant_id`,`reminder_key`), KEY `idx_notice_owner` (`tenant_id`,`user_id`,`created_at`), - KEY `idx_notice_due` (`status`,`scheduled_at`) + KEY `idx_notice_due` (`status`,`scheduled_at`), + KEY `idx_notice_task` (`tenant_id`,`task_id`) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT='流程通知事件'; SET FOREIGN_KEY_CHECKS = 1; diff --git a/db/migration/20261013_flow_node_version.md b/db/migration/20261013_flow_node_version.md deleted file mode 100644 index d4c5f2670..000000000 --- a/db/migration/20261013_flow_node_version.md +++ /dev/null @@ -1,69 +0,0 @@ -# 节点配置版本隔离升级 - -本次只隔离节点配置。业务 flowId 不变；流程图、表单定义的完整历史快照不在本次范围。任务、抄送和实例详情使用对应版本的字段权限，当前表单中未被该快照声明的字段默认隐藏；这不替代服务端的提交字段鉴权。已结束实例的详情依赖 Flowable 历史数据来确定版本，历史已清理时会明确报错，不能回退最新版。 - -## 上线顺序 - -1. 暂停流程发布、发起、任务处理和异步执行器，备份流程业务表及 Flowable 数据。不得新旧版本混跑：旧代码会按 flowId + nodeId 删除节点记录。 -2. 对已有库执行同目录 20261013_flow_node_version.sql；新库直接使用初始化脚本。升级 SQL 只执行一次，不删除旧配置。 -3. 核对存量定义与节点配置，建立明确的版本映射。至少覆盖所有运行中定义；如需要查询历史任务、抄送或使用旧定义，也必须覆盖对应历史定义。 -4. 将已核验配置写入对应 process_definition_id，包含 root 和所有业务节点。每个租户、定义、节点只能有一条快照。按该版本的已核验表单字段集合补齐各节点 formPerms：原来未配置/空值的字段按原有前端默认语义补 E，已有 H/R/E 保留；包含 children 中的嵌套字段及带 field 的容器。不能用当前表单的新增字段补旧版权限。无法恢复原字段集合时，此版本迁移尚未完成。 -5. 复查覆盖情况，部署全部新代码后恢复业务。无法确认映射时不要恢复受影响流程；运行期会明确报“缺少节点版本快照”，不会回退读取最新版。 - -## 只读盘点 - -在工作流所在数据库执行以下查询；若自定义了 Flowable 表名前缀，请先调整。 - - SELECT DISTINCT d.TENANT_ID_, d.KEY_, d.ID_, d.VERSION_, d.DEPLOYMENT_ID_ - FROM ACT_RU_EXECUTION e - JOIN ACT_RE_PROCDEF d ON d.ID_ = e.PROC_DEF_ID_ - ORDER BY d.TENANT_ID_, d.KEY_, d.VERSION_; - - SELECT p.tenant_id, p.flow_id, p.unique_id, d.ID_, d.VERSION_, p.process - FROM process p - JOIN ACT_RE_PROCDEF d - ON d.DEPLOYMENT_ID_ = p.unique_id - AND d.KEY_ = p.flow_id - AND d.TENANT_ID_ = CAST(p.tenant_id AS CHAR) - WHERE p.del_flag = '0'; - - SELECT id, tenant_id, flow_id, node_id, del_flag, process_definition_id, data - FROM process_node_data - ORDER BY tenant_id, flow_id, node_id, id; - -process.unique_id 存储部署 ID，可以定位当前模板的引擎定义。它只能帮助定位当前发布版本，不代表所有旧节点记录都属于该版本。旧节点表保留了逻辑删除数据，但没有发布版本标记；不能按行 ID 或时间顺序自动配对。 - -核对节点 JSON 与相应发布资料/备份，包括审批人、多实例模式、拒绝策略、字段权限、触发器和通知地址。配置中的敏感业务内容不要写入审查日志。 - -## 存量写入模板 - -下面只是单行映射示例，所有占位值都必须替换为已核验结果。先 SELECT 确认目标行和定义的租户、业务 key、版本一致，再在受控事务中执行。不要按 flowId 批量绑定最新版。 - - UPDATE process_node_data n - JOIN ACT_RE_PROCDEF d - ON d.ID_ = '<已核验的流程定义ID>' - AND d.KEY_ = n.flow_id - AND d.TENANT_ID_ = CAST(n.tenant_id AS CHAR) - SET n.process_definition_id = d.ID_, n.del_flag = '0' - WHERE n.id = <已核验的节点记录ID> - AND n.tenant_id = <目标租户ID> - AND n.process_definition_id IS NULL; - -一条旧记录如果需属于多个已验证版本，应分别插入不同 ID 的快照副本；不要反复改写同一行的版本 ID。若需从原始设计 JSON 重建节点，注意前端的 nodeName/childNode 与数据库 JSON 的 name/children 映射，并保存所有节点。无法恢复的历史版本应明确列为阻塞，不能用最新版替代。 - -迁移后检查所有运行实例是否有 root 快照： - - SELECT DISTINCT d.ID_, d.KEY_, d.TENANT_ID_ - FROM ACT_RU_EXECUTION e - JOIN ACT_RE_PROCDEF d ON d.ID_ = e.PROC_DEF_ID_ - LEFT JOIN process_node_data n - ON n.process_definition_id = d.ID_ - AND CAST(n.tenant_id AS CHAR) = d.TENANT_ID_ - AND n.node_id = 'root' AND n.del_flag = '0' - WHERE n.id IS NULL; - -该查询必须无遗漏；还需对照设计检查全部业务节点，不能仅用 root 存在代替完整性核验。新发布时程序先按当时的表单字段集合补齐默认 E 权限，再在部署事务中写快照，后续不得人工修改已发布数据。 - -## 回退 - -不能在产生新快照后直接恢复旧代码继续写流程。需要先冻结流程写入，再根据备份和期间新增实例单独制定回退方案。不要直接删除快照或批量清空版本列。 diff --git a/db/migration/20261013_flow_node_version.sql b/db/migration/20261013_flow_node_version.sql deleted file mode 100644 index 4951bf3ef..000000000 --- a/db/migration/20261013_flow_node_version.sql +++ /dev/null @@ -1,6 +0,0 @@ --- 升级已有数据库；全新数据库使用 0pigxx_boot.sql 或 3pigxx_flow.sql。 --- 执行前停止流程写入并备份；本脚本不猜测存量节点与流程版本的对应关系。 --- 必须完成同目录说明中的存量核对后再上线应用。 -ALTER TABLE process_node_data - ADD COLUMN process_definition_id varchar(64) DEFAULT NULL COMMENT '引擎流程定义ID，发布版本快照', - ADD UNIQUE KEY uk_node_definition (tenant_id, process_definition_id, node_id); diff --git a/db/migration/20261014_flow_first_batch.md b/db/migration/20261014_flow_first_batch.md deleted file mode 100644 index 2c03ac50a..000000000 --- a/db/migration/20261014_flow_first_batch.md +++ /dev/null @@ -1,61 +0,0 @@ -# 流程第一批能力升级 - -## 范围与入口 - -- 流程设计器：保存设计草稿、设计草稿箱；保存基本信息、表单和节点树，不部署流程。 -- 流程列表：申请草稿箱；动态表单发起抽屉可保存表单值和自选审批人。按用户和租户隔离，版本号防止多窗口覆盖。 -- 待办页：流程通知，包含任务分配、发起人催办和超时提醒；点击通知沿用现有任务详情权限检查。历史通知可能对应已完成或已转交任务。 -- 我发起的：运行中流程可催办，同一实例十分钟内最多一次；仅发起人可以操作。回调记录可查看状态和手动重试失败事件。 -- 审批节点 / 事件通知：配置自然时长分钟数，0 或留空关闭；最多 525600 分钟。计时基于任务创建时间，不因转交重新计时，到期只提醒当时的当前审批人一次。 - -本批申请草稿适用于流程引擎的动态表单。自定义业务组件自行提交、拥有独立的数据模型，继续由业务页面保存草稿；未擅自修改业务表单合同。申请草稿记录发布 deployment ID，恢复时版本变化会拒绝自动恢复并保留原草稿。普通草稿只在明确保存或设计器点击下一步时持久化，未保存的输入不会自动落库。 - -## 升级顺序 - -1. 备份数据库并停止本次流程写入，确认上一次节点版本隔离迁移已经完成。 -2. 在流程与引擎使用的同一 MySQL 数据源执行 `20261014_flow_first_batch.sql`。新增 `flow_draft`、`flow_notice`、`flow_callback` 三张表，不改动存量业务表；新装脚本 `0pigxx_boot.sql`、`3pigxx_flow.sql` 已包含相同建表语句。 -3. 部署后端与管理端。不要混用新旧后端：旧版本不会创建持久回调与通知记录。 -4. 使用测试租户检查保存和恢复两类草稿、任务分配与转交、催办限频、到期提醒、回调 503 后重试。 - -新部署的监听器负责此后发生的任务事件；不自动重放过去的任务通知或已完成回调，不自动为已存在且没有再次分配的任务补建超时记录。 - -## 已安装五表版本的兼容升级 - -仅已经执行过旧五表脚本的环境需要此步骤；新装或尚未安装本批能力的用户只执行三表建表脚本。 - -1. 停止旧后端及其他流程写入，完整备份原五张表。恢复备份前也必须停止新后端，不能混跑两版。 -2. 执行三表建表脚本 `20261014_flow_first_batch.sql`，再执行 `20261014_flow_merge_tables.sql`（使用支持 `DELIMITER` 的 MySQL 客户端）。合并脚本只补字段和索引；数据部分使用事务。发生异常时数据部分回滚，已完成的 DDL 保留，排查后可重新执行。 -3. 合并脚本保留所有草稿和现有通知的 ID、已读状态；把超时监听关联到已有超时通知，或转换为待触发记录。已提醒的监听不会重新排队；未启用截止时间的旧记录为 `DISABLED`。旧回调的 ID、状态、次数、租约、地址和载荷保持原值，尝试历史按原 ID 合入 JSON 数组。孤立审计记录会报错并回滚，不会静默丢弃。 -4. 核对下面两个查询结果均为 0，并核对备份中的草稿、通知和回调业务字段。合并脚本可重复执行，不会重复追加历史或重置已处理的提醒。 -5. 旧 `flow_task_watch` 和 `flow_callback_attempt` 默认保留为只读核对来源，新代码不再访问。确认备份可用、迁移完整后可清理这两张旧表。旧监听中的 `last_urge_at` 仅写不读，原值保存在旧表和备份；催办限频仍以通知记录为准。 - -```sql --- 未迁移的监听数 -SELECT COUNT(*) FROM flow_task_watch w -WHERE NOT EXISTS ( - SELECT 1 FROM flow_notice n WHERE n.tenant_id=w.tenant_id AND n.reminder_key=w.task_id -); --- 未迁移的回调尝试数 -SELECT COUNT(*) FROM flow_callback_attempt a -LEFT JOIN flow_callback c ON c.id=a.callback_id AND c.tenant_id=a.tenant_id -WHERE c.id IS NULL OR JSON_SEARCH(COALESCE(c.history,'[]'),'one',a.id,NULL,'$[*].id') IS NULL; -``` - -## 事务和投递约定 - -- 通知与回调写入使用 `JdbcTemplate`，与 Flowable 共用 Spring 事务及数据源；独立监听器 `isFailOnException=true`，写入失败会回滚任务操作，避免任务完成但回调丢失。 -- 新表 SQL 显式指定租户、用户或引擎事件身份；后台跨租户扫描仅取内部记录，HTTP API 不允许指定他人的用户或租户。 -- `flow_notice` 同时保存已生成通知和待触发提醒。`status` 区分 `PENDING`、`SENT`、`CANCELLED`、`DISABLED`；通知列表和已读操作只接受 `SENT`。`reminder_key` 按租户和任务唯一，到期才确定当前接收人；任务结束取消尚未发送的提醒，已发送记录保留。任务暂停或暂无处理人时保留 `PENDING`，将 `scheduled_at` 推迟十秒后再检查，避免占满最早一批扫描记录；恢复后继续发送，不重新计算完整超时时长。 -- 后台默认每轮完成后间隔 10 秒扫描，可配置 `flow.delivery.scan-delay`（毫秒）。回调与提醒使用本模块独立的两线程调度器；每批上限为 20 个回调、100 个提醒，不承诺精确到秒的触发。 -- HTTP 回调使用任务完成时固定的目标地址、JSON 载荷和事件 ID。只接受 2xx；重定向、其他状态码、网络错误均进入失败处理。请求超时 10 秒，不携带用户 Bearer token。 -- 首次共尝试 5 次，失败后按 30、60、120、240 秒退避。60 秒租约与条件更新允许多实例竞争；进程崩溃后可回收过期租约继续投递，达到尝试上限的未知结果标为失败。 -- 请求头 `X-Flow-Event-Id` 和 `Idempotency-Key` 使用同一个稳定事件 ID。语义是**至少一次投递**：接收方必须把事件 ID 去重与自身业务写入放在同一事务中，才能避免“已处理但响应丢失”造成重复副作用。 -- 手动重试仅允许实例发起人对失败记录操作，保留事件 ID、目标和载荷，再给予 5 次尝试。累计次数不清零。 -- `flow_callback.history` 以 JSON 数组保存每次尝试和手动重试的审计记录，与主记录状态在同一事务中更新，不截断历史。审计归属由父记录的租户和事件 ID 确定。STARTED 表示尝试已开始但没有确认结果；可能是进程退出或租约失效。页面只展示状态、次数和最近结果，不返回回调载荷和目标 URL。 -- 本批可靠投递针对节点“事件通知”HTTP 回调；同步触发器业务调用、Redis 流程状态广播维持原有行为。 - -## 验证证据 - -`FlowWorkspaceTests` 使用真实 Spring Flowable、共享事务的 H2、生产建表语句（H2 MySQL 兼容模式）和本地 HTTP 服务，覆盖草稿持久化、所有者隔离、乐观锁、分配/转交通知、超时去重、催办限频、503 重试、固定事件 ID，以及回调写入失败后引擎任务回滚。H2 验证不等同于真实 MySQL 或完整应用运行验证。 - -相关既有节点版本、会签和任务服务测试一并执行。管理端进行指定文件 ESLint、Prettier、Vue SFC 编译，以及已有版本隔离浏览器回归。三表合并额外验证了历史追加、手动重试保留旧历史、已完成任务不再提醒。迁移在本地 OceanBase seekdb 的临时数据库使用带数据样本验证，并检查重复执行和孤立审计回滚；临时库测试后清理。真实账号完整联调尚未执行，不把接口隔离回归当成真实后端验收。 diff --git a/db/migration/20261014_flow_first_batch.sql b/db/migration/20261014_flow_first_batch.sql deleted file mode 100644 index cd363ea47..000000000 --- a/db/migration/20261014_flow_first_batch.sql +++ /dev/null @@ -1,27 +0,0 @@ --- 第一批持久化能力：三张业务表。已安装旧五表版本时，建表后另执行 20261014_flow_merge_tables.sql。 --- 升级前备份并停写，先执行 DDL 再部署应用；不自动迁移运行中的任务。 -CREATE TABLE IF NOT EXISTS flow_draft ( - id varchar(36) NOT NULL, tenant_id bigint NOT NULL, user_id bigint NOT NULL, - 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; -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, - task_id varchar(64) NOT NULL, instance_id varchar(64) NOT NULL, - created_at timestamp(3) NOT NULL, read_at timestamp(3) NULL, - status varchar(16) NOT NULL DEFAULT 'SENT', scheduled_at timestamp(3) NULL, sent_at timestamp(3) NULL, - 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; -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, - status varchar(16) NOT NULL, attempts int NOT NULL DEFAULT 0, max_attempts int NOT NULL DEFAULT 5, next_at timestamp(3) NOT NULL, - history longtext NULL, - 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; diff --git a/db/migration/20261014_flow_merge_tables.sql b/db/migration/20261014_flow_merge_tables.sql deleted file mode 100644 index 0200d9a4f..000000000 --- a/db/migration/20261014_flow_merge_tables.sql +++ /dev/null @@ -1,100 +0,0 @@ --- 仅用于已经安装旧五表版本的库。先备份、停止流程写入及后台扫描，并执行三表建表脚本。 --- 可重复执行；旧表暂保留作为核对来源，新代码不再访问。核验并备份后再清理旧表。 -DELIMITER $$ -DROP PROCEDURE IF EXISTS migrate_flow_three_tables$$ -CREATE PROCEDURE migrate_flow_three_tables() -BEGIN - DECLARE finished INT DEFAULT 0; - DECLARE v_task VARCHAR(64); - DECLARE v_tenant BIGINT; - DECLARE v_instance VARCHAR(64); - DECLARE v_due TIMESTAMP(3); - DECLARE v_reminded TIMESTAMP(3); - DECLARE v_id VARCHAR(36); - DECLARE v_callback VARCHAR(36); - DECLARE v_attempt INT; - DECLARE v_result VARCHAR(16); - DECLARE v_detail VARCHAR(500); - DECLARE v_created TIMESTAMP(3); - DECLARE v_history LONGTEXT; - DECLARE watches CURSOR FOR SELECT task_id,tenant_id,instance_id,due_at,reminded_at FROM flow_task_watch; - DECLARE audits CURSOR FOR SELECT id,callback_id,tenant_id,attempt_no,result,detail,created_at FROM flow_callback_attempt ORDER BY created_at,id; - DECLARE CONTINUE HANDLER FOR NOT FOUND SET finished=1; - - IF NOT EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='flow_notice' AND column_name='status') THEN - ALTER TABLE flow_notice ADD COLUMN status VARCHAR(16) NOT NULL DEFAULT 'SENT'; - END IF; - IF NOT EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='flow_notice' AND column_name='scheduled_at') THEN - ALTER TABLE flow_notice ADD COLUMN scheduled_at TIMESTAMP(3) NULL; - END IF; - IF NOT EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='flow_notice' AND column_name='sent_at') THEN - ALTER TABLE flow_notice ADD COLUMN sent_at TIMESTAMP(3) NULL; - END IF; - IF NOT EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='flow_notice' AND column_name='reminder_key') THEN - ALTER TABLE flow_notice ADD COLUMN reminder_key VARCHAR(64) NULL; - END IF; - IF NOT EXISTS (SELECT 1 FROM information_schema.statistics WHERE table_schema=DATABASE() AND table_name='flow_notice' AND index_name='uk_notice_reminder') THEN - ALTER TABLE flow_notice ADD UNIQUE KEY uk_notice_reminder(tenant_id,reminder_key); - END IF; - IF NOT EXISTS (SELECT 1 FROM information_schema.statistics WHERE table_schema=DATABASE() AND table_name='flow_notice' AND index_name='idx_notice_due') THEN - ALTER TABLE flow_notice ADD KEY idx_notice_due(status,scheduled_at); - END IF; - IF NOT EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema=DATABASE() AND table_name='flow_callback' AND column_name='history') THEN - ALTER TABLE flow_callback ADD COLUMN history LONGTEXT NULL; - END IF; - - -- DDL 自动提交；以下数据搬迁独立事务，任何异常回滚数据部分，修正后可重跑。 - BEGIN - DECLARE EXIT HANDLER FOR SQLEXCEPTION - BEGIN - ROLLBACK; - RESIGNAL; - END; - START TRANSACTION; - UPDATE flow_notice SET sent_at=created_at WHERE status='SENT' AND sent_at IS NULL; - IF EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema=DATABASE() AND table_name='flow_task_watch') THEN - SET finished=0; - OPEN watches; - watch_loop: LOOP - FETCH watches INTO v_task,v_tenant,v_instance,v_due,v_reminded; - IF finished=1 THEN LEAVE watch_loop; END IF; - IF NOT EXISTS (SELECT 1 FROM flow_notice WHERE tenant_id=v_tenant AND reminder_key=v_task) THEN - IF (SELECT COUNT(*) FROM flow_notice WHERE tenant_id=v_tenant AND event_key=CONCAT('OVERDUE:',v_task) AND kind='OVERDUE') > 1 THEN - SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='Multiple overdue notices for one watch; inspect before migration'; - END IF; - IF EXISTS (SELECT 1 FROM flow_notice WHERE tenant_id=v_tenant AND event_key=CONCAT('OVERDUE:',v_task) AND kind='OVERDUE') THEN - UPDATE flow_notice SET reminder_key=v_task,scheduled_at=v_due,sent_at=COALESCE(v_reminded,sent_at) - WHERE tenant_id=v_tenant AND event_key=CONCAT('OVERDUE:',v_task) AND kind='OVERDUE'; - ELSE - INSERT INTO flow_notice(id,tenant_id,user_id,event_key,kind,title,task_id,instance_id,created_at,status,scheduled_at,sent_at,reminder_key) - VALUES(UUID(),v_tenant,'',CONCAT('OVERDUE:',v_task),'OVERDUE','流程任务',v_task,v_instance,CURRENT_TIMESTAMP(3), - CASE WHEN v_reminded IS NOT NULL THEN 'SENT' WHEN v_due IS NULL THEN 'DISABLED' ELSE 'PENDING' END,v_due,v_reminded,v_task); - END IF; - END IF; - END LOOP; - CLOSE watches; - END IF; - IF EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema=DATABASE() AND table_name='flow_callback_attempt') THEN - SET finished=0; - OPEN audits; - audit_loop: LOOP - FETCH audits INTO v_id,v_callback,v_tenant,v_attempt,v_result,v_detail,v_created; - IF finished=1 THEN LEAVE audit_loop; END IF; - IF NOT EXISTS (SELECT 1 FROM flow_callback WHERE id=v_callback AND tenant_id=v_tenant) THEN - SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT='Orphan callback attempt; retain old tables and inspect'; - END IF; - SELECT COALESCE(history,'[]') INTO v_history FROM flow_callback WHERE id=v_callback AND tenant_id=v_tenant FOR UPDATE; - IF JSON_SEARCH(v_history,'one',v_id,NULL,'$[*].id') IS NULL THEN - UPDATE flow_callback SET history=JSON_ARRAY_APPEND(v_history,'$',JSON_OBJECT('id',v_id,'attempt_no',v_attempt, - 'result',v_result,'detail',v_detail,'created_at',DATE_FORMAT(v_created,'%Y-%m-%d %H:%i:%s.%f'))) - WHERE id=v_callback AND tenant_id=v_tenant; - END IF; - END LOOP; - CLOSE audits; - END IF; - COMMIT; - END; -END$$ -CALL migrate_flow_three_tables()$$ -DROP PROCEDURE migrate_flow_three_tables$$ -DELIMITER ; 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 485819e71..5b4f5ec89 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 @@ -19,7 +19,10 @@ public enum FlowNoticeStatusEnum { /** 已投递到统一站内信，已读状态由 sys_message_relation 维护。 */ SENT("SENT"), - /** 任务已经结束，取消尚未发送的提醒。 */ + /** 任务结束或接收人变更，等待删除统一站内信的接收记录。 */ + REMOVING("REMOVING"), + + /** 任务已经结束，提醒已取消或接收记录已删除。 */ CANCELLED("CANCELLED"), /** 旧监听没有截止时间，迁移后保留但不参与扫描。 */ diff --git a/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/entity/FlowNotice.java b/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/entity/FlowNotice.java index aafa17a82..5e86df6b2 100644 --- a/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/entity/FlowNotice.java +++ b/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/entity/FlowNotice.java @@ -1,6 +1,8 @@ package com.pig4cloud.pigx.flow.entity; +import com.baomidou.mybatisplus.annotation.FieldStrategy; import com.baomidou.mybatisplus.annotation.IdType; +import com.baomidou.mybatisplus.annotation.TableField; import com.baomidou.mybatisplus.annotation.TableId; import com.baomidou.mybatisplus.annotation.TableName; import com.pig4cloud.pigx.common.core.util.TenantTable; @@ -21,7 +23,8 @@ public class FlowNotice { /** 所属租户 ID。 */ private Long tenantId; - /** 接收人 ID；超时提醒到期前为空串。 */ + /** 接收人 ID；超时提醒到期前为空串，插入时保留该值。 */ + @TableField(insertStrategy = FieldStrategy.NOT_NULL) private String userId; /** 事件去重键，与租户、接收人组成唯一键。 */ 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 80a302744..57e4a34ed 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 @@ -2,6 +2,8 @@ package com.pig4cloud.pigx.flow.config; import com.pig4cloud.pigx.flow.service.IFlowNoticeService; import com.pig4cloud.pigx.flow.support.listeners.FlowDurableEventListener; +import com.pig4cloud.pigx.flow.support.listeners.FlowNoticeRemovalEvent; +import org.springframework.transaction.event.TransactionalEventListener; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.flowable.spring.SpringProcessEngineConfiguration; @@ -95,4 +97,22 @@ public class FlowDurableConfiguration { } } + /** + * 引擎事务提交后立即撤销失效通知；事务回滚时不执行，避免丢失提醒。失败只记录日志，由扫描器重试。 + * @param event 已置为 REMOVING 的通知事件 + */ + @TransactionalEventListener(fallbackExecution = true) + public void removeNoticesAfterCommit(FlowNoticeRemovalEvent event) { + IFlowNoticeService service = noticeService.getObject(); + for (String id : event.noticeIds()) { + try { + service.deliverNotice(id); + } + catch (Exception exception) { + // REMOVING 已随引擎事务提交，扫描器会再次尝试；不能把已成功的审批报告为失败。 + log.warn("流程站内信暂未删除，事件 {}", id, exception); + } + } + } + } diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/IFlowNoticeService.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/IFlowNoticeService.java index 00c13031b..56c941f94 100644 --- a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/IFlowNoticeService.java +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/IFlowNoticeService.java @@ -43,20 +43,22 @@ public interface IFlowNoticeService extends IService { void remind(long tenant, String taskId); /** - * 查询最多 100 条待投递的通知事件。 + * 查询最多 100 条到期的待投递或待撤销通知事件。 * @return 只包含 ID 的通知事件 */ List listDueNotices(); /** - * 投递通知事件到统一站内信；行锁防止扫描器并发发送同一事件。 + * 在独立事务中处理一条通知事件，行锁防止扫描器与提交后回调并发处理同一事件。 + *

+ * READY 且任务仍归原接收人时投递，否则撤销；REMOVING 删除原接收人的接收记录后置为 CANCELLED。 * @param id 通知事件 ID - * @throws IllegalStateException 站内信服务返回失败，由扫描器延迟重试 + * @throws IllegalStateException 站内信发送或删除失败，状态保持不变，由扫描器延迟重试 */ void deliverNotice(String id); /** - * 投递失败后延迟一分钟再试。 + * 投递或撤销失败后延迟一分钟再试。 * @param id 通知事件 ID */ void delayNotice(String id); diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/FlowNoticeServiceImpl.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/FlowNoticeServiceImpl.java index f7e43c308..56aa228a3 100644 --- a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/FlowNoticeServiceImpl.java +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/FlowNoticeServiceImpl.java @@ -5,8 +5,10 @@ import cn.hutool.core.util.EnumUtil; import cn.hutool.core.util.StrUtil; import com.baomidou.mybatisplus.extension.plugins.pagination.Page; import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl; +import com.pig4cloud.pigx.admin.api.dto.FlowMessageRemovalDTO; import com.pig4cloud.pigx.admin.api.dto.MessageNoticeDTO; import com.pig4cloud.pigx.admin.api.feign.RemoteMessageService; +import com.pig4cloud.pigx.common.core.constant.CommonConstants; import com.pig4cloud.pigx.common.core.constant.enums.YesNoEnum; import com.pig4cloud.pigx.common.data.tenant.TenantBroker; import com.pig4cloud.pigx.flow.constant.FlowNoticeKindEnum; @@ -19,6 +21,7 @@ import com.pig4cloud.pigx.flow.entity.FlowNotice; import com.pig4cloud.pigx.flow.entity.ProcessInstanceRecord; import com.pig4cloud.pigx.flow.mapper.FlowNoticeMapper; import com.pig4cloud.pigx.flow.support.listeners.NodeTaskCompleteNotify; +import com.pig4cloud.pigx.flow.support.listeners.FlowNoticeRemovalEvent; import com.pig4cloud.pigx.flow.service.IFlowNoticeService; import com.pig4cloud.pigx.flow.service.IProcessInstanceRecordService; import com.pig4cloud.pigx.flow.service.IProcessNodeDataService; @@ -28,6 +31,8 @@ import org.flowable.engine.TaskService; import org.flowable.task.service.impl.persistence.entity.TaskEntity; import org.springframework.dao.DuplicateKeyException; import org.springframework.stereotype.Service; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.transaction.annotation.Propagation; import org.springframework.transaction.annotation.Transactional; import java.time.LocalDateTime; import java.util.List; @@ -49,6 +54,8 @@ public class FlowNoticeServiceImpl extends ServiceImpl nodeService.getPublishedNode(task.getProcessDefinitionId(), task.getTaskDefinitionKey()) .getData()); if (Objects.isNull(node)) return; if (FlowableEngineEventType.TASK_COMPLETED.name().equals(event)) { - cancelReminder(tenant, task.getId()); if (StrUtil.isNotBlank(node.getEventConfig())) nodeTaskCompleteNotify.notifyTaskCompleted(task, node.getEventConfig()); return; @@ -222,7 +235,7 @@ public class FlowNoticeServiceImpl extends ServiceImpl listDueNotices() { return lambdaQuery().select(FlowNotice::getId) - .eq(FlowNotice::getStatus, FlowNoticeStatusEnum.READY.getCode()) + .in(FlowNotice::getStatus, FlowNoticeStatusEnum.READY.getCode(), FlowNoticeStatusEnum.REMOVING.getCode()) .le(FlowNotice::getScheduledAt, LocalDateTime.now()) .orderByAsc(FlowNotice::getScheduledAt) .page(new Page<>(1, 100, false)) @@ -233,14 +246,25 @@ public class FlowNoticeServiceImpl extends ServiceImpl ids = rows.stream().map(FlowNotice::getId).toList(); + lambdaUpdate().in(FlowNotice::getId, ids) + .eq(FlowNotice::getTenantId, tenant) + .in(FlowNotice::getStatus, FlowNoticeStatusEnum.READY.getCode(), FlowNoticeStatusEnum.SENT.getCode()) + .set(FlowNotice::getStatus, FlowNoticeStatusEnum.REMOVING.getCode()) + .set(FlowNotice::getScheduledAt, LocalDateTime.now()).update(); + eventPublisher.publishEvent(new FlowNoticeRemovalEvent(ids)); + } + + private void removeNotice(FlowNotice row) { + FlowMessageRemovalDTO removal = new FlowMessageRemovalDTO(); + removal.setTaskId(row.getTaskId()); + removal.setRequestKey("flow:" + row.getId()); + removal.setUserId(Long.parseLong(row.getUserId())); + var result = TenantBroker.applyAs(row.getTenantId(), t -> messageService.removeFlowNotice(removal)); + if (Objects.isNull(result) || result.getCode() != CommonConstants.SUCCESS + || !Boolean.TRUE.equals(result.getData())) + throw new IllegalStateException("流程站内信删除失败"); + lambdaUpdate().eq(FlowNotice::getId, row.getId()) + .set(FlowNotice::getStatus, FlowNoticeStatusEnum.CANCELLED.getCode()).update(); + } + private void cancelReminder(long tenant, String task) { lambdaUpdate().set(FlowNotice::getStatus, FlowNoticeStatusEnum.CANCELLED.getCode()) .eq(FlowNotice::getTenantId, tenant) diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/listeners/FlowNoticeRemovalEvent.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/listeners/FlowNoticeRemovalEvent.java new file mode 100644 index 000000000..7fc7ac655 --- /dev/null +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/listeners/FlowNoticeRemovalEvent.java @@ -0,0 +1,11 @@ +package com.pig4cloud.pigx.flow.support.listeners; + +import java.util.List; + +/** + * 引擎事务提交后立即撤销通知；失败仍由持久化状态驱动重试。 + * + * @param noticeIds 已置为 REMOVING 的通知事件 ID + */ +public record FlowNoticeRemovalEvent(List noticeIds) { +} diff --git a/pigx-flow/pigx-flow-biz/src/test/java/com/pig4cloud/pigx/flow/service/impl/FlowNoticeLifecycleTests.java b/pigx-flow/pigx-flow-biz/src/test/java/com/pig4cloud/pigx/flow/service/impl/FlowNoticeLifecycleTests.java new file mode 100644 index 000000000..a51cad9ba --- /dev/null +++ b/pigx-flow/pigx-flow-biz/src/test/java/com/pig4cloud/pigx/flow/service/impl/FlowNoticeLifecycleTests.java @@ -0,0 +1,153 @@ +package com.pig4cloud.pigx.flow.service.impl; + +import com.baomidou.mybatisplus.annotation.FieldStrategy; +import com.baomidou.mybatisplus.core.config.GlobalConfig; +import com.baomidou.mybatisplus.core.toolkit.GlobalConfigUtils; +import com.baomidou.mybatisplus.core.MybatisConfiguration; +import com.baomidou.mybatisplus.core.MybatisSqlSessionFactoryBuilder; +import com.pig4cloud.pigx.admin.api.dto.FlowMessageRemovalDTO; +import com.pig4cloud.pigx.admin.api.feign.RemoteMessageService; +import com.pig4cloud.pigx.common.core.util.R; +import com.pig4cloud.pigx.flow.entity.FlowNotice; +import com.pig4cloud.pigx.flow.mapper.FlowNoticeMapper; +import com.pig4cloud.pigx.flow.service.IProcessNodeDataService; +import com.pig4cloud.pigx.flow.support.listeners.FlowNoticeRemovalEvent; +import org.apache.ibatis.mapping.Environment; +import org.h2.jdbcx.JdbcDataSource; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import org.mybatis.spring.SqlSessionTemplate; +import org.mybatis.spring.transaction.SpringManagedTransactionFactory; +import org.flowable.engine.TaskService; +import org.flowable.task.api.TaskQuery; +import org.flowable.task.service.impl.persistence.entity.TaskEntity; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.datasource.DataSourceTransactionManager; +import org.springframework.test.util.ReflectionTestUtils; +import org.springframework.transaction.support.TransactionTemplate; + +import java.util.UUID; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.*; + +class FlowNoticeLifecycleTests { + + private JdbcTemplate jdbc; + private FlowNoticeServiceImpl service; + private RemoteMessageService messages; + private ApplicationEventPublisher events; + private TransactionTemplate transaction; + + @BeforeEach + void setUp() { + var dataSource = new JdbcDataSource(); + dataSource.setURL("jdbc:h2:mem:" + UUID.randomUUID() + ";DB_CLOSE_DELAY=-1"); + jdbc = new JdbcTemplate(dataSource); + jdbc.execute("create table flow_notice(id varchar(64) primary key, tenant_id bigint, user_id varchar(64) not null, " + + "event_key varchar(255), kind varchar(32), title varchar(255), task_id varchar(64), " + + "instance_id varchar(64), created_at timestamp, status varchar(32), scheduled_at timestamp, " + + "sent_at timestamp, reminder_key varchar(64))"); + var configuration = new MybatisConfiguration(); + configuration.setMapUnderscoreToCamelCase(true); + configuration.setEnvironment(new Environment("test", new SpringManagedTransactionFactory(), dataSource)); + GlobalConfigUtils.setGlobalConfig(configuration, new GlobalConfig() + .setDbConfig(new GlobalConfig.DbConfig().setInsertStrategy(FieldStrategy.NOT_EMPTY))); + configuration.addMapper(FlowNoticeMapper.class); + var factory = new MybatisSqlSessionFactoryBuilder().build(configuration); + messages = mock(RemoteMessageService.class); + events = mock(ApplicationEventPublisher.class); + var tasks = mock(TaskService.class); + var query = mock(TaskQuery.class, RETURNS_SELF); + when(tasks.createTaskQuery()).thenReturn(query); + var nodes = mock(IProcessNodeDataService.class); + when(nodes.getPublishedNode(any(), any())).thenReturn(R.ok()); + service = new FlowNoticeServiceImpl(null, null, nodes, tasks, messages, events); + ReflectionTestUtils.setField(service, "baseMapper", new SqlSessionTemplate(factory).getMapper(FlowNoticeMapper.class)); + transaction = new TransactionTemplate(new DataSourceTransactionManager(dataSource)); + when(messages.removeFlowNotice(any())).thenReturn(R.ok(true)); + } + + @Test + void pendingReminderPersistsEmptyRecipientWithNotEmptyGlobalStrategy() { + var reminder = new FlowNotice(); + reminder.setId("pending-reminder"); + reminder.setUserId(""); + reminder.setStatus("PENDING"); + reminder.setKind("OVERDUE"); + assertThat(service.save(reminder)).isTrue(); + assertThat(jdbc.queryForObject("select user_id from flow_notice where id=?", + String.class, reminder.getId())).isEmpty(); + } + + @ParameterizedTest + @ValueSource(strings = { "ENTITY_DELETED", "TASK_COMPLETED", "TASK_ASSIGNED" }) + void endedOrTransferredTaskRemovesOnlyObsoleteNotices(String event) { + insert("todo", 1, "task-A", "SENT", null); + insert("urge", 1, "task-A", "SENT", null); + insert("queued", 1, "task-A", "READY", null); + insert("overdue", 1, "task-A", "PENDING", "task-A"); + insert("next-task", 1, "task-B", "SENT", null); + insert("other-tenant", 2, "task-A", "SENT", null); + insert("current-owner", 1, "task-A", "SENT", null); + jdbc.update("update flow_notice set user_id='2002' where id='current-owner'"); + var task = mock(TaskEntity.class); + when(task.getId()).thenReturn("task-A"); + when(task.getTenantId()).thenReturn("1"); + when(task.getProcessDefinitionId()).thenReturn("process-A"); + when(task.getTaskDefinitionKey()).thenReturn("node-A"); + when(task.getAssignee()).thenReturn("2002"); + transaction.executeWithoutResult(status -> service.record(task, event)); + verifyNoInteractions(messages); + verify(events).publishEvent(any(FlowNoticeRemovalEvent.class)); + assertThat(state("overdue")).isEqualTo(event.equals("TASK_ASSIGNED") ? "PENDING" : "CANCELLED"); + for (String id : new String[] { "todo", "urge", "queued" }) { + assertThat(state(id)).isEqualTo("REMOVING"); + transaction.executeWithoutResult(status -> service.deliverNotice(id)); + assertThat(state(id)).isEqualTo("CANCELLED"); + } + verify(messages, times(3)).removeFlowNotice(argThat(removal -> + removal.getTaskId().equals("task-A") && removal.getUserId().equals(1001L) + && removal.getRequestKey().startsWith("flow:"))); + verify(messages, never()).sendNotice(any()); + if (event.equals("TASK_ASSIGNED")) { + assertThat(state("current-owner")).isEqualTo("SENT"); + } + else { + assertThat(state("current-owner")).isEqualTo("REMOVING"); + transaction.executeWithoutResult(status -> service.deliverNotice("current-owner")); + assertThat(state("current-owner")).isEqualTo("CANCELLED"); + } + assertThat(state("next-task")).isEqualTo("SENT"); + assertThat(state("other-tenant")).isEqualTo("SENT"); + } + + @Test + void failedRemovalCanRetryAndAnEndedTaskIsNeverDeliveredAgain() { + // READY 可能是发送成功但响应丢失的重试；即使任务已结束也要尝试清理原消息。 + insert("queued", 1, "task-A", "READY", null); + when(messages.removeFlowNotice(any())).thenReturn(R.failed("暂时不可用")); + assertThatThrownBy(() -> transaction.executeWithoutResult(status -> service.deliverNotice("queued"))) + .isInstanceOf(IllegalStateException.class); + assertThat(state("queued")).isEqualTo("READY"); + when(messages.removeFlowNotice(any())).thenReturn(R.ok(true)); + transaction.executeWithoutResult(status -> service.deliverNotice("queued")); + assertThat(state("queued")).isEqualTo("CANCELLED"); + verify(messages, never()).sendNotice(any()); + verify(messages, times(2)).removeFlowNotice(any(FlowMessageRemovalDTO.class)); + } + + private void insert(String id, long tenant, String task, String status, String reminder) { + jdbc.update("insert into flow_notice(id,tenant_id,user_id,task_id,status,reminder_key,scheduled_at) values(?,?,?,?,?,?,CURRENT_TIMESTAMP)", + id, tenant, "1001", task, status, reminder); + } + + private String state(String id) { + return jdbc.queryForObject("select status from flow_notice where id=?", String.class, id); + } +} diff --git a/pigx-upms/pigx-upms-api/src/main/java/com/pig4cloud/pigx/admin/api/dto/FlowMessageRemovalDTO.java b/pigx-upms/pigx-upms-api/src/main/java/com/pig4cloud/pigx/admin/api/dto/FlowMessageRemovalDTO.java new file mode 100644 index 000000000..b79a8151c --- /dev/null +++ b/pigx-upms/pigx-upms-api/src/main/java/com/pig4cloud/pigx/admin/api/dto/FlowMessageRemovalDTO.java @@ -0,0 +1,29 @@ +package com.pig4cloud.pigx.admin.api.dto; + +import jakarta.validation.constraints.NotBlank; +import jakarta.validation.constraints.NotNull; +import lombok.Data; + +/** 精确撤销某次流程通知的接收记录，避免误删转办后新产生的通知。 */ +@Data +public class FlowMessageRemovalDTO { + + /** + * 流程任务 ID，对应站内信 business_id + */ + @NotBlank + private String taskId; + + /** + * 通知事件标识，格式为 flow: 加通知事件 ID，对应站内信 request_key + */ + @NotBlank + private String requestKey; + + /** + * 原接收人用户 ID + */ + @NotNull + private Long userId; + +} 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 7492dc9e2..f8ea4d61f 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 @@ -19,6 +19,7 @@ package com.pig4cloud.pigx.admin.api.feign; +import com.pig4cloud.pigx.admin.api.dto.FlowMessageRemovalDTO; import com.pig4cloud.pigx.admin.api.dto.MessageEmailDTO; import com.pig4cloud.pigx.admin.api.dto.MessageHookDTO; import com.pig4cloud.pigx.admin.api.dto.MessageNoticeDTO; @@ -76,4 +77,14 @@ public interface RemoteMessageService { @PostMapping("/sysMessage/internal/notice") R sendNotice(@RequestBody MessageNoticeDTO noticeDTO); + /** + * 逻辑删除指定流程通知在原接收人下的接收记录；仅供流程服务内部调用。 + * + * @param removal 任务 ID、通知事件标识和原接收人 + * @return 处理成功为 true，记录不存在也视为成功 + */ + @NoToken + @PostMapping("/sysMessage/internal/flow-notice/remove") + R removeFlowNotice(@RequestBody FlowMessageRemovalDTO removal); + } 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 21895b275..cafc7fadb 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 @@ -6,6 +6,7 @@ import cn.hutool.core.util.StrUtil; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import com.baomidou.mybatisplus.core.toolkit.Wrappers; import com.baomidou.mybatisplus.extension.plugins.pagination.Page; +import com.pig4cloud.pigx.admin.api.dto.FlowMessageRemovalDTO; import com.pig4cloud.pigx.admin.api.dto.MessageEmailDTO; import com.pig4cloud.pigx.admin.api.dto.MessageHookDTO; import com.pig4cloud.pigx.admin.api.dto.MessageNoticeDTO; @@ -285,4 +286,16 @@ public class SysMessageController { return sysMessageService.sendNotice(noticeDTO); } + /** + * 流程完成、取消或转办后，逻辑删除对应通知在原接收人下的接收记录；仅限内部服务调用。 + * @param removal 任务 ID、通知事件标识和原接收人 + * @return 处理成功为 true，记录不存在也视为成功 + */ + @Inner + @PostMapping("/internal/flow-notice/remove") + @Operation(summary = "内部撤销流程站内信") + public R removeFlowNotice(@Valid @RequestBody FlowMessageRemovalDTO removal) { + return R.ok(sysMessageService.removeFlowNotice(removal)); + } + } diff --git a/pigx-upms/pigx-upms-biz/src/main/java/com/pig4cloud/pigx/admin/service/SysMessageService.java b/pigx-upms/pigx-upms-biz/src/main/java/com/pig4cloud/pigx/admin/service/SysMessageService.java index 5074696d8..3dee9d5ea 100644 --- a/pigx-upms/pigx-upms-biz/src/main/java/com/pig4cloud/pigx/admin/service/SysMessageService.java +++ b/pigx-upms/pigx-upms-biz/src/main/java/com/pig4cloud/pigx/admin/service/SysMessageService.java @@ -3,6 +3,7 @@ package com.pig4cloud.pigx.admin.service; import com.baomidou.mybatisplus.core.metadata.IPage; import com.baomidou.mybatisplus.extension.plugins.pagination.Page; import com.baomidou.mybatisplus.extension.service.IService; +import com.pig4cloud.pigx.admin.api.dto.FlowMessageRemovalDTO; import com.pig4cloud.pigx.admin.api.dto.MessageEmailDTO; import com.pig4cloud.pigx.admin.api.dto.MessageHookDTO; import com.pig4cloud.pigx.admin.api.dto.MessageNoticeDTO; @@ -121,4 +122,13 @@ public interface SysMessageService extends IService { */ R sendNotice(MessageNoticeDTO noticeDTO); + /** + * 逻辑删除指定流程通知在原接收人下的接收记录，不删除消息正文和其他接收人的记录。 + * + * @param removal 任务 ID、通知事件标识（须以 flow: 开头）和原接收人 + * @return 处理成功为 true，记录不存在也视为成功 + * @throws IllegalArgumentException 参数缺失或通知事件标识不属于流程通知 + */ + Boolean removeFlowNotice(FlowMessageRemovalDTO removal); + } 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 d1dc1a6a6..dde39949c 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 @@ -654,4 +654,24 @@ public class SysMessageServiceImpl extends ServiceImpllambdaQuery() + .select(SysMessageEntity::getId) + .eq(SysMessageEntity::getBusinessType, "FLOW_TASK") + .eq(SysMessageEntity::getBusinessId, removal.getTaskId()) + .eq(SysMessageEntity::getRequestKey, removal.getRequestKey())); + for (var message : messages) { + relationMapper.delete(Wrappers.lambdaQuery() + .eq(SysMessageRelationEntity::getMsgId, message.getId()) + .eq(SysMessageRelationEntity::getUserId, removal.getUserId())); + } + return Boolean.TRUE; + } + } diff --git a/pigx-upms/pigx-upms-biz/src/test/java/com/pig4cloud/pigx/admin/service/impl/SysMessageDeletionTests.java b/pigx-upms/pigx-upms-biz/src/test/java/com/pig4cloud/pigx/admin/service/impl/SysMessageDeletionTests.java index e31a1fb19..54b5ef944 100644 --- a/pigx-upms/pigx-upms-biz/src/test/java/com/pig4cloud/pigx/admin/service/impl/SysMessageDeletionTests.java +++ b/pigx-upms/pigx-upms-biz/src/test/java/com/pig4cloud/pigx/admin/service/impl/SysMessageDeletionTests.java @@ -4,6 +4,11 @@ import com.baomidou.mybatisplus.core.MybatisConfiguration; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import com.baomidou.mybatisplus.core.metadata.TableInfoHelper; import com.pig4cloud.pigx.admin.api.entity.SysMessageRelationEntity; +import com.pig4cloud.pigx.admin.api.entity.SysMessageEntity; +import com.pig4cloud.pigx.admin.api.dto.FlowMessageRemovalDTO; +import com.pig4cloud.pigx.admin.mapper.SysMessageMapper; +import org.springframework.test.util.ReflectionTestUtils; +import java.util.List; import com.pig4cloud.pigx.admin.mapper.SysMessageRelationMapper; import com.pig4cloud.pigx.common.security.service.PigxUser; import org.springframework.security.authentication.UsernamePasswordAuthenticationToken; @@ -35,5 +40,26 @@ class SysMessageDeletionTests { assertThat(condition.getValue().getSqlSegment()).contains("msg_id =", "user_id ="); assertThat(condition.getValue().getParamNameValuePairs().values()).containsExactlyInAnyOrder(123L, 1001L); verifyNoMoreInteractions(relations); + + // 流程撤销必须同时匹配任务、事件和接收人，不能误删新分配的通知。 + TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new MybatisConfiguration(), "message"), SysMessageEntity.class); + var messages = mock(SysMessageMapper.class); + ReflectionTestUtils.setField(service, "baseMapper", messages); + var message = new SysMessageEntity(); + message.setId(456L); + when(messages.selectList(any(LambdaQueryWrapper.class))).thenReturn(List.of(message)); + var removal = new FlowMessageRemovalDTO(); + removal.setTaskId("task-A"); + removal.setRequestKey("flow:event-A"); + removal.setUserId(1002L); + service.removeFlowNotice(removal); + var messageCondition = ArgumentCaptor.forClass(LambdaQueryWrapper.class); + verify(messages).selectList(messageCondition.capture()); + assertThat(messageCondition.getValue().getSqlSegment()).contains("business_type =", "business_id =", "request_key ="); + assertThat(messageCondition.getValue().getParamNameValuePairs().values()) + .containsExactlyInAnyOrder("FLOW_TASK", "task-A", "flow:event-A"); + verify(relations, times(2)).delete(condition.capture()); + assertThat(condition.getValue().getSqlSegment()).contains("msg_id =", "user_id ="); + assertThat(condition.getValue().getParamNameValuePairs().values()).containsExactlyInAnyOrder(456L, 1002L); } }