package cluster

import (
	"context"
	"fmt"
	"reflect"
	"testing"

	"braces.dev/errtrace"
	"github.com/smartystreets/goconvey/convey"
	"go.uber.org/mock/gomock"

	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/mock_adapter"
	"github.com/tidbcloud/tidb-management-service/serverless-global-service/domain/repository/mock_repository"
	"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/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/clusterevent"
	"github.com/tidbcloud/tidb-management-service/utilities/clusterevent/mock_clusterevent"
	"github.com/tidbcloud/tidb-management-service/utilities/featuregate"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow/mock_workflow"
)

var ctx = context.Background()

type MockTransaction struct {
	ctrl     *gomock.Controller
	recorder *MockTransactionMockRecorder
}

type MockTransactionMockRecorder struct {
	mock *MockTransaction
}

func NewMockMockTransaction(ctrl *gomock.Controller) *MockTransaction {
	m := &MockTransaction{ctrl: ctrl}
	m.recorder = &MockTransactionMockRecorder{mock: m}

	return m
}

func (m *MockTransaction) EXPECT() *MockTransactionMockRecorder {
	return m.recorder
}

func (m *MockTransaction) Transaction(ctx context.Context, cb func(ctx context.Context) error) error {
	m.ctrl.T.Helper()
	ret := m.ctrl.Call(m, "Transaction", ctx, cb)
	ret0, ok := ret[0].(bizerr.BizError)
	if !ok {
		return nil
	}
	return ret0
}

