package sdk

import (
	"context"
	"encoding/json"
	"fmt"

	"github.com/go-logr/logr"
	"github.com/google/go-cmp/cmp"
	"github.com/pingcap/errors"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	operatorconfig "github.com/pingcap/tidb-operator/pkg/apis/util/config"
	apiequality "k8s.io/apimachinery/pkg/api/equality"
	"sigs.k8s.io/controller-runtime/pkg/client"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-api/gitops-sdk/utils"
)

var (
	ConfigUpdateAllComponents = []cloudv1.ComponentType{cloudv1.ComponentTiDB, cloudv1.ComponentTiKV, cloudv1.ComponentPD,
		cloudv1.ComponentTiFlash, cloudv1.ComponentTiCDC}
)

const (
	// recommendedEvictLeaderTimeout is set according to
	// https://docs.pingcap.com/tidb-in-kubernetes/dev/configure-a-tidb-cluster#configure-graceful-upgrade-for-tikv-cluster
	recommendedEvictLeaderTimeout = "10000m"
	ConfigRevisionIdKey           = "tidbcloud.com/config-revision-id"
	defaultTiDBImage              = "gcr.io/pingcap-public/dbaas/tidb"
	defaultTiKVImage              = "gcr.io/pingcap-public/dbaas/tikv"
	defaultPDImage                = "gcr.io/pingcap-public/dbaas/pd"
	defaultTiCDCImage             = "gcr.io/pingcap-public/dbaas/ticdc"
	defaultTiFlashImage           = "gcr.io/pingcap-public/dbaas/tiflash"
	defaultBRImage                = "gcr.io/pingcap-public/dbaas/br"
)

type configManager struct {
	client.Client
	// preConfigKeys defines the keys that are not managed by this configManager,
	// and they will not be modified during configuration updates.
	preConfigKeys map[cloudv1.ComponentType][]string
	// dryrun here controls whether to execute the update, used for old code migration only.
	// The dryrun updatePolicy in ClusterConfig is already handled by ClusterConfig controller.
	dryrun bool
	Log    logr.Logger
}

func NewConfigManager(infraCli client.Client, log logr.Logger, preConfig map[cloudv1.ComponentType][]string, dryrun bool) Interface {
	return &configManager{
		Client:        infraCli,
		preConfigKeys: preConfig,
		dryrun:        dryrun,
		Log:           log,
	}
}

func (m *configManager) InitConfig(cluster *cloudv1.Cluster, revFinder RevisionFinder, tc *v1alpha1.TidbCluster) error {
	revision, err := revFinder(cluster)
	if err != nil {
		return err
	}
	// set the initConfig and image version for components via desired config
	return m.patchTiDBCluster(tc, revision, ConfigUpdateAllComponents, false /* update */)
}

func (m *configManager) UpdateConfig(cluster *cloudv1.Cluster, revFinder RevisionFinder, tc *v1alpha1.TidbCluster,
	tcClient client.Client, pdCli DynamicPDClient, comps []cloudv1.ComponentType, apply bool) error {
	revision, err := revFinder(cluster)
	if err != nil {
		return err
	}

	pdNeeded := false
	for _, comp := range comps {
		if comp == cloudv1.ComponentPD {
			pdNeeded = true
			break
		}
	}
	if tc.Spec.PD != nil && pdNeeded {
		if pdCli == nil {
			return errors.New("pdClient should not be nil when PD is present")
		}
		// get pd cluster config that cloud be online tuned from desired config
		pcc := cloudv1.GetPDClusterConfig(revision.Spec.PD.Config)
		// refer to https://docs.pingcap.com/tidb-in-kubernetes/dev/modify-tidb-configuration#modify-pd-configuration
		// pd config can't be updated in tidb-operator, update it via pdClient.
		err = m.updatePDClusterConfig(pdCli, pcc, cluster)
		if err != nil {
			return err
		}
	}
	// after mutate all the component configs, update tc,
	// if conflicted, let provider do the retry thing.
	return m.updateTidbClusterConfig(tc, tcClient, cluster, apply, func() error {
		return m.patchTiDBCluster(tc, revision, comps, true /* update */)
	})
}

