package config

import (
	"fmt"
	"os"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/Shopify/sarama"
	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/session"
	"github.com/aws/aws-sdk-go/service/s3"
	"github.com/aws/aws-sdk-go/service/s3/s3iface"
	"github.com/go-sql-driver/mysql"
	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"go.uber.org/zap"
	"gopkg.in/yaml.v3"

	"github.com/pingcap-inc/clinic/pkg/o11y"
	"github.com/pingcap-inc/clinic/pkg/stream"
)

var globalCfg *Config

func GetConfig() *Config {
	return globalCfg
}

// GetDeploymentEnv returns the configured deployment environment (Config.Env), e.g. dev, staging, prod, or "" if unset.
func GetDeploymentEnv() string {
	cfg := GetConfig()
	if cfg == nil {
		return ""
	}
	return cfg.Env
}

// FilterTags has format: key1:value1;key2:value2;key3:value3
type FilterTags string

func (ft FilterTags) ToMap() map[string]string {
	if len(ft) == 0 {
		return nil
	}
	tags := make(map[string]string)
	for _, pair := range strings.Split(string(ft), ";") {
		kv := strings.Split(pair, ":")
		if len(kv) != 2 {
			continue
		}
		tags[kv[0]] = kv[1]
	}
	return tags
}

type Config struct {
	DisableInNewDeployment bool `yaml:"disable-in-new-deployment"`

	// Env is the deployment environment (e.g. dev, staging, prod). Used for environment-specific behavior such as shared-cluster region overrides.
	Env string `yaml:"env"`

	AllowMaxDataRange int64 `yaml:"allow-max-data-range"`

	LinkTiDBCloud    bool             `yaml:"link-tidb-cloud"`
	UseAPIGateWay    bool             `yaml:"use-api-gate-way"`
	ServerConfig     *ServerConfig    `yaml:"server"`
	DBConfig         *DBConfig        `yaml:"db"`
	FileStorage      FileStorage      `yaml:"file-storage"`
	SinkConfig       SinkConfig       `yaml:"sink"`
	ReportConfig     ReportConfig     `yaml:"report-config"`
	DiagnosisConfig  DiagnosisConfig  `yaml:"diagnosis"`
	DatasourceConfig DatasourceConfig `yaml:"datasource"`

	UploadConfig      UploadConfig      `yaml:"upload-config"`
	TaskManagerConfig TaskManagerConfig `yaml:"task-manager-config"`

	K8sConfig             K8sConfig              `yaml:"k8s-config"`
	EBSFilterTags         FilterTags             `yaml:"ebs-filter-tags"`
	EKSConfig             EKSConfig              `yaml:"eks"`
	DBaasConfig           DBaasConfig            `yaml:"dbaas-config"`
	AllowCIDRS            []string               `yaml:"allow-cidrs"`
	DisableIPWhitelist    bool                   `yaml:"disable-ip-whitelist"`
	SecretStore           *SecretConfig          `yaml:"secret"`
	EnableO11yCanary      bool                   `yaml:"enable-o11y-canary"`
	CentralO11yConfig     *CentralO11yConfig     `yaml:"central-o11y"`
	SecondO11yConfig      *CentralO11yConfig     `yaml:"second-o11y"`
	MixPanelConfig        MixPanelConfig         `yaml:"mix-panel-config"`
	ActivityManagerConfig *ActivityManagerConfig `yaml:"activity-manager"`
	ExternalDatasource    ExternalDatasource     `yaml:"external-datasource"`
	ActivityFileConfig    ActivityFileConfig     `yaml:"activity-file"`

	EnablePerformanceInsight bool         `yaml:"enable-performance-insight"`
	ExternalFeatureGates     FeatureGates `yaml:"external-feature-gates"`

	DataServiceConfig DataServiceConfig `yaml:"data-service"`
	OpsDBConfig       OpsDBConfig       `yaml:"ops-db"`

	// KafkaExt configures KafkaExt SASL/TLS for tiinf auditlog / tieventbus when audit_kafka.sasl.enable is true.
	// Broker addresses still come from OneConfig (kafka_brokers); this only sets transport auth.
	KafkaExt KafkaExtConfig `yaml:"kafka_ext"`
}

