From 085235710b76c43eec999ce28cc47d7ac2adadc5 Mon Sep 17 00:00:00 2001 From: PIGCLOUD Date: Wed, 14 Oct 2026 10:27:12 +0800 Subject: [PATCH] =?UTF-8?q?feat(flow):=20=E6=8C=81=E4=B9=85=E5=8C=96?= =?UTF-8?q?=E6=B5=81=E7=A8=8B=E8=8D=89=E7=A8=BF=E6=8F=90=E9=86=92=E4=B8=8E?= =?UTF-8?q?=E5=8F=AF=E9=9D=A0=E5=9B=9E=E8=B0=83?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit (cherry picked from commit f6871bee7c5cddaaa5650a284ac3cc3fec16dd18) --- db/0pigxx_boot.sql | 28 ++ db/3pigxx_flow.sql | 28 ++ db/migration/20261014_flow_first_batch.md | 61 ++++ db/migration/20261014_flow_first_batch.sql | 27 ++ db/migration/20261014_flow_merge_tables.sql | 100 +++++++ .../flow/constant/FlowCallbackStatusEnum.java | 23 ++ .../flow/constant/FlowNoticeStatusEnum.java | 29 ++ .../com/pig4cloud/pigx/flow/dto/Node.java | 3 + .../flow/config/FlowDurableConfiguration.java | 100 +++++++ .../controller/FlowWorkspaceController.java | 150 ++++++++++ .../flow/service/FlowCallbackDelivery.java | 107 +++++++ .../pigx/flow/service/FlowDraftService.java | 121 ++++++++ .../flow/service/FlowReliabilityService.java | 260 ++++++++++++++++++ .../flow/service/impl/ProcessServiceImpl.java | 15 + .../listeners/FlowDurableEventListener.java | 56 ++++ .../listeners/FlowProcessEventListener.java | 3 - .../listeners/NodeTaskCompleteNotify.java | 54 +--- .../pigx/flow/service/FlowWorkspaceTests.java | 253 +++++++++++++++++ 18 files changed, 1363 insertions(+), 55 deletions(-) create mode 100644 db/migration/20261014_flow_first_batch.md create mode 100644 db/migration/20261014_flow_first_batch.sql create mode 100644 db/migration/20261014_flow_merge_tables.sql create mode 100644 pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/constant/FlowCallbackStatusEnum.java create mode 100644 pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/constant/FlowNoticeStatusEnum.java create mode 100644 pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/config/FlowDurableConfiguration.java create mode 100644 pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/controller/FlowWorkspaceController.java create mode 100644 pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/FlowCallbackDelivery.java create mode 100644 pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/FlowDraftService.java create mode 100644 pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/FlowReliabilityService.java create mode 100644 pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/listeners/FlowDurableEventListener.java create mode 100644 pigx-flow/pigx-flow-biz/src/test/java/com/pig4cloud/pigx/flow/service/FlowWorkspaceTests.java diff --git a/db/0pigxx_boot.sql b/db/0pigxx_boot.sql index f97e2bcca..d2b53de90 100644 --- a/db/0pigxx_boot.sql +++ b/db/0pigxx_boot.sql @@ -3235,3 +3235,31 @@ COMMIT; -- ------------------------------------------- SET FOREIGN_KEY_CHECKS = 1; + +-- 第一批持久化能力：三张业务表。已安装旧五表版本时，建表后另执行 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/3pigxx_flow.sql b/db/3pigxx_flow.sql index bc6fecafe..6533693c7 100644 --- a/db/3pigxx_flow.sql +++ b/db/3pigxx_flow.sql @@ -288,3 +288,31 @@ CREATE TABLE `bmp_oa_leave` ( ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT='请假表单'; SET FOREIGN_KEY_CHECKS = 1; + +-- 第一批持久化能力：三张业务表。已安装旧五表版本时，建表后另执行 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_first_batch.md b/db/migration/20261014_flow_first_batch.md new file mode 100644 index 000000000..2c03ac50a --- /dev/null +++ b/db/migration/20261014_flow_first_batch.md @@ -0,0 +1,61 @@ +# 流程第一批能力升级 + +## 范围与入口 + +- 流程设计器：保存设计草稿、设计草稿箱；保存基本信息、表单和节点树，不部署流程。 +- 流程列表：申请草稿箱；动态表单发起抽屉可保存表单值和自选审批人。按用户和租户隔离，版本号防止多窗口覆盖。 +- 待办页：流程通知，包含任务分配、发起人催办和超时提醒；点击通知沿用现有任务详情权限检查。历史通知可能对应已完成或已转交任务。 +- 我发起的：运行中流程可催办，同一实例十分钟内最多一次；仅发起人可以操作。回调记录可查看状态和手动重试失败事件。 +- 审批节点 / 事件通知：配置自然时长分钟数，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 new file mode 100644 index 000000000..cd363ea47 --- /dev/null +++ b/db/migration/20261014_flow_first_batch.sql @@ -0,0 +1,27 @@ +-- 第一批持久化能力：三张业务表。已安装旧五表版本时，建表后另执行 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 new file mode 100644 index 000000000..0200d9a4f --- /dev/null +++ b/db/migration/20261014_flow_merge_tables.sql @@ -0,0 +1,100 @@ +-- 仅用于已经安装旧五表版本的库。先备份、停止流程写入及后台扫描，并执行三表建表脚本。 +-- 可重复执行；旧表暂保留作为核对来源，新代码不再访问。核验并备份后再清理旧表。 +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/FlowCallbackStatusEnum.java b/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/constant/FlowCallbackStatusEnum.java new file mode 100644 index 000000000..db029a00c --- /dev/null +++ b/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/constant/FlowCallbackStatusEnum.java @@ -0,0 +1,23 @@ +package com.pig4cloud.pigx.flow.constant; + +import lombok.Getter; +import lombok.RequiredArgsConstructor; + +/** 回调投递状态；编码与 flow_callback.status 保持兼容，不表示单次尝试的审计结果。 */ +@Getter +@RequiredArgsConstructor +public enum FlowCallbackStatusEnum { + + /** 等待首次投递或到期重试。 */ + PENDING("PENDING"), + /** 已领取租约，正在投递；过期后可重新领取。 */ + SENDING("SENDING"), + /** 接收方返回 HTTP 2xx，投递结束。 */ + DELIVERED("DELIVERED"), + /** 尝试耗尽或最终租约过期，允许发起人手动重试。 */ + FAILED("FAILED"); + + /** 持久化编码，不能随枚举重命名改变。 */ + private final String code; + +} 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 new file mode 100644 index 000000000..46c61f15e --- /dev/null +++ b/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/constant/FlowNoticeStatusEnum.java @@ -0,0 +1,29 @@ +package com.pig4cloud.pigx.flow.constant; + +import lombok.Getter; +import lombok.RequiredArgsConstructor; + +/** + * 流程通知的持久化状态；编码与 flow_notice.status 及升级 SQL 保持一致。 待触发提醒发送后转为 SENT，任务结束时转为 CANCELLED；列表仅展示 + * SENT。 + */ +@Getter +@RequiredArgsConstructor +public enum FlowNoticeStatusEnum { + + /** 等待截止时间，到期后确定当前审批人。 */ + PENDING("PENDING"), + + /** 通知已生成，允许接收人查看及标记已读。 */ + SENT("SENT"), + + /** 任务已经结束，取消尚未发送的提醒。 */ + CANCELLED("CANCELLED"), + + /** 旧监听没有截止时间，迁移后保留但不参与扫描。 */ + DISABLED("DISABLED"); + + /** 数据库状态值，保持与已发布的三表结构兼容。 */ + private final String code; + +} diff --git a/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/dto/Node.java b/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/dto/Node.java index d4f93c34d..3f5804035 100644 --- a/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/dto/Node.java +++ b/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/dto/Node.java @@ -70,6 +70,9 @@ public class Node { */ private String eventConfig; + /** 超时提醒分钟数；空或 0 表示关闭，按任务创建时间计算自然时长。 */ + private Integer reminderMinutes; + /** * 执行人分配类型 对应AssignedTypeClass中的常量（指定用户、部门主管、角色等） */ 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 new file mode 100644 index 000000000..16ebbf55b --- /dev/null +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/config/FlowDurableConfiguration.java @@ -0,0 +1,100 @@ +package com.pig4cloud.pigx.flow.config; + +import com.pig4cloud.pigx.flow.service.FlowCallbackDelivery; +import com.pig4cloud.pigx.flow.constant.FlowNoticeStatusEnum; +import com.pig4cloud.pigx.flow.service.FlowReliabilityService; +import com.pig4cloud.pigx.flow.support.listeners.FlowDurableEventListener; +import com.pig4cloud.pigx.flow.constant.FlowCallbackStatusEnum; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.flowable.spring.SpringProcessEngineConfiguration; +import org.flowable.spring.boot.EngineConfigurationConfigurer; +import org.springframework.beans.factory.ObjectProvider; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.scheduling.annotation.EnableScheduling; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; +import java.util.ArrayList; +import java.util.Objects; +import org.flowable.common.engine.api.delegate.event.FlowableEventListener; + +/** 引擎持久化监听和有限批次扫描，不依赖请求令牌。 */ +@Configuration +@EnableScheduling +@RequiredArgsConstructor +@Slf4j +public class FlowDurableConfiguration { + + private final JdbcTemplate jdbc; + + private final ObjectProvider reliability; + + private final ObjectProvider delivery; + + /** + * 注册持久化监听器，保留引擎已配置的监听器。 + * @return 引擎配置器 + */ + @Bean + public EngineConfigurationConfigurer durableEvents() { + return engine -> { + var listeners = new ArrayList(); + if (Objects.nonNull(engine.getEventListeners())) + listeners.addAll(engine.getEventListeners()); + listeners.add(new FlowDurableEventListener(reliability)); + engine.setEventListeners(listeners); + }; + } + + /** + * 使用独立的两个调度线程，慢回调不会阻塞超时提醒，也不替换应用默认调度器。 + * @return 本模块专用调度器 + */ + @Bean(name = "flowDeliveryScheduler", defaultCandidate = false) + public ThreadPoolTaskScheduler flowDeliveryScheduler() { + var scheduler = new ThreadPoolTaskScheduler(); + scheduler.setPoolSize(2); + scheduler.setThreadNamePrefix("flow-delivery-"); + return scheduler; + } + + /** 每轮最多处理 20 条回调，持久记录支持重启后继续。 */ + @Scheduled(fixedDelayString = "${flow.delivery.scan-delay:10000}", scheduler = "flowDeliveryScheduler") + public void scanCallbacks() { + // 最后一次尝试进程崩溃时，结果未知，保留审计并置失败供人工核对后重试。 + jdbc.update( + "UPDATE flow_callback SET status=?,last_error='租约过期，投递结果未知',lease_token=NULL,lease_until=NULL,updated_at=CURRENT_TIMESTAMP(3) WHERE status=? AND lease_until=max_attempts", + FlowCallbackStatusEnum.FAILED.getCode(), FlowCallbackStatusEnum.SENDING.getCode()); + var callbacks = jdbc.queryForList( + "SELECT id FROM flow_callback WHERE (status=? AND next_at<=CURRENT_TIMESTAMP(3)) OR (status=? AND lease_until> save(@RequestBody FlowDraftService.Draft draft) { + return R.ok(drafts.save(tenant(), user(), draft)); + } + + /** + * 查询个人草稿。 + * @param kind DESIGN 或 APPLICATION + * @param page 页码 + * @return 草稿摘要 + */ + @Operation(summary = "分页查询个人草稿") + @GetMapping("/drafts") + public R>> list(@RequestParam String kind, @RequestParam(defaultValue = "1") int page) { + return R.ok(drafts.list(tenant(), user(), kind, page)); + } + + /** + * 读取草稿详情。 + * @param id 草稿 ID + * @return 当前用户草稿 + * @throws IllegalStateException 草稿不存在或不属于当前用户 + */ + @Operation(summary = "读取个人草稿") + @GetMapping("/drafts/{id}") + public R> get(@PathVariable String id) { + return R.ok(drafts.get(tenant(), user(), id)); + } + + /** + * 按版本删除草稿。 + * @param id 草稿 ID + * @param revision 草稿版本 + * @return 是否成功删除；不存在、不属于当前用户或版本冲突时为 false + */ + @Operation(summary = "按版本删除个人草稿") + @SysLog("按版本删除个人草稿") + @DeleteMapping("/drafts/{id}") + public R delete(@PathVariable String id, @RequestParam long revision) { + 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 + * @return 接收催办任务数 + * @throws IllegalStateException 非发起人、实例已结束、无可催办任务或十分钟内重复催办 + */ + @Operation(summary = "催办本人发起的实例") + @SysLog("催办本人发起的实例") + @PostMapping("/instances/{id}/urge") + public R urge(@PathVariable String id) { + return R.ok(reliability.urge(tenant(), user(), id)); + } + + /** + * 查看自己发起实例的回调。 + * @param id 实例 ID + * @return 最多 100 条回调摘要，不含表单载荷与目标地址 + * @throws IllegalStateException 实例不存在或不属于当前发起人 + */ + @Operation(summary = "查询本人发起实例的回调") + @GetMapping("/instances/{id}/callbacks") + public R>> callbacks(@PathVariable String id) { + return R.ok(reliability.callbacks(tenant(), user(), id)); + } + + /** + * 重试失败回调；保留事件 ID、载荷和历史，追加最多五次投递机会。 + * @param id 回调 ID + * @return 操作结果 + * @throws IllegalStateException 回调不存在、非实例发起人或回调并非失败状态 + */ + @Operation(summary = "重试本人发起实例的失败回调") + @SysLog("重试本人发起实例的失败回调") + @PostMapping("/callbacks/{id}/retry") + public R retry(@PathVariable String id) { + reliability.retry(tenant(), user(), id); + return R.ok(); + } + +} diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/FlowCallbackDelivery.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/FlowCallbackDelivery.java new file mode 100644 index 000000000..6d38c4727 --- /dev/null +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/FlowCallbackDelivery.java @@ -0,0 +1,107 @@ +package com.pig4cloud.pigx.flow.service; + +import cn.hutool.core.util.StrUtil; +import cn.hutool.http.HttpRequest; +import cn.hutool.http.HttpResponse; +import cn.hutool.json.JSONArray; +import cn.hutool.json.JSONObject; +import cn.hutool.json.JSONUtil; +import com.pig4cloud.pigx.flow.constant.FlowCallbackStatusEnum; +import lombok.RequiredArgsConstructor; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.stereotype.Service; +import org.springframework.transaction.support.TransactionTemplate; +import java.sql.Timestamp; +import java.time.Instant; +import java.util.Objects; +import java.util.UUID; + +/** HTTP 回调至少一次投递；租约用于多实例互斥，接收方使用事件 ID 实现幂等。 */ +@Service +@RequiredArgsConstructor +public class FlowCallbackDelivery { + + private final JdbcTemplate jdbc; + + private final TransactionTemplate transactions; + + /** + * 尝试投递一条到期回调，不在网络请求期间持有数据库事务。 + * @param id 内部扫描得到的事件 ID + */ + public void deliver(String id) { + String lease = UUID.randomUUID().toString(); + var record = transactions.execute(tx -> { + int changed = jdbc.update( + "UPDATE flow_callback SET status=?,attempts=attempts+1,lease_token=?,lease_until=?,updated_at=CURRENT_TIMESTAMP(3) WHERE id=? AND ((status=? AND next_at<=CURRENT_TIMESTAMP(3)) OR (status=? AND lease_until= 200 && response.getStatus() < 300; + detail = "HTTP " + response.getStatus(); + } + catch (Exception exception) { + // 只记录异常类型，避免响应正文、URL 参数或表单数据进入日志。 + detail = exception.getClass().getSimpleName(); + } + int attempts = ((Number) record.get("attempts")).intValue(); + int maximum = ((Number) record.get("max_attempts")).intValue(); + FlowCallbackStatusEnum state = success ? FlowCallbackStatusEnum.DELIVERED + : attempts >= maximum ? FlowCallbackStatusEnum.FAILED : FlowCallbackStatusEnum.PENDING; + String result = success ? "DELIVERED" : "ERROR"; + String error = success ? null : StrUtil.maxLength(detail, 500); + String auditDetail = detail; + transactions.executeWithoutResult(tx -> { + int changed = jdbc.update( + "UPDATE flow_callback SET status=?,next_at=?,last_error=?,lease_token=NULL,lease_until=NULL,updated_at=CURRENT_TIMESTAMP(3) WHERE id=? AND lease_token=?", + state.getCode(), + Timestamp.from(Instant.now().plusSeconds(Math.min(3600, 30L * (1L << Math.min(attempts - 1, 7))))), + error, id, lease); + if (changed == 1) { + JSONArray history = history(jdbc, id); + JSONObject attempt = history.stream() + .map(value -> (JSONObject) value) + .filter(value -> lease.equals(value.getStr("id"))) + .findFirst() + .orElseThrow(() -> new IllegalStateException("回调尝试记录缺失")); + attempt.set("result", result); + attempt.set("detail", auditDetail); + jdbc.update("UPDATE flow_callback SET history=? WHERE id=?", history.toString(), id); + } + }); + } + + /** 调用方必须在同一事务中先锁定回调行；历史不截断，避免手动重试覆盖旧记录。 */ + static void appendHistory(JdbcTemplate jdbc, String callback, String entry, int number, String result, + String detail) { + JSONArray history = history(jdbc, callback); + history.add(new JSONObject().set("id", entry) + .set("attempt_no", number) + .set("result", result) + .set("detail", detail) + .set("created_at", Instant.now().toString())); + jdbc.update("UPDATE flow_callback SET history=? WHERE id=?", history.toString(), callback); + } + + private static JSONArray history(JdbcTemplate jdbc, String callback) { + String value = jdbc.queryForObject("SELECT history FROM flow_callback WHERE id=?", String.class, callback); + return StrUtil.isBlank(value) ? new JSONArray() : JSONUtil.parseArray(value); + } + +} 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 new file mode 100644 index 000000000..9f2068d36 --- /dev/null +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/FlowDraftService.java @@ -0,0 +1,121 @@ +package com.pig4cloud.pigx.flow.service; + +import cn.hutool.core.util.StrUtil; +import cn.hutool.json.JSONUtil; +import com.fasterxml.jackson.annotation.JsonProperty; +import io.swagger.v3.oas.annotations.media.Schema; +import lombok.RequiredArgsConstructor; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; +import java.nio.charset.StandardCharsets; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.UUID; + +/** 个人流程草稿存储；所有操作同时限定租户和用户，保存不部署或启动流程。 */ +@Service +@RequiredArgsConstructor +public class FlowDraftService { + + private final JdbcTemplate jdbc; + + /** + * 保存草稿，使用版本号防止多窗口覆盖。 + * @param tenant 当前租户 + * @param user 当前用户 + * @param draft 完整草稿，首次保存的 id 为空、revision 为 0 + * @return 保存后的草稿（包括新的版本号） + * @throws IllegalArgumentException 类型或 JSON 无效 + * @throws IllegalStateException 草稿不存在、无权访问或版本冲突 + */ + @Transactional + public Map save(long tenant, long user, Draft draft) { + if (Objects.isNull(draft.kind()) || !List.of("DESIGN", "APPLICATION").contains(draft.kind()) + || StrUtil.isBlank(draft.payload()) || draft.payload().getBytes(StandardCharsets.UTF_8).length > 1000000 + || !JSONUtil.isTypeJSONObject(draft.payload()) || StrUtil.isBlank(draft.title()) + || draft.title().length() > 200 || Objects.isNull(draft.flowId()) || draft.flowId().length() > 64) { + throw new IllegalArgumentException("草稿类型、标题或内容无效（内容最多 1 MB）"); + } + JSONUtil.parseObj(draft.payload()); + String id = draft.id(); + if (StrUtil.isBlank(id)) { + id = UUID.randomUUID().toString(); + jdbc.update( + "INSERT INTO flow_draft(id,tenant_id,user_id,kind,flow_id,title,payload,revision,updated_at) VALUES(?,?,?,?,?,?,?,1,CURRENT_TIMESTAMP(3))", + id, tenant, user, draft.kind(), draft.flowId(), draft.title(), draft.payload()); + } + else if (jdbc.update( + "UPDATE flow_draft SET title=?,payload=?,flow_id=?,revision=revision+1,updated_at=CURRENT_TIMESTAMP(3) WHERE id=? AND tenant_id=? AND user_id=? AND kind=? AND revision=?", + draft.title(), draft.payload(), draft.flowId(), id, tenant, user, draft.kind(), + draft.revision()) != 1) { + throw new IllegalStateException("草稿已被修改或删除，请重新打开后再保存"); + } + return get(tenant, user, id); + } + + /** + * 读取当前用户的草稿。 + * @param tenant 当前租户 + * @param user 当前用户 + * @param id 草稿 ID + * @return 草稿详情 + * @throws IllegalStateException 草稿不存在或不属于当前用户 + */ + public Map get(long tenant, long user, String id) { + var rows = jdbc.queryForList("SELECT * FROM flow_draft WHERE id=? AND tenant_id=? AND user_id=?", id, tenant, + user); + if (rows.isEmpty()) + throw new IllegalStateException("草稿不存在或无权访问"); + return rows.get(0); + } + + /** + * 分页列出个人草稿摘要，不返回表单数据。 + * @param tenant 当前租户 + * @param user 当前用户 + * @param kind DESIGN 或 APPLICATION + * @param page 从 1 开始的页码 + * @return 每页最多 50 个草稿 + */ + public List> list(long tenant, long user, String kind, int page) { + 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); + } + + /** + * 按版本删除自己的草稿，冲突时不删除。 + * @param tenant 当前租户 + * @param user 当前用户 + * @param id 草稿 ID + * @param revision 打开时的版本号 + * @return 是否成功删除 + */ + public boolean delete(long tenant, long user, String id, long revision) { + return jdbc.update("DELETE FROM flow_draft WHERE id=? AND tenant_id=? AND user_id=? AND revision=?", id, tenant, + user, revision) == 1; + } + + /** + * 草稿提交数据；tenant/user 只能由认证上下文提供。 标题和载荷仅供反序列化输入，避免操作日志序列化个人表单；响应由查询结果 Map 提供。 + * + * @param id 首次保存为空；更新时使用已有草稿 ID + * @param kind DESIGN 表示流程设计，APPLICATION 表示动态表单申请 + * @param flowId 关联流程 ID；未建立流程的设计草稿可为空串 + * @param title 非空标题，最多 200 字符 + * @param payload JSON 对象字符串，最多 1 MB UTF-8；内容结构由对应 kind 的客户端维护 + * @param revision 首次保存为 0；更新携带读取版本，冲突时拒绝覆盖 + */ + public record Draft(@Schema(description = "首次保存为空，更新使用已有草稿 ID") String id, + @Schema(description = "草稿类型", allowableValues = { + "DESIGN", "APPLICATION" }) String kind, + @Schema(description = "关联流程 ID，尚未建立流程的设计草稿可为空串") String flowId, + @Schema(description = "非空标题，最多 200 字符") @JsonProperty(access = JsonProperty.Access.WRITE_ONLY) String title, + @Schema(description = "与 kind 对应的 JSON 对象字符串，最多 1 MB UTF-8") @JsonProperty( + access = JsonProperty.Access.WRITE_ONLY) String payload, + @Schema(description = "首次保存为 0，更新为读取时的版本号") long revision){ + } + +} 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 new file mode 100644 index 000000000..e7fd59e18 --- /dev/null +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/FlowReliabilityService.java @@ -0,0 +1,260 @@ +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.common.data.tenant.TenantBroker; +import com.pig4cloud.pigx.flow.dto.Node; +import com.pig4cloud.pigx.flow.constant.NodeStatusEnum; +import com.pig4cloud.pigx.flow.constant.FlowNoticeStatusEnum; +import com.pig4cloud.pigx.flow.constant.NodeTypeEnum; +import com.pig4cloud.pigx.flow.constant.ProcessInstanceConstant; +import com.pig4cloud.pigx.flow.dto.ProcessNodeRecordAssignUserParamDto; +import com.pig4cloud.pigx.flow.support.utils.NodeUtil; +import com.pig4cloud.pigx.flow.constant.FlowCallbackStatusEnum; +import lombok.RequiredArgsConstructor; +import org.flowable.engine.TaskService; +import org.flowable.task.service.impl.persistence.entity.TaskEntity; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; +import java.sql.Timestamp; +import java.time.Instant; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.UUID; + +/** 流程通知、催办与 HTTP 回调的持久化边界。SQL 始终显式带租户，不依赖请求线程上下文。 */ +@Service +@RequiredArgsConstructor +public class FlowReliabilityService { + + private final JdbcTemplate jdbc; + + private final IProcessNodeDataService nodeService; + + private final TaskService taskService; + + /** + * 在引擎事务内记录任务事件；持久化失败必须回滚引擎操作。 + * @param task 引擎当前任务 + * @param event TASK_CREATED、TASK_ASSIGNED、TASK_COMPLETED 或 ENTITY_DELETED + */ + public void record(TaskEntity task, String event) { + if (StrUtil.isBlank(task.getTenantId()) || StrUtil.isBlank(task.getProcessDefinitionId())) + return; + long tenant = Long.parseLong(task.getTenantId()); + if ("ENTITY_DELETED".equals(event)) { + cancelReminder(tenant, task.getId()); + return; + } + Node node = TenantBroker.applyAs(tenant, + t -> nodeService.getPublishedNode(task.getProcessDefinitionId(), task.getTaskDefinitionKey()) + .getData()); + if (Objects.isNull(node)) + return; + if ("TASK_COMPLETED".equals(event)) { + cancelReminder(tenant, task.getId()); + if (StrUtil.isNotBlank(node.getEventConfig())) + enqueue(task, tenant, node.getEventConfig()); + return; + } + if (Objects.equals(node.getType(), NodeTypeEnum.ROOT.getValue()) + && !Boolean.TRUE.equals(task.getVariable(ProcessInstanceConstant.REJECT_TO_STARTER_VAR))) + return; + Timestamp due = null; + if (Objects.nonNull(node.getReminderMinutes()) && node.getReminderMinutes() > 0) { + Instant created = Objects.nonNull(task.getCreateTime()) ? task.getCreateTime().toInstant() : Instant.now(); + due = Timestamp.from(created.plusSeconds(node.getReminderMinutes() * 60L)); + } + if (Objects.nonNull(due)) { + jdbc.update( + "INSERT INTO flow_notice(id,tenant_id,user_id,event_key,kind,title,task_id,instance_id,created_at,status,scheduled_at,reminder_key) VALUES(?,?,'',?,'OVERDUE',?,?,?,CURRENT_TIMESTAMP(3),?,?,?) ON DUPLICATE KEY UPDATE id=id", + UUID.randomUUID().toString(), tenant, "OVERDUE:" + task.getId(), + StrUtil.maxLength(StrUtil.blankToDefault(task.getName(), "流程任务"), 200), task.getId(), + task.getProcessInstanceId(), FlowNoticeStatusEnum.PENDING.getCode(), due, task.getId()); + } + if (StrUtil.isNotBlank(task.getAssignee())) { + // 每次分配可追踪，TASK_CREATED 和首次 TASK_ASSIGNED 按同一接收人去重。 + notice(tenant, task.getAssignee(), "ASSIGNED:" + task.getId() + StrUtil.COLON + task.getAssignee() + + StrUtil.COLON + task.getRevision(), "TODO", task.getName(), task.getId(), + task.getProcessInstanceId()); + } + } + + private void enqueue(TaskEntity task, long tenant, String target) { + if (!target.startsWith("http://") && !target.startsWith("https://")) + throw new IllegalArgumentException("节点回调地址必须使用 HTTP(S)"); + ProcessNodeRecordAssignUserParamDto payload = new ProcessNodeRecordAssignUserParamDto(); + payload.setFlowId(NodeUtil.getFlowId(task.getProcessDefinitionId())); + payload.setProcessDefinitionId(task.getProcessDefinitionId()); + payload.setProcessInstanceId(task.getProcessInstanceId()); + payload.setNodeId(task.getTaskDefinitionKey()); + payload.setTaskId(task.getId()); + payload.setExecutionId(task.getExecutionId()); + payload.setUserId(Convert.toLong(task.getAssignee())); + payload.setNodeName(task.getName()); + payload.setTaskType("COMPLETE"); + payload.setApproveDesc(Convert.toStr(task.getVariableLocal("approveDesc"))); + payload.setData(JSONUtil.toJsonStr(task.getVariables())); + payload.setLocalData(JSONUtil.toJsonStr(task.getVariablesLocal())); + jdbc.update( + "INSERT INTO flow_callback(id,tenant_id,task_id,instance_id,target_url,payload,status,next_at,created_at,updated_at) VALUES(?,?,?,?,?,?,?,CURRENT_TIMESTAMP(3),CURRENT_TIMESTAMP(3),CURRENT_TIMESTAMP(3)) ON DUPLICATE KEY UPDATE task_id=task_id", + UUID.randomUUID().toString(), tenant, task.getId(), task.getProcessInstanceId(), target, + JSONUtil.toJsonStr(payload), FlowCallbackStatusEnum.PENDING.getCode()); + } + + 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", + UUID.randomUUID().toString(), tenant, user, event, kind, + StrUtil.maxLength(StrUtil.blankToDefault(title, "流程任务"), 200), task, instance, + FlowNoticeStatusEnum.SENT.getCode()); + } + + /** + * 发起人催办；锁定实例实现跨任务、跨进程的每十分钟限频。 + * @param tenant 当前租户 + * @param user 当前用户 + * @param instance 流程实例 ID + * @return 本次接收催办的任务数 + * @throws IllegalStateException 非发起人、已结束、无待办或催办过频 + */ + @Transactional + public int urge(long tenant, long user, String instance) { + requireStarter(tenant, user, instance, true); + var tasks = taskService.createTaskQuery() + .processInstanceId(instance) + .taskTenantId(String.valueOf(tenant)) + .active() + .list(); + if (tasks.isEmpty()) + 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("已发送催办，请十分钟后再试"); + 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(), + task.getId(), instance); + count++; + } + if (count == 0) + throw new IllegalStateException("流程当前没有已分配的待办任务"); + return count; + } + + /** + * 扫描到期任务并记录一次超时提醒；只使用任务当前接收人。 暂停或暂无接收人的任务保留待发状态，十秒后再检查，避免长期占满最早一批记录。 + * @param tenant 任务所属租户 + * @param taskId 到期任务 ID + */ + @Transactional + public void remind(long tenant, String taskId) { + var watches = jdbc.queryForList( + "SELECT id FROM flow_notice WHERE reminder_key=? AND tenant_id=? AND status=? AND scheduled_at<=CURRENT_TIMESTAMP(3) FOR UPDATE", + taskId, tenant, FlowNoticeStatusEnum.PENDING.getCode()); + if (watches.isEmpty()) + return; + var task = taskService.createTaskQuery().taskId(taskId).taskTenantId(String.valueOf(tenant)).singleResult(); + if (Objects.isNull(task)) { + cancelReminder(tenant, taskId); + return; + } + if (task.isSuspended() || StrUtil.isBlank(task.getAssignee())) { + jdbc.update( + "UPDATE flow_notice SET scheduled_at=TIMESTAMPADD(SECOND,10,CURRENT_TIMESTAMP(3)) WHERE id=? AND tenant_id=? AND status=?", + watches.get(0).get("id"), tenant, FlowNoticeStatusEnum.PENDING.getCode()); + 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=?", + task.getAssignee(), StrUtil.maxLength(StrUtil.blankToDefault(task.getName(), "流程任务"), 200), + FlowNoticeStatusEnum.SENT.getCode(), watches.get(0).get("id"), tenant, + FlowNoticeStatusEnum.PENDING.getCode()); + } + + /** + * 分页读取个人通知。 + * @param tenant 当前租户 + * @param user 当前用户 + * @param page 从 1 开始的页码 + * @return 最近通知，每页最多 50 条 + */ + 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()); + } + + /** + * 查询自己发起流程的回调摘要；不向页面返回表单载荷和目标地址。 + * @param tenant 当前租户 + * @param user 当前用户 + * @param instance 实例 ID + * @return 最近 100 条回调及其状态 + */ + public List> callbacks(long tenant, long user, String instance) { + requireStarter(tenant, user, instance, false); + return jdbc.queryForList( + "SELECT id,task_id,status,attempts,last_error,created_at,updated_at FROM flow_callback WHERE tenant_id=? AND instance_id=? ORDER BY created_at DESC LIMIT 100", + tenant, instance); + } + + /** + * 将失败回调重新排队，事件 ID 和载荷保持不变。 + * @param tenant 当前租户 + * @param user 当前用户 + * @param id 回调 ID + * @throws IllegalStateException 无权限或记录并非失败状态 + */ + @Transactional + public void retry(long tenant, long user, String id) { + var rows = jdbc.queryForList("SELECT instance_id FROM flow_callback WHERE id=? AND tenant_id=?", id, tenant); + if (rows.isEmpty()) + throw new IllegalStateException("回调不存在"); + requireStarter(tenant, user, String.valueOf(rows.get(0).get("instance_id")), false); + if (jdbc.update( + "UPDATE flow_callback SET status=?,max_attempts=attempts+5,next_at=CURRENT_TIMESTAMP(3),updated_at=CURRENT_TIMESTAMP(3),lease_token=NULL,lease_until=NULL WHERE id=? AND tenant_id=? AND status=?", + FlowCallbackStatusEnum.PENDING.getCode(), id, tenant, FlowCallbackStatusEnum.FAILED.getCode()) != 1) + throw new IllegalStateException("只能重试失败的回调"); + FlowCallbackDelivery.appendHistory(jdbc, id, UUID.randomUUID().toString(), 0, "REQUEUED", + "用户 " + user + " 手动重试"); + } + + private void cancelReminder(long tenant, String task) { + jdbc.update("UPDATE flow_notice SET status=? WHERE tenant_id=? AND reminder_key=? AND status=?", + FlowNoticeStatusEnum.CANCELLED.getCode(), tenant, task, FlowNoticeStatusEnum.PENDING.getCode()); + } + + private void 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" + + (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("流程已结束，不能催办"); + } + +} diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/ProcessServiceImpl.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/ProcessServiceImpl.java index a3d47dcf8..904306eba 100644 --- a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/ProcessServiceImpl.java +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/ProcessServiceImpl.java @@ -195,6 +195,7 @@ public class ProcessServiceImpl extends ServiceImpl impl } Node publishedRoot = objectMapper.readValue(process.getProcess(), Node.class); + validateDeliveryConfig(publishedRoot); NodeUtil.snapshotFormPermissions(publishedRoot, process.getFormItems()); BpmnModel bpmnModel = ModelUtil.buildBpmnModel(publishedRoot, process.getName(), flowId); { @@ -330,4 +331,18 @@ public class ProcessServiceImpl extends ServiceImpl impl return R.ok(flowStatusEventPublisher.isFlowIdRegistered(flowId)); } + private void validateDeliveryConfig(Node node) { + if (java.util.Objects.isNull(node)) + return; + if (java.util.Objects.nonNull(node.getReminderMinutes()) + && (node.getReminderMinutes() < 0 || node.getReminderMinutes() > 525600)) + throw new IllegalArgumentException("超时提醒必须为 0 到 525600 分钟"); + if (StrUtil.isNotBlank(node.getEventConfig()) && !node.getEventConfig().startsWith("http://") + && !node.getEventConfig().startsWith("https://")) + throw new IllegalArgumentException("节点回调地址必须使用 HTTP(S)"); + if (java.util.Objects.nonNull(node.getConditionNodes())) + node.getConditionNodes().forEach(this::validateDeliveryConfig); + validateDeliveryConfig(node.getChildren()); + } + } diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/listeners/FlowDurableEventListener.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/listeners/FlowDurableEventListener.java new file mode 100644 index 000000000..a65503b4f --- /dev/null +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/listeners/FlowDurableEventListener.java @@ -0,0 +1,56 @@ +package com.pig4cloud.pigx.flow.support.listeners; + +import com.pig4cloud.pigx.flow.service.FlowReliabilityService; +import lombok.RequiredArgsConstructor; +import org.flowable.common.engine.api.delegate.event.FlowableEntityEvent; +import org.flowable.common.engine.api.delegate.event.FlowableEvent; +import org.flowable.common.engine.api.delegate.event.FlowableEventListener; +import org.flowable.task.service.impl.persistence.entity.TaskEntity; +import org.springframework.beans.factory.ObjectProvider; +import java.util.Set; + +/** 与引擎共用事务的持久化监听器，独立于允许失败的旧业务事件监听器。 */ +@RequiredArgsConstructor +public class FlowDurableEventListener implements FlowableEventListener { + + private static final Set EVENTS = Set.of("TASK_CREATED", "TASK_ASSIGNED", "TASK_COMPLETED", + "ENTITY_DELETED"); + + private final ObjectProvider service; + + /** + * 记录目标任务事件。 + * @param event 引擎事件 + */ + @Override + public void onEvent(FlowableEvent event) { + if (EVENTS.contains(event.getType().toString()) && event instanceof FlowableEntityEvent entity + && entity.getEntity() instanceof TaskEntity task) + service.getObject().record(task, event.getType().toString()); + } + + /** + * @return true，写入失败时回滚任务操作 + */ + @Override + public boolean isFailOnException() { + return true; + } + + /** + * @return false，在引擎事务中立即写入 + */ + @Override + public boolean isFireOnTransactionLifecycleEvent() { + return false; + } + + /** + * @return null，不使用提交后回调 + */ + @Override + public String getOnTransaction() { + return null; + } + +} diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/listeners/FlowProcessEventListener.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/listeners/FlowProcessEventListener.java index 1525b2c2a..179e403d6 100644 --- a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/listeners/FlowProcessEventListener.java +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/listeners/FlowProcessEventListener.java @@ -233,9 +233,6 @@ public class FlowProcessEventListener implements FlowableEventListener { remoteService.taskEndEvent(processNodeRecordAssignUserParamDto); - // 发送任务完成通知 - SpringUtil.getBean(NodeTaskCompleteNotify.class) - .sendNotify(SecurityUtils.getToken(), processNodeRecordAssignUserParamDto); } // 任务分配事件处理 diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/listeners/NodeTaskCompleteNotify.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/listeners/NodeTaskCompleteNotify.java index 7c80cbeed..baa281274 100644 --- a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/listeners/NodeTaskCompleteNotify.java +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/listeners/NodeTaskCompleteNotify.java @@ -1,78 +1,28 @@ package com.pig4cloud.pigx.flow.support.listeners; -import cn.hutool.core.lang.Validator; -import cn.hutool.http.HttpRequest; -import cn.hutool.http.HttpResponse; -import cn.hutool.json.JSONUtil; import com.baomidou.mybatisplus.core.toolkit.Wrappers; -import com.pig4cloud.pigx.common.core.util.RetOps; import com.pig4cloud.pigx.common.security.util.NonWebTokenContextHolder; import com.pig4cloud.pigx.flow.dto.ProcessInstanceParamDto; -import com.pig4cloud.pigx.flow.dto.ProcessNodeRecordAssignUserParamDto; import com.pig4cloud.pigx.flow.entity.ProcessInstanceRecord; import com.pig4cloud.pigx.flow.event.FlowStatusEventPublisher; import com.pig4cloud.pigx.flow.mapper.ProcessInstanceRecordMapper; -import com.pig4cloud.pigx.flow.service.IProcessNodeDataService; import lombok.RequiredArgsConstructor; -import lombok.extern.slf4j.Slf4j; -import org.springframework.scheduling.annotation.Async; import org.springframework.stereotype.Service; /** - * 节点任务完成通知服务 - *

- * 该服务负责在流程节点任务完成时发送异步通知。当节点配置了外部事件回调URL时， 系统会自动调用该URL通知任务完成事件。这种机制允许流程与外部系统进行集成， - * 实现业务流程的自动化。通知采用异步方式发送，不会阻塞主流程的执行。 - *

+ * 流程实例状态通知服务。HTTP 节点回调由 FlowDurableEventListener 持久化， FlowCallbackDelivery 在提交后投递；此类保留现有 + * Redis 状态发布行为。 * * @author lengleng - * @date 2026-02-10 */ -@Slf4j @Service @RequiredArgsConstructor public class NodeTaskCompleteNotify { - private final IProcessNodeDataService processNodeDataService; - private final ProcessInstanceRecordMapper processInstanceRecordMapper; private final FlowStatusEventPublisher flowStatusEventPublisher; - /** - * 异步发送任务完成通知 - *

- * 当流程节点任务完成时，如果节点配置了事件通知URL，则通过HTTP POST请求 将任务完成信息发送到指定的URL。通知内容包括任务ID、处理人、处理结果等 - * 完整信息。该方法使用异步执行，确保不影响主流程的性能。通知失败时会 记录日志但不会影响流程执行。 - *

- * @param token 安全令牌，用于在异步线程中传递认证信息 - * @param taskCompleteParamDto 任务完成参数，包含： - flowId: 流程定义ID - nodeId: 节点ID - taskId: - * 任务ID - userId: 处理人ID - result: 处理结果 - comment: 处理意见 - completeTime: 完成时间 - - * variables: 流程变量 - */ - @Async - public void sendNotify(String token, ProcessNodeRecordAssignUserParamDto taskCompleteParamDto) { - // 在异步线程中设置认证令牌 - NonWebTokenContextHolder.setToken(token); - - // 获取节点配置并发送通知 - RetOps - .of(processNodeDataService.getPublishedNode(taskCompleteParamDto.getProcessDefinitionId(), - taskCompleteParamDto.getNodeId())) - .getData() - .ifPresent(node -> { - // 验证事件配置是否为有效的URL - if (Validator.isUrl(node.getEventConfig())) { - // 发送HTTP POST请求到配置的URL - HttpResponse response = HttpRequest.post(node.getEventConfig()) - .body(JSONUtil.toJsonStr(taskCompleteParamDto)) - .timeout(10000) // 设置10秒超时 - .execute(); - log.debug("节点任务完成通知处理完成:{}，数据：{}", response.getStatus(), taskCompleteParamDto); - } - }); - } - /** * 异步发送流程实例状态变更通知 *

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 new file mode 100644 index 000000000..188158231 --- /dev/null +++ b/pigx-flow/pigx-flow-biz/src/test/java/com/pig4cloud/pigx/flow/service/FlowWorkspaceTests.java @@ -0,0 +1,253 @@ +package com.pig4cloud.pigx.flow.service; + +import com.pig4cloud.pigx.common.core.util.R; +import cn.hutool.json.JSONUtil; +import com.pig4cloud.pigx.flow.dto.Node; +import com.pig4cloud.pigx.flow.constant.FlowNoticeStatusEnum; +import com.pig4cloud.pigx.flow.config.FlowDurableConfiguration; +import com.pig4cloud.pigx.flow.support.listeners.FlowDurableEventListener; +import com.sun.net.httpserver.HttpServer; +import org.flowable.engine.ProcessEngine; +import org.flowable.engine.ProcessEngineConfiguration; +import org.flowable.spring.SpringProcessEngineConfiguration; +import org.h2.jdbcx.JdbcDataSource; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.springframework.aop.framework.ProxyFactory; +import org.springframework.beans.factory.ObjectProvider; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.datasource.DataSourceTransactionManager; +import org.springframework.transaction.annotation.AnnotationTransactionAttributeSource; +import org.springframework.transaction.interceptor.TransactionInterceptor; +import org.springframework.transaction.support.TransactionTemplate; +import java.net.InetSocketAddress; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; +import java.util.UUID; +import java.util.concurrent.atomic.AtomicInteger; +import static org.assertj.core.api.Assertions.*; +import static org.mockito.Mockito.*; + +class FlowWorkspaceTests { + + private ProcessEngine engine; + + private JdbcTemplate jdbc; + + private TransactionTemplate transaction; + + private FlowReliabilityService reliability; + + private FlowDraftService drafts; + + private HttpServer server; + + private final List eventIds = new ArrayList<>(); + + private final AtomicInteger requests = new AtomicInteger(); + + @BeforeEach + void setUp() throws Exception { + JdbcDataSource source = new JdbcDataSource(); + source.setURL("jdbc:h2:mem:" + UUID.randomUUID() + ";DB_CLOSE_DELAY=-1;DATABASE_TO_LOWER=TRUE"); + source.setUser("sa"); + jdbc = new JdbcTemplate(source); + jdbc.execute("SET MODE MySQL"); + String schema = Files.readString(Path.of("../../db/migration/20261014_flow_first_batch.sql")); + for (String sql : schema.split(";")) + if (!sql.isBlank()) + jdbc.execute(sql); + jdbc.execute( + "CREATE TABLE process_instance_record(id bigint primary key,process_instance_id varchar(64),tenant_id bigint,user_id bigint,status int,del_flag char(1))"); + var manager = new DataSourceTransactionManager(source); + transaction = new TransactionTemplate(manager); + var configuration = new SpringProcessEngineConfiguration(); + configuration.setDataSource(source); + configuration.setTransactionManager(manager); + configuration.setDatabaseSchemaUpdate(ProcessEngineConfiguration.DB_SCHEMA_UPDATE_TRUE); + configuration.setDisableIdmEngine(true); + configuration.setDisableEventRegistry(true); + configuration.setAsyncExecutorActivate(false); + ObjectProvider provider = mock(ObjectProvider.class); + when(provider.getObject()).thenAnswer(invocation -> reliability); + configuration.setEventListeners(List.of(new FlowDurableEventListener(provider))); + jdbc.execute("SET MODE LEGACY"); + engine = configuration.buildProcessEngine(); + jdbc.execute("SET MODE MySQL"); + server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext("/callback", exchange -> { + synchronized (eventIds) { + eventIds.add(exchange.getRequestHeaders().getFirst("X-Flow-Event-Id")); + } + exchange.getRequestBody().readAllBytes(); + exchange.sendResponseHeaders(requests.incrementAndGet() == 1 ? 503 : 204, -1); + exchange.close(); + }); + server.start(); + Node node = new Node(); + node.setReminderMinutes(1); + node.setEventConfig("http://127.0.0.1:" + server.getAddress().getPort() + "/callback"); + var nodes = mock(IProcessNodeDataService.class); + when(nodes.getPublishedNode(anyString(), anyString())).thenReturn(R.ok(node)); + var target = new FlowReliabilityService(jdbc, nodes, engine.getTaskService()); + var proxy = new ProxyFactory(target); + proxy.addAdvice(new TransactionInterceptor(manager, new AnnotationTransactionAttributeSource())); + reliability = (FlowReliabilityService) proxy.getProxy(); + drafts = new FlowDraftService(jdbc); + engine.getRepositoryService() + .createDeployment() + .tenantId("1") + .addString("test.bpmn20.xml", + """ + + + + + + + + """) + .deploy(); + } + + @AfterEach + void tearDown() { + if (server != null) + server.stop(0); + if (engine != null) + engine.close(); + if (jdbc != null) + jdbc.execute("SHUTDOWN"); + } + + @Test + void draftsAndTaskEventsSurviveNewServiceInstancesAndHttpRetries() throws Exception { + // 请求仍能读取个人表单，但操作日志使用的 Jackson 不得输出标题和载荷。 + var logMapper = com.pig4cloud.pigx.common.log.util.JacksonSensitiveFieldUtil.getObjectMapper().copy(); + logMapper.setFilterProvider(new com.fasterxml.jackson.databind.ser.impl.SimpleFilterProvider() + .setDefaultFilter(com.fasterxml.jackson.databind.ser.impl.SimpleBeanPropertyFilter.serializeAll())); + var input = logMapper.readValue( + "{\"kind\":\"APPLICATION\",\"flowId\":\"approval\",\"title\":\"private-title\",\"payload\":\"{\\\"secret\\\":1}\",\"revision\":0}", + FlowDraftService.Draft.class); + assertThat(input.title()).isEqualTo("private-title"); + assertThat(input.payload()).contains("secret"); + assertThat(logMapper.writeValueAsString(new Object[] { input })).doesNotContain("private-title", "secret", + "payload", "title"); + assertThat(FlowNoticeStatusEnum.values()).extracting(FlowNoticeStatusEnum::getCode) + .containsExactly("PENDING", "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.get(1, 7, application.get("id").toString()).get("payload")) + .isEqualTo("{\"formData\":{\"amount\":8}}"); + var instance = engine.getRuntimeService() + .startProcessInstanceByKeyAndTenantId("approval", java.util.Map.of(), "1"); + jdbc.update("INSERT INTO process_instance_record VALUES(1,?,1,7,1,'0')", instance.getId()); + var task = engine.getTaskService().createTaskQuery().singleResult(); + 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(); + assertThat(jdbc.queryForObject("SELECT count(*) FROM flow_notice WHERE status='PENDING'", Integer.class)) + .isEqualTo(1); + engine.getRuntimeService().suspendProcessInstanceById(instance.getId()); + jdbc.update( + "UPDATE flow_notice SET scheduled_at=DATEADD('MINUTE',-1,CURRENT_TIMESTAMP) WHERE status='PENDING'"); + reliability.remind(1, task.getId()); + assertThat( + jdbc.queryForObject("SELECT status FROM flow_notice WHERE reminder_key=?", String.class, task.getId())) + .isEqualTo("PENDING"); + assertThat(jdbc.queryForObject("SELECT scheduled_at>CURRENT_TIMESTAMP(3) FROM flow_notice WHERE reminder_key=?", + Boolean.class, task.getId())) + .isTrue(); + engine.getRuntimeService().activateProcessInstanceById(instance.getId()); + engine.getTaskService().setAssignee(task.getId(), "2002"); + jdbc.update( + "UPDATE flow_notice SET scheduled_at=DATEADD('MINUTE',-1,CURRENT_TIMESTAMP) WHERE status='PENDING'"); + ObjectProvider provider = mock(ObjectProvider.class); + when(provider.getObject()).thenReturn(reliability); + var scheduler = new FlowDurableConfiguration(jdbc, provider, mock(ObjectProvider.class)); + scheduler.scanReminders(); + scheduler.scanReminders(); + assertThat(reliability.notices(1, 2002, 1)).hasSize(2); + engine.getTaskService().setAssignee(task.getId(), "1001"); + assertThat(reliability.notices(1, 1001, 1)).hasSize(2); + engine.getTaskService().setAssignee(task.getId(), "2002"); + assertThat(reliability.urge(1, 7, instance.getId())).isEqualTo(1); + assertThatThrownBy(() -> reliability.urge(1, 7, instance.getId())).hasMessageContaining("十分钟"); + engine.getTaskService().complete(task.getId()); + assertThat(jdbc.queryForObject("SELECT count(*) FROM flow_notice WHERE status='PENDING'", Integer.class)) + .isZero(); + var callback = reliability.callbacks(1, 7, instance.getId()).get(0); + String id = callback.get("id").toString(); + new FlowCallbackDelivery(jdbc, transaction).deliver(id); + assertThat(jdbc.queryForObject("SELECT status FROM flow_callback", String.class)).isEqualTo("PENDING"); + jdbc.update("UPDATE flow_callback SET next_at=CURRENT_TIMESTAMP"); + new FlowCallbackDelivery(jdbc, transaction).deliver(id); + assertThat(jdbc.queryForObject("SELECT status FROM flow_callback", String.class)).isEqualTo("DELIVERED"); + var history = JSONUtil.parseArray(jdbc.queryForObject("SELECT history FROM flow_callback", String.class)); + assertThat(history).hasSize(2); + assertThat(history.getJSONObject(0).getStr("result")).isEqualTo("ERROR"); + assertThat(history.getJSONObject(1).getStr("result")).isEqualTo("DELIVERED"); + jdbc.update("UPDATE flow_callback SET status='FAILED'"); + reliability.retry(1, 7, id); + history = JSONUtil.parseArray(jdbc.queryForObject("SELECT history FROM flow_callback", String.class)); + assertThat(history).hasSize(3); + assertThat(history.getJSONObject(2).getStr("result")).isEqualTo("REQUEUED"); + new FlowCallbackDelivery(jdbc, transaction).deliver(id); + history = JSONUtil.parseArray(jdbc.queryForObject("SELECT history FROM flow_callback", String.class)); + assertThat(history).hasSize(4); + assertThat(history.getJSONObject(0).getStr("result")).isEqualTo("ERROR"); + assertThat(history.getJSONObject(3).getStr("result")).isEqualTo("DELIVERED"); + assertThat(eventIds).containsExactly(id, id, id); + engine.getRuntimeService().startProcessInstanceByKeyAndTenantId("approval", java.util.Map.of(), "1"); + String completed = engine.getTaskService().createTaskQuery().singleResult().getId(); + engine.getTaskService().complete(completed); + jdbc.update("UPDATE flow_notice SET scheduled_at=DATEADD('MINUTE',-1,CURRENT_TIMESTAMP) WHERE reminder_key=?", + completed); + reliability.remind(1, completed); + assertThat(jdbc.queryForObject("SELECT status FROM flow_notice WHERE reminder_key=?", String.class, completed)) + .isEqualTo("CANCELLED"); + assertThat(jdbc.queryForObject("SELECT count(*) FROM flow_notice WHERE reminder_key=? AND status='SENT'", + Integer.class, completed)) + .isZero(); + } + + @Test + void ownershipConflictsAndStorageFailuresCannotLoseCommittedWork() { + var saved = drafts.save(1, 7, new FlowDraftService.Draft(null, "APPLICATION", "approval", "申请", "{}", 0)); + String id = saved.get("id").toString(); + assertThatThrownBy(() -> drafts.get(2, 7, id)).hasMessageContaining("无权"); + assertThatThrownBy(() -> drafts.get(1, 8, id)).hasMessageContaining("无权"); + assertThatThrownBy( + () -> drafts.save(1, 7, new FlowDraftService.Draft(id, "APPLICATION", "approval", "申请", "{}", 0))) + .hasMessageContaining("重新打开"); + assertThat(drafts.delete(1, 8, id, 1)).isFalse(); + assertThatThrownBy(() -> transaction.executeWithoutResult(tx -> { + engine.getRuntimeService().startProcessInstanceByKeyAndTenantId("approval", java.util.Map.of(), "1"); + throw new IllegalStateException("rollback"); + })).hasMessage("rollback"); + assertThat(jdbc.queryForObject("SELECT count(*) FROM flow_notice", Integer.class)).isZero(); + assertThat(engine.getTaskService().createTaskQuery().count()).isZero(); + var instance = engine.getRuntimeService() + .startProcessInstanceByKeyAndTenantId("approval", java.util.Map.of(), "1"); + jdbc.update("INSERT INTO process_instance_record VALUES(1,?,1,7,1,'0')", instance.getId()); + assertThatThrownBy(() -> reliability.urge(2, 7, instance.getId())).hasMessageContaining("无权"); + assertThatThrownBy(() -> reliability.callbacks(1, 8, instance.getId())).hasMessageContaining("无权"); + String task = engine.getTaskService().createTaskQuery().singleResult().getId(); + jdbc.execute("DROP TABLE flow_callback"); + assertThatThrownBy(() -> engine.getTaskService().complete(task)).isInstanceOf(RuntimeException.class); + assertThat(engine.getTaskService().createTaskQuery().taskId(task).count()).isEqualTo(1); + assertThat(jdbc.queryForObject("SELECT count(*) FROM flow_notice WHERE status='PENDING'", Integer.class)) + .isEqualTo(1); + } + +}