package impl

import (
	"context"
	"testing"

	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	biz_changefeed_mock "github.com/tidbcloud/dataflow-service-ng/app/models/biz_changefeed/mock"
	data_pipeline "github.com/tidbcloud/dataflow-service-ng/app/models/data_pipeline"
	"github.com/tidbcloud/dataflow-service-ng/app/models/domain"
	dataflow "github.com/tidbcloud/dataflow-service-ng/idl/pbgen/proto/consoleapi"
	"github.com/tidbcloud/dataflow-service-ng/pkg/lake"
	"github.com/tidbcloud/pkgv2/bizerr"
)

func TestExternalResourceValidatorValidateSource(t *testing.T) {
	changefeeds := biz_changefeed_mock.NewBizChangefeedModel(t)
	changefeeds.On(
		"DescribeSchemaTable",
		mock.Anything,
		mock.MatchedBy(func(req *domain.DescribeSchemaTableReq) bool {
			// The rules reach the source check exactly as the user wrote them;
			// this service does not expand or normalize them.
			return req.ClusterID == 101 && req.Username == "test-source-user" &&
				req.Password == "test-source-password" &&
				len(req.Filters) == 1 && req.Filters[0] == "test_app.*"
		}),
	).Return(&domain.DescribeSchemaTableResp{
		ReplicateAllowSchemaTable: []string{"test_app.orders"},
	}, nil).Once()

	validator := NewExternalResourceValidator(changefeeds, nil)
	matched, err := validator.ValidateSource(context.Background(), 101,
		domain.DataPipelineSourceInput{
			Username: "test-source-user", Password: "test-source-password",
			TableFilter: []string{"test_app.*"},
		})
	require.NoError(t, err)
	require.EqualValues(t, 1, matched)
}

func TestExternalResourceValidatorValidateSourceSurfacesFailures(t *testing.T) {
	changefeeds := biz_changefeed_mock.NewBizChangefeedModel(t)
	changefeeds.On("DescribeSchemaTable", mock.Anything, mock.Anything).
		Return(nil, bizerr.New(bizerr.ErrCodeParams, "test source unreachable")).Once()

	validator := NewExternalResourceValidator(changefeeds, nil)
	_, err := validator.ValidateSource(context.Background(), 101, domain.DataPipelineSourceInput{
		Username: "test-source-user", Password: "test-source-password",
	})
	require.Error(t, err)
	require.Equal(t, bizerr.ErrCodeParams, err.RawErrCode())
}

// TestExternalResourceValidatorAllowsSelectionMatchingNothing pins that a
// filter which parses and authenticates but selects no table is accepted. The
// Export does not fail on it - Dumpling exits 0 on an unmatched filter and
// writes only a metadata file - and an incremental Pipeline set up before its
// tables exist picks them up when they arrive, so refusing here would block a
// reasonable order of work.
func TestExternalResourceValidatorAllowsSelectionMatchingNothing(t *testing.T) {
	changefeeds := biz_changefeed_mock.NewBizChangefeedModel(t)
	changefeeds.On("DescribeSchemaTable", mock.Anything, mock.Anything).
		Return(&domain.DescribeSchemaTableResp{
			ReplicateDenySchemaTable: []string{"test_app.no_primary_key"},
		}, nil).Once()

	validator := NewExternalResourceValidator(changefeeds, nil)
	_, err := validator.ValidateSource(context.Background(), 101, domain.DataPipelineSourceInput{
		Username: "test-source-user", Password: "test-source-password",
		TableFilter: []string{"test_app.absent"},
	})
	require.NoError(t, err)
}

// TestExternalResourceValidatorAllowsEmptyCluster covers the credential-only
// endpoint against a cluster with nothing in it. Reaching the cluster and
// seeing no table is not a credential problem, and reporting one used to send
// the caller after grants that were already correct.
func TestExternalResourceValidatorAllowsEmptyCluster(t *testing.T) {
	changefeeds := biz_changefeed_mock.NewBizChangefeedModel(t)
	changefeeds.On("DescribeSchemaTable", mock.Anything, mock.Anything).
		Return(&domain.DescribeSchemaTableResp{}, nil).Once()

	validator := NewExternalResourceValidator(changefeeds, nil)
	_, err := validator.ValidateSource(context.Background(), 101, domain.DataPipelineSourceInput{
		Username: "test-source-user", Password: "test-source-password",
	})
	require.NoError(t, err)
}

// TestExternalResourceValidatorHidesUpstreamSourceError pins that the source
// check reports what the caller can fix instead of the upstream's own text.
// These calls carry the source password, so the downstream error stays in the
// log rather than the response.
func TestExternalResourceValidatorHidesUpstreamSourceError(t *testing.T) {
	changefeeds := biz_changefeed_mock.NewBizChangefeedModel(t)
	changefeeds.On("DescribeSchemaTable", mock.Anything, mock.Anything).
		Return(nil, bizerr.New(bizerr.ErrCodeUnknown, "cluster-agent test-source-password leak")).Once()

	validator := NewExternalResourceValidator(changefeeds, nil)
	_, err := validator.ValidateSource(context.Background(), 101, domain.DataPipelineSourceInput{
		Username: "test-source-user", Password: "test-source-password",
	})
	require.Error(t, err)
	require.Equal(t, bizerr.ErrCodeParams, err.RawErrCode())
	require.Contains(t, err.Error(), "check the username and password")
	require.NotContains(t, err.Error(), "test-source-password")
}

