package service

import (
	"context"
	"encoding/json"
	"testing"
	"time"

	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/db/mock_db"
	"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/vo"
	"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"
)

func TestCreateExportPersistsPreparingWithoutCreatingCR(t *testing.T) {
	originalConfig := config.Global()
	config.Set(&config.CentralConfig{BizConfig: config.BizConfig{
		MaxExportConcurrencyPerOrg:         100,
		SkipExportSizeCheckTenants:         []uint64{3001},
		SkipExportLocalStorageCheckTenants: []uint64{3001},
	}})
	t.Cleanup(func() { config.Set(originalConfig) })

	ctrl := gomock.NewController(t)
	exportRepo := mock_repo.NewMockExportRepo(ctrl)
	infraFacade := mock_facade.NewMockInfraFacade(ctrl)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	encryptor := mock_facade.NewMockEncryptorFacade(ctrl)
	transProvider := mock_db.NewMockTransProvider(ctrl)
	eventbusService := eventbus.NewMockEventBusService(ctrl)
	svc := &exportService{
		exportRepo:      exportRepo,
		infraFacade:     infraFacade,
		mgmtSvc:         mgmtSvc,
		encryptorFacade: encryptor,
		transProvider:   transProvider,
		eventbusService: eventbusService,
	}

	cluster := &tidbmgmtservicev1.Cluster{
		ClusterId:         1001,
		ProjectId:         2001,
		OriginalProjectId: 2001,
		OrganizationId:    3001,
		CloudProvider:     "aws",
		Region:            "us-west-2",
		DeployType:        tidbmgmtservicev1.DeployType_DEPLOY_TYPE_SERVERLESS,
		ClusterInfo: &tidbmgmtservicev1.Cluster_ServerlessCluster{
			ServerlessCluster: &globalserverlessv1.ServerlessServiceCluster{
				UserStatus: globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_NORMAL,
				Cluster: &regionalserverlessv1.ServerlessRegionalServiceClusterInfo{
					Kubernetes:    "test-k8s",
					ProxyEndpoint: "https://proxy.example",
				},
			},
		},
	}
	mgmtSvc.EXPECT().
		GetClusterIncludeDedicated(gomock.Any(), uint64(1001), tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC).
		Return(cluster, nil)
	encryptor.EXPECT().EncryptBase64(gomock.Any()).Return("encrypted-data-key", nil)
	exportRepo.EXPECT().ListByFilter(gomock.Any(), entity.ExportFilter{
		TenantID: "3001",
		States: []constant.ExportState{
			constant.ExportStatePreparing,
			constant.ExportStateRunning,
		},
	}).Return(nil, nil)
	mgmtSvc.EXPECT().SetClusterExportStatus(gomock.Any(), uint64(1001), gomock.Any()).Return(uint64(88), nil)
	transProvider.EXPECT().Transaction(gomock.Any(), gomock.Any()).DoAndReturn(
		func(ctx context.Context, callback func(context.Context) error) error {
			return callback(ctx)
		},
	)
	exportRepo.EXPECT().Create(gomock.Any(), gomock.Any()).DoAndReturn(
		func(_ context.Context, export *entity.Export) bizerr.BizError {
			require.Equal(t, constant.ExportStatePreparing, export.State)
			var envelope exportCreatePayloadEnvelope
			require.NoError(t, json.Unmarshal([]byte(export.CreatePayload), &envelope))
			require.Equal(t, "encrypted-data-key", envelope.EncryptedDataKey)
			require.NotEmpty(t, envelope.Nonce)
			require.NotEmpty(t, envelope.Ciphertext)
			require.Equal(t, "https://proxy.example", export.ProxyEndpoint)
			return nil
		},
	)
	published := make(chan *tieventbus.Event, 1)
	eventbusService.EXPECT().Publish(gomock.Any()).DoAndReturn(func(event *tieventbus.Event) error {
		published <- event
		return nil
	})

	result, err := svc.CreateExport(context.Background(), &vo.Export{
		ClusterID:  "1001",
		TargetType: constant.ExportTargetTypeLocal,
	})
	require.NoError(t, err)
	require.Equal(t, constant.ExportStatePreparing, result.State)
	waitForExportEvent(t, published)
}

func waitForExportEvent(t *testing.T, events <-chan *tieventbus.Event) {
	t.Helper()
	select {
	case <-events:
	case <-time.After(time.Second):
		t.Fatal("timed out waiting for export event")
	}
}