package export_svc

import (
	"context"
	"errors"
	"sync/atomic"
	"testing"
	"time"

	"github.com/smartystreets/goconvey/convey"
	"github.com/stretchr/testify/require"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/serverless-service/internal/config"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	"github.com/tidbcloud/serverless-service/internal/domain/facade/mock_facade"
	"github.com/tidbcloud/serverless-service/internal/domain/repo/mock_repo"
	"github.com/tidbcloud/serverless-service/internal/domain/service"
	"github.com/tidbcloud/serverless-service/internal/domain/service/mock_service"
	"github.com/tidbcloud/serverless-service/internal/errorcode"
	"github.com/tidbcloud/serverless-service/internal/infra/eventbus"
	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	regionalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/regional/serverless/v1"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"github.com/tidbcloud/tiinf/transport/tieventbus"
	"go.uber.org/mock/gomock"
	"go.uber.org/multierr"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/runtime/schema"
)

func TestExportSyncerService_Run(t *testing.T) {
	convey.Convey("ExportSyncerService.Run", t, func() {
		convey.Convey("completed exports use persisted proxy endpoints", func() {
			ctrl := gomock.NewController(t)
			defer ctrl.Finish()

			exportRepo := mock_repo.NewMockExportRepo(ctrl)
			infraFacade := mock_facade.NewMockInfraFacade(ctrl)
			mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
			eventbusService := eventbus.NewMockEventBusService(ctrl)

			svc := NewExportSyncerService(exportRepo, infraFacade, mgmtSvc, nil, eventbusService)
			ctx := context.Background()
			cluster := buildTestCluster(1001, "https://proxy.example")
			exports := []*entity.Export{
				buildRunningExport("exp-1", "1001", "2001", "https://proxy.example"),
				buildRunningExport("exp-2", "1001", "2001", "https://proxy.example"),
			}

			exportRepo.EXPECT().
				ListByFilter(gomock.Any(), activeExportFilter()).
				Return(exports, nil)
			mgmtSvc.EXPECT().
				GetCluster(gomock.Any(), uint64(1001), tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC).
				Return(cluster, nil).
				Times(2)

			for _, export := range exports {
				infraFacade.EXPECT().
					GetExport(gomock.Any(), "https://proxy.example", uint64(2001), export.ID).
					Return(&cloudv1.Export{
						Status: cloudv1.ExportStatus{Phase: cloudv1.ExportPhaseCompleted},
					}, nil)
			}
			exportRepo.EXPECT().
				UpdateRunningState(gomock.Any(), gomock.AssignableToTypeOf(&entity.Export{})).
				DoAndReturn(
					func(_ context.Context, got *entity.Export) (*entity.Export, bizerr.BizError) {
						require.Equal(t, constant.ExportStateSucceed, got.State)
						return got, nil
					},
				).
				Times(len(exports))
			expectListTerminalExports(t, exportRepo)

			published := make(chan *tieventbus.Event, len(exports))
			eventbusService.EXPECT().Publish(gomock.AssignableToTypeOf(&tieventbus.Event{})).DoAndReturn(
				func(evt *tieventbus.Event) error {
					published <- evt
					return nil
				},
			).Times(len(exports))

			err := svc.Run(ctx)
			convey.So(err, convey.ShouldBeNil)
			waitPublishedEvents(t, published, len(exports))
		})

		convey.Convey("failed export transition copies reason", func() {
			ctrl := gomock.NewController(t)
			defer ctrl.Finish()

			exportRepo := mock_repo.NewMockExportRepo(ctrl)
			infraFacade := mock_facade.NewMockInfraFacade(ctrl)
			mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
			eventbusService := eventbus.NewMockEventBusService(ctrl)

			svc := NewExportSyncerService(exportRepo, infraFacade, mgmtSvc, nil, eventbusService)
			ctx := context.Background()
			export := buildRunningExport("exp-failed", "1002", "2002", "https://proxy-2.example")
			cluster := buildTestCluster(1002, "https://proxy-2.example")

			exportRepo.EXPECT().
				ListByFilter(gomock.Any(), activeExportFilter()).
				Return([]*entity.Export{export}, nil)
			mgmtSvc.EXPECT().
				GetCluster(gomock.Any(), uint64(1002), tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC).
				Return(cluster, nil)
			infraFacade.EXPECT().
				GetExport(gomock.Any(), "https://proxy-2.example", uint64(2002), "exp-failed").
				Return(&cloudv1.Export{
					Status: cloudv1.ExportStatus{
						Phase:   cloudv1.ExportPhaseFailed,
						Message: "infra failed",
					},
				}, nil)
			exportRepo.EXPECT().
				UpdateRunningState(gomock.Any(), gomock.AssignableToTypeOf(&entity.Export{})).
				DoAndReturn(
					func(_ context.Context, got *entity.Export) (*entity.Export, bizerr.BizError) {
						require.Equal(t, constant.ExportStateFailed, got.State)
						require.Equal(t, "infra failed", got.Reason)
						return got, nil
					},
				)
			expectListTerminalExports(t, exportRepo)

			published := make(chan *tieventbus.Event, 1)
			eventbusService.EXPECT().Publish(gomock.AssignableToTypeOf(&tieventbus.Event{})).DoAndReturn(
				func(evt *tieventbus.Event) error {
					published <- evt
					return nil
				},
			)

			err := svc.Run(ctx)
			convey.So(err, convey.ShouldBeNil)
			waitPublishedEvents(t, published, 1)
		})

		convey.Convey("missing export CR marks export failed with deterministic reason", func() {
			ctrl := gomock.NewController(t)
			defer ctrl.Finish()

			exportRepo := mock_repo.NewMockExportRepo(ctrl)
			infraFacade := mock_facade.NewMockInfraFacade(ctrl)
			mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
			eventbusService := eventbus.NewMockEventBusService(ctrl)

			svc := NewExportSyncerService(exportRepo, infraFacade, mgmtSvc, nil, eventbusService)
			ctx := context.Background()
			export := buildRunningExport("exp-missing", "1003", "2003", "https://proxy-3.example")
			cluster := buildTestCluster(1003, "https://proxy-3.example")

			exportRepo.EXPECT().
				ListByFilter(gomock.Any(), activeExportFilter()).
				Return([]*entity.Export{export}, nil)
			mgmtSvc.EXPECT().
				GetCluster(gomock.Any(), uint64(1003), tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC).
				Return(cluster, nil)
			infraFacade.EXPECT().
				GetExport(gomock.Any(), "https://proxy-3.example", uint64(2003), "exp-missing").
				Return(nil, apierrors.NewNotFound(schema.GroupResource{Group: "infra", Resource: "exports"}, "exp-missing"))
			exportRepo.EXPECT().
				UpdateRunningState(gomock.Any(), gomock.AssignableToTypeOf(&entity.Export{})).
				DoAndReturn(
					func(_ context.Context, got *entity.Export) (*entity.Export, bizerr.BizError) {
						require.Equal(t, constant.ExportStateFailed, got.State)
						require.Equal(t, exportCRNotFoundReason, got.Reason)
						return got, nil
					},
				)
			expectListTerminalExports(t, exportRepo)

			published := make(chan *tieventbus.Event, 1)
			eventbusService.EXPECT().Publish(gomock.AssignableToTypeOf(&tieventbus.Event{})).DoAndReturn(
				func(evt *tieventbus.Event) error {
					published <- evt
					return nil
				},
			)

			err := svc.Run(ctx)
			convey.So(err, convey.ShouldBeNil)
			waitPublishedEvents(t, published, 1)
		})

		convey.Convey("no rows update is treated as idempotent no-op", func() {
			ctrl := gomock.NewController(t)
			defer ctrl.Finish()

			exportRepo := mock_repo.NewMockExportRepo(ctrl)
			infraFacade := mock_facade.NewMockInfraFacade(ctrl)
			mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
			eventbusService := eventbus.NewMockEventBusService(ctrl)

			svc := NewExportSyncerService(exportRepo, infraFacade, mgmtSvc, nil, eventbusService)
			ctx := context.Background()
			export := buildRunningExport("exp-idempotent", "1004", "2004", "https://proxy-4.example")
			cluster := buildTestCluster(1004, "https://proxy-4.example")

			exportRepo.EXPECT().
				ListByFilter(gomock.Any(), activeExportFilter()).
				Return([]*entity.Export{export}, nil)
			mgmtSvc.EXPECT().
				GetCluster(gomock.Any(), uint64(1004), tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC).
				Return(cluster, nil)
			infraFacade.EXPECT().
				GetExport(gomock.Any(), "https://proxy-4.example", uint64(2004), "exp-idempotent").
				Return(&cloudv1.Export{
					Status: cloudv1.ExportStatus{Phase: cloudv1.ExportPhaseCompleted},
				}, nil)
			exportRepo.EXPECT().
				UpdateRunningState(gomock.Any(), gomock.AssignableToTypeOf(&entity.Export{})).
				DoAndReturn(
					func(_ context.Context, got *entity.Export) (*entity.Export, bizerr.BizError) {
						require.Equal(t, constant.ExportStateSucceed, got.State)
						return nil, bizerr.New(
							errorcode.ErrCodeExportUpdateStateAffectedNoRows,
							"fail to update status for rowsAffected is 0",
						)
					},
				)
			expectListTerminalExports(t, exportRepo)

			err := svc.Run(ctx)
			convey.So(err, convey.ShouldBeNil)
		})

		convey.Convey("preparing export creates CR then transitions to running", func() {
			ctrl := gomock.NewController(t)
			defer ctrl.Finish()

			exportRepo := mock_repo.NewMockExportRepo(ctrl)
			infraFacade := mock_facade.NewMockInfraFacade(ctrl)
			mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
			exportSvc := mock_service.NewMockExportService(ctrl)
			eventbusService := eventbus.NewMockEventBusService(ctrl)
			svc := NewExportSyncerService(exportRepo, infraFacade, mgmtSvc, exportSvc, eventbusService)
			export := buildPreparingExport("exp-preparing", "1006", "2006", "https://proxy-6.example")
			cluster := buildTestCluster(1006, "https://proxy-6.example")
			exportCR := &cloudv1.Export{}

			exportRepo.EXPECT().ListByFilter(gomock.Any(), activeExportFilter()).Return([]*entity.Export{export}, nil)
			exportRepo.EXPECT().GetById(gomock.Any(), "1006", "exp-preparing").Return(export, nil).Times(2)
			mgmtSvc.EXPECT().
				GetCluster(gomock.Any(), uint64(1006), tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC).
				Return(cluster, nil)
			exportSvc.EXPECT().NewInfraAPIExportObject(gomock.Any(), export, cluster).Return(exportCR, nil)
			infraFacade.EXPECT().CreateExport(gomock.Any(), "https://proxy-6.example", exportCR).Return(nil)
			exportRepo.EXPECT().TransitionPreparingState(gomock.Any(), export).DoAndReturn(
				func(_ context.Context, got *entity.Export) (bool, bizerr.BizError) {
					require.Equal(t, constant.ExportStateRunning, got.State)
					return true, nil
				},
			)
			expectListTerminalExports(t, exportRepo)

			convey.So(svc.Run(context.Background()), convey.ShouldBeNil)
		})

		convey.Convey("temporary CR creation error leaves export preparing", func() {
			ctrl := gomock.NewController(t)
			defer ctrl.Finish()

			exportRepo := mock_repo.NewMockExportRepo(ctrl)
			infraFacade := mock_facade.NewMockInfraFacade(ctrl)
			mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
			exportSvc := mock_service.NewMockExportService(ctrl)
			eventbusService := eventbus.NewMockEventBusService(ctrl)
			svc := NewExportSyncerService(exportRepo, infraFacade, mgmtSvc, exportSvc, eventbusService)
			export := buildPreparingExport("exp-retry", "1007", "2007", "https://proxy-7.example")
			cluster := buildTestCluster(1007, "https://proxy-7.example")
			exportCR := &cloudv1.Export{}
			createErr := errors.New("infra unavailable")

			exportRepo.EXPECT().ListByFilter(gomock.Any(), activeExportFilter()).Return([]*entity.Export{export}, nil)
			exportRepo.EXPECT().GetById(gomock.Any(), "1007", "exp-retry").Return(export, nil).Times(2)
			mgmtSvc.EXPECT().
				GetCluster(gomock.Any(), uint64(1007), tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC).
				Return(cluster, nil)
			exportSvc.EXPECT().NewInfraAPIExportObject(gomock.Any(), export, cluster).Return(exportCR, nil)
			infraFacade.EXPECT().CreateExport(gomock.Any(), "https://proxy-7.example", exportCR).Return(createErr)
			expectListTerminalExports(t, exportRepo)

			convey.So(svc.Run(context.Background()), convey.ShouldEqual, createErr)
			convey.So(export.State, convey.ShouldEqual, constant.ExportStatePreparing)
		})

		convey.Convey("running export with persisted endpoint skips cluster lookup", func() {
			ctrl := gomock.NewController(t)
			defer ctrl.Finish()

			exportRepo := mock_repo.NewMockExportRepo(ctrl)
			infraFacade := mock_facade.NewMockInfraFacade(ctrl)
			mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
			svc := NewExportSyncerService(exportRepo, infraFacade, mgmtSvc, nil, nil)
			export := buildRunningExport("exp-running", "1010", "2010", "https://proxy-10.example")

			exportRepo.EXPECT().ListByFilter(gomock.Any(), activeExportFilter()).Return([]*entity.Export{export}, nil)
			infraFacade.EXPECT().
				GetExport(gomock.Any(), "https://proxy-10.example", uint64(2010), "exp-running").
				Return(&cloudv1.Export{}, nil)
			expectListTerminalExports(t, exportRepo)

			convey.So(svc.Run(context.Background()), convey.ShouldBeNil)
		})

		convey.Convey("running export without endpoint resolves and backfills it", func() {
			ctrl := gomock.NewController(t)
			defer ctrl.Finish()

			exportRepo := mock_repo.NewMockExportRepo(ctrl)
			infraFacade := mock_facade.NewMockInfraFacade(ctrl)
			mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
			svc := NewExportSyncerService(exportRepo, infraFacade, mgmtSvc, nil, nil)
			export := buildRunningExport("exp-backfill", "1011", "2011", "")
			cluster := buildTestCluster(1011, "https://proxy-11.example")

			exportRepo.EXPECT().ListByFilter(gomock.Any(), activeExportFilter()).Return([]*entity.Export{export}, nil)
			mgmtSvc.EXPECT().
				GetCluster(gomock.Any(), uint64(1011), tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC).
				Return(cluster, nil)
			exportRepo.EXPECT().
				UpdateProxyEndpoint(gomock.Any(), "exp-backfill", "1011", "https://proxy-11.example").
				Return(nil)
			infraFacade.EXPECT().
				GetExport(gomock.Any(), "https://proxy-11.example", uint64(2011), "exp-backfill").
				Return(&cloudv1.Export{}, nil)
			expectListTerminalExports(t, exportRepo)

			convey.So(svc.Run(context.Background()), convey.ShouldBeNil)
			convey.So(export.ProxyEndpoint, convey.ShouldEqual, "https://proxy-11.example")
		})

		convey.Convey("terminal exports with cluster status id are cleaned up idempotently", func() {
			ctrl := gomock.NewController(t)
			defer ctrl.Finish()

			exportRepo := mock_repo.NewMockExportRepo(ctrl)
			infraFacade := mock_facade.NewMockInfraFacade(ctrl)
			mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
			eventbusService := eventbus.NewMockEventBusService(ctrl)

			svc := NewExportSyncerService(exportRepo, infraFacade, mgmtSvc, nil, eventbusService)
			ctx := context.Background()
			exp := &entity.Export{
				ID:              "exp-terminal",
				ClusterID:       "1005",
				ProjectID:       "2005",
				State:           constant.ExportStateCanceled,
				ClusterStatusID: 88,
			}

			exportRepo.EXPECT().
				ListByFilter(gomock.Any(), activeExportFilter()).
				Return([]*entity.Export{}, nil)
			exportRepo.EXPECT().
				ListByFilter(gomock.Any(), gomock.AssignableToTypeOf(entity.ExportFilter{})).
				DoAndReturn(func(_ context.Context, filter entity.ExportFilter) ([]*entity.Export, bizerr.BizError) {
					require.Equal(t, constant.ExportStateSucceed, filter.State)
					require.NotNil(t, filter.HasClusterStatus)
					require.True(t, *filter.HasClusterStatus)
					require.True(t, filter.IncludeDeleted)
					return []*entity.Export{}, nil
				})
			exportRepo.EXPECT().
				ListByFilter(gomock.Any(), gomock.AssignableToTypeOf(entity.ExportFilter{})).
				DoAndReturn(func(_ context.Context, filter entity.ExportFilter) ([]*entity.Export, bizerr.BizError) {
					require.Equal(t, constant.ExportStateFailed, filter.State)
					require.NotNil(t, filter.HasClusterStatus)
					require.True(t, *filter.HasClusterStatus)
					require.True(t, filter.IncludeDeleted)
					return []*entity.Export{}, nil
				})
			exportRepo.EXPECT().
				ListByFilter(gomock.Any(), gomock.AssignableToTypeOf(entity.ExportFilter{})).
				DoAndReturn(func(_ context.Context, filter entity.ExportFilter) ([]*entity.Export, bizerr.BizError) {
					require.Equal(t, constant.ExportStateCanceled, filter.State)
					require.NotNil(t, filter.HasClusterStatus)
					require.True(t, *filter.HasClusterStatus)
					require.True(t, filter.IncludeDeleted)
					return []*entity.Export{exp}, nil
				})
			exportRepo.EXPECT().
				ListByFilter(gomock.Any(), gomock.AssignableToTypeOf(entity.ExportFilter{})).
				DoAndReturn(func(_ context.Context, filter entity.ExportFilter) ([]*entity.Export, bizerr.BizError) {
					require.Equal(t, constant.ExportStateDeleted, filter.State)
					require.NotNil(t, filter.HasClusterStatus)
					require.True(t, *filter.HasClusterStatus)
					require.True(t, filter.IncludeDeleted)
					return []*entity.Export{}, nil
				})
			mgmtSvc.EXPECT().
				DeleteClusterCustomStatus(gomock.Any(), uint64(1005), uint64(88)).
				Return(nil)
			exportRepo.EXPECT().
				ClearClusterStatusID(gomock.Any(), "exp-terminal", "1005", uint64(88)).
				Return(nil)

			err := svc.Run(ctx)
			convey.So(err, convey.ShouldBeNil)
			convey.So(exp.ClusterStatusID, convey.ShouldEqual, 0)
		})
	})
}

