package console

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"slices"
	"strconv"
	"strings"
	"sync"
	"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/import/precheck"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/pkgv2/tracing"
	"github.com/tidbcloud/serverless-service/idl/pbgen/proto/console"
	"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/domain/facade"
	"github.com/tidbcloud/serverless-service/internal/domain/vo"
	"github.com/tidbcloud/serverless-service/internal/errorcode"
	"github.com/tidbcloud/serverless-service/internal/service/import_svc"
	"github.com/tidbcloud/serverless-service/internal/utils"
	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	regionalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/regional/serverless/v1"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/emptypb"
)

func (s *importService) Precheck(ctx context.Context, req *console.PrecheckReq) (_ *console.PrecheckResp, err error) {
	ctx, span := Tracer.Start(ctx, "console.import.Precheck")
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareImportErrorMessage(ctx, err)
	}()

	// Add 50s timeout
	ctx, cancel := context.WithTimeout(ctx, 50*time.Second)
	defer cancel()

	logger := log.WithContextL(ctx).With(zap.String("cluster_id", req.ClusterId))
	bizErr := validatePrecheckReq(req)
	if bizErr != nil {
		return nil, bizErr
	}

	cID, err := strconv.ParseUint(req.ClusterId, 10, 64)
	if err != nil {
		logger.Error("invalid cluster id", zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid cluster id")
	}

	user, err := utils.GetUserFromContext(ctx)
	if err != nil {
		logger.Error("user not found", zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeUnknown, "user not found")
	}
	span.AddEvent("get JWT")
	jwt, err := s.accountFacade.GetJwtByUser(ctx, user.Email, user.ID, cID)
	if err != nil {
		logger.Error("failed to get jwt", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeInternalError, "")
	}

	infos, err := vo.GetTargetTableInfos(ctx, req.GetSource())
	if err != nil {
		return nil, err
	}
	fileRules, bizErr := generateFileRouter(ctx, req.GetImportOptions().GetFileType(), infos)
	if bizErr != nil {
		return nil, bizErr
	}

	span.AddEvent("get cluster")
	cluster, err := s.mgmtSvc.GetCluster(ctx, cID, tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC)
	if err != nil {
		logger.Error("failed to get cluster", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeInternalError, "")
	}
	if cluster.GetServerlessCluster().
		GetUserStatus() !=
		globalserverlessv1.UserClusterStatus_USER_CLUSTER_STATUS_NORMAL {
		return nil, bizerr.New(errorcode.ErrCodeClusterNotRunning, "cluster is not running")
	}

	getExternalIdFunc := func() (string, bizerr.BizError) {
		profile, bizErr := s.importSvc.GetDataIngestionProfile(ctx, cluster)
		if bizErr != nil {
			logger.Error("failed to get data ingestion profile", zap.Error(bizErr))
			return "", bizErr
		}
		return profile.ExternalId, nil
	}

	cs, bizErr := buildCloudStorageFromImportSource(req.GetSource(), getExternalIdFunc)
	if bizErr != nil {
		logger.Error("failed to build cloud storage from import source", zap.Error(bizErr))
		return nil, bizErr
	}
	span.AddEvent("call Precheck")
	logger.Debug("call Precheck", zap.String("provider", cluster.CloudProvider), zap.String("region", cluster.Region))
	res, err := s.mgmtSvc.Precheck(ctx, &globalserverlessv1.PrecheckRequest{
		CloudStorage: cs,
		DbCredential: &regionalserverlessv1.DatabaseCredential{
			UserName:   jwt.UserName,
			Credential: jwt.Jwt,
		},
		FileRouteRules:       fileRules,
		UseCustomFileMapping: len(infos) != 0,
		ClusterId:            strconv.FormatUint(cID, 10),
	})
	if err != nil {
		if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
			return &console.PrecheckResp{
				Status: console.ImportPrecheckStatusType_TIMEOUT,
			}, nil
		}
		return nil, err
	}
	return convertToPrecheckResp(res), nil
}

