package service

import (
	"context"
	"net/url"
	"testing"

	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/config-center/proto/config_center"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/clusterproxy"
	"github.com/tidbcloud/pkgv2/clusterproxy/serverless"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	"github.com/tidbcloud/serverless-service/internal/grpc/grpcli"
	"google.golang.org/protobuf/encoding/protowire"
)

type changefeedConfigCenterStub struct {
	grpcli.ConfigCenterClient
	files        map[string]*config_center.ConfigurationFile
	createdFiles []string
}

type invalidReplicaConfig struct {
	Unsupported chan int `json:"unsupported"`
}

func (s *changefeedConfigCenterStub) GetClusterConfigurationFile(
	_ context.Context, req *config_center.GetClusterConfigurationFileRequest,
) (*config_center.ConfigurationFile, bizerr.BizError) {
	file, ok := s.files[req.GetConfigurationFilename()]
	if !ok {
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "configuration not found")
	}
	return file, nil
}

func (s *changefeedConfigCenterStub) SearchDefaultConfigurationTemplate(
	_ context.Context, _ *config_center.SearchDefaultConfigurationTemplateRequest,
) (*config_center.ConfigurationTemplate, bizerr.BizError) {
	return &config_center.ConfigurationTemplate{TemplateId: 1}, nil
}

func (s *changefeedConfigCenterStub) UpdateClusterMeta(
	_ context.Context, req *config_center.UpdateClusterMetaRequest,
) (*config_center.UpdateClusterMetaResponse, bizerr.BizError) {
	if s.files == nil {
		s.files = make(map[string]*config_center.ConfigurationFile)
	}
	for _, fileReq := range req.GetConfigurationFileRequests() {
		filename := fileReq.GetConfigurationFilename()
		s.files[filename] = &config_center.ConfigurationFile{
			ConfigurationFilename: filename,
			Text:                  `{}`,
		}
		s.createdFiles = append(s.createdFiles, filename)
	}
	return &config_center.UpdateClusterMetaResponse{}, nil
}

func TestChangefeedDownstreamType(t *testing.T) {
	tests := []struct {
		name           string
		changefeed     *entity.CDCChangefeed
		downstreamType string
	}{
		{
			name:           "kafka",
			changefeed:     &entity.CDCChangefeed{SinkType: constant.ChangefeedTypeKafka},
			downstreamType: "kafka",
		},
		{
			name:           "mysql",
			changefeed:     &entity.CDCChangefeed{SinkType: constant.ChangefeedTypeMySQL},
			downstreamType: "mysql",
		},
		{
			name:           "s3",
			changefeed:     cloudStorageChangefeed(constant.CloudStorageTypeS3),
			downstreamType: "s3",
		},
		{
			name:           "oss uses s3",
			changefeed:     cloudStorageChangefeed(constant.CloudStorageTypeOSS),
			downstreamType: "s3",
		},
		{
			name:           "gcs",
			changefeed:     cloudStorageChangefeed(constant.CloudStorageTypeGCS),
			downstreamType: "gcs",
		},
		{
			name:           "azure blob",
			changefeed:     cloudStorageChangefeed(constant.CloudStorageTypeAzureBlob),
			downstreamType: "azblob",
		},
		{
			name:           "tidb cloud",
			changefeed:     cloudStorageChangefeed(constant.CloudStorageTypeTiDBCloud),
			downstreamType: "serverless_tidb",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			downstreamType, err := changefeedDownstreamType(tt.changefeed)
			require.NoError(t, err)
			require.Equal(t, tt.downstreamType, downstreamType)
		})
	}
}

func TestEnsureChangefeedConfigurationsCreatesFilesForExistingChangefeed(t *testing.T) {
	client := &changefeedConfigCenterStub{files: map[string]*config_center.ConfigurationFile{}}
	changefeed := &entity.CDCChangefeed{
		ID:       "existing-changefeed",
		SinkType: constant.ChangefeedTypeKafka,
	}

	err := ensureChangefeedConfigurationsWithClient(context.Background(), client, changefeed)
	require.NoError(t, err)
	require.Contains(t, client.files, essentialChangefeedConfigFileName)
	require.Contains(t, client.files, essentialChangefeedURIFileName)
	require.ElementsMatch(t, []string{
		essentialChangefeedConfigFileName,
		essentialChangefeedURIFileName,
	}, client.createdFiles)
}

