package kubernetesagent

import (
	"context"
	"fmt"
	"time"

	vmoperatorv1beta1 "github.com/VictoriaMetrics/operator/api/operator/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/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/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/kubernetesagent/metadata"
	"github.com/tidbcloud/observability/agent-control-server/controllers/kubernetesagent/task"
	taskalicloud "github.com/tidbcloud/observability/agent-control-server/controllers/kubernetesagent/task/alicloud"
	taskaws "github.com/tidbcloud/observability/agent-control-server/controllers/kubernetesagent/task/aws"
	"github.com/tidbcloud/observability/agent-control-server/pkg/commontasks"
	commonazure "github.com/tidbcloud/observability/agent-control-server/pkg/commontasks/azure"
	"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/k8scliv2"
	"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 KubernetesAgent object
type AgentReconciler struct {
	reconciler.Reconciler
}

// +kubebuilder:rbac:groups=agent.observability.pingcap.com,resources=kubernetesagents,verbs=get;list;watch;create;update;patch;delete
// +kubebuilder:rbac:groups=agent.observability.pingcap.com,resources=kubernetesagents/status,verbs=get;update;patch
// +kubebuilder:rbac:groups=agent.observability.pingcap.com,resources=kubernetesagents/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 KubernetesAgent 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.AgentKubernetes, ctx, req, r.reconcile)
}

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

	// Fetch the KubernetesAgent instance.
	var agent agentv2.KubernetesAgent
	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 kubernetesagent %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 := k8scliv2.NetworkClientGetter(
		r.Client,
		loader.RemoteScheme(vmoperatorv1beta1.AddToScheme),
		consts.VendorToProvider(r.Config.O11yConfig.Vendor),
		agent.Spec.KubernetesClusterName,
		r.Config,
	)
	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.KubernetesAgent]{
		Agent:     &agent,
		L:         l,
		Local:     r.Client,
		Remote:    k8scliv2.NewCacheClientGetter(clientGetter),
		Config:    r.Config,
		Templates: r.Templates,
		AWS:       r.AWS,
		GCP:       r.GCP,
		AliCloud:  r.AliCloud,
		O11yAPI:   r.O11yAPI,
	}

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

	// Statistics of the first creation time by hook.
	if agent.DeletionTimestamp.IsZero() && (agent.Status.Phase == "" || agent.Status.Phase == agentv2.O11yAgentPhaseCreating) {
		hook := func(ctx context.Context, task deploy.Task) {
			age := time.Since(c.Agent.CreationTimestamp.Time)
			l.Info("Pass task when creating o11y agent", "task", task.Name(), "sinceCreate", age)
		}
		for i, task := range tasks {
			tasks[i] = deploy.NewTaskWrapper(task, hook)
		}
	}

	// Run tasks and update status.
	var res ctrl.Result
	d := deploy.NewDeployer(tasks, nil, l)
	if agent.DeletionTimestamp.IsZero() {
		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.KubernetesAgent{}).
		Named("kubernetesagent").
		WithOptions(controller.Options{
			MaxConcurrentReconciles: r.MaxConcurrentReconciles,
			RateLimiter:             workqueue.DefaultTypedControllerRateLimiter[reconcile.Request](),
		}).
		Complete(r)
}

