package dualdisk

import (
	"context"
	"fmt"

	"github.com/mitchellh/mapstructure"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	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/dataplane"
	astsv1 "github.com/tidbcloud/infra-provider/pkg/pingcap/apis/v1"
)

var _ action.Action = &deleteOneASTS{}

// deleteOneASTS deletes one ASTS of a tidb cluster.
type deleteOneASTS struct {
	cli            client.Client
	cachedProvider dataplane.EnvProviderManager
}

type deleteOneASTSInputs 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 ASTS
	Namespace string `mapstructure:"namespace" json:"namespace" validate:"required"`
	// Name is the name of the ASTS
	Name string `mapstructure:"name" json:"name" validate:"required"`
}

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

func (a *deleteOneASTS) Name() cloudv1.WorkflowActionName {
	return "DeleteOneASTSAction"
}

func (a *deleteOneASTS) PreCheck(ctx context.Context, wf *cloudv1.Workflow, step cloudv1.WorkflowStep) error {
	inputs := &deleteOneASTSInputs{}
	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 *deleteOneASTS) Handle(ctx context.Context, wf *cloudv1.Workflow, step cloudv1.WorkflowStep) error {
	inputs := &deleteOneASTSInputs{}
	_ = 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)
	}

	asts := &astsv1.StatefulSet{}
	if err := remoteCli.Get(ctx, client.ObjectKey{Namespace: inputs.Namespace, Name: inputs.Name}, asts); err != nil {
		// If the ASTS is deleted by a previous call but interrupted accidentally,
		// we can retry this error and delete it again.
		return fmt.Errorf("failed to get asts %s/%s: %w", inputs.Namespace, inputs.Name, err)
	}
	// NOTE: Orphan deletion is used here to avoid deleting the Pods
	if err := remoteCli.Delete(ctx, asts, client.PropagationPolicy(metav1.DeletePropagationOrphan)); err != nil {
		return fmt.Errorf("failed to delete asts %s/%s: %w", inputs.Namespace, inputs.Name, err)
	}

	return nil
}