package cdc

import (
	"context"
	"fmt"
	"testing"
	"time"

	"github.com/golang/mock/gomock"
	"github.com/pingcap/dbaas/central/config"
	"github.com/pingcap/dbaas/central/db"
	mock_cdcdao "github.com/pingcap/dbaas/central/db/dbutil/cdc/mock"
	mock_clusterdao "github.com/pingcap/dbaas/central/db/dbutil/cluster/mock"
	"github.com/pingcap/dbaas/central/db/entity"
	"github.com/pingcap/dbaas/central/db/tenant"
	"github.com/pingcap/dbaas/central/handler/billing2"
	"github.com/pingcap/dbaas/central/infra-api/mock_infraapi"
	"github.com/pingcap/dbaas/central/model/billing"
	"github.com/pingcap/dbaas/central/test"
	"github.com/pingcap/log"
	. "github.com/smartystreets/goconvey/convey"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/idl/pbgen/proto/central"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/pkgv2/clusterproxy"
	"github.com/tidbcloud/pkgv2/clusterproxy/mock_clusterproxy"
	"github.com/tidbcloud/pkgv2/secret/encrypt"
	mock_encrypt "github.com/tidbcloud/pkgv2/secret/encrypt/mock_encryptor"
	"gorm.io/gorm"
)

func TestCleanUnunsedCdcCluster(t *testing.T) {
	Convey("TestCleanUnusedCdcCluster", t, func() {
		ctrl := gomock.NewController(t)
		defer ctrl.Finish()
		logger := log.L()
		ctx := context.Background()
		mockInfraCli := mock_infraapi.NewMockInfraFacade(ctrl)
		mockClusterDaoCli := mock_clusterdao.NewMockClusterDaoFacade(ctrl)
		mockCdcDaoCli := mock_cdcdao.NewMockCDCDaoFacade(ctrl)
		model := Model{
			log:        logger,
			api:        mockInfraCli,
			clusterdao: mockClusterDaoCli,
			dao:        mockCdcDaoCli,
		}
		userContext := UserContext{
			Model: model,
			log:   logger,
			tenantAndUser: &tenant.User{
				TenantID: entity.ID(111),
			},
		}
		clusterList := []entity.Cluster{
			{
				Updatable: entity.Updatable{
					Base: entity.Base{
						ID:        entity.ID(1),
						CreatedAt: time.Now().Add(-24 * time.Hour),
					},
				},
				ProjectID: entity.ID(2),
				Status:    entity.ClusterStatusScaling,
			},
		}
		cdcList := []entity.CDCTask{}

		// return empty cdc list for test
		mockCdcDaoCli.EXPECT().FindCDCTasks(gomock.Any()).Return(cdcList, nil)
		// mock infra api return cr list
		mockInfraCli.EXPECT().DeleteTicdcIfExist(ctx, clusterList[0].ProjectID.Uint64(), clusterList[0].ID.Uint64()).Return(nil)
		// return backup entity list when use dao layer
		mockClusterDaoCli.EXPECT().FindClusters(gomock.Any()).Return(clusterList, nil)

		err := BgContext{userContext}.CleanUnusedCdcCluster(ctx)
		So(err, ShouldBeNil)
	})
}

