package service

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net/http"
	"regexp"
	"strconv"
	"strings"

	"dataservices/commons/config"
	"dataservices/commons/constants"
	comerrs "dataservices/commons/errs"
	"dataservices/commons/grpc"
	http2 "dataservices/commons/http"
	"dataservices/commons/metrics"
	"dataservices/commons/utils"
	"dataservices/dbhubsdk"
	"dataservices/ddb"
	dto "dataservices/internal/dto/dataapi"
	dto2 "dataservices/internal/dto/dataapi_gateway"
	"dataservices/internal/entity"
	"dataservices/internal/repository"
	"dataservices/internal/repository_ddb"

	"github.com/pingcap/tidb/pkg/parser"
	"github.com/tidbcloud/tiinf/errs"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.uber.org/zap"
)

var (
	sqlArgsRegexp     = regexp.MustCompile(constants.SqlArgsRegexp)
	sqlCommentsRegexp = regexp.MustCompile(constants.SqlCommentsRegexp)
)

func CreateOnlineEndpoint(ctx context.Context, endpointId uint64, createdBy string) (version string, err error) {

	defer recordCounterMetric(metrics.EndpointOnlineCounter, err)

	endpoint, err := repository.GetEndpoint(endpointId)
	if err != nil {
		return "", err
	}
	if err := checkEndpointCanRun(ctx, endpoint); err != nil {
		return "", err
	}
	publishedVersion, err := getLastPublishedVersion(endpointId)
	if err != nil {
		return "", err
	}
	version = getReleaseVersion(publishedVersion)

	err = repository.Transaction(ctx, func(ctx context.Context) error {
		if err = repository.CreateOnlineEndpointWithTX(ctx, endpoint, version, createdBy); err != nil {
			return err
		}

		// push sync deploy endpoint event
		if err = publishEndpointEvent(ctx, endpointId, createdBy, dto.EndpointActionDeploy); err != nil {
			return err
		}

		return nil
	})
	if err != nil {
		return "", err
	}
	return version, nil
}

func OfflineEndpoint(ctx context.Context, endpointId uint64, createdBy string) (err error) {

	defer recordCounterMetric(metrics.EndpointOfflineCounter, err)

	_, err = repository.GetPublishedEndpoint(endpointId)
	if err != nil {
		if errors.Is(err, errs.ErrNotFound) {
			return errs.NewBadRequest("Endpoint is not online", err)
		}
		return err
	}

	err = repository.Transaction(ctx, func(ctx context.Context) error {
		if err = repository.OfflineEndpointWithTX(ctx, endpointId, createdBy); err != nil {
			return err
		}

		// push sync undeploy endpoint event
		if err = publishEndpointEvent(ctx, endpointId, createdBy, dto.EndpointActionUndeploy); err != nil {
			return err
		}
		return nil
	})

	return err
}

func ListEndpointPublishLog(ctx context.Context, endpointId uint64) ([]*entity.EndpointPublishVersions, error) {

	publishedVersion, err := getCurPublishedVersion(endpointId)
	if err != nil {
		return nil, err
	}

	res, err := repository.ListEndpointPublishVersions(endpointId)
	if err != nil {
		return nil, err
	}
	result := make([]*entity.EndpointPublishVersions, 0, len(res))
	versionExists := make(map[string]bool, len(res))
	for _, versionInfo := range res {
		if versionExists[versionInfo.Version] {
			result[len(result)-1].CreatedAt = versionInfo.CreatedAt
			continue
		}
		versionExists[versionInfo.Version] = true
		if versionInfo.Version == publishedVersion {
			versionInfo.IsOnline = 1
		} else {
			if versionInfo.IsOnline == 1 && len(result) > 0 {
				versionInfo.UpdatedAt = result[len(result)-1].CreatedAt
			}
			versionInfo.IsOnline = 0
		}
		result = append(result, versionInfo)
	}
	return result, nil
}

