feat(flow): 持久化流程草稿提醒与可靠回调

(cherry picked from commit f6871bee7c5cddaaa5650a284ac3cc3fec16dd18)

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;

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;

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 的临时数据库使用带数据样本验证，并检查重复执行和孤立审计回滚；临时库测试后清理。真实账号完整联调尚未执行，不把接口隔离回归当成真实后端验收。

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;

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 ;

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;
+
+}

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;
+
+}

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中的常量（指定用户、部门主管、角色等）
 	 */

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<FlowReliabilityService> reliability;
+
+	private final ObjectProvider<FlowCallbackDelivery> delivery;
+
+	/**
+	 * 注册持久化监听器，保留引擎已配置的监听器。
+	 * @return 引擎配置器
+	 */
+	@Bean
+	public EngineConfigurationConfigurer<SpringProcessEngineConfiguration> durableEvents() {
+		return engine -> {
+			var listeners = new ArrayList<FlowableEventListener>();
+			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<CURRENT_TIMESTAMP(3) AND attempts>=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<CURRENT_TIMESTAMP(3)) ORDER BY next_at LIMIT 20",
+				FlowCallbackStatusEnum.PENDING.getCode(), FlowCallbackStatusEnum.SENDING.getCode());
+		for (var row : callbacks) {
+			try {
+				delivery.getObject().deliver(String.valueOf(row.get("id")));
+			}
+			catch (Exception exception) {
+				log.warn("流程回调投递暂未完成，事件 {}", row.get("id"), exception);
+			}
+		}
+	}
+
+	/** 独立扫描最多 100 条超时任务。 */
+	@Scheduled(fixedDelayString = "${flow.delivery.scan-delay:10000}", scheduler = "flowDeliveryScheduler")
+	public void scanReminders() {
+		var tasks = jdbc.queryForList(
+				"SELECT task_id,tenant_id FROM flow_notice WHERE status=? AND scheduled_at<=CURRENT_TIMESTAMP(3) ORDER BY scheduled_at LIMIT 100",
+				FlowNoticeStatusEnum.PENDING.getCode());
+		for (var row : tasks) {
+			try {
+				reliability.getObject()
+					.remind(((Number) row.get("tenant_id")).longValue(), String.valueOf(row.get("task_id")));
+			}
+			catch (Exception exception) {
+				log.warn("流程超时提醒暂未完成，任务 {}", row.get("task_id"), exception);
+			}
+		}
+	}
+
+}

pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/controller/FlowWorkspaceController.java

