package o11yk8sagent

import (
	"context"
	"fmt"

	vmoperatorv1beta1 "github.com/VictoriaMetrics/operator/api/operator/v1beta1"
	eksv1alpha1 "github.com/crossplane-contrib/provider-aws/apis/eks/manualv1alpha1"
	gkecontainerv1beta1 "github.com/crossplane-contrib/provider-gcp/apis/container/v1beta1"
	"github.com/go-logr/logr"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/utils/ptr"
	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/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/reconcile"

	agentv2 "github.com/tidbcloud/observability/agent-control-server/api/v2"
	"github.com/tidbcloud/observability/agent-control-server/controllers/o11yk8sagent/metadata"
	"github.com/tidbcloud/observability/agent-control-server/controllers/o11yk8sagent/task"
	taskaws "github.com/tidbcloud/observability/agent-control-server/controllers/o11yk8sagent/task/aws"
	taskgcp "github.com/tidbcloud/observability/agent-control-server/controllers/o11yk8sagent/task/gcp"
	"github.com/tidbcloud/observability/agent-control-server/pkg/commontasks"
	commonaws "github.com/tidbcloud/observability/agent-control-server/pkg/commontasks/aws"
	commonazure "github.com/tidbcloud/observability/agent-control-server/pkg/commontasks/azure"
	commongcp "github.com/tidbcloud/observability/agent-control-server/pkg/commontasks/gcp"
	"github.com/tidbcloud/observability/agent-control-server/pkg/consts"
	"github.com/tidbcloud/observability/agent-control-server/pkg/deploy"
	"github.com/tidbcloud/observability/agent-control-server/pkg/k8sclient/k8scliv1"
	"github.com/tidbcloud/observability/agent-control-server/pkg/loader"
	"github.com/tidbcloud/observability/agent-control-server/pkg/metrics"
	"github.com/tidbcloud/observability/agent-control-server/pkg/reconciler"
)

// AgentReconciler reconciles a O11yK8sAgent object
type AgentReconciler struct {
	reconciler.Reconciler
}

// +kubebuilder:rbac:groups=agent.observability.pingcap.com,resources=o11yk8sagents,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=agent.observability.pingcap.com,resources=o11yk8sagents/status,verbs=get;update;patch
// +kubebuilder:rbac:groups=agent.observability.pingcap.com,resources=o11yk8sagents/finalizers,verbs=update

// Reconcile is part of the main kubernetes reconciliation loop which aims to
// move the current state of the cluster closer to the desired state.
// the O11yK8sAgent object against the actual cluster state, and then
// perform operations to make the cluster state reflect the state specified by
// the user.
//
// For more details, check Reconcile and its Result here:
// - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.21.0/pkg/reconcile
func (r *AgentReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	return metrics.WrapReconciledFunc(consts.AgentO11yK8s, ctx, req, r.reconcile)
}

