/*
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 serverless

import (
	"context"
	"fmt"
	"strings"
	"time"

	vmv1beta1 "github.com/VictoriaMetrics/operator/api/operator/v1beta1"
	"github.com/go-logr/logr"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/client-go/tools/record"
	"k8s.io/client-go/util/workqueue"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller"
	"sigs.k8s.io/controller-runtime/pkg/handler"
	"sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/predicate"
	"sigs.k8s.io/controller-runtime/pkg/reconcile"
	k8ssource "sigs.k8s.io/controller-runtime/pkg/source"

	observabilitypingcapcomv1 "github.com/tidbcloud/observability/api/regionalserver/resourceworker/v1"
	"github.com/tidbcloud/observability/pkg/common/constants"
	"github.com/tidbcloud/observability/pkg/common/deploy"
	"github.com/tidbcloud/observability/pkg/common/structs"
	"github.com/tidbcloud/observability/pkg/mysql/source"
	"github.com/tidbcloud/observability/pkg/resourceworker/controllers/serverless/task"
)

const (
	defaultMaxConcurrency = 5
)

// ServerlessClusterReconciler reconciles a Cluster object
type ServerlessClusterReconciler struct {
	client.Client
	// Local is used to operate with rds clusters
	RdsClient client.Client
	Scheme    *runtime.Scheme
	Logger    logr.Logger

	EventRecorder record.EventRecorder
	Config        structs.ServiceInfo
}

type typedServerlessReconciler struct {
	inner *ServerlessClusterReconciler
}

func (t typedServerlessReconciler) Reconcile(ctx context.Context, obj any) (ctrl.Result, error) {
	req, ok := obj.(reconcile.Request)
	if !ok {
		return ctrl.Result{}, nil
	}
	return t.inner.Reconcile(ctx, ctrl.Request{NamespacedName: req.NamespacedName})
}

func (r *ServerlessClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	logger := log.FromContext(ctx)
	var cluster observabilitypingcapcomv1.Cluster
	logger.Info("Reconciling ServerLess Cluster")

	if err := r.RdsClient.Get(ctx, req.NamespacedName, &cluster); err != nil {
		if apierrors.IsNotFound(err) {
			return ctrl.Result{}, nil
		}
		return ctrl.Result{}, fmt.Errorf("unable to get cluster %s: %v", req.NamespacedName, err)
	}

	c, err := (&task.Context{
		Local:   r.RdsClient,
		Remote:  r.Client,
		Config:  r.Config,
		Cluster: &cluster,
		Logger:  r.Logger,
	}).Default()
	if err != nil {
		return ctrl.Result{}, fmt.Errorf("unable to get task context: %v", err)
	}

	// in regional server , we do not need set remote client for serverless cluster

	var tasks []deploy.Task = GetO11yTasks(r, c)

	//statistics of the first creation time by hook
	if cluster.DeletionTimestamp.IsZero() &&
		(cluster.Status.Phase == "" || cluster.Status.Phase == constants.CREATING) {
		hook := func(ctx context.Context, task deploy.Task) {

			name := task.Name()
			key := "first-pass-when-creating.cloud.pingcap.com/" + name
			_, ok := c.Cluster.Annotations[key]
			if ok {
				return
			}

			if c.Cluster.Annotations == nil {
				c.Cluster.Annotations = make(map[string]string)
			}
			age := time.Since(c.Cluster.CreationTimestamp.Time)
			c.Cluster.Annotations[key] = age.String()
			log.FromContext(ctx).Info("pass task when creating serverless cluster", "task", name, "sinceCreate", age)
			// metrics.NetworkFirstPassStageSeconds.WithLabelValues(name).Observe(age.Seconds())
		}
		for i, task := range tasks {
			tasks[i] = deploy.NewTaskWraper(task, hook)
		}
	}

	d := deploy.NewDeployer(tasks, nil, logger)

	var res ctrl.Result
	if cluster.DeletionTimestamp.IsZero() {
		res, err = updateCreate(ctx, c, d.Deploy(ctx))
	} else {
		res, err = updateDelete(ctx, c, d.Destroy(ctx))
	}

	if err == nil {
		logger.Info("ReconcileSuccess")
	} else {
		logger.Error(err, "ReconcileFail")
	}

	return res, err
}

// SetupWithManager sets up the controller with the Manager.
func (r *ServerlessClusterReconciler) SetupWithManager(mgr ctrl.Manager, pds ...predicate.Predicate) error {
	logConstructor := func(in *any) logr.Logger {
		if in != nil {
			if req, ok := (*in).(reconcile.Request); ok {
				return r.Logger.WithValues(
					"namespace", req.Namespace, "name", req.Name,
				)
			}
		}
		return r.Logger
	}

	maxConcurrency := max(r.Config.ControllerConfig.ServerlessConfig.MaxConcurrency, defaultMaxConcurrency)

	// returns a typed Controller of type any, in order to enqueue MysqlItem rather than reconcile.Request in mysql.Consumer.
	c, err := controller.NewTyped("ServerlessCluster", mgr, controller.TypedOptions[any]{
		NewQueue: func(
			controllerName string,
			rateLimiter workqueue.TypedRateLimiter[any],
		) workqueue.TypedRateLimitingInterface[any] {
			return source.NewMysqlRateLimitingQueueWithConfig(
				rateLimiter,
				workqueue.TypedRateLimitingQueueConfig[any]{Name: controllerName},
				1*time.Minute,
			)
		},
		RateLimiter:             source.ControllerRateLimiter(),
		MaxConcurrentReconciles: maxConcurrency,
		Reconciler:              typedServerlessReconciler{inner: r},
		LogConstructor:          logConstructor,
	})
	if err != nil {
		return fmt.Errorf("can't new a serverless cluster controller: %w", err)
	}

	if err := c.Watch(&source.MysqlSource{
		Logger:  c.GetLogger(),
		ConnCfg: r.Config.ControllerConfig.ServerlessConfig.RdsConfig.ToGormutilConfig(),
	}); err != nil {
		return err
	}

	return c.Watch(k8ssource.TypedKind[client.Object, any](mgr.GetCache(), &vmv1beta1.VMRule{},
		handler.TypedEnqueueRequestsFromMapFunc(r.findClustersForVMRule),
		predicate.NewPredicateFuncs(func(obj client.Object) bool {
			labels := obj.GetLabels()
			// only watch the event of rules labeled with "external-overlay=true"
			return labels != nil && labels[constants.ExternalLabelOverlay] == "true"
		}), predicate.GenerationChangedPredicate{}))
}

// findClustersForVMRule returns reconcile requests for clusters whose template ID label matches vmRule.Name.
func (r *ServerlessClusterReconciler) findClustersForVMRule(ctx context.Context, obj client.Object) []any {
	vmRule, ok := obj.(*vmv1beta1.VMRule)
	if !ok {
		return nil
	}

	var reqs []any
	if strings.HasPrefix(vmRule.Name, "overlay-") {
		// enqueue the cluster for overlay rule
		clusterName := strings.TrimPrefix(vmRule.Name, "overlay-")
		reqs = append(reqs, reconcile.Request{
			NamespacedName: types.NamespacedName{
				Name:      clusterName,
				Namespace: vmRule.Namespace,
			},
		})
	} else {
		// enqueue clusters using the rule template.
		var clusters observabilitypingcapcomv1.ClusterList
		err := r.RdsClient.List(ctx, &clusters, client.MatchingLabels{
			constants.LabelTemplateID: vmRule.Name,
		})

		if err != nil {
			r.Logger.Error(err, "failed to list clusters for vmrule", "vmrule", vmRule.Name)
			// Record metric and event for alerting - update event may be lost
			VMRuleMapErrorCounter.WithLabelValues(vmRule.Name, "list_clusters").Inc()
			r.EventRecorder.Event(vmRule, "Warning", "ClusterListFailed",
				fmt.Sprintf("Failed to list clusters for rule template %s, update event may be lost: %v", vmRule.Name, err))
			return nil
		}
		for _, cluster := range clusters.Items {
			reqs = append(reqs, reconcile.Request{
				NamespacedName: types.NamespacedName{
					Name:      cluster.Name,
					Namespace: cluster.Namespace,
				},
			})
		}
	}

	return reqs
}