@@ -0,0 +1,150 @@
+package com.pig4cloud.pigx.flow.controller;
+
+import com.pig4cloud.pigx.common.core.util.R;
+import com.pig4cloud.pigx.common.data.tenant.TenantContextHolder;
+import com.pig4cloud.pigx.common.security.util.SecurityUtils;
+import com.pig4cloud.pigx.flow.service.FlowDraftService;
+import com.pig4cloud.pigx.flow.service.FlowReliabilityService;
+import com.pig4cloud.pigx.common.log.annotation.SysLog;
+import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import lombok.RequiredArgsConstructor;
+import org.springframework.web.bind.annotation.*;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+/** 当前登录用户的流程工作区；不接受客户端指定租户或用户身份。 */
+@Tag(name = "流程工作区", description = "当前用户草稿和通知，以及本人发起实例的催办与回调")
+@RestController
+@RequestMapping("/workspace")
+@RequiredArgsConstructor
+public class FlowWorkspaceController {
+
+	private final FlowDraftService drafts;
+
+	private final FlowReliabilityService reliability;
+
+	private long tenant() {
+		return Objects.requireNonNull(TenantContextHolder.getTenantId(), "缺少租户");
+	}
+
+	private long user() {
+		return Objects.requireNonNull(SecurityUtils.getUser(), "请登录").getId();
+	}
+
+	/**
+	 * 保存个人草稿。
+	 * @param draft 草稿内容与版本号
+	 * @return 新版本草稿
+	 * @throws IllegalArgumentException 草稿类型、标题或 JSON 内容无效
+	 * @throws IllegalStateException 草稿不属于当前用户、已删除或版本冲突
+	 */
+	@Operation(summary = "保存个人草稿")
+	@SysLog("保存个人草稿")
+	@PostMapping("/drafts")
+	public R<Map<String, Object>> 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<Map<String, Object>>> 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<Map<String, Object>> 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<Boolean> 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<List<Map<String, Object>>> 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<Void> 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<Integer> 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<List<Map<String, Object>>> 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<Void> retry(@PathVariable String id) {
+		reliability.retry(tenant(), user(), id);
+		return R.ok();
+	}
+
+}

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<CURRENT_TIMESTAMP(3))) AND attempts<max_attempts",
+					FlowCallbackStatusEnum.SENDING.getCode(), lease, Timestamp.from(Instant.now().plusSeconds(60)), id,
+					FlowCallbackStatusEnum.PENDING.getCode(), FlowCallbackStatusEnum.SENDING.getCode());
+			if (changed != 1)
+				return null;
+			var row = jdbc.queryForMap("SELECT * FROM flow_callback WHERE id=? AND lease_token=?", id, lease);
+			appendHistory(jdbc, id, lease, ((Number) row.get("attempts")).intValue(), "STARTED", null);
+			return row;
+		});
+		if (Objects.isNull(record))
+			return;
+		boolean success = false;
+		String detail;
+		try (HttpResponse response = HttpRequest.post(String.valueOf(record.get("target_url")))
+			.header("X-Flow-Event-Id", id)
+			.header("Idempotency-Key", id)
+			.body(String.valueOf(record.get("payload")))
+			.timeout(10000)
+			.setFollowRedirects(false)
+			.execute()) {
+			success = response.getStatus() >= 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);
+	}
+
+}

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<String, Object> 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<String, Object> 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<Map<String, Object>> 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){
+	}
+
+}

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<Map<String, Object>> 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<Map<String, Object>> 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("流程已结束，不能催办");
+	}
+
+}

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<ProcessMapper, Process> 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<ProcessMapper, Process> 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());
+	}
+
 }

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<String> EVENTS = Set.of("TASK_CREATED", "TASK_ASSIGNED", "TASK_COMPLETED",
+			"ENTITY_DELETED");
+
+	private final ObjectProvider<FlowReliabilityService> 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;
+	}
+
+}

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);
 		}
 
 		// 任务分配事件处理

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;
 
 /**
- * 节点任务完成通知服务
- * <p>
- * 该服务负责在流程节点任务完成时发送异步通知。当节点配置了外部事件回调URL时， 系统会自动调用该URL通知任务完成事件。这种机制允许流程与外部系统进行集成，
- * 实现业务流程的自动化。通知采用异步方式发送，不会阻塞主流程的执行。
- * </p>
+ * 流程实例状态通知服务。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;
 
-	/**
-	 * 异步发送任务完成通知
-	 * <p>
-	 * 当流程节点任务完成时，如果节点配置了事件通知URL，则通过HTTP POST请求 将任务完成信息发送到指定的URL。通知内容包括任务ID、处理人、处理结果等
-	 * 完整信息。该方法使用异步执行，确保不影响主流程的性能。通知失败时会 记录日志但不会影响流程执行。
-	 * </p>
-	 * @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);
-				}
-			});
-	}
-
 	/**
 	 * 异步发送流程实例状态变更通知
 	 * <p>

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<String> eventIds = new ArrayList<>();
+
+	private final AtomicInteger requests = new AtomicInteger();
+
+	@BeforeEach
+	void setUp() throws Exception {
+		JdbcDataSource source = new JdbcDataSource();
+		source.setURL("jdbc:h2:mem:" + UUID.randomUUID() + ";DB_CLOSE_DELAY=-1;DATABASE_TO_LOWER=TRUE");
+		source.setUser("sa");
+		jdbc = new JdbcTemplate(source);
+		jdbc.execute("SET MODE MySQL");
+		String schema = Files.readString(Path.of("../../db/migration/20261014_flow_first_batch.sql"));
+		for (String sql : schema.split(";"))
+			if (!sql.isBlank())
+				jdbc.execute(sql);
+		jdbc.execute(
+				"CREATE TABLE process_instance_record(id bigint primary key,process_instance_id varchar(64),tenant_id bigint,user_id bigint,status int,del_flag char(1))");
+		var manager = new DataSourceTransactionManager(source);
+		transaction = new TransactionTemplate(manager);
+		var configuration = new SpringProcessEngineConfiguration();
+		configuration.setDataSource(source);
+		configuration.setTransactionManager(manager);
+		configuration.setDatabaseSchemaUpdate(ProcessEngineConfiguration.DB_SCHEMA_UPDATE_TRUE);
+		configuration.setDisableIdmEngine(true);
+		configuration.setDisableEventRegistry(true);
+		configuration.setAsyncExecutorActivate(false);
+		ObjectProvider<FlowReliabilityService> provider = mock(ObjectProvider.class);
+		when(provider.getObject()).thenAnswer(invocation -> reliability);
+		configuration.setEventListeners(List.of(new FlowDurableEventListener(provider)));
+		jdbc.execute("SET MODE LEGACY");
+		engine = configuration.buildProcessEngine();
+		jdbc.execute("SET MODE MySQL");
+		server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+		server.createContext("/callback", exchange -> {
+			synchronized (eventIds) {
+				eventIds.add(exchange.getRequestHeaders().getFirst("X-Flow-Event-Id"));
+			}
+			exchange.getRequestBody().readAllBytes();
+			exchange.sendResponseHeaders(requests.incrementAndGet() == 1 ? 503 : 204, -1);
+			exchange.close();
+		});
+		server.start();
+		Node node = new Node();
+		node.setReminderMinutes(1);
+		node.setEventConfig("http://127.0.0.1:" + server.getAddress().getPort() + "/callback");
+		var nodes = mock(IProcessNodeDataService.class);
+		when(nodes.getPublishedNode(anyString(), anyString())).thenReturn(R.ok(node));
+		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",
+					"""
+							<?xml version="1.0" encoding="UTF-8"?>
+							<definitions xmlns="http://www.omg.org/spec/BPMN/20100524/MODEL" xmlns:flowable="http://flowable.org/bpmn" targetNamespace="test">
+							 <process id="approval" isExecutable="true">
+							  <startEvent id="start"/><sequenceFlow id="s1" sourceRef="start" targetRef="approve"/>
+							  <userTask id="approve" name="审批" flowable:assignee="1001"/><sequenceFlow id="s2" sourceRef="approve" targetRef="end"/><endEvent id="end"/>
+							 </process>
+							</definitions>
+							""")
+			.deploy();
+	}
+
+	@AfterEach
+	void tearDown() {
+		if (server != null)
+			server.stop(0);
+		if (engine != null)
+			engine.close();
+		if (jdbc != null)
+			jdbc.execute("SHUTDOWN");
+	}
+
+	@Test
+	void draftsAndTaskEventsSurviveNewServiceInstancesAndHttpRetries() throws Exception {
+		// 请求仍能读取个人表单，但操作日志使用的 Jackson 不得输出标题和载荷。
+		var logMapper = com.pig4cloud.pigx.common.log.util.JacksonSensitiveFieldUtil.getObjectMapper().copy();
+		logMapper.setFilterProvider(new com.fasterxml.jackson.databind.ser.impl.SimpleFilterProvider()
+			.setDefaultFilter(com.fasterxml.jackson.databind.ser.impl.SimpleBeanPropertyFilter.serializeAll()));
+		var input = logMapper.readValue(
+				"{\"kind\":\"APPLICATION\",\"flowId\":\"approval\",\"title\":\"private-title\",\"payload\":\"{\\\"secret\\\":1}\",\"revision\":0}",
+				FlowDraftService.Draft.class);
+		assertThat(input.title()).isEqualTo("private-title");
+		assertThat(input.payload()).contains("secret");
+		assertThat(logMapper.writeValueAsString(new Object[] { input })).doesNotContain("private-title", "secret",
+				"payload", "title");
+		assertThat(FlowNoticeStatusEnum.values()).extracting(FlowNoticeStatusEnum::getCode)
+			.containsExactly("PENDING", "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<FlowReliabilityService> 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);
+	}
+
+}