package export_svc

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

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/pkgv2/tracing"
	"github.com/tidbcloud/serverless-service/idl/pbgen/proto/v1beta1"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/entity"
	"github.com/tidbcloud/serverless-service/internal/domain/service"
	"github.com/tidbcloud/serverless-service/internal/domain/vo"
	"github.com/tidbcloud/serverless-service/internal/errorcode"
	stashing "github.com/tidbcloud/serverless-service/internal/infra/stashing_area"
	"github.com/tidbcloud/serverless-service/internal/utils"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/attribute"
	"go.uber.org/zap"
)

var v1beta1Tracer = otel.Tracer("export.v1beta1")

const (
	listExportFilesDefaultPageSize = 100
	listExportFilesMaxPageSize     = 1000
	perSignedUrlExpireTime         = 8 * time.Hour
	metaFileName                   = "metadata"
	downloadExportFileMaxNumber    = 100
)

var (
	ListSupportedOrderByField = []string{"create_time"}
	ListOrderByFieldMap       = map[string]string{
		"create_time": "created_at",
	}
)

type exportV1beta1Service struct {
	es                   service.ExportService
	stashingAreaProvider *stashing.StashingAreaProvider
	v1beta1.UnimplementedExportServiceServer
}

func NewExportV1beta1Service(
	es service.ExportService,
	stashingAreaProvider *stashing.StashingAreaProvider,
) v1beta1.ExportServiceServer {
	return &exportV1beta1Service{
		es:                   es,
		stashingAreaProvider: stashingAreaProvider,
	}
}

// ListExports Provides a list of export jobs.
func (e *exportV1beta1Service) ListExports(
	ctx context.Context,
	request *v1beta1.ListExportsRequest,
) (_ *v1beta1.ListExportsResponse, err error) {
	ctx, span := v1beta1Tracer.Start(ctx, "exportV1beta1Service.ListExports")
	logger := log.WithContextL(ctx)
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareExportErrorMessage(ctx, err)
	}()
	span.SetAttributes(
		attribute.String("clusterId", request.ClusterId),
		attribute.String("PageSize", fmt.Sprint(request.PageSize)),
		attribute.String("PageToken", request.PageToken),
	)

	pager := utils.NewAipPager()
	err = pager.Normalize(request)
	if err != nil {
		return nil, err
	}

	// check orderBy
	err = utils.CheckOrderBy(request, ListSupportedOrderByField)
	if err != nil {
		logger.Error("check order by failed", zap.Error(err))
		return nil, bizerr.New(
			errorcode.ErrCodeInvalidOrderby,
			"%s",
			errorcode.ExportErrorMessages[errorcode.ErrCodeInvalidOrderby]+" Error: "+err.Error(),
		)
	}
	orderBy := request.OrderBy
	for k, v := range ListOrderByFieldMap {
		orderBy = strings.ReplaceAll(orderBy, k, v)
	}

	// list exports
	exportVOs, total, err := e.es.ListExports(ctx, &vo.ListExportOpt{
		ClusterID: request.ClusterId,
		PageSize:  int32(pager.PageSize()),
		Offset:    int64(pager.Offset()),
		OrderBy:   orderBy,
	})
	if err != nil {
		logger.Error("list exports failed", zap.Error(err))
		return nil, err
	}

	// build response
	exports := make([]*v1beta1.Export, 0, len(exportVOs))
	for _, exportVO := range exportVOs {
		exports = append(exports, exportVO.ConvertToV1beta1(ctx))
	}
	return &v1beta1.ListExportsResponse{
		Exports:       exports,
		NextPageToken: pager.NextPageToken(total),
		TotalSize:     total,
	}, nil
}

