package resourcepool

import (
	"context"
	"errors"
	"testing"

	"github.com/stretchr/testify/require"
	"go.uber.org/mock/gomock"
	"google.golang.org/grpc"
	grpcmetadata "google.golang.org/grpc/metadata"

	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/adapter/nextgenglobal/mock_nextgenglobal"
	"github.com/tidbcloud/tidb-management-service/tidb-mgmt-service/domain/entity"
	grpcutil "github.com/tidbcloud/tidb-management-service/utilities/grpc"
	requestmetadata "github.com/tidbcloud/tidb-management-service/utilities/grpc/metadata"
)

type fakePauseCustomStatusRepository struct {
	statuses map[uint64][]*entity.ClusterCustomStatus
	err      error
	locked   bool
}

func (f *fakePauseCustomStatusRepository) LockClustersForUpdate(_ context.Context, _ []uint64) error {
	f.locked = true
	return f.err
}

func (f *fakePauseCustomStatusRepository) ListClusterCustomStatus(
	context.Context,
	[]uint64,
	[]tidbmgmtservicev1.CustomStatusType,
	bool,
) (map[uint64][]*entity.ClusterCustomStatus, error) {
	return f.statuses, f.err
}

type fakeTransactioner struct{}

func (fakeTransactioner) Transaction(ctx context.Context, callback func(context.Context) error) error {
	return callback(ctx)
}

func requireResourcePoolMetadata(ctx context.Context, t *testing.T, orgID, resourcePoolID string) {
	t.Helper()
	err := grpcutil.AppendRequestMetaInterceptor(
		ctx,
		"/test.Service/Method",
		nil,
		nil,
		nil,
		func(ctx context.Context, _ string, _, _ any, _ *grpc.ClientConn, _ ...grpc.CallOption) error {
			md, ok := grpcmetadata.FromOutgoingContext(ctx)
			require.True(t, ok)
			require.Equal(t, []string{orgID}, md.Get(requestmetadata.OrgIDHeaderKey))
			require.Equal(t, []string{resourcePoolID}, md.Get(requestmetadata.ResourceIDKey))
			return nil
		},
	)
	require.NoError(t, err)
}

func newRequestContext(t *testing.T) context.Context {
	t.Helper()
	return requestmetadata.AppendToIncomingMeta(t.Context(), &requestmetadata.RequestMeta{
		OrgID:            456,
		UserID:           "user-1",
		DefaultProjectID: 789,
		ValidProjectList: []uint64{789},
	})
}

func TestPauseResourcePoolForwardsRequestMetadata(t *testing.T) {
	ctrl := gomock.NewController(t)
	nextgenCli := mock_nextgenglobal.NewMockNextGenGlobal(ctrl)
	nextgenCli.EXPECT().BeginPauseResourcePoolAdmission(gomock.Any(), gomock.Any()).DoAndReturn(
		func(ctx context.Context, _ *globalngv1.BeginPauseResourcePoolAdmissionRequest, _ ...grpc.CallOption) (*globalngv1.BeginPauseResourcePoolAdmissionResponse, error) {
			requireResourcePoolMetadata(ctx, t, "456", "100")
			return &globalngv1.BeginPauseResourcePoolAdmissionResponse{OperationId: "pause-100"}, nil
		},
	)
	nextgenCli.EXPECT().GetResourcePool(gomock.Any(), gomock.Any()).Return(
		&globalngv1.ResourcePoolProfile{}, nil,
	)
	nextgenCli.EXPECT().ResolvePauseResourcePoolAdmission(gomock.Any(), gomock.Any()).DoAndReturn(
		func(ctx context.Context, _ *globalngv1.ResolvePauseResourcePoolAdmissionRequest, _ ...grpc.CallOption) (*globalngv1.ResolvePauseResourcePoolAdmissionResponse, error) {
			requireResourcePoolMetadata(ctx, t, "456", "100")
			return &globalngv1.ResolvePauseResourcePoolAdmissionResponse{OperationId: "pause-100"}, nil
		},
	)

	ctx := newRequestContext(t)
	_, err := (&serverImpl{nextgenCli: nextgenCli}).PauseResourcePool(ctx, &globalngv1.PauseResourcePoolRequest{
		ResourcePoolId: 100,
		OperationId:    "pause-100",
	})
	require.NoError(t, err)
}

