From 83de4793fd11068562468715c530d3b0fb3a0680 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Salavisa?= Date: Tue, 5 May 2026 08:28:09 +0100 Subject: [PATCH] feat: add Linear issue watchers (#805) --- AGENTS.md | 6 +- apps/backend/cmd/kandev/main.go | 5 +- apps/backend/cmd/kandev/orchestrator.go | 19 + apps/backend/cmd/kandev/services.go | 8 +- .../prompts/linear-issue-watch-default.md | 22 + apps/backend/internal/events/types.go | 5 + apps/backend/internal/linear/handlers.go | 6 + .../internal/linear/handlers_issue_watch.go | 134 ++++ apps/backend/internal/linear/models.go | 81 +++ apps/backend/internal/linear/poller.go | 162 ++++- .../linear/poller_issue_watch_test.go | 213 +++++++ apps/backend/internal/linear/provider.go | 14 +- apps/backend/internal/linear/service.go | 2 + .../internal/linear/service_issue_watch.go | 266 ++++++++ .../linear/service_issue_watch_test.go | 225 +++++++ apps/backend/internal/linear/store.go | 29 + .../internal/linear/store_issue_watch.go | 267 ++++++++ .../internal/linear/store_issue_watch_test.go | 261 ++++++++ .../orchestrator/event_handlers_linear.go | 187 ++++++ .../event_handlers_linear_test.go | 43 ++ apps/backend/internal/orchestrator/service.go | 6 + .../linear/linear-issue-watch-dialog.tsx | 589 ++++++++++++++++++ .../linear/linear-issue-watch-placeholders.ts | 86 +++ .../linear/linear-issue-watch-table.tsx | 187 ++++++ .../linear/linear-issue-watchers-section.tsx | 177 ++++++ .../web/components/linear/linear-settings.tsx | 2 + .../linear/use-linear-issue-watches.ts | 100 +++ apps/web/lib/api/domains/linear-api.test.ts | 84 +++ apps/web/lib/api/domains/linear-api.ts | 65 ++ apps/web/lib/state/default-state.ts | 6 + apps/web/lib/state/slices/index.ts | 7 + .../state/slices/linear/linear-slice.test.ts | 96 +++ .../lib/state/slices/linear/linear-slice.ts | 48 ++ apps/web/lib/state/slices/linear/types.ts | 27 + apps/web/lib/state/store.ts | 21 + apps/web/lib/types/linear.ts | 46 ++ 36 files changed, 3473 insertions(+), 29 deletions(-) create mode 100644 apps/backend/config/prompts/linear-issue-watch-default.md create mode 100644 apps/backend/internal/linear/handlers_issue_watch.go create mode 100644 apps/backend/internal/linear/poller_issue_watch_test.go create mode 100644 apps/backend/internal/linear/service_issue_watch.go create mode 100644 apps/backend/internal/linear/service_issue_watch_test.go create mode 100644 apps/backend/internal/linear/store_issue_watch.go create mode 100644 apps/backend/internal/linear/store_issue_watch_test.go create mode 100644 apps/backend/internal/orchestrator/event_handlers_linear.go create mode 100644 apps/backend/internal/orchestrator/event_handlers_linear_test.go create mode 100644 apps/web/components/linear/linear-issue-watch-dialog.tsx create mode 100644 apps/web/components/linear/linear-issue-watch-placeholders.ts create mode 100644 apps/web/components/linear/linear-issue-watch-table.tsx create mode 100644 apps/web/components/linear/linear-issue-watchers-section.tsx create mode 100644 apps/web/hooks/domains/linear/use-linear-issue-watches.ts create mode 100644 apps/web/lib/state/slices/linear/linear-slice.test.ts create mode 100644 apps/web/lib/state/slices/linear/linear-slice.ts create mode 100644 apps/web/lib/state/slices/linear/types.ts diff --git a/AGENTS.md b/AGENTS.md index 42dcd27818..0ef8c64649 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -362,7 +362,7 @@ Skills use `gh` CLI by default. If a `gh` command fails (not installed, not auth Jira and Linear are the model: per-workspace credentials, a 90s auth-health poller, a settings page with status banner + reconnect CTA, link/import buttons that gate on availability. New integrations should reuse the shared shapes rather than copying either. **Backend** (`apps/backend/internal//`): -- Mirror the package layout: `service.go`, `store.go`, `client.go`, `provider.go`, `handlers.go`, `models.go`, `poller.go`. Expose `Provide(writer, reader *sqlx.DB, secrets SecretStore, log *logger.Logger) (*Service, func() error, error)` (add `eventBus bus.EventBus` only if the integration actually publishes events — Linear does not). +- Mirror the package layout: `service.go`, `store.go`, `client.go`, `provider.go`, `handlers.go`, `models.go`, `poller.go`. Expose `Provide(writer, reader *sqlx.DB, secrets SecretStore, eventBus bus.EventBus, log *logger.Logger) (*Service, func() error, error)`. Pass `nil` for `eventBus` when the integration doesn't publish events; both Jira and Linear take and use it for issue-watch publishing. - Use `internal/integrations/secretadapter` instead of writing your own upsert wrapper around `secrets.SecretStore`. The adapter satisfies any per-integration `SecretStore` interface shaped as `{Reveal, Set, Delete, Exists}`. - Use `internal/integrations/healthpoll` for the auth-health loop. Implement the `Prober` interface (`ListConfiguredWorkspaces` + `RecordAuthHealth`) on a small adapter and let `healthpoll.New("name", prober, log)` own Start/Stop/ticker. Keep integration-specific loops (JQL polling, webhook reconciliation, etc.) separate, like jira's issue-watch loop. - Wire the service via a per-domain `initService(...)` helper in `cmd/kandev/services.go`, not inline in `provideServices`. @@ -377,7 +377,7 @@ Jira and Linear are the model: per-workspace credentials, a 90s auth-health poll **Where Jira and Linear deliberately diverge:** - Issue model: Jira uses transitions + JQL; Linear uses state IDs + structured filters. Don't merge these schemas — the upstream APIs are genuinely different. -- Event publishing: only Jira accepts `eventBus` and emits `NewJiraIssueEvent` from its issue-watch loop. Linear has no equivalent feature today. +- Watch filter persistence: Jira stores the JQL string verbatim; Linear stores the structured `SearchFilter` as JSON in `filter_json` (Linear has no JQL equivalent). The orchestrator emits `NewJiraIssueEvent` / `NewLinearIssueEvent` respectively and dedups by issue key (Jira) vs identifier (Linear). - Health column extras: Linear's `linear_configs` row carries an `org_slug` captured from successful probes; Jira's row does not. --- @@ -388,4 +388,4 @@ This file is read by AI coding agents (Claude Code via `CLAUDE.md` symlink, Code --- -**Last Updated**: 2026-05-02 +**Last Updated**: 2026-05-03 diff --git a/apps/backend/cmd/kandev/main.go b/apps/backend/cmd/kandev/main.go index c1c3461663..f3cc9648f6 100644 --- a/apps/backend/cmd/kandev/main.go +++ b/apps/backend/cmd/kandev/main.go @@ -358,8 +358,11 @@ func startAgentInfrastructure( addCleanup(func() error { jiraPoller.Stop(); return nil }) } - // Start Linear auth-health poller, mirroring the Jira poller. + // Start Linear poller. Mirrors the Jira shape: auth-health probe plus an + // issue-watch loop that runs configured filters and emits + // NewLinearIssueEvent for the orchestrator to turn into tasks. if services.Linear != nil { + orchestratorSvc.SetLinearService(&linearServiceAdapter{svc: services.Linear}) linearPoller := linearpkg.NewPoller(services.Linear, log) linearPoller.Start(ctx) addCleanup(func() error { linearPoller.Stop(); return nil }) diff --git a/apps/backend/cmd/kandev/orchestrator.go b/apps/backend/cmd/kandev/orchestrator.go index 157a719dfa..0db11f75d7 100644 --- a/apps/backend/cmd/kandev/orchestrator.go +++ b/apps/backend/cmd/kandev/orchestrator.go @@ -17,6 +17,7 @@ import ( "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/events/bus" jirapkg "github.com/kandev/kandev/internal/jira" + linearpkg "github.com/kandev/kandev/internal/linear" "github.com/kandev/kandev/internal/orchestrator" "github.com/kandev/kandev/internal/repoclone" "github.com/kandev/kandev/internal/secrets" @@ -267,6 +268,24 @@ func (a *jiraServiceAdapter) ReleaseIssueWatchTask(ctx context.Context, watchID, return a.svc.Store().ReleaseIssueWatchTask(ctx, watchID, issueKey) } +// linearServiceAdapter exposes the Linear service's issue-watch dedup methods +// to the orchestrator without leaking the rest of the package surface area. +type linearServiceAdapter struct { + svc *linearpkg.Service +} + +func (a *linearServiceAdapter) ReserveIssueWatchTask(ctx context.Context, watchID, identifier, issueURL string) (bool, error) { + return a.svc.Store().ReserveIssueWatchTask(ctx, watchID, identifier, issueURL) +} + +func (a *linearServiceAdapter) AssignIssueWatchTaskID(ctx context.Context, watchID, identifier, taskID string) error { + return a.svc.Store().AssignIssueWatchTaskID(ctx, watchID, identifier, taskID) +} + +func (a *linearServiceAdapter) ReleaseIssueWatchTask(ctx context.Context, watchID, identifier string) error { + return a.svc.Store().ReleaseIssueWatchTask(ctx, watchID, identifier) +} + // repoLocalPathUpdater adapts the task service's UpdateRepository to the executor.RepoUpdater interface. type repoLocalPathUpdater struct { svc *taskservice.Service diff --git a/apps/backend/cmd/kandev/services.go b/apps/backend/cmd/kandev/services.go index 0df0e4f4c7..4261354217 100644 --- a/apps/backend/cmd/kandev/services.go +++ b/apps/backend/cmd/kandev/services.go @@ -86,7 +86,7 @@ func provideServices(cfg *config.Config, log *logger.Logger, repos *Repositories githubSvc := initGitHubService(dbPool, eventBus, repos.Secrets, log) jiraSvc := initJiraService(dbPool, eventBus, repos.Secrets, log) - linearSvc := initLinearService(dbPool, repos.Secrets, log) + linearSvc := initLinearService(dbPool, eventBus, repos.Secrets, log) return &Services{ Task: taskSvc, @@ -238,10 +238,8 @@ func initJiraService(dbPool *db.Pool, eventBus bus.EventBus, secretsStore secret } // initLinearService wires up the Linear integration. Failures are non-fatal. -// Linear doesn't publish events today so it doesn't take an eventBus — -// adding one would be speculative until a feature actually needs it. -func initLinearService(dbPool *db.Pool, secretsStore secrets.SecretStore, log *logger.Logger) *linear.Service { - svc, _, err := linear.Provide(dbPool.Writer(), dbPool.Reader(), secretadapter.New(secretsStore), log) +func initLinearService(dbPool *db.Pool, eventBus bus.EventBus, secretsStore secrets.SecretStore, log *logger.Logger) *linear.Service { + svc, _, err := linear.Provide(dbPool.Writer(), dbPool.Reader(), secretadapter.New(secretsStore), eventBus, log) if err != nil { log.Warn("Linear service initialization failed (non-fatal)", zap.Error(err)) } diff --git a/apps/backend/config/prompts/linear-issue-watch-default.md b/apps/backend/config/prompts/linear-issue-watch-default.md new file mode 100644 index 0000000000..451b3a6395 --- /dev/null +++ b/apps/backend/config/prompts/linear-issue-watch-default.md @@ -0,0 +1,22 @@ +You have been assigned a Linear issue to work on. + +**Issue:** {{issue.url}} +**Identifier:** {{issue.identifier}} +**Title:** {{issue.title}} +**Team:** {{issue.team}} +**State:** {{issue.state}} +**Priority:** {{issue.priority}} +**Assignee:** {{issue.assignee}} + +## Description + +{{issue.description}} + +## 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 {{issue.identifier}}. diff --git a/apps/backend/internal/events/types.go b/apps/backend/internal/events/types.go index 8b8ff9767d..750d6d595a 100644 --- a/apps/backend/internal/events/types.go +++ b/apps/backend/internal/events/types.go @@ -211,6 +211,11 @@ const ( JiraNewIssue = "jira.new_issue" // New issue found matching a Jira issue watch ) +// Event types for Linear integration +const ( + LinearNewIssue = "linear.new_issue" // New issue found matching a Linear issue watch +) + // BuildShellOutputSubject creates a shell output subject for a specific session func BuildShellOutputSubject(sessionID string) string { return ShellOutput + "." + sessionID diff --git a/apps/backend/internal/linear/handlers.go b/apps/backend/internal/linear/handlers.go index b119b775e7..e0ace4bc85 100644 --- a/apps/backend/internal/linear/handlers.go +++ b/apps/backend/internal/linear/handlers.go @@ -39,6 +39,12 @@ func (c *Controller) RegisterHTTPRoutes(router *gin.Engine) { api.GET("/issues", c.httpSearchIssues) api.GET("/issues/:id", c.httpGetIssue) api.POST("/issues/:id/state", c.httpSetIssueState) + + api.GET("/watches/issue", c.httpListIssueWatches) + api.POST("/watches/issue", c.httpCreateIssueWatch) + api.PATCH("/watches/issue/:id", c.httpUpdateIssueWatch) + api.DELETE("/watches/issue/:id", c.httpDeleteIssueWatch) + api.POST("/watches/issue/:id/trigger", c.httpTriggerIssueWatch) } // --- HTTP handlers --- diff --git a/apps/backend/internal/linear/handlers_issue_watch.go b/apps/backend/internal/linear/handlers_issue_watch.go new file mode 100644 index 0000000000..2265cbe1a3 --- /dev/null +++ b/apps/backend/internal/linear/handlers_issue_watch.go @@ -0,0 +1,134 @@ +package linear + +import ( + "errors" + "net/http" + + "github.com/gin-gonic/gin" +) + +// httpListIssueWatches returns watches scoped to one workspace when +// `workspace_id` is supplied, or every watch across all workspaces when it is +// absent (used by the install-wide settings page). +func (c *Controller) httpListIssueWatches(ctx *gin.Context) { + workspaceID := ctx.Query("workspace_id") + var ( + watches []*IssueWatch + err error + ) + if workspaceID == "" { + watches, err = c.service.ListAllIssueWatches(ctx.Request.Context()) + } else { + 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 + } + w, err := c.service.CreateIssueWatch(ctx.Request.Context(), &req) + if err != nil { + c.writeIssueWatchError(ctx, err) + return + } + ctx.JSON(http.StatusOK, w) +} + +func (c *Controller) httpUpdateIssueWatch(ctx *gin.Context) { + id := ctx.Param("id") + if !c.assertWatchInWorkspace(ctx, id) { + return + } + var req UpdateIssueWatchRequest + if err := ctx.ShouldBindJSON(&req); err != nil { + ctx.JSON(http.StatusBadRequest, gin.H{"error": "invalid payload"}) + return + } + w, err := c.service.UpdateIssueWatch(ctx.Request.Context(), id, &req) + if err != nil { + c.writeIssueWatchError(ctx, err) + return + } + ctx.JSON(http.StatusOK, w) +} + +func (c *Controller) httpDeleteIssueWatch(ctx *gin.Context) { + id := ctx.Param("id") + if !c.assertWatchInWorkspace(ctx, id) { + return + } + 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}) +} + +// httpTriggerIssueWatch runs a single immediate poll, ignoring the per-watch +// interval gating. Returns the count of newly-discovered issues so the user +// gets feedback without waiting for the orchestrator to fan tasks out. +func (c *Controller) httpTriggerIssueWatch(ctx *gin.Context) { + id := ctx.Param("id") + w, err := c.service.GetIssueWatch(ctx.Request.Context(), id) + if err != nil { + c.writeIssueWatchError(ctx, err) + return + } + if !workspaceMatches(ctx, w.WorkspaceID) { + ctx.JSON(http.StatusNotFound, gin.H{"error": ErrIssueWatchNotFound.Error()}) + return + } + issues, err := c.service.CheckIssueWatch(ctx.Request.Context(), w) + if err != nil { + c.writeClientError(ctx, err) + return + } + for _, issue := range issues { + c.service.publishNewLinearIssueEvent(ctx.Request.Context(), w, issue) + } + ctx.JSON(http.StatusOK, gin.H{"newIssues": len(issues)}) +} + +// 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. +func (c *Controller) assertWatchInWorkspace(ctx *gin.Context, id string) bool { + w, err := c.service.GetIssueWatch(ctx.Request.Context(), id) + if err != nil { + c.writeIssueWatchError(ctx, err) + return false + } + if !workspaceMatches(ctx, w.WorkspaceID) { + ctx.JSON(http.StatusNotFound, gin.H{"error": ErrIssueWatchNotFound.Error()}) + return false + } + return true +} + +// workspaceMatches returns true when the request's `workspace_id` query +// matches the resource's stored workspace. Empty query is rejected so callers +// can't bypass the check by omitting the parameter. +func workspaceMatches(ctx *gin.Context, resourceWorkspace string) bool { + q := ctx.Query("workspace_id") + return q != "" && q == resourceWorkspace +} + +func (c *Controller) writeIssueWatchError(ctx *gin.Context, err error) { + if errors.Is(err, ErrIssueWatchNotFound) { + ctx.JSON(http.StatusNotFound, gin.H{"error": err.Error()}) + return + } + if errors.Is(err, ErrInvalidConfig) { + ctx.JSON(http.StatusBadRequest, gin.H{"error": err.Error()}) + return + } + c.writeClientError(ctx, err) +} diff --git a/apps/backend/internal/linear/models.go b/apps/backend/internal/linear/models.go index 6e21cc53cc..40b68514e2 100644 --- a/apps/backend/internal/linear/models.go +++ b/apps/backend/internal/linear/models.go @@ -135,3 +135,84 @@ const SecretKey = "linear:singleton:token" func LegacySecretKeyForWorkspace(workspaceID string) string { return "linear:" + workspaceID + ":token" } + +// DefaultIssueWatchPollInterval is the polling cadence assigned to a watcher +// when the caller does not specify one. Five minutes balances freshness +// against Linear rate limits when many workspaces have watches configured. +const DefaultIssueWatchPollInterval = 300 + +// IssueWatch configures periodic Linear search-polling. The filter is a +// structured SearchFilter (Linear has no JQL equivalent) persisted as JSON; +// the poller deserialises it back to SearchFilter at the store boundary. +// +// As with Jira, Linear issues have no repository affinity — the target +// workflow step's defaults determine where the resulting task runs. +type IssueWatch struct { + ID string `json:"id" db:"id"` + WorkspaceID string `json:"workspaceId" db:"workspace_id"` + WorkflowID string `json:"workflowId" db:"workflow_id"` + WorkflowStepID string `json:"workflowStepId" db:"workflow_step_id"` + Filter SearchFilter `json:"filter"` + AgentProfileID string `json:"agentProfileId" db:"agent_profile_id"` + ExecutorProfileID string `json:"executorProfileId" db:"executor_profile_id"` + Prompt string `json:"prompt" db:"prompt"` + Enabled bool `json:"enabled" db:"enabled"` + PollIntervalSeconds int `json:"pollIntervalSeconds" db:"poll_interval_seconds"` + LastPolledAt *time.Time `json:"lastPolledAt,omitempty" db:"last_polled_at"` + CreatedAt time.Time `json:"createdAt" db:"created_at"` + UpdatedAt time.Time `json:"updatedAt" db:"updated_at"` +} + +// IssueWatchTask deduplicates task creation per (watch, issue) tuple. The +// UNIQUE constraint on (issue_watch_id, issue_identifier) prevents two +// concurrent pollers from racing to create duplicate tasks for the same +// issue. We key on Identifier (e.g. "ENG-123") rather than the GraphQL UUID +// because it's stable, what humans see, and present on every search result. +type IssueWatchTask struct { + ID string `json:"id" db:"id"` + IssueWatchID string `json:"issueWatchId" db:"issue_watch_id"` + IssueIdentifier string `json:"issueIdentifier" db:"issue_identifier"` + IssueURL string `json:"issueUrl" db:"issue_url"` + TaskID string `json:"taskId" db:"task_id"` + CreatedAt time.Time `json:"createdAt" db:"created_at"` +} + +// NewLinearIssueEvent is published on the bus whenever the poller observes an +// issue matching a watch that has no existing dedup row. The orchestrator +// consumes this to create (and optionally auto-start) a Kandev task. +type NewLinearIssueEvent struct { + IssueWatchID string `json:"issueWatchId"` + WorkspaceID string `json:"workspaceId"` + WorkflowID string `json:"workflowId"` + WorkflowStepID string `json:"workflowStepId"` + AgentProfileID string `json:"agentProfileId"` + ExecutorProfileID string `json:"executorProfileId"` + Prompt string `json:"prompt"` + Issue *LinearIssue `json:"issue"` +} + +// CreateIssueWatchRequest is the payload for POST /api/v1/linear/watches/issue. +type CreateIssueWatchRequest struct { + WorkspaceID string `json:"workspaceId"` + WorkflowID string `json:"workflowId"` + WorkflowStepID string `json:"workflowStepId"` + Filter SearchFilter `json:"filter"` + AgentProfileID string `json:"agentProfileId"` + ExecutorProfileID string `json:"executorProfileId"` + Prompt string `json:"prompt"` + PollIntervalSeconds int `json:"pollIntervalSeconds"` + Enabled *bool `json:"enabled,omitempty"` +} + +// UpdateIssueWatchRequest is the payload for PATCH /api/v1/linear/watches/issue/:id. +// All fields are pointers so the caller can omit ones it doesn't want to change. +type UpdateIssueWatchRequest struct { + WorkflowID *string `json:"workflowId,omitempty"` + WorkflowStepID *string `json:"workflowStepId,omitempty"` + Filter *SearchFilter `json:"filter,omitempty"` + AgentProfileID *string `json:"agentProfileId,omitempty"` + ExecutorProfileID *string `json:"executorProfileId,omitempty"` + Prompt *string `json:"prompt,omitempty"` + Enabled *bool `json:"enabled,omitempty"` + PollIntervalSeconds *int `json:"pollIntervalSeconds,omitempty"` +} diff --git a/apps/backend/internal/linear/poller.go b/apps/backend/internal/linear/poller.go index f98d85edb9..a1da701728 100644 --- a/apps/backend/internal/linear/poller.go +++ b/apps/backend/internal/linear/poller.go @@ -2,44 +2,174 @@ package linear import ( "context" + "sync" + "time" + + "go.uber.org/zap" "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/integrations/healthpoll" ) -// Poller probes the stored Linear credentials of every configured workspace on -// the shared 90s auth-health cadence. The actual loop lives in -// internal/integrations/healthpoll; this type exists so callers can keep the -// familiar `linear.NewPoller(...).Start(ctx)` shape without depending on the -// shared package directly. +// defaultIssuePollTickInterval is how often the issue-watch loop wakes up to +// consider every enabled watcher. The actual GraphQL search is only re-run for +// a watcher when its per-watch `PollIntervalSeconds` has elapsed since +// `LastPolledAt`, so this is the *minimum* granularity, not the *actual* +// cadence. +const defaultIssuePollTickInterval = 60 * time.Second + +// Poller drives two background loops sharing a single Service: +// - auth health: probes stored credentials so the UI can show connect status. +// Delegated to internal/integrations/healthpoll for the loop semantics. +// - issue watches: runs each enabled watcher's filter and emits +// NewLinearIssueEvent for every matching issue the orchestrator hasn't yet +// seen. +// +// Both loops are cancelled together via Stop. type Poller struct { - inner *healthpoll.Poller + service *Service + logger *logger.Logger + auth *healthpoll.Poller + issueInterval time.Duration + issueTickHook func() // tests use this to observe each issue-watch tick. + + mu sync.Mutex + cancelIssueLoop context.CancelFunc + wg sync.WaitGroup + started bool } -// NewPoller returns a poller using the default 90s cadence. Returns nil when -// svc is nil so the caller's nil-check still works after the indirection. +// NewPoller returns a poller using the default cadences. Returns nil when svc +// is nil so the caller's nil-check still works after the indirection. func NewPoller(svc *Service, log *logger.Logger) *Poller { if svc == nil { return nil } - return &Poller{inner: healthpoll.New("linear", svcProber{svc}, log)} + return &Poller{ + service: svc, + logger: log, + auth: healthpoll.New("linear", svcProber{svc}, log), + issueInterval: defaultIssuePollTickInterval, + } } -// Start launches the background loop. Calling Start more than once without -// Stop is a no-op. +// SetIssueTickHook installs a callback fired at the end of each issue-watch +// tick. Production code never sets this; tests use it to wait for a tick +// without sleep-polling. +func (p *Poller) SetIssueTickHook(fn func()) { + p.mu.Lock() + p.issueTickHook = fn + p.mu.Unlock() +} + +// Start launches both background loops. Calling Start more than once without +// Stop is a no-op. A nil receiver is also a no-op. func (p *Poller) Start(ctx context.Context) { - if p == nil || p.inner == nil { + if p == nil { return } - p.inner.Start(ctx) + p.mu.Lock() + defer p.mu.Unlock() + if p.started || p.service == nil { + return + } + p.started = true + issueCtx, cancel := context.WithCancel(ctx) + p.cancelIssueLoop = cancel + p.auth.Start(ctx) + p.wg.Add(1) + go p.issueWatchLoop(issueCtx) } -// Stop cancels the loop and waits for it to drain. +// Stop cancels both loops and waits for them to drain. func (p *Poller) Stop() { - if p == nil || p.inner == nil { + if p == nil { return } - p.inner.Stop() + p.mu.Lock() + if !p.started { + p.mu.Unlock() + return + } + cancel := p.cancelIssueLoop + p.mu.Unlock() + if cancel != nil { + cancel() + } + p.auth.Stop() + p.wg.Wait() + p.mu.Lock() + p.started = false + p.mu.Unlock() +} + +func (p *Poller) issueWatchLoop(ctx context.Context) { + defer p.wg.Done() + ticker := time.NewTicker(p.issueInterval) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + p.checkIssueWatches(ctx) + p.fireIssueTickHook() + } + } +} + +func (p *Poller) checkIssueWatches(ctx context.Context) { + watches, err := p.service.Store().ListEnabledIssueWatches(ctx) + if err != nil { + p.logger.Warn("linear poller: list enabled issue watches failed", zap.Error(err)) + return + } + if len(watches) == 0 { + return + } + for _, w := range watches { + if ctx.Err() != nil { + return + } + if !isIssueWatchDue(w, time.Now()) { + continue + } + newIssues, err := p.service.CheckIssueWatch(ctx, w) + if err != nil { + p.logger.Debug("linear poller: check issue watch failed", + zap.String("watch_id", w.ID), zap.Error(err)) + continue + } + for _, issue := range newIssues { + p.logger.Info("new linear issue found for watch", + zap.String("watch_id", w.ID), + zap.String("identifier", issue.Identifier), + zap.String("title", issue.Title)) + p.service.publishNewLinearIssueEvent(ctx, w, issue) + } + } +} + +// isIssueWatchDue reports whether enough time has passed since the watch was +// last polled to re-run its search. +func isIssueWatchDue(w *IssueWatch, now time.Time) bool { + if w.LastPolledAt == nil { + return true + } + interval := w.PollIntervalSeconds + if interval <= 0 { + interval = DefaultIssueWatchPollInterval + } + return now.Sub(*w.LastPolledAt) >= time.Duration(interval)*time.Second +} + +func (p *Poller) fireIssueTickHook() { + p.mu.Lock() + hook := p.issueTickHook + p.mu.Unlock() + if hook != nil { + hook() + } } // svcProber adapts *Service to healthpoll.Prober without leaking the shared diff --git a/apps/backend/internal/linear/poller_issue_watch_test.go b/apps/backend/internal/linear/poller_issue_watch_test.go new file mode 100644 index 0000000000..03d8db61c5 --- /dev/null +++ b/apps/backend/internal/linear/poller_issue_watch_test.go @@ -0,0 +1,213 @@ +package linear + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/internal/events" + "github.com/kandev/kandev/internal/events/bus" +) + +// recordingSubscriber subscribes to LinearNewIssue on the in-memory bus and +// captures payloads so the test can assert on them without touching internals. +type recordingSubscriber struct { + mu sync.Mutex + payload []*NewLinearIssueEvent + done chan struct{} + want int +} + +func newRecordingSubscriber(eb bus.EventBus, want int) (*recordingSubscriber, error) { + r := &recordingSubscriber{done: make(chan struct{}), want: want} + _, err := eb.Subscribe(events.LinearNewIssue, func(_ context.Context, e *bus.Event) error { + r.mu.Lock() + defer r.mu.Unlock() + evt, ok := e.Data.(*NewLinearIssueEvent) + if !ok { + return nil + } + r.payload = append(r.payload, evt) + if r.want > 0 && len(r.payload) == r.want { + close(r.done) + } + return nil + }) + return r, err +} + +func (r *recordingSubscriber) snapshot() []*NewLinearIssueEvent { + r.mu.Lock() + defer r.mu.Unlock() + out := make([]*NewLinearIssueEvent, len(r.payload)) + copy(out, r.payload) + return out +} + +func TestPoller_CheckIssueWatches_PublishesNewIssuesOnly(t *testing.T) { + f := newPollerFixture(t) + ctx := context.Background() + f.saveConfig(t, "tok") + + eb := bus.NewMemoryEventBus(logger.Default()) + defer eb.Close() + f.svc.SetEventBus(eb) + + sub, err := newRecordingSubscriber(eb, 2) + if err != nil { + t.Fatalf("subscribe: %v", err) + } + + w := newTestIssueWatch("ws-1") + if err := f.store.CreateIssueWatch(ctx, w); err != nil { + t.Fatalf("create watch: %v", err) + } + + f.client.searchIssuesFn = func(_ SearchFilter, _ string, _ int) (*SearchResult, error) { + return &SearchResult{ + Issues: []LinearIssue{ + {Identifier: "ENG-1", Title: "first", URL: "https://linear.app/x/issue/ENG-1"}, + {Identifier: "ENG-2", Title: "second", URL: "https://linear.app/x/issue/ENG-2"}, + }, + IsLast: true, + }, nil + } + + // First tick: both issues are new, two events publish. + f.poller.checkIssueWatches(ctx) + + <-sub.done + first := sub.snapshot() + if len(first) != 2 { + t.Fatalf("expected 2 events on first tick, got %d", len(first)) + } + idents := []string{first[0].Issue.Identifier, first[1].Issue.Identifier} + if !contains(idents, "ENG-1") || !contains(idents, "ENG-2") { + t.Errorf("expected events for ENG-1 and ENG-2, got %v", idents) + } + if first[0].WorkspaceID != "ws-1" || first[0].WorkflowID != "wf-1" { + t.Errorf("event missing watch context: %+v", first[0]) + } + + // Reserve both identifiers to simulate the orchestrator having created tasks. + for _, id := range []string{"ENG-1", "ENG-2"} { + if _, err := f.store.ReserveIssueWatchTask(ctx, w.ID, id, "https://linear.app/x/issue/"+id); err != nil { + t.Fatalf("reserve %s: %v", id, err) + } + } + + // Second tick: nothing new, no additional events should publish. + f.poller.checkIssueWatches(ctx) + if got := len(sub.snapshot()); got != 2 { + t.Errorf("expected event count to stay at 2 after dedup, got %d", got) + } +} + +func TestPoller_CheckIssueWatches_SkipsDisabled(t *testing.T) { + f := newPollerFixture(t) + ctx := context.Background() + f.saveConfig(t, "tok") + + eb := bus.NewMemoryEventBus(logger.Default()) + defer eb.Close() + f.svc.SetEventBus(eb) + + disabled := newTestIssueWatch("ws-1") + disabled.Enabled = false + if err := f.store.CreateIssueWatch(ctx, disabled); err != nil { + t.Fatalf("create disabled: %v", err) + } + + calls := 0 + f.client.searchIssuesFn = func(_ SearchFilter, _ string, _ int) (*SearchResult, error) { + calls++ + return &SearchResult{Issues: []LinearIssue{{Identifier: "ENG-1"}}, IsLast: true}, nil + } + + f.poller.checkIssueWatches(ctx) + + if calls != 0 { + t.Errorf("expected disabled watch to be skipped (no Linear call), got %d calls", calls) + } +} + +func contains(haystack []string, needle string) bool { + for _, h := range haystack { + if h == needle { + return true + } + } + return false +} + +func TestIsIssueWatchDue(t *testing.T) { + now := time.Now() + stamp := now.Add(-30 * time.Second) + cases := []struct { + name string + watch *IssueWatch + expected bool + }{ + { + name: "never polled is always due", + watch: &IssueWatch{PollIntervalSeconds: 300}, + expected: true, + }, + { + name: "polled less than interval ago is not due", + watch: &IssueWatch{PollIntervalSeconds: 300, LastPolledAt: &stamp}, + expected: false, + }, + { + name: "polled at exactly the interval boundary is due", + watch: &IssueWatch{PollIntervalSeconds: 30, LastPolledAt: &stamp}, + expected: true, + }, + { + name: "zero interval falls back to default and gates accordingly", + watch: &IssueWatch{PollIntervalSeconds: 0, LastPolledAt: &stamp}, + expected: false, + }, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := isIssueWatchDue(tc.watch, now); got != tc.expected { + t.Errorf("expected %v, got %v", tc.expected, got) + } + }) + } +} + +func TestPoller_CheckIssueWatches_RespectsPerWatchInterval(t *testing.T) { + f := newPollerFixture(t) + ctx := context.Background() + f.saveConfig(t, "tok") + + eb := bus.NewMemoryEventBus(logger.Default()) + defer eb.Close() + f.svc.SetEventBus(eb) + + w := newTestIssueWatch("ws-1") + w.PollIntervalSeconds = 300 + if err := f.store.CreateIssueWatch(ctx, w); err != nil { + t.Fatalf("create: %v", err) + } + now := time.Now().UTC() + if err := f.store.UpdateIssueWatchLastPolled(ctx, w.ID, now); err != nil { + t.Fatalf("stamp: %v", err) + } + + calls := 0 + f.client.searchIssuesFn = func(_ SearchFilter, _ string, _ int) (*SearchResult, error) { + calls++ + return &SearchResult{Issues: []LinearIssue{{Identifier: "ENG-1"}}, IsLast: true}, nil + } + + f.poller.checkIssueWatches(ctx) + + if calls != 0 { + t.Errorf("expected gating to skip the Linear search, got %d call(s)", calls) + } +} diff --git a/apps/backend/internal/linear/provider.go b/apps/backend/internal/linear/provider.go index 7eb4e9933f..62c385e62c 100644 --- a/apps/backend/internal/linear/provider.go +++ b/apps/backend/internal/linear/provider.go @@ -10,6 +10,7 @@ import ( "go.uber.org/zap" "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/internal/events/bus" ) // mockEnvVar gates the in-memory mock client used in E2E tests. @@ -20,14 +21,16 @@ func MockEnabled() bool { return os.Getenv(mockEnvVar) == "true" } -// Provide builds the Linear service. Cleanup is a no-op today — the service -// holds only in-memory client caches — but the signature mirrors other -// integration providers so callers can register it uniformly. +// Provide builds the Linear service. eventBus may be nil — used in tests and +// during early boot before the bus is ready; the service falls back to a +// no-op publish path. Cleanup is a no-op today — the service holds only +// in-memory client caches — but the signature mirrors other integration +// providers so callers can register it uniformly. // // When KANDEV_MOCK_LINEAR=true, the service is wired to a process-wide // MockClient and the same instance is exposed via Service.MockClient() so the // E2E mock controller can drive it. -func Provide(writer, reader *sqlx.DB, secrets SecretStore, log *logger.Logger) (*Service, func() error, error) { +func Provide(writer, reader *sqlx.DB, secrets SecretStore, eventBus bus.EventBus, log *logger.Logger) (*Service, func() error, error) { store, err := NewStore(writer, reader) if err != nil { return nil, nil, err @@ -42,6 +45,9 @@ func Provide(writer, reader *sqlx.DB, secrets SecretStore, log *logger.Logger) ( } svc := NewService(store, secrets, clientFn, log) svc.mockClient = mock + if eventBus != nil { + svc.SetEventBus(eventBus) + } cleanup := func() error { return nil } return svc, cleanup, nil } diff --git a/apps/backend/internal/linear/service.go b/apps/backend/internal/linear/service.go index b42b2df417..25bf49d655 100644 --- a/apps/backend/internal/linear/service.go +++ b/apps/backend/internal/linear/service.go @@ -10,6 +10,7 @@ import ( "go.uber.org/zap" "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/internal/events/bus" ) // SecretStore is the subset of the secrets store the service needs. @@ -30,6 +31,7 @@ type Service struct { clientFn ClientFactory client Client // singleton, cleared on config change. probeHook func() + eventBus bus.EventBus // 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. diff --git a/apps/backend/internal/linear/service_issue_watch.go b/apps/backend/internal/linear/service_issue_watch.go new file mode 100644 index 0000000000..f999a14491 --- /dev/null +++ b/apps/backend/internal/linear/service_issue_watch.go @@ -0,0 +1,266 @@ +package linear + +import ( + "context" + "errors" + "fmt" + "strings" + "time" + + "go.uber.org/zap" + + "github.com/kandev/kandev/internal/events" + "github.com/kandev/kandev/internal/events/bus" +) + +// ErrIssueWatchNotFound is returned when GetIssueWatch's caller looks up an ID +// that doesn't exist. Callers map this to HTTP 404. +var ErrIssueWatchNotFound = errors.New("linear: issue watch not found") + +// SetEventBus wires the bus used to publish NewLinearIssueEvent. Optional: if +// unset the poller still runs but observed issues do not become Kandev tasks. +func (s *Service) SetEventBus(eb bus.EventBus) { + s.mu.Lock() + s.eventBus = eb + s.mu.Unlock() +} + +// CreateIssueWatch validates the request and persists a new watch row. +func (s *Service) CreateIssueWatch(ctx context.Context, req *CreateIssueWatchRequest) (*IssueWatch, error) { + if err := validateIssueWatchCreate(req); err != nil { + return nil, err + } + w := &IssueWatch{ + WorkspaceID: req.WorkspaceID, + WorkflowID: req.WorkflowID, + WorkflowStepID: req.WorkflowStepID, + Filter: normalizeFilter(req.Filter), + AgentProfileID: req.AgentProfileID, + ExecutorProfileID: req.ExecutorProfileID, + Prompt: req.Prompt, + PollIntervalSeconds: req.PollIntervalSeconds, + Enabled: true, + } + if req.Enabled != nil { + w.Enabled = *req.Enabled + } + if err := s.store.CreateIssueWatch(ctx, w); err != nil { + return nil, err + } + return w, nil +} + +// ListIssueWatches returns the watches configured for a workspace. +func (s *Service) ListIssueWatches(ctx context.Context, workspaceID string) ([]*IssueWatch, error) { + return s.store.ListIssueWatches(ctx, workspaceID) +} + +// ListAllIssueWatches returns every watch across all workspaces. +func (s *Service) ListAllIssueWatches(ctx context.Context) ([]*IssueWatch, error) { + return s.store.ListAllIssueWatches(ctx) +} + +// GetIssueWatch returns a single watch by ID or ErrIssueWatchNotFound. +func (s *Service) GetIssueWatch(ctx context.Context, id string) (*IssueWatch, error) { + w, err := s.store.GetIssueWatch(ctx, id) + if err != nil { + return nil, err + } + if w == nil { + return nil, ErrIssueWatchNotFound + } + return w, nil +} + +// UpdateIssueWatch applies a partial update by patching only the fields the +// caller explicitly set, then persists the result. +func (s *Service) UpdateIssueWatch(ctx context.Context, id string, req *UpdateIssueWatchRequest) (*IssueWatch, error) { + w, err := s.GetIssueWatch(ctx, id) + if err != nil { + return nil, err + } + applyIssueWatchPatch(w, req) + if filterIsEmpty(w.Filter) { + return nil, fmt.Errorf("%w: filter must specify at least one of query, teamKey, stateIds, or assigned", ErrInvalidConfig) + } + if w.WorkflowID == "" || w.WorkflowStepID == "" { + return nil, fmt.Errorf("%w: workflowId and workflowStepId cannot be empty", ErrInvalidConfig) + } + if err := validatePollInterval(w.PollIntervalSeconds); err != nil { + return nil, err + } + if err := s.store.UpdateIssueWatch(ctx, w); err != nil { + return nil, err + } + return w, nil +} + +// DeleteIssueWatch removes the watch and its dedup rows. Idempotent. +func (s *Service) DeleteIssueWatch(ctx context.Context, id string) error { + return s.store.DeleteIssueWatch(ctx, id) +} + +// 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". +// +// Concurrency note: callers must tolerate being handed an issue that gets +// stolen by a concurrent reserver. We query the seen-set and return unseen +// identifiers, but we do NOT insert the dedup row here — that happens in the +// orchestrator's ReserveIssueWatchTask via INSERT OR IGNORE. If the manual +// /trigger endpoint and the poller tick fire for the same watch in quick +// succession, both calls can see the same identifier as unseen before either +// has reserved it. The duplicate publish is harmless (the second reserver +// loses the race and bails) but the goroutine work is wasted. Same pattern as +// the JIRA watcher. +func (s *Service) CheckIssueWatch(ctx context.Context, w *IssueWatch) ([]*LinearIssue, error) { + defer s.stampWatchLastPolled(w.ID) + client, err := s.clientFor(ctx) + if err != nil { + return nil, err + } + res, err := client.SearchIssues(ctx, w.Filter, "", issueWatchSearchPageSize) + if err != nil { + return nil, err + } + seen, err := s.store.ListSeenIssueIdentifiers(ctx, w.ID) + if err != nil { + s.log.Warn("linear: dedup set fetch failed", + zap.String("watch_id", w.ID), zap.Error(err)) + seen = nil + } + out := make([]*LinearIssue, 0, len(res.Issues)) + for i := range res.Issues { + issue := res.Issues[i] + if _, ok := seen[issue.Identifier]; ok { + continue + } + out = append(out, &issue) + } + return out, nil +} + +// stampWatchLastPolled writes the current timestamp using a fresh background +// context with a short write deadline, so a cancelled caller ctx (e.g. shutdown) +// doesn't drop the liveness record. +func (s *Service) stampWatchLastPolled(watchID string) { + ctx, cancel := context.WithTimeout(context.Background(), authHealthWriteTimeout) + defer cancel() + if err := s.store.UpdateIssueWatchLastPolled(ctx, watchID, time.Now().UTC()); err != nil { + s.log.Warn("linear: update last_polled_at failed", + zap.String("watch_id", watchID), zap.Error(err)) + } +} + +// publishNewLinearIssueEvent emits the orchestrator-facing event for one +// freshly-observed issue. No-op when the event bus is not wired (tests, early +// boot). +func (s *Service) publishNewLinearIssueEvent(ctx context.Context, w *IssueWatch, issue *LinearIssue) { + s.mu.Lock() + eb := s.eventBus + s.mu.Unlock() + if eb == nil { + return + } + evt := bus.NewEvent(events.LinearNewIssue, "linear", &NewLinearIssueEvent{ + IssueWatchID: w.ID, + WorkspaceID: w.WorkspaceID, + WorkflowID: w.WorkflowID, + WorkflowStepID: w.WorkflowStepID, + AgentProfileID: w.AgentProfileID, + ExecutorProfileID: w.ExecutorProfileID, + Prompt: w.Prompt, + Issue: issue, + }) + if err := eb.Publish(ctx, events.LinearNewIssue, evt); err != nil { + s.log.Debug("linear: publish new issue event failed", + zap.String("watch_id", w.ID), zap.String("identifier", issue.Identifier), zap.Error(err)) + } +} + +// issueWatchSearchPageSize caps how many issues a single CheckIssueWatch call +// pulls from Linear. Mirrors the Jira watcher's limit so per-tick cost stays +// bounded for very broad filters. +const issueWatchSearchPageSize = 50 + +// MinIssueWatchPollInterval / MaxIssueWatchPollInterval bound the per-watch +// search re-run cadence. +const ( + MinIssueWatchPollInterval = 60 + MaxIssueWatchPollInterval = 3600 +) + +func validateIssueWatchCreate(req *CreateIssueWatchRequest) error { + if req.WorkspaceID == "" { + return fmt.Errorf("%w: workspaceId required", ErrInvalidConfig) + } + if req.WorkflowID == "" || req.WorkflowStepID == "" { + return fmt.Errorf("%w: workflowId and workflowStepId required", ErrInvalidConfig) + } + if filterIsEmpty(normalizeFilter(req.Filter)) { + return fmt.Errorf("%w: filter must specify at least one of query, teamKey, stateIds, or assigned", ErrInvalidConfig) + } + if req.PollIntervalSeconds != 0 { + if err := validatePollInterval(req.PollIntervalSeconds); err != nil { + return err + } + } + return nil +} + +func validatePollInterval(seconds int) error { + if seconds < MinIssueWatchPollInterval || seconds > MaxIssueWatchPollInterval { + return fmt.Errorf("%w: pollIntervalSeconds must be between %d and %d", + ErrInvalidConfig, MinIssueWatchPollInterval, MaxIssueWatchPollInterval) + } + return nil +} + +// normalizeFilter trims string fields and drops empty stateIds entries, so a +// filter that looks empty after normalization fails the at-least-one check +// instead of slipping through with whitespace. +func normalizeFilter(f SearchFilter) SearchFilter { + out := SearchFilter{ + Query: strings.TrimSpace(f.Query), + TeamKey: strings.TrimSpace(f.TeamKey), + Assigned: strings.TrimSpace(f.Assigned), + } + for _, id := range f.StateIDs { + id = strings.TrimSpace(id) + if id != "" { + out.StateIDs = append(out.StateIDs, id) + } + } + return out +} + +func filterIsEmpty(f SearchFilter) bool { + return f.Query == "" && f.TeamKey == "" && f.Assigned == "" && len(f.StateIDs) == 0 +} + +func applyIssueWatchPatch(w *IssueWatch, req *UpdateIssueWatchRequest) { + if req.WorkflowID != nil { + w.WorkflowID = *req.WorkflowID + } + if req.WorkflowStepID != nil { + w.WorkflowStepID = *req.WorkflowStepID + } + if req.Filter != nil { + w.Filter = normalizeFilter(*req.Filter) + } + if req.AgentProfileID != nil { + w.AgentProfileID = *req.AgentProfileID + } + if req.ExecutorProfileID != nil { + w.ExecutorProfileID = *req.ExecutorProfileID + } + if req.Prompt != nil { + w.Prompt = *req.Prompt + } + if req.Enabled != nil { + w.Enabled = *req.Enabled + } + if req.PollIntervalSeconds != nil { + w.PollIntervalSeconds = *req.PollIntervalSeconds + } +} diff --git a/apps/backend/internal/linear/service_issue_watch_test.go b/apps/backend/internal/linear/service_issue_watch_test.go new file mode 100644 index 0000000000..2f15c62d02 --- /dev/null +++ b/apps/backend/internal/linear/service_issue_watch_test.go @@ -0,0 +1,225 @@ +package linear + +import ( + "context" + "errors" + "testing" +) + +// withSearchResults returns a fakeClient that always returns the given issues +// for SearchIssues, ignoring the filter. +func (c *fakeClient) withSearchResults(issues []LinearIssue) *fakeClient { + c.searchIssuesFn = func(_ SearchFilter, _ string, _ int) (*SearchResult, error) { + return &SearchResult{Issues: issues, IsLast: true}, nil + } + return c +} + +func TestService_CreateIssueWatch_DefaultsAndValidation(t *testing.T) { + f := newSvcFixture(t) + ctx := context.Background() + + // Empty filter is rejected. + if _, err := f.svc.CreateIssueWatch(ctx, &CreateIssueWatchRequest{ + WorkspaceID: "ws-1", + WorkflowID: "wf", + WorkflowStepID: "step", + }); !errors.Is(err, ErrInvalidConfig) { + t.Errorf("expected ErrInvalidConfig for empty filter, got %v", err) + } + + // Whitespace-only fields are also rejected (normalize then check). + if _, err := f.svc.CreateIssueWatch(ctx, &CreateIssueWatchRequest{ + WorkspaceID: "ws-1", + WorkflowID: "wf", + WorkflowStepID: "step", + Filter: SearchFilter{Query: " ", TeamKey: " ", StateIDs: []string{" ", ""}}, + }); !errors.Is(err, ErrInvalidConfig) { + t.Errorf("expected ErrInvalidConfig for whitespace-only filter, got %v", err) + } + + // Happy path assigns ID + defaults Enabled=true. + w, err := f.svc.CreateIssueWatch(ctx, &CreateIssueWatchRequest{ + WorkspaceID: "ws-1", + WorkflowID: "wf", + WorkflowStepID: "step", + Filter: SearchFilter{TeamKey: "ENG"}, + }) + if err != nil { + t.Fatalf("create: %v", err) + } + if w.ID == "" { + t.Fatal("expected ID assigned") + } + if !w.Enabled { + t.Error("expected Enabled defaulted to true") + } + if w.Filter.TeamKey != "ENG" { + t.Errorf("filter not persisted: %+v", w.Filter) + } +} + +func TestService_UpdateIssueWatch_PartialPatch(t *testing.T) { + f := newSvcFixture(t) + ctx := context.Background() + + created, err := f.svc.CreateIssueWatch(ctx, &CreateIssueWatchRequest{ + WorkspaceID: "ws-1", + WorkflowID: "wf", + WorkflowStepID: "step", + Filter: SearchFilter{TeamKey: "ENG"}, + Prompt: "original", + }) + if err != nil { + t.Fatalf("create: %v", err) + } + + // Patch only the prompt; everything else must remain. + newPrompt := "updated" + updated, err := f.svc.UpdateIssueWatch(ctx, created.ID, &UpdateIssueWatchRequest{ + Prompt: &newPrompt, + }) + if err != nil { + t.Fatalf("update: %v", err) + } + if updated.Prompt != "updated" { + t.Errorf("prompt not patched: %q", updated.Prompt) + } + if updated.Filter.TeamKey != "ENG" || updated.WorkspaceID != created.WorkspaceID { + t.Errorf("unexpected mutation of unset fields: %+v", updated) + } + + // Patching filter to something empty is rejected to keep watch rows valid. + empty := SearchFilter{} + if _, err := f.svc.UpdateIssueWatch(ctx, created.ID, &UpdateIssueWatchRequest{Filter: &empty}); !errors.Is(err, ErrInvalidConfig) { + t.Errorf("expected ErrInvalidConfig for empty filter patch, got %v", err) + } +} + +func TestService_CreateIssueWatch_RejectsOutOfRangeInterval(t *testing.T) { + f := newSvcFixture(t) + ctx := context.Background() + for _, n := range []int{1, 30, 59, 3601, 86400} { + _, err := f.svc.CreateIssueWatch(ctx, &CreateIssueWatchRequest{ + WorkspaceID: "ws-1", + WorkflowID: "wf", + WorkflowStepID: "step", + Filter: SearchFilter{TeamKey: "ENG"}, + PollIntervalSeconds: n, + }) + if !errors.Is(err, ErrInvalidConfig) { + t.Errorf("expected ErrInvalidConfig for pollIntervalSeconds=%d, got %v", n, err) + } + } + // Zero is allowed (the store coerces to default). + if _, err := f.svc.CreateIssueWatch(ctx, &CreateIssueWatchRequest{ + WorkspaceID: "ws-1", + WorkflowID: "wf", + WorkflowStepID: "step", + Filter: SearchFilter{TeamKey: "ENG"}, + }); err != nil { + t.Errorf("expected zero pollIntervalSeconds to be accepted, got %v", err) + } +} + +func TestService_UpdateIssueWatch_RejectsEmptyWorkflowFields(t *testing.T) { + f := newSvcFixture(t) + ctx := context.Background() + created, err := f.svc.CreateIssueWatch(ctx, &CreateIssueWatchRequest{ + WorkspaceID: "ws-1", WorkflowID: "wf", WorkflowStepID: "step", + Filter: SearchFilter{TeamKey: "ENG"}, + }) + if err != nil { + t.Fatalf("create: %v", err) + } + empty := "" + for _, req := range []*UpdateIssueWatchRequest{ + {WorkflowID: &empty}, + {WorkflowStepID: &empty}, + } { + if _, err := f.svc.UpdateIssueWatch(ctx, created.ID, req); !errors.Is(err, ErrInvalidConfig) { + t.Errorf("expected ErrInvalidConfig for %+v, got %v", req, err) + } + } +} + +func TestService_UpdateIssueWatch_NotFound(t *testing.T) { + f := newSvcFixture(t) + prompt := "x" + _, err := f.svc.UpdateIssueWatch(context.Background(), "ghost", &UpdateIssueWatchRequest{Prompt: &prompt}) + if !errors.Is(err, ErrIssueWatchNotFound) { + t.Errorf("expected ErrIssueWatchNotFound, got %v", err) + } +} + +func TestService_CheckIssueWatch_FiltersAlreadySeen(t *testing.T) { + f := newSvcFixture(t) + ctx := context.Background() + + if _, err := f.svc.SetConfig(ctx, &SetConfigRequest{ + AuthMethod: AuthMethodAPIKey, Secret: "lin_api", + }); err != nil { + t.Fatalf("set config: %v", err) + } + + f.client.withSearchResults([]LinearIssue{ + {Identifier: "ENG-1", Title: "one", URL: "https://linear.app/x/issue/ENG-1"}, + {Identifier: "ENG-2", Title: "two", URL: "https://linear.app/x/issue/ENG-2"}, + {Identifier: "ENG-3", Title: "three", URL: "https://linear.app/x/issue/ENG-3"}, + }) + + w, err := f.svc.CreateIssueWatch(ctx, &CreateIssueWatchRequest{ + WorkspaceID: "ws-1", WorkflowID: "wf", WorkflowStepID: "step", + Filter: SearchFilter{TeamKey: "ENG"}, + }) + if err != nil { + t.Fatalf("create watch: %v", err) + } + + // Pre-seed ENG-2 as already turned into a task. + if _, err := f.store.ReserveIssueWatchTask(ctx, w.ID, "ENG-2", "https://linear.app/x/issue/ENG-2"); err != nil { + t.Fatalf("seed reservation: %v", err) + } + + got, err := f.svc.CheckIssueWatch(ctx, w) + if err != nil { + t.Fatalf("check: %v", err) + } + if len(got) != 2 { + t.Fatalf("expected 2 unseen issues, got %d", len(got)) + } + for _, i := range got { + if i.Identifier == "ENG-2" { + t.Error("ENG-2 should have been filtered as already seen") + } + } + + refreshed, _ := f.store.GetIssueWatch(ctx, w.ID) + if refreshed.LastPolledAt == nil { + t.Error("expected last_polled_at stamped after check") + } +} + +func TestService_CheckIssueWatch_StampsLastPolledOnError(t *testing.T) { + f := newSvcFixture(t) + ctx := context.Background() + if _, err := f.svc.SetConfig(ctx, &SetConfigRequest{ + AuthMethod: AuthMethodAPIKey, Secret: "lin_api", + }); err != nil { + t.Fatalf("set config: %v", err) + } + f.client.searchIssuesFn = func(_ SearchFilter, _ string, _ int) (*SearchResult, error) { + return nil, errors.New("upstream 500") + } + w, _ := f.svc.CreateIssueWatch(ctx, &CreateIssueWatchRequest{ + WorkspaceID: "ws-1", WorkflowID: "wf", WorkflowStepID: "step", + Filter: SearchFilter{TeamKey: "ENG"}, + }) + if _, err := f.svc.CheckIssueWatch(ctx, w); err == nil { + t.Error("expected error from search to surface to caller") + } + refreshed, _ := f.store.GetIssueWatch(ctx, w.ID) + if refreshed.LastPolledAt == nil { + t.Error("expected last_polled_at stamped even on search failure (liveness signal)") + } +} diff --git a/apps/backend/internal/linear/store.go b/apps/backend/internal/linear/store.go index 3ace6e0ed4..0e913e4b5a 100644 --- a/apps/backend/internal/linear/store.go +++ b/apps/backend/internal/linear/store.go @@ -51,6 +51,35 @@ const createTablesSQL = ` created_at DATETIME NOT NULL, updated_at DATETIME NOT NULL ); + + CREATE TABLE IF NOT EXISTS linear_issue_watches ( + id TEXT PRIMARY KEY, + workspace_id TEXT NOT NULL, + workflow_id TEXT NOT NULL, + workflow_step_id TEXT NOT NULL, + filter_json TEXT NOT NULL DEFAULT '{}', + agent_profile_id TEXT NOT NULL DEFAULT '', + executor_profile_id TEXT NOT NULL DEFAULT '', + prompt TEXT NOT NULL DEFAULT '', + enabled BOOLEAN NOT NULL DEFAULT 1, + poll_interval_seconds INTEGER NOT NULL DEFAULT 300, + last_polled_at DATETIME, + created_at DATETIME NOT NULL, + updated_at DATETIME NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_linear_issue_watches_workspace + ON linear_issue_watches(workspace_id); + + CREATE TABLE IF NOT EXISTS linear_issue_watch_tasks ( + id TEXT PRIMARY KEY, + issue_watch_id TEXT NOT NULL, + issue_identifier TEXT NOT NULL, + issue_url TEXT NOT NULL, + task_id TEXT NOT NULL DEFAULT '', + created_at DATETIME NOT NULL, + UNIQUE(issue_watch_id, issue_identifier), + FOREIGN KEY(issue_watch_id) REFERENCES linear_issue_watches(id) ON DELETE CASCADE + ); ` // singletonID is the synthetic primary key of the (only) row in linear_configs. diff --git a/apps/backend/internal/linear/store_issue_watch.go b/apps/backend/internal/linear/store_issue_watch.go new file mode 100644 index 0000000000..3219704433 --- /dev/null +++ b/apps/backend/internal/linear/store_issue_watch.go @@ -0,0 +1,267 @@ +package linear + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + "time" + + "github.com/google/uuid" +) + +// issueWatchRow mirrors IssueWatch but stores the filter as a raw JSON string +// the way SQLite holds it. The store marshals/unmarshals at this boundary so +// service callers see a typed SearchFilter and never have to think about JSON. +type issueWatchRow struct { + ID string `db:"id"` + WorkspaceID string `db:"workspace_id"` + WorkflowID string `db:"workflow_id"` + WorkflowStepID string `db:"workflow_step_id"` + FilterJSON string `db:"filter_json"` + AgentProfileID string `db:"agent_profile_id"` + ExecutorProfileID string `db:"executor_profile_id"` + Prompt string `db:"prompt"` + Enabled bool `db:"enabled"` + PollIntervalSeconds int `db:"poll_interval_seconds"` + LastPolledAt *time.Time `db:"last_polled_at"` + CreatedAt time.Time `db:"created_at"` + UpdatedAt time.Time `db:"updated_at"` +} + +func (r *issueWatchRow) toIssueWatch() (*IssueWatch, error) { + var filter SearchFilter + if r.FilterJSON != "" { + if err := json.Unmarshal([]byte(r.FilterJSON), &filter); err != nil { + return nil, fmt.Errorf("decode filter: %w", err) + } + } + return &IssueWatch{ + ID: r.ID, + WorkspaceID: r.WorkspaceID, + WorkflowID: r.WorkflowID, + WorkflowStepID: r.WorkflowStepID, + Filter: filter, + AgentProfileID: r.AgentProfileID, + ExecutorProfileID: r.ExecutorProfileID, + Prompt: r.Prompt, + Enabled: r.Enabled, + PollIntervalSeconds: r.PollIntervalSeconds, + LastPolledAt: r.LastPolledAt, + CreatedAt: r.CreatedAt, + UpdatedAt: r.UpdatedAt, + }, nil +} + +func encodeFilter(f SearchFilter) (string, error) { + b, err := json.Marshal(f) + if err != nil { + return "", fmt.Errorf("encode filter: %w", err) + } + return string(b), nil +} + +const issueWatchColumns = `id, workspace_id, workflow_id, workflow_step_id, filter_json, + agent_profile_id, executor_profile_id, prompt, enabled, + poll_interval_seconds, last_polled_at, created_at, updated_at` + +// CreateIssueWatch persists a new issue watch row. ID and timestamps are +// assigned here so callers can pass a partially-populated struct. +func (s *Store) CreateIssueWatch(ctx context.Context, w *IssueWatch) error { + if w.ID == "" { + w.ID = uuid.New().String() + } + now := time.Now().UTC() + w.CreatedAt = now + w.UpdatedAt = now + if w.PollIntervalSeconds <= 0 { + w.PollIntervalSeconds = DefaultIssueWatchPollInterval + } + filterJSON, err := encodeFilter(w.Filter) + if err != nil { + return err + } + _, err = s.db.ExecContext(ctx, ` + INSERT INTO linear_issue_watches (`+issueWatchColumns+`) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + w.ID, w.WorkspaceID, w.WorkflowID, w.WorkflowStepID, filterJSON, + w.AgentProfileID, w.ExecutorProfileID, w.Prompt, w.Enabled, + w.PollIntervalSeconds, w.LastPolledAt, w.CreatedAt, w.UpdatedAt) + return err +} + +// GetIssueWatch returns a single watch by ID, or nil when no row matches. +func (s *Store) GetIssueWatch(ctx context.Context, id string) (*IssueWatch, error) { + var row issueWatchRow + err := s.ro.GetContext(ctx, &row, + `SELECT `+issueWatchColumns+` FROM linear_issue_watches WHERE id = ?`, id) + if errors.Is(err, sql.ErrNoRows) { + return nil, nil + } + if err != nil { + return nil, err + } + return row.toIssueWatch() +} + +// ListIssueWatches returns all watches configured for a workspace. +func (s *Store) ListIssueWatches(ctx context.Context, workspaceID string) ([]*IssueWatch, error) { + var rows []issueWatchRow + err := s.ro.SelectContext(ctx, &rows, + `SELECT `+issueWatchColumns+` FROM linear_issue_watches + WHERE workspace_id = ? ORDER BY created_at`, workspaceID) + if err != nil { + return nil, err + } + return materializeWatches(rows) +} + +// ListAllIssueWatches returns every watch across all workspaces. +func (s *Store) ListAllIssueWatches(ctx context.Context) ([]*IssueWatch, error) { + var rows []issueWatchRow + err := s.ro.SelectContext(ctx, &rows, + `SELECT `+issueWatchColumns+` FROM linear_issue_watches ORDER BY workspace_id, created_at`) + if err != nil { + return nil, err + } + return materializeWatches(rows) +} + +// ListEnabledIssueWatches returns every enabled watch across all workspaces, +// used by the poller to decide what to query each tick. +func (s *Store) ListEnabledIssueWatches(ctx context.Context) ([]*IssueWatch, error) { + var rows []issueWatchRow + err := s.ro.SelectContext(ctx, &rows, + `SELECT `+issueWatchColumns+` FROM linear_issue_watches + WHERE enabled = 1 ORDER BY created_at`) + if err != nil { + return nil, err + } + return materializeWatches(rows) +} + +func materializeWatches(rows []issueWatchRow) ([]*IssueWatch, error) { + out := make([]*IssueWatch, 0, len(rows)) + for i := range rows { + w, err := rows[i].toIssueWatch() + if err != nil { + return nil, err + } + out = append(out, w) + } + return out, nil +} + +// UpdateIssueWatch overwrites the mutable fields of an existing watch row. +// updated_at is bumped automatically; last_polled_at is preserved unless the +// caller explicitly sets it. +func (s *Store) UpdateIssueWatch(ctx context.Context, w *IssueWatch) error { + w.UpdatedAt = time.Now().UTC() + if w.PollIntervalSeconds <= 0 { + w.PollIntervalSeconds = DefaultIssueWatchPollInterval + } + filterJSON, err := encodeFilter(w.Filter) + if err != nil { + return err + } + _, err = s.db.ExecContext(ctx, ` + UPDATE linear_issue_watches SET workflow_id = ?, workflow_step_id = ?, filter_json = ?, + agent_profile_id = ?, executor_profile_id = ?, prompt = ?, + enabled = ?, poll_interval_seconds = ?, last_polled_at = ?, updated_at = ? + WHERE id = ?`, + w.WorkflowID, w.WorkflowStepID, filterJSON, + w.AgentProfileID, w.ExecutorProfileID, w.Prompt, + w.Enabled, w.PollIntervalSeconds, w.LastPolledAt, w.UpdatedAt, w.ID) + return err +} + +// UpdateIssueWatchLastPolled stamps the last-polled timestamp without touching +// the rest of the row. +func (s *Store) UpdateIssueWatchLastPolled(ctx context.Context, id string, t time.Time) error { + _, err := s.db.ExecContext(ctx, + `UPDATE linear_issue_watches SET last_polled_at = ?, updated_at = ? WHERE id = ?`, + t, time.Now().UTC(), id) + return err +} + +// DeleteIssueWatch removes a watch and its dedup rows in a single transaction. +// The explicit child DELETE guards older databases where foreign_keys may not +// have been enabled at attach time. +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 linear_issue_watch_tasks WHERE issue_watch_id = ?`, id); err != nil { + return err + } + if _, err := tx.ExecContext(ctx, `DELETE FROM linear_issue_watches WHERE id = ?`, id); err != nil { + return err + } + return tx.Commit() +} + +// ReserveIssueWatchTask atomically claims a slot for a (watch, issue) pair via +// INSERT OR IGNORE. Returns true when this caller won the race and should +// proceed to create the task. +func (s *Store) ReserveIssueWatchTask(ctx context.Context, watchID, identifier, issueURL string) (bool, error) { + res, err := s.db.ExecContext(ctx, ` + INSERT OR IGNORE INTO linear_issue_watch_tasks (id, issue_watch_id, issue_identifier, issue_url, task_id, created_at) + VALUES (?, ?, ?, ?, ?, ?)`, + uuid.New().String(), watchID, identifier, 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 stamps the created task ID onto a previously-reserved +// dedup row. +func (s *Store) AssignIssueWatchTaskID(ctx context.Context, watchID, identifier, taskID string) error { + res, err := s.db.ExecContext(ctx, ` + UPDATE linear_issue_watch_tasks SET task_id = ? + WHERE issue_watch_id = ? AND issue_identifier = ?`, + taskID, watchID, identifier) + 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=%s", watchID, identifier) + } + return nil +} + +// ReleaseIssueWatchTask drops a reservation so the next poll can retry. Used +// when task creation fails after a successful reserve. +func (s *Store) ReleaseIssueWatchTask(ctx context.Context, watchID, identifier string) error { + _, err := s.db.ExecContext(ctx, + `DELETE FROM linear_issue_watch_tasks WHERE issue_watch_id = ? AND issue_identifier = ?`, + watchID, identifier) + return err +} + +// ListSeenIssueIdentifiers returns the set of issue identifiers already +// reserved against a watch. +func (s *Store) ListSeenIssueIdentifiers(ctx context.Context, watchID string) (map[string]struct{}, error) { + var keys []string + err := s.ro.SelectContext(ctx, &keys, + `SELECT issue_identifier FROM linear_issue_watch_tasks WHERE issue_watch_id = ?`, watchID) + if err != nil { + return nil, err + } + out := make(map[string]struct{}, len(keys)) + for _, k := range keys { + out[k] = struct{}{} + } + return out, nil +} diff --git a/apps/backend/internal/linear/store_issue_watch_test.go b/apps/backend/internal/linear/store_issue_watch_test.go new file mode 100644 index 0000000000..300e55e979 --- /dev/null +++ b/apps/backend/internal/linear/store_issue_watch_test.go @@ -0,0 +1,261 @@ +package linear + +import ( + "context" + "testing" + "time" +) + +func newTestIssueWatch(workspaceID string) *IssueWatch { + return &IssueWatch{ + WorkspaceID: workspaceID, + WorkflowID: "wf-1", + WorkflowStepID: "step-1", + Filter: SearchFilter{TeamKey: "ENG", StateIDs: []string{"state-started"}}, + Prompt: "Investigate {{issue.identifier}}", + Enabled: true, + } +} + +func TestStore_IssueWatch_CreateGet(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 w.ID == "" { + t.Fatal("expected ID assigned") + } + if w.PollIntervalSeconds != DefaultIssueWatchPollInterval { + t.Errorf("expected default poll interval %d, got %d", DefaultIssueWatchPollInterval, w.PollIntervalSeconds) + } + if w.CreatedAt.IsZero() || w.UpdatedAt.IsZero() { + t.Error("expected timestamps assigned") + } + + got, err := store.GetIssueWatch(ctx, w.ID) + if err != nil { + t.Fatalf("get: %v", err) + } + if got == nil { + t.Fatal("expected watch, got nil") + } + if got.WorkspaceID != w.WorkspaceID || got.Prompt != w.Prompt { + t.Errorf("round-trip mismatch: %+v vs %+v", got, w) + } + // Filter survives the JSON round-trip. + if got.Filter.TeamKey != "ENG" || len(got.Filter.StateIDs) != 1 || got.Filter.StateIDs[0] != "state-started" { + t.Errorf("filter round-trip failed: %+v", got.Filter) + } +} + +func TestStore_IssueWatch_ListByWorkspace(t *testing.T) { + store := newTestStore(t) + ctx := context.Background() + + w1 := newTestIssueWatch("ws-1") + w1.Filter = SearchFilter{TeamKey: "A"} + w2 := newTestIssueWatch("ws-1") + w2.Filter = SearchFilter{TeamKey: "B"} + w3 := newTestIssueWatch("ws-2") + + for _, w := range []*IssueWatch{w1, w2, w3} { + if err := store.CreateIssueWatch(ctx, w); err != nil { + t.Fatalf("create: %v", err) + } + } + + got, err := store.ListIssueWatches(ctx, "ws-1") + if err != nil { + t.Fatalf("list: %v", err) + } + if len(got) != 2 { + t.Fatalf("expected 2 watches for ws-1, got %d", len(got)) + } + for _, w := range got { + if w.WorkspaceID != "ws-1" { + t.Errorf("workspace leaked into list: %s", w.WorkspaceID) + } + } +} + +func TestStore_IssueWatch_ListEnabled(t *testing.T) { + store := newTestStore(t) + ctx := context.Background() + + enabled := newTestIssueWatch("ws-1") + disabled := newTestIssueWatch("ws-2") + disabled.Enabled = false + for _, w := range []*IssueWatch{enabled, disabled} { + if err := store.CreateIssueWatch(ctx, w); err != nil { + t.Fatalf("create: %v", err) + } + } + + got, err := store.ListEnabledIssueWatches(ctx) + if err != nil { + t.Fatalf("list enabled: %v", err) + } + if len(got) != 1 { + t.Fatalf("expected only the enabled watch, got %d", len(got)) + } + if got[0].ID != enabled.ID { + t.Errorf("expected enabled watch returned, got %s", got[0].ID) + } +} + +func TestStore_IssueWatch_Update(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) + } + originalCreated := w.CreatedAt + w.Filter = SearchFilter{Query: "auth bug"} + w.Enabled = false + w.PollIntervalSeconds = 60 + if err := store.UpdateIssueWatch(ctx, w); err != nil { + t.Fatalf("update: %v", err) + } + got, _ := store.GetIssueWatch(ctx, w.ID) + if got.Filter.Query != "auth bug" || got.Enabled || got.PollIntervalSeconds != 60 { + t.Errorf("update did not persist: %+v", got) + } + if !got.CreatedAt.Equal(originalCreated) { + t.Errorf("update must not change created_at: %v vs %v", got.CreatedAt, originalCreated) + } +} + +func TestStore_IssueWatch_LastPolledStamp(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) + } + t1 := time.Now().UTC().Truncate(time.Second) + if err := store.UpdateIssueWatchLastPolled(ctx, w.ID, t1); err != nil { + t.Fatalf("stamp: %v", err) + } + got, _ := store.GetIssueWatch(ctx, w.ID) + if got.LastPolledAt == nil || !got.LastPolledAt.Equal(t1) { + t.Errorf("expected last_polled_at=%v, got %v", t1, got.LastPolledAt) + } +} + +func TestStore_IssueWatch_DeleteCascadesDedupRows(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.ReserveIssueWatchTask(ctx, w.ID, "ENG-1", "https://linear.app/x/issue/ENG-1"); err != nil { + t.Fatalf("reserve: %v", err) + } + + if err := store.DeleteIssueWatch(ctx, w.ID); err != nil { + t.Fatalf("delete: %v", err) + } + gone, _ := store.GetIssueWatch(ctx, w.ID) + if gone != nil { + t.Errorf("expected watch deleted, got %+v", gone) + } + // Re-creating a watch with the same ID and reserving the same identifier + // should succeed without a UNIQUE collision — proves cascade ran. + w2 := newTestIssueWatch("ws-1") + w2.ID = w.ID + if err := store.CreateIssueWatch(ctx, w2); err != nil { + t.Fatalf("recreate watch: %v", err) + } + ok, err := store.ReserveIssueWatchTask(ctx, w2.ID, "ENG-1", "https://linear.app/x/issue/ENG-1") + if err != nil { + t.Fatalf("re-reserve: %v", err) + } + if !ok { + t.Error("expected re-reserve to succeed after cascade delete") + } +} + +func TestStore_IssueWatchTask_ReserveDedup(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) + } + + first, err := store.ReserveIssueWatchTask(ctx, w.ID, "ENG-7", "https://linear.app/x/issue/ENG-7") + if err != nil { + t.Fatalf("first reserve: %v", err) + } + if !first { + t.Error("expected first reservation to win") + } + + second, err := store.ReserveIssueWatchTask(ctx, w.ID, "ENG-7", "https://linear.app/x/issue/ENG-7") + if err != nil { + t.Fatalf("second reserve: %v", err) + } + if second { + t.Error("expected second reservation to lose due to UNIQUE constraint") + } + + seen, err := store.ListSeenIssueIdentifiers(ctx, w.ID) + if err != nil { + t.Fatalf("list seen: %v", err) + } + if _, ok := seen["ENG-7"]; !ok { + t.Error("expected ENG-7 in seen set after reservation") + } +} + +func TestStore_IssueWatchTask_AssignTaskID(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.ReserveIssueWatchTask(ctx, w.ID, "ENG-1", "https://linear.app/x/issue/ENG-1"); err != nil { + t.Fatalf("reserve: %v", err) + } + if err := store.AssignIssueWatchTaskID(ctx, w.ID, "ENG-1", "task-abc"); err != nil { + t.Fatalf("assign: %v", err) + } + + if err := store.AssignIssueWatchTaskID(ctx, w.ID, "ENG-NOPE", "task-zzz"); err == nil { + t.Error("expected error for missing reservation, got nil") + } +} + +func TestStore_IssueWatchTask_Release(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.ReserveIssueWatchTask(ctx, w.ID, "ENG-1", "https://linear.app/x/issue/ENG-1"); err != nil { + t.Fatalf("reserve: %v", err) + } + if err := store.ReleaseIssueWatchTask(ctx, w.ID, "ENG-1"); err != nil { + t.Fatalf("release: %v", err) + } + again, err := store.ReserveIssueWatchTask(ctx, w.ID, "ENG-1", "https://linear.app/x/issue/ENG-1") + if err != nil { + t.Fatalf("re-reserve: %v", err) + } + if !again { + t.Error("expected reservation to succeed after release") + } +} diff --git a/apps/backend/internal/orchestrator/event_handlers_linear.go b/apps/backend/internal/orchestrator/event_handlers_linear.go new file mode 100644 index 0000000000..e0e6e80e7d --- /dev/null +++ b/apps/backend/internal/orchestrator/event_handlers_linear.go @@ -0,0 +1,187 @@ +package orchestrator + +import ( + "context" + "fmt" + "strings" + + "go.uber.org/zap" + + promptcfg "github.com/kandev/kandev/config/prompts" + "github.com/kandev/kandev/internal/events" + "github.com/kandev/kandev/internal/events/bus" + "github.com/kandev/kandev/internal/linear" +) + +// LinearService is the subset of linear.Service the orchestrator needs to +// deduplicate issue→task mappings. Mirrors the JIRA equivalent. +type LinearService interface { + ReserveIssueWatchTask(ctx context.Context, watchID, identifier, issueURL string) (bool, error) + AssignIssueWatchTaskID(ctx context.Context, watchID, identifier, taskID string) error + ReleaseIssueWatchTask(ctx context.Context, watchID, identifier string) error +} + +// SetLinearService wires the Linear dedup helpers into the orchestrator so +// linear-watch handlers can claim issue slots before creating tasks. +func (s *Service) SetLinearService(l LinearService) { + s.linearService = l +} + +// subscribeLinearEvents wires the Linear event handlers onto the bus. +func (s *Service) subscribeLinearEvents() { + if s.eventBus == nil { + return + } + if _, err := s.eventBus.Subscribe(events.LinearNewIssue, s.handleNewLinearIssue); err != nil { + s.logger.Error("failed to subscribe to linear.new_issue events", zap.Error(err)) + } +} + +// handleNewLinearIssue creates a Kandev task for a freshly-observed Linear issue. +func (s *Service) handleNewLinearIssue(_ context.Context, event *bus.Event) error { + evt, ok := event.Data.(*linear.NewLinearIssueEvent) + if !ok { + return nil + } + if evt.Issue == nil { + return nil + } + s.logger.Info("new linear issue detected from watch", + zap.String("issue_watch_id", evt.IssueWatchID), + zap.String("identifier", evt.Issue.Identifier)) + + if s.issueTaskCreator == nil { + s.logger.Warn("issue task creator not configured, skipping linear task creation") + return nil + } + + // Background context: the bus delivery context may be cancelled before + // task creation finishes. The work is independent of the publisher. + go s.createLinearIssueTask(context.Background(), evt) + return nil +} + +// createLinearIssueTask mirrors createJiraIssueTask intentionally: the dedup- +// reserve → create → assign-or-release lifecycle is the same shape across +// both integrations. Refactoring out the symmetry would couple two +// integration packages we'd rather keep independent (their event payloads, +// metadata keys, and prompt placeholders all differ). +// +//nolint:dupl // intentional symmetry with createJiraIssueTask; see comment above +func (s *Service) createLinearIssueTask(ctx context.Context, evt *linear.NewLinearIssueEvent) { + issue := evt.Issue + if !s.reserveLinearIssue(ctx, evt) { + return + } + + req := &IssueTaskRequest{ + WorkspaceID: evt.WorkspaceID, + WorkflowID: evt.WorkflowID, + WorkflowStepID: evt.WorkflowStepID, + Title: fmt.Sprintf("[%s] %s", issue.Identifier, issue.Title), + Description: interpolateLinearPrompt(evt.Prompt, issue), + Metadata: map[string]interface{}{ + "linear_issue_watch_id": evt.IssueWatchID, + "linear_issue_identifier": issue.Identifier, + "linear_issue_url": issue.URL, + "linear_state": issue.StateName, + "linear_assignee": issue.AssigneeName, + "agent_profile_id": evt.AgentProfileID, + "executor_profile_id": evt.ExecutorProfileID, + }, + } + + task, err := s.issueTaskCreator.CreateIssueTask(ctx, req) + if err != nil { + s.logger.Error("failed to create linear issue task", + zap.String("issue_watch_id", evt.IssueWatchID), + zap.String("identifier", issue.Identifier), + zap.Error(err)) + s.releaseLinearIssue(ctx, evt) + return + } + + s.attachLinearIssueTaskID(ctx, evt, task.ID) + s.logger.Info("created linear issue task", + zap.String("task_id", task.ID), + zap.String("identifier", issue.Identifier)) + + if !s.shouldAutoStartStep(ctx, evt.WorkflowStepID) { + return + } + if _, err := s.StartTask( + ctx, task.ID, evt.AgentProfileID, "", evt.ExecutorProfileID, + 0, task.Description, evt.WorkflowStepID, false, nil, + ); err != nil { + s.logger.Error("failed to auto-start linear issue task", + zap.String("task_id", task.ID), zap.Error(err)) + return + } + s.logger.Info("auto-started linear issue task", + zap.String("task_id", task.ID), + zap.String("identifier", issue.Identifier)) +} + +// reserveLinearIssue claims the dedup slot before task creation. When the +// linear service isn't wired (boot order corner case), proceed anyway — +// better to risk a duplicate task than silently drop the event. +func (s *Service) reserveLinearIssue(ctx context.Context, evt *linear.NewLinearIssueEvent) bool { + if s.linearService == nil { + return true + } + reserved, err := s.linearService.ReserveIssueWatchTask(ctx, evt.IssueWatchID, evt.Issue.Identifier, evt.Issue.URL) + if err != nil { + s.logger.Error("failed to reserve linear issue slot", + zap.String("identifier", evt.Issue.Identifier), zap.Error(err)) + return false + } + if !reserved { + s.logger.Debug("linear issue already reserved by concurrent handler, skipping", + zap.String("identifier", evt.Issue.Identifier)) + } + return reserved +} + +func (s *Service) releaseLinearIssue(ctx context.Context, evt *linear.NewLinearIssueEvent) { + if s.linearService == nil { + return + } + if err := s.linearService.ReleaseIssueWatchTask(ctx, evt.IssueWatchID, evt.Issue.Identifier); err != nil { + s.logger.Warn("failed to release linear issue reservation after task-create failure", + zap.String("identifier", evt.Issue.Identifier), zap.Error(err)) + } +} + +func (s *Service) attachLinearIssueTaskID(ctx context.Context, evt *linear.NewLinearIssueEvent, taskID string) { + if s.linearService == nil { + return + } + if err := s.linearService.AssignIssueWatchTaskID(ctx, evt.IssueWatchID, evt.Issue.Identifier, taskID); err != nil { + s.logger.Error("failed to assign task ID to linear issue reservation", + zap.String("task_id", taskID), + zap.String("identifier", evt.Issue.Identifier), zap.Error(err)) + } +} + +// interpolateLinearPrompt replaces {{issue.*}} placeholders with issue fields. +// When the template is empty (user didn't configure a custom prompt), it falls +// back to the embedded default at config/prompts/linear-issue-watch-default.md +// — same pattern as the github + jira watchers, so prompt content is editable +// without redeploying. +func interpolateLinearPrompt(template string, i *linear.LinearIssue) string { + if strings.TrimSpace(template) == "" { + template = promptcfg.Get("linear-issue-watch-default") + } + r := strings.NewReplacer( + "{{issue.identifier}}", i.Identifier, + "{{issue.title}}", i.Title, + "{{issue.url}}", i.URL, + "{{issue.state}}", i.StateName, + "{{issue.priority}}", i.PriorityLabel, + "{{issue.team}}", i.TeamKey, + "{{issue.assignee}}", i.AssigneeName, + "{{issue.creator}}", i.CreatorName, + "{{issue.description}}", i.Description, + ) + return r.Replace(template) +} diff --git a/apps/backend/internal/orchestrator/event_handlers_linear_test.go b/apps/backend/internal/orchestrator/event_handlers_linear_test.go new file mode 100644 index 0000000000..6a1f83e8ac --- /dev/null +++ b/apps/backend/internal/orchestrator/event_handlers_linear_test.go @@ -0,0 +1,43 @@ +package orchestrator + +import ( + "strings" + "testing" + + "github.com/kandev/kandev/internal/linear" +) + +func TestInterpolateLinearPrompt(t *testing.T) { + issue := &linear.LinearIssue{ + Identifier: "ENG-7", + Title: "Login fails on mobile", + URL: "https://linear.app/acme/issue/ENG-7", + StateName: "In Progress", + PriorityLabel: "High", + TeamKey: "ENG", + AssigneeName: "Alice", + CreatorName: "Bob", + Description: "Tap submit, nothing happens.", + } + + // Empty template falls back to the embedded default; assert on the load- + // bearing fields rather than the full string so editing the markdown + // doesn't break the test. + got := interpolateLinearPrompt("", issue) + if !strings.Contains(got, "ENG-7") { + t.Errorf("default template missing identifier: %q", got) + } + if !strings.Contains(got, "https://linear.app/acme/issue/ENG-7") { + t.Errorf("default template missing URL: %q", got) + } + + // All placeholders interpolate. + got = interpolateLinearPrompt( + "{{issue.identifier}} | {{issue.title}} | {{issue.state}} | {{issue.priority}} | {{issue.team}} | {{issue.assignee}} | {{issue.creator}}", + issue, + ) + want := "ENG-7 | Login fails on mobile | In Progress | High | ENG | Alice | Bob" + if got != want { + t.Errorf("got %q, want %q", got, want) + } +} diff --git a/apps/backend/internal/orchestrator/service.go b/apps/backend/internal/orchestrator/service.go index 2a78e877b2..ba91807405 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -225,6 +225,9 @@ type Service struct { // Jira service for issue watch dedup operations jiraService JiraService + // Linear service for issue watch dedup operations + linearService LinearService + // Repository resolver for cloning + finding/creating repos for review tasks repositoryResolver RepositoryResolver @@ -662,6 +665,9 @@ func (s *Service) Start(ctx context.Context) error { // Subscribe to JIRA integration events s.subscribeJiraEvents() + // Subscribe to Linear integration events + s.subscribeLinearEvents() + // Subscribe to clarification events (cancel-and-resume flow) s.subscribeClarificationEvents() diff --git a/apps/web/components/linear/linear-issue-watch-dialog.tsx b/apps/web/components/linear/linear-issue-watch-dialog.tsx new file mode 100644 index 0000000000..3d790bb0e4 --- /dev/null +++ b/apps/web/components/linear/linear-issue-watch-dialog.tsx @@ -0,0 +1,589 @@ +"use client"; + +import { useState, useEffect, useCallback, useMemo, useRef } 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 { Badge } from "@kandev/ui/badge"; +import { + Dialog, + DialogContent, + DialogHeader, + DialogTitle, + DialogDescription, + DialogFooter, +} from "@kandev/ui/dialog"; +import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@kandev/ui/select"; +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 { useWorkflowSteps, stepPlaceholder } from "@/hooks/use-workflow-steps"; +import { listLinearTeams, listLinearStates } from "@/lib/api/domains/linear-api"; +import { + ScriptEditor, + computeEditorHeight, +} from "@/components/settings/profile-edit/script-editor"; +import { + LINEAR_ISSUE_WATCH_PLACEHOLDERS, + DEFAULT_LINEAR_ISSUE_WATCH_PROMPT, +} from "./linear-issue-watch-placeholders"; +import type { + CreateLinearIssueWatchInput, + LinearIssueWatch, + LinearTeam, + LinearWorkflowState, + UpdateLinearIssueWatchInput, +} from "@/lib/types/linear"; + +type Props = { + open: boolean; + onOpenChange: (open: boolean) => void; + watch: LinearIssueWatch | null; + workspaceId?: string; + onCreate: (req: CreateLinearIssueWatchInput) => Promise; + onUpdate: (id: string, req: UpdateLinearIssueWatchInput) => Promise; +}; + +type FormState = { + workspaceId: string; + query: string; + teamKey: string; + stateIds: string[]; + assigned: string; + workflowId: string; + workflowStepId: string; + agentProfileId: string; + executorProfileId: string; + prompt: string; + enabled: boolean; + pollInterval: number; +}; + +const ASSIGNED_ANY = "__any__"; + +function makeEmptyForm(workspaceId: string): FormState { + return { + workspaceId, + query: "", + teamKey: "", + stateIds: [], + assigned: "", + workflowId: "", + workflowStepId: "", + agentProfileId: "", + executorProfileId: "", + prompt: DEFAULT_LINEAR_ISSUE_WATCH_PROMPT, + enabled: true, + pollInterval: 300, + }; +} + +function formStateFromWatch(w: LinearIssueWatch): FormState { + return { + workspaceId: w.workspaceId, + query: w.filter?.query ?? "", + teamKey: w.filter?.teamKey ?? "", + stateIds: w.filter?.stateIds ?? [], + assigned: w.filter?.assigned ?? "", + workflowId: w.workflowId, + workflowStepId: w.workflowStepId, + agentProfileId: w.agentProfileId, + executorProfileId: w.executorProfileId, + prompt: w.prompt || DEFAULT_LINEAR_ISSUE_WATCH_PROMPT, + enabled: w.enabled, + pollInterval: w.pollIntervalSeconds, + }; +} + +function useFormData(workspaceId: string) { + useSettingsData(true); + useWorkflows(workspaceId, true); + const allWorkflows = useAppStore((s) => s.workflows.items); + const workflows = useMemo(() => allWorkflows.filter((w) => !w.hidden), [allWorkflows]); + const agentProfiles = useAppStore((s) => s.agentProfiles.items); + const executors = useAppStore((s) => s.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 SelectField(props: { + label: string; + description?: string; + value: string; + onChange: (v: string) => void; + placeholder: string; + items: { id: string; label: string }[]; + disabled?: boolean; +}) { + return ( +
+ + {props.description &&

{props.description}

} + +
+ ); +} + +function StateMultiSelect({ + states, + loading, + selected, + onToggle, + disabled, +}: { + states: LinearWorkflowState[]; + loading: boolean; + selected: string[]; + onToggle: (id: string) => void; + disabled: boolean; +}) { + if (disabled) { + return ( +

+ Pick a team to choose specific workflow states. +

+ ); + } + if (loading) { + return

Loading states…

; + } + if (states.length === 0) { + return

No workflow states available.

; + } + return ( +
+ {states.map((s) => { + const active = selected.includes(s.id); + return ( + onToggle(s.id)} + > + {s.name} + + ); + })} +
+ ); +} + +// useTeamsAndStates loads the team list once Linear is configured, plus the +// states for the currently-selected team. The states map is keyed by teamKey +// so switching teams renders an empty list without us having to setState in +// an effect — only the lookup expression changes. +function useTeamsAndStates(teamKey: string) { + const [teams, setTeams] = useState([]); + const [statesByTeam, setStatesByTeam] = useState>({}); + // Track which teams we've already kicked off a fetch for so the effect + // doesn't list `statesByTeam` as a dep (which would re-fire whenever any + // team's states load). A ref mutation isn't reactive, so the sole input + // that schedules a fetch is `teamKey`. + const fetchedTeams = useRef>(new Set()); + + useEffect(() => { + let cancelled = false; + listLinearTeams() + .then((res) => { + if (!cancelled) setTeams(res.teams ?? []); + }) + .catch(() => { + if (!cancelled) setTeams([]); + }); + return () => { + cancelled = true; + }; + }, []); + + useEffect(() => { + if (!teamKey || fetchedTeams.current.has(teamKey)) return; + fetchedTeams.current.add(teamKey); + let cancelled = false; + listLinearStates(teamKey) + .then((res) => { + if (!cancelled) setStatesByTeam((prev) => ({ ...prev, [teamKey]: res.states ?? [] })); + }) + .catch(() => { + if (!cancelled) setStatesByTeam((prev) => ({ ...prev, [teamKey]: [] })); + }); + return () => { + cancelled = true; + }; + }, [teamKey]); + + const states = teamKey ? (statesByTeam[teamKey] ?? []) : []; + // "loading" = a team is selected but we haven't received a result yet. + const loadingStates = !!teamKey && statesByTeam[teamKey] === undefined; + return { teams, states, loadingStates }; +} + +function FilterFields({ + form, + setForm, +}: { + form: FormState; + setForm: React.Dispatch>; +}) { + const { teams, states, loadingStates } = useTeamsAndStates(form.teamKey); + const toggleState = useCallback( + (id: string) => + setForm((p) => ({ + ...p, + stateIds: p.stateIds.includes(id) + ? p.stateIds.filter((s) => s !== id) + : [...p.stateIds, id], + })), + [setForm], + ); + + return ( + <> +
+ setForm((p) => ({ ...p, teamKey: v, stateIds: [] }))} + placeholder="(any team)" + items={teams.map((t) => ({ id: t.key, label: `${t.name} (${t.key})` }))} + /> + setForm((p) => ({ ...p, assigned: v === ASSIGNED_ANY ? "" : v }))} + placeholder="(any)" + items={[ + { id: ASSIGNED_ANY, label: "(any)" }, + { id: "me", label: "Me" }, + { id: "unassigned", label: "Unassigned" }, + ]} + /> +
+
+ +

