From c11a6bbfd2332982eec2f89bf54ad86d26c5f79f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Salavisa?= Date: Tue, 21 Apr 2026 17:32:46 +0100 Subject: [PATCH] feat(github): add issue watcher with task creation and auto-cleanup (#672) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(github): add issue watcher with task creation and auto-cleanup Add a GitHub Issue Watcher feature that polls for new issues matching configurable filters and automatically creates tasks for them. This complements the existing PR Review Watcher with the same patterns. Backend: - Issue watch CRUD (models, store, service, controller, WS handlers) - Poller integration with configurable intervals - Event-driven task creation via orchestrator - Repository resolution and worktree support for issue tasks - Auto-cleanup of tasks for closed issues (CleanupClosedIssueTasks) - GetIssueState client method across all implementations - Default branch auto-detection from cloned repo HEAD Frontend: - Issue watch settings UI (table, dialog, placeholders) - Issue link icon on task cards (green circle-dot) - API client, store slice, and hook for issue watches - SSR and WS hydration of issue metadata * fix: resolve golangci-lint errors (goconst, nestif, dupl) and prettier formatting * fix(github): fix issue watch update shape, trigger key, and label quoting - httpUpdateIssueWatch now returns the full IssueWatch object instead of {updated:true} so the frontend store correctly refreshes after edits - httpTriggerIssueWatch returns new_issues_found to match TriggerIssueResponse type; was returning new_issues so the toast count was always 0 - buildIssueFilter uses conditional quoting instead of fmt.Sprintf %q to avoid Go escape sequences in GitHub search qualifiers Co-authored-by: João Salavisa * fix: resolve eslint warnings (max-lines-per-function, no-nested-ternary) * test(github): add issue watcher tests and fix review feedback - Add TestInterpolateIssuePrompt (4 cases: empty/all/none/partial) - Add TestCreateIssueTask (3 cases: skip/reserve+assign/release on fail) - Add TestShouldDeleteIssueTask (4 cases: open/closed+sessions/closed/error) - Fix WS response key new_issues -> new_issues_found - Fix fmt.Printf -> fmt.Fprintf(os.Stderr) in store.go - Add CleanupClosedIssueTasks call in WS trigger handler * fix: address PR review feedback and CI failures - Split test files to fix revive file-length-limit (service_test.go, event_handlers_github_test.go) - Add session checker guard in CleanupClosedIssueTasks - Normalize poll intervals on UpdateIssueWatch - Add cleanup call in TriggerAllIssueChecks loop - Log cleanup errors in controller and WS handler - Prefer refs/remotes/origin/HEAD in detectGitDefaultBranch - Add aria-label to issue task icon - Fix TypeScript type errors: add linkedIssues to GitHubSlice, fix hook exports, remove invalid imports * fix: split test files exceeding revive limit and clean up minor issues - Split service_test.go (883→779 lines): move TestTriggerPRSync_* to service_pr_sync_test.go - Split event_handlers_github_test.go (806→677 lines): move TestCreateReviewTask_* and helpers to event_handlers_github_review_test.go - Remove redundant s.taskSessionChecker nil-check inside shouldDeleteIssueTask (early guard at CleanupClosedIssueTasks already ensures it is non-nil) - Remove unreachable dead guard in detectAndPersistDefaultBranch (repo.DefaultBranch is always "" at that call site) - Fix bogus #0 issue badge in task-session-sidebar.tsx when issueNumber is absent Co-authored-by: João Salavisa * fix: address Claude review — poll clamp and prompt injection - Handle v <= 0 in UpdateIssueWatch poll interval (fall back to default) - Remove {{issue.body}} from default template to prevent prompt injection - Update test expected output for new default template * fix: remove {{issue.body}} from frontend DEFAULT_ISSUE_WATCH_PROMPT Mirrors the backend fix — the UI pre-fills the prompt so the backend fallback was never reached. The placeholder stays in ISSUE_WATCH_PLACEHOLDERS for explicit opt-in. * feat(github): add isIssueWatch sidebar filter dimension Add isIssueWatch boolean to KanbanState, TaskSwitcherItem, and sidebar filter registry — mirroring the isPRReview pattern. Issue-watch tasks (created from issue_watch_id in metadata) can now be explicitly filtered in the sidebar, just like PR review tasks. Files changed: - sidebar-view-types: add isIssueWatch dimension - filter-dimension-registry: add Issue watch filter meta - apply-view: add isIssueWatch extractor - kanban types: add isIssueWatch field - TaskSwitcherItem: add isIssueWatch field - use-all-workflow-snapshots: derive isIssueWatch from metadata - ssr/mapper: derive isIssueWatch from metadata - task-session-sidebar: pass isIssueWatch, extract helpers to fix complexity * fix: correct deleteIssueWatch response type to match backend Backend httpDeleteIssueWatch returns {"deleted": true}; the frontend type was { success: boolean } — a mismatch with no functional impact (return value is ignored) but wrong per TypeScript contract. Co-authored-by: João Salavisa * refactor: extract metadata helpers to shared lib/metadata-utils.ts - Move isPRReviewFromMetadata, isIssueWatchFromMetadata, and issueFieldsFromMetadata to a shared module to eliminate duplicate definitions (sonarjs/no-identical-functions). - Fix deleteIssueWatch response type: { success: boolean } -> { deleted: boolean } to match the actual backend response. * style: fix prettier formatting * test(web): add tests for isIssueWatch filter dimension and metadata-utils Co-authored-by: João Salavisa Co-authored-by: Claude Sonnet 4.6 * style: fix prettier formatting in test files * fix(github): cascade-delete issue watch tasks on watch deletion and add isPRReview to SSR mapper - DeleteIssueWatch now wraps in a transaction and deletes related github_issue_watch_tasks rows before removing the watch; prevents orphaned dedup rows accumulating after a watch is deleted - snapshotToState in mapper.ts now also sets isPRReview from metadata, matching the mapping already done in use-all-workflow-snapshots.ts Co-authored-by: João Salavisa * style: extract duplicate string constant in metadata-utils test * style: extract repeated WS error strings into constants in handlers.go goconst min-occurrences: 3 — 'invalid payload' (6x) and 'id required' (4x) both exceed the threshold. Extract to package-level constants. Co-authored-by: João Salavisa * style: extract repeated issue key strings into constants in event_handlers_github.go goconst min-occurrences: 3 — 'issue_number' (9x) and 'issue_watch_id' (5x) both newly introduced in this PR exceed the threshold. Co-authored-by: João Salavisa * style: fix prettier formatting in metadata-utils test --------- Co-authored-by: Kandev Agent Co-authored-by: claude[bot] <41898282+claude[bot]@users.noreply.github.com> Co-authored-by: João Salavisa Co-authored-by: Claude Sonnet 4.6 --- apps/backend/cmd/kandev/orchestrator.go | 79 +++ .../config/prompts/issue-watch-default.md | 16 + apps/backend/internal/events/types.go | 1 + apps/backend/internal/github/client.go | 8 + .../backend/internal/github/client_helpers.go | 84 +++ apps/backend/internal/github/controller.go | 106 ++++ .../internal/github/controller_test.go | 6 + apps/backend/internal/github/gh_client.go | 27 + apps/backend/internal/github/handlers.go | 175 +++++-- apps/backend/internal/github/mock_client.go | 8 + apps/backend/internal/github/models.go | 94 ++++ apps/backend/internal/github/noop_client.go | 8 + apps/backend/internal/github/pat_client.go | 23 + apps/backend/internal/github/poller.go | 69 ++- apps/backend/internal/github/service.go | 383 ++++++++++++++ .../internal/github/service_issue_test.go | 91 ++++ .../internal/github/service_pr_sync_test.go | 110 ++++ .../github/service_review_cleanup_test.go | 126 +++++ apps/backend/internal/github/service_test.go | 223 -------- apps/backend/internal/github/store.go | 239 ++++++++- .../orchestrator/event_handlers_github.go | 239 +++++++++ .../event_handlers_github_issue_test.go | 188 +++++++ .../event_handlers_github_review_test.go | 139 +++++ .../event_handlers_github_test.go | 149 +----- apps/backend/internal/orchestrator/service.go | 3 + apps/backend/pkg/websocket/actions.go | 9 + .../web/components/github/github-settings.tsx | 234 +++++++-- .../web/components/github/issue-task-icon.tsx | 27 + .../components/github/issue-watch-dialog.tsx | 489 ++++++++++++++++++ .../github/issue-watch-placeholders.ts | 63 +++ .../components/github/issue-watch-table.tsx | 168 ++++++ .../filter-dimension-registry.ts | 8 + apps/web/components/task/task-item.tsx | 7 + .../components/task/task-session-sidebar.tsx | 35 +- apps/web/components/task/task-switcher.tsx | 3 + .../hooks/domains/github/use-issue-watches.ts | 85 +++ .../kanban/use-all-workflow-snapshots.ts | 13 +- apps/web/lib/api/domains/github-api.ts | 60 +++ apps/web/lib/metadata-utils.test.ts | 81 +++ apps/web/lib/metadata-utils.ts | 31 ++ apps/web/lib/sidebar/apply-view.test.ts | 18 + apps/web/lib/sidebar/apply-view.ts | 1 + apps/web/lib/ssr/mapper.ts | 8 + apps/web/lib/state/default-state.ts | 2 + .../lib/state/slices/github/github-slice.ts | 34 ++ apps/web/lib/state/slices/github/types.ts | 14 +- apps/web/lib/state/slices/index.ts | 1 + apps/web/lib/state/slices/kanban/types.ts | 3 + .../lib/state/slices/ui/sidebar-view-types.ts | 1 + apps/web/lib/state/store.ts | 9 + apps/web/lib/types/github.ts | 45 ++ 51 files changed, 3580 insertions(+), 463 deletions(-) create mode 100644 apps/backend/config/prompts/issue-watch-default.md create mode 100644 apps/backend/internal/github/service_issue_test.go create mode 100644 apps/backend/internal/github/service_pr_sync_test.go create mode 100644 apps/backend/internal/github/service_review_cleanup_test.go create mode 100644 apps/backend/internal/orchestrator/event_handlers_github_issue_test.go create mode 100644 apps/backend/internal/orchestrator/event_handlers_github_review_test.go create mode 100644 apps/web/components/github/issue-task-icon.tsx create mode 100644 apps/web/components/github/issue-watch-dialog.tsx create mode 100644 apps/web/components/github/issue-watch-placeholders.ts create mode 100644 apps/web/components/github/issue-watch-table.tsx create mode 100644 apps/web/hooks/domains/github/use-issue-watches.ts create mode 100644 apps/web/lib/metadata-utils.test.ts create mode 100644 apps/web/lib/metadata-utils.ts diff --git a/apps/backend/cmd/kandev/orchestrator.go b/apps/backend/cmd/kandev/orchestrator.go index 33ba18027f..c5832e33d3 100644 --- a/apps/backend/cmd/kandev/orchestrator.go +++ b/apps/backend/cmd/kandev/orchestrator.go @@ -5,6 +5,7 @@ import ( "crypto/sha1" "errors" "fmt" + "os" "path/filepath" "strings" @@ -90,6 +91,9 @@ func provideOrchestrator( // Wire review task creator for auto-creating tasks from review watch PRs orchestratorSvc.SetReviewTaskCreator(&reviewTaskCreatorAdapter{svc: taskSvc}) + // Wire issue task creator for auto-creating tasks from issue watch events + orchestratorSvc.SetIssueTaskCreator(&issueTaskCreatorAdapter{svc: taskSvc}) + // Wire repository resolver for auto-cloning repos during review task creation if repoCloner != nil { orchestratorSvc.SetRepositoryResolver(&repositoryResolverAdapter{ @@ -214,6 +218,31 @@ func (a *reviewTaskCreatorAdapter) CreateReviewTask(ctx context.Context, req *or }) } +// issueTaskCreatorAdapter adapts the task service to the orchestrator's IssueTaskCreator interface. +type issueTaskCreatorAdapter struct { + svc *taskservice.Service +} + +// CreateIssueTask implements orchestrator.IssueTaskCreator. +func (a *issueTaskCreatorAdapter) CreateIssueTask(ctx context.Context, req *orchestrator.IssueTaskRequest) (*taskmodels.Task, error) { + var repos []taskservice.TaskRepositoryInput + for _, r := range req.Repositories { + repos = append(repos, taskservice.TaskRepositoryInput{ + RepositoryID: r.RepositoryID, + BaseBranch: r.BaseBranch, + }) + } + return a.svc.CreateTask(ctx, &taskservice.CreateTaskRequest{ + WorkspaceID: req.WorkspaceID, + WorkflowID: req.WorkflowID, + WorkflowStepID: req.WorkflowStepID, + Title: req.Title, + Description: req.Description, + Metadata: req.Metadata, + Repositories: repos, + }) +} + // repoLocalPathUpdater adapts the task service's UpdateRepository to the executor.RepoUpdater interface. type repoLocalPathUpdater struct { svc *taskservice.Service @@ -267,5 +296,55 @@ func (a *repositoryResolverAdapter) ResolveForReview( if baseBranch == "" { baseBranch = repo.DefaultBranch } + // When no default branch is known (e.g. issue watch with no PR context), + // detect it from the cloned repo's HEAD. + if baseBranch == "" && localPath != "" { + baseBranch = a.detectAndPersistDefaultBranch(ctx, repo, localPath) + } return repo.ID, baseBranch, nil } + +// detectAndPersistDefaultBranch reads the default branch from the local clone +// and persists it to the repository record for future lookups. +func (a *repositoryResolverAdapter) detectAndPersistDefaultBranch( + ctx context.Context, repo *taskmodels.Repository, localPath string, +) string { + detected := detectGitDefaultBranch(localPath) + if detected == "" { + return "" + } + if _, err := a.taskSvc.UpdateRepository(ctx, repo.ID, &taskservice.UpdateRepositoryRequest{ + DefaultBranch: &detected, + }); err != nil { + a.logger.Warn("failed to persist detected default branch", + zap.String("repository_id", repo.ID), + zap.String("branch", detected), + zap.Error(err)) + } + return detected +} + +// detectGitDefaultBranch reads the default branch of a git repository. +// It first checks refs/remotes/origin/HEAD (set by `git clone`), then +// falls back to .git/HEAD. Returns empty string on any failure. +func detectGitDefaultBranch(repoPath string) string { + // Prefer the remote default branch pointer set by `git clone`. + originHead := filepath.Join(repoPath, ".git", "refs", "remotes", "origin", "HEAD") + if content, err := os.ReadFile(originHead); err == nil { + trimmed := strings.TrimSpace(string(content)) + if after, ok := strings.CutPrefix(trimmed, "ref: refs/remotes/origin/"); ok { + return after + } + } + // Fall back to the local HEAD (works for fresh clones that lack origin/HEAD). + headPath := filepath.Join(repoPath, ".git", "HEAD") + content, err := os.ReadFile(headPath) + if err != nil { + return "" + } + trimmed := strings.TrimSpace(string(content)) + if after, ok := strings.CutPrefix(trimmed, "ref: refs/heads/"); ok { + return after + } + return "" +} diff --git a/apps/backend/config/prompts/issue-watch-default.md b/apps/backend/config/prompts/issue-watch-default.md new file mode 100644 index 0000000000..5630e07869 --- /dev/null +++ b/apps/backend/config/prompts/issue-watch-default.md @@ -0,0 +1,16 @@ +You have been assigned a GitHub issue to work on. + +**Issue:** {{issue.link}} +**Title:** {{issue.title}} (#{{issue.number}}) +**Repository:** {{issue.repo}} +**Author:** {{issue.author}} +**Labels:** {{issue.labels}} + +## Instructions + +1. Read the issue description carefully and understand the requirements. +2. Explore the codebase to understand the relevant code and architecture. +3. Implement the changes described in the issue. +4. Write or update tests to cover the changes. +5. Run the test suite to ensure nothing is broken. +6. Commit your changes with a descriptive commit message referencing the issue. diff --git a/apps/backend/internal/events/types.go b/apps/backend/internal/events/types.go index 4437b1580c..06febd1272 100644 --- a/apps/backend/internal/events/types.go +++ b/apps/backend/internal/events/types.go @@ -198,6 +198,7 @@ const ( GitHubPRFeedback = "github.pr_feedback" // PR has new feedback (UI notification only) GitHubPRStateChanged = "github.pr_state_changed" // PR state changed (merged, closed, etc.) GitHubNewReviewPR = "github.new_pr_to_review" // New PR found needing review + GitHubNewIssue = "github.new_issue" // New issue found matching issue watch GitHubTaskPRUpdated = "github.task_pr.updated" // TaskPR record updated (for UI refresh) GitHubWatchEvent = "github.watch.event" // Watch created/deleted ) diff --git a/apps/backend/internal/github/client.go b/apps/backend/internal/github/client.go index e423fc6c62..7140695544 100644 --- a/apps/backend/internal/github/client.go +++ b/apps/backend/internal/github/client.go @@ -64,4 +64,12 @@ type Client interface { // ListRepoBranches lists branches for a repository. ListRepoBranches(ctx context.Context, owner, repo string) ([]RepoBranch, error) + + // ListIssues searches for open issues (not PRs) matching the given query. + // filter is an optional additional search qualifier (e.g. "repo:owner/name" or "label:bug"). + // customQuery, when non-empty, replaces the entire generated query. + ListIssues(ctx context.Context, filter, customQuery string) ([]*Issue, error) + + // GetIssueState returns the state of a single issue ("open" or "closed"). + GetIssueState(ctx context.Context, owner, repo string, number int) (string, error) } diff --git a/apps/backend/internal/github/client_helpers.go b/apps/backend/internal/github/client_helpers.go index 1c0073baca..6ebfc81ee1 100644 --- a/apps/backend/internal/github/client_helpers.go +++ b/apps/backend/internal/github/client_helpers.go @@ -304,6 +304,90 @@ func convertSearchItemToPR( } } +// buildIssueSearchQuery assembles a GitHub search query for issues. +// When customQuery is non-empty, it is used verbatim as the entire query. +// Otherwise a query is built from optional filter qualifiers. +func buildIssueSearchQuery(filter, customQuery string) string { + if customQuery != "" { + return customQuery + } + base := "type:issue state:open" + if filter != "" { + base += " " + filter + } + return base +} + +// issueSearchItem is an issue item from the GitHub search API. +type issueSearchItem struct { + Number int `json:"number"` + Title string `json:"title"` + Body string `json:"body"` + HTMLURL string `json:"html_url"` + State string `json:"state"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` + ClosedAt *string `json:"closed_at"` + RepositoryURL string `json:"repository_url"` + User struct { + Login string `json:"login"` + } `json:"user"` + Labels []struct { + Name string `json:"name"` + } `json:"labels"` + Assignees []struct { + Login string `json:"login"` + } `json:"assignees"` + PullRequest *struct { + URL string `json:"url"` + } `json:"pull_request"` +} + +// convertSearchItemToIssue converts a search API item into an Issue struct. +func convertSearchItemToIssue(item issueSearchItem) *Issue { + owner, repo := parseRepoURL(item.RepositoryURL) + labels := make([]string, len(item.Labels)) + for i, l := range item.Labels { + labels[i] = l.Name + } + assignees := make([]string, len(item.Assignees)) + for i, a := range item.Assignees { + assignees[i] = a.Login + } + issue := &Issue{ + Number: item.Number, + Title: item.Title, + Body: item.Body, + HTMLURL: item.HTMLURL, + State: item.State, + AuthorLogin: item.User.Login, + RepoOwner: owner, + RepoName: repo, + Labels: labels, + Assignees: assignees, + CreatedAt: item.CreatedAt, + UpdatedAt: item.UpdatedAt, + } + if item.ClosedAt != nil { + issue.ClosedAt = parseTimePtr(*item.ClosedAt) + } + return issue +} + +// parseIssueSearchResults parses GitHub search API response items into Issue structs, +// filtering out any pull requests (which also appear in search/issues results). +func parseIssueSearchResults(items []issueSearchItem) []*Issue { + var issues []*Issue + for _, item := range items { + // GitHub search/issues returns both issues and PRs; skip PRs + if item.PullRequest != nil { + continue + } + issues = append(issues, convertSearchItemToIssue(item)) + } + return issues +} + // latestReviewByAuthor returns a map of the most recent review per author. func latestReviewByAuthor(reviews []PRReview) map[string]PRReview { latest := make(map[string]PRReview) diff --git a/apps/backend/internal/github/controller.go b/apps/backend/internal/github/controller.go index 336257dcdf..f8080c95bd 100644 --- a/apps/backend/internal/github/controller.go +++ b/apps/backend/internal/github/controller.go @@ -46,6 +46,13 @@ func (c *Controller) RegisterHTTPRoutes(router *gin.Engine) { api.POST("/watches/review/:id/trigger", c.httpTriggerReviewWatch) api.POST("/watches/review/trigger-all", c.httpTriggerAllReviewChecks) + api.GET("/watches/issue", c.httpListIssueWatches) + api.POST("/watches/issue", c.httpCreateIssueWatch) + api.PUT("/watches/issue/:id", c.httpUpdateIssueWatch) + api.DELETE("/watches/issue/:id", c.httpDeleteIssueWatch) + api.POST("/watches/issue/:id/trigger", c.httpTriggerIssueWatch) + api.POST("/watches/issue/trigger-all", c.httpTriggerAllIssueChecks) + api.GET("/orgs", c.httpListUserOrgs) api.GET("/repos/search", c.httpSearchRepos) api.GET("/repos/:owner/:repo/branches", c.httpListRepoBranches) @@ -365,3 +372,102 @@ func (c *Controller) httpGetStats(ctx *gin.Context) { } ctx.JSON(http.StatusOK, stats) } + +// --- Issue watch HTTP handlers --- + +func (c *Controller) httpListIssueWatches(ctx *gin.Context) { + workspaceID := ctx.Query("workspace_id") + if workspaceID == "" { + ctx.JSON(http.StatusBadRequest, gin.H{"error": "workspace_id query parameter required"}) + return + } + watches, err := c.service.ListIssueWatches(ctx.Request.Context(), workspaceID) + if err != nil { + ctx.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + ctx.JSON(http.StatusOK, gin.H{"watches": watches}) +} + +func (c *Controller) httpCreateIssueWatch(ctx *gin.Context) { + var req CreateIssueWatchRequest + if err := ctx.ShouldBindJSON(&req); err != nil { + ctx.JSON(http.StatusBadRequest, gin.H{"error": "invalid payload"}) + return + } + iw, err := c.service.CreateIssueWatch(ctx.Request.Context(), &req) + if err != nil { + ctx.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + ctx.JSON(http.StatusCreated, iw) +} + +func (c *Controller) httpUpdateIssueWatch(ctx *gin.Context) { + id := ctx.Param("id") + var req UpdateIssueWatchRequest + if err := ctx.ShouldBindJSON(&req); err != nil { + ctx.JSON(http.StatusBadRequest, gin.H{"error": "invalid payload"}) + return + } + if err := c.service.UpdateIssueWatch(ctx.Request.Context(), id, &req); err != nil { + ctx.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + iw, err := c.service.GetIssueWatch(ctx.Request.Context(), id) + if err != nil { + ctx.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + ctx.JSON(http.StatusOK, iw) +} + +func (c *Controller) httpDeleteIssueWatch(ctx *gin.Context) { + id := ctx.Param("id") + if err := c.service.DeleteIssueWatch(ctx.Request.Context(), id); err != nil { + ctx.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + ctx.JSON(http.StatusOK, gin.H{"deleted": true}) +} + +func (c *Controller) httpTriggerIssueWatch(ctx *gin.Context) { + id := ctx.Param("id") + watch, err := c.service.GetIssueWatch(ctx.Request.Context(), id) + if err != nil { + ctx.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + if watch == nil { + ctx.JSON(http.StatusNotFound, gin.H{"error": "issue watch not found"}) + return + } + newIssues, err := c.service.CheckIssueWatch(ctx.Request.Context(), watch) + if err != nil { + ctx.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + for _, issue := range newIssues { + c.service.publishNewIssueEvent(ctx.Request.Context(), watch, issue) + } + // Clean up tasks for closed issues that haven't been started. + cleaned, cleanErr := c.service.CleanupClosedIssueTasks(ctx.Request.Context(), watch) + if cleanErr != nil { + c.service.logger.Warn("cleanup closed issue tasks failed", zap.String("watch_id", id), zap.Error(cleanErr)) + } + ctx.JSON(http.StatusOK, gin.H{"new_issues_found": len(newIssues), "issues": newIssues, "cleaned": cleaned}) +} + +func (c *Controller) httpTriggerAllIssueChecks(ctx *gin.Context) { + workspaceID := ctx.Query("workspace_id") + if workspaceID == "" { + ctx.JSON(http.StatusBadRequest, gin.H{"error": "workspace_id query parameter required"}) + return + } + count, err := c.service.TriggerAllIssueChecks(ctx.Request.Context(), workspaceID) + if err != nil { + ctx.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + ctx.JSON(http.StatusOK, gin.H{"new_issues_found": count}) +} diff --git a/apps/backend/internal/github/controller_test.go b/apps/backend/internal/github/controller_test.go index 6939910197..6dff463892 100644 --- a/apps/backend/internal/github/controller_test.go +++ b/apps/backend/internal/github/controller_test.go @@ -66,6 +66,12 @@ func (s *stubClient) SubmitReview(context.Context, string, string, int, string, func (s *stubClient) ListRepoBranches(context.Context, string, string) ([]RepoBranch, error) { return nil, nil } +func (s *stubClient) ListIssues(context.Context, string, string) ([]*Issue, error) { + return nil, nil +} +func (s *stubClient) GetIssueState(context.Context, string, string, int) (string, error) { + return defaultPRState, nil +} func (s *stubClient) GetPRStatus(context.Context, string, string, int) (*PRStatus, error) { return nil, nil } diff --git a/apps/backend/internal/github/gh_client.go b/apps/backend/internal/github/gh_client.go index 1ef34d80a9..6cd071b333 100644 --- a/apps/backend/internal/github/gh_client.go +++ b/apps/backend/internal/github/gh_client.go @@ -171,6 +171,33 @@ func (c *GHClient) ListReviewRequestedPRs(ctx context.Context, scope, filter, cu return c.parseSearchResults(out) } +func (c *GHClient) ListIssues(ctx context.Context, filter, customQuery string) ([]*Issue, error) { + query := buildIssueSearchQuery(filter, customQuery) + out, err := c.run(ctx, "api", "search/issues", + "-X", "GET", + "-f", "q="+query, + "-f", "per_page=50", + "--jq", ".items") + if err != nil { + return nil, fmt.Errorf("list issues: %w", err) + } + var items []issueSearchItem + if err := json.Unmarshal([]byte(out), &items); err != nil { + return nil, fmt.Errorf("parse issue search results: %w", err) + } + return parseIssueSearchResults(items), nil +} + +func (c *GHClient) GetIssueState(ctx context.Context, owner, repo string, number int) (string, error) { + out, err := c.run(ctx, "api", + fmt.Sprintf("repos/%s/%s/issues/%d", owner, repo, number), + "--jq", ".state") + if err != nil { + return "", fmt.Errorf("get issue state: %w", err) + } + return strings.TrimSpace(out), nil +} + func (c *GHClient) ListUserOrgs(ctx context.Context) ([]GitHubOrg, error) { out, err := c.run(ctx, "api", "user/orgs", "--paginate") if err != nil { diff --git a/apps/backend/internal/github/handlers.go b/apps/backend/internal/github/handlers.go index 23badc6cb9..957cbace7e 100644 --- a/apps/backend/internal/github/handlers.go +++ b/apps/backend/internal/github/handlers.go @@ -11,6 +11,11 @@ import ( ws "github.com/kandev/kandev/pkg/websocket" ) +const ( + errMsgInvalidPayload = "invalid payload" + errMsgIDRequired = "id required" +) + // RegisterRoutes registers HTTP and WebSocket routes for GitHub integration. func RegisterRoutes(router *gin.Engine, dispatcher *ws.Dispatcher, svc *Service, log *logger.Logger) { ctrl := NewController(svc, log) @@ -47,6 +52,14 @@ func registerWSHandlers(dispatcher *ws.Dispatcher, svc *Service, log *logger.Log dispatcher.RegisterFunc(ws.ActionGitHubPRCommitsGet, wsGetPRCommits(svc, log)) dispatcher.RegisterFunc(ws.ActionGitHubTaskPRSync, wsSyncTaskPR(svc, log)) dispatcher.RegisterFunc(ws.ActionGitHubStats, wsGetStats(svc, log)) + + // Issue watch handlers + dispatcher.RegisterFunc(ws.ActionGitHubIssueWatchesList, wsListIssueWatches(svc, log)) + dispatcher.RegisterFunc(ws.ActionGitHubIssueWatchCreate, wsCreateIssueWatch(svc, log)) + dispatcher.RegisterFunc(ws.ActionGitHubIssueWatchUpdate, wsUpdateIssueWatch(svc, log)) + dispatcher.RegisterFunc(ws.ActionGitHubIssueWatchDelete, wsDeleteIssueWatch(svc, log)) + dispatcher.RegisterFunc(ws.ActionGitHubIssueTrigger, wsTriggerIssueWatch(svc, log)) + dispatcher.RegisterFunc(ws.ActionGitHubIssueTriggerAll, wsTriggerAllIssueChecks(svc, log)) } // parseMap parses the WS message payload into a map for simple field lookups. @@ -90,7 +103,7 @@ func wsDeleteByID(deleteFn func(ctx context.Context, id string) error) func(ctx } id, _ := payload["id"].(string) if id == "" { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "id required", nil) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, errMsgIDRequired, nil) } if err := deleteFn(ctx, id); err != nil { return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, err.Error(), nil) @@ -99,6 +112,54 @@ func wsDeleteByID(deleteFn func(ctx context.Context, id string) error) func(ctx } } +// wsUpdateByPayload builds a WS handler that parses an ID + typed request from +// the payload and delegates to updateFn. +func wsUpdateByPayload[T any](updateFn func(ctx context.Context, id string, req *T) error) func(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + return func(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + idHolder := struct { + ID string `json:"id"` + }{} + if err := msg.ParsePayload(&idHolder); err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, errMsgInvalidPayload, nil) + } + if idHolder.ID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, errMsgIDRequired, nil) + } + var req T + if err := msg.ParsePayload(&req); err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, errMsgInvalidPayload, nil) + } + if err := updateFn(ctx, idHolder.ID, &req); err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, err.Error(), nil) + } + return ws.NewResponse(msg.ID, msg.Action, map[string]bool{"updated": true}) + } +} + +// wsTriggerAllByWorkspace builds a WS handler that extracts workspace_id from +// the payload, calls the given trigger function, and returns the count under the +// specified response key. +func wsTriggerAllByWorkspace( + responseKey string, + triggerFn func(ctx context.Context, workspaceID string) (int, error), +) func(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + return func(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + payload, parseErr := parseMap(msg) + if parseErr != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "invalid payload: "+parseErr.Error(), nil) + } + workspaceID, _ := payload["workspace_id"].(string) + if workspaceID == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "workspace_id required", nil) + } + count, err := triggerFn(ctx, workspaceID) + if err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, err.Error(), nil) + } + return ws.NewResponse(msg.ID, msg.Action, map[string]int{responseKey: count}) + } +} + func wsStatus(svc *Service, _ *logger.Logger) func(ctx context.Context, msg *ws.Message) (*ws.Message, error) { return func(ctx context.Context, msg *ws.Message) (*ws.Message, error) { status, err := svc.GetStatus(ctx) @@ -188,7 +249,7 @@ func wsCreateReviewWatch(svc *Service, _ *logger.Logger) func(ctx context.Contex return func(ctx context.Context, msg *ws.Message) (*ws.Message, error) { var req CreateReviewWatchRequest if err := msg.ParsePayload(&req); err != nil { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "invalid payload", nil) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, errMsgInvalidPayload, nil) } rw, err := svc.CreateReviewWatch(ctx, &req) if err != nil { @@ -199,23 +260,9 @@ func wsCreateReviewWatch(svc *Service, _ *logger.Logger) func(ctx context.Contex } func wsUpdateReviewWatch(svc *Service, _ *logger.Logger) func(ctx context.Context, msg *ws.Message) (*ws.Message, error) { - return func(ctx context.Context, msg *ws.Message) (*ws.Message, error) { - var payload struct { - ID string `json:"id"` - UpdateReviewWatchRequest - } - if err := msg.ParsePayload(&payload); err != nil { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "invalid payload", nil) - } - if payload.ID == "" { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "id required", nil) - } - req := payload.UpdateReviewWatchRequest - if err := svc.UpdateReviewWatch(ctx, payload.ID, &req); err != nil { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, err.Error(), nil) - } - return ws.NewResponse(msg.ID, msg.Action, map[string]bool{"updated": true}) - } + return wsUpdateByPayload(func(ctx context.Context, id string, req *UpdateReviewWatchRequest) error { + return svc.UpdateReviewWatch(ctx, id, req) + }) } func wsDeleteReviewWatch(svc *Service, _ *logger.Logger) func(ctx context.Context, msg *ws.Message) (*ws.Message, error) { @@ -230,7 +277,7 @@ func wsTriggerReviewWatch(svc *Service, _ *logger.Logger) func(ctx context.Conte } id, _ := payload["id"].(string) if id == "" { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "id required", nil) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, errMsgIDRequired, nil) } watch, err := svc.GetReviewWatch(ctx, id) if err != nil { @@ -248,21 +295,7 @@ func wsTriggerReviewWatch(svc *Service, _ *logger.Logger) func(ctx context.Conte } func wsTriggerAllReviewChecks(svc *Service, _ *logger.Logger) func(ctx context.Context, msg *ws.Message) (*ws.Message, error) { - return func(ctx context.Context, msg *ws.Message) (*ws.Message, error) { - payload, parseErr := parseMap(msg) - if parseErr != nil { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "invalid payload: "+parseErr.Error(), nil) - } - workspaceID, _ := payload["workspace_id"].(string) - if workspaceID == "" { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "workspace_id required", nil) - } - count, err := svc.TriggerAllReviewChecks(ctx, workspaceID) - if err != nil { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, err.Error(), nil) - } - return ws.NewResponse(msg.ID, msg.Action, map[string]int{"new_prs_found": count}) - } + return wsTriggerAllByWorkspace("new_prs_found", svc.TriggerAllReviewChecks) } func wsListPRWatches(svc *Service, _ *logger.Logger) func(ctx context.Context, msg *ws.Message) (*ws.Message, error) { @@ -283,7 +316,7 @@ func wsGetStats(svc *Service, _ *logger.Logger) func(ctx context.Context, msg *w return func(ctx context.Context, msg *ws.Message) (*ws.Message, error) { var req PRStatsRequest if err := msg.ParsePayload(&req); err != nil { - return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "invalid payload", nil) + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, errMsgInvalidPayload, nil) } stats, err := svc.GetPRStats(ctx, &req) if err != nil { @@ -298,7 +331,7 @@ func wsGetStats(svc *Service, _ *logger.Logger) func(ctx context.Context, msg *w func parsePRParams(msg *ws.Message) (string, string, int, *ws.Message) { payload, parseErr := parseMap(msg) if parseErr != nil { - resp, _ := ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "invalid payload", nil) + resp, _ := ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, errMsgInvalidPayload, nil) return "", "", 0, resp } owner, _ := payload["owner"].(string) @@ -339,3 +372,71 @@ func wsGetPRCommits(svc *Service, _ *logger.Logger) func(ctx context.Context, ms return ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{"commits": commits}) } } + +// --- Issue watch WS handlers --- + +func wsListIssueWatches(svc *Service, _ *logger.Logger) func(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + return wsWithField("workspace_id", func(ctx context.Context, workspaceID string) (interface{}, error) { + return svc.ListIssueWatches(ctx, workspaceID) + }) +} + +func wsCreateIssueWatch(svc *Service, _ *logger.Logger) func(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + return func(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + var req CreateIssueWatchRequest + if err := msg.ParsePayload(&req); err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, errMsgInvalidPayload, nil) + } + iw, err := svc.CreateIssueWatch(ctx, &req) + if err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, err.Error(), nil) + } + return ws.NewResponse(msg.ID, msg.Action, iw) + } +} + +func wsUpdateIssueWatch(svc *Service, _ *logger.Logger) func(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + return wsUpdateByPayload(func(ctx context.Context, id string, req *UpdateIssueWatchRequest) error { + return svc.UpdateIssueWatch(ctx, id, req) + }) +} + +func wsDeleteIssueWatch(svc *Service, _ *logger.Logger) func(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + return wsDeleteByID(svc.DeleteIssueWatch) +} + +func wsTriggerIssueWatch(svc *Service, log *logger.Logger) func(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + return func(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + payload, parseErr := parseMap(msg) + if parseErr != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, "invalid payload: "+parseErr.Error(), nil) + } + id, _ := payload["id"].(string) + if id == "" { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeBadRequest, errMsgIDRequired, nil) + } + watch, err := svc.GetIssueWatch(ctx, id) + if err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, err.Error(), nil) + } + if watch == nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeNotFound, "issue watch not found", nil) + } + newIssues, err := svc.CheckIssueWatch(ctx, watch) + if err != nil { + return ws.NewError(msg.ID, msg.Action, ws.ErrorCodeInternalError, err.Error(), nil) + } + for _, issue := range newIssues { + svc.publishNewIssueEvent(ctx, watch, issue) + } + cleaned, cleanErr := svc.CleanupClosedIssueTasks(ctx, watch) + if cleanErr != nil { + log.Warn("cleanup closed issue tasks failed", zap.String("watch_id", id), zap.Error(cleanErr)) + } + return ws.NewResponse(msg.ID, msg.Action, map[string]interface{}{"new_issues_found": len(newIssues), "issues": newIssues, "cleaned": cleaned}) + } +} + +func wsTriggerAllIssueChecks(svc *Service, _ *logger.Logger) func(ctx context.Context, msg *ws.Message) (*ws.Message, error) { + return wsTriggerAllByWorkspace("new_issues_found", svc.TriggerAllIssueChecks) +} diff --git a/apps/backend/internal/github/mock_client.go b/apps/backend/internal/github/mock_client.go index d2ef7a2517..58053c81e5 100644 --- a/apps/backend/internal/github/mock_client.go +++ b/apps/backend/internal/github/mock_client.go @@ -133,6 +133,14 @@ func (m *MockClient) ListReviewRequestedPRs(context.Context, string, string, str return result, nil } +func (m *MockClient) ListIssues(context.Context, string, string) ([]*Issue, error) { + return nil, nil +} + +func (m *MockClient) GetIssueState(context.Context, string, string, int) (string, error) { + return defaultPRState, nil +} + func (m *MockClient) ListUserOrgs(context.Context) ([]GitHubOrg, error) { m.mu.RLock() defer m.mu.RUnlock() diff --git a/apps/backend/internal/github/models.go b/apps/backend/internal/github/models.go index 623384ec16..f6f0ee0b70 100644 --- a/apps/backend/internal/github/models.go +++ b/apps/backend/internal/github/models.go @@ -317,3 +317,97 @@ type DailyCount struct { Date string `json:"date"` Count int `json:"count"` } + +// --- Issue Watch models --- + +// Issue represents a GitHub Issue (not a PR). +type Issue struct { + Number int `json:"number"` + Title string `json:"title"` + Body string `json:"body"` + URL string `json:"url"` + HTMLURL string `json:"html_url"` + State string `json:"state"` // open, closed + AuthorLogin string `json:"author_login"` + RepoOwner string `json:"repo_owner"` + RepoName string `json:"repo_name"` + Labels []string `json:"labels"` + Assignees []string `json:"assignees"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` + ClosedAt *time.Time `json:"closed_at,omitempty"` +} + +// IssueWatch configures periodic polling for GitHub issues matching a query. +// Repos holds the list of repositories to monitor. An empty list means all repos. +type IssueWatch struct { + ID string `json:"id" db:"id"` + WorkspaceID string `json:"workspace_id" db:"workspace_id"` + WorkflowID string `json:"workflow_id" db:"workflow_id"` + WorkflowStepID string `json:"workflow_step_id" db:"workflow_step_id"` + Repos []RepoFilter `json:"repos" db:"-"` + ReposJSON string `json:"-" db:"repos"` + AgentProfileID string `json:"agent_profile_id" db:"agent_profile_id"` + ExecutorProfileID string `json:"executor_profile_id" db:"executor_profile_id"` + Prompt string `json:"prompt" db:"prompt"` + Labels []string `json:"labels" db:"-"` + LabelsJSON string `json:"-" db:"labels"` + CustomQuery string `json:"custom_query" db:"custom_query"` + Enabled bool `json:"enabled" db:"enabled"` + PollIntervalSeconds int `json:"poll_interval_seconds" db:"poll_interval_seconds"` + LastPolledAt *time.Time `json:"last_polled_at,omitempty" db:"last_polled_at"` + CreatedAt time.Time `json:"created_at" db:"created_at"` + UpdatedAt time.Time `json:"updated_at" db:"updated_at"` +} + +// IssueWatchTask records which issues have already had tasks created (deduplication). +type IssueWatchTask struct { + ID string `json:"id" db:"id"` + IssueWatchID string `json:"issue_watch_id" db:"issue_watch_id"` + RepoOwner string `json:"repo_owner" db:"repo_owner"` + RepoName string `json:"repo_name" db:"repo_name"` + IssueNumber int `json:"issue_number" db:"issue_number"` + IssueURL string `json:"issue_url" db:"issue_url"` + TaskID string `json:"task_id" db:"task_id"` + CreatedAt time.Time `json:"created_at" db:"created_at"` +} + +// NewIssueEvent is published when a new issue matching a watch is found. +type NewIssueEvent struct { + IssueWatchID string `json:"issue_watch_id"` + WorkspaceID string `json:"workspace_id"` + WorkflowID string `json:"workflow_id"` + WorkflowStepID string `json:"workflow_step_id"` + AgentProfileID string `json:"agent_profile_id"` + ExecutorProfileID string `json:"executor_profile_id"` + Prompt string `json:"prompt"` + Issue *Issue `json:"issue"` +} + +// CreateIssueWatchRequest is the request body for creating an issue watch. +type CreateIssueWatchRequest struct { + WorkspaceID string `json:"workspace_id"` + WorkflowID string `json:"workflow_id"` + WorkflowStepID string `json:"workflow_step_id"` + Repos []RepoFilter `json:"repos"` + AgentProfileID string `json:"agent_profile_id"` + ExecutorProfileID string `json:"executor_profile_id"` + Prompt string `json:"prompt"` + Labels []string `json:"labels"` + CustomQuery string `json:"custom_query"` + PollIntervalSeconds int `json:"poll_interval_seconds"` +} + +// UpdateIssueWatchRequest is the request body for updating an issue watch. +type UpdateIssueWatchRequest struct { + WorkflowID *string `json:"workflow_id,omitempty"` + WorkflowStepID *string `json:"workflow_step_id,omitempty"` + Repos *[]RepoFilter `json:"repos,omitempty"` + AgentProfileID *string `json:"agent_profile_id,omitempty"` + ExecutorProfileID *string `json:"executor_profile_id,omitempty"` + Prompt *string `json:"prompt,omitempty"` + Labels *[]string `json:"labels,omitempty"` + CustomQuery *string `json:"custom_query,omitempty"` + Enabled *bool `json:"enabled,omitempty"` + PollIntervalSeconds *int `json:"poll_interval_seconds,omitempty"` +} diff --git a/apps/backend/internal/github/noop_client.go b/apps/backend/internal/github/noop_client.go index 1184184fe5..e5cc67ae2e 100644 --- a/apps/backend/internal/github/noop_client.go +++ b/apps/backend/internal/github/noop_client.go @@ -80,3 +80,11 @@ func (c *NoopClient) ListRepoBranches(context.Context, string, string) ([]RepoBr func (c *NoopClient) SubmitReview(context.Context, string, string, int, string, string) error { return ErrNoClient } + +func (c *NoopClient) ListIssues(context.Context, string, string) ([]*Issue, error) { + return nil, ErrNoClient +} + +func (c *NoopClient) GetIssueState(context.Context, string, string, int) (string, error) { + return "", ErrNoClient +} diff --git a/apps/backend/internal/github/pat_client.go b/apps/backend/internal/github/pat_client.go index 6188478307..62f137b976 100644 --- a/apps/backend/internal/github/pat_client.go +++ b/apps/backend/internal/github/pat_client.go @@ -133,6 +133,29 @@ func (c *PATClient) ListReviewRequestedPRs(ctx context.Context, scope, filter, c return prs, nil } +func (c *PATClient) ListIssues(ctx context.Context, filter, customQuery string) ([]*Issue, error) { + query := buildIssueSearchQuery(filter, customQuery) + var result struct { + Items []issueSearchItem `json:"items"` + } + endpoint := "/search/issues?q=" + url.QueryEscape(query) + "&per_page=50" + if err := c.get(ctx, endpoint, &result); err != nil { + return nil, fmt.Errorf("search issues: %w", err) + } + return parseIssueSearchResults(result.Items), nil +} + +func (c *PATClient) GetIssueState(ctx context.Context, owner, repo string, number int) (string, error) { + var result struct { + State string `json:"state"` + } + endpoint := fmt.Sprintf("/repos/%s/%s/issues/%d", owner, repo, number) + if err := c.get(ctx, endpoint, &result); err != nil { + return "", fmt.Errorf("get issue state: %w", err) + } + return result.State, nil +} + func (c *PATClient) ListUserOrgs(ctx context.Context) ([]GitHubOrg, error) { var raw []struct { Login string `json:"login"` diff --git a/apps/backend/internal/github/poller.go b/apps/backend/internal/github/poller.go index d23149c427..6d7a72cb8a 100644 --- a/apps/backend/internal/github/poller.go +++ b/apps/backend/internal/github/poller.go @@ -15,6 +15,7 @@ import ( const ( defaultPRPollInterval = 30 * time.Second defaultReviewPollInterval = 5 * time.Minute + defaultIssuePollInterval = 5 * time.Minute ) // TaskBranchInfo describes a task+session that may need a PR watch. @@ -64,9 +65,10 @@ func (p *Poller) Start(ctx context.Context) { p.started = true ctx, p.cancel = context.WithCancel(ctx) - p.wg.Add(2) //nolint:mnd + p.wg.Add(3) //nolint:mnd go p.prMonitorLoop(ctx) go p.reviewQueueLoop(ctx) + go p.issueWatchLoop(ctx) p.logger.Info("GitHub poller started") } @@ -358,3 +360,68 @@ func (p *Poller) checkReviewWatches(ctx context.Context) { } } } + +// issueWatchLoop polls issue watches for new GitHub issues. +func (p *Poller) issueWatchLoop(ctx context.Context) { + defer p.wg.Done() + + // Run an initial check immediately so existing watches are evaluated on startup. + p.checkIssueWatches(ctx) + + ticker := time.NewTicker(defaultIssuePollInterval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + p.checkIssueWatches(ctx) + } + } +} + +func (p *Poller) checkIssueWatches(ctx context.Context) { + watches, err := p.service.store.ListEnabledIssueWatches(ctx) + if err != nil { + p.logger.Error("failed to list issue watches", zap.Error(err)) + return + } + if len(watches) == 0 { + return + } + p.logger.Debug("checking issue watches", zap.Int("count", len(watches))) + for _, watch := range watches { + p.logger.Debug("polling issue watch", + zap.String("watch_id", watch.ID), + zap.String("workspace_id", watch.WorkspaceID), + zap.String("custom_query", watch.CustomQuery), + zap.Int("repo_filters", len(watch.Repos))) + + newIssues, err := p.service.CheckIssueWatch(ctx, watch) + if err != nil { + p.logger.Debug("failed to check issue watch", + zap.String("id", watch.ID), zap.Error(err)) + continue + } + p.logger.Debug("issue watch checked", + zap.String("watch_id", watch.ID), + zap.Int("new_issues", len(newIssues))) + for _, issue := range newIssues { + p.logger.Info("new issue found for watch", + zap.String("watch_id", watch.ID), + zap.String("repo", issue.RepoOwner+"/"+issue.RepoName), + zap.Int("issue_number", issue.Number), + zap.String("title", issue.Title)) + p.service.publishNewIssueEvent(ctx, watch, issue) + } + // Clean up tasks for closed issues that the user hasn't opened. + if cleaned, err := p.service.CleanupClosedIssueTasks(ctx, watch); err != nil { + p.logger.Warn("failed to cleanup closed issue tasks", + zap.String("watch_id", watch.ID), zap.Error(err)) + } else if cleaned > 0 { + p.logger.Info("cleaned up closed issue tasks", + zap.String("watch_id", watch.ID), zap.Int("deleted", cleaned)) + } + } +} diff --git a/apps/backend/internal/github/service.go b/apps/backend/internal/github/service.go index aa966e5a3d..b6a9e38682 100644 --- a/apps/backend/internal/github/service.go +++ b/apps/backend/internal/github/service.go @@ -1149,6 +1149,389 @@ func (s *Service) publishNewReviewPREvent(ctx context.Context, watch *ReviewWatc } } +// --- Issue Watch service methods --- + +// CreateIssueWatch creates a new issue watch and triggers an initial poll. +func (s *Service) CreateIssueWatch(ctx context.Context, req *CreateIssueWatchRequest) (*IssueWatch, error) { + if req.PollIntervalSeconds <= 0 { + req.PollIntervalSeconds = defaultWatchPollIntervalSec + } + if req.PollIntervalSeconds < minWatchPollIntervalSec { + req.PollIntervalSeconds = minWatchPollIntervalSec + } + repos := req.Repos + if repos == nil { + repos = []RepoFilter{} + } + labels := req.Labels + if labels == nil { + labels = []string{} + } + iw := &IssueWatch{ + WorkspaceID: req.WorkspaceID, + WorkflowID: req.WorkflowID, + WorkflowStepID: req.WorkflowStepID, + Repos: repos, + AgentProfileID: req.AgentProfileID, + ExecutorProfileID: req.ExecutorProfileID, + Prompt: req.Prompt, + Labels: labels, + CustomQuery: req.CustomQuery, + Enabled: true, + PollIntervalSeconds: req.PollIntervalSeconds, + } + if err := s.store.CreateIssueWatch(ctx, iw); err != nil { + return nil, fmt.Errorf("create issue watch: %w", err) + } + go s.initialIssueCheck(context.Background(), iw) + return iw, nil +} + +// initialIssueCheck runs a single poll for a newly created issue watch. +func (s *Service) initialIssueCheck(ctx context.Context, watch *IssueWatch) { + newIssues, err := s.CheckIssueWatch(ctx, watch) + if err != nil { + s.logger.Debug("initial issue check failed", + zap.String("watch_id", watch.ID), zap.Error(err)) + return + } + for _, issue := range newIssues { + s.publishNewIssueEvent(ctx, watch, issue) + } + if len(newIssues) > 0 { + s.logger.Info("initial issue check found issues", + zap.String("watch_id", watch.ID), + zap.Int("new_issues", len(newIssues))) + } +} + +// GetIssueWatch returns a single issue watch by ID. +func (s *Service) GetIssueWatch(ctx context.Context, id string) (*IssueWatch, error) { + return s.store.GetIssueWatch(ctx, id) +} + +// ListIssueWatches returns all issue watches for a workspace. +func (s *Service) ListIssueWatches(ctx context.Context, workspaceID string) ([]*IssueWatch, error) { + return s.store.ListIssueWatches(ctx, workspaceID) +} + +// UpdateIssueWatch updates an issue watch. +// +//nolint:dupl // mirrors UpdateReviewWatch — different types, same structure +func (s *Service) UpdateIssueWatch(ctx context.Context, id string, req *UpdateIssueWatchRequest) error { + iw, err := s.store.GetIssueWatch(ctx, id) + if err != nil { + return err + } + if iw == nil { + return fmt.Errorf("issue watch not found: %s", id) + } + if req.WorkflowID != nil { + iw.WorkflowID = *req.WorkflowID + } + if req.WorkflowStepID != nil { + iw.WorkflowStepID = *req.WorkflowStepID + } + if req.Repos != nil { + iw.Repos = *req.Repos + } + if req.AgentProfileID != nil { + iw.AgentProfileID = *req.AgentProfileID + } + if req.ExecutorProfileID != nil { + iw.ExecutorProfileID = *req.ExecutorProfileID + } + if req.Prompt != nil { + iw.Prompt = *req.Prompt + } + if req.Labels != nil { + iw.Labels = *req.Labels + } + if req.CustomQuery != nil { + iw.CustomQuery = *req.CustomQuery + } + if req.Enabled != nil { + iw.Enabled = *req.Enabled + } + if req.PollIntervalSeconds != nil { + v := *req.PollIntervalSeconds + if v <= 0 { + v = defaultWatchPollIntervalSec + } + if v < minWatchPollIntervalSec { + v = minWatchPollIntervalSec + } + iw.PollIntervalSeconds = v + } + return s.store.UpdateIssueWatch(ctx, iw) +} + +// DeleteIssueWatch deletes an issue watch. +func (s *Service) DeleteIssueWatch(ctx context.Context, id string) error { + return s.store.DeleteIssueWatch(ctx, id) +} + +// CheckIssueWatch checks for new issues matching the watch and returns ones not yet tracked. +func (s *Service) CheckIssueWatch(ctx context.Context, watch *IssueWatch) ([]*Issue, error) { + if s.client == nil { + return nil, fmt.Errorf("github client not available") + } + + s.logger.Debug("checking issue watch for new issues", + zap.String("watch_id", watch.ID), + zap.Int("repo_filters", len(watch.Repos)), + zap.String("custom_query", watch.CustomQuery), + zap.Bool("enabled", watch.Enabled)) + + issues, err := s.fetchIssues(ctx, watch) + if err != nil { + return nil, err + } + + var newIssues []*Issue + for _, issue := range issues { + exists, checkErr := s.store.HasIssueWatchTask(ctx, watch.ID, issue.RepoOwner, issue.RepoName, issue.Number) + if checkErr != nil { + s.logger.Error("failed to check issue watch task", zap.Error(checkErr)) + continue + } + if !exists { + newIssues = append(newIssues, issue) + } + } + + now := time.Now().UTC() + watch.LastPolledAt = &now + _ = s.store.UpdateIssueWatch(ctx, watch) + + return newIssues, nil +} + +// fetchIssues fetches issues based on the watch configuration. +func (s *Service) fetchIssues(ctx context.Context, watch *IssueWatch) ([]*Issue, error) { + hasRepos := len(watch.Repos) > 0 + + if !hasRepos { + filter := s.buildIssueFilter(watch) + return s.client.ListIssues(ctx, filter, watch.CustomQuery) + } + + return s.fetchIssuesWithRepoFilter(ctx, watch), nil +} + +// buildIssueFilter builds the filter qualifier from watch labels. +func (s *Service) buildIssueFilter(watch *IssueWatch) string { + var parts []string + for _, label := range watch.Labels { + if strings.ContainsRune(label, ' ') { + parts = append(parts, `label:"`+label+`"`) + } else { + parts = append(parts, "label:"+label) + } + } + return strings.Join(parts, " ") +} + +// fetchIssuesWithRepoFilter queries each repo individually and deduplicates. +func (s *Service) fetchIssuesWithRepoFilter(ctx context.Context, watch *IssueWatch) []*Issue { + var allIssues []*Issue + seen := make(map[string]bool) + + labelFilter := s.buildIssueFilter(watch) + + for _, repo := range watch.Repos { + qualifier := repoFilterToQualifier(repo) + filter := qualifier + if labelFilter != "" { + filter += " " + labelFilter + } + + var issues []*Issue + var err error + if watch.CustomQuery != "" { + query := watch.CustomQuery + " " + qualifier + issues, err = s.client.ListIssues(ctx, "", query) + } else { + issues, err = s.client.ListIssues(ctx, filter, "") + } + if err != nil { + if isConnectivityError(err) { + s.logger.Warn("failed to list issues (connectivity)", + zap.String("filter", qualifier), zap.Error(err)) + } else { + s.logger.Error("failed to list issues", + zap.String("filter", qualifier), zap.Error(err)) + } + continue + } + + for _, issue := range issues { + key := fmt.Sprintf("%s/%s#%d", issue.RepoOwner, issue.RepoName, issue.Number) + if !seen[key] { + seen[key] = true + allIssues = append(allIssues, issue) + } + } + } + return allIssues +} + +// ReserveIssueWatchTask atomically claims a dedup slot. +func (s *Service) ReserveIssueWatchTask(ctx context.Context, watchID, repoOwner, repoName string, issueNumber int, issueURL string) (bool, error) { + return s.store.ReserveIssueWatchTask(ctx, watchID, repoOwner, repoName, issueNumber, issueURL) +} + +// AssignIssueWatchTaskID attaches a task ID to a previously reserved slot. +func (s *Service) AssignIssueWatchTaskID(ctx context.Context, watchID, repoOwner, repoName string, issueNumber int, taskID string) error { + return s.store.AssignIssueWatchTaskID(ctx, watchID, repoOwner, repoName, issueNumber, taskID) +} + +// ReleaseIssueWatchTask removes a reservation when task creation fails. +func (s *Service) ReleaseIssueWatchTask(ctx context.Context, watchID, repoOwner, repoName string, issueNumber int) error { + return s.store.ReleaseIssueWatchTask(ctx, watchID, repoOwner, repoName, issueNumber) +} + +// TriggerAllIssueChecks triggers all issue watches for a workspace. +// +//nolint:dupl // mirrors TriggerAllReviewChecks — different types, same structure +func (s *Service) TriggerAllIssueChecks(ctx context.Context, workspaceID string) (int, error) { + watches, err := s.store.ListIssueWatches(ctx, workspaceID) + if err != nil { + return 0, err + } + enabled := 0 + for _, w := range watches { + if w.Enabled { + enabled++ + } + } + s.logger.Info("triggering issue checks", + zap.String("workspace_id", workspaceID), + zap.Int("total_watches", len(watches)), + zap.Int("enabled_watches", enabled)) + + totalNew := 0 + for _, watch := range watches { + if !watch.Enabled { + continue + } + newIssues, checkErr := s.CheckIssueWatch(ctx, watch) + if checkErr != nil { + s.logger.Error("failed to check issue watch", + zap.String("watch_id", watch.ID), zap.Error(checkErr)) + continue + } + for _, issue := range newIssues { + s.publishNewIssueEvent(ctx, watch, issue) + } + totalNew += len(newIssues) + if _, cleanErr := s.CleanupClosedIssueTasks(ctx, watch); cleanErr != nil { + s.logger.Warn("cleanup closed issue tasks failed", + zap.String("watch_id", watch.ID), zap.Error(cleanErr)) + } + } + s.logger.Info("issue checks completed", + zap.String("workspace_id", workspaceID), + zap.Int("new_issues_found", totalNew)) + return totalNew, nil +} + +// CleanupClosedIssueTasks checks issues tracked by a watch and deletes +// tasks whose issues are closed and the user hasn't interacted with. +// +//nolint:dupl // mirrors CleanupMergedReviewTasks — different types, same structure +func (s *Service) CleanupClosedIssueTasks(ctx context.Context, watch *IssueWatch) (int, error) { + if s.client == nil || s.taskDeleter == nil || s.taskSessionChecker == nil { + return 0, nil + } + issueTasks, err := s.store.ListIssueWatchTasksByWatch(ctx, watch.ID) + if err != nil { + return 0, fmt.Errorf("list issue watch tasks: %w", err) + } + deleted := 0 + for _, it := range issueTasks { + // Orphan reservation: task_id is empty because the process was + // killed between Reserve and Assign. Clean up once the issue is closed. + if it.TaskID == "" { + if should, _ := s.shouldDeleteIssueTask(ctx, it); should { + _ = s.store.DeleteIssueWatchTask(ctx, it.ID) + deleted++ + } + continue + } + shouldDelete, reason := s.shouldDeleteIssueTask(ctx, it) + if !shouldDelete { + continue + } + if err := s.taskDeleter.DeleteTask(ctx, it.TaskID); err != nil { + if strings.Contains(err.Error(), "not found") { + _ = s.store.DeleteIssueWatchTask(ctx, it.ID) + deleted++ + continue + } + s.logger.Warn("failed to delete issue task", + zap.String("task_id", it.TaskID), zap.Error(err)) + continue + } + _ = s.store.DeleteIssueWatchTask(ctx, it.ID) + s.logger.Info("deleted issue task", + zap.String("task_id", it.TaskID), + zap.String("reason", reason), + zap.Int("issue_number", it.IssueNumber), + zap.String("repo", it.RepoOwner+"/"+it.RepoName)) + deleted++ + } + return deleted, nil +} + +// shouldDeleteIssueTask checks if an issue task should be cleaned up. +// Returns true + reason if the issue is closed AND the user hasn't interacted +// with the task yet (no sessions). +func (s *Service) shouldDeleteIssueTask(ctx context.Context, it *IssueWatchTask) (bool, string) { + state, err := s.client.GetIssueState(ctx, it.RepoOwner, it.RepoName, it.IssueNumber) + if err != nil { + s.logger.Debug("failed to fetch issue state for cleanup", + zap.Int("issue_number", it.IssueNumber), zap.Error(err)) + return false, "" + } + if state != "closed" { + return false, "" + } + reason := "issue_closed" + // Don't delete tasks the user has interacted with (has sessions). + if it.TaskID != "" { + hasSessions, err := s.taskSessionChecker.HasTaskSessions(ctx, it.TaskID) + if err != nil { + s.logger.Debug("failed to check task sessions", + zap.String("task_id", it.TaskID), zap.Error(err)) + return false, "" + } + if hasSessions { + return false, "" + } + } + return true, reason +} + +func (s *Service) publishNewIssueEvent(ctx context.Context, watch *IssueWatch, issue *Issue) { + if s.eventBus == nil { + return + } + event := bus.NewEvent(events.GitHubNewIssue, "github", &NewIssueEvent{ + IssueWatchID: watch.ID, + WorkspaceID: watch.WorkspaceID, + WorkflowID: watch.WorkflowID, + WorkflowStepID: watch.WorkflowStepID, + AgentProfileID: watch.AgentProfileID, + ExecutorProfileID: watch.ExecutorProfileID, + Prompt: watch.Prompt, + Issue: issue, + }) + if err := s.eventBus.Publish(ctx, events.GitHubNewIssue, event); err != nil { + s.logger.Debug("failed to publish new issue event", zap.Error(err)) + } +} + func findLatestCommentTime(comments []PRComment) *time.Time { var latest *time.Time for _, c := range comments { diff --git a/apps/backend/internal/github/service_issue_test.go b/apps/backend/internal/github/service_issue_test.go new file mode 100644 index 0000000000..fd1b74de6d --- /dev/null +++ b/apps/backend/internal/github/service_issue_test.go @@ -0,0 +1,91 @@ +package github + +import ( + "context" + "fmt" + "testing" + + "github.com/kandev/kandev/internal/common/logger" +) + +// --- shouldDeleteIssueTask tests --- + +// issueStateClient is a minimal Client stub for shouldDeleteIssueTask tests. +type issueStateClient struct { + NoopClient + state string + err error +} + +func (c *issueStateClient) GetIssueState(_ context.Context, _, _ string, _ int) (string, error) { + return c.state, c.err +} + +// stubSessionChecker implements TaskSessionChecker for tests. +type stubSessionChecker struct { + has bool + err error +} + +func (s *stubSessionChecker) HasTaskSessions(_ context.Context, _ string) (bool, error) { + return s.has, s.err +} + +func TestShouldDeleteIssueTask(t *testing.T) { + log, _ := logger.NewLogger(logger.LoggingConfig{Level: "error", Format: "console"}) + + task := &IssueWatchTask{ + ID: "iwt-1", + IssueWatchID: "iw-1", + RepoOwner: "acme", + RepoName: "widget", + IssueNumber: 7, + TaskID: "task-1", + } + + t.Run("open issue returns false", func(t *testing.T) { + svc := &Service{client: &issueStateClient{state: "open"}, logger: log} + del, _ := svc.shouldDeleteIssueTask(context.Background(), task) + if del { + t.Error("expected false for open issue") + } + }) + + t.Run("closed issue without sessions returns true", func(t *testing.T) { + svc := &Service{ + client: &issueStateClient{state: "closed"}, + logger: log, + taskSessionChecker: &stubSessionChecker{has: false}, + } + del, reason := svc.shouldDeleteIssueTask(context.Background(), task) + if !del { + t.Error("expected true for closed issue without sessions") + } + if reason != "issue_closed" { + t.Errorf("reason = %q, want %q", reason, "issue_closed") + } + }) + + t.Run("closed issue with sessions returns false", func(t *testing.T) { + svc := &Service{ + client: &issueStateClient{state: "closed"}, + logger: log, + taskSessionChecker: &stubSessionChecker{has: true}, + } + del, _ := svc.shouldDeleteIssueTask(context.Background(), task) + if del { + t.Error("expected false for closed issue with active sessions") + } + }) + + t.Run("API error returns false", func(t *testing.T) { + svc := &Service{ + client: &issueStateClient{err: fmt.Errorf("api error")}, + logger: log, + } + del, _ := svc.shouldDeleteIssueTask(context.Background(), task) + if del { + t.Error("expected false when API returns error") + } + }) +} diff --git a/apps/backend/internal/github/service_pr_sync_test.go b/apps/backend/internal/github/service_pr_sync_test.go new file mode 100644 index 0000000000..d1be09f36e --- /dev/null +++ b/apps/backend/internal/github/service_pr_sync_test.go @@ -0,0 +1,110 @@ +package github + +import ( + "context" + "testing" +) + +func TestTriggerPRSync_SyncsExistingWatch(t *testing.T) { + _, svc, mockClient, store := setupPollerTest(t) + ctx := context.Background() + + // Set up a PR in mock client. + mockClient.AddPR(&PR{ + Number: 10, + Title: "Feature PR", + State: "open", + HeadSHA: "abc123", + HeadBranch: "feat/x", + RepoOwner: "org", + RepoName: "repo", + }) + + // Create a PR watch and TaskPR in DB. + watch := &PRWatch{ + SessionID: "s1", + TaskID: "t1", + Owner: "org", + Repo: "repo", + PRNumber: 10, + Branch: "feat/x", + } + if err := store.CreatePRWatch(ctx, watch); err != nil { + t.Fatal(err) + } + tp := &TaskPR{ + TaskID: "t1", + Owner: "org", + Repo: "repo", + PRNumber: 10, + PRTitle: "Feature PR", + State: "open", + } + if err := store.CreateTaskPR(ctx, tp); err != nil { + t.Fatal(err) + } + + // Trigger sync. + result, err := svc.TriggerPRSync(ctx, "t1") + if err != nil { + t.Fatalf("TriggerPRSync: %v", err) + } + if result == nil { + t.Fatal("expected non-nil TaskPR") + } + if result.LastSyncedAt == nil { + t.Error("expected LastSyncedAt to be set after sync") + } +} + +func TestTriggerPRSync_DetectsPR(t *testing.T) { + _, svc, mockClient, store := setupPollerTest(t) + ctx := context.Background() + + // Set up a PR findable by branch. + mockClient.AddPR(&PR{ + Number: 20, + Title: "New PR", + State: "open", + HeadBranch: "feat/y", + RepoOwner: "org", + RepoName: "repo", + }) + + // Create a watch with pr_number=0 (still searching). + watch := &PRWatch{ + SessionID: "s2", + TaskID: "t2", + Owner: "org", + Repo: "repo", + PRNumber: 0, + Branch: "feat/y", + } + if err := store.CreatePRWatch(ctx, watch); err != nil { + t.Fatal(err) + } + + result, err := svc.TriggerPRSync(ctx, "t2") + if err != nil { + t.Fatalf("TriggerPRSync: %v", err) + } + if result == nil { + t.Fatal("expected TaskPR after detection") + } + if result.PRNumber != 20 { + t.Errorf("expected PR #20, got #%d", result.PRNumber) + } +} + +func TestTriggerPRSync_NoWatch(t *testing.T) { + _, svc, _, _ := setupPollerTest(t) + ctx := context.Background() + + result, err := svc.TriggerPRSync(ctx, "nonexistent") + if err != nil { + t.Fatalf("TriggerPRSync: %v", err) + } + if result != nil { + t.Errorf("expected nil TaskPR for task without watch, got %+v", result) + } +} diff --git a/apps/backend/internal/github/service_review_cleanup_test.go b/apps/backend/internal/github/service_review_cleanup_test.go new file mode 100644 index 0000000000..eaa8227745 --- /dev/null +++ b/apps/backend/internal/github/service_review_cleanup_test.go @@ -0,0 +1,126 @@ +package github + +import ( + "context" + "fmt" + "testing" +) + +// stubTaskDeleter implements TaskDeleter for testing. +type stubTaskDeleter struct { + err error +} + +func (s *stubTaskDeleter) DeleteTask(_ context.Context, _ string) error { + return s.err +} + +// TestCleanupMergedReviewTasks_TaskAlreadyDeleted verifies that when DeleteTask +// returns "not found" the orphaned dedup record is still removed, preventing the +// 5-minute poller from logging the same warning indefinitely. +func TestCleanupMergedReviewTasks_TaskAlreadyDeleted(t *testing.T) { + _, svc, mockClient, store := setupPollerTest(t) + ctx := context.Background() + + // Create a review watch. + watch := &ReviewWatch{WorkspaceID: "ws-1", Enabled: true} + if err := store.CreateReviewWatch(ctx, watch); err != nil { + t.Fatalf("CreateReviewWatch: %v", err) + } + + // Create a dedup record pointing to an already-deleted task. + taskID := "task-already-gone" + rpt := &ReviewPRTask{ + ReviewWatchID: watch.ID, + RepoOwner: "acme", + RepoName: "widget", + PRNumber: 42, + PRURL: "https://github.com/acme/widget/pull/42", + TaskID: taskID, + } + if err := store.CreateReviewPRTask(ctx, rpt); err != nil { + t.Fatalf("CreateReviewPRTask: %v", err) + } + + // Mock: PR is merged so shouldDeleteReviewTask returns true. + mockClient.AddPR(&PR{ + Number: 42, + State: prStateMerged, + RepoOwner: "acme", + RepoName: "widget", + }) + + // Stub: DeleteTask returns "not found" as the real service does. + svc.SetTaskDeleter(&stubTaskDeleter{ + err: fmt.Errorf("task not found: %s", taskID), + }) + + deleted, err := svc.CleanupMergedReviewTasks(ctx, watch) + if err != nil { + t.Fatalf("CleanupMergedReviewTasks returned error: %v", err) + } + if deleted != 1 { + t.Errorf("expected 1 deleted, got %d", deleted) + } + + // The orphaned dedup record must be gone. + remaining, err := store.ListReviewPRTasksByWatch(ctx, watch.ID) + if err != nil { + t.Fatalf("ListReviewPRTasksByWatch: %v", err) + } + if len(remaining) != 0 { + t.Errorf("expected 0 remaining dedup records, got %d", len(remaining)) + } +} + +// Regression: when a task already has a TaskPR row pointing to an old PR +// (e.g. the first PR was closed and a second one opened on the same or a new +// branch), AssociatePRWithTask must replace the stale row so downstream +// consumers (UI, GetTaskPR) observe the current PR rather than the old one. +func TestAssociatePRWithTask_ReplacesStaleAssociation(t *testing.T) { + svc, store, _ := setupSyncTest(t) + ctx := context.Background() + + // Seed an existing association for PR #1. + if err := store.CreateTaskPR(ctx, &TaskPR{ + TaskID: "t1", + Owner: "owner", + Repo: "repo", + PRNumber: 1, + PRURL: "https://github.com/owner/repo/pull/1", + PRTitle: "First", + HeadBranch: "feat-a", + BaseBranch: "main", + State: "closed", + }); err != nil { + t.Fatalf("seed TaskPR: %v", err) + } + + // Associate a new PR #2 (could be on same or different branch). + newPR := &PR{ + Number: 2, + Title: "Second", + HTMLURL: "https://github.com/owner/repo/pull/2", + HeadBranch: "feat-b", + BaseBranch: "main", + State: "open", + RepoOwner: "owner", + RepoName: "repo", + AuthorLogin: "alice", + } + tp, err := svc.AssociatePRWithTask(ctx, "t1", newPR) + if err != nil { + t.Fatalf("AssociatePRWithTask: %v", err) + } + if tp.PRNumber != 2 { + t.Errorf("returned TaskPR.PRNumber=%d, want 2", tp.PRNumber) + } + + got, err := store.GetTaskPR(ctx, "t1") + if err != nil { + t.Fatalf("GetTaskPR: %v", err) + } + if got == nil || got.PRNumber != 2 { + t.Errorf("GetTaskPR after replace = %+v, want PRNumber=2", got) + } +} diff --git a/apps/backend/internal/github/service_test.go b/apps/backend/internal/github/service_test.go index 84e1b0d843..667fab3657 100644 --- a/apps/backend/internal/github/service_test.go +++ b/apps/backend/internal/github/service_test.go @@ -3,7 +3,6 @@ package github import ( "context" "database/sql" - "fmt" "strings" "sync" "testing" @@ -778,225 +777,3 @@ func TestTimeEqual(t *testing.T) { } } -func TestTriggerPRSync_SyncsExistingWatch(t *testing.T) { - _, svc, mockClient, store := setupPollerTest(t) - ctx := context.Background() - - // Set up a PR in mock client. - mockClient.AddPR(&PR{ - Number: 10, - Title: "Feature PR", - State: "open", - HeadSHA: "abc123", - HeadBranch: "feat/x", - RepoOwner: "org", - RepoName: "repo", - }) - - // Create a PR watch and TaskPR in DB. - watch := &PRWatch{ - SessionID: "s1", - TaskID: "t1", - Owner: "org", - Repo: "repo", - PRNumber: 10, - Branch: "feat/x", - } - if err := store.CreatePRWatch(ctx, watch); err != nil { - t.Fatal(err) - } - tp := &TaskPR{ - TaskID: "t1", - Owner: "org", - Repo: "repo", - PRNumber: 10, - PRTitle: "Feature PR", - State: "open", - } - if err := store.CreateTaskPR(ctx, tp); err != nil { - t.Fatal(err) - } - - // Trigger sync. - result, err := svc.TriggerPRSync(ctx, "t1") - if err != nil { - t.Fatalf("TriggerPRSync: %v", err) - } - if result == nil { - t.Fatal("expected non-nil TaskPR") - } - if result.LastSyncedAt == nil { - t.Error("expected LastSyncedAt to be set after sync") - } -} - -func TestTriggerPRSync_DetectsPR(t *testing.T) { - _, svc, mockClient, store := setupPollerTest(t) - ctx := context.Background() - - // Set up a PR findable by branch. - mockClient.AddPR(&PR{ - Number: 20, - Title: "New PR", - State: "open", - HeadBranch: "feat/y", - RepoOwner: "org", - RepoName: "repo", - }) - - // Create a watch with pr_number=0 (still searching). - watch := &PRWatch{ - SessionID: "s2", - TaskID: "t2", - Owner: "org", - Repo: "repo", - PRNumber: 0, - Branch: "feat/y", - } - if err := store.CreatePRWatch(ctx, watch); err != nil { - t.Fatal(err) - } - - result, err := svc.TriggerPRSync(ctx, "t2") - if err != nil { - t.Fatalf("TriggerPRSync: %v", err) - } - if result == nil { - t.Fatal("expected TaskPR after detection") - } - if result.PRNumber != 20 { - t.Errorf("expected PR #20, got #%d", result.PRNumber) - } -} - -func TestTriggerPRSync_NoWatch(t *testing.T) { - _, svc, _, _ := setupPollerTest(t) - ctx := context.Background() - - result, err := svc.TriggerPRSync(ctx, "nonexistent") - if err != nil { - t.Fatalf("TriggerPRSync: %v", err) - } - if result != nil { - t.Errorf("expected nil TaskPR for task without watch, got %+v", result) - } -} - -// stubTaskDeleter implements TaskDeleter for testing. -type stubTaskDeleter struct { - err error -} - -func (s *stubTaskDeleter) DeleteTask(_ context.Context, _ string) error { - return s.err -} - -// TestCleanupMergedReviewTasks_TaskAlreadyDeleted verifies that when DeleteTask -// returns "not found" the orphaned dedup record is still removed, preventing the -// 5-minute poller from logging the same warning indefinitely. -func TestCleanupMergedReviewTasks_TaskAlreadyDeleted(t *testing.T) { - _, svc, mockClient, store := setupPollerTest(t) - ctx := context.Background() - - // Create a review watch. - watch := &ReviewWatch{WorkspaceID: "ws-1", Enabled: true} - if err := store.CreateReviewWatch(ctx, watch); err != nil { - t.Fatalf("CreateReviewWatch: %v", err) - } - - // Create a dedup record pointing to an already-deleted task. - taskID := "task-already-gone" - rpt := &ReviewPRTask{ - ReviewWatchID: watch.ID, - RepoOwner: "acme", - RepoName: "widget", - PRNumber: 42, - PRURL: "https://github.com/acme/widget/pull/42", - TaskID: taskID, - } - if err := store.CreateReviewPRTask(ctx, rpt); err != nil { - t.Fatalf("CreateReviewPRTask: %v", err) - } - - // Mock: PR is merged so shouldDeleteReviewTask returns true. - mockClient.AddPR(&PR{ - Number: 42, - State: prStateMerged, - RepoOwner: "acme", - RepoName: "widget", - }) - - // Stub: DeleteTask returns "not found" as the real service does. - svc.SetTaskDeleter(&stubTaskDeleter{ - err: fmt.Errorf("task not found: %s", taskID), - }) - - deleted, err := svc.CleanupMergedReviewTasks(ctx, watch) - if err != nil { - t.Fatalf("CleanupMergedReviewTasks returned error: %v", err) - } - if deleted != 1 { - t.Errorf("expected 1 deleted, got %d", deleted) - } - - // The orphaned dedup record must be gone. - remaining, err := store.ListReviewPRTasksByWatch(ctx, watch.ID) - if err != nil { - t.Fatalf("ListReviewPRTasksByWatch: %v", err) - } - if len(remaining) != 0 { - t.Errorf("expected 0 remaining dedup records, got %d", len(remaining)) - } -} - -// Regression: when a task already has a TaskPR row pointing to an old PR -// (e.g. the first PR was closed and a second one opened on the same or a new -// branch), AssociatePRWithTask must replace the stale row so downstream -// consumers (UI, GetTaskPR) observe the current PR rather than the old one. -func TestAssociatePRWithTask_ReplacesStaleAssociation(t *testing.T) { - svc, store, _ := setupSyncTest(t) - ctx := context.Background() - - // Seed an existing association for PR #1. - if err := store.CreateTaskPR(ctx, &TaskPR{ - TaskID: "t1", - Owner: "owner", - Repo: "repo", - PRNumber: 1, - PRURL: "https://github.com/owner/repo/pull/1", - PRTitle: "First", - HeadBranch: "feat-a", - BaseBranch: "main", - State: "closed", - }); err != nil { - t.Fatalf("seed TaskPR: %v", err) - } - - // Associate a new PR #2 (could be on same or different branch). - newPR := &PR{ - Number: 2, - Title: "Second", - HTMLURL: "https://github.com/owner/repo/pull/2", - HeadBranch: "feat-b", - BaseBranch: "main", - State: "open", - RepoOwner: "owner", - RepoName: "repo", - AuthorLogin: "alice", - } - tp, err := svc.AssociatePRWithTask(ctx, "t1", newPR) - if err != nil { - t.Fatalf("AssociatePRWithTask: %v", err) - } - if tp.PRNumber != 2 { - t.Errorf("returned TaskPR.PRNumber=%d, want 2", tp.PRNumber) - } - - got, err := store.GetTaskPR(ctx, "t1") - if err != nil { - t.Fatalf("GetTaskPR: %v", err) - } - if got == nil || got.PRNumber != 2 { - t.Errorf("GetTaskPR after replace = %+v, want PRNumber=2", got) - } -} diff --git a/apps/backend/internal/github/store.go b/apps/backend/internal/github/store.go index 60cfc0eed9..83afb541d1 100644 --- a/apps/backend/internal/github/store.go +++ b/apps/backend/internal/github/store.go @@ -6,6 +6,7 @@ import ( "encoding/json" "errors" "fmt" + "os" "time" "github.com/google/uuid" @@ -102,6 +103,36 @@ const createTablesSQL = ` created_at DATETIME NOT NULL, UNIQUE(review_watch_id, repo_owner, repo_name, pr_number) ); + + CREATE TABLE IF NOT EXISTS github_issue_watches ( + id TEXT PRIMARY KEY, + workspace_id TEXT NOT NULL, + workflow_id TEXT NOT NULL, + workflow_step_id TEXT NOT NULL, + repos TEXT NOT NULL DEFAULT '[]', + agent_profile_id TEXT NOT NULL, + executor_profile_id TEXT NOT NULL, + prompt TEXT DEFAULT '', + labels TEXT NOT NULL DEFAULT '[]', + custom_query TEXT NOT NULL DEFAULT '', + enabled BOOLEAN DEFAULT 1, + poll_interval_seconds INTEGER DEFAULT 300, + last_polled_at DATETIME, + created_at DATETIME NOT NULL, + updated_at DATETIME NOT NULL + ); + + CREATE TABLE IF NOT EXISTS github_issue_watch_tasks ( + id TEXT PRIMARY KEY, + issue_watch_id TEXT NOT NULL, + repo_owner TEXT NOT NULL DEFAULT '', + repo_name TEXT NOT NULL DEFAULT '', + issue_number INTEGER NOT NULL, + issue_url TEXT NOT NULL, + task_id TEXT NOT NULL, + created_at DATETIME NOT NULL, + UNIQUE(issue_watch_id, repo_owner, repo_name, issue_number) + ); ` func (s *Store) initSchema() error { @@ -371,7 +402,7 @@ func hydrateReviewWatchRepos(rw *ReviewWatch) { if rw.ReposJSON != "" { if err := json.Unmarshal([]byte(rw.ReposJSON), &rw.Repos); err != nil { // Log but don't fail — the watch can still function with no repo filter. - fmt.Printf("WARN: failed to unmarshal repos JSON for review watch %s: %v\n", rw.ID, err) + fmt.Fprintf(os.Stderr, "WARN: failed to unmarshal repos JSON for review watch %s: %v\n", rw.ID, err) } } if rw.Repos == nil { @@ -651,3 +682,209 @@ func (s *Store) fetchApprovalRate(ctx context.Context, q *prStatsQuery, stats *P } return nil } + +// --- Issue Watch operations --- + +// hydrateIssueWatch unmarshals JSON fields into their Go slices. +func hydrateIssueWatch(iw *IssueWatch) { + if iw.ReposJSON != "" { + if err := json.Unmarshal([]byte(iw.ReposJSON), &iw.Repos); err != nil { + fmt.Fprintf(os.Stderr, "WARN: failed to unmarshal repos JSON for issue watch %s: %v\n", iw.ID, err) + } + } + if iw.Repos == nil { + iw.Repos = []RepoFilter{} + } + if iw.LabelsJSON != "" { + if err := json.Unmarshal([]byte(iw.LabelsJSON), &iw.Labels); err != nil { + fmt.Fprintf(os.Stderr, "WARN: failed to unmarshal labels JSON for issue watch %s: %v\n", iw.ID, err) + } + } + if iw.Labels == nil { + iw.Labels = []string{} + } +} + +// CreateIssueWatch creates a new issue watch configuration. +func (s *Store) CreateIssueWatch(ctx context.Context, iw *IssueWatch) error { + if iw.ID == "" { + iw.ID = uuid.New().String() + } + now := time.Now().UTC() + iw.CreatedAt = now + iw.UpdatedAt = now + reposJSON, err := json.Marshal(iw.Repos) + if err != nil { + return fmt.Errorf("marshal repos: %w", err) + } + iw.ReposJSON = string(reposJSON) + labelsJSON, err := json.Marshal(iw.Labels) + if err != nil { + return fmt.Errorf("marshal labels: %w", err) + } + iw.LabelsJSON = string(labelsJSON) + _, err = s.db.ExecContext(ctx, ` + INSERT INTO github_issue_watches (id, workspace_id, workflow_id, workflow_step_id, repos, + agent_profile_id, executor_profile_id, prompt, labels, custom_query, + enabled, poll_interval_seconds, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + iw.ID, iw.WorkspaceID, iw.WorkflowID, iw.WorkflowStepID, iw.ReposJSON, + iw.AgentProfileID, iw.ExecutorProfileID, iw.Prompt, iw.LabelsJSON, iw.CustomQuery, + iw.Enabled, iw.PollIntervalSeconds, iw.CreatedAt, iw.UpdatedAt) + return err +} + +// GetIssueWatch returns an issue watch by ID. +func (s *Store) GetIssueWatch(ctx context.Context, id string) (*IssueWatch, error) { + var iw IssueWatch + err := s.ro.GetContext(ctx, &iw, `SELECT * FROM github_issue_watches WHERE id = ?`, id) + if errors.Is(err, sql.ErrNoRows) { + return nil, nil + } + if err != nil { + return nil, err + } + hydrateIssueWatch(&iw) + return &iw, nil +} + +// ListIssueWatches returns all issue watches for a workspace. +func (s *Store) ListIssueWatches(ctx context.Context, workspaceID string) ([]*IssueWatch, error) { + var watches []*IssueWatch + err := s.ro.SelectContext(ctx, &watches, + `SELECT * FROM github_issue_watches WHERE workspace_id = ? ORDER BY created_at`, workspaceID) + if err != nil { + return nil, err + } + for _, w := range watches { + hydrateIssueWatch(w) + } + return watches, nil +} + +// ListEnabledIssueWatches returns all enabled issue watches. +func (s *Store) ListEnabledIssueWatches(ctx context.Context) ([]*IssueWatch, error) { + var watches []*IssueWatch + err := s.ro.SelectContext(ctx, &watches, + `SELECT * FROM github_issue_watches WHERE enabled = 1 ORDER BY created_at`) + if err != nil { + return nil, err + } + for _, w := range watches { + hydrateIssueWatch(w) + } + return watches, nil +} + +// UpdateIssueWatch updates an issue watch. +func (s *Store) UpdateIssueWatch(ctx context.Context, iw *IssueWatch) error { + iw.UpdatedAt = time.Now().UTC() + reposJSON, err := json.Marshal(iw.Repos) + if err != nil { + return fmt.Errorf("marshal repos: %w", err) + } + iw.ReposJSON = string(reposJSON) + labelsJSON, err := json.Marshal(iw.Labels) + if err != nil { + return fmt.Errorf("marshal labels: %w", err) + } + iw.LabelsJSON = string(labelsJSON) + _, err = s.db.ExecContext(ctx, ` + UPDATE github_issue_watches SET workflow_id = ?, workflow_step_id = ?, repos = ?, + agent_profile_id = ?, executor_profile_id = ?, + prompt = ?, labels = ?, custom_query = ?, + enabled = ?, poll_interval_seconds = ?, last_polled_at = ?, updated_at = ? + WHERE id = ?`, + iw.WorkflowID, iw.WorkflowStepID, iw.ReposJSON, + iw.AgentProfileID, iw.ExecutorProfileID, + iw.Prompt, iw.LabelsJSON, iw.CustomQuery, + iw.Enabled, iw.PollIntervalSeconds, iw.LastPolledAt, iw.UpdatedAt, iw.ID) + return err +} + +// DeleteIssueWatch deletes an issue watch and all its associated dedup task rows. +func (s *Store) DeleteIssueWatch(ctx context.Context, id string) error { + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return err + } + defer func() { _ = tx.Rollback() }() + if _, err := tx.ExecContext(ctx, `DELETE FROM github_issue_watch_tasks WHERE issue_watch_id = ?`, id); err != nil { + return err + } + if _, err := tx.ExecContext(ctx, `DELETE FROM github_issue_watches WHERE id = ?`, id); err != nil { + return err + } + return tx.Commit() +} + +// --- Issue Watch Task deduplication --- + +// ReserveIssueWatchTask atomically claims a slot for a (watch, repo, issue) tuple. +// Returns true if this caller won the race and should proceed to create the task. +func (s *Store) ReserveIssueWatchTask(ctx context.Context, issueWatchID, repoOwner, repoName string, issueNumber int, issueURL string) (bool, error) { + res, err := s.db.ExecContext(ctx, ` + INSERT OR IGNORE INTO github_issue_watch_tasks (id, issue_watch_id, repo_owner, repo_name, issue_number, issue_url, task_id, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, + uuid.New().String(), issueWatchID, repoOwner, repoName, issueNumber, issueURL, "", time.Now().UTC()) + if err != nil { + return false, err + } + rows, err := res.RowsAffected() + if err != nil { + return false, err + } + return rows == 1, nil +} + +// AssignIssueWatchTaskID sets the task_id on a reserved dedup row. +func (s *Store) AssignIssueWatchTaskID(ctx context.Context, issueWatchID, repoOwner, repoName string, issueNumber int, taskID string) error { + res, err := s.db.ExecContext(ctx, ` + UPDATE github_issue_watch_tasks SET task_id = ? + WHERE issue_watch_id = ? AND repo_owner = ? AND repo_name = ? AND issue_number = ?`, + taskID, issueWatchID, repoOwner, repoName, issueNumber) + if err != nil { + return err + } + rows, err := res.RowsAffected() + if err != nil { + return err + } + if rows == 0 { + return fmt.Errorf("assign task ID: reservation row not found for watch=%s issue=%d", issueWatchID, issueNumber) + } + return nil +} + +// ReleaseIssueWatchTask removes a reservation for a (watch, repo, issue) tuple. +func (s *Store) ReleaseIssueWatchTask(ctx context.Context, issueWatchID, repoOwner, repoName string, issueNumber int) error { + _, err := s.db.ExecContext(ctx, ` + DELETE FROM github_issue_watch_tasks + WHERE issue_watch_id = ? AND repo_owner = ? AND repo_name = ? AND issue_number = ?`, + issueWatchID, repoOwner, repoName, issueNumber) + return err +} + +// HasIssueWatchTask checks if a task was already created for an issue in an issue watch. +func (s *Store) HasIssueWatchTask(ctx context.Context, issueWatchID, repoOwner, repoName string, issueNumber int) (bool, error) { + var count int + err := s.ro.GetContext(ctx, &count, + `SELECT COUNT(*) FROM github_issue_watch_tasks WHERE issue_watch_id = ? AND repo_owner = ? AND repo_name = ? AND issue_number = ?`, + issueWatchID, repoOwner, repoName, issueNumber) + return count > 0, err +} + +// ListIssueWatchTasksByWatch lists all dedup records for a given issue watch. +func (s *Store) ListIssueWatchTasksByWatch(ctx context.Context, watchID string) ([]*IssueWatchTask, error) { + var tasks []*IssueWatchTask + err := s.ro.SelectContext(ctx, &tasks, + `SELECT id, issue_watch_id, repo_owner, repo_name, issue_number, issue_url, task_id, created_at + FROM github_issue_watch_tasks WHERE issue_watch_id = ?`, watchID) + return tasks, err +} + +// DeleteIssueWatchTask deletes a dedup record by ID. +func (s *Store) DeleteIssueWatchTask(ctx context.Context, id string) error { + _, err := s.db.ExecContext(ctx, `DELETE FROM github_issue_watch_tasks WHERE id = ?`, id) + return err +} diff --git a/apps/backend/internal/orchestrator/event_handlers_github.go b/apps/backend/internal/orchestrator/event_handlers_github.go index d7ef389300..cab6fb75b3 100644 --- a/apps/backend/internal/orchestrator/event_handlers_github.go +++ b/apps/backend/internal/orchestrator/event_handlers_github.go @@ -18,6 +18,11 @@ import ( wfmodels "github.com/kandev/kandev/internal/workflow/models" ) +const ( + issueWatchIDKey = "issue_watch_id" + issueNumberKey = "issue_number" +) + // GitHubService is the interface the orchestrator uses for GitHub operations. type GitHubService interface { Client() github.Client @@ -33,6 +38,11 @@ type GitHubService interface { ReserveReviewPRTask(ctx context.Context, watchID, repoOwner, repoName string, prNumber int, prURL string) (bool, error) AssignReviewPRTaskID(ctx context.Context, watchID, repoOwner, repoName string, prNumber int, taskID string) error ReleaseReviewPRTask(ctx context.Context, watchID, repoOwner, repoName string, prNumber int) error + + // Issue watch dedup operations + ReserveIssueWatchTask(ctx context.Context, watchID, repoOwner, repoName string, issueNumber int, issueURL string) (bool, error) + AssignIssueWatchTaskID(ctx context.Context, watchID, repoOwner, repoName string, issueNumber int, taskID string) error + ReleaseIssueWatchTask(ctx context.Context, watchID, repoOwner, repoName string, issueNumber int) error } // ReviewTaskCreator creates tasks from review watch events. @@ -40,6 +50,28 @@ type ReviewTaskCreator interface { CreateReviewTask(ctx context.Context, req *ReviewTaskRequest) (*models.Task, error) } +// IssueTaskCreator creates tasks from issue watch events. +type IssueTaskCreator interface { + CreateIssueTask(ctx context.Context, req *IssueTaskRequest) (*models.Task, error) +} + +// IssueTaskRequest contains the data for creating a task from an issue watch event. +type IssueTaskRequest struct { + WorkspaceID string + WorkflowID string + WorkflowStepID string + Title string + Description string + Metadata map[string]interface{} + Repositories []IssueTaskRepository +} + +// IssueTaskRepository associates a repository with an issue task. +type IssueTaskRepository struct { + RepositoryID string + BaseBranch string +} + // RepositoryResolver resolves a GitHub repo to a local clone + Repository DB record. type RepositoryResolver interface { ResolveForReview(ctx context.Context, workspaceID, provider, owner, name, defaultBranch string) (repositoryID, baseBranch string, err error) @@ -78,6 +110,11 @@ func (s *Service) SetRepositoryResolver(rr RepositoryResolver) { s.repositoryResolver = rr } +// SetIssueTaskCreator sets the task creator for issue watch auto-task creation. +func (s *Service) SetIssueTaskCreator(tc IssueTaskCreator) { + s.issueTaskCreator = tc +} + // handlePRFeedback logs PR feedback events. WS broadcasting is handled in main.go. func (s *Service) handlePRFeedback(_ context.Context, event *bus.Event) error { feedbackEvt, ok := event.Data.(*github.PRFeedbackEvent) @@ -708,6 +745,189 @@ func (s *Service) subscribeGitHubEvents() { if _, err := s.eventBus.Subscribe(events.GitHubNewReviewPR, s.handleNewReviewPR); err != nil { s.logger.Error("failed to subscribe to github.new_pr_to_review events", zap.Error(err)) } + if _, err := s.eventBus.Subscribe(events.GitHubNewIssue, s.handleNewIssue); err != nil { + s.logger.Error("failed to subscribe to github.new_issue events", zap.Error(err)) + } +} + +// handleNewIssue creates a task for a new GitHub issue matching an issue watch. +func (s *Service) handleNewIssue(ctx context.Context, event *bus.Event) error { + issueEvt, ok := event.Data.(*github.NewIssueEvent) + if !ok { + return nil + } + + issue := issueEvt.Issue + s.logger.Info("new issue detected from watch", + zap.String(issueWatchIDKey, issueEvt.IssueWatchID), + zap.String("repo", fmt.Sprintf("%s/%s", issue.RepoOwner, issue.RepoName)), + zap.Int(issueNumberKey, issue.Number)) + + if s.issueTaskCreator == nil { + s.logger.Warn("issue task creator not configured, skipping task creation") + return nil + } + + go s.createIssueTask(context.Background(), issueEvt) + return nil +} + +func (s *Service) createIssueTask(ctx context.Context, evt *github.NewIssueEvent) { + issue := evt.Issue + repoSlug := fmt.Sprintf("%s/%s", issue.RepoOwner, issue.RepoName) + + if !s.reserveIssueWatch(ctx, evt) { + return + } + + repositories := s.resolveIssueRepository(ctx, evt.WorkspaceID, issue) + + req := &IssueTaskRequest{ + WorkspaceID: evt.WorkspaceID, + WorkflowID: evt.WorkflowID, + WorkflowStepID: evt.WorkflowStepID, + Title: fmt.Sprintf("Issue #%d: %s", issue.Number, issue.Title), + Description: interpolateIssuePrompt(evt.Prompt, issue), + Metadata: map[string]interface{}{ + issueWatchIDKey: evt.IssueWatchID, + issueNumberKey: issue.Number, + "issue_url": issue.HTMLURL, + "issue_repo": repoSlug, + "issue_author": issue.AuthorLogin, + "agent_profile_id": evt.AgentProfileID, + "executor_profile_id": evt.ExecutorProfileID, + }, + Repositories: repositories, + } + + task, err := s.issueTaskCreator.CreateIssueTask(ctx, req) + if err != nil { + s.logger.Error("failed to create issue task", + zap.String(issueWatchIDKey, evt.IssueWatchID), + zap.Int(issueNumberKey, issue.Number), + zap.Error(err)) + s.releaseIssueWatch(ctx, evt) + return + } + + s.attachIssueTaskToReservation(ctx, evt, task.ID) + + s.logger.Info("created issue task", + zap.String("task_id", task.ID), + zap.Int(issueNumberKey, issue.Number), + zap.String("repo", repoSlug)) + + if !s.shouldAutoStartStep(ctx, evt.WorkflowStepID) { + return + } + s.autoStartIssueTask(ctx, evt, task) +} + +func (s *Service) reserveIssueWatch(ctx context.Context, evt *github.NewIssueEvent) bool { + if s.githubService == nil { + return true + } + issue := evt.Issue + reserved, err := s.githubService.ReserveIssueWatchTask( + ctx, evt.IssueWatchID, issue.RepoOwner, issue.RepoName, issue.Number, issue.HTMLURL, + ) + if err != nil { + s.logger.Error("failed to reserve issue watch slot", + zap.String(issueWatchIDKey, evt.IssueWatchID), + zap.Int(issueNumberKey, issue.Number), + zap.Error(err)) + return false + } + if !reserved { + s.logger.Debug("issue already reserved by concurrent handler, skipping", + zap.String(issueWatchIDKey, evt.IssueWatchID), + zap.Int(issueNumberKey, issue.Number)) + return false + } + return true +} + +func (s *Service) releaseIssueWatch(ctx context.Context, evt *github.NewIssueEvent) { + if s.githubService == nil { + return + } + issue := evt.Issue + if err := s.githubService.ReleaseIssueWatchTask( + ctx, evt.IssueWatchID, issue.RepoOwner, issue.RepoName, issue.Number, + ); err != nil { + s.logger.Warn("failed to release issue watch reservation after task-create failure", + zap.Int(issueNumberKey, issue.Number), + zap.Error(err)) + } +} + +func (s *Service) attachIssueTaskToReservation(ctx context.Context, evt *github.NewIssueEvent, taskID string) { + if s.githubService == nil { + return + } + issue := evt.Issue + if err := s.githubService.AssignIssueWatchTaskID( + ctx, evt.IssueWatchID, issue.RepoOwner, issue.RepoName, issue.Number, taskID, + ); err != nil { + s.logger.Error("failed to assign task ID to issue watch reservation", + zap.String("task_id", taskID), + zap.Int(issueNumberKey, issue.Number), + zap.Error(err)) + } +} + +// resolveIssueRepository attempts to resolve and clone the issue's repository. +// Returns a slice with one entry on success, or nil on failure (graceful degradation). +func (s *Service) resolveIssueRepository(ctx context.Context, workspaceID string, issue *github.Issue) []IssueTaskRepository { + if s.repositoryResolver == nil { + return nil + } + repoSlug := fmt.Sprintf("%s/%s", issue.RepoOwner, issue.RepoName) + s.logger.Debug("resolving issue repository", zap.String("repo", repoSlug)) + + repoID, baseBranch, err := s.repositoryResolver.ResolveForReview( + ctx, workspaceID, "github", issue.RepoOwner, issue.RepoName, "", + ) + if err != nil { + s.logger.Warn("failed to resolve repository for issue task (continuing without repo)", + zap.String("repo", repoSlug), + zap.Error(err)) + return nil + } + if repoID == "" { + return nil + } + s.logger.Debug("resolved issue repository", + zap.String("repo", repoSlug), + zap.String("repo_id", repoID), + zap.String("base_branch", baseBranch)) + return []IssueTaskRepository{{RepositoryID: repoID, BaseBranch: baseBranch}} +} + +func (s *Service) autoStartIssueTask( + ctx context.Context, evt *github.NewIssueEvent, task *models.Task, +) { + _, err := s.StartTask( + ctx, + task.ID, + evt.AgentProfileID, + "", + evt.ExecutorProfileID, + 0, + task.Description, + evt.WorkflowStepID, + false, + nil, + ) + if err != nil { + s.logger.Error("failed to auto-start issue task", + zap.String("task_id", task.ID), + zap.Error(err)) + return + } + s.logger.Info("auto-started issue task", + zap.String("task_id", task.ID), + zap.Int(issueNumberKey, evt.Issue.Number)) } // interpolateReviewPrompt replaces {{pr.*}} placeholders in the prompt template with actual PR values. @@ -729,3 +949,22 @@ func interpolateReviewPrompt(promptTemplate string, pr *github.PR) string { ) return replacer.Replace(promptTemplate) } + +// interpolateIssuePrompt replaces {{issue.*}} placeholders in the prompt template with actual issue values. +func interpolateIssuePrompt(promptTemplate string, issue *github.Issue) string { + if promptTemplate == "" { + promptTemplate = promptcfg.Get("issue-watch-default") + } + repoSlug := fmt.Sprintf("%s/%s", issue.RepoOwner, issue.RepoName) + labelsStr := strings.Join(issue.Labels, ", ") + replacer := strings.NewReplacer( + "{{issue.link}}", issue.HTMLURL, + "{{issue.number}}", strconv.Itoa(issue.Number), + "{{issue.title}}", issue.Title, + "{{issue.author}}", issue.AuthorLogin, + "{{issue.repo}}", repoSlug, + "{{issue.labels}}", labelsStr, + "{{issue.body}}", issue.Body, + ) + return replacer.Replace(promptTemplate) +} diff --git a/apps/backend/internal/orchestrator/event_handlers_github_issue_test.go b/apps/backend/internal/orchestrator/event_handlers_github_issue_test.go new file mode 100644 index 0000000000..f935b0bf40 --- /dev/null +++ b/apps/backend/internal/orchestrator/event_handlers_github_issue_test.go @@ -0,0 +1,188 @@ +package orchestrator + +import ( + "context" + "testing" + + "github.com/kandev/kandev/internal/github" + "github.com/kandev/kandev/internal/task/models" + wfmodels "github.com/kandev/kandev/internal/workflow/models" +) + +// --- Issue Watch Tests --- + +func TestInterpolateIssuePrompt(t *testing.T) { + issue := &github.Issue{ + Number: 7, + Title: "Fix login bug", + Body: "Login fails on mobile", + HTMLURL: "https://github.com/acme/widget/issues/7", + AuthorLogin: "bob", + RepoOwner: "acme", + RepoName: "widget", + Labels: []string{"bug", "priority:high"}, + } + + tests := []struct { + name string + template string + want string + }{ + { + "empty template uses embedded default", + "", + "You have been assigned a GitHub issue to work on.\n\n" + + "**Issue:** https://github.com/acme/widget/issues/7\n" + + "**Title:** Fix login bug (#7)\n" + + "**Repository:** acme/widget\n" + + "**Author:** bob\n" + + "**Labels:** bug, priority:high\n\n" + + "## Instructions\n\n" + + "1. Read the issue description carefully and understand the requirements.\n" + + "2. Explore the codebase to understand the relevant code and architecture.\n" + + "3. Implement the changes described in the issue.\n" + + "4. Write or update tests to cover the changes.\n" + + "5. Run the test suite to ensure nothing is broken.\n" + + "6. Commit your changes with a descriptive commit message referencing the issue.", + }, + { + "all placeholders", + "Fix {{issue.link}} (#{{issue.number}}) by {{issue.author}} in {{issue.repo}}: {{issue.title}} [{{issue.labels}}]\n{{issue.body}}", + "Fix https://github.com/acme/widget/issues/7 (#7) by bob in acme/widget: Fix login bug [bug, priority:high]\nLogin fails on mobile", + }, + { + "no placeholders", + "Please fix this issue", + "Please fix this issue", + }, + { + "partial placeholders", + "Check {{issue.link}}", + "Check https://github.com/acme/widget/issues/7", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := interpolateIssuePrompt(tt.template, issue) + if got != tt.want { + t.Errorf("got:\n%s\nwant:\n%s", got, tt.want) + } + }) + } +} + +// countingIssueTaskCreator records how many times CreateIssueTask was called. +type countingIssueTaskCreator struct { + calls int + err error + taskID string +} + +func (c *countingIssueTaskCreator) CreateIssueTask(_ context.Context, _ *IssueTaskRequest) (*models.Task, error) { + c.calls++ + if c.err != nil { + return nil, c.err + } + id := c.taskID + if id == "" { + id = "issue-task-created" + } + return &models.Task{ID: id}, nil +} + +// newIssueEvent builds a NewIssueEvent for createIssueTask tests. +func newIssueEvent() *github.NewIssueEvent { + return &github.NewIssueEvent{ + IssueWatchID: "iw1", + WorkspaceID: "ws1", + WorkflowID: "wf1", + WorkflowStepID: "step1", + Issue: &github.Issue{ + Number: 7, Title: "Fix login bug", HTMLURL: "https://gh/acme/widget/issues/7", + RepoOwner: "acme", RepoName: "widget", + }, + } +} + +// setupIssueTaskTest builds a Service with a seeded workflow step. +func setupIssueTaskTest(t *testing.T) (*Service, *mockStepGetter) { + t.Helper() + repo := setupTestRepo(t) + stepGetter := newMockStepGetter() + stepGetter.steps["step1"] = &wfmodels.WorkflowStep{ + ID: "step1", WorkflowID: "wf1", Name: "Step 1", Position: 0, + Events: wfmodels.StepEvents{}, + } + return createTestService(repo, stepGetter, newMockTaskRepo()), stepGetter +} + + +func TestCreateIssueTask_SkipsWhenAlreadyReserved(t *testing.T) { + svc, _ := setupIssueTaskTest(t) + ghSvc := &mockGitHubService{issueReserveReturn: false} + svc.SetGitHubService(ghSvc) + creator := &countingIssueTaskCreator{} + svc.SetIssueTaskCreator(creator) + + svc.createIssueTask(context.Background(), newIssueEvent()) + + if ghSvc.issueReserveCalls != 1 { + t.Errorf("expected 1 Reserve call, got %d", ghSvc.issueReserveCalls) + } + if creator.calls != 0 { + t.Errorf("expected CreateIssueTask NOT to be called when reservation lost, got %d calls", creator.calls) + } + if ghSvc.issueReleaseCalls != 0 { + t.Errorf("expected no Release calls when reservation was never held, got %d", ghSvc.issueReleaseCalls) + } +} + +func TestCreateIssueTask_ReservesThenAssignsTaskID(t *testing.T) { + svc, _ := setupIssueTaskTest(t) + ghSvc := &mockGitHubService{issueReserveReturn: true} + svc.SetGitHubService(ghSvc) + creator := &countingIssueTaskCreator{taskID: "issue-task-999"} + svc.SetIssueTaskCreator(creator) + + svc.createIssueTask(context.Background(), newIssueEvent()) + + if ghSvc.issueReserveCalls != 1 { + t.Errorf("expected 1 Reserve call, got %d", ghSvc.issueReserveCalls) + } + if creator.calls != 1 { + t.Errorf("expected 1 CreateIssueTask call, got %d", creator.calls) + } + if ghSvc.issueAssignCalls != 1 { + t.Errorf("expected 1 AssignIssueWatchTaskID call, got %d", ghSvc.issueAssignCalls) + } + if ghSvc.issueAssignedID != "issue-task-999" { + t.Errorf("AssignIssueWatchTaskID got taskID=%q, want %q", ghSvc.issueAssignedID, "issue-task-999") + } + if ghSvc.issueReleaseCalls != 0 { + t.Errorf("expected no Release calls on happy path, got %d", ghSvc.issueReleaseCalls) + } +} + +func TestCreateIssueTask_ReleasesOnTaskCreationFailure(t *testing.T) { + svc, _ := setupIssueTaskTest(t) + ghSvc := &mockGitHubService{issueReserveReturn: true} + svc.SetGitHubService(ghSvc) + creator := &countingIssueTaskCreator{err: assertErrTaskCreate} + svc.SetIssueTaskCreator(creator) + + svc.createIssueTask(context.Background(), newIssueEvent()) + + if ghSvc.issueReserveCalls != 1 { + t.Errorf("expected 1 Reserve call, got %d", ghSvc.issueReserveCalls) + } + if creator.calls != 1 { + t.Errorf("expected 1 CreateIssueTask call, got %d", creator.calls) + } + if ghSvc.issueReleaseCalls != 1 { + t.Errorf("expected 1 Release call after task-create failure, got %d", ghSvc.issueReleaseCalls) + } + if ghSvc.issueAssignCalls != 0 { + t.Errorf("expected no Assign call after failure, got %d", ghSvc.issueAssignCalls) + } +} \ No newline at end of file diff --git a/apps/backend/internal/orchestrator/event_handlers_github_review_test.go b/apps/backend/internal/orchestrator/event_handlers_github_review_test.go new file mode 100644 index 0000000000..3de9ccd88f --- /dev/null +++ b/apps/backend/internal/orchestrator/event_handlers_github_review_test.go @@ -0,0 +1,139 @@ +package orchestrator + +import ( + "context" + "testing" + + "github.com/kandev/kandev/internal/github" + "github.com/kandev/kandev/internal/task/models" + wfmodels "github.com/kandev/kandev/internal/workflow/models" +) + +// countingReviewTaskCreator records how many times CreateReviewTask was called. +type countingReviewTaskCreator struct { + calls int + err error + taskID string +} + +func (c *countingReviewTaskCreator) CreateReviewTask(_ context.Context, _ *ReviewTaskRequest) (*models.Task, error) { + c.calls++ + if c.err != nil { + return nil, c.err + } + id := c.taskID + if id == "" { + id = "task-created" + } + return &models.Task{ID: id}, nil +} + +// newReviewEvent builds a NewReviewPREvent for createReviewTask tests. +func newReviewEvent() *github.NewReviewPREvent { + return &github.NewReviewPREvent{ + ReviewWatchID: "w1", + WorkspaceID: "ws1", + WorkflowID: "wf1", + WorkflowStepID: "step1", + PR: &github.PR{ + Number: 42, Title: "Some PR", HTMLURL: "https://gh/acme/widget/pull/42", + RepoOwner: "acme", RepoName: "widget", + }, + } +} + +// setupReviewTaskTest builds a Service with a seeded workflow step (needed +// because createReviewTask runs shouldAutoStartStep after task creation). +func setupReviewTaskTest(t *testing.T) (*Service, *mockStepGetter) { + t.Helper() + repo := setupTestRepo(t) + stepGetter := newMockStepGetter() + stepGetter.steps["step1"] = &wfmodels.WorkflowStep{ + ID: "step1", WorkflowID: "wf1", Name: "Step 1", Position: 0, + Events: wfmodels.StepEvents{}, // no auto-start action + } + return createTestService(repo, stepGetter, newMockTaskRepo()), stepGetter +} + +var assertErrTaskCreate = &taskCreateErr{"simulated task creation failure"} + +type taskCreateErr struct{ msg string } + +func (e *taskCreateErr) Error() string { return e.msg } + +// TestCreateReviewTask_SkipsWhenAlreadyReserved is the regression test for the +// duplicate review-task bug: if another handler has already reserved the +// dedup slot for this PR, createReviewTask must NOT call CreateReviewTask. +func TestCreateReviewTask_SkipsWhenAlreadyReserved(t *testing.T) { + svc, _ := setupReviewTaskTest(t) + ghSvc := &mockGitHubService{reserveReturn: false} // reservation lost to concurrent handler + svc.SetGitHubService(ghSvc) + creator := &countingReviewTaskCreator{} + svc.SetReviewTaskCreator(creator) + + svc.createReviewTask(context.Background(), newReviewEvent()) + + if ghSvc.reserveCalls != 1 { + t.Errorf("expected 1 Reserve call, got %d", ghSvc.reserveCalls) + } + if creator.calls != 0 { + t.Errorf("expected CreateReviewTask NOT to be called when reservation lost, got %d calls", creator.calls) + } + if ghSvc.releaseCalls != 0 { + t.Errorf("expected no Release calls when reservation was never held, got %d", ghSvc.releaseCalls) + } +} + +// TestCreateReviewTask_ReservesThenAssignsTaskID verifies the happy path: +// Reserve -> CreateReviewTask -> AssignReviewPRTaskID. +func TestCreateReviewTask_ReservesThenAssignsTaskID(t *testing.T) { + svc, _ := setupReviewTaskTest(t) + ghSvc := &mockGitHubService{reserveReturn: true} + svc.SetGitHubService(ghSvc) + creator := &countingReviewTaskCreator{taskID: "task-999"} + svc.SetReviewTaskCreator(creator) + + svc.createReviewTask(context.Background(), newReviewEvent()) + + if ghSvc.reserveCalls != 1 { + t.Errorf("expected 1 Reserve call, got %d", ghSvc.reserveCalls) + } + if creator.calls != 1 { + t.Errorf("expected 1 CreateReviewTask call, got %d", creator.calls) + } + if ghSvc.assignCalls != 1 { + t.Errorf("expected 1 AssignReviewPRTaskID call, got %d", ghSvc.assignCalls) + } + if ghSvc.assignedTaskID != "task-999" { + t.Errorf("AssignReviewPRTaskID got taskID=%q, want %q", ghSvc.assignedTaskID, "task-999") + } + if ghSvc.releaseCalls != 0 { + t.Errorf("expected no Release calls on happy path, got %d", ghSvc.releaseCalls) + } +} + +// TestCreateReviewTask_ReleasesOnTaskCreationFailure verifies that a failed +// CreateReviewTask triggers a Release so the slot can be retried on the next +// poll instead of being permanently blocked by an orphan reservation. +func TestCreateReviewTask_ReleasesOnTaskCreationFailure(t *testing.T) { + svc, _ := setupReviewTaskTest(t) + ghSvc := &mockGitHubService{reserveReturn: true} + svc.SetGitHubService(ghSvc) + creator := &countingReviewTaskCreator{err: assertErrTaskCreate} + svc.SetReviewTaskCreator(creator) + + svc.createReviewTask(context.Background(), newReviewEvent()) + + if ghSvc.reserveCalls != 1 { + t.Errorf("expected 1 Reserve call, got %d", ghSvc.reserveCalls) + } + if creator.calls != 1 { + t.Errorf("expected 1 CreateReviewTask call, got %d", creator.calls) + } + if ghSvc.releaseCalls != 1 { + t.Errorf("expected 1 Release call after task-create failure, got %d", ghSvc.releaseCalls) + } + if ghSvc.assignCalls != 0 { + t.Errorf("expected no Assign call after failure, got %d", ghSvc.assignCalls) + } +} diff --git a/apps/backend/internal/orchestrator/event_handlers_github_test.go b/apps/backend/internal/orchestrator/event_handlers_github_test.go index 904b881a5b..3fbfa4d531 100644 --- a/apps/backend/internal/orchestrator/event_handlers_github_test.go +++ b/apps/backend/internal/orchestrator/event_handlers_github_test.go @@ -8,7 +8,6 @@ import ( "github.com/kandev/kandev/internal/github" "github.com/kandev/kandev/internal/task/models" sqliterepo "github.com/kandev/kandev/internal/task/repository/sqlite" - wfmodels "github.com/kandev/kandev/internal/workflow/models" v1 "github.com/kandev/kandev/pkg/api/v1" ) @@ -37,6 +36,13 @@ type mockGitHubService struct { assignCalls int assignedTaskID string releaseCalls int + + // Issue reservation tracking. + issueReserveCalls int + issueReserveReturn bool + issueAssignCalls int + issueAssignedID string + issueReleaseCalls int } func (m *mockGitHubService) Client() github.Client { return m.client } @@ -91,6 +97,19 @@ func (m *mockGitHubService) ReleaseReviewPRTask(_ context.Context, _, _, _ strin m.releaseCalls++ return nil } +func (m *mockGitHubService) ReserveIssueWatchTask(_ context.Context, _, _, _ string, _ int, _ string) (bool, error) { + m.issueReserveCalls++ + return m.issueReserveReturn, nil +} +func (m *mockGitHubService) AssignIssueWatchTaskID(_ context.Context, _, _, _ string, _ int, taskID string) error { + m.issueAssignCalls++ + m.issueAssignedID = taskID + return nil +} +func (m *mockGitHubService) ReleaseIssueWatchTask(_ context.Context, _, _, _ string, _ int) error { + m.issueReleaseCalls++ + return nil +} func TestInterpolateReviewPrompt(t *testing.T) { pr := &github.PR{ @@ -656,131 +675,3 @@ func TestListTasksNeedingPRWatch(t *testing.T) { }) } -// countingReviewTaskCreator records how many times CreateReviewTask was called. -type countingReviewTaskCreator struct { - calls int - err error - taskID string -} - -func (c *countingReviewTaskCreator) CreateReviewTask(_ context.Context, _ *ReviewTaskRequest) (*models.Task, error) { - c.calls++ - if c.err != nil { - return nil, c.err - } - id := c.taskID - if id == "" { - id = "task-created" - } - return &models.Task{ID: id}, nil -} - -// newReviewEvent builds a NewReviewPREvent for createReviewTask tests. -func newReviewEvent() *github.NewReviewPREvent { - return &github.NewReviewPREvent{ - ReviewWatchID: "w1", - WorkspaceID: "ws1", - WorkflowID: "wf1", - WorkflowStepID: "step1", - PR: &github.PR{ - Number: 42, Title: "Some PR", HTMLURL: "https://gh/acme/widget/pull/42", - RepoOwner: "acme", RepoName: "widget", - }, - } -} - -// setupReviewTaskTest builds a Service with a seeded workflow step (needed -// because createReviewTask runs shouldAutoStartStep after task creation). -func setupReviewTaskTest(t *testing.T) (*Service, *mockStepGetter) { - t.Helper() - repo := setupTestRepo(t) - stepGetter := newMockStepGetter() - stepGetter.steps["step1"] = &wfmodels.WorkflowStep{ - ID: "step1", WorkflowID: "wf1", Name: "Step 1", Position: 0, - Events: wfmodels.StepEvents{}, // no auto-start action - } - return createTestService(repo, stepGetter, newMockTaskRepo()), stepGetter -} - -// TestCreateReviewTask_SkipsWhenAlreadyReserved is the regression test for the -// duplicate review-task bug: if another handler has already reserved the -// dedup slot for this PR, createReviewTask must NOT call CreateReviewTask. -func TestCreateReviewTask_SkipsWhenAlreadyReserved(t *testing.T) { - svc, _ := setupReviewTaskTest(t) - ghSvc := &mockGitHubService{reserveReturn: false} // reservation lost to concurrent handler - svc.SetGitHubService(ghSvc) - creator := &countingReviewTaskCreator{} - svc.SetReviewTaskCreator(creator) - - svc.createReviewTask(context.Background(), newReviewEvent()) - - if ghSvc.reserveCalls != 1 { - t.Errorf("expected 1 Reserve call, got %d", ghSvc.reserveCalls) - } - if creator.calls != 0 { - t.Errorf("expected CreateReviewTask NOT to be called when reservation lost, got %d calls", creator.calls) - } - if ghSvc.releaseCalls != 0 { - t.Errorf("expected no Release calls when reservation was never held, got %d", ghSvc.releaseCalls) - } -} - -// TestCreateReviewTask_ReservesThenAssignsTaskID verifies the happy path: -// Reserve -> CreateReviewTask -> AssignReviewPRTaskID. -func TestCreateReviewTask_ReservesThenAssignsTaskID(t *testing.T) { - svc, _ := setupReviewTaskTest(t) - ghSvc := &mockGitHubService{reserveReturn: true} - svc.SetGitHubService(ghSvc) - creator := &countingReviewTaskCreator{taskID: "task-999"} - svc.SetReviewTaskCreator(creator) - - svc.createReviewTask(context.Background(), newReviewEvent()) - - if ghSvc.reserveCalls != 1 { - t.Errorf("expected 1 Reserve call, got %d", ghSvc.reserveCalls) - } - if creator.calls != 1 { - t.Errorf("expected 1 CreateReviewTask call, got %d", creator.calls) - } - if ghSvc.assignCalls != 1 { - t.Errorf("expected 1 AssignReviewPRTaskID call, got %d", ghSvc.assignCalls) - } - if ghSvc.assignedTaskID != "task-999" { - t.Errorf("AssignReviewPRTaskID got taskID=%q, want %q", ghSvc.assignedTaskID, "task-999") - } - if ghSvc.releaseCalls != 0 { - t.Errorf("expected no Release calls on happy path, got %d", ghSvc.releaseCalls) - } -} - -// TestCreateReviewTask_ReleasesOnTaskCreationFailure verifies that a failed -// CreateReviewTask triggers a Release so the slot can be retried on the next -// poll instead of being permanently blocked by an orphan reservation. -func TestCreateReviewTask_ReleasesOnTaskCreationFailure(t *testing.T) { - svc, _ := setupReviewTaskTest(t) - ghSvc := &mockGitHubService{reserveReturn: true} - svc.SetGitHubService(ghSvc) - creator := &countingReviewTaskCreator{err: assertErrTaskCreate} - svc.SetReviewTaskCreator(creator) - - svc.createReviewTask(context.Background(), newReviewEvent()) - - if ghSvc.reserveCalls != 1 { - t.Errorf("expected 1 Reserve call, got %d", ghSvc.reserveCalls) - } - if creator.calls != 1 { - t.Errorf("expected 1 CreateReviewTask call, got %d", creator.calls) - } - if ghSvc.releaseCalls != 1 { - t.Errorf("expected 1 Release call after task-create failure, got %d", ghSvc.releaseCalls) - } - if ghSvc.assignCalls != 0 { - t.Errorf("expected no Assign call after failure, got %d", ghSvc.assignCalls) - } -} - -var assertErrTaskCreate = &taskCreateErr{"simulated task creation failure"} - -type taskCreateErr struct{ msg string } - -func (e *taskCreateErr) Error() string { return e.msg } diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index 986a911bfe..a28c6dba61 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -200,6 +200,9 @@ type Service struct { // Review task creator for auto-creating tasks from review watch PRs reviewTaskCreator ReviewTaskCreator + // Issue task creator for auto-creating tasks from issue watch events + issueTaskCreator IssueTaskCreator + // Repository resolver for cloning + finding/creating repos for review tasks repositoryResolver RepositoryResolver diff --git a/apps/backend/pkg/websocket/actions.go b/apps/backend/pkg/websocket/actions.go index ee37e6c92c..71925f714d 100644 --- a/apps/backend/pkg/websocket/actions.go +++ b/apps/backend/pkg/websocket/actions.go @@ -368,6 +368,15 @@ const ( ActionGitHubTaskPRSync = "github.task_pr.sync" ActionGitHubStats = "github.stats" ActionGitHubCheckSessionPR = "github.check_session_pr" + + // Issue watch actions + ActionGitHubIssueWatchesList = "github.issue_watches.list" + ActionGitHubIssueWatchCreate = "github.issue_watches.create" + ActionGitHubIssueWatchUpdate = "github.issue_watches.update" + ActionGitHubIssueWatchDelete = "github.issue_watches.delete" + ActionGitHubIssueTrigger = "github.issue_watches.trigger" + ActionGitHubIssueTriggerAll = "github.issue_watches.trigger_all" + ActionGitHubNewIssueNotify = "github.new_issue.notify" // Notification ) // Error codes diff --git a/apps/web/components/github/github-settings.tsx b/apps/web/components/github/github-settings.tsx index 3ae57da6f4..0b2d222027 100644 --- a/apps/web/components/github/github-settings.tsx +++ b/apps/web/components/github/github-settings.tsx @@ -11,9 +11,12 @@ import { SettingsSection } from "@/components/settings/settings-section"; import { GitHubStatusCard } from "./github-status"; import { ReviewWatchTable } from "./review-watch-table"; import { ReviewWatchDialog } from "./review-watch-dialog"; +import { IssueWatchTable } from "./issue-watch-table"; +import { IssueWatchDialog } from "./issue-watch-dialog"; import { PRStatsPanel } from "./pr-stats"; import { useReviewWatches } from "@/hooks/domains/github/use-review-watches"; -import type { ReviewWatch } from "@/lib/types/github"; +import { useIssueWatches } from "@/hooks/domains/github/use-issue-watches"; +import type { ReviewWatch, IssueWatch } from "@/lib/types/github"; type GitHubSettingsProps = { workspaceId: string; @@ -73,19 +76,61 @@ function useWatchActions(workspaceId: string | null) { return { watches, create, update, handleDelete, handleTrigger, handleToggleEnabled }; } -export function GitHubSettings({ workspaceId }: GitHubSettingsProps) { - const { watches, create, update, handleDelete, handleTrigger, handleToggleEnabled } = - useWatchActions(workspaceId); +function useIssueWatchActions(workspaceId: string | null) { + const { items: watches, create, update, remove, trigger } = useIssueWatches(workspaceId); const { toast } = useToast(); - const [dialogOpen, setDialogOpen] = useState(false); - const [editingWatch, setEditingWatch] = useState(null); + const handleDelete = useCallback( + async (id: string) => { + try { + await remove(id); + toast({ description: "Issue watch deleted", variant: "success" }); + } catch { + toast({ description: "Failed to delete issue watch", variant: "error" }); + } + }, + [remove, toast], + ); - const handleEdit = useCallback((watch: ReviewWatch) => { - setEditingWatch(watch); - setDialogOpen(true); - }, []); + const handleTrigger = useCallback( + async (id: string) => { + try { + const result = await trigger(id); + const count = result?.new_issues_found ?? 0; + if (count > 0) { + toast({ + description: `Found ${count} new issue${count > 1 ? "s" : ""}`, + variant: "success", + }); + } else { + toast({ description: "No new issues found" }); + } + } catch { + toast({ description: "Failed to check for issues", variant: "error" }); + } + }, + [trigger, toast], + ); + const handleToggleEnabled = useCallback( + async (watch: IssueWatch) => { + try { + await update(watch.id, { enabled: !watch.enabled }); + toast({ + description: watch.enabled ? "Watch paused" : "Watch enabled", + variant: "success", + }); + } catch { + toast({ description: "Failed to update watch", variant: "error" }); + } + }, + [update, toast], + ); + + return { watches, create, update, handleDelete, handleTrigger, handleToggleEnabled }; +} + +export function GitHubSettings({ workspaceId }: GitHubSettingsProps) { return (
@@ -95,7 +140,7 @@ export function GitHubSettings({ workspaceId }: GitHubSettingsProps) { GitHub Integration

- Configure GitHub PR monitoring and review queue automation. + Configure GitHub PR monitoring, issue tracking, and review queue automation.

@@ -109,35 +154,8 @@ export function GitHubSettings({ workspaceId }: GitHubSettingsProps) { - { - setEditingWatch(null); - setDialogOpen(true); - }} - className="cursor-pointer" - > - - Add Watch - - } - > - - - - - - + + - - { - await create(req); - toast({ description: "Review watch created", variant: "success" }); - }} - onUpdate={async (id, req) => { - await update(id, req); - toast({ description: "Review watch updated", variant: "success" }); - }} - /> ); } + +function ReviewWatchSection({ workspaceId }: { workspaceId: string }) { + const { watches, create, update, handleDelete, handleTrigger, handleToggleEnabled } = + useWatchActions(workspaceId); + const { toast } = useToast(); + const [dialogOpen, setDialogOpen] = useState(false); + const [editingWatch, setEditingWatch] = useState(null); + + const handleEdit = useCallback((watch: ReviewWatch) => { + setEditingWatch(watch); + setDialogOpen(true); + }, []); + + return ( + <> + { + setEditingWatch(null); + setDialogOpen(true); + }} + className="cursor-pointer" + > + + Add Watch + + } + > + + + + + + + { + await create(req); + toast({ description: "Review watch created", variant: "success" }); + }} + onUpdate={async (id, req) => { + await update(id, req); + toast({ description: "Review watch updated", variant: "success" }); + }} + /> + + ); +} + +function IssueWatchSection({ workspaceId }: { workspaceId: string }) { + const issueActions = useIssueWatchActions(workspaceId); + const { toast } = useToast(); + const [dialogOpen, setDialogOpen] = useState(false); + const [editingWatch, setEditingIssueWatch] = useState(null); + + const handleEdit = useCallback((watch: IssueWatch) => { + setEditingIssueWatch(watch); + setDialogOpen(true); + }, []); + + return ( + <> + { + setEditingIssueWatch(null); + setDialogOpen(true); + }} + className="cursor-pointer" + > + + Add Watch + + } + > + + + + + + + { + await issueActions.create(req); + toast({ description: "Issue watch created", variant: "success" }); + }} + onUpdate={async (id, req) => { + await issueActions.update(id, req); + toast({ description: "Issue watch updated", variant: "success" }); + }} + /> + + ); +} diff --git a/apps/web/components/github/issue-task-icon.tsx b/apps/web/components/github/issue-task-icon.tsx new file mode 100644 index 0000000000..e4bea70dfc --- /dev/null +++ b/apps/web/components/github/issue-task-icon.tsx @@ -0,0 +1,27 @@ +"use client"; + +import { IconCircleDot } from "@tabler/icons-react"; +import { Tooltip, TooltipContent, TooltipTrigger } from "@kandev/ui/tooltip"; + +type IssueInfo = { url: string; number: number }; + +export function IssueTaskIcon({ issueInfo }: { issueInfo: IssueInfo }) { + return ( + + + e.stopPropagation()} + className="inline-flex items-center shrink-0 text-green-600 hover:text-green-500 cursor-pointer" + data-testid="issue-task-icon" + aria-label={`Issue #${issueInfo.number}`} + > + + + + Issue #{issueInfo.number} + + ); +} diff --git a/apps/web/components/github/issue-watch-dialog.tsx b/apps/web/components/github/issue-watch-dialog.tsx new file mode 100644 index 0000000000..25c0d3ae1c --- /dev/null +++ b/apps/web/components/github/issue-watch-dialog.tsx @@ -0,0 +1,489 @@ +"use client"; + +import { useState, useEffect, useCallback, useMemo } from "react"; +import { Button } from "@kandev/ui/button"; +import { Switch } from "@kandev/ui/switch"; +import { Label } from "@kandev/ui/label"; +import { Input } from "@kandev/ui/input"; +import { + Dialog, + DialogContent, + DialogHeader, + DialogTitle, + DialogDescription, + DialogFooter, +} from "@kandev/ui/dialog"; +import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@kandev/ui/select"; +import { Textarea } from "@kandev/ui/textarea"; +import { IconInfoCircle } from "@tabler/icons-react"; +import { Tooltip, TooltipContent, TooltipProvider, TooltipTrigger } from "@kandev/ui/tooltip"; +import { useAppStore } from "@/components/state-provider"; +import { useSettingsData } from "@/hooks/domains/settings/use-settings-data"; +import { useWorkflows } from "@/hooks/use-workflows"; +import { listWorkflowSteps } from "@/lib/api/domains/workflow-api"; +import { + ScriptEditor, + computeEditorHeight, +} from "@/components/settings/profile-edit/script-editor"; +import { + ISSUE_WATCH_PLACEHOLDERS, + DEFAULT_ISSUE_WATCH_PROMPT, +} from "@/components/github/issue-watch-placeholders"; +import { RepoFilterSelector } from "@/components/github/repo-filter-selector"; +import type { + RepoFilter, + IssueWatch, + CreateIssueWatchRequest, + UpdateIssueWatchRequest, +} from "@/lib/types/github"; + +type IssueWatchDialogProps = { + open: boolean; + onOpenChange: (open: boolean) => void; + watch: IssueWatch | null; + workspaceId: string; + onCreate: (req: CreateIssueWatchRequest) => Promise; + onUpdate: (id: string, req: UpdateIssueWatchRequest) => Promise; +}; + +type FormState = { + selectedRepos: RepoFilter[]; + allRepos: boolean; + workflowId: string; + workflowStepId: string; + agentProfileId: string; + executorProfileId: string; + prompt: string; + labels: string; + customQuery: string; + enabled: boolean; + pollInterval: number; +}; + +type WorkflowStepOption = { id: string; name: string }; + +const DEFAULT_QUERY = "type:issue state:open"; + +const defaultFormState: FormState = { + selectedRepos: [], + allRepos: false, + workflowId: "", + workflowStepId: "", + agentProfileId: "", + executorProfileId: "", + prompt: DEFAULT_ISSUE_WATCH_PROMPT, + labels: "", + customQuery: DEFAULT_QUERY, + enabled: true, + pollInterval: 300, +}; + +function formStateFromWatch(watch: IssueWatch): FormState { + const hasRepos = watch.repos && watch.repos.length > 0; + return { + selectedRepos: hasRepos ? watch.repos : [], + allRepos: !hasRepos, + workflowId: watch.workflow_id, + workflowStepId: watch.workflow_step_id, + agentProfileId: watch.agent_profile_id, + executorProfileId: watch.executor_profile_id, + prompt: watch.prompt || DEFAULT_ISSUE_WATCH_PROMPT, + labels: (watch.labels ?? []).join(", "), + customQuery: watch.custom_query || DEFAULT_QUERY, + enabled: watch.enabled, + pollInterval: watch.poll_interval_seconds, + }; +} + +function useWorkflowSteps(workflowId: string) { + const [steps, setSteps] = useState([]); + + useEffect(() => { + if (!workflowId) { + void Promise.resolve().then(() => setSteps([])); + return; + } + let cancelled = false; + listWorkflowSteps(workflowId) + .then((response) => { + if (cancelled) return; + const sorted = [...response.steps].sort((a, b) => a.position - b.position); + setSteps(sorted.map((s) => ({ id: s.id, name: s.name }))); + }) + .catch(() => { + if (!cancelled) setSteps([]); + }); + return () => { + cancelled = true; + }; + }, [workflowId]); + + return steps; +} + +function useWatchFormData(workspaceId: string) { + useSettingsData(true); + useWorkflows(workspaceId, true); + + const workflows = useAppStore((state) => state.workflows.items); + const agentProfiles = useAppStore((state) => state.agentProfiles.items); + const executors = useAppStore((state) => state.executors.items); + const allExecutorProfiles = useMemo( + () => executors.filter((e) => e.type !== "local").flatMap((e) => e.profiles ?? []), + [executors], + ); + + const filteredAgentProfiles = useMemo( + () => agentProfiles.filter((p) => !p.cli_passthrough), + [agentProfiles], + ); + + return { workflows, agentProfiles: filteredAgentProfiles, allExecutorProfiles }; +} + +function SectionHeader({ children }: { children: React.ReactNode }) { + return ( +
+ + {children} + +
+
+ ); +} + +function PlaceholdersHelp() { + return ( + + + + + + +

Available placeholders:

+
    + {ISSUE_WATCH_PLACEHOLDERS.map((p) => ( +
  • + {`{{${p.key}}}`}{" "} + {p.description} +
    • + ))} +
+ + + + ); +} + +function IssueWatchFormFields({ + form, + setForm, + workspaceId, + onAllReposChange, + onSelectedReposChange, +}: { + form: FormState; + setForm: React.Dispatch>; + workspaceId: string; + onAllReposChange: (checked: boolean) => void; + onSelectedReposChange: (repos: RepoFilter[]) => void; +}) { + return ( +
+ + + +
+ ); +} + +function IssueFilterFields({ + form, + setForm, + onAllReposChange, + onSelectedReposChange, +}: { + form: FormState; + setForm: React.Dispatch>; + onAllReposChange: (checked: boolean) => void; + onSelectedReposChange: (repos: RepoFilter[]) => void; +}) { + return ( + <> + Filter + +
+ + setForm((prev) => ({ ...prev, labels: e.target.value }))} + placeholder="e.g. bug, enhancement, priority:high" + /> +

+ Only match issues with these labels. Leave empty for all issues. +

+
+
+ +