package sdk

import (
	"context"
	"fmt"
	"os"
	"sort"
	"strings"

	"github.com/pingcap/errors"
	"golang.org/x/mod/semver"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"
	clog "sigs.k8s.io/controller-runtime/pkg/log"

	v1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/gitops-sdk/configeval"
	"github.com/tidbcloud/infra-provider/pkg/gitops-sdk/consts"
	"github.com/tidbcloud/infra-provider/util/collections"
	"github.com/tidbcloud/infra-provider/util/config"
)

// TODO: It is hard-coded temporarily, later serverless will have a separate config to set the Lightning image
const minServerlessLigntningVersion string = "v6.6.0"

// CCInterface is an interface to get ClusterConfig for a cluster
type CCInterface interface {
	DescribeClusterConfiguration(cluster *v1.Cluster) (*v1.ClusterConfig, error)
}

type DescribeClusterConfigurationFunc func(cluster *v1.Cluster) (*v1.ClusterConfig, error)

func (f DescribeClusterConfigurationFunc) DescribeClusterConfiguration(cluster *v1.Cluster) (*v1.ClusterConfig, error) {
	return f(cluster)
}

type Gitops struct {
	Cli client.Client
}

func (g *Gitops) DescribeClusterConfiguration(cluster *v1.Cluster) (*v1.ClusterConfig, error) {
	ls := cluster.GetLabels()
	ccl, err := g.allClusterConfig()
	if err != nil {
		return nil, err
	}

	tenantAndClusterLayer, err := g.matchClusterConfig(ccl, labels.Set(ls))
	if err != nil {
		return nil, err
	}

	clusterAndTenantOverlay, err := mergeClusterConfig(tenantAndClusterLayer)
	if err != nil {
		return nil, err
	}
	labels := map[string]string{
		v1.TenantTagKey:  cluster.GetTenantID(),
		v1.ProjectTagKey: cluster.GetProjectID(),
		// NOTE: Should use cluster.GetClusterID() instead
		// But now just keep compatibility
		v1.ClusterTagKey:     cluster.Name,
		v1.ProviderTypeLabel: cluster.GetProviderType(),
	}
	if clusterAndTenantOverlay != nil {
		for _, l := range tenantAndClusterLayer {
			labels = collections.MergeMaps(l.Spec.Selector.MatchLabels, labels)
		}
	}
	version := getTargetVersion(cluster, clusterAndTenantOverlay)
	baseObj, err := g.getBaseClusterConfig(ccl, version)
	if err != nil {
		return nil, errors.Annotatef(err, "get base cluster config for version '%s' failed", version)
	}
	if version == "" {
		version = baseObj.Spec.Version
	}
	labels[v1.VersionLabel] = version

	instanceTypeOverlay, err := g.InstanceTypeLevelOverlay(ccl, version, getAllInstanceType(cluster))
	if err != nil {
		return nil, err
	}

	if cluster.GetProviderType() == consts.ProviderTypeAWSFreeTier && semver.Compare(version, minServerlessLigntningVersion) < 0 {
		// replace base object's lightning config for serverless tier
		// NOTE: here we don't use the provider overlay, but to simply replace the whole Lightning config,
		// because here the Lightning is a whole version change, rather than just changing some configs.
		// There might be some configs that is removed in the new Lightning version,
		// and overlaying the new values won't remove those configs.
		clog.Log.Info("Lightning version doesn't meet the minimum version requirement. Need to change Lightning version", "current_version", version, "new_version", minServerlessLigntningVersion)
		lnConfig, err := g.getLightningConfig(ccl, minServerlessLigntningVersion)
		if err != nil {
			clog.Log.Error(err, "get new lightning config failed", "version", minServerlessLigntningVersion)
			// no need to do any further error handling besides log the error
		} else {
			clog.Log.Info("retrieved new Lightning config", "new_version", minServerlessLigntningVersion, "lightning_config", lnConfig)
			baseObj.Spec.Lightning = lnConfig
		}
	}

	var selectedClusterconfig []v1.ClusterConfig
	selectedClusterconfig = append(selectedClusterconfig, *baseObj)
	selectedClusterconfig = append(selectedClusterconfig, *g.ProviderLevelOverlay(cluster.GetProviderType()))
	selectedClusterconfig = append(selectedClusterconfig, *instanceTypeOverlay)
	if clusterAndTenantOverlay != nil {
		selectedClusterconfig = append(selectedClusterconfig, *clusterAndTenantOverlay)
	}

	cc, err := mergeClusterConfig(selectedClusterconfig)
	if err != nil {
		return nil, err
	}
	cc.ObjectMeta.Labels = labels
	err = configeval.EvalClusterConfig(g.Cli, cc, cluster)
	if err != nil {
		return nil, err
	}
	return cc, nil
}

