package action

import (
	"context"
	"testing"
	"time"

	"github.com/stretchr/testify/require"
	clusterngv1beta "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"go.uber.org/mock/gomock"
	"google.golang.org/protobuf/encoding/protojson"
	"google.golang.org/protobuf/proto"

	globalcommonv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/common/v1"
	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	mockdataflow "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/adapter/dataflow/mock_dataflow"
	mockregionalserver "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/adapter/regionalserver/mock_regionalserver"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/entity"
	mockbrrepo "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/br/mock_br"
	mockcluster "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/cluster/mock_cluster"
	mockregion "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/region/mock_region"
	mockresourcepool "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/repository/resourcepool/mock_resourcepool"
	workflowcommon "github.com/tidbcloud/tidb-management-service/nextgen-global-service/domain/service/workflow/common"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/infra/repoimpl/schema"
	"github.com/tidbcloud/tidb-management-service/nextgen-global-service/pkg/config"
	"github.com/tidbcloud/tidb-management-service/utilities/bizerr"
	"github.com/tidbcloud/tidb-management-service/utilities/workflow"
)

func TestCreateManualBackupRegionalFailureKeepsPendingState(t *testing.T) {
	ctrl := gomock.NewController(t)
	backupRepo := mockbrrepo.NewMockBackupRepo(ctrl)
	clusterRepo := mockcluster.NewMockRepo(ctrl)
	regionRepo := mockregion.NewMockRepo(ctrl)
	resourcePoolRepo := mockresourcepool.NewMockRepo(ctrl)
	regionalCli := mockregionalserver.NewMockClient(ctrl)
	transaction := &countingTransactioner{}
	cluster, region, resourcePool := submissionTestResources()
	backup := &entity.Backup{Backup: &schema.Backup{ID: 42, State: schema.BackupStatePending}}

	requestJSON := marshalSubmissionRequest(t, &clusterngv1beta.CreateManualBackupRequest{
		BackupId:  "42",
		ClusterId: "100",
	})
	clusterRepo.EXPECT().Get(gomock.Any(), uint64(100), true).Return(cluster, nil)
	regionRepo.EXPECT().Get(gomock.Any(), uint64(1)).Return(region, nil)
	backupRepo.EXPECT().Get(gomock.Any(), uint64(42)).Return(backup, nil)
	resourcePoolRepo.EXPECT().Load(gomock.Any(), uint64(200), true).Return(resourcePool, nil)
	regionalCli.EXPECT().CreateManualBackup(gomock.Any(), region, gomock.Any()).
		Return(bizerr.New(bizerr.ErrCodeUnknown, "regional failure"))

	action := NewCreateManualBackup(
		backupRepo,
		regionRepo,
		clusterRepo,
		resourcePoolRepo,
		nil,
		regionalCli,
		transaction,
		nil,
	)
	err := action.Run(&fakeExecuteContext{vars: map[string]string{workflowcommon.KeyRequest: requestJSON}}, nil)

	require.Error(t, err)
	require.Equal(t, schema.BackupStatePending, backup.State)
	require.Zero(t, transaction.calls)
}

