package executor

import (
	"fmt"
	"os"
	"sync"
	"testing"
	"time"

	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/actions"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/dag"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/storage"
)

type MockBuiltinAction struct {
	*actions.MockActionBase
	fn func()
}

func (action *MockBuiltinAction) Invoke(args actions.ActionArgs) error {
	err := action.MockInvoke()
	if err != nil {
		return err
	}
	action.fn()
	return nil
}

func TestExecutorBasic(t *testing.T) {
	logicPlans := dag.NewPlans()
	actionLib := actions.NewActionLibrary()

	result1 := NewSafeString()
	regAction1 := func(char string) {
		actionLib.SafeRegister(&MockBuiltinAction{
			MockActionBase: &actions.MockActionBase{
				ActionName: char,
			},
			fn: func() {
				result1.Append(char)
			},
		})
	}
	regAction1("a")
	regAction1("b")
	regAction1("c")
	logicPlans.NewPlanWithRootNode("test1").
		SafeAddDependNode("a").
		SafeAddDependNode("b").
		SafeAddDependNode("c")
	expected1 := "cba"

	result2 := NewSafeString()
	regAction2 := func(char string) {
		actionLib.SafeRegister(&MockBuiltinAction{
			MockActionBase: &actions.MockActionBase{
				ActionName: char,
			},
			fn: func() {
				result2.Append(char)
			},
		})
	}
	regAction2("11")
	regAction2("22")
	regAction2("33")
	logicPlans.NewPlanWithRootNode("test2").
		SafeAddDependNode("11").
		SafeAddDependNode("22").
		SafeAddDependNode("33")
	expected2 := "332211"

	result3 := NewSafeString()
	regAction3 := func(char string) {
		actionLib.SafeRegister(&MockBuiltinAction{
			MockActionBase: &actions.MockActionBase{
				MockElapsed: time.Second / 10,
				ActionName:  char,
			},
			fn: func() {
				result3.Append(char)
			},
		})
	}
	regAction3("x")
	regAction3("y")
	regAction3("z")
	regAction3("m")
	regAction3("n")
	regAction3("o")
	logicPlans.NewPlanWithRootNode("test3").
		SafeAddDependNode("x").
		SafeAddDependNode("y").
		SafeAddDependNode("z").
		SafeAddDependNode("m").
		SafeAddDependNode("n").
		SafeAddDependNode("o")
	expected3 := "onmzyx"

	exec, err := NewExecutor(actionLib, logicPlans, nil, nil, 10)
	if err != nil {
		t.Fatal(err)
	}

	logger := NewGeneralLogger("TestExecutorBasic", os.Stderr)
	exec.RegisterEventReceiver(logger)
	asyncLogger := NewGeneralLogger("TestExecutorBasicAsyncLog", os.Stderr)
	exec.RegisterAsyncEventReceiver(asyncLogger)
	metrics := NewMetricsRecorder("TestExecutorBasic")
	exec.RegisterAsyncEventReceiver(metrics)

	err = exec.Submit("test1", actions.ActionArgs{}, "test-exec-id1")
	if err != nil {
		t.Fatal(err)
	}
	err = exec.Submit("test2", actions.ActionArgs{}, "test-exec-id2")
	if err != nil {
		t.Fatal(err)
	}
	err = exec.Submit("test3", actions.ActionArgs{}, "test-exec-id3")
	if err != nil {
		t.Fatal(err)
	}
	err = exec.Run(ExitPolicyStopOnFinish)
	if err != nil {
		t.Fatal(err)
	}

	if result1.Load() != expected1 {
		t.Fatal(fmt.Errorf("result should be `%v` but got `%v`", expected1, result1.Load()))
	}
	if result2.Load() != expected2 {
		t.Fatal(fmt.Errorf("result should be `%v` but got `%v`", expected2, result2.Load()))
	}
	if result3.Load() != expected3 {
		t.Fatal(fmt.Errorf("result should be `%v` but got `%v`", expected3, result3.Load()))
	}
}

func TestExecutorWithErrors(t *testing.T) {
	logicPlans := dag.NewPlans()
	actionLib := actions.NewActionLibrary()

	rootName := "test"
	execId := "test-exec-id"
	result := NewSafeString()
	regAction := func(char string) {
		actionLib.SafeRegister(&MockBuiltinAction{
			MockActionBase: &actions.MockActionBase{
				MockElapsed:   time.Second / 10,
				MockErrorRate: 0.8,
				ActionName:    char,
			},
			fn: func() {
				result.Append(char)
			},
		})
	}
	regAction("x")
	regAction("y")
	regAction("z")
	regAction("m")
	regAction("n")
	regAction("o")
	logicPlans.NewPlanWithRootNode(rootName).
		SafeAddDependNode("x").
		SafeAddDependNode("y").
		SafeAddDependNode("z").
		SafeAddDependNode("m").
		SafeAddDependNode("n").
		SafeAddDependNode("o")
	expected := "onmzyx"

	exec, err := NewExecutor(actionLib, logicPlans, nil, nil, 10)
	if err != nil {
		t.Fatal(err)
	}

	logger := NewGeneralLogger("TestExecutorWithErrors", os.Stderr)
	exec.RegisterEventReceiver(logger)

	err = exec.Submit(rootName, actions.ActionArgs{}, execId)
	if err != nil {
		t.Fatal(err)
	}
	err = exec.Run(ExitPolicyStopOnFinish)
	if err != nil {
		t.Fatal(err)
	}

	if result.Load() != expected {
		t.Fatal(fmt.Errorf("result should be `%v` but got `%v`", expected, result.Load()))
	}
}