func TestNewChangefeedConfigUpdateRequest(t *testing.T) {
	t.Parallel()

	withParams := newChangefeedConfigUpdateRequest(
		essentialChangefeedConfigFileName,
		map[string]*config_center.ParamItem{"force-replicate": {}},
		nil,
	)
	require.Len(t, withParams.GetCustomParams(), 1)
	require.Empty(t, withParams.ProtoReflect().GetUnknown())

	withoutParams := newChangefeedConfigUpdateRequest(
		essentialChangefeedConfigFileName,
		map[string]*config_center.ParamItem{},
		nil,
	)
	require.Empty(t, withoutParams.GetCustomParams())
	require.Equal(t,
		protowire.AppendVarint(
			protowire.AppendTag(nil, clearCustomParamsFieldNumber, protowire.VarintType),
			1,
		),
		[]byte(withoutParams.ProtoReflect().GetUnknown()),
	)
}

func cloudStorageChangefeed(storageType constant.CloudStorageType) *entity.CDCChangefeed {
	return &entity.CDCChangefeed{
		SinkType: constant.ChangefeedTypeCloudStorage,
		CloudStorageInfo: &entity.CloudStorageInfo{
			CloudStorage: &entity.CloudStorage{Type: storageType},
		},
	}
}

func TestMergeLegacyReplicaConfig(t *testing.T) {
	currentMemoryQuota := uint64(128)
	currentCaseSensitive := false
	userCaseSensitive := true
	current := &serverless.ReplicaConfig{
		MemoryQuota:   &currentMemoryQuota,
		CaseSensitive: &currentCaseSensitive,
		Filter:        &serverless.FilterConfig{Rules: []string{"current.*"}},
	}
	userDefined := &serverless.ReplicaConfig{CaseSensitive: &userCaseSensitive}

	merged, err := mergeLegacyReplicaConfig(
		current,
		userDefined,
		`{"memory_quota":256,"case_sensitive":false}`,
	)
	require.NoError(t, err)
	require.Equal(t, uint64(256), *merged.MemoryQuota)
	require.True(t, *merged.CaseSensitive, "the user-defined value has the same precedence as dataflow-service-ng")
	require.Equal(t, []string{"current.*"}, merged.Filter.Rules)
	require.Equal(t, uint64(128), *current.MemoryQuota, "the kernel response must not be mutated")
}

func TestAssembleLegacyChangefeedConfigAlwaysReadsConfigCenter(t *testing.T) {
	currentMemoryQuota := uint64(128)
	userCaseSensitive := true
	protocol := "canal-json"
	kafkaVersion := "3.0.0"
	client := &changefeedConfigCenterStub{files: map[string]*config_center.ConfigurationFile{
		essentialChangefeedConfigFileName: {
			ConfigurationFilename: essentialChangefeedConfigFileName,
			Text:                  `{"memory_quota":256,"case_sensitive":false}`,
		},
		essentialChangefeedURIFileName: {
			ConfigurationFilename: essentialChangefeedURIFileName,
			Text:                  `{"max-message-bytes":1048576}`,
		},
	}}
	userConfig := &serverless.ChangefeedConfig{
		SinkURI: stringPointer("kafka://broker-1:9092/topic-a"),
		ReplicaConfig: &serverless.ReplicaConfig{
			CaseSensitive: &userCaseSensitive,
			Sink: &serverless.SinkConfig{
				Protocol: &protocol,
				KafkaConfig: &serverless.KafkaConfig{
					KafkaVersion: &kafkaVersion,
				},
			},
		},
	}
	currentConfig := &serverless.ReplicaConfig{
		MemoryQuota: &currentMemoryQuota,
		Filter:      &serverless.FilterConfig{Rules: []string{"current.*"}},
	}

	effective, err := assembleLegacyChangefeedConfig(
		context.Background(),
		client,
		&entity.CDCChangefeed{ID: "cf-1", SinkType: constant.ChangefeedTypeKafka},
		userConfig,
		currentConfig,
	)
	require.NoError(t, err)
	require.Equal(t, uint64(256), *effective.ReplicaConfig.MemoryQuota)
	require.True(t, *effective.ReplicaConfig.CaseSensitive)
	require.Equal(t, []string{"current.*"}, effective.ReplicaConfig.Filter.Rules)
	parsed, parseErr := url.Parse(*effective.SinkURI)
	require.NoError(t, parseErr)
	require.Equal(t, "1048576", parsed.Query().Get("max-message-bytes"))
}

