package impl

import (
	"context"
	"fmt"
	"math"
	"strconv"
	"strings"
	"time"

	"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"
	data_pipeline_repo "github.com/tidbcloud/dataflow-service-ng/app/repo/data_pipeline"
	"github.com/tidbcloud/dataflow-service-ng/app/repo/entity"
	"github.com/tidbcloud/dataflow-service-ng/app/workflow"
	encryptor_wrapper "github.com/tidbcloud/dataflow-service-ng/pkg/encryptor-wrapper"
	"github.com/tidbcloud/dataflow-service-ng/pkg/lake"
	"github.com/tidbcloud/dataflow-service-ng/pkg/log"
	"github.com/tidbcloud/pkgv2/bizerr"
	mgmt "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
)

const (
	dataPipelineParamsVersion = 1
	maxPipelineNameLength     = 64
	maxPageSize               = 100
)

type RuntimeConfig struct {
	Environment string
}

type ClusterResolver interface {
	GetClusterByID(context.Context, uint64) (*mgmt.Cluster, bizerr.BizError)
}

var _ data_pipeline.Model = (*dataPipelineModel)(nil)

type dataPipelineModel struct {
	repo         data_pipeline_repo.DataPipelineRepo
	clusters     ClusterResolver
	changefeeds  biz_changefeed.BizChangefeedModel
	destinations lake.Client
	validator    data_pipeline.ExternalResourceValidator
	encryptor    encryptor_wrapper.Encryptor
	workflow     workflow.Workflow
	runtime      RuntimeConfig
	clock        planner.Clock
}

func NewDataPipelineModel(
	repo data_pipeline_repo.DataPipelineRepo,
	clusters ClusterResolver,
	changefeeds biz_changefeed.BizChangefeedModel,
	destinations lake.Client,
	validator data_pipeline.ExternalResourceValidator,
	encryptor encryptor_wrapper.Encryptor,
	workflowModel workflow.Workflow,
	runtime RuntimeConfig,
	clock planner.Clock,
) data_pipeline.Model {
	return &dataPipelineModel{
		repo: repo, clusters: clusters, changefeeds: changefeeds,
		destinations: destinations, validator: validator, encryptor: encryptor,
		workflow: workflowModel, runtime: runtime, clock: clock,
	}
}

// validateCreateAgainstSource runs the checks that need the upstream cluster,
// and returns the size of the selection so the caller can record it. The
// specification check is here rather than beside it because it belongs to the
// same round trip to the source and only applies to a Pipeline that will run a
// changefeed.
func (m *dataPipelineModel) validateCreateAgainstSource(
	ctx context.Context,
	req *domain.CreateDataPipelineReq,
	executionPlan domain.DataPipelineExecutionPlan,
) (int32, bizerr.BizError) {
	matchedTables, err := m.validator.ValidateSource(ctx, req.ClusterID, req.Source)
	if err != nil {
		return 0, err.Chain("validate data pipeline source")
	}
	if executionPlan.Incremental != nil {
		if err = m.validator.ValidateChangefeedSpecification(
			ctx, req.ClusterID, strings.TrimSpace(req.ChangefeedSpecification)); err != nil {
			return 0, err.Chain("validate data pipeline changefeed specification")
		}
	}
	return matchedTables, nil
}

func (m *dataPipelineModel) CreateDataPipeline(
	ctx context.Context,
	req *domain.CreateDataPipelineReq,
) (*domain.CreateDataPipelineResp, bizerr.BizError) {
	if err := validateCreateRequest(req); err != nil {
		return nil, err
	}
	_, tenantID, err := m.resolveScope(ctx, req.ClusterID)
	if err != nil {
		return nil, err
	}

	executionPlan, intervalPlan, bucket, basePrefix, err := m.planCreate(req)
	if err != nil {
		return nil, err
	}
	if err = m.ensureDependencies(); err != nil {
		return nil, err
	}
	// Fail here rather than part-way through the DAG: by the time
	// CreateDestination runs, a real Export and a real changefeed already exist.
	if err = m.ensureDestinationAdapter(req.Destination.Type); err != nil {
		return nil, err
	}

	matchedTables, err := m.validateCreateAgainstSource(ctx, req, executionPlan)
	if err != nil {
		return nil, err
	}

	pipelineID := m.repo.AllocateID().Uint64()
	children, childErr := m.planChildren(pipelineID, executionPlan)
	if childErr != nil {
		return nil, childErr
	}
	paths, planErr := planner.BuildStagingPaths(
		basePrefix, m.runtime.Environment, req.ClusterID, pipelineID)
	if planErr != nil {
		return nil, invalidPlanError(planErr)
	}
	snapshotTSO, planErr := planner.EnsureSnapshotTSO(0, m.clock)
	if planErr != nil {
		return nil, invalidPlanError(planErr)
	}

	validationScope := data_pipeline.ValidationScope{
		TenantID: tenantID, ClusterID: req.ClusterID, PipelineID: pipelineID,
		UserID: req.UserID, LakeOrgID: strings.TrimSpace(req.Destination.LakeOrgID),
	}
	if err = m.validator.ValidateStaging(ctx, validationScope, req.Staging); err != nil {
		return nil, err.Chain("validate data pipeline staging")
	}

	passwordCiphertext, err := m.encryptor.EncryptBase64(req.Source.Password)
	if err != nil {
		return nil, err.Chain("encrypt source credential")
	}
	// A role-ARN Pipeline has no secret to keep, so nothing is encrypted and
	// nothing is stored.
	secretCiphertext := ""
	if !req.Staging.S3.UsesRoleARN() {
		secretCiphertext, err = m.encryptor.EncryptBase64(req.Staging.S3.SecretAccessKey)
		if err != nil {
			return nil, err.Chain("encrypt staging credential")
		}
	}

	createdAt := m.clock.Now().UTC()
	pipeline := &entity.DataPipeline{
		ID:                      entity.ID(pipelineID),
		TenantID:                entity.ID(tenantID),
		ClusterID:               entity.ID(req.ClusterID),
		CreatorID:               entity.ID(req.UserID),
		Creator:                 strings.TrimSpace(req.UserEmail),
		Name:                    strings.TrimSpace(req.Name),
		Status:                  entity.DataPipelineStatusCreating,
		StatusEnteredAt:         &createdAt,
		ChangefeedSpecification: incrementalSpecification(executionPlan, req.ChangefeedSpecification),
		Params: entity.DataPipelineParams{
			Version:  dataPipelineParamsVersion,
			SyncMode: req.SyncMode,
			Source: entity.DataPipelineSourceConfig{
				Username: req.Source.Username, PasswordCiphertext: passwordCiphertext,
				TableFilter:       append([]string(nil), req.Source.TableFilter...),
				CaseSensitive:     req.Source.CaseSensitive,
				MatchedTableCount: &matchedTables,
			},
			Staging: entity.DataPipelineStagingConfig{
				Type: req.Staging.Type,
				S3: entity.DataPipelineS3Config{
					Bucket: bucket, BasePrefix: basePrefix, Region: req.Staging.S3.Region,
					Endpoint:                  req.Staging.S3.Endpoint,
					AuthType:                  req.Staging.S3.AuthType,
					AccessKeyID:               req.Staging.S3.AccessKeyID,
					SecretAccessKeyCiphertext: secretCiphertext,
					RoleARN:                   req.Staging.S3.RoleARN,
					SQSQueueURL:               strings.TrimSpace(req.Staging.S3.SQSQueueURL),
					SQSRegion:                 strings.TrimSpace(req.Staging.S3.SQSRegion),
					Provider:                  req.Staging.S3.Provider,
				},
			},
			Destination: entity.DataPipelineDestinationConfig{
				Type:         req.Destination.Type,
				LakeOrgID:    strings.TrimSpace(req.Destination.LakeOrgID),
				Warehouse:    strings.TrimSpace(req.Destination.Warehouse),
				TargetNaming: req.Destination.TargetNaming,
			},
		},
		ExecutionPlan: executionPlan,
		IntervalPlan:  intervalPlan,
		SnapshotTSO:   snapshotTSO,
		StagingRoot:   paths.Root,
		Children:      children,
	}
	if err = m.repo.Create(ctx, pipeline); err != nil {
		return nil, err.Chain("persist data pipeline")
	}
	logPipelineCreated(ctx, pipeline)

	if workflowErr := m.startWorkflow(
		workflow.DagCreateDataPipeline,
		workflow.BusinessActionCreateDataPipeline,
		pipeline,
	); workflowErr != nil {
		m.logWorkflowStartFailure(ctx, pipeline, workflowErr)
	}
	return &domain.CreateDataPipelineResp{PipelineID: pipelineID}, nil
}

