package cluster_job_test

import (
	"context"
	"testing"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	"github.com/stretchr/testify/suite"
	"github.com/tidbcloud/dataflow-service-ng/app/internal/cluster_job"
	grpcli_mock "github.com/tidbcloud/dataflow-service-ng/pkg/grpc/grpcli/mock"
	"github.com/tidbcloud/pkgv2/bizerr"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
)

const (
	clusterID = 1234
	taskID    = 4567
)

type clusterJobTestSuite struct {
	suite.Suite
	mockCoreGrpcClient *grpcli_mock.MgmtSvcGrpcClient

	clusterJobModel cluster_job.Model
}

func (job *clusterJobTestSuite) SetupSuite() {
	mockCoreGrpcClient := grpcli_mock.NewMgmtSvcGrpcClient(job.T())

	job.mockCoreGrpcClient = mockCoreGrpcClient
	job.clusterJobModel = cluster_job.NewClusterJobModelImpl(mockCoreGrpcClient)
}

func (job *clusterJobTestSuite) TestRegisterClusterJob() {
	ctx := context.Background()
	clusterJobResp := &mgmt.ClusterCustomStatus{}

	createClusterJobCall := job.mockCoreGrpcClient.
		On("SetClusterCustomStatus", mock.Anything, mock.Anything).
		Return(clusterJobResp, nil)
	defer createClusterJobCall.Unset()

	// normal case
	err := job.clusterJobModel.RegisterCDCClusterJob(
		ctx, uint64(clusterID), uint64(taskID))
	require.NoError(job.T(), err)
}

func (job *clusterJobTestSuite) TestRegisterClusterJobGrpcError() {
	ctx := context.Background()
	// abnormal case
	// grpc error
	clusterJobResp := &mgmt.ClusterCustomStatus{}

	createClusterJobCall := job.mockCoreGrpcClient.
		On("SetClusterCustomStatus", mock.Anything, mock.Anything).
		Return(clusterJobResp, bizerr.New(0, "grpc error"))
	defer createClusterJobCall.Unset()

	err := job.clusterJobModel.RegisterCDCClusterJob(
		ctx, uint64(clusterID), uint64(taskID))
	require.Error(job.T(), err)
}

func (job *clusterJobTestSuite) TestRegisterImportClusterJob() {
	job.mockCoreGrpcClient.
		On("SetClusterCustomStatus", mock.Anything, mock.MatchedBy(func(req *mgmt.SetClusterCustomStatusRequest) bool {
			status := req.GetCustomStatus()
			return status.GetClusterId() == clusterID &&
				status.GetStatusName() == "4567" &&
				status.GetStatusType() == mgmt.CustomStatusType_CUSTOM_STATUS_TYPE_IMPORT &&
				assert.Contains(job.T(), status.GetMutex().GetClusterAction(), mgmt.ClusterAction_CLUSTER_ACTION_PAUSE_CLUSTER)
		})).
		Return(&mgmt.ClusterCustomStatus{}, nil).
		Once()

	err := job.clusterJobModel.RegisterImportClusterJob(context.Background(), clusterID, taskID)
	require.NoError(job.T(), err)
}

func (job *clusterJobTestSuite) TestDeregisterImportClusterJob() {
	statusID := uint64(9876)
	job.mockCoreGrpcClient.
		On("IsClusterCustomStatusExisted", mock.Anything, mock.MatchedBy(func(req *mgmt.IsClusterCustomStatusExistedRequest) bool {
			return req.GetClusterId() == clusterID && req.GetStatusType() == mgmt.CustomStatusType_CUSTOM_STATUS_TYPE_IMPORT
		})).
		Return(&mgmt.IsClusterCustomStatusExistedResponse{
			IsExisted: true,
			ExistedCustomStatuses: []*mgmt.ClusterCustomStatus{{
				StatusId:   statusID,
				StatusName: "4567",
			}},
		}, nil).
		Once()
	job.mockCoreGrpcClient.
		On("DeleteClusterCustomStatus", mock.Anything, mock.MatchedBy(func(req *mgmt.DeleteClusterCustomStatusRequest) bool {
			return req.GetClusterId() == clusterID && req.GetStatusId() == statusID
		})).
		Return(nil).
		Once()

	err := job.clusterJobModel.DeregisterImportClusterJob(context.Background(), clusterID, taskID)
	require.NoError(job.T(), err)
}

func TestEnsureImportClusterJobAlreadyExists(t *testing.T) {
	mgmtClient := grpcli_mock.NewMgmtSvcGrpcClient(t)
	mgmtClient.On("IsClusterCustomStatusExisted", mock.Anything, mock.Anything).
		Return(&mgmt.IsClusterCustomStatusExistedResponse{
			IsExisted: true,
			ExistedCustomStatuses: []*mgmt.ClusterCustomStatus{{
				StatusName: "4567",
			}},
		}, nil).
		Once()

	err := cluster_job.NewClusterJobModelImpl(mgmtClient).
		EnsureImportClusterJob(context.Background(), clusterID, taskID)
	require.NoError(t, err)
	mgmtClient.AssertNotCalled(t, "SetClusterCustomStatus", mock.Anything, mock.Anything)
}