func TestManageCdcRecords(t *testing.T) {
	Convey("Test_ManagerCDCRecord", t, func() {
		var err error
		ctrl := gomock.NewController(t)
		defer ctrl.Finish()
		logger := log.L()
		ctx := context.Background()
		mockInfraCli := mock_infraapi.NewMockInfraFacade(ctrl)
		mockClusterDaoCli := mock_clusterdao.NewMockClusterDaoFacade(ctrl)
		mockCdcDaoCli := mock_cdcdao.NewMockCDCDaoFacade(ctrl)
		mockClusterproxy := mock_clusterproxy.NewMockClusterProxy(ctrl)
		mockEncryptorCli := mock_encrypt.NewMockEncryptor(ctrl)

		mockGetProxy := func(clusterID uint64, status cloudv1.ClusterStatus, timeout int) (clusterproxy.ClusterProxy, error) {
			return mockClusterproxy, nil
		}

		model := Model{
			log:        logger,
			api:        mockInfraCli,
			clusterdao: mockClusterDaoCli,
			dao:        mockCdcDaoCli,
			encryptor:  mockEncryptorCli,
		}
		userContext := UserContext{
			Model: model,
			log:   logger,
			tenantAndUser: &tenant.User{
				TenantID: entity.ID(111),
			},
		}

		cdcTaskCreatingKafka := entity.CDCTask{
			Updatable:      entity.Updatable{Base: entity.Base{ID: entity.ID(1)}},
			Status:         entity.CDCStatusCreating,
			DownStreamType: entity.CDCDownStreamTypeKafka,
			Cluster: &entity.Cluster{
				Updatable: entity.Updatable{Base: entity.Base{ID: entity.ID(1)}},
				ProjectID: entity.ID(1),
			},
		}

		cdcTaskDeleteCluster := entity.CDCTask{
			Updatable:      entity.Updatable{Base: entity.Base{ID: entity.ID(1)}},
			Status:         entity.CDCStatusNormal,
			DownStreamType: entity.CDCDownStreamTypeKafka,
			Cluster: &entity.Cluster{
				Updatable: entity.Updatable{Base: entity.Base{ID: entity.ID(1)}},
				ProjectID: entity.ID(1),
				Status:    entity.ClusterStatusDeleted,
			},
		}

		cdcTaskCreatingMysql := entity.CDCTask{
			Updatable:      entity.Updatable{Base: entity.Base{ID: entity.ID(1)}},
			Status:         entity.CDCStatusCreating,
			DownStreamType: entity.CDCDownStreamTypeMysql,
			Cluster: &entity.Cluster{
				Updatable: entity.Updatable{Base: entity.Base{ID: entity.ID(1)}},
				ProjectID: entity.ID(1),
			},
		}

		cdcTaskDeletingKafka := entity.CDCTask{
			Updatable:      entity.Updatable{Base: entity.Base{ID: entity.ID(1)}},
			Status:         entity.CDCStatusDeleting,
			DownStreamType: entity.CDCDownStreamTypeKafka,
			Cluster: &entity.Cluster{
				Updatable: entity.Updatable{Base: entity.Base{ID: entity.ID(1)}},
				ProjectID: entity.ID(1),
			},
		}

		cdcTaskDeletingMysql := entity.CDCTask{
			Updatable:      entity.Updatable{Base: entity.Base{ID: entity.ID(1)}},
			Status:         entity.CDCStatusDeleting,
			DownStreamType: entity.CDCDownStreamTypeMysql,
			Cluster: &entity.Cluster{
				Updatable: entity.Updatable{Base: entity.Base{ID: entity.ID(1)}},
				ProjectID: entity.ID(1),
			},
		}

		cdcTaskNormalKafka := entity.CDCTask{
			Updatable:      entity.Updatable{Base: entity.Base{ID: entity.ID(2)}},
			Status:         entity.CDCStatusNormal,
			DownStreamType: entity.CDCDownStreamTypeKafka,
			Cluster: &entity.Cluster{
				Updatable: entity.Updatable{Base: entity.Base{ID: entity.ID(1)}},
				ProjectID: entity.ID(1),
			},
		}

		cdcTaskFailedMysql := entity.CDCTask{
			Updatable:      entity.Updatable{Base: entity.Base{ID: entity.ID(2)}},
			Status:         entity.CDCStatusFailed,
			DownStreamType: entity.CDCDownStreamTypeMysql,
			Cluster: &entity.Cluster{
				Updatable: entity.Updatable{Base: entity.Base{ID: entity.ID(1)}},
				ProjectID: entity.ID(1),
			},
		}

		cdcTaskNormalMysql := entity.CDCTask{
			Updatable:      entity.Updatable{Base: entity.Base{ID: entity.ID(1)}},
			Status:         entity.CDCStatusFailed,
			DownStreamType: entity.CDCDownStreamTypeMysql,
			Cluster: &entity.Cluster{
				Updatable: entity.Updatable{Base: entity.Base{ID: entity.ID(1)}},
				ProjectID: entity.ID(1),
			},
		}

		cdcInfoList1 := []clusterproxy.CdcInfo{
			{
				ID:    "changefeed-1",
				State: "normal",
			},
		}

		cdcInfoList2 := []clusterproxy.CdcInfo{
			{
				ID:    "changefeed-2",
				State: "normal",
			},
		}

		Convey("should be create", func() {
			mockInfraCli.EXPECT().GetCluster(ctx, gomock.Any(), gomock.Any()).Return(&cloudv1.Cluster{}, nil)
			mockClusterproxy.EXPECT().GetCDCTaskInfo(ctx).Return(cdcInfoList2, nil)
			mockCdcDaoCli.EXPECT().GetCdcKafkaTask(gomock.Any()).Return(&entity.CDCKafkaTask{}, nil)
			mockInfraCli.EXPECT().GetCluster(ctx, gomock.Any(), gomock.Any()).Return(&cloudv1.Cluster{}, nil)
			mockClusterproxy.EXPECT().CreateCDCTask(ctx, gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil)
			err = BgContext{userContext}.manageCdcRecord(ctx, false, cdcTaskCreatingKafka, nil, mockGetProxy)
			So(err, ShouldBeNil)

			mockInfraCli.EXPECT().GetCluster(ctx, gomock.Any(), gomock.Any()).Return(&cloudv1.Cluster{}, nil)
			mockClusterproxy.EXPECT().GetCDCTaskInfo(ctx).Return(cdcInfoList2, nil)
			mockCdcDaoCli.EXPECT().GetCdcSqlTask(gomock.Any()).Return(&entity.CDCSQLTask{}, nil)
			mockInfraCli.EXPECT().GetCluster(ctx, gomock.Any(), gomock.Any()).Return(&cloudv1.Cluster{}, nil)
			mockClusterproxy.EXPECT().CreateCDCTask(ctx, gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil)
			mockEncryptorCli.EXPECT().Decrypt(gomock.Any()).Return(encrypt.Secret(""), nil)
			err = BgContext{userContext}.manageCdcRecord(ctx, false, cdcTaskCreatingMysql, nil, mockGetProxy)
			So(err, ShouldBeNil)
		})
		Convey("should be delete", func() {
			mockInfraCli.EXPECT().GetCluster(ctx, gomock.Any(), gomock.Any()).Return(&cloudv1.Cluster{}, nil)
			mockClusterproxy.EXPECT().GetCDCTaskInfo(ctx).Return(cdcInfoList1, nil)
			mockInfraCli.EXPECT().GetCluster(ctx, gomock.Any(), gomock.Any()).Return(&cloudv1.Cluster{}, nil)
			mockClusterproxy.EXPECT().DeleteCDCTask(ctx, cdcInfoList1[0].ID)
			err = BgContext{userContext}.manageCdcRecord(ctx, false, cdcTaskDeletingKafka, nil, mockGetProxy)
			So(err, ShouldBeNil)

			mockInfraCli.EXPECT().GetCluster(ctx, gomock.Any(), gomock.Any()).Return(&cloudv1.Cluster{}, nil)
			mockClusterproxy.EXPECT().GetCDCTaskInfo(ctx).Return(cdcInfoList1, nil)
			mockInfraCli.EXPECT().GetCluster(ctx, gomock.Any(), gomock.Any()).Return(&cloudv1.Cluster{}, nil)
			mockClusterproxy.EXPECT().DeleteCDCTask(ctx, cdcInfoList1[0].ID)
			err = BgContext{userContext}.manageCdcRecord(ctx, false, cdcTaskDeletingMysql, nil, mockGetProxy)
			So(err, ShouldBeNil)

			mockCdcDaoCli.EXPECT().UpdateCdcTask(gomock.Any(), gomock.Any()).Return(nil)
			err = BgContext{userContext}.manageCdcRecord(ctx, true, cdcTaskDeleteCluster, nil, mockGetProxy)
			So(err, ShouldBeNil)
		})
		Convey("should be update status", func() {
			mockInfraCli.EXPECT().GetCluster(ctx, gomock.Any(), gomock.Any()).Return(&cloudv1.Cluster{}, nil)
			mockClusterproxy.EXPECT().GetCDCTaskInfo(ctx).Return(cdcInfoList2, nil)
			mockCdcDaoCli.EXPECT().UpdateCdcTask(gomock.Any(), gomock.Any()).Return(nil)
			err = BgContext{userContext}.manageCdcRecord(ctx, false, cdcTaskNormalKafka, nil, mockGetProxy)
			So(err, ShouldBeNil)

			mockInfraCli.EXPECT().GetCluster(ctx, gomock.Any(), gomock.Any()).Return(&cloudv1.Cluster{}, nil)
			mockClusterproxy.EXPECT().GetCDCTaskInfo(ctx).Return(cdcInfoList2, nil)
			mockCdcDaoCli.EXPECT().UpdateCdcTask(gomock.Any(), gomock.Any()).Return(nil)
			err = BgContext{userContext}.manageCdcRecord(ctx, false, cdcTaskFailedMysql, nil, mockGetProxy)
			So(err, ShouldBeNil)

			mockInfraCli.EXPECT().GetCluster(ctx, gomock.Any(), gomock.Any()).Return(&cloudv1.Cluster{}, nil)
			mockClusterproxy.EXPECT().GetCDCTaskInfo(ctx).Return(cdcInfoList2, nil)
			mockCdcDaoCli.EXPECT().UpdateCdcTask(gomock.Any(), gomock.Any()).Return(nil)
			err = BgContext{userContext}.manageCdcRecord(ctx, false, cdcTaskNormalMysql, nil, mockGetProxy)
			So(err, ShouldBeNil)
		})
		Convey("should be deleted", func() {
			mockInfraCli.EXPECT().GetCluster(ctx, gomock.Any(), gomock.Any()).Return(&cloudv1.Cluster{}, nil)
			mockClusterproxy.EXPECT().GetCDCTaskInfo(ctx).Return(cdcInfoList2, nil)
			mockCdcDaoCli.EXPECT().UpdateCdcTask(gomock.Any(), gomock.Any()).Return(nil)
			err = BgContext{userContext}.manageCdcRecord(ctx, false, cdcTaskDeletingKafka, nil, mockGetProxy)
			So(err, ShouldBeNil)

			mockInfraCli.EXPECT().GetCluster(ctx, gomock.Any(), gomock.Any()).Return(&cloudv1.Cluster{}, nil)
			mockClusterproxy.EXPECT().GetCDCTaskInfo(ctx).Return(cdcInfoList2, nil)
			mockCdcDaoCli.EXPECT().UpdateCdcTask(gomock.Any(), gomock.Any()).Return(nil)
			err = BgContext{userContext}.manageCdcRecord(ctx, false, cdcTaskDeletingMysql, nil, mockGetProxy)
			So(err, ShouldBeNil)
		})
	})
}

