package console

import (
	"context"
	"testing"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/dataflow-service-ng/app/models/domain"
	sink_endpoint_mock "github.com/tidbcloud/dataflow-service-ng/app/models/sink_endpoint/mock"
	"github.com/tidbcloud/dataflow-service-ng/app/repo/entity"
	"github.com/tidbcloud/dataflow-service-ng/app/utils"
	dataflow "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	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"
)

func TestMSKChangefeedAuthType(t *testing.T) {
	t.Run("sasl/scram maps to scram-sha-512", func(t *testing.T) {
		got, err := mskChangefeedAuthType(entity.BizSinkEndpointAuthSASLScram)
		require.NoError(t, err)
		assert.Equal(t, dataflow.KafkaAuthType_SASL_SCRAM_SHA_512, got)
	})

	t.Run("unsupported auth type is rejected", func(t *testing.T) {
		_, err := mskChangefeedAuthType(entity.BizSinkEndpointAuthSASLIAM)
		require.Error(t, err)
	})

	t.Run("empty auth type is rejected", func(t *testing.T) {
		_, err := mskChangefeedAuthType("")
		require.Error(t, err)
	})
}

func TestKafkaBizChangefeedDownstreamAdaptor_MSKPrivateLink(t *testing.T) {
	endpointID := uint64(456)
	cluster := utils.NewCluster(&mgmt.Cluster{
		ClusterId: 123,
		ClusterInfo: &mgmt.Cluster_NextGenCluster{
			NextGenCluster: &v12.Cluster{},
		},
	})

	newReq := func() *dataflow.ChangefeedCreateReq {
		return &dataflow.ChangefeedCreateReq{
			NetworkInfo: &dataflow.NetworkInfo{
				NetworkType:    dataflow.NetworkType_NetworkType_Private_Link,
				SinkEndpointId: &endpointID,
			},
			KafkaInfo: &dataflow.KafkaInfo{
				Broker: &dataflow.Broker{
					BrokerEndpoints: "user-supplied:9092",
				},
				Authentication: &dataflow.Authentication{
					AuthType: dataflow.KafkaAuthType_DISABLE,
					UserName: "alice",
					Password: "secret",
				},
			},
		}
	}

	setupAdaptor := func(t *testing.T, endpoint *entity.BizSinkEndpoint) BizChangefeedDownstreamAdaptor {
		t.Helper()
		sinkEndpointModel := sink_endpoint_mock.NewSinkEndpointModel(t)
		queryMatcher := mock.MatchedBy(func(req *domain.QuerySinkEndpointReq) bool {
			return req.ID == endpointID && req.ClusterID == cluster.GetClusterID()
		})
		sinkEndpointModel.
			On("QuerySinkEndpoint", mock.Anything, queryMatcher).
			Return(&domain.QuerySinkEndpointResp{SinkEndpoint: endpoint}, nil).
			Once()
		return newKafkaBizChangefeedDownstreamAdaptor(sinkEndpointModel)
	}

	t.Run("overwrites broker and derives auth from endpoint", func(t *testing.T) {
		req := newReq()
		adaptor := setupAdaptor(t, &entity.BizSinkEndpoint{
			Status:          entity.BizPLEndpointStatusAvailable,
			EndpointType:    entity.BizSinkEndpointTypeAmazonMSK,
			AuthType:        entity.BizSinkEndpointAuthSASLScram,
			BootstrapBroker: "b-1.demo:9196,b-2.demo:9196",
		})

		params, err := adaptor.ParseCreateParams(context.Background(), req, cluster)
		require.NoError(t, err)
		assert.Equal(t, req.KafkaInfo, params)
		assert.Equal(t, "b-1.demo:9196,b-2.demo:9196", req.KafkaInfo.Broker.BrokerEndpoints)
		assert.Equal(t, dataflow.KafkaAuthType_SASL_SCRAM_SHA_512, req.KafkaInfo.Authentication.AuthType)
	})

	t.Run("requires bootstrap broker to be ready", func(t *testing.T) {
		req := newReq()
		adaptor := setupAdaptor(t, &entity.BizSinkEndpoint{
			Status:       entity.BizPLEndpointStatusAvailable,
			EndpointType: entity.BizSinkEndpointTypeAmazonMSK,
			AuthType:     entity.BizSinkEndpointAuthSASLScram,
		})

		_, err := adaptor.ParseCreateParams(context.Background(), req, cluster)
		require.Error(t, err)
		assert.Contains(t, err.Error(), "bootstrap broker")
	})

	t.Run("requires broker section", func(t *testing.T) {
		req := newReq()
		req.KafkaInfo.Broker = nil
		adaptor := setupAdaptor(t, &entity.BizSinkEndpoint{
			Status:          entity.BizPLEndpointStatusAvailable,
			EndpointType:    entity.BizSinkEndpointTypeAmazonMSK,
			AuthType:        entity.BizSinkEndpointAuthSASLScram,
			BootstrapBroker: "b-1.demo:9196",
		})

		_, err := adaptor.ParseCreateParams(context.Background(), req, cluster)
		require.Error(t, err)
		assert.Contains(t, err.Error(), "broker")
	})

	t.Run("requires authentication section", func(t *testing.T) {
		req := newReq()
		req.KafkaInfo.Authentication = nil
		adaptor := setupAdaptor(t, &entity.BizSinkEndpoint{
			Status:          entity.BizPLEndpointStatusAvailable,
			EndpointType:    entity.BizSinkEndpointTypeAmazonMSK,
			AuthType:        entity.BizSinkEndpointAuthSASLScram,
			BootstrapBroker: "b-1.demo:9196",
		})

		_, err := adaptor.ParseCreateParams(context.Background(), req, cluster)
		require.Error(t, err)
		assert.Contains(t, err.Error(), "username and password")
	})
}

