package child

import (
	"fmt"
	"strconv"

	"github.com/tidbcloud/dataflow-service-ng/app/models/domain"
	globalngv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
	"google.golang.org/protobuf/proto"
)

const (
	csvSeparator = ","
	csvDelimiter = `"`
	csvNullValue = `\N`
)

// escapeBackslash turns off Dumpling's backslash escaping so the snapshot is
// RFC 4180, which is the only dialect the destination reads.
//
// With escaping on, Dumpling writes a quote inside a field as \" and a
// backslash as \\. The destination's CSV reader runs with no escape character
// - Databend's CsvFileFormatParams defaults escape to "", and only TSV defaults
// it to a backslash - so it ends the field at the quote that follows the
// backslash. Every comma after that point then splits the row, and the load
// fails with a column count that does not match the table. A JSON column hits
// this for any value that is not null or a bare number; so does any text value
// holding a double quote.
//
// With escaping off and a delimiter set, Dumpling doubles the delimiter instead
// (escapeCSV in dumpling/export/sql_type.go) and writes backslashes verbatim -
// exactly what a reader with no escape character expects on both counts.
var escapeBackslash = false

// csvDialect selects how Dumpling encodes binary columns. The dialect controls
// nothing else - separator, delimiter and NULL are set explicitly above and are
// the same for every dialect.
//
// SNOWFLAKE means hex. The default writes the raw bytes, which a CSV reader
// cannot reliably recover: a BLOB is free to contain the separator, the
// delimiter, or a newline. Hex costs twice the size and is unambiguous.
var csvDialect = globalngv1.ExportOptions_ExportCSVDialect_SNOWFLAKE

// ExportCreateRequest is safe to print and marshal. Request returns a cloned
// plaintext request solely for the existing management Export client boundary.
type ExportCreateRequest struct {
	request     *globalngv1.CreateExportRequest
	snapshotTSO uint64
}

func (r ExportCreateRequest) Request() *globalngv1.CreateExportRequest {
	if r.request == nil {
		return nil
	}
	return proto.Clone(r.request).(*globalngv1.CreateExportRequest)
}

func (r ExportCreateRequest) SnapshotTSO() uint64 {
	return r.snapshotTSO
}

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

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

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

// BuildExportCreateRequest builds the management Export request and returns
// snapshot TSO separately. Export.snapshot_tso is output-only and is always
// left zero; the adapter passes SnapshotTSO through the request-level boundary.
func BuildExportCreateRequest(
	pipeline *domain.DataPipeline,
	credentials DecryptedCredentials,
) (ExportCreateRequest, error) {
	if err := validatePipelineIdentity(pipeline); err != nil {
		return ExportCreateRequest{}, err
	}
	if err := validateSnapshotPlan(pipeline); err != nil {
		return ExportCreateRequest{}, err
	}
	if credentials.TiDBPassword.Empty() {
		return ExportCreateRequest{}, ErrMissingDecryptedCredential
	}
	if err := validateStagingCredentials(pipeline, credentials.S3SecretAccessKey); err != nil {
		return ExportCreateRequest{}, err
	}
	if pipeline.Params.Source.Username == "" {
		return ExportCreateRequest{}, fmt.Errorf("%w: source username and S3 access key are required", ErrInvalidPipeline)
	}
	// A custom endpoint is meaningful only for Alibaba OSS, which Export
	// addresses through its own oss:// storage shape rather than the S3 one.
	// The staging validator already rejects an endpoint on AWS, so reaching
	// here with one and no OSS provider means the row predates that rule.
	if pipeline.Params.Staging.S3.Endpoint != "" && !pipeline.Params.Staging.S3.UsesOSS() {
		return ExportCreateRequest{}, ErrExportCustomEndpoint
	}
	filters, err := tableFilters(pipeline)
	if err != nil {
		return ExportCreateRequest{}, err
	}
	snapshot, err := buildStagingLocation(pipeline, snapshotDirectory, stagingSchemeFor(pipeline))
	if err != nil {
		return ExportCreateRequest{}, err
	}
	name := pipeline.Children.ExportName
	if name == "" {
		return ExportCreateRequest{}, fmt.Errorf("%w: Export name is required", ErrInvalidPipeline)
	}
	exportID := pipeline.Children.ExportID
	if exportID == "" {
		return ExportCreateRequest{}, fmt.Errorf("%w: Export id is required", ErrInvalidPipeline)
	}

	cloudStorage, storageErr := exportCloudStorage(pipeline, snapshot.URI, credentials.S3SecretAccessKey)
	if storageErr != nil {
		return ExportCreateRequest{}, storageErr
	}

	request := &globalngv1.CreateExportRequest{
		ClusterId: strconv.FormatUint(pipeline.ClusterID, 10),
		Export: &globalngv1.Export{
			// The Pipeline chooses the Export's identity so a create whose
			// result was lost can be resolved with a plain Get.
			ExportId:    exportID,
			DisplayName: name,
			TidbId:      strconv.FormatUint(pipeline.ClusterID, 10),
			Username:    pipeline.Params.Source.Username,
			Password:    credentials.TiDBPassword.reveal(),
			ExportOptions: &globalngv1.ExportOptions{
				FileType:    globalngv1.ExportOptions_CSV,
				Compression: globalngv1.ExportOptions_NONE,
				Filter: &globalngv1.ExportOptions_Filter{
					Type: &globalngv1.ExportOptions_Filter_Table_{
						Table: &globalngv1.ExportOptions_Filter_Table{
							Patterns: append([]string(nil), filters...),
						},
					},
				},
				CsvFormat: &globalngv1.ExportOptions_CSVFormat{
					Separator:  csvSeparator,
					Delimiter:  stringPointer(csvDelimiter),
					NullValue:  stringPointer(csvNullValue),
					SkipHeader: false,
					Dialect:    csvDialect.Enum(),
				},
				EscapeBackslash: &escapeBackslash,
			},
			CloudStorage: cloudStorage,
		},
	}
	return ExportCreateRequest{request: request, snapshotTSO: pipeline.SnapshotTSO}, nil
}

