package com.easyagents.flow.core.test;

import com.easyagents.flow.core.chain.Chain;
import com.easyagents.flow.core.chain.ChainDefinition;
import com.easyagents.flow.core.chain.ChainState;
import com.easyagents.flow.core.chain.ChainStatus;
import com.easyagents.flow.core.chain.EventManager;
import com.easyagents.flow.core.chain.NodeState;
import com.easyagents.flow.core.chain.NodeStatus;
import com.easyagents.flow.core.chain.repository.InMemoryChainStateRepository;
import com.easyagents.flow.core.chain.repository.InMemoryNodeStateRepository;
import com.easyagents.flow.core.node.StartNode;
import com.easyagents.flow.core.chain.runtime.InMemoryTriggerStore;
import com.easyagents.flow.core.chain.runtime.Trigger;
import com.easyagents.flow.core.chain.runtime.TriggerScheduler;
import org.junit.Assert;
import org.junit.Test;

import java.util.Collections;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 验证入口触发器先行持久化和崩溃重放协议。
 */
public class ChainRecoverableStartTest {

    /**
     * 验证 RUNNING 状态提交前已经存在入口意图，且 RUNNING 节点允许重放。
     */
    @Test
    public void shouldPersistIntentBeforeRunningAndReplayInterruptedStart() {
        InMemoryChainStateRepository stateRepository =
                new InMemoryChainStateRepository();
        ObservingTriggerStore triggerStore =
                new ObservingTriggerStore(stateRepository, "recoverable-1");
        ScheduledExecutorService schedulerExecutor =
                Executors.newSingleThreadScheduledExecutor();
        java.util.concurrent.ExecutorService worker =
                Executors.newSingleThreadExecutor();
        TriggerScheduler scheduler = new TriggerScheduler(
                triggerStore, schedulerExecutor, worker, 60_000L);
        try {
            Chain chain = chain(
                    stateRepository, triggerStore, scheduler);
            chain.start(Collections.singletonMap("value", "v"));

            Assert.assertEquals(
                    ChainStatus.READY,
                    triggerStore.statusObservedAtFirstSave);
            ChainState running = stateRepository.load("recoverable-1");
            Assert.assertEquals(ChainStatus.RUNNING, running.getStatus());
            Trigger pending = triggerStore.find(
                    triggerStore.lastTriggerId);
            Assert.assertNotNull(pending);

            NodeState startState = chain.getNodeState("start");
            startState.setStatus(NodeStatus.RUNNING);
            startState.getExecuteCount().incrementAndGet();
            chain.start(Collections.emptyMap());

            Assert.assertNotNull(
                    "interrupted RUNNING start must remain replayable",
                    triggerStore.find(triggerStore.lastTriggerId));
        } finally {
            scheduler.shutdown();
        }
    }

    /**
     * 验证入口意图保存后、RUNNING 提交前崩溃时，真实调度消费可恢复变量并继续执行。
     *
     * @throws Exception 等待异步调度被中断
     */
    @Test
    public void shouldRecoverReadyStartThroughRealScheduler()
            throws Exception {
        InMemoryChainStateRepository stateRepository =
                new InMemoryChainStateRepository();
        FailAfterSaveTriggerStore triggerStore =
                new FailAfterSaveTriggerStore();
        ScheduledExecutorService schedulerExecutor =
                Executors.newSingleThreadScheduledExecutor();
        java.util.concurrent.ExecutorService worker =
                Executors.newSingleThreadExecutor();
        TriggerScheduler scheduler = new TriggerScheduler(
                triggerStore, schedulerExecutor, worker, 1_000L);
        try {
            Chain chain = chain(
                    stateRepository, triggerStore, scheduler,
                    "recoverable-crash");
            StartNode startNode = (StartNode) chain.getDefinition()
                    .getNodeById("start");
            scheduler.registerConsumer(
                    (trigger, ignored) ->
                            chain.executeNode(startNode, trigger));

            try {
                chain.start(Collections.singletonMap(
                        "value", "persisted-input"));
                Assert.fail("simulated crash must interrupt start");
            } catch (SimulatedCrashException expected) {
                // 触发器已经保存，异常模拟进程在 RUNNING 提交前退出。
            }
            Assert.assertEquals(
                    ChainStatus.READY,
                    stateRepository.load(
                            "recoverable-crash").getStatus());

            long deadline = System.nanoTime()
                    + TimeUnit.SECONDS.toNanos(5L);
            while (System.nanoTime() < deadline) {
                ChainState recovered = stateRepository.load(
                        "recoverable-crash");
                NodeState nodeState = chain.getNodeState("start");
                if (recovered.getStatus() != ChainStatus.READY
                        && nodeState.getStatus()
                        == NodeStatus.SUCCEEDED) {
                    break;
                }
                Thread.sleep(25L);
            }

            ChainState recovered = stateRepository.load(
                    "recoverable-crash");
            Assert.assertNotEquals(
                    ChainStatus.READY, recovered.getStatus());
            Assert.assertEquals(
                    "persisted-input",
                    recovered.getMemory().get("value"));
            Assert.assertEquals(
                    NodeStatus.SUCCEEDED,
                    chain.getNodeState("start").getStatus());
            Assert.assertNull(triggerStore.find(
                    triggerStore.lastTriggerId));
        } finally {
            scheduler.shutdown();
        }
    }