func TestGetAndValidateSinkEndpoint_TypeEnforcement(t *testing.T) {
	cluster := utils.NewCluster(&mgmt.Cluster{
		ClusterId:      100,
		OrganizationId: 200,
		CloudProvider:  "aws",
		ClusterInfo:    &mgmt.Cluster_NextGenCluster{NextGenCluster: &v12.Cluster{SharedPoolId: "sp-1"}},
	})
	mskEndpoint := func() *domain.QuerySinkEndpointResp {
		return &domain.QuerySinkEndpointResp{SinkEndpoint: &entity.BizSinkEndpoint{
			Status:          entity.BizPLEndpointStatusAvailable,
			EndpointType:    entity.BizSinkEndpointTypeAmazonMSK,
			BootstrapBroker: "b-1:9196",
		}}
	}

	t.Run("mysql path rejects an msk endpoint", func(t *testing.T) {
		model := sink_endpoint_mock.NewSinkEndpointModel(t)
		model.On("QuerySinkEndpoint", mock.Anything, mock.Anything).Return(mskEndpoint(), nil).Once()
		_, err := getAndValidateSinkEndpoint(context.Background(), model, cluster, 300,
			entity.BizSinkEndpointTypeVPCEndpointService)
		require.Error(t, err)
		assert.Contains(t, err.Error(), "not supported")
	})

	t.Run("kafka path accepts an msk endpoint", func(t *testing.T) {
		model := sink_endpoint_mock.NewSinkEndpointModel(t)
		model.On("QuerySinkEndpoint", mock.Anything, mock.Anything).Return(mskEndpoint(), nil).Once()
		ep, err := getAndValidateSinkEndpoint(context.Background(), model, cluster, 300,
			entity.BizSinkEndpointTypeVPCEndpointService, entity.BizSinkEndpointTypeAmazonMSK)
		require.NoError(t, err)
		assert.Equal(t, entity.BizSinkEndpointTypeAmazonMSK, ep.EndpointType)
	})

	t.Run("legacy empty type is treated as vpc endpoint service", func(t *testing.T) {
		model := sink_endpoint_mock.NewSinkEndpointModel(t)
		model.On("QuerySinkEndpoint", mock.Anything, mock.Anything).Return(
			&domain.QuerySinkEndpointResp{SinkEndpoint: &entity.BizSinkEndpoint{
				Status: entity.BizPLEndpointStatusAvailable, EndpointType: "",
			}}, nil).Once()
		_, err := getAndValidateSinkEndpoint(context.Background(), model, cluster, 300,
			entity.BizSinkEndpointTypeVPCEndpointService)
		require.NoError(t, err)
	})
}