func TestExecutorServiceStopNowDontWait(t *testing.T) {
	logicPlans := dag.NewPlans()
	actionLib := actions.NewActionLibrary()
	actionLib.SafeRegister(&actions.MockActionBase{
		ActionName:  "test-long-exec",
		MockElapsed: time.Second * 10,
	})
	logicPlans.NewPlanWithRootNode("test-long-exec")
	exec, err := NewExecutor(actionLib, logicPlans, nil, nil, 10)
	if err != nil {
		t.Fatal(err)
	}

	logger := NewGeneralLogger("TestExecutorServiceStopNowDontWait", os.Stderr)
	exec.RegisterEventReceiver(logger)
	exec.RegisterEventReceiver(NewDragSlow(time.Second / 50))

	err = exec.Start(ExitPolicyStopOnFinish)
	if err != nil {
		t.Fatal(err)
	}

	err = exec.Submit("test-long-exec", actions.ActionArgs{}, "test-exec-id")
	if err != nil {
		t.Fatal(err)
	}

	err = exec.Stop(ExitPolicyStopNowDontWait)
	if err != nil {
		t.Fatal(err)
	}
}

func TestExecutorServiceStopNowWaitForRunningActions(t *testing.T) {
	logicPlans := dag.NewPlans()
	actionLib := actions.NewActionLibrary()

	rootName := "test"
	execId := "test-exec-id"
	result := NewSafeString()
	regAction := func(char string) {
		actionLib.SafeRegister(&MockBuiltinAction{
			MockActionBase: &actions.MockActionBase{
				ActionName:  char,
				MockElapsed: time.Second * 2,
			},
			fn: func() {
				result.Append(char)
			},
		})
	}
	regAction("i2")
	regAction("l2")
	regAction("k2")
	logicPlans.NewPlanWithRootNode(rootName).
		SafeAddDependNode("i2").
		SafeAddDependNode("l2").
		SafeAddDependNode("k2")

	exec, err := NewExecutor(actionLib, logicPlans, nil, nil, 10)
	if err != nil {
		t.Fatal(err)
	}

	logger := NewGeneralLogger("TestExecutorServiceStopNowWaitForRunningActions", os.Stderr)
	exec.RegisterEventReceiver(logger)

	err = exec.Start(ExitPolicyNonStop)
	if err != nil {
		t.Fatal(err)
	}

	err = exec.Submit(rootName, actions.ActionArgs{}, execId)
	if err != nil {
		t.Fatal(err)
	}

	err = exec.Stop(ExitPolicyStopNowWaitForRunningActions)
	if err != nil {
		t.Fatal(err)
	}

	expected := "k2"

	if result.Load() != expected {
		t.Fatal(fmt.Errorf("result should be `%v` but got `%v`", expected, result.Load()))
	}
}

func TestExecutorServiceMode(t *testing.T) {
	logicPlans := dag.NewPlans()
	actionLib := actions.NewActionLibrary()

	rootName := "test"
	execId := "test-exec-id"
	result := NewSafeString()
	regAction := func(char string) {
		actionLib.SafeRegister(&MockBuiltinAction{
			MockActionBase: &actions.MockActionBase{
				ActionName: char,
			},
			fn: func() {
				result.Append(char)
			},
		})
	}
	regAction("i")
	regAction("l")
	regAction("k")
	logicPlans.NewPlanWithRootNode(rootName).
		SafeAddDependNode("i").
		SafeAddDependNode("l").
		SafeAddDependNode("k")
	expected := "kli"

	exec, err := NewExecutor(actionLib, logicPlans, nil, nil, 10)
	if err != nil {
		t.Fatal(err)
	}

	logger := NewGeneralLogger("TestExecutorServiceMode", os.Stderr)
	exec.RegisterEventReceiver(logger)

	err = exec.Start(ExitPolicyStopOnFinish)
	if err != nil {
		t.Fatal(err)
	}

	err = exec.Submit(rootName, actions.ActionArgs{}, execId)
	if err != nil {
		t.Fatal(err)
	}

	err = exec.Wait()
	if err != nil {
		t.Fatal(err)
	}

	if result.Load() != expected {
		t.Fatal(fmt.Errorf("result should be `%v` but got `%v`", expected, result.Load()))
	}
}

func TestExecutorReboot(t *testing.T) {
	logicPlans := dag.NewPlans()
	actionLib := actions.NewActionLibrary()

	rootName := "test"
	execId := "test-exec-id"
	result := NewSafeString()
	regAction := func(char string) {
		actionLib.SafeRegister(&MockBuiltinAction{
			MockActionBase: &actions.MockActionBase{
				ActionName:  char,
				MockElapsed: time.Second * 2,
			},
			fn: func() {
				result.Append(char)
			},
		})
	}
	regAction(rootName)
	regAction("l3")
	logicPlans.NewPlanWithRootNode(rootName).
		SafeAddDependNode("l3")

	stg := storage.NewMemKvStg()
	execKvStg := storage.NewPrefixWrapKvStg([]string{"executor"}, stg)
	execStatusStg := storage.NewKvBasedExecStatusStg(execKvStg)
	dagKvStg := storage.NewPrefixWrapKvStg([]string{"dags"}, stg)
	dagStatusStg := storage.NewKvBasedDagStatusStg(dagKvStg)

	exec, err := NewExecutor(actionLib, logicPlans, execStatusStg, dagStatusStg, 10)
	if err != nil {
		t.Fatal(err)
	}

	logger := NewGeneralLogger("TestExecutorReboot1", os.Stderr)
	exec.RegisterEventReceiver(logger)

	err = exec.Start(ExitPolicyNonStop)
	if err != nil {
		t.Fatal(err)
	}

	err = exec.Submit(rootName, actions.ActionArgs{}, execId)
	if err != nil {
		t.Fatal(err)
	}

	err = exec.Stop(ExitPolicyStopNowWaitForRunningActions)
	if err != nil {
		t.Fatal(err)
	}

	expected := "l3"

	if result.Load() != expected {
		t.Fatal(fmt.Errorf("result should be `%v` but got `%v`", expected, result.Load()))
	}

	// reboot with old storage, empty plans (rely on plan reload)
	logicPlans2 := dag.NewPlans()
	exec2, err := NewExecutor(actionLib, logicPlans2, execStatusStg, dagStatusStg, 10)
	if err != nil {
		t.Fatal(err)
	}
	logger2 := NewGeneralLogger("TestExecutorReboot2", os.Stderr)
	exec2.RegisterEventReceiver(logger2)

	err = exec2.Start(ExitPolicyNonStop)
	if err != nil {
		t.Fatal(err)
	}
	err = exec2.Stop(ExitPolicyStopOnFinish)
	if err != nil {
		t.Fatal(err)
	}
	expected = "l3" + rootName
	if result.Load() != expected {
		t.Fatal(fmt.Errorf("result should be `%v` but got `%v`", expected, result.Load()))
	}
}

