package vo

import (
	"context"
	"encoding/base64"
	"fmt"
	"strconv"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/serverless-service/idl/pbgen/proto/console"
	"github.com/tidbcloud/serverless-service/idl/pbgen/proto/v1beta1"
	"github.com/tidbcloud/serverless-service/internal/config"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	"github.com/tidbcloud/serverless-service/internal/errorcode"
	"github.com/tidbcloud/serverless-service/internal/utils"
	serverlesscs "github.com/tidbcloud/serverless-service/pkg/cloud_storage"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/timestamppb"
)

const maxPartitionNum = 5

type Export struct {
	ID            string                    `json:"id,omitempty"`
	Name          string                    `json:"name,omitempty"`
	State         constant.ExportState      `json:"state,omitempty"`
	Database      string                    `json:"database,omitempty"`
	Table         string                    `json:"table,omitempty"`
	TargetType    constant.ExportTargetType `json:"type,omitempty"`
	LocalURL      string                    `json:"local_url,omitempty"`
	CreatedAt     *time.Time                `json:"created_at,omitempty"`
	UpdatedAt     *time.Time                `json:"updated_at,omitempty"`
	ClusterID     string                    `json:"cluster_id,omitempty"`
	CreatedBy     string                    `json:"created_by,omitempty"`
	CloudProvider string                    `json:"cloud_provider,omitempty"`
	Reason        string                    `json:"reason,omitempty"`
	Region        string                    `json:"region,omitempty"`
	TenantID      string                    `json:"tenant_id,omitempty"`
	CompletedAt   *time.Time                `json:"completed_at,omitempty"`
	Compression   constant.Compression      `json:"compression,omitempty"`
	SnapshotTime  *time.Time                `json:"snapshot_time,omitempty"`
	ExpiredAt     *time.Time                `json:"expired_at,omitempty"`
	DisplayName   string                    `json:"display_mame,omitempty"`
	// Filter
	SQL             string                  `json:"sql,omitempty"`
	TableFilter     *TableFilter            `json:"table_filter,omitempty"`
	ColumnSelectors []entity.ColumnSelector `json:"column_selectors,omitempty"`
	// File Type
	FileType      constant.ExportFileType `json:"file_type,omitempty"`
	CSVFormat     *CSVFormat              `json:"csv_format,omitempty"`
	ParquetFormat *ParquetFormat          `json:"parquet_format,omitempty"`
	// EscapeBackslash is tri-state: nil keeps the server/dumpling default.
	EscapeBackslash *bool `json:"escape_backslash,omitempty"`
	// S3
	S3URI             string `json:"s3_uri,omitempty"`
	S3AccessKeyID     string `json:"s3_access_key_id,omitempty"`
	S3SecretAccessKey string `json:"s3_secret_access_key,omitempty"`
	S3RoleArn         string `json:"s3_role_arn,omitempty"`
	S3Endpoint        string `json:"s3_endpoint,omitempty"`
	// GCS
	GCSURI               string `json:"gcs_uri,omitempty"`
	GCSServiceAccountKey string `json:"gcs_service_account_key,omitempty"`
	// Azure blob
	AzBlobURI      string `json:"az_blob_uri,omitempty"`
	AzBlobSasToken string `json:"az_blob_sas_token,omitempty"`
	// OSS
	OSSURI             string `json:"oss_uri,omitempty"`
	OSSAccessKeyID     string `json:"oss_access_key_id,omitempty"`
	OSSAccessKeySecret string `json:"oss_access_key_secret,omitempty"`
}

type ParquetFormat struct {
	Compression constant.ParquetCompression `json:"compression,omitempty"`
}

type CSVFormat struct {
	Separator  string  `json:"separator,omitempty"`
	Delimiter  *string `json:"delimiter,omitempty"`
	NullValue  *string `json:"null_value,omitempty"`
	SkipHeader bool    `json:"skip_header,omitempty"`
	// Dialect is the dumpling CSV output dialect (e.g. "snowflake"). Empty means
	// the dumpling default dialect.
	Dialect string `json:"dialect,omitempty"`
}

type TableFilter struct {
	Patterns   []string
	Where      string
	Partitions []string
}