// GetExport Retrieves details of an export job.
func (e *exportV1beta1Service) GetExport(
	ctx context.Context,
	request *v1beta1.GetExportRequest,
) (_ *v1beta1.Export, err error) {
	ctx, span := v1beta1Tracer.Start(ctx, "exportV1beta1Service.GetExport")
	logger := log.WithContextL(ctx)
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareExportErrorMessage(ctx, err)
	}()
	span.SetAttributes(
		attribute.String("clusterId", request.ClusterId),
		attribute.String("exportId", request.ExportId),
	)

	exportVo, err := e.es.GetExport(ctx, request.ClusterId, request.ExportId)
	if err != nil {
		logger.Error("get export failed", zap.Error(err))
		return nil, err
	}

	return exportVo.ConvertToV1beta1(ctx), nil
}

// DeleteExport Deletes a specific export job.
func (e *exportV1beta1Service) DeleteExport(
	ctx context.Context,
	request *v1beta1.DeleteExportRequest,
) (_ *v1beta1.Export, err error) {
	ctx, span := v1beta1Tracer.Start(ctx, "exportV1beta1Service.DeleteExport")
	logger := log.WithContextL(ctx).With(zap.String("exportId", request.ExportId))
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareExportErrorMessage(ctx, err)
	}()
	span.SetAttributes(
		attribute.String("clusterId", request.ClusterId),
		attribute.String("exportId", request.ExportId),
	)

	exportVo, err := e.es.DeleteExport(ctx, request.ClusterId, request.ExportId)
	if err != nil {
		logger.Error("delete export failed", zap.Error(err))
		return nil, err
	}

	logger.Info("delete export success")

	return exportVo.ConvertToV1beta1(ctx), nil
}

// CancelExport Cancel a specific export job.
func (e *exportV1beta1Service) CancelExport(
	ctx context.Context,
	request *v1beta1.CancelExportRequest,
) (_ *v1beta1.Export, err error) {
	ctx, span := v1beta1Tracer.Start(ctx, "exportV1beta1Service.CancelExport")
	logger := log.WithContextL(ctx).With(zap.String("exportId", request.ExportId))
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareExportErrorMessage(ctx, err)
	}()
	span.SetAttributes(
		attribute.String("clusterId", request.ClusterId),
		attribute.String("exportId", request.ExportId),
	)

	exportVo, err := e.es.CancelExport(ctx, request.ClusterId, request.ExportId)
	if err != nil {
		logger.Error("cancel export failed", zap.Error(err))
		return nil, err
	}

	logger.Info("cancel export success")

	return exportVo.ConvertToV1beta1(ctx), nil
}

// CreateExport Create an export job
func (e *exportV1beta1Service) CreateExport(
	ctx context.Context,
	request *v1beta1.CreateExportRequest,
) (_ *v1beta1.Export, err error) {
	ctx, span := v1beta1Tracer.Start(ctx, "exportV1beta1Service.CreateExport")
	logger := log.WithContextL(ctx)
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareExportErrorMessage(ctx, err)
	}()
	span.SetAttributes(
		attribute.String("clusterId", request.ClusterId),
	)

	span.AddEvent("check and build exportVO")
	// for backward compatibility
	if request.Target != nil && request.Target.GetS3() != nil {
		// s3.uri takes precedence over s3.bucketUri
		if request.Target.GetS3().BucketUri != "" && request.Target.GetS3().Uri == "" {
			request.Target.GetS3().Uri = request.Target.GetS3().BucketUri
		}
	}
	exportVo, err := e.buildExportVO(request)
	if err != nil {
		logger.Error("check and build exportVO failed", zap.Error(err))
		return nil, err
	}

	span.AddEvent("create export")
	exportVo, err = e.es.CreateExport(ctx, exportVo)
	if err != nil {
		logger.Error("create export failed", zap.Error(err))
		return nil, err
	}

	logger.Info("create export success", zap.String("exportId", exportVo.ID))
	span.AddEvent("convert export")
	return exportVo.ConvertToV1beta1(ctx), nil
}