func TestExecutorEventFilter(t *testing.T) {
	logicPlans := dag.NewPlans()
	actionLib := actions.NewActionLibrary()

	rootName := "test"
	execId := "test-exec-id"
	result := NewSafeString()
	regAction := func(char string) {
		actionLib.SafeRegister(&MockBuiltinAction{
			MockActionBase: &actions.MockActionBase{
				ActionName: char,
			},
			fn: func() {
				result.Append(char)
			},
		})
	}
	regAction("a")
	regAction("b")
	logicPlans.NewPlanWithRootNode(rootName).
		SafeAddDependNode("a").
		SafeAddDependNode("b")
	expected := "ba"

	exec, err := NewExecutor(actionLib, logicPlans, nil, nil, 10)
	if err != nil {
		t.Fatal(err)
	}

	actionLogger := NewGeneralLogger("action logger", os.Stderr)
	exec.RegisterEventReceiver(NewEventAllowFilter(actionLogger, EventExecutorRunLoopOnActionResult))
	otherLogger := NewGeneralLogger("other logger", os.Stderr)
	exec.RegisterEventReceiver(NewEventDenyFilter(otherLogger, EventExecutorRunLoopOnActionResult))

	err = exec.Submit(rootName, actions.ActionArgs{}, execId)
	if err != nil {
		t.Fatal(err)
	}
	err = exec.Run(ExitPolicyStopOnFinish)
	if err != nil {
		t.Fatal(err)
	}

	if result.Load() != expected {
		t.Fatal(fmt.Errorf("result should be `%v` but got `%v`", expected, result.Load()))
	}
}

func TestExecutorSubmitFromAction(t *testing.T) {
	logicPlans := dag.NewPlans()
	actionLib := actions.NewActionLibrary()

	result1 := NewSafeString()
	regAction1 := func(char string) {
		actionLib.SafeRegister(&MockBuiltinAction{
			MockActionBase: &actions.MockActionBase{
				ActionName: char,
			},
			fn: func() {
				result1.Append(char)
			},
		})
	}
	regAction1("a")
	regAction1("b")
	regAction1("c")
	logicPlans.NewPlanWithRootNode("test1").
		SafeAddDependNode("a").
		SafeAddDependNode("b").
		SafeAddDependNode("c")
	expected1 := "cba"

	var exec *Executor

	result2 := NewSafeString()
	regAction2 := func(char string) {
		actionLib.SafeRegister(&MockBuiltinAction{
			MockActionBase: &actions.MockActionBase{
				MockElapsed: time.Second / 10,
				ActionName:  char,
			},
			fn: func() {
				result2.Append(char)
			},
		})
	}
	regAction2("x")
	actionLib.SafeRegister(&MockBuiltinAction{
		MockActionBase: &actions.MockActionBase{
			MockElapsed: time.Second / 10,
			ActionName:  "y",
		},
		fn: func() {
			err := exec.Submit("test1", actions.ActionArgs{}, "test-exec-id1")
			if err != nil {
				t.Fatal(err)
			}
			result2.Append("y")
		},
	})
	regAction2("z")
	logicPlans.NewPlanWithRootNode("test2").
		SafeAddDependNode("x").
		SafeAddDependNode("y").
		SafeAddDependNode("z")
	expected2 := "zyx"

	var err error
	exec, err = NewExecutor(actionLib, logicPlans, nil, nil, 10)
	if err != nil {
		t.Fatal(err)
	}

	logger := NewGeneralLogger("TestExecutorSubmitFromAction", os.Stderr)
	exec.RegisterEventReceiver(logger)

	err = exec.Start(ExitPolicyStopOnFinish)
	if err != nil {
		t.Fatal(err)
	}
	err = exec.Submit("test2", actions.ActionArgs{}, "test-exec-id2")
	if err != nil {
		t.Fatal(err)
	}
	err = exec.Stop(ExitPolicyStopOnFinish)
	if err != nil {
		t.Fatal(err)
	}

	if result1.Load() != expected1 {
		t.Fatal(fmt.Errorf("result should be `%v` but got `%v`", expected1, result1.Load()))
	}
	if result2.Load() != expected2 {
		t.Fatal(fmt.Errorf("result should be `%v` but got `%v`", expected2, result2.Load()))
	}
}

func TestExecutorDependency(t *testing.T) {
	test := func() {
		logicPlans := dag.NewPlans()
		actionLib := actions.NewActionLibrary()

		rootName := "test"
		execId := "test-exec-id"
		result := NewSafeString()
		regAction := func(char string) {
			actionLib.SafeRegister(&MockBuiltinAction{
				MockActionBase: &actions.MockActionBase{
					ActionName: char,
				},
				fn: func() {
					result.Append(char)
				},
			})
		}
		regAction("i")
		regAction("l")
		regAction("k")
		nodeI := logicPlans.NewPlanWithRootNode(rootName).
			SafeAddDependNode("i")
		nodeI.SafeAddDependNode("l")
		nodeI.SafeAddDependNode("k")
		expected1 := "kli"
		expected2 := "lki"

		exec, err := NewExecutor(actionLib, logicPlans, nil, nil, 10)
		if err != nil {
			t.Fatal(err)
		}

		logger := NewGeneralLogger("TestExecutorDependency", os.Stderr)
		exec.RegisterEventReceiver(logger)

		err = exec.Start(ExitPolicyStopOnFinish)
		if err != nil {
			t.Fatal(err)
		}

		err = exec.Submit(rootName, actions.ActionArgs{}, execId)
		if err != nil {
			t.Fatal(err)
		}

		err = exec.Wait()
		if err != nil {
			t.Fatal(err)
		}
		if result.Load() != expected1 && result.Load() != expected2 {
			t.Fatal(fmt.Errorf("result should be `%v` or `%v` but got `%v`", expected1, expected2, result.Load()))
		}
	}

	for i := 0; i < 10; i++ {
		test()
	}
}