func CodeExample(ctx context.Context, endpointId uint64, in entity.CodeExampleInput) (res *dto.CodeExampleOut, err error) {
	defer recordCounterMetric(metrics.EndpointCodeExampleCounter, err)

	draftEndpoint, err := repository.GetEndpoint(endpointId)
	if err != nil {
		return nil, err
	}
	publishedEndpoint, err := repository.GetPublishedEndpoint(endpointId)
	if err != nil && !errors.Is(err, errs.ErrNotFound) {
		return nil, err
	}
	app, err := repository.GetDataApp(draftEndpoint.DataappId)
	if err != nil {
		return nil, err
	}
	regionSuccessfulDomainMap, err := GetRegionSuccessfulDomainsMap(ctx, app.ID)
	if err != nil {
		return nil, err
	}

	in.FillDefaultVal = true
	res = new(dto.CodeExampleOut)
	if draftEndpoint.Type != entity.EndpointTypeChat2Data &&
		draftEndpoint.Type != entity.EndpointTypeChat2EDA &&
		draftEndpoint.Type != entity.EndpointTypeSystemData {
		in.ClusterID = strconv.FormatUint(draftEndpoint.ClusterId, 10)
		res.Test = getEndpointCodeExample(ctx, app, draftEndpoint, true, in, regionSuccessfulDomainMap)
	}
	if publishedEndpoint != nil {
		if in.ClusterID == "" {
			in.ClusterID = strconv.FormatUint(draftEndpoint.ClusterId, 10)
		}
		res.Prod = getEndpointCodeExample(ctx, app, publishedEndpoint.ToEndpoint(), false, in, regionSuccessfulDomainMap)
	}
	return res, nil
}

func getEndpointCodeExample(ctx context.Context, app *entity.DataApp, endpoint *entity.Endpoint, isDraft bool, in entity.CodeExampleInput, regionSuccessfulDomainsMap map[string]*entity.CustomDomain) []dto.CodeExample {
	tilog.Info(ctx, "getEndpointCodeExample", zap.Any("in", in))

	path := endpoint.Endpoint
	method := strings.ToUpper(endpoint.Method)
	endpointType := endpoint.Type
	batch := endpoint.IsBatch()

	var args []entity.EndpointArgs
	if endpoint.Args != "" {
		if err := utils.FromJson(endpoint.Args, &args); err != nil {
			tilog.Error(ctx, "getEndpointCodeExample failed to parse endpoint arguments", zap.Uint64("endpoint_id", endpoint.ID), zap.Error(err))
			return nil
		}
	}

	var (
		uri       string
		auth      string
		body      string
		clusterID = in.ClusterID
		database  = in.Database
		ak        = in.AK
		sk        = in.SK
	)

	if path == "" {
		path = "/<endpoint>"
	}
	var region string
	if clusterID == "" || clusterID == "0" {
		clusterID = "<your cluster id>"
	} else {
		clusterIDInt, err := strconv.ParseUint(clusterID, 10, 64)
		if err != nil {
			tilog.Error(ctx, "getEndpointCodeExample failed to parse cluster_id", zap.String("cluster_id", clusterID), zap.Error(err))
			return nil
		}
		if !config.IsProd() {
			region = constants.DefaultAWSRegion
		} else {
			clusterInfo, err := grpc.GetClusterWithoutPermCheck(context.TODO(), clusterIDInt)
			if err != nil {
				tilog.Error(ctx, "getEndpointCodeExample failed to get cluster info", zap.String("cluster_id", clusterID), zap.Error(err))
				return nil
			}
			region = clusterInfo.RegionName
		}
	}
	if database == "" {
		database = "<your database>"
	}
	if ak == "" {
		ak = "${PUBLIC_KEY}"
	}
	if sk == "" {
		sk = "${PRIVATE_KEY}"
	}
	uri = fmt.Sprintf("%s%s", GenAPPServiceUrl(app, region, regionSuccessfulDomainsMap), path)
	if region != "" {
		if in.AuthType == "basic" {
			auth = fmt.Sprintf(`--user %s:%s`, ak, sk)
		} else {
			auth = fmt.Sprintf(`--digest --user %s:%s`, ak, sk)
		}
	} else {
		if in.AuthType == "basic" {
			auth = fmt.Sprintf(`--location-trusted --user %s:%s`, ak, sk)
		} else {
			auth = fmt.Sprintf(`--location-trusted --digest --user %s:%s`, ak, sk)
		}
	}

	params := map[string]map[string]string{
		"system": {
			"cluster_id": clusterID,
			"database":   database,
			"org_id":     strconv.FormatUint(app.OrgId, 10),
		},
		"user": {},
	}
	if endpoint.IsSystemDataEndpoint() {
		playgroundContent, err := loadUserEndpointPlayground(ctx, endpoint.DataappId, endpoint.ID, in.Operator)
		if err != nil {
			tilog.Warn(ctx, "loadUserEndpointPlayground failed", zap.Uint64("dataapp_id", endpoint.DataappId), zap.Uint64("endpoint_id", endpoint.ID), zap.Error(err))
		} else if playgroundContent.SqlTemplate != "" {
			params["system"]["sql"] = convertSqlToJsonField(playgroundContent.SqlTemplate)
		}
	}
	if in.FillDefaultVal {
		for _, arg := range args {
			params["user"][arg.Name] = arg.Default
		}
	}
	query, body, headers := convertEndpointRequest(endpointType, method, path, args, batch, params)
	if query != "" {
		uri += "?" + query
	}
	uri = paginationConvert(ctx, endpoint, uri)

	if isDraft {
		headers["endpoint-type"] = "draft"
	}
	if in.Language == "js" {
		jsCode := fmt.Sprintf(`const { request } = require('urllib');`)
		jsCode += fmt.Sprintf(`\nconst url = '%s'`, uri)
		jsCode += fmt.Sprintf(`\nconst endpoint = async () => {\n	const { err, data, res } = await request( url,		\n		{ `)
		if in.AuthType == "basic" {
			jsCode += fmt.Sprintf("\\n				auth: `%s:%s`,", ak, sk)
		} else {
			jsCode += fmt.Sprintf("\\n				digestAuth: `%s:%s`,", ak, sk)
		}
		header := ""
		for k, v := range headers {
			header += fmt.Sprintf(`\n					'%s': '%s',`, k, v)
		}
		if header != "" {
			jsCode += fmt.Sprintf(`\n				headers: { %s \n				},`, header)
		}
		jsCode += fmt.Sprintf(`\n				method: '%s',`, method)
		if body != "" {
			jsCode += fmt.Sprintf(`\n				data: %s`, body)
		}

		jsCode += fmt.Sprintf(`\n			}\n		)\n	console.log(data.toString()) \n}\nendpoint()`)

		return []dto.CodeExample{
			{
				Type: "js",
				Code: jsCode,
			},
		}
	} else {
		curlCmd := fmt.Sprintf(`curl %s --request %s '%s'`, auth, method, uri)
		for k, v := range headers {
			curlCmd += fmt.Sprintf(`\\n --header '%s: %s'`, k, v)
		}
		if body != "" {
			curlCmd += fmt.Sprintf(`\\n --data-raw '%s'`, body)
		}

		return []dto.CodeExample{
			{
				Type: "curl",
				Code: curlCmd,
			},
		}
	}

}