// DownloadExport Download an export job
func (e *exportV1beta1Service) DownloadExport(
	ctx context.Context,
	request *v1beta1.DownloadExportRequest,
) (_ *v1beta1.DownloadExportsResponse, err error) {
	ctx, span := v1beta1Tracer.Start(ctx, "exportV1beta1Service.DownloadExport")
	logger := log.WithContextL(ctx).With(zap.String("exportId", request.ExportId))
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareExportErrorMessage(ctx, err)
	}()
	span.SetAttributes(
		attribute.String("clusterId", request.ClusterId),
		attribute.String("exportId", request.ExportId),
	)

	exportVo, err := e.es.GetExport(ctx, request.ClusterId, request.ExportId)
	if err != nil {
		logger.Error("get export failed", zap.Error(err))
		return nil, err
	}

	if exportVo.TargetType != constant.ExportTargetTypeLocal {
		return nil, bizerr.New(errorcode.ErrCodeExportNotLocal, "")
	}
	if exportVo.State != constant.ExportStateSucceed {
		return nil, bizerr.New(errorcode.ErrCodeExportNotSuccessful, "")
	}
	// expiredAt can be nil only when list old export downloads
	if exportVo.ExpiredAt == nil || exportVo.ExpiredAt.Before(time.Now()) {
		return nil, bizerr.New(errorcode.ErrCodeExportExpired, "")
	}

	urls, err := e.createPreSignedUrl(ctx, exportVo)
	if err != nil {
		return nil, err
	}
	return &v1beta1.DownloadExportsResponse{
		Downloads: urls,
	}, nil
}

func (e *exportV1beta1Service) ListExportFiles(
	ctx context.Context,
	request *v1beta1.ListExportFilesRequest,
) (_ *v1beta1.ListExportFilesResponse, err error) {
	ctx, span := v1beta1Tracer.Start(ctx, "exportV1beta1Service.ListExportFiles")
	logger := log.WithContextL(ctx).With(zap.String("exportId", request.ExportId))
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareExportErrorMessage(ctx, err)
	}()
	span.SetAttributes(
		attribute.String("clusterId", request.ClusterId),
		attribute.String("exportId", request.ExportId),
	)

	// check request
	switch {
	case request.PageSize < 0:
		return nil, bizerr.New(errorcode.ErrCodeNegativePageSize, "page size must be positive")
	case request.PageSize == 0:
		request.PageSize = listExportFilesDefaultPageSize
	case request.PageSize > listExportFilesMaxPageSize:
		logger.Warn("specify page size is larger than max page size",
			zap.Int32("specify page size", request.PageSize), zap.Int("max page size", listExportFilesMaxPageSize))
		request.PageSize = listExportFilesMaxPageSize
	}

	exportVo, err := e.es.GetExport(ctx, request.ClusterId, request.ExportId)
	if err != nil {
		logger.Error("get export failed", zap.Error(err))
		return nil, err
	}
	if exportVo.TargetType != constant.ExportTargetTypeLocal {
		return nil, bizerr.New(errorcode.ErrCodeExportNotLocal, "")
	}
	if exportVo.State != constant.ExportStateSucceed {
		return nil, bizerr.New(errorcode.ErrCodeExportNotSuccessful, "")
	}
	if exportVo.ExpiredAt == nil || exportVo.ExpiredAt.Before(time.Now()) {
		return nil, bizerr.New(errorcode.ErrCodeExportExpired, "")
	}

	files, nextPageToken, err := e.listExportFilesFromS3(
		ctx,
		request.PageSize,
		request.PageToken,
		exportVo,
		request.GenerateUrl,
	)
	if err != nil {
		return nil, err
	}

	return &v1beta1.ListExportFilesResponse{
		Files:         files,
		NextPageToken: nextPageToken,
	}, nil
}

