/*
Copyright AppsCode Inc. and Contributors

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package v1alpha1

import (
	"context"
	"fmt"
	"strings"

	catalog "kubedb.dev/apimachinery/apis/catalog/v1alpha1"
	olddbapi "kubedb.dev/apimachinery/apis/kubedb/v1alpha2"
	opsapi "kubedb.dev/apimachinery/apis/ops/v1alpha1"
	opsutil "kubedb.dev/apimachinery/pkg/webhooks/ops"

	"github.com/pkg/errors"
	"gomodules.xyz/x/arrays"
	storagev1 "k8s.io/api/storage/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/apimachinery/pkg/util/mergepatch"
	"k8s.io/apimachinery/pkg/util/sets"
	"k8s.io/apimachinery/pkg/util/validation/field"
	meta_util "kmodules.xyz/client-go/meta"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	logf "sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/webhook"
	"sigs.k8s.io/controller-runtime/pkg/webhook/admission"
)

// SetupCassandraOpsRequestWebhookWithManager registers the webhook for CassandraOpsRequest in the manager.
func SetupCassandraOpsRequestWebhookWithManager(mgr ctrl.Manager) error {
	return ctrl.NewWebhookManagedBy(mgr).For(&opsapi.CassandraOpsRequest{}).
		WithValidator(&CassandraOpsRequestCustomWebhook{mgr.GetClient()}).
		Complete()
}

type CassandraOpsRequestCustomWebhook struct {
	DefaultClient client.Client
}

// log is for logging in this package.
var cassandraLog = logf.Log.WithName("cassandra-opsrequest")

var _ webhook.CustomValidator = &CassandraOpsRequestCustomWebhook{}

// ValidateCreate implements webhook.Validator so a webhook will be registered for the type
func (w *CassandraOpsRequestCustomWebhook) ValidateCreate(ctx context.Context, obj runtime.Object) (admission.Warnings, error) {
	ops, ok := obj.(*opsapi.CassandraOpsRequest)
	if !ok {
		return nil, fmt.Errorf("expected an CassandraOpsRequest object but got %T", obj)
	}
	cassandraLog.Info("validate create", "name", ops.Name)
	return w.validateCreateOrUpdate(ops)
}

// ValidateUpdate implements webhook.Validator so a webhook will be registered for the type
func (w *CassandraOpsRequestCustomWebhook) ValidateUpdate(ctx context.Context, oldObj, newObj runtime.Object) (admission.Warnings, error) {
	ops, ok := newObj.(*opsapi.CassandraOpsRequest)
	if !ok {
		return nil, fmt.Errorf("expected an CassandraOpsRequest object but got %T", newObj)
	}
	cassandraLog.Info("validate update", "name", ops.Name)

	oldOps, ok := oldObj.(*opsapi.CassandraOpsRequest)
	if !ok {
		return nil, fmt.Errorf("expected an CassandraOpsRequest object but got %T", oldObj)
	}

	if err := validateCassandraOpsRequest(ops, oldOps); err != nil {
		return nil, err
	}

	warnings, err := w.validateCreateOrUpdate(ops)
	if err != nil {
		return warnings, err
	}

	if isOpsReqCompleted(ops.Status.Phase) && !isOpsReqCompleted(oldOps.Status.Phase) { // just completed
		var db olddbapi.Cassandra
		err := w.DefaultClient.Get(context.TODO(), types.NamespacedName{Name: ops.Spec.DatabaseRef.Name, Namespace: ops.Namespace}, &db)
		if err != nil {
			return warnings, err
		}
		return warnings, resumeDatabase(w.DefaultClient, &db)
	}
	return warnings, nil
}

func validateCassandraOpsRequest(req *opsapi.CassandraOpsRequest, oldReq *opsapi.CassandraOpsRequest) error {
	preconditions := meta_util.PreConditionSet{Set: sets.New[string]("spec")}
	_, err := meta_util.CreateStrategicPatch(oldReq, req, preconditions.PreconditionFunc()...)
	if err != nil {
		if mergepatch.IsPreconditionFailed(err) {
			return fmt.Errorf("%v.%v", err, preconditions.Error())
		}
		return err
	}
	return nil
}

func (w *CassandraOpsRequestCustomWebhook) ValidateDelete(ctx context.Context, obj runtime.Object) (admission.Warnings, error) {
	return nil, nil
}

func (w *CassandraOpsRequestCustomWebhook) validateCreateOrUpdate(req *opsapi.CassandraOpsRequest) (admission.Warnings, error) {
	if validType, _ := arrays.Contains(opsapi.CassandraOpsRequestTypeNames(), string(req.Spec.Type)); !validType {
		return nil, field.Invalid(field.NewPath("spec").Child("type"), req.Name,
			fmt.Sprintf("defined OpsRequestType %s is not supported, supported types for Cassandra are %s", req.Spec.Type, strings.Join(opsapi.CassandraOpsRequestTypeNames(), ", ")))
	}
	db, err := w.hasDatabaseRef(req)
	if err != nil {
		return nil, err
	}
	var allErr field.ErrorList
	var warnings admission.Warnings

	switch opsapi.CassandraOpsRequestType(req.GetRequestType()) {
	case opsapi.CassandraOpsRequestTypeRestart:

	case opsapi.CassandraOpsRequestTypeHorizontalScaling:
		if err := w.validateCassandraHorizontalScalingOpsRequest(req); err != nil {
			allErr = append(allErr, field.Invalid(field.NewPath("spec").Child("horizontalScaling"),
				req.Name,
				err.Error()))
		}
	case opsapi.CassandraOpsRequestTypeVerticalScaling:
		warns, err := w.validateCassandraVerticalScalingOpsRequest(req)
		warnings = append(warnings, warns...)
		if err != nil {
			allErr = append(allErr, field.Invalid(field.NewPath("spec").Child("verticalScaling"),
				req.Name,
				err.Error()))
		}
	case opsapi.CassandraOpsRequestTypeVolumeExpansion:
		if err := w.validateCassandraVolumeExpansionOpsRequest(db, req); err != nil {
			allErr = append(allErr, field.Invalid(field.NewPath("spec").Child("volumeExpansion"),
				req.Name,
				err.Error()))
		}

	case opsapi.CassandraOpsRequestTypeUpdateVersion:
		if err := w.validateCassandraUpdateVersionOpsRequest(db, req); err != nil {
			allErr = append(allErr, field.Invalid(field.NewPath("spec").Child("updateVersion"),
				req.Name,
				err.Error()))
		}
	case opsapi.CassandraOpsRequestTypeReconfigure:
		if err := w.validateCassandraReconfigurationOpsRequest(req); err != nil {
			allErr = append(allErr, field.Invalid(field.NewPath("spec").Child("configuration"),
				req.Name,
				err.Error()))
		}
	case opsapi.CassandraOpsRequestTypeReconfigureTLS:
		if err := w.validateCassandraReconfigurationTLSOpsRequest(req); err != nil {
			allErr = append(allErr, field.Invalid(field.NewPath("spec").Child("tls"),
				req.Name,
				err.Error()))
		}
	case opsapi.CassandraOpsRequestTypeRotateAuth:
		if err := w.validateCassandraRotateAuthenticationOpsRequest(req); err != nil {
			allErr = append(allErr, field.Invalid(field.NewPath("spec").Child("authentication"),
				req.Name,
				err.Error()))
		}
	case opsapi.CassandraOpsRequestTypeStorageMigration:
		if err := w.validateCassandraStorageMigrationOpsRequest(req, db); err != nil {
			allErr = append(allErr, field.Invalid(field.NewPath("spec").Child("migration"),
				req.Name,
				err.Error()))
		}
	}

	if len(allErr) == 0 {
		return warnings, nil
	}
	return warnings, apierrors.NewInvalid(schema.GroupKind{Group: "Cassandraopsrequests.kubedb.com", Kind: "CassandraOpsRequest"}, req.Name, allErr)
}

func (w *CassandraOpsRequestCustomWebhook) hasDatabaseRef(req *opsapi.CassandraOpsRequest) (*olddbapi.Cassandra, error) {
	cassandra := &olddbapi.Cassandra{}
	if err := w.DefaultClient.Get(context.TODO(), types.NamespacedName{
		Name:      req.GetDBRefName(),
		Namespace: req.GetNamespace(),
	}, cassandra); err != nil {
		return nil, fmt.Errorf("spec.databaseRef %s/%s, is invalid or not found", req.GetNamespace(), req.GetDBRefName())
	}
	return cassandra, nil
}

func (w *CassandraOpsRequestCustomWebhook) validateCassandraVerticalScalingOpsRequest(req *opsapi.CassandraOpsRequest) (admission.Warnings, error) {
	verticalScalingSpec := req.Spec.VerticalScaling
	if verticalScalingSpec == nil {
		return nil, errors.New("spec.verticalScaling nil not supported in VerticalScaling type")
	}

	if verticalScalingSpec.Node == nil {
		return nil, errors.New("spec.verticalScaling.Node can't be empty")
	}

	var warnings admission.Warnings
	if verticalScalingSpec.Mode == opsapi.VerticalScalingModeInPlace {
		warnings = append(warnings, "in-place vertical scaling is not recommended for Cassandra: JVM heap is derived from container resources and requires a pod restart to take effect")
	}

	return warnings, nil
}

func (w *CassandraOpsRequestCustomWebhook) validateCassandraHorizontalScalingOpsRequest(req *opsapi.CassandraOpsRequest) error {
	horizontalScalingSpec := req.Spec.HorizontalScaling
	if horizontalScalingSpec == nil {
		return errors.New("spec.horizontalScaling nil not supported in HorizontalScaling type")
	}

	if horizontalScalingSpec.Node == nil {
		return errors.New("spec.horizontalScaling.node can not be empty")
	}

	if *horizontalScalingSpec.Node <= 0 {
		return errors.New("spec.horizontalScaling.node must be positive")
	}

	return nil
}

func (w *CassandraOpsRequestCustomWebhook) validateCassandraVolumeExpansionOpsRequest(db *olddbapi.Cassandra, req *opsapi.CassandraOpsRequest) error {
	volumeExpansionSpec := req.Spec.VolumeExpansion
	if volumeExpansionSpec == nil {
		return errors.New("spec.volumeExpansion nil not supported in VolumeExpansion type")
	}

	if volumeExpansionSpec.Node == nil {
		return errors.New("spec.volumeExpansion.Node can't be empty")
	}

	if db.Spec.Topology == nil {
		if err := opsutil.ValidateStorageExpansion(db.Spec.Storage, volumeExpansionSpec.Node, req.Status.Phase, "Cassandra"); err != nil {
			return err
		}
	} else {
		for _, rack := range db.Spec.Topology.Rack {
			if err := opsutil.ValidateStorageExpansion(rack.Storage, volumeExpansionSpec.Node, req.Status.Phase, fmt.Sprintf("Cassandra rack %q", rack.Name)); err != nil {
				return err
			}
		}
	}

	return nil
}

func (w *CassandraOpsRequestCustomWebhook) validateCassandraUpdateVersionOpsRequest(db *olddbapi.Cassandra, req *opsapi.CassandraOpsRequest) error {
	updateVersionSpec := req.Spec.UpdateVersion
	if updateVersionSpec == nil {
		return errors.New("spec.updateVersion nil not supported in UpdateVersion type")
	}

	yes, err := IsUpgradable(w.DefaultClient, catalog.ResourceKindCassandraVersion, db.Spec.Version, updateVersionSpec.TargetVersion)
	if err != nil {
		return err
	}
	if !yes {
		return fmt.Errorf("upgrade from version %v to %v is not supported", db.Spec.Version, req.Spec.UpdateVersion.TargetVersion)
	}

	return nil
}

func (w *CassandraOpsRequestCustomWebhook) validateCassandraReconfigurationOpsRequest(req *opsapi.CassandraOpsRequest) error {
	configurationSpec := req.Spec.Configuration
	if configurationSpec == nil {
		return errors.New("spec.configuration nil not supported in Reconfigure type")
	}

	if !configurationSpec.RemoveCustomConfig && configurationSpec.ConfigSecret == nil && len(configurationSpec.ApplyConfig) == 0 {
		return errors.New("at least one of `RemoveCustomConfig`, `ConfigSecret`, or `ApplyConfig` must be specified")
	}
	return nil
}

func (w *CassandraOpsRequestCustomWebhook) validateCassandraReconfigurationTLSOpsRequest(req *opsapi.CassandraOpsRequest) error {
	TLSSpec := req.Spec.TLS
	if TLSSpec == nil {
		return errors.New("spec.TLS nil not supported in ReconfigureTLS type")
	}
	configCount := 0
	if req.Spec.TLS.Remove {
		configCount++
	}
	if req.Spec.TLS.RotateCertificates {
		configCount++
	}
	if req.Spec.TLS.IssuerRef != nil || req.Spec.TLS.Certificates != nil {
		configCount++
	}

	if configCount == 0 {
		return errors.New("no reconfiguration is provided in TLS spec")
	}

	if configCount > 1 {
		return errors.New("more than 1 field have assigned to spec.reconfigureTLS but at a time one is allowed to run one operation")
	}
	return nil
}

func (w *CassandraOpsRequestCustomWebhook) validateCassandraStorageMigrationOpsRequest(req *opsapi.CassandraOpsRequest, db *olddbapi.Cassandra) error {
	m := req.Spec.Migration
	if m == nil {
		return errors.New("spec.migration is required for StorageMigration type")
	}
	if m.StorageClassName == nil {
		return errors.New("spec.migration.storageClassName is required")
	}
	if req.Spec.Timeout == nil {
		return errors.New("spec.timeout is required for Storage Migration ops request, adjust timeout according to the size of your database")
	}
	if db.Spec.Storage == nil || db.Spec.Storage.StorageClassName == nil {
		return fmt.Errorf("db.Spec.Storage.StorageClassName can't be nil in the database yaml")
	}
	var newstorage, oldstorage storagev1.StorageClass
	if err := w.DefaultClient.Get(context.TODO(), types.NamespacedName{Name: *m.StorageClassName}, &newstorage); err != nil {
		if apierrors.IsNotFound(err) {
			return errors.Wrap(err, fmt.Sprintf("storage class %s not found", *m.StorageClassName))
		}
		return err
	}
	if err := w.DefaultClient.Get(context.TODO(), types.NamespacedName{Name: *db.Spec.Storage.StorageClassName}, &oldstorage); err != nil {
		return err
	}
	if *oldstorage.VolumeBindingMode == storagev1.VolumeBindingWaitForFirstConsumer {
		if *newstorage.VolumeBindingMode != storagev1.VolumeBindingWaitForFirstConsumer {
			return fmt.Errorf("volume binding mode should be WaitForFirstConsumer for %s storageClass", newstorage.Name)
		}
	}
	return nil
}

func (w *CassandraOpsRequestCustomWebhook) validateCassandraRotateAuthenticationOpsRequest(req *opsapi.CassandraOpsRequest) error {
	cassandra := &olddbapi.Cassandra{}
	err := w.DefaultClient.Get(context.TODO(), types.NamespacedName{
		Namespace: req.Namespace,
		Name:      req.GetDBRefName(),
	}, cassandra)
	if err != nil {
		return err
	}
	if cassandra.Spec.DisableSecurity {
		return fmt.Errorf("disableSecurity is on, RotateAuth is not applicable")
	}
	authSpec := req.Spec.Authentication
	if authSpec != nil && authSpec.SecretRef != nil {
		if err := validateAuthSecretRef(context.TODO(), w.DefaultClient, req.Namespace, authSpec.SecretRef); err != nil {
			return err
		}
	}
	return nil
}