func generateFileRouter(
	ctx context.Context,
	fileType console.ImportFileType_Enum,
	infos []*console.ImportTargetTableInfo,
) ([]*regionalserverlessv1.FileRouteRule, bizerr.BizError) {
	var rules []*regionalserverlessv1.FileRouteRule

	if len(infos) == 0 {
		rules = append(
			rules,
			&regionalserverlessv1.FileRouteRule{
				Pattern: utils.DefaultDBSchemaFilePattern,
				Schema:  "$1",
				Type:    precheck.SchemaSchema,
			},
			&regionalserverlessv1.FileRouteRule{
				Pattern: utils.DefaultTableSchemaFilePattern,
				Schema:  "$1",
				Table:   "$2",
				Type:    precheck.TableSchema,
			},
		)
		rule, err := getDefaultFileRule(fileType)
		if err != nil {
			return nil, err
		}
		if rule != nil {
			rules = append(rules, rule)
		}
	} else {
		rs, err := getCustomFileRules(infos, fileType)
		if err != nil {
			return nil, err
		}
		rules = append(rules, rs...)
	}
	log.WithContextL(ctx).Debug("create file router rule", zap.Any("rules", rules))

	return rules, nil
}

func getDefaultFileRule(fileType console.ImportFileType_Enum) (*regionalserverlessv1.FileRouteRule, bizerr.BizError) {
	switch fileType {
	case console.ImportFileType_CSV:
		return &regionalserverlessv1.FileRouteRule{
			Pattern: utils.DefaultCSVFilePattern,
			Schema:  "$1",
			Table:   "$2",
			Type:    precheck.TypeCSV,
		}, nil
	case console.ImportFileType_PARQUET:
		return &regionalserverlessv1.FileRouteRule{
			Pattern: utils.DefaultRawParquetFilePattern,
			Schema:  "$1",
			Table:   "$2",
			Type:    precheck.TypeParquet,
		}, nil
	case console.ImportFileType_SQL:
		return &regionalserverlessv1.FileRouteRule{
			Pattern: utils.DefaultSQLFilePattern,
			Schema:  "$1",
			Table:   "$2",
			Type:    precheck.TypeSQL,
		}, nil
	case console.ImportFileType_AURORA_SNAPSHOT:
		return &regionalserverlessv1.FileRouteRule{
			Pattern: utils.DefaultAuroraSnapshotFilePattern,
			Schema:  "$1",
			Table:   "$2",
			Type:    precheck.TypeParquet,
		}, nil
	default:
		return nil, bizerr.New(bizerr.ErrCodeParams, "unsupported file type")
	}
}

func getCustomFileRules(
	infos []*console.ImportTargetTableInfo,
	fileType console.ImportFileType_Enum,
) ([]*regionalserverlessv1.FileRouteRule, bizerr.BizError) {
	rules := make([]*regionalserverlessv1.FileRouteRule, 0, len(infos))

	for _, info := range infos {
		if info.CustomFile == "" {
			return nil, bizerr.New(bizerr.ErrCodeParams, "file rule pattern is empty")
		}

		var t string
		switch fileType {
		case console.ImportFileType_CSV:
			t = precheck.TypeCSV
		case console.ImportFileType_PARQUET, console.ImportFileType_AURORA_SNAPSHOT:
			t = precheck.TypeParquet
		case console.ImportFileType_SQL:
			t = precheck.TypeSQL
		default:
			return nil, bizerr.New(bizerr.ErrCodeParams, "unsupported file type")
		}

		rules = append(rules, &regionalserverlessv1.FileRouteRule{
			Pattern: utils.GenCustomNamingRuleRegexp(info.CustomFile),
			Schema:  utils.EscapeFieldExtractor(info.GetTargetTable().Database),
			Table:   utils.EscapeFieldExtractor(info.GetTargetTable().Table),
			Type:    t,
		})
	}

	return rules, nil
}

