package com.agent.management.engine;

import com.agent.management.external.SseEventBus;
import com.agent.management.model.entity.Workflow;
import com.agent.management.model.entity.WorkflowRun;
import com.agent.management.repository.WorkflowRepository;
import com.agent.management.repository.WorkflowRunNodeRepository;
import com.agent.management.repository.WorkflowRunRepository;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.*;

/**
 * 工作流调度引擎
 * 解析 DAG 图，按拓扑层级逐层执行节点，通过 SSE 推送实时状态
 */
@Slf4j
@Component
public class WorkflowEngine {

    private static final ObjectMapper MAPPER = new ObjectMapper();

    private final WorkflowRepository workflowRepo;
    private final WorkflowRunRepository runRepo;
    private final WorkflowRunNodeRepository runNodeRepo;
    private final WorkflowRunDirManager runDirManager;
    private final WorkflowLevelExecutor levelExecutor;
    private final SseEventBus eventBus;
    private final ExecutorService executor = Executors.newFixedThreadPool(
            Runtime.getRuntime().availableProcessors(),
            r -> {
                Thread t = new Thread(r, "workflow-engine");
                t.setDaemon(true);
                return t;
            }
    );

    /** 工作流最大执行时间：无限制（Long.MAX_VALUE 秒 ≈ 2920 亿年，scheduler.schedule 实际永不触发） */
    private static final long MAX_EXECUTION_SECONDS = Long.MAX_VALUE;

    private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(r -> {
        Thread t = new Thread(r, "workflow-timeout");
        t.setDaemon(true);
        return t;
    });

    public WorkflowEngine(WorkflowRepository workflowRepo,
                          WorkflowRunRepository runRepo, WorkflowRunNodeRepository runNodeRepo,
                          WorkflowRunDirManager runDirManager,
                          WorkflowLevelExecutor levelExecutor, SseEventBus eventBus) {
        this.workflowRepo = workflowRepo;
        this.runRepo = runRepo;
        this.runNodeRepo = runNodeRepo;
        this.runDirManager = runDirManager;
        this.levelExecutor = levelExecutor;
        this.eventBus = eventBus;
        log.info("[WorkflowEngine] 已初始化，使用 WorkflowLevelExecutor 处理节点遍历");
    }

    /**
     * 外部 API 入口：使用调用方指定的 runId 触发执行，事件通过 SseEventBus 派发。
     * 工作目录由调用方提前创建并填充文件，inputs 中不需要 _preRunId。
     */
    public void executeForExternal(Long workflowId, String runId, Map<String, Object> inputs) {
        eventBus.register(runId);
        Future<?> future = executor.submit(() -> {
            try {
                execute(workflowId, runId, inputs, new NoopSseEmitter());
            } catch (Exception e) {
                log.error("[WorkflowEngine] 外部执行异常: {}", e.getMessage(), e);
                eventBus.publish(runId, "workflow_error", WorkflowRunEvent.workflowError(runId, e.getMessage()));
            }
        });
        scheduler.schedule(() -> {
            if (!future.isDone()) {
                future.cancel(true);
                log.warn("[WorkflowEngine] 外部执行超时: runId={}", runId);
                eventBus.publish(runId, "workflow_error", WorkflowRunEvent.workflowError(runId, "执行超时"));
            }
        }, MAX_EXECUTION_SECONDS, TimeUnit.SECONDS);
    }

    /**
     * 异步执行工作流
     */
    public void executeAsync(Long workflowId, Map<String, Object> inputs, SseEmitter emitter) {
        // 支持预生成的 runId（文件上传场景：Controller 已提前创建工作目录）
        String runId;
        if (inputs != null && inputs.containsKey("_preRunId")) {
            runId = (String) inputs.remove("_preRunId");
        } else {
            runId = UUID.randomUUID().toString().substring(0, 8);
        }
        final String finalRunId = runId;
        Future<?> future = executor.submit(() -> {
            try {
                execute(workflowId, finalRunId, inputs, emitter);
            } catch (Exception e) {
                log.error("[WorkflowEngine] 执行异常: {}", e.getMessage(), e);
                safeSend(emitter, WorkflowRunEvent.workflowError(finalRunId, e.getMessage()));
                emitter.completeWithError(e);
            }
        });

        // 超时保护：强制终止超时任务
        scheduler.schedule(() -> {
            if (!future.isDone()) {
                future.cancel(true);
                log.warn("[WorkflowEngine] 执行超时，强制终止: runId={}", finalRunId);
                safeSend(emitter, WorkflowRunEvent.workflowError(finalRunId, "执行超时"));
                emitter.completeWithError(new TimeoutException("工作流执行超时"));
            }
        }, MAX_EXECUTION_SECONDS, TimeUnit.SECONDS);
    }

