package task

import (
	"context"
	"errors"
	"fmt"
	"maps"
	"strings"

	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"

	"github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/config"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/consts"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/deploy"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/namer"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/util"
)

type NodePool struct {
	*BaseTask
}

func NewNodePool(ctx *Context) *NodePool {
	return &NodePool{BaseTask: &BaseTask{Context: ctx}}
}

func (t *NodePool) Name() string {
	return "NodePool"
}

func (t *NodePool) Create(ctx context.Context) deploy.Result {
	if t.DataPlane.Status.AppBox == nil {
		return deploy.Error(fmt.Errorf("appbox ref is missing")).WithRequeuAfter(requeueAfterDuration)
	}

	appbox := new(v1alpha1.APPBox)
	if err := t.AppBoxRW.Get(ctx, client.ObjectKey{
		Namespace: t.DataPlane.Status.AppBox.Namespace,
		Name:      t.DataPlane.Status.AppBox.Name,
	}, appbox); err != nil {
		return deploy.Error(err).WithRequeuAfter(requeueAfterDuration)
	}

	version := appbox.Spec.KubernetesConfig.Version
	versionName := version
	// ACK nodepool supports in-place upgrade, so remove version part from nodepool name
	// so the same NodePool CR is reused across k8s version upgrades.
	if t.Config.Provider == consts.ProviderTypeACK {
		versionName = ""
	}

	// The pools created by this loop are worker NodePool templates sourced
	// from APPBoxConfig.NodePools. Admin/o11y are created by the appbox
	// controller as EKS NodeGroups (Upbound resources), so they never enter
	// this code path. The annotation tidbcloud.com/builtin-nodepool=true
	// still excludes these from CSNG's sibling-inheritance filter (the
	// workload chain), which is the only consumer of that annotation.
	//
	// Placement-group eligibility is topology-derived per
	// docs/design/single-az-ha-placement-groups.md: a built-in worker
	// pool is eligible for an EC2 partition placement group iff the
	// dataplane is AWS + single-AZ. The CSNG gRPC CreateNodePool path
	// (the producer-side source of truth for non-AppBox workload
	// NodePools) does NOT fire on AppBox bootstrap, so this controller
	// IS the producer for these pools and must stamp
	// Spec.PlacementGroupEnabled at create time. The EKS PG controller
	// reads the durable spec and creates the partition group on its
	// next reconcile.

	for i := range t.Config.ControllerConfig.APPBoxConfig.NodePools {
		pool := t.Config.ControllerConfig.APPBoxConfig.NodePools[i]
		name := t.buildNodePoolName(
			t.DataPlane.Name,
			pool.InstanceType,
			versionName,
		)
		labels := map[string]string{
			v1alpha1.RegionKey:      t.Config.RegionConfig.Name,
			v1alpha1.VPCNameKey:     t.DataPlane.Status.Vpc.Name,
			v1alpha1.ManageKey:      v1alpha1.ManageFull,
			v1alpha1.DataPlaneIDKey: t.DataPlane.Name,
			v1alpha1.APPBoxNameKey:  t.DataPlane.Status.AppBox.Name,
		}
		labels = util.MergeMaps(t.Config.ResourceLabels, labels, t.DataPlane.Spec.CloudTags)
		nodePoolCR := &v1alpha1.NodePool{
			ObjectMeta: metav1.ObjectMeta{
				Name:      name,
				Namespace: t.Config.Namespace,
				Labels:    labels,
				Annotations: map[string]string{
					consts.AnnotationKeyBuiltinNodePool: "true",
				},
			},
			Spec: v1alpha1.NodePoolSpec{
				DataPlane: v1alpha1.Reference{
					Namespace: t.Config.Namespace,
					Name:      t.DataPlane.Name,
				},
				InstanceType:          pool.InstanceType,
				KubernetesVersion:     version,
				CpuCore:               uint32(pool.CPU),
				AvailableCpuMilliCore: nodePoolConfigAvailableCPUMilli(pool),
				MaxCpuMilli:           nodePoolConfigMaxCPUMilli(pool),
				MemoryGi:              uint32(pool.MemoryGi),
				AvailableMemoryMi:     uint32(pool.AvailableMemoryMi),
				MaxMemoryMi:           nodePoolConfigMaxMemoryMi(pool),
				NodeType:              consts.NodeTypeDedicated,
				WithLocalDisk:         pool.HasLocalDisk,
				LocalDiskSizeGi:       uint32(pool.LocalDiskSize),
				PlacementGroupEnabled: singleAzEKS(t.Config.Provider, t.DataPlane.Spec.NodeZones),
			},
		}
		existing := &v1alpha1.NodePool{}
		key := client.ObjectKey{Namespace: nodePoolCR.Namespace, Name: nodePoolCR.Name}
		if err := t.NodePoolRW.Get(ctx, key, existing); err != nil {
			if !apierrors.IsNotFound(err) {
				return deploy.Error(err)
			}
			if err := t.NodePoolRW.Create(ctx, nodePoolCR); err != nil {
				return deploy.Error(err)
			}
			continue
		}

		// Spec.PlacementGroupEnabled is the durable record stamped at
		// create time above. Reconcile does not rewrite the spec: a
		// topology change after creation must not silently migrate a
		// legacy Spec=true pool off the partition path (which would
		// orphan its PG and 3 partition NodeGroups). Legacy intent
		// survives reconciles per docs/design/single-az-ha-placement-groups.md.

		if nodePoolMutableEqual(existing, nodePoolCR) {
			continue
		}

		existing.Labels = nodePoolCR.Labels
		existing.Annotations = nodePoolCR.Annotations
		existing.Spec.Labels = nodePoolCR.Spec.Labels
		existing.Spec.Taints = nodePoolCR.Spec.Taints

		if err := t.NodePoolRW.Update(ctx, existing); err != nil {
			return deploy.Error(err)
		}
	}
	return deploy.Done()
}

