package controller

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"sort"
	"strconv"
	"strings"
	"time"

	"github.com/pingcap/errors"

	"dataservices/dbhubsdk"

	"dataservices/commons/config"
	"dataservices/commons/constants"
	"dataservices/commons/contextvar"
	"dataservices/commons/endpointcache"
	comerrs "dataservices/commons/errs"
	comhttp "dataservices/commons/http"
	"dataservices/commons/logger"
	"dataservices/commons/metrics"
	"dataservices/commons/redisutil"
	"dataservices/commons/sessioncache"
	"dataservices/commons/utils"
	"dataservices/ddb"
	dto "dataservices/internal/dto/dataapi_gateway"
	"dataservices/internal/entity"
	"dataservices/internal/repository_ddb"
	service_central "dataservices/internal/service/central"
	service "dataservices/internal/service/dataapi_gateway"

	"github.com/gin-gonic/gin"
	"github.com/gin-gonic/gin/binding"
	"github.com/tidbcloud/pkgv2/auth/kong"
	"github.com/tidbcloud/pkgv2/bizerr"
	"github.com/tidbcloud/tiinf/errs"
	"github.com/tidbcloud/tiinf/monitor/tilog"
	"go.buf.build/protocolbuffers/go/linuxgit/tidbcloud/base"
	"go.uber.org/zap"

	"github.com/tidbcloud/dataservices/idl/pbgen/proto/dataapi"
)

// InTest will be set in testing.
var InTest = false

func isEDAEndpoint(endpoint *entity.Endpoint) bool {
	return endpoint.Type == entity.EndpointTypeChat2EDA
}

func isChat2QueryAppEndpoint(endpoint *entity.Endpoint) bool {
	return endpoint.Type == entity.EndpointTypeChat2EDA || endpoint.Type == entity.EndpointTypeChat2Data
}

func Advanced(c *gin.Context) {
	handleQuery(c, c.Param("external_app_id"), c.Param("path"))
}

func HandleCustomDomainQuery(c *gin.Context) {
	ctx := c.Request.Context()
	endpointType := c.GetHeader("endpoint-type")
	epResType := constants.EndpointResTypeSQL
	host := c.Request.Host
	path := c.Request.URL.Path
	region := config.Global().AwsConf.Region
	isDraft := endpointType == "draft"

	logger := tilog.WithContextL(ctx).With(
		zap.String("region", region),
		zap.String("endpoint-type", endpointType),
		zap.String("host", host),
		zap.String("path", path),
		zap.String("method", c.Request.Method),
	)
	ctx = contextvar.WithLogger(ctx, logger)

	if strings.HasSuffix(c.Request.Host, "tidbcloud.com") {
		fail(c, ctx, epResType, nil, "", isDraft, errs.NewNotFound("Not Found", errors.New("invalid domain")))
		return
	}
	domains, err := service.ListCustomDomains(ctx, c.Request.Host)
	if err != nil {
		fail(c, ctx, epResType, nil, "", isDraft, err)
		return
	}

	var endpointPath, externalAppId string
	for _, domain := range domains {
		if !strings.HasPrefix(domain.DefaultDomainName, region) {
			continue
		}
		if !strings.HasPrefix(path, domain.CustomPath) {
			continue
		}
		if domain.Expired {
			continue
		}

		externalAppId = domain.ExternalAppID
		if domain.CustomPath == "/" {
			endpointPath = path
		} else {
			p := strings.TrimPrefix(path, domain.CustomPath)
			if p == "" {
				p = "/"
			} else if !strings.HasPrefix(p, "/") {
				continue
			}
			endpointPath = p
		}
	}
	if endpointPath == "" {
		fail(c, ctx, epResType, nil, "", isDraft, errs.NewNotFound("Not Found", errors.New("endpointPath not found")))
		return
	}

	handleQuery(c, externalAppId, endpointPath)
}