func (m *dataPipelineModel) GetDataPipeline(
	ctx context.Context,
	req *domain.GetDataPipelineReq,
) (*domain.GetDataPipelineResp, bizerr.BizError) {
	if req == nil || req.ClusterID == 0 || req.PipelineID == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cluster and data pipeline IDs are required")
	}
	_, tenantID, err := m.resolveScope(ctx, req.ClusterID)
	if err != nil {
		return nil, err
	}
	pipeline, err := m.repo.Get(
		ctx,
		data_pipeline_repo.WithIDs(req.PipelineID),
		data_pipeline_repo.WithTenantID(tenantID),
		data_pipeline_repo.WithClusterID(req.ClusterID),
	)
	if err != nil {
		return nil, err.Chain("get scoped data pipeline")
	}
	return &domain.GetDataPipelineResp{DataPipeline: dataPipelineDetails(pipeline)}, nil
}

func (m *dataPipelineModel) ListDataPipelines(
	ctx context.Context,
	req *domain.ListDataPipelinesReq,
) (*domain.ListDataPipelinesResp, bizerr.BizError) {
	if req == nil || req.ClusterID == 0 || req.Page == 0 || req.PerPage == 0 || req.PerPage > maxPageSize {
		return nil, bizerr.New(bizerr.ErrCodeParams, "valid cluster ID and pagination are required")
	}
	_, tenantID, err := m.resolveScope(ctx, req.ClusterID)
	if err != nil {
		return nil, err
	}
	offset := uint64(req.Page-1) * uint64(req.PerPage)
	if offset > math.MaxInt {
		return nil, bizerr.New(bizerr.ErrCodeParams, "data pipeline page offset is too large")
	}
	baseOps := []data_pipeline_repo.FilterOp{
		data_pipeline_repo.WithTenantID(tenantID),
		data_pipeline_repo.WithClusterID(req.ClusterID),
	}
	if req.Status != nil {
		baseOps = append(baseOps, data_pipeline_repo.WithStatuses(*req.Status))
	}
	total, err := m.repo.Count(ctx, baseOps...)
	if err != nil {
		return nil, err.Chain("count scoped data pipelines")
	}
	listOps := append([]data_pipeline_repo.FilterOp(nil), baseOps...)
	listOps = append(
		listOps,
		data_pipeline_repo.OrderByIDDesc(),
		data_pipeline_repo.WithPagination(int(offset), int(req.PerPage)),
	)
	rows, err := m.repo.List(ctx, listOps...)
	if err != nil {
		return nil, err.Chain("list scoped data pipelines")
	}
	items := make([]*domain.DataPipelineDetails, 0, len(rows))
	for index := range rows {
		items = append(items, dataPipelineDetails(&rows[index]))
	}
	if total > math.MaxUint32 {
		total = math.MaxUint32
	}
	return &domain.ListDataPipelinesResp{DataPipelines: items, Total: uint32(total)}, nil
}

func (m *dataPipelineModel) PauseDataPipeline(
	ctx context.Context,
	req *domain.PauseDataPipelineReq,
) bizerr.BizError {
	if req == nil {
		return bizerr.New(bizerr.ErrCodeParams, "pause data pipeline request is required")
	}
	pipeline, err := m.loadScoped(ctx, req.ClusterID, req.PipelineID)
	if err != nil {
		return err
	}
	if pipeline.Params.SyncMode != entity.DataPipelineSyncModeFullAndIncremental ||
		pipeline.Status != entity.DataPipelineStatusRunning {
		return invalidStateError(pipeline.Status, "pause")
	}
	return m.transitionAndStart(
		ctx, pipeline,
		entity.DataPipelineStatusPausing,
		workflow.DagPauseDataPipeline,
		workflow.BusinessActionPauseDataPipeline,
	)
}

