package common

import (
	"context"
	"fmt"

	v1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/labels"
	"sigs.k8s.io/controller-runtime/pkg/client"

	"github.com/tidbcloud/eks-provider/controllers/dm/task"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
)

// PVCGC delete persistent volumes during dmjob deletion
type PVCGC struct {
	*task.BaseTask
}

func NewPVCGC(ctx *task.Context) *PVCGC {
	return &PVCGC{BaseTask: &task.BaseTask{Context: ctx}}
}

func (t *PVCGC) Name() string {
	return "pvcgc"
}

// Delete the pvc of the dmcluster
func (t *PVCGC) Delete(ctx context.Context) deploy.Result {

	if !t.IsDataplaneExist() {
		return deploy.Done()
	}
	cli := t.EnvProvider.GetClient()
	pvcs := &v1.PersistentVolumeClaimList{}
	ops := k8s.ListOpts[*v1.PersistentVolumeClaim]{
		ObjectList: pvcs,
		Unpack:     func() []*v1.PersistentVolumeClaim { return k8s.MapPtr(pvcs.Items) },
	}
	list, err := ops.ListFilterBy(ctx, cli, func(_ *v1.PersistentVolumeClaim) bool { return true }, t.pvcSelector()...)
	if client.IgnoreNotFound(err) != nil {
		return deploy.Error(err)
	}

	for _, pvc := range list {
		t.Logger.Info("deleting dmcluster pvc", "name", pvc.Name)
		if err := cli.Delete(ctx, pvc); client.IgnoreNotFound(err) != nil {
			return deploy.Error(err)
		}
	}

	return deploy.Done()
}

func (t *PVCGC) PostDelete(ctx context.Context) deploy.Result {

	// check pvc is deleted
	cli := t.EnvProvider.GetClient()
	pvcs := &v1.PersistentVolumeClaimList{}
	ops := k8s.ListOpts[*v1.PersistentVolumeClaim]{
		ObjectList: pvcs,
		Unpack:     func() []*v1.PersistentVolumeClaim { return k8s.MapPtr(pvcs.Items) },
	}
	list, err := ops.ListFilterBy(ctx, cli, func(_ *v1.PersistentVolumeClaim) bool { return true }, t.pvcSelector()...)
	if client.IgnoreNotFound(err) != nil {
		return deploy.Error(err)
	}

	if len(list) != 0 {
		return deploy.Error(fmt.Errorf("dmcluster pvc still exists, remain %d", len(list)))
	}
	return deploy.Done()
}

func (t *PVCGC) pvcSelector() []client.ListOption {
	return []client.ListOption{
		client.MatchingLabelsSelector{
			Selector: labels.Set{
				consts.InstanceLabelKey: GetDMClusterName(t.DMJob),
				consts.NameLabelKey:     "dm-cluster",
			}.AsSelector(),
		},
		client.InNamespace(GetDMClusterNamespace(t.DMJob, t.GetRegion())),
	}
}