agent-management/backend/src/main/java/com/agent/management/external/ExternalApiController.java

package com.agent.management.external;

import com.agent.management.config.ExternalApiAuthFilter;
import com.agent.management.config.ExternalApiProperties;
import com.agent.management.engine.AgentIgnoreFilter;
import com.agent.management.engine.WorkflowEngine;
import com.agent.management.engine.WorkflowRunDirManager;
import com.agent.management.engine.WorkflowRunEvent;
import com.agent.management.external.dto.ApiError;
import com.agent.management.external.dto.ExternalDtos.CreateRunRequest;
import com.agent.management.external.dto.ExternalDtos.CreateRunResponse;
import com.agent.management.external.dto.ExternalDtos.FieldDef;
import com.agent.management.external.dto.ExternalDtos.Links;
import com.agent.management.external.dto.ExternalDtos.LogEntry;
import com.agent.management.external.dto.ExternalDtos.NodeLogsResponse;
import com.agent.management.external.dto.ExternalDtos.NodeResult;
import com.agent.management.external.dto.ExternalDtos.RunResultResponse;
import com.agent.management.external.dto.ExternalDtos.StartRunResponse;
import com.agent.management.external.dto.ExternalDtos.WorkflowListResponse;
import com.agent.management.external.dto.ExternalDtos.WorkflowSummary;
import com.agent.management.external.dto.ExternalDtos.WorkspaceInfo;
import com.agent.management.model.entity.Workflow;
import com.agent.management.model.entity.WorkflowRun;
import com.agent.management.model.entity.WorkflowRunNode;
import com.agent.management.repository.WorkflowRepository;
import com.agent.management.repository.WorkflowRunNodeRepository;
import com.agent.management.repository.WorkflowRunRepository;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.io.Resource;
import org.springframework.core.io.UrlResource;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.multipart.MultipartFile;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.net.URLEncoder;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.*;
import java.util.stream.Collectors;
import java.util.zip.ZipEntry;
import java.util.zip.ZipOutputStream;

/**
 * 外部 API（/api/v1/**）入口：第三方系统通过此 Controller 调用平台工作流。
 *
 * <p>URL 路径使用 kebab-case 唯一标识 {@code workflowName}（与 Skill 一致）。
 * 旧调用方使用数字 id 的 URL 仍可通过 by-id 兜底逻辑访问（用户手动改名为非数字前都自动兼容）。
 *
 * 流程：
 * - POST /workflows/{workflowName}/runs           创建运行
 * - POST /workflows/{workflowName}/runs/{runId}/start  启动执行
 * - GET  /workflows/{workflowName}/runs/{runId}/stream 订阅 SSE 流
 * - GET  /workflows/{workflowName}/runs/{runId}        查询运行结果
 * - GET  /workflows/{workflowName}/runs/{runId}/workspace  下载工作空间 zip
 * - GET  /workflows/{workflowName}/runs/{runId}/nodes/{nodeId}/logs  查询节点日志
 * - GET  /workflows                          列出当前 API Key 可调用的工作流
 */
@Slf4j
@RestController
@RequestMapping("/api/v1")
@RequiredArgsConstructor
public class ExternalApiController {

    private final ExternalApiProperties properties;
    private final ExternalRunRegistry runRegistry;
    private final SseEventBus eventBus;
    private final WorkflowEngine workflowEngine;
    private final WorkflowRepository workflowRepository;
    private final WorkflowRunRepository workflowRunRepository;
    private final WorkflowRunNodeRepository workflowRunNodeRepository;
    private final WorkflowRunDirManager runDirManager;
    private final AgentIgnoreFilter agentIgnoreFilter;

    private static final ObjectMapper MAPPER = new ObjectMapper().registerModule(new JavaTimeModule());
    private static final int DEFAULT_TTL_HOURS = 720;
    private static final long SSE_TIMEOUT_MS = 1_800_000L;

    // ============================================================
    // POST /workflows/{workflowName}/runs — 创建运行
    // ============================================================