func (m *dataPipelineModel) EditDataPipeline(
	ctx context.Context,
	req *domain.EditDataPipelineReq,
) bizerr.BizError {
	if err := validateEditRequest(req); err != nil {
		return err
	}
	pipeline, err := m.loadScoped(ctx, req.ClusterID, req.PipelineID)
	if err != nil {
		return err
	}
	// A revision the children rejected leaves the row Failed at an applying
	// stage. Retry would only replay the same bad value, so Edit is accepted
	// there too: the children are still stopped, and a new revision replaces
	// the one that failed.
	if !editableStatus(pipeline) {
		return invalidStateError(pipeline.Status, "edit")
	}
	if err = m.ensureDependencies(); err != nil {
		return err
	}

	// Planned against the queue the Edit asks for, not the one the Pipeline
	// currently has: attaching or detaching a queue is exactly what moves the
	// budget between the two halves.
	intervalPlan, planErr := planner.PlanDeliveryIntervals(
		pipeline.Params.SyncMode,
		int32(req.Destination.FlushIntervalSeconds),
		req.Staging.UsesSQS(),
	)
	if planErr != nil {
		return invalidPlanError(planErr)
	}

	password, err := m.encryptor.DecryptBase64(pipeline.Params.Source.PasswordCiphertext)
	if err != nil {
		return err.Chain("decrypt source credential")
	}
	// A role-ARN Pipeline stores no secret, so there is nothing to decrypt.
	secret := ""
	if !pipeline.Params.Staging.S3.UsesRoleARN() {
		secret, err = m.encryptor.DecryptBase64(pipeline.Params.Staging.S3.SecretAccessKeyCiphertext)
		if err != nil {
			return err.Chain("decrypt staging credential")
		}
	}
	source := domain.DataPipelineSourceInput{
		Username: pipeline.Params.Source.Username, Password: password,
		TableFilter: req.Source.TableFilter, CaseSensitive: req.Source.CaseSensitive,
	}
	matchedTables, err := m.validator.ValidateSource(ctx, req.ClusterID, source)
	if err != nil {
		return err.Chain("validate edited data pipeline source")
	}
	baseURI, planErr := planner.BuildS3URI(
		planner.SchemeForProvider(pipeline.Params.Staging.S3.Provider),
		pipeline.Params.Staging.S3.Bucket, pipeline.Params.Staging.S3.BasePrefix)
	if planErr != nil {
		return invalidPlanError(planErr)
	}
	staging := domain.DataPipelineStagingInput{
		Type: pipeline.Params.Staging.Type,
		S3: &domain.DataPipelineS3StagingInput{
			URI: baseURI, Region: pipeline.Params.Staging.S3.Region,
			Endpoint:    pipeline.Params.Staging.S3.Endpoint,
			Provider:    pipeline.Params.Staging.S3.Provider,
			AuthType:    stagingAuthType(pipeline.Params.Staging.S3),
			AccessKeyID: pipeline.Params.Staging.S3.AccessKeyID, SecretAccessKey: secret,
			RoleARN: pipeline.Params.Staging.S3.RoleARN,
			// The requested queue rather than the stored one, so a queue that
			// cannot be reached is rejected here instead of silently leaving
			// the destination waiting for notifications that never arrive.
			SQSQueueURL: strings.TrimSpace(req.Staging.SQSQueueURL),
			SQSRegion:   strings.TrimSpace(req.Staging.SQSRegion),
		},
	}
	if err = m.validator.ValidateStaging(ctx, data_pipeline.ValidationScope{
		TenantID: pipeline.TenantID.Uint64(), ClusterID: req.ClusterID,
		PipelineID: req.PipelineID, UserID: req.UserID,
		// Not from the request: an Edit cannot move a Pipeline to a different
		// Lake, so the stored one is the only correct answer here.
		LakeOrgID: pipeline.Params.Destination.LakeOrgID,
	}, staging); err != nil {
		return err.Chain("validate edited data pipeline staging")
	}

	pipeline.Name = strings.TrimSpace(req.Name)
	pipeline.Params.Source.TableFilter = append([]string(nil), req.Source.TableFilter...)
	pipeline.Params.Source.CaseSensitive = req.Source.CaseSensitive
	pipeline.Params.Source.MatchedTableCount = &matchedTables
	pipeline.Params.Destination.Warehouse = strings.TrimSpace(req.Destination.Warehouse)
	pipeline.Params.Destination.TargetNaming = req.Destination.TargetNaming
	// Trimmed on the way in, so the stored value is the one Lake is sent and
	// the one the spec matcher compares - all three have to be the same string
	// or the datasource never converges.
	pipeline.Params.Staging.S3.SQSQueueURL = strings.TrimSpace(req.Staging.SQSQueueURL)
	pipeline.Params.Staging.S3.SQSRegion = strings.TrimSpace(req.Staging.SQSRegion)
	pipeline.IntervalPlan = intervalPlan
	pipeline.LastError = nil
	// Edit is asynchronous on purpose. Persisting the desired revision without
	// applying it would defer every child rejection to Resume, and a Resume
	// failure leaves the Pipeline in Failed where Edit is no longer accepted -
	// a bad value would then only be recoverable by deleting the Pipeline.
	setPipelineStatus(ctx, pipeline, entity.DataPipelineStatusEditing, m.clock.Now())
	if err = m.repo.Save(ctx, pipeline); err != nil {
		return err.Chain("save edited data pipeline")
	}
	if workflowErr := m.startWorkflow(
		workflow.DagEditDataPipeline, workflow.BusinessActionEditDataPipeline, pipeline,
	); workflowErr != nil {
		m.logWorkflowStartFailure(ctx, pipeline, workflowErr)
	}
	return nil
}