func TestAssembleNewArchChangefeedConfigAlwaysReadsConfigCenter(t *testing.T) {
	protocol := "canal-json"
	kafkaVersion := "3.0.0"
	client := &changefeedConfigCenterStub{files: map[string]*config_center.ConfigurationFile{
		essentialChangefeedConfigFileName: {
			ConfigurationFilename: essentialChangefeedConfigFileName,
			Text:                  `{"memory_quota":256}`,
		},
		essentialChangefeedURIFileName: {
			ConfigurationFilename: essentialChangefeedURIFileName,
			Text:                  `{"max-message-bytes":1048576}`,
		},
	}}
	userConfig := &serverless.ChangefeedConfig{
		SinkURI: stringPointer("kafka://broker-1:9092/topic-a"),
		ReplicaConfig: &serverless.ReplicaConfig{Sink: &serverless.SinkConfig{
			Protocol: &protocol,
			KafkaConfig: &serverless.KafkaConfig{
				KafkaVersion: &kafkaVersion,
			},
		}},
	}
	currentConfig := &clusterproxy.ReplicaConfig{
		MemoryQuota: 128,
		Filter:      &clusterproxy.FilterConfig{Rules: []string{"current.*"}},
	}

	effective, err := assembleNewArchChangefeedConfig(
		context.Background(),
		client,
		&entity.CDCChangefeed{ID: "cf-1", SinkType: constant.ChangefeedTypeKafka},
		userConfig,
		currentConfig,
	)
	require.NoError(t, err)
	require.Equal(t, uint64(256), effective.ReplicaConfig.MemoryQuota)
	require.Equal(t, []string{"current.*"}, effective.ReplicaConfig.Filter.Rules)
	parsed, parseErr := url.Parse(effective.SinkURI)
	require.NoError(t, parseErr)
	require.Equal(t, "1048576", parsed.Query().Get("max-message-bytes"))
}

func TestMergeNewArchReplicaConfig(t *testing.T) {
	currentCaseSensitive := false
	userCaseSensitive := true
	current := &clusterproxy.ReplicaConfig{
		MemoryQuota:   128,
		CaseSensitive: &currentCaseSensitive,
		Filter:        &clusterproxy.FilterConfig{Rules: []string{"current.*"}},
	}
	userDefined := &clusterproxy.ReplicaConfig{CaseSensitive: &userCaseSensitive}

	merged, err := mergeNewArchReplicaConfig(
		current,
		userDefined,
		`{"memory_quota":256,"case_sensitive":false}`,
	)
	require.NoError(t, err)
	require.Equal(t, uint64(256), merged.MemoryQuota)
	require.True(t, *merged.CaseSensitive)
	require.Equal(t, []string{"current.*"}, merged.Filter.Rules)
}

func TestCloneReplicaConfigReturnsMarshalError(t *testing.T) {
	_, err := cloneReplicaConfig(&invalidReplicaConfig{Unsupported: make(chan int)})
	require.Error(t, err)
	require.Contains(t, err.Error(), "marshal current changefeed replica config")
}