func (e *exportV1beta1Service) buildExportVO(request *v1beta1.CreateExportRequest) (_ *vo.Export, err bizerr.BizError) {
	// Truncate with millisecond to avoid rounding when saving to database
	snapshotTime := time.Now().UTC().Truncate(time.Millisecond)
	exportVo := &vo.Export{
		ClusterID:    request.ClusterId,
		SnapshotTime: &snapshotTime,
	}
	exportVo.SetDisplayName(request.DisplayName)
	// add options
	if err := e.addExportOptions(exportVo, request.ExportOptions); err != nil {
		return nil, err
	}
	// add target
	if request.Target != nil {
		switch request.Target.Type {
		case v1beta1.ExportTargetType_LOCAL:
			exportVo.TargetType = constant.ExportTargetTypeLocal
		case v1beta1.ExportTargetType_S3:
			err = e.addExportS3Target(exportVo, request.Target.GetS3())
		case v1beta1.ExportTargetType_GCS:
			err = e.addExportGCSTarget(exportVo, request.Target.GetGcs())
		case v1beta1.ExportTargetType_AZURE_BLOB:
			err = e.addExportAzBlobTarget(exportVo, request.Target.GetAzureBlob())
		case v1beta1.ExportTargetType_OSS:
			err = e.addExportOSSTarget(exportVo, request.Target.GetOss())
		default:
			return nil, bizerr.New(errorcode.ErrCodeInvalidTargetType, "")
		}
	}
	return exportVo, err
}

func (e *exportV1beta1Service) addExportAzBlobTarget(
	exportVo *vo.Export,
	target *v1beta1.AzureBlobTarget,
) bizerr.BizError {
	exportVo.TargetType = constant.ExportTargetTypeAzureBlob
	if target == nil || target.Uri == "" {
		return bizerr.New(errorcode.ErrCodeAzBlobAuth, "Empty Azure Blob URI")
	}
	switch target.AuthType {
	case v1beta1.ExportAzureBlobAuthType_SAS_TOKEN:
		if target.GetSasToken() == "" {
			return bizerr.New(errorcode.ErrCodeAzBlobAuth, "Empty Azure Blob SAS token")
		}
		exportVo.AzBlobURI = target.Uri
		exportVo.AzBlobSasToken = target.GetSasToken()
	default:
		return bizerr.New(errorcode.ErrCodeAzBlobAuth, "Empty Azure Blob auth type, please specify one of [SAS_TOKEN]")
	}
	return nil
}

func (e *exportV1beta1Service) addExportGCSTarget(exportVo *vo.Export, target *v1beta1.GCSTarget) bizerr.BizError {
	exportVo.TargetType = constant.ExportTargetTypeGCS
	if target == nil || target.Uri == "" {
		return bizerr.New(errorcode.ErrCodeGCSAuth, "Empty GCS URI")
	}
	exportVo.GCSURI = target.Uri
	switch target.AuthType {
	case v1beta1.ExportGcsAuthType_SERVICE_ACCOUNT_KEY:
		if err := exportVo.SetGCSServiceAccountKey(target.GetServiceAccountKey()); err != nil {
			return err
		}
	default:
		return bizerr.New(
			errorcode.ErrCodeGCSAuth,
			"Empty GCS auth type, please specify one of the [SERVICE_ACCOUNT_KEY]",
		)
	}
	return nil
}

func (e *exportV1beta1Service) addExportOSSTarget(exportVo *vo.Export, target *v1beta1.OSSTarget) bizerr.BizError {
	exportVo.TargetType = constant.ExportTargetTypeOSS
	if target == nil || target.Uri == "" {
		return bizerr.New(bizerr.ErrCodeParams, "Empty OSS URI")
	}
	exportVo.OSSURI = target.Uri
	switch target.AuthType {
	case v1beta1.ExportOSSAuthType_ACCESS_KEY:
		accessKey := target.GetAccessKey()
		if accessKey == nil || accessKey.Id == "" || accessKey.Secret == "" {
			return bizerr.New(bizerr.ErrCodeParams, "Empty OSS access key id or secret")
		}
		exportVo.OSSAccessKeyID = accessKey.Id
		exportVo.OSSAccessKeySecret = accessKey.Secret
	default:
		return bizerr.New(bizerr.ErrCodeParams, "Empty OSS auth type, please specify one of the [ACCESS_KEY]")
	}
	return nil
}

