package executor

import (
	"context"
	"fmt"
	"sync/atomic"
	"time"

	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/actions"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/base"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/dag"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/storage"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
)

// TODO(shenjun): come up a better name, may be `runtime`?
type RemoteExecutor interface {
	Abort(ctx context.Context, execId string) error
	Submit(ctx context.Context, planName string, args actions.ActionArgs, execId string) error
	QueryRunnings(ctx context.Context, planName string, matchFilter MatchFilter) (ExecTaskIdStatus, error)
}

type ExitPolicyType uint32

const (
	ExitPolicyNonStop                      ExitPolicyType = 0
	ExitPolicyStopOnFinish                 ExitPolicyType = 1
	ExitPolicyStopNowDontWait              ExitPolicyType = 2
	ExitPolicyStopNowWaitForRunningActions ExitPolicyType = 3
)

const (
	defaultTickerInterval      time.Duration = 5 * time.Second
	defaultCleanupInterval     time.Duration = 3 * time.Hour
	defaultKeepDeletedDuration time.Duration = 48 * time.Hour
)

func (policy ExitPolicyType) String() string {
	switch policy {
	case ExitPolicyNonStop:
		return "non-stop"
	case ExitPolicyStopOnFinish:
		return "stop-on-finish"
	case ExitPolicyStopNowDontWait:
		return "stop-now-dont-wait"
	case ExitPolicyStopNowWaitForRunningActions:
		return "stop-now-wait-for-running-actions"
	default:
		return "unknown-stop-policy"
	}
}

type Executor struct {
	actions    *actions.ActionLibrary
	logicPlans *dag.Plans
	// renders    *ActionRenderLibrary

	physicPlanBuilder   *PhysicPlanBuilder
	execStatusStg       storage.ExecStatusStg
	dagStatusStg        storage.DagStatusStg
	actionTaskScheduler *SafeWrapScheduler

	eventHub                 *SyncEventHub
	exitPolicyChangeNotifyCh chan ExitPolicyType
	execTaskSubmitCh         chan *ExecutorTask
	runEndCh                 chan error
	// actionTaskResultCh       chan *ActionTaskStatus
	// `runningExecTasks` is thread-safe, the elements in it are read-only and only used by the run-thread.
	// it's for:
	//   1. judge exit when all executor-tasks done;
	//   2. query running executor-tasks
	// `runningExecTasksStatus` can do the same jobs too,
	// but `runningExecTasks` is more simple and easy to output to event-hub
	runningExecTasks *SafeExecutorTasks

	// `runningExecTasksStatus` is thread-safe.
	// it won't return it's elements, all modifications will be under it's mutex lock.
	// it's for:
	//   1. return detail dag-action status in running-query
	//   2. check and judge dag-actions in an executor is all done
	//   3. prevent dag-actions duplicatedly add to scheduler
	runningExecTasksStatus *SafeExecTaskIdStatus

	// as extra info for scheduler, only used in the run-thread
	runningActionTasks RunningActionTasks

	// for detecting duplicated exec-task in a short time
	submittingExecTasks *SafeStrSet

	// pendingAbortExecIds tracks exec-tasks that have been asked to abort but not yet
	// processed by the run-loop. It is updated synchronously in Abort() so that
	// QueryRunnings immediately stops returning these tasks.
	pendingAbortExecIds *SafeStrSet

	abortingExecTaskCh chan *ExecutorTask
	// abortingExecTasks must only be accessed from the run-loop goroutine;
	// cross-goroutine readers use pendingAbortExecIds instead
	abortingExecTasks map[string]*ExecutorTask

	stgErrorWaitInterval time.Duration
	tickerInterval       time.Duration
	cleanupInterval      time.Duration
	cleanupRunning       atomic.Bool
	maxFinishedTasks     int64
}

func NewExecutor(
	actionLib *actions.ActionLibrary,
	logicPlans *dag.Plans,
	execStatusStg storage.ExecStatusStg,
	dagStatusStg storage.DagStatusStg,
	submitBufferSize uint,
) (*Executor, error) {
	keepDeletedDuration := defaultKeepDeletedDuration

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

	physicPlanBuilder := NewPhysicPlanBuilder()
	physicPlanBuilder.RegisterRender(NewDagNodeRenderByActions(actionLib))

	// full speed scheduler is the default scheduler
	actionTaskScheduler := NewFullSpeedScheduler()

	return &Executor{
		actions:                  actionLib,
		logicPlans:               logicPlans,
		physicPlanBuilder:        physicPlanBuilder,
		execStatusStg:            execStatusStg,
		dagStatusStg:             dagStatusStg,
		runningExecTasks:         NewSafeExecutorTasks(keepDeletedDuration),
		runningExecTasksStatus:   NewSafeExecTaskIdStatus(keepDeletedDuration),
		runningActionTasks:       NewRunningActionTasks(),
		actionTaskScheduler:      NewSafeWrapScheduler(actionTaskScheduler),
		eventHub:                 NewSyncEventHub(),
		exitPolicyChangeNotifyCh: make(chan ExitPolicyType),
		execTaskSubmitCh:         make(chan *ExecutorTask, submitBufferSize),
		runEndCh:                 make(chan error, 9),
		abortingExecTaskCh:       make(chan *ExecutorTask, submitBufferSize),
		abortingExecTasks:        make(map[string]*ExecutorTask),
		submittingExecTasks:      NewSafeStrSet(),
		pendingAbortExecIds:      NewSafeStrSet(),
		stgErrorWaitInterval:     time.Second,
		tickerInterval:           defaultTickerInterval,
		cleanupInterval:          defaultCleanupInterval,
		maxFinishedTasks:         1000,
	}, nil
}

