/*
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"
	"errors"
	"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"

	"gomodules.xyz/x/arrays"
	core "k8s.io/api/core/v1"
	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"
)

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

type ClickHouseOpsRequestCustomWebhook struct {
	DefaultClient client.Client
}

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

var _ webhook.CustomValidator = &ClickHouseOpsRequestCustomWebhook{}

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

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

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

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

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

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

	switch opsapi.ClickHouseOpsRequestType(req.GetRequestType()) {
	case opsapi.ClickHouseOpsRequestTypeRestart:

	case opsapi.ClickHouseOpsRequestTypeVerticalScaling:
		if err := rv.validateClickHouseVerticalScalingOpsRequest(db, req); err != nil {
			allErr = append(allErr, field.Invalid(field.NewPath("spec").Child("verticalScaling"),
				req.Name,
				err.Error()))
		}
	case opsapi.ClickHouseOpsRequestTypeUpdateVersion:
		if err := rv.validateClickHouseUpdateVersionOpsRequest(db, req); err != nil {
			allErr = append(allErr, field.Invalid(field.NewPath("spec").Child("updateVersion"),
				req.Name,
				err.Error()))
		}
	case opsapi.ClickHouseOpsRequestTypeVolumeExpansion:
		if err := rv.validateClickHouseVolumeExpansionOpsRequest(db, req); err != nil {
			allErr = append(allErr, field.Invalid(field.NewPath("spec").Child("volumeExpansion"),
				req.Name,
				err.Error()))
		}
	case opsapi.ClickHouseOpsRequestTypeHorizontalScaling:
		if err := rv.validateClickHouseHorizontalScalingOpsRequest(db, req); err != nil {
			allErr = append(allErr, field.Invalid(field.NewPath("spec").Child("horizontalScaling"),
				req.Name,
				err.Error()))
		}
	case opsapi.ClickHouseOpsRequestTypeReconfigure:
		if err := rv.validateClickHouseReconfigurationOpsRequest(req); err != nil {
			allErr = append(allErr, field.Invalid(field.NewPath("spec").Child("configuration"),
				req.Name,
				err.Error()))
		}
	case opsapi.ClickHouseOpsRequestTypeReconfigureTLS:
		if err := rv.validateClickHouseReconfigurationTLSOpsRequest(req); err != nil {
			allErr = append(allErr, field.Invalid(field.NewPath("spec").Child("tls"),
				req.Name,
				err.Error()))
		}
	case opsapi.ClickHouseOpsRequestTypeRotateAuth:
		if err := rv.validateClickHouseRotateAuthenticationOpsRequest(req); err != nil {
			allErr = append(allErr, field.Invalid(field.NewPath("spec").Child("authentication"),
				req.Name,
				err.Error()))
		}
	case opsapi.ClickHouseOpsRequestTypeStorageMigration:
		if err := rv.validateClickHouseStorageMigrationOpsRequest(req, db); err != nil {
			allErr = append(allErr, field.Invalid(field.NewPath("spec").Child("migration"),
				req.Name,
				err.Error()))
		}
	}

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

func (rv *ClickHouseOpsRequestCustomWebhook) validateClickHouseStorageMigrationOpsRequest(req *opsapi.ClickHouseOpsRequest, db *olddbapi.ClickHouse) error {
	m := req.Spec.Migration
	if m == nil {
		return errors.New("spec.migration is required for StorageMigration type")
	}
	if m.Standalone == nil && m.Cluster == nil && m.ClickHouseKeeper == nil {
		return errors.New("at least one component 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, storage *core.PersistentVolumeClaimSpec) error {
		if spec == nil {
			return nil
		}
		if spec.StorageClassName == nil {
			return errors.New("storageClassName is required in migration spec")
		}
		if storage == nil || storage.StorageClassName == nil {
			return nil
		}
		var newstorage, oldstorage storagev1.StorageClass
		if err := rv.DefaultClient.Get(context.TODO(), types.NamespacedName{Name: *spec.StorageClassName}, &newstorage); err != nil {
			if apierrors.IsNotFound(err) {
				return fmt.Errorf("storage class %s not found: %w", *spec.StorageClassName, err)
			}
			return err
		}
		if err := rv.DefaultClient.Get(context.TODO(), types.NamespacedName{Name: *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
	}

	if db.Spec.ClusterTopology != nil {
		t := db.Spec.ClusterTopology
		if err := validateComponent(m.Cluster, t.Cluster.Storage); err != nil {
			return fmt.Errorf("cluster: %w", err)
		}
		if m.ClickHouseKeeper != nil && t.ClickHouseKeeper != nil && t.ClickHouseKeeper.Spec != nil {
			if err := validateComponent(m.ClickHouseKeeper, t.ClickHouseKeeper.Spec.Storage); err != nil {
				return fmt.Errorf("clickHouseKeeper: %w", err)
			}
		}
	} else {
		if err := validateComponent(m.Standalone, db.Spec.Storage); err != nil {
			return fmt.Errorf("standalone: %w", err)
		}
	}
	return nil
}

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

func (rv *ClickHouseOpsRequestCustomWebhook) validateClickHouseVerticalScalingOpsRequest(db *olddbapi.ClickHouse, req *opsapi.ClickHouseOpsRequest) error {
	verticalScalingSpec := req.Spec.VerticalScaling
	if verticalScalingSpec == nil {
		return errors.New("spec.verticalScaling nil not supported in VerticalScaling type")
	}
	if verticalScalingSpec.Node == nil && verticalScalingSpec.ClickHouseKeeper == nil {
		return errors.New("at least one of `spec.verticalScaling.node` or `spec.verticalScaling.clickHouseKeeper` must be provided")
	}
	if verticalScalingSpec.ClickHouseKeeper != nil {
		if err := rv.validateClickHouseKeeperOpsRequest(db); err != nil {
			return fmt.Errorf("clickHouseKeeper: %w", err)
		}
	}

	return nil
}

func (rv *ClickHouseOpsRequestCustomWebhook) validateClickHouseUpdateVersionOpsRequest(db *olddbapi.ClickHouse, req *opsapi.ClickHouseOpsRequest) error {
	updateVersionSpec := req.Spec.UpdateVersion
	if updateVersionSpec == nil {
		return errors.New("spec.updateVersion nil not supported in UpdateVersion type")
	}

	yes, err := IsUpgradable(rv.DefaultClient, catalog.ResourceKindClickHouseVersion, 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 (rv *ClickHouseOpsRequestCustomWebhook) validateClickHouseVolumeExpansionOpsRequest(db *olddbapi.ClickHouse, req *opsapi.ClickHouseOpsRequest) error {
	volumeExpansionSpec := req.Spec.VolumeExpansion
	if volumeExpansionSpec == nil {
		return errors.New("spec.volumeExpansion nil not supported in VolumeExpansion type")
	}

	if volumeExpansionSpec.Node == nil && volumeExpansionSpec.ClickHouseKeeper == nil {
		return errors.New("at least one of `spec.volumeExpansion.node` or `spec.volumeExpansion.clickHouseKeeper` must be provided")
	}

	if volumeExpansionSpec.Node != nil {
		var storage *core.PersistentVolumeClaimSpec
		if db.Spec.ClusterTopology != nil {
			storage = db.Spec.ClusterTopology.Cluster.Storage
		} else {
			storage = db.Spec.Storage
		}
		if err := opsutil.ValidateStorageExpansion(storage, volumeExpansionSpec.Node, req.Status.Phase, "ClickHouse"); err != nil {
			return err
		}
	}

	if volumeExpansionSpec.ClickHouseKeeper != nil {
		if err := rv.validateClickHouseKeeperOpsRequest(db); err != nil {
			return fmt.Errorf("clickHouseKeeper: %w", err)
		}
		if db.Spec.ClusterTopology != nil && db.Spec.ClusterTopology.ClickHouseKeeper != nil && !db.Spec.ClusterTopology.ClickHouseKeeper.ExternallyManaged && db.Spec.ClusterTopology.ClickHouseKeeper.Spec != nil {
			if err := opsutil.ValidateStorageExpansion(db.Spec.ClusterTopology.ClickHouseKeeper.Spec.Storage, volumeExpansionSpec.ClickHouseKeeper, req.Status.Phase, "ClickHouseKeeper"); err != nil {
				return err
			}
		}
	}

	return nil
}

func (rv *ClickHouseOpsRequestCustomWebhook) validateClickHouseHorizontalScalingOpsRequest(db *olddbapi.ClickHouse, req *opsapi.ClickHouseOpsRequest) error {
	horizontalScalingSpec := req.Spec.HorizontalScaling
	if horizontalScalingSpec == nil {
		return errors.New("`spec.horizontalScaling` nil not supported in HorizontalScaling type")
	}

	if horizontalScalingSpec.Replicas == nil && horizontalScalingSpec.Shards == nil && horizontalScalingSpec.ClickHouseKeeper == nil {
		return errors.New("at least one of `spec.horizontalScaling.replicas`, `spec.horizontalScaling.shards`, or `spec.horizontalScaling.clickHouseKeeper` must be provided")
	}

	if horizontalScalingSpec.Replicas != nil && *horizontalScalingSpec.Replicas <= 0 {
		return errors.New("`spec.horizontalScaling.replicas` must be greater than 0")
	}

	if horizontalScalingSpec.ClickHouseKeeper != nil {
		if err := rv.validateClickHouseKeeperOpsRequest(db); err != nil {
			return fmt.Errorf("clickHouseKeeper: %w", err)
		}
		target := *horizontalScalingSpec.ClickHouseKeeper
		if target <= 0 {
			return errors.New("`spec.horizontalScaling.clickHouseKeeper` must be greater than 0")
		}
		// Only meaningful before the request starts: once it succeeds the database
		// really does have the requested replica count, and re-validating would
		// reject the operator's own closing status update.
		current := db.Spec.ClusterTopology.ClickHouseKeeper.Spec.Replicas
		if (req.Status.Phase == opsapi.OpsRequestPhasePending || req.Status.Phase == "") &&
			current != nil && target == *current {
			return fmt.Errorf("clickHouseKeeper already has %d replicas", target)
		}
	}

	if horizontalScalingSpec.Shards != nil {
		if *horizontalScalingSpec.Shards <= 0 {
			return errors.New("`spec.horizontalScaling.shards` must be greater than 0")
		}
		// shard scaling only applies to sharded (topology) clusters
		if db.Spec.ClusterTopology == nil {
			return errors.New("`spec.horizontalScaling.shards` is only supported for a sharded cluster (spec.clusterTopology)")
		}
		// scale-up only — validate at creation time to avoid re-validation failures
		if req.Status.Phase == opsapi.OpsRequestPhasePending || req.Status.Phase == "" {
			currentShards := int(*db.Spec.ClusterTopology.Cluster.Shards)
			if int(*horizontalScalingSpec.Shards) < currentShards {
				return fmt.Errorf("shard scale-up only: `spec.horizontalScaling.shards` (%d) must be greater than current shard count (%d)", *horizontalScalingSpec.Shards, currentShards)
			}
		}
	}

	return nil
}

func (rv *ClickHouseOpsRequestCustomWebhook) validateClickHouseReconfigurationOpsRequest(req *opsapi.ClickHouseOpsRequest) error {
	reconfigureSpec := req.Spec.Configuration
	if reconfigureSpec == nil {
		return errors.New("spec.configuration nil not supported in Reconfigure type")
	}

	if !reconfigureSpec.RemoveCustomConfig && reconfigureSpec.ConfigSecret == nil && req.Spec.Configuration.ApplyConfig == nil {
		return fmt.Errorf("at least one of `RemoveCustomConfig`, `ConfigSecret`, or `ApplyConfig` must be specified")
	}

	return nil
}

func (rv *ClickHouseOpsRequestCustomWebhook) validateClickHouseReconfigurationTLSOpsRequest(req *opsapi.ClickHouseOpsRequest) 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 (rv *ClickHouseOpsRequestCustomWebhook) validateClickHouseRotateAuthenticationOpsRequest(req *opsapi.ClickHouseOpsRequest) error {
	clickhouse := &olddbapi.ClickHouse{}
	err := rv.DefaultClient.Get(context.TODO(), types.NamespacedName{
		Namespace: req.Namespace,
		Name:      req.GetDBRefName(),
	}, clickhouse)
	if err != nil {
		return err
	}
	if clickhouse.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(), rv.DefaultClient, req.Namespace, authSpec.SecretRef); err != nil {
			return err
		}
	}
	return nil
}

func (rv *ClickHouseOpsRequestCustomWebhook) validateClickHouseKeeperOpsRequest(db *olddbapi.ClickHouse) error {
	if db.Spec.ClusterTopology == nil {
		return errors.New("clusterTopology is not set; clickHouseKeeper ops are only supported for topology mode")
	}
	if db.Spec.ClusterTopology.ClickHouseKeeper == nil {
		return errors.New("clickHouseKeeper is not configured in spec.clusterTopology")
	}
	if db.Spec.ClusterTopology.ClickHouseKeeper.ExternallyManaged {
		return errors.New("clickHouseKeeper is externally managed; ops are not supported")
	}
	if db.Spec.ClusterTopology.ClickHouseKeeper.Spec == nil {
		return errors.New("spec.clusterTopology.clickHouseKeeper.spec is not set")
	}
	return nil
}

func (rv *ClickHouseOpsRequestCustomWebhook) validateClickHouseOpsRequest(req *opsapi.ClickHouseOpsRequest, oldReq *opsapi.ClickHouseOpsRequest) 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
}
