refactor(flow): migrate durable workspace services to MyBatis Plus

(cherry picked from commit 1b0f4b664cb310fe236dfb046bef9c8248543d71)
2026-11-08 04:39:00 +08:00 · 2026-11-03 14:56:16 +08:00
1 parent b04588ffae
commit e0d43f41a1
@@ -1814,6 +1814,75 @@ CREATE TABLE `process_task_add_sign_record` (
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci ROW_FORMAT=DYNAMIC COMMENT='流程动态加签批次记录';


+-- ----------------------------
+-- Table structure for flow_draft
+-- ----------------------------
+DROP TABLE IF EXISTS `flow_draft`;
+CREATE TABLE `flow_draft` (
+  `id` varchar(36) COLLATE utf8mb4_general_ci NOT NULL COMMENT '草稿ID',
+  `tenant_id` bigint NOT NULL COMMENT '所属租户ID',
+  `user_id` bigint NOT NULL COMMENT '草稿所属用户ID',
+  `kind` varchar(16) COLLATE utf8mb4_general_ci NOT NULL COMMENT '草稿类型：DESIGN 流程设计，APPLICATION 申请',
+  `flow_id` varchar(64) COLLATE utf8mb4_general_ci NOT NULL DEFAULT '' COMMENT '关联流程ID，未建立流程时为空串',
+  `title` varchar(200) COLLATE utf8mb4_general_ci NOT NULL COMMENT '草稿标题',
+  `payload` longtext COLLATE utf8mb4_general_ci NOT NULL COMMENT '草稿内容 JSON，最多 1 MB',
+  `revision` bigint NOT NULL COMMENT '乐观锁版本号',
+  `updated_at` timestamp(3) NOT NULL COMMENT '最后保存时间',
+  PRIMARY KEY (`id`),
+  KEY `idx_draft_owner` (`tenant_id`,`user_id`,`kind`,`updated_at`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT='个人流程草稿';
+
+-- ----------------------------
+-- Table structure for flow_notice
+-- ----------------------------
+DROP TABLE IF EXISTS `flow_notice`;
+CREATE TABLE `flow_notice` (
+  `id` varchar(36) COLLATE utf8mb4_general_ci NOT NULL COMMENT '通知事件ID',
+  `tenant_id` bigint NOT NULL COMMENT '所属租户ID',
+  `user_id` varchar(64) COLLATE utf8mb4_general_ci NOT NULL COMMENT '接收人ID，超时提醒到期前为空串',
+  `event_key` varchar(180) COLLATE utf8mb4_general_ci NOT NULL COMMENT '事件去重键',
+  `kind` varchar(16) COLLATE utf8mb4_general_ci NOT NULL COMMENT '通知类型：TODO 待办，URGE 催办，OVERDUE 超时提醒',
+  `title` varchar(200) COLLATE utf8mb4_general_ci NOT NULL COMMENT '通知标题',
+  `task_id` varchar(64) COLLATE utf8mb4_general_ci NOT NULL COMMENT '引擎任务ID',
+  `instance_id` varchar(64) COLLATE utf8mb4_general_ci NOT NULL COMMENT '引擎流程实例ID',
+  `created_at` timestamp(3) NOT NULL COMMENT '创建时间',
+  `status` varchar(16) COLLATE utf8mb4_general_ci NOT NULL DEFAULT 'SENT' COMMENT '状态：PENDING 待到期，READY 待投递，SENT 已投递，CANCELLED 已取消，DISABLED 已停用',
+  `scheduled_at` timestamp(3) NULL DEFAULT NULL COMMENT '计划投递或到期时间',
+  `sent_at` timestamp(3) NULL DEFAULT NULL COMMENT '投递到统一站内信的时间',
+  `reminder_key` varchar(64) COLLATE utf8mb4_general_ci DEFAULT NULL COMMENT '超时提醒任务ID，每个任务最多一条',
+  PRIMARY KEY (`id`),
+  UNIQUE KEY `uk_notice_event` (`tenant_id`,`user_id`,`event_key`),
+  UNIQUE KEY `uk_notice_reminder` (`tenant_id`,`reminder_key`),
+  KEY `idx_notice_owner` (`tenant_id`,`user_id`,`created_at`),
+  KEY `idx_notice_due` (`status`,`scheduled_at`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT='流程通知事件';
+
+-- ----------------------------
+-- Table structure for flow_callback
+-- ----------------------------
+DROP TABLE IF EXISTS `flow_callback`;
+CREATE TABLE `flow_callback` (
+  `id` varchar(36) COLLATE utf8mb4_general_ci NOT NULL COMMENT '回调事件ID，同时作为接收方幂等键',
+  `tenant_id` bigint NOT NULL COMMENT '所属租户ID',
+  `task_id` varchar(64) COLLATE utf8mb4_general_ci NOT NULL COMMENT '引擎任务ID',
+  `instance_id` varchar(64) COLLATE utf8mb4_general_ci NOT NULL COMMENT '引擎流程实例ID',
+  `target_url` varchar(2000) COLLATE utf8mb4_general_ci NOT NULL COMMENT '回调目标地址',
+  `payload` longtext COLLATE utf8mb4_general_ci NOT NULL COMMENT '回调请求体 JSON',
+  `status` varchar(16) COLLATE utf8mb4_general_ci NOT NULL COMMENT '状态：PENDING 待投递，SENDING 投递中，DELIVERED 已送达，FAILED 失败',
+  `attempts` int NOT NULL DEFAULT '0' COMMENT '已尝试次数',
+  `max_attempts` int NOT NULL DEFAULT '5' COMMENT '最大尝试次数',
+  `next_at` timestamp(3) NOT NULL COMMENT '下次投递时间',
+  `history` longtext COLLATE utf8mb4_general_ci COMMENT '投递历史 JSON',
+  `lease_token` varchar(36) COLLATE utf8mb4_general_ci DEFAULT NULL COMMENT '投递租约令牌',
+  `lease_until` timestamp(3) NULL DEFAULT NULL COMMENT '租约到期时间',
+  `last_error` varchar(500) COLLATE utf8mb4_general_ci DEFAULT NULL COMMENT '最近一次失败原因',
+  `created_at` timestamp(3) NOT NULL COMMENT '创建时间',
+  `updated_at` timestamp(3) NOT NULL COMMENT '更新时间',
+  PRIMARY KEY (`id`),
+  UNIQUE KEY `uk_callback_task` (`tenant_id`,`task_id`),
+  KEY `idx_callback_due` (`status`,`next_at`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT='流程节点 HTTP 回调';
+
 -- ----------------------------
 -- Table structure for process_starter
 -- ----------------------------
@@ -3240,30 +3309,3 @@ 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 COLLATE=utf8mb4_general_ci;
-CREATE TABLE IF NOT EXISTS flow_notice (
- id varchar(36) NOT NULL, tenant_id bigint NOT NULL, user_id varchar(64) NOT NULL,
- event_key varchar(180) NOT NULL, kind varchar(16) NOT NULL, title varchar(200) NOT NULL,
- 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 COLLATE=utf8mb4_general_ci;
-CREATE TABLE IF NOT EXISTS flow_callback (
- id varchar(36) NOT NULL, tenant_id bigint NOT NULL, task_id varchar(64) NOT NULL,
- instance_id varchar(64) NOT NULL, target_url varchar(2000) NOT NULL, payload longtext NOT NULL,
- 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 COLLATE=utf8mb4_general_ci;
@@ -287,32 +287,73 @@ CREATE TABLE `bmp_oa_leave` (
  PRIMARY KEY (`id`)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT='请假表单';

-SET FOREIGN_KEY_CHECKS = 1;
+-- ----------------------------
+-- Table structure for flow_draft
+-- ----------------------------
+DROP TABLE IF EXISTS `flow_draft`;
+CREATE TABLE `flow_draft` (
+  `id` varchar(36) COLLATE utf8mb4_general_ci NOT NULL COMMENT '草稿ID',
+  `tenant_id` bigint NOT NULL COMMENT '所属租户ID',
+  `user_id` bigint NOT NULL COMMENT '草稿所属用户ID',
+  `kind` varchar(16) COLLATE utf8mb4_general_ci NOT NULL COMMENT '草稿类型：DESIGN 流程设计，APPLICATION 申请',
+  `flow_id` varchar(64) COLLATE utf8mb4_general_ci NOT NULL DEFAULT '' COMMENT '关联流程ID，未建立流程时为空串',
+  `title` varchar(200) COLLATE utf8mb4_general_ci NOT NULL COMMENT '草稿标题',
+  `payload` longtext COLLATE utf8mb4_general_ci NOT NULL COMMENT '草稿内容 JSON，最多 1 MB',
+  `revision` bigint NOT NULL COMMENT '乐观锁版本号',
+  `updated_at` timestamp(3) NOT NULL COMMENT '最后保存时间',
+  PRIMARY KEY (`id`),
+  KEY `idx_draft_owner` (`tenant_id`,`user_id`,`kind`,`updated_at`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT='个人流程草稿';

-- 第一批持久化能力：三张业务表。已安装旧五表版本时，建表后另执行 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 COLLATE=utf8mb4_general_ci;
-CREATE TABLE IF NOT EXISTS flow_notice (
- id varchar(36) NOT NULL, tenant_id bigint NOT NULL, user_id varchar(64) NOT NULL,
- event_key varchar(180) NOT NULL, kind varchar(16) NOT NULL, title varchar(200) NOT NULL,
- 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 COLLATE=utf8mb4_general_ci;
-CREATE TABLE IF NOT EXISTS flow_callback (
- id varchar(36) NOT NULL, tenant_id bigint NOT NULL, task_id varchar(64) NOT NULL,
- instance_id varchar(64) NOT NULL, target_url varchar(2000) NOT NULL, payload longtext NOT NULL,
- 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 COLLATE=utf8mb4_general_ci;
+-- ----------------------------
+-- Table structure for flow_notice
+-- ----------------------------
+DROP TABLE IF EXISTS `flow_notice`;
+CREATE TABLE `flow_notice` (
+  `id` varchar(36) COLLATE utf8mb4_general_ci NOT NULL COMMENT '通知事件ID',
+  `tenant_id` bigint NOT NULL COMMENT '所属租户ID',
+  `user_id` varchar(64) COLLATE utf8mb4_general_ci NOT NULL COMMENT '接收人ID，超时提醒到期前为空串',
+  `event_key` varchar(180) COLLATE utf8mb4_general_ci NOT NULL COMMENT '事件去重键',
+  `kind` varchar(16) COLLATE utf8mb4_general_ci NOT NULL COMMENT '通知类型：TODO 待办，URGE 催办，OVERDUE 超时提醒',
+  `title` varchar(200) COLLATE utf8mb4_general_ci NOT NULL COMMENT '通知标题',
+  `task_id` varchar(64) COLLATE utf8mb4_general_ci NOT NULL COMMENT '引擎任务ID',
+  `instance_id` varchar(64) COLLATE utf8mb4_general_ci NOT NULL COMMENT '引擎流程实例ID',
+  `created_at` timestamp(3) NOT NULL COMMENT '创建时间',
+  `status` varchar(16) COLLATE utf8mb4_general_ci NOT NULL DEFAULT 'SENT' COMMENT '状态：PENDING 待到期，READY 待投递，SENT 已投递，CANCELLED 已取消，DISABLED 已停用',
+  `scheduled_at` timestamp(3) NULL DEFAULT NULL COMMENT '计划投递或到期时间',
+  `sent_at` timestamp(3) NULL DEFAULT NULL COMMENT '投递到统一站内信的时间',
+  `reminder_key` varchar(64) COLLATE utf8mb4_general_ci DEFAULT NULL COMMENT '超时提醒任务ID，每个任务最多一条',
+  PRIMARY KEY (`id`),
+  UNIQUE KEY `uk_notice_event` (`tenant_id`,`user_id`,`event_key`),
+  UNIQUE KEY `uk_notice_reminder` (`tenant_id`,`reminder_key`),
+  KEY `idx_notice_owner` (`tenant_id`,`user_id`,`created_at`),
+  KEY `idx_notice_due` (`status`,`scheduled_at`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT='流程通知事件';
+
+-- ----------------------------
+-- Table structure for flow_callback
+-- ----------------------------
+DROP TABLE IF EXISTS `flow_callback`;
+CREATE TABLE `flow_callback` (
+  `id` varchar(36) COLLATE utf8mb4_general_ci NOT NULL COMMENT '回调事件ID，同时作为接收方幂等键',
+  `tenant_id` bigint NOT NULL COMMENT '所属租户ID',
+  `task_id` varchar(64) COLLATE utf8mb4_general_ci NOT NULL COMMENT '引擎任务ID',
+  `instance_id` varchar(64) COLLATE utf8mb4_general_ci NOT NULL COMMENT '引擎流程实例ID',
+  `target_url` varchar(2000) COLLATE utf8mb4_general_ci NOT NULL COMMENT '回调目标地址',
+  `payload` longtext COLLATE utf8mb4_general_ci NOT NULL COMMENT '回调请求体 JSON',
+  `status` varchar(16) COLLATE utf8mb4_general_ci NOT NULL COMMENT '状态：PENDING 待投递，SENDING 投递中，DELIVERED 已送达，FAILED 失败',
+  `attempts` int NOT NULL DEFAULT '0' COMMENT '已尝试次数',
+  `max_attempts` int NOT NULL DEFAULT '5' COMMENT '最大尝试次数',
+  `next_at` timestamp(3) NOT NULL COMMENT '下次投递时间',
+  `history` longtext COLLATE utf8mb4_general_ci COMMENT '投递历史 JSON',
+  `lease_token` varchar(36) COLLATE utf8mb4_general_ci DEFAULT NULL COMMENT '投递租约令牌',
+  `lease_until` timestamp(3) NULL DEFAULT NULL COMMENT '租约到期时间',
+  `last_error` varchar(500) COLLATE utf8mb4_general_ci DEFAULT NULL COMMENT '最近一次失败原因',
+  `created_at` timestamp(3) NOT NULL COMMENT '创建时间',
+  `updated_at` timestamp(3) NOT NULL COMMENT '更新时间',
+  PRIMARY KEY (`id`),
+  UNIQUE KEY `uk_callback_task` (`tenant_id`,`task_id`),
+  KEY `idx_callback_due` (`status`,`next_at`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT='流程节点 HTTP 回调';
+
+SET FOREIGN_KEY_CHECKS = 1;
@@ -14,7 +14,7 @@ public enum FlowCallbackStatusEnum {
 	SENDING("SENDING"),
 	/** 接收方返回 HTTP 2xx，投递结束。 */
 	DELIVERED("DELIVERED"),
-	/** 尝试耗尽或最终租约过期，允许发起人手动重试。 */
+	/** 尝试耗尽或最终租约过期，需人工核对接收方后处理。 */
 	FAILED("FAILED");

 	/** 持久化编码，不能随枚举重命名改变。 */
@@ -0,0 +1,26 @@
+package com.pig4cloud.pigx.flow.constant;
+
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+/** 流程通知事件类型；编码与 flow_notice.kind 保持兼容，标签用于统一站内信标题。 */
+@Getter
+@RequiredArgsConstructor
+public enum FlowNoticeKindEnum {
+
+	/** 新分配的待办任务。 */
+	TODO("TODO", "待办"),
+
+	/** 发起人手动催办。 */
+	URGE("URGE", "催办"),
+
+	/** 审批节点超时提醒。 */
+	OVERDUE("OVERDUE", "超时提醒");
+
+	/** 持久化编码，不能随枚举重命名改变。 */
+	private final String code;
+
+	/** 站内信标题前缀。 */
+	private final String label;
+
+}
@@ -0,0 +1,51 @@
+package com.pig4cloud.pigx.flow.dto;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import io.swagger.v3.oas.annotations.media.Schema;
+import jakarta.validation.constraints.NotBlank;
+import jakarta.validation.constraints.NotNull;
+import jakarta.validation.constraints.Pattern;
+import jakarta.validation.constraints.Size;
+import lombok.Data;
+
+/**
+ * 个人流程草稿保存请求；租户和用户只能由认证上下文提供。标题和内容仅反序列化，避免操作日志记录个人表单。
+ */
+@Data
+@Schema(description = "个人流程草稿保存请求")
+public class FlowDraftDTO {
+
+	/** 首次保存为空；更新时为已有草稿 ID。 */
+	@Schema(description = "首次保存为空，更新使用已有草稿 ID")
+	private String id;
+
+	/** 草稿类型：DESIGN 流程设计，APPLICATION 动态表单申请。 */
+	@NotBlank
+	@Pattern(regexp = "DESIGN|APPLICATION", message = "草稿类型无效")
+	@Schema(description = "草稿类型", allowableValues = { "DESIGN", "APPLICATION" })
+	private String kind;
+
+	/** 关联流程 ID；尚未建立流程的设计草稿为空串。 */
+	@NotNull
+	@Size(max = 64)
+	@Schema(description = "关联流程 ID，尚未建立流程的设计草稿可为空串")
+	private String flowId;
+
+	/** 草稿标题，最多 200 字符。 */
+	@NotBlank
+	@Size(max = 200)
+	@JsonProperty(access = JsonProperty.Access.WRITE_ONLY)
+	@Schema(description = "非空标题，最多 200 字符")
+	private String title;
+
+	/** 与 kind 对应的 JSON 对象字符串，最多 1 MB UTF-8；字节数和 JSON 结构由服务端校验。 */
+	@NotBlank
+	@JsonProperty(access = JsonProperty.Access.WRITE_ONLY)
+	@Schema(description = "与 kind 对应的 JSON 对象字符串，最多 1 MB UTF-8")
+	private String payload;
+
+	/** 首次保存为 0；更新时为读取到的版本号，冲突时拒绝覆盖。 */
+	@Schema(description = "首次保存为 0，更新为读取时的版本号")
+	private long revision;
+
+}
@@ -0,0 +1,66 @@
+package com.pig4cloud.pigx.flow.entity;
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
+import com.pig4cloud.pigx.common.core.util.TenantTable;
+import lombok.Data;
+
+import java.time.LocalDateTime;
+
+/** 节点完成 HTTP 回调；租约保证多实例互斥，事件 ID 供接收方幂等。 */
+@Data
+@TenantTable
+@TableName("flow_callback")
+public class FlowCallback {
+
+	/** 回调事件 ID。 */
+	@TableId(type = IdType.ASSIGN_UUID)
+	private String id;
+
+	/** 所属租户 ID。 */
+	private Long tenantId;
+
+	/** 引擎任务 ID，同一任务只入队一次。 */
+	private String taskId;
+
+	/** 引擎流程实例 ID。 */
+	private String instanceId;
+
+	/** 回调目标地址。 */
+	private String targetUrl;
+
+	/** 回调请求体 JSON。 */
+	private String payload;
+
+	/** 状态，见 FlowCallbackStatusEnum。 */
+	private String status;
+
+	/** 已尝试次数；为空时使用数据库默认值 0。 */
+	private Integer attempts;
+
+	/** 最大尝试次数；为空时使用数据库默认值 5。 */
+	private Integer maxAttempts;
+
+	/** 下次投递时间。 */
+	private LocalDateTime nextAt;
+
+	/** 投递历史 JSON 数组。 */
+	private String history;
+
+	/** 当前投递租约令牌。 */
+	private String leaseToken;
+
+	/** 租约到期时间。 */
+	private LocalDateTime leaseUntil;
+
+	/** 最近一次失败原因。 */
+	private String lastError;
+
+	/** 创建时间。 */
+	private LocalDateTime createdAt;
+
+	/** 更新时间。 */
+	private LocalDateTime updatedAt;
+
+}
@@ -0,0 +1,58 @@
+package com.pig4cloud.pigx.flow.entity;
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
+import com.baomidou.mybatisplus.annotation.Version;
+import com.pig4cloud.pigx.common.core.util.TenantTable;
+import io.swagger.v3.oas.annotations.media.Schema;
+import lombok.Data;
+
+import java.time.LocalDateTime;
+
+/** 个人流程草稿；只允许所属租户和用户读写，保存不部署或启动流程。 */
+@Data
+@TenantTable
+@TableName("flow_draft")
+@Schema(description = "个人流程草稿")
+public class FlowDraft {
+
+	/** 草稿 ID。 */
+	@TableId(type = IdType.ASSIGN_UUID)
+	@Schema(description = "草稿ID")
+	private String id;
+
+	/** 所属租户 ID。 */
+	@Schema(description = "所属租户ID")
+	private Long tenantId;
+
+	/** 草稿所属用户 ID。 */
+	@Schema(description = "草稿所属用户ID")
+	private Long userId;
+
+	/** 草稿类型：DESIGN 流程设计，APPLICATION 申请。 */
+	@Schema(description = "草稿类型", allowableValues = { "DESIGN", "APPLICATION" })
+	private String kind;
+
+	/** 关联流程 ID，未建立流程的设计草稿为空串。 */
+	@Schema(description = "关联流程ID")
+	private String flowId;
+
+	/** 草稿标题，最多 200 字符。 */
+	@Schema(description = "草稿标题")
+	private String title;
+
+	/** 草稿内容 JSON，列表查询不返回。 */
+	@Schema(description = "草稿内容 JSON")
+	private String payload;
+
+	/** 乐观锁版本号，首次保存为 1，由 MyBatis-Plus 乐观锁插件校验并递增。 */
+	@Version
+	@Schema(description = "版本号")
+	private Long revision;
+
+	/** 最后保存时间。 */
+	@Schema(description = "最后保存时间")
+	private LocalDateTime updatedAt;
+
+}
@@ -0,0 +1,57 @@
+package com.pig4cloud.pigx.flow.entity;
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
+import com.pig4cloud.pigx.common.core.util.TenantTable;
+import lombok.Data;
+
+import java.time.LocalDateTime;
+
+/** 流程通知事件；READY 由扫描器投递到统一站内信，PENDING 为等待到期的超时提醒。 */
+@Data
+@TenantTable
+@TableName("flow_notice")
+public class FlowNotice {
+
+	/** 通知事件 ID，同时作为站内信幂等键的一部分。 */
+	@TableId(type = IdType.ASSIGN_UUID)
+	private String id;
+
+	/** 所属租户 ID。 */
+	private Long tenantId;
+
+	/** 接收人 ID；超时提醒到期前为空串。 */
+	private String userId;
+
+	/** 事件去重键，与租户、接收人组成唯一键。 */
+	private String eventKey;
+
+	/** 通知类型，见 FlowNoticeKindEnum。 */
+	private String kind;
+
+	/** 通知标题。 */
+	private String title;
+
+	/** 引擎任务 ID。 */
+	private String taskId;
+
+	/** 引擎流程实例 ID。 */
+	private String instanceId;
+
+	/** 创建时间；超时提醒到期时更新为提醒时间。 */
+	private LocalDateTime createdAt;
+
+	/** 状态，见 FlowNoticeStatusEnum。 */
+	private String status;
+
+	/** 计划投递或到期时间。 */
+	private LocalDateTime scheduledAt;
+
+	/** 投递到统一站内信的时间。 */
+	private LocalDateTime sentAt;
+
+	/** 超时提醒对应的任务 ID，每个任务最多一条。 */
+	private String reminderKey;
+
+}
@@ -1,10 +1,8 @@
 package com.pig4cloud.pigx.flow.config;

-import com.pig4cloud.pigx.flow.support.workspace.FlowCallbackDelivery;
-import com.pig4cloud.pigx.flow.constant.FlowNoticeStatusEnum;
-import com.pig4cloud.pigx.flow.support.workspace.FlowReliabilityService;
+import com.pig4cloud.pigx.flow.service.IFlowCallbackService;
+import com.pig4cloud.pigx.flow.service.IFlowNoticeService;
 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;
@@ -12,7 +10,6 @@ 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;
@@ -27,11 +24,9 @@ import org.flowable.common.engine.api.delegate.event.FlowableEventListener;
@Slf4j
 public class FlowDurableConfiguration {

-	private final JdbcTemplate jdbc;
+	private final ObjectProvider<IFlowNoticeService> noticeService;

-	private final ObjectProvider<FlowReliabilityService> reliability;
-
-	private final ObjectProvider<FlowCallbackDelivery> delivery;
+	private final ObjectProvider<IFlowCallbackService> callbackService;

 	/**
 	 * 注册持久化监听器，保留引擎已配置的监听器。
@@ -43,7 +38,7 @@ public class FlowDurableConfiguration {
 			var listeners = new ArrayList<FlowableEventListener>();
 			if (Objects.nonNull(engine.getEventListeners()))
 				listeners.addAll(engine.getEventListeners());
-			listeners.add(new FlowDurableEventListener(reliability));
+			listeners.add(new FlowDurableEventListener(noticeService));
 			engine.setEventListeners(listeners);
 		};
 	}
@@ -63,19 +58,13 @@ public class FlowDurableConfiguration {
 	/** 每轮最多处理 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) {
+		IFlowCallbackService service = callbackService.getObject();
+		for (var row : service.listDue()) {
 			try {
-				delivery.getObject().deliver(String.valueOf(row.get("id")));
+				service.deliver(row.getId());
 			}
 			catch (Exception exception) {
-				log.warn("流程回调投递暂未完成，事件 {}", row.get("id"), exception);
+				log.warn("流程回调投递暂未完成，事件 {}", row.getId(), exception);
 			}
 		}
 	}
@@ -83,34 +72,28 @@ public class FlowDurableConfiguration {
 	/** 独立扫描最多 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) {
+		IFlowNoticeService service = noticeService.getObject();
+		for (var row : service.listDueReminders()) {
 			try {
-				reliability.getObject()
-					.remind(((Number) row.get("tenant_id")).longValue(), String.valueOf(row.get("task_id")));
+				service.remind(row.getTenantId(), row.getTaskId());
 			}
 			catch (Exception exception) {
-				log.warn("流程超时提醒暂未完成，任务 {}", row.get("task_id"), exception);
+				log.warn("流程超时提醒暂未完成，任务 {}", row.getTaskId(), exception);
 			}
 		}
 	}

-	/** 复用流程事件表投递站内信；失败延迟一分钟再试，不阻塞引擎事务。 */
+	/** 投递流程通知到统一站内信；失败延迟一分钟再试，不阻塞引擎事务。 */
 	@Scheduled(fixedDelayString = "${flow.delivery.scan-delay:10000}", scheduler = "flowDeliveryScheduler")
 	public void scanNotices() {
-		var notices = jdbc.queryForList(
-				"SELECT id FROM flow_notice WHERE status=? AND scheduled_at<=CURRENT_TIMESTAMP(3) ORDER BY scheduled_at LIMIT 100",
-				FlowNoticeStatusEnum.READY.getCode());
-		for (var row : notices) {
+		IFlowNoticeService service = noticeService.getObject();
+		for (var row : service.listDueNotices()) {
 			try {
-				reliability.getObject().deliverNotice(String.valueOf(row.get("id")));
+				service.deliverNotice(row.getId());
 			}
 			catch (Exception exception) {
-				jdbc.update("UPDATE flow_notice SET scheduled_at=TIMESTAMPADD(SECOND,60,CURRENT_TIMESTAMP(3)) WHERE id=? AND status=?",
-						row.get("id"), FlowNoticeStatusEnum.READY.getCode());
-				log.warn("流程站内信暂未发送，事件 {}", row.get("id"), exception);
+				service.delayNotice(row.getId());
+				log.warn("流程站内信暂未发送，事件 {}", row.getId(), exception);
 			}
 		}
 	}
@@ -3,27 +3,31 @@ 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.support.workspace.FlowDraftService;
-import com.pig4cloud.pigx.flow.support.workspace.FlowReliabilityService;
+import com.pig4cloud.pigx.flow.entity.FlowDraft;
+import com.pig4cloud.pigx.flow.dto.FlowDraftDTO;
+import com.pig4cloud.pigx.flow.service.IFlowDraftService;
+import com.pig4cloud.pigx.flow.service.IFlowNoticeService;
+import com.baomidou.mybatisplus.core.metadata.IPage;
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+import jakarta.validation.Valid;
+import org.springdoc.core.annotations.ParameterObject;
 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 = "当前用户草稿和通知，以及本人发起实例的催办与回调")
+@Tag(name = "流程工作区", description = "当前用户草稿，以及本人发起实例的催办")
@RestController
@RequestMapping("/workspace")
@RequiredArgsConstructor
 public class FlowWorkspaceController {

-	private final FlowDraftService drafts;
+	private final IFlowDraftService draftService;

-	private final FlowReliabilityService reliability;
+	private final IFlowNoticeService noticeService;

 	private long tenant() {
 		return Objects.requireNonNull(TenantContextHolder.getTenantId(), "缺少租户");
@@ -35,29 +39,32 @@ public class FlowWorkspaceController {

 	/**
 	 * 保存个人草稿。
-	 * @param draft 草稿内容与版本号
+	 * @param draft 草稿内容与版本号，字段约束由注解校验
 	 * @return 新版本草稿
-	 * @throws IllegalArgumentException 草稿类型、标题或 JSON 内容无效
+	 * @throws org.springframework.web.bind.MethodArgumentNotValidException 草稿类型、标题或流程 ID
+	 * 不满足注解约束
+	 * @throws IllegalArgumentException 内容超过 1 MB 或不是 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));
+	public R<FlowDraft> save(@Valid @RequestBody FlowDraftDTO draft) {
+		return R.ok(draftService.saveDraft(tenant(), user(), draft));
 	}

 	/**
 	 * 查询个人草稿。
+	 * @param page 分页参数 current、size，受全局分页上限约束
 	 * @param kind DESIGN 或 APPLICATION
-	 * @param page 页码
-	 * @return 草稿摘要
+	 * @param flowId 可选的流程 ID，为空时查询全部流程
+	 * @return 草稿摘要分页，不含表单载荷
 	 */
 	@Operation(summary = "分页查询个人草稿")
 	@GetMapping("/drafts")
-	public R<List<Map<String, Object>>> list(@RequestParam String kind, @RequestParam(defaultValue = "1") int page,
+	public R<IPage<FlowDraft>> page(@ParameterObject Page<FlowDraft> page, @RequestParam String kind,
 			@RequestParam(required = false) String flowId) {
-		return R.ok(drafts.list(tenant(), user(), kind, page, flowId));
+		return R.ok(draftService.pageDraft(page, tenant(), user(), kind, flowId));
 	}

 	/**
@@ -68,8 +75,8 @@ public class FlowWorkspaceController {
 	 */
 	@Operation(summary = "读取个人草稿")
 	@GetMapping("/drafts/{id}")
-	public R<Map<String, Object>> get(@PathVariable String id) {
-		return R.ok(drafts.get(tenant(), user(), id));
+	public R<FlowDraft> get(@PathVariable String id) {
+		return R.ok(draftService.getDraft(tenant(), user(), id));
 	}

 	/**
@@ -82,7 +89,7 @@ public class FlowWorkspaceController {
 	@SysLog("按版本删除个人草稿")
 	@DeleteMapping("/drafts/{id}")
 	public R<Boolean> delete(@PathVariable String id, @RequestParam long revision) {
-		return R.ok(drafts.delete(tenant(), user(), id, revision));
+		return R.ok(draftService.removeDraft(tenant(), user(), id, revision));
 	}

 	/**
@@ -95,33 +102,7 @@ public class FlowWorkspaceController {
 	@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();
+		return R.ok(noticeService.urge(tenant(), user(), id));
 	}

 }
@@ -0,0 +1,11 @@
+package com.pig4cloud.pigx.flow.mapper;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.pig4cloud.pigx.flow.entity.FlowCallback;
+import org.apache.ibatis.annotations.Mapper;
+
+/** 节点 HTTP 回调表 flow_callback 数据访问。 */
+@Mapper
+public interface FlowCallbackMapper extends BaseMapper<FlowCallback> {
+
+}
@@ -0,0 +1,11 @@
+package com.pig4cloud.pigx.flow.mapper;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.pig4cloud.pigx.flow.entity.FlowDraft;
+import org.apache.ibatis.annotations.Mapper;
+
+/** 个人流程草稿表 flow_draft 数据访问。 */
+@Mapper
+public interface FlowDraftMapper extends BaseMapper<FlowDraft> {
+
+}
@@ -0,0 +1,11 @@
+package com.pig4cloud.pigx.flow.mapper;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.pig4cloud.pigx.flow.entity.FlowNotice;
+import org.apache.ibatis.annotations.Mapper;
+
+/** 流程通知事件表 flow_notice 数据访问。 */
+@Mapper
+public interface FlowNoticeMapper extends BaseMapper<FlowNotice> {
+
+}
@@ -0,0 +1,33 @@
+package com.pig4cloud.pigx.flow.service;
+
+import com.baomidou.mybatisplus.extension.service.IService;
+import com.pig4cloud.pigx.flow.entity.FlowCallback;
+import org.flowable.task.service.impl.persistence.entity.TaskEntity;
+
+import java.util.List;
+
+/** 节点完成 HTTP 回调的至少一次投递；租约用于多实例互斥，接收方使用事件 ID 实现幂等。 */
+public interface IFlowCallbackService extends IService<FlowCallback> {
+
+	/**
+	 * 在引擎事务内为已完成任务入队回调；同一任务只入队一次。
+	 * @param task 已完成的引擎任务
+	 * @param tenant 任务所属租户
+	 * @param target 节点配置的 HTTP(S) 回调地址
+	 * @throws IllegalArgumentException 回调地址不是 HTTP(S)
+	 */
+	void enqueue(TaskEntity task, long tenant, String target);
+
+	/**
+	 * 将最后一次尝试租约已过期的回调置为失败，并返回最多 20 条到期回调。
+	 * @return 到期或租约过期的回调，只包含 ID
+	 */
+	List<FlowCallback> listDue();
+
+	/**
+	 * 尝试投递一条到期回调，不在网络请求期间持有数据库事务。
+	 * @param id 扫描得到的回调 ID
+	 */
+	void deliver(String id);
+
+}
@@ -0,0 +1,54 @@
+package com.pig4cloud.pigx.flow.service;
+
+import com.baomidou.mybatisplus.core.metadata.IPage;
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+import com.baomidou.mybatisplus.extension.service.IService;
+import com.pig4cloud.pigx.flow.dto.FlowDraftDTO;
+import com.pig4cloud.pigx.flow.entity.FlowDraft;
+
+/** 个人流程草稿服务；所有操作同时限定租户和用户，保存不部署或启动流程。 */
+public interface IFlowDraftService extends IService<FlowDraft> {
+
+	/**
+	 * 保存草稿，使用乐观锁版本号防止多窗口覆盖。
+	 * @param tenant 当前租户
+	 * @param user 当前用户
+	 * @param draft 已通过注解校验的草稿；首次保存 id 为空、revision 为 0
+	 * @return 保存后的草稿（包括新的版本号）
+	 * @throws IllegalArgumentException 内容超过 1 MB 或不是 JSON 对象
+	 * @throws IllegalStateException 草稿不存在、无权访问或版本冲突
+	 */
+	FlowDraft saveDraft(long tenant, long user, FlowDraftDTO draft);
+
+	/**
+	 * 读取当前用户的草稿。
+	 * @param tenant 当前租户
+	 * @param user 当前用户
+	 * @param id 草稿 ID
+	 * @return 草稿详情，包含内容 JSON
+	 * @throws IllegalStateException 草稿不存在或不属于当前用户
+	 */
+	FlowDraft getDraft(long tenant, long user, String id);
+
+	/**
+	 * 分页查询个人草稿摘要，不返回内容 JSON。
+	 * @param page 分页参数，受全局分页上限约束
+	 * @param tenant 当前租户
+	 * @param user 当前用户
+	 * @param kind DESIGN 或 APPLICATION
+	 * @param flowId 可选的流程 ID，为 null 时查询全部流程
+	 * @return 按更新时间倒序的草稿分页，payload 为 null
+	 */
+	IPage<FlowDraft> pageDraft(Page<FlowDraft> page, long tenant, long user, String kind, String flowId);
+
+	/**
+	 * 按版本删除自己的草稿，版本冲突时不删除。
+	 * @param tenant 当前租户
+	 * @param user 当前用户
+	 * @param id 草稿 ID
+	 * @param revision 打开时的版本号
+	 * @return 是否成功删除
+	 */
+	boolean removeDraft(long tenant, long user, String id, long revision);
+
+}
@@ -0,0 +1,64 @@
+package com.pig4cloud.pigx.flow.service;
+
+import com.baomidou.mybatisplus.extension.service.IService;
+import com.pig4cloud.pigx.flow.entity.FlowNotice;
+import org.flowable.task.service.impl.persistence.entity.TaskEntity;
+
+import java.util.List;
+
+/**
+ * 流程通知事件：待办、催办与超时提醒的持久化及统一站内信投递。写入显式带租户，扫描线程没有租户上下文时由 {@code @TenantTable} 跳过租户过滤。
+ */
+public interface IFlowNoticeService extends IService<FlowNotice> {
+
+	/**
+	 * 在引擎事务内记录任务事件；持久化失败必须回滚引擎操作。
+	 * @param task 引擎当前任务
+	 * @param event FlowableEngineEventType 名称：TASK_CREATED、TASK_ASSIGNED、TASK_COMPLETED 或
+	 * ENTITY_DELETED
+	 */
+	void record(TaskEntity task, String event);
+
+	/**
+	 * 发起人催办；锁定实例实现跨任务、跨进程的每十分钟限频。
+	 * @param tenant 当前租户
+	 * @param user 当前用户
+	 * @param instance 流程实例 ID
+	 * @return 本次接收催办的任务数
+	 * @throws IllegalStateException 非发起人、已结束、无待办或催办过频
+	 */
+	int urge(long tenant, long user, String instance);
+
+	/**
+	 * 查询最多 100 条已到期的超时提醒。
+	 * @return 只包含任务 ID 和租户 ID 的提醒
+	 */
+	List<FlowNotice> listDueReminders();
+
+	/**
+	 * 为到期任务记录一次超时提醒；只使用任务当前接收人，暂停或暂无接收人时十秒后再检查。
+	 * @param tenant 任务所属租户
+	 * @param taskId 到期任务 ID
+	 */
+	void remind(long tenant, String taskId);
+
+	/**
+	 * 查询最多 100 条待投递的通知事件。
+	 * @return 只包含 ID 的通知事件
+	 */
+	List<FlowNotice> listDueNotices();
+
+	/**
+	 * 投递通知事件到统一站内信；行锁防止扫描器并发发送同一事件。
+	 * @param id 通知事件 ID
+	 * @throws IllegalStateException 站内信服务返回失败，由扫描器延迟重试
+	 */
+	void deliverNotice(String id);
+
+	/**
+	 * 投递失败后延迟一分钟再试。
+	 * @param id 通知事件 ID
+	 */
+	void delayNotice(String id);
+
+}
@@ -0,0 +1,183 @@
+package com.pig4cloud.pigx.flow.service.impl;
+
+import cn.hutool.core.convert.Convert;
+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.baomidou.mybatisplus.extension.plugins.pagination.Page;
+import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import com.pig4cloud.pigx.flow.constant.FlowCallbackStatusEnum;
+import com.pig4cloud.pigx.flow.dto.ProcessNodeRecordAssignUserParamDto;
+import com.pig4cloud.pigx.flow.entity.FlowCallback;
+import com.pig4cloud.pigx.flow.mapper.FlowCallbackMapper;
+import com.pig4cloud.pigx.flow.service.IFlowCallbackService;
+import com.pig4cloud.pigx.flow.support.utils.NodeUtil;
+import lombok.RequiredArgsConstructor;
+import org.flowable.task.service.impl.persistence.entity.TaskEntity;
+import org.springframework.dao.DuplicateKeyException;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.support.TransactionTemplate;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.Objects;
+import java.util.UUID;
+
+/** HTTP 回调持久化与投递；租约令牌条件保证只有持有租约的实例能写回结果和历史。 */
+@Service
+@RequiredArgsConstructor
+public class FlowCallbackServiceImpl extends ServiceImpl<FlowCallbackMapper, FlowCallback>
+		implements IFlowCallbackService {
+
+	private final TransactionTemplate transactions;
+
+	/**
+	 * {@inheritDoc}
+	 */
+	@Override
+	public void enqueue(TaskEntity task, long tenant, String target) {
+		if (!NodeUtil.isCallbackUrl(target))
+			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()));
+		LocalDateTime now = LocalDateTime.now();
+		FlowCallback callback = new FlowCallback();
+		callback.setTenantId(tenant);
+		callback.setTaskId(task.getId());
+		callback.setInstanceId(task.getProcessInstanceId());
+		callback.setTargetUrl(target);
+		callback.setPayload(JSONUtil.toJsonStr(payload));
+		callback.setStatus(FlowCallbackStatusEnum.PENDING.getCode());
+		callback.setNextAt(now);
+		callback.setCreatedAt(now);
+		callback.setUpdatedAt(now);
+		try {
+			save(callback);
+		}
+		catch (DuplicateKeyException ignored) {
+			// 唯一键保证同一任务只入队一次；MySQL 唯一键冲突只回滚当前语句，不影响引擎事务。
+		}
+	}
+
+	/**
+	 * {@inheritDoc}
+	 */
+	@Override
+	public List<FlowCallback> listDue() {
+		LocalDateTime now = LocalDateTime.now();
+		// 最后一次尝试进程崩溃时，结果未知，保留审计并置失败供人工核对。
+		lambdaUpdate().set(FlowCallback::getStatus, FlowCallbackStatusEnum.FAILED.getCode())
+			.set(FlowCallback::getLastError, "租约过期，投递结果未知")
+			.set(FlowCallback::getLeaseToken, null)
+			.set(FlowCallback::getLeaseUntil, null)
+			.set(FlowCallback::getUpdatedAt, now)
+			.eq(FlowCallback::getStatus, FlowCallbackStatusEnum.SENDING.getCode())
+			.lt(FlowCallback::getLeaseUntil, now)
+			.apply("attempts >= max_attempts")
+			.update();
+		return lambdaQuery().select(FlowCallback::getId)
+			.and(w -> w
+				.nested(due -> due.eq(FlowCallback::getStatus, FlowCallbackStatusEnum.PENDING.getCode())
+					.le(FlowCallback::getNextAt, now))
+				.or(expired -> expired.eq(FlowCallback::getStatus, FlowCallbackStatusEnum.SENDING.getCode())
+					.lt(FlowCallback::getLeaseUntil, now)))
+			.orderByAsc(FlowCallback::getNextAt)
+			.page(new Page<>(1, 20, false))
+			.getRecords();
+	}
+
+	/**
+	 * {@inheritDoc}
+	 */
+	@Override
+	public void deliver(String id) {
+		String lease = UUID.randomUUID().toString();
+		FlowCallback record = transactions.execute(tx -> {
+			LocalDateTime now = LocalDateTime.now();
+			boolean claimed = lambdaUpdate().set(FlowCallback::getStatus, FlowCallbackStatusEnum.SENDING.getCode())
+				.set(FlowCallback::getLeaseToken, lease)
+				.set(FlowCallback::getLeaseUntil, now.plusSeconds(60))
+				.set(FlowCallback::getUpdatedAt, now)
+				.setSql("attempts = attempts + 1")
+				.eq(FlowCallback::getId, id)
+				.and(w -> w
+					.nested(due -> due.eq(FlowCallback::getStatus, FlowCallbackStatusEnum.PENDING.getCode())
+						.le(FlowCallback::getNextAt, now))
+					.or(expired -> expired.eq(FlowCallback::getStatus, FlowCallbackStatusEnum.SENDING.getCode())
+						.lt(FlowCallback::getLeaseUntil, now)))
+				.apply("attempts < max_attempts")
+				.update();
+			if (!claimed)
+				return null;
+			FlowCallback row = lambdaQuery().eq(FlowCallback::getId, id).eq(FlowCallback::getLeaseToken, lease).one();
+			// 历史不截断，保留每次领取租约的审计记录。
+			JSONArray history = StrUtil.isBlank(row.getHistory()) ? new JSONArray()
+					: JSONUtil.parseArray(row.getHistory());
+			history.add(new JSONObject().set("id", lease)
+				.set("attempt_no", row.getAttempts())
+				.set("result", "STARTED")
+				.set("created_at", Instant.now().toString()));
+			row.setHistory(history.toString());
+			lambdaUpdate().set(FlowCallback::getHistory, row.getHistory())
+				.eq(FlowCallback::getId, id)
+				.eq(FlowCallback::getLeaseToken, lease)
+				.update();
+			return row;
+		});
+		if (Objects.isNull(record))
+			return;
+		boolean success = false;
+		String detail;
+		try (HttpResponse response = HttpRequest.post(record.getTargetUrl())
+			.header("X-Flow-Event-Id", id)
+			.header("Idempotency-Key", id)
+			.body(record.getPayload())
+			.timeout(10000)
+			.setFollowRedirects(false)
+			.execute()) {
+			success = response.isOk();
+			detail = "HTTP " + response.getStatus();
+		}
+		catch (Exception exception) {
+			// 只记录异常类型，避免响应正文、URL 参数或表单数据进入日志。
+			detail = exception.getClass().getSimpleName();
+		}
+		int attempts = record.getAttempts();
+		FlowCallbackStatusEnum state = success ? FlowCallbackStatusEnum.DELIVERED
+				: attempts >= record.getMaxAttempts() ? FlowCallbackStatusEnum.FAILED : FlowCallbackStatusEnum.PENDING;
+		JSONArray history = JSONUtil.parseArray(record.getHistory());
+		JSONObject attempt = history.stream()
+			.map(value -> (JSONObject) value)
+			.filter(value -> lease.equals(value.getStr("id")))
+			.findFirst()
+			.orElseThrow(() -> new IllegalStateException("回调尝试记录缺失"));
+		attempt.set("result", success ? "DELIVERED" : "ERROR");
+		attempt.set("detail", detail);
+		LocalDateTime now = LocalDateTime.now();
+		lambdaUpdate().set(FlowCallback::getStatus, state.getCode())
+			.set(FlowCallback::getNextAt, now.plusSeconds(Math.min(3600, 30L * (1L << Math.min(attempts - 1, 7)))))
+			.set(FlowCallback::getLastError, success ? null : StrUtil.maxLength(detail, 500))
+			.set(FlowCallback::getLeaseToken, null)
+			.set(FlowCallback::getLeaseUntil, null)
+			.set(FlowCallback::getHistory, history.toString())
+			.set(FlowCallback::getUpdatedAt, now)
+			.eq(FlowCallback::getId, id)
+			.eq(FlowCallback::getLeaseToken, lease)
+			.update();
+	}
+
+}
@@ -0,0 +1,104 @@
+package com.pig4cloud.pigx.flow.service.impl;
+
+import cn.hutool.core.util.StrUtil;
+import cn.hutool.json.JSONUtil;
+import com.baomidou.mybatisplus.core.metadata.IPage;
+import com.baomidou.mybatisplus.core.toolkit.Wrappers;
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import com.pig4cloud.pigx.flow.dto.FlowDraftDTO;
+import com.pig4cloud.pigx.flow.entity.FlowDraft;
+import com.pig4cloud.pigx.flow.mapper.FlowDraftMapper;
+import com.pig4cloud.pigx.flow.service.IFlowDraftService;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+import java.nio.charset.StandardCharsets;
+import java.time.LocalDateTime;
+import java.util.Objects;
+
+/** 个人流程草稿存储；字段约束由 {@link FlowDraftDTO} 注解校验，这里只校验注解无法表达的内容大小与结构。 */
+@Service
+public class FlowDraftServiceImpl extends ServiceImpl<FlowDraftMapper, FlowDraft> implements IFlowDraftService {
+
+	/**
+	 * {@inheritDoc}
+	 */
+	@Override
+	@Transactional(rollbackFor = Exception.class)
+	public FlowDraft saveDraft(long tenant, long user, FlowDraftDTO draft) {
+		if (draft.getPayload().getBytes(StandardCharsets.UTF_8).length > 1000000
+				|| !JSONUtil.isTypeJSONObject(draft.getPayload())) {
+			throw new IllegalArgumentException("草稿内容必须是 JSON 对象，最多 1 MB");
+		}
+		JSONUtil.parseObj(draft.getPayload());
+		FlowDraft entity = new FlowDraft();
+		entity.setFlowId(draft.getFlowId());
+		entity.setTitle(draft.getTitle());
+		entity.setPayload(draft.getPayload());
+		entity.setUpdatedAt(LocalDateTime.now());
+		if (StrUtil.isBlank(draft.getId())) {
+			entity.setTenantId(tenant);
+			entity.setUserId(user);
+			entity.setKind(draft.getKind());
+			entity.setRevision(1L);
+			save(entity);
+			return getDraft(tenant, user, entity.getId());
+		}
+		// 乐观锁插件追加 revision 条件并递增版本号。
+		entity.setRevision(draft.getRevision());
+		if (!update(entity,
+				Wrappers.<FlowDraft>lambdaUpdate()
+					.eq(FlowDraft::getId, draft.getId())
+					.eq(FlowDraft::getTenantId, tenant)
+					.eq(FlowDraft::getUserId, user)
+					.eq(FlowDraft::getKind, draft.getKind()))) {
+			throw new IllegalStateException("草稿已被修改或删除，请重新打开后再保存");
+		}
+		return getDraft(tenant, user, draft.getId());
+	}
+
+	/**
+	 * {@inheritDoc}
+	 */
+	@Override
+	public FlowDraft getDraft(long tenant, long user, String id) {
+		FlowDraft draft = lambdaQuery().eq(FlowDraft::getId, id)
+			.eq(FlowDraft::getTenantId, tenant)
+			.eq(FlowDraft::getUserId, user)
+			.one();
+		if (Objects.isNull(draft)) {
+			throw new IllegalStateException("草稿不存在或无权访问");
+		}
+		return draft;
+	}
+
+	/**
+	 * {@inheritDoc}
+	 */
+	@Override
+	public IPage<FlowDraft> pageDraft(Page<FlowDraft> page, long tenant, long user, String kind, String flowId) {
+		return lambdaQuery()
+			.select(FlowDraft::getId, FlowDraft::getKind, FlowDraft::getFlowId, FlowDraft::getTitle,
+					FlowDraft::getRevision, FlowDraft::getUpdatedAt)
+			.eq(FlowDraft::getTenantId, tenant)
+			.eq(FlowDraft::getUserId, user)
+			.eq(FlowDraft::getKind, kind)
+			.eq(Objects.nonNull(flowId), FlowDraft::getFlowId, flowId)
+			.orderByDesc(FlowDraft::getUpdatedAt)
+			.orderByDesc(FlowDraft::getId)
+			.page(page);
+	}
+
+	/**
+	 * {@inheritDoc}
+	 */
+	@Override
+	public boolean removeDraft(long tenant, long user, String id, long revision) {
+		return remove(Wrappers.<FlowDraft>lambdaQuery()
+			.eq(FlowDraft::getId, id)
+			.eq(FlowDraft::getTenantId, tenant)
+			.eq(FlowDraft::getUserId, user)
+			.eq(FlowDraft::getRevision, revision));
+	}
+
+}
@@ -0,0 +1,283 @@
+package com.pig4cloud.pigx.flow.service.impl;
+
+import cn.hutool.core.date.DateUtil;
+import cn.hutool.core.util.EnumUtil;
+import cn.hutool.core.util.StrUtil;
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import com.pig4cloud.pigx.admin.api.dto.MessageNoticeDTO;
+import com.pig4cloud.pigx.admin.api.feign.RemoteMessageService;
+import com.pig4cloud.pigx.common.core.constant.enums.YesNoEnum;
+import com.pig4cloud.pigx.common.data.tenant.TenantBroker;
+import com.pig4cloud.pigx.flow.constant.FlowNoticeKindEnum;
+import com.pig4cloud.pigx.flow.constant.FlowNoticeStatusEnum;
+import com.pig4cloud.pigx.flow.constant.NodeStatusEnum;
+import com.pig4cloud.pigx.flow.constant.NodeTypeEnum;
+import com.pig4cloud.pigx.flow.constant.ProcessInstanceConstant;
+import com.pig4cloud.pigx.flow.dto.Node;
+import com.pig4cloud.pigx.flow.entity.FlowNotice;
+import com.pig4cloud.pigx.flow.entity.ProcessInstanceRecord;
+import com.pig4cloud.pigx.flow.mapper.FlowNoticeMapper;
+import com.pig4cloud.pigx.flow.service.IFlowCallbackService;
+import com.pig4cloud.pigx.flow.service.IFlowNoticeService;
+import com.pig4cloud.pigx.flow.service.IProcessInstanceRecordService;
+import com.pig4cloud.pigx.flow.service.IProcessNodeDataService;
+import lombok.RequiredArgsConstructor;
+import org.flowable.common.engine.api.delegate.event.FlowableEngineEventType;
+import org.flowable.engine.TaskService;
+import org.flowable.task.service.impl.persistence.entity.TaskEntity;
+import org.springframework.dao.DuplicateKeyException;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+import java.time.LocalDateTime;
+import java.util.List;
+import java.util.Objects;
+import java.util.UUID;
+
+/** 流程通知事件存储与投递；统一站内信由 upms 维护已读和删除状态。 */
+@Service
+@RequiredArgsConstructor
+public class FlowNoticeServiceImpl extends ServiceImpl<FlowNoticeMapper, FlowNotice> implements IFlowNoticeService {
+
+	private final IFlowCallbackService callbackService;
+
+	private final IProcessInstanceRecordService instanceRecordService;
+
+	private final IProcessNodeDataService nodeService;
+
+	private final TaskService taskService;
+
+	private final RemoteMessageService messageService;
+
+	/**
+	 * {@inheritDoc}
+	 */
+	@Override
+	public void record(TaskEntity task, String event) {
+		if (StrUtil.isBlank(task.getTenantId()) || StrUtil.isBlank(task.getProcessDefinitionId()))
+			return;
+		long tenant = Long.parseLong(task.getTenantId());
+		if (FlowableEngineEventType.ENTITY_DELETED.name().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 (FlowableEngineEventType.TASK_COMPLETED.name().equals(event)) {
+			cancelReminder(tenant, task.getId());
+			if (StrUtil.isNotBlank(node.getEventConfig()))
+				callbackService.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;
+		if (Objects.nonNull(node.getReminderMinutes()) && node.getReminderMinutes() > 0) {
+			LocalDateTime created = Objects.nonNull(task.getCreateTime())
+					? DateUtil.toLocalDateTime(task.getCreateTime()) : LocalDateTime.now();
+			FlowNotice reminder = notice(tenant, StrUtil.EMPTY,
+					FlowNoticeKindEnum.OVERDUE.getCode() + StrUtil.COLON + task.getId(), FlowNoticeKindEnum.OVERDUE,
+					task.getName(), task.getId(), task.getProcessInstanceId());
+			reminder.setStatus(FlowNoticeStatusEnum.PENDING.getCode());
+			reminder.setScheduledAt(created.plusMinutes(node.getReminderMinutes()));
+			reminder.setReminderKey(task.getId());
+			saveOnce(reminder);
+		}
+		if (StrUtil.isNotBlank(task.getAssignee())) {
+			// 每次分配可追踪，TASK_CREATED 和首次 TASK_ASSIGNED 按同一接收人去重。
+			saveOnce(notice(tenant, task.getAssignee(),
+					"ASSIGNED:" + task.getId() + StrUtil.COLON + task.getAssignee() + StrUtil.COLON
+							+ task.getRevision(),
+					FlowNoticeKindEnum.TODO, task.getName(), task.getId(), task.getProcessInstanceId()));
+		}
+	}
+
+	/** 构造立即投递的通知事件；超时提醒由调用方改为待到期状态。 */
+	private FlowNotice notice(long tenant, String user, String event, FlowNoticeKindEnum kind, String title,
+			String task, String instance) {
+		LocalDateTime now = LocalDateTime.now();
+		FlowNotice notice = new FlowNotice();
+		notice.setTenantId(tenant);
+		notice.setUserId(user);
+		notice.setEventKey(event);
+		notice.setKind(kind.getCode());
+		notice.setTitle(StrUtil.maxLength(StrUtil.blankToDefault(title, "流程任务"), 200));
+		notice.setTaskId(task);
+		notice.setInstanceId(instance);
+		notice.setCreatedAt(now);
+		notice.setScheduledAt(now);
+		notice.setStatus(FlowNoticeStatusEnum.READY.getCode());
+		return notice;
+	}
+
+	/** 唯一键负责去重；MySQL 唯一键冲突只回滚当前语句，不影响外层引擎事务。 */
+	private void saveOnce(FlowNotice notice) {
+		try {
+			save(notice);
+		}
+		catch (DuplicateKeyException ignored) {
+			// 同一事件已记录。
+		}
+	}
+
+	/**
+	 * {@inheritDoc}
+	 */
+	@Override
+	@Transactional(rollbackFor = Exception.class)
+	public int urge(long tenant, long user, String instance) {
+		ProcessInstanceRecord record = instanceRecordService.lambdaQuery()
+			.eq(ProcessInstanceRecord::getProcessInstanceId, instance)
+			.eq(ProcessInstanceRecord::getUserId, user)
+			.last("LIMIT 1 FOR UPDATE")
+			.one();
+		if (Objects.isNull(record))
+			throw new IllegalStateException("流程不存在或无权访问");
+		if (!Objects.equals(record.getStatus(), NodeStatusEnum.JXZ.getCode()))
+			throw new IllegalStateException("流程已结束，不能催办");
+		String flowName = StrUtil.blankToDefault(record.getName(), "流程申请");
+		var tasks = taskService.createTaskQuery()
+			.processInstanceId(instance)
+			.taskTenantId(String.valueOf(tenant))
+			.active()
+			.list();
+		if (tasks.isEmpty())
+			throw new IllegalStateException("当前没有待处理的审批，暂时无法催办");
+		if (lambdaQuery().eq(FlowNotice::getTenantId, tenant)
+			.eq(FlowNotice::getInstanceId, instance)
+			.eq(FlowNotice::getKind, FlowNoticeKindEnum.URGE.getCode())
+			.gt(FlowNotice::getCreatedAt, LocalDateTime.now().minusMinutes(10))
+			.exists())
+			throw new IllegalStateException("已发送催办，请 10 分钟后重试。");
+		int count = 0;
+		String event = UUID.randomUUID().toString();
+		for (var task : tasks) {
+			if (StrUtil.isBlank(task.getAssignee()))
+				continue;
+			// 建模时会在节点名称末尾追加五位随机数，通知展示时移除内部后缀。
+			String nodeName = StrUtil.blankToDefault(task.getName(), "审批").replaceFirst("\\[\\d{5}\\]$", StrUtil.EMPTY);
+			save(notice(tenant, task.getAssignee(),
+					FlowNoticeKindEnum.URGE.getCode() + StrUtil.COLON + event + StrUtil.COLON + task.getId(),
+					FlowNoticeKindEnum.URGE, flowName + " · " + nodeName, task.getId(), instance));
+			count++;
+		}
+		if (count == 0)
+			throw new IllegalStateException("当前待办尚未分配处理人，暂时无法催办");
+		return count;
+	}
+
+	/**
+	 * {@inheritDoc}
+	 */
+	@Override
+	public List<FlowNotice> listDueReminders() {
+		return lambdaQuery().select(FlowNotice::getTaskId, FlowNotice::getTenantId)
+			.eq(FlowNotice::getStatus, FlowNoticeStatusEnum.PENDING.getCode())
+			.le(FlowNotice::getScheduledAt, LocalDateTime.now())
+			.orderByAsc(FlowNotice::getScheduledAt)
+			.page(new Page<>(1, 100, false))
+			.getRecords();
+	}
+
+	/**
+	 * {@inheritDoc}
+	 */
+	@Override
+	@Transactional(rollbackFor = Exception.class)
+	public void remind(long tenant, String taskId) {
+		LocalDateTime now = LocalDateTime.now();
+		FlowNotice watch = lambdaQuery().eq(FlowNotice::getReminderKey, taskId)
+			.eq(FlowNotice::getTenantId, tenant)
+			.eq(FlowNotice::getStatus, FlowNoticeStatusEnum.PENDING.getCode())
+			.le(FlowNotice::getScheduledAt, now)
+			.last("FOR UPDATE")
+			.one();
+		if (Objects.isNull(watch))
+			return;
+		var task = taskService.createTaskQuery().taskId(taskId).taskTenantId(String.valueOf(tenant)).singleResult();
+		if (Objects.isNull(task)) {
+			cancelReminder(tenant, taskId);
+			return;
+		}
+		var update = lambdaUpdate().eq(FlowNotice::getId, watch.getId())
+			.eq(FlowNotice::getStatus, FlowNoticeStatusEnum.PENDING.getCode());
+		if (task.isSuspended() || StrUtil.isBlank(task.getAssignee())) {
+			update.set(FlowNotice::getScheduledAt, now.plusSeconds(10)).update();
+			return;
+		}
+		update.set(FlowNotice::getUserId, task.getAssignee())
+			.set(FlowNotice::getTitle, StrUtil.maxLength(StrUtil.blankToDefault(task.getName(), "流程任务"), 200))
+			.set(FlowNotice::getStatus, FlowNoticeStatusEnum.READY.getCode())
+			.set(FlowNotice::getScheduledAt, now)
+			.set(FlowNotice::getCreatedAt, now)
+			.update();
+	}
+
+	/**
+	 * {@inheritDoc}
+	 */
+	@Override
+	public List<FlowNotice> listDueNotices() {
+		return lambdaQuery().select(FlowNotice::getId)
+			.eq(FlowNotice::getStatus, FlowNoticeStatusEnum.READY.getCode())
+			.le(FlowNotice::getScheduledAt, LocalDateTime.now())
+			.orderByAsc(FlowNotice::getScheduledAt)
+			.page(new Page<>(1, 100, false))
+			.getRecords();
+	}
+
+	/**
+	 * {@inheritDoc}
+	 */
+	@Override
+	@Transactional(rollbackFor = Exception.class)
+	public void deliverNotice(String id) {
+		FlowNotice row = lambdaQuery().eq(FlowNotice::getId, id)
+			.eq(FlowNotice::getStatus, FlowNoticeStatusEnum.READY.getCode())
+			.last("FOR UPDATE")
+			.one();
+		if (Objects.isNull(row))
+			return;
+		MessageNoticeDTO notice = new MessageNoticeDTO();
+		// 站内信分类，upms 尚无对应常量。
+		notice.setCategory("1");
+		notice.setAllFlag(YesNoEnum.NO.getCode());
+		notice.setUserIds(List.of(Long.parseLong(row.getUserId())));
+		String label = EnumUtil.getBy(FlowNoticeKindEnum::getCode, row.getKind(), FlowNoticeKindEnum.TODO).getLabel();
+		notice.setTitle(label + StrUtil.COLON + row.getTitle());
+		notice.setContent("您有一条" + label + "消息，请打开对应审批任务处理。");
+		notice.setBusinessType("FLOW_TASK");
+		notice.setBusinessId(row.getTaskId());
+		notice.setRequestKey("flow:" + id);
+		var result = TenantBroker.applyAs(row.getTenantId(), t -> messageService.sendNotice(notice));
+		if (Objects.isNull(result) || result.getCode() != 0)
+			throw new IllegalStateException("流程站内信发送失败");
+		lambdaUpdate().set(FlowNotice::getStatus, FlowNoticeStatusEnum.SENT.getCode())
+			.set(FlowNotice::getSentAt, LocalDateTime.now())
+			.eq(FlowNotice::getId, id)
+			.update();
+	}
+
+	/**
+	 * {@inheritDoc}
+	 */
+	@Override
+	public void delayNotice(String id) {
+		lambdaUpdate().set(FlowNotice::getScheduledAt, LocalDateTime.now().plusSeconds(60))
+			.eq(FlowNotice::getId, id)
+			.eq(FlowNotice::getStatus, FlowNoticeStatusEnum.READY.getCode())
+			.update();
+	}
+
+	private void cancelReminder(long tenant, String task) {
+		lambdaUpdate().set(FlowNotice::getStatus, FlowNoticeStatusEnum.CANCELLED.getCode())
+			.eq(FlowNotice::getTenantId, tenant)
+			.eq(FlowNotice::getReminderKey, task)
+			.eq(FlowNotice::getStatus, FlowNoticeStatusEnum.PENDING.getCode())
+			.update();
+	}
+
+}
@@ -112,10 +112,8 @@ public class ProcessNodeDataServiceImpl extends ServiceImpl<ProcessNodeDataMappe
 		List<ProcessNodeData> nodes = new ArrayList<>();
 		collectNodes(definition, root, new HashSet<>(), nodes);
 		// 数据库唯一约束防止同一版本被重复写入，禁止删除或覆盖已有快照。
-		for (ProcessNodeData node : nodes) {
-			if (baseMapper.insert(node) != 1) {
-				throw new IllegalStateException("保存节点版本快照失败");
-			}
+		if (!saveBatch(nodes)) {
+			throw new IllegalStateException("保存节点版本快照失败");
 		}
 	}

@@ -337,8 +337,7 @@ public class ProcessServiceImpl extends ServiceImpl<ProcessMapper, Process> impl
 		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://"))
+		if (StrUtil.isNotBlank(node.getEventConfig()) && !NodeUtil.isCallbackUrl(node.getEventConfig()))
 			throw new IllegalArgumentException("节点回调地址必须使用 HTTP(S)");
 		if (java.util.Objects.nonNull(node.getConditionNodes()))
 			node.getConditionNodes().forEach(this::validateDeliveryConfig);
@@ -1,7 +1,8 @@
 package com.pig4cloud.pigx.flow.support.listeners;

-import com.pig4cloud.pigx.flow.support.workspace.FlowReliabilityService;
+import com.pig4cloud.pigx.flow.service.IFlowNoticeService;
 import lombok.RequiredArgsConstructor;
+import org.flowable.common.engine.api.delegate.event.FlowableEngineEventType;
 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;
@@ -13,10 +14,11 @@ 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 static final Set<String> EVENTS = Set.of(FlowableEngineEventType.TASK_CREATED.name(),
+			FlowableEngineEventType.TASK_ASSIGNED.name(), FlowableEngineEventType.TASK_COMPLETED.name(),
+			FlowableEngineEventType.ENTITY_DELETED.name());

-	private final ObjectProvider<FlowReliabilityService> service;
+	private final ObjectProvider<IFlowNoticeService> service;

 	/**
 	 * 记录目标任务事件。
@@ -10,7 +10,7 @@ import lombok.RequiredArgsConstructor;
 import org.springframework.stereotype.Service;

 /**
- * 流程实例状态通知服务。HTTP 节点回调由 FlowDurableEventListener 持久化， FlowCallbackDelivery 在提交后投递；此类保留现有
+ * 流程实例状态通知服务。HTTP 节点回调由 FlowDurableEventListener 持久化， IFlowCallbackService 在提交后投递；此类保留现有
 * Redis 状态发布行为。
 *
 * @author lengleng
@@ -1,6 +1,7 @@
 package com.pig4cloud.pigx.flow.support.utils;

 import cn.hutool.core.util.StrUtil;
+import cn.hutool.http.HttpUtil;
 import cn.hutool.json.JSONArray;
 import cn.hutool.json.JSONObject;
 import cn.hutool.json.JSONUtil;
@@ -93,6 +94,15 @@ public class NodeUtil {
 		return StrUtil.subBefore(processDefinitionId, ":", false);
 	}

+	/**
+	 * 判断节点完成回调地址是否使用 HTTP(S)；发布校验与运行期入队共用同一规则。
+	 * @param url 回调地址，可为 null
+	 * @return 以 http: 或 https: 开头（忽略大小写）时返回 true，null 或其他协议返回 false
+	 */
+	public static boolean isCallbackUrl(String url) {
+		return HttpUtil.isHttp(url) || HttpUtil.isHttps(url);
+	}
+
 	/**
 	 * 判断是否为有效节点
 	 * <p>
@@ -1,107 +0,0 @@
-package com.pig4cloud.pigx.flow.support.workspace;
-
-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);
-	}
-
-}
@@ -1,122 +0,0 @@
-package com.pig4cloud.pigx.flow.support.workspace;
-
-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 开始的页码
-	 * @param flowId 可选的流程 ID，为空时查询全部流程
-	 * @return 每页最多 50 个草稿
-	 */
-	public List<Map<String, Object>> list(long tenant, long user, String kind, int page, String flowId) {
-		return jdbc.queryForList(
-				"SELECT id,kind,flow_id,title,revision,updated_at FROM flow_draft WHERE tenant_id=? AND user_id=? AND kind=? AND (? IS NULL OR flow_id=?) ORDER BY updated_at DESC, id DESC LIMIT 50 OFFSET ?",
-				tenant, user, kind, flowId, flowId, Math.max(0L, (long) page - 1) * 50);
-	}
-
-	/**
-	 * 按版本删除自己的草稿，冲突时不删除。
-	 * @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){
-	}
-
-}
@@ -1,278 +0,0 @@
-package com.pig4cloud.pigx.flow.support.workspace;
-
-import com.pig4cloud.pigx.flow.service.IProcessNodeDataService;
-
-import cn.hutool.core.convert.Convert;
-import cn.hutool.core.util.StrUtil;
-import cn.hutool.json.JSONUtil;
-import com.pig4cloud.pigx.admin.api.dto.MessageNoticeDTO;
-import com.pig4cloud.pigx.admin.api.feign.RemoteMessageService;
-import com.pig4cloud.pigx.common.data.tenant.TenantBroker;
-import com.pig4cloud.pigx.flow.dto.Node;
-import com.pig4cloud.pigx.flow.constant.NodeStatusEnum;
-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;
-
-	private final RemoteMessageService messageService;
-
-	/**
-	 * 在引擎事务内记录任务事件；持久化失败必须回滚引擎操作。
-	 * @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,scheduled_at,status) VALUES(?,?,?,?,?,?,?,?,CURRENT_TIMESTAMP(3),CURRENT_TIMESTAMP(3),?) ON DUPLICATE KEY UPDATE event_key=event_key",
-				UUID.randomUUID().toString(), tenant, user, event, kind,
-				StrUtil.maxLength(StrUtil.blankToDefault(title, "流程任务"), 200), task, instance,
-				FlowNoticeStatusEnum.READY.getCode());
-	}
-
-	/**
-	 * 发起人催办；锁定实例实现跨任务、跨进程的每十分钟限频。
-	 * @param tenant 当前租户
-	 * @param user 当前用户
-	 * @param instance 流程实例 ID
-	 * @return 本次接收催办的任务数
-	 * @throws IllegalStateException 非发起人、已结束、无待办或催办过频
-	 */
-	@Transactional
-	public int urge(long tenant, long user, String instance) {
-		String flowName = requireStarter(tenant, user, instance, true);
-		var tasks = taskService.createTaskQuery()
-			.processInstanceId(instance)
-			.taskTenantId(String.valueOf(tenant))
-			.active()
-			.list();
-		if (tasks.isEmpty())
-			throw new IllegalStateException("当前没有待处理的审批，暂时无法催办");
-		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;
-			// 建模时会在节点名称末尾追加五位随机数，通知展示时移除内部后缀。
-			String nodeName = StrUtil.blankToDefault(task.getName(), "审批").replaceFirst("\\[\\d{5}\\]$", StrUtil.EMPTY);
-			notice(tenant, task.getAssignee(), "URGE:" + event + StrUtil.COLON + task.getId(), "URGE",
-					flowName + " · " + nodeName,
-					task.getId(), instance);
-			count++;
-		}
-		if (count == 0)
-			throw new IllegalStateException("当前待办尚未分配处理人，暂时无法催办");
-		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=?,scheduled_at=CURRENT_TIMESTAMP(3),created_at=CURRENT_TIMESTAMP(3) WHERE id=? AND tenant_id=? AND status=?",
-				task.getAssignee(), StrUtil.maxLength(StrUtil.blankToDefault(task.getName(), "流程任务"), 200),
-				FlowNoticeStatusEnum.READY.getCode(), watches.get(0).get("id"), tenant,
-				FlowNoticeStatusEnum.PENDING.getCode());
-	}
-
-	/**
-	 * 投递已提交的流程事件到统一站内信；行锁防止扫描器并发发送同一事件。
-	 * @param id 流程通知事件 ID
-	 */
-	@Transactional
-	public void deliverNotice(String id) {
-		var rows = jdbc.queryForList("SELECT * FROM flow_notice WHERE id=? AND status=? FOR UPDATE", id,
-				FlowNoticeStatusEnum.READY.getCode());
-		if (rows.isEmpty())
-			return;
-		var row = rows.get(0);
-		long tenant = ((Number) row.get("tenant_id")).longValue();
-		MessageNoticeDTO notice = new MessageNoticeDTO();
-		notice.setCategory("1");
-		notice.setAllFlag("0");
-		notice.setUserIds(List.of(Long.parseLong(String.valueOf(row.get("user_id")))));
-		String label = switch (String.valueOf(row.get("kind"))) {
-			case "URGE" -> "催办";
-			case "OVERDUE" -> "超时提醒";
-			default -> "待办";
-		};
-		notice.setTitle(label + StrUtil.COLON + row.get("title"));
-		notice.setContent("您有一条" + label + "消息，请打开对应审批任务处理。");
-		notice.setBusinessType("FLOW_TASK");
-		notice.setBusinessId(String.valueOf(row.get("task_id")));
-		notice.setRequestKey("flow:" + id);
-		var result = TenantBroker.applyAs(tenant, t -> messageService.sendNotice(notice));
-		if (result == null || result.getCode() != 0)
-			throw new IllegalStateException("流程站内信发送失败");
-		jdbc.update("UPDATE flow_notice SET status=?,sent_at=CURRENT_TIMESTAMP(3) WHERE id=? AND tenant_id=?",
-				FlowNoticeStatusEnum.SENT.getCode(), id, tenant);
-	}
-
-	/**
-	 * 查询自己发起流程的回调摘要；不向页面返回表单载荷和目标地址。
-	 * @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 String requireStarter(long tenant, long user, String instance, boolean running) {
-		var rows = jdbc.queryForList(
-				"SELECT status,name FROM process_instance_record WHERE process_instance_id=? AND user_id=? AND tenant_id=? AND del_flag='0' LIMIT 1"
-						+ (running ? " FOR UPDATE" : StrUtil.EMPTY),
-				instance, user, tenant);
-		if (rows.isEmpty())
-			throw new IllegalStateException("流程不存在或无权访问");
-		if (running && !Objects.equals(Convert.toInt(rows.get(0).get("status")), NodeStatusEnum.JXZ.getCode()))
-			throw new IllegalStateException("流程已结束，不能催办");
-		return StrUtil.blankToDefault(Convert.toStr(rows.get(0).get("name")), "流程申请");
-	}
-
-}
@@ -157,6 +157,12 @@ public class SysMessageController {
        return R.ok(sysMessageService.readMessage(id));
    }

+    /**
+     * 删除当前用户的消息，仅移除本人的接收记录，不需要菜单权限。
+     *
+     * @param id 消息 ID
+     * @return 始终返回 true；记录不存在或不属于当前用户时不删除任何数据
+     */
    @Operation(summary = "删除当前用户消息", description = "仅删除当前用户的接收记录")
    @SysLog("删除当前用户消息")
    @DeleteMapping("/user/{id}")
@@ -44,7 +44,12 @@ public interface SysMessageService extends IService<SysMessageEntity> {
     */
    Boolean readMessage(Long id);

-    /** 删除当前用户的消息接收记录。 */
+    /**
+     * 删除当前用户的消息接收记录，不影响消息本身及其他接收人。
+     *
+     * @param id 消息 ID
+     * @return 始终返回 true；记录不存在或不属于当前用户时不删除任何数据
+     */
    Boolean removeUserMessage(Long id);

    /**
@@ -217,6 +217,12 @@ public class SysMessageServiceImpl extends ServiceImpl<SysMessageMapper, SysMess
        return Boolean.TRUE;
    }

+    /**
+     * 删除当前用户的消息接收记录，不影响消息本身及其他接收人。
+     *
+     * @param id 消息 ID
+     * @return 始终返回 true；记录不存在或不属于当前用户时不删除任何数据
+     */
    @Override
    public Boolean removeUserMessage(Long id) {
        relationMapper.delete(Wrappers.<SysMessageRelationEntity>lambdaQuery()
@@ -611,7 +617,7 @@ public class SysMessageServiceImpl extends ServiceImpl<SysMessageMapper, SysMess
        if (StrUtil.isNotBlank(noticeDTO.getRequestKey())) {
            SysMessageEntity existing = baseMapper.selectOne(Wrappers.<SysMessageEntity>lambdaQuery()
                    .eq(SysMessageEntity::getRequestKey, noticeDTO.getRequestKey()));
-            if (existing != null) {
+            if (Objects.nonNull(existing)) {
                return R.ok(existing.getId());
            }
        }