package tidbcluster

import (
	"context"
	"fmt"
	"github.com/pingcap/log"
	"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/operator"
	"github.com/pingcap/tep/pkg/rbac"
	store "github.com/pingcap/tep/pkg/registry/apiserver"
	platformsetting "github.com/pingcap/tep/pkg/registry/apiserver/platform_setting"
	"github.com/pingcap/tep/pkg/utils"
	"github.com/pingcap/tep/pkg/utils/alert"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"github.com/pingcap/tidb-operator/pkg/label"
	"go.uber.org/zap"
	"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/registry/rest"
	"strings"
	"time"
)

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

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

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

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

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"))
	}
	instance := obj.(*apiserver.TiDBCluster)
	instance.Name = utils.RandomInstanceID()
	instance.Version = strings.TrimSpace(instance.Version)

	if instance.Monitor.AlertmanagerURL == "" {
		instance.Monitor.AlertmanagerURL = s.store.GetAlertmanagerURL()
	}

	log.Info("create tidb cluster payload", zap.Any("instance", instance))

	if createValidation != nil {
		if err := createValidation(ctx, obj.DeepCopyObject()); err != nil {
			return nil, err
		}
	}

	if err := s.validate(instance, nil, true); err != nil {
		log.Error("invalid creating instance request payload", zap.String("Name", instance.Name), zap.Error(err))
		return nil, errors.NewBadRequest(err.Error())
	}

	if err := s.store.Controller().CreateNamespace(instance.Name); err != nil {
		msg := fmt.Errorf("failed to create namespace[%s]. err: %v", instance.Name, err)
		return nil, errors.NewInternalError(msg)
	}

	if err := s.store.Controller().CreateTiDBSecret(instance); err != nil {
		msg := fmt.Errorf("failed to create tidb secret. err: %v", err)
		return nil, errors.NewInternalError(msg)
	}

	if err := s.store.Controller().CreateTiDBBackupRBAC(instance.Name); err != nil {
		msg := fmt.Errorf("failed to create tidb backup RBAC rules. err: %v", err)
		return nil, errors.NewInternalError(msg)
	}

	err := s.store.Operator.CreateTiDBCluster(instance)
	if err != nil {
		msg := fmt.Errorf("failed to create instance. err: %v", err)
		log.Error("failed to create instance", zap.String("Name", instance.Name), zap.Error(msg))
		go s.store.Controller().DeleteNamespace(instance.Name)
		return nil, errors.NewInternalError(msg)
	}
	utils.AddEventAndLogError(s.store.DB, &utils.Event{
		UserID:         operatorName,
		Payload:        instance,
		OperationType:  entity.CreateOperation,
		ManagementType: entity.InstanceManagement,
		CostTime:       time.Now().Sub(startTime).Nanoseconds(),
	})
	i18n.PlatformMessage.AddPlatformMessageAndLogError(s.store.DB, i18n.OperateMessageType, i18n.CreateCluster, instance.Name)
	return instance, nil
}

func (s *Storage) Get(ctx context.Context, name string, options *metav1.GetOptions) (runtime.Object, error) {
	tc, err := s.store.GetTiDBCluster(name)
	if err != nil {
		if errors.IsNotFound(err) {
			return nil, errors.NewNotFound(s.groupResource, name)
		}
		return nil, errors.NewInternalError(err)
	}
	return s.k8sTC2APIItem(tc, true)
}

