From 0dbfaa8b67c47b83b2e789764032bdb1bee18ca4 Mon Sep 17 00:00:00 2001 From: mkaaad <119158371+mkaaad@users.noreply.github.com> Date: Thu, 13 Aug 2026 18:39:29 +0800 Subject: [PATCH] feat(cli): add sync logs listing commands for the user CLI (#18195) ### Summary ``` RAGFlow(api/default)> LIST SYNC LOGS WITH PAGE_SIZE 2 PAGE 2; Total: 764 +----------------------------------+-------------------------+-------------+-----------+----------------------------------+----------------------------------+---------+------------------+------------+--------------+--------+-----------+---------------------+--------------------+---------------------+ | connector_id | docs_removed_from_index | error_count | error_msg | id | kb_id | kb_name | new_docs_indexed | prune_freq | refresh_freq | status | task_type | time_started | total_docs_indexed | update_date | +----------------------------------+-------------------------+-------------+-----------+----------------------------------+----------------------------------+---------+------------------+------------+--------------+--------+-----------+---------------------+--------------------+---------------------+ | 02b5b14183884445bd55ad6b9d322204 | 0 | 0 | | fb55b5f1fb754475834cd9eac8b52054 | 0e6e28e65b7044e58070e0e4252319a6 | test | 0 | 5 | 0 | 3 | sync | 2026-08-13 09:49:45 | 16 | 2026-08-13 09:49:46 | | 02b5b14183884445bd55ad6b9d322204 | 0 | 0 | | 3cad779d7d2e4efb8d8d2dbf05d45653 | 0e6e28e65b7044e58070e0e4252319a6 | test | 0 | 5 | 0 | 3 | sync | 2026-08-13 09:49:42 | 16 | 2026-08-13 09:49:43 | +----------------------------------+-------------------------+-------------+-----------+----------------------------------+----------------------------------+---------+------------------+------------+--------------+--------+-----------+---------------------+--------------------+---------------------+ ``` ``` RAGFlow(api/default)> LIST SYNC LOGS; Total: 764 +----------------------------------+-------------------------+-------------+--------------------------------------------------------------------------------------------+----------------------------------+----------------------------------+---------+------------------+------------+--------------+--------+-----------+---------------------+--------------------+---------------------+ | connector_id | docs_removed_from_index | error_count | error_msg | id | kb_id | kb_name | new_docs_indexed | prune_freq | refresh_freq | status | task_type | time_started | total_docs_indexed | update_date | +----------------------------------+-------------------------+-------------+--------------------------------------------------------------------------------------------+----------------------------------+----------------------------------+---------+------------------+------------+--------------+--------+-----------+---------------------+--------------------+---------------------+ | 02b5b14183884445bd55ad6b9d322204 | 0 | 1 | sync task failed: failed to fetch URL: Get "https://xxxxxxxxxxxx/rss.xml": EOF | 87592e5623fc4284bf69b9290f322718 | 0e6e28e65b7044e58070e0e4252319a6 | test | 0 | 5 | 0 | 4 | sync | 2026-08-13 09:49:48 | 16 | 2026-08-13 09:51:01 | | 570138201cf44e1180711f1ebdf8492f | 0 | 1 | sync task failed: failed to fetch URL: Get "https://www.ruanyifeng.com/blog/atom.xml": EOF | 456adbbf21a147af99adfce4405abd99 | 0e6e28e65b7044e58070e0e4252319a6 | test | 0 | 5 | 1 | 4 | sync | 2026-08-13 09:50:00 | 3 | 2026-08-13 09:50:18 | | 02b5b14183884445bd55ad6b9d322204 | 0 | 0 | | fb55b5f1fb754475834cd9eac8b52054 | 0e6e28e65b7044e58070e0e4252319a6 | test | 0 | 5 | 0 | 3 | sync | 2026-08-13 09:49:45 | 16 | 2026-08-13 09:49:46 | | 02b5b14183884445bd55ad6b9d322204 | 0 | 0 | | 3cad779d7d2e4efb8d8d2dbf05d45653 | 0e6e28e65b7044e58070e0e4252319a6 | test | 0 | 5 | 0 | 3 | sync | 2026-08-13 09:49:42 | 16 | 2026-08-13 09:49:43 | | 02b5b14183884445bd55ad6b9d322204 | 0 | 0 | | dd09320aa5b34d6cbcfa7b03ede68c4a | 0e6e28e65b7044e58070e0e4252319a6 | test | 0 | 5 | 0 | 3 | sync | 2026-08-13 09:49:36 | 16 | 2026-08-13 09:49:38 | | 02b5b14183884445bd55ad6b9d322204 | 0 | 0 | | afbcee73240f420ca1997e6506f387fb | 0e6e28e65b7044e58070e0e4252319a6 | test | 0 | 5 | 0 | 3 | sync | 2026-08-13 09:49:33 | 16 | 2026-08-13 09:49:34 | | 02b5b14183884445bd55ad6b9d322204 | 0 | 0 | | 7d018146a0f14a25b7263d5980fdc30f | 0e6e28e65b7044e58070e0e4252319a6 | test | 0 | 5 | 0 | 3 | sync | 2026-08-13 09:49:30 | 16 | 2026-08-13 09:49:32 | | 02b5b14183884445bd55ad6b9d322204 | 0 | 0 | | 520e0d6f421e40628ab541fc4fee1a69 | 0e6e28e65b7044e58070e0e4252319a6 | test | 0 | 5 | 0 | 3 | sync | 2026-08-13 09:49:27 | 16 | 2026-08-13 09:49:28 | | 02b5b14183884445bd55ad6b9d322204 | 0 | 0 | | 3d3ce49825de4ef99642e7994ae954e3 | 0e6e28e65b7044e58070e0e4252319a6 | test | 0 | 5 | 0 | 3 | sync | 2026-08-13 09:49:21 | 16 | 2026-08-13 09:49:22 | | 02b5b14183884445bd55ad6b9d322204 | 0 | 0 | | 1c511f9b4a7c425fb30a956589ca56fa | 0e6e28e65b7044e58070e0e4252319a6 | test | 0 | 5 | 0 | 3 | sync | 2026-08-13 09:49:18 | 16 | 2026-08-13 09:49:19 | | 02b5b14183884445bd55ad6b9d322204 | 0 | 0 | | b9ab73c3f48c4f47a84fd55b542b32b4 | 0e6e28e65b7044e58070e0e4252319a6 | test | 0 | 5 | 0 | 3 | sync | 2026-08-13 09:49:15 | 16 | 2026-08-13 09:49:16 | | 02b5b14183884445bd55ad6b9d322204 | 0 | 0 | | db0cad90a7744d758b24d2e3d3828963 | 0e6e28e65b7044e58070e0e4252319a6 | test | 0 | 5 | 0 | 3 | sync | 2026-08-13 09:49:12 | 16 | 2026-08-13 09:49:13 | | 02b5b14183884445bd55ad6b9d322204 | 0 | 0 | | a501eadb2cce452187f3c86431352c70 | 0e6e28e65b7044e58070e0e4252319a6 | test | 0 | 5 | 0 | 3 | sync | 2026-08-13 09:49:08 | 16 | 2026-08-13 09:49:10 | | 02b5b14183884445bd55ad6b9d322204 | 0 | 0 | | b4265a3b7baf4466a525b91a9fe21b71 | 0e6e28e65b7044e58070e0e4252319a6 | test | 0 | 5 | 0 | 3 | sync | 2026-08-13 09:49:03 | 16 | 2026-08-13 09:49:05 | | 02b5b14183884445bd55ad6b9d322204 | 0 | 0 | | a6ad590931c046dda5fc3222b4010c06 | 0e6e28e65b7044e58070e0e4252319a6 | test | 0 | 5 | 0 | 3 | sync | 2026-08-13 09:49:00 | 16 | 2026-08-13 09:49:02 | ...... +----------------------------------+-------------------------+-------------+--------------------------------------------------------------------------------------------+----------------------------------+----------------------------------+---------+------------------+------------+--------------+--------+-----------+---------------------+--------------------+---------------------+ ``` list all logs ``` RAGFlow(api/default)> LIST DATASET 'test' SYNC LOGS; Total: 764 +----------------------------------+-------------------------+-------------+--------------------------------------------------------------------------------------------+----------------------------------+----------------------------------+---------+------------------+------------+--------------+--------+-----------+---------------------+--------------------+---------------------+ | connector_id | docs_removed_from_index | error_count | error_msg | id | kb_id | kb_name | new_docs_indexed | prune_freq | refresh_freq | status | task_type | time_started | total_docs_indexed | update_date | +----------------------------------+-------------------------+-------------+--------------------------------------------------------------------------------------------+----------------------------------+----------------------------------+---------+------------------+------------+--------------+--------+-----------+---------------------+--------------------+---------------------+ | 02b5b14183884445bd55ad6b9d322204 | 0 | 1 | sync task failed: failed to fetch URL: Get "https://xxxxxxxxxxxx/rss.xml": EOF | 87592e5623fc4284bf69b9290f322718 | 0e6e28e65b7044e58070e0e4252319a6 | test | 0 | 5 | 0 | 4 | sync | 2026-08-13 09:49:48 | 16 | 2026-08-13 09:51:01 | | 570138201cf44e1180711f1ebdf8492f | 0 | 1 | sync task failed: failed to fetch URL: Get "https://www.ruanyifeng.com/blog/atom.xml": EOF | 456adbbf21a147af99adfce4405abd99 | 0e6e28e65b7044e58070e0e4252319a6 | test | 0 | 5 | 1 | 4 | sync | 2026-08-13 09:50:00 | 3 | 2026-08-13 09:50:18 | | 02b5b14183884445bd55ad6b9d322204 | 0 | 0 | | fb55b5f1fb754475834cd9eac8b52054 | 0e6e28e65b7044e58070e0e4252319a6 | test | 0 | 5 | 0 | 3 | sync | 2026-08-13 09:49:45 | 16 | 2026-08-13 09:49:46 | | 02b5b14183884445bd55ad6b9d322204 | 0 | 0 | | 3cad779d7d2e4efb8d8d2dbf05d45653 | 0e6e28e65b7044e58070e0e4252319a6 | test | 0 | 5 | 0 | 3 | sync | 2026-08-13 09:49:42 | 16 | 2026-08-13 09:49:43 | | 02b5b14183884445bd55ad6b9d322204 | 0 | 0 | | dd09320aa5b34d6cbcfa7b03ede68c4a | 0e6e28e65b7044e58070e0e4252319a6 | test | 0 | 5 | 0 | 3 | sync | 2026-08-13 09:49:36 | 16 | 2026-08-13 09:49:38 | | 02b5b14183884445bd55ad6b9d322204 | 0 | 0 | | afbcee73240f420ca1997e6506f387fb | 0e6e28e65b7044e58070e0e4252319a6 | test | 0 | 5 | 0 | 3 | sync | 2026-08-13 09:49:33 | 16 | 2026-08-13 09:49:34 | | 02b5b14183884445bd55ad6b9d322204 | 0 | 0 | | 7d018146a0f14a25b7263d5980fdc30f | 0e6e28e65b7044e58070e0e4252319a6 | test | 0 | 5 | 0 | 3 | sync | 2026-08-13 09:49:30 | 16 | 2026-08-13 09:49:32 | | 02b5b14183884445bd55ad6b9d322204 | 0 | 0 | | 520e0d6f421e40628ab541fc4fee1a69 | 0e6e28e65b7044e58070e0e4252319a6 | test | 0 | 5 | 0 | 3 | sync | 2026-08-13 09:49:27 | 16 | 2026-08-13 09:49:28 | | 02b5b14183884445bd55ad6b9d322204 | 0 | 0 | | 3d3ce49825de4ef99642e7994ae954e3 | 0e6e28e65b7044e58070e0e4252319a6 | test | 0 | 5 | 0 | 3 | sync | 2026-08-13 09:49:21 | 16 | 2026-08-13 09:49:22 | | 02b5b14183884445bd55ad6b9d322204 | 0 | 0 | | 1c511f9b4a7c425fb30a956589ca56fa | 0e6e28e65b7044e58070e0e4252319a6 | test | 0 | 5 | 0 | 3 | sync | 2026-08-13 09:49:18 | 16 | 2026-08-13 09:49:19 | | 02b5b14183884445bd55ad6b9d322204 | 0 | 0 | | b9ab73c3f48c4f47a84fd55b542b32b4 | 0e6e28e65b7044e58070e0e4252319a6 | test | 0 | 5 | 0 | 3 | sync | 2026-08-13 09:49:15 | 16 | 2026-08-13 09:49:16 | | 02b5b14183884445bd55ad6b9d322204 | 0 | 0 | | db0cad90a7744d758b24d2e3d3828963 | 0e6e28e65b7044e58070e0e4252319a6 | test | 0 | 5 | 0 | 3 | sync | 2026-08-13 09:49:12 | 16 | 2026-08-13 09:49:13 | | 02b5b14183884445bd55ad6b9d322204 | 0 | 0 | | a501eadb2cce452187f3c86431352c70 | 0e6e28e65b7044e58070e0e4252319a6 | test | 0 | 5 | 0 | 3 | sync | 2026-08-13 09:49:08 | 16 | 2026-08-13 09:49:10 | | 02b5b14183884445bd55ad6b9d322204 | 0 | 0 | | b4265a3b7baf4466a525b91a9fe21b71 | 0e6e28e65b7044e58070e0e4252319a6 | test | 0 | 5 | 0 | 3 | sync | 2026-08-13 09:49:03 | 16 | 2026-08-13 09:49:05 | | 02b5b14183884445bd55ad6b9d322204 | 0 | 0 | | a6ad590931c046dda5fc3222b4010c06 | 0e6e28e65b7044e58070e0e4252319a6 | test | 0 | 5 | 0 | 3 | sync | 2026-08-13 09:49:00 | 16 | 2026-08-13 09:49:02 | ...... +----------------------------------+-------------------------+-------------+--------------------------------------------------------------------------------------------+----------------------------------+----------------------------------+---------+------------------+------------+--------------+--------+-----------+---------------------+--------------------+---------------------+ ``` --- internal/cli/README.md | 4 + internal/cli/cli_http.go | 2 + internal/cli/lexer.go | 2 + internal/cli/response.go | 36 +++++++ internal/cli/types.go | 1 + internal/cli/user_command.go | 70 +++++++++++++ internal/cli/user_parser.go | 92 +++++++++++++++++ internal/cli/user_parser_test.go | 130 ++++++++++++++++++++++++ internal/dao/connector.go | 50 +++++++++ internal/handler/connector.go | 53 ++++++++++ internal/handler/connector_test.go | 107 +++++++++++++++++++ internal/router/router.go | 2 + internal/service/connector.go | 38 +++++++ internal/service/connector_logs_test.go | 114 +++++++++++++++++++++ internal/syncer/scheduler.go | 1 - 15 files changed, 701 insertions(+), 1 deletion(-) create mode 100644 internal/service/connector_logs_test.go diff --git a/internal/cli/README.md b/internal/cli/README.md index f4c6f9df21..5442a8edf9 100644 --- a/internal/cli/README.md +++ b/internal/cli/README.md @@ -149,6 +149,10 @@ CREATE DATASET 'my_dataset' WITH EMBEDDING 'text-embedding-ada-002' PARSER 'naiv LIST DATASETS; DROP DATASET 'my_dataset'; +-- Data sync logs +LIST SYNC_LOGS; +LIST DATASET 'my_dataset' SYNC_LOGS; + -- Model configuration SET DEFAULT LLM 'gpt-4'; SET DEFAULT EMBEDDING 'text-embedding-ada-002'; diff --git a/internal/cli/cli_http.go b/internal/cli/cli_http.go index 51961d5297..d1d77165ff 100644 --- a/internal/cli/cli_http.go +++ b/internal/cli/cli_http.go @@ -484,6 +484,8 @@ func (c *CLI) ExecuteUserCommand(cmd *Command) (ResponseIf, error) { return c.APIStopIngestionCommand(cmd) case "api_list_ingestion_tasks": return c.APIListIngestionTasks(cmd) + case "api_list_sync_logs": + return c.APIListSyncLogsCommand(cmd) case "api_remove_task": return c.APIRemoveTaskCommand(cmd) case "user_parse_local_file_command": diff --git a/internal/cli/lexer.go b/internal/cli/lexer.go index 207438c8c0..93fc0a5324 100644 --- a/internal/cli/lexer.go +++ b/internal/cli/lexer.go @@ -411,6 +411,8 @@ func (l *Lexer) lookupIdent(ident string) Token { return Token{Type: TokenAsync, Value: ident} case "SYNC": return Token{Type: TokenSync, Value: ident} + case "SYNC_LOGS": + return Token{Type: TokenSyncLogs, Value: ident} case "BENCHMARK": return Token{Type: TokenBenchmark, Value: ident} case "PING": diff --git a/internal/cli/response.go b/internal/cli/response.go index 24086b0506..3b9c5372bd 100644 --- a/internal/cli/response.go +++ b/internal/cli/response.go @@ -213,6 +213,42 @@ func (r *ListDocumentsResponse) PrintOut() { } } +type ListSyncLogsResponse struct { + Code int `json:"code"` + Data map[string]interface{} `json:"data"` + Message string `json:"message"` + Duration float64 + OutputFormat OutputFormat +} + +func (r *ListSyncLogsResponse) Type() string { + return "list_sync_logs" +} + +func (r *ListSyncLogsResponse) TimeCost() float64 { + return r.Duration +} + +func (r *ListSyncLogsResponse) SetOutputFormat(format OutputFormat) { + r.OutputFormat = format +} + +func (r *ListSyncLogsResponse) PrintOut() { + if r.Code == 0 { + total := r.Data["total"].(float64) + fmt.Printf("Total: %0.0f\n", total) + logs := r.Data["logs"].([]interface{}) + table := make([]map[string]interface{}, 0) + for _, log := range logs { + table = append(table, log.(map[string]interface{})) + } + PrintTableSimpleByFormat(table, r.OutputFormat) + } else { + fmt.Println("ERROR") + fmt.Printf("%d, %s\n", r.Code, r.Message) + } +} + type ListAgentsResponse struct { Code int `json:"code"` Data map[string]interface{} `json:"data"` diff --git a/internal/cli/types.go b/internal/cli/types.go index af258d6d1c..f6f48c435c 100644 --- a/internal/cli/types.go +++ b/internal/cli/types.go @@ -129,6 +129,7 @@ const ( TokenDimension TokenAsync TokenSync + TokenSyncLogs TokenBenchmark TokenPing TokenToken diff --git a/internal/cli/user_command.go b/internal/cli/user_command.go index d67386c03f..a8b3277320 100644 --- a/internal/cli/user_command.go +++ b/internal/cli/user_command.go @@ -3467,6 +3467,76 @@ func (c *CLI) APIListIngestionTasks(cmd *Command) (ResponseIf, error) { return HandleCommonResponse(resp, "list ingestion tasks") } +// APIListSyncLogsCommand lists sync logs (user mode). +// LIST SYNC_LOGS; lists the sync logs of all datasets. +// LIST SYNC_LOGS FROM 'dataset_id'; and LIST DATASET 'dataset_name' SYNC_LOGS; +// restrict the listing to one dataset. +func (c *CLI) APIListSyncLogsCommand(cmd *Command) (ResponseIf, error) { + if c.APIServerClientMap[c.Config.APIClientConfig.CurrentAPIServer].APIKey == nil && c.APIServerClientMap[c.Config.APIClientConfig.CurrentAPIServer].LoginToken == nil { + return nil, fmt.Errorf("API key not set. Please login first") + } + + if c.Config.CLIMode != APIMode { + return nil, fmt.Errorf("this command is only allowed in USER mode") + } + + datasetID := "" + if rawID, ok := cmd.Params["dataset_id"].(string); ok { + datasetID = strings.TrimSpace(rawID) + } + if datasetName, ok := cmd.Params["dataset_name"].(string); ok && datasetName != "" { + id, err := c.getDatasetID(datasetName) + if err != nil { + return nil, err + } + datasetID = id + } + + url := "/connectors/sync_logs" + query := netUrl.Values{} + if datasetID != "" { + query.Set("dataset_id", datasetID) + } + page, hasPage := cmd.Params["page"].(int) + pageSize, hasPageSize := cmd.Params["page_size"].(int) + switch { + case hasPage && hasPageSize: + query.Set("page", fmt.Sprintf("%d", page)) + query.Set("page_size", fmt.Sprintf("%d", pageSize)) + case hasPage: + query.Set("page", fmt.Sprintf("%d", page)) + case hasPageSize: + query.Set("page_size", fmt.Sprintf("%d", pageSize)) + default: + // No pagination requested: ask the API for every matching row. + query.Set("page_size", "0") + } + if encoded := query.Encode(); encoded != "" { + url += "?" + encoded + } + + resp, err := c.APIServerClientMap[c.Config.APIClientConfig.CurrentAPIServer].Request("GET", url, "web", nil, nil) + if err != nil { + return nil, fmt.Errorf("failed to list sync logs: %w", err) + } + + if resp.StatusCode != 200 { + return nil, fmt.Errorf("failed to list sync logs: HTTP %d, body: %s", resp.StatusCode, string(resp.Body)) + } + + var result ListSyncLogsResponse + if err = json.Unmarshal(resp.Body, &result); err != nil { + return nil, fmt.Errorf("list sync logs failed: invalid JSON (%w)", err) + } + + if result.Code != 0 { + return nil, fmt.Errorf("%s", result.Message) + } + result.Duration = resp.Duration + + return &result, nil +} + // APIShowLogLevelCommand sets the log level for the system. func (c *CLI) APIShowLogLevelCommand(cmd *Command) (ResponseIf, error) { if c.Config.CLIMode != APIMode { diff --git a/internal/cli/user_parser.go b/internal/cli/user_parser.go index 18d4d23c0a..eecc6b846e 100644 --- a/internal/cli/user_parser.go +++ b/internal/cli/user_parser.go @@ -151,6 +151,8 @@ func (p *Parser) parseAPIListCommands() (*Command, error) { return p.parseAPIListAllModels() case TokenIngestion: return p.parseAPIListIngestionTasks() + case TokenSyncLogs: + return p.parseAPIListSyncLogs() case TokenDefault: return p.parseAPIListDefaultModels() case TokenAvailable: @@ -194,6 +196,8 @@ func (p *Parser) parseAPIListDatasetCommands() (*Command, error) { return p.parseAPIListDatasetFiles(datasetID) case TokenIngestion: return p.parseAPIListDatasetIngestionTasks(datasetID) + case TokenSyncLogs: + return p.parseAPIListDatasetSyncLogs(datasetID) default: return nil, fmt.Errorf("unknown LIST target: %s", p.curToken.Value) } @@ -242,6 +246,94 @@ func (p *Parser) parseAPIListDatasetIngestionTasks(datasetName string) (*Command return cmd, nil } +// LIST SYNC_LOGS; +// LIST SYNC_LOGS FROM 'dataset_id' +// LIST SYNC_LOGS [FROM 'dataset_id'] WITH PAGE 2 PAGE_SIZE 50 +func (p *Parser) parseAPIListSyncLogs() (*Command, error) { + p.nextToken() // consume SYNC_LOGS + + cmd := NewCommand("api_list_sync_logs") + + if p.curToken.Type == TokenFrom { + p.nextToken() + datasetID, err := p.parseQuotedString() + if err != nil { + return nil, err + } + cmd.Params["dataset_id"] = datasetID + p.nextToken() // move past the dataset id + } + + if err := p.parseSyncLogsWithOptions(cmd); err != nil { + return nil, err + } + + // Semicolon is optional + if p.curToken.Type == TokenSemicolon { + p.nextToken() + } + return cmd, nil +} + +// LIST DATASET 'dataset_name' SYNC_LOGS [WITH PAGE 2 PAGE_SIZE 50]; +func (p *Parser) parseAPIListDatasetSyncLogs(datasetName string) (*Command, error) { + p.nextToken() // consume SYNC_LOGS + + cmd := NewCommand("api_list_sync_logs") + cmd.Params["dataset_name"] = datasetName + + if err := p.parseSyncLogsWithOptions(cmd); err != nil { + return nil, err + } + + // Semicolon is optional + if p.curToken.Type == TokenSemicolon { + p.nextToken() + } + return cmd, nil +} + +// parseSyncLogsWithOptions parses the optional WITH clause of the sync logs +// listing commands. Only PAGE and PAGE_SIZE are accepted, both as integers, +// mirroring the search command's space-separated WITH syntax. +func (p *Parser) parseSyncLogsWithOptions(cmd *Command) error { + if p.curToken.Type != TokenWith && !(p.curToken.Type == TokenIdentifier && strings.EqualFold(p.curToken.Value, "with")) { + return nil + } + p.nextToken() // consume WITH + + for p.curToken.Type != TokenEOF && p.curToken.Type != TokenSemicolon { + if p.curToken.Type == TokenComma { + return fmt.Errorf("syntax error: WITH options must be space-separated, not comma-separated") + } + if p.curToken.Type != TokenIdentifier { + break + } + paramName := strings.ToLower(p.curToken.Value) + p.nextToken() + + if p.curToken.Type != TokenInteger { + if p.curToken.Type == TokenEOF || p.curToken.Type == TokenSemicolon { + return fmt.Errorf("WITH option %q is missing a value", paramName) + } + return fmt.Errorf("WITH option %q must be an integer, got %s", paramName, tokenTypeDescription(p.curToken.Type, p.curToken)) + } + value, err := p.parseNumber() + if err != nil { + return err + } + p.nextToken() + + switch paramName { + case "page", "page_size": + cmd.Params[paramName] = value + default: + return fmt.Errorf("unknown WITH option %q", paramName) + } + } + return nil +} + func (p *Parser) parseAPIListAgents() (*Command, error) { p.nextToken() // consume AGENTS diff --git a/internal/cli/user_parser_test.go b/internal/cli/user_parser_test.go index 0e1c46e4d6..8ce6bf3982 100644 --- a/internal/cli/user_parser_test.go +++ b/internal/cli/user_parser_test.go @@ -170,3 +170,133 @@ func TestParseAddModelWithDimensions(t *testing.T) { }) } } + +func TestParseListSyncLogs(t *testing.T) { + tests := []struct { + name string + input string + expected *Command + wantErr bool + }{ + { + name: "LIST SYNC_LOGS", + input: "LIST SYNC_LOGS;", + expected: &Command{ + Type: "api_list_sync_logs", + Params: map[string]interface{}{}, + }, + }, + { + name: "LIST SYNC_LOGS FROM dataset_id", + input: "LIST SYNC_LOGS FROM 'kb-1';", + expected: &Command{ + Type: "api_list_sync_logs", + Params: map[string]interface{}{ + "dataset_id": "kb-1", + }, + }, + }, + { + name: "LIST DATASET name SYNC_LOGS", + input: "LIST DATASET 'my dataset' SYNC_LOGS;", + expected: &Command{ + Type: "api_list_sync_logs", + Params: map[string]interface{}{ + "dataset_name": "my dataset", + }, + }, + }, + { + name: "LIST SYNC_LOGS WITH page and page_size", + input: "LIST SYNC_LOGS WITH PAGE 2 PAGE_SIZE 50;", + expected: &Command{ + Type: "api_list_sync_logs", + Params: map[string]interface{}{ + "page": 2, + "page_size": 50, + }, + }, + }, + { + name: "LIST SYNC_LOGS FROM dataset_id WITH page", + input: "LIST SYNC_LOGS FROM 'kb-1' WITH PAGE 3;", + expected: &Command{ + Type: "api_list_sync_logs", + Params: map[string]interface{}{ + "dataset_id": "kb-1", + "page": 3, + }, + }, + }, + { + name: "LIST DATASET name SYNC_LOGS WITH page_size", + input: "LIST DATASET 'my dataset' SYNC_LOGS WITH PAGE_SIZE 20;", + expected: &Command{ + Type: "api_list_sync_logs", + Params: map[string]interface{}{ + "dataset_name": "my dataset", + "page_size": 20, + }, + }, + }, + { + name: "unknown WITH option", + input: "LIST SYNC_LOGS WITH FOO 1;", + wantErr: true, + }, + { + name: "non-integer WITH option", + input: "LIST SYNC_LOGS WITH PAGE 'x';", + wantErr: true, + }, + { + name: "comma-separated WITH options", + input: "LIST SYNC_LOGS WITH PAGE 2, PAGE_SIZE 50;", + wantErr: true, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + p := NewParser(tt.input) + cmd, err := p.Parse(APIMode) + if tt.wantErr { + if err == nil { + t.Fatalf("Parse() error = nil, want error") + } + return + } + if err != nil { + t.Fatalf("Parse() error = %v", err) + } + + if cmd.Type != tt.expected.Type { + t.Errorf("Command Type = %v, expected = %v", cmd.Type, tt.expected.Type) + } + + gotDatasetID, _ := cmd.Params["dataset_id"].(string) + expectedDatasetID, _ := tt.expected.Params["dataset_id"].(string) + if gotDatasetID != expectedDatasetID { + t.Errorf("dataset_id = %v, expected = %v", gotDatasetID, expectedDatasetID) + } + + gotDatasetName, _ := cmd.Params["dataset_name"].(string) + expectedDatasetName, _ := tt.expected.Params["dataset_name"].(string) + if gotDatasetName != expectedDatasetName { + t.Errorf("dataset_name = %v, expected = %v", gotDatasetName, expectedDatasetName) + } + + gotPage, _ := cmd.Params["page"].(int) + expectedPage, _ := tt.expected.Params["page"].(int) + if gotPage != expectedPage { + t.Errorf("page = %v, expected = %v", gotPage, expectedPage) + } + + gotPageSize, _ := cmd.Params["page_size"].(int) + expectedPageSize, _ := tt.expected.Params["page_size"].(int) + if gotPageSize != expectedPageSize { + t.Errorf("page_size = %v, expected = %v", gotPageSize, expectedPageSize) + } + }) + } +} diff --git a/internal/dao/connector.go b/internal/dao/connector.go index fff493bf7e..3a8647b473 100644 --- a/internal/dao/connector.go +++ b/internal/dao/connector.go @@ -544,3 +544,53 @@ func (dao *ConnectorDAO) ListLogsByConnectorID(ctx context.Context, db *gorm.DB, return logs, total, nil } + +// ListLogs lists sync logs for the given tenant IDs with pagination. +// When datasetID is non-empty, only logs of that dataset are returned. +func (dao *ConnectorDAO) ListLogs(ctx context.Context, db *gorm.DB, tenantIDs []string, datasetID string, offset, limit int) ([]*entity.ConnectorSyncLog, int64, error) { + baseQuery := db.WithContext(ctx).Model(&entity.SyncLogs{}). + Joins("JOIN connector ON sync_logs.connector_id = connector.id"). + Joins("JOIN connector2kb ON sync_logs.connector_id = connector2kb.connector_id AND sync_logs.kb_id = connector2kb.kb_id"). + Joins("JOIN knowledgebase ON sync_logs.kb_id = knowledgebase.id"). + Where("connector.tenant_id IN ?", tenantIDs) + + if datasetID != "" { + baseQuery = baseQuery.Where("sync_logs.kb_id = ?", datasetID) + } + + var total int64 + if err := baseQuery.Distinct("sync_logs.id").Count(&total).Error; err != nil { + return nil, 0, err + } + + var logs []*entity.ConnectorSyncLog + query := baseQuery. + Select( + "sync_logs.id", + "sync_logs.connector_id", + "sync_logs.task_type", + "sync_logs.kb_id", + "sync_logs.update_date", + "sync_logs.new_docs_indexed", + "sync_logs.total_docs_indexed", + "sync_logs.docs_removed_from_index", + "sync_logs.error_msg", + "sync_logs.error_count", + "sync_logs.time_started", + "connector.refresh_freq AS refresh_freq", + "connector.prune_freq AS prune_freq", + "knowledgebase.name AS kb_name", + "sync_logs.status", + ). + Distinct(). + Order("sync_logs.update_date DESC") + if limit > 0 { + query = query.Offset(offset).Limit(limit) + } + err := query.Scan(&logs).Error + if err != nil { + return nil, 0, err + } + + return logs, total, nil +} diff --git a/internal/handler/connector.go b/internal/handler/connector.go index d71e22ef00..63d314e460 100644 --- a/internal/handler/connector.go +++ b/internal/handler/connector.go @@ -36,6 +36,7 @@ type connectorServiceIface interface { CreateConnector(ctx context.Context, userID string, req *service.CreateConnectorRequest) (*entity.Connector, error) GetConnector(ctx context.Context, connectorID, userID string) (*entity.Connector, common.ErrorCode, error) ListLog(ctx context.Context, connectorID, userID string, page, pageSize int) ([]*entity.ConnectorSyncLog, int64, common.ErrorCode, error) + ListLogs(ctx context.Context, userID, datasetID string, page, pageSize int) ([]*entity.ConnectorSyncLog, int64, common.ErrorCode, error) DeleteConnector(ctx context.Context, connectorID, userID string) (bool, common.ErrorCode, error) RebuildConnector(ctx context.Context, connectorID, userID, kbID string) (bool, common.ErrorCode, error) ResumeFailedSync(ctx context.Context, connectorID, userID string, req *service.ResumeFailedSyncRequest) (bool, common.ErrorCode, error) @@ -235,6 +236,58 @@ func (h *ConnectorHandler) ListLogs(c *gin.Context) { common.SuccessWithData(c, gin.H{"total": total, "logs": logs}, "success") } +// ListSyncLogs handles GET /api/v1/connectors/sync_logs. +// Lists sync logs for the current user; when dataset_id is provided, only +// the logs of that dataset are returned. +// @Summary list sync logs +// @Description list sync logs for the current user, optionally filtered by dataset_id +// @Tags connector +// @Accept json +// @Produce json +// @Success 200 {object} map[string]interface{} +// @Router /api/v1/connectors/sync_logs [get] +func (h *ConnectorHandler) ListSyncLogs(c *gin.Context) { + user, errorCode, errorMessage := GetUser(c) + if errorCode != common.CodeSuccess { + common.ErrorWithCode(c, errorCode, errorMessage) + return + } + + page := 1 + pageSize := 15 + if rawPage := strings.TrimSpace(c.Query("page")); rawPage != "" { + parsedPage, err := strconv.Atoi(rawPage) + if err != nil { + common.ErrorWithCode(c, common.CodeArgumentError, "page must be an integer") + return + } + page = parsedPage + } + if rawPageSize := strings.TrimSpace(c.Query("page_size")); rawPageSize != "" { + parsedPageSize, err := strconv.Atoi(rawPageSize) + if err != nil { + common.ErrorWithCode(c, common.CodeArgumentError, "page_size must be an integer") + return + } + pageSize = parsedPageSize + } + + datasetID := strings.TrimSpace(c.Query("dataset_id")) + + ctx := c.Request.Context() + + logs, total, code, err := h.connectorService.ListLogs(ctx, user.ID, datasetID, page, pageSize) + if err != nil { + common.ErrorWithCode(c, code, err.Error()) + return + } + if logs == nil { + logs = []*entity.ConnectorSyncLog{} + } + + common.SuccessWithData(c, gin.H{"total": total, "logs": logs}, "success") +} + // CreateConnector create connector // @Summary create Connectors // @Description create a connectors for the current user diff --git a/internal/handler/connector_test.go b/internal/handler/connector_test.go index 57080390b4..3d420bf5a1 100644 --- a/internal/handler/connector_test.go +++ b/internal/handler/connector_test.go @@ -24,6 +24,12 @@ type fakeConnectorService struct { code common.ErrorCode err error html string + capture *logListCapture +} + +type logListCapture struct { + page int + pageSize int } func (s fakeConnectorService) ListConnectors(context.Context, string) (*service.ListConnectorsResponse, error) { @@ -105,6 +111,17 @@ func (s fakeConnectorService) ListLog(context.Context, string, string, int, int) return s.logs, s.total, common.CodeSuccess, nil } +func (s fakeConnectorService) ListLogs(_ context.Context, _, _ string, page, pageSize int) ([]*entity.ConnectorSyncLog, int64, common.ErrorCode, error) { + if s.err != nil { + return nil, 0, s.code, s.err + } + if s.capture != nil { + s.capture.page = page + s.capture.pageSize = pageSize + } + return s.logs, s.total, common.CodeSuccess, nil +} + func (s fakeConnectorService) DeleteConnector(context.Context, string, string) (bool, common.ErrorCode, error) { if s.err != nil { return false, s.code, s.err @@ -451,3 +468,93 @@ func TestConnectorHandlerListLogs(t *testing.T) { }) } } + +func TestConnectorHandlerListSyncLogs(t *testing.T) { + gin.SetMode(gin.TestMode) + + tests := []struct { + name string + query string + wantCode common.ErrorCode + wantMsg string + wantPage int + wantPageSize int + }{ + { + name: "no pagination params uses defaults", + query: "", + wantCode: common.CodeSuccess, + wantPage: 1, + wantPageSize: 15, + }, + { + name: "page only defaults page_size", + query: "?page=3", + wantCode: common.CodeSuccess, + wantPage: 3, + wantPageSize: 15, + }, + { + name: "page_size only defaults page", + query: "?page_size=50", + wantCode: common.CodeSuccess, + wantPage: 1, + wantPageSize: 50, + }, + { + name: "both params", + query: "?page=2&page_size=10", + wantCode: common.CodeSuccess, + wantPage: 2, + wantPageSize: 10, + }, + { + name: "bad page", + query: "?page=abc", + wantCode: common.CodeArgumentError, + wantMsg: "page must be an integer", + }, + { + name: "bad page_size", + query: "?page_size=xyz", + wantCode: common.CodeArgumentError, + wantMsg: "page_size must be an integer", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + capture := &logListCapture{} + h := &ConnectorHandler{connectorService: fakeConnectorService{ + logs: []*entity.ConnectorSyncLog{{ID: "log-1"}}, + total: 1, + capture: capture, + }} + router := gin.New() + router.GET("/api/v1/connectors/sync_logs", func(c *gin.Context) { + c.Set("user", &entity.User{ID: "tenant-1"}) + h.ListSyncLogs(c) + }) + + resp := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, "/api/v1/connectors/sync_logs"+tt.query, nil) + router.ServeHTTP(resp, req) + + var body map[string]interface{} + if err := json.Unmarshal(resp.Body.Bytes(), &body); err != nil { + t.Fatalf("unmarshal response: %v", err) + } + if body["code"] != float64(tt.wantCode) { + t.Fatalf("code=%v body=%v", body["code"], body) + } + if tt.wantMsg != "" && body["message"] != tt.wantMsg { + t.Fatalf("message=%v body=%v", body["message"], body) + } + if tt.wantCode == common.CodeSuccess { + if capture.page != tt.wantPage || capture.pageSize != tt.wantPageSize { + t.Fatalf("service called with page=%d page_size=%d, want %d/%d", capture.page, capture.pageSize, tt.wantPage, tt.wantPageSize) + } + } + }) + } +} diff --git a/internal/router/router.go b/internal/router/router.go index 0dc12816c4..10ebdd764d 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -626,6 +626,8 @@ func (r *Router) Setup(engine *gin.Engine) { { connectors.GET("", r.connectorHandler.ListConnectors) connectors.POST("", r.connectorHandler.CreateConnector) + // Sync logs for the current user, optionally filtered by dataset. + connectors.GET("/sync_logs", r.connectorHandler.ListSyncLogs) connectors.POST("/google/oauth/web/start", r.connectorHandler.StartGoogleWebOAuth) connectors.POST("/google/oauth/web/result", r.connectorHandler.PollGoogleWebOAuthResult) connectors.POST("/box/oauth/web/start", r.connectorHandler.StartBoxWebOAuth) diff --git a/internal/service/connector.go b/internal/service/connector.go index e3fd4133a1..1458a73d84 100644 --- a/internal/service/connector.go +++ b/internal/service/connector.go @@ -1210,6 +1210,44 @@ func (s *ConnectorService) ListLog(ctx context.Context, connectorID, userID stri return logs, total, common.CodeSuccess, nil } +// ListLogs lists sync logs for the current user with pagination. +// When datasetID is non-empty, only logs of that dataset are returned. +func (s *ConnectorService) ListLogs(ctx context.Context, userID, datasetID string, page, pageSize int) ([]*entity.ConnectorSyncLog, int64, common.ErrorCode, error) { + userID = strings.TrimSpace(userID) + if userID == "" { + return nil, 0, common.CodeDataError, fmt.Errorf("user_id is required") + } + + tenantIDs, err := s.userTenantDAO.GetTenantIDsByUserID(ctx, dao.DB, userID) + if err != nil { + return nil, 0, common.CodeServerError, err + } + tenantIDs = append(tenantIDs, userID) + + offset, limit := 0, pageSize + if pageSize <= 0 { + // pageSize == 0 means no pagination: return every matching row. + limit = 0 + } else { + if page < 1 { + page = 1 + } + if pageSize > 100 { + limit = 15 + } + offset = (page - 1) * limit + } + + logs, total, err := s.connectorDAO.ListLogs(ctx, dao.DB, tenantIDs, datasetID, offset, limit) + if err != nil { + return nil, 0, common.CodeServerError, fmt.Errorf("failed to fetch sync logs: %w", err) + } + if logs == nil { + logs = []*entity.ConnectorSyncLog{} + } + return logs, total, common.CodeSuccess, nil +} + func (s *ConnectorService) StartBoxWebOAuth(ctx context.Context, userID string, req *StartBoxWebOAuthRequest) (*StartBoxWebOAuthResponse, common.ErrorCode, error) { var clientID, clientSecret, redirectURI string if req != nil { diff --git a/internal/service/connector_logs_test.go b/internal/service/connector_logs_test.go new file mode 100644 index 0000000000..308aa32380 --- /dev/null +++ b/internal/service/connector_logs_test.go @@ -0,0 +1,114 @@ +// +// Copyright 2026 The InfiniFlow Authors. All Rights Reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +// + +package service + +import ( + "context" + "fmt" + "testing" + + "ragflow/internal/common" + "ragflow/internal/dao" + "ragflow/internal/entity" +) + +func TestListLogsPaginationModes(t *testing.T) { + db := setupServiceTestDB(t) + pushServiceDB(t, db) + if err := db.AutoMigrate(&entity.Connector{}, &entity.Connector2Kb{}, &entity.Knowledgebase{}, &entity.SyncLogs{}); err != nil { + t.Fatalf("migrate connector tables: %v", err) + } + + if err := db.Create(&entity.Connector{ + ID: "conn-1", + TenantID: "user-1", + Name: "conn-1", + Source: "rss", + InputType: "poll", + Config: entity.JSONMap{}, + Status: string(entity.TaskStatusDone), + RefreshFreq: 5, + PruneFreq: 5, + TimeoutSecs: 60, + }).Error; err != nil { + t.Fatalf("insert connector: %v", err) + } + if err := db.Create(&entity.Knowledgebase{ + ID: "kb-1", + TenantID: "user-1", + Name: "kb-1", + CreatedBy: "user-1", + EmbdID: "embd", + }).Error; err != nil { + t.Fatalf("insert kb: %v", err) + } + if err := db.Create(&entity.Connector2Kb{ + ID: "conn-1-kb-1", + ConnectorID: "conn-1", + KbID: "kb-1", + AutoParse: "1", + }).Error; err != nil { + t.Fatalf("insert connector2kb: %v", err) + } + for i := 0; i < 5; i++ { + if err := db.Create(&entity.SyncLogs{ + ID: fmt.Sprintf("task-%d", i), + ConnectorID: "conn-1", + KbID: "kb-1", + TaskType: dao.TaskTypeSync, + Status: dao.SyncStatusDone, + ErrorMsg: "", + }).Error; err != nil { + t.Fatalf("insert sync log %d: %v", i, err) + } + } + + svc := NewConnectorService() + ctx := context.Background() + + all, total, code, err := svc.ListLogs(ctx, "user-1", "", 1, 0) + if err != nil || code != common.CodeSuccess { + t.Fatalf("ListLogs(all): code=%v err=%v", code, err) + } + if total != 5 || len(all) != 5 { + t.Fatalf("ListLogs(all) = total %d len %d, want 5/5", total, len(all)) + } + + first, total, code, err := svc.ListLogs(ctx, "user-1", "", 1, 2) + if err != nil || code != common.CodeSuccess { + t.Fatalf("ListLogs(page): code=%v err=%v", code, err) + } + if total != 5 || len(first) != 2 { + t.Fatalf("ListLogs(page) = total %d len %d, want 5/2", total, len(first)) + } + + second, total, code, err := svc.ListLogs(ctx, "user-1", "", 2, 2) + if err != nil || code != common.CodeSuccess { + t.Fatalf("ListLogs(page2): code=%v err=%v", code, err) + } + if total != 5 || len(second) != 2 { + t.Fatalf("ListLogs(page2) = total %d len %d, want 5/2", total, len(second)) + } + + normalized, total, code, err := svc.ListLogs(ctx, "user-1", "", 0, 101) + if err != nil || code != common.CodeSuccess { + t.Fatalf("ListLogs(normalize): code=%v err=%v", code, err) + } + if total != 5 || len(normalized) != 5 { + t.Fatalf("ListLogs(normalize) = total %d len %d, want 5/5", total, len(normalized)) + } +} diff --git a/internal/syncer/scheduler.go b/internal/syncer/scheduler.go index fe17e2954c..650dddc17c 100644 --- a/internal/syncer/scheduler.go +++ b/internal/syncer/scheduler.go @@ -125,7 +125,6 @@ func (s *Scheduler) queueAvailable() int { // publishStartupTasks scans DB once for startup reconciliation. func (s *Scheduler) publishStartupTasks(ctx context.Context) error { - // TODO restores running tasks during syncer startup. if err := s.taskService.RecoverRunning(ctx); err != nil { return err }