package service import ( "context" httpRegion "dataservices/commons/http" dtoGateway "dataservices/internal/dto/dataapi_gateway" "encoding/json" "errors" "fmt" "net/http" "regexp" "strconv" "strings" "dataservices/commons/constants" commonErr "dataservices/commons/errs" "dataservices/commons/grpc" "dataservices/commons/metrics" "dataservices/commons/utils" dto "dataservices/internal/dto/dataapi" "dataservices/internal/entity" "dataservices/internal/repository" "github.com/prometheus/client_golang/prometheus" "github.com/tidbcloud/account/idl/pbgen/proto/account" "github.com/tidbcloud/tiinf/errs" "github.com/tidbcloud/tiinf/monitor/tilog" "go.uber.org/zap" ) var ( defaultSettings = fmt.Sprintf(`{"row_limit": %d, "timeout": %d}`, constants.EndpointDefaultRowlimit, constants.EndpointDefaultTimeoutMs) internalOnlyEndpointPath = map[string]bool{ endpointPathExecuteSQL: true, endpointPathExecuteQuery: true, endpointPathUserFeedback: true, } defaultEndpoint = entity.Endpoint{ Name: "untitled endpoint", Type: entity.EndpointTypeChat2Sql, Method: http.MethodGet, Settings: defaultSettings, Batch: false, Tag: "Default", RWMode: constants.SqlModeReadOnly, SqlTemplate: `/* Getting Started: Enter "USE {database};" before entering your SQL statements. Type "--your question" + Enter to try out AI-generated SQL queries Declare a parameter like "Where id = ${arg}". */ SELECT "Hello World"; `} defaultChatSettings = fmt.Sprintf(`{"row_limit": %d, "timeout": 60000}`, constants.EndpointDefaultRowlimit) defaultChat2DataEndpoint = entity.Endpoint{ Name: "/v1/chat2data", Type: entity.EndpointTypeChat2Data, Endpoint: "/v1/chat2data", Method: http.MethodPost, Settings: defaultChatSettings, Description: `"/v1/chat2data" is a RESTful interface that utilizes the POST method. It allows you to generate and execute SQL statements using artificial intelligence by providing instructions. The interface has four parameters, of which the first three are mandatory: cluster_id, database, and your instruction. The fourth parameter, tables, is optional and can be used if you can determine the content of the table you want to query.`, Position: 10000, } defaultChat2ChartEndpoint = entity.Endpoint{ Name: "Chat2Chart", Type: entity.EndpointTypeChat2Chart, Endpoint: "/chat2chart", Method: http.MethodPost, Settings: defaultChatSettings, } defaultChat2QuestionEndpoint = entity.Endpoint{ Name: "Chat2Question", Type: entity.EndpointTypeChat2Question, Endpoint: "/chat2question", Method: http.MethodPost, Settings: defaultChatSettings, } defaultChatEndpointList = []entity.Endpoint{ // order matters here: // data summary defaultChat2QueryCreateDataSummary, defaultChat2QueryListDataSummaries, defaultChat2QueryGetDataSummary, defaultChat2QueryUpdateDataSummary, defaultChat2QueryUpdateTableDescription, defaultChat2QueryUpdateColumnDescription, // knowledge base defaultChat2QueryCreateKnowledgeBase, defaultChat2QueryListKnowledgeBases, defaultChat2QueryGetKnowledgeBase, defaultChat2QueryUpdateKnowledgeBase, // knowledge base data defaultChat2QueryCreateDataItems, defaultChat2QueryListDataItems, defaultChat2QueryUpdateDataItems, defaultChat2QueryDeleteDataItems, // session defaultChat2QueryCreateSession, defaultChat2QueryListSessions, defaultChat2QueryGetSession, defaultChat2QueryUpdateSession, defaultChat2QueryResetSession, defaultChat2QuerySessionChat, // chat2data defaultChat2QueryChat2Data, // refine sql defaultChat2QueryRefineSQL, // suggest questions defaultChat2QuerySuggestQuestions, // v2 endpoints defaultChat2QueryCreateDataSummaryV2, defaultChat2QueryListDataSummariesV2, defaultChat2QueryChat2DataV2, defaultChat2QueryQueryJobDetail, // defaultChat2QueryExecuteSQL, defaultChat2QueryExecuteQuery, defaultChat2QueryUserFeedback, // always last: old chat2data defaultChat2DataEndpoint, } defaultChat2QuerySettings = fmt.Sprintf(`{"row_limit": %d, "timeout": 60000}`, constants.ExecSQLDefaultRowlimit) endpointPathCreateDataSummaries = "/v3/dataSummaries" endpointPathListDataSummaries = "/v3/dataSummaries" endpointPathGetDataSummary = "/v3/dataSummaries/{data_summary_id}" endpointPathUpdateDataSummary = "/v3/dataSummaries/{data_summary_id}" endpointPathUpdateTableDescription = "/v3/dataSummaries/{data_summary_id}/tables/{table_name}" endpointPathUpdateColumnDescription = "/v3/dataSummaries/{data_summary_id}/tables/{table_name}/columns" endpointPathSuggestQuestions = "/v3/suggestQuestions" endpointPathCreateKnowledgeBase = "/v3/knowledgeBases" endpointPathListKnowledgeBases = "/v3/knowledgeBases" endpointPathGetKnowledgeBase = "/v3/knowledgeBases/{knowledge_base_id}" endpointPathUpdateKnowledgeBase = "/v3/knowledgeBases/{knowledge_base_id}" endpointPathListDataItems = "/v3/knowledgeBases/{knowledge_base_id}/data" endpointPathCreateDataItems = "/v3/knowledgeBases/{knowledge_base_id}/data" endpointPathUpdateDataItems = "/v3/knowledgeBases/{knowledge_base_id}/data/{knowledge_data_id}" endpointPathDeleteDataItems = "/v3/knowledgeBases/{knowledge_base_id}/data/{knowledge_data_id}" endpointPathListSessions = "/v3/sessions" endpointPathCreateSessionContext = "/v3/sessions" endpointPathGetSessionContext = "/v3/sessions/{session_id}" endpointPathSessionChat = "/v3/sessions/{session_id}/chat2data" endpointPathUpdateSessionContext = "/v3/sessions/{session_id}" endpointPathResetSessionContext = "/v3/sessions/{session_id}/reset" endpointPathChat2Data = "/v3/chat2data" endpointPathRefineSQL = "/v3/refineSql" endpointPathCreateDataSummariesV2 = "/v2/dataSummaries" endpointPathListDataSummaryV2 = "/v2/dataSummaries" endpointPathChat2DataV2 = "/v2/chat2data" endpointPathQueryJobDetail = "/v2/jobs/{job_id}" endpointPathExecuteSQL = "/v2/executeSQL" endpointPathExecuteQuery = "/v2/executeQuery" endpointPathUserFeedback = "/v2/feedback" // data summary defaultChat2QueryCreateDataSummary = entity.Endpoint{ Name: endpointPathCreateDataSummaries, Type: entity.EndpointTypeChat2EDA, Endpoint: endpointPathCreateDataSummaries, Method: http.MethodPost, Settings: defaultChat2QuerySettings, Description: "This is an async API, create data summary analysis the database schema, column schema, explore the structure of your database with AI powered, the response contains a data_summary_id and job_id, use the /v2/jobs to query the job status, when the status changed from running to done, start to explore your dataset by using /v2/chat2data.", Position: 1, } defaultChat2QueryListDataSummaries = entity.Endpoint{ Name: endpointPathListDataSummaries, Type: entity.EndpointTypeChat2EDA, Endpoint: endpointPathListDataSummaries, Method: http.MethodGet, Settings: defaultChat2QuerySettings, Description: "List all data summaries", Position: 2, } defaultChat2QueryGetDataSummary = entity.Endpoint{ Name: endpointPathGetDataSummary, Type: entity.EndpointTypeChat2EDA, Endpoint: endpointPathGetDataSummary, Method: http.MethodGet, Settings: defaultChat2QuerySettings, Description: "Get data summary by cluster_id and database", Position: 3, } defaultChat2QueryUpdateDataSummary = entity.Endpoint{ Name: endpointPathUpdateDataSummary, Type: entity.EndpointTypeChat2EDA, Endpoint: endpointPathUpdateDataSummary, Method: http.MethodPut, Settings: defaultChat2QuerySettings, Description: "Update data summary description and default", Position: 4, } defaultChat2QueryUpdateTableDescription = entity.Endpoint{ Name: endpointPathUpdateTableDescription, Type: entity.EndpointTypeChat2EDA, Endpoint: endpointPathUpdateTableDescription, Method: http.MethodPut, Settings: defaultChat2QuerySettings, Description: "Update table description", Position: 5, } defaultChat2QueryUpdateColumnDescription = entity.Endpoint{ Name: endpointPathUpdateColumnDescription, Type: entity.EndpointTypeChat2EDA, Endpoint: endpointPathUpdateColumnDescription, Method: http.MethodPut, Settings: defaultChat2QuerySettings, Description: "Update column description", Position: 6, } // knowledge base defaultChat2QueryCreateKnowledgeBase = entity.Endpoint{ Name: endpointPathCreateKnowledgeBase, Type: entity.EndpointTypeChat2EDA, Endpoint: endpointPathCreateKnowledgeBase, Method: http.MethodPost, Settings: defaultChat2QuerySettings, Description: "Create knowledge base", Position: 10, } defaultChat2QueryListKnowledgeBases = entity.Endpoint{ Name: endpointPathListKnowledgeBases, Type: entity.EndpointTypeChat2EDA, Endpoint: endpointPathListKnowledgeBases, Method: http.MethodGet, Settings: defaultChat2QuerySettings, Description: "List all knowledge bases", Position: 11, } defaultChat2QueryGetKnowledgeBase = entity.Endpoint{ Name: endpointPathGetKnowledgeBase, Type: entity.EndpointTypeChat2EDA, Endpoint: endpointPathGetKnowledgeBase, Method: http.MethodGet, Settings: defaultChat2QuerySettings, Description: "Get knowledge base by knowledge_base_id", Position: 12, } defaultChat2QueryUpdateKnowledgeBase = entity.Endpoint{ Name: endpointPathUpdateKnowledgeBase, Type: entity.EndpointTypeChat2EDA, Endpoint: endpointPathUpdateKnowledgeBase, Method: http.MethodPut, Settings: defaultChat2QuerySettings, Description: "Update knowledge base by knowledge_base_id", Position: 13, } // knowledge base data defaultChat2QueryCreateDataItems = entity.Endpoint{ Name: endpointPathCreateDataItems, Type: entity.EndpointTypeChat2EDA, Endpoint: endpointPathCreateDataItems, Method: http.MethodPost, Settings: defaultChat2QuerySettings, Description: "Create knowledge base data", Position: 20, } defaultChat2QueryListDataItems = entity.Endpoint{ Name: endpointPathListDataItems, Type: entity.EndpointTypeChat2EDA, Endpoint: endpointPathListDataItems, Method: http.MethodGet, Settings: defaultChat2QuerySettings, Description: "List all knowledge base data by knowledge_base_id", Position: 21, } defaultChat2QueryUpdateDataItems = entity.Endpoint{ Name: endpointPathUpdateDataItems, Type: entity.EndpointTypeChat2EDA, Endpoint: endpointPathUpdateDataItems, Method: http.MethodPut, Settings: defaultChat2QuerySettings, Description: "Update knowledge base data by knowledge_data_id", Position: 22, } defaultChat2QueryDeleteDataItems = entity.Endpoint{ Name: endpointPathDeleteDataItems, Type: entity.EndpointTypeChat2EDA, Endpoint: endpointPathDeleteDataItems, Method: http.MethodDelete, Settings: defaultChat2QuerySettings, Description: "Delete knowledge base data by knowledge_data_id", Position: 23, } // session defaultChat2QueryCreateSession = entity.Endpoint{ Name: endpointPathCreateSessionContext, Type: entity.EndpointTypeChat2EDA, Endpoint: endpointPathCreateSessionContext, Method: http.MethodPost, Settings: defaultChat2QuerySettings, Description: "Create session context", Position: 30, } defaultChat2QueryListSessions = entity.Endpoint{ Name: endpointPathListSessions, Type: entity.EndpointTypeChat2EDA, Endpoint: endpointPathListSessions, Method: http.MethodGet, Settings: defaultChat2QuerySettings, Description: "Get all session contexts", Position: 31, } defaultChat2QueryGetSession = entity.Endpoint{ Name: endpointPathGetSessionContext, Type: entity.EndpointTypeChat2EDA, Endpoint: endpointPathGetSessionContext, Method: http.MethodGet, Settings: defaultChat2QuerySettings, Description: "Get session detail by session_id", Position: 32, } defaultChat2QueryUpdateSession = entity.Endpoint{ Name: endpointPathUpdateSessionContext, Type: entity.EndpointTypeChat2EDA, Endpoint: endpointPathUpdateSessionContext, Method: http.MethodPut, Settings: defaultChat2QuerySettings, Description: "Update session by session_id", Position: 33, } defaultChat2QueryResetSession = entity.Endpoint{ Name: endpointPathResetSessionContext, Type: entity.EndpointTypeChat2EDA, Endpoint: endpointPathResetSessionContext, Method: http.MethodPut, Settings: defaultChat2QuerySettings, Description: "Reset session by session_id", Position: 34, } defaultChat2QuerySessionChat = entity.Endpoint{ Name: endpointPathSessionChat, Type: entity.EndpointTypeChat2EDA, Endpoint: endpointPathSessionChat, Method: http.MethodPost, Settings: defaultChat2QuerySettings, Description: "This is an async API, chat with session by session_id", Position: 35, } // chat2data defaultChat2QueryChat2Data = entity.Endpoint{ Name: endpointPathChat2Data, Type: entity.EndpointTypeChat2EDA, Endpoint: endpointPathChat2Data, Method: http.MethodPost, Settings: defaultChat2QuerySettings, Description: "This is an async API, explore dataset by nature language described question. Chat2Data generate a SQL by your given question described in nature language, and return the result executed by the SQL, the response contains a job_id, use the /v2/jobs to query the job status, when the status changed from running to done, get generated sql and execute result from response.", Position: 40, } // sql refine defaultChat2QueryRefineSQL = entity.Endpoint{ Name: endpointPathRefineSQL, Type: entity.EndpointTypeChat2EDA, Endpoint: endpointPathRefineSQL, Method: http.MethodPost, Settings: defaultChat2QuerySettings, Description: "Refine SQL", Position: 50, } // suggest questions defaultChat2QuerySuggestQuestions = entity.Endpoint{ Name: endpointPathSuggestQuestions, Type: entity.EndpointTypeChat2EDA, Endpoint: endpointPathSuggestQuestions, Method: http.MethodPost, Settings: defaultChat2QuerySettings, Description: "Suggest questions", Position: 60, } // v2 endpoints defaultChat2QueryCreateDataSummaryV2 = entity.Endpoint{ Name: endpointPathCreateDataSummariesV2, Type: entity.EndpointTypeChat2EDA, Endpoint: endpointPathCreateDataSummariesV2, Method: http.MethodPost, Settings: defaultChat2QuerySettings, Description: "Create data summary analysis the database schema, column schema, explore the structure of your database with AI powered, the response contains a data_summary_id and job_id, use the /v2/jobs to query the job status, when the status changed from running to done, start to explore your dataset by using /v2/chat2data.", Position: 70, } defaultChat2QueryListDataSummariesV2 = entity.Endpoint{ Name: endpointPathListDataSummaryV2, Type: entity.EndpointTypeChat2EDA, Endpoint: endpointPathListDataSummaryV2, Method: http.MethodGet, Settings: defaultChat2QuerySettings, Description: "List all data summaries", Position: 71, } defaultChat2QueryChat2DataV2 = entity.Endpoint{ Name: endpointPathChat2DataV2, Type: entity.EndpointTypeChat2EDA, Endpoint: endpointPathChat2DataV2, Method: http.MethodPost, Settings: defaultChat2QuerySettings, Description: "Explore dataset by nature language described question. Chat2Data generate a SQL by your given question described in nature language, and return the result executed by the SQL, the response contains a job_id, use the /v2/jobs to query the job status, when the status changed from running to done, get generated sql and execute result from response.", Position: 72, } defaultChat2QueryQueryJobDetail = entity.Endpoint{ Name: endpointPathQueryJobDetail, Type: entity.EndpointTypeChat2EDA, Endpoint: endpointPathQueryJobDetail, Method: http.MethodGet, Settings: defaultChat2QuerySettings, Description: "Query the job detail of async job, the status indicates the job status, which can be init, running, done, failed.", Position: 73, } // execute sql defaultChat2QueryExecuteQuery = entity.Endpoint{ Name: endpointPathExecuteQuery, Type: entity.EndpointTypeChat2EDA, Endpoint: endpointPathExecuteQuery, Method: http.MethodPost, Settings: defaultChat2QuerySettings, Description: "Execute SQL query by given cluster_id and database, returns the execution result of the query.", Position: 80, } defaultChat2QueryUserFeedback = entity.Endpoint{ Name: endpointPathUserFeedback, Type: entity.EndpointTypeChat2EDA, Endpoint: endpointPathUserFeedback, Method: http.MethodPost, Settings: defaultChat2QuerySettings, Description: "User feedback for the chat2data response", Position: 81, } defaultExecuteQueryEndpoint = entity.Endpoint{ Type: entity.EndpointTypeSystemData, Name: "/system/query", Endpoint: "/system/query", Method: http.MethodPost, Settings: defaultSettings, Args: `[{"name":"cluster_id","type":"string","required":0,"default":"%s","description":"TiDB Cluster to use"},{"name":"database","type":"string","required":0,"default":"%s","description":"Database within the cluster"},{"name":"sql","type":"string","required":1,"description":"SQL statement"}]`, Batch: false, Tag: "System/Data", RWMode: constants.SqlModeINVALID, Description: `This API is used for executing any SQL statement for specified cluster and database.`, } defaultSystemDataEndpoints = []entity.Endpoint{defaultExecuteQueryEndpoint} defaultSystemDataEndpointMap = map[string]entity.Endpoint{} validMethod = map[string]bool{ http.MethodGet: true, http.MethodPost: true, http.MethodPut: true, http.MethodDelete: true, "B-" + http.MethodPost: true, "B-" + http.MethodPut: true, "B-" + http.MethodDelete: true, } validDataType = map[string]bool{ "string": true, "number": true, "boolean": true, "integer": true, "array": true, } validArrayItemType = map[string]bool{ "string": true, "number": true, "integer": true, } endpointPathRegexp = regexp.MustCompile(`^(\/\w+|\/{[A-Za-z_]\w+})+$`) endpointNameRegexp = regexp.MustCompile(`^[A-Za-z0-9 \_\-\/\[\]]*$`) // AGE defaultAutoGenerateMaxRowExcludeGET = 1000 defaultAutoGenerateMaxRowGET = 1000 defaultAutoGenerateVectorSearchMaxRow = 200 defaultAutoGenerateBatchMaxRow = 0 defaultMatchThreshold = "10" defaultMatchCount = "10" PK = "primary key" UK = "unique key" suffixBatch = "/batch" ErrInvalidEndpointPathOrMethod = errs.NewBadRequest("invalid endpoint path or method", nil) ) func init() { for _, systemEndpoint := range defaultSystemDataEndpoints { key := systemEndpoint.Key() defaultSystemDataEndpointMap[key] = systemEndpoint } } func IsEDAInternalUser(ctx context.Context, email string, orgID uint64) bool { resp, err := grpc.IsHitExp(ctx, &account.HitExpReq{ ExpName: constants.ExpWhitelistChat2QueryBetaAPI, Email: email, OrgId: orgID, }) if err != nil { tilog.Error(ctx, "IsEDAInternalUser failed", zap.Error(err)) return false } return resp.IsHit } func GetDataAppEndpointsDeployStat(ctx context.Context, appId uint64) (online, total int, err error) { list, err := repository.ListEndpointsByAppIDs([]uint64{appId}) if err != nil { return 0, 0, err } for _, e := range list { if e.Label > 0 { online += 1 } } return online, len(list), nil } func GetEndpointsNameByIdStr(ctx context.Context, endpoints []string) (map[string]string, error) { ids := make([]uint64, 0, len(endpoints)) for _, s := range endpoints { id, err := strconv.ParseUint(s, 10, 64) if err != nil { tilog.Error(ctx, "GetEndpointsNameByIdStr invalid endpoints id str", zap.String("input", s)) } else { ids = append(ids, id) } } list, err := repository.ListEndpointByIDs(ctx, ids) if err != nil { return nil, err } res := make(map[string]string) for _, e := range list { res[fmt.Sprintf("%d", e.ID)] = fmt.Sprintf("%s (%s)", e.Name, e.Method) } for _, s := range endpoints { _, ok := res[s] if !ok { res[s] = "unknown" } } return res, nil } func ListEndpointsByAppIDs(ctx context.Context, appIds []uint64) ([]*entity.EndpointV1, error) { return repository.ListEndpointsByAppIDs(appIds) } func IsOnlineEndpoint(ctx context.Context, endpointId uint64) (bool, error) { return repository.IsOnlineEndpoint(endpointId) } func ListEndpoints(ctx context.Context, appId uint64, endpointTypes ...string) ([]*entity.Endpoint, error) { return repository.ListEndpoints(ctx, appId, endpointTypes...) } func ListPublishedEndpointByIDs(ctx context.Context, endpointIDs []uint64) (map[uint64]*entity.EndpointPublish, error) { endpointList, err := repository.ListPublishedEndpointByIDs(ctx, endpointIDs) if err != nil { return nil, err } result := make(map[uint64]*entity.EndpointPublish, len(endpointList)) for _, endpoint := range endpointList { result[endpoint.EndpointId] = endpoint } return result, nil } func isEndpointParamValid(ctx context.Context, appInfo *entity.DataApp, endpoint *entity.Endpoint, param *entity.CreateEndpointParam, selectedKeys map[string]interface{}) (eerr error) { appId := appInfo.ID // chat2query/system endpoint is not allowed to be create/update var allowedKeys map[string]bool if appInfo.AppType == entity.DataAppTypeChat2Query { allowedKeys = map[string]bool{"settings": true} } else if endpoint.IsSystemDataEndpoint() { allowedKeys = map[string]bool{"settings": true, "sql_template": true, "cluster_id": true, "database": true} } if allowedKeys != nil { for key := range selectedKeys { if !allowedKeys[key] { return commonErr.ErrOperationNotAllowed } } } if len(selectedKeys) == 0 || selectedKeys["endpoint"] != nil || selectedKeys["method"] != nil { method, path := param.Method, param.Endpoint if param.EndpointID > 0 && len(selectedKeys) != 0 && (selectedKeys["endpoint"] == nil || selectedKeys["method"] == nil) { var err error if endpoint == nil { endpoint, err = repository.GetEndpoint(param.EndpointID) } if err != nil { return err } if selectedKeys["endpoint"] == nil { path = endpoint.Endpoint } if selectedKeys["method"] == nil { method = endpoint.Method } } if err := checkEndpointDuplicated(ctx, appId, param.EndpointID, method, path); err != nil { if errs.IsError(err, errs.ErrInternalServerError) { return err } // ignore endpoint path and method when check failed for update defer func() { if eerr == nil { if param.EndpointID == 0 { // Create eerr = err } else { // Update param.Endpoint = endpoint.Endpoint param.Method = endpoint.Method eerr = errs.NewError(ErrInvalidEndpointPathOrMethod, err.Error(), err) } } }() } if param.Batch { if err := VerifyBatchOn(ctx, entity.VerifyBatch{Method: param.Method, Sql: param.SqlTemplate}); err != nil { return err } } } if len(selectedKeys) == 0 || selectedKeys["description"] != nil { if err := validateEndpointDescription(ctx, param.Description); err != nil { return err } } if len(selectedKeys) == 0 || selectedKeys["name"] != nil { if param.Name == "" { return errs.NewBadRequest("Endpoint name is required.", nil) } if !endpointNameRegexp.MatchString(param.Name) { return errs.NewBadRequest("Endpoint name can only include letters, numbers, hyphens(-), underscores(_), slashes(/) and space", nil) } } if len(selectedKeys) == 0 || selectedKeys["settings"] != nil { if endpoint.IsSystemDataEndpoint() { if err := checkSystemDataEndpointSettingsValid(param.Settings); err != nil { return err } } var maxTimeout int if appInfo.AppType == entity.DataAppTypeChat2Query { maxTimeout = constants.Chat2queryMaxTimeoutMs } else { maxTimeout = constants.ExecSQLMaxTimeoutMs } if param.Settings.TimeOut < 1 || param.Settings.TimeOut > maxTimeout { return errs.NewBadRequest("Invalid timeout value. Value should be between 1 and "+strconv.Itoa(maxTimeout)+".", nil) } if !param.Batch && param.Settings.RowsLimit < 1 || param.Settings.RowsLimit > constants.ExecSQLMaxRowlimit { return errs.NewBadRequest("Invalid max rows value. Value should be between 1 and "+strconv.Itoa(constants.ExecSQLMaxRowlimit)+".", nil) } if param.Settings.EnablePagination == 1 && param.Method != http.MethodGet { return errs.NewBadRequest("Sorry, Only GET requests support pagination.", nil) } // if param.Batch && param.Settings.RowsLimit > constants.ExeSQLBatchMaxRowlimit { // return errs.NewBadRequest("Invalid arguments of batch row_limit.", nil) // } if param.Settings.CacheEnabled != 0 && param.Settings.CacheEnabled != 1 { return errs.NewBadRequest("Invalid arguments of cache_enabled, it should be either 0 or 1.", nil) } if !param.Settings.IsCacheEnabled() && param.Settings.CacheTTL > 0 { return errs.NewBadRequest("Invalid arguments of cache_ttl, it should not be set if cache is disabled.", nil) } if param.Settings.IsCacheEnabled() && param.Method != http.MethodGet { return errs.NewBadRequest("Invalid arguments of cache_enabled, it should only be enabled on GET endpoint.", nil) } if param.Settings.IsCacheEnabled() && !param.Settings.TTLValid() { return errs.NewBadRequest(fmt.Sprintf("Invalid arguments of cache_ttl, it should not be in range of [%d, %d].", entity.MinCacheTTL, entity.MaxCacheTTL), nil) } } if len(selectedKeys) == 0 || selectedKeys["args"] != nil { if err := checkEndpointArg(param.Endpoint, param.Args, false); err != nil { return err } } return nil } func checkEndpointDuplicated(ctx context.Context, appId, endpointId uint64, method, path string) error { if path == "" || method == "" { return nil } if path != "" && !endpointPathRegexp.MatchString(path) { return errs.NewBadRequest("Invalid endpoint path", nil) } if method != "" && !validMethod[method] { return errs.NewBadRequest("Invalid endpoint method", nil) } endpoints, err := repository.ListEndpoints(ctx, appId) if err != nil { return err } pathTrie := utils.NewPathTrie() for _, endpoint := range endpoints { if endpointId == endpoint.ID || endpoint.Method == "" || endpoint.Endpoint == "" { continue } if err := pathTrie.Insert(endpoint.Method, endpoint.Endpoint, nil); err != nil { return err } } if err := pathTrie.Insert(method, path, nil); err != nil { return err } // exists, _, err := repository.IsExistEndpoint(appId, endpointId, path, method) // if err != nil { // return errs.NewInternalDB("Check endpoint failed", err) // } // if exists { // return errs.NewBadRequest(fmt.Sprintf(`Endpoint with path "%s:%s" already exists under the dataapp`, method, path), nil) // } return nil } func SaveEndpoint(ctx context.Context, appId uint64, param entity.CreateEndpointParam, createdBy string, selectedKeys map[string]interface{}) (res *entity.Endpoint, err error) { defer recordCounterMetric(metrics.EndpointSaveCounter, err) // chat2query app only support modify settings appInfo, err := repository.GetDataApp(appId) if err != nil { return nil, err } var endpoint *entity.Endpoint if param.EndpointID > 0 { endpoint, err = repository.GetEndpoint(param.EndpointID) if err != nil { return nil, err } } param.Method = strings.ToUpper(param.Method) if verr := isEndpointParamValid(ctx, appInfo, endpoint, ¶m, selectedKeys); verr != nil { if !errs.IsError(verr, ErrInvalidEndpointPathOrMethod) { return nil, verr } defer func() { if err == nil { err = verr } }() } // Check for any changes, if not, return directly to avoid version updates if endpoint != nil { if param.EqualEndpoint(endpoint, selectedKeys) { tilog.Info(ctx, "The endpoint has not changed, ignoring this update", zap.Uint64("endpoint_id", param.EndpointID)) if endpoint.IsSystemDataEndpoint() { saveUserEndpointPlayground(ctx, endpoint, entity.EndpointPlaygroundContent{ ClusterID: param.ClusterId, Database: param.Database, SqlTemplate: param.SqlTemplate, }, createdBy) } return endpoint, nil } // for the batch endpoint, the modification of the row limit field is ignored if endpoint.IsBatch() { param.Settings.RowsLimit = endpoint.GetSettings().RowsLimit } } else { // ignore the row limit settings when create batch endpoint if param.Batch { param.Settings.RowsLimit = 0 } } settings, _ := json.Marshal(param.Settings) var args []byte if param.Args != nil { args, _ = json.Marshal(param.Args) } if endpoint == nil { endpoint = new(entity.Endpoint) endpoint.CreatedBy = createdBy endpoint.CreatedAt = utils.Now() endpoint.Type = entity.EndpointTypeChat2Sql } endpoint.DataappId = appId endpoint.Name = param.Name endpoint.Description = param.Description endpoint.Endpoint = param.Endpoint endpoint.Method = param.Method if !endpoint.IsSystemDataEndpoint() { endpoint.SqlTemplate = param.SqlTemplate endpoint.ClusterId = param.ClusterId endpoint.Database = param.Database } endpoint.Args = string(args) endpoint.Settings = string(settings) endpoint.UpdatedBy = createdBy endpoint.RWMode = getSqlRwMode(ctx, param.SqlTemplate) endpoint.Vid = param.Vid endpoint.Tag = param.Tag endpoint.Batch = param.Batch if endpoint.ID == 0 { if endpoint.Name == "" { endpoint.Name = defaultEndpoint.Name } if endpoint.Method == "" { endpoint.Method = defaultEndpoint.Method } if param.Settings.TimeOut <= 0 && param.Settings.RowsLimit <= 0 { endpoint.Settings = defaultEndpoint.Settings } if endpoint.SqlTemplate == "" { endpoint.SqlTemplate = defaultEndpoint.SqlTemplate } if endpoint.Tag == "" { endpoint.Tag = defaultEndpoint.Tag } endpoint.Batch = defaultEndpoint.Batch } err = repository.Transaction(ctx, func(ctx context.Context) error { res, err = repository.CreateEndpoint(ctx, endpoint, selectedKeys) if err != nil { return err } if endpoint.IsSystemDataEndpoint() { saveUserEndpointPlayground(ctx, endpoint, entity.EndpointPlaygroundContent{ ClusterID: param.ClusterId, Database: param.Database, SqlTemplate: param.SqlTemplate, }, createdBy) } if appInfo.AppType == entity.DataAppTypeChat2Query { endpoint, err = repository.GetEndpointWithCtx(ctx, endpoint.ID) if err != nil { return err } publishedVersion, err := getLastPublishedVersion(endpoint.ID) if err != nil { return err } version := getReleaseVersion(publishedVersion) err = repository.CreateOnlineEndpoint(ctx, endpoint, version, createdBy) if err != nil { return err } } // Auto trigger deployment when creating system data endpoint if endpoint.IsSystemDataEndpoint() && param.EndpointID == 0 { if _, err = TriggerEndpointDeployFromUI(ctx, appId, endpoint.ID, createdBy, "Deploy Endpoint: "+endpoint.Endpoint); err != nil { return err } } // push sync endpoint event if err = publishEndpointEvent(ctx, endpoint.ID, createdBy, dto.EndpointActionSave); err != nil { return err } return nil }) return res, err } func DumpEndpoint(ctx context.Context, param *entity.Endpoint, createdBy string) (res *entity.Endpoint, err error) { defer recordCounterMetric(metrics.EndpointDumpCounter, err) if err := checkEndpointDuplicated(ctx, param.DataappId, 0, param.Method, param.Endpoint); err != nil { return nil, err } endpoint := *param endpoint.ID = 0 endpoint.CreatedBy = createdBy endpoint.UpdatedBy = createdBy endpoint.CreatedAt = utils.Now() endpoint.UpdatedAt = utils.Now() endpoint.Vid = 0 err = repository.Transaction(ctx, func(ctx context.Context) error { res, err = repository.CreateEndpoint(ctx, &endpoint, nil) if err != nil { return err } // push sync endpoint event if err = publishEndpointEvent(ctx, res.ID, createdBy, dto.EndpointActionSave); err != nil { return err } return nil }) return } func DeleteEndpoint(ctx context.Context, dataAppID, endpointId uint64, createdBy string) (err error) { if err := CheckDataAppCanDeploy(ctx, dataAppID); err != nil { return err } defer recordCounterMetric(metrics.EndpointDeleteCounter, err) endpoint, err := repository.GetEndpoint(endpointId) if err != nil { return err } err = repository.Transaction(ctx, func(ctx context.Context) error { dataApp, err := repository.GetDataApp(dataAppID) if err != nil { return err } if dataApp.AppType == entity.DataAppTypeChat2Query { return errs.NewBadRequest("Delete endpoint is not allowed for chat2query data app.", nil) } if err = repository.DeleteEndpoint(ctx, endpointId, createdBy); err != nil { return err } // push delete endpoint event if err = publishEndpointEvent(ctx, endpointId, createdBy, dto.EndpointActionDelete); err != nil { return err } isOnline, err := IsOnlineEndpoint(ctx, endpointId) if err != nil { return err } // endpoints of chat2query is read only in the UI if isOnline { // push delete endpoint deployment message := "Deployment for deleting endpoint" if endpoint.IsSystemDataEndpoint() { message = "Deployment for removing endpoint" } _, err = TriggerEndpointDeleteFromUI(ctx, dataAppID, endpointId, createdBy, message) if err != nil { return err } } return nil }) if err != nil { return err } tilog.Info(ctx, "DeleteEndpoint success", zap.Uint64("endpoint_id", endpointId), zap.String("user", createdBy)) return nil } func GetEndpoint(ctx context.Context, endpointId uint64) (res *entity.Endpoint, err error) { defer recordCounterMetric(metrics.EndpointGetCounter, err) return repository.GetEndpoint(endpointId) } func HasEndpointModified(ctx context.Context, endpoint *entity.Endpoint) (bool, error) { publishedEndpoint, err := repository.GetPublishedEndpoint(endpoint.ID) if err != nil { if errors.Is(err, errs.ErrNotFound) { return true, nil } return false, err } return !endpoint.EqualPublished(publishedEndpoint), nil } func recordCounterMetric(counter *prometheus.CounterVec, err error) { code := "0" if err != nil { if tiErr, ok := errs.AsTiError(err); ok { code = strconv.FormatInt(tiErr.BizCode(), 10) } else { code = "-1" } } counter.WithLabelValues(code).Inc() } func recordFailedCounterMetric(counter *prometheus.CounterVec, err error, lvs ...string) { if err == nil { return } if len(lvs) > 0 { counter.WithLabelValues(lvs...).Inc() } else { counter.With(nil).Inc() } } func CreateAutoGenerateEndpoints(ctx context.Context, appId, projectId uint64, clusterIdStr string, token, createdBy string, in *entity.AutoGenerateEndpoint) ([]*dto.EndpointFormat, error) { if len(in.Operations) == 0 && !in.VectorSearchOperations.IsEnabled { tilog.Info(ctx, "no operation", zap.String("user", createdBy)) return nil, nil } clusterId, err := strconv.ParseUint(clusterIdStr, 10, 64) if err != nil { return nil, err } clusterInfo, err := grpc.GetClusterWithoutPermCheck(ctx, clusterId) if err != nil { return nil, err } // Global fast-fail keeps unsupported AliCloud endpoint-generation flows out of regional control. if err := GuardRegionalCapabilityForCluster(clusterInfo, RegionalCapabilityAutoGenerateRoute); err != nil { return nil, err } endpointList := make([]*dto.EndpointFormat, 0) endpointListRaw := make([]entity.CreateEndpointParam, 0) // basic operation if len(in.Operations) != 0 { //get table meta for param sqlMeta := "select column_name,column_default,is_nullable,data_type,column_key,extra from information_schema.columns where table_schema='" + in.DB + "' and table_name ='" + in.Table + "';" metaRes, err := httpRegion.ExecSql(ctx, in.Region, &dtoGateway.ExecSqlReq{ ProjectId: projectId, ClusterId: clusterId, Database: in.DB, Sql: sqlMeta, Credential: dtoGateway.Credential{Type: dtoGateway.TypeCredentialAuth0, Key: token, AccountID: createdBy}, }) if err != nil { return nil, err } // select key keys := selectKey(metaRes.Data.Rows) for _, operation := range in.Operations { switch operation { case entity.AutoGenerateOperationGET: // generate path name, endpointPath, err := generateNameAndPath(ctx, in.Table, entity.AutoGenerateOperationGET, appId, false) if err != nil { tilog.Error(ctx, "verify if path exist failed!") return nil, err } sql, argsList := genSelectSQL(ctx, in.DB, in.Table, keys) getEndpointParam := entity.CreateEndpointParam{ Name: name, Endpoint: endpointPath, Method: entity.AutoGenerateOperationGET, Database: in.DB, SqlTemplate: sql, Settings: entity.EndpointSettings{TimeOut: in.Configuration.TimeOut, RowsLimit: defaultAutoGenerateMaxRowGET}, Args: argsList, Tag: in.Configuration.Tag, Batch: false, ClusterId: clusterId, } endpointListRaw = append(endpointListRaw, getEndpointParam) case entity.AutoGenerateOperationPOST, entity.AutoGenerateOperationBatchPOST: // generate path name, endpointPath, err := generateNameAndPath(ctx, in.Table, entity.AutoGenerateOperationPOST, appId, false) if err != nil { tilog.Error(ctx, "verify if path exist failed!") return nil, err } sql, argsList := genInsertSQL(ctx, in.DB, in.Table, metaRes.Data.Rows) postEndpointParam := entity.CreateEndpointParam{ Name: name, Endpoint: endpointPath, Method: entity.AutoGenerateOperationPOST, Database: in.DB, SqlTemplate: sql, Settings: entity.EndpointSettings{TimeOut: in.Configuration.TimeOut, RowsLimit: defaultAutoGenerateMaxRowExcludeGET}, Args: argsList, Tag: in.Configuration.Tag, Batch: false, ClusterId: clusterId, } if operation == entity.AutoGenerateOperationPOST { endpointListRaw = append(endpointListRaw, postEndpointParam) } else { postEndpointParam.Settings.RowsLimit = defaultAutoGenerateBatchMaxRow postEndpointParam.Batch = true postEndpointParam.Name, postEndpointParam.Endpoint, err = generateNameAndPath(ctx, in.Table+suffixBatch, entity.AutoGenerateOperationPOST, appId, false) if err != nil { return nil, err } endpointListRaw = append(endpointListRaw, postEndpointParam) } case entity.AutoGenerateOperationPUT, entity.AutoGenerateOperationBatchPUT: // generate path name, endpointPath, err := generateNameAndPath(ctx, in.Table, entity.AutoGenerateOperationPUT, appId, false) if err != nil { tilog.Error(ctx, "verify if path exist failed!") return nil, err } sql, argsList := genUpdateSQL(ctx, in.DB, in.Table, keys, metaRes.Data.Rows) putEndpointParam := entity.CreateEndpointParam{ Name: name, Endpoint: endpointPath, Method: entity.AutoGenerateOperationPUT, Database: in.DB, SqlTemplate: sql, Settings: entity.EndpointSettings{TimeOut: in.Configuration.TimeOut, RowsLimit: defaultAutoGenerateMaxRowExcludeGET}, Args: argsList, Tag: in.Configuration.Tag, Batch: false, ClusterId: clusterId, } if operation == entity.AutoGenerateOperationPUT { endpointListRaw = append(endpointListRaw, putEndpointParam) } else { putEndpointParam.Settings.RowsLimit = defaultAutoGenerateBatchMaxRow putEndpointParam.Batch = true putEndpointParam.Name, putEndpointParam.Endpoint, err = generateNameAndPath(ctx, in.Table+suffixBatch, entity.AutoGenerateOperationPUT, appId, false) if err != nil { return nil, err } endpointListRaw = append(endpointListRaw, putEndpointParam) } case entity.AutoGenerateOperationDELETE: // generate path name, endpointPath, err := generateNameAndPath(ctx, in.Table, entity.AutoGenerateOperationDELETE, appId, false) if err != nil { tilog.Error(ctx, "verify if path exist failed!") return nil, err } sql, argsList := genDeleteSQL(ctx, in.DB, in.Table, keys) deleteEndpointParam := entity.CreateEndpointParam{ Name: name, Endpoint: endpointPath, Method: entity.AutoGenerateOperationDELETE, Database: in.DB, SqlTemplate: sql, Settings: entity.EndpointSettings{TimeOut: in.Configuration.TimeOut, RowsLimit: defaultAutoGenerateMaxRowExcludeGET}, Args: argsList, Tag: in.Configuration.Tag, Batch: false, ClusterId: clusterId, } endpointListRaw = append(endpointListRaw, deleteEndpointParam) default: tilog.Error(ctx, "not support operation!", zap.String("user", createdBy)) } } } //vector search operation if in.VectorSearchOperations != nil && in.VectorSearchOperations.IsEnabled { name, endpointPath, err := generateNameAndPath(ctx, in.Table, entity.AutoGenerateOperationPOST, appId, true) if err != nil { tilog.Error(ctx, "verify if path exist failed!") return nil, err } argsList := createVectorSearchParams() sql := genVectorSearchSQL(ctx, in.DB, in.Table, in.VectorSearchOperations.Field, in.VectorSearchOperations.DistanceFunc) vectorSearchEndpointParam := entity.CreateEndpointParam{ Name: name, Endpoint: endpointPath, Method: entity.AutoGenerateOperationPOST, Database: in.DB, SqlTemplate: sql, Settings: entity.EndpointSettings{TimeOut: in.Configuration.TimeOut, RowsLimit: defaultAutoGenerateVectorSearchMaxRow}, Args: argsList, Tag: in.Configuration.Tag, Batch: false, ClusterId: clusterId, } endpointListRaw = append(endpointListRaw, vectorSearchEndpointParam) } //batch save endpoints endpointList, err = SaveBatchAutoGenerateEndpointsDraft(ctx, appId, endpointListRaw, createdBy, endpointList, in.Configuration.IsDeploy) if err != nil { tilog.Error(ctx, "save batch endpoint failed!", zap.Error(err)) return endpointList, err } // trigger deploy if in.Configuration.IsDeploy { endpointIds := make([]uint64, 0) for _, v := range endpointList { endpointIds = append(endpointIds, v.ID) } _, err := TriggerAutoGenerateDeployFromUI(ctx, appId, endpointIds, entity.GenSnapshotTypeEndpointDeploy, createdBy) if err != nil { tilog.Error(ctx, "auto generate deploy failed!", zap.Error(err), zap.Any("endpointIds", endpointIds), zap.String("user", createdBy)) return endpointList, err } } return endpointList, nil } func createVectorSearchParams() []entity.EndpointArgs { argsList := make([]entity.EndpointArgs, 0) argsList = append(argsList, entity.EndpointArgs{Name: "query_vector", Type: entity.EndpointArgTypeString, Required: 1, Description: "The variable obtained by converting the embedding vector into a VECTOR type.", IsPathParameter: false}) argsList = append(argsList, entity.EndpointArgs{Name: "match_threshold", Type: entity.EndpointArgTypeNumber, Required: 0, Default: defaultMatchThreshold, Description: "The distance threshold used for similarity filtering.", IsPathParameter: false}) argsList = append(argsList, entity.EndpointArgs{Name: "match_count", Type: entity.EndpointArgTypeInteger, Required: 0, Default: defaultMatchCount, Description: "The desired maximum number of match results.", IsPathParameter: false}) return argsList } func generateNameAndPath(ctx context.Context, table, method string, appId uint64, ifVector bool) (string, string, error) { endpointPath := "/" + table if ifVector { endpointPath = endpointPath + "/vector_search" } isPathExist, dumpE, err := repository.IsExistEndpoint(appId, 0, endpointPath, method) if err != nil { return "", "", err } if isPathExist { endpointPath = dumpE.Endpoint + "_dump_" + utils.RandStr(6) } name := endpointPath return name, endpointPath, nil } func SaveBatchAutoGenerateEndpointsDraft(ctx context.Context, appId uint64, params []entity.CreateEndpointParam, createdBy string, endpointFormatList []*dto.EndpointFormat, isDeploy bool) ([]*dto.EndpointFormat, error) { var err error defer recordCounterMetric(metrics.EndpointSaveCounter, err) appInfo, err := repository.GetDataApp(appId) if err != nil { return nil, err } endpointList := make([]*entity.Endpoint, 0) for _, param := range params { param.Method = strings.ToUpper(param.Method) if err := isEndpointParamValid(ctx, appInfo, nil, ¶m, nil); err != nil { return nil, err } settings, _ := json.Marshal(param.Settings) var args []byte if param.Args != nil { args, _ = json.Marshal(param.Args) } rwMode := getSqlRwMode(ctx, param.SqlTemplate) endpointType := entity.EndpointTypeChat2Sql endpoint := &entity.Endpoint{ ID: param.EndpointID, DataappId: appId, Name: param.Name, Description: param.Description, Endpoint: param.Endpoint, Method: param.Method, ClusterId: param.ClusterId, Database: param.Database, SqlTemplate: param.SqlTemplate, Settings: string(settings), Args: string(args), CreatedBy: createdBy, UpdatedBy: createdBy, Type: endpointType, RWMode: rwMode, Tag: param.Tag, Batch: param.Batch, Vid: 1, } endpointList = append(endpointList, endpoint) } err = repository.Transaction(ctx, func(ctx context.Context) error { res, err := repository.CreateAutoGenerateEndpoints(ctx, endpointList) if err != nil { return err } for _, endpoint := range res { endpointFormat := ConvertEndpoint(ctx, endpoint, nil) if isDeploy { endpointFormat.HasModified = false } else { endpointFormat.HasModified = true } endpointFormatList = append(endpointFormatList, endpointFormat) // push sync endpoint event if err = publishEndpointEvent(ctx, endpoint.ID, createdBy, dto.EndpointActionSave); err != nil { return err } } return nil }) return endpointFormatList, err } func selectKey(rows []entity.Row) []entity.EndpointArgs { keysPri := make([]entity.EndpointArgs, 0) keysUniq := make([]entity.EndpointArgs, 0) for _, v := range rows { for k1, v1 := range v { if k1 == "column_key" && v1 == "PRI" { columnType := getArgsType(v["data_type"]) keysPri = append(keysPri, entity.EndpointArgs{Name: v["column_name"], Type: columnType, Description: PK}) } else if k1 == "column_key" && v1 == "UNI" { columnType := getArgsType(v["data_type"]) keysUniq = append(keysUniq, entity.EndpointArgs{Name: v["column_name"], Type: columnType, Description: UK}) } } } if len(keysPri) > 0 { return keysPri } else if len(keysUniq) > 0 { return keysUniq } else { return nil } } func getArgsType(v string) string { columnType := entity.EndpointArgTypeString if utils.ArrayIn(strings.ToLower(v), entity.ENDPOINTARGSNUMBER) { columnType = entity.EndpointArgTypeNumber } return columnType } func isNumberBegin(ctx context.Context, s string) (bool, error) { regexpNumber := `^[0-9]+\w*$` beginNumber, err := regexp.MatchString(regexpNumber, s) if err != nil { tilog.Error(ctx, "regexp number begin failed", zap.Error(err)) return false, err } return beginNumber, err } func genSelectSQL(ctx context.Context, db, table string, keys []entity.EndpointArgs) (string, []entity.EndpointArgs) { sql := "" argsList := make([]entity.EndpointArgs, 0) if len(keys) == 0 { sql = "USE " + db + ";\nSELECT * FROM `" + table + "`;" } else { ele := "" fName := "" for _, pri := range keys { // whether number begin,if number begin, param =${p..} beginNumber, _ := isNumberBegin(ctx, pri.Name) if beginNumber { ele = ele + "`" + pri.Name + "` = ${p" + pri.Name + "} and " fName = "p" + pri.Name } else { ele = ele + "`" + pri.Name + "` = ${" + pri.Name + "} and " fName = pri.Name } sql = "USE " + db + ";\nSELECT * FROM `" + table + "` WHERE " + ele[:len(ele)-4] + ";" args := entity.EndpointArgs{ Name: fName, Type: pri.Type, Required: 1, } argsList = append(argsList, args) } } return sql, argsList } func genInsertSQL(ctx context.Context, db, table string, meta []entity.Row) (string, []entity.EndpointArgs) { field := "" values := "" sql := "" fName := "" argsList := make([]entity.EndpointArgs, 0) for _, v := range meta { if v["extra"] == "" && v["column_default"] != "CURRENT_TIMESTAMP" { field = field + "`" + v["column_name"] + "`," beginNumber, _ := isNumberBegin(ctx, v["column_name"]) if beginNumber { values = values + "${p" + v["column_name"] + "}," fName = "p" + v["column_name"] } else { values = values + "${" + v["column_name"] + "}," fName = v["column_name"] } if v["is_nullable"] == "YES" { args := entity.EndpointArgs{ Name: fName, Type: getArgsType(v["data_type"]), Required: 0, } argsList = append(argsList, args) } else { args := entity.EndpointArgs{ Name: fName, Type: getArgsType(v["data_type"]), Required: 1, } argsList = append(argsList, args) } } } sql = "USE " + db + ";\nINSERT INTO `" + table + "` (" + field[:len(field)-1] + ") VALUES(" + values[:len(values)-1] + ");" return sql, argsList } func genUpdateSQL(ctx context.Context, db, table string, keys []entity.EndpointArgs, meta []entity.Row) (string, []entity.EndpointArgs) { sql := "" field := "" where := "" fName := "" argsList := make([]entity.EndpointArgs, 0) keyList := make([]string, 0, len(keys)) if len(keys) == 0 { where = "1=0 and " } else { for _, v := range keys { beginNumber, _ := isNumberBegin(ctx, v.Name) if beginNumber { where = where + "`" + v.Name + "` = ${p" + v.Name + "} and " fName = "p" + v.Name } else { where = where + "`" + v.Name + "` = ${" + v.Name + "} and " fName = v.Name } args := entity.EndpointArgs{ Name: fName, Type: v.Type, Required: 1, } argsList = append(argsList, args) // Parameters for the UPDATE statement MUST pertain to all columns in the table except primary, composite key columns. if v.Description != UK { keyList = append(keyList, v.Name) } } } for _, v := range meta { //only PK doesn't exist in update content if v["extra"] == "" && !utils.ArrayIn(v["column_name"], keyList) && v["column_default"] != "CURRENT_TIMESTAMP" { beginNumber, _ := isNumberBegin(ctx, v["column_name"]) if beginNumber { if getArgsType(v["data_type"]) == entity.EndpointArgTypeNumber { field = field + "`" + v["column_name"] + "` = IF(${p" + v["column_name"] + "}<>0,${p" + v["column_name"] + "},`" + v["column_name"] + "`),\n" } else { field = field + "`" + v["column_name"] + "` = IF(length(${p" + v["column_name"] + "})>0,${p" + v["column_name"] + "},`" + v["column_name"] + "`),\n" } fName = "p" + v["column_name"] } else { if getArgsType(v["data_type"]) == entity.EndpointArgTypeNumber { field = field + "`" + v["column_name"] + "` = IF(${" + v["column_name"] + "}<>0,${" + v["column_name"] + "},`" + v["column_name"] + "`),\n" } else { field = field + "`" + v["column_name"] + "` = IF(length(${" + v["column_name"] + "})>0,${" + v["column_name"] + "},`" + v["column_name"] + "`),\n" } fName = v["column_name"] } args := entity.EndpointArgs{ Name: fName, Type: getArgsType(v["data_type"]), Required: 0, } argsList = append(argsList, args) } } argsList = removeDuplicates(argsList) // if field all PRI if len(field) < 3 { sql = "USE " + db + ";\nUPDATE `" + table + "` SET " + strings.ReplaceAll(where[:len(where)-4], "and", ",") + " \n WHERE " + where[:len(where)-4] + ";" } else { sql = "USE " + db + ";\nUPDATE `" + table + "` SET " + field[:len(field)-2] + " \n WHERE " + where[:len(where)-4] + ";" } return sql, argsList } func genDeleteSQL(ctx context.Context, db, table string, keys []entity.EndpointArgs) (string, []entity.EndpointArgs) { sql := "" where := "" fName := "" argsList := make([]entity.EndpointArgs, 0) if len(keys) == 0 { where = "1=0 and " } else { for _, v := range keys { beginNumber, _ := isNumberBegin(ctx, v.Name) if beginNumber { where = where + "`" + v.Name + "` = ${p" + v.Name + "} and " fName = "p" + v.Name } else { where = where + "`" + v.Name + "` = ${" + v.Name + "} and " fName = v.Name } args := entity.EndpointArgs{ Name: fName, Type: v.Type, Required: 1, } argsList = append(argsList, args) } } sql = "USE " + db + ";\nDELETE FROM `" + table + "` WHERE " + where[:len(where)-4] + ";" return sql, argsList } func genVectorSearchSQL(ctx context.Context, db, table, vectorField, distanceFunc string) string { sql := "USE " + db + ";\nSELECT *\nFROM " + table + "\nWHERE " + distanceFunc + "(" + vectorField + ", ${query_vector}) < ${match_threshold}\nORDER BY " + distanceFunc + "(" + vectorField + ", ${query_vector})\nLIMIT ${match_count};" return sql } func removeDuplicates(arr []entity.EndpointArgs) []entity.EndpointArgs { uniqueSet := make(map[string]bool) var uniqueArr []entity.EndpointArgs for _, v := range arr { if !uniqueSet[v.Name] { uniqueSet[v.Name] = true uniqueArr = append(uniqueArr, v) } } return uniqueArr } func VerifyBatchOn(ctx context.Context, input entity.VerifyBatch) error { if input.Method == entity.AutoGenerateOperationGET || input.Method == entity.AutoGenerateOperationDELETE { return errs.NewBadRequest("Method GET/DELETE doesn't support batch!", nil) } return nil } func ConvertEndpoint(ctx context.Context, in *entity.Endpoint, operator *string) *dto.EndpointFormat { if in.ID == 0 { return nil } var config entity.EndpointSettings if in.Settings != "" { err := json.Unmarshal([]byte(in.Settings), &config) if err != nil { tilog.Error(ctx, "ConvertEndpoint settings parse failed.", zap.Error(err), zap.Uint64("endpoint_id", in.ID), zap.String("settings", in.Settings)) } } var args []*entity.EndpointArgs if in.Args != "" { err := json.Unmarshal([]byte(in.Args), &args) if err != nil { tilog.Error(ctx, "ConvertEndpoint args parse failed.", zap.Error(err), zap.Uint64("endpoint_id", in.ID), zap.String("args", in.Args)) } } isOnline, err := repository.IsOnlineEndpoint(in.ID) if err != nil { tilog.Error(ctx, "IsOnlineEndpoint Failed!", zap.Uint64("endpointId", in.ID)) isOnline = false } sqlTemplate := in.SqlTemplate clusterID := in.ClusterId database := in.Database if operator != nil && in.IsSystemDataEndpoint() { if playgroundContent, err := loadUserEndpointPlayground(ctx, in.DataappId, in.ID, *operator); err != nil { tilog.Error(ctx, "LoadUserEndpointPlayground failed", zap.Uint64("endpoint_id", in.ID), zap.Any("user", operator), zap.Error(err)) } else { sqlTemplate = playgroundContent.SqlTemplate clusterID = playgroundContent.ClusterID database = playgroundContent.Database } if sqlTemplate == "" { sqlTemplate = `/* Select a cluster and a database before providing the SQL statement. SQL statements and parameters used in this Playground (SQL Editor) are not saved and are intended solely for development and testing purposes */` } } return &dto.EndpointFormat{ ID: in.ID, DataappId: in.DataappId, Name: in.Name, Type: in.Type, Description: in.Description, Endpoint: in.Endpoint, Method: in.Method, ClusterId: strconv.FormatUint(clusterID, 10), Database: database, SqlTemplate: sqlTemplate, Settings: config, Args: args, Tag: in.Tag, Batch: in.IsBatch(), CreatedBy: in.CreatedBy, UpdatedBy: in.UpdatedBy, CreatedAt: in.CreatedAt, UpdatedAt: in.UpdatedAt, Vid: in.Vid, IsOnline: isOnline, } } func saveUserEndpointPlayground(ctx context.Context, endpoint *entity.Endpoint, content entity.EndpointPlaygroundContent, operator string) { if err := repository.SaveUserEndpointPlayground(ctx, &entity.UserEndpointPlayground{ DataappID: endpoint.DataappId, EndpointID: endpoint.ID, CreatedBy: operator, Content: content.JsonMarshal(), }); err != nil { tilog.Error(ctx, "SaveUserEndpointPlayground failed", zap.Error(err)) } } func loadUserEndpointPlayground(ctx context.Context, dataappID, endpointID uint64, operator string) (*entity.EndpointPlaygroundContent, error) { playground, err := repository.LoadUserEndpointPlayground(ctx, dataappID, endpointID, operator) if err != nil { return nil, err } content := new(entity.EndpointPlaygroundContent) content.JsonUnmarshal(playground.Content) return content, nil } func ListEndpointsWithPreID(ctx context.Context, dataappID uint64, preID uint64, limit int) ([]*entity.Endpoint, error) { return repository.ListEndpointsWithPreID(ctx, dataappID, preID, limit) } func LoadEndpointPublishID(ctx context.Context, endpoints ...*entity.Endpoint) error { endpointIDs := make([]uint64, 0, len(endpoints)) for _, endpoint := range endpoints { endpointIDs = append(endpointIDs, endpoint.ID) } deployedEndpoints, err := repository.ListPublishedEndpointByIDs(ctx, endpointIDs) if err != nil { return err } deployedEndpointMap := utils.ConvertArrToMap(deployedEndpoints, func(item *entity.EndpointPublish) uint64 { return item.EndpointId }) for _, endpoint := range endpoints { if deployedEndpointMap[endpoint.ID] != nil { endpoint.PublishID = deployedEndpointMap[endpoint.ID].ID } } return nil }