package domain

import (
	"net/url"
	"testing"

	"github.com/AlekSi/pointer"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

func Test_TiCDCChangefeedMySQLURIConfigData_Generate(t *testing.T) {
	uriConfig := &TiCDCChangefeedMySQLURIConfigData{
		Endpoint:              pointer.ToString("abc:3306"),
		User:                  pointer.ToString("user-abc"),
		Password:              pointer.ToString("password-abc"),
		MaxTxnRow:             pointer.ToInt(1000),
		WorkerCount:           pointer.ToInt(16),
		CachePrepStmts:        pointer.ToBool(true),
		TimeZone:              pointer.ToString("Asia/Shanghai"),
		TransactionAtomicity:  pointer.ToString("Atomicity"),
		BatchDmlEnable:        pointer.ToBool(true),
		MultiStmtEnable:       pointer.ToBool(true),
		MaxMultiUpdateRow:     pointer.ToInt32(800),
		MaxMultiUpdateRowSize: pointer.ToInt64(1024),
		SslCa:                 pointer.ToString("ssl-ca"),
		SslCert:               pointer.ToString("ssl-cert"),
		SslKey:                pointer.ToString("ssl-key"),
		ReadTimeout:           pointer.ToString("2m"),
		WriteTimeout:          pointer.ToString("5m"),
		Timeout:               pointer.ToString("10m"),
		TidbTxnMode:           pointer.ToString("txn-mode-a"),
		SafeMode:              pointer.ToBool(true),
	}

	expectedQuery := url.Values{
		"max-txn-row":               []string{"1000"},
		"worker-count":              []string{"16"},
		"cache-prep-stmts":          []string{"true"},
		"time-zone":                 []string{"Asia/Shanghai"},
		"transaction-atomicity":     []string{"Atomicity"},
		"batch-dml-enable":          []string{"true"},
		"multi-stmt-enable":         []string{"true"},
		"max-multi-update-row":      []string{"800"},
		"max-multi-update-row-size": []string{"1024"},
		"ssl-ca":                    []string{"ssl-ca"},
		"ssl-cert":                  []string{"ssl-cert"},
		"ssl-key":                   []string{"ssl-key"},
		"read-timeout":              []string{"2m"},
		"write-timeout":             []string{"5m"},
		"timeout":                   []string{"10m"},
		"tidb-txn-mode":             []string{"txn-mode-a"},
		"safe-mode":                 []string{"true"},
	}

	sinkUri := uriConfig.GenerateURI()
	sinkU, err := url.Parse(sinkUri)
	require.NoError(t, err)

	assert.Equal(t, "mysql", sinkU.Scheme)
	assert.Equal(t, "user-abc", sinkU.User.Username())
	password, ok := sinkU.User.Password()
	assert.True(t, ok)
	assert.Equal(t, "password-abc", password)
	assert.Equal(t, "abc:3306", sinkU.Host)

	assert.Equal(t, expectedQuery, sinkU.Query())
}

func Test_TiCDCChangefeedKafkaURIConfigData_Generate(t *testing.T) {
	uriConfig := &TiCDCChangefeedKafkaURIConfigData{
		BrokerEndpoints:                "abc:9092,def:9092,ghi:9092",
		DefaultTopic:                   "topic-a",
		Protocol:                       "canal-json",
		EnableTLS:                      true,
		ReplicationFactor:              3,
		PartitionNum:                   16,
		KafkaVersion:                   "v2",
		Compression:                    pointer.ToString("snappy"),
		SASLMechanism:                  pointer.ToString("plain"),
		SASLUser:                       pointer.ToString("user"),
		SASLPassword:                   pointer.ToString("password"),
		AvroDecimalHandlingMode:        pointer.ToString("decimal"),
		AvroBigintUnsignedHandlingMode: pointer.ToString("bigint"),
		EnableTiDBExtension:            pointer.ToBool(true),
		AutoCreateTopic:                pointer.ToBool(true),
		InsecureSkipVerify:             pointer.ToBool(true),
		MaxMessageBytes:                pointer.ToUint64(4096),
		DialTimeout:                    pointer.ToString("5m"),
		ReadTimeout:                    pointer.ToString("2m"),
		WriteTimeout:                   pointer.ToString("3m"),
		KafkaClientID:                  pointer.ToString("client-id-123"),
		RequiredAcks:                   pointer.ToInt32(-1),
		MaxBatchSize:                   pointer.ToUint64(2048),
		Ca:                             pointer.ToString("ssl-ca"),
		Cert:                           pointer.ToString("ssl-cert"),
		Key:                            pointer.ToString("ssl-key"),
		SASLGssapiAuthType:             pointer.ToString("gssapi-auth"),
		SASLGssapiKeytabPath:           pointer.ToString("gssapi-key-tab-path"),
		SASLGssapiKerberosConfigPath:   pointer.ToString("gssapi-kerberos-config-path"),
		SASLGssapiServiceName:          pointer.ToString("gssapi-service-name"),
		SASLGssapiUser:                 pointer.ToString("gssapi-user"),
		SASLGssapiPassword:             pointer.ToString("gssapi-password"),
		SASLGssapiRealm:                pointer.ToString("gssapi-realm"),
		SASLGssapiDisablePafxfast:      pointer.ToBool(true),
	}

	expectedQuery := url.Values{
		"protocol":                           []string{"canal-json"},
		"enable-tls":                         []string{"true"},
		"replication-factor":                 []string{"3"},
		"partition-num":                      []string{"16"},
		"kafka-version":                      []string{"v2"},
		"compression":                        []string{"snappy"},
		"sasl-mechanism":                     []string{"plain"},
		"sasl-user":                          []string{"user"},
		"sasl-password":                      []string{"password"},
		"avro-decimal-handling-mode":         []string{"decimal"},
		"avro-bigint-unsigned-handling-mode": []string{"bigint"},
		"enable-tidb-extension":              []string{"true"},
		"auto-create-topic":                  []string{"true"},
		"insecure-skip-verify":               []string{"true"},
		"max-message-bytes":                  []string{"4096"},
		"dial-timeout":                       []string{"5m"},
		"read-timeout":                       []string{"2m"},
		"write-timeout":                      []string{"3m"},
		"kafka-client-id":                    []string{"client-id-123"},
		"required-acks":                      []string{"-1"},
		"max-batch-size":                     []string{"2048"},
		"ca":                                 []string{"ssl-ca"},
		"cert":                               []string{"ssl-cert"},
		"key":                                []string{"ssl-key"},
		"sasl-gssapi-auth-type":              []string{"gssapi-auth"},
		"sasl-gssapi-keytab-path":            []string{"gssapi-key-tab-path"},
		"sasl-gssapi-kerberos-config-path":   []string{"gssapi-kerberos-config-path"},
		"sasl-gssapi-service-name":           []string{"gssapi-service-name"},
		"sasl-gssapi-user":                   []string{"gssapi-user"},
		"sasl-gssapi-password":               []string{"gssapi-password"},
		"sasl-gssapi-realm":                  []string{"gssapi-realm"},
		"sasl-gssapi-disable-pafxfast":       []string{"true"},
	}

	sinkUri := uriConfig.GenerateURI()
	sinkU, err := url.Parse(sinkUri)
	require.NoError(t, err)

	assert.Equal(t, "kafka", sinkU.Scheme)
	assert.Equal(t, "abc:9092,def:9092,ghi:9092", sinkU.Host)
	assert.Equal(t, "/topic-a", sinkU.Path)
	assert.Equal(t, expectedQuery, sinkU.Query())

	uriConfig.KafkaClientID = pointer.ToString("")
	assert.NotContains(t, uriConfig.GenerateURI(), "kafka-client-id=")
}

func Test_TiCDCChangefeedPulsarURIConfigData_Generate(t *testing.T) {
	uriConfig := &TiCDCChangefeedPulsarURIConfigData{
		PulsarProtocol:      "pulsar",
		Endpoint:            "abcd:29095",
		PulsarTenant:        "tenant",
		PulsarNamespace:     "namespace",
		DefaultTopic:        "topic-c",
		DataProtocol:        "avro",
		EnableTiDBExtension: pointer.ToBool(true),
	}

	expectedQuery := url.Values{
		"protocol":              []string{"avro"},
		"enable-tidb-extension": []string{"true"},
	}

	sinkUri := uriConfig.GenerateURI()
	sinkU, err := url.Parse(sinkUri)
	require.NoError(t, err)

	assert.Equal(t, "pulsar", sinkU.Scheme)
	assert.Equal(t, "abcd:29095", sinkU.Host)
	assert.Equal(t, "/persistent://tenant/namespace/topic-c", sinkU.Path)
	assert.Equal(t, expectedQuery, sinkU.Query())
}

func Test_TiCDCChangefeedS3URIConfigData_Generate(t *testing.T) {
	t.Run("ak sk auth", func(t *testing.T) {
		uriConfig := &TiCDCChangefeedS3URIConfigData{
			S3URI:                "s3://bucket/prefix",
			AccessKey:            pointer.ToString("accesskey"),
			SecretAccessKey:      pointer.ToString("secretkey"),
			ForcePathStyle:       pointer.ToBool(true),
			FlushInterval:        "10m",
			FileSize:             2048,
			TransactionAtomicity: pointer.ToString("atomicity"),
			EnableTiDBExtension:  pointer.ToBool(true),
			WorkerCount:          pointer.ToUint32(8),
			Protocol:             pointer.ToString("csv"),
		}

		expectedQuery := url.Values{
			"access-key":            []string{"accesskey"},
			"secret-access-key":     []string{"secretkey"},
			"force-path-style":      []string{"true"},
			"flush-interval":        []string{"10m"},
			"file-size":             []string{"2048"},
			"transaction-atomicity": []string{"atomicity"},
			"enable-tidb-extension": []string{"true"},
			"worker-count":          []string{"8"},
			"protocol":              []string{"csv"},
		}

		sinkUri := uriConfig.GenerateURI()
		sinkU, err := url.Parse(sinkUri)
		require.NoError(t, err)

		assert.Equal(t, "s3", sinkU.Scheme)
		assert.Equal(t, "bucket", sinkU.Host)
		assert.Equal(t, "/prefix", sinkU.Path)
		assert.Equal(t, expectedQuery, sinkU.Query())
	})

	t.Run("iam role auth", func(t *testing.T) {
		uriConfig := &TiCDCChangefeedS3URIConfigData{
			S3URI:                "s3://bucket/prefix",
			RoleARN:              pointer.ToString("role123"),
			ExternalID:           pointer.ToString("123"),
			ForcePathStyle:       pointer.ToBool(true),
			FlushInterval:        "10m",
			FileSize:             2048,
			TransactionAtomicity: pointer.ToString("atomicity"),
			EnableTiDBExtension:  pointer.ToBool(true),
			WorkerCount:          pointer.ToUint32(8),
			Protocol:             pointer.ToString("csv"),
		}

		expectedQuery := url.Values{
			"role-arn":              []string{"role123"},
			"external-id":           []string{"123"},
			"force-path-style":      []string{"true"},
			"flush-interval":        []string{"10m"},
			"file-size":             []string{"2048"},
			"transaction-atomicity": []string{"atomicity"},
			"enable-tidb-extension": []string{"true"},
			"worker-count":          []string{"8"},
			"protocol":              []string{"csv"},
		}

		sinkUri := uriConfig.GenerateURI()
		sinkU, err := url.Parse(sinkUri)
		require.NoError(t, err)

		assert.Equal(t, "s3", sinkU.Scheme)
		assert.Equal(t, "bucket", sinkU.Host)
		assert.Equal(t, "/prefix", sinkU.Path)
		assert.Equal(t, expectedQuery, sinkU.Query())
	})
}

func Test_TiCDCChangefeedGcsURIConfigData_Generate(t *testing.T) {
	uriConfig := &TiCDCChangefeedGcsURIConfigData{
		GcsURI:               "gcs://bucket/prefix",
		ForcePathStyle:       pointer.ToBool(true),
		FlushInterval:        "10m",
		FileSize:             2048,
		TransactionAtomicity: pointer.ToString("atomicity"),
		EnableTiDBExtension:  pointer.ToBool(true),
		WorkerCount:          pointer.ToUint32(8),
		Protocol:             pointer.ToString("csv"),
	}

	expectedQuery := url.Values{
		"force-path-style":      []string{"true"},
		"flush-interval":        []string{"10m"},
		"file-size":             []string{"2048"},
		"transaction-atomicity": []string{"atomicity"},
		"enable-tidb-extension": []string{"true"},
		"worker-count":          []string{"8"},
		"protocol":              []string{"csv"},
	}

	sinkUri := uriConfig.GenerateURI()
	sinkU, err := url.Parse(sinkUri)
	require.NoError(t, err)

	assert.Equal(t, "gcs", sinkU.Scheme)
	assert.Equal(t, "bucket", sinkU.Host)
	assert.Equal(t, "/prefix", sinkU.Path)
	assert.Equal(t, expectedQuery, sinkU.Query())
}

func Test_TiCDCChangefeedAzureBlobStorageURIConfigData_Generate(t *testing.T) {
	uriConfig := &TiCDCChangefeedAzureBlobStorageURIConfigData{
		BlobPath:             "container/prefix",
		AccountName:          "storageaccount",
		SASToken:             pointer.ToString(";token%"),
		ForcePathStyle:       pointer.ToBool(true),
		FlushInterval:        "10m",
		FileSize:             2048,
		TransactionAtomicity: pointer.ToString("atomicity"),
		EnableTiDBExtension:  pointer.ToBool(true),
		WorkerCount:          pointer.ToUint32(8),
		Protocol:             pointer.ToString("csv"),
	}

	expectedQuery := url.Values{
		"account-name":          []string{"storageaccount"},
		"sas-token":             []string{";token%"},
		"force-path-style":      []string{"true"},
		"flush-interval":        []string{"10m"},
		"file-size":             []string{"2048"},
		"transaction-atomicity": []string{"atomicity"},
		"enable-tidb-extension": []string{"true"},
		"worker-count":          []string{"8"},
		"protocol":              []string{"csv"},
	}

	sinkUri := uriConfig.GenerateURI()
	sinkU, err := url.Parse(sinkUri)
	require.NoError(t, err)

	assert.Equal(t, "azure", sinkU.Scheme)
	assert.Equal(t, "container", sinkU.Host)
	assert.Equal(t, "/prefix", sinkU.Path)
	assert.Equal(t, expectedQuery, sinkU.Query())
}