func (exec *Executor) GetExecStatusStorage() storage.ExecStatusStg {
	return exec.execStatusStg
}

// TODO(shenjun): tickerInterval can not be changed after executor start.
func (exec *Executor) SetTickerInterval(dur time.Duration) error {
	exec.tickerInterval = dur
	return nil
}

// SetCleanupInterval sets the interval at which the executor automatically cleans up finished tasks.
// If dur <= 0, it falls back to the default cleanup interval.
func (exec *Executor) SetCleanupInterval(dur time.Duration) {
	if dur <= 0 {
		dur = defaultCleanupInterval
	}
	exec.cleanupInterval = dur
}

// SetKeepDeletedDuration sets how long finished tasks are retained before cleanup.
// Default is 48h. If dur <= 0, it falls back to the default.
func (exec *Executor) SetKeepDeletedDuration(dur time.Duration) {
	if dur <= 0 {
		dur = defaultKeepDeletedDuration
	}
	exec.runningExecTasks.SetKeepDeletedDuration(dur)
	exec.runningExecTasksStatus.SetKeepDeletedDuration(dur)
}

// SetMaxFinishedTasks sets the maximum number of finished tasks to retain.
// When the count exceeds this limit, the oldest finished tasks are deleted first.
// If maxTasks <= 0, the count-based limit is disabled (only time-based cleanup applies).
func (exec *Executor) SetMaxFinishedTasks(maxTasks int64) {
	if maxTasks < 0 {
		maxTasks = 0
	}
	exec.maxFinishedTasks = maxTasks
}

// the receivers will handle events in the main running thread:
// - that can give receivers the opportunity to modify things in executor, like scheduler policy.
// - expect receivers are lightweight, won't run for a long time.
func (exec *Executor) RegisterEventReceiver(receiver EventReceiver) {
	exec.eventHub.RegisterReceiver(receiver)
}

// the receivers will handle events in a speparated thread:
// - expect receivers can run for a long time, though it will affect other receivers.
func (exec *Executor) RegisterAsyncEventReceiver(receiver EventReceiver) {
	exec.eventHub.GetAsyncHub().RegisterReceiver(receiver)
}

func (exec *Executor) Start(exitPolicy ExitPolicyType) error {
	err := exec.prerun()
	if err != nil {
		return err
	}
	go func() {
		err := exec.run(exitPolicy)
		exec.runEndCh <- err
	}()
	return nil
}

func (exec *Executor) Wait() error {
	err := <-exec.runEndCh
	return err
}

func (exec *Executor) Stop(exitPolicy ExitPolicyType) error {
	exec.exitPolicyChangeNotifyCh <- exitPolicy
	err := <-exec.runEndCh
	return err
}

func (exec *Executor) Abort(execId string) error {
	execTask := exec.runningExecTasks.Get(execId)
	if execTask == nil {
		return fmt.Errorf("Executor.Abort: executor task `%v` not found", execId)
	}
	exec.eventHub.OnEvent(EventExecTaskAbortPresend,
		EventDataWithName("exec-id", base.CloneableString(execId)))

	// Mark as pending-abort immediately so QueryRunnings stops returning this task
	// before the run-loop has a chance to process the abort signal.
	exec.pendingAbortExecIds.Add(execId)
	exec.abortingExecTaskCh <- execTask

	exec.eventHub.OnEvent(EventExecTaskAbortPresend,
		EventDataWithName("exec-id", base.CloneableString(execId)))
	return nil
}