func TestRunChangefeedConfigApplyWorkflow(t *testing.T) {
	tests := []struct {
		name        string
		pauseErr    bizerr.BizError
		updateErr   bizerr.BizError
		applyErr    bizerr.BizError
		resumeErr   bizerr.BizError
		expected    []string
		errContains string
	}{
		{
			name:     "running changefeed is paused updated applied and resumed",
			expected: []string{"pause", "update", "apply", "resume"},
		},
		{
			name:        "pause failure stops workflow",
			pauseErr:    bizerr.New(bizerr.ErrCodeUnknown, "pause failed"),
			expected:    []string{"pause"},
			errContains: "pause changefeed",
		},
		{
			name:        "config center failure still resumes",
			updateErr:   bizerr.New(bizerr.ErrCodeUnknown, "update failed"),
			expected:    []string{"pause", "update", "resume"},
			errContains: "update configuration",
		},
		{
			name:        "kernel failure still resumes",
			applyErr:    bizerr.New(bizerr.ErrCodeUnknown, "apply failed"),
			expected:    []string{"pause", "update", "apply", "resume"},
			errContains: "update changefeed config",
		},
		{
			name:        "resume failure is returned after successful apply",
			resumeErr:   bizerr.New(bizerr.ErrCodeUnknown, "resume failed"),
			expected:    []string{"pause", "update", "apply", "resume"},
			errContains: "resume changefeed",
		},
		{
			name:        "original failure takes precedence over resume failure",
			updateErr:   bizerr.New(bizerr.ErrCodeUnknown, "update failed"),
			resumeErr:   bizerr.New(bizerr.ErrCodeUnknown, "resume failed"),
			expected:    []string{"pause", "update", "resume"},
			errContains: "update configuration",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			events := make([]string, 0, 4)
			step := func(name string, stepErr bizerr.BizError) func() bizerr.BizError {
				return func() bizerr.BizError {
					events = append(events, name)
					return stepErr
				}
			}

			err := runChangefeedConfigApplyWorkflow(
				context.Background(),
				"cluster-1",
				"changefeed-1",
				step("pause", tt.pauseErr),
				step("update", tt.updateErr),
				step("apply", tt.applyErr),
				step("resume", tt.resumeErr),
			)
			require.Equal(t, tt.expected, events)
			if tt.errContains == "" {
				require.NoError(t, err)
				return
			}
			require.Error(t, err)
			require.Contains(t, err.Error(), tt.errContains)
			if tt.updateErr != nil && tt.resumeErr != nil {
				require.NotContains(t, err.Error(), "resume changefeed")
			}
		})
	}
}

func TestAssembleKafkaSinkURI(t *testing.T) {
	protocol := "canal-json"
	kafkaVersion := "3.0.0"
	enableTLS := true
	replicationFactor := int16(3)
	partitionNum := int32(6)
	userConfig := &serverless.ChangefeedConfig{
		SinkURI: stringPointer("kafka://broker-1:9092/topic-a"),
		ReplicaConfig: &serverless.ReplicaConfig{Sink: &serverless.SinkConfig{
			Protocol: &protocol,
			KafkaConfig: &serverless.KafkaConfig{
				EnableTLS:         &enableTLS,
				ReplicationFactor: &replicationFactor,
				PartitionNum:      &partitionNum,
				KafkaVersion:      &kafkaVersion,
			},
		}},
	}

	sinkURI, err := assembleChangefeedSinkURI(
		&entity.CDCChangefeed{SinkType: constant.ChangefeedTypeKafka},
		userConfig,
		`{"protocol":"open-protocol","max-message-bytes":1048576}`,
	)
	require.NoError(t, err)
	parsed, parseErr := url.Parse(sinkURI)
	require.NoError(t, parseErr)
	require.Equal(t, "broker-1:9092", parsed.Host)
	require.Equal(t, "/topic-a", parsed.Path)
	require.Equal(t, "canal-json", parsed.Query().Get("protocol"))
	require.Equal(t, "1048576", parsed.Query().Get("max-message-bytes"))
}

func TestAssembleKafkaSinkURIWithoutCompressionPreservesFollowingParameters(t *testing.T) {
	protocol := "canal-json"
	kafkaVersion := "3.0.0"
	saslUser := "test-user"
	kafkaClientID := "test-client"
	userConfig := &serverless.ChangefeedConfig{
		SinkURI: stringPointer("kafka://broker-1:9092/topic-a"),
		ReplicaConfig: &serverless.ReplicaConfig{Sink: &serverless.SinkConfig{
			Protocol: &protocol,
			KafkaConfig: &serverless.KafkaConfig{
				KafkaVersion:  &kafkaVersion,
				KafkaClientID: &kafkaClientID,
				SASLUser:      &saslUser,
			},
		}},
	}

	sinkURI, err := assembleChangefeedSinkURI(
		&entity.CDCChangefeed{SinkType: constant.ChangefeedTypeKafka},
		userConfig,
		`{}`,
	)
	require.NoError(t, err)
	parsed, parseErr := url.Parse(sinkURI)
	require.NoError(t, parseErr)
	require.Equal(t, "test-client", parsed.Query().Get("kafka-client-id"))
	require.Equal(t, "test-user", parsed.Query().Get("sasl-user"))
}