func paginationConvert(ctx context.Context, endpoint *entity.Endpoint, uri string) string {

	if strings.ToUpper(endpoint.Method) != http.MethodGet {
		return uri
	}

	var settings entity.EndpointSettings
	if len(endpoint.Settings) > 0 {
		json.Unmarshal([]byte(endpoint.Settings), &settings)
	}

	if settings.EnablePagination == 0 {
		return uri
	}

	sql := sqlArgsRegexp.ReplaceAllString(endpoint.SqlTemplate, "?")
	stmts, _, err := parser.New().ParseSQL(sql)
	if err != nil {
		tilog.Error(ctx, "Parsing sql failed, Please check if sql syntax is correct", zap.Error(err))
		return uri
	} else if len(stmts) == 0 ||
		dbhubsdk.GetStmtType(ctx, stmts[len(stmts)-1]) != dbhubsdk.SELECT {
		return uri
	}

	pageSize := settings.RowsLimit
	if pageSize > 20 {
		pageSize = 20
	}
	splitChar := "?"
	if strings.Contains(uri, "?") {
		splitChar = "&"
	}
	uri += splitChar + fmt.Sprintf("%s=1&%s=%d", constants.EndpointPageParam, constants.EndpointPageSizeParam, pageSize)
	return uri
}

func formatEDACodeExampleQuery(method, path string, params map[string]map[string]string, clusterID, database string) string {
	genKey := func(method, path string) string {
		return fmt.Sprintf("%s:%s", method, path)
	}

	mapper := map[string]string{
		genKey(http.MethodGet, endpointPathListDataSummaries):  fmt.Sprintf("cluster_id=%s&database=%s", clusterID, database),
		genKey(http.MethodGet, endpointPathListDataSummaryV2):  fmt.Sprintf("cluster_id=%s&database=%s", clusterID, database),
		genKey(http.MethodGet, endpointPathListKnowledgeBases): fmt.Sprintf("cluster_id=%s&database=%s", clusterID, database),
		genKey(http.MethodGet, endpointPathListSessions):       fmt.Sprintf("page=1&per_page=200"),
	}

	return mapper[genKey(method, path)]
}