// ValidateDataPipelineDestination checks a destination before any Pipeline row
// or external resource exists.
func (m *dataPipelineModel) ValidateDataPipelineDestination(
	ctx context.Context,
	req *domain.ValidateDataPipelineDestinationReq,
) bizerr.BizError {
	if req == nil || req.ClusterID == 0 {
		return bizerr.New(bizerr.ErrCodeParams, "cluster ID and destination are required")
	}
	if _, _, err := m.resolveScope(ctx, req.ClusterID); err != nil {
		return err
	}
	if _, planErr := planner.SelectExecutionPlan(
		req.SyncMode, req.Destination.Type,
		domain.DataPipelineStagingTypeCustomerManagedS3,
	); planErr != nil {
		return invalidPlanError(planErr)
	}
	if strings.TrimSpace(req.Destination.LakeOrgID) == "" {
		return bizerr.New(bizerr.ErrCodeParams, "destination Lake is required")
	}
	if strings.TrimSpace(req.Destination.Warehouse) == "" {
		return bizerr.New(bizerr.ErrCodeParams, "destination warehouse is required")
	}
	if req.Destination.FlushIntervalSeconds > math.MaxInt32 {
		return bizerr.New(bizerr.ErrCodeParams, "flush interval is too large")
	}
	// This endpoint is given a destination and nothing else, so it cannot know
	// whether the eventual staging area will notify or be polled - and the two
	// accept different budgets. Asking whether the budget is plannable either
	// way is the strongest question that can honestly be answered here; Create
	// knows the staging mode and holds the budget to that mode exactly.
	if planErr := validateDeliveryBudget(
		req.SyncMode, int32(req.Destination.FlushIntervalSeconds),
	); planErr != nil {
		return invalidPlanError(planErr)
	}
	return m.ensureDestinationAdapter(req.Destination.Type)
}

// validateDeliveryBudget accepts a budget that at least one staging mode can
// deliver. A budget that neither can is wrong under any later choice.
func validateDeliveryBudget(syncMode domain.DataPipelineSyncMode, totalSeconds int32) error {
	_, polledErr := planner.PlanDeliveryIntervals(syncMode, totalSeconds, false)
	if polledErr == nil {
		return nil
	}
	if _, notifiedErr := planner.PlanDeliveryIntervals(syncMode, totalSeconds, true); notifiedErr == nil {
		return nil
	}
	return polledErr
}

// ensureDestinationAdapter reports whether this deployment can actually drive
// the requested destination, so an unconfigured environment fails here instead
// of part-way through Create.
func (m *dataPipelineModel) ensureDestinationAdapter(
	destination domain.DataPipelineDestinationType,
) bizerr.BizError {
	if destination != domain.DataPipelineDestinationTypeTiDBLake {
		return bizerr.New(
			bizerr.ErrCodeParams, "unsupported data pipeline destination %q", destination)
	}
	if m.destinations == nil {
		return bizerr.New(
			bizerr.ErrCodeUnknown, "data pipeline destination client is not configured")
	}
	return nil
}

// ScaleDataPipeline changes the TiCDC capacity of a running Pipeline.
//
// This is not part of Edit. Resizing declares a new expected specification and
// then waits for the shared TiCDC cluster to be reconciled, which the existing
// Changefeed scale workflow already owns; the Changefeed status matrix also only
// allows Scale from a running feed. Delegating keeps that asynchronous
// convergence out of the Pipeline pause/edit/resume path entirely, so the
// Pipeline needs no Scaling state of its own: the child Changefeed reports its
// own progress through the existing observation read model.
func (m *dataPipelineModel) ScaleDataPipeline(
	ctx context.Context,
	req *domain.ScaleDataPipelineReq,
) bizerr.BizError {
	if req == nil || req.UserID == 0 {
		return bizerr.New(bizerr.ErrCodeParams, "scale data pipeline request and user ID are required")
	}
	specification := strings.TrimSpace(req.ChangefeedSpecification)
	if specification == "" {
		return bizerr.New(bizerr.ErrCodeParams, "changefeed specification is required")
	}
	pipeline, err := m.loadScoped(ctx, req.ClusterID, req.PipelineID)
	if err != nil {
		return err
	}
	if pipeline.ExecutionPlan.Incremental == nil {
		return bizerr.New(
			bizerr.ErrCodeParams,
			"data pipeline without incremental sync has no changefeed capacity to scale",
		)
	}
	if pipeline.Status != entity.DataPipelineStatusRunning {
		return invalidStateError(pipeline.Status, "scale")
	}
	if pipeline.Children.ChangefeedID == 0 {
		return bizerr.New(bizerr.ErrCodeParams, "data pipeline has no changefeed to scale")
	}
	if pipeline.ChangefeedSpecification == specification {
		// The child rejects an unchanged specification, and a repeated request
		// is a normal client retry rather than an error.
		return nil
	}
	if err = m.ensureDependencies(); err != nil {
		return err
	}
	if err = m.validator.ValidateChangefeedSpecification(ctx, req.ClusterID, specification); err != nil {
		return err.Chain("validate data pipeline changefeed specification")
	}
	if m.changefeeds == nil {
		return bizerr.New(bizerr.ErrCodeUnknown, "data pipeline changefeed model is unavailable")
	}
	// Scale the child first. The child transaction persists its own desired
	// specification and starts its workflow, so a failure here leaves both rows
	// on the previous value. If the Pipeline row save below fails afterwards,
	// the observation worker re-reads the child and repairs the read model.
	if _, err = m.changefeeds.ScaleBizChangefeed(ctx, &domain.ScaleBizChangefeedReq{
		BizChangefeedID:   pipeline.Children.ChangefeedID,
		SpecificationCode: specification,
		UserID:            req.UserID,
	}); err != nil {
		return err.Chain("scale data pipeline changefeed")
	}
	pipeline.ChangefeedSpecification = specification
	if err = m.repo.Save(ctx, pipeline); err != nil {
		return err.Chain("save scaled data pipeline specification")
	}
	return nil
}

func (m *dataPipelineModel) ResumeDataPipeline(
	ctx context.Context,
	req *domain.ResumeDataPipelineReq,
) bizerr.BizError {
	if req == nil {
		return bizerr.New(bizerr.ErrCodeParams, "resume data pipeline request is required")
	}
	pipeline, err := m.loadScoped(ctx, req.ClusterID, req.PipelineID)
	if err != nil {
		return err
	}
	if pipeline.Params.SyncMode != entity.DataPipelineSyncModeFullAndIncremental ||
		pipeline.Status != entity.DataPipelineStatusPaused {
		return invalidStateError(pipeline.Status, "resume")
	}
	return m.transitionAndStart(
		ctx, pipeline,
		entity.DataPipelineStatusResuming,
		workflow.DagResumeDataPipeline,
		workflow.BusinessActionResumeDataPipeline,
	)
}

