package facade_imp

import (
	"context"
	"errors"
	"fmt"
	"regexp"
	"strings"

	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/pkgv2/bizerr"
	proxy_pkg "github.com/tidbcloud/pkgv2/clusterproxy"
	"github.com/tidbcloud/pkgv2/clusterproxy/serverless"
	"github.com/tidbcloud/pkgv2/clusterproxy/serverless/dm"
	"github.com/tidbcloud/pkgv2/clusterproxy/tidbcloud"
	"github.com/tidbcloud/pkgv2/log"
	"github.com/tidbcloud/pkgv2/secret/encrypt"
	"github.com/tidbcloud/serverless-service/internal/config"
	"github.com/tidbcloud/serverless-service/internal/domain/constant"
	"github.com/tidbcloud/serverless-service/internal/domain/facade"
	"github.com/tidbcloud/serverless-service/internal/errorcode"
	tidbmgmtservicev1 "github.com/tidbcloud/tidb-management-service/api/spec/tidb_mgmt_service/v1"
	"go.uber.org/zap"
)

const (
	kafkaInvalidConfigErrorCode       = "ErrKafkaInvalidConfig"
	kafkaCreateTopicErrorCode         = "ErrKafkaCreateTopic"
	kafkaInvalidPartitionNumErrorCode = "ErrKafkaInvalidPartitionNum"
	kafkaTopicExprInvalidErrorCode    = "ErrKafkaTopicExprInvalid"
	tableIneligibleErrorCode          = "ErrTableIneligible"
	apiInvalidParamCode               = "ErrAPIInvalidParam"
	startTsBeforeGC                   = "ErrStartTsBeforeGC"
	kafkaNewProducerErrorCode         = "ErrKafkaNewProducer" // parent err of ErrAvroSchemaAPIError
	changefeedUpdateRefusedErrorCode  = "CDC:ErrChangefeedUpdateRefused"
	changefeedConfigUnchangedError    = "changefeed config is the same with the old one"

	ServerlessCDCAlreadyExistsErrMsg  = "keyspace already exists"
	ServerlessCDCAlreadyDeletedErrMsg = "keyspace not found"

	defaultServerlessService = "serverless-cluster"
)

var CreateChangefeedKKernelErrorCodes = []string{
	kafkaInvalidConfigErrorCode,
	kafkaCreateTopicErrorCode,
	kafkaInvalidPartitionNumErrorCode,
	kafkaTopicExprInvalidErrorCode,
	tableIneligibleErrorCode,
	apiInvalidParamCode,
	startTsBeforeGC,
	kafkaNewProducerErrorCode,
}

var serverlessProxyURLRegex = regexp.MustCompile("^(.*)/proxy/namespace/([^/]*)/cluster/([^/]*)")

type clusterProxyProvider struct{}

func NewClusterProxyProviderFacade() facade.ClusterProxyProviderFacade {
	return &clusterProxyProvider{}
}

func (c *clusterProxyProvider) GetClusterProxy(
	ctx context.Context,
	cluster *tidbmgmtservicev1.Cluster,
	operator *proxy_pkg.ClusterOperator,
) (facade.ClusterProxyFacade, bizerr.BizError) {
	if cluster == nil || cluster.GetServerlessCluster().GetCluster().GetProxyEndpoint() == "" {
		return nil, bizerr.New(errorcode.ErrCodeSvcClusterProxy, "the cluster is not ready for access it via proxy")
	}

	proxy, err := c.getProxy(
		ctx,
		cluster.GetServerlessCluster().GetCluster().GetProxyEndpoint(),
		operator,
		0,
	)
	if err != nil {
		return nil, err
	}
	return proxy, nil
}

func (c *clusterProxyProvider) GetClusterProxyWithTimeout(
	ctx context.Context,
	cluster *tidbmgmtservicev1.Cluster,
	operator *proxy_pkg.ClusterOperator,
	timeoutSec int,
) (facade.ClusterProxyFacade, bizerr.BizError) {
	if cluster == nil || cluster.GetServerlessCluster().GetCluster().GetProxyEndpoint() == "" {
		return nil, bizerr.New(errorcode.ErrCodeSvcClusterProxy, "the cluster is not ready for access it via proxy")
	}

	proxy, err := c.getProxy(
		ctx,
		cluster.GetServerlessCluster().GetCluster().GetProxyEndpoint(),
		operator,
		timeoutSec,
	)
	if err != nil {
		return nil, err
	}
	return proxy, nil
}