type ListExportOpt struct {
	ClusterID string
	PageSize  int32
	Offset    int64
	OrderBy   string
	NameLike  string
}

var ExportDefaultValue = &Export{
	FileType:    constant.ExportFileTypeCSV,
	TargetType:  constant.ExportTargetTypeLocal,
	Compression: constant.CompressionGzip,
	CSVFormat: &CSVFormat{
		Separator:  ",",
		Delimiter:  aws.String("\""),
		NullValue:  aws.String("\\N"),
		SkipHeader: false,
	},
	ParquetFormat: &ParquetFormat{
		Compression: constant.ParquetCompressionZstd,
	},
}

func (e *Export) SetDisplayName(name string) {
	e.DisplayName = name
	if e.DisplayName == "" && e.SnapshotTime != nil {
		e.DisplayName = fmt.Sprintf("SNAPSHOT_%s", e.SnapshotTime.UTC().Format(time.RFC3339))
	}
}

func (e *Export) SetGCSServiceAccountKey(key string) bizerr.BizError {
	if key == "" {
		return bizerr.New(errorcode.ErrCodeGCSAuth, "Empty GCS service account key")
	}
	decodeString, err := DecodeBase64(key)
	if err != nil {
		return bizerr.New(bizerr.ErrCodeParams, "invalid service account key")
	}
	e.GCSServiceAccountKey = base64.URLEncoding.EncodeToString(decodeString)
	return nil
}

func (e *Export) ApplyDefaultValue() {
	// set file type default value
	if e.FileType == "" {
		e.FileType = ExportDefaultValue.FileType
	}
	switch e.FileType {
	case constant.ExportFileTypeCSV:
		if e.CSVFormat != nil {
			if e.CSVFormat.Delimiter == nil {
				e.CSVFormat.Delimiter = ExportDefaultValue.CSVFormat.Delimiter
			}
			// separator' empty string is invalid, regard as default value
			if e.CSVFormat.Separator == "" {
				e.CSVFormat.Separator = ExportDefaultValue.CSVFormat.Separator
			}
			if e.CSVFormat.NullValue == nil {
				e.CSVFormat.NullValue = ExportDefaultValue.CSVFormat.NullValue
			}
		}
	case constant.ExportFileTypePARQUET:
		if e.ParquetFormat == nil {
			e.ParquetFormat = &ParquetFormat{}
		}
		if e.ParquetFormat.Compression == "" {
			e.ParquetFormat.Compression = ExportDefaultValue.ParquetFormat.Compression
		}
		// set compression to none with parquet file type
		e.Compression = constant.CompressionNone
	}
	// set compression default value
	if e.Compression == "" {
		e.Compression = ExportDefaultValue.Compression
	}
	// set target type default value
	if e.TargetType == "" {
		e.TargetType = ExportDefaultValue.TargetType
	}
}

func (e *Export) checkFilter(ctx context.Context) bizerr.BizError {
	logger := log.WithContextL(ctx)
	if e.SQL != "" {
		_, err := utils.ParseSQL(e.SQL)
		if err != nil {
			logger.Error("parse sql failed", zap.Error(err))
			return bizerr.New(errorcode.ErrCodeInvalidSQL, "%s SQL error: %s",
				errorcode.ExportErrorMessages[errorcode.ErrCodeInvalidSQL], err.Error())
		}
	}
	if e.TableFilter != nil {
		if e.TableFilter.Patterns != nil {
			for _, pattern := range e.TableFilter.Patterns {
				if err := utils.ParseTableFilter(pattern); err != nil {
					return bizerr.New(errorcode.ErrCodeInvalidTableFilter, "%s Error filter: %s",
						errorcode.ExportErrorMessages[errorcode.ErrCodeInvalidTableFilter], pattern)
				}
			}
		}
		if e.TableFilter.Where != "" {
			sql := fmt.Sprintf("SELECT * FROM test.t WHERE %s", e.TableFilter.Where)
			_, err := utils.ParseSQL(sql)
			if err != nil {
				logger.Error("parse where failed", zap.Error(err))
				return bizerr.New(errorcode.ErrCodeInvalidWhere, "")
			}
		}
		if len(e.TableFilter.Partitions) > maxPartitionNum {
			return bizerr.New(bizerr.ErrCodeParams, "too many partitions, max to %d", maxPartitionNum)
		}
	}
	for _, columnSelector := range e.ColumnSelectors {
		if len(columnSelector.Matcher) == 0 || len(columnSelector.Columns) == 0 {
			return bizerr.New(bizerr.ErrCodeParams, "column selector matcher and columns are required")
		}
		if err := utils.ParseColumnFilter(columnSelector.Columns); err != nil {
			logger.Error(
				"invalid column selector columns",
				zap.Strings("columns", columnSelector.Columns),
				zap.Error(err),
			)
			return bizerr.New(bizerr.ErrCodeParams, "invalid column selector columns")
		}
		for _, matcher := range columnSelector.Matcher {
			if err := utils.ParseTableFilter(matcher); err != nil {
				logger.Error("invalid column selector matcher", zap.String("matcher", matcher), zap.Error(err))
				return bizerr.New(bizerr.ErrCodeParams, "invalid column selector matcher: %s", matcher)
			}
		}
	}
	return nil
}