+ Click states to toggle. Empty matches every state on the team. +

+ +
+
+ +

+ Free-text match across title and description (optional). +

+ setForm((p) => ({ ...p, query: e.target.value }))} + placeholder="auth bug" + /> +
+ + ); +} + +function PlaceholdersHelp() { + return ( + + + + + + +

Available placeholders:

+
    + {LINEAR_ISSUE_WATCH_PLACEHOLDERS.map((p) => ( +
  • + {`{{${p.key}}}`}{" "} + {p.description} +
    • + ))} +
+ + + + ); +} + +function PromptField({ value, onChange }: { value: string; onChange: (v: string) => void }) { + return ( +
+
+ + +
+

+ The prompt sent to the agent for each new issue. Type {"{{"} to insert placeholders. +

+
+ +
+
+ ); +} + +function WorkspacePicker({ + value, + onChange, + disabled, +}: { + value: string; + onChange: (v: string) => void; + disabled?: boolean; +}) { + const workspaces = useAppStore((s) => s.workspaces.items); + return ( + ({ id: w.id, label: w.name }))} + disabled={disabled} + /> + ); +} + +function AutomationFields({ + form, + setForm, +}: { + form: FormState; + setForm: React.Dispatch>; +}) { + const { workflows, agentProfiles, allExecutorProfiles } = useFormData(form.workspaceId); + const { steps, loading: stepsLoading } = useWorkflowSteps(form.workflowId); + return ( + <> +
+ setForm((p) => ({ ...p, workflowId: v, workflowStepId: "" }))} + placeholder="Select workflow" + items={workflows.map((w) => ({ id: w.id, label: w.name }))} + /> + setForm((p) => ({ ...p, workflowStepId: v }))} + placeholder={stepPlaceholder(form.workflowId, stepsLoading, steps.length)} + items={steps.map((s) => ({ id: s.id, label: s.name }))} + disabled={!form.workflowId || stepsLoading || steps.length === 0} + /> +
+
+ setForm((p) => ({ ...p, agentProfileId: v }))} + placeholder="(use step default)" + items={agentProfiles.map((p) => ({ id: p.id, label: p.label }))} + /> + setForm((p) => ({ ...p, executorProfileId: v }))} + placeholder="(use step default)" + items={allExecutorProfiles.map((p) => ({ id: p.id, label: p.name }))} + /> +
+ + ); +} + +function SettingsFields({ + form, + setForm, +}: { + form: FormState; + setForm: React.Dispatch>; +}) { + return ( + <> +
+ +

+ How often to re-run the search. Minimum 60s, maximum 3600s. +

+ setForm((p) => ({ ...p, pollInterval: Number(e.target.value) }))} + min={60} + max={3600} + /> +
+
+
+ +

Pause or resume polling.

+
+ setForm((p) => ({ ...p, enabled: v }))} + className="cursor-pointer" + /> +
+ + ); +} + +function savingLabel(saving: boolean, isEdit: boolean): string { + if (saving) return "Saving…"; + return isEdit ? "Update" : "Create"; +} + +function filterIsEmpty(form: FormState): boolean { + return ( + form.query.trim() === "" && + form.teamKey.trim() === "" && + form.assigned.trim() === "" && + form.stateIds.length === 0 + ); +} + +export function LinearIssueWatchDialog({ + open, + onOpenChange, + watch, + workspaceId, + onCreate, + onUpdate, +}: Props) { + const activeWorkspaceId = useAppStore((s) => s.workspaces.activeId); + const [saving, setSaving] = useState(false); + const [form, setForm] = useState(() => makeEmptyForm(workspaceId ?? "")); + + useEffect(() => { + if (watch) { + setForm(formStateFromWatch(watch)); + } else { + setForm(makeEmptyForm(workspaceId ?? activeWorkspaceId ?? "")); + } + }, [watch, open, workspaceId, activeWorkspaceId]); + + const workspaceLocked = !!watch || !!workspaceId; + + const canSave = + !!form.workspaceId && + !filterIsEmpty(form) && + !!form.workflowId && + !!form.workflowStepId && + !!form.prompt.trim(); + + const handleSave = useCallback(async () => { + setSaving(true); + try { + const filter = { + query: form.query.trim() || undefined, + teamKey: form.teamKey.trim() || undefined, + stateIds: form.stateIds.length > 0 ? form.stateIds : undefined, + assigned: form.assigned.trim() || undefined, + }; + const payload = { + filter, + workflowId: form.workflowId, + workflowStepId: form.workflowStepId, + agentProfileId: form.agentProfileId, + executorProfileId: form.executorProfileId, + prompt: form.prompt, + enabled: form.enabled, + pollIntervalSeconds: form.pollInterval, + }; + if (watch) { + await onUpdate(watch.id, payload); + } else { + await onCreate({ ...payload, workspaceId: form.workspaceId }); + } + onOpenChange(false); + } catch { + // Error surfaced by caller's toast. + } finally { + setSaving(false); + } + }, [form, watch, onCreate, onUpdate, onOpenChange]); + + return ( + + + + {watch ? "Edit Linear Watcher" : "Create Linear Watcher"} + + Poll Linear with a structured filter and auto-create a Kandev task for each + newly-matching issue. Issues are not bound to a repository — the workflow step's + defaults decide where the task runs. + + +
+ + setForm((p) => ({ ...p, workspaceId: v, workflowId: "", workflowStepId: "" })) + } + disabled={workspaceLocked} + /> + + + setForm((p) => ({ ...p, prompt: v }))} + /> + +
+ + + + + + + ); +} diff --git a/apps/web/components/linear/linear-issue-watch-placeholders.ts b/apps/web/components/linear/linear-issue-watch-placeholders.ts new file mode 100644 index 0000000000..39faf2c10c --- /dev/null +++ b/apps/web/components/linear/linear-issue-watch-placeholders.ts @@ -0,0 +1,86 @@ +import type { ScriptPlaceholder } from "@/components/settings/profile-edit/script-editor-completions"; + +export const LINEAR_ISSUE_WATCH_PLACEHOLDERS: ScriptPlaceholder[] = [ + { + key: "issue.url", + description: "Linear issue URL", + example: "https://linear.app/acme/issue/ENG-7", + executor_types: [], + }, + { + key: "issue.identifier", + description: "Issue identifier", + example: "ENG-7", + executor_types: [], + }, + { + key: "issue.title", + description: "Issue title", + example: "Login fails on mobile", + executor_types: [], + }, + { + key: "issue.team", + description: "Team key", + example: "ENG", + executor_types: [], + }, + { + key: "issue.state", + description: "Workflow state name", + example: "In Progress", + executor_types: [], + }, + { + key: "issue.priority", + description: "Priority label", + example: "High", + executor_types: [], + }, + { + key: "issue.assignee", + description: "Assignee display name", + example: "Alice", + executor_types: [], + }, + { + key: "issue.creator", + description: "Issue creator display name", + example: "Bob", + executor_types: [], + }, + { + key: "issue.description", + description: "Issue description body", + example: "Tap submit, nothing happens.", + executor_types: [], + }, +]; + +// DEFAULT_LINEAR_ISSUE_WATCH_PROMPT mirrors apps/backend/config/prompts/linear-issue-watch-default.md. +// Kept in sync by hand: the UI shows this when the user clears the field, and +// the backend reads the .md when the saved prompt is empty. Diverging would +// surprise the user — they'd see one default in the dialog and another get +// sent to the agent. +export const DEFAULT_LINEAR_ISSUE_WATCH_PROMPT = `You have been assigned a Linear issue to work on. + +**Issue:** {{issue.url}} +**Identifier:** {{issue.identifier}} +**Title:** {{issue.title}} +**Team:** {{issue.team}} +**State:** {{issue.state}} +**Priority:** {{issue.priority}} +**Assignee:** {{issue.assignee}} + +## Description + +{{issue.description}} + +## 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 {{issue.identifier}}.`; diff --git a/apps/web/components/linear/linear-issue-watch-table.tsx b/apps/web/components/linear/linear-issue-watch-table.tsx new file mode 100644 index 0000000000..96de193788 --- /dev/null +++ b/apps/web/components/linear/linear-issue-watch-table.tsx @@ -0,0 +1,187 @@ +"use client"; + +import { IconTrash, IconRefresh, IconPlayerPlay, IconPlayerPause } 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"; +import { Tooltip, TooltipContent, TooltipTrigger } from "@kandev/ui/tooltip"; +import { useAppStore } from "@/components/state-provider"; +import type { LinearIssueWatch, LinearSearchFilter } from "@/lib/types/linear"; + +type LinearIssueWatchTableProps = { + watches: LinearIssueWatch[]; + // showWorkspace renders a Workspace column when the table aggregates rows + // from every workspace (install-wide settings page). + showWorkspace?: boolean; + onEdit: (watch: LinearIssueWatch) => void; + onDelete: (id: string) => void; + onTrigger: (id: string) => void; + onToggleEnabled: (watch: LinearIssueWatch) => void; +}; + +function formatLastPolled(dateStr?: string | null): string { + if (!dateStr) return "Never"; + const diff = Date.now() - new Date(dateStr).getTime(); + const minutes = Math.floor(diff / 60000); + if (minutes < 1) return "Just now"; + if (minutes < 60) return `${minutes}m ago`; + const hours = Math.floor(minutes / 60); + if (hours < 24) return `${hours}h ago`; + return `${Math.floor(hours / 24)}d ago`; +} + +// summarizeFilter renders the structured filter as a short tag-style label +// the user can scan at a glance. Falls back to "(any)" when somehow empty — +// the backend rejects empty filters at create/update time, so this is just +// defensive for forward-compat. +function summarizeFilter(filter: LinearSearchFilter | undefined): string { + if (!filter) return "(any)"; + const parts: string[] = []; + if (filter.teamKey) parts.push(`team:${filter.teamKey}`); + if (filter.stateIds && filter.stateIds.length > 0) { + parts.push(`state:${filter.stateIds.length} selected`); + } + if (filter.assigned) parts.push(`assigned:${filter.assigned}`); + if (filter.query) parts.push(`q:"${filter.query}"`); + return parts.length === 0 ? "(any)" : parts.join(" · "); +} + +function WatchActions({ + watch, + onToggleEnabled, + onTrigger, + onDelete, +}: { + watch: LinearIssueWatch; + onToggleEnabled: (watch: LinearIssueWatch) => void; + onTrigger: (id: string) => void; + onDelete: (id: string) => void; +}) { + return ( +
+ + + + + {watch.enabled ? "Pause" : "Enable"} + + + + + + Check now + + + + + + Delete + +
+ ); +} + +export function LinearIssueWatchTable({ + watches, + showWorkspace, + onEdit, + onDelete, + onTrigger, + onToggleEnabled, +}: LinearIssueWatchTableProps) { + const workspaces = useAppStore((s) => s.workspaces.items); + const workspaceName = (id: string) => workspaces.find((w) => w.id === id)?.name ?? id; + + if (watches.length === 0) { + return ( +

+ No Linear watchers configured. Create one to auto-create tasks from filtered issues. +

+ ); + } + + return ( + + + + {showWorkspace && Workspace} + Filter + Interval + Last Polled + Status + Actions + + + + {watches.map((watch) => ( + onEdit(watch)}> + {showWorkspace && ( + + {workspaceName(watch.workspaceId)} + + )} + + {summarizeFilter(watch.filter)} + + + {Math.round(watch.pollIntervalSeconds / 60)}m + + + {formatLastPolled(watch.lastPolledAt)} + + + + {watch.enabled ? "Active" : "Paused"} + + + + + + + ))} + +
+ ); +} diff --git a/apps/web/components/linear/linear-issue-watchers-section.tsx b/apps/web/components/linear/linear-issue-watchers-section.tsx new file mode 100644 index 0000000000..8c627cb742 --- /dev/null +++ b/apps/web/components/linear/linear-issue-watchers-section.tsx @@ -0,0 +1,177 @@ +"use client"; + +import { useCallback, useState } from "react"; +import { IconBellRinging, IconPlus } from "@tabler/icons-react"; +import { Button } from "@kandev/ui/button"; +import { Card, CardContent } from "@kandev/ui/card"; +import { SettingsSection } from "@/components/settings/settings-section"; +import { useToast } from "@/components/toast-provider"; +import { useLinearIssueWatches } from "@/hooks/domains/linear/use-linear-issue-watches"; +import { LinearIssueWatchTable } from "./linear-issue-watch-table"; +import { LinearIssueWatchDialog } from "./linear-issue-watch-dialog"; +import type { LinearIssueWatch } from "@/lib/types/linear"; + +// LinearIssueWatchersSection lists watches across every workspace in a single +// flat table on the install-wide settings page. The dialog's create flow asks +// the user to pick the workspace; per-row mutations forward each watch's +// stored workspaceId so the backend's IDOR guard accepts them. +type RawActions = { + create: ReturnType["create"]; + update: ReturnType["update"]; + remove: ReturnType["remove"]; + trigger: ReturnType["trigger"]; +}; + +function useToastedActions({ create, update, remove, trigger }: RawActions) { + const { toast } = useToast(); + + const wrappedCreate = useCallback( + async (req: Parameters[0]) => { + try { + await create(req); + toast({ description: "Watcher created", variant: "success" }); + } catch (err) { + toast({ description: `Create failed: ${String(err)}`, variant: "error" }); + throw err; + } + }, + [create, toast], + ); + + const wrappedUpdate = useCallback( + async (id: string, req: Parameters[1], rowWorkspaceId: string) => { + try { + await update(id, req, rowWorkspaceId); + toast({ description: "Watcher updated", variant: "success" }); + } catch (err) { + toast({ description: `Update failed: ${String(err)}`, variant: "error" }); + throw err; + } + }, + [update, toast], + ); + + const wrappedDelete = useCallback( + async (w: LinearIssueWatch) => { + if (!confirm("Delete this Linear watcher?")) return; + try { + await remove(w.id, w.workspaceId); + toast({ description: "Watcher deleted", variant: "success" }); + } catch (err) { + toast({ description: `Delete failed: ${String(err)}`, variant: "error" }); + } + }, + [remove, toast], + ); + + const wrappedTrigger = useCallback( + async (w: LinearIssueWatch) => { + try { + const res = await trigger(w.id, w.workspaceId); + const n = res?.newIssues ?? 0; + const description = + n > 0 ? `Found ${n} new issue(s) — tasks will appear shortly.` : "No new issues matched."; + toast({ description, variant: "success" }); + } catch (err) { + toast({ description: `Check failed: ${String(err)}`, variant: "error" }); + } + }, + [trigger, toast], + ); + + const toggleEnabled = useCallback( + async (w: LinearIssueWatch) => { + try { + await update(w.id, { enabled: !w.enabled }, w.workspaceId); + } catch (err) { + toast({ description: `Toggle failed: ${String(err)}`, variant: "error" }); + } + }, + [update, toast], + ); + + return { + create: wrappedCreate, + update: wrappedUpdate, + remove: wrappedDelete, + trigger: wrappedTrigger, + toggleEnabled, + }; +} + +export function LinearIssueWatchersSection() { + const { items, loading, create, update, remove, trigger } = useLinearIssueWatches(); + const actions = useToastedActions({ create, update, remove, trigger }); + + const [dialogOpen, setDialogOpen] = useState(false); + const [editing, setEditing] = useState(null); + + const openCreate = useCallback(() => { + setEditing(null); + setDialogOpen(true); + }, []); + const openEdit = useCallback((w: LinearIssueWatch) => { + setEditing(w); + setDialogOpen(true); + }, []); + + // Adapt the watch-aware actions back to id-keyed callbacks the table expects; + // the table looks up the watch by id when it needs to forward the per-row + // workspaceId to mutations. + const handleDelete = useCallback( + (id: string) => { + const w = items.find((item) => item.id === id); + if (w) actions.remove(w); + }, + [items, actions], + ); + const handleTrigger = useCallback( + (id: string) => { + const w = items.find((item) => item.id === id); + if (w) actions.trigger(w); + }, + [items, actions], + ); + + return ( + } + title="Linear watchers" + description="Poll a Linear filter and auto-create a Kandev task for each newly-matching issue." + action={ + + } + > + + + {loading && items.length === 0 ? ( +

