package config

import (
	"time"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/client"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/secretsmanager"
	"github.com/creasty/defaults"
	"github.com/go-playground/validator/v10"
	"github.com/imdario/mergo"
	"github.com/mitchellh/mapstructure"
	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"github.com/spf13/viper"
	ov "github.com/tidbcloud/config-service/oneconfig/viper"
	"go.uber.org/zap"
	"gopkg.in/yaml.v2"
	"k8s.io/apimachinery/pkg/api/resource"
)

// use defaults <- config map (config file) <- remote config <- secrets
// defaults for tag default in the config struct.

// use a single instance of Validate, it caches struct info
var validate = validator.New()

func NormalizeConfigDefaults(config *CentralConfig) error {
	return defaults.Set(config)
}

func NormalizeConfigSetup(config *CentralConfig) error {
	err := viper.Unmarshal(config)
	if err != nil {
		return errors.Annotate(err, "failed unmarshal from viper")
	}

	// let's load some secret from aws and merge it
	if config.ServerConfig.SecretStore != nil {
		secretConfig, err := GetSecretFromAwsSecretManager(config.ServerConfig.SecretStore.SecretKey, config.ServerConfig.SecretStore.Region)
		if err != nil {
			return errors.Annotate(err, "failed to load secret config from aws")
		}
		m := make(map[interface{}]interface{})
		if err := yaml.Unmarshal([]byte(secretConfig), &m); err != nil {
			return errors.Annotate(err, "failed to unmarshal secret config")
		}

		var s CentralConfig

		if err := mapstructure.Decode(m, &s); err != nil {
			return errors.Annotate(err, "failed to mapsturcture secret config")
		}

		if err := mergo.Merge(config, s, mergo.WithOverride); err != nil {
			return errors.Annotate(err, "failed to merge secret config")
		}
	}

	// merge the config from constants, because the viper ignore case of the map key, we need to merge the config from constants to the config from viper
	config.RBACConfig.RBACPathPattern2ResourceMap = mergeMaps(config.RBACConfig.RBACPathPattern2ResourceMap, RBACPathPattern2ResourceMap)
	return nil
}

// MergeMaps is a function that is authored (except the name) by Copilot.
func mergeMaps[K comparable, V any](maps ...map[K]V) map[K]V {
	result := make(map[K]V)
	for _, m := range maps {
		for k, v := range m {
			result[k] = v
		}
	}
	return result
}

func NormalizeConfigValidate(config *CentralConfig) error {
	// validate the values
	if err := validate.Struct(config); err != nil {
		return errors.Annotate(err, "failed to validate the config values")
	}

	if _, err := resource.ParseQuantity(config.ImportConfig.StorageRequest); err != nil {
		return errors.Annotatef(err, "invalid import.storageRequest for importer value: %s", config.ImportConfig.StorageRequest)
	}

	if err := config.BackupConfig.Init(); err != nil {
		return errors.Annotate(err, "init log backup enable min ver constraint")
	}
	if err := config.CompactConfig.InitCompactResourceConfig(); err != nil {
		return errors.Annotate(err, "init compact resource config")
	}
	if err := config.RBACConfig.Init(); err != nil {
		return errors.Annotate(err, "init rbac config")
	}
	if err := config.ChangefeedConfig.Init(); err != nil {
		return errors.Annotate(err, "init changefeed config")
	}

	return nil
}

type MockConfigFunc func(config *CentralConfig)

// MockConfig only used for test
func MockConfig(opts ...MockConfigFunc) error {
	config := &CentralConfig{}
	err := NormalizeConfigDefaults(config)
	if err != nil {
		return errors.Annotate(err, "config defaults")
	}

	for _, opt := range opts {
		opt(config)
	}
	if err := config.BackupConfig.Init(); err != nil {
		return errors.Annotate(err, "init backup constraint")
	}
	if err := config.CompactConfig.InitCompactResourceConfig(); err != nil {
		return errors.Annotate(err, "init compact resource config")
	}
	if err := config.ChangefeedConfig.Init(); err != nil {
		return errors.Annotate(err, "init changefeed config")
	}
	globalConf = config
	return nil
}

func NormalizeConfig() error {
	config := &CentralConfig{}

	err := NormalizeConfigDefaults(config)
	if err != nil {
		return errors.Annotate(err, "config defaults")
	}
	err = NormalizeConfigSetup(config)
	if err != nil {
		return errors.Annotate(err, "config setup")
	}
	err = NormalizeConfigValidate(config)
	if err != nil {
		return errors.Annotate(err, "config validate")
	}
	globalConf = config
	log.L().Debug("set new conf", zap.Reflect("global conf", globalConf))
	return nil
}

func DefaultConfig4Test() error {
	config := &CentralConfig{}
	err := NormalizeConfigDefaults(config)
	if err != nil {
		return errors.Annotate(err, "config defaults")
	}
	globalConf = config
	log.Debug("set new test conf", zap.Reflect("global conf", globalConf))
	return nil
}

// InitConfig parse the central configuration file and init the global config var
func InitConfig(path string) error {
	viper.SetConfigFile(path)
	log.L().Info("using config file", zap.String("file", viper.ConfigFileUsed()))
	err := viper.ReadInConfig()
	if err != nil {
		return errors.Annotate(err, "read config")
	}

	err = LoadRemote()
	if err != nil {
		log.L().Warn("load remote config failed, just ignore it in personal project", zap.Error(err))
	}

	err = NormalizeConfig()
	if err != nil {
		return errors.Annotate(err, "normalize config")
	}

	InitInfraContext()
	return nil
}

func LoadRemote() error {
	remote := "oneconfig-dataflow/config.yaml"
	log.L().Info("Read From Remote", zap.String("remote", remote))
	err := viper.AddRemoteProvider(ov.Provider, ov.Endpoint, remote)
	if err != nil {
		return errors.Annotate(err, "failed to load remote config")
	}

	err = viper.ReadRemoteConfig()
	if err != nil {
		return errors.Annotate(err, "failed to load remote config")
	}

	// open a goroutine to watch remote changes forever
	go func() {
		watchErrTimes := 0
		for {
			time.Sleep(time.Second * 5)

			err := viper.WatchRemoteConfig()
			if err != nil {
				// Config reload will temporarily cause file not found
				// Only log error when retry reach 10 times to avoid noise
				if watchErrTimes++; watchErrTimes%10 == 0 {
					log.L().Error("unable to read remote config", zap.Error(err))
				} else {
					log.L().Warn("unable to read remote config", zap.Error(err))
				}
				continue
			}
			watchErrTimes = 0

			err = NormalizeConfig()
			if err != nil {
				log.L().Error("failed to normalize config", zap.Error(err))
				continue
			}

		}
	}()
	return nil
}

func Global() *CentralConfig {
	return globalConf
}

func GetSecretFromAwsSecretManager(secretName, region string) (string, error) {
	awsConfig := aws.NewConfig().WithRegion(region)
	// retry config using exponential backoff
	awsConfig.Retryer = client.DefaultRetryer{NumMaxRetries: 3}
	opts := session.Options{Config: *awsConfig}
	sess, err := session.NewSessionWithOptions(opts)
	if err != nil {
		return "", err
	}
	svc := secretsmanager.New(sess)
	input := &secretsmanager.GetSecretValueInput{
		SecretId:     aws.String(secretName),
		VersionStage: aws.String("AWSCURRENT"), // VersionStage defaults to AWSCURRENT if unspecified
	}
	result, err := svc.GetSecretValue(input)
	if err != nil {
		return "", err
	}
	return *result.SecretString, nil
}