package controller

import (
	"fmt"
	"strings"

	"github.com/pingcap/log"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"github.com/pingcap/tidb-operator/pkg/label"
	"go.uber.org/zap"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	ktypes "k8s.io/apimachinery/pkg/types"
)

// TODO: Because of the limited number of PVs in an IDC environment,
// in the test environment, after tc is removed, the associated PV is
// defined as the Delete reclaim policy. Prevents the PV from being
// deleted after it is deleted and is not available.
// Remove these codes when going Online.
func (c *Controller) CleanTiDBClusterPV(namespace string, instanceID string) error {
	// At first delete all pvc of this namespace
	selector, err := label.New().Instance(instanceID).Selector()
	if err != nil {
		return fmt.Errorf("failed to create label selector, namespace: %s pvc: %v", instanceID, err)
	}
	listOptions := metav1.ListOptions{LabelSelector: selector.String()}
	err = c.kubeCli.CoreV1().PersistentVolumeClaims(namespace).DeleteCollection(&metav1.DeleteOptions{}, listOptions)
	if err != nil {
		log.Error("failed to batch delete pvc", zap.String("namespace", namespace), zap.Error(err))
		return fmt.Errorf("failed to batch delete pvc, namespace: %s err: %v", instanceID, err)
	}

	// patch convert related PVs' recleam policy to Delete
	selector, err = label.New().Instance(instanceID).Namespace(namespace).Selector()
	if err != nil {
		return fmt.Errorf("failed to create label selector, namespace: %s pvc: %v", instanceID, err)
	}
	listOptions = metav1.ListOptions{LabelSelector: selector.String()}
	pvList, err := c.kubeCli.CoreV1().PersistentVolumes().List(listOptions)
	if err != nil {
		log.Error("failed to list pv", zap.String("namespace", namespace), zap.Error(err))
		return fmt.Errorf("failed to list pv, err: %v", err)
	}
	for _, pv := range pvList.Items {
		_, err := c.kubeCli.CoreV1().PersistentVolumes().Patch(pv.Name, ktypes.StrategicMergePatchType, []byte(`{"spec":{"persistentVolumeReclaimPolicy":"Delete"}}`))
		if err != nil {
			log.Error("failed to patch pv", zap.String("namespace", namespace), zap.Error(err))
			return fmt.Errorf("failed to patch pv, name: %s err: %v", pv.Name, err)
		}
	}
	return nil
}

func (c *Controller) GetDiskMountpoints(ns string, tc *v1alpha1.TidbCluster) []map[string]string {
	var kvPVList *corev1.PersistentVolumeList
	var pdPVList *corev1.PersistentVolumeList
	var err error
	var hostIP string
	var pv corev1.PersistentVolume

	// get pds' and kvs' persistentvolumes from apiserver and combine them
	if kvPVList, err = c.GetPVList(ns, tc.Name, label.TiKVLabelVal); err != nil {
		log.Error("GetDiskMountpoints: failed to get tikv PV list",
			zap.String("namespace", ns), zap.Error(err))
	}
	if pdPVList, err = c.GetPVList(ns, tc.Name, label.PDLabelVal); err != nil {
		log.Error("GetDiskMountpoints: failed to get pd pv list",
			zap.String("namespace", ns), zap.Error(err))
	}
	pvList := append(kvPVList.Items, pdPVList.Items...)

	// claim prometheus disk statement list
	diskMountpointList := make([]map[string]string, 0, len(pvList))

	for _, pv = range pvList {
		if provisionedBy, ok := pv.Annotations["pv.kubernetes.io/provisioned-by"]; ok && strings.HasPrefix(provisionedBy, "local-volume-provisioner-") && len(strings.Split(provisionedBy, "-")) > 3 {
			hostIP = strings.Split(provisionedBy, "-")[3]
		} else {
			continue
		}
		tmp := make(map[string]string)
		tmp[hostIP] = pv.Spec.Local.Path
		diskMountpointList = append(diskMountpointList, tmp)
	}

	return diskMountpointList
}