package impl

import (
	"context"
	"fmt"
	"strconv"
	"strings"

	"github.com/tidbcloud/dataflow-service-ng/app/models/biz_changefeed"
	data_pipeline "github.com/tidbcloud/dataflow-service-ng/app/models/data_pipeline"
	"github.com/tidbcloud/dataflow-service-ng/app/models/data_pipeline/child"
	"github.com/tidbcloud/dataflow-service-ng/app/models/data_pipeline/planner"
	"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/dataflow-service-ng/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	"go.uber.org/zap"
)

var _ data_pipeline.ExternalResourceValidator = (*externalResourceValidator)(nil)

type externalResourceValidator struct {
	changefeeds biz_changefeed.BizChangefeedModel
	lake        lake.Client
}

func NewExternalResourceValidator(
	changefeeds biz_changefeed.BizChangefeedModel,
	lakeClient lake.Client,
) data_pipeline.ExternalResourceValidator {
	return &externalResourceValidator{changefeeds: changefeeds, lake: lakeClient}
}

func (v *externalResourceValidator) ValidateSource(
	ctx context.Context,
	clusterID uint64,
	source domain.DataPipelineSourceInput,
) (int32, bizerr.BizError) {
	if v.changefeeds == nil {
		return 0, bizerr.New(bizerr.ErrCodeUnknown, "changefeed source validator is unavailable")
	}
	// An absent filter means the caller is checking the account rather than a
	// selection, so the upstream call needs something that matches everything.
	filters := append([]string(nil), source.TableFilter...)
	checkingSelection := len(filters) > 0
	if !checkingSelection {
		filters = []string{"*.*"}
	}
	response, err := v.changefeeds.DescribeSchemaTable(ctx, &domain.DescribeSchemaTableReq{
		ClusterID: clusterID, Username: source.Username, Password: source.Password,
		Filters: filters, CaseSensitive: source.CaseSensitive,
	})
	if err != nil {
		// The upstream check answers with a bare 500 for everything that can go
		// wrong inside it, so the cause cannot be narrowed here. Report the
		// three things the caller can actually fix, as a parameter error rather
		// than an internal one, and keep the upstream text in the log instead of
		// the response: these calls carry the source password, and this boundary
		// must stay safe even if a downstream echoes its request.
		log.GetLogger(ctx).Warn("data pipeline source validation failed",
			zap.Uint64("clusterID", clusterID), zap.Error(err))
		// Name only what this caller supplied. Telling someone who passed just
		// credentials to check their table filter sends them after a setting
		// they did not set and cannot see.
		if checkingSelection {
			return 0, bizerr.New(bizerr.ErrCodeParams,
				"could not verify the source TiDB: check the username, password, "+
					"and that the table filter is valid for this cluster")
		}
		return 0, bizerr.New(bizerr.ErrCodeParams,
			"could not verify the source TiDB: check the username and password")
	}
	// Selecting nothing is allowed. It used to be rejected here on the theory
	// that the Export would fail on an empty selection, which is not what
	// happens: Dumpling treats an unmatched filter as an empty table list
	// rather than an error, exits 0, and writes only a metadata file, so the
	// regional job reports Complete and the Export reaches Succeeded.
	//
	// What is left is that a Pipeline over nothing has nothing to carry, and
	// that is only true of a snapshot. An incremental Pipeline is a standing
	// subscription: an empty cluster now says nothing about whether it will
	// have tables later, and the rest of the chain already handles ones that
	// arrive afterwards - the changefeed filter covers subsequent DDL, and the
	// destination is given table rules and creates targets on demand instead of
	// a list fixed at create time. Rejecting that would block setting a
	// Pipeline up before loading the data, which is a reasonable order to work
	// in. So this is logged and allowed rather than refused.
	if response == nil || len(response.ReplicateAllowSchemaTable) == 0 {
		log.GetLogger(ctx).Warn("data pipeline source selects no table",
			zap.Uint64("clusterID", clusterID),
			zap.Bool("checkingSelection", checkingSelection))
		return 0, nil
	}
	return int32(len(response.ReplicateAllowSchemaTable)), nil
}

