package task

import (
	"context"
	"crypto/md5"
	"encoding/hex"
	"fmt"
	"os"
	"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"
	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"

	"github.com/tidbcloud/eks-provider/controllers/utils"
	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"
)

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.ComponentBackup),
		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.ComponentBackup),
						},
					},
					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.ComponentBackup)},
							},
							// 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.Name)
	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),
		})
	}

	var instanceTypes []cloudv1.InstanceType
	instanceTypes, err = p.config.GetInstanceTypes()
	if err != nil {
		logger.Error(err, "backup: 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:       ptr.To(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
}

func (p *awsProvider) nodePoolName(name string) string {
	// source example: np-backup-t3.medium-dedicated-dev-us-west-2-af0534b1cc83eb32932b1fc30d9d6c-8487a920-1.32-
	// inject cluster id to the nodepool name hash id
	name = fmt.Sprintf("%s-%s", name, p.clusterID)
	hash := md5.Sum([]byte(name))
	hashString := hex.EncodeToString(hash[:])
	// prefix with `backup-` to avoid name starts with a number
	return fmt.Sprintf("backup-%s", hashString[:11])
}

func (p *awsProvider) GetKubernetesID() string {
	// the same as cluster name
	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 := p.aws.SubnetGroups.SubnetsFilterByGroupNameOrZones([]cloudv1.AWSSubnetGroupName{
		cloudv1.AWSSubnetGroupNodes, cloudv1.AWSSubnetGroupSecondaryCIDR,
	}, []string{})

	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
}

func (p *awsProvider) Overlay(backup *opv1alpha1.Backup) error {
	// follow previous convention: append backup CR name to prefix
	prefix := strings.Trim(fmt.Sprintf("%s/%s", p.prefix, backup.Name), "/")
	backup.Spec.StorageProvider = opv1alpha1.StorageProvider{
		S3: &opv1alpha1.S3StorageProvider{
			Bucket:   p.bucket,
			Prefix:   prefix,
			Region:   p.region,
			Provider: opv1alpha1.S3StorageProviderTypeAWS,
		},
	}
	backup.Spec.BR.SendCredToTikv = ptr.To(false)

	// workaround for BR unsupported region
	if ep := os.Getenv(utils.S3GatewayEndpointEnv); ep != "" {
		backup.Spec.BR.Options = append(backup.Spec.BR.Options,
			fmt.Sprintf("--s3.endpoint=%s", ep),
		)
	}
	return nil
}

func (p *awsProvider) OverlayStatus(cloudBackup *cloudv1.Backup, opBackup *opv1alpha1.Backup) error {
	cloudBackup.Status.BucketName = p.bucket
	return nil
}