func (e *exportV1beta1Service) addExportS3Target(exportVo *vo.Export, target *v1beta1.S3Target) bizerr.BizError {
	exportVo.TargetType = constant.ExportTargetTypeS3
	if target == nil || target.Uri == "" {
		return bizerr.New(errorcode.ErrCodeEmptyS3Uri, "")
	}
	exportVo.S3URI = target.Uri

	switch target.AuthType {
	case v1beta1.ExportS3AuthType_ACCESS_KEY:
		accessKey := target.GetAccessKey()
		if accessKey == nil || accessKey.Id == "" || accessKey.Secret == "" {
			return bizerr.New(errorcode.ErrCodeEmptyS3AccessKey, "")
		}
		exportVo.S3AccessKeyID = accessKey.Id
		exportVo.S3SecretAccessKey = accessKey.Secret
	case v1beta1.ExportS3AuthType_ROLE_ARN:
		roleArn := target.GetRoleArn()
		if roleArn == "" {
			return bizerr.New(errorcode.ErrCodeEmptyS3RoleArn, "")
		}
		exportVo.S3RoleArn = roleArn
	default:
		return bizerr.New(errorcode.ErrCodeEmptyS3Auth, "")
	}
	return nil
}

func (e *exportV1beta1Service) addExportOptions(exportVo *vo.Export, options *v1beta1.ExportOptions) bizerr.BizError {
	if options == nil {
		// forward compatible.
		exportVo.Database = "*"
		exportVo.Table = "*"
		return nil
	}
	// forward compatible. filter takes precedence over db&table
	if options.Filter == nil {
		db := "*"
		table := "*"
		if options.Database != "" {
			db = options.Database
		}
		if options.Table != "" {
			table = options.Table
		}
		exportVo.Database = db
		exportVo.Table = table
		dbTable := fmt.Sprintf("%s.%s", db, table)
		if dbTable != "*.*" {
			options.Filter = &v1beta1.ExportOptions_Filter{
				Type: &v1beta1.ExportOptions_Filter_Table_{
					Table: &v1beta1.ExportOptions_Filter_Table{
						Patterns: []string{dbTable},
					},
				},
			}
		}
	}
	// build filter
	if options.Filter != nil {
		switch options.Filter.Type.(type) {
		case *v1beta1.ExportOptions_Filter_Sql:
			sql := options.Filter.GetSql()
			exportVo.SQL = sql
		case *v1beta1.ExportOptions_Filter_Table_:
			dbTable := options.Filter.GetTable()
			if dbTable != nil {
				exportVo.TableFilter = &vo.TableFilter{
					Patterns:   dbTable.Patterns,
					Where:      dbTable.Where,
					Partitions: dbTable.Partitions,
				}
			}
		}
	}
	if len(options.ColumnSelectors) > 0 {
		exportVo.ColumnSelectors = make([]entity.ColumnSelector, 0, len(options.ColumnSelectors))
		for _, columnSelector := range options.ColumnSelectors {
			if columnSelector == nil {
				return bizerr.New(bizerr.ErrCodeParams, "column selector cannot be nil")
			}
			exportVo.ColumnSelectors = append(exportVo.ColumnSelectors, entity.ColumnSelector{
				Matcher: columnSelector.Matcher,
				Columns: columnSelector.Columns,
			})
		}
	}
	// build file type
	switch options.FileType {
	case v1beta1.ExportFileType_SQL:
		exportVo.FileType = constant.ExportFileTypeSQL
	case v1beta1.ExportFileType_CSV:
		exportVo.FileType = constant.ExportFileTypeCSV
	case v1beta1.ExportFileType_PARQUET:
		exportVo.FileType = constant.ExportFileTypePARQUET
		if options.ParquetFormat != nil {
			exportVo.ParquetFormat = &vo.ParquetFormat{}
			switch options.ParquetFormat.Compression {
			case v1beta1.ExportParquetCompressionType_GZIP:
				exportVo.ParquetFormat.Compression = constant.ParquetCompressionGzip
			case v1beta1.ExportParquetCompressionType_SNAPPY:
				exportVo.ParquetFormat.Compression = constant.ParquetCompressionSnappy
			case v1beta1.ExportParquetCompressionType_ZSTD:
				exportVo.ParquetFormat.Compression = constant.ParquetCompressionZstd
			case v1beta1.ExportParquetCompressionType_NONE:
				exportVo.ParquetFormat.Compression = constant.ParquetCompressionNone
			case v1beta1.ExportParquetCompressionType_UNSPECIFIED:
				exportVo.ParquetFormat.Compression = constant.ParquetCompressionZstd
			default:
				return bizerr.New(errorcode.ErrCodeInvalidParquetCompression, "")
			}
		}
	case v1beta1.ExportFileType_TYPE_UNSPECIFIED:
	default:
		return bizerr.New(errorcode.ErrCodeInvalidFileType, "")
	}
	// build csvFormat
	if exportVo.FileType == constant.ExportFileTypeCSV && options.CsvFormat != nil {
		exportVo.CSVFormat = &vo.CSVFormat{
			Delimiter:  options.CsvFormat.Delimiter,
			Separator:  options.CsvFormat.Separator,
			NullValue:  options.CsvFormat.NullValue,
			SkipHeader: options.CsvFormat.SkipHeader,
			Dialect:    csvDialectToDumpling(options.CsvFormat.Dialect),
		}
	}
	// escape backslash is an export-level option (tri-state)
	exportVo.EscapeBackslash = options.EscapeBackslash
	// build compression
	if exportVo.FileType == constant.ExportFileTypePARQUET &&
		options.Compression != v1beta1.ExportCompressionType_NONE &&
		options.Compression != v1beta1.ExportCompressionType_UNSPECIFIED {
		// can not specify compression for parquet
		return bizerr.New(errorcode.ErrCodeSpecifyCompressionInParquet, "")
	}
	switch options.Compression {
	case v1beta1.ExportCompressionType_GZIP:
		exportVo.Compression = constant.CompressionGzip
	case v1beta1.ExportCompressionType_NONE:
		exportVo.Compression = constant.CompressionNone
	case v1beta1.ExportCompressionType_SNAPPY:
		exportVo.Compression = constant.CompressionSnappy
	case v1beta1.ExportCompressionType_ZSTD:
		exportVo.Compression = constant.CompressionZstd
	case v1beta1.ExportCompressionType_UNSPECIFIED:
	default:
		return bizerr.New(errorcode.ErrCodeInvalidCompression, "")
	}
	// snapshot TSO: when set, pin the export snapshot to this exact point so a
	// downstream changefeed can start at the same TSO for a consistent handoff.
	// When unset, buildExportVO has already defaulted the snapshot to now.
	if options.SnapshotTso != nil && *options.SnapshotTso != "" {
		snapshotTime, err := parseSnapshotTSO(*options.SnapshotTso)
		if err != nil {
			return err
		}
		exportVo.SnapshotTime = snapshotTime
	}
	return nil
}