func TestExportSyncerService_RunSharesConcurrencyLimitAcrossStates(t *testing.T) {
	originalConcurrency := config.Global().BizConfig.ExportSyncerConcurrency
	config.Global().BizConfig.ExportSyncerConcurrency = 2
	t.Cleanup(func() {
		config.Global().BizConfig.ExportSyncerConcurrency = originalConcurrency
	})

	ctrl := gomock.NewController(t)
	exportRepo := mock_repo.NewMockExportRepo(ctrl)
	infraFacade := mock_facade.NewMockInfraFacade(ctrl)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	exportSvc := mock_service.NewMockExportService(ctrl)
	svc := NewExportSyncerService(exportRepo, infraFacade, mgmtSvc, exportSvc, nil)

	exports := []*entity.Export{
		buildPreparingExport("exp-preparing-1", "1101", "2101", ""),
		buildRunningExport("exp-running-1", "1102", "2102", ""),
		buildPreparingExport("exp-preparing-2", "1103", "2103", ""),
		buildRunningExport("exp-running-2", "1104", "2104", ""),
	}
	exportsByID := make(map[string]*entity.Export, len(exports))
	for _, export := range exports {
		exportsByID[export.ID] = export
	}

	exportRepo.EXPECT().ListByFilter(gomock.Any(), activeExportFilter()).Return(exports, nil)
	exportRepo.EXPECT().GetById(gomock.Any(), gomock.Any(), gomock.Any()).DoAndReturn(
		func(_ context.Context, _, exportID string) (*entity.Export, bizerr.BizError) {
			return exportsByID[exportID], nil
		},
	).Times(2)
	expectListTerminalExports(t, exportRepo)

	started := make(chan struct{}, len(exports))
	release := make(chan struct{})
	var inFlight atomic.Int32
	var maxInFlight atomic.Int32
	syncErr := bizerr.New(bizerr.ErrCodeUnknown, "sync failed")
	mgmtSvc.EXPECT().
		GetCluster(gomock.Any(), gomock.Any(), tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC).
		DoAndReturn(func(
			_ context.Context,
			_ uint64,
			_ tidbmgmtservicev1.ClusterFilter_ClusterView,
		) (*tidbmgmtservicev1.Cluster, bizerr.BizError) {
			current := inFlight.Add(1)
			for {
				observedMax := maxInFlight.Load()
				if current <= observedMax || maxInFlight.CompareAndSwap(observedMax, current) {
					break
				}
			}
			started <- struct{}{}
			<-release
			inFlight.Add(-1)
			return nil, syncErr
		}).
		Times(len(exports))

	done := make(chan error, 1)
	go func() {
		done <- svc.Run(context.Background())
	}()

	for range 2 {
		select {
		case <-started:
		case <-time.After(time.Second):
			t.Fatal("timed out waiting for initial sync tasks")
		}
	}
	select {
	case <-started:
		t.Fatal("more than two export sync tasks started concurrently")
	case <-time.After(100 * time.Millisecond):
	}

	release <- struct{}{}
	select {
	case <-started:
	case <-time.After(time.Second):
		t.Fatal("timed out waiting for the next sync task")
	}
	close(release)

	runErr := <-done
	require.ErrorContains(t, runErr, "sync failed")
	require.Len(t, multierr.Errors(runErr), len(exports))
	require.EqualValues(t, 2, maxInFlight.Load())
}

