diff --git a/apps/backend/internal/github/poller_test.go b/apps/backend/internal/github/poller_test.go index 14b699d966..8b47a41684 100644 --- a/apps/backend/internal/github/poller_test.go +++ b/apps/backend/internal/github/poller_test.go @@ -14,6 +14,8 @@ import ( ) // setupPollerTest creates a Poller backed by an in-memory SQLite store and MockClient. +// The tasks table is created so the JOIN in ListActivePRWatches works; tests that +// want a watch to appear in that listing must seed a matching task row via seedTask. func setupPollerTest(t *testing.T) (*Poller, *Service, *MockClient, *Store) { t.Helper() @@ -26,6 +28,15 @@ func setupPollerTest(t *testing.T) (*Poller, *Service, *MockClient, *Store) { sqlxDB := sqlx.NewDb(rawDB, "sqlite3") t.Cleanup(func() { _ = sqlxDB.Close() }) + // Minimal tasks schema: ListActivePRWatches only joins on id and filters by archived_at. + if _, err := sqlxDB.Exec(` + CREATE TABLE tasks ( + id TEXT PRIMARY KEY, + archived_at DATETIME + )`); err != nil { + t.Fatalf("create tasks table: %v", err) + } + store, err := NewStore(sqlxDB, sqlxDB) if err != nil { t.Fatalf("new store: %v", err) @@ -41,6 +52,19 @@ func setupPollerTest(t *testing.T) (*Poller, *Service, *MockClient, *Store) { return poller, svc, mockClient, store } +// seedTask inserts a minimal task row for JOIN-based queries. Pass archived=true +// to seed an archived task (archived_at set to now). +func seedTask(t *testing.T, store *Store, taskID string, archived bool) { + t.Helper() + var archivedAt interface{} + if archived { + archivedAt = time.Now().UTC() + } + if _, err := store.db.Exec(`INSERT INTO tasks (id, archived_at) VALUES (?, ?)`, taskID, archivedAt); err != nil { + t.Fatalf("seed task %s: %v", taskID, err) + } +} + func TestCheckSinglePRWatch_MergedPR_SyncsThenResets(t *testing.T) { poller, _, mockClient, store := setupPollerTest(t) ctx := context.Background() @@ -377,6 +401,7 @@ func TestRefreshStaleBranches_UpdatesBranchWhenChanged(t *testing.T) { ctx := context.Background() // Create a watch with pr_number=0 on old branch. + seedTask(t, store, "t1", false) watch := &PRWatch{ SessionID: "s1", TaskID: "t1", @@ -416,6 +441,7 @@ func TestRefreshStaleBranches_SkipsWhenBranchUnchanged(t *testing.T) { poller, _, _, store := setupPollerTest(t) ctx := context.Background() + seedTask(t, store, "t1", false) watch := &PRWatch{ SessionID: "s1", TaskID: "t1", @@ -448,6 +474,7 @@ func TestRefreshStaleBranches_SkipsWatchesWithPR(t *testing.T) { ctx := context.Background() // Watch that already found a PR (pr_number > 0). + seedTask(t, store, "t1", false) watch := &PRWatch{ SessionID: "s1", TaskID: "t1", @@ -480,6 +507,7 @@ func TestRefreshStaleBranches_SkipsWhenResolverReturnsEmpty(t *testing.T) { poller, _, _, store := setupPollerTest(t) ctx := context.Background() + seedTask(t, store, "t1", false) watch := &PRWatch{ SessionID: "s1", TaskID: "t1", diff --git a/apps/backend/internal/github/provider.go b/apps/backend/internal/github/provider.go index fcb04f9c1c..4030a4a755 100644 --- a/apps/backend/internal/github/provider.go +++ b/apps/backend/internal/github/provider.go @@ -30,7 +30,11 @@ func Provide( } svc := NewService(client, authMethod, secrets, store, eventBus, log) + svc.subscribeTaskEvents() - cleanup := func() error { return nil } + cleanup := func() error { + svc.unsubscribeTaskEvents() + return nil + } return svc, cleanup, nil } diff --git a/apps/backend/internal/github/service.go b/apps/backend/internal/github/service.go index 18a3a7df39..aa966e5a3d 100644 --- a/apps/backend/internal/github/service.go +++ b/apps/backend/internal/github/service.go @@ -47,6 +47,7 @@ type Service struct { taskDeleter TaskDeleter taskSessionChecker TaskSessionChecker syncGroup singleflight.Group + taskEventSubs []bus.Subscription } // NewService creates a new GitHub service. diff --git a/apps/backend/internal/github/service_task_events.go b/apps/backend/internal/github/service_task_events.go new file mode 100644 index 0000000000..c1142d4818 --- /dev/null +++ b/apps/backend/internal/github/service_task_events.go @@ -0,0 +1,98 @@ +package github + +import ( + "context" + + "go.uber.org/zap" + + "github.com/kandev/kandev/internal/events" + "github.com/kandev/kandev/internal/events/bus" +) + +// subscribeTaskEvents wires the github service to task archive/delete events so +// PR watches are pruned proactively. The ListActivePRWatches query already hides +// archived-task watches from the poller, but pruning keeps the table bounded. +// Subscriptions are tracked so they can be released via unsubscribeTaskEvents. +func (s *Service) subscribeTaskEvents() { + if s.eventBus == nil { + return + } + if sub, err := s.eventBus.Subscribe(events.TaskUpdated, s.handleTaskUpdated); err != nil { + s.logger.Error("failed to subscribe to task.updated events", zap.Error(err)) + } else { + s.taskEventSubs = append(s.taskEventSubs, sub) + } + if sub, err := s.eventBus.Subscribe(events.TaskDeleted, s.handleTaskDeleted); err != nil { + s.logger.Error("failed to subscribe to task.deleted events", zap.Error(err)) + } else { + s.taskEventSubs = append(s.taskEventSubs, sub) + } +} + +// unsubscribeTaskEvents releases the subscriptions created in subscribeTaskEvents. +// Called from the provider cleanup so handlers don't accumulate if the service is +// torn down and re-created while the event bus persists. +func (s *Service) unsubscribeTaskEvents() { + for _, sub := range s.taskEventSubs { + if err := sub.Unsubscribe(); err != nil { + s.logger.Error("failed to unsubscribe from task event", zap.Error(err)) + } + } + s.taskEventSubs = nil +} + +// handleTaskUpdated deletes PR watches when a task is archived. Non-archive +// updates are ignored so we don't interfere with normal task edits. +func (s *Service) handleTaskUpdated(ctx context.Context, event *bus.Event) error { + taskID, archived := taskIDAndArchivedFrom(event) + if taskID == "" || !archived { + return nil + } + s.pruneWatchesForTask(ctx, taskID, "archived") + return nil +} + +// handleTaskDeleted deletes PR watches when a task is hard-deleted. +func (s *Service) handleTaskDeleted(ctx context.Context, event *bus.Event) error { + taskID, _ := taskIDAndArchivedFrom(event) + if taskID == "" { + return nil + } + s.pruneWatchesForTask(ctx, taskID, "deleted") + return nil +} + +func (s *Service) pruneWatchesForTask(ctx context.Context, taskID, reason string) { + n, err := s.store.DeletePRWatchesByTaskID(ctx, taskID) + if err != nil { + s.logger.Error("failed to delete PR watches for task", + zap.String("task_id", taskID), + zap.String("reason", reason), + zap.Error(err)) + return + } + if n > 0 { + s.logger.Info("pruned PR watches after task change", + zap.String("task_id", taskID), + zap.String("reason", reason), + zap.Int64("deleted", n)) + } +} + +// taskIDAndArchivedFrom extracts task_id + archived flag from a task event +// payload (task-service publishes a map[string]interface{} — see +// internal/task/service/service_events.go publishTaskEvent). Archived is +// detected from a non-empty string value so a future null/zero archived_at +// in the payload doesn't silently prune watches on non-archive updates. +func taskIDAndArchivedFrom(event *bus.Event) (taskID string, archived bool) { + if event == nil { + return "", false + } + data, ok := event.Data.(map[string]interface{}) + if !ok { + return "", false + } + id, _ := data["task_id"].(string) + archivedStr, _ := data["archived_at"].(string) + return id, archivedStr != "" +} diff --git a/apps/backend/internal/github/service_task_events_test.go b/apps/backend/internal/github/service_task_events_test.go new file mode 100644 index 0000000000..a5456968ad --- /dev/null +++ b/apps/backend/internal/github/service_task_events_test.go @@ -0,0 +1,171 @@ +package github + +import ( + "context" + "testing" + + "github.com/kandev/kandev/internal/events" + "github.com/kandev/kandev/internal/events/bus" +) + +func TestListActivePRWatches_ExcludesArchivedTasks(t *testing.T) { + _, svc, _, store := setupPollerTest(t) + ctx := context.Background() + + seedTask(t, store, "t-active", false) + seedTask(t, store, "t-archived", true) + + mustCreateWatch(t, store, "s-active", "t-active") + mustCreateWatch(t, store, "s-archived", "t-archived") + + watches, err := svc.ListActivePRWatches(ctx) + if err != nil { + t.Fatalf("list watches: %v", err) + } + if len(watches) != 1 { + t.Fatalf("expected 1 watch (active only), got %d", len(watches)) + } + if watches[0].TaskID != "t-active" { + t.Errorf("expected watch for t-active, got %q", watches[0].TaskID) + } +} + +func TestListActivePRWatches_ExcludesOrphanedWatches(t *testing.T) { + _, svc, _, store := setupPollerTest(t) + ctx := context.Background() + + // Orphan: watch exists but no matching task row (task was hard-deleted). + mustCreateWatch(t, store, "s-orphan", "t-gone") + + watches, err := svc.ListActivePRWatches(ctx) + if err != nil { + t.Fatalf("list watches: %v", err) + } + if len(watches) != 0 { + t.Fatalf("expected orphaned watch to be excluded, got %d watches", len(watches)) + } +} + +func TestHandleTaskUpdated_ArchiveDeletesWatches(t *testing.T) { + _, svc, _, store := setupPollerTest(t) + ctx := context.Background() + + seedTask(t, store, "t1", false) + mustCreateWatch(t, store, "s1", "t1") + + // Simulate task-service publishing task.updated with archived_at set. + event := bus.NewEvent(events.TaskUpdated, "task-service", map[string]interface{}{ + "task_id": "t1", + "archived_at": "2026-04-19T12:00:00Z", + }) + if err := svc.handleTaskUpdated(ctx, event); err != nil { + t.Fatalf("handleTaskUpdated: %v", err) + } + + if got, _ := store.GetPRWatchBySession(ctx, "s1"); got != nil { + t.Errorf("expected watch to be deleted after archive event, got %+v", got) + } +} + +func TestHandleTaskUpdated_NonArchiveUpdateLeavesWatches(t *testing.T) { + _, svc, _, store := setupPollerTest(t) + ctx := context.Background() + + seedTask(t, store, "t1", false) + mustCreateWatch(t, store, "s1", "t1") + + // Regular edit: no archived_at in payload. + event := bus.NewEvent(events.TaskUpdated, "task-service", map[string]interface{}{ + "task_id": "t1", + "title": "Edited title", + }) + if err := svc.handleTaskUpdated(ctx, event); err != nil { + t.Fatalf("handleTaskUpdated: %v", err) + } + + if got, _ := store.GetPRWatchBySession(ctx, "s1"); got == nil { + t.Error("expected watch to persist after non-archive update") + } +} + +func TestHandleTaskDeleted_DeletesWatches(t *testing.T) { + _, svc, _, store := setupPollerTest(t) + ctx := context.Background() + + seedTask(t, store, "t1", false) + mustCreateWatch(t, store, "s1", "t1") + + event := bus.NewEvent(events.TaskDeleted, "task-service", map[string]interface{}{ + "task_id": "t1", + }) + if err := svc.handleTaskDeleted(ctx, event); err != nil { + t.Fatalf("handleTaskDeleted: %v", err) + } + + if got, _ := store.GetPRWatchBySession(ctx, "s1"); got != nil { + t.Errorf("expected watch to be deleted after delete event, got %+v", got) + } +} + +func TestHandleTaskEvents_MalformedPayloadIsNoop(t *testing.T) { + _, svc, _, store := setupPollerTest(t) + ctx := context.Background() + + seedTask(t, store, "t1", false) + mustCreateWatch(t, store, "s1", "t1") + + // Wrong payload type — should be ignored, not crash. + bad := bus.NewEvent(events.TaskUpdated, "task-service", "not-a-map") + if err := svc.handleTaskUpdated(ctx, bad); err != nil { + t.Fatalf("handleTaskUpdated: %v", err) + } + + // Missing task_id — should be ignored. + empty := bus.NewEvent(events.TaskDeleted, "task-service", map[string]interface{}{}) + if err := svc.handleTaskDeleted(ctx, empty); err != nil { + t.Fatalf("handleTaskDeleted: %v", err) + } + + if got, _ := store.GetPRWatchBySession(ctx, "s1"); got == nil { + t.Error("expected watch to persist when payload is malformed") + } +} + +func TestSubscribeTaskEvents_EndToEnd(t *testing.T) { + _, svc, _, store := setupPollerTest(t) + ctx := context.Background() + + svc.subscribeTaskEvents() + + seedTask(t, store, "t1", false) + mustCreateWatch(t, store, "s1", "t1") + + event := bus.NewEvent(events.TaskUpdated, "task-service", map[string]interface{}{ + "task_id": "t1", + "archived_at": "2026-04-19T12:00:00Z", + }) + if err := svc.eventBus.Publish(ctx, events.TaskUpdated, event); err != nil { + t.Fatalf("publish: %v", err) + } + + // MemoryEventBus delivers synchronously, so the watch should already be gone. + if got, _ := store.GetPRWatchBySession(ctx, "s1"); got != nil { + t.Errorf("expected watch to be deleted by subscribed handler, got %+v", got) + } +} + +// mustCreateWatch is a test helper for creating a PR watch with minimal boilerplate. +func mustCreateWatch(t *testing.T, store *Store, sessionID, taskID string) { + t.Helper() + w := &PRWatch{ + SessionID: sessionID, + TaskID: taskID, + Owner: "owner", + Repo: "repo", + PRNumber: 0, + Branch: "main", + } + if err := store.CreatePRWatch(context.Background(), w); err != nil { + t.Fatalf("create PR watch: %v", err) + } +} diff --git a/apps/backend/internal/github/store.go b/apps/backend/internal/github/store.go index 8f1b160680..60cfc0eed9 100644 --- a/apps/backend/internal/github/store.go +++ b/apps/backend/internal/github/store.go @@ -151,10 +151,17 @@ func (s *Store) GetPRWatchByTask(ctx context.Context, taskID string) (*PRWatch, return &w, err } -// ListActivePRWatches returns all active PR watches. +// ListActivePRWatches returns all active PR watches whose task is not archived. +// Watches for archived tasks (and orphaned watches whose task row was hard-deleted) +// are excluded so the poller stops making GitHub API calls for them. An INNER JOIN +// on `tasks` is used so orphans are dropped automatically. func (s *Store) ListActivePRWatches(ctx context.Context) ([]*PRWatch, error) { var watches []*PRWatch - err := s.ro.SelectContext(ctx, &watches, `SELECT * FROM github_pr_watches ORDER BY created_at`) + err := s.ro.SelectContext(ctx, &watches, ` + SELECT w.* FROM github_pr_watches w + INNER JOIN tasks t ON t.id = w.task_id + WHERE t.archived_at IS NULL + ORDER BY w.created_at`) return watches, err } @@ -173,6 +180,20 @@ func (s *Store) DeletePRWatch(ctx context.Context, id string) error { return err } +// DeletePRWatchesByTaskID deletes all PR watches for a task. Returns the number +// of rows removed so callers can log meaningful diagnostics. +func (s *Store) DeletePRWatchesByTaskID(ctx context.Context, taskID string) (int64, error) { + res, err := s.db.ExecContext(ctx, `DELETE FROM github_pr_watches WHERE task_id = ?`, taskID) + if err != nil { + return 0, err + } + n, err := res.RowsAffected() + if err != nil { + return 0, err + } + return n, nil +} + // UpdatePRWatchPRNumber updates a PR watch's PR number after discovery. func (s *Store) UpdatePRWatchPRNumber(ctx context.Context, id string, prNumber int) error { _, err := s.db.ExecContext(ctx,