package configchange

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"reflect"
	"strings"

	"github.com/samber/lo"
	"k8s.io/apimachinery/pkg/api/resource"
	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
	"sigs.k8s.io/controller-runtime/pkg/log"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	config2 "github.com/tidbcloud/infra-provider/pkg/config"
	"github.com/tidbcloud/infra-provider/util/config"
)

var (
	SQLExecutorGetter func(ctx context.Context, cluster *cloudv1.Cluster, cloudAdminPasswdGetter CloudAdminPasswdGetter, conf config2.RootConfig) (SQLExecutor, error)
)

func init() {
	SQLExecutorGetter = getSQLExecutor
}

type hotReloadApplier struct {
	cloudAdminPasswdGetter CloudAdminPasswdGetter
	conf                   config2.RootConfig
}

func newHotReloadApplier(conf config2.RootConfig) *hotReloadApplier {
	out := &hotReloadApplier{
		cloudAdminPasswdGetter: newCloudAdminPasswdGetter(&conf.ClusterController.ConfigChange),
		conf:                   conf,
	}
	return out
}

func newHotReloadComponentStep(comp cloudv1.ComponentType) cloudv1.ConfigChangeStepName {
	return cloudv1.ConfigChangeStepName(fmt.Sprintf("HotReloadComponent/%s", comp))
}

func (r *hotReloadApplier) InitSteps(ctx context.Context, cc *cloudv1.ConfigChange) {
	utilruntime.Must(cc.AddStep(ctx, cloudv1.CCStepUpdateCR2Cluster, cloudv1.CCStepStatusPending, ""))
	utilruntime.Must(cc.AddStep(ctx, cloudv1.CCStepUpdateConfig2CM, cloudv1.CCStepStatusPending, ""))

	for _, comp := range r.filterComponentTypes(cc) {
		utilruntime.Must(cc.AddStep(ctx, newHotReloadComponentStep(comp), cloudv1.CCStepStatusPending, ""))
	}
}

func (r *hotReloadApplier) RegisterStepAction(re *ConfigChangeReconciler) {
	for _, comp := range hotReloadComponents {
		comp := comp
		re.RegisterAction(newHotReloadComponentStep(comp), func(ctx context.Context, rx *ReconcileContext) error {
			return r.hotReload(ctx, rx, []cloudv1.ComponentType{comp})
		})
	}
}

func (r *hotReloadApplier) Apply(ctx context.Context, rx *ReconcileContext) (err error) {
	if err := r.hotReload(ctx, rx, r.filterComponentTypes(rx.ConfigChange)); err != nil {
		return fmt.Errorf("hotReload: %w", err)
	}

	return nil
}

func (r *hotReloadApplier) filterComponentTypes(cc *cloudv1.ConfigChange) []cloudv1.ComponentType {
	// should place `hotReloadComponents` as the secondary parameter to ensure the order of result is sorted same as it.
	return lo.Intersect(cc.Status.ChangedComponents, hotReloadComponents)
}

