package cluster

import (
	"reflect"
	"testing"

	"github.com/smartystreets/goconvey/convey"
	workflowentity "github.com/tidbcloud/fastflow/pkg/entity"
	"go.uber.org/mock/gomock"
	"google.golang.org/protobuf/encoding/protojson"

	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	regionalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/regional/serverless/v1"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/adapter"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/adapter/mock_adapter"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/entity"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/repository"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/repository/mock_repository"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/service/cvt"
	workflowcommon "github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/service/workflow/common"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/testdata"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/pkg/config"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/pkg/utils"
	"github.com/tidbcloud/tidb-management-service/utilities/featuregate"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow/mock_workflow"
)

func TestListCluster(t *testing.T) {
	convey.Convey("TestListCluster", t, func() {
		ctrl := gomock.NewController(t)
		defer ctrl.Finish()

		clusterRepo := mock_repository.NewMockClusterRepo(ctrl)
		cloudProviderRepo := mock_repository.NewMockCloudProviderRepo(ctrl)
		regionRepo := mock_repository.NewMockRegionRepo(ctrl)

		encryptSvc := mock_adapter.NewMockEncryptorService(ctrl)
		serverlessSvc := mock_adapter.NewMockServerlessService(ctrl)
		regionalSvc := mock_adapter.NewMockRegionalService(ctrl)
		sys9Lifecycle := mock_adapter.NewMockSys9Lifecycle(ctrl)
		rbSvc := mock_adapter.NewMockBRService(ctrl)
		billingSvc := mock_adapter.NewMockBillingService(ctrl)
		trans := NewMockMockTransaction(ctrl)
		featGateSvc := featuregate.NewFeatureGate(config.Global().ServerConfig.FeatureGates, nil)
		flow := mock_workflow.NewMockWorkflow(ctrl)
		flow.EXPECT().RegisterDag(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
		flow.EXPECT().ListRunningDagInstances(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, nil).AnyTimes()
		svc, err := NewClusterService(
			flow,
			cloudProviderRepo,
			regionRepo,
			clusterRepo,
			encryptSvc,
			regionalSvc,
			sys9Lifecycle,
			serverlessSvc,
			rbSvc,
			billingSvc,
			nil,
			nil,
			nil,
			trans,
			nil,
			featGateSvc,
		)
		convey.So(err, convey.ShouldBeNil)

		var clusters []*regionalserverlessv1.ServerlessRegionalServiceClusterInfo

		en := testdata.GetCluster()
		req := &globalserverlessv1.ServerlessServiceListRequest{}

		convey.Convey("List", func() {
			convey.Convey("no filter will cause err", func() {
				_, err := svc.List(testdata.Ctx, req)
				convey.So(err, convey.ShouldNotBeNil)
			})

			convey.Convey("have filter and the result exceed 100", func() {
				for range 150 {
					clusters = append(clusters, testdata.GetRegionalCluster())
				}
				clusterRepo.EXPECT().Query(gomock.Any(), gomock.Any()).Return([]*entity.Cluster{en}, nil)

				regionalSvc.EXPECT().ListCluster(gomock.Any(), en.Region, gomock.Any()).Return(clusters, nil)
				req.Filter = &globalserverlessv1.ClusterFilter{ClusterIds: []string{"123"}}
				results, err := svc.List(testdata.Ctx, req)
				convey.So(err, convey.ShouldBeNil)
				convey.So(len(results) == len(clusters), convey.ShouldBeTrue)
			})
		})
	})
}

func TestClusterQueryParameterZero(t *testing.T) {
	convey.Convey("TestClusterQueryParameterZero", t, func() {
		req := &globalserverlessv1.ServerlessServiceListRequest{
			Filter: &globalserverlessv1.ClusterFilter{},
		}
		deleted := false
		if req.Filter != nil && req.Filter.Deleted != nil {
			deleted = *req.Filter.Deleted
		}
		parameter := repository.ClusterQueryParameter{
			ClusterIds:         req.Filter.ClusterIds,
			TenantIds:          req.Filter.TenantIds,
			CloudProviderNames: []globalcommonv1.CloudProvider_Name{},
			ProjectIds:         []uint64{},
			Deleted:            deleted,
		}

		convey.Convey("should be zero value", func() {
			r1 := utils.IsZero(reflect.ValueOf(parameter))
			convey.So(r1, convey.ShouldBeTrue)
		})

		convey.Convey("should not be zero value", func() {
			parameter.Deleted = true

			r1 := utils.IsZero(reflect.ValueOf(parameter))
			convey.So(r1, convey.ShouldBeFalse)
		})

		convey.Convey("should be zero value2", func() {
			parameter.Deleted = false

			r1 := utils.IsZero(reflect.ValueOf(parameter))
			convey.So(r1, convey.ShouldBeTrue)
		})
	})
}

func TestClusterRegionalToGlobalOverlayPrivateAuthorizedNetworks(t *testing.T) {
	convey.Convey("TestClusterRegionalToGlobalOverlayPrivateAuthorizedNetworks", t, func() {
		ctrl := gomock.NewController(t)
		defer ctrl.Finish()

		flow := mock_workflow.NewMockWorkflow(ctrl)
		reqBody, err := protojson.Marshal(&globalserverlessv1.ServerlessServiceUpdateRequest{
			Cluster: &globalserverlessv1.ServerlessServiceCluster{
				Cluster: &regionalserverlessv1.ServerlessRegionalServiceClusterInfo{
					Endpoints: &regionalserverlessv1.Endpoints{
						Private: &regionalserverlessv1.PrivateEndpoint{
							AuthorizedNetwork: []*regionalserverlessv1.PrivateEndpoint_AuthorizedNetwork{
								{
									VpcEndpointId: "vpce-1234567890abcdef0",
									DisplayName:   "private-network-1",
								},
							},
						},
					},
				},
			},
		})
		convey.So(err, convey.ShouldBeNil)

		flow.EXPECT().ListRunningDagInstances(
			gomock.Any(),
			[]string{workflowcommon.UpdateCluster},
			testdata.GetCluster().ClusterId,
		).Return([]*workflowentity.DagInstance{
			{
				DagID: workflowcommon.UpdateCluster,
				Vars: workflowentity.DagInstanceVars{
					workflowcommon.KeyRequestBody: {Value: string(reqBody)},
				},
			},
		}, nil)

		svc := &clusterServiceImpl{flow: flow}
		result, err := svc.clusterRegionalToGlobal(
			testdata.Ctx,
			testdata.GetRegionalCluster(),
			testdata.GetCluster(),
			false,
		)
		convey.So(err, convey.ShouldBeNil)
		convey.So(result.GetCluster().GetEndpoints().GetPrivate().GetAuthorizedNetwork(), convey.ShouldHaveLength, 1)
		convey.So(
			result.GetCluster().GetEndpoints().GetPrivate().GetAuthorizedNetwork()[0].GetVpcEndpointId(),
			convey.ShouldEqual,
			"vpce-1234567890abcdef0",
		)
		convey.So(
			result.GetCluster().GetEndpoints().GetPrivate().GetAuthorizedNetwork()[0].GetDisplayName(),
			convey.ShouldEqual,
			"private-network-1",
		)
	})
}

func TestMergeCreatingClusterOverlayPrivateAuthorizedNetworks(t *testing.T) {
	convey.Convey("TestMergeCreatingClusterOverlayPrivateAuthorizedNetworks", t, func() {
		ctrl := gomock.NewController(t)
		defer ctrl.Finish()

		flow := mock_workflow.NewMockWorkflow(ctrl)
		reqBody, err := protojson.Marshal(&globalserverlessv1.ServerlessServiceCreateRequest{
			Endpoints: &regionalserverlessv1.Endpoints{
				Private: &regionalserverlessv1.PrivateEndpoint{
					AuthorizedNetwork: []*regionalserverlessv1.PrivateEndpoint_AuthorizedNetwork{
						{
							VpcEndpointId: "vpce-1234567890abcdef0",
							DisplayName:   "private-network-1",
						},
					},
				},
			},
		})
		convey.So(err, convey.ShouldBeNil)

		flow.EXPECT().ListRunningDagInstances(
			gomock.Any(),
			[]string{workflowcommon.CreateCluster},
			testdata.ClusterId,
		).Return([]*workflowentity.DagInstance{
			{
				DagID: workflowcommon.CreateCluster,
				Vars: workflowentity.DagInstanceVars{
					workflowcommon.KeyRequestBody: {Value: string(reqBody)},
				},
			},
		}, nil)

		creatingCluster := testdata.GetRegionalCluster()
		creatingCluster.State = regionalserverlessv1.ClusterState_CLUSTER_STATE_CREATING
		globalCluster := cvt.ClusterRegionalToGlobal(creatingCluster)
		globalCluster.UserStatus = globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_CREATING

		svc := &clusterServiceImpl{flow: flow}
		result, err := svc.mergeCreatingCluster(testdata.Ctx, globalCluster, false)
		convey.So(err, convey.ShouldBeNil)
		convey.So(result.GetCluster().GetEndpoints().GetPrivate().GetAuthorizedNetwork(), convey.ShouldHaveLength, 1)
		convey.So(
			result.GetCluster().GetEndpoints().GetPrivate().GetAuthorizedNetwork()[0].GetVpcEndpointId(),
			convey.ShouldEqual,
			"vpce-1234567890abcdef0",
		)
		convey.So(
			result.GetCluster().GetEndpoints().GetPrivate().GetAuthorizedNetwork()[0].GetDisplayName(),
			convey.ShouldEqual,
			"private-network-1",
		)
	})
}

func TestListClusterPGEndpointFromSys9(t *testing.T) {
	convey.Convey("TestListClusterPGEndpointFromSys9", t, func() {
		ctrl := gomock.NewController(t)
		defer ctrl.Finish()

		clusterRepo := mock_repository.NewMockClusterRepo(ctrl)
		cloudProviderRepo := mock_repository.NewMockCloudProviderRepo(ctrl)
		regionRepo := mock_repository.NewMockRegionRepo(ctrl)

		encryptSvc := mock_adapter.NewMockEncryptorService(ctrl)
		serverlessSvc := mock_adapter.NewMockServerlessService(ctrl)
		regionalSvc := mock_adapter.NewMockRegionalService(ctrl)
		sys9Lifecycle := mock_adapter.NewMockSys9Lifecycle(ctrl)
		rbSvc := mock_adapter.NewMockBRService(ctrl)
		billingSvc := mock_adapter.NewMockBillingService(ctrl)
		trans := NewMockMockTransaction(ctrl)
		featGateSvc := featuregate.NewFeatureGate(config.Global().ServerConfig.FeatureGates, nil)
		flow := mock_workflow.NewMockWorkflow(ctrl)
		flow.EXPECT().RegisterDag(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
		flow.EXPECT().ListRunningDagInstances(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, nil).AnyTimes()

		svc, err := NewClusterService(
			flow,
			cloudProviderRepo,
			regionRepo,
			clusterRepo,
			encryptSvc,
			regionalSvc,
			sys9Lifecycle,
			serverlessSvc,
			rbSvc,
			billingSvc,
			nil,
			nil,
			nil,
			trans,
			nil,
			featGateSvc,
		)
		convey.So(err, convey.ShouldBeNil)

		pgCluster := testdata.GetCluster()
		pgCluster.EngineType = "POSTGRESQL_COMPATIBLE"
		pgCluster.ServicePlan = globalcommonv1.ServicePlan_Starter
		pgCluster.UserStatus = globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_NORMAL
		pgCluster.ExternalInstanceID = "inst-1"

		clusterRepo.EXPECT().Query(gomock.Any(), gomock.Any()).Return([]*entity.Cluster{pgCluster}, nil)
		sys9Lifecycle.EXPECT().GetManagedDatabaseConnectionInfo(gomock.Any(), gomock.Any(), "inst-1").
			Return(&adapter.Sys9ManagedDatabaseConnectionInfo{Host: "pg-host", Port: 5432}, nil)

		resp, err := svc.List(testdata.Ctx, &globalserverlessv1.ServerlessServiceListRequest{
			Filter: &globalserverlessv1.ClusterFilter{ClusterIds: []string{testdata.ClusterId}},
		})
		convey.So(err, convey.ShouldBeNil)
		convey.So(resp, convey.ShouldHaveLength, 1)
		convey.So(resp[0].GetCluster().GetEndpoints().GetPublic().GetHost(), convey.ShouldEqual, "pg-host")
		convey.So(resp[0].GetCluster().GetEndpoints().GetPublic().GetPort(), convey.ShouldEqual, int64(5432))
	})
}

func TestListClusterPGEndpointFromSys9WhenRegionalIncluded(t *testing.T) {
	convey.Convey("TestListClusterPGEndpointFromSys9WhenRegionalIncluded", t, func() {
		ctrl := gomock.NewController(t)
		defer ctrl.Finish()

		clusterRepo := mock_repository.NewMockClusterRepo(ctrl)
		cloudProviderRepo := mock_repository.NewMockCloudProviderRepo(ctrl)
		regionRepo := mock_repository.NewMockRegionRepo(ctrl)

		encryptSvc := mock_adapter.NewMockEncryptorService(ctrl)
		serverlessSvc := mock_adapter.NewMockServerlessService(ctrl)
		regionalSvc := mock_adapter.NewMockRegionalService(ctrl)
		sys9Lifecycle := mock_adapter.NewMockSys9Lifecycle(ctrl)
		rbSvc := mock_adapter.NewMockBRService(ctrl)
		billingSvc := mock_adapter.NewMockBillingService(ctrl)
		trans := NewMockMockTransaction(ctrl)
		featGateSvc := featuregate.NewFeatureGate(config.Global().ServerConfig.FeatureGates, nil)
		flow := mock_workflow.NewMockWorkflow(ctrl)
		flow.EXPECT().RegisterDag(gomock.Any(), gomock.Any()).Return(nil).AnyTimes()
		flow.EXPECT().ListRunningDagInstances(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, nil).AnyTimes()

		svc, err := NewClusterService(
			flow,
			cloudProviderRepo,
			regionRepo,
			clusterRepo,
			encryptSvc,
			regionalSvc,
			sys9Lifecycle,
			serverlessSvc,
			rbSvc,
			billingSvc,
			nil,
			nil,
			nil,
			trans,
			nil,
			featGateSvc,
		)
		convey.So(err, convey.ShouldBeNil)

		pgCluster := testdata.GetCluster()
		pgCluster.EngineType = "POSTGRESQL_COMPATIBLE"
		pgCluster.ServicePlan = globalcommonv1.ServicePlan_Starter
		pgCluster.UserStatus = globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_NORMAL
		pgCluster.ExternalInstanceID = "inst-2"

		mysqlCluster := testdata.GetCluster()
		mysqlCluster.ClusterId = "222"
		mysqlCluster.EngineType = "MYSQL_COMPATIBLE"
		mysqlCluster.ServicePlan = globalcommonv1.ServicePlan_Starter

		clusterRepo.EXPECT().Query(gomock.Any(), gomock.Any()).Return([]*entity.Cluster{pgCluster, mysqlCluster}, nil)
		regionalCluster := testdata.GetRegionalCluster()
		regionalCluster.Endpoints = nil
		regionalSvc.EXPECT().
			ListCluster(gomock.Any(), pgCluster.Region, gomock.Any()).
			Return([]*regionalserverlessv1.ServerlessRegionalServiceClusterInfo{
				regionalCluster,
			}, nil)
		sys9Lifecycle.EXPECT().GetManagedDatabaseConnectionInfo(gomock.Any(), gomock.Any(), "inst-2").
			Return(&adapter.Sys9ManagedDatabaseConnectionInfo{Host: "pg-host-2", Port: 5432}, nil)

		resp, err := svc.List(testdata.Ctx, &globalserverlessv1.ServerlessServiceListRequest{
			Filter: &globalserverlessv1.ClusterFilter{ClusterIds: []string{testdata.ClusterId, "222"}},
		})
		convey.So(err, convey.ShouldBeNil)
		convey.So(resp, convey.ShouldNotBeEmpty)

		var pgResp *globalserverlessv1.ServerlessServiceCluster
		for _, c := range resp {
			if c.GetClusterId() == testdata.ClusterId {
				pgResp = c
				break
			}
		}
		convey.So(pgResp, convey.ShouldNotBeNil)
		convey.So(pgResp.GetCluster().GetEndpoints().GetPublic().GetHost(), convey.ShouldEqual, "pg-host-2")
		convey.So(pgResp.GetCluster().GetEndpoints().GetPublic().GetPort(), convey.ShouldEqual, int64(5432))
	})
}