func TestExecutorRunningStatus(t *testing.T) {
	logicPlans := dag.NewPlans()
	actionLib := actions.NewActionLibrary()

	actionTimeCost := time.Second / 2

	rootName := "test"
	execId := "test-exec-id"
	result := NewSafeString()

	// Use channels to synchronize action execution
	lStarted := make(chan struct{})
	lCanFinish := make(chan struct{})
	iStarted := make(chan struct{})
	iCanFinish := make(chan struct{})

	// Register action "l" with synchronization
	actionLib.SafeRegister(&MockBuiltinAction{
		MockActionBase: &actions.MockActionBase{
			ActionName:  "l",
			MockElapsed: 0, // We control timing via channels
		},
		fn: func() {
			close(lStarted)
			<-lCanFinish
			result.Append("l")
		},
	})

	// Register action "i" with synchronization
	actionLib.SafeRegister(&MockBuiltinAction{
		MockActionBase: &actions.MockActionBase{
			ActionName:  "i",
			MockElapsed: 0, // We control timing via channels
		},
		fn: func() {
			close(iStarted)
			<-iCanFinish
			result.Append("i")
		},
	})

	logicPlans.NewPlanWithRootNode(rootName).
		SafeAddDependNode("i").
		SafeAddDependNode("l")
	expected := "li"

	exec, err := NewExecutor(actionLib, logicPlans, nil, nil, 10)
	if err != nil {
		t.Fatal(err)
	}

	logger := NewGeneralLogger("TestExecutorRunningStatus", os.Stderr)
	exec.RegisterEventReceiver(logger)

	err = exec.Start(ExitPolicyStopOnFinish)
	if err != nil {
		t.Fatal(err)
	}

	err = exec.Submit(rootName, actions.ActionArgs{}, execId)
	if err != nil {
		t.Fatal(err)
	}

	// Wait for action "l" to start
	select {
	case <-lStarted:
	case <-time.After(5 * time.Second):
		t.Fatal("timeout waiting for action 'l' to start")
	}

	// Small delay to let status update
	time.Sleep(50 * time.Millisecond)

	status := exec.GetRunningStatus(execId)
	if status == nil {
		t.Fatal(fmt.Errorf("submitted task `%v` doesn't have status", rootName))
	}
	if len(status.ActionsStatus) != 2 {
		t.Fatalf("submitted task `%v` should have 2 action statuses but got `%v`", rootName, status.ActionsStatus)
	}
	checkActionStatus(t, status, "i", false, false, false)
	checkActionStatus(t, status, "l", true, true, false)

	statusMap, err := exec.QueryRunnings(rootName+"-useless-subfix", func(args actions.ActionArgs) (bool, error) {
		return true, nil
	})
	if err != nil {
		t.Fatal(err)
	}
	if len(statusMap) != 0 {
		t.Fatal(fmt.Errorf("QueryRunnings should return empty"))
	}
	statusMap, err = exec.QueryRunnings(rootName, func(args actions.ActionArgs) (bool, error) {
		return true, nil
	})
	if err != nil {
		t.Fatal(err)
	}
	status = statusMap[execId]
	if status == nil {
		t.Fatal(fmt.Errorf("submitted exec-id `%v` not in map (%v)", rootName, statusMap))
	}
	if len(status.ActionsStatus) != 2 {
		t.Fatalf("submitted task `%v` should have 2 action statuses but got `%v`", rootName, status.ActionsStatus)
	}
	checkActionStatus(t, status, "i", false, false, false)
	checkActionStatus(t, status, "l", true, true, false)

	// Allow action "l" to finish
	close(lCanFinish)

	// Wait for action "i" to start
	select {
	case <-iStarted:
	case <-time.After(5 * time.Second):
		t.Fatal("timeout waiting for action 'i' to start")
	}

	// Small delay to let status update
	time.Sleep(50 * time.Millisecond)

	status = exec.GetRunningStatus(execId)
	checkActionStatus(t, status, "i", true, true, false)
	checkActionStatus(t, status, "l", false, true, true)

	// Allow action "i" to finish
	close(iCanFinish)

	// Wait for completion with timeout
	done := make(chan struct{})
	go func() {
		for {
			localStatus := exec.GetRunningStatus(execId)
			if localStatus == nil {
				close(done)
				return
			}
			actionI := localStatus.ActionsStatus.SafeGetActionTaskStatus("i")
			if actionI != nil && actionI.Elapsed > 0 {
				close(done)
				return
			}
			time.Sleep(10 * time.Millisecond)
		}
	}()

	select {
	case <-done:
	case <-time.After(5 * time.Second):
		t.Fatal("timeout waiting for actions to complete")
	}

	// Give time for task to be removed from running list
	time.Sleep(actionTimeCost)

	statusMap, err = exec.QueryRunnings(rootName, func(args actions.ActionArgs) (bool, error) {
		return true, nil
	})
	if err != nil {
		t.Fatal(err)
	}
	if len(statusMap) != 0 {
		t.Fatal(fmt.Errorf("QueryRunnings should return empty"))
	}

	if result.Load() != expected {
		t.Fatal(fmt.Errorf("result should be `%v` but got `%v`", expected, result.Load()))
	}
}

