package tidbtools

import (
	"fmt"
	"net/url"
	"strconv"
	"strings"

	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	nextgenglobal "github.com/tidbcloud/tidb-management-service/api/spec/global/nextgen/v1"
)

// DedicatedBackupSetting is the backup configuration reported for Dedicated clusters.
type DedicatedBackupSetting struct {
	Schedule          string                `json:"schedule"`
	RetentionDays     int32                 `json:"retention_days"`
	EncryptionEnabled bool                  `json:"encryption_enabled,omitempty"`
	LogBackupEnabled  bool                  `json:"log_backup_enabled"`
	ScheduledStatus   BackupScheduledStatus `json:"scheduled_status" enums:"enabled,paused,disabled"`
	DualRegionEnabled bool                  `json:"dual_region_enabled,omitempty"`
	DualRegions       []string              `json:"dual_regions,omitempty"`
	NextRunAt         int64                 `json:"next_run_at,omitempty"`
	CurrentActions    []string              `json:"current_actions"`
}

// NextGenBackupSetting is the backup configuration reported for Premium and
// Essential v2 clusters.
type NextGenBackupSetting struct {
	Mode               string                 `json:"mode,omitempty" enums:"standard_bundle,flexible_retention"`
	RetentionDays      *int32                 `json:"retention_days,omitempty"`
	DailyBackupTimeUTC string                 `json:"daily_backup_time_utc,omitempty"`
	PITRStartTime      int64                  `json:"pitr_start_time,omitempty"`
	PITREndTime        int64                  `json:"pitr_end_time,omitempty"`
	LastModifyTime     int64                  `json:"last_modify_time,omitempty"`
	BackupPolicies     []*NextGenBackupPolicy `json:"backup_policies,omitempty"`
	// Unavailable is true while a Premium or Essential v2 restore has not
	// created the target instance's backup-setting record yet.
	Unavailable bool `json:"unavailable,omitempty"`
}

type NextGenBackupPolicy struct {
	Name                string `json:"name"`
	Description         string `json:"description,omitempty"`
	PITRRetentionDays   *int32 `json:"pitr_retention_days,omitempty"`
	HourlyRetentionDays *int32 `json:"hourly_retention_days,omitempty"`
	DailyRetentionDays  *int32 `json:"daily_retention_days,omitempty"`
	DailyStartTimeUTC   string `json:"daily_start_time_utc,omitempty"`
}

// dailyTimeToCron converts a "HH:MM" daily-schedule time string to a
// standard 5-field cron expression ("MIN HOUR * * *").
// If the string cannot be parsed it is returned unchanged.
func dailyTimeToCron(timeStr string) string {
	parts := strings.SplitN(timeStr, ":", 2)
	if len(parts) != 2 {
		return timeStr
	}
	return fmt.Sprintf("%s %s * * *", parts[1], parts[0])
}

// ToGetBackupSettingResp converts a proto QueryBackupSettingResp to the
// FE-facing DedicatedBackupSetting.
func ToGetBackupSettingResp(pb *dataflow.QueryBackupSettingResp) *DedicatedBackupSetting {
	if pb == nil {
		return &DedicatedBackupSetting{CurrentActions: []string{}}
	}
	setting := &DedicatedBackupSetting{
		CurrentActions: []string{},
	}
	// Status
	switch pb.Status {
	case dataflow.BackupSettingStatus_Deleted:
		setting.ScheduledStatus = BackupScheduledStatusDisabled
	default:
		setting.ScheduledStatus = BackupScheduledStatusEnabled
	}
	// Current actions
	for _, a := range pb.Actions {
		setting.CurrentActions = append(setting.CurrentActions, a.String())
	}
	// Snapshot schedule
	if ss := pb.SnapshotBackupSchedule; ss != nil {
		setting.Schedule = dailyTimeToCron(ss.DailyScheduleTime)
		if rp := ss.RetentionPolicy; rp != nil {
			// RetentionPeriod is in days (not seconds).
			setting.RetentionDays = int32(rp.RetentionPeriod)
		}
		if nt := ss.NextBackupTime; nt != nil {
			setting.NextRunAt = nt.AsTime().Unix()
		}
		if ss.Disable {
			setting.ScheduledStatus = BackupScheduledStatusDisabled
		}
	}
	// Log backup (PiTR / continuous)
	if cb := pb.ContinuousBackupSchedule; cb != nil {
		setting.LogBackupEnabled = cb.Enable
	}
	// Dual-region
	setting.DualRegionEnabled = len(pb.Regions) > 1
	for _, r := range pb.Regions {
		if r == nil {
			continue
		}
		name := r.DisplayName
		if name == "" {
			name = r.Name
		}
		if name != "" {
			setting.DualRegions = append(setting.DualRegions, name)
		}
	}
	return setting
}

