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"
	"github.com/tidbcloud/dataflow-service-ng/pkg/lake"
)

const dumplingOnErrorAbort = "abort"

// csvHeaderLines is the number of leading CSV lines the destination discards.
var csvHeaderLines = int32(1)

func BuildLakeDataSourceSpec(
	pipeline *domain.DataPipeline,
	s3SecretAccessKey Secret,
) (lake.DataSourceSpec, error) {
	if err := validateLakePipeline(pipeline); err != nil {
		return lake.DataSourceSpec{}, err
	}
	if err := validateStagingCredentials(pipeline, s3SecretAccessKey); err != nil {
		return lake.DataSourceSpec{}, err
	}
	if pipeline.Params.Staging.S3.Region == "" {
		return lake.DataSourceSpec{}, fmt.Errorf("%w: S3 region is required", ErrInvalidPipeline)
	}
	if _, err := buildStagingLocation(pipeline, snapshotDirectory, planner.StagingSchemeS3); err != nil {
		return lake.DataSourceSpec{}, err
	}
	name := pipeline.Children.LakeDataSourceName
	if name == "" {
		return lake.DataSourceSpec{}, fmt.Errorf("%w: Lake datasource name is required", ErrInvalidPipeline)
	}
	return lake.DataSourceSpec{
		Name: name,
		S3:   lakeS3Credentials(pipeline, s3SecretAccessKey),
	}, nil
}

func BuildLakeTaskSpec(pipeline *domain.DataPipeline) (lake.TaskSpec, error) {
	if err := validateLakePipeline(pipeline); err != nil {
		return lake.TaskSpec{}, err
	}
	if strings.TrimSpace(pipeline.Params.Destination.Warehouse) == "" {
		return lake.TaskSpec{}, fmt.Errorf("%w: Lake warehouse is required", ErrInvalidPipeline)
	}
	// The Pipeline's filters are already TiCDC rules, which is the language
	// Lake's tableRules takes, so they travel verbatim. Anything else would be
	// a translation, and the previous one lost information: it kept only the
	// first database a filter named and dropped every wildcard table pattern.
	rules := lake.TableRules(append([]string(nil), pipeline.Params.Source.TableFilter...))
	if len(rules) == 0 {
		return lake.TaskSpec{}, fmt.Errorf("%w: at least one table filter is required", ErrInvalidPipeline)
	}
	name := pipeline.Children.LakeTaskName
	if name == "" {
		return lake.TaskSpec{}, fmt.Errorf("%w: Lake task name is required", ErrInvalidPipeline)
	}
	snapshot, err := buildStagingLocation(pipeline, snapshotDirectory, planner.StagingSchemeS3)
	if err != nil {
		return lake.TaskSpec{}, err
	}

	dataSourceName := pipeline.Children.LakeDataSourceName
	if dataSourceName == "" {
		return lake.TaskSpec{}, ErrInvalidLakeDataSource
	}

	task := lake.TaskSpec{
		Name:           name,
		DataSourceName: dataSourceName,
		Warehouse:      pipeline.Params.Destination.Warehouse,
		TiDB: lake.TiDBTaskSpec{
			SyncMode:             lake.SyncModeSnapshot,
			TableRules:           rules,
			DumplingS3Prefix:     snapshot.Prefix,
			DumplingCSVSeparator: csvSeparator,
			DumplingCSVNull:      csvNullValue,
			// The Export emits a header row (ExportOptions.skip_header is false),
			// so the destination must discard exactly one line or the column
			// names are ingested as data.
			DumplingCSVSkipHeader:   &csvHeaderLines,
			DumplingAutoCreateTable: true,
			DumplingOnError:         dumplingOnErrorAbort,
			DumplingPurge:           false,
			// Target naming is Lake's own contract; it is stored and
			// forwarded verbatim.
			TargetNaming: lake.TargetNaming{
				DatabasePrefix: pipeline.Params.Destination.TargetNaming.DatabasePrefix,
				DatabaseSuffix: pipeline.Params.Destination.TargetNaming.DatabaseSuffix,
				TablePrefix:    pipeline.Params.Destination.TargetNaming.TablePrefix,
				TableSuffix:    pipeline.Params.Destination.TargetNaming.TableSuffix,
			},
		},
	}

	switch pipeline.Params.SyncMode {
	case domain.DataPipelineSyncModeFull:
		if pipeline.ExecutionPlan.Incremental != nil {
			return lake.TaskSpec{}, fmt.Errorf("%w: full mode cannot have an incremental plan", ErrUnsupportedExecutionPlan)
		}
	case domain.DataPipelineSyncModeFullAndIncremental:
		if err := validateIncrementalPlan(pipeline); err != nil {
			return lake.TaskSpec{}, err
		}
		incremental, locationErr := buildStagingLocation(pipeline, incrementalDirectory, planner.StagingSchemeS3)
		if locationErr != nil {
			return lake.TaskSpec{}, locationErr
		}
		task.ContinuousIngestion = true
		task.TiDB.SyncMode = lake.SyncModeSnapshotAndCDC
		// Always sent, including the zero a notified task is planned with.
		// Omitting it instead would leave a Pipeline that used to poll still
		// carrying its old interval after an Edit attached a queue: an absent
		// field cannot clear a value Lake already holds. Lake stores an
		// explicit zero by clearing the field, so a notified task reads back
		// with no interval at all - which is what lakeTaskMatchesSpec has to
		// accept for the update to ever converge.
		consumerInterval := pipeline.IntervalPlan.ConsumerPollIntervalSeconds
		task.TiDB.PollIntervalSeconds = &consumerInterval
		task.TiDB.AllowDelete = true
		task.TiDB.ChangefeedS3Prefix = incremental.Prefix
	default:
		return lake.TaskSpec{}, fmt.Errorf("%w: unsupported sync mode", ErrUnsupportedExecutionPlan)
	}
	return task, nil
}

func validateLakePipeline(pipeline *domain.DataPipeline) error {
	if err := validatePipelineIdentity(pipeline); err != nil {
		return err
	}
	if err := validateSnapshotPlan(pipeline); err != nil {
		return err
	}
	if pipeline.Params.Destination.Type != domain.DataPipelineDestinationTypeTiDBLake ||
		pipeline.ExecutionPlan.DestinationType != domain.DataPipelineDestinationTypeTiDBLake {
		return fmt.Errorf("%w: TiDB Lake destination is required", ErrUnsupportedExecutionPlan)
	}
	return nil
}

// lakeS3Credentials renders the staging credentials in the shape Lake takes.
// The external ID is deliberately left empty: Lake fills it with the org slug,
// and that is the value its own guide tells users to put in the trust policy.
func lakeS3Credentials(pipeline *domain.DataPipeline, s3Secret Secret) lake.S3Credentials {
	s3 := pipeline.Params.Staging.S3
	credentials := lake.S3Credentials{
		Bucket: s3.Bucket,
		// Forwarded as given for both providers. What the destination does with
		// a region when an endpoint is also set, and which of the two wins if
		// they disagree, is not established for OSS - the AWS path is the only
		// one exercised end to end so far. Left as a straight pass-through
		// rather than guessed at, and to be settled against a real OSS
		// staging area.
		Region:   s3.Region,
		Endpoint: s3.Endpoint,
		// Empty leaves Lake polling the prefix, which is the behavior every
		// Pipeline had before queues were configurable.
		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(s3Secret.reveal())
	return credentials
}