func TestResumeResourcePoolForwardsRequestMetadata(t *testing.T) {
	ctrl := gomock.NewController(t)
	nextgenCli := mock_nextgenglobal.NewMockNextGenGlobal(ctrl)
	nextgenCli.EXPECT().ResumeResourcePool(gomock.Any(), gomock.Any()).DoAndReturn(
		func(ctx context.Context, _ *globalngv1.ResumeResourcePoolRequest, _ ...grpc.CallOption) (*globalngv1.ResumeResourcePoolResponse, error) {
			requireResourcePoolMetadata(ctx, t, "456", "100")
			return &globalngv1.ResumeResourcePoolResponse{}, nil
		},
	)

	ctx := newRequestContext(t)
	_, err := (&serverImpl{nextgenCli: nextgenCli}).ResumeResourcePool(ctx, &globalngv1.ResumeResourcePoolRequest{
		ResourcePoolId: 100,
	})
	require.NoError(t, err)
}

func TestPauseResourcePoolReturnsImportAndChangefeedBlockers(t *testing.T) {
	ctrl := gomock.NewController(t)
	nextgenCli := mock_nextgenglobal.NewMockNextGenGlobal(ctrl)
	nextgenCli.EXPECT().BeginPauseResourcePoolAdmission(gomock.Any(), &globalngv1.BeginPauseResourcePoolAdmissionRequest{
		ResourcePoolId: 100,
		OperationId:    "pause-100",
	}).Return(&globalngv1.BeginPauseResourcePoolAdmissionResponse{OperationId: "pause-100"}, nil)
	nextgenCli.EXPECT().GetResourcePool(gomock.Any(), &globalngv1.GetResourcePoolRequest{
		ResourcePoolId: 100,
	}).Return(&globalngv1.ResourcePoolProfile{
		Instances: []*globalngv1.ResourcePoolProfile_Instance{
			{ClusterId: 1},
			{ClusterId: 2},
		},
	}, nil)
	nextgenCli.EXPECT().ResolvePauseResourcePoolAdmission(gomock.Any(), gomock.Any()).DoAndReturn(
		func(_ context.Context, request *globalngv1.ResolvePauseResourcePoolAdmissionRequest, _ ...grpc.CallOption) (*globalngv1.ResolvePauseResourcePoolAdmissionResponse, error) {
			require.False(t, request.GetProceed())
			require.Len(t, request.GetValidationResults(), 2)
			return &globalngv1.ResolvePauseResourcePoolAdmissionResponse{OperationId: request.GetOperationId()}, nil
		},
	)

	pauseMutex := &tidbmgmtservicev1.ClusterCustomStatus_CustomMutex{
		ClusterAction: []tidbmgmtservicev1.ClusterAction{
			tidbmgmtservicev1.ClusterAction_CLUSTER_ACTION_PAUSE_CLUSTER,
		},
	}
	repo := &fakePauseCustomStatusRepository{statuses: map[uint64][]*entity.ClusterCustomStatus{
		1: {{
			ClusterId:   1,
			StatusType:  tidbmgmtservicev1.CustomStatusType_CUSTOM_STATUS_TYPE_IMPORT,
			CustomMutex: pauseMutex,
		}},
		2: {{
			ClusterId:   2,
			StatusType:  tidbmgmtservicev1.CustomStatusType_CUSTOM_STATUS_TYPE_CDC,
			CustomMutex: pauseMutex,
		}},
	}}
	server := serverImpl{
		nextgenCli:       nextgenCli,
		customStatusRepo: repo,
		transaction:      fakeTransactioner{},
	}

	response, err := server.PauseResourcePool(t.Context(), &globalngv1.PauseResourcePoolRequest{
		ResourcePoolId: 100,
		OperationId:    "pause-100",
	})

	require.NoError(t, err)
	require.True(t, repo.locked)
	require.Equal(t, []*globalngv1.ResourcePoolPauseValidationResult{
		{CheckCode: importExportInProgressCheckCode, Detail: "resource pool has a running import"},
		{CheckCode: changefeedInProgressCheckCode, Detail: "resource pool has a running changefeed"},
	}, response.GetValidationResults())
}

