package cluster_job_test

import (
	"context"
	"strconv"
	"testing"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	"github.com/stretchr/testify/suite"
	"github.com/tidbcloud/core-svc/idl/pbgen/proto/core"
	"github.com/tidbcloud/dataflow-service/app/internal/cluster_job"
	mock_data "github.com/tidbcloud/dataflow-service/app/internal/cluster_job/mock"
	"github.com/tidbcloud/dataflow-service/app/repo/entity"
	grpcli_mock "github.com/tidbcloud/dataflow-service/pkg/grpc/grpcli/mock"
	"github.com/tidbcloud/pkgv2/bizerr"
	openapicommonv1beta1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_cloud_open_api/common/v1beta1"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
)

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

	clusterJobModel cluster_job.Model
}

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

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

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

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

	// normal case
	err := job.clusterJobModel.RegisterCDCClusterJob(
		ctx, uint64(mock_data.ClusterID),
		cluster_job.GetBizChangefeedTaskID(ctx, uint64(mock_data.TaskID)))
	require.NoError(job.T(), err)

	err = job.clusterJobModel.RegisterBackupClusterJob(
		ctx, uint64(mock_data.ClusterID), uint64(mock_data.TaskID))
	require.NoError(job.T(), err)
}

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

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

	err := job.clusterJobModel.RegisterCDCClusterJob(
		ctx, uint64(mock_data.ClusterID),
		cluster_job.GetBizChangefeedTaskID(ctx, uint64(mock_data.TaskID)))
	require.Error(job.T(), err)

	err = job.clusterJobModel.RegisterBackupClusterJob(
		ctx, uint64(mock_data.ClusterID), uint64(mock_data.TaskID))
	require.Error(job.T(), err)
}

func (job *clusterJobTestSuite) TestDeregisterClusterJob() {
	testcases := []struct {
		name            string
		mockRestoreJobs []*core.ClusterJob
	}{
		{
			name:            "existing restore job is nil",
			mockRestoreJobs: nil,
		},
		{
			name:            "existing restore job is empty",
			mockRestoreJobs: []*core.ClusterJob{},
		},
		{
			name: "existing restore job is more than 1",
			mockRestoreJobs: []*core.ClusterJob{
				{
					ClusterId: 1,
					Name:      "a",
					JobType:   core.ClusterJobType_CLUSTER_JOB_TYPE_RESTORE,
				},
				{
					ClusterId: 1,
					Name:      "b",
					JobType:   core.ClusterJobType_CLUSTER_JOB_TYPE_RESTORE,
				},
			},
		},
		{
			name: "existing restore job is just 1",
			mockRestoreJobs: []*core.ClusterJob{
				{
					ClusterId: 1,
					Name:      "a",
					JobType:   core.ClusterJobType_CLUSTER_JOB_TYPE_RESTORE,
				},
			},
		},
	}

	for _, tc := range testcases {
		job.Run(tc.name, func() {
			listClusterJobsResp := &core.ListClusterJobsResp{
				Jobs: tc.mockRestoreJobs,
			}
			call1 := job.mockCoreGrpcClient.
				On("ListClusterJobs", mock.Anything, mock.Anything).
				Return(listClusterJobsResp, nil)
			defer call1.Unset()

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

			ctx := context.Background()
			err := job.clusterJobModel.DeregisterRestoreClusterJob(
				ctx, uint64(mock_data.ClusterID), entity.RestoreModeSnapshot)
			if len(tc.mockRestoreJobs) <= 1 {
				assert.NoError(job.T(), err)
			} else {
				assert.Error(job.T(), err)
			}
			job.mockCoreGrpcClient.AssertExpectations(job.T())
		})
	}
}

func (job *clusterJobTestSuite) TestRegisterPiTRPauseLock() {
	ctx := context.Background()
	clusterID := uint64(mock_data.ClusterID)
	turn := uint64(7)
	setStatusCallCnt := 0

	setStatusCall := job.mockMgmtSvcCli.
		On("SetClusterCustomStatus", mock.Anything, mock.Anything).
		Run(func(args mock.Arguments) {
			setStatusCallCnt++
			req, ok := args.Get(1).(*mgmt.SetClusterCustomStatusRequest)
			require.True(job.T(), ok)
			require.NotNil(job.T(), req.CustomStatus)
			customStatus := req.CustomStatus
			require.Equal(job.T(), clusterID, customStatus.ClusterId)
			require.Equal(job.T(), mgmt.CustomStatusType_CUSTOM_STATUS_TYPE_PITR, customStatus.StatusType)
			require.Equal(job.T(), "pitr-job-7", customStatus.StatusName)
			require.Nil(job.T(), customStatus.OverrideStatus)
			require.NotNil(job.T(), customStatus.Mutex)
			require.Contains(job.T(), customStatus.Mutex.ClusterAction, mgmt.ClusterAction_CLUSTER_ACTION_PAUSE_CLUSTER)
		}).
		Return(&mgmt.ClusterCustomStatus{}, nil)
	defer setStatusCall.Unset()

	err := job.clusterJobModel.RegisterPiTRPauseLock(ctx, clusterID, turn)
	require.NoError(job.T(), err)
	require.Equal(job.T(), 1, setStatusCallCnt)
}