func formatEDACodeExamplePayload(method, path string, params map[string]map[string]string, clusterID, database string) string {
	ctx := context.TODO()
	tilog.Info(ctx, "format eda code example", zap.String("method", method), zap.String("path", path), zap.Any("params", params))

	genKey := func(method, path string) string {
		return fmt.Sprintf("%s:%s", method, path)
	}

	mapper := map[string]string{
		genKey(http.MethodPost, endpointPathCreateDataSummaries): fmt.Sprintf(`{
    "cluster_id": "%s",
    "database": "%s",
    "description": "<Your data summary description>",
    "reuse": false,
    "default": false
}`, clusterID, database),
		genKey(http.MethodPost, endpointPathCreateDataSummariesV2): fmt.Sprintf(`{
    "cluster_id": "%s",
    "database": "%s",
    "description": "<Your data summary description>",
    "reuse": false
}`, clusterID, database),
		genKey(http.MethodPut, endpointPathUpdateDataSummary): `{
	"description": "<Your data summary description>",
	"default": false
}`,
		genKey(http.MethodPut, endpointPathUpdateTableDescription): `{
	"description": "<Your table description>"
}`,
		genKey(http.MethodPut, endpointPathUpdateColumnDescription): `{
    "columns":[
       {
           "name": "<Your column name>",
           "description": "<Your column description>"
       }
   ]
}`,
		genKey(http.MethodPost, endpointPathCreateKnowledgeBase): fmt.Sprintf(`{
    "cluster_id": "%s",
    "database": "%s",
    "description": "<Your knowledge base description>"
}`, clusterID, database),
		genKey(http.MethodPut, endpointPathUpdateKnowledgeBase): `{
    "description": "<Your knowledge base description>",
    "default": true
}`,
		genKey(http.MethodPut, endpointPathUpdateSessionContext): `{
    "name": "<New session name>"
}`,
		genKey(http.MethodPut, endpointPathResetSessionContext): `{}`,
		genKey(http.MethodPost, endpointPathCreateDataItems): `{
    "type": "few_shot",
    "meta_data": {
        "involved_tables": ["<Your table name>"]
    },
    "raw_data": {
        "question": "Count users",
        "answer": "SELECT COUNT(*) FROM users"
    }
}`,
		genKey(http.MethodPut, endpointPathUpdateDataItems): `{
    "meta_data": {
        "involved_tables": ["<Your table name>"]
    },
    "raw_data": {
        "question": "Count users",
        "answer": "SELECT COUNT(*) FROM users"
    }
}`,
		genKey(http.MethodPost, endpointPathCreateSessionContext): fmt.Sprintf(`{
    "cluster_id": "%s",
    "database": "%s",
    "name": "<Your session name>"
}`, clusterID, database),
		genKey(http.MethodPost, endpointPathSessionChat): `{
    "question": "<Your question>",
    "feedback_answer_id": "<Optional, feedback answer id>",
    "feedback_task_id": "<Optional, feedback task id>",
    "sql_generate_mode": "<Optional, sql generate mode, by default is direct, choices: direct/auto_breakdown>"
}`,
		genKey(http.MethodPost, endpointPathSuggestQuestions): fmt.Sprintf(`{
    "cluster_id": "%s",
    "database": "%s"
}`, clusterID, database),
		genKey(http.MethodPost, endpointPathChat2Data): fmt.Sprintf(`{
    "cluster_id": "%s",
    "database": "%s",
    "data_summary_id": "<Optional, use the given data summary id to generate data, only works when 'cluster_id' and 'database' are not provided>",
    "question": "<Your question to generate data>",
    "sql_generate_mode": "<Optional, sql generate mode, by default is direct, choices: direct/auto_breakdown>"
}`, clusterID, database),
		genKey(http.MethodPost, endpointPathChat2DataV2): fmt.Sprintf(`{
	"cluster_id": "%s",
	"database": "%s",
	"question": "<Your question to generate data>"
}`, clusterID, database),
		genKey(http.MethodPost, endpointPathRefineSQL): fmt.Sprintf(`{
	"cluster_id": "%s",
	"database": "%s",
	"sql": "<The SQL you want to refine. Required.>",
    "feedback": "<The feedback for the SQL. Optional.>"
}`, clusterID, database),
		genKey(http.MethodPost, endpointPathExecuteSQL): `{
	"session_context_id": <Session context id returned in chat2data/chat2sql>,
	"task_id": "<Task ID returned in chat2data/chat2sql>",
	"question_id": "<Question ID returned in chat2data/chat2sql>"
}`,
		genKey(http.MethodPost, endpointPathExecuteQuery): fmt.Sprintf(`{
	"cluster_id": "%s",
	"database": "%s",
	"sql": "<Your SQL to execute>"
}`, clusterID, database),
		genKey(http.MethodPost, endpointPathUserFeedback): `{
	"session_context_id": <Session context id returned in chat2data>,
	"question_id": "<Question ID returned in chat2data>",
	"feedback": "<Your feedback, like or dislike>",
	"comment": "<Your comment, which is optional>"
}`,
	}

	return mapper[genKey(method, path)]
}