func (exec *Executor) Submit(planName string, args actions.ActionArgs, execId string) error {
	tilog.PlainInfo("Submit is called", zap.String("exec-id", execId), zap.String("plan-name", planName))
	// we have internal duplicated-detecting, but it's nice to have another detecting outside the core loop
	if exec.runningExecTasks.Get(execId) != nil {
		exec.eventHub.OnEvent(EventExecTaskDuplicatedRunning,
			EventDataWithName("exec-id", base.CloneableString(execId)),
			EventDataWithName("timing", base.CloneableString("on-submit")))
		return nil
	}

	if exec.runningExecTasks.IsDeleted(execId) {
		exec.eventHub.OnEvent(EventExecTaskDuplicatedFinished,
			EventDataWithName("exec-id", base.CloneableString(execId)),
			EventDataWithName("timing", base.CloneableString("on-submit")))
		return nil
	}

	submitting := exec.submittingExecTasks.CheckHasAndAdd(execId)
	if submitting {
		exec.eventHub.OnEvent(EventExecTaskDuplicatedSubmitting,
			EventDataWithName("exec-id", base.CloneableString(execId)),
			EventDataWithName("timing", base.CloneableString("on-submit")))
		return nil
	}

	logicPlan := exec.logicPlans.GetAndCloneDagByRoot(planName)
	if logicPlan == nil {
		return fmt.Errorf("Executor.Submit: logic plan `%v` not found", planName)
	}

	physicPlan, err := exec.physicPlanBuilder.BuildPlan(planName, logicPlan, execId)
	if err != nil {
		return err
	}
	execTask := &ExecutorTask{
		LogicPlan:  logicPlan,
		PhysicPlan: physicPlan,
		RootName:   planName,
		Args:       args,
		ExecId:     execId,
		StartTime:  time.Now(),
	}
	marshalledExecTask, err := execTask.ToString()
	if err != nil {
		return err
	}

	if err := exec.execStatusStg.OnNewTask(execId, marshalledExecTask); err != nil {
		return err
	}
	// caller may submit an existed and already finished task, do not add finished task to the running queue.
	finished, err := exec.execStatusStg.IsTaskFinished(execId)
	if err != nil {
		return err
	}
	tilog.PlainInfo("OnNewTask", zap.String("exec-id", execId), zap.String("plan-name", planName), zap.Bool("finished", finished))
	if finished {
		return nil
	}

	exec.addExecTaskToRunning(execTask, "on-submit-presend")
	exec.eventHub.OnEvent(EventExecTaskSubmitPresend,
		EventDataWithType(EventDataExecTask, execTask))
	exec.execTaskSubmitCh <- execTask
	exec.eventHub.OnEvent(EventExecTaskSubmitSent,
		EventDataWithType(EventDataExecTask, execTask))
	return nil
}

func (exec *Executor) Run(exitPolicy ExitPolicyType) error {
	err := exec.prerun()
	if err != nil {
		return err
	}
	return exec.run(exitPolicy)
}

type QueryMatchFunc func(args actions.ActionArgs) (bool, error)

func (exec *Executor) QueryExecTasks(planName string, matchFunc QueryMatchFunc) (execIds []string, err error) {
	return exec.queryExecTasks(planName, true, matchFunc)
}

func (exec *Executor) queryExecTasks(planName string, includeFinished bool, matchFunc QueryMatchFunc) (execIds []string, err error) {
	err = exec.runningExecTasks.Iterate(includeFinished, func(execTask *ExecutorTask) error {
		if planName != execTask.RootName {
			return nil
		}
		// Skip tasks that are pending abort: Abort() has been called but the run-loop
		// has not yet removed them from runningExecTasks.
		if exec.pendingAbortExecIds.Has(execTask.ExecId) {
			return nil
		}
		matched, err := matchFunc(execTask.Args)
		if err != nil {
			return err
		}
		if matched {
			execIds = append(execIds, execTask.ExecId)
		}
		return nil
	})
	if err != nil {
		return nil, err
	}
	return execIds, nil
}

// QueryRunningsWithFilter queries running tasks using a MatchFilter for server-side filtering.
func (exec *Executor) QueryRunningsWithFilter(planName string, filter MatchFilter) (ExecTaskIdStatus, error) {
	return exec.QueryRunnings(planName, filter.ToMatchFunc())
}

func (exec *Executor) QueryRunnings(planName string, matchFunc QueryMatchFunc) (ExecTaskIdStatus, error) {
	execIds, err := exec.queryExecTasks(planName, false, matchFunc)
	if err != nil {
		return nil, err
	}
	execTasksStatus := NewExecTaskIdStatus()
	for _, execId := range execIds {
		status, ok := exec.runningExecTasksStatus.GetAndClone(execId)
		if !ok && status == nil {
			// the task might be submitted but not start yet
			execTask := exec.runningExecTasks.Get(execId)
			if execTask != nil {
				status = NewExecTaskStatus(execTask)
			}
		}
		if status != nil {
			execTasksStatus[execId] = NewExecTaskStatusEx(status, exec.getExecTaskState(execId))
		}
	}
	return execTasksStatus, nil
}

func (exec *Executor) GetRunningStatus(execId string) *ExecTaskStatusEx {
	status, _ := exec.runningExecTasksStatus.GetAndClone(execId)
	execTask := exec.runningExecTasks.Get(execId)
	if status == nil && execTask != nil {
		status = NewExecTaskStatus(execTask)
	}
	return NewExecTaskStatusEx(status, exec.getExecTaskState(execId))
}

// Change Action after executor Start may loss some action.
func (exec *Executor) ChangeActionTaskScheduler(origin ActionTaskScheduler) {
	exec.actionTaskScheduler.ChangeScheduler(origin)
}

