package console

import (
	"context"
	"testing"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/app/internal/feature_gate"
	feature_gate_impl "github.com/tidbcloud/dataflow-service/app/internal/feature_gate/impl"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	"github.com/tidbcloud/dataflow-service/internal/config"
	bizerr_wrapper "github.com/tidbcloud/dataflow-service/pkg/bizerr-wrapper"
	client_mock "github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli/mock"
)

func TestIsMultiTiCDCClustersEnabled(t *testing.T) {
	originFeatureList := config.Global().ServerConfig.FeaturesSwitchers.FeaturesAllowList
	originFeatureGate := config.Global().ChangefeedConfig.MultiTiCDCClustersFeatureGate
	defer func() {
		config.Global().ServerConfig.FeaturesSwitchers.FeaturesAllowList = originFeatureList
		config.Global().ChangefeedConfig.MultiTiCDCClustersFeatureGate = originFeatureGate
	}()

	config.Global().ServerConfig.FeaturesSwitchers.FeaturesAllowList = map[string]config.AllowListMatcher{
		feature_gate.MultiTiCDCClusters: {
			TenantID:  []string{"1"},
			ProjectID: []string{"11"},
			ClusterID: []string{"111"},
		},
	}
	config.Global().ChangefeedConfig.MultiTiCDCClustersFeatureGate = config.MultiTiCDCClustersFeatureGate{
		EnabledCloudProviders:  []string{"aws"},
		EnabledDownstreamTypes: []int32{int32(dataflow.ConnectorType_MYSQL)},
	}
	featureGate := feature_gate_impl.NewFeatureGate()

	testcases := []struct {
		name          string
		cluster       *core.Cluster
		connectorType dataflow.ConnectorType
		result        bool
	}{
		{
			name: "not int allow list",
			cluster: &core.Cluster{
				TenantId:          2,
				ProjectId:         22,
				Id:                222,
				CloudProviderName: "aws",
			},
			connectorType: dataflow.ConnectorType_MYSQL,
			result:        false,
		},
		{
			name: "cloud provider is not enabled",
			cluster: &core.Cluster{
				TenantId:          1,
				ProjectId:         11,
				Id:                111,
				CloudProviderName: "gcp",
			},
			connectorType: dataflow.ConnectorType_MYSQL,
			result:        false,
		},
		{
			name: "connector type is not enabled",
			cluster: &core.Cluster{
				TenantId:          1,
				ProjectId:         11,
				Id:                111,
				CloudProviderName: "aws",
			},
			connectorType: dataflow.ConnectorType_KAFKA,
			result:        false,
		},
		{
			name: "tenant id is in allow list",
			cluster: &core.Cluster{
				TenantId:          1,
				ProjectId:         22,
				Id:                222,
				CloudProviderName: "aws",
			},
			connectorType: dataflow.ConnectorType_MYSQL,
			result:        true,
		},
		{
			name: "project id is in allow list",
			cluster: &core.Cluster{
				TenantId:          2,
				ProjectId:         11,
				Id:                222,
				CloudProviderName: "aws",
			},
			connectorType: dataflow.ConnectorType_MYSQL,
			result:        true,
		},
		{
			name: "cluster id is in allow list",
			cluster: &core.Cluster{
				TenantId:          2,
				ProjectId:         22,
				Id:                111,
				CloudProviderName: "aws",
			},
			connectorType: dataflow.ConnectorType_MYSQL,
			result:        true,
		},
		{
			name: "all ids are in allow list",
			cluster: &core.Cluster{
				TenantId:          1,
				ProjectId:         11,
				Id:                111,
				CloudProviderName: "aws",
			},
			connectorType: dataflow.ConnectorType_MYSQL,
			result:        true,
		},
	}

	for _, tc := range testcases {
		t.Run(tc.name, func(t *testing.T) {
			featureGateReq := &feature_gate.QueryFeatureGateReq{
				TenantID:  tc.cluster.TenantId,
				ProjectID: tc.cluster.ProjectId,
				ClusterID: tc.cluster.Id,
			}
			result := isMultiTiCDCClustersEnabled(featureGate, featureGateReq, tc.cluster, tc.connectorType)
			assert.Equal(t, tc.result, result)
		})
	}
}

