package importadmin

import (
	"context"
	"fmt"
	"strconv"

	"github.com/pingcap/errcode"
	importadminbp "github.com/tidbcloud/dataflow-service/idl/pbgen/proto/adminapi/v1beta1"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	"github.com/tidbcloud/dataflow-service/internal/common/pagetoken"
	"github.com/tidbcloud/dataflow-service/internal/db"
	"github.com/tidbcloud/dataflow-service/internal/db/dbutil"
	"github.com/tidbcloud/dataflow-service/internal/db/entity"
	"github.com/tidbcloud/dataflow-service/internal/grpc/grpcli"
	"github.com/tidbcloud/dataflow-service/pkg/redact"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
	"google.golang.org/protobuf/types/known/emptypb"
	"google.golang.org/protobuf/types/known/timestamppb"
)

var _ importadminbp.ImportServiceServer = &Service{}

type Service struct {
	importadminbp.UnimplementedImportServiceServer

	dataflowImportService dataflow.ImportServiceServer
}

func NewService(dataflowImportService dataflow.ImportServiceServer) *Service {
	return &Service{
		dataflowImportService: dataflowImportService,
	}
}

func (s *Service) parseClusterID(clusterID string) (uint64, error) {
	clusterIDUint64, err := strconv.ParseUint(clusterID, 10, 64)
	if err != nil {
		return 0, err
	}
	return clusterIDUint64, nil
}

func (s *Service) parseImportID(importID string) (uint64, error) {
	importIDUint64, err := strconv.ParseUint(importID, 10, 64)
	if err != nil {
		return 0, err
	}
	return importIDUint64, nil
}

func (s *Service) ListImports(ctx context.Context, req *importadminbp.ListImportsRequest) (*importadminbp.ListImportsResponse, error) {
	clusterID, err := s.parseClusterID(req.ClusterId)
	if err != nil {
		return nil, err
	}

	cluster, err := grpcli.GetClusterByIDAndExcludeStatuses(ctx, clusterID, nil)
	if err != nil {
		return nil, err
	}

	pageSize := req.PageSize
	token := pagetoken.PageToken[uint64]{}
	if req.PageToken != "" {
		if err := token.DecodeFromString(req.PageToken); err != nil {
			return nil, err
		}
	}
	page := int32(token.Prev) + 1 // page is 1-based

	imports, err := s.dataflowImportService.List(ctx, &dataflow.ListImportReq{
		OrgId:     cluster.TenantId,
		ProjectId: cluster.ProjectId,
		ClusterId: clusterID,
		Page:      page,
		PageSize:  pageSize,
	})
	if err != nil {
		return nil, err
	}

	importAdminImports := make([]*importadminbp.Import, 0, len(imports.Imports))
	for _, imp := range imports.Imports {
		impdb, err := dbutil.GetImportByID(db.GetDBWithContext(ctx), entity.ID(imp.ClusterId), entity.ID(imp.Id))
		if err != nil {
			return nil, err
		}
		importAdminImport, err := s.convertImportToImportAdminImport(imp, impdb)
		if err != nil {
			return nil, err
		}

		importAdminImports = append(importAdminImports, importAdminImport)
	}

	token.Prev = uint64(page)
	nextPageToken, err := token.EncodeToString()
	if err != nil {
		return nil, err
	}

	return &importadminbp.ListImportsResponse{
		Imports:       importAdminImports,
		TotalSize:     int32(imports.Total),
		NextPageToken: nextPageToken,
	}, nil
}

