package action

import (
	"context"
	"errors"
	"strings"
	"testing"
	"time"
	"unicode/utf8"

	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/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow"
	dataplanemocks "github.com/tidbcloud/cluster-service-ng/pkg/dataplane/mocks"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/actions"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/executor"
	repomocks "github.com/tidbcloud/cluster-service-ng/pkg/storage/repo/mocks"
	"google.golang.org/protobuf/types/known/timestamppb"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"
)

type stageOperationRepo struct {
	operation *entity.PauseResumeOperation
	saves     int
}

func (r *stageOperationRepo) Create(context.Context, *entity.PauseResumeOperation) error { return nil }
func (r *stageOperationRepo) LoadByID(context.Context, string, string) (*entity.PauseResumeOperation, error) {
	return r.operation, nil
}
func (r *stageOperationRepo) LoadLatestBySharedPoolID(context.Context, string) (*entity.PauseResumeOperation, error) {
	return r.operation, nil
}
func (r *stageOperationRepo) UpdateStageIfRunning(_ context.Context, _, _ string, stage entity.PauseResumeOperationStage) (bool, error) {
	if r.operation.GetStatus() != apiv1.PauseResumeOperation_RUNNING {
		return false, nil
	}
	r.operation.Stage = stage
	r.saves++
	return true, nil
}
func (r *stageOperationRepo) MarkSucceededIfRunning(context.Context, string, string, string, time.Time) (bool, error) {
	return false, nil
}
func (r *stageOperationRepo) MarkFailedIfRunning(_ context.Context, _, _, summary string, completedAt time.Time) (bool, error) {
	if r.operation.GetStatus() != apiv1.PauseResumeOperation_RUNNING {
		return false, nil
	}
	r.operation.Status = apiv1.PauseResumeOperation_FAILED
	r.operation.ResultSummary = summary
	r.operation.CompletedAt = timestamppb.New(completedAt)
	r.saves++
	return true, nil
}
func (r *stageOperationRepo) CancelIfRunning(context.Context, string, string, string, time.Time) (bool, error) {
	return false, nil
}

type errorAction struct{ err error }

func (a errorAction) Name() string                    { return "stage-action" }
func (a errorAction) Invoke(actions.ActionArgs) error { return a.err }

func TestPauseResumeResultSummaryPreservesUTF8AtLimit(t *testing.T) {
	summary := pauseResumeResultSummary(errors.New(strings.Repeat("中", 400)))
	require.LessOrEqual(t, len(summary), 1024)
	require.True(t, utf8.ValidString(summary))
}

func TestPauseResumeStageDecoratorPersistsTerminalFailure(t *testing.T) {
	repo := &stageOperationRepo{operation: &entity.PauseResumeOperation{
		SharedPoolId: "100", OperationId: "op-1", Type: apiv1.PauseResumeOperation_PAUSE, Status: apiv1.PauseResumeOperation_RUNNING,
	}}
	action := DecoratePauseResumeStage(repo, apiv1.PauseResumeOperation_FLUSH_WAL, errorAction{err: actions.NewTerminalError(errors.New("safe failure"))})
	err := action.Invoke(actions.ActionArgs{
		workflow.SharedPoolIdKey:             actions.ActionArgString("100"),
		workflow.PauseResumeOperationIdKey:   actions.ActionArgString("op-1"),
		workflow.PauseResumeOperationTypeKey: actions.ActionArgString(apiv1.PauseResumeOperation_PAUSE.String()),
	})
	require.Error(t, err)
	require.True(t, actions.IsTerminalError(err))
	require.Equal(t, 2, repo.saves)
	require.Equal(t, apiv1.PauseResumeOperation_FAILED, repo.operation.GetStatus())
	require.Equal(t, apiv1.PauseResumeOperation_FLUSH_WAL, repo.operation.GetStage())
	require.NotNil(t, repo.operation.GetCompletedAt())
}

func TestPauseResumeStageDecoratorKeepsRetryableFailureRunning(t *testing.T) {
	repo := &stageOperationRepo{operation: &entity.PauseResumeOperation{
		SharedPoolId: "100", OperationId: "op-1", Type: apiv1.PauseResumeOperation_RESUME, Status: apiv1.PauseResumeOperation_RUNNING,
	}}
	action := DecoratePauseResumeStage(repo, apiv1.PauseResumeOperation_APPLY_RESUME, errorAction{err: errors.New("temporary")})
	err := action.Invoke(actions.ActionArgs{
		workflow.SharedPoolIdKey:             actions.ActionArgString("100"),
		workflow.PauseResumeOperationIdKey:   actions.ActionArgString("op-1"),
		workflow.PauseResumeOperationTypeKey: actions.ActionArgString(apiv1.PauseResumeOperation_RESUME.String()),
	})
	require.EqualError(t, err, "temporary")
	require.Equal(t, 1, repo.saves)
	require.Equal(t, apiv1.PauseResumeOperation_RUNNING, repo.operation.GetStatus())
}

