package service

import (
	"context"
	"errors"
	"strings"
	"testing"
	"time"

	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/pkgv2/bizerr"
	"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/domain/facade/mock_facade"
	"go.uber.org/mock/gomock"
)

type migrateTestEncryptor struct{}

func (m migrateTestEncryptor) EncryptBase64(ext string) (string, error) {
	return "enc:" + ext, nil
}

func (m migrateTestEncryptor) DecryptBase64(encryptedEncoded string) (string, error) {
	if !strings.HasPrefix(encryptedEncoded, "enc:") {
		return "", errors.New("unexpected ciphertext")
	}
	return strings.TrimPrefix(encryptedEncoded, "enc:"), nil
}

func TestBuildMigratedChangefeedVoDecryptsMySQLSink(t *testing.T) {
	svc := &cdcService{encryptor: migrateTestEncryptor{}}
	source := &entity.CDCChangefeed{
		ClusterID: "cluster-1",
		Arch:      constant.ChangefeedArchReplicationWorker,
		RCU:       3,
		Name:      "source",
		SinkType:  constant.ChangefeedTypeMySQL,
		MySQL: &entity.MySQLInfo{
			Network: entity.MySQLNetwork{
				NetworkType:    constant.MySQLNetworkTypePublic,
				PublicEndpoint: "127.0.0.1:3306",
			},
			Authentication: entity.MySQLAuthentication{
				User:     "root",
				Password: "enc:secret",
			},
		},
		Filter: &entity.CDCFilter{
			FilterRule: []string{"*.*"},
			Mode:       constant.TableModeIgnoreNotSupportTable,
		},
	}

	got, err := svc.buildMigratedChangefeedVo(context.Background(), source, 4, "target", 123)
	require.Nil(t, err)
	require.Equal(t, constant.ChangefeedArchTiCDCNewArch, got.Arch)
	require.Equal(t, uint32(4), got.RCU)
	require.Equal(t, "target", got.Name)
	require.Equal(t, constant.StartModeFromTSO, got.StartPosition.Mode)
	require.Equal(t, uint64(123), got.StartPosition.TSO)
	require.Equal(t, "secret", got.Sink.MySQL.Authentication.Password)
	require.Equal(t, "enc:secret", source.MySQL.Authentication.Password)
	require.NotSame(t, source.MySQL, got.Sink.MySQL)
	require.NotSame(t, source.Filter, got.Filter)
}

func TestBuildMigratedChangefeedVoDefaultsToSourceName(t *testing.T) {
	svc := &cdcService{encryptor: migrateTestEncryptor{}}
	source := &entity.CDCChangefeed{
		ClusterID: "cluster-1",
		Name:      "source-name",
		SinkType:  constant.ChangefeedTypeMySQL,
		MySQL:     &entity.MySQLInfo{},
	}

	got, err := svc.buildMigratedChangefeedVo(context.Background(), source, 3, "", 123)
	require.Nil(t, err)
	require.Equal(t, "source-name", got.Name)
}

