package impl

import (
	"context"
	"testing"

	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/suite"
	ticdc_v1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/ticdc-service/v1"
	"github.com/tidbcloud/dataflow-service-ng/app/models/domain"
	"github.com/tidbcloud/dataflow-service-ng/app/repo/entity"
	repo_mock "github.com/tidbcloud/dataflow-service-ng/app/repo/mock"
	grpcli_mock "github.com/tidbcloud/dataflow-service-ng/pkg/grpc/grpcli/mock"
	regionalapi "github.com/tidbcloud/dataflow-service-ng/pkg/regional-api"
	regionalapi_mock "github.com/tidbcloud/dataflow-service-ng/pkg/regional-api/mock"
	"github.com/tidbcloud/pkgv2/bizerr"
	v12 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
)

type sinkEndpointTestSuite struct {
	suite.Suite
	ctx                   context.Context
	model                 *sinkEndpointImpl
	mockBizPLEndpointRepo *repo_mock.BizSinkEndpointRepo
	mockMgmtCli           *grpcli_mock.MgmtSvcGrpcClient
	mockRegionalCli       *regionalapi_mock.TiCDCPrivateLinkClient
}

func TestSinkEndpointTestSuite(t *testing.T) {
	suite.Run(t, new(sinkEndpointTestSuite))
}

func (s *sinkEndpointTestSuite) SetupTest() {
	s.ctx = context.Background()
	s.mockBizPLEndpointRepo = repo_mock.NewBizSinkEndpointRepo(s.T())
	s.mockMgmtCli = grpcli_mock.NewMgmtSvcGrpcClient(s.T())
	s.mockRegionalCli = regionalapi_mock.NewTiCDCPrivateLinkClient(s.T())
	s.model = &sinkEndpointImpl{
		bizPLEndpointRepo: s.mockBizPLEndpointRepo,
		mgmtCli:           s.mockMgmtCli,
		cli:               s.mockRegionalCli,
	}
}

func (s *sinkEndpointTestSuite) TestValidateRootDomainName() {
	s.Run("conflict across tenants in same sharedpool", func() {
		s.mockBizPLEndpointRepo.
			On("List", mock.Anything, mock.Anything).
			Return([]entity.BizSinkEndpoint{
				{
					TenantID:       2001,
					SharedPoolID:   "sp-1",
					RootDomainName: "example.com.",
				},
			}, nil).Once()

		err := s.model.validateRootDomainName(s.ctx, "sp-1", "example.com.")
		s.Error(err)
		s.Contains(err.Error(), "conflict")
	})

	s.Run("same domain in another sharedpool does not conflict", func() {
		s.mockBizPLEndpointRepo.
			On("List", mock.Anything, mock.Anything).
			Return([]entity.BizSinkEndpoint{}, nil).Once()

		err := s.model.validateRootDomainName(s.ctx, "sp-2", "example.com.")
		s.NoError(err)
	})
}

func (s *sinkEndpointTestSuite) TestQuerySinkEndpointWithClusterScope() {
	const clusterID = uint64(100)
	const tenantID = uint64(200)
	const endpointID = uint64(300)
	cluster := &mgmt.Cluster{
		ClusterId:      clusterID,
		OrganizationId: tenantID,
		ClusterInfo: &mgmt.Cluster_NextGenCluster{
			NextGenCluster: &v12.Cluster{SharedPoolId: "sp-1"},
		},
	}

	s.Run("scoped query returns matching endpoint", func() {
		s.mockMgmtCli.
			On("GetClusterByID", mock.Anything, clusterID).
			Return(cluster, nil).Once()
		s.mockBizPLEndpointRepo.
			On("List", mock.Anything, mock.Anything).
			Return([]entity.BizSinkEndpoint{{
				Updatable:    entity.Updatable{Base: entity.Base{ID: entity.ID(endpointID)}},
				TenantID:     entity.ID(tenantID),
				SharedPoolID: "sp-1",
			}}, nil).Once()

		resp, err := s.model.QuerySinkEndpoint(s.ctx, &domain.QuerySinkEndpointReq{
			ID:        endpointID,
			ClusterID: clusterID,
		})
		s.NoError(err)
		s.Equal(endpointID, resp.SinkEndpoint.ID.Uint64())
	})

	s.Run("scoped query returns not found when repo has no matching endpoint", func() {
		s.mockMgmtCli.
			On("GetClusterByID", mock.Anything, clusterID).
			Return(cluster, nil).Once()
		s.mockBizPLEndpointRepo.
			On("List", mock.Anything, mock.Anything).
			Return([]entity.BizSinkEndpoint{}, nil).Once()

		_, err := s.model.QuerySinkEndpoint(s.ctx, &domain.QuerySinkEndpointReq{
			ID:        endpointID,
			ClusterID: clusterID,
		})
		s.Error(err)
		s.Equal(bizerr.ErrCodeRecordNotFound, err.RawErrCode())
	})

	s.Run("query requires cluster id", func() {
		_, err := s.model.QuerySinkEndpoint(s.ctx, &domain.QuerySinkEndpointReq{
			ID: endpointID,
		})
		s.Error(err)
		s.Equal(bizerr.ErrCodeParams, err.RawErrCode())
	})
}

