package task

import (
	"context"
	"crypto/md5"
	"encoding/hex"
	"fmt"
	"os"

	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
	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

	aws cloudv1.CloudAWS

	// resource tag
	tenantID  string
	projectID string
	clusterID string

	snapshotBackupPrefix string
	// PITR needed
	logBackupPrefix string
	config          config.CloudConfigProvider
}

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

	var logBackupPrefix string
	if logBackupBucket != "" {
		// Log backup and snapshot backup use the same bucket
		_, logBackupPrefix, err = util.ParseS3OrGCSBucketURI(logBackupBucket)
		if err != nil {
			return nil, err
		}
	}

	// log backup and snapshot backup use the same bucket

	return &awsProvider{
		kubernetesName:       kubernetesName,
		kubeVersion:          kubeVersion,
		nodeRole:             nodeRole,
		aws:                  aws,
		accountID:            accountID,
		assumeRoleARN:        assumeRoleARN,
		externalID:           externalID,
		bucket:               bucket,
		region:               region,
		providerConfigName:   providerConfigName,
		tenantID:             tenantID,
		projectID:            projectID,
		clusterID:            clusterID,
		snapshotBackupPrefix: snapshotBackupPrefix,
		// PITR needed
		logBackupPrefix: logBackupPrefix,
		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.ComponentRestore),
		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.ComponentRestore),
						},
					},
					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.ComponentRestore)},
							},
							// 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: pc},
			},
		},
	}

	// 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, "restore: 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-restore-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 `restore-` to avoid name starts with a number
	return fmt.Sprintf("restore-%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 {
	// TODO: unimplemented
	return []string{}
}

func (p *awsProvider) GetPodSubnets() []string {
	// TODO: unimplemented
	return []string{}
}

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

func (p *awsProvider) Overlay(cloudRestore *cloudv1.Restore, restore *opv1alpha1.Restore) error {
	bucketRegion := ptr.Deref(cloudRestore.Spec.From.Region, p.region)
	restore.Spec.StorageProvider = opv1alpha1.StorageProvider{
		S3: &opv1alpha1.S3StorageProvider{
			Bucket:   p.bucket,
			Prefix:   p.snapshotBackupPrefix,
			Provider: opv1alpha1.S3StorageProviderTypeAWS,
			Region:   bucketRegion,
		},
	}
	restore.Spec.BR.SendCredToTikv = ptr.To(false)

	if restore.Spec.Mode == opv1alpha1.RestoreModePiTR {
		// log backup and snapshot backup use the same bucket
		restore.Spec.StorageProvider.S3.Prefix = p.logBackupPrefix
		restore.Spec.PitrFullBackupStorageProvider = opv1alpha1.StorageProvider{
			S3: &opv1alpha1.S3StorageProvider{
				Bucket:   p.bucket,
				Prefix:   p.snapshotBackupPrefix,
				Provider: opv1alpha1.S3StorageProviderTypeAWS,
				Region:   bucketRegion,
			},
		}
	}

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