diff --git a/db/0pigxx_boot.sql b/db/0pigxx_boot.sql index 019602827..769f89834 100644 --- a/db/0pigxx_boot.sql +++ b/db/0pigxx_boot.sql @@ -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; diff --git a/db/3pigxx_flow.sql b/db/3pigxx_flow.sql index 48cca9109..7840937af 100644 --- a/db/3pigxx_flow.sql +++ b/db/3pigxx_flow.sql @@ -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; diff --git a/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/constant/FlowCallbackStatusEnum.java b/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/constant/FlowCallbackStatusEnum.java index db029a00c..e71ed1d25 100644 --- a/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/constant/FlowCallbackStatusEnum.java +++ b/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/constant/FlowCallbackStatusEnum.java @@ -14,7 +14,7 @@ public enum FlowCallbackStatusEnum { SENDING("SENDING"), /** 接收方返回 HTTP 2xx，投递结束。 */ DELIVERED("DELIVERED"), - /** 尝试耗尽或最终租约过期，允许发起人手动重试。 */ + /** 尝试耗尽或最终租约过期，需人工核对接收方后处理。 */ FAILED("FAILED"); /** 持久化编码，不能随枚举重命名改变。 */ diff --git a/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/constant/FlowNoticeKindEnum.java b/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/constant/FlowNoticeKindEnum.java new file mode 100644 index 000000000..f951c54cf --- /dev/null +++ b/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/constant/FlowNoticeKindEnum.java @@ -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; + +} diff --git a/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/dto/FlowDraftDTO.java b/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/dto/FlowDraftDTO.java new file mode 100644 index 000000000..b314ce87b --- /dev/null +++ b/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/dto/FlowDraftDTO.java @@ -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; + +} diff --git a/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/entity/FlowCallback.java b/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/entity/FlowCallback.java new file mode 100644 index 000000000..e723c49eb --- /dev/null +++ b/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/entity/FlowCallback.java @@ -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; + +} diff --git a/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/entity/FlowDraft.java b/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/entity/FlowDraft.java new file mode 100644 index 000000000..0263e05ed --- /dev/null +++ b/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/entity/FlowDraft.java @@ -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; + +} diff --git a/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/entity/FlowNotice.java b/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/entity/FlowNotice.java new file mode 100644 index 000000000..aafa17a82 --- /dev/null +++ b/pigx-flow/pigx-flow-api/src/main/java/com/pig4cloud/pigx/flow/entity/FlowNotice.java @@ -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; + +} diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/config/FlowDurableConfiguration.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/config/FlowDurableConfiguration.java index af3b56c3c..501b3bf99 100644 --- a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/config/FlowDurableConfiguration.java +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/config/FlowDurableConfiguration.java @@ -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 noticeService; - private final ObjectProvider reliability; - - private final ObjectProvider delivery; + private final ObjectProvider callbackService; /** * 注册持久化监听器，保留引擎已配置的监听器。 @@ -43,7 +38,7 @@ public class FlowDurableConfiguration { var listeners = new ArrayList(); 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=max_attempts", - FlowCallbackStatusEnum.FAILED.getCode(), FlowCallbackStatusEnum.SENDING.getCode()); - var callbacks = jdbc.queryForList( - "SELECT id FROM flow_callback WHERE (status=? AND next_at<=CURRENT_TIMESTAMP(3)) OR (status=? AND lease_until> save(@RequestBody FlowDraftService.Draft draft) { - return R.ok(drafts.save(tenant(), user(), draft)); + public R 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(@RequestParam String kind, @RequestParam(defaultValue = "1") int page, + public R> page(@ParameterObject Page 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> get(@PathVariable String id) { - return R.ok(drafts.get(tenant(), user(), id)); + public R 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 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 urge(@PathVariable String id) { - return R.ok(reliability.urge(tenant(), user(), id)); - } - - /** - * 查看自己发起实例的回调。 - * @param id 实例 ID - * @return 最多 100 条回调摘要，不含表单载荷与目标地址 - * @throws IllegalStateException 实例不存在或不属于当前发起人 - */ - @Operation(summary = "查询本人发起实例的回调") - @GetMapping("/instances/{id}/callbacks") - public R>> callbacks(@PathVariable String id) { - return R.ok(reliability.callbacks(tenant(), user(), id)); - } - - /** - * 重试失败回调；保留事件 ID、载荷和历史，追加最多五次投递机会。 - * @param id 回调 ID - * @return 操作结果 - * @throws IllegalStateException 回调不存在、非实例发起人或回调并非失败状态 - */ - @Operation(summary = "重试本人发起实例的失败回调") - @SysLog("重试本人发起实例的失败回调") - @PostMapping("/callbacks/{id}/retry") - public R retry(@PathVariable String id) { - reliability.retry(tenant(), user(), id); - return R.ok(); + return R.ok(noticeService.urge(tenant(), user(), id)); } } diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/mapper/FlowCallbackMapper.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/mapper/FlowCallbackMapper.java new file mode 100644 index 000000000..9b3e22951 --- /dev/null +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/mapper/FlowCallbackMapper.java @@ -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 { + +} diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/mapper/FlowDraftMapper.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/mapper/FlowDraftMapper.java new file mode 100644 index 000000000..23d0b1747 --- /dev/null +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/mapper/FlowDraftMapper.java @@ -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 { + +} diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/mapper/FlowNoticeMapper.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/mapper/FlowNoticeMapper.java new file mode 100644 index 000000000..13773ef98 --- /dev/null +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/mapper/FlowNoticeMapper.java @@ -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 { + +} diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/IFlowCallbackService.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/IFlowCallbackService.java new file mode 100644 index 000000000..e8dc9b0a3 --- /dev/null +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/IFlowCallbackService.java @@ -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 { + + /** + * 在引擎事务内为已完成任务入队回调；同一任务只入队一次。 + * @param task 已完成的引擎任务 + * @param tenant 任务所属租户 + * @param target 节点配置的 HTTP(S) 回调地址 + * @throws IllegalArgumentException 回调地址不是 HTTP(S) + */ + void enqueue(TaskEntity task, long tenant, String target); + + /** + * 将最后一次尝试租约已过期的回调置为失败，并返回最多 20 条到期回调。 + * @return 到期或租约过期的回调，只包含 ID + */ + List listDue(); + + /** + * 尝试投递一条到期回调，不在网络请求期间持有数据库事务。 + * @param id 扫描得到的回调 ID + */ + void deliver(String id); + +} diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/IFlowDraftService.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/IFlowDraftService.java new file mode 100644 index 000000000..937740891 --- /dev/null +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/IFlowDraftService.java @@ -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 { + + /** + * 保存草稿，使用乐观锁版本号防止多窗口覆盖。 + * @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 pageDraft(Page 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); + +} diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/IFlowNoticeService.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/IFlowNoticeService.java new file mode 100644 index 000000000..00c13031b --- /dev/null +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/IFlowNoticeService.java @@ -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 { + + /** + * 在引擎事务内记录任务事件；持久化失败必须回滚引擎操作。 + * @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 listDueReminders(); + + /** + * 为到期任务记录一次超时提醒；只使用任务当前接收人，暂停或暂无接收人时十秒后再检查。 + * @param tenant 任务所属租户 + * @param taskId 到期任务 ID + */ + void remind(long tenant, String taskId); + + /** + * 查询最多 100 条待投递的通知事件。 + * @return 只包含 ID 的通知事件 + */ + List listDueNotices(); + + /** + * 投递通知事件到统一站内信；行锁防止扫描器并发发送同一事件。 + * @param id 通知事件 ID + * @throws IllegalStateException 站内信服务返回失败，由扫描器延迟重试 + */ + void deliverNotice(String id); + + /** + * 投递失败后延迟一分钟再试。 + * @param id 通知事件 ID + */ + void delayNotice(String id); + +} diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/FlowCallbackServiceImpl.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/FlowCallbackServiceImpl.java new file mode 100644 index 000000000..366004128 --- /dev/null +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/FlowCallbackServiceImpl.java @@ -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 + 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 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(); + } + +} diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/FlowDraftServiceImpl.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/FlowDraftServiceImpl.java new file mode 100644 index 000000000..279e534a3 --- /dev/null +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/FlowDraftServiceImpl.java @@ -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 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.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 pageDraft(Page 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.lambdaQuery() + .eq(FlowDraft::getId, id) + .eq(FlowDraft::getTenantId, tenant) + .eq(FlowDraft::getUserId, user) + .eq(FlowDraft::getRevision, revision)); + } + +} diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/FlowNoticeServiceImpl.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/FlowNoticeServiceImpl.java new file mode 100644 index 000000000..585923e5e --- /dev/null +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/FlowNoticeServiceImpl.java @@ -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 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 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 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(); + } + +} diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/ProcessNodeDataServiceImpl.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/ProcessNodeDataServiceImpl.java index 15ebb51e4..22794c22d 100644 --- a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/ProcessNodeDataServiceImpl.java +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/ProcessNodeDataServiceImpl.java @@ -112,10 +112,8 @@ public class ProcessNodeDataServiceImpl extends ServiceImpl 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("保存节点版本快照失败"); } } diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/ProcessServiceImpl.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/ProcessServiceImpl.java index 904306eba..0e018a76a 100644 --- a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/ProcessServiceImpl.java +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/service/impl/ProcessServiceImpl.java @@ -337,8 +337,7 @@ public class ProcessServiceImpl extends ServiceImpl 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); diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/listeners/FlowDurableEventListener.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/listeners/FlowDurableEventListener.java index 275e705c4..dfd923324 100644 --- a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/listeners/FlowDurableEventListener.java +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/listeners/FlowDurableEventListener.java @@ -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 EVENTS = Set.of("TASK_CREATED", "TASK_ASSIGNED", "TASK_COMPLETED", - "ENTITY_DELETED"); + private static final Set EVENTS = Set.of(FlowableEngineEventType.TASK_CREATED.name(), + FlowableEngineEventType.TASK_ASSIGNED.name(), FlowableEngineEventType.TASK_COMPLETED.name(), + FlowableEngineEventType.ENTITY_DELETED.name()); - private final ObjectProvider service; + private final ObjectProvider service; /** * 记录目标任务事件。 diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/listeners/NodeTaskCompleteNotify.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/listeners/NodeTaskCompleteNotify.java index baa281274..3838da4e2 100644 --- a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/listeners/NodeTaskCompleteNotify.java +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/listeners/NodeTaskCompleteNotify.java @@ -10,7 +10,7 @@ import lombok.RequiredArgsConstructor; import org.springframework.stereotype.Service; /** - * 流程实例状态通知服务。HTTP 节点回调由 FlowDurableEventListener 持久化， FlowCallbackDelivery 在提交后投递；此类保留现有 + * 流程实例状态通知服务。HTTP 节点回调由 FlowDurableEventListener 持久化， IFlowCallbackService 在提交后投递；此类保留现有 * Redis 状态发布行为。 * * @author lengleng diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/utils/NodeUtil.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/utils/NodeUtil.java index 750707fc9..8cd88cd73 100644 --- a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/utils/NodeUtil.java +++ b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/utils/NodeUtil.java @@ -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); + } + /** * 判断是否为有效节点 *

diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/workspace/FlowCallbackDelivery.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/workspace/FlowCallbackDelivery.java deleted file mode 100644 index a095f45e0..000000000 --- a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/workspace/FlowCallbackDelivery.java +++ /dev/null @@ -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= 200 && response.getStatus() < 300; - detail = "HTTP " + response.getStatus(); - } - catch (Exception exception) { - // 只记录异常类型，避免响应正文、URL 参数或表单数据进入日志。 - detail = exception.getClass().getSimpleName(); - } - int attempts = ((Number) record.get("attempts")).intValue(); - int maximum = ((Number) record.get("max_attempts")).intValue(); - FlowCallbackStatusEnum state = success ? FlowCallbackStatusEnum.DELIVERED - : attempts >= maximum ? FlowCallbackStatusEnum.FAILED : FlowCallbackStatusEnum.PENDING; - String result = success ? "DELIVERED" : "ERROR"; - String error = success ? null : StrUtil.maxLength(detail, 500); - String auditDetail = detail; - transactions.executeWithoutResult(tx -> { - int changed = jdbc.update( - "UPDATE flow_callback SET status=?,next_at=?,last_error=?,lease_token=NULL,lease_until=NULL,updated_at=CURRENT_TIMESTAMP(3) WHERE id=? AND lease_token=?", - state.getCode(), - Timestamp.from(Instant.now().plusSeconds(Math.min(3600, 30L * (1L << Math.min(attempts - 1, 7))))), - error, id, lease); - if (changed == 1) { - JSONArray history = history(jdbc, id); - JSONObject attempt = history.stream() - .map(value -> (JSONObject) value) - .filter(value -> lease.equals(value.getStr("id"))) - .findFirst() - .orElseThrow(() -> new IllegalStateException("回调尝试记录缺失")); - attempt.set("result", result); - attempt.set("detail", auditDetail); - jdbc.update("UPDATE flow_callback SET history=? WHERE id=?", history.toString(), id); - } - }); - } - - /** 调用方必须在同一事务中先锁定回调行；历史不截断，避免手动重试覆盖旧记录。 */ - static void appendHistory(JdbcTemplate jdbc, String callback, String entry, int number, String result, - String detail) { - JSONArray history = history(jdbc, callback); - history.add(new JSONObject().set("id", entry) - .set("attempt_no", number) - .set("result", result) - .set("detail", detail) - .set("created_at", Instant.now().toString())); - jdbc.update("UPDATE flow_callback SET history=? WHERE id=?", history.toString(), callback); - } - - private static JSONArray history(JdbcTemplate jdbc, String callback) { - String value = jdbc.queryForObject("SELECT history FROM flow_callback WHERE id=?", String.class, callback); - return StrUtil.isBlank(value) ? new JSONArray() : JSONUtil.parseArray(value); - } - -} diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/workspace/FlowDraftService.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/workspace/FlowDraftService.java deleted file mode 100644 index 4bb610b66..000000000 --- a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/workspace/FlowDraftService.java +++ /dev/null @@ -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 save(long tenant, long user, Draft draft) { - if (Objects.isNull(draft.kind()) || !List.of("DESIGN", "APPLICATION").contains(draft.kind()) - || StrUtil.isBlank(draft.payload()) || draft.payload().getBytes(StandardCharsets.UTF_8).length > 1000000 - || !JSONUtil.isTypeJSONObject(draft.payload()) || StrUtil.isBlank(draft.title()) - || draft.title().length() > 200 || Objects.isNull(draft.flowId()) || draft.flowId().length() > 64) { - throw new IllegalArgumentException("草稿类型、标题或内容无效（内容最多 1 MB）"); - } - JSONUtil.parseObj(draft.payload()); - String id = draft.id(); - if (StrUtil.isBlank(id)) { - id = UUID.randomUUID().toString(); - jdbc.update( - "INSERT INTO flow_draft(id,tenant_id,user_id,kind,flow_id,title,payload,revision,updated_at) VALUES(?,?,?,?,?,?,?,1,CURRENT_TIMESTAMP(3))", - id, tenant, user, draft.kind(), draft.flowId(), draft.title(), draft.payload()); - } - else if (jdbc.update( - "UPDATE flow_draft SET title=?,payload=?,flow_id=?,revision=revision+1,updated_at=CURRENT_TIMESTAMP(3) WHERE id=? AND tenant_id=? AND user_id=? AND kind=? AND revision=?", - draft.title(), draft.payload(), draft.flowId(), id, tenant, user, draft.kind(), - draft.revision()) != 1) { - throw new IllegalStateException("草稿已被修改或删除，请重新打开后再保存"); - } - return get(tenant, user, id); - } - - /** - * 读取当前用户的草稿。 - * @param tenant 当前租户 - * @param user 当前用户 - * @param id 草稿 ID - * @return 草稿详情 - * @throws IllegalStateException 草稿不存在或不属于当前用户 - */ - public Map get(long tenant, long user, String id) { - var rows = jdbc.queryForList("SELECT * FROM flow_draft WHERE id=? AND tenant_id=? AND user_id=?", id, tenant, - user); - if (rows.isEmpty()) - throw new IllegalStateException("草稿不存在或无权访问"); - return rows.get(0); - } - - /** - * 分页列出个人草稿摘要，不返回表单数据。 - * @param tenant 当前租户 - * @param user 当前用户 - * @param kind DESIGN 或 APPLICATION - * @param page 从 1 开始的页码 - * @param flowId 可选的流程 ID，为空时查询全部流程 - * @return 每页最多 50 个草稿 - */ - public List> list(long tenant, long user, String kind, int page, String flowId) { - return jdbc.queryForList( - "SELECT id,kind,flow_id,title,revision,updated_at FROM flow_draft WHERE tenant_id=? AND user_id=? AND kind=? 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){ - } - -} diff --git a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/workspace/FlowReliabilityService.java b/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/workspace/FlowReliabilityService.java deleted file mode 100644 index 2a3064517..000000000 --- a/pigx-flow/pigx-flow-biz/src/main/java/com/pig4cloud/pigx/flow/support/workspace/FlowReliabilityService.java +++ /dev/null @@ -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> 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")), "流程申请"); - } - -} diff --git a/pigx-upms/pigx-upms-biz/src/main/java/com/pig4cloud/pigx/admin/controller/SysMessageController.java b/pigx-upms/pigx-upms-biz/src/main/java/com/pig4cloud/pigx/admin/controller/SysMessageController.java index edbfd2858..21895b275 100644 --- a/pigx-upms/pigx-upms-biz/src/main/java/com/pig4cloud/pigx/admin/controller/SysMessageController.java +++ b/pigx-upms/pigx-upms-biz/src/main/java/com/pig4cloud/pigx/admin/controller/SysMessageController.java @@ -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}") diff --git a/pigx-upms/pigx-upms-biz/src/main/java/com/pig4cloud/pigx/admin/service/SysMessageService.java b/pigx-upms/pigx-upms-biz/src/main/java/com/pig4cloud/pigx/admin/service/SysMessageService.java index 5b9c79de2..5074696d8 100644 --- a/pigx-upms/pigx-upms-biz/src/main/java/com/pig4cloud/pigx/admin/service/SysMessageService.java +++ b/pigx-upms/pigx-upms-biz/src/main/java/com/pig4cloud/pigx/admin/service/SysMessageService.java @@ -44,7 +44,12 @@ public interface SysMessageService extends IService { */ Boolean readMessage(Long id); - /** 删除当前用户的消息接收记录。 */ + /** + * 删除当前用户的消息接收记录，不影响消息本身及其他接收人。 + * + * @param id 消息 ID + * @return 始终返回 true；记录不存在或不属于当前用户时不删除任何数据 + */ Boolean removeUserMessage(Long id); /** diff --git a/pigx-upms/pigx-upms-biz/src/main/java/com/pig4cloud/pigx/admin/service/impl/SysMessageServiceImpl.java b/pigx-upms/pigx-upms-biz/src/main/java/com/pig4cloud/pigx/admin/service/impl/SysMessageServiceImpl.java index af4ba0555..a3aea3667 100644 --- a/pigx-upms/pigx-upms-biz/src/main/java/com/pig4cloud/pigx/admin/service/impl/SysMessageServiceImpl.java +++ b/pigx-upms/pigx-upms-biz/src/main/java/com/pig4cloud/pigx/admin/service/impl/SysMessageServiceImpl.java @@ -217,6 +217,12 @@ public class SysMessageServiceImpl extends ServiceImpllambdaQuery() @@ -611,7 +617,7 @@ public class SysMessageServiceImpl extends ServiceImpllambdaQuery() .eq(SysMessageEntity::getRequestKey, noticeDTO.getRequestKey())); - if (existing != null) { + if (Objects.nonNull(existing)) { return R.ok(existing.getId()); } }