package scheduledbackup

import (
	"context"
	"fmt"
	"time"

	"github.com/pingcap/log"
	"github.com/pingcap/tep/db/entity"
	"github.com/pingcap/tep/pkg/apis/apiserver"
	"github.com/pingcap/tep/pkg/operator"
	"github.com/pingcap/tep/pkg/rbac"
	store "github.com/pingcap/tep/pkg/registry/apiserver"
	"github.com/pingcap/tep/pkg/utils"
	"github.com/pingcap/tep/pkg/utils/validate"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"go.uber.org/zap"
	"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"
)

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) Kind() string {
	return "TiDBClusterScheduledBackup"
}

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

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"))
	}
	var err, msg error
	var backup, scheduledBackup *apiserver.TiDBClusterScheduledBackup
	var scheduledBackupCR *v1alpha1.BackupSchedule
	backup = obj.(*apiserver.TiDBClusterScheduledBackup)
	namespace := request.NamespaceValue(ctx)

	// use pointer to detect weither pause field exists,
	// if not, default to false
	if backup.Pause == nil {
		defaultPause := false
		backup.Pause = &defaultPause
	}

	// set default secret name
	if backup.TiDBSecretName == "" {
		backup.TiDBSecretName = operator.BackupSecretName
	}

	if err = validate.ValidateScheduledBackup(backup); err != nil {
		log.Error("", zap.Error(err))
		return nil, errors.NewBadRequest(err.Error())
	}

	if err = s.store.Controller().ReplaceS3Secret(namespace, backup.S3.AccessKey, backup.S3.SecretKey); err != nil {
		msg = fmt.Errorf("failed to create ReplaceS3Secret namespace[%s]. err: %v", namespace, err)
		log.Error("", zap.Error(msg))
		return nil, errors.NewInternalError(msg)
	}

	scheduledBackupCR = s.store.Controller().AssembleScheduledBackup(operator.S3SecretName, namespace, backup)
	if scheduledBackupCR.Spec.Schedule, err = s.store.ParseScheduleViaUTCOffset(scheduledBackupCR.Spec.Schedule, false); err != nil {
		msg := fmt.Errorf("failed to parse backupCR[%s] schedule[%s]. err: %v", scheduledBackupCR.Name, scheduledBackupCR.Spec.Schedule, err)
		return nil, errors.NewInternalError(msg)
	}

	if scheduledBackupCR, err = s.store.Controller().CreateScheduledBackup(scheduledBackupCR, namespace); err != nil {
		if errors.IsAlreadyExists(err) {
			return nil, errors.NewConflict(s.groupResource, backup.Name, err)
		}
		msg = fmt.Errorf("failed to create backupCR[%s] namespace[%s]. err: %v", scheduledBackupCR.Name, namespace, err)
		return nil, errors.NewInternalError(msg)
	}

	scheduledBackup = s.store.Controller().AssembleTiDBClusterScheduledBackup(scheduledBackupCR)

	err = s.store.AddEvent(&utils.Event{
		UserID:         operatorName,
		Payload:        backup,
		OperationType:  entity.CreateOperation,
		ManagementType: entity.InstanceManagement,
		CostTime:       time.Now().Sub(startTime).Nanoseconds(),
	})
	if err != nil {
		log.Error("", zap.Error(err))
		return nil, err
	}

	return scheduledBackup, nil
}

func (s *Storage) NewGetOptions() (runtime.Object, bool, string) {
	return &apiserver.TiDBClusterScheduledBackup{}, true, ""
}

