package task

import (
	"context"
	"sort"
	"strings"

	ec2types "github.com/aws/aws-sdk-go-v2/service/ec2/types"
	"github.com/aws/aws-sdk-go/aws"
	ec2v1alpha1 "github.com/crossplane-contrib/provider-aws/apis/ec2/v1alpha1"
	"github.com/crossplane-contrib/provider-aws/apis/eks/manualv1alpha1"
	xpv1 "github.com/crossplane/crossplane-runtime/apis/common/v1"
	opv1alpha1 "github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"github.com/samber/lo"
	corev1 "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/log"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/config"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	"github.com/tidbcloud/infra-provider/pkg/nodegroup"
	"github.com/tidbcloud/infra-provider/pkg/util"
)

// compactbackup nodepools are isolated per TiDB cluster on AWS.
// Even when clusters share the same dataplane EKS, each TiDB cluster may use a
// different node role, so we must scope nodegroup labels and names by cluster.
type awsProvider struct {
	bucket string
	prefix string
	region string
	// kubernetesName is the name of the cluster
	kubernetesName string
	kubeVersion    string
	// nodeRole is the role of the node
	nodeRole           string
	accountID          string
	assumeRoleARN      string
	externalID         string
	providerConfigName string

	// resource tag
	tenantID  string
	projectID string
	clusterID string

	aws    cloudv1.CloudAWS
	config config.CloudConfigProvider
}

func NewAWSProvider(bucket string, region string, kubernetesName string, kubeVersion string, nodeRole string, accountID string, assumeRoleARN string, externalID string, providerConfigName string, tenantID string, projectID string, clusterID string, aws cloudv1.CloudAWS, cfg config.CloudConfigProvider) (CloudProvider, error) {
	bucket, prefix, err := util.ParseS3OrGCSBucketURI(bucket)
	if err != nil {
		return nil, err
	}
	return &awsProvider{
		bucket:             bucket,
		prefix:             prefix,
		region:             region,
		kubernetesName:     kubernetesName,
		kubeVersion:        kubeVersion,
		nodeRole:           nodeRole,
		aws:                aws,
		accountID:          accountID,
		assumeRoleARN:      assumeRoleARN,
		externalID:         externalID,
		providerConfigName: providerConfigName,
		tenantID:           tenantID,
		projectID:          projectID,
		clusterID:          clusterID,
		config:             cfg,
	}, nil
}

func (p *awsProvider) GetProviderConfigName() string {
	// use the providerconfig of cluster
	return p.providerConfigName
}

// Nodepool and node labels should be
func (p *awsProvider) GetNodepoolLabels() map[string]string {
	return map[string]string{
		consts.TagKeyTenant:         p.tenantID,
		consts.TagKeyProject:        p.projectID,
		consts.TagKeyCluster:        p.clusterID,
		consts.TagKeyUsedBy:         consts.UsedByCustomerTiDB,
		cloudv1.TagLabelServiceType: cloudv1.ServiceTypeDedicated,
		// used for node schedule
		namer.ComponentSelectorKey: string(cloudv1.ComponentCompactBackup),
		consts.K8sVersionLabelKey:  p.kubeVersion,
	}
}

func (p *awsProvider) GetPodAffinity() *corev1.Affinity {
	return &corev1.Affinity{
		PodAntiAffinity: &corev1.PodAntiAffinity{
			RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{
				{
					LabelSelector: &metav1.LabelSelector{
						MatchLabels: map[string]string{
							consts.ComponentLabelKey: string(cloudv1.ComponentCompactBackup),
						},
					},
					TopologyKey: "kubernetes.io/hostname",
				},
			},
		},
		NodeAffinity: &corev1.NodeAffinity{
			RequiredDuringSchedulingIgnoredDuringExecution: &corev1.NodeSelector{
				NodeSelectorTerms: []corev1.NodeSelectorTerm{
					{
						MatchExpressions: []corev1.NodeSelectorRequirement{
							{
								Key:      namer.ComponentSelectorKey,
								Operator: corev1.NodeSelectorOpIn,
								Values:   []string{string(cloudv1.ComponentCompactBackup)},
							},
							// cluster-level node selector
							{
								Key:      consts.TagKeyCluster,
								Operator: corev1.NodeSelectorOpIn,
								Values:   []string{p.clusterID},
							},
						},
					},
				},
			},
		},
	}
}

