/*
Copyright 2023.

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 vpcpeering

import (
	"context"
	"fmt"
	"time"

	"github.com/go-logr/logr"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/trace"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/client-go/tools/record"

	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/controller"
	"sigs.k8s.io/controller-runtime/pkg/event"
	"sigs.k8s.io/controller-runtime/pkg/log"
	"sigs.k8s.io/controller-runtime/pkg/predicate"

	"github.com/tidbcloud/infra-provider/pkg/cloud/aws"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/deploy"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	metricsctrl "github.com/tidbcloud/infra-provider/pkg/metrics/controller"
	"github.com/tidbcloud/infra-provider/pkg/o11y/tracing"
	tracerutil "github.com/tidbcloud/infra-provider/pkg/tracer"
	"github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/common/constants"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/common/ec2client"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/common/utils"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/controllers/vpcpeering/task"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/resourcemanager/vpcpeering"
)

var (
	tracer = otel.Tracer("VPCPeeringReconciler")
)

type VPCPeeringReconciler struct {
	kubeClient    client.Client
	kubeAPIReader client.Reader
	awsClient     aws.Client
	ec2Client     ec2client.Client
	logger        logr.Logger
	scheme        *runtime.Scheme
	eventRecorder record.EventRecorder

	// vpcpeeringRW is an extra abstraction over kube client
	// only allow caller to interact with vpcpeering CR
	vpcpeeringRW vpcpeering.VPCPeeringReaderWriter
}

func NewVPCPeeringReconciler(client client.Client, apiReader client.Reader, logger logr.Logger, scheme *runtime.Scheme, recorder record.EventRecorder) *VPCPeeringReconciler {
	return &VPCPeeringReconciler{
		kubeClient:    client,
		kubeAPIReader: apiReader,
		logger:        logger,
		scheme:        scheme,
		eventRecorder: recorder,

		vpcpeeringRW: vpcpeering.NewVPCPeeringReaderWriter(client),
	}
}

func (r *VPCPeeringReconciler) WithEC2Client(client ec2client.Client) *VPCPeeringReconciler {
	r.ec2Client = client
	return r
}

func (r *VPCPeeringReconciler) WithAWSClient(client aws.Client) *VPCPeeringReconciler {
	r.awsClient = client
	return r
}

func (r *VPCPeeringReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	logger := r.logger.WithValues("vpcpeering", req.NamespacedName)

	logger.Info("reconciling regional-server vpcpeering")
	start := time.Now()

	defer func() {
		metricsctrl.ReconcileDuration.
			WithLabelValues("vpcpeering").Observe(time.Since(start).Seconds())
		logger.Info("finish reconcile")
	}()

	peering := &v1alpha1.VPCPeering{}
	if err := r.vpcpeeringRW.Get(ctx, req.NamespacedName, peering); err != nil {
		return ctrl.Result{}, client.IgnoreNotFound(err)
	}
	// do nothing if the cr is paused
	if utils.IsPaused(peering.GetObjectMeta()) {
		logger.Info("reconciliation is paused either through the pause annotation",
			"annotation", v1alpha1.AnnotationKeyReconciliationPaused,
			"vpcpeering", req.NamespacedName.String(),
		)
		r.eventRecorder.Event(peering, "ReconciliationPaused", "reconciliation is paused through the pause annotation", v1alpha1.AnnotationKeyReconciliationPaused)
		return ctrl.Result{}, nil
	}

	providerType := peering.Spec.Provider
	// provider not support, do nothing and return without error
	switch providerType {
	case consts.ProviderTypeEKS, consts.ProviderTypeGKE, string(constants.AWS), string(constants.GCP):
	// do nothing
	default:
		logger.Info("reconciliation is skipped because provider is not supported",
			"provider", providerType,
			"vpcpeering", req.NamespacedName.String(),
		)
		return ctrl.Result{}, nil
	}

	c, err := (&task.Context{
		Local:        r.kubeClient,
		EC2Client:    r.ec2Client,
		ProviderType: providerType,
		VPCPeering:   peering,
	}).Default()
	if err != nil {
		logger.Error(err, "new task context failed",
			"vpcpeering", client.ObjectKeyFromObject(peering).String(),
		)
		return ctrl.Result{}, fmt.Errorf("new task context failed: %v", err)
	}
	// set ec2client with role arn resumed
	switch providerType {
	case consts.ProviderTypeEKS, string(constants.AWS):
		c.EC2Client = r.awsClient.EC2(&aws.AssumedRole{
			RoleARN: "arn:aws:iam::" + peering.Spec.AccountRef.AccountID + ":role/" + peering.Spec.AccountRef.Role,
		})
	}

	tasks := BuildVPCPeeringTasks(c)
	if peering.DeletionTimestamp.IsZero() && isNotReadyVPCPeering(peering) {
		tasks = r.setHookOnTasks(tasks, c)

		// TODO: refactor tracing
		var span trace.Span
		ctx, span = tracer.Start(tracing.ExtractTraceContext(ctx, peering), "VPCPeering Reconcile")
		span.SetAttributes(
			attribute.String("vpcpeering", peering.Name),
			attribute.String("namespace", peering.Namespace),
			attribute.String("provider", peering.Spec.Provider),
			attribute.String("region", peering.Spec.Region),
			attribute.String("sourcevpc", peering.Spec.SourceVPCRef.Name),
		)
		defer func() { tracerutil.EndSpan(span, err) }()
	}

	d := deploy.NewDeployer(tasks, nil, logger)
	var res ctrl.Result
	if peering.DeletionTimestamp.IsZero() {
		res, err = updateCreate(ctx, c, d.Deploy(ctx))
	} else {
		peering.Status.Phase = v1alpha1.VPCPeeringPhaseDeleting
		if err := r.vpcpeeringRW.UpdateStatus(ctx, peering); err != nil {
			return ctrl.Result{}, fmt.Errorf("failed to update peering status: %v", err)
		}
		res, err = updateDelete(ctx, c, d.Destroy(ctx))
	}
	if err == nil {
		r.eventRecorder.Event(peering, corev1.EventTypeNormal, "ReconcileSuccess", "")
	} else {
		r.eventRecorder.Event(peering, corev1.EventTypeWarning, "ReconcileFail", err.Error())
	}

	return res, err
}

func isNotReadyVPCPeering(peering *v1alpha1.VPCPeering) bool {
	return peering.Status.Phase != v1alpha1.VPCPeeringPhaseActive &&
		peering.Status.Phase != v1alpha1.VPCPeeringPhaseDeleting
}

func (r *VPCPeeringReconciler) setHookOnTasks(tasks []deploy.Task, c *task.Context) []deploy.Task {
	hook := func(ctc context.Context, task deploy.Task) {
		name := task.Name()
		passTimeKey := utils.GenTaskFirstPastAnnotationKey(name)
		_, ok := c.VPCPeering.Annotations[passTimeKey]
		if ok {
			return
		}
		if c.VPCPeering.Annotations == nil {
			c.VPCPeering.Annotations = map[string]string{}
		}
		age := time.Since(c.VPCPeering.CreationTimestamp.Time)
		c.VPCPeering.Annotations[passTimeKey] = age.String()
		log.FromContext(ctc).Info("pass task when creating vpcpeering", "task", name, "age", age.String())
	}

	for idx, task := range tasks {
		tasks[idx] = deploy.NewTaskWraper(task, hook, tracer)
	}
	return tasks
}

type AnnotationChangedPredicate struct {
	predicate.Funcs
	includes []string
}

// Update implements default UpdateEvent filter for validating annotation change.
func (a AnnotationChangedPredicate) Update(e event.UpdateEvent) bool {
	if e.ObjectOld == nil {
		// Update event has no old object to update
		return false
	}
	if e.ObjectNew == nil {
		// Update event has no new object for update
		return false
	}
	if len(a.includes) == 0 {
		return false
	}
	extract := func(an map[string]string, keys []string) map[string]string {
		r := make(map[string]string, len(an))
		for _, key := range keys {
			if v := an[key]; v != "" {
				r[key] = v
			}
		}
		return r
	}

	na := extract(e.ObjectNew.GetAnnotations(), a.includes)
	oa := extract(e.ObjectOld.GetAnnotations(), a.includes)
	// Below is the same as controller-runtime's AnnotationChangedPredicate
	// implementation but optimized to avoid using reflect.DeepEqual.
	if len(na) != len(oa) {
		// annotation length changed
		return true
	}

	for k, v := range na {
		if oa[k] != v {
			// annotation value changed
			return true
		}
	}

	// annotations unchanged.
	return false
}

func DesiredStateChanged() predicate.Predicate {
	return predicate.Or(
		AnnotationChangedPredicate{
			includes: []string{
				v1alpha1.AnnotationKeyReconciliationPaused,
				v1alpha1.AnnotationKeySourceAccountID,
			},
		},
		predicate.GenerationChangedPredicate{},
	)
}

func (r *VPCPeeringReconciler) SetupWithManager(mgr ctrl.Manager, pds ...predicate.Predicate) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(&v1alpha1.VPCPeering{}, builder.WithPredicates(pds...)).
		WithEventFilter(DesiredStateChanged()).
		WithOptions(controller.Options{RateLimiter: k8s.RetryLimiter, MaxConcurrentReconciles: 5}).
		Complete(r)
}