package config

import (
	"io/ioutil"
	"os"
	"testing"
	"time"

	"github.com/imdario/mergo"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

func TestAPIConfig(t *testing.T) {
	str := `
infraAPI:

# the flowing are just required
server:
  pocManualBackupReservedDays: 1
  pocAutoBackupReservedDays: 2
  onDemandAutoBackupReservedDays: 3
  frontendDNS: aa
  backendDNS: bb
  stage: stage
  centralRoleARN: centralRoleARN
session:
    secret: secret
`

	f, err := ioutil.TempFile("", "config-test*.yaml")
	assert.Nil(t, err)
	defer func() {
		_ = os.Remove(f.Name())
	}()
	_, err = f.WriteString(str)
	assert.Nil(t, err)

	err = InitConfig(f.Name())
	assert.Nil(t, err)
	require.Equal(t, 30*time.Minute, Global().DMConfig.PreCheckTimeout)
}

func TestHibernationConfig(t *testing.T) {
	str := `
hibernation:
  dryRun: true
  idleDurationThresholdMinutes: 120
  checkIdleIntervalMinutes: 10

# the flowing are just required
server:
  pocManualBackupReservedDays: 1
  pocAutoBackupReservedDays: 2
  onDemandAutoBackupReservedDays: 3
  frontendDNS: aa
  backendDNS: bb
  stage: stage
  centralRoleARN: centralRoleARN
session:
    secret: secret
dm:
    preCheckTimeout: 10m
`
	f, err := ioutil.TempFile("", "config-test*.yaml")
	assert.Nil(t, err)
	defer func() {
		_ = os.Remove(f.Name())
	}()
	_, err = f.WriteString(str)
	assert.Nil(t, err)

	err = InitConfig(f.Name())
	assert.Nil(t, err)
	require.Equal(t, 10*time.Minute, Global().DMConfig.PreCheckTimeout)
}

type TestMapIn struct {
	A string
	B string
	C string
}

type TestMap struct {
	Test TestMapIn
}

func TestMergoConfig(t *testing.T) {
	assert := assert.New(t)
	a := TestMap{
		TestMapIn{
			A: "a",
			C: "c1",
		},
	}
	b := TestMap{
		TestMapIn{
			B: "b",
			C: "c2",
		},
	}

	err := mergo.Merge(&a, b, mergo.WithOverride)
	assert.NoError(err)
	assert.Equal(a.Test.B, "b")
	assert.Equal(a.Test.A, "a")
	assert.Equal(a.Test.C, "c2")
}

func TestConfigRemote(t *testing.T) {
	t.Skip("not able test config server in git pr test")
	str := `
infraAPI:

# the flowing are just required
server:
  pocManualBackupReservedDays: 1
  pocAutoBackupReservedDays: 2
  onDemandAutoBackupReservedDays: 3
  frontendDNS: aa
  backendDNS: bb
  stage: stage
  centralRoleARN: centralRoleARN
session:
    secret: secret
`

	f, err := ioutil.TempFile("", "config-test*.yaml")
	assert.Nil(t, err)
	defer func() {
		_ = os.Remove(f.Name())
	}()
	_, err = f.WriteString(str)
	assert.Nil(t, err)

	err = InitConfig(f.Name())
	assert.Nil(t, err)
}

func TestAllowListMatcher(t *testing.T) {
	m := AllowListMatcher{
		TenantID:  []string{"1", "2"},
		ProjectID: []string{"1", "2"},
		ClusterID: []string{"1", "2"},
	}
	// without cluster property matcher
	require.False(t, m.Match(11, 11, 11, "us-west-2", "eks"))
	require.True(t, m.Match(1, 11, 11, "us-west-2", "eks"))
	require.True(t, m.Match(11, 1, 11, "us-west-2", "eks"))
	require.True(t, m.Match(11, 11, 1, "us-west-2", "eks"))
	require.True(t, m.Match(11, 11, 2, "us-west-2", "eks"))

	// with cluster property matcher
	m.DeploymentMatcher = &DeploymentMatcher{
		Regions:   []string{"us-east-2"},
		Providers: []string{"eks", "gke"},
	}
	require.False(t, m.Match(11, 11, 11, "us-east-2", "gke"))
	require.False(t, m.Match(11, 11, 2, "us-west-2", "eks"))
	require.False(t, m.Match(11, 11, 2, "us-west-2", "gke"))
	require.True(t, m.Match(11, 11, 2, "us-east-2", "gke"))
}

func TestClusterPropertyMatcher(t *testing.T) {
	matcher := DeploymentMatcher{
		Regions:   []string{"us-east-2", "us-west-2"},
		Providers: []string{"eks", "gke"},
	}
	require.False(t, matcher.Match("xx", "eks"))
	require.False(t, matcher.Match("us-east-2", "xxx"))
	require.True(t, matcher.Match("us-east-2", "eks"))
	require.True(t, matcher.Match("us-east-2", "gke"))
	require.True(t, matcher.Match("us-west-2", "gke"))
}

func TestGetMatchedSQLMode(t *testing.T) {
	testcases := []struct {
		name            string
		conf            ChangefeedConfig
		tenantID        uint64
		projectID       uint64
		clusterID       uint64
		expectedSQLMode string
	}{
		{
			name: "empty config",
			conf: ChangefeedConfig{
				SQLModeConfig: []SQLModeConfig{},
			},
			tenantID:        1,
			projectID:       11,
			clusterID:       111,
			expectedSQLMode: "",
		},
		{
			name: "no matched sql mode",
			conf: ChangefeedConfig{
				SQLModeConfig: []SQLModeConfig{
					{
						SQLMode:   "a",
						TenantID:  []string{"2"},
						ProjectID: []string{"22"},
						ClusterID: []string{"222"},
					},
				},
			},
			tenantID:        1,
			projectID:       11,
			clusterID:       111,
			expectedSQLMode: "",
		},
		{
			name: "match tenant id",
			conf: ChangefeedConfig{
				SQLModeConfig: []SQLModeConfig{
					{
						SQLMode:   "a",
						TenantID:  []string{"1"},
						ProjectID: []string{"22"},
						ClusterID: []string{"222"},
					},
				},
			},
			tenantID:        1,
			projectID:       11,
			clusterID:       111,
			expectedSQLMode: "a",
		},
		{
			name: "match project id",
			conf: ChangefeedConfig{
				SQLModeConfig: []SQLModeConfig{
					{
						SQLMode:   "a",
						TenantID:  []string{"2"},
						ProjectID: []string{"11"},
						ClusterID: []string{"222"},
					},
				},
			},
			tenantID:        1,
			projectID:       11,
			clusterID:       111,
			expectedSQLMode: "a",
		},
		{
			name: "match cluster id",
			conf: ChangefeedConfig{
				SQLModeConfig: []SQLModeConfig{
					{
						SQLMode:   "a",
						TenantID:  []string{"2"},
						ProjectID: []string{"22"},
						ClusterID: []string{"111"},
					},
				},
			},
			tenantID:        1,
			projectID:       11,
			clusterID:       111,
			expectedSQLMode: "a",
		},
		{
			name: "multi matches",
			conf: ChangefeedConfig{
				SQLModeConfig: []SQLModeConfig{
					{
						SQLMode:   "b",
						TenantID:  []string{"1"},
						ProjectID: []string{"22"},
						ClusterID: []string{"222"},
					},
					{
						SQLMode:   "a",
						TenantID:  []string{"2"},
						ProjectID: []string{"11"},
						ClusterID: []string{"222"},
					},
				},
			},
			tenantID:        1,
			projectID:       11,
			clusterID:       111,
			expectedSQLMode: "b",
		},
	}

	for _, tc := range testcases {
		t.Run(tc.name, func(t *testing.T) {
			matchedSQLMode := tc.conf.GetMatchedSQLMode(tc.tenantID, tc.projectID, tc.clusterID)
			assert.Equal(t, tc.expectedSQLMode, matchedSQLMode)
		})
	}
}

// func TestBizChangefeedFeatureGate(t *testing.T) {
// 	cloudProvider := "aws"
// 	downstreamType := dataflow.ConnectorType_MYSQL
// 	testcases := []struct {
// 		featureGate BizChangefeedFeatureGate
// 		enabled     bool
// 	}{
// 		{
// 			featureGate: BizChangefeedFeatureGate{},
// 			enabled:     false,
// 		},
// 		{
// 			featureGate: BizChangefeedFeatureGate{
// 				EnabledCloudProviders:  []string{},
// 				EnabledDownstreamTypes: []int32{},
// 			},
// 			enabled: false,
// 		},
// 		{
// 			featureGate: BizChangefeedFeatureGate{
// 				EnabledCloudProviders:  []string{"aws"},
// 				EnabledDownstreamTypes: []int32{},
// 			},
// 			enabled: false,
// 		},
// 		{
// 			featureGate: BizChangefeedFeatureGate{
// 				EnabledCloudProviders:  []string{},
// 				EnabledDownstreamTypes: []int32{int32(dataflow.ConnectorType_MYSQL)},
// 			},
// 			enabled: false,
// 		},
// 		{
// 			featureGate: BizChangefeedFeatureGate{
// 				EnabledCloudProviders:  []string{"aws"},
// 				EnabledDownstreamTypes: []int32{int32(dataflow.ConnectorType_MYSQL)},
// 			},
// 			enabled: true,
// 		},
// 		{
// 			featureGate: BizChangefeedFeatureGate{
// 				EnabledCloudProviders:  []string{"gcp", "aws"},
// 				EnabledDownstreamTypes: []int32{int32(dataflow.ConnectorType_MYSQL)},
// 			},
// 			enabled: true,
// 		},
// 		{
// 			featureGate: BizChangefeedFeatureGate{
// 				EnabledCloudProviders: []string{"aws"},
// 				EnabledDownstreamTypes: []int32{
// 					int32(dataflow.ConnectorType_KAFKA), int32(dataflow.ConnectorType_MYSQL)},
// 			},
// 			enabled: true,
// 		},
// 		{
// 			featureGate: BizChangefeedFeatureGate{
// 				EnabledCloudProviders: []string{"gcp", "aws"},
// 				EnabledDownstreamTypes: []int32{
// 					int32(dataflow.ConnectorType_KAFKA), int32(dataflow.ConnectorType_MYSQL)},
// 			},
// 			enabled: true,
// 		},
// 	}

// 	for _, tc := range testcases {
// 		enabled := tc.featureGate.IsEnabled(cloudProvider, downstreamType)
// 		assert.Equal(t, tc.enabled, enabled)
// 	}
// }

func TestGetGCPDualRegionCandidateSet(t *testing.T) {
	testcases := []struct {
		region     string
		candidates map[string]struct{}
	}{
		{
			region: "asia-northeast1",
			candidates: map[string]struct{}{
				"asia-northeast2": {},
			},
		},
		{
			region: "asia-northeast2",
			candidates: map[string]struct{}{
				"asia-northeast1": {},
			},
		},
		{
			region: "us-east5",
			candidates: map[string]struct{}{
				"us-central1": {},
				"us-east1":    {},
			},
		},
		{
			region: "us-east1",
			candidates: map[string]struct{}{
				"us-central1": {},
				"us-east4":    {},
				"us-east5":    {},
				"us-south1":   {},
				"us-west1":    {},
				"us-west2":    {},
				"us-west3":    {},
				"us-west4":    {},
			},
		},
		{
			region: "asia-south1",
			candidates: map[string]struct{}{
				"asia-south2": {},
			},
		},
		{
			region: "asia-south2",
			candidates: map[string]struct{}{
				"asia-south1": {},
			},
		},
		{
			region: "europe-west2",
			candidates: map[string]struct{}{
				"europe-west1": {},
				"europe-west3": {},
			},
		},
		{
			region: "europe-west1",
			candidates: map[string]struct{}{
				"europe-central2":   {},
				"europe-north1":     {},
				"europe-southwest1": {},
				"europe-west3":      {},
				"europe-west4":      {},
				"europe-west8":      {},
				"europe-west9":      {},
				"europe-west2":      {},
			},
		},
		{
			region:     "asia-east2",
			candidates: map[string]struct{}{},
		},
	}

	InitGCPDualRegionConfigForTest()
	for _, tc := range testcases {
		candidates := Global().ServerConfig.BucketOptions.GetGCPDualRegionCandidateSet(tc.region)
		assert.Equal(t, tc.candidates, candidates)
	}
}

func TestGetGCPDualRegionPredefinedName(t *testing.T) {
	testcases := []struct {
		region1, region2 string
		predefinedName   string
	}{
		{
			region1:        "asia-northeast1",
			region2:        "asia-northeast2",
			predefinedName: "asia1",
		},
		{
			region1:        "europe-north1",
			region2:        "europe-west4",
			predefinedName: "eur4",
		},
		{
			region1:        "us-central1",
			region2:        "us-east1",
			predefinedName: "nam4",
		},
		{
			region1:        "asia-northeast1",
			region2:        "asia-east2",
			predefinedName: "",
		},
		{
			region1:        "asia-east1",
			region2:        "asia-east2",
			predefinedName: "",
		},
	}

	InitGCPDualRegionConfigForTest()
	for _, tc := range testcases {
		name := Global().ServerConfig.BucketOptions.GetGCPDualRegionPredefinedName(tc.region1, tc.region2)
		assert.Equal(t, tc.predefinedName, name)
	}
}

func TestGetGCPDualRegionLocationCode(t *testing.T) {
	testcases := []struct {
		region1, region2 string
		locationCode     string
	}{
		{
			region1:      "asia-east1",
			region2:      "asia-southeast1",
			locationCode: "asia",
		},
		{
			region1:      "australia-southeast1",
			region2:      "australia-southeast2",
			locationCode: "au",
		},
		{
			region1:      "northamerica-northeast1",
			region2:      "northamerica-northeast2",
			locationCode: "ca",
		},
		{
			region1:      "europe-central2",
			region2:      "europe-north1",
			locationCode: "eu",
		},
		{
			region1:      "asia-south1",
			region2:      "asia-south2",
			locationCode: "in",
		},
		{
			region1:      "us-central1",
			region2:      "us-east1",
			locationCode: "us",
		},
		{
			region1:      "asia-east1",
			region2:      "asia-northeast1",
			locationCode: "",
		},
		{
			region1:      "asia-east1",
			region2:      "us-central1",
			locationCode: "",
		},
		{
			region1:      "asia-northeast1",
			region2:      "asia-northeast2",
			locationCode: "",
		},
	}

	InitGCPDualRegionConfigForTest()
	for _, tc := range testcases {
		code := Global().ServerConfig.BucketOptions.GetGCPDualRegionLocationCode(tc.region1, tc.region2)
		assert.Equal(t, tc.locationCode, code)
	}
}

func TestChangefeedConfig_GetEnabledFeatures(t *testing.T) {
	changefeedConfig := &ChangefeedConfig{
		FeatureEnabledVersions: []*ChangefeedFeatureEnabledVersions{
			{
				Name:     "csv-header",
				NewArch:  true,
				Versions: []string{">= v8.5.4"},
			},
			{
				Name:     "pulsar-sink",
				NewArch:  false,
				Versions: []string{">= v7.5.1"},
			},
			{
				Name:     "output-raw-change-event",
				NewArch:  false,
				Versions: []string{"~v6.5.10", ">= v7.1.6, < v7.2.0", "~v7.5.3", ">= v8.1.1"},
			},
		},
	}
	err := changefeedConfig.Init()
	require.Nil(t, err)

	testcases := []struct {
		newArch          bool
		version          string
		err              bool
		expectedFeatures []string
	}{
		{
			version: "a6.5.0",
			err:     true,
		},
		{
			newArch:          false,
			version:          "v6.5.0",
			err:              false,
			expectedFeatures: []string{},
		},
		{
			newArch:          false,
			version:          "v6.5.10",
			err:              false,
			expectedFeatures: []string{"output-raw-change-event"},
		},
		{
			newArch:          false,
			version:          "v7.0.2",
			err:              false,
			expectedFeatures: []string{},
		},
		{
			newArch:          false,
			version:          "v7.1.10",
			err:              false,
			expectedFeatures: []string{"output-raw-change-event"},
		},
		{
			newArch:          false,
			version:          "v7.3.0",
			err:              false,
			expectedFeatures: []string{},
		},
		{
			newArch:          false,
			version:          "v7.5.1",
			err:              false,
			expectedFeatures: []string{"pulsar-sink"},
		},
		{
			newArch:          false,
			version:          "v7.5.5",
			err:              false,
			expectedFeatures: []string{"output-raw-change-event", "pulsar-sink"},
		},
		{
			newArch:          false,
			version:          "v7.6.0",
			err:              false,
			expectedFeatures: []string{"pulsar-sink"},
		},
		{
			newArch:          false,
			version:          "v8.5.4",
			err:              false,
			expectedFeatures: []string{"output-raw-change-event", "pulsar-sink"},
		},
		{
			newArch:          true,
			version:          "v8.5.4",
			err:              false,
			expectedFeatures: []string{"csv-header", "output-raw-change-event", "pulsar-sink"},
		},
		{
			newArch:          true,
			version:          "v8.5.5-release.1", // there is no constraint with pre-release version
			err:              false,
			expectedFeatures: []string{},
		},
	}

	for _, tc := range testcases {
		features, err := changefeedConfig.GetEnabledFeatures(tc.version, tc.newArch)
		if tc.err {
			assert.Error(t, err, "version: %s", tc.version)
		} else {
			assert.NoError(t, err, "version: %s", tc.version)
			assert.Equal(t, tc.expectedFeatures, features, "version: %s", tc.version)
		}
	}
}