package facade_imp

import (
	"context"
	"encoding/json"
	stderrors "errors"
	"fmt"
	"net/http"
	"strconv"
	"time"

	"github.com/pingcap/errors"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	proxy_pkg "github.com/tidbcloud/pkgv2/clusterproxy"
	"github.com/tidbcloud/pkgv2/clusterproxy/tidbcloud"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/pkgv2/tracing"
	"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/repo"
	"github.com/tidbcloud/serverless-service/internal/metric"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/attribute"
	"go.uber.org/zap"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/runtime/schema"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

var (
	Tracer = otel.Tracer("facade_imp")

	errBranchCRNotFoundInConfiguredProxies = errors.New(
		"branch CR not found in configured cluster proxy endpoints",
	)
)

type infraFacadeImp struct {
	projectName          string
	clusterProxyProvider facade.ClusterProxyProviderFacade
	branchRepo           repo.BranchRepo
	mgmtSvc              facade.MgmtSvcFacade
}

func NewInfraFacade(
	branchRepo repo.BranchRepo,
	mgmtSvc facade.MgmtSvcFacade,
) facade.InfraFacade {
	return &infraFacadeImp{
		projectName:          config.Global().GardenerConfig.GardenerProjectName,
		clusterProxyProvider: NewClusterProxyProviderFacade(),
		branchRepo:           branchRepo,
		mgmtSvc:              mgmtSvc,
	}
}

func (f *infraFacadeImp) ListBranches(
	ctx context.Context,
	proxyEndpoint string,
) (_ []cloudv1.Branch, err error) {
	start := time.Now()
	ctx, span := Tracer.Start(ctx, "infra.ListBranches")
	defer func() {
		tracing.EndSpan(span, err)
		status := metric.StatusOK
		if err != nil {
			status = metric.StatusErr
		}
		metric.FacadeRequestedCounter.WithLabelValues("infra", "ListBranches", status).Inc()
		metric.FacadeRequestedHistogram.WithLabelValues("infra", "ListBranches", status).
			Observe(time.Since(start).Seconds())
	}()

	return f.listBranchesViaProxy(ctx, proxyEndpoint)
}

func (f *infraFacadeImp) CreateBranch(ctx context.Context, b *cloudv1.Branch) (err error) {
	start := time.Now()
	ctx, span := Tracer.Start(ctx, "infra.CreateBranch")
	defer func() {
		tracing.EndSpan(span, err)
		status := metric.StatusOK
		if err != nil {
			status = metric.StatusErr
		}
		metric.FacadeRequestedCounter.WithLabelValues("infra", "CreateBranch", status).Inc()
		metric.FacadeRequestedHistogram.WithLabelValues("infra", "CreateBranch", status).
			Observe(time.Since(start).Seconds())
	}()
	span.SetAttributes(attribute.String("branch_id", b.Name))

	if config.IsLocalStage() {
		b.Labels[constant.LabelUsage] = constant.UsageE2E          // exclude reconciler in dev env
		b.Labels[constant.LabelTestNamespace] = config.MyNamespace // exclude reconcilers in other presonal envs
	}

	proxyEndpoint, resolveErr := f.resolveBranchProxyEndpointByBranchID(ctx, b.Name)
	if resolveErr != nil {
		return resolveErr
	}
	if err := f.createBranchViaProxy(ctx, proxyEndpoint, b); err != nil {
		tilog.Error(ctx, "create branch via proxy failed", zap.String("branch_id", b.Name), zap.Error(err))
		return err
	}
	return nil
}

func (f *infraFacadeImp) GetBranch(
	ctx context.Context,
	projectId uint64,
	branchId string,
) (b *cloudv1.Branch, err error) {
	start := time.Now()
	ctx, span := Tracer.Start(ctx, "infraFacade.GetBranch")
	defer func() {
		tracing.EndSpan(span, err)
		status := metric.StatusOK
		if err != nil {
			status = metric.StatusErr
		}
		metric.FacadeRequestedCounter.WithLabelValues("infra", "GetBranch", status).Inc()
		metric.FacadeRequestedHistogram.WithLabelValues("infra", "GetBranch", status).
			Observe(time.Since(start).Seconds())
	}()
	span.SetAttributes(
		attribute.String("branch_id", branchId),
		attribute.String("project_id", fmt.Sprint(projectId)),
	)

	proxyEndpoint, resolveErr := f.resolveBranchProxyEndpointByBranchID(ctx, branchId)
	if resolveErr != nil {
		return nil, resolveErr
	}
	return f.getBranchFromProxy(ctx, proxyEndpoint, projectId, branchId)
}

func (f *infraFacadeImp) GetBranchDirectly(
	ctx context.Context,
	projectId uint64,
	branchId string,
) (b *cloudv1.Branch, err error) {
	start := time.Now()
	ctx, span := Tracer.Start(ctx, "infraFacade.GetBranchDirectly")
	defer func() {
		tracing.EndSpan(span, err)
		status := metric.StatusOK
		if err != nil {
			status = metric.StatusErr
		}
		metric.FacadeRequestedCounter.WithLabelValues("infra", "GetBranchDirectly", status).Inc()
		metric.FacadeRequestedHistogram.WithLabelValues("infra", "GetBranchDirectly", status).
			Observe(time.Since(start).Seconds())
	}()
	span.SetAttributes(
		attribute.String("branch_id", branchId),
		attribute.String("project_id", fmt.Sprint(projectId)),
	)

	proxyEndpoint, resolveErr := f.resolveBranchProxyEndpointByBranchID(ctx, branchId)
	if resolveErr != nil {
		return nil, resolveErr
	}
	return f.getBranchDirectlyFromProxy(ctx, proxyEndpoint, projectId, branchId)
}

func (f *infraFacadeImp) UpdateBranch(ctx context.Context, b *cloudv1.Branch) (err error) {
	start := time.Now()
	ctx, span := Tracer.Start(ctx, "infra.UpdateBranch")
	defer func() {
		tracing.EndSpan(span, err)
		status := metric.StatusOK
		if err != nil {
			status = metric.StatusErr
		}
		metric.FacadeRequestedCounter.WithLabelValues("infra", "UpdateBranch", status).Inc()
		metric.FacadeRequestedHistogram.WithLabelValues("infra", "UpdateBranch", status).
			Observe(time.Since(start).Seconds())
	}()
	span.SetAttributes(attribute.String("branch_id", b.Name))

	// must update regional cr first
	proxyEndpoint, resolveErr := f.resolveBranchProxyEndpointByBranchID(ctx, b.Name)
	if resolveErr != nil {
		return resolveErr
	}
	if err := f.updateBranchViaProxy(ctx, proxyEndpoint, b); err != nil {
		tilog.Error(ctx, "update branch via proxy failed", zap.String("branch_id", b.Name), zap.Error(err))
		return err
	}
	return nil
}

func (f *infraFacadeImp) UpdateBranchStatus(ctx context.Context, b *cloudv1.Branch) (err error) {
	start := time.Now()
	ctx, span := Tracer.Start(ctx, "infra.UpdateBranchStatus")
	defer func() {
		tracing.EndSpan(span, err)
		status := metric.StatusOK
		if err != nil {
			status = metric.StatusErr
		}
		metric.FacadeRequestedCounter.WithLabelValues("infra", "UpdateBranchStatus", status).Inc()
		metric.FacadeRequestedHistogram.WithLabelValues("infra", "UpdateBranchStatus", status).
			Observe(time.Since(start).Seconds())
	}()
	span.SetAttributes(
		attribute.String("branch_id", b.Name),
		attribute.String("status", string(b.Status.Phase)),
		attribute.String("reason", b.Status.Reason),
	)

	tilog.Info(ctx, "branch status", zap.Any("status", b.Status))

	proxyEndpoint, resolveErr := f.resolveBranchProxyEndpointByBranchID(ctx, b.Name)
	if resolveErr != nil {
		return resolveErr
	}
	if err := f.updateBranchStatusViaProxy(ctx, proxyEndpoint, b); err != nil {
		tilog.Error(ctx, "update branch status via proxy failed", zap.String("branch_id", b.Name), zap.Error(err))
		return err
	}
	return nil
}

func (f *infraFacadeImp) DeleteBranch(ctx context.Context, projectID uint64, branchID string) (err error) {
	start := time.Now()
	ctx, span := Tracer.Start(ctx, "infra.DeleteBranch")
	defer func() {
		tracing.EndSpan(span, err)
		status := metric.StatusOK
		if err != nil {
			status = metric.StatusErr
		}
		metric.FacadeRequestedCounter.WithLabelValues("infra", "DeleteBranch", status).Inc()
		metric.FacadeRequestedHistogram.WithLabelValues("infra", "DeleteBranch", status).
			Observe(time.Since(start).Seconds())
	}()
	span.SetAttributes(
		attribute.String("branch_id", branchID),
		attribute.String("project_id", fmt.Sprint(projectID)),
	)

	proxyEndpoint, resolveErr := f.resolveBranchProxyEndpointByBranchID(ctx, branchID)
	if resolveErr != nil {
		if errors.Cause(resolveErr) == errBranchCRNotFoundInConfiguredProxies {
			return nil
		}
		return resolveErr
	}
	if err := f.deleteBranchViaProxy(ctx, proxyEndpoint, projectID, branchID); err != nil {
		tilog.Error(ctx, "delete branch failed", zap.Error(err))
		return err
	}
	return nil
}

func (f *infraFacadeImp) DeleteImport(
	ctx context.Context,
	proxyEndpoint string,
	projectID uint64,
	importID string,
) (err error) {
	start := time.Now()
	ctx, span := Tracer.Start(ctx, "infra.DeleteImport")
	defer func() {
		tracing.EndSpan(span, err)
		status := metric.StatusOK
		if err != nil {
			status = metric.StatusErr
		}
		metric.FacadeRequestedCounter.WithLabelValues("infra", "DeleteImport", status).Inc()
		metric.FacadeRequestedHistogram.WithLabelValues("infra", "DeleteImport", status).
			Observe(time.Since(start).Seconds())
	}()
	span.SetAttributes(
		attribute.String("import_id", importID),
		attribute.String("project_id", fmt.Sprint(projectID)),
	)

	proxy, bizErr := f.clusterProxyProvider.GetServerlessHTTPProxy(ctx, proxyEndpoint, nil, nil)
	if bizErr != nil {
		return errors.AddStack(bizErr)
	}
	_, err = proxy.CallK8sResource(ctx, &proxy_pkg.K8sResourceRequest{
		OperateType: "Delete",
		APIVersion:  "cloud.pingcap.com/v1",
		Kind:        "Import",
		Namespace:   f.Namespace(projectID),
		Name:        importID,
	})
	if err != nil {
		err = mapK8sResourceError(err, importID)
		if client.IgnoreNotFound(err) != nil {
			if isK8sAPIError(err) {
				return err
			}
			return errors.AddStack(err)
		}
	}
	log.WithContextL(ctx).With(zap.String("import_id", importID)).Info("import cr deleted")
	return nil
}

func (f *infraFacadeImp) CreateImportIgnoreExists(
	ctx context.Context,
	proxyEndpoint string,
	cr *cloudv1.Import,
) (err error) {
	start := time.Now()
	ctx, span := Tracer.Start(ctx, "infra.CreateImportIgnoreExists")
	defer func() {
		tracing.EndSpan(span, err)
		status := metric.StatusOK
		if err != nil {
			status = metric.StatusErr
		}
		metric.FacadeRequestedCounter.WithLabelValues("infra", "CreateImportIgnoreExists", status).Inc()
		metric.FacadeRequestedHistogram.WithLabelValues("infra", "CreateImportIgnoreExists", status).
			Observe(time.Since(start).Seconds())
	}()
	namespace := cr.Namespace
	span.SetAttributes(
		attribute.String("import_id", cr.Name),
		attribute.String("namespace", namespace),
	)

	proxy, bizErr := f.clusterProxyProvider.GetServerlessHTTPProxy(ctx, proxyEndpoint, nil, nil)
	if bizErr != nil {
		return errors.AddStack(bizErr)
	}
	_, err = proxy.CallK8sResource(ctx, &proxy_pkg.K8sResourceRequest{
		OperateType: "Create",
		APIVersion:  "cloud.pingcap.com/v1",
		Kind:        "Import",
		Namespace:   namespace,
		Object:      cr,
	})
	if err != nil {
		err = mapK8sResourceError(err, cr.Name)
		if apierrors.IsAlreadyExists(err) {
			return nil
		}
		if isK8sAPIError(err) {
			return err
		}
		return errors.AddStack(err)
	}
	log.WithContextL(ctx).With(zap.String("import_id", cr.Name)).Info("import cr created")
	return nil
}

func (f *infraFacadeImp) GetImport(
	ctx context.Context,
	proxyEndpoint string,
	projectID uint64,
	importID string,
) (i *cloudv1.Import, err error) {
	start := time.Now()
	ctx, span := Tracer.Start(ctx, "infraFacade.GetImport")
	defer func() {
		tracing.EndSpan(span, err)
		status := metric.StatusOK
		if err != nil {
			status = metric.StatusErr
		}
		metric.FacadeRequestedCounter.WithLabelValues("infra", "GetImport", status).Inc()
		metric.FacadeRequestedHistogram.WithLabelValues("infra", "GetImport", status).
			Observe(time.Since(start).Seconds())
	}()
	span.SetAttributes(
		attribute.String("import_id", importID),
		attribute.String("project_id", fmt.Sprint(projectID)),
	)

	proxy, bizErr := f.clusterProxyProvider.GetServerlessHTTPProxy(ctx, proxyEndpoint, nil, nil)
	if bizErr != nil {
		return nil, errors.AddStack(bizErr)
	}
	data, err := proxy.CallK8sResource(ctx, &proxy_pkg.K8sResourceRequest{
		OperateType: "Get",
		APIVersion:  "cloud.pingcap.com/v1",
		Kind:        "Import",
		Namespace:   f.Namespace(projectID),
		Name:        importID,
	})
	if err != nil {
		err = mapK8sResourceError(err, importID)
		if isK8sAPIError(err) {
			return nil, err
		}
		return nil, errors.AddStack(err)
	}

	payload, err := json.Marshal(data)
	if err != nil {
		return nil, errors.AddStack(err)
	}
	i = &cloudv1.Import{}
	if err := json.Unmarshal(payload, i); err != nil {
		return nil, errors.AddStack(err)
	}
	return i, nil
}

func (f *infraFacadeImp) GetExport(
	ctx context.Context,
	proxyEndpoint string,
	projectId uint64,
	exportId string,
) (_ *cloudv1.Export, err error) {
	start := time.Now()
	ctx, span := Tracer.Start(ctx, "infraFacade.GetExport")
	defer func() {
		tracing.EndSpan(span, err)
		status := metric.StatusOK
		if err != nil {
			status = metric.StatusErr
		}
		metric.FacadeRequestedCounter.WithLabelValues("infra", "GetExport", status).Inc()
		metric.FacadeRequestedHistogram.WithLabelValues("infra", "GetExport", status).
			Observe(time.Since(start).Seconds())
	}()
	span.SetAttributes(
		attribute.String("export_id", exportId),
		attribute.String("project_id", fmt.Sprint(projectId)),
	)

	proxy, bizErr := f.clusterProxyProvider.GetServerlessHTTPProxy(ctx, proxyEndpoint, nil, nil)
	if bizErr != nil {
		return nil, errors.AddStack(bizErr)
	}
	data, err := proxy.CallK8sResource(ctx, &proxy_pkg.K8sResourceRequest{
		OperateType: "Get",
		APIVersion:  "cloud.pingcap.com/v1",
		Kind:        "Export",
		Namespace:   f.Namespace(projectId),
		Name:        exportId,
	})
	if err != nil {
		err = mapK8sResourceErrorWithResource(err, exportId, "exports", true)
		if isK8sAPIError(err) {
			return nil, err
		}
		return nil, errors.AddStack(err)
	}

	e := &cloudv1.Export{}
	if err := decodeK8sResourcePayload(data, e); err != nil {
		return nil, err
	}
	return e, nil
}

func (f *infraFacadeImp) CreateExport(ctx context.Context, proxyEndpoint string, e *cloudv1.Export) (err error) {
	start := time.Now()
	ctx, span := Tracer.Start(ctx, "infra.CreateExport")
	defer func() {
		tracing.EndSpan(span, err)
		status := metric.StatusOK
		if err != nil {
			status = metric.StatusErr
		}
		metric.FacadeRequestedCounter.WithLabelValues("infra", "CreateExport", status).Inc()
		metric.FacadeRequestedHistogram.WithLabelValues("infra", "CreateExport", status).
			Observe(time.Since(start).Seconds())
	}()
	span.SetAttributes(attribute.String("export_id", e.Name))

	proxy, bizErr := f.clusterProxyProvider.GetServerlessHTTPProxy(ctx, proxyEndpoint, nil, nil)
	if bizErr != nil {
		return errors.AddStack(bizErr)
	}
	_, err = proxy.CallK8sResource(ctx, &proxy_pkg.K8sResourceRequest{
		OperateType: "Create",
		APIVersion:  "cloud.pingcap.com/v1",
		Kind:        "Export",
		Namespace:   e.Namespace,
		Object:      e,
	})
	if err != nil {
		err = mapK8sResourceErrorWithResource(err, e.Name, "exports", true)
		if apierrors.IsAlreadyExists(err) {
			return nil
		}
		if isK8sAPIError(err) {
			return err
		}
		return errors.AddStack(err)
	}
	log.WithContextL(ctx).Info("export cr created")
	return nil
}

func (f *infraFacadeImp) UpdateExport(ctx context.Context, proxyEndpoint string, e *cloudv1.Export) (err error) {
	start := time.Now()
	ctx, span := Tracer.Start(ctx, "infra.UpdateExport")
	defer func() {
		tracing.EndSpan(span, err)
		status := metric.StatusOK
		if err != nil {
			status = metric.StatusErr
		}
		metric.FacadeRequestedCounter.WithLabelValues("infra", "UpdateExport", status).Inc()
		metric.FacadeRequestedHistogram.WithLabelValues("infra", "UpdateExport", status).
			Observe(time.Since(start).Seconds())
	}()
	span.SetAttributes(attribute.String("export_id", e.Name))

	proxy, bizErr := f.clusterProxyProvider.GetServerlessHTTPProxy(ctx, proxyEndpoint, nil, nil)
	if bizErr != nil {
		return errors.AddStack(bizErr)
	}
	_, err = proxy.CallK8sResource(ctx, &proxy_pkg.K8sResourceRequest{
		OperateType: "Update",
		APIVersion:  "cloud.pingcap.com/v1",
		Kind:        "Export",
		Namespace:   e.Namespace,
		Object:      e,
	})
	if err != nil {
		err = mapK8sResourceErrorWithResource(err, e.Name, "exports", false)
		if isK8sAPIError(err) {
			return err
		}
		return errors.AddStack(err)
	}
	log.WithContextL(ctx).Info("export cr updated")
	return nil
}

func (f *infraFacadeImp) DeleteExport(
	ctx context.Context,
	proxyEndpoint string,
	projectID uint64,
	exportID string,
) (err error) {
	start := time.Now()
	ctx, span := Tracer.Start(ctx, "infra.DeleteExport")
	defer func() {
		tracing.EndSpan(span, err)
		status := metric.StatusOK
		if err != nil {
			status = metric.StatusErr
		}
		metric.FacadeRequestedCounter.WithLabelValues("infra", "DeleteExport", status).Inc()
		metric.FacadeRequestedHistogram.WithLabelValues("infra", "DeleteExport", status).
			Observe(time.Since(start).Seconds())
	}()
	span.SetAttributes(
		attribute.String("project_id", fmt.Sprint(projectID)),
		attribute.String("export_id", exportID),
	)

	proxy, bizErr := f.clusterProxyProvider.GetServerlessHTTPProxy(ctx, proxyEndpoint, nil, nil)
	if bizErr != nil {
		return errors.AddStack(bizErr)
	}
	_, err = proxy.CallK8sResource(ctx, &proxy_pkg.K8sResourceRequest{
		OperateType: "Delete",
		APIVersion:  "cloud.pingcap.com/v1",
		Kind:        "Export",
		Namespace:   f.Namespace(projectID),
		Name:        exportID,
	})
	if err != nil {
		err = mapK8sResourceErrorWithResource(err, exportID, "exports", true)
		if client.IgnoreNotFound(err) != nil {
			if isK8sAPIError(err) {
				return err
			}
			return errors.AddStack(err)
		}
	}
	log.WithContextL(ctx).Info("export cr deleted")
	return nil
}

func (f *infraFacadeImp) GetExportDirectly(
	ctx context.Context,
	proxyEndpoint string,
	projectId uint64,
	exportId string,
) (_ *cloudv1.Export, err error) {
	start := time.Now()
	ctx, span := Tracer.Start(ctx, "infraFacade.GetExportDirectly")
	defer func() {
		tracing.EndSpan(span, err)
		status := metric.StatusOK
		if err != nil {
			status = metric.StatusErr
		}
		metric.FacadeRequestedCounter.WithLabelValues("infra", "GetExportDirectly", status).Inc()
		metric.FacadeRequestedHistogram.WithLabelValues("infra", "GetExportDirectly", status).
			Observe(time.Since(start).Seconds())
	}()
	span.SetAttributes(
		attribute.String("export_id", exportId),
		attribute.String("project_id", fmt.Sprint(projectId)),
	)

	proxy, bizErr := f.clusterProxyProvider.GetServerlessHTTPProxy(ctx, proxyEndpoint, nil, nil)
	if bizErr != nil {
		return nil, errors.AddStack(bizErr)
	}
	data, err := proxy.CallK8sResource(ctx, &proxy_pkg.K8sResourceRequest{
		OperateType: "Get",
		APIVersion:  "cloud.pingcap.com/v1",
		Kind:        "Export",
		Namespace:   f.Namespace(projectId),
		Name:        exportId,
	})
	if err != nil {
		err = mapK8sResourceErrorWithResource(err, exportId, "exports", true)
		if isK8sAPIError(err) {
			return nil, err
		}
		return nil, errors.AddStack(err)
	}

	e := &cloudv1.Export{}
	if err := decodeK8sResourcePayload(data, e); err != nil {
		return nil, err
	}
	return e, nil
}

func (f *infraFacadeImp) listBranchesViaProxy(
	ctx context.Context,
	proxyEndpoint string,
) ([]cloudv1.Branch, error) {
	proxy, bizErr := f.clusterProxyProvider.GetServerlessHTTPProxy(ctx, proxyEndpoint, nil, nil)
	if bizErr != nil {
		return nil, errors.AddStack(bizErr)
	}

	data, err := proxy.CallK8sResource(ctx, &proxy_pkg.K8sResourceRequest{
		OperateType: "List",
		APIVersion:  "cloud.pingcap.com/v1",
		Kind:        "Branch",
		Namespace:   corev1.NamespaceAll,
	})
	if err != nil {
		if isK8sAPIError(err) {
			return nil, err
		}
		return nil, errors.AddStack(err)
	}

	items, err := decodeBranchListPayload(data)
	if err != nil {
		return nil, err
	}

	return items, nil
}

func decodeBranchListPayload(data any) ([]cloudv1.Branch, error) {
	if data == nil {
		return nil, nil
	}

	items := make([]cloudv1.Branch, 0)
	if err := decodeK8sResourcePayload(data, &items); err == nil {
		return items, nil
	}

	return nil, errors.Errorf("invalid branch list payload type: %T", data)
}

func (f *infraFacadeImp) resolveBranchProxyEndpointByBranchID(ctx context.Context, branchID string) (string, error) {
	branch, bizErr := f.branchRepo.GetBranchByID(ctx, branchID)
	if bizErr != nil {
		return "", errors.AddStack(bizErr)
	}

	clusterID, err := strconv.ParseUint(branch.MainBranchID, 10, 64)
	if err != nil {
		return "", errors.AddStack(err)
	}

	proxyEndpoint, err := f.resolveBranchProxyEndpointFromCluster(ctx, clusterID)
	if err == nil {
		return proxyEndpoint, nil
	}
	tilog.Warn(
		ctx,
		"failed to resolve branch proxy endpoint from cluster, fallback to config",
		zap.String("branch_id", branchID),
		zap.String("cluster_id", branch.MainBranchID),
		zap.Error(err),
	)

	proxyEndpoint, fallbackErr := f.resolveBranchProxyEndpointFromConfig(ctx, branch)
	if fallbackErr != nil {
		return "", errors.Annotatef(
			fallbackErr,
			"failed to fallback branch proxy endpoint from config after cluster resolution failed: %v",
			err,
		)
	}
	return proxyEndpoint, nil
}

func (f *infraFacadeImp) resolveBranchProxyEndpointFromCluster(ctx context.Context, clusterID uint64) (string, error) {
	cluster, bizErr := f.mgmtSvc.GetClusterIncludeDeleted(ctx, clusterID)
	if bizErr != nil {
		cluster, bizErr = f.mgmtSvc.GetCluster(ctx, clusterID, tidbmgmtservicev1.ClusterFilter_CLUSTER_VIEW_BASIC)
		if bizErr != nil {
			return "", errors.AddStack(bizErr)
		}
	}

	serverlessCluster := cluster.GetServerlessCluster()
	if serverlessCluster == nil || serverlessCluster.GetCluster() == nil {
		return "", errors.New("serverless cluster is nil")
	}
	proxyEndpoint := serverlessCluster.GetCluster().GetProxyEndpoint()
	if proxyEndpoint == "" {
		return "", errors.New("proxy endpoint is empty")
	}

	return proxyEndpoint, nil
}

func (f *infraFacadeImp) resolveBranchProxyEndpointFromConfig(
	ctx context.Context,
	branch *entity.Branch,
) (string, error) {
	if branch.CloudProviderName == "" || branch.RegionName == "" {
		return "", errors.New("branch provider or region is empty")
	}

	if branch.Kubernetes != "" {
		proxyEndpoint := config.GetClusterProxyUrl(branch.CloudProviderName, branch.RegionName, branch.Kubernetes)
		if proxyEndpoint == "" {
			return "", errors.Errorf(
				"cluster proxy endpoint is empty for provider %s region %s kubernetes %s",
				branch.CloudProviderName,
				branch.RegionName,
				branch.Kubernetes,
			)
		}
		return proxyEndpoint, nil
	}

	foundNonNotFoundErr := false
	for _, regionalCfg := range config.Global().WorkerConfig.RegionalConfig {
		if regionalCfg.Provider != branch.CloudProviderName || regionalCfg.Region != branch.RegionName {
			continue
		}
		for _, k8sCfg := range regionalCfg.K8sConfig {
			if k8sCfg.ClusterProxyUrl == "" {
				continue
			}
			_, err := f.getBranchFromProxy(ctx, k8sCfg.ClusterProxyUrl, branch.ProjectID, branch.ID)
			if err == nil {
				return k8sCfg.ClusterProxyUrl, nil
			}
			if apierrors.IsNotFound(err) {
				continue
			}
			foundNonNotFoundErr = true
			tilog.Warn(
				ctx,
				"failed to probe branch CR from configured cluster proxy endpoint",
				zap.String("branch_id", branch.ID),
				zap.String("cluster_proxy_url", k8sCfg.ClusterProxyUrl),
				zap.Error(err),
			)
		}
	}
	if !foundNonNotFoundErr {
		return "", errBranchCRNotFoundInConfiguredProxies
	}

	return "", errors.Errorf(
		"failed to find branch CR in configured cluster proxy endpoints for provider %s region %s",
		branch.CloudProviderName,
		branch.RegionName,
	)
}

func (f *infraFacadeImp) getBranchFromProxy(
	ctx context.Context,
	proxyEndpoint string,
	projectID uint64,
	branchID string,
) (_ *cloudv1.Branch, err error) {
	proxy, bizErr := f.clusterProxyProvider.GetServerlessHTTPProxy(ctx, proxyEndpoint, nil, nil)
	if bizErr != nil {
		return nil, errors.AddStack(bizErr)
	}
	data, err := proxy.CallK8sResource(ctx, &proxy_pkg.K8sResourceRequest{
		OperateType: "Get",
		APIVersion:  "cloud.pingcap.com/v1",
		Kind:        "Branch",
		Namespace:   f.Namespace(projectID),
		Name:        branchID,
	})
	if err != nil {
		err = mapK8sResourceErrorWithResource(err, branchID, "branches", true)
		if isK8sAPIError(err) {
			return nil, err
		}
		return nil, errors.AddStack(err)
	}

	branch := &cloudv1.Branch{}
	if err := decodeK8sResourcePayload(data, branch); err != nil {
		return nil, err
	}
	return branch, nil
}

func (f *infraFacadeImp) getBranchDirectlyFromProxy(
	ctx context.Context,
	proxyEndpoint string,
	projectID uint64,
	branchID string,
) (_ *cloudv1.Branch, err error) {
	proxy, bizErr := f.clusterProxyProvider.GetServerlessHTTPProxy(ctx, proxyEndpoint, nil, nil)
	if bizErr != nil {
		return nil, errors.AddStack(bizErr)
	}
	data, err := proxy.CallK8sResource(ctx, &proxy_pkg.K8sResourceRequest{
		OperateType: "GetDirectly",
		APIVersion:  "cloud.pingcap.com/v1",
		Kind:        "Branch",
		Namespace:   f.Namespace(projectID),
		Name:        branchID,
	})
	if err != nil {
		err = mapK8sResourceErrorWithResource(err, branchID, "branches", true)
		if isK8sAPIError(err) {
			return nil, err
		}
		return nil, errors.AddStack(err)
	}

	branch := &cloudv1.Branch{}
	if err := decodeK8sResourcePayload(data, branch); err != nil {
		return nil, err
	}
	return branch, nil
}

func (f *infraFacadeImp) createBranchViaProxy(ctx context.Context, proxyEndpoint string, b *cloudv1.Branch) error {
	proxy, bizErr := f.clusterProxyProvider.GetServerlessHTTPProxy(ctx, proxyEndpoint, nil, nil)
	if bizErr != nil {
		return errors.AddStack(bizErr)
	}
	_, err := proxy.CallK8sResource(ctx, &proxy_pkg.K8sResourceRequest{
		OperateType: "Create",
		APIVersion:  "cloud.pingcap.com/v1",
		Kind:        "Branch",
		Namespace:   b.Namespace,
		Object:      b,
	})
	if err != nil {
		err = mapK8sResourceErrorWithResource(err, b.Name, "branches", true)
		if isK8sAPIError(err) {
			return err
		}
		return errors.AddStack(err)
	}
	return nil
}

func (f *infraFacadeImp) updateBranchViaProxy(ctx context.Context, proxyEndpoint string, b *cloudv1.Branch) error {
	proxy, bizErr := f.clusterProxyProvider.GetServerlessHTTPProxy(ctx, proxyEndpoint, nil, nil)
	if bizErr != nil {
		return errors.AddStack(bizErr)
	}
	_, err := proxy.CallK8sResource(ctx, &proxy_pkg.K8sResourceRequest{
		OperateType: "Update",
		APIVersion:  "cloud.pingcap.com/v1",
		Kind:        "Branch",
		Namespace:   b.Namespace,
		Object:      b,
	})
	if err != nil {
		err = mapK8sResourceErrorWithResource(err, b.Name, "branches", false)
		if isK8sAPIError(err) {
			return err
		}
		return errors.AddStack(err)
	}
	return nil
}

func (f *infraFacadeImp) updateBranchStatusViaProxy(
	ctx context.Context,
	proxyEndpoint string,
	b *cloudv1.Branch,
) error {
	proxy, bizErr := f.clusterProxyProvider.GetServerlessHTTPProxy(ctx, proxyEndpoint, nil, nil)
	if bizErr != nil {
		return errors.AddStack(bizErr)
	}
	_, err := proxy.CallK8sResource(ctx, &proxy_pkg.K8sResourceRequest{
		OperateType: "UpdateStatus",
		APIVersion:  "cloud.pingcap.com/v1",
		Kind:        "Branch",
		Namespace:   b.Namespace,
		Object:      b,
		Status:      b.Status,
		Name:        b.Name,
	})
	if err != nil {
		err = mapK8sResourceErrorWithResource(err, b.Name, "branches", false)
		if isK8sAPIError(err) {
			return err
		}
		return errors.AddStack(err)
	}
	return nil
}

func (f *infraFacadeImp) deleteBranchViaProxy(
	ctx context.Context,
	proxyEndpoint string,
	projectID uint64,
	branchID string,
) error {
	proxy, bizErr := f.clusterProxyProvider.GetServerlessHTTPProxy(ctx, proxyEndpoint, nil, nil)
	if bizErr != nil {
		return errors.AddStack(bizErr)
	}
	_, err := proxy.CallK8sResource(ctx, &proxy_pkg.K8sResourceRequest{
		OperateType: "Delete",
		APIVersion:  "cloud.pingcap.com/v1",
		Kind:        "Branch",
		Namespace:   f.Namespace(projectID),
		Name:        branchID,
	})
	if err != nil {
		err = mapK8sResourceErrorWithResource(err, branchID, "branches", true)
		if client.IgnoreNotFound(err) != nil {
			if isK8sAPIError(err) {
				return err
			}
			return errors.AddStack(err)
		}
	}
	return nil
}

func (f *infraFacadeImp) Namespace(projectID uint64) string {
	return f.projectName + "-" + strconv.FormatUint(projectID, 10)
}

func mapK8sResourceError(err error, name string) error {
	return mapK8sResourceErrorWithResource(err, name, "imports", true)
}

func mapK8sResourceErrorWithResource(err error, name, resourceName string, conflictAsAlreadyExists bool) error {
	var respErr *tidbcloud.ResponseError
	if !stderrors.As(err, &respErr) {
		return err
	}
	resource := schema.GroupResource{Group: "cloud.pingcap.com", Resource: resourceName}
	switch respErr.StatusCode {
	case http.StatusNotFound:
		return apierrors.NewNotFound(resource, name)
	case http.StatusConflict:
		if !conflictAsAlreadyExists {
			return apierrors.NewConflict(resource, name, nil)
		}
		return apierrors.NewAlreadyExists(resource, name)
	default:
		return err
	}
}

func decodeK8sResourcePayload(data, target any) error {
	payload, err := json.Marshal(data)
	if err != nil {
		return errors.AddStack(err)
	}
	if err := json.Unmarshal(payload, target); err != nil {
		return errors.AddStack(err)
	}
	return nil
}

func isK8sAPIError(err error) bool {
	_, ok := err.(apierrors.APIStatus)
	return ok
}