func TestCreateCdcTaskImpl(t *testing.T) {
	Convey("Test_CreateCdcTaskImpl", t, func() {
		ctrl := gomock.NewController(t)
		defer ctrl.Finish()
		logger := log.L()
		mockClusterDaoCli := mock_clusterdao.NewMockClusterDaoFacade(ctrl)
		mockEncryptorCli := mock_encrypt.NewMockEncryptor(ctrl)
		importID := uint64(1)

		clusterRecovering := entity.Cluster{
			Updatable: entity.Updatable{Base: entity.Base{ID: entity.ID(1)}},
			ProjectID: entity.ID(1),
			Status:    entity.ClusterStatusRecovering,
		}

		clusterImporting := entity.Cluster{
			Updatable: entity.Updatable{Base: entity.Base{ID: entity.ID(1)}},
			ProjectID: entity.ID(1),
			ImportID:  &importID,
		}

		clusterCreating := entity.Cluster{
			Updatable: entity.Updatable{Base: entity.Base{ID: entity.ID(1)}},
			ProjectID: entity.ID(1),
			Status:    entity.ClusterStatusInitializing,
		}

		clusterDevTier := entity.Cluster{
			Updatable: entity.Updatable{Base: entity.Base{ID: entity.ID(1)}},
			ProjectID: entity.ID(1),
			Status:    entity.ClusterStatusNormal,
			Free:      true,
		}

		Convey("should be recovering error", func() {
			mockClusterDaoCli.EXPECT().GetCluster(entity.ID(1)).Return(clusterRecovering, nil)

			_, err := CreateCdcTaskImpl(
				entity.ID(1),
				central.ConnectorType_KAFKA,
				"a.com:100",
				"test",
				"a",
				"pwd",
				12345,
				mockClusterDaoCli,
				mockEncryptorCli,
				logger,
				false,
			)
			So(err.Error(), ShouldEqual, RecoveringErr)
		})

		Convey("should be creating error", func() {
			mockClusterDaoCli.EXPECT().GetCluster(entity.ID(1)).Return(clusterCreating, nil)

			_, err := CreateCdcTaskImpl(
				entity.ID(1),
				central.ConnectorType_KAFKA,
				"a.com:100",
				"test",
				"a",
				"pwd",
				12345,
				mockClusterDaoCli,
				mockEncryptorCli,
				logger,
				false,
			)
			So(err.Error(), ShouldEqual, CreatingErr)
		})

		Convey("should be importing error", func() {
			mockClusterDaoCli.EXPECT().GetCluster(entity.ID(1)).Return(clusterImporting, nil)

			_, err := CreateCdcTaskImpl(
				entity.ID(1),
				central.ConnectorType_KAFKA,
				"a.com:100",
				"test",
				"a",
				"pwd",
				12345,
				mockClusterDaoCli,
				mockEncryptorCli,
				logger,
				false,
			)
			So(err.Error(), ShouldEqual, ImportingErr)
		})

		Convey("should be dev tier error", func() {
			mockClusterDaoCli.EXPECT().GetCluster(entity.ID(1)).Return(clusterDevTier, nil)

			_, err := CreateCdcTaskImpl(
				entity.ID(1),
				central.ConnectorType_KAFKA,
				"a.com:100",
				"test",
				"a",
				"pwd",
				12345,
				mockClusterDaoCli,
				mockEncryptorCli,
				logger,
				false,
			)
			So(err.Error(), ShouldEqual, NotallowDevTierErr)
		})
	})
}