const DefaultAllowMaxDataRange int64 = 90 * 24 * 60 * 60

func (c *Config) GetAllowMaxDataRange() int64 {
	if c == nil || c.AllowMaxDataRange <= 0 {
		return DefaultAllowMaxDataRange
	}
	return c.AllowMaxDataRange
}

// KafkaExtConfig is used by cloud_server/pkg/auditloginit for Sarama SASL/TLS.
type KafkaExtConfig struct {
	SASL KafkaSASL `yaml:"sasl"`
	TLS  KafkaTLS  `yaml:"tls"`
}

// KafkaSASL enables PLAIN or SCRAM-* when Enable is true.
type KafkaSASL struct {
	Enabled   bool                 `yaml:"enabled"`
	Mechanism sarama.SASLMechanism `yaml:"mechanism"` // PLAIN, SCRAM-SHA-256, SCRAM-SHA-512
	Username  string               `yaml:"username"`
	Password  string               `yaml:"password"`
}

// KafkaTLS is applied when tls.enable is true or sasl.enable is true (TLS defaults on with SASL).
type KafkaTLS struct {
	Enabled            bool `yaml:"enabled"`
	InsecureSkipVerify bool `yaml:"insecure_skip_verify"`
}
type OpsDBConfig struct {
	Host        string `yaml:"host"`
	Port        int    `yaml:"port"`
	DBName      string `yaml:"db-name"`
	Salt        string `yaml:"salt"`
	AESKey      string `yaml:"aes-key"`
	UserEnc     string `yaml:"user-enc"`
	PasswordEnc string `yaml:"password-enc"`
}

type DataServiceConfig struct {
	Addr  string `yaml:"addr"`  // e.g. "http://clinic-data-service:8087"; defaults to that if empty
	Token string `yaml:"token"` // Bearer token for service-to-service auth
}

func (c *DataServiceConfig) GetAddr() string {
	if c.Addr != "" {
		return c.Addr
	}
	return "http://clinic-data-service:8087"
}

type ExternalDatasource struct {
	TelemetryDB *DBConfig `yaml:"telemetry-db"`
}

type MixPanelConfig struct {
	ProjectID int    `yaml:"project-id"`
	UserName  string `yaml:"user-name"`
	Secret    string `yaml:"secret"`
}

type SecretConfig struct {
	SecretKey string `yaml:"secretKey"`
	Region    string `yaml:"region"`
}

type DBaasConfig struct {
	DashboardConfig          DashboardConfig `yaml:"dashboard-proxy"`
	CoreSvcAddr              string          `yaml:"core-svc-addr"`
	ServerlessEndpoint       string          `yaml:"serverless-svc-addr"`
	AccountSvcAddr           string          `yaml:"account-svc-addr"`
	ClinicCloudServerAddr    string          `yaml:"clinic-cloud-server-addr"`
	ClinicInternalServerAddr string          `yaml:"clinic-internal-server-addr"`
	MgmtSvcAddr              string          `yaml:"mgmt-svc-addr"`
	ConfigCenterSvcAddr      string          `yaml:"config-center-svc-addr"`
	NextgenSvcAddr           string          `yaml:"nextgen-svc-addr"`
	NextgenSvcToken          string          `yaml:"nextgen-svc-token"`
}

type DashboardConfig struct {
	Endpoint      string `yaml:"endpoint"`
	SessionSecret string `yaml:"session-secret"`
}

type EKSConfig struct {
	Region          string `yaml:"region"`
	ClusterName     string `yaml:"cluster-name"`
	ImportNodeGroup string `yaml:"import-node-group"`
	AccessKey       string `yaml:"access-key"`
	SecretKey       string `yaml:"secret-key"`
	Token           string `yaml:"token"`
}

