package executor

import (
	"encoding/json"
	"fmt"
	"sync"
	"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"
)

const (
	ExecTaskStateTypeUnknown          ExecTaskStateType = "unknown"
	ExecTaskStateTypePending          ExecTaskStateType = "pending"
	ExecTaskStateTypeAborting         ExecTaskStateType = "aborting"
	ExecTaskStateTypeRunningOK        ExecTaskStateType = "running.ok"
	ExecTaskStateTypeRunningWithError ExecTaskStateType = "running.error"
	ExecTaskStateTypeFinishedAborted  ExecTaskStateType = "finished.aborted"
	ExecTaskStateTypeFinishedDone     ExecTaskStateType = "finished.done"
)

type ExecTaskStateType string

type ExecTaskStatusEx struct {
	ExecTaskStatus
	State ExecTaskStateType
}

func NewExecTaskStatusEx(origin *ExecTaskStatus, state ExecTaskStateType) *ExecTaskStatusEx {
	if origin == nil {
		return nil
	}
	return &ExecTaskStatusEx{
		ExecTaskStatus: *origin,
		State:          state,
	}
}

type ExecTaskStatus struct {
	ExecTask      *ExecutorTask
	ActionsStatus ActionTaskIdStatus
}

func NewExecTaskStatus(execTask *ExecutorTask) *ExecTaskStatus {
	return &ExecTaskStatus{
		ExecTask:      execTask,
		ActionsStatus: make(ActionTaskIdStatus),
	}
}

func (status *ExecTaskStatus) Clone() *ExecTaskStatus {
	return &ExecTaskStatus{
		ExecTask:      status.ExecTask,
		ActionsStatus: status.ActionsStatus.Clone(),
	}
}

func (status *ExecTaskStatus) HasError() bool {
	if status.ActionsStatus == nil {
		return false
	}
	return status.ActionsStatus.HasError()
}

type ExecTaskIdStatus map[string]*ExecTaskStatusEx

func (status ExecTaskIdStatus) First() (string, *ExecTaskStatusEx) {
	for key, execStatus := range status {
		return key, execStatus
	}
	return "", nil
}

func NewExecTaskIdStatus() ExecTaskIdStatus {
	return make(ExecTaskIdStatus)
}

type ExecutorTask struct {
	LogicPlan  dag.Dag            `json:"logicPlan,omitempty"`
	PhysicPlan dag.Dag            `json:"-"`
	RootName   string             `json:"rootName,omitempty"`
	ExecId     string             `json:"execId,omitempty"`
	Args       actions.ActionArgs `json:"args,omitempty"`
	StartTime  time.Time
}

func (task *ExecutorTask) Clone() *ExecutorTask {
	return &ExecutorTask{
		LogicPlan:  task.LogicPlan,
		PhysicPlan: task.PhysicPlan,
		RootName:   task.RootName,
		ExecId:     task.ExecId,
		Args:       task.Args,
		StartTime:  task.StartTime,
	}
}

func (task *ExecutorTask) CloneableClone() base.Cloneable {
	return task.Clone()
}

func (task *ExecutorTask) String() string {
	return fmt.Sprintf("%v@%v => %v", task.RootName, task.PhysicPlan, task.ExecId)
}

func (task *ExecutorTask) ToString() (string, error) {
	bytes, err := json.Marshal(task)
	if err != nil {
		return "", err
	}
	return string(bytes), nil
}

type ExecutorTasks []*ExecutorTask

func (tasks ExecutorTasks) Clone() ExecutorTasks {
	var clone ExecutorTasks
	for _, task := range tasks {
		clone = append(clone, task.Clone())
	}
	return clone
}

func (tasks ExecutorTasks) CloneableClone() base.Cloneable {
	return tasks.Clone()
}

func (task *ExecutorTask) FromString(str string, planBuilder *PhysicPlanBuilder) error {
	err := json.Unmarshal([]byte(str), task)
	if err != nil {
		return err
	}
	if task.StartTime.IsZero() {
		task.StartTime = time.Now()
	}
	task.PhysicPlan, err = planBuilder.BuildPlan(task.RootName, task.LogicPlan, task.ExecId)
	return err
}

type DeletedExecutorTask struct {
	task    *ExecutorTask
	delTime time.Time
}