func TestMustNotReplicate2Itself(t *testing.T) {
	tests := []struct {
		name                     string
		changefeedType           dataflow.ConnectorType
		mysqlInfo                *dataflow.MySQLInfo
		accessPoints             []*core.AccessPoint
		mockGetAccessPointsError error
		expectedError            error
	}{
		{
			name:           "Missing MySQL endpoint",
			changefeedType: dataflow.ConnectorType_MYSQL,
			mysqlInfo:      &dataflow.MySQLInfo{},
			expectedError:  nil,
		},
		{
			name:           "Invalid MySQL endpoint format",
			changefeedType: dataflow.ConnectorType_MYSQL,
			mysqlInfo: &dataflow.MySQLInfo{
				Connection: &dataflow.Connection{
					Endpoint: "invalid-endpoint",
				},
			},
			expectedError: bizerr_wrapper.NewInvalidInputError("MySQL endpoint should be formatted as {host}:{port}"),
		},
		{
			name:           "Error fetching access points",
			changefeedType: dataflow.ConnectorType_MYSQL,
			mysqlInfo: &dataflow.MySQLInfo{
				Connection: &dataflow.Connection{
					Endpoint: "localhost:3306",
				},
			},
			mockGetAccessPointsError: bizerr_wrapper.NewInternalErr("access points error"),
			expectedError:            bizerr_wrapper.NewInternalErr("Unexpected internal error, please retry later. If problem can not be solved, please contact technical support."),
		},
		{
			name:           "MySQL endpoint matches an access point",
			changefeedType: dataflow.ConnectorType_MYSQL,
			mysqlInfo: &dataflow.MySQLInfo{
				Connection: &dataflow.Connection{
					Endpoint: "localhost:3306",
				},
			},
			accessPoints: []*core.AccessPoint{
				{PublicEndpoint: &core.AccessPoint_PublicEndpoint{Host: "localhost", Port: 3306}},
			},
			expectedError: bizerr_wrapper.NewInvalidInputError("It doesn't support creating a changefeed with the same TiDB cluster as both the source and the target for the changefeed."),
		},
		{
			name:           "Valid MySQL endpoint and no access point match",
			changefeedType: dataflow.ConnectorType_MYSQL,
			mysqlInfo: &dataflow.MySQLInfo{
				Connection: &dataflow.Connection{
					Endpoint: "localhost:3306",
				},
			},
			accessPoints: []*core.AccessPoint{
				{PublicEndpoint: &core.AccessPoint_PublicEndpoint{Host: "127.0.0.1", Port: 3306}},
			},
			expectedError: nil,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			// Setup the mock coreCli
			var mockCoreCli *client_mock.CoreGrpcClient
			needMock := tt.mockGetAccessPointsError != nil || tt.accessPoints != nil
			if needMock {
				mockCoreCli = client_mock.NewCoreGrpcClient(t)
				mockCoreCli.On("GetClusterAccessPoints", mock.Anything, mock.Anything).Return(tt.accessPoints, tt.mockGetAccessPointsError)
			}

			// Create bizChangefeedControllerImpl with the mocked coreCli
			controller := &bizChangefeedControllerImpl{
				coreCli: mockCoreCli,
			}

			// Call the method under test
			err := controller.mustNotReplicate2Itself(context.Background(), 1, tt.changefeedType, tt.mysqlInfo)

			// Assert the expected error
			if tt.expectedError != nil {
				assert.EqualError(t, err, tt.expectedError.Error())
			} else {
				assert.NoError(t, err)
			}

			// Verify that the mock was called
			if needMock {
				mockCoreCli.AssertExpectations(t)
			}
		})
	}
}

func TestParseMySQLEndpoint(t *testing.T) {
	tests := []struct {
		name          string
		endpoint      string
		expectedHost  string
		expectedPort  int32
		expectedError error
	}{
		{
			name:          "Valid MySQL endpoint",
			endpoint:      "localhost:3306",
			expectedHost:  "localhost",
			expectedPort:  3306,
			expectedError: nil,
		},
		{
			name:          "Missing port",
			endpoint:      "localhost",
			expectedHost:  "",
			expectedPort:  0,
			expectedError: bizerr_wrapper.NewInvalidInputError("MySQL endpoint should be formatted as {host}:{port}"),
		},
		{
			name:          "Empty endpoint",
			endpoint:      "",
			expectedHost:  "",
			expectedPort:  0,
			expectedError: bizerr_wrapper.NewInvalidInputError("MySQL endpoint should be formatted as {host}:{port}"),
		},
		{
			name:          "Invalid port (non-numeric)",
			endpoint:      "localhost:abc",
			expectedHost:  "",
			expectedPort:  0,
			expectedError: bizerr_wrapper.NewInvalidInputError("MySQL endpoint's port(%s) is not valid", "abc"),
		},
		{
			name:          "Invalid port (negative number)",
			endpoint:      "localhost:-3306",
			expectedHost:  "",
			expectedPort:  0,
			expectedError: bizerr_wrapper.NewInvalidInputError("MySQL endpoint's port(%s) is not valid", "-3306"),
		},
		{
			name:          "Valid endpoint with port 0",
			endpoint:      "localhost:0",
			expectedHost:  "localhost",
			expectedPort:  0,
			expectedError: nil,
		},
		{
			name:          "Endpoint with extra spaces",
			endpoint:      "  localhost : 3306 ",
			expectedHost:  "localhost",
			expectedPort:  3306,
			expectedError: nil,
		},
		{
			name:          "Multiple colons in endpoint",
			endpoint:      "localhost:3306:extra",
			expectedHost:  "",
			expectedPort:  0,
			expectedError: bizerr_wrapper.NewInvalidInputError("MySQL endpoint should be formatted as {host}:{port}"),
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			host, port, err := parseMySQLEndpoint(tt.endpoint)
			// Check the expected error
			if tt.expectedError != nil {
				assert.EqualError(t, err, tt.expectedError.Error())
			} else {
				assert.NoError(t, err)
			}
			// Check the expected host and port
			assert.Equal(t, tt.expectedHost, host)
			assert.Equal(t, tt.expectedPort, port)
		})
	}
}