// LoadDashboard call MetricConfig.LoadDashboard to read grafana dashboard related config.
func (c *Config) LoadDashboard() error {
	_, err := c.SinkConfig.MetricConfig.LoadDashboard()
	if err != nil {
		log.Error("load dashboard failed", zap.Error(err))
		return err
	}
	return nil
}

type UploadConfig struct {
	AddressConfig `yaml:",inline"`
}

type TaskManagerConfig struct {
	AddressConfig `yaml:",inline"`
}

type ActivityManagerConfig struct {
	GroupRule         map[string]ActivityGroupRule `yaml:"group-rule"`
	ActivityHubAddr   string                       `yaml:"activityhub-endpoint"`
	ActivityHubSource string                       `yaml:"activityhub-source"`
	ActivityHubKey    string                       `yaml:"activityhub-key"`
}

type ActivityGroupRule struct {
	Rule    []string `yaml:"rule"`
	Timeout int64    `yaml:"timeout"`
}
type ReportConfig struct {
	TemplatePath          string `yaml:"template-path"`
	TempFilePath          string `yaml:"temp-file-path"`
	ComputeEngineEndPoint string `yaml:"compute-engine-end-point"`

	QpsReportCronPlan     string `yaml:"qps-report-cron-plan" default:"0 1 * * ?"`
	QpsReportThreshold    []int  `yaml:"qps-report-threshold"`
	QpsReportRequestLimit int    `yaml:"qps-report-request-limit" default:"100"`
	QpsReportConcurrency  int    `yaml:"qps-report-concurrency" default:"50"`
	QpsReportDateRange    int    `yaml:"qps-report-date-range" default:"1"`
}

type K8sConfig struct {
	Namespace string `yaml:"namespace"`
	InCluster bool   `yaml:"in-cluster"`
	Path      string `yaml:"path"`

	ImageName string `yaml:"image-name"`
}

type SinkConfig struct {
	MetricConfig MetricConfig `yaml:"metric"`
	LogConfig    LogConfig    `yaml:"log"`
}

// DiagnosisConfig does not support op and cloud in runtime
type DiagnosisConfig struct {
	MetricConfig MetricSingleEndpointConfig `yaml:"metric"`
	LogConfig    LogConfig                  `yaml:"log"`
}

type DatasourceConfig struct {
	MetricConfig MetricConfig        `yaml:"metric"`
	LogConfig    DatasourceLogConfig `yaml:"log"`
}

type DatasourceLogConfig struct {
	CloudAddress LogConfig `yaml:"cloud"`
	OpAddress    LogConfig `yaml:"op"`
}

type MetricConfig struct {
	VMClusterAddressConfig        `yaml:",inline"`
	SelectAddr                    MetricSelectAddressConfig `yaml:"select-addr"`
	DefaultAccountId              uint32                    `yaml:"default-account-id"`
	DashboardPath                 string                    `yaml:"dashboard-path"`
	SortedDashboardVersion        []string
	SortedNextGenDashboardVersion []string
	GrafanaEndpoints              map[string]string
	GrafanaDashboardMetas         map[string][]*MetricDashboardMeta
	EnableInternalGrafana         bool `yaml:"enable-internal-grafana" default:"false"`
}

// LoadDashboard read yaml form DashboardPath and set related fields.
func (cfg *MetricConfig) LoadDashboard() (*MetricConfig, error) {
	if cfg.DashboardPath == "" {
		// do nothing
		return cfg, nil
	}

	// read available from interal grafana configmap
	internalMetas, versions, ngVersions, err := parseDashboardConfigMap(filepath.Join(cfg.DashboardPath, "dashboard-internal"))
	if err != nil {
		return nil, err
	}
	externalMetas, _, _, err := parseDashboardConfigMap(filepath.Join(cfg.DashboardPath, "dashboard-external"))
	if err != nil {
		return nil, err
	}
	metas := make(map[string][]*MetricDashboardMeta)
	for k, v := range internalMetas {
		metas[k+"-internal"] = v
	}
	for k, v := range externalMetas {
		metas[k+"-external"] = v
	}

	data, err := os.ReadFile(fmt.Sprintf("%s/endpoints.yaml", cfg.DashboardPath))
	if err != nil {
		return nil, err
	}
	endpoints := make(map[string]string)
	if err = yaml.Unmarshal(data, &endpoints); err != nil {
		return nil, err
	}

	cfg.GrafanaDashboardMetas = metas
	cfg.SortedDashboardVersion = versions
	cfg.SortedNextGenDashboardVersion = ngVersions
	cfg.GrafanaEndpoints = endpoints
	return cfg, nil
}