func (r *AgentReconciler) reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	l := ctrl.Log.WithValues("agent", consts.AgentO11yK8s, "namespace", req.Namespace, "name", req.Name)

	// Fetch the O11yK8sAgent instance.
	var agent agentv2.O11yK8sAgent
	if err := r.Get(ctx, req.NamespacedName, &agent); err != nil {
		if apierrors.IsNotFound(err) {
			return ctrl.Result{}, nil
		}
		return ctrl.Result{}, fmt.Errorf("unable to get o11yk8sagent %s: %v", req.NamespacedName, err)
	}

	// Check if the agent is paused or in migration state.
	if agent.GetAnnotations()[consts.AnnotationKeyPauseReconcile4Migration] == "true" {
		return ctrl.Result{}, nil
	}
	if agent.GetAnnotations()[consts.AnnotationKeyPauseReconcile] == "true" && agent.DeletionTimestamp.IsZero() {
		return ctrl.Result{}, nil
	}

	// Ensure the tenant label is set on the agent (backfill for agents created
	// before the label was required by PrivateLinkTask).
	if agent.Labels[agentv2.TagKeyTenant] != agent.Spec.O11y.TenantID {
		if agent.Labels == nil {
			agent.Labels = make(map[string]string)
		}
		agent.Labels[agentv2.TagKeyTenant] = agent.Spec.O11y.TenantID
		if err := r.Update(ctx, &agent); err != nil {
			return ctrl.Result{}, fmt.Errorf("unable to set tenant label on o11yk8sagent %s: %v", req.NamespacedName, err)
		}
	}

	// Prepare local and remote k8s clients.
	clientGetter, err := k8scliv1.NetworkClientGetter(
		r.Client,
		loader.RemoteScheme(vmoperatorv1beta1.AddToScheme),
		consts.VendorToProvider(r.Config.O11yConfig.Vendor),
		r.Config.O11yConfig.Region,
		agent.Spec.KubernetesClusterName)
	if err != nil {
		return ctrl.Result{}, fmt.Errorf("unable to get k8s network client getter: %v", err)
	}

	// Core task context for the reconciler.
	c := &reconciler.Context[*agentv2.O11yK8sAgent]{
		Agent:     &agent,
		L:         l,
		Local:     r.Client,
		Remote:    k8scliv1.NewCacheClientGetter(clientGetter),
		Config:    r.Config,
		Templates: r.Templates,
		AWS:       r.AWS,
		GCP:       r.GCP,
		O11yAPI:   r.O11yAPI,
	}

	// Pre-check remote k8s cluster availability.
	// If the remote k8s CR (e.g., EKS/GKE Cluster) is not found, skip remote-dependent tasks.
	remoteAvailable := true
	if _, err := c.Remote.GetClient(ctx); err != nil {
		if apierrors.IsNotFound(err) {
			l.Info("remote k8s cluster CR not found, skipping remote-dependent tasks",
				"clusterName", agent.Spec.KubernetesClusterName)
			remoteAvailable = false
		} else {
			return ctrl.Result{}, fmt.Errorf("unable to get remote k8s client: %v", err)
		}
	}

	// Prepare the tasks based on the agent's status.
	var tasks []deploy.Task
	if agent.Status.Phase != agentv2.O11yAgentPhaseHibernated {
		tasks = r.buildTasks(c, remoteAvailable)
	} else {
		tasks = r.buildHibernatedTasks(c)
	}

	// Run tasks and update status.
	var res ctrl.Result
	d := deploy.NewDeployer(tasks, nil, l)
	if agent.DeletionTimestamp.IsZero() {
		if agent.Spec.Pause {
			res, err = updateStatusToHibernated(ctx, l, c, d.Deploy(ctx))
		} else {
			res, err = updateStatusToNormal(ctx, l, c, d.Deploy(ctx))
		}
	} else {
		res, err = updateStatusToDeleting(ctx, l, c, d.Destroy(ctx))
	}
	if err == nil {
		r.EventRecorder.Event(&agent, corev1.EventTypeNormal, "ReconcileSuccess", "")
	} else {
		r.EventRecorder.Event(&agent, corev1.EventTypeWarning, "ReconcileFail", err.Error())
	}
	return res, err
}

// SetupWithManager sets up the controller with the Manager.
func (r *AgentReconciler) SetupWithManager(mgr ctrl.Manager) error {
	return ctrl.NewControllerManagedBy(mgr).
		For(&agentv2.O11yK8sAgent{}).
		Named("o11yk8sagent").
		WithOptions(controller.Options{
			MaxConcurrentReconciles: r.MaxConcurrentReconciles,
			RateLimiter:             reconciler.DefaultControllerRateLimiter[reconcile.Request](),
		}).
		Complete(r)
}