func convertEndpointRequest(endpointType, method, path string, args []entity.EndpointArgs, batch bool, params map[string]map[string]string) (query, body string, headers map[string]string) {
	tilog.Info(context.TODO(), "code example, convert endpoint request", zap.String("endpointType", endpointType), zap.String("path", path), zap.Any("params", params))

	clusterID := params["system"]["cluster_id"]
	database := params["system"]["database"]
	sql := params["system"]["sql"]
	if clusterID == "" {
		clusterID = "<your cluster id>"
	}
	if database == "" {
		database = "<your database>"
	}
	if sql == "" {
		sql = "<your SQL to execute>"
	}

	headers = make(map[string]string)
	switch endpointType {
	case entity.EndpointTypeSystemData:
		headers["content-type"] = "application/json"
		body = fmt.Sprintf(`{
    "cluster_id": "%s",
    "database": "%s",
    "sql": "%s"
}`, clusterID, database, sql)
	case entity.EndpointTypeChat2Data:
		headers["content-type"] = "application/json"
		body = fmt.Sprintf(`{
    "cluster_id": "%s",
    "database": "%s",
    "tables": ["<your table name, optional>"],
    "instruction": "<your instruction>"
}`, clusterID, database)
	case entity.EndpointTypeChat2EDA:
		headers["content-type"] = "application/json"
		body = formatEDACodeExamplePayload(method, path, params, clusterID, database)
		query = formatEDACodeExampleQuery(method, path, params, clusterID, database)
	case entity.EndpointTypeChat2Sql:
		fallthrough
	default:
		if method == http.MethodGet || method == http.MethodDelete {
			query = buildQueryFormArgs(args, params)
		} else {
			headers["content-type"] = "application/json"
			body = buildJsonFormArgs(args, params, batch)
		}
	}
	return
}

func buildQueryFormArgs(args []entity.EndpointArgs, params map[string]map[string]string) string {
	query := make([]string, 0, len(args))
	for _, v := range args {
		if v.IsPathParameter {
			continue
		}
		name, value := v.Name, fmt.Sprintf("${%s}", v.Name)
		if params["user"][name] != "" {
			value = params["user"][name]
		}
		query = append(query, name+"="+value)
	}
	return strings.Join(query, "&")
}

func buildJsonFormArgs(args []entity.EndpointArgs, params map[string]map[string]string, batch bool) string {
	bodyParams := make(map[string]string, len(args))
	for _, v := range args {
		if v.IsPathParameter {
			continue
		}
		name, value := v.Name, fmt.Sprintf("${%s}", v.Name)
		if params["user"][name] != "" {
			value = params["user"][name]
		}
		bodyParams[name] = value
	}
	var bodyBytes []byte
	if !batch {
		bodyBytes, _ = json.MarshalIndent(bodyParams, "", "  ")
	} else {
		batchParam := map[string][]map[string]string{
			"items": {bodyParams},
		}
		bodyBytes, _ = json.MarshalIndent(batchParam, "", "  ")
	}
	return string(bodyBytes)
}