// SnapshotBackup mirrors `SnapshotBackup` in FE types.
type SnapshotBackup struct {
	ID             string             `json:"id"`
	Name           string             `json:"name"`
	Type           SnapshotBackupType `json:"type" enums:"Manual,Auto"`
	Status         BackupStatus       `json:"status" enums:"running,succeeded,failed,canceled,pending"`
	ErrorMsg       string             `json:"error_msg,omitempty"`
	SizeBytes      int64              `json:"size_bytes"`
	StartedAt      int64              `json:"started_at"`
	FinishedAt     int64              `json:"finished_at,omitempty"`
	StorageURI     string             `json:"storage_uri,omitempty"`
	TSO            string             `json:"tso,omitempty"`
	Regions        []string           `json:"regions,omitempty"`
	ClusterVersion string             `json:"cluster_version,omitempty"`
	ExpireAt       int64              `json:"expire_at,omitempty"`
	Protected      bool               `json:"protected,omitempty"`
}

// ListBackupsResp is the response body for `GET .../backups`.
type ListBackupsResp struct {
	Items []*SnapshotBackup `json:"items"`
	Total int64             `json:"total"`
}

// LogBackup mirrors `LogBackup` in FE types.
type LogBackup struct {
	ID                  string          `json:"id"`
	Name                string          `json:"name"`
	Status              LogBackupStatus `json:"status" enums:"pending,running,stopping,stopped,pausing,paused,resuming,failed,scheduled"`
	ErrorMsg            string          `json:"error_msg,omitempty"`
	CheckpointTimestamp int64           `json:"checkpoint_timestamp"`
	SizeBytes           int64           `json:"size_bytes,omitempty"`
	CreatedAt           int64           `json:"created_at,omitempty"`
	StartedAt           int64           `json:"started_at"`
	StorageURI          string          `json:"storage_uri,omitempty"`
	Regions             []string        `json:"regions,omitempty"`
	TruncateUntil       int64           `json:"truncate_until,omitempty"`
}

// ListLogBackupsResp is the response body for `GET .../log-backups` (Dedicated).
type ListLogBackupsResp struct {
	Items []*LogBackup `json:"items"`
	Total int64        `json:"total"`
}

// ToListBackupsResp converts a proto Backups to the FE-facing ListBackupsResp.
func ToListBackupsResp(pb *dataflow.Backups) *ListBackupsResp {
	out := &ListBackupsResp{Items: make([]*SnapshotBackup, 0)}
	if pb == nil {
		return out
	}
	out.Total = int64(pb.Total)
	for _, b := range pb.Items {
		if b == nil {
			continue
		}
		var startedAt int64
		if b.CreatedAt != nil {
			startedAt = b.CreatedAt.AsTime().Unix()
		}
		var finishedAt int64
		if b.FinishedAt != nil {
			finishedAt = b.FinishedAt.AsTime().Unix()
		} else if b.BackupTs != nil {
			finishedAt = b.BackupTs.AsTime().Unix()
		}
		var expireAt int64
		if b.ExpirationTime != nil {
			expireAt = b.ExpirationTime.AsTime().Unix()
		}
		regions := make([]string, 0, len(b.Regions))
		for _, r := range b.Regions {
			if r != nil {
				regions = append(regions, r.GetName())
			}
		}
		var clusterVersion string
		if b.Cluster != nil {
			clusterVersion = b.Cluster.ClusterVersion
		}
		out.Items = append(out.Items, &SnapshotBackup{
			ID:             strconv.FormatUint(b.Id, 10),
			Name:           b.Name,
			Type:           snapshotBackupType(b.Type),
			Status:         snapshotBackupStatus(b.Status),
			SizeBytes:      int64(b.Size),
			StartedAt:      startedAt,
			FinishedAt:     finishedAt,
			TSO:            strconv.FormatUint(b.BackupTso, 10),
			Regions:        regions,
			ClusterVersion: clusterVersion,
			ExpireAt:       expireAt,
			Protected:      b.Protected,
		})
	}
	return out
}