// parseSnapshotTSO converts a TiDB TSO (a uint64 encoded as a string) into the
// UTC wall-clock time used as the export snapshot. A TiDB TSO packs the physical
// time (milliseconds since epoch) in its high bits and an 18-bit logical
// counter in its low bits, so the physical time is `tso >> 18`. The export
// pipeline later reconstructs the effective TSO as `SnapshotTime.UnixMilli() << 18`,
// which is also what GetExport returns as `snapshot_tso`.
func parseSnapshotTSO(tso string) (*time.Time, bizerr.BizError) {
	v, parseErr := strconv.ParseUint(tso, 10, 64)
	if parseErr != nil || v == 0 {
		return nil, bizerr.New(
			errorcode.ErrCodeInvalidSnapshotTime,
			"snapshot_tso must be a positive uint64 encoded as a string",
		)
	}
	physicalMs := int64(v >> 18)
	t := time.UnixMilli(physicalMs).UTC().Truncate(time.Millisecond)
	if t.After(time.Now().UTC()) {
		return nil, bizerr.New(errorcode.ErrCodeInvalidSnapshotTime, "")
	}
	return &t, nil
}

// csvDialectToDumpling maps the API CSV dialect enum to the dumpling
// --csv-output-dialect value. An empty string means the dumpling default.
func csvDialectToDumpling(d v1beta1.ExportCSVDialect_Enum) string {
	switch d {
	case v1beta1.ExportCSVDialect_SNOWFLAKE:
		return "snowflake"
	case v1beta1.ExportCSVDialect_BIGQUERY:
		return "bigquery"
	case v1beta1.ExportCSVDialect_REDSHIFT:
		return "redshift"
	default:
		return ""
	}
}

