feat(mr): background mergeability checks, their Watch event and API field

commit: 23c66c065d92af4179bd23f31397038a5486c8b7
author: brice · 3d ago · 2026-10-19 13:05 UTC
committer: brice · 2d ago · 2026-10-19 18:30 UTC
parent: 8f36a2f
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

17 files · +1014 -20

M internal/adapters/client/features.go +3 -0


@@ -54,6 +54,9 @@ const (
54	54		// FeatureCommitSignatures is TrustedKeyService and Commit.verification
55	55		// (`ermite signing-key trust`).
56	56		FeatureCommitSignatures = "commit_signatures"
	57	+	// FeatureMergeability is MergeRequest.mergeability and the
	58	+	// merge_request_mergeability event.
	59	+	FeatureMergeability = "merge_requests.mergeability"
57	60		)
58	61
59	62		// UsedFeatures lists the features the CLI uses, in the order they are reported.


@@ -54,6 +54,9 @@ const (
54	// FeatureCommitSignatures is TrustedKeyService and Commit.verification	54		// FeatureCommitSignatures is TrustedKeyService and Commit.verification
55	// (`ermite signing-key trust`).	55		// (`ermite signing-key trust`).
56	FeatureCommitSignatures = "commit_signatures"	56		FeatureCommitSignatures = "commit_signatures"
		57	+	// FeatureMergeability is MergeRequest.mergeability and the
		58	+	// merge_request_mergeability event.
		59	+	FeatureMergeability = "merge_requests.mergeability"
57	)	60		)
58		61
59	// UsedFeatures lists the features the CLI uses, in the order they are reported.	62		// UsedFeatures lists the features the CLI uses, in the order they are reported.

M internal/adapters/kvstore/mergerequests.go +1 -0


@@ -679,6 +679,7 @@ func encodeMergeability(m domain.Mergeability) ([]byte, error) {
679	679		})
680	680		}
681	681
	682	+	//hexcheck:ignore suspicious-business-logic-in-adapter decodes the stored record into the domain value; no rule is applied
682	683		func decodeMergeability(repo domain.RepoName, number int64, item kv.Item) (domain.Mergeability, error) {
683	684		var pb storagepb.Mergeability
684	685		if err := proto.Unmarshal(item.Value, &pb); err != nil {


@@ -679,6 +679,7 @@ func encodeMergeability(m domain.Mergeability) ([]byte, error) {
679	})	679		})
680	}	680		}
681		681
		682	+	//hexcheck:ignore suspicious-business-logic-in-adapter decodes the stored record into the domain value; no rule is applied