func TestValidateOnlyCombinesCustomStatusAndNextgenResults(t *testing.T) {
	ctrl := gomock.NewController(t)
	nextgenCli := mock_nextgenglobal.NewMockNextGenGlobal(ctrl)
	nextgenCli.EXPECT().GetResourcePool(gomock.Any(), gomock.Any()).Return(&globalngv1.ResourcePoolProfile{
		Instances: []*globalngv1.ResourcePoolProfile_Instance{{ClusterId: 1}},
	}, nil)
	nextgenCli.EXPECT().PauseResourcePool(gomock.Any(), gomock.Any()).Return(&globalngv1.PauseResourcePoolResponse{
		ValidationResults: []*globalngv1.ResourcePoolPauseValidationResult{{
			CheckCode: "INSTANCE_STATE_CONFLICT",
			Detail:    "instance is not active",
		}},
	}, nil)

	server := serverImpl{
		nextgenCli: nextgenCli,
		customStatusRepo: &fakePauseCustomStatusRepository{statuses: map[uint64][]*entity.ClusterCustomStatus{
			1: {{
				ClusterId:  1,
				StatusType: tidbmgmtservicev1.CustomStatusType_CUSTOM_STATUS_TYPE_CDC,
				CustomMutex: &tidbmgmtservicev1.ClusterCustomStatus_CustomMutex{
					ClusterAction: []tidbmgmtservicev1.ClusterAction{
						tidbmgmtservicev1.ClusterAction_CLUSTER_ACTION_PAUSE_CLUSTER,
					},
				},
			}},
		}},
	}

	response, err := server.PauseResourcePool(t.Context(), &globalngv1.PauseResourcePoolRequest{
		ResourcePoolId: 100,
		ValidateOnly:   true,
	})

	require.NoError(t, err)
	require.Equal(t, []*globalngv1.ResourcePoolPauseValidationResult{
		{CheckCode: "INSTANCE_STATE_CONFLICT", Detail: "instance is not active"},
		{CheckCode: changefeedInProgressCheckCode, Detail: "resource pool has a running changefeed"},
	}, response.GetValidationResults())
}

func TestPauseResourcePoolProceedsAfterLockedCustomStatusCheck(t *testing.T) {
	ctrl := gomock.NewController(t)
	nextgenCli := mock_nextgenglobal.NewMockNextGenGlobal(ctrl)
	nextgenCli.EXPECT().BeginPauseResourcePoolAdmission(gomock.Any(), gomock.Any()).Return(
		&globalngv1.BeginPauseResourcePoolAdmissionResponse{OperationId: "pause-100"}, nil,
	)
	nextgenCli.EXPECT().GetResourcePool(gomock.Any(), gomock.Any()).Return(&globalngv1.ResourcePoolProfile{
		Instances: []*globalngv1.ResourcePoolProfile_Instance{{ClusterId: 1}},
	}, nil)
	nextgenCli.EXPECT().ResolvePauseResourcePoolAdmission(gomock.Any(), gomock.Any()).DoAndReturn(
		func(_ context.Context, request *globalngv1.ResolvePauseResourcePoolAdmissionRequest, _ ...grpc.CallOption) (*globalngv1.ResolvePauseResourcePoolAdmissionResponse, error) {
			require.True(t, request.GetProceed())
			return &globalngv1.ResolvePauseResourcePoolAdmissionResponse{OperationId: request.GetOperationId()}, nil
		},
	)
	repo := &fakePauseCustomStatusRepository{}
	server := serverImpl{
		nextgenCli:       nextgenCli,
		customStatusRepo: repo,
		transaction:      fakeTransactioner{},
	}

	response, err := server.PauseResourcePool(t.Context(), &globalngv1.PauseResourcePoolRequest{
		ResourcePoolId: 100,
		OperationId:    "pause-100",
	})

	require.NoError(t, err)
	require.True(t, repo.locked)
	require.Equal(t, "pause-100", response.GetOperationId())
}

func TestPauseResourcePoolReturnsAdmissionCompensationFailure(t *testing.T) {
	t.Parallel()

	ctrl := gomock.NewController(t)
	nextgenCli := mock_nextgenglobal.NewMockNextGenGlobal(ctrl)
	precheckErr := errors.New("cluster discovery failed")
	compensationErr := errors.New("resolve admission failed")
	nextgenCli.EXPECT().BeginPauseResourcePoolAdmission(gomock.Any(), gomock.Any()).Return(
		&globalngv1.BeginPauseResourcePoolAdmissionResponse{OperationId: "pause-100"}, nil,
	)
	nextgenCli.EXPECT().GetResourcePool(gomock.Any(), gomock.Any()).Return(nil, precheckErr)
	nextgenCli.EXPECT().ResolvePauseResourcePoolAdmission(gomock.Any(), gomock.Any()).DoAndReturn(
		func(_ context.Context, request *globalngv1.ResolvePauseResourcePoolAdmissionRequest, _ ...grpc.CallOption) (*globalngv1.ResolvePauseResourcePoolAdmissionResponse, error) {
			require.Equal(t, uint64(100), request.GetResourcePoolId())
			require.Equal(t, "pause-100", request.GetOperationId())
			require.False(t, request.GetProceed())
			return nil, compensationErr
		},
	)
	server := serverImpl{nextgenCli: nextgenCli}

	response, err := server.PauseResourcePool(t.Context(), &globalngv1.PauseResourcePoolRequest{
		ResourcePoolId: 100,
		OperationId:    "pause-100",
	})

	require.Nil(t, response)
	require.ErrorIs(t, err, precheckErr)
	require.ErrorIs(t, err, compensationErr)
	require.ErrorContains(t, err, "resource pool 100")
	require.ErrorContains(t, err, "operation pause-100")
}