func (e *exportV1beta1Service) createPreSignedUrl(
	ctx context.Context,
	export *vo.Export,
) ([]*v1beta1.DownloadUrl, bizerr.BizError) {
	logger := log.WithContextL(ctx)

	stashingArea, err := e.stashingAreaProvider.GetStashingArea(export.CloudProvider, export.Region)
	if err != nil {
		logger.Error("get stashing area failed", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeInternalError, "")
	}

	exportPath := stashingArea.GetObjectPath(export.TenantID, stashing.Export, export.ID)
	objs, err := stashingArea.ListAllObjects(ctx, &s3.ListObjectsV2Input{
		Bucket: aws.String(stashingArea.GetBucketName()),
		Prefix: aws.String(exportPath),
	})
	if err != nil {
		logger.Error("list objects failed", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeInternalError, "")
	}
	if len(objs) == 0 {
		logger.Error("no object found", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeExportNoData, "")
	}
	// get all objects
	urls := make([]*v1beta1.DownloadUrl, 0)
	for _, obj := range objs {
		key := obj.Key
		objectUri := strings.Split(*key, "/")
		objectName := objectUri[len(objectUri)-1]
		if objectName == metaFileName {
			continue
		}
		res, err := stashingArea.PresignGetObject(ctx, *key, perSignedUrlExpireTime)
		if err != nil {
			logger.Error("failed to generate pre-signed url", zap.String("key", *key), zap.Error(err))
			return nil, bizerr.New(errorcode.ErrCodeInternalError, "")
		}
		urls = append(urls, &v1beta1.DownloadUrl{
			Url:  res.URL,
			Name: objectName,
			Size: uint32(*obj.Size),
		})
	}
	return urls, nil
}

