diff --git a/docs/self-host.md b/docs/self-host.md index 5e39f93fa..83341b474 100644 --- a/docs/self-host.md +++ b/docs/self-host.md @@ -321,7 +321,7 @@ overwritten by accident. ### Forge secrets -The forge integration adds up to six more secrets. Each row below is a *name* +The forge integration adds up to seven more secrets. Each row below is a *name* you choose (via the flag or its environment variable), whose *value* you then set in the provider under the `SERVER_`-prefixed spelling. Compass declares a name only when the feature that needs it is configured, so you only provision @@ -335,9 +335,11 @@ the rows for the lanes you run. | `--forge-linear-client-id` | Linear OAuth client id | Set, together with the client secret | | `--forge-linear-client-secret` | Linear OAuth client secret | Set, together with the client id | | `--forge-linear-webhook-secret` | Linear webhook signing secret | Set | +| `--forge-trunk-token-secret` | Trunk API token | Merge-queue polling is configured with `--forge-merge-queues` | Each flag also reads an environment variable when the flag is unset: -`$COMPASS_FORGE_APP_KEY_SECRET` and so on, one per row. +`$COMPASS_FORGE_APP_KEY_SECRET` and so on, one per row. The Trunk token name +uses `$COMPASS_FORGE_TRUNK_TOKEN_SECRET`. Set every name you intend to use. The two Linear rows have built-in default names in the code, but those apply only when compass looks a value *up* -- diff --git a/go/cmd/compass-server/main.go b/go/cmd/compass-server/main.go index cd1b71770..b5fc723b6 100644 --- a/go/cmd/compass-server/main.go +++ b/go/cmd/compass-server/main.go @@ -488,6 +488,8 @@ type forgeFlags struct { linearClientID *string linearClientSecret *string linearWebhook *string + mergeQueues *string + trunkTokenSecret *string scopeEnforcementDisabled *string scopeGrants *string } @@ -558,6 +560,12 @@ func registerForgeFlags(fs *flag.FlagSet) forgeFlags { "Comma-separated user scope grants seeded at boot, each "+ "account:provider:host:repo with provider github or linear and repo * for a "+ "whole host. Insert only. Defaults to $COMPASS_FORGE_SCOPE_GRANTS."), + mergeQueues: fs.String("forge-merge-queues", "", + "Comma-separated owner/name=trunk:branch entries for merge-queue polling. "+ + "Defaults to $COMPASS_FORGE_MERGE_QUEUES."), + trunkTokenSecret: fs.String("forge-trunk-token-secret", "", + "Declared server-only secret NAME for the Trunk API token. Required when "+ + "merge queues are configured; defaults to $COMPASS_FORGE_TRUNK_TOKEN_SECRET."), } } @@ -580,6 +588,8 @@ func (f forgeFlags) resolve() (server.ForgeConfig, error) { linearWebhook: firstNonEmpty(*f.linearWebhook, os.Getenv("COMPASS_FORGE_LINEAR_WEBHOOK_SECRET")), scopeEnforcementDisabled: firstNonEmpty(*f.scopeEnforcementDisabled, os.Getenv("COMPASS_FORGE_DISABLE_SCOPE_ENFORCEMENT")), scopeGrants: firstNonEmpty(*f.scopeGrants, os.Getenv("COMPASS_FORGE_SCOPE_GRANTS")), + mergeQueues: firstNonEmpty(*f.mergeQueues, os.Getenv("COMPASS_FORGE_MERGE_QUEUES")), + trunkTokenSecret: firstNonEmpty(*f.trunkTokenSecret, os.Getenv("COMPASS_FORGE_TRUNK_TOKEN_SECRET")), }) } @@ -600,6 +610,8 @@ type forgeInputs struct { linearClientID string linearClientSecret string linearWebhook string + mergeQueues string + trunkTokenSecret string scopeEnforcementDisabled string scopeGrants string } @@ -616,6 +628,10 @@ func resolveForge(in forgeInputs) (server.ForgeConfig, error) { if err != nil { return server.ForgeConfig{}, err } + queues, err := parseMergeQueues(in.mergeQueues) + if err != nil { + return server.ForgeConfig{}, err + } id, err := parseForgeInt(in.appID, "--forge-app-id") if err != nil { return server.ForgeConfig{}, err @@ -646,6 +662,8 @@ func resolveForge(in forgeInputs) (server.ForgeConfig, error) { Host: in.host, ForgeCAPath: in.ca, SeedRepos: seed, + TrunkTokenSecretName: in.trunkTokenSecret, + MergeQueues: queues, LinearClientIDSecretName: in.linearClientID, LinearClientSecretName: in.linearClientSecret, LinearWebhookSecretName: in.linearWebhook, @@ -702,6 +720,39 @@ func parseForgeRepos(repos string) ([]string, error) { return out, nil } +func parseMergeQueues(raw string) ([]server.MergeQueueConfig, error) { + if strings.TrimSpace(raw) == "" { + return nil, nil + } + configs := make([]server.MergeQueueConfig, 0, strings.Count(raw, ",")+1) + seen := make(map[string]struct{}, cap(configs)) + for item := range strings.SplitSeq(raw, ",") { + item = strings.TrimSpace(item) + if item == "" { + continue + } + repo, spec, ok := strings.Cut(item, "=") + if !ok || strings.Contains(spec, "=") { + return nil, fmt.Errorf("invalid --forge-merge-queues entry %q: want owner/name=trunk:branch", item) + } + owner, name, repoOK := strings.Cut(repo, "/") + provider, branch, providerOK := strings.Cut(spec, ":") + if !repoOK || owner == "" || name == "" || strings.Contains(name, "/") || !providerOK || provider == "" || branch == "" || strings.Contains(branch, ":") { + return nil, fmt.Errorf("invalid --forge-merge-queues entry %q: want owner/name=trunk:branch", item) + } + if provider != "trunk" { + return nil, fmt.Errorf("unsupported merge queue provider %q", provider) + } + repo = strings.ToLower(repo) + if _, exists := seen[repo]; exists { + return nil, fmt.Errorf("duplicate --forge-merge-queues repo %q", repo) + } + seen[repo] = struct{}{} + configs = append(configs, server.MergeQueueConfig{Repo: repo, Provider: provider, Branch: branch}) + } + return configs, nil +} + // parseForgeScopeGrants parses account:provider:host:repo entries. The store // normalizes repo case, so this only validates shape and the provider name. func parseForgeScopeGrants(v string) ([]store.ForgeScope, error) { diff --git a/go/cmd/compass-server/main_forge_test.go b/go/cmd/compass-server/main_forge_test.go index 402ef4c17..3410a3a5a 100644 --- a/go/cmd/compass-server/main_forge_test.go +++ b/go/cmd/compass-server/main_forge_test.go @@ -47,6 +47,8 @@ func TestResolveForgeMapping(t *testing.T) { linearClientID: "LINEAR_CID", linearClientSecret: "LINEAR_CSECRET", linearWebhook: "LINEAR_WEBHOOK_SECRET", + mergeQueues: "Owner/Repo=trunk:main", + trunkTokenSecret: "TRUNK_TOKEN", }) if err != nil { t.Fatalf("resolveForge: %v", err) @@ -60,6 +62,8 @@ func TestResolveForgeMapping(t *testing.T) { LinearClientIDSecretName: "LINEAR_CID", LinearClientSecretName: "LINEAR_CSECRET", LinearWebhookSecretName: "LINEAR_WEBHOOK_SECRET", + TrunkTokenSecretName: "TRUNK_TOKEN", + MergeQueues: []server.MergeQueueConfig{{Repo: "owner/repo", Provider: "trunk", Branch: "main"}}, App: server.ForgeAppConfig{ AppID: 12345, InstallationID: 678, diff --git a/go/cmd/compass-server/merge_queue_test.go b/go/cmd/compass-server/merge_queue_test.go new file mode 100644 index 000000000..c02ef447c --- /dev/null +++ b/go/cmd/compass-server/merge_queue_test.go @@ -0,0 +1,39 @@ +//go:build unix + +package main + +import ( + "reflect" + "testing" + + "github.com/RigelBuild/compass/go/server" +) + +func TestParseMergeQueues(t *testing.T) { + configs, err := parseMergeQueues("Owner/Repo=trunk:main, other/project=trunk:release/next") + if err != nil { + t.Fatalf("parseMergeQueues: %v", err) + } + want := []server.MergeQueueConfig{ + {Repo: "owner/repo", Provider: "trunk", Branch: "main"}, + {Repo: "other/project", Provider: "trunk", Branch: "release/next"}, + } + if !reflect.DeepEqual(configs, want) { + t.Fatalf("parseMergeQueues = %#v, want %#v", configs, want) + } +} + +func TestParseMergeQueuesRejectsInvalidEntries(t *testing.T) { + for _, raw := range []string{ + "owner=trunk:main", + "owner/repo=trunk:", + "owner/repo=github:main", + "owner/repo=trunk:main,owner/repo=trunk:other", + } { + t.Run(raw, func(t *testing.T) { + if _, err := parseMergeQueues(raw); err == nil { + t.Fatalf("parseMergeQueues(%q) returned no error", raw) + } + }) + } +} diff --git a/go/internal/board/issue_prs.go b/go/internal/board/issue_prs.go index ef1c9bc88..896a85cc8 100644 --- a/go/internal/board/issue_prs.go +++ b/go/internal/board/issue_prs.go @@ -6,11 +6,13 @@ import ( "context" "fmt" "slices" + "time" compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1" "github.com/RigelBuild/compass/go/internal/ingest" "github.com/RigelBuild/compass/go/internal/store" "google.golang.org/protobuf/encoding/protojson" + "google.golang.org/protobuf/proto" ) // PublishPullRequestUpdate stores a hydrated PR and its closing references, then @@ -26,12 +28,51 @@ func (p *IssueProjection) PublishPullRequestUpdate(ctx context.Context, in inges if err != nil { return fmt.Errorf("board: upsert pull request: %w", err) } + if in.MergeQueueObserved { + entry := []byte("null") + if in.PR.GetMergeQueue() != nil { + entry, err = protojson.Marshal(in.PR.GetMergeQueue()) + if err != nil { + return fmt.Errorf("board: marshal merge queue entry: %w", err) + } + } + changed, err := p.store.SetPullRequestMergeQueue(ctx, row.Coord, entry, in.MergeQueueObservedAt) + if err != nil { + return err + } + if changed { + affected = append(affected, row.Coord) + } + } return p.republishPrs(ctx, affected) } +// PublishPullRequestMergeQueue republishes issues after a changed stored queue entry. +func (p *IssueProjection) PublishPullRequestMergeQueue( + ctx context.Context, + coord store.ForgeCoord, + _ *compassv1.MergeQueueEntry, +) error { + links, err := p.store.PullRequestLinks(ctx, coord) + if err != nil { + return fmt.Errorf("board: load pull request links for queue update: %w", err) + } + return p.republishPrs(ctx, links) +} + +// UpdatePullRequestMergeQueue persists an observed entry and reports whether it changed. +func (p *IssueProjection) UpdatePullRequestMergeQueue(ctx context.Context, coord store.ForgeCoord, entry []byte, observedAt time.Time) (bool, error) { + return p.store.SetPullRequestMergeQueue(ctx, coord, entry, observedAt) +} + // pullRequestRow maps an ingested PR to the store row plus its closing refs on the PR's forge. +// The queue entry is stripped: only SetPullRequestMergeQueue writes it, under its observation guard. func pullRequestRow(in ingest.IngestedPullRequest) (store.PullRequestRow, []store.ForgeCoord, error) { pr := in.PR + if pr.GetMergeQueue() != nil { + pr = proto.CloneOf(pr) + pr.MergeQueue = nil + } body, err := protojson.Marshal(pr) if err != nil { return store.PullRequestRow{}, nil, fmt.Errorf("board: marshal pull request: %w", err) diff --git a/go/internal/board/issue_prs_pgtest_test.go b/go/internal/board/issue_prs_pgtest_test.go index ae0f4be13..8a984bc5f 100644 --- a/go/internal/board/issue_prs_pgtest_test.go +++ b/go/internal/board/issue_prs_pgtest_test.go @@ -7,6 +7,8 @@ package board import ( "context" + "encoding/json" + "errors" "slices" "testing" "time" @@ -39,6 +41,78 @@ func ingestedPR(number uint32, age, updated time.Duration, closes ...uint64) ing } } +func TestPullRequestHydrateLookupErrorPreservesStoredMergeQueue(t *testing.T) { + ctx := context.Background() + p, _, st := newIssueBoard(t) + mustPublishIssue(t, p, 1) + initial := ingestedPR(10, 0, time.Hour, 1) + initial.PR.MergeQueue = &compassv1.MergeQueueEntry{Provider: "trunk", State: "testing"} + initial.MergeQueueObserved = true + initial.MergeQueueObservedAt = prT0.Add(90 * time.Minute) + mustPublishPR(t, p, initial) + ref := &compassv1.ForgeRef{Provider: compassv1.ForgeProvider_FORGE_PROVIDER_GITHUB, Host: "github.com"} + hydrator := ingest.NewPullRequestHydrator( + errorQueuePullReader{}, p, ref, + map[string]forge.MergeQueue{"trunk": errorMergeQueue{}}, errorQueueConfig{}, nil, + ) + if err := hydrator.Hydrate(ctx, "RigelBuild/compass", 10); err != nil { + t.Fatalf("Hydrate: %v", err) + } + issue := store.ForgeCoord{Provider: store.ForgeProviderGitHub, Host: "github.com", Repo: "RigelBuild/compass", Number: 1} + rows, err := st.PullRequestsForIssues(ctx, []store.ForgeCoord{issue}) + if err != nil { + t.Fatalf("PullRequestsForIssues: %v", err) + } + prs := rows[issue.Normalized()] + if len(prs) != 1 { + t.Fatalf("linked pull requests = %d, want 1", len(prs)) + } + var got struct { + MergeQueue *struct { + State string `json:"state"` + } `json:"mergeQueue"` + } + if err := json.Unmarshal(prs[0].PR, &got); err != nil { + t.Fatalf("decode PR JSON: %v", err) + } + if got.MergeQueue == nil || got.MergeQueue.State != "testing" { + t.Fatalf("hydrated PR mergeQueue = %v, want testing", got.MergeQueue) + } +} + +type errorQueuePullReader struct{} + +func (errorQueuePullReader) GetPullRequest(_ context.Context, repo string, number uint64) (forge.PullRequest, error) { + return forge.PullRequest{ + Number: number, State: "open", CreatedAt: prT0, UpdatedAt: prT0.Add(2 * time.Hour), + ClosingRefs: []forge.IssueRef{{Repo: repo, Number: 1}}, + }, nil +} + +func (errorQueuePullReader) ListOpenPullRequests(context.Context, string) ([]forge.UpdatedPull, error) { + return nil, nil +} + +type errorQueueConfig struct{} + +func (errorQueueConfig) MergeQueueFor(context.Context, string) (string, string, error) { + return "trunk", "main", nil +} + +func (errorQueueConfig) MergeQueueRepos(context.Context) ([]string, error) { + return []string{"RigelBuild/compass"}, nil +} + +func (errorQueueConfig) MergeQueueHost() string { return "github.com" } + +type errorMergeQueue struct{} + +func (errorMergeQueue) Name() string { return "trunk" } + +func (errorMergeQueue) Entries(context.Context, string, string, []uint64) (map[uint64]forge.QueueEntry, error) { + return nil, errors.New("offline") +} + func prNumbers(iss *compassv1.Issue) []uint32 { out := make([]uint32, 0, len(iss.GetPrs())) for _, pr := range iss.GetPrs() { diff --git a/go/internal/forge/mergequeue.go b/go/internal/forge/mergequeue.go new file mode 100644 index 000000000..4ee6b984c --- /dev/null +++ b/go/internal/forge/mergequeue.go @@ -0,0 +1,31 @@ +package forge + +import ( + "context" + "time" +) + +// QueueState is the normalized merge-queue state for a pull request. +type QueueState int + +const ( + QueueUnknown QueueState = iota + QueueQueued + QueueTesting + QueueFailed + QueueRemoved + QueueMerged +) + +// QueueEntry is one observed merge-queue entry. +type QueueEntry struct { + Number uint64 + State QueueState + ChangedAt time.Time +} + +// MergeQueue reads pull-request state from a configured merge queue. +type MergeQueue interface { + Name() string + Entries(ctx context.Context, repo, targetBranch string, numbers []uint64) (map[uint64]QueueEntry, error) +} diff --git a/go/internal/forge/trunk.go b/go/internal/forge/trunk.go new file mode 100644 index 000000000..a1bff46ae --- /dev/null +++ b/go/internal/forge/trunk.go @@ -0,0 +1,186 @@ +package forge + +import ( + "bytes" + "context" + "encoding/json" + "errors" + "fmt" + "maps" + "net/http" + "net/url" + "slices" + "strings" + "time" +) + +const trunkBatchSize = 50 + +// TrunkConfig configures read access to the Trunk Merge Queue API. +type TrunkConfig struct { + Token func(ctx context.Context) ([]byte, error) + Host string + HTTP *http.Client +} + +// Trunk reads submitted pull request state from Trunk Merge Queue. +type Trunk struct { + cfg TrunkConfig +} + +// NewTrunk returns a Trunk merge-queue provider. +func NewTrunk(cfg TrunkConfig) *Trunk { + if cfg.Host == "" { + cfg.Host = "github.com" + } + if cfg.HTTP == nil { + cfg.HTTP = &http.Client{Timeout: 30 * time.Second} + } else { + copyClient := *cfg.HTTP + cfg.HTTP = ©Client + } + cfg.HTTP.CheckRedirect = noTrunkRedirects + return &Trunk{cfg: cfg} +} + +func noTrunkRedirects(*http.Request, []*http.Request) error { + return http.ErrUseLastResponse +} + +// Name returns the provider key used by merge-queue configuration. +func (*Trunk) Name() string { return "trunk" } + +// Entries reads queue state for one repository in batches of at most 50 PRs. +func (t *Trunk) Entries(ctx context.Context, repo, targetBranch string, numbers []uint64) (map[uint64]QueueEntry, error) { + out := make(map[uint64]QueueEntry, len(numbers)) + if len(numbers) == 0 { + return out, nil + } + for start := 0; start < len(numbers); start += trunkBatchSize { + end := min(start+trunkBatchSize, len(numbers)) + batch, err := t.entriesAt(ctx, trunkAPIBase, repo, targetBranch, numbers[start:end]) + if err != nil { + return nil, err + } + maps.Copy(out, batch) + } + return out, nil +} + +const trunkAPIBase = "https://api.trunk.io" + +func (t *Trunk) entriesAt(ctx context.Context, base, repo, targetBranch string, numbers []uint64) (map[uint64]QueueEntry, error) { + if len(numbers) > trunkBatchSize { + return nil, fmt.Errorf("forge: Trunk batch has %d pull requests, maximum is %d", len(numbers), trunkBatchSize) + } + owner, name, ok := strings.Cut(repo, "/") + if !ok || owner == "" || name == "" || strings.Contains(name, "/") { + return nil, fmt.Errorf("forge: invalid Trunk repo %q: want owner/name", repo) + } + if t.cfg.Token == nil { + return nil, errors.New("forge: Trunk token resolver is required") + } + token, err := t.cfg.Token(ctx) + if err != nil { + return nil, fmt.Errorf("forge: resolve Trunk token: %w", err) + } + prs := make([]trunkPRRequest, 0, len(numbers)) + for _, number := range numbers { + prs = append(prs, trunkPRRequest{Number: number}) + } + body, err := json.Marshal(trunkRequest{ + Repo: trunkRepo{Host: t.cfg.Host, Owner: owner, Name: name}, + TargetBranch: targetBranch, + PRs: prs, + }) + if err != nil { + return nil, fmt.Errorf("forge: marshal Trunk request: %w", err) + } + endpoint, err := url.JoinPath(base, "/v1/getSubmittedPullRequests") + if err != nil { + return nil, fmt.Errorf("forge: build Trunk endpoint: %w", err) + } + req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(body)) + if err != nil { + return nil, fmt.Errorf("forge: build Trunk request: %w", err) + } + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-Api-Token", string(token)) + resp, err := t.cfg.HTTP.Do(req) + if err != nil { + return nil, fmt.Errorf("forge: request Trunk merge queue: %w", err) + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices { + return nil, fmt.Errorf("forge: Trunk merge queue returned %s", resp.Status) + } + var result trunkResponse + if err := json.NewDecoder(resp.Body).Decode(&result); err != nil { + return nil, fmt.Errorf("forge: decode Trunk response: %w", err) + } + out := make(map[uint64]QueueEntry, len(result.PullRequests)) + for _, item := range result.PullRequests { + state, ok := trunkQueueState(item.State) + if !ok { + return nil, fmt.Errorf("forge: unknown Trunk queue state %q", item.State) + } + if !slices.Contains(numbers, item.PRNumber) { + return nil, fmt.Errorf("forge: Trunk returned unrequested pull request #%d", item.PRNumber) + } + if !item.IsCurrentlySubmittedToQueue && (state == QueueQueued || state == QueueTesting) { + state = QueueRemoved + } + + changedAt, err := time.Parse(time.RFC3339Nano, item.StateChangedAt) + if err != nil { + return nil, fmt.Errorf("forge: parse Trunk stateChangedAt for pull request #%d: %w", item.PRNumber, err) + } + out[item.PRNumber] = QueueEntry{Number: item.PRNumber, State: state, ChangedAt: changedAt} + } + return out, nil +} + +func trunkQueueState(state string) (QueueState, bool) { + switch state { + case "not_ready", "pending": + return QueueQueued, true + case "testing", "tests_passed": + return QueueTesting, true + case "failed", "pending_failure": + return QueueFailed, true + case "cancelled": + return QueueRemoved, true + case "merged": + return QueueMerged, true + default: + return 0, false + } +} + +type trunkRequest struct { + Repo trunkRepo `json:"repo"` + TargetBranch string `json:"targetBranch"` + PRs []trunkPRRequest `json:"prs"` +} + +type trunkRepo struct { + Host string `json:"host"` + Owner string `json:"owner"` + Name string `json:"name"` +} + +type trunkPRRequest struct { + Number uint64 `json:"number"` +} + +type trunkResponse struct { + PullRequests []trunkPullRequest `json:"pullRequests"` + NotFound []trunkPRRequest `json:"notFound"` +} + +type trunkPullRequest struct { + PRNumber uint64 `json:"prNumber"` + State string `json:"state"` + StateChangedAt string `json:"stateChangedAt"` + IsCurrentlySubmittedToQueue bool `json:"isCurrentlySubmittedToQueue"` +} diff --git a/go/internal/forge/trunk_test.go b/go/internal/forge/trunk_test.go new file mode 100644 index 000000000..0a99aeb70 --- /dev/null +++ b/go/internal/forge/trunk_test.go @@ -0,0 +1,190 @@ +package forge + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "reflect" + "strings" + "sync/atomic" + "testing" + "time" +) + +func TestTrunkEntriesMapsStatesAndRequest(t *testing.T) { + changedAt := "2026-10-11T12:34:56.123Z" + var request struct { + Repo struct { + Host string `json:"host"` + Owner string `json:"owner"` + Name string `json:"name"` + } `json:"repo"` + TargetBranch string `json:"targetBranch"` + PRs []struct { + Number uint64 `json:"number"` + } `json:"prs"` + } + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost || r.URL.Path != "/v1/getSubmittedPullRequests" { + t.Errorf("request = %s %s, want POST /v1/getSubmittedPullRequests", r.Method, r.URL.Path) + } + if got := r.Header.Get("X-Api-Token"); got != "fixture-token" { + t.Errorf("X-Api-Token = %q, want fixture-token", got) + + } + if err := json.NewDecoder(r.Body).Decode(&request); err != nil { + t.Errorf("decode request: %v", err) + } + w.Header().Set("Content-Type", "application/json") + _, _ = w.Write([]byte(`{"pullRequests":[` + + `{"prNumber":1,"state":"not_ready","stateChangedAt":"` + changedAt + `","isCurrentlySubmittedToQueue":true},` + + `{"prNumber":2,"state":"pending","stateChangedAt":"` + changedAt + `","isCurrentlySubmittedToQueue":true},` + + `{"prNumber":3,"state":"testing","stateChangedAt":"` + changedAt + `","isCurrentlySubmittedToQueue":true},` + + `{"prNumber":4,"state":"tests_passed","stateChangedAt":"` + changedAt + `","isCurrentlySubmittedToQueue":true},` + + `{"prNumber":5,"state":"failed","stateChangedAt":"` + changedAt + `","isCurrentlySubmittedToQueue":false},` + + `{"prNumber":6,"state":"pending_failure","stateChangedAt":"` + changedAt + `","isCurrentlySubmittedToQueue":false},` + + `{"prNumber":7,"state":"cancelled","stateChangedAt":"` + changedAt + `","isCurrentlySubmittedToQueue":false},` + + `{"prNumber":8,"state":"merged","stateChangedAt":"` + changedAt + `","isCurrentlySubmittedToQueue":false},` + + `{"prNumber":9,"state":"pending","stateChangedAt":"` + changedAt + `","isCurrentlySubmittedToQueue":false}` + + `],"notFound":[{"number":10}]}`)) + })) + queue := NewTrunk(TrunkConfig{ + Token: func(context.Context) ([]byte, error) { return []byte("fixture-token"), nil }, + Host: "github.com", + HTTP: &http.Client{Transport: roundTripTo(server)}, + }) + got, err := queue.Entries(context.Background(), "owner/repo", "main", []uint64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}) + if err != nil { + t.Fatalf("Entries: %v", err) + } + wantRepo := struct { + Host string `json:"host"` + Owner string `json:"owner"` + Name string `json:"name"` + }{Host: "github.com", Owner: "owner", Name: "repo"} + if request.Repo != wantRepo || request.TargetBranch != "main" || len(request.PRs) != 10 { + t.Fatalf("request repo/branch/PRs = %+v/%q/%d, want %+v/main/10", request.Repo, request.TargetBranch, len(request.PRs), wantRepo) + } + wantAt, err := time.Parse(time.RFC3339Nano, changedAt) + if err != nil { + t.Fatal(err) + } + want := map[uint64]QueueEntry{ + 1: {Number: 1, State: QueueQueued, ChangedAt: wantAt}, + 2: {Number: 2, State: QueueQueued, ChangedAt: wantAt}, + 3: {Number: 3, State: QueueTesting, ChangedAt: wantAt}, + 4: {Number: 4, State: QueueTesting, ChangedAt: wantAt}, + 5: {Number: 5, State: QueueFailed, ChangedAt: wantAt}, + 6: {Number: 6, State: QueueFailed, ChangedAt: wantAt}, + 7: {Number: 7, State: QueueRemoved, ChangedAt: wantAt}, + 8: {Number: 8, State: QueueMerged, ChangedAt: wantAt}, + 9: {Number: 9, State: QueueRemoved, ChangedAt: wantAt}, + } + if !reflect.DeepEqual(got, want) { + t.Fatalf("Entries = %#v, want %#v", got, want) + } + if _, ok := got[10]; ok { + t.Fatalf("notFound PR #10 unexpectedly returned: %#v", got[10]) + } +} + +func TestTrunkEntriesMapsCancelledSubmissionToRemoved(t *testing.T) { + changedAt := "2026-10-11T12:34:56Z" + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = w.Write([]byte(`{"pullRequests":[{"prNumber":7,"state":"pending","stateChangedAt":"` + changedAt + `","isCurrentlySubmittedToQueue":false}],"notFound":[]}`)) + })) + defer server.Close() + queue := NewTrunk(TrunkConfig{ + Token: func(context.Context) ([]byte, error) { return []byte("fixture-token"), nil }, + HTTP: &http.Client{Transport: roundTripTo(server)}, + }) + got, err := queue.entriesAt(context.Background(), server.URL, "owner/repo", "main", []uint64{7}) + if err != nil { + t.Fatalf("Entries: %v", err) + } + if entry, ok := got[7]; !ok || entry.State != QueueRemoved { + t.Fatalf("Entries = %#v, want PR #7 removed", got) + } +} + +func TestTrunkEntriesBatchesAtFifty(t *testing.T) { + var sizes []int + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var request trunkRequest + if err := json.NewDecoder(r.Body).Decode(&request); err != nil { + t.Errorf("decode request: %v", err) + w.WriteHeader(http.StatusBadRequest) + return + } + sizes = append(sizes, len(request.PRs)) + response := trunkResponse{PullRequests: make([]trunkPullRequest, 0, len(request.PRs))} + for _, pr := range request.PRs { + response.PullRequests = append(response.PullRequests, trunkPullRequest{ + PRNumber: pr.Number, State: "pending", StateChangedAt: "2026-10-11T12:34:56Z", IsCurrentlySubmittedToQueue: true, + }) + } + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(response); err != nil { + t.Errorf("encode response: %v", err) + } + })) + defer server.Close() + queue := NewTrunk(TrunkConfig{ + Token: func(context.Context) ([]byte, error) { return []byte("fixture-token"), nil }, + HTTP: &http.Client{Transport: roundTripTo(server)}, + }) + numbers := make([]uint64, 51) + for i := range numbers { + numbers[i] = uint64(i + 1) + } + entries, err := queue.Entries(context.Background(), "owner/repo", "main", numbers) + if err != nil { + t.Fatalf("Entries: %v", err) + } + if !reflect.DeepEqual(sizes, []int{50, 1}) { + t.Fatalf("batch sizes = %v, want [50 1]", sizes) + } + if len(entries) != len(numbers) { + t.Fatalf("Entries returned %d PRs, want %d", len(entries), len(numbers)) + } +} + +func TestTrunkDoesNotFollowRedirectsOrExposeResponseBody(t *testing.T) { + var redirected atomic.Int32 + target := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + redirected.Add(1) + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{"pullRequests":[],"notFound":[]}`)) + })) + defer target.Close() + origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Location", target.URL) + w.WriteHeader(http.StatusFound) + _, _ = w.Write([]byte("response-body-must-not-appear")) + })) + defer origin.Close() + queue := NewTrunk(TrunkConfig{Token: func(context.Context) ([]byte, error) { return []byte("fixture-token"), nil }}) + _, err := queue.entriesAt(context.Background(), origin.URL, "owner/repo", "main", []uint64{1}) + if err == nil || !strings.Contains(err.Error(), "302") { + t.Fatalf("Entries error = %v, want HTTP 302", err) + } + if strings.Contains(err.Error(), "response-body-must-not-appear") { + t.Fatalf("Entries error exposes response body: %v", err) + } + if redirected.Load() != 0 { + t.Fatalf("redirect target received %d requests, want none", redirected.Load()) + } +} + +type roundTripFunc func(*http.Request) (*http.Response, error) + +func (f roundTripFunc) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) } + +func roundTripTo(server *httptest.Server) http.RoundTripper { + return roundTripFunc(func(r *http.Request) (*http.Response, error) { + r.URL.Scheme = "http" + r.URL.Host = server.Listener.Addr().String() + return http.DefaultTransport.RoundTrip(r) + }) +} diff --git a/go/internal/ingest/board_reconcile.go b/go/internal/ingest/board_reconcile.go index c19559e5b..375914d81 100644 --- a/go/internal/ingest/board_reconcile.go +++ b/go/internal/ingest/board_reconcile.go @@ -58,15 +58,10 @@ type updatedLister interface { // BoardReconcileConfig configures the board reconciliation sweep. type BoardReconcileConfig struct { - // Backstop is the ticker cadence between sweeps; <= 0 uses defaultBackstop. Backstop time.Duration - // Pace is the inter-repo delay within a sweep (anti-burst); 0 uses - // defaultPace, negative disables pacing. - Pace time.Duration - // Log is the sweep logger; nil uses slog.Default(). - Log *slog.Logger - // Pulls hydrates listed PR rows and runs the backfill; nil skips PR rows. - Pulls *PullRequestHydrator + Pace time.Duration + Log *slog.Logger + Pulls *PullRequestHydrator } // BoardReconciler drives the backstop sweep over the enabled-repo set, listing diff --git a/go/internal/ingest/mergequeue.go b/go/internal/ingest/mergequeue.go new file mode 100644 index 000000000..19d80034e --- /dev/null +++ b/go/internal/ingest/mergequeue.go @@ -0,0 +1,143 @@ +package ingest + +import ( + "context" + "fmt" + "log/slog" + "time" + + compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1" + "github.com/RigelBuild/compass/go/internal/forge" + "github.com/RigelBuild/compass/go/internal/store" + "google.golang.org/protobuf/types/known/timestamppb" +) + +const mergeQueueBatchSize = 50 + +// QueueConfigs resolves the queue config for each configured repository and +// lists the repositories the sweep refreshes. +type QueueConfigs interface { + MergeQueueFor(ctx context.Context, repo string) (provider, branch string, err error) + MergeQueueRepos(ctx context.Context) ([]string, error) + MergeQueueHost() string +} + +// MergeQueueStore lists open pull requests for one repository. +type MergeQueueStore interface { + OpenPullRequests(ctx context.Context, repo string) ([]uint64, error) +} + +// MergeQueueSink stores a queue entry and republishes linked board issues when it changes. +type MergeQueueSink interface { + PublishPullRequestMergeQueue(ctx context.Context, coord store.ForgeCoord, entry *compassv1.MergeQueueEntry, observedAt time.Time) error +} + +// MergeQueueSweeper polls configured queue providers and publishes state changes. +type MergeQueueSweeper struct { + queues map[string]forge.MergeQueue + cfgs QueueConfigs + store MergeQueueStore + sink MergeQueueSink + every time.Duration + log *slog.Logger + now func() time.Time +} + +// NewMergeQueueSweeper builds the periodic merge-queue refresh worker. +func NewMergeQueueSweeper(queues map[string]forge.MergeQueue, cfgs QueueConfigs, st MergeQueueStore, sink MergeQueueSink, every time.Duration) *MergeQueueSweeper { + if every <= 0 { + every = 2 * time.Minute + } + return &MergeQueueSweeper{queues: queues, cfgs: cfgs, store: st, sink: sink, every: every, log: slog.Default(), now: time.Now} +} + +// Run sweeps once at startup, then at every cadence until ctx ends. +func (s *MergeQueueSweeper) Run(ctx context.Context) error { + s.sweep(ctx) + ticker := time.NewTicker(s.every) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return nil + case <-ticker.C: + s.sweep(ctx) + } + } +} + +func (s *MergeQueueSweeper) sweep(ctx context.Context) { + if ctx.Err() != nil { + return + } + repos, err := s.cfgs.MergeQueueRepos(ctx) + if err != nil { + s.log.ErrorContext(ctx, "merge queue sweep: list configured repos", "error", err) + return + } + for _, repo := range repos { + if ctx.Err() != nil { + return + } + provider, branch, err := s.cfgs.MergeQueueFor(ctx, repo) + if err != nil { + s.log.ErrorContext(ctx, "merge queue sweep: resolve repo config", "repo", repo, "error", err) + continue + } + queue := s.queues[provider] + if queue == nil { + s.log.ErrorContext(ctx, "merge queue sweep: provider unavailable", "repo", repo, "provider", provider) + continue + } + numbers, err := s.store.OpenPullRequests(ctx, repo) + if err != nil { + s.log.ErrorContext(ctx, "merge queue sweep: list open pull requests", "repo", repo, "error", err) + continue + } + for start := 0; start < len(numbers); start += mergeQueueBatchSize { + end := min(start+mergeQueueBatchSize, len(numbers)) + observedAt := s.now() + entries, err := queue.Entries(ctx, repo, branch, numbers[start:end]) + if err != nil { + s.log.ErrorContext(ctx, "merge queue sweep: read provider entries", "repo", repo, "error", err) + break + } + if err := s.publishChanged(ctx, s.cfgs.MergeQueueHost(), repo, provider, numbers[start:end], entries, observedAt); err != nil { + s.log.ErrorContext(ctx, "merge queue sweep: publish entries", "repo", repo, "error", err) + break + } + } + } +} + +func (s *MergeQueueSweeper) publishChanged(ctx context.Context, host, repo, provider string, numbers []uint64, entries map[uint64]forge.QueueEntry, observedAt time.Time) error { + for _, number := range numbers { + entry, ok := entries[number] + var wire *compassv1.MergeQueueEntry + if ok { + wire = &compassv1.MergeQueueEntry{Provider: provider, State: queueStateName(entry.State), ChangedAt: timestamppb.New(entry.ChangedAt)} + } + coord := store.ForgeCoord{Provider: store.ForgeProviderGitHub, Host: host, Repo: repo, Number: number} + if err := s.sink.PublishPullRequestMergeQueue(ctx, coord, wire, observedAt); err != nil { + return fmt.Errorf("publish pull request #%d queue state: %w", number, err) + } + } + return nil +} + +func queueStateName(state forge.QueueState) string { + switch state { + case forge.QueueQueued: + return "queued" + case forge.QueueTesting: + return "testing" + case forge.QueueFailed: + return "failed" + case forge.QueueRemoved: + return "removed" + case forge.QueueMerged: + return "merged" + default: + return "" + } +} diff --git a/go/internal/ingest/mergequeue_test.go b/go/internal/ingest/mergequeue_test.go new file mode 100644 index 000000000..f076692fc --- /dev/null +++ b/go/internal/ingest/mergequeue_test.go @@ -0,0 +1,205 @@ +package ingest + +import ( + "context" + "errors" + "log/slog" + "slices" + "testing" + "time" + + compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1" + "github.com/RigelBuild/compass/go/internal/forge" + "github.com/RigelBuild/compass/go/internal/store" +) + +func TestPullRequestHydratorMergeQueue(t *testing.T) { + queueEntry := forge.QueueEntry{Number: 7, State: forge.QueueTesting, ChangedAt: time.Date(2026, 10, 11, 12, 0, 0, 0, time.UTC)} + for _, tt := range []struct { + name string + configs QueueConfigs + queues map[string]forge.MergeQueue + queue *fixedQueue + wantCalls int + wantState string + wantProvider string + wantObserved bool + wantLogError bool + }{ + {name: "configured queue attaches entry", configs: fixedQueueConfig{provider: "trunk", branch: "main", repos: []string{"owner/repo"}}, queue: &fixedQueue{entry: queueEntry}, wantCalls: 1, wantState: "testing", wantProvider: "trunk", wantObserved: true}, + {name: "configured queue failure keeps hydration green", configs: fixedQueueConfig{provider: "trunk", branch: "main", repos: []string{"owner/repo"}}, queue: &fixedQueue{err: errors.New("offline")}, wantCalls: 1, wantLogError: true}, + {name: "unconfigured repo makes no provider call", queue: &fixedQueue{entry: queueEntry}, wantCalls: 0}, + } { + t.Run(tt.name, func(t *testing.T) { + if tt.queue != nil { + tt.queues = map[string]forge.MergeQueue{"trunk": tt.queue} + } + pulls := &fakePulls{} + sink := &recordingPRSink{} + log := &queueLogHandler{} + hydrator := NewPullRequestHydrator(pulls, sink, testForgeRef(), tt.queues, tt.configs, slog.New(log)) + if err := hydrator.Hydrate(context.Background(), "owner/repo", 7); err != nil { + t.Fatalf("Hydrate: %v", err) + } + if tt.queue != nil && tt.queue.calls != tt.wantCalls { + t.Fatalf("queue calls = %d, want %d", tt.queue.calls, tt.wantCalls) + } + if len(sink.got) != 1 { + t.Fatalf("sink calls = %d, want 1", len(sink.got)) + } + in := sink.got[0] + entry := in.PR.GetMergeQueue() + if tt.wantProvider == "" { + if entry != nil { + t.Fatalf("merge queue = %v, want nil", entry) + } + } else if entry.GetProvider() != tt.wantProvider || entry.GetState() != tt.wantState || entry.GetChangedAt().AsTime() != queueEntry.ChangedAt { + t.Fatalf("merge queue = %v, want provider=%s state=%s changed_at=%v", entry, tt.wantProvider, tt.wantState, queueEntry.ChangedAt) + } + if in.MergeQueueObserved != tt.wantObserved { + t.Fatalf("queue observed = %v, want %v", in.MergeQueueObserved, tt.wantObserved) + } + if log.errors != tt.wantLogError { + t.Fatalf("queue error logged = %v, want %v", log.errors, tt.wantLogError) + } + }) + } +} + +type fixedQueueConfig struct { + provider string + branch string + repos []string +} + +func (c fixedQueueConfig) MergeQueueFor(context.Context, string) (string, string, error) { + return c.provider, c.branch, nil +} + +func (c fixedQueueConfig) MergeQueueRepos(context.Context) ([]string, error) { + return c.repos, nil +} + +func (c fixedQueueConfig) MergeQueueHost() string { return "github.com" } + +type fixedQueue struct { + entry forge.QueueEntry + err error + calls int +} + +func (q *fixedQueue) Name() string { return "trunk" } + +func (q *fixedQueue) Entries(_ context.Context, _, _ string, numbers []uint64) (map[uint64]forge.QueueEntry, error) { + q.calls++ + if q.err != nil { + return nil, q.err + } + if len(numbers) == 0 { + return map[uint64]forge.QueueEntry{}, nil + } + return map[uint64]forge.QueueEntry{q.entry.Number: q.entry}, nil +} + +type queueLogHandler struct{ errors bool } + +func (h *queueLogHandler) Enabled(context.Context, slog.Level) bool { return true } +func (h *queueLogHandler) Handle(_ context.Context, r slog.Record) error { + if r.Level >= slog.LevelError { + h.errors = true + } + return nil +} +func (h *queueLogHandler) WithAttrs([]slog.Attr) slog.Handler { return h } +func (h *queueLogHandler) WithGroup(string) slog.Handler { return h } + +func TestMergeQueueSweeperBatchesAtFifty(t *testing.T) { + numbers := make([]uint64, 51) + for i := range numbers { + numbers[i] = uint64(i + 1) + } + queue := &recordingBatchQueue{} + store := &fixedQueuePRStore{numbers: numbers} + sink := &recordingQueueSink{} + configs := fixedQueueConfig{provider: "trunk", branch: "main", repos: []string{"owner/repo"}} + clock := &batchClock{at: time.Date(2026, 10, 11, 12, 0, 0, 0, time.UTC)} + sweeper := NewMergeQueueSweeper(map[string]forge.MergeQueue{"trunk": queue}, configs, store, sink, time.Hour) + sweeper.now = clock.Now + sweeper.sweep(context.Background()) + if !slices.Equal(queue.batchSizes, []int{50, 1}) { + t.Fatalf("provider batch sizes = %v, want [50 1]", queue.batchSizes) + } + if len(sink.entries) != 51 { + t.Fatalf("published entries = %d, want 51", len(sink.entries)) + } + if len(sink.observedAt) != 51 || !sink.observedAt[0].Equal(sink.observedAt[49]) || sink.observedAt[49].Equal(sink.observedAt[50]) { + t.Fatalf("observedAt = %v; want one timestamp per batch", sink.observedAt) + } +} + +type batchClock struct { + at time.Time + calls int +} + +func (c *batchClock) Now() time.Time { + c.calls++ + return c.at.Add(time.Duration(c.calls) * time.Second) +} + +type recordingBatchQueue struct{ batchSizes []int } + +func (*recordingBatchQueue) Name() string { return "trunk" } + +func (q *recordingBatchQueue) Entries(_ context.Context, _, _ string, numbers []uint64) (map[uint64]forge.QueueEntry, error) { + q.batchSizes = append(q.batchSizes, len(numbers)) + return map[uint64]forge.QueueEntry{}, nil +} + +func TestMergeQueueSweeperPublishesConfiguredEntries(t *testing.T) { + entry := forge.QueueEntry{Number: 7, State: forge.QueueTesting, ChangedAt: time.Date(2026, 10, 11, 12, 0, 0, 0, time.UTC)} + queue := &fixedQueue{entry: entry} + configs := fixedQueueConfig{provider: "trunk", branch: "main", repos: []string{"owner/repo"}} + store := &fixedQueuePRStore{numbers: []uint64{7}} + sink := &recordingQueueSink{} + sweeper := NewMergeQueueSweeper(map[string]forge.MergeQueue{"trunk": queue}, configs, store, sink, time.Hour) + sweeper.sweep(context.Background()) + if queue.calls != 1 || store.calls != 1 || len(sink.entries) != 1 { + t.Fatalf("calls = provider %d, store %d, sink %d; want 1 each", queue.calls, store.calls, len(sink.entries)) + } + got := sink.entries[0] + if got.GetProvider() != "trunk" || got.GetState() != "testing" || !got.GetChangedAt().AsTime().Equal(entry.ChangedAt) { + t.Fatalf("published entry = %v, want testing entry for PR #7", got) + } +} + +func TestMergeQueueSweeperClearsMissingEntries(t *testing.T) { + queue := &fixedQueue{entry: forge.QueueEntry{Number: 99}} + configs := fixedQueueConfig{provider: "trunk", branch: "main", repos: []string{"owner/repo"}} + sink := &recordingQueueSink{} + NewMergeQueueSweeper(map[string]forge.MergeQueue{"trunk": queue}, configs, &fixedQueuePRStore{numbers: []uint64{7}}, sink, time.Hour).sweep(context.Background()) + if len(sink.entries) != 1 || sink.entries[0] != nil { + t.Fatalf("missing queue entry published = %v, want nil removal", sink.entries) + } +} + +type fixedQueuePRStore struct { + numbers []uint64 + calls int +} + +func (s *fixedQueuePRStore) OpenPullRequests(context.Context, string) ([]uint64, error) { + s.calls++ + return s.numbers, nil +} + +type recordingQueueSink struct { + entries []*compassv1.MergeQueueEntry + observedAt []time.Time +} + +func (s *recordingQueueSink) PublishPullRequestMergeQueue(_ context.Context, _ store.ForgeCoord, entry *compassv1.MergeQueueEntry, observedAt time.Time) error { + s.entries = append(s.entries, entry) + s.observedAt = append(s.observedAt, observedAt) + return nil +} diff --git a/go/internal/ingest/pull_request_sink.go b/go/internal/ingest/pull_request_sink.go index a21d0094c..d9bc5fb73 100644 --- a/go/internal/ingest/pull_request_sink.go +++ b/go/internal/ingest/pull_request_sink.go @@ -3,18 +3,22 @@ package ingest import ( "context" "fmt" + "log/slog" "time" compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1" "github.com/RigelBuild/compass/go/internal/forge" + "google.golang.org/protobuf/types/known/timestamppb" ) -// IngestedPullRequest carries what the wire PullRequest lacks: forge times and closing refs. +// IngestedPullRequest carries forge times, closing refs, and queue observation state. type IngestedPullRequest struct { - PR *compassv1.PullRequest - CreatedAt time.Time - UpdatedAt time.Time - ClosingRefs []forge.IssueRef + PR *compassv1.PullRequest + CreatedAt time.Time + UpdatedAt time.Time + ClosingRefs []forge.IssueRef + MergeQueueObserved bool + MergeQueueObservedAt time.Time } // pullRequestReader is the PR read surface, satisfied by *forge.GitHub. @@ -31,15 +35,28 @@ type prSink interface { // PullRequestHydrator reads one PR, translates it the way the create path does, // and sinks it to the board. type PullRequestHydrator struct { - reader pullRequestReader - sink prSink - forgeRef *compassv1.ForgeRef + reader pullRequestReader + sink prSink + forgeRef *compassv1.ForgeRef + queues map[string]forge.MergeQueue + queueCfgs QueueConfigs + log *slog.Logger } // NewPullRequestHydrator returns a hydrator reading from r, stamping ref and -// sinking to s. -func NewPullRequestHydrator(r pullRequestReader, s prSink, ref *compassv1.ForgeRef) *PullRequestHydrator { - return &PullRequestHydrator{reader: r, sink: s, forgeRef: ref} +// sinking to s. Queue lookup is optional and never blocks a successful PR hydrate. +func NewPullRequestHydrator( + r pullRequestReader, + s prSink, + ref *compassv1.ForgeRef, + queues map[string]forge.MergeQueue, + queueCfgs QueueConfigs, + log *slog.Logger, +) *PullRequestHydrator { + if log == nil { + log = slog.Default() + } + return &PullRequestHydrator{reader: r, sink: s, forgeRef: ref, queues: queues, queueCfgs: queueCfgs, log: log} } // Hydrate reads PR number in repo and sinks it. Errors wrap the cause, so a @@ -57,13 +74,54 @@ func (h *PullRequestHydrator) Hydrate(ctx context.Context, repo string, number u pr := forge.TranslatePullRequest(raw, attr) pr.Forge = &compassv1.ForgeRef{Provider: h.forgeRef.GetProvider(), Host: h.forgeRef.GetHost()} pr.Repo = repo - in := IngestedPullRequest{PR: pr, CreatedAt: raw.CreatedAt, UpdatedAt: raw.UpdatedAt, ClosingRefs: raw.ClosingRefs} + observedAt := time.Time{} + queueObserved := false + if entry, observed, at := h.mergeQueueEntry(ctx, repo, number); observed { + pr.MergeQueue = entry + observedAt = at + queueObserved = true + } + in := IngestedPullRequest{ + PR: pr, CreatedAt: raw.CreatedAt, UpdatedAt: raw.UpdatedAt, ClosingRefs: raw.ClosingRefs, + MergeQueueObserved: queueObserved, MergeQueueObservedAt: observedAt, + } if err := h.sink.PublishPullRequestUpdate(ctx, in); err != nil { return fmt.Errorf("ingest: publish pull request #%d for %q: %w", number, repo, err) } return nil } +func (h *PullRequestHydrator) mergeQueueEntry(ctx context.Context, repo string, number uint64) (*compassv1.MergeQueueEntry, bool, time.Time) { + if h.queueCfgs == nil { + return nil, false, time.Time{} + } + provider, branch, err := h.queueCfgs.MergeQueueFor(ctx, repo) + if err != nil { + h.log.ErrorContext(ctx, "pull request hydrate: merge queue config", "repo", repo, "error", err) + return nil, false, time.Time{} + } + if provider == "" { + return nil, false, time.Time{} + } + queue := h.queues[provider] + if queue == nil { + h.log.ErrorContext(ctx, "pull request hydrate: merge queue provider unavailable", "repo", repo, "provider", provider) + return nil, false, time.Time{} + } + observedAt := time.Now() + entries, err := queue.Entries(ctx, repo, branch, []uint64{number}) + if err != nil { + h.log.ErrorContext(ctx, "pull request hydrate: merge queue lookup", "repo", repo, "number", number, "error", err) + return nil, false, time.Time{} + } + entry, ok := entries[number] + if !ok { + return nil, true, observedAt + } + wire := &compassv1.MergeQueueEntry{Provider: provider, State: queueStateName(entry.State), ChangedAt: timestamppb.New(entry.ChangedAt)} + return wire, true, observedAt +} + // listOpen lists the repo's open PRs for the backfill pass. func (h *PullRequestHydrator) listOpen(ctx context.Context, repo string) ([]forge.UpdatedPull, error) { return h.reader.ListOpenPullRequests(ctx, repo) diff --git a/go/internal/ingest/pull_request_sink_test.go b/go/internal/ingest/pull_request_sink_test.go index 9904ee061..cd8c39672 100644 --- a/go/internal/ingest/pull_request_sink_test.go +++ b/go/internal/ingest/pull_request_sink_test.go @@ -74,7 +74,7 @@ func newPRArm(t *testing.T, pulls *fakePulls, numbers PullNumberResolver) (*Boar func newPRArmFor(t *testing.T, pulls *fakePulls, numbers PullNumberResolver, targets TargetChecker) (*BoardWebhookArm, *recordingPRSink) { t.Helper() sink := &recordingPRSink{} - h := NewPullRequestHydrator(pulls, sink, testForgeRef()) + h := NewPullRequestHydrator(pulls, sink, testForgeRef(), nil, nil, nil) ing := NewIngester(forge.NewFakeProvider("gh"), &recordingSink{}, testForgeRef()) arm := NewBoardWebhookArm(&fakeHydrator{}, ing, targets, BoardArmConfig{QueueSize: 64, Pulls: h, PullNumbers: numbers}) return arm, sink @@ -224,7 +224,7 @@ func TestArmIssueAndPRSameNumberStayDistinct(t *testing.T) { func newPRReconciler(l updatedLister, st *fakeBoardStore, pulls *fakePulls) (*BoardReconciler, *recordingSink) { sink := &recordingSink{} - h := NewPullRequestHydrator(pulls, &recordingPRSink{}, testForgeRef()) + h := NewPullRequestHydrator(pulls, &recordingPRSink{}, testForgeRef(), nil, nil, nil) return NewBoardReconciler(l, NewIngester(nil, sink, testForgeRef()), st, BoardReconcileConfig{Pace: -1, Pulls: h}), sink } diff --git a/go/internal/store/db/forge_cursors.sql.go b/go/internal/store/db/forge_cursors.sql.go index f81be7f3d..7056e7c56 100644 --- a/go/internal/store/db/forge_cursors.sql.go +++ b/go/internal/store/db/forge_cursors.sql.go @@ -11,17 +11,35 @@ import ( "github.com/jackc/pgx/v5/pgtype" ) +const clearForgeRepoMergeQueueConfigs = `-- name: ClearForgeRepoMergeQueueConfigs :exec +UPDATE forge_repo_subscriptions + SET merge_queue = '', merge_queue_branch = '' + WHERE forge_provider = $1 AND forge_host = $2 +` + +type ClearForgeRepoMergeQueueConfigsParams struct { + ForgeProvider int16 + ForgeHost string +} + +func (q *Queries) ClearForgeRepoMergeQueueConfigs(ctx context.Context, arg ClearForgeRepoMergeQueueConfigsParams) error { + _, err := q.db.Exec(ctx, clearForgeRepoMergeQueueConfigs, arg.ForgeProvider, arg.ForgeHost) + return err +} + const ensureForgeRepoSubscription = `-- name: EnsureForgeRepoSubscription :exec -INSERT INTO forge_repo_subscriptions (forge_provider, forge_host, repo, enabled) -VALUES ($1, $2, $3, $4) +INSERT INTO forge_repo_subscriptions (forge_provider, forge_host, repo, enabled, merge_queue, merge_queue_branch) +VALUES ($1, $2, $3, $4, $5, $6) ON CONFLICT (tenant_id, forge_provider, forge_host, repo) DO NOTHING ` type EnsureForgeRepoSubscriptionParams struct { - ForgeProvider int16 - ForgeHost string - Repo string - Enabled bool + ForgeProvider int16 + ForgeHost string + Repo string + Enabled bool + MergeQueue string + MergeQueueBranch string } func (q *Queries) EnsureForgeRepoSubscription(ctx context.Context, arg EnsureForgeRepoSubscriptionParams) error { @@ -30,10 +48,36 @@ func (q *Queries) EnsureForgeRepoSubscription(ctx context.Context, arg EnsureFor arg.ForgeHost, arg.Repo, arg.Enabled, + arg.MergeQueue, + arg.MergeQueueBranch, ) return err } +const getForgeRepoMergeQueueConfig = `-- name: GetForgeRepoMergeQueueConfig :one +SELECT merge_queue, merge_queue_branch + FROM forge_repo_subscriptions + WHERE forge_provider = $1 AND forge_host = $2 AND repo = $3 +` + +type GetForgeRepoMergeQueueConfigParams struct { + ForgeProvider int16 + ForgeHost string + Repo string +} + +type GetForgeRepoMergeQueueConfigRow struct { + MergeQueue string + MergeQueueBranch string +} + +func (q *Queries) GetForgeRepoMergeQueueConfig(ctx context.Context, arg GetForgeRepoMergeQueueConfigParams) (GetForgeRepoMergeQueueConfigRow, error) { + row := q.db.QueryRow(ctx, getForgeRepoMergeQueueConfig, arg.ForgeProvider, arg.ForgeHost, arg.Repo) + var i GetForgeRepoMergeQueueConfigRow + err := row.Scan(&i.MergeQueue, &i.MergeQueueBranch) + return i, err +} + const isEnabledForgeRepo = `-- name: IsEnabledForgeRepo :one SELECT EXISTS ( SELECT 1 FROM forge_repo_subscriptions @@ -48,7 +92,7 @@ func (q *Queries) IsEnabledForgeRepo(ctx context.Context, repo string) (bool, er } const listEnabledForgeRepoSubscriptions = `-- name: ListEnabledForgeRepoSubscriptions :many -SELECT forge_provider, forge_host, repo, enabled +SELECT forge_provider, forge_host, repo, enabled, merge_queue, merge_queue_branch FROM forge_repo_subscriptions WHERE forge_provider = $1 AND forge_host = $2 AND enabled = TRUE ORDER BY repo ASC @@ -60,10 +104,12 @@ type ListEnabledForgeRepoSubscriptionsParams struct { } type ListEnabledForgeRepoSubscriptionsRow struct { - ForgeProvider int16 - ForgeHost string - Repo string - Enabled bool + ForgeProvider int16 + ForgeHost string + Repo string + Enabled bool + MergeQueue string + MergeQueueBranch string } func (q *Queries) ListEnabledForgeRepoSubscriptions(ctx context.Context, arg ListEnabledForgeRepoSubscriptionsParams) ([]ListEnabledForgeRepoSubscriptionsRow, error) { @@ -80,6 +126,8 @@ func (q *Queries) ListEnabledForgeRepoSubscriptions(ctx context.Context, arg Lis &i.ForgeHost, &i.Repo, &i.Enabled, + &i.MergeQueue, + &i.MergeQueueBranch, ); err != nil { return nil, err } @@ -118,6 +166,77 @@ func (q *Queries) ListEnabledForgeRepos(ctx context.Context) ([]string, error) { return items, nil } +const listForgeRepoMergeQueueConfigs = `-- name: ListForgeRepoMergeQueueConfigs :many +SELECT repo, merge_queue, merge_queue_branch + FROM forge_repo_subscriptions + WHERE forge_provider = $1 AND forge_host = $2 AND enabled AND merge_queue <> '' + ORDER BY repo ASC +` + +type ListForgeRepoMergeQueueConfigsParams struct { + ForgeProvider int16 + ForgeHost string +} + +type ListForgeRepoMergeQueueConfigsRow struct { + Repo string + MergeQueue string + MergeQueueBranch string +} + +func (q *Queries) ListForgeRepoMergeQueueConfigs(ctx context.Context, arg ListForgeRepoMergeQueueConfigsParams) ([]ListForgeRepoMergeQueueConfigsRow, error) { + rows, err := q.db.Query(ctx, listForgeRepoMergeQueueConfigs, arg.ForgeProvider, arg.ForgeHost) + if err != nil { + return nil, err + } + defer rows.Close() + var items []ListForgeRepoMergeQueueConfigsRow + for rows.Next() { + var i ListForgeRepoMergeQueueConfigsRow + if err := rows.Scan(&i.Repo, &i.MergeQueue, &i.MergeQueueBranch); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listOpenPullRequests = `-- name: ListOpenPullRequests :many +SELECT number + FROM pull_requests + WHERE forge_provider = $1 AND forge_host = $2 AND repo = $3 AND forge_state = 'open' + ORDER BY number ASC +` + +type ListOpenPullRequestsParams struct { + ForgeProvider int16 + ForgeHost string + Repo string +} + +func (q *Queries) ListOpenPullRequests(ctx context.Context, arg ListOpenPullRequestsParams) ([]int64, error) { + rows, err := q.db.Query(ctx, listOpenPullRequests, arg.ForgeProvider, arg.ForgeHost, arg.Repo) + if err != nil { + return nil, err + } + defer rows.Close() + var items []int64 + for rows.Next() { + var number int64 + if err := rows.Scan(&number); err != nil { + return nil, err + } + items = append(items, number) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const loadForgeRepoWatermark = `-- name: LoadForgeRepoWatermark :one SELECT swept_updated_at, list_etag @@ -193,6 +312,35 @@ func (q *Queries) MarkPRsBackfilled(ctx context.Context, arg MarkPRsBackfilledPa return result.RowsAffected(), nil } +const setForgeRepoMergeQueue = `-- name: SetForgeRepoMergeQueue :execrows +UPDATE forge_repo_subscriptions + SET merge_queue = $4, + merge_queue_branch = $5 + WHERE forge_provider = $1 AND forge_host = $2 AND repo = $3 +` + +type SetForgeRepoMergeQueueParams struct { + ForgeProvider int16 + ForgeHost string + Repo string + MergeQueue string + MergeQueueBranch string +} + +func (q *Queries) SetForgeRepoMergeQueue(ctx context.Context, arg SetForgeRepoMergeQueueParams) (int64, error) { + result, err := q.db.Exec(ctx, setForgeRepoMergeQueue, + arg.ForgeProvider, + arg.ForgeHost, + arg.Repo, + arg.MergeQueue, + arg.MergeQueueBranch, + ) + if err != nil { + return 0, err + } + return result.RowsAffected(), nil +} + const setForgeRepoSubscriptionEnabled = `-- name: SetForgeRepoSubscriptionEnabled :execrows UPDATE forge_repo_subscriptions SET enabled = $4 diff --git a/go/internal/store/db/models.go b/go/internal/store/db/models.go index e77bc49ff..d95972b2f 100644 --- a/go/internal/store/db/models.go +++ b/go/internal/store/db/models.go @@ -260,16 +260,18 @@ type ForgeAuthoredArtifact struct { } type ForgeRepoSubscription struct { - ForgeProvider int16 - ForgeHost string - Repo string - Enabled bool - SweptUpdatedAt pgtype.Timestamptz - ListEtag string - PrsBackfilledAt pgtype.Timestamptz - CreatedAt pgtype.Timestamptz - UpdatedAt pgtype.Timestamptz - TenantID string + ForgeProvider int16 + ForgeHost string + Repo string + MergeQueue string + MergeQueueBranch string + Enabled bool + SweptUpdatedAt pgtype.Timestamptz + ListEtag string + PrsBackfilledAt pgtype.Timestamptz + CreatedAt pgtype.Timestamptz + UpdatedAt pgtype.Timestamptz + TenantID string } type ForgeStateTransition struct { @@ -380,17 +382,18 @@ type OwedMention struct { } type PullRequest struct { - ForgeProvider int16 - ForgeHost string - Repo string - Number int64 - ForgeState string - ForgeCreatedAt pgtype.Timestamptz - ForgeUpdatedAt pgtype.Timestamptz - Pr []byte - CreatedAt pgtype.Timestamptz - UpdatedAt pgtype.Timestamptz - TenantID string + ForgeProvider int16 + ForgeHost string + Repo string + Number int64 + ForgeState string + ForgeCreatedAt pgtype.Timestamptz + ForgeUpdatedAt pgtype.Timestamptz + Pr []byte + MergeQueueObservedAt pgtype.Timestamptz + CreatedAt pgtype.Timestamptz + UpdatedAt pgtype.Timestamptz + TenantID string } type PullRequestIssueLink struct { diff --git a/go/internal/store/db/pull_requests.sql.go b/go/internal/store/db/pull_requests.sql.go index d2775ec37..e7dd5deef 100644 --- a/go/internal/store/db/pull_requests.sql.go +++ b/go/internal/store/db/pull_requests.sql.go @@ -250,6 +250,56 @@ func (q *Queries) PullRequestForgeUpdatedAt(ctx context.Context, arg PullRequest return forge_updated_at, err } +const pullRequestLinks = `-- name: PullRequestLinks :many +SELECT issue_forge_provider, issue_forge_host, issue_repo, issue_number + FROM pull_request_issue_links + WHERE pr_forge_provider = $1 AND pr_forge_host = $2 AND pr_repo = $3 AND pr_number = $4 +` + +type PullRequestLinksParams struct { + PrForgeProvider int16 + PrForgeHost string + PrRepo string + PrNumber int64 +} + +type PullRequestLinksRow struct { + IssueForgeProvider int16 + IssueForgeHost string + IssueRepo string + IssueNumber int64 +} + +func (q *Queries) PullRequestLinks(ctx context.Context, arg PullRequestLinksParams) ([]PullRequestLinksRow, error) { + rows, err := q.db.Query(ctx, pullRequestLinks, + arg.PrForgeProvider, + arg.PrForgeHost, + arg.PrRepo, + arg.PrNumber, + ) + if err != nil { + return nil, err + } + defer rows.Close() + var items []PullRequestLinksRow + for rows.Next() { + var i PullRequestLinksRow + if err := rows.Scan( + &i.IssueForgeProvider, + &i.IssueForgeHost, + &i.IssueRepo, + &i.IssueNumber, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const pullRequestsForIssues = `-- name: PullRequestsForIssues :many WITH want AS ( SELECT unnest($1::smallint[]) AS provider, @@ -344,6 +394,52 @@ func (q *Queries) PullRequestsForIssues(ctx context.Context, arg PullRequestsFor return items, nil } +const setPullRequestMergeQueue = `-- name: SetPullRequestMergeQueue :one +WITH current AS MATERIALIZED ( + SELECT tenant_id, forge_provider, forge_host, repo, number, + COALESCE(pr -> 'mergeQueue', 'null'::jsonb) IS DISTINCT FROM $5::jsonb AS changed + FROM pull_requests + WHERE pull_requests.forge_provider = $1 AND pull_requests.forge_host = $2 AND pull_requests.repo = $3 AND pull_requests.number = $4 + AND $6 >= COALESCE(pull_requests.merge_queue_observed_at, '-infinity'::timestamptz) + FOR UPDATE +), updated AS ( + UPDATE pull_requests AS p + SET pr = jsonb_set(p.pr, '{mergeQueue}', $5::jsonb, true), + merge_queue_observed_at = $6 + FROM current + WHERE p.tenant_id = current.tenant_id + AND p.forge_provider = current.forge_provider + AND p.forge_host = current.forge_host + AND p.repo = current.repo + AND p.number = current.number + RETURNING current.changed +) +SELECT changed FROM updated +` + +type SetPullRequestMergeQueueParams struct { + ForgeProvider int16 + ForgeHost string + Repo string + Number int64 + Column5 []byte + MergeQueueObservedAt pgtype.Timestamptz +} + +func (q *Queries) SetPullRequestMergeQueue(ctx context.Context, arg SetPullRequestMergeQueueParams) (bool, error) { + row := q.db.QueryRow(ctx, setPullRequestMergeQueue, + arg.ForgeProvider, + arg.ForgeHost, + arg.Repo, + arg.Number, + arg.Column5, + arg.MergeQueueObservedAt, + ) + var changed bool + err := row.Scan(&changed) + return changed, err +} + const upsertExplicitPullRequestLink = `-- name: UpsertExplicitPullRequestLink :exec INSERT INTO pull_request_issue_links (pr_forge_provider, pr_forge_host, pr_repo, pr_number, @@ -381,15 +477,19 @@ func (q *Queries) UpsertExplicitPullRequestLink(ctx context.Context, arg UpsertE const upsertPullRequestGuarded = `-- name: UpsertPullRequestGuarded :execrows INSERT INTO pull_requests - (forge_provider, forge_host, repo, number, forge_state, - forge_created_at, forge_updated_at, pr) + (forge_provider, forge_host, repo, number, forge_state, + forge_created_at, forge_updated_at, pr) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) ON CONFLICT (tenant_id, forge_provider, forge_host, repo, number) DO UPDATE - SET forge_state = EXCLUDED.forge_state, - forge_created_at = EXCLUDED.forge_created_at, - forge_updated_at = EXCLUDED.forge_updated_at, - pr = EXCLUDED.pr - WHERE EXCLUDED.forge_updated_at >= pull_requests.forge_updated_at + SET forge_state = EXCLUDED.forge_state, + forge_created_at = EXCLUDED.forge_created_at, + forge_updated_at = EXCLUDED.forge_updated_at, + pr = CASE + WHEN pull_requests.pr ? 'mergeQueue' THEN + jsonb_set(EXCLUDED.pr - 'mergeQueue', '{mergeQueue}', pull_requests.pr -> 'mergeQueue', true) + ELSE EXCLUDED.pr - 'mergeQueue' + END +WHERE EXCLUDED.forge_updated_at >= pull_requests.forge_updated_at ` type UpsertPullRequestGuardedParams struct { diff --git a/go/internal/store/db/querier.go b/go/internal/store/db/querier.go index 81fb04309..7e3306215 100644 --- a/go/internal/store/db/querier.go +++ b/go/internal/store/db/querier.go @@ -95,6 +95,7 @@ type Querier interface { ChannelsByNameForViewer(ctx context.Context, arg ChannelsByNameForViewerParams) ([]ChannelsByNameForViewerRow, error) // The unique key makes concurrent first-run attempts a single-winner claim. ClaimTourStart(ctx context.Context, arg ClaimTourStartParams) (string, error) + ClearForgeRepoMergeQueueConfigs(ctx context.Context, arg ClearForgeRepoMergeQueueConfigsParams) error ClearOwedMention(ctx context.Context, arg ClearOwedMentionParams) (int64, error) // Close intervals only after the binding and matching end event are absent. // The tenant id comes from each start because this sweep has no tenant GUC. @@ -228,6 +229,7 @@ type Querier interface { // DeleteChannelMember (channels.sql) — the statements are identical. GetCoordinationGroup(ctx context.Context, arg GetCoordinationGroupParams) (string, error) GetDMChannelByName(ctx context.Context, arg GetDMChannelByNameParams) (GetDMChannelByNameRow, error) + GetForgeRepoMergeQueueConfig(ctx context.Context, arg GetForgeRepoMergeQueueConfigParams) (GetForgeRepoMergeQueueConfigRow, error) GetGatewayCredentialForUpdate(ctx context.Context, id string) (GetGatewayCredentialForUpdateRow, error) // Feeds isReservedGroupTx: the reserved-group discriminator (top-level AND a reserved name AND // VisibilityOwner) the CreateChannel and CreateChannelGroup create-guards key on. @@ -398,10 +400,12 @@ type Querier interface { // observed) and the subscriber rows, container-scope rows collapsed per // (repo, kind) to coord_number 0. The Go groups the flat rows into targets. ListForgeNotifyTargets(ctx context.Context, arg ListForgeNotifyTargetsParams) ([]ListForgeNotifyTargetsRow, error) + ListForgeRepoMergeQueueConfigs(ctx context.Context, arg ListForgeRepoMergeQueueConfigsParams) ([]ListForgeRepoMergeQueueConfigsRow, error) ListForgeScopeRepos(ctx context.Context, arg ListForgeScopeReposParams) ([]string, error) ListGatewayCredentials(ctx context.Context, arg ListGatewayCredentialsParams) ([]ListGatewayCredentialsRow, error) ListIssues(ctx context.Context) ([]ListIssuesRow, error) ListMessages(ctx context.Context, arg ListMessagesParams) ([]ListMessagesRow, error) + ListOpenPullRequests(ctx context.Context, arg ListOpenPullRequestsParams) ([]int64, error) ListPullRequestLinks(ctx context.Context, arg ListPullRequestLinksParams) ([]ListPullRequestLinksRow, error) // ListTenantIDs lists every tenant. tenants has no row-level security, so the // app role sees them all without the BYPASSRLS system role. @@ -531,6 +535,7 @@ type Querier interface { PlacementForAgent(ctx context.Context, agentAccountID string) (PlacementForAgentRow, error) PruneTranscriptEntries(ctx context.Context, arg PruneTranscriptEntriesParams) error PullRequestForgeUpdatedAt(ctx context.Context, arg PullRequestForgeUpdatedAtParams) (pgtype.Timestamptz, error) + PullRequestLinks(ctx context.Context, arg PullRequestLinksParams) ([]PullRequestLinksRow, error) // Explicit links attach directly; a closing reference attaches only when none // of the PR's explicit targets is an issue on the board. PullRequestsForIssues(ctx context.Context, arg PullRequestsForIssuesParams) ([]PullRequestsForIssuesRow, error) @@ -662,8 +667,10 @@ type Querier interface { // AgentActivity domain struct (agentActivityFromRow-equivalent, done inline in // the Go — absent-from-table means absent-from-map). SetActivity(ctx context.Context, arg SetActivityParams) error + SetForgeRepoMergeQueue(ctx context.Context, arg SetForgeRepoMergeQueueParams) (int64, error) SetForgeRepoSubscriptionEnabled(ctx context.Context, arg SetForgeRepoSubscriptionEnabledParams) (int64, error) SetIssueState(ctx context.Context, arg SetIssueStateParams) (int64, error) + SetPullRequestMergeQueue(ctx context.Context, arg SetPullRequestMergeQueueParams) (bool, error) SetTopicArchived(ctx context.Context, arg SetTopicArchivedParams) error SetTourState(ctx context.Context, arg SetTourStateParams) error SharesVisibleChannel(ctx context.Context, arg SharesVisibleChannelParams) (bool, error) diff --git a/go/internal/store/forge_cursors.go b/go/internal/store/forge_cursors.go index b353d33a6..77a73172f 100644 --- a/go/internal/store/forge_cursors.go +++ b/go/internal/store/forge_cursors.go @@ -19,10 +19,12 @@ import ( // (OQ-C's table model). Enabled=false soft-disables the target without deleting // its watermark history. type ForgeRepoSubscription struct { - Provider ForgeProvider - Host string - Repo string - Enabled bool + Provider ForgeProvider + Host string + Repo string + Enabled bool + MergeQueue string + MergeBranch string } // validCoordinate rejects the zero/empty coordinate fields every forge method @@ -106,10 +108,12 @@ func (s *Store) EnsureForgeRepoSubscription(ctx context.Context, sub ForgeRepoSu return err } if err := s.q.EnsureForgeRepoSubscription(ctx, db.EnsureForgeRepoSubscriptionParams{ - ForgeProvider: int16(sub.Provider), //nolint:gosec // G115: ForgeProvider is a CHECK-constrained 1..4 enum, always within int16 - ForgeHost: sub.Host, - Repo: sub.Repo, - Enabled: sub.Enabled, + ForgeProvider: int16(sub.Provider), //nolint:gosec // G115: ForgeProvider is a CHECK-constrained 1..4 enum, always within int16 + ForgeHost: sub.Host, + Repo: sub.Repo, + Enabled: sub.Enabled, + MergeQueue: sub.MergeQueue, + MergeQueueBranch: sub.MergeBranch, }); err != nil { return fmt.Errorf("store: ensure forge repo subscription: %w", err) } @@ -170,10 +174,12 @@ func (s *Store) ListEnabledForgeRepoSubscriptions(ctx context.Context, provider var out []ForgeRepoSubscription for _, r := range rows { out = append(out, ForgeRepoSubscription{ - Provider: ForgeProvider(r.ForgeProvider), - Host: r.ForgeHost, - Repo: r.Repo, - Enabled: r.Enabled, + Provider: ForgeProvider(r.ForgeProvider), + Host: r.ForgeHost, + Repo: r.Repo, + Enabled: r.Enabled, + MergeQueue: r.MergeQueue, + MergeBranch: r.MergeQueueBranch, }) } return out, nil diff --git a/go/internal/store/forge_cursors_pgtest_test.go b/go/internal/store/forge_cursors_pgtest_test.go index e7458703d..b4ff8a27d 100644 --- a/go/internal/store/forge_cursors_pgtest_test.go +++ b/go/internal/store/forge_cursors_pgtest_test.go @@ -198,6 +198,24 @@ func TestForgeRepoSubscriptionCRUD(t *testing.T) { sentinelIs(t, err, ErrNotFound, "set enabled on unknown coordinate") } +func TestMergeQueueConfigForReturnsDisabledRowsForHydrator(t *testing.T) { + ctx := context.Background() + s := newTestStore(t) + if err := s.EnsureForgeRepoSubscription(ctx, ForgeRepoSubscription{Provider: ForgeProviderGitHub, Host: "github.com", Repo: "owner/repo", Enabled: true}); err != nil { + t.Fatalf("EnsureForgeRepoSubscription: %v", err) + } + if err := s.SetForgeRepoMergeQueueConfig(ctx, "github.com", "owner/repo", "trunk", "main"); err != nil { + t.Fatalf("SetForgeRepoMergeQueueConfig: %v", err) + } + if err := s.SetForgeRepoSubscriptionEnabled(ctx, ForgeProviderGitHub, "github.com", "owner/repo", false); err != nil { + t.Fatalf("disable forge repo: %v", err) + } + provider, branch, err := s.MergeQueueConfigFor(ctx, "github.com", "owner/repo") + if err != nil || provider != "trunk" || branch != "main" { + t.Fatalf("disabled merge queue config = %q/%q, %v; want trunk/main", provider, branch, err) + } +} + // ── Test 9: invalid input → ErrInvalidArgument on every method ──────────────── func TestForgeCursorInvalidArgument(t *testing.T) { diff --git a/go/internal/store/merge_queue.go b/go/internal/store/merge_queue.go new file mode 100644 index 000000000..3a2f9b8d8 --- /dev/null +++ b/go/internal/store/merge_queue.go @@ -0,0 +1,127 @@ +package store + +import ( + "context" + "fmt" + "time" + + "github.com/RigelBuild/compass/go/internal/store/db" +) + +// MergeQueueConfig is the configured queue provider and target branch for a repo. +type MergeQueueConfig struct { + Repo string + Provider string + Branch string +} + +// MergeQueueConfigs returns enabled GitHub repositories configured for queue polling. +func (s *Store) MergeQueueConfigs(ctx context.Context, host string) ([]MergeQueueConfig, error) { + rows, err := s.q.ListForgeRepoMergeQueueConfigs(ctx, db.ListForgeRepoMergeQueueConfigsParams{ + ForgeProvider: int16(ForgeProviderGitHub), + ForgeHost: host, + }) + if err != nil { + return nil, fmt.Errorf("store: list forge merge queue configs: %w", err) + } + out := make([]MergeQueueConfig, 0, len(rows)) + for _, row := range rows { + out = append(out, MergeQueueConfig{Repo: row.Repo, Provider: row.MergeQueue, Branch: row.MergeQueueBranch}) + } + return out, nil +} + +// MergeQueueConfigFor returns the queue configured for a repository. +func (s *Store) MergeQueueConfigFor(ctx context.Context, host, repo string) (string, string, error) { + row, err := s.q.GetForgeRepoMergeQueueConfig(ctx, db.GetForgeRepoMergeQueueConfigParams{ + ForgeProvider: int16(ForgeProviderGitHub), + ForgeHost: host, + Repo: repo, + }) + if noRows(err) { + return "", "", nil + } + if err != nil { + return "", "", fmt.Errorf("store: get forge repo merge queue config: %w", err) + } + return row.MergeQueue, row.MergeQueueBranch, nil +} + +// SetForgeRepoMergeQueueConfig updates a GitHub repo's provider and queue branch. +func (s *Store) SetForgeRepoMergeQueueConfig(ctx context.Context, host, repo, provider, branch string) error { + if err := validCoordinate(ForgeProviderGitHub, host, repo); err != nil { + return err + } + affected, err := s.q.SetForgeRepoMergeQueue(ctx, db.SetForgeRepoMergeQueueParams{ + ForgeProvider: int16(ForgeProviderGitHub), + ForgeHost: host, + Repo: repo, + MergeQueue: provider, + MergeQueueBranch: branch, + }) + if err != nil { + return fmt.Errorf("store: set forge merge queue config: %w", err) + } + if affected == 0 { + return fmt.Errorf("%w: forge repo subscription (%q, %q)", ErrNotFound, host, repo) + } + return nil +} + +// ClearForgeRepoMergeQueueConfigs disables merge-queue polling for every GitHub repo at host. +func (s *Store) ClearForgeRepoMergeQueueConfigs(ctx context.Context, host string) error { + if host == "" { + return fmt.Errorf("%w: forge host is required", ErrInvalidArgument) + } + if err := s.q.ClearForgeRepoMergeQueueConfigs(ctx, db.ClearForgeRepoMergeQueueConfigsParams{ + ForgeProvider: int16(ForgeProviderGitHub), + ForgeHost: host, + }); err != nil { + return fmt.Errorf("store: clear forge merge queue configs: %w", err) + } + return nil +} + +// OpenPullRequests returns the open PR numbers for one GitHub repository. +func (s *Store) OpenPullRequests(ctx context.Context, host, repo string) ([]uint64, error) { + rows, err := s.q.ListOpenPullRequests(ctx, db.ListOpenPullRequestsParams{ + ForgeProvider: int16(ForgeProviderGitHub), + ForgeHost: host, + Repo: repo, + }) + if err != nil { + return nil, fmt.Errorf("store: list open pull requests: %w", err) + } + out := make([]uint64, 0, len(rows)) + for _, number := range rows { + out = append(out, uint64(number)) //nolint:gosec // G115: a stored PR number is a positive BIGINT written from a canonical uint64 forge number. + } + return out, nil +} + +// SetPullRequestMergeQueue stores an observation, returning true only when the value changed. +func (s *Store) SetPullRequestMergeQueue(ctx context.Context, coord ForgeCoord, entry []byte, observedAt time.Time) (bool, error) { + coord = coord.Normalized() + if err := coord.valid(); err != nil { + return false, err + } + if len(entry) == 0 { + return false, fmt.Errorf("%w: merge queue entry is required", ErrInvalidArgument) + } + p := coord.db() + changed, err := s.q.SetPullRequestMergeQueue(ctx, db.SetPullRequestMergeQueueParams{ + ForgeProvider: p.provider, + ForgeHost: p.host, + Repo: p.repo, + Number: p.number, + Column5: entry, + MergeQueueObservedAt: timestamptz(observedAt), + }) + if err != nil { + if noRows(err) { + return false, nil + } + return false, fmt.Errorf("store: set pull request merge queue: %w", err) + } + return changed, nil +} diff --git a/go/internal/store/migrations/0001_init.sql b/go/internal/store/migrations/0001_init.sql index b00d65289..7f27eddfa 100644 --- a/go/internal/store/migrations/0001_init.sql +++ b/go/internal/store/migrations/0001_init.sql @@ -1041,6 +1041,8 @@ CREATE TABLE forge_repo_subscriptions ( forge_provider SMALLINT NOT NULL CHECK (forge_provider IN (1, 2, 3, 4)), forge_host TEXT NOT NULL, repo TEXT NOT NULL, + merge_queue TEXT NOT NULL DEFAULT '', + merge_queue_branch TEXT NOT NULL DEFAULT '', enabled BOOLEAN NOT NULL DEFAULT TRUE, swept_updated_at TIMESTAMPTZ, -- last swept forge updated_at watermark; NULL = never swept list_etag TEXT NOT NULL DEFAULT '', -- conditional-GET etag for the repo LIST walk @@ -1062,6 +1064,7 @@ CREATE TABLE pull_requests ( forge_created_at TIMESTAMPTZ NOT NULL, forge_updated_at TIMESTAMPTZ NOT NULL, pr JSONB NOT NULL, + merge_queue_observed_at TIMESTAMPTZ, created_at TIMESTAMPTZ NOT NULL DEFAULT now(), updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), tenant_id TEXT NOT NULL DEFAULT current_setting('compass.tenant_id', TRUE) REFERENCES tenants (id) ON DELETE RESTRICT, diff --git a/go/internal/store/pull_requests.go b/go/internal/store/pull_requests.go index 3fa04eeea..27e2ad949 100644 --- a/go/internal/store/pull_requests.go +++ b/go/internal/store/pull_requests.go @@ -323,6 +323,29 @@ func (s *Store) FallbackIssuesForTarget(ctx context.Context, issue ForgeCoord) ( return out, nil } +// PullRequestLinks returns every issue coordinate linked to a pull request. +func (s *Store) PullRequestLinks(ctx context.Context, pr ForgeCoord) ([]ForgeCoord, error) { + pr = pr.Normalized() + if err := pr.valid(); err != nil { + return nil, err + } + p := pr.db() + rows, err := s.q.PullRequestLinks(ctx, db.PullRequestLinksParams{ + PrForgeProvider: p.provider, + PrForgeHost: p.host, + PrRepo: p.repo, + PrNumber: p.number, + }) + if err != nil { + return nil, fmt.Errorf("store: pull request links: %w", err) + } + out := make([]ForgeCoord, 0, len(rows)) + for _, row := range rows { + out = append(out, coordFromDB(row.IssueForgeProvider, row.IssueForgeHost, row.IssueRepo, row.IssueNumber)) + } + return out, nil +} + // PullRequestUpdatedAt reads the stored forge_updated_at of pr; ok is false when // the PR was never stored. The sweep hydrates only rows newer than this. func (s *Store) PullRequestUpdatedAt(ctx context.Context, pr ForgeCoord) (time.Time, bool, error) { diff --git a/go/internal/store/pull_requests_pgtest_test.go b/go/internal/store/pull_requests_pgtest_test.go index 7e4e4aa53..31f3f1e50 100644 --- a/go/internal/store/pull_requests_pgtest_test.go +++ b/go/internal/store/pull_requests_pgtest_test.go @@ -8,6 +8,7 @@ package store import ( "context" + "encoding/json" "errors" "slices" "testing" @@ -394,3 +395,101 @@ func TestPullRequestMixedCaseBoardIssueResolvesExplicitLink(t *testing.T) { t.Fatalf("closing ref attached although the explicit target is on the board: %v", got) } } + +func TestSetPullRequestMergeQueueRejectsStaleClear(t *testing.T) { + ctx := context.Background() + s, coord, issue := seedMergeQueuePullRequest(t) + entry := []byte(`{"provider":"trunk","state":"testing"}`) + observedAt := time.Date(2026, 10, 11, 12, 0, 0, 0, time.UTC) + if changed, err := s.SetPullRequestMergeQueue(ctx, coord, entry, observedAt); err != nil || !changed { + t.Fatalf("store entry = %v, %v; want changed", changed, err) + } + changed, err := s.SetPullRequestMergeQueue(ctx, coord, []byte("null"), observedAt.Add(-time.Second)) + if err != nil || changed { + t.Fatalf("stale clear = %v, %v; want unchanged", changed, err) + } + if got := storedQueueState(t, ctx, s, issue); got != "testing" { + t.Fatalf("queue after stale clear = %q, want testing", got) + } +} + +func TestSetPullRequestMergeQueueRejectsStaleEntryAfterClear(t *testing.T) { + ctx := context.Background() + s, coord, issue := seedMergeQueuePullRequest(t) + observedAt := time.Date(2026, 10, 11, 12, 0, 0, 0, time.UTC) + if changed, err := s.SetPullRequestMergeQueue(ctx, coord, []byte("null"), observedAt); err != nil || changed { + t.Fatalf("store clear = %v, %v; want unchanged", changed, err) + } + entry := []byte(`{"provider":"trunk","state":"testing"}`) + changed, err := s.SetPullRequestMergeQueue(ctx, coord, entry, observedAt.Add(-time.Second)) + if err != nil || changed { + t.Fatalf("stale entry = %v, %v; want unchanged", changed, err) + } + if got := storedQueueState(t, ctx, s, issue); got != "" { + t.Fatalf("queue after stale entry = %q, want absent", got) + } +} + +func TestSetPullRequestMergeQueueEqualValueAdvancesObservation(t *testing.T) { + ctx := context.Background() + s, coord, issue := seedMergeQueuePullRequest(t) + entry := []byte(`{"provider":"trunk","state":"testing"}`) + observedAt := time.Date(2026, 10, 11, 12, 0, 0, 0, time.UTC) + if changed, err := s.SetPullRequestMergeQueue(ctx, coord, entry, observedAt); err != nil || !changed { + t.Fatalf("store entry = %v, %v; want changed", changed, err) + } + changed, err := s.SetPullRequestMergeQueue(ctx, coord, entry, observedAt.Add(time.Second)) + if err != nil || changed { + t.Fatalf("equal newer entry = %v, %v; want unchanged", changed, err) + } + var watermark time.Time + if err := s.pool.QueryRow(ctx, `SELECT merge_queue_observed_at FROM pull_requests WHERE forge_provider = $1 AND forge_host = $2 AND repo = $3 AND number = $4`, int16(coord.Provider), coord.Host, coord.Repo, int64(coord.Number)).Scan(&watermark); err != nil { + t.Fatalf("read observation watermark: %v", err) + } + if !watermark.Equal(observedAt.Add(time.Second)) { + t.Fatalf("watermark = %v, want %v", watermark, observedAt.Add(time.Second)) + } + olderEntry := []byte(`{"provider":"trunk","state":"failed"}`) + changed, err = s.SetPullRequestMergeQueue(ctx, coord, olderEntry, observedAt.Add(500*time.Millisecond)) + if err != nil || changed { + t.Fatalf("entry older than equal observation = %v, %v; want unchanged", changed, err) + } + if got := storedQueueState(t, ctx, s, issue); got != "testing" { + t.Fatalf("queue after old entry = %q, want testing", got) + } +} + +func seedMergeQueuePullRequest(t *testing.T) (*Store, ForgeCoord, ForgeCoord) { + t.Helper() + ctx := context.Background() + s := newTestStore(t) + coord := ghCoord("owner/repo", 7) + issue := ghCoord("owner/repo", 1) + seedBoardIssue(t, ctx, s, issue) + mustUpsertPR(t, ctx, s, prRow(coord.Repo, coord.Number, "open", prBase, prBase.Add(time.Hour)), issue) + return s, coord, issue +} + +func storedQueueState(t *testing.T, ctx context.Context, s *Store, issue ForgeCoord) string { + t.Helper() + rows, err := s.PullRequestsForIssues(ctx, []ForgeCoord{issue}) + if err != nil { + t.Fatalf("PullRequestsForIssues: %v", err) + } + prs := rows[issue] + if len(prs) != 1 { + t.Fatalf("linked pull requests = %d, want 1", len(prs)) + } + var got struct { + MergeQueue *struct { + State string `json:"state"` + } `json:"mergeQueue"` + } + if err := json.Unmarshal(prs[0].PR, &got); err != nil { + t.Fatalf("decode PR JSON: %v", err) + } + if got.MergeQueue == nil { + return "" + } + return got.MergeQueue.State +} diff --git a/go/internal/store/queries/forge_cursors.sql b/go/internal/store/queries/forge_cursors.sql index 2a10ffb88..d4f15f80a 100644 --- a/go/internal/store/queries/forge_cursors.sql +++ b/go/internal/store/queries/forge_cursors.sql @@ -17,8 +17,8 @@ UPDATE forge_repo_subscriptions WHERE forge_provider = $1 AND forge_host = $2 AND repo = $3; -- name: EnsureForgeRepoSubscription :exec -INSERT INTO forge_repo_subscriptions (forge_provider, forge_host, repo, enabled) -VALUES ($1, $2, $3, $4) +INSERT INTO forge_repo_subscriptions (forge_provider, forge_host, repo, enabled, merge_queue, merge_queue_branch) +VALUES ($1, $2, $3, $4, $5, $6) ON CONFLICT (tenant_id, forge_provider, forge_host, repo) DO NOTHING; -- name: ListEnabledForgeRepos :many @@ -33,7 +33,7 @@ SELECT EXISTS ( WHERE repo = $1 AND enabled = TRUE); -- name: ListEnabledForgeRepoSubscriptions :many -SELECT forge_provider, forge_host, repo, enabled +SELECT forge_provider, forge_host, repo, enabled, merge_queue, merge_queue_branch FROM forge_repo_subscriptions WHERE forge_provider = $1 AND forge_host = $2 AND enabled = TRUE ORDER BY repo ASC; @@ -50,3 +50,31 @@ SELECT prs_backfilled_at FROM forge_repo_subscriptions -- name: MarkPRsBackfilled :execrows UPDATE forge_repo_subscriptions SET prs_backfilled_at = $4 WHERE forge_provider = $1 AND forge_host = $2 AND repo = $3; + +-- name: SetForgeRepoMergeQueue :execrows +UPDATE forge_repo_subscriptions + SET merge_queue = $4, + merge_queue_branch = $5 + WHERE forge_provider = $1 AND forge_host = $2 AND repo = $3; + +-- name: ListForgeRepoMergeQueueConfigs :many +SELECT repo, merge_queue, merge_queue_branch + FROM forge_repo_subscriptions + WHERE forge_provider = $1 AND forge_host = $2 AND enabled AND merge_queue <> '' + ORDER BY repo ASC; + +-- name: ListOpenPullRequests :many +SELECT number + FROM pull_requests + WHERE forge_provider = $1 AND forge_host = $2 AND repo = $3 AND forge_state = 'open' + ORDER BY number ASC; + +-- name: ClearForgeRepoMergeQueueConfigs :exec +UPDATE forge_repo_subscriptions + SET merge_queue = '', merge_queue_branch = '' + WHERE forge_provider = $1 AND forge_host = $2; + +-- name: GetForgeRepoMergeQueueConfig :one +SELECT merge_queue, merge_queue_branch + FROM forge_repo_subscriptions + WHERE forge_provider = $1 AND forge_host = $2 AND repo = $3; diff --git a/go/internal/store/queries/pull_requests.sql b/go/internal/store/queries/pull_requests.sql index e95bd305a..f26ebae4e 100644 --- a/go/internal/store/queries/pull_requests.sql +++ b/go/internal/store/queries/pull_requests.sql @@ -14,16 +14,19 @@ ON CONFLICT (tenant_id, forge_provider, forge_host, repo, number) DO NOTHING; -- name: UpsertPullRequestGuarded :execrows -- Zero rows affected means the stored row is newer and the write was skipped. INSERT INTO pull_requests - (forge_provider, forge_host, repo, number, forge_state, - forge_created_at, forge_updated_at, pr) + (forge_provider, forge_host, repo, number, forge_state, + forge_created_at, forge_updated_at, pr) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) ON CONFLICT (tenant_id, forge_provider, forge_host, repo, number) DO UPDATE - SET forge_state = EXCLUDED.forge_state, - forge_created_at = EXCLUDED.forge_created_at, - forge_updated_at = EXCLUDED.forge_updated_at, - pr = EXCLUDED.pr - WHERE EXCLUDED.forge_updated_at >= pull_requests.forge_updated_at; - + SET forge_state = EXCLUDED.forge_state, + forge_created_at = EXCLUDED.forge_created_at, + forge_updated_at = EXCLUDED.forge_updated_at, + pr = CASE + WHEN pull_requests.pr ? 'mergeQueue' THEN + jsonb_set(EXCLUDED.pr - 'mergeQueue', '{mergeQueue}', pull_requests.pr -> 'mergeQueue', true) + ELSE EXCLUDED.pr - 'mergeQueue' + END +WHERE EXCLUDED.forge_updated_at >= pull_requests.forge_updated_at; -- name: UpsertExplicitPullRequestLink :exec INSERT INTO pull_request_issue_links (pr_forge_provider, pr_forge_host, pr_repo, pr_number, @@ -101,3 +104,29 @@ SELECT DISTINCT c.issue_forge_provider, c.issue_forge_host, c.issue_repo, c.issu -- name: PullRequestForgeUpdatedAt :one SELECT forge_updated_at FROM pull_requests WHERE forge_provider = $1 AND forge_host = $2 AND repo = $3 AND number = $4; + +-- name: PullRequestLinks :many +SELECT issue_forge_provider, issue_forge_host, issue_repo, issue_number + FROM pull_request_issue_links + WHERE pr_forge_provider = $1 AND pr_forge_host = $2 AND pr_repo = $3 AND pr_number = $4; +-- name: SetPullRequestMergeQueue :one +WITH current AS MATERIALIZED ( + SELECT tenant_id, forge_provider, forge_host, repo, number, + COALESCE(pr -> 'mergeQueue', 'null'::jsonb) IS DISTINCT FROM $5::jsonb AS changed + FROM pull_requests + WHERE pull_requests.forge_provider = $1 AND pull_requests.forge_host = $2 AND pull_requests.repo = $3 AND pull_requests.number = $4 + AND $6 >= COALESCE(pull_requests.merge_queue_observed_at, '-infinity'::timestamptz) + FOR UPDATE +), updated AS ( + UPDATE pull_requests AS p + SET pr = jsonb_set(p.pr, '{mergeQueue}', $5::jsonb, true), + merge_queue_observed_at = $6 + FROM current + WHERE p.tenant_id = current.tenant_id + AND p.forge_provider = current.forge_provider + AND p.forge_host = current.forge_host + AND p.repo = current.repo + AND p.number = current.number + RETURNING current.changed +) +SELECT changed FROM updated; diff --git a/go/server/board_pr_ingest_pgtest_test.go b/go/server/board_pr_ingest_pgtest_test.go index 38f70923d..2c35ade17 100644 --- a/go/server/board_pr_ingest_pgtest_test.go +++ b/go/server/board_pr_ingest_pgtest_test.go @@ -4,6 +4,7 @@ package server import ( "context" + "log/slog" "testing" "time" @@ -60,7 +61,7 @@ func TestBoardReconcileBackfillAttachesOpenPR(t *testing.T) { ClosingRefs: []forge.IssueRef{{Repo: repo, Number: 2}}}} marked := &markSignal{boardReconcileStore: adapter, done: make(chan struct{})} rc := ingest.NewBoardReconciler(notModifiedLister{}, ingest.NewIngester(nil, brd, ref), marked, ingest.BoardReconcileConfig{ - Pace: -1, Pulls: ingest.NewPullRequestHydrator(reads, brd, ref), + Pace: -1, Pulls: ingest.NewPullRequestHydrator(reads, brd, ref, nil, nil, slog.Default()), }) runCtx, cancel := context.WithCancel(ctx) defer cancel() diff --git a/go/server/merge_queue.go b/go/server/merge_queue.go new file mode 100644 index 000000000..8b628a010 --- /dev/null +++ b/go/server/merge_queue.go @@ -0,0 +1,133 @@ +//go:build unix + +package server + +import ( + "context" + "errors" + "fmt" + "net/http" + "time" + + compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1" + "github.com/RigelBuild/compass/go/internal/board" + "github.com/RigelBuild/compass/go/internal/forge" + "github.com/RigelBuild/compass/go/internal/ingest" + "github.com/RigelBuild/compass/go/internal/store" + "google.golang.org/protobuf/encoding/protojson" +) + +type mergeQueueConfigStore struct { + st *store.Store + host string +} + +func (q *mergeQueueConfigStore) MergeQueueFor(ctx context.Context, repo string) (string, string, error) { + return q.st.MergeQueueConfigFor(ctx, q.host, repo) +} +func (q *mergeQueueConfigStore) MergeQueueRepos(ctx context.Context) ([]string, error) { + configs, err := q.st.MergeQueueConfigs(ctx, q.host) + if err != nil { + return nil, err + } + repos := make([]string, 0, len(configs)) + for _, config := range configs { + repos = append(repos, config.Repo) + } + return repos, nil +} + +func (q *mergeQueueConfigStore) MergeQueueHost() string { return q.host } + +type mergeQueuePullRequestStore struct { + st *store.Store + host string +} + +func (s mergeQueuePullRequestStore) OpenPullRequests(ctx context.Context, repo string) ([]uint64, error) { + return s.st.OpenPullRequests(ctx, s.host, repo) +} + +type mergeQueueProjection struct { + brd *board.IssueProjection +} + +func (p mergeQueueProjection) PublishPullRequestMergeQueue(ctx context.Context, coord store.ForgeCoord, entry *compassv1.MergeQueueEntry, observedAt time.Time) error { + body := []byte("null") + if entry != nil { + var err error + body, err = protojson.Marshal(entry) + if err != nil { + return fmt.Errorf("marshal merge queue entry: %w", err) + } + } + changed, err := p.brd.UpdatePullRequestMergeQueue(ctx, coord, body, observedAt) + if err != nil { + return err + } + if changed { + return p.brd.PublishPullRequestMergeQueue(ctx, coord, entry) + } + return nil +} + +func mergeQueueProviders(host string, token func(context.Context) ([]byte, error), httpClient *http.Client) map[string]forge.MergeQueue { + trunk := forge.NewTrunk(forge.TrunkConfig{Token: token, Host: host, HTTP: httpClient}) + return map[string]forge.MergeQueue{trunk.Name(): trunk} +} + +func validateMergeQueueConfigs(configs []MergeQueueConfig) error { + seen := make(map[string]struct{}, len(configs)) + for _, config := range configs { + if config.Repo == "" || config.Branch == "" { + return errors.New("merge queue repo and branch are required") + } + if config.Provider != "trunk" { + return fmt.Errorf("unsupported merge queue provider %q for %q", config.Provider, config.Repo) + } + if _, exists := seen[config.Repo]; exists { + return fmt.Errorf("duplicate merge queue repo %q", config.Repo) + } + seen[config.Repo] = struct{}{} + } + return nil +} + +var _ ingest.QueueConfigs = (*mergeQueueConfigStore)(nil) +var _ ingest.MergeQueueStore = mergeQueuePullRequestStore{} +var _ ingest.MergeQueueSink = mergeQueueProjection{} + +func reconcileMergeQueueConfigs(ctx context.Context, st *store.Store, host string, configs []MergeQueueConfig) error { + for i := range configs { + repo, err := normalizeMergeQueueRepo(configs[i].Repo) + if err != nil { + return err + } + configs[i].Repo = repo + } + if err := validateMergeQueueConfigs(configs); err != nil { + return err + } + if len(configs) == 0 && st == nil { + return nil + } + if st == nil { + return errors.New("merge queue configuration requires a store") + } + if err := st.ClearForgeRepoMergeQueueConfigs(ctx, host); err != nil { + return fmt.Errorf("clear forge merge queues: %w", err) + } + for _, config := range configs { + if err := st.SetForgeRepoMergeQueueConfig(ctx, host, config.Repo, config.Provider, config.Branch); err != nil { + if errors.Is(err, store.ErrNotFound) { + return fmt.Errorf("merge queue repo %q has no forge subscription: %w", config.Repo, err) + } + return fmt.Errorf("set forge merge queue %q: %w", config.Repo, err) + } + } + return nil +} + +func normalizeMergeQueueRepo(raw string) (string, error) { + return normalizeGitHubRepo(raw) +} diff --git a/go/server/merge_queue_pgtest_test.go b/go/server/merge_queue_pgtest_test.go new file mode 100644 index 000000000..eb736777b --- /dev/null +++ b/go/server/merge_queue_pgtest_test.go @@ -0,0 +1,270 @@ +//go:build pgtest && unix + +package server + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "sync/atomic" + "testing" + "time" + + "google.golang.org/protobuf/encoding/protojson" + + "github.com/RigelBuild/compass/go/events" + compassv1 "github.com/RigelBuild/compass/go/gen/compass/v1" + "github.com/RigelBuild/compass/go/internal/board" + "github.com/RigelBuild/compass/go/internal/pgtest" + "github.com/RigelBuild/compass/go/internal/store" +) + +func TestServeMergeQueueSweeperPersistsChangedEntryWithinOneTick(t *testing.T) { + api, requests := startFakeTrunk(t) + st, issueCoord, dsn := seedMergeQueueServeStore(t) + config := mergeQueueServeConfig(t, dsn, api) + serveCtx, cancel := context.WithCancel(context.Background()) + serveDone := make(chan error, 1) + go func() { serveDone <- Serve(serveCtx, config) }() + t.Cleanup(func() { + cancel() + select { + case err := <-serveDone: + if err != nil { + t.Errorf("Serve shutdown: %v", err) + } + case <-time.After(testTimeout): + t.Error("Serve did not stop after cancellation") + } + }) + waitListening(t, config.SocketPath) + select { + case <-requests: + case <-time.After(testTimeout): + t.Fatal("Serve did not call the fake Trunk API within one configured tick") + } + waitForQueueState(t, st, issueCoord, "testing") + select { + case <-requests: + case <-time.After(testTimeout): + t.Fatal("Serve did not call fake Trunk on its recurring tick") + } + waitForQueueState(t, st, issueCoord, "failed") +} + +func startFakeTrunk(t *testing.T) (*httptest.Server, <-chan struct{}) { + t.Helper() + const token = "fake-trunk-test-token" + var calls atomic.Int32 + requests := make(chan struct{}, 1) + api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost || r.URL.Path != "/v1/getSubmittedPullRequests" { + t.Errorf("Trunk request = %s %s", r.Method, r.URL.Path) + w.WriteHeader(http.StatusNotFound) + return + } + if got := r.Header.Get("X-Api-Token"); got != token { + t.Errorf("Trunk token header = %q, want fake test token", got) + } + var body struct { + TargetBranch string `json:"targetBranch"` + PRs []struct { + Number uint64 `json:"number"` + } `json:"prs"` + } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + t.Errorf("decode Trunk request: %v", err) + } + if body.TargetBranch != "main" || len(body.PRs) != 1 || body.PRs[0].Number != 7 { + t.Errorf("Trunk request branch/PRs = %q/%+v, want main/[7]", body.TargetBranch, body.PRs) + } + select { + case requests <- struct{}{}: + default: + } + w.Header().Set("Content-Type", "application/json") + state := "testing" + if calls.Add(1) > 1 { + state = "failed" + } + _, _ = fmt.Fprintf(w, `{"pullRequests":[{"prNumber":7,"state":%q,"stateChangedAt":"2026-10-11T12:34:56Z","isCurrentlySubmittedToQueue":true}],"notFound":[]}`, state) + })) + t.Cleanup(api.Close) + return api, requests +} + +func seedMergeQueueServeStore(t *testing.T) (*store.Store, store.ForgeCoord, string) { + t.Helper() + dsn := pgtest.RequireDSN(t) + ctx := context.Background() + st, err := store.Open(ctx, dsn) + if err != nil { + t.Fatalf("open seed store: %v", err) + } + t.Cleanup(st.Close) + issueCoord := store.ForgeCoord{Provider: store.ForgeProviderGitHub, Host: forgeTestHost, Repo: "owner/repo", Number: 1} + if err := st.EnsureForgeRepoSubscription(ctx, store.ForgeRepoSubscription{Provider: store.ForgeProviderGitHub, Host: forgeTestHost, Repo: "owner/repo", Enabled: true}); err != nil { + t.Fatalf("seed forge subscription: %v", err) + } + if _, err := st.UpsertIssueForgeFields(ctx, store.IssueForgeFields{ + ForgeProvider: store.ForgeProviderGitHub, + ForgeHost: forgeTestHost, + Repo: "owner/repo", + Number: 1, + Title: "linked issue", + ForgeState: "open", + URL: "https://github.com/owner/repo/issues/1", + ForgeAccount: "octocat", + }); err != nil { + t.Fatalf("seed board issue: %v", err) + } + + pr := &compassv1.PullRequest{ + Forge: &compassv1.ForgeRef{Provider: compassv1.ForgeProvider_FORGE_PROVIDER_GITHUB, Host: forgeTestHost}, + Repo: "owner/repo", + Number: 7, + ForgeState: "open", + } + prJSON, err := protojson.Marshal(pr) + if err != nil { + t.Fatal(err) + } + if _, err := st.UpsertPullRequest(ctx, store.PullRequestRow{ + Coord: store.ForgeCoord{Provider: store.ForgeProviderGitHub, Host: forgeTestHost, Repo: "owner/repo", Number: 7}, + State: "open", + CreatedAt: time.Date(2026, 10, 1, 0, 0, 0, 0, time.UTC), + UpdatedAt: time.Date(2026, 10, 10, 0, 0, 0, 0, time.UTC), + PR: prJSON, + }, []store.ForgeCoord{issueCoord}); err != nil { + t.Fatalf("seed linked pull request: %v", err) + } + if err := st.DeclareServerSecret(ctx, "", store.ServerSecretPrefix+"TRUNK_TOKEN"); err != nil { + t.Fatalf("declare Trunk token secret: %v", err) + } + return st, issueCoord, dsn +} + +func mergeQueueServeConfig(t *testing.T, dsn string, api *httptest.Server) ServeConfig { + t.Helper() + const token = "fake-trunk-test-token" + secretFile := filepath.Join(t.TempDir(), "serve.env") + secretContent := store.MasterKeyName + "=" + dotenvValue(testMasterKeyHex) + "\n" + + serverSecretName("TRUNK_TOKEN") + "=" + dotenvValue(token) + "\n" + if err := os.WriteFile(secretFile, []byte(secretContent), 0o600); err != nil { + t.Fatalf("write fake test secrets: %v", err) + } + config := ServeConfig{ + SocketPath: filepath.Join(t.TempDir(), "compass.sock"), + Version: "merge-queue-test", + DatabaseDSN: dsn, + SecretProvider: "dotenv://" + secretFile, + MergeQueueInterval: 25 * time.Millisecond, + Forge: ForgeConfig{ + Host: forgeTestHost, + MergeQueues: []MergeQueueConfig{{Repo: "owner/repo", Provider: "trunk", Branch: "main"}}, + TrunkTokenSecretName: "TRUNK_TOKEN", + TrunkHTTP: &http.Client{Transport: queueHTTPTransport{target: api}}, + }, + } + provisionNats(t, &config) + return config +} + +func waitForQueueState(t *testing.T, st *store.Store, issueCoord store.ForgeCoord, state string) { + t.Helper() + ctx := context.Background() + deadline := time.NewTimer(testTimeout) + defer deadline.Stop() + poll := time.NewTicker(5 * time.Millisecond) + defer poll.Stop() + for { + rows, err := st.PullRequestsForIssues(ctx, []store.ForgeCoord{issueCoord}) + if err != nil { + t.Fatalf("read linked pull request: %v", err) + } + prs := rows[issueCoord] + if len(prs) == 1 { + got := &compassv1.PullRequest{} + if err := (protojson.UnmarshalOptions{DiscardUnknown: true}).Unmarshal(prs[0].PR, got); err != nil { + t.Fatalf("decode stored linked pull request: %v", err) + } + if got.GetMergeQueue().GetState() == state { + return + } + } + select { + case <-deadline.C: + t.Fatalf("Serve did not persist queue entry state %q: linkedPRs=%d", state, len(prs)) + case <-poll.C: + } + } +} + +func TestReconcileMergeQueueConfigsDoesNotEnrollUnwatchedRepo(t *testing.T) { + ctx := context.Background() + st := forgeTestStore(t) + err := reconcileMergeQueueConfigs(ctx, st, forgeTestHost, []MergeQueueConfig{{Repo: "owner/repo", Provider: "trunk", Branch: "main"}}) + if !errors.Is(err, store.ErrNotFound) { + t.Fatalf("reconcileMergeQueueConfigs error = %v, want missing subscription", err) + } + enabled, err := st.IsEnabledForgeRepo(ctx, "owner/repo") + if err != nil || enabled { + t.Fatalf("queue-only subscription enabled = %v, %v; want false", enabled, err) + } +} + +func TestMergeQueueProjectionPublishesOnlyChangedQueueEntries(t *testing.T) { + ctx := context.Background() + st, issueCoord, _ := seedMergeQueueServeStore(t) + if err := st.SetForgeRepoMergeQueueConfig(ctx, forgeTestHost, "owner/repo", "trunk", "main"); err != nil { + t.Fatalf("seed queue config: %v", err) + } + bus := events.NewBus[busPayload]() + t.Cleanup(bus.Close) + issueBrd := board.NewIssueProjection(bus, st) + if err := issueBrd.Rehydrate(ctx); err != nil { + t.Fatalf("rehydrate board: %v", err) + } + subscription, err := bus.Subscribe(0, bus.InstanceEpoch()) + if err != nil { + t.Fatalf("subscribe: %v", err) + } + t.Cleanup(subscription.Cancel) + projection := mergeQueueProjection{brd: issueBrd} + coord := store.ForgeCoord{Provider: store.ForgeProviderGitHub, Host: forgeTestHost, Repo: "owner/repo", Number: 7} + entry := &compassv1.MergeQueueEntry{Provider: "trunk", State: "testing"} + observedAt := time.Date(2026, 10, 11, 12, 0, 0, 0, time.UTC) + if err := projection.PublishPullRequestMergeQueue(ctx, coord, entry, observedAt); err != nil { + t.Fatalf("publish new entry: %v", err) + } + select { + case event := <-subscription.Live: + if event.Payload.GetIssue().GetNumber() != uint32(issueCoord.Number) { + t.Fatalf("first event issue = %d, want %d", event.Payload.GetIssue().GetNumber(), issueCoord.Number) + } + case <-time.After(testTimeout): + t.Fatal("changed queue entry did not republish linked issue") + } + if err := projection.PublishPullRequestMergeQueue(ctx, coord, entry, observedAt.Add(time.Second)); err != nil { + t.Fatalf("publish unchanged entry: %v", err) + } + select { + case event := <-subscription.Live: + t.Fatalf("unchanged queue entry republished issue: %v", event.Payload) + default: + } +} + +type queueHTTPTransport struct{ target *httptest.Server } + +func (r queueHTTPTransport) RoundTrip(req *http.Request) (*http.Response, error) { + copyReq := req.Clone(req.Context()) + copyReq.URL.Scheme = "http" + copyReq.URL.Host = r.target.Listener.Addr().String() + return http.DefaultTransport.RoundTrip(copyReq) +} diff --git a/go/server/merge_queue_test.go b/go/server/merge_queue_test.go new file mode 100644 index 000000000..1ccdcf303 --- /dev/null +++ b/go/server/merge_queue_test.go @@ -0,0 +1,39 @@ +//go:build unix + +package server + +import ( + "context" + "strings" + "testing" +) + +func TestValidateMergeQueueConfigs(t *testing.T) { + for _, tc := range []struct { + name string + configs []MergeQueueConfig + wantErr string + }{ + {name: "valid", configs: []MergeQueueConfig{{Repo: "owner/repo", Provider: "trunk", Branch: "main"}}}, + {name: "missing branch", configs: []MergeQueueConfig{{Repo: "owner/repo", Provider: "trunk"}}, wantErr: "required"}, + {name: "unsupported provider", configs: []MergeQueueConfig{{Repo: "owner/repo", Provider: "github", Branch: "main"}}, wantErr: "unsupported"}, + {name: "duplicate repo", configs: []MergeQueueConfig{{Repo: "owner/repo", Provider: "trunk", Branch: "main"}, {Repo: "owner/repo", Provider: "trunk", Branch: "release"}}, wantErr: "duplicate"}, + } { + t.Run(tc.name, func(t *testing.T) { + err := validateMergeQueueConfigs(tc.configs) + if tc.wantErr == "" && err != nil { + t.Fatalf("validateMergeQueueConfigs: %v", err) + } + if tc.wantErr != "" && (err == nil || !strings.Contains(err.Error(), tc.wantErr)) { + t.Fatalf("validateMergeQueueConfigs = %v, want %q", err, tc.wantErr) + } + }) + } +} + +func TestPrepareMergeQueueSweeperDoesNotBuildWhenNoReposConfigured(t *testing.T) { + runner, err := prepareMergeQueueSweeper(context.Background(), ServeConfig{}, nil, nil, nil) + if err != nil || runner.sweeper != nil { + t.Fatalf("prepareMergeQueueSweeper = %v, %v; want empty runner, nil", runner, err) + } +} diff --git a/go/server/serve.go b/go/server/serve.go index 64add91cd..51ddd6ef3 100644 --- a/go/server/serve.go +++ b/go/server/serve.go @@ -154,6 +154,7 @@ type ServeConfig struct { // RunnerClustersPath is the Runner cluster file (--runner-clusters). When // set, the RunnerService door also accepts projected ServiceAccount tokens. RunnerClustersPath string + MergeQueueInterval time.Duration } // ForgeConfig configures the board webhook-ingestion lane (RIG-2883) and the @@ -206,6 +207,12 @@ type ForgeConfig struct { // INDEPENDENT of the GitHub App gate (a deployment can run Linear // notifications without a GitHub App and vice versa). LinearWebhookSecretName string + // TrunkTokenSecretName is the declared server-only secret name for Trunk API access. + TrunkTokenSecretName string + // MergeQueues maps GitHub repositories to Trunk target branches. + MergeQueues []MergeQueueConfig + // TrunkHTTP overrides the Trunk API client in tests. + TrunkHTTP *http.Client // ScopeEnforcementDisabled opts out of grant-gated forge writes. The zero // value enforces; Dogfood opts out to keep the single-trust-domain posture. ScopeEnforcementDisabled bool @@ -214,6 +221,13 @@ type ForgeConfig struct { ScopeGrants []store.ForgeScope } +// MergeQueueConfig binds a repository to a provider and queue target branch. +type MergeQueueConfig struct { + Repo string + Provider string + Branch string +} + // ForgeAppConfig is the GitHub App credential the board webhook-ingestion lane // runs on (RIG-2883, specified at // docs/designs/server/compass-forge-agent-notification/design.md:1035-1048). @@ -235,6 +249,40 @@ type ForgeAppConfig struct { ReconcileBackstop time.Duration } +const defaultMergeQueueSweepInterval = 2 * time.Minute + +type mergeQueueRunner struct { + sweeper *ingest.MergeQueueSweeper +} + +func (r mergeQueueRunner) start(g *errgroup.Group, ctx context.Context) { + if r.sweeper != nil { + g.Go(func() error { return r.sweeper.Run(ctx) }) + } +} + +func prepareMergeQueueSweeper(ctx context.Context, cfg ServeConfig, st *store.Store, issueBrd *board.IssueProjection, resolver secrets.Resolver) (mergeQueueRunner, error) { + fc := cfg.Forge.resolved() + if err := reconcileMergeQueueConfigs(ctx, st, fc.Host, fc.MergeQueues); err != nil { + return mergeQueueRunner{}, fmt.Errorf("configure forge merge queues: %w", err) + } + if len(fc.MergeQueues) == 0 { + return mergeQueueRunner{}, nil + } + if fc.TrunkTokenSecretName == "" { + return mergeQueueRunner{}, errors.New("--forge-trunk-token-secret is required when merge queues are configured") + } + queues := mergeQueueProviders(fc.Host, newTrunkTokenResolver(resolver, fc.TrunkTokenSecretName), fc.TrunkHTTP) + cfgs := &mergeQueueConfigStore{st: st, host: fc.Host} + storeAdapter := mergeQueuePullRequestStore{st: st, host: fc.Host} + sink := mergeQueueProjection{brd: issueBrd} + every := cfg.MergeQueueInterval + if every <= 0 { + every = defaultMergeQueueSweepInterval + } + return mergeQueueRunner{sweeper: ingest.NewMergeQueueSweeper(queues, cfgs, storeAdapter, sink, every)}, nil +} + // Forge config defaults, applied by resolveForge when a field is zero. const ( defaultForgeHost = "github.com" @@ -602,7 +650,9 @@ func declareServerSecretNames(ctx context.Context, st *store.Store, cfg ServeCon if raw.LinearWebhookSecretName != "" { names = append(names, raw.LinearWebhookSecretName) } - + if fc.TrunkTokenSecretName != "" { + names = append(names, fc.TrunkTokenSecretName) + } for _, name := range names { if name == "" { // An unset name is "not configured", never a bare prefix. @@ -886,6 +936,10 @@ func Serve(ctx context.Context, cfg ServeConfig) error { if err != nil { return failStartup(udsListener, listeners, err) } + mergeQueueRunner, err := prepareMergeQueueSweeper(ctx, cfg, st, issueBrd, serverResolver) + if err != nil { + return failStartup(udsListener, listeners, err) + } // The forge-WRITE caller: enabled iff BOTH primary and reviewer Apps are // configured (the 2-App cutover). The author leg REUSES the shared primary App @@ -900,19 +954,7 @@ func Serve(ctx context.Context, cfg ServeConfig) error { // error and always ends by Wait, so the verifier's key refresh never outlives // Serve. The UDS door is primary; its error wins over the drain result below. g, gctx := newServeGroup(ctx, doors.runnerVerifier) - g.Go(func() error { return classifyServe(doors.uds.Serve(udsListener), "compass.v1 UDS server") }) - if doors.dev != nil { - g.Go(func() error { return classifyServe(doors.dev.Serve(devListener), "dev gRPC-Web server") }) - } - if doors.net != nil { - // ServeTLS wraps the pre-bound listener with the configured keypair - // (already in TLSConfig, so the empty cert/key args skip a file load) and - // advertises ALPN h2 from Protocols. A clean shutdown returns - // ErrServerClosed, mapped to nil by classifyServe. - g.Go(func() error { - return classifyServe(doors.net.ServeTLS(netListener, "", ""), "compass.v1 network door") - }) - } + startServeDoors(g, gctx, doors, udsListener, devListener, netListener, mergeQueueRunner) // The forge webhook-ingestion lanes (RIG-2883 board + RIG-2732 T7 notify): // each lane joins the SAME scoped group so it inherits the doors' lifecycle. // The board + GitHub notify lanes are nil when the GitHub App is absent; the @@ -947,6 +989,19 @@ func Serve(ctx context.Context, cfg ServeConfig) error { return g.Wait() } +func startServeDoors(g *errgroup.Group, ctx context.Context, doors serveDoors, uds, dev, network net.Listener, mergeQueue mergeQueueRunner) { + g.Go(func() error { return classifyServe(doors.uds.Serve(uds), "compass.v1 UDS server") }) + if doors.dev != nil { + g.Go(func() error { return classifyServe(doors.dev.Serve(dev), "dev gRPC-Web server") }) + } + if doors.net != nil { + g.Go(func() error { + return classifyServe(doors.net.ServeTLS(network, "", ""), "compass.v1 network door") + }) + } + mergeQueue.start(g, ctx) +} + // serveDoors holds the three compass.v1 doors Serve drives: the shipped Unix // socket (always built), the optional dev loopback browser door, and the // optional authenticated network door. dev and net are nil when their listener @@ -1451,20 +1506,7 @@ func buildLinearWebhookWiring( return handler, nil } -// buildBoardIngestLane assembles the App-only board webhook-ingestion lane -// (RIG-2883 T5) over the shared GitHub client. The caller -// (buildBoardWebhookWiring) owns the App gate, the two App-secret validations, -// and the client construction — this builder is only reached when the App is -// configured and validated, so it assembles unconditionally. -// -// In order it: (1) reconciles the seed repos into forge_repo_subscriptions -// (bootstrap-only insert, ON CONFLICT DO NOTHING, lowercased for GITHUB) so the -// target rows are visible before the first sweep; (2) assembles the pipeline -// over the shared client — the Ingester (sharing issueBrd) -> the T1 webhook arm -// (over the store's IsEnabledForgeRepo target check) and the T3 reconciler (over -// the store's enabled-repo + watermark seam). Both ingest seams are adapted onto -// *store.Store binding (provider, host) — the ingest package owns no store type, -// so go/server supplies these thin adapters. +// buildBoardIngestLane assembles the GitHub board-ingestion pipeline with optional queue lookup. func buildBoardIngestLane( ctx context.Context, cfg ServeConfig, @@ -1475,35 +1517,24 @@ func buildBoardIngestLane( log *slog.Logger, ) (*boardIngestLane, error) { fc := cfg.Forge.resolved() - // The board lane ships a GitHub client only this slice (Global Constraints); - // the bound provider is GITHUB. const provider = store.ForgeProviderGitHub - - // (1) Seed reconcile BEFORE the first sweep: each seed repo lowercased (for - // GITHUB) and inserted enabled under ON CONFLICT DO NOTHING — additive, never - // destructive, never re-enabling. Bad repo format fails startup. if err := reconcileForgeSeed(ctx, st, provider, fc.Host, cfg.Forge.SeedRepos); err != nil { return nil, err } - - // (2) Assemble the pipeline over the shared client: the Ingester (sharing the - // existing issue projection), and the two arms over store adapters binding - // (provider, host). - ref := &compassv1.ForgeRef{ - Provider: compassv1.ForgeProvider_FORGE_PROVIDER_GITHUB, - Host: fc.Host, + var queues map[string]forge.MergeQueue + var queueConfigs ingest.QueueConfigs + if len(fc.MergeQueues) > 0 { + queues = mergeQueueProviders(fc.Host, newTrunkTokenResolver(buildServerSecretResolver(st, cfg), fc.TrunkTokenSecretName), fc.TrunkHTTP) + queueConfigs = &mergeQueueConfigStore{st: st, host: fc.Host} } + ref := &compassv1.ForgeRef{Provider: compassv1.ForgeProvider_FORGE_PROVIDER_GITHUB, Host: fc.Host} ing := ingest.NewIngester(client, issueBrd, ref) - pulls := ingest.NewPullRequestHydrator(client, issueBrd, ref) + pulls := ingest.NewPullRequestHydrator(client, issueBrd, ref, queues, queueConfigs, log) arm := ingest.NewBoardWebhookArm(client, ing, &boardTargetStore{st: st}, ingest.BoardArmConfig{ - Log: log, - Pulls: pulls, - PullNumbers: pullNumbers, + Log: log, Pulls: pulls, PullNumbers: pullNumbers, }) reconciler := ingest.NewBoardReconciler(client, ing, &boardReconcileStore{st: st, provider: provider, host: fc.Host}, ingest.BoardReconcileConfig{ - Backstop: fc.App.ReconcileBackstop, - Log: log, - Pulls: pulls, + Backstop: fc.App.ReconcileBackstop, Log: log, Pulls: pulls, }) return &boardIngestLane{arm: arm, reconciler: reconciler, sink: arm, client: client}, nil } @@ -1978,18 +2009,26 @@ func buildLinearNotifyLane( // and verifies webhook HMACs, so an unwrapped comparison here fails every // resolve at runtime, not at boot. func newDeclaredSecretResolver(resolver secrets.Resolver, name string) func(ctx context.Context) ([]byte, error) { + return resolveDeclaredSecret(resolver, name, "board webhook") +} + +func newTrunkTokenResolver(resolver secrets.Resolver, name string) func(ctx context.Context) ([]byte, error) { + return resolveDeclaredSecret(resolver, name, "Trunk API") +} + +func resolveDeclaredSecret(resolver secrets.Resolver, name, label string) func(ctx context.Context) ([]byte, error) { want := serverSecretName(name) return func(ctx context.Context) ([]byte, error) { - resolved, err := resolver.Resolve(ctx, "board webhook") + resolved, err := resolver.Resolve(ctx, label) if err != nil { - return nil, fmt.Errorf("board webhook secret resolve: %w", err) + return nil, fmt.Errorf("%s secret resolve: %w", label, err) } - for _, s := range resolved { - if s.Name == want { - return []byte(s.Value), nil + for _, secret := range resolved { + if secret.Name == want { + return []byte(secret.Value), nil } } - return nil, fmt.Errorf("board webhook secret %q not declared", name) + return nil, fmt.Errorf("%s secret %q not declared", label, name) } }