func (v *externalResourceValidator) ValidateStaging(
	ctx context.Context,
	scope data_pipeline.ValidationScope,
	staging domain.DataPipelineStagingInput,
) bizerr.BizError {
	bucket, prefix, parseErr := validateStagingInput(staging)
	if parseErr != nil {
		return parseErr
	}
	if v.changefeeds == nil {
		return bizerr.New(bizerr.ErrCodeUnknown, "changefeed staging validator is unavailable")
	}
	// TiCDC keeps the s3:// form for every provider and is told which one it
	// is separately, so the changefeed probe is built that way even for OSS.
	baseURI, uriErr := planner.BuildS3URI(planner.StagingSchemeS3, bucket, prefix)
	if uriErr != nil {
		return invalidPlanError(uriErr)
	}
	s3Info := &dataflow.S3Info{
		Endpoint: validationS3Endpoint(*staging.S3, baseURI),
		DataFormat: &dataflow.S3Dataformat{
			Protocol:            dataflow.S3Protocol_S3_PROTOCOL_CANAL_JSON,
			EnableTidbExtension: true,
		},
	}
	if err := v.changefeeds.TestBizChangefeed(ctx, &domain.TestBizChangefeedReq{
		ClusterID: scope.ClusterID, DownstreamType: dataflow.ChangefeedType_S3,
		DownstreamParameter: s3Info,
	}); err != nil {
		return err.Chain("validate S3 producer access")
	}

	// Both halves of the staging area are checked, because passing only the
	// producer side is how a caller gets told the bucket is fine and then
	// discovers at create time that the destination cannot read it - which is
	// the failure this endpoint exists to prevent. The Lake test used to be
	// skipped without a table filter, from when the datasource carried the
	// source selection; it no longer does, and DataSourceSpec has no filter to
	// give, so there is nothing left to wait for.
	if v.lake == nil {
		return bizerr.New(bizerr.ErrCodeUnknown, "Lake staging validator is unavailable")
	}
	invocation := lake.Invocation{
		TiDBOrgID:   strconv.FormatUint(scope.TenantID, 10),
		LakeOrgID:   scope.LakeOrgID,
		InitiatedBy: strconv.FormatUint(scope.UserID, 10),
	}
	if scope.PipelineID != 0 {
		invocation.ContextType = lakeContextTypeDataPipeline
		invocation.ContextID = strconv.FormatUint(scope.PipelineID, 10)
		invocation.ContextVersion = lakeContextVersion
	}
	name := fmt.Sprintf("tidb-dp-validate-%d-%d", scope.ClusterID, scope.PipelineID)
	if err := v.lake.TestDataSource(ctx, invocation, lake.DataSourceSpec{
		Name: name,
		S3:   validationLakeCredentials(*staging.S3, bucket),
	}); err != nil {
		return bizerr.WrapError(err, "validate S3 consumer access through Lake")
	}
	return nil
}

// validationS3Endpoint and validationLakeCredentials mirror what the child
// builders send, so validating a staging configuration exercises the same auth
// shape the Pipeline will actually be created with.
func validationS3Endpoint(
	s3 domain.DataPipelineS3StagingInput,
	uri string,
) *dataflow.S3Endpoint {
	endpoint := &dataflow.S3Endpoint{
		S3Uri:    uri,
		Endpoint: s3.Endpoint,
		Provider: child.ChangefeedS3Provider(s3.Provider),
	}
	if s3.UsesRoleARN() {
		endpoint.AuthType = dataflow.S3AuthType_S3_AUTH_TYPE_ASSUME_ROLE
		endpoint.IamAssumedRoleArn = s3.RoleARN
		return endpoint
	}
	endpoint.AuthType = dataflow.S3AuthType_S3_AUTH_TYPE_AK_SK
	endpoint.AccessKey = s3.AccessKeyID
	endpoint.SecretKey = s3.SecretAccessKey
	return endpoint
}

func validationLakeCredentials(
	s3 domain.DataPipelineS3StagingInput,
	bucket string,
) lake.S3Credentials {
	credentials := lake.S3Credentials{
		Bucket: bucket, Region: s3.Region, Endpoint: s3.Endpoint,
		// Lake probes the queue as part of validating the credentials, so
		// passing it here is what turns "the bucket is reachable" into "the
		// whole staging area is". An empty SQS region is Lake's to default to
		// the bucket region.
		SQSQueueURL: s3.SQSQueueURL, SQSRegion: s3.SQSRegion,
	}
	if s3.UsesRoleARN() {
		credentials.RoleARN = s3.RoleARN
		return credentials
	}
	credentials.AccessKey = lake.NewSecret(s3.AccessKeyID)
	credentials.SecretKey = lake.NewSecret(s3.SecretAccessKey)
	return credentials
}

func (v *externalResourceValidator) ValidateChangefeedSpecification(
	ctx context.Context,
	clusterID uint64,
	specification string,
) bizerr.BizError {
	if v.changefeeds == nil {
		return bizerr.New(bizerr.ErrCodeUnknown, "changefeed specification validator is unavailable")
	}
	specification = strings.TrimSpace(specification)
	if specification == "" {
		return bizerr.New(bizerr.ErrCodeParams, "changefeed specification is required")
	}
	response, err := v.changefeeds.ListBizChangefeedSpecifications(
		ctx,
		&domain.ListBizChangefeedSpecificationsReq{ClusterID: clusterID},
	)
	if err != nil {
		return err.Chain("list changefeed specifications")
	}
	if response != nil {
		for _, candidate := range response.Specifications {
			if candidate != nil && candidate.SpecificationCode == specification {
				return nil
			}
		}
	}
	return bizerr.New(
		bizerr.ErrCodeParams,
		"changefeed specification %q is unavailable for this cluster",
		specification,
	)
}