type SafeExecutorTasks struct {
	tasks               map[string]*ExecutorTask
	deleted             []DeletedExecutorTask
	deletedSet          map[string]bool
	keepDeletedDuration time.Duration
	locker              sync.Mutex
}

func NewSafeExecutorTasks(keepDeletedDuration time.Duration) *SafeExecutorTasks {
	return &SafeExecutorTasks{
		tasks:               make(map[string]*ExecutorTask),
		deletedSet:          make(map[string]bool),
		keepDeletedDuration: keepDeletedDuration,
	}
}

func removeFromDeletedList(deletedList []DeletedExecutorTask, execId string) (out []DeletedExecutorTask) {
	for _, it := range deletedList {
		if it.task.ExecId != execId {
			out = append(out, it)
		}
	}
	return out
}

func (tasks *SafeExecutorTasks) Add(execTask *ExecutorTask) (failToAddByDuplicated bool) {
	tasks.locker.Lock()
	defer tasks.locker.Unlock()
	execId := execTask.ExecId
	if tasks.tasks[execId] != nil {
		return true
	}
	tasks.tasks[execId] = execTask
	// handle duplicated tasks:
	//   a deleted task might be a cancaled task
	//   in that case if it's been adding back, it should be removed from deleted-list and run
	if tasks.deletedSet[execId] {
		delete(tasks.deletedSet, execId)
		tasks.deleted = removeFromDeletedList(tasks.deleted, execId)
	}
	return false
}

func (tasks *SafeExecutorTasks) Delete(execId string) {
	tasks.locker.Lock()
	defer tasks.locker.Unlock()

	task := tasks.tasks[execId]
	if task == nil {
		return
	}
	delete(tasks.tasks, execId)
	tasks.deleted = append(tasks.deleted, DeletedExecutorTask{
		task:    task,
		delTime: time.Now(),
	})
	tasks.deletedSet[execId] = true

	// only trigger deleting on `Delete` call,
	// this will make deleted data stay in `SafeExecTaskIdStatus` longer than expected, but it's ok
	expiredTime := time.Now().Add(-tasks.keepDeletedDuration)
	for i, it := range tasks.deleted {
		if it.delTime.Before(expiredTime) {
			continue
		}
		tasks.deleted = tasks.deleted[i:]
		delete(tasks.deletedSet, execId)
		return
	}
	tasks.deleted = nil
}

func (tasks *SafeExecutorTasks) Get(execId string) *ExecutorTask {
	tasks.locker.Lock()
	defer tasks.locker.Unlock()
	return tasks.tasks[execId]
}

// TODO: if the process(instance) rebooted, the finished task won't be in deleted-list
func (tasks *SafeExecutorTasks) IsDeleted(execId string) bool {
	tasks.locker.Lock()
	defer tasks.locker.Unlock()
	return tasks.deletedSet[execId]
}

func (tasks *SafeExecutorTasks) Len() int {
	tasks.locker.Lock()
	defer tasks.locker.Unlock()
	return len(tasks.tasks)
}

func (tasks *SafeExecutorTasks) ExecIds() (result []string) {
	tasks.locker.Lock()
	defer tasks.locker.Unlock()
	for _, task := range tasks.tasks {
		result = append(result, task.ExecId)
	}
	return result
}

func (tasks *SafeExecutorTasks) List() (result ExecutorTasks) {
	tasks.locker.Lock()
	defer tasks.locker.Unlock()
	for _, task := range tasks.tasks {
		result = append(result, task)
	}
	return result
}

func (tasks *SafeExecutorTasks) Iterate(includeDeleted bool, fn func(*ExecutorTask) error) error {
	tasks.locker.Lock()
	defer tasks.locker.Unlock()
	for _, execTask := range tasks.tasks {
		err := fn(execTask)
		if err != nil {
			return err
		}
	}
	if includeDeleted {
		for _, it := range tasks.deleted {
			err := fn(it.task)
			if err != nil {
				return err
			}
		}
	}
	return nil
}

// GetKeepDeletedDuration returns the configured retention duration for deleted tasks.
func (tasks *SafeExecutorTasks) GetKeepDeletedDuration() time.Duration {
	tasks.locker.Lock()
	defer tasks.locker.Unlock()
	return tasks.keepDeletedDuration
}