func TestExecutorAbort(t *testing.T) {
	logicPlans := dag.NewPlans()
	actionLib := actions.NewActionLibrary()

	actionTimeCost := time.Second / 2

	rootName := "test"
	execId := "test-exec-id"
	result := NewSafeString()
	regAction := func(char string) {
		actionLib.SafeRegister(&MockBuiltinAction{
			MockActionBase: &actions.MockActionBase{
				ActionName:  char,
				MockElapsed: actionTimeCost,
			},
			fn: func() {
				result.Append(char)
			},
		})
	}
	regAction("i")
	regAction("l")
	logicPlans.NewPlanWithRootNode(rootName).
		SafeAddDependNode("i").
		SafeAddDependNode("l")
	expected := "l"

	exec, err := NewExecutor(actionLib, logicPlans, nil, nil, 10)
	if err != nil {
		t.Fatal(err)
	}

	logger := NewGeneralLogger("TestExecutorAbort", os.Stderr)
	exec.RegisterEventReceiver(logger)

	err = exec.Start(ExitPolicyStopOnFinish)
	if err != nil {
		t.Fatal(err)
	}

	err = exec.Submit(rootName, actions.ActionArgs{}, execId)
	if err != nil {
		t.Fatal(err)
	}

	time.Sleep(actionTimeCost / 2)
	status := exec.GetRunningStatus(execId)
	if status == nil {
		t.Fatal(fmt.Errorf("submitted task `%v` doesn't have status", rootName))
	}
	if len(status.ActionsStatus) != 2 {
		t.Fatalf("submitted task `%v` should have 2 action statuses but got `%v`", rootName, status.ActionsStatus)
	}
	checkActionStatus(t, status, "i", false, false, false)
	checkActionStatus(t, status, "l", true, true, false)

	err = exec.Abort(execId)
	if err != nil {
		t.Fatal(err)
	}

	time.Sleep(actionTimeCost)
	status = exec.GetRunningStatus(execId)
	checkActionStatus(t, status, "i", false, false, false)
	checkActionStatus(t, status, "l", false, true, true)

	time.Sleep(actionTimeCost)
	status = exec.GetRunningStatus(execId)
	checkActionStatus(t, status, "i", false, false, false)
	checkActionStatus(t, status, "l", false, true, true)

	statusMap, err := exec.QueryRunnings(rootName, func(args actions.ActionArgs) (bool, error) {
		return true, nil
	})
	if err != nil {
		t.Fatal(err)
	}
	if len(statusMap) != 0 {
		t.Fatal(fmt.Errorf("QueryRunnings should return empty"))
	}

	if result.Load() != expected {
		t.Fatal(fmt.Errorf("result should be `%v` but got `%v`", expected, result.Load()))
	}
}

func TestExecutorAbortOnFinished(t *testing.T) {
	logicPlans := dag.NewPlans()
	actionLib := actions.NewActionLibrary()

	actionTimeCost := time.Second / 2

	rootName := "test"
	execId := "test-exec-id"
	result := NewSafeString()
	regAction := func(char string) {
		actionLib.SafeRegister(&MockBuiltinAction{
			MockActionBase: &actions.MockActionBase{
				ActionName:  char,
				MockElapsed: actionTimeCost,
			},
			fn: func() {
				result.Append(char)
			},
		})
	}
	regAction("i")
	regAction("l")
	logicPlans.NewPlanWithRootNode(rootName).
		SafeAddDependNode("i").
		SafeAddDependNode("l")
	expected := "l"

	exec, err := NewExecutor(actionLib, logicPlans, nil, nil, 10)
	if err != nil {
		t.Fatal(err)
	}

	logger := NewGeneralLogger("TestExecutorAbortOnFinished", os.Stderr)
	exec.RegisterEventReceiver(logger)

	err = exec.Start(ExitPolicyStopOnFinish)
	if err != nil {
		t.Fatal(err)
	}

	err = exec.Submit(rootName, actions.ActionArgs{}, execId)
	if err != nil {
		t.Fatal(err)
	}

	time.Sleep(actionTimeCost / 2)
	status := exec.GetRunningStatus(execId)
	if status == nil {
		t.Fatal(fmt.Errorf("submitted task `%v` doesn't have status", rootName))
	}
	if len(status.ActionsStatus) != 2 {
		t.Fatal(fmt.Errorf("submitted task `%v` should have 2 action statuses but got `%v`", rootName, status.ActionsStatus))
	}
	checkActionStatus(t, status, "i", false, false, false)
	checkActionStatus(t, status, "l", true, true, false)

	for i := 0; i < 100; i++ {
		err = exec.Abort(execId)
		if err != nil {
			t.Fatal(err)
		}
	}

	time.Sleep(actionTimeCost)
	status = exec.GetRunningStatus(execId)
	checkActionStatus(t, status, "i", false, false, false)
	checkActionStatus(t, status, "l", false, true, true)

	time.Sleep(actionTimeCost)
	status = exec.GetRunningStatus(execId)
	checkActionStatus(t, status, "i", false, false, false)
	checkActionStatus(t, status, "l", false, true, true)

	statusMap, err := exec.QueryRunnings(rootName, func(args actions.ActionArgs) (bool, error) {
		return true, nil
	})
	if err != nil {
		t.Fatal(err)
	}
	if len(statusMap) != 0 {
		t.Fatal(fmt.Errorf("QueryRunnings should return empty"))
	}

	for i := 0; i < 100; i++ {
		_ = exec.Abort(execId)
	}

	statusMap, err = exec.QueryRunnings(rootName, func(args actions.ActionArgs) (bool, error) {
		return true, nil
	})
	if err != nil {
		t.Fatal(err)
	}
	if len(statusMap) != 0 {
		t.Fatal(fmt.Errorf("QueryRunnings should return empty"))
	}

	if result.Load() != expected {
		t.Fatal(fmt.Errorf("result should be `%v` but got `%v`", expected, result.Load()))
	}
}