func (s *Service) convertImportToImportAdminImport(imp *dataflow.ImportGetResp, impdb *entity.Import) (*importadminbp.Import, error) {
	var state importadminbp.ImportState_Enum
	switch imp.Status {
	case dataflow.ImportGetResp_PREPARING:
		state = importadminbp.ImportState_PREPARING
	case dataflow.ImportGetResp_IMPORTING:
		state = importadminbp.ImportState_IMPORTING
	case dataflow.ImportGetResp_COMPLETED:
		state = importadminbp.ImportState_COMPLETED
	case dataflow.ImportGetResp_FAILED:
		state = importadminbp.ImportState_FAILED
	case dataflow.ImportGetResp_CANCELING:
		state = importadminbp.ImportState_CANCELING
	case dataflow.ImportGetResp_CANCELED:
		state = importadminbp.ImportState_CANCELED
	default:
		state = importadminbp.ImportState_UNSPECIFIED
	}

	var customCSVFormat *importadminbp.CSVFormat
	var fileType importadminbp.ImportFileType_Enum
	switch imp.DataFormat {
	case uint32(dataflow.ImportDataFormat_SQL):
		fileType = importadminbp.ImportFileType_SQL
	case uint32(dataflow.ImportDataFormat_AURORA_SNAPSHOT):
		fileType = importadminbp.ImportFileType_AURORA_SNAPSHOT
	case uint32(dataflow.ImportDataFormat_CSV):
		fileType = importadminbp.ImportFileType_CSV

		if imp.ImportCreateReq != nil {
			customCSVFormat = &importadminbp.CSVFormat{
				Separator:         imp.ImportCreateReq.CsvFormat.Separator,
				Delimiter:         &imp.ImportCreateReq.CsvFormat.Delimiter,
				Header:            &imp.ImportCreateReq.CsvFormat.Header,
				NotNull:           imp.ImportCreateReq.CsvFormat.NotNull,
				NullValue:         imp.ImportCreateReq.CsvFormat.Null,
				BackslashEscape:   &imp.ImportCreateReq.CsvFormat.BackslashEscape,
				TrimLastSeparator: imp.ImportCreateReq.CsvFormat.TrimLastSeparator,
			}
		}
	case uint32(dataflow.ImportDataFormat_PARQUET):
		fileType = importadminbp.ImportFileType_PARQUET
	default:
		fileType = importadminbp.ImportFileType_UNSPECIFIED
	}

	importOptions := &importadminbp.ImportOptions{
		FileType: fileType,
		Format:   &importadminbp.ImportOptions_CsvFormat{CsvFormat: customCSVFormat},
	}
	if imp.ImportCreateReq != nil {
		importOptions.SkipChecksum = imp.ImportCreateReq.GetSkipChecksum()
	}

	var importSource *importadminbp.ImportSource

	var sourceType importadminbp.ImportSourceType_Enum
	switch imp.DataSourceType {
	case uint32(dataflow.ImportDataSource_S3_SOURCE):
		sourceType = importadminbp.ImportSourceType_S3

		s3Source := &importadminbp.S3Source{
			Uri: imp.SourceUrl,
		}

		if imp.ImportCreateReq != nil {
			if imp.ImportCreateReq.AwsRoleArn != "" {
				s3Source.AuthType = importadminbp.ImportS3AuthType_ROLE_ARN
				s3Source.Auth = &importadminbp.S3Source_RoleArn{
					RoleArn: imp.ImportCreateReq.AwsRoleArn,
				}
			} else if imp.ImportCreateReq.AwsAccessKeyId != "" || imp.ImportCreateReq.AwsSecretAccessKey != "" {
				s3Source.AuthType = importadminbp.ImportS3AuthType_ACCESS_KEY
				s3Source.Auth = &importadminbp.S3Source_AccessKey_{
					AccessKey: &importadminbp.S3Source_AccessKey{
						Id:     imp.ImportCreateReq.AwsAccessKeyId,
						Secret: imp.ImportCreateReq.AwsSecretAccessKey,
					},
				}
			} else {
				s3Source.AuthType = importadminbp.ImportS3AuthType_UNSPECIFIED
			}
		} else {
			s3Source.AuthType = importadminbp.ImportS3AuthType_UNSPECIFIED
		}

		importSource = &importadminbp.ImportSource{
			Type: sourceType,
			Source: &importadminbp.ImportSource_S3{
				S3: s3Source,
			},
		}
	case uint32(dataflow.ImportDataSource_GCS_SOURCE):
		sourceType = importadminbp.ImportSourceType_GCS

		gcsSource := &importadminbp.GCSSource{
			Uri:      imp.SourceUrl,
			AuthType: importadminbp.ImportGcsAuthType_UNSPECIFIED,
		}

		importSource = &importadminbp.ImportSource{
			Type: sourceType,
			Source: &importadminbp.ImportSource_Gcs{
				Gcs: gcsSource,
			},
		}
	case uint32(dataflow.ImportDataSource_AZBLOB_SOURCE):
		sourceType = importadminbp.ImportSourceType_AZURE_BLOB

		azureBlobSource := &importadminbp.AzureBlobSource{
			Uri: imp.SourceUrl,
		}

		if imp.ImportCreateReq != nil && imp.ImportCreateReq.AzblobSasToken != "" {
			azureBlobSource.AuthType = importadminbp.ImportAzureBlobAuthType_SAS_TOKEN
			azureBlobSource.Auth = &importadminbp.AzureBlobSource_SasToken{
				SasToken: imp.ImportCreateReq.AzblobSasToken,
			}
		} else {
			azureBlobSource.AuthType = importadminbp.ImportAzureBlobAuthType_UNSPECIFIED
		}

		importSource = &importadminbp.ImportSource{
			Type: sourceType,
			Source: &importadminbp.ImportSource_AzureBlob{
				AzureBlob: azureBlobSource,
			},
		}
	default:
		sourceType = importadminbp.ImportSourceType_UNSPECIFIED
	}

	targetTableInfos := make([]*importadminbp.ImportTargetTableInfo, 0, len(imp.ImportCreateReq.TargetTableInfos))
	for _, targetTableInfo := range imp.ImportCreateReq.TargetTableInfos {
		targetTableInfos = append(targetTableInfos, &importadminbp.ImportTargetTableInfo{
			TargetTable: targetTableInfo.TargetTable,
			CustomFile:  targetTableInfo.CustomFile,
			MatchedUri:  targetTableInfo.MatchedUri,
		})
	}

	creationDetails := &importadminbp.CreationDetails{
		ImportOptions:    importOptions,
		Source:           importSource,
		TargetTableInfos: targetTableInfos,
	}

	var completeTime *timestamppb.Timestamp
	if impdb.FinishedAt != nil {
		completeTime = timestamppb.New(*impdb.FinishedAt)
	}

	return &importadminbp.Import{
		Name:            fmt.Sprintf("clusters/%d/imports/%d", imp.ClusterId, imp.Id),
		ClusterId:       strconv.FormatUint(imp.ClusterId, 10),
		TotalSize:       int64(imp.TotalSize),
		CreateTime:      timestamppb.New(impdb.CreatedAt),
		CompleteTime:    completeTime,
		State:           state,
		CompletePercent: int32(imp.CompletedPercent),
		Message:         imp.Message,
		Creator:         imp.CreatedBy,
		CreationDetails: creationDetails,
		ImportId:        strconv.FormatUint(imp.Id, 10),
	}, nil
}