// ToListPremiumBackupsResp converts the same Premium backup catalog used by
// the TiDB Cloud console. These customer-facing records are synchronized from
// cluster-service-ng and include stable IDs, retention and trigger metadata.
func ToListPremiumBackupsResp(pb *nextgenglobal.ListClusterBackupsResponse) *ListBackupsResp {
	out := &ListBackupsResp{Items: make([]*SnapshotBackup, 0)}
	if pb == nil {
		return out
	}
	out.Total = int64(pb.GetTotalSize())
	for _, backup := range pb.GetBackups() {
		if backup == nil {
			continue
		}
		var backupAt int64
		if tso, err := strconv.ParseUint(backup.GetBackupTs(), 10, 64); err == nil {
			backupAt = int64(tso>>18) / 1000
		}
		var expireAt int64
		if backup.GetExpirationTime() != nil {
			expireAt = backup.GetExpirationTime().AsTime().Unix()
		}
		out.Items = append(out.Items, &SnapshotBackup{
			ID:        backup.GetId(),
			Name:      backup.GetDisplayName(),
			Type:      premiumSnapshotBackupType(backup.GetTriggerType()),
			Status:    premiumSnapshotBackupStatus(backup.GetState()),
			SizeBytes: backup.GetSize(),
			StartedAt: backupAt,
			// CreateTime records creation of the backup resource, not completion
			// of the backup operation. The NextGen catalog does not expose a
			// completion timestamp, so do not fabricate one for active backups.
			FinishedAt: 0,
			TSO:        backup.GetBackupTs(),
			Regions:    []string{backup.GetRegionId()},
			ExpireAt:   expireAt,
			// The NextGen Backup contract exposes expiry but no protection flag.
			// A manual trigger alone does not establish that a backup is permanent.
			Protected: false,
		})
	}
	return out
}

func ToGetNextGenBackupSettingResp(setting *nextgenglobal.BackupSetting, policies *nextgenglobal.ListBackupPoliciesResponse) *NextGenBackupSetting {
	out := &NextGenBackupSetting{}
	if setting == nil {
		out.Unavailable = true
		return out
	}
	if setting.RetentionDays != nil {
		retentionDays := setting.GetRetentionDays()
		out.RetentionDays = &retentionDays
	}
	out.DailyBackupTimeUTC = setting.GetDailyBackupTimeUtc()
	switch setting.GetMode() {
	case nextgenglobal.BackupSetting_STANDARD_BUNDLE:
		out.Mode = "standard_bundle"
	case nextgenglobal.BackupSetting_FLEXIBLE_RETENTION:
		out.Mode = "flexible_retention"
	}
	if setting.GetPitrStartTime() != nil {
		out.PITRStartTime = setting.GetPitrStartTime().AsTime().Unix()
	}
	if setting.GetPitrEndTime() != nil {
		out.PITREndTime = setting.GetPitrEndTime().AsTime().Unix()
	}
	if setting.GetLastModifyTime() != nil {
		out.LastModifyTime = setting.GetLastModifyTime().AsTime().Unix()
	}
	for _, policy := range policies.GetBackupPolicies() {
		if policy == nil {
			continue
		}
		outPolicy := &NextGenBackupPolicy{
			Name:        policy.GetName().String(),
			Description: policy.GetDescription(),
		}
		if pitr := policy.GetPitr(); pitr != nil {
			retentionDays := pitr.GetRetentionDays()
			outPolicy.PITRRetentionDays = &retentionDays
		}
		if hourly := policy.GetHourlyBackup(); hourly != nil {
			retentionDays := hourly.GetRetentionDays()
			outPolicy.HourlyRetentionDays = &retentionDays
		}
		if daily := policy.GetDailyBackup(); daily != nil {
			retentionDays := daily.GetRetentionDays()
			outPolicy.DailyRetentionDays = &retentionDays
			outPolicy.DailyStartTimeUTC = daily.GetStartTimeUtc()
		}
		out.BackupPolicies = append(out.BackupPolicies, outPolicy)
	}
	return out
}