func TestExecutorAbortErrorTask(t *testing.T) {
	logicPlans := dag.NewPlans()
	actionLib := actions.NewActionLibrary()

	actionTimeCost := time.Second / 2

	rootName := "test"
	execId := "test-exec-id"
	result := NewSafeString()
	regAction := func(char string, errRate float64) {
		actionLib.SafeRegister(&MockBuiltinAction{
			MockActionBase: &actions.MockActionBase{
				ActionName:    char,
				MockElapsed:   actionTimeCost,
				MockErrorRate: errRate,
			},
			fn: func() {
				result.Append(char)
			},
		})
	}
	actionLib.SafeRegister(actions.CreateActionFromAsync(
		actions.NewMockAsyncAction("h", time.Hour*24*365),
		time.Second,
		time.Hour*24*365,
	))
	regAction("i", 1.0)
	regAction("l", 0.0)
	rootNode := logicPlans.NewPlanWithRootNode(rootName)
	rootNode.SafeAddDependNode("h")
	rootNode.SafeAddDependNode("i").
		SafeAddDependNode("l")
	expected := "l"

	exec, err := NewExecutor(actionLib, logicPlans, nil, nil, 10)
	if err != nil {
		t.Fatal(err)
	}

	logger := NewGeneralLogger("TestExecutorAbortError", os.Stderr)
	exec.RegisterEventReceiver(logger)

	err = exec.Start(ExitPolicyStopOnFinish)
	if err != nil {
		t.Fatal(err)
	}

	err = exec.Submit(rootName, actions.ActionArgs{}, execId)
	if err != nil {
		t.Fatal(err)
	}

	time.Sleep(actionTimeCost / 2)
	status := exec.GetRunningStatus(execId)
	if status == nil {
		t.Fatal(fmt.Errorf("submitted task `%v` doesn't have status", rootName))
	}
	if len(status.ActionsStatus) != 3 {
		t.Fatal(fmt.Errorf("submitted task `%v` should have 3 action statuses but got `%v`", rootName, status.ActionsStatus))
	}
	checkActionStatus(t, status, "h", true, true, false)
	checkActionStatus(t, status, "i", false, false, false)
	checkActionStatus(t, status, "l", true, true, false)

	time.Sleep(actionTimeCost * 4)
	status = exec.GetRunningStatus(execId)
	checkActionStatus(t, status, "h", true, true, false)
	checkActionStatus(t, status, "i", true, true, false)
	checkActionStatus(t, status, "l", false, true, true)

	err = exec.Abort(execId)
	if err != nil {
		t.Fatal(err)
	}

	err = exec.Stop(ExitPolicyStopOnFinish)
	if err != nil {
		t.Fatal(err)
	}

	if result.Load() != expected {
		t.Fatal(fmt.Errorf("result should be `%v` but got `%v`", expected, result.Load()))
	}
}

func TestExecutorDuplicatedSubmit(t *testing.T) {
	logicPlans := dag.NewPlans()
	actionLib := actions.NewActionLibrary()

	rootName := "test"
	execId := "test-exec-id"
	result := NewSafeString()
	regAction := func(char string) {
		actionLib.SafeRegister(&MockBuiltinAction{
			MockActionBase: &actions.MockActionBase{
				MockElapsed: time.Second / 10,
				ActionName:  char,
			},
			fn: func() {
				result.Append(char)
			},
		})
	}
	regAction("i")
	logicPlans.NewPlanWithRootNode(rootName).
		SafeAddDependNode("i")
	expected := "i"

	exec, err := NewExecutor(actionLib, logicPlans, nil, nil, 10)
	if err != nil {
		t.Fatal(err)
	}

	logger := NewGeneralLogger("TestExecutorDuplicatedSubmit", os.Stderr)
	exec.RegisterEventReceiver(logger)
	counter := NewEventsCounter()
	exec.RegisterEventReceiver(counter)

	err = exec.Start(ExitPolicyStopOnFinish)
	if err != nil {
		t.Fatal(err)
	}

	err = exec.Submit(rootName, actions.ActionArgs{}, execId)
	if err != nil {
		t.Fatal(err)
	}
	err = exec.Submit(rootName, actions.ActionArgs{}, execId)
	if err != nil {
		t.Fatal(err)
	}

	err = exec.Stop(ExitPolicyStopOnFinish)
	if err != nil {
		t.Fatal(err)
	}

	if result.Load() != expected {
		t.Fatal(fmt.Errorf("result should be `%v` but got `%v`", expected, result.Load()))
	}
	statusMap, err := exec.QueryRunnings(rootName, func(args actions.ActionArgs) (bool, error) {
		return true, nil
	})
	if err != nil {
		t.Fatal(err)
	}
	if len(statusMap) != 0 {
		t.Fatal(fmt.Errorf("QueryRunnings should return empty"))
	}

	duplicated := counter.GetCount(EventExecTaskDuplicatedRunning) +
		counter.GetCount(EventExecTaskDuplicatedFinished) +
		counter.GetCount(EventExecTaskDuplicatedSubmitting)
	if duplicated != 1 {
		t.Fatal(fmt.Errorf("Should be a duplicated event"))
	}
}

func TestExecutorDynamicChangeScheduler(t *testing.T) {
	logicPlans := dag.NewPlans()
	actionLib := actions.NewActionLibrary()

	exec, err := NewExecutor(actionLib, logicPlans, nil, nil, 10)
	if err != nil {
		t.Fatal(err)
	}

	result := NewSafeString()
	regAction := func(char string) {
		actionLib.SafeRegister(&MockBuiltinAction{
			MockActionBase: &actions.MockActionBase{
				ActionName: char,
			},
			fn: func() {
				result.Append(char)
				if char == "b" {
					exec.ChangeActionTaskScheduler(NewFifoSigleThreadScheduler())
				}
			},
		})
	}
	regAction("a")
	regAction("b")
	regAction("c")
	logicPlans.NewPlanWithRootNode("test").
		SafeAddDependNode("a").
		SafeAddDependNode("b").
		SafeAddDependNode("c")
	expected := "cba"

	logger := NewGeneralLogger("TestExecutorDynamicChangeScheduler", os.Stderr)
	exec.RegisterEventReceiver(logger)
	exec.RegisterEventReceiver(NewDragSlow(time.Second / 50))

	err = exec.Start(ExitPolicyStopOnFinish)
	if err != nil {
		t.Fatal(err)
	}

	err = exec.Submit("test", actions.ActionArgs{}, "test-exec-id")
	if err != nil {
		t.Fatal(err)
	}

	err = exec.Wait()
	if err != nil {
		t.Fatal(err)
	}

	if result.Load() != expected {
		t.Fatal(fmt.Errorf("result should be `%v` but got `%v`", expected, result.Load()))
	}
}

