package child

import (
	"fmt"
	"strconv"

	"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"
	"google.golang.org/protobuf/proto"
)

const (
	canalJSONTerminator    = "\n"
	canalJSONDateSeparator = "none"

	// changefeedFileSizeBytes matches the Console default for S3 changefeeds.
	changefeedFileSizeBytes = 67108864
)

type ChangefeedCreateRequest struct {
	request *domain.CreateBizChangefeedReq
}

func (r ChangefeedCreateRequest) Request() *domain.CreateBizChangefeedReq {
	if r.request == nil {
		return nil
	}
	result := *r.request
	result.DownstreamParameter = proto.Clone(
		r.request.DownstreamParameter.(*dataflow.S3Info),
	).(*dataflow.S3Info)
	return &result
}

func (ChangefeedCreateRequest) String() string {
	return "ChangefeedCreateRequest{" + redactedValue + "}"
}

func (r ChangefeedCreateRequest) GoString() string {
	return r.String()
}

func (ChangefeedCreateRequest) MarshalJSON() ([]byte, error) {
	return marshalRedactedRequest()
}

type ChangefeedEditRequest struct {
	request *domain.ApplyStoppedBizChangefeedConfigReq
}

func (r ChangefeedEditRequest) Request() *domain.ApplyStoppedBizChangefeedConfigReq {
	if r.request == nil {
		return nil
	}
	result := *r.request
	// The Pipeline always stages through S3, so the parameter is an S3Info
	// even though the request it travels in is downstream-agnostic.
	if message, ok := r.request.DownstreamParameter.(proto.Message); ok {
		result.DownstreamParameter = proto.Clone(message)
	}
	return &result
}

func (ChangefeedEditRequest) String() string {
	return "ChangefeedEditRequest{" + redactedValue + "}"
}

func (r ChangefeedEditRequest) GoString() string {
	return r.String()
}

func (ChangefeedEditRequest) MarshalJSON() ([]byte, error) {
	return marshalRedactedRequest()
}

func BuildChangefeedCreateRequest(
	pipeline *domain.DataPipeline,
	s3SecretAccessKey Secret,
) (ChangefeedCreateRequest, error) {
	params, name, err := buildChangefeedParams(pipeline, s3SecretAccessKey)
	if err != nil {
		return ChangefeedCreateRequest{}, err
	}
	return ChangefeedCreateRequest{request: &domain.CreateBizChangefeedReq{
		Name:                name,
		ClusterID:           pipeline.ClusterID,
		DownstreamType:      dataflow.ChangefeedType_S3,
		DownstreamParameter: params,
		SpecificationCode:   pipeline.ChangefeedSpecification,
		UserID:              pipeline.CreatorID,
	}}, nil
}

// BuildChangefeedEditRequest builds the internal stopped-edit request used by
// Data Pipeline Resume. It updates desired S3/filter/interval configuration
// while leaving the Changefeed paused. The specification is deliberately absent:
// resizing goes through the Changefeed scale workflow.
func BuildChangefeedEditRequest(
	pipeline *domain.DataPipeline,
	userID uint64,
	s3SecretAccessKey Secret,
) (ChangefeedEditRequest, error) {
	if userID == 0 {
		return ChangefeedEditRequest{}, ErrInvalidChangefeedActor
	}
	if pipeline == nil || pipeline.Children.ChangefeedID == 0 {
		return ChangefeedEditRequest{}, ErrMissingChangefeedChild
	}
	params, name, err := buildChangefeedParams(pipeline, s3SecretAccessKey)
	if err != nil {
		return ChangefeedEditRequest{}, err
	}
	return ChangefeedEditRequest{request: &domain.ApplyStoppedBizChangefeedConfigReq{
		BizChangefeedID:     pipeline.Children.ChangefeedID,
		Name:                name,
		DownstreamParameter: params,
		UserID:              userID,
	}}, nil
}

func buildChangefeedParams(
	pipeline *domain.DataPipeline,
	s3SecretAccessKey Secret,
) (*dataflow.S3Info, string, error) {
	if err := validatePipelineIdentity(pipeline); err != nil {
		return nil, "", err
	}
	if err := validateSnapshotPlan(pipeline); err != nil {
		return nil, "", err
	}
	if err := validateIncrementalPlan(pipeline); err != nil {
		return nil, "", err
	}
	if err := validateStagingCredentials(pipeline, s3SecretAccessKey); err != nil {
		return nil, "", err
	}
	if pipeline.ChangefeedSpecification == "" {
		return nil, "", fmt.Errorf("%w: changefeed specification is required", ErrInvalidPipeline)
	}
	filters, err := tableFilters(pipeline)
	if err != nil {
		return nil, "", err
	}
	incremental, err := buildStagingLocation(pipeline, incrementalDirectory, planner.StagingSchemeS3)
	if err != nil {
		return nil, "", err
	}
	return &dataflow.S3Info{
		Endpoint: changefeedS3Endpoint(pipeline, incremental.URI, s3SecretAccessKey),
		DataFormat: &dataflow.S3Dataformat{
			Protocol:            dataflow.S3Protocol_S3_PROTOCOL_CANAL_JSON,
			Terminator:          canalJSONTerminator,
			DateSeparator:       canalJSONDateSeparator,
			EnableTidbExtension: true,
		},
		TableConfig: &dataflow.TableConfig{
			FilterRule:    append([]string(nil), filters...),
			Mode:          dataflow.TableMode_IGNORE_NOT_SUPPORT_TABLE,
			CaseSensitive: pipeline.Params.Source.CaseSensitive,
		},
		StartPosition: &dataflow.StartPosition{
			Mode: dataflow.StartMode_FROM_TSO,
			Tso:  pipeline.SnapshotTSO,
		},
		StorageFlushConfig: &dataflow.StorageFlushConfig{
			FlushInterval: strconv.FormatInt(
				int64(pipeline.IntervalPlan.ProducerFlushIntervalSeconds), 10) + "s",
			// The Console fills this in for user-created changefeeds; this path
			// calls the model directly, and TiCDC rejects a zero file size.
			FileSize: changefeedFileSizeBytes,
		},
	}, pipeline.Children.ChangefeedName, nil
}

// changefeedS3Endpoint renders the staging credentials in the shape TiCDC takes:
// a flat endpoint whose auth type selects between the key pair and the role.
func changefeedS3Endpoint(
	pipeline *domain.DataPipeline,
	uri string,
	s3Secret Secret,
) *dataflow.S3Endpoint {
	s3 := pipeline.Params.Staging.S3
	endpoint := &dataflow.S3Endpoint{
		S3Uri:    uri,
		Endpoint: s3.Endpoint,
		Provider: 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 = s3Secret.reveal()
	return endpoint
}