package commonazure

import (
	"context"
	"fmt"
	"hash/fnv"
	"strconv"
	"time"

	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	aksv1beta1 "github.com/upbound/provider-azure/apis/containerservice/v1beta1"
	corev1 "k8s.io/api/core/v1"
	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/controllerutil"

	"github.com/tidbcloud/observability/agent-control-server/pkg/agentbuild"
	"github.com/tidbcloud/observability/agent-control-server/pkg/config"
	"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/reconciler"
	"github.com/tidbcloud/observability/agent-control-server/pkg/util"
)

const (
	defaultMinSize     = 1
	defaultMaxSize     = 3
	defaultDesiredSize = 1
	defaultDiskSize    = 100 // in GB
	defaultSpecName    = "default"
)

// sanitizeAKSNodePoolName converts an arbitrary name into an AKS-compliant agent pool
// name: max 12 characters, lowercase alphanumeric only, starting with a letter.
// It uses a deterministic FNV-1a hash to avoid collisions while staying within limits.
func sanitizeAKSNodePoolName(name string) string {
	h := fnv.New64a()
	h.Write([]byte(name))
	return fmt.Sprintf("ny%010x", h.Sum64())[:12]
}

// ptrOrNil returns a pointer to the string if non-empty, or nil.
func ptrOrNil(s string) *string {
	if s == "" {
		return nil
	}
	return &s
}

// AKSConfigProvider is a function that extracts AKS configuration from an agent.
// Returns subscriptionID, resourceGroup, subnetID, podSubnetID, and error.
type AKSConfigProvider func() (subscriptionID, resourceGroup, subnetID, podSubnetID string, err error)

// NodePoolTask manages the AKS NodePool for the observability agent.
// This shared implementation eliminates code duplication between o11yk8sagent and o11ytidbagent.
type NodePoolTask[T client.Object] struct {
	*reconciler.BaseTask[T]
	clusterName      string
	nameFunc         func(string) string
	specs            config.CustomerAgentSpecs
	taints           []corev1.Taint
	nodeMatchLabels  map[string]string
	nodeConfigLabels map[string]string
	getAKSConfig     AKSConfigProvider
	clusterMetaFunc  func() deploy.ClusterMeta
}

func NewNodePoolTask[T client.Object](
	ctx *reconciler.Context[T],
	clusterName string,
	nameFunc func(string) string,
	nodeMatchLabels map[string]string,
	nodeConfigLabels map[string]string,
	specs config.CustomerAgentSpecs,
	taints []corev1.Taint,
	getAKSConfig AKSConfigProvider,
	clusterMetaFunc func() deploy.ClusterMeta,
) *NodePoolTask[T] {
	return &NodePoolTask[T]{
		BaseTask:         &reconciler.BaseTask[T]{Context: ctx},
		clusterName:      clusterName,
		nameFunc:         func(s string) string { return sanitizeAKSNodePoolName(nameFunc(s)) },
		specs:            specs,
		taints:           taints,
		nodeMatchLabels:  nodeMatchLabels,
		nodeConfigLabels: nodeConfigLabels,
		getAKSConfig:     getAKSConfig,
		clusterMetaFunc:  clusterMetaFunc,
	}
}

func (t *NodePoolTask[T]) Name() string {
	return "NodePool"
}

func (t *NodePoolTask[T]) ClusterMeta() deploy.ClusterMeta {
	return t.clusterMetaFunc()
}

func (t *NodePoolTask[T]) Create(ctx context.Context) deploy.Result {
	return deploy.PanicOrDone(t.create(ctx))
}

func (t *NodePoolTask[T]) PostCreate(ctx context.Context) deploy.Result {
	desiredSpec := util.GetAnnotationWithDefault(t.Agent, consts.AnnotationKeyO11yDesiredSpec, defaultSpecName)
	name := t.nameFunc(desiredSpec)
	if err := t.checkNodePool(ctx, name); err != nil {
		t.clearStalledExternalCreate(ctx, name)
		return deploy.Panic(err).WithRequeueAfter(time.Minute)
	}
	switch util.GetAnnotation(t.Agent, consts.AnnotationKeyO11ySpecScaleStatus) {
	case consts.SpecScaleStatusPendingScale:
		util.SetAnnotation(t.Agent, consts.AnnotationKeyO11ySpecScaleStatus, consts.SpecScaleStatusNewNodeGroupSynced)
	}
	return deploy.Done()
}