func TestEnsureImportClusterJobRecoversConcurrentRegistration(t *testing.T) {
	mgmtClient := grpcli_mock.NewMgmtSvcGrpcClient(t)
	mgmtClient.On("IsClusterCustomStatusExisted", mock.Anything, mock.Anything).
		Return(&mgmt.IsClusterCustomStatusExistedResponse{}, nil).
		Once()
	mgmtClient.On("SetClusterCustomStatus", mock.Anything, mock.Anything).
		Return((*mgmt.ClusterCustomStatus)(nil), bizerr.New(bizerr.ErrCodeGRPC, "duplicate or lost response")).
		Once()
	mgmtClient.On("IsClusterCustomStatusExisted", mock.Anything, mock.Anything).
		Return(&mgmt.IsClusterCustomStatusExistedResponse{
			IsExisted: true,
			ExistedCustomStatuses: []*mgmt.ClusterCustomStatus{{
				StatusName: "4567",
			}},
		}, nil).
		Once()

	err := cluster_job.NewClusterJobModelImpl(mgmtClient).
		EnsureImportClusterJob(context.Background(), clusterID, taskID)
	require.NoError(t, err)
}

func TestEnsureImportClusterJobReturnsRegistrationErrorWhenStillAbsent(t *testing.T) {
	mgmtClient := grpcli_mock.NewMgmtSvcGrpcClient(t)
	mgmtClient.On("IsClusterCustomStatusExisted", mock.Anything, mock.Anything).
		Return(&mgmt.IsClusterCustomStatusExistedResponse{}, nil).
		Twice()
	registerErr := bizerr.New(bizerr.ErrCodeGRPC, "resource pool is pausing")
	mgmtClient.On("SetClusterCustomStatus", mock.Anything, mock.Anything).
		Return((*mgmt.ClusterCustomStatus)(nil), registerErr).
		Once()

	err := cluster_job.NewClusterJobModelImpl(mgmtClient).
		EnsureImportClusterJob(context.Background(), clusterID, taskID)
	require.Error(t, err)
	require.Equal(t, registerErr.RawErrCode(), err.RawErrCode())
	require.Contains(t, err.Error(), "resource pool is pausing")
}

func (job *clusterJobTestSuite) TestDeregisterClusterJob() {
	testcases := []struct {
		name        string
		mockCDCJobs []*mgmt.ClusterCustomStatus
	}{
		{
			name:        "existing cdc job is nil",
			mockCDCJobs: nil,
		},
		{
			name:        "existing cdc job is empty",
			mockCDCJobs: []*mgmt.ClusterCustomStatus{},
		},
		{
			name: "existing cdc job is more than 1",
			mockCDCJobs: []*mgmt.ClusterCustomStatus{
				{
					ClusterId:  1,
					StatusName: "1",
					StatusType: mgmt.CustomStatusType_CUSTOM_STATUS_TYPE_CDC,
				},
				{
					ClusterId:  1,
					StatusName: "b",
					StatusType: mgmt.CustomStatusType_CUSTOM_STATUS_TYPE_CDC,
				},
			},
		},
		{
			name: "existing cdc job is just 1",
			mockCDCJobs: []*mgmt.ClusterCustomStatus{
				{
					ClusterId:  1,
					StatusName: "1",
					StatusType: mgmt.CustomStatusType_CUSTOM_STATUS_TYPE_CDC,
				},
			},
		},
	}

	for _, tc := range testcases {
		job.Run(tc.name, func() {
			listClusterJobsResp := &mgmt.IsClusterCustomStatusExistedResponse{
				IsExisted:             len(tc.mockCDCJobs) > 0,
				ExistedCustomStatuses: tc.mockCDCJobs,
			}
			call1 := job.mockCoreGrpcClient.
				On("IsClusterCustomStatusExisted", mock.Anything, mock.Anything).
				Return(listClusterJobsResp, nil)
			defer call1.Unset()

			if len(tc.mockCDCJobs) == 1 {
				call2 := job.mockCoreGrpcClient.
					On("DeleteClusterJob", mock.Anything, mock.Anything).
					Return(nil, nil)
				defer call2.Unset()
			}

			ctx := context.Background()
			err := job.clusterJobModel.DeregisterCDCClusterJob(
				ctx, uint64(clusterID), uint64(taskID))
			assert.NoError(job.T(), err)
		})
	}
}

func TestClusterJobTestSuite(t *testing.T) {
	suite.Run(t, new(clusterJobTestSuite))
}