func (p *awsProvider) BuildNodePool(ctx context.Context, pc string, np *NodePool) ([]client.Object, error) {
	logger := log.FromContext(ctx)
	ngName := p.nodePoolName(np)
	tags := np.Labels
	tags["Name"] = ngName

	ngLaunchTemplate := &ec2v1alpha1.LaunchTemplate{
		ObjectMeta: metav1.ObjectMeta{
			Name:   ngName,
			Labels: np.Labels,
		},
		Spec: ec2v1alpha1.LaunchTemplateSpec{
			ForProvider: ec2v1alpha1.LaunchTemplateParameters{
				Region:             np.Region,
				LaunchTemplateName: aws.String(ngName),
				LaunchTemplateData: &ec2v1alpha1.RequestLaunchTemplateData{
					TagSpecifications: []*ec2v1alpha1.LaunchTemplateTagSpecificationRequest{
						{
							ResourceType: aws.String(string(ec2types.ResourceTypeVolume)),
							Tags:         util.GetEC2V1AlphaTags(tags),
						},
						{
							ResourceType: aws.String(string(ec2types.ResourceTypeInstance)),
							Tags:         util.GetEC2V1AlphaTags(tags),
						},
						{
							ResourceType: aws.String(string(ec2types.ResourceTypeNetworkInterface)),
							Tags:         util.GetEC2V1AlphaTags(tags),
						},
					},
				},
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{Name: p.providerConfigName},
			},
		},
	}

	// hardcode system disk size
	size, err := util.ParseDiskSizeAsGiB("50Gib")
	if err != nil {
		return nil, err
	}
	// set system disk map
	ngLaunchTemplate.Spec.ForProvider.LaunchTemplateData.BlockDeviceMappings = nodegroup.GetBlockDeviceMappingsForBottlerocket(size)

	nodeTains := []manualv1alpha1.Taint{}
	for _, taint := range np.Taints {
		effect := util.ToCrossplaneTaintEffect(taint.Effect)
		if effect == "" {
			continue
		}
		nodeTains = append(nodeTains, manualv1alpha1.Taint{
			Effect: effect,
			Key:    ptr.To(taint.Key),
			Value:  ptr.To(taint.Value),
		})
	}

	instanceTypes, err := p.config.GetInstanceTypes()
	if err != nil {
		logger.Error(err, "compactbackup: failed to load instance types, falling back to x86 AMI")
	}
	amiType, _ := nodegroup.GetAMITypeByInstanceType(instanceTypes, np.InstanceType)

	group := &manualv1alpha1.NodeGroup{
		ObjectMeta: metav1.ObjectMeta{
			Name:   ngName,
			Labels: np.Labels,
		},
		Spec: manualv1alpha1.NodeGroupSpec{
			ForProvider: manualv1alpha1.NodeGroupParameters{
				AMIType:       lo.ToPtr(amiType),
				ClusterName:   np.KubeClusterID,
				Labels:        np.Labels,
				Region:        np.Region,
				Taints:        nodeTains,
				InstanceTypes: []string{np.InstanceType},
				ScalingConfig: &manualv1alpha1.NodeGroupScalingConfig{
					MinSize: ptr.To[int32](0),
					MaxSize: ptr.To[int32](100),
				},
				Subnets: p.GetNodeSubnets(),
				LaunchTemplate: &manualv1alpha1.LaunchTemplateSpecification{
					NameRef: &xpv1.Reference{
						Name: ngName,
					},
				},
				NodeRole: np.Identity,
				Tags:     np.Tags,
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{Name: pc},
			},
		},
	}

	return []client.Object{
		ngLaunchTemplate,
		group,
	}, nil
}

// nodePoolName builds the EKS nodegroup/LaunchTemplate name for a compact backup nodepool.
// Format: compact-{instanceType}-{clusterID}-{k8sVersion}, bounded to 63 chars.
// On AWS, each TiDB cluster gets its own nodegroup (different node roles per cluster),
// so clusterID is used as the identity to guarantee uniqueness.
func (p *awsProvider) nodePoolName(np *NodePool) string {
	normalizedType := strings.ReplaceAll(np.InstanceType, ".", "-")
	normalizedType = strings.ReplaceAll(normalizedType, "_", "-")
	k8sVersion := p.GetKubernetesVersion()

	return namer.NewNameHashOnlyExceed(p.clusterID, "compact-"+normalizedType, nodegroup.NormalizeVersion(k8sVersion), 63)
}

func (p *awsProvider) GetKubernetesID() string {
	return p.kubernetesName
}

func (p *awsProvider) GetKubernetesName() string {
	return p.kubernetesName
}

func (p *awsProvider) GetKubernetesVersion() string {
	return p.kubeVersion
}

func (p *awsProvider) GetNodeSubnets() []string {
	subnetIDs := []string{}
	for _, subnetGroups := range p.aws.SubnetGroups {
		if subnetGroups.Name == cloudv1.AWSSubnetGroupNodes {
			for _, subnet := range subnetGroups.Subnets {
				subnetIDs = append(subnetIDs, subnet.SubnetID)
			}
			break
		}
	}
	// sort subnet IDs to make sure the order is consistent
	sort.Strings(subnetIDs)

	return subnetIDs
}

func (p *awsProvider) GetNodeZones() []string {
	// unused
	return []string{}
}

func (p *awsProvider) GetPodSubnets() []string {
	// unused
	return []string{}
}

func (p *awsProvider) GetNodeIdentity() string {
	return p.nodeRole
}

// OverlayCompact implements storage overlay for CompactBackup
func (p *awsProvider) OverlayCompact(compactBackup *opv1alpha1.CompactBackup) error {
	compactBackup.Spec.StorageProvider = opv1alpha1.StorageProvider{
		S3: &opv1alpha1.S3StorageProvider{
			Bucket:   p.bucket,
			Prefix:   p.prefix,
			Region:   p.region,
			Provider: opv1alpha1.S3StorageProviderTypeAWS,
		},
	}
	compactBackup.Spec.BR.SendCredToTikv = ptr.To(false)
	return nil
}

// OverlayCompactStatus implements status overlay for CompactBackup
func (p *awsProvider) OverlayCompactStatus(cloudCompactBackup *cloudv1.CompactBackup, opCompactBackup *opv1alpha1.CompactBackup) error {
	// Set backup path for compact backup, similar to regular backup
	backupPath := ptr.Deref(cloudCompactBackup.Spec.Storage, "")
	if backupPath != "" {
		// Set the backup path to the original storage path without appending the CompactBackup name
		cloudCompactBackup.Status.BackupPath = backupPath
	}
	return nil
}