func (s *Service) GetImport(ctx context.Context, req *importadminbp.GetImportRequest) (*importadminbp.Import, error) {
	clusterID, err := s.parseClusterID(req.ClusterId)
	if err != nil {
		return nil, err
	}

	importID, err := s.parseImportID(req.ImportId)
	if err != nil {
		return nil, err
	}

	impdb, err := dbutil.GetImportByID(db.GetDBWithContext(ctx), entity.ID(clusterID), entity.ID(importID))
	if err != nil {
		return nil, err
	}

	imp, err := s.dataflowImportService.Get(ctx, &dataflow.ClusterImportId{
		ClusterId: clusterID,
		Id:        importID,
		OrgId:     impdb.TenantID.Uint64(),
		ProjectId: impdb.ProjectID.Uint64(),
	})
	if err != nil {
		return nil, err
	}

	return s.convertImportToImportAdminImport(imp, impdb)
}

func (s *Service) CancelImport(ctx context.Context, req *importadminbp.CancelImportRequest) (*emptypb.Empty, error) {
	clusterID, err := s.parseClusterID(req.ClusterId)
	if err != nil {
		return nil, err
	}

	importID, err := s.parseImportID(req.ImportId)
	if err != nil {
		return nil, err
	}

	imp, err := dbutil.GetImportByID(db.GetDBWithContext(ctx), entity.ID(clusterID), entity.ID(importID))
	if err != nil {
		tilog.Error(ctx, "CancelImport", zap.Any("err", err), zap.Any("clusterID", clusterID), zap.Any("importID", importID))
		return nil, err
	}

	return s.dataflowImportService.Cancel(ctx, &dataflow.ClusterImportId{
		ClusterId: clusterID,
		Id:        importID,
		OrgId:     imp.TenantID.Uint64(),
		ProjectId: imp.ProjectID.Uint64(),
	})
}