func (s *Storage) List(ctx context.Context, options *metainternalversion.ListOptions, extraOptions runtime.Object) (runtime.Object, error) {
	listOpts := extraOptions.(*apiserver.TiDBClusterListOptions)

	opts := metav1.ListOptions{}
	opts.LabelSelector = fmt.Sprintf("%s=%s", operator.CreatedByLabelKey, operator.CreatedByLabelVal)

	if listOpts.Name != "" {
		if err := ValidateName(listOpts.Name); err != nil {
			msg := fmt.Sprintf("instance id is invalid: %v", err)
			return nil, errors.NewBadRequest(msg)
		}
		opts.LabelSelector += fmt.Sprintf(",%s=%s", label.InstanceLabelKey, listOpts.Name)
	}

	if listOpts.ClusterName != "" {
		if err := ValidateClusterName(listOpts.ClusterName); err != nil {
			msg := fmt.Sprintf("instance name is invalid: %v", err)
			return nil, errors.NewBadRequest(msg)
		}
		opts.LabelSelector += fmt.Sprintf(",%s=%s", operator.NameLabelKey, listOpts.ClusterName)
	}

	var tidbClusters []v1alpha1.TidbCluster
	// get instance list with tag filter
	// multiple tags separated by comma，e.g. tags=pingcap,tep,cloud
	if listOpts.Tags != "" {
		tags := strings.Split(listOpts.Tags, ",")
		tagLables := make([]string, 0, len(tags))
		for _, tag := range tags {
			tagLables = append(tagLables, operator.TagLabelKeyPrefix+strings.TrimSpace(tag))
		}

		var tidbClusterList []v1alpha1.TidbCluster
		for _, tagLabel := range tagLables {
			opts.LabelSelector = tagLabel
			tcl, err := s.store.Controller().GetTidbClusters(metav1.NamespaceAll, opts)
			if err != nil {
				msg := fmt.Sprintf("failed to get instance list: %v", err)
				return nil, errors.NewBadRequest(msg)
			}
			tidbClusterList = append(tidbClusterList, tcl.Items...)
		}

		m := make(map[string]bool)
		for _, tc := range tidbClusterList {
			ns := tc.GetNamespace()
			if _, ok := m[ns]; !ok {
				m[ns] = true
				tidbClusters = append(tidbClusters, tc)
			}
		}
	} else {
		tcl, err := s.store.Controller().GetTidbClusters(metav1.NamespaceAll, opts)
		if err != nil {
			msg := fmt.Errorf("failed to get instance list: %v", err)
			return nil, errors.NewInternalError(msg)
		}
		tidbClusters = tcl.Items
	}

	apiList := []apiserver.TiDBCluster{}
	for _, item := range tidbClusters {
		apiItem, err := s.k8sTC2APIItem(&item, false)
		if err != nil {
			log.Warn("failed to construct api item", zap.String("name", item.Namespace), zap.Error(err))
		} else {
			if listOpts.Status != "" && listOpts.Status != apiItem.Status {
				continue
			}
			apiList = append(apiList, *apiItem)
		}
	}

	res := &apiserver.TiDBClusterList{
		Total: int64(len(apiList)),
		Items: utils.Pagination(int(listOpts.Page), int(listOpts.ItemsPerPage), apiList).([]apiserver.TiDBCluster),
	}
	for index, item := range res.Items {
		res.Items[index] = *s.setTimeCostFields(&item)
	}
	return res, nil
}

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

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

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

func (s *Storage) k8sTC2APIItem(tc *v1alpha1.TidbCluster, initTimeCostFields bool) (*apiserver.TiDBCluster, error) {
	name := tc.Name
	instance := apiserver.TiDBCluster{}
	instance.Name = name
	instance.CreationTimestamp = metav1.NewTime(tc.CreationTimestamp.Time)
	instance.ClusterName, instance.Tags = s.store.Controller().GetNameAndTags(tc)
	instance.Version = tc.Spec.Version
	instance.EnableTLS = tc.Spec.EnableTLSCluster

	pdResourceTemplate, err := s.store.Controller().GetComponentResourceTemplate(tc, label.PDLabelVal)
	if err != nil {
		msg := fmt.Errorf("failed to get ResourceTemplate of component[%s], err: %s", label.PDLabelVal, err)
		return nil, errors.NewInternalError(msg)
	}

	tidbResourceTemplate, err := s.store.Controller().GetComponentResourceTemplate(tc, label.TiDBLabelVal)
	if err != nil {
		msg := fmt.Errorf("failed to get ResourceTemplate of component[%s], err: %s", label.TiDBLabelVal, err)
		return nil, errors.NewInternalError(msg)
	}

	tikvResourceTemplate, err := s.store.Controller().GetComponentResourceTemplate(tc, label.PDLabelVal)
	if err != nil {
		msg := fmt.Errorf("failed to get ResourceTemplate of component[%s], err: %s", label.PDLabelVal, err)
		return nil, errors.NewInternalError(msg)
	}

	instance.PD = apiserver.PDSpec{
		Replicas:         uint32(tc.Spec.PD.Replicas),
		ResourceTemplate: *pdResourceTemplate,
		StorageClassName: s.store.StatusMonitor.GetPDStorageClass(name),
	}

	instance.TiKV = apiserver.TiKVSpec{
		Replicas:         uint32(tc.Spec.TiKV.Replicas),
		ResourceTemplate: *tikvResourceTemplate,
		StorageClassName: s.store.StatusMonitor.GetTiKVStorageClass(name),
	}

	instance.TiDB = apiserver.TiDBSpec{
		Replicas:         uint32(tc.Spec.TiDB.Replicas),
		ResourceTemplate: *tidbResourceTemplate,
	}

	instance.Monitor = apiserver.MonitorSpec{
		StorageClassName: s.store.StatusMonitor.GetMonitorStorageClass(name),
	}

	instance.AZs = s.store.Controller().GetInstanceAzs(tc)
	instance.Status = s.store.StatusMonitor.GetClusterStatus(tc)
	instance.UID = tc.UID
	if initTimeCostFields {
		s.setTimeCostFields(&instance)
	}
	return &instance, nil
}

