package service

import (
	"context"
	"testing"

	"github.com/golang/mock/gomock"
	"github.com/stretchr/testify/assert"
	apiv1 "github.com/tidbcloud/infra-provider/api/tidbcloud/ticdc-service/v1"
	"github.com/tidbcloud/ticdc-service/pkg/clusterproxy"
	"github.com/tidbcloud/ticdc-service/pkg/clusterproxy/api"
	apiMock "github.com/tidbcloud/ticdc-service/pkg/clusterproxy/api/mocks"
	proxyModelMock "github.com/tidbcloud/ticdc-service/pkg/clusterproxy/mocks"
)

func TestPreCheckServerImpl_TestNetworkConnection(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	mockProxy := proxyModelMock.NewMockClusterProxy(ctrl)
	modelMock := proxyModelMock.NewMockModel(ctrl)

	server := NewPreCheckServiceServer(modelMock)

	req := &apiv1.TestNetworkConnectionRequest{
		ClusterId: "test-cluster",
		Type:      "mysql",
		MysqlParam: &apiv1.TestNetworkConnectionRequest_MySQLParameter{
			User:     "user",
			Password: "password",
			Url:      "url",
		},
	}

	modelMock.EXPECT().GetClusterProxy(gomock.Any(), req.ClusterId, clusterproxy.CDCHeavyOperationTimeout).Return(mockProxy, nil)
	cfCli := apiMock.NewMockChangefeedAPIV2(ctrl)
	mockProxy.EXPECT().ChangefeedClient().Return(cfCli)
	cfCli.EXPECT().TestConnection(gomock.Any(), gomock.Any()).Return(nil)

	_, err := server.TestNetworkConnection(context.Background(), req)
	assert.NoError(t, err)
}

func TestPreCheckServerImpl_ValidateChangefeedParameters(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	mockProxy := proxyModelMock.NewMockClusterProxy(ctrl)
	modelMock := proxyModelMock.NewMockModel(ctrl)

	server := NewPreCheckServiceServer(modelMock)

	req := &apiv1.ValidateChangefeedParametersRequest{
		ClusterId: "test-cluster",
		Type:      "kafka",
		KafkaParam: &apiv1.ValidateChangefeedParametersRequest_KafkaParameter{
			User:      "user",
			Password:  "password",
			Url:       "url",
			EnableTls: true,
			AuthType:  "auth",
			Topic:     "topic",
		},
	}

	modelMock.EXPECT().GetClusterProxy(gomock.Any(), req.ClusterId, clusterproxy.CDCHeavyOperationTimeout).Return(mockProxy, nil)
	cfCli := apiMock.NewMockChangefeedAPIV2(ctrl)
	mockProxy.EXPECT().ChangefeedClient().Return(cfCli)
	cfCli.EXPECT().TestConnection(gomock.Any(), gomock.Any()).Return(nil)

	_, err := server.ValidateChangefeedParameters(context.Background(), req)
	assert.NoError(t, err)
}

func TestPreCheckServerImpl_VerifyTable(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	mockProxy := proxyModelMock.NewMockClusterProxy(ctrl)
	modelMock := proxyModelMock.NewMockModel(ctrl)

	server := NewPreCheckServiceServer(modelMock)

	clusterID := "test-cluster"
	modelMock.EXPECT().GetClusterProxy(gomock.Any(), clusterID, clusterproxy.CDCHeavyOperationTimeout).Return(mockProxy, nil).AnyTimes()
	sqlCli := apiMock.NewMockSQLClient(ctrl)
	mockProxy.EXPECT().SQLClient().Return(sqlCli).AnyTimes()

	sqlCli.EXPECT().QuerySQL(gomock.Any(), gomock.Any(), QueryEligibleSchemaTableClaus, gomock.Any()).DoAndReturn(
		func(ctx context.Context, operator *api.ClusterOperator, sql string, dest interface{}) error {
			*dest.(*TableSchemaInfo) = TableSchemaInfo{
				TableSchema: []string{"dGVzdA==", "VEVTVA==", "dGlkYl9jZGM=", "bGlnaHRuaW5nX3Rhc2tfaW5mbw=="},
				TableName:   []string{"dGVzdA==", "VEVTVA==", "dGlkYl9jZGM=", "bGlnaHRuaW5nX3Rhc2tfaW5mbw=="},
			}
			return nil
		}).AnyTimes()
	sqlCli.EXPECT().QuerySQL(gomock.Any(), gomock.Any(), QueryIneligibleSchemaTableClaus, gomock.Any()).DoAndReturn(
		func(ctx context.Context, operator *api.ClusterOperator, sql string, dest interface{}) error {
			*dest.(*TableSchemaInfo) = TableSchemaInfo{
				TableSchema: []string{"dGVzdA==", "VEVTVA==", "dGlkYl9jZGM=", "bGlnaHRuaW5nX3Rhc2tfaW5mbw=="},
				TableName:   []string{"dGVzdA==", "VEVTVA==", "dGlkYl9jZGM=", "bGlnaHRuaW5nX3Rhc2tfaW5mbw=="},
			}
			return nil
		}).AnyTimes()

	t.Run("case sensitive", func(t *testing.T) {
		req := &apiv1.VerifyTableRequest{
			TidbClusterId: clusterID,
			User:          "user",
			Password:      "password",
			FilterRules:   []string{"test.*"},
			CaseSensitive: true,
		}

		resp, err := server.VerifyTable(context.Background(), req)
		assert.NoError(t, err)
		assert.NotNil(t, resp)
		assert.Equal(t, 1, len(resp.EligibleTables))
		assert.Equal(t, 1, len(resp.IneligibleTables))
	})

	t.Run("case insensitive", func(t *testing.T) {
		req := &apiv1.VerifyTableRequest{
			TidbClusterId: clusterID,
			User:          "user",
			Password:      "password",
			FilterRules:   []string{"test.*"},
			CaseSensitive: false,
		}

		resp, err := server.VerifyTable(context.Background(), req)
		assert.NoError(t, err)
		assert.NotNil(t, resp)
		assert.Equal(t, 2, len(resp.EligibleTables))
		assert.Equal(t, 2, len(resp.IneligibleTables))
	})
}