func (s *Service) convertCreateImportRequest(ctx context.Context, req *importadminbp.CreateImportRequest) (*dataflow.ImportCreateReq, error) {
	imp := req.Import

	clusterID, err := s.parseClusterID(req.ClusterId)
	if err != nil {
		return nil, err
	}

	cluster, err := grpcli.GetClusterByIDAndExcludeStatuses(ctx, clusterID, nil)
	if err != nil {
		return nil, err
	}

	region := cluster.RegionName

	dataSourceType, sourceURL, awsRoleArn, awsAccessKeyId, awsSecretAccessKey, azblobSasToken, err := s.convertSourceURL(ctx, imp)
	if err != nil {
		return nil, err
	}

	dataFormat, err := s.convertDataFormat(imp)
	if err != nil {
		return nil, err
	}

	targetTableInfos, err := s.convertTargetTableInfos(imp)
	if err != nil {
		return nil, err
	}

	csvFormat, err := s.convertCsvFormat(ctx, imp)
	if err != nil {
		return nil, err
	}

	return &dataflow.ImportCreateReq{
		ClusterId:          clusterID,
		SourceUrl:          sourceURL,
		Region:             region,
		DataFormat:         dataFormat,
		DataSourceType:     dataSourceType,
		OrgId:              cluster.TenantId,
		ProjectId:          cluster.ProjectId,
		CsvFormat:          csvFormat,
		AwsRoleArn:         awsRoleArn,
		AwsAccessKeyId:     awsAccessKeyId,
		AwsSecretAccessKey: awsSecretAccessKey,
		AzblobSasToken:     azblobSasToken,
		Name:               imp.Name,
		DupHandlingForSql:  dataflow.DupHandlingForSql(imp.CreationDetails.ImportOptions.DuplicationHandling),
		SkipChecksum:       imp.CreationDetails.ImportOptions.GetSkipChecksum(),
		//NeedDropTables:     imp.CreationDetails.ImportOptions.NeedDropTables, // openapi disable this option
		TargetTableInfos: targetTableInfos,
	}, nil
}

func (s *Service) CreateImport(ctx context.Context, req *importadminbp.CreateImportRequest) (*importadminbp.Import, error) {
	if req.Import != nil && req.Import.CreationDetails != nil && req.Import.CreationDetails.ImportOptions != nil && req.Import.CreationDetails.ImportOptions.ValidateOnly {
		return s.validateImport(ctx, req)
	}

	dataflowCreateImportReq, err := s.convertCreateImportRequest(ctx, req)
	if err != nil {
		return nil, err
	}

	importID, err := s.dataflowImportService.Create(ctx, dataflowCreateImportReq)
	if err != nil {
		return nil, err
	}

	impdb, err := dbutil.GetImportByID(db.GetDBWithContext(ctx), entity.ID(importID.ClusterId), entity.ID(importID.Id))
	if err != nil {
		return nil, err
	}

	imp, err := s.dataflowImportService.Get(ctx, importID)
	if err != nil {
		return nil, err
	}

	return s.convertImportToImportAdminImport(imp, impdb)
}

func (s *Service) convertSourceURL(ctx context.Context, imp *importadminbp.Import) (uint32, string, string, string, string, string, error) {
	var (
		dataSourceType     uint32
		sourceURL          string
		awsRoleArn         string
		awsAccessKeyId     string
		awsSecretAccessKey string
		azblobSasToken     string
	)

	if imp == nil || imp.CreationDetails == nil || imp.CreationDetails.Source == nil {
		tilog.Error(ctx, "invalid import", zap.Any("import", imp))
		return 0, "", "", "", "", "", errcode.NewInvalidInputErr(fmt.Errorf("invalid import: %v", imp))
	}

	if sourcePayload, redactErr := redact.RedactKeysToJSON(redact.PasswordFields, imp.CreationDetails.Source); redactErr == nil {
		tilog.Info(ctx, "convertSourceURL", zap.ByteString("source", sourcePayload))
	} else {
		tilog.Warn(ctx, "convertSourceURL: failed to redact source payload for logging", zap.Error(redactErr))
	}

	switch imp.CreationDetails.Source.Type {
	case importadminbp.ImportSourceType_S3:
		dataSourceType = uint32(dataflow.ImportDataSource_S3_SOURCE)
		s3 := imp.CreationDetails.Source.GetS3()
		if s3 == nil {
			return 0, "", "", "", "", "", errcode.NewInvalidInputErr(fmt.Errorf("invalid s3 source: %v", imp.CreationDetails.Source))
		}

		sourceURL = s3.Uri
		awsRoleArn = s3.GetRoleArn()
		awsAccessKeyId = s3.GetAccessKey().GetId()
		awsSecretAccessKey = s3.GetAccessKey().GetSecret()
	case importadminbp.ImportSourceType_GCS:
		dataSourceType = uint32(dataflow.ImportDataSource_GCS_SOURCE)
		gcs := imp.CreationDetails.Source.GetGcs()
		if gcs == nil {
			return 0, "", "", "", "", "", errcode.NewInvalidInputErr(fmt.Errorf("invalid gcs source: %v", imp.CreationDetails.Source))
		}

		sourceURL = gcs.Uri
	case importadminbp.ImportSourceType_AZURE_BLOB:
		dataSourceType = uint32(dataflow.ImportDataSource_AZBLOB_SOURCE)
		azblob := imp.CreationDetails.Source.GetAzureBlob()
		if azblob == nil {
			return 0, "", "", "", "", "", errcode.NewInvalidInputErr(fmt.Errorf("invalid azure blob source: %v", imp.CreationDetails.Source))
		}

		sourceURL = azblob.GetUri()
		azblobSasToken = azblob.GetSasToken()
	default:
		return 0, "", "", "", "", "", errcode.NewInvalidInputErr(fmt.Errorf("invalid source type: %v", imp.CreationDetails.Source.Type))
	}

	return dataSourceType, sourceURL, awsRoleArn, awsAccessKeyId, awsSecretAccessKey, azblobSasToken, nil
}