// CheckOptions check export options like file type, filter, etc.
func (e *Export) CheckOptions(ctx context.Context) bizerr.BizError {
	// file type
	if e.FileType == constant.ExportFileTypeSQL && e.SQL != "" {
		return bizerr.New(errorcode.ErrCodeNotSupportSQLWithSQLQuery, "")
	}
	// filter
	if e.SQL != "" && e.TableFilter != nil {
		return bizerr.New(errorcode.ErrCodeBothSQLAndTableExist, "")
	}
	if e.SQL != "" && len(e.ColumnSelectors) > 0 {
		return bizerr.New(bizerr.ErrCodeParams, "column selectors cannot be used with SQL statement filters")
	}
	return e.checkFilter(ctx)
}

func (e *Export) ConvertToV1beta1(ctx context.Context) *v1beta1.Export {
	export := &v1beta1.Export{
		ExportId:   e.ID,
		Name:       e.Name,
		State:      convertToV1beta1State(e.State, e.ExpiredAt),
		CreateTime: timestamppb.New(*e.CreatedAt),
		ClusterId:  e.ClusterID,
		CreatedBy:  e.CreatedBy,
		Target:     &v1beta1.ExportTarget{},
		ExportOptions: &v1beta1.ExportOptions{
			Database: e.Database,
			Table:    e.Table,
		},
		DisplayName: e.DisplayName,
	}

	if e.Reason != "" {
		export.Reason = &e.Reason
	}

	if e.UpdatedAt != nil {
		export.UpdateTime = timestamppb.New(*e.UpdatedAt)
	}

	if e.CompletedAt != nil {
		export.CompleteTime = timestamppb.New(*e.CompletedAt)
	}

	if e.SnapshotTime != nil {
		export.SnapshotTime = timestamppb.New(*e.SnapshotTime)
		// The effective TSO the export pipeline uses for the snapshot, matching
		// the conversion in the export domain service. Clients use this exact
		// value as the changefeed start position for a consistent handoff.
		snapshotTso := strconv.FormatInt(e.SnapshotTime.UnixMilli()<<18, 10)
		export.SnapshotTso = &snapshotTso
	}

	if e.ExpiredAt != nil {
		export.ExpireTime = timestamppb.New(*e.ExpiredAt)
	}

	switch e.TargetType {
	case constant.ExportTargetTypeLocal:
		export.Target = &v1beta1.ExportTarget{
			Type: v1beta1.ExportTargetType_LOCAL,
		}
	case constant.ExportTargetTypeS3:
		export.Target = &v1beta1.ExportTarget{
			Type: v1beta1.ExportTargetType_S3,
			Storage: &v1beta1.ExportTarget_S3{
				S3: &v1beta1.S3Target{
					BucketUri: e.S3URI,
					Uri:       e.S3URI,
				},
			},
		}
		if e.S3RoleArn != "" {
			export.Target.GetS3().AuthType = v1beta1.ExportS3AuthType_ROLE_ARN
			export.Target.GetS3().Auth = &v1beta1.S3Target_RoleArn{
				RoleArn: e.S3RoleArn,
			}
		} else if e.S3AccessKeyID != "" {
			export.Target.GetS3().AuthType = v1beta1.ExportS3AuthType_ACCESS_KEY
			export.Target.GetS3().Auth = &v1beta1.S3Target_AccessKey_{
				AccessKey: &v1beta1.S3Target_AccessKey{
					Id: e.S3AccessKeyID,
				},
			}
		}
	case constant.ExportTargetTypeGCS:
		export.Target = &v1beta1.ExportTarget{
			Type: v1beta1.ExportTargetType_GCS,
			Storage: &v1beta1.ExportTarget_Gcs{
				Gcs: &v1beta1.GCSTarget{
					Uri:      e.GCSURI,
					AuthType: v1beta1.ExportGcsAuthType_SERVICE_ACCOUNT_KEY,
				},
			},
		}
	case constant.ExportTargetTypeAzureBlob:
		export.Target = &v1beta1.ExportTarget{
			Type: v1beta1.ExportTargetType_AZURE_BLOB,
			Storage: &v1beta1.ExportTarget_AzureBlob{
				AzureBlob: &v1beta1.AzureBlobTarget{
					Uri:      e.AzBlobURI,
					AuthType: v1beta1.ExportAzureBlobAuthType_SAS_TOKEN,
				},
			},
		}
	case constant.ExportTargetTypeOSS:
		export.Target = &v1beta1.ExportTarget{
			Type: v1beta1.ExportTargetType_OSS,
			Storage: &v1beta1.ExportTarget_Oss{
				Oss: &v1beta1.OSSTarget{
					Uri:      e.OSSURI,
					AuthType: v1beta1.ExportOSSAuthType_ACCESS_KEY,
					Auth: &v1beta1.OSSTarget_AccessKey_{
						AccessKey: &v1beta1.OSSTarget_AccessKey{
							Id: e.OSSAccessKeyID,
						},
					},
				},
			},
		}
	}
	export.ExportOptions = &v1beta1.ExportOptions{
		Database: e.Database,
		Table:    e.Table,
	}
	switch e.FileType {
	case constant.ExportFileTypeSQL:
		export.ExportOptions.FileType = v1beta1.ExportFileType_SQL
	case constant.ExportFileTypeCSV:
		export.ExportOptions.FileType = v1beta1.ExportFileType_CSV
	case constant.ExportFileTypePARQUET:
		export.ExportOptions.FileType = v1beta1.ExportFileType_PARQUET
	}

	switch e.Compression {
	case constant.CompressionGzip:
		export.ExportOptions.Compression = v1beta1.ExportCompressionType_GZIP
	case constant.CompressionNone:
		export.ExportOptions.Compression = v1beta1.ExportCompressionType_NONE
	case constant.CompressionSnappy:
		export.ExportOptions.Compression = v1beta1.ExportCompressionType_SNAPPY
	case constant.CompressionZstd:
		export.ExportOptions.Compression = v1beta1.ExportCompressionType_ZSTD
	}

	// filter
	if e.SQL != "" || e.TableFilter != nil {
		export.ExportOptions.Filter = &v1beta1.ExportOptions_Filter{}
		if e.SQL != "" {
			export.ExportOptions.Filter = &v1beta1.ExportOptions_Filter{
				Type: &v1beta1.ExportOptions_Filter_Sql{
					Sql: e.SQL,
				},
			}
		}
		if e.TableFilter != nil {
			export.ExportOptions.Filter = &v1beta1.ExportOptions_Filter{
				Type: &v1beta1.ExportOptions_Filter_Table_{
					Table: &v1beta1.ExportOptions_Filter_Table{
						Where:      e.TableFilter.Where,
						Patterns:   e.TableFilter.Patterns,
						Partitions: e.TableFilter.Partitions,
					},
				},
			}
		}
	}

	// Column selectors
	if len(e.ColumnSelectors) > 0 {
		export.ExportOptions.ColumnSelectors = make([]*v1beta1.ExportOptions_ColumnSelector, 0, len(e.ColumnSelectors))
		for _, columnSelector := range e.ColumnSelectors {
			export.ExportOptions.ColumnSelectors = append(
				export.ExportOptions.ColumnSelectors,
				&v1beta1.ExportOptions_ColumnSelector{
					Matcher: columnSelector.Matcher,
					Columns: columnSelector.Columns,
				},
			)
		}
	}

	// CSV
	if e.CSVFormat != nil {
		export.ExportOptions.CsvFormat = &v1beta1.ExportOptions_CSVFormat{
			Delimiter:  e.CSVFormat.Delimiter,
			Separator:  e.CSVFormat.Separator,
			NullValue:  e.CSVFormat.NullValue,
			SkipHeader: e.CSVFormat.SkipHeader,
		}
	}
	// Parquet
	if e.ParquetFormat != nil {
		parquetCompression := v1beta1.ExportParquetCompressionType_UNSPECIFIED
		switch e.ParquetFormat.Compression {
		case constant.ParquetCompressionGzip:
			parquetCompression = v1beta1.ExportParquetCompressionType_GZIP
		case constant.ParquetCompressionZstd:
			parquetCompression = v1beta1.ExportParquetCompressionType_ZSTD
		case constant.ParquetCompressionSnappy:
			parquetCompression = v1beta1.ExportParquetCompressionType_SNAPPY
		case constant.ParquetCompressionNone:
			parquetCompression = v1beta1.ExportParquetCompressionType_NONE
		}
		export.ExportOptions.ParquetFormat = &v1beta1.ExportOptions_ParquetFormat{
			Compression: parquetCompression,
		}
	}
	return export
}