func TestTestCdcTask(t *testing.T) {
	Convey("Test_TestCdcTask", t, func() {
		ctrl := gomock.NewController(t)
		defer ctrl.Finish()
		logger := log.L()
		ctx := context.Background()
		mockClusterDao := mock_clusterdao.NewMockClusterDaoFacade(ctrl)
		mockClusterproxy := mock_clusterproxy.NewMockClusterProxy(ctrl)
		mockInfraCli := mock_infraapi.NewMockInfraFacade(ctrl)
		model := Model{
			clusterdao: mockClusterDao,
			log:        logger,
			api:        mockInfraCli,
		}
		userContext := UserContext{
			Model: model,
			log:   logger,
			tenantAndUser: &tenant.User{
				TenantID: entity.ID(111),
			},
		}
		safepoint := clusterproxy.Safepoint{
			SP: uint64(12345678),
		}
		mockGetProxy := func(clusterID uint64, status cloudv1.ClusterStatus, timeout int) (clusterproxy.ClusterProxy, error) {
			return mockClusterproxy, nil
		}
		cidr := entity.ProjectNetwork{
			CIDR: "192.168.0.0/24",
		}

		Convey("should be failed in get start tso", func() {
			mockClusterDao.EXPECT().GetProjectCIDRByClusterID(gomock.Any(), gomock.Any()).Return(&cidr, nil)
			mockInfraCli.EXPECT().GetCluster(ctx, gomock.Any(), gomock.Any()).Return(&cloudv1.Cluster{}, nil)
			mockClusterproxy.EXPECT().GetPDSafepoint(ctx).Return(&safepoint, nil)
			err := userContext.TestCdcTask(
				ctx,
				entity.ID(1),
				entity.ID(1),
				central.ConnectorType_MYSQL,
				"192.168.1.1:100",
				"test",
				"root",
				"123456",
				12345,
				mockGetProxy,
			)
			So(err.Error(), ShouldEqual, FailedToGetTSO)
		})

		Convey("should be failed in get start tso tidb", func() {
			mockClusterDao.EXPECT().GetProjectCIDRByClusterID(gomock.Any(), gomock.Any()).Return(&cidr, nil)
			mockInfraCli.EXPECT().GetCluster(ctx, gomock.Any(), gomock.Any()).Return(&cloudv1.Cluster{}, nil)
			mockClusterproxy.EXPECT().GetPDSafepoint(ctx).Return(&safepoint, nil)
			err := userContext.TestCdcTask(
				ctx,
				entity.ID(1),
				entity.ID(1),
				central.ConnectorType_TIDB,
				"192.168.1.1:100",
				"test",
				"root",
				"123456",
				12345,
				mockGetProxy,
			)
			So(err.Error(), ShouldEqual, FailedToGetTSO)
		})

		Convey("kafka will not check start tso", func() {
			mockClusterDao.EXPECT().GetProjectCIDRByClusterID(gomock.Any(), gomock.Any()).Return(&cidr, nil)
			mockInfraCli.EXPECT().GetCluster(ctx, gomock.Any(), gomock.Any()).Return(&cloudv1.Cluster{}, nil)
			mockClusterproxy.EXPECT().TestConnection(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil)
			err := userContext.TestCdcTask(
				ctx,
				entity.ID(1),
				entity.ID(1),
				central.ConnectorType_KAFKA,
				"192.168.1.1:100",
				"test",
				"root",
				"123456",
				12345,
				mockGetProxy,
			)
			So(err, ShouldBeNil)
		})

		Convey("should be nil when start tso is zore", func() {
			mockClusterDao.EXPECT().GetProjectCIDRByClusterID(gomock.Any(), gomock.Any()).Return(&cidr, nil)
			mockInfraCli.EXPECT().GetCluster(ctx, gomock.Any(), gomock.Any()).Return(&cloudv1.Cluster{}, nil)
			mockClusterproxy.EXPECT().TestConnection(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil)
			err := userContext.TestCdcTask(
				ctx,
				entity.ID(1),
				entity.ID(1),
				central.ConnectorType_TIDB,
				"192.168.1.1:100",
				"test",
				"root",
				"123456",
				0,
				mockGetProxy,
			)
			So(err, ShouldBeNil)
		})

		Convey("should be nil when start tso is correct", func() {
			mockClusterDao.EXPECT().GetProjectCIDRByClusterID(gomock.Any(), gomock.Any()).Return(&cidr, nil)
			mockInfraCli.EXPECT().GetCluster(ctx, gomock.Any(), gomock.Any()).Return(&cloudv1.Cluster{}, nil)
			mockClusterproxy.EXPECT().GetPDSafepoint(ctx).Return(&safepoint, nil)
			mockClusterproxy.EXPECT().TestConnection(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil)
			err := userContext.TestCdcTask(
				ctx,
				entity.ID(1),
				entity.ID(1),
				central.ConnectorType_TIDB,
				"192.168.1.1:100",
				"test",
				"root",
				"123456",
				123456789,
				mockGetProxy,
			)
			So(err, ShouldBeNil)
		})
	})
}