func TestCreatePausePreparationBackupCancelsStaleOperationChild(t *testing.T) {
	ctrl := gomock.NewController(t)
	backupRepo := mockbrrepo.NewMockBackupRepo(ctrl)
	clusterRepo := mockcluster.NewMockRepo(ctrl)
	regionRepo := mockregion.NewMockRepo(ctrl)
	resourcePoolRepo := mockresourcepool.NewMockRepo(ctrl)
	pauseOperationRepo := mockresourcepool.NewMockPauseOperationRepo(ctrl)
	regionalCli := mockregionalserver.NewMockClient(ctrl)
	transaction := &countingTransactioner{}
	cluster, region, resourcePool := submissionTestResources()
	resourcePool.LifecycleState = globalngv1.ResourcePool_PAUSE_PRECHECKING
	pauseOperationID := "pause-a"
	backup := &entity.Backup{Backup: &schema.Backup{
		ID:               42,
		State:            schema.BackupStatePending,
		PauseOperationId: &pauseOperationID,
	}}
	requestJSON := marshalSubmissionRequest(t, &clusterngv1beta.CreateManualBackupRequest{
		BackupId:  "42",
		ClusterId: "100",
	})

	clusterRepo.EXPECT().Get(gomock.Any(), uint64(100), true).Return(cluster, nil)
	regionRepo.EXPECT().Get(gomock.Any(), uint64(1)).Return(region, nil)
	backupRepo.EXPECT().Get(gomock.Any(), uint64(42)).Return(backup, nil).Times(2)
	resourcePoolRepo.EXPECT().LoadForUpdate(gomock.Any(), uint64(200), false).Return(resourcePool, nil)
	pauseOperationRepo.EXPECT().GetRunning(gomock.Any(), uint64(200)).Return(
		&entity.ResourcePoolPauseOperation{
			ResourcePoolID: 200,
			OperationID:    "pause-b",
			Status:         entity.ResourcePoolPauseOperationRunning,
		},
		nil,
	)
	backupRepo.EXPECT().Save(gomock.Any(), backup).DoAndReturn(func(_ any, saved *entity.Backup) bizerr.BizError {
		require.Equal(t, schema.BackupStateCancelled, saved.State)
		return nil
	})

	action := NewCreateManualBackup(
		backupRepo,
		regionRepo,
		clusterRepo,
		resourcePoolRepo,
		pauseOperationRepo,
		regionalCli,
		transaction,
		nil,
	)
	err := action.Run(&fakeExecuteContext{vars: map[string]string{workflowcommon.KeyRequest: requestJSON}}, nil)

	require.Error(t, err)
	require.True(t, workflow.IsTerminalError(err))
	require.Equal(t, schema.BackupStateCancelled, backup.State)
	require.Equal(t, 1, transaction.calls)
}

func TestCreatePausePreparationBackupDoesNotHoldPoolLockAcrossRegionalCall(t *testing.T) {
	ctrl := gomock.NewController(t)
	backupRepo := mockbrrepo.NewMockBackupRepo(ctrl)
	clusterRepo := mockcluster.NewMockRepo(ctrl)
	regionRepo := mockregion.NewMockRepo(ctrl)
	resourcePoolRepo := mockresourcepool.NewMockRepo(ctrl)
	pauseOperationRepo := mockresourcepool.NewMockPauseOperationRepo(ctrl)
	regionalCli := mockregionalserver.NewMockClient(ctrl)
	transaction := &trackingTransactioner{}
	cluster, region, resourcePool := submissionTestResources()
	resourcePool.LifecycleState = globalngv1.ResourcePool_PAUSE_PRECHECKING
	pauseOperationID := "pause-a"
	backup := &entity.Backup{Backup: &schema.Backup{
		ID:               42,
		State:            schema.BackupStatePending,
		PauseOperationId: &pauseOperationID,
	}}
	requestJSON := marshalSubmissionRequest(t, &clusterngv1beta.CreateManualBackupRequest{
		BackupId:  "42",
		ClusterId: "100",
	})

	clusterRepo.EXPECT().Get(gomock.Any(), uint64(100), true).Return(cluster, nil)
	regionRepo.EXPECT().Get(gomock.Any(), uint64(1)).Return(region, nil)
	backupRepo.EXPECT().Get(gomock.Any(), uint64(42)).Return(backup, nil).Times(3)
	resourcePoolRepo.EXPECT().LoadForUpdate(gomock.Any(), uint64(200), false).Return(resourcePool, nil)
	pauseOperationRepo.EXPECT().GetRunning(gomock.Any(), uint64(200)).Return(
		&entity.ResourcePoolPauseOperation{
			ResourcePoolID: 200,
			OperationID:    "pause-a",
			Status:         entity.ResourcePoolPauseOperationRunning,
			DeadlineAt:     time.Now().Add(time.Hour),
		},
		nil,
	)
	states := make([]schema.BackupState, 0, 2)
	backupRepo.EXPECT().Save(gomock.Any(), backup).DoAndReturn(func(_ any, saved *entity.Backup) bizerr.BizError {
		states = append(states, saved.State)
		return nil
	}).Times(2)
	regionalCli.EXPECT().CreateManualBackup(gomock.Any(), region, gomock.Any()).DoAndReturn(
		func(context.Context, *entity.Region, *clusterngv1beta.CreateManualBackupRequest) bizerr.BizError {
			require.False(t, transaction.active)
			require.Equal(t, schema.BackupStateSubmitting, backup.State)
			return nil
		},
	)

	action := NewCreateManualBackup(
		backupRepo,
		regionRepo,
		clusterRepo,
		resourcePoolRepo,
		pauseOperationRepo,
		regionalCli,
		transaction,
		nil,
	)
	err := action.Run(&fakeExecuteContext{vars: map[string]string{workflowcommon.KeyRequest: requestJSON}}, nil)

	require.NoError(t, err)
	require.Equal(t, []schema.BackupState{schema.BackupStateSubmitting, schema.BackupStateRunning}, states)
	require.Equal(t, 2, transaction.calls)
}