func TestWaitForLegacyChangefeedStopped(t *testing.T) {
	t.Run("waits for stopped state", func(t *testing.T) {
		ctrl := gomock.NewController(t)
		proxy := mock_facade.NewMockServerlessClusterProxyFacade(ctrl)
		source := &entity.CDCChangefeed{KeySpaceID: 42, ChangefeedID: "cf-old"}
		checkpoint := uint64(123)

		gomock.InOrder(
			proxy.EXPECT().
				GetServerlessChangefeed(gomock.Any(), uint32(42), "cf-old").
				Return(&serverless.Changefeed{State: serverless.StateNormal}, nil),
			proxy.EXPECT().
				GetServerlessChangefeed(gomock.Any(), uint32(42), "cf-old").
				Return(&serverless.Changefeed{State: serverless.StateStopped, CheckpointTS: &checkpoint}, nil),
		)

		got, err := waitForLegacyChangefeedStopped(context.Background(), proxy, source, time.Millisecond)
		require.Nil(t, err)
		require.Equal(t, checkpoint, got)
	})

	t.Run("returns immediately when already stopped", func(t *testing.T) {
		ctrl := gomock.NewController(t)
		proxy := mock_facade.NewMockServerlessClusterProxyFacade(ctrl)
		source := &entity.CDCChangefeed{KeySpaceID: 42, ChangefeedID: "cf-old"}
		checkpoint := uint64(456)

		proxy.EXPECT().
			GetServerlessChangefeed(gomock.Any(), uint32(42), "cf-old").
			Return(&serverless.Changefeed{State: serverless.StateStopped, CheckpointTS: &checkpoint}, nil)

		got, err := waitForLegacyChangefeedStopped(context.Background(), proxy, source, time.Hour)
		require.Nil(t, err)
		require.Equal(t, checkpoint, got)
	})

	t.Run("returns when context ends", func(t *testing.T) {
		ctrl := gomock.NewController(t)
		proxy := mock_facade.NewMockServerlessClusterProxyFacade(ctrl)
		source := &entity.CDCChangefeed{KeySpaceID: 42, ChangefeedID: "cf-old"}
		ctx, cancel := context.WithCancel(context.Background())

		proxy.EXPECT().
			GetServerlessChangefeed(gomock.Any(), uint32(42), "cf-old").
			DoAndReturn(func(context.Context, uint32, string) (*serverless.Changefeed, bizerr.BizError) {
				cancel()
				return &serverless.Changefeed{State: serverless.StateNormal}, nil
			})

		got, err := waitForLegacyChangefeedStopped(ctx, proxy, source, time.Hour)
		require.Zero(t, got)
		require.NotNil(t, err)
		require.Contains(t, err.Error(), "last state: normal")
		require.Contains(t, err.Error(), context.Canceled.Error())
	})

	t.Run("returns kernel lookup error", func(t *testing.T) {
		ctrl := gomock.NewController(t)
		proxy := mock_facade.NewMockServerlessClusterProxyFacade(ctrl)
		source := &entity.CDCChangefeed{KeySpaceID: 42, ChangefeedID: "cf-old"}
		lookupErr := bizerr.New(bizerr.ErrCodeUnknown, "lookup failed")

		proxy.EXPECT().
			GetServerlessChangefeed(gomock.Any(), uint32(42), "cf-old").
			Return(nil, lookupErr)

		got, err := waitForLegacyChangefeedStopped(context.Background(), proxy, source, time.Hour)
		require.Zero(t, got)
		require.Equal(t, lookupErr, err)
	})
}

func TestCanonicalSinkURI(t *testing.T) {
	left, leftSafe, err := canonicalSinkURI("kafka://user:password@broker:9092/topic?b=2&a=1")
	require.NoError(t, err)
	right, rightSafe, err := canonicalSinkURI("kafka://user:password@broker:9092/topic?a=1&b=2")
	require.NoError(t, err)
	require.Equal(t, left, right)
	require.NotContains(t, leftSafe, "password")
	require.NotContains(t, rightSafe, "password")
	require.Contains(t, leftSafe, "broker:9092/topic")
}

func TestCanonicalSinkURI_RedactsSensitiveQuery(t *testing.T) {
	_, safe, err := canonicalSinkURI("s3://bucket/prefix?access_key=abc&secret-access-key=def&token=ghi&format=csv")
	require.NoError(t, err)
	require.NotContains(t, safe, "abc")
	require.NotContains(t, safe, "def")
	require.NotContains(t, safe, "ghi")
	require.Contains(t, safe, "format=csv")
}

func TestCanonicalKernelConfig_IgnoresRuntimeTuning(t *testing.T) {
	left, err := canonicalKernelConfig(map[string]any{
		"memory_quota": float64(1024),
		"filter":       map[string]any{"rules": []string{"db.*"}},
	}, "mysql")
	require.NoError(t, err)
	right, err := canonicalKernelConfig(map[string]any{
		"memory_quota": float64(2048),
		"filter":       map[string]any{"rules": []string{"db.*"}},
	}, "mysql")
	require.NoError(t, err)
	require.Equal(t, left, right)
}

func TestCanonicalKernelConfig_MySQLIgnoresUnrelatedExpandedDefaults(t *testing.T) {
	legacy, err := canonicalKernelConfig(map[string]any{
		"filter": map[string]any{"rules": []string{"*.*"}},
		"sink": map[string]any{
			"mysql_config":    map[string]any{},
			"advance_timeout": float64(150),
		},
	}, "mysql")
	require.NoError(t, err)

	newarch, err := canonicalKernelConfig(map[string]any{
		"filter": map[string]any{"rules": []string{"*.*"}},
		"sink": map[string]any{
			"csv":                        map[string]any{"delimiter": ","},
			"encoder_concurrency":        float64(32),
			"enable_partition_separator": true,
			"mysql_config":               map[string]any{},
			"advance_timeout":            float64(150),
		},
	}, "mysql")
	require.NoError(t, err)
	require.Equal(t, legacy, newarch)
}