// NOTE: do not forget to update `ConfigUpdateAllComponents` if any new component is added.
func (m *configManager) patchTiDBCluster(tc *v1alpha1.TidbCluster, revision *cloudv1.ConfigRevision,
	comps []cloudv1.ComponentType, update bool) error {
	if tc.Annotations == nil {
		tc.Annotations = map[string]string{}
	}
	// cluster level fileds will always be updated
	tc.Annotations[ConfigRevisionIdKey] = revision.Name
	tc.Spec.Version = revision.Spec.Version

	for _, comp := range comps {
		switch comp {
		case cloudv1.ComponentTiDB:
			if tc.Spec.TiDB != nil {
				confMap := make(map[string]interface{})
				if tc.Spec.TiDB.Config != nil {
					confMap = tc.Spec.TiDB.Config.MP
					if update {
						confMap = m.getPreConfig(tc.Spec.TiDB.Config.GenericConfig, cloudv1.ComponentTiDB)
					}
				}
				desired := revision.Spec.TiDB
				conf := operatorconfig.New(utils.MergeMaps(desired.Config.MP, confMap))
				tc.Spec.TiDB.Config = &v1alpha1.TiDBConfigWraper{GenericConfig: conf}
				tc.Spec.TiDB.Version = desired.Tag
				tc.Spec.TiDB.BaseImage = utils.DefaultIfEmpty(desired.BaseImage, defaultTiDBImage)
			}
		case cloudv1.ComponentTiKV:
			if tc.Spec.TiKV != nil {
				confMap := make(map[string]interface{})
				if tc.Spec.TiKV.Config != nil {
					confMap = tc.Spec.TiKV.Config.MP
					if update {
						confMap = m.getPreConfig(tc.Spec.TiKV.Config.GenericConfig, cloudv1.ComponentTiKV)
					}
				}
				desired := revision.Spec.TiKV
				conf := operatorconfig.New(utils.MergeMaps(desired.Config.MP, confMap))
				tc.Spec.TiKV.Config = &v1alpha1.TiKVConfigWraper{GenericConfig: conf}
				tc.Spec.TiKV.Version = desired.Tag
				tc.Spec.TiKV.BaseImage = utils.DefaultIfEmpty(desired.BaseImage, defaultTiKVImage)

				// Turn up the `evictLeaderTimeout`.
				// The default is 10 minutes, which may cause an error.
				// See more details in https://jira.tidbcloud.com/browse/DM-1051
				// Note: this modification will not make TiKV restart.
				evictLeaderTimeout := recommendedEvictLeaderTimeout
				tc.Spec.TiKV.EvictLeaderTimeout = &evictLeaderTimeout
			}
		case cloudv1.ComponentPD:
			if tc.Spec.PD != nil {
				confMap := make(map[string]interface{})
				if tc.Spec.PD.Config != nil {
					confMap = tc.Spec.PD.Config.MP
					if update {
						currentPCC := cloudv1.ExtractPDClusterConfig(tc.Spec.PD.Config.MP)
						// combine pd cluster-level config and pre defined config
						confMap = utils.MergeMaps(m.getPreConfig(tc.Spec.PD.Config.GenericConfig, cloudv1.ComponentPD), currentPCC.MP)
						// drop the cluster-level config
					}
				}
				desired := revision.Spec.PD
				// to avoid unnecessary rolling-restart, remove the cluster-level config from desired config,
				// then overlay the desired instance-level config by combined config,
				// thus to keep the cluster config untouched at instance level.
				desiredInstanceConf := cloudv1.GetPDInstanceConfig(desired.Config)
				conf := operatorconfig.New(utils.MergeMaps(desiredInstanceConf.MP, confMap))
				tc.Spec.PD.Config = &v1alpha1.PDConfigWraper{GenericConfig: conf}
				tc.Spec.PD.BaseImage = utils.DefaultIfEmpty(desired.BaseImage, defaultPDImage)
				tc.Spec.PD.Version = desired.Tag
			}
		case cloudv1.ComponentTiFlash:
			if tc.Spec.TiFlash != nil {
				commonConfMap, proxyConfMap := make(map[string]interface{}), make(map[string]interface{})
				if tc.Spec.TiFlash.Config != nil {
					if tc.Spec.TiFlash.Config.Common != nil {
						commonConfMap = tc.Spec.TiFlash.Config.Common.MP
						if update {
							commonConfMap = m.getPreConfig(tc.Spec.TiFlash.Config.Common.GenericConfig, cloudv1.ComponentTiFlash)
						}
					}
					if tc.Spec.TiFlash.Config.Proxy != nil {
						proxyConfMap = tc.Spec.TiFlash.Config.Proxy.MP
						if update {
							proxyConfMap = m.getPreConfig(tc.Spec.TiFlash.Config.Proxy.GenericConfig, cloudv1.ComponentTiFlashLearner)
						}
					}
				}

				desired := revision.Spec.TiFlash
				conf := v1alpha1.NewTiFlashConfig()
				conf.Common = &v1alpha1.TiFlashCommonConfigWraper{GenericConfig: operatorconfig.New(utils.MergeMaps(desired.Config.MP, commonConfMap))}
				if revision.Spec.TiFlashLearner != nil {
					proxyConfMap = utils.MergeMaps(revision.Spec.TiFlashLearner.Config.MP, proxyConfMap)
				}
				conf.Proxy = &v1alpha1.TiFlashProxyConfigWraper{GenericConfig: operatorconfig.New(proxyConfMap)}
				tc.Spec.TiFlash.Config = conf
				tc.Spec.TiFlash.Version = desired.Tag
				tc.Spec.TiFlash.BaseImage = utils.DefaultIfEmpty(desired.BaseImage, defaultTiFlashImage)
			}
		case cloudv1.ComponentTiCDC:
			if tc.Spec.TiCDC != nil {
				confMap := make(map[string]interface{})
				if tc.Spec.TiCDC.Config != nil {
					confMap = tc.Spec.TiCDC.Config.GenericConfig.MP
					if update {
						confMap = m.getPreConfig(tc.Spec.TiCDC.Config.GenericConfig, cloudv1.ComponentTiCDC)
					}
				}
				desired := revision.Spec.TiCDC
				conf := operatorconfig.New(utils.MergeMaps(desired.Config.MP, confMap))
				tc.Spec.TiCDC.Config = &v1alpha1.CDCConfigWraper{GenericConfig: conf}
				tc.Spec.TiCDC.Version = desired.Tag
				tc.Spec.TiCDC.BaseImage = utils.DefaultIfEmpty(desired.BaseImage, defaultTiCDCImage)
			}
		}
	}

	return nil
}

