package transferpdleader

import (
	"context"
	"fmt"
	"github.com/pingcap/tep/db/entity"
	"github.com/pingcap/tep/pkg/apis/apiserver"
	"github.com/pingcap/tep/pkg/controller"
	"github.com/pingcap/tep/pkg/i18n"
	"github.com/pingcap/tep/pkg/rbac"
	store "github.com/pingcap/tep/pkg/registry/apiserver"
	"github.com/pingcap/tep/pkg/utils"
	"k8s.io/apimachinery/pkg/api/errors"
	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"
	"k8s.io/apiserver/pkg/registry/rest"
	"time"
)

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

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

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

func (s *Storage) Create(ctx context.Context, obj runtime.Object, createValidation rest.ValidateObjectFunc, options *metav1.CreateOptions) (runtime.Object, error) {
	startTime := time.Now()
	operatorName := utils.GetAPIUsername(ctx)
	if !rbac.CanOperate(s.store.DB, operatorName, rbac.OtherClusterOperation) {
		return nil, errors.NewForbidden(s.groupResource, operatorName, fmt.Errorf("permission denied"))
	}
	transferPDLeader := obj.(*apiserver.TransferPDLeader)
	_, err := s.store.Controller().GetPod(transferPDLeader.NextLeader, request.NamespaceValue(ctx))
	if err != nil {
		if errors.IsNotFound(err) {
			return nil, errors.NewBadRequest(fmt.Sprintf("pd node named %s is not found", transferPDLeader.NextLeader))
		}
		return nil, errors.NewInternalError(err)
	}

	client := s.store.Controller().GetPDClient(controller.Namespace(request.NamespaceValue(ctx)), request.NamespaceValue(ctx), false)
	if err := client.TransferPDLeader(transferPDLeader.NextLeader); err != nil {
		return nil, err
	}
	utils.AddEventAndLogError(s.store.DB, &utils.Event{
		UserID:         operatorName,
		Payload:        fmt.Sprintf("cluster new pd leader is %s", transferPDLeader.NextLeader),
		OperationType:  entity.OperateOperation,
		ManagementType: entity.InstanceManagement,
		CostTime:       time.Now().Sub(startTime).Nanoseconds(),
	})
	i18n.PlatformMessage.AddPlatformMessageAndLogError(s.store.DB, i18n.OperateMessageType, i18n.TransferPDLeader, request.NamespaceValue(ctx), transferPDLeader.NextLeader)
	return transferPDLeader, nil
}