func (e *exportV1beta1Service) listExportFilesFromS3(ctx context.Context, pageSize int32, pageToken string,
	export *vo.Export, isGenerateUrl bool,
) ([]*v1beta1.ExportFile, string, bizerr.BizError) {
	logger := log.WithContextL(ctx).With(zap.String("export_id", export.ID))

	logger.Info("list export files from s3", zap.Int32("page_size", pageSize), zap.String("page_token", pageToken),
		zap.Bool("isGenerateUrl", isGenerateUrl))

	stashingArea, err := e.stashingAreaProvider.GetStashingArea(export.CloudProvider, export.Region)
	if err != nil {
		logger.Error("get stashing area failed", zap.Error(err))
		return nil, "", bizerr.New(errorcode.ErrCodeInternalError, "")
	}
	// list files
	exportPath := stashingArea.GetObjectPath(export.TenantID, stashing.Export, export.ID)
	input := &s3.ListObjectsV2Input{
		Bucket:  aws.String(stashingArea.GetBucketName()),
		Prefix:  aws.String(exportPath),
		MaxKeys: aws.Int32(pageSize),
	}
	if pageToken != "" {
		input.ContinuationToken = aws.String(pageToken)
	}
	output, err := stashingArea.ListObjects(ctx, input)
	if err != nil {
		logger.Error("list objects failed", zap.Error(err))
		return nil, "", bizerr.New(errorcode.ErrCodeInternalError, "")
	}
	// build response
	files := make([]*v1beta1.ExportFile, 0)
	for _, obj := range output.Contents {
		objectUri := strings.Split(*obj.Key, "/")
		objectName := objectUri[len(objectUri)-1]
		var fileSize uint32
		if obj.Size != nil {
			fileSize = uint32(*obj.Size)
		}
		exportFile := &v1beta1.ExportFile{
			Name: objectName,
			Size: fileSize,
		}
		if isGenerateUrl {
			key := fmt.Sprintf("%s/%s", exportPath, objectName)
			res, err := stashingArea.PresignGetObject(ctx, key, perSignedUrlExpireTime)
			if err != nil {
				logger.Error("failed to generate pre-signed url", zap.String("key", key), zap.Error(err))
				return nil, "", bizerr.New(errorcode.ErrCodeInternalError, "")
			}
			exportFile.Url = res.URL
		}
		files = append(files, exportFile)
	}
	nextPageToken := ""
	if output.NextContinuationToken != nil {
		nextPageToken = *output.NextContinuationToken
	}
	return files, nextPageToken, nil
}

func (e *exportV1beta1Service) DownloadExportFiles(
	ctx context.Context,
	request *v1beta1.DownloadExportFilesRequest,
) (_ *v1beta1.DownloadExportFilesResponse, err error) {
	ctx, span := v1beta1Tracer.Start(ctx, "exportV1beta1Service.DownloadExportFiles")
	logger := log.WithContextL(ctx).With(zap.String("exportId", request.ExportId))
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareExportErrorMessage(ctx, err)
	}()
	span.SetAttributes(
		attribute.String("clusterId", request.ClusterId),
		attribute.String("exportId", request.ExportId),
	)

	if len(request.FileNames) > downloadExportFileMaxNumber {
		return nil, bizerr.New(errorcode.ErrCodeExceedMaxFileNumber, "")
	}

	exportVo, err := e.es.GetExport(ctx, request.ClusterId, request.ExportId)
	if err != nil {
		logger.Error("get export failed", zap.Error(err))
		return nil, err
	}
	if exportVo.TargetType != constant.ExportTargetTypeLocal {
		return nil, bizerr.New(errorcode.ErrCodeExportNotLocal, "")
	}
	if exportVo.State != constant.ExportStateSucceed {
		return nil, bizerr.New(errorcode.ErrCodeExportNotSuccessful, "")
	}
	if exportVo.ExpiredAt == nil || exportVo.ExpiredAt.Before(time.Now()) {
		return nil, bizerr.New(errorcode.ErrCodeExportExpired, "")
	}

	// generate pre-signed urls
	stashingArea, err := e.stashingAreaProvider.GetStashingArea(exportVo.CloudProvider, exportVo.Region)
	if err != nil {
		logger.Error("get stashing area failed", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeInternalError, "")
	}
	exportPath := stashingArea.GetObjectPath(exportVo.TenantID, stashing.Export, exportVo.ID)
	files := make([]*v1beta1.ExportFile, 0)
	for _, fileName := range request.FileNames {
		key := fmt.Sprintf("%s/%s", exportPath, fileName)
		res, err := stashingArea.PresignGetObject(ctx, key, perSignedUrlExpireTime)
		if err != nil {
			logger.Error("failed to generate pre-signed url", zap.String("key", key), zap.Error(err))
			return nil, bizerr.New(errorcode.ErrCodeInternalError, "")
		}
		files = append(files, &v1beta1.ExportFile{
			Name: fileName,
			Url:  res.URL,
		})
	}
	return &v1beta1.DownloadExportFilesResponse{
		Files: files,
	}, nil
}