|
|
@@ -1,207 +1,207 @@
|
|
|
-package com.fs.company.service.workflow.scheduler;
|
|
|
-
|
|
|
-import com.fs.company.domain.CompanyWorkflowLobster;
|
|
|
-import com.fs.company.mapper.CompanyWorkflowLobsterMapper;
|
|
|
-import com.fs.company.service.workflow.LobsterWorkflowExecutor;
|
|
|
-import org.slf4j.Logger;
|
|
|
-import org.slf4j.LoggerFactory;
|
|
|
-import org.springframework.beans.factory.annotation.Autowired;
|
|
|
-import org.springframework.scheduling.annotation.Scheduled;
|
|
|
-import org.springframework.stereotype.Component;
|
|
|
-
|
|
|
-import javax.annotation.PostConstruct;
|
|
|
-import java.time.LocalDateTime;
|
|
|
-import java.time.LocalTime;
|
|
|
-import java.util.HashMap;
|
|
|
-import java.util.List;
|
|
|
-import java.util.Map;
|
|
|
-import java.util.concurrent.ConcurrentHashMap;
|
|
|
-
|
|
|
-/**
|
|
|
- * 工作流触发调度器（skill.md 要求）
|
|
|
- * <p>
|
|
|
- * 三种触发机制：
|
|
|
- * 1. 时间触发 — Cron / 固定时间 → 扫描启用的工作流并触发
|
|
|
- * 2. 事件触发 — 客户消息到达时由消息处理器主动调用 fire(eventType)
|
|
|
- * 3. 条件触发 — 定时巡检触发条件（活跃天数/沉默小时数等）
|
|
|
- * <p>
|
|
|
- * 实现：使用 Spring @Scheduled（每分钟检查一次），由 @EnableScheduling 启用
|
|
|
- */
|
|
|
-@Component
|
|
|
-public class WorkflowTriggerScheduler {
|
|
|
-
|
|
|
- private static final Logger log = LoggerFactory.getLogger(WorkflowTriggerScheduler.class);
|
|
|
-
|
|
|
- @Autowired(required = false)
|
|
|
- private CompanyWorkflowLobsterMapper workflowMapper;
|
|
|
-
|
|
|
- @Autowired(required = false)
|
|
|
- private LobsterWorkflowExecutor workflowExecutor;
|
|
|
-
|
|
|
- @Autowired(required = false)
|
|
|
- private com.fs.company.service.workflow.LobsterTestScenarioService testScenarioService;
|
|
|
-
|
|
|
- /** 最近一次触发时间（防重复） — workflowId → lastFireTime */
|
|
|
- private final ConcurrentHashMap<Long, LocalDateTime> lastFireTime = new ConcurrentHashMap<>();
|
|
|
-
|
|
|
- @PostConstruct
|
|
|
- public void init() {
|
|
|
- log.info("[WorkflowTriggerScheduler] 启动，每分钟扫描一次工作流触发");
|
|
|
- }
|
|
|
-
|
|
|
- /**
|
|
|
- * 每天凌晨 3 点跑全部启用的 E2E 测试场景（回归测试）
|
|
|
- */
|
|
|
- @Scheduled(cron = "0 0 3 * * ?")
|
|
|
- public void runDailyE2eRegression() {
|
|
|
- if (testScenarioService == null) return;
|
|
|
- try {
|
|
|
- int n = testScenarioService.runAllEnabledScenarios();
|
|
|
- log.info("[E2E回归] 已触发 {} 个测试场景", n);
|
|
|
- } catch (Exception ex) {
|
|
|
- log.error("[E2E回归] 失败: {}", ex.getMessage(), ex);
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- /**
|
|
|
- * 每分钟扫描一次启用状态的工作流，按其触发配置决定是否触发
|
|
|
- */
|
|
|
- @Scheduled(cron = "0 * * * * ?")
|
|
|
- public void scanAndTrigger() {
|
|
|
- if (workflowMapper == null) return;
|
|
|
- try {
|
|
|
- CompanyWorkflowLobster q = new CompanyWorkflowLobster();
|
|
|
- q.setStatus(1); // 仅启用
|
|
|
- List<CompanyWorkflowLobster> list = workflowMapper.selectList(new com.baomidou.mybatisplus.core.conditions.query.QueryWrapper<>(q));
|
|
|
- if (list == null || list.isEmpty()) return;
|
|
|
-
|
|
|
- LocalDateTime now = LocalDateTime.now();
|
|
|
- int triggered = 0;
|
|
|
- for (CompanyWorkflowLobster wf : list) {
|
|
|
- if (shouldTrigger(wf, now)) {
|
|
|
- fire(wf, now);
|
|
|
- triggered++;
|
|
|
- }
|
|
|
- }
|
|
|
- if (triggered > 0) {
|
|
|
- log.info("[WorkflowTriggerScheduler] 本轮触发 {} 个工作流", triggered);
|
|
|
- }
|
|
|
- } catch (Exception e) {
|
|
|
- log.error("[WorkflowTriggerScheduler] 扫描异常: {}", e.getMessage(), e);
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- /** 事件触发入口（消息到达/订单创建等场景，由业务方主动调用） */
|
|
|
- public void fireByEvent(String eventType, Long companyId, Object payload) {
|
|
|
- log.info("[WorkflowTriggerScheduler] 事件触发 type={} company={} payload={}", eventType, companyId, payload);
|
|
|
- if (workflowMapper == null || workflowExecutor == null) return;
|
|
|
- try {
|
|
|
- CompanyWorkflowLobster q = new CompanyWorkflowLobster();
|
|
|
- q.setCompanyId(companyId);
|
|
|
- q.setStatus(1);
|
|
|
- List<CompanyWorkflowLobster> list = workflowMapper.selectList(new com.baomidou.mybatisplus.core.conditions.query.QueryWrapper<>(q));
|
|
|
- if (list == null) return;
|
|
|
- for (CompanyWorkflowLobster wf : list) {
|
|
|
- String evtCfg = parseEventTypeFromCanvas(wf.getCanvasData());
|
|
|
- if (eventType != null && eventType.equalsIgnoreCase(evtCfg)) {
|
|
|
- Map<String, Object> vars = new HashMap<>();
|
|
|
- vars.put("eventType", eventType);
|
|
|
- if (payload != null) vars.put("payload", payload);
|
|
|
- Long contactId = payload instanceof Map ?
|
|
|
- (Long) ((Map<?, ?>) payload).get("contactId") : null;
|
|
|
- try {
|
|
|
- workflowExecutor.startWorkflow(companyId, wf.getId(), contactId, vars);
|
|
|
- log.info("[WorkflowTriggerScheduler] 事件 {} 启动工作流 id={}", eventType, wf.getId());
|
|
|
- } catch (Exception ex) {
|
|
|
- log.warn("[WorkflowTriggerScheduler] 启动工作流 {} 失败: {}", wf.getId(), ex.getMessage());
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
- } catch (Exception e) {
|
|
|
- log.error("[WorkflowTriggerScheduler] 事件触发异常: {}", e.getMessage(), e);
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- // ════════════ 私有方法 ════════════
|
|
|
-
|
|
|
- /**
|
|
|
- * 判断工作流是否应在当前时刻触发
|
|
|
- * 规则：
|
|
|
- * - 优先解析 canvasData 中的 startNode.config.fireTime（HH:mm）
|
|
|
- * - 同一工作流 1 分钟内不重复触发
|
|
|
- */
|
|
|
- private boolean shouldTrigger(CompanyWorkflowLobster wf, LocalDateTime now) {
|
|
|
- if (wf == null || wf.getId() == null) return false;
|
|
|
- LocalDateTime last = lastFireTime.get(wf.getId());
|
|
|
- if (last != null && java.time.Duration.between(last, now).getSeconds() < 60) {
|
|
|
- return false;
|
|
|
- }
|
|
|
- String fireTime = parseFireTimeFromCanvas(wf.getCanvasData());
|
|
|
- if (fireTime != null && fireTime.matches("\\d{2}:\\d{2}")) {
|
|
|
- try {
|
|
|
- LocalTime target = LocalTime.parse(fireTime);
|
|
|
- LocalTime curr = now.toLocalTime();
|
|
|
- return target.getHour() == curr.getHour() && target.getMinute() == curr.getMinute();
|
|
|
- } catch (Exception ignored) {}
|
|
|
- }
|
|
|
- return false;
|
|
|
- }
|
|
|
-
|
|
|
- /** 从 canvasData JSON 抓 startNode.config.fireTime（HH:mm） */
|
|
|
- private String parseFireTimeFromCanvas(String canvasData) {
|
|
|
- if (canvasData == null || canvasData.isEmpty()) return null;
|
|
|
- try {
|
|
|
- com.alibaba.fastjson.JSONObject json = com.alibaba.fastjson.JSON.parseObject(canvasData);
|
|
|
- com.alibaba.fastjson.JSONArray nodes = json.getJSONArray("nodes");
|
|
|
- if (nodes == null) return null;
|
|
|
- for (int i = 0; i < nodes.size(); i++) {
|
|
|
- com.alibaba.fastjson.JSONObject node = nodes.getJSONObject(i);
|
|
|
- if (node == null) continue;
|
|
|
- Integer type = node.getInteger("type");
|
|
|
- if (type != null && type == 1) { // 1=开始节点
|
|
|
- com.alibaba.fastjson.JSONObject cfg = node.getJSONObject("config");
|
|
|
- if (cfg != null) return cfg.getString("fireTime");
|
|
|
- }
|
|
|
- }
|
|
|
- } catch (Exception ignored) {}
|
|
|
- return null;
|
|
|
- }
|
|
|
-
|
|
|
- private void fire(CompanyWorkflowLobster wf, LocalDateTime now) {
|
|
|
- lastFireTime.put(wf.getId(), now);
|
|
|
- log.info("[WorkflowTriggerScheduler] 触发工作流 id={} template={}", wf.getId(), wf.getTemplateName());
|
|
|
- if (workflowExecutor == null) {
|
|
|
- log.warn("[WorkflowTriggerScheduler] workflowExecutor 未注入，跳过");
|
|
|
- return;
|
|
|
- }
|
|
|
- try {
|
|
|
- Map<String, Object> vars = new HashMap<>();
|
|
|
- vars.put("triggerSource", "scheduler");
|
|
|
- vars.put("triggerTime", now.toString());
|
|
|
- // 定时触发不绑定具体 contact，contactId=null
|
|
|
- workflowExecutor.startWorkflow(wf.getCompanyId(), wf.getId(), null, vars);
|
|
|
- } catch (Exception e) {
|
|
|
- log.error("[WorkflowTriggerScheduler] 工作流 {} 执行失败: {}", wf.getId(), e.getMessage(), e);
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- /** 从 canvasData startNode.config.eventType 解析事件触发类型 */
|
|
|
- private String parseEventTypeFromCanvas(String canvasData) {
|
|
|
- if (canvasData == null || canvasData.isEmpty()) return null;
|
|
|
- try {
|
|
|
- com.alibaba.fastjson.JSONObject json = com.alibaba.fastjson.JSON.parseObject(canvasData);
|
|
|
- com.alibaba.fastjson.JSONArray nodes = json.getJSONArray("nodes");
|
|
|
- if (nodes == null) return null;
|
|
|
- for (int i = 0; i < nodes.size(); i++) {
|
|
|
- com.alibaba.fastjson.JSONObject node = nodes.getJSONObject(i);
|
|
|
- if (node == null) continue;
|
|
|
- Integer type = node.getInteger("type");
|
|
|
- if (type != null && type == 1) {
|
|
|
- com.alibaba.fastjson.JSONObject cfg = node.getJSONObject("config");
|
|
|
- if (cfg != null) return cfg.getString("eventType");
|
|
|
- }
|
|
|
- }
|
|
|
- } catch (Exception ignored) {}
|
|
|
- return null;
|
|
|
- }
|
|
|
-}
|
|
|
+//package com.fs.company.service.workflow.scheduler;
|
|
|
+//
|
|
|
+//import com.fs.company.domain.CompanyWorkflowLobster;
|
|
|
+//import com.fs.company.mapper.CompanyWorkflowLobsterMapper;
|
|
|
+//import com.fs.company.service.workflow.LobsterWorkflowExecutor;
|
|
|
+//import org.slf4j.Logger;
|
|
|
+//import org.slf4j.LoggerFactory;
|
|
|
+//import org.springframework.beans.factory.annotation.Autowired;
|
|
|
+//import org.springframework.scheduling.annotation.Scheduled;
|
|
|
+//import org.springframework.stereotype.Component;
|
|
|
+//
|
|
|
+//import javax.annotation.PostConstruct;
|
|
|
+//import java.time.LocalDateTime;
|
|
|
+//import java.time.LocalTime;
|
|
|
+//import java.util.HashMap;
|
|
|
+//import java.util.List;
|
|
|
+//import java.util.Map;
|
|
|
+//import java.util.concurrent.ConcurrentHashMap;
|
|
|
+//
|
|
|
+///**
|
|
|
+// * 工作流触发调度器（skill.md 要求）
|
|
|
+// * <p>
|
|
|
+// * 三种触发机制：
|
|
|
+// * 1. 时间触发 — Cron / 固定时间 → 扫描启用的工作流并触发
|
|
|
+// * 2. 事件触发 — 客户消息到达时由消息处理器主动调用 fire(eventType)
|
|
|
+// * 3. 条件触发 — 定时巡检触发条件（活跃天数/沉默小时数等）
|
|
|
+// * <p>
|
|
|
+// * 实现：使用 Spring @Scheduled（每分钟检查一次），由 @EnableScheduling 启用
|
|
|
+// */
|
|
|
+//@Component
|
|
|
+//public class WorkflowTriggerScheduler {
|
|
|
+//
|
|
|
+// private static final Logger log = LoggerFactory.getLogger(WorkflowTriggerScheduler.class);
|
|
|
+//
|
|
|
+// @Autowired(required = false)
|
|
|
+// private CompanyWorkflowLobsterMapper workflowMapper;
|
|
|
+//
|
|
|
+// @Autowired(required = false)
|
|
|
+// private LobsterWorkflowExecutor workflowExecutor;
|
|
|
+//
|
|
|
+// @Autowired(required = false)
|
|
|
+// private com.fs.company.service.workflow.LobsterTestScenarioService testScenarioService;
|
|
|
+//
|
|
|
+// /** 最近一次触发时间（防重复） — workflowId → lastFireTime */
|
|
|
+// private final ConcurrentHashMap<Long, LocalDateTime> lastFireTime = new ConcurrentHashMap<>();
|
|
|
+//
|
|
|
+// @PostConstruct
|
|
|
+// public void init() {
|
|
|
+// log.info("[WorkflowTriggerScheduler] 启动，每分钟扫描一次工作流触发");
|
|
|
+// }
|
|
|
+//
|
|
|
+// /**
|
|
|
+// * 每天凌晨 3 点跑全部启用的 E2E 测试场景（回归测试）
|
|
|
+// */
|
|
|
+// @Scheduled(cron = "0 0 3 * * ?")
|
|
|
+// public void runDailyE2eRegression() {
|
|
|
+// if (testScenarioService == null) return;
|
|
|
+// try {
|
|
|
+// int n = testScenarioService.runAllEnabledScenarios();
|
|
|
+// log.info("[E2E回归] 已触发 {} 个测试场景", n);
|
|
|
+// } catch (Exception ex) {
|
|
|
+// log.error("[E2E回归] 失败: {}", ex.getMessage(), ex);
|
|
|
+// }
|
|
|
+// }
|
|
|
+//
|
|
|
+// /**
|
|
|
+// * 每分钟扫描一次启用状态的工作流，按其触发配置决定是否触发
|
|
|
+// */
|
|
|
+// @Scheduled(cron = "0 * * * * ?")
|
|
|
+// public void scanAndTrigger() {
|
|
|
+// if (workflowMapper == null) return;
|
|
|
+// try {
|
|
|
+// CompanyWorkflowLobster q = new CompanyWorkflowLobster();
|
|
|
+// q.setStatus(1); // 仅启用
|
|
|
+// List<CompanyWorkflowLobster> list = workflowMapper.selectList(new com.baomidou.mybatisplus.core.conditions.query.QueryWrapper<>(q));
|
|
|
+// if (list == null || list.isEmpty()) return;
|
|
|
+//
|
|
|
+// LocalDateTime now = LocalDateTime.now();
|
|
|
+// int triggered = 0;
|
|
|
+// for (CompanyWorkflowLobster wf : list) {
|
|
|
+// if (shouldTrigger(wf, now)) {
|
|
|
+// fire(wf, now);
|
|
|
+// triggered++;
|
|
|
+// }
|
|
|
+// }
|
|
|
+// if (triggered > 0) {
|
|
|
+// log.info("[WorkflowTriggerScheduler] 本轮触发 {} 个工作流", triggered);
|
|
|
+// }
|
|
|
+// } catch (Exception e) {
|
|
|
+// log.error("[WorkflowTriggerScheduler] 扫描异常: {}", e.getMessage(), e);
|
|
|
+// }
|
|
|
+// }
|
|
|
+//
|
|
|
+// /** 事件触发入口（消息到达/订单创建等场景，由业务方主动调用） */
|
|
|
+// public void fireByEvent(String eventType, Long companyId, Object payload) {
|
|
|
+// log.info("[WorkflowTriggerScheduler] 事件触发 type={} company={} payload={}", eventType, companyId, payload);
|
|
|
+// if (workflowMapper == null || workflowExecutor == null) return;
|
|
|
+// try {
|
|
|
+// CompanyWorkflowLobster q = new CompanyWorkflowLobster();
|
|
|
+// q.setCompanyId(companyId);
|
|
|
+// q.setStatus(1);
|
|
|
+// List<CompanyWorkflowLobster> list = workflowMapper.selectList(new com.baomidou.mybatisplus.core.conditions.query.QueryWrapper<>(q));
|
|
|
+// if (list == null) return;
|
|
|
+// for (CompanyWorkflowLobster wf : list) {
|
|
|
+// String evtCfg = parseEventTypeFromCanvas(wf.getCanvasData());
|
|
|
+// if (eventType != null && eventType.equalsIgnoreCase(evtCfg)) {
|
|
|
+// Map<String, Object> vars = new HashMap<>();
|
|
|
+// vars.put("eventType", eventType);
|
|
|
+// if (payload != null) vars.put("payload", payload);
|
|
|
+// Long contactId = payload instanceof Map ?
|
|
|
+// (Long) ((Map<?, ?>) payload).get("contactId") : null;
|
|
|
+// try {
|
|
|
+// workflowExecutor.startWorkflow(companyId, wf.getId(), contactId, vars);
|
|
|
+// log.info("[WorkflowTriggerScheduler] 事件 {} 启动工作流 id={}", eventType, wf.getId());
|
|
|
+// } catch (Exception ex) {
|
|
|
+// log.warn("[WorkflowTriggerScheduler] 启动工作流 {} 失败: {}", wf.getId(), ex.getMessage());
|
|
|
+// }
|
|
|
+// }
|
|
|
+// }
|
|
|
+// } catch (Exception e) {
|
|
|
+// log.error("[WorkflowTriggerScheduler] 事件触发异常: {}", e.getMessage(), e);
|
|
|
+// }
|
|
|
+// }
|
|
|
+//
|
|
|
+// // ════════════ 私有方法 ════════════
|
|
|
+//
|
|
|
+// /**
|
|
|
+// * 判断工作流是否应在当前时刻触发
|
|
|
+// * 规则：
|
|
|
+// * - 优先解析 canvasData 中的 startNode.config.fireTime（HH:mm）
|
|
|
+// * - 同一工作流 1 分钟内不重复触发
|
|
|
+// */
|
|
|
+// private boolean shouldTrigger(CompanyWorkflowLobster wf, LocalDateTime now) {
|
|
|
+// if (wf == null || wf.getId() == null) return false;
|
|
|
+// LocalDateTime last = lastFireTime.get(wf.getId());
|
|
|
+// if (last != null && java.time.Duration.between(last, now).getSeconds() < 60) {
|
|
|
+// return false;
|
|
|
+// }
|
|
|
+// String fireTime = parseFireTimeFromCanvas(wf.getCanvasData());
|
|
|
+// if (fireTime != null && fireTime.matches("\\d{2}:\\d{2}")) {
|
|
|
+// try {
|
|
|
+// LocalTime target = LocalTime.parse(fireTime);
|
|
|
+// LocalTime curr = now.toLocalTime();
|
|
|
+// return target.getHour() == curr.getHour() && target.getMinute() == curr.getMinute();
|
|
|
+// } catch (Exception ignored) {}
|
|
|
+// }
|
|
|
+// return false;
|
|
|
+// }
|
|
|
+//
|
|
|
+// /** 从 canvasData JSON 抓 startNode.config.fireTime（HH:mm） */
|
|
|
+// private String parseFireTimeFromCanvas(String canvasData) {
|
|
|
+// if (canvasData == null || canvasData.isEmpty()) return null;
|
|
|
+// try {
|
|
|
+// com.alibaba.fastjson.JSONObject json = com.alibaba.fastjson.JSON.parseObject(canvasData);
|
|
|
+// com.alibaba.fastjson.JSONArray nodes = json.getJSONArray("nodes");
|
|
|
+// if (nodes == null) return null;
|
|
|
+// for (int i = 0; i < nodes.size(); i++) {
|
|
|
+// com.alibaba.fastjson.JSONObject node = nodes.getJSONObject(i);
|
|
|
+// if (node == null) continue;
|
|
|
+// Integer type = node.getInteger("type");
|
|
|
+// if (type != null && type == 1) { // 1=开始节点
|
|
|
+// com.alibaba.fastjson.JSONObject cfg = node.getJSONObject("config");
|
|
|
+// if (cfg != null) return cfg.getString("fireTime");
|
|
|
+// }
|
|
|
+// }
|
|
|
+// } catch (Exception ignored) {}
|
|
|
+// return null;
|
|
|
+// }
|
|
|
+//
|
|
|
+// private void fire(CompanyWorkflowLobster wf, LocalDateTime now) {
|
|
|
+// lastFireTime.put(wf.getId(), now);
|
|
|
+// log.info("[WorkflowTriggerScheduler] 触发工作流 id={} template={}", wf.getId(), wf.getTemplateName());
|
|
|
+// if (workflowExecutor == null) {
|
|
|
+// log.warn("[WorkflowTriggerScheduler] workflowExecutor 未注入，跳过");
|
|
|
+// return;
|
|
|
+// }
|
|
|
+// try {
|
|
|
+// Map<String, Object> vars = new HashMap<>();
|
|
|
+// vars.put("triggerSource", "scheduler");
|
|
|
+// vars.put("triggerTime", now.toString());
|
|
|
+// // 定时触发不绑定具体 contact，contactId=null
|
|
|
+// workflowExecutor.startWorkflow(wf.getCompanyId(), wf.getId(), null, vars);
|
|
|
+// } catch (Exception e) {
|
|
|
+// log.error("[WorkflowTriggerScheduler] 工作流 {} 执行失败: {}", wf.getId(), e.getMessage(), e);
|
|
|
+// }
|
|
|
+// }
|
|
|
+//
|
|
|
+// /** 从 canvasData startNode.config.eventType 解析事件触发类型 */
|
|
|
+// private String parseEventTypeFromCanvas(String canvasData) {
|
|
|
+// if (canvasData == null || canvasData.isEmpty()) return null;
|
|
|
+// try {
|
|
|
+// com.alibaba.fastjson.JSONObject json = com.alibaba.fastjson.JSON.parseObject(canvasData);
|
|
|
+// com.alibaba.fastjson.JSONArray nodes = json.getJSONArray("nodes");
|
|
|
+// if (nodes == null) return null;
|
|
|
+// for (int i = 0; i < nodes.size(); i++) {
|
|
|
+// com.alibaba.fastjson.JSONObject node = nodes.getJSONObject(i);
|
|
|
+// if (node == null) continue;
|
|
|
+// Integer type = node.getInteger("type");
|
|
|
+// if (type != null && type == 1) {
|
|
|
+// com.alibaba.fastjson.JSONObject cfg = node.getJSONObject("config");
|
|
|
+// if (cfg != null) return cfg.getString("eventType");
|
|
|
+// }
|
|
|
+// }
|
|
|
+// } catch (Exception ignored) {}
|
|
|
+// return null;
|
|
|
+// }
|
|
|
+//}
|
