server/sino-mci
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(inbound): add flow execution trace recording and query

Browse Source
This commit is contained in:
王宇航
2026-07-07 15:18:55 +08:00
parent ee82e6e447
commit 891ded382e
7 changed files with 116 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
backend/mci-server/src/main/java/com/sino/mci/config/FlywayConfig.java
View File

@@ -21,6 +21,7 @@ public class FlywayConfig {
.dataSource(dataSource)
.locations(locations)
.baselineOnMigrate(baselineOnMigrate)
.outOfOrder(true)
.load();
}
}

8
backend/mci-server/src/main/java/com/sino/mci/inbound/controller/InboundMessageController.java
View File

@@ -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<List<FlowExecutionTrace>> listTraces(@PathVariable String eventId) {
return ApiResponse.ok(inboundMessageService.listTraces(eventId));
}
}

47
backend/mci-server/src/main/java/com/sino/mci/inbound/engine/FlowEngine.java
View File

@@ -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<String, Object> 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();
}
}

21
backend/mci-server/src/main/java/com/sino/mci/inbound/entity/FlowExecutionTrace.java Normal file
View File

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

9
backend/mci-server/src/main/java/com/sino/mci/inbound/repository/FlowExecutionTraceMapper.java Normal file
View File

@@ -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<FlowExecutionTrace> {
}

21
backend/mci-server/src/main/java/com/sino/mci/inbound/service/InboundMessageService.java
View File

@@ -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<FlowExecutionTrace> listTraces(String eventId) {
return flowExecutionTraceMapper.selectList(
new LambdaQueryWrapper<FlowExecutionTrace>()
.eq(FlowExecutionTrace::getEventId, eventId)
.orderByAsc(FlowExecutionTrace::getId));
}
}

14
backend/mci-server/src/main/resources/db/migration/V8__add_flow_trace.sql Normal file
View File

@@ -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='流程执行轨迹表';