// clearStalledExternalCreate best-effort clears crossplane's stalled
// external-create-pending annotation on the node pool so crossplane resumes
// reconciling it. See agentbuild.ClearStalledExternalCreate for why this is safe.
func (t *NodePoolTask[T]) clearStalledExternalCreate(ctx context.Context, name string) {
	cleared, err := agentbuild.ClearStalledExternalCreate(ctx, t.Local,
		&aksv1beta1.KubernetesClusterNodePool{ObjectMeta: metav1.ObjectMeta{Name: name}})
	switch {
	case err != nil:
		t.L.Error(err, "failed to clear stalled crossplane external-create-pending annotation", "name", name)
	case cleared:
		t.L.Info("cleared stalled crossplane external-create-pending annotation for crossplane to resume", "name", name)
	}
}

func (t *NodePoolTask[T]) Delete(ctx context.Context) deploy.Result {
	// capable with legacy
	if err := t.deleteNodePool(ctx, t.nameFunc(defaultSpecName)); err != nil {
		return deploy.Panic(err)
	}
	return deploy.PanicOrDone(t.deleteNodePoolByLabels(ctx, t.nodeMatchLabels))
}

func (t *NodePoolTask[T]) PostDelete(ctx context.Context) deploy.Result {
	// capable with legacy
	name := t.nameFunc(defaultSpecName)
	if err := t.checkNodePoolDeleted(ctx, name); err != nil {
		t.clearStalledExternalCreate(ctx, name)
		return deploy.Panic(err).WithRequeueAfter(time.Minute)
	}
	if err := t.checkNodePoolDeletedByLabels(ctx, t.nodeMatchLabels); err != nil {
		return deploy.Panic(err).WithRequeueAfter(time.Minute)
	}
	return deploy.Done()
}

func (t *NodePoolTask[T]) create(ctx context.Context) error {
	desiredSpec := util.GetAnnotationWithDefault(t.Agent, consts.AnnotationKeyO11yDesiredSpec, defaultSpecName)
	spec := t.specs.Select(desiredSpec)
	if spec == nil {
		return fmt.Errorf("desired spec not found: %s", desiredSpec)
	}
	if err := t.reconcile(ctx, t.nameFunc(desiredSpec), spec.InstanceType); err != nil {
		return err
	}

	switch util.GetAnnotation(t.Agent, consts.AnnotationKeyO11ySpecScaleStatus) {
	case consts.SpecScaleStatusPendingScale:
		// If we are in a pending-scale state, remember to reconcile the original spec
		// until the new spec is ready. This is to avoid a situation where the old spec
		// cannot be further edited when the new spec is delayed.
		previousSpec := util.GetAnnotation(t.Agent, consts.AnnotationKeyO11yPreviousSpec)
		if previousSpec == "" {
			return fmt.Errorf("unexpected empty previous spec name")
		}
		spec := t.specs.Select(previousSpec)
		if spec == nil {
			return fmt.Errorf("previous spec not found: %s", previousSpec)
		}
		if err := t.reconcile(ctx, t.nameFunc(previousSpec), spec.InstanceType); err != nil {
			return err
		}

	case consts.SpecScaleStatusResourceUpdated:
		previousSpec := util.GetAnnotation(t.Agent, consts.AnnotationKeyO11yPreviousSpec)
		if previousSpec == "" {
			return fmt.Errorf("unexpected empty previous spec name")
		}
		prevName := t.nameFunc(previousSpec)
		if err := t.deleteNodePool(ctx, prevName); err != nil {
			return fmt.Errorf("failed to delete previous nodepool: %w", err)
		}
		if err := t.checkNodePoolDeleted(ctx, prevName); err != nil {
			t.clearStalledExternalCreate(ctx, prevName)
			return deploy.NewRetriableError(err, ctrl.Result{Requeue: true, RequeueAfter: time.Minute})
		}
		util.SetAnnotation(t.Agent, consts.AnnotationKeyO11ySpecScaleStatus, consts.SpecScaleStatusNormal)
	}
	return nil
}

