package ticdc

import (
	"testing"

	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	operatorconfig "github.com/pingcap/tidb-operator/pkg/apis/util/config"
	"github.com/stretchr/testify/assert"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-api/pkg/provider"
	"k8s.io/utils/pointer"
)

func TestGenerateTiCDCConfig(t *testing.T) {
	testcases := []struct {
		name            string
		cdcSpec         *v1alpha1.TiCDCSpec
		conf            *cloudv1.DPConfiguration
		expectedError   bool
		expectedUpdated bool
	}{
		{
			name: "empty tc config and empty new config",
			cdcSpec: &v1alpha1.TiCDCSpec{
				Config: v1alpha1.NewCDCConfig(),
				ComponentSpec: v1alpha1.ComponentSpec{
					Version: pointer.String(""),
				},
				BaseImage: "gcr.io/pingcap-public/dbaas/ticdc",
			},
			conf:            &cloudv1.DPConfiguration{},
			expectedError:   false,
			expectedUpdated: false,
		},
		{
			name: "empty tc config and new config",
			cdcSpec: &v1alpha1.TiCDCSpec{
				Config: v1alpha1.NewCDCConfig(),
				ComponentSpec: v1alpha1.ComponentSpec{
					Version: pointer.String("v7.5.0"),
				},
				BaseImage: "gcr.io/pingcap-public/dbaas/ticdc",
			},
			conf: &cloudv1.DPConfiguration{
				Config:    mustGenerateSameTiCDCConfigWithoutPredefined(),
				Tag:       "v7.5.0",
				BaseImage: "380838443567.dkr.ecr.us-east-1.amazonaws.com/tidbcloud/ticdc",
			},
			expectedError:   false,
			expectedUpdated: true,
		},
		{
			name: "new config with wrong format",
			cdcSpec: &v1alpha1.TiCDCSpec{
				Config: &v1alpha1.CDCConfigWraper{GenericConfig: generateTCConfigWithPredefined()},
				ComponentSpec: v1alpha1.ComponentSpec{
					Version: pointer.String("v7.5.0"),
				},
				BaseImage: "gcr.io/pingcap-public/dbaas/ticdc",
			},
			conf: &cloudv1.DPConfiguration{
				Config:    mustGenerateJsonTiCDCConfig(),
				Tag:       "v7.5.0",
				BaseImage: "380838443567.dkr.ecr.us-east-1.amazonaws.com/tidbcloud/ticdc",
			},
			expectedError:   true,
			expectedUpdated: false,
		},
		{
			name: "tc config and new config are same",
			cdcSpec: &v1alpha1.TiCDCSpec{
				Config: &v1alpha1.CDCConfigWraper{GenericConfig: generateTCConfigWithPredefined()},
				ComponentSpec: v1alpha1.ComponentSpec{
					Version: pointer.String("v7.5.0"),
				},
				BaseImage: "gcr.io/pingcap-public/dbaas/ticdc",
			},
			conf: &cloudv1.DPConfiguration{
				Config:    mustGenerateSameTiCDCConfigWithoutPredefined(),
				Tag:       "v7.5.0",
				BaseImage: "380838443567.dkr.ecr.us-east-1.amazonaws.com/tidbcloud/ticdc",
			},
			expectedError:   false,
			expectedUpdated: false,
		},
		{
			name: "only predefined config changed",
			cdcSpec: &v1alpha1.TiCDCSpec{
				Config: &v1alpha1.CDCConfigWraper{GenericConfig: generateTCConfigWithPredefined()},
				ComponentSpec: v1alpha1.ComponentSpec{
					Version: pointer.String("v7.5.0"),
				},
				BaseImage: "gcr.io/pingcap-public/dbaas/ticdc",
			},
			conf: &cloudv1.DPConfiguration{
				Config:    mustGenerateNewTiCDCConfigWithPredefinedChanged(),
				Tag:       "v7.5.0",
				BaseImage: "380838443567.dkr.ecr.us-east-1.amazonaws.com/tidbcloud/ticdc",
			},
			expectedError:   false,
			expectedUpdated: false,
		},
		{
			name: "config same but tag changed",
			cdcSpec: &v1alpha1.TiCDCSpec{
				Config: &v1alpha1.CDCConfigWraper{GenericConfig: generateTCConfigWithPredefined()},
				ComponentSpec: v1alpha1.ComponentSpec{
					Version: pointer.String("v7.5.0"),
				},
				BaseImage: "gcr.io/pingcap-public/dbaas/ticdc",
			},
			conf: &cloudv1.DPConfiguration{
				Config:    mustGenerateSameTiCDCConfigWithoutPredefined(),
				Tag:       "v7.6.0",
				BaseImage: "380838443567.dkr.ecr.us-east-1.amazonaws.com/tidbcloud/ticdc",
			},
			expectedError:   false,
			expectedUpdated: true,
		},
		{
			name: "existing config changed",
			cdcSpec: &v1alpha1.TiCDCSpec{
				Config: &v1alpha1.CDCConfigWraper{GenericConfig: generateTCConfigWithPredefined()},
				ComponentSpec: v1alpha1.ComponentSpec{
					Version: pointer.String("v7.5.0"),
				},
				BaseImage: "gcr.io/pingcap-public/dbaas/ticdc",
			},
			conf: &cloudv1.DPConfiguration{
				Config:    mustGenerateNewTiCDCConfigWithNewValueWithoutPredefined(),
				Tag:       "v7.5.0",
				BaseImage: "380838443567.dkr.ecr.us-east-1.amazonaws.com/tidbcloud/ticdc",
			},
			expectedError:   false,
			expectedUpdated: true,
		},
		{
			name: "new config key added",
			cdcSpec: &v1alpha1.TiCDCSpec{
				Config: &v1alpha1.CDCConfigWraper{GenericConfig: generateTCConfigWithPredefined()},
				ComponentSpec: v1alpha1.ComponentSpec{
					Version: pointer.String("v7.5.0"),
				},
				BaseImage: "gcr.io/pingcap-public/dbaas/ticdc",
			},
			conf: &cloudv1.DPConfiguration{
				Config:    mustGenerateNewTiCDCConfigWithNewKeyWithoutPredefined(),
				Tag:       "v7.5.0",
				BaseImage: "380838443567.dkr.ecr.us-east-1.amazonaws.com/tidbcloud/ticdc",
			},
			expectedError:   false,
			expectedUpdated: true,
		},
	}

	for _, c := range testcases {
		t.Run(c.name, func(t *testing.T) {
			ticdcClusterMgr := &NewArchTiCDCCluster{
				ticdcCluster: ticdcCluster{
					cdc:            &provider.Unimplemented{},
					referenceOwner: &cloudv1.Cluster{},
				},
			}
			tc := &v1alpha1.TidbCluster{
				Spec: v1alpha1.TidbClusterSpec{
					TiCDC: c.cdcSpec,
				},
			}
			updated, err := ticdcClusterMgr.generateConfig(tc, c.conf)
			if c.expectedError {
				assert.Error(t, err)
				return
			}

			assert.NoError(t, err)
			if !c.expectedUpdated {
				assert.False(t, updated)
				return
			}

			assert.True(t, updated)
			assert.Equal(t, c.conf.Tag, *tc.Spec.TiCDC.Version)
			assert.Equal(t, c.conf.BaseImage, tc.Spec.TiCDC.BaseImage)
		})
	}
}