func TestExternalResourceValidatorValidateStaging(t *testing.T) {
	staging := domain.DataPipelineStagingInput{
		Type: domain.DataPipelineStagingTypeCustomerManagedS3,
		S3: &domain.DataPipelineS3StagingInput{
			URI: "s3://test-pipeline-bucket/customer/base", Region: "us-west-2",
			AuthType: domain.DataPipelineS3AuthTypeAccessKey,

			AccessKeyID: "test-access-key", SecretAccessKey: "test-secret-key",
		},
	}
	scope := data_pipeline.ValidationScope{
		TenantID: 10, ClusterID: 101, PipelineID: 202, UserID: 303, LakeOrgID: "7",
	}
	t.Run("validation tests the producer and the destination", func(t *testing.T) {
		changefeeds := biz_changefeed_mock.NewBizChangefeedModel(t)
		changefeeds.On(
			"TestBizChangefeed",
			mock.Anything,
			mock.MatchedBy(func(req *domain.TestBizChangefeedReq) bool {
				s3Info, ok := req.DownstreamParameter.(*dataflow.S3Info)
				return ok && req.ClusterID == 101 &&
					s3Info.GetEndpoint().GetS3Uri() == "s3://test-pipeline-bucket/customer/base" &&
					s3Info.GetEndpoint().GetSecretKey() == "test-secret-key"
			}),
		).Return(nil).Once()
		lakeClient := lake.NewFakeClient()
		validator := NewExternalResourceValidator(changefeeds, lakeClient)

		// Both halves, on every caller. Checking only the producer is how a
		// caller is told the bucket is fine and then discovers at create time
		// that the destination cannot read it.
		require.NoError(t, validator.ValidateStaging(context.Background(), scope, staging))
		require.Equal(t, []lake.Operation{lake.OperationTestDataSource}, lakeClient.Calls())
	})

	t.Run("a custom endpoint is rejected before any external call", func(t *testing.T) {
		customEndpoint := staging
		s3 := *staging.S3
		s3.Endpoint = "https://test-s3.example.invalid"
		customEndpoint.S3 = &s3

		changefeeds := biz_changefeed_mock.NewBizChangefeedModel(t)
		lakeClient := lake.NewFakeClient()
		validator := NewExternalResourceValidator(changefeeds, lakeClient)

		err := validator.ValidateStaging(context.Background(), scope, customEndpoint)
		require.Error(t, err)
		require.Equal(t, bizerr.ErrCodeParams, err.RawErrCode())
		require.Empty(t, lakeClient.Calls())
	})
}

func TestExternalResourceValidatorValidateChangefeedSpecification(t *testing.T) {
	changefeeds := biz_changefeed_mock.NewBizChangefeedModel(t)
	changefeeds.On("ListBizChangefeedSpecifications", mock.Anything, mock.Anything).
		Return(&domain.ListBizChangefeedSpecificationsResp{
			Specifications: []*domain.BizChangefeedSpecification{{SpecificationCode: "standard-2"}},
		}, nil).Twice()
	validator := NewExternalResourceValidator(changefeeds, nil)

	require.NoError(t,
		validator.ValidateChangefeedSpecification(context.Background(), 101, " standard-2 "))

	err := validator.ValidateChangefeedSpecification(context.Background(), 101, "standard-9")
	require.Error(t, err)
	require.Equal(t, bizerr.ErrCodeParams, err.RawErrCode())
}

// The two callers get different advice because they supplied different things.
// Telling someone who passed only credentials to check their table filter
// sends them after a setting they never set.
func TestValidateSourceNamesOnlyWhatTheCallerSupplied(t *testing.T) {
	t.Run("credentials only", func(t *testing.T) {
		changefeeds := biz_changefeed_mock.NewBizChangefeedModel(t)
		changefeeds.On("DescribeSchemaTable", mock.Anything,
			mock.MatchedBy(func(req *domain.DescribeSchemaTableReq) bool {
				// The upstream call still needs a selection; a wildcard is
				// substituted so the account is exercised against everything.
				return len(req.Filters) == 1 && req.Filters[0] == "*.*"
			}),
		).Return(nil, bizerr.New(bizerr.ErrCodeUnknown, "upstream 500")).Once()

		_, err := NewExternalResourceValidator(changefeeds, lake.NewFakeClient()).
			ValidateSource(context.Background(), 101, domain.DataPipelineSourceInput{
				Username: "u", Password: "p",
			})
		require.Error(t, err)
		require.Contains(t, err.Error(), "check the username and password")
		require.NotContains(t, err.Error(), "table filter")
	})

	t.Run("a selection was supplied", func(t *testing.T) {
		changefeeds := biz_changefeed_mock.NewBizChangefeedModel(t)
		changefeeds.On("DescribeSchemaTable", mock.Anything,
			mock.MatchedBy(func(req *domain.DescribeSchemaTableReq) bool {
				return len(req.Filters) == 1 && req.Filters[0] == "app.orders"
			}),
		).Return(nil, bizerr.New(bizerr.ErrCodeUnknown, "upstream 500")).Once()

		_, err := NewExternalResourceValidator(changefeeds, lake.NewFakeClient()).
			ValidateSource(context.Background(), 101, domain.DataPipelineSourceInput{
				Username: "u", Password: "p", TableFilter: []string{"app.orders"},
			})
		require.Error(t, err)
		require.Contains(t, err.Error(), "table filter")
	})

}