func (c *clusterProxyProvider) GetBranchProxy(
	ctx context.Context,
	branchCr *cloudv1.Branch,
	operator *proxy_pkg.ClusterOperator,
) (facade.ClusterProxyFacade, bizerr.BizError) {
	if branchCr == nil || branchCr.Status.Proxy == nil || branchCr.Status.DB == nil {
		return nil, bizerr.New(errorcode.ErrCodeSvcClusterProxy, "the branchCr is not ready for access it via proxy")
	}

	proxy, err := c.getProxy(ctx, branchCr.Status.Proxy.Endpoint, operator, 0)
	if err != nil {
		return nil, err
	}
	return proxy, nil
}

// GetServerlessHTTPProxy creates a lightweight serverless HTTP proxy and will not call PD or fetch
// pdClusterID during initialization. Use it for K8s CR, DM, and CDC keyspace APIs that do not need cluster_id.
func (c *clusterProxyProvider) GetServerlessHTTPProxy(
	ctx context.Context,
	clusterProxyURL string,
	namespace *string,
	service *string,
) (facade.ServerlessClusterProxyFacade, bizerr.BizError) {
	cfg := config.Global()
	baseURL := strings.Split(clusterProxyURL, "/")[0]
	baseURL = fmt.Sprintf("%s/proxy/namespace/fake/cluster/fake", baseURL)
	baseURL, err := buildServerlessHTTPProxyBaseURL(baseURL, namespace, service)
	if err != nil {
		return nil, bizerr.WrapError(err, "format serverless http proxy failed")
	}
	proxy, err := tidbcloud.NewClusterProxyV2(ctx, tidbcloud.ClusterProxyInput{
		PcaARN: cfg.SvcAddrConfig.PCAArn,
		Cfg: tidbcloud.ProxyConfig{
			Auth0Domain:       cfg.Auth0Config.ProxyAuth0Domain,
			Auth0ClientID:     cfg.Auth0Config.ProxyAuth0ClientID,
			Auth0ClientSecret: cfg.Auth0Config.ProxyAuth0ClientSecret,
			Auth0Audience:     cfg.Auth0Config.ProxyAuth0Audience,
			DevMode:           cfg.ServerConfig.Stage != config.StageProd,
		},
		BaseURL: baseURL,
		Timeout: 300,
	})
	if err != nil {
		return nil, bizerr.WrapError(err, "get serverless http proxy failed")
	}
	return &clusterProxy{
		proxy: proxy,
	}, nil
}

func buildServerlessHTTPProxyBaseURL(clusterProxyURL string, namespace, service *string) (string, error) {
	matches := serverlessProxyURLRegex.FindStringSubmatch(clusterProxyURL)
	if len(matches) < 4 {
		return "", fmt.Errorf("invalid BaseUrl %s", clusterProxyURL)
	}

	ns := constant.ServerlessNamespace
	if namespace != nil {
		ns = *namespace
	}
	svc := defaultServerlessService
	if service != nil {
		svc = *service
	}

	return fmt.Sprintf("%s/proxy/namespace/%s/cluster/%s", matches[1], ns, svc), nil
}

// GetServerlessProxy creates a full serverless proxy and will call PD to fetch pdClusterID during initialization.
// Use it only for backup/restore APIs that require cluster_id. Do not use it for K8s CR, DM, or CDC keyspace APIs.
func (c *clusterProxyProvider) GetServerlessProxy(
	ctx context.Context,
	clusterProxyURL string,
) (facade.ServerlessClusterProxyFacade, bizerr.BizError) {
	cfg := config.Global()
	// use fake cluster name and namespace to make NewServerlessClusterProxy happy
	baseURL := strings.Split(clusterProxyURL, "/")[0]
	baseURL = fmt.Sprintf("%s/proxy/namespace/fake/cluster/fake", baseURL)
	proxy, err := tidbcloud.NewServerlessClusterProxy(ctx, tidbcloud.ClusterProxyInput{
		PcaARN: cfg.SvcAddrConfig.PCAArn,
		Cfg: tidbcloud.ProxyConfig{
			Auth0Domain:       cfg.Auth0Config.ProxyAuth0Domain,
			Auth0ClientID:     cfg.Auth0Config.ProxyAuth0ClientID,
			Auth0ClientSecret: cfg.Auth0Config.ProxyAuth0ClientSecret,
			Auth0Audience:     cfg.Auth0Config.ProxyAuth0Audience,
			DevMode:           cfg.ServerConfig.Stage != config.StageProd,
		},
		BaseURL: baseURL,
		Timeout: 60,
	})
	if err != nil {
		return nil, bizerr.WrapError(err, "get serverless cluster proxy failed")
	}
	return &clusterProxy{
		proxy: proxy,
	}, nil
}

