package child

import (
	"fmt"
	"strings"

	"github.com/tidbcloud/dataflow-service-ng/app/models/data_pipeline/planner"
	"github.com/tidbcloud/dataflow-service-ng/app/models/domain"
)

const (
	snapshotDirectory    = "snapshot"
	incrementalDirectory = "incremental"
)

type stagingLocation struct {
	URI    string
	Prefix string
}

func validatePipelineIdentity(pipeline *domain.DataPipeline) error {
	if pipeline == nil || pipeline.ID == 0 || pipeline.ClusterID == 0 {
		return fmt.Errorf("%w: pipeline and cluster ids are required", ErrInvalidPipeline)
	}
	if pipeline.Params.Staging.Type != domain.DataPipelineStagingTypeCustomerManagedS3 {
		return fmt.Errorf("%w: customer-managed S3 is required", ErrUnsupportedExecutionPlan)
	}
	return nil
}

// buildStagingLocation renders the staging area the way one child addresses it.
//
// The scheme is the caller's because the two children disagree about it for the
// same bucket: Export names an Alibaba staging area with an oss:// URI, while
// TiCDC keeps s3:// and is told the provider separately, so that it uses
// path-style addressing and skips the AWS-only bucket-region probe.
func buildStagingLocation(
	pipeline *domain.DataPipeline, directory string, scheme planner.StagingScheme,
) (stagingLocation, error) {
	if err := validatePipelineIdentity(pipeline); err != nil {
		return stagingLocation{}, err
	}
	root := strings.TrimSpace(pipeline.StagingRoot)
	if root == "" {
		return stagingLocation{}, fmt.Errorf("%w: staging root is required", ErrInvalidPipeline)
	}
	prefix := strings.Trim(root, "/") + "/" + directory
	uri, err := planner.BuildS3URI(scheme, pipeline.Params.Staging.S3.Bucket, prefix)
	if err != nil {
		return stagingLocation{}, fmt.Errorf("%w: invalid staging location", ErrInvalidPipeline)
	}
	_, normalizedPrefix, _, err := planner.ParseS3BaseURI(uri)
	if err != nil {
		return stagingLocation{}, fmt.Errorf("%w: invalid staging location", ErrInvalidPipeline)
	}
	return stagingLocation{
		URI:    uri + "/",
		Prefix: normalizedPrefix + "/",
	}, nil
}

// stagingSchemeFor is how a child that follows the provider picks its scheme.
func stagingSchemeFor(pipeline *domain.DataPipeline) planner.StagingScheme {
	return planner.SchemeForProvider(pipeline.Params.Staging.S3.Provider)
}

// tableFilters forwards the user's rules unchanged. Export and TiCDC each own
// their own filter semantics, so re-interpreting the rules here would only add
// a third opinion.
func tableFilters(pipeline *domain.DataPipeline) ([]string, error) {
	if len(pipeline.Params.Source.TableFilter) == 0 {
		return nil, fmt.Errorf("%w: at least one table filter rule is required", ErrInvalidPipeline)
	}
	return append([]string(nil), pipeline.Params.Source.TableFilter...), nil
}

func validateSnapshotPlan(pipeline *domain.DataPipeline) error {
	if pipeline.ExecutionPlan.Snapshot.Engine != domain.DataPipelineSnapshotEngineDumpling ||
		pipeline.ExecutionPlan.Snapshot.Format != domain.DataPipelineSnapshotFormatCSV {
		return fmt.Errorf("%w: snapshot must use Dumpling CSV", ErrUnsupportedExecutionPlan)
	}
	if pipeline.SnapshotTSO == 0 {
		return fmt.Errorf("%w: snapshot TSO is required", ErrInvalidPipeline)
	}
	return nil
}

func validateIncrementalPlan(pipeline *domain.DataPipeline) error {
	if pipeline.Params.SyncMode != domain.DataPipelineSyncModeFullAndIncremental ||
		pipeline.ExecutionPlan.Incremental == nil ||
		pipeline.ExecutionPlan.Incremental.Engine != domain.DataPipelineIncrementalEngineTiCDC ||
		pipeline.ExecutionPlan.Incremental.Format != domain.DataPipelineIncrementalFormatCanalJSON {
		return fmt.Errorf("%w: incremental replication must use TiCDC Canal JSON", ErrUnsupportedExecutionPlan)
	}
	if pipeline.IntervalPlan.ProducerFlushIntervalSeconds <= 0 {
		return fmt.Errorf("%w: incremental intervals must be positive", ErrInvalidPipeline)
	}
	// A notified destination is planned with no poll interval at all, so zero
	// is only valid alongside a queue. Without one it means nothing would ever
	// pick the staged objects up, which is worth failing over rather than
	// building a task that quietly ingests nothing.
	if pipeline.IntervalPlan.ConsumerPollIntervalSeconds <= 0 &&
		!pipeline.Params.Staging.S3.UsesSQS() {
		return fmt.Errorf("%w: incremental intervals must be positive", ErrInvalidPipeline)
	}
	return nil
}

// validateStagingCredentials checks the credential shape the staging auth type
// requires. The two modes are mutually exclusive, and the secret is only
// decrypted - and only exists - for the access-key mode.
func validateStagingCredentials(pipeline *domain.DataPipeline, s3Secret Secret) error {
	s3 := pipeline.Params.Staging.S3
	if s3.UsesRoleARN() {
		if s3.RoleARN == "" {
			return fmt.Errorf("%w: staging role ARN is required", ErrInvalidPipeline)
		}
		return nil
	}
	if s3.AccessKeyID == "" {
		return fmt.Errorf("%w: staging access key id is required", ErrInvalidPipeline)
	}
	if s3Secret.Empty() {
		return ErrMissingDecryptedCredential
	}
	return nil
}