func handleQuery(c *gin.Context, externalAppId, path string) {
	ctx := c.Request.Context()
	endpointType := c.GetHeader("endpoint-type")
	var publicKey string
	epResType := constants.EndpointResTypeSQL
	method := c.Request.Method
	if strings.HasPrefix(externalAppId, "chat2query") {
		epResType = constants.EndpointResTypeChat2Data
	}
	isDraft := endpointType == "draft"

	logger := tilog.WithContextL(ctx).With(
		zap.String("region", config.Global().AwsConf.Region),
		zap.String("external_app_id", externalAppId),
		zap.String("endpoint-type", endpointType),
		zap.String("path", path),
		zap.String("method", c.Request.Method),
	)

	metricEndpointId := "unknown_endpoint"
	defer func() {
		latency := time.Since(contextvar.RequestBeginTime(c.Request.Context()))
		latencyHistogram.WithLabelValues(externalAppId, metricEndpointId).Observe(float64(latency.Microseconds()) / 1000.0)
	}()

	authResult, err := kong.GetOpenAPIAuthResult(c.Request)
	if err != nil {
		fail(c, contextvar.WithLogger(ctx, logger), epResType, nil, "", isDraft, errs.NewUnauthorized("unauthorized", err))
		errorCounter.WithLabelValues(externalAppId, metricEndpointId).Inc()
		return
	}
	publicKey = authResult.PublicKey
	logger = logger.With(zap.String("public_key", publicKey))
	ctx = contextvar.WithLogger(ctx, logger)
	logger.Info("advanced endpoint begin")

	if path == "/chat2data" { // for compatibility
		path = "/v1/chat2data"
	}

	entityApp, entityEndpoint, pathParams, err := GetEndpoint(ctx, externalAppId, path, c.Request.Method, isDraft)
	if err != nil {
		metrics.ExecEndpointFailedCounter.WithLabelValues("get_endpoint_failed").Inc()
		errorCounter.WithLabelValues(externalAppId, metricEndpointId).Inc()
		fail(c, ctx, epResType, nil, publicKey, isDraft, err)
		return
	}
	logger.Info("endpoint path parameters:", zap.Any("params", pathParams))

	metricEndpointId = fmt.Sprintf("%d", entityEndpoint.ID)
	var copiedBody string
	succResAndLog := func(result *dto.AdvancedRes, cached bool) {
		if result == nil {
			defaultCustomLogger.LogExecuteSuccess(c, result, entityEndpoint, entityApp, cached, copiedBody)
			tilog.Info(ctx, "advanced endpoint succeed, but result is nil", zap.String("endpoint_type", epResType), zap.String("caller", publicKey))
			return
		}

		succ(c, ctx, result, entityEndpoint, publicKey, isDraft)
		defaultCustomLogger.LogExecuteSuccess(c, result, entityEndpoint, entityApp, cached, copiedBody)
		if result.Data.Result.Code != 200 {
			errorCounter.WithLabelValues(externalAppId, metricEndpointId).Inc()
		}
	}
	failResAndLog := func(err error) {
		failedRes := fail(c, ctx, epResType, entityEndpoint, publicKey, isDraft, err)
		defaultCustomLogger.LogExecuteError(c, failedRes, entityEndpoint, entityApp, copiedBody)
		errorCounter.WithLabelValues(externalAppId, metricEndpointId).Inc()
	}

	paramsArray, cbody, err := ParamFormat4Http(c, ctx, entityEndpoint)
	copiedBody = cbody
	if err != nil {
		metrics.ExecEndpointFailedCounter.WithLabelValues("parse_param_failed").Inc()
		failResAndLog(err)
		return
	}
	for idx, params := range paramsArray {
		for key, value := range pathParams {
			if params == nil {
				params = make(map[string]any)
				paramsArray[idx] = params
			}
			params[key] = value
		}
	}
	if !InTest && !isChat2QueryAppEndpoint(entityEndpoint) {
		redirect, err := Redirect(ctx, c, entityApp.ProjectId, entityEndpoint.ClusterId, true)
		if err != nil {
			failResAndLog(err)
			return
		}
		if redirect {
			return
		}
	}

	if !InTest {
		// auth
		if !config.IsLocalMode() {
			if isEDAEndpoint(entityEndpoint) {
				// for EDA endpoints, we check permission in eda service, so skip here to let the process goto chat2queryEndpoint function
			} else {
				err = service.CheckApiKeyPermission(ctx, entityEndpoint.DataappId, entityEndpoint.ClusterId, entityEndpoint.RWMode, publicKey)
				if err != nil {
					failResAndLog(err)
					return
				}
			}
		}

		// check if the dataapp has permission to access the cluster
		if !isChat2QueryAppEndpoint(entityEndpoint) {
			has, err := service.HasClusterPermission(ctx, entityEndpoint.DataappId, entityEndpoint.ClusterId)
			if err != nil {
				failResAndLog(err)
				return
			} else if !has {
				failResAndLog(errs.NewUnauthorized("No cluster permissions", nil))
				return
			}
		}
	}

	if isChat2QueryAppEndpoint(entityEndpoint) {
		res, redirect, err := chat2queryEndpoint(c, entityApp, entityEndpoint, publicKey)
		if err != nil {
			failResAndLog(err)
		} else if !redirect {
			succResAndLog(res, false)
		}
		return
	}

	returnAllStmtResult := c.GetHeader("return_all_stmt_result")
	var sessionID string
	if config.Global().EndpointConf.EnableSession {
		sessionID = sessioncache.GenEndpointSessionID(entityEndpoint.ID)
	}
	endpoint, err := parseEntityEndpointToExec(entityEndpoint, entityApp.ProjectId, sessionID, strings.ToLower(returnAllStmtResult) == "true")
	if err != nil {
		metrics.ExecEndpointFailedCounter.WithLabelValues("parse_endpoint_failed").Inc()
		failResAndLog(err)
		return
	}

	credential := &dataapi.Credential{
		Type: dataapi.CrtType_APIKEY,
		Key:  publicKey,
	}
	resJson, resp, err1 := ExecEndpoints(ctx, endpoint, paramsArray, credential, &ExecEndpointOption{
		RequestMethod: method,
		RequestPath:   path,
		Operator:      publicKey,
		ExternalAppID: externalAppId,
		EndpointType:  endpointType,
		OrgID:         entityApp.OrgId,
		MeasureType:   constants.NeedMeasure,
		Source:        entity.DataAPISourceExecEndpoint,
	})
	if err1 != nil {
		failResAndLog(err1)
		return
	}
	if resp.BaseResp != nil && resp.BaseResp.ErrCode > 0 && resp.BaseResp.ErrMsg != "" {
		failResAndLog(errs.GetTiErrFromBaseResp(resp.BaseResp))
		return
	}
	var result dto.AdvancedRes
	json.Unmarshal(resJson, &result)

	succResAndLog(&result, (!isDraft) && endpoint.method == http.MethodGet && endpoint.setting.IsCacheEnabled() && endpoint.setting.TTLValid())
}

