package metadatascheduledbackup

import (
	"context"
	"fmt"
	"github.com/pingcap/tep/pkg/rbac"
	"time"

	"github.com/pingcap/log"
	"github.com/pingcap/tep/db/entity"
	"github.com/pingcap/tep/pkg/apis/apiserver"
	"github.com/pingcap/tep/pkg/operator"
	store "github.com/pingcap/tep/pkg/registry/apiserver"
	"github.com/pingcap/tep/pkg/utils"
	"github.com/pingcap/tep/pkg/utils/validate"
	"go.uber.org/zap"
	batchv1beta1 "k8s.io/api/batch/v1beta1"
	"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 "MetadataScheduledBackup"
}

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

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.PlatformOperation) {
		return nil, errors.NewForbidden(s.groupResource, operatorName, fmt.Errorf("permission denied"))
	}
	var err, msg error
	var cronJob *batchv1beta1.CronJob
	backup := obj.(*apiserver.MetadataScheduledBackup)
	//if no bucket field, use the default one
	if backup.S3.Bucket == "" {
		backup.S3.Bucket = operator.MetadataBucketName
	}

	if err = validate.ValidateScheduledMetadataBackup(backup); err != nil {
		log.Error("failed to validate metadata schedule backup", zap.Error(err))
		return nil, errors.NewBadRequest(err.Error())
	}

	namespace := s.store.Namespace
	cronJob = s.store.Controller().AssembleMetadataBackupCronJob(s.store.MetadataBackupImage, namespace, s.store.DSN, backup)
	if cronJob.Spec.Schedule, err = s.store.ParseScheduleViaUTCOffset(cronJob.Spec.Schedule, false); err != nil {
		msg := fmt.Errorf("failed to parse cronJob[%s] schedule[%s]. err: %v", cronJob.Name, cronJob.Spec.Schedule, err)
		return nil, errors.NewInternalError(msg)
	}

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

	backup = s.store.Controller().AssembleMetadataScheduledBackup(cronJob)

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

	return backup, nil
}

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

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

	namespace = s.store.Namespace

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

	return s.store.Controller().AssembleMetadataScheduledBackup(cronJob), 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()
	var err, msg error
	var oldCronJob, newCronJob *batchv1beta1.CronJob
	var oldBackup, newBackup *apiserver.MetadataScheduledBackup
	var newObj runtime.Object
	var userID, namespace, accessKey, secretKey string

	if user, found := request.UserFrom(ctx); found {
		userID = user.GetUID()
	}

	namespace = s.store.Namespace

	if oldCronJob, err = s.store.Controller().GetCronJob(name, namespace); err != nil {
		msg = fmt.Errorf("failed to get old cronJob name [%s] namespace[%s]. err: %v", name, namespace, err)
		log.Error("", zap.Error(msg))
		return nil, false, msg
	}

	accessKey, secretKey = s.store.Controller().GetSecretKeyViaEnv(oldCronJob.Spec.JobTemplate.Spec.Template.Spec.Containers[0].Env)

	oldBackup = s.store.Controller().AssembleMetadataScheduledBackup(oldCronJob)

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

	newBackup = newObj.(*apiserver.MetadataScheduledBackup)
	if newBackup.S3.AccessKey == "" {
		newBackup.S3.AccessKey = accessKey
	}
	if newBackup.S3.SecretKey == "" {
		newBackup.S3.SecretKey = secretKey
	}

	//if no bucket field, use the default one
	if newBackup.S3.Bucket == "" {
		newBackup.S3.Bucket = operator.MetadataBucketName
	}

	if err = validate.ValidateScheduledMetadataBackup(newBackup); err != nil {
		log.Error("failed to validate metadata schedule backup", zap.Error(err))
		return nil, false, errors.NewBadRequest(err.Error())
	}

	newCronJob = s.store.Controller().AssembleMetadataBackupCronJob(s.store.MetadataBackupImage, namespace, s.store.DSN, newBackup)

	newCronJob.ObjectMeta.ResourceVersion = oldCronJob.ObjectMeta.ResourceVersion
	if newCronJob.Spec.Schedule, err = s.store.ParseScheduleViaUTCOffset(newCronJob.Spec.Schedule, false); err != nil {
		msg := fmt.Errorf("failed to parse cronJob[%s] schedule[%s]. err: %v", newCronJob.Name, newCronJob.Spec.Schedule, err)
		return nil, false, errors.NewInternalError(msg)
	}
	if newCronJob, err = s.store.Controller().UpdateCronJob(namespace, newCronJob); err != nil {
		msg = fmt.Errorf("failed to update CronJob [%s]. err: %v", namespace, err)
		log.Error("", zap.Error(msg))
		return nil, false, errors.NewInternalError(msg)
	}

	if err := s.store.AddEvent(&utils.Event{
		UserID:         userID,
		Payload:        newObj,
		OperationType:  entity.UpdateOperation,
		ManagementType: entity.PlatformManagement,
		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 s.store.Controller().AssembleMetadataScheduledBackup(newCronJob), true, nil
}

func (s *Storage) Delete(ctx context.Context, name string, deleteValidation rest.ValidateObjectFunc, options *metav1.DeleteOptions) (runtime.Object, bool, error) {
	startTime := time.Now()
	var err error
	namespace := s.store.Namespace
	userID := ""
	if user, found := request.UserFrom(ctx); found {
		userID = user.GetUID()
	}
	if err = s.store.Controller().DeleteCronJob(name, namespace); err != nil {
		msg := fmt.Errorf("failed to delete CronJob[%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:         userID,
		Payload:        fmt.Sprintf("delete cluster schedule backup: %s", name),
		OperationType:  entity.DeleteOperation,
		ManagementType: entity.PlatformManagement,
		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.MetadataScheduledBackupList{}
}

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

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