func TestExportSyncerService_CreatePreparingExportCancelRace(t *testing.T) {
	ctrl := gomock.NewController(t)
	exportRepo := mock_repo.NewMockExportRepo(ctrl)
	infraFacade := mock_facade.NewMockInfraFacade(ctrl)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	exportSvc := mock_service.NewMockExportService(ctrl)
	svc := NewExportSyncerService(exportRepo, infraFacade, mgmtSvc, exportSvc, nil)

	export := buildPreparingExport("exp-race", "1008", "2008", "https://proxy-8.example")
	canceled := *export
	canceled.State = constant.ExportStateCanceled
	cluster := buildTestCluster(1008, "https://proxy-8.example")
	exportCR := &cloudv1.Export{}

	exportRepo.EXPECT().GetById(gomock.Any(), "1008", "exp-race").Return(export, nil).Times(2)
	mgmtSvc.EXPECT().
		GetCluster(gomock.Any(), uint64(1008), tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC).
		Return(cluster, nil)
	exportSvc.EXPECT().NewInfraAPIExportObject(gomock.Any(), export, cluster).Return(exportCR, nil)
	infraFacade.EXPECT().CreateExport(gomock.Any(), "https://proxy-8.example", exportCR).Return(nil)
	exportRepo.EXPECT().TransitionPreparingState(gomock.Any(), export).Return(false, nil)
	exportRepo.EXPECT().GetById(gomock.Any(), "1008", "exp-race").Return(&canceled, nil)
	infraFacade.EXPECT().DeleteExport(gomock.Any(), "https://proxy-8.example", uint64(2008), "exp-race").Return(nil)

	require.NoError(t, svc.createPreparingExport(context.Background(), export))
}