func fail(c *gin.Context, ctx context.Context, resType string, endpoint *entity.Endpoint, caller string, isDraft bool, err error) *dto.AdvancedRes {
	var httpCode int
	var message string
	switch e := err.(type) {
	case errs.Error:
		httpCode = e.HTTPCode()
		message = e.ErrMsg()
	case bizerr.BizError:
		httpCode = e.HttpStatus()
		message = e.ErrMsg()
	default:
		httpCode = http.StatusInternalServerError
		message = "System error"
	}
	var internalErr string
	if httpCode == http.StatusInternalServerError {
		internalErr = message
		message = "System error"
	}

	startTime := contextvar.RequestBeginTime(c.Request.Context())
	endTime := time.Now()
	result := dto.ConvertErrorToAdvancedRes(httpCode, message, resType)
	result.Data.Result.StartMs = startTime.UnixMilli()
	result.Data.Result.EndMs = endTime.UnixMilli()
	result.Data.Result.Latency = endTime.Sub(startTime).String()

	logger.LogWithError(ctx, err, "DS-Region-advanced endpoint failed",
		zap.String("endpoint_type", result.Type),
		zap.String("caller", caller),
		zap.Duration("cost", time.Since(startTime)),
		zap.Int64("result.code", result.Data.Result.Code),
		zap.String("resp.message", result.Data.Result.Message))

	reportEndpointExecRecord(c, endpoint, &result.Data, caller, isDraft, "", internalErr)
	c.JSON(httpCode, result)
	return result
}

func succ(c *gin.Context, ctx context.Context, result *dto.AdvancedRes, endpoint *entity.Endpoint, caller string, isDraft bool) {
	startTime := contextvar.RequestBeginTime(c.Request.Context())
	endTime := time.Now()
	result.Data.Result.StartMs = startTime.UnixMilli()
	result.Data.Result.EndMs = endTime.UnixMilli()
	result.Data.Result.Latency = endTime.Sub(startTime).String()
	logger.Info(ctx, "advanced endpoint succeed",
		zap.String("endpoint_type", result.Type),
		zap.String("caller", caller),
		zap.Duration("cost", time.Since(startTime)),
		zap.Int64("result.code", result.Data.Result.Code),
		zap.String("resp.message", result.Data.Result.Message),
	)
	// do not report suc log
	if result.Data.Result.Code != 200 {
		reportEndpointExecRecord(c, endpoint, &result.Data, caller, isDraft, "", "")
	}
	result.Data.Result.Databases = nil
	result.Data.Result.WarnCount = 0
	result.Data.Result.WarnMessages = nil
	c.JSON(http.StatusOK, result)
}

func reportEndpointExecRecord(c *gin.Context, endpoint *entity.Endpoint, result *entity.DataRes, caller string, isDraft bool, channel string, internalErr string) {
	if endpoint == nil || result == nil {
		return
	}
	ctx := utils.TraceContext(c.Request.Context())
	traceID := utils.TraceIDFromContext(ctx)
	go func() {
		var status int8
		if result.Result.Code == 200 {
			status = 1 // Succ
		} else {
			status = 2 // Fail
		}
		if endpoint.ID > 0 && endpoint.DataappId == 0 {
			dbEndpoint, err := GetTestEndpointById(ctx, endpoint.ID)
			if err != nil {
				tilog.Error(ctx, "get endpoint info failed", zap.Uint64("endpoint_id", endpoint.ID), zap.Error(err))
			} else {
				endpoint = dbEndpoint
			}
		}

		if err := comhttp.ReportEndpointExecRecord(ctx, &entity.EndpointExecRecord{
			DataAppID:  endpoint.DataappId,
			EndpointID: endpoint.ID,
			PublishID:  endpoint.PublishID,
			Type:       endpoint.Type,
			Region:     config.Global().AwsConf.Region,
			Caller:     caller,
			Status:     status,
			Channel:    channel,
			Code:       result.Result.Code,
			Message:    result.Result.Message + " " + internalErr,
			StartMs:    result.Result.StartMs,
			EndMs:      result.Result.EndMs,
			Latency:    result.Result.Latency,
			AiLatency:  result.Result.AiLatency,
			RowCount:   result.Result.RowCount,
			RowAffect:  result.Result.RowAffect,
			Limit:      result.Result.Limit,
			TraceID:    traceID,
			IsDraft:    isDraft,
			CreatedAt:  utils.Now(),
		}); err != nil {
			tilog.Error(ctx, "failed to report endpoint execute record", zap.Error(err), zap.String("region", config.Global().AwsConf.Region))
		}
	}()
}