func (*clusterProxyProvider) getProxy(
	ctx context.Context,
	endpoint string,
	operator *proxy_pkg.ClusterOperator,
	timeoutSec int,
) (facade.ClusterProxyFacade, bizerr.BizError) {
	timeout := 300
	if timeoutSec > 0 {
		timeout = timeoutSec
	}
	cfg := config.Global()
	proxy, err := tidbcloud.NewClusterProxyV2(ctx, tidbcloud.ClusterProxyInput{
		PcaARN: cfg.SvcAddrConfig.PCAArn,
		Cfg: tidbcloud.ProxyConfig{
			Auth0Domain:       cfg.Auth0Config.ProxyAuth0Domain,
			Auth0ClientID:     cfg.Auth0Config.ProxyAuth0ClientID,
			Auth0ClientSecret: cfg.Auth0Config.ProxyAuth0ClientSecret,
			Auth0Audience:     cfg.Auth0Config.ProxyAuth0Audience,
			DevMode:           cfg.ServerConfig.Stage != config.StageProd,
		},
		BaseURL: endpoint,
		Timeout: timeout,
	})
	if err != nil {
		return nil, bizerr.WrapError(err, "get cluster proxy failed")
	}
	return &clusterProxy{
		proxy:    proxy,
		operator: operator,
	}, nil
}

type clusterProxy struct {
	proxy    proxy_pkg.ClusterProxy
	operator *proxy_pkg.ClusterOperator
}

func (p *clusterProxy) SetOperator(operator *proxy_pkg.ClusterOperator) facade.ClusterProxyFacade {
	out := &clusterProxy{
		proxy:    p.proxy,
		operator: operator,
	}
	return out
}

func (p *clusterProxy) ExecuteSQL(ctx context.Context, sql string) bizerr.BizError {
	result, err := p.proxy.ExecuteSQLWithSqlError(ctx, p.operator, sql)
	if err != nil {
		return bizerr.WrapErrorWithCode(err, errorcode.ErrCodeSvcClusterProxy, "ExecuteSQL")
	}
	return wrapClusterProxyErr(result.SqlError)
}

func (p *clusterProxy) ExecuteMultiSQL(ctx context.Context, sqls []string) bizerr.BizError {
	result, err := p.proxy.ExecuteMultiSQLWithSqlError(ctx, p.operator, sqls)
	if err != nil {
		return bizerr.WrapErrorWithCode(err, errorcode.ErrCodeSvcClusterProxy, "ExecuteMultiSQL")
	}
	return wrapClusterProxyErr(result.SqlError)
}

func (p *clusterProxy) InitCloudAdminPassword(ctx context.Context, cloudAdminPwd string) bizerr.BizError {
	return bizerr.WrapErrorWithCode(
		p.proxy.CreateCloudAdmin(ctx, encrypt.Secret(cloudAdminPwd)),
		errorcode.ErrCodeSvcClusterProxy,
		"InitCloudAdminPassword",
	)
}

func (p *clusterProxy) ChangePassword(ctx context.Context, username, newPwd string) bizerr.BizError {
	return bizerr.WrapErrorWithCode(
		p.proxy.ChangePasswordV2(ctx, p.operator, username, newPwd),
		errorcode.ErrCodeSvcClusterProxy,
		"ChangePassword",
	)
}

func (p *clusterProxy) QuerySQL(ctx context.Context, sql string) ([]byte, bizerr.BizError) {
	result, err := p.proxy.QuerySQLWithSqlError(ctx, p.operator, sql)
	if err != nil {
		return nil, bizerr.WrapErrorWithCode(err, errorcode.ErrCodeSvcClusterProxy, "QuerySQL")
	}
	return result.Result, wrapClusterProxyErr(result.SqlError)
}