func TestExportSyncerService_InvalidPayloadFailsPreparingExport(t *testing.T) {
	ctrl := gomock.NewController(t)
	exportRepo := mock_repo.NewMockExportRepo(ctrl)
	infraFacade := mock_facade.NewMockInfraFacade(ctrl)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	exportSvc := mock_service.NewMockExportService(ctrl)
	eventbusService := eventbus.NewMockEventBusService(ctrl)
	svc := NewExportSyncerService(exportRepo, infraFacade, mgmtSvc, exportSvc, eventbusService)

	export := buildPreparingExport("exp-invalid", "1009", "2009", "https://proxy-9.example")
	cluster := buildTestCluster(1009, "https://proxy-9.example")
	exportRepo.EXPECT().GetById(gomock.Any(), "1009", "exp-invalid").Return(export, nil)
	mgmtSvc.EXPECT().
		GetCluster(gomock.Any(), uint64(1009), tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC).
		Return(cluster, nil)
	exportSvc.EXPECT().
		NewInfraAPIExportObject(gomock.Any(), export, cluster).
		Return(nil, bizerr.New(bizerr.ErrCodeParams, service.InvalidExportCreatePayloadMessage))
	exportRepo.EXPECT().TransitionPreparingState(gomock.Any(), export).DoAndReturn(
		func(_ context.Context, got *entity.Export) (bool, bizerr.BizError) {
			require.Equal(t, constant.ExportStateFailed, got.State)
			require.Equal(t, service.InvalidExportCreatePayloadMessage, got.Reason)
			return true, nil
		},
	)
	published := make(chan *tieventbus.Event, 1)
	eventbusService.EXPECT().Publish(gomock.Any()).DoAndReturn(func(evt *tieventbus.Event) error {
		published <- evt
		return nil
	})

	require.NoError(t, svc.createPreparingExport(context.Background(), export))
	waitPublishedEvents(t, published, 1)
}