func (e *Export) ConvertToConsole(ctx context.Context) *console.Export {
	export := &console.Export{
		ExportId:      e.ID,
		Name:          e.Name,
		State:         convertToConsoleState(e.State, e.ExpiredAt),
		CreateTime:    timestamppb.New(*e.CreatedAt),
		ClusterId:     e.ClusterID,
		CreatedBy:     e.CreatedBy,
		Target:        &console.ExportTarget{},
		ExportOptions: &console.ExportOptions{},
		DisplayName:   e.DisplayName,
	}

	if e.Reason != "" {
		export.Reason = &e.Reason
	}

	if e.UpdatedAt != nil {
		export.UpdateTime = timestamppb.New(*e.UpdatedAt)
	}

	if e.CompletedAt != nil {
		export.CompleteTime = timestamppb.New(*e.CompletedAt)
	}

	if e.SnapshotTime != nil {
		export.SnapshotTime = timestamppb.New(*e.SnapshotTime)
		// The effective TSO the export pipeline uses for the snapshot, matching
		// the conversion in the export domain service. Clients use this exact
		// value as the changefeed start position for a consistent handoff.
		snapshotTso := strconv.FormatInt(e.SnapshotTime.UnixMilli()<<18, 10)
		export.SnapshotTso = &snapshotTso
	}

	if e.ExpiredAt != nil {
		export.ExpireTime = timestamppb.New(*e.ExpiredAt)
	}

	switch e.TargetType {
	case constant.ExportTargetTypeLocal:
		export.Target = &console.ExportTarget{
			Type: console.ExportTargetType_LOCAL,
		}
		if e.State == constant.ExportStateSucceed {
			export.Target.LocalDownloadCommand = fmt.Sprintf("ticloud s export download -c %s -e %s", e.ClusterID, e.ID)
		}
	case constant.ExportTargetTypeS3:
		export.Target = &console.ExportTarget{
			Type: console.ExportTargetType_S3,
			Storage: &console.ExportTarget_S3{
				S3: &console.S3Target{
					Uri: e.S3URI,
				},
			},
		}
		if e.S3RoleArn != "" {
			export.Target.GetS3().AuthType = console.ExportS3AuthType_ROLE_ARN
			export.Target.GetS3().Auth = &console.S3Target_RoleArn{
				RoleArn: e.S3RoleArn,
			}
		} else if e.S3AccessKeyID != "" {
			export.Target.GetS3().AuthType = console.ExportS3AuthType_ACCESS_KEY
			export.Target.GetS3().Auth = &console.S3Target_AccessKey_{
				AccessKey: &console.S3Target_AccessKey{
					Id: e.S3AccessKeyID,
				},
			}
		}
	case constant.ExportTargetTypeGCS:
		export.Target = &console.ExportTarget{
			Type: console.ExportTargetType_GCS,
			Storage: &console.ExportTarget_Gcs{
				Gcs: &console.GCSTarget{
					Uri:      e.GCSURI,
					AuthType: console.ExportGcsAuthType_SERVICE_ACCOUNT_KEY,
				},
			},
		}
	case constant.ExportTargetTypeAzureBlob:
		export.Target = &console.ExportTarget{
			Type: console.ExportTargetType_AZURE_BLOB,
			Storage: &console.ExportTarget_AzureBlob{
				AzureBlob: &console.AzureBlobTarget{
					Uri:      e.AzBlobURI,
					AuthType: console.ExportAzureBlobAuthType_SAS_TOKEN,
				},
			},
		}
	case constant.ExportTargetTypeOSS:
		export.Target = &console.ExportTarget{
			Type: console.ExportTargetType_OSS,
			Storage: &console.ExportTarget_Oss{
				Oss: &console.OSSTarget{
					Uri:      e.OSSURI,
					AuthType: console.ExportOSSAuthType_ACCESS_KEY,
					Auth: &console.OSSTarget_AccessKey_{
						AccessKey: &console.OSSTarget_AccessKey{
							Id: e.OSSAccessKeyID,
						},
					},
				},
			},
		}
	}
	switch e.FileType {
	case constant.ExportFileTypeSQL:
		export.ExportOptions.FileType = console.ExportFileType_SQL
	case constant.ExportFileTypeCSV:
		export.ExportOptions.FileType = console.ExportFileType_CSV
	case constant.ExportFileTypePARQUET:
		export.ExportOptions.FileType = console.ExportFileType_PARQUET
	}

	switch e.Compression {
	case constant.CompressionGzip:
		export.ExportOptions.Compression = console.ExportCompressionType_GZIP
	case constant.CompressionNone:
		export.ExportOptions.Compression = console.ExportCompressionType_NONE
	case constant.CompressionSnappy:
		export.ExportOptions.Compression = console.ExportCompressionType_SNAPPY
	case constant.CompressionZstd:
		export.ExportOptions.Compression = console.ExportCompressionType_ZSTD
	}

	// filter
	if e.SQL != "" {
		export.ExportOptions.Filter = &console.ExportOptions_Filter{
			Type: &console.ExportOptions_Filter_Sql{
				Sql: e.SQL,
			},
		}
	}
	if e.TableFilter != nil {
		export.ExportOptions.Filter = &console.ExportOptions_Filter{
			Type: &console.ExportOptions_Filter_Table_{
				Table: &console.ExportOptions_Filter_Table{
					Where:      e.TableFilter.Where,
					Patterns:   e.TableFilter.Patterns,
					Partitions: e.TableFilter.Partitions,
				},
			},
		}
	}
	// Column selectors
	if len(e.ColumnSelectors) > 0 {
		export.ExportOptions.ColumnSelectors = make([]*console.ExportOptions_ColumnSelector, 0, len(e.ColumnSelectors))
		for _, columnSelector := range e.ColumnSelectors {
			export.ExportOptions.ColumnSelectors = append(
				export.ExportOptions.ColumnSelectors,
				&console.ExportOptions_ColumnSelector{
					Matcher: columnSelector.Matcher,
					Columns: columnSelector.Columns,
				},
			)
		}
	}

	// CSV
	if e.CSVFormat != nil {
		export.ExportOptions.CsvFormat = &console.ExportOptions_CSVFormat{
			Delimiter:  e.CSVFormat.Delimiter,
			Separator:  e.CSVFormat.Separator,
			NullValue:  e.CSVFormat.NullValue,
			SkipHeader: e.CSVFormat.SkipHeader,
		}
	}
	// Parquet
	if e.ParquetFormat != nil {
		parquetCompression := console.ExportParquetCompressionType_UNSPECIFIED
		switch e.ParquetFormat.Compression {
		case constant.ParquetCompressionGzip:
			parquetCompression = console.ExportParquetCompressionType_GZIP
		case constant.ParquetCompressionZstd:
			parquetCompression = console.ExportParquetCompressionType_ZSTD
		case constant.ParquetCompressionSnappy:
			parquetCompression = console.ExportParquetCompressionType_SNAPPY
		case constant.ParquetCompressionNone:
			parquetCompression = console.ExportParquetCompressionType_NONE
		}
		export.ExportOptions.ParquetFormat = &console.ExportOptions_ParquetFormat{
			Compression: parquetCompression,
		}
	}
	return export
}