type MetricSingleEndpointConfig struct {
	VMClusterAddressConfig `yaml:",inline"`

	SelectAddr VMClusterAddressConfig `yaml:"select-addr"`

	DefaultAccountId uint32 `yaml:"default-account-id"`

	//DashboardPath string `yaml:"dashboard-path"`
	//
	//GrafanaConfig GrafanaConfig `yaml:"grafana"`
}

type GrafanaConfig struct {
	AddressConfig `yaml:",inline"`
	Url           string `yaml:"url"`
}

type LogConfig struct {
	// Endpoint include scheme
	Endpoint     string       `yaml:"endpoint"`
	User         string       `yaml:"user"`
	Password     string       `yaml:"password"`
	KibanaConfig KibanaConfig `yaml:"kibana"`
}

type SessionConfig struct {
	Secret   string `yaml:"secret"`
	Disabled bool   `yaml:"disabled"`
	Secure   bool   `yaml:"secure"`
}

type KibanaConfig struct {
	AddressConfig `yaml:",inline"`
}

type AddressConfig struct {
	Host string `yaml:"host"`
	Port uint16 `yaml:"port"`
}

// https://datatracker.ietf.org/doc/html/rfc3986/#section-3
func (cfg *AddressConfig) Authority() string {
	if cfg.Port == 0 {
		return cfg.Host
	}
	return fmt.Sprintf("%s:%d", cfg.Host, cfg.Port)
}

type VMClusterAddressConfig struct {
	AddressConfig `yaml:",inline"`
	VMClusterMode bool `yaml:"vm-cluster-mode"`
}

type MetricSelectAddressConfig struct {
	CloudAddress VMClusterAddressConfig `yaml:"cloud"`
	OpAddress    VMClusterAddressConfig `yaml:"op"`
}

func (cfg *VMClusterAddressConfig) InVMCluster() bool {
	return cfg.VMClusterMode
}

type ServerConfig struct {
	Host           string        `yaml:"host"`
	Addr           string        `yaml:"addr"`
	TrustedCAFile  string        `yaml:"trusted-ca-file"`
	CertFile       string        `yaml:"cert-file"`
	KeyFile        string        `yaml:"key-file"`
	RbacModelPath  string        `yaml:"rbac-model-path"`
	ListFilesLimit int           `yaml:"list-files-limit"`
	Accounts       Accounts      `yaml:"accounts"`
	OAuthConfigs   []OAuthConfig `yaml:"oauthes"`

	APIsLimit       string         `yaml:"apis-limit"`
	APILimitConfigs []*LimitConfig `yaml:"-"`

	SessionConfig SessionConfig `yaml:"session"`
}

type Accounts struct {
	Users  []User `yaml:"users"`
	Admins []User `yaml:"admins"`
}

type OAuthConfig struct {
	Name         string `yaml:"name"`
	ClientId     string `yaml:"client-id"`
	ClientSecret string `yaml:"client-secret"`
	Redirect     string `yaml:"redirect"`
	AuthUrl      string `yaml:"auth-url"`
	TokenUrl     string `yaml:"token-url"`
	ApiUrl       string `yaml:"api-url"`
	Scopes       string `yaml:"scopes"`
	Enabled      bool   `yaml:"enabled"`
}

type User struct {
	Name     string `yaml:"name"`
	Password string `yaml:"password"`
}

type ImporterConfig struct {
	Name          string `yaml:"name"`
	AddressConfig `yaml:",inline"`
}