func (s *Service) convertDataFormat(imp *importadminbp.Import) (uint32, error) {
	var dataFormat uint32 = uint32(dataflow.ImportDataFormat_UNKNOWN_FORMAT)
	switch imp.CreationDetails.ImportOptions.FileType {
	case importadminbp.ImportFileType_SQL:
		dataFormat = uint32(dataflow.ImportDataFormat_SQL)
	case importadminbp.ImportFileType_AURORA_SNAPSHOT:
		dataFormat = uint32(dataflow.ImportDataFormat_AURORA_SNAPSHOT)
	case importadminbp.ImportFileType_CSV:
		dataFormat = uint32(dataflow.ImportDataFormat_CSV)
	case importadminbp.ImportFileType_PARQUET:
		dataFormat = uint32(dataflow.ImportDataFormat_PARQUET)
	default:
		return 0, fmt.Errorf("invalid file type: %v", imp.CreationDetails.ImportOptions.FileType)
	}

	return dataFormat, nil
}

func (s *Service) convertCsvFormat(ctx context.Context, imp *importadminbp.Import) (*dataflow.CustomCSVFormat, error) {
	var (
		defaultSeparator         = ","
		defaultDelimiter         = "\""
		defaultHeader            = true
		defaultNotNull           = false
		defaultNullValue         = "\\N"
		defaultBackslashEscape   = true
		defaultTrimLastSeparator = false
	)
	defaultCSVFormat := &dataflow.CustomCSVFormat{
		Separator:         defaultSeparator,
		Delimiter:         defaultDelimiter,
		Header:            defaultHeader,
		NotNull:           &defaultNotNull,
		Null:              &defaultNullValue,
		BackslashEscape:   defaultBackslashEscape,
		TrimLastSeparator: &defaultTrimLastSeparator,
	}

	if imp.CreationDetails.ImportOptions.FileType == importadminbp.ImportFileType_CSV {
		reqCsvFormat := imp.CreationDetails.ImportOptions.GetCsvFormat()
		if reqCsvFormat == nil {
			tilog.Info(ctx, "no csv format, use default", zap.Any("imp", imp), zap.Any("defaultCSVFormat", defaultCSVFormat))
			return defaultCSVFormat, nil
		}

		var (
			separator         = defaultSeparator
			delimiter         = defaultDelimiter
			header            = defaultHeader
			notNull           = defaultNotNull
			nullValue         = defaultNullValue
			backslashEscape   = defaultBackslashEscape
			trimLastSeparator = defaultTrimLastSeparator
		)

		if reqCsvFormat.Separator != "" {
			separator = reqCsvFormat.Separator
		}

		if reqCsvFormat.Delimiter != nil {
			delimiter = *reqCsvFormat.Delimiter
		}

		if reqCsvFormat.Header != nil {
			header = *reqCsvFormat.Header
		}

		if reqCsvFormat.NotNull != nil {
			notNull = *reqCsvFormat.NotNull
		}

		if reqCsvFormat.NullValue != nil {
			nullValue = *reqCsvFormat.NullValue
		}

		if reqCsvFormat.BackslashEscape != nil {
			backslashEscape = *reqCsvFormat.BackslashEscape
		}

		if reqCsvFormat.TrimLastSeparator != nil {
			trimLastSeparator = *reqCsvFormat.TrimLastSeparator
		}

		csvFormat := &dataflow.CustomCSVFormat{
			Separator:         separator,
			Delimiter:         delimiter,
			Header:            header,
			NotNull:           &notNull,
			Null:              &nullValue,
			BackslashEscape:   backslashEscape,
			TrimLastSeparator: &trimLastSeparator,
		}

		tilog.Info(ctx, "convertCsvFormat", zap.Any("csvFormat", csvFormat))

		return csvFormat, nil
	}

	return nil, nil
}