func (m *configManager) getPreConfig(config *operatorconfig.GenericConfig, component cloudv1.ComponentType) map[string]interface{} {
	preConfig := operatorconfig.New(make(map[string]interface{}))
	for _, key := range m.preConfigKeys[component] {
		v := config.Get(key)
		if v != nil {
			preConfig.Set(key, v.Interface())
		}
	}
	return preConfig.MP
}

func (m *configManager) updatePDClusterConfig(pdCli DynamicPDClient, desired *cloudv1.PDClusterConfig, c *cloudv1.Cluster) error {
	conf, err := pdCli.GetConfig()
	if err != nil {
		return err
	}
	pcc := cloudv1.ExtractPDClusterConfig(conf)
	current := pcc.MP
	updated := utils.MergeMaps(current, desired.MP)

	// current is json.Unmarshal with json from pd api, type will be float64 for float type in json.
	// updated will contains some int type, so compare `current` and `updated` will not be true.
	// we use json marshal to make it with same type before compare it.
	data, err := json.Marshal(updated)
	if err != nil {
		return err
	}
	updated = make(map[string]interface{})
	err = json.Unmarshal(data, &updated)
	if err != nil {
		return err
	}

	// no update
	if apiequality.Semantic.DeepEqual(current, updated) {
		return nil
	}
	logPlan(m.Log, current, updated, m.dryrun, fmt.Sprintf("PD cluster config of cluster %s/%s", c.Namespace, c.Name))
	if !m.dryrun {
		return pdCli.PutConfig(desired.MP)
	}
	return nil
}

