// Copyright 2017 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// See the License for the specific language governing permissions and
// limitations under the License.

package controller

import (
	"fmt"
	"time"

	"github.com/juju/errors"
	"github.com/ngaut/log"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap.com/v1"
	"github.com/pingcap/tidb-operator/pkg/util/label"
	corev1 "k8s.io/api/core/v1"
	apierrs "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

func (syncc *SyncController) syncPVC(key string) error {
	startTime := time.Now()
	defer func() {
		log.Infof("Finished syncing PVC[%s] (%v)", key, time.Since(startTime))
	}()

	obj, exists, getErr := syncc.pvcStore.GetByKey(key)
	if getErr != nil {
		return errors.Trace(getErr)
	}

	if !exists {
		return nil
	}

	pvc, ok := obj.(*corev1.PersistentVolumeClaim)
	if !ok {
		return errors.Errorf("object %v is not a *corev1.PersistentVolumeClaim", obj)
	}

	ns := pvc.GetNamespace()
	name := pvc.GetName()
	clusterName := pvc.Labels[label.ClusterLabelKey]
	if len(clusterName) == 0 {
		return errors.Errorf("PVC: [%s/%s]'s labels: %s is empty", ns, name, label.ClusterLabelKey)
	}

	{
		pod, err := syncc.getPodFrom(pvc, clusterName)
		if err != nil {
			return errors.Trace(err)
		}
		if pod != nil {
			err := syncc.syncPVCLabels(pvc, pod, clusterName)
			if err != nil {
				return errors.Trace(err)
			}
			return nil
		}
	}

	app := pvc.Labels[label.AppLabelKey]
	if len(app) == 0 {
		return errors.Errorf("PVC: [%s/%s]'s labels: %s is empty", ns, name, label.AppLabelKey)
	}

	// TODO add e2e spec to ensure pvc is deleted when pod disappeared
	// TODO two pvc related to one pd pod
	log.Infof("PVC: [%s/%s]'s pod is disappeared", ns, name)
	timeIsZero := pvc.CreationTimestamp.IsZero()
	ltGCDuration := time.Since(pvc.CreationTimestamp.Time).Minutes() < gcDuration
	if timeIsZero || ltGCDuration {
		log.Debugf("PVC: [%s/%s] was created less than %d minutes, no need to do gc", ns, name, gcDuration)
		return nil
	}

	switch app {
	case label.TiKVLabelVal:
		tikvset, err := syncc.cli.PingcapV1().TidbSets(ns).Get(fmt.Sprintf("%s-tikvset", clusterName), metav1.GetOptions{})
		// paused
		if err == nil && tikvset.Paused() {
			log.Infof("TidbSet: [%s/%s] paused, skip deleting PVC: [%s/%s]", ns, tikvset.GetName(), ns, name)
			return nil
		}
		// error is not notFound
		if err != nil && !apierrs.IsNotFound(err) {
			return errors.Trace(err)
		}
		// error is notFound or TidbSet is not paused
	case label.PDLabelVal:
		pdset, err := syncc.cli.PingcapV1().TidbSets(ns).Get(fmt.Sprintf("%s-pdset", clusterName), metav1.GetOptions{})
		// paused
		if err == nil && pdset.Paused() {
			log.Infof("TidbSet: [%s/%s] paused, skip deleting PVC: [%s/%s]", ns, pdset.GetName(), ns, name)
			return nil
		}
		// error is not notFound
		if err != nil && !apierrs.IsNotFound(err) {
			return errors.Trace(err)
		}
		// error is notFound or TidbSet is not paused
	}

	log.Infof("Deleting PVC: [%s/%s], timeIsZero: [%v], ltGCDuration: [%v]", ns, name, timeIsZero, ltGCDuration)
	err := syncc.kubeCli.CoreV1().PersistentVolumeClaims(ns).Delete(name, nil)
	if err != nil {
		log.Errorf("failed to delete PVC: [%s/%s], %v", ns, name, err)
		return errors.Trace(err)
	}

	log.Infof("PVC: [%s/%s] deleted successfully.", ns, name)
	return nil
}

// Ensure PVC have:
//   Labels: label.New().Namespace(ns).Cluster(clusterName).PD()/TiDB()/TiKV()
//   Annotations: label.AnnPodNameKey
func (syncc *SyncController) syncPVCLabels(pvc *corev1.PersistentVolumeClaim, pod *corev1.Pod, clusterName string) error {
	ns := pvc.GetNamespace()
	pvcName := pvc.GetName()
	podName := pod.GetName()
	podLabels := pod.GetLabels()
	pvcLabels := pvc.GetLabels()

	app := podLabels[label.AppLabelKey]
	clusterID := podLabels[label.ClusterIDLabelKey]
	storeID := podLabels[label.StoreIDLabelKey]
	memberID := podLabels[label.MemberIDLabelKey]

	if pvc.Annotations == nil {
		pvc.Annotations = make(map[string]string)
	}

	if pvcLabels[label.AppLabelKey] == app &&
		pvcLabels[label.ClusterIDLabelKey] == clusterID &&
		pvcLabels[label.MemberIDLabelKey] == memberID &&
		pvcLabels[label.StoreIDLabelKey] == storeID &&
		pvc.Annotations[label.AnnPodNameKey] == podName {
		log.Debugf("pvc %s/%s already has labels and annotations synced, skipping", ns, pvcName)
		return nil
	}

	if app != "" {
		pvcLabels[label.AppLabelKey] = app
	}
	if clusterID != "" {
		pvcLabels[label.ClusterIDLabelKey] = clusterID
	}
	if memberID != "" {
		pvcLabels[label.MemberIDLabelKey] = memberID
	}
	if storeID != "" {
		pvcLabels[label.StoreIDLabelKey] = storeID
	}
	if podName != "" {
		pvc.Annotations[label.AnnPodNameKey] = podName
	}

	_, err := syncc.kubeCli.CoreV1().PersistentVolumeClaims(ns).Update(pvc)
	if err != nil {
		log.Errorf("failed to update PVC: [%s/%s]: %v", ns, pvcName, err)
		return errors.Trace(err)
	}

	log.Infof("update PVC: [%s/%s] successfully", ns, pvcName)
	return nil
}

func (syncc *SyncController) getPodFrom(pvc *corev1.PersistentVolumeClaim, clusterName string) (*corev1.Pod, error) {
	ns := pvc.GetNamespace()
	pvcName := pvc.GetName()
	podList, err := syncc.kubeCli.CoreV1().Pods(ns).List(label.ClusterListOptions(clusterName))
	if err != nil {
		return nil, errors.Trace(err)
	}

	for _, p := range podList.Items {
		for _, vol := range p.Spec.Volumes {
			if vol.Name == v1.TiDBVolumeName && vol.VolumeSource.PersistentVolumeClaim != nil && pvcName == vol.VolumeSource.PersistentVolumeClaim.ClaimName {
				return &p, nil
			}
		}
	}

	return nil, nil
}