package cdc

import (
	"context"
	"fmt"
	"regexp"
	"sort"
	"strings"

	"github.com/crossplane-contrib/provider-gcp/apis/container/v1beta1"
	"gomodules.xyz/jsonpatch/v2"
	corev1 "k8s.io/api/core/v1"
	storagev1 "k8s.io/api/storage/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/utils/pointer"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
	"sigs.k8s.io/controller-runtime/pkg/log"

	eksv1alpha1 "github.com/tidbcloud/eks-provider/api/v1alpha1"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/cloud/gcp"
	compatnodepool "github.com/tidbcloud/infra-provider/pkg/compat/nodepool"
	"github.com/tidbcloud/infra-provider/pkg/config"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	"github.com/tidbcloud/infra-provider/pkg/controllers/common"
	pkgk8s "github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	"github.com/tidbcloud/infra-provider/pkg/nodegroup"
	"github.com/tidbcloud/infra-provider/pkg/nodepool"
	"github.com/tidbcloud/infra-provider/pkg/util"
)

const annoKeyResourceLabels = "gcp.cloud.pingcap.com/resource-labels"

var (
	// parse name from selfLink for request param
	// full selfLink: https://container.googleapis.com/v1/{name}
	containerSelfLinkToNameRe = regexp.MustCompile(`^https://container.googleapis.com/v1/(\S+)$`)
	// check if the key of a tag is valid
	tagKeyRegex = regexp.MustCompile("^[a-z][a-z0-9-_]{0,34}$")
)

type GcpCdc struct {
	Cdc
	gcpCli gcp.Client
}

func NewGcpCdc(base *Cdc) *GcpCdc {
	gcpCli, err := gcp.NewClientSet()
	if err != nil {
		panic(err)
	}

	return &GcpCdc{
		Cdc:    *base,
		gcpCli: gcpCli,
	}
}

func (p *GcpCdc) CreateCdcNodeGroup(ctx context.Context, ticdc *cloudv1.TiCDC, cls *cloudv1.Cluster, ngSet *nodegroup.Set) (map[string]string, error) {
	network, err := p.getCdcNetwork(ctx, ticdc)
	if err != nil {
		return nil, fmt.Errorf("failed find cdc located network: %v", err)
	}
	if len(ngSet.NodeGroups) == 0 {
		return nil, fmt.Errorf("no node group found")
	}

	ng := ngSet.NodeGroups[0]
	ng.Name = tidbNodeGroupName(ngSet.Name, string(ticdc.ObjectMeta.GetUID()))
	ngNamer := namer.NewNamer().LegacyWithNetworkName(network.Name).WithComponent(string(ng.ComponentType))
	ng.Labels = util.ConcatenateTags(ng.Labels, cloudv1.TagsFromLabels(network.Labels))
	ng.Labels = util.ConcatenateTags(ng.Labels, map[string]string{
		// override the network "usedby":"dbaas-control-plane-shoot" label
		common.UsedBy:      common.CustomerTiDB,
		common.Component:   string(cloudv1.ComponentTiCDC),
		common.Cluster:     ticdc.Spec.ClusterName,
		common.ServiceType: common.Dedicated,
		common.Environment: p.Config.GetEnvironment(),
	})
	ng.Labels = util.ConcatenateTags(ng.Labels, ngNamer.Labels())
	ng.MachineSpec.SystemVolumeSize = defaultSystemVolumeSize

	networkSpec, err := eksv1alpha1.GKENetworkSpecForNetwork(network)
	if err != nil {
		return nil, fmt.Errorf("unable to get network spec: %v", err)
	}
	if ng.Maximum == 0 {
		ng.Maximum = 1
	}
	np, npMut := compatnodepool.GetNodePoolByNGSpecAndNetWork(ng, network, networkSpec.ProjectID)
	result, err := controllerutil.CreateOrUpdate(ctx, p.Cli, np, npMut)
	if err != nil {
		return nil, err
	}
	if result == controllerutil.OperationResultCreated {
		log.FromContext(ctx).Info(
			"create gcp node pool for ticdc successfully", "ticdc", ticdc.Name, "node pool", np)
	} else if result == controllerutil.OperationResultUpdated {
		log.FromContext(ctx).Info(
			"update gcp node pool for ticdc successfully", "ticdc", ticdc.Name, "node pool", np)
	}
	return ng.Labels, nil

}

func (p *GcpCdc) DeleteCdcNodeGroup(ctx context.Context, ticdc *cloudv1.TiCDC, ngSet *nodegroup.Set) error {
	network, err := p.getCdcNetwork(ctx, ticdc)
	if err != nil {
		return err
	}
	specName := tidbNodeGroupName(ngSet.Name, string(ticdc.ObjectMeta.GetUID()))
	return nodepool.DeleteNodePoolBySpecNameAndNetWork(ctx, p.Cli, specName, network)

}