func TestAssembleMySQLSinkURI(t *testing.T) {
	workerCount := 8
	userConfig := &serverless.ChangefeedConfig{
		SinkURI: stringPointer("mysql://user:p%40ssword@mysql.example.com:4000"),
		ReplicaConfig: &serverless.ReplicaConfig{Sink: &serverless.SinkConfig{
			MySQLConfig: &serverless.MySQLConfig{WorkerCount: &workerCount},
		}},
	}

	sinkURI, err := assembleChangefeedSinkURI(
		&entity.CDCChangefeed{SinkType: constant.ChangefeedTypeMySQL},
		userConfig,
		`{"worker-count":16,"max-txn-row":512}`,
	)
	require.NoError(t, err)
	parsed, parseErr := url.Parse(sinkURI)
	require.NoError(t, parseErr)
	require.Equal(t, "8", parsed.Query().Get("worker-count"))
	require.Equal(t, "512", parsed.Query().Get("max-txn-row"))
	require.Contains(t, sinkURI, "p%40ssword@", "password must stay URL-escaped")
}

func TestAssembleMySQLSinkURIEscapesUsername(t *testing.T) {
	userConfig := &serverless.ChangefeedConfig{
		SinkURI: stringPointer("mysql://user%40tenant:password@mysql.example.com:4000"),
		ReplicaConfig: &serverless.ReplicaConfig{Sink: &serverless.SinkConfig{
			MySQLConfig: &serverless.MySQLConfig{},
		}},
	}

	sinkURI, err := assembleChangefeedSinkURI(
		&entity.CDCChangefeed{SinkType: constant.ChangefeedTypeMySQL},
		userConfig,
		`{}`,
	)
	require.NoError(t, err)
	require.Contains(t, sinkURI, "user%40tenant:password@")
}

func TestAssembleCloudStorageSinkURIRejectsMalformedBoolean(t *testing.T) {
	userConfig := &serverless.ChangefeedConfig{
		SinkURI:       stringPointer("s3://bucket/path?force-path-style=notabool"),
		ReplicaConfig: &serverless.ReplicaConfig{},
	}

	_, err := assembleChangefeedSinkURI(
		cloudStorageChangefeed(constant.CloudStorageTypeS3),
		userConfig,
		`{}`,
	)
	require.Error(t, err)
	require.Contains(t, err.Error(), "invalid boolean sink uri parameter force-path-style")
}

func TestAssembleGCSSinkURIPreservesCredentials(t *testing.T) {
	credentials := `{"type":"service_account","private_key":"test-key&value"}`
	userConfig := &serverless.ChangefeedConfig{
		SinkURI: stringPointer("gcs://test-bucket/path?credentials=" + url.QueryEscape(credentials)),
		ReplicaConfig: &serverless.ReplicaConfig{Sink: &serverless.SinkConfig{
			CloudStorageConfig: &serverless.CloudStorageConfig{},
		}},
	}

	sinkURI, err := assembleChangefeedSinkURI(
		cloudStorageChangefeed(constant.CloudStorageTypeGCS),
		userConfig,
		`{}`,
	)
	require.NoError(t, err)
	parsed, parseErr := url.Parse(sinkURI)
	require.NoError(t, parseErr)
	require.Equal(t, credentials, parsed.Query().Get("credentials"))
}

func TestAssembleSinkURIRejectsUnknownField(t *testing.T) {
	userConfig := &serverless.ChangefeedConfig{
		SinkURI: stringPointer("kafka://broker-1:9092/topic-a"),
		ReplicaConfig: &serverless.ReplicaConfig{Sink: &serverless.SinkConfig{
			KafkaConfig: &serverless.KafkaConfig{},
		}},
	}
	_, err := assembleChangefeedSinkURI(
		&entity.CDCChangefeed{SinkType: constant.ChangefeedTypeKafka},
		userConfig,
		`{"unknown-option":true}`,
	)
	require.Error(t, err)
	require.Contains(t, err.Error(), "unknown field")
}