func (p *clusterProxy) TestCDCConnection(ctx context.Context, model *proxy_pkg.ConnectionModel) bizerr.BizError {
	err := p.proxy.TestConnection(ctx, model)
	if err != nil {
		switch model.Type {
		case "kafka":
			return formatKafkaTestConnectionErrorContent(ctx, err)
		case "mysql":
			return formatMySQLTestConnectionErrorContent(ctx, err)
		default:
			return bizerr.WrapErrorWithCode(err, errorcode.ErrCodeSvcClusterProxy, "TestConnection")
		}
	}
	return nil
}

// formatKafkaTestConnectionErrorContent
// Currently we got 3 types of error:
// 1. Client.Timeout exceeded while awaiting headers => test connection timeout
// 2. Request fail, 400 => test connection failed
// 3. BizError => test connection failed with specific error message
func formatKafkaTestConnectionErrorContent(ctx context.Context, err error) bizerr.BizError {
	logger := log.WithContextL(ctx)
	logger.Error("formatKafkaTestConnectionErrorContent", zap.Error(err))
	// Get the raw error content
	content := err.Error()
	var bizErr bizerr.BizError
	isFormattedError := false

	// Check if error is already a BizError
	if errors.As(err, &bizErr) {
		content = bizErr.ErrMsg()
		isFormattedError = true
	}

	// Handle timeout errors
	if strings.Contains(content, "Client.Timeout exceeded while awaiting headers") {
		return bizerr.New(errorcode.TestConnectionTimeout, "")
	}

	// Handle test failure errors
	const testFailedPrefix = "Request fail, 400-"
	if strings.HasPrefix(content, testFailedPrefix) {
		content = strings.TrimPrefix(content, testFailedPrefix)
		isFormattedError = true
	}

	if isFormattedError {
		if strings.Contains(content, "none of the kafka bootstrap endpoints can be connected") {
			return bizerr.New(errorcode.TestConnectionTimeout, "")
		}
		// parts := strings.Split(content, "\n")
		// reason := strings.TrimRight(parts[0], ".") + "."
		return bizerr.New(bizerr.ErrCodeParams, "connect kafka failed, reason: %s", content)
	}
	// For unknown errors, return unknown error
	return bizerr.New(bizerr.ErrCodeUnknown, "unknown error: %s", content)
}

func formatMySQLTestConnectionErrorContent(ctx context.Context, err error) bizerr.BizError {
	logger := log.WithContextL(ctx)
	logger.Error("formatMySQLTestConnectionErrorContent", zap.Error(err))
	// Get the raw error content
	content := err.Error()
	var bizErr bizerr.BizError
	isFormattedError := false

	// Check if error is already a BizError
	if errors.As(err, &bizErr) {
		content = bizErr.ErrMsg()
		isFormattedError = true
	}

	// Handle timeout errors
	if strings.Contains(content, "Client.Timeout exceeded while awaiting headers") {
		return bizerr.New(errorcode.TestConnectionTimeout, "")
	}

	// Handle test failure errors
	const testFailedPrefix = "Request fail, 400-"
	if strings.HasPrefix(content, testFailedPrefix) {
		content = strings.TrimPrefix(content, testFailedPrefix)
		isFormattedError = true
	}

	if isFormattedError {
		return bizerr.New(bizerr.ErrCodeParams, "connect mysql failed, reason: %s", content)
	}
	// For unknown errors, return unknown error
	return bizerr.New(bizerr.ErrCodeUnknown, "unknown error: %s", content)
}

func (p *clusterProxy) ListBackups(
	ctx context.Context,
	req serverless.ListBackupRequest,
) (*serverless.ListBackupResponse, bizerr.BizError) {
	result, err := p.proxy.ListServerlessBackups(ctx, req)
	return result, bizerr.WrapErrorWithCode(err, errorcode.ErrCodeSvcClusterProxy, "ListBackups")
}

func (p *clusterProxy) RestoreServerlessCluster(ctx context.Context, req serverless.RestoreRequest) bizerr.BizError {
	err := p.proxy.RestoreServerlessCluster(ctx, req)
	if err != nil {
		return bizerr.WrapErrorWithCode(err, errorcode.ErrCodeSvcClusterProxy, "RestoreServerlessCluster")
	}
	return nil
}

