package com.cmvr.test.flow.control;

import cn.hutool.core.bean.BeanUtil;
import com.cmvr.common.exception.GlobalException;
import com.cmvr.edge.client.service.EdgeSystemService;
import com.cmvr.test.enums.NodeTypeEnum;
import com.cmvr.test.enums.TaskStatusEnum;
import com.cmvr.test.enums.TerminalStatusEnum;
import com.cmvr.test.flow.builder.FlowGraph;
import com.cmvr.test.flow.context.TaskContext;
import com.cmvr.test.flow.context.TaskInstHolder;
import com.cmvr.test.flow.context.TaskThreadRegistry;
import com.cmvr.test.flow.runtime.engine.FlowItemExecutor;
import com.cmvr.test.flow.runtime.message.TaskNodeExecuteContext;
import com.cmvr.test.flow.runtime.message.TaskNodeExecuteMessage;
import com.cmvr.test.service.ITeNodeInstService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.stream.Collectors;

/**
 * 流程控制服务(检测项内部)
 * 终止/暂停/继续
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class FlowControlService {

    @Resource(name = "threadPoolTaskExecutor")
    private final ThreadPoolTaskExecutor executor;
    private final TaskInstHolder taskInstHolder;
    private final ITeNodeInstService nodeInstService;
    private final TaskThreadRegistry taskThreadRegistry;
    private final FlowItemExecutor flowItemExecutor;
    private final EdgeSystemService edgeSystemService;

    /**
     * 终止流程
     */
    public void stop(String instId) {

        TaskContext ctx = taskInstHolder.getContext(instId);
        if (ctx == null) {
            throw new GlobalException("任务不存在");
        }
        if (ctx.isStopped()) {
            throw new GlobalException("任务已终止，无需重复操作");
        }
        // 异步终止
        executor.execute(() -> edgeSystemService.stopAll(ctx.getTerminalId()));

        // 统一记录日志 + 设置上下文状态 + 数据库状态
        taskInstHolder.syncStatus(instId, TaskStatusEnum.STOPPED);
        taskInstHolder.unRegister(instId);
        taskThreadRegistry.interruptAll(instId);
        // 被中断的节点 状态修改成 PAUSED
        for (TaskNodeExecuteContext pendingNode : taskThreadRegistry.getPendingNodes(instId)) {
            String nodeId = pendingNode.getNode().getNodeId();
            nodeInstService.update(instId, nodeId, TaskStatusEnum.STOPPED);
        }
    }

    /**
     * 暂停任务
     */
    public void pause(String instId) {
        TaskContext ctx = taskInstHolder.getContext(instId);
        if (ctx == null || ctx.isStopped()) {
            throw new GlobalException("任务不存在或已终止，无法暂停");
        }
        if (ctx.isPaused()) {
            throw new GlobalException("任务已处于暂停状态");
        }
        // 异步停止终端
        executor.execute(() -> edgeSystemService.stopAll(ctx.getTerminalId()));
        ctx.setPaused(true);
        ctx.setTerminalStatus(TerminalStatusEnum.RUNNING);
        ctx.setStatus(TaskStatusEnum.PAUSED);
        ctx.updateTimestamp();
        // 统一记录日志 + 设置上下文状态 + 数据库状态
        taskInstHolder.syncStatus(instId, TaskStatusEnum.PAUSED);
        taskThreadRegistry.interruptAll(instId);
        // 被中断的节点 状态修改成 PAUSED
        for (TaskNodeExecuteContext pendingNode : taskThreadRegistry.getPendingNodes(instId)) {
            String nodeId = pendingNode.getNode().getNodeId();
            nodeInstService.update(instId, nodeId, TaskStatusEnum.PAUSED);
        }
    }

    /**
     * 继续任务
     */
    public void resume(String instId) {
        TaskContext ctx = taskInstHolder.getContext(instId);
        if (ctx == null) {
            throw new GlobalException("任务不存在");
        }
        if (!ctx.isPaused()) {
            throw new GlobalException("当前任务未处于暂停状态，无法继续运行");
        }

        ctx.setPaused(false);

        List<TaskNodeExecuteContext> pendingNodes = taskThreadRegistry.getPendingNodes(instId);
        List<TaskNodeExecuteContext> pendingLoops = pendingNodes.stream()
                .filter(node -> node.getNode().getNodeType().equals(NodeTypeEnum.LOOP))
                .collect(Collectors.toList());

        // 恢复节点运行
        for (TaskNodeExecuteContext pendingNode : pendingNodes) {
            FlowGraph graph = pendingNode.getGraph();

            String nodeId = pendingNode.getNode().getNodeId();
            if (pendingNode.getNode().getNodeType().equals(NodeTypeEnum.LOOP)) {
                // LOOP 节点更新日志，并且等子图中断节点执行完再唤醒 loop
                nodeInstService.update(instId, nodeId, TaskStatusEnum.RUNNING);
                continue;
            } else {
                nodeInstService.delete(instId, nodeId, pendingNode.getIterations());
            }
            if (graph.isSub()) {
                // 找到子图对应的loop节点
                TaskNodeExecuteContext loopCtx = pendingLoops.stream()
                        .filter(loopNode -> {
                            FlowGraph loopSubGraph = loopNode.getGraph().getNodeMap().get(loopNode.getNode().getNodeId()).getSubGraph();
                            return loopSubGraph.getNodeMap().containsKey(nodeId);
                        })
                        .findFirst()
                        .orElse(null);
                if (loopCtx == null) {
                    throw new GlobalException("找不到子图对应的loop节点");
                }
                // 子图中断点，先执行中断的子图节点
                executor.execute(() -> {
                    CountDownLatch latch = taskThreadRegistry.getLatch(instId, loopCtx.getNode().getNodeId());

                    // 恢复时构造新的 message，带上 loopNum 和 iterations
                    TaskNodeExecuteMessage resumeMessage = new TaskNodeExecuteMessage();
                    BeanUtil.copyProperties(pendingNode.getRootMessage(), resumeMessage);
                    resumeMessage.setLoopNum(pendingNode.getLoopIteration());              // 保留中断时的循环次数
                    resumeMessage.setLoopArray(pendingNode.getLoopArray());              // 保留中断时的循环次数
                    resumeMessage.setIterations(new ArrayList<>(pendingNode.getIterations())); // 保留中断时的路径

                    flowItemExecutor.executeNode(
                            pendingNode.getGraph(),
                            pendingNode.getNode(),
                            pendingNode.getStartNodeId(),
                            pendingNode.getStartInputDefs(),
                            resumeMessage,
                            latch::countDown,
                            pendingNode.getIterations()
                    );

                    try {
                        latch.await(); // 等待子图恢复执行完成
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                });

            } else {
                // 普通节点直接恢复
                executor.execute(() -> {
                    TaskNodeExecuteMessage resumeMessage = new TaskNodeExecuteMessage();
                    BeanUtil.copyProperties(pendingNode.getRootMessage(), resumeMessage);
                    resumeMessage.setLoopNum(pendingNode.getLoopIteration());              // 保留循环次数
                    resumeMessage.setLoopArray(pendingNode.getLoopArray());              // 保留循环次数
                    resumeMessage.setIterations(new ArrayList<>(pendingNode.getIterations())); // 保留路径

                    flowItemExecutor.executeNode(
                            pendingNode.getGraph(),
                            pendingNode.getNode(),
                            pendingNode.getStartNodeId(),
                            pendingNode.getStartInputDefs(),
                            resumeMessage,
                            pendingNode.getOnFinished(),
                            pendingNode.getIterations()
                    );
                });
            }
        }

        // 最终设置上下文和数据库状态为 RUNNING
        ctx.setTerminalStatus(TerminalStatusEnum.RUNNING);
        ctx.setStatus(TaskStatusEnum.RUNNING);
        ctx.updateTimestamp();
        taskInstHolder.syncStatus(instId, TaskStatusEnum.RUNNING);
        // 发布任务恢复事件
        taskInstHolder.syncStatusAsResumed(instId);
    }
}
