/*
Copyright 2022.

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 clusterconfig

import (
	"context"

	"k8s.io/apimachinery/pkg/fields"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/types"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/builder"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/handler"
	"sigs.k8s.io/controller-runtime/pkg/predicate"
	"sigs.k8s.io/controller-runtime/pkg/reconcile"
	"sigs.k8s.io/controller-runtime/pkg/source"

	"github.com/go-logr/logr"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-api/controllers/revision"
	gitopsv2 "github.com/tidbcloud/infra-api/pkg/gitops-v2"
)

const (
	ConfigRevisionRefField = ".spec.configRevisionRef.name"
)

// ClusterConfigReconciler reconciles a ClusterConfig object
type ClusterConfigReconciler struct {
	client.Client
	revisioner revision.Interface
	g          gitopsv2.Interface
	Log        logr.Logger
}

// Reconcile with cluster object, for every enqueued cluster,
// 1. find related raw ClusterConfigs in infra-api
// 2. evaluate ClusterConfig by layered ClusterConfigs and Cluster
// 3. calculate hash over evaluated ClusterConfig and create ConfigRevision
// 4. change RevisionRef on the Cluster.
//+kubebuilder:rbac:groups=cloud.pingcap.com,resources=clusterconfigs,verbs=get;list;watch;create;update;patch;delete
//+kubebuilder:rbac:groups=cloud.pingcap.com,resources=clusterconfigs/status,verbs=get;update;patch
//+kubebuilder:rbac:groups=cloud.pingcap.com,resources=clusterconfigs/finalizers,verbs=update
//+kubebuilder:rbac:groups=cloud.pingcap.com,resources=clusters,verbs=get;list;watch;update;patch
//+kubebuilder:rbac:groups=cloud.pingcap.com,resources=configrevisions,verbs=get;list;watch;create;update;patch;delete
func (r *ClusterConfigReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	log := r.Log.WithValues("cluster", req.NamespacedName)

	c := &cloudv1.Cluster{}
	if err := r.Get(ctx, req.NamespacedName, c); err != nil {
		// we'll ignore not-found errors, since they can't be fixed by an immediate
		// requeue (we'll need to wait for a new notification), and we can get them
		// on deleted requests.
		return ctrl.Result{}, client.IgnoreNotFound(err)
	}

	// Do nothing if the cluster is deleting.
	// ClusterConfig will be GCed by background workers.
	if !c.GetDeletionTimestamp().IsZero() {
		return ctrl.Result{}, nil
	}

	// calculate overlayed ClusterConfig, we treat this as VirtualNode
	cc, err := r.g.DescribeClusterConfiguration(c)
	if err != nil {
		return ctrl.Result{}, err
	}
	if cc.IsDryrun() {
		return ctrl.Result{}, nil
	}

	cr, err := GetClusterRevision(r.revisioner, cc)
	if err != nil {
		return ctrl.Result{}, err
	}

	err = r.updateClusterRevision(ctx, c, cr)
	if err != nil {
		log.Error(err, "update cluster failed")
		return ctrl.Result{}, err
	}

	return ctrl.Result{}, nil
}

func (r *ClusterConfigReconciler) updateClusterRevision(ctx context.Context, c *cloudv1.Cluster, rev *cloudv1.ConfigRevision) error {
	if c.Spec.ConfigRevisionRef != nil && c.Spec.ConfigRevisionRef.Name == rev.Name {
		return nil
	}
	c.Spec.ConfigRevisionRef = &cloudv1.Reference{
		Name: rev.Name,
	}
	return r.Update(ctx, c)
}

// GetClusterRevision returns the desired ConfigRevision for the cluster.
// This method may create a new revision.
func GetClusterRevision(rv revision.Interface, cc *cloudv1.ClusterConfig) (*cloudv1.ConfigRevision, error) {
	crs, err := rv.ListConfigRevisions(labels.SelectorFromSet(labels.Set(cc.Labels)))
	if err != nil {
		return nil, err
	}

	// find if ConfigRevision exists for this ClusterConfig
	cr := revision.NewConfigRevision(cc)
	for i := range crs {
		if revision.EqualRevision(&crs[i], cr) {
			err := rv.RenewConfigRevision(&crs[i])
			if err != nil {
				return nil, err
			}
			return &crs[i], nil
		}
	}

	created, err := rv.CreateConfigRevision(cr)
	if err != nil {
		return nil, err
	}
	return created, nil
}

func NewClusterConfigReconciler(mgr ctrl.Manager, log logr.Logger) *ClusterConfigReconciler {
	return &ClusterConfigReconciler{
		Log:        log,
		Client:     mgr.GetClient(),
		revisioner: revision.NewRevisioner(mgr.GetClient()),
		g:          &gitopsv2.Gitops{Cli: mgr.GetClient()},
	}
}

// SetupWithManager sets up the controller with the Manager.
func (r *ClusterConfigReconciler) SetupWithManager(mgr ctrl.Manager) error {
	if err := mgr.GetFieldIndexer().IndexField(context.Background(), &cloudv1.Cluster{}, ConfigRevisionRefField, func(o client.Object) []string {
		c := o.(*cloudv1.Cluster)
		if c.Spec.ConfigRevisionRef == nil {
			return nil
		}
		return []string{c.Spec.ConfigRevisionRef.Name}
	}); err != nil {
		return err
	}

	// enqueue all clusters and clusters that are related to ClusterConfig.
	return ctrl.NewControllerManagedBy(mgr).
		For(&cloudv1.Cluster{}, builder.WithPredicates(instanceTypeChangedPredicate{})).
		Watches(
			&source.Kind{Type: &cloudv1.ClusterConfig{}},
			handler.EnqueueRequestsFromMapFunc(r.findClusterFromClusterConfig),
			builder.WithPredicates(predicate.And(predicate.GenerationChangedPredicate{}, omitConfigMetaChangePredicate{})),
		).
		Complete(r)
}

func (r *ClusterConfigReconciler) findClusterFromClusterConfig(clusterConfig client.Object) []reconcile.Request {
	cc, ok := clusterConfig.(*cloudv1.ClusterConfig)
	if !ok {
		r.Log.Error(nil, "object is not ClusterConfig", "object", clusterConfig)
		return []reconcile.Request{}
	}
	relatedClusters := &cloudv1.ClusterList{}
	if _, exist := cc.Spec.Selector.MatchLabels[cloudv1.TenantLabel]; exist {
		// find matched clusters via tenant/cluster selector
		selector, err := metav1.LabelSelectorAsSelector(cc.Spec.Selector)
		if err != nil {
			r.Log.Error(err, "invalid label selector", "labelselector", cc.Spec.Selector)
			return []reconcile.Request{}
		}
		err = r.List(context.TODO(), relatedClusters, &client.ListOptions{
			LabelSelector: selector,
		})
		if err != nil {
			r.Log.Error(err, "failed to list clusters for clusterconfig %s", cc.Name)
			return []reconcile.Request{}
		}
	} else {
		// for base and instanceType clusterconfig, find all the clusters via version selector.
		selector := labels.SelectorFromSet(labels.Set(map[string]string{
			cloudv1.VersionLabel: cc.Labels[cloudv1.VersionLabel],
		}))

		crs, err := r.revisioner.ListConfigRevisions(selector)
		if err != nil {
			r.Log.Error(err, "failed to list config revisions for clusterconfig %s", cc.Name)
			return []reconcile.Request{}
		}

		for _, cr := range crs {
			cs := &cloudv1.ClusterList{}
			if err = r.Client.List(context.TODO(), cs, &client.ListOptions{
				FieldSelector: fields.OneTermEqualSelector(ConfigRevisionRefField, cr.Name),
			}); err != nil {
				continue
			}
			relatedClusters.Items = append(relatedClusters.Items, cs.Items...)
		}
	}

	requests := make([]reconcile.Request, len(relatedClusters.Items))
	for i, item := range relatedClusters.Items {
		requests[i] = reconcile.Request{
			NamespacedName: types.NamespacedName{
				Name:      item.GetName(),
				Namespace: item.GetNamespace(),
			},
		}
	}
	return requests
}