// SetKeepDeletedDuration updates the retention duration for deleted tasks.
func (tasks *SafeExecutorTasks) SetKeepDeletedDuration(dur time.Duration) {
	tasks.locker.Lock()
	defer tasks.locker.Unlock()
	tasks.keepDeletedDuration = dur
}

// CleanExpired removes only entries older than keepDeletedDuration from the deleted list.
func (tasks *SafeExecutorTasks) CleanExpired() {
	tasks.locker.Lock()
	defer tasks.locker.Unlock()
	expiredTime := time.Now().Add(-tasks.keepDeletedDuration)
	var remaining []DeletedExecutorTask
	for _, it := range tasks.deleted {
		if it.delTime.After(expiredTime) {
			remaining = append(remaining, it)
		} else {
			delete(tasks.deletedSet, it.task.ExecId)
		}
	}
	tasks.deleted = remaining
}

type DeletingExecTaskStatus struct {
	execId         string
	delTime        time.Time
	doneNotAborted bool
}

type SafeExecTaskIdStatus struct {
	data                map[string]*ExecTaskStatus
	deletingList        []*DeletingExecTaskStatus
	deletingNames       map[string]*DeletingExecTaskStatus
	keepDeletedDuration time.Duration
	locker              sync.Mutex
}

func NewSafeExecTaskIdStatus(keepDeletedDuration time.Duration) *SafeExecTaskIdStatus {
	return &SafeExecTaskIdStatus{
		data:                make(map[string]*ExecTaskStatus),
		deletingNames:       make(map[string]*DeletingExecTaskStatus),
		keepDeletedDuration: keepDeletedDuration,
	}
}

func (status *SafeExecTaskIdStatus) EnsureExisted(execTask *ExecutorTask) (created bool) {
	status.locker.Lock()
	defer status.locker.Unlock()

	// handle duplicated tasks:
	//   a deleted task might be a cancaled task
	//   in that case if it's been adding back, it should be removed from deleted-list and run
	if status.deletingNames[execTask.ExecId] != nil {
		delete(status.deletingNames, execTask.ExecId)
		status.deletingList = removeFromDeletingStatusList(status.deletingList, execTask.ExecId)
	}

	it := status.data[execTask.ExecId]
	if it == nil {
		it = NewExecTaskStatus(execTask.Clone())
		status.data[execTask.ExecId] = it
		return true
	}
	return false
}

func removeFromDeletingStatusList(deletingList []*DeletingExecTaskStatus, execId string) (out []*DeletingExecTaskStatus) {
	for _, it := range deletingList {
		if it.execId != execId {
			out = append(out, it)
		}
	}
	return out
}

func (status *SafeExecTaskIdStatus) IsRunning(execId string) (isRunning, hasError bool) {
	status.locker.Lock()
	defer status.locker.Unlock()
	runningStatus := status.data[execId]
	if runningStatus == nil {
		return false, false
	}
	return true, runningStatus.HasError()
}

func (status *SafeExecTaskIdStatus) IsDeleting(execId string) (isDeleting, doneNotAborted bool) {
	status.locker.Lock()
	defer status.locker.Unlock()
	deletingStatus := status.deletingNames[execId]
	if deletingStatus == nil {
		return false, false
	}
	return true, deletingStatus.doneNotAborted
}

func (status *SafeExecTaskIdStatus) Delete(execId string, doneNotAborted bool) {
	status.locker.Lock()
	defer status.locker.Unlock()

	if status.deletingNames[execId] != nil {
		return
	}
	deletingStatus := &DeletingExecTaskStatus{
		execId:         execId,
		delTime:        time.Now(),
		doneNotAborted: doneNotAborted,
	}
	status.deletingNames[execId] = deletingStatus
	status.deletingList = append(status.deletingList, deletingStatus)

	// only trigger deleting on `Delete` call,
	// this will make deleted data stay in `SafeExecTaskIdStatus` longer than expected, but it's ok
	expiredTime := time.Now().Add(-status.keepDeletedDuration)
	for i, it := range status.deletingList {
		if it.delTime.Before(expiredTime) {
			delete(status.data, it.execId)
			delete(status.deletingNames, it.execId)
		} else {
			status.deletingList = status.deletingList[i:]
			return
		}
	}
	status.deletingList = nil
	status.deletingNames = make(map[string]*DeletingExecTaskStatus)
}