func expectListTerminalExports(t *testing.T, exportRepo *mock_repo.MockExportRepo) {
	t.Helper()
	exportRepo.EXPECT().
		ListByFilter(gomock.Any(), gomock.AssignableToTypeOf(entity.ExportFilter{})).
		DoAndReturn(func(_ context.Context, filter entity.ExportFilter) ([]*entity.Export, bizerr.BizError) {
			require.Equal(t, constant.ExportStateSucceed, filter.State)
			require.NotNil(t, filter.HasClusterStatus)
			require.True(t, *filter.HasClusterStatus)
			require.True(t, filter.IncludeDeleted)
			return []*entity.Export{}, nil
		})
	exportRepo.EXPECT().
		ListByFilter(gomock.Any(), gomock.AssignableToTypeOf(entity.ExportFilter{})).
		DoAndReturn(func(_ context.Context, filter entity.ExportFilter) ([]*entity.Export, bizerr.BizError) {
			require.Equal(t, constant.ExportStateFailed, filter.State)
			require.NotNil(t, filter.HasClusterStatus)
			require.True(t, *filter.HasClusterStatus)
			require.True(t, filter.IncludeDeleted)
			return []*entity.Export{}, nil
		})
	exportRepo.EXPECT().
		ListByFilter(gomock.Any(), gomock.AssignableToTypeOf(entity.ExportFilter{})).
		DoAndReturn(func(_ context.Context, filter entity.ExportFilter) ([]*entity.Export, bizerr.BizError) {
			require.Equal(t, constant.ExportStateCanceled, filter.State)
			require.NotNil(t, filter.HasClusterStatus)
			require.True(t, *filter.HasClusterStatus)
			require.True(t, filter.IncludeDeleted)
			return []*entity.Export{}, nil
		})
	exportRepo.EXPECT().
		ListByFilter(gomock.Any(), gomock.AssignableToTypeOf(entity.ExportFilter{})).
		DoAndReturn(func(_ context.Context, filter entity.ExportFilter) ([]*entity.Export, bizerr.BizError) {
			require.Equal(t, constant.ExportStateDeleted, filter.State)
			require.NotNil(t, filter.HasClusterStatus)
			require.True(t, *filter.HasClusterStatus)
			require.True(t, filter.IncludeDeleted)
			return []*entity.Export{}, nil
		})
}