func (r *hotReloadApplier) hotReload(ctx context.Context, rx *ReconcileContext, componentTypes []cloudv1.ComponentType) error {
	logger := rx.Logger
	logger.Info("try to hot reload", "componentTypes", componentTypes)
	cc := rx.ConfigChange
	cluster := rx.Cluster
	remoteCli := rx.RemoteClient
	tcGetter, err := newTCGetter(ctx, cluster, remoteCli)
	if err != nil {
		return fmt.Errorf("failed to get tc: %w", err)
	}

	for _, comp := range componentTypes {
		compStatus := cc.ReadOnlyComponentStatus(comp)
		// component is not required manually hot apply config via SQL, and can be handled properly by component itself
		if comp == cloudv1.ComponentPD || comp == cloudv1.ComponentTiFlash || comp == cloudv1.ComponentTiProxy {
			msg := "PD config have been updated by cluster-controller via HTTP API"
			if comp == cloudv1.ComponentTiFlash {
				msg = "TiFlash config have been auto reloaded by itself"
			} else if comp == cloudv1.ComponentTiProxy {
				msg = "TiProxy config have been auto reloaded by itself"
			}
			compStatus.UpdatedReplicas = int32(cluster.GetComponent(comp).GetReplicas())
			// no corresponding resources to be changed, directly update the step to Completed
			compStatus.Done = true
			if err := cc.UpdateComponentStatus(ctx, comp, &compStatus); err != nil {
				return fmt.Errorf("update component(%s) status: %w", comp, err)
			}
			cc.UpdateStep(ctx, newHotReloadComponentStep(comp), cloudv1.CCStepStatusCompleted, msg)
			if err := rx.LocalClient.Status().Update(ctx, cc); err != nil {
				return fmt.Errorf("update configChange status: %w", err)
			}

			continue
		}

		// component is not deployed
		if tcGetter.ComponentReplicas(comp) == 0 {
			logger.Info("component not found in tc", "component", comp)
			if err := markComponentAndStepDoneForComponentNotDeployed(ctx, cc, comp, newHotReloadComponentStep(comp)); err != nil {
				return fmt.Errorf("markComponentAndStepDoneForComponentNotDeployed: %w", err)
			}
			if err := rx.LocalClient.Status().Update(ctx, cc); err != nil {
				return fmt.Errorf("update configChange status: %w", err)
			}

			continue
		}

		// currently, only tikv, tiflashLearner would reach here
		if err := r.hotReloadOneComponent(ctx, rx, comp); err != nil {
			return fmt.Errorf("hotReloadOneComponent %s: %w", comp, err)
		}
	}

	return nil
}

func (r *hotReloadApplier) hotReloadOneComponent(ctx context.Context, rx *ReconcileContext, comp cloudv1.ComponentType) error {
	cc := rx.ConfigChange
	cluster := rx.Cluster
	logger := rx.Logger
	oldConf := config.New(nil)
	if conf := rx.LastConfigRevision.CompConfByType(comp); conf != nil && conf.Config != nil {
		oldConf = config.New(conf.Config.MP)
	}
	newConf := config.New(rx.TargetConfigRevision.CompConfByType(comp).Config.MP)
	flattedOldConf := oldConf.FlatMP()
	flattedNewConf := newConf.FlatMP()
	step := newHotReloadComponentStep(comp)
	sqlExecutor, err := SQLExecutorGetter(ctx, rx.Cluster, r.cloudAdminPasswdGetter, r.conf)
	if err != nil {
		return err
	}
	cc.UpdateStep(ctx, newHotReloadComponentStep(comp), cloudv1.CCStepStatusRunning, "try to hot reload config")

	// handle config item removed
	for k := range flattedOldConf {
		if _, ok := flattedNewConf[k]; !ok {
			err := fmt.Errorf("there are some conf are removed, can't do hotReload")
			logger.Error(err, "")
			return err
		}
	}

	// handle config item changed(added or modified)
	for k, newValue := range flattedNewConf {
		oldValue := flattedOldConf[k]
		if !reflect.DeepEqual(oldValue, newValue) {
			sql := buildConfigChangeSQL(comp, k, newValue)
			logger.Info("execute sql to change config", "sql", sql, "key", k)
			// 1. the key(or configuration item) can't be modified dynamically (only return warning with code 1105)
			// 2. the value is invalid to parse (only return warning with code 1105)
			// 3. persist change to config file failed (only return warning with code 1105)
			if err := sqlExecutor.Execute(ctx, sql); err != nil {
				return fmt.Errorf("execute sql: %w", err)
			}

			if inconsistent, err := compareConfig(ctx, rx, sqlExecutor, comp, k, newValue); err != nil {
				return fmt.Errorf("compare config: %w", err)
			} else if len(inconsistent) != 0 {
				cc.UpdateStep(ctx, step, cloudv1.CCStepStatusRunning, fmt.Sprintf("dynamically update config failed on some instances, key=%s, inconsistent=%s", k, inconsistent))
				return rx.LocalClient.Status().Update(ctx, cc)
			}
			logger.Info("update one config success", "key", k)
		}
	}

	compStatus := cc.ReadOnlyComponentStatus(comp)
	compStatus.UpdatedReplicas = int32(cluster.GetComponent(comp).GetReplicas())
	compStatus.Done = true
	if err := cc.UpdateComponentStatus(ctx, comp, &compStatus); err != nil {
		return fmt.Errorf("update component(%s) status: %w", comp, err)
	}
	cc.UpdateStep(ctx, step, cloudv1.CCStepStatusCompleted, "all config changes are applied via SQL")
	if err := rx.LocalClient.Status().Update(ctx, cc); err != nil {
		return fmt.Errorf("update configChange status: %w", err)
	}

	return nil
}

