diff --git a/go/internal/forge/github.go b/go/internal/forge/github.go index e02886e9e..fede6eb5b 100644 --- a/go/internal/forge/github.go +++ b/go/internal/forge/github.go @@ -143,6 +143,12 @@ type ghIssue struct { PullRequest *json.RawMessage `json:"pull_request"` } +// jsonNull is the literal a *json.RawMessage holds when GitHub sends an explicit +// `"pull_request": null` (a non-nil RawMessage wrapping the four bytes), as +// opposed to omitting the key for a plain issue. The interleaved-PR guards test +// against it so a null marker never drops a real issue. +const jsonNull = "null" + // ghError is the wire shape of a GitHub error body (the message field feeds // StatusError.Message). type ghError struct { @@ -223,7 +229,7 @@ func (g *GitHub) ListIssuesPage(ctx context.Context, repo string, f IssueFilter, // OMITS the pull_request key for a plain issue, but a *json.RawMessage // unmarshals an explicit "pull_request": null to a non-nil // RawMessage("null") — guard that so a null never drops a real issue. - if raw := r.PullRequest; raw != nil && len(*raw) > 0 && string(*raw) != "null" { + if raw := r.PullRequest; raw != nil && len(*raw) > 0 && string(*raw) != jsonNull { continue } issues = append(issues, r.toIssue()) diff --git a/go/internal/forge/notify_reader.go b/go/internal/forge/notify_reader.go index 51a925bf7..8cc785401 100644 --- a/go/internal/forge/notify_reader.go +++ b/go/internal/forge/notify_reader.go @@ -23,6 +23,7 @@ import ( "io" "net/http" "strconv" + "time" compassv1internal "github.com/RigelBuild/compass/go/internal/gen/compass/v1" ) @@ -238,7 +239,7 @@ func (g *GitHub) ListNewArtifacts(ctx context.Context, repo string, kind compass // Drop the PR rows GitHub interleaves into /issues (a *json.Raw // unmarshals an explicit "pull_request": null to a non-nil // RawMessage("null") — guard that, mirroring ListIssuesPage). - if raw := r.PullRequest; raw != nil && len(*raw) > 0 && string(*raw) != "null" { + if raw := r.PullRequest; raw != nil && len(*raw) > 0 && string(*raw) != jsonNull { return Issue{}, false } return r.toIssue(), true @@ -297,6 +298,69 @@ func ghWalkNewArtifacts[R any](ctx context.Context, g *GitHub, base string, sinc return ConditionalResult[[]Issue]{V: out, ETag: pageETag}, nil } +// ListUpdatedIssues walks /repos/{repo}/issues?state=all&sort=updated& +// direction=desc newest-updated-first (page 1 conditioned on etag; a 304 => +// NotModified), collecting issue rows (PR rows dropped by the pull_request +// marker, mirroring ListNewArtifacts) until a page's oldest updated_at is +// strictly < since, or no rel="next" remains. Rows with updated_at == since are +// RE-included: GitHub's updated_at is second-granularity, so a <= stop would +// permanently exclude an issue updated in the same second as the stored +// watermark after the sweep read it; the duplicates are free by coordinate +// idempotency. A zero since walks ALL pages (cold start). It returns page 1's +// ETag to re-store. This is the updated-order sibling of ListNewArtifacts +// (created-order, number-keyed): the reconcile/backfill read cannot see updates +// to existing issues via the created-order walk. +func (g *GitHub) ListUpdatedIssues(ctx context.Context, repo string, since time.Time, etag string) (ConditionalResult[[]Issue], error) { + base := g.apiBase() + "/repos/" + repo + "/issues?state=all&sort=updated&direction=desc" + var out []Issue + pageETag := "" + for page := 1; ; page++ { + u := base + "&per_page=" + strconv.Itoa(perPage) + "&page=" + strconv.Itoa(page) + sendETag := "" + if page == 1 { + sendETag = etag + } + var rows []ghIssue + notMod, e, hasNext, err := g.getJSONCond(ctx, u, sendETag, &rows) + if err != nil { + return ConditionalResult[[]Issue]{}, fmt.Errorf("forge: github list updated issues %q: %w", repo, err) + } + if page == 1 { + if notMod { + return ConditionalResult[[]Issue]{NotModified: true}, nil + } + pageETag = e + } + reachedOld := false + for _, r := range rows { + iss := r.toIssue() + if !iss.UpdatedAt.IsZero() && iss.UpdatedAt.Before(since) { + // Newest-updated-first: strictly older than the watermark, so this + // and everything after it is old. A row == since is NOT Before it, + // so it is re-included (second-granularity dedup safety). A row + // whose updated_at failed to parse (zero time) is NOT a stop + // signal — treating it as one would let a single malformed row + // truncate the whole sweep persistently; skip it and keep walking. + reachedOld = true + continue + } + if iss.UpdatedAt.IsZero() { + continue + } + // Drop the PR rows GitHub interleaves into /issues (mirroring + // ListNewArtifacts / ListIssuesPage's pull_request-marker guard). + if raw := r.PullRequest; raw != nil && len(*raw) > 0 && string(*raw) != jsonNull { + continue + } + out = append(out, iss) + } + if reachedOld || !hasNext { + break + } + } + return ConditionalResult[[]Issue]{V: out, ETag: pageETag}, nil +} + // --- Linear arm -------------------------------------------------------------- // Compile-time proof the Linear client satisfies the conditional-read surface. diff --git a/go/internal/forge/notify_reader_test.go b/go/internal/forge/notify_reader_test.go index b40c86d8d..ddbfdde32 100644 --- a/go/internal/forge/notify_reader_test.go +++ b/go/internal/forge/notify_reader_test.go @@ -12,6 +12,7 @@ import ( "errors" "net/http" "testing" + "time" compassv1internal "github.com/RigelBuild/compass/go/internal/gen/compass/v1" ) @@ -274,6 +275,179 @@ func TestListNewArtifactsPage1_304(t *testing.T) { } } +// --- GitHub: ListUpdatedIssues contract points ------------------------------- + +// TestListUpdatedIssuesStopsStrictlyBelowSince: a multi-page updated-order walk +// stops when a page's oldest updated_at is strictly < since. A row whose +// updated_at == since is RE-included (second-granularity dedup safety — a <= +// stop would permanently drop a same-second issue). +func TestListUpdatedIssuesStopsStrictlyBelowSince(t *testing.T) { + since := time.Date(2026, 8, 1, 12, 0, 0, 0, time.UTC) + // Page 1 all >= since (newest-updated-first); its oldest (== since) is NOT + // below since, so the walk continues to page 2. + page1 := `[ + {"number":50,"state":"open","html_url":"u50","updated_at":"2026-08-01T12:00:02Z"}, + {"number":49,"state":"open","html_url":"u49","updated_at":"2026-08-01T12:00:00Z"} + ]` + // Page 2's first row == since (re-included), its second is strictly below + // (dropped, and stops the walk). + page2 := `[ + {"number":48,"state":"open","html_url":"u48","updated_at":"2026-08-01T12:00:00Z"}, + {"number":40,"state":"open","html_url":"u40","updated_at":"2026-07-31T23:59:59Z"} + ]` + rt := &scriptedRoundTripper{responses: []scriptedResponse{ + {status: 200, body: page1, headers: map[string]string{"ETag": `"n1"`, "Link": `; rel="next"`}}, + {status: 200, body: page2, headers: map[string]string{"Link": `; rel="next"`}}, + }} + g := newTestGitHub(rt, &fakeTokenSource{token: "t"}) + + res, err := g.ListUpdatedIssues(context.Background(), "org/repo", since, "") + if err != nil { + t.Fatalf("ListUpdatedIssues: %v", err) + } + // 50,49 (page 1) + 48 (== since, re-included) = 3; 40 is strictly below and + // stops the walk before any page 3 is requested. + if len(res.V) != 3 { + t.Fatalf("kept %d (%+v), want 3 (50,49,48; 48 re-included at == since, 40 stops the walk)", len(res.V), res.V) + } + if res.V[2].Number != 48 { + t.Errorf("V[2].Number = %d, want 48 (the == since row is re-included)", res.V[2].Number) + } + if rt.calls != 2 { + t.Errorf("calls = %d, want 2 (the walk stopped once a page's oldest was strictly < since)", rt.calls) + } + if res.ETag != `"n1"` { + t.Errorf("ETag = %q, want page 1's %q (re-store)", res.ETag, `"n1"`) + } +} + +// TestListUpdatedIssuesMalformedRowDoesNotTruncate: a row whose updated_at fails +// to parse (zero time) must NOT be treated as the strictly-below-since stop +// signal — otherwise one malformed row truncates the whole sweep persistently. +// It is skipped (no output), and the walk continues to the valid rows behind it. +func TestListUpdatedIssuesMalformedRowDoesNotTruncate(t *testing.T) { + since := time.Date(2026, 8, 1, 12, 0, 0, 0, time.UTC) + // A malformed updated_at sits BETWEEN two valid >= since rows on page 1. If + // the zero time were a stop signal it would drop #48 (a real, fresh issue). + page1 := `[ + {"number":50,"state":"open","html_url":"u50","updated_at":"2026-08-01T12:00:05Z"}, + {"number":49,"state":"open","html_url":"u49","updated_at":"not-a-timestamp"}, + {"number":48,"state":"open","html_url":"u48","updated_at":"2026-08-01T12:00:03Z"} + ]` + rt := &scriptedRoundTripper{responses: []scriptedResponse{ + {status: 200, body: page1, headers: map[string]string{"ETag": `"n1"`}}, + }} + g := newTestGitHub(rt, &fakeTokenSource{token: "t"}) + + res, err := g.ListUpdatedIssues(context.Background(), "org/repo", since, "") + if err != nil { + t.Fatalf("ListUpdatedIssues: %v", err) + } + // 50 and 48 are collected; the malformed 49 is skipped, not a stop signal. + if len(res.V) != 2 { + t.Fatalf("kept %d (%+v), want 2 (50,48; the malformed 49 is skipped without truncating)", len(res.V), res.V) + } + if res.V[0].Number != 50 || res.V[1].Number != 48 { + t.Errorf("kept %d,%d, want 50,48 (the row behind the malformed one still collected)", res.V[0].Number, res.V[1].Number) + } +} + +// TestListUpdatedIssuesPage1_304: a 304 on page 1 short-circuits to NotModified +// without walking (nothing updated since the last sweep). +func TestListUpdatedIssuesPage1_304(t *testing.T) { + rt := &scriptedRoundTripper{responses: []scriptedResponse{ + {status: http.StatusNotModified, headers: map[string]string{"ETag": `"u0"`}}, + }} + g := newTestGitHub(rt, &fakeTokenSource{token: "t"}) + + res, err := g.ListUpdatedIssues(context.Background(), "org/repo", time.Time{}, `"u0"`) + if err != nil { + t.Fatalf("ListUpdatedIssues: %v", err) + } + if !res.NotModified || rt.calls != 1 { + t.Errorf("NotModified=%v calls=%d, want true/1", res.NotModified, rt.calls) + } +} + +// TestListUpdatedIssuesZeroSinceWalksAll: a zero since (cold start) walks every +// page to the last (no rel="next"), collecting all issue rows. +func TestListUpdatedIssuesZeroSinceWalksAll(t *testing.T) { + page1 := `[{"number":50,"state":"open","html_url":"u50","updated_at":"2026-08-01T12:00:02Z"}]` + page2 := `[{"number":10,"state":"closed","html_url":"u10","updated_at":"2020-01-01T00:00:00Z"}]` + rt := &scriptedRoundTripper{responses: []scriptedResponse{ + {status: 200, body: page1, headers: map[string]string{"ETag": `"n1"`, "Link": `; rel="next"`}}, + {status: 200, body: page2}, + }} + g := newTestGitHub(rt, &fakeTokenSource{token: "t"}) + + res, err := g.ListUpdatedIssues(context.Background(), "org/repo", time.Time{}, "") + if err != nil { + t.Fatalf("ListUpdatedIssues: %v", err) + } + // A zero since is never strictly greater than any updated_at, so nothing + // stops the walk short — both pages collected, walk ends at no rel="next". + if len(res.V) != 2 { + t.Fatalf("kept %d (%+v), want 2 (zero since walks to the last page)", len(res.V), res.V) + } + if rt.calls != 2 { + t.Errorf("calls = %d, want 2 (walk ended only at no rel=\"next\")", rt.calls) + } +} + +// TestListUpdatedIssuesFiltersPRs: /repos/{repo}/issues interleaves PR rows +// (pull_request marker); the updated-order walk drops them, keeping only +// issue-shaped rows, on the updated-order endpoint. +func TestListUpdatedIssuesFiltersPRs(t *testing.T) { + body := `[ + {"number":45,"state":"open","html_url":"u45","updated_at":"2026-08-01T12:00:02Z","pull_request":{"url":"pr"}}, + {"number":44,"state":"open","html_url":"u44","updated_at":"2026-08-01T12:00:01Z"} + ]` + rt := &scriptedRoundTripper{responses: []scriptedResponse{ + {status: 200, body: body, headers: map[string]string{"ETag": `"n1"`}}, + }} + g := newTestGitHub(rt, &fakeTokenSource{token: "t"}) + + res, err := g.ListUpdatedIssues(context.Background(), "org/repo", time.Time{}, "") + if err != nil { + t.Fatalf("ListUpdatedIssues: %v", err) + } + if len(res.V) != 1 || res.V[0].Number != 44 { + t.Fatalf("kept %+v, want only issue #44 (PR #45 filtered)", res.V) + } + if got := rt.requests[0].URL.Path; got != "/repos/org/repo/issues" { + t.Errorf("path = %q, want the issues endpoint", got) + } + if q := rt.requests[0].URL.Query(); q.Get("sort") != "updated" || q.Get("direction") != "desc" { + t.Errorf("query sort=%q direction=%q, want updated/desc", q.Get("sort"), q.Get("direction")) + } +} + +// TestListUpdatedIssuesBudgetErrorPropagates: an armed budget gate fails the +// page-1 read fast with ErrBudgetExhausted, propagated to the caller. +func TestListUpdatedIssuesBudgetErrorPropagates(t *testing.T) { + base := time.Date(2026, 8, 9, 12, 0, 0, 0, time.UTC) + // Page 1 arms the gate (remaining==0); page 2's read then fails fast. + page1 := `[{"number":50,"state":"open","html_url":"u50","updated_at":"2026-08-01T12:00:02Z"}]` + rt := &scriptedRoundTripper{responses: []scriptedResponse{ + {status: 200, body: page1, headers: map[string]string{ + "ETag": `"n1"`, + "Link": `; rel="next"`, + "x-ratelimit-remaining": "0", + "x-ratelimit-reset": "9999999999", + }}, + }} + g := newTestGitHub(rt, &fakeTokenSource{token: "t"}) + g.now = func() time.Time { return base } + + _, err := g.ListUpdatedIssues(context.Background(), "org/repo", time.Time{}, "") + if !errors.Is(err, ErrBudgetExhausted) { + t.Fatalf("err = %v, want ErrBudgetExhausted (gate armed on page 1 fails page 2 fast)", err) + } + if rt.calls != 1 { + t.Errorf("calls = %d, want 1 (page 2 fails fast without a request)", rt.calls) + } +} + // --- Linear arms ------------------------------------------------------------- // TestLinearReaderNoETags: a Linear issue read returns a 200-equivalent with an