func getClusterVersion(cluster *v1.Cluster) string {
	if cluster == nil {
		return ""
	}
	dbStatus := cluster.Status.DB
	if dbStatus == nil {
		return ""
	}
	return dbStatus.Version
}

func getTargetVersion(cluster *v1.Cluster, clusterAndTenantOverlay *v1.ClusterConfig) string {
	if cluster.GetProviderType() != consts.ProviderTypeAWSFreeTier && clusterAndTenantOverlay != nil {
		customVersion := clusterAndTenantOverlay.Spec.Version
		if len(customVersion) > 0 {
			return customVersion
		}
	}
	return getClusterVersion(cluster)
}

func (g *Gitops) allClusterConfig() (*v1.ClusterConfigList, error) {
	ccs := new(v1.ClusterConfigList)
	err := g.Cli.List(context.TODO(), ccs)
	if err != nil {
		return nil, err
	}
	return ccs, nil
}

func (g *Gitops) matchClusterConfig(ccl *v1.ClusterConfigList, labels labels.Labels) ([]v1.ClusterConfig, error) {
	var res []v1.ClusterConfig
	for _, cc := range ccl.Items {
		selector, err := metav1.LabelSelectorAsSelector(cc.Spec.Selector)
		if err != nil {
			return nil, err
		}

		if selector.Matches(labels) {
			res = append(res, cc)
		}
	}

	return res, nil
}

func getAllInstanceType(cluster *v1.Cluster) map[v1.ComponentType]string {
	providerType := cluster.Labels[v1.ProviderTypeLabel]

	if providerType == v1.ProviderTypeEKS {
		providerType = v1.ProviderTypeAWS
	}

	if providerType == v1.ProviderTypeGKE {
		providerType = v1.ProviderTypeGCP
	}

	res := make(map[v1.ComponentType]string, 4)

	if cluster.Spec.Cluster.TiFlash.InstanceType != "" {
		res[v1.ComponentTiFlash] = fmt.Sprintf("%s.%s", providerType, cluster.Spec.Cluster.TiFlash.InstanceType)
	}
	if cluster.Spec.Cluster.TiProxy.InstanceType != "" {
		res[v1.ComponentTiProxy] = fmt.Sprintf("%s.%s", providerType, cluster.Spec.Cluster.TiProxy.InstanceType)
	}
	if cluster.Spec.Cluster.TiDB.InstanceType != "" {
		res[v1.ComponentTiDB] = fmt.Sprintf("%s.%s", providerType, cluster.Spec.Cluster.TiDB.InstanceType)
	}
	if cluster.Spec.Cluster.TiKV.InstanceType != "" {
		res[v1.ComponentTiKV] = fmt.Sprintf("%s.%s", providerType, cluster.Spec.Cluster.TiKV.InstanceType)
	}
	if cluster.Spec.Cluster.PD.InstanceType != "" {
		res[v1.ComponentPD] = fmt.Sprintf("%s.%s", providerType, cluster.Spec.Cluster.PD.InstanceType)
	}
	return res
}