func (p *clusterProxy) GetServerlessRestoreProgress(
	ctx context.Context,
	token string,
	restoreId uint64,
) (*serverless.RestoreProgressResponse, bizerr.BizError) {
	resp, err := p.proxy.GetServerlessRestoreProgress(ctx, token, restoreId)
	if err != nil {
		return nil, bizerr.WrapErrorWithCode(err, errorcode.ErrCodeSvcClusterProxy, "GetServerlessRestoreProgress")
	}
	return resp, nil
}

func (p *clusterProxy) CreateServerlessCDC(ctx context.Context, keyspaceId uint32) bizerr.BizError {
	err := p.proxy.CreateServerlessCDC(ctx, keyspaceId)
	if err != nil {
		if strings.Contains(err.Error(), ServerlessCDCAlreadyExistsErrMsg) {
			return nil
		}
		return bizerr.WrapErrorWithCode(err, errorcode.ErrCodeSvcClusterProxy, "CreateServerlessCDC")
	}
	return nil
}

func (p *clusterProxy) DeleteServerlessCDC(ctx context.Context, keyspaceId uint32) bizerr.BizError {
	err := p.proxy.DeleteServerlessCDC(ctx, keyspaceId)
	if err != nil {
		// already deleted
		if strings.Contains(err.Error(), ServerlessCDCAlreadyDeletedErrMsg) {
			return nil
		}
		return bizerr.WrapErrorWithCode(err, errorcode.ErrCodeSvcClusterProxy, "DeleteServerlessCDC")
	}
	return nil
}

func (p *clusterProxy) CreateServerlessChangefeed(
	ctx context.Context,
	keyspaceId uint32,
	req *serverless.ChangefeedConfig,
) (*serverless.Changefeed, bizerr.BizError) {
	cf, err := p.proxy.CreateServerlessChangefeed(ctx, keyspaceId, req)
	if err != nil {
		return nil, formatCreateServerlessChangefeedError(ctx, err, req)
	}
	return cf, nil
}

func formatCreateServerlessChangefeedError(
	ctx context.Context,
	err error,
	req *serverless.ChangefeedConfig,
) bizerr.BizError {
	logger := log.WithContextL(ctx)
	logger.Error("format CreateServerlessChangefeedError", zap.Any("req", req), zap.Error(err))
	for _, code := range CreateChangefeedKKernelErrorCodes {
		if strings.Contains(err.Error(), code) {
			errMsg := extractCDCErrorMsg(err.Error())
			return bizerr.New(errorcode.CreateChangefeedUnretryableErr, "create changefeed failed: %s", errMsg)
		}
	}
	return bizerr.WrapErrorWithCode(err, errorcode.ErrCodeSvcClusterProxy, "CreateServerlessChangefeed")
}

func extractCDCErrorMsg(errMsg string) string {
	re := regexp.MustCompile(`"error_msg"\s*:\s*"([^"]+)"`)
	matches := re.FindStringSubmatch(errMsg)
	if len(matches) > 1 {
		return matches[1]
	}
	return "unknown error"
}

func (p *clusterProxy) GetServerlessChangefeed(
	ctx context.Context,
	keyspaceId uint32,
	changefeedId string,
) (*serverless.Changefeed, bizerr.BizError) {
	cf, err := p.proxy.GetServerlessChangefeed(ctx, keyspaceId, changefeedId)
	if err != nil {
		return nil, bizerr.WrapErrorWithCode(err, errorcode.ErrCodeSvcClusterProxy, "GetServerlessChangefeed")
	}
	return cf, nil
}

func (p *clusterProxy) ListServerlessChangefeeds(
	ctx context.Context,
	keyspaceId uint32,
	state *serverless.ChangefeedState,
) (*serverless.ListChangefeedsResponse, bizerr.BizError) {
	list, err := p.proxy.ListServerlessChangefeeds(ctx, keyspaceId, state)
	if err != nil {
		return nil, bizerr.WrapErrorWithCode(err, errorcode.ErrCodeSvcClusterProxy, "ListServerlessChangefeeds")
	}
	return list, nil
}

