package backup

import (
	"fmt"
	"sort"
	"strings"

	"github.com/Masterminds/semver"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"github.com/tidbcloud/config-service/sdk/api"
	v1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/util/json"

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

const (
	OptionWithSystemTable  = "--with-sys-table"
	DisableSysTableRestore = "DisableSysTableRestore"
)

func BRConfig(output *api.DescribeBRConfigOutput) (*v1alpha1.BRConfig, error) {
	jsonBody, err := json.Marshal(output.Config)
	if err != nil {
		return nil, err
	}
	var brConfig v1alpha1.BRConfig
	err = json.Unmarshal(jsonBody, &brConfig)
	if err != nil {
		return nil, err
	}
	return &brConfig, nil
}

func addRestoreOptionOldVersion(cr *cloudv1.Restore, restore *v1alpha1.Restore) *v1alpha1.Restore {
	images := strings.Split(restore.Spec.ToolImage, ":")
	if len(images) != 2 {
		return restore
	}
	tag := images[1]
	if checkSupportSysTableRestore(tag) && (!checkDisableSysTableRestore(cr)) {
		restore.Spec.BR.Options = append(restore.Spec.BR.Options, OptionWithSystemTable)
		restore.Spec.BR.Options = append(restore.Spec.BR.Options, "--filter=*.*")
		restore.Spec.BR.Options = append(restore.Spec.BR.Options, "--filter=!__TiDB_BR_Temporary_*.*")
		restore.Spec.BR.Options = append(restore.Spec.BR.Options, "--filter=!mysql.*")
		restore.Spec.BR.Options = append(restore.Spec.BR.Options, "--filter=mysql.user")
		restore.Spec.BR.Options = append(restore.Spec.BR.Options, "--filter=mysql.db")
		restore.Spec.BR.Options = append(restore.Spec.BR.Options, "--filter=mysql.tables_priv")
		restore.Spec.BR.Options = append(restore.Spec.BR.Options, "--filter=mysql.columns_priv")
		restore.Spec.BR.Options = append(restore.Spec.BR.Options, "--filter=mysql.global_priv")
		restore.Spec.BR.Options = append(restore.Spec.BR.Options, "--filter=mysql.global_grants")
		restore.Spec.BR.Options = append(restore.Spec.BR.Options, "--filter=mysql.default_roles")
		restore.Spec.BR.Options = append(restore.Spec.BR.Options, "--filter=mysql.role_edges")
		restore.Spec.BR.Options = append(restore.Spec.BR.Options, "--filter=!sys.*")
		restore.Spec.BR.Options = append(restore.Spec.BR.Options, "--filter=!INFORMATION_SCHEMA.*")
		restore.Spec.BR.Options = append(restore.Spec.BR.Options, "--filter=!PERFORMANCE_SCHEMA.*")
		restore.Spec.BR.Options = append(restore.Spec.BR.Options, "--filter=!METRICS_SCHEMA.*")
		restore.Spec.BR.Options = append(restore.Spec.BR.Options, "--filter=!INSPECTION_SCHEMA.*")
		restore.Spec.BR.Options = append(restore.Spec.BR.Options, "--filter=mysql.bind_info")
	}
	return restore
}

func AddRestoreOption(cr *cloudv1.Restore, restore *v1alpha1.Restore) *v1alpha1.Restore {
	if len(cr.Spec.Options) == 0 {
		return addRestoreOptionOldVersion(cr, restore)
	}

	restore.Spec.BR.Options = append(restore.Spec.BR.Options, cr.Spec.Options...)
	return restore
}

func checkSupportSysTableRestore(tag string) bool {
	versions := strings.Split(tag, "-")
	sysTableCheck, _ := semver.NewConstraint(">= v6.2.0")
	if curVersion, err := semver.NewVersion(versions[0]); err == nil {
		return sysTableCheck.Check(curVersion)
	}
	return false
}

func checkDisableSysTableRestore(restore *cloudv1.Restore) bool {
	if restore.Annotations == nil {
		return false
	}
	if value, exist := restore.Annotations[DisableSysTableRestore]; exist && value == "true" {
		return true
	}
	return false
}

func commonAnnotationsForBR() map[string]string {
	// do not evit workload pods with cluster-autoscaler, https://github.com/kubernetes/autoscaler/tree/master/cluster-autoscaler
	return map[string]string{
		"cluster-autoscaler.kubernetes.io/safe-to-evict": "false",
	}
}

func backupNodeGroupName(b *cloudv1.Backup, inNewArch bool, isGKE bool) string {
	if isGKE {
		name := fmt.Sprintf("%s-%s", b.Namespace, b.Name)
		hash := util.FnvHash(name)
		return fmt.Sprint("backup-", hash)
	}
	name := fmt.Sprint("backup-", b.Name)
	if !inNewArch {
		name = util.FnvHash(name)
	}
	return name
}

func backupNodeAffinityKey(b *cloudv1.Backup) string {
	return fmt.Sprintf("affinity-key-backup-%s", util.FnvHash(b.Name))
}

func backupNodeAffinityValue(b *cloudv1.Backup) string {
	return fmt.Sprintf("affinity-value-backup-%s", util.FnvHash(b.Name))
}

func buildBackupLabels(b *cloudv1.Backup) map[string]string {
	//labels := utils.CopyLabelsForWorkloads(b.Labels)
	labels := cloudv1.TagsFromLabels(b.Labels)
	// this label is to make sure node affinity is satisfied
	labels[backupNodeAffinityKey(b)] = backupNodeAffinityValue(b)
	return labels
}

func getSortedInstanceTypes(provider *cloudv1.Provider) []cloudv1.InstanceType {
	sortedInstance := provider.Spec.DeepCopy().InstanceTypes
	sort.Slice(sortedInstance, func(i, j int) bool {
		// when memory is equal, sort by cpu
		if sortedInstance[i].Memory.Cmp(sortedInstance[j].Memory) == 0 {
			return sortedInstance[i].CPU.Cmp(sortedInstance[j].CPU) < 0
		}
		// sort instance types by memory
		return sortedInstance[i].Memory.Cmp(sortedInstance[j].Memory) < 0
	})
	return sortedInstance
}

func inSameRegion(region *cloudv1.Region, regions []string) bool {
	for _, r := range regions {
		if r == region.Name {
			return true
		}
	}
	return false
}

func getMachineSpecByResourceRequirements(
	provider *cloudv1.Provider, region *cloudv1.Region, resourceRequirements *v1.ResourceRequirements) *cloudv1.MachineSpec {
	machine := &cloudv1.MachineSpec{}
	cpuLimit := resourceRequirements.Limits.Cpu()
	cpuRequest := resourceRequirements.Requests.Cpu()
	memoryLimit := resourceRequirements.Limits.Memory()
	memoryRequest := resourceRequirements.Requests.Memory()
	// find the instance that meets the minimum requirements of cpu and memory in provider.Spec.InstanceTypes
	// and the instance is in the region of the cluster
	for _, ins := range getSortedInstanceTypes(provider) {
		if ins.CPU.Cmp(cpuRequest.DeepCopy()) > 0 && ins.CPU.Cmp(cpuLimit.DeepCopy()) >= 0 &&
			ins.Memory.Cmp(memoryRequest.DeepCopy()) > 0 && ins.Memory.Cmp(memoryLimit.DeepCopy()) >= 0 &&
			ins.Arch == consts.AMD64ArchName && inSameRegion(region, ins.Regions) {
			machine.MachineType = ins.Name
			machine.SystemVolumeSize = "50Gi"
			machine.MachineArch = ins.Arch
			break
		}
	}
	if machine.MachineType == "" {
		return nil
	}
	return machine
}

// create a node group with tainted node for backup, taints contains backup name
func buildBackupNodeGroupSpecWithTaints(
	ngName string,
	b *cloudv1.Backup,
	provider *cloudv1.Provider,
	region *cloudv1.Region) (*cloudv1.NodegroupSpec, error) {
	machine := getMachineSpecByResourceRequirements(provider, region, b.Spec.ResourceRequirements)
	if machine == nil {
		return nil, fmt.Errorf("no instance type found for backup")
	}
	ngSpec := &cloudv1.NodegroupSpec{
		MachineSpec:   machine,
		Zones:         region.Zones,
		Labels:        buildBackupLabels(b),
		ComponentType: cloudv1.ComponentBackup,
		Name:          ngName,
		Taints:        utils.TaintWithUk(cloudv1.ComponentBackup, b.Name),
		Maximum:       1,
		Minimum:       1,
		NodeRole:      b.Spec.NodeRole,
	}
	return ngSpec, nil
}

func restoreNodeGroupName(b *cloudv1.Restore, inNewArch bool) string {
	name := fmt.Sprint("restore-", b.Name)
	if !inNewArch {
		name = util.FnvHash(name)
	}
	return name
}

func restoreNodeAffinityKey(b *cloudv1.Restore) string {
	return fmt.Sprintf("affinity-key-restore-%s", util.FnvHash(b.Name))
}

func restoreNodeAffinityValue(b *cloudv1.Restore) string {
	return fmt.Sprintf("affinity-value-restore-%s", util.FnvHash(b.Name))
}

func buildRestoreLabels(b *cloudv1.Restore) map[string]string {
	//labels := utils.CopyLabelsForWorkloads(b.Labels)
	labels := cloudv1.TagsFromLabels(b.Labels)
	// this label is to make sure node affinity is satisfied
	labels[restoreNodeAffinityKey(b)] = restoreNodeAffinityValue(b)
	return labels
}

// create a node group with tainted node for restore, taints contains restore name
func buildRestoreNodeGroupSpecWithTaints(
	ngName string,
	r *cloudv1.Restore,
	provider *cloudv1.Provider,
	region *cloudv1.Region) (*cloudv1.NodegroupSpec, error) {
	machine := getMachineSpecByResourceRequirements(provider, region, r.Spec.ResourceRequirements)
	if machine == nil {
		return nil, fmt.Errorf("no instance type found for restore")
	}
	ngSpec := &cloudv1.NodegroupSpec{
		MachineSpec:   machine,
		Zones:         []string{region.Zones[0]}, // only one zone is enough
		Labels:        buildRestoreLabels(r),
		ComponentType: cloudv1.ComponentRestore,
		Name:          ngName,
		Taints:        utils.TaintWithUk(cloudv1.ComponentRestore, r.Name),
		Maximum:       1,
		Minimum:       1,
		NodeRole:      r.Spec.NodeRole,
	}
	return ngSpec, nil
}

func networkInNewArch(network *cloudv1.Network) bool {
	return network.Spec.Provider == consts.ProviderTypeEKS || network.Spec.Provider == consts.ProviderTypeGKE
}