func (g *Gitops) getBaseClusterConfig(ccl *v1.ClusterConfigList, version string) (*v1.ClusterConfig, error) {
	// query label that contains the specified version
	sets := labels.Set{}
	sets[v1.VersionLabel] = version
	ccs, err := g.matchClusterConfig(ccl, sets)
	if err != nil {
		return nil, err
	}

	// if the specified version is not found, query label that contains the default version
	if len(ccs) == 0 {
		sets := labels.Set{}
		sets[v1.DefaultVersionLabel] = "true"
		ccs, err := g.matchClusterConfig(ccl, sets)
		if err != nil {
			return nil, err
		}
		if len(ccs) != 1 {
			return nil, fmt.Errorf("should be only one base clusterconfig, but now: %d", len(ccs))
		}
		return &ccs[0], nil
	}

	if len(ccs) != 1 {
		return nil, fmt.Errorf("should be only one base clusterconfig, but now: %d", len(ccs))
	}

	return &ccs[0], nil
}

func mergeClusterConfig(configs []v1.ClusterConfig) (*v1.ClusterConfig, error) {
	if len(configs) == 0 {
		return nil, nil
	}
	sort.Slice(configs, func(i, j int) bool {
		return configs[i].Spec.Priority <= configs[j].Spec.Priority
	})
	if !isMetaConsistent(configs) {
		return nil, fmt.Errorf("ClusterConfigList is not fully updated")
	}

	base := new(v1.ClusterConfig)
	for _, overlay := range configs {
		if overlay.Spec.Version != "" {
			base.Spec.Version = overlay.Spec.Version
		}
		if overlay.Spec.UpdatePolicy != "" {
			base.Spec.UpdatePolicy = overlay.Spec.UpdatePolicy
		}

		base.Spec.ConfigMeta = mergeConfigMeta(base.Spec.ConfigMeta, overlay.Spec.ConfigMeta)
		base.Spec.TiDB = mergeConfiguration(base.Spec.TiDB, overlay.Spec.TiDB)
		base.Spec.TiKV = mergeConfiguration(base.Spec.TiKV, overlay.Spec.TiKV)
		base.Spec.PD = mergeConfiguration(base.Spec.PD, overlay.Spec.PD)
		base.Spec.TiFlash = mergeConfiguration(base.Spec.TiFlash, overlay.Spec.TiFlash)
		base.Spec.TiProxy = mergeConfiguration(base.Spec.TiProxy, overlay.Spec.TiProxy)
		base.Spec.TiFlashLearner = mergeConfiguration(base.Spec.TiFlashLearner, overlay.Spec.TiFlashLearner)
		base.Spec.Restore = mergeConfiguration(base.Spec.Restore, overlay.Spec.Restore)
		base.Spec.Backup = mergeConfiguration(base.Spec.Backup, overlay.Spec.Backup)
		base.Spec.Lightning = mergeConfiguration(base.Spec.Lightning, overlay.Spec.Lightning)
		base.Spec.TiCDC = mergeConfiguration(base.Spec.TiCDC, overlay.Spec.TiCDC)
		base.Spec.Priority = overlay.Spec.Priority
	}
	return base, nil
}

func mergeConfigMeta(c1, c2 *v1.ConfigMeta) *v1.ConfigMeta {
	if c2 == nil {
		return c1
	}
	if c1 == nil {
		c1 = new(v1.ConfigMeta)
	}

	if c2.GitCommitID != "" {
		c1.GitCommitID = c2.GitCommitID
	}
	c1.ConfigVersion = c2.ConfigVersion

	return c1
}

func mergeConfiguration(c1, c2 *v1.Configuration) *v1.Configuration {
	if c2 == nil {
		return c1
	}
	if c1 == nil {
		c1 = new(v1.Configuration)
	}

	if c2.BaseImage != nil {
		c1.BaseImage = c2.BaseImage
	}
	if c2.Tag != nil {
		c1.Tag = c2.Tag
	}
	if c1.Config == nil {
		c1.Config = v1.NewTomlConfig(map[string]interface{}{})
	}
	if c2.Config != nil {
		err := Merge(c1.Config.MP, c2.Config.MP)
		if err != nil {
			fmt.Fprintf(os.Stderr, "failed merge config: %v", err)
		}
	}

	return c1
}