func (m *dataPipelineModel) DeleteDataPipeline(
	ctx context.Context,
	req *domain.DeleteDataPipelineReq,
) bizerr.BizError {
	if req == nil {
		return bizerr.New(bizerr.ErrCodeParams, "delete data pipeline request is required")
	}
	pipeline, err := m.loadScoped(ctx, req.ClusterID, req.PipelineID)
	if err != nil {
		return err
	}
	// A repeat is refused rather than accepted and ignored. Answering 200 to a
	// delete that changes nothing tells the caller their request was taken,
	// which is what a Console shows the user; the state is the honest answer,
	// and it is the same one every other lifecycle operation gives.
	//
	// It used to re-submit the workflow here, on the theory that a DAG whose
	// worker died leaves the row with nothing driving it. That recovered a
	// stranded delete at the cost of racing a healthy one: staging ended up
	// with two instances retrying the same step, 206 times and 127 times, over
	// the same row and the same children. Recovering a stranded workflow is
	// the workflow layer's job, not something to buy with duplicate DAGs.
	if pipeline.Status == entity.DataPipelineStatusDeleting {
		return invalidStateError(pipeline.Status, "delete")
	}
	return m.transitionAndStart(
		ctx, pipeline,
		entity.DataPipelineStatusDeleting,
		workflow.DagDeleteDataPipeline,
		workflow.BusinessActionDeleteDataPipeline,
	)
}

func (m *dataPipelineModel) ValidateDataPipelineSourceConnection(
	ctx context.Context,
	req *domain.ValidateDataPipelineSourceConnectionReq,
) bizerr.BizError {
	if req == nil || req.ClusterID == 0 || strings.TrimSpace(req.Username) == "" || req.Password == "" {
		return bizerr.New(bizerr.ErrCodeParams, "cluster ID, source username, and password are required")
	}
	if _, _, err := m.resolveScope(ctx, req.ClusterID); err != nil {
		return err
	}
	if err := m.ensureValidator(); err != nil {
		return err
	}
	// This endpoint checks the account, not the selection. Leaving the filter
	// empty says so: the validator substitutes a wildcard for the upstream
	// call and reports a failure in terms of the credentials alone, which are
	// the only thing this caller supplied and so the only thing it can act on.
	_, err := m.validator.ValidateSource(ctx, req.ClusterID, domain.DataPipelineSourceInput{
		Username: req.Username, Password: req.Password,
	})
	return err
}

func (m *dataPipelineModel) ValidateDataPipelineStaging(
	ctx context.Context,
	req *domain.ValidateDataPipelineStagingReq,
) bizerr.BizError {
	if req == nil || req.ClusterID == 0 {
		return bizerr.New(bizerr.ErrCodeParams, "cluster ID and staging configuration are required")
	}
	if strings.TrimSpace(req.LakeOrgID) == "" {
		return bizerr.New(bizerr.ErrCodeParams, "destination Lake is required")
	}
	if _, _, err := validateStagingInput(req.Staging); err != nil {
		return err
	}
	_, tenantID, err := m.resolveScope(ctx, req.ClusterID)
	if err != nil {
		return err
	}
	if err = m.ensureValidator(); err != nil {
		return err
	}
	return m.validator.ValidateStaging(ctx, data_pipeline.ValidationScope{
		TenantID: tenantID, ClusterID: req.ClusterID, UserID: req.UserID,
		LakeOrgID: req.LakeOrgID,
	}, req.Staging)
}

func (m *dataPipelineModel) planCreate(
	req *domain.CreateDataPipelineReq,
) (domain.DataPipelineExecutionPlan, domain.DeliveryIntervalPlan, string, string, bizerr.BizError) {
	executionPlan, err := planner.SelectExecutionPlan(req.SyncMode, req.Destination.Type, req.Staging.Type)
	if err != nil {
		return domain.DataPipelineExecutionPlan{}, domain.DeliveryIntervalPlan{}, "", "", invalidPlanError(err)
	}
	if req.Destination.FlushIntervalSeconds > math.MaxInt32 {
		return domain.DataPipelineExecutionPlan{}, domain.DeliveryIntervalPlan{}, "", "",
			bizerr.New(bizerr.ErrCodeParams, "flush interval is too large")
	}
	intervalPlan, err := planner.PlanDeliveryIntervals(
		req.SyncMode, int32(req.Destination.FlushIntervalSeconds),
		req.Staging.S3 != nil && req.Staging.S3.UsesSQS())
	if err != nil {
		return domain.DataPipelineExecutionPlan{}, domain.DeliveryIntervalPlan{}, "", "", invalidPlanError(err)
	}
	bucket, prefix, stagingErr := validateStagingInput(req.Staging)
	if stagingErr != nil {
		return domain.DataPipelineExecutionPlan{}, domain.DeliveryIntervalPlan{}, "", "", stagingErr
	}
	return executionPlan, intervalPlan, bucket, prefix, nil
}

// incrementalSpecification keeps the capacity field empty for a full-only
// Pipeline, which has no changefeed to size and rejects Scale.
// planChildren settles every child identity before the row is written. The
// names are pure functions of the Pipeline ID and the Changefeed ID is
// allocated here, so from this point on "does my child already exist?" is a
// lookup on a key this Pipeline owns - there is nothing to discover, and a
// create whose result was lost needs no reconciliation.
//
// The two Lake IDs stay empty: Lake assigns them. Their names are fixed here
// all the same, and the name is what the create path looks up.
func (m *dataPipelineModel) planChildren(
	pipelineID uint64,
	plan entity.DataPipelineExecutionPlan,
) (entity.DataPipelineChildResources, bizerr.BizError) {
	names, err := child.DeterministicResourceNames(pipelineID)
	if err != nil {
		return entity.DataPipelineChildResources{},
			bizerr.New(bizerr.ErrCodeUnknown, "derive data pipeline child names")
	}
	children := entity.DataPipelineChildResources{
		ExportName:         names.Export,
		ExportID:           strconv.FormatUint(m.repo.AllocateID().Uint64(), 10),
		ChangefeedName:     names.Changefeed,
		LakeDataSourceName: names.LakeDataSource,
		LakeTaskName:       names.LakeTask,
	}
	// A full-only Pipeline has no incremental stream, so a zero Changefeed ID
	// keeps its meaning: this Pipeline never has a changefeed.
	if plan.Incremental != nil {
		children.ChangefeedID = m.repo.AllocateID().Uint64()
	}
	return children, nil
}