func TestPauseResumeStageDecoratorRejectsDifferentOperationType(t *testing.T) {
	repo := &stageOperationRepo{operation: &entity.PauseResumeOperation{
		SharedPoolId: "100", OperationId: "op-1", Type: apiv1.PauseResumeOperation_RESUME, Status: apiv1.PauseResumeOperation_RUNNING,
	}}
	action := DecoratePauseResumeStage(repo, apiv1.PauseResumeOperation_APPLY_SUSPEND, errorAction{})
	err := action.Invoke(actions.ActionArgs{
		workflow.SharedPoolIdKey:             actions.ActionArgString("100"),
		workflow.PauseResumeOperationIdKey:   actions.ActionArgString("op-1"),
		workflow.PauseResumeOperationTypeKey: actions.ActionArgString(apiv1.PauseResumeOperation_PAUSE.String()),
	})
	require.Error(t, err)
	require.True(t, actions.IsTerminalError(err))
	require.Zero(t, repo.saves)
	require.Equal(t, apiv1.PauseResumeOperation_RUNNING, repo.operation.GetStatus())
}

func TestPauseResumeOwnerGuardFencesCancelledOperation(t *testing.T) {
	operationRepo := repomocks.NewPauseResumeOperationRepo(t)
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	lockRepo := repomocks.NewLockRepo(t)
	lockRepo.EXPECT().Lock(mock.Anything, mock.Anything).Return(func() error { return nil }, nil).Once()
	operationRepo.EXPECT().LoadByID(mock.Anything, "100", "op-1").Return(&entity.PauseResumeOperation{
		SharedPoolId: "100", OperationId: "op-1", Type: apiv1.PauseResumeOperation_PAUSE, Status: apiv1.PauseResumeOperation_CANCELLED,
	}, nil).Once()
	wrote := false
	err := NewPauseResumeOwnerGuard(operationRepo, sharedPoolRepo, lockRepo).Run(pauseResumeTestArgs(apiv1.PauseResumeOperation_PAUSE), func() error {
		wrote = true
		return nil
	})
	require.Error(t, err)
	require.True(t, actions.IsTerminalError(err))
	require.False(t, wrote)
}

func TestPauseResumeOwnerGuardRunsWriteForCurrentOwner(t *testing.T) {
	operationRepo := repomocks.NewPauseResumeOperationRepo(t)
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	lockRepo := repomocks.NewLockRepo(t)
	lockRepo.EXPECT().Lock(mock.Anything, mock.Anything).Return(func() error { return nil }, nil).Once()
	operationRepo.EXPECT().LoadByID(mock.Anything, "100", "op-1").Return(&entity.PauseResumeOperation{
		SharedPoolId: "100", OperationId: "op-1", Type: apiv1.PauseResumeOperation_PAUSE, Status: apiv1.PauseResumeOperation_RUNNING,
	}, nil).Once()
	sharedPoolRepo.EXPECT().Load(mock.Anything, "100", false).Return(&entity.SharedPool{PauseResumeState: entity.PauseResumeStatePausing}, nil).Once()
	wrote := false
	require.NoError(t, NewPauseResumeOwnerGuard(operationRepo, sharedPoolRepo, lockRepo).Run(pauseResumeTestArgs(apiv1.PauseResumeOperation_PAUSE), func() error {
		wrote = true
		return nil
	}))
	require.True(t, wrote)
}

func pauseResumeTestArgs(operationType apiv1.PauseResumeOperation_Type) actions.ActionArgs {
	return actions.ActionArgs{
		workflow.SharedPoolIdKey: actions.ActionArgString("100"), workflow.PauseResumeOperationIdKey: actions.ActionArgString("op-1"),
		workflow.PauseResumeOperationTypeKey: actions.ActionArgString(operationType.String()),
	}
}