func premiumSnapshotBackupType(t nextgenglobal.Backup_TriggerType) SnapshotBackupType {
	if t == nextgenglobal.Backup_TRIGGER_TYPE_MANUAL {
		return SnapshotBackupTypeManual
	}
	return SnapshotBackupTypeAuto
}

func premiumSnapshotBackupStatus(s nextgenglobal.Backup_State) BackupStatus {
	switch s {
	case nextgenglobal.Backup_STATE_PENDING:
		return BackupStatusPending
	case nextgenglobal.Backup_STATE_RUNNING:
		return BackupStatusRunning
	case nextgenglobal.Backup_STATE_SUCCESS:
		return BackupStatusSucceeded
	case nextgenglobal.Backup_STATE_FAILED:
		return BackupStatusFailed
	case nextgenglobal.Backup_STATE_CANCELLED:
		return BackupStatusCanceled
	default:
		return ""
	}
}

func snapshotBackupType(t dataflow.BackupType) SnapshotBackupType {
	switch t {
	case dataflow.BackupType_MANUAL:
		return SnapshotBackupTypeManual
	case dataflow.BackupType_AUTO:
		return SnapshotBackupTypeAuto
	default:
		return ""
	}
}

func snapshotBackupStatus(s dataflow.Backup_ExternalStatus) BackupStatus {
	switch s {
	case dataflow.Backup_RUNNING:
		return BackupStatusRunning
	case dataflow.Backup_SUCCESS:
		return BackupStatusSucceeded
	case dataflow.Backup_FAILED:
		return BackupStatusFailed
	case dataflow.Backup_PENDING:
		return BackupStatusPending
	default:
		return ""
	}
}

// ToListLogBackupsResp converts a proto LogBackupListResp to the
// FE-facing ListLogBackupsResp.
func ToListLogBackupsResp(pb *dataflow.LogBackupListResp) *ListLogBackupsResp {
	out := &ListLogBackupsResp{Items: make([]*LogBackup, 0)}
	if pb == nil {
		return out
	}
	out.Total = int64(pb.Total)
	for _, lb := range pb.Items {
		if lb == nil {
			continue
		}
		var createdAt int64
		if lb.CreatedAt != nil {
			createdAt = lb.CreatedAt.AsTime().Unix()
		}
		var startedAt int64
		if lb.StartTs != nil {
			startedAt = lb.StartTs.AsTime().Unix()
		}
		var checkpointTimestamp int64
		if lb.CheckPointTs != nil {
			checkpointTimestamp = lb.CheckPointTs.AsTime().Unix()
		}
		var truncateUntil int64
		if lb.TruncateUntil != nil {
			truncateUntil = lb.TruncateUntil.AsTime().Unix()
		}
		out.Items = append(out.Items, &LogBackup{
			ID:                  strconv.FormatUint(lb.Id, 10),
			Name:                lb.Name,
			Status:              logBackupStatus(lb.Status),
			CheckpointTimestamp: checkpointTimestamp,
			CreatedAt:           createdAt,
			StartedAt:           startedAt,
			TruncateUntil:       truncateUntil,
		})
	}
	return out
}

func logBackupStatus(s dataflow.LogBackup_Status) LogBackupStatus {
	switch s {
	case dataflow.LogBackup_PENDING:
		return LogBackupStatusPending
	case dataflow.LogBackup_RUNNING:
		return LogBackupStatusRunning
	case dataflow.LogBackup_STOPPING:
		return LogBackupStatusStopping
	case dataflow.LogBackup_STOPPED:
		return LogBackupStatusStopped
	case dataflow.LogBackup_PAUSING:
		return LogBackupStatusPausing
	case dataflow.LogBackup_PAUSED:
		return LogBackupStatusPaused
	case dataflow.LogBackup_RESUMING:
		return LogBackupStatusResuming
	case dataflow.LogBackup_FAILED:
		return LogBackupStatusFailed
	case dataflow.LogBackup_SCHEDULED:
		return LogBackupStatusScheduled
	default:
		return ""
	}
}

