package dualdisk

import (
	"context"
	"fmt"
	"sort"
	"strconv"
	"strings"

	"github.com/mitchellh/mapstructure"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"

	"github.com/tidbcloud/eks-provider/controllers/workflow/action"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/dataplane"
)

var _ action.Action = &createPVCs{}

// createPVCs creates PVCs for a tidb cluster component.
type createPVCs struct {
	cli            client.Client
	cachedProvider dataplane.EnvProviderManager
}

type createPVCsInputs struct {
	// ClusterNamespace is the namespace of the infra Cluster
	ClusterNamespace string `mapstructure:"clusterNamespace" json:"clusterNamespace" validate:"required"`
	// ClusterName is the name of the infra Cluster
	ClusterName string `mapstructure:"clusterName" json:"clusterName" validate:"required"`
	// Namespace is the namespace of the dataplane TiDBCluster
	Namespace string `mapstructure:"namespace" json:"namespace" validate:"required"`
	// ComponentName is the name of the component, only TiKV supported now
	ComponentType string `mapstructure:"componentType" json:"componentType" validate:"required"`
	// BaseName is the base name of the PVCs to be created
	BaseName         string `mapstructure:"baseName" json:"baseName" validate:"required"`
	StorageClassName string `mapstructure:"storageClassName" json:"storageClassName" validate:"required"`
	SizeGi           string `mapstructure:"sizeGi" json:"sizeGi" validate:"required"` // string for decode
}

func NewCreatePVCs(cli client.Client, cachedProvider dataplane.EnvProviderManager) action.Action {
	return &createPVCs{cli: cli, cachedProvider: cachedProvider}
}

func (a *createPVCs) Name() cloudv1.WorkflowActionName {
	return "CreatePVCsAction"
}

func (a *createPVCs) PreCheck(ctx context.Context, wf *cloudv1.Workflow, step cloudv1.WorkflowStep) error {
	inputs := &createPVCsInputs{}
	if err := mapstructure.Decode(step.Inputs.Object, inputs); err != nil {
		return fmt.Errorf("decode inputs: %w", err)
	}
	if err := validate.Struct(inputs); err != nil {
		return fmt.Errorf("validate inputs: %w", err)
	}
	return nil
}

func (a *createPVCs) Handle(ctx context.Context, wf *cloudv1.Workflow, step cloudv1.WorkflowStep) error {
	inputs := &createPVCsInputs{}
	_ = mapstructure.Decode(step.Inputs.Object, inputs)

	cluster := &cloudv1.Cluster{}
	if err := a.cli.Get(ctx, client.ObjectKey{Namespace: inputs.ClusterNamespace, Name: inputs.ClusterName}, cluster); err != nil {
		if apierrors.IsNotFound(err) {
			return cloudv1.NewWFNonRetriableError(fmt.Errorf("cluster %s/%s not found", inputs.ClusterNamespace, inputs.ClusterName))
		}
		return fmt.Errorf("failed to get cluster %s/%s: %w", inputs.ClusterNamespace, inputs.ClusterName, err)
	}

	remoteCli, err := a.cachedProvider.GetClient(ctx, cluster)
	if err != nil {
		return fmt.Errorf("get dataplane cli: %w", err)
	}

	if inputs.ComponentType != string(cloudv1.ComponentTiKV) {
		return cloudv1.NewWFNonRetriableError(fmt.Errorf("component type %s not supported", inputs.ComponentType))
	}

	// as we may have "delete-slots", so we need to get the current real TiKV pods name
	tikvPodList := &corev1.PodList{}
	if err := remoteCli.List(ctx, tikvPodList,
		client.InNamespace(inputs.Namespace),
		client.MatchingLabels{
			"app.kubernetes.io/component":  "tikv",
			"app.kubernetes.io/instance":   consts.ClusterPrefix, // "db"
			"app.kubernetes.io/managed-by": "tidb-operator",
			"app.kubernetes.io/name":       "tidb-cluster",
		}); err != nil {
		return fmt.Errorf("list tikv pods: %w", err)
	}

	if len(tikvPodList.Items) != cluster.Spec.Cluster.TiKV.Replicas {
		return cloudv1.NewWFNonRetriableError(fmt.Errorf(
			"tikv pod count %d not match spec %d", len(tikvPodList.Items), cluster.Spec.Cluster.TiKV.Replicas))
	}

	tikvPodNames := make([]string, 0, len(tikvPodList.Items))
	for _, pod := range tikvPodList.Items {
		tikvPodNames = append(tikvPodNames, pod.Name)
	}
	// sort the pod names based on the ordinal index
	sortPodNames(tikvPodNames)

	for _, podName := range tikvPodNames {
		pvc := &corev1.PersistentVolumeClaim{
			ObjectMeta: metav1.ObjectMeta{
				Namespace: inputs.Namespace,
				Name:      fmt.Sprintf("%s-%s", inputs.BaseName, podName),
				Labels: map[string]string{
					// some other labels (e.g. "tidb.pingcap.com/pod-name") will be added by the TiDB Operator
					"app.kubernetes.io/component":  "tikv",
					"app.kubernetes.io/instance":   consts.ClusterPrefix, // "db"
					"app.kubernetes.io/managed-by": "tidb-operator",
					"app.kubernetes.io/name":       "tidb-cluster",
				},
			},
			Spec: corev1.PersistentVolumeClaimSpec{
				AccessModes: []corev1.PersistentVolumeAccessMode{corev1.ReadWriteOnce},
				Resources: corev1.VolumeResourceRequirements{
					Requests: corev1.ResourceList{
						corev1.ResourceStorage: resource.MustParse(fmt.Sprintf("%sGi", inputs.SizeGi)),
					},
				},
				StorageClassName: &inputs.StorageClassName,
			},
		}
		if err := remoteCli.Create(ctx, pvc); err != nil && !apierrors.IsAlreadyExists(err) {
			return fmt.Errorf("create pvc %s/%s: %w", inputs.Namespace, pvc.Name, err)
		}
	}

	// the podNames will be used by another step in the raft log migration case
	stepStatus, _ := wf.StepStatus(ctx, step.Name)
	return stepStatus.UpdateOutput(ctx, map[string]string{
		"podNames": strings.Join(tikvPodNames, ","),
	})
}

// sortPodNames based on the ordinal index of the pod name, e.g. "tikv-0", "tikv-1", ...
// NOTE: "tikv-0" < "tikv-1" < "tikv-2" < ... < "tikv-9" < "tikv-10" < "tikv-11" < ... < "tikv-20"
func sortPodNames(podNames []string) {
	sort.Slice(podNames, func(i, j int) bool {
		// extract the ordinal index from the pod name, get the last part after the last '-'
		iIndex := strings.LastIndex(podNames[i], "-")
		jIndex := strings.LastIndex(podNames[j], "-")
		if iIndex == -1 || jIndex == -1 {
			return podNames[i] < podNames[j] // fallback to string comparison
		}
		iOrdinal := podNames[i][iIndex+1:]
		jOrdinal := podNames[j][jIndex+1:]
		// convert the ordinal index to int for comparison
		ii, err := strconv.Atoi(iOrdinal)
		if err != nil {
			return podNames[i] < podNames[j] // fallback to string comparison
		}
		jj, err := strconv.Atoi(jOrdinal)
		if err != nil {
			return podNames[i] < podNames[j] // fallback to string comparison
		}
		return ii < jj || (ii == jj && podNames[i] < podNames[j]) // compare the ordinal index
	})
}