func (t *NodePoolTask[T]) updateSizeAnnotation(key string, val int) (int, error) {
	s := util.SetAnnotationIfEmpty(t.Agent, key, strconv.Itoa(val))
	i, err := strconv.Atoi(s)
	if err != nil {
		return 0, fmt.Errorf("invalid key '%s': %w", key, err)
	}
	return i, nil
}

// reconcile ensures that the NodePool with the given name and instance type exists
// and is configured according to the specifications derived from the agent annotations
// and the AKS configuration from the agent spec.
func (t *NodePoolTask[T]) reconcile(ctx context.Context, name, instanceType string) error {
	_, _, subnetID, podSubnetID, err := t.getAKSConfig()
	if err != nil {
		return err
	}

	aksCluster, err := t.getCluster(ctx)
	if err != nil {
		return err
	}

	// Prefer the annotation (set by /ops/nodegroups/upgrade); fall back to the
	// observed cluster version for bootstrap and write it back for idempotency.
	orchestratorVersion := util.GetAnnotation(t.Agent, consts.AnnotationKeyO11yNodeVersion)
	if orchestratorVersion == "" && aksCluster.Status.AtProvider.CurrentKubernetesVersion != nil {
		orchestratorVersion = *aksCluster.Status.AtProvider.CurrentKubernetesVersion
		if orchestratorVersion != "" {
			util.SetAnnotation(t.Agent, consts.AnnotationKeyO11yNodeVersion, orchestratorVersion)
		}
	}
	if orchestratorVersion == "" {
		return fmt.Errorf("orchestrator version is empty: neither annotation %s nor AKS cluster %s observed version is set",
			consts.AnnotationKeyO11yNodeVersion, t.clusterName)
	}
	if aksCluster.Spec.ProviderConfigReference == nil {
		return fmt.Errorf("AKS cluster %s providerConfigReference is empty", t.clusterName)
	}
	providerConfigRef := *aksCluster.Spec.ProviderConfigReference

	minSize, err := t.updateSizeAnnotation(consts.AnnotationKeyO11yNodeMinSize, defaultMinSize)
	if err != nil {
		return err
	}
	maxSize, err := t.updateSizeAnnotation(consts.AnnotationKeyO11yNodeMaxSize, defaultMaxSize)
	if err != nil {
		return err
	}
	desiredSize, err := t.updateSizeAnnotation(consts.AnnotationKeyO11yNodeDesiredSize, defaultDesiredSize)
	if err != nil {
		return err
	}
	diskSize, err := t.updateSizeAnnotation(consts.AnnotationKeyO11yNodeDiskSize, defaultDiskSize)
	if err != nil {
		return err
	}

	// Convert taints to Azure format (key=value:Effect)
	taints := make([]*string, 0, len(t.taints))
	for _, taint := range t.taints {
		taintStr := fmt.Sprintf("%s=%s:%s", taint.Key, taint.Value, taint.Effect)
		taints = append(taints, &taintStr)
	}

	// Convert node labels to Azure format
	nodeLabels := make(map[string]*string)
	for k, v := range t.nodeConfigLabels {
		v := v
		nodeLabels[k] = &v
	}
	nodeLabels["Name"] = ptr.To(name)
	nodeLabels["tidbcloud.com/nodegroup-name"] = ptr.To(name)

	nodePool := &aksv1beta1.KubernetesClusterNodePool{
		ObjectMeta: metav1.ObjectMeta{Name: name},
	}
	_, err = controllerutil.CreateOrUpdate(ctx, t.Local, nodePool, func() error {
		if nodePool.Labels == nil {
			nodePool.Labels = make(map[string]string)
		}
		for k, v := range t.nodeMatchLabels {
			nodePool.Labels[k] = v
		}

		nodePool.Spec = aksv1beta1.KubernetesClusterNodePoolSpec{
			ResourceSpec: xpv1.ResourceSpec{
				DeletionPolicy: xpv1.DeletionDelete,
				ProviderConfigReference: &xpv1.Reference{
					Name: providerConfigRef.Name,
				},
			},
			ForProvider: aksv1beta1.KubernetesClusterNodePoolParameters{
				KubernetesClusterIDRef: &xpv1.Reference{Name: t.clusterName},
				VMSize:                 ptr.To(instanceType),
				OsDiskSizeGb:           ptr.To(float64(diskSize)),
				NodeCount:              ptr.To(float64(desiredSize)),
				MinCount:               ptr.To(float64(minSize)),
				MaxCount:               ptr.To(float64(maxSize)),
				EnableAutoScaling:      ptr.To(true),
				OrchestratorVersion:    ptr.To(orchestratorVersion),
				Mode:                   ptr.To("User"),
				NodeLabels:             nodeLabels,
				NodeTaints:             taints,
				VnetSubnetID:           ptrOrNil(subnetID),
				PodSubnetID:            ptrOrNil(podSubnetID),
			},
		}
		return nil
	})
	if err != nil {
		return fmt.Errorf("failed to create or update node pool %s: %w", name, err)
	}

	return nil
}

