package common

import (
	"context"
	"fmt"

	oplabel "github.com/pingcap/tidb-operator/pkg/apis/label"
	v1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/labels"
	"sigs.k8s.io/controller-runtime/pkg/client"

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

var _ deploy.Task = &PV{}

// PV reclaims the persistent volumes during cluster deletion
type PV struct {
	*task.BaseTask
}

func NewPV(ctx *task.Context) *PV {
	return &PV{BaseTask: &task.BaseTask{Context: ctx}}
}

func (t *PV) Name() string {
	return "pv"
}

func (t *PV) PostCreate(ctx context.Context) deploy.Result {
	if t.TCGetter == nil || !t.TCGetter.IsReady() {
		return deploy.Done() // do nothing
	}

	cli := t.EnvProvider.GetClient()

	if err := t.deleteDeferDeletingPVCs(ctx, cli); err != nil {
		return deploy.Error(fmt.Errorf("delete defer deleting PVCs: %v", err))
	}

	return deploy.Done()
}

func (t *PV) Delete(ctx context.Context) deploy.Result {
	// Delete the PVCs in the namespace
	// if delete PVs directly, it would lead to resource leak
	cli := t.EnvProvider.GetClient()
	return deploy.ErrorOrDone(cli.DeleteAllOf(ctx, &v1.PersistentVolumeClaim{}, client.InNamespace(t.Namer.GetTiDBNamespace())))
}

func (t *PV) PostDelete(ctx context.Context) deploy.Result {
	cli := t.EnvProvider.GetClient()
	pvs := &v1.PersistentVolumeList{}
	ops := k8s.ListOpts[*v1.PersistentVolume]{
		ObjectList: pvs,
		Unpack:     func() []*v1.PersistentVolume { return k8s.MapPtr(pvs.Items) },
	}
	list, err := ops.ListFilterBy(ctx, cli, func(_ *v1.PersistentVolume) bool { return true }, t.pvSelector())
	if client.IgnoreNotFound(err) != nil {
		return deploy.Error(err)
	}
	if len(list) != 0 {
		return deploy.Error(fmt.Errorf("persistent volume deleted, remain %d", len(list)))
	}
	return deploy.Done()
}

func (t *PV) pvSelector() client.MatchingLabelsSelector {
	return client.MatchingLabelsSelector{
		Selector: labels.Set{
			consts.NamespaceLabelKey: t.TidbClusterMetaObject().Namespace,
		}.AsSelector(),
	}
}

// after scale in, the PVCs of the scaled-in pods are still in the cluster (tidb-operator will not delete them),
// we need to delete them.
func (t *PV) deleteDeferDeletingPVCs(ctx context.Context, cli client.Client) error {
	pvcs := &v1.PersistentVolumeClaimList{}
	ops := k8s.ListOpts[*v1.PersistentVolumeClaim]{
		ObjectList: pvcs,
		Unpack:     func() []*v1.PersistentVolumeClaim { return k8s.MapPtr(pvcs.Items) },
	}

	// list all PVCs in the namespace with `tidb.pingcap.com/pvc-defer-deleting` annotation
	list, err := ops.ListFilterBy(ctx, cli, func(pvc *v1.PersistentVolumeClaim) bool {
		if _, ok := pvc.Annotations[oplabel.AnnPVCDeferDeleting]; ok {
			return true
		}
		return false
	}, client.InNamespace(t.TidbClusterMetaObject().Namespace))
	if client.IgnoreNotFound(err) != nil {
		return err
	}

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

	return nil
}