// singleAzEKS returns whether the given provider+zone-shape is the only
// topology where EC2 partition placement groups apply. It is the
// topology-derived value stamped on Spec.PlacementGroupEnabled for
// built-in worker NodePools at create time.
//
// Returns false when:
//   - the provider is anything other than EKS (Azure / GCP / ACK have
//     their own single-AZ primitives and must not opt into the
//     partition path);
//   - NodeZones is empty or has more than one entry (multi-AZ pools
//     span zones that a partition PG cannot straddle).
//
// Exposed at package scope so the test fixture and any future producer
// can reuse the same predicate.
func singleAzEKS(provider string, nodeZones []string) bool {
	return provider == consts.ProviderTypeEKS && len(nodeZones) == 1
}

// Built-in worker NodePools created by this controller have their
// Spec.PlacementGroupEnabled set at create time from
// singleAzEKS(Config.Provider, DataPlane.Spec.NodeZones) above. The
// legacy "resolvePlacementGroupEnabled" helper that used to derive it
// from a per-pool config knob was removed when the per-pool override
// was dropped (see #3040 and docs/design/single-az-ha-placement-groups.md).

func (t *NodePool) Delete(ctx context.Context) deploy.Result {
	nodePools := new(v1alpha1.NodePoolList)
	selector := client.MatchingLabels{v1alpha1.DataPlaneIDKey: t.DataPlane.Name}
	if err := t.NodePoolRW.List(ctx, nodePools, client.InNamespace(t.Config.Namespace), selector); err != nil {
		return deploy.Error(err)
	}
	var errs []error
	for i := range nodePools.Items {
		if err := t.NodePoolRW.Delete(ctx, &nodePools.Items[i]); err != nil && !apierrors.IsNotFound(err) {
			errs = append(errs, err)
		}
	}
	if len(errs) > 0 {
		return deploy.Error(errors.Join(errs...))
	}
	return deploy.Done()
}

func (t *NodePool) PostDelete(ctx context.Context) deploy.Result {
	nodePools := new(v1alpha1.NodePoolList)
	selector := client.MatchingLabels{v1alpha1.DataPlaneIDKey: t.DataPlane.Name}
	if err := t.NodePoolRW.List(ctx, nodePools, client.InNamespace(t.Config.Namespace), selector); err != nil {
		return deploy.Error(err)
	}
	if len(nodePools.Items) > 0 {
		return deploy.Error(fmt.Errorf("nodepool crs not deleted: %d remaining", len(nodePools.Items))).
			WithRequeuAfter(requeueAfterDuration)
	}
	return deploy.Done()
}

func (t *NodePool) buildNodePoolName(dataPlaneID, instanceType, k8sVersion string) string {
	versionPart := strings.ReplaceAll(k8sVersion, ".", "-")
	instancePart := strings.ToLower(strings.NewReplacer(".", "-", "_", "-").Replace(instanceType))
	prefix := fmt.Sprintf("%s-%s", t.Config.ResourceNamePrefix, instancePart)
	return namer.NewName(dataPlaneID, prefix, versionPart, -1)
}

func nodePoolMutableEqual(existing, desired *v1alpha1.NodePool) bool {
	if existing == nil || desired == nil {
		return existing == desired
	}
	if !maps.Equal(existing.Labels, desired.Labels) {
		return false
	}
	if !maps.Equal(existing.Spec.Labels, desired.Spec.Labels) {
		return false
	}
	if !taintsEqual(existing.Spec.Taints, desired.Spec.Taints) {
		return false
	}
	return true
}

func taintsEqual(a, b []v1alpha1.NodePoolTaint) bool {
	if len(a) == 0 && len(b) == 0 {
		return true
	}
	if len(a) != len(b) {
		return false
	}
	for i := range a {
		if a[i] != b[i] {
			return false
		}
	}
	return true
}

func nodePoolConfigAvailableCPUMilli(pool config.NodePoolConfig) uint32 {
	if pool.AvailableCPUMilli > 0 {
		return uint32(pool.AvailableCPUMilli)
	}
	if pool.AvailableCPU > 0 {
		return uint32(pool.AvailableCPU) * 1000
	}
	if pool.CPU > 1 {
		return uint32(pool.CPU-1) * 1000
	}
	return 0
}

func nodePoolConfigMaxCPUMilli(pool config.NodePoolConfig) uint32 {
	if pool.MaxCPUMilli > 0 {
		return uint32(pool.MaxCPUMilli)
	}
	return nodePoolConfigAvailableCPUMilli(pool)
}

func nodePoolConfigMaxMemoryMi(pool config.NodePoolConfig) uint32 {
	if pool.MaxMemoryMi > 0 {
		return uint32(pool.MaxMemoryMi)
	}
	return uint32(pool.AvailableMemoryMi)
}