682	func decodeMergeability(repo domain.RepoName, number int64, item kv.Item) (domain.Mergeability, error) {	683		func decodeMergeability(repo domain.RepoName, number int64, item kv.Item) (domain.Mergeability, error) {
683	var pb storagepb.Mergeability	684		var pb storagepb.Mergeability
684	if err := proto.Unmarshal(item.Value, &pb); err != nil {	685		if err := proto.Unmarshal(item.Value, &pb); err != nil {

M internal/adapters/rpc/api.go +1 -1


@@ -35,7 +35,7 @@ const RunnerMaxRequestBytes = MaxRequestBytes + 64<<10
35	35		// always implements, as GetVersion lists them. FeaturePasskeys comes on top
36	36		// when Config.Passkeys is set, and FeatureCISecrets when the settings services
37	37		// are, and FeatureCommitSignatures when Config.TrustedKeys is set.
38		-	var Features = []string{"repo.update", "events.resume", "merge_requests", "merge_requests.delete_branch", "issues", "signing_key", "session", "git.list_paths", "token.branch_prefixes", "ci.runs", "ci.runs.filters", "ci.runners", "ci.actions", "ci.annotations", "ci.summaries", "ci.concurrency", "ci.dispatch", "ci.cache", "ci.workflows"}
	38	+	var Features = []string{"repo.update", "events.resume", "merge_requests", "merge_requests.delete_branch", "issues", "merge_requests.mergeability", "signing_key", "session", "git.list_paths", "token.branch_prefixes", "ci.runs", "ci.runs.filters", "ci.runners", "ci.actions", "ci.annotations", "ci.summaries", "ci.concurrency", "ci.dispatch", "ci.cache", "ci.workflows"}
39	39
40	40		// FeaturePasskeys is listed by GetVersion when passkeys are available.
41	41		// FeatureCIArtifacts (artifact.go) when the artifact services are served.


@@ -35,7 +35,7 @@ const RunnerMaxRequestBytes = MaxRequestBytes + 64<<10
35		// always implements, as GetVersion lists them. FeaturePasskeys comes on top	35		// always implements, as GetVersion lists them. FeaturePasskeys comes on top
36		// when Config.Passkeys is set, and FeatureCISecrets when the settings services	36		// when Config.Passkeys is set, and FeatureCISecrets when the settings services
37		// are, and FeatureCommitSignatures when Config.TrustedKeys is set.	37		// are, and FeatureCommitSignatures when Config.TrustedKeys is set.
38	-	var Features = []string{"repo.update", "events.resume", "merge_requests", "merge_requests.delete_branch", "issues", "signing_key", "session", "git.list_paths", "token.branch_prefixes", "ci.runs", "ci.runs.filters", "ci.runners", "ci.actions", "ci.annotations", "ci.summaries", "ci.concurrency", "ci.dispatch", "ci.cache", "ci.workflows"}	38	+	var Features = []string{"repo.update", "events.resume", "merge_requests", "merge_requests.delete_branch", "issues", "merge_requests.mergeability", "signing_key", "session", "git.list_paths", "token.branch_prefixes", "ci.runs", "ci.runs.filters", "ci.runners", "ci.actions", "ci.annotations", "ci.summaries", "ci.concurrency", "ci.dispatch", "ci.cache", "ci.workflows"}
39			39
40		// FeaturePasskeys is listed by GetVersion when passkeys are available.	40		// FeaturePasskeys is listed by GetVersion when passkeys are available.
41		// FeatureCIArtifacts (artifact.go) when the artifact services are served.	41		// FeatureCIArtifacts (artifact.go) when the artifact services are served.

M internal/adapters/rpc/convert.go +3 -0


@@ -501,6 +501,9 @@ func watchEventToPB(we ports.WatchEvent) *pb.Event {
501	501		case domain.CommentResolved:
502	502		out.Time = timestamp(ev.At)
503	503		out.Kind = &pb.Event_CommentResolved{CommentResolved: &pb.CommentResolved{Repo: ev.Repo.String(), Number: ev.Number, Comment: commentToPB(ev.Comment)}}
	504	+	case domain.MergeRequestMergeability:
	505	+	out.Time = timestamp(ev.At)
	506	+	out.Kind = &pb.Event_MergeRequestMergeability{MergeRequestMergeability: &pb.MergeRequestMergeability{Repo: ev.Repo.String(), Number: ev.Number, Mergeability: mergeabilityToPB(ev.Mergeability)}}
504	507		case domain.RunUpdated:
505	508		out.Time = timestamp(ev.At)
506	509		out.Kind = runEventToPB(ev)


@@ -501,6 +501,9 @@ func watchEventToPB(we ports.WatchEvent) *pb.Event {
501	case domain.CommentResolved:	501		case domain.CommentResolved:
502	out.Time = timestamp(ev.At)	502		out.Time = timestamp(ev.At)
503	out.Kind = &pb.Event_CommentResolved{CommentResolved: &pb.CommentResolved{Repo: ev.Repo.String(), Number: ev.Number, Comment: commentToPB(ev.Comment)}}	503		out.Kind = &pb.Event_CommentResolved{CommentResolved: &pb.CommentResolved{Repo: ev.Repo.String(), Number: ev.Number, Comment: commentToPB(ev.Comment)}}
		504	+	case domain.MergeRequestMergeability:
		505	+	out.Time = timestamp(ev.At)
		506	+	out.Kind = &pb.Event_MergeRequestMergeability{MergeRequestMergeability: &pb.MergeRequestMergeability{Repo: ev.Repo.String(), Number: ev.Number, Mergeability: mergeabilityToPB(ev.Mergeability)}}
504	case domain.RunUpdated:	507		case domain.RunUpdated:
505	out.Time = timestamp(ev.At)	508		out.Time = timestamp(ev.At)
506	out.Kind = runEventToPB(ev)	509		out.Kind = runEventToPB(ev)

M internal/adapters/rpc/mergerequest.go +39 -2


@@ -133,6 +133,43 @@ func mergeRequestToPB(mr domain.MergeRequest) *pb.MergeRequest {
133	133		}
134	134		}
135	135
	136	+	// mergeRequestWithMergeabilityToPB is mergeRequestToPB with the mergeability of
	137	+	// an open request; m is ignored for a request that is not open.
	138	+	func mergeRequestWithMergeabilityToPB(mr domain.MergeRequest, m domain.Mergeability) *pb.MergeRequest {
	139	+	out := mergeRequestToPB(mr)
	140	+	if mr.IsOpen() && m.State != "" {
	141	+	out.Mergeability = mergeabilityToPB(m)
	142	+	}
	143	+	return out
	144	+	}
	145	+
	146	+	func mergeabilityStateToPB(s domain.MergeabilityState) pb.MergeabilityState {
	147	+	switch s {
	148	+	case domain.MergeabilityChecking:
	149	+	return pb.MergeabilityState_MERGEABILITY_STATE_CHECKING
	150	+	case domain.MergeabilityUpToDate:
	151	+	return pb.MergeabilityState_MERGEABILITY_STATE_UP_TO_DATE
	152	+	case domain.MergeabilityBehind:
	153	+	return pb.MergeabilityState_MERGEABILITY_STATE_BEHIND
	154	+	case domain.MergeabilityConflict:
	155	+	return pb.MergeabilityState_MERGEABILITY_STATE_CONFLICT
	156	+	}
	157	+	return pb.MergeabilityState_MERGEABILITY_STATE_UNSPECIFIED
	158	+	}
	159	+
	160	+	func mergeabilityToPB(m domain.Mergeability) *pb.Mergeability {
	161	+	return &pb.Mergeability{
	162	+	State: mergeabilityStateToPB(m.State),
	163	+	TargetId: m.TargetID.String(),
	164	+	HeadId: m.HeadID.String(),
	165	+	Behind: int32(m.Behind), //nolint:gosec // capped at domain.MaxBehindCount
	166	+	BehindCapped: m.BehindCapped,
	167	+	ConflictPaths: m.ConflictPaths,
	168	+	ConflictPathsTruncated: m.ConflictPathsTruncated,
	169	+	CheckedAt: timestamp(m.CheckedAt),
	170	+	}
	171	+	}
	172	+
136	173		func commentToPB(c domain.Comment) *pb.Comment {
137	174		return &pb.Comment{
138	175		Id: strconv.FormatInt(c.ID, 10),

@@ -214,7 +251,7 @@ func (s mergeRequestService) ListMergeRequests(ctx context.Context, req connec
214	251		}
215	252		out := &pb.ListMergeRequestsResponse{NextPageToken: page.NextPageToken, MergeRequests: make([]*pb.MergeRequest, 0, len(page.MergeRequests))}
216	253		for _, mr := range page.MergeRequests {
217		-	out.MergeRequests = append(out.MergeRequests, mergeRequestToPB(mr))
	254	+	out.MergeRequests = append(out.MergeRequests, mergeRequestWithMergeabilityToPB(mr, page.Mergeability[mr.Number]))
218	255		}
219	256		return connect.NewResponse(out), nil
220	257		}

@@ -229,7 +266,7 @@ func (s mergeRequestService) GetMergeRequest(ctx context.Context, req connect.
229	266		return nil, connectError(ctx, err)
230	267		}
231	268		out := &pb.GetMergeRequestResponse{
232		-	MergeRequest: mergeRequestToPB(d.MergeRequest),
	269	+	MergeRequest: mergeRequestWithMergeabilityToPB(d.MergeRequest, d.Mergeability),
233	270		CommitsTruncated: d.CommitsTruncated,
234	271		DiffBaseId: d.DiffBaseID.String(),
235	272		Stats: diffStatsToPB(d.Stats),


@@ -133,6 +133,43 @@ func mergeRequestToPB(mr domain.MergeRequest) *pb.MergeRequest {
133		}	133		}
134		}	134		}
135			135
			136	+	// mergeRequestWithMergeabilityToPB is mergeRequestToPB with the mergeability of
			137	+	// an open request; m is ignored for a request that is not open.
			138	+	func mergeRequestWithMergeabilityToPB(mr domain.MergeRequest, m domain.Mergeability) *pb.MergeRequest {
			139	+	out := mergeRequestToPB(mr)
			140	+	if mr.IsOpen() && m.State != "" {
			141	+	out.Mergeability = mergeabilityToPB(m)
			142	+	}
			143	+	return out
			144	+	}
			145	+
			146	+	func mergeabilityStateToPB(s domain.MergeabilityState) pb.MergeabilityState {
			147	+	switch s {
			148	+	case domain.MergeabilityChecking:
			149	+	return pb.MergeabilityState_MERGEABILITY_STATE_CHECKING
			150	+	case domain.MergeabilityUpToDate:
			151	+	return pb.MergeabilityState_MERGEABILITY_STATE_UP_TO_DATE
			152	+	case domain.MergeabilityBehind:
			153	+	return pb.MergeabilityState_MERGEABILITY_STATE_BEHIND
			154	+	case domain.MergeabilityConflict:
			155	+	return pb.MergeabilityState_MERGEABILITY_STATE_CONFLICT
			156	+	}
			157	+	return pb.MergeabilityState_MERGEABILITY_STATE_UNSPECIFIED
			158	+	}
			159	+
			160	+	func mergeabilityToPB(m domain.Mergeability) *pb.Mergeability {
			161	+	return &pb.Mergeability{
			162	+	State: mergeabilityStateToPB(m.State),
			163	+	TargetId: m.TargetID.String(),
			164	+	HeadId: m.HeadID.String(),
			165	+	Behind: int32(m.Behind), //nolint:gosec // capped at domain.MaxBehindCount
			166	+	BehindCapped: m.BehindCapped,
			167	+	ConflictPaths: m.ConflictPaths,
			168	+	ConflictPathsTruncated: m.ConflictPathsTruncated,
			169	+	CheckedAt: timestamp(m.CheckedAt),
			170	+	}
			171	+	}
			172	+
136		func commentToPB(c domain.Comment) *pb.Comment {	173		func commentToPB(c domain.Comment) *pb.Comment {
137		return &pb.Comment{	174		return &pb.Comment{
138		Id: strconv.FormatInt(c.ID, 10),	175		Id: strconv.FormatInt(c.ID, 10),

@@ -214,7 +251,7 @@ func (s mergeRequestService) ListMergeRequests(ctx context.Context, req connec
214		}	251		}
215		out := &pb.ListMergeRequestsResponse{NextPageToken: page.NextPageToken, MergeRequests: make([]*pb.MergeRequest, 0, len(page.MergeRequests))}	252		out := &pb.ListMergeRequestsResponse{NextPageToken: page.NextPageToken, MergeRequests: make([]*pb.MergeRequest, 0, len(page.MergeRequests))}
216		for _, mr := range page.MergeRequests {	253		for _, mr := range page.MergeRequests {
217	-	out.MergeRequests = append(out.MergeRequests, mergeRequestToPB(mr))	254	+	out.MergeRequests = append(out.MergeRequests, mergeRequestWithMergeabilityToPB(mr, page.Mergeability[mr.Number]))
218		}	255		}
219		return connect.NewResponse(out), nil	256		return connect.NewResponse(out), nil
220		}	257		}

@@ -229,7 +266,7 @@ func (s mergeRequestService) GetMergeRequest(ctx context.Context, req connect.
229		return nil, connectError(ctx, err)	266		return nil, connectError(ctx, err)
230		}	267		}
231		out := &pb.GetMergeRequestResponse{	268		out := &pb.GetMergeRequestResponse{
232	-	MergeRequest: mergeRequestToPB(d.MergeRequest),	269	+	MergeRequest: mergeRequestWithMergeabilityToPB(d.MergeRequest, d.Mergeability),
233		CommitsTruncated: d.CommitsTruncated,	270		CommitsTruncated: d.CommitsTruncated,
234		DiffBaseId: d.DiffBaseID.String(),	271		DiffBaseId: d.DiffBaseID.String(),
235		Stats: diffStatsToPB(d.Stats),	272		Stats: diffStatsToPB(d.Stats),

M internal/app/api_helpers_test.go +14 -1


@@ -167,8 +167,21 @@ func (h harness) watch(c clients, resume string, topics ...string) *stream {
167	167		return s
168	168		}
169	169
170		-	// next returns the next event or fails after d.
	170	+	// next returns the next event or fails after d. Mergeability events, which
	171	+	// arrive asynchronously after a push or an opening, are skipped: use
	172	+	// nextAny to see them.
171	173		func (s stream) next(t testing.T, d time.Duration) *pb.Event {
	174	+	t.Helper()
	175	+	for {
	176	+	ev := s.nextAny(t, d)
	177	+	if ev.GetMergeRequestMergeability() == nil {
	178	+	return ev
	179	+	}
	180	+	}
	181	+	}
	182	+
	183	+	// nextAny returns the next event, whatever its kind, or fails after d.
	184	+	func (s stream) nextAny(t testing.T, d time.Duration) *pb.Event {
172	185		t.Helper()
173	186		select {
174	187		case ev, ok := <-s.events:


@@ -167,8 +167,21 @@ func (h harness) watch(c clients, resume string, topics ...string) *stream {
167		return s	167		return s
168		}	168		}
169			169
170	-	// next returns the next event or fails after d.	170	+	// next returns the next event or fails after d. Mergeability events, which
			171	+	// arrive asynchronously after a push or an opening, are skipped: use
			172	+	// nextAny to see them.
171		func (s stream) next(t testing.T, d time.Duration) *pb.Event {	173		func (s stream) next(t testing.T, d time.Duration) *pb.Event {
			174	+	t.Helper()
			175	+	for {
			176	+	ev := s.nextAny(t, d)
			177	+	if ev.GetMergeRequestMergeability() == nil {
			178	+	return ev
			179	+	}
			180	+	}
			181	+	}
			182	+
			183	+	// nextAny returns the next event, whatever its kind, or fails after d.
			184	+	func (s stream) nextAny(t testing.T, d time.Duration) *pb.Event {
172		t.Helper()	185		t.Helper()
173		select {	186		select {
174		case ev, ok := <-s.events:	187		case ev, ok := <-s.events:

M internal/app/api_test.go +1 -1


@@ -196,7 +196,7 @@ func TestAC2ReadAccess(t *testing.T) {
196	196		for _, who := range allPrincipals {
197	197		resp, err := r.who[who].server.GetVersion(ctx, connect.NewRequest(&pb.GetVersionRequest{}))
198	198		require.NoError(t, err, who)
199		-	assert.ElementsMatch(t, []string{"repo.update", "events.resume", "merge_requests", "merge_requests.delete_branch", "issues", "signing_key", "session", "git.list_paths", "token.branch_prefixes", "ci.runs", "ci.runs.filters", "ci.runners", "ci.actions", "ci.annotations", "ci.summaries", "ci.concurrency", "ci.dispatch", "ci.workflows", "ci.cache", "ci.secrets", "ci.artifacts", "commit_signatures"}, resp.Msg.GetFeatures(), who)
	199	+	assert.ElementsMatch(t, []string{"repo.update", "events.resume", "merge_requests", "merge_requests.delete_branch", "issues", "merge_requests.mergeability", "signing_key", "session", "git.list_paths", "token.branch_prefixes", "ci.runs", "ci.runs.filters", "ci.runners", "ci.actions", "ci.annotations", "ci.summaries", "ci.concurrency", "ci.dispatch", "ci.workflows", "ci.cache", "ci.secrets", "ci.artifacts", "commit_signatures"}, resp.Msg.GetFeatures(), who)
200	200		assert.NotEmpty(t, resp.Msg.GetVersion())
201	201		}
202	202		})


@@ -196,7 +196,7 @@ func TestAC2ReadAccess(t *testing.T) {
196		for _, who := range allPrincipals {	196		for _, who := range allPrincipals {
197		resp, err := r.who[who].server.GetVersion(ctx, connect.NewRequest(&pb.GetVersionRequest{}))	197		resp, err := r.who[who].server.GetVersion(ctx, connect.NewRequest(&pb.GetVersionRequest{}))
198		require.NoError(t, err, who)	198		require.NoError(t, err, who)
199	-	assert.ElementsMatch(t, []string{"repo.update", "events.resume", "merge_requests", "merge_requests.delete_branch", "issues", "signing_key", "session", "git.list_paths", "token.branch_prefixes", "ci.runs", "ci.runs.filters", "ci.runners", "ci.actions", "ci.annotations", "ci.summaries", "ci.concurrency", "ci.dispatch", "ci.workflows", "ci.cache", "ci.secrets", "ci.artifacts", "commit_signatures"}, resp.Msg.GetFeatures(), who)	199	+	assert.ElementsMatch(t, []string{"repo.update", "events.resume", "merge_requests", "merge_requests.delete_branch", "issues", "merge_requests.mergeability", "signing_key", "session", "git.list_paths", "token.branch_prefixes", "ci.runs", "ci.runs.filters", "ci.runners", "ci.actions", "ci.annotations", "ci.summaries", "ci.concurrency", "ci.dispatch", "ci.workflows", "ci.cache", "ci.secrets", "ci.artifacts", "commit_signatures"}, resp.Msg.GetFeatures(), who)
200		assert.NotEmpty(t, resp.Msg.GetVersion())	200		assert.NotEmpty(t, resp.Msg.GetVersion())
201		}	201		}
202		})	202		})

M internal/app/cli_e2e_test.go +14 -2


@@ -123,8 +123,20 @@ func (c cliHarness) startWatch(t testing.T, secret string, args ...string) *wa
123	123		return w
124	124		}
125	125
126		-	// next decodes the next event line or fails after d.
	126	+	// next returns the next event line or fails after d. Mergeability events,
	127	+	// which arrive asynchronously after a push or an opening, are skipped.
127	128		func (w watchProc) next(t testing.T, d time.Duration) (raw string) {
	129	+	t.Helper()
	130	+	for {
	131	+	l := w.nextAny(t, d)
	132	+	if !strings.Contains(l, `"type":"`+eventMRMergeability+`"`) {
	133	+	return l
	134	+	}
	135	+	}
	136	+	}
	137	+
	138	+	// nextAny returns the next event line, whatever its kind, or fails after d.
	139	+	func (w watchProc) nextAny(t testing.T, d time.Duration) (raw string) {
128	140		t.Helper()
129	141		select {
130	142		case l, ok := <-w.lines:

@@ -338,7 +350,7 @@ func TestAC6CLIEndToEnd(t *testing.T) {
338	350		ver := decode[versionOut](t, res.stdout)
339	351		assert.Equal(t, c.url, ver.Server.URL)
340	352		assert.NotEmpty(t, ver.Server.Version)
341		-	assert.ElementsMatch(t, []string{"repo.update", "events.resume", "merge_requests", "merge_requests.delete_branch", "issues", "signing_key", "session", "git.list_paths", "token.branch_prefixes", "ci.runs", "ci.runs.filters", "ci.runners", "ci.actions", "ci.annotations", "ci.summaries", "ci.concurrency", "ci.dispatch", "ci.workflows", "ci.cache", "ci.secrets", "ci.artifacts", "commit_signatures"}, ver.Server.Features)
	353	+	assert.ElementsMatch(t, []string{"repo.update", "events.resume", "merge_requests", "merge_requests.delete_branch", "issues", "merge_requests.mergeability", "signing_key", "session", "git.list_paths", "token.branch_prefixes", "ci.runs", "ci.runs.filters", "ci.runners", "ci.actions", "ci.annotations", "ci.summaries", "ci.concurrency", "ci.dispatch", "ci.workflows", "ci.cache", "ci.secrets", "ci.artifacts", "commit_signatures"}, ver.Server.Features)
342	354		assert.Equal(t, []string{"passkeys"}, ver.Server.MissingFeatures)
343	355		stdout, stderr, code = c.bin(t, secret, "version")
344	356		require.Equal(t, 0, code, stderr)

 	return w
 }
-// next decodes the next event line or fails after d.
+// next returns the next event line or fails after d. Mergeability events,
+// which arrive asynchronously after a push or an opening, are skipped.
 func (w *watchProc) next(t *testing.T, d time.Duration) (raw string) {
+	t.Helper()
+	for {
+		l := w.nextAny(t, d)
+		if !strings.Contains(l, `"type":"`+eventMRMergeability+`"`) {
+			return l
+		}
+	}
+}
+// nextAny returns the next event line, whatever its kind, or fails after d.
+func (w *watchProc) nextAny(t *testing.T, d time.Duration) (raw string) {
 	t.Helper()
 	select {
 	case l, ok := <-w.lines:
 	ver := decode[versionOut](t, res.stdout)
 	assert.Equal(t, c.url, ver.Server.URL)
 	assert.NotEmpty(t, ver.Server.Version)
-	assert.ElementsMatch(t, []string{"repo.update", "events.resume", "merge_requests", "merge_requests.delete_branch", "issues", "signing_key", "session", "git.list_paths", "token.branch_prefixes", "ci.runs", "ci.runs.filters", "ci.runners", "ci.actions", "ci.annotations", "ci.summaries", "ci.concurrency", "ci.dispatch", "ci.workflows", "ci.cache", "ci.secrets", "ci.artifacts", "commit_signatures"}, ver.Server.Features)
+	assert.ElementsMatch(t, []string{"repo.update", "events.resume", "merge_requests", "merge_requests.delete_branch", "issues", "merge_requests.mergeability", "signing_key", "session", "git.list_paths", "token.branch_prefixes", "ci.runs", "ci.runs.filters", "ci.runners", "ci.actions", "ci.annotations", "ci.summaries", "ci.concurrency", "ci.dispatch", "ci.workflows", "ci.cache", "ci.secrets", "ci.artifacts", "commit_signatures"}, ver.Server.Features)
 	assert.Equal(t, []string{"passkeys"}, ver.Server.MissingFeatures)
 	stdout, stderr, code = c.bin(t, secret, "version")
 	require.Equal(t, 0, code, stderr)

M internal/app/cli_watch.go +8 -0


@@ -29,6 +29,8 @@ type jsonEvent struct {
29	29		MergeRequest *jsonMergeRequest `json:"merge_request,omitempty"`
30	30		Number int64 `json:"number,omitempty"`
31	31		Comment *jsonComment `json:"comment,omitempty"`
	32	+	// Mergeability is the state of a merge_request_mergeability event.
	33	+	Mergeability string `json:"mergeability,omitempty"`
32	34		}
33	35
34	36		type jsonRefUpdate struct {

@@ -52,6 +54,7 @@ const (
52	54		eventCommentEdited = "comment_edited"
53	55		eventCommentDeleted = "comment_deleted"
54	56		eventCommentResolved = "comment_resolved"
	57	+	eventMRMergeability = "merge_request_mergeability"
55	58		)
56	59
57	60		// toJSONEvent maps an event; ok is false for kinds this client does not know,

@@ -94,6 +97,9 @@ func toJSONEvent(ev *ermite.Event, baseURL string) (jsonEvent, bool) {
94	97		case *ermite.Event_CommentResolved:
95	98		c := toJSONComment(k.CommentResolved.GetComment())
96	99		out.Type, out.Repo, out.Number, out.Comment = eventCommentResolved, k.CommentResolved.GetRepo(), k.CommentResolved.GetNumber(), &c
	100	+	case *ermite.Event_MergeRequestMergeability:
	101	+	out.Type, out.Repo, out.Number = eventMRMergeability, k.MergeRequestMergeability.GetRepo(), k.MergeRequestMergeability.GetNumber()
	102	+	out.Mergeability = k.MergeRequestMergeability.GetMergeability().GetState().String()
97	103		case *ermite.Event_Gap:
98	104		out.Type = eventGap
99	105		default:

@@ -126,6 +132,8 @@ func eventLine(e jsonEvent) string {
126	132		detail = fmt.Sprintf("%s!%d %s -> %s", e.Repo, e.MergeRequest.Number, e.MergeRequest.SourceBranch, e.MergeRequest.TargetBranch)
127	133		case eventCommentAdded, eventCommentEdited, eventCommentDeleted, eventCommentResolved:
128	134		detail = fmt.Sprintf("%s!%d by %s", e.Repo, e.Number, e.Comment.Author)
	135	+	case eventMRMergeability:
	136	+	detail = fmt.Sprintf("%s!%d %s", e.Repo, e.Number, e.Mergeability)
129	137		case eventGap:
130	138		detail = "events were lost, reload what you display"
131	139		}


@@ -29,6 +29,8 @@ type jsonEvent struct {
29	MergeRequest *jsonMergeRequest `json:"merge_request,omitempty"`	29		MergeRequest *jsonMergeRequest `json:"merge_request,omitempty"`
30	Number int64 `json:"number,omitempty"`	30		Number int64 `json:"number,omitempty"`
31	Comment *jsonComment `json:"comment,omitempty"`	31		Comment *jsonComment `json:"comment,omitempty"`
		32	+	// Mergeability is the state of a merge_request_mergeability event.
		33	+	Mergeability string `json:"mergeability,omitempty"`
32	}	34		}
33		35
34	type jsonRefUpdate struct {	36		type jsonRefUpdate struct {

@@ -52,6 +54,7 @@ const (
52	eventCommentEdited = "comment_edited"	54		eventCommentEdited = "comment_edited"
53	eventCommentDeleted = "comment_deleted"	55		eventCommentDeleted = "comment_deleted"
54	eventCommentResolved = "comment_resolved"	56		eventCommentResolved = "comment_resolved"
		57	+	eventMRMergeability = "merge_request_mergeability"
55	)	58		)
56		59
57	// toJSONEvent maps an event; ok is false for kinds this client does not know,	60		// toJSONEvent maps an event; ok is false for kinds this client does not know,

@@ -94,6 +97,9 @@ func toJSONEvent(ev *ermite.Event, baseURL string) (jsonEvent, bool) {
94	case *ermite.Event_CommentResolved:	97		case *ermite.Event_CommentResolved:
95	c := toJSONComment(k.CommentResolved.GetComment())	98		c := toJSONComment(k.CommentResolved.GetComment())
96	out.Type, out.Repo, out.Number, out.Comment = eventCommentResolved, k.CommentResolved.GetRepo(), k.CommentResolved.GetNumber(), &c	99		out.Type, out.Repo, out.Number, out.Comment = eventCommentResolved, k.CommentResolved.GetRepo(), k.CommentResolved.GetNumber(), &c
		100	+	case *ermite.Event_MergeRequestMergeability:
		101	+	out.Type, out.Repo, out.Number = eventMRMergeability, k.MergeRequestMergeability.GetRepo(), k.MergeRequestMergeability.GetNumber()
		102	+	out.Mergeability = k.MergeRequestMergeability.GetMergeability().GetState().String()
97	case *ermite.Event_Gap:	103		case *ermite.Event_Gap:
98	out.Type = eventGap	104		out.Type = eventGap
99	default:	105		default:

@@ -126,6 +132,8 @@ func eventLine(e jsonEvent) string {
126	detail = fmt.Sprintf("%s!%d %s -> %s", e.Repo, e.MergeRequest.Number, e.MergeRequest.SourceBranch, e.MergeRequest.TargetBranch)	132		detail = fmt.Sprintf("%s!%d %s -> %s", e.Repo, e.MergeRequest.Number, e.MergeRequest.SourceBranch, e.MergeRequest.TargetBranch)
127	case eventCommentAdded, eventCommentEdited, eventCommentDeleted, eventCommentResolved:	133		case eventCommentAdded, eventCommentEdited, eventCommentDeleted, eventCommentResolved:
128	detail = fmt.Sprintf("%s!%d by %s", e.Repo, e.Number, e.Comment.Author)	134		detail = fmt.Sprintf("%s!%d by %s", e.Repo, e.Number, e.Comment.Author)
		135	+	case eventMRMergeability:
		136	+	detail = fmt.Sprintf("%s!%d %s", e.Repo, e.Number, e.Mergeability)
129	case eventGap:	137		case eventGap:
130	detail = "events were lost, reload what you display"	138		detail = "events were lost, reload what you display"
131	}	139		}

M internal/app/forge.go +11 -1


@@ -137,6 +137,8 @@ type forge struct {
137	137		// Inbox serves the owner's inbox of the web.
138	138		Inbox *usecase.Inbox
139	139		SigningKey *usecase.SigningKey
	140	+	// Mergeability keeps the mergeability of the open merge requests up to date.
	141	+	Mergeability *usecase.Mergeability
140	142		// Signatures are the trusted keys and the commit signature checks
141	143		// (ADR 0030).
142	144		Signatures *usecase.Signatures

@@ -331,11 +333,16 @@ func newForge(ctx context.Context, cfg config.Config, opts options) (_ *forge, e
331	333		if err := f.signatures(store, reader, signer.PublicKey(), committer, clock); err != nil {
332	334		return nil, err
333	335		}
	336	+	f.Mergeability = usecase.NewMergeability(ctx, usecase.MergeabilityConfig{
	337	+	Store: store.MergeRequests(), Meta: store, Git: git, Bus: f.Bus, Clock: clock,
	338	+	})
	339	+	f.Pushes.WithMergeability(f.Mergeability)
334	340		f.MergeRequests = usecase.NewMergeRequests(usecase.MergeRequestsConfig{
335	341		Meta: store, Store: store.MergeRequests(), Repos: git, Git: git, Signer: signer,
336	342		Committer: committer, Bus: f.Bus, Clock: clock, Maint: f.maint,
337		-	CI: f.CI, Checks: f.CI, PullRequests: f.CI, RefTips: store.CI(),
	343	+	CI: f.CI, Checks: f.CI, PullRequests: f.CI, RefTips: store.CI(), Mergeability: f.Mergeability,
338	344		})
	345	+	f.Mergeability.Start()
339	346		f.MergeRequests.WithVerifier(f.Signatures)
340	347		f.Issues = usecase.NewIssues(usecase.IssuesConfig{
341	348		Meta: store, Store: store.Issues(), MergeRequests: store.MergeRequests(), Bus: f.Bus, Clock: clock,

@@ -515,6 +522,9 @@ func (f *forge) Close(ctx context.Context) {
515	522		if f.CI != nil {
516	523		f.CI.Close(ctx)
517	524		}
	525	+	if f.Mergeability != nil {
	526	+	f.Mergeability.Close(ctx)
	527	+	}
518	528		if f.EventLog != nil {
519	529		f.EventLog.Close(ctx)
520	530		}

 	// Inbox serves the owner's inbox of the web.
 	Inbox      *usecase.Inbox
 	SigningKey *usecase.SigningKey
+	// Mergeability keeps the mergeability of the open merge requests up to date.
+	Mergeability *usecase.Mergeability
 	// Signatures are the trusted keys and the commit signature checks
 	// (ADR 0030).
 	Signatures *usecase.Signatures
 	if err := f.signatures(store, reader, signer.PublicKey(), committer, clock); err != nil {
 		return nil, err
 	}
+	f.Mergeability = usecase.NewMergeability(ctx, usecase.MergeabilityConfig{
+		Store: store.MergeRequests(), Meta: store, Git: git, Bus: f.Bus, Clock: clock,
+	})
+	f.Pushes.WithMergeability(f.Mergeability)
 	f.MergeRequests = usecase.NewMergeRequests(usecase.MergeRequestsConfig{
 		Meta: store, Store: store.MergeRequests(), Repos: git, Git: git, Signer: signer,
 		Committer: committer, Bus: f.Bus, Clock: clock, Maint: f.maint,
-		CI: f.CI, Checks: f.CI, PullRequests: f.CI, RefTips: store.CI(),
+		CI: f.CI, Checks: f.CI, PullRequests: f.CI, RefTips: store.CI(), Mergeability: f.Mergeability,
 	})
+	f.Mergeability.Start()
 	f.MergeRequests.WithVerifier(f.Signatures)
 	f.Issues = usecase.NewIssues(usecase.IssuesConfig{
 		Meta: store, Store: store.Issues(), MergeRequests: store.MergeRequests(), Bus: f.Bus, Clock: clock,
 	if f.CI != nil {
 		f.CI.Close(ctx)
 	}
+	if f.Mergeability != nil {
+		f.Mergeability.Close(ctx)
+	}
 	if f.EventLog != nil {
 		f.EventLog.Close(ctx)
 	}

M internal/ports/dto.go +3 -0


@@ -198,6 +198,9 @@ type MergeRequestPage struct {
198	198		MergeRequests []domain.MergeRequest
199	199		// NextPageToken is empty on the last page.
200	200		NextPageToken string
	201	+	// Mergeability holds, by number, the mergeability of the open requests of
	202	+	// the page (nil when the server does not compute it).
	203	+	Mergeability map[int64]domain.Mergeability
201	204		}
202	205
203	206		// ListIssuesParams selects a page of issues from an IssueStore.


@@ -198,6 +198,9 @@ type MergeRequestPage struct {
198	MergeRequests []domain.MergeRequest	198		MergeRequests []domain.MergeRequest
199	// NextPageToken is empty on the last page.	199		// NextPageToken is empty on the last page.
200	NextPageToken string	200		NextPageToken string
		201	+	// Mergeability holds, by number, the mergeability of the open requests of
		202	+	// the page (nil when the server does not compute it).
		203	+	Mergeability map[int64]domain.Mergeability
201	}	204		}
202		205
203	// ListIssuesParams selects a page of issues from an IssueStore.	206		// ListIssuesParams selects a page of issues from an IssueStore.

M internal/usecase/events.go +4 -0


@@ -198,6 +198,8 @@ func (e *Events) snapshot(ctx context.Context, ev domain.Event) bufferedEvent {
198	198		return repoSnapshot(ev.Visibility, ev.CreatedAt)
199	199		case domain.CommentResolved:
200	200		return repoSnapshot(ev.Visibility, ev.CreatedAt)
	201	+	case domain.MergeRequestMergeability:
	202	+	return repoSnapshot(ev.Visibility, ev.CreatedAt)
201	203		case domain.RunUpdated:
202	204		return repoSnapshot(ev.Visibility, ev.CreatedAt)
203	205		case domain.JobUpdated:

@@ -541,6 +543,8 @@ func (e Events) deliver(ctx context.Context, w watch, be bufferedEvent, send f
541	543		return e.deliverRepo(ctx, w, ev.Repo, false, be, out, send)
542	544		case domain.CommentResolved:
543	545		return e.deliverRepo(ctx, w, ev.Repo, false, be, out, send)
	546	+	case domain.MergeRequestMergeability:
	547	+	return e.deliverRepo(ctx, w, ev.Repo, false, be, out, send)
544	548		case domain.RunUpdated:
545	549		// Home shows the CI state of each repository.
546	550		return e.deliverRepo(ctx, w, ev.Run.Repo, false, be, out, send)


@@ -198,6 +198,8 @@ func (e *Events) snapshot(ctx context.Context, ev domain.Event) bufferedEvent {
198	return repoSnapshot(ev.Visibility, ev.CreatedAt)	198		return repoSnapshot(ev.Visibility, ev.CreatedAt)
199	case domain.CommentResolved:	199		case domain.CommentResolved:
200	return repoSnapshot(ev.Visibility, ev.CreatedAt)	200		return repoSnapshot(ev.Visibility, ev.CreatedAt)
		201	+	case domain.MergeRequestMergeability:
		202	+	return repoSnapshot(ev.Visibility, ev.CreatedAt)
201	case domain.RunUpdated:	203		case domain.RunUpdated:
202	return repoSnapshot(ev.Visibility, ev.CreatedAt)	204		return repoSnapshot(ev.Visibility, ev.CreatedAt)
203	case domain.JobUpdated:	205		case domain.JobUpdated:

@@ -541,6 +543,8 @@ func (e Events) deliver(ctx context.Context, w watch, be bufferedEvent, send f
541	return e.deliverRepo(ctx, w, ev.Repo, false, be, out, send)	543		return e.deliverRepo(ctx, w, ev.Repo, false, be, out, send)
542	case domain.CommentResolved:	544		case domain.CommentResolved:
543	return e.deliverRepo(ctx, w, ev.Repo, false, be, out, send)	545		return e.deliverRepo(ctx, w, ev.Repo, false, be, out, send)
		546	+	case domain.MergeRequestMergeability:
		547	+	return e.deliverRepo(ctx, w, ev.Repo, false, be, out, send)
544	case domain.RunUpdated:	548		case domain.RunUpdated:
545	// Home shows the CI state of each repository.	549		// Home shows the CI state of each repository.
546	return e.deliverRepo(ctx, w, ev.Run.Repo, false, be, out, send)	550		return e.deliverRepo(ctx, w, ev.Run.Repo, false, be, out, send)

A internal/usecase/mergeability.go +359 -0

@@ -0,0 +1,359 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package usecase
	4	+
	5	+	import (
	6	+	"context"
	7	+	"errors"
	8	+	"slices"
	9	+	"strings"
	10	+	"sync"
	11	+
	12	+	"bnema.dev/ermite/internal/domain"
	13	+	"bnema.dev/ermite/internal/ports"
	14	+	)
	15	+
	16	+	const (
	17	+	defaultMergeabilityWorkers = 2
	18	+	defaultMergeabilityQueue = 1024
	19	+	// mergeabilityAttempts bounds the computations of one check that lose a
	20	+	// race with a moving branch or a concurrent write; the check is then queued
	21	+	// again.
	22	+	mergeabilityAttempts = 3
	23	+	)
	24	+
	25	+	// MergeabilityConfig holds what the mergeability service needs. Workers and
	26	+	// Queue default to 2 and 1024.
	27	+	type MergeabilityConfig struct {
	28	+	Store ports.MergeRequestStore
	29	+	Meta ports.RepoMetaStore
	30	+	Git ports.MergeGit
	31	+	Bus ports.EventBus
	32	+	Clock ports.Clock
	33	+	// Workers is the number of checks that run at once, Queue the number of
	34	+	// checks that may wait; past it a repository is rescanned.
	35	+	Workers, Queue int
	36	+	}
	37	+
	38	+	// Mergeability keeps up to date whether each open merge request merges with
	39	+	// its target (see domain.Mergeability). It is independent of CI: it never
	40	+	// touches refs/pull, workflows or runs. It computes with Git reads only, holds
	41	+	// no repository lock, and stores its result with a compare-and-swap.
	42	+	type Mergeability struct {
	43	+	cfg MergeabilityConfig
	44	+	q *mergeQueue
	45	+	ctx context.Context
	46	+	cancel context.CancelFunc
	47	+	wg sync.WaitGroup
	48	+	}
	49	+
	50	+	// MergeabilityHook is what the merge request use cases need from the
	51	+	// mergeability service: queue checks, read results and drop the record of a
	52	+	// request that is no longer open.
	53	+	type MergeabilityHook interface {
	54	+	ports.MergeabilityTrigger
	55	+	ports.MergeabilityReader
	56	+	// OnClosed deletes the record of a merge request that left the open state.
	57	+	OnClosed(ctx context.Context, mr domain.MergeRequest)
	58	+	}
	59	+
	60	+	var _ MergeabilityHook = (*Mergeability)(nil)
	61	+
	62	+	// NewMergeability builds the service. Call Start to run it, Close to stop it.
	63	+	func NewMergeability(ctx context.Context, cfg MergeabilityConfig) *Mergeability {
	64	+	if cfg.Workers <= 0 {
	65	+	cfg.Workers = defaultMergeabilityWorkers
	66	+	}
	67	+	if cfg.Queue <= 0 {
	68	+	cfg.Queue = defaultMergeabilityQueue
	69	+	}
	70	+	wctx, cancel := context.WithCancel(ctx)
	71	+	return &Mergeability{cfg: cfg, q: newMergeQueue(cfg.Queue), ctx: wctx, cancel: cancel}
	72	+	}
	73	+
	74	+	// Start rescans the open merge requests of every repository (the records may
	75	+	// be missing or stale after a restart) and starts the workers.
	76	+	func (s *Mergeability) Start() {
	77	+	s.rescanAll()
	78	+	for range s.cfg.Workers {
	79	+	s.wg.Add(1)
	80	+	go s.loop()
	81	+	}
	82	+	}
	83	+
	84	+	// Close stops the workers and waits for them.
	85	+	func (s *Mergeability) Close(_ context.Context) {
	86	+	s.cancel()
	87	+	s.wg.Wait()
	88	+	}
	89	+
	90	+	func (s *Mergeability) rescanAll() {
	91	+	repos, err := s.cfg.Meta.List(s.ctx)
	92	+	if err != nil {
	93	+	ctxLog(s.ctx).Error().Err(err).Msg("mergeability: list the repositories to rescan")
	94	+	return
	95	+	}
	96	+	for _, r := range repos {
	97	+	n, err := s.cfg.Store.CountOpen(s.ctx, r.Name)
	98	+	if err != nil {
	99	+	ctxLog(s.ctx).Error().Err(err).Str("repo", r.Name.String()).Msg("mergeability: count the open merge requests")
	100	+	s.q.markDirty(r.Name)
	101	+	continue
	102	+	}
	103	+	if n > 0 {
	104	+	s.q.markDirty(r.Name)
	105	+	}
	106	+	}
	107	+	}
	108	+
	109	+	// OnOpened implements ports.MergeabilityTrigger.
	110	+	func (s *Mergeability) OnOpened(_ context.Context, mr domain.MergeRequest) {
	111	+	s.q.add(mergeKey{mr.Repo, mr.Number})
	112	+	}
	113	+
	114	+	// OnClosed deletes the record of a merge request that is no longer open. The
	115	+	// store already drops it when the request leaves the open state; this makes it
	116	+	// explicit and is idempotent.
	117	+	func (s *Mergeability) OnClosed(ctx context.Context, mr domain.MergeRequest) {
	118	+	if err := s.cfg.Store.DeleteMergeability(context.WithoutCancel(ctx), mr.Repo, mr.Number); err != nil {
	119	+	ctxLog(ctx).Error().Err(err).Str("repo", mr.Repo.String()).Int64("number", mr.Number).Msg("mergeability: delete the record of a closed merge request")
	120	+	}
	121	+	}
	122	+
	123	+	// OnBranchesMoved implements ports.MergeabilityTrigger: the open merge
	124	+	// requests whose source or target is one of the branches are queued. Past the
	125	+	// bound, or when the lookup fails, the repository is rescanned.
	126	+	func (s *Mergeability) OnBranchesMoved(ctx context.Context, repo domain.RepoName, updates []domain.RefUpdate) {
	127	+	var branches []string
	128	+	for _, u := range updates {
	129	+	if b, ok := strings.CutPrefix(u.Ref, "refs/heads/"); ok && !slices.Contains(branches, b) {
	130	+	branches = append(branches, b)
	131	+	}
	132	+	}
	133	+	if len(branches) == 0 {
	134	+	return
	135	+	}
	136	+	lookups := []func(context.Context, domain.RepoName, []string, int) ([]int64, bool, error){
	137	+	s.cfg.Store.OpenBySource, s.cfg.Store.OpenByTarget,
	138	+	}
	139	+	for _, lookup := range lookups {
	140	+	numbers, truncated, err := lookup(ctx, repo, branches, s.cfg.Queue)
	141	+	if err != nil \|\| truncated {
	142	+	if err != nil {
	143	+	ctxLog(ctx).Error().Err(err).Str("repo", repo.String()).Msg("mergeability: look up the merge requests of moved branches: the repository is rescanned")
	144	+	}
	145	+	s.q.markDirty(repo)
	146	+	}
	147	+	for _, n := range numbers {
	148	+	s.q.add(mergeKey{repo, n})
	149	+	}
	150	+	}
	151	+	}
	152	+
	153	+	// Read implements ports.MergeabilityReader.
	154	+	func (s *Mergeability) Read(ctx context.Context, mr domain.MergeRequest, targetTip domain.ObjectID) domain.Mergeability {
	155	+	checking := domain.Mergeability{Repo: mr.Repo, Number: mr.Number, State: domain.MergeabilityChecking}
	156	+	if mr.HeadID.IsZero() \|\| targetTip.IsZero() {
	157	+	return checking // a branch is gone: there is nothing to check
	158	+	}
	159	+	rec, err := s.cfg.Store.GetMergeability(ctx, mr.Repo, mr.Number)
	160	+	if err == nil && rec.IsFor(targetTip, mr.HeadID) {
	161	+	return rec
	162	+	}
	163	+	k := mergeKey{mr.Repo, mr.Number}
	164	+	s.q.markSeen(k)
	165	+	s.q.add(k)
	166	+	return checking
	167	+	}
	168	+
	169	+	func (s *Mergeability) loop() {
	170	+	defer s.wg.Done()
	171	+	for {
	172	+	key, dirty := s.q.next()
	173	+	switch {
	174	+	case key != nil:
	175	+	s.runKey(*key)
	176	+	case dirty != nil:
	177	+	s.rescan(*dirty)
	178	+	default:
	179	+	select {
	180	+	case <-s.ctx.Done():
	181	+	return
	182	+	case <-s.q.wake:
	183	+	}
	184	+	}
	185	+	if s.ctx.Err() != nil {
	186	+	return
	187	+	}
	188	+	}
	189	+	}
	190	+
	191	+	// runKey checks k, which is marked running, as often as it was queued again
	192	+	// meanwhile.
	193	+	func (s *Mergeability) runKey(k mergeKey) {
	194	+	for {
	195	+	s.check(k)
	196	+	if !s.q.finish(k) \|\| s.ctx.Err() != nil {
	197	+	return
	198	+	}
	199	+	}
	200	+	}
	201	+
	202	+	// rescan checks every open merge request of a repository, here and not through
	203	+	// the bounded queue, so that a rescan never marks its repository dirty again.
	204	+	func (s *Mergeability) rescan(repo domain.RepoName) {
	205	+	mrs, err := s.cfg.Store.ListOpen(s.ctx, repo)
	206	+	if err != nil {
	207	+	ctxLog(s.ctx).Error().Err(err).Str("repo", repo.String()).Msg("mergeability: list the open merge requests to rescan")
	208	+	return
	209	+	}
	210	+	for _, mr := range mrs {
	211	+	if s.ctx.Err() != nil {
	212	+	return
	213	+	}
	214	+	k := mergeKey{repo, mr.Number}
	215	+	if s.q.tryRun(k) {
	216	+	s.runKey(k)
	217	+	}
	218	+	}
	219	+	}
	220	+
	221	+	// Check computes the mergeability of a merge request now, on the calling
	222	+	// goroutine, as a worker does. It is for tests; the server never calls it.
	223	+	func (s *Mergeability) Check(repo domain.RepoName, number int64) {
	224	+	k := mergeKey{repo, number}
	225	+	if s.q.tryRun(k) {
	226	+	s.runKey(k)
	227	+	}
	228	+	}
	229	+
	230	+	// check brings the record of k up to date.
	231	+	func (s *Mergeability) check(k mergeKey) {
	232	+	ctx := s.ctx
	233	+	log := ctxLog(ctx).With().Str("repo", k.repo.String()).Int64("number", k.number).Logger()
	234	+	for range mergeabilityAttempts {
	235	+	mr, err := s.cfg.Store.Get(ctx, k.repo, k.number)
	236	+	if err != nil {
	237	+	if !errors.Is(err, domain.ErrNotFound) && ctx.Err() == nil {
	238	+	log.Error().Err(err).Msg("mergeability: read the merge request")
	239	+	}
	240	+	return
	241	+	}
	242	+	if !mr.IsOpen() {
	243	+	return // the store dropped the record when it left the open state
	244	+	}
	245	+	head, tip, ok := s.tips(ctx, mr)
	246	+	if !ok {
	247	+	return
	248	+	}
	249	+	prev, err := s.cfg.Store.GetMergeability(ctx, k.repo, k.number)
	250	+	if err != nil && !errors.Is(err, domain.ErrNotFound) {
	251	+	log.Error().Err(err).Msg("mergeability: read the record")
	252	+	return
	253	+	}
	254	+	if prev.IsFor(tip, head) {
	255	+	s.q.takeSeen(k)
	256	+	return
	257	+	}
	258	+	res, err := s.compute(ctx, k, tip, head)
	259	+	if err != nil {
	260	+	if ctx.Err() == nil {
	261	+	log.Error().Err(err).Msg("mergeability: compute")
	262	+	}
	263	+	return
	264	+	}
	265	+	// The branches may have moved during the computation: the result is
	266	+	// then for tips nobody asks about, and is dropped.
	267	+	if h2, t2, ok := s.tips(ctx, mr); !ok {
	268	+	return
	269	+	} else if h2 != head \|\| t2 != tip {
	270	+	continue
	271	+	}
	272	+	res.Version = prev.Version
	273	+	switch err := s.cfg.Store.PutMergeability(ctx, res); {
	274	+	case errors.Is(err, domain.ErrConflict):
	275	+	continue
	276	+	case errors.Is(err, domain.ErrNotFound):
	277	+	return // closed meanwhile
	278	+	case err != nil:
	279	+	log.Error().Err(err).Msg("mergeability: store the record")
	280	+	return
	281	+	}
	282	+	seen := s.q.takeSeen(k)
	283	+	if seen \|\| prev.State == "" \|\| changedMergeability(prev, res) {
	284	+	s.publish(ctx, res)
	285	+	}
	286	+	return
	287	+	}
	288	+	s.q.add(k) // kept losing races: try again later
	289	+	}
	290	+
	291	+	// tips reads the heads of the source and the target of mr. ok is false when
	292	+	// the check is over: a branch is gone (the record is deleted) or Git failed.
	293	+	func (s *Mergeability) tips(ctx context.Context, mr domain.MergeRequest) (head, tip domain.ObjectID, ok bool) {
	294	+	head, herr := s.cfg.Git.BranchHead(ctx, mr.Repo, mr.Source)
	295	+	tip, terr := s.cfg.Git.BranchHead(ctx, mr.Repo, mr.Target)
	296	+	switch {
	297	+	case errors.Is(herr, domain.ErrNotFound) \|\| errors.Is(terr, domain.ErrNotFound):
	298	+	if err := s.cfg.Store.DeleteMergeability(ctx, mr.Repo, mr.Number); err != nil && ctx.Err() == nil {
	299	+	ctxLog(ctx).Error().Err(err).Str("repo", mr.Repo.String()).Int64("number", mr.Number).Msg("mergeability: delete the record of a merge request without a branch")
	300	+	}
	301	+	return domain.ObjectID{}, domain.ObjectID{}, false
	302	+	case herr != nil \|\| terr != nil:
	303	+	if ctx.Err() == nil {
	304	+	ctxLog(ctx).Error().Err(errors.Join(herr, terr)).Str("repo", mr.Repo.String()).Int64("number", mr.Number).Msg("mergeability: read the branches")
	305	+	}
	306	+	return domain.ObjectID{}, domain.ObjectID{}, false
	307	+	}
	308	+	return head, tip, true
	309	+	}
	310	+
	311	+	// compute runs the Git reads that decide the state of head against tip.
	312	+	func (s *Mergeability) compute(ctx context.Context, k mergeKey, tip, head domain.ObjectID) (domain.Mergeability, error) {
	313	+	m := domain.Mergeability{Repo: k.repo, Number: k.number, TargetID: tip, HeadID: head, CheckedAt: s.cfg.Clock.Now().UTC()}
	314	+	ancestor := tip == head
	315	+	if !ancestor {
	316	+	var err error
	317	+	if ancestor, err = s.cfg.Git.IsAncestor(ctx, k.repo, tip, head); err != nil {
	318	+	return m, err
	319	+	}
	320	+	}
	321	+	if ancestor {
	322	+	m.State = domain.MergeabilityUpToDate
	323	+	return m, nil
	324	+	}
	325	+	tree, err := s.cfg.Git.MergeTree(ctx, k.repo, tip, head)
	326	+	if err != nil {
	327	+	return m, err
	328	+	}
	329	+	if !tree.Clean {
	330	+	m.State = domain.MergeabilityConflict
	331	+	m.ConflictPaths, m.ConflictPathsTruncated = tree.ConflictPaths, tree.Truncated
	332	+	return m, nil
	333	+	}
	334	+	m.State = domain.MergeabilityBehind
	335	+	if m.Behind, m.BehindCapped, err = s.cfg.Git.CountBehind(ctx, k.repo, head, tip, domain.MaxBehindCount); err != nil {
	336	+	return m, err
	337	+	}
	338	+	return m, nil
	339	+	}
	340	+
	341	+	// changedMergeability reports whether what a reader shows differs.
	342	+	func changedMergeability(a, b domain.Mergeability) bool {
	343	+	return a.State != b.State \|\| a.Behind != b.Behind \|\| a.BehindCapped != b.BehindCapped \|\|
	344	+	a.ConflictPathsTruncated != b.ConflictPathsTruncated \|\| !slices.Equal(a.ConflictPaths, b.ConflictPaths)
	345	+	}
	346	+
	347	+	func (s *Mergeability) publish(ctx context.Context, m domain.Mergeability) {
	348	+	repo, err := s.cfg.Meta.Get(ctx, m.Repo)
	349	+	if err != nil {
	350	+	if !errors.Is(err, domain.ErrNotFound) && ctx.Err() == nil {
	351	+	ctxLog(ctx).Error().Err(err).Str("repo", m.Repo.String()).Msg("mergeability: read the repository to publish a result")
	352	+	}
	353	+	return
	354	+	}
	355	+	s.cfg.Bus.Publish(ctx, domain.MergeRequestMergeability{
	356	+	Repo: m.Repo, Number: m.Number, Mergeability: m, At: m.CheckedAt,
	357	+	Visibility: repo.Visibility, CreatedAt: repo.CreatedAt,
	358	+	})
	359	+	}

@@ -0,0 +1,359 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package usecase
	4	+
	5	+	import (
	6	+	"context"
	7	+	"errors"
	8	+	"slices"
	9	+	"strings"
	10	+	"sync"
	11	+
	12	+	"bnema.dev/ermite/internal/domain"
	13	+	"bnema.dev/ermite/internal/ports"
	14	+	)
	15	+
	16	+	const (
	17	+	defaultMergeabilityWorkers = 2
	18	+	defaultMergeabilityQueue = 1024
	19	+	// mergeabilityAttempts bounds the computations of one check that lose a
	20	+	// race with a moving branch or a concurrent write; the check is then queued
	21	+	// again.
	22	+	mergeabilityAttempts = 3
	23	+	)
	24	+
	25	+	// MergeabilityConfig holds what the mergeability service needs. Workers and
	26	+	// Queue default to 2 and 1024.
	27	+	type MergeabilityConfig struct {
	28	+	Store ports.MergeRequestStore
	29	+	Meta ports.RepoMetaStore
	30	+	Git ports.MergeGit
	31	+	Bus ports.EventBus
	32	+	Clock ports.Clock
	33	+	// Workers is the number of checks that run at once, Queue the number of
	34	+	// checks that may wait; past it a repository is rescanned.
	35	+	Workers, Queue int
	36	+	}
	37	+
	38	+	// Mergeability keeps up to date whether each open merge request merges with
	39	+	// its target (see domain.Mergeability). It is independent of CI: it never
	40	+	// touches refs/pull, workflows or runs. It computes with Git reads only, holds
	41	+	// no repository lock, and stores its result with a compare-and-swap.
	42	+	type Mergeability struct {
	43	+	cfg MergeabilityConfig
	44	+	q *mergeQueue
	45	+	ctx context.Context
	46	+	cancel context.CancelFunc
	47	+	wg sync.WaitGroup
	48	+	}
	49	+
	50	+	// MergeabilityHook is what the merge request use cases need from the
	51	+	// mergeability service: queue checks, read results and drop the record of a
	52	+	// request that is no longer open.
	53	+	type MergeabilityHook interface {
	54	+	ports.MergeabilityTrigger
	55	+	ports.MergeabilityReader
	56	+	// OnClosed deletes the record of a merge request that left the open state.
	57	+	OnClosed(ctx context.Context, mr domain.MergeRequest)
	58	+	}
	59	+
	60	+	var _ MergeabilityHook = (*Mergeability)(nil)
	61	+
	62	+	// NewMergeability builds the service. Call Start to run it, Close to stop it.
	63	+	func NewMergeability(ctx context.Context, cfg MergeabilityConfig) *Mergeability {
	64	+	if cfg.Workers <= 0 {
	65	+	cfg.Workers = defaultMergeabilityWorkers
	66	+	}
	67	+	if cfg.Queue <= 0 {
	68	+	cfg.Queue = defaultMergeabilityQueue
	69	+	}
	70	+	wctx, cancel := context.WithCancel(ctx)
	71	+	return &Mergeability{cfg: cfg, q: newMergeQueue(cfg.Queue), ctx: wctx, cancel: cancel}
	72	+	}
	73	+
	74	+	// Start rescans the open merge requests of every repository (the records may
	75	+	// be missing or stale after a restart) and starts the workers.
	76	+	func (s *Mergeability) Start() {
	77	+	s.rescanAll()
	78	+	for range s.cfg.Workers {
	79	+	s.wg.Add(1)
	80	+	go s.loop()
	81	+	}
	82	+	}
	83	+
	84	+	// Close stops the workers and waits for them.
	85	+	func (s *Mergeability) Close(_ context.Context) {
	86	+	s.cancel()
	87	+	s.wg.Wait()
	88	+	}
	89	+
	90	+	func (s *Mergeability) rescanAll() {
	91	+	repos, err := s.cfg.Meta.List(s.ctx)
	92	+	if err != nil {
	93	+	ctxLog(s.ctx).Error().Err(err).Msg("mergeability: list the repositories to rescan")
	94	+	return
	95	+	}
	96	+	for _, r := range repos {
	97	+	n, err := s.cfg.Store.CountOpen(s.ctx, r.Name)
	98	+	if err != nil {
	99	+	ctxLog(s.ctx).Error().Err(err).Str("repo", r.Name.String()).Msg("mergeability: count the open merge requests")
	100	+	s.q.markDirty(r.Name)
	101	+	continue
	102	+	}
	103	+	if n > 0 {
	104	+	s.q.markDirty(r.Name)
	105	+	}
	106	+	}
	107	+	}
	108	+
	109	+	// OnOpened implements ports.MergeabilityTrigger.
	110	+	func (s *Mergeability) OnOpened(_ context.Context, mr domain.MergeRequest) {
	111	+	s.q.add(mergeKey{mr.Repo, mr.Number})
	112	+	}
	113	+
	114	+	// OnClosed deletes the record of a merge request that is no longer open. The
	115	+	// store already drops it when the request leaves the open state; this makes it
	116	+	// explicit and is idempotent.
	117	+	func (s *Mergeability) OnClosed(ctx context.Context, mr domain.MergeRequest) {
	118	+	if err := s.cfg.Store.DeleteMergeability(context.WithoutCancel(ctx), mr.Repo, mr.Number); err != nil {
	119	+	ctxLog(ctx).Error().Err(err).Str("repo", mr.Repo.String()).Int64("number", mr.Number).Msg("mergeability: delete the record of a closed merge request")
	120	+	}
	121	+	}
	122	+
	123	+	// OnBranchesMoved implements ports.MergeabilityTrigger: the open merge
	124	+	// requests whose source or target is one of the branches are queued. Past the
	125	+	// bound, or when the lookup fails, the repository is rescanned.
	126	+	func (s *Mergeability) OnBranchesMoved(ctx context.Context, repo domain.RepoName, updates []domain.RefUpdate) {
	127	+	var branches []string
	128	+	for _, u := range updates {
	129	+	if b, ok := strings.CutPrefix(u.Ref, "refs/heads/"); ok && !slices.Contains(branches, b) {
	130	+	branches = append(branches, b)
	131	+	}
	132	+	}
	133	+	if len(branches) == 0 {
	134	+	return
	135	+	}
	136	+	lookups := []func(context.Context, domain.RepoName, []string, int) ([]int64, bool, error){
	137	+	s.cfg.Store.OpenBySource, s.cfg.Store.OpenByTarget,
	138	+	}
	139	+	for _, lookup := range lookups {
	140	+	numbers, truncated, err := lookup(ctx, repo, branches, s.cfg.Queue)
	141	+	if err != nil \|\| truncated {
	142	+	if err != nil {
	143	+	ctxLog(ctx).Error().Err(err).Str("repo", repo.String()).Msg("mergeability: look up the merge requests of moved branches: the repository is rescanned")
	144	+	}
	145	+	s.q.markDirty(repo)
	146	+	}
	147	+	for _, n := range numbers {
	148	+	s.q.add(mergeKey{repo, n})
	149	+	}
	150	+	}
	151	+	}
	152	+
	153	+	// Read implements ports.MergeabilityReader.
	154	+	func (s *Mergeability) Read(ctx context.Context, mr domain.MergeRequest, targetTip domain.ObjectID) domain.Mergeability {
	155	+	checking := domain.Mergeability{Repo: mr.Repo, Number: mr.Number, State: domain.MergeabilityChecking}
	156	+	if mr.HeadID.IsZero() \|\| targetTip.IsZero() {
	157	+	return checking // a branch is gone: there is nothing to check
	158	+	}
	159	+	rec, err := s.cfg.Store.GetMergeability(ctx, mr.Repo, mr.Number)
	160	+	if err == nil && rec.IsFor(targetTip, mr.HeadID) {
	161	+	return rec
	162	+	}
	163	+	k := mergeKey{mr.Repo, mr.Number}
	164	+	s.q.markSeen(k)
	165	+	s.q.add(k)
	166	+	return checking
	167	+	}
	168	+
	169	+	func (s *Mergeability) loop() {
	170	+	defer s.wg.Done()
	171	+	for {
	172	+	key, dirty := s.q.next()
	173	+	switch {
	174	+	case key != nil:
	175	+	s.runKey(*key)
	176	+	case dirty != nil:
	177	+	s.rescan(*dirty)
	178	+	default:
	179	+	select {
	180	+	case <-s.ctx.Done():
	181	+	return
	182	+	case <-s.q.wake:
	183	+	}
	184	+	}
	185	+	if s.ctx.Err() != nil {
	186	+	return
	187	+	}
	188	+	}
	189	+	}
	190	+
	191	+	// runKey checks k, which is marked running, as often as it was queued again
	192	+	// meanwhile.
	193	+	func (s *Mergeability) runKey(k mergeKey) {
	194	+	for {
	195	+	s.check(k)
	196	+	if !s.q.finish(k) \|\| s.ctx.Err() != nil {
	197	+	return
	198	+	}
	199	+	}
	200	+	}
	201	+
	202	+	// rescan checks every open merge request of a repository, here and not through
	203	+	// the bounded queue, so that a rescan never marks its repository dirty again.
	204	+	func (s *Mergeability) rescan(repo domain.RepoName) {
	205	+	mrs, err := s.cfg.Store.ListOpen(s.ctx, repo)
	206	+	if err != nil {
	207	+	ctxLog(s.ctx).Error().Err(err).Str("repo", repo.String()).Msg("mergeability: list the open merge requests to rescan")
	208	+	return
	209	+	}
	210	+	for _, mr := range mrs {
	211	+	if s.ctx.Err() != nil {
	212	+	return
	213	+	}
	214	+	k := mergeKey{repo, mr.Number}
	215	+	if s.q.tryRun(k) {
	216	+	s.runKey(k)
	217	+	}
	218	+	}
	219	+	}
	220	+
	221	+	// Check computes the mergeability of a merge request now, on the calling
	222	+	// goroutine, as a worker does. It is for tests; the server never calls it.
	223	+	func (s *Mergeability) Check(repo domain.RepoName, number int64) {
	224	+	k := mergeKey{repo, number}
	225	+	if s.q.tryRun(k) {
	226	+	s.runKey(k)
	227	+	}
	228	+	}
	229	+
	230	+	// check brings the record of k up to date.
	231	+	func (s *Mergeability) check(k mergeKey) {
	232	+	ctx := s.ctx
	233	+	log := ctxLog(ctx).With().Str("repo", k.repo.String()).Int64("number", k.number).Logger()
	234	+	for range mergeabilityAttempts {
	235	+	mr, err := s.cfg.Store.Get(ctx, k.repo, k.number)
	236	+	if err != nil {
	237	+	if !errors.Is(err, domain.ErrNotFound) && ctx.Err() == nil {
	238	+	log.Error().Err(err).Msg("mergeability: read the merge request")
	239	+	}
	240	+	return
	241	+	}
	242	+	if !mr.IsOpen() {
	243	+	return // the store dropped the record when it left the open state
	244	+	}
	245	+	head, tip, ok := s.tips(ctx, mr)
	246	+	if !ok {
	247	+	return
	248	+	}
	249	+	prev, err := s.cfg.Store.GetMergeability(ctx, k.repo, k.number)
	250	+	if err != nil && !errors.Is(err, domain.ErrNotFound) {
	251	+	log.Error().Err(err).Msg("mergeability: read the record")
	252	+	return
	253	+	}
	254	+	if prev.IsFor(tip, head) {
	255	+	s.q.takeSeen(k)
	256	+	return
	257	+	}
	258	+	res, err := s.compute(ctx, k, tip, head)
	259	+	if err != nil {
	260	+	if ctx.Err() == nil {
	261	+	log.Error().Err(err).Msg("mergeability: compute")
	262	+	}
	263	+	return
	264	+	}
	265	+	// The branches may have moved during the computation: the result is
	266	+	// then for tips nobody asks about, and is dropped.
	267	+	if h2, t2, ok := s.tips(ctx, mr); !ok {
	268	+	return
	269	+	} else if h2 != head \|\| t2 != tip {
	270	+	continue
	271	+	}
	272	+	res.Version = prev.Version
	273	+	switch err := s.cfg.Store.PutMergeability(ctx, res); {
	274	+	case errors.Is(err, domain.ErrConflict):
	275	+	continue
	276	+	case errors.Is(err, domain.ErrNotFound):
	277	+	return // closed meanwhile
	278	+	case err != nil:
	279	+	log.Error().Err(err).Msg("mergeability: store the record")
	280	+	return
	281	+	}
	282	+	seen := s.q.takeSeen(k)
	283	+	if seen \|\| prev.State == "" \|\| changedMergeability(prev, res) {
	284	+	s.publish(ctx, res)
	285	+	}
	286	+	return
	287	+	}
	288	+	s.q.add(k) // kept losing races: try again later
	289	+	}
	290	+
	291	+	// tips reads the heads of the source and the target of mr. ok is false when
	292	+	// the check is over: a branch is gone (the record is deleted) or Git failed.
	293	+	func (s *Mergeability) tips(ctx context.Context, mr domain.MergeRequest) (head, tip domain.ObjectID, ok bool) {
	294	+	head, herr := s.cfg.Git.BranchHead(ctx, mr.Repo, mr.Source)
	295	+	tip, terr := s.cfg.Git.BranchHead(ctx, mr.Repo, mr.Target)
	296	+	switch {
	297	+	case errors.Is(herr, domain.ErrNotFound) \|\| errors.Is(terr, domain.ErrNotFound):
	298	+	if err := s.cfg.Store.DeleteMergeability(ctx, mr.Repo, mr.Number); err != nil && ctx.Err() == nil {
	299	+	ctxLog(ctx).Error().Err(err).Str("repo", mr.Repo.String()).Int64("number", mr.Number).Msg("mergeability: delete the record of a merge request without a branch")
	300	+	}
	301	+	return domain.ObjectID{}, domain.ObjectID{}, false
	302	+	case herr != nil \|\| terr != nil:
	303	+	if ctx.Err() == nil {
	304	+	ctxLog(ctx).Error().Err(errors.Join(herr, terr)).Str("repo", mr.Repo.String()).Int64("number", mr.Number).Msg("mergeability: read the branches")
	305	+	}
	306	+	return domain.ObjectID{}, domain.ObjectID{}, false
	307	+	}
	308	+	return head, tip, true
	309	+	}
	310	+
	311	+	// compute runs the Git reads that decide the state of head against tip.
	312	+	func (s *Mergeability) compute(ctx context.Context, k mergeKey, tip, head domain.ObjectID) (domain.Mergeability, error) {
	313	+	m := domain.Mergeability{Repo: k.repo, Number: k.number, TargetID: tip, HeadID: head, CheckedAt: s.cfg.Clock.Now().UTC()}
	314	+	ancestor := tip == head
	315	+	if !ancestor {
	316	+	var err error
	317	+	if ancestor, err = s.cfg.Git.IsAncestor(ctx, k.repo, tip, head); err != nil {
	318	+	return m, err
	319	+	}
	320	+	}
	321	+	if ancestor {
	322	+	m.State = domain.MergeabilityUpToDate
	323	+	return m, nil
	324	+	}
	325	+	tree, err := s.cfg.Git.MergeTree(ctx, k.repo, tip, head)
	326	+	if err != nil {
	327	+	return m, err
	328	+	}
	329	+	if !tree.Clean {
	330	+	m.State = domain.MergeabilityConflict
	331	+	m.ConflictPaths, m.ConflictPathsTruncated = tree.ConflictPaths, tree.Truncated
	332	+	return m, nil
	333	+	}
	334	+	m.State = domain.MergeabilityBehind
	335	+	if m.Behind, m.BehindCapped, err = s.cfg.Git.CountBehind(ctx, k.repo, head, tip, domain.MaxBehindCount); err != nil {
	336	+	return m, err
	337	+	}
	338	+	return m, nil
	339	+	}
	340	+
	341	+	// changedMergeability reports whether what a reader shows differs.
	342	+	func changedMergeability(a, b domain.Mergeability) bool {
	343	+	return a.State != b.State \|\| a.Behind != b.Behind \|\| a.BehindCapped != b.BehindCapped \|\|
	344	+	a.ConflictPathsTruncated != b.ConflictPathsTruncated \|\| !slices.Equal(a.ConflictPaths, b.ConflictPaths)
	345	+	}
	346	+
	347	+	func (s *Mergeability) publish(ctx context.Context, m domain.Mergeability) {
	348	+	repo, err := s.cfg.Meta.Get(ctx, m.Repo)
	349	+	if err != nil {
	350	+	if !errors.Is(err, domain.ErrNotFound) && ctx.Err() == nil {
	351	+	ctxLog(ctx).Error().Err(err).Str("repo", m.Repo.String()).Msg("mergeability: read the repository to publish a result")
	352	+	}
	353	+	return
	354	+	}
	355	+	s.cfg.Bus.Publish(ctx, domain.MergeRequestMergeability{
	356	+	Repo: m.Repo, Number: m.Number, Mergeability: m, At: m.CheckedAt,
	357	+	Visibility: repo.Visibility, CreatedAt: repo.CreatedAt,
	358	+	})
	359	+	}

A internal/usecase/mergeability_queue.go +146 -0

@@ -0,0 +1,146 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package usecase
	4	+
	5	+	import (
	6	+	"sync"
	7	+
	8	+	"bnema.dev/ermite/internal/domain"
	9	+	)
	10	+
	11	+	// mergeKey names one open merge request.
	12	+	type mergeKey struct {
	13	+	repo domain.RepoName
	14	+	number int64
	15	+	}
	16	+
	17	+	// mergeQueue is the bounded work queue of the mergeability checks: a key
	18	+	// waits at most once (pending), is checked by one worker at a time (running),
	19	+	// and a key queued while it runs is checked again afterwards (again). When
	20	+	// the queue is full the repository is marked dirty instead: a worker then
	21	+	// rescans all its open merge requests.
	22	+	type mergeQueue struct {
	23	+	mu sync.Mutex
	24	+	max int
	25	+	pending map[mergeKey]bool
	26	+	order []mergeKey
	27	+	running map[mergeKey]bool
	28	+	again map[mergeKey]bool
	29	+	dirty map[domain.RepoName]bool
	30	+	dirtyOrder []domain.RepoName
	31	+	// seen holds the keys a reader answered "checking" for: the result of
	32	+	// their next check is published whatever it is.
	33	+	seen map[mergeKey]bool
	34	+	wake chan struct{}
	35	+	}
	36	+
	37	+	func newMergeQueue(max int) *mergeQueue {
	38	+	return &mergeQueue{
	39	+	max: max, pending: map[mergeKey]bool{}, running: map[mergeKey]bool{}, again: map[mergeKey]bool{},
	40	+	dirty: map[domain.RepoName]bool{}, seen: map[mergeKey]bool{}, wake: make(chan struct{}, 1),
	41	+	}
	42	+	}
	43	+
	44	+	func (q *mergeQueue) signal() {
	45	+	select {
	46	+	case q.wake <- struct{}{}:
	47	+	default:
	48	+	}
	49	+	}
	50	+
	51	+	// add queues a check. The caller does not hold mu.
	52	+	func (q *mergeQueue) add(k mergeKey) {
	53	+	q.mu.Lock()
	54	+	switch {
	55	+	case q.running[k]:
	56	+	q.again[k] = true
	57	+	case q.pending[k]:
	58	+	case len(q.pending) < q.max:
	59	+	q.pending[k] = true
	60	+	q.order = append(q.order, k)
	61	+	default:
	62	+	q.markDirtyLocked(k.repo)
	63	+	}
	64	+	q.mu.Unlock()
	65	+	q.signal()
	66	+	}
	67	+
	68	+	func (q *mergeQueue) markDirty(repo domain.RepoName) {
	69	+	q.mu.Lock()
	70	+	q.markDirtyLocked(repo)
	71	+	q.mu.Unlock()
	72	+	q.signal()
	73	+	}
	74	+
	75	+	func (q *mergeQueue) markDirtyLocked(repo domain.RepoName) {
	76	+	if !q.dirty[repo] {
	77	+	q.dirty[repo] = true
	78	+	q.dirtyOrder = append(q.dirtyOrder, repo)
	79	+	}
	80	+	}
	81	+
	82	+	// next takes the next check, which it marks running, or the next dirty
	83	+	// repository to rescan; pending checks first.
	84	+	func (q mergeQueue) next() (key mergeKey, dirty *domain.RepoName) {
	85	+	q.mu.Lock()
	86	+	defer q.mu.Unlock()
	87	+	switch {
	88	+	case len(q.order) > 0:
	89	+	k := q.order[0]
	90	+	q.order = q.order[1:]
	91	+	delete(q.pending, k)
	92	+	q.running[k] = true
	93	+	key = &k
	94	+	case len(q.dirtyOrder) > 0:
	95	+	r := q.dirtyOrder[0]
	96	+	q.dirtyOrder = q.dirtyOrder[1:]
	97	+	delete(q.dirty, r)
	98	+	dirty = &r
	99	+	}
	100	+	if len(q.order) > 0 \|\| len(q.dirtyOrder) > 0 {
	101	+	q.signal() // let another worker take the rest
	102	+	}
	103	+	return key, dirty
	104	+	}
	105	+
	106	+	// tryRun marks k running for a rescan; when it already runs it asks for
	107	+	// another pass instead and reports false.
	108	+	func (q *mergeQueue) tryRun(k mergeKey) bool {
	109	+	q.mu.Lock()
	110	+	defer q.mu.Unlock()
	111	+	if q.running[k] {
	112	+	q.again[k] = true
	113	+	return false
	114	+	}
	115	+	delete(q.pending, k) // a pending check is covered by this run
	116	+	q.running[k] = true
	117	+	return true
	118	+	}
	119	+
	120	+	// finish ends a pass of a running key; it reports whether another is due.
	121	+	func (q *mergeQueue) finish(k mergeKey) (again bool) {
	122	+	q.mu.Lock()
	123	+	defer q.mu.Unlock()
	124	+	if q.again[k] {
	125	+	delete(q.again, k)
	126	+	return true
	127	+	}
	128	+	delete(q.running, k)
	129	+	return false
	130	+	}
	131	+
	132	+	// markSeen records that a reader was told "checking" for k.
	133	+	func (q *mergeQueue) markSeen(k mergeKey) {
	134	+	q.mu.Lock()
	135	+	q.seen[k] = true
	136	+	q.mu.Unlock()
	137	+	}
	138	+
	139	+	// takeSeen reports and clears the seen flag.
	140	+	func (q *mergeQueue) takeSeen(k mergeKey) bool {
	141	+	q.mu.Lock()
	142	+	defer q.mu.Unlock()
	143	+	s := q.seen[k]
	144	+	delete(q.seen, k)
	145	+	return s
	146	+	}

@@ -0,0 +1,146 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package usecase
	4	+
	5	+	import (
	6	+	"sync"
	7	+
	8	+	"bnema.dev/ermite/internal/domain"
	9	+	)
	10	+
	11	+	// mergeKey names one open merge request.
	12	+	type mergeKey struct {
	13	+	repo domain.RepoName
	14	+	number int64
	15	+	}
	16	+
	17	+	// mergeQueue is the bounded work queue of the mergeability checks: a key
	18	+	// waits at most once (pending), is checked by one worker at a time (running),
	19	+	// and a key queued while it runs is checked again afterwards (again). When
	20	+	// the queue is full the repository is marked dirty instead: a worker then
	21	+	// rescans all its open merge requests.
	22	+	type mergeQueue struct {
	23	+	mu sync.Mutex
	24	+	max int
	25	+	pending map[mergeKey]bool
	26	+	order []mergeKey
	27	+	running map[mergeKey]bool
	28	+	again map[mergeKey]bool
	29	+	dirty map[domain.RepoName]bool
	30	+	dirtyOrder []domain.RepoName
	31	+	// seen holds the keys a reader answered "checking" for: the result of
	32	+	// their next check is published whatever it is.
	33	+	seen map[mergeKey]bool
	34	+	wake chan struct{}
	35	+	}
	36	+
	37	+	func newMergeQueue(max int) *mergeQueue {
	38	+	return &mergeQueue{
	39	+	max: max, pending: map[mergeKey]bool{}, running: map[mergeKey]bool{}, again: map[mergeKey]bool{},
	40	+	dirty: map[domain.RepoName]bool{}, seen: map[mergeKey]bool{}, wake: make(chan struct{}, 1),
	41	+	}
	42	+	}
	43	+
	44	+	func (q *mergeQueue) signal() {
	45	+	select {
	46	+	case q.wake <- struct{}{}:
	47	+	default:
	48	+	}
	49	+	}
	50	+
	51	+	// add queues a check. The caller does not hold mu.
	52	+	func (q *mergeQueue) add(k mergeKey) {
	53	+	q.mu.Lock()
	54	+	switch {
	55	+	case q.running[k]:
	56	+	q.again[k] = true
	57	+	case q.pending[k]:
	58	+	case len(q.pending) < q.max:
	59	+	q.pending[k] = true
	60	+	q.order = append(q.order, k)
	61	+	default:
	62	+	q.markDirtyLocked(k.repo)
	63	+	}
	64	+	q.mu.Unlock()
	65	+	q.signal()
	66	+	}
	67	+
	68	+	func (q *mergeQueue) markDirty(repo domain.RepoName) {
	69	+	q.mu.Lock()
	70	+	q.markDirtyLocked(repo)
	71	+	q.mu.Unlock()
	72	+	q.signal()
	73	+	}
	74	+
	75	+	func (q *mergeQueue) markDirtyLocked(repo domain.RepoName) {
	76	+	if !q.dirty[repo] {
	77	+	q.dirty[repo] = true
	78	+	q.dirtyOrder = append(q.dirtyOrder, repo)
	79	+	}
	80	+	}
	81	+
	82	+	// next takes the next check, which it marks running, or the next dirty
	83	+	// repository to rescan; pending checks first.
	84	+	func (q mergeQueue) next() (key mergeKey, dirty *domain.RepoName) {
	85	+	q.mu.Lock()
	86	+	defer q.mu.Unlock()
	87	+	switch {
	88	+	case len(q.order) > 0:
	89	+	k := q.order[0]
	90	+	q.order = q.order[1:]
	91	+	delete(q.pending, k)
	92	+	q.running[k] = true
	93	+	key = &k
	94	+	case len(q.dirtyOrder) > 0:
	95	+	r := q.dirtyOrder[0]
	96	+	q.dirtyOrder = q.dirtyOrder[1:]
	97	+	delete(q.dirty, r)
	98	+	dirty = &r
	99	+	}
	100	+	if len(q.order) > 0 \|\| len(q.dirtyOrder) > 0 {
	101	+	q.signal() // let another worker take the rest
	102	+	}
	103	+	return key, dirty
	104	+	}
	105	+
	106	+	// tryRun marks k running for a rescan; when it already runs it asks for
	107	+	// another pass instead and reports false.
	108	+	func (q *mergeQueue) tryRun(k mergeKey) bool {
	109	+	q.mu.Lock()
	110	+	defer q.mu.Unlock()
	111	+	if q.running[k] {
	112	+	q.again[k] = true
	113	+	return false
	114	+	}
	115	+	delete(q.pending, k) // a pending check is covered by this run
	116	+	q.running[k] = true
	117	+	return true
	118	+	}
	119	+
	120	+	// finish ends a pass of a running key; it reports whether another is due.
	121	+	func (q *mergeQueue) finish(k mergeKey) (again bool) {
	122	+	q.mu.Lock()
	123	+	defer q.mu.Unlock()
	124	+	if q.again[k] {
	125	+	delete(q.again, k)
	126	+	return true
	127	+	}
	128	+	delete(q.running, k)
	129	+	return false
	130	+	}
	131	+
	132	+	// markSeen records that a reader was told "checking" for k.
	133	+	func (q *mergeQueue) markSeen(k mergeKey) {
	134	+	q.mu.Lock()
	135	+	q.seen[k] = true
	136	+	q.mu.Unlock()
	137	+	}
	138	+
	139	+	// takeSeen reports and clears the seen flag.
	140	+	func (q *mergeQueue) takeSeen(k mergeKey) bool {
	141	+	q.mu.Lock()
	142	+	defer q.mu.Unlock()
	143	+	s := q.seen[k]
	144	+	delete(q.seen, k)
	145	+	return s
	146	+	}

A internal/usecase/mergeability_test.go +347 -0

@@ -0,0 +1,347 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package usecase_test
	4	+
	5	+	import (
	6	+	"context"
	7	+	"strings"
	8	+	"sync/atomic"
	9	+	"testing"
	10	+	"time"
	11	+
	12	+	"github.com/stretchr/testify/assert"
	13	+	"github.com/stretchr/testify/mock"
	14	+	"github.com/stretchr/testify/require"
	15	+
	16	+	"bnema.dev/ermite/internal/domain"
	17	+	"bnema.dev/ermite/internal/ports"
	18	+	"bnema.dev/ermite/internal/ports/mocks"
	19	+	"bnema.dev/ermite/internal/usecase"
	20	+	)
	21	+
	22	+	func fakeOID(t *testing.T, c string) domain.ObjectID {
	23	+	t.Helper()
	24	+	return oidFrom(t, strings.Repeat(c, 40))
	25	+	}
	26	+
	27	+	// mergeRig runs the mergeability service over the real adapters of mrRig.
	28	+	type mergeRig struct {
	29	+	*mrRig
	30	+	svc *usecase.Mergeability
	31	+	mev <-chan domain.Event
	32	+	}
	33	+
	34	+	func newMergeRig(t testing.T) mergeRig {
	35	+	t.Helper()
	36	+	r := newMRRig(t)
	37	+	svc := usecase.NewMergeability(r.ctx, usecase.MergeabilityConfig{
	38	+	Store: r.store.MergeRequests(), Meta: r.store, Git: r.git, Bus: r.bus, Clock: fixedClock{}, Workers: 2,
	39	+	})
	40	+	r.mrs = usecase.NewMergeRequests(usecase.MergeRequestsConfig{
	41	+	Meta: r.store, Store: r.store.MergeRequests(), Repos: r.git, Git: r.git, Signer: nil,
	42	+	Committer: domain.Identity{Name: "ermite", Email: "ermite@git.example.org"}, Bus: r.bus, Clock: fixedClock{}, Maint: r.sched,
	43	+	Mergeability: svc,
	44	+	})
	45	+	t.Cleanup(func() { svc.Close(context.Background()) })
	46	+	return &mergeRig{mrRig: r, svc: svc, mev: r.bus.Subscribe(r.ctx)}
	47	+	}
	48	+
	49	+	func (r *mergeRig) record(n int64) (domain.Mergeability, error) {
	50	+	return r.store.MergeRequests().GetMergeability(r.ctx, r.repo, n)
	51	+	}
	52	+
	53	+	func (r *mergeRig) waitState(n int64, want domain.MergeabilityState, target, head domain.ObjectID) domain.Mergeability {
	54	+	r.t.Helper()
	55	+	var m domain.Mergeability
	56	+	require.Eventually(r.t, func() bool {
	57	+	var err error
	58	+	m, err = r.record(n)
	59	+	return err == nil && m.State == want && m.IsFor(target, head)
	60	+	}, 10time.Second, 10time.Millisecond, "mergeability of !%d did not become %s", n, want)
	61	+	return m
	62	+	}
	63	+
	64	+	// events returns the mergeability events published so far.
	65	+	func (r *mergeRig) events() []domain.MergeRequestMergeability {
	66	+	var out []domain.MergeRequestMergeability
	67	+	for {
	68	+	select {
	69	+	case ev := <-r.mev:
	70	+	if m, ok := ev.(domain.MergeRequestMergeability); ok {
	71	+	out = append(out, m)
	72	+	}
	73	+	default:
	74	+	return out
	75	+	}
	76	+	}
	77	+	}
	78	+
	79	+	func (r *mergeRig) push2(branch, path, content string) domain.ObjectID {
	80	+	r.t.Helper()
	81	+	r.sh("checkout", "-q", branch)
	82	+	id := r.commit(author, path, content, "edit "+path)
	83	+	r.push(branch)
	84	+	return id
	85	+	}
	86	+
	87	+	func TestMergeabilityStatesEndToEnd(t *testing.T) {
	88	+	r := newMergeRig(t)
	89	+	r.svc.Start()
	90	+	main := r.seed()
	91	+	head := r.feature("feat")
	92	+	mr := r.open("feat")
	93	+
	94	+	// up to date: the target is an ancestor of the head.
	95	+	m := r.waitState(mr.Number, domain.MergeabilityUpToDate, main, head)
	96	+	assert.False(t, m.CheckedAt.IsZero())
	97	+
	98	+	// behind: the target moved with a change that merges cleanly.
	99	+	r.sh("checkout", "-q", "main")
	100	+	main2 := r.commit(author, "d.txt", "dee\n", "add d")
	101	+	r.push("main")
	102	+	r.svc.OnBranchesMoved(r.ctx, r.repo, []domain.RefUpdate{{Old: main.String(), New: main2.String(), Ref: "refs/heads/main"}})
	103	+	m = r.waitState(mr.Number, domain.MergeabilityBehind, main2, head)
	104	+	assert.Equal(t, 1, m.Behind)
	105	+	assert.False(t, m.BehindCapped)
	106	+
	107	+	// conflict: the target changes a line the head changed.
	108	+	main3 := r.commit(author, "a.txt", "one\nTWO\nthree\n", "conflict a")
	109	+	r.push("main")
	110	+	r.svc.OnBranchesMoved(r.ctx, r.repo, []domain.RefUpdate{{Old: main2.String(), New: main3.String(), Ref: "refs/heads/main"}})
	111	+	m = r.waitState(mr.Number, domain.MergeabilityConflict, main3, head)
	112	+	assert.Equal(t, []string{"a.txt"}, m.ConflictPaths)
	113	+
	114	+	got := r.events()
	115	+	require.Len(t, got, 3)
	116	+	for i, s := range []domain.MergeabilityState{domain.MergeabilityUpToDate, domain.MergeabilityBehind, domain.MergeabilityConflict} {
	117	+	assert.Equal(t, s, got[i].Mergeability.State)
	118	+	assert.Equal(t, mr.Number, got[i].Number)
	119	+	}
	120	+	}
	121	+
	122	+	func TestMergeabilityReaderIsLazy(t *testing.T) {
	123	+	r := newMergeRig(t)
	124	+	main := r.seed()
	125	+	head := r.feature("feat")
	126	+	mr := r.open("feat") // queued, but the workers are not started
	127	+
	128	+	d, err := r.mrs.Get(r.ctx, root, r.repo, mr.Number, ports.GetMergeRequestOptions{})
	129	+	require.NoError(t, err)
	130	+	assert.Equal(t, domain.MergeabilityChecking, d.Mergeability.State)
	131	+
	132	+	r.svc.Start()
	133	+	r.waitState(mr.Number, domain.MergeabilityUpToDate, main, head)
	134	+	d, err = r.mrs.Get(r.ctx, root, r.repo, mr.Number, ports.GetMergeRequestOptions{})
	135	+	require.NoError(t, err)
	136	+	assert.Equal(t, domain.MergeabilityUpToDate, d.Mergeability.State)
	137	+
	138	+	page, err := r.mrs.List(r.ctx, root, r.repo, ports.ListMergeRequestsParams{})
	139	+	require.NoError(t, err)
	140	+	assert.Equal(t, domain.MergeabilityUpToDate, page.Mergeability[mr.Number].State)
	141	+	}
	142	+
	143	+	func TestMergeabilityTransitionOnlyPublication(t *testing.T) {
	144	+	r := newMergeRig(t)
	145	+	r.seed()
	146	+	r.feature("feat")
	147	+	mr := r.open("feat")
	148	+	r.svc.Check(r.repo, mr.Number)
	149	+	require.Len(t, r.events(), 1)
	150	+
	151	+	// Same tips: nothing is computed, nothing is published.
	152	+	r.svc.Check(r.repo, mr.Number)
	153	+	assert.Empty(t, r.events())
	154	+
	155	+	// The target moves without changing the state: a new record but, being the
	156	+	// same state and counts, no event.
	157	+	r.sh("checkout", "-q", "feat")
	158	+	r.commit(author, "e.txt", "e\n", "more")
	159	+	r.push("feat")
	160	+	r.svc.Check(r.repo, mr.Number)
	161	+	assert.Empty(t, r.events())
	162	+	m, err := r.record(mr.Number)
	163	+	require.NoError(t, err)
	164	+	assert.Equal(t, domain.MergeabilityUpToDate, m.State)
	165	+	}
	166	+
	167	+	func TestMergeabilityPublishesWhenReadersSawChecking(t *testing.T) {
	168	+	r := newMergeRig(t)
	169	+	r.seed()
	170	+	r.feature("feat")
	171	+	mr := r.open("feat")
	172	+	r.svc.Check(r.repo, mr.Number)
	173	+	require.Len(t, r.events(), 1)
	174	+
	175	+	r.sh("checkout", "-q", "feat")
	176	+	r.commit(author, "e.txt", "e\n", "more")
	177	+	r.push("feat")
	178	+	_, err := r.mrs.Get(r.ctx, root, r.repo, mr.Number, ports.GetMergeRequestOptions{}) // sees "checking"
	179	+	require.NoError(t, err)
	180	+	r.svc.Check(r.repo, mr.Number)
	181	+	got := r.events()
	182	+	require.Len(t, got, 1)
	183	+	assert.Equal(t, domain.MergeabilityUpToDate, got[0].Mergeability.State)
	184	+	}
	185	+
	186	+	func TestMergeabilityMergeReevaluatesOtherRequests(t *testing.T) {
	187	+	r := newMergeRig(t)
	188	+	r.svc.Start()
	189	+	main := r.seed()
	190	+	r.sh("checkout", "-q", "-B", "one", "main")
	191	+	oneHead := r.commit(author, "one.txt", "1\n", "one")
	192	+	r.push("one")
	193	+	r.sh("checkout", "-q", "-B", "two", "main")
	194	+	twoHead := r.commit(author, "two.txt", "2\n", "two")
	195	+	r.push("two")
	196	+	one, two := r.open("one"), r.open("two")
	197	+	r.waitState(one.Number, domain.MergeabilityUpToDate, main, oneHead)
	198	+	r.waitState(two.Number, domain.MergeabilityUpToDate, main, twoHead)
	199	+
	200	+	_, err := r.mrs.Merge(r.ctx, mrM, r.repo, one.Number, ports.MergeParams{HeadID: oneHead})
	201	+	require.NoError(t, err)
	202	+	m := r.waitState(two.Number, domain.MergeabilityBehind, oneHead, twoHead)
	203	+	assert.Equal(t, 1, m.Behind)
	204	+	_, err = r.record(one.Number)
	205	+	assert.ErrorIs(t, err, domain.ErrNotFound, "the merged request keeps no record")
	206	+	}
	207	+
	208	+	func TestMergeabilityClosedRequestRecordDeleted(t *testing.T) {
	209	+	r := newMergeRig(t)
	210	+	r.seed()
	211	+	r.feature("feat")
	212	+	mr := r.open("feat")
	213	+	r.svc.Check(r.repo, mr.Number)
	214	+	_, err := r.record(mr.Number)
	215	+	require.NoError(t, err)
	216	+
	217	+	_, err = r.mrs.Close(r.ctx, mrW, r.repo, mr.Number)
	218	+	require.NoError(t, err)
	219	+	_, err = r.record(mr.Number)
	220	+	assert.ErrorIs(t, err, domain.ErrNotFound)
	221	+	r.svc.Check(r.repo, mr.Number) // a late check writes nothing
	222	+	_, err = r.record(mr.Number)
	223	+	assert.ErrorIs(t, err, domain.ErrNotFound)
	224	+	}
	225	+
	226	+	func TestMergeabilityStartupRescanComputesMissingRecords(t *testing.T) {
	227	+	r := newMergeRig(t)
	228	+	main := r.seed()
	229	+	head := r.feature("feat")
	230	+	mr := r.open("feat")
	231	+	_, err := r.record(mr.Number)
	232	+	require.ErrorIs(t, err, domain.ErrNotFound)
	233	+
	234	+	r.svc.Start()
	235	+	r.waitState(mr.Number, domain.MergeabilityUpToDate, main, head)
	236	+	}
	237	+
	238	+	func TestMergeabilityNeedsNoCI(t *testing.T) {
	239	+	// The service is built without any CI: a target move still updates the
	240	+	// record, and the repository gets no pull ref (the service has no CI to create a run with).
	241	+	r := newMergeRig(t)
	242	+	r.svc.Start()
	243	+	main := r.seed()
	244	+	head := r.feature("feat")
	245	+	mr := r.open("feat")
	246	+	r.waitState(mr.Number, domain.MergeabilityUpToDate, main, head)
	247	+	r.sh("checkout", "-q", "main")
	248	+	main2 := r.commit(author, "d.txt", "dee\n", "add d")
	249	+	r.push("main")
	250	+	r.svc.OnBranchesMoved(r.ctx, r.repo, []domain.RefUpdate{{Old: main.String(), New: main2.String(), Ref: "refs/heads/main"}})
	251	+	r.waitState(mr.Number, domain.MergeabilityBehind, main2, head)
	252	+
	253	+	assert.Empty(t, r.run(r.bare(), r.env(author), "for-each-ref", "refs/pull"))
	254	+	}
	255	+
	256	+	// --- unit tests over mocks ---
	257	+
	258	+	type mockRig struct {
	259	+	store *mocks.MergeRequestStore
	260	+	meta *mocks.RepoMetaStore
	261	+	git *mocks.MergeGit
	262	+	bus *mocks.EventBus
	263	+	svc *usecase.Mergeability
	264	+	mr domain.MergeRequest
	265	+	}
	266	+
	267	+	func newMockRig(t testing.T) mockRig {
	268	+	t.Helper()
	269	+	repo := name(t, "proj")
	270	+	r := &mockRig{
	271	+	store: mocks.NewMergeRequestStore(t), meta: mocks.NewRepoMetaStore(t), git: mocks.NewMergeGit(t), bus: mocks.NewEventBus(t),
	272	+	mr: domain.MergeRequest{Repo: repo, Number: 7, Source: "feat", Target: "main", State: domain.MRStateOpen},
	273	+	}
	274	+	r.store.EXPECT().Get(mock.Anything, repo, int64(7)).Return(r.mr, nil).Maybe()
	275	+	r.store.EXPECT().GetMergeability(mock.Anything, repo, int64(7)).Return(domain.Mergeability{}, domain.ErrNotFound).Maybe()
	276	+	r.meta.EXPECT().List(mock.Anything).Return(nil, nil).Maybe()
	277	+	r.meta.EXPECT().Get(mock.Anything, repo).Return(domain.Repo{Name: repo, CreatedAt: now}, nil).Maybe()
	278	+	r.svc = usecase.NewMergeability(context.Background(), usecase.MergeabilityConfig{
	279	+	Store: r.store, Meta: r.meta, Git: r.git, Bus: r.bus, Clock: fixedClock{},
	280	+	})
	281	+	t.Cleanup(func() { r.svc.Close(context.Background()) })
	282	+	return r
	283	+	}
	284	+
	285	+	func TestMergeabilityStaleResultIsDiscarded(t *testing.T) {
	286	+	r := newMockRig(t)
	287	+	repo := r.mr.Repo
	288	+	head, t1, t2 := fakeOID(t, "a"), fakeOID(t, "b"), fakeOID(t, "c")
	289	+	var moved atomic.Bool
	290	+	r.git.EXPECT().BranchHead(mock.Anything, repo, "feat").Return(head, nil)
	291	+	r.git.EXPECT().BranchHead(mock.Anything, repo, "main").RunAndReturn(
	292	+	func(context.Context, domain.RepoName, string) (domain.ObjectID, error) {
	293	+	if moved.Load() {
	294	+	return t2, nil
	295	+	}
	296	+	return t1, nil
	297	+	})
	298	+	// The target moves while the first computation runs.
	299	+	r.git.EXPECT().IsAncestor(mock.Anything, repo, mock.Anything, head).RunAndReturn(
	300	+	func(context.Context, domain.RepoName, domain.ObjectID, domain.ObjectID) (bool, error) {
	301	+	moved.Store(true)
	302	+	return true, nil
	303	+	})
	304	+	var puts []domain.Mergeability
	305	+	r.store.EXPECT().PutMergeability(mock.Anything, mock.Anything).RunAndReturn(
	306	+	func(_ context.Context, m domain.Mergeability) error { puts = append(puts, m); return nil })
	307	+	r.bus.EXPECT().Publish(mock.Anything, mock.Anything).Return()
	308	+
	309	+	r.svc.Check(repo, 7)
	310	+	require.Len(t, puts, 1, "the result for the old tip is dropped, the one for the new tip is stored")
	311	+	assert.Equal(t, t2, puts[0].TargetID)
	312	+	}
	313	+
	314	+	func TestMergeabilityDedupsTriggers(t *testing.T) {
	315	+	r := newMockRig(t)
	316	+	repo := r.mr.Repo
	317	+	head, tip := fakeOID(t, "a"), fakeOID(t, "b")
	318	+	r.git.EXPECT().BranchHead(mock.Anything, repo, "feat").Return(head, nil)
	319	+	r.git.EXPECT().BranchHead(mock.Anything, repo, "main").Return(tip, nil)
	320	+	var computes atomic.Int32
	321	+	r.git.EXPECT().IsAncestor(mock.Anything, repo, tip, head).RunAndReturn(
	322	+	func(context.Context, domain.RepoName, domain.ObjectID, domain.ObjectID) (bool, error) {
	323	+	computes.Add(1)
	324	+	return true, nil
	325	+	})
	326	+	r.store.EXPECT().PutMergeability(mock.Anything, mock.Anything).Return(nil)
	327	+	r.bus.EXPECT().Publish(mock.Anything, mock.Anything).Return()
	328	+
	329	+	for range 50 {
	330	+	r.svc.OnOpened(context.Background(), r.mr) // queued before any worker runs
	331	+	}
	332	+	r.svc.Start()
	333	+	require.Eventually(t, func() bool { return computes.Load() >= 1 }, 5time.Second, 5time.Millisecond)
	334	+	time.Sleep(50 * time.Millisecond)
	335	+	assert.EqualValues(t, 1, computes.Load())
	336	+	}
	337	+
	338	+	func TestMergeabilityNoWriteForClosedRequest(t *testing.T) {
	339	+	r := newMockRig(t)
	340	+	repo := r.mr.Repo
	341	+	head, tip := fakeOID(t, "a"), fakeOID(t, "b")
	342	+	r.git.EXPECT().BranchHead(mock.Anything, repo, "feat").Return(head, nil)
	343	+	r.git.EXPECT().BranchHead(mock.Anything, repo, "main").Return(tip, nil)
	344	+	r.git.EXPECT().IsAncestor(mock.Anything, repo, tip, head).Return(true, nil)
	345	+	r.store.EXPECT().PutMergeability(mock.Anything, mock.Anything).Return(domain.ErrNotFound).Once() // closed meanwhile
	346	+	r.svc.Check(repo, 7) // no event, no further write
	347	+	}

@@ -0,0 +1,347 @@
	1	+	// SPDX-License-Identifier: AGPL-3.0-or-later
	2	+
	3	+	package usecase_test
	4	+
	5	+	import (
	6	+	"context"
	7	+	"strings"
	8	+	"sync/atomic"
	9	+	"testing"
	10	+	"time"
	11	+
	12	+	"github.com/stretchr/testify/assert"
	13	+	"github.com/stretchr/testify/mock"
	14	+	"github.com/stretchr/testify/require"
	15	+
	16	+	"bnema.dev/ermite/internal/domain"
	17	+	"bnema.dev/ermite/internal/ports"
	18	+	"bnema.dev/ermite/internal/ports/mocks"
	19	+	"bnema.dev/ermite/internal/usecase"
	20	+	)
	21	+
	22	+	func fakeOID(t *testing.T, c string) domain.ObjectID {
	23	+	t.Helper()
	24	+	return oidFrom(t, strings.Repeat(c, 40))
	25	+	}
	26	+
	27	+	// mergeRig runs the mergeability service over the real adapters of mrRig.
	28	+	type mergeRig struct {
	29	+	*mrRig
	30	+	svc *usecase.Mergeability
	31	+	mev <-chan domain.Event
	32	+	}
	33	+
	34	+	func newMergeRig(t testing.T) mergeRig {
	35	+	t.Helper()
	36	+	r := newMRRig(t)
	37	+	svc := usecase.NewMergeability(r.ctx, usecase.MergeabilityConfig{
	38	+	Store: r.store.MergeRequests(), Meta: r.store, Git: r.git, Bus: r.bus, Clock: fixedClock{}, Workers: 2,
	39	+	})
	40	+	r.mrs = usecase.NewMergeRequests(usecase.MergeRequestsConfig{
	41	+	Meta: r.store, Store: r.store.MergeRequests(), Repos: r.git, Git: r.git, Signer: nil,
	42	+	Committer: domain.Identity{Name: "ermite", Email: "ermite@git.example.org"}, Bus: r.bus, Clock: fixedClock{}, Maint: r.sched,
	43	+	Mergeability: svc,
	44	+	})
	45	+	t.Cleanup(func() { svc.Close(context.Background()) })
	46	+	return &mergeRig{mrRig: r, svc: svc, mev: r.bus.Subscribe(r.ctx)}
	47	+	}
	48	+
	49	+	func (r *mergeRig) record(n int64) (domain.Mergeability, error) {
	50	+	return r.store.MergeRequests().GetMergeability(r.ctx, r.repo, n)
	51	+	}
	52	+
	53	+	func (r *mergeRig) waitState(n int64, want domain.MergeabilityState, target, head domain.ObjectID) domain.Mergeability {
	54	+	r.t.Helper()
	55	+	var m domain.Mergeability
	56	+	require.Eventually(r.t, func() bool {
	57	+	var err error
	58	+	m, err = r.record(n)
	59	+	return err == nil && m.State == want && m.IsFor(target, head)
	60	+	}, 10time.Second, 10time.Millisecond, "mergeability of !%d did not become %s", n, want)
	61	+	return m
	62	+	}
	63	+
	64	+	// events returns the mergeability events published so far.
	65	+	func (r *mergeRig) events() []domain.MergeRequestMergeability {
	66	+	var out []domain.MergeRequestMergeability
	67	+	for {
	68	+	select {
	69	+	case ev := <-r.mev:
	70	+	if m, ok := ev.(domain.MergeRequestMergeability); ok {
	71	+	out = append(out, m)
	72	+	}
	73	+	default:
	74	+	return out
	75	+	}
	76	+	}
	77	+	}
	78	+
	79	+	func (r *mergeRig) push2(branch, path, content string) domain.ObjectID {
	80	+	r.t.Helper()
	81	+	r.sh("checkout", "-q", branch)
	82	+	id := r.commit(author, path, content, "edit "+path)
	83	+	r.push(branch)
	84	+	return id
	85	+	}
	86	+
	87	+	func TestMergeabilityStatesEndToEnd(t *testing.T) {
	88	+	r := newMergeRig(t)
	89	+	r.svc.Start()
	90	+	main := r.seed()
	91	+	head := r.feature("feat")
	92	+	mr := r.open("feat")
	93	+
	94	+	// up to date: the target is an ancestor of the head.
	95	+	m := r.waitState(mr.Number, domain.MergeabilityUpToDate, main, head)
	96	+	assert.False(t, m.CheckedAt.IsZero())
	97	+
	98	+	// behind: the target moved with a change that merges cleanly.
	99	+	r.sh("checkout", "-q", "main")
	100	+	main2 := r.commit(author, "d.txt", "dee\n", "add d")
	101	+	r.push("main")
	102	+	r.svc.OnBranchesMoved(r.ctx, r.repo, []domain.RefUpdate{{Old: main.String(), New: main2.String(), Ref: "refs/heads/main"}})
	103	+	m = r.waitState(mr.Number, domain.MergeabilityBehind, main2, head)
	104	+	assert.Equal(t, 1, m.Behind)
	105	+	assert.False(t, m.BehindCapped)
	106	+
	107	+	// conflict: the target changes a line the head changed.
	108	+	main3 := r.commit(author, "a.txt", "one\nTWO\nthree\n", "conflict a")
	109	+	r.push("main")
	110	+	r.svc.OnBranchesMoved(r.ctx, r.repo, []domain.RefUpdate{{Old: main2.String(), New: main3.String(), Ref: "refs/heads/main"}})
	111	+	m = r.waitState(mr.Number, domain.MergeabilityConflict, main3, head)
	112	+	assert.Equal(t, []string{"a.txt"}, m.ConflictPaths)
	113	+
	114	+	got := r.events()
	115	+	require.Len(t, got, 3)
	116	+	for i, s := range []domain.MergeabilityState{domain.MergeabilityUpToDate, domain.MergeabilityBehind, domain.MergeabilityConflict} {
	117	+	assert.Equal(t, s, got[i].Mergeability.State)
	118	+	assert.Equal(t, mr.Number, got[i].Number)
	119	+	}
	120	+	}
	121	+
	122	+	func TestMergeabilityReaderIsLazy(t *testing.T) {
	123	+	r := newMergeRig(t)
	124	+	main := r.seed()
	125	+	head := r.feature("feat")
	126	+	mr := r.open("feat") // queued, but the workers are not started
	127	+
	128	+	d, err := r.mrs.Get(r.ctx, root, r.repo, mr.Number, ports.GetMergeRequestOptions{})
	129	+	require.NoError(t, err)
	130	+	assert.Equal(t, domain.MergeabilityChecking, d.Mergeability.State)
	131	+
	132	+	r.svc.Start()
	133	+	r.waitState(mr.Number, domain.MergeabilityUpToDate, main, head)
	134	+	d, err = r.mrs.Get(r.ctx, root, r.repo, mr.Number, ports.GetMergeRequestOptions{})
	135	+	require.NoError(t, err)
	136	+	assert.Equal(t, domain.MergeabilityUpToDate, d.Mergeability.State)
	137	+
	138	+	page, err := r.mrs.List(r.ctx, root, r.repo, ports.ListMergeRequestsParams{})
	139	+	require.NoError(t, err)
	140	+	assert.Equal(t, domain.MergeabilityUpToDate, page.Mergeability[mr.Number].State)
	141	+	}
	142	+
	143	+	func TestMergeabilityTransitionOnlyPublication(t *testing.T) {
	144	+	r := newMergeRig(t)
	145	+	r.seed()
	146	+	r.feature("feat")
	147	+	mr := r.open("feat")
	148	+	r.svc.Check(r.repo, mr.Number)
	149	+	require.Len(t, r.events(), 1)
	150	+
	151	+	// Same tips: nothing is computed, nothing is published.
	152	+	r.svc.Check(r.repo, mr.Number)
	153	+	assert.Empty(t, r.events())
	154	+
	155	+	// The target moves without changing the state: a new record but, being the
	156	+	// same state and counts, no event.
	157	+	r.sh("checkout", "-q", "feat")
	158	+	r.commit(author, "e.txt", "e\n", "more")
	159	+	r.push("feat")
	160	+	r.svc.Check(r.repo, mr.Number)
	161	+	assert.Empty(t, r.events())
	162	+	m, err := r.record(mr.Number)
	163	+	require.NoError(t, err)
	164	+	assert.Equal(t, domain.MergeabilityUpToDate, m.State)
	165	+	}
	166	+
	167	+	func TestMergeabilityPublishesWhenReadersSawChecking(t *testing.T) {
	168	+	r := newMergeRig(t)
	169	+	r.seed()
	170	+	r.feature("feat")
	171	+	mr := r.open("feat")
	172	+	r.svc.Check(r.repo, mr.Number)
	173	+	require.Len(t, r.events(), 1)
	174	+
	175	+	r.sh("checkout", "-q", "feat")
	176	+	r.commit(author, "e.txt", "e\n", "more")
	177	+	r.push("feat")
	178	+	_, err := r.mrs.Get(r.ctx, root, r.repo, mr.Number, ports.GetMergeRequestOptions{}) // sees "checking"
	179	+	require.NoError(t, err)
	180	+	r.svc.Check(r.repo, mr.Number)
	181	+	got := r.events()
	182	+	require.Len(t, got, 1)
	183	+	assert.Equal(t, domain.MergeabilityUpToDate, got[0].Mergeability.State)
	184	+	}
	185	+
	186	+	func TestMergeabilityMergeReevaluatesOtherRequests(t *testing.T) {
	187	+	r := newMergeRig(t)
	188	+	r.svc.Start()
	189	+	main := r.seed()
	190	+	r.sh("checkout", "-q", "-B", "one", "main")
	191	+	oneHead := r.commit(author, "one.txt", "1\n", "one")
	192	+	r.push("one")
	193	+	r.sh("checkout", "-q", "-B", "two", "main")
	194	+	twoHead := r.commit(author, "two.txt", "2\n", "two")
	195	+	r.push("two")
	196	+	one, two := r.open("one"), r.open("two")
	197	+	r.waitState(one.Number, domain.MergeabilityUpToDate, main, oneHead)
	198	+	r.waitState(two.Number, domain.MergeabilityUpToDate, main, twoHead)
	199	+
	200	+	_, err := r.mrs.Merge(r.ctx, mrM, r.repo, one.Number, ports.MergeParams{HeadID: oneHead})
	201	+	require.NoError(t, err)
	202	+	m := r.waitState(two.Number, domain.MergeabilityBehind, oneHead, twoHead)
	203	+	assert.Equal(t, 1, m.Behind)
	204	+	_, err = r.record(one.Number)
	205	+	assert.ErrorIs(t, err, domain.ErrNotFound, "the merged request keeps no record")
	206	+	}
	207	+
	208	+	func TestMergeabilityClosedRequestRecordDeleted(t *testing.T) {
	209	+	r := newMergeRig(t)
	210	+	r.seed()
	211	+	r.feature("feat")
	212	+	mr := r.open("feat")
	213	+	r.svc.Check(r.repo, mr.Number)
	214	+	_, err := r.record(mr.Number)
	215	+	require.NoError(t, err)
	216	+
	217	+	_, err = r.mrs.Close(r.ctx, mrW, r.repo, mr.Number)
	218	+	require.NoError(t, err)
	219	+	_, err = r.record(mr.Number)
	220	+	assert.ErrorIs(t, err, domain.ErrNotFound)
	221	+	r.svc.Check(r.repo, mr.Number) // a late check writes nothing
	222	+	_, err = r.record(mr.Number)
	223	+	assert.ErrorIs(t, err, domain.ErrNotFound)
	224	+	}
	225	+
	226	+	func TestMergeabilityStartupRescanComputesMissingRecords(t *testing.T) {
	227	+	r := newMergeRig(t)
	228	+	main := r.seed()
	229	+	head := r.feature("feat")
	230	+	mr := r.open("feat")
	231	+	_, err := r.record(mr.Number)
	232	+	require.ErrorIs(t, err, domain.ErrNotFound)
	233	+
	234	+	r.svc.Start()
	235	+	r.waitState(mr.Number, domain.MergeabilityUpToDate, main, head)
	236	+	}
	237	+
	238	+	func TestMergeabilityNeedsNoCI(t *testing.T) {
	239	+	// The service is built without any CI: a target move still updates the
	240	+	// record, and the repository gets no pull ref (the service has no CI to create a run with).
	241	+	r := newMergeRig(t)
	242	+	r.svc.Start()
	243	+	main := r.seed()
	244	+	head := r.feature("feat")
	245	+	mr := r.open("feat")
	246	+	r.waitState(mr.Number, domain.MergeabilityUpToDate, main, head)
	247	+	r.sh("checkout", "-q", "main")
	248	+	main2 := r.commit(author, "d.txt", "dee\n", "add d")
	249	+	r.push("main")
	250	+	r.svc.OnBranchesMoved(r.ctx, r.repo, []domain.RefUpdate{{Old: main.String(), New: main2.String(), Ref: "refs/heads/main"}})
	251	+	r.waitState(mr.Number, domain.MergeabilityBehind, main2, head)
	252	+
	253	+	assert.Empty(t, r.run(r.bare(), r.env(author), "for-each-ref", "refs/pull"))
	254	+	}
	255	+
	256	+	// --- unit tests over mocks ---
	257	+
	258	+	type mockRig struct {
	259	+	store *mocks.MergeRequestStore
	260	+	meta *mocks.RepoMetaStore
	261	+	git *mocks.MergeGit
	262	+	bus *mocks.EventBus
	263	+	svc *usecase.Mergeability
	264	+	mr domain.MergeRequest
	265	+	}
	266	+
	267	+	func newMockRig(t testing.T) mockRig {
	268	+	t.Helper()
	269	+	repo := name(t, "proj")
	270	+	r := &mockRig{
	271	+	store: mocks.NewMergeRequestStore(t), meta: mocks.NewRepoMetaStore(t), git: mocks.NewMergeGit(t), bus: mocks.NewEventBus(t),
	272	+	mr: domain.MergeRequest{Repo: repo, Number: 7, Source: "feat", Target: "main", State: domain.MRStateOpen},
	273	+	}
	274	+	r.store.EXPECT().Get(mock.Anything, repo, int64(7)).Return(r.mr, nil).Maybe()
	275	+	r.store.EXPECT().GetMergeability(mock.Anything, repo, int64(7)).Return(domain.Mergeability{}, domain.ErrNotFound).Maybe()
	276	+	r.meta.EXPECT().List(mock.Anything).Return(nil, nil).Maybe()
	277	+	r.meta.EXPECT().Get(mock.Anything, repo).Return(domain.Repo{Name: repo, CreatedAt: now}, nil).Maybe()
	278	+	r.svc = usecase.NewMergeability(context.Background(), usecase.MergeabilityConfig{
	279	+	Store: r.store, Meta: r.meta, Git: r.git, Bus: r.bus, Clock: fixedClock{},
	280	+	})
	281	+	t.Cleanup(func() { r.svc.Close(context.Background()) })
	282	+	return r
	283	+	}
	284	+
	285	+	func TestMergeabilityStaleResultIsDiscarded(t *testing.T) {
	286	+	r := newMockRig(t)
	287	+	repo := r.mr.Repo
	288	+	head, t1, t2 := fakeOID(t, "a"), fakeOID(t, "b"), fakeOID(t, "c")
	289	+	var moved atomic.Bool
	290	+	r.git.EXPECT().BranchHead(mock.Anything, repo, "feat").Return(head, nil)
	291	+	r.git.EXPECT().BranchHead(mock.Anything, repo, "main").RunAndReturn(
	292	+	func(context.Context, domain.RepoName, string) (domain.ObjectID, error) {
	293	+	if moved.Load() {
	294	+	return t2, nil
	295	+	}
	296	+	return t1, nil
	297	+	})
	298	+	// The target moves while the first computation runs.
	299	+	r.git.EXPECT().IsAncestor(mock.Anything, repo, mock.Anything, head).RunAndReturn(
	300	+	func(context.Context, domain.RepoName, domain.ObjectID, domain.ObjectID) (bool, error) {
	301	+	moved.Store(true)
	302	+	return true, nil
	303	+	})
	304	+	var puts []domain.Mergeability
	305	+	r.store.EXPECT().PutMergeability(mock.Anything, mock.Anything).RunAndReturn(
	306	+	func(_ context.Context, m domain.Mergeability) error { puts = append(puts, m); return nil })
	307	+	r.bus.EXPECT().Publish(mock.Anything, mock.Anything).Return()
	308	+
	309	+	r.svc.Check(repo, 7)
	310	+	require.Len(t, puts, 1, "the result for the old tip is dropped, the one for the new tip is stored")
	311	+	assert.Equal(t, t2, puts[0].TargetID)
	312	+	}
	313	+
	314	+	func TestMergeabilityDedupsTriggers(t *testing.T) {
	315	+	r := newMockRig(t)
	316	+	repo := r.mr.Repo
	317	+	head, tip := fakeOID(t, "a"), fakeOID(t, "b")
	318	+	r.git.EXPECT().BranchHead(mock.Anything, repo, "feat").Return(head, nil)
	319	+	r.git.EXPECT().BranchHead(mock.Anything, repo, "main").Return(tip, nil)
	320	+	var computes atomic.Int32
	321	+	r.git.EXPECT().IsAncestor(mock.Anything, repo, tip, head).RunAndReturn(
	322	+	func(context.Context, domain.RepoName, domain.ObjectID, domain.ObjectID) (bool, error) {
	323	+	computes.Add(1)
	324	+	return true, nil
	325	+	})
	326	+	r.store.EXPECT().PutMergeability(mock.Anything, mock.Anything).Return(nil)
	327	+	r.bus.EXPECT().Publish(mock.Anything, mock.Anything).Return()
	328	+
	329	+	for range 50 {
	330	+	r.svc.OnOpened(context.Background(), r.mr) // queued before any worker runs
	331	+	}
	332	+	r.svc.Start()
	333	+	require.Eventually(t, func() bool { return computes.Load() >= 1 }, 5time.Second, 5time.Millisecond)
	334	+	time.Sleep(50 * time.Millisecond)
	335	+	assert.EqualValues(t, 1, computes.Load())
	336	+	}
	337	+
	338	+	func TestMergeabilityNoWriteForClosedRequest(t *testing.T) {
	339	+	r := newMockRig(t)
	340	+	repo := r.mr.Repo
	341	+	head, tip := fakeOID(t, "a"), fakeOID(t, "b")
	342	+	r.git.EXPECT().BranchHead(mock.Anything, repo, "feat").Return(head, nil)
	343	+	r.git.EXPECT().BranchHead(mock.Anything, repo, "main").Return(tip, nil)
	344	+	r.git.EXPECT().IsAncestor(mock.Anything, repo, tip, head).Return(true, nil)
	345	+	r.store.EXPECT().PutMergeability(mock.Anything, mock.Anything).Return(domain.ErrNotFound).Once() // closed meanwhile
	346	+	r.svc.Check(repo, 7) // no event, no further write
	347	+	}

M internal/usecase/mergerequests.go +37 -1


@@ -56,6 +56,7 @@ type MergeRequests struct {
56	56		pulls ports.PullRequestTrigger // nil: no pull_request workflows
57	57		tips ports.CIRefTipStore // nil: who merges is not recorded
58	58		verifier ports.CommitVerifier // nil: commits are not checked (ADR 0030)
	59	+	mergeab MergeabilityHook // nil: no mergeability
59	60
60	61		mu sync.Mutex
61	62		locks map[domain.RepoName]chan struct{}

@@ -95,13 +96,16 @@ type MergeRequestsConfig struct {
95	96		// deleted source branch) before the merge does, as a push does
96	97		// (domain.CIRefTip). Optional.
97	98		RefTips ports.CIRefTipStore
	99	+	// Mergeability is queued when a request is opened, closed or merged and
	100	+	// read by Get and List. Optional: without it Mergeability stays zero.
	101	+	Mergeability MergeabilityHook
98	102		}
99	103
100	104		// NewMergeRequests builds the merge request use cases.
101	105		func NewMergeRequests(cfg MergeRequestsConfig) *MergeRequests {
102	106		return &MergeRequests{
103	107		meta: cfg.Meta, store: cfg.Store, repos: cfg.Repos, git: cfg.Git, signer: cfg.Signer,
104		-	committer: cfg.Committer, bus: cfg.Bus, clock: cfg.Clock, maint: cfg.Maint, ci: cfg.CI, checks: cfg.Checks, pulls: cfg.PullRequests, tips: cfg.RefTips,
	108	+	committer: cfg.Committer, bus: cfg.Bus, clock: cfg.Clock, maint: cfg.Maint, ci: cfg.CI, checks: cfg.Checks, pulls: cfg.PullRequests, tips: cfg.RefTips, mergeab: cfg.Mergeability,
105	109		locks: make(map[domain.RepoName]chan struct{}),
106	110		}
107	111		}

@@ -256,6 +260,9 @@ func (m *MergeRequests) Create(ctx context.Context, p domain.Principal, name dom
256	260		}
257	261		recordActivity(ctx, m.meta, repo.Name, now)
258	262		m.bus.Publish(ctx, domain.MergeRequestOpened{MergeRequest: mr, At: now, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt})
	263	+	if m.mergeab != nil {
	264	+	m.mergeab.OnOpened(ctx, mr)
	265	+	}
259	266		if m.pulls != nil {
260	267		// The commits the opening is for are the ones of now.
261	268		base, _ := m.branchOrZero(ctx, name, target)

@@ -336,10 +343,25 @@ func (m *MergeRequests) List(ctx context.Context, p domain.Principal, name domai
336	343		return (params.State == "" \|\| mr.State == params.State) && (after == 0 \|\| mr.Number < after)
337	344		},
338	345		encodeMRToken)
	346	+	tips := map[string]domain.ObjectID{} // one BranchHead per distinct target
339	347		for i, mr := range page.MergeRequests {
340	348		if page.MergeRequests[i], err = m.withHead(ctx, mr); err != nil {
341	349		return ports.MergeRequestPage{}, err
342	350		}
	351	+	if m.mergeab == nil \|\| !mr.IsOpen() {
	352	+	continue
	353	+	}
	354	+	tip, ok := tips[mr.Target]
	355	+	if !ok {
	356	+	if tip, err = m.branchOrZero(ctx, name, mr.Target); err != nil {
	357	+	return ports.MergeRequestPage{}, err
	358	+	}
	359	+	tips[mr.Target] = tip
	360	+	}
	361	+	if page.Mergeability == nil {
	362	+	page.Mergeability = map[int64]domain.Mergeability{}
	363	+	}
	364	+	page.Mergeability[mr.Number] = m.mergeab.Read(ctx, page.MergeRequests[i], tip)
343	365		}
344	366		return page, nil
345	367		}

@@ -467,6 +489,13 @@ func (m *MergeRequests) Get(ctx context.Context, p domain.Principal, name domain
467	489		return nil
468	490		}
469	491		d := ports.MergeRequestDetail{MergeRequest: mr}
	492	+	if m.mergeab != nil && mr.IsOpen() {
	493	+	tip, err := m.branchOrZero(ctx, name, mr.Target)
	494	+	if err != nil {
	495	+	return ports.MergeRequestDetail{}, err
	496	+	}
	497	+	d.Mergeability = m.mergeab.Read(ctx, mr, tip)
	498	+	}
470	499		if d.Comments, err = m.store.ListComments(ctx, name, number); err != nil {
471	500		return ports.MergeRequestDetail{}, fmt.Errorf("list comments of !%d in %q: %w", number, name, err)
472	501		}

@@ -801,6 +830,9 @@ func (m *MergeRequests) Close(ctx context.Context, p domain.Principal, name doma
801	830		Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("merge request closed")
802	831		recordActivity(ctx, m.meta, repo.Name, mr.UpdatedAt)
803	832		m.bus.Publish(ctx, domain.MergeRequestClosed{MergeRequest: mr, At: mr.UpdatedAt, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt})
	833	+	if m.mergeab != nil {
	834	+	m.mergeab.OnClosed(ctx, mr)
	835	+	}
804	836		if m.pulls != nil {
805	837		m.pulls.OnPullRequest(ctx, name, ports.PullRequestEvent{Number: number, Action: domain.PullClosed, Head: mr.HeadID, Base: mr.BaseID}, p)
806	838		}

@@ -1007,6 +1039,10 @@ func (m *MergeRequests) merge(ctx context.Context, p domain.Principal, name doma
1007	1039		m.bus.Publish(ctx, domain.PushEvent{
1008	1040		Repo: name, At: at, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt, Updates: updates,
1009	1041		})
	1042	+	if m.mergeab != nil {
	1043	+	m.mergeab.OnClosed(ctx, mr)
	1044	+	m.mergeab.OnBranchesMoved(ctx, name, updates) // the other requests of the moved target
	1045	+	}
1010	1046		if m.pulls != nil {
1011	1047		m.pulls.OnPullRequest(ctx, name, ports.PullRequestEvent{Number: number, Action: domain.PullClosed, Head: mr.HeadID, Base: mr.BaseID, Merged: true}, p)
1012	1048		}

 	pulls     ports.PullRequestTrigger // nil: no pull_request workflows
 	tips      ports.CIRefTipStore      // nil: who merges is not recorded
 	verifier  ports.CommitVerifier     // nil: commits are not checked (ADR 0030)
+	mergeab   MergeabilityHook         // nil: no mergeability
 	mu    sync.Mutex
 	locks map[domain.RepoName]chan struct{}
 	// deleted source branch) before the merge does, as a push does
 	// (domain.CIRefTip). Optional.
 	RefTips ports.CIRefTipStore
+	// Mergeability is queued when a request is opened, closed or merged and
+	// read by Get and List. Optional: without it Mergeability stays zero.
+	Mergeability MergeabilityHook
 }
 // NewMergeRequests builds the merge request use cases.
 func NewMergeRequests(cfg MergeRequestsConfig) *MergeRequests {
 	return &MergeRequests{
 		meta: cfg.Meta, store: cfg.Store, repos: cfg.Repos, git: cfg.Git, signer: cfg.Signer,
-		committer: cfg.Committer, bus: cfg.Bus, clock: cfg.Clock, maint: cfg.Maint, ci: cfg.CI, checks: cfg.Checks, pulls: cfg.PullRequests, tips: cfg.RefTips,
+		committer: cfg.Committer, bus: cfg.Bus, clock: cfg.Clock, maint: cfg.Maint, ci: cfg.CI, checks: cfg.Checks, pulls: cfg.PullRequests, tips: cfg.RefTips, mergeab: cfg.Mergeability,
 		locks: make(map[domain.RepoName]chan struct{}),
 	}
 }
 	}
 	recordActivity(ctx, m.meta, repo.Name, now)
 	m.bus.Publish(ctx, domain.MergeRequestOpened{MergeRequest: mr, At: now, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt})
+	if m.mergeab != nil {
+		m.mergeab.OnOpened(ctx, mr)
+	}
 	if m.pulls != nil {
 		// The commits the opening is for are the ones of now.
 		base, _ := m.branchOrZero(ctx, name, target)
 			return (params.State == "" || mr.State == params.State) && (after == 0 || mr.Number < after)
 		},
 		encodeMRToken)
+	tips := map[string]domain.ObjectID{} // one BranchHead per distinct target
 	for i, mr := range page.MergeRequests {
 		if page.MergeRequests[i], err = m.withHead(ctx, mr); err != nil {
 			return ports.MergeRequestPage{}, err
 		}
+		if m.mergeab == nil || !mr.IsOpen() {
+			continue
+		}
+		tip, ok := tips[mr.Target]
+		if !ok {
+			if tip, err = m.branchOrZero(ctx, name, mr.Target); err != nil {
+				return ports.MergeRequestPage{}, err
+			}
+			tips[mr.Target] = tip
+		}
+		if page.Mergeability == nil {
+			page.Mergeability = map[int64]domain.Mergeability{}
+		}
+		page.Mergeability[mr.Number] = m.mergeab.Read(ctx, page.MergeRequests[i], tip)
 	}
 	return page, nil
 }
 		return nil
 	}
 	d := ports.MergeRequestDetail{MergeRequest: mr}
+	if m.mergeab != nil && mr.IsOpen() {
+		tip, err := m.branchOrZero(ctx, name, mr.Target)
+		if err != nil {
+			return ports.MergeRequestDetail{}, err
+		}
+		d.Mergeability = m.mergeab.Read(ctx, mr, tip)
+	}
 	if d.Comments, err = m.store.ListComments(ctx, name, number); err != nil {
 		return ports.MergeRequestDetail{}, fmt.Errorf("list comments of !%d in %q: %w", number, name, err)
 	}
 			Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("merge request closed")
 		recordActivity(ctx, m.meta, repo.Name, mr.UpdatedAt)
 		m.bus.Publish(ctx, domain.MergeRequestClosed{MergeRequest: mr, At: mr.UpdatedAt, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt})
+		if m.mergeab != nil {
+			m.mergeab.OnClosed(ctx, mr)
+		}
 		if m.pulls != nil {
 			m.pulls.OnPullRequest(ctx, name, ports.PullRequestEvent{Number: number, Action: domain.PullClosed, Head: mr.HeadID, Base: mr.BaseID}, p)
 		}
 	m.bus.Publish(ctx, domain.PushEvent{
 		Repo: name, At: at, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt, Updates: updates,
 	})
+	if m.mergeab != nil {
+		m.mergeab.OnClosed(ctx, mr)
+		m.mergeab.OnBranchesMoved(ctx, name, updates) // the other requests of the moved target
+	}
 	if m.pulls != nil {
 		m.pulls.OnPullRequest(ctx, name, ports.PullRequestEvent{Number: number, Action: domain.PullClosed, Head: mr.HeadID, Base: mr.BaseID, Merged: true}, p)
 	}

M internal/usecase/pushes.go +23 -11


@@ -17,17 +17,18 @@ import (
17	17
18	18		// Pushes are the use cases around a Git push.
19	19		type Pushes struct {
20		-	meta ports.RepoMetaStore
21		-	repos *Repos
22		-	git ports.GitRepos
23		-	bus ports.EventBus
24		-	clock ports.Clock
25		-	maint ports.MaintenanceScheduler
26		-	auth ports.Revalidator
27		-	fresh ports.RepoFreshener // nil: the reader needs no telling
28		-	ci ports.WorkflowTrigger // nil: no CI
29		-	wf ports.WorkflowGuard // nil: pushes by job tokens are refused
30		-	tips ports.CIRefTipStore // nil: who pushes a ref is not recorded
	20	+	meta ports.RepoMetaStore
	21	+	repos *Repos
	22	+	git ports.GitRepos
	23	+	bus ports.EventBus
	24	+	clock ports.Clock
	25	+	maint ports.MaintenanceScheduler
	26	+	auth ports.Revalidator
	27	+	fresh ports.RepoFreshener // nil: the reader needs no telling
	28	+	ci ports.WorkflowTrigger // nil: no CI
	29	+	wf ports.WorkflowGuard // nil: pushes by job tokens are refused
	30	+	tips ports.CIRefTipStore // nil: who pushes a ref is not recorded
	31	+	mergeab ports.MergeabilityTrigger // nil: mergeability is not tracked
31	32		}
32	33
33	34		var (

@@ -56,6 +57,14 @@ func (u Pushes) WithWorkflowTrigger(t ports.WorkflowTrigger) Pushes {
56	57		return u
57	58		}
58	59
	60	+	// WithMergeability makes Record queue the mergeability checks of the merge
	61	+	// requests whose branches the push moved. Call it before the first push; it
	62	+	// returns u.
	63	+	func (u Pushes) WithMergeability(t ports.MergeabilityTrigger) Pushes {
	64	+	u.mergeab = t
	65	+	return u
	66	+	}
	67	+
59	68		// WithRefTips makes Check record who is about to move each branch and tag
60	69		// (domain.CIRefTip), before Git moves it. Call it before the first push; it
61	70		// returns u.

@@ -230,6 +239,9 @@ func (u *Pushes) Record(ctx context.Context, p domain.Principal, name domain.Rep
230	239		if u.ci != nil {
231	240		u.ci.OnPush(ctx, name, updates, p) // queues only: never slows or fails the push
232	241		}
	242	+	if u.mergeab != nil {
	243	+	u.mergeab.OnBranchesMoved(ctx, name, updates) // queues only
	244	+	}
233	245		u.bus.Publish(ctx, domain.PushEvent{
234	246		Repo: name, Updates: append([]domain.RefUpdate(nil), updates...), At: now,
235	247		Visibility: repo.Visibility, CreatedAt: repo.CreatedAt,

 // Pushes are the use cases around a Git push.
 type Pushes struct {
-	meta  ports.RepoMetaStore
+	meta    ports.RepoMetaStore
-	repos *Repos
+	repos   *Repos
-	git   ports.GitRepos
+	git     ports.GitRepos
-	bus   ports.EventBus
+	bus     ports.EventBus
-	clock ports.Clock
+	clock   ports.Clock
-	maint ports.MaintenanceScheduler
+	maint   ports.MaintenanceScheduler
-	auth  ports.Revalidator
+	auth    ports.Revalidator
-	fresh ports.RepoFreshener   // nil: the reader needs no telling
+	fresh   ports.RepoFreshener       // nil: the reader needs no telling
-	ci    ports.WorkflowTrigger // nil: no CI
+	ci      ports.WorkflowTrigger     // nil: no CI
-	wf    ports.WorkflowGuard   // nil: pushes by job tokens are refused
+	wf      ports.WorkflowGuard       // nil: pushes by job tokens are refused
-	tips  ports.CIRefTipStore   // nil: who pushes a ref is not recorded
+	tips    ports.CIRefTipStore       // nil: who pushes a ref is not recorded
+	mergeab ports.MergeabilityTrigger // nil: mergeability is not tracked
 }
 var (
 	return u
 }
+// WithMergeability makes Record queue the mergeability checks of the merge
+// requests whose branches the push moved. Call it before the first push; it
+// returns u.
+func (u *Pushes) WithMergeability(t ports.MergeabilityTrigger) *Pushes {
+	u.mergeab = t
+	return u
+}
 // WithRefTips makes Check record who is about to move each branch and tag
 // (domain.CIRefTip), before Git moves it. Call it before the first push; it
 // returns u.
 	if u.ci != nil {
 		u.ci.OnPush(ctx, name, updates, p) // queues only: never slows or fails the push
 	}
+	if u.mergeab != nil {
+		u.mergeab.OnBranchesMoved(ctx, name, updates) // queues only
+	}
 	u.bus.Publish(ctx, domain.PushEvent{
 		Repo: name, Updates: append([]domain.RefUpdate(nil), updates...), At: now,
 		Visibility: repo.Visibility, CreatedAt: repo.CreatedAt,