type trackingTransactioner struct {
	active bool
	calls  int
}

func (t *trackingTransactioner) Transaction(ctx context.Context, callback func(context.Context) error) error {
	t.calls++
	t.active = true
	defer func() { t.active = false }()
	return callback(ctx)
}

func TestCreateLightweightRestoreRegionalFailureKeepsPendingState(t *testing.T) {
	ctrl := gomock.NewController(t)
	restoreRepo := mockbrrepo.NewMockRestoreRepo(ctrl)
	clusterRepo := mockcluster.NewMockRepo(ctrl)
	regionRepo := mockregion.NewMockRepo(ctrl)
	resourcePoolRepo := mockresourcepool.NewMockRepo(ctrl)
	regionalCli := mockregionalserver.NewMockClient(ctrl)
	transaction := &countingTransactioner{}
	cluster, region, resourcePool := submissionTestResources()
	restore := submissionTestRestore()

	requestJSON := marshalSubmissionRequest(t, &clusterngv1beta.CreateLightweightRestoreRequest{
		RestoreId:          "restore-100",
		SourceSharedPoolId: "300",
		SourceClusterId:    "400",
		TargetSharedPoolId: "200",
		TargetClusterId:    "100",
	})
	clusterRepo.EXPECT().Get(gomock.Any(), uint64(100), true).Return(cluster, nil).Times(2)
	regionRepo.EXPECT().Get(gomock.Any(), uint64(1)).Return(region, nil)
	restoreRepo.EXPECT().Get(gomock.Any(), uint64(100)).Return(restore, nil)
	resourcePoolRepo.EXPECT().Load(gomock.Any(), uint64(200), true).Return(resourcePool, nil)
	regionalCli.EXPECT().CreateLightweightRestore(gomock.Any(), region, gomock.Any()).
		Return(bizerr.New(bizerr.ErrCodeUnknown, "regional failure"))

	action := NewCreateLightweightRestoreAction(
		restoreRepo,
		regionRepo,
		clusterRepo,
		nil,
		regionalCli,
		transaction,
		nil,
		resourcePoolRepo,
	)
	err := action.Run(&fakeExecuteContext{vars: map[string]string{workflowcommon.KeyRequest: requestJSON}}, nil)

	require.Error(t, err)
	require.Equal(t, "pending", restore.State)
	require.Zero(t, transaction.calls)
}