type DBConfig struct {
	User     string `yaml:"user"`
	Host     string `yaml:"host"`
	Database string `yaml:"database"`
	// InitDatabase is used to connect to the very first created database.
	InitDatabase      string `yaml:"init-database"`
	Password          string `yaml:"password"`
	Port              uint16 `yaml:"port"`
	EnableDetailedLog bool   `yaml:"enable-detailed-log"`
	LogLevel          string `yaml:"log-level"`
	ConnMaxLifetime   time.Duration
	MaxIdleConns      int
	MaxOpenConns      int
	TLSConfigKey      string `yaml:"tls-config-key"`
}

type FileStorage struct {
	Type          string     `yaml:"type"`
	RSAPrivateKey string     `yaml:"key"`
	AWSStorage    AWSStorage `yaml:"aws"`
}

type LimitConfig struct {
	URI   string
	Limit int
	Burst int
	// time.
}

func (fs FileStorage) StorageType() StorageType {
	if fs.Type == "" {
		return AWSStorageType
	}

	return StorageType(fs.Type)
}

type StorageType string

const (
	AWSStorageType StorageType = "aws"

	DefaultListFilesLimit = 3
)

type AWSStorage struct {
	Endpoint       string `yaml:"endpoint"`
	AccessKey      string `yaml:"access-key"`
	SecretKey      string `yaml:"secret-key"`
	Bucket         string `yaml:"bucket"`
	Region         string `yaml:"region"`
	ForcePathStyle bool   `yaml:"forcepathstyle"`
}

func InitConfig(cfgPath string) (*Config, error) {
	b, err := os.ReadFile(cfgPath)
	if err != nil {
		return nil, err
	}
	cfg, err := initConfig(b)
	if err != nil {
		return nil, err
	}

	err = merge(cfg)
	if err != nil {
		return nil, err
	}

	globalCfg = cfg
	return cfg, nil
}

func (c *ServerConfig) GetFilesLimit() int {
	limit := c.ListFilesLimit
	if limit <= 0 {
		limit = DefaultListFilesLimit
	}

	return limit
}

func initConfig(body []byte) (*Config, error) {
	var cfg Config
	err := yaml.Unmarshal(body, &cfg)
	if err != nil {
		return nil, err
	}

	if cfg.SinkConfig.MetricConfig.SelectAddr.OpAddress.Host == "" && cfg.SinkConfig.MetricConfig.SelectAddr.CloudAddress.Host == "" {
		return nil, errors.New("miss metrics select address")
	}

	cfg.ServerConfig.APILimitConfigs = formatAPILimitConfig(cfg.ServerConfig.APIsLimit)
	return &cfg, nil
}

type MetricDashboardMeta struct {
	Name string
	UID  string
}

func parseDashboardConfigMap(dirPath string) (map[string][]*MetricDashboardMeta, []string, []string, error) {
	versions := make([]string, 0)
	ngVersions := make([]string, 0)
	metas := make(map[string][]*MetricDashboardMeta)

	fileList, err := os.ReadDir(dirPath)
	if err != nil {
		log.Error("read dashboard path", zap.Error(err))
		return nil, nil, nil, err
	}

	for _, f := range fileList {
		if f.IsDir() || strings.HasPrefix(f.Name(), ".") {
			continue
		}
		name := f.Name()
		data, err := os.ReadFile(filepath.Join(dirPath, name))
		if err != nil {
			return nil, nil, nil, err
		}

		uidMap := map[string]string{}
		err = yaml.Unmarshal(data, &uidMap)
		if err != nil {
			return nil, nil, nil, err
		}

		meta := make([]*MetricDashboardMeta, 0)
		for uid, title := range uidMap {
			meta = append(meta, &MetricDashboardMeta{Name: title, UID: uid})
		}
		sort.Slice(meta, func(i, j int) bool {
			return strings.Compare(meta[i].Name, meta[j].Name) < 0
		})
		metas[name] = meta

		if strings.HasPrefix(name, "clinic-private-cluster-") {
			versions = append(versions, strings.TrimPrefix(name, "clinic-private-cluster-"))
		}
		if strings.HasPrefix(name, "clinic-private-nextgen-") {
			ngVersions = append(ngVersions, strings.TrimPrefix(name, "clinic-private-nextgen-"))
		}
	}
	sort.Slice(versions, func(i, j int) bool {
		return strings.Compare(versions[i], versions[j]) < 0
	})
	sort.Slice(ngVersions, func(i, j int) bool {
		return strings.Compare(ngVersions[i], ngVersions[j]) < 0
	})

	return metas, versions, ngVersions, nil
}