func (s *Storage) setTimeCostFields(instance *apiserver.TiDBCluster) *apiserver.TiDBCluster {
	tidbSvcName := fmt.Sprintf("%s-tidb", instance.Name)
	addrs, port, err := s.store.Controller().GetTiDBEndpoints(instance.Name, tidbSvcName)
	if err != nil {
		log.Error("failed to get tidb endpoints", zap.String("cluster", instance.Name), zap.Error(err))
	} else {
		instance.Endpoints = addrs
		instance.Port = port
	}
	config, err := s.store.Controller().GetPDClient(controller.Namespace(instance.Name), instance.Name, false).GetReplicate()
	if err != nil {
		log.Warn("can not get pd config", zap.String("cluster", instance.Name), zap.Error(err))
	} else {
		instance.ReplicationFactor = config.MaxReplicas
	}
	return instance
}

func (s *Storage) Delete(ctx context.Context, name string, deleteValidation rest.ValidateObjectFunc, options *metav1.DeleteOptions) (runtime.Object, bool, error) {
	startTime := time.Now()
	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"))
	}
	_, err := s.store.GetTiDBCluster(name)
	if err != nil {
		if errors.IsNotFound(err) {
			return nil, false, errors.NewNotFound(s.groupResource, name)
		}
		return nil, false, errors.NewInternalError(err)
	}
	err = s.store.Operator.DeleteTiDBCluster(name)
	if err != nil {
		msg := fmt.Errorf("failed to delete tidb cluster[%s]: %v", name, err)
		return nil, false, errors.NewInternalError(msg)
	}
	utils.AddEventAndLogError(s.store.DB, &utils.Event{
		UserID:         operatorName,
		Payload:        name,
		OperationType:  entity.DeleteOperation,
		ManagementType: entity.InstanceManagement,
		CostTime:       time.Now().Sub(startTime).Nanoseconds(),
	})

	err = s.store.Controller().DeleteTiDBSecret(name)
	if err != nil {
		return nil, false, errors.NewInternalError(err)
	}
	i18n.PlatformMessage.AddPlatformMessageAndLogError(s.store.DB, i18n.OperateMessageType, i18n.DeleteCluster, name)
	alert.DeleteClusterAlerts(name, s.store.DB.DB())
	return nil, false, s.deleteBackupsAndRestores(name)
}