// SetKeepDeletedDuration updates the retention duration for deleting task status.
func (status *SafeExecTaskIdStatus) SetKeepDeletedDuration(dur time.Duration) {
	status.locker.Lock()
	defer status.locker.Unlock()
	status.keepDeletedDuration = dur
}

// CleanExpired removes only entries older than keepDeletedDuration from the deleting list and their data.
func (status *SafeExecTaskIdStatus) CleanExpired() {
	status.locker.Lock()
	defer status.locker.Unlock()
	expiredTime := time.Now().Add(-status.keepDeletedDuration)
	var remaining []*DeletingExecTaskStatus
	for _, it := range status.deletingList {
		if it.delTime.After(expiredTime) {
			remaining = append(remaining, it)
		} else {
			delete(status.data, it.execId)
			delete(status.deletingNames, it.execId)
		}
	}
	status.deletingList = remaining
}

func (status *SafeExecTaskIdStatus) GetAndClone(execId string) (*ExecTaskStatus, bool) {
	status.locker.Lock()
	defer status.locker.Unlock()
	it := status.data[execId]
	if it == nil {
		return nil, false
	}
	return it.Clone(), true
}

func (status *SafeExecTaskIdStatus) IsDagActionQueued(execId, dagActionId string) bool {
	status.locker.Lock()
	defer status.locker.Unlock()
	it := status.data[execId]
	if it == nil {
		return false
	}
	actionStatus := it.ActionsStatus[dagActionId]
	if actionStatus == nil {
		return false
	}
	return actionStatus.Queued
}

func (status *SafeExecTaskIdStatus) IsDagActionStatusExisted(execId, dagActionId string) bool {
	status.locker.Lock()
	defer status.locker.Unlock()
	it := status.data[execId]
	if it == nil {
		return false
	}
	return it.ActionsStatus[dagActionId] != nil
}

func (status *SafeExecTaskIdStatus) IsAllDone(execId string, includeUnqueued bool) bool {
	status.locker.Lock()
	defer status.locker.Unlock()
	it := status.data[execId]
	if it == nil {
		return false
	}
	return it.ActionsStatus.IsAllDone(includeUnqueued)
}

func (status *SafeExecTaskIdStatus) PutActionTaskStatus(target *ActionTaskStatus) {
	status.locker.Lock()
	defer status.locker.Unlock()
	execTask := target.ActionTask.ExecTask
	execId := execTask.ExecId
	it, ok := status.data[execId]
	if !ok {
		execTask = execTask.Clone()
		it = NewExecTaskStatus(execTask)
	}
	dagActionId := target.ActionTask.DagAction.Id()
	it.ActionsStatus[dagActionId] = target.Clone()
	status.data[execId] = it
}

func (status *SafeExecTaskIdStatus) TryAbortActionsOfExecTask(execId string) (succeeded, failed []*ActionTaskStatus) {
	status.locker.Lock()
	defer status.locker.Unlock()

	it := status.data[execId]
	if it == nil {
		return nil, nil
	}

	for _, actionStatus := range it.ActionsStatus {
		if !actionStatus.Running {
			continue
		}
		action := actionStatus.ActionTask.DagAction.Action
		ok := actions.TryAbortAction(action)
		if ok {
			succeeded = append(succeeded, actionStatus.Clone())
		} else {
			failed = append(failed, actionStatus.Clone())
		}
	}
	return succeeded, failed
}

type SafeStrSet struct {
	strs   map[string]bool
	locker sync.Mutex
}

func NewSafeStrSet() *SafeStrSet {
	return &SafeStrSet{
		strs: make(map[string]bool),
	}
}

func (sss *SafeStrSet) Add(str string) {
	sss.locker.Lock()
	defer sss.locker.Unlock()
	sss.strs[str] = true
}

func (sss *SafeStrSet) Delete(str string) {
	sss.locker.Lock()
	defer sss.locker.Unlock()
	delete(sss.strs, str)
}

func (sss *SafeStrSet) Has(str string) bool {
	sss.locker.Lock()
	defer sss.locker.Unlock()
	return sss.strs[str]
}

func (sss *SafeStrSet) CheckHasAndAdd(str string) bool {
	sss.locker.Lock()
	defer sss.locker.Unlock()
	has := sss.strs[str]
	sss.strs[str] = true
	return has
}