    /**
     * 同步执行工作流（内部方法）
     */
    private void execute(Long workflowId, String runId, Map<String, Object> inputs, SseEmitter emitter) {
        log.info("[WorkflowEngine] 开始执行 workflowId={}, runId={}", workflowId, runId);

        // 1. 加载工作流
        Workflow wf = workflowRepo.findById(workflowId)
                .orElseThrow(() -> new IllegalArgumentException("工作流不存在: " + workflowId));

        String graphData = wf.getGraphData();
        if (graphData == null || graphData.equals("{\"nodes\":[],\"edges\":[]}")) {
            safeSend(emitter, WorkflowRunEvent.workflowError(runId, "工作流图为空"));
            emitter.complete();
            return;
        }

        // 2. 解析 DAG
        DagResolver.ResolvedDag dag = DagResolver.resolve(graphData);
        log.info("[WorkflowEngine] DAG 解析完成: {} 个节点, {} 层", dag.getNodeDataMap().size(), dag.getLevels().size());

        // 3. 创建运行记录
        WorkflowRun runRecord = new WorkflowRun();
        runRecord.setWorkflowId(workflowId);
        runRecord.setWorkflowName(wf.getName());
        runRecord.setRunId(runId);
        runRecord.setStatus("RUNNING");
        try {
            runRecord.setInputs(inputs != null ? MAPPER.writeValueAsString(inputs) : "{}");
        } catch (JsonProcessingException e) {
            runRecord.setInputs("{}");
        }
        runRecord = runRepo.save(runRecord);

        // 4. 创建工作目录（按 name 命名）
        java.nio.file.Path workingDir = runDirManager.createRunDir(workflowId, wf.getName(), runId);

        // 5. 初始化上下文
        WorkflowContext context = new WorkflowContext(workflowId, runId, workingDir, inputs);

        // 6. 委托 levelExecutor 逐层执行节点
        WorkflowLevelExecutor.ExecutionOutcome outcome;
        try {
            outcome = levelExecutor.executeByLevel(dag, context, runId, runRecord.getId(), emitter);
        } catch (Exception e) {
            // 引擎内部异常：标记为 FAILED 并持久化部分节点记录
            log.error("[WorkflowEngine] levelExecutor 抛出异常: {}", e.getMessage(), e);
            safeSend(emitter, WorkflowRunEvent.workflowError(runId, e.getMessage()));
            persistRunResult(runRecord, new WorkflowLevelExecutor.ExecutionOutcome(
                    new LinkedHashMap<>(), new ArrayList<>(), true, e.getMessage()));
            emitter.complete();
            return;
        }

        // 7. 持久化 + 推送完成/错误事件
        persistRunResult(runRecord, outcome);
        if (outcome.failed) {
            emitter.complete();
            return;
        }
        safeSend(emitter, WorkflowRunEvent.workflowComplete(runId, outcome.finalOutputs));
        emitter.complete();
        log.info("[WorkflowEngine] 执行完成, runId={}, 输出变量数={}", runId, outcome.finalOutputs.size());
    }

    /**
     * 把执行结果回写到 WorkflowRun + 节点记录表
     */
    private void persistRunResult(WorkflowRun runRecord, WorkflowLevelExecutor.ExecutionOutcome outcome) {
        runRecord.setStatus(outcome.failed ? "FAILED" : "SUCCESS");
        runRecord.setCompletedAt(LocalDateTime.now());
        if (outcome.failed) {
            runRecord.setError(outcome.errorMsg);
        } else {
            try {
                runRecord.setOutputs(outcome.finalOutputs.isEmpty() ? null : MAPPER.writeValueAsString(outcome.finalOutputs));
            } catch (JsonProcessingException e) {
                log.warn("[WorkflowEngine] 序列化输出失败", e);
            }
        }
        runRepo.save(runRecord);
        runNodeRepo.saveAll(outcome.nodeRecords);
    }

    /**
     * 安全发送 SSE 事件
     */
    private void safeSend(SseEmitter emitter, WorkflowRunEvent event) {
        // 同时发布到事件总线（外部 API 通过 /stream 订阅消费）
        eventBus.publish(event.getRunId(), event.getType(), event);
        try {
            emitter.send(SseEmitter.event()
                    .name(event.getType())
                    .data(event));
        } catch (IOException e) {
            log.warn("[WorkflowEngine] SSE 推送失败: {}", e.getMessage());
        }
    }
}