func TestIsOneOfAccessPoints(t *testing.T) {
	tests := []struct {
		name           string
		accessPoints   []*core.AccessPoint
		mysqlHost      string
		mysqlPort      int32
		expectedResult bool
	}{
		{
			name: "Match public endpoint",
			accessPoints: []*core.AccessPoint{
				{PublicEndpoint: &core.AccessPoint_PublicEndpoint{Host: "localhost", Port: 3306}},
				{PrivateEndpoint: &core.AccessPoint_PrivateEndpoint{Host: "127.0.0.1", Port: 3307}},
				{VpcPeeringEndpoint: &core.AccessPoint_VpcPeeringEndpoint{Host: "vpc.local", Port: 3308}},
			},
			mysqlHost:      "localhost",
			mysqlPort:      3306,
			expectedResult: true,
		},
		{
			name: "Match private endpoint",
			accessPoints: []*core.AccessPoint{
				{PublicEndpoint: &core.AccessPoint_PublicEndpoint{Host: "localhost", Port: 3306}},
				{PrivateEndpoint: &core.AccessPoint_PrivateEndpoint{Host: "127.0.0.1", Port: 3307}},
				{VpcPeeringEndpoint: &core.AccessPoint_VpcPeeringEndpoint{Host: "vpc.local", Port: 3308}}},
			mysqlHost:      "127.0.0.1",
			mysqlPort:      3307,
			expectedResult: true,
		},
		{
			name: "Match VPC peering endpoint",
			accessPoints: []*core.AccessPoint{
				{PublicEndpoint: &core.AccessPoint_PublicEndpoint{Host: "localhost", Port: 3306}},
				{PrivateEndpoint: &core.AccessPoint_PrivateEndpoint{Host: "127.0.0.1", Port: 3307}},
				{VpcPeeringEndpoint: &core.AccessPoint_VpcPeeringEndpoint{Host: "vpc.local", Port: 3308}}},
			mysqlHost:      "vpc.local",
			mysqlPort:      3308,
			expectedResult: true,
		},
		{
			name: "No match with any endpoint",
			accessPoints: []*core.AccessPoint{
				{PublicEndpoint: &core.AccessPoint_PublicEndpoint{Host: "localhost", Port: 3306}},
				{PrivateEndpoint: &core.AccessPoint_PrivateEndpoint{Host: "127.0.0.1", Port: 3307}},
			},
			mysqlHost:      "10.0.0.1",
			mysqlPort:      3309,
			expectedResult: false,
		},
		{
			name: "Match public endpoint with extra spaces in host",
			accessPoints: []*core.AccessPoint{
				{PublicEndpoint: &core.AccessPoint_PublicEndpoint{Host: "  localhost  ", Port: 3306}},
			},
			mysqlHost:      "localhost",
			mysqlPort:      3306,
			expectedResult: true,
		},
		{
			name:           "Empty access points list",
			accessPoints:   []*core.AccessPoint{},
			mysqlHost:      "localhost",
			mysqlPort:      3306,
			expectedResult: false,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			result := isOneOfAccessPoints(tt.accessPoints, tt.mysqlHost, tt.mysqlPort)
			assert.Equal(t, tt.expectedResult, result)
		})
	}
}

func TestAddressEqual(t *testing.T) {
	tests := []struct {
		name           string
		tidbHost       string
		tidbPort       int32
		mysqlHost      string
		mysqlPort      int32
		expectedResult bool
	}{
		{
			name:           "Both address and ports are equal",
			tidbHost:       "localhost",
			tidbPort:       3306,
			mysqlHost:      "localhost",
			mysqlPort:      3306,
			expectedResult: true,
		},
		{
			name:           "Different hosts, same ports",
			tidbHost:       "localhost",
			tidbPort:       3306,
			mysqlHost:      "127.0.0.1",
			mysqlPort:      3306,
			expectedResult: false,
		},
		{
			name:           "Same hosts, different ports",
			tidbHost:       "localhost",
			tidbPort:       3306,
			mysqlHost:      "localhost",
			mysqlPort:      3307,
			expectedResult: false,
		},
		{
			name:           "Different hosts and different ports",
			tidbHost:       "localhost",
			tidbPort:       3306,
			mysqlHost:      "127.0.0.1",
			mysqlPort:      3307,
			expectedResult: false,
		},
		{
			name:           "Hosts with leading/trailing spaces, same ports",
			tidbHost:       "  localhost  ",
			tidbPort:       3306,
			mysqlHost:      "localhost",
			mysqlPort:      3306,
			expectedResult: true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			result := addressEqual(tt.tidbHost, tt.tidbPort, tt.mysqlHost, tt.mysqlPort)
			if result != tt.expectedResult {
				t.Errorf("expected %v, got %v", tt.expectedResult, result)
			}
		})
	}
}