// Package runtime provides runtime storage implementations for executor tasks in etcd.
package runtime

import (
	"context"
	"fmt"
	"sort"
	"time"

	v1 "github.com/tidbcloud/cluster-service-ng/pkg/storage/etcd/schema/v1"
	"github.com/tidbcloud/pkgv2/log"
	"go.uber.org/zap"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	apitypes "k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

const (
	executorTaskScopeKey = "executorTaskScope"

	executorTaskFinishedKey     = "executorTaskFinished"
	executorTaskUnFinishedValue = "false"
	executorTaskFinishedValue   = "true"

	executorTaskFinishedAtKey = "executorTaskFinishedAt"
)

type executorTaskStorage struct {
	scope      string
	storageCli client.Client
}

func NewExecutorTaskStorage(scope string, storageCli client.Client) *executorTaskStorage {
	return &executorTaskStorage{
		scope:      scope,
		storageCli: storageCli,
	}
}

func (stg *executorTaskStorage) OnNewTask(execId, marshalledExecTask string) error {
	ctx := context.TODO()
	err := stg.storageCli.Get(ctx, client.ObjectKey{Name: execId}, &v1.Task{})
	if err == nil {
		log.Info(ctx, "executor task is existed", zap.String("execId", execId))
		return nil
	}
	if !apierrors.IsNotFound(err) {
		return err
	}

	return stg.storageCli.Create(ctx, &v1.Task{
		ObjectMeta: metav1.ObjectMeta{
			Name: execId,
			Labels: map[string]string{
				executorTaskScopeKey:    stg.scope,
				executorTaskFinishedKey: executorTaskUnFinishedValue,
			},
		},
		Spec: v1.TaskSpec{
			ExecutorTask: marshalledExecTask,
			DoneTaskList: make(map[string]bool),
		},
	})
}

func (stg *executorTaskStorage) SetTaskFinished(execId string) error {
	ctx := context.TODO()
	task := new(v1.Task)
	task.Name = execId

	// Patch both the finished label and the finished-at annotation in one merge-patch.
	now := time.Now().UTC().Format(time.RFC3339)
	data := []byte(fmt.Sprintf(
		`{"metadata":{"labels":{%q:%q},"annotations":{%q:%q}}}`,
		executorTaskFinishedKey, executorTaskFinishedValue,
		executorTaskFinishedAtKey, now,
	))
	patch := client.RawPatch(apitypes.MergePatchType, data)
	return stg.storageCli.Patch(ctx, task, patch)
}

func (stg *executorTaskStorage) IsTaskFinished(execId string) (bool, error) {
	ctx := context.TODO()
	var task v1.Task
	err := stg.storageCli.Get(ctx, client.ObjectKey{Name: execId}, &task)
	if err != nil {
		return false, err
	}
	return task.Labels[executorTaskFinishedKey] == executorTaskFinishedValue, nil
}

func (stg *executorTaskStorage) LoadUnfinishedTasks() (marshalledExecTasks map[string]string, err error) {
	ctx := context.TODO()
	var taskList v1.TaskList
	selector := client.MatchingLabelsSelector{
		Selector: labels.Set{
			executorTaskScopeKey:    stg.scope,
			executorTaskFinishedKey: executorTaskUnFinishedValue,
		}.AsSelector(),
	}

	if err := stg.storageCli.List(ctx, &taskList, selector); err != nil {
		return nil, err
	}

	out := make(map[string]string)
	for i := 0; i < len(taskList.Items); i++ {
		item := taskList.Items[i]
		out[item.Name] = item.Spec.ExecutorTask
	}

	return out, nil
}

func (stg *executorTaskStorage) SetDone(scope, id string) error {
	ctx := context.TODO()
	var task v1.Task
	err := stg.storageCli.Get(ctx, client.ObjectKey{Name: scope}, &task)
	if err != nil {
		return err
	}

	if task.Spec.DoneTaskList == nil {
		task.Spec.DoneTaskList = make(map[string]bool)
	}
	task.Spec.DoneTaskList[id] = true
	if err := stg.storageCli.Update(ctx, &task); err != nil {
		return err
	}
	condType := buildConditionType(id)
	cond := &metav1.Condition{
		Type:               condType,
		Status:             metav1.ConditionTrue,
		Reason:             "ActionSuccessed",
		Message:            "successed",
		LastTransitionTime: metav1.Now(),
	}
	task.Status.Conditions.SetCondition(cond)
	_ = stg.storageCli.Status().Update(ctx, &task)
	return nil
}

func (stg *executorTaskStorage) IsDone(scope, id string) (bool, error) {
	ctx := context.TODO()
	var task v1.Task
	err := stg.storageCli.Get(ctx, client.ObjectKey{Name: scope}, &task)
	if err != nil {
		return false, err
	}
	done, ok := task.Spec.DoneTaskList[id]
	if !ok {
		return false, nil
	}
	return done, nil
}

func (stg *executorTaskStorage) SetError(scope, id string, actionErr error) error {
	ctx := context.TODO()
	var task v1.Task
	err := stg.storageCli.Get(ctx, client.ObjectKey{Name: scope}, &task)
	if err != nil {
		return err
	}
	condType := buildConditionType(id)
	msg := actionErr.Error()
	if prev, _ := task.Status.Conditions.GetCondition(condType); prev != nil {
		// no need to update the status if the fields equal
		if prev.Status == metav1.ConditionFalse &&
			prev.Reason == "ActionFailed" &&
			prev.Message == msg {
			return nil
		}
	}
	cond := &metav1.Condition{
		Type:               condType,
		Status:             metav1.ConditionFalse,
		Reason:             "ActionFailed",
		Message:            msg,
		LastTransitionTime: metav1.Now(),
	}
	task.Status.Conditions.SetCondition(cond)
	return stg.storageCli.Status().Update(ctx, &task)
}

func buildConditionType(id string) string {
	return fmt.Sprintf("%s/Done", id)
}

// CleanFinishedTasks deletes finished task CRs from etcd storage.
// Time-based: tasks finished longer than retainDuration ago are deleted.
// Count-based: if maxFinishedTasks > 0, remaining finished tasks are capped
// at maxFinishedTasks with the oldest deleted first.
func (stg *executorTaskStorage) CleanFinishedTasks(retainDuration time.Duration, maxFinishedTasks int64) (int64, error) {
	ctx := context.TODO()
	var taskList v1.TaskList
	selector := client.MatchingLabelsSelector{
		Selector: labels.Set{
			executorTaskScopeKey:    stg.scope,
			executorTaskFinishedKey: executorTaskFinishedValue,
		}.AsSelector(),
	}

	if err := stg.storageCli.List(ctx, &taskList, selector); err != nil {
		return 0, fmt.Errorf("failed to list finished tasks: %w", err)
	}

	// First pass: time-based cleanup. Track surviving tasks for count-based pass.
	type taskWithTime struct {
		index      int
		finishTime time.Time
	}
	cutoff := time.Now().Add(-retainDuration)
	var cleaned int64
	var surviving []taskWithTime

	for i := range taskList.Items {
		ann := taskList.Items[i].Annotations
		var finishTime time.Time
		if ann != nil {
			if ts, ok := ann[executorTaskFinishedAtKey]; ok {
				parsed, err := time.Parse(time.RFC3339, ts)
				if err == nil {
					finishTime = parsed
				}
			}
		}
		// Legacy tasks without annotation: use CreationTimestamp as conservative finish time.
		// CreationTimestamp <= actual finish time, so if it's already past the cutoff, safe to delete.
		if finishTime.IsZero() {
			finishTime = taskList.Items[i].CreationTimestamp.Time
		}
		// Time-based: delete tasks older than or equal to retainDuration cutoff.
		if !finishTime.After(cutoff) {
			if err := stg.storageCli.Delete(ctx, &taskList.Items[i]); err != nil {
				if apierrors.IsNotFound(err) {
					continue
				}
				return cleaned, fmt.Errorf("failed to delete finished task %s: %w", taskList.Items[i].Name, err)
			}
			cleaned++
			continue
		}
		// Task survived time-based cleanup; track for count-based pass.
		surviving = append(surviving, taskWithTime{index: i, finishTime: finishTime})
	}

	// Second pass: enforce max finished tasks count limit.
	if maxFinishedTasks > 0 && int64(len(surviving)) > maxFinishedTasks {
		sort.Slice(surviving, func(a, b int) bool {
			return surviving[a].finishTime.Before(surviving[b].finishTime)
		})
		excess := int64(len(surviving)) - maxFinishedTasks
		for i := int64(0); i < excess; i++ {
			idx := surviving[i].index
			if err := stg.storageCli.Delete(ctx, &taskList.Items[idx]); err != nil {
				if apierrors.IsNotFound(err) {
					continue
				}
				return cleaned, fmt.Errorf("failed to delete excess finished task %s: %w", taskList.Items[idx].Name, err)
			}
			cleaned++
		}
	}

	return cleaned, nil
}