package task

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

	"github.com/crossplane-contrib/provider-gcp/apis/container/v1beta1"
	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"

	"github.com/tidbcloud/eks-provider/controllers/utils"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	"github.com/tidbcloud/infra-provider/pkg/util"
)

type gcpProvider struct {
	bucket string
	prefix string
	region string
	// gcp account id
	gcpProjectID   string
	kubernetesName string
	kubeVersion    string
	nodeRole       string

	providerConfigName string

	// resource tag
	backupTenantID  string
	backupProjectID string
}

func NewGCPProvider(bucket string, region string, gcpProjectID string, kubernetesName string, kubeVersion string, nodeRole string, providerConfigName string, backupTenantID string, backupProjectID string) (CloudProvider, error) {
	bucket, prefix, err := util.ParseS3OrGCSBucketURI(bucket)
	if err != nil {
		return nil, err
	}
	return &gcpProvider{
		bucket:             bucket,
		prefix:             prefix,
		region:             region,
		gcpProjectID:       gcpProjectID,
		kubernetesName:     kubernetesName,
		kubeVersion:        kubeVersion,
		nodeRole:           nodeRole,
		providerConfigName: providerConfigName,
		backupTenantID:     backupTenantID,
		backupProjectID:    backupProjectID,
	}, nil
}

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

// Nodepool and node labels should be
func (p *gcpProvider) GetNodepoolLabels() map[string]string {
	return map[string]string{
		consts.TagKeyTenant:         p.backupTenantID,
		consts.TagKeyProject:        p.backupProjectID,
		consts.TagKeyCluster:        consts.ClusterShared,
		consts.TagKeyUsedBy:         consts.UsedByCustomerTiDB,
		cloudv1.TagLabelServiceType: cloudv1.ServiceTypeDedicated,
		// used for node schedule
		namer.ComponentSelectorKey: string(cloudv1.ComponentBackup),
		consts.K8sVersionLabelKey:  p.kubeVersion,
	}
}

func (p *gcpProvider) 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)},
							},
						},
					},
				},
			},
		},
	}
}

func (p *gcpProvider) BuildNodePool(ctx context.Context, pc string, np *NodePool) ([]client.Object, error) {
	taints := []*v1beta1.NodeTaint{}
	for _, t := range np.Taints {
		taint := v1beta1.NodeTaint{
			Key:   t.Key,
			Value: t.Value,
			// hardcode to simplify convert NoSchedule Effect
			Effect: string("NO_SCHEDULE"),
		}
		taints = append(taints, &taint)
	}

	zones := make([]string, len(np.Zones))
	for i, z := range np.Zones {
		s := strings.Split(z, "-")
		if len(s) != 2 {
			return nil, fmt.Errorf("invalid zone %s", z)
		}
		zones[i] = s[len(s)-1]
	}

	kcnp := &v1beta1.NodePool{
		ObjectMeta: metav1.ObjectMeta{
			Name:   p.nodePoolName(np.Name),
			Labels: np.Labels,
		},
		Spec: v1beta1.NodePoolSpec{
			ForProvider: v1beta1.NodePoolParameters{
				Cluster: np.KubeClusterID,
				Autoscaling: &v1beta1.NodePoolAutoscaling{
					Enabled:      ptr.To(true),
					MaxNodeCount: ptr.To[int64](maxNodeCountPerZone),
					MinNodeCount: ptr.To[int64](0),
				},
				Config: &v1beta1.NodeConfig{
					Labels:         np.Labels,
					DiskSizeGb:     ptr.To[int64](100),
					DiskType:       ptr.To(utils.GetGcpNodepoolDiskType(np.InstanceType)),
					ImageType:      ptr.To("cos_containerd"),
					MachineType:    ptr.To(np.InstanceType),
					ServiceAccount: ptr.To(p.nodeRole), // node sa
					OauthScopes: []string{
						"https://www.googleapis.com/auth/cloud-platform",
					},
					Taints: taints,
				},
				InitialNodeCount: ptr.To(int64(0)),
				Locations:        zones,
				Management: &v1beta1.NodeManagementSpec{
					// Do it manually
					AutoUpgrade: ptr.To(false),
					AutoRepair:  ptr.To(false),
				},
			},
			ResourceSpec: xpv1.ResourceSpec{
				ProviderConfigReference: &xpv1.Reference{
					Name: pc,
				},
			},
		},
	}

	return []client.Object{kcnp}, nil
}

func (p *gcpProvider) nodePoolName(name string) string {
	hash := md5.Sum([]byte(name))
	hashString := hex.EncodeToString(hash[:])
	// prefix with `n` to avoid name starts with a number
	return fmt.Sprintf("n%s", hashString[:11])
}

func (p *gcpProvider) GetKubernetesID() string {
	return fmt.Sprintf("projects/%s/locations/%s/clusters/%s", p.gcpProjectID, p.region, p.kubernetesName)
}

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

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

func (p *gcpProvider) GetNodeSubnets() []string {
	// unused
	return []string{}
}

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

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

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

func (p *gcpProvider) 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{
		Gcs: &opv1alpha1.GcsStorageProvider{
			Bucket:     p.bucket,
			Prefix:     prefix,
			Location:   p.region,
			ProjectId:  p.gcpProjectID,
			SecretName: consts.BRGCPSecretName,
		},
	}
	backup.Spec.BR.SendCredToTikv = ptr.To(true)
	return nil
}

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