package server

import (
	"context"
	"errors"
	"fmt"
	"testing"

	mysqlDriver "github.com/go-sql-driver/mysql"
	"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/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow"
	braction "github.com/tidbcloud/cluster-service-ng/domain/workflow/action/br"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow/dag"
	"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"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"gorm.io/gorm"
)

func TestPauseResumeDuplicateKeyDetectionHandlesWrappedMySQLError(t *testing.T) {
	err := fmt.Errorf("create operation: %w", &mysqlDriver.MySQLError{Number: 1062, Message: "Duplicate entry"})
	require.True(t, isPauseResumeDuplicateKeyError(err))
	require.True(t, isPauseResumeDuplicateKeyError(fmt.Errorf("translated: %w", gorm.ErrDuplicatedKey)))
	require.False(t, isPauseResumeDuplicateKeyError(fmt.Errorf("other")))
}

func TestProjectSharedPoolPhase(t *testing.T) {
	tests := []struct {
		name    string
		base    apiv1.SharedPool_Phase
		state   entity.PauseResumeState
		desired bool
		want    apiv1.SharedPool_Phase
		wantErr bool
	}{
		{name: "pausing before flush", base: apiv1.SharedPool_AVAILABLE, state: entity.PauseResumeStatePausing, want: apiv1.SharedPool_PAUSING},
		{name: "pausing after flush", base: apiv1.SharedPool_UNAVAILABLE, state: entity.PauseResumeStatePausing, desired: true, want: apiv1.SharedPool_PAUSING},
		{name: "paused", base: apiv1.SharedPool_UNAVAILABLE, state: entity.PauseResumeStatePaused, desired: true, want: apiv1.SharedPool_PAUSED},
		{name: "resuming", base: apiv1.SharedPool_MODIFYING, state: entity.PauseResumeStateResuming, want: apiv1.SharedPool_RESUMING},
		{name: "deleting wins", base: apiv1.SharedPool_DELETING, state: entity.PauseResumeStatePaused, desired: true, want: apiv1.SharedPool_DELETING},
		{name: "invalid combination", base: apiv1.SharedPool_AVAILABLE, desired: true, want: apiv1.SharedPool_AVAILABLE, wantErr: true},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			sharedPool := &entity.SharedPool{SharedPool: apiv1.SharedPool{Phase: tt.base}, PauseResumeState: tt.state, SuspendComputeDesired: tt.desired}
			err := projectSharedPoolPhase(sharedPool)
			if tt.wantErr {
				require.Error(t, err)
				return
			}
			require.NoError(t, err)
			require.Equal(t, tt.want, sharedPool.GetPhase())
		})
	}
}

func TestPauseResumeTransitionalStateDistinguishesRunningAndStranded(t *testing.T) {
	tests := []struct {
		name    string
		running bool
		loadErr error
		code    codes.Code
		reason  string
		wantErr bool
	}{
		{name: "running operation", running: true, code: codes.FailedPrecondition, reason: reasonPauseResumeOperationInProgress, wantErr: true},
		{name: "latest operation dependency failure", loadErr: errors.New("database unavailable"), code: codes.Unavailable, reason: reasonPauseResumeDependencyUnavailable, wantErr: true},
		{name: "failed operation permits new-id retry"},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			dagExecutor := executormocks.NewRemoteExecutor(t)
			running := executor.NewExecTaskIdStatus()
			if tt.running {
				running["current-pause"] = &executor.ExecTaskStatusEx{}
			}
			dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanPauseSharedPool, mock.Anything).Return(running, nil).Once()
			dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanResumeSharedPool, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil).Once()
			operationRepo := repomocks.NewPauseResumeOperationRepo(t)
			if !tt.running {
				if tt.loadErr != nil {
					operationRepo.On("LoadLatestBySharedPoolID", mock.Anything, "100").Return(nil, tt.loadErr).Once()
				} else {
					operationRepo.On("LoadLatestBySharedPoolID", mock.Anything, "100").Return(&entity.PauseResumeOperation{
						SharedPoolId: "100", OperationId: "old-op", Type: apiv1.PauseResumeOperation_PAUSE, Status: apiv1.PauseResumeOperation_FAILED,
					}, nil).Once()
				}
			}
			s := &serverImpl{dagExecutor: dagExecutor, pauseResumeOperationRepo: operationRepo}
			sharedPool := &entity.SharedPool{
				SharedPool:       apiv1.SharedPool{Id: "100", Phase: apiv1.SharedPool_PAUSING},
				PauseResumeState: entity.PauseResumeStatePausing,
			}

			err := s.validatePauseResumeLifecycle(context.Background(), sharedPool, apiv1.PauseResumeOperation_PAUSE, "")
			if tt.wantErr {
				require.Error(t, err)
				require.Equal(t, tt.code, status.Code(err))
				require.Equal(t, tt.reason, errorInfoReason(err))
			} else {
				require.NoError(t, err)
			}
		})
	}
}