func execEndpointWriteCache(
	params map[string]any,
	endpoint *EndpointExec,
	ctx context.Context,
	publicKey,
	externalAppId,
	endpointType string,
	resp *dataapi.ExecSqlRespV2,
) {
	cacheKey := endpoint.cacheKey(ctx, params, externalAppId)
	if endpointType != "draft" && resp.BaseResp.ErrCode == 0 && endpoint.method == http.MethodGet && endpoint.setting.IsCacheEnabled() && endpoint.setting.TTLValid() {
		value, _ := json.Marshal(resp.StmtResults)
		cacheLimiter := endpointcache.GlobalCacheLimiter()
		success := cacheLimiter.PutItem(&endpointcache.CacheItem{
			ExternalAppId: externalAppId,
			WrittenTime:   time.Now(),
			WrittenBytes:  len(value),
			TTL:           endpoint.setting.CacheTTL,
		})
		if !success {
			tilog.Info(
				ctx,
				"redis cache reach storage limit",
				zap.String("key", cacheKey),
				zap.String("public_key", publicKey),
				zap.String("external_app_id", externalAppId),
				zap.Int("bytes", len(value)),
			)
		} else {
			masterRedisClient := redisutil.MasterRedisClient()
			if err := masterRedisClient.Set(ctx, cacheKey, value, time.Second*time.Duration(endpoint.setting.CacheTTL)).Err(); err != nil {
				tilog.Error(
					ctx,
					"set redis cache failed",
					zap.Error(err),
					zap.String("key", cacheKey),
					zap.String("public_key", publicKey),
					zap.String("external_app_id", externalAppId),
				)
			} else {
				tilog.Info(
					ctx,
					"set redis cache success",
					zap.String("key", cacheKey),
					zap.String("public_key", publicKey),
					zap.String("external_app_id", externalAppId),
				)
			}
		}
	}

}

func execEndpointReadCache(
	params map[string]any,
	endpoint *EndpointExec,
	ctx context.Context,
	publicKey,
	externalAppId,
	endpointType string,
) ([]byte, *dataapi.ExecSqlRespV2) {
	startMs := time.Now()
	cacheKey := endpoint.cacheKey(ctx, params, externalAppId)
	if endpointType != "draft" && endpoint.method == http.MethodGet && endpoint.setting.IsCacheEnabled() && endpoint.setting.TTLValid() {
		replicaRedisClient := redisutil.MasterRedisClient()
		cacheValue, err := replicaRedisClient.Get(ctx, cacheKey).Bytes()
		if err != nil {
			metrics.CacheHitCounter.WithLabelValues(strconv.FormatUint(endpoint.id, 10), externalAppId, "miss").Inc()
			tilog.Info(
				ctx,
				"want to use cache, but read redis failed",
				zap.Error(err),
				zap.String("key", cacheKey),
				zap.String("public_key", publicKey),
				zap.String("external_app_id", externalAppId),
			)
		} else if len(cacheValue) == 0 {
			metrics.CacheHitCounter.WithLabelValues(strconv.FormatUint(endpoint.id, 10), externalAppId, "miss").Inc()
			tilog.Info(
				ctx,
				"want to use cache, but get empty result",
				zap.String("key", cacheKey),
				zap.String("public_key", publicKey),
				zap.String("external_app_id", externalAppId),
			)
		} else {
			metrics.CacheHitCounter.WithLabelValues(strconv.FormatUint(endpoint.id, 10), externalAppId, "hit").Inc()
			var results []*dataapi.ExecStmtResult
			err = json.Unmarshal(cacheValue, &results)
			if err != nil {
				tilog.Error(
					ctx, "unmarshal cache result failed",
					zap.Error(err),
					zap.String("key", cacheKey),
					zap.String("public_key", publicKey),
					zap.String("external_app_id", externalAppId),
				)
			} else {
				tilog.Info(
					ctx, "get result from cache success",
					zap.String("key", cacheKey),
					zap.String("public_key", publicKey),
					zap.String("external_app_id", externalAppId),
				)
				resJson := dto.ConvertToAdvancedRes(results, constants.EndpointResTypeSQL)
				resJson.Data.Result.StartMs = startMs.UnixMilli()
				resJson.Data.Result.EndMs = time.Now().UnixMilli()
				resJson.Data.Result.Latency = time.Since(startMs).String()
				output, _ := json.Marshal(resJson)
				return output, &dataapi.ExecSqlRespV2{StmtResults: results, BaseResp: &base.BaseResp{}}
			}
		}
	}
	return nil, nil
}

type ExecEndpointOption struct {
	Operator      string
	ExternalAppID string
	EndpointType  string
	RequestPath   string
	RequestMethod string
	MeasureType   string
	OrgID         uint64
	Source        string
}