func formatAPILimitConfig(limits string) []*LimitConfig {
	if limits == "" {
		return nil
	}

	apiLimits := strings.Split(limits, ";")

	configs := make([]*LimitConfig, 0)
	stream.OfSlice(apiLimits).Map(func(apiLimit string) *LimitConfig {
		lc := strings.Split(apiLimit, ":")
		if len(lc) != 3 {
			return nil
		}

		limit, err := strconv.ParseInt(lc[1], 10, 32)
		if err != nil {
			return nil
		}
		brust, err := strconv.ParseInt(lc[2], 10, 32)
		if err != nil {
			return nil
		}

		return &LimitConfig{
			URI:   lc[0],
			Limit: int(limit),
			Burst: int(brust),
		}
	}).Filter(func(lc *LimitConfig) bool {
		return lc != nil
	}).CollectEmptySlice(&configs)

	sort.Slice(configs, func(i, j int) bool {
		return len(configs[i].URI) > len(configs[j].URI)
	})

	return configs
}

func (c DBConfig) DSN() string {
	return c.toMySQLConfigWithDefaults().FormatDSN()
}

func (c DBConfig) DSNWithTestDB() string {
	localConfig := c.toMySQLConfigWithDefaults()

	initDatabase := c.InitDatabase
	if initDatabase == "" {
		initDatabase = "test"
	}
	localConfig.DBName = initDatabase
	return localConfig.FormatDSN()
}

func (c DBConfig) toMySQLConfigWithDefaults() *mysql.Config {
	defaultConfig := mysql.NewConfig()

	defaultConfig.Net = "tcp"
	defaultConfig.User = c.User
	defaultConfig.Addr = fmt.Sprintf("%s:%d", c.Host, c.Port)
	defaultConfig.Passwd = c.Password
	defaultConfig.DBName = c.Database
	defaultConfig.ParseTime = true
	defaultConfig.Loc = time.UTC
	defaultConfig.Timeout = 30 * time.Second
	defaultConfig.ReadTimeout = 30 * time.Second
	defaultConfig.WriteTimeout = 30 * time.Second
	defaultConfig.Params = map[string]string{"charset": "utf8mb4"}
	defaultConfig.Collation = "utf8mb4_bin"
	defaultConfig.AllowNativePasswords = true
	defaultConfig.TLSConfig = c.TLSConfigKey

	return defaultConfig
}

func (c *Config) GetUploadEndpoint() string {
	return fmt.Sprintf("%s:%d", c.UploadConfig.Host, c.UploadConfig.Port)
}

func (c *Config) GetComputeEngineEndPoint() string {
	return c.ReportConfig.ComputeEngineEndPoint
}

type CentralO11yConfig struct {
	ClientId       string `yaml:"client-id"`
	ClientSecret   string `yaml:"client-secret"`
	GSEndpoint     string `yaml:"gs-endpoint"`
	DSEndpointTPL  string `yaml:"ds-endpoint-tpl"`  // example: www.rs.%s.%s.observability-staging.pingcap.cloud
	ByocDSEndpoint string `yaml:"byoc-ds-endpoint"` // host only, without scheme
	Env            string `yaml:"env"`              // example: dev, staging, prod
}