func TestExecutorComplicatedDag(t *testing.T) {
	// TODO: test complicated dag
}

// TestExecutorAbortSkipPendingActions tests that when abort is called,
// pending actions that haven't started yet are skipped instead of executed.
// DAG: root -> a -> b -> c -> d (execution order: d -> c -> b -> a)
func TestExecutorAbortSkipPendingActions(t *testing.T) {
	logicPlans := dag.NewPlans()
	actionLib := actions.NewActionLibrary()

	actionTimeCost := time.Second / 2

	rootName := "test"
	execId := "test-exec-id"
	result := NewSafeString()

	// Create a chain of actions: root -> a -> b -> c -> d
	// Execution order is: d -> c -> b -> a (leaf first)
	regAction := func(char string) {
		actionLib.SafeRegister(&MockBuiltinAction{
			MockActionBase: &actions.MockActionBase{
				ActionName:  char,
				MockElapsed: actionTimeCost,
			},
			fn: func() {
				result.Append(char)
			},
		})
	}
	regAction("a")
	regAction("b")
	regAction("c")
	regAction("d")

	logicPlans.NewPlanWithRootNode(rootName).
		SafeAddDependNode("a").
		SafeAddDependNode("b").
		SafeAddDependNode("c").
		SafeAddDependNode("d")

	exec, err := NewExecutor(actionLib, logicPlans, nil, nil, 10)
	if err != nil {
		t.Fatal(err)
	}

	logger := NewGeneralLogger("TestExecutorAbortSkipPendingActions", os.Stderr)
	exec.RegisterEventReceiver(logger)

	err = exec.Start(ExitPolicyStopOnFinish)
	if err != nil {
		t.Fatal(err)
	}

	err = exec.Submit(rootName, actions.ActionArgs{}, execId)
	if err != nil {
		t.Fatal(err)
	}

	// Wait for 'd' to complete and 'c' to start running
	time.Sleep(actionTimeCost + actionTimeCost/4) // 625ms, d done, c running

	// Abort while 'c' is running - 'b' and 'a' should be skipped
	err = exec.Abort(execId)
	if err != nil {
		t.Fatal(err)
	}

	// Wait for executor to finish
	err = exec.Wait()
	if err != nil {
		t.Fatal(err)
	}

	// Verify results:
	// - 'd' should definitely complete (finished before abort)
	// - 'c' should complete (was already running when abort called)
	// - 'a' and 'b' should NOT execute (skipped due to abort)
	resultStr := result.Load()
	t.Logf("Execution result: %s (expected: dc)", resultStr)

	if resultStr == "" {
		t.Fatal("at least one action should have executed")
	}

	// 'd' must be first (leaf node executes first)
	if resultStr[0] != 'd' {
		t.Fatalf("first action should be 'd' but got '%c'", resultStr[0])
	}

	// Should have at most 'd' and 'c', not 'b' or 'a'
	if len(resultStr) > 2 {
		t.Errorf("expected at most 'dc' but got '%s' - abort may not have taken effect properly", resultStr)
	}

	// 'a' should NOT be in result (skipped due to abort)
	for _, c := range resultStr {
		if c == 'a' {
			t.Errorf("'a' should have been skipped due to abort, but got result: %s", resultStr)
		}
	}
}

// TestExecutorAbortPriorityHandling tests that abort events are processed with priority.
// This test creates parallel actions that all start at once and verifies the abort mechanism works.
// Note: When all actions start simultaneously, abort may not prevent any from executing.
// The test verifies that the abort mechanism doesn't cause errors and the task finishes properly.
func TestExecutorAbortPriorityHandling(t *testing.T) {
	logicPlans := dag.NewPlans()
	actionLib := actions.NewActionLibrary()

	actionTimeCost := time.Millisecond * 100

	rootName := "test"
	execId := "test-exec-id"
	executedActionCount := 0
	var countLock sync.Mutex

	// Register multiple fast actions that all depend on root (will run in parallel)
	for i := 0; i < 5; i++ {
		char := fmt.Sprintf("action%d", i)
		actionLib.SafeRegister(&MockBuiltinAction{
			MockActionBase: &actions.MockActionBase{
				ActionName:  char,
				MockElapsed: actionTimeCost,
			},
			fn: func() {
				countLock.Lock()
				executedActionCount++
				countLock.Unlock()
			},
		})
	}

	// Create parallel actions (all depend on root, no chain between them)
	// DAG: root -> {action0, action1, action2, action3, action4}
	root := logicPlans.NewPlanWithRootNode(rootName)
	root.SafeAddDependNodes("action0", "action1", "action2", "action3", "action4")

	exec, err := NewExecutor(actionLib, logicPlans, nil, nil, 10)
	if err != nil {
		t.Fatal(err)
	}

	err = exec.Start(ExitPolicyStopOnFinish)
	if err != nil {
		t.Fatal(err)
	}

	err = exec.Submit(rootName, actions.ActionArgs{}, execId)
	if err != nil {
		t.Fatal(err)
	}

	// All 5 actions start immediately (parallel), abort while they are running
	time.Sleep(actionTimeCost / 2)
	err = exec.Abort(execId)
	if err != nil {
		t.Fatal(err)
	}

	err = exec.Wait()
	if err != nil {
		t.Fatal(err)
	}

	// Since all actions start in parallel before abort, they may all complete.
	// This is expected behavior - abort can only prevent actions that haven't started yet.
	t.Logf("Actions executed: %d out of 5 (all may execute since they start in parallel)", executedActionCount)

	// Verify executor finished without error
	if executedActionCount == 0 {
		t.Error("Expected at least some actions to execute")
	}
}