func (t *NodePoolTask[T]) getCluster(ctx context.Context) (*aksv1beta1.KubernetesCluster, error) {
	aksCluster := &aksv1beta1.KubernetesCluster{}
	if err := t.Local.Get(ctx, client.ObjectKey{Name: t.clusterName}, aksCluster); err != nil {
		return nil, fmt.Errorf("failed to get AKS cluster %s: %w", t.clusterName, err)
	}
	return aksCluster, nil
}

func (t *NodePoolTask[T]) checkNodePool(ctx context.Context, name string) error {
	nodePool := &aksv1beta1.KubernetesClusterNodePool{}
	err := t.Local.Get(ctx, client.ObjectKey{Name: name}, nodePool)
	if err != nil {
		return fmt.Errorf("failed to get node pool %s: %w", name, err)
	}

	if condition := nodePool.Status.GetCondition(xpv1.TypeReady); condition.Status != corev1.ConditionTrue {
		return fmt.Errorf("node pool %s is not ready, reason: %s, message: %s", name, condition.Reason, condition.Message)
	}
	return nil
}

func (t *NodePoolTask[T]) checkNodePoolDeleted(ctx context.Context, name string) error {
	err := t.Local.Get(ctx, client.ObjectKey{Name: name}, &aksv1beta1.KubernetesClusterNodePool{})
	if err == nil {
		return fmt.Errorf("node pool %s is still present", name)
	}
	if client.IgnoreNotFound(err) == nil {
		return nil
	}
	return fmt.Errorf("failed to get node pool %s: %w", name, err)
}

func (t *NodePoolTask[T]) deleteNodePool(ctx context.Context, name string) error {
	return client.IgnoreNotFound(t.Local.Delete(ctx, &aksv1beta1.KubernetesClusterNodePool{
		ObjectMeta: metav1.ObjectMeta{Name: name},
	}))
}

// deleteNodePoolByLabels deletes all node pools that match the given labels.
// This fixes the node leak issue by properly cleaning up node pools during agent deletion.
func (t *NodePoolTask[T]) deleteNodePoolByLabels(ctx context.Context, labels map[string]string) error {
	if len(labels) == 0 {
		return nil
	}

	var nodePools aksv1beta1.KubernetesClusterNodePoolList
	if err := t.Local.List(ctx, &nodePools, client.MatchingLabels(labels)); err != nil {
		return fmt.Errorf("failed to list node pools: %w", err)
	}
	for _, nodePool := range nodePools.Items {
		if err := client.IgnoreNotFound(t.Local.Delete(ctx, &nodePool)); err != nil {
			return fmt.Errorf("failed to delete node pool %s: %w", nodePool.Name, err)
		}
	}

	return nil
}

// checkNodePoolDeletedByLabels checks that all node pools matching the given labels have been deleted.
// This fixes the node leak issue by verifying cleanup during agent deletion.
func (t *NodePoolTask[T]) checkNodePoolDeletedByLabels(ctx context.Context, labels map[string]string) error {
	if len(labels) == 0 {
		return nil
	}

	var nodePools aksv1beta1.KubernetesClusterNodePoolList
	if err := t.Local.List(ctx, &nodePools, client.MatchingLabels(labels)); err != nil {
		return fmt.Errorf("failed to list node pools: %w", err)
	}
	if len(nodePools.Items) > 0 {
		return fmt.Errorf("node pools with labels %v are still present", labels)
	}

	return nil
}