func compareConfig(ctx context.Context, rx *ReconcileContext, sqlExecutor SQLExecutor, comp cloudv1.ComponentType, key string, expectedValue interface{}) (inconsistent []string, _ error) {
	logger := log.FromContext(ctx)
	queryValueSQL := buildGetConfigSQL(comp, key)
	logger.Info("query to get config", "sql", queryValueSQL, "key", key)
	resultBytes, err := sqlExecutor.Query(ctx, queryValueSQL)
	if err != nil {
		return nil, fmt.Errorf("query %s: %w", queryValueSQL, err)
	}
	var result showConfigResult
	if err := json.Unmarshal(resultBytes, &result); err != nil {
		return nil, fmt.Errorf("unmarshal show config result: %w", err)
	}
	// Some instances down may cause len(instance) of show config result is less than replica
	if replica := rx.Cluster.GetComponent(comp).GetReplicas(); replica > len(result.Instance) {
		return nil, fmt.Errorf("len(result.instance)=%d is not equal to replicas(%d)", len(result.Instance), replica)
	}
	for idx, each := range result.Value {
		actualValueBytes, err := base64.StdEncoding.DecodeString(each)
		if err != nil {
			return nil, fmt.Errorf("failed to base64-decode of show config result.value: %w", err)
		}
		actualValueStr := string(actualValueBytes)
		instanceBytes, err := base64.StdEncoding.DecodeString(result.Instance[idx])
		if err != nil {
			return nil, fmt.Errorf("failed to base64-decode of show config result.instance: %w", err)
		}
		instance := string(instanceBytes)
		expectedValueStr := fmt.Sprintf("%v", expectedValue)
		logger.Info(fmt.Sprintf("value on %s is %s", instance, actualValueStr))

		if !compareConfigurationValue(ctx, actualValueStr, expectedValueStr) {
			inconsistent = append(inconsistent, fmt.Sprintf("%s,actual=%s", instance, actualValueStr))
			continue
		}
	}

	return inconsistent, nil
}

var (
	// MisusedUnitsOfConfigurationValue defines a unit map, its Key is the unit may be improperly used by user,
	// and its Value is the unit always used by tidb kernel. Both of Key and Value are acceptable and treated as the same
	// in tidb kernel, so we should ignore them when comparing configuration value pair having these unit suffix.
	// For example, if the expected value is '20MB' and the actual value is '20MiB', we should regard them as the same.
	MisusedUnitsOfConfigurationValue = map[string]string{
		"KB": "KiB",
		"MB": "MiB",
		"GB": "GiB",
		"TB": "TiB",
	}
)