func (e *Export) GenerateExportName() {
	e.Name = fmt.Sprintf("clusters/%s/exports/%s", e.ClusterID, e.ID)
}

func convertToV1beta1State(state constant.ExportState, expiredAt *time.Time) v1beta1.ExportState_Enum {
	switch state {
	case constant.ExportStatePreparing, constant.ExportStateRunning:
		return v1beta1.ExportState_RUNNING
	case constant.ExportStateCanceled:
		return v1beta1.ExportState_CANCELED
	case constant.ExportStateSucceed:
		if expiredAt != nil && expiredAt.Before(time.Now()) {
			return v1beta1.ExportState_EXPIRED
		}
		return v1beta1.ExportState_SUCCEEDED
	case constant.ExportStateFailed:
		return v1beta1.ExportState_FAILED
	case constant.ExportStateDeleted:
		return v1beta1.ExportState_DELETED
	default:
		return v1beta1.ExportState_UNSPECIFIED
	}
}

func convertToConsoleState(state constant.ExportState, expiredAt *time.Time) console.ExportState_Enum {
	switch state {
	case constant.ExportStatePreparing, constant.ExportStateRunning:
		return console.ExportState_RUNNING
	case constant.ExportStateCanceled:
		return console.ExportState_CANCELED
	case constant.ExportStateSucceed:
		if expiredAt != nil && expiredAt.Before(time.Now()) {
			return console.ExportState_EXPIRED
		}
		return console.ExportState_SUCCEEDED
	case constant.ExportStateFailed:
		return console.ExportState_FAILED
	case constant.ExportStateDeleted:
		return console.ExportState_DELETED
	default:
		return console.ExportState_UNSPECIFIED
	}
}