func TestAssembleSinkUrl(t *testing.T) {
	Convey("Test_assembleSinkUrl", t, func() {
		ctrl := gomock.NewController(t)
		defer ctrl.Finish()
		logger := log.L()
		mockCdcDaoCli := mock_cdcdao.NewMockCDCDaoFacade(ctrl)
		mockEncryptorCli := mock_encrypt.NewMockEncryptor(ctrl)
		model := Model{
			log:       logger,
			dao:       mockCdcDaoCli,
			encryptor: mockEncryptorCli,
		}
		userContext := UserContext{
			Model: model,
			log:   logger,
			tenantAndUser: &tenant.User{
				TenantID: entity.ID(111),
			},
		}

		cdcTaskKafka := &entity.CDCTask{
			DownStreamType: entity.CDCDownStreamTypeKafka,
		}

		cdcTaskMysql := &entity.CDCTask{
			DownStreamType: entity.CDCDownStreamTypeMysql,
		}

		cdcTaskTiDB := &entity.CDCTask{
			DownStreamType: entity.CDCDownStreamTypeTiDB,
		}

		kafkaTask := &entity.CDCKafkaTask{
			DownStreamEndpoint: "a.com",
			DownStreamPort:     "200",
			Topic:              "test",
			Version:            "v2.0",
			MaxMessageBytes:    "123",
			Protocol:           "abc",
		}

		mysqlTask := &entity.CDCSQLTask{
			User:               "root",
			Password:           "",
			DownStreamEndpoint: "a.com",
			DownStreamPort:     "200",
			WorkCount:          2,
			MaxTxnRow:          30,
		}

		kafkaSink := "kafka://a.com:200/test?kafka-version=v2.0&max-message-bytes=123&protocol=abc"
		mysqlSink := "mysql://root:@a.com:200?worker-count=2&max-txn-row=30"
		tidbSink := "tidb://root:@a.com:200?worker-count=2&max-txn-row=30"
		Convey("kafkasink should be nil", func() {
			mockCdcDaoCli.EXPECT().GetCdcKafkaTask(gomock.Any()).Return(kafkaTask, nil)
			ks, err := BgContext{userContext}.assembleSinkUrl(cdcTaskKafka)
			So(err, ShouldBeNil)
			So(ks, ShouldEqual, kafkaSink)
		})

		Convey("mysql should be nil", func() {
			mockCdcDaoCli.EXPECT().GetCdcSqlTask(gomock.Any()).Return(mysqlTask, nil)
			mockEncryptorCli.EXPECT().Decrypt(gomock.Any()).Return(encrypt.Secret(""), nil)
			ms, err := BgContext{userContext}.assembleSinkUrl(cdcTaskMysql)
			So(err, ShouldBeNil)
			So(ms, ShouldEqual, mysqlSink)
		})

		Convey("tidb should be nil", func() {
			mockCdcDaoCli.EXPECT().GetCdcSqlTask(gomock.Any()).Return(mysqlTask, nil)
			mockEncryptorCli.EXPECT().Decrypt(gomock.Any()).Return(encrypt.Secret(""), nil)
			ms, err := BgContext{userContext}.assembleSinkUrl(cdcTaskTiDB)
			So(err, ShouldBeNil)
			So(ms, ShouldEqual, tidbSink)
		})
	})
}

