From 891ded382ee4520b7497d8c10ebb27da81e09b73 Mon Sep 17 00:00:00 2001 From: marsal Date: Tue, 7 Jul 2026 15:18:55 +0800 Subject: [PATCH] feat(inbound): add flow execution trace recording and query --- .../com/sino/mci/config/FlywayConfig.java | 1 + .../controller/InboundMessageController.java | 8 ++++ .../sino/mci/inbound/engine/FlowEngine.java | 47 +++++++++++++++++-- .../inbound/entity/FlowExecutionTrace.java | 21 +++++++++ .../repository/FlowExecutionTraceMapper.java | 9 ++++ .../service/InboundMessageService.java | 21 +++++++++ .../db/migration/V8__add_flow_trace.sql | 14 ++++++ 7 files changed, 116 insertions(+), 5 deletions(-) create mode 100644 backend/mci-server/src/main/java/com/sino/mci/inbound/entity/FlowExecutionTrace.java create mode 100644 backend/mci-server/src/main/java/com/sino/mci/inbound/repository/FlowExecutionTraceMapper.java create mode 100644 backend/mci-server/src/main/resources/db/migration/V8__add_flow_trace.sql diff --git a/backend/mci-server/src/main/java/com/sino/mci/config/FlywayConfig.java b/backend/mci-server/src/main/java/com/sino/mci/config/FlywayConfig.java index 363df29..27da840 100644 --- a/backend/mci-server/src/main/java/com/sino/mci/config/FlywayConfig.java +++ b/backend/mci-server/src/main/java/com/sino/mci/config/FlywayConfig.java @@ -21,6 +21,7 @@ public class FlywayConfig { .dataSource(dataSource) .locations(locations) .baselineOnMigrate(baselineOnMigrate) + .outOfOrder(true) .load(); } } diff --git a/backend/mci-server/src/main/java/com/sino/mci/inbound/controller/InboundMessageController.java b/backend/mci-server/src/main/java/com/sino/mci/inbound/controller/InboundMessageController.java index 11ddc6c..775945b 100644 --- a/backend/mci-server/src/main/java/com/sino/mci/inbound/controller/InboundMessageController.java +++ b/backend/mci-server/src/main/java/com/sino/mci/inbound/controller/InboundMessageController.java @@ -1,12 +1,15 @@ package com.sino.mci.inbound.controller; import com.sino.mci.inbound.dto.ReceiveMessageRequest; +import com.sino.mci.inbound.entity.FlowExecutionTrace; import com.sino.mci.inbound.service.InboundMessageService; import com.sino.mci.shared.web.ApiResponse; import jakarta.validation.Valid; import lombok.RequiredArgsConstructor; import org.springframework.web.bind.annotation.*; +import java.util.List; + @RestController @RequestMapping("/api/v1/inbound") @RequiredArgsConstructor @@ -20,4 +23,9 @@ public class InboundMessageController { @Valid @RequestBody ReceiveMessageRequest request) { return ApiResponse.ok(inboundMessageService.receiveMessage(tenantCode, request)); } + + @GetMapping("/{eventId}/trace") + public ApiResponse> listTraces(@PathVariable String eventId) { + return ApiResponse.ok(inboundMessageService.listTraces(eventId)); + } } diff --git a/backend/mci-server/src/main/java/com/sino/mci/inbound/engine/FlowEngine.java b/backend/mci-server/src/main/java/com/sino/mci/inbound/engine/FlowEngine.java index 58d623d..aea5ea7 100644 --- a/backend/mci-server/src/main/java/com/sino/mci/inbound/engine/FlowEngine.java +++ b/backend/mci-server/src/main/java/com/sino/mci/inbound/engine/FlowEngine.java @@ -3,7 +3,9 @@ package com.sino.mci.inbound.engine; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import com.sino.mci.flow.entity.FlowDefinition; import com.sino.mci.flow.repository.FlowDefinitionMapper; +import com.sino.mci.inbound.entity.FlowExecutionTrace; import com.sino.mci.inbound.entity.MessageEvent; +import com.sino.mci.inbound.repository.FlowExecutionTraceMapper; import com.sino.mci.inbound.repository.MessageEventMapper; import com.sino.mci.shared.util.JsonUtils; import lombok.RequiredArgsConstructor; @@ -22,6 +24,7 @@ public class FlowEngine { private final MessageEventMapper messageEventMapper; private final IntentRecognitionService intentRecognitionService; private final WebhookPushService webhookPushService; + private final FlowExecutionTraceMapper flowExecutionTraceMapper; public void executeInboundFlow(Long eventId) { MessageEvent event = messageEventMapper.selectById(eventId); @@ -30,10 +33,13 @@ public class FlowEngine { return; } + trace(event, null, "flow_start", event.getEventId(), null, 1, now(), null); + FlowDefinition flow = findPublishedFlow(event.getTenantCode()); + Long flowDefinitionId = flow != null ? flow.getId() : null; if (flow == null) { log.info("No published inbound flow for tenant {}, skipping flow execution", event.getTenantCode()); - markProcessed(event); + markProcessed(event, flowDefinitionId); return; } @@ -41,16 +47,27 @@ public class FlowEngine { Map definition = JsonUtils.fromJson(flow.getDefinitionJson(), Map.class); if (definition == null) { log.warn("Invalid flow definition JSON for flow {}", flow.getFlowCode()); - markProcessed(event); + markProcessed(event, flowDefinitionId); return; } + long recognizeStart = now(); IntentRecognitionService.IntentResult intent = intentRecognitionService.recognize(event); + trace(event, flowDefinitionId, "intent_recognition", + event.getContent(), JsonUtils.toJson(intent), 1, recognizeStart, null); log.info("Intent recognized: eventId={}, intent={}, confidence={}", event.getEventId(), intent.intent(), intent.confidence()); - webhookPushService.pushToBusiness(event.getTenantCode(), event, intent); - markProcessed(event); + long pushStart = now(); + try { + webhookPushService.pushToBusiness(event.getTenantCode(), event, intent); + trace(event, flowDefinitionId, "webhook_push", event.getEventId(), "ok", 1, pushStart, null); + } catch (Exception e) { + trace(event, flowDefinitionId, "webhook_push", event.getEventId(), null, 2, pushStart, e.getMessage()); + throw e; + } + + markProcessed(event, flowDefinitionId); } catch (Exception e) { log.error("Flow execution failed for event {}", eventId, e); } @@ -66,8 +83,28 @@ public class FlowEngine { return flows.isEmpty() ? null : flows.get(0); } - private void markProcessed(MessageEvent event) { + private void markProcessed(MessageEvent event, Long flowDefinitionId) { event.setIsProcessed(1); messageEventMapper.updateById(event); + trace(event, flowDefinitionId, "mark_processed", event.getEventId(), "processed", 1, now(), null); + } + + private void trace(MessageEvent event, Long flowDefinitionId, String nodeType, + String input, String output, int status, long startMs, String errorMsg) { + FlowExecutionTrace record = new FlowExecutionTrace(); + record.setEventId(event.getEventId()); + record.setFlowDefinitionId(flowDefinitionId); + record.setNodeType(nodeType); + record.setNodeConfig(null); + record.setInput(input); + record.setOutput(output); + record.setStatus(status); + record.setDurationMs((int) (now() - startMs)); + record.setErrorMsg(errorMsg); + flowExecutionTraceMapper.insert(record); + } + + private long now() { + return System.currentTimeMillis(); } } diff --git a/backend/mci-server/src/main/java/com/sino/mci/inbound/entity/FlowExecutionTrace.java b/backend/mci-server/src/main/java/com/sino/mci/inbound/entity/FlowExecutionTrace.java new file mode 100644 index 0000000..5a40577 --- /dev/null +++ b/backend/mci-server/src/main/java/com/sino/mci/inbound/entity/FlowExecutionTrace.java @@ -0,0 +1,21 @@ +package com.sino.mci.inbound.entity; + +import com.baomidou.mybatisplus.annotation.TableName; +import lombok.Data; +import java.time.LocalDateTime; + +@Data +@TableName("mci_flow_execution_trace") +public class FlowExecutionTrace { + private Long id; + private String eventId; + private Long flowDefinitionId; + private String nodeType; + private String nodeConfig; + private String input; + private String output; + private Integer status; + private Integer durationMs; + private String errorMsg; + private LocalDateTime createTime; +} diff --git a/backend/mci-server/src/main/java/com/sino/mci/inbound/repository/FlowExecutionTraceMapper.java b/backend/mci-server/src/main/java/com/sino/mci/inbound/repository/FlowExecutionTraceMapper.java new file mode 100644 index 0000000..6d1e506 --- /dev/null +++ b/backend/mci-server/src/main/java/com/sino/mci/inbound/repository/FlowExecutionTraceMapper.java @@ -0,0 +1,9 @@ +package com.sino.mci.inbound.repository; + +import com.baomidou.mybatisplus.core.mapper.BaseMapper; +import com.sino.mci.inbound.entity.FlowExecutionTrace; +import org.apache.ibatis.annotations.Mapper; + +@Mapper +public interface FlowExecutionTraceMapper extends BaseMapper { +} diff --git a/backend/mci-server/src/main/java/com/sino/mci/inbound/service/InboundMessageService.java b/backend/mci-server/src/main/java/com/sino/mci/inbound/service/InboundMessageService.java index 25a775b..e8379fd 100644 --- a/backend/mci-server/src/main/java/com/sino/mci/inbound/service/InboundMessageService.java +++ b/backend/mci-server/src/main/java/com/sino/mci/inbound/service/InboundMessageService.java @@ -1,8 +1,11 @@ package com.sino.mci.inbound.service; +import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; import com.sino.mci.inbound.dto.ReceiveMessageRequest; +import com.sino.mci.inbound.entity.FlowExecutionTrace; import com.sino.mci.inbound.entity.MessageEvent; import com.sino.mci.inbound.engine.FlowEngine; +import com.sino.mci.inbound.repository.FlowExecutionTraceMapper; import com.sino.mci.inbound.repository.MessageEventMapper; import com.sino.mci.shared.util.JsonUtils; import lombok.RequiredArgsConstructor; @@ -10,12 +13,15 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; +import java.util.List; + @Slf4j @Service @RequiredArgsConstructor public class InboundMessageService { private final MessageEventMapper messageEventMapper; + private final FlowExecutionTraceMapper flowExecutionTraceMapper; private final FlowEngine flowEngine; @Transactional @@ -40,7 +46,22 @@ public class InboundMessageService { log.info("Inbound message received: tenant={}, eventId={}, conversationId={}", tenantCode, request.getEventId(), request.getConversationId()); + FlowExecutionTrace receiveTrace = new FlowExecutionTrace(); + receiveTrace.setEventId(event.getEventId()); + receiveTrace.setNodeType("receive"); + receiveTrace.setInput(JsonUtils.toJson(request)); + receiveTrace.setOutput("event_id=" + event.getId()); + receiveTrace.setStatus(1); + flowExecutionTraceMapper.insert(receiveTrace); + flowEngine.executeInboundFlow(event.getId()); return messageEventMapper.selectById(event.getId()); } + + public List listTraces(String eventId) { + return flowExecutionTraceMapper.selectList( + new LambdaQueryWrapper() + .eq(FlowExecutionTrace::getEventId, eventId) + .orderByAsc(FlowExecutionTrace::getId)); + } } diff --git a/backend/mci-server/src/main/resources/db/migration/V8__add_flow_trace.sql b/backend/mci-server/src/main/resources/db/migration/V8__add_flow_trace.sql new file mode 100644 index 0000000..52c0345 --- /dev/null +++ b/backend/mci-server/src/main/resources/db/migration/V8__add_flow_trace.sql @@ -0,0 +1,14 @@ +CREATE TABLE IF NOT EXISTS mci_flow_execution_trace ( + id BIGINT AUTO_INCREMENT PRIMARY KEY, + event_id VARCHAR(128) NOT NULL COMMENT '入站事件ID', + flow_definition_id BIGINT COMMENT '流程定义ID', + node_type VARCHAR(64) NOT NULL COMMENT '节点类型', + node_config JSON COMMENT '节点配置快照', + input TEXT COMMENT '节点输入', + output TEXT COMMENT '节点输出', + status TINYINT NOT NULL DEFAULT 1 COMMENT '1成功 2失败', + duration_ms INT COMMENT '执行耗时毫秒', + error_msg VARCHAR(512) COMMENT '错误信息', + create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + KEY idx_event_id (event_id) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='流程执行轨迹表';