func TestCanonicalKernelConfig_MySQLStillComparesBehavior(t *testing.T) {
	left, err := canonicalKernelConfig(map[string]any{
		"sink": map[string]any{"mysql_config": map[string]any{"enable_batch_dml": true}},
	}, "mysql")
	require.NoError(t, err)
	right, err := canonicalKernelConfig(map[string]any{
		"sink": map[string]any{"mysql_config": map[string]any{"enable_batch_dml": false}},
	}, "mysql")
	require.NoError(t, err)
	require.NotEqual(t, left, right)
}

func TestCanonicalKernelConfig_S3IgnoresExpandedDefaults(t *testing.T) {
	legacy, err := canonicalKernelConfig(map[string]any{
		"case_sensitive": true,
		"filter":         map[string]any{"rules": []string{"test.*", "!test.excluded"}},
		"sink": map[string]any{
			"protocol": "csv",
			"csv": map[string]any{
				"delimiter": ",", "quote": "\"", "null": "\\N",
				"include_commit_ts": true, "binary_encoding_method": "base64",
				"output_old_value": false, "output_handle_key": false,
			},
			"content_compatible":                    nil,
			"delete_only_output_handle_key_columns": nil,
			"cloud_storage_config":                  map[string]any{"file_size": float64(67108864)},
		},
	}, "s3")
	require.NoError(t, err)

	newarch, err := canonicalKernelConfig(map[string]any{
		"case_sensitive":                    true,
		"enable_sync_point":                 false,
		"enable_table_monitor":              false,
		"bdr_mode":                          false,
		"enable_active_active":              false,
		"active_active_progress_interval":   float64(30 * time.Minute),
		"active_active_sync_stats_interval": float64(time.Minute),
		"sync_point_interval":               float64(10 * time.Minute),
		"sync_point_retention":              float64(24 * time.Hour),
		"filter":                            map[string]any{"rules": []string{"test.*", "!test.excluded"}},
		"sink": map[string]any{
			"protocol": "csv",
			"csv": map[string]any{
				"delimiter": ",", "quote": "\"", "null": "\\N",
				"include_commit_ts": true, "binary_encoding_method": "base64",
				"output_old_value": false, "output_handle_key": false,
				"output_field_header": false,
			},
			"encoder_concurrency":                   float64(32),
			"only_output_updated_columns":           false,
			"content_compatible":                    false,
			"delete_only_output_handle_key_columns": false,
			"cloud_storage_config": map[string]any{
				"file_size": float64(67108864), "output_column_id": false,
			},
		},
	}, "s3")
	require.NoError(t, err)
	require.Equal(t, legacy, newarch)
}

func TestCanonicalKernelConfig_NormalizesDurationRepresentations(t *testing.T) {
	legacy, err := canonicalKernelConfig(map[string]any{
		"sync_point_interval":  float64(10 * time.Minute),
		"sync_point_retention": float64(24 * time.Hour),
	}, "mysql")
	require.NoError(t, err)
	newarch, err := canonicalKernelConfig(map[string]any{
		"sync_point_interval":  "10m0s",
		"sync_point_retention": "24h0m0s",
	}, "mysql")
	require.NoError(t, err)
	require.Equal(t, legacy, newarch)
}

func TestCanonicalKernelConfig_NormalizesCrossArchDefaults(t *testing.T) {
	legacy, err := canonicalKernelConfig(map[string]any{
		"filter": map[string]any{"rules": []string{"test.*"}},
		"sink":   map[string]any{"mysql_config": map[string]any{}},
	}, "mysql")
	require.NoError(t, err)

	newarch, err := canonicalKernelConfig(map[string]any{
		"filter":                            map[string]any{"rules": []string{"test.*"}},
		"enable_sync_point":                 false,
		"enable_table_monitor":              false,
		"bdr_mode":                          false,
		"enable_active_active":              false,
		"active_active_progress_interval":   float64(30 * time.Minute),
		"active_active_sync_stats_interval": float64(time.Minute),
		"sink": map[string]any{
			"mysql_config":    map[string]any{},
			"advance_timeout": float64(150),
		},
	}, "mysql")
	require.NoError(t, err)
	require.Equal(t, legacy, newarch)
}

func TestCanonicalConfigDiffPrintsPathsAndRedactsSecrets(t *testing.T) {
	diff := canonicalConfigDiff(
		map[string]any{"sink": map[string]any{
			"mysql_config": map[string]any{"password": "source-secret"},
		}},
		map[string]any{"sink": map[string]any{
			"mysql_config": map[string]any{"password": "target-secret"},
			"safe_mode":    true,
		}},
	)
	require.Contains(t, diff, "sink.mysql_config.password")
	require.Contains(t, diff, "<redacted>")
	require.NotContains(t, diff, "source-secret")
	require.NotContains(t, diff, "target-secret")
	require.Contains(t, diff, "sink.safe_mode")
}