    @PostMapping(value = "/workflows/{workflowName}/runs", consumes = MediaType.APPLICATION_JSON_VALUE)
    public ResponseEntity<?> createRunJson(@PathVariable String workflowName,
                                           @RequestAttribute(ExternalApiAuthFilter.ATTR_KEY_ENTRY) ExternalApiProperties.KeyEntry keyEntry,
                                           @RequestBody(required = false) CreateRunRequest req) {
        return doCreateRun(workflowName, req, null, null);
    }

    @PostMapping(value = "/workflows/{workflowName}/runs", consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
    public ResponseEntity<?> createRunMultipart(@PathVariable String workflowName,
                                                @RequestAttribute(ExternalApiAuthFilter.ATTR_KEY_ENTRY) ExternalApiProperties.KeyEntry keyEntry,
                                                @RequestParam(value = "payload", required = false) String payloadJson,
                                                @RequestParam(value = "files", required = false) MultipartFile[] files,
                                                @RequestHeader(value = "X-Relative-Path", required = false) List<String> relativePaths) {
        CreateRunRequest req = parsePayload(payloadJson);
        return doCreateRun(workflowName, req, files, relativePaths);
    }

    private ResponseEntity<?> doCreateRun(String workflowName, CreateRunRequest req,
                                          MultipartFile[] files, List<String> relativePaths) {
        // 1. 工作流解析
        Workflow wf = resolveWorkflow(workflowName);
        if (wf == null) {
            return error(404, "WORKFLOW_NOT_FOUND", "工作流不存在: " + workflowName);
        }
        if (wf.getGraphData() == null || wf.getGraphData().isBlank()
                || wf.getGraphData().equals("{\"nodes\":[],\"edges\":[]}")) {
            return error(422, "GRAPH_INVALID", "工作流图为空，无法执行");
        }

        // 2. 准备 inputs
        Map<String, Object> inputs = new HashMap<>();
        if (req != null && req.getVariables() != null) {
            inputs.putAll(req.getVariables());
        }

        // 3. 生成 runId + 创建工作目录 + 写入上传文件
        String runId;
        try {
            runId = generateRunId();
            Path runDir = runDirManager.createRunDir(wf.getId(), wf.getName(), runId);
            List<String> uploadedFiles = new ArrayList<>();
            if (files != null) {
                for (int i = 0; i < files.length; i++) {
                    MultipartFile file = files[i];
                    if (file == null || file.isEmpty()) continue;
                    String rel = (relativePaths != null && i < relativePaths.size() && !relativePaths.get(i).isBlank())
                            ? relativePaths.get(i) : file.getOriginalFilename();
                    if (rel == null || rel.isBlank()) continue;
                    Path dest = runDir.resolve(rel).normalize();
                    if (!dest.startsWith(runDir)) {
                        return error(400, "VALIDATION_FAILED", "非法上传路径: " + rel);
                    }
                    Files.createDirectories(dest.getParent());
                    file.transferTo(dest.toFile());
                    uploadedFiles.add(runDir.relativize(dest).toString().replace('\\', '/'));
                }
            }
            if (!uploadedFiles.isEmpty()) {
                inputs.put("uploadedFiles", uploadedFiles);
            }
        } catch (IllegalArgumentException e) {
            return error(400, "VALIDATION_FAILED", e.getMessage());
        } catch (IOException e) {
            log.error("[ExternalApi] 创建运行失败 workflowName={}", workflowName, e);
            return error(500, "INTERNAL_ERROR", "工作目录创建失败");
        }

        // 4. 注册 CREATED 状态
        int ttlHours = (req != null && req.getTtlHours() != null && req.getTtlHours() > 0)
                ? req.getTtlHours() : DEFAULT_TTL_HOURS;
        runRegistry.register(runId, wf.getId(), wf.getName(), inputs, ttlHours);
        eventBus.register(runId);

        // 5. async=false：立即启动并直接进入 SSE 流
        boolean async = req == null || req.getAsync() == null || req.getAsync();
        if (!async) {
            return startRun(wf.getName(), wf.getId(), runId, inputs);
        }

        // 6. async=true：返回 runId + links（URL 使用 workflowName）
        CreateRunResponse body = new CreateRunResponse();
        body.setRunId(runId);
        body.setWorkflowId(wf.getId());
        body.setWorkflowName(wf.getName());
        body.setDisplayName(wf.getDisplayName());
        body.setStatus("CREATED");
        body.setCreatedAt(Instant.now());
        Links links = new Links();
        String base = "/api/v1/workflows/" + wf.getName() + "/runs/" + runId;
        links.setStart(base + "/start");
        links.setStream(base + "/stream");
        links.setResult(base);
        links.setWorkspace(base + "/workspace");
        body.setLinks(links);
        return ResponseEntity.ok(ApiError.ok(body));
    }

    // ============================================================
    // POST /workflows/{workflowName}/runs/{runId}/start — 启动运行
    // ============================================================

    @PostMapping("/workflows/{workflowName}/runs/{runId}/start")
    public ResponseEntity<?> startRun(@PathVariable String workflowName,
                                      @PathVariable String runId) {
        ExternalRunRegistry.CreatedRun created = runRegistry.consume(runId);
        if (created == null) {
            return error(404, "RUN_NOT_FOUND", "运行不存在或已启动: " + runId);
        }
        // 校验 pathVariable 与 created 一致（兼容数字字符串）
        if (!matchesWorkflow(created, workflowName)) {
            return error(400, "VALIDATION_FAILED", "runId 与 workflowName 不匹配");
        }
        return startRun(created.workflowName(), created.workflowId(), runId, created.inputs());
    }

    private ResponseEntity<?> startRun(String workflowName, Long workflowId, String runId, Map<String, Object> inputs) {
        workflowEngine.executeForExternal(workflowId, runId, inputs);

        StartRunResponse body = new StartRunResponse();
        body.setRunId(runId);
        body.setStatus("RUNNING");
        body.setStreamUrl("/api/v1/workflows/" + workflowName + "/runs/" + runId + "/stream");
        return ResponseEntity.ok(ApiError.ok(body));
    }

    // ============================================================
    // GET /workflows/{workflowName}/runs/{runId}/stream — SSE 流
    // ============================================================

    @GetMapping(value = "/workflows/{workflowName}/runs/{runId}/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public ResponseEntity<SseEmitter> streamRun(@PathVariable String workflowName,
                                                @PathVariable String runId,
                                                @RequestHeader(value = "Last-Event-ID", required = false) Long lastEventId) {
        if (!eventBus.exists(runId)) {
            WorkflowRun record = findRunRecord(runId);
            if (record == null) {
                return ResponseEntity.notFound().build();
            }
            eventBus.register(runId);
            eventBus.publish(runId,
                    record.getStatus().equals("FAILED") ? "workflow_error" : "workflow_complete",
                    WorkflowRunEvent.workflowComplete(runId, parseOutputs(record)));
        }

        SseEmitter emitter = eventBus.subscribe(runId, lastEventId);
        if (emitter == null) {
            return ResponseEntity.notFound().build();
        }
        return ResponseEntity.ok()
                .header("X-Accel-Buffering", "no")
                .header("Cache-Control", "no-cache")
                .body(emitter);
    }

    // ============================================================
    // GET /workflows/{workflowName}/runs/{runId} — 查询运行结果
    // ============================================================

    @GetMapping("/workflows/{workflowName}/runs/{runId}")
    public ResponseEntity<?> getRunResult(@PathVariable String workflowName,
                                          @PathVariable String runId) {
        Workflow wf = resolveWorkflow(workflowName);
        if (wf == null) {
            return error(404, "WORKFLOW_NOT_FOUND", "工作流不存在: " + workflowName);
        }
        RunResultResponse body = new RunResultResponse();
        body.setRunId(runId);
        body.setWorkflowId(wf.getId());
        body.setWorkflowName(wf.getName());
        body.setDisplayName(wf.getDisplayName());

        SseEventBus.RunStatus live = eventBus.getStatus(runId);
        if (live != null) {
            body.setStatus(live.status());
        }

        WorkflowRun record = findRunRecord(runId);
        if (record == null) {
            ExternalRunRegistry.CreatedRun created = runRegistry.peek(runId);
            if (created != null) {
                body.setStatus("CREATED");
                return ResponseEntity.ok(ApiError.ok(body));
            }
            return error(404, "RUN_NOT_FOUND", "运行不存在: " + runId);
        }

        body.setStatus(record.getStatus());
        if (record.getStartedAt() != null) {
            body.setStartedAt(record.getStartedAt().toInstant(ZoneOffset.UTC));
        }
        if (record.getCompletedAt() != null) {
            body.setCompletedAt(record.getCompletedAt().toInstant(ZoneOffset.UTC));
        }
        body.setError(record.getError());
        body.setOutputs(parseOutputs(record));

        List<WorkflowRunNode> nodes = workflowRunNodeRepository.findByRunIdOrderBySortOrderAsc(record.getId());
        List<NodeResult> nodeDtos = nodes.stream().map(n -> {
            NodeResult dto = new NodeResult();
            dto.setNodeId(n.getNodeId());
            dto.setNodeType(n.getNodeType());
            dto.setLabel(n.getLabel());
            dto.setStatus(n.getStatus());
            dto.setOutput(parseJsonToMap(n.getOutput()));
            dto.setError(n.getError());
            dto.setLogsCount(countLogs(n.getLogs()));
            dto.setLogsUrl("/api/v1/workflows/" + wf.getName() + "/runs/" + runId + "/nodes/" + n.getNodeId() + "/logs");
            return dto;
        }).collect(Collectors.toList());
        body.setNodes(nodeDtos);

        Path runDir = runDirManager.getRunDir(wf.getId(), wf.getName(), runId);
        if (runDir != null) {
            try {
                final int[] fileCount = {0};
                final long[] totalBytes = {0};
                Files.walk(runDir)
                        .filter(p -> !Files.isDirectory(p) && shouldIncludeInZip(runDir, p))
                        .forEach(p -> {
                            fileCount[0]++;
                            try { totalBytes[0] += Files.size(p); } catch (IOException ignore) {}
                        });
                WorkspaceInfo ws = new WorkspaceInfo();
                ws.setFileCount(fileCount[0]);
                ws.setTotalBytes(totalBytes[0]);
                ws.setDownloadUrl("/api/v1/workflows/" + wf.getName() + "/runs/" + runId + "/workspace");
                body.setWorkspace(ws);
            } catch (IOException e) {
                log.warn("[ExternalApi] 统计工作目录失败 runId={}: {}", runId, e.getMessage());
            }
        }

        return ResponseEntity.ok(ApiError.ok(body));
    }

    // ============================================================
    // GET /workflows/{workflowName}/runs/{runId}/workspace
    // ============================================================

    @GetMapping("/workflows/{workflowName}/runs/{runId}/workspace")
    public ResponseEntity<Resource> downloadWorkspace(@PathVariable String workflowName,
                                                      @PathVariable String runId) throws IOException {
        Workflow wf = resolveWorkflow(workflowName);
        if (wf == null) {
            return ResponseEntity.notFound().build();
        }
        Path runDir = runDirManager.getRunDir(wf.getId(), wf.getName(), runId);
        if (runDir == null) {
            return ResponseEntity.notFound().build();
        }

        Path zipPath = Files.createTempFile("external-run-" + runId, ".zip");
        try (ZipOutputStream zos = new ZipOutputStream(Files.newOutputStream(zipPath))) {
            Files.walk(runDir)
                    .filter(p -> !Files.isDirectory(p) && shouldIncludeInZip(runDir, p))
                    .forEach(p -> {
                        ZipEntry entry = new ZipEntry(runDir.relativize(p).toString().replace('\\', '/'));
                        try {
                            zos.putNextEntry(entry);
                            Files.copy(p, zos);
                            zos.closeEntry();
                        } catch (IOException e) {
                            log.warn("[ExternalApi] zip 条目写入失败: {}", p, e);
                        }
                    });
        }

        Resource resource = new UrlResource(zipPath.toUri());
        String filename = URLEncoder.encode("workflow-" + wf.getName() + "-run-" + runId + ".zip", StandardCharsets.UTF_8);
        return ResponseEntity.ok()
                .header(HttpHeaders.CONTENT_DISPOSITION, "attachment; filename*=UTF-8''" + filename)
                .contentType(MediaType.APPLICATION_OCTET_STREAM)
                .body(resource);
    }

    // ============================================================
    // GET /workflows/{workflowName}/runs/{runId}/nodes/{nodeId}/logs
    // ============================================================

    @GetMapping("/workflows/{workflowName}/runs/{runId}/nodes/{nodeId}/logs")
    public ResponseEntity<?> getNodeLogs(@PathVariable String workflowName,
                                         @PathVariable String runId,
                                         @PathVariable String nodeId) {
        WorkflowRun record = findRunRecord(runId);
        if (record == null) {
            return error(404, "RUN_NOT_FOUND", "运行不存在: " + runId);
        }
        List<WorkflowRunNode> nodes = workflowRunNodeRepository.findByRunIdOrderBySortOrderAsc(record.getId());
        WorkflowRunNode target = nodes.stream()
                .filter(n -> nodeId.equals(n.getNodeId()))
                .findFirst().orElse(null);
        if (target == null) {
            return error(404, "NODE_NOT_FOUND", "节点不存在: " + nodeId);
        }

        List<LogEntry> logs = new ArrayList<>();
        if (target.getLogs() != null && !target.getLogs().isBlank()) {
            try {
                List<JsonNode> raw = MAPPER.readValue(target.getLogs(), new TypeReference<List<JsonNode>>() {});
                for (JsonNode n : raw) {
                    LogEntry e = new LogEntry();
                    e.setType(n.path("type").asText(""));
                    e.setMessage(n.path("message").asText(""));
                    e.setDetail(n.path("detail").isNull() ? null : n.path("detail").asText(""));
                    String ts = n.path("timestamp").asText("");
                    if (!ts.isEmpty()) {
                        try { e.setTimestamp(Instant.parse(ts)); } catch (Exception ignore) {}
                    }
                    logs.add(e);
                }
            } catch (Exception e) {
                log.warn("[ExternalApi] 解析节点日志失败 nodeId={}: {}", nodeId, e.getMessage());
            }
        }

        NodeLogsResponse body = new NodeLogsResponse();
        body.setRunId(runId);
        body.setNodeId(nodeId);
        body.setLogs(logs);
        return ResponseEntity.ok(ApiError.ok(body));
    }

    // ============================================================
    // GET /workflows — 列出可调用工作流
    // ============================================================

    @GetMapping("/workflows")
    public ResponseEntity<?> listWorkflows(@RequestAttribute(ExternalApiAuthFilter.ATTR_KEY_ENTRY) ExternalApiProperties.KeyEntry keyEntry) {
        List<Workflow> all = workflowRepository.findAllByOrderByUpdatedAtDesc();
        List<Workflow> accessible = all.stream()
                .filter(w -> properties.canAccess(keyEntry, w.getName(), w.getId()))
                .filter(w -> w.getGraphData() != null && !w.getGraphData().isBlank())
                .collect(Collectors.toList());

        List<WorkflowSummary> items = new ArrayList<>();
        for (Workflow w : accessible) {
            WorkflowSummary s = new WorkflowSummary();
            s.setId(w.getId());
            s.setName(w.getName());
            s.setDisplayName(w.getDisplayName());
            s.setDescription(w.getDescription());
            try {
                JsonNode root = MAPPER.readTree(w.getGraphData());
                s.setInputs(extractFields(root, "userInput"));
                s.setOutputs(extractFields(root, "output"));
            } catch (Exception e) {
                s.setInputs(List.of());
                s.setOutputs(List.of());
            }
            items.add(s);
        }

        WorkflowListResponse body = new WorkflowListResponse();
        body.setTotal(items.size());
        body.setItems(items);
        return ResponseEntity.ok(ApiError.ok(body));
    }

    // ============================================================
    // 工具方法
    // ============================================================

    /**
     * 解析工作流：优先 findByName，回退 findById（兼容数字 id URL）。
     */
    private Workflow resolveWorkflow(String nameOrId) {
        if (nameOrId == null || nameOrId.isEmpty()) return null;
        return workflowRepository.findByName(nameOrId)
                .orElseGet(() -> {
                    if (nameOrId.matches("\\d+")) {
                        return workflowRepository.findById(Long.valueOf(nameOrId)).orElse(null);
                    }
                    return null;
                });
    }

    /** 判断 created run 是否与 pathVariable workflowName 匹配（name 或数字 id 兼容） */
    private boolean matchesWorkflow(ExternalRunRegistry.CreatedRun created, String workflowName) {
        if (created == null || workflowName == null) return false;
        if (workflowName.equals(created.workflowName())) return true;
        if (workflowName.matches("\\d+") && Long.valueOf(workflowName).equals(created.workflowId())) return true;
        return false;
    }

    private ResponseEntity<ApiError> error(int httpStatus, String code, String message) {
        return ResponseEntity.status(httpStatus).body(ApiError.error(code, message));
    }

    private String generateRunId() {
        return UUID.randomUUID().toString().substring(0, 8);
    }

    private CreateRunRequest parsePayload(String payloadJson) {
        if (payloadJson == null || payloadJson.isBlank()) return new CreateRunRequest();
        try {
            return MAPPER.readValue(payloadJson, CreateRunRequest.class);
        } catch (Exception e) {
            log.warn("[ExternalApi] 解析 payload 失败，使用默认值: {}", e.getMessage());
            return new CreateRunRequest();
        }
    }

    private WorkflowRun findRunRecord(String runId) {
        return workflowRunRepository.findAllByOrderByStartedAtDesc().stream()
                .filter(r -> runId.equals(r.getRunId()))
                .findFirst().orElse(null);
    }

    @SuppressWarnings("unchecked")
    private Map<String, Object> parseOutputs(WorkflowRun record) {
        if (record.getOutputs() == null || record.getOutputs().isBlank()) return Map.of();
        try {
            return MAPPER.readValue(record.getOutputs(), Map.class);
        } catch (Exception e) {
            return Map.of();
        }
    }

    @SuppressWarnings("unchecked")
    private Map<String, Object> parseJsonToMap(String json) {
        if (json == null || json.isBlank()) return null;
        try {
            return MAPPER.readValue(json, Map.class);
        } catch (Exception e) {
            return null;
        }
    }

    private int countLogs(String logsJson) {
        if (logsJson == null || logsJson.isBlank()) return 0;
        try {
            return MAPPER.readTree(logsJson).size();
        } catch (Exception e) {
            return 0;
        }
    }

    private List<FieldDef> extractFields(JsonNode root, String nodeType) {
        List<FieldDef> result = new ArrayList<>();
        JsonNode nodes = root.path("nodes");
        if (!nodes.isArray()) return result;
        for (JsonNode node : nodes) {
            if (!nodeType.equals(node.path("type").asText(""))) continue;
            JsonNode data = node.path("data");
            JsonNode varArr = data.path("variables");
            if (varArr.isArray()) {
                for (JsonNode v : varArr) {
                    result.add(toFieldDef(v));
                }
            }
            JsonNode fieldArr = data.path("fields");
            if (fieldArr.isArray()) {
                for (JsonNode f : fieldArr) {
                    result.add(toFieldDef(f));
                }
            }
        }
        return result;
    }

    private FieldDef toFieldDef(JsonNode n) {
        FieldDef f = new FieldDef();
        f.setName(n.path("name").asText(""));
        f.setType(n.path("type").asText("string"));
        f.setRequired(n.path("required").asBoolean(false));
        f.setDescription(n.path("description").asText(""));
        return f;
    }

    private boolean shouldIncludeInZip(Path runDir, Path p) {
        Path rel = runDir.relativize(p);
        String relStr = rel.toString().replace('\\', '/');
        if (agentIgnoreFilter.shouldIgnore(relStr, false)) return false;
        Path cur = rel;
        while (cur != null && cur.getNameCount() > 1) {
            cur = cur.getParent();
            if (cur == null) break;
            if (agentIgnoreFilter.shouldIgnore(cur.toString().replace('\\', '/'), true)) return false;
        }
        return true;
    }
}