func ExecEndpoints(
	ctx context.Context,
	endpoint *EndpointExec,
	paramsArray []map[string]any,
	credential *dataapi.Credential,
	opt *ExecEndpointOption,
) ([]byte, *dataapi.ExecSqlRespV2, errs.Error) {
	start := time.Now()
	defer func() {
		cost := time.Since(start)
		metrics.ExecutionLinkDuration.WithLabelValues("endpoint_exec", "exec_endpoints").Observe(cost.Seconds())
	}()

	var (
		publicKey     string
		externalAppId string
		endpointType  string
		orgId         uint64
		measureType   string
		path          string
		method        string
		source        string
	)

	if opt != nil {
		publicKey = opt.Operator
		externalAppId = opt.ExternalAppID
		endpointType = opt.EndpointType
		orgId = opt.OrgID
		measureType = opt.MeasureType
		path = opt.RequestPath
		method = opt.RequestMethod
		source = opt.Source
	}

	startTime := contextvar.RequestBeginTime(ctx)
	if !endpoint.batch || len(paramsArray) < 1 {
		var params map[string]any
		if len(paramsArray) < 1 {
			params = nil
		} else {
			params = paramsArray[0]
		}

		// read from cache
		cacheOutput, cacheResp := execEndpointReadCache(params, endpoint, ctx, publicKey, externalAppId, endpointType)
		if cacheResp != nil {
			return cacheOutput, cacheResp, nil
		}

		_, resp, execErr := endpoint.exec(ctx, params, credential, orgId, measureType, source)
		if execErr != nil {
			return nil, nil, execErr
		}

		tilog.Info(ctx, "advanced endpoint result",
			zap.String("public_key", publicKey),
			zap.String("external_app_id", externalAppId),
			zap.String("endpoint_type", endpointType),
			zap.String("path", path),
			zap.String("method", method),
			zap.String("source", source),
			zap.Duration("cost", time.Since(startTime)),
			zap.Int64("resp.err_code", resp.BaseResp.ErrCode),
			zap.String("resp.err_msg", resp.BaseResp.ErrMsg))

		// write to cache
		execEndpointWriteCache(params, endpoint, ctx, publicKey, externalAppId, endpointType, resp)

		resJson := dto.ConvertToAdvancedRes(resp.GetStmtResults(), constants.EndpointResTypeSQL)
		output, err1 := json.Marshal(resJson)
		if err1 != nil {
			return nil, nil, errs.NewInternalServerError("ConvertToAdvancedRes failed!", err1)
		}
		return output, resp, nil
	} else {
		maxDatasetSize := constants.BatchRequestMaxDatasetSize
		if len(paramsArray) > maxDatasetSize {
			metrics.ExecEndpointFailedCounter.WithLabelValues("batch_exceed_limit").Inc()
			return nil, nil, errs.NewBadRequest(fmt.Sprintf("The submitted dataset is too large, exceeding the limit of %d entries.", maxDatasetSize), nil)
		}
		affect := int64(0)
		batchRows := make([]entity.BatchRowsStr, 0)
		batchResult := &dataapi.ExecStmtResult{}
		queryResult := make([]*dataapi.ExecStmtResult, 0)
		respS := &dataapi.ExecSqlRespV2{}
		batchResult.StartMs = time.Now().UnixNano() / int64(time.Millisecond)
		for index, v := range paramsArray {
			req, resp, err := endpoint.exec(ctx, v, credential, orgId, measureType, source)
			if err != nil {
				batchRows = append(batchRows, entity.BatchRowsStr{
					Index:           strconv.Itoa(index),
					Success:         strconv.FormatBool(false),
					Message:         fmt.Sprintf("Execution was failed, error:%s", err.Error()),
					AutoIncrementID: "null",
				})
			} else {
				tilog.Info(ctx, "advanced endpoint result",
					zap.String("public_key", publicKey),
					zap.String("external_app_id", externalAppId),
					zap.String("endpoint_type", endpointType),
					zap.String("path", path),
					zap.String("method", method),
					zap.String("source", source),
					zap.Duration("cost", time.Since(startTime)))
				stmtType := service_central.GetLastSQLStmtType(ctx, req.Sql)
				if stmtType == dbhubsdk.UPDATE || stmtType == dbhubsdk.INSERT || stmtType == dbhubsdk.DELETE {
					if resp.BaseResp != nil && resp.BaseResp.ErrCode != 0 {
						// Internal Error
						batchRows = append(batchRows, entity.BatchRowsStr{
							Index:           strconv.Itoa(index),
							Success:         strconv.FormatBool(false),
							Message:         fmt.Sprintf("Row failed to %v, code:%v, error:%s", stmtType, resp.BaseResp.ErrCode, resp.BaseResp.ErrMsg),
							AutoIncrementID: "null",
						})
					} else if len(resp.StmtResults) > 0 {
						// !Internal Error
						batchRows = GenerateBatchRows(resp, batchRows, index, stmtType)
						affect = affect + resp.StmtResults[0].Affect
						if index == 0 {
							batchResult.Databases = resp.StmtResults[0].Databases
							batchResult.StmtType = resp.StmtResults[0].StmtType
							batchResult.LimitRows = resp.StmtResults[0].LimitRows
						}
					}
					if index == len(paramsArray)-1 {
						batchResult.ErrCode = 0
						batchResult.RowNum = int64(len(paramsArray))
						batchResult.Affect = affect
					}
				} else {
					batchRows = GenerateBatchRows(resp, batchRows, index, stmtType)
					if resp != nil && index == len(paramsArray)-1 {
						queryResult = resp.GetStmtResults()
						respS = resp
					}
				}
			}
		}
		batchResult.EndMs = time.Now().UnixNano() / int64(time.Millisecond)
		if len(batchRows) > 0 && len(queryResult) == 0 {
			//resJson := dto.ConvertToBatchAdvancedRes(batchResult, batchRows, constants.EndpointResTypeSQL)
			//resp = dto.BatchConvertToExecSqlRespV2(resJson, batchResult.StmtType)
			//output, err1 := json.Marshal(resJson)
			output, resp, err1 := toBatchResult(batchResult, batchRows)
			if err1 != nil {
				return nil, nil, errs.NewInternalServerError("ConvertToBatchAdvancedRes failed!", err1)
			}
			return output, resp, nil
		} else if !verifyAllSuccess(batchRows) && len(queryResult) > 0 {
			output, resp, err1 := toBatchResult(batchResult, batchRows)
			if err1 != nil {
				return nil, nil, errs.NewInternalServerError("ConvertToBatchAdvancedRes failed!", err1)
			}
			return output, resp, nil
		} else {
			resJson := dto.ConvertToAdvancedRes(queryResult, constants.EndpointResTypeSQL)
			output, err1 := json.Marshal(resJson)
			if err1 != nil {
				return nil, nil, errs.NewInternalServerError("ConvertToAdvancedRes failed!", err1)
			}
			return output, respS, nil
		}
	}
}

