package export

import (
	"bytes"
	"context"
	"encoding/csv"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"slices"
	"strings"
	"time"

	"github.com/pingcap/errors"
	"github.com/tidbcloud/aws-shared-provider/pkg/types"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils/labels"
	"github.com/tidbcloud/aws-shared-provider/pkg/utils/pd"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-api/pkg/consts"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

var (
	exportSpeedLimit = pd.SpeedLimitSetting{FillRate: 2_000_000, BurstLimit: 50_000_000_000} //nolint:mnd // ignore
	ExportGVK        = cloudv1.GroupVersion.WithKind("Export")
)

const (
	ResourceGroupModeRU  = 1
	DialBackendInterval  = 20 * time.Second
	AssignTiDBMaxAttempt = 1
	PdAddr               = "serverless-cluster-pd.tidb-serverless.svc.cluster.local:2379"
	//nolint:lll // nolint
	// DefaultTableFilter follow https://github.com/tidbcloud/dataservices/blob/16a450b4cc18360c92c6cf316842f55b6cb127bf/internal/service/central/query.go#L34
	DefaultTableFilter = "!/^(mysql|sys|INFORMATION_SCHEMA|PERFORMANCE_SCHEMA|METRICS_SCHEMA|INSPECTION_SCHEMA" +
		"|lightning_task_info|__TiDB_BR_Temporary_mysql|tidbcloud_dm_meta|tidb_cdc)$/.*"

	certDir  = "/server-cert"
	caPath   = certDir + "/ca.crt"
	certPath = certDir + "/tls.crt"
	keyPath  = certDir + "/tls.key"
)

func (r *reconcileContext) startExport() error {
	r.Info("start export")

	if r.export.Annotations[types.AnnoExportType] == "" {
		if err := r.ensureExportType(); err != nil {
			return errors.Annotate(err, "failed to ensure export type")
		}
		return nil
	}

	if r.export.Annotations[types.AnnoExportEtcdKey] == "" {
		if err := r.ensureEtcdToken(); err != nil {
			return errors.Annotate(err, "failed to create etcd token")
		}
		return nil
	}

	if r.export.Annotations[types.AnnoExportResourceGroup] == "" {
		if err := r.ensureResourceGroup(); err != nil {
			return errors.Annotate(err, "failed to create resource group")
		}
		return nil
	}

	r.Info("start export by keyspace", "keyspace-name", r.keyspaceToken)
	if err := r.checkKeyspaceNotMigrating(); err != nil {
		return err
	}

	if r.export.Annotations[types.AnnoExportTiDBPod] == "" {
		if err := r.ensureTiDBPod(); err != nil {
			// resolveReconcileErr type-asserts on *RequeueAfterError, so it must not be wrapped.
			if _, ok := err.(*utils.RequeueAfterError); ok {
				return err
			}
			// Exhausting the wait budget is terminal: move to Failed so cleanExport releases the
			// etcd token and the resource group instead of retrying this export forever.
			if unRetryableError, ok := err.(*UnRetryableError); ok {
				_, err = r.updateProgress(cloudv1.ExportPhaseFailed, unRetryableError.Error())
			}
			if err != nil {
				return errors.Annotate(err, "failed to assign tidb pod")
			}
		}
		return nil
	}

	if r.export.Annotations[types.AnnoExportDumplingJob] == "" {
		if err := r.ensureDumplingJob(); err != nil {
			if unRetryableError, ok := err.(*UnRetryableError); ok {
				_, err = r.updateProgress(cloudv1.ExportPhaseFailed, unRetryableError.Error())
			}
			if err != nil {
				return errors.Annotate(err, "failed to start dumpling job")
			}
		}
		return nil
	}

	if _, err := r.updateProgress(cloudv1.ExportPhaseRunning, "export is running"); err != nil {
		return err
	}
	return nil
}

func (r *reconcileContext) checkKeyspaceNotMigrating() error {
	if r.MigrationLocker != nil && r.MigrationLocker.IsLocked(r.keyspaceToken) {
		return utils.ErrKeyspaceMetaServiceGroupMigrating
	}
	return nil
}

func (r *reconcileContext) ensureExportType() error {
	exportCm := &corev1.ConfigMap{}
	if err := r.Client.Get(context.TODO(),
		client.ObjectKey{
			Name:      ExportConfigCM,
			Namespace: r.ExportJobNS,
		}, exportCm); err != nil {
		return err
	}

	exportType := ExportTypeDefault
	if r.poolSelector == vipTidbPool || r.poolSelector == canaryVipTidbPool || r.poolSelector == superVipTidbPool {
		exportType = ExportTypeVIP
	}
	superVipClusters := []string{}
	if val, ok := exportCm.Data[superVipWhiteList]; ok {
		superVipClusters = strings.Split(val, ",")
	}
	if r.export.Annotations[types.AnnoExportSuperVip] == "true" ||
		slices.Contains(superVipClusters, r.export.Spec.ClusterName) {
		exportType = ExportTypeSuperVIP
	}
	r.export.Annotations[types.AnnoExportType] = string(exportType)
	return r.InfraClient.Update(r.ctx, r.export)
}

// ensureEtcdToken is idempotent
func (r *reconcileContext) ensureEtcdToken() error {
	exportType := r.export.Annotations[types.AnnoExportType]
	exportPoolNs := ExportTiDBPoolNS
	if exportType == string(ExportTypeVIP) {
		exportPoolNs = ExportVipTiDBPoolNS
	} else if exportType == string(ExportTypeSuperVIP) {
		exportPoolNs = ExportSuperVipTiDBPoolNS
	}

	etcdTokenPrefix := r.EtcdTokenPrefixMap[exportPoolNs]
	token := r.keyspaceToken + ":" + r.export.Name
	keyEndpoints := etcdTokenPrefix + token + "/" + utils.KeyEndpoints
	resp, err := r.EtcdKV.Get(r.ctx, keyEndpoints)
	if err != nil {
		return errors.Annotatef(err, "failed to get etcd endpoint path %s", keyEndpoints)
	}
	if len(resp.Kvs) == 0 {
		if _, err := r.EtcdKV.Put(r.ctx, keyEndpoints, ""); err != nil {
			return errors.Annotatef(err, "failed to put endpoint etcd path %s", keyEndpoints)
		}
		r.InfraRecorder.Event(r.export, corev1.EventTypeNormal, "created etcd token", "etcd token endpoint is set to "+token)
	}

	keyReplicas := etcdTokenPrefix + token + "/" + utils.KeyReplicas
	resp, err = r.EtcdKV.Get(r.ctx, keyReplicas)
	if err != nil {
		return errors.Annotatef(err, "failed to get replicas etcd path %s", keyEndpoints)
	}
	if len(resp.Kvs) == 0 {
		if _, err := r.EtcdKV.Put(r.ctx, keyReplicas, "1"); err != nil {
			return errors.Annotatef(err, "failed to put replicas etcd path %s", keyReplicas)
		}
		r.InfraRecorder.Event(r.export, corev1.EventTypeNormal,
			"created etcd token", "etcd token replica is set to "+token)
	}

	r.export.Annotations[types.AnnoExportEtcdKey] = token
	return r.InfraClient.Update(r.ctx, r.export)
}

func (r *reconcileContext) ensureResourceGroup() error {
	resourceGroupName, err := r.exportResourceGroupName()
	if err != nil {
		return err
	}
	if err := r.updateResourceGroup(resourceGroupName); err != nil {
		return err
	}
	r.export.Annotations[types.AnnoExportResourceGroup] = resourceGroupName
	return r.InfraClient.Update(r.ctx, r.export)
}

func (r *reconcileContext) updateResourceGroup(name string) error {
	created, err := r.ensureResourceGroupLimit(name, exportSpeedLimit)
	if err != nil {
		return err
	}
	if created {
		r.InfraRecorder.Event(r.export, corev1.EventTypeNormal, "create resource group",
			"resource group name is "+name)
	}
	return nil
}

func (r *reconcileContext) ensureResourceGroupLimit(name string, setting pd.SpeedLimitSetting) (bool, error) {
	err := r.sendResourceLimit(r.PdClient.PutResourceGroup, name, setting)
	if err == nil {
		return false, nil
	}

	// PD AddResourceGroup is currently an idempotent upsert, so POST is used
	// as the fallback when updating an existing group fails.
	err = r.sendResourceLimit(r.PdClient.PostResourceGroup, name, setting)
	if err == nil {
		return true, nil
	}
	r.Error(err, "try to create resource limit", "resource-group", name)
	return false, errors.Annotatef(err, "failed to create resource group: %s", name)
}

func (r *reconcileContext) sendResourceLimit(
	writer func(ctx context.Context, group *pd.ResourceGroup) error,
	name string,
	setting pd.SpeedLimitSetting,
) error {
	group := &pd.ResourceGroup{
		Name: name,
		Mode: ResourceGroupModeRU,
		RUSettings: pd.RUSettings{
			RU: pd.RU{
				Settings: setting,
			},
		},
	}

	return writer(r.ctx, group)
}

func (r *reconcileContext) exportResourceGroupName() (string, error) {
	exportCm := &corev1.ConfigMap{}
	if err := r.Client.Get(r.ctx,
		client.ObjectKey{
			Name:      ExportConfigCM,
			Namespace: r.ExportJobNS,
		}, exportCm); err != nil {
		if apierrors.IsNotFound(err) {
			return r.export.Name, nil
		}
		return "", err
	}
	if strings.EqualFold(strings.TrimSpace(exportCm.Data[enableNewResourceGroupName]), "true") {
		return buildExportResourceGroupName(r.keyspaceID), nil
	}
	return r.export.Name, nil
}

func (r *reconcileContext) ensureTiDBPod() error {
	exportType := r.export.Annotations[types.AnnoExportType]
	exportPoolNs := ExportTiDBPoolNS
	if exportType == string(ExportTypeVIP) {
		exportPoolNs = ExportVipTiDBPoolNS
	} else if exportType == string(ExportTypeSuperVIP) {
		exportPoolNs = ExportSuperVipTiDBPoolNS
	}

	r.Info("start to assign tidb pod", "exportPoolNs", exportPoolNs)
	backends, err := r.assignTiDB(r.ctx, AssignTiDBMaxAttempt, exportPoolNs)
	if err != nil {
		maxWait := r.assignTiDBMaxWait()
		if waited := time.Since(r.export.CreationTimestamp.Time); waited < maxWait {
			r.Info("waiting for export tidb pod to be ready",
				"exportPoolNs", exportPoolNs, "waited", waited, "maxWait", maxWait)
			return &utils.RequeueAfterError{Cause: err, RequeueAfter: assignTiDBRequeueInterval}
		}
		r.Error(err, "give up assigning tidb pod", "exportPoolNs", exportPoolNs, "maxWait", maxWait)
		return &UnRetryableError{
			message: fmt.Sprintf("no available tidb pod in %s after waiting %s", exportPoolNs, maxWait),
		}
	}
	r.Info("assign tidb pod success", "exportPoolNs", exportPoolNs)

	r.export.Annotations[types.AnnoExportTiDBPod] = strings.Join(backends, ",")
	return r.InfraClient.Update(r.ctx, r.export)
}

// assignTiDBMaxWait is the budget for an on-demand export TiDB pod to show up. It is
// tunable per region through the export config map.
//
// The budget is deliberately counted from the export creation rather than from the first
// assign: it caps how long the user waits in total, so the earlier prepare steps (etcd
// token, resource group), which each take one reconcile round, eat into the same budget.
func (r *reconcileContext) assignTiDBMaxWait() time.Duration {
	exportCm := &corev1.ConfigMap{}
	if err := r.Client.Get(r.ctx,
		client.ObjectKey{
			Name:      ExportConfigCM,
			Namespace: r.ExportJobNS,
		}, exportCm); err != nil {
		return DefaultAssignTiDBMaxWait
	}

	value := strings.TrimSpace(exportCm.Data[assignTiDBMaxWaitKey])
	if value == "" {
		return DefaultAssignTiDBMaxWait
	}
	maxWait, err := time.ParseDuration(value)
	if err != nil {
		r.Error(err, "unparsable assign tidb max wait, fallback to default", "value", value)
		return DefaultAssignTiDBMaxWait
	}
	if maxWait <= 0 {
		r.Info("non-positive assign tidb max wait, fallback to default", "value", value)
		return DefaultAssignTiDBMaxWait
	}
	return maxWait
}

func (r *reconcileContext) assignTiDB(ctx context.Context, maxAttempts int, ns string) (backends []string, err error) {
	token := r.keyspaceToken + ":" + r.export.Name
	tenantID := r.tenantID
	projectID := r.projectID
	clusterID := r.clusterID

	urlFormat := "https://manager-server:8000/api/tidb/assign?token=%s&ns=%s" +
		"&project_id=%s" +
		"&cluster_id=%s" +
		"&tenant_id=%s"
	url := fmt.Sprintf(urlFormat, token, ns, projectID, clusterID, tenantID)
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
	if err != nil {
		return nil, err
	}

	for i := 1; i <= maxAttempts; i++ {
		backends, err = func() ([]string, error) {
			resp, err := r.HttpClient.Do(req)
			if err != nil {
				return nil, err
			}
			defer resp.Body.Close() //nolint:errcheck // no need check error

			body, err := io.ReadAll(resp.Body)
			if err != nil {
				return nil, err
			}
			if resp.StatusCode != http.StatusOK {
				return nil, errors.Errorf("manager return %s, body: %s", resp.Status, body)
			}
			return strings.Split(string(body), ","), nil
		}()
		if err == nil {
			break
		}
		if i < maxAttempts {
			time.Sleep(DialBackendInterval)
		}
	}
	if err == nil && len(backends) == 0 {
		return nil, errors.New("get empty pod list from manager")
	}
	return backends, err
}

//nolint:funlen // by design
func (r *reconcileContext) ensureDumplingJob() error {
	exportJob, err := r.getExportJob(context.TODO())
	if err != nil {
		return fmt.Errorf("can't get job template: %w", err)
	}

	jo, err := r.applyCfgFromCm(context.TODO(), exportJob)
	if err != nil {
		return fmt.Errorf("can't build job: %w", err)
	}
	jo.Name = r.BuildDumplingJobName()
	jo.Namespace = r.ExportJobNS

	// build command
	tidbBackends := r.export.Annotations[types.AnnoExportTiDBPod]
	tidb := strings.Split(tidbBackends, ",")[0]
	command := []string{
		"--pd=" + PdAddr,
		"-u=" + r.export.Spec.Secret.User,
		"-p=" + r.export.Spec.Secret.Password,
		"-h=" + tidb,
		"--filetype=" + dumplingFileType(r.export.Spec.Options.FileType),
		"--keyspace-name=" + r.keyspaceToken,
	}

	var volumeMounts []corev1.VolumeMount
	if r.EnableTLS {
		// Add the cluster secret to the job
		jo.Spec.Template.Spec.Volumes = append(jo.Spec.Template.Spec.Volumes, corev1.Volume{
			Name: "server-cert",
			VolumeSource: corev1.VolumeSource{
				Secret: &corev1.SecretVolumeSource{
					SecretName: utils.ExportClusterSecretName,
				},
			},
		})
		// Mount the cluster secret to the container
		volumeMounts = append(volumeMounts, corev1.VolumeMount{
			Name:      "server-cert",
			MountPath: certDir,
		})
		// Enable TLS in the app
		command = append(
			command,
			fmt.Sprintf("--ca=%s", caPath),
			fmt.Sprintf("--cert=%s", certPath),
			fmt.Sprintf("--key=%s", keyPath),
		)
	}

	// add output option
	outputURL, err := r.buildOutputURL()
	if err != nil {
		return err
	}
	output := outputURL.String()
	command = append(command, "-o="+output)
	// add filter option
	filterConst := "--filter="
	if r.export.Spec.Options.SQL != "" {
		command = append(command, "--sql="+r.export.Spec.Options.SQL)
	} else {
		if r.export.Spec.Options.Where != "" {
			command = append(command, "--where="+r.export.Spec.Options.Where)
		}
		// we need to exclude the system tables
		if r.export.Spec.Options.TableFilter == nil {
			command = append(command, "--filter=*.*")
		} else {
			dumplingFilter, err := buildDumplingFilter(r.export.Spec.Options.TableFilter)
			if err != nil {
				r.Error(err, "failed to build filters")
				return &UnRetryableError{message: "invalid table filters"}
			}
			command = append(command, filterConst+dumplingFilter)
		}
		command = append(command, filterConst+DefaultTableFilter)
	}
	// Dumpling accepts each selector as one inline TOML --column-filter rule.
	// Keep the legacy command byte-for-byte unchanged when no selector is set.
	for _, columnSelector := range r.export.Spec.Options.ColumnSelectors {
		columnFilter, err := buildDumplingColumnFilter(columnSelector)
		if err != nil {
			r.Error(err, "failed to build column filter")
			return &UnRetryableError{message: "invalid column selector"}
		}
		command = append(command, "--column-filter="+columnFilter)
	}

	// add compression option
	if r.export.Spec.Options.Compression != "" {
		command = append(command, "--compress="+string(r.export.Spec.Options.Compression))
	}
	// add snapshot option
	if r.export.Spec.Options.Snapshot != "" {
		command = append(command, "--snapshot="+r.export.Spec.Options.Snapshot)
	}
	// add csv options, delimiter and null-value can be empty string
	if r.export.Spec.Options.FileType == cloudv1.FileTypeCSV {
		if r.export.Spec.Options.CSVOptions.Separator != "" {
			command = append(command, "--csv-separator="+r.export.Spec.Options.CSVOptions.Separator)
		}
		if r.export.Spec.Options.CSVOptions.Delimiter != nil {
			command = append(command, "--csv-delimiter="+*r.export.Spec.Options.CSVOptions.Delimiter)
		}
		if r.export.Spec.Options.CSVOptions.Null != nil {
			command = append(command, "--csv-null-value="+*r.export.Spec.Options.CSVOptions.Null)
		}
		if r.export.Spec.Options.CSVOptions.NoHeader {
			command = append(command, "--no-header")
		}
		if r.export.Spec.Options.CSVOptions.Dialect != "" {
			command = append(command, "--csv-output-dialect="+r.export.Spec.Options.CSVOptions.Dialect)
		}
	}
	// add escape-backslash option; tri-state so the dumpling default is kept
	// unless the export explicitly sets it.
	if r.export.Spec.Options.EscapeBackslash != nil {
		command = append(command, fmt.Sprintf("--escape-backslash=%t", *r.export.Spec.Options.EscapeBackslash))
	}
	// add parquet options
	if r.export.Spec.Options.FileType == "PARQUET" {
		compression := string(r.export.Spec.Options.ParquetOptions.Compression)
		if compression != "" {
			if compression == "none" {
				compression = "no-compression"
			}
			command = append(command, "--parquet-compress="+compression)
		}
	}

	// add partition option
	if len(r.export.Spec.Options.Partition) > 0 {
		for _, partition := range r.export.Spec.Options.Partition {
			command = append(command, "--partitions="+partition)
		}
	}

	// build container
	containers := jo.Spec.Template.Spec.Containers
	for i := range containers {
		if containers[i].Name == DumplingContainerName {
			containers[i].Image = r.DumplingImage
			containers[i].Command = append(containers[i].Command, command...)
			containers[i].VolumeMounts = append(containers[i].VolumeMounts, volumeMounts...)
			break
		}
	}

	// Add pod finalizer to ensure controller can collect termination message before pod is deleted
	if jo.Spec.Template.ObjectMeta.Finalizers == nil {
		jo.Spec.Template.ObjectMeta.Finalizers = []string{exportFinalizer}
	}

	if jo.Spec.Template.Labels == nil {
		jo.Spec.Template.Labels = map[string]string{}
	}

	serverlessClusterInfoLabels := r.getServerlessClusterInfoLabels()
	mergeLables(jo.Spec.Template.Labels, serverlessClusterInfoLabels)

	// add labels
	jobLabels := map[string]string{
		types.LabelJobType:               DumplingJobType,
		cloudv1.ProviderTypeLabel:        r.ProviderType,
		labels.LabelKeyKubernetesCluster: r.KubeClusterName,
		labels.ZREvicterIgnoredKey:       "true",
	}

	mergeLables(jobLabels, serverlessClusterInfoLabels)

	jobLabels = labels.NewOwnerLabels(r.export, jobLabels)
	jo.SetLabels(jobLabels)

	err = r.Client.Create(context.TODO(), jo)
	if err != nil {
		return err
	}
	r.export.Annotations[types.AnnoExportDumplingJob] = r.BuildDumplingJobName()
	return r.InfraClient.Update(r.ctx, r.export)
}

func mergeLables(dst, src map[string]string) {
	for k, v := range src {
		dst[k] = v
	}
}

func (r *reconcileContext) getServerlessClusterInfoLabels() map[string]string {
	labels := map[string]string{
		consts.ClusterTagKey:    r.clusterID,
		consts.ProjectTagKey:    r.projectID,
		consts.TenantTagKey:     r.tenantID,
		types.KeyspaceNameLabel: r.keyspaceToken,
	}
	return labels
}

//nolint:funlen // by design
func (r *reconcileContext) buildOutputURL() (outputURL *url.URL, err error) {
	switch r.export.Spec.Target.ExportType {
	case cloudv1.ExportTypeS3:
		outputURL, err = url.Parse(escapeS3BucketUri(r.export.Spec.Target.S3.BucketUri))
		if err != nil {
			return nil, fmt.Errorf("can't parese url: %w", err)
		}
		q := outputURL.Query()
		if r.export.Spec.Target.S3.RoleArn != "" {
			externalID := utils.GetExternalId(r.AWSAccountID, r.export.Labels[labels.LabelKeyKubernetesCluster])
			q.Set("role-arn", r.export.Spec.Target.S3.RoleArn)
			q.Set("external-id", externalID)
		}
		if r.export.Spec.Target.S3.AccessKey != "" && r.export.Spec.Target.S3.SecretAccess != "" {
			q.Set("access-key", r.export.Spec.Target.S3.AccessKey)
			q.Set("secret-access-key", r.export.Spec.Target.S3.SecretAccess)
		}
		if r.export.Spec.Target.S3.Endpoint != "" {
			q.Set("endpoint", r.export.Spec.Target.S3.Endpoint)
		}
		if r.export.Spec.Target.S3.Provider != "" {
			q.Set("provider", r.export.Spec.Target.S3.Provider)
		}
		outputURL.RawQuery = q.Encode()
		return outputURL, nil
	case cloudv1.ExportTypeGCS:
		outputURL, err = url.Parse(escapeGCSBucketUri(r.export.Spec.Target.GCS.BucketUri))
		if err != nil {
			return nil, fmt.Errorf("can't parese url: %w", err)
		}
		q := outputURL.Query()
		if r.export.Spec.Target.GCS.ServiceAccountKey != "" {
			q.Set("credentials", r.export.Spec.Target.GCS.ServiceAccountKey)
		}
		outputURL.RawQuery = q.Encode()
		return outputURL, nil
	case cloudv1.ExportTypeAzure:
		azure := r.export.Spec.Target.Azure
		outputURL, err = url.Parse(escapeAzureUri(azure.ContainerName))
		if err != nil {
			return nil, fmt.Errorf("can't parese url: %w", err)
		}
		q := outputURL.Query()
		if azure.AccountName != "" {
			q.Set("account-name", azure.AccountName)
		}
		if azure.SasToken != "" {
			q.Set("sas-token", azure.SasToken)
		}
		outputURL.RawQuery = q.Encode()
		return outputURL, nil
	case cloudv1.ExportTypeLOCAL:
		switch r.ProviderType {
		case types.ProviderTypeAWSFreeTier:
			outputURL, err = url.Parse(r.BuildBucketPath())
			if err != nil {
				return nil, fmt.Errorf("can't parese url: %w", err)
			}
			return outputURL, nil
		case types.ProviderTypeAliCloudFreeTier:
			// using RSSA in the future
			parts := strings.Split(r.ExportStashingArea, "-")
			region := strings.Join(parts[1:len(parts)-2], "-")
			endpoint := fmt.Sprintf("https://oss-%s-internal.aliyuncs.com", region)
			outputURL, err = url.Parse(r.BuildBucketPath())
			if err != nil {
				return nil, fmt.Errorf("can't parese url: %w", err)
			}
			q := outputURL.Query()
			q.Set("endpoint", endpoint)
			q.Set("access-key", r.export.Spec.Target.S3.AccessKey)
			q.Set("secret-access-key", r.export.Spec.Target.S3.SecretAccess)
			q.Set("provider", "alibaba")
			outputURL.RawQuery = q.Encode()
			return outputURL, nil
		}
	}
	return nil, fmt.Errorf("unsupported export type %s", r.export.Spec.Target.ExportType)
}

func (r *reconcileContext) BuildDumplingJobName() string {
	return r.export.Name + "-dumpling"
}

func dumplingFileType(fileType cloudv1.FileType) string {
	return strings.ToLower(string(fileType))
}

func (r *reconcileContext) BuildBucketPath() string {
	return fmt.Sprintf("s3://%s/org-%s/export/%s", r.ExportStashingArea, r.tenantID, r.export.Name)
}

func buildDumplingFilter(tableFilters []string) (string, error) {
	buf := &bytes.Buffer{}
	csvWriter := csv.NewWriter(buf)
	err := csvWriter.Write(tableFilters)
	if err != nil {
		return "", err
	}
	csvWriter.Flush()
	return strings.TrimSuffix(buf.String(), "\n"), nil
}

func buildDumplingColumnFilter(selector cloudv1.ColumnSelector) (string, error) {
	if len(selector.Matcher) == 0 || len(selector.Columns) == 0 {
		return "", errors.New("column selector matcher and columns are required")
	}
	matcher, err := json.Marshal(selector.Matcher)
	if err != nil {
		return "", errors.Annotate(err, "marshal column selector matcher")
	}
	columns, err := json.Marshal(selector.Columns)
	if err != nil {
		return "", errors.Annotate(err, "marshal column selector columns")
	}
	return fmt.Sprintf("{ matcher = %s, columns = %s }", matcher, columns), nil
}

func escapeS3BucketUri(s3Uri string) string {
	// cannot use url.Parse since the s3 uri might be invalid
	// for example: s3://bucket/folder?%/
	schema := "s3://"
	bucketUri := strings.TrimPrefix(s3Uri, schema)
	decodeUri := schema
	pathSegments := strings.Split(bucketUri, "/")
	for i, s := range pathSegments {
		if i > 0 {
			// folder
			decodeUri = decodeUri + "/" + url.PathEscape(s)
		} else {
			// host/bucket
			decodeUri += s
		}
	}
	return decodeUri
}

func escapeGCSBucketUri(uri string) string {
	schema := "gcs://"
	if strings.Contains(uri, "gs://") {
		schema = "gs://"
	}
	bucketUri := strings.TrimPrefix(uri, schema)
	decodeUri := schema
	pathSegments := strings.Split(bucketUri, "/")
	for i, s := range pathSegments {
		if i > 0 {
			// folder
			decodeUri = decodeUri + "/" + url.PathEscape(s)
		} else {
			// host/bucket
			decodeUri += s
		}
	}
	return decodeUri
}

func escapeAzureUri(uri string) string {
	schema := "azure://"
	if strings.Contains(uri, "azblob://") {
		schema = "azblob://"
	}
	bucketUri := strings.TrimPrefix(uri, schema)
	decodeUri := schema
	pathSegments := strings.Split(bucketUri, "/")
	for i, s := range pathSegments {
		if i > 0 {
			// folder
			decodeUri = decodeUri + "/" + url.PathEscape(s)
		} else {
			// container
			decodeUri += s
		}
	}
	return decodeUri
}

func buildExportResourceGroupName(keyspaceID uint32) string {
	return fmt.Sprintf("exp-%d", keyspaceID)
}