package server

import (
	"context"
	"errors"
	"testing"

	v1alpha1br "github.com/pingcap/tidb-operator/api/v2/br/v1alpha1"
	operatorv1alpha1 "github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"github.com/tidbcloud/cluster-service-ng/config"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow/dag"
	dataplanemocks "github.com/tidbcloud/cluster-service-ng/pkg/dataplane/mocks"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/executor"
	executormocks "github.com/tidbcloud/cluster-service-ng/pkg/runtime/executor/mocks"
	repomocks "github.com/tidbcloud/cluster-service-ng/pkg/storage/repo/mocks"
	"github.com/tidbcloud/cluster-service-ng/pkg/tioperator"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"gorm.io/gorm"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"
)

type commitFailureTransactioner struct {
	events *[]string
	err    error
}

func (t *commitFailureTransactioner) Transaction(ctx context.Context, fn func(context.Context) error) error {
	*t.events = append(*t.events, "transaction")
	if err := fn(ctx); err != nil {
		return err
	}
	*t.events = append(*t.events, "commit-error")
	return t.err
}

func TestPauseSharedPoolAbortsSubmittedDAGBeforeUnlockOnCommitFailure(t *testing.T) {
	events := []string{}
	commitErr := errors.New("commit failed")
	sharedPool := &entity.SharedPool{SharedPool: apiv1.SharedPool{
		Id: "pool-1", Phase: apiv1.SharedPool_AVAILABLE, DataplaneName: "dp-1",
	}}

	scheme := runtime.NewScheme()
	require.NoError(t, v1alpha1br.AddToScheme(scheme))
	require.NoError(t, operatorv1alpha1.AddToScheme(scheme))
	namespace := entity.GetDataPlaneNamespace("pool-1")
	cluster := &operatorv1alpha1.Cluster{ObjectMeta: metav1.ObjectMeta{Name: "db", Namespace: namespace}}
	tibr := &v1alpha1br.TiBR{
		ObjectMeta: metav1.ObjectMeta{Name: tioperator.DefaultTiBRName, Namespace: namespace},
		Status: v1alpha1br.TiBRStatus{CommonStatus: operatorv1alpha1.CommonStatus{Conditions: []metav1.Condition{
			{Type: "Ready", Status: metav1.ConditionTrue},
			{Type: "Synced", Status: metav1.ConditionTrue},
		}}},
	}
	remoteClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(cluster, tibr).Build()

	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	sharedPoolRepo.On("Load", mock.Anything, "pool-1", false).Return(sharedPool, nil).Times(3)
	sharedPoolRepo.On("UpdatePauseResumeState", mock.Anything, "pool-1", entity.PauseResumeStatePausing, false).Return(nil).Once()
	operationRepo := repomocks.NewPauseResumeOperationRepo(t)
	operationRepo.On("LoadByID", mock.Anything, "pool-1", "op-1").Return(nil, gorm.ErrRecordNotFound).Twice()
	operationRepo.On("Create", mock.Anything, mock.Anything).Return(nil).Once()
	dataplaneRepo := repomocks.NewDataplaneRepo(t)
	dataplaneRepo.On("LoadByName", mock.Anything, "dp-1").Return(&entity.DataPlaneProvider{Name: "dp-1", GatewayUrl: "gateway.test"}, nil).Once()
	dataplaneManager := dataplanemocks.NewManager(t)
	dataplaneManager.On("GetClient", "dp-1").Return(remoteClient, nil).Once()
	lockRepo := repomocks.NewLockRepo(t)
	lockRepo.On("Lock", mock.Anything, mock.Anything).Run(func(mock.Arguments) {
		events = append(events, "lock")
	}).Return(func() error {
		events = append(events, "unlock")
		return nil
	}, nil).Once()
	dagExecutor := executormocks.NewRemoteExecutor(t)
	dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)
	execID := workflow.PauseResumeExecID("pool-1", "op-1", apiv1.PauseResumeOperation_PAUSE.String())
	dagExecutor.On("Submit", mock.Anything, dag.PlanPauseSharedPool, mock.Anything, execID).Run(func(mock.Arguments) {
		events = append(events, "submit")
	}).Return(nil).Once()
	dagExecutor.On("Abort", mock.Anything, execID).Run(func(mock.Arguments) {
		events = append(events, "abort")
	}).Return(nil).Once()

	s := &serverImpl{
		transactioner:            &commitFailureTransactioner{events: &events, err: commitErr},
		sharedPoolRepo:           sharedPoolRepo,
		pauseResumeOperationRepo: operationRepo,
		dataplaneRepo:            dataplaneRepo,
		dataplaneManager:         dataplaneManager,
		lockRepo:                 lockRepo,
		dagExecutor:              dagExecutor,
		bizCfg:                   &config.BizConfig{},
	}

	_, err := s.PauseSharedPool(context.Background(), "pool-1", "op-1")
	require.ErrorIs(t, err, commitErr)
	require.Equal(t, []string{"lock", "transaction", "submit", "commit-error", "abort", "unlock"}, events)
}

func TestPauseAdmissionBlocksCrossPoolRestoreUsingPoolAsSource(t *testing.T) {
	dagExecutor := executormocks.NewRemoteExecutor(t)
	running := executor.NewExecTaskIdStatus()
	running["cross-restore"] = nil
	dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanCreateLightweightXRestore, mock.MatchedBy(matchesSourcePoolFilter("pool-1"))).Return(running, nil).Once()
	dagExecutor.On("QueryRunnings", mock.Anything, mock.Anything, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil)

	s := &serverImpl{dagExecutor: dagExecutor}
	err := s.checkPauseResumeConflictingDAGs(context.Background(), "pool-1", false)
	require.Equal(t, codes.FailedPrecondition, status.Code(err))
	require.ErrorContains(t, err, "a conflicting operation is in progress")
}

func TestPauseResumeConflictPlansSkipCreateOnlyDuringRecovery(t *testing.T) {
	plans := []string{dag.PlanCreateTiDBCluster, dag.PlanUpdateTiDBCluster}

	require.Equal(t, plans, pauseResumeConflictPlans(plans, false))
	require.Equal(t, []string{dag.PlanUpdateTiDBCluster}, pauseResumeConflictPlans(plans, true))
}