func TestPauseResumeLifecycleIgnoresDAGForSameOperationID(t *testing.T) {
	const operationID = "op-1"
	dagExecutor := executormocks.NewRemoteExecutor(t)
	running := executor.NewExecTaskIdStatus()
	running[workflow.PauseResumeExecID("100", operationID, apiv1.PauseResumeOperation_PAUSE.String())] = &executor.ExecTaskStatusEx{}
	dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanPauseSharedPool, mock.Anything).Return(running, nil).Once()
	dagExecutor.On("QueryRunnings", mock.Anything, dag.PlanResumeSharedPool, mock.Anything).Return(executor.NewExecTaskIdStatus(), nil).Once()
	s := &serverImpl{dagExecutor: dagExecutor}
	sharedPool := &entity.SharedPool{SharedPool: apiv1.SharedPool{Id: "100", Phase: apiv1.SharedPool_AVAILABLE}}

	require.NoError(t, s.validatePauseResumeLifecycle(context.Background(), sharedPool, apiv1.PauseResumeOperation_PAUSE, operationID))
}

func TestLoadPauseResumeSharedPoolMapsDependencyErrors(t *testing.T) {
	tests := []struct {
		name       string
		loadErr    error
		wantCode   codes.Code
		wantReason string
	}{
		{name: "dependency failure", loadErr: errors.New("operator unavailable"), wantCode: codes.Unavailable, wantReason: reasonPauseResumeDependencyUnavailable},
		{name: "not found", loadErr: status.Error(codes.NotFound, "missing"), wantCode: codes.NotFound},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			repo := repomocks.NewSharedPoolRepo(t)
			repo.On("Load", mock.Anything, "100", false).Return(nil, tt.loadErr).Once()
			s := &serverImpl{sharedPoolRepo: repo}

			_, err := s.loadPauseResumeSharedPool(context.Background(), "100")
			require.Equal(t, tt.wantCode, status.Code(err))
			require.Equal(t, tt.wantReason, errorInfoReason(err))
		})
	}
}

func TestPauseFlushWALPrerequisiteErrorMapping(t *testing.T) {
	tests := []struct {
		name       string
		disabled   bool
		loadErr    error
		wantCode   codes.Code
		wantReason string
	}{
		{name: "terminal configuration failure", disabled: true, wantCode: codes.FailedPrecondition, wantReason: reasonPauseResumeInvalidState},
		{name: "retryable dependency failure", loadErr: errors.New("storage unavailable"), wantCode: codes.Unavailable, wantReason: reasonPauseResumeDependencyUnavailable},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
			if !tt.disabled {
				sharedPoolRepo.On("Load", mock.Anything, "100", false).Return(nil, tt.loadErr).Once()
			}
			_, prerequisiteErr := braction.ValidateFlushWALPrerequisites(context.Background(), "100", sharedPoolRepo, nil, nil, nil, tt.disabled)
			require.Error(t, prerequisiteErr)
			err := pauseFlushWALPrerequisiteStatusError(prerequisiteErr)
			require.Equal(t, tt.wantCode, status.Code(err))
			require.Equal(t, tt.wantReason, errorInfoReason(err))
		})
	}
}