func convertToPrecheckResp(result *globalserverlessv1.PrecheckResponse) *console.PrecheckResp {
	resp := &console.PrecheckResp{
		Status: convertPrecheckStatusToEnum(result.Status),
	}

	if len(result.TableResults) == 0 {
		return resp
	}

	var totalFiles uint64
	tablesWithErrors := make([]string, 0)

	// Process all results - each result represents a unique table
	for _, res := range result.TableResults {
		tableKey := res.Table.Schema + "." + res.Table.Table

		// Create table result and add to response
		tableResult := &console.TableResult{
			TargetTable: &console.Table{
				Database: res.Table.Schema,
				Table:    res.Table.Table,
			},
			MatchedDataFiles:  res.DataFiles,
			MatchedSchemaFile: res.SchemaFile,
			UseSchemaFile:     res.UseSchemaFile,
		}

		// Handle tables with errors
		if res.ErrorMsg != "" {
			tablesWithErrors = append(tablesWithErrors, tableKey)
			tableResult.ErrorMessage = res.ErrorMsg
		}

		resp.TargetTables = append(resp.TargetTables, tableResult)
		totalFiles += uint64(len(res.DataFiles))
	}

	// Set counts and warning message
	resp.TotalTablesCount = uint64(len(result.TableResults))
	resp.TotalDataFilesCount = totalFiles

	if len(tablesWithErrors) > 0 {
		resp.ErrorMessage = fmt.Sprintf("Found %d table(s) with error: %s",
			len(tablesWithErrors), strings.Join(tablesWithErrors, ", "))
	}

	return resp
}

func convertPrecheckStatusToEnum(
	status regionalserverlessv1.ImportPrecheckStatus,
) console.ImportPrecheckStatusType_Enum {
	switch status {
	case regionalserverlessv1.ImportPrecheckStatus_COMPLETED:
		return console.ImportPrecheckStatusType_COMPLETED
	case regionalserverlessv1.ImportPrecheckStatus_TRUNCATED:
		return console.ImportPrecheckStatusType_TRUNCATED
	case regionalserverlessv1.ImportPrecheckStatus_TIMEOUT:
		return console.ImportPrecheckStatusType_TIMEOUT
	default:
		return console.ImportPrecheckStatusType_UNSPECIFIED
	}
}

