package entity

import (
	"context"
	"encoding/json"
	"testing"

	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/pkgv2/clusterproxy/serverless"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
)

// TestCloudStorageSinkConfig_OutputColumnID verifies that the OutputColumnID
// option is propagated to the cloud-storage sink config, and defaults to false.
func TestCloudStorageSinkConfig_OutputColumnID(t *testing.T) {
	newChangefeed := func(outputColumnID bool) *CDCChangefeed {
		return &CDCChangefeed{
			CloudStorageInfo: &CloudStorageInfo{
				CloudStorage:      &CloudStorage{},
				IntervalInSeconds: 5,
				SizeInMiB:         64,
				OutputColumnID:    outputColumnID,
				DateSeparator:     constant.DateSeparatorDay,
				DataFormat: CloudStorageDataFormat{
					Protocol: constant.CloudStorageProtocolCSV,
					CSVConfig: &CloudStorageCSVConfig{
						IncludeCommitTs: true,
						Encoding:        constant.CloudStorageBinaryEncodingHex,
					},
				},
			},
		}
	}

	t.Run("true is propagated", func(t *testing.T) {
		sink := &serverless.SinkConfig{}
		err := newChangefeed(true).CloudStorageSinkConfig(sink, nil)
		require.Nil(t, err)
		require.NotNil(t, sink.CloudStorageConfig.OutputColumnID)
		require.True(t, *sink.CloudStorageConfig.OutputColumnID)
	})

	t.Run("false by default", func(t *testing.T) {
		sink := &serverless.SinkConfig{}
		err := newChangefeed(false).CloudStorageSinkConfig(sink, nil)
		require.Nil(t, err)
		require.NotNil(t, sink.CloudStorageConfig.OutputColumnID)
		require.False(t, *sink.CloudStorageConfig.OutputColumnID)
	})
}

func TestValidateColumnSelectors(t *testing.T) {
	tests := []struct {
		name      string
		selectors []ColumnSelector
	}{
		{name: "empty matcher", selectors: []ColumnSelector{{Columns: []string{"id"}}}},
		{name: "empty columns", selectors: []ColumnSelector{{Matcher: []string{"app.orders"}}}},
		{
			name: "invalid column rule",
			selectors: []ColumnSelector{{
				Matcher: []string{"app.orders"},
				Columns: []string{"["},
			}},
		},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			require.Error(t, validateColumnSelectors(context.Background(), tt.selectors))
		})
	}

	require.NoError(t, validateColumnSelectors(context.Background(), []ColumnSelector{{
		Matcher: []string{"app.orders"},
		Columns: []string{"*", "!secret"},
	}}))
}

func TestValidateCloudStorageColumnSelectorsArch(t *testing.T) {
	storage := &CloudStorageInfo{ColumnSelectors: []ColumnSelector{{
		Matcher: []string{"app.orders"}, Columns: []string{"*"},
	}}}
	require.Error(t, ValidateCloudStorageColumnSelectorsArch(constant.ChangefeedArchReplicationWorker, storage))
	require.NoError(t, ValidateCloudStorageColumnSelectorsArch(constant.ChangefeedArchTiCDCNewArch, storage))
	require.NoError(
		t,
		ValidateCloudStorageColumnSelectorsArch(constant.ChangefeedArchReplicationWorker, &CloudStorageInfo{}),
	)
}

func TestCloudStorageSinkConfig_ColumnSelectors(t *testing.T) {
	changefeed := &CDCChangefeed{CloudStorageInfo: &CloudStorageInfo{
		CloudStorage:      &CloudStorage{},
		IntervalInSeconds: 5,
		SizeInMiB:         64,
		DateSeparator:     constant.DateSeparatorDay,
		DataFormat: CloudStorageDataFormat{
			Protocol: constant.CloudStorageProtocolCanalJSON,
		},
		ColumnSelectors: []ColumnSelector{{
			Matcher: []string{"app.orders"},
			Columns: []string{"id", "status"},
		}},
	}}

	sink := &serverless.SinkConfig{}
	require.NoError(t, changefeed.CloudStorageSinkConfig(sink, nil))
	require.Len(t, sink.ColumnSelectors, 1)
	require.Equal(t, []string{"app.orders"}, sink.ColumnSelectors[0].Matcher)
	require.Equal(t, []string{"id", "status"}, sink.ColumnSelectors[0].Columns)
}

func TestCloudStorageSinkConfig_CSVDefaults(t *testing.T) {
	newChangefeed := func(csvConfig *CloudStorageCSVConfig) *CDCChangefeed {
		return &CDCChangefeed{
			CloudStorageInfo: &CloudStorageInfo{
				CloudStorage:      &CloudStorage{},
				IntervalInSeconds: 5,
				SizeInMiB:         64,
				DateSeparator:     constant.DateSeparatorDay,
				DataFormat: CloudStorageDataFormat{
					Protocol:  constant.CloudStorageProtocolCSV,
					CSVConfig: csvConfig,
				},
			},
		}
	}

	t.Run("omitted values are materialized in the downstream request", func(t *testing.T) {
		sink := &serverless.SinkConfig{}
		err := newChangefeed(&CloudStorageCSVConfig{}).CloudStorageSinkConfig(sink, nil)
		require.Nil(t, err)
		require.NotNil(t, sink.CSV)
		require.Equal(t, ",", *sink.CSV.Delimiter)
		require.Equal(t, "\"", *sink.CSV.Quote)
		require.Equal(t, "\\N", *sink.CSV.Null)
		require.Equal(t, "base64", *sink.CSV.BinaryEncodingMethod)

		body, marshalErr := json.Marshal(sink)
		require.NoError(t, marshalErr)
		require.JSONEq(t, `{
			"protocol": "csv",
			"csv": {
				"delimiter": ",",
				"quote": "\"",
				"null": "\\N",
				"include_commit_ts": false,
				"binary_encoding_method": "base64"
			},
			"date_separator": "day",
			"cloud_storage_config": {
				"flush_interval": "5s",
				"file_size": 67108864,
				"output_column_id": false,
				"output_raw_change_event": false
			}
		}`, string(body))
	})

	t.Run("explicit values are preserved", func(t *testing.T) {
		delimiter := "|"
		quote := "'"
		nullValue := "NULL"
		sink := &serverless.SinkConfig{}
		err := newChangefeed(&CloudStorageCSVConfig{
			Delimiter: &delimiter,
			Quote:     &quote,
			NullValue: &nullValue,
			Encoding:  constant.CloudStorageBinaryEncodingHex,
		}).CloudStorageSinkConfig(sink, nil)
		require.Nil(t, err)
		require.Equal(t, delimiter, *sink.CSV.Delimiter)
		require.Equal(t, quote, *sink.CSV.Quote)
		require.Equal(t, nullValue, *sink.CSV.Null)
		require.Equal(t, "hex", *sink.CSV.BinaryEncodingMethod)
	})
}