func incrementalSpecification(
	plan domain.DataPipelineExecutionPlan, specification string,
) string {
	if plan.Incremental == nil {
		return ""
	}
	return strings.TrimSpace(specification)
}

func validateCreateRequest(req *domain.CreateDataPipelineReq) bizerr.BizError {
	if req == nil {
		return bizerr.New(bizerr.ErrCodeParams, "create data pipeline request is required")
	}
	if req.ClusterID == 0 || req.UserID == 0 {
		return bizerr.New(bizerr.ErrCodeParams, "cluster and user IDs are required")
	}
	if name := strings.TrimSpace(req.Name); name == "" || len(name) > maxPipelineNameLength {
		return bizerr.New(bizerr.ErrCodeParams, "data pipeline name must contain 1 to %d bytes", maxPipelineNameLength)
	}
	if strings.TrimSpace(req.Source.Username) == "" || req.Source.Password == "" {
		return bizerr.New(bizerr.ErrCodeParams, "source username and password are required")
	}
	// The warehouse alone does not identify a destination: a TiDB organization
	// may own several Lakes, and two of them can hold a warehouse of the same
	// name.
	if strings.TrimSpace(req.Destination.LakeOrgID) == "" {
		return bizerr.New(bizerr.ErrCodeParams, "destination Lake is required")
	}
	if strings.TrimSpace(req.Destination.Warehouse) == "" {
		return bizerr.New(bizerr.ErrCodeParams, "destination warehouse is required")
	}
	if req.SyncMode == domain.DataPipelineSyncModeFullAndIncremental &&
		strings.TrimSpace(req.ChangefeedSpecification) == "" {
		return bizerr.New(bizerr.ErrCodeParams, "changefeed specification is required for incremental sync")
	}
	return nil
}

// editableStatus reports whether a new desired revision may be submitted. A
// revision the children rejected leaves the row Failed, and Edit has to stay
// open there: replacing the bad value is the only way back, and the Edit
// workflow is idempotent whatever state it finds the stopped children in.
func editableStatus(pipeline *entity.DataPipeline) bool {
	return pipeline.Status == entity.DataPipelineStatusPaused ||
		pipeline.Status == entity.DataPipelineStatusFailed
}

func validateEditRequest(req *domain.EditDataPipelineReq) bizerr.BizError {
	if req == nil || req.ClusterID == 0 || req.PipelineID == 0 || req.UserID == 0 {
		return bizerr.New(bizerr.ErrCodeParams, "cluster, data pipeline, and user IDs are required")
	}
	if name := strings.TrimSpace(req.Name); name == "" || len(name) > maxPipelineNameLength {
		return bizerr.New(bizerr.ErrCodeParams, "data pipeline name must contain 1 to %d bytes", maxPipelineNameLength)
	}
	if strings.TrimSpace(req.Destination.Warehouse) == "" {
		return bizerr.New(bizerr.ErrCodeParams, "destination warehouse is required")
	}
	if req.Destination.FlushIntervalSeconds > math.MaxInt32 {
		return bizerr.New(bizerr.ErrCodeParams, "flush interval is too large")
	}
	return nil
}

// stagingAuthType reads a stored configuration. Rows written before role ARN
// support carry no auth type and always hold a key pair.
func stagingAuthType(s3 entity.DataPipelineS3Config) domain.DataPipelineS3AuthType {
	if s3.UsesRoleARN() {
		return domain.DataPipelineS3AuthTypeRoleARN
	}
	return domain.DataPipelineS3AuthTypeAccessKey
}

func (m *dataPipelineModel) resolveScope(
	ctx context.Context,
	clusterID uint64,
) (*mgmt.Cluster, uint64, bizerr.BizError) {
	if clusterID == 0 || m.clusters == nil {
		return nil, 0, bizerr.New(bizerr.ErrCodeParams, "valid cluster ID and resolver are required")
	}
	cluster, err := m.clusters.GetClusterByID(ctx, clusterID)
	if err != nil {
		return nil, 0, err.Chain("resolve data pipeline cluster")
	}
	if cluster == nil || cluster.GetClusterId() != clusterID || cluster.GetOrganizationId() == 0 {
		return nil, 0, bizerr.New(bizerr.ErrCodePermission, "data pipeline cluster scope is invalid")
	}
	if cluster.GetDeployType() != mgmt.DeployType_DEPLOY_TYPE_NEXTGEN {
		return nil, 0, bizerr.New(bizerr.ErrCodeParams, "data pipeline currently supports Premium clusters only")
	}
	return cluster, cluster.GetOrganizationId(), nil
}

func (m *dataPipelineModel) ensureDependencies() bizerr.BizError {
	if err := m.ensureValidator(); err != nil {
		return err
	}
	if m.repo == nil || m.encryptor == nil || m.workflow == nil || m.clock == nil {
		return bizerr.New(bizerr.ErrCodeUnknown, "data pipeline runtime dependencies are unavailable")
	}
	return nil
}

func (m *dataPipelineModel) ensureValidator() bizerr.BizError {
	if m.validator == nil {
		return bizerr.New(bizerr.ErrCodeUnknown, "data pipeline external validator is unavailable")
	}
	return nil
}

func (m *dataPipelineModel) loadScoped(
	ctx context.Context,
	clusterID uint64,
	pipelineID uint64,
) (*entity.DataPipeline, bizerr.BizError) {
	if clusterID == 0 || pipelineID == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cluster and data pipeline IDs are required")
	}
	_, tenantID, err := m.resolveScope(ctx, clusterID)
	if err != nil {
		return nil, err
	}
	pipeline, err := m.repo.Get(
		ctx,
		data_pipeline_repo.WithIDs(pipelineID),
		data_pipeline_repo.WithTenantID(tenantID),
		data_pipeline_repo.WithClusterID(clusterID),
	)
	if err != nil {
		return nil, err.Chain("load scoped data pipeline")
	}
	return pipeline, nil
}

