// 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/util/label"
	"github.com/pingcap/tidb-operator/pkg/volumeprovisioner"
	"k8s.io/api/core/v1"
	apierrs "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)

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

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

	if !exists {
		return nil
	}

	pv, ok := obj.(*v1.PersistentVolume)
	if !ok {
		return errors.Trace(fmt.Errorf("object %v is not a *v1.PersistentVolume", obj))
	}

	err = syncc.syncPVLabels(pv)
	if err != nil {
		return errors.Trace(err)
	}

	err = syncc.syncPVClusterDeleted(pv)
	if err != nil {
		return errors.Trace(err)
	}

	if syncc.testMode {
		return syncc.syncManualDeletePV(pv)
	}
	return nil
}

// Ensure PV have:
//   Labels: label.New().Namespace(ns).Cluster(clusterName).PD()/TiDB()/TiKV()
//   Annotations: label.AnnPodNameKey
func (syncc *SyncController) syncPVLabels(pv *v1.PersistentVolume) error {
	pvLabels := pv.GetLabels()

	pvName := pv.GetName()
	pvcRef := pv.Spec.ClaimRef
	if pvcRef == nil {
		log.Warnf("PV: [%s] doesn't have a ClaimRef, skipping", pvName)
		return nil
	}

	pvcNs := pvcRef.Namespace
	pvcName := pvcRef.Name
	pvc, err := syncc.kubeCli.CoreV1().PersistentVolumeClaims(pvcNs).Get(pvcName, metav1.GetOptions{})
	if err != nil {
		if !apierrs.IsNotFound(err) {
			return errors.Trace(err)
		}

		log.Warnf("PV: [%s]'s PVC: [%s/%s] doesn't exist, skipping", pvName, pvcNs, pvcName)
		return nil
	}

	pvcLabels := pvc.GetLabels()
	app := pvc.Labels[label.AppLabelKey]
	podName := pvc.Annotations[label.AnnPodNameKey]
	clusterID := pvcLabels[label.ClusterIDLabelKey]
	memberID := pvcLabels[label.MemberIDLabelKey]
	storeID := pvcLabels[label.StoreIDLabelKey]

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

	if pvLabels[label.NamespaceLabelKey] == pvcNs &&
		pvLabels[label.AppLabelKey] == app &&
		pvLabels[label.ClusterIDLabelKey] == clusterID &&
		pvLabels[label.MemberIDLabelKey] == memberID &&
		pvLabels[label.StoreIDLabelKey] == storeID &&
		pv.Annotations[label.AnnPodNameKey] == podName {
		log.Debugf("pv %s already has labels and annotations synced, skipping", pvName)
		return nil
	}

	if pvcNs != "" {
		pvLabels[label.NamespaceLabelKey] = pvcNs
	}
	if app != "" {
		pvLabels[label.AppLabelKey] = app
	}
	if clusterID != "" {
		pvLabels[label.ClusterIDLabelKey] = clusterID
	}
	if memberID != "" {
		pvLabels[label.MemberIDLabelKey] = memberID
	}
	if storeID != "" {
		pvLabels[label.StoreIDLabelKey] = storeID
	}
	if podName != "" {
		pv.Annotations[label.AnnPodNameKey] = podName
	}

	_, err = syncc.kubeCli.CoreV1().PersistentVolumes().Update(pv)
	if err != nil {
		log.Errorf("failed to update PV: [%s]: %v", pvName, err)
		return errors.Trace(err)
	}

	log.Infof("PV: [%s] updated successfully", pvName)
	return nil
}

func (syncc *SyncController) syncPVClusterDeleted(pv *v1.PersistentVolume) error {
	ns := pv.Labels[label.NamespaceLabelKey]
	clusterName := pv.Labels[label.ClusterLabelKey]
	pvs := syncc.kubeCli.CoreV1().PersistentVolumes()
	clusters := syncc.cli.PingcapV1().TidbClusters(ns)
	if _, getErr := clusters.Get(clusterName, metav1.GetOptions{}); getErr != nil {
		if apierrs.IsNotFound(getErr) {
			if pv.Annotations[volumeprovisioner.AnnPVClusterDeleted] == "" {
				t := time.Now().Format(time.RFC3339)
				pv.Annotations[volumeprovisioner.AnnPVClusterDeleted] = t
				if _, upErr := pvs.Update(pv); upErr != nil {
					log.Errorf("failed to set TidbCluster: [%s/%s]'s PV: [%s] Annotations[%s] to [%s]",
						ns, clusterName, pv.GetName(), volumeprovisioner.AnnPVClusterDeleted, t)
					return errors.Trace(upErr)
				}
				log.Infof("set TidbCluster: [%s/%s]'s PV[%s] Annotations[%s] to [%s] successfully",
					ns, clusterName, pv.GetName(), volumeprovisioner.AnnPVClusterDeleted, t)
				return nil
			}

			return nil
		}

		return errors.Trace(getErr)
	}

	if pv.Annotations[volumeprovisioner.AnnPVClusterDeleted] != "" {
		pv.Annotations[volumeprovisioner.AnnPVClusterDeleted] = ""
		_, err := pvs.Update(pv)
		if err != nil {
			log.Errorf("failed to clear TidbCluster: [%s/%s]'s PV: [%s] Annotations[%s]",
				ns, clusterName, pv.GetName(), volumeprovisioner.AnnPVClusterDeleted)
			return errors.Trace(err)
		}

		log.Infof("clear TidbCluster: [%s/%s]'s PV: [%s] Annotations[%s] successfully",
			ns, clusterName, pv.GetName(), volumeprovisioner.AnnPVClusterDeleted)
		return nil
	}

	return nil
}

func (syncc *SyncController) syncManualDeletePV(pv *v1.PersistentVolume) error {
	pvName := pv.GetName()
	pvs := syncc.kubeCli.CoreV1().PersistentVolumes()
	var err error
	if pv.Annotations[volumeprovisioner.AnnPVTombstoned] != "" || pv.Annotations[volumeprovisioner.AnnPVClusterDeleted] != "" {
		now := time.Now().Format(time.RFC3339)
		pv, err = pvs.Get(pvName, metav1.GetOptions{})
		if err != nil {
			return errors.Trace(err)
		}
		pv.Annotations[volumeprovisioner.AnnPVManualDeleted] = now
		_, err := pvs.Update(pv)
		if err != nil {
			log.Errorf("failed to update PV: [%s]'s Annotations[%s] to [%s], %v",
				pvName, volumeprovisioner.AnnPVManualDeleted, now, err)
			return errors.Trace(err)
		}

		log.Infof("Update PV: [%s]'s Annotations[%s] to [%s] successfully", pvName, volumeprovisioner.AnnPVManualDeleted, now)
		return nil
	}

	return nil
}