package tikv

import (
	"context"
	"encoding/json"
	"fmt"
	"github.com/pingcap/log"
	"github.com/pingcap/tep/pkg/apis/apiserver"
	"github.com/pingcap/tep/pkg/rbac"
	store "github.com/pingcap/tep/pkg/registry/apiserver"
	"github.com/pingcap/tep/pkg/utils"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"github.com/pingcap/tidb-operator/pkg/label"
	"go.uber.org/zap"
	"k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/errors"
	metainternalversion "k8s.io/apimachinery/pkg/apis/meta/internalversion"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/util/sets"
	"k8s.io/apiserver/pkg/endpoints/request"
	"k8s.io/apiserver/pkg/registry/rest"
	"strconv"
	"strings"
)

type Storage struct {
	store         *store.Store
	groupResource schema.GroupResource
}

func NewStorage(store *store.Store, groupResource schema.GroupResource) *Storage {
	return &Storage{
		store:         store,
		groupResource: groupResource,
	}
}

func (s *Storage) New() runtime.Object {
	return &apiserver.TiKV{}
}

func (s *Storage) Kind() string {
	return "TiKV"
}

func (s *Storage) NamespaceScoped() bool {
	return true
}

func (s *Storage) Get(ctx context.Context, name string, options *metav1.GetOptions) (runtime.Object, error) {
	tidbClusterID := request.NamespaceValue(ctx)
	tc, err := s.store.GetTiDBCluster(tidbClusterID)
	if err != nil {
		return nil, err
	}

	pod, err := s.store.Controller().GetPod(name, tc.Namespace)
	if err != nil {
		if errors.IsNotFound(err) {
			return nil, errors.NewNotFound(s.groupResource, name)
		}
		msg := fmt.Errorf("failed to get pod[%s]: %v", name, err)
		return nil, errors.NewInternalError(msg)
	}
	storeMap := MakePodName2TiKVStoreMap(tc.Status.TiKV.Stores, tc.Status.TiKV.TombstoneStores)
	return s.buildAPIItem(tc, pod, storeMap), nil
}

func (s *Storage) List(ctx context.Context, options *metainternalversion.ListOptions, extraOptions runtime.Object) (runtime.Object, error) {
	tidbClusterID := request.NamespaceValue(ctx)
	tc, err := s.store.GetTiDBCluster(tidbClusterID)
	if err != nil {
		if errors.IsBadRequest(err) {
			return s.NewList(), nil
		}
		return nil, err
	}

	pods, err := s.store.Controller().GetPodList(tc.Namespace, label.TiKVLabelVal)
	storeMap := MakePodName2TiKVStoreMap(tc.Status.TiKV.Stores, tc.Status.TiKV.TombstoneStores)
	if err == nil {
		nodes := []apiserver.TiKV{}
		for _, pod := range pods.Items {
			nodes = append(nodes, *s.buildAPIItem(tc, &pod, storeMap))
		}

		listOpts := extraOptions.(*apiserver.ListOptions)

		return &apiserver.TiKVList{
			Total: int64(len(nodes)),
			Items: utils.Pagination(listOpts.Page, listOpts.ItemsPerPage, nodes).([]apiserver.TiKV),
		}, nil
	}
	return nil, err
}

func (s *Storage) buildAPIItem(tc *v1alpha1.TidbCluster, pod *v1.Pod, storeMap map[string]v1alpha1.TiKVStore) *apiserver.TiKV {
	node := apiserver.TiKV{}
	componentInfo, err := s.store.Controller().GetComponentInfo(tc, label.TiKVLabelVal, *pod, s.store.DB)
	if err != nil {
		log.Warn("failed to tikv component info", zap.String("tc", tc.Namespace), zap.String("tikv", pod.Name), zap.Error(err))
	}
	node.ComponentInfo = componentInfo
	node.State = storeMap[pod.Name].State
	node.Name = pod.Name
	node.CreationTimestamp = pod.GetCreationTimestamp()
	return &node
}

func (s *Storage) NewList() runtime.Object {
	return &apiserver.TiKVList{}
}

func (s *Storage) NewExtraListOptions() runtime.Object {
	return &apiserver.ListOptions{}
}