func isMetaConsistent(configs []v1.ClusterConfig) bool {
	list := make([]v1.ClusterConfig, 0, len(configs))
	for _, each := range configs {
		// Since BR may create some ClusterConfigs with the name prefix "restore--", and these ClusterConfigs
		// don't have the ConfigMeta, we need to filter them out
		if !strings.HasPrefix(each.Name, "base--") {
			continue
		}

		// The test-infra will create some ClusterConfigs having GitCommitID 'hacked-config-by-test-infra'
		if each.Spec.ConfigMeta != nil && each.Spec.ConfigMeta.GitCommitID == "hacked-config-by-test-infra" {
			continue
		}

		list = append(list, each)
	}
	if len(list) == 0 {
		return true
	}

	base := list[0]
	for _, each := range list {
		if each.Spec.ConfigMeta.GitCommitID != base.Spec.ConfigMeta.GitCommitID {
			clog.Log.Info("config meta is inconsistent",
				"baseName", base.Name, "baseCommitID", base.Spec.ConfigMeta.GitCommitID,
				"name", each.Name, "commitID", each.Spec.ConfigMeta.GitCommitID,
			)
			return false
		}
		if each.Spec.ConfigMeta.ConfigVersion != base.Spec.ConfigMeta.ConfigVersion {
			clog.Log.Info("config meta is inconsistent",
				"baseName", base.Name, "baseConfigVersion", base.Spec.ConfigMeta.ConfigVersion,
				"name", each.Name, "configVersion", each.Spec.ConfigMeta.ConfigVersion,
			)
			return false
		}
	}

	return true
}

// ProviderLevelOverlay provider level configs
// todo: add another layer in tidb-configuration later? right now we only need to change config of import
// in free-tier
func (g *Gitops) ProviderLevelOverlay(providerType string) *v1.ClusterConfig {
	base := new(v1.ClusterConfig)
	base.Spec.Priority = v1.PriorityProvider
	if providerType == consts.ProviderTypeAWSFreeTier {
		// For serverless tier, we use the default config. The default config
		// of the backend is "tidb", which maybe is changed to "remote" by
		// serverless manger. The config of the addr will be used by "remote"
		// backend.
		lightningConfigTpl := `[lightning]
region-concurrency = 2
table-concurrency = 1
index-concurrency = 1
[tikv-importer]
range-concurrency = 1
backend = "tidb"
on-duplicate = "ignore"
addr = "http://tikv-worker-service.tidb-serverless.svc.cluster.local:19000"
[tidb]
build-stats-concurrency = 1
distsql-scan-concurrency = 1
index-serial-scan-concurrency = 1
checksum-table-concurrency = 1
`
		cfg := config.New(map[string]interface{}{})
		_ = cfg.UnmarshalTOML([]byte(lightningConfigTpl))

		tikvConfigTpl := `[import]
num-threads = 1
`
		tikvCfg := config.New(map[string]interface{}{})
		_ = tikvCfg.UnmarshalTOML([]byte(tikvConfigTpl))

		base.Spec.Lightning = &v1.Configuration{
			Config: v1.NewTomlConfig(cfg.MP),
		}
		base.Spec.TiKV = &v1.Configuration{
			Config: v1.NewTomlConfig(tikvCfg.MP),
		}
	}
	return base
}

func (g *Gitops) getLightningConfig(ccl *v1.ClusterConfigList, version string) (*v1.Configuration, error) {
	baseConfig, err := g.getBaseClusterConfig(ccl, version)
	if err != nil {
		return nil, errors.Annotatef(err, "get base cluster config for version '%s' failed", version)
	}
	if semver.Compare(version, baseConfig.Spec.Version) != 0 {
		return nil, errors.Errorf("The returned base object's version ('%s') is different from the requested version ('%s')", baseConfig.Spec.Version, version)
	}
	lnConfig := baseConfig.Spec.Lightning
	if lnConfig == nil {
		return nil, errors.Errorf("retrieved Lightning config for version ('%s') is empty", version)
	}
	if lnConfig.BaseImage == nil {
		return nil, errors.Errorf("cannot fetch Lightning base image URL for version ('%s')", version)
	}
	if lnConfig.Tag == nil {
		return nil, errors.Errorf("cannot fetch Lightning image tag for version ('%s')", version)
	}
	if lnConfig.Config == nil {
		return nil, errors.Errorf("cannot fetch Lightning config template content for version ('%s')", version)
	}
	return lnConfig, nil
}

