package service

import (
	"context"
	"testing"

	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	"github.com/tidbcloud/serverless-service/internal/domain/facade/mock_facade"
	"github.com/tidbcloud/serverless-service/internal/domain/repo/mock_repo"
	"github.com/tidbcloud/serverless-service/internal/domain/service/mock_service"
	"go.uber.org/mock/gomock"
)

const testRegisterClusterID = "10194411360446690844"

func testRegisterClusterIDUint() uint64 { return 10194411360446690844 }

// happy path: registration succeeds on the first try, no healing.
func TestRegisterCDCClusterJob_Success(t *testing.T) {
	ctrl := gomock.NewController(t)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)

	mgmtSvc.EXPECT().
		RegisterCDCClusterJob(gomock.Any(), testRegisterClusterIDUint(), "cf-1").
		Return(nil)

	hook := &cdcNewArchWorkflowHook{mgmtSvc: mgmtSvc}
	err := hook.RegisterCDCClusterJob(context.Background(), testRegisterClusterID, "cf-1")
	require.NoError(t, err)
}

// forbidden + legacy lock present: heal, repoint old-arch bookkeeping, then retry succeeds.
func TestRegisterCDCClusterJob_HealsLegacyLockAndRetries(t *testing.T) {
	ctrl := gomock.NewController(t)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)
	cdcClusterRepo := mock_repo.NewMockCDCClusterRepo(ctrl)

	forbidden := bizerr.New(bizerr.ErrCodePermission, "Forbidden. The cluster is in the CDC state")

	gomock.InOrder(
		mgmtSvc.EXPECT().
			RegisterCDCClusterJob(gomock.Any(), testRegisterClusterIDUint(), "cf-1").
			Return(forbidden),
		mgmtSvc.EXPECT().
			ReplaceLegacyCDCClusterLock(gomock.Any(), testRegisterClusterIDUint()).
			Return(true, uint64(100), uint64(200), nil),
		mgmtSvc.EXPECT().
			RegisterCDCClusterJob(gomock.Any(), testRegisterClusterIDUint(), "cf-1").
			Return(nil),
	)

	// old-arch record still points at the legacy lock id -> must be repointed.
	cdcClusterRepo.EXPECT().
		GetById(gomock.Any(), testRegisterClusterID).
		Return(&entity.CDCCluster{ClusterID: testRegisterClusterID, ClusterStatusId: 100}, nil)
	cdcClusterRepo.EXPECT().
		SetClusterStatusId(gomock.Any(), testRegisterClusterID, uint64(200)).
		Return(nil)

	hook := &cdcNewArchWorkflowHook{mgmtSvc: mgmtSvc, cdcClusterRepo: cdcClusterRepo}
	err := hook.RegisterCDCClusterJob(context.Background(), testRegisterClusterID, "cf-1")
	require.NoError(t, err)
}

// forbidden but no legacy lock found: surface the original forbidden error, do not heal.
func TestRegisterCDCClusterJob_ForbiddenWithoutLegacyLock(t *testing.T) {
	ctrl := gomock.NewController(t)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)

	forbidden := bizerr.New(bizerr.ErrCodePermission, "Forbidden. The cluster is in the IMPORT state")

	gomock.InOrder(
		mgmtSvc.EXPECT().
			RegisterCDCClusterJob(gomock.Any(), testRegisterClusterIDUint(), "cf-1").
			Return(forbidden),
		mgmtSvc.EXPECT().
			ReplaceLegacyCDCClusterLock(gomock.Any(), testRegisterClusterIDUint()).
			Return(false, uint64(0), uint64(0), nil),
	)

	hook := &cdcNewArchWorkflowHook{mgmtSvc: mgmtSvc}
	err := hook.RegisterCDCClusterJob(context.Background(), testRegisterClusterID, "cf-1")
	require.Error(t, err)
	require.Equal(t, bizerr.ErrCodePermission, err.(bizerr.BizError).ErrCode())
}

// non-permission error: returned as-is without attempting to heal.
func TestRegisterCDCClusterJob_NonPermissionErrorNotHealed(t *testing.T) {
	ctrl := gomock.NewController(t)
	mgmtSvc := mock_facade.NewMockMgmtSvcFacade(ctrl)

	other := bizerr.New(bizerr.ErrCodeUnknown, "boom")
	mgmtSvc.EXPECT().
		RegisterCDCClusterJob(gomock.Any(), testRegisterClusterIDUint(), "cf-1").
		Return(other)

	hook := &cdcNewArchWorkflowHook{mgmtSvc: mgmtSvc}
	err := hook.RegisterCDCClusterJob(context.Background(), testRegisterClusterID, "cf-1")
	require.Error(t, err)
	require.Equal(t, bizerr.ErrCodeUnknown, err.(bizerr.BizError).RawErrCode())
}

func TestNewCDCNewArchWorkflowHookWiresPrivateLinkService(t *testing.T) {
	ctrl := gomock.NewController(t)
	privateLinkSvc := mock_service.NewMockPrivateLinkConnectionService(ctrl)

	hook := NewCDCNewArchWorkflowHook(
		nil,
		nil,
		nil,
		nil,
		nil,
		nil,
		nil,
		nil,
		nil,
		privateLinkSvc,
		nil,
	).(*cdcNewArchWorkflowHook)

	require.Same(t, privateLinkSvc, hook.newCDCService().privateLinkConnectionSvc)
}