// SetSpendLimit indicates an expected call of SetSpendLimit.
func (mr *MockTransactionMockRecorder) Transaction(ctx, cb any) *gomock.Call {
	mr.mock.ctrl.T.Helper()
	return mr.mock.ctrl.RecordCallWithMethodType(
		mr.mock,
		"Transaction",
		reflect.TypeFor[func(ctx context.Context, cb func(ctx context.Context) error) error](),
		ctx,
		cb,
	)
}

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

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

		encryptorSvc := 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)

		committer := mock_clusterevent.NewMockWALCommitter(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()
		clusterSvc, err := NewClusterService(
			flow,
			cloudProviderRepo,
			regionRepo,
			mockedClusterRepo,
			encryptorSvc,
			regionalSvc,
			sys9Lifecycle,
			serverlessSvc,
			rbSvc,
			billingSvc,
			nil,
			nil,
			nil,
			trans,
			committer,
			featGateSvc,
		)
		convey.So(err, convey.ShouldBeNil)
		svc, ok := clusterSvc.(*clusterServiceImpl)
		convey.So(ok, convey.ShouldBeTrue)

		cluster := testdata.GetCluster()
		cloudProviders := testdata.GetCloudProviders()
		createParams := &globalserverlessv1.ServerlessServiceCreateRequest{
			Id:            cluster.ClusterId,
			TenantId:      cluster.TenantId,
			ProjectId:     cluster.ProjectId,
			CloudProvider: cluster.CloudProvider.Id.String(),
			Region:        cluster.Region.Id.String(),
			SpendLimit: &globalserverlessv1.SpendLimit{
				Monthly: 1,
			},
			BackupPolicy: &globalserverlessv1.AutomatedBackupPolicy{
				BackupTime: "06:00",
			},
			EncryptionConfig: &globalserverlessv1.EncryptionConfig{
				EnhancedEncryptionEnabled: false,
			},
		}

		convey.Convey("Execute", func() {
			convey.Convey("should be ok if all is well", func() {
				cluster.UserStatus = globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_CREATING
				encryptorSvc.EXPECT().
					EncryptBase64(gomock.Any(), gomock.Any()).
					Return(testdata.EncryptedCloudAdminPassword, nil)
				encryptorSvc.EXPECT().
					EncryptBase64(gomock.Any(), gomock.Any()).
					Return(testdata.EncryptedCloudAdminPassword, nil)
				cloudProviderRepo.EXPECT().LoadAll(gomock.Any()).Return(cloudProviders, nil)

				mockedClusterRepo.EXPECT().Save(gomock.Any(), gomock.Any()).Return(nil)
				committer.EXPECT().
					BatchWrite(gomock.Any(), clusterevent.TopicClusterEvent, gomock.Any()).
					Do(func(ctx context.Context, topic clusterevent.Topic, events []clusterevent.Event) {
						ev, ok := events[0].(*clusterevent.ClusterEvent)
						convey.So(ok, convey.ShouldBeTrue)
						convey.So(fmt.Sprintf("%d", ev.ClusterID), convey.ShouldEqual, cluster.ClusterId)
					})
				trans.EXPECT().
					Transaction(gomock.Any(), gomock.Any()).
					DoAndReturn(func(ctx context.Context, cb func(ctx context.Context) error) error {
						events, err := cluster.CreateEvent(ctx)
						convey.So(err, convey.ShouldBeNil)

						err = committer.BatchWrite(ctx, clusterevent.TopicClusterEvent, events)
						convey.So(err, convey.ShouldBeNil)

						return errtrace.Wrap(mockedClusterRepo.Save(ctx, cluster))
					})

				_, err := svc.Create(ctx, createParams)
				convey.So(err, convey.ShouldBeNil)
			})

			convey.Convey("should include private authorized networks in creating response", func() {
				cluster = testdata.GetCluster()
				cluster.UserStatus = globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_CREATING

				createReq := &globalserverlessv1.ServerlessServiceCreateRequest{
					Id:            cluster.ClusterId,
					TenantId:      cluster.TenantId,
					ProjectId:     cluster.ProjectId,
					CloudProvider: cluster.CloudProvider.Id.String(),
					Region:        cluster.Region.Id.String(),
					SpendLimit: &globalserverlessv1.SpendLimit{
						Monthly: 1,
					},
					BackupPolicy: &globalserverlessv1.AutomatedBackupPolicy{
						BackupTime: "06:00",
					},
					EncryptionConfig: &globalserverlessv1.EncryptionConfig{
						EnhancedEncryptionEnabled: false,
					},
					Endpoints: &regionalserverlessv1.Endpoints{
						Private: &regionalserverlessv1.PrivateEndpoint{
							AuthorizedNetwork: []*regionalserverlessv1.PrivateEndpoint_AuthorizedNetwork{
								{
									VpcEndpointId: "vpce-1234567890abcdef0",
									DisplayName:   "private-network-1",
								},
							},
						},
					},
				}

				encryptorSvc.EXPECT().
					EncryptBase64(gomock.Any(), gomock.Any()).
					Return(testdata.EncryptedCloudAdminPassword, nil)
				encryptorSvc.EXPECT().
					EncryptBase64(gomock.Any(), gomock.Any()).
					Return(testdata.EncryptedCloudAdminPassword, nil)
				cloudProviderRepo.EXPECT().LoadAll(gomock.Any()).Return(cloudProviders, nil)

				mockedClusterRepo.EXPECT().Save(gomock.Any(), gomock.Any()).Return(nil)
				committer.EXPECT().
					BatchWrite(gomock.Any(), clusterevent.TopicClusterEvent, gomock.Any()).
					AnyTimes()
				trans.EXPECT().
					Transaction(gomock.Any(), gomock.Any()).
					DoAndReturn(func(ctx context.Context, cb func(ctx context.Context) error) error {
						return errtrace.Wrap(mockedClusterRepo.Save(ctx, cluster))
					})

				result, err := svc.Create(ctx, createReq)
				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",
				)
			})

			convey.Convey("should fail when region doesn't support Essential plan", func() {
				// Switch to GCP region which supports only Starter in testdata
				createParams.CloudProvider = testdata.GetGcpProvider().Id.String()
				createParams.Region = testdata.GetGcpRegionUsWest1().Id.String()
				createParams.AutoScaling = &regionalserverlessv1.ServerlessRegionalServiceClusterInfo_AutoScaling{
					MinRcu: 10,
					MaxRcu: 100,
				}
				cloudProviderRepo.EXPECT().LoadAll(gomock.Any()).Return(cloudProviders, nil)

				_, err := svc.Create(ctx, createParams)
				convey.So(err, convey.ShouldNotBeNil)
				convey.So(err.Error(), convey.ShouldContainSubstring, "unsupported region")
			})

			convey.Convey("should pass when region supports Essential plan", func() {
				// us-west-1 supports Starter and Essential in testdata
				createParams.CloudProvider = testdata.GetAwsProvider().Id.String()
				createParams.Region = testdata.GetAwsRegionUsWest1().Id.String()
				createParams.AutoScaling = &regionalserverlessv1.ServerlessRegionalServiceClusterInfo_AutoScaling{
					MinRcu: 11,
					MaxRcu: 100,
				}

				cluster.UserStatus = globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_CREATING
				encryptorSvc.EXPECT().
					EncryptBase64(gomock.Any(), gomock.Any()).
					Return(testdata.EncryptedCloudAdminPassword, nil)
				encryptorSvc.EXPECT().
					EncryptBase64(gomock.Any(), gomock.Any()).
					Return(testdata.EncryptedCloudAdminPassword, nil)
				cloudProviderRepo.EXPECT().LoadAll(gomock.Any()).Return(cloudProviders, nil)

				mockedClusterRepo.EXPECT().Save(gomock.Any(), gomock.Any()).Return(nil)
				committer.EXPECT().
					BatchWrite(gomock.Any(), clusterevent.TopicClusterEvent, gomock.Any()).
					AnyTimes()
				trans.EXPECT().
					Transaction(gomock.Any(), gomock.Any()).
					DoAndReturn(func(ctx context.Context, cb func(ctx context.Context) error) error {
						return errtrace.Wrap(mockedClusterRepo.Save(ctx, cluster))
					})

				_, err := svc.Create(ctx, createParams)
				convey.So(err, convey.ShouldBeNil)
			})
		})
	})
}