package impl

import (
	"context"
	"testing"

	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/dataflow-service-ng/app/models/domain"
	"github.com/tidbcloud/dataflow-service-ng/app/repo/entity"
)

func ossStaging(mutate func(*domain.DataPipelineS3StagingInput)) domain.DataPipelineStagingInput {
	s3 := domain.DataPipelineS3StagingInput{
		URI:             "oss://test-oss-bucket/base",
		Region:          "cn-hongkong",
		Endpoint:        "https://oss-cn-hongkong.aliyuncs.com",
		Provider:        domain.DataPipelineS3ProviderAlibabaOSS,
		AuthType:        domain.DataPipelineS3AuthTypeAccessKey,
		AccessKeyID:     "test-access-key",
		SecretAccessKey: "test-secret",
	}
	if mutate != nil {
		mutate(&s3)
	}
	return domain.DataPipelineStagingInput{
		Type: domain.DataPipelineStagingTypeCustomerManagedS3,
		S3:   &s3,
	}
}

// The provider, the URI scheme, the endpoint and the queue are each valid on
// their own and only wrong together, which is why they are checked as a group.
func TestValidateStagingInputHoldsTheProviderAndItsURITogether(t *testing.T) {
	t.Parallel()

	t.Run("a complete OSS staging area is accepted", func(t *testing.T) {
		t.Parallel()
		bucket, prefix, err := validateStagingInput(ossStaging(nil))
		require.Nil(t, err)
		require.Equal(t, "test-oss-bucket", bucket)
		require.Equal(t, "base", prefix)
	})

	// The region is not required to match the cluster's: the changefeed skips
	// its AWS bucket-region check for OSS, so a Hong Kong bucket reached from
	// elsewhere is a supported configuration rather than an oversight.
	t.Run("an OSS region unrelated to the cluster is accepted", func(t *testing.T) {
		t.Parallel()
		_, _, err := validateStagingInput(ossStaging(func(s3 *domain.DataPipelineS3StagingInput) {
			s3.Region = "ap-northeast-1"
			s3.Endpoint = "https://oss-ap-northeast-1.aliyuncs.com"
		}))
		require.Nil(t, err)
	})

	t.Run("an s3 scheme is rejected for the OSS provider", func(t *testing.T) {
		t.Parallel()
		_, _, err := validateStagingInput(ossStaging(func(s3 *domain.DataPipelineS3StagingInput) {
			s3.URI = "s3://test-oss-bucket/base"
		}))
		require.NotNil(t, err)
		require.Contains(t, err.Error(), "scheme")
	})

	t.Run("an oss scheme is rejected for AWS", func(t *testing.T) {
		t.Parallel()
		_, _, err := validateStagingInput(ossStaging(func(s3 *domain.DataPipelineS3StagingInput) {
			s3.Provider = domain.DataPipelineS3ProviderAWS
			s3.Endpoint = ""
		}))
		require.NotNil(t, err)
		require.Contains(t, err.Error(), "scheme")
	})

	t.Run("OSS without an endpoint is rejected", func(t *testing.T) {
		t.Parallel()
		_, _, err := validateStagingInput(ossStaging(func(s3 *domain.DataPipelineS3StagingInput) {
			s3.Endpoint = ""
		}))
		require.NotNil(t, err)
		require.Contains(t, err.Error(), "endpoint is required")
	})

	// Export answers "oss role_arn is not supported yet", so accepting it here
	// would only defer the same refusal into the snapshot task.
	t.Run("OSS with a role ARN is rejected at the request", func(t *testing.T) {
		t.Parallel()
		_, _, err := validateStagingInput(ossStaging(func(s3 *domain.DataPipelineS3StagingInput) {
			s3.AuthType = domain.DataPipelineS3AuthTypeRoleARN
			s3.AccessKeyID = ""
			s3.SecretAccessKey = ""
			s3.RoleARN = "acs:ram::1234:role/test"
		}))
		require.NotNil(t, err)
		require.Contains(t, err.Error(), "role_arn is not supported")
	})

	// The destination has no notification support for OSS, so a queue would be
	// stored and never read.
	t.Run("OSS with a queue is rejected", func(t *testing.T) {
		t.Parallel()
		_, _, err := validateStagingInput(ossStaging(func(s3 *domain.DataPipelineS3StagingInput) {
			s3.SQSQueueURL = "https://sqs.us-west-2.amazonaws.com/000000000000/q"
		}))
		require.NotNil(t, err)
		require.Contains(t, err.Error(), "sqs_queue_url is not supported")
	})

	t.Run("an endpoint on AWS is rejected", func(t *testing.T) {
		t.Parallel()
		_, _, err := validateStagingInput(ossStaging(func(s3 *domain.DataPipelineS3StagingInput) {
			s3.Provider = domain.DataPipelineS3ProviderAWS
			s3.URI = "s3://test-oss-bucket/base"
		}))
		require.NotNil(t, err)
		require.Contains(t, err.Error(), "endpoint is only supported")
	})
}

// The creator travelled from the row to the domain layer and was then dropped,
// because the Console read model had nowhere to put it: a Console could not say
// who created a Pipeline. The email is recorded at create time rather than
// resolved on read, because the account service resolves a user by token and
// not by id.
func TestCreateDataPipelineRecordsWhoCreatedIt(t *testing.T) {
	t.Parallel()

	h := newModelHarness()
	req := validCreateRequest()
	req.UserID = 4242
	req.UserEmail = "  someone@example.invalid  "

	_, err := h.model.CreateDataPipeline(context.Background(), req)
	require.Nil(t, err)
	require.Len(t, h.repo.creates, 1)

	persisted := h.repo.creates[0]
	require.Equal(t, entity.ID(4242), persisted.CreatorID)
	require.Equal(t, "someone@example.invalid", persisted.Creator,
		"the recorded address is trimmed, like every other stored string")
}

// TestCreateDataPipelineRecordsTheSelectionSize pins that the table count the
// source check already computes is kept rather than discarded. It is what tells
// a caller a Pipeline was set up over an empty selection, which is accepted and
// is otherwise indistinguishable from a healthy one: a snapshot-only Pipeline
// has no other source for the answer, because the destination's live table
// counters are written by the CDC run it does not have.
func TestCreateDataPipelineRecordsTheSelectionSize(t *testing.T) {
	t.Parallel()

	h := newModelHarness()
	h.validator.sourceMatched = 7

	_, err := h.model.CreateDataPipeline(context.Background(), validCreateRequest())
	require.Nil(t, err)
	require.Len(t, h.repo.creates, 1)

	recorded := h.repo.creates[0].Params.Source.MatchedTableCount
	require.NotNil(t, recorded)
	require.EqualValues(t, 7, *recorded)
}

// TestCreateDataPipelineDistinguishesNoTablesFromNotRecorded covers the reason
// the count is a pointer: zero is a real answer, and a row that predates the
// field must not be read as having matched nothing.
func TestCreateDataPipelineDistinguishesNoTablesFromNotRecorded(t *testing.T) {
	t.Parallel()

	h := newModelHarness()
	h.validator.sourceMatched = 0

	_, err := h.model.CreateDataPipeline(context.Background(), validCreateRequest())
	require.Nil(t, err)

	recorded := h.repo.creates[0].Params.Source.MatchedTableCount
	require.NotNil(t, recorded, "an empty selection is recorded as zero, not left absent")
	require.EqualValues(t, 0, *recorded)
}