func (exec *Executor) getExecTaskState(execId string) ExecTaskStateType {
	isSubmitting := exec.submittingExecTasks.Has(execId)
	if isSubmitting {
		return ExecTaskStateTypePending
	}
	// abortingExecTasks is owned by the run-loop goroutine; query paths must
	// use the thread-safe pendingAbortExecIds, which shares its lifecycle
	// (added on Abort()/terminal error, removed in onExecTaskFinish)
	isAborting := exec.pendingAbortExecIds.Has(execId)
	if isAborting {
		return ExecTaskStateTypeAborting
	}
	isRunning, hasError := exec.runningExecTasksStatus.IsRunning(execId)
	if isRunning {
		if hasError {
			return ExecTaskStateTypeRunningWithError
		} else {
			// TODO: for async-task, the error might be neglected
			return ExecTaskStateTypeRunningOK
		}
	}
	isDeleting, doneNotAborted := exec.runningExecTasksStatus.IsDeleting(execId)
	if isDeleting {
		if doneNotAborted {
			return ExecTaskStateTypeFinishedAborted
		} else {
			return ExecTaskStateTypeFinishedDone
		}
	}
	return ExecTaskStateTypeUnknown
}

func (exec *Executor) run(exitPolicy ExitPolicyType) (err error) {
	exec.eventHub.OnEvent(EventExecutorStart,
		EventDataWithNameType("logic plans", EventDataDag, exec.logicPlans))

	actionTaskResultCh := make(chan *ActionTaskStatus)
	ticker := time.NewTicker(exec.tickerInterval)
	cleanupTicker := time.NewTicker(exec.cleanupInterval)
	loop := true

	// TODO: use a ticker for timeout check ?
	// TODO: stall/stucked exec-task interval check

	for loop {
		exec.eventHub.OnEvent(EventExecutorRunLoopStart,
			EventDataWithNameType("running exec tasks", EventDataStrList, base.CloneableStrings(exec.runningExecTasks.ExecIds())),
			EventDataWithName("running exec tasks len", base.CloneableInt(exec.runningExecTasks.Len())),
			EventDataWithName("exitPolicy", base.CloneableString(exitPolicy.String())),
			EventDataWithName("scheduler queue cnt", base.CloneableInt(exec.actionTaskScheduler.Len())),
			EventDataWithName("running action cnt", base.CloneableInt(len(exec.runningActionTasks))),
		)

		if exec.actionTaskScheduler.Len() != 0 && len(exec.runningActionTasks) == 0 {
			exec.popAndRunActionTasks(nil, EventSchedulerIdlePop, actionTaskResultCh, "on-run-loop-start-and-no-running-tasks")
		}

		// Priority handling: process all pending abort requests first before handling other events
		// This ensures abort events are processed with higher priority
	drainAbort:
		for {
			select {
			case execTask := <-exec.abortingExecTaskCh:
				exec.abortingExecTasks[execTask.ExecId] = execTask
				exec.eventHub.OnEvent(EventExecutorRunLoopOnExecTaskAbort,
					EventDataWithType(EventDataExecTask, execTask),
					EventDataWithNameType("aborting tasks cnt",
						EventDataStr, base.CloneableInt(len(exec.abortingExecTasks))))
			default:
				break drainAbort
			}
		}

		select {
		case <-cleanupTicker.C:
			if exec.cleanupRunning.CompareAndSwap(false, true) {
				go func() {
					defer exec.cleanupRunning.Store(false)
					tilog.PlainInfo("auto cleanup started",
						zap.Duration("cleanupInterval", exec.cleanupInterval),
						zap.Duration("keepDeletedDuration", exec.runningExecTasks.GetKeepDeletedDuration()),
						zap.Int64("maxFinishedTasks", exec.maxFinishedTasks))
					cleaned, err := exec.CleanFinished()
					if err != nil {
						tilog.PlainWarn("auto cleanup finished tasks failed", zap.Error(err))
					} else {
						tilog.PlainInfo("auto cleanup finished tasks", zap.Int64("cleaned", cleaned))
					}
				}()
			}

		case <-ticker.C:
			if exec.actionTaskScheduler.Len() == 0 {
				continue
			}
			toBeRuns := exec.actionTaskScheduler.PopToBeRuns(exec.runningActionTasks)
			exec.eventHub.OnEvent(EventSchedulerTickPop,
				EventDataWithName("scheduler queue cnt", base.CloneableInt(exec.actionTaskScheduler.Len())),
				EventDataWithName("pop tasks", toBeRuns))
			if len(toBeRuns) > 0 {
				exec.asyncRunActionTasks(toBeRuns, actionTaskResultCh)
			}

		case execTask := <-exec.execTaskSubmitCh:
			exec.eventHub.OnEvent(EventExecutorRunLoopOnExecTaskSubmit,
				EventDataWithType(EventDataExecTask, execTask))
			exec.submittingExecTasks.Delete(execTask.ExecId)
			reason := "receive-submit-in-loop-run"
			exec.safeQueueTaskActions(execTask, reason)
			exec.popAndRunActionTasks(execTask, EventSchedulerPop, actionTaskResultCh, reason)

		case execTask := <-exec.abortingExecTaskCh:
			exec.abortingExecTasks[execTask.ExecId] = execTask
			exec.eventHub.OnEvent(EventExecutorRunLoopOnExecTaskAbort,
				EventDataWithType(EventDataExecTask, execTask),
				EventDataWithNameType("aborting tasks cnt",
					EventDataStr, base.CloneableInt(len(exec.abortingExecTasks))))

		case actionTaskResult := <-actionTaskResultCh:
			exec.eventHub.OnEvent(EventExecutorRunLoopOnActionResult,
				EventDataWithType(EventDataActionResult, actionTaskResult))

			exec.runningExecTasksStatus.PutActionTaskStatus(actionTaskResult)

			// Only remove from runningActionTasks if it was actually running (Queued == true)
			// Actions skipped due to abort are never added to runningActionTasks
			if actionTaskResult.Queued {
				exec.runningActionTasks.Delete(actionTaskResult.ActionTask)
			}

			execId := actionTaskResult.ActionTask.ExecTask.ExecId
			dagAction := actionTaskResult.ActionTask.DagAction
			dagActionId := dagAction.Id()
			reasonToDoNext := fmt.Sprintf("after-dag-action-done:%s", dagActionId)

			// set action-task done
			if actionTaskResult.Error == nil {
				exec.safeSetDagActionDone(execId, dagActionId)
				exec.eventHub.OnEvent(EventDagActionStgSetDone,
					EventDataWithName("exec-id", base.CloneableString(execId)),
					EventDataWithName("dag-action-id", base.CloneableString(dagActionId)),
					EventDataWithType(EventDataDagAction, dagAction))
			}

			if exitPolicy == ExitPolicyStopNowDontWait || exitPolicy == ExitPolicyStopNowWaitForRunningActions {
				exec.eventHub.OnEvent(EventSchedulerSkip,
					EventDataWithName("scheduler queue cnt", base.CloneableInt(exec.actionTaskScheduler.Len())),
					EventDataWithType(EventDataActionTask, actionTaskResult.ActionTask),
					EventDataWithName("exit policy", base.CloneableString(exitPolicy.String())))
				if exitPolicy == ExitPolicyStopNowDontWait {
					loop = false
				}
				if exitPolicy == ExitPolicyStopNowWaitForRunningActions &&
					exec.actionTaskScheduler.Len() == 0 {
					loop = false
				}
				continue
			}

			execTask := actionTaskResult.ActionTask.ExecTask

			// a terminal error can never be resolved by retrying, stop the whole
			// executor-task through the abort machinery instead of rescheduling
			if actionTaskResult.Error != nil &&
				actions.IsTerminalError(actionTaskResult.Error) &&
				exec.abortingExecTasks[execTask.ExecId] == nil {
				exec.abortingExecTasks[execTask.ExecId] = execTask
				exec.pendingAbortExecIds.Add(execTask.ExecId)
				exec.eventHub.OnEvent(EventExecTaskTerminalError,
					EventDataWithType(EventDataExecTask, execTask),
					EventDataWithName("dag-action-id", base.CloneableString(dagActionId)),
					EventDataWithType(EventDataError, base.NewCloneableError(actionTaskResult.Error)))
			}

			execTaskIsAborting := (exec.abortingExecTasks[execTask.ExecId] != nil)

			// re-schedule this action-task if it failed
			// TODO: improve backoff, store all errors in actionTaskResult, let scheduler do the judge
			if actionTaskResult.Error != nil {
				exec.safeSetDagActionError(execId, dagActionId, actionTaskResult.Error)
				if !execTaskIsAborting {
					exec.actionTaskScheduler.Add(actionTaskResult.ActionTask)
					exec.eventHub.OnEvent(EventSchedulerReschedule,
						EventDataWithName("scheduler queue cnt", base.CloneableInt(exec.actionTaskScheduler.Len())),
						EventDataWithType(EventDataActionTask, actionTaskResult.ActionTask))
				} else {
					exec.tryAbortRunningActionsOfExecTask(execTask.ExecId)
					action := actionTaskResult.ActionTask.DagAction.Action
					actionAborted := actions.TryAbortAction(action)
					exec.eventHub.OnEvent(EventSchedulerSendAbortSignalToAction,
						EventDataWithName("action name", base.CloneableString(action.Name())),
						EventDataWithName("in running", base.CloneableBool(false)),
						EventDataWithName("send succeeded", base.CloneableBool(actionAborted)))
					actionTaskResult.Queued = false
					exec.runningExecTasksStatus.PutActionTaskStatus(actionTaskResult)
					exec.eventHub.OnEvent(EventSchedulerSkipRescheduleOnAbort,
						EventDataWithName("scheduler queue cnt", base.CloneableInt(exec.actionTaskScheduler.Len())),
						EventDataWithType(EventDataActionTask, actionTaskResult.ActionTask))
				}
			}

			var waitingActionCnt int

			// if current executor-task is aborting, dont schedule more actions from it,
			// else schedule more action-tasks from it
			if !execTaskIsAborting {
				waitingActionCnt = exec.safeQueueTaskActions(execTask, reasonToDoNext)
			}

			// check the current executor-task is finished or not
			includeUnqueuedOnDoneCheck := !execTaskIsAborting
			readyActionsAreAllDone := exec.runningExecTasksStatus.IsAllDone(execTask.ExecId, includeUnqueuedOnDoneCheck)
			if waitingActionCnt == 0 && readyActionsAreAllDone {
				exec.onExecTaskFinish(execTask)
				// check should exit or not
				if exec.runningExecTasks.Len() == 0 {
					exec.eventHub.OnEvent(EventExecTaskListEmpty,
						EventDataWithName("exit policy", base.CloneableString(exitPolicy.String())))
					if exitPolicy == ExitPolicyStopOnFinish ||
						exitPolicy == ExitPolicyStopNowWaitForRunningActions {
						loop = false
						break // break the select statement
					}
				}
			}

			// fetch new action-tasks from scheduler and run them
			exec.popAndRunActionTasks(execTask, EventSchedulerPop, actionTaskResultCh, reasonToDoNext)

		case exitPolicy = <-exec.exitPolicyChangeNotifyCh:
			exec.eventHub.OnEvent(EventExecutorRunLoopOnStopPolicyNotify,
				EventDataWithName("exit policy", base.CloneableString(exitPolicy.String())))
			if exitPolicy == ExitPolicyStopNowDontWait {
				loop = false
				break // break the select statement
			}
			// Check if should exit immediately when policy changes to StopOnFinish
			// This handles the case where Stop() is called after all tasks are already done
			if (exitPolicy == ExitPolicyStopOnFinish ||
				exitPolicy == ExitPolicyStopNowWaitForRunningActions) &&
				exec.runningExecTasks.Len() == 0 && len(exec.runningActionTasks) == 0 {
				exec.eventHub.OnEvent(EventExecTaskListEmpty,
					EventDataWithName("exit policy", base.CloneableString(exitPolicy.String())))
				loop = false
			}
		}
	}

	exec.eventHub.OnEvent(EventExecutorDone)
	ticker.Stop()
	cleanupTicker.Stop()
	return nil
}