func (r *AgentReconciler) buildTasks(c *reconciler.Context[*agentv2.O11yK8sAgent], remoteAvailable bool) []deploy.Task {
	clusterMeta := metadata.ClusterMeta(c.Agent)
	tasks := []deploy.Task{
		commontasks.NewGuardianTask(c.Agent, clusterMeta),
	}

	// For Azure, fetch kubeletIdentity before backend task to set annotations for vector task.
	if r.Config.O11yConfig.Vendor == consts.VendorTypeAzure {
		tasks = append(tasks, commonazure.NewKubeletIdentityTask(c, clusterMeta))
	}

	tasks = append(tasks,
		task.NewBackendTask(c),
		task.NewAutoResourcingTask(c),
	)

	o11yCfg := c.Agent.Spec.O11y
	specs := c.Config.O11yConfig.CustomerConfigsV2.Dedicated.Specs.K8s
	clusterName := c.Agent.Spec.KubernetesClusterName
	nodeNameFunc := func(specName string) string {
		base := fmt.Sprintf("tidbcloud-o11y-k8s-%s", o11yCfg.ProjectID)
		if r.Config.O11yConfig.Vendor == consts.VendorTypeGCP {
			return commongcp.DedicatedNodePoolName(base, specName)
		}
		if specName != "default" {
			base = fmt.Sprintf("%s-%s", base, specName)
		}
		if r.Config.O11yConfig.Vendor == consts.VendorTypeAWS {
			annotations := c.Agent.GetAnnotations()
			base += commonaws.NameSuffix(
				annotations[consts.AnnotationKeyO11yNodeAMIType],
				annotations[consts.AnnotationKeyO11yNodeAMISuffixRequired],
				annotations[consts.AnnotationKeyO11yNodeVersion],
			)
		}
		return base
	}
	nodeMatchLabels := map[string]string{
		"o11y.tidbcloud.com/component": "o11y",
		"o11y.tidbcloud.com/project":   o11yCfg.ProjectID,
	}

	switch r.Config.O11yConfig.Vendor {
	case consts.VendorTypeAWS:
		if remoteAvailable {
			tasks = append(tasks, taskaws.NewStorageClassTask(c))
		}
		tasks = append(tasks,
			taskaws.NewPrivateLinkTask(c),
			commonaws.NewNodeGroupTask(c, clusterName, nodeNameFunc, nodeMatchLabels,
				map[string]string{
					consts.TagKeyTenant:  o11yCfg.TenantID,
					consts.TagKeyProject: o11yCfg.ProjectID,
				},
				specs,
				[]eksv1alpha1.Taint{
					{Effect: "NO_EXECUTE", Key: ptr.To("o11y-dedicated"), Value: ptr.To("o11y")},
					{Effect: "NO_EXECUTE", Key: ptr.To("o11y-project"), Value: &o11yCfg.ProjectID},
				},
			),
		)

	case consts.VendorTypeGCP:
		if remoteAvailable {
			tasks = append(tasks, taskgcp.NewStorageClassTask(c))
		}
		tasks = append(tasks,
			taskgcp.NewPrivateLinkTask(c),
			commongcp.NewNodePoolTask(c, clusterName, nodeNameFunc, nodeMatchLabels,
				map[string]string{
					"servicetype":                  "dedicated",
					"usedby":                       "o11y",
					"environment":                  c.Config.O11yConfig.Env,
					"tenant":                       o11yCfg.TenantID,
					"project":                      o11yCfg.ProjectID,
					"o11y.tidbcloud.com/component": "o11y",
					"o11y.tidbcloud.com/project":   o11yCfg.ProjectID,
				},
				specs,
				[]*gkecontainerv1beta1.NodeTaint{
					{Effect: "NO_EXECUTE", Key: "o11y-dedicated", Value: "o11y"},
					{Effect: "NO_EXECUTE", Key: "o11y-project", Value: o11yCfg.ProjectID},
				},
			),
		)
	case consts.VendorTypeAzure:
		if remoteAvailable {
			tasks = append(tasks, commonazure.NewStorageClassTask(c, commonazure.StorageClassConfig{
				Tags: func() map[string]string {
					return map[string]string{
						consts.TagKeyBillingServiceType: "o11y",
						consts.TagKeyBillingTenant:      o11yCfg.TenantID,
						consts.TagKeyBillingProject:     o11yCfg.ProjectID,
						consts.TagKeyBillingUsedBy:      "o11y-agent",
					}
				},
			}))
		}
		tasks = append(tasks,
			commonazure.NewPrivateLinkTask(c, commonazure.PrivateLinkConfig{
				AKS:                   func() *agentv2.AKSConfig { return c.Agent.Spec.AKS },
				KubernetesClusterName: func() string { return c.Agent.Spec.KubernetesClusterName },
				ObjectMeta:            func() metav1.ObjectMeta { return c.Agent.ObjectMeta },
			}),
			commonazure.NewNodePoolTask(c, clusterName, nodeNameFunc, nodeMatchLabels,
				map[string]string{
					"servicetype":                  "dedicated",
					"usedby":                       "o11y",
					"environment":                  c.Config.O11yConfig.Env,
					"tenant":                       o11yCfg.TenantID,
					"project":                      o11yCfg.ProjectID,
					"o11y.tidbcloud.com/component": "o11y",
					"o11y.tidbcloud.com/project":   o11yCfg.ProjectID,
				},
				specs,
				[]corev1.Taint{
					{Effect: corev1.TaintEffectNoExecute, Key: "o11y-dedicated", Value: "o11y"},
					{Effect: corev1.TaintEffectNoExecute, Key: "o11y-project", Value: o11yCfg.ProjectID},
				},
				func() (string, string, string, string, error) {
					if c.Agent.Spec.AKS == nil {
						return "", "", "", "", fmt.Errorf("AKS configuration is not set")
					}
					if c.Agent.Spec.AKS.SubscriptionID == "" || c.Agent.Spec.AKS.ResourceGroup == "" || c.Agent.Spec.AKS.SubnetID == "" || c.Agent.Spec.AKS.PodSubnetID == "" {
						return "", "", "", "", fmt.Errorf("AKS configuration is incomplete")
					}
					return c.Agent.Spec.AKS.SubscriptionID, c.Agent.Spec.AKS.ResourceGroup, c.Agent.Spec.AKS.SubnetID, c.Agent.Spec.AKS.PodSubnetID, nil
				},
				func() deploy.ClusterMeta {
					return clusterMeta
				},
			),
		)
	}

	if remoteAvailable {
		tasks = append(tasks,
			commontasks.NewHelmRepoTask(c, clusterMeta, r.Config.O11yConfig.CustomerConfigsV2.Dedicated.Repos),
			task.NewOperatorTask(c),
			commontasks.NewPriorityClassTask(c, clusterMeta),
			task.NewVMAgentTask(c),
			task.NewVectorTask(c),
			commontasks.NewVMAgentRestartTask(c, clusterMeta, consts.AgentO11yK8s, consts.O11ySystemNamespace),
			commontasks.NewDrainNodeTask(c, clusterMeta),
		)
	}
	tasks = append(tasks,
		commontasks.NewUpdateVersionTask(c.Agent, c.Config.O11yConfig.Version, clusterMeta),
	)
	return tasks
}