func (s *importService) PreviewLocal(
	ctx context.Context,
	req *console.ImportsPreviewLocalReq,
) (_ *console.ImportsPreviewLocalResp, err error) {
	ctx, span := Tracer.Start(ctx, "console.import.PreviewLocal")
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareImportErrorMessage(ctx, err)
	}()

	logger := log.WithContextL(ctx).With(zap.String("cluster_id", req.ClusterId))
	if req.CsvFormat == nil {
		req.CsvFormat = &console.CSVFormat{}
	}
	bizErr := validateCSVFormat(ctx, req.CsvFormat)
	if bizErr != nil {
		return nil, bizErr
	}

	span.AddEvent("call GetUploadByID")
	idCompat, err := strconv.ParseUint(req.GetUploadId(), 10, 64)
	var upload *entity.Upload
	if err == nil {
		upload, bizErr = s.uploadRepo.GetUploadByIDCompat(ctx, idCompat)
		if bizErr != nil {
			return nil, bizErr
		}
	} else {
		upload, bizErr = s.uploadRepo.GetUploadByID(ctx, req.GetUploadId())
		if bizErr != nil {
			return nil, bizErr
		}
	}
	clusterID, err := strconv.ParseUint(req.ClusterId, 10, 64)
	if err != nil {
		logger.Error("invalid cluster id", zap.Error(err), zap.String("cluster_id", req.ClusterId))
		return nil, bizerr.New(bizerr.ErrCodeParams, "invalid cluster id")
	}
	bizErr = import_svc.ValidateUpload(ctx, upload, clusterID)
	if bizErr != nil {
		return nil, bizErr
	}

	span.AddEvent("get import file from stashing area")
	area, err := s.stashingAreaProvider.GetStashingArea(upload.GetCloudProvider(), upload.RegionName)
	if err != nil {
		logger.Error("GetStashingArea failed", zap.Error(err))
		return nil, bizerr.New(errorcode.ErrCodeInternalError, errorcode.InternalErrorMessage)
	}

	object, err := area.GetObject(ctx, &s3.GetObjectInput{
		Bucket: aws.String(area.GetBucketName()),
		Key:    aws.String(upload.BucketKey),
		Range:  aws.String(fmt.Sprintf("bytes=0-%d", constant.PreviewReadSize)),
	})
	if err != nil {
		logger.Error(
			"GetObject failed",
			zap.Error(err),
			zap.String("bucket", area.GetBucketName()),
			zap.String("key", upload.BucketKey),
		)
		return nil, bizerr.New(errorcode.ErrCodeInternalError, errorcode.InternalErrorMessage)
	}
	defer object.Body.Close()

	dbPreviewResult, bizErr := import_svc.PreviewLocalImport(
		ctx,
		object.Body,
		vo.Console2CSVConfig(req.CsvFormat),
		vo.ConsoleFileType2Entity(req.GetFileType()),
	)
	if bizErr != nil {
		logger.Error("PreviewLocalImport failed", zap.Error(err))
		return nil, bizErr
	}
	if len(dbPreviewResult.GetTables()) < 1 {
		logger.Error("no preview result for the table")
		return nil, bizerr.New(
			bizerr.ErrCodeParams,
			"no preview result for the table, please make sure the CSV file is correct",
		)
	}
	return &console.ImportsPreviewLocalResp{
		Table: vo.Table2Console(dbPreviewResult.GetTables()[0]),
	}, nil
}

func (s *importService) GetTableColumns(
	ctx context.Context,
	req *console.ImportsGetTableColumnsReq,
) (_ *console.ImportsGetTableColumnsResp, err error) {
	ctx, span := Tracer.Start(ctx, "console.import.GetTableColumns")
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareImportErrorMessage(ctx, err)
	}()

	logger := log.WithContextL(ctx).With(zap.String("cluster_id", req.ClusterId))

	if req.Database == "" {
		return nil, bizerr.New(bizerr.ErrCodeParams, "database name cannot be empty")
	}
	if req.Table == "" {
		return nil, bizerr.New(bizerr.ErrCodeParams, "table name cannot be empty")
	}

	cluster, err := s.validateCluster(ctx, req.ClusterId, logger)
	if err != nil {
		return nil, err
	}

	span.AddEvent("call GetTableColumns")
	tableColumns, bizErr := s.importHelper.GetTableColumns(ctx, cluster, req.Database, req.Table)
	if bizErr != nil {
		if bizErr.RawErrCode() != errorcode.ErrCodeTableNotExist {
			logger.Error("GetTableColumns failed", zap.Error(bizErr))
			return nil, bizErr
		}
		logger.Warn("table not exist", zap.String("database", req.Database), zap.String("table", req.Table))
		return nil, bizerr.New(bizerr.ErrCodeRecordNotFound, "table not exist")
	}

	return &console.ImportsGetTableColumnsResp{
		Table: vo.Table2Console(tableColumns),
	}, nil
}

func (s *importService) GetDataIngestionProfile(
	ctx context.Context,
	req *console.GetProfileReq,
) (_ *console.DataIngestionProfile, err error) {
	ctx, span := Tracer.Start(ctx, "console.import.GetDataIngestionProfile")
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareImportErrorMessage(ctx, err)
	}()

	cluster, bizErr := s.validateCluster(ctx, req.ClusterId, log.WithContextL(ctx))
	if bizErr != nil {
		return nil, bizErr
	}

	if cluster.CloudProvider != config.ProviderAWS {
		return nil, bizerr.New(bizerr.ErrCodeParams, "only AWS provider is supported")
	}
	p, bizErr := s.importSvc.GetDataIngestionProfile(ctx, cluster)
	if bizErr != nil {
		return nil, bizErr
	}

	return &console.DataIngestionProfile{
		Type: console.ImportDataIngestionProfileType_S3,
		Profile: &console.DataIngestionProfile_S3Profile{
			S3Profile: &console.S3Profile{
				AwsExternalId: p.ExternalId,
				AwsAccountId:  p.AccountId,
			},
		},
		BaseResp: nil,
	}, nil
}