func generateTCConfigWithPredefined() *operatorconfig.GenericConfig {
	cdcConf := operatorconfig.New(make(map[string]interface{}))
	cdcConf.Set("data-dir", "a")
	cdcConf.Set("gc-ttl", int64(86400))
	cdcConf.Set("kv-client.worker-concurrent", int64(16))
	return cdcConf
}

func generateTCConfigWithoutPredefined() *operatorconfig.GenericConfig {
	cdcConf := operatorconfig.New(make(map[string]interface{}))
	cdcConf.Set("gc-ttl", int64(86400))
	cdcConf.Set("kv-client.worker-concurrent", int64(16))
	return cdcConf
}

func mustGenerateNewTiCDCConfigWithNewKeyWithoutPredefined() string {
	cdcConf := generateTCConfigWithoutPredefined()
	cdcConf.Set("kv-client.worker-pool-size", int64(4))
	return mustMarshalTOML(cdcConf)
}

func mustGenerateNewTiCDCConfigWithNewValueWithoutPredefined() string {
	cdcConf := generateTCConfigWithoutPredefined()
	cdcConf.Set("gc-ttl", int64(43200))
	return mustMarshalTOML(cdcConf)
}

func mustGenerateNewTiCDCConfigWithPredefinedChanged() string {
	cdcConf := generateTCConfigWithoutPredefined()
	cdcConf.Set("data-dir", "b")
	return mustMarshalTOML(cdcConf)
}

func mustGenerateSameTiCDCConfigWithoutPredefined() string {
	cdcConf := generateTCConfigWithoutPredefined()
	return mustMarshalTOML(cdcConf)
}

func mustGenerateJsonTiCDCConfig() string {
	cdcConf := generateTCConfigWithPredefined()
	return mustMarshalJSON(cdcConf)
}

func mustMarshalTOML(config *operatorconfig.GenericConfig) string {
	confBytes, err := config.MarshalTOML()
	if err != nil {
		panic(err)
	}
	return string(confBytes)
}

func mustMarshalJSON(config *operatorconfig.GenericConfig) string {
	confBytes, err := config.MarshalJSON()
	if err != nil {
		panic(err)
	}
	return string(confBytes)
}