func (m *configManager) updateTidbClusterConfig(tc *v1alpha1.TidbCluster, tcCli client.Client, c *cloudv1.Cluster, apply bool, mutateFn func() error) error {
	old := tc.DeepCopy()
	err := mutateFn()
	if err != nil {
		return err
	}
	// no update
	if apiequality.Semantic.DeepEqual(old, tc) {
		return nil
	}
	logPlan(m.Log, old, tc, m.dryrun, fmt.Sprintf("TidbCluster %s/%s resource of cluster %s/%s", tc.Namespace, tc.Name, c.Namespace, c.Name))
	if !m.dryrun && apply {
		// since we perform comprehensive pre-check, simple back-off to next reconciliation chance on any conflicts
		return tcCli.Update(context.TODO(), tc)
	} else if m.dryrun {
		// rest tc to old
		old.DeepCopyInto(tc)
	}
	return nil
}

func logPlan(logger logr.Logger, before, after interface{}, dryrun bool, target string) {
	mode := "execute"
	if dryrun {
		mode = "dryrun"
	}
	d := cmp.Diff(before, after)
	msg := fmt.Sprintf("[%s] update %s, diff:\n%s", mode, target, d)
	logger.Info(msg)
}

func (m *configManager) InitBackupConfig(cluster *cloudv1.Cluster, revFinder RevisionFinder, backup *v1alpha1.Backup) error {
	revision, err := revFinder(cluster)
	if err != nil {
		return err
	}
	if backup.Spec.BR == nil {
		backup.Spec.BR = &v1alpha1.BRConfig{}
	}
	err = mergeBRConfig(backup.Spec.BR, revision.Spec.Backup.Config)
	if err != nil {
		return err
	}

	tag := utils.DefaultIfEmpty(revision.Spec.Backup.Tag, revision.Spec.Version)
	baseImage := utils.DefaultIfEmpty(revision.Spec.Backup.BaseImage, defaultBRImage)
	backup.Spec.ToolImage = fmt.Sprintf("%s:%s", baseImage, tag)
	return nil
}

func (m *configManager) InitRestoreConfig(cluster *cloudv1.Cluster, revFinder RevisionFinder, restore *v1alpha1.Restore) error {
	revision, err := revFinder(cluster)
	if err != nil {
		return err
	}
	if restore.Spec.BR == nil {
		restore.Spec.BR = &v1alpha1.BRConfig{}
	}
	err = mergeBRConfig(restore.Spec.BR, revision.Spec.Restore.Config)
	if err != nil {
		return err
	}

	tag := utils.DefaultIfEmpty(revision.Spec.Restore.Tag, revision.Spec.Version)
	baseImage := utils.DefaultIfEmpty(revision.Spec.Restore.BaseImage, defaultBRImage)
	restore.Spec.ToolImage = fmt.Sprintf("%s:%s", baseImage, tag)
	return nil
}

func mergeBRConfig(BR *v1alpha1.BRConfig, brConf *cloudv1.TomlConfig) error {
	jsonBody, err := json.Marshal(brConf.MP)
	if err != nil {
		return err
	}
	var brConfig v1alpha1.BRConfig
	err = json.Unmarshal(jsonBody, &brConfig)
	if err != nil {
		return err
	}

	BR.LogLevel = brConfig.LogLevel
	BR.StatusAddr = brConfig.StatusAddr
	BR.TimeAgo = brConfig.TimeAgo
	BR.Options = brConfig.Options
	if brConfig.Concurrency != nil {
		BR.Concurrency = brConfig.Concurrency
	}
	if brConfig.RateLimit != nil {
		BR.RateLimit = brConfig.RateLimit
	}
	if brConfig.Checksum != nil {
		BR.Checksum = brConfig.Checksum
	}
	if brConfig.OnLine != nil {
		BR.OnLine = brConfig.OnLine
	}
	return nil
}