func toBatchResult(batchResult *dataapi.ExecStmtResult, batchRows []entity.BatchRowsStr) ([]byte, *dataapi.ExecSqlRespV2, error) {
	resJson := dto.ConvertToBatchAdvancedRes(batchResult, batchRows, constants.EndpointResTypeSQL)
	resp := dto.BatchConvertToExecSqlRespV2(resJson, batchResult.StmtType)
	output, err1 := json.Marshal(resJson)
	if err1 != nil {
		return nil, nil, errs.NewInternalServerError("ConvertToBatchAdvancedRes failed!", err1)
	}
	return output, resp, err1
}

func verifyAllSuccess(batchRows []entity.BatchRowsStr) bool {
	for _, v := range batchRows {
		if v.Success == "false" {
			return false
		}
	}
	return true
}

func GenerateBatchRows(resp *dataapi.ExecSqlRespV2, batchRowsStr []entity.BatchRowsStr, index int, stmtType dbhubsdk.StmtType) []entity.BatchRowsStr {
	autoIncrementID := "null"
	if resp.BaseResp != nil && resp.BaseResp.ErrCode != 0 {
		batchRowsStr = append(batchRowsStr, entity.BatchRowsStr{
			Index:           strconv.Itoa(index),
			Success:         strconv.FormatBool(false),
			Message:         fmt.Sprintf("Row failed to %v, code:%v,error:%s", stmtType, resp.BaseResp.ErrCode, resp.BaseResp.ErrMsg),
			AutoIncrementID: autoIncrementID,
		})
		return batchRowsStr
	}
	if len(resp.StmtResults) > 0 {
		if resp.StmtResults[0].ErrCode == 0 {
			autoIncrementID = strconv.FormatInt(resp.StmtResults[0].LastInsertId, 10)
			if autoIncrementID == "0" {
				autoIncrementID = "null"
			}
			message := "Execution was successful, but 0 rows affected."
			if resp.StmtResults[0].Affect > 0 {
				message = fmt.Sprintf("Row %v successfully", stmtType)
			}
			batchRowsStr = append(batchRowsStr, entity.BatchRowsStr{
				Index:           strconv.Itoa(index),
				Success:         strconv.FormatBool(true),
				Message:         message,
				AutoIncrementID: autoIncrementID,
			})
		} else {
			batchRowsStr = append(batchRowsStr, entity.BatchRowsStr{
				Index:           strconv.Itoa(index),
				Success:         strconv.FormatBool(false),
				Message:         fmt.Sprintf("Row failed to %v, code:%v,error:%s", stmtType, resp.StmtResults[0].ErrCode, resp.StmtResults[0].ErrMessage),
				AutoIncrementID: autoIncrementID,
			})
		}
	} else {
		batchRowsStr = append(batchRowsStr, entity.BatchRowsStr{
			Index:           strconv.Itoa(index),
			Success:         strconv.FormatBool(false),
			Message:         fmt.Sprintf("Row failed to %v", stmtType),
			AutoIncrementID: autoIncrementID,
		})
	}
	return batchRowsStr
}

var ErrsCodeToHttpCode = map[int64]int{
	errs.ErrCodeBadRequest:          400,
	errs.ErrCodeUnauthorized:        401,
	errs.ErrCodePaymentRequired:     402,
	errs.ErrCodeForbidden:           403,
	errs.ErrCodeNotFound:            404,
	errs.ErrCodeInternalServerError: 500,
	errs.ErrCodeDB:                  500,
}