func TestPausePrecheckBlocksCrossPoolRestoreUsingSourcePoolFilter(t *testing.T) {
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	sharedPoolRepo.EXPECT().Load(mock.Anything, "pool-1", false).Return(&entity.SharedPool{
		PauseResumeState: entity.PauseResumeStatePausing,
	}, nil)
	queriedSourcePool := false
	precheck := NewPausePrecheckAction(sharedPoolRepo, nil, nil, true,
		func(plan string, filter executor.MatchFilter) (executor.ExecTaskIdStatus, error) {
			if plan == "create-lightweight-x-restore" && len(filter[workflow.SourceSharedPoolIdKey]) == 1 && filter[workflow.SourceSharedPoolIdKey][0] == "pool-1" {
				queriedSourcePool = true
				return executor.ExecTaskIdStatus{"cross-restore": nil}, nil
			}
			return executor.NewExecTaskIdStatus(), nil
		}, nil, []string{"create-lightweight-x-restore"})

	err := precheck.Invoke(actions.ActionArgs{
		workflow.SharedPoolIdKey:             actions.ActionArgString("pool-1"),
		workflow.PauseResumeOperationIdKey:   actions.ActionArgString("pause-1"),
		workflow.PauseResumeOperationTypeKey: actions.ActionArgString(apiv1.PauseResumeOperation_PAUSE.String()),
	})
	require.Error(t, err)
	require.True(t, queriedSourcePool)
}

func TestPausePrecheckRejectsPendingOperatorSuspension(t *testing.T) {
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	sharedPoolRepo.EXPECT().Load(mock.Anything, "100", false).Return(&entity.SharedPool{
		SharedPool:       apiv1.SharedPool{Id: "100", DataplaneName: "dp-1"},
		PauseResumeState: entity.PauseResumeStatePausing,
	}, nil).Once()
	manager := dataplanemocks.NewManager(t)
	manager.EXPECT().GetClient("dp-1").Return(newPendingSuspendComputeClient(t), nil).Once()
	precheck := NewPausePrecheckAction(sharedPoolRepo, nil, manager, true,
		func(string, executor.MatchFilter) (executor.ExecTaskIdStatus, error) {
			return executor.NewExecTaskIdStatus(), nil
		}, nil, nil)

	err := precheck.Invoke(actions.ActionArgs{
		workflow.SharedPoolIdKey:             actions.ActionArgString("100"),
		workflow.PauseResumeOperationIdKey:   actions.ActionArgString("pause-1"),
		workflow.PauseResumeOperationTypeKey: actions.ActionArgString(apiv1.PauseResumeOperation_PAUSE.String()),
	})
	require.ErrorContains(t, err, "operator suspend spec is true")
	require.True(t, actions.IsTerminalError(err))
}

func TestPausePrecheckChecksFlushWALWhileOperatorSuspensionIsPending(t *testing.T) {
	sharedPoolRepo := repomocks.NewSharedPoolRepo(t)
	sharedPoolRepo.EXPECT().Load(mock.Anything, "100", false).Return(&entity.SharedPool{
		SharedPool:            apiv1.SharedPool{Id: "100", DataplaneName: "dp-1"},
		PauseResumeState:      entity.PauseResumeStatePausing,
		SuspendComputeDesired: true,
	}, nil).Once()
	manager := dataplanemocks.NewManager(t)
	manager.EXPECT().GetClient("dp-1").Return(newPendingSuspendComputeClient(t), nil).Once()
	precheck := NewPausePrecheckAction(sharedPoolRepo, nil, manager, true,
		func(string, executor.MatchFilter) (executor.ExecTaskIdStatus, error) {
			return executor.NewExecTaskIdStatus(), nil
		}, nil, nil)

	err := precheck.Invoke(actions.ActionArgs{
		workflow.SharedPoolIdKey:             actions.ActionArgString("100"),
		workflow.PauseResumeOperationIdKey:   actions.ActionArgString("pause-1"),
		workflow.PauseResumeOperationTypeKey: actions.ActionArgString(apiv1.PauseResumeOperation_PAUSE.String()),
	})
	require.ErrorContains(t, err, "FlushWAL is disabled")
	require.True(t, actions.IsTerminalError(err))
}

func newPendingSuspendComputeClient(t *testing.T) client.Client {
	t.Helper()
	scheme := runtime.NewScheme()
	require.NoError(t, operatorv1alpha1.AddToScheme(scheme))
	cluster := &operatorv1alpha1.Cluster{
		ObjectMeta: metav1.ObjectMeta{Name: "db", Namespace: entity.GetDataPlaneNamespace("100"), Generation: 2},
		Spec:       operatorv1alpha1.ClusterSpec{SuspendAction: &operatorv1alpha1.SuspendAction{SuspendCompute: true}},
		Status:     operatorv1alpha1.ClusterStatus{ObservedGeneration: 1},
	}
	return fake.NewClientBuilder().WithScheme(scheme).WithObjects(cluster).Build()
}