Filter upstream PR rows that already match main

Author: codex

internal/remote/dolthub.go

@@ -1040,12 +1040,63 @@ func (d *DoltHubProvider) ListPendingWantedIDs(upstreamOrg, db string) (map[stri
 	wg.Wait()
 	close(diffCh)
 
-	ids := make(map[string][]PendingWantedState)
+	branchResults := make([]pendingBranchRows, 0, len(branchPRs))
+	compareIDs := make(map[string]struct{})
 	for result := range diffCh {
 		if result.err != nil {
 			return nil, result.err
 		}
+		branchResults = append(branchResults, result)
+		if result.source.branch == "main" {
+			continue
+		}
+		for _, row := range result.rows {
+			if row.wantedID != "" {
+				compareIDs[row.wantedID] = struct{}{}
+			}
+		}
+	}
+
+	if len(compareIDs) > 0 && upstreamItems == nil {
+		wantedIDs := make([]string, 0, len(compareIDs))
+		for wantedID := range compareIDs {
+			wantedIDs = append(wantedIDs, wantedID)
+		}
+		slices.Sort(wantedIDs)
+		quotedIDs := make([]string, 0, len(wantedIDs))
+		for _, wantedID := range wantedIDs {
+			quotedIDs = append(quotedIDs, fmt.Sprintf("'%s'", strings.ReplaceAll(wantedID, "'", "''")))
+		}
+		upstreamItems = make(map[string]wantedItem, len(wantedIDs))
+		upstreamCompareQuery := fmt.Sprintf(
+			"SELECT id, status, COALESCE(claimed_by, '') as claimed_by FROM wanted WHERE id IN (%s)",
+			strings.Join(quotedIDs, ","),
+		)
+		upstreamURL := fmt.Sprintf("%s/%s/%s/main?q=%s",
+			dolthubAPIBase, upstreamOrg, db, url.QueryEscape(upstreamCompareQuery))
+		body, err := d.dolthubGet(upstreamURL)
+		if err == nil {
+			var qr queryResponse
+			if json.Unmarshal(body, &qr) == nil {
+				for _, row := range qr.Rows {
+					upstreamItems[row["id"]] = wantedItem{
+						status:    row["status"],
+						claimedBy: row["claimed_by"],
+					}
+				}
+			}
+		}
+	}
+
+	ids := make(map[string][]PendingWantedState)
+	for _, result := range branchResults {
 		for _, row := range result.rows {
+			if result.source.branch != "main" {
+				if upstream, exists := upstreamItems[row.wantedID]; exists &&
+					upstream.status == row.status && upstream.claimedBy == row.claimedBy {
+					continue
+				}
+			}
 			for _, pr := range branchPRs[result.source] {
 				branchURL := fmt.Sprintf("%s/%s/%s/data/%s",
 					dolthubRepoBase, result.source.owner, db, url.PathEscape(pr.fromBranch))

internal/remote/dolthub_test.go

@@ -1862,6 +1862,84 @@ func TestDoltHubProvider_ListPendingWantedIDs_NonStandardBranch(t *testing.T) {
 	}
 }
 
+func TestDoltHubProvider_ListPendingWantedIDs_FiltersRowsMatchingUpstreamMain(t *testing.T) {
+	mux := http.NewServeMux()
+	mux.HandleFunc("/org/db/pulls", func(w http.ResponseWriter, r *http.Request) {
+		if strings.Contains(r.URL.Path, "/pulls/") {
+			return
+		}
+		w.Header().Set("Content-Type", "application/json")
+		_ = json.NewEncoder(w).Encode(map[string]any{
+			"pulls": []map[string]any{{"pull_id": "1", "state": "open"}},
+		})
+	})
+	mux.HandleFunc("/org/db/pulls/1", func(w http.ResponseWriter, _ *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		_ = json.NewEncoder(w).Encode(map[string]any{
+			"from_branch":       "wl/bob/w-real",
+			"from_branch_owner": "bob-fork",
+			"author":            "bob",
+		})
+	})
+	mux.HandleFunc("/org/db/main", func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		q := r.URL.Query().Get("q")
+		if !strings.Contains(q, "'w-open'") || !strings.Contains(q, "'w-same-claim'") || !strings.Contains(q, "'w-real'") {
+			t.Fatalf("upstream comparison query missing wanted IDs: %s", q)
+		}
+		_ = json.NewEncoder(w).Encode(map[string]any{
+			"rows": []map[string]string{
+				{"id": "w-open", "status": "open", "claimed_by": ""},
+				{"id": "w-same-claim", "status": "claimed", "claimed_by": "alice"},
+				{"id": "w-real", "status": "open", "claimed_by": ""},
+			},
+		})
+	})
+	mux.HandleFunc("/bob-fork/db/", func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "application/json")
+		if strings.Contains(r.URL.Query().Get("q"), "FROM completions") {
+			_ = json.NewEncoder(w).Encode(map[string]any{"rows": []map[string]string{}})
+			return
+		}
+		_ = json.NewEncoder(w).Encode(map[string]any{
+			"rows": []map[string]string{
+				{"id": "w-open", "status": "open", "claimed_by": "", "diff_type": "added"},
+				{"id": "w-same-claim", "status": "claimed", "claimed_by": "alice", "diff_type": "modified"},
+				{"id": "w-real", "status": "claimed", "claimed_by": "bob", "diff_type": "modified"},
+			},
+		})
+	})
+
+	server := httptest.NewServer(mux)
+	defer server.Close()
+	oldAPIBase, oldRepoBase := dolthubAPIBase, dolthubRepoBase
+	dolthubAPIBase = server.URL
+	dolthubRepoBase = server.URL + "/repositories"
+	defer func() {
+		dolthubAPIBase = oldAPIBase
+		dolthubRepoBase = oldRepoBase
+	}()
+
+	provider := NewDoltHubProvider("token")
+	ids, err := provider.ListPendingWantedIDs("org", "db")
+	if err != nil {
+		t.Fatalf("ListPendingWantedIDs() error: %v", err)
+	}
+
+	if len(ids) != 1 {
+		t.Fatalf("expected 1 pending ID after upstream filtering, got %d: %+v", len(ids), ids)
+	}
+	if _, ok := ids["w-open"]; ok {
+		t.Fatalf("w-open should be filtered as identical to upstream: %+v", ids["w-open"])
+	}
+	if _, ok := ids["w-same-claim"]; ok {
+		t.Fatalf("w-same-claim should be filtered as identical to upstream: %+v", ids["w-same-claim"])
+	}
+	if pending := ids["w-real"]; len(pending) != 1 || pending[0].RigHandle != "bob" || pending[0].Status != "claimed" {
+		t.Fatalf("w-real pending = %+v, want only the real branch-owned change", pending)
+	}
+}
+
 func TestDoltHubProvider_ListPendingWantedIDs_PRFromMain(t *testing.T) {
 	mux := http.NewServeMux()
 	mux.HandleFunc("/org/db/pulls", func(w http.ResponseWriter, r *http.Request) {