func (p *GcpCdc) ReleaseNodes(ctx context.Context, cr *cloudv1.TiCDC, ngNamePrefix string, labelSet labels.Set) error {
	set := &nodegroup.Set{
		Name: ngNamePrefix,
	}
	if err := p.DeleteCdcNodeGroup(ctx, cr, set); client.IgnoreNotFound(err) != nil {
		return err
	}

	// protect delete label set not specify ticdc.cr.id
	if _, exist := labelSet[TiCDCCrIdLabel]; !exist {
		labelSet[TiCDCCrIdLabel] = string(cr.UID)
	}
	// delete all nodegroup of target ticdc cluster
	if err := p.Cli.DeleteAllOf(ctx, &v1beta1.NodePool{}, client.MatchingLabelsSelector{
		Selector: labelSet.AsSelector(),
	}); client.IgnoreNotFound(err) != nil {
		return err
	}

	return nil
}

func (p *GcpCdc) UpdateNodegroupTopology(ctx context.Context, cr *cloudv1.TiCDC, topology map[string]int) error {

	var list v1beta1.NodePoolList
	if err := p.Cli.List(ctx, &list, client.MatchingLabels{
		TiCDCCrIdLabel:    string(cr.UID),
		TiCDCClusterLabel: *cr.Spec.CDCClusterID,
	}); client.IgnoreNotFound(err) != nil {
		return err
	}
	// ensure current exist one group np
	if len(list.Items) != DefaultTiCDCZoneNum {
		return fmt.Errorf("current np num %d is not equal to default ticdc zone %d", len(list.Items), DefaultTiCDCZoneNum)
	}
	for _, np := range list.Items {
		az, exist := np.Labels[TiCDCZoneLabel]
		if !exist {
			return fmt.Errorf("np %s has a empty zone label", np.Name)
		}
		num, exist := topology[az]
		if !exist {
			return fmt.Errorf("topology has no az %s info", az)
		}
		np.Spec.ForProvider.Autoscaling.MinNodeCount = pointer.Int64(int64(num))
		np.Spec.ForProvider.Autoscaling.MaxNodeCount = pointer.Int64(int64(num))
		if num == 0 {
			np.Spec.ForProvider.Autoscaling.MaxNodeCount = pointer.Int64(1)
		}
		if err := p.Cli.Update(ctx, &np); err != nil {
			return err
		}
	}

	return nil
}

func (p *GcpCdc) SetResourceLabels(ctx context.Context, cr *cloudv1.TiCDC) error {
	var nodePoolList v1beta1.NodePoolList
	if err := p.Cli.List(ctx, &nodePoolList, client.MatchingLabels{
		TiCDCCrIdLabel:    string(cr.UID),
		TiCDCClusterLabel: *cr.Spec.CDCClusterID,
	}); client.IgnoreNotFound(err) != nil {
		return err
	}

	logger := log.FromContext(ctx, "ticdc", fmt.Sprintf("%s/%s", cr.Namespace, cr.Name))
	for _, np := range nodePoolList.Items {
		status := np.Status.AtProvider.Status
		if status != v1beta1.NodePoolStateRunning {
			logger.Info("node pool is not running, skip it", "node_pool_name", np.Name)
			continue
		}

		_, ok := np.Annotations[annoKeyResourceLabels]
		if ok {
			continue
		}

		// updating resourceLabels after they are running
		// v1beta1.NodeConfig not support resourceLabels, set it manually

		// the name of nodePool must be specified in the format
		// 'projects/*/locations/*/clusters/*/nodePools/*'
		nodePoolName, err := parseFullNodePoolNameFromSelfLink(np.Status.AtProvider.SelfLink)
		if err != nil {
			return fmt.Errorf("node pool %s parse self link %s error: %w",
				np.Name, np.Status.AtProvider.SelfLink, err)
		}
		ls, err := p.gcpCli.GetNodePoolResourceLabels(ctx, nodePoolName)
		if err != nil {
			return fmt.Errorf("node pool %s get resource labels error: %w", np.Name, err)
		}

		changed := false
		customTags := getValidTags(np.Spec.ForProvider.Config.Labels)
		for k, v := range customTags {
			if _, ok := ls[k]; !ok {
				ls[k] = v
				changed = true
			}
		}

		if !changed {
			// set anno to avoid GCP api call
			patch := newResourceLabelsPatch(ls)
			if err := pkgk8s.PatchObject(ctx, p.Cli, &np, patch); err != nil {
				return fmt.Errorf("node pool %s record resource labels in anno error: %w", np.Name, err)
			}
			continue
		}

		// only set resource labels if they are not existing in gke node pools
		_, err = p.gcpCli.UpdateNodePoolResourceLabels(ctx, nodePoolName, ls)
		if err != nil {
			return fmt.Errorf("node pool %s update resource labels error: %w", np.Name, err)
		}
		logger.Info("set resource label to node pool successfully", "node_pool_name", np.Name)
	}
	return nil
}