func TestBuildTestConnectionModel(t *testing.T) {
	tests := []struct {
		name string
		req  *apiv1.TestNetworkConnectionRequest
		want *api.ConnectionModel
	}{
		{
			name: "MySQL parameters",
			req: &apiv1.TestNetworkConnectionRequest{
				Type: "mysql",
				MysqlParam: &apiv1.TestNetworkConnectionRequest_MySQLParameter{
					User:     "user",
					Password: "password",
					Url:      "url",
				},
			},
			want: &api.ConnectionModel{
				Type:     "mysql",
				User:     "user",
				Password: "password",
				Url:      "url",
			},
		},
		{
			name: "Kafka parameters",
			req: &apiv1.TestNetworkConnectionRequest{
				Type: "kafka",
				KafkaParam: &apiv1.TestNetworkConnectionRequest_KafkaParameter{
					User:              "user",
					Password:          "password",
					EnableTls:         true,
					AuthType:          "auth",
					BoostrapEndpoints: "kafka://127.0.0.1:9092",
				},
			},
			want: &api.ConnectionModel{
				Type:      "kafka",
				User:      "user",
				Password:  "password",
				Url:       "kafka://127.0.0.1:9092",
				EnableTls: true,
				AuthType:  "auth",
			},
		},
		{
			name: "S3 parameters",
			req: &apiv1.TestNetworkConnectionRequest{
				Type: "s3",
				S3Param: &apiv1.TestNetworkConnectionRequest_S3Parameter{
					S3Uri:           "s3://bucket",
					AccessKeyId:     "accessKeyId",
					SecretAccessKey: "secretAccessKey",
					RoleArn:         "roleArn",
					ExternalId:      "externalId",
				},
			},
			want: &api.ConnectionModel{
				Type: "s3",
				S3Param: api.S3Parameter{
					S3Uri:           "s3://bucket",
					AccessKeyId:     "accessKeyId",
					SecretAccessKey: "secretAccessKey",
					RoleArn:         "roleArn",
					ExternalID:      "externalId",
				},
			},
		},
		{
			name: "GCS parameters",
			req: &apiv1.TestNetworkConnectionRequest{
				Type: "gcs",
				GcsParam: &apiv1.TestNetworkConnectionRequest_GCSParameter{
					GcsUrl:                    "gcs://bucket",
					Region:                    "region",
					ImpersonateServiceAccount: "serviceAccount",
				},
			},
			want: &api.ConnectionModel{
				Type: "gcs",
				GCSParam: api.GcsParameter{
					GcsUrl:                    "gcs://bucket",
					Region:                    "region",
					ImpersonateServiceAccount: "serviceAccount",
				},
			},
		},
		{
			name: "pulsar parameters",
			req: &apiv1.TestNetworkConnectionRequest{
				Type: "pulsar",
				PulsarParam: &apiv1.TestNetworkConnectionRequest_PulsarParameter{
					ServiceUrl:         "pulsar://pulsar-service",
					AuthenticationType: "basic",
					Username:           "username",
					Password:           "password",
					Token:              "",
					ConnectionTimeout:  5,
					Tenant:             "public",
					Namespace:          "default",
					Topic:              "test",
				},
			},
			want: &api.ConnectionModel{
				Type: "pulsar",
				PulsarParam: api.PulsarParameter{
					ServiceURL:         "pulsar://pulsar-service",
					AuthenticationType: "basic",
					Username:           "username",
					Password:           "password",
					Token:              "",
					ConnectionTimeout:  5,
					Tenant:             "public",
					Namespace:          "default",
					Topic:              "test",
				},
			},
		},
		{
			name: "azure blob storage parameters",
			req: &apiv1.TestNetworkConnectionRequest{
				Type: "azblob",
				AzureBlobStorageParam: &apiv1.TestNetworkConnectionRequest_AzureBlobStorageParameter{
					BlobUri:  "https://account.blob.core.windows.net/container",
					SasToken: "token123",
				},
			},
			want: &api.ConnectionModel{
				Type: "azblob",
				AzureBlobStorageParam: api.AzureBlobStorageParameter{
					BlobUri:  "https://account.blob.core.windows.net/container",
					SASToken: "token123",
				},
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			got := buildTestConnectionModel(tt.req)
			assert.Equal(t, tt.want, got)
		})
	}
}