func (exec *Executor) addExecTaskToRunning(execTask *ExecutorTask, reason string) {
	failToAddByDuplicated := exec.runningExecTasks.Add(execTask)
	if failToAddByDuplicated {
		exec.eventHub.OnEvent(EventExecTaskDuplicated,
			EventDataWithNameType("old", EventDataExecTask, exec.runningExecTasks.Get(execTask.ExecId)),
			EventDataWithNameType("new", EventDataExecTask, execTask),
			EventDataWithName("timing", base.CloneableString(reason)))
	} else {
		exec.eventHub.OnEvent(EventExecTaskAddToRunning,
			EventDataWithType(EventDataExecTask, execTask))
	}
}

func (exec *Executor) popAndRunActionTasks(
	relatedExecTask *ExecutorTask,
	event EventName,
	actionTaskResultCh chan *ActionTaskStatus,
	reason string,
) {
	toBeRuns := exec.actionTaskScheduler.PopToBeRuns(exec.runningActionTasks)
	exec.eventHub.OnEvent(event,
		EventDataWithName("scheduler queue cnt", base.CloneableInt(exec.actionTaskScheduler.Len())),
		EventDataWithName("pop tasks", toBeRuns),
		EventDataWithName("timing", base.CloneableString(reason)))
	if relatedExecTask != nil && len(toBeRuns) == 0 {
		if exec.runningActionTasks.Has(relatedExecTask.ExecId) {
			exec.eventHub.OnEvent(EventExecTaskStall,
				EventDataWithType(EventDataExecTask, relatedExecTask),
				EventDataWithName("timing", base.CloneableString(reason)))
		} else {
			exec.eventHub.OnEvent(EventExecTaskStucked,
				EventDataWithType(EventDataExecTask, relatedExecTask),
				EventDataWithName("timing", base.CloneableString(reason)))
		}
	}
	if len(toBeRuns) > 0 {
		exec.asyncRunActionTasks(toBeRuns, actionTaskResultCh)
	}
}