func (s *Storage) Update(ctx context.Context, name string, objInfo rest.UpdatedObjectInfo, createValidation rest.ValidateObjectFunc, updateValidation rest.ValidateObjectUpdateFunc, forceAllowCreate bool, options *metav1.UpdateOptions) (runtime.Object, bool, error) {
	startTime := time.Now()
	operatorName := utils.GetAPIUsername(ctx)
	if !rbac.CanOperate(s.store.DB, operatorName, rbac.OtherClusterOperation) {
		return nil, false, errors.NewForbidden(s.groupResource, operatorName, fmt.Errorf("permission denied"))
	}
	tc, err := s.store.GetTiDBCluster(name)
	if err != nil {
		if errors.IsNotFound(err) {
			return nil, false, errors.NewNotFound(s.groupResource, name)
		}
		return nil, false, errors.NewInternalError(err)
	}
	instance, err := s.k8sTC2APIItem(tc, true)
	if err != nil {
		return nil, false, errors.NewInternalError(err)
	}

	newObj, err := objInfo.UpdatedObject(ctx, instance)
	if err != nil {
		return nil, false, errors.NewInternalError(err)
	}
	newInstance := newObj.(*apiserver.TiDBCluster)
	mergeUnUpdateField(newInstance, instance)
	if err := s.validate(newInstance, instance, false); err != nil {
		return nil, false, errors.NewBadRequest(err.Error())
	}
	newObj, updated, err := s.UpdateTiDBCluster(newInstance)
	if err == nil {
		utils.AddEventAndLogError(s.store.DB, &utils.Event{
			UserID:         operatorName,
			Payload:        newObj,
			OperationType:  entity.OperateOperation,
			ManagementType: entity.InstanceManagement,
			CostTime:       time.Now().Sub(startTime).Nanoseconds(),
		})
		if newInstance.Version != instance.Version {
			i18n.PlatformMessage.AddPlatformMessageAndLogError(s.store.DB, i18n.OperateMessageType, i18n.UpgradeClusterVersion, name)
		}
		if newInstance.ClusterName != instance.ClusterName || !IsArrayEquals(instance.Tags, newInstance.Tags) {
			i18n.PlatformMessage.AddPlatformMessageAndLogError(s.store.DB, i18n.OperateMessageType, i18n.UpdateClusterMetadataInformation, name)
		}
		if newInstance.TiDB.Replicas != instance.TiDB.Replicas || newInstance.TiKV.Replicas != instance.TiKV.Replicas || newInstance.PD.Replicas != instance.PD.Replicas {
			i18n.PlatformMessage.AddPlatformMessageAndLogError(s.store.DB, i18n.OperateMessageType, i18n.UpdateClusterConfiguration, name)
		}
	}
	if err != nil {
		return newObj, updated, errors.NewInternalError(err)
	}
	return newObj, updated, nil
}

func IsArrayEquals(old []string, new []string) bool {
	if len(old) != len(new) {
		return false
	}
	var tagMap = make(map[string]bool)
	for _, item := range old {
		tagMap[item] = true
	}
	for _, item := range new {
		if !tagMap[item] {
			return false
		}
	}
	return true
}

func (s *Storage) UpdateTiDBCluster(instance *apiserver.TiDBCluster) (runtime.Object, bool, error) {
	err := s.store.Operator.UpdateTiDBCluster(instance)
	if err != nil {
		msg := fmt.Errorf("failed to update tidb cluster: %v", err)
		log.Error("failed to update tidb cluster", zap.String("ID", instance.Name), zap.Error(err))
		return nil, false, errors.NewInternalError(msg)
	}
	return instance, true, nil
}

func (s *Storage) validate(newObj, oldObj *apiserver.TiDBCluster, isCreating bool) error {
	setting, err := s.platformStorage.GetPlatform(store.UserSettingsPath)
	if err != nil {
		return err
	}
	return validateCluster(newObj, oldObj, setting, isCreating)
}

func mergeUnUpdateField(newObj, oldObj *apiserver.TiDBCluster) {
	newObj.Name = oldObj.Name
	newObj.Password = ""
	newObj.ReplicationFactor = oldObj.ReplicationFactor
	newObj.AZs = oldObj.AZs
	newObj.EnableTLS = oldObj.EnableTLS

	newObj.TiDB.ResourceTemplate = oldObj.TiDB.ResourceTemplate

	newObj.PD = oldObj.PD

	newObj.TiKV.ResourceTemplate = oldObj.TiKV.ResourceTemplate
	newObj.TiKV.StorageClassName = oldObj.TiKV.StorageClassName

	newObj.Monitor.StorageClassName = oldObj.Monitor.StorageClassName
}

func (s *Storage) deleteBackupsAndRestores(namespace string) error {
	if err := s.store.Controller().DeleteAllBackups(namespace); err != nil {
		return errors.NewInternalError(err)
	}
	if err := s.store.Controller().DeleteScheduledBackup(operator.ScheduledBackupName, namespace); err != nil {
		return errors.NewInternalError(err)
	}
	if err := s.store.Controller().DeleteAllRestores(namespace); err != nil {
		return errors.NewInternalError(err)
	}
	return nil
}