func TestGetTCU(t *testing.T) {
	assert := require.New(t)
	err := config.NormalizeConfigDefaults(config.Global())
	assert.NoError(err)
	cluster := entity.Cluster{}
	assert.EqualValues(getTCU(cluster), 0)
	cluster.Replicas = 1
	cluster.ComponentsV2 = []*entity.ClusterComponent{
		{
			ComponentType: entity.TiKV,
			Replicas:      3,
			InternalProfile: &entity.InternalProfile{
				InfraCompute: &entity.InfraCompute{
					CPUNum: 8,
				},
			},
		},
	}
	assert.EqualValues(getTCU(cluster), 16)
	cluster.ComponentsV2[0].Replicas = 6
	assert.EqualValues(getTCU(cluster), 24)
	cluster.ComponentsV2[0].Replicas = 15
	assert.EqualValues(getTCU(cluster), 32)

	cluster.TCUNumber = 4
	assert.EqualValues(getTCU(cluster), 4)
}

func TestGetClusterTCU(t *testing.T) {
	assert := require.New(t)
	err := config.NormalizeConfigDefaults(config.Global())
	assert.NoError(err)
	ctrl := gomock.NewController(t)
	mockClusterDaoCli := mock_clusterdao.NewMockClusterDaoFacade(ctrl)
	mockCdcDaoCli := mock_cdcdao.NewMockCDCDaoFacade(ctrl)

	logger := log.L()
	billingModel := billing.NewModel(logger, nil, nil)
	model := Model{
		log:          logger,
		clusterdao:   mockClusterDaoCli,
		dao:          mockCdcDaoCli,
		billingModel: billingModel,
	}

	userContext := UserContext{
		Model: model,
		log:   logger,
		tenantAndUser: &tenant.User{
			TenantID: entity.ID(111),
		},
	}

	cluster := entity.Cluster{}
	cluster.ComponentsV2 = []*entity.ClusterComponent{
		{
			ComponentType: entity.TiKV,
			Replicas:      3,
			InternalProfile: &entity.InternalProfile{
				InfraCompute: &entity.InfraCompute{
					CPUNum: 8,
				},
			},
		},
	}
	cdcList := []entity.CDCTask{
		{},
	}

	mockCdcDaoCli.EXPECT().FindCDCTasks(gomock.Any()).Return(cdcList, nil)
	mockClusterDaoCli.EXPECT().GetCluster(gomock.Any()).Return(cluster, nil)
	tcu, err := userContext.GetClusterTCU(0)
	assert.NoError(err)
	assert.EqualValues(tcu, 16)
}