func (cfg *CentralO11yConfig) Validate() error {
	if len(cfg.ClientId) == 0 {
		return fmt.Errorf("central-o11y.client-id is empty")
	}
	if len(cfg.ClientSecret) == 0 {
		return fmt.Errorf("central-o11y.client-sercret is empty")
	}
	if len(cfg.GSEndpoint) == 0 {
		return fmt.Errorf("central-o11y.gs-endpoint is empty")
	}
	if len(cfg.DSEndpointTPL) == 0 {
		return fmt.Errorf("central-o11y.ds-endpoint-tpl is empty")
	}
	return nil
}

// NewClient builds an o11y client from this block.
// sharedOpts must match the transport options used with the primary central-o11y client in this process
// (e.g. TimeoutOption, MaxIdleConnOption). ByocDSEndpointOption is always taken from this block's byoc-ds-endpoint.
func (cfg *CentralO11yConfig) NewClient(sharedOpts ...o11y.Option) (*o11y.Client, error) {
	if cfg == nil {
		return nil, fmt.Errorf("central o11y config is nil")
	}
	opts := make([]o11y.Option, 0, len(sharedOpts)+1)
	opts = append(opts, sharedOpts...)
	opts = append(opts, o11y.ByocDSEndpointOption(cfg.ByocDSEndpoint))
	return o11y.New(
		cfg.GSEndpoint,
		cfg.DSEndpointTPL,
		cfg.ClientId,
		cfg.ClientSecret,
		cfg.Env,
		opts...,
	)
}

type ActivityFileConfig struct {
	Bucket string `yaml:"bucket"`
	Region string `yaml:"region"`
}

func (cfg *ActivityFileConfig) S3API() (s3iface.S3API, error) {
	awsConfig := &aws.Config{
		Region:           aws.String("us-west-2"),
		S3ForcePathStyle: aws.Bool(true),
	}
	awsSessionOpts := session.Options{
		Config: *awsConfig,
	}
	ses, err := session.NewSessionWithOptions(awsSessionOpts)
	if err != nil {
		return nil, err
	}
	return s3.New(ses), nil
}

type FeatureGates map[FeatureGateName]FeatureGate

type FeatureGate struct {
	// If `All` is true, this feature gate will be enabled for all except DisabledTenants.
	// If `All` is false, this feature gate will be disabled for all except EnabledTenants.
	All bool `yaml:"all" default:"false"`
	// Enable the feature for the specified tenants.
	EnabledOrgs []string `yaml:"enabledOrgs"`
	// Disable the feature for the specified tenants.
	DisabledOrgs []string `yaml:"disabledOrgs"`
}

type FeatureGateName string

const (
	FeatureGateOrgs              FeatureGateName = "orgs"
	FeatureGateMetrics           FeatureGateName = "metrics"
	FeatureGateLogs              FeatureGateName = "logs"
	FeatureGateSlowQuery         FeatureGateName = "slow-query"
	FeatureGateSlowQueryVisual   FeatureGateName = "slow-query-visual"
	FeatureGateTopSQL            FeatureGateName = "top-sql"
	FeatureGateConProf           FeatureGateName = "con-prof"
	FeatureGateBenchmarkReport   FeatureGateName = "benchmark-report"
	FeatureGateComparisonReport  FeatureGateName = "comparison-report"
	FeatureGateSystemCheckReport FeatureGateName = "system-check-report"
	FeatureGateMaterializedView  FeatureGateName = "materialized-view"
)

func (cfg FeatureGates) FeatureEnabled(featureName FeatureGateName, dbaasOrgID uint64) bool {
	gates, ok := cfg[featureName]
	log.Debug("feature gates",
		zap.String("feature", string(featureName)),
		zap.Uint64("orgID", dbaasOrgID),
		zap.Any("gates", gates),
		zap.Bool("ok", ok),
	)
	if !ok {
		return false
	}
	// deny list
	// Check if it's disabled for the org.
	for _, oid := range gates.DisabledOrgs {
		if oid == strconv.FormatUint(dbaasOrgID, 10) {
			return false
		}
	}
	// allow list
	// Check if it's enabled for the org.
	for _, oid := range gates.EnabledOrgs {
		if oid == strconv.FormatUint(dbaasOrgID, 10) {
			return true
		}
	}
	return gates.All
}