package error_wrapper

import (
	"testing"
)

func TestClassifyChangefeedError(t *testing.T) {
	tests := []struct {
		name     string
		errMsg   string
		expected CommonChangefeedErr
	}{
		{
			name:     "ErrKafkaInvalidPartitionNum in message",
			errMsg:   "CDC:ErrKafkaInvalidPartitionNum partition count 0 is invalid",
			expected: ErrChangefeedKafkaInvalidPartitionNum,
		},
		{
			name:     "ErrKafkaInvalidPartitionNum wins over ErrKafkaNewProducer when both present",
			errMsg:   "CDC:ErrKafkaNewProducer failed to create producer: CDC:ErrKafkaInvalidPartitionNum partition count 0",
			expected: ErrChangefeedKafkaInvalidPartitionNum,
		},
		{
			name:     "ErrKafkaNewProducer alone (no inner code)",
			errMsg:   "CDC:ErrKafkaNewProducer failed to create Kafka producer for unknown reason",
			expected: ErrChangefeedKafkaNewProducer,
		},
		{
			name:     "ErrKafkaInvalidConfig",
			errMsg:   "CDC:ErrKafkaInvalidConfig broker list is empty",
			expected: ErrChangefeedKafkaInvalidConfig,
		},
		{
			name:     "ErrKafkaTopicExprInvalid",
			errMsg:   "CDC:ErrKafkaTopicExprInvalid expression syntax error",
			expected: ErrChangefeedKafkaTopicExprInvalid,
		},
		{
			name:     "ErrTableIneligible",
			errMsg:   "CDC:ErrTableIneligible table does not have a primary key",
			expected: ErrChangefeedTableIneligible,
		},
		{
			name:     "ErrAPIInvalidParam",
			errMsg:   "CDC:ErrAPIInvalidParam sink_uri is required",
			expected: ErrChangefeedAPIInvalidParam,
		},
		{
			name:     "ErrStartTsBeforeGC",
			errMsg:   "CDC:ErrStartTsBeforeGC start-ts 1234 is before GC safe point 9999",
			expected: ErrChangefeedStartTsBeforeGC,
		},
		{
			name:     "unknown system error returns NotCommonErr",
			errMsg:   "CDC:ErrUnknownSomethingElse some random system failure",
			expected: ErrChangefeedNotCommonErr,
		},
		{
			name:     "empty string returns NotCommonErr",
			errMsg:   "",
			expected: ErrChangefeedNotCommonErr,
		},
	}

	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			got := ClassifyChangefeedError(tc.errMsg)
			if got != tc.expected {
				t.Errorf("ClassifyChangefeedError(%q) = %q, want %q", tc.errMsg, got, tc.expected)
			}
		})
	}
}

func TestChangefeedErrType(t *testing.T) {
	tests := []struct {
		name     string
		errMsg   string
		expected string
	}{
		{
			name:     "ErrKafkaInvalidPartitionNum returns specific label",
			errMsg:   "CDC:ErrKafkaInvalidPartitionNum partition count 0 is invalid",
			expected: "KafkaInvalidPartitionNum",
		},
		{
			name:     "ErrKafkaInvalidPartitionNum wins over ErrKafkaNewProducer",
			errMsg:   "CDC:ErrKafkaNewProducer failed: CDC:ErrKafkaInvalidPartitionNum partition count 0",
			expected: "KafkaInvalidPartitionNum",
		},
		{
			name:     "ErrKafkaNewProducer alone returns KafkaNewProducer",
			errMsg:   "CDC:ErrKafkaNewProducer some other inner error",
			expected: "KafkaNewProducer",
		},
		{
			name:     "unknown error returns SystemError",
			errMsg:   "connection refused",
			expected: "SystemError",
		},
		{
			name:     "empty string returns SystemError",
			errMsg:   "",
			expected: "SystemError",
		},
	}

	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			got := ChangefeedErrType(tc.errMsg)
			if got != tc.expected {
				t.Errorf("ChangefeedErrType(%q) = %q, want %q", tc.errMsg, got, tc.expected)
			}
		})
	}
}