func (exec *Executor) asyncRunActionTasks(actionTasks ActionTasks, actionTaskResultCh chan *ActionTaskStatus) {
	exec.eventHub.OnEvent(EventActionTasksToBeRun,
		EventDataWithType(EventDataActionTaskList, actionTasks))
	for _, actionTask := range actionTasks {
		execId := actionTask.ExecTask.ExecId
		// Check abort status before starting action, skip if already aborting
		if exec.abortingExecTasks[execId] != nil {
			exec.eventHub.OnEvent(EventSchedulerSkipRescheduleOnAbort,
				EventDataWithName("scheduler queue cnt", base.CloneableInt(exec.actionTaskScheduler.Len())),
				EventDataWithType(EventDataActionTask, actionTask))
			// Send a result to channel so the main loop can check if execTask is finished
			go func(actionTask *ActionTask) {
				actionTaskResultCh <- &ActionTaskStatus{
					ActionTask: actionTask,
					Running:    false,
					Queued:     false, // Mark as not queued since it was skipped
					Error:      fmt.Errorf("action skipped due to abort"),
				}
			}(actionTask)
			continue
		}
		exec.runningActionTasks.Add(actionTask)
		now := time.Now()
		exec.eventHub.OnEvent(EventActionTaskStart,
			EventDataWithType(EventDataActionTask, actionTask))
		status := &ActionTaskStatus{
			ActionTask: actionTask,
			StartTime:  now,
			Running:    true,
			Queued:     true,
		}
		exec.runningExecTasksStatus.PutActionTaskStatus(status)
		go func(actionTask *ActionTask) {
			err := actionTask.DagAction.Invoke(actionTask.ExecTask.Args)
			status := &ActionTaskStatus{
				ActionTask: actionTask,
				StartTime:  now,
				Elapsed:    time.Since(now),
				Running:    false,
				Queued:     true,
				Error:      err,
			}
			actionTaskResultCh <- status
		}(actionTask)
	}
}