func (m *dataPipelineModel) transitionAndStart(
	ctx context.Context,
	pipeline *entity.DataPipeline,
	status entity.DataPipelineStatus,
	dagID string,
	action workflow.BusinessAction,
) bizerr.BizError {
	setPipelineStatus(ctx, pipeline, status, m.clock.Now())
	pipeline.LastError = nil
	if err := m.repo.Save(ctx, pipeline); err != nil {
		return err.Chain("save data pipeline transition")
	}
	if err := m.startWorkflow(dagID, action, pipeline); err != nil {
		// Marking the row Failed here would turn a transient workflow-store
		// error into a permanent verdict, so the row keeps the status the
		// caller asked for and the failure is logged.
		m.logWorkflowStartFailure(ctx, pipeline, err)
	}
	return nil
}

func (m *dataPipelineModel) startWorkflow(
	dagID string,
	action workflow.BusinessAction,
	pipeline *entity.DataPipeline,
) error {
	return runDataPipelineWorkflow(m.workflow, dagID, action, pipeline)
}

func runDataPipelineWorkflow(
	workflowModel workflow.Workflow,
	dagID string,
	action workflow.BusinessAction,
	pipeline *entity.DataPipeline,
) error {
	_, err := workflowModel.RunDagWithTags(
		dagID,
		dataPipelineWorkflowVars(pipeline),
		dataPipelineWorkflowTags(pipeline, action),
	)
	return err
}

func dataPipelineWorkflowVars(pipeline *entity.DataPipeline) map[string]string {
	return map[string]string{
		workflow.KeyDataPipelineID: strconv.FormatUint(pipeline.ID.Uint64(), 10),
	}
}

func dataPipelineWorkflowTags(
	pipeline *entity.DataPipeline,
	action workflow.BusinessAction,
) map[string]string {
	return map[string]string{
		workflow.ClusterID:         strconv.FormatUint(pipeline.ClusterID.Uint64(), 10),
		workflow.BusinessTypeKey:   string(workflow.BusinessTypeDataPipeline),
		workflow.BusinessActionKey: string(action),
		workflow.BusinessID:        strconv.FormatUint(pipeline.ID.Uint64(), 10),
	}
}

// logPipelineCreated records everything an investigation needs to find this
// Pipeline's parts in the systems that hold them. The identities are settled at
// insert time and never change, and without this the first step of every
// investigation is an API call to discover the names to search for.
func logPipelineCreated(ctx context.Context, pipeline *entity.DataPipeline) {
	log.GetLogger(ctx).Info(
		"data pipeline created",
		zap.Uint64("pipelineID", pipeline.ID.Uint64()),
		zap.Uint64("clusterID", pipeline.ClusterID.Uint64()),
		zap.String("syncMode", string(pipeline.Params.SyncMode)),
		zap.Any("children", pipeline.Children),
		zap.String("stagingRoot", pipeline.StagingRoot),
		zap.String("stagingBucket", pipeline.Params.Staging.S3.Bucket),
		zap.Bool("notified", pipeline.Params.Staging.S3.UsesSQS()),
		zap.String("lakeOrgID", pipeline.Params.Destination.LakeOrgID),
		zap.String("warehouse", pipeline.Params.Destination.Warehouse),
		zap.Uint64("snapshotTSO", pipeline.SnapshotTSO),
		// How the delivery budget was split, and therefore why the data
		// arrives as fast or as slowly as it does.
		zap.Int32("producerFlushSeconds", pipeline.IntervalPlan.ProducerFlushIntervalSeconds),
		zap.Int32("consumerPollSeconds", pipeline.IntervalPlan.ConsumerPollIntervalSeconds),
	)
}

// logWorkflowStartFailure records a failed DAG submission without changing the
// persisted status. Re-issuing the operation is how the row moves again: every
// step is idempotent - it reads the persisted child reference and adopts by
// deterministic name - so a second submission repeats no work.
func (m *dataPipelineModel) logWorkflowStartFailure(
	ctx context.Context,
	pipeline *entity.DataPipeline,
	workflowErr error,
) {
	log.GetLogger(ctx).Error(
		"failed to start data pipeline workflow; the row keeps its status and the operation must be re-issued",
		zap.Uint64("pipelineID", pipeline.ID.Uint64()),
		zap.String("status", string(pipeline.Status)),
		zap.Error(workflowErr),
	)
}

func dataPipelineDetails(pipeline *entity.DataPipeline) *domain.DataPipelineDetails {
	domainPipeline := domain.DataPipelineEntity2Domain(pipeline)
	observation := pipeline.Observation
	latency := uint64(0)
	if observation.LatencySeconds > 0 {
		latency = uint64(observation.LatencySeconds)
	}
	return &domain.DataPipelineDetails{
		Pipeline: domainPipeline,
		ChildResources: domain.DataPipelineChildResourceDetails{
			Export: domain.DataPipelineResourceDetails{
				Status: observation.ExportStatus, ErrorMessage: observation.ExportErrorMessage,
			},
			Changefeed: domain.DataPipelineResourceDetails{
				Status: observation.ChangefeedStatus, ErrorMessage: observation.ChangefeedErrorMessage,
			},
			Lake: domain.DataPipelineLakeResourceDetails{
				DataSourceStatus:  observation.LakeDataSourceStatus,
				IntegrationStatus: observation.LakeIntegrationStatus,
				ErrorMessage:      observation.LakeErrorMessage,
			},
		},
		Observation: domain.DataPipelineObservationDetails{
			ChangefeedCheckpointTSO:  observation.ChangefeedCheckpointTSO,
			ChangefeedCheckpointTime: cloneTime(observation.ChangefeedCheckpointAt),
			LakeResolvedTSO:          observation.LakeResolvedTSO,
			LakeResolvedTime:         cloneTime(observation.LakeResolvedAt),
			LatencySeconds:           latency,
			ObservedAt:               cloneTime(observation.LastSucceededAt),
			Stale:                    observation.Stale,
			Error:                    clonePipelineError(observation.PollError),
			RowsSynced:               observation.RowsSynced,
			InsertCount:              observation.InsertCount,
			UpdateCount:              observation.UpdateCount,
			DeleteCount:              observation.DeleteCount,
			TablesActive:             observation.TablesActive,
			TablesError:              observation.TablesError,
			DatabasesActive:          observation.DatabasesActive,
		},
	}
}

func cloneTime(value *time.Time) *time.Time {
	if value == nil {
		return nil
	}
	cloned := *value
	return &cloned
}