// BackupExport is a cluster-scoped backup export job for Dedicated.
type BackupExport struct {
	ID          string             `json:"id"`
	Name        string             `json:"name"`
	BackupID    string             `json:"backup_id"`
	BackupName  string             `json:"backup_name"`
	BackupAt    int64              `json:"backup_at,omitempty"`
	Provider    string             `json:"provider,omitempty"`
	Region      string             `json:"region,omitempty"`
	StorageType string             `json:"storage_type,omitempty" enums:"s3,gcs,azblob,oss"`
	StorageURI  string             `json:"storage_uri,omitempty"`
	Status      BackupExportStatus `json:"status" enums:"pending,running,succeeded,failed,canceled,unknown"`
	StatusMsg   string             `json:"status_msg,omitempty"`
	ErrorMsg    string             `json:"error_msg,omitempty"`
	CreatedAt   int64              `json:"created_at,omitempty"`
	FinishedAt  int64              `json:"finished_at,omitempty"`
}

// ListBackupExportsResp is the response body for `GET .../backup-exports`.
type ListBackupExportsResp struct {
	Items []*BackupExport `json:"items"`
	Total int64           `json:"total"`
}

// Export is a Premium/Essential v2 data-export task. It is intentionally
// separate from Dedicated BackupExport: NextGen exports originate from the
// Export service and are not exports of Dedicated backup records.
type Export struct {
	ID              string       `json:"id"`
	Name            string       `json:"name"`
	SourceClusterID string       `json:"source_cluster_id,omitempty"`
	SnapshotTSO     string       `json:"snapshot_tso,omitempty"`
	SnapshotAt      int64        `json:"snapshot_at,omitempty"`
	StorageType     string       `json:"storage_type,omitempty" enums:"s3,gcs,azblob,oss"`
	StorageURI      string       `json:"storage_uri,omitempty"`
	StorageAuthType string       `json:"storage_auth_type,omitempty"`
	Status          ExportStatus `json:"status" enums:"pending,running,succeeded,failed,canceled,unknown"`
	StatusMsg       string       `json:"status_msg,omitempty"`
	ErrorMsg        string       `json:"error_msg,omitempty"`
	CreatedAt       int64        `json:"created_at,omitempty"`
	UpdatedAt       int64        `json:"updated_at,omitempty"`
	FinishedAt      int64        `json:"finished_at,omitempty"`
	FileType        string       `json:"file_type,omitempty"`
	Compression     string       `json:"compression,omitempty"`
	FilterType      string       `json:"filter_type,omitempty"`
	TablePatterns   []string     `json:"table_patterns,omitempty"`
	CSVSeparator    string       `json:"csv_separator,omitempty"`
	CSVDelimiter    string       `json:"csv_delimiter,omitempty"`
	CSVNullValue    string       `json:"csv_null_value,omitempty"`
	CSVSkipHeader   *bool        `json:"csv_skip_header,omitempty"`
	CreatedBy       string       `json:"created_by,omitempty"`
}

// ListExportsResp is the response body for Premium/Essential v2
// `GET .../exports`.
type ListExportsResp struct {
	Items []*Export `json:"items"`
	Total int64     `json:"total"`
}

// ToListBackupExportsResp converts dataflow backup-export records to the
// stable, frontend-facing Tools response.
func ToListBackupExportsResp(pb *dataflow.ListBackupExportsResp) *ListBackupExportsResp {
	out := &ListBackupExportsResp{Items: make([]*BackupExport, 0)}
	if pb == nil {
		return out
	}
	out.Total = int64(pb.Total)
	for _, export := range pb.Items {
		if export == nil {
			continue
		}

		var backupAt, createdAt, finishedAt int64
		if export.BackupTs != nil {
			backupAt = export.BackupTs.AsTime().Unix()
		}
		if export.CreatedAt != nil {
			createdAt = export.CreatedAt.AsTime().Unix()
		}
		if export.FinishedAt != nil {
			finishedAt = export.FinishedAt.AsTime().Unix()
		}

		storageType, storageURI := backupExportStorage(export)
		item := &BackupExport{
			ID:          strconv.FormatUint(export.Id, 10),
			Name:        export.Name,
			BackupID:    strconv.FormatUint(export.BackupId, 10),
			BackupName:  export.BackupName,
			BackupAt:    backupAt,
			Provider:    export.Provider,
			Region:      export.Region,
			StorageType: storageType,
			StorageURI:  sanitizeExportStorageURI(storageURI),
			Status:      backupExportStatus(export.Status),
			StatusMsg:   export.StatusMsg,
			ErrorMsg:    export.ErrorMsg,
			CreatedAt:   createdAt,
			FinishedAt:  finishedAt,
		}
		out.Items = append(out.Items, item)
	}
	return out
}