// queueTaskActions() do two things:
// 1. Add actionStatus into runningExecTasksStatus for ready actions and waiting actions
// 2. Add all ready actions under the given execTask into actionTaskScheduler
func (exec *Executor) queueTaskActions(execTask *ExecutorTask, reason string) (waitingActionCnt int, err error) {
	exec.eventHub.OnEvent(EventExecTaskQueueActionsStart,
		EventDataWithType(EventDataExecTask, execTask),
		EventDataWithName("queue-reason", base.CloneableString(reason)))

	queried := make(map[string]bool)
	isDoneFunc := func(action *dag.DagAction) (bool, error) {
		actionId := action.Id()
		isDone, err := exec.dagStatusStg.IsDone(execTask.ExecId, actionId)
		tempId := fmt.Sprintf("%s:%s", execTask.ExecId, actionId)
		if !queried[tempId] {
			exec.eventHub.OnEvent(EventDagActionStgQueryIsDone,
				EventDataWithName("exec-id", base.CloneableString(execTask.ExecId)),
				EventDataWithName("dag-action-id", base.CloneableString(actionId)),
				EventDataWithName("is done", base.CloneableBool(isDone)),
				EventDataWithName("query err", base.NewCloneableError(err)))
		}
		queried[tempId] = true
		return isDone, err
	}
	dagActions, waitingActions, err := execTask.PhysicPlan.GetReadyActionTasks(execTask.RootName, isDoneFunc)
	if err != nil {
		return -1, err
	}

	var newReadyActionTasks []*ActionTask
	created := exec.runningExecTasksStatus.EnsureExisted(execTask)
	exec.eventHub.OnEvent(EventGetReadyActions,
		EventDataWithType(EventDataDagActionList, dag.DagActions(dagActions)),
		EventDataWithName("exec task status created", base.CloneableBool(created)))

	// put waiting actions in `runningExecTasksStatus` for `QueryRunnings`
	for _, dagAction := range waitingActions {
		dagActionId := dagAction.Id()
		if exec.runningExecTasksStatus.IsDagActionStatusExisted(execTask.ExecId, dagActionId) {
			continue
		}
		actionTask := &ActionTask{
			ExecTask:  execTask,
			DagAction: dagAction,
		}
		exec.runningExecTasksStatus.PutActionTaskStatus(&ActionTaskStatus{
			ActionTask: actionTask,
			Queued:     false,
		})
	}

	for _, dagAction := range dagActions {
		dagActionId := dagAction.Id()
		if exec.runningExecTasksStatus.IsDagActionQueued(execTask.ExecId, dagActionId) {
			continue
		}
		actionTask := &ActionTask{
			ExecTask:  execTask,
			DagAction: dagAction,
		}
		newReadyActionTasks = append(newReadyActionTasks, actionTask)
		exec.runningExecTasksStatus.PutActionTaskStatus(&ActionTaskStatus{
			ActionTask: actionTask,
			Queued:     true,
		})
	}
	exec.actionTaskScheduler.Add(newReadyActionTasks...)
	exec.eventHub.OnEvent(EventSchedulerAdd,
		EventDataWithName("scheduler queue cnt", base.CloneableInt(exec.actionTaskScheduler.Len())),
		EventDataWithType(EventDataActionTaskList, ActionTasks(newReadyActionTasks)))

	exec.eventHub.OnEvent(EventNewReadyActions,
		EventDataWithType(EventDataActionTaskList, ActionTasks(newReadyActionTasks)),
		EventDataWithName("exec-id", base.CloneableString(execTask.ExecId)),
		EventDataWithNameType("waiting dag action", EventDataActionTaskList, dag.DagActions(waitingActions)))
	return len(waitingActions), nil
}

func (exec *Executor) onExecTaskFinish(execTask *ExecutorTask) {
	exec.eventHub.OnEvent(EventExecTaskDone,
		EventDataWithType(EventDataExecTask, execTask))
	exec.safeSetExecTaskFinished(execTask.ExecId)
	isAborted := (exec.abortingExecTasks[execTask.ExecId] != nil)
	// TODO: keep it for a while to prevent duplcated request
	exec.runningExecTasks.Delete(execTask.ExecId)
	exec.runningExecTasksStatus.Delete(execTask.ExecId, !isAborted)
	if isAborted {
		delete(exec.abortingExecTasks, execTask.ExecId)
	}
	exec.pendingAbortExecIds.Delete(execTask.ExecId)
}