func (s *Storage) Get(ctx context.Context, name string, options *metav1.GetOptions) (runtime.Object, error) {
	var err, msg error
	var namespace string
	var backupSchedule *v1alpha1.BackupSchedule

	namespace = request.NamespaceValue(ctx)
	if backupSchedule, err = s.store.Controller().GetScheduledBackup(name, namespace); err != nil {
		if errors.IsNotFound(err) {
			return nil, errors.NewNotFound(s.groupResource, fmt.Sprintf("schedulebackup %s is not found", name))
		}
		msg = fmt.Errorf("failed to get scheduledBackupCR[%s] namespace[%s]. err: %v", name, namespace, err)
		log.Error("", zap.Error(msg))
		return nil, errors.NewInternalError(msg)
	}
	if backupSchedule.Spec.Schedule, err = s.store.ParseScheduleViaUTCOffset(backupSchedule.Spec.Schedule, true); err != nil {
		msg := fmt.Errorf("failed to parse scheduledBackupCR[%s] schedule[%s]. err: %v", backupSchedule.Name, backupSchedule.Spec.Schedule, err)
		return nil, errors.NewInternalError(msg)
	}

	return s.store.Controller().AssembleTiDBClusterScheduledBackup(backupSchedule), nil
}

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"))
	}
	var err, msg error
	var oldScheduledBackup, newScheduledBackup *apiserver.TiDBClusterScheduledBackup
	var oldBackupSchedule, scheduledBackupCR *v1alpha1.BackupSchedule
	var newObj runtime.Object
	var namespace, accessKey, secretKey string

	namespace = request.NamespaceValue(ctx)
	if oldBackupSchedule, err = s.store.Controller().GetScheduledBackup(name, namespace); err != nil {
		msg = fmt.Errorf("failed to get v1alpha1 BackupSchedule namespace[%s]. err: %v", namespace, err)
		log.Error("", zap.Error(msg))
		return nil, false, msg
	}

	if newObj, err = objInfo.UpdatedObject(ctx,
		&apiserver.TiDBClusterScheduledBackup{
			ObjectMeta: metav1.ObjectMeta{
				Name:      oldBackupSchedule.Name,
				Namespace: oldBackupSchedule.Namespace,
				UID:       oldBackupSchedule.UID,
			},
		}); err != nil {
		msg = fmt.Errorf("failed to get TiDBScheduledBackup namespace[%s]. err: %v", namespace, err)
		log.Error("", zap.Error(msg))
		return nil, false, errors.NewInternalError(msg)
	}

	newScheduledBackup = newObj.(*apiserver.TiDBClusterScheduledBackup)

	oldScheduledBackup = s.store.Controller().AssembleTiDBClusterScheduledBackup(oldBackupSchedule)

	// if secretkey or accesskey is update, validate it and then update the s3-object-storage secret
	if accessKey, secretKey, err = s.store.Controller().GetCredentialFromSecret(namespace, oldBackupSchedule.Spec.BackupTemplate.S3.SecretName); err != nil {
		msg = fmt.Errorf("failed to get secret[%s] in namespace[%s]. err: %v", oldBackupSchedule.Spec.BackupTemplate.S3.SecretName, namespace, err)
		log.Error("", zap.Error(msg))
		return nil, false, errors.NewInternalError(msg)
	}
	if newScheduledBackup.S3 != nil && newScheduledBackup.S3.AccessKey != "" {
		accessKey = newScheduledBackup.S3.AccessKey
	}
	if newScheduledBackup.S3 != nil && newScheduledBackup.S3.SecretKey != "" {
		secretKey = newScheduledBackup.S3.SecretKey
	}

	if newObj, err = objInfo.UpdatedObject(ctx, oldScheduledBackup); err != nil {
		msg = fmt.Errorf("failed to get TiDBScheduledBackup namespace[%s]. err: %v", namespace, err)
		log.Error("", zap.Error(msg))
		return nil, false, errors.NewInternalError(msg)
	}

	newScheduledBackup = newObj.(*apiserver.TiDBClusterScheduledBackup)
	newScheduledBackup.S3.AccessKey = accessKey
	newScheduledBackup.S3.SecretKey = secretKey
	// use pointer to detect weither pause field exists,
	// if not, merge old one.
	if newScheduledBackup.Pause == nil {
		newScheduledBackup.Pause = oldScheduledBackup.Pause
	}

	// set default secret name
	if newScheduledBackup.TiDBSecretName == "" {
		newScheduledBackup.TiDBSecretName = operator.BackupSecretName
	}

	if err = validate.ValidateScheduledBackup(newScheduledBackup); err != nil {
		log.Error("", zap.Error(err))
		return nil, false, errors.NewBadRequest(err.Error())
	}

	if newScheduledBackup.S3.AccessKey != "" || newScheduledBackup.S3.SecretKey != "" {
		if err = s.store.Controller().ReplaceS3Secret(namespace, newScheduledBackup.S3.AccessKey, newScheduledBackup.S3.SecretKey); err != nil {
			msg = fmt.Errorf("failed to create ReplaceS3Secret namespace[%s]. err: %v", namespace, err)
			log.Error("", zap.Error(msg))
			return nil, false, errors.NewInternalError(msg)
		}
	}

	scheduledBackupCR = s.store.Controller().AssembleScheduledBackup(operator.S3SecretName, namespace, newScheduledBackup)

	scheduledBackupCR.ObjectMeta.ResourceVersion = oldBackupSchedule.ObjectMeta.ResourceVersion

	if scheduledBackupCR.Spec.Schedule, err = s.store.ParseScheduleViaUTCOffset(scheduledBackupCR.Spec.Schedule, false); err != nil {
		msg := fmt.Errorf("failed to parse backupCR[%s] schedule[%s]. err: %v", scheduledBackupCR.Name, scheduledBackupCR.Spec.Schedule, err)
		return nil, false, errors.NewInternalError(msg)
	}

	if scheduledBackupCR, err = s.store.Controller().UpdateScheduledBackup(namespace, scheduledBackupCR); err != nil {
		msg = fmt.Errorf("failed to update BackupSchedule [%s]. err: %v", namespace, err)
		log.Error("", zap.Error(msg))
		return nil, false, errors.NewInternalError(msg)
	}

	newScheduledBackup = s.store.Controller().AssembleTiDBClusterScheduledBackup(scheduledBackupCR)

	if err := s.store.AddEvent(&utils.Event{
		UserID:         operatorName,
		Payload:        newObj,
		OperationType:  entity.UpdateOperation,
		ManagementType: entity.InstanceManagement,
		CostTime:       time.Now().Sub(startTime).Nanoseconds(),
	}); err != nil {
		msg = fmt.Errorf("failed to update operator event [%s]. err: %v", namespace, err)
		log.Error("", zap.Error(msg))
		return nil, false, errors.NewInternalError(msg)
	}

	return newScheduledBackup, true, nil
}

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.OtherClusterOperation) {
		return nil, false, errors.NewForbidden(s.groupResource, operatorName, fmt.Errorf("permission denied"))
	}
	var err error
	namespace := request.NamespaceValue(ctx)
	if err = s.store.Controller().DeleteScheduledBackup(name, namespace); err != nil {
		msg := fmt.Errorf("failed to delete schedulebackup[%s]. err: %v", namespace, err)
		if errors.IsNotFound(err) {
			return nil, false, errors.NewNotFound(s.groupResource, fmt.Sprintf("backup %s is not found", name))
		}
		log.Error("", zap.Error(msg))
		return nil, false, errors.NewInternalError(msg)
	}
	if err = s.store.AddEvent(&utils.Event{
		UserID:         operatorName,
		Payload:        fmt.Sprintf("delete cluster schedule backup: %s", name),
		OperationType:  entity.DeleteOperation,
		ManagementType: entity.InstanceManagement,
		CostTime:       time.Now().Sub(startTime).Nanoseconds(),
	}); err != nil {
		log.Error("", zap.Error(err))
		return nil, false, err
	}

	return nil, true, nil
}

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

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

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