// compareConfigurationValue
// `actual` is the value tidb returned
// `expected` is the value user set and expected
func compareConfigurationValue(ctx context.Context, actual, expected string) bool {
	logger := log.FromContext(ctx)

	if actual == expected {
		return true
	}

	// v is the unit actually used by tidb kernel, k is the unit may be improperly used by user.
	for k, v := range MisusedUnitsOfConfigurationValue {
		if !strings.HasSuffix(expected, k) {
			continue
		}

		logger.Info("hit MisusedUnitsOfConfigurationValue", "actual", actual, "expected", expected)
		expected = strings.TrimSuffix(expected, k) + v
	}
	if actual == expected {
		return true
	}

	// tidb automatically convert the unit of storage size value between KiB, MiB, GiB, etc. So if your expected value
	// is "31457280KiB", the actual value may be "30GiB". To deal with this case, we parse them and then compare.
	logger.Info("parse and then compare actual and amendedExpected")
	// `resource` lib supports only 'Ki' not 'KiB', so trim it
	// in here, we don't need to ensure the unit of both actual and expected are storage size, just ignore by error or
	// compare fail
	actualValue, err1 := resource.ParseQuantity(strings.TrimSuffix(actual, "B"))
	expectedValue, err2 := resource.ParseQuantity(strings.TrimSuffix(expected, "B"))
	if err1 != nil || err2 != nil {
		return false
	}
	if actualValue.Cmp(expectedValue) == 0 {
		return true
	}
	return false
}

func getSQLExecutor(ctx context.Context, cluster *cloudv1.Cluster, cloudAdminPasswdGetter CloudAdminPasswdGetter, conf config2.RootConfig) (SQLExecutor, error) {
	// TODO(chenfei): add cache
	cloudAdminPasswd, err := cloudAdminPasswdGetter.GetCloudAdminPasswd(ctx, cluster.Name)
	if err != nil {
		return nil, fmt.Errorf("get cloud admin password: %w", err)
	}
	return newClusterProxySQLExecutor(ctx, cluster, cloudAdminPasswd, conf)
}

type SQLExecutor interface {
	Execute(ctx context.Context, sql string) error
	Query(ctx context.Context, sql string) ([]byte, error)
}

type CloudAdminPasswdGetter interface {
	GetCloudAdminPasswd(ctx context.Context, clusterID string) (string, error)
}

func buildConfigChangeSQL(comp cloudv1.ComponentType, key string, value interface{}) string {
	key = decorateConfigKeyInSQL(comp, key)
	if comp == cloudv1.ComponentTiFlashLearner {
		// Set config for tiflashLearner should use `tiflash` as the component name.
		comp = cloudv1.ComponentTiFlash
	}
	val := reflect.ValueOf(value).Interface()
	sqlTemplate := fmt.Sprintf("set config %s `%s`=?", comp, key)

	return buildSQL(sqlTemplate, val)
}

func buildGetConfigSQL(comp cloudv1.ComponentType, key string) string {
	key = decorateConfigKeyInSQL(comp, key)
	if comp == cloudv1.ComponentTiFlashLearner {
		// Get config for tiflashLearner should use `tiflash` as the component name.
		comp = cloudv1.ComponentTiFlash
	}
	sql := fmt.Sprintf("show config where type='%s' and name='%s'", comp, key)

	return sql
}

// decorateConfigKeyInSQL decorates a configuration key if need
func decorateConfigKeyInSQL(comp cloudv1.ComponentType, key string) string {
	if comp == cloudv1.ComponentTiFlashLearner {
		// 1. Currently, tiflash has two categories of config (and two components), which can be retrieved by the same SQL `show config where type='tiflash'`.
		// One has prefix like 'raftstore-proxy.' (is for tiflash-learner) and another has prefix like 'engine-store.' which is not supported to hot reload config.
		// 2. For the keys having prefix 'raftstore-proxy', they are mirror config (subset) of tikv's config. So if the key can be hot reloaded by tikv,
		//the tiflash too. To get list of config keys supporting hot reload, click
		// [here](https://docs-archive.pingcap.com/tidb/v6.3/dynamic-config#modify-tikv-configuration-dynamically)
		// 3. As these two prefixes are only required when setting configuration via SQL, we automatically add this prefix 'raftstore-proxy.' here.
		// Refer to https://pingcap.feishu.cn/wiki/UsvuwWM25i5YIHkKXZBcfzADnnf
		key = "raftstore-proxy." + key
	}

	return key
}

type showConfigResult struct {
	//Type     []string `json:"Type"`
	Instance []string `json:"Instance"`
	//Name     []string `json:"Name"`
	Value []string `json:"Value"`
}