func (s *Service) convertCheckItems(imp *importadminbp.Import) ([]dataflow.ImportValidationItem, error) {
	var checkItems []dataflow.ImportValidationItem
	for _, checkItem := range imp.CreationDetails.CheckItems {
		checkItems = append(checkItems, dataflow.ImportValidationItem(checkItem))
	}
	return checkItems, nil
}

func (s *Service) convertTargetTableInfos(imp *importadminbp.Import) ([]*dataflow.ImportTargetTableInfo, error) {
	var targetTableInfos []*dataflow.ImportTargetTableInfo
	for _, targetTableInfo := range imp.CreationDetails.TargetTableInfos {
		targetTableInfos = append(targetTableInfos, &dataflow.ImportTargetTableInfo{
			TargetTable: targetTableInfo.TargetTable,
			CustomFile:  targetTableInfo.CustomFile,
			MatchedUri:  targetTableInfo.MatchedUri,
		})
	}
	return targetTableInfos, nil
}

func (s *Service) convertValidateImportRequest(ctx context.Context, req *importadminbp.CreateImportRequest) (*dataflow.ImportValidateReq, error) {
	clusterID, err := s.parseClusterID(req.ClusterId)
	if err != nil {
		return nil, err
	}

	imp := req.Import

	dataSourceType, sourceURL, awsRoleArn, awsAccessKeyId, awsSecretAccessKey, azblobSasToken, err := s.convertSourceURL(ctx, imp)
	tilog.Info(ctx, "convertSourceURL", zap.Uint32("dataSourceType", dataSourceType))
	if err != nil {
		return nil, err
	}

	dataFormat, err := s.convertDataFormat(imp)
	if err != nil {
		return nil, err
	}

	csvFormat, err := s.convertCsvFormat(ctx, imp)
	if err != nil {
		return nil, err
	}

	cluster, err := grpcli.GetClusterByIDAndExcludeStatuses(ctx, clusterID, nil)
	if err != nil {
		return nil, err
	}

	region := cluster.RegionName

	targetTableInfos, err := s.convertTargetTableInfos(imp)
	if err != nil {
		return nil, err
	}

	checkItems, err := s.convertCheckItems(imp)
	if err != nil {
		return nil, err
	}

	return &dataflow.ImportValidateReq{
		ClusterId:      clusterID,
		SourceUrl:      sourceURL,
		Region:         region,
		DataSourceType: dataSourceType,
		/* deprecated fields
		// TargetUsername: imp.CreationDetails.Target.Username,
		// TargetPassword: imp.CreationDetails.Target.Password,
		*/
		OrgId:      cluster.TenantId,
		ProjectId:  cluster.ProjectId,
		AwsRoleArn: awsRoleArn,
		/* deprecated fields
		// DbTblFilter:    dbTblFilter,
		// ObjNamePattern: objNamePattern,
		// TargetTblName:  targetTblName,
		*/
		DataFormat:         dataFormat,
		CsvFormat:          csvFormat,
		CheckItems:         checkItems,
		AwsAccessKeyId:     awsAccessKeyId,
		AwsSecretAccessKey: awsSecretAccessKey,
		TargetTableInfos:   targetTableInfos,
		AzblobSasToken:     azblobSasToken,
	}, nil
}

func (s *Service) validateImport(ctx context.Context, req *importadminbp.CreateImportRequest) (*importadminbp.Import, error) {
	if importPayload, redactErr := redact.RedactKeysToJSON(redact.PasswordFields, req.Import); redactErr == nil {
		tilog.Info(ctx, "ValidateImport", zap.ByteString("import", importPayload))
	}
	dataflowReq, err := s.convertValidateImportRequest(ctx, req)
	if err != nil {
		return nil, err
	}

	_, err = s.dataflowImportService.Validate(ctx, dataflowReq)
	if err != nil {
		return nil, err
	}

	return &importadminbp.Import{
		Name:    req.Import.Name,
		Message: "validate success",
	}, nil
}