func (job *clusterJobTestSuite) TestDeregisterPiTRPauseLock() {
	ctx := context.Background()
	clusterID := uint64(mock_data.ClusterID)
	turn := uint64(7)
	queryStatusCallCnt := 0

	queryStatusCall := job.mockMgmtSvcCli.
		On("IsClusterCustomStatusExisted", mock.Anything, mock.Anything).
		Run(func(args mock.Arguments) {
			queryStatusCallCnt++
			req, ok := args.Get(1).(*mgmt.IsClusterCustomStatusExistedRequest)
			require.True(job.T(), ok)
			require.Equal(job.T(), clusterID, req.ClusterId)
			require.Equal(job.T(), mgmt.CustomStatusType_CUSTOM_STATUS_TYPE_PITR, req.StatusType)
		}).
		Return(&mgmt.IsClusterCustomStatusExistedResponse{
			IsExisted: true,
			ExistedCustomStatuses: []*mgmt.ClusterCustomStatus{
				{
					StatusId:   1,
					StatusName: "snapshot-backup",
				},
				{
					StatusId:   2,
					StatusName: "pitr-job-7",
				},
			},
		}, nil)
	defer queryStatusCall.Unset()

	deleteStatusCallCnt := 0
	deleteStatusCall := job.mockMgmtSvcCli.
		On("DeleteClusterCustomStatus", mock.Anything, mock.Anything).
		Run(func(args mock.Arguments) {
			deleteStatusCallCnt++
			req, ok := args.Get(1).(*mgmt.DeleteClusterCustomStatusRequest)
			require.True(job.T(), ok)
			require.Equal(job.T(), clusterID, req.ClusterId)
			require.Equal(job.T(), uint64(2), req.StatusId)
		}).
		Return(nil)
	defer deleteStatusCall.Unset()

	err := job.clusterJobModel.DeregisterPiTRPauseLock(ctx, clusterID, turn)
	require.NoError(job.T(), err)
	require.Equal(job.T(), 1, queryStatusCallCnt)
	require.Equal(job.T(), 1, deleteStatusCallCnt)
}

func (job *clusterJobTestSuite) TestRegisterRestoreClusterV2() {
	ctx := context.Background()
	clusterID := uint64(mock_data.ClusterID)
	taskID := uint64(mock_data.TaskID)
	setStatusCallCnt := 0

	setStatusCall := job.mockMgmtSvcCli.
		On("SetClusterCustomStatus", mock.Anything, mock.Anything).
		Run(func(args mock.Arguments) {
			setStatusCallCnt++
			req, ok := args.Get(1).(*mgmt.SetClusterCustomStatusRequest)
			require.True(job.T(), ok)
			require.NotNil(job.T(), req.CustomStatus)
			customStatus := req.CustomStatus
			require.Equal(job.T(), clusterID, customStatus.ClusterId)
			require.Equal(job.T(), mgmt.CustomStatusType_CUSTOM_STATUS_TYPE_RESTORE, customStatus.StatusType)
			require.Equal(job.T(), strconv.FormatUint(taskID, 10), customStatus.StatusName)
			require.NotNil(job.T(), customStatus.OverrideStatus)
			require.Equal(job.T(), openapicommonv1beta1.ClusterState_RESTORING, *customStatus.OverrideStatus)
			require.NotNil(job.T(), customStatus.Mutex)
			require.Contains(job.T(), customStatus.Mutex.ClusterAction, mgmt.ClusterAction_CLUSTER_ACTION_PAUSE_CLUSTER)
		}).
		Return(&mgmt.ClusterCustomStatus{}, nil)
	defer setStatusCall.Unset()

	err := job.clusterJobModel.RegisterRestoreClusterV2(ctx, clusterID, taskID)
	require.NoError(job.T(), err)
	require.Equal(job.T(), 1, setStatusCallCnt)
}