type MockBillingHandler struct {
	billing2.BillingHandler
	mock.Mock
}

func (m *MockBillingHandler) MeasureCosts(tenantID uint64, cost *billing2.UnitCostsInput) error {
	m.Called(tenantID, cost)
	return nil
}

func TestMeteringMeasures(t *testing.T) {
	test.Setup()
	config.Global().BillingConfig.EnableCDCTCUMetering = true
	defer test.Shutdown()
	assert := require.New(t)
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()
	logger := log.L()
	ctx := context.Background()
	mockBilling := &MockBillingHandler{}
	billing2.SetHandler(mockBilling)
	mockClusterDaoCli := mock_clusterdao.NewMockClusterDaoFacade(ctrl)
	mockCdcDaoCli := mock_cdcdao.NewMockCDCDaoFacade(ctrl)
	billingModel := billing.NewModel(logger, nil, nil)
	model := Model{
		log:          logger,
		clusterdao:   mockClusterDaoCli,
		dao:          mockCdcDaoCli,
		billingModel: billingModel,
	}

	userContext := UserContext{
		Model: model,
		log:   logger,
		tenantAndUser: &tenant.User{
			TenantID: entity.ID(111),
		},
	}

	userContext.tenantAndUser.ID = entity.ID(222)
	clusterList := []entity.Cluster{
		{
			Updatable: entity.Updatable{
				Base: entity.Base{
					ID:        entity.ID(1),
					CreatedAt: time.Now().Add(-24 * time.Hour),
				},
			},
			ProjectID: entity.ID(2),
			Project: &entity.Project{
				Name: "default",
			},
			TCUNumber: 5,
			Status:    entity.ClusterStatusNormal,
			Provider: &entity.Provider{
				Name: "provider",
			},
			ComponentsV2: []*entity.ClusterComponent{
				{
					Region: &entity.Region{
						Name: "region",
					},
				},
			},
		},
	}
	cdcList := []entity.CDCTask{
		{},
	}
	mockCdcDaoCli.EXPECT().FindCDCTasks(gomock.Any()).Return(cdcList, nil)
	mockClusterDaoCli.EXPECT().FindClusters(gomock.Any()).Return(clusterList, nil)

	mockBilling.On("MeasureCosts", uint64(111), mock.Anything).Return(nil)
	err := db.Transaction(func(tx *gorm.DB) error {
		tenant := entity.Tenant{}
		tenant.ID = 111
		tenant.Plan = "poc"
		tenant.Status = "active"
		err := tenant.Create(tx)
		if err != nil {
			return err
		}
		trial := entity.Trial{}
		trial.StartedAt = time.Now()
		trial.TenantID = 111
		trial.TrialType = "poc"
		return trial.Create(tx)
	})
	assert.NoError(err)

	err = BgContext{userContext}.SendTicdcMesaures(ctx)
	assert.NoError(err)

	mockBilling.AssertExpectations(t)
	cost := mockBilling.Calls[0].Arguments[1].(*billing2.UnitCostsInput)
	assert.Equal(cost.Cost, 5*60)
	assert.Equal(cost.Unit, fmt.Sprintf(TCUCostUnitPattern, "provider", "region"))
	assert.Equal(cost.Labels["poc"], "true")

	err = db.Transaction(func(tx *gorm.DB) error {
		cnt, err := entity.NewCDCTCUMeteringRecordQuerySet(tx).Count()
		assert.NoError(err)
		assert.EqualValues(cnt, 1)
		record := &entity.CDCTCUMeteringRecord{}
		err = entity.NewCDCTCUMeteringRecordQuerySet(tx).One(record)
		assert.NoError(err)
		assert.Equal(record.Status, entity.CDCMeteringStatusDone)
		return nil
	})

	assert.NoError(err)
	clusterList[0].ID = 2
	cdcList[0].Status = entity.CDCStatusFailed
	mockCdcDaoCli.EXPECT().FindCDCTasks(gomock.Any()).Return(cdcList, nil)
	mockClusterDaoCli.EXPECT().FindClusters(gomock.Any()).Return(clusterList, nil)

	mockBilling.On("MeasureCosts", uint64(111), mock.Anything).Return(nil)

	err = BgContext{userContext}.SendTicdcMesaures(ctx)
	assert.NoError(err)

	err = db.Transaction(func(tx *gorm.DB) error {
		cnt, err := entity.NewCDCTCUMeteringRecordQuerySet(tx).Count()
		assert.NoError(err)
		assert.EqualValues(cnt, 2)
		return nil
	})

	assert.NoError(err)
}