func PublishTest(ctx context.Context, input entity.TestEndpointParamV2) (res *dto.EndpointResult, err error) {
	defer recordCounterMetric(metrics.EndpointTestCounter, err)

	endpoint, err := repository.GetEndpoint(input.EndpointId)
	if err != nil {
		return nil, err
	}

	argValues := make([]map[string]any, 0, len(input.Args))
	for _, v := range input.Args {
		_argValues := make(map[string]any, len(v))
		for _, arg := range v {
			_argValues[arg.Name] = arg.Value
			if endpoint.IsSystemDataEndpoint() {
				switch arg.Name {
				case "cluster_id":
					clusterIDStr, _ := arg.Value.(string)
					clusterID, err := strconv.ParseUint(clusterIDStr, 10, 64)
					if err != nil || clusterID == 0 {
						return nil, errs.NewBadRequest(fmt.Sprintf("Invlaid cluster parameter: %s", clusterIDStr), err)
					}
					endpoint.ClusterId = clusterID
				case "database":
					endpoint.Database, _ = arg.Value.(string)
				case "sql":
					endpoint.SqlTemplate, _ = arg.Value.(string)
					if endpoint.SqlTemplate == "" {
						return nil, errs.NewBadRequest("Invalid sql parameter: can not be empty", nil)
					}
				}
			}
		}
		argValues = append(argValues, _argValues)
	}

	if err := checkEndpointCanRun(ctx, endpoint); err != nil {
		return nil, err
	}

	cluster, err := grpc.GetClusterWithoutPermCheck(ctx, endpoint.ClusterId)
	if err != nil {
		tilog.Error(ctx, "GetClusterWithoutPermCheck failed", zap.Error(err), zap.String("region", cluster.RegionName), zap.Uint64("cluster_id", endpoint.ClusterId))
		return nil, err
	}
	if usesProviderScopedRegionalHost(cluster) {
		return nil, errs.NewBadRequest("test endpoint is unsupported for alicloud serverless", nil)
	}

	endpointRes, err := http2.TestEndpointV2WithProvider(ctx, cluster.RegionName, http2.ProviderForRegionalHostPrefix(cluster.CloudProviderName, cluster.GetDeployType()), &dto2.TestEndpointReqV2{
		ProjectId:  input.ProjectId,
		AppId:      input.AppId,
		EndpointId: input.EndpointId,
		ArgValues:  argValues,
		Credential: dto2.Credential{
			Type:      input.CredType,
			Key:       input.CredKey,
			AccountID: input.Operator,
		},
	})
	if err != nil {
		tilog.Error(ctx, "TestEndpointV2 failed", zap.Error(err), zap.String("region", config.Global().AwsConf.Region))
		return nil, err
	}
	res = convertExecSqlRespToEndpointResult(endpointRes)
	return res, nil
}

func CheckEndpointDeployComplete(ctx context.Context, endpointID uint64, version string) (bool, error) {

	endpointInfo, err := repository.GetPublishedEndpoint(endpointID)
	if err != nil {
		return false, err
	}
	region := config.Global().AwsConf.Region
	if endpointInfo.ClusterId != 0 {
		clusterInfo, err := grpc.GetClusterWithoutPermCheck(ctx, endpointInfo.ClusterId)
		if err != nil {
			return false, err
		}
		region = clusterInfo.RegionName
	}
	ddbcli, err := ddb.NewDynamodbClient(ctx, region)
	if err != nil {
		return false, errs.NewInternalDB("check endpoint deploy status failed", err)
	}
	ddbEndpointInfo, err := repository_ddb.GetPublishedEndpoint(ctx, ddbcli, endpointID)
	if err != nil {
		return false, err
	}
	if cmp, err := endpointVersionComparse(ddbEndpointInfo.Version, version); err != nil {
		return false, err
	} else {
		return cmp >= 0, nil
	}
}

func getReleaseVersion(latestVersion string) string {
	version := ""
	if latestVersion == "" {
		version = "v1"
	} else {
		s := strings.Split(latestVersion, "v")
		if len(s) > 1 {
			n, _ := strconv.Atoi(s[1])
			n = n + 1
			version = fmt.Sprintf("v%d", n)
		}
	}
	return version
}

func endpointVersionComparse(a, b string) (int, error) {
	a = strings.TrimPrefix(a, "v")
	b = strings.TrimPrefix(b, "v")
	versionA, err := strconv.ParseUint(a, 10, 64)
	if err != nil {
		return 0, errs.NewBadRequest("Invalid endpoint version", err)
	}
	versionB, err := strconv.ParseUint(b, 10, 64)
	if err != nil {
		return 0, errs.NewBadRequest("Invalid endpoint version", err)
	}
	if versionA < versionB {
		return -1, nil
	} else if versionA > versionB {
		return 1, nil
	} else {
		return 0, nil
	}
}

func argsParamsRegexp(sqlTemplate string) [][]string {
	// Remove all comments
	sqlTemplate = sqlCommentsRegexp.ReplaceAllString(sqlTemplate, "")
	// Find all arguments
	matchList := sqlArgsRegexp.FindAllStringSubmatch(sqlTemplate, -1)
	return removeDuplicateElement(matchList)
}

func removeDuplicateElement(ele [][]string) [][]string {
	result := make([][]string, 0, len(ele))
	temp := map[string]struct{}{}
	for _, item := range ele {
		if _, ok := temp[item[0]]; !ok {
			temp[item[0]] = struct{}{}
			result = append(result, item)
		}
	}
	return result
}

func getCurPublishedVersion(endpointID uint64) (string, error) {
	latestRelease, err := repository.GetPublishedEndpoint(endpointID)
	if err != nil {
		if errors.Is(err, errs.ErrNotFound) {
			return "", nil
		}
		return "", err
	}
	return latestRelease.Version, nil
}

