package impl

import (
	"testing"
	"time"

	"github.com/AlekSi/pointer"
	"github.com/stretchr/testify/assert"
	"github.com/tidbcloud/dataflow-service-ng/app/utils"
	dataflow "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	"github.com/tidbcloud/pkgv2/clusterproxy"
)

func TestParseStartTSO(t *testing.T) {
	var startTSO uint64 = 123123
	utcTime := time.Now()
	invalidUTCTimeStr := utcTime.Format(time.RFC1123)
	validUTCTimeStr := utcTime.Format(time.RFC3339)
	utcTimeTSO := utils.GoTimeToTSO(utcTime.Truncate(time.Second))

	testcases := []struct {
		name           string
		startPosition  *dataflow.StartPosition
		expectedError  bool
		expectedResult uint64
	}{
		{
			name:           "nil start position",
			startPosition:  nil,
			expectedError:  false,
			expectedResult: 0,
		},
		{
			name: "unsupported mode",
			startPosition: &dataflow.StartPosition{
				Mode: dataflow.StartMode(11),
			},
			expectedError:  true,
			expectedResult: 0,
		},
		{
			name: "from_now mode",
			startPosition: &dataflow.StartPosition{
				Mode: dataflow.StartMode_FROM_NOW,
			},
			expectedError:  false,
			expectedResult: 0,
		},
		{
			name: "tso mode",
			startPosition: &dataflow.StartPosition{
				Mode: dataflow.StartMode_FROM_TSO,
				Tso:  startTSO,
			},
			expectedError:  false,
			expectedResult: startTSO,
		},
		{
			name: "utc mode with invalid format",
			startPosition: &dataflow.StartPosition{
				Mode: dataflow.StartMode_FROM_UTC,
				Utc:  invalidUTCTimeStr,
			},
			expectedError:  true,
			expectedResult: 0,
		},
		{
			name: "utc mode with valid format",
			startPosition: &dataflow.StartPosition{
				Mode: dataflow.StartMode_FROM_UTC,
				Utc:  validUTCTimeStr,
			},
			expectedError:  false,
			expectedResult: utcTimeTSO,
		},
	}

	for _, tc := range testcases {
		t.Run(tc.name, func(t *testing.T) {
			tso, err := parseStartTSO(tc.startPosition)
			if tc.expectedError {
				assert.Error(t, err)
			} else {
				assert.NoError(t, err)
				assert.Equal(t, tc.expectedResult, tso)
			}
		})
	}
}

func TestConvertEventFilterRules(t *testing.T) {
	testcases := []struct {
		name             string
		rules            []*dataflow.EventFilterRule
		expectedRulesNum int
	}{
		{
			name:             "nil rule",
			rules:            nil,
			expectedRulesNum: 0,
		},
		{
			name:             "empty rule",
			rules:            []*dataflow.EventFilterRule{},
			expectedRulesNum: 0,
		},
		{
			name: "one rule",
			rules: []*dataflow.EventFilterRule{
				{
					Matcher:               []string{"a"},
					IgnoreDeleteValueExpr: "b",
				},
			},
			expectedRulesNum: 1,
		},
		{
			name: "multiple rules",
			rules: []*dataflow.EventFilterRule{
				{
					Matcher:               []string{"a"},
					IgnoreDeleteValueExpr: "b",
				},
				{
					Matcher:               []string{"c"},
					IgnoreDeleteValueExpr: "d",
				},
				{
					Matcher:               []string{"e"},
					IgnoreDeleteValueExpr: "f",
				},
			},
			expectedRulesNum: 3,
		},
	}

	for _, tc := range testcases {
		t.Run(tc.name, func(t *testing.T) {
			rules := convertEventFilterRules(tc.rules)
			assert.Equal(t, tc.expectedRulesNum, len(rules))
		})
	}
}

func TestAssembleChangefeedSinkConfigTableConfig(t *testing.T) {
	tableConfig := &dataflow.TableConfig{
		FilterRule: []string{"*.*"},
		Mode:       dataflow.TableMode_FORCE_SYNC,
		EventFilterRule: []*dataflow.EventFilterRule{
			{
				Matcher: []string{"test.*"},
			},
		},
		BdrMode:          pointer.ToBool(true),
		IgnoreTxnStartTs: []uint64{123, 456},
	}
	replicaConfig := &clusterproxy.ReplicaConfig{}
	assembleChangefeedSinkConfigTableConfig(tableConfig, replicaConfig)
	assert.NotNil(t, replicaConfig.Filter)
	assert.Equal(t, tableConfig.FilterRule, replicaConfig.Filter.Rules)
	assert.Equal(t, tableConfig.IgnoreTxnStartTs, replicaConfig.Filter.IgnoreTxnStartTs)
	assert.Equal(t, tableConfig.EventFilterRule[0].Matcher, replicaConfig.Filter.EventFilters[0].Matcher)
	assert.NotNil(t, replicaConfig.ForceReplicate)
	assert.True(t, *replicaConfig.ForceReplicate)
	assert.NotNil(t, replicaConfig.BDRMode)
	assert.True(t, *replicaConfig.BDRMode)

	tableConfig.Mode = dataflow.TableMode_IGNORE_NOT_SUPPORT_TABLE
	tableConfig.BdrMode = pointer.ToBool(false)
	replicaConfig = &clusterproxy.ReplicaConfig{}
	assembleChangefeedSinkConfigTableConfig(tableConfig, replicaConfig)
	assert.NotNil(t, replicaConfig.Filter)
	assert.Equal(t, tableConfig.FilterRule, replicaConfig.Filter.Rules)
	assert.Equal(t, tableConfig.IgnoreTxnStartTs, replicaConfig.Filter.IgnoreTxnStartTs)
	assert.Equal(t, tableConfig.EventFilterRule[0].Matcher, replicaConfig.Filter.EventFilters[0].Matcher)
	assert.NotNil(t, replicaConfig.IgnoreIneligibleTable)
	assert.True(t, *replicaConfig.IgnoreIneligibleTable)
	assert.NotNil(t, replicaConfig.BDRMode)
	assert.False(t, *replicaConfig.BDRMode)
}