func (p *clusterProxy) DeleteServerlessChangefeed(
	ctx context.Context,
	keyspaceId uint32,
	changefeedId string,
) bizerr.BizError {
	err := p.proxy.DeleteServerlessChangefeed(ctx, keyspaceId, changefeedId)
	if err != nil {
		// The changefeed/CDC component is already gone or never deployed, so the
		// delete is effectively a no-op. These are treated as success for
		// idempotency, but we log them so the swallowed error stays observable
		// (the string matching is fragile and worth alerting on if it shifts).
		if strings.Contains(err.Error(), "No CDC server found") ||
			strings.Contains(err.Error(), "502-dial tcp") {
			log.WithContextL(ctx).Warn("DeleteServerlessChangefeed swallowed as no-op",
				zap.Uint32("keyspaceId", keyspaceId),
				zap.String("changefeedId", changefeedId),
				zap.Error(err))
			return nil
		}
		return bizerr.WrapErrorWithCode(err, errorcode.ErrCodeSvcClusterProxy, "DeleteServerlessChangefeed")
	}
	return nil
}

func (p *clusterProxy) UpdateServerlessChangefeed(
	ctx context.Context,
	keyspaceId uint32,
	changefeedId string,
	req *serverless.ChangefeedConfig,
) (*serverless.Changefeed, bizerr.BizError) {
	cf, err := p.proxy.UpdateServerlessChangefeed(ctx, keyspaceId, changefeedId, req)
	if err != nil {
		if isNoopUpdateServerlessChangefeedError(err) {
			log.WithContextL(ctx).Info(
				"serverless changefeed config is unchanged, skip update error",
				zap.String("changefeedID", changefeedId),
			)
			return nil, nil
		}
		return nil, formatUpdateServerlessChangefeedError(ctx, err, req)
	}
	return cf, nil
}

func isNoopUpdateServerlessChangefeedError(err error) bool {
	if err == nil {
		return false
	}
	errMsg := err.Error()
	return strings.Contains(errMsg, changefeedUpdateRefusedErrorCode) &&
		strings.Contains(errMsg, changefeedConfigUnchangedError)
}

func formatUpdateServerlessChangefeedError(
	ctx context.Context,
	err error,
	req *serverless.ChangefeedConfig,
) bizerr.BizError {
	logger := log.WithContextL(ctx)
	logger.Error("format UpdateServerlessChangefeedError", zap.Any("req", req), zap.Error(err))
	for _, code := range CreateChangefeedKKernelErrorCodes {
		if strings.Contains(err.Error(), code) {
			errMsg := extractCDCErrorMsg(err.Error())
			return bizerr.New(bizerr.ErrCodeParams, "update changefeed failed: %s", errMsg)
		}
	}
	return bizerr.WrapErrorWithCode(err, errorcode.ErrCodeSvcClusterProxy, "UpdateServerlessChangefeed")
}

func (p *clusterProxy) PauseServerlessChangefeed(
	ctx context.Context,
	keyspaceId uint32,
	changefeedId string,
) bizerr.BizError {
	err := p.proxy.PauseServerlessChangefeed(ctx, keyspaceId, changefeedId)
	if err != nil {
		return bizerr.WrapErrorWithCode(err, errorcode.ErrCodeSvcClusterProxy, "PauseServerlessChangefeed")
	}
	return nil
}

func (p *clusterProxy) ResumeServerlessChangefeed(
	ctx context.Context,
	keyspaceId uint32,
	changefeedId string,
	req *serverless.ResumeChangefeedConfig,
) bizerr.BizError {
	// missing this will cause "-d null" when request cdc, which will cause panic
	if req == nil {
		req = &serverless.ResumeChangefeedConfig{}
	}
	err := p.proxy.ResumeServerlessChangefeed(ctx, keyspaceId, changefeedId, req)
	if err != nil {
		return bizerr.WrapErrorWithCode(err, errorcode.ErrCodeSvcClusterProxy, "ResumeServerlessChangefeed")
	}
	return nil
}

func (p *clusterProxy) GetServerlessGcSafePoint(ctx context.Context, keyspaceId uint32) (uint64, bizerr.BizError) {
	resp, err := p.proxy.GetServerlessGCSafePoint(ctx, keyspaceId)
	if err != nil {
		return 0, bizerr.WrapErrorWithCode(err, errorcode.ErrCodeSvcClusterProxy, "GetServerlessGcSafePoint")
	}
	if resp == nil {
		return 0, bizerr.New(bizerr.ErrCodeUnknown, "no gc safe point body in response")
	}
	// `resp.SafePoint == 0` is a valid case, which means the GC safe point is not set yet (e.g. the cluster is just created).
	return resp.SafePoint, nil
}

