package task

import (
	"context"
	"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
	compactBackupTenantID  string
	compactBackupProjectID string
}

func NewGCPProvider(bucket string, region string, gcpProjectID string, kubernetesName string, kubeVersion string, nodeRole string, providerConfigName string, compactBackupTenantID string, compactBackupProjectID 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,
		compactBackupTenantID:  compactBackupTenantID,
		compactBackupProjectID: compactBackupProjectID,
	}, 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.compactBackupTenantID,
		consts.TagKeyProject:        p.compactBackupProjectID,
		consts.TagKeyCluster:        consts.ClusterShared,
		consts.TagKeyUsedBy:         consts.UsedByCustomerTiDB,
		cloudv1.TagLabelServiceType: cloudv1.ServiceTypeDedicated,
		// used for node schedule
		namer.ComponentSelectorKey: string(cloudv1.ComponentCompactBackup),
		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.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)},
							},
						},
					},
				},
			},
		},
	}
}

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
}

// nodePoolName sanitizes a GKE node pool name.
// GKE node pool name constraints: lowercase alphanumeric and hyphens only,
// must start with a letter, max 63 chars.
// getNodePoolName() already produces a lowercase, letter-prefixed, ≤63-char name,
// so this function is a defensive guard against unexpected dots or underscores.
func (p *gcpProvider) nodePoolName(name string) string {
	name = strings.ReplaceAll(name, ".", "-")
	name = strings.ReplaceAll(name, "_", "-")
	return namer.NewNameHashOnlyExceed(name, "", "", 40)
}

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
}

// OverlayCompact implements storage overlay for CompactBackup
func (p *gcpProvider) OverlayCompact(compactBackup *opv1alpha1.CompactBackup) error {
	compactBackup.Spec.StorageProvider = opv1alpha1.StorageProvider{
		Gcs: &opv1alpha1.GcsStorageProvider{
			Bucket:     p.bucket,
			Prefix:     p.prefix,
			Location:   p.region,
			ProjectId:  p.gcpProjectID,
			SecretName: consts.BRGCPSecretName,
		},
	}
	compactBackup.Spec.BR.SendCredToTikv = ptr.To(true)
	return nil
}

// OverlayCompactStatus implements status overlay for CompactBackup
func (p *gcpProvider) 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
}