func (r *AgentReconciler) buildTasks(c *reconciler.Context[*agentv2.KubernetesAgent]) []deploy.Task {
	clusterMeta := metadata.ClusterMeta(c.Agent)
	account := ""
	if c.Agent.Spec.EKS != nil {
		account = c.Agent.Spec.EKS.AccountID
	} else if c.Agent.Spec.AKS != nil {
		account = c.Agent.Spec.AKS.SubscriptionID
	} else if c.Agent.Spec.ACK != nil {
		account = c.Agent.Spec.ACK.AccountID
	}
	backendConfig := commontasks.ControlPlaneBackendConfig(account)
	backendConfig.RequirePeering = true
	if r.Config.O11yConfig.Vendor == consts.VendorTypeAzure {
		backendConfig.SkipBackendDNSName = true
		backendConfig.WithAgentAccessARN = true
	}

	tasks := []deploy.Task{
		commontasks.NewGuardianTask(c.Agent, clusterMeta),
	}

	// For Azure, fetch kubeletIdentity before creating the backend.
	if r.Config.O11yConfig.Vendor == consts.VendorTypeAzure {
		tasks = append(tasks, commonazure.NewKubeletIdentityTask(c, clusterMeta))
	}

	tasks = append(tasks,
		commontasks.NewBaseBackendTask(c, clusterMeta, c.Agent.Spec.O11y, c.Agent.Spec.O11y.BizType, backendConfig),
	)

	switch r.Config.O11yConfig.Vendor {
	case consts.VendorTypeAWS:
		tasks = append(tasks, taskaws.NewStorageClassTask(c))
		// When peeringNLB is set, the o11y VPC and dataplane VPC are connected via
		// VPC peering, so the NLB is directly reachable and no PrivateLink endpoint
		// is needed.
		if r.Config.O11yConfig.PeeringNLB == "" {
			tasks = append(tasks, taskaws.NewPrivateLinkTask(c))
		}
	case consts.VendorTypeGCP:
	case consts.VendorTypeAzure:
		tasks = append(tasks,
			commonazure.NewStorageClassTask(c, commonazure.StorageClassConfig{
				Tags: func() map[string]string {
					return map[string]string{
						consts.TagKeyBillingServiceType: "o11y",
						consts.TagKeyBillingTenant:      c.Agent.Spec.O11y.TenantID,
						consts.TagKeyBillingUsedBy:      "o11y-agent",
						consts.CrossplaneKeyVendor:      consts.CrossplaneValueVendor,
					}
				},
				ClusterMetaFunc: func() deploy.ClusterMeta { return metadata.ClusterMeta(c.Agent) },
			}),
			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 },
				ClusterMetaFunc:       func() deploy.ClusterMeta { return metadata.ClusterMeta(c.Agent) },
			}))
	case consts.VendorTypeAliCloud:
		tasks = append(tasks,
			taskalicloud.NewStorageClassTask(c),
			taskalicloud.NewPrivateLinkTask(c))
	}

	tasks = append(tasks,
		commontasks.NewPriorityClassTask(c, clusterMeta),
		commontasks.NewHelmRepoTask(c, clusterMeta, r.Config.O11yConfig.CustomerConfigsV2.NextGen.Repos),
		task.NewOperatorTask(c),
		task.NewVMAgentTask(c),
	)

	// In serverless mode, VMPodScrapes are managed by ACS after the operator installs VM CRDs.
	// VMServiceScrape selects the ACS Service so VMAgent scrapes ACS's own metrics.
	if r.Config.IsServerlessMode() {
		tasks = append(tasks,
			task.NewVMPodScrapeTask(c),
			task.NewVMServiceScrapeTask(c),
		)
	}

	tasks = append(tasks,
		task.NewVectorTask(c),
		commontasks.NewVMAgentRestartTask(c, clusterMeta, consts.AgentKubernetes, consts.O11ySystemNamespace),
		commontasks.NewDrainNodeTask(c, clusterMeta),
		commontasks.NewUpdateVersionTask(c.Agent, c.Config.O11yConfig.Version, clusterMeta),
	)

	if c.Agent.Annotations[consts.AnnotationKeySkipDataPlaneDelete] == "true" {
		switch r.Config.O11yConfig.Vendor {
		case consts.VendorTypeAWS:
			awsTasks := []deploy.Task{
				commontasks.NewGuardianTask(c.Agent, clusterMeta),
				commontasks.NewBaseBackendTask(c, clusterMeta, c.Agent.Spec.O11y, c.Agent.Spec.O11y.BizType, backendConfig),
			}
			if r.Config.O11yConfig.PeeringNLB == "" {
				awsTasks = append(awsTasks, taskaws.NewPrivateLinkTask(c))
			}
			return awsTasks
		case consts.VendorTypeAliCloud:
			return []deploy.Task{
				commontasks.NewGuardianTask(c.Agent, clusterMeta),
				commontasks.NewBaseBackendTask(c, clusterMeta, c.Agent.Spec.O11y, c.Agent.Spec.O11y.BizType, backendConfig),
				taskalicloud.NewPrivateLinkTask(c),
			}
		case consts.VendorTypeAzure:
			return []deploy.Task{
				commontasks.NewGuardianTask(c.Agent, clusterMeta),
				commonazure.NewKubeletIdentityTask(c, clusterMeta),
				commontasks.NewBaseBackendTask(c, clusterMeta, c.Agent.Spec.O11y, c.Agent.Spec.O11y.BizType, backendConfig),
				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 },
					ClusterMetaFunc:       func() deploy.ClusterMeta { return metadata.ClusterMeta(c.Agent) },
				}),
			}
		default:
			return []deploy.Task{
				commontasks.NewGuardianTask(c.Agent, clusterMeta),
				commontasks.NewBaseBackendTask(c, clusterMeta, c.Agent.Spec.O11y, c.Agent.Spec.O11y.BizType, backendConfig),
			}
		}
	}
	return tasks
}

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

func updateStatus(ctx context.Context, cli client.Client, agent *agentv2.KubernetesAgent) 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.KubernetesAgent], 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 {
			age := time.Since(c.Agent.CreationTimestamp.Time)
			l.Info(fmt.Sprintf("mark o11y become running when creating o11y, since=%v", age))
			c.Agent.Status.Phase = agentv2.O11yAgentPhaseNormal
		}
	}
	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 updateStatusToDeleting(ctx context.Context, l logr.Logger, c *reconciler.Context[*agentv2.KubernetesAgent], results []deploy.Result) (ctrl.Result, error) {
	var deleted bool
	var res ctrl.Result
	var err error

	// If 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 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
}