func (g *Gitops) InstanceTypeLevelOverlay(ccl *v1.ClusterConfigList, version string, instancesTypes map[v1.ComponentType]string) (*v1.ClusterConfig, error) {
	base := new(v1.ClusterConfig)
	base.Spec.Priority = v1.PriorityInstanceType
	for component, instancesType := range instancesTypes {
		sets := labels.Set{}
		sets[v1.VersionLabel] = version
		sets[v1.InstanceTypeLabel] = instancesType
		ccs, err := g.matchClusterConfig(ccl, sets)
		if err != nil {
			return nil, err
		}
		// sort by priority
		sort.Slice(ccs, func(i, j int) bool {
			return ccs[i].Spec.Priority <= ccs[j].Spec.Priority
		})

		// only get the configuration of the specified component, since the ClusterConfig of instance types
		// may contain the configuration of multiple components.
		for _, c := range ccs {
			switch component {
			case v1.ComponentTiDB:
				base.Spec.TiDB = mergeConfiguration(base.Spec.TiDB, c.Spec.TiDB)
			case v1.ComponentTiKV:
				base.Spec.TiKV = mergeConfiguration(base.Spec.TiKV, c.Spec.TiKV)
			case v1.ComponentPD:
				base.Spec.PD = mergeConfiguration(base.Spec.PD, c.Spec.PD)
			case v1.ComponentTiFlash:
				base.Spec.TiFlash = mergeConfiguration(base.Spec.TiFlash, c.Spec.TiFlash)
				// TODO(chenfei): After we fix this bug, any component config change will cause config change of TiflashLearner
				// for clusters having version v6.5.x. This change may be not expected and ignored, and then finally cause
				// unexpected restart of tiflash. Therefore, we plan to postpone this fix until the config-center online.
				// Because in config-center, changing config for a component will only cause config change for this component rather
				// than recalculating for entire cluster.
				//base.Spec.TiFlashLearner = mergeConfiguration(base.Spec.TiFlashLearner, c.Spec.TiFlashLearner)
			case v1.ComponentTiProxy:
				base.Spec.TiProxy = mergeConfiguration(base.Spec.TiProxy, c.Spec.TiProxy)
			}
		}
	}

	return base, nil
}

// GetCurrentClusterConfiguration gets the cluster's current configuration
// It's used to regenerate ClusterRevision with component configuration overlay whose instance type is changed
func GetCurrentClusterConfiguration(infraCli client.Client, ref *v1.Reference) (*v1.ClusterConfig, error) {
	rev := new(v1.ConfigRevision)
	if err := infraCli.Get(context.TODO(), types.NamespacedName{Name: ref.Name}, rev); err != nil {
		return nil, err
	}
	// generate ClusterConfig from existing ConfigRevision
	cc := &v1.ClusterConfig{
		ObjectMeta: metav1.ObjectMeta{
			Name:   "temp", // the name is unimportant as it's unused in revision.NewConfigRevision() function
			Labels: rev.Labels,
		},
		Spec: v1.ClusterConfigSpec{
			Version: rev.Spec.Version,
			// code in cluster-controller/controllers/cluster/task/common/cluster_context.go:AssembleTidbClusterValues
			// is using this, such as 'api.ConfigIDKey:      cc.ConfigMeta.ID,'
			ConfigMeta: &v1.ConfigMeta{
				GitCommitID:   "",
				ConfigVersion: 0,
			},
			TiDB:           rev.Spec.TiDB,
			PD:             rev.Spec.PD,
			TiProxy:        rev.Spec.TiProxy,
			TiKV:           rev.Spec.TiKV,
			TiFlash:        rev.Spec.TiFlash,
			TiFlashLearner: rev.Spec.TiFlashLearner,
			Backup:         rev.Spec.Backup,
			Restore:        rev.Spec.Restore,
			Lightning:      rev.Spec.Lightning,
			TiCDC:          rev.Spec.TiCDC,
			TiflowMaster:   nil,
			TiflowExecutor: nil,
			UpdatePolicy:   v1.UpdatePolicyDryrun, // This field doesn't matter and is not used.
			Priority:       0,
			Selector:       nil,
		},
	}
	return cc, nil
}