//nolint:thelper // need cong to resolve it
func checkActionStatus(t *testing.T, status *ExecTaskStatusEx, name string, expectRunning, expectQueued, expectDone bool) {
	actionStatus := status.ActionsStatus.SafeGetActionTaskStatus(name)
	if actionStatus == nil {
		t.Fatalf("action `%v` status not found", name)
	}
	if actionStatus.Running != expectRunning {
		t.Fatalf("action `%v` running should be `%v` but got `%v`, %v", name, expectRunning, actionStatus.Running, actionStatus)
	}
	if actionStatus.Queued != expectQueued {
		t.Fatalf("action `%v` queue-status should be `%v` but got `%v`, %v", name, expectQueued, actionStatus.Queued, actionStatus)
	}
	var zero time.Duration
	if expectDone && actionStatus.Elapsed == zero {
		t.Fatalf("action `%v` should be done but got not-done", name)
	}
	if !expectDone && actionStatus.Elapsed != zero {
		t.Fatalf("action `%v` should be not-done but got done", name)
	}
}

type SafeString struct {
	str    string
	locker sync.Mutex
}

func NewSafeString() *SafeString {
	return &SafeString{}
}

func (str *SafeString) Append(x string) {
	str.locker.Lock()
	defer str.locker.Unlock()
	str.str += x
}

func (str *SafeString) Load() string {
	str.locker.Lock()
	defer str.locker.Unlock()
	return str.str
}

func TestAsyncActionIsolationOnAbort(t *testing.T) {
	logicPlans := dag.NewPlans()
	actionLib := actions.NewActionLibrary()

	asyncActionDuration := time.Second * 2
	checkInterval := time.Millisecond * 100

	asyncAction := actions.CreateActionFromAsync(
		actions.NewMockAsyncAction("shared-async", asyncActionDuration),
		checkInterval,
		asyncActionDuration*2,
	)
	actionLib.SafeRegister(asyncAction)

	logicPlans.NewPlanWithRootNode("plan1").SafeAddDependNode("shared-async")
	logicPlans.NewPlanWithRootNode("plan2").SafeAddDependNode("shared-async")

	exec, err := NewExecutor(actionLib, logicPlans, nil, nil, 10)
	if err != nil {
		t.Fatal(err)
	}

	logger := NewGeneralLogger("TestAsyncActionIsolationOnAbort", os.Stderr)
	exec.RegisterEventReceiver(logger)

	err = exec.Start(ExitPolicyNonStop)
	if err != nil {
		t.Fatal(err)
	}

	err = exec.Submit("plan1", actions.ActionArgs{}, "exec-1")
	if err != nil {
		t.Fatal(err)
	}
	err = exec.Submit("plan2", actions.ActionArgs{}, "exec-2")
	if err != nil {
		t.Fatal(err)
	}

	time.Sleep(asyncActionDuration / 4)

	status1 := exec.GetRunningStatus("exec-1")
	status2 := exec.GetRunningStatus("exec-2")
	if status1 == nil || status2 == nil {
		t.Fatal("both tasks should be running")
	}

	err = exec.Abort("exec-1")
	if err != nil {
		t.Fatal(err)
	}

	time.Sleep(asyncActionDuration * 2)

	status2AfterAbort := exec.GetRunningStatus("exec-2")
	if status2AfterAbort != nil {
		actionStatus := status2AfterAbort.ActionsStatus.SafeGetActionTaskStatus("shared-async")
		if actionStatus != nil && actionStatus.Error != nil {
			t.Fatalf("exec-2 should complete without error, but got: %v", actionStatus.Error)
		}
	}

	err = exec.Stop(ExitPolicyStopOnFinish)
	if err != nil {
		t.Fatal(err)
	}

	finalStatus1 := exec.GetRunningStatus("exec-1")
	finalStatus2 := exec.GetRunningStatus("exec-2")

	if finalStatus1 != nil && finalStatus1.State != ExecTaskStateTypeFinishedAborted {
		t.Logf("exec-1 state: %v", finalStatus1.State)
	}

	if finalStatus2 != nil {
		action2 := finalStatus2.ActionsStatus.SafeGetActionTaskStatus("shared-async")
		if action2 != nil && !action2.IsDone() && action2.Error != nil {
			t.Errorf("exec-2 async action should complete successfully, got error: %v", action2.Error)
		}
	}
}

// TestQueryRunningsImmediatelyAfterAbort verifies that QueryRunnings returns no result
// for an exec-task immediately after Abort() returns, without any sleep.
// This guards against the race where the abort signal is only enqueued into a channel
// and the run-loop has not yet removed the task from its in-memory state.
func TestQueryRunningsImmediatelyAfterAbort(t *testing.T) {
	logicPlans := dag.NewPlans()
	actionLib := actions.NewActionLibrary()

	actionTimeCost := time.Second * 2
	rootName := "test"
	execId := "test-exec-id"

	actionLib.SafeRegister(&MockBuiltinAction{
		MockActionBase: &actions.MockActionBase{
			ActionName:  "slow",
			MockElapsed: actionTimeCost,
		},
		fn: func() {},
	})
	logicPlans.NewPlanWithRootNode(rootName).SafeAddDependNode("slow")

	exec, err := NewExecutor(actionLib, logicPlans, nil, nil, 10)
	if err != nil {
		t.Fatal(err)
	}

	if err := exec.Start(ExitPolicyStopOnFinish); err != nil {
		t.Fatal(err)
	}

	if err := exec.Submit(rootName, actions.ActionArgs{}, execId); err != nil {
		t.Fatal(err)
	}

	// Wait until the action is actually running before aborting.
	time.Sleep(actionTimeCost / 4)

	if err := exec.Abort(execId); err != nil {
		t.Fatal(err)
	}

	// Immediately query — no sleep. The task must already be gone from results.
	statusMap, err := exec.QueryRunnings(rootName, func(args actions.ActionArgs) (bool, error) {
		return true, nil
	})
	if err != nil {
		t.Fatal(err)
	}
	if _, found := statusMap[execId]; found {
		t.Fatalf("QueryRunnings returned execId %q immediately after Abort(); expected it to be filtered out", execId)
	}

	// Let the executor finish cleanly.
	_ = exec.Wait()
}