package o11ytidbagent

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"
	"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"
	logf "sigs.k8s.io/controller-runtime/pkg/log"
	"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/o11ytidbagent/metadata"
	"github.com/tidbcloud/observability/agent-control-server/controllers/o11ytidbagent/task"
	taskaws "github.com/tidbcloud/observability/agent-control-server/controllers/o11ytidbagent/task/aws"
	taskgcp "github.com/tidbcloud/observability/agent-control-server/controllers/o11ytidbagent/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 O11yTiDBAgent 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 O11yTiDBAgent 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.AgentO11yTidb, ctx, req, r.reconcile)
}

func (r *AgentReconciler) reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
	l := logf.FromContext(ctx)

	// Fetch the O11yTiDBAgent instance.
	var agent agentv2.O11yTiDBAgent
	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 o11ytidbagent %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
	}

	// 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.O11yTiDBAgent]{
		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.Spec.Pause && agent.DeletionTimestamp.IsZero() {
		// During pause, use an explicit hibernation deploy task list.
		tasks = r.buildPauseDeployTasks(c, remoteAvailable)
	} else 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
	if agent.DeletionTimestamp.IsZero() {
		if agent.Spec.Pause {
			// Pause semantics:
			// 1) scale VMAgent/Vector to 0 via deploy path,
			// 2) then clean up compute pools (NodeGroup/NodePool) via destroy path.
			results := deploy.NewDeployer(tasks, nil, l).Deploy(ctx)

			deleteTasks := r.buildPauseCleanupTasks(c)
			if len(deleteTasks) > 0 {
				deleteResults := deploy.NewDeployer(deleteTasks, nil, l).Destroy(ctx)
				results = append(results, deleteResults...)
			}

			res, err = updateStatusToHibernated(ctx, l, c, results)
		} else {
			d := deploy.NewDeployer(tasks, nil, l)
			res, err = updateStatusToNormal(ctx, l, c, d.Deploy(ctx))
		}
	} else {
		d := deploy.NewDeployer(tasks, nil, l)
		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.O11yTiDBAgent{}).
		Named("o11ytidbagent").
		WithOptions(controller.Options{
			MaxConcurrentReconciles: r.MaxConcurrentReconciles,
			RateLimiter:             reconciler.DefaultControllerRateLimiter[reconcile.Request](),
		}).
		Complete(r)
}

func (r *AgentReconciler) buildTasks(c *reconciler.Context[*agentv2.O11yTiDBAgent], 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.TiDBCluster
	clusterName := c.Agent.Spec.KubernetesClusterName
	nodeNameFunc := func(specName string) string {
		base := fmt.Sprintf("tidbcloud-o11y-tidb-%s", o11yCfg.ApplicationID)
		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/cluster":   o11yCfg.ApplicationID,
	}

	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-cluster"), Value: &o11yCfg.ApplicationID},
				},
			),
		)

	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/cluster":   o11yCfg.ApplicationID,
				},
				specs,
				[]*gkecontainerv1beta1.NodeTaint{
					{Effect: "NO_EXECUTE", Key: "o11y-dedicated", Value: "o11y"},
					{Effect: "NO_EXECUTE", Key: "o11y-cluster", Value: o11yCfg.ApplicationID},
				},
			),
		)
	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.NewPrivateLinkIPTask(c, c.Agent.Spec.KubernetesClusterName),
			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/cluster":   o11yCfg.ApplicationID,
				},
				specs,
				[]corev1.Taint{
					{Effect: corev1.TaintEffectNoExecute, Key: "o11y-dedicated", Value: "o11y"},
					{Effect: corev1.TaintEffectNoExecute, Key: "o11y-cluster", Value: o11yCfg.ApplicationID},
				},
				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.NewPriorityClassTask(c, clusterMeta),
			task.NewVMAgentTask(c),
			task.NewVectorTask(c),
			commontasks.NewVectorVMAgentRestartTask(c, clusterMeta, consts.AgentO11yTidb, fmt.Sprintf("tidb%s", c.Agent.Spec.O11y.ApplicationID)),
			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.O11yTiDBAgent]) []deploy.Task {
	return []deploy.Task{
		commontasks.NewGuardianTask(c.Agent, metadata.ClusterMeta(c.Agent)),
	}
}

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

	if remoteAvailable {
		tasks = append(tasks,
			task.NewVMAgentTask(c),
			task.NewVectorTask(c),
		)
	}

	return tasks
}

func (r *AgentReconciler) buildPauseCleanupTasks(c *reconciler.Context[*agentv2.O11yTiDBAgent]) []deploy.Task {
	o11yCfg := c.Agent.Spec.O11y
	specs := c.Config.O11yConfig.CustomerConfigsV2.Dedicated.Specs.TiDBCluster
	clusterName := c.Agent.Spec.KubernetesClusterName
	nodeNameFunc := func(specName string) string {
		base := fmt.Sprintf("tidbcloud-o11y-tidb-%s", o11yCfg.ApplicationID)
		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/cluster":   o11yCfg.ApplicationID,
	}

	switch r.Config.O11yConfig.Vendor {
	case consts.VendorTypeAWS:
		return []deploy.Task{
			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-cluster"), Value: &o11yCfg.ApplicationID},
				},
			),
		}
	case consts.VendorTypeGCP:
		return []deploy.Task{
			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/cluster":   o11yCfg.ApplicationID,
				},
				specs,
				[]*gkecontainerv1beta1.NodeTaint{
					{Effect: "NO_EXECUTE", Key: "o11y-dedicated", Value: "o11y"},
					{Effect: "NO_EXECUTE", Key: "o11y-cluster", Value: o11yCfg.ApplicationID},
				},
			),
		}
	case consts.VendorTypeAzure:
		clusterMeta := metadata.ClusterMeta(c.Agent)
		return []deploy.Task{
			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/cluster":   o11yCfg.ApplicationID,
				},
				specs,
				[]corev1.Taint{
					{Effect: corev1.TaintEffectNoExecute, Key: "o11y-dedicated", Value: "o11y"},
					{Effect: corev1.TaintEffectNoExecute, Key: "o11y-cluster", Value: o11yCfg.ApplicationID},
				},
				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
				},
			),
		}
	default:
		return nil
	}
}

func updateStatus(ctx context.Context, cli client.Client, agent *agentv2.O11yTiDBAgent) 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.O11yTiDBAgent], 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.O11yTiDBAgent], 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.O11yTiDBAgent], 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
}