func (p *GcpCdc) EnsureAllNodegroupsReady(ctx context.Context, cr *cloudv1.TiCDC) error {
	var list v1beta1.NodePoolList
	if err := p.Cli.List(ctx, &list, client.MatchingLabels{
		TiCDCCrIdLabel:    string(cr.UID),
		TiCDCClusterLabel: *cr.Spec.CDCClusterID,
	}); client.IgnoreNotFound(err) != nil {
		return err
	}
	for _, np := range list.Items {
		if np.Status.AtProvider.Status != v1beta1.NodePoolStateRunning {
			return fmt.Errorf("wait nodepool %s status %s, current status %s", np.Name, v1beta1.NodePoolStateRunning, np.Status.AtProvider.Status)
		}
	}

	return nil
}

func (p *GcpCdc) DeleteNodegroupsByFilter(ctx context.Context, cr *cloudv1.TiCDC, selector labels.Selector) error {
	var npList v1beta1.NodePoolList
	err := p.Cli.List(ctx, &npList, &client.MatchingLabelsSelector{Selector: selector})
	if err != nil {
		return err
	}
	for _, np := range npList.Items {
		if np.Labels[TiCDCCrIdLabel] != string(cr.UID) {
			continue
		}
		group := &v1beta1.NodePool{
			ObjectMeta: metav1.ObjectMeta{
				Name: np.Name,
			},
		}
		if err := p.Cli.Delete(ctx, group, &client.DeleteOptions{}); err != nil {
			return err
		}
	}
	return nil
}

func (p *GcpCdc) GetNodegroupsByFilter(ctx context.Context, cr *cloudv1.TiCDC, selector labels.Selector) (map[string]int, error) {
	topology := make(map[string]int)
	var npList v1beta1.NodePoolList
	err := p.Cli.List(ctx, &npList, &client.MatchingLabelsSelector{Selector: selector})
	if err != nil {
		return nil, err
	}
	for _, np := range npList.Items {
		zone, exist := np.Labels[TiCDCZoneLabel]
		if !exist {
			return nil, fmt.Errorf("current node pool %s dose not have zone lable", np.Name)
		}
		num := *np.Spec.ForProvider.Autoscaling.MinNodeCount
		topology[zone] = int(num)
	}
	return topology, nil
}

func (p *GcpCdc) Credentials(ctx context.Context, network *cloudv1.Network) ([]byte, error) {
	saKeyName := namer.LegacyBuildServiceAccountKeyNameByNetwork(network)
	saKeySecret := corev1.Secret{}
	err := p.Cli.Get(ctx, types.NamespacedName{Namespace: network.Namespace, Name: saKeyName}, &saKeySecret)
	if err != nil {
		return nil, err
	}

	return saKeySecret.Data[GcpCredentialsKey], nil
}

func (p *GcpCdc) IsDPCompatMode() bool {
	return config.FeatureEnabled(p.Config.FeatureGates, config.EnableDPCompatibilityMode, "", "")
}

func (p *GcpCdc) BuildStorageClass() *storagev1.StorageClass {
	bindingMode := storagev1.VolumeBindingWaitForFirstConsumer
	reclaimPolicy := corev1.PersistentVolumeReclaimDelete

	return &storagev1.StorageClass{
		ObjectMeta: metav1.ObjectMeta{
			Name: consts.TiCDCGp3StorageClassName,
		},
		VolumeBindingMode:    &bindingMode,
		AllowVolumeExpansion: pointer.Bool(true),
		ReclaimPolicy:        &reclaimPolicy,
		Provisioner:          "pd.csi.storage.gke.io",
		Parameters: map[string]string{
			"csi.storage.k8s.io/fstype": "ext4",
			"type":                      consts.GCPVolumeTypePDBalanced,
		},
		MountOptions: []string{"nodelalloc,noatime"},
	}
}

// parseFullNodePoolNameFromSelfLink get the name like 'projects/*/locations/*/clusters/*/nodePools/*' from self link
func parseFullNodePoolNameFromSelfLink(selfLink string) (name string, err error) {
	strs := containerSelfLinkToNameRe.FindStringSubmatch(selfLink)
	if len(strs) != 2 {
		return "", fmt.Errorf("parse nodepool name from selflink error, parsed strings: %q", strs)
	}
	return strs[1], nil
}

func newResourceLabelsPatch(ls map[string]string) pkgk8s.ObjectPatch {
	var kvs []string
	for k, v := range ls {
		kvs = append(kvs, k+":"+v)
	}
	sort.Strings(kvs)

	return func(obj client.Object) *jsonpatch.Operation {
		anno, op := obj.GetAnnotations(), "replace"
		if anno == nil || anno[annoKeyResourceLabels] == "" {
			op = "add"
		}
		return &jsonpatch.Operation{
			Operation: op,
			Path:      `/metadata/annotations/` + strings.ReplaceAll(annoKeyResourceLabels, "/", "~1"),
			Value:     strings.Join(kvs, ","),
		}
	}
}

func getValidTags(ls map[string]string) map[string]string {
	validTags := make(map[string]string, len(ls))
	for k, v := range ls {
		if tagKeyRegex.MatchString(k) {
			validTags[k] = v
		}
	}
	return validTags
}