func TestCreateManualBackupRestoreRegionalFailureKeepsPendingState(t *testing.T) {
	ctrl := gomock.NewController(t)
	restoreRepo := mockbrrepo.NewMockRestoreRepo(ctrl)
	clusterRepo := mockcluster.NewMockRepo(ctrl)
	regionRepo := mockregion.NewMockRepo(ctrl)
	resourcePoolRepo := mockresourcepool.NewMockRepo(ctrl)
	regionalCli := mockregionalserver.NewMockClient(ctrl)
	transaction := &countingTransactioner{}
	cluster, region, resourcePool := submissionTestResources()
	restore := submissionTestRestore()

	requestJSON := marshalSubmissionRequest(t, &clusterngv1beta.CreateManualBackupRestoreRequest{
		RestoreId:          "restore-100",
		SourceClusterId:    "400",
		BackupId:           "backup-1",
		TargetSharedPoolId: "200",
		TargetClusterId:    "100",
	})
	clusterRepo.EXPECT().Get(gomock.Any(), uint64(100), true).Return(cluster, nil).Times(2)
	regionRepo.EXPECT().Get(gomock.Any(), uint64(1)).Return(region, nil)
	restoreRepo.EXPECT().Get(gomock.Any(), uint64(100)).Return(restore, nil)
	resourcePoolRepo.EXPECT().Load(gomock.Any(), uint64(200), true).Return(resourcePool, nil)
	regionalCli.EXPECT().CreateManualBackupRestore(gomock.Any(), region, gomock.Any()).
		Return(bizerr.New(bizerr.ErrCodeUnknown, "regional failure"))

	action := NewCreateManualBackupRestore(
		restoreRepo,
		regionRepo,
		clusterRepo,
		nil,
		regionalCli,
		transaction,
		nil,
		resourcePoolRepo,
	)
	err := action.Run(&fakeExecuteContext{vars: map[string]string{workflowcommon.KeyRequest: requestJSON}}, nil)

	require.Error(t, err)
	require.Equal(t, "pending", restore.State)
	require.Zero(t, transaction.calls)
}

func TestCreateClassicRestoreRegionalFailureKeepsPendingState(t *testing.T) {
	ctrl := gomock.NewController(t)
	restoreRepo := mockbrrepo.NewMockRestoreRepo(ctrl)
	clusterRepo := mockcluster.NewMockRepo(ctrl)
	regionRepo := mockregion.NewMockRepo(ctrl)
	resourcePoolRepo := mockresourcepool.NewMockRepo(ctrl)
	regionalCli := mockregionalserver.NewMockClient(ctrl)
	dataflowCli := mockdataflow.NewMockDataFlow(ctrl)
	transaction := &countingTransactioner{}
	cluster, region, resourcePool := submissionTestResources()
	restore := submissionTestRestore()
	restore.SourceRoleArn = "source-role"

	restoreRepo.EXPECT().Get(gomock.Any(), uint64(100)).Return(restore, nil)
	clusterRepo.EXPECT().Get(gomock.Any(), uint64(100), false).Return(cluster, nil)
	clusterRepo.EXPECT().Get(gomock.Any(), uint64(100), true).Return(cluster, nil)
	regionRepo.EXPECT().Get(gomock.Any(), uint64(1)).Return(region, nil)
	resourcePoolRepo.EXPECT().Load(gomock.Any(), uint64(200), true).Return(resourcePool, nil)
	regionalCli.EXPECT().CreateClassicRestore(gomock.Any(), region, gomock.Any()).
		Return(bizerr.New(bizerr.ErrCodeUnknown, "regional failure"))

	action := NewCreateClassicRestoreAction(
		restoreRepo,
		clusterRepo,
		regionRepo,
		dataflowCli,
		regionalCli,
		&config.Config{},
		transaction,
		nil,
		resourcePoolRepo,
	)
	err := action.Run(&fakeExecuteContext{vars: map[string]string{workflowcommon.KeyTargetClusterId: "100"}}, nil)

	require.Error(t, err)
	require.Equal(t, "pending", restore.State)
	require.Zero(t, transaction.calls)
}

func submissionTestResources() (*entity.Cluster, *entity.Region, *entity.ResourcePool) {
	cluster := &entity.Cluster{
		Cluster: &globalngv1.Cluster{
			ClusterId:    100,
			SharedPoolId: "200",
		},
		RegionId: 1,
	}
	region := &entity.Region{
		Region:   &globalcommonv1.Region{},
		RegionID: 1,
	}
	resourcePool := &entity.ResourcePool{
		ResourcePool: &globalngv1.ResourcePool{
			Id:             200,
			LifecycleState: globalngv1.ResourcePool_ACTIVE,
		},
	}
	return cluster, region, resourcePool
}

func submissionTestRestore() *entity.Restore {
	return &entity.Restore{Restore: &schema.Restore{
		ID:                 1,
		State:              "pending",
		TargetClusterId:    100,
		TargetSharedPoolId: 200,
	}}
}

func marshalSubmissionRequest(t *testing.T, request proto.Message) string {
	t.Helper()
	requestJSON, err := protojson.Marshal(request)
	require.NoError(t, err)
	return string(requestJSON)
}