func (p *clusterProxy) OpenWebsocket(ctx context.Context) (proxy_pkg.WebsocketConn, bizerr.BizError) {
	ws, err := p.proxy.OpenWebsocket(ctx, p.operator)
	if err != nil {
		return nil, bizerr.WrapErrorWithCode(err, errorcode.ErrCodeSvcClusterProxy, "OpenWebsocket")
	}
	return ws, nil
}

func (p *clusterProxy) GetProxyInfo(ctx context.Context) (proxy_pkg.ClusterProxy, *proxy_pkg.ClusterOperator) {
	return p.proxy, p.operator
}

func (p *clusterProxy) CreateServerlessDMSource(ctx context.Context, req *dm.CreateSourceRequest) (*dm.Source, error) {
	source, err := p.proxy.CreateServerlessDMSource(ctx, req)
	if err != nil {
		return nil, err
	}
	return source, nil
}

func (p *clusterProxy) GetServerlessDMSource(ctx context.Context, name string) (*dm.Source, error) {
	source, err := p.proxy.GetServerlessDMSource(ctx, name)
	if err != nil {
		return nil, err
	}
	return source, nil
}

func (p *clusterProxy) DeleteServerlessDMSource(ctx context.Context, name string) error {
	if err := p.proxy.DeleteServerlessDMSource(ctx, name); err != nil {
		return err
	}
	return nil
}

// --- DM Task ---
func (p *clusterProxy) CreateServerlessDMTask(
	ctx context.Context,
	req *dm.CreateTaskRequest,
) (*dm.OperateTaskResponse, error) {
	task, err := p.proxy.CreateServerlessDMTask(ctx, req)
	if err != nil {
		return nil, err
	}
	return task, nil
}

func (p *clusterProxy) DeleteServerlessDMTask(ctx context.Context, name string) error {
	if err := p.proxy.DeleteServerlessDMTask(ctx, name); err != nil {
		return err
	}
	return nil
}

func (p *clusterProxy) StartServerlessDMTask(ctx context.Context, name string, req *dm.StartTaskRequest) error {
	if err := p.proxy.StartServerlessDMTask(ctx, name, req); err != nil {
		return err
	}
	return nil
}

func (p *clusterProxy) StopServerlessDMTask(ctx context.Context, name string, req *dm.StopTaskRequest) error {
	if err := p.proxy.StopServerlessDMTask(ctx, name, req); err != nil {
		return err
	}
	return nil
}

func (p *clusterProxy) QueryServerlessDMTaskStatus(
	ctx context.Context,
	name string,
	sourceNameList []string,
) (*dm.GetTaskStatusResponse, error) {
	resp, err := p.proxy.QueryServerlessDMTaskStatus(ctx, name, sourceNameList)
	if err != nil {
		return nil, err
	}
	return resp, nil
}

func (p *clusterProxy) ListServerlessDMWorkers(ctx context.Context) (*dm.GetClusterWorkerListResponse, error) {
	resp, err := p.proxy.ListServerlessDMWorkers(ctx)
	if err != nil {
		return nil, err
	}
	return resp, nil
}

// GetSourceTables returns the tables available for a DM source.
func (p *clusterProxy) GetSourceTables(
	ctx context.Context,
	model *proxy_pkg.ConnectionModel,
) ([]proxy_pkg.TableInfo, error) {
	return p.proxy.GetSourceTables(ctx, model)
}

// TestConnection tests connectivity for a given DM source or other connection request.
func (p *clusterProxy) TestConnection(ctx context.Context, model *proxy_pkg.ConnectionModel) error {
	return p.proxy.TestConnection(ctx, model)
}

func (p *clusterProxy) CallK8sResource(ctx context.Context, req *proxy_pkg.K8sResourceRequest) (any, error) {
	return p.proxy.CallK8sResource(ctx, *req)
}

func wrapClusterProxyErr(sqlError proxy_pkg.SqlError) bizerr.BizError {
	if sqlError.ErrNumber == 0 {
		return nil
	}
	return facade.NewClusterProxyErr(sqlError.ErrNumber, sqlError.ErrMessage)
}