func buildRunningExport(exportID, clusterID, projectID, proxyEndpoint string) *entity.Export {
	return &entity.Export{
		ID:            exportID,
		ClusterID:     clusterID,
		ProjectID:     projectID,
		ProxyEndpoint: proxyEndpoint,
		TenantID:      "3001",
		CreatedBy:     "tester@example.com",
		State:         constant.ExportStateRunning,
	}
}

func buildPreparingExport(exportID, clusterID, projectID, proxyEndpoint string) *entity.Export {
	export := buildRunningExport(exportID, clusterID, projectID, proxyEndpoint)
	export.State = constant.ExportStatePreparing
	export.CreatePayload = "encrypted"
	return export
}

func activeExportFilter() entity.ExportFilter {
	return entity.ExportFilter{States: []constant.ExportState{
		constant.ExportStatePreparing,
		constant.ExportStateRunning,
	}}
}

func buildTestCluster(clusterID uint64, proxyEndpoint string) *tidbmgmtservicev1.Cluster {
	return &tidbmgmtservicev1.Cluster{
		ClusterId:      clusterID,
		ProjectId:      4001,
		OrganizationId: 5001,
		ClusterInfo: &tidbmgmtservicev1.Cluster_ServerlessCluster{
			ServerlessCluster: &globalserverlessv1.ServerlessServiceCluster{
				Cluster: &regionalserverlessv1.ServerlessRegionalServiceClusterInfo{
					ProxyEndpoint: proxyEndpoint,
				},
			},
		},
	}
}

func waitPublishedEvents(t *testing.T, published <-chan *tieventbus.Event, count int) {
	t.Helper()
	deadline := time.After(2 * time.Second)
	for i := 0; i < count; i++ {
		select {
		case <-published:
		case <-deadline:
			t.Fatalf("timed out waiting for %d published events", count)
		}
	}
}