From a4ec4f84cf73ff1b82acb973c5794009e576b920 Mon Sep 17 00:00:00 2001 From: mkaaad <119158371+mkaaad@users.noreply.github.com> Date: Tue, 18 Aug 2026 19:10:51 +0800 Subject: [PATCH] feat(go-syncer): add GitLab datasource connector with checkpoint resume (#18425) --- internal/syncer/connector/builtin.go | 1 + internal/syncer/connector/gitlab.go | 1052 ++++++++++++++++++++++ internal/syncer/connector/gitlab_test.go | 458 ++++++++++ 3 files changed, 1511 insertions(+) create mode 100644 internal/syncer/connector/gitlab.go create mode 100644 internal/syncer/connector/gitlab_test.go diff --git a/internal/syncer/connector/builtin.go b/internal/syncer/connector/builtin.go index 9e8256b0b2..5327280214 100644 --- a/internal/syncer/connector/builtin.go +++ b/internal/syncer/connector/builtin.go @@ -29,6 +29,7 @@ import ( func RegisterBuiltIns(registry *Registry) { registerBuiltIn(registry, "rss", NewRSSConnector) registerBuiltIn(registry, "github", NewGitHubConnector) + registerBuiltIn(registry, "gitlab", NewGitlabConnector) registerBuiltIn(registry, "gmail", NewGmailConnector) registerBuiltIn(registry, "google-drive", NewGoogleDriveConnector) registerBuiltIn(registry, "google_drive", NewGoogleDriveConnector) diff --git a/internal/syncer/connector/gitlab.go b/internal/syncer/connector/gitlab.go new file mode 100644 index 0000000000..12a783951f --- /dev/null +++ b/internal/syncer/connector/gitlab.go @@ -0,0 +1,1052 @@ +// +// 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 connector + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net/http" + "net/url" + "path/filepath" + "strconv" + "strings" + "time" + + "ragflow/internal/utility" +) + +const ( + defaultGitlabBatchSize = 32 + gitlabRequestTimeout = 60 * time.Second +) + +var gitlabExcludePatterns = []string{"logs", ".github", ".gitlab"} + +// GitlabConnector reads GitLab merge requests, issues, and code files. +type GitlabConnector struct { + projectOwner string + projectName string + gitlabURL string + token string + batchSize int + includeMRs bool + includeIssues bool + includeCodeFiles bool + baseURL string + doJSON func(ctx context.Context, apiURL string, out any) (http.Header, error) + doRaw func(ctx context.Context, apiURL string) ([]byte, error) +} + +// NewGitlabConnector creates a GitLab connector from config. +func NewGitlabConnector(config map[string]any) (*GitlabConnector, error) { + credentials, _ := config["credentials"].(map[string]any) + token, _ := credentials["gitlab_access_token"].(string) + gitlabURL := strings.TrimRight(stringConfig(config["gitlab_url"]), "/") + if gitlabURL == "" { + gitlabURL = "https://gitlab.com" + } + return &GitlabConnector{ + projectOwner: strings.TrimSpace(stringConfig(config["project_owner"])), + projectName: strings.TrimSpace(stringConfig(config["project_name"])), + gitlabURL: gitlabURL, + token: strings.TrimSpace(token), + batchSize: configInt(config["batch_size"], defaultGitlabBatchSize), + includeMRs: configBoolDefault(config["include_mrs"], true), + includeIssues: configBoolDefault(config["include_issues"], true), + includeCodeFiles: configBoolDefault(config["include_code_files"], true), + baseURL: gitlabURL + "/api/v4", + }, nil +} + +// Validate validates GitLab connector settings and credentials. +func (c *GitlabConnector) Validate(ctx context.Context) error { + if c == nil { + return fmt.Errorf("gitlab connector is nil") + } + if c.projectOwner == "" { + return fmt.Errorf("Invalid connector settings: 'project_owner' must be provided") + } + if c.projectName == "" { + return fmt.Errorf("Invalid connector settings: 'project_name' must be provided") + } + if c.token == "" { + return fmt.Errorf("Missing gitlab_access_token in credentials") + } + if c.batchSize <= 0 { + return fmt.Errorf("batch_size must be a positive integer") + } + if _, err := c.getProject(ctx); err != nil { + return err + } + return nil +} + +// ValidateConnectorSetting validates GitLab settings from an unsaved config. +func (c *GitlabConnector) ValidateConnectorSetting(ctx context.Context, request map[string]any) error { + ctx, cancel := context.WithTimeout(ctx, connectorSettingValidationTimeout) + defer cancel() + if c == nil { + return fmt.Errorf("gitlab connector is nil") + } + if c.projectOwner == "" { + return fmt.Errorf("Invalid connector settings: 'project_owner' must be provided") + } + if c.projectName == "" { + return fmt.Errorf("Invalid connector settings: 'project_name' must be provided") + } + if c.token == "" { + return fmt.Errorf("Missing gitlab_access_token in credentials") + } + if c.batchSize <= 0 { + return fmt.Errorf("batch_size must be a positive integer") + } + var user map[string]any + if _, err := c.getJSON(ctx, c.apiURL("/user", nil), &user); err != nil { + return err + } + return nil +} + +// OpenSync opens one GitLab sync session. +func (c *GitlabConnector) OpenSync(ctx context.Context, request SyncRequest) (SyncSession, error) { + project, err := c.getProject(ctx) + if err != nil { + return nil, err + } + session := &gitlabSyncSession{ + connector: c, + project: project, + batchSize: c.batchSize, + stage: gitlabStageCodeFiles, + treeQueue: []string{""}, + treePage: 1, + page: 1, + windowStart: request.WindowStart, + windowEnd: request.WindowEnd, + } + if !c.includeCodeFiles { + session.stage = gitlabStageMRs + session.treeQueue = nil + } + session.applyResume(request.Resume) + return session, nil +} + +// OpenPrune opens one complete GitLab prune snapshot session. +func (c *GitlabConnector) OpenPrune(ctx context.Context, request PruneRequest) (PruneSession, error) { + project, err := c.getProject(ctx) + if err != nil { + return nil, err + } + session := &gitlabPruneSession{ + connector: c, + project: project, + batchSize: c.batchSize, + stage: gitlabStageCodeFiles, + treeQueue: []string{""}, + treePage: 1, + page: 1, + } + if !c.includeCodeFiles { + session.stage = gitlabStageMRs + session.treeQueue = nil + } + return session, nil +} + +// getProject fetches the GitLab project info. +func (c *GitlabConnector) getProject(ctx context.Context) (gitlabProject, error) { + encoded := gitlabProjectPath(c.projectOwner, c.projectName) + var project gitlabProject + if _, err := c.getJSON(ctx, c.apiURL("/projects/"+encoded, nil), &project); err != nil { + return gitlabProject{}, err + } + return project, nil +} + +// listMergeRequestPage returns one page of GitLab merge requests. +func (c *GitlabConnector) listMergeRequestPage(ctx context.Context, projectID int, page, pageSize int, windowStart *time.Time, windowEnd time.Time) ([]gitlabBufferedDocument, bool, error) { + query := gitlabListQuery(page, pageSize) + query.Set("order_by", "updated_at") + query.Set("sort", "desc") + var batch []gitlabMergeRequest + headers, err := c.getJSON(ctx, c.apiURL(fmt.Sprintf("/projects/%d/merge_requests", projectID), query), &batch) + if err != nil { + return nil, false, err + } + documents := make([]gitlabBufferedDocument, 0, len(batch)) + doneByWindow := false + pageOffset := 0 + for _, mr := range batch { + if beforeOrAtWindowStart(mr.UpdatedAt, windowStart) { + doneByWindow = true + break + } + if afterWindowEnd(mr.UpdatedAt, windowEnd) { + continue + } + doc := mr.toSourceDocument() + pageOffset++ + documents = append(documents, gitlabBufferedDocument{ + document: doc, + checkpoint: gitlabSyncCheckpoint(gitlabSyncCursor{Stage: gitlabStageMRs, Page: page, Offset: pageOffset, SourceID: doc.SourceID}, doc), + offset: pageOffset, + sourceID: doc.SourceID, + }) + } + done := doneByWindow || !gitlabHasNextPage(headers) || len(batch) == 0 + return documents, done, nil +} + +// listIssuePage returns one page of GitLab issues. +func (c *GitlabConnector) listIssuePage(ctx context.Context, projectID int, page, pageSize int, windowStart *time.Time, windowEnd time.Time) ([]gitlabBufferedDocument, bool, error) { + query := gitlabListQuery(page, pageSize) + query.Set("order_by", "updated_at") + query.Set("sort", "desc") + var batch []gitlabIssue + headers, err := c.getJSON(ctx, c.apiURL(fmt.Sprintf("/projects/%d/issues", projectID), query), &batch) + if err != nil { + return nil, false, err + } + documents := make([]gitlabBufferedDocument, 0, len(batch)) + doneByWindow := false + pageOffset := 0 + for _, issue := range batch { + if beforeOrAtWindowStart(issue.UpdatedAt, windowStart) { + doneByWindow = true + break + } + if afterWindowEnd(issue.UpdatedAt, windowEnd) { + continue + } + doc := issue.toSourceDocument() + pageOffset++ + documents = append(documents, gitlabBufferedDocument{ + document: doc, + checkpoint: gitlabSyncCheckpoint(gitlabSyncCursor{Stage: gitlabStageIssues, Page: page, Offset: pageOffset, SourceID: doc.SourceID}, doc), + offset: pageOffset, + sourceID: doc.SourceID, + }) + } + done := doneByWindow || !gitlabHasNextPage(headers) || len(batch) == 0 + return documents, done, nil +} + +// listTreePage returns one page of GitLab repository tree items. +func (c *GitlabConnector) listTreePage(ctx context.Context, projectID int, branch, path string, page, pageSize int) ([]gitlabTreeItem, bool, error) { + query := url.Values{ + "per_page": {strconv.Itoa(pageSize)}, + "page": {strconv.Itoa(page)}, + "ref": {branch}, + } + if path != "" { + query.Set("path", path) + } + var batch []gitlabTreeItem + headers, err := c.getJSON(ctx, c.apiURL(fmt.Sprintf("/projects/%d/repository/tree", projectID), query), &batch) + if err != nil { + return nil, false, err + } + done := !gitlabHasNextPage(headers) || len(batch) == 0 + return batch, done, nil +} + +// fetchLastCommit returns the most recent commit for a file path. +func (c *GitlabConnector) fetchLastCommit(ctx context.Context, projectID int, branch, path string) (gitlabCommit, error) { + query := url.Values{ + "ref_name": {branch}, + "path": {path}, + "per_page": {"1"}, + } + var commits []gitlabCommit + if _, err := c.getJSON(ctx, c.apiURL(fmt.Sprintf("/projects/%d/repository/commits", projectID), query), &commits); err != nil { + return gitlabCommit{}, err + } + if len(commits) == 0 { + return gitlabCommit{}, nil + } + return commits[0], nil +} + +// fetchFileRaw downloads raw file content from GitLab. +func (c *GitlabConnector) fetchFileRaw(ctx context.Context, projectID int, branch, path string) ([]byte, error) { + encoded := url.PathEscape(path) + apiURL := c.apiURL(fmt.Sprintf("/projects/%d/repository/files/%s/raw", projectID, encoded), url.Values{"ref": {branch}}) + return c.getRaw(ctx, apiURL) +} + +// getJSON fetches a GitLab API JSON response into out. +func (c *GitlabConnector) getJSON(ctx context.Context, apiURL string, out any) (http.Header, error) { + if c.doJSON != nil { + return c.doJSON(ctx, apiURL, out) + } + hostname, resolvedIP, err := utility.AssertURLSafe(apiURL) + if err != nil { + return nil, err + } + client := utility.PinnedHTTPClient(hostname, resolvedIP, gitlabRequestTimeout) + req, err := http.NewRequestWithContext(ctx, http.MethodGet, apiURL, nil) + if err != nil { + return nil, err + } + req.Header.Set("Accept", "application/json") + req.Header.Set("PRIVATE-TOKEN", c.token) + resp, err := client.Do(req) + if err != nil { + return nil, fmt.Errorf("failed to fetch GitLab API: %w", err) + } + defer resp.Body.Close() + if resp.StatusCode >= 400 { + body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) + return nil, fmt.Errorf("GitLab API returned HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(body))) + } + if err = json.NewDecoder(resp.Body).Decode(out); err != nil { + return nil, err + } + return resp.Header.Clone(), nil +} + +// getRaw fetches raw bytes from a GitLab API URL. +func (c *GitlabConnector) getRaw(ctx context.Context, apiURL string) ([]byte, error) { + if c.doRaw != nil { + return c.doRaw(ctx, apiURL) + } + hostname, resolvedIP, err := utility.AssertURLSafe(apiURL) + if err != nil { + return nil, err + } + client := utility.PinnedHTTPClient(hostname, resolvedIP, gitlabRequestTimeout) + req, err := http.NewRequestWithContext(ctx, http.MethodGet, apiURL, nil) + if err != nil { + return nil, err + } + req.Header.Set("PRIVATE-TOKEN", c.token) + resp, err := client.Do(req) + if err != nil { + return nil, fmt.Errorf("failed to fetch GitLab raw file: %w", err) + } + defer resp.Body.Close() + if resp.StatusCode >= 400 { + body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096)) + return nil, fmt.Errorf("GitLab API returned HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(body))) + } + return io.ReadAll(resp.Body) +} + +// apiURL builds a GitLab API URL. +func (c *GitlabConnector) apiURL(path string, query url.Values) string { + path = strings.TrimLeft(path, "/") + u := strings.TrimRight(c.baseURL, "/") + "/" + path + if len(query) == 0 { + return u + } + return u + "?" + query.Encode() +} + +// fileURL builds the web URL for a repository file. +func (c *GitlabConnector) fileURL(branch, path string) string { + return fmt.Sprintf("%s/%s/%s/-/blob/%s/%s", c.gitlabURL, c.projectOwner, c.projectName, branch, path) +} + +// projectFullName returns "owner/name". +func (c *GitlabConnector) projectFullName() string { + return c.projectOwner + "/" + c.projectName +} + +// gitlabSyncSession streams GitLab documents for one fixed sync window. +type gitlabSyncSession struct { + connector *GitlabConnector + project gitlabProject + batchSize int + stage string + page int + treeQueue []string + treePage int + windowStart *time.Time + windowEnd time.Time + buffer []gitlabBufferedDocument + resumeStage string + resumePage int + resumeOffset int + resumeSourceID string +} + +// NextBatch returns the next GitLab document batch. +func (s *gitlabSyncSession) NextBatch(ctx context.Context) (SyncBatch, error) { + documents := make([]SourceDocument, 0, s.batchSize) + var checkpoint *SyncCheckpoint + if len(s.buffer) > 0 { + n := s.batchSize + if n > len(s.buffer) { + n = len(s.buffer) + } + for _, buffered := range s.buffer[:n] { + documents = append(documents, buffered.document) + checkpoint = buffered.checkpoint + } + s.buffer = s.buffer[n:] + } + + for len(documents) < s.batchSize { + batch, err := s.nextDocumentPage(ctx) + if err != nil { + return SyncBatch{}, err + } + if len(batch) == 0 { + if s.stage == gitlabStageDone { + if len(documents) == 0 { + return SyncBatch{}, io.EOF + } + break + } + continue + } + remaining := s.batchSize - len(documents) + if len(batch) > remaining { + for _, buffered := range batch[:remaining] { + documents = append(documents, buffered.document) + checkpoint = buffered.checkpoint + } + s.buffer = append(s.buffer, batch[remaining:]...) + break + } + for _, buffered := range batch { + documents = append(documents, buffered.document) + checkpoint = buffered.checkpoint + } + } + return SyncBatch{Documents: documents, Checkpoint: checkpoint}, nil +} + +// Close closes the GitLab sync session. +func (s *gitlabSyncSession) Close() error { + return nil +} + +// Fetch downloads a GitLab file body for a delayed source document. +func (s *gitlabSyncSession) Fetch(ctx context.Context, ref FetchReference) ([]byte, error) { + var fetch gitlabFetchReference + if err := json.Unmarshal([]byte(ref.Key), &fetch); err != nil { + return nil, err + } + return s.connector.fetchFileRaw(ctx, fetch.ProjectID, fetch.Ref, fetch.FilePath) +} + +const ( + gitlabStageCodeFiles = "code_files" + gitlabStageMRs = "merge_requests" + gitlabStageIssues = "issues" + gitlabStageDone = "done" +) + +// nextDocumentPage fetches one GitLab API page for sync. +func (s *gitlabSyncSession) nextDocumentPage(ctx context.Context) ([]gitlabBufferedDocument, error) { + switch s.stage { + case gitlabStageCodeFiles: + if !s.connector.includeCodeFiles { + s.advanceStage() + return nil, nil + } + return s.nextCodeFilesPage(ctx) + case gitlabStageMRs: + if !s.connector.includeMRs { + s.advanceStage() + return nil, nil + } + docs, done, err := s.connector.listMergeRequestPage(ctx, s.project.ID, s.page, s.batchSize, s.windowStart, s.windowEnd) + if err != nil { + return nil, err + } + docs = s.filterResumedDocuments(gitlabStageMRs, s.page, docs) + if done { + s.advanceStage() + } else { + s.page++ + } + return docs, nil + case gitlabStageIssues: + if !s.connector.includeIssues { + s.advanceStage() + return nil, nil + } + docs, done, err := s.connector.listIssuePage(ctx, s.project.ID, s.page, s.batchSize, s.windowStart, s.windowEnd) + if err != nil { + return nil, err + } + docs = s.filterResumedDocuments(gitlabStageIssues, s.page, docs) + if done { + s.advanceStage() + } else { + s.page++ + } + return docs, nil + default: + s.stage = gitlabStageDone + return nil, nil + } +} + +// nextCodeFilesPage fetches one tree page and converts blobs into documents. +// pendingPaths is captured before processing so resume re-discovers subtrees +// without duplicating queue entries. +func (s *gitlabSyncSession) nextCodeFilesPage(ctx context.Context) ([]gitlabBufferedDocument, error) { + if len(s.treeQueue) == 0 { + s.advanceStage() + return nil, nil + } + currentPath := s.treeQueue[0] + pendingPaths := append([]string(nil), s.treeQueue[1:]...) + items, done, err := s.connector.listTreePage(ctx, s.project.ID, s.project.DefaultBranch, currentPath, s.treePage, s.batchSize) + if err != nil { + return nil, err + } + documents := make([]gitlabBufferedDocument, 0, len(items)) + pageOffset := 0 + for _, item := range items { + if shouldExcludeGitlabPath(item.Path) { + continue + } + if item.Type == "tree" { + s.treeQueue = append(s.treeQueue, item.Path) + continue + } + if item.Type != "blob" { + continue + } + // The tree API does not expose a committed date, so this request still + // supplies the update time used by window filtering. + commit, err := s.connector.fetchLastCommit(ctx, s.project.ID, s.project.DefaultBranch, item.Path) + if err != nil { + return nil, err + } + updatedAt := time.Now().UTC() + if !commit.CommittedDate.IsZero() { + updatedAt = commit.CommittedDate.UTC() + } + if beforeOrAtWindowStart(updatedAt, s.windowStart) { + continue + } + if afterWindowEnd(updatedAt, s.windowEnd) { + continue + } + fileURL := s.connector.fileURL(s.project.DefaultBranch, item.Path) + fetchRef := gitlabFetchReference{ProjectID: s.project.ID, FilePath: item.Path, Ref: s.project.DefaultBranch} + fetchKey, _ := json.Marshal(fetchRef) + doc := SourceDocument{ + SourceID: fileURL, + SemanticIdentifier: item.Name, + Extension: gitlabFileExt(item.Name), + FetchRef: &FetchReference{Key: string(fetchKey), SizeHint: 0}, + UpdatedAt: updatedAt, + Metadata: map[string]any{ + "type": "CodeFile", + "path": item.Path, + "ref": s.project.DefaultBranch, + "project": s.connector.projectFullName(), + "web_url": fileURL, + }, + Fingerprint: gitlabCodeFileFingerprint(item, s.project.DefaultBranch), + } + pageOffset++ + cursor := gitlabSyncCursor{ + Stage: gitlabStageCodeFiles, + Page: s.treePage, + Offset: pageOffset, + SourceID: doc.SourceID, + TreePath: currentPath, + PendingPaths: append([]string(nil), pendingPaths...), + } + documents = append(documents, gitlabBufferedDocument{ + document: doc, + checkpoint: gitlabSyncCheckpoint(cursor, doc), + offset: pageOffset, + sourceID: doc.SourceID, + }) + } + documents = s.filterResumedDocuments(gitlabStageCodeFiles, s.treePage, documents) + if done { + s.treeQueue = s.treeQueue[1:] + s.treePage = 1 + if len(s.treeQueue) == 0 { + s.advanceStage() + } + } else { + s.treePage++ + } + return documents, nil +} + +// advanceStage moves a sync session to the next GitLab stage. +func (s *gitlabSyncSession) advanceStage() { + switch s.stage { + case gitlabStageCodeFiles: + s.stage = gitlabStageMRs + case gitlabStageMRs: + s.stage = gitlabStageIssues + default: + s.stage = gitlabStageDone + } + s.page = 1 + s.treeQueue = nil + s.treePage = 1 + s.clearResume() +} + +// applyResume advances a sync session to the last committed GitLab position. +func (s *gitlabSyncSession) applyResume(checkpoint *SyncCheckpoint) { + if checkpoint == nil || checkpoint.Cursor == "" { + return + } + var cursor gitlabSyncCursor + if err := json.Unmarshal([]byte(checkpoint.Cursor), &cursor); err != nil { + return + } + if cursor.Stage == "" || cursor.Page <= 0 { + return + } + s.stage = cursor.Stage + s.resumeStage = cursor.Stage + s.resumePage = cursor.Page + s.resumeOffset = cursor.Offset + s.resumeSourceID = firstNonEmpty(cursor.SourceID, checkpoint.SourceID) + if cursor.Stage == gitlabStageCodeFiles { + s.treeQueue = append([]string{cursor.TreePath}, cursor.PendingPaths...) + s.treePage = cursor.Page + } else { + s.page = cursor.Page + } +} + +// filterResumedDocuments drops documents through the committed checkpoint. +func (s *gitlabSyncSession) filterResumedDocuments(stage string, page int, candidates []gitlabBufferedDocument) []gitlabBufferedDocument { + if s.resumeStage == "" || stage != s.resumeStage || page != s.resumePage { + return candidates + } + if s.resumeSourceID != "" { + for index, candidate := range candidates { + if candidate.sourceID == s.resumeSourceID { + s.clearResume() + return candidates[index+1:] + } + } + } + if s.resumeOffset <= 0 { + s.clearResume() + return candidates + } + filtered := candidates[:0] + for _, candidate := range candidates { + if candidate.offset > s.resumeOffset { + filtered = append(filtered, candidate) + } + } + s.clearResume() + return filtered +} + +func (s *gitlabSyncSession) clearResume() { + s.resumeStage = "" + s.resumePage = 0 + s.resumeOffset = 0 + s.resumeSourceID = "" +} + +// gitlabPruneSession streams a complete GitLab slim snapshot. +type gitlabPruneSession struct { + connector *GitlabConnector + project gitlabProject + batchSize int + stage string + page int + treeQueue []string + treePage int + buffer []SlimDocument +} + +// NextBatch returns the next GitLab prune snapshot batch. +func (s *gitlabPruneSession) NextBatch(ctx context.Context) (PruneBatch, error) { + documents := make([]SlimDocument, 0, s.batchSize) + if len(s.buffer) > 0 { + n := s.batchSize + if n > len(s.buffer) { + n = len(s.buffer) + } + documents = append(documents, s.buffer[:n]...) + s.buffer = s.buffer[n:] + } + for len(documents) < s.batchSize { + batch, err := s.nextSlimPage(ctx) + if err != nil { + return PruneBatch{}, err + } + if len(batch) == 0 { + if s.stage == gitlabStageDone { + if len(documents) == 0 { + return PruneBatch{}, io.EOF + } + break + } + continue + } + remaining := s.batchSize - len(documents) + if len(batch) > remaining { + documents = append(documents, batch[:remaining]...) + s.buffer = append(s.buffer, batch[remaining:]...) + break + } + documents = append(documents, batch...) + } + return PruneBatch{Documents: documents}, nil +} + +// Close closes the GitLab prune session. +func (s *gitlabPruneSession) Close() error { + return nil +} + +// nextSlimPage fetches one GitLab API page for prune. +func (s *gitlabPruneSession) nextSlimPage(ctx context.Context) ([]SlimDocument, error) { + switch s.stage { + case gitlabStageCodeFiles: + if !s.connector.includeCodeFiles { + s.advanceStage() + return nil, nil + } + return s.nextCodeFilesSlimPage(ctx) + case gitlabStageMRs: + if !s.connector.includeMRs { + s.advanceStage() + return nil, nil + } + return s.nextMRSlimPage(ctx) + case gitlabStageIssues: + if !s.connector.includeIssues { + s.advanceStage() + return nil, nil + } + return s.nextIssueSlimPage(ctx) + default: + s.stage = gitlabStageDone + return nil, nil + } +} + +// nextCodeFilesSlimPage fetches one tree page for prune. +func (s *gitlabPruneSession) nextCodeFilesSlimPage(ctx context.Context) ([]SlimDocument, error) { + if len(s.treeQueue) == 0 { + s.advanceStage() + return nil, nil + } + currentPath := s.treeQueue[0] + items, done, err := s.connector.listTreePage(ctx, s.project.ID, s.project.DefaultBranch, currentPath, s.treePage, s.batchSize) + if err != nil { + return nil, err + } + documents := make([]SlimDocument, 0, len(items)) + for _, item := range items { + if shouldExcludeGitlabPath(item.Path) { + continue + } + if item.Type == "tree" { + s.treeQueue = append(s.treeQueue, item.Path) + continue + } + if item.Type != "blob" { + continue + } + documents = append(documents, SlimDocument{SourceID: s.connector.fileURL(s.project.DefaultBranch, item.Path)}) + } + if done { + s.treeQueue = s.treeQueue[1:] + s.treePage = 1 + if len(s.treeQueue) == 0 { + s.advanceStage() + } + } else { + s.treePage++ + } + return documents, nil +} + +// nextMRSlimPage fetches one MR page for prune. +func (s *gitlabPruneSession) nextMRSlimPage(ctx context.Context) ([]SlimDocument, error) { + query := gitlabListQuery(s.page, s.batchSize) + var batch []gitlabMergeRequest + headers, err := s.connector.getJSON(ctx, s.connector.apiURL(fmt.Sprintf("/projects/%d/merge_requests", s.project.ID), query), &batch) + if err != nil { + return nil, err + } + documents := make([]SlimDocument, 0, len(batch)) + for _, mr := range batch { + documents = append(documents, SlimDocument{SourceID: mr.WebURL}) + } + if !gitlabHasNextPage(headers) || len(batch) == 0 { + s.advanceStage() + } else { + s.page++ + } + return documents, nil +} + +// nextIssueSlimPage fetches one issue page for prune. +func (s *gitlabPruneSession) nextIssueSlimPage(ctx context.Context) ([]SlimDocument, error) { + query := gitlabListQuery(s.page, s.batchSize) + var batch []gitlabIssue + headers, err := s.connector.getJSON(ctx, s.connector.apiURL(fmt.Sprintf("/projects/%d/issues", s.project.ID), query), &batch) + if err != nil { + return nil, err + } + documents := make([]SlimDocument, 0, len(batch)) + for _, issue := range batch { + documents = append(documents, SlimDocument{SourceID: issue.WebURL}) + } + if !gitlabHasNextPage(headers) || len(batch) == 0 { + s.advanceStage() + } else { + s.page++ + } + return documents, nil +} + +// advanceStage moves a prune session to the next GitLab stage. +func (s *gitlabPruneSession) advanceStage() { + switch s.stage { + case gitlabStageCodeFiles: + s.stage = gitlabStageMRs + case gitlabStageMRs: + s.stage = gitlabStageIssues + default: + s.stage = gitlabStageDone + } + s.page = 1 + s.treeQueue = nil + s.treePage = 1 +} + +type gitlabProject struct { + ID int `json:"id"` + DefaultBranch string `json:"default_branch"` + WebURL string `json:"web_url"` +} + +type gitlabSyncCursor struct { + Stage string `json:"stage"` + Page int `json:"page"` + Offset int `json:"offset"` + SourceID string `json:"source_id"` + TreePath string `json:"tree_path,omitempty"` + PendingPaths []string `json:"pending_paths,omitempty"` +} + +type gitlabBufferedDocument struct { + document SourceDocument + checkpoint *SyncCheckpoint + offset int + sourceID string +} + +type gitlabFetchReference struct { + ProjectID int `json:"project_id"` + FilePath string `json:"file_path"` + Ref string `json:"ref"` +} + +func gitlabSyncCheckpoint(cursor gitlabSyncCursor, doc SourceDocument) *SyncCheckpoint { + data, err := json.Marshal(cursor) + if err != nil { + return nil + } + updatedAt := doc.UpdatedAt + return &SyncCheckpoint{ + Cursor: string(data), + SourceID: doc.SourceID, + UpdatedAt: &updatedAt, + } +} + +type gitlabMergeRequest struct { + IID int `json:"iid"` + Title string `json:"title"` + Description string `json:"description"` + State string `json:"state"` + WebURL string `json:"web_url"` + UpdatedAt time.Time `json:"updated_at"` + Author *gitlabUser `json:"author"` +} + +// toSourceDocument converts a merge request into the syncer model. +func (m gitlabMergeRequest) toSourceDocument() SourceDocument { + body := []byte(m.Description) + author := m.Author.metadata() + return SourceDocument{ + SourceID: m.WebURL, + SemanticIdentifier: m.Title, + Extension: ".md", + Blob: body, + UpdatedAt: m.UpdatedAt.UTC(), + SizeBytes: int64(len(body)), + Metadata: map[string]any{ + "state": m.State, + "type": "MergeRequest", + "web_url": m.WebURL, + "author": author, + }, + Fingerprint: stableFingerprint(map[string]any{ + "type": "MergeRequest", + "url": m.WebURL, + "title": m.Title, + "description": m.Description, + "state": m.State, + "updated_at": m.UpdatedAt.UTC(), + "author": author, + }), + } +} + +type gitlabIssue struct { + IID int `json:"iid"` + Title string `json:"title"` + Description string `json:"description"` + State string `json:"state"` + WebURL string `json:"web_url"` + UpdatedAt time.Time `json:"updated_at"` + Author *gitlabUser `json:"author"` + Type string `json:"type"` +} + +// toSourceDocument converts an issue into the syncer model. +func (i gitlabIssue) toSourceDocument() SourceDocument { + body := []byte(i.Description) + author := i.Author.metadata() + issueType := i.Type + if issueType == "" { + issueType = "Issue" + } + return SourceDocument{ + SourceID: i.WebURL, + SemanticIdentifier: i.Title, + Extension: ".md", + Blob: body, + UpdatedAt: i.UpdatedAt.UTC(), + SizeBytes: int64(len(body)), + Metadata: map[string]any{ + "state": i.State, + "type": issueType, + "web_url": i.WebURL, + "author": author, + }, + Fingerprint: stableFingerprint(map[string]any{ + "type": issueType, + "url": i.WebURL, + "title": i.Title, + "description": i.Description, + "state": i.State, + "updated_at": i.UpdatedAt.UTC(), + "author": author, + }), + } +} + +type gitlabTreeItem struct { + ID string `json:"id"` + Name string `json:"name"` + Type string `json:"type"` + Path string `json:"path"` + Mode string `json:"mode"` +} + +func gitlabCodeFileFingerprint(item gitlabTreeItem, ref string) string { + return stableFingerprint(map[string]any{ + "type": "CodeFile", + "path": item.Path, + "ref": ref, + "id": item.ID, + }) +} + +type gitlabCommit struct { + ID string `json:"id"` + CommittedDate time.Time `json:"committed_date"` +} + +type gitlabUser struct { + Name string `json:"name"` + Username string `json:"username"` +} + +// metadata returns user metadata. +func (u *gitlabUser) metadata() map[string]string { + if u == nil { + return nil + } + out := map[string]string{} + if u.Name != "" { + out["name"] = u.Name + } + if u.Username != "" { + out["username"] = u.Username + } + return out +} + +// gitlabListQuery builds standard GitLab list query parameters. +func gitlabListQuery(page, pageSize int) url.Values { + if pageSize <= 0 { + pageSize = defaultGitlabBatchSize + } + return url.Values{ + "state": {"all"}, + "per_page": {strconv.Itoa(pageSize)}, + "page": {strconv.Itoa(page)}, + } +} + +// gitlabHasNextPage reports whether a GitLab response has a next page. +func gitlabHasNextPage(headers http.Header) bool { + nextPage := strings.TrimSpace(headers.Get("X-Next-Page")) + if nextPage != "" && nextPage != "0" { + return true + } + return strings.Contains(headers.Get("Link"), `rel="next"`) +} + +// shouldExcludeGitlabPath reports whether a path matches an exclude pattern. +func shouldExcludeGitlabPath(path string) bool { + for _, pattern := range gitlabExcludePatterns { + pattern = strings.TrimSuffix(pattern, "/") + if path == pattern || strings.HasPrefix(path, pattern+"/") { + return true + } + } + return false +} + +// gitlabFileExt returns the lowercased file extension. +func gitlabFileExt(name string) string { + ext := filepath.Ext(name) + if ext == "" { + return "" + } + return strings.ToLower(ext) +} + +// gitlabProjectPath URL-encodes "owner/name" for the GitLab projects API. +func gitlabProjectPath(owner, name string) string { + return url.QueryEscape(owner + "/" + name) +} diff --git a/internal/syncer/connector/gitlab_test.go b/internal/syncer/connector/gitlab_test.go new file mode 100644 index 0000000000..c2863ca49d --- /dev/null +++ b/internal/syncer/connector/gitlab_test.go @@ -0,0 +1,458 @@ +package connector + +import ( + "context" + "encoding/json" + "errors" + "io" + "net/http" + "net/url" + "strings" + "testing" +) + +// TestGitlabConnectorOpenSyncUsesWindowAndFingerprint verifies incremental sync emits only updated docs with fingerprints. +func TestGitlabConnectorOpenSyncUsesWindowAndFingerprint(t *testing.T) { + connector, err := NewGitlabConnector(map[string]any{ + "project_owner": "owner", + "project_name": "repo", + "include_mrs": true, + "include_issues": true, + "include_code_files": false, + "batch_size": 10, + "credentials": map[string]any{"gitlab_access_token": "token"}, + }) + if err != nil { + t.Fatalf("NewGitlabConnector failed: %v", err) + } + connector.baseURL = "https://gitlab.com/api/v4" + connector.doJSON = gitlabFixtureDoJSON(t) + + start := mustTime(t, "2026-01-02T12:00:00Z") + end := mustTime(t, "2026-01-04T00:00:00Z") + session, err := connector.OpenSync(context.Background(), SyncRequest{WindowStart: &start, WindowEnd: end}) + if err != nil { + t.Fatalf("OpenSync failed: %v", err) + } + batch, err := session.NextBatch(context.Background()) + if err != nil { + t.Fatalf("NextBatch failed: %v", err) + } + if len(batch.Documents) != 1 { + t.Fatalf("documents len = %d, want 1", len(batch.Documents)) + } + doc := batch.Documents[0] + if doc.SourceID != "https://gitlab.com/owner/repo/-/merge_requests/7" { + t.Fatalf("source id = %q", doc.SourceID) + } + if doc.Fingerprint == "" { + t.Fatalf("fingerprint is empty") + } + if _, err = session.NextBatch(context.Background()); !errors.Is(err, io.EOF) { + t.Fatalf("NextBatch EOF = %v", err) + } +} + +// TestGitlabFingerprintStable verifies GitLab fingerprints are stable and content-sensitive. +func TestGitlabFingerprintStable(t *testing.T) { + updatedAt := mustTime(t, "2026-01-03T00:00:00Z") + mr := gitlabMergeRequest{ + IID: 7, + Title: "Add syncer", + Description: "MR body", + State: "opened", + WebURL: "https://gitlab.com/owner/repo/-/merge_requests/7", + UpdatedAt: updatedAt, + Author: &gitlabUser{Name: "Alice", Username: "alice"}, + } + fp1 := mr.toSourceDocument().Fingerprint + fp2 := mr.toSourceDocument().Fingerprint + if fp1 == "" || fp1 != fp2 { + t.Fatalf("fingerprint unstable: %q %q", fp1, fp2) + } + + changed := mr + changed.Title = "Add syncer v2" + if got := changed.toSourceDocument().Fingerprint; got == fp1 { + t.Fatalf("fingerprint did not change after title update") + } +} + +func TestGitlabCodeFileFingerprintUsesTreeItemID(t *testing.T) { + item := gitlabTreeItem{ID: "abc", Name: "main.go", Type: "blob", Path: "main.go"} + base := gitlabCodeFileFingerprint(item, "main") + if base == "" { + t.Fatalf("fingerprint is empty") + } + if got := gitlabCodeFileFingerprint(item, "main"); got != base { + t.Fatalf("fingerprint unstable: %q %q", base, got) + } + + changed := item + changed.ID = "def" + if got := gitlabCodeFileFingerprint(changed, "main"); got == base { + t.Fatalf("fingerprint did not change after tree item ID update") + } +} + +// TestGitlabConnectorOpenPrune verifies PRUNE returns correct web_url IDs. +func TestGitlabConnectorOpenPrune(t *testing.T) { + connector, err := NewGitlabConnector(map[string]any{ + "project_owner": "owner", + "project_name": "repo", + "include_mrs": true, + "include_issues": true, + "include_code_files": false, + "batch_size": 10, + "credentials": map[string]any{"gitlab_access_token": "token"}, + }) + if err != nil { + t.Fatalf("NewGitlabConnector failed: %v", err) + } + connector.baseURL = "https://gitlab.com/api/v4" + connector.doJSON = gitlabFixtureDoJSON(t) + + session, err := connector.OpenPrune(context.Background(), PruneRequest{}) + if err != nil { + t.Fatalf("OpenPrune failed: %v", err) + } + batch, err := session.NextBatch(context.Background()) + if err != nil { + t.Fatalf("NextBatch failed: %v", err) + } + got := []string{} + for _, doc := range batch.Documents { + got = append(got, doc.SourceID) + } + want := []string{ + "https://gitlab.com/owner/repo/-/merge_requests/7", + "https://gitlab.com/owner/repo/-/issues/3", + } + if len(got) != len(want) { + t.Fatalf("ids len = %d, want %d: %v", len(got), len(want), got) + } + for i := range want { + if got[i] != want[i] { + t.Fatalf("ids[%d] = %q, want %q", i, got[i], want[i]) + } + } +} + +// TestGitlabConnectorOpenSyncResumesAfterCheckpoint verifies retry skips committed GitLab documents. +func TestGitlabConnectorOpenSyncResumesAfterCheckpoint(t *testing.T) { + connector, err := NewGitlabConnector(map[string]any{ + "project_owner": "owner", + "project_name": "repo", + "include_mrs": true, + "include_issues": true, + "include_code_files": false, + "batch_size": 1, + "credentials": map[string]any{"gitlab_access_token": "token"}, + }) + if err != nil { + t.Fatalf("NewGitlabConnector failed: %v", err) + } + connector.baseURL = "https://gitlab.com/api/v4" + connector.doJSON = gitlabFixtureDoJSON(t) + + end := mustTime(t, "2026-01-04T00:00:00Z") + session, err := connector.OpenSync(context.Background(), SyncRequest{FromBeginning: true, WindowEnd: end}) + if err != nil { + t.Fatalf("OpenSync failed: %v", err) + } + first, err := session.NextBatch(context.Background()) + if err != nil { + t.Fatalf("NextBatch first failed: %v", err) + } + if len(first.Documents) != 1 || first.Documents[0].SourceID != "https://gitlab.com/owner/repo/-/merge_requests/7" { + t.Fatalf("first documents = %+v, want MR 7", first.Documents) + } + if first.Checkpoint == nil || first.Checkpoint.SourceID != "https://gitlab.com/owner/repo/-/merge_requests/7" { + t.Fatalf("first checkpoint = %+v, want MR 7", first.Checkpoint) + } + + resumed, err := connector.OpenSync(context.Background(), SyncRequest{FromBeginning: true, WindowEnd: end, Resume: first.Checkpoint}) + if err != nil { + t.Fatalf("resume OpenSync failed: %v", err) + } + second, err := resumed.NextBatch(context.Background()) + if err != nil { + t.Fatalf("resume NextBatch failed: %v", err) + } + if len(second.Documents) != 1 || second.Documents[0].SourceID != "https://gitlab.com/owner/repo/-/issues/3" { + t.Fatalf("resume documents = %+v, want issue 3", second.Documents) + } + if second.Checkpoint == nil || second.Checkpoint.SourceID != "https://gitlab.com/owner/repo/-/issues/3" { + t.Fatalf("resume checkpoint = %+v, want issue 3", second.Checkpoint) + } + if _, err = resumed.NextBatch(context.Background()); !errors.Is(err, io.EOF) { + t.Fatalf("resume EOF = %v", err) + } +} + +// TestGitlabConnectorOpenSyncResumeOffsetFallbackSkipsCommittedDocument verifies offset fallback excludes the committed offset. +func TestGitlabConnectorOpenSyncResumeOffsetFallbackSkipsCommittedDocument(t *testing.T) { + connector, err := NewGitlabConnector(map[string]any{ + "project_owner": "owner", + "project_name": "repo", + "include_mrs": true, + "include_issues": true, + "include_code_files": false, + "batch_size": 1, + "credentials": map[string]any{"gitlab_access_token": "token"}, + }) + if err != nil { + t.Fatalf("NewGitlabConnector failed: %v", err) + } + connector.baseURL = "https://gitlab.com/api/v4" + connector.doJSON = gitlabFixtureDoJSON(t) + + end := mustTime(t, "2026-01-04T00:00:00Z") + session, err := connector.OpenSync(context.Background(), SyncRequest{FromBeginning: true, WindowEnd: end}) + if err != nil { + t.Fatalf("OpenSync failed: %v", err) + } + first, err := session.NextBatch(context.Background()) + if err != nil { + t.Fatalf("NextBatch first failed: %v", err) + } + if len(first.Documents) != 1 || first.Documents[0].SourceID != "https://gitlab.com/owner/repo/-/merge_requests/7" { + t.Fatalf("first documents = %+v, want MR 7", first.Documents) + } + if first.Checkpoint == nil { + t.Fatalf("first checkpoint is nil") + } + + resumeCheckpoint := cloneGitlabCheckpointWithMissingSourceID(t, first.Checkpoint) + resumed, err := connector.OpenSync(context.Background(), SyncRequest{FromBeginning: true, WindowEnd: end, Resume: resumeCheckpoint}) + if err != nil { + t.Fatalf("resume OpenSync failed: %v", err) + } + second, err := resumed.NextBatch(context.Background()) + if err != nil { + t.Fatalf("resume NextBatch failed: %v", err) + } + if len(second.Documents) != 1 || second.Documents[0].SourceID != "https://gitlab.com/owner/repo/-/issues/3" { + t.Fatalf("resume documents = %+v, want issue 3", second.Documents) + } + if second.Documents[0].SourceID == first.Documents[0].SourceID { + t.Fatalf("committed document was redelivered: %s", second.Documents[0].SourceID) + } +} + +// TestGitlabConnectorOpenSyncCodeFiles verifies BFS traversal of code files with lazy Fetch. +func TestGitlabConnectorOpenSyncCodeFiles(t *testing.T) { + connector, err := NewGitlabConnector(map[string]any{ + "project_owner": "owner", + "project_name": "repo", + "include_mrs": false, + "include_issues": false, + "include_code_files": true, + "batch_size": 10, + "credentials": map[string]any{"gitlab_access_token": "token"}, + }) + if err != nil { + t.Fatalf("NewGitlabConnector failed: %v", err) + } + connector.baseURL = "https://gitlab.com/api/v4" + connector.doJSON = gitlabFixtureDoJSON(t) + connector.doRaw = gitlabFixtureDoRaw(t) + + end := mustTime(t, "2026-01-04T00:00:00Z") + session, err := connector.OpenSync(context.Background(), SyncRequest{FromBeginning: true, WindowEnd: end}) + if err != nil { + t.Fatalf("OpenSync failed: %v", err) + } + batch, err := session.NextBatch(context.Background()) + if err != nil { + t.Fatalf("NextBatch failed: %v", err) + } + if len(batch.Documents) != 2 { + t.Fatalf("documents len = %d, want 2 (main.go + src/helper.go)", len(batch.Documents)) + } + if batch.Documents[0].SourceID != "https://gitlab.com/owner/repo/-/blob/main/main.go" { + t.Fatalf("doc[0] source id = %q", batch.Documents[0].SourceID) + } + if batch.Documents[1].SourceID != "https://gitlab.com/owner/repo/-/blob/main/src/helper.go" { + t.Fatalf("doc[1] source id = %q", batch.Documents[1].SourceID) + } + if batch.Documents[0].FetchRef == nil { + t.Fatalf("doc[0] has no FetchRef") + } + fetcher, ok := session.(Fetcher) + if !ok { + t.Fatalf("session does not implement Fetcher") + } + blob, err := fetcher.Fetch(context.Background(), *batch.Documents[0].FetchRef) + if err != nil { + t.Fatalf("Fetch failed: %v", err) + } + if string(blob) != "package main\n" { + t.Fatalf("blob = %q, want %q", string(blob), "package main\n") + } + if _, err = session.NextBatch(context.Background()); !errors.Is(err, io.EOF) { + t.Fatalf("NextBatch EOF = %v", err) + } +} + +// TestGitlabFetchFileRawEscapesFilePath verifies raw file paths use path-safe %20/%2F encoding. +func TestGitlabFetchFileRawEscapesFilePath(t *testing.T) { + connector := &GitlabConnector{baseURL: "https://gitlab.com/api/v4"} + connector.doRaw = func(ctx context.Context, apiURL string) ([]byte, error) { + parsed, err := url.Parse(apiURL) + if err != nil { + t.Fatalf("parse api url: %v", err) + } + if got := parsed.EscapedPath(); got != "/api/v4/projects/1/repository/files/src%20dir%2Fhelper%20one.go/raw" { + t.Fatalf("escaped path = %q", got) + } + if got := parsed.Query().Get("ref"); got != "main" { + t.Fatalf("ref = %q, want %q", got, "main") + } + return []byte("ok"), nil + } + body, err := connector.fetchFileRaw(context.Background(), 1, "main", "src dir/helper one.go") + if err != nil { + t.Fatalf("fetchFileRaw failed: %v", err) + } + if string(body) != "ok" { + t.Fatalf("body = %q, want %q", body, "ok") + } +} + +// TestGitlabConnectorValidate verifies missing config returns clear errors. +func TestGitlabConnectorValidate(t *testing.T) { + cases := []struct { + name string + config map[string]any + want string + }{ + {"missing owner", map[string]any{ + "project_name": "repo", + "credentials": map[string]any{"gitlab_access_token": "token"}, + }, "project_owner"}, + {"missing token", map[string]any{ + "project_owner": "owner", + "project_name": "repo", + "credentials": map[string]any{}, + }, "gitlab_access_token"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + connector, err := NewGitlabConnector(tc.config) + if err != nil { + t.Fatalf("NewGitlabConnector failed: %v", err) + } + err = connector.Validate(context.Background()) + if err == nil || !strings.Contains(err.Error(), tc.want) { + t.Fatalf("err = %v, want contains %q", err, tc.want) + } + }) + } +} + +func cloneGitlabCheckpointWithMissingSourceID(t *testing.T, checkpoint *SyncCheckpoint) *SyncCheckpoint { + t.Helper() + var cursor gitlabSyncCursor + if err := json.Unmarshal([]byte(checkpoint.Cursor), &cursor); err != nil { + t.Fatalf("decode checkpoint cursor: %v", err) + } + cursor.SourceID = "" + data, err := json.Marshal(cursor) + if err != nil { + t.Fatalf("encode checkpoint cursor: %v", err) + } + clone := *checkpoint + clone.Cursor = string(data) + clone.SourceID = "" + return &clone +} + +// gitlabFixtureDoJSON returns a fixture GitLab JSON transport. +func gitlabFixtureDoJSON(t *testing.T) func(ctx context.Context, apiURL string, out any) (http.Header, error) { + t.Helper() + return func(ctx context.Context, apiURL string, out any) (http.Header, error) { + parsed, err := url.Parse(apiURL) + if err != nil { + t.Fatalf("parse api url: %v", err) + } + path := parsed.EscapedPath() + query := parsed.Query() + + var body string + switch { + case strings.HasSuffix(path, "/projects/owner%2Frepo"): + body = `{"id":1,"default_branch":"main","web_url":"https://gitlab.com/owner/repo"}` + case strings.HasSuffix(path, "/projects/1/merge_requests"): + body = `[{ + "iid":7, + "title":"Add syncer", + "description":"MR body", + "state":"opened", + "web_url":"https://gitlab.com/owner/repo/-/merge_requests/7", + "updated_at":"2026-01-03T00:00:00.000Z", + "author":{"name":"Alice","username":"alice"} + }]` + case strings.HasSuffix(path, "/projects/1/issues"): + body = `[{ + "iid":3, + "title":"Prune bug", + "description":"Issue body", + "state":"opened", + "web_url":"https://gitlab.com/owner/repo/-/issues/3", + "updated_at":"2026-01-02T00:00:00.000Z", + "author":{"name":"Bob","username":"bob"}, + "type":"ISSUE" + }]` + case strings.HasSuffix(path, "/projects/1/repository/tree"): + treePath := query.Get("path") + if treePath == "" { + body = `[ + {"id":"abc","name":"main.go","type":"blob","path":"main.go","mode":"100644"}, + {"id":"def","name":"src","type":"tree","path":"src","mode":"040000"} + ]` + } else if treePath == "src" { + body = `[ + {"id":"ghi","name":"helper.go","type":"blob","path":"src/helper.go","mode":"100644"} + ]` + } else { + t.Fatalf("unexpected tree path %s", treePath) + } + case strings.HasSuffix(path, "/projects/1/repository/commits"): + commitPath := query.Get("path") + if commitPath == "main.go" { + body = `[{"id":"commit1","committed_date":"2026-01-03T00:00:00.000Z"}]` + } else if commitPath == "src/helper.go" { + body = `[{"id":"commit2","committed_date":"2026-01-02T00:00:00.000Z"}]` + } else { + t.Fatalf("unexpected commit path %s", commitPath) + } + default: + t.Fatalf("unexpected api path %s", path) + } + if err = json.Unmarshal([]byte(body), out); err != nil { + t.Fatalf("decode fixture: %v", err) + } + return http.Header{}, nil + } +} + +// gitlabFixtureDoRaw returns fixture raw file content. +func gitlabFixtureDoRaw(t *testing.T) func(ctx context.Context, apiURL string) ([]byte, error) { + t.Helper() + return func(ctx context.Context, apiURL string) ([]byte, error) { + parsed, err := url.Parse(apiURL) + if err != nil { + t.Fatalf("parse api url: %v", err) + } + path := parsed.EscapedPath() + if strings.HasSuffix(path, "/files/main.go/raw") { + return []byte("package main\n"), nil + } + if strings.HasSuffix(path, "/files/src%2Fhelper.go/raw") { + return []byte("package src\n"), nil + } + t.Fatalf("unexpected raw path %s", path) + return nil, nil + } +}