func ParamFormat4Http(c *gin.Context, ctx context.Context, endpoint *entity.Endpoint) (paramsArray []map[string]any, copiedBody string, err errs.Error) {

	start := time.Now()
	defer func() {
		cost := time.Since(start)
		metrics.ExecutionLinkDuration.WithLabelValues("endpoint_exec", "read_body").Observe(cost.Seconds())
		logger.Info(ctx, "ParamFormat4Http complated", zap.Any("params", paramsArray), zap.Duration("cost", cost), zap.Error(err))
	}()

	if c.Request.Method == http.MethodGet || c.Request.Method == http.MethodDelete {
		params := make(map[string]any)
		for key, values := range c.Request.URL.Query() {
			var value string
			if len(values) > 0 {
				value = values[len(values)-1]
			}
			params[key] = value
		}
		paramsArray = append(paramsArray, params)
		return paramsArray, "", nil
	}

	if c.ContentType() != binding.MIMEJSON && c.Request.ContentLength != 0 {
		return nil, "", errs.NewBadRequest("Parse param failed, Only json formatted content is supported", nil)
	}

	body, berr := io.ReadAll(c.Request.Body)
	if berr != nil && berr != io.EOF {
		return nil, "", errs.NewBadRequest("failed to read request body:"+berr.Error(), berr)
	}
	if len(body) > 0 {
		c.Request.Body = io.NopCloser(bytes.NewReader(body))
		defer func() {
			if err == nil {
				c.Request.Body = io.NopCloser(bytes.NewReader(body))
			}
		}()
	}
	copiedBody = string(body)

	if endpoint.IsBatch() {
		if len(body) == 0 {
			return nil, "", errs.NewBadRequest("Parse param failed, body can not be empty", nil)
		}
		// request body can be either [item] or {"items": [item]}
		if body[0] == '[' {
			cerr := c.ShouldBindBodyWith(&paramsArray, binding.JSON)
			if cerr != nil {
				return nil, "", errs.NewBadRequest("Parse param failed. Tips: The endpoint supports batch, please check if request body format.", err)
			}
		} else {
			var paramMapObj map[string][]map[string]any
			cerr := c.ShouldBindBodyWith(&paramMapObj, binding.JSON)
			if cerr != nil {
				return nil, "", errs.NewBadRequest("Parse param failed. Tips: The endpoint supports batch, please check if request body format.", err)
			}
			var ok bool
			paramsArray, ok = paramMapObj["items"]
			if !ok {
				return nil, "", errs.NewBadRequest("Parse param failed. Tips: The endpoint supports batch, please check if request body format. error: items field not found", nil)
			}
		}
		return
	}

	var bodyMap map[string]any
	if berr := c.ShouldBindBodyWith(&bodyMap, binding.JSON); berr != nil && berr != io.EOF {
		return nil, "", errs.NewBadRequest("Parse param failed. Tips: The endpoint doesn't support batch, the data type must be an object.", berr)
	}
	paramsArray = append(paramsArray, bodyMap)
	err = parseParamsWithEndpoint(ctx, endpoint, paramsArray)
	return
}

func parseParamsWithEndpoint(ctx context.Context, endpoint *entity.Endpoint, paramsArray []map[string]any) errs.Error {

	if endpoint.IsSystemDataEndpoint() {
		var args entity.EndpointArgsList
		if err := utils.FromJson(endpoint.Args, &args); err != nil {
			logger.Error(ctx, "parseParamsWithEndpoint failed to parse endpoint args", zap.Error(err), zap.Uint64("endpoint_id", endpoint.ID))
		}
		if len(paramsArray) == 0 {
			return errs.NewBadRequest("Parameters is required", nil)
		}
		var (
			params      = paramsArray[0]
			clusterID   uint64
			database    string
			sqltemplate string
			err         error
		)

		dftErr := errs.NewBadRequest("Failed to parse endpoint arguments", nil)

		for _, arg := range args {
			argVal, err := arg.Parse(params[arg.Name])
			if err != nil {
				return utils.ConvertToTiError(err, dftErr)
			}
			switch arg.Name {
			case "cluster_id":
				clusterID, err = argVal.Uint64()
				if err != nil {
					return utils.ConvertToTiError(err, dftErr)
				}
			case "database":
				database = argVal.String()
			case "sql":
				sqltemplate = argVal.String()
			}
		}

		if clusterID == 0 {
			return errs.NewBadRequest("Invalid parameters: cluster_id", err)
		}
		if database == "" {
			return errs.NewBadRequest("Invalid parameters: database is required", err)
		}
		if sqltemplate == "" {
			return errs.NewBadRequest("Invalid parameters: sql is required", err)
		}
		endpoint.ClusterId = clusterID
		endpoint.Database = database
		endpoint.SqlTemplate = sqltemplate
		endpoint.RWMode = dbhubsdk.RWMode(ctx, sqltemplate)
		delete(params, "cluster_id")
		delete(params, "database")
		delete(params, "sql")
	}
	return nil
}

func GetEndpoint(ctx context.Context, appExtId, endpointPath, method string, isDraft bool) (app *entity.DataApp, endpoint *entity.Endpoint, params map[string]string, err error) {
	start := time.Now()
	defer func() {
		metricTpEndpoint := metrics.TpPublishEndpoint
		if isDraft {
			metricTpEndpoint = metrics.TpDraftEndpoint
		}
		ok := strconv.FormatBool(err == nil)
		cost := time.Since(start)
		metrics.GetEndpointDuration.WithLabelValues(metricTpEndpoint, "by_name", ok).Observe(cost.Seconds())
		metrics.ExecutionLinkDuration.WithLabelValues("endpoint_exec", "get_endpoint_conf").Observe(cost.Seconds())
		tilog.Info(ctx, "get endpoint and dataapp info completed",
			zap.String("external_app_id", appExtId),
			zap.String("endpoint_path", endpointPath),
			zap.String("method", method),
			zap.Bool("is_draft", isDraft),
			zap.Duration("cost", cost),
			zap.Error(err),
		)
	}()

	app, err = service.GetDataAppByExtId(ctx, appExtId)
	if err != nil {
		return
	}

	method = strings.ToUpper(method)
	if isDraft {
		endpoint, params, err = service.GetDraftEndpointByPath(ctx, app, method, endpointPath)
		return
	}

	endpointPublish, params, err := service.GetDeployedEndpointByPath(ctx, app, method, endpointPath)
	if err != nil {
		return
	}
	endpoint = endpointPublish.ToEndpoint()

	// get endpoint name for logger monitor
	endpointdraft, nerr := service.GetDraftEndpointByID(ctx, app, endpoint.ID)
	if nerr == nil && endpointdraft != nil {
		endpoint.Name = endpointdraft.Name
	}
	return
}

