package pd

import (
	"context"
	"fmt"
	"github.com/pingcap/log"
	"github.com/pingcap/tep/pkg/apis/apiserver"
	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/apiserver/pkg/endpoints/request"
)

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.PD{}
}

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

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

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

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

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
	}

	listOpts := extraOptions.(*apiserver.PDListOptions)
	pods, err := s.store.Controller().GetPodList(tc.Namespace, label.PDLabelVal)
	if err == nil {
		nodes := []apiserver.PD{}
		for _, pod := range pods.Items {
			pd := s.toAPIItem(&pod, tc)
			if listOpts.Role == "" || listOpts.Role == pd.Role {
				nodes = append(nodes, *pd)
			}
		}

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

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)
	}

	return s.toAPIItem(pod, tc), nil
}

func (s *Storage) toAPIItem(pod *v1.Pod, tc *v1alpha1.TidbCluster) *apiserver.PD {
	pd := apiserver.PD{}
	pd.Name = pod.Name
	pd.CreationTimestamp = pod.GetCreationTimestamp()
	db := s.store.DB
	componentInfo, err := s.store.Controller().GetComponentInfo(tc, label.PDLabelVal, *pod, db)
	if err != nil {
		log.Warn("failed to get pod component info", zap.String("tc", tc.Namespace), zap.String("pd", pod.Name), zap.Error(err))
	}
	pd.ComponentInfo = componentInfo
	pdStatus := tc.Status.PD
	pdLeaderName := pdStatus.Leader.Name
	if pd.Name == pdLeaderName {
		pd.Role = "leader"
	} else {
		pd.Role = "follower"
	}
	member, exist := pdStatus.Members[pd.Name]
	if exist {
		pd.Health = member.Health
	}
	return &pd
}