func (s *sinkEndpointTestSuite) TestDescribeEndpointServiceAvailabilityZones() {
	const clusterID = uint64(100)
	cluster := &mgmt.Cluster{
		ClusterId:      clusterID,
		CloudProvider:  string(entity.CloudProviderTypeAWS),
		OrganizationId: 200,
		ClusterInfo: &mgmt.Cluster_NextGenCluster{
			NextGenCluster: &v12.Cluster{SharedPoolId: "sp-1"},
		},
	}

	s.Run("success", func() {
		s.mockMgmtCli.
			On("GetClusterByID", mock.Anything, clusterID).
			Return(cluster, nil).Once()
		s.mockRegionalCli.
			On("DescribeEndpointServiceAvailabilityZones",
				mock.Anything,
				mock.MatchedBy(func(region *regionalapi.Region) bool {
					return region.SharedPoolID == "sp-1"
				}),
				mock.MatchedBy(func(req *ticdc_v1.DescribeEndpointServiceAvailabilityZonesRequest) bool {
					return req.SharedPoolId == "sp-1" &&
						req.ServiceName == "com.amazonaws.vpce.us-east-1.vpce-svc-123" &&
						req.ServiceRegion == "us-east-1"
				})).
			Return(&ticdc_v1.DescribeEndpointServiceAvailabilityZonesResponse{
				Zones: []*ticdc_v1.EndpointServiceAvailabilityZone{
					{AvailabilityZone: "us-east-1b", AvailabilityZoneId: "use1-az1"},
					{AvailabilityZone: "us-east-1c", AvailabilityZoneId: "use1-az2"},
				},
			}, nil).Once()

		resp, err := s.model.DescribeEndpointServiceAvailabilityZones(s.ctx,
			&domain.DescribeEndpointServiceAvailabilityZonesReq{
				ClusterID:     clusterID,
				ServiceName:   "com.amazonaws.vpce.us-east-1.vpce-svc-123",
				ServiceRegion: "us-east-1",
			})

		s.NoError(err)
		s.Equal([]domain.EndpointServiceAvailabilityZone{
			{AvailabilityZone: "us-east-1b", AvailabilityZoneID: "use1-az1"},
			{AvailabilityZone: "us-east-1c", AvailabilityZoneID: "use1-az2"},
		}, resp.Zones)
	})

	s.Run("regional error is propagated", func() {
		s.mockMgmtCli.
			On("GetClusterByID", mock.Anything, clusterID).
			Return(cluster, nil).Once()
		s.mockRegionalCli.
			On("DescribeEndpointServiceAvailabilityZones", mock.Anything, mock.Anything, mock.Anything).
			Return(nil, bizerr.New(bizerr.ErrCodeParams, "cross-region endpoint service availability zone discovery is not supported")).Once()

		_, err := s.model.DescribeEndpointServiceAvailabilityZones(s.ctx,
			&domain.DescribeEndpointServiceAvailabilityZonesReq{
				ClusterID:     clusterID,
				ServiceName:   "com.amazonaws.vpce.us-west-2.vpce-svc-123",
				ServiceRegion: "us-west-2",
			})

		s.Error(err)
		s.Equal(bizerr.ErrCodeParams, err.RawErrCode())
		s.Contains(err.ErrMsg(), "cross-region")
	})

	s.Run("normalizes AWS logical service region", func() {
		s.mockMgmtCli.
			On("GetClusterByID", mock.Anything, clusterID).
			Return(cluster, nil).Once()
		s.mockRegionalCli.
			On("DescribeEndpointServiceAvailabilityZones",
				mock.Anything,
				mock.Anything,
				mock.MatchedBy(func(req *ticdc_v1.DescribeEndpointServiceAvailabilityZonesRequest) bool {
					return req.ServiceRegion == "us-east-1"
				})).
			Return(&ticdc_v1.DescribeEndpointServiceAvailabilityZonesResponse{}, nil).Once()

		_, err := s.model.DescribeEndpointServiceAvailabilityZones(s.ctx,
			&domain.DescribeEndpointServiceAvailabilityZonesReq{
				ClusterID:     clusterID,
				ServiceName:   "com.amazonaws.vpce.us-east-1.vpce-svc-123",
				ServiceRegion: "aws-us-east-1",
			})

		s.NoError(err)
	})

	s.Run("normalizes AliCloud logical service region", func() {
		aliCloudCluster := &mgmt.Cluster{
			ClusterId:      clusterID,
			CloudProvider:  string(entity.CloudProviderTypeAliCloud),
			OrganizationId: 200,
			ClusterInfo: &mgmt.Cluster_NextGenCluster{
				NextGenCluster: &v12.Cluster{SharedPoolId: "sp-1"},
			},
		}
		s.mockMgmtCli.
			On("GetClusterByID", mock.Anything, clusterID).
			Return(aliCloudCluster, nil).Once()
		s.mockRegionalCli.
			On("DescribeEndpointServiceAvailabilityZones",
				mock.Anything,
				mock.Anything,
				mock.MatchedBy(func(req *ticdc_v1.DescribeEndpointServiceAvailabilityZonesRequest) bool {
					return req.ServiceRegion == "cn-hongkong"
				})).
			Return(&ticdc_v1.DescribeEndpointServiceAvailabilityZonesResponse{}, nil).Once()

		_, err := s.model.DescribeEndpointServiceAvailabilityZones(s.ctx,
			&domain.DescribeEndpointServiceAvailabilityZonesReq{
				ClusterID:     clusterID,
				ServiceName:   "com.aliyuncs.privatelink.cn-hongkong.epsrv-example",
				ServiceRegion: "alicloud-cn-hongkong",
			})

		s.NoError(err)
	})
}