func (s *importService) ListSupportedDataTypes(
	ctx context.Context,
	_ *emptypb.Empty,
) (_ *console.ImportsSupportedDataTypesResp, err error) {
	ctx, span := Tracer.Start(ctx, "console.import.ListDataTypes")
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareImportErrorMessage(ctx, err)
	}()

	return &console.ImportsSupportedDataTypesResp{
		DataTypes: entity.GetDefaultDataType(),
	}, nil
}

func (s *importService) ListTargetTables(
	ctx context.Context,
	req *console.ListTargetTablesReq,
) (_ *console.ListTargetTablesResp, err error) {
	ctx, span := Tracer.Start(ctx, "console.import.ListTargetTables")
	defer func() {
		tracing.EndSpan(span, err)
		err = errorcode.PrepareImportErrorMessage(ctx, err)
	}()
	logger := log.WithContextL(ctx).With(zap.String("cluster_id", req.ClusterId))

	proxy, bizErr := s.getClusterProxy(ctx, span, logger, req.ClusterId)
	if bizErr != nil {
		return nil, bizErr
	}

	span.AddEvent("list databases sql")
	res, err := proxy.QuerySQL(ctx, "SHOW DATABASES")
	if err != nil {
		logger.Error("failed to query sql", zap.Error(err))
		return nil, err
	}

	type DatabaseResult struct {
		Databases [][]byte `json:"Database"`
	}
	result := DatabaseResult{}
	err = json.Unmarshal(res, &result)
	if err != nil {
		logger.Error("failed to unmarshal json", zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeJsonMarshal, "failed to unmarshal json")
	}
	var databases []string
	for _, database := range result.Databases {
		if !slices.Contains(constant.ExcludeDatabases, string(database)) {
			databases = append(databases, string(database))
		}
	}

	ds, err := s.getAllDatabases(ctx, databases, logger, proxy)
	if err != nil {
		return nil, err
	}

	return &console.ListTargetTablesResp{
		Databases: ds,
	}, nil
}

func (s *importService) getAllDatabases(
	ctx context.Context,
	databases []string,
	logger *zap.Logger,
	proxy facade.ClusterProxyFacade,
) ([]*console.Database, error) {
	ds := make([]*console.Database, 0, len(databases))
	var wg sync.WaitGroup
	var mu sync.Mutex
	wg.Add(len(databases))
	var errorInRoutine error
	var errorOnce sync.Once
	for _, databaseName := range databases {
		go func(ctx context.Context, logger *zap.Logger, proxy facade.ClusterProxyFacade, dbName string) {
			defer wg.Done()
			result, e := s.getAllTables(ctx, logger, proxy, dbName)
			if e != nil {
				logger.Error("failed to get all tables", zap.Error(e))
				errorOnce.Do(func() {
					errorInRoutine = e
				})
				return
			}

			mu.Lock()
			ds = append(ds, result)
			mu.Unlock()
		}(ctx, logger, proxy, databaseName)
	}

	wg.Wait()
	if errorInRoutine != nil {
		return nil, errorInRoutine
	}
	return ds, nil
}