func getLastPublishedVersion(endpointID uint64) (string, error) {
	latestRelease, err := repository.GetLastPublishVersion(endpointID)
	if err != nil {
		if errors.Is(err, errs.ErrNotFound) {
			return "", nil
		}
		return "", err
	}
	return latestRelease.Version, nil
}

func checkEndpointCanRun(ctx context.Context, endpoint *entity.Endpoint) error {
	return checkEndpointCanRunWithOpts(ctx, endpoint, true)
}
func checkEndpointCanRunWithoutCluster(ctx context.Context, endpoint *entity.Endpoint) error {
	return checkEndpointCanRunWithOpts(ctx, endpoint, false)
}
func checkEndpointCanRunWithOpts(ctx context.Context, endpoint *entity.Endpoint, checkCluster bool) error {
	tilog.Info(ctx, "checkEndpointCanRunWithOpts", zap.Any("endpoint", endpoint), zap.Bool("checkCluster", checkCluster))

	if endpoint.Endpoint == "" {
		return errs.NewBadRequest("Endpoint path is required", nil)
	}

	if !endpointPathRegexp.MatchString(endpoint.Endpoint) {
		return errs.NewBadRequest("Invalid endpoint path", nil)
	}
	if endpoint.Method == "" {
		return errs.NewBadRequest("Endpoint method is required", nil)
	}
	if !validMethod[endpoint.Method] {
		return errs.NewBadRequest("Invalid endpoint method", nil)
	}
	if !endpoint.IsSystemDataEndpoint() {
		if endpoint.ClusterId == 0 {
			return errs.NewBadRequest("Invalid endpoint cluster", nil)
		}
		if endpoint.SqlTemplate == "" {
			return errs.NewBadRequest("Invalid sql content", nil)
		}
	}

	var args []entity.EndpointArgs
	if endpoint.Args != "" {
		if err := json.Unmarshal([]byte(endpoint.Args), &args); err != nil {
			return errs.NewBadRequest("Invalid endpoint parameters config", nil)
		}
	}
	if err := checkEndpointArg(endpoint.Endpoint, args, true); err != nil {
		return err
	}
	sqlArgsMap := make(map[string]bool, len(args))
	for _, arg := range args {
		sqlArgsMap[arg.Name] = true
	}

	if !endpoint.IsSystemDataEndpoint() || endpoint.SqlTemplate != "" {
		matchs := argsParamsRegexp(endpoint.SqlTemplate)
		for _, match := range matchs {
			if match[1] == "" {
				return errs.NewBadRequest("Invalid endpoint sql parameters definition", nil)
			}
			if match[1] == constants.EndpointPageParam || match[1] == constants.EndpointPageSizeParam {
				return comerrs.ErrPaginationConflict
			}
			if !sqlArgsMap[match[1]] {
				return errs.NewBadRequest("Endpoint parameters not match", nil)
			}
		}

		if err := checkSqlSyntax(endpoint.SqlTemplate); err != nil {
			return err
		}
	}

	// check cluster status is valid
	if checkCluster && !endpoint.IsSystemDataEndpoint() {
		appInfo, err := repository.GetDataApp(endpoint.DataappId)
		if err != nil {
			return err
		}
		if err := checkClusterStatus(ctx, appInfo.OrgId, appInfo.ProjectId, []uint64{endpoint.ClusterId}, ""); err != nil {
			return err
		}
	}

	return nil
}

func checkSqlSyntax(sql string) error {
	sql = sqlCommentsRegexp.ReplaceAllString(sql, "")
	sql = sqlArgsRegexp.ReplaceAllString(sql, "?")
	sql = strings.TrimSpace(sql)
	if sql == "" {
		return errs.NewBadRequest("SQL cannot be empty", nil)
	}
	if _, _, err := parser.New().ParseSQL(sql); err != nil {
		return errs.NewBadRequest("Parsing sql failed. Please check if your sql syntax is correct, "+err.Error(), nil)
	}
	return nil
}

func getSqlRwMode(ctx context.Context, sql string) string {
	return dbhubsdk.RWMode(ctx, sql)
}

func getEndpointHost() string {
	switch {
	case config.IsDev():
		return "data.dev.tidbcloud.com"
	case config.IsStaging():
		return "data.staging.tidbcloud.com"
	default:
		return "data.tidbcloud.com"
	}
}

func GetWebConsoleHost() string {
	switch {
	case config.IsDev():
		return "us-west-2.nightly.shared.aws.tidbcloud.com"
	case config.IsStaging():
		return "us-west-2.staging.shared.aws.tidbcloud.com"
	default:
		return "us-west-2.prod.aws.tidbcloud.com"
	}
}

