package ticdc

import (
	"errors"
	"fmt"
	"sort"
	"strconv"
	"strings"

	corev1 "k8s.io/api/core/v1"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	infrautil "github.com/tidbcloud/infra-api/pkg/util"
)

const (
	TiCDCCrIdLabel           = "ticdc.cr.id"
	TiCDCClusterLabel        = "ticdc.cluster.id"
	TiCDCZoneLabel           = "ticdc.zone"
	TiCDCResourceKeyLabel    = "ticdc.resource.key"
	TiCDCScalingDisableLabel = "ticdc.scaling.disable"
	TiCDCNodegroupNameLabel  = "Name"

	TiCDCDefaultDirName = "sort-kv"
	TiCDCDefaultDirPath = "/var/lib/ticdc"
	TiCDCFinalizer      = "ticdc.cloud.pingcap.com/finalizer"

	DefaultTiCDCZoneNum = 2
	DefaultMemoryQuota  = 20971520
	// FIXME: remove these hardcode
	DefaultTiCDCVersion = "v5.3.0-20211129"
	DefaultTiCDCImage   = "gcr.io/pingcap-public/dbaas/ticdc"
	// default gc ttl is 3 days, per table memory is 20M
	DefaultTiCDCGcTTL                 = 259200
	DefaultTiCDCDefaultMemoryPerTable = 20971520

	GcpCredentialsKey  = "privateKey"
	GcpCredentialsBase = "/root/gcp"
	GcpCredentialsPath = "key.json"
)

var predefinedConfigKeys = []string{
	"cluster-id",
	"data-dir",
	"log-file",
}

var kubeDNSConf = `
{
  tidb%s.svc.cluster.local: [
	%s
  ]
}
`

var coreDNSConf = `
.:%d {
	log
	errors
	health
	ready
	kubernetes cluster.local in-addr.arpa ip6.arpa {
	   pods insecure
	   fallthrough in-addr.arpa ip6.arpa
	}
    prometheus :9153
    forward . /etc/resolv.conf
	cache 30
	loop
	reload
	loadbalance round_robin
}
tidb%s.svc.cluster.local:%d {
	errors
	cache 30
	forward . %s {
	    %s
	}
}
`

type NgFilter struct {
	SpecifyLabels map[string]string
	ExcludeLabels map[string]string
}

func buildTopology(zones []string, replicaNum int) (map[string]int, error) {
	if len(zones) <= 0 || replicaNum <= 0 {
		return nil, errors.New("invalid zone num or replica num of ticdc")
	}

	nodeArray := make([]int, len(zones))
	index := 0
	for i := 0; i < replicaNum; i++ {
		if index == len(zones) {
			index = 0
		}
		nodeArray[index]++
		index++
	}

	topology := make(map[string]int)
	for i := 0; i < len(zones); i++ {
		topology[zones[i]] = nodeArray[i]
	}
	return topology, nil
}

type zoneNodeInfo struct {
	zone    string
	nodeNum int
}

/*
calculate desired topology of map[zone]nodeNum from current to desiredReplicas
will return topology of balance zone node to cover desired replicas
*/
func calculateDesiredTopology(current map[string]int, desiredReplicas int) map[string]int {
	curReplicas := 0
	var infos []zoneNodeInfo
	for az, num := range current {
		curReplicas += num
		infos = append(infos, zoneNodeInfo{zone: az, nodeNum: num})
	}

	if curReplicas == desiredReplicas {
		return current
	}

	sort.Slice(infos, func(i, j int) bool {
		// descending order
		return infos[i].nodeNum > infos[j].nodeNum
	})

	curIndex := 0
	curNum := 0
	for index := 0; index < len(infos); index++ {
		if curNum == 0 {
			curNum = infos[index].nodeNum
		}
		if curNum > infos[index].nodeNum {
			curIndex = index
			break
		}
	}

	if desiredReplicas > curReplicas {
		addIndex := curIndex
		for cnt := 0; cnt < desiredReplicas-curReplicas; cnt++ {
			if addIndex == len(infos) {
				addIndex = 0
			}
			infos[addIndex].nodeNum++
			addIndex++
		}
	} else {
		deleteIndex := curIndex - 1
		for cnt := 0; cnt < curReplicas-desiredReplicas; cnt++ {
			if deleteIndex < 0 {
				deleteIndex = len(infos) - 1
			}
			infos[deleteIndex].nodeNum--
			deleteIndex--
		}
	}

	desired := make(map[string]int)
	for _, info := range infos {
		desired[info.zone] = info.nodeNum
	}
	return BalanceTopology(desired)
}

func BalanceTopology(topo map[string]int) map[string]int {
	if len(topo) == 0 {
		return topo
	}
	max := 0
	balanceTopo := make(map[string]int)
	for _, v := range topo {
		if v > max {
			max = v
		}
	}
	for k := range topo {
		balanceTopo[k] = max
	}
	return balanceTopo
}

func TiCDCNodeGroupName(cr *cloudv1.TiCDC, zone string) string {
	return "cdc-" + infrautil.FnvHash(fmt.Sprintf("%s-%s-%s", cr.UID, zone, strings.ReplaceAll(TiCDCResourceKey(cr.Spec.InstanceType, cr.Spec.Replicas), ".", "-")))
}

func TiCDCResourceKey(instanceType string, replicas int) string {
	return fmt.Sprintf("%s_%d", instanceType, replicas)
}

func TiCDCResourceValue(key string) (instanceType string, replicas int, err error) {
	strArr := strings.Split(key, "_")
	if len(strArr) != 2 {
		return "", 0, fmt.Errorf("resource key %s invalid", key)
	}
	instanceType = strArr[0]
	replicas, err = strconv.Atoi(strArr[1])
	return
}

type AccountCredentialData struct {
	volume corev1.Volume
	mount  corev1.VolumeMount
	env    corev1.EnvVar
}

func TiCDCGcpCredentialsName(cr *cloudv1.TiCDC) string {
	return fmt.Sprintf("cdc-gcp-credentials-%s", cr.UID)
}