func (s *importService) getAllTables(
	ctx context.Context,
	logger *zap.Logger,
	proxy facade.ClusterProxyFacade,
	databaseName string,
) (*console.Database, error) {
	d := &console.Database{
		Database: databaseName,
	}
	data, bizErr := proxy.QuerySQL(
		ctx,
		fmt.Sprintf("show full tables in `%s` where Table_Type != 'VIEW';", utils.EscapeBackticks(databaseName)),
	)
	if bizErr != nil {
		logger.Error("failed to query sql", zap.Error(bizErr), zap.String("database", databaseName))
		return nil, bizErr
	}
	temp := map[string][][]byte{}
	err := json.Unmarshal(data, &temp)
	if err != nil {
		logger.Error("failed to unmarshal json", zap.Error(err))
		return nil, bizerr.New(bizerr.ErrCodeJsonMarshal, "failed to unmarshal json")
	}
	tables, ok := temp[fmt.Sprintf("Tables_in_%s", databaseName)]
	ts := make([]string, 0, len(tables))
	if ok {
		for j := range tables {
			ts = append(ts, string(tables[j]))
		}
	} else {
		logger.Warn(fmt.Sprintf("no \"Tables_in_%s\" in sql result", databaseName), zap.String("database", databaseName))
	}
	d.Tables = ts
	return d, nil
}

func convertToCloudStorageType(t console.ImportSourceType_Enum) regionalserverlessv1.CloudStorageType {
	switch t {
	case console.ImportSourceType_S3:
		return regionalserverlessv1.CloudStorageType_S3
	case console.ImportSourceType_GCS:
		return regionalserverlessv1.CloudStorageType_GCS
	case console.ImportSourceType_OSS:
		return regionalserverlessv1.CloudStorageType_OSS
	case console.ImportSourceType_AZURE_BLOB:
		return regionalserverlessv1.CloudStorageType_AZURE_BLOB
	case console.ImportSourceType_S3_COMPATIBLE:
		return regionalserverlessv1.CloudStorageType_S3_COMPATIBLE
	default:
		return regionalserverlessv1.CloudStorageType_CLOUD_STORAGE_TYPE_UNSPECIFIED
	}
}