func BatchCheckEndpointCanRun(ctx context.Context, endpoints []*entity.Endpoint) error {
	for _, endpoint := range endpoints {
		if err := checkEndpointCanRunWithoutCluster(ctx, endpoint); err != nil {
			return errs.NewBadRequest(fmt.Sprintf(`Endpoint "%s" check failed: %s`, endpoint.Name, err.Error()), err)
		}
	}
	return nil
}

func GetLastSQLStmtType(ctx context.Context, sql string) dbhubsdk.StmtType {
	sql = sqlCommentsRegexp.ReplaceAllString(sql, "")
	sql = sqlArgsRegexp.ReplaceAllString(sql, "?")
	stmts, _, err := parser.New().ParseSQL(sql)
	if err != nil {
		return dbhubsdk.OTHER
	}
	stmtType := dbhubsdk.OTHER
	if len(stmts) > 0 {
		lastSql := stmts[len(stmts)-1]
		stmtType = dbhubsdk.GetStmtType(ctx, lastSql)
	}

	return stmtType
}

func convertSqlToJsonField(sql string) string {
	sql = sqlCommentsRegexp.ReplaceAllString(sql, "")
	sql = strings.ReplaceAll(sql, "\n", ` `)
	sql = strings.ReplaceAll(sql, `'`, `'\''`)
	sql = strings.ReplaceAll(sql, `"`, `\"`)
	sql = strings.TrimSpace(sql)
	return sql
}

func checkEndpointArg(endpointPath string, args []entity.EndpointArgs, strictMode bool) error {
	if endpointPath != "" && !endpointPathRegexp.MatchString(endpointPath) {
		return errs.NewBadRequest("Invalid endpoint path", nil)
	}
	pathParameterMap := make(map[string]bool, len(args))
	for _, segment := range strings.Split(endpointPath, "/") {
		n := len(segment)
		if n > 2 && segment[0] == '{' && segment[n-1] == '}' {
			if pathParameterMap[segment[1:n-1]] {
				return errs.NewBadRequest("duplicate path parameters:"+segment[1:n-1], nil)
			}
			pathParameterMap[segment[1:n-1]] = true
		}
	}
	for _, arg := range args {
		argType := strings.ToLower(arg.Type)
		argItemType := strings.ToLower(arg.ItemType)
		if arg.Name == "" {
			return errs.NewBadRequest("Endpoint parameter name is required", nil)
		}
		if len([]rune(arg.Description)) > 1000 {
			return errs.NewBadRequest("Argument description maximum length is 1000 characters: "+arg.Name, nil)
		}
		if !validDataType[argType] {
			return errs.NewBadRequest(fmt.Sprintf("Invalid parameter type for `%s`: %s", arg.Name, arg.Type), nil)
		}
		if arg.IsPathParameter {
			if arg.Type != entity.EndpointArgTypeString && arg.Type != entity.EndpointArgTypeInteger {
				return errs.NewBadRequest(fmt.Sprintf("Invalid path parameter type for `%s`: %s, expected types are string,integer", arg.Name, arg.Type), nil)
			}
			if !pathParameterMap[arg.Name] {
				return errs.NewBadRequest(fmt.Sprintf("Parameter `%s` is not a path parameter", arg.Name), nil)
			}
			if arg.Required != 1 {
				return errs.NewBadRequest(fmt.Sprintf("Path parameter `%s` needs to be required", arg.Name), nil)
			}
			delete(pathParameterMap, arg.Name)
		} else {
			if pathParameterMap[arg.Name] {
				return errs.NewBadRequest(fmt.Sprintf("Parameter `%s` is a path parameter", arg.Name), nil)
			}
		}
		if strictMode {
			if arg.Type == entity.EndpointArgTypeArray && !validArrayItemType[argItemType] {
				return errs.NewBadRequest(fmt.Sprintf("Invalid parameter type for `%s`: %s[%s]", arg.Name, arg.Type, arg.ItemType), nil)
			}
			if arg.Enum != "" && arg.Default != "" {
				enum := strings.Split(arg.Enum, ",")
				if !utils.IsSliceContain(enum, arg.Default) {
					return errs.NewBadRequest(fmt.Sprintf("Invalid parameter default value for `%s`, expected values:%s", arg.Name, arg.Enum), nil)
				}
			}
		}
	}
	if len(pathParameterMap) > 0 {
		var pathParameters string
		for parameter := range pathParameterMap {
			pathParameters += "," + parameter
		}
		return errs.NewBadRequest(fmt.Sprintf("The path parameter `%s` is invalid. Please review your path definition.", pathParameters[1:]), nil)
	}
	return nil
}