func (r *AgentReconciler) buildHibernatedTasks(c *reconciler.Context[*agentv2.O11yK8sAgent]) []deploy.Task {
	return []deploy.Task{
		commontasks.NewGuardianTask(c.Agent, metadata.ClusterMeta(c.Agent)),
	}
}

func updateStatus(ctx context.Context, cli client.Client, agent *agentv2.O11yK8sAgent) error {
	status := agent.Status.DeepCopy()
	if err := cli.Update(ctx, agent); err != nil {
		return err
	}
	agent.Status = *status
	return cli.Status().Update(ctx, agent)
}

func updateStatusToNormal(ctx context.Context, l logr.Logger, c *reconciler.Context[*agentv2.O11yK8sAgent], results []deploy.Result) (ctrl.Result, error) {
	var retriableErr error
	var res ctrl.Result
	var err error

	res, err, retriableErr = deploy.MinRequeue(results)
	if deploy.HasPanic(results) {
		c.Agent.Status.Phase = agentv2.O11yAgentPhasePanic
	} else if c.Agent.Status.Phase == agentv2.O11yAgentPhasePanic {
		// Clear Panic on the first reconcile without Panic, even if Error remains.
		c.Agent.Status.Phase = agentv2.O11yAgentPhaseNormal
	} else if err == nil && retriableErr == nil {
		// Hibernated status will update to WakingUp status when create a cluster in the network
		// Hibernated network status can not update to normal
		if c.Agent.Status.Phase != agentv2.O11yAgentPhaseNormal && c.Agent.Status.Phase != agentv2.O11yAgentPhaseHibernated {
			c.Agent.Status.Phase = agentv2.O11yAgentPhaseNormal
		}
	}
	if err != nil {
		l.Error(err, "failed in MinRequeue for o11y cluster")
	}
	if retriableErr != nil {
		l.Error(retriableErr, "update status to normal failed")
	}

	if updateErr := updateStatus(ctx, c.Local, c.Agent); updateErr != nil {
		return res, updateErr
	}
	return res, err
}

func updateStatusToHibernated(ctx context.Context, l logr.Logger, c *reconciler.Context[*agentv2.O11yK8sAgent], results []deploy.Result) (ctrl.Result, error) {
	var retriableErr error
	var res ctrl.Result
	var err error

	res, err, retriableErr = deploy.MinRequeue(results)
	if err == nil && retriableErr == nil {
		c.Agent.Status.Phase = agentv2.O11yAgentPhaseHibernated
	}
	if err != nil {
		l.Error(err, "failed in MinRequeue for o11y cluster")
	}
	if retriableErr != nil {
		l.Error(retriableErr, "update status to hibernated failed")
	}

	if updateErr := updateStatus(ctx, c.Local, c.Agent); updateErr != nil {
		return res, updateErr
	}
	return res, err
}

func updateStatusToDeleting(ctx context.Context, l logr.Logger, c *reconciler.Context[*agentv2.O11yK8sAgent], results []deploy.Result) (ctrl.Result, error) {
	var deleted bool
	var res ctrl.Result
	var err error

	// If the o11y cluster is pending delete, then the deletion tasks are not actually performed
	if c.Agent.Status.Phase != agentv2.O11yAgentPhasePendingDelete {
		var retriableErr error
		res, err, retriableErr = deploy.MinRequeue(results)
		if err == nil && retriableErr == nil {
			deleted = true
		}
		c.Agent.Status.Phase = agentv2.O11yAgentPhaseDeleting
		if err != nil {
			l.Error(err, "failed in MinRequeue for o11y cluster")
		}
		if retriableErr != nil {
			l.Error(retriableErr, "update status to deleting failed")
		}
	}
	if deleted {
		controllerutil.RemoveFinalizer(c.Agent, consts.AnnotationKeyO11yAgentFinalizer)
		l.Info("remove finalizer when agent deleted")
	}

	if updateErr := updateStatus(ctx, c.Local, c.Agent); updateErr != nil {
		return res, updateErr
	}
	return res, err
}