func (job *clusterJobTestSuite) TestDeregisterRestoreClusterJobV2() {
	ctx := context.Background()
	clusterID := uint64(mock_data.ClusterID)
	taskID := uint64(mock_data.TaskID)
	queryStatusCallCnt := 0

	queryStatusCall := job.mockMgmtSvcCli.
		On("IsClusterCustomStatusExisted", mock.Anything, mock.Anything).
		Run(func(args mock.Arguments) {
			queryStatusCallCnt++
			req, ok := args.Get(1).(*mgmt.IsClusterCustomStatusExistedRequest)
			require.True(job.T(), ok)
			require.Equal(job.T(), clusterID, req.ClusterId)
			require.Equal(job.T(), mgmt.CustomStatusType_CUSTOM_STATUS_TYPE_RESTORE, req.StatusType)
		}).
		Return(&mgmt.IsClusterCustomStatusExistedResponse{
			IsExisted: true,
			ExistedCustomStatuses: []*mgmt.ClusterCustomStatus{
				{
					StatusId:   1,
					StatusName: "snapshot-backup",
				},
				{
					StatusId:   2,
					StatusName: strconv.FormatUint(taskID, 10),
				},
			},
		}, nil)
	defer queryStatusCall.Unset()

	deleteStatusCallCnt := 0
	deleteStatusCall := job.mockMgmtSvcCli.
		On("DeleteClusterCustomStatus", mock.Anything, mock.Anything).
		Run(func(args mock.Arguments) {
			deleteStatusCallCnt++
			req, ok := args.Get(1).(*mgmt.DeleteClusterCustomStatusRequest)
			require.True(job.T(), ok)
			require.Equal(job.T(), clusterID, req.ClusterId)
			require.Equal(job.T(), uint64(2), req.StatusId)
		}).
		Return(nil)
	defer deleteStatusCall.Unset()

	err := job.clusterJobModel.DeregisterRestoreClusterJobV2(ctx, clusterID, taskID)
	require.NoError(job.T(), err)
	require.Equal(job.T(), 1, queryStatusCallCnt)
	require.Equal(job.T(), 1, deleteStatusCallCnt)
}

func (job *clusterJobTestSuite) TestIsRegisterClusterJob() {
	ctx := context.Background()
	clusterJobResp := &core.IsClusterJobRegisteredResp{IsRegistered: true}

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

	// normal case
	isRegister := job.clusterJobModel.IsBackupClusterJobRegistered(
		ctx, uint64(mock_data.ClusterID), uint64(mock_data.TaskID))
	require.Equal(job.T(), true, isRegister)
}

func (job *clusterJobTestSuite) TestIsPiTRPauseLockRegistered() {
	ctx := context.Background()
	clusterID := uint64(mock_data.ClusterID)
	turn := uint64(7)

	testcases := []struct {
		name     string
		response *mgmt.IsClusterCustomStatusExistedResponse
		existed  bool
	}{
		{
			name: "lock exists",
			response: &mgmt.IsClusterCustomStatusExistedResponse{
				IsExisted: true,
				ExistedCustomStatuses: []*mgmt.ClusterCustomStatus{
					{StatusName: "pitr-job-7"},
				},
			},
			existed: true,
		},
		{
			name: "lock not found",
			response: &mgmt.IsClusterCustomStatusExistedResponse{
				IsExisted: true,
				ExistedCustomStatuses: []*mgmt.ClusterCustomStatus{
					{StatusName: "other"},
				},
			},
			existed: false,
		},
		{
			name: "status type absent",
			response: &mgmt.IsClusterCustomStatusExistedResponse{
				IsExisted: false,
			},
			existed: false,
		},
	}

	for _, tc := range testcases {
		job.Run(tc.name, func() {
			queryCall := job.mockMgmtSvcCli.
				On("IsClusterCustomStatusExisted", mock.Anything, mock.Anything).
				Run(func(args mock.Arguments) {
					req, ok := args.Get(1).(*mgmt.IsClusterCustomStatusExistedRequest)
					require.True(job.T(), ok)
					require.Equal(job.T(), clusterID, req.ClusterId)
					require.Equal(job.T(), mgmt.CustomStatusType_CUSTOM_STATUS_TYPE_PITR, req.StatusType)
				}).
				Return(tc.response, nil)
			defer queryCall.Unset()

			existed, err := job.clusterJobModel.IsPiTRPauseLockRegistered(ctx, clusterID, turn)
			require.NoError(job.T(), err)
			require.Equal(job.T(), tc.existed, existed)
		})
	}
}

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