func buildCloudStorageFromImportSource(
	source *console.ImportSource,
	getExternalId func() (string, bizerr.BizError),
) (*regionalserverlessv1.CloudStorage, bizerr.BizError) {
	cs := &regionalserverlessv1.CloudStorage{
		CloudStorageType: convertToCloudStorageType(source.GetType()),
	}
	switch source.GetType() {
	case console.ImportSourceType_S3:
		s3 := source.GetS3()
		cs.Uri = s3.GetUri()
		switch s3.GetAuthType() {
		case console.ImportS3AuthType_ROLE_ARN:
			externalId, bizErr := getExternalId()
			if bizErr != nil {
				return nil, bizErr
			}
			cs.CloudStorage = &regionalserverlessv1.CloudStorage_S3{
				S3: &regionalserverlessv1.S3CloudStorage{
					AuthType: regionalserverlessv1.S3CloudStorage_ROLE_ARN,
					Auth: &regionalserverlessv1.S3CloudStorage_RoleArn_{
						RoleArn: &regionalserverlessv1.S3CloudStorage_RoleArn{
							RoleArn:    s3.GetRoleArn(),
							ExternalId: externalId,
						},
					},
				},
			}
		case console.ImportS3AuthType_ACCESS_KEY:
			cs.CloudStorage = &regionalserverlessv1.CloudStorage_S3{
				S3: &regionalserverlessv1.S3CloudStorage{
					AuthType: regionalserverlessv1.S3CloudStorage_ACCESS_KEY,
					Auth: &regionalserverlessv1.S3CloudStorage_AccessKey_{
						AccessKey: &regionalserverlessv1.S3CloudStorage_AccessKey{
							AccessKeyId:     s3.GetAccessKey().GetId(),
							SecretAccessKey: s3.GetAccessKey().GetSecret(),
						},
					},
				},
			}
		}
	case console.ImportSourceType_AZURE_BLOB:
		azure := source.GetAzureBlob()
		_, _, bizErr := utils.ParseAzureURI(azure.GetUri())
		if bizErr != nil {
			return nil, bizErr
		}
		cs.Uri = azure.GetUri()
		switch azure.GetAuthType() {
		case console.ImportAzureBlobAuthType_SAS_TOKEN:
			cs.CloudStorage = &regionalserverlessv1.CloudStorage_AzureBlob{
				AzureBlob: &regionalserverlessv1.AzureBlobCloudStorage{
					AuthType: regionalserverlessv1.AzureBlobCloudStorage_SAS_TOKEN,
					Auth: &regionalserverlessv1.AzureBlobCloudStorage_SasToken{
						SasToken: azure.GetSasToken(),
					},
				},
			}
		}
	case console.ImportSourceType_GCS:
		gcs := source.GetGcs()
		cs.Uri = gcs.GetUri()
		switch gcs.GetAuthType() {
		case console.ImportGcsAuthType_SERVICE_ACCOUNT_KEY:
			cs.CloudStorage = &regionalserverlessv1.CloudStorage_Gcs{
				Gcs: &regionalserverlessv1.GCSCloudStorage{
					AuthType: regionalserverlessv1.GCSCloudStorage_SERVICE_ACCOUNT_KEY,
					Auth: &regionalserverlessv1.GCSCloudStorage_ServiceAccountKey{
						ServiceAccountKey: gcs.GetServiceAccountKey(),
					},
				},
			}
		}
	case console.ImportSourceType_OSS:
		oss := source.GetOss()
		cs.Uri = oss.GetUri()
		switch oss.GetAuthType() {
		case console.ImportOSSAuthType_ACCESS_KEY:
			cs.CloudStorage = &regionalserverlessv1.CloudStorage_Oss{
				Oss: &regionalserverlessv1.OSSCloudStorage{
					AuthType: regionalserverlessv1.OSSCloudStorage_ACCESS_KEY,
					Auth: &regionalserverlessv1.OSSCloudStorage_AccessKey_{
						AccessKey: &regionalserverlessv1.OSSCloudStorage_AccessKey{
							AccessKeyId:     oss.GetAccessKey().GetId(),
							AccessKeySecret: oss.GetAccessKey().GetSecret(),
						},
					},
				},
			}
		case console.ImportOSSAuthType_ROLE_ARN:
			externalId, bizErr := getExternalId()
			if bizErr != nil {
				return nil, bizErr
			}
			cs.CloudStorage = &regionalserverlessv1.CloudStorage_Oss{
				Oss: &regionalserverlessv1.OSSCloudStorage{
					AuthType: regionalserverlessv1.OSSCloudStorage_ROLE_ARN,
					Auth: &regionalserverlessv1.OSSCloudStorage_RoleArn_{
						RoleArn: &regionalserverlessv1.OSSCloudStorage_RoleArn{
							RoleArn:    oss.GetRoleArn(),
							ExternalId: externalId,
						},
					},
				},
			}
		}
	case console.ImportSourceType_S3_COMPATIBLE:
		s3Comp := source.GetS3Compatible()
		cs.Uri = s3Comp.GetUri()
		switch s3Comp.GetAuthType() {
		case console.ImportS3CompatibleAuthType_ACCESS_KEY:
			cs.CloudStorage = &regionalserverlessv1.CloudStorage_S3Compatible{
				S3Compatible: &regionalserverlessv1.S3CompatibleCloudStorage{
					AuthType: regionalserverlessv1.S3CompatibleCloudStorage_ACCESS_KEY,
					Auth: &regionalserverlessv1.S3CompatibleCloudStorage_AccessKey_{
						AccessKey: &regionalserverlessv1.S3CompatibleCloudStorage_AccessKey{
							AccessKeyId:     s3Comp.GetAccessKey().GetId(),
							SecretAccessKey: s3Comp.GetAccessKey().GetSecret(),
						},
					},
					Endpoint: s3Comp.GetEndpoint(),
				},
			}
		}
	}
	return cs, nil
}