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
	region             string
	projectID          string
	kubernetesName     string
	kubeVersion        string
	nodeRole           string
	providerConfigName string

	// resource tag
	restoreTenantID  string
	restoreProjectID string

	snapshotBackupPrefix string
	logBackupPrefix      string
}

func NewGCPProvider(snapshotBucket string, logBackupBucket string, region string, projectID string, kubernetesName string, kubeVersion string, nodeRole string, providerConfigName string, restoreTenantID string, restoreProjectID string) (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 &gcpProvider{
		bucket:               bucket,
		region:               region,
		projectID:            projectID,
		kubernetesName:       kubernetesName,
		kubeVersion:          kubeVersion,
		nodeRole:             nodeRole,
		providerConfigName:   providerConfigName,
		restoreTenantID:      restoreTenantID,
		restoreProjectID:     restoreProjectID,
		snapshotBackupPrefix: snapshotBackupPrefix,
		// PITR needed
		logBackupPrefix: logBackupPrefix,
	}, 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.restoreTenantID,
		consts.TagKeyProject:        p.restoreProjectID,
		consts.TagKeyCluster:        consts.ClusterShared,
		consts.TagKeyUsedBy:         consts.UsedByCustomerTiDB,
		cloudv1.TagLabelServiceType: cloudv1.ServiceTypeDedicated,
		// used for node schedule
		namer.ComponentSelectorKey: string(cloudv1.ComponentRestore),
		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.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)},
							},
						},
					},
				},
			},
		},
	}
}

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.projectID, 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(cloudRestore *cloudv1.Restore, restore *opv1alpha1.Restore) error {
	bucketRegion := ptr.Deref(cloudRestore.Spec.From.Region, p.region)
	restore.Spec.StorageProvider = opv1alpha1.StorageProvider{
		Gcs: &opv1alpha1.GcsStorageProvider{
			ProjectId:  p.projectID,
			Bucket:     p.bucket,
			Prefix:     p.snapshotBackupPrefix,
			Location:   bucketRegion,
			SecretName: consts.BRGCPSecretName,
		},
	}
	restore.Spec.BR.SendCredToTikv = ptr.To(true)

	if restore.Spec.Mode == opv1alpha1.RestoreModePiTR {
		// log backup and snapshot backup use the same bucket
		restore.Spec.StorageProvider.Gcs.Prefix = p.logBackupPrefix
		restore.Spec.PitrFullBackupStorageProvider = opv1alpha1.StorageProvider{
			Gcs: &opv1alpha1.GcsStorageProvider{
				ProjectId:  p.projectID,
				Bucket:     p.bucket,
				Prefix:     p.snapshotBackupPrefix,
				Location:   bucketRegion,
				SecretName: consts.BRGCPSecretName,
			},
		}
	}

	return nil
}