// ToListPremiumExportsResp converts the Premium/Essential v2 Export service
// catalog used by the TiDB Cloud console to the stable Tools response.
func ToListPremiumExportsResp(pb *nextgenglobal.ListExportsResponse) *ListExportsResp {
	out := &ListExportsResp{Items: make([]*Export, 0)}
	if pb == nil {
		return out
	}
	out.Total = int64(pb.GetTotalSize())
	for _, export := range pb.GetExports() {
		if export == nil {
			continue
		}
		name := export.GetDisplayName()
		if name == "" {
			name = export.GetExportId()
		}
		var backupAt, createdAt, updatedAt, finishedAt int64
		if export.GetSnapshotTime() != nil {
			backupAt = export.GetSnapshotTime().AsTime().Unix()
		}
		if export.GetCreateTime() != nil {
			createdAt = export.GetCreateTime().AsTime().Unix()
		}
		if export.GetUpdateTime() != nil {
			updatedAt = export.GetUpdateTime().AsTime().Unix()
		}
		if export.GetCompleteTime() != nil {
			finishedAt = export.GetCompleteTime().AsTime().Unix()
		}
		storageType, storageURI := premiumBackupExportStorage(export.GetCloudStorage())
		metadata := premiumBackupExportMetadata(export)
		snapshotTSO := ""
		if export.GetSnapshotTso() > 0 {
			snapshotTSO = strconv.FormatUint(export.GetSnapshotTso(), 10)
		}
		errorMsg := ""
		if export.GetState() == nextgenglobal.Export_FAILED {
			errorMsg = export.GetReason()
		}
		out.Items = append(out.Items, &Export{
			ID:              export.GetExportId(),
			Name:            name,
			SourceClusterID: export.GetTidbId(),
			SnapshotTSO:     snapshotTSO,
			SnapshotAt:      backupAt,
			StorageType:     storageType,
			StorageURI:      sanitizeExportStorageURI(storageURI),
			StorageAuthType: metadata.StorageAuthType,
			Status:          premiumExportStatus(export.GetState()),
			StatusMsg:       export.GetReason(),
			ErrorMsg:        errorMsg,
			CreatedAt:       createdAt,
			UpdatedAt:       updatedAt,
			FinishedAt:      finishedAt,
			FileType:        export.GetExportOptions().GetFileType().String(),
			Compression:     export.GetExportOptions().GetCompression().String(),
			FilterType:      metadata.FilterType,
			TablePatterns:   metadata.TablePatterns,
			CSVSeparator:    metadata.CSVSeparator,
			CSVDelimiter:    metadata.CSVDelimiter,
			CSVNullValue:    metadata.CSVNullValue,
			CSVSkipHeader:   metadata.CSVSkipHeader,
			CreatedBy:       export.GetCreatedBy(),
		})
	}
	return out
}

const redactedExportValue = "[REDACTED]"

type premiumExportMetadata struct {
	StorageAuthType string
	FilterType      string
	TablePatterns   []string
	CSVSeparator    string
	CSVDelimiter    string
	CSVNullValue    string
	CSVSkipHeader   *bool
}