func (s *Storage) Delete(ctx context.Context, name string, deleteValidation rest.ValidateObjectFunc, options *metav1.DeleteOptions) (runtime.Object, bool, error) {
	tidbClusterID := request.NamespaceValue(ctx)
	operatorName := utils.GetAPIUsername(ctx)
	if !rbac.CanOperate(s.store.DB, operatorName, rbac.DeleteAndRestartCluster) {
		return nil, false, errors.NewForbidden(s.groupResource, operatorName, fmt.Errorf("permission denied"))
	}
	tc, err := s.store.GetTiDBCluster(tidbClusterID)
	if err != nil {
		if errors.IsNotFound(err) {
			return nil, false, errors.NewNotFound(s.groupResource, name)
		}
		return nil, false, errors.NewInternalError(err)
	}
	pod, err := s.store.Controller().GetPod(name, tidbClusterID)
	if err != nil {
		if errors.IsNotFound(err) {
			return nil, false, errors.NewNotFound(s.groupResource, name)
		}
		msg := fmt.Errorf("failed to get pod[%s]: %v", name, err)
		return nil, false, errors.NewInternalError(msg)
	}
	if pod.Labels[label.ComponentLabelKey] != label.TiKVLabelVal {
		return nil, false, errors.NewBadRequest(fmt.Sprintf("%s is not a tikv pod", name))
	}

	parts := strings.Split(pod.Name, "-")
	if len(parts) != 3 {
		return nil, false, errors.NewBadRequest(fmt.Sprintf("%s is not a valid tikv pod name", name))
	}
	id, err := strconv.Atoi(parts[2])
	if err != nil {
		return nil, false, errors.NewBadRequest(fmt.Sprintf("%s is not a valid tikv pod name, %v", name, err))
	}

	deleteSlots := getDeleteSlots(tc)
	if deleteSlots.Has(int32(id)) {
		//already deleted
		return pod, false, nil
	}
	deleteSlots.Insert(int32(id))
	if err := setDeleteSlots(tc, deleteSlots); err != nil {
		msg := fmt.Errorf("failed to delete pod[%s]: %v", name, err)
		return nil, false, errors.NewInternalError(msg)
	}

	if _, err := s.store.TidbClusterClient.PingcapV1alpha1().TidbClusters(tidbClusterID).Update(tc); err != nil {
		msg := fmt.Errorf("failed to delete pod[%s]: %v", name, err)
		return nil, false, errors.NewInternalError(msg)
	}
	//the pod will be deleted asynchronously.
	return pod, false, nil
}

func MakePodName2TiKVStoreMap(store map[string]v1alpha1.TiKVStore, tombstoneStores map[string]v1alpha1.TiKVStore) map[string]v1alpha1.TiKVStore {
	m := make(map[string]v1alpha1.TiKVStore)
	//FIXME: has duplicate pod name in store and tombstoneStores?
	for _, item := range store {
		m[item.PodName] = item
	}
	for _, item := range tombstoneStores {
		m[item.PodName] = item
	}
	return m
}

func getDeleteSlots(tc *v1alpha1.TidbCluster) (deleteSlots sets.Int32) {
	deleteSlots = sets.NewInt32()
	annotations := tc.GetAnnotations()
	if annotations == nil {
		return
	}
	value, ok := annotations[label.AnnTiKVDeleteSlots]
	if !ok {
		return
	}
	var slice []int32
	err := json.Unmarshal([]byte(value), &slice)
	if err != nil {
		return
	}
	deleteSlots.Insert(slice...)
	return
}

func setDeleteSlots(tc *v1alpha1.TidbCluster, deleteSlots sets.Int32) (err error) {
	annotations := tc.GetAnnotations()
	if deleteSlots == nil || deleteSlots.Len() == 0 {
		// clear
		delete(annotations, label.AnnTiKVDeleteSlots)
	} else {
		// set
		b, err := json.Marshal(deleteSlots.List())
		if err != nil {
			return err
		}
		if annotations == nil {
			annotations = make(map[string]string)
		}
		annotations[label.AnnTiKVDeleteSlots] = string(b)
	}
	tc.SetAnnotations(annotations)
	return
}