func GetTestEndpointById(ctx context.Context, endpointId uint64) (draftEndpoint *entity.Endpoint, err errs.Error) {
	start := time.Now()
	defer func() {
		ok := strconv.FormatBool(err == nil)
		metrics.GetEndpointDuration.WithLabelValues(metrics.TpDraftEndpoint, "by_id", ok).Observe(time.Since(start).Seconds())
	}()

	draftEndpoint, err = service.GetDraftEndpoint(ctx, endpointId)
	if err != nil {
		return nil, err
	}
	return draftEndpoint, nil
}

func BuildEndpoint(sqlTemplate string, argsList []*entity.EndpointArgs, params map[string]any) (string, []entity.SqlArgValue, error) {
	sqlArgs := make([]SqlArg, 0)
	avMap := make(map[string]entity.ArgValue, len(argsList))
	for _, arg := range argsList {
		name := "${" + arg.Name + "}"
		av, err := arg.Parse(params[arg.Name])
		if err != nil {
			return "", nil, err
		}
		avMap[arg.Name] = av

		offset := 0
		template := sqlTemplate
		if err != nil {
			return "", nil, err
		}
		for {
			idx := strings.Index(template, name)
			if idx < 0 {
				break
			}
			sqlArgs = append(sqlArgs, SqlArg{
				idx: idx + offset,
				av:  av,
			})
			template = template[idx+len(name):]
			offset = offset + idx + len(name)
		}
	}

	for _, arg := range argsList {
		name := "${" + arg.Name + "}"
		placeholder := strings.Repeat("?,", len(avMap[arg.Name].Values()))
		placeholder = strings.TrimRight(placeholder, ",")
		sqlTemplate = strings.ReplaceAll(sqlTemplate, name, placeholder)
	}

	sqlTemplate = strings.ReplaceAll(sqlTemplate, `\\n`, " ")
	sqlTemplate = strings.ReplaceAll(sqlTemplate, `\n`, " ")
	sqlTemplate = strings.TrimSpace(sqlTemplate)
	sort.Slice(sqlArgs, func(i, j int) bool {
		return sqlArgs[i].idx < sqlArgs[j].idx
	})
	args := make([]entity.SqlArgValue, 0, len(sqlArgs))
	for _, sqlArg := range sqlArgs {
		for _, sqlValue := range sqlArg.av.Values() {
			args = append(args, entity.SqlArgValue{
				Type: sqlArg.av.ItemType(),
				Val:  sqlValue,
			})
		}
	}
	return sqlTemplate, args, nil
}

type SqlArg struct {
	idx int
	av  entity.ArgValue
}

func Redirect(ctx context.Context, c *gin.Context, projectId, clusterId uint64, dedicatedAllowed bool) (bool, errs.Error) {
	start := time.Now()
	defer func() {
		cost := time.Since(start)
		metrics.ExecutionLinkDuration.WithLabelValues("endpoint_exec", "check_redirect").Observe(cost.Seconds())
		tilog.Info(ctx, "check redirect completed",
			zap.Uint64("project_id", projectId),
			zap.Uint64("cluster_id", clusterId),
			zap.Duration("cost", cost),
		)
	}()
	cfg := config.Global()
	if !cfg.RegionalConf.GatewayRedirect {
		return false, nil
	}
	clusterInfo, err := repository_ddb.GetUserClusterInfo(ctx, ddb.Client(), projectId, clusterId)
	if err != nil {
		if errs.IsError(err, comerrs.ErrUserClusterNotFound) {
			tierr, _ := errs.AsTiError(err)
			return false, tierr
		}
		return false, errs.NewInternalDB("GetUserClusterInfo failed", err)
	}
	if clusterInfo.IsDedicate && !dedicatedAllowed {
		return false, errs.NewBadRequest("dedicated tier not support", nil)
	}
	if clusterInfo.Region == "" {
		if config.IsProd() {
			tilog.Error(ctx, "user cluster region info is empty", zap.Uint64("cluster_id", clusterId))
			return false, nil
		} else {
			return false, errs.NewInternalServerError("user cluster region info is empty", nil)
		}
	}
	region := clusterInfo.Region
	if !config.IsProd() {
		region = "us-west-2"
	}
	if !cfg.IsCentralMode() && clusterInfo.Region == cfg.AwsConf.Region {
		return false, nil
	}
	location := fmt.Sprintf("https://%s.%s%s", region, cfg.RegionalConf.GatewayRedirectAddr, c.Request.RequestURI)
	c.Redirect(http.StatusTemporaryRedirect, location)
	return true, nil
}

func isArrayBody(c *gin.Context) (bool, errs.Error) {
	var buf [1]byte
	n, err := c.Request.Body.Read(buf[:])
	if err != nil && err != io.EOF {
		return false, errs.NewBadRequest("Failed to read request body", err)
	}
	if n == 0 {
		return false, errs.NewBadRequest("Request body can not be empty", nil)
	}
	c.Request.Body = struct {
		io.Reader
		io.Closer
	}{
		io.MultiReader(bytes.NewReader(buf[:1]), c.Request.Body),
		c.Request.Body,
	}
	return buf[0] == '[', nil
}