func stringPointer(value string) *string {
	return &value
}

// exportS3Storage renders the staging credentials in the shape the Export API
// takes: an auth type plus a oneof carrying either the key pair or the role.
// exportCloudStorage picks the storage shape the provider is addressed by.
// Export models S3 and OSS as separate branches of one oneof rather than as an
// S3 bucket with an endpoint, so the choice is made here and not smuggled into
// the URI.
func exportCloudStorage(
	pipeline *domain.DataPipeline,
	uri string,
	s3Secret Secret,
) (*globalngv1.CloudStorage, error) {
	if pipeline.Params.Staging.S3.UsesOSS() {
		oss, err := exportOSSStorage(pipeline, uri, s3Secret)
		if err != nil {
			return nil, err
		}
		return &globalngv1.CloudStorage{
			Type:    globalngv1.CloudStorageType_OSS,
			Storage: &globalngv1.CloudStorage_Oss{Oss: oss},
		}, nil
	}
	return &globalngv1.CloudStorage{
		Type: globalngv1.CloudStorageType_S3,
		Storage: &globalngv1.CloudStorage_S3{
			S3: exportS3Storage(pipeline, uri, s3Secret),
		},
	}, nil
}

// exportOSSStorage mirrors exportS3Storage for Alibaba.
//
// Only the key pair is rendered: Export refuses OSS with a role outright
// ("oss role_arn is not supported yet"), so the staging validator rejects that
// combination before a Pipeline exists and there is nothing here to translate.
func exportOSSStorage(
	pipeline *domain.DataPipeline,
	uri string,
	s3Secret Secret,
) (*globalngv1.OSSCloudStorage, error) {
	s3 := pipeline.Params.Staging.S3
	if s3.UsesRoleARN() {
		return nil, fmt.Errorf(
			"%w: role ARN staging is not supported for the Alibaba OSS provider", ErrInvalidPipeline)
	}
	return &globalngv1.OSSCloudStorage{
		Uri:      uri,
		AuthType: globalngv1.OSSCloudStorage_ACCESS_KEY,
		Auth: &globalngv1.OSSCloudStorage_AccessKey_{
			AccessKey: &globalngv1.OSSCloudStorage_AccessKey{
				Id:     s3.AccessKeyID,
				Secret: s3Secret.reveal(),
			},
		},
	}, nil
}

func exportS3Storage(
	pipeline *domain.DataPipeline,
	uri string,
	s3Secret Secret,
) *globalngv1.S3CloudStorage {
	s3 := pipeline.Params.Staging.S3
	if s3.UsesRoleARN() {
		return &globalngv1.S3CloudStorage{
			Uri:      uri,
			AuthType: globalngv1.S3CloudStorage_ROLE_ARN,
			Auth:     &globalngv1.S3CloudStorage_RoleArn{RoleArn: s3.RoleARN},
		}
	}
	return &globalngv1.S3CloudStorage{
		Uri:      uri,
		AuthType: globalngv1.S3CloudStorage_ACCESS_KEY,
		Auth: &globalngv1.S3CloudStorage_AccessKey_{
			AccessKey: &globalngv1.S3CloudStorage_AccessKey{
				Id:     s3.AccessKeyID,
				Secret: s3Secret.reveal(),
			},
		},
	}
}