    /**
     * 创建仅含入口节点的测试链。
     *
     * @param stateRepository 状态仓储
     * @param triggerStore 触发器仓储
     * @param scheduler 调度器
     * @return 已配置链
     */
    private Chain chain(
            InMemoryChainStateRepository stateRepository,
            InMemoryTriggerStore triggerStore,
            TriggerScheduler scheduler) {
        ChainDefinition definition = new ChainDefinition();
        definition.setId("recoverable-definition");
        StartNode start = new StartNode();
        start.setId("start");
        definition.addNode(start);
        definition.setEdges(Collections.emptyList());

        Chain chain = new Chain(definition, "recoverable-1");
        chain.setChainStateRepository(stateRepository);
        chain.setNodeStateRepository(
                new InMemoryNodeStateRepository());
        chain.setTriggerScheduler(scheduler);
        chain.setEventManager(new EventManager());
        return chain;
    }

    /**
     * 创建指定实例 ID 的仅入口测试链。
     *
     * @param stateRepository 状态仓储
     * @param triggerStore 触发器仓储
     * @param scheduler 调度器
     * @param instanceId 实例 ID
     * @return 已配置链
     */
    private Chain chain(
            InMemoryChainStateRepository stateRepository,
            InMemoryTriggerStore triggerStore,
            TriggerScheduler scheduler,
            String instanceId) {
        ChainDefinition definition = new ChainDefinition();
        definition.setId("recoverable-definition");
        StartNode start = new StartNode();
        start.setId("start");
        definition.addNode(start);
        definition.setEdges(Collections.emptyList());

        Chain chain = new Chain(definition, instanceId);
        chain.setChainStateRepository(stateRepository);
        chain.setNodeStateRepository(
                new InMemoryNodeStateRepository());
        chain.setTriggerScheduler(scheduler);
        chain.setEventManager(new EventManager());
        return chain;
    }

    /**
     * 记录首次触发器保存时的实例状态。
     */
    private static final class ObservingTriggerStore
            extends InMemoryTriggerStore {

        private final InMemoryChainStateRepository stateRepository;
        private final String instanceId;
        private ChainStatus statusObservedAtFirstSave;
        private String lastTriggerId;

        /**
         * 创建观察仓储。
         *
         * @param stateRepository 状态仓储
         * @param instanceId 实例 ID
         */
        private ObservingTriggerStore(
                InMemoryChainStateRepository stateRepository,
                String instanceId) {
            this.stateRepository = stateRepository;
            this.instanceId = instanceId;
        }

        /**
         * {@inheritDoc}
         */
        @Override
        public Trigger save(Trigger trigger) {
            observe(trigger);
            return super.save(trigger);
        }

        /**
         * {@inheritDoc}
         */
        @Override
        public boolean saveIfAbsent(
                Trigger trigger) {
            observe(trigger);
            return super.saveIfAbsent(trigger);
        }

        /**
         * 记录首次稳定入口保存时的状态。
         *
         * @param trigger 待保存触发器
         */
        private void observe(Trigger trigger) {
            if (statusObservedAtFirstSave == null) {
                ChainState state = stateRepository.load(instanceId);
                statusObservedAtFirstSave =
                        state == null ? null : state.getStatus();
            }
            lastTriggerId = trigger.getId();
        }
    }

    /**
     * 首次保存成功后抛错，用于模拟状态提交前进程退出。
     */
    private static final class FailAfterSaveTriggerStore
            extends InMemoryTriggerStore {

        private boolean fail = true;
        private String lastTriggerId;

        /**
         * {@inheritDoc}
         */
        @Override
        public Trigger save(Trigger trigger) {
            Trigger saved = super.save(trigger);
            failAfterFirstSave(trigger);
            return saved;
        }

        /**
         * {@inheritDoc}
         */
        @Override
        public boolean saveIfAbsent(
                Trigger trigger) {
            boolean saved =
                    super.saveIfAbsent(trigger);
            if (saved) {
                failAfterFirstSave(trigger);
            }
            return saved;
        }

        /**
         * 首次成功保存后抛出模拟崩溃。
         *
         * @param trigger 已保存触发器
         */
        private void failAfterFirstSave(
                Trigger trigger) {
            lastTriggerId = trigger.getId();
            if (fail) {
                fail = false;
                throw new SimulatedCrashException();
            }
        }
    }

    /**
     * 测试专用崩溃异常。
     */
    private static final class SimulatedCrashException
            extends RuntimeException {
        private static final long serialVersionUID = 1L;
    }
}