func (exec *Executor) prerun() error {
	// if Submit a request before Run, it will be in both stg and mem and will be load here again,
	// but that's ok, will be dedup by the same exec-id
	err := exec.loadUnfinished()
	if err != nil {
		return err
	}
	err = exec.runningExecTasks.Iterate(false, func(execTask *ExecutorTask) error {
		_, err := exec.queueTaskActions(execTask, "load-unfinisheds-and-submitteds-before-started")
		return err
	})
	return err
}

func (exec *Executor) loadUnfinished() error {
	exec.eventHub.OnEvent(EventExecTaskLoadUnfinishedStart)
	unfinishedTasks, err := exec.execStatusStg.LoadUnfinishedTasks()
	if err != nil {
		return err
	}
	exec.eventHub.OnEvent(EventExecTaskLoadUnfinishedDone,
		EventDataWithName("load unfinished exec-tasks cnt", base.CloneableInt(len(unfinishedTasks))))
	for _, marshalledTask := range unfinishedTasks {
		var execTask ExecutorTask
		err := execTask.FromString(marshalledTask, exec.physicPlanBuilder)
		if err != nil {
			return err
		}
		exec.addExecTaskToRunning(&execTask, "on-load-unifinished")
	}
	return nil
}

// TODO: better backoff, can't wait in main thread
func (exec *Executor) safeSetDagActionDone(execId, dagActionId string) {
	var err error
	for {
		err = exec.dagStatusStg.SetDone(execId, dagActionId)
		if err != nil {
			time.Sleep(time.Millisecond * 100)
		} else {
			return
		}
	}
}

func (exec *Executor) safeSetDagActionError(execId, dagActionId string, actionErr error) {
	if err := exec.dagStatusStg.SetError(execId, dagActionId, actionErr); err != nil {
		tilog.PlainWarn("set error for action failed",
			zap.String("id", dagActionId),
			zap.Error(err),
		)
	}
}

// should not wait in main thread, do this in goroutine, use chan to return to main thread
func (exec *Executor) safeSetExecTaskFinished(execId string) {
	var err error
	for {
		err = exec.execStatusStg.SetTaskFinished(execId)
		if err != nil {
			exec.eventHub.OnEvent(EventExecStgError,
				EventDataWithType(EventDataError, base.NewCloneableError(err)))
			time.Sleep(exec.stgErrorWaitInterval)
		} else {
			return
		}
	}
}

// should not wait in main thread, do this in goroutine, use chan to return to main thread
func (exec *Executor) safeQueueTaskActions(execTask *ExecutorTask, reason string) (waitingActionCnt int) {
	var err error
	for i := 0; ; i++ {
		waitingActionCnt, err = exec.queueTaskActions(execTask, fmt.Sprintf("%s-#%v-try", reason, i))
		if err != nil {
			exec.eventHub.OnEvent(EventExecStgError,
				EventDataWithType(EventDataError, base.NewCloneableError(err)))
			time.Sleep(exec.stgErrorWaitInterval)
		} else {
			return waitingActionCnt
		}
	}
}

func (exec *Executor) tryAbortRunningActionsOfExecTask(execId string) {
	succeededAborts, failedAborts := exec.runningExecTasksStatus.TryAbortActionsOfExecTask(execId)
	for _, it := range succeededAborts {
		exec.eventHub.OnEvent(EventSchedulerSendAbortSignalToAction,
			EventDataWithName("action name", base.CloneableString(it.ActionTask.DagAction.Action.Name())),
			EventDataWithName("in running", base.CloneableBool(true)),
			EventDataWithName("send succeeded", base.CloneableBool(true)))
	}
	for _, it := range failedAborts {
		exec.eventHub.OnEvent(EventSchedulerSendAbortSignalToAction,
			EventDataWithName("action name", base.CloneableString(it.ActionTask.DagAction.Action.Name())),
			EventDataWithName("in running", base.CloneableBool(true)),
			EventDataWithName("send succeeded", base.CloneableBool(false)))
	}
}

// CleanFinished cleans up finished task data that has been retained longer than keepDeletedDuration
// from both memory and storage. It also enforces the maxFinishedTasks count limit by deleting
// the oldest finished tasks when the count exceeds the limit.
// Returns the total number of tasks cleaned from storage.
func (exec *Executor) CleanFinished() (int64, error) {
	// Clean expired finished tasks from in-memory deleted lists
	exec.runningExecTasks.CleanExpired()
	exec.runningExecTasksStatus.CleanExpired()

	// Clean expired finished task records from storage (time-based + count-based)
	cleaned, err := exec.execStatusStg.CleanFinishedTasks(exec.runningExecTasks.GetKeepDeletedDuration(), exec.maxFinishedTasks)
	if err != nil {
		return 0, fmt.Errorf("failed to clean finished tasks from storage: %w", err)
	}

	return cleaned, nil
}