diff --git a/internal/cli/poll.go b/internal/cli/poll.go new file mode 100644 index 000000000..69ad0cc31 --- /dev/null +++ b/internal/cli/poll.go @@ -0,0 +1,96 @@ +package cli + +import ( + "fmt" + "os" + + "github.com/spf13/cobra" + + "github.com/fullsend-ai/fullsend/internal/poll" +) + +func newPollCmd() *cobra.Command { + var ( + forgeFlag string + projectPath string + gitlabURL string + outputPath string + pollModeFlag string + ) + + cmd := &cobra.Command{ + Use: "poll", + Short: "Poll GitLab API for new events and dispatch agent stages", + RunE: func(cmd *cobra.Command, args []string) error { + if forgeFlag != "gitlab" { + return fmt.Errorf("poll command currently supports --forge gitlab only (got %q)", forgeFlag) + } + + forgeToken := os.Getenv("FULLSEND_FORGE_TOKEN") + if forgeToken == "" { + return fmt.Errorf("FULLSEND_FORGE_TOKEN is required") + } + + if projectPath == "" { + projectPath = os.Getenv("CI_PROJECT_PATH") + } + if projectPath == "" { + return fmt.Errorf("--project or CI_PROJECT_PATH is required") + } + + slashCommandsOnly := pollModeFlag == "fast" || os.Getenv("FULLSEND_POLL_MODE") == "fast" + + // The GitLab client is not yet implemented (Phase 1). + // This command will be fully wired when the GitLab forge + // client provides a type satisfying poll.GitLabClient. + _ = forgeToken + _ = gitlabURL + + var botUserID int + // botUserID will be resolved via client.GetAuthenticatedUser + // once the GitLab client is available. + + opts := poll.Options{ + SlashCommandsOnly: slashCommandsOnly, + BotUserID: botUserID, + OutputPath: outputPath, + GitLabURL: gitlabURL, + } + + // TODO(phase1): Replace nil client and router with real + // implementations once the GitLab forge client exists. + poller := poll.New(nil, nil, projectPath, opts) + return poller.Run(cmd.Context()) + }, + } + + cmd.Flags().StringVar(&forgeFlag, "forge", "", "Forge platform (required: gitlab)") + _ = cmd.MarkFlagRequired("forge") + cmd.Flags().StringVar(&projectPath, "project", "", "GitLab project path (default: $CI_PROJECT_PATH)") + cmd.Flags().StringVar(&gitlabURL, "gitlab-url", "https://gitlab.com", "GitLab instance URL") + cmd.Flags().StringVar(&outputPath, "output", "", "Path to write dispatches JSON") + cmd.Flags().StringVar(&pollModeFlag, "poll-mode", "", "Poll mode: fast (slash commands only) or full") + + cmd.AddCommand(newPollGenerateChildPipelineCmd()) + return cmd +} + +func newPollGenerateChildPipelineCmd() *cobra.Command { + var ( + dispatchesPath string + outputPath string + ) + + cmd := &cobra.Command{ + Use: "generate-child-pipeline", + Short: "Generate child pipeline YAML from dispatches JSON", + RunE: func(cmd *cobra.Command, args []string) error { + return poll.GenerateChildPipelineFromFile(dispatchesPath, outputPath) + }, + } + + cmd.Flags().StringVar(&dispatchesPath, "dispatches", "dispatches.json", "Path to dispatches JSON file") + cmd.Flags().StringVar(&outputPath, "output", "child-pipeline.yml", "Path to write child pipeline YAML") + + return cmd +} diff --git a/internal/cli/root.go b/internal/cli/root.go index 74173e177..a4fc82d76 100644 --- a/internal/cli/root.go +++ b/internal/cli/root.go @@ -52,6 +52,7 @@ func newRootCmd() *cobra.Command { cmd.AddCommand(newPostReviewCmd()) cmd.AddCommand(newPostCommentCmd()) cmd.AddCommand(newReconcileStatusCmd()) + cmd.AddCommand(newPollCmd()) return cmd } diff --git a/internal/dispatch/event.go b/internal/dispatch/event.go new file mode 100644 index 000000000..fd564a821 --- /dev/null +++ b/internal/dispatch/event.go @@ -0,0 +1,92 @@ +package dispatch + +// NormalizedEvent is the forge-neutral routing input for dispatch and +// harness CEL trigger evaluation. See docs/normative/normalized-event/v1/. +type NormalizedEvent struct { + Repo string `json:"repo"` + Entity Entity `json:"entity"` + Transition Transition `json:"transition"` + Actor Actor `json:"actor"` + State State `json:"state"` + Source Source `json:"source"` +} + +// Entity identifies the work item or change proposal the event acts on. +type Entity struct { + Kind string `json:"kind"` + ID int `json:"id"` + URL string `json:"url"` + Key string `json:"key,omitempty"` + LinkedChangeProposal *LinkedChangeProposal `json:"linked_change_proposal,omitempty"` +} + +// LinkedChangeProposal links a work_item entity to its associated change proposal. +type LinkedChangeProposal struct { + ID int `json:"id"` + URL string `json:"url"` +} + +// Transition describes the lifecycle event that occurred. +type Transition struct { + Kind string `json:"kind"` + Label *TransitionLabel `json:"label,omitempty"` + Comment *TransitionComment `json:"comment,omitempty"` + Review *TransitionReview `json:"review,omitempty"` +} + +// TransitionLabel carries label change details (kind == "label_changed"). +type TransitionLabel struct { + Name string `json:"name"` + Action string `json:"action"` +} + +// TransitionComment carries comment details (kind == "comment_added"). +type TransitionComment struct { + Command string `json:"command,omitempty"` + Body string `json:"body"` + Instruction string `json:"instruction,omitempty"` +} + +// TransitionReview carries review details (kind == "review_submitted"). +type TransitionReview struct { + State string `json:"state"` + ReviewerID string `json:"reviewer_id"` +} + +// Actor identifies who triggered the event. +type Actor struct { + ID string `json:"id"` + Kind string `json:"kind"` + Role string `json:"role"` + IsEntityAuthor bool `json:"is_entity_author"` +} + +// State captures the entity's state at event time. +type State struct { + Labels []string `json:"labels"` + ChangeProposal *ChangeProposalState `json:"change_proposal,omitempty"` +} + +// ChangeProposalState carries MR/PR metadata needed by stages. +type ChangeProposalState struct { + ID int `json:"id"` + HeadRepo string `json:"head_repo"` + BaseRepo string `json:"base_repo"` + HeadRef string `json:"head_ref"` + BaseRef string `json:"base_ref"` + HeadSHA string `json:"head_sha,omitempty"` + AuthorID string `json:"author_id"` + IsFork bool `json:"is_fork"` +} + +// Source records event provenance. +type Source struct { + System string `json:"system"` + RawType string `json:"raw_type"` + RawAction string `json:"raw_action,omitempty"` +} + +// EventRouter routes a NormalizedEvent to zero or more stage names. +type EventRouter interface { + Route(event *NormalizedEvent) ([]string, error) +} diff --git a/internal/poll/client.go b/internal/poll/client.go new file mode 100644 index 000000000..47c77cecc --- /dev/null +++ b/internal/poll/client.go @@ -0,0 +1,97 @@ +// Package poll implements the GitLab cron-polling event dispatch loop. +// It discovers events from the GitLab API, converts them to +// NormalizedEvents, routes them through the dispatch core, and +// triggers child pipelines for matched stages. +package poll + +import ( + "context" + "time" +) + +// GitLabClient defines the GitLab API surface the poller requires. +// The interface is satisfied by the GitLab forge client (Phase 1). +type GitLabClient interface { + ListIssuesUpdatedSince(ctx context.Context, owner, repo string, since time.Time) ([]Issue, error) + ListMergeRequestsUpdatedSince(ctx context.Context, owner, repo string, since time.Time) ([]MergeRequest, error) + ListProjectEvents(ctx context.Context, owner, repo string, targetType string, after time.Time) ([]ProjectEvent, error) + ListIssueNotes(ctx context.Context, owner, repo string, issueIID int) ([]Note, error) + ListMergeRequestNotes(ctx context.Context, owner, repo string, mrIID int) ([]Note, error) + ListResourceLabelEvents(ctx context.Context, owner, repo string, issueIID int) ([]ResourceLabelEvent, error) + GetCIVariable(ctx context.Context, owner, repo, name string) (string, error) + UpdateCIVariable(ctx context.Context, owner, repo, name, value string, protected bool) error + GetAuthenticatedUser(ctx context.Context) (string, error) + CreateNoteAwardEmoji(ctx context.Context, owner, repo string, noteableIID, noteID int, emoji string) error + GetIssue(ctx context.Context, owner, repo string, issueIID int) (*Issue, error) + GetMemberAccessLevel(ctx context.Context, owner, repo string, userID int) (int, error) + GetProjectPath(ctx context.Context, projectID int) (string, error) +} + +// Issue represents a GitLab issue as returned by the API. +type Issue struct { + IID int `json:"iid"` + Title string `json:"title"` + State string `json:"state"` + Labels []string `json:"labels"` + AuthorID int `json:"author_id"` + UpdatedAt time.Time `json:"updated_at"` +} + +// MergeRequest represents a GitLab merge request as returned by the API. +type MergeRequest struct { + IID int `json:"iid"` + Title string `json:"title"` + State string `json:"state"` + Labels []string `json:"labels"` + SourceProjectID int `json:"source_project_id"` + TargetProjectID int `json:"target_project_id"` + SourceBranch string `json:"source_branch"` + TargetBranch string `json:"target_branch"` + Author UserRef `json:"author"` + AuthorID int `json:"author_id"` + MergedByID int `json:"merged_by_id"` + MergedBy UserRef `json:"merged_by"` + MergedAt time.Time `json:"merged_at"` + UpdatedAt time.Time `json:"updated_at"` +} + +// Note represents a GitLab note (comment) on an issue or MR. +type Note struct { + ID int `json:"id"` + Body string `json:"body"` + Author UserRef `json:"author"` + CreatedAt time.Time `json:"created_at"` +} + +// UserRef is a minimal user reference from the GitLab API. +type UserRef struct { + ID int `json:"id"` + Username string `json:"username"` + Bot bool `json:"bot"` +} + +// ProjectEvent represents an event from the GitLab Events API. +type ProjectEvent struct { + ID int `json:"id"` + Author UserRef `json:"author"` + Note EventNote `json:"note"` + CreatedAt time.Time `json:"created_at"` +} + +// EventNote is the embedded note object in a project event. +type EventNote struct { + ID int `json:"id"` + NoteableType string `json:"noteable_type"` + NoteableIID int `json:"noteable_iid"` + Body string `json:"body"` +} + +// ResourceLabelEvent represents a label change event from the GitLab API. +type ResourceLabelEvent struct { + ID int `json:"id"` + Action string `json:"action"` + Label struct { + Name string `json:"name"` + } `json:"label"` + User UserRef `json:"user"` +} diff --git a/internal/poll/convert.go b/internal/poll/convert.go new file mode 100644 index 000000000..ab06a4b71 --- /dev/null +++ b/internal/poll/convert.go @@ -0,0 +1,298 @@ +package poll + +import ( + "context" + "fmt" + "log" + "strconv" + "strings" + + "github.com/fullsend-ai/fullsend/internal/dispatch" +) + +// toNormalizedEvent converts a RoutableEvent into a dispatch.NormalizedEvent +// suitable for stage matching and child-pipeline triggering. +func (p *Poller) toNormalizedEvent(ctx context.Context, event RoutableEvent) (dispatch.NormalizedEvent, error) { + ne := dispatch.NormalizedEvent{ + Repo: p.projectPath, + Source: dispatch.Source{ + System: "gitlab", + RawType: mapRawType(event.Type), + RawAction: mapRawAction(event.Type), + }, + Entity: dispatch.Entity{ + Kind: entityKind(event.Type), + ID: event.IID, + URL: entityURL(p.gitlabURL, p.projectPath, event.Type, event.IID), + }, + Transition: dispatch.Transition{ + Kind: translateEventType(event.Type), + }, + State: dispatch.State{ + Labels: event.Labels, + }, + } + + // Populate transition details based on event type. + switch event.Type { + case "issue_label": + if event.ChangedLabel != "" { + ne.Transition.Label = &dispatch.TransitionLabel{ + Name: event.ChangedLabel, + Action: "added", + } + } + case "issue_note", "mr_note": + cmd, instruction := extractCommand(event.NoteBody) + ne.Transition.Comment = &dispatch.TransitionComment{ + Body: truncate(event.NoteBody, 4096), + Command: cmd, + Instruction: truncate(instruction, 4096), + } + } + + // Resolve actor identity. + var authorID int + var actorLogin string + var isBot bool + + switch event.Type { + case "issue_note", "mr_note": + authorID = event.NoteAuthorID + actorLogin = event.NoteAuthorLogin + isBot = event.IsBot + case "issue_label": + if event.ChangedLabel != "" { + la, err := p.resolveLabelAuthor(ctx, event.IID, event.ChangedLabel) + if err != nil { + return dispatch.NormalizedEvent{}, fmt.Errorf("resolve label author: %w", err) + } + authorID = la.ID + actorLogin = la.Username + isBot = la.IsBot + } + case "mr_event": + authorID = event.NoteAuthorID + actorLogin = event.MergedByLogin + isBot = event.IsBot + } + + if authorID == 0 || actorLogin == "" { + return dispatch.NormalizedEvent{}, fmt.Errorf("unresolvable actor") + } + + actorKind := "human" + if isBot { + actorKind = "bot" + } + + ne.Actor = dispatch.Actor{ + ID: actorLogin, + Kind: actorKind, + Role: p.resolveActorRole(ctx, authorID), + } + + // Best-effort: determine if the actor authored the entity. + ne.Actor.IsEntityAuthor = p.isEntityAuthor(ctx, event, authorID) + + // For MR events, populate ChangeProposalState. + if event.Type == "mr_note" || event.Type == "mr_event" { + cpState, err := p.buildChangeProposalState(ctx, event) + if err != nil { + log.Printf("WARNING: buildChangeProposalState for %s IID %d: %v", event.Type, event.IID, err) + } else { + ne.State.ChangeProposal = cpState + } + } + + return ne, nil +} + +// translateEventType maps a RoutableEvent type to a NormalizedEvent +// transition kind. +func translateEventType(eventType string) string { + switch eventType { + case "issue_label": + return "label_changed" + case "issue_note": + return "comment_added" + case "mr_note": + return "comment_added" + case "mr_event": + return "merged" + default: + return eventType + } +} + +// resolveLabelAuthor determines who applied a label by inspecting the +// resource label events API for the issue. +func (p *Poller) resolveLabelAuthor(ctx context.Context, issueIID int, labelName string) (LabelAuthor, error) { + events, err := p.client.ListResourceLabelEvents(ctx, p.owner, p.repo, issueIID) + if err != nil { + return LabelAuthor{}, fmt.Errorf("list resource label events: %w", err) + } + + // Iterate in reverse to find the most recent "add" event for the label. + for i := len(events) - 1; i >= 0; i-- { + e := events[i] + if e.Action == "add" && e.Label.Name == labelName { + return LabelAuthor{ + ID: e.User.ID, + Username: e.User.Username, + IsBot: e.User.Bot, + }, nil + } + } + + return LabelAuthor{}, fmt.Errorf("no add event found for label %q on issue %d", labelName, issueIID) +} + +// resolveActorRole maps a GitLab project member's access level to a +// normalized role string. +func (p *Poller) resolveActorRole(ctx context.Context, userID int) string { + level, err := p.client.GetMemberAccessLevel(ctx, p.owner, p.repo, userID) + if err != nil { + return "none" + } + + switch level { + case 10: // Guest + return "read" + case 20: // Reporter + return "triage" + case 30: // Developer + return "write" + case 40: // Maintainer + return "maintain" + case 50: // Owner + return "admin" + default: + return "none" + } +} + +// extractCommand parses a note body for a /fs- slash command. It returns +// the command token and the remaining instruction text. If no slash command +// is found, both return values are empty strings. +func extractCommand(body string) (command, instruction string) { + trimmed := strings.TrimSpace(body) + if trimmed == "" { + return "", "" + } + + firstLine := strings.SplitN(trimmed, "\n", 2)[0] + tokens := strings.Fields(firstLine) + if len(tokens) == 0 { + return "", "" + } + + if !strings.HasPrefix(tokens[0], "/fs-") { + return "", "" + } + + command = tokens[0] + + // Instruction is everything after the command token in the full body. + after := strings.TrimSpace(trimmed[len(command):]) + return command, after +} + +func mapRawType(eventType string) string { + switch eventType { + case "issue_label", "issue_note": + return "issues" + case "mr_note", "mr_event": + return "merge_request" + default: + return eventType + } +} + +func mapRawAction(eventType string) string { + switch eventType { + case "issue_label": + return "labeled" + case "issue_note", "mr_note": + return "commented" + case "mr_event": + return "merged" + default: + return "" + } +} + +// entityKind returns the Entity.Kind based on the event type. +func entityKind(eventType string) string { + switch eventType { + case "issue_label", "issue_note": + return "work_item" + case "mr_note", "mr_event": + return "change_proposal" + default: + return "work_item" + } +} + +// entityURL constructs the full URL for the entity on the GitLab instance. +func entityURL(gitlabURL, projectPath, eventType string, iid int) string { + base := strings.TrimRight(gitlabURL, "/") + "/" + projectPath + switch eventType { + case "issue_label", "issue_note": + return base + "/-/issues/" + strconv.Itoa(iid) + case "mr_note", "mr_event": + return base + "/-/merge_requests/" + strconv.Itoa(iid) + default: + return base + "/-/issues/" + strconv.Itoa(iid) + } +} + +// isEntityAuthor determines whether the actor is the author of the entity. +// It uses a best-effort approach, returning false if the author cannot be +// determined. +func (p *Poller) isEntityAuthor(ctx context.Context, event RoutableEvent, actorID int) bool { + switch event.Type { + case "issue_label", "issue_note": + issue, err := p.client.GetIssue(ctx, p.owner, p.repo, event.IID) + if err != nil { + return false + } + return issue.AuthorID == actorID + case "mr_note", "mr_event": + return event.MRAuthorID != 0 && event.MRAuthorID == actorID + default: + return false + } +} + +// buildChangeProposalState populates the ChangeProposalState for MR events +// by resolving project paths from project IDs. +func (p *Poller) buildChangeProposalState(ctx context.Context, event RoutableEvent) (*dispatch.ChangeProposalState, error) { + headRepo, err := p.client.GetProjectPath(ctx, event.MRSource) + if err != nil { + return nil, fmt.Errorf("resolve head repo: %w", err) + } + + baseRepo, err := p.client.GetProjectPath(ctx, event.MRTarget) + if err != nil { + return nil, fmt.Errorf("resolve base repo: %w", err) + } + + return &dispatch.ChangeProposalState{ + ID: event.IID, + HeadRepo: headRepo, + BaseRepo: baseRepo, + HeadRef: event.SourceBranch, + BaseRef: event.TargetBranch, + AuthorID: event.MRAuthorLogin, + IsFork: event.MRSource != event.MRTarget, + }, nil +} + +func truncate(s string, maxLen int) string { + r := []rune(s) + if len(r) <= maxLen { + return s + } + return string(r[:maxLen]) +} diff --git a/internal/poll/convert_test.go b/internal/poll/convert_test.go new file mode 100644 index 000000000..d460680b0 --- /dev/null +++ b/internal/poll/convert_test.go @@ -0,0 +1,669 @@ +package poll + +import ( + "context" + "fmt" + "strings" + "testing" +) + +func TestToNormalizedEvent_IssueNote(t *testing.T) { + mc := newMockClient() + mc.memberLevel[42] = 30 // Developer -> "write" + mc.issue[1] = &Issue{IID: 1, AuthorID: 42} // actor is the author + p := newEventsPoller(mc) + + event := RoutableEvent{ + Type: "issue_note", + IID: 1, + NoteBody: "looks good to me", + NoteID: 10, + NoteAuthorID: 42, + NoteAuthorLogin: "alice", + IsBot: false, + Labels: []string{"bug"}, + } + + ne, err := p.toNormalizedEvent(context.Background(), event) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if ne.Repo != "group/project" { + t.Errorf("Repo = %q, want %q", ne.Repo, "group/project") + } + if ne.Source.System != "gitlab" { + t.Errorf("Source.System = %q, want %q", ne.Source.System, "gitlab") + } + if ne.Source.RawType != "issues" { + t.Errorf("Source.RawType = %q, want %q", ne.Source.RawType, "issues") + } + if ne.Entity.Kind != "work_item" { + t.Errorf("Entity.Kind = %q, want %q", ne.Entity.Kind, "work_item") + } + if ne.Entity.ID != 1 { + t.Errorf("Entity.ID = %d, want 1", ne.Entity.ID) + } + if ne.Entity.URL != "https://gitlab.com/group/project/-/issues/1" { + t.Errorf("Entity.URL = %q, want %q", ne.Entity.URL, "https://gitlab.com/group/project/-/issues/1") + } + if ne.Transition.Kind != "comment_added" { + t.Errorf("Transition.Kind = %q, want %q", ne.Transition.Kind, "comment_added") + } + if ne.Transition.Comment == nil { + t.Fatal("expected Transition.Comment to be set") + } + if ne.Transition.Comment.Body != "looks good to me" { + t.Errorf("Comment.Body = %q, want %q", ne.Transition.Comment.Body, "looks good to me") + } + if ne.Actor.ID != "alice" { + t.Errorf("Actor.ID = %q, want %q", ne.Actor.ID, "alice") + } + if ne.Actor.Kind != "human" { + t.Errorf("Actor.Kind = %q, want %q", ne.Actor.Kind, "human") + } + if ne.Actor.Role != "write" { + t.Errorf("Actor.Role = %q, want %q", ne.Actor.Role, "write") + } + if !ne.Actor.IsEntityAuthor { + t.Error("expected Actor.IsEntityAuthor to be true") + } + if len(ne.State.Labels) != 1 || ne.State.Labels[0] != "bug" { + t.Errorf("State.Labels = %v, want [bug]", ne.State.Labels) + } +} + +func TestToNormalizedEvent_IssueLabel(t *testing.T) { + mc := newMockClient() + mc.labelEvents[1] = []ResourceLabelEvent{ + { + ID: 100, + Action: "add", + Label: struct { + Name string `json:"name"` + }{Name: "ready-to-code"}, + User: UserRef{ID: 55, Username: "carol", Bot: false}, + }, + } + mc.memberLevel[55] = 40 // Maintainer -> "maintain" + mc.issue[1] = &Issue{IID: 1, AuthorID: 99} // actor 55 is NOT the author + p := newEventsPoller(mc) + + event := RoutableEvent{ + Type: "issue_label", + IID: 1, + Labels: []string{"ready-to-code"}, + ChangedLabel: "ready-to-code", + } + + ne, err := p.toNormalizedEvent(context.Background(), event) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if ne.Transition.Kind != "label_changed" { + t.Errorf("Transition.Kind = %q, want %q", ne.Transition.Kind, "label_changed") + } + if ne.Transition.Label == nil { + t.Fatal("expected Transition.Label to be set") + } + if ne.Transition.Label.Name != "ready-to-code" { + t.Errorf("Label.Name = %q, want %q", ne.Transition.Label.Name, "ready-to-code") + } + if ne.Transition.Label.Action != "added" { + t.Errorf("Label.Action = %q, want %q", ne.Transition.Label.Action, "added") + } + if ne.Actor.ID != "carol" { + t.Errorf("Actor.ID = %q, want %q (resolved from label event)", ne.Actor.ID, "carol") + } + if ne.Actor.Kind != "human" { + t.Errorf("Actor.Kind = %q, want %q", ne.Actor.Kind, "human") + } + if ne.Actor.Role != "maintain" { + t.Errorf("Actor.Role = %q, want %q", ne.Actor.Role, "maintain") + } + if ne.Actor.IsEntityAuthor { + t.Error("expected Actor.IsEntityAuthor to be false") + } +} + +func TestToNormalizedEvent_MRNote(t *testing.T) { + mc := newMockClient() + mc.memberLevel[42] = 30 + mc.projectPaths[1] = "group/project" + mc.projectPaths[2] = "fork-owner/project" + p := newEventsPoller(mc) + + event := RoutableEvent{ + Type: "mr_note", + IID: 5, + NoteBody: "/fs-review check this", + NoteID: 20, + NoteAuthorID: 42, + NoteAuthorLogin: "alice", + IsBot: false, + MRSource: 2, + MRTarget: 1, + MRAuthorID: 42, + MRAuthorLogin: "alice", + SourceBranch: "feature", + TargetBranch: "main", + } + + ne, err := p.toNormalizedEvent(context.Background(), event) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if ne.Entity.Kind != "change_proposal" { + t.Errorf("Entity.Kind = %q, want %q", ne.Entity.Kind, "change_proposal") + } + if ne.Entity.URL != "https://gitlab.com/group/project/-/merge_requests/5" { + t.Errorf("Entity.URL = %q, want %q", ne.Entity.URL, "https://gitlab.com/group/project/-/merge_requests/5") + } + if ne.Source.RawType != "merge_request" { + t.Errorf("Source.RawType = %q, want %q", ne.Source.RawType, "merge_request") + } + if ne.Transition.Kind != "comment_added" { + t.Errorf("Transition.Kind = %q, want %q", ne.Transition.Kind, "comment_added") + } + if ne.Transition.Comment == nil { + t.Fatal("expected Transition.Comment to be set") + } + if ne.Transition.Comment.Command != "/fs-review" { + t.Errorf("Comment.Command = %q, want %q", ne.Transition.Comment.Command, "/fs-review") + } + if ne.Transition.Comment.Instruction != "check this" { + t.Errorf("Comment.Instruction = %q, want %q", ne.Transition.Comment.Instruction, "check this") + } + + // MR note should have change proposal state populated. + if ne.State.ChangeProposal == nil { + t.Fatal("expected State.ChangeProposal to be set for mr_note") + } + if ne.State.ChangeProposal.HeadRepo != "fork-owner/project" { + t.Errorf("ChangeProposal.HeadRepo = %q, want %q", ne.State.ChangeProposal.HeadRepo, "fork-owner/project") + } + if ne.State.ChangeProposal.BaseRepo != "group/project" { + t.Errorf("ChangeProposal.BaseRepo = %q, want %q", ne.State.ChangeProposal.BaseRepo, "group/project") + } + if !ne.State.ChangeProposal.IsFork { + t.Error("expected ChangeProposal.IsFork to be true") + } +} + +func TestToNormalizedEvent_MREventMerge(t *testing.T) { + mc := newMockClient() + mc.memberLevel[10] = 50 // Owner -> "admin" + mc.projectPaths[1] = "group/project" + p := newEventsPoller(mc) + + event := RoutableEvent{ + Type: "mr_event", + IID: 5, + NoteAuthorID: 10, + MergedByLogin: "maintainer-user", + IsBot: false, + MRSource: 1, + MRTarget: 1, + MRAuthorID: 10, + MRAuthorLogin: "maintainer-user", + SourceBranch: "feature", + TargetBranch: "main", + } + + ne, err := p.toNormalizedEvent(context.Background(), event) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if ne.Transition.Kind != "merged" { + t.Errorf("Transition.Kind = %q, want %q", ne.Transition.Kind, "merged") + } + if ne.Actor.ID != "maintainer-user" { + t.Errorf("Actor.ID = %q, want %q", ne.Actor.ID, "maintainer-user") + } + if ne.Actor.Role != "admin" { + t.Errorf("Actor.Role = %q, want %q", ne.Actor.Role, "admin") + } + if ne.State.ChangeProposal == nil { + t.Fatal("expected State.ChangeProposal to be set for mr_event") + } + if ne.State.ChangeProposal.IsFork { + t.Error("expected ChangeProposal.IsFork to be false for same project") + } +} + +func TestToNormalizedEvent_UnresolvableActorError(t *testing.T) { + mc := newMockClient() + p := newEventsPoller(mc) + + // issue_note with NoteAuthorID=0 -> authorID stays 0 -> error. + event := RoutableEvent{ + Type: "issue_note", + IID: 1, + NoteAuthorID: 0, + NoteBody: "orphaned", + } + + _, err := p.toNormalizedEvent(context.Background(), event) + if err == nil { + t.Fatal("expected error for unresolvable actor") + } + if got := err.Error(); got != "unresolvable actor" { + t.Errorf("error = %q, want %q", got, "unresolvable actor") + } +} + +func TestTranslateEventType(t *testing.T) { + tests := []struct { + input string + want string + }{ + {"issue_label", "label_changed"}, + {"issue_note", "comment_added"}, + {"mr_note", "comment_added"}, + {"mr_event", "merged"}, + {"unknown", "unknown"}, + } + for _, tt := range tests { + t.Run(tt.input, func(t *testing.T) { + got := translateEventType(tt.input) + if got != tt.want { + t.Errorf("translateEventType(%q) = %q, want %q", tt.input, got, tt.want) + } + }) + } +} + +func TestResolveLabelAuthor_FindsMatch(t *testing.T) { + mc := newMockClient() + mc.labelEvents[1] = []ResourceLabelEvent{ + { + ID: 1, + Action: "add", + Label: struct { + Name string `json:"name"` + }{Name: "bug"}, + User: UserRef{ID: 10, Username: "alice", Bot: false}, + }, + { + ID: 2, + Action: "add", + Label: struct { + Name string `json:"name"` + }{Name: "ready-to-code"}, + User: UserRef{ID: 20, Username: "bob", Bot: false}, + }, + { + ID: 3, + Action: "add", + Label: struct { + Name string `json:"name"` + }{Name: "ready-to-code"}, + User: UserRef{ID: 30, Username: "carol", Bot: true}, + }, + } + p := newEventsPoller(mc) + + // Should find the most recent "add" for "ready-to-code" (ID 3, user 30). + la, err := p.resolveLabelAuthor(context.Background(), 1, "ready-to-code") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if la.ID != 30 { + t.Errorf("LabelAuthor.ID = %d, want 30 (most recent add)", la.ID) + } + if !la.IsBot { + t.Error("expected LabelAuthor.IsBot to be true") + } +} + +func TestResolveLabelAuthor_NoMatch(t *testing.T) { + mc := newMockClient() + mc.labelEvents[1] = []ResourceLabelEvent{ + { + ID: 1, + Action: "remove", + Label: struct { + Name string `json:"name"` + }{Name: "ready-to-code"}, + User: UserRef{ID: 10, Username: "alice"}, + }, + } + p := newEventsPoller(mc) + + _, err := p.resolveLabelAuthor(context.Background(), 1, "ready-to-code") + if err == nil { + t.Fatal("expected error when no matching add event found") + } +} + +func TestResolveActorRole_AllLevels(t *testing.T) { + tests := []struct { + level int + want string + }{ + {10, "read"}, + {20, "triage"}, + {30, "write"}, + {40, "maintain"}, + {50, "admin"}, + } + for _, tt := range tests { + t.Run(fmt.Sprintf("level_%d", tt.level), func(t *testing.T) { + mc := newMockClient() + mc.memberLevel[1] = tt.level + p := newEventsPoller(mc) + + got := p.resolveActorRole(context.Background(), 1) + if got != tt.want { + t.Errorf("resolveActorRole(level=%d) = %q, want %q", tt.level, got, tt.want) + } + }) + } +} + +func TestResolveActorRole_UnknownLevel(t *testing.T) { + mc := newMockClient() + mc.memberLevel[1] = 99 // unknown level + p := newEventsPoller(mc) + + got := p.resolveActorRole(context.Background(), 1) + if got != "none" { + t.Errorf("resolveActorRole(level=99) = %q, want %q", got, "none") + } +} + +func TestResolveActorRole_MemberNotFound(t *testing.T) { + mc := newMockClient() + // No memberLevel set for userID 1 -> GetMemberAccessLevel returns error. + p := newEventsPoller(mc) + + got := p.resolveActorRole(context.Background(), 1) + if got != "none" { + t.Errorf("resolveActorRole(not found) = %q, want %q", got, "none") + } +} + +func TestExtractCommand_Triage(t *testing.T) { + cmd, instruction := extractCommand("/fs-triage") + if cmd != "/fs-triage" { + t.Errorf("command = %q, want %q", cmd, "/fs-triage") + } + if instruction != "" { + t.Errorf("instruction = %q, want empty", instruction) + } +} + +func TestExtractCommand_WithInstruction(t *testing.T) { + cmd, instruction := extractCommand("/fs-code please implement the fix\nwith tests") + if cmd != "/fs-code" { + t.Errorf("command = %q, want %q", cmd, "/fs-code") + } + want := "please implement the fix\nwith tests" + if instruction != want { + t.Errorf("instruction = %q, want %q", instruction, want) + } +} + +func TestExtractCommand_NoCommand(t *testing.T) { + tests := []struct { + name string + body string + }{ + {"regular comment", "just a regular comment"}, + {"non-fs slash", "/label ~bug"}, + {"empty body", ""}, + {"whitespace only", " \n "}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cmd, instruction := extractCommand(tt.body) + if cmd != "" { + t.Errorf("command = %q, want empty", cmd) + } + if instruction != "" { + t.Errorf("instruction = %q, want empty", instruction) + } + }) + } +} + +func TestEntityURL_Issue(t *testing.T) { + got := entityURL("https://gitlab.com", "group/project", "issue_note", 42) + want := "https://gitlab.com/group/project/-/issues/42" + if got != want { + t.Errorf("entityURL(issue_note) = %q, want %q", got, want) + } + + // Also test issue_label. + got = entityURL("https://gitlab.com", "group/project", "issue_label", 7) + want = "https://gitlab.com/group/project/-/issues/7" + if got != want { + t.Errorf("entityURL(issue_label) = %q, want %q", got, want) + } +} + +func TestEntityURL_MR(t *testing.T) { + got := entityURL("https://gitlab.com", "group/project", "mr_note", 15) + want := "https://gitlab.com/group/project/-/merge_requests/15" + if got != want { + t.Errorf("entityURL(mr_note) = %q, want %q", got, want) + } + + // Also test mr_event. + got = entityURL("https://gitlab.com", "group/project", "mr_event", 3) + want = "https://gitlab.com/group/project/-/merge_requests/3" + if got != want { + t.Errorf("entityURL(mr_event) = %q, want %q", got, want) + } +} + +func TestEntityURL_TrailingSlash(t *testing.T) { + got := entityURL("https://gitlab.com/", "group/project", "issue_note", 1) + want := "https://gitlab.com/group/project/-/issues/1" + if got != want { + t.Errorf("entityURL with trailing slash = %q, want %q", got, want) + } +} + +func TestMapRawType(t *testing.T) { + tests := []struct { + input string + want string + }{ + {"issue_label", "issues"}, + {"issue_note", "issues"}, + {"mr_note", "merge_request"}, + {"mr_event", "merge_request"}, + {"unknown", "unknown"}, + } + for _, tt := range tests { + t.Run(tt.input, func(t *testing.T) { + got := mapRawType(tt.input) + if got != tt.want { + t.Errorf("mapRawType(%q) = %q, want %q", tt.input, got, tt.want) + } + }) + } +} + +func TestMapRawAction(t *testing.T) { + tests := []struct { + input string + want string + }{ + {"issue_label", "labeled"}, + {"issue_note", "commented"}, + {"mr_note", "commented"}, + {"mr_event", "merged"}, + {"unknown", ""}, + } + for _, tt := range tests { + t.Run(tt.input, func(t *testing.T) { + got := mapRawAction(tt.input) + if got != tt.want { + t.Errorf("mapRawAction(%q) = %q, want %q", tt.input, got, tt.want) + } + }) + } +} + +func TestEntityKind(t *testing.T) { + tests := []struct { + input string + want string + }{ + {"issue_label", "work_item"}, + {"issue_note", "work_item"}, + {"mr_note", "change_proposal"}, + {"mr_event", "change_proposal"}, + {"unknown", "work_item"}, + } + for _, tt := range tests { + t.Run(tt.input, func(t *testing.T) { + got := entityKind(tt.input) + if got != tt.want { + t.Errorf("entityKind(%q) = %q, want %q", tt.input, got, tt.want) + } + }) + } +} + +func TestBuildChangeProposalState(t *testing.T) { + mc := newMockClient() + mc.projectPaths[10] = "fork/repo" + mc.projectPaths[20] = "upstream/repo" + p := newEventsPoller(mc) + + event := RoutableEvent{ + Type: "mr_note", + IID: 5, + MRSource: 10, + MRTarget: 20, + SourceBranch: "feature-branch", + TargetBranch: "main", + MRAuthorID: 77, + MRAuthorLogin: "dev-user", + } + + cps, err := p.buildChangeProposalState(context.Background(), event) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if cps.ID != 5 { + t.Errorf("ID = %d, want 5", cps.ID) + } + if cps.HeadRepo != "fork/repo" { + t.Errorf("HeadRepo = %q, want %q", cps.HeadRepo, "fork/repo") + } + if cps.BaseRepo != "upstream/repo" { + t.Errorf("BaseRepo = %q, want %q", cps.BaseRepo, "upstream/repo") + } + if cps.HeadRef != "feature-branch" { + t.Errorf("HeadRef = %q, want %q", cps.HeadRef, "feature-branch") + } + if cps.BaseRef != "main" { + t.Errorf("BaseRef = %q, want %q", cps.BaseRef, "main") + } + if cps.AuthorID != "dev-user" { + t.Errorf("AuthorID = %q, want %q", cps.AuthorID, "dev-user") + } + if !cps.IsFork { + t.Error("expected IsFork to be true for different source/target projects") + } +} + +func TestBuildChangeProposalState_SameProject(t *testing.T) { + mc := newMockClient() + mc.projectPaths[1] = "group/project" + p := newEventsPoller(mc) + + event := RoutableEvent{ + Type: "mr_event", + IID: 3, + MRSource: 1, + MRTarget: 1, + } + + cps, err := p.buildChangeProposalState(context.Background(), event) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if cps.IsFork { + t.Error("expected IsFork to be false for same source/target project") + } +} + +func TestIsEntityAuthor_IssueNote(t *testing.T) { + mc := newMockClient() + mc.issue[1] = &Issue{IID: 1, AuthorID: 42} + p := newEventsPoller(mc) + + event := RoutableEvent{Type: "issue_note", IID: 1} + + if !p.isEntityAuthor(context.Background(), event, 42) { + t.Error("expected isEntityAuthor to return true when actor is issue author") + } + if p.isEntityAuthor(context.Background(), event, 99) { + t.Error("expected isEntityAuthor to return false when actor is not issue author") + } +} + +func TestIsEntityAuthor_MREvent(t *testing.T) { + mc := newMockClient() + p := newEventsPoller(mc) + + event := RoutableEvent{Type: "mr_event", IID: 5, MRAuthorID: 42} + if !p.isEntityAuthor(context.Background(), event, 42) { + t.Error("expected isEntityAuthor to return true when actor is MR author") + } + if p.isEntityAuthor(context.Background(), event, 99) { + t.Error("expected isEntityAuthor to return false when actor is not MR author") + } + + noAuthor := RoutableEvent{Type: "mr_event", IID: 5, MRAuthorID: 0} + if p.isEntityAuthor(context.Background(), noAuthor, 42) { + t.Error("expected isEntityAuthor to return false when MRAuthorID is 0") + } +} + +func TestToNormalizedEvent_BotActor(t *testing.T) { + mc := newMockClient() + mc.memberLevel[200] = 30 + p := newEventsPoller(mc) + + event := RoutableEvent{ + Type: "issue_note", + IID: 1, + NoteBody: "automated comment", + NoteID: 10, + NoteAuthorID: 200, + NoteAuthorLogin: "project_1_bot_abc", + IsBot: true, + Labels: []string{}, + } + + ne, err := p.toNormalizedEvent(context.Background(), event) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if ne.Actor.Kind != "bot" { + t.Errorf("Actor.Kind = %q, want %q for bot actor", ne.Actor.Kind, "bot") + } +} + +func TestTruncate(t *testing.T) { + if got := truncate("hello", 10); got != "hello" { + t.Errorf("short string: got %q, want %q", got, "hello") + } + if got := truncate("hello", 3); got != "hel" { + t.Errorf("ASCII truncate: got %q, want %q", got, "hel") + } + multi := strings.Repeat("\U0001F600", 5) + got := truncate(multi, 3) + if []rune(got)[0] != '\U0001F600' || len([]rune(got)) != 3 { + t.Errorf("multi-byte truncate: got %q (len %d runes), want 3 runes", got, len([]rune(got))) + } +} diff --git a/internal/poll/dispatch.go b/internal/poll/dispatch.go new file mode 100644 index 000000000..53c808535 --- /dev/null +++ b/internal/poll/dispatch.go @@ -0,0 +1,155 @@ +package poll + +import ( + "bytes" + "context" + "encoding/base64" + "encoding/json" + "fmt" + "os" + "regexp" + "strings" + "time" +) + +var validStage = regexp.MustCompile(`^[a-z][a-z0-9-]*$`) +var validYAMLField = regexp.MustCompile(`^[a-z][a-z0-9_-]*$`) + +// appendDispatch appends a dispatch record to the poller's in-memory list. +func (p *Poller) appendDispatch(d Dispatch) error { + p.dispatches = append(p.dispatches, d) + return nil +} + +// writeDispatches marshals the accumulated dispatches as a JSON array +// and writes them to the given file path. +func (p *Poller) writeDispatches(path string) error { + dispatches := p.dispatches + if len(dispatches) == 0 { + return os.WriteFile(path, []byte("[]\n"), 0o644) + } + data, err := json.MarshalIndent(dispatches, "", " ") + if err != nil { + return fmt.Errorf("marshal dispatches: %w", err) + } + data = append(data, '\n') + return os.WriteFile(path, data, 0o644) +} + +// resourceKey returns a stable entity-based key for concurrency control. +func resourceKey(event RoutableEvent) string { + prefix := "issue" + if strings.HasPrefix(event.Type, "mr_") { + prefix = "mr" + } + return fmt.Sprintf("%s-%d", prefix, event.IID) +} + +// dispatch builds an event payload, base64-encodes it, and appends +// a Dispatch record for child pipeline generation. +func (p *Poller) dispatch(ctx context.Context, owner, repo, stage string, event RoutableEvent) error { + _ = ctx // reserved for future use + _ = owner // included in signature for routing context + _ = repo + payload, err := buildEventPayload(event) + if err != nil { + return fmt.Errorf("build event payload: %w", err) + } + encoded := base64.StdEncoding.EncodeToString(payload) + return p.appendDispatch(Dispatch{ + Stage: stage, + EventType: event.Type, + EventPayloadB64: encoded, + ResourceKey: resourceKey(event), + }) +} + +// buildEventPayload creates a JSON payload from a RoutableEvent, +// including only non-zero/non-empty optional fields. +func buildEventPayload(event RoutableEvent) ([]byte, error) { + m := map[string]interface{}{ + "type": event.Type, + "iid": event.IID, + "updated_at": event.UpdatedAt.Format(time.RFC3339), + } + if event.NoteBody != "" { + m["note_body"] = truncate(event.NoteBody, 4096) + } + if event.NoteID != 0 { + m["note_id"] = event.NoteID + } + if event.NoteAuthorID != 0 { + m["note_author_id"] = event.NoteAuthorID + } + if event.Labels != nil { + m["labels"] = event.Labels + } + if event.MRSource != 0 { + m["mr_source_project_id"] = event.MRSource + } + if event.MRTarget != 0 { + m["mr_target_project_id"] = event.MRTarget + } + if event.SourceBranch != "" { + m["source_branch"] = event.SourceBranch + } + if event.TargetBranch != "" { + m["target_branch"] = event.TargetBranch + } + if event.MRAuthorID != 0 { + m["mr_author_id"] = event.MRAuthorID + } + if event.IsBot { + m["is_bot"] = true + } + return json.Marshal(m) +} + +// generateChildPipelineYAML produces GitLab CI YAML that triggers +// one child pipeline job per dispatch record. Returns an error if +// any stage name contains invalid characters. +func generateChildPipelineYAML(dispatches []Dispatch) (string, error) { + var buf bytes.Buffer + for i, d := range dispatches { + if !validStage.MatchString(d.Stage) { + return "", fmt.Errorf("invalid stage name %q: must match %s", d.Stage, validStage.String()) + } + if !validYAMLField.MatchString(d.EventType) { + return "", fmt.Errorf("invalid event type %q: must match %s", d.EventType, validYAMLField.String()) + } + if !validYAMLField.MatchString(d.ResourceKey) { + return "", fmt.Errorf("invalid resource key %q: must match %s", d.ResourceKey, validYAMLField.String()) + } + fmt.Fprintf(&buf, "agent-%d:\n", i) + fmt.Fprintf(&buf, " trigger:\n") + fmt.Fprintf(&buf, " include: .gitlab/ci/fullsend-%s.yml\n", d.Stage) + fmt.Fprintf(&buf, " strategy: depend\n") + fmt.Fprintf(&buf, " variables:\n") + fmt.Fprintf(&buf, " STAGE: %q\n", d.Stage) + fmt.Fprintf(&buf, " EVENT_TYPE: %q\n", d.EventType) + fmt.Fprintf(&buf, " EVENT_PAYLOAD_B64: %q\n", d.EventPayloadB64) + fmt.Fprintf(&buf, " RESOURCE_KEY: %q\n", d.ResourceKey) + fmt.Fprintf(&buf, " rules:\n") + fmt.Fprintf(&buf, " - when: always\n") + } + return buf.String(), nil +} + +// GenerateChildPipelineFromFile reads a dispatches JSON file and +// writes the corresponding child pipeline YAML to outputPath. +// It is used by the CLI "fullsend poll generate-child-pipeline" subcommand. +func GenerateChildPipelineFromFile(dispatchesPath, outputPath string) error { + data, err := os.ReadFile(dispatchesPath) + if err != nil { + return fmt.Errorf("read dispatches file: %w", err) + } + var dispatches []Dispatch + if err := json.Unmarshal(data, &dispatches); err != nil { + return fmt.Errorf("unmarshal dispatches: %w", err) + } + yaml, err := generateChildPipelineYAML(dispatches) + if err != nil { + return err + } + return os.WriteFile(outputPath, []byte(yaml), 0o644) +} diff --git a/internal/poll/dispatch_test.go b/internal/poll/dispatch_test.go new file mode 100644 index 000000000..e9615d2d2 --- /dev/null +++ b/internal/poll/dispatch_test.go @@ -0,0 +1,494 @@ +package poll + +import ( + "context" + "encoding/base64" + "encoding/json" + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +// --- appendDispatch tests --- + +func TestAppendDispatch_AddsToList(t *testing.T) { + mc := newMockClient() + p := newTestPoller(mc, Options{}) + + d := Dispatch{ + Stage: "triage", + EventType: "issue_comment", + EventPayloadB64: "dGVzdA==", + ResourceKey: "issue_comment-1", + } + if err := p.appendDispatch(d); err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(p.dispatches) != 1 { + t.Fatalf("expected 1 dispatch, got %d", len(p.dispatches)) + } + if p.dispatches[0].Stage != "triage" { + t.Errorf("got stage %q, want triage", p.dispatches[0].Stage) + } +} + +func TestAppendDispatch_AccumulatesMultiple(t *testing.T) { + mc := newMockClient() + p := newTestPoller(mc, Options{}) + + for i := 0; i < 3; i++ { + _ = p.appendDispatch(Dispatch{Stage: "triage", ResourceKey: "k"}) + } + if len(p.dispatches) != 3 { + t.Errorf("expected 3 dispatches, got %d", len(p.dispatches)) + } +} + +// --- writeDispatches tests --- + +func TestWriteDispatches_EmptyWritesEmptyArray(t *testing.T) { + mc := newMockClient() + p := newTestPoller(mc, Options{}) + + path := filepath.Join(t.TempDir(), "dispatches.json") + if err := p.writeDispatches(path); err != nil { + t.Fatalf("unexpected error: %v", err) + } + + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read file: %v", err) + } + if string(data) != "[]\n" { + t.Errorf("got %q, want %q", string(data), "[]\n") + } +} + +func TestWriteDispatches_WritesValidJSONArray(t *testing.T) { + mc := newMockClient() + p := newTestPoller(mc, Options{}) + + _ = p.appendDispatch(Dispatch{ + Stage: "triage", + EventType: "issue_comment", + EventPayloadB64: "cGF5bG9hZA==", + ResourceKey: "issue_comment-5", + }) + _ = p.appendDispatch(Dispatch{ + Stage: "code", + EventType: "issue_label", + EventPayloadB64: "bGFiZWw=", + ResourceKey: "issue_label-10", + }) + + path := filepath.Join(t.TempDir(), "dispatches.json") + if err := p.writeDispatches(path); err != nil { + t.Fatalf("unexpected error: %v", err) + } + + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read file: %v", err) + } + + var parsed []Dispatch + if err := json.Unmarshal(data, &parsed); err != nil { + t.Fatalf("invalid JSON: %v", err) + } + if len(parsed) != 2 { + t.Fatalf("expected 2 dispatches, got %d", len(parsed)) + } + if parsed[0].Stage != "triage" { + t.Errorf("dispatch 0: got stage %q, want triage", parsed[0].Stage) + } + if parsed[1].Stage != "code" { + t.Errorf("dispatch 1: got stage %q, want code", parsed[1].Stage) + } +} + +func TestWriteDispatches_TrailingNewline(t *testing.T) { + mc := newMockClient() + p := newTestPoller(mc, Options{}) + + _ = p.appendDispatch(Dispatch{Stage: "triage", ResourceKey: "k-1"}) + + path := filepath.Join(t.TempDir(), "dispatches.json") + if err := p.writeDispatches(path); err != nil { + t.Fatalf("unexpected error: %v", err) + } + + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read file: %v", err) + } + if !strings.HasSuffix(string(data), "\n") { + t.Error("expected trailing newline") + } +} + +// --- dispatch method tests --- + +func TestDispatch_BuildsPayloadAndAppends(t *testing.T) { + mc := newMockClient() + p := newTestPoller(mc, Options{}) + + ts := time.Date(2025, 6, 15, 12, 0, 0, 0, time.UTC) + event := RoutableEvent{ + Type: "issue_comment", + IID: 42, + UpdatedAt: ts, + NoteBody: "/fs-triage", + NoteID: 100, + } + + err := p.dispatch(context.Background(), "owner", "repo", "triage", event) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if len(p.dispatches) != 1 { + t.Fatalf("expected 1 dispatch, got %d", len(p.dispatches)) + } + d := p.dispatches[0] + if d.Stage != "triage" { + t.Errorf("stage: got %q, want triage", d.Stage) + } + if d.EventType != "issue_comment" { + t.Errorf("event_type: got %q, want issue_comment", d.EventType) + } + if d.ResourceKey != "issue-42" { + t.Errorf("resource_key: got %q, want issue-42", d.ResourceKey) + } + + // Verify the payload is valid base64-encoded JSON. + decoded, err := base64.StdEncoding.DecodeString(d.EventPayloadB64) + if err != nil { + t.Fatalf("decode payload: %v", err) + } + var payload map[string]interface{} + if err := json.Unmarshal(decoded, &payload); err != nil { + t.Fatalf("unmarshal payload: %v", err) + } + if payload["type"] != "issue_comment" { + t.Errorf("payload type: got %v, want issue_comment", payload["type"]) + } + if int(payload["iid"].(float64)) != 42 { + t.Errorf("payload iid: got %v, want 42", payload["iid"]) + } +} + +// --- buildEventPayload tests --- + +func TestBuildEventPayload_IncludesAllFields(t *testing.T) { + ts := time.Date(2025, 6, 15, 12, 0, 0, 0, time.UTC) + event := RoutableEvent{ + Type: "issue_comment", + IID: 7, + UpdatedAt: ts, + NoteBody: "/fs-triage", + NoteID: 200, + NoteAuthorID: 55, + Labels: []string{"ready-to-code"}, + MRSource: 100, + MRTarget: 200, + } + + data, err := buildEventPayload(event) + if err != nil { + t.Fatalf("buildEventPayload: %v", err) + } + var m map[string]interface{} + if err := json.Unmarshal(data, &m); err != nil { + t.Fatalf("unmarshal: %v", err) + } + + checks := map[string]interface{}{ + "type": "issue_comment", + "iid": float64(7), + "note_body": "/fs-triage", + "note_id": float64(200), + "note_author_id": float64(55), + "mr_source_project_id": float64(100), + "mr_target_project_id": float64(200), + } + for key, want := range checks { + got, ok := m[key] + if !ok { + t.Errorf("missing key %q", key) + continue + } + if got != want { + t.Errorf("%s: got %v, want %v", key, got, want) + } + } + + // Check updated_at. + if m["updated_at"] != ts.Format(time.RFC3339) { + t.Errorf("updated_at: got %v, want %v", m["updated_at"], ts.Format(time.RFC3339)) + } + + // Check labels array. + labels, ok := m["labels"].([]interface{}) + if !ok { + t.Fatal("labels should be an array") + } + if len(labels) != 1 || labels[0] != "ready-to-code" { + t.Errorf("labels: got %v, want [ready-to-code]", labels) + } +} + +func TestBuildEventPayload_OmitsZeroOptionalFields(t *testing.T) { + ts := time.Date(2025, 6, 15, 12, 0, 0, 0, time.UTC) + event := RoutableEvent{ + Type: "issue_label", + IID: 10, + UpdatedAt: ts, + // All optional fields are zero values. + } + + data, err := buildEventPayload(event) + if err != nil { + t.Fatalf("buildEventPayload: %v", err) + } + var m map[string]interface{} + if err := json.Unmarshal(data, &m); err != nil { + t.Fatalf("unmarshal: %v", err) + } + + // Must have required fields. + for _, required := range []string{"type", "iid", "updated_at"} { + if _, ok := m[required]; !ok { + t.Errorf("missing required key %q", required) + } + } + + // Must NOT have optional fields when zero. + optionals := []string{"note_body", "note_id", "note_author_id", "labels", "mr_source_project_id", "mr_target_project_id"} + for _, key := range optionals { + if _, ok := m[key]; ok { + t.Errorf("expected optional key %q to be omitted, but it was present", key) + } + } +} + +// --- generateChildPipelineYAML tests --- + +func TestGenerateChildPipelineYAML_SingleDispatch(t *testing.T) { + dispatches := []Dispatch{ + { + Stage: "triage", + EventType: "issue_comment", + EventPayloadB64: "cGF5bG9hZA==", + ResourceKey: "issue_comment-1", + }, + } + + yaml, err := generateChildPipelineYAML(dispatches) + if err != nil { + t.Fatalf("generateChildPipelineYAML: %v", err) + } + + expected := []string{ + "agent-0:", + "trigger:", + ".gitlab/ci/fullsend-triage.yml", + "strategy: depend", + `STAGE: "triage"`, + `EVENT_TYPE: "issue_comment"`, + `EVENT_PAYLOAD_B64: "cGF5bG9hZA=="`, + `RESOURCE_KEY: "issue_comment-1"`, + "rules:", + "- when: always", + } + for _, substr := range expected { + if !strings.Contains(yaml, substr) { + t.Errorf("missing %q in output:\n%s", substr, yaml) + } + } +} + +func TestGenerateChildPipelineYAML_MultipleDispatches(t *testing.T) { + dispatches := []Dispatch{ + {Stage: "triage", EventType: "issue_comment", EventPayloadB64: "a", ResourceKey: "k1"}, + {Stage: "code", EventType: "issue_label", EventPayloadB64: "b", ResourceKey: "k2"}, + {Stage: "review", EventType: "mr_comment", EventPayloadB64: "c", ResourceKey: "k3"}, + } + + yaml, err := generateChildPipelineYAML(dispatches) + if err != nil { + t.Fatalf("generateChildPipelineYAML: %v", err) + } + + for i := 0; i < 3; i++ { + marker := "agent-" + strings.Repeat("", 0) + string(rune('0'+i)) + ":" + if !strings.Contains(yaml, marker) { + t.Errorf("missing %q in output:\n%s", marker, yaml) + } + } + + if !strings.Contains(yaml, "fullsend-triage.yml") { + t.Error("missing triage stage include") + } + if !strings.Contains(yaml, "fullsend-code.yml") { + t.Error("missing code stage include") + } + if !strings.Contains(yaml, "fullsend-review.yml") { + t.Error("missing review stage include") + } +} + +func TestGenerateChildPipelineYAML_EmptyDispatches(t *testing.T) { + yaml, err := generateChildPipelineYAML(nil) + if err != nil { + t.Fatalf("generateChildPipelineYAML(nil): %v", err) + } + if yaml != "" { + t.Errorf("expected empty string, got %q", yaml) + } + + yaml, err = generateChildPipelineYAML([]Dispatch{}) + if err != nil { + t.Fatalf("generateChildPipelineYAML([]): %v", err) + } + if yaml != "" { + t.Errorf("expected empty string for empty slice, got %q", yaml) + } +} + +func TestGenerateChildPipelineYAML_InvalidStageReturnsError(t *testing.T) { + dispatches := []Dispatch{ + {Stage: "INVALID STAGE!", EventType: "issue_comment", EventPayloadB64: "a", ResourceKey: "k"}, + } + _, err := generateChildPipelineYAML(dispatches) + if err == nil { + t.Fatal("expected error for invalid stage name") + } + if !strings.Contains(err.Error(), "invalid stage name") { + t.Errorf("unexpected error: %v", err) + } +} + +func TestResourceKey_EntityBased(t *testing.T) { + tests := []struct { + event RoutableEvent + want string + }{ + {RoutableEvent{Type: "issue_label", IID: 42}, "issue-42"}, + {RoutableEvent{Type: "issue_note", IID: 7}, "issue-7"}, + {RoutableEvent{Type: "mr_event", IID: 10}, "mr-10"}, + {RoutableEvent{Type: "mr_note", IID: 3}, "mr-3"}, + } + for _, tt := range tests { + got := resourceKey(tt.event) + if got != tt.want { + t.Errorf("resourceKey(%s, IID=%d) = %q, want %q", tt.event.Type, tt.event.IID, got, tt.want) + } + } +} + +// --- GenerateChildPipelineFromFile tests --- + +func TestGenerateChildPipelineFromFile_ReadsAndWrites(t *testing.T) { + tmpDir := t.TempDir() + inputPath := filepath.Join(tmpDir, "dispatches.json") + outputPath := filepath.Join(tmpDir, "child-pipeline.yml") + + dispatches := []Dispatch{ + { + Stage: "triage", + EventType: "issue_comment", + EventPayloadB64: "cGF5bG9hZA==", + ResourceKey: "issue_comment-5", + }, + { + Stage: "code", + EventType: "issue_label", + EventPayloadB64: "bGFiZWw=", + ResourceKey: "issue_label-10", + }, + } + data, err := json.Marshal(dispatches) + if err != nil { + t.Fatalf("marshal: %v", err) + } + if err := os.WriteFile(inputPath, data, 0o644); err != nil { + t.Fatalf("write input: %v", err) + } + + if err := GenerateChildPipelineFromFile(inputPath, outputPath); err != nil { + t.Fatalf("unexpected error: %v", err) + } + + output, err := os.ReadFile(outputPath) + if err != nil { + t.Fatalf("read output: %v", err) + } + yamlStr := string(output) + + if !strings.Contains(yamlStr, "agent-0:") { + t.Error("missing agent-0 in output") + } + if !strings.Contains(yamlStr, "agent-1:") { + t.Error("missing agent-1 in output") + } + if !strings.Contains(yamlStr, "fullsend-triage.yml") { + t.Error("missing triage include") + } + if !strings.Contains(yamlStr, "fullsend-code.yml") { + t.Error("missing code include") + } +} + +func TestGenerateChildPipelineFromFile_MissingInputFile(t *testing.T) { + tmpDir := t.TempDir() + err := GenerateChildPipelineFromFile( + filepath.Join(tmpDir, "nonexistent.json"), + filepath.Join(tmpDir, "output.yml"), + ) + if err == nil { + t.Fatal("expected error for missing input file") + } + if !strings.Contains(err.Error(), "read dispatches file") { + t.Errorf("unexpected error message: %v", err) + } +} + +func TestGenerateChildPipelineFromFile_InvalidJSON(t *testing.T) { + tmpDir := t.TempDir() + inputPath := filepath.Join(tmpDir, "bad.json") + if err := os.WriteFile(inputPath, []byte("{not json"), 0o644); err != nil { + t.Fatalf("write: %v", err) + } + + err := GenerateChildPipelineFromFile(inputPath, filepath.Join(tmpDir, "output.yml")) + if err == nil { + t.Fatal("expected error for invalid JSON") + } + if !strings.Contains(err.Error(), "unmarshal dispatches") { + t.Errorf("unexpected error message: %v", err) + } +} + +func TestGenerateChildPipelineFromFile_EmptyDispatches(t *testing.T) { + tmpDir := t.TempDir() + inputPath := filepath.Join(tmpDir, "empty.json") + outputPath := filepath.Join(tmpDir, "output.yml") + + if err := os.WriteFile(inputPath, []byte("[]"), 0o644); err != nil { + t.Fatalf("write: %v", err) + } + + if err := GenerateChildPipelineFromFile(inputPath, outputPath); err != nil { + t.Fatalf("unexpected error: %v", err) + } + + output, err := os.ReadFile(outputPath) + if err != nil { + t.Fatalf("read output: %v", err) + } + if string(output) != "" { + t.Errorf("expected empty output for empty dispatches, got %q", string(output)) + } +} diff --git a/internal/poll/events.go b/internal/poll/events.go new file mode 100644 index 000000000..a3b296c5e --- /dev/null +++ b/internal/poll/events.go @@ -0,0 +1,229 @@ +package poll + +import ( + "context" + "log" + "strings" + "time" +) + +var routableLabels = map[string]bool{ + "ready-to-code": true, + "ready-for-review": true, +} + +func filterRoutableLabels(labels []string) []string { + var out []string + for _, l := range labels { + if routableLabels[l] { + out = append(out, l) + } + } + return out +} + +// discoverAllEvents finds all routable events since the given time. +// Returns events, updated label state (for persistence after dispatch), +// minSkippedAt (earliest UpdatedAt among skipped items), and error. +func (p *Poller) discoverAllEvents(ctx context.Context, owner, repo string, since time.Time) ([]RoutableEvent, LabelState, time.Time, error) { + var events []RoutableEvent + + issues, err := p.client.ListIssuesUpdatedSince(ctx, owner, repo, since) + if err != nil { + return nil, nil, time.Time{}, err + } + + newLabels, updatedLabelState, previousLabels, err := p.detectNewLabels(ctx, owner, repo, issues) + if err != nil { + return nil, nil, time.Time{}, err + } + + var minSkippedAt time.Time + for _, issue := range issues { + notes, err := p.client.ListIssueNotes(ctx, owner, repo, issue.IID) + if err != nil { + log.Printf("list notes for issue %d: %v (skipping issue entirely)", issue.IID, err) + if prev, ok := previousLabels[issue.IID]; ok { + updatedLabelState[issue.IID] = prev + } else { + delete(updatedLabelState, issue.IID) + } + if minSkippedAt.IsZero() || issue.UpdatedAt.Before(minSkippedAt) { + minSkippedAt = issue.UpdatedAt + } + continue + } + + if added, ok := newLabels[issue.IID]; ok { + for _, label := range added { + events = append(events, RoutableEvent{ + Type: "issue_label", + IID: issue.IID, + UpdatedAt: issue.UpdatedAt, + Labels: issue.Labels, + ChangedLabel: label, + }) + } + } + + for _, note := range notes { + if note.CreatedAt.Before(since) { + continue + } + events = append(events, RoutableEvent{ + Type: "issue_note", + IID: issue.IID, + UpdatedAt: note.CreatedAt, + NoteBody: note.Body, + NoteID: note.ID, + NoteAuthorID: note.Author.ID, + NoteAuthorLogin: note.Author.Username, + IsBot: note.Author.Bot, + Labels: issue.Labels, + }) + } + } + + mrs, err := p.client.ListMergeRequestsUpdatedSince(ctx, owner, repo, since) + if err != nil { + log.Printf("list merge requests: %v (continuing with issue events only)", err) + if minSkippedAt.IsZero() || since.Before(minSkippedAt) { + minSkippedAt = since + } + return events, updatedLabelState, minSkippedAt, nil + } + + for _, mr := range mrs { + if !mr.MergedAt.IsZero() && mr.MergedAt.After(since) { + events = append(events, RoutableEvent{ + Type: "mr_event", + IID: mr.IID, + UpdatedAt: mr.MergedAt, + NoteAuthorID: mr.MergedByID, + NoteAuthorLogin: mr.MergedBy.Username, + IsBot: mr.MergedBy.Bot, + MRSource: mr.SourceProjectID, + MRTarget: mr.TargetProjectID, + Labels: mr.Labels, + SourceBranch: mr.SourceBranch, + TargetBranch: mr.TargetBranch, + MRAuthorID: mr.AuthorID, + MRAuthorLogin: mr.Author.Username, + MergedByLogin: mr.MergedBy.Username, + }) + } + + notes, err := p.client.ListMergeRequestNotes(ctx, owner, repo, mr.IID) + if err != nil { + log.Printf("list notes for MR %d: %v (skipping MR entirely)", mr.IID, err) + if minSkippedAt.IsZero() || mr.UpdatedAt.Before(minSkippedAt) { + minSkippedAt = mr.UpdatedAt + } + continue + } + for _, note := range notes { + if note.CreatedAt.Before(since) { + continue + } + events = append(events, RoutableEvent{ + Type: "mr_note", + IID: mr.IID, + UpdatedAt: note.CreatedAt, + NoteBody: note.Body, + NoteID: note.ID, + NoteAuthorID: note.Author.ID, + NoteAuthorLogin: note.Author.Username, + IsBot: note.Author.Bot, + MRSource: mr.SourceProjectID, + MRTarget: mr.TargetProjectID, + Labels: mr.Labels, + SourceBranch: mr.SourceBranch, + TargetBranch: mr.TargetBranch, + MRAuthorID: mr.AuthorID, + MRAuthorLogin: mr.Author.Username, + }) + } + } + + return events, updatedLabelState, minSkippedAt, nil +} + +// discoverSlashCommands finds notes containing /fs-* commands using the +// lightweight Events API (fast-poll mode). +func (p *Poller) discoverSlashCommands(ctx context.Context, owner, repo string, since time.Time) ([]RoutableEvent, error) { + projectEvents, err := p.client.ListProjectEvents(ctx, owner, repo, "Note", since) + if err != nil { + return nil, err + } + + var events []RoutableEvent + for _, evt := range projectEvents { + if evt.CreatedAt.Before(since) { + continue + } + if !strings.HasPrefix(strings.TrimSpace(evt.Note.Body), "/fs-") { + continue + } + var eventType string + switch evt.Note.NoteableType { + case "Issue": + eventType = "issue_note" + case "MergeRequest": + eventType = "mr_note" + default: + continue + } + events = append(events, RoutableEvent{ + Type: eventType, + IID: evt.Note.NoteableIID, + UpdatedAt: evt.CreatedAt, + NoteBody: evt.Note.Body, + NoteID: evt.Note.ID, + NoteAuthorID: evt.Author.ID, + NoteAuthorLogin: evt.Author.Username, + IsBot: isProjectAccessTokenBot(evt.Author.Username), + Labels: []string{}, + }) + } + return events, nil +} + +// isProjectAccessTokenBot detects GitLab project access token bot users +// by username pattern. +func isProjectAccessTokenBot(username string) bool { + return strings.HasPrefix(username, "project_") && strings.Contains(username, "_bot_") +} + +// filterBotEvents removes bot-authored events, except for the enrolled +// bot's changes-requested markers which trigger the fix stage. +func (p *Poller) filterBotEvents(events []RoutableEvent) []RoutableEvent { + var filtered []RoutableEvent + for _, event := range events { + if !event.IsBot { + filtered = append(filtered, event) + continue + } + if event.Type == "mr_note" && + strings.Contains(event.NoteBody, "") && + event.NoteAuthorID == p.botUserID { + filtered = append(filtered, event) + continue + } + } + return filtered +} + +// deduplicate removes duplicate events based on their Key(). +func (p *Poller) deduplicate(events []RoutableEvent) []RoutableEvent { + seen := make(map[string]bool) + var unique []RoutableEvent + for _, event := range events { + key := event.Key() + if seen[key] { + continue + } + seen[key] = true + unique = append(unique, event) + } + return unique +} diff --git a/internal/poll/events_test.go b/internal/poll/events_test.go new file mode 100644 index 000000000..c0f67c870 --- /dev/null +++ b/internal/poll/events_test.go @@ -0,0 +1,542 @@ +package poll + +import ( + "context" + "fmt" + "testing" + "time" +) + +// newEventsPoller creates a Poller with test defaults suitable for events +// and convert tests (sets projectPath, gitlabURL, and botUserID). +func newEventsPoller(client GitLabClient) *Poller { + return New(client, nil, "group/project", Options{ + BotUserID: 100, + GitLabURL: "https://gitlab.com", + }) +} + +func TestDiscoverAllEvents_IssueNotes(t *testing.T) { + now := time.Now().Truncate(time.Second) + since := now.Add(-time.Minute) + mc := newMockClient() + mc.issues = []Issue{ + {IID: 1, UpdatedAt: now, Labels: []string{"bug"}}, + } + mc.notes[1] = []Note{ + {ID: 10, Body: "hello world", Author: UserRef{ID: 42, Username: "alice"}, CreatedAt: now}, + } + + p := newEventsPoller(mc) + events, _, minSkipped, err := p.discoverAllEvents(context.Background(), "group", "project", since) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !minSkipped.IsZero() { + t.Errorf("expected zero minSkippedAt, got %v", minSkipped) + } + + var noteEvents []RoutableEvent + for _, e := range events { + if e.Type == "issue_note" { + noteEvents = append(noteEvents, e) + } + } + if len(noteEvents) != 1 { + t.Fatalf("expected 1 issue_note event, got %d", len(noteEvents)) + } + got := noteEvents[0] + if got.IID != 1 { + t.Errorf("IID = %d, want 1", got.IID) + } + if got.NoteID != 10 { + t.Errorf("NoteID = %d, want 10", got.NoteID) + } + if got.NoteBody != "hello world" { + t.Errorf("NoteBody = %q, want %q", got.NoteBody, "hello world") + } + if got.NoteAuthorID != 42 { + t.Errorf("NoteAuthorID = %d, want 42", got.NoteAuthorID) + } +} + +func TestDiscoverAllEvents_LabelEvents(t *testing.T) { + now := time.Now().Truncate(time.Second) + since := now.Add(-time.Minute) + mc := newMockClient() + mc.issues = []Issue{ + {IID: 1, UpdatedAt: now, Labels: []string{"ready-to-code"}}, + } + // No previous label state -> ErrNotFound -> empty state -> "ready-to-code" is new. + mc.notes[1] = []Note{} // no notes + + p := newEventsPoller(mc) + events, labelState, _, err := p.discoverAllEvents(context.Background(), "group", "project", since) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + var labelEvents []RoutableEvent + for _, e := range events { + if e.Type == "issue_label" { + labelEvents = append(labelEvents, e) + } + } + if len(labelEvents) != 1 { + t.Fatalf("expected 1 issue_label event, got %d", len(labelEvents)) + } + got := labelEvents[0] + if got.IID != 1 { + t.Errorf("IID = %d, want 1", got.IID) + } + if len(got.Labels) != 1 || got.Labels[0] != "ready-to-code" { + t.Errorf("Labels = %v, want [ready-to-code]", got.Labels) + } + + // Label state should track "ready-to-code" for IID 1. + if ls, ok := labelState[1]; !ok { + t.Error("labelState missing IID 1") + } else if len(ls) != 1 || ls[0] != "ready-to-code" { + t.Errorf("labelState[1] = %v, want [ready-to-code]", ls) + } +} + +func TestDiscoverAllEvents_MRMergeEvents(t *testing.T) { + now := time.Now().Truncate(time.Second) + since := now.Add(-time.Minute) + mc := newMockClient() + mc.mrs = []MergeRequest{ + { + IID: 5, + MergedAt: now, + MergedByID: 10, + MergedBy: UserRef{ID: 10, Username: "bob", Bot: false}, + SourceProjectID: 1, + TargetProjectID: 1, + UpdatedAt: now, + }, + } + mc.mrNotes[5] = []Note{} // no MR notes + + p := newEventsPoller(mc) + events, _, _, err := p.discoverAllEvents(context.Background(), "group", "project", since) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + var mrEvents []RoutableEvent + for _, e := range events { + if e.Type == "mr_event" { + mrEvents = append(mrEvents, e) + } + } + if len(mrEvents) != 1 { + t.Fatalf("expected 1 mr_event, got %d", len(mrEvents)) + } + got := mrEvents[0] + if got.IID != 5 { + t.Errorf("IID = %d, want 5", got.IID) + } + if got.NoteAuthorID != 10 { + t.Errorf("NoteAuthorID (MergedByID) = %d, want 10", got.NoteAuthorID) + } + if got.MRSource != 1 || got.MRTarget != 1 { + t.Errorf("MRSource=%d, MRTarget=%d, want 1, 1", got.MRSource, got.MRTarget) + } +} + +func TestDiscoverAllEvents_MRNotes(t *testing.T) { + now := time.Now().Truncate(time.Second) + since := now.Add(-time.Minute) + mc := newMockClient() + mc.mrs = []MergeRequest{ + { + IID: 5, + SourceProjectID: 1, + TargetProjectID: 1, + UpdatedAt: now, + }, + } + mc.mrNotes[5] = []Note{ + {ID: 20, Body: "looks good", Author: UserRef{ID: 42, Username: "alice"}, CreatedAt: now}, + } + + p := newEventsPoller(mc) + events, _, _, err := p.discoverAllEvents(context.Background(), "group", "project", since) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + var mrNotes []RoutableEvent + for _, e := range events { + if e.Type == "mr_note" { + mrNotes = append(mrNotes, e) + } + } + if len(mrNotes) != 1 { + t.Fatalf("expected 1 mr_note event, got %d", len(mrNotes)) + } + got := mrNotes[0] + if got.IID != 5 { + t.Errorf("IID = %d, want 5", got.IID) + } + if got.NoteID != 20 { + t.Errorf("NoteID = %d, want 20", got.NoteID) + } + if got.NoteBody != "looks good" { + t.Errorf("NoteBody = %q, want %q", got.NoteBody, "looks good") + } + if got.MRSource != 1 || got.MRTarget != 1 { + t.Errorf("MRSource=%d, MRTarget=%d, want 1, 1", got.MRSource, got.MRTarget) + } +} + +func TestDiscoverAllEvents_SkipsOldNotes(t *testing.T) { + now := time.Now().Truncate(time.Second) + since := now.Add(-time.Minute) + mc := newMockClient() + mc.issues = []Issue{ + {IID: 1, UpdatedAt: now, Labels: []string{"bug"}}, + } + mc.notes[1] = []Note{ + {ID: 10, Body: "old comment", Author: UserRef{ID: 42}, CreatedAt: since.Add(-time.Hour)}, + } + + p := newEventsPoller(mc) + events, _, _, err := p.discoverAllEvents(context.Background(), "group", "project", since) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + for _, e := range events { + if e.Type == "issue_note" { + t.Errorf("expected no issue_note events for old notes, got event with NoteID=%d", e.NoteID) + } + } +} + +func TestDiscoverAllEvents_NoteFetchFailure(t *testing.T) { + now := time.Now().Truncate(time.Second) + since := now.Add(-time.Minute) + mc := newMockClient() + + // Set up existing label state so we can verify restoration. + mc.variables["FULLSEND_LABEL_STATE"] = `{"1":["ready-to-code"]}` + + mc.issues = []Issue{ + {IID: 1, UpdatedAt: now, Labels: []string{"ready-to-code", "ready-for-review"}}, + } + // ListIssueNotes fails for IID 1. + mc.noteErr[1] = fmt.Errorf("API timeout") + + p := newEventsPoller(mc) + events, labelState, minSkipped, err := p.discoverAllEvents(context.Background(), "group", "project", since) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + // No events should be returned for the failed issue. + for _, e := range events { + if e.IID == 1 { + t.Errorf("expected no events for IID 1 after note failure, got %s", e.Type) + } + } + + // minSkippedAt should be set to the issue's UpdatedAt. + if minSkipped.IsZero() { + t.Fatal("expected minSkippedAt to be set") + } + if !minSkipped.Equal(now) { + t.Errorf("minSkippedAt = %v, want %v", minSkipped, now) + } + + // Label state should be restored to previous state ["ready-to-code"], + // not the updated state ["ready-to-code", "ready-for-review"]. + ls, ok := labelState[1] + if !ok { + t.Fatal("expected labelState to contain IID 1") + } + if len(ls) != 1 || ls[0] != "ready-to-code" { + t.Errorf("labelState[1] = %v, want [ready-to-code] (restored)", ls) + } +} + +func TestDiscoverAllEvents_MRListFailure(t *testing.T) { + now := time.Now().Truncate(time.Second) + since := now.Add(-time.Minute) + mc := newMockClient() + mc.issues = []Issue{ + {IID: 1, UpdatedAt: now, Labels: []string{"bug"}}, + } + mc.notes[1] = []Note{ + {ID: 10, Body: "a comment", Author: UserRef{ID: 42}, CreatedAt: now}, + } + mc.mrsErr = fmt.Errorf("MR API down") + + p := newEventsPoller(mc) + events, _, minSkipped, err := p.discoverAllEvents(context.Background(), "group", "project", since) + if err != nil { + t.Fatalf("unexpected error: %v (should be nil, MR failure is non-fatal)", err) + } + + // Issue events should still be returned. + var noteEvents []RoutableEvent + for _, e := range events { + if e.Type == "issue_note" { + noteEvents = append(noteEvents, e) + } + } + if len(noteEvents) != 1 { + t.Fatalf("expected 1 issue_note event despite MR failure, got %d", len(noteEvents)) + } + + // minSkippedAt should be set. + if minSkipped.IsZero() { + t.Error("expected minSkippedAt to be set after MR list failure") + } +} + +func TestDiscoverSlashCommands_IssueCommands(t *testing.T) { + now := time.Now().Truncate(time.Second) + since := now.Add(-time.Minute) + mc := newMockClient() + mc.events = []ProjectEvent{ + { + ID: 1, + Author: UserRef{ID: 42, Username: "alice"}, + CreatedAt: now, + Note: EventNote{ + ID: 100, + NoteableType: "Issue", + NoteableIID: 1, + Body: "/fs-triage please handle this", + }, + }, + } + + p := newEventsPoller(mc) + events, err := p.discoverSlashCommands(context.Background(), "group", "project", since) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(events) != 1 { + t.Fatalf("expected 1 event, got %d", len(events)) + } + got := events[0] + if got.Type != "issue_note" { + t.Errorf("Type = %q, want %q", got.Type, "issue_note") + } + if got.NoteID != 100 { + t.Errorf("NoteID = %d, want 100", got.NoteID) + } + if got.NoteBody != "/fs-triage please handle this" { + t.Errorf("NoteBody = %q, want %q", got.NoteBody, "/fs-triage please handle this") + } + if got.NoteAuthorID != 42 { + t.Errorf("NoteAuthorID = %d, want 42", got.NoteAuthorID) + } +} + +func TestDiscoverSlashCommands_MRCommands(t *testing.T) { + now := time.Now().Truncate(time.Second) + since := now.Add(-time.Minute) + mc := newMockClient() + mc.events = []ProjectEvent{ + { + ID: 2, + Author: UserRef{ID: 55, Username: "bob"}, + CreatedAt: now, + Note: EventNote{ + ID: 200, + NoteableType: "MergeRequest", + NoteableIID: 10, + Body: "/fs-review", + }, + }, + } + + p := newEventsPoller(mc) + events, err := p.discoverSlashCommands(context.Background(), "group", "project", since) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(events) != 1 { + t.Fatalf("expected 1 event, got %d", len(events)) + } + got := events[0] + if got.Type != "mr_note" { + t.Errorf("Type = %q, want %q", got.Type, "mr_note") + } + if got.IID != 10 { + t.Errorf("IID = %d, want 10", got.IID) + } +} + +func TestDiscoverSlashCommands_SkipsNonSlashCommands(t *testing.T) { + now := time.Now().Truncate(time.Second) + since := now.Add(-time.Minute) + mc := newMockClient() + mc.events = []ProjectEvent{ + { + ID: 3, + Author: UserRef{ID: 42, Username: "alice"}, + CreatedAt: now, + Note: EventNote{ + ID: 300, + NoteableType: "Issue", + NoteableIID: 1, + Body: "just a regular comment", + }, + }, + { + ID: 4, + Author: UserRef{ID: 42, Username: "alice"}, + CreatedAt: now, + Note: EventNote{ + ID: 301, + NoteableType: "Issue", + NoteableIID: 2, + Body: "/label ~bug", + }, + }, + } + + p := newEventsPoller(mc) + events, err := p.discoverSlashCommands(context.Background(), "group", "project", since) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(events) != 0 { + t.Errorf("expected 0 events for non-slash-command notes, got %d", len(events)) + } +} + +func TestIsProjectAccessTokenBot(t *testing.T) { + tests := []struct { + username string + want bool + }{ + {"project_123_bot_456", true}, + {"project_1_bot_2", true}, + {"project_abc_bot_xyz", true}, + {"alice", false}, + {"project_123", false}, + {"bot_user", false}, + {"", false}, + } + for _, tt := range tests { + t.Run(tt.username, func(t *testing.T) { + got := isProjectAccessTokenBot(tt.username) + if got != tt.want { + t.Errorf("isProjectAccessTokenBot(%q) = %v, want %v", tt.username, got, tt.want) + } + }) + } +} + +func TestFilterBotEvents_RemovesBotEvents(t *testing.T) { + mc := newMockClient() + p := newEventsPoller(mc) + + events := []RoutableEvent{ + {Type: "issue_note", IID: 1, NoteBody: "bot comment", IsBot: true, NoteAuthorID: 999}, + } + filtered := p.filterBotEvents(events) + if len(filtered) != 0 { + t.Errorf("expected bot event to be removed, got %d events", len(filtered)) + } +} + +func TestFilterBotEvents_RetainsBotChangesRequested(t *testing.T) { + mc := newMockClient() + p := newEventsPoller(mc) // botUserID = 100 + + events := []RoutableEvent{ + { + Type: "mr_note", + IID: 5, + NoteBody: "Changes needed here", + IsBot: true, + NoteAuthorID: 100, // matches botUserID + }, + } + filtered := p.filterBotEvents(events) + if len(filtered) != 1 { + t.Fatalf("expected bot changes-requested marker to be retained, got %d events", len(filtered)) + } +} + +func TestFilterBotEvents_RetainsNonBotEvents(t *testing.T) { + mc := newMockClient() + p := newEventsPoller(mc) + + events := []RoutableEvent{ + {Type: "issue_note", IID: 1, NoteBody: "human comment", IsBot: false, NoteAuthorID: 42}, + } + filtered := p.filterBotEvents(events) + if len(filtered) != 1 { + t.Errorf("expected non-bot event to be retained, got %d events", len(filtered)) + } +} + +func TestDeduplicate(t *testing.T) { + mc := newMockClient() + p := newEventsPoller(mc) + + events := []RoutableEvent{ + {Type: "issue_note", IID: 1, NoteID: 10, NoteBody: "hello"}, + {Type: "issue_note", IID: 1, NoteID: 10, NoteBody: "hello"}, // duplicate + {Type: "issue_note", IID: 1, NoteID: 20, NoteBody: "world"}, // different note + {Type: "issue_label", IID: 1, Labels: []string{"ready-to-code"}}, + {Type: "issue_label", IID: 1, Labels: []string{"ready-to-code"}}, // duplicate + } + unique := p.deduplicate(events) + if len(unique) != 3 { + t.Errorf("expected 3 unique events, got %d", len(unique)) + for _, e := range unique { + t.Logf(" %s (key=%s)", e.Type, e.Key()) + } + } +} + +func TestFilterRoutableLabels(t *testing.T) { + tests := []struct { + name string + labels []string + want []string + }{ + { + name: "filters to known labels", + labels: []string{"bug", "ready-to-code", "enhancement", "ready-for-review"}, + want: []string{"ready-to-code", "ready-for-review"}, + }, + { + name: "no routable labels", + labels: []string{"bug", "enhancement"}, + want: nil, + }, + { + name: "empty input", + labels: nil, + want: nil, + }, + { + name: "all routable", + labels: []string{"ready-to-code", "ready-for-review"}, + want: []string{"ready-to-code", "ready-for-review"}, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := filterRoutableLabels(tt.labels) + if len(got) != len(tt.want) { + t.Fatalf("filterRoutableLabels(%v) returned %v, want %v", tt.labels, got, tt.want) + } + for i := range got { + if got[i] != tt.want[i] { + t.Errorf("filterRoutableLabels(%v)[%d] = %q, want %q", tt.labels, i, got[i], tt.want[i]) + } + } + }) + } +} diff --git a/internal/poll/mock_test.go b/internal/poll/mock_test.go new file mode 100644 index 000000000..29e869bfe --- /dev/null +++ b/internal/poll/mock_test.go @@ -0,0 +1,173 @@ +package poll + +import ( + "context" + "fmt" + "sync" + "time" + + "github.com/fullsend-ai/fullsend/internal/forge" +) + +// emojiCall records a CreateNoteAwardEmoji invocation. +type emojiCall struct { + NoteableIID int + NoteID int + Emoji string +} + +// mockClient implements GitLabClient with configurable return values +// and error injection for deterministic testing. +type mockClient struct { + mu sync.Mutex + + issues []Issue + issuesErr error + + mrs []MergeRequest + mrsErr error + + notes map[int][]Note // keyed by issue IID + noteErr map[int]error + + mrNotes map[int][]Note // keyed by MR IID + mrNoteErr map[int]error + + events []ProjectEvent + eventsErr error + + labelEvents map[int][]ResourceLabelEvent // keyed by issue IID + labelEventsErr map[int]error + + variables map[string]string + variableErr map[string]error + updatedVars map[string]string // records UpdateCIVariable calls + + issue map[int]*Issue // keyed by IID + issueErr map[int]error + + memberLevel map[int]int // keyed by userID + memberErr map[int]error // keyed by userID + + projectPaths map[int]string // keyed by project ID + + emojis []emojiCall + + authenticatedUser string + authErr error +} + +func newMockClient() *mockClient { + return &mockClient{ + notes: make(map[int][]Note), + noteErr: make(map[int]error), + mrNotes: make(map[int][]Note), + mrNoteErr: make(map[int]error), + labelEvents: make(map[int][]ResourceLabelEvent), + labelEventsErr: make(map[int]error), + variables: make(map[string]string), + variableErr: make(map[string]error), + updatedVars: make(map[string]string), + issue: make(map[int]*Issue), + issueErr: make(map[int]error), + memberLevel: make(map[int]int), + memberErr: make(map[int]error), + projectPaths: make(map[int]string), + } +} + +func (m *mockClient) ListIssuesUpdatedSince(_ context.Context, _, _ string, _ time.Time) ([]Issue, error) { + return m.issues, m.issuesErr +} + +func (m *mockClient) ListMergeRequestsUpdatedSince(_ context.Context, _, _ string, _ time.Time) ([]MergeRequest, error) { + return m.mrs, m.mrsErr +} + +func (m *mockClient) ListProjectEvents(_ context.Context, _, _ string, _ string, _ time.Time) ([]ProjectEvent, error) { + return m.events, m.eventsErr +} + +func (m *mockClient) ListIssueNotes(_ context.Context, _, _ string, issueIID int) ([]Note, error) { + if err, ok := m.noteErr[issueIID]; ok && err != nil { + return nil, err + } + return m.notes[issueIID], nil +} + +func (m *mockClient) ListMergeRequestNotes(_ context.Context, _, _ string, mrIID int) ([]Note, error) { + if err, ok := m.mrNoteErr[mrIID]; ok && err != nil { + return nil, err + } + return m.mrNotes[mrIID], nil +} + +func (m *mockClient) ListResourceLabelEvents(_ context.Context, _, _ string, issueIID int) ([]ResourceLabelEvent, error) { + if err, ok := m.labelEventsErr[issueIID]; ok && err != nil { + return nil, err + } + return m.labelEvents[issueIID], nil +} + +func (m *mockClient) GetCIVariable(_ context.Context, _, _, name string) (string, error) { + if err, ok := m.variableErr[name]; ok && err != nil { + return "", err + } + val, ok := m.variables[name] + if !ok { + return "", forge.ErrNotFound + } + return val, nil +} + +func (m *mockClient) UpdateCIVariable(_ context.Context, _, _, name, value string, _ bool) error { + m.mu.Lock() + defer m.mu.Unlock() + m.updatedVars[name] = value + return nil +} + +func (m *mockClient) GetAuthenticatedUser(_ context.Context) (string, error) { + return m.authenticatedUser, m.authErr +} + +func (m *mockClient) CreateNoteAwardEmoji(_ context.Context, _, _ string, noteableIID, noteID int, emoji string) error { + m.mu.Lock() + defer m.mu.Unlock() + m.emojis = append(m.emojis, emojiCall{ + NoteableIID: noteableIID, + NoteID: noteID, + Emoji: emoji, + }) + return nil +} + +func (m *mockClient) GetIssue(_ context.Context, _, _ string, issueIID int) (*Issue, error) { + if err, ok := m.issueErr[issueIID]; ok && err != nil { + return nil, err + } + iss, ok := m.issue[issueIID] + if !ok { + return nil, forge.ErrNotFound + } + return iss, nil +} + +func (m *mockClient) GetMemberAccessLevel(_ context.Context, _, _ string, userID int) (int, error) { + if err, ok := m.memberErr[userID]; ok && err != nil { + return 0, err + } + level, ok := m.memberLevel[userID] + if !ok { + return 0, fmt.Errorf("member not found") + } + return level, nil +} + +func (m *mockClient) GetProjectPath(_ context.Context, projectID int) (string, error) { + path, ok := m.projectPaths[projectID] + if !ok { + return "", forge.ErrNotFound + } + return path, nil +} diff --git a/internal/poll/poll.go b/internal/poll/poll.go new file mode 100644 index 000000000..525b9b31b --- /dev/null +++ b/internal/poll/poll.go @@ -0,0 +1,190 @@ +package poll + +import ( + "context" + "fmt" + "log" + "strings" + "time" + + "github.com/fullsend-ai/fullsend/internal/dispatch" +) + +// Poller discovers GitLab events and dispatches agent stages. +type Poller struct { + client GitLabClient + router dispatch.EventRouter + projectPath string + owner string + repo string + botUserID int + gitlabURL string + opts Options + dispatches []Dispatch +} + +// New creates a Poller for the given project. +func New(client GitLabClient, router dispatch.EventRouter, projectPath string, opts Options) *Poller { + owner, repo := splitOwnerRepo(projectPath) + gitlabURL := opts.GitLabURL + if gitlabURL == "" { + gitlabURL = "https://gitlab.com" + } + return &Poller{ + client: client, + router: router, + projectPath: projectPath, + owner: owner, + repo: repo, + botUserID: opts.BotUserID, + gitlabURL: strings.TrimRight(gitlabURL, "/"), + opts: opts, + } +} + +// Run executes a single poll cycle: read watermark, discover events, +// filter, deduplicate, convert to NormalizedEvent, route, dispatch, +// and advance the watermark. +func (p *Poller) Run(ctx context.Context) error { + if p.client == nil { + return fmt.Errorf("poller requires a GitLab client (Phase 1 wiring incomplete)") + } + + lastPollAt, err := p.readWatermark(ctx, p.owner, p.repo) + if err != nil { + return fmt.Errorf("read watermark: %w", err) + } + + var events []RoutableEvent + var labelState LabelState + var minSkippedAt time.Time + if p.opts.SlashCommandsOnly { + events, err = p.discoverSlashCommands(ctx, p.owner, p.repo, lastPollAt) + } else { + events, labelState, minSkippedAt, err = p.discoverAllEvents(ctx, p.owner, p.repo, lastPollAt) + } + if err != nil { + return fmt.Errorf("discover events: %w", err) + } + + events = p.filterBotEvents(events) + events = p.deduplicate(events) + + dispatched := 0 + var maxUpdatedAt time.Time + var minFailedAt time.Time + failedLabelEvents := make(map[int]map[string]bool) + + for _, event := range events { + normalizedEvent, err := p.toNormalizedEvent(ctx, event) + if err != nil { + log.Printf("WARNING: skipping %s event on IID %d: %v", event.Type, event.IID, err) + trackFailure(&minFailedAt, event.UpdatedAt) + trackLabelFailure(failedLabelEvents, event) + continue + } + + var stages []string + if p.router != nil { + stages, err = p.router.Route(&normalizedEvent) + if err != nil { + log.Printf("dispatch core error for %s: %v", event.Key(), err) + trackFailure(&minFailedAt, event.UpdatedAt) + trackLabelFailure(failedLabelEvents, event) + continue + } + } + + if len(stages) == 0 { + if event.UpdatedAt.After(maxUpdatedAt) { + maxUpdatedAt = event.UpdatedAt + } + continue + } + + for _, stage := range stages { + if err := p.dispatch(ctx, p.owner, p.repo, stage, event); err != nil { + log.Printf("dispatch %s for %s failed: %v", stage, event.Key(), err) + trackFailure(&minFailedAt, event.UpdatedAt) + trackLabelFailure(failedLabelEvents, event) + continue + } + dispatched++ + if event.UpdatedAt.After(maxUpdatedAt) { + maxUpdatedAt = event.UpdatedAt + } + if event.NoteID != 0 && strings.HasPrefix(strings.TrimSpace(event.NoteBody), "/fs-") { + _ = p.client.CreateNoteAwardEmoji(ctx, p.owner, p.repo, event.IID, event.NoteID, "eyes") + } + } + } + + if p.opts.OutputPath != "" { + if err := p.writeDispatches(p.opts.OutputPath); err != nil { + return fmt.Errorf("write dispatches: %w", err) + } + } + + if maxUpdatedAt.IsZero() && len(events) == 0 { + maxUpdatedAt = time.Now() + } + if maxUpdatedAt.IsZero() { + log.Printf("WARNING: all %d dispatches failed, watermark not advanced", len(events)) + return nil + } + if !minFailedAt.IsZero() && minFailedAt.Before(maxUpdatedAt) { + maxUpdatedAt = minFailedAt + } + if !minSkippedAt.IsZero() && minSkippedAt.Before(maxUpdatedAt) { + maxUpdatedAt = minSkippedAt + } + newWatermark := maxUpdatedAt.Add(-30 * time.Second) + if err := p.updateWatermark(ctx, p.owner, p.repo, newWatermark); err != nil { + log.Printf("WARNING: failed to update watermark: %v", err) + } + + if labelState != nil { + for iid, failedLabels := range failedLabelEvents { + if current, ok := labelState[iid]; ok { + var kept []string + for _, label := range current { + if !failedLabels[label] { + kept = append(kept, label) + } + } + labelState[iid] = kept + } + } + if err := p.persistLabelState(ctx, p.owner, p.repo, labelState); err != nil { + log.Printf("WARNING: %v (next poll may re-dispatch label events)", err) + } + } + + log.Printf("poll complete: %d events discovered, %d dispatched", len(events), dispatched) + return nil +} + +func trackFailure(minFailedAt *time.Time, updatedAt time.Time) { + if minFailedAt.IsZero() || updatedAt.Before(*minFailedAt) { + *minFailedAt = updatedAt + } +} + +func trackLabelFailure(failedLabelEvents map[int]map[string]bool, event RoutableEvent) { + if event.Type != "issue_label" || event.ChangedLabel == "" { + return + } + if failedLabelEvents[event.IID] == nil { + failedLabelEvents[event.IID] = make(map[string]bool) + } + failedLabelEvents[event.IID][event.ChangedLabel] = true +} + +// splitOwnerRepo splits "group/subgroup/project" into owner="group/subgroup" and repo="project". +func splitOwnerRepo(projectPath string) (string, string) { + idx := strings.LastIndex(projectPath, "/") + if idx < 0 { + return "", projectPath + } + return projectPath[:idx], projectPath[idx+1:] +} diff --git a/internal/poll/poll_test.go b/internal/poll/poll_test.go new file mode 100644 index 000000000..08e7b7e4c --- /dev/null +++ b/internal/poll/poll_test.go @@ -0,0 +1,457 @@ +package poll + +import ( + "context" + "encoding/json" + "fmt" + "os" + "path/filepath" + "testing" + "time" + + "github.com/fullsend-ai/fullsend/internal/dispatch" +) + +func TestSplitOwnerRepo(t *testing.T) { + tests := []struct { + input string + wantOwner string + wantRepo string + }{ + {"org/project", "org", "project"}, + {"org/sub/project", "org/sub", "project"}, + {"org/sub1/sub2/project", "org/sub1/sub2", "project"}, + {"project", "", "project"}, + } + for _, tc := range tests { + owner, repo := splitOwnerRepo(tc.input) + if owner != tc.wantOwner || repo != tc.wantRepo { + t.Errorf("splitOwnerRepo(%q) = (%q, %q), want (%q, %q)", + tc.input, owner, repo, tc.wantOwner, tc.wantRepo) + } + } +} + +func TestNew(t *testing.T) { + mc := newMockClient() + p := New(mc, nil, "org/sub/project", Options{ + SlashCommandsOnly: true, + BotUserID: 42, + GitLabURL: "https://gitlab.example.com", + }) + if p.owner != "org/sub" { + t.Errorf("owner = %q, want %q", p.owner, "org/sub") + } + if p.repo != "project" { + t.Errorf("repo = %q, want %q", p.repo, "project") + } + if p.botUserID != 42 { + t.Errorf("botUserID = %d, want 42", p.botUserID) + } + if p.gitlabURL != "https://gitlab.example.com" { + t.Errorf("gitlabURL = %q, want %q", p.gitlabURL, "https://gitlab.example.com") + } +} + +func TestNewDefaultGitLabURL(t *testing.T) { + p := New(newMockClient(), nil, "org/project", Options{}) + if p.gitlabURL != "https://gitlab.com" { + t.Errorf("gitlabURL = %q, want default %q", p.gitlabURL, "https://gitlab.com") + } +} + +type stubRouter struct { + stages []string + err error +} + +func (r *stubRouter) Route(_ *dispatch.NormalizedEvent) ([]string, error) { + return r.stages, r.err +} + +func TestRunEmptyPoll(t *testing.T) { + now := time.Now() + mc := newMockClient() + mc.variables["FULLSEND_LAST_POLL_AT_FULL"] = now.Add(-10 * time.Minute).Format(time.RFC3339) + + dir := t.TempDir() + outputPath := filepath.Join(dir, "dispatches.json") + + p := New(mc, nil, "org/project", Options{ + OutputPath: outputPath, + }) + + err := p.Run(context.Background()) + if err != nil { + t.Fatalf("Run() error: %v", err) + } + + data, err := os.ReadFile(outputPath) + if err != nil { + t.Fatalf("read output: %v", err) + } + if string(data) != "[]\n" { + t.Errorf("output = %q, want empty JSON array", string(data)) + } + + if _, ok := mc.updatedVars["FULLSEND_LAST_POLL_AT_FULL"]; !ok { + t.Error("watermark not updated") + } +} + +func TestRunSlashCommandsOnlyMode(t *testing.T) { + now := time.Now() + mc := newMockClient() + mc.variables["FULLSEND_LAST_POLL_AT_FAST"] = now.Add(-5 * time.Minute).Format(time.RFC3339) + + dir := t.TempDir() + outputPath := filepath.Join(dir, "dispatches.json") + + p := New(mc, nil, "org/project", Options{ + SlashCommandsOnly: true, + OutputPath: outputPath, + }) + + err := p.Run(context.Background()) + if err != nil { + t.Fatalf("Run() error: %v", err) + } + + if _, ok := mc.updatedVars["FULLSEND_LAST_POLL_AT_FAST"]; !ok { + t.Error("fast watermark not updated") + } +} + +func TestTrackFailure(t *testing.T) { + var min time.Time + t1 := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC) + t2 := time.Date(2026, 1, 1, 11, 0, 0, 0, time.UTC) + + trackFailure(&min, t1) + if !min.Equal(t1) { + t.Errorf("first track: got %v, want %v", min, t1) + } + + trackFailure(&min, t2) + if !min.Equal(t2) { + t.Errorf("second track (earlier): got %v, want %v", min, t2) + } + + t3 := time.Date(2026, 1, 1, 13, 0, 0, 0, time.UTC) + trackFailure(&min, t3) + if !min.Equal(t2) { + t.Errorf("third track (later): got %v, want %v", min, t2) + } +} + +func TestTrackLabelFailure(t *testing.T) { + failed := make(map[int]map[string]bool) + + trackLabelFailure(failed, RoutableEvent{Type: "issue_note", IID: 1}) + if len(failed) != 0 { + t.Error("non-label event should not be tracked") + } + + trackLabelFailure(failed, RoutableEvent{ + Type: "issue_label", + IID: 5, + Labels: []string{"ready-to-code"}, + ChangedLabel: "ready-to-code", + }) + if !failed[5]["ready-to-code"] { + t.Error("label failure not tracked") + } +} + +func TestRunFullPollWithRouterAndDispatch(t *testing.T) { + now := time.Now().Truncate(time.Second) + since := now.Add(-10 * time.Minute) + mc := newMockClient() + mc.variables["FULLSEND_LAST_POLL_AT_FULL"] = since.Format(time.RFC3339) + mc.issues = []Issue{ + {IID: 1, Labels: []string{"bug"}, UpdatedAt: now, AuthorID: 42}, + } + mc.notes[1] = []Note{ + {ID: 10, Body: "/fs-triage handle this", CreatedAt: now, Author: UserRef{ID: 42, Username: "alice"}}, + } + mc.memberLevel[42] = 30 + mc.issue[1] = &Issue{IID: 1, AuthorID: 42} + + dir := t.TempDir() + outputPath := filepath.Join(dir, "dispatches.json") + + router := &stubRouter{stages: []string{"triage"}} + p := New(mc, router, "group/project", Options{OutputPath: outputPath}) + + if err := p.Run(context.Background()); err != nil { + t.Fatalf("Run() error: %v", err) + } + + data, err := os.ReadFile(outputPath) + if err != nil { + t.Fatalf("read output: %v", err) + } + var dispatches []Dispatch + if err := json.Unmarshal(data, &dispatches); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if len(dispatches) != 1 { + t.Fatalf("expected 1 dispatch, got %d", len(dispatches)) + } + if dispatches[0].Stage != "triage" { + t.Errorf("stage = %q, want %q", dispatches[0].Stage, "triage") + } + + if len(mc.emojis) != 1 { + t.Fatalf("expected 1 emoji reaction, got %d", len(mc.emojis)) + } + if mc.emojis[0].Emoji != "eyes" { + t.Errorf("emoji = %q, want %q", mc.emojis[0].Emoji, "eyes") + } +} + +func TestRunMultipleStages(t *testing.T) { + now := time.Now().Truncate(time.Second) + since := now.Add(-10 * time.Minute) + mc := newMockClient() + mc.variables["FULLSEND_LAST_POLL_AT_FULL"] = since.Format(time.RFC3339) + mc.issues = []Issue{ + {IID: 1, Labels: []string{"ready-to-code"}, UpdatedAt: now, AuthorID: 5}, + } + mc.notes[1] = []Note{} + mc.issue[1] = &Issue{IID: 1, AuthorID: 5} + mc.labelEvents[1] = []ResourceLabelEvent{ + { + ID: 100, + Action: "add", + Label: struct { + Name string `json:"name"` + }{Name: "ready-to-code"}, + User: UserRef{ID: 5, Username: "dev"}, + }, + } + mc.memberLevel[5] = 30 + + dir := t.TempDir() + outputPath := filepath.Join(dir, "dispatches.json") + + router := &stubRouter{stages: []string{"triage", "code"}} + p := New(mc, router, "group/project", Options{OutputPath: outputPath}) + + if err := p.Run(context.Background()); err != nil { + t.Fatalf("Run() error: %v", err) + } + + data, err := os.ReadFile(outputPath) + if err != nil { + t.Fatalf("read output: %v", err) + } + var dispatches []Dispatch + if err := json.Unmarshal(data, &dispatches); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if len(dispatches) != 2 { + t.Fatalf("expected 2 dispatches, got %d", len(dispatches)) + } + + if _, ok := mc.updatedVars["FULLSEND_LABEL_STATE"]; !ok { + t.Error("expected label state to be persisted") + } +} + +func TestRunNoMatchingStages(t *testing.T) { + now := time.Now().Truncate(time.Second) + since := now.Add(-10 * time.Minute) + mc := newMockClient() + mc.variables["FULLSEND_LAST_POLL_AT_FULL"] = since.Format(time.RFC3339) + mc.issues = []Issue{ + {IID: 1, Labels: []string{"bug"}, UpdatedAt: now, AuthorID: 42}, + } + mc.notes[1] = []Note{ + {ID: 10, Body: "just a comment", CreatedAt: now, Author: UserRef{ID: 42, Username: "alice"}}, + } + mc.memberLevel[42] = 30 + mc.issue[1] = &Issue{IID: 1, AuthorID: 42} + + dir := t.TempDir() + outputPath := filepath.Join(dir, "dispatches.json") + + router := &stubRouter{stages: nil} + p := New(mc, router, "group/project", Options{OutputPath: outputPath}) + + if err := p.Run(context.Background()); err != nil { + t.Fatalf("Run() error: %v", err) + } + + data, err := os.ReadFile(outputPath) + if err != nil { + t.Fatalf("read output: %v", err) + } + if string(data) != "[]\n" { + t.Errorf("expected empty dispatches, got %q", string(data)) + } +} + +func TestRunRouterError(t *testing.T) { + now := time.Now().Truncate(time.Second) + since := now.Add(-10 * time.Minute) + mc := newMockClient() + mc.variables["FULLSEND_LAST_POLL_AT_FULL"] = since.Format(time.RFC3339) + mc.issues = []Issue{ + {IID: 1, Labels: []string{"bug"}, UpdatedAt: now, AuthorID: 42}, + } + mc.notes[1] = []Note{ + {ID: 10, Body: "note", CreatedAt: now, Author: UserRef{ID: 42, Username: "alice"}}, + } + mc.memberLevel[42] = 30 + mc.issue[1] = &Issue{IID: 1, AuthorID: 42} + + dir := t.TempDir() + outputPath := filepath.Join(dir, "dispatches.json") + + router := &stubRouter{err: fmt.Errorf("routing failed")} + p := New(mc, router, "group/project", Options{OutputPath: outputPath}) + + if err := p.Run(context.Background()); err != nil { + t.Fatalf("Run() should not return error on router failure, got: %v", err) + } +} + +func TestRunConversionErrorSkipsEvent(t *testing.T) { + now := time.Now().Truncate(time.Second) + since := now.Add(-10 * time.Minute) + mc := newMockClient() + mc.variables["FULLSEND_LAST_POLL_AT_FULL"] = since.Format(time.RFC3339) + mc.issues = []Issue{ + {IID: 1, Labels: []string{"bug"}, UpdatedAt: now}, + } + mc.notes[1] = []Note{ + {ID: 10, Body: "note", CreatedAt: now, Author: UserRef{ID: 0}}, + } + + dir := t.TempDir() + outputPath := filepath.Join(dir, "dispatches.json") + + router := &stubRouter{stages: []string{"triage"}} + p := New(mc, router, "group/project", Options{OutputPath: outputPath}) + + if err := p.Run(context.Background()); err != nil { + t.Fatalf("Run() error: %v", err) + } + + data, err := os.ReadFile(outputPath) + if err != nil { + t.Fatalf("read output: %v", err) + } + if string(data) != "[]\n" { + t.Errorf("expected empty dispatches for unresolvable actor, got %q", string(data)) + } +} + +func TestRunAllEventsFailWatermarkNotAdvanced(t *testing.T) { + now := time.Now().Truncate(time.Second) + since := now.Add(-10 * time.Minute) + mc := newMockClient() + mc.variables["FULLSEND_LAST_POLL_AT_FULL"] = since.Format(time.RFC3339) + mc.issues = []Issue{ + {IID: 1, Labels: []string{"bug"}, UpdatedAt: now}, + {IID: 2, Labels: []string{"bug"}, UpdatedAt: now}, + } + mc.notes[1] = []Note{ + {ID: 10, Body: "note", CreatedAt: now, Author: UserRef{ID: 0}}, + } + mc.notes[2] = []Note{ + {ID: 11, Body: "note", CreatedAt: now, Author: UserRef{ID: 0}}, + } + + dir := t.TempDir() + outputPath := filepath.Join(dir, "dispatches.json") + + router := &stubRouter{stages: []string{"triage"}} + p := New(mc, router, "group/project", Options{OutputPath: outputPath}) + + if err := p.Run(context.Background()); err != nil { + t.Fatalf("Run() error: %v", err) + } + + if _, ok := mc.updatedVars["FULLSEND_LAST_POLL_AT_FULL"]; ok { + t.Error("watermark should not be advanced when all events fail") + } +} + +func TestRunNilRouter(t *testing.T) { + now := time.Now().Truncate(time.Second) + since := now.Add(-10 * time.Minute) + mc := newMockClient() + mc.variables["FULLSEND_LAST_POLL_AT_FULL"] = since.Format(time.RFC3339) + mc.issues = []Issue{ + {IID: 1, Labels: []string{"bug"}, UpdatedAt: now, AuthorID: 42}, + } + mc.notes[1] = []Note{ + {ID: 10, Body: "just a comment", CreatedAt: now, Author: UserRef{ID: 42, Username: "alice"}}, + } + mc.memberLevel[42] = 30 + mc.issue[1] = &Issue{IID: 1, AuthorID: 42} + + dir := t.TempDir() + outputPath := filepath.Join(dir, "dispatches.json") + + p := New(mc, nil, "group/project", Options{OutputPath: outputPath}) + + if err := p.Run(context.Background()); err != nil { + t.Fatalf("Run() error: %v", err) + } +} + +func TestRunLabelFailureRollback(t *testing.T) { + now := time.Now().Truncate(time.Second) + since := now.Add(-10 * time.Minute) + mc := newMockClient() + mc.variables["FULLSEND_LAST_POLL_AT_FULL"] = since.Format(time.RFC3339) + mc.issues = []Issue{ + {IID: 1, Labels: []string{"ready-to-code"}, UpdatedAt: now, AuthorID: 5}, + {IID: 2, Labels: []string{"bug"}, UpdatedAt: now, AuthorID: 42}, + } + mc.notes[1] = []Note{} + mc.notes[2] = []Note{ + {ID: 20, Body: "hello", CreatedAt: now, Author: UserRef{ID: 42, Username: "alice"}}, + } + mc.labelEvents[1] = []ResourceLabelEvent{ + { + ID: 100, + Action: "add", + Label: struct { + Name string `json:"name"` + }{Name: "ready-to-code"}, + User: UserRef{ID: 0}, // zero ID -> conversion will fail + }, + } + mc.memberLevel[42] = 30 + mc.issue[2] = &Issue{IID: 2, AuthorID: 42} + + dir := t.TempDir() + outputPath := filepath.Join(dir, "dispatches.json") + + router := &stubRouter{stages: []string{"triage"}} + p := New(mc, router, "group/project", Options{OutputPath: outputPath}) + + if err := p.Run(context.Background()); err != nil { + t.Fatalf("Run() error: %v", err) + } + + persisted, ok := mc.updatedVars["FULLSEND_LABEL_STATE"] + if !ok { + t.Fatal("expected label state to be persisted") + } + var ls LabelState + if err := json.Unmarshal([]byte(persisted), &ls); err != nil { + t.Fatalf("unmarshal label state: %v", err) + } + if labels, ok := ls[1]; ok && len(labels) > 0 { + for _, l := range labels { + if l == "ready-to-code" { + t.Error("expected ready-to-code to be rolled back from label state after dispatch failure") + } + } + } +} diff --git a/internal/poll/state.go b/internal/poll/state.go new file mode 100644 index 000000000..30266e94a --- /dev/null +++ b/internal/poll/state.go @@ -0,0 +1,138 @@ +package poll + +import ( + "context" + "encoding/json" + "errors" + "log" + "time" + + "github.com/fullsend-ai/fullsend/internal/forge" +) + +// readWatermark reads the last-polled timestamp from a CI variable. +// On first run (variable not found), it defaults to one hour ago. +func (p *Poller) readWatermark(ctx context.Context, owner, repo string) (time.Time, error) { + val, err := p.client.GetCIVariable(ctx, owner, repo, p.watermarkVarName()) + if err != nil { + if errors.Is(err, forge.ErrNotFound) { + return time.Now().Add(-1 * time.Hour), nil + } + return time.Time{}, err + } + t, err := time.Parse(time.RFC3339, val) + if err != nil { + return time.Time{}, err + } + return t, nil +} + +// watermarkVarName returns the CI variable name used for the poll watermark. +// Slash-command-only mode uses a faster polling cadence with its own variable. +func (p *Poller) watermarkVarName() string { + if p.opts.SlashCommandsOnly { + return "FULLSEND_LAST_POLL_AT_FAST" + } + return "FULLSEND_LAST_POLL_AT_FULL" +} + +// updateWatermark persists the given timestamp as the poll watermark. +func (p *Poller) updateWatermark(ctx context.Context, owner, repo string, t time.Time) error { + return p.client.UpdateCIVariable(ctx, owner, repo, p.watermarkVarName(), t.Format(time.RFC3339), true) +} + +// detectNewLabels compares current issue labels against stored state to find +// newly-added routable labels. It returns: +// - newLabels: IID → labels that were added since the last poll +// - updatedState: the new label state (caller persists after dispatch) +// - previousLabels: snapshot of prior state per issue (for rollback) +// - error +func (p *Poller) detectNewLabels(ctx context.Context, owner, repo string, issues []Issue) (map[int][]string, LabelState, map[int][]string, error) { + // Read persisted label state from CI variable. + raw, err := p.client.GetCIVariable(ctx, owner, repo, "FULLSEND_LABEL_STATE") + if err != nil && !errors.Is(err, forge.ErrNotFound) { + return nil, nil, nil, err + } + if errors.Is(err, forge.ErrNotFound) { + raw = "{}" + } + + var state LabelState + if err := json.Unmarshal([]byte(raw), &state); err != nil { + log.Printf("warning: failed to unmarshal label state, starting fresh: %v", err) + state = LabelState{} + } + + // Snapshot previous labels for each issue (used for rollback). + previousLabels := make(map[int][]string, len(issues)) + for _, iss := range issues { + if prev, ok := state[iss.IID]; ok { + cp := make([]string, len(prev)) + copy(cp, prev) + previousLabels[iss.IID] = cp + } + } + + // Detect newly-added routable labels per issue. + newLabels := make(map[int][]string, len(issues)) + polledIIDs := make(map[int]bool, len(issues)) + for _, iss := range issues { + polledIIDs[iss.IID] = true + + currentRoutable := filterRoutableLabels(iss.Labels) + prevSet := toSet(state[iss.IID]) + + var added []string + for _, lbl := range currentRoutable { + if !prevSet[lbl] { + added = append(added, lbl) + } + } + if len(added) > 0 { + newLabels[iss.IID] = added + } + + // Update state with current routable labels. + state[iss.IID] = currentRoutable + } + + // Prune closed issues that were NOT in the current poll set. + for iid := range state { + if polledIIDs[iid] { + continue + } + if p.isIssueClosed(ctx, owner, repo, iid) { + delete(state, iid) + } + } + + return newLabels, state, previousLabels, nil +} + +// persistLabelState writes the label state to a CI variable. +func (p *Poller) persistLabelState(ctx context.Context, owner, repo string, state LabelState) error { + data, err := json.Marshal(state) + if err != nil { + return err + } + return p.client.UpdateCIVariable(ctx, owner, repo, "FULLSEND_LABEL_STATE", string(data), true) +} + +// isIssueClosed checks whether the given issue is closed. +// Returns false on any error (including not-found). +func (p *Poller) isIssueClosed(ctx context.Context, owner, repo string, iid int) bool { + iss, err := p.client.GetIssue(ctx, owner, repo, iid) + if err != nil { + return false + } + return iss.State == "closed" +} + +// toSet converts a string slice to a set for O(1) lookups. +func toSet(labels []string) map[string]bool { + s := make(map[string]bool, len(labels)) + for _, l := range labels { + s[l] = true + } + return s +} diff --git a/internal/poll/state_test.go b/internal/poll/state_test.go new file mode 100644 index 000000000..6439947fd --- /dev/null +++ b/internal/poll/state_test.go @@ -0,0 +1,424 @@ +package poll + +import ( + "context" + "encoding/json" + "fmt" + "testing" + "time" +) + +func newTestPoller(client GitLabClient, opts Options) *Poller { + return &Poller{ + client: client, + owner: "testgroup", + repo: "testrepo", + opts: opts, + } +} + +// --- readWatermark tests --- + +func TestReadWatermark_ReturnsStoredTime(t *testing.T) { + stored := time.Date(2025, 6, 15, 10, 0, 0, 0, time.UTC) + mc := newMockClient() + mc.variables["FULLSEND_LAST_POLL_AT_FULL"] = stored.Format(time.RFC3339) + + p := newTestPoller(mc, Options{}) + got, err := p.readWatermark(context.Background(), "testgroup", "testrepo") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !got.Equal(stored) { + t.Errorf("got %v, want %v", got, stored) + } +} + +func TestReadWatermark_FirstRunDefaultsToOneHourAgo(t *testing.T) { + mc := newMockClient() + // No variable stored -- GetCIVariable returns ErrNotFound. + + p := newTestPoller(mc, Options{}) + before := time.Now().Add(-1*time.Hour - time.Second) + got, err := p.readWatermark(context.Background(), "testgroup", "testrepo") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + after := time.Now().Add(-1*time.Hour + time.Second) + + if got.Before(before) || got.After(after) { + t.Errorf("expected watermark ~1 hour ago, got %v (window %v to %v)", got, before, after) + } +} + +func TestReadWatermark_ParseError(t *testing.T) { + mc := newMockClient() + mc.variables["FULLSEND_LAST_POLL_AT_FULL"] = "not-a-timestamp" + + p := newTestPoller(mc, Options{}) + _, err := p.readWatermark(context.Background(), "testgroup", "testrepo") + if err == nil { + t.Fatal("expected parse error, got nil") + } +} + +func TestReadWatermark_ClientError(t *testing.T) { + mc := newMockClient() + mc.variableErr["FULLSEND_LAST_POLL_AT_FULL"] = fmt.Errorf("network failure") + + p := newTestPoller(mc, Options{}) + _, err := p.readWatermark(context.Background(), "testgroup", "testrepo") + if err == nil { + t.Fatal("expected error, got nil") + } + if err.Error() != "network failure" { + t.Errorf("unexpected error: %v", err) + } +} + +// --- watermarkVarName tests --- + +func TestWatermarkVarName_FastMode(t *testing.T) { + p := newTestPoller(nil, Options{SlashCommandsOnly: true}) + got := p.watermarkVarName() + if got != "FULLSEND_LAST_POLL_AT_FAST" { + t.Errorf("got %q, want FULLSEND_LAST_POLL_AT_FAST", got) + } +} + +func TestWatermarkVarName_FullMode(t *testing.T) { + p := newTestPoller(nil, Options{SlashCommandsOnly: false}) + got := p.watermarkVarName() + if got != "FULLSEND_LAST_POLL_AT_FULL" { + t.Errorf("got %q, want FULLSEND_LAST_POLL_AT_FULL", got) + } +} + +// --- updateWatermark tests --- + +func TestUpdateWatermark_StoresRFC3339(t *testing.T) { + mc := newMockClient() + p := newTestPoller(mc, Options{}) + + ts := time.Date(2025, 7, 1, 14, 30, 0, 0, time.UTC) + err := p.updateWatermark(context.Background(), "testgroup", "testrepo", ts) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + want := ts.Format(time.RFC3339) + got := mc.updatedVars["FULLSEND_LAST_POLL_AT_FULL"] + if got != want { + t.Errorf("stored %q, want %q", got, want) + } +} + +func TestUpdateWatermark_UsesFastVarForSlashOnly(t *testing.T) { + mc := newMockClient() + p := newTestPoller(mc, Options{SlashCommandsOnly: true}) + + ts := time.Date(2025, 7, 1, 14, 30, 0, 0, time.UTC) + err := p.updateWatermark(context.Background(), "testgroup", "testrepo", ts) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if _, ok := mc.updatedVars["FULLSEND_LAST_POLL_AT_FAST"]; !ok { + t.Error("expected FULLSEND_LAST_POLL_AT_FAST to be set, got nothing") + } +} + +// --- detectNewLabels tests --- + +func TestDetectNewLabels_NewLabelsDetected(t *testing.T) { + mc := newMockClient() + // Pre-existing state: issue 1 had "ready-to-code" + mc.variables["FULLSEND_LABEL_STATE"] = `{"1":["ready-to-code"]}` + + p := newTestPoller(mc, Options{}) + issues := []Issue{ + {IID: 1, Labels: []string{"ready-to-code", "ready-for-review"}}, + } + + newLabels, _, _, err := p.detectNewLabels(context.Background(), "testgroup", "testrepo", issues) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + labels, ok := newLabels[1] + if !ok { + t.Fatal("expected new labels for issue 1") + } + if len(labels) != 1 || labels[0] != "ready-for-review" { + t.Errorf("got %v, want [ready-for-review]", labels) + } +} + +func TestDetectNewLabels_PreExistingNotRedetected(t *testing.T) { + mc := newMockClient() + // Issue 1 already had both routable labels. + mc.variables["FULLSEND_LABEL_STATE"] = `{"1":["ready-to-code","ready-for-review"]}` + + p := newTestPoller(mc, Options{}) + issues := []Issue{ + {IID: 1, Labels: []string{"ready-to-code", "ready-for-review"}}, + } + + newLabels, _, _, err := p.detectNewLabels(context.Background(), "testgroup", "testrepo", issues) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if len(newLabels) != 0 { + t.Errorf("expected no new labels, got %v", newLabels) + } +} + +func TestDetectNewLabels_FirstRunNoStoredState(t *testing.T) { + mc := newMockClient() + // No FULLSEND_LABEL_STATE variable -- GetCIVariable returns ErrNotFound. + + p := newTestPoller(mc, Options{}) + issues := []Issue{ + {IID: 42, Labels: []string{"ready-to-code", "bug"}}, + } + + newLabels, state, _, err := p.detectNewLabels(context.Background(), "testgroup", "testrepo", issues) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + // On first run, all routable labels are "new". + labels, ok := newLabels[42] + if !ok { + t.Fatal("expected new labels for issue 42") + } + if len(labels) != 1 || labels[0] != "ready-to-code" { + t.Errorf("got %v, want [ready-to-code]", labels) + } + + // State should now track the routable labels. + if tracked, ok := state[42]; !ok || len(tracked) != 1 { + t.Errorf("expected state to track issue 42 with 1 label, got %v", state) + } +} + +func TestDetectNewLabels_CorruptStateGracefulDegradation(t *testing.T) { + mc := newMockClient() + mc.variables["FULLSEND_LABEL_STATE"] = `{invalid json` + + p := newTestPoller(mc, Options{}) + issues := []Issue{ + {IID: 5, Labels: []string{"ready-for-review"}}, + } + + newLabels, _, _, err := p.detectNewLabels(context.Background(), "testgroup", "testrepo", issues) + if err != nil { + t.Fatalf("expected graceful degradation, got error: %v", err) + } + + // With corrupt state treated as empty, all routable labels are "new". + labels, ok := newLabels[5] + if !ok { + t.Fatal("expected new labels for issue 5") + } + if len(labels) != 1 || labels[0] != "ready-for-review" { + t.Errorf("got %v, want [ready-for-review]", labels) + } +} + +func TestDetectNewLabels_PrunesClosedIssues(t *testing.T) { + mc := newMockClient() + // State has issue 99 which is NOT in the current poll set. + mc.variables["FULLSEND_LABEL_STATE"] = `{"10":["ready-to-code"],"99":["ready-to-code"]}` + // Issue 99 is closed. + mc.issue[99] = &Issue{IID: 99, State: "closed"} + + p := newTestPoller(mc, Options{}) + issues := []Issue{ + {IID: 10, Labels: []string{"ready-to-code"}}, + } + + _, state, _, err := p.detectNewLabels(context.Background(), "testgroup", "testrepo", issues) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + // Issue 99 should be pruned from state. + if _, ok := state[99]; ok { + t.Error("expected closed issue 99 to be pruned from state") + } + // Issue 10 should remain. + if _, ok := state[10]; !ok { + t.Error("expected issue 10 to remain in state") + } +} + +func TestDetectNewLabels_DoesNotPruneOpenIssues(t *testing.T) { + mc := newMockClient() + mc.variables["FULLSEND_LABEL_STATE"] = `{"10":["ready-to-code"],"88":["ready-for-review"]}` + // Issue 88 is open -- not in poll set but not closed. + mc.issue[88] = &Issue{IID: 88, State: "opened"} + + p := newTestPoller(mc, Options{}) + issues := []Issue{ + {IID: 10, Labels: []string{"ready-to-code"}}, + } + + _, state, _, err := p.detectNewLabels(context.Background(), "testgroup", "testrepo", issues) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + if _, ok := state[88]; !ok { + t.Error("expected open issue 88 to remain in state") + } +} + +func TestDetectNewLabels_PreviousLabelsSnapshot(t *testing.T) { + mc := newMockClient() + mc.variables["FULLSEND_LABEL_STATE"] = `{"7":["ready-to-code"]}` + + p := newTestPoller(mc, Options{}) + issues := []Issue{ + {IID: 7, Labels: []string{"ready-to-code", "ready-for-review"}}, + } + + _, _, previousLabels, err := p.detectNewLabels(context.Background(), "testgroup", "testrepo", issues) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + prev, ok := previousLabels[7] + if !ok { + t.Fatal("expected previous labels for issue 7") + } + if len(prev) != 1 || prev[0] != "ready-to-code" { + t.Errorf("got %v, want [ready-to-code]", prev) + } +} + +func TestDetectNewLabels_ClientError(t *testing.T) { + mc := newMockClient() + mc.variableErr["FULLSEND_LABEL_STATE"] = fmt.Errorf("api timeout") + + p := newTestPoller(mc, Options{}) + _, _, _, err := p.detectNewLabels(context.Background(), "testgroup", "testrepo", nil) + if err == nil { + t.Fatal("expected error, got nil") + } +} + +// --- persistLabelState tests --- + +func TestPersistLabelState_MarshalsAndWrites(t *testing.T) { + mc := newMockClient() + p := newTestPoller(mc, Options{}) + + state := LabelState{ + 1: {"ready-to-code"}, + 2: {"ready-for-review", "ready-to-code"}, + } + + err := p.persistLabelState(context.Background(), "testgroup", "testrepo", state) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + raw, ok := mc.updatedVars["FULLSEND_LABEL_STATE"] + if !ok { + t.Fatal("expected FULLSEND_LABEL_STATE to be set") + } + + var roundtripped LabelState + if err := json.Unmarshal([]byte(raw), &roundtripped); err != nil { + t.Fatalf("stored value is not valid JSON: %v", err) + } + + for iid, wantLabels := range state { + gotLabels := roundtripped[iid] + if len(gotLabels) != len(wantLabels) { + t.Errorf("issue %d: got %v, want %v", iid, gotLabels, wantLabels) + } + } +} + +// --- isIssueClosed tests --- + +func TestIsIssueClosed_ReturnsTrue(t *testing.T) { + mc := newMockClient() + mc.issue[42] = &Issue{IID: 42, State: "closed"} + + p := newTestPoller(mc, Options{}) + if !p.isIssueClosed(context.Background(), "testgroup", "testrepo", 42) { + t.Error("expected true for closed issue") + } +} + +func TestIsIssueClosed_ReturnsFalseForOpen(t *testing.T) { + mc := newMockClient() + mc.issue[42] = &Issue{IID: 42, State: "opened"} + + p := newTestPoller(mc, Options{}) + if p.isIssueClosed(context.Background(), "testgroup", "testrepo", 42) { + t.Error("expected false for open issue") + } +} + +func TestIsIssueClosed_ReturnsFalseOnError(t *testing.T) { + mc := newMockClient() + mc.issueErr[42] = fmt.Errorf("server error") + + p := newTestPoller(mc, Options{}) + if p.isIssueClosed(context.Background(), "testgroup", "testrepo", 42) { + t.Error("expected false on error") + } +} + +func TestIsIssueClosed_ReturnsFalseOnNotFound(t *testing.T) { + mc := newMockClient() + // No issue[42] entry -- GetIssue returns forge.ErrNotFound. + + p := newTestPoller(mc, Options{}) + if p.isIssueClosed(context.Background(), "testgroup", "testrepo", 42) { + t.Error("expected false when issue not found") + } +} + +// --- toSet tests --- + +func TestToSet_BasicConversion(t *testing.T) { + s := toSet([]string{"a", "b", "c"}) + if len(s) != 3 { + t.Fatalf("expected 3 elements, got %d", len(s)) + } + for _, k := range []string{"a", "b", "c"} { + if !s[k] { + t.Errorf("expected %q in set", k) + } + } +} + +func TestToSet_EmptySlice(t *testing.T) { + s := toSet([]string{}) + if len(s) != 0 { + t.Errorf("expected empty set, got %d elements", len(s)) + } +} + +func TestToSet_NilSlice(t *testing.T) { + s := toSet(nil) + if s == nil { + t.Fatal("expected non-nil map") + } + if len(s) != 0 { + t.Errorf("expected empty set, got %d elements", len(s)) + } +} + +func TestToSet_Duplicates(t *testing.T) { + s := toSet([]string{"x", "x", "y"}) + if len(s) != 2 { + t.Errorf("expected 2 elements, got %d", len(s)) + } +} diff --git a/internal/poll/types.go b/internal/poll/types.go new file mode 100644 index 000000000..90deb2c99 --- /dev/null +++ b/internal/poll/types.go @@ -0,0 +1,66 @@ +package poll + +import ( + "fmt" + "time" +) + +// Options configures the poller. +type Options struct { + SlashCommandsOnly bool + BotUserID int + OutputPath string + GitLabURL string +} + +// RoutableEvent is an intermediate representation of a detected change, +// produced by event discovery and consumed by the poll loop for +// NormalizedEvent conversion and dispatch. +type RoutableEvent struct { + Type string + IID int + UpdatedAt time.Time + Labels []string // full label set at event time + ChangedLabel string // the label that was added/removed (label events only) + NoteBody string + NoteID int + NoteAuthorID int + NoteAuthorLogin string + IsBot bool + MRSource int + MRTarget int + SourceBranch string + TargetBranch string + MRAuthorID int + MRAuthorLogin string + MergedByLogin string +} + +// Key returns a deduplication key for the event. +func (e RoutableEvent) Key() string { + if e.NoteID != 0 { + return fmt.Sprintf("note-%d", e.NoteID) + } + if e.ChangedLabel != "" { + return fmt.Sprintf("%s-%d-%s", e.Type, e.IID, e.ChangedLabel) + } + return fmt.Sprintf("%s-%d", e.Type, e.IID) +} + +// LabelState tracks previously-seen labels per issue IID. +type LabelState map[int][]string + +// Dispatch represents a single child pipeline dispatch record. +type Dispatch struct { + Stage string `json:"stage"` + EventType string `json:"event_type"` + EventPayloadB64 string `json:"event_payload_b64"` + ResourceKey string `json:"resource_key"` +} + +// LabelAuthor identifies who applied a label. +type LabelAuthor struct { + ID int + Username string + IsBot bool +}