Loading…

+ ) : ( + + )} + + + { + const w = editing; + if (!w) throw new Error("update without editing watch"); + return actions.update(id, req, w.workspaceId); + }} + /> + + ); +} diff --git a/apps/web/components/linear/linear-settings.tsx b/apps/web/components/linear/linear-settings.tsx index 7ca805f299..d7dcc35cfa 100644 --- a/apps/web/components/linear/linear-settings.tsx +++ b/apps/web/components/linear/linear-settings.tsx @@ -26,6 +26,7 @@ import { listLinearTeams, } from "@/lib/api/domains/linear-api"; import type { LinearConfig, LinearTeam, TestLinearConnectionResult } from "@/lib/types/linear"; +import { LinearIssueWatchersSection } from "./linear-issue-watchers-section"; type FormState = { defaultTeamKey: string; @@ -432,6 +433,7 @@ export function LinearIntegrationPage() { return (
+
); } diff --git a/apps/web/hooks/domains/linear/use-linear-issue-watches.ts b/apps/web/hooks/domains/linear/use-linear-issue-watches.ts new file mode 100644 index 0000000000..85eb5163c0 --- /dev/null +++ b/apps/web/hooks/domains/linear/use-linear-issue-watches.ts @@ -0,0 +1,100 @@ +"use client"; + +import { useEffect, useCallback, useRef } from "react"; +import { + listLinearIssueWatches, + createLinearIssueWatch, + updateLinearIssueWatch, + deleteLinearIssueWatch, + triggerLinearIssueWatch, +} from "@/lib/api/domains/linear-api"; +import { useAppStore } from "@/components/state-provider"; +import type { CreateLinearIssueWatchInput, UpdateLinearIssueWatchInput } from "@/lib/types/linear"; + +/** + * useLinearIssueWatches owns the Linear-watcher list: + * - workspaceId: string → fetch and operate on watches in one workspace + * - workspaceId: undefined → fetch every watch across all workspaces; the + * caller supplies workspaceId to update/remove/trigger + * calls per-row (those endpoints still validate it + * against the watch's stored workspace as an IDOR guard) + * - workspaceId: null → don't fetch + * + * Mirrors `useJiraIssueWatches`. Workspace changes reset the cached list so + * the user doesn't see the previous workspace's stale rows during the swap. + */ +export function useLinearIssueWatches(workspaceId?: string | null) { + const items = useAppStore((s) => s.linearIssueWatches.items); + const loaded = useAppStore((s) => s.linearIssueWatches.loaded); + const loading = useAppStore((s) => s.linearIssueWatches.loading); + const setWatches = useAppStore((s) => s.setLinearIssueWatches); + const resetWatches = useAppStore((s) => s.resetLinearIssueWatches); + const setLoading = useAppStore((s) => s.setLinearIssueWatchesLoading); + const addWatch = useAppStore((s) => s.addLinearIssueWatch); + const updateWatch = useAppStore((s) => s.updateLinearIssueWatch); + const removeWatch = useAppStore((s) => s.removeLinearIssueWatch); + + const lastScope = useRef(undefined); + const scope: string | null = workspaceId ?? null; + + useEffect(() => { + if (workspaceId === null) return; + if (lastScope.current !== undefined && lastScope.current !== scope) { + resetWatches(); + } + lastScope.current = scope; + }, [workspaceId, scope, resetWatches]); + + useEffect(() => { + if (workspaceId === null || loaded || loading) return; + setLoading(true); + listLinearIssueWatches(workspaceId ?? undefined, { cache: "no-store" }) + .then((res) => setWatches(res ?? [])) + .catch(() => setWatches([])) + .finally(() => setLoading(false)); + }, [workspaceId, loaded, loading, setWatches, setLoading]); + + const create = useCallback( + async (req: CreateLinearIssueWatchInput) => { + const watch = await createLinearIssueWatch(req); + addWatch(watch); + return watch; + }, + [addWatch], + ); + + // Per-row mutations require the row's own workspace_id to satisfy the + // backend's IDOR guard. Callers pass it explicitly when the hook itself + // wasn't bound to a single workspace (the install-wide listing case). + const update = useCallback( + async (id: string, req: UpdateLinearIssueWatchInput, rowWorkspaceId?: string) => { + const ws = rowWorkspaceId ?? workspaceId; + if (!ws) throw new Error("workspaceId required"); + const watch = await updateLinearIssueWatch(ws, id, req); + updateWatch(watch); + return watch; + }, + [workspaceId, updateWatch], + ); + + const remove = useCallback( + async (id: string, rowWorkspaceId?: string) => { + const ws = rowWorkspaceId ?? workspaceId; + if (!ws) throw new Error("workspaceId required"); + await deleteLinearIssueWatch(ws, id); + removeWatch(id); + }, + [workspaceId, removeWatch], + ); + + const trigger = useCallback( + async (id: string, rowWorkspaceId?: string) => { + const ws = rowWorkspaceId ?? workspaceId; + if (!ws) throw new Error("workspaceId required"); + return triggerLinearIssueWatch(ws, id); + }, + [workspaceId], + ); + + return { items, loaded, loading, create, update, remove, trigger }; +} diff --git a/apps/web/lib/api/domains/linear-api.test.ts b/apps/web/lib/api/domains/linear-api.test.ts index f4a3fc6624..e4e409915e 100644 --- a/apps/web/lib/api/domains/linear-api.test.ts +++ b/apps/web/lib/api/domains/linear-api.test.ts @@ -7,15 +7,20 @@ vi.mock("@/lib/config", () => ({ })); import { + createLinearIssueWatch, deleteLinearConfig, + deleteLinearIssueWatch, getLinearConfig, getLinearIssue, + listLinearIssueWatches, listLinearStates, listLinearTeams, searchLinearIssues, setLinearConfig, setLinearIssueState, testLinearConnection, + triggerLinearIssueWatch, + updateLinearIssueWatch, } from "./linear-api"; const BASE = "http://api.test/api/v1/linear"; @@ -173,3 +178,82 @@ describe("setLinearIssueState", () => { expect(JSON.parse(String(init?.body))).toEqual({ stateId: "state-id-123" }); }); }); + +// Issue watches: workspace_id is a query param on list/update/delete/trigger +// but absent on create — exactly the kind of subtle URL-construction +// difference that's worth catching at the API client boundary. +describe("listLinearIssueWatches", () => { + it("hits /watches/issue with no query param when workspaceId omitted", async () => { + fetchSpy.mockResolvedValueOnce(jsonResponse({ watches: [] })); + await listLinearIssueWatches(); + expect(lastCall().url).toBe(`${BASE}/watches/issue`); + }); + + it("appends workspace_id query param when provided", async () => { + fetchSpy.mockResolvedValueOnce(jsonResponse({ watches: [] })); + await listLinearIssueWatches("ws-1"); + expect(lastCall().url).toBe(`${BASE}/watches/issue?workspace_id=ws-1`); + }); + + it("returns an empty array when the response has no watches field", async () => { + fetchSpy.mockResolvedValueOnce(jsonResponse({})); + await expect(listLinearIssueWatches()).resolves.toEqual([]); + }); +}); + +describe("createLinearIssueWatch", () => { + it("POSTs to /watches/issue without workspace_id in the URL", async () => { + fetchSpy.mockResolvedValueOnce(jsonResponse({ id: "w1" })); + await createLinearIssueWatch({ + workspaceId: "ws-1", + workflowId: "wf-1", + workflowStepId: "step-1", + filter: { teamKey: "ENG" }, + }); + const { url, init } = lastCall(); + expect(url).toBe(`${BASE}/watches/issue`); + expect(init?.method).toBe("POST"); + expect(JSON.parse(String(init?.body))).toMatchObject({ + workspaceId: "ws-1", + filter: { teamKey: "ENG" }, + }); + }); +}); + +describe("updateLinearIssueWatch", () => { + it("PATCHes /watches/issue/:id with workspace_id query param", async () => { + fetchSpy.mockResolvedValueOnce(jsonResponse({ id: "w1" })); + await updateLinearIssueWatch("ws-1", "w1", { enabled: false }); + const { url, init } = lastCall(); + expect(url).toBe(`${BASE}/watches/issue/w1?workspace_id=ws-1`); + expect(init?.method).toBe("PATCH"); + expect(JSON.parse(String(init?.body))).toEqual({ enabled: false }); + }); + + it("URL-encodes both id and workspace_id", async () => { + fetchSpy.mockResolvedValueOnce(jsonResponse({ id: "w/1" })); + await updateLinearIssueWatch("ws/space", "w/1", {}); + expect(lastCall().url).toBe(`${BASE}/watches/issue/w%2F1?workspace_id=ws%2Fspace`); + }); +}); + +describe("deleteLinearIssueWatch", () => { + it("issues DELETE on /watches/issue/:id with workspace_id query param", async () => { + fetchSpy.mockResolvedValueOnce(jsonResponse({ deleted: true })); + await deleteLinearIssueWatch("ws-1", "w1"); + const { url, init } = lastCall(); + expect(url).toBe(`${BASE}/watches/issue/w1?workspace_id=ws-1`); + expect(init?.method).toBe("DELETE"); + }); +}); + +describe("triggerLinearIssueWatch", () => { + it("POSTs to /watches/issue/:id/trigger with workspace_id query param", async () => { + fetchSpy.mockResolvedValueOnce(jsonResponse({ newIssues: 3 })); + const res = await triggerLinearIssueWatch("ws-1", "w1"); + const { url, init } = lastCall(); + expect(url).toBe(`${BASE}/watches/issue/w1/trigger?workspace_id=ws-1`); + expect(init?.method).toBe("POST"); + expect(res.newIssues).toBe(3); + }); +}); diff --git a/apps/web/lib/api/domains/linear-api.ts b/apps/web/lib/api/domains/linear-api.ts index 8de4c09e8b..702b488465 100644 --- a/apps/web/lib/api/domains/linear-api.ts +++ b/apps/web/lib/api/domains/linear-api.ts @@ -1,13 +1,16 @@ import { fetchJson, type ApiRequestOptions } from "../client"; import type { + CreateLinearIssueWatchInput, LinearConfig, LinearIssue, + LinearIssueWatch, LinearSearchFilter, LinearSearchResult, LinearTeam, LinearWorkflowState, SetLinearConfigRequest, TestLinearConnectionResult, + UpdateLinearIssueWatchInput, } from "@/lib/types/linear"; // getLinearConfig returns null when the backend responds 204 (no config yet). @@ -90,3 +93,65 @@ export async function setLinearIssueState( }, ); } + +// --- Issue watches --- + +// listLinearIssueWatches fetches watches across all workspaces when +// workspaceId is omitted, or scoped to one workspace when provided. +export async function listLinearIssueWatches(workspaceId?: string, options?: ApiRequestOptions) { + const path = workspaceId + ? `/api/v1/linear/watches/issue?workspace_id=${encodeURIComponent(workspaceId)}` + : `/api/v1/linear/watches/issue`; + const res = await fetchJson<{ watches: LinearIssueWatch[] }>(path, options); + return res.watches ?? []; +} + +export async function createLinearIssueWatch( + payload: CreateLinearIssueWatchInput, + options?: ApiRequestOptions, +) { + return fetchJson(`/api/v1/linear/watches/issue`, { + ...options, + init: { ...(options?.init ?? {}), method: "POST", body: JSON.stringify(payload) }, + }); +} + +// All mutation/trigger endpoints require `workspace_id` so the backend can +// reject cross-workspace IDOR. + +export async function updateLinearIssueWatch( + workspaceId: string, + id: string, + payload: UpdateLinearIssueWatchInput, + options?: ApiRequestOptions, +) { + return fetchJson( + `/api/v1/linear/watches/issue/${encodeURIComponent(id)}?workspace_id=${encodeURIComponent(workspaceId)}`, + { + ...options, + init: { ...(options?.init ?? {}), method: "PATCH", body: JSON.stringify(payload) }, + }, + ); +} + +export async function deleteLinearIssueWatch( + workspaceId: string, + id: string, + options?: ApiRequestOptions, +) { + return fetchJson<{ deleted: boolean }>( + `/api/v1/linear/watches/issue/${encodeURIComponent(id)}?workspace_id=${encodeURIComponent(workspaceId)}`, + { ...options, init: { ...(options?.init ?? {}), method: "DELETE" } }, + ); +} + +export async function triggerLinearIssueWatch( + workspaceId: string, + id: string, + options?: ApiRequestOptions, +) { + return fetchJson<{ newIssues: number }>( + `/api/v1/linear/watches/issue/${encodeURIComponent(id)}/trigger?workspace_id=${encodeURIComponent(workspaceId)}`, + { ...options, init: { ...(options?.init ?? {}), method: "POST" } }, + ); +} diff --git a/apps/web/lib/state/default-state.ts b/apps/web/lib/state/default-state.ts index 327a33681d..8126f74070 100644 --- a/apps/web/lib/state/default-state.ts +++ b/apps/web/lib/state/default-state.ts @@ -7,6 +7,7 @@ import { defaultUIState, defaultGitHubState, defaultJiraState, + defaultLinearState, } from "./slices"; export const defaultState = { @@ -64,6 +65,7 @@ export const defaultState = { issueWatches: defaultGitHubState.issueWatches, actionPresets: defaultGitHubState.actionPresets, jiraIssueWatches: defaultJiraState.jiraIssueWatches, + linearIssueWatches: defaultLinearState.linearIssueWatches, previewPanel: defaultUIState.previewPanel, rightPanel: defaultUIState.rightPanel, diffs: defaultUIState.diffs, @@ -148,6 +150,10 @@ export function mergeInitialState(initialState?: Partial): Default issueWatches: { ...defaultState.issueWatches, ...initialState.issueWatches }, actionPresets: { ...defaultState.actionPresets, ...initialState.actionPresets }, jiraIssueWatches: { ...defaultState.jiraIssueWatches, ...initialState.jiraIssueWatches }, + linearIssueWatches: { + ...defaultState.linearIssueWatches, + ...initialState.linearIssueWatches, + }, previewPanel: { ...defaultState.previewPanel, ...initialState.previewPanel }, rightPanel: { ...defaultState.rightPanel, ...initialState.rightPanel }, diffs: { ...defaultState.diffs, ...initialState.diffs }, diff --git a/apps/web/lib/state/slices/index.ts b/apps/web/lib/state/slices/index.ts index 9bae93e8a8..84e93dd633 100644 --- a/apps/web/lib/state/slices/index.ts +++ b/apps/web/lib/state/slices/index.ts @@ -10,6 +10,7 @@ export { export { createUISlice, defaultUIState } from "./ui/ui-slice"; export { createGitHubSlice, defaultGitHubState } from "./github/github-slice"; export { createJiraSlice, defaultJiraState } from "./jira/jira-slice"; +export { createLinearSlice, defaultLinearState } from "./linear/linear-slice"; // Export types export type { KanbanSlice, KanbanSliceState, KanbanSliceActions } from "./kanban/types"; @@ -29,6 +30,12 @@ export type { JiraSliceActions, JiraIssueWatchesState, } from "./jira/types"; +export type { + LinearSlice, + LinearSliceState, + LinearSliceActions, + LinearIssueWatchesState, +} from "./linear/types"; // Re-export commonly used types from each domain export type { diff --git a/apps/web/lib/state/slices/linear/linear-slice.test.ts b/apps/web/lib/state/slices/linear/linear-slice.test.ts new file mode 100644 index 0000000000..85f0b23f37 --- /dev/null +++ b/apps/web/lib/state/slices/linear/linear-slice.test.ts @@ -0,0 +1,96 @@ +import { describe, expect, it } from "vitest"; +import { create } from "zustand"; +import { immer } from "zustand/middleware/immer"; +import { createLinearSlice } from "./linear-slice"; +import type { LinearSlice } from "./types"; +import type { LinearIssueWatch } from "@/lib/types/linear"; + +function makeStore() { + return create()( + // eslint-disable-next-line @typescript-eslint/no-explicit-any + immer((...a) => ({ ...(createLinearSlice as any)(...a) })), + ); +} + +function watch(id: string, overrides: Partial = {}): LinearIssueWatch { + return { + id, + workspaceId: "ws-1", + workflowId: "wf-1", + workflowStepId: "step-1", + filter: { teamKey: "ENG" }, + agentProfileId: "", + executorProfileId: "", + prompt: "", + enabled: true, + pollIntervalSeconds: 300, + createdAt: "2026-01-01T00:00:00Z", + updatedAt: "2026-01-01T00:00:00Z", + ...overrides, + }; +} + +describe("linear issue-watches slice", () => { + it("starts empty and not loaded", () => { + const store = makeStore(); + const s = store.getState(); + expect(s.linearIssueWatches.items).toEqual([]); + expect(s.linearIssueWatches.loaded).toBe(false); + expect(s.linearIssueWatches.loading).toBe(false); + }); + + it("setLinearIssueWatches replaces items and flips loaded=true", () => { + const store = makeStore(); + store.getState().setLinearIssueWatches([watch("a"), watch("b")]); + const s = store.getState(); + expect(s.linearIssueWatches.items.map((w) => w.id)).toEqual(["a", "b"]); + expect(s.linearIssueWatches.loaded).toBe(true); + }); + + it("setLinearIssueWatchesLoading toggles loading independently", () => { + const store = makeStore(); + store.getState().setLinearIssueWatchesLoading(true); + expect(store.getState().linearIssueWatches.loading).toBe(true); + store.getState().setLinearIssueWatchesLoading(false); + expect(store.getState().linearIssueWatches.loading).toBe(false); + }); + + it("addLinearIssueWatch appends a new entry", () => { + const store = makeStore(); + store.getState().setLinearIssueWatches([watch("a")]); + store.getState().addLinearIssueWatch(watch("b")); + expect(store.getState().linearIssueWatches.items.map((w) => w.id)).toEqual(["a", "b"]); + }); + + it("updateLinearIssueWatch replaces in place by id; missing id is a no-op", () => { + const store = makeStore(); + store + .getState() + .setLinearIssueWatches([watch("a", { filter: { teamKey: "OLD" } }), watch("b")]); + store.getState().updateLinearIssueWatch(watch("a", { filter: { teamKey: "NEW" } })); + expect(store.getState().linearIssueWatches.items[0].filter.teamKey).toBe("NEW"); + + store.getState().updateLinearIssueWatch(watch("ghost", { filter: { teamKey: "X" } })); + expect(store.getState().linearIssueWatches.items.map((w) => w.id)).toEqual(["a", "b"]); + }); + + it("removeLinearIssueWatch filters by id", () => { + const store = makeStore(); + store.getState().setLinearIssueWatches([watch("a"), watch("b"), watch("c")]); + store.getState().removeLinearIssueWatch("b"); + expect(store.getState().linearIssueWatches.items.map((w) => w.id)).toEqual(["a", "c"]); + }); + + it("resetLinearIssueWatches clears items AND loaded so a refetch is triggered", () => { + // The whole point of this action vs. setLinearIssueWatches([]) — an empty + // setWatches keeps loaded=true and would block the fetch effect from + // re-running on workspace switch. + const store = makeStore(); + store.getState().setLinearIssueWatches([watch("a")]); + expect(store.getState().linearIssueWatches.loaded).toBe(true); + + store.getState().resetLinearIssueWatches(); + expect(store.getState().linearIssueWatches.items).toEqual([]); + expect(store.getState().linearIssueWatches.loaded).toBe(false); + }); +}); diff --git a/apps/web/lib/state/slices/linear/linear-slice.ts b/apps/web/lib/state/slices/linear/linear-slice.ts new file mode 100644 index 0000000000..03d6dc1086 --- /dev/null +++ b/apps/web/lib/state/slices/linear/linear-slice.ts @@ -0,0 +1,48 @@ +import type { StateCreator } from "zustand"; +import type { LinearSlice, LinearSliceState } from "./types"; + +export const defaultLinearState: LinearSliceState = { + linearIssueWatches: { items: [], loaded: false, loading: false }, +}; + +type ImmerSet = Parameters< + StateCreator +>[0]; + +export const createLinearSlice: StateCreator< + LinearSlice, + [["zustand/immer", never]], + [], + LinearSlice +> = (set: ImmerSet) => ({ + ...defaultLinearState, + setLinearIssueWatches: (watches) => + set((draft) => { + draft.linearIssueWatches.items = watches; + draft.linearIssueWatches.loaded = true; + }), + setLinearIssueWatchesLoading: (loading) => + set((draft) => { + draft.linearIssueWatches.loading = loading; + }), + addLinearIssueWatch: (watch) => + set((draft) => { + draft.linearIssueWatches.items.push(watch); + }), + updateLinearIssueWatch: (watch) => + set((draft) => { + const idx = draft.linearIssueWatches.items.findIndex((w) => w.id === watch.id); + if (idx >= 0) { + draft.linearIssueWatches.items[idx] = watch; + } + }), + removeLinearIssueWatch: (id) => + set((draft) => { + draft.linearIssueWatches.items = draft.linearIssueWatches.items.filter((w) => w.id !== id); + }), + resetLinearIssueWatches: () => + set((draft) => { + draft.linearIssueWatches.items = []; + draft.linearIssueWatches.loaded = false; + }), +}); diff --git a/apps/web/lib/state/slices/linear/types.ts b/apps/web/lib/state/slices/linear/types.ts new file mode 100644 index 0000000000..4412aa27b7 --- /dev/null +++ b/apps/web/lib/state/slices/linear/types.ts @@ -0,0 +1,27 @@ +import type { LinearIssueWatch } from "@/lib/types/linear"; + +export type LinearIssueWatchesState = { + items: LinearIssueWatch[]; + loaded: boolean; + loading: boolean; +}; + +export type LinearSliceState = { + linearIssueWatches: LinearIssueWatchesState; +}; + +export type LinearSliceActions = { + setLinearIssueWatches: (watches: LinearIssueWatch[]) => void; + setLinearIssueWatchesLoading: (loading: boolean) => void; + addLinearIssueWatch: (watch: LinearIssueWatch) => void; + updateLinearIssueWatch: (watch: LinearIssueWatch) => void; + removeLinearIssueWatch: (id: string) => void; + /** + * Clears items AND `loaded` so the next fetch effect runs again. Distinct + * from `setLinearIssueWatches([])`, which marks the empty list as loaded + * and would prevent a refetch on workspace switch. + */ + resetLinearIssueWatches: () => void; +}; + +export type LinearSlice = LinearSliceState & LinearSliceActions; diff --git a/apps/web/lib/state/store.ts b/apps/web/lib/state/store.ts index c8944b6f2e..5b968cbd49 100644 --- a/apps/web/lib/state/store.ts +++ b/apps/web/lib/state/store.ts @@ -29,6 +29,7 @@ import { createUISlice, createGitHubSlice, createJiraSlice, + createLinearSlice, defaultKanbanState, defaultWorkspaceState, defaultSettingsState, @@ -37,6 +38,7 @@ import { defaultUIState, defaultGitHubState, defaultJiraState, + defaultLinearState, type WorkspaceState, type WorkflowsState, type ExecutorsState, @@ -139,8 +141,13 @@ export type { JiraSliceState, JiraSliceActions, JiraIssueWatchesState, + LinearSlice, + LinearSliceState, + LinearSliceActions, + LinearIssueWatchesState, } from "./slices"; import type { JiraIssueWatch } from "@/lib/types/jira"; +import type { LinearIssueWatch } from "@/lib/types/linear"; // Combined AppState type export type AppState = { @@ -213,6 +220,9 @@ export type AppState = { // JIRA slice jiraIssueWatches: (typeof defaultJiraState)["jiraIssueWatches"]; + // Linear slice + linearIssueWatches: (typeof defaultLinearState)["linearIssueWatches"]; + // UI slice previewPanel: (typeof defaultUIState)["previewPanel"]; rightPanel: (typeof defaultUIState)["rightPanel"]; @@ -260,6 +270,14 @@ export type AppState = { removeJiraIssueWatch: (id: string) => void; resetJiraIssueWatches: () => void; + // Linear actions + setLinearIssueWatches: (watches: LinearIssueWatch[]) => void; + setLinearIssueWatchesLoading: (loading: boolean) => void; + addLinearIssueWatch: (watch: LinearIssueWatch) => void; + updateLinearIssueWatch: (watch: LinearIssueWatch) => void; + removeLinearIssueWatch: (id: string) => void; + resetLinearIssueWatches: () => void; + // Actions from all slices hydrate: (state: Partial, options?: HydrationOptions) => void; setActiveWorkspace: (workspaceId: string | null) => void; @@ -506,6 +524,8 @@ export function createAppStore(initialState?: Partial) { // eslint-disable-next-line @typescript-eslint/no-explicit-any ...createJiraSlice(set as any, get as any, api as any), // eslint-disable-next-line @typescript-eslint/no-explicit-any + ...createLinearSlice(set as any, get as any, api as any), + // eslint-disable-next-line @typescript-eslint/no-explicit-any ...createUISlice(set as any, get as any, api as any), // Override state with merged initial state kanban: merged.kanban, @@ -558,6 +578,7 @@ export function createAppStore(initialState?: Partial) { issueWatches: merged.issueWatches, actionPresets: merged.actionPresets, jiraIssueWatches: merged.jiraIssueWatches, + linearIssueWatches: merged.linearIssueWatches, previewPanel: merged.previewPanel, rightPanel: merged.rightPanel, diffs: merged.diffs, diff --git a/apps/web/lib/types/linear.ts b/apps/web/lib/types/linear.ts index 12f0792751..73ce0613bd 100644 --- a/apps/web/lib/types/linear.ts +++ b/apps/web/lib/types/linear.ts @@ -89,3 +89,49 @@ export interface LinearSearchResult { isLast: boolean; nextPageToken?: string; } + +/** + * A workspace-scoped Linear poller. The backend re-evaluates the structured + * filter on `pollIntervalSeconds` cadence and creates a Kandev task in the + * configured workflow step for each newly-matching issue. + */ +export interface LinearIssueWatch { + id: string; + workspaceId: string; + workflowId: string; + workflowStepId: string; + filter: LinearSearchFilter; + agentProfileId: string; + executorProfileId: string; + prompt: string; + enabled: boolean; + pollIntervalSeconds: number; + /** Last poll timestamp, or null when the watch has never run. */ + lastPolledAt?: string | null; + createdAt: string; + updatedAt: string; +} + +export interface CreateLinearIssueWatchInput { + workspaceId: string; + workflowId: string; + workflowStepId: string; + filter: LinearSearchFilter; + agentProfileId?: string; + executorProfileId?: string; + prompt?: string; + pollIntervalSeconds?: number; + enabled?: boolean; +} + +/** Patch shape: every field is optional so the UI can change one knob at a time. */ +export interface UpdateLinearIssueWatchInput { + workflowId?: string; + workflowStepId?: string; + filter?: LinearSearchFilter; + agentProfileId?: string; + executorProfileId?: string; + prompt?: string; + enabled?: boolean; + pollIntervalSeconds?: number; +}