// ToCloudStorage converts Export's cloud storage fields to entity.CloudStorage.
func (e *Export) ToCloudStorage(region string) *entity.CloudStorage {
	if e == nil {
		return nil
	}
	out := &entity.CloudStorage{}
	switch e.TargetType {
	case constant.ExportTargetTypeLocal:
		out.Type = constant.CloudStorageTypeTiDBCloud
		return out
	case constant.ExportTargetTypeS3:
		out.Type = constant.CloudStorageTypeS3
		out.Uri = e.S3URI
		out.S3 = &entity.S3CloudStorage{}
		if e.S3RoleArn != "" {
			profile := serverlesscs.GetAWSProfile(config.Global().ServerConfig.Stage, region)
			externalId := ""
			if profile != nil {
				externalId = profile.ExternalId
			}
			out.S3.RoleArn = &entity.S3CloudStorageRoleArn{
				RoleArn:    e.S3RoleArn,
				ExternalId: externalId,
			}
		}
		if e.S3AccessKeyID != "" && e.S3SecretAccessKey != "" {
			out.S3.AccessKey = &entity.S3CloudStorageAccessKey{
				AccessKeyId:     e.S3AccessKeyID,
				SecretAccessKey: e.S3SecretAccessKey,
			}
		}
	case constant.ExportTargetTypeGCS:
		out.Type = constant.CloudStorageTypeGCS
		out.Uri = e.GCSURI
		out.GCS = &entity.GCSCloudStorage{
			ServiceAccountKey: e.GCSServiceAccountKey,
		}
	case constant.ExportTargetTypeAzureBlob:
		out.Type = constant.CloudStorageTypeAzureBlob
		out.Uri = e.AzBlobURI
		out.AzureBlob = &entity.AzureBlobCloudStorage{
			SasToken: e.AzBlobSasToken,
		}
	case constant.ExportTargetTypeOSS:
		out.Type = constant.CloudStorageTypeOSS
		out.Uri = e.OSSURI
		out.OSS = &entity.OSSCloudStorage{}
		if e.OSSAccessKeyID != "" && e.OSSAccessKeySecret != "" {
			out.OSS.AccessKey = &entity.OSSCloudStorageAccessKey{
				AccessKeyId:     e.OSSAccessKeyID,
				AccessKeySecret: e.OSSAccessKeySecret,
			}
		}
	default:
		return nil
	}
	return out
}