diff --git a/apps/backend/cmd/kandev/helpers.go b/apps/backend/cmd/kandev/helpers.go index 1b2fe313cb..c09720b2d6 100644 --- a/apps/backend/cmd/kandev/helpers.go +++ b/apps/backend/cmd/kandev/helpers.go @@ -504,6 +504,22 @@ func registerRoutes(p routeParams) { // runCanceller but its container leaks because the cascade bypasses // Service.ArchiveTask's runAsyncTaskCleanup branch. handoffSvc.SetTaskResourceCleaner(p.taskSvc) + // Watch reset (Reset button on integration settings) cascade-deletes + // every task a watch previously created. The integrations re-use the + // shared HandoffService so the reset path goes through the same + // cleanup machinery as the regular delete-task surface. + if p.services.GitHub != nil { + p.services.GitHub.SetCascadeTaskDeleter(handoffSvc) + } + if p.services.Jira != nil { + p.services.Jira.SetTaskDeleter(handoffSvc) + } + if p.services.Linear != nil { + p.services.Linear.SetTaskDeleter(handoffSvc) + } + if p.services.Sentry != nil { + p.services.Sentry.SetTaskDeleter(handoffSvc) + } p.orchestratorSvc.SetWorkspaceMaterializer(handoffSvc) // Phase 8 prompt enrichment — wire the office scheduler's // TaskContextProvider so every run prompt rendered by the active diff --git a/apps/backend/internal/github/controller.go b/apps/backend/internal/github/controller.go index 451c959590..4a56345f4f 100644 --- a/apps/backend/internal/github/controller.go +++ b/apps/backend/internal/github/controller.go @@ -51,6 +51,8 @@ func (c *Controller) RegisterHTTPRoutes(router *gin.Engine) { api.PUT("/watches/review/:id", c.httpUpdateReviewWatch) api.DELETE("/watches/review/:id", c.httpDeleteReviewWatch) api.POST("/watches/review/:id/trigger", c.httpTriggerReviewWatch) + api.GET("/watches/review/:id/reset/preview", c.httpPreviewResetReviewWatch) + api.POST("/watches/review/:id/reset", c.httpResetReviewWatch) api.POST("/watches/review/trigger-all", c.httpTriggerAllReviewChecks) api.GET("/watches/issue", c.httpListIssueWatches) @@ -58,6 +60,8 @@ func (c *Controller) RegisterHTTPRoutes(router *gin.Engine) { api.PUT("/watches/issue/:id", c.httpUpdateIssueWatch) api.DELETE("/watches/issue/:id", c.httpDeleteIssueWatch) api.POST("/watches/issue/:id/trigger", c.httpTriggerIssueWatch) + api.GET("/watches/issue/:id/reset/preview", c.httpPreviewResetIssueWatch) + api.POST("/watches/issue/:id/reset", c.httpResetIssueWatch) api.POST("/watches/issue/trigger-all", c.httpTriggerAllIssueChecks) api.POST("/cleanup/review-tasks", c.httpCleanupReviewTasks) diff --git a/apps/backend/internal/github/controller_watch_reset.go b/apps/backend/internal/github/controller_watch_reset.go new file mode 100644 index 0000000000..e58c13a42e --- /dev/null +++ b/apps/backend/internal/github/controller_watch_reset.go @@ -0,0 +1,109 @@ +package github + +import ( + "net/http" + + "github.com/gin-gonic/gin" + "go.uber.org/zap" +) + +// httpPreviewResetReviewWatch returns the count of tasks a reset on the +// review watch would cascade-delete. Used by the confirmation dialog so the +// user sees "delete N task(s)" before they commit. +func (c *Controller) httpPreviewResetReviewWatch(ctx *gin.Context) { + id := ctx.Param("id") + if !c.requireReviewWatchInWorkspace(ctx, id) { + return + } + n, err := c.service.PreviewResetReviewWatch(ctx.Request.Context(), id) + if err != nil { + c.writeResetError(ctx, "preview reset review watch", err) + return + } + ctx.JSON(http.StatusOK, gin.H{"taskCount": n}) +} + +// httpResetReviewWatch executes the destructive reset: cascade-deletes +// every task previously created by the review watch (including archived), +// wipes its dedup table, and nulls last_polled_at so the next poll +// re-imports every currently-matching PR. +func (c *Controller) httpResetReviewWatch(ctx *gin.Context) { + id := ctx.Param("id") + if !c.requireReviewWatchInWorkspace(ctx, id) { + return + } + n, err := c.service.ResetReviewWatch(ctx.Request.Context(), id) + if err != nil { + c.writeResetError(ctx, "reset review watch", err) + return + } + ctx.JSON(http.StatusOK, gin.H{"tasksDeleted": n}) +} + +// httpPreviewResetIssueWatch returns the count of tasks a reset on the +// issue watch would cascade-delete. +func (c *Controller) httpPreviewResetIssueWatch(ctx *gin.Context) { + id := ctx.Param("id") + if !c.requireIssueWatchInWorkspace(ctx, id) { + return + } + n, err := c.service.PreviewResetIssueWatch(ctx.Request.Context(), id) + if err != nil { + c.writeResetError(ctx, "preview reset issue watch", err) + return + } + ctx.JSON(http.StatusOK, gin.H{"taskCount": n}) +} + +// httpResetIssueWatch executes the destructive reset for an issue watch. +func (c *Controller) httpResetIssueWatch(ctx *gin.Context) { + id := ctx.Param("id") + if !c.requireIssueWatchInWorkspace(ctx, id) { + return + } + n, err := c.service.ResetIssueWatch(ctx.Request.Context(), id) + if err != nil { + c.writeResetError(ctx, "reset issue watch", err) + return + } + ctx.JSON(http.StatusOK, gin.H{"tasksDeleted": n}) +} + +// requireReviewWatchInWorkspace guards the reset endpoints against IDOR: +// the caller must supply `?workspace_id=...` matching the watch's stored +// workspace. Mismatch and not-found both return 404 so a probing client +// can't tell whether a watch ID exists in another workspace. +func (c *Controller) requireReviewWatchInWorkspace(ctx *gin.Context, id string) bool { + watch, err := c.service.GetReviewWatch(ctx.Request.Context(), id) + if err != nil { + c.writeResetError(ctx, "load review watch", err) + return false + } + if watch == nil || ctx.Query("workspace_id") == "" || ctx.Query("workspace_id") != watch.WorkspaceID { + ctx.JSON(http.StatusNotFound, gin.H{"error": "review watch not found"}) + return false + } + return true +} + +// requireIssueWatchInWorkspace mirrors requireReviewWatchInWorkspace for +// the issue watch reset endpoints. +func (c *Controller) requireIssueWatchInWorkspace(ctx *gin.Context, id string) bool { + watch, err := c.service.GetIssueWatch(ctx.Request.Context(), id) + if err != nil { + c.writeResetError(ctx, "load issue watch", err) + return false + } + if watch == nil || ctx.Query("workspace_id") == "" || ctx.Query("workspace_id") != watch.WorkspaceID { + ctx.JSON(http.StatusNotFound, gin.H{"error": "issue watch not found"}) + return false + } + return true +} + +// writeResetError logs the raw error and returns a generic message to the +// client so internal store/SQL details don't leak through the HTTP response. +func (c *Controller) writeResetError(ctx *gin.Context, op string, err error) { + c.logger.Error(op, zap.Error(err)) + ctx.JSON(http.StatusInternalServerError, gin.H{"error": op + " failed"}) +} diff --git a/apps/backend/internal/github/errors.go b/apps/backend/internal/github/errors.go index d2f4ebafa0..ca74c08e82 100644 --- a/apps/backend/internal/github/errors.go +++ b/apps/backend/internal/github/errors.go @@ -43,6 +43,13 @@ var ErrInvalidToken = errors.New("invalid token") // transient upstream blip. var ErrRepoNotResolvable = errors.New("github: repository not resolvable") +// errStoreUnavailable is returned from service methods when no Store is +// wired (Provide can return a Service with store == nil when the SQLite +// repos aren't configured). Returning a typed error instead of nil- +// dereferencing keeps the watch reset handlers from panicking under that +// degenerate config — the caller surfaces a 5xx, the process keeps running. +var errStoreUnavailable = errors.New("github: store not configured") + // repoNotResolvableSubstrings matches the wire-format substrings GitHub's // GraphQL/REST APIs use for a deterministic "repository does not exist / // not accessible" outcome: diff --git a/apps/backend/internal/github/service.go b/apps/backend/internal/github/service.go index e3e6c297d7..82a5c75b52 100644 --- a/apps/backend/internal/github/service.go +++ b/apps/backend/internal/github/service.go @@ -10,6 +10,7 @@ import ( "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/events/bus" + "github.com/kandev/kandev/internal/watchreset" ) // Auth method constants. @@ -70,15 +71,21 @@ type SecretManager interface { // Service coordinates GitHub integration operations. type Service struct { - mu sync.Mutex - client Client - authMethod string - secrets SecretProvider - secretManager SecretManager - store *Store - eventBus bus.EventBus - logger *logger.Logger - taskDeleter TaskDeleter + mu sync.Mutex + client Client + authMethod string + secrets SecretProvider + secretManager SecretManager + store *Store + eventBus bus.EventBus + logger *logger.Logger + taskDeleter TaskDeleter + // cascadeTaskDeleter is the cascade-delete entry point used by the + // watch reset flow. It is distinct from taskDeleter (which only deletes + // a single task by ID) because reset must walk the task tree and clean + // up subtasks, runs, and worktrees too. Wired post-construction via + // SetCascadeTaskDeleter to avoid an import cycle with the task service. + cascadeTaskDeleter watchreset.TaskDeleter taskSessionChecker TaskSessionChecker syncGroup singleflight.Group taskEventSubs []bus.Subscription @@ -176,6 +183,16 @@ func (s *Service) newPATClient(token string) *PATClient { // SetTaskDeleter sets the task deletion dependency for cleanup operations. func (s *Service) SetTaskDeleter(d TaskDeleter) { s.taskDeleter = d } +// SetCascadeTaskDeleter wires the cascade-delete dependency used by the +// watch reset flow (ResetIssueWatch / ResetReviewWatch). Optional — when +// unset, reset returns an error so the missing wiring is surfaced instead +// of silently no-op'ing. +func (s *Service) SetCascadeTaskDeleter(d watchreset.TaskDeleter) { + s.mu.Lock() + defer s.mu.Unlock() + s.cascadeTaskDeleter = d +} + // SetTaskSessionChecker sets the session checker for cleanup operations. func (s *Service) SetTaskSessionChecker(c TaskSessionChecker) { s.taskSessionChecker = c } diff --git a/apps/backend/internal/github/service_watch_reset.go b/apps/backend/internal/github/service_watch_reset.go new file mode 100644 index 0000000000..5efb8173a9 --- /dev/null +++ b/apps/backend/internal/github/service_watch_reset.go @@ -0,0 +1,101 @@ +package github + +import ( + "context" + "errors" + + "github.com/kandev/kandev/internal/watchreset" +) + +// --- Issue watch reset --- + +// githubIssueWatchResetter is the watchreset.Resetter adapter for a single +// GitHub issue watch. Closes over the store and watch ID so the shared +// watchreset.Run helper stays integration-agnostic. +type githubIssueWatchResetter struct { + store *Store + watchID string +} + +func (r *githubIssueWatchResetter) ListTaskIDs(ctx context.Context) ([]string, error) { + return r.store.ListIssueWatchTaskIDsByWatch(ctx, r.watchID) +} + +func (r *githubIssueWatchResetter) Clear(ctx context.Context) error { + return r.store.ResetIssueWatchState(ctx, r.watchID) +} + +// PreviewResetIssueWatch returns how many tasks ResetIssueWatch would +// cascade-delete. Used by the frontend to populate the confirmation dialog. +func (s *Service) PreviewResetIssueWatch(ctx context.Context, watchID string) (int, error) { + if s.store == nil { + return 0, errStoreUnavailable + } + return watchreset.Preview(ctx, &githubIssueWatchResetter{store: s.store, watchID: watchID}) +} + +// ResetIssueWatch is destructive: cascade-deletes every task previously +// created by the issue watch (including archived), wipes the per-watch +// dedup rows, and nulls last_polled_at so the next poll re-imports every +// currently-matching issue. Returns the count of tasks deleted. +func (s *Service) ResetIssueWatch(ctx context.Context, watchID string) (int, error) { + if s.store == nil { + return 0, errStoreUnavailable + } + s.mu.Lock() + td := s.cascadeTaskDeleter + s.mu.Unlock() + if td == nil { + return 0, errors.New("github: cascade task deleter not wired; reset unavailable") + } + res, err := watchreset.Run(ctx, + &githubIssueWatchResetter{store: s.store, watchID: watchID}, + td, s.logger) + return res.TasksDeleted, err +} + +// --- Review watch reset --- + +// githubReviewWatchResetter is the watchreset.Resetter adapter for a single +// GitHub review watch. +type githubReviewWatchResetter struct { + store *Store + watchID string +} + +func (r *githubReviewWatchResetter) ListTaskIDs(ctx context.Context) ([]string, error) { + return r.store.ListReviewPRTaskIDsByWatch(ctx, r.watchID) +} + +func (r *githubReviewWatchResetter) Clear(ctx context.Context) error { + return r.store.ResetReviewWatchState(ctx, r.watchID) +} + +// PreviewResetReviewWatch returns how many tasks ResetReviewWatch would +// cascade-delete. Used by the frontend to populate the confirmation dialog. +func (s *Service) PreviewResetReviewWatch(ctx context.Context, watchID string) (int, error) { + if s.store == nil { + return 0, errStoreUnavailable + } + return watchreset.Preview(ctx, &githubReviewWatchResetter{store: s.store, watchID: watchID}) +} + +// ResetReviewWatch is destructive: cascade-deletes every task previously +// created by the review watch (including archived), wipes the per-watch +// dedup rows, and nulls last_polled_at so the next poll re-imports every +// currently-matching PR. Returns the count of tasks deleted. +func (s *Service) ResetReviewWatch(ctx context.Context, watchID string) (int, error) { + if s.store == nil { + return 0, errStoreUnavailable + } + s.mu.Lock() + td := s.cascadeTaskDeleter + s.mu.Unlock() + if td == nil { + return 0, errors.New("github: cascade task deleter not wired; reset unavailable") + } + res, err := watchreset.Run(ctx, + &githubReviewWatchResetter{store: s.store, watchID: watchID}, + td, s.logger) + return res.TasksDeleted, err +} diff --git a/apps/backend/internal/github/store.go b/apps/backend/internal/github/store.go index ddb5449ed8..062ee674ca 100644 --- a/apps/backend/internal/github/store.go +++ b/apps/backend/internal/github/store.go @@ -1236,6 +1236,38 @@ func (s *Store) DeleteReviewPRTask(ctx context.Context, id string) error { return err } +// ListReviewPRTaskIDsByWatch returns every task_id recorded against a +// review watch, including empty-string reservations. Used by the watch +// reset flow to enumerate the tasks to cascade-delete. +func (s *Store) ListReviewPRTaskIDsByWatch(ctx context.Context, watchID string) ([]string, error) { + var ids []string + err := s.ro.SelectContext(ctx, &ids, + `SELECT task_id FROM github_review_pr_tasks WHERE review_watch_id = ?`, watchID) + return ids, err +} + +// ResetReviewWatchState wipes a review watch's dedup rows and nulls its +// last_polled_at in a single transaction. Used by the reset flow after +// the cascade-delete loop so the next poll re-imports every currently +// matching PR as if the watch were freshly created. +func (s *Store) ResetReviewWatchState(ctx context.Context, watchID 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_review_pr_tasks WHERE review_watch_id = ?`, watchID); err != nil { + return err + } + if _, err := tx.ExecContext(ctx, + `UPDATE github_review_watches SET last_polled_at = NULL, updated_at = ? WHERE id = ?`, + time.Now().UTC(), watchID); err != nil { + return err + } + return tx.Commit() +} + // --- Stats queries --- // prStatsQuery builds parameterised SELECT queries against the github_task_prs table. @@ -1595,6 +1627,38 @@ func (s *Store) DeleteIssueWatchTask(ctx context.Context, id string) error { return err } +// ListIssueWatchTaskIDsByWatch returns every task_id recorded against an +// issue watch, including empty-string reservations. Used by the watch +// reset flow to enumerate the tasks to cascade-delete. +func (s *Store) ListIssueWatchTaskIDsByWatch(ctx context.Context, watchID string) ([]string, error) { + var ids []string + err := s.ro.SelectContext(ctx, &ids, + `SELECT task_id FROM github_issue_watch_tasks WHERE issue_watch_id = ?`, watchID) + return ids, err +} + +// ResetIssueWatchState wipes an issue watch's dedup rows and nulls its +// last_polled_at in a single transaction. Used by the reset flow after +// the cascade-delete loop so the next poll re-imports every currently +// matching issue as if the watch were freshly created. +func (s *Store) ResetIssueWatchState(ctx context.Context, watchID 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 = ?`, watchID); err != nil { + return err + } + if _, err := tx.ExecContext(ctx, + `UPDATE github_issue_watches SET last_polled_at = NULL, updated_at = ? WHERE id = ?`, + time.Now().UTC(), watchID); err != nil { + return err + } + return tx.Commit() +} + // --- Action preset operations --- // GetActionPresets returns stored PR/Issue presets for a workspace. Returns diff --git a/apps/backend/internal/github/store_watch_reset_test.go b/apps/backend/internal/github/store_watch_reset_test.go new file mode 100644 index 0000000000..4af1b75ca4 --- /dev/null +++ b/apps/backend/internal/github/store_watch_reset_test.go @@ -0,0 +1,148 @@ +package github + +import ( + "context" + "testing" + "time" +) + +// TestStore_ReviewWatch_ListTaskIDsAndReset pins the contract used by the +// review watch reset flow: every dedup row's task_id (including empty +// reservations) is enumerable, and ResetReviewWatchState wipes those rows +// atomically alongside last_polled_at. +func TestStore_ReviewWatch_ListTaskIDsAndReset(t *testing.T) { + store := newTestStore(t) + ctx := context.Background() + + rw := &ReviewWatch{ + WorkspaceID: "ws-1", + WorkflowID: "wf-1", + WorkflowStepID: "step-1", + AgentProfileID: "agent-1", + ExecutorProfileID: "exec-1", + ReviewScope: "user_and_teams", + Enabled: true, + } + if err := store.CreateReviewWatch(ctx, rw); err != nil { + t.Fatalf("create review watch: %v", err) + } + now := time.Now().UTC() + rw.LastPolledAt = &now + rw.UpdatedAt = now + if err := store.UpdateReviewWatch(ctx, rw); err != nil { + t.Fatalf("stamp last polled: %v", err) + } + + tasks := []*ReviewPRTask{ + {ReviewWatchID: rw.ID, RepoOwner: "acme", RepoName: "widget", PRNumber: 1, PRURL: "https://github.com/acme/widget/pull/1", TaskID: "task-a", CreatedAt: now}, + {ReviewWatchID: rw.ID, RepoOwner: "acme", RepoName: "widget", PRNumber: 2, PRURL: "https://github.com/acme/widget/pull/2", TaskID: "task-b", CreatedAt: now}, + {ReviewWatchID: rw.ID, RepoOwner: "acme", RepoName: "widget", PRNumber: 3, PRURL: "https://github.com/acme/widget/pull/3", TaskID: "", CreatedAt: now}, + } + for _, tp := range tasks { + if err := store.CreateReviewPRTask(ctx, tp); err != nil { + t.Fatalf("create review pr task: %v", err) + } + } + + ids, err := store.ListReviewPRTaskIDsByWatch(ctx, rw.ID) + if err != nil { + t.Fatalf("list ids: %v", err) + } + if len(ids) != 3 { + t.Fatalf("ListReviewPRTaskIDsByWatch returned %d rows, want 3 (including empty reservation)", len(ids)) + } + nonEmpty := 0 + for _, id := range ids { + if id != "" { + nonEmpty++ + } + } + if nonEmpty != 2 { + t.Errorf("expected 2 non-empty task IDs, got %d", nonEmpty) + } + + if err := store.ResetReviewWatchState(ctx, rw.ID); err != nil { + t.Fatalf("reset: %v", err) + } + + idsAfter, err := store.ListReviewPRTaskIDsByWatch(ctx, rw.ID) + if err != nil { + t.Fatalf("list ids after reset: %v", err) + } + if len(idsAfter) != 0 { + t.Errorf("expected 0 dedup rows after reset, got %d", len(idsAfter)) + } + got, err := store.GetReviewWatch(ctx, rw.ID) + if err != nil { + t.Fatalf("get watch: %v", err) + } + if got.LastPolledAt != nil { + t.Errorf("expected LastPolledAt to be nil after reset, got %v", got.LastPolledAt) + } +} + +// TestStore_IssueWatch_ListTaskIDsAndReset mirrors the review watch test for +// the issue watch reset path. +func TestStore_IssueWatch_ListTaskIDsAndReset(t *testing.T) { + store := newTestStore(t) + ctx := context.Background() + + iw := &IssueWatch{ + WorkspaceID: "ws-1", + WorkflowID: "wf-1", + WorkflowStepID: "step-1", + AgentProfileID: "agent-1", + ExecutorProfileID: "exec-1", + Enabled: true, + } + if err := store.CreateIssueWatch(ctx, iw); err != nil { + t.Fatalf("create issue watch: %v", err) + } + now := time.Now().UTC() + iw.LastPolledAt = &now + iw.UpdatedAt = now + if err := store.UpdateIssueWatch(ctx, iw); err != nil { + t.Fatalf("stamp last polled: %v", err) + } + + // Three reservations, only two get task IDs assigned so the third + // exercises the empty-ID inclusion the reset flow depends on. + for _, n := range []int{1, 2, 3} { + if _, err := store.ReserveIssueWatchTask(ctx, iw.ID, "acme", "widget", n, "https://github.com/acme/widget/issues/0"); err != nil { + t.Fatalf("reserve %d: %v", n, err) + } + } + if err := store.AssignIssueWatchTaskID(ctx, iw.ID, "acme", "widget", 1, "task-a"); err != nil { + t.Fatalf("assign 1: %v", err) + } + if err := store.AssignIssueWatchTaskID(ctx, iw.ID, "acme", "widget", 2, "task-b"); err != nil { + t.Fatalf("assign 2: %v", err) + } + + ids, err := store.ListIssueWatchTaskIDsByWatch(ctx, iw.ID) + if err != nil { + t.Fatalf("list ids: %v", err) + } + if len(ids) != 3 { + t.Fatalf("ListIssueWatchTaskIDsByWatch returned %d rows, want 3 (including empty reservation)", len(ids)) + } + + if err := store.ResetIssueWatchState(ctx, iw.ID); err != nil { + t.Fatalf("reset: %v", err) + } + + idsAfter, err := store.ListIssueWatchTaskIDsByWatch(ctx, iw.ID) + if err != nil { + t.Fatalf("list ids after reset: %v", err) + } + if len(idsAfter) != 0 { + t.Errorf("expected 0 dedup rows after reset, got %d", len(idsAfter)) + } + got, err := store.GetIssueWatch(ctx, iw.ID) + if err != nil { + t.Fatalf("get watch: %v", err) + } + if got.LastPolledAt != nil { + t.Errorf("expected LastPolledAt to be nil after reset, got %v", got.LastPolledAt) + } +} diff --git a/apps/backend/internal/jira/handlers.go b/apps/backend/internal/jira/handlers.go index c0449dd87b..73c0ef54c7 100644 --- a/apps/backend/internal/jira/handlers.go +++ b/apps/backend/internal/jira/handlers.go @@ -43,6 +43,8 @@ func (c *Controller) RegisterHTTPRoutes(router *gin.Engine) { api.PATCH("/watches/issue/:id", c.httpUpdateIssueWatch) api.DELETE("/watches/issue/:id", c.httpDeleteIssueWatch) api.POST("/watches/issue/:id/trigger", c.httpTriggerIssueWatch) + api.GET("/watches/issue/:id/reset/preview", c.httpPreviewResetIssueWatch) + api.POST("/watches/issue/:id/reset", c.httpResetIssueWatch) } // --- HTTP handlers --- @@ -242,6 +244,39 @@ func (c *Controller) httpTriggerIssueWatch(ctx *gin.Context) { ctx.JSON(http.StatusOK, gin.H{"newIssues": len(tickets)}) } +// httpPreviewResetIssueWatch returns the count of tasks that a reset on the +// watch would cascade-delete. Used by the confirmation dialog so the user +// sees "delete N task(s)" before they commit. +func (c *Controller) httpPreviewResetIssueWatch(ctx *gin.Context) { + id := ctx.Param("id") + if !c.assertWatchInWorkspace(ctx, id) { + return + } + n, err := c.service.PreviewResetIssueWatch(ctx.Request.Context(), id) + if err != nil { + c.writeIssueWatchError(ctx, err) + return + } + ctx.JSON(http.StatusOK, gin.H{"taskCount": n}) +} + +// httpResetIssueWatch executes the destructive reset: cascade-deletes all +// tasks previously created by the watch (including archived), wipes its +// dedup table, and nulls last_polled_at so the next poll re-imports +// everything currently matching the JQL. +func (c *Controller) httpResetIssueWatch(ctx *gin.Context) { + id := ctx.Param("id") + if !c.assertWatchInWorkspace(ctx, id) { + return + } + n, err := c.service.ResetIssueWatch(ctx.Request.Context(), id) + if err != nil { + c.writeIssueWatchError(ctx, err) + return + } + ctx.JSON(http.StatusOK, gin.H{"tasksDeleted": n}) +} + // assertWatchInWorkspace guards mutation/trigger endpoints against IDOR: the // caller must supply `?workspace_id=...` matching the watch's workspace. The // list/create endpoints already require this query; mirroring it here closes diff --git a/apps/backend/internal/jira/service.go b/apps/backend/internal/jira/service.go index 71ad13be8e..cb5f6f90e1 100644 --- a/apps/backend/internal/jira/service.go +++ b/apps/backend/internal/jira/service.go @@ -13,6 +13,7 @@ import ( "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/events" "github.com/kandev/kandev/internal/events/bus" + "github.com/kandev/kandev/internal/watchreset" ) // SecretStore is the subset of the secrets store the service needs. Kept small @@ -35,12 +36,26 @@ type Service struct { client Client // singleton, cleared on config change. probeHook func() eventBus bus.EventBus + // taskDeleter is the cascade-delete entry point used by the watch + // reset flow. Wired post-construction via SetTaskDeleter to avoid + // the import cycle that would result from passing the concrete + // *task.HandoffService into NewService. + taskDeleter watchreset.TaskDeleter // mockClient is non-nil only when Provide built the service with a MockClient // (KANDEV_MOCK_JIRA=true). Exposed via MockClient() so the e2e control routes // can drive the same instance the clientFn returns. mockClient *MockClient } +// SetTaskDeleter wires the cascade-delete dependency used by ResetIssueWatch. +// Optional — when unset, reset returns an error so the missing wiring is +// surfaced instead of silently no-op'ing. +func (s *Service) SetTaskDeleter(td watchreset.TaskDeleter) { + s.mu.Lock() + defer s.mu.Unlock() + s.taskDeleter = td +} + // MockClient returns the shared mock client when the service was built in mock // mode, or nil for production builds. func (s *Service) MockClient() *MockClient { @@ -581,6 +596,45 @@ func (s *Service) DeleteIssueWatch(ctx context.Context, id string) error { return s.store.DeleteIssueWatch(ctx, id) } +// jiraIssueWatchResetter is the watchreset.Resetter adapter for a single +// Jira issue watch. Closes over the store and watch ID so the shared +// watchreset.Run helper stays integration-agnostic. +type jiraIssueWatchResetter struct { + store *Store + watchID string +} + +func (r *jiraIssueWatchResetter) ListTaskIDs(ctx context.Context) ([]string, error) { + return r.store.ListIssueWatchTaskIDs(ctx, r.watchID) +} + +func (r *jiraIssueWatchResetter) Clear(ctx context.Context) error { + return r.store.ResetIssueWatchState(ctx, r.watchID) +} + +// PreviewResetIssueWatch returns how many tasks ResetIssueWatch would +// cascade-delete. Used by the frontend to populate the confirmation dialog. +func (s *Service) PreviewResetIssueWatch(ctx context.Context, watchID string) (int, error) { + return watchreset.Preview(ctx, &jiraIssueWatchResetter{store: s.store, watchID: watchID}) +} + +// ResetIssueWatch is destructive: cascade-deletes every task previously +// created by the watch (including archived), wipes the per-watch dedup +// rows, and nulls last_polled_at so the next poll re-imports every +// currently-matching ticket. Returns the count of tasks deleted. +func (s *Service) ResetIssueWatch(ctx context.Context, watchID string) (int, error) { + s.mu.Lock() + td := s.taskDeleter + s.mu.Unlock() + if td == nil { + return 0, errors.New("jira: task deleter not wired; reset unavailable") + } + res, err := watchreset.Run(ctx, + &jiraIssueWatchResetter{store: s.store, watchID: watchID}, + td, s.log) + return res.TasksDeleted, err +} + // CheckIssueWatch runs the watch's JQL once and returns the tickets that // haven't been turned into tasks yet. last_polled_at is stamped regardless of // whether the search succeeded — a failing search still counts as "we tried" diff --git a/apps/backend/internal/jira/store.go b/apps/backend/internal/jira/store.go index 5955fdee11..5b72311596 100644 --- a/apps/backend/internal/jira/store.go +++ b/apps/backend/internal/jira/store.go @@ -627,3 +627,35 @@ func (s *Store) ListSeenIssueKeys(ctx context.Context, watchID string) (map[stri } return out, nil } + +// ListIssueWatchTaskIDs returns every task_id recorded against a watch, +// including the empty-string sentinel reservations that never completed. +// Used by the reset flow to enumerate the tasks to cascade-delete. +func (s *Store) ListIssueWatchTaskIDs(ctx context.Context, watchID string) ([]string, error) { + var ids []string + err := s.ro.SelectContext(ctx, &ids, + `SELECT task_id FROM jira_issue_watch_tasks WHERE issue_watch_id = ?`, watchID) + return ids, err +} + +// ResetIssueWatchState wipes the watch's dedup rows and nulls its +// last_polled_at in one transaction. Used by the reset flow after the +// cascade-delete loop so the next poll re-imports every currently-matching +// ticket as if the watch were freshly created. +func (s *Store) ResetIssueWatchState(ctx context.Context, watchID 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 jira_issue_watch_tasks WHERE issue_watch_id = ?`, watchID); err != nil { + return err + } + if _, err := tx.ExecContext(ctx, + `UPDATE jira_issue_watches SET last_polled_at = NULL, updated_at = ? WHERE id = ?`, + time.Now().UTC(), watchID); err != nil { + return err + } + return tx.Commit() +} diff --git a/apps/backend/internal/jira/store_issue_watch_test.go b/apps/backend/internal/jira/store_issue_watch_test.go index 0359f73e7a..ea986fe3a6 100644 --- a/apps/backend/internal/jira/store_issue_watch_test.go +++ b/apps/backend/internal/jira/store_issue_watch_test.go @@ -267,6 +267,71 @@ func TestStore_IssueWatchTask_ListSeenIssueKeys(t *testing.T) { } } +func TestStore_IssueWatchTask_ListTaskIDsAndReset(t *testing.T) { + store := newTestStore(t) + ctx := context.Background() + + w := newTestIssueWatch("ws-1") + if err := store.CreateIssueWatch(ctx, w); err != nil { + t.Fatalf("create: %v", err) + } + // Stamp a last_polled_at so we can assert the reset clears it. + if err := store.UpdateIssueWatchLastPolled(ctx, w.ID, time.Now().UTC()); err != nil { + t.Fatalf("stamp last polled: %v", err) + } + // Reserve three slots; only two get assigned task IDs (mirrors the + // "task creation in flight" state the reset must handle). + for _, k := range []string{"PROJ-1", "PROJ-2", "PROJ-3"} { + if _, err := store.ReserveIssueWatchTask(ctx, w.ID, k, "https://x/"+k); err != nil { + t.Fatalf("reserve %s: %v", k, err) + } + } + if err := store.AssignIssueWatchTaskID(ctx, w.ID, "PROJ-1", "task-a"); err != nil { + t.Fatalf("assign 1: %v", err) + } + if err := store.AssignIssueWatchTaskID(ctx, w.ID, "PROJ-2", "task-b"); err != nil { + t.Fatalf("assign 2: %v", err) + } + + ids, err := store.ListIssueWatchTaskIDs(ctx, w.ID) + if err != nil { + t.Fatalf("list ids: %v", err) + } + if len(ids) != 3 { + t.Fatalf("ListIssueWatchTaskIDs returned %d rows, want 3 (including empty reservation)", len(ids)) + } + nonEmpty := 0 + for _, id := range ids { + if id != "" { + nonEmpty++ + } + } + if nonEmpty != 2 { + t.Errorf("expected 2 non-empty task IDs, got %d", nonEmpty) + } + + if err := store.ResetIssueWatchState(ctx, w.ID); err != nil { + t.Fatalf("reset: %v", err) + } + + // Dedup rows wiped. + idsAfter, err := store.ListIssueWatchTaskIDs(ctx, w.ID) + if err != nil { + t.Fatalf("list ids after reset: %v", err) + } + if len(idsAfter) != 0 { + t.Errorf("expected 0 dedup rows after reset, got %d", len(idsAfter)) + } + // last_polled_at cleared. + got, err := store.GetIssueWatch(ctx, w.ID) + if err != nil { + t.Fatalf("get watch: %v", err) + } + if got.LastPolledAt != nil { + t.Errorf("expected LastPolledAt to be nil after reset, got %v", got.LastPolledAt) + } +} + func TestStore_IssueWatchTask_Release(t *testing.T) { store := newTestStore(t) ctx := context.Background() diff --git a/apps/backend/internal/linear/handlers.go b/apps/backend/internal/linear/handlers.go index b9e2016004..fa2554231f 100644 --- a/apps/backend/internal/linear/handlers.go +++ b/apps/backend/internal/linear/handlers.go @@ -49,6 +49,8 @@ func (c *Controller) RegisterHTTPRoutes(router *gin.Engine) { api.PATCH("/watches/issue/:id", c.httpUpdateIssueWatch) api.DELETE("/watches/issue/:id", c.httpDeleteIssueWatch) api.POST("/watches/issue/:id/trigger", c.httpTriggerIssueWatch) + api.GET("/watches/issue/:id/reset/preview", c.httpPreviewResetIssueWatch) + api.POST("/watches/issue/:id/reset", c.httpResetIssueWatch) } // --- HTTP handlers --- diff --git a/apps/backend/internal/linear/handlers_issue_watch.go b/apps/backend/internal/linear/handlers_issue_watch.go index 2265cbe1a3..523e6adc77 100644 --- a/apps/backend/internal/linear/handlers_issue_watch.go +++ b/apps/backend/internal/linear/handlers_issue_watch.go @@ -97,6 +97,39 @@ func (c *Controller) httpTriggerIssueWatch(ctx *gin.Context) { ctx.JSON(http.StatusOK, gin.H{"newIssues": len(issues)}) } +// httpPreviewResetIssueWatch returns the count of tasks that a reset on the +// watch would cascade-delete. Used by the confirmation dialog so the user +// sees "delete N task(s)" before they commit. +func (c *Controller) httpPreviewResetIssueWatch(ctx *gin.Context) { + id := ctx.Param("id") + if !c.assertWatchInWorkspace(ctx, id) { + return + } + n, err := c.service.PreviewResetIssueWatch(ctx.Request.Context(), id) + if err != nil { + c.writeIssueWatchError(ctx, err) + return + } + ctx.JSON(http.StatusOK, gin.H{"taskCount": n}) +} + +// httpResetIssueWatch executes the destructive reset: cascade-deletes all +// tasks previously created by the watch (including archived), wipes its +// dedup table, and nulls last_polled_at so the next poll re-imports +// everything currently matching the filter. +func (c *Controller) httpResetIssueWatch(ctx *gin.Context) { + id := ctx.Param("id") + if !c.assertWatchInWorkspace(ctx, id) { + return + } + n, err := c.service.ResetIssueWatch(ctx.Request.Context(), id) + if err != nil { + c.writeIssueWatchError(ctx, err) + return + } + ctx.JSON(http.StatusOK, gin.H{"tasksDeleted": n}) +} + // assertWatchInWorkspace guards mutation/trigger endpoints against IDOR: the // caller must supply `?workspace_id=...` matching the watch's workspace. // Writes the response and returns false on mismatch. diff --git a/apps/backend/internal/linear/service.go b/apps/backend/internal/linear/service.go index 63f1f8acc9..4495a393dd 100644 --- a/apps/backend/internal/linear/service.go +++ b/apps/backend/internal/linear/service.go @@ -11,6 +11,7 @@ import ( "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/events/bus" + "github.com/kandev/kandev/internal/watchreset" ) // SecretStore is the subset of the secrets store the service needs. @@ -32,12 +33,25 @@ type Service struct { client Client // singleton, cleared on config change. probeHook func() eventBus bus.EventBus + // taskDeleter is the cascade-delete entry point used by ResetIssueWatch. + // Wired post-construction via SetTaskDeleter to avoid an import cycle + // with the task service. + taskDeleter watchreset.TaskDeleter // mockClient is non-nil only when Provide built the service with a MockClient // (KANDEV_MOCK_LINEAR=true). Exposed via MockClient() so the e2e control // routes can drive the same instance the clientFn returns. mockClient *MockClient } +// SetTaskDeleter wires the cascade-delete dependency used by ResetIssueWatch. +// Optional — when unset, reset returns an error so the missing wiring is +// surfaced instead of silently no-op'ing. +func (s *Service) SetTaskDeleter(td watchreset.TaskDeleter) { + s.mu.Lock() + defer s.mu.Unlock() + s.taskDeleter = td +} + // MockClient returns the shared mock client when the service was built in mock // mode, or nil for production builds. func (s *Service) MockClient() *MockClient { diff --git a/apps/backend/internal/linear/service_issue_watch.go b/apps/backend/internal/linear/service_issue_watch.go index 7d3526045f..93c648fd69 100644 --- a/apps/backend/internal/linear/service_issue_watch.go +++ b/apps/backend/internal/linear/service_issue_watch.go @@ -12,6 +12,7 @@ import ( "github.com/kandev/kandev/internal/events" "github.com/kandev/kandev/internal/events/bus" + "github.com/kandev/kandev/internal/watchreset" ) // ErrIssueWatchNotFound is returned when GetIssueWatch's caller looks up an ID @@ -108,6 +109,45 @@ func (s *Service) DeleteIssueWatch(ctx context.Context, id string) error { return s.store.DeleteIssueWatch(ctx, id) } +// linearIssueWatchResetter is the watchreset.Resetter adapter for a single +// Linear issue watch. Closes over the store and watch ID so the shared +// watchreset.Run helper stays integration-agnostic. +type linearIssueWatchResetter struct { + store *Store + watchID string +} + +func (r *linearIssueWatchResetter) ListTaskIDs(ctx context.Context) ([]string, error) { + return r.store.ListIssueWatchTaskIDs(ctx, r.watchID) +} + +func (r *linearIssueWatchResetter) Clear(ctx context.Context) error { + return r.store.ResetIssueWatchState(ctx, r.watchID) +} + +// PreviewResetIssueWatch returns how many tasks ResetIssueWatch would +// cascade-delete. Used by the frontend to populate the confirmation dialog. +func (s *Service) PreviewResetIssueWatch(ctx context.Context, watchID string) (int, error) { + return watchreset.Preview(ctx, &linearIssueWatchResetter{store: s.store, watchID: watchID}) +} + +// ResetIssueWatch is destructive: cascade-deletes every task previously +// created by the watch (including archived), wipes the per-watch dedup +// rows, and nulls last_polled_at so the next poll re-imports every +// currently-matching issue. Returns the count of tasks deleted. +func (s *Service) ResetIssueWatch(ctx context.Context, watchID string) (int, error) { + s.mu.Lock() + td := s.taskDeleter + s.mu.Unlock() + if td == nil { + return 0, errors.New("linear: task deleter not wired; reset unavailable") + } + res, err := watchreset.Run(ctx, + &linearIssueWatchResetter{store: s.store, watchID: watchID}, + td, s.log) + return res.TasksDeleted, err +} + // CheckIssueWatch runs the watch's filter once and returns the issues that // haven't been turned into tasks yet. last_polled_at is stamped regardless of // whether the search succeeded — a failing search still counts as "we tried". diff --git a/apps/backend/internal/linear/store_issue_watch.go b/apps/backend/internal/linear/store_issue_watch.go index e6398b2f10..ae9b4b58c0 100644 --- a/apps/backend/internal/linear/store_issue_watch.go +++ b/apps/backend/internal/linear/store_issue_watch.go @@ -317,3 +317,35 @@ func (s *Store) ListSeenIssueIdentifiers(ctx context.Context, watchID string) (m } return out, nil } + +// ListIssueWatchTaskIDs returns every task_id recorded against a watch, +// including the empty-string sentinel reservations that never completed. +// Used by the reset flow to enumerate the tasks to cascade-delete. +func (s *Store) ListIssueWatchTaskIDs(ctx context.Context, watchID string) ([]string, error) { + var ids []string + err := s.ro.SelectContext(ctx, &ids, + `SELECT task_id FROM linear_issue_watch_tasks WHERE issue_watch_id = ?`, watchID) + return ids, err +} + +// ResetIssueWatchState wipes the watch's dedup rows and nulls its +// last_polled_at in one transaction. Used by the reset flow after the +// cascade-delete loop so the next poll re-imports every currently-matching +// issue as if the watch were freshly created. +func (s *Store) ResetIssueWatchState(ctx context.Context, watchID 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 linear_issue_watch_tasks WHERE issue_watch_id = ?`, watchID); err != nil { + return err + } + if _, err := tx.ExecContext(ctx, + `UPDATE linear_issue_watches SET last_polled_at = NULL, updated_at = ? WHERE id = ?`, + time.Now().UTC(), watchID); err != nil { + return err + } + return tx.Commit() +} diff --git a/apps/backend/internal/linear/store_issue_watch_test.go b/apps/backend/internal/linear/store_issue_watch_test.go index 300e55e979..828b4678c1 100644 --- a/apps/backend/internal/linear/store_issue_watch_test.go +++ b/apps/backend/internal/linear/store_issue_watch_test.go @@ -259,3 +259,65 @@ func TestStore_IssueWatchTask_Release(t *testing.T) { t.Error("expected reservation to succeed after release") } } + +func TestStore_IssueWatchTask_ListTaskIDsAndReset(t *testing.T) { + store := newTestStore(t) + ctx := context.Background() + + w := newTestIssueWatch("ws-1") + if err := store.CreateIssueWatch(ctx, w); err != nil { + t.Fatalf("create: %v", err) + } + if err := store.UpdateIssueWatchLastPolled(ctx, w.ID, time.Now().UTC()); err != nil { + t.Fatalf("stamp last polled: %v", err) + } + // Three reservations; only two get task IDs assigned so the third row + // exercises the empty-ID inclusion the reset flow depends on. + for _, k := range []string{"ENG-1", "ENG-2", "ENG-3"} { + if _, err := store.ReserveIssueWatchTask(ctx, w.ID, k, "https://linear.app/x/issue/"+k); err != nil { + t.Fatalf("reserve %s: %v", k, err) + } + } + if err := store.AssignIssueWatchTaskID(ctx, w.ID, "ENG-1", "task-a"); err != nil { + t.Fatalf("assign 1: %v", err) + } + if err := store.AssignIssueWatchTaskID(ctx, w.ID, "ENG-2", "task-b"); err != nil { + t.Fatalf("assign 2: %v", err) + } + + ids, err := store.ListIssueWatchTaskIDs(ctx, w.ID) + if err != nil { + t.Fatalf("list ids: %v", err) + } + if len(ids) != 3 { + t.Fatalf("ListIssueWatchTaskIDs returned %d rows, want 3 (including empty reservation)", len(ids)) + } + nonEmpty := 0 + for _, id := range ids { + if id != "" { + nonEmpty++ + } + } + if nonEmpty != 2 { + t.Errorf("expected 2 non-empty task IDs, got %d", nonEmpty) + } + + if err := store.ResetIssueWatchState(ctx, w.ID); err != nil { + t.Fatalf("reset: %v", err) + } + + idsAfter, err := store.ListIssueWatchTaskIDs(ctx, w.ID) + if err != nil { + t.Fatalf("list ids after reset: %v", err) + } + if len(idsAfter) != 0 { + t.Errorf("expected 0 dedup rows after reset, got %d", len(idsAfter)) + } + got, err := store.GetIssueWatch(ctx, w.ID) + if err != nil { + t.Fatalf("get watch: %v", err) + } + if got.LastPolledAt != nil { + t.Errorf("expected LastPolledAt to be nil after reset, got %v", got.LastPolledAt) + } +} diff --git a/apps/backend/internal/sentry/handlers_issue_watch.go b/apps/backend/internal/sentry/handlers_issue_watch.go index 331ac71e6f..53fea34989 100644 --- a/apps/backend/internal/sentry/handlers_issue_watch.go +++ b/apps/backend/internal/sentry/handlers_issue_watch.go @@ -16,6 +16,8 @@ func (c *Controller) registerIssueWatchRoutes(api *gin.RouterGroup) { api.PATCH("/watches/issue/:id", c.httpUpdateIssueWatch) api.DELETE("/watches/issue/:id", c.httpDeleteIssueWatch) api.POST("/watches/issue/:id/trigger", c.httpTriggerIssueWatch) + api.GET("/watches/issue/:id/reset/preview", c.httpPreviewResetIssueWatch) + api.POST("/watches/issue/:id/reset", c.httpResetIssueWatch) } // httpListIssueWatches returns watches scoped to one workspace when @@ -123,6 +125,39 @@ func (c *Controller) httpTriggerIssueWatch(ctx *gin.Context) { ctx.JSON(http.StatusOK, gin.H{"published": len(issues)}) } +// httpPreviewResetIssueWatch returns the count of tasks that a reset on the +// watch would cascade-delete. Used by the confirmation dialog so the user +// sees "delete N task(s)" before they commit. +func (c *Controller) httpPreviewResetIssueWatch(ctx *gin.Context) { + id := ctx.Param("id") + if !c.assertWatchInWorkspace(ctx, id) { + return + } + n, err := c.service.PreviewResetIssueWatch(ctx.Request.Context(), id) + if err != nil { + c.writeIssueWatchError(ctx, err) + return + } + ctx.JSON(http.StatusOK, gin.H{"taskCount": n}) +} + +// httpResetIssueWatch executes the destructive reset: cascade-deletes all +// tasks previously created by the watch (including archived), wipes its +// dedup table, and nulls last_polled_at so the next poll re-imports +// everything currently matching the filter. +func (c *Controller) httpResetIssueWatch(ctx *gin.Context) { + id := ctx.Param("id") + if !c.assertWatchInWorkspace(ctx, id) { + return + } + n, err := c.service.ResetIssueWatch(ctx.Request.Context(), id) + if err != nil { + c.writeIssueWatchError(ctx, err) + return + } + ctx.JSON(http.StatusOK, gin.H{"tasksDeleted": n}) +} + // assertWatchInWorkspace guards mutation/trigger endpoints against IDOR: the // caller must supply `?workspace_id=...` matching the watch's workspace. // Writes the response and returns false on mismatch. diff --git a/apps/backend/internal/sentry/service.go b/apps/backend/internal/sentry/service.go index 668813cbd1..412ed53e92 100644 --- a/apps/backend/internal/sentry/service.go +++ b/apps/backend/internal/sentry/service.go @@ -13,6 +13,7 @@ import ( "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/events/bus" + "github.com/kandev/kandev/internal/watchreset" ) // SecretStore is the subset of the secrets store the service needs. @@ -44,6 +45,19 @@ type Service struct { // NewSentryIssueEvent. Optional: when nil, observed issues are not // surfaced to the orchestrator. eventBus bus.EventBus + // taskDeleter is the cascade-delete entry point used by ResetIssueWatch. + // Wired post-construction via SetTaskDeleter to avoid an import cycle + // with the task service. + taskDeleter watchreset.TaskDeleter +} + +// SetTaskDeleter wires the cascade-delete dependency used by ResetIssueWatch. +// Optional — when unset, reset returns an error so the missing wiring is +// surfaced instead of silently no-op'ing. +func (s *Service) SetTaskDeleter(td watchreset.TaskDeleter) { + s.mu.Lock() + defer s.mu.Unlock() + s.taskDeleter = td } // MockClient returns the shared mock client when the service was built in mock diff --git a/apps/backend/internal/sentry/service_issue_watch.go b/apps/backend/internal/sentry/service_issue_watch.go index 62f2b75976..df362e956e 100644 --- a/apps/backend/internal/sentry/service_issue_watch.go +++ b/apps/backend/internal/sentry/service_issue_watch.go @@ -11,6 +11,7 @@ import ( "github.com/kandev/kandev/internal/events" "github.com/kandev/kandev/internal/events/bus" + "github.com/kandev/kandev/internal/watchreset" ) // ErrIssueWatchNotFound is returned when GetIssueWatch's caller looks up an ID @@ -104,6 +105,45 @@ func (s *Service) DeleteIssueWatch(ctx context.Context, id string) error { return s.store.DeleteIssueWatch(ctx, id) } +// sentryIssueWatchResetter is the watchreset.Resetter adapter for a single +// Sentry issue watch. Closes over the store and watch ID so the shared +// watchreset.Run helper stays integration-agnostic. +type sentryIssueWatchResetter struct { + store *Store + watchID string +} + +func (r *sentryIssueWatchResetter) ListTaskIDs(ctx context.Context) ([]string, error) { + return r.store.ListIssueWatchTaskIDs(ctx, r.watchID) +} + +func (r *sentryIssueWatchResetter) Clear(ctx context.Context) error { + return r.store.ResetIssueWatchState(ctx, r.watchID) +} + +// PreviewResetIssueWatch returns how many tasks ResetIssueWatch would +// cascade-delete. Used by the frontend to populate the confirmation dialog. +func (s *Service) PreviewResetIssueWatch(ctx context.Context, watchID string) (int, error) { + return watchreset.Preview(ctx, &sentryIssueWatchResetter{store: s.store, watchID: watchID}) +} + +// ResetIssueWatch is destructive: cascade-deletes every task previously +// created by the watch (including archived), wipes the per-watch dedup +// rows, and nulls last_polled_at so the next poll re-imports every +// currently-matching issue. Returns the count of tasks deleted. +func (s *Service) ResetIssueWatch(ctx context.Context, watchID string) (int, error) { + s.mu.Lock() + td := s.taskDeleter + s.mu.Unlock() + if td == nil { + return 0, errors.New("sentry: task deleter not wired; reset unavailable") + } + res, err := watchreset.Run(ctx, + &sentryIssueWatchResetter{store: s.store, watchID: watchID}, + td, s.log) + return res.TasksDeleted, err +} + // CheckIssueWatch runs the watch's filter once and returns the issues that // haven't been turned into tasks yet. last_polled_at is stamped regardless of // whether the search succeeded — a failing search still counts as "we tried". diff --git a/apps/backend/internal/sentry/store_issue_watch.go b/apps/backend/internal/sentry/store_issue_watch.go index 5c1c4cc97d..50928bed5d 100644 --- a/apps/backend/internal/sentry/store_issue_watch.go +++ b/apps/backend/internal/sentry/store_issue_watch.go @@ -316,3 +316,35 @@ func (s *Store) ListSeenIssueShortIDs(ctx context.Context, watchID string) (map[ } return out, nil } + +// ListIssueWatchTaskIDs returns every task_id recorded against a watch, +// including the empty-string sentinel reservations that never completed. +// Used by the reset flow to enumerate the tasks to cascade-delete. +func (s *Store) ListIssueWatchTaskIDs(ctx context.Context, watchID string) ([]string, error) { + var ids []string + err := s.ro.SelectContext(ctx, &ids, + `SELECT task_id FROM sentry_issue_watch_tasks WHERE issue_watch_id = ?`, watchID) + return ids, err +} + +// ResetIssueWatchState wipes the watch's dedup rows and nulls its +// last_polled_at in one transaction. Used by the reset flow after the +// cascade-delete loop so the next poll re-imports every currently-matching +// issue as if the watch were freshly created. +func (s *Store) ResetIssueWatchState(ctx context.Context, watchID 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 sentry_issue_watch_tasks WHERE issue_watch_id = ?`, watchID); err != nil { + return err + } + if _, err := tx.ExecContext(ctx, + `UPDATE sentry_issue_watches SET last_polled_at = NULL, updated_at = ? WHERE id = ?`, + time.Now().UTC(), watchID); err != nil { + return err + } + return tx.Commit() +} diff --git a/apps/backend/internal/sentry/store_issue_watch_test.go b/apps/backend/internal/sentry/store_issue_watch_test.go index a90e5a4aa0..4e84d4d50d 100644 --- a/apps/backend/internal/sentry/store_issue_watch_test.go +++ b/apps/backend/internal/sentry/store_issue_watch_test.go @@ -262,3 +262,65 @@ func TestStore_IssueWatchTask_Release(t *testing.T) { t.Error("expected reservation to succeed after release") } } + +func TestStore_IssueWatchTask_ListTaskIDsAndReset(t *testing.T) { + store := newTestStore(t) + ctx := context.Background() + + w := newTestIssueWatch("ws-1") + if err := store.CreateIssueWatch(ctx, w); err != nil { + t.Fatalf("create: %v", err) + } + if err := store.UpdateIssueWatchLastPolled(ctx, w.ID, time.Now().UTC()); err != nil { + t.Fatalf("stamp last polled: %v", err) + } + // Three reservations; only two get task IDs assigned so the third row + // exercises the empty-ID inclusion the reset flow depends on. + for _, k := range []string{"PROJ-1", "PROJ-2", "PROJ-3"} { + if _, err := store.ReserveIssueWatchTask(ctx, w.ID, k, "https://sentry.io/issues/"+k); err != nil { + t.Fatalf("reserve %s: %v", k, err) + } + } + if err := store.AssignIssueWatchTaskID(ctx, w.ID, "PROJ-1", "task-a"); err != nil { + t.Fatalf("assign 1: %v", err) + } + if err := store.AssignIssueWatchTaskID(ctx, w.ID, "PROJ-2", "task-b"); err != nil { + t.Fatalf("assign 2: %v", err) + } + + ids, err := store.ListIssueWatchTaskIDs(ctx, w.ID) + if err != nil { + t.Fatalf("list ids: %v", err) + } + if len(ids) != 3 { + t.Fatalf("ListIssueWatchTaskIDs returned %d rows, want 3 (including empty reservation)", len(ids)) + } + nonEmpty := 0 + for _, id := range ids { + if id != "" { + nonEmpty++ + } + } + if nonEmpty != 2 { + t.Errorf("expected 2 non-empty task IDs, got %d", nonEmpty) + } + + if err := store.ResetIssueWatchState(ctx, w.ID); err != nil { + t.Fatalf("reset: %v", err) + } + + idsAfter, err := store.ListIssueWatchTaskIDs(ctx, w.ID) + if err != nil { + t.Fatalf("list ids after reset: %v", err) + } + if len(idsAfter) != 0 { + t.Errorf("expected 0 dedup rows after reset, got %d", len(idsAfter)) + } + got, err := store.GetIssueWatch(ctx, w.ID) + if err != nil { + t.Fatalf("get watch: %v", err) + } + if got.LastPolledAt != nil { + t.Errorf("expected LastPolledAt to be nil after reset, got %v", got.LastPolledAt) + } +} diff --git a/apps/backend/internal/system/metrics/collector.go b/apps/backend/internal/system/metrics/collector.go index d9977462ba..034665047c 100644 --- a/apps/backend/internal/system/metrics/collector.go +++ b/apps/backend/internal/system/metrics/collector.go @@ -1,15 +1,9 @@ package metrics import ( - "bufio" "context" - "errors" "fmt" "math" - "os" - "path/filepath" - "strconv" - "strings" "sync" "time" ) @@ -81,10 +75,10 @@ func (c *Collector) sampleMetric(ctx context.Context, id string, diskPath string value, err := diskPercent(ctx, diskPath) return sample(id, "Disk", "%", value, err) case MetricCPUTemp: - value, err := cpuTemp(c.sysRoot) + value, err := c.cpuTempValue() return sample(id, "CPU temp", "C", value, err) case MetricIOLoad: - value, err := ioLoad(c.procRoot) + value, err := c.ioLoadValue() return sample(id, "Load avg", "", value, err) default: return MetricSample{ID: id, Label: id, Available: false, Error: "unknown metric"} @@ -146,7 +140,7 @@ type cpuTimes struct { } func (c *Collector) cpuPercent() (float64, error) { - current, err := readProcStat(filepath.Join(c.procRoot, "stat")) + current, err := readCPUTimes(c) if err != nil { return 0, err } @@ -165,30 +159,6 @@ func (c *Collector) cpuPercent() (float64, error) { return value, nil } -func readProcStat(path string) (cpuTimes, error) { - data, err := os.ReadFile(path) - if err != nil { - return cpuTimes{}, err - } - fields := strings.Fields(strings.SplitN(string(data), "\n", 2)[0]) - if len(fields) < 8 || fields[0] != "cpu" { - return cpuTimes{}, errors.New("invalid /proc/stat cpu line") - } - var total uint64 - var idle uint64 - for i, field := range fields[1:] { - v, err := strconv.ParseUint(field, 10, 64) - if err != nil { - return cpuTimes{}, err - } - total += v - if i == 3 || i == 4 { - idle += v - } - } - return cpuTimes{total: total, idle: idle}, nil -} - func calculateCPUPercent(prev, current cpuTimes) float64 { totalDelta := current.total - prev.total if totalDelta == 0 { @@ -198,101 +168,6 @@ func calculateCPUPercent(prev, current cpuTimes) float64 { return (1 - float64(idleDelta)/float64(totalDelta)) * 100 } -func (c *Collector) memoryPercent() (float64, error) { - if value, ok := c.cgroupMemoryPercent(); ok { - return value, nil - } - return memInfoPercent(filepath.Join(c.procRoot, "meminfo")) -} - -func (c *Collector) cgroupMemoryPercent() (float64, bool) { - usage, err := readUintFile(filepath.Join(c.cgroupRoot, "memory.current")) - if err != nil { - return 0, false - } - maxRaw, err := os.ReadFile(filepath.Join(c.cgroupRoot, "memory.max")) - if err != nil { - return 0, false - } - maxText := strings.TrimSpace(string(maxRaw)) - if maxText == "max" { - return 0, false - } - limit, err := strconv.ParseUint(maxText, 10, 64) - if err != nil || limit == 0 { - return 0, false - } - return float64(usage) / float64(limit) * 100, true -} - -func memInfoPercent(path string) (float64, error) { - data, err := os.ReadFile(path) - if err != nil { - return 0, err - } - values := map[string]uint64{} - scanner := bufio.NewScanner(strings.NewReader(string(data))) - for scanner.Scan() { - fields := strings.Fields(scanner.Text()) - if len(fields) < 2 { - continue - } - key := strings.TrimSuffix(fields[0], ":") - v, err := strconv.ParseUint(fields[1], 10, 64) - if err == nil { - values[key] = v - } - } - total := values["MemTotal"] - available := values["MemAvailable"] - if total == 0 { - return 0, errors.New("MemTotal missing") - } - if available == 0 { - available = values["MemFree"] - } - return (1 - float64(available)/float64(total)) * 100, nil -} - -func cpuTemp(sysRoot string) (float64, error) { - paths, err := filepath.Glob(filepath.Join(sysRoot, "class/thermal/thermal_zone*/temp")) - if err != nil || len(paths) == 0 { - return 0, errors.New("cpu temperature unavailable") - } - for _, path := range paths { - raw, err := readUintFile(path) - if err != nil { - continue - } - value := float64(raw) - if value > 1000 { - value /= 1000 - } - return value, nil - } - return 0, errors.New("cpu temperature unavailable") -} - -func ioLoad(procRoot string) (float64, error) { - data, err := os.ReadFile(filepath.Join(procRoot, "loadavg")) - if err != nil { - return 0, err - } - fields := strings.Fields(string(data)) - if len(fields) == 0 { - return 0, errors.New("loadavg missing") - } - return strconv.ParseFloat(fields[0], 64) -} - -func readUintFile(path string) (uint64, error) { - data, err := os.ReadFile(path) - if err != nil { - return 0, err - } - return strconv.ParseUint(strings.TrimSpace(string(data)), 10, 64) -} - func (c *Collector) SampleWithTimestamp(ctx context.Context, settings GlobalSettings) Snapshot { source := c.Sample(ctx, settings.Metrics, settings.BackendDiskPath) return Snapshot{ diff --git a/apps/backend/internal/system/metrics/collector_darwin.go b/apps/backend/internal/system/metrics/collector_darwin.go new file mode 100644 index 0000000000..7a6b847b1f --- /dev/null +++ b/apps/backend/internal/system/metrics/collector_darwin.go @@ -0,0 +1,119 @@ +//go:build darwin + +package metrics + +/* +#include +#include +#include +*/ +import "C" + +import ( + "encoding/binary" + "errors" + "fmt" + "os" + "unsafe" + + "golang.org/x/sys/unix" +) + +// readCPUTimes uses Mach host_statistics(HOST_CPU_LOAD_INFO). Darwin does not +// expose kern.cp_time the way BSDs do, so we go straight to Mach. The returned +// cpu_ticks array has CPU_STATE_MAX entries (USER, SYSTEM, IDLE, NICE) of +// natural_t (uint32). Wraparound is rare at 100 Hz (~497 days) and the delta +// arithmetic in calculateCPUPercent already tolerates it via uint64 promotion. +func readCPUTimes(_ *Collector) (cpuTimes, error) { + var info C.host_cpu_load_info_data_t + count := C.mach_msg_type_number_t(C.HOST_CPU_LOAD_INFO_COUNT) + rc := C.host_statistics( + C.mach_host_self(), + C.HOST_CPU_LOAD_INFO, + C.host_info_t(unsafe.Pointer(&info)), + &count, + ) + if rc != C.KERN_SUCCESS { + return cpuTimes{}, fmt.Errorf("host_statistics HOST_CPU_LOAD_INFO: kern_return %d", int(rc)) + } + var total, idle uint64 + for i := 0; i < int(C.CPU_STATE_MAX); i++ { + v := uint64(info.cpu_ticks[i]) + total += v + if i == int(C.CPU_STATE_IDLE) { + idle = v + } + } + return cpuTimes{total: total, idle: idle}, nil +} + +func (c *Collector) memoryPercent() (float64, error) { + return darwinMemoryPercent() +} + +func (c *Collector) cpuTempValue() (float64, error) { + return 0, errors.New("cpu temperature unavailable on darwin") +} + +func (c *Collector) ioLoadValue() (float64, error) { + return darwinLoadAvg1() +} + +// vm.loadavg returns struct loadavg { fixed_pt_t ldavg[3]; long fscale; }. +// On darwin amd64/arm64 long is 8 bytes; the struct is padded to 24 bytes. +func darwinLoadAvg1() (float64, error) { + raw, err := unix.SysctlRaw("vm.loadavg") + if err != nil { + return 0, fmt.Errorf("sysctl vm.loadavg: %w", err) + } + if len(raw) < 24 { + return 0, errors.New("sysctl vm.loadavg response too short") + } + ld0 := binary.LittleEndian.Uint32(raw[0:4]) + fscale := binary.LittleEndian.Uint64(raw[16:24]) + if fscale == 0 { + return 0, errors.New("sysctl vm.loadavg fscale is zero") + } + return float64(ld0) / float64(fscale), nil +} + +// darwinMemoryPercent treats (free + inactive + speculative) pages as +// available, matching Activity Monitor's "Memory Used / Total" approximation. +// Wired, active, and compressor pages count as used. +func darwinMemoryPercent() (float64, error) { + total, err := unix.SysctlUint64("hw.memsize") + if err != nil { + return 0, fmt.Errorf("sysctl hw.memsize: %w", err) + } + if total == 0 { + return 0, errors.New("hw.memsize is zero") + } + pageSize := uint64(os.Getpagesize()) + if pageSize == 0 { + return 0, errors.New("page size is zero") + } + free, inactive, speculative, err := readVMStatistics64() + if err != nil { + return 0, err + } + available := (free + inactive + speculative) * pageSize + if available > total { + available = total + } + return (1.0 - float64(available)/float64(total)) * 100.0, nil +} + +func readVMStatistics64() (free, inactive, speculative uint64, err error) { + var vm C.vm_statistics64_data_t + count := C.mach_msg_type_number_t(C.HOST_VM_INFO64_COUNT) + rc := C.host_statistics64( + C.mach_host_self(), + C.HOST_VM_INFO64, + C.host_info64_t(unsafe.Pointer(&vm)), + &count, + ) + if rc != C.KERN_SUCCESS { + return 0, 0, 0, fmt.Errorf("host_statistics64 HOST_VM_INFO64: kern_return %d", int(rc)) + } + return uint64(vm.free_count), uint64(vm.inactive_count), uint64(vm.speculative_count), nil +} diff --git a/apps/backend/internal/system/metrics/collector_darwin_test.go b/apps/backend/internal/system/metrics/collector_darwin_test.go new file mode 100644 index 0000000000..bd79ecf480 --- /dev/null +++ b/apps/backend/internal/system/metrics/collector_darwin_test.go @@ -0,0 +1,60 @@ +//go:build darwin + +package metrics + +import ( + "testing" +) + +func TestDarwinReadCPUTimes(t *testing.T) { + got, err := readCPUTimes(NewCollector()) + if err != nil { + t.Fatalf("readCPUTimes: %v", err) + } + if got.total == 0 { + t.Fatal("expected total cpu ticks > 0") + } + if got.idle > got.total { + t.Fatalf("idle=%d > total=%d", got.idle, got.total) + } +} + +func TestDarwinCPUPercentNonNegative(t *testing.T) { + c := NewCollector() + if _, err := c.cpuPercent(); err != nil { + t.Fatalf("first cpuPercent (baseline): %v", err) + } + v, err := c.cpuPercent() + if err != nil { + t.Fatalf("second cpuPercent: %v", err) + } + if v < 0 || v > 100 { + t.Fatalf("cpu percent=%v, want 0..100", v) + } +} + +func TestDarwinLoadAvg1(t *testing.T) { + v, err := darwinLoadAvg1() + if err != nil { + t.Fatalf("darwinLoadAvg1: %v", err) + } + if v < 0 { + t.Fatalf("loadavg=%v, want >= 0", v) + } +} + +func TestDarwinMemoryPercent(t *testing.T) { + v, err := darwinMemoryPercent() + if err != nil { + t.Fatalf("darwinMemoryPercent: %v", err) + } + if v < 0 || v > 100 { + t.Fatalf("memory percent=%v, want 0..100", v) + } +} + +func TestDarwinCPUTempUnavailable(t *testing.T) { + if _, err := NewCollector().cpuTempValue(); err == nil { + t.Fatal("expected cpu temperature to be unavailable on darwin") + } +} diff --git a/apps/backend/internal/system/metrics/collector_linux.go b/apps/backend/internal/system/metrics/collector_linux.go new file mode 100644 index 0000000000..719f002779 --- /dev/null +++ b/apps/backend/internal/system/metrics/collector_linux.go @@ -0,0 +1,143 @@ +//go:build linux + +package metrics + +import ( + "bufio" + "errors" + "os" + "path/filepath" + "strconv" + "strings" +) + +func readCPUTimes(c *Collector) (cpuTimes, error) { + return readProcStat(filepath.Join(c.procRoot, "stat")) +} + +func (c *Collector) memoryPercent() (float64, error) { + if value, ok := c.cgroupMemoryPercent(); ok { + return value, nil + } + return memInfoPercent(filepath.Join(c.procRoot, "meminfo")) +} + +func (c *Collector) cpuTempValue() (float64, error) { + return cpuTemp(c.sysRoot) +} + +func (c *Collector) ioLoadValue() (float64, error) { + return ioLoad(c.procRoot) +} + +func readProcStat(path string) (cpuTimes, error) { + data, err := os.ReadFile(path) + if err != nil { + return cpuTimes{}, err + } + fields := strings.Fields(strings.SplitN(string(data), "\n", 2)[0]) + if len(fields) < 8 || fields[0] != "cpu" { + return cpuTimes{}, errors.New("invalid /proc/stat cpu line") + } + var total uint64 + var idle uint64 + for i, field := range fields[1:] { + v, err := strconv.ParseUint(field, 10, 64) + if err != nil { + return cpuTimes{}, err + } + total += v + if i == 3 || i == 4 { + idle += v + } + } + return cpuTimes{total: total, idle: idle}, nil +} + +func (c *Collector) cgroupMemoryPercent() (float64, bool) { + usage, err := readUintFile(filepath.Join(c.cgroupRoot, "memory.current")) + if err != nil { + return 0, false + } + maxRaw, err := os.ReadFile(filepath.Join(c.cgroupRoot, "memory.max")) + if err != nil { + return 0, false + } + maxText := strings.TrimSpace(string(maxRaw)) + if maxText == "max" { + return 0, false + } + limit, err := strconv.ParseUint(maxText, 10, 64) + if err != nil || limit == 0 { + return 0, false + } + return float64(usage) / float64(limit) * 100, true +} + +func memInfoPercent(path string) (float64, error) { + data, err := os.ReadFile(path) + if err != nil { + return 0, err + } + values := map[string]uint64{} + scanner := bufio.NewScanner(strings.NewReader(string(data))) + for scanner.Scan() { + fields := strings.Fields(scanner.Text()) + if len(fields) < 2 { + continue + } + key := strings.TrimSuffix(fields[0], ":") + v, err := strconv.ParseUint(fields[1], 10, 64) + if err == nil { + values[key] = v + } + } + total := values["MemTotal"] + if total == 0 { + return 0, errors.New("MemTotal missing") + } + available, ok := values["MemAvailable"] + if !ok { + available = values["MemFree"] + } + return (1 - float64(available)/float64(total)) * 100, nil +} + +func cpuTemp(sysRoot string) (float64, error) { + paths, err := filepath.Glob(filepath.Join(sysRoot, "class/thermal/thermal_zone*/temp")) + if err != nil || len(paths) == 0 { + return 0, errors.New("cpu temperature unavailable") + } + for _, path := range paths { + raw, err := readUintFile(path) + if err != nil { + continue + } + value := float64(raw) + if value > 1000 { + value /= 1000 + } + return value, nil + } + return 0, errors.New("cpu temperature unavailable") +} + +func ioLoad(procRoot string) (float64, error) { + data, err := os.ReadFile(filepath.Join(procRoot, "loadavg")) + if err != nil { + return 0, err + } + fields := strings.Fields(string(data)) + if len(fields) == 0 { + return 0, errors.New("loadavg missing") + } + return strconv.ParseFloat(fields[0], 64) +} + +func readUintFile(path string) (uint64, error) { + data, err := os.ReadFile(path) + if err != nil { + return 0, err + } + return strconv.ParseUint(strings.TrimSpace(string(data)), 10, 64) +} diff --git a/apps/backend/internal/system/metrics/collector_other.go b/apps/backend/internal/system/metrics/collector_other.go new file mode 100644 index 0000000000..91982fa980 --- /dev/null +++ b/apps/backend/internal/system/metrics/collector_other.go @@ -0,0 +1,21 @@ +//go:build !linux && !darwin && !windows + +package metrics + +import "errors" + +func readCPUTimes(_ *Collector) (cpuTimes, error) { + return cpuTimes{}, errors.New("cpu metrics unavailable on this platform") +} + +func (c *Collector) memoryPercent() (float64, error) { + return 0, errors.New("memory metrics unavailable on this platform") +} + +func (c *Collector) cpuTempValue() (float64, error) { + return 0, errors.New("cpu temperature unavailable on this platform") +} + +func (c *Collector) ioLoadValue() (float64, error) { + return 0, errors.New("load average unavailable on this platform") +} diff --git a/apps/backend/internal/system/metrics/collector_windows.go b/apps/backend/internal/system/metrics/collector_windows.go new file mode 100644 index 0000000000..3925c3cc20 --- /dev/null +++ b/apps/backend/internal/system/metrics/collector_windows.go @@ -0,0 +1,76 @@ +//go:build windows + +package metrics + +import ( + "errors" + "fmt" + "unsafe" + + "golang.org/x/sys/windows" +) + +var ( + kernel32 = windows.NewLazySystemDLL("kernel32.dll") + procGetSystemTimes = kernel32.NewProc("GetSystemTimes") + procGlobalMemoryStatusEx = kernel32.NewProc("GlobalMemoryStatusEx") +) + +// memoryStatusEx mirrors MEMORYSTATUSEX from windows.h. The first field is the +// struct length in bytes and must be set by the caller before invoking +// GlobalMemoryStatusEx. +type memoryStatusEx struct { + Length uint32 + MemoryLoad uint32 + TotalPhys uint64 + AvailPhys uint64 + TotalPageFile uint64 + AvailPageFile uint64 + TotalVirtual uint64 + AvailVirtual uint64 + AvailExtendedVirtual uint64 +} + +// readCPUTimes uses GetSystemTimes. On Windows kernelTime includes idleTime, +// so total non-idle = (kernel + user) - idle and we return total = kernel + user. +func readCPUTimes(_ *Collector) (cpuTimes, error) { + var idle, kernel, user windows.Filetime + r1, _, e := procGetSystemTimes.Call( + uintptr(unsafe.Pointer(&idle)), + uintptr(unsafe.Pointer(&kernel)), + uintptr(unsafe.Pointer(&user)), + ) + if r1 == 0 { + return cpuTimes{}, fmt.Errorf("GetSystemTimes: %w", e) + } + return cpuTimes{ + total: filetimeToUint64(kernel) + filetimeToUint64(user), + idle: filetimeToUint64(idle), + }, nil +} + +func filetimeToUint64(ft windows.Filetime) uint64 { + return (uint64(ft.HighDateTime) << 32) | uint64(ft.LowDateTime) +} + +func (c *Collector) memoryPercent() (float64, error) { + var status memoryStatusEx + status.Length = uint32(unsafe.Sizeof(status)) + r1, _, e := procGlobalMemoryStatusEx.Call(uintptr(unsafe.Pointer(&status))) + if r1 == 0 { + return 0, fmt.Errorf("GlobalMemoryStatusEx: %w", e) + } + if status.TotalPhys == 0 { + return 0, errors.New("total physical memory is zero") + } + used := status.TotalPhys - status.AvailPhys + return float64(used) / float64(status.TotalPhys) * 100, nil +} + +func (c *Collector) cpuTempValue() (float64, error) { + return 0, errors.New("cpu temperature unavailable on windows") +} + +func (c *Collector) ioLoadValue() (float64, error) { + return 0, errors.New("load average unavailable on windows") +} diff --git a/apps/backend/internal/system/metrics/collector_windows_test.go b/apps/backend/internal/system/metrics/collector_windows_test.go new file mode 100644 index 0000000000..2e5c19c42e --- /dev/null +++ b/apps/backend/internal/system/metrics/collector_windows_test.go @@ -0,0 +1,54 @@ +//go:build windows + +package metrics + +import "testing" + +func TestWindowsReadCPUTimes(t *testing.T) { + got, err := readCPUTimes(NewCollector()) + if err != nil { + t.Fatalf("readCPUTimes: %v", err) + } + if got.total == 0 { + t.Fatal("expected total cpu time > 0") + } + if got.idle > got.total { + t.Fatalf("idle=%d > total=%d", got.idle, got.total) + } +} + +func TestWindowsCPUPercentNonNegative(t *testing.T) { + c := NewCollector() + if _, err := c.cpuPercent(); err != nil { + t.Fatalf("first cpuPercent (baseline): %v", err) + } + v, err := c.cpuPercent() + if err != nil { + t.Fatalf("second cpuPercent: %v", err) + } + if v < 0 || v > 100 { + t.Fatalf("cpu percent=%v, want 0..100", v) + } +} + +func TestWindowsMemoryPercent(t *testing.T) { + v, err := NewCollector().memoryPercent() + if err != nil { + t.Fatalf("memoryPercent: %v", err) + } + if v < 0 || v > 100 { + t.Fatalf("memory percent=%v, want 0..100", v) + } +} + +func TestWindowsLoadAvgUnavailable(t *testing.T) { + if _, err := NewCollector().ioLoadValue(); err == nil { + t.Fatal("expected load average to be unavailable on windows") + } +} + +func TestWindowsCPUTempUnavailable(t *testing.T) { + if _, err := NewCollector().cpuTempValue(); err == nil { + t.Fatal("expected cpu temperature to be unavailable on windows") + } +} diff --git a/apps/backend/internal/system/metrics/disk_windows.go b/apps/backend/internal/system/metrics/disk_windows.go index f538414d99..80e44a5b14 100644 --- a/apps/backend/internal/system/metrics/disk_windows.go +++ b/apps/backend/internal/system/metrics/disk_windows.go @@ -5,8 +5,55 @@ package metrics import ( "context" "errors" + "fmt" + "time" + + "golang.org/x/sys/windows" ) -func diskPercent(_ context.Context, _ string) (float64, error) { - return 0, errors.New("disk usage unavailable on windows") +const diskQueryTimeout = 2 * time.Second + +// getDiskFreeSpaceEx is a package-level seam so tests can substitute a fake. +var getDiskFreeSpaceEx = windows.GetDiskFreeSpaceEx + +type diskQueryResult struct { + totalBytes uint64 + freeBytes uint64 + err error +} + +func diskPercent(ctx context.Context, path string) (float64, error) { + dir, err := windows.UTF16PtrFromString(path) + if err != nil { + return 0, fmt.Errorf("convert path %q: %w", path, err) + } + + result := make(chan diskQueryResult, 1) + go func() { + var freeToCaller, totalBytes, totalFree uint64 + callErr := getDiskFreeSpaceEx(dir, &freeToCaller, &totalBytes, &totalFree) + result <- diskQueryResult{totalBytes: totalBytes, freeBytes: totalFree, err: callErr} + }() + + timer := time.NewTimer(diskQueryTimeout) + defer timer.Stop() + + select { + case <-ctx.Done(): + return 0, ctx.Err() + case <-timer.C: + return 0, fmt.Errorf("disk usage timed out for %q", path) + case res := <-result: + if res.err != nil { + return 0, fmt.Errorf("GetDiskFreeSpaceEx %q: %w", path, res.err) + } + return diskPercentFromBytes(res.totalBytes, res.freeBytes) + } +} + +func diskPercentFromBytes(total, free uint64) (float64, error) { + if total == 0 { + return 0, errors.New("disk total is zero") + } + return (1 - float64(free)/float64(total)) * 100, nil } diff --git a/apps/backend/internal/system/metrics/disk_windows_test.go b/apps/backend/internal/system/metrics/disk_windows_test.go new file mode 100644 index 0000000000..4d7972585f --- /dev/null +++ b/apps/backend/internal/system/metrics/disk_windows_test.go @@ -0,0 +1,71 @@ +//go:build windows + +package metrics + +import ( + "context" + "errors" + "testing" + "time" +) + +func TestDiskPercentReturnsWhenContextCancelsWhileQueryBlocks(t *testing.T) { + original := getDiskFreeSpaceEx + block := make(chan struct{}) + started := make(chan struct{}) + getDiskFreeSpaceEx = func(_ *uint16, _ *uint64, _ *uint64, _ *uint64) error { + close(started) + <-block + return errors.New("unblocked") + } + t.Cleanup(func() { + close(block) + getDiskFreeSpaceEx = original + }) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { + _, err := diskPercent(ctx, `C:\`) + done <- err + }() + + select { + case <-started: + case <-time.After(time.Second): + t.Fatal("GetDiskFreeSpaceEx was not called") + } + cancel() + + select { + case err := <-done: + if !errors.Is(err, context.Canceled) { + t.Fatalf("diskPercent error = %v, want context.Canceled", err) + } + case <-time.After(500 * time.Millisecond): + t.Fatal("diskPercent did not return after context cancellation") + } +} + +func TestDiskPercentLive(t *testing.T) { + v, err := diskPercent(context.Background(), `C:\`) + if err != nil { + t.Fatalf("diskPercent: %v", err) + } + if v < 0 || v > 100 { + t.Fatalf("disk percent=%v, want 0..100", v) + } +} + +func TestDiskPercentFromBytes(t *testing.T) { + if _, err := diskPercentFromBytes(0, 0); err == nil { + t.Fatal("expected error when total is zero") + } + v, err := diskPercentFromBytes(100, 25) + if err != nil { + t.Fatalf("diskPercentFromBytes: %v", err) + } + if v != 75 { + t.Fatalf("got %v, want 75", v) + } +} diff --git a/apps/backend/internal/watchreset/reset.go b/apps/backend/internal/watchreset/reset.go new file mode 100644 index 0000000000..62f1d9c0ab --- /dev/null +++ b/apps/backend/internal/watchreset/reset.go @@ -0,0 +1,132 @@ +// Package watchreset orchestrates "reset" for integration watches +// (GitHub issue / GitHub review / Jira / Linear / Sentry). A reset is a +// destructive operation that: +// +// 1. Deletes every Kandev task previously created by the watch (cascade, +// including archived tasks — "wipe the slate"). +// 2. Wipes the per-watch dedup rows so the next poll re-considers every +// currently-matching external item. +// 3. Clears the watch's last_polled_at so it polls immediately. +// +// Each integration implements the small Resetter interface against its own +// store, and calls Run with a *task.HandoffService for the cascade delete. +package watchreset + +import ( + "context" + "fmt" + + "go.uber.org/zap" + + "github.com/kandev/kandev/internal/common/logger" + taskservice "github.com/kandev/kandev/internal/task/service" +) + +// Resetter is the per-watch surface watchreset needs. Implementations +// close over the watch ID and the integration's store. Methods are called +// in a fixed order by Run: ListTaskIDs first, then Clear after the cascade +// delete loop completes. +type Resetter interface { + // ListTaskIDs returns every task ID previously created by the watch, + // including rows whose task_id is empty (those are skipped by Run but + // the dedup row will still be wiped by Clear). + ListTaskIDs(ctx context.Context) ([]string, error) + // Clear wipes all dedup rows for the watch and nulls its + // last_polled_at column. Implementations should do both in a single + // transaction so a partial reset cannot leave the watch in a state + // where it skips items it just decided to forget about. + Clear(ctx context.Context) error +} + +// TaskDeleter is the cascade-delete entry point. Satisfied by +// *task.HandoffService.DeleteTaskTree. +type TaskDeleter interface { + DeleteTaskTree(ctx context.Context, rootID string, cascade bool) (*taskservice.CascadeOutcome, error) +} + +// Result reports what Run did. TasksDeleted counts task trees that were +// successfully cascade-deleted; tasks that failed to delete are logged but +// don't fail the reset (the dedup table is still wiped so the watch +// re-imports them on the next poll). +type Result struct { + TasksDeleted int +} + +// Run executes a watch reset. Deletes are best-effort: a delete failure +// for one task is logged and skipped so the dedup wipe still runs — a +// reset that fails halfway would leave the watch unable to re-import the +// items it just lost references to. Returns the final Result regardless; +// the only fatal error is the dedup/clear step itself. +// +// Once started, the reset runs to completion even if the caller's context +// is cancelled (e.g. the HTTP client disconnects mid-reset). A half-applied +// reset is worse than a slow one: the user already confirmed the +// destructive action, and the dedup wipe is the step that lets them retry +// safely. The cancel signal is preserved for deadlines that arrive +// before Run is invoked, but not for ones that fire after. +func Run(ctx context.Context, r Resetter, td TaskDeleter, log *logger.Logger) (Result, error) { + if r == nil { + return Result{}, fmt.Errorf("watchreset: nil Resetter") + } + if td == nil { + return Result{}, fmt.Errorf("watchreset: nil TaskDeleter") + } + // Respect a caller deadline that's already expired before we detach. + // WithoutCancel intentionally drops every signal, so swallowing an + // already-cancelled context here would silently run an unwanted reset. + if err := ctx.Err(); err != nil { + return Result{}, err + } + ctx = context.WithoutCancel(ctx) + + ids, err := r.ListTaskIDs(ctx) + if err != nil { + return Result{}, fmt.Errorf("list watch task ids: %w", err) + } + + deleted := 0 + for _, id := range ids { + if id == "" { + continue + } + if _, err := td.DeleteTaskTree(ctx, id, true); err != nil { + if log != nil { + log.Warn("watchreset: delete task failed", + zap.String("task_id", id), zap.Error(err)) + } + continue + } + deleted++ + } + + if err := r.Clear(ctx); err != nil { + return Result{TasksDeleted: deleted}, fmt.Errorf("clear watch state: %w", err) + } + + if log != nil { + log.Info("watch.reset", + zap.Int("tasks_deleted", deleted), + zap.Int("tasks_considered", len(ids))) + } + return Result{TasksDeleted: deleted}, nil +} + +// Preview returns the number of tasks that would be deleted by Run. It +// counts non-empty task IDs only — dedup rows whose task creation never +// completed don't have a Kandev task to delete. +func Preview(ctx context.Context, r Resetter) (int, error) { + if r == nil { + return 0, fmt.Errorf("watchreset: nil Resetter") + } + ids, err := r.ListTaskIDs(ctx) + if err != nil { + return 0, fmt.Errorf("list watch task ids: %w", err) + } + n := 0 + for _, id := range ids { + if id != "" { + n++ + } + } + return n, nil +} diff --git a/apps/backend/internal/watchreset/reset_test.go b/apps/backend/internal/watchreset/reset_test.go new file mode 100644 index 0000000000..67548f469e --- /dev/null +++ b/apps/backend/internal/watchreset/reset_test.go @@ -0,0 +1,133 @@ +package watchreset + +import ( + "context" + "errors" + "testing" + + taskservice "github.com/kandev/kandev/internal/task/service" +) + +type fakeResetter struct { + ids []string + listErr error + cleared bool + clearErr error +} + +func (f *fakeResetter) ListTaskIDs(_ context.Context) ([]string, error) { + return f.ids, f.listErr +} +func (f *fakeResetter) Clear(_ context.Context) error { + f.cleared = true + return f.clearErr +} + +type fakeDeleter struct { + called []string + failOn map[string]error + outcome *taskservice.CascadeOutcome +} + +func (f *fakeDeleter) DeleteTaskTree(_ context.Context, id string, _ bool) (*taskservice.CascadeOutcome, error) { + f.called = append(f.called, id) + if err, ok := f.failOn[id]; ok { + return nil, err + } + return f.outcome, nil +} + +func TestRun_DeletesEveryNonEmptyIDAndClears(t *testing.T) { + r := &fakeResetter{ids: []string{"a", "b", "", "c"}} + td := &fakeDeleter{} + res, err := Run(context.Background(), r, td, nil) + if err != nil { + t.Fatalf("Run: %v", err) + } + if res.TasksDeleted != 3 { + t.Fatalf("TasksDeleted=%d want 3", res.TasksDeleted) + } + if got := len(td.called); got != 3 { + t.Fatalf("DeleteTaskTree calls=%d want 3", got) + } + if !r.cleared { + t.Fatalf("Clear not called") + } +} + +func TestRun_ContinuesOnDeleteFailure(t *testing.T) { + r := &fakeResetter{ids: []string{"ok1", "bad", "ok2"}} + td := &fakeDeleter{failOn: map[string]error{"bad": errors.New("boom")}} + res, err := Run(context.Background(), r, td, nil) + if err != nil { + t.Fatalf("Run: %v", err) + } + if res.TasksDeleted != 2 { + t.Fatalf("TasksDeleted=%d want 2 (skip failing)", res.TasksDeleted) + } + if !r.cleared { + t.Fatalf("Clear must still run even when a delete failed") + } +} + +func TestRun_PropagatesListError(t *testing.T) { + r := &fakeResetter{listErr: errors.New("db down")} + td := &fakeDeleter{} + _, err := Run(context.Background(), r, td, nil) + if err == nil { + t.Fatalf("expected list error to surface") + } +} + +func TestRun_PropagatesClearError(t *testing.T) { + r := &fakeResetter{ids: []string{"a"}, clearErr: errors.New("write fail")} + td := &fakeDeleter{} + res, err := Run(context.Background(), r, td, nil) + if err == nil { + t.Fatalf("expected clear error to surface") + } + if res.TasksDeleted != 1 { + t.Fatalf("partial TasksDeleted should still be reported, got %d", res.TasksDeleted) + } +} + +func TestPreview_CountsOnlyNonEmptyIDs(t *testing.T) { + r := &fakeResetter{ids: []string{"", "a", "", "b", "c"}} + n, err := Preview(context.Background(), r) + if err != nil { + t.Fatalf("Preview: %v", err) + } + if n != 3 { + t.Fatalf("n=%d want 3", n) + } +} + +func TestRun_RejectsNilArgs(t *testing.T) { + if _, err := Run(context.Background(), nil, &fakeDeleter{}, nil); err == nil { + t.Fatalf("nil Resetter must error") + } + if _, err := Run(context.Background(), &fakeResetter{}, nil, nil); err == nil { + t.Fatalf("nil TaskDeleter must error") + } +} + +// TestRun_RejectsAlreadyCancelledContext pins the ctx.Err() pre-check: +// WithoutCancel intentionally drops every cancellation signal, so without +// this guard a caller passing an already-expired ctx (deadline missed +// before Run was even invoked) would silently get a destructive reset. +func TestRun_RejectsAlreadyCancelledContext(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + r := &fakeResetter{ids: []string{"a"}} + td := &fakeDeleter{} + _, err := Run(ctx, r, td, nil) + if err == nil { + t.Fatalf("expected error for already-cancelled context") + } + if r.cleared { + t.Fatalf("Clear must not run when context is already cancelled") + } + if len(td.called) != 0 { + t.Fatalf("DeleteTaskTree must not run when context is already cancelled, got %d calls", len(td.called)) + } +} diff --git a/apps/web/components/github/github-settings.tsx b/apps/web/components/github/github-settings.tsx index d6abbbf974..d6554a92f5 100644 --- a/apps/web/components/github/github-settings.tsx +++ b/apps/web/components/github/github-settings.tsx @@ -19,6 +19,7 @@ import { PRStatsPanel } from "./pr-stats"; import { useReviewWatches } from "@/hooks/domains/github/use-review-watches"; import { useIssueWatches } from "@/hooks/domains/github/use-issue-watches"; import { WorkspaceScopedSection } from "@/components/integrations/workspace-scoped-section"; +import { ResetWatchDialog, useWatchResetController } from "@/components/watches/reset-watch-dialog"; import { cleanupMergedReviewTasks, cleanupClosedIssueTasks } from "@/lib/api/domains/github-api"; import type { ReviewWatch, IssueWatch } from "@/lib/types/github"; @@ -66,7 +67,15 @@ function CleanupNowButton({ } function useWatchActions(workspaceId?: string | null) { - const { items: watches, create, update, remove, trigger } = useReviewWatches(workspaceId); + const { + items: watches, + create, + update, + remove, + trigger, + previewReset, + reset, + } = useReviewWatches(workspaceId); const { toast } = useToast(); const handleDelete = useCallback( @@ -116,11 +125,47 @@ function useWatchActions(workspaceId?: string | null) { [update, toast], ); - return { watches, create, update, handleDelete, handleTrigger, handleToggleEnabled }; + const handleReset = useCallback( + async (id: string, workspaceId: string) => { + try { + const { tasksDeleted } = await reset(id, workspaceId); + toast({ + description: + tasksDeleted > 0 + ? `Reset complete — deleted ${tasksDeleted} task(s); next poll will re-import.` + : "Reset complete — next poll will re-import matches.", + variant: "success", + }); + } catch { + toast({ description: "Failed to reset review watch", variant: "error" }); + throw new Error("reset failed"); + } + }, + [reset, toast], + ); + + return { + watches, + create, + update, + previewReset, + handleDelete, + handleTrigger, + handleReset, + handleToggleEnabled, + }; } function useIssueWatchActions(workspaceId?: string | null) { - const { items: watches, create, update, remove, trigger } = useIssueWatches(workspaceId); + const { + items: watches, + create, + update, + remove, + trigger, + previewReset, + reset, + } = useIssueWatches(workspaceId); const { toast } = useToast(); const handleDelete = useCallback( @@ -170,7 +215,35 @@ function useIssueWatchActions(workspaceId?: string | null) { [update, toast], ); - return { watches, create, update, handleDelete, handleTrigger, handleToggleEnabled }; + const handleReset = useCallback( + async (id: string, workspaceId: string) => { + try { + const { tasksDeleted } = await reset(id, workspaceId); + toast({ + description: + tasksDeleted > 0 + ? `Reset complete — deleted ${tasksDeleted} task(s); next poll will re-import.` + : "Reset complete — next poll will re-import matches.", + variant: "success", + }); + } catch { + toast({ description: "Failed to reset issue watch", variant: "error" }); + throw new Error("reset failed"); + } + }, + [reset, toast], + ); + + return { + watches, + create, + update, + previewReset, + handleDelete, + handleTrigger, + handleReset, + handleToggleEnabled, + }; } export function GitHubConnectionSection() { @@ -232,17 +305,38 @@ export function GitHubIntegrationPage() { // single flat table. Pass `undefined` to the hook so it fetches all rows; the // dialog's create flow asks the user to pick the workspace. function ReviewWatchSection() { - const { watches, create, update, handleDelete, handleTrigger, handleToggleEnabled } = - useWatchActions(undefined); + const { + watches, + create, + update, + previewReset, + handleDelete, + handleTrigger, + handleReset, + handleToggleEnabled, + } = useWatchActions(undefined); const { toast } = useToast(); const [dialogOpen, setDialogOpen] = useState(false); const [editingWatch, setEditingWatch] = useState(null); + const resetCtrl = useWatchResetController({ + preview: (w) => previewReset(w.id, w.workspace_id), + reset: (w) => handleReset(w.id, w.workspace_id), + }); const handleEdit = useCallback((watch: ReviewWatch) => { setEditingWatch(watch); setDialogOpen(true); }, []); + const { setResetting: setReviewResetting } = resetCtrl; + const onResetClick = useCallback( + (id: string) => { + const w = watches.find((item) => item.id === id); + if (w) setReviewResetting(w); + }, + [watches, setReviewResetting], + ); + return ( <> @@ -292,6 +387,15 @@ function ReviewWatchSection() { toast({ description: "Review watch updated", variant: "success" }); }} /> + {resetCtrl.resetting && ( + + )} ); } @@ -301,12 +405,25 @@ function IssueWatchSection() { const { toast } = useToast(); const [dialogOpen, setDialogOpen] = useState(false); const [editingWatch, setEditingIssueWatch] = useState(null); + const resetCtrl = useWatchResetController({ + preview: (w) => issueActions.previewReset(w.id, w.workspace_id), + reset: (w) => issueActions.handleReset(w.id, w.workspace_id), + }); const handleEdit = useCallback((watch: IssueWatch) => { setEditingIssueWatch(watch); setDialogOpen(true); }, []); + const { setResetting: setIssueResetting } = resetCtrl; + const onResetClick = useCallback( + (id: string) => { + const w = issueActions.watches.find((item) => item.id === id); + if (w) setIssueResetting(w); + }, + [issueActions.watches, setIssueResetting], + ); + return ( <> @@ -355,6 +473,15 @@ function IssueWatchSection() { toast({ description: "Issue watch updated", variant: "success" }); }} /> + {resetCtrl.resetting && ( + + )} ); } diff --git a/apps/web/components/github/issue-watch-table.tsx b/apps/web/components/github/issue-watch-table.tsx index 85d769024a..5cdd9ca7da 100644 --- a/apps/web/components/github/issue-watch-table.tsx +++ b/apps/web/components/github/issue-watch-table.tsx @@ -1,6 +1,12 @@ "use client"; -import { IconTrash, IconRefresh, IconPlayerPlay, IconPlayerPause } from "@tabler/icons-react"; +import { + IconTrash, + IconRefresh, + IconPlayerPlay, + IconPlayerPause, + IconRestore, +} from "@tabler/icons-react"; import { Button } from "@kandev/ui/button"; import { Badge } from "@kandev/ui/badge"; import { Table, TableBody, TableCell, TableHead, TableHeader, TableRow } from "@kandev/ui/table"; @@ -17,6 +23,7 @@ type IssueWatchTableProps = { onEdit: (watch: IssueWatch) => void; onDelete: (id: string) => void; onTrigger: (id: string) => void; + onReset: (id: string) => void; onToggleEnabled: (watch: IssueWatch) => void; }; @@ -47,10 +54,11 @@ type WatchActionsProps = { watch: IssueWatch; onToggleEnabled: (watch: IssueWatch) => void; onTrigger: (id: string) => void; + onReset: (id: string) => void; onDelete: (id: string) => void; }; -function WatchActions({ watch, onToggleEnabled, onTrigger, onDelete }: WatchActionsProps) { +function WatchActions({ watch, onToggleEnabled, onTrigger, onReset, onDelete }: WatchActionsProps) { const { toast } = useToast(); return (
@@ -91,6 +99,24 @@ function WatchActions({ watch, onToggleEnabled, onTrigger, onDelete }: WatchActi Check now + + + + + Reset + + + Reset + + + Reset + + + Reset + + + Reset +