/*
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"
	dbapi "kubedb.dev/apimachinery/apis/kubedb/v1"
	opsapi "kubedb.dev/apimachinery/apis/ops/v1alpha1"
	opsutil "kubedb.dev/apimachinery/pkg/webhooks/ops"

	"github.com/Masterminds/semver/v3"
	"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/api/resource"
	"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"
)

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

type ElasticsearchOpsRequestCustomWebhook struct {
	DefaultClient client.Client
}

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

var _ webhook.CustomValidator = &ElasticsearchOpsRequestCustomWebhook{}

// ValidateCreate implements webhook.Validator so a webhook will be registered for the type
func (w *ElasticsearchOpsRequestCustomWebhook) ValidateCreate(ctx context.Context, obj runtime.Object) (admission.Warnings, error) {
	ops, ok := obj.(*opsapi.ElasticsearchOpsRequest)
	if !ok {
		return nil, fmt.Errorf("expected an ElasticsearchOpsRequest object but got %T", obj)
	}
	esLog.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 *ElasticsearchOpsRequestCustomWebhook) ValidateUpdate(ctx context.Context, oldObj, newObj runtime.Object) (admission.Warnings, error) {
	ops, ok := newObj.(*opsapi.ElasticsearchOpsRequest)
	if !ok {
		return nil, fmt.Errorf("expected an ElasticsearchOpsRequest object but got %T", newObj)
	}
	esLog.Info("validate update", "name", ops.Name)

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

	if err := validateElasticsearchOpsRequest(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 dbapi.Elasticsearch
		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 (w *ElasticsearchOpsRequestCustomWebhook) ValidateDelete(ctx context.Context, obj runtime.Object) (admission.Warnings, error) {
	return nil, nil
}

func validateElasticsearchOpsRequest(obj, oldObj runtime.Object) error {
	preconditions := meta_util.PreConditionSet{Set: sets.New[string]("spec")}
	_, err := meta_util.CreateStrategicPatch(oldObj, obj, preconditions.PreconditionFunc()...)
	if err != nil {
		if mergepatch.IsPreconditionFailed(err) {
			return fmt.Errorf("%v.%v", err, preconditions.Error())
		}
		return err
	}
	return nil
}

func (w *ElasticsearchOpsRequestCustomWebhook) validateOpensearchVersionCompatibility(req *opsapi.ElasticsearchOpsRequest, db *dbapi.Elasticsearch) (bool, error) {
	if req.Spec.Type != opsapi.ElasticsearchOpsRequestTypeReconfigure && req.Spec.Type != opsapi.ElasticsearchOpsRequestTypeRotateAuth {
		return false, nil
	}
	esversion := &catalog.ElasticsearchVersion{}
	err := w.DefaultClient.Get(context.TODO(), types.NamespacedName{
		Name: db.Spec.Version,
	}, esversion)
	if err != nil {
		return true, err
	}
	version, err := semver.NewVersion(esversion.Spec.Version)
	if err != nil {
		return true, err
	}

	if !db.Spec.EnableSSL && esversion.Spec.AuthPlugin == catalog.ElasticsearchAuthPluginOpenSearch && version.Major() > 1 {
		return true, nil
	}

	return false, nil
}

func (w *ElasticsearchOpsRequestCustomWebhook) validateCreateOrUpdate(req *opsapi.ElasticsearchOpsRequest) (admission.Warnings, error) {
	if validType, _ := arrays.Contains(opsapi.ElasticsearchOpsRequestTypeNames(), 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 Elasticsearch are %s", req.Spec.Type, strings.Join(opsapi.ElasticsearchOpsRequestTypeNames(), ", ")))
	}
	var allErr field.ErrorList
	var warnings admission.Warnings
	db := &dbapi.Elasticsearch{}
	err := w.DefaultClient.Get(context.TODO(), types.NamespacedName{
		Name:      req.GetDBRefName(),
		Namespace: req.GetNamespace(),
	}, db)
	if err != nil && !apierrors.IsNotFound(err) {
		allErr = append(allErr, field.Invalid(field.NewPath("spec").Child("databaseRef"), req.Name,
			fmt.Sprintf("referenced database %s/%s is not found", req.Namespace, req.Spec.DatabaseRef.Name)))
	}
	switch opsapi.ElasticsearchOpsRequestType(req.GetRequestType()) {
	case opsapi.ElasticsearchOpsRequestTypeUpdateVersion:
		if err := w.validateElasticsearchUpdateVersionOpsRequest(req, db); err != nil {
			allErr = append(allErr, field.Invalid(field.NewPath("spec").Child("updateVersion"),
				req.Name,
				err.Error()))
		}
	case opsapi.ElasticsearchOpsRequestTypeReconfigure:
		if err := w.validateElasticsearchReconfigureOpsRequest(req, db); err != nil {
			allErr = append(allErr, field.Invalid(field.NewPath("spec").Child("configuration"),
				req.Name,
				err.Error()))
		}
	case opsapi.ElasticsearchOpsRequestTypeRotateAuth:
		if err := w.validateElasticsearchRotateAuthenticationOpsRequest(req, db); err != nil {
			allErr = append(allErr, field.Invalid(field.NewPath("spec").Child("authentication"),
				req.Name,
				err.Error()))
		}
	case opsapi.ElasticsearchOpsRequestTypeStorageMigration:
		if err := w.validateElasticsearchStorageMigrationOpsRequest(req, db); err != nil {
			allErr = append(allErr, field.Invalid(field.NewPath("spec").Child("migration"),
				req.Name,
				err.Error()))
		}
	case opsapi.ElasticsearchOpsRequestTypeVolumeExpansion:
		if err := w.validateElasticsearchVolumeExpansionOpsRequest(req, db); err != nil {
			allErr = append(allErr, field.Invalid(field.NewPath("spec").Child("volumeExpansion"),
				req.Name,
				err.Error()))
		}
	case opsapi.ElasticsearchOpsRequestTypeVerticalScaling:
		warns, err := w.validateElasticsearchVerticalScalingOpsRequest(req)
		warnings = append(warnings, warns...)
		if err != nil {
			allErr = append(allErr, field.Invalid(field.NewPath("spec").Child("verticalScaling"),
				req.Name,
				err.Error()))
		}
	}

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

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

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

	return warnings, nil
}

func (w *ElasticsearchOpsRequestCustomWebhook) validateElasticsearchUpdateVersionOpsRequest(req *opsapi.ElasticsearchOpsRequest, db *dbapi.Elasticsearch) 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.ResourceKindElasticsearchVersion, 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 *ElasticsearchOpsRequestCustomWebhook) validateElasticsearchStorageMigrationOpsRequest(req *opsapi.ElasticsearchOpsRequest, db *dbapi.Elasticsearch) error {
	m := req.Spec.Migration
	if m == nil {
		return errors.New("spec.migration is required for StorageMigration type")
	}
	if m.Node == nil && m.Master == nil && m.Ingest == nil && m.Data == nil &&
		m.DataContent == nil && m.DataHot == nil && m.DataWarm == nil &&
		m.DataCold == nil && m.DataFrozen == nil && m.ML == nil && m.Transform == nil && m.Coordinating == nil {
		return errors.New("at least one node migration spec is required in spec.migration")
	}
	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")
	}

	validateComponent := func(spec *opsapi.StorageMigrationSpec, oldClassName string) error {
		if spec == nil {
			return nil
		}
		if spec.StorageClassName == nil {
			return errors.New("storageClassName is required in migration spec")
		}
		var newstorage, oldstorage storagev1.StorageClass
		if err := w.DefaultClient.Get(context.TODO(), types.NamespacedName{Name: *spec.StorageClassName}, &newstorage); err != nil {
			if apierrors.IsNotFound(err) {
				return errors.Wrap(err, fmt.Sprintf("storage class %s not found", *spec.StorageClassName))
			}
			return err
		}
		if err := w.DefaultClient.Get(context.TODO(), types.NamespacedName{Name: oldClassName}, &oldstorage); err != nil {
			return err
		}
		if *oldstorage.VolumeBindingMode == storagev1.VolumeBindingWaitForFirstConsumer {
			if *newstorage.VolumeBindingMode != storagev1.VolumeBindingWaitForFirstConsumer {
				return errors.New(fmt.Sprintf("volume binding mode should be WaitForFirstConsumer for %s storageClass", newstorage.Name))
			}
		}
		return nil
	}

	if db.Spec.Topology != nil {
		t := db.Spec.Topology
		if err := validateComponent(m.Master, *t.Master.Storage.StorageClassName); err != nil {
			return err
		}
		if err := validateComponent(m.Ingest, *t.Ingest.Storage.StorageClassName); err != nil {
			return err
		}
		if m.Data != nil {
			if err := validateComponent(m.Data, *t.Data.Storage.StorageClassName); err != nil {
				return err
			}
		}
		if m.DataContent != nil {
			if err := validateComponent(m.DataContent, *t.DataContent.Storage.StorageClassName); err != nil {
				return err
			}
		}
		if m.DataHot != nil {
			if err := validateComponent(m.DataHot, *t.DataHot.Storage.StorageClassName); err != nil {
				return err
			}
		}
		if m.DataWarm != nil {
			if err := validateComponent(m.DataWarm, *t.DataWarm.Storage.StorageClassName); err != nil {
				return err
			}
		}
		if m.DataCold != nil {
			if err := validateComponent(m.DataCold, *t.DataCold.Storage.StorageClassName); err != nil {
				return err
			}
		}
		if m.DataFrozen != nil {
			if err := validateComponent(m.DataFrozen, *t.DataFrozen.Storage.StorageClassName); err != nil {
				return err
			}
		}
		if m.ML != nil {
			if err := validateComponent(m.ML, *t.ML.Storage.StorageClassName); err != nil {
				return err
			}
		}
		if m.Transform != nil {
			if err := validateComponent(m.Transform, *t.Transform.Storage.StorageClassName); err != nil {
				return err
			}
		}
		if m.Coordinating != nil {
			if err := validateComponent(m.Coordinating, *t.Coordinating.Storage.StorageClassName); err != nil {
				return err
			}
		}
	} else {
		if err := validateComponent(m.Node, db.GetStorageClassName()); err != nil {
			return err
		}
	}

	return nil
}

func (w *ElasticsearchOpsRequestCustomWebhook) validateElasticsearchRotateAuthenticationOpsRequest(req *opsapi.ElasticsearchOpsRequest, db *dbapi.Elasticsearch) error {
	issue, err := w.validateOpensearchVersionCompatibility(req, db)
	if err != nil {
		return err
	}
	if issue {
		return fmt.Errorf("opsrequest %s/%s is not compatible for version %s", req.Namespace, req.Name, db.Spec.Version)
	}
	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
}

func (w *ElasticsearchOpsRequestCustomWebhook) validateElasticsearchReconfigureOpsRequest(req *opsapi.ElasticsearchOpsRequest, db *dbapi.Elasticsearch) error {
	issue, err := w.validateOpensearchVersionCompatibility(req, db)
	if err != nil {
		return err
	}
	if issue {
		return fmt.Errorf("opsrequest %s/%s is not compatible for version %s", req.Namespace, req.Name, db.Spec.Version)
	}
	configuration := req.Spec.Configuration
	if configuration == nil {
		return fmt.Errorf("configuration can not be empty for %s/%s", req.Namespace, req.Name)
	}

	return nil
}

func (w *ElasticsearchOpsRequestCustomWebhook) validateElasticsearchVolumeExpansionOpsRequest(req *opsapi.ElasticsearchOpsRequest, db *dbapi.Elasticsearch) error {
	volumeExpansionSpec := req.Spec.VolumeExpansion
	if volumeExpansionSpec == nil {
		return errors.New("spec.volumeExpansion nil not supported in VolumeExpansion type")
	}

	// combined mode: db.Spec.Topology == nil
	if db.Spec.Topology == nil && volumeExpansionSpec.Node != nil {
		if err := opsutil.ValidateStorageExpansion(db.Spec.Storage, volumeExpansionSpec.Node, req.Status.Phase, "combined Elasticsearch"); err != nil {
			return err
		}
	}

	checkNodeStorage := func(nodeName string, nodeSpec *dbapi.ElasticsearchNode, desired *resource.Quantity) error {
		if desired == nil || nodeSpec == nil {
			return nil
		}
		return opsutil.ValidateStorageExpansion(nodeSpec.Storage, desired, req.Status.Phase, nodeName)
	}

	if db.Spec.Topology != nil {
		if err := checkNodeStorage("master", &db.Spec.Topology.Master, volumeExpansionSpec.Master); err != nil {
			return err
		}
		if err := checkNodeStorage("ingest", &db.Spec.Topology.Ingest, volumeExpansionSpec.Ingest); err != nil {
			return err
		}
		if err := checkNodeStorage("data", db.Spec.Topology.Data, volumeExpansionSpec.Data); err != nil {
			return err
		}
		if err := checkNodeStorage("dataContent", db.Spec.Topology.DataContent, volumeExpansionSpec.DataContent); err != nil {
			return err
		}
		if err := checkNodeStorage("dataHot", db.Spec.Topology.DataHot, volumeExpansionSpec.DataHot); err != nil {
			return err
		}
		if err := checkNodeStorage("dataWarm", db.Spec.Topology.DataWarm, volumeExpansionSpec.DataWarm); err != nil {
			return err
		}
		if err := checkNodeStorage("dataCold", db.Spec.Topology.DataCold, volumeExpansionSpec.DataCold); err != nil {
			return err
		}
		if err := checkNodeStorage("dataFrozen", db.Spec.Topology.DataFrozen, volumeExpansionSpec.DataFrozen); err != nil {
			return err
		}
		if err := checkNodeStorage("ml", db.Spec.Topology.ML, volumeExpansionSpec.ML); err != nil {
			return err
		}
		if err := checkNodeStorage("transform", db.Spec.Topology.Transform, volumeExpansionSpec.Transform); err != nil {
			return err
		}
		if err := checkNodeStorage("coordinating", db.Spec.Topology.Coordinating, volumeExpansionSpec.Coordinating); err != nil {
			return err
		}
	}

	return nil
}
