feat(forge): merge-queue provider with Trunk and a queue sweep (RIG-5183)

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* --

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) {

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,

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)
+			}
+		})
+	}
+}

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)

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() {

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)
+}