func clonePipelineError(value *entity.DataPipelineError) *domain.DataPipelineError {
	if value == nil {
		return nil
	}
	cloned := *value
	cloned.OccurredAt = cloneTime(value.OccurredAt)
	return &cloned
}

func invalidStateError(status entity.DataPipelineStatus, action string) bizerr.BizError {
	return bizerr.New(
		bizerr.ErrCodeParams,
		"data pipeline status %s does not allow %s",
		status,
		action,
	)
}

func (*dataPipelineModel) String() string {
	return "dataPipelineModel{[REDACTED]}"
}

// changefeedRoleARN renders the changefeed identity as something a customer can
// paste into a trust policy.
//
// The regional service reports this identity as a bare role name, not an ARN -
// observed on staging as "tidbcloud-<hash>-<suffix>". A trust policy needs the
// full ARN, so this composes one from the account the identity lives in. An
// identity that already arrives as an ARN is left alone, and a non-AWS provider
// has no ARN to build: its identity is a service account, and it passes through
// untouched.
func changefeedRoleARN(profile *domain.DescribeAdminProfileResp, region string) string {
	identity := strings.TrimSpace(profile.CloudIAMIdentity)
	if identity == "" || strings.HasPrefix(identity, "arn:") {
		return identity
	}
	if !strings.EqualFold(profile.ProviderType, "aws") || strings.TrimSpace(profile.CloudAccountID) == "" {
		return identity
	}
	// China regions live in their own partition, and an arn:aws: prefix there
	// names nothing.
	partition := "aws"
	if strings.HasPrefix(strings.ToLower(strings.TrimSpace(region)), "cn-") {
		partition = "aws-cn"
	}
	return fmt.Sprintf("arn:%s:iam::%s:role/%s", partition, profile.CloudAccountID, identity)
}

// GetDataPipelineStagingTrust gathers, in one call, every identity that will
// assume the customer's staging role.
//
// Three principals do, from two systems that report them separately: the
// snapshot and incremental producers share the cluster's changefeed identity,
// and the destination uses one identity to validate the connection and another
// to load the data. A customer who trusts only what a single system reports
// gets a role that passes validation and then fails partway through delivery,
// so an incomplete answer is worse than none: any of the three failing to
// resolve fails the whole call.
func (m *dataPipelineModel) GetDataPipelineStagingTrust(
	ctx context.Context,
	req *domain.GetDataPipelineStagingTrustReq,
) (*domain.GetDataPipelineStagingTrustResp, bizerr.BizError) {
	if req == nil || req.ClusterID == 0 {
		return nil, bizerr.New(bizerr.ErrCodeParams, "cluster ID is required")
	}
	if strings.TrimSpace(req.LakeOrgID) == "" {
		return nil, bizerr.New(bizerr.ErrCodeParams, "destination Lake is required")
	}
	cluster, tenantID, err := m.resolveScope(ctx, req.ClusterID)
	if err != nil {
		return nil, err
	}
	if m.changefeeds == nil || m.destinations == nil {
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "data pipeline runtime dependencies are unavailable")
	}

	profile, profileErr := m.changefeeds.DescribeAdminProfile(
		ctx, &domain.DescribeAdminProfileReq{ClusterID: req.ClusterID})
	if profileErr != nil {
		return nil, profileErr.Chain("get staging trust for the snapshot and incremental producers")
	}
	if strings.TrimSpace(profile.CloudIAMIdentity) == "" {
		return nil, bizerr.New(bizerr.ErrCodeUnknown,
			"the cluster has no changefeed IAM identity yet; retry once it is provisioned")
	}

	setup, setupErr := m.destinations.SetupInfo(ctx, lake.Invocation{
		TiDBOrgID:   strconv.FormatUint(tenantID, 10),
		LakeOrgID:   req.LakeOrgID,
		InitiatedBy: strconv.FormatUint(req.UserID, 10),
	})
	if setupErr != nil {
		return nil, bizerr.WrapError(setupErr, "get staging trust for the destination")
	}

	principals := make([]domain.DataPipelineStagingTrustPrincipal, 0, len(setup.TrustPrincipals)+1)
	principals = append(principals, domain.DataPipelineStagingTrustPrincipal{
		Name:       "TiDB Cloud snapshot and changefeed",
		RoleARN:    changefeedRoleARN(profile, cluster.GetRegion()),
		ExternalID: profile.AWSExternalID,
	})
	for _, principal := range setup.TrustPrincipals {
		if strings.TrimSpace(principal.ARN) == "" {
			return nil, bizerr.New(bizerr.ErrCodeUnknown,
				"the destination reported a trust principal without a role ARN")
		}
		principals = append(principals, domain.DataPipelineStagingTrustPrincipal{
			Name:       principal.Name,
			RoleARN:    principal.ARN,
			ExternalID: setup.ExternalID,
		})
	}
	return &domain.GetDataPipelineStagingTrustResp{
		Principals:    principals,
		BucketActions: append([]string(nil), stagingBucketActions...),
		QueueActions:  append([]string(nil), stagingQueueActions...),
	}, nil
}

// The actions the principals above perform on the staging area. They are
// constants rather than something asked of the two systems, because neither
// reports them: what a customer gets today is a trust policy that names the
// right roles and an access policy they have to guess at, and the guess fails
// during delivery rather than during validation.
//
// The producers write the staged objects and the destination reads them; the
// multipart actions are there because a snapshot of any size is uploaded in
// parts, and an abort permission missing only shows up as orphaned parts the
// customer keeps paying for.
var stagingBucketActions = []string{
	"s3:ListBucket",
	"s3:GetObject",
	"s3:PutObject",
	"s3:DeleteObject",
	"s3:AbortMultipartUpload",
	"s3:ListBucketMultipartUploads",
	"s3:ListMultipartUploadParts",
}

// Consuming a notification queue, for the pipelines that use one. Deleting the
// message is as necessary as receiving it - a consumer that may receive but not
// delete reprocesses the same notification until the queue's retention expires.
//
// This is what the destination needs on the queue. Letting events reach the
// queue in the first place is a separate grant on the queue's own policy, made
// to the s3 or events service principal depending on how the customer wired it,
// rather than to any role named here.
var stagingQueueActions = []string{
	"sqs:ReceiveMessage",
	"sqs:DeleteMessage",
	"sqs:GetQueueAttributes",
	"sqs:GetQueueUrl",
}