Add real-time notification support via sprinkler

cmd/goose/cache.go

@@ -24,8 +24,7 @@
 }
 
 // turnData fetches Turn API data with caching.
 func (app *App) turnData(ctx context.Context, url string, updatedAt time.Time) (*turn.CheckResponse, bool, error) {
-	prAge := time.Since(updatedAt)
 	hasRunningTests := false
 	// Validate URL before processing
 	if err := validateURL(url); err != nil {
@@ -57,13 +56,16 @@
 				}
 			} else if time.Since(entry.CachedAt) < cacheTTL && entry.UpdatedAt.Equal(updatedAt) {
 				// Check if cache is still valid (10 day TTL, but PR UpdatedAt is primary check)
-				// But invalidate cache for PRs with running tests if they're fresh (< 90 minutes old)
-				if entry.Data != nil && entry.Data.PullRequest.TestState == "running" && prAge < runningTestsCacheBypass {
+				// But invalidate cache for PRs with incomplete tests if cache entry is fresh (< 90 minutes old)
+				cacheAge := time.Since(entry.CachedAt)
+				testState := entry.Data.PullRequest.TestState
+				isTestIncomplete := testState == "running" || testState == "queued" || testState == "pending"
+				if entry.Data != nil && isTestIncomplete && cacheAge < runningTestsCacheBypass {
 					hasRunningTests = true
-					slog.Debug("[CACHE] Cache invalidated - PR has running tests and is fresh",
+					slog.Debug("[CACHE] Cache invalidated - tests incomplete and cache entry is fresh",
 						"url", url,
-						"test_state", entry.Data.PullRequest.TestState,
-						"pr_age", prAge.Round(time.Minute),
+						"test_state", testState,
+						"cache_age", cacheAge.Round(time.Minute),
 						"cached_at", entry.CachedAt.Format(time.RFC3339))
 					// Don't return cached data - fall through to fetch fresh data with current time
 				} else {
@@ -160,18 +162,21 @@
 	}
 
 	// Save to cache (don't fail if caching fails) - skip if --no-cache is set
-	// Also skip caching if tests are running and PR is fresh (updated in last 90 minutes)
+	// Don't cache when tests are incomplete - always re-poll to catch completion
 	if !app.noCache {
 		shouldCache := true
-		prAge := time.Since(updatedAt)
 
-		// Don't cache PRs with running tests unless they're older than 90 minutes
-		if data != nil && data.PullRequest.TestState == "running" && prAge < runningTestsCacheBypass {
+		// Never cache PRs with incomplete tests - we want fresh data on every poll
+		testState := ""
+		if data != nil {
+			testState = data.PullRequest.TestState
+		}
+		isTestIncomplete := testState == "running" || testState == "queued" || testState == "pending"
+		if data != nil && isTestIncomplete {
 			shouldCache = false
-			slog.Debug("[CACHE] Skipping cache for PR with running tests",
+			slog.Debug("[CACHE] Skipping cache for PR with incomplete tests",
 				"url", url,
-				"test_state", data.PullRequest.TestState,
-				"pr_age", prAge.Round(time.Minute),
+				"test_state", testState,
 				"pending_checks", len(data.PullRequest.CheckSummary.PendingStatuses))
 		}
 

cmd/goose/github.go

@@ -53,6 +53,9 @@ func (app *App) initClients(ctx context.Context) error {
 	turnClient.SetAuthToken(token)
 	app.turnClient = turnClient
 
+	// Initialize sprinkler monitor for real-time events
+	app.sprinklerMonitor = newSprinklerMonitor(app, token)
+
 	return nil
 }
 
@@ -396,15 +399,33 @@ func (app *App) fetchPRsInternal(ctx context.Context) (incoming []PR, outgoing [
 		// Categorize as incoming or outgoing
 		// When viewing another user's PRs, we're looking at it from their perspective
 		if issue.GetUser().GetLogin() == user {
+			slog.Info("[GITHUB] Found outgoing PR", "repo", repo, "number", pr.Number, "author", pr.Author, "url", pr.URL)
 			outgoing = append(outgoing, pr)
 		} else {
+			slog.Info("[GITHUB] Found incoming PR", "repo", repo, "number", pr.Number, "author", pr.Author, "url", pr.URL)
 			incoming = append(incoming, pr)
 		}
 	}
 
 	// Only log summary, not individual PRs
 	slog.Info("[GITHUB] GitHub PR summary", "incoming", len(incoming), "outgoing", len(outgoing))
 
+	// Update sprinkler monitor with discovered orgs
+	app.mu.RLock()
+	orgs := make([]string, 0, len(app.seenOrgs))
+	for org := range app.seenOrgs {
+		orgs = append(orgs, org)
+	}
+	app.mu.RUnlock()
+
+	if app.sprinklerMonitor != nil && len(orgs) > 0 {
+		app.sprinklerMonitor.updateOrgs(orgs)
+		// Start monitor if not already running
+		if err := app.sprinklerMonitor.start(); err != nil {
+			slog.Warn("[SPRINKLER] Failed to start monitor", "error", err)
+		}
+	}
+
 	// Fetch Turn API data
 	// Always synchronous now for simplicity - Turn API calls are fast with caching
 	app.fetchTurnDataSync(ctx, allIssues, user, &incoming, &outgoing)

cmd/goose/main.go

@@ -82,6 +82,7 @@
 	hiddenOrgs                   map[string]bool
 	seenOrgs                     map[string]bool
 	turnClient                   *turn.Client
+	sprinklerMonitor             *sprinklerMonitor
 	previousBlockedPRs           map[string]bool
 	authError                    string
 	lastFetchError               string
@@ -94,6 +95,7 @@
 	consecutiveFailures          int
 	mu                           sync.RWMutex
 	menuMutex                    sync.Mutex // Mutex to prevent concurrent menu rebuilds
+	updateMutex                  sync.Mutex // Mutex to prevent concurrent PR updates
 	enableAutoBrowser            bool
 	hideStaleIncoming            bool
 	hasPerformedInitialDiscovery bool // Track if we've done the first poll to distinguish from real state changes
@@ -257,7 +259,7 @@
 			}),
 			retry.Context(ctx),
 		)
 		if err != nil {
 			slog.Warn("Failed to load current user after retries", "maxRetries", maxRetries, "error", err)
 			if app.authError == "" {
 				app.authError = fmt.Sprintf("Failed to load user: %v", err)
@@ -283,11 +285,14 @@
 	systray.Run(func() { app.onReady(appCtx) }, func() {
 		slog.Info("Shutting down application")
 		cancel() // Cancel the context to stop goroutines
+		if app.sprinklerMonitor != nil {
+			app.sprinklerMonitor.stop()
+		}
 		app.cleanupOldCache()
 	})
 }
 
 func (app *App) onReady(ctx context.Context) {
 	slog.Info("System tray ready")
 
 	// On Linux, immediately build a minimal menu to ensure it's visible
@@ -504,6 +509,13 @@
 }
 
 func (app *App) updatePRs(ctx context.Context) {
+	// Prevent concurrent updates
+	if !app.updateMutex.TryLock() {
+		slog.Debug("[UPDATE] Update already in progress, skipping")
+		return
+	}
+	defer app.updateMutex.Unlock()
+
 	var incoming, outgoing []PR
 	err := safeExecute("fetchPRs", func() error {
 		var fetchErr error
@@ -615,7 +627,7 @@
 		"outgoing_count", len(outgoing))
 	// Log ALL outgoing PRs for debugging
 	slog.Debug("[UPDATE] Listing ALL outgoing PRs for debugging")
 	for i, pr := range outgoing {
 		slog.Debug("[UPDATE] Outgoing PR details",
 			"index", i,
 			"repo", pr.Repository,
@@ -683,6 +695,13 @@
 
 // updatePRsWithWait fetches PRs and waits for Turn data before building initial menu.
 func (app *App) updatePRsWithWait(ctx context.Context) {
+	// Prevent concurrent updates
+	if !app.updateMutex.TryLock() {
+		slog.Debug("[UPDATE] Update already in progress, skipping")
+		return
+	}
+	defer app.updateMutex.Unlock()
+
 	incoming, outgoing, err := app.fetchPRsInternal(ctx)
 	if err != nil {
 		slog.Error("Error fetching PRs", "error", err)
@@ -802,7 +821,7 @@
 }
 
 // tryAutoOpenPR attempts to open a PR in the browser if enabled and rate limits allow.
 func (app *App) tryAutoOpenPR(ctx context.Context, pr PR, autoBrowserEnabled bool, startTime time.Time) {
 	if !autoBrowserEnabled {
 		return
 	}