diff --git a/cmd/reviewGOOSE/cache.go b/cmd/reviewGOOSE/cache.go index 7aa4218..10bc07c 100644 --- a/cmd/reviewGOOSE/cache.go +++ b/cmd/reviewGOOSE/cache.go @@ -96,6 +96,12 @@ func (app *App) checkCache(cacheFile, url string, updatedAt time.Time) (cachedDa // turnData fetches Turn API data with caching. func (app *App) turnData(ctx context.Context, url string, updatedAt time.Time) (*turn.CheckResponse, bool, error) { + // If Turn API is disabled, return nil without error + if app.turnClient == nil { + slog.Debug("[TURN] Turn API disabled, skipping", "url", url) + return nil, false, nil + } + hasRunningTests := false // Validate URL before processing if err := safebrowse.ValidateURL(url); err != nil { diff --git a/cmd/reviewGOOSE/github.go b/cmd/reviewGOOSE/github.go index 34050dc..b5deab3 100644 --- a/cmd/reviewGOOSE/github.go +++ b/cmd/reviewGOOSE/github.go @@ -44,24 +44,51 @@ func (app *App) initClients(ctx context.Context) error { tc := oauth2.NewClient(ctx, ts) app.client = github.NewClient(tc) - // Initialize Turn client using default backend - turnClient, err := turn.NewDefaultClient() - if err != nil { - return fmt.Errorf("create turn client: %w", err) + // Check for custom turn server hostname (for self-hosting) + // Set TURNSERVER=disabled to run without Turn API + turnServer := os.Getenv("TURNSERVER") + if turnServer == "disabled" { + slog.Info("Turn API disabled via TURNSERVER=disabled") + } else { + var turnClient *turn.Client + if turnServer != "" { + slog.Info("Using custom turn server", "hostname", turnServer) + turnClient, err = turn.NewClient("https://" + turnServer) + } else { + turnClient, err = turn.NewDefaultClient() + } + if err != nil { + return fmt.Errorf("create turn client: %w", err) + } + turnClient.SetAuthToken(token) + app.turnClient = turnClient } - turnClient.SetAuthToken(token) - app.turnClient = turnClient // Initialize sprinkler monitor for real-time events - app.sprinklerMonitor = newSprinklerMonitor(app, token) + // Check for custom sprinkler server hostname (for self-hosting) + // Set SPRINKLER=disabled to run without real-time events + sprinklerServer := os.Getenv("SPRINKLER") + if sprinklerServer == "disabled" { + slog.Info("Sprinkler disabled via SPRINKLER=disabled") + } else { + if sprinklerServer != "" { + slog.Info("Using custom sprinkler server", "hostname", sprinklerServer) + } + app.sprinklerMonitor = newSprinklerMonitor(app, token, sprinklerServer) + } return nil } // initSprinklerOrgs fetches the user's organizations and starts sprinkler monitoring. func (app *App) initSprinklerOrgs(ctx context.Context) error { - if app.client == nil || app.sprinklerMonitor == nil { - return errors.New("client or sprinkler not initialized") + if app.client == nil { + return errors.New("github client not initialized") + } + // If sprinkler is disabled, skip silently + if app.sprinklerMonitor == nil { + slog.Debug("[SPRINKLER] Sprinkler disabled, skipping org initialization") + return nil } // Get current user @@ -80,22 +107,21 @@ func (app *App) initSprinklerOrgs(ctx context.Context) error { // Fetch all orgs the user is a member of with retry opts := &github.ListOptions{PerPage: 100} - var allOrgs []string + var orgs []string for { - var orgs []*github.Organization + var page []*github.Organization var resp *github.Response err := retry.Do(func() error { - // Create timeout context for API call apiCtx, cancel := context.WithTimeout(ctx, 30*time.Second) defer cancel() - var retryErr error - orgs, resp, retryErr = app.client.Organizations.List(apiCtx, user, opts) - if retryErr != nil { - slog.Debug("[SPRINKLER] Organizations.List failed (will retry)", "error", retryErr, "page", opts.Page) - return retryErr + var err error + page, resp, err = app.client.Organizations.List(apiCtx, user, opts) + if err != nil { + slog.Debug("[SPRINKLER] Organizations.List failed (will retry)", "error", err, "page", opts.Page) + return err } return nil }, @@ -115,9 +141,9 @@ func (app *App) initSprinklerOrgs(ctx context.Context) error { return nil // Return nil to avoid blocking startup } - for _, org := range orgs { - if org.Login != nil { - allOrgs = append(allOrgs, *org.Login) + for _, o := range page { + if o.Login != nil { + orgs = append(orgs, *o.Login) } } @@ -129,12 +155,12 @@ func (app *App) initSprinklerOrgs(ctx context.Context) error { slog.Info("[SPRINKLER] Discovered user organizations", "user", user, - "orgs", allOrgs, - "count", len(allOrgs)) + "orgs", orgs, + "count", len(orgs)) // Update sprinkler with all orgs at once - if len(allOrgs) > 0 { - app.sprinklerMonitor.updateOrgs(allOrgs) + if len(orgs) > 0 { + app.sprinklerMonitor.updateOrgs(orgs) if err := app.sprinklerMonitor.start(ctx); err != nil { return fmt.Errorf("start sprinkler: %w", err) } @@ -387,79 +413,78 @@ func (app *App) fetchPRsInternal(ctx context.Context) (incoming []PR, outgoing [ searchStart := time.Now() // Run both queries in parallel - type queryResult struct { + type qResult struct { err error query string issues []*github.Issue } - queryResults := make(chan queryResult, 2) + results := make(chan qResult, 2) // Query 1: PRs involving the user go func() { - query := fmt.Sprintf("is:open is:pr involves:%s archived:false", user) - slog.Debug("[GITHUB] Searching for PRs", "query", query) + q := fmt.Sprintf("is:open is:pr involves:%s archived:false", user) + slog.Debug("[GITHUB] Searching for PRs", "query", q) - result, err := app.executeGitHubQuery(ctx, query, opts) + res, err := app.executeGitHubQuery(ctx, q, opts) if err != nil { - queryResults <- queryResult{err: err, query: query} + results <- qResult{err: err, query: q} } else { - queryResults <- queryResult{issues: result.Issues, query: query} + results <- qResult{issues: res.Issues, query: q} } }() // Query 2: PRs in user-owned repos with no reviewers go func() { - query := fmt.Sprintf("is:open is:pr user:%s review:none archived:false", user) - slog.Debug("[GITHUB] Searching for PRs", "query", query) + q := fmt.Sprintf("is:open is:pr user:%s review:none archived:false", user) + slog.Debug("[GITHUB] Searching for PRs", "query", q) - result, err := app.executeGitHubQuery(ctx, query, opts) + res, err := app.executeGitHubQuery(ctx, q, opts) if err != nil { - queryResults <- queryResult{err: err, query: query} + results <- qResult{err: err, query: q} } else { - queryResults <- queryResult{issues: result.Issues, query: query} + results <- qResult{issues: res.Issues, query: q} } }() // Collect results from both queries - var allIssues []*github.Issue - seenURLs := make(map[string]bool) - var queryErrors []error + var issues []*github.Issue + seen := make(map[string]bool) + var errs []error for range 2 { - result := <-queryResults - if result.err != nil { - slog.Error("[GITHUB] Query failed", "query", result.query, "error", result.err) - queryErrors = append(queryErrors, result.err) - // Continue processing other query results even if one fails + r := <-results + if r.err != nil { + slog.Error("[GITHUB] Query failed", "query", r.query, "error", r.err) + errs = append(errs, r.err) continue } - slog.Debug("[GITHUB] Query completed", "query", result.query, "prCount", len(result.issues)) + slog.Debug("[GITHUB] Query completed", "query", r.query, "prCount", len(r.issues)) // Deduplicate PRs based on URL - for _, issue := range result.issues { + for _, issue := range r.issues { url := issue.GetHTMLURL() - if !seenURLs[url] { - seenURLs[url] = true - allIssues = append(allIssues, issue) + if !seen[url] { + seen[url] = true + issues = append(issues, issue) } } } - slog.Info("[GITHUB] Both searches completed", "duration", time.Since(searchStart), "uniquePRs", len(allIssues)) + slog.Info("[GITHUB] Both searches completed", "duration", time.Since(searchStart), "uniquePRs", len(issues)) // If both queries failed, return an error - if len(queryErrors) == 2 { - return nil, nil, fmt.Errorf("all GitHub queries failed: %v", queryErrors) + if len(errs) == 2 { + return nil, nil, fmt.Errorf("all GitHub queries failed: %v", errs) } // Limit PRs for performance - if len(allIssues) > maxPRsToProcess { - slog.Info("Limiting PRs for performance", "limit", maxPRsToProcess, "total", len(allIssues)) - allIssues = allIssues[:maxPRsToProcess] + if len(issues) > maxPRsToProcess { + slog.Info("Limiting PRs for performance", "limit", maxPRsToProcess, "total", len(issues)) + issues = issues[:maxPRsToProcess] } // Process GitHub results immediately - for _, issue := range allIssues { + for _, issue := range issues { if !issue.IsPullRequest() { continue } @@ -503,7 +528,7 @@ func (app *App) fetchPRsInternal(ctx context.Context) (incoming []PR, outgoing [ // Fetch Turn API data // Always synchronous now for simplicity - Turn API calls are fast with caching - app.fetchTurnDataSync(ctx, allIssues, user, &incoming, &outgoing) + app.fetchTurnDataSync(ctx, issues, user, &incoming, &outgoing) return incoming, outgoing, nil } diff --git a/cmd/reviewGOOSE/main.go b/cmd/reviewGOOSE/main.go index 6b0ce20..bd220f7 100644 --- a/cmd/reviewGOOSE/main.go +++ b/cmd/reviewGOOSE/main.go @@ -59,7 +59,7 @@ const ( runningTestsCacheBypass = 90 * time.Minute // Don't cache PRs with running tests if fresher than this maxPRsToProcess = 200 minUpdateInterval = 10 * time.Second - defaultUpdateInterval = 1 * time.Minute + defaultUpdateInterval = 2 * time.Minute blockedPRIconDuration = 5 * time.Minute maxRetryDelay = 2 * time.Minute maxRetries = 10 diff --git a/cmd/reviewGOOSE/main_test.go b/cmd/reviewGOOSE/main_test.go index 30cc666..2b15995 100644 --- a/cmd/reviewGOOSE/main_test.go +++ b/cmd/reviewGOOSE/main_test.go @@ -2,7 +2,10 @@ package main import ( "context" + "encoding/json" "fmt" + "net/http" + "net/http/httptest" "os" "runtime" "strings" @@ -11,6 +14,7 @@ import ( "time" "github.com/codeGROOVE-dev/turnclient/pkg/turn" + "github.com/google/go-github/v57/github" ) func TestMain(m *testing.M) { @@ -1031,3 +1035,160 @@ func TestAuthErrorStatePreservation(t *testing.T) { } app.mu.RUnlock() } + +// TestTurnDataDisabled tests that turnData returns nil gracefully when Turn API is disabled. +func TestTurnDataDisabled(t *testing.T) { + ctx := context.Background() + + app := &App{ + mu: sync.RWMutex{}, + turnClient: nil, // Simulates TURNSERVER=disabled + cacheDir: t.TempDir(), + } + + // turnData should return nil without error when disabled + data, cached, err := app.turnData(ctx, "https://github.com/test/repo/pull/1", time.Now()) + if err != nil { + t.Errorf("Expected no error when Turn API disabled, got: %v", err) + } + if data != nil { + t.Error("Expected nil data when Turn API disabled") + } + if cached { + t.Error("Expected cached=false when Turn API disabled") + } +} + +// TestSprinklerDisabled tests that initSprinklerOrgs returns nil gracefully when Sprinkler is disabled. +func TestSprinklerDisabled(t *testing.T) { + ctx := context.Background() + + app := &App{ + mu: sync.RWMutex{}, + client: github.NewClient(nil), // Need a non-nil client + sprinklerMonitor: nil, // Simulates SPRINKLER=disabled + } + + // initSprinklerOrgs should return nil without error when disabled + err := app.initSprinklerOrgs(ctx) + if err != nil { + t.Errorf("Expected no error when Sprinkler disabled, got: %v", err) + } +} + +// TestCustomTurnServer tests that a custom TURNSERVER hostname routes requests correctly. +func TestCustomTurnServer(t *testing.T) { + ctx := context.Background() + + // Create a mock Turn API server + requestReceived := false + mockServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requestReceived = true + + // Verify the request is to the validate endpoint + if r.URL.Path != "/v1/validate" { + t.Errorf("Expected request to /v1/validate, got: %s", r.URL.Path) + } + + // Return a valid Turn API response matching the expected schema: + // - turn.CheckResponse contains prx.PullRequest and turn.Analysis + // - CheckSummary fields are map[string]string (check name -> status description) + resp := map[string]any{ + "timestamp": time.Now().Format(time.RFC3339), + "commit": "abc123def456", + "pull_request": map[string]any{ + "number": 1, + "state": "open", + "title": "Test PR", + "author": "testauthor", + "author_bot": false, + "draft": false, + "merged": false, + "test_state": "passing", + "created_at": time.Now().Add(-24 * time.Hour).Format(time.RFC3339), + "updated_at": time.Now().Format(time.RFC3339), + "head_sha": "abc123def456", + "check_summary": map[string]any{ + "success": map[string]string{"ci/test": "All tests passed"}, + "failing": map[string]string{}, + "pending": map[string]string{}, + "cancelled": map[string]string{}, + "skipped": map[string]string{}, + "stale": map[string]string{}, + "neutral": map[string]string{}, + }, + }, + "analysis": map[string]any{ + "workflow_state": "WAITING_FOR_REVIEW", + "next_action": map[string]any{ + "testuser": map[string]any{ + "kind": "review", + "reason": "PR is ready for review", + "critical": true, + "since": time.Now().Format(time.RFC3339), + }, + }, + "last_activity": map[string]any{ + "timestamp": time.Now().Format(time.RFC3339), + "kind": "push", + "actor": "testauthor", + "message": "Pushed new commits", + }, + "size": "S", + "ready_merge": false, + }, + } + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(resp); err != nil { + t.Errorf("Failed to encode response: %v", err) + } + })) + defer mockServer.Close() + + // Create a Turn client pointing to our mock server + turnClient, err := turn.NewClient(mockServer.URL) + if err != nil { + t.Fatalf("Failed to create turn client: %v", err) + } + turnClient.SetAuthToken("test-token") + + // Create app with the custom turn client + login := "testuser" + app := &App{ + mu: sync.RWMutex{}, + turnClient: turnClient, + cacheDir: t.TempDir(), + noCache: true, // Skip cache to ensure we hit the API + currentUser: &github.User{ + Login: &login, + }, + } + + // Make a request + data, _, err := app.turnData(ctx, "https://github.com/test/repo/pull/1", time.Now()) + if err != nil { + t.Fatalf("turnData failed: %v", err) + } + + // Verify the mock server received the request + if !requestReceived { + t.Error("Expected request to be sent to custom Turn server") + } + + // Verify we got a valid response + if data == nil { + t.Fatal("Expected non-nil response from custom Turn server") + } + if data.PullRequest.State != "open" { + t.Errorf("Expected state 'open', got: %s", data.PullRequest.State) + } + if data.PullRequest.TestState != "passing" { + t.Errorf("Expected test_state 'passing', got: %s", data.PullRequest.TestState) + } + if data.Analysis.WorkflowState != "WAITING_FOR_REVIEW" { + t.Errorf("Expected workflow_state 'WAITING_FOR_REVIEW', got: %s", data.Analysis.WorkflowState) + } + if _, hasAction := data.Analysis.NextAction["testuser"]; !hasAction { + t.Error("Expected NextAction to contain 'testuser'") + } +} diff --git a/cmd/reviewGOOSE/sprinkler.go b/cmd/reviewGOOSE/sprinkler.go index c25c65c..2e5eb45 100644 --- a/cmd/reviewGOOSE/sprinkler.go +++ b/cmd/reviewGOOSE/sprinkler.go @@ -40,6 +40,7 @@ type sprinklerMonitor struct { eventChan chan prEvent lastEventMap map[string]time.Time token string + serverAddress string // Custom server hostname (empty = use default) orgs []string mu sync.RWMutex isRunning bool @@ -47,13 +48,15 @@ type sprinklerMonitor struct { } // newSprinklerMonitor creates a new sprinkler monitor for real-time PR events. -func newSprinklerMonitor(app *App, token string) *sprinklerMonitor { +// If sprinklerServer is non-empty, it will be used as the WebSocket server hostname. +func newSprinklerMonitor(app *App, token, sprinklerServer string) *sprinklerMonitor { return &sprinklerMonitor{ - app: app, - token: token, - orgs: make([]string, 0), - eventChan: make(chan prEvent, eventChannelSize), - lastEventMap: make(map[string]time.Time), + app: app, + token: token, + serverAddress: sprinklerServer, + orgs: make([]string, 0), + eventChan: make(chan prEvent, eventChannelSize), + lastEventMap: make(map[string]time.Time), } } @@ -106,8 +109,14 @@ func (sm *sprinklerMonitor) start(ctx context.Context) error { })) } + // Use custom server address if configured, otherwise default + serverAddr := client.DefaultServerAddress + if sm.serverAddress != "" { + serverAddr = sm.serverAddress + } + config := client.Config{ - ServerURL: "wss://" + client.DefaultServerAddress + "/ws", + ServerURL: "wss://" + serverAddr + "/ws", Token: sm.token, Organization: "*", // Monitor all orgs EventTypes: []string{"pull_request"}, @@ -441,14 +450,24 @@ func validateUserAction(data *turn.CheckResponse, user, repo string, n int, cach // handleNewPR triggers a refresh for PRs not in our lists and returns true if handled. func (sm *sprinklerMonitor) handleNewPR(ctx context.Context, url, repo string, n int, act *turn.Action) bool { sm.app.mu.RLock() - inIncoming := findPRInList(sm.app.incoming, url) - inOutgoing := false - if !inIncoming { - inOutgoing = findPRInList(sm.app.outgoing, url) + found := false + for i := range sm.app.incoming { + if sm.app.incoming[i].URL == url { + found = true + break + } + } + if !found { + for i := range sm.app.outgoing { + if sm.app.outgoing[i].URL == url { + found = true + break + } + } } sm.app.mu.RUnlock() - if !inIncoming && !inOutgoing { + if !found { slog.Info("[SPRINKLER] New PR detected, triggering refresh", "repo", repo, "number", n, @@ -460,16 +479,6 @@ func (sm *sprinklerMonitor) handleNewPR(ctx context.Context, url, repo string, n return false } -// findPRInList searches for a PR URL in the given list. -func findPRInList(prs []PR, url string) bool { - for i := range prs { - if prs[i].URL == url { - return true - } - } - return false -} - // isAlreadyTrackedAsBlocked checks if the PR is already tracked as blocked. func (sm *sprinklerMonitor) isAlreadyTrackedAsBlocked(url, repo string, n int) bool { sm.app.mu.RLock()