diff --git a/apps/backend/cmd/kandev/main.go b/apps/backend/cmd/kandev/main.go index f863fa620c..e55425580b 100644 --- a/apps/backend/cmd/kandev/main.go +++ b/apps/backend/cmd/kandev/main.go @@ -333,10 +333,12 @@ func startAgentInfrastructure( log.Info("GitHub poller started") } - // Start JIRA auth-health poller. Probes stored credentials for every - // configured workspace on a fixed cadence so the UI can show a real - // connected/disconnected status without doing its own polling. + // Start JIRA poller. Drives two background loops sharing one service: an + // auth-health probe (so the UI can show connect status without polling + // JIRA itself) and an issue-watch loop that runs configured JQL queries + // and emits NewJiraIssueEvent for the orchestrator to turn into tasks. if services.Jira != nil { + orchestratorSvc.SetJiraService(&jiraServiceAdapter{svc: services.Jira}) jiraPoller := jirapkg.NewPoller(services.Jira, log) jiraPoller.Start(ctx) addCleanup(func() error { jiraPoller.Stop(); return nil }) diff --git a/apps/backend/cmd/kandev/orchestrator.go b/apps/backend/cmd/kandev/orchestrator.go index fa6514b6cf..157a719dfa 100644 --- a/apps/backend/cmd/kandev/orchestrator.go +++ b/apps/backend/cmd/kandev/orchestrator.go @@ -16,6 +16,7 @@ import ( "github.com/kandev/kandev/internal/common/config" "github.com/kandev/kandev/internal/common/logger" "github.com/kandev/kandev/internal/events/bus" + jirapkg "github.com/kandev/kandev/internal/jira" "github.com/kandev/kandev/internal/orchestrator" "github.com/kandev/kandev/internal/repoclone" "github.com/kandev/kandev/internal/secrets" @@ -248,6 +249,24 @@ func (a *issueTaskCreatorAdapter) CreateIssueTask(ctx context.Context, req *orch }) } +// jiraServiceAdapter exposes the JIRA service's issue-watch dedup methods to +// the orchestrator without leaking the rest of the package surface area. +type jiraServiceAdapter struct { + svc *jirapkg.Service +} + +func (a *jiraServiceAdapter) ReserveIssueWatchTask(ctx context.Context, watchID, issueKey, issueURL string) (bool, error) { + return a.svc.Store().ReserveIssueWatchTask(ctx, watchID, issueKey, issueURL) +} + +func (a *jiraServiceAdapter) AssignIssueWatchTaskID(ctx context.Context, watchID, issueKey, taskID string) error { + return a.svc.Store().AssignIssueWatchTaskID(ctx, watchID, issueKey, taskID) +} + +func (a *jiraServiceAdapter) ReleaseIssueWatchTask(ctx context.Context, watchID, issueKey string) error { + return a.svc.Store().ReleaseIssueWatchTask(ctx, watchID, issueKey) +} + // 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 0805f8584f..ea89ac6913 100644 --- a/apps/backend/cmd/kandev/services.go +++ b/apps/backend/cmd/kandev/services.go @@ -97,7 +97,7 @@ func provideServices(cfg *config.Config, log *logger.Logger, repos *Repositories // Initialize JIRA service jiraSecrets := &jiraSecretAdapter{store: repos.Secrets} - jiraSvc, _, jiraErr := jira.Provide(dbPool.Writer(), dbPool.Reader(), jiraSecrets, log) + jiraSvc, _, jiraErr := jira.Provide(dbPool.Writer(), dbPool.Reader(), jiraSecrets, eventBus, log) if jiraErr != nil { log.Warn("JIRA service initialization failed (non-fatal)", zap.Error(jiraErr)) } diff --git a/apps/backend/internal/events/types.go b/apps/backend/internal/events/types.go index 0b4813ca5d..8b8ff9767d 100644 --- a/apps/backend/internal/events/types.go +++ b/apps/backend/internal/events/types.go @@ -206,6 +206,11 @@ const ( GitHubWatchEvent = "github.watch.event" // Watch created/deleted ) +// Event types for Jira integration +const ( + JiraNewIssue = "jira.new_issue" // New issue found matching a Jira 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/jira/handlers.go b/apps/backend/internal/jira/handlers.go index da43f1f719..9efee14591 100644 --- a/apps/backend/internal/jira/handlers.go +++ b/apps/backend/internal/jira/handlers.go @@ -37,6 +37,12 @@ func (c *Controller) RegisterHTTPRoutes(router *gin.Engine) { api.GET("/tickets", c.httpSearchTickets) api.GET("/tickets/:key", c.httpGetTicket) api.POST("/tickets/:key/transitions", c.httpDoTransition) + + 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 --- @@ -171,6 +177,133 @@ func (c *Controller) httpDoTransition(ctx *gin.Context) { ctx.JSON(http.StatusOK, gin.H{"transitioned": true}) } +// --- Issue watch HTTP handlers --- + +func (c *Controller) httpListIssueWatches(ctx *gin.Context) { + workspaceID := ctx.Query("workspace_id") + if workspaceID == "" { + ctx.JSON(http.StatusBadRequest, gin.H{"error": "workspace_id required"}) + return + } + watches, err := c.service.ListIssueWatches(ctx.Request.Context(), workspaceID) + if err != nil { + ctx.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()}) + return + } + ctx.JSON(http.StatusOK, gin.H{"watches": watches}) +} + +func (c *Controller) httpCreateIssueWatch(ctx *gin.Context) { + var req CreateIssueWatchRequest + if err := ctx.ShouldBindJSON(&req); err != nil { + ctx.JSON(http.StatusBadRequest, gin.H{"error": "invalid payload"}) + return + } + 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 of the watch. Useful from +// the UI to verify a JQL change without waiting for the next 5-minute tick. +// Returns the count of newly-discovered tickets so the user gets feedback even +// when the matching tickets fan out asynchronously through the orchestrator. +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) { + // 404 not 403 — don't reveal whether the ID exists. + ctx.JSON(http.StatusNotFound, gin.H{"error": ErrIssueWatchNotFound.Error()}) + return + } + tickets, err := c.service.CheckIssueWatch(ctx.Request.Context(), w) + if err != nil { + c.writeClientError(ctx, err) + return + } + for _, t := range tickets { + c.service.publishNewJiraIssueEvent(ctx.Request.Context(), w, t) + } + ctx.JSON(http.StatusOK, gin.H{"newIssues": len(tickets)}) +} + +// assertWatchInWorkspace guards mutation/trigger endpoints against IDOR: the +// caller must supply `?workspace_id=...` matching the watch's workspace. The +// list/create endpoints already require this query; mirroring it here closes +// the gap where a known watch UUID from another workspace could be mutated. +// Writes the response and returns false on mismatch — caller bails out. +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) { + // Use 404 so a probing client can't tell whether a watch ID exists in + // another workspace. + ctx.JSON(http.StatusNotFound, gin.H{"error": ErrIssueWatchNotFound.Error()}) + return false + } + return true +} + +// workspaceMatches returns true when the request's `workspace_id` query param +// matches the resource's stored workspace. Empty query is rejected so the +// caller 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) +} + // errCodeJiraNotConfigured is the wire-level code surfaced to the UI when the // workspace has no saved Jira credentials. The same string is used by both the // HTTP and WebSocket layers so the frontend can branch on a single value. diff --git a/apps/backend/internal/jira/models.go b/apps/backend/internal/jira/models.go index c01ae370c7..798fb3c79f 100644 --- a/apps/backend/internal/jira/models.go +++ b/apps/backend/internal/jira/models.go @@ -118,3 +118,83 @@ type SearchResult struct { func SecretKeyForWorkspace(workspaceID string) string { return "jira:" + workspaceID + ":token" } + +// DefaultIssueWatchPollInterval is the polling cadence assigned to a watcher +// when the caller does not specify one. Five minutes balances freshness against +// Atlassian rate limits when many workspaces have watches configured. +const DefaultIssueWatchPollInterval = 300 + +// IssueWatch configures periodic JQL polling: a workspace-scoped watcher runs +// the JQL on a schedule and emits a NewJiraIssueEvent for each matching ticket +// the orchestrator hasn't already turned into a Kandev task. +// +// Unlike the GitHub equivalent, JIRA issues have no repository affinity — the +// target workflow step's defaults determine where the resulting task runs, so +// there's no `repos` column. +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"` + JQL string `json:"jql" db:"jql"` + 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, ticket) tuple. The +// UNIQUE constraint on (issue_watch_id, issue_key) prevents two concurrent +// pollers from racing to create duplicate tasks for the same ticket. +type IssueWatchTask struct { + ID string `json:"id" db:"id"` + IssueWatchID string `json:"issueWatchId" db:"issue_watch_id"` + IssueKey string `json:"issueKey" db:"issue_key"` + IssueURL string `json:"issueUrl" db:"issue_url"` + TaskID string `json:"taskId" db:"task_id"` + CreatedAt time.Time `json:"createdAt" db:"created_at"` +} + +// NewJiraIssueEvent is published on the bus whenever the poller observes a +// ticket matching a watch that has no existing dedup row. The orchestrator +// consumes this to create (and optionally auto-start) a Kandev task. +type NewJiraIssueEvent 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 *JiraTicket `json:"issue"` +} + +// CreateIssueWatchRequest is the payload for POST /api/v1/jira/watches/issue. +type CreateIssueWatchRequest struct { + WorkspaceID string `json:"workspaceId"` + WorkflowID string `json:"workflowId"` + WorkflowStepID string `json:"workflowStepId"` + JQL string `json:"jql"` + 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/jira/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"` + JQL *string `json:"jql,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/jira/poller.go b/apps/backend/internal/jira/poller.go index dfa0706830..822a550793 100644 --- a/apps/backend/internal/jira/poller.go +++ b/apps/backend/internal/jira/poller.go @@ -17,14 +17,26 @@ import ( // don't hammer Atlassian when many workspaces are configured. const defaultAuthPollInterval = 90 * time.Second -// Poller probes the stored Jira credentials of every configured workspace on a -// fixed cadence and persists the result on the JiraConfig row. The frontend -// reads those fields via getJiraConfig to render the connected/disconnected -// indicator without doing its own probing. +// defaultIssuePollTickInterval is how often the issue-watch loop wakes up to +// look at every enabled watcher. The actual JQL 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. 60 seconds +// matches the smallest interval the UI accepts; the gating check then rate- +// limits each individual watcher. +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. +// - issue watches: runs each enabled watcher's JQL and emits NewJiraIssueEvent +// for every matching ticket the orchestrator hasn't yet seen. +// +// Both loops are cancelled together via Stop. type Poller struct { - service *Service - logger *logger.Logger - interval time.Duration + service *Service + logger *logger.Logger + authInterval time.Duration + issueInterval time.Duration + issueTickHook func() // tests use this to observe each issue-watch tick. // mu guards started/cancel/wg against concurrent Start/Stop calls. mu sync.Mutex @@ -33,12 +45,26 @@ type Poller struct { started bool } -// NewPoller returns a poller that uses the default 90s cadence. +// NewPoller returns a poller using the default cadences. func NewPoller(svc *Service, log *logger.Logger) *Poller { - return &Poller{service: svc, logger: log, interval: defaultAuthPollInterval} + return &Poller{ + service: svc, + logger: log, + authInterval: defaultAuthPollInterval, + issueInterval: defaultIssuePollTickInterval, + } } -// Start launches the background loop. Calling Start more than once without +// 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. func (p *Poller) Start(ctx context.Context) { p.mu.Lock() @@ -48,9 +74,10 @@ func (p *Poller) Start(ctx context.Context) { } p.started = true ctx, p.cancel = context.WithCancel(ctx) - p.wg.Add(1) + p.wg.Add(2) go p.loop(ctx) - p.logger.Info("Jira auth poller started") + go p.issueWatchLoop(ctx) + p.logger.Info("Jira poller started") } // Stop cancels the loop and waits for it to drain. @@ -69,7 +96,7 @@ func (p *Poller) Stop() { p.mu.Lock() p.started = false p.mu.Unlock() - p.logger.Info("Jira auth poller stopped") + p.logger.Info("Jira poller stopped") } func (p *Poller) loop(ctx context.Context) { @@ -77,7 +104,7 @@ func (p *Poller) loop(ctx context.Context) { // Run an initial probe immediately so the UI gets a status without waiting // the full interval after backend startup. p.probeAll(ctx) - ticker := time.NewTicker(p.interval) + ticker := time.NewTicker(p.authInterval) defer ticker.Stop() for { select { @@ -89,6 +116,81 @@ func (p *Poller) loop(ctx context.Context) { } } +// issueWatchLoop drives the periodic JQL-poll → publish-event flow. Unlike +// the auth loop, this one waits a full interval before its first tick so the +// backend doesn't hammer JIRA the moment it starts. +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("jira 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 + } + newTickets, err := p.service.CheckIssueWatch(ctx, w) + if err != nil { + p.logger.Debug("jira poller: check issue watch failed", + zap.String("watch_id", w.ID), zap.Error(err)) + continue + } + for _, t := range newTickets { + p.logger.Info("new jira issue found for watch", + zap.String("watch_id", w.ID), + zap.String("issue_key", t.Key), + zap.String("summary", t.Summary)) + p.service.publishNewJiraIssueEvent(ctx, w, t) + } + } +} + +// isIssueWatchDue reports whether enough time has passed since the watch was +// last polled to re-run its JQL on this tick. A watch with no LastPolledAt +// (never polled, or DB row freshly created) is always due. PollIntervalSeconds +// <= 0 falls back to the default — the same normalisation the store applies on +// write — so a corrupt row never blocks polling forever. +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() + } +} + func (p *Poller) probeAll(ctx context.Context) { ids, err := p.service.Store().ListConfiguredWorkspaces(ctx) if err != nil { diff --git a/apps/backend/internal/jira/poller_issue_watch_test.go b/apps/backend/internal/jira/poller_issue_watch_test.go new file mode 100644 index 0000000000..5489b68996 --- /dev/null +++ b/apps/backend/internal/jira/poller_issue_watch_test.go @@ -0,0 +1,214 @@ +package jira + +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 JiraNewIssue 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 []*NewJiraIssueEvent + done chan struct{} // closed once `expected` events have arrived. + want int +} + +func newRecordingSubscriber(eb bus.EventBus, want int) (*recordingSubscriber, error) { + r := &recordingSubscriber{done: make(chan struct{}), want: want} + _, err := eb.Subscribe(events.JiraNewIssue, func(_ context.Context, e *bus.Event) error { + r.mu.Lock() + defer r.mu.Unlock() + evt, ok := e.Data.(*NewJiraIssueEvent) + 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() []*NewJiraIssueEvent { + r.mu.Lock() + defer r.mu.Unlock() + out := make([]*NewJiraIssueEvent, len(r.payload)) + copy(out, r.payload) + return out +} + +func TestPoller_CheckIssueWatches_PublishesNewTicketsOnly(t *testing.T) { + f := newPollerFixture(t) + ctx := context.Background() + f.saveConfig(t, "ws-1", "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.searchFn = func(_ string) (*SearchResult, error) { + return &SearchResult{ + Tickets: []JiraTicket{ + {Key: "PROJ-1", Summary: "first", URL: "https://x/browse/PROJ-1"}, + {Key: "PROJ-2", Summary: "second", URL: "https://x/browse/PROJ-2"}, + }, + IsLast: true, + }, nil + } + + // First tick: both tickets are new, so two events publish. + f.poller.checkIssueWatches(ctx) + + <-sub.done // wait for both events to arrive on the in-memory bus + first := sub.snapshot() + if len(first) != 2 { + t.Fatalf("expected 2 events on first tick, got %d", len(first)) + } + keys := []string{first[0].Issue.Key, first[1].Issue.Key} + if !contains(keys, "PROJ-1") || !contains(keys, "PROJ-2") { + t.Errorf("expected events for PROJ-1 and PROJ-2, got %v", keys) + } + if first[0].WorkspaceID != "ws-1" || first[0].WorkflowID != "wf-1" { + t.Errorf("event missing watch context: %+v", first[0]) + } + + // Reserve both keys to simulate the orchestrator having created tasks. + for _, key := range []string{"PROJ-1", "PROJ-2"} { + if _, err := f.store.ReserveIssueWatchTask(ctx, w.ID, key, "https://x/browse/"+key); err != nil { + t.Fatalf("reserve %s: %v", key, 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, "ws-1", "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.searchFn = func(_ string) (*SearchResult, error) { + calls++ + return &SearchResult{Tickets: []JiraTicket{{Key: "X-1"}}, IsLast: true}, nil + } + + f.poller.checkIssueWatches(ctx) + + if calls != 0 { + t.Errorf("expected disabled watch to be skipped (no JIRA 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, // default 300s > 30s elapsed + }, + } + 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, "ws-1", "tok") + + eb := bus.NewMemoryEventBus(logger.Default()) + defer eb.Close() + f.svc.SetEventBus(eb) + + // Watch was just polled — should be skipped on this tick. + 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.searchFn = func(_ string) (*SearchResult, error) { + calls++ + return &SearchResult{Tickets: []JiraTicket{{Key: "X-1"}}, IsLast: true}, nil + } + + f.poller.checkIssueWatches(ctx) + + if calls != 0 { + t.Errorf("expected gating to skip the JIRA search, got %d call(s)", calls) + } +} diff --git a/apps/backend/internal/jira/provider.go b/apps/backend/internal/jira/provider.go index e0ebcfa136..7daf55faac 100644 --- a/apps/backend/internal/jira/provider.go +++ b/apps/backend/internal/jira/provider.go @@ -4,17 +4,23 @@ import ( "github.com/jmoiron/sqlx" "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/internal/events/bus" ) -// Provide builds the Jira 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. -func Provide(writer, reader *sqlx.DB, secrets SecretStore, log *logger.Logger) (*Service, func() error, error) { +// Provide builds the Jira 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. +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 } svc := NewService(store, secrets, DefaultClientFactory, log) + if eventBus != nil { + svc.SetEventBus(eventBus) + } cleanup := func() error { return nil } return svc, cleanup, nil } diff --git a/apps/backend/internal/jira/service.go b/apps/backend/internal/jira/service.go index 0cad41fe18..b33769f0b4 100644 --- a/apps/backend/internal/jira/service.go +++ b/apps/backend/internal/jira/service.go @@ -11,6 +11,8 @@ import ( "go.uber.org/zap" "github.com/kandev/kandev/internal/common/logger" + "github.com/kandev/kandev/internal/events" + "github.com/kandev/kandev/internal/events/bus" ) // SecretStore is the subset of the secrets store the service needs. Kept small @@ -32,6 +34,7 @@ type Service struct { clientFn ClientFactory cache map[string]Client // workspaceID → client, cleared on config change. probeHook func(workspaceID string) + eventBus bus.EventBus } // ClientFactory builds a Client for the given config + secret. Overridable so @@ -404,3 +407,238 @@ func normalizeSiteURL(raw string) string { } return s } + +// SetEventBus wires the bus used to publish NewJiraIssueEvent. Optional: if +// unset the poller still runs but observed tickets do not become Kandev tasks +// — useful in tests that exercise the polling loop in isolation. +func (s *Service) SetEventBus(eb bus.EventBus) { + s.mu.Lock() + s.eventBus = eb + s.mu.Unlock() +} + +// --- Issue watch CRUD (thin pass-throughs to the store) --- + +// 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("jira: issue watch not found") + +// 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, + JQL: strings.TrimSpace(req.JQL), + 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) +} + +// 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) + // Empty-string PATCH writes (`{"workflowId": ""}` etc.) bypass the nil + // guard in applyIssueWatchPatch — Go's JSON decoder returns a non-nil + // pointer to "". Guard the post-patch state so a PATCH can't strip a row + // of the fields the orchestrator needs to create tasks. + if w.JQL == "" { + return nil, fmt.Errorf("%w: jql cannot be empty", 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: deleting +// a missing ID is a silent success. +func (s *Service) DeleteIssueWatch(ctx context.Context, id string) error { + return s.store.DeleteIssueWatch(ctx, id) +} + +// CheckIssueWatch runs the watch's JQL once and returns the tickets that +// haven't been turned into tasks yet. last_polled_at is stamped regardless of +// whether the search succeeded — a failing search still counts as "we tried" +// so the UI can show liveness. +func (s *Service) CheckIssueWatch(ctx context.Context, w *IssueWatch) ([]*JiraTicket, error) { + defer s.stampLastPolled(w.ID) + client, err := s.clientFor(ctx, w.WorkspaceID) + if err != nil { + return nil, err + } + // Single page is enough per tick: the next poll picks up anything that + // overflowed maxResults. Pagination would let one chatty workspace starve + // the others without any real freshness benefit. + res, err := client.SearchTickets(ctx, w.JQL, "", issueWatchSearchPageSize) + if err != nil { + return nil, err + } + // Bulk-fetch the dedup set once per call instead of one query per ticket — + // a broad JQL can return up to issueWatchSearchPageSize tickets, so a + // per-ticket round-trip would multiply the per-tick cost. + seen, err := s.store.ListSeenIssueKeys(ctx, w.ID) + if err != nil { + s.log.Warn("jira: dedup set fetch failed", + zap.String("watch_id", w.ID), zap.Error(err)) + seen = nil // fall through: a missing dedup set is safer than dropping all tickets + } + out := make([]*JiraTicket, 0, len(res.Tickets)) + for i := range res.Tickets { + t := res.Tickets[i] + if _, ok := seen[t.Key]; ok { + continue + } + out = append(out, &t) + } + return out, nil +} + +// stampLastPolled writes the current timestamp onto the watch row using a +// fresh background context with a short write deadline. The caller's ctx may +// already be cancelled (typical at poller shutdown), which would cause the +// DB write to fail silently and the watch would look "never polled" on the +// next backend restart. Mirrors the pattern in RecordAuthHealth. +func (s *Service) stampLastPolled(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("jira: update last_polled_at failed", + zap.String("watch_id", watchID), zap.Error(err)) + } +} + +// publishNewJiraIssueEvent emits the orchestrator-facing event for one freshly +// observed ticket. No-op when the event bus is not wired (tests, early boot). +func (s *Service) publishNewJiraIssueEvent(ctx context.Context, w *IssueWatch, ticket *JiraTicket) { + s.mu.Lock() + eb := s.eventBus + s.mu.Unlock() + if eb == nil { + return + } + evt := bus.NewEvent(events.JiraNewIssue, "jira", &NewJiraIssueEvent{ + IssueWatchID: w.ID, + WorkspaceID: w.WorkspaceID, + WorkflowID: w.WorkflowID, + WorkflowStepID: w.WorkflowStepID, + AgentProfileID: w.AgentProfileID, + ExecutorProfileID: w.ExecutorProfileID, + Prompt: w.Prompt, + Issue: ticket, + }) + if err := eb.Publish(ctx, events.JiraNewIssue, evt); err != nil { + s.log.Debug("jira: publish new issue event failed", + zap.String("watch_id", w.ID), zap.String("issue_key", ticket.Key), zap.Error(err)) + } +} + +// issueWatchSearchPageSize caps how many tickets a single CheckIssueWatch call +// pulls from JIRA. 50 is well under the API's 100-result limit and keeps the +// per-tick cost bounded even for very broad JQL queries. +const issueWatchSearchPageSize = 50 + +// MinIssueWatchPollInterval / MaxIssueWatchPollInterval bound the per-watch +// JQL re-run cadence. The lower limit protects Atlassian from rapid-fire +// polling; the upper limit keeps "stuck" rows from polling so rarely they +// look broken to the user. +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 strings.TrimSpace(req.JQL) == "" { + return fmt.Errorf("%w: jql required", ErrInvalidConfig) + } + // 0 / unset is allowed — the store coerces to DefaultIssueWatchPollInterval. + // Any positive value must fall inside the documented bounds. + 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 +} + +func applyIssueWatchPatch(w *IssueWatch, req *UpdateIssueWatchRequest) { + if req.WorkflowID != nil { + w.WorkflowID = *req.WorkflowID + } + if req.WorkflowStepID != nil { + w.WorkflowStepID = *req.WorkflowStepID + } + if req.JQL != nil { + w.JQL = strings.TrimSpace(*req.JQL) + } + 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/jira/service_issue_watch_test.go b/apps/backend/internal/jira/service_issue_watch_test.go new file mode 100644 index 0000000000..0babe28e66 --- /dev/null +++ b/apps/backend/internal/jira/service_issue_watch_test.go @@ -0,0 +1,220 @@ +package jira + +import ( + "context" + "errors" + "testing" +) + +// withSearchResults returns a fakeClient that always returns the given tickets +// for SearchTickets, ignoring the JQL. +func (c *fakeClient) withSearchResults(tickets []JiraTicket) *fakeClient { + c.searchFn = func(_ string) (*SearchResult, error) { + return &SearchResult{Tickets: tickets, IsLast: true}, nil + } + return c +} + +func TestService_CreateIssueWatch_DefaultsAndValidation(t *testing.T) { + f := newSvcFixture(t) + ctx := context.Background() + + // Missing JQL 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 missing JQL, got %v", err) + } + + // Happy path assigns ID + defaults Enabled=true. + w, err := f.svc.CreateIssueWatch(ctx, &CreateIssueWatchRequest{ + WorkspaceID: "ws-1", + WorkflowID: "wf", + WorkflowStepID: "step", + JQL: "project = PROJ", + }) + 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") + } +} + +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", + JQL: "project = PROJ", + 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.JQL != created.JQL || updated.WorkspaceID != created.WorkspaceID { + t.Errorf("unexpected mutation of unset fields: %+v", updated) + } + + // Patching JQL to empty is rejected to keep watch rows valid. + empty := " " + if _, err := f.svc.UpdateIssueWatch(ctx, created.ID, &UpdateIssueWatchRequest{JQL: &empty}); !errors.Is(err, ErrInvalidConfig) { + t.Errorf("expected ErrInvalidConfig for empty JQL, 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", + JQL: "project = PROJ", + 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", + JQL: "project = PROJ", + }); err != nil { + t.Errorf("expected zero pollIntervalSeconds to be accepted (default fill), 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", + JQL: "project = PROJ", + }) + if err != nil { + t.Fatalf("create: %v", err) + } + empty := "" + // `{"workflowId": ""}` would land here — the nil guard in applyIssueWatchPatch + // doesn't catch explicit empty strings, and a watch with empty WorkflowID + // would silently drop every event the orchestrator tries to handle. + 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() + + // Configure JIRA so clientFor succeeds. + if _, err := f.svc.SetConfig(ctx, &SetConfigRequest{ + WorkspaceID: "ws-1", SiteURL: "https://a.net", Email: "e", + AuthMethod: AuthMethodAPIToken, Secret: "tok", + }); err != nil { + t.Fatalf("set config: %v", err) + } + + // Search returns three tickets; one of them is already in the dedup table. + f.client.withSearchResults([]JiraTicket{ + {Key: "PROJ-1", Summary: "one", URL: "https://a.net/browse/PROJ-1"}, + {Key: "PROJ-2", Summary: "two", URL: "https://a.net/browse/PROJ-2"}, + {Key: "PROJ-3", Summary: "three", URL: "https://a.net/browse/PROJ-3"}, + }) + + w, err := f.svc.CreateIssueWatch(ctx, &CreateIssueWatchRequest{ + WorkspaceID: "ws-1", WorkflowID: "wf", WorkflowStepID: "step", + JQL: "project = PROJ", + }) + if err != nil { + t.Fatalf("create watch: %v", err) + } + + // Pre-seed PROJ-2 as already turned into a task. + if _, err := f.store.ReserveIssueWatchTask(ctx, w.ID, "PROJ-2", "https://a.net/browse/PROJ-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 tickets, got %d", len(got)) + } + for _, tk := range got { + if tk.Key == "PROJ-2" { + t.Error("PROJ-2 should have been filtered as already seen") + } + } + + // last_polled_at must have been stamped. + 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{ + WorkspaceID: "ws-1", SiteURL: "https://a.net", Email: "e", + AuthMethod: AuthMethodAPIToken, Secret: "tok", + }); err != nil { + t.Fatalf("set config: %v", err) + } + f.client.searchFn = func(_ string) (*SearchResult, error) { + return nil, errors.New("upstream 500") + } + w, _ := f.svc.CreateIssueWatch(ctx, &CreateIssueWatchRequest{ + WorkspaceID: "ws-1", WorkflowID: "wf", WorkflowStepID: "step", + JQL: "project = PROJ", + }) + 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/jira/service_test.go b/apps/backend/internal/jira/service_test.go index 269b61a941..1f076e4b7e 100644 --- a/apps/backend/internal/jira/service_test.go +++ b/apps/backend/internal/jira/service_test.go @@ -58,6 +58,7 @@ type fakeClient struct { getTicketFn func(key string) (*JiraTicket, error) transitionFn func(key, id string) error listProjects func() ([]JiraProject, error) + searchFn func(jql string) (*SearchResult, error) transitionLog []string // "key:id" } @@ -89,7 +90,10 @@ func (c *fakeClient) ListProjects(_ context.Context) ([]JiraProject, error) { } return nil, nil } -func (c *fakeClient) SearchTickets(_ context.Context, _, _ string, _ int) (*SearchResult, error) { +func (c *fakeClient) SearchTickets(_ context.Context, jql, _ string, _ int) (*SearchResult, error) { + if c.searchFn != nil { + return c.searchFn(jql) + } return &SearchResult{}, nil } diff --git a/apps/backend/internal/jira/store.go b/apps/backend/internal/jira/store.go index 304080a230..ed4ba555ae 100644 --- a/apps/backend/internal/jira/store.go +++ b/apps/backend/internal/jira/store.go @@ -7,6 +7,7 @@ import ( "fmt" "time" + "github.com/google/uuid" "github.com/jmoiron/sqlx" ) @@ -39,6 +40,35 @@ const createTablesSQL = ` created_at DATETIME NOT NULL, updated_at DATETIME NOT NULL ); + + CREATE TABLE IF NOT EXISTS jira_issue_watches ( + id TEXT PRIMARY KEY, + workspace_id TEXT NOT NULL, + workflow_id TEXT NOT NULL, + workflow_step_id TEXT NOT NULL, + jql TEXT NOT NULL, + 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_jira_issue_watches_workspace + ON jira_issue_watches(workspace_id); + + CREATE TABLE IF NOT EXISTS jira_issue_watch_tasks ( + id TEXT PRIMARY KEY, + issue_watch_id TEXT NOT NULL, + issue_key TEXT NOT NULL, + issue_url TEXT NOT NULL, + task_id TEXT NOT NULL DEFAULT '', + created_at DATETIME NOT NULL, + UNIQUE(issue_watch_id, issue_key), + FOREIGN KEY(issue_watch_id) REFERENCES jira_issue_watches(id) ON DELETE CASCADE + ); ` // addedColumns lists columns introduced after the initial schema. SQLite has no @@ -175,3 +205,185 @@ func (s *Store) UpdateAuthHealth(ctx context.Context, workspaceID string, ok boo checkedAt, ok, errMsg, workspaceID) return err } + +// --- Issue watch operations --- + +const issueWatchColumns = `id, workspace_id, workflow_id, workflow_step_id, jql, + 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 + } + _, err := s.db.ExecContext(ctx, ` + INSERT INTO jira_issue_watches (`+issueWatchColumns+`) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + w.ID, w.WorkspaceID, w.WorkflowID, w.WorkflowStepID, w.JQL, + 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 w IssueWatch + err := s.ro.GetContext(ctx, &w, + `SELECT `+issueWatchColumns+` FROM jira_issue_watches WHERE id = ?`, id) + if errors.Is(err, sql.ErrNoRows) { + return nil, nil + } + if err != nil { + return nil, err + } + return &w, nil +} + +// ListIssueWatches returns all watches configured for a workspace, in +// insertion order. The UI uses this to render the watcher table. +func (s *Store) ListIssueWatches(ctx context.Context, workspaceID string) ([]*IssueWatch, error) { + var watches []*IssueWatch + err := s.ro.SelectContext(ctx, &watches, + `SELECT `+issueWatchColumns+` FROM jira_issue_watches + WHERE workspace_id = ? ORDER BY created_at`, workspaceID) + if err != nil { + return nil, err + } + return watches, nil +} + +// 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 watches []*IssueWatch + err := s.ro.SelectContext(ctx, &watches, + `SELECT `+issueWatchColumns+` FROM jira_issue_watches + WHERE enabled = 1 ORDER BY created_at`) + if err != nil { + return nil, err + } + return watches, 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 + } + _, err := s.db.ExecContext(ctx, ` + UPDATE jira_issue_watches SET workflow_id = ?, workflow_step_id = ?, jql = ?, + agent_profile_id = ?, executor_profile_id = ?, prompt = ?, + enabled = ?, poll_interval_seconds = ?, last_polled_at = ?, updated_at = ? + WHERE id = ?`, + w.WorkflowID, w.WorkflowStepID, w.JQL, + 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. The poller calls this after every check so the UI can +// show "polled X seconds ago". +func (s *Store) UpdateIssueWatchLastPolled(ctx context.Context, id string, t time.Time) error { + _, err := s.db.ExecContext(ctx, + `UPDATE jira_issue_watches SET last_polled_at = ?, updated_at = ? WHERE id = ?`, + t, time.Now().UTC(), id) + return err +} + +// DeleteIssueWatch removes a watch and (via FK ON DELETE CASCADE) its dedup +// rows in a single transaction. The explicit DELETE on the child table 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 jira_issue_watch_tasks WHERE issue_watch_id = ?`, id); err != nil { + return err + } + if _, err := tx.ExecContext(ctx, `DELETE FROM jira_issue_watches WHERE id = ?`, id); err != nil { + return err + } + return tx.Commit() +} + +// ReserveIssueWatchTask atomically claims a slot for a (watch, ticket) pair via +// INSERT OR IGNORE. Returns true when this caller won the race and should +// proceed to create the task. False either means another handler already +// reserved the same ticket or the row already exists from a prior run. +func (s *Store) ReserveIssueWatchTask(ctx context.Context, watchID, issueKey, issueURL string) (bool, error) { + res, err := s.db.ExecContext(ctx, ` + INSERT OR IGNORE INTO jira_issue_watch_tasks (id, issue_watch_id, issue_key, issue_url, task_id, created_at) + VALUES (?, ?, ?, ?, ?, ?)`, + uuid.New().String(), watchID, issueKey, 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. Returns an error if no reservation matches — callers should treat +// that as a programming bug since they only call this after a successful +// reservation. +func (s *Store) AssignIssueWatchTaskID(ctx context.Context, watchID, issueKey, taskID string) error { + res, err := s.db.ExecContext(ctx, ` + UPDATE jira_issue_watch_tasks SET task_id = ? + WHERE issue_watch_id = ? AND issue_key = ?`, + taskID, watchID, issueKey) + 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, issueKey) + } + 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, issueKey string) error { + _, err := s.db.ExecContext(ctx, + `DELETE FROM jira_issue_watch_tasks WHERE issue_watch_id = ? AND issue_key = ?`, + watchID, issueKey) + return err +} + +// ListSeenIssueKeys returns the set of ticket keys already reserved against a +// watch. Returning a set in one query is cheaper than per-ticket existence +// checks — a single JQL search can return up to 50 tickets per tick, so the +// per-call savings scale with the workspace's watch count. +func (s *Store) ListSeenIssueKeys(ctx context.Context, watchID string) (map[string]struct{}, error) { + var keys []string + err := s.ro.SelectContext(ctx, &keys, + `SELECT issue_key FROM jira_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/jira/store_issue_watch_test.go b/apps/backend/internal/jira/store_issue_watch_test.go new file mode 100644 index 0000000000..0359f73e7a --- /dev/null +++ b/apps/backend/internal/jira/store_issue_watch_test.go @@ -0,0 +1,292 @@ +package jira + +import ( + "context" + "testing" + "time" +) + +func newTestIssueWatch(workspaceID string) *IssueWatch { + return &IssueWatch{ + WorkspaceID: workspaceID, + WorkflowID: "wf-1", + WorkflowStepID: "step-1", + JQL: `project = PROJ AND status = "Open"`, + Prompt: "Investigate {{issue.key}}", + 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.JQL != w.JQL || got.WorkspaceID != w.WorkspaceID || got.Prompt != w.Prompt { + t.Errorf("round-trip mismatch: %+v vs %+v", got, w) + } +} + +func TestStore_IssueWatch_ListByWorkspace(t *testing.T) { + store := newTestStore(t) + ctx := context.Background() + + w1 := newTestIssueWatch("ws-1") + w1.JQL = "project = A" + w2 := newTestIssueWatch("ws-1") + w2.JQL = "project = 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)) + } + // ws-2's watch must not appear. + 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.JQL = "project = NEW" + 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.JQL != "project = NEW" || 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, "PROJ-1", "https://example.atlassian.net/browse/PROJ-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) + } + // Dedup row must be gone too — re-creating a watch with the same ID and + // reserving the same key should succeed without a UNIQUE collision. + 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, "PROJ-1", "https://example.atlassian.net/browse/PROJ-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 reservation wins. + first, err := store.ReserveIssueWatchTask(ctx, w.ID, "PROJ-7", "https://example.atlassian.net/browse/PROJ-7") + if err != nil { + t.Fatalf("first reserve: %v", err) + } + if !first { + t.Error("expected first reservation to win") + } + + // Second reservation for the same (watch, key) loses (dedup). + second, err := store.ReserveIssueWatchTask(ctx, w.ID, "PROJ-7", "https://example.atlassian.net/browse/PROJ-7") + if err != nil { + t.Fatalf("second reserve: %v", err) + } + if second { + t.Error("expected second reservation to lose due to UNIQUE constraint") + } + + // ListSeenIssueKeys reflects the reservation regardless of who won. + seen, err := store.ListSeenIssueKeys(ctx, w.ID) + if err != nil { + t.Fatalf("list seen: %v", err) + } + if _, ok := seen["PROJ-7"]; !ok { + t.Error("expected PROJ-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, "PROJ-1", "https://example.atlassian.net/browse/PROJ-1"); err != nil { + t.Fatalf("reserve: %v", err) + } + if err := store.AssignIssueWatchTaskID(ctx, w.ID, "PROJ-1", "task-abc"); err != nil { + t.Fatalf("assign: %v", err) + } + + // Assigning to a missing reservation surfaces a clear error. + err := store.AssignIssueWatchTaskID(ctx, w.ID, "PROJ-NOPE", "task-zzz") + if err == nil { + t.Error("expected error for missing reservation, got nil") + } +} + +func TestStore_IssueWatchTask_ListSeenIssueKeys(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) + } + for _, k := range []string{"PROJ-1", "PROJ-2", "PROJ-3"} { + if _, err := store.ReserveIssueWatchTask(ctx, w.ID, k, "https://x/"+k); err != nil { + t.Fatalf("reserve %s: %v", k, err) + } + } + + seen, err := store.ListSeenIssueKeys(ctx, w.ID) + if err != nil { + t.Fatalf("list seen: %v", err) + } + for _, k := range []string{"PROJ-1", "PROJ-2", "PROJ-3"} { + if _, ok := seen[k]; !ok { + t.Errorf("expected %s in seen set, got %v", k, seen) + } + } + if _, ok := seen["PROJ-99"]; ok { + t.Error("expected PROJ-99 not in seen set") + } +} + +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, "PROJ-1", "https://example.atlassian.net/browse/PROJ-1"); err != nil { + t.Fatalf("reserve: %v", err) + } + if err := store.ReleaseIssueWatchTask(ctx, w.ID, "PROJ-1"); err != nil { + t.Fatalf("release: %v", err) + } + // After release, the slot can be reserved again. + again, err := store.ReserveIssueWatchTask(ctx, w.ID, "PROJ-1", "https://example.atlassian.net/browse/PROJ-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_jira.go b/apps/backend/internal/orchestrator/event_handlers_jira.go new file mode 100644 index 0000000000..2edc5591b5 --- /dev/null +++ b/apps/backend/internal/orchestrator/event_handlers_jira.go @@ -0,0 +1,185 @@ +package orchestrator + +import ( + "context" + "fmt" + "strings" + + "go.uber.org/zap" + + "github.com/kandev/kandev/internal/events" + "github.com/kandev/kandev/internal/events/bus" + "github.com/kandev/kandev/internal/jira" +) + +// JiraService is the subset of jira.Service the orchestrator needs to +// deduplicate ticket→task mappings. Mirrors the GitHub equivalent so the +// orchestrator stays decoupled from the concrete jira package types. +type JiraService interface { + ReserveIssueWatchTask(ctx context.Context, watchID, issueKey, issueURL string) (bool, error) + AssignIssueWatchTaskID(ctx context.Context, watchID, issueKey, taskID string) error + ReleaseIssueWatchTask(ctx context.Context, watchID, issueKey string) error +} + +// SetJiraService wires the JIRA dedup helpers into the orchestrator so +// jira-watch handlers can claim ticket slots before creating tasks. +func (s *Service) SetJiraService(j JiraService) { + s.jiraService = j +} + +// subscribeJiraEvents wires the JIRA event handlers onto the bus. Called from +// the existing subscribeGitHubEvents-style boot path so all integration +// subscriptions stay grouped together. +func (s *Service) subscribeJiraEvents() { + if s.eventBus == nil { + return + } + if _, err := s.eventBus.Subscribe(events.JiraNewIssue, s.handleNewJiraIssue); err != nil { + s.logger.Error("failed to subscribe to jira.new_issue events", zap.Error(err)) + } +} + +// handleNewJiraIssue creates a Kandev task for a freshly-observed JIRA ticket. +// Reuses the same IssueTaskCreator the GitHub flow uses so task creation logic +// (workflow placement, on-enter actions, etc.) stays in one place. +func (s *Service) handleNewJiraIssue(_ context.Context, event *bus.Event) error { + evt, ok := event.Data.(*jira.NewJiraIssueEvent) + if !ok { + return nil + } + if evt.Issue == nil { + return nil + } + s.logger.Info("new jira issue detected from watch", + zap.String("issue_watch_id", evt.IssueWatchID), + zap.String("issue_key", evt.Issue.Key)) + + if s.issueTaskCreator == nil { + s.logger.Warn("issue task creator not configured, skipping jira 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.createJiraIssueTask(context.Background(), evt) + return nil +} + +func (s *Service) createJiraIssueTask(ctx context.Context, evt *jira.NewJiraIssueEvent) { + ticket := evt.Issue + if !s.reserveJiraIssue(ctx, evt) { + return + } + + req := &IssueTaskRequest{ + WorkspaceID: evt.WorkspaceID, + WorkflowID: evt.WorkflowID, + WorkflowStepID: evt.WorkflowStepID, + Title: fmt.Sprintf("[%s] %s", ticket.Key, ticket.Summary), + Description: interpolateJiraPrompt(evt.Prompt, ticket), + Metadata: map[string]interface{}{ + "jira_issue_watch_id": evt.IssueWatchID, + "jira_issue_key": ticket.Key, + "jira_issue_url": ticket.URL, + "jira_status": ticket.StatusName, + "jira_assignee": ticket.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 jira issue task", + zap.String("issue_watch_id", evt.IssueWatchID), + zap.String("issue_key", ticket.Key), + zap.Error(err)) + s.releaseJiraIssue(ctx, evt) + return + } + + s.attachJiraIssueTaskID(ctx, evt, task.ID) + s.logger.Info("created jira issue task", + zap.String("task_id", task.ID), + zap.String("issue_key", ticket.Key)) + + 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 jira issue task", + zap.String("task_id", task.ID), zap.Error(err)) + return + } + s.logger.Info("auto-started jira issue task", + zap.String("task_id", task.ID), + zap.String("issue_key", ticket.Key)) +} + +// reserveJiraIssue claims the dedup slot before the (relatively expensive) +// task creation. Returns false if another handler beat us to it. When the +// jira service isn't wired (boot order corner case), proceed anyway — better +// to risk a duplicate task than silently drop the event. +func (s *Service) reserveJiraIssue(ctx context.Context, evt *jira.NewJiraIssueEvent) bool { + if s.jiraService == nil { + return true + } + reserved, err := s.jiraService.ReserveIssueWatchTask(ctx, evt.IssueWatchID, evt.Issue.Key, evt.Issue.URL) + if err != nil { + s.logger.Error("failed to reserve jira issue slot", + zap.String("issue_key", evt.Issue.Key), zap.Error(err)) + return false + } + if !reserved { + s.logger.Debug("jira issue already reserved by concurrent handler, skipping", + zap.String("issue_key", evt.Issue.Key)) + } + return reserved +} + +func (s *Service) releaseJiraIssue(ctx context.Context, evt *jira.NewJiraIssueEvent) { + if s.jiraService == nil { + return + } + if err := s.jiraService.ReleaseIssueWatchTask(ctx, evt.IssueWatchID, evt.Issue.Key); err != nil { + s.logger.Warn("failed to release jira issue reservation after task-create failure", + zap.String("issue_key", evt.Issue.Key), zap.Error(err)) + } +} + +func (s *Service) attachJiraIssueTaskID(ctx context.Context, evt *jira.NewJiraIssueEvent, taskID string) { + if s.jiraService == nil { + return + } + if err := s.jiraService.AssignIssueWatchTaskID(ctx, evt.IssueWatchID, evt.Issue.Key, taskID); err != nil { + s.logger.Error("failed to assign task ID to jira issue reservation", + zap.String("task_id", taskID), + zap.String("issue_key", evt.Issue.Key), zap.Error(err)) + } +} + +// interpolateJiraPrompt replaces {{issue.*}} placeholders with ticket fields. +// When the template is empty, the agent receives a minimal default that links +// to the JIRA ticket — enough context to start without forcing every watch to +// supply boilerplate. +func interpolateJiraPrompt(template string, t *jira.JiraTicket) string { + if strings.TrimSpace(template) == "" { + template = "Investigate JIRA ticket {{issue.key}}: {{issue.summary}}\n\n{{issue.url}}" + } + r := strings.NewReplacer( + "{{issue.key}}", t.Key, + "{{issue.summary}}", t.Summary, + "{{issue.url}}", t.URL, + "{{issue.status}}", t.StatusName, + "{{issue.priority}}", t.Priority, + "{{issue.type}}", t.IssueType, + "{{issue.assignee}}", t.AssigneeName, + "{{issue.reporter}}", t.ReporterName, + "{{issue.project}}", t.ProjectKey, + "{{issue.description}}", t.Description, + ) + return r.Replace(template) +} diff --git a/apps/backend/internal/orchestrator/event_handlers_jira_test.go b/apps/backend/internal/orchestrator/event_handlers_jira_test.go new file mode 100644 index 0000000000..2fe339710d --- /dev/null +++ b/apps/backend/internal/orchestrator/event_handlers_jira_test.go @@ -0,0 +1,159 @@ +package orchestrator + +import ( + "context" + "errors" + "strings" + "testing" + + "github.com/kandev/kandev/internal/jira" + wfmodels "github.com/kandev/kandev/internal/workflow/models" +) + +// mockJiraService records dedup calls so tests can assert on the +// reserve→assign→release contract used by handleNewJiraIssue. +type mockJiraService struct { + reserveReturn bool + reserveErr error + reserveCalls int + assignCalls int + releaseCalls int + lastWatchID string + lastIssueKey string + assignedTaskID string +} + +func (m *mockJiraService) ReserveIssueWatchTask(_ context.Context, watchID, issueKey, _ string) (bool, error) { + m.reserveCalls++ + m.lastWatchID = watchID + m.lastIssueKey = issueKey + return m.reserveReturn, m.reserveErr +} + +func (m *mockJiraService) AssignIssueWatchTaskID(_ context.Context, _, _ string, taskID string) error { + m.assignCalls++ + m.assignedTaskID = taskID + return nil +} + +func (m *mockJiraService) ReleaseIssueWatchTask(_ context.Context, _, _ string) error { + m.releaseCalls++ + return nil +} + +func newJiraIssueEvent() *jira.NewJiraIssueEvent { + return &jira.NewJiraIssueEvent{ + IssueWatchID: "iw-1", + WorkspaceID: "ws-1", + WorkflowID: "wf1", + WorkflowStepID: "step1", + Issue: &jira.JiraTicket{ + Key: "PROJ-42", + Summary: "Login fails on mobile", + URL: "https://acme.atlassian.net/browse/PROJ-42", + }, + } +} + +func setupJiraTaskTest(t *testing.T) *Service { + t.Helper() + repo := setupTestRepo(t) + stepGetter := newMockStepGetter() + stepGetter.steps["step1"] = &wfmodels.WorkflowStep{ + ID: "step1", WorkflowID: "wf1", Name: "Step 1", Position: 0, + Events: wfmodels.StepEvents{}, + } + return createTestService(repo, stepGetter, newMockTaskRepo()) +} + +func TestCreateJiraIssueTask_HappyPath(t *testing.T) { + svc := setupJiraTaskTest(t) + jiraSvc := &mockJiraService{reserveReturn: true} + svc.SetJiraService(jiraSvc) + creator := &countingIssueTaskCreator{taskID: "task-jira-1"} + svc.SetIssueTaskCreator(creator) + + svc.createJiraIssueTask(context.Background(), newJiraIssueEvent()) + + if jiraSvc.reserveCalls != 1 { + t.Errorf("expected 1 Reserve call, got %d", jiraSvc.reserveCalls) + } + if creator.calls != 1 { + t.Errorf("expected 1 CreateIssueTask call, got %d", creator.calls) + } + if jiraSvc.assignCalls != 1 { + t.Errorf("expected 1 AssignIssueWatchTaskID call, got %d", jiraSvc.assignCalls) + } + if jiraSvc.assignedTaskID != "task-jira-1" { + t.Errorf("expected assigned task id task-jira-1, got %q", jiraSvc.assignedTaskID) + } + if jiraSvc.releaseCalls != 0 { + t.Errorf("expected no Release on happy path, got %d", jiraSvc.releaseCalls) + } + if jiraSvc.lastIssueKey != "PROJ-42" { + t.Errorf("expected reservation keyed on PROJ-42, got %q", jiraSvc.lastIssueKey) + } +} + +func TestCreateJiraIssueTask_SkipsWhenAlreadyReserved(t *testing.T) { + svc := setupJiraTaskTest(t) + jiraSvc := &mockJiraService{reserveReturn: false} + svc.SetJiraService(jiraSvc) + creator := &countingIssueTaskCreator{} + svc.SetIssueTaskCreator(creator) + + svc.createJiraIssueTask(context.Background(), newJiraIssueEvent()) + + if creator.calls != 0 { + t.Errorf("expected CreateIssueTask NOT to be called when reservation is lost, got %d", creator.calls) + } + if jiraSvc.releaseCalls != 0 { + t.Errorf("expected no Release when reservation was never held, got %d", jiraSvc.releaseCalls) + } +} + +func TestCreateJiraIssueTask_ReleasesWhenCreateFails(t *testing.T) { + svc := setupJiraTaskTest(t) + jiraSvc := &mockJiraService{reserveReturn: true} + svc.SetJiraService(jiraSvc) + creator := &countingIssueTaskCreator{err: errors.New("task creation failed")} + svc.SetIssueTaskCreator(creator) + + svc.createJiraIssueTask(context.Background(), newJiraIssueEvent()) + + if jiraSvc.assignCalls != 0 { + t.Errorf("expected no Assign when task creation failed, got %d", jiraSvc.assignCalls) + } + if jiraSvc.releaseCalls != 1 { + t.Errorf("expected Release after task creation failure, got %d", jiraSvc.releaseCalls) + } +} + +func TestInterpolateJiraPrompt(t *testing.T) { + ticket := &jira.JiraTicket{ + Key: "PROJ-7", + Summary: "Login fails on mobile", + URL: "https://acme.atlassian.net/browse/PROJ-7", + StatusName: "In Progress", + Priority: "High", + IssueType: "Bug", + AssigneeName: "Alice", + ProjectKey: "PROJ", + } + + // Empty template falls back to a default that mentions key + URL. + got := interpolateJiraPrompt("", ticket) + if !strings.Contains(got, "PROJ-7") || !strings.Contains(got, "https://acme.atlassian.net/browse/PROJ-7") { + t.Errorf("default template missing key or URL: %q", got) + } + + // All placeholders. + got = interpolateJiraPrompt( + "{{issue.key}} | {{issue.summary}} | {{issue.status}} | {{issue.priority}} | {{issue.assignee}}", + ticket, + ) + want := "PROJ-7 | Login fails on mobile | In Progress | High | Alice" + 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 67fa51338d..e89717edb7 100644 --- a/apps/backend/internal/orchestrator/service.go +++ b/apps/backend/internal/orchestrator/service.go @@ -217,6 +217,9 @@ type Service struct { // Issue task creator for auto-creating tasks from issue watch events issueTaskCreator IssueTaskCreator + // Jira service for issue watch dedup operations + jiraService JiraService + // Repository resolver for cloning + finding/creating repos for review tasks repositoryResolver RepositoryResolver @@ -637,6 +640,9 @@ func (s *Service) Start(ctx context.Context) error { // Subscribe to GitHub integration events s.subscribeGitHubEvents() + // Subscribe to JIRA integration events + s.subscribeJiraEvents() + // Subscribe to clarification events (cancel-and-resume flow) s.subscribeClarificationEvents() diff --git a/apps/web/components/jira/jira-issue-watch-dialog.tsx b/apps/web/components/jira/jira-issue-watch-dialog.tsx new file mode 100644 index 0000000000..4e6a7eaccd --- /dev/null +++ b/apps/web/components/jira/jira-issue-watch-dialog.tsx @@ -0,0 +1,400 @@ +"use client"; + +import { useState, useEffect, useCallback, useMemo } from "react"; +import { Button } from "@kandev/ui/button"; +import { Switch } from "@kandev/ui/switch"; +import { Label } from "@kandev/ui/label"; +import { Input } from "@kandev/ui/input"; +import { + Dialog, + DialogContent, + DialogHeader, + DialogTitle, + DialogDescription, + DialogFooter, +} from "@kandev/ui/dialog"; +import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from "@kandev/ui/select"; +import { Textarea } from "@kandev/ui/textarea"; +import { useAppStore } from "@/components/state-provider"; +import { useSettingsData } from "@/hooks/domains/settings/use-settings-data"; +import { useWorkflows } from "@/hooks/use-workflows"; +import { listWorkflowSteps } from "@/lib/api/domains/workflow-api"; +import { searchJiraTickets } from "@/lib/api/domains/jira-api"; +import type { + CreateJiraIssueWatchInput, + JiraIssueWatch, + UpdateJiraIssueWatchInput, +} from "@/lib/types/jira"; + +type Props = { + open: boolean; + onOpenChange: (open: boolean) => void; + watch: JiraIssueWatch | null; + workspaceId: string; + onCreate: (req: CreateJiraIssueWatchInput) => Promise; + onUpdate: (id: string, req: UpdateJiraIssueWatchInput) => Promise; +}; + +type FormState = { + jql: string; + workflowId: string; + workflowStepId: string; + agentProfileId: string; + executorProfileId: string; + prompt: string; + enabled: boolean; + pollInterval: number; +}; + +const DEFAULT_JQL = `project = PROJ AND status = "Open" ORDER BY created DESC`; +const DEFAULT_PROMPT = `Investigate JIRA ticket {{issue.key}}: {{issue.summary}}\n\n{{issue.url}}`; + +const emptyForm: FormState = { + jql: DEFAULT_JQL, + workflowId: "", + workflowStepId: "", + agentProfileId: "", + executorProfileId: "", + prompt: DEFAULT_PROMPT, + enabled: true, + pollInterval: 300, +}; + +function formStateFromWatch(w: JiraIssueWatch): FormState { + return { + jql: w.jql, + workflowId: w.workflowId, + workflowStepId: w.workflowStepId, + agentProfileId: w.agentProfileId, + executorProfileId: w.executorProfileId, + prompt: w.prompt || DEFAULT_PROMPT, + enabled: w.enabled, + pollInterval: w.pollIntervalSeconds, + }; +} + +type StepOption = { id: string; name: string }; + +function useWorkflowSteps(workflowId: string) { + const [steps, setSteps] = useState([]); + useEffect(() => { + if (!workflowId) { + // Defer to next tick so we don't sync-set state inside an effect body. + void Promise.resolve().then(() => setSteps([])); + return; + } + let cancelled = false; + listWorkflowSteps(workflowId) + .then((res) => { + if (cancelled) return; + const sorted = [...res.steps].sort((a, b) => a.position - b.position); + setSteps(sorted.map((s) => ({ id: s.id, name: s.name }))); + }) + .catch(() => { + if (!cancelled) setSteps([]); + }); + return () => { + cancelled = true; + }; + }, [workflowId]); + return steps; +} + +function useFormData(workspaceId: string) { + useSettingsData(true); + useWorkflows(workspaceId, true); + const workflows = useAppStore((s) => s.workflows.items); + 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 }[]; +}) { + return ( +
+ + {props.description &&

{props.description}

} + +
+ ); +} + +function JQLField({ + workspaceId, + jql, + onChange, +}: { + workspaceId: string; + jql: string; + onChange: (v: string) => void; +}) { + const [result, setResult] = useState<{ ok: boolean; message: string } | null>(null); + const [testing, setTesting] = useState(false); + const handleTest = useCallback(async () => { + setTesting(true); + setResult(null); + try { + const res = await searchJiraTickets(workspaceId, { jql, maxResults: 5 }); + setResult({ ok: true, message: `Matched ${res.tickets.length} ticket(s) in this page.` }); + } catch (err) { + setResult({ ok: false, message: `JQL error: ${String(err)}` }); + } finally { + setTesting(false); + } + }, [workspaceId, jql]); + + return ( +
+
+ + +
+