func premiumBackupExportMetadata(export *nextgenglobal.Export) premiumExportMetadata {
	metadata := premiumExportMetadata{}
	if export == nil {
		return metadata
	}

	if options := export.GetExportOptions(); options != nil {
		optionDetails := map[string]any{
			"file_type":   options.GetFileType().String(),
			"compression": options.GetCompression().String(),
		}
		if filter := options.GetFilter(); filter != nil {
			filterDetails := map[string]any{}
			switch {
			case filter.GetSql() != "":
				metadata.FilterType = "SQL"
				filterDetails["type"] = "SQL"
				filterDetails["sql"] = redactedExportValue
			case filter.GetTable() != nil:
				metadata.FilterType = "TABLE"
				metadata.TablePatterns = append([]string(nil), filter.GetTable().GetPatterns()...)
				filterDetails["type"] = "TABLE"
				filterDetails["patterns"] = metadata.TablePatterns
				if filter.GetTable().GetWhere() != "" {
					filterDetails["where"] = redactedExportValue
				}
			}
			if len(filterDetails) > 0 {
				optionDetails["filter"] = filterDetails
			}
		}
		if csv := options.GetCsvFormat(); csv != nil {
			skipHeader := csv.GetSkipHeader()
			metadata.CSVSeparator = csv.GetSeparator()
			metadata.CSVDelimiter = csv.GetDelimiter()
			metadata.CSVNullValue = csv.GetNullValue()
			metadata.CSVSkipHeader = &skipHeader
			optionDetails["csv_format"] = map[string]any{
				"separator":   metadata.CSVSeparator,
				"delimiter":   metadata.CSVDelimiter,
				"null_value":  metadata.CSVNullValue,
				"skip_header": skipHeader,
			}
		}
	}

	metadata.StorageAuthType = getPremiumCloudStorageAuthType(export.GetCloudStorage())

	return metadata
}

func getPremiumCloudStorageAuthType(storage *nextgenglobal.CloudStorage) string {
	if storage == nil {
		return ""
	}
	var authType string
	switch {
	case storage.GetS3() != nil:
		value := storage.GetS3()
		authType = value.GetAuthType().String()
	case storage.GetGcs() != nil:
		value := storage.GetGcs()
		authType = value.GetAuthType().String()
	case storage.GetAzureBlob() != nil:
		value := storage.GetAzureBlob()
		authType = value.GetAuthType().String()
	case storage.GetOss() != nil:
		value := storage.GetOss()
		authType = value.GetAuthType().String()
	}
	return authType
}

func sanitizeExportStorageURI(value string) string {
	parsed, err := url.Parse(value)
	if err != nil {
		// A malformed URI cannot be safely normalized: it may contain userinfo
		// that url.Parse did not expose. Omit it rather than returning credentials.
		return ""
	}
	parsed.User = nil
	parsed.RawQuery = ""
	parsed.ForceQuery = false
	parsed.Fragment = ""
	return parsed.String()
}

func premiumBackupExportStorage(storage *nextgenglobal.CloudStorage) (string, string) {
	if storage == nil {
		return "", ""
	}
	if value := storage.GetS3(); value != nil {
		return "s3", value.GetUri()
	}
	if value := storage.GetGcs(); value != nil {
		return "gcs", value.GetUri()
	}
	if value := storage.GetAzureBlob(); value != nil {
		return "azblob", value.GetUri()
	}
	if value := storage.GetOss(); value != nil {
		return "oss", value.GetUri()
	}
	return "", ""
}

func premiumExportStatus(status nextgenglobal.Export_State) ExportStatus {
	switch status {
	case nextgenglobal.Export_STATE_UNSPECIFIED:
		return ExportStatusPending
	case nextgenglobal.Export_RUNNING:
		return ExportStatusRunning
	case nextgenglobal.Export_SUCCEEDED:
		return ExportStatusSucceeded
	case nextgenglobal.Export_FAILED:
		return ExportStatusFailed
	case nextgenglobal.Export_CANCELED, nextgenglobal.Export_DELETED:
		return ExportStatusCanceled
	default:
		return ExportStatusUnknown
	}
}

func backupExportStorage(export *dataflow.BackupExport) (string, string) {
	if storage := export.GetS3(); storage != nil {
		return "s3", storage.TargetUri
	}
	if storage := export.GetGcs(); storage != nil {
		return "gcs", storage.TargetUri
	}
	if storage := export.GetAzBlob(); storage != nil {
		return "azblob", storage.TargetUri
	}
	return "", ""
}

func backupExportStatus(status dataflow.BackupExport_Status) BackupExportStatus {
	switch status {
	case dataflow.BackupExport_PENDING:
		return BackupExportStatusPending
	case dataflow.BackupExport_RUNNING:
		return BackupExportStatusRunning
	case dataflow.BackupExport_SUCCESS:
		return BackupExportStatusSucceeded
	case dataflow.BackupExport_FAILED:
		return BackupExportStatusFailed
	case dataflow.BackupExport_CANCELED:
		return BackupExportStatusCanceled
	default:
		return BackupExportStatusUnknown
	}
}