1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package usecase
4
5import (
6 "context"
7 "encoding/base64"
8 "errors"
9 "fmt"
10 "strconv"
11 "strings"
12 "sync"
13
14 "github.com/bnema/zerowrap"
15
16 "bnema.dev/ermite/internal/domain"
17 "bnema.dev/ermite/internal/ports"
18)
19
20// MaxMergeRequestCommits is the most commits GetMergeRequest returns.
21const MaxMergeRequestCommits = 100
22
23// maxWriteRetries bounds the retries of a write that loses a race with another
24// write of the same merge request. Each lost race means another writer won, so
25// a writer loses at most as many times as there are other writers at once; the
26// retry is a read and a write of ermite-kv, cheap enough for a bound this high.
27const maxWriteRetries = 32
28
29// MaxCommentsPerMergeRequest is the most comments a merge request holds. Past
30// it AddComment fails with domain.ErrResourceExhausted (resource_exhausted in
31// the API). Comments are never deleted, so the limit counts the ones ever added.
32const MaxCommentsPerMergeRequest = 1000
33
34const mrTokenPrefix = "mp1:"
35
36// MergeRequests are the merge request use cases (ADR 0017): review, comment,
37// close and merge a branch. Reading needs the access of reading the
38// repository: a repository the principal may not read, a missing one and an
39// unknown number all answer domain.ErrNotFound. Writing needs mr:write, and
40// merging mr:merge, on the repository (see scopeAccess), and that scope also
41// makes a private repository visible for the action (see writable).
42type MergeRequests struct {
43 meta ports.RepoMetaStore
44 store ports.MergeRequestStore
45 repos ports.GitRepos
46 git ports.MergeGit
47 signer ports.CommitSigner
48 committer domain.Identity
49 bus ports.EventBus
50 clock ports.Clock
51 maint ports.MaintenanceScheduler
52 ci ports.WorkflowTrigger // nil: no CI
53 checks ports.CIChecksReader // nil: no CI, so no check blocks a merge
54 pulls ports.PullRequestTrigger // nil: no pull_request workflows
55
56 mu sync.Mutex
57 locks map[domain.RepoName]chan struct{}
58}
59
60var (
61 _ ports.MergeRequestService = (*MergeRequests)(nil)
62 _ ports.PullGate = (*MergeRequests)(nil)
63)
64
65// MergeRequestsConfig holds what the merge request use cases need. Every field
66// is required.
67type MergeRequestsConfig struct {
68 Meta ports.RepoMetaStore
69 Store ports.MergeRequestStore
70 // Repos gives the default branch of a repository.
71 Repos ports.GitRepos
72 Git ports.MergeGit
73 // Signer signs the squash commits.
74 Signer ports.CommitSigner
75 // Committer is recorded as the committer of the squash commits.
76 Committer domain.Identity
77 Bus ports.EventBus
78 Clock ports.Clock
79 // Maint queues the maintenance that follows a merge, as it does a push.
80 Maint ports.MaintenanceScheduler
81 // CI starts the workflows a merge triggers (GitHub runs "push" workflows
82 // on a merge) and Checks gives the checks the merge rule reads. Both are
83 // optional: without them there is no CI.
84 CI ports.WorkflowTrigger
85 Checks ports.CIChecksReader
86 // PullRequests starts the pull_request workflows when a request is opened,
87 // closed or merged (the push of its source branch reaches CI instead).
88 // Optional.
89 PullRequests ports.PullRequestTrigger
90}
91
92// NewMergeRequests builds the merge request use cases.
93func NewMergeRequests(cfg MergeRequestsConfig) *MergeRequests {
94 return &MergeRequests{
95 meta: cfg.Meta, store: cfg.Store, repos: cfg.Repos, git: cfg.Git, signer: cfg.Signer,
96 committer: cfg.Committer, bus: cfg.Bus, clock: cfg.Clock, maint: cfg.Maint, ci: cfg.CI, checks: cfg.Checks, pulls: cfg.PullRequests,
97 locks: make(map[domain.RepoName]chan struct{}),
98 }
99}
100
101// lockRepo takes the merge lock of a repository, waiting until ctx ends. The
102// merges of one repository, and the creation of its merge requests, run one at
103// a time. The returned function releases the lock.
104func (m *MergeRequests) lockRepo(ctx context.Context, name domain.RepoName) (unlock func(), err error) {
105 m.mu.Lock()
106 ch, ok := m.locks[name]
107 if !ok {
108 ch = make(chan struct{}, 1)
109 m.locks[name] = ch
110 }
111 m.mu.Unlock()
112 select {
113 case ch <- struct{}{}:
114 return func() { <-ch }, nil
115 case <-ctx.Done():
116 return nil, ctx.Err()
117 }
118}
119
120// readable looks the repository up and applies the read rule: reading merge
121// requests needs the access of reading the repository (canRead). repo:write
122// alone does not read them, as it does not fetch a private repository either.
123func (m *MergeRequests) readable(ctx context.Context, p domain.Principal, name domain.RepoName) (domain.Repo, error) {
124 repo, err := lookupRepo(ctx, m.meta, name)
125 if err != nil {
126 return domain.Repo{}, err
127 }
128 if !canRead(p, repo) {
129 return domain.Repo{}, domain.ErrNotFound
130 }
131 return repo, nil
132}
133
134// writable looks the repository up for an action that needs a merge request
135// scope. A principal whose token holds that scope on the repository sees the
136// repository for this action even without repo:read, so that the agent preset
137// (repo:write, mr:write) can open a request on the private repository it just
138// pushed. A principal with neither repo:read nor the scope, and an unknown
139// repository, still get domain.ErrNotFound; one that can read the repository
140// but lacks the scope gets the error of scopeAccess.
141func (m *MergeRequests) writable(ctx context.Context, p domain.Principal, name domain.RepoName, scope domain.Scope) (domain.Repo, error) {
142 repo, err := lookupRepo(ctx, m.meta, name)
143 if err != nil {
144 return domain.Repo{}, err
145 }
146 if !canRead(p, repo) && !p.Allows(scope, name) {
147 return domain.Repo{}, domain.ErrNotFound
148 }
149 if err := scopeAccess(p, repo, scope); err != nil {
150 return domain.Repo{}, err
151 }
152 return repo, nil
153}
154
155// withHead fills HeadID of an open request from its source branch: zero when
156// the branch is gone. Merged and closed requests keep the head they stored.
157func (m *MergeRequests) withHead(ctx context.Context, mr domain.MergeRequest) (domain.MergeRequest, error) {
158 if !mr.IsOpen() {
159 return mr, nil
160 }
161 head, err := m.git.BranchHead(ctx, mr.Repo, mr.Source)
162 switch {
163 case errors.Is(err, domain.ErrNotFound):
164 mr.HeadID = domain.ObjectID{}
165 case err != nil:
166 return domain.MergeRequest{}, err
167 default:
168 mr.HeadID = head
169 }
170 return mr, nil
171}
172
173// branchOrZero returns the head of a branch, the zero id when it is gone.
174func (m *MergeRequests) branchOrZero(ctx context.Context, repo domain.RepoName, branch string) (domain.ObjectID, error) {
175 id, err := m.git.BranchHead(ctx, repo, branch)
176 if errors.Is(err, domain.ErrNotFound) {
177 return domain.ObjectID{}, nil
178 }
179 return id, err
180}
181
182// Create opens a merge request and publishes MergeRequestOpened. The principal
183// needs mr:write. A request that breaks a rule of ADR 0017 (the texts, a
184// branch that does not exist, source and target the same) is rejected whole
185// with an *domain.InvalidFieldsError naming the fields. An open request with
186// the same source and target gives domain.ErrAlreadyExists.
187func (m *MergeRequests) Create(ctx context.Context, p domain.Principal, name domain.RepoName, params ports.CreateMergeRequestParams) (domain.MergeRequest, error) {
188 repo, err := m.writable(ctx, p, name, domain.ScopeMRWrite)
189 if err != nil {
190 return domain.MergeRequest{}, err
191 }
192 var v domain.Violations
193 title, err := domain.ParseMRTitle(params.Title)
194 if err != nil {
195 v.AddErr("title", err)
196 }
197 desc, err := domain.ParsePlainText(params.Description)
198 if err != nil {
199 v.AddErr("description", err)
200 }
201 // Under the lock, so that two requests for the same pair cannot both pass
202 // the check for an existing one, and a merge that deletes its source
203 // cannot remove a branch between its check here and the creation.
204 unlock, err := m.lockRepo(ctx, name)
205 if err != nil {
206 return domain.MergeRequest{}, err
207 }
208 defer unlock()
209 source, target, err := m.checkBranches(ctx, name, params, &v)
210 if err != nil {
211 return domain.MergeRequest{}, err
212 }
213 if err := v.Err(); err != nil {
214 return domain.MergeRequest{}, err
215 }
216 existing, err := m.store.List(ctx, name)
217 if err != nil {
218 return domain.MergeRequest{}, fmt.Errorf("list merge requests of %q: %w", name, err)
219 }
220 for _, e := range existing {
221 if e.IsOpen() && e.Source == source && e.Target == target {
222 return domain.MergeRequest{}, fmt.Errorf("%w: !%d already merges %s into %s", domain.ErrAlreadyExists, e.Number, source, target)
223 }
224 }
225
226 now := m.clock.Now().UTC()
227 mr, err := m.store.Create(ctx, domain.MergeRequest{
228 Repo: name, Title: title, Description: desc, Source: source, Target: target,
229 State: domain.MRStateOpen, Author: authorOf(p), CreatedAt: now, UpdatedAt: now,
230 })
231 if err != nil {
232 return domain.MergeRequest{}, fmt.Errorf("create merge request of %q: %w", name, err)
233 }
234 log := zerowrap.FromCtx(ctx)
235 log.Info().Str("repo", name.String()).Int64("number", mr.Number).
236 Str("source", source).Str("target", target).
237 Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("merge request opened")
238 mr, err = m.withHead(ctx, mr)
239 if err != nil {
240 return domain.MergeRequest{}, err
241 }
242 m.bus.Publish(ctx, domain.MergeRequestOpened{MergeRequest: mr, At: now, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt})
243 if m.pulls != nil {
244 // The commits the opening is for are the ones of now.
245 base, _ := m.branchOrZero(ctx, name, target)
246 m.pulls.OnPullRequest(ctx, name, ports.PullRequestEvent{Number: mr.Number, Action: domain.PullOpened, Head: mr.HeadID, Base: base}, p)
247 }
248 return mr, nil
249}
250
251// checkBranches validates the branches of a new request and returns them, with
252// the default branch filled in for the target.
253//
254// Only a branch that does not exist is a violation. Any other failure to read
255// Git is returned as an error, so that it is logged and masked, not shown in
256// the field violations.
257func (m *MergeRequests) checkBranches(ctx context.Context, name domain.RepoName, params ports.CreateMergeRequestParams, v *domain.Violations) (source, target string, err error) {
258 source, target = params.Source, params.Target
259 if source == "" {
260 v.Add("source_branch", "is required")
261 } else if _, err := domain.ParseBranchName(source); err != nil {
262 v.Add("source_branch", "is not a valid branch name")
263 source = ""
264 }
265 if target == "" {
266 def, exists, err := m.repos.DefaultBranch(ctx, name)
267 switch {
268 case err != nil: // includes domain.ErrNotFound: the repository is gone
269 return "", "", fmt.Errorf("read the default branch of %q: %w", name, err)
270 case !exists:
271 v.Add("target_branch", fmt.Sprintf("is empty and the default branch %q does not exist", def))
272 default:
273 target = def
274 }
275 } else if _, err := domain.ParseBranchName(target); err != nil {
276 v.Add("target_branch", "is not a valid branch name")
277 target = ""
278 }
279 if source != "" && source == target {
280 v.Add("target_branch", "must differ from source_branch")
281 return source, target, nil
282 }
283 for _, b := range []struct{ field, name string }{{"source_branch", source}, {"target_branch", target}} {
284 if b.name == "" {
285 continue
286 }
287 _, err := m.git.BranchHead(ctx, name, b.name)
288 if errors.Is(err, domain.ErrNotFound) {
289 v.Add(b.field, fmt.Sprintf("branch %q does not exist", b.name))
290 } else if err != nil {
291 return "", "", fmt.Errorf("read branch %q of %q: %w", b.name, name, err)
292 }
293 }
294 return source, target, nil
295}
296
297// List returns a page of the merge requests of a repository, highest number
298// first. The open ones carry the head of their source branch. The page token
299// names the last number of the previous page, so requests created between two
300// pages never shift the others.
301func (m *MergeRequests) List(ctx context.Context, p domain.Principal, name domain.RepoName, params ports.ListMergeRequestsParams) (ports.MergeRequestPage, error) {
302 size, err := pageSize(params.PageSize)
303 if err != nil {
304 return ports.MergeRequestPage{}, err
305 }
306 after, err := decodeMRToken(params.PageToken)
307 if err != nil {
308 return ports.MergeRequestPage{}, err
309 }
310 if _, err := m.readable(ctx, p, name); err != nil {
311 return ports.MergeRequestPage{}, err
312 }
313 all, err := m.store.List(ctx, name)
314 if err != nil {
315 return ports.MergeRequestPage{}, fmt.Errorf("list merge requests of %q: %w", name, err)
316 }
317 var page ports.MergeRequestPage
318 page.MergeRequests, page.NextPageToken = pageAfter(all, size,
319 func(mr domain.MergeRequest) bool {
320 return (params.State == "" || mr.State == params.State) && (after == 0 || mr.Number < after)
321 },
322 encodeMRToken)
323 for i, mr := range page.MergeRequests {
324 if page.MergeRequests[i], err = m.withHead(ctx, mr); err != nil {
325 return ports.MergeRequestPage{}, err
326 }
327 }
328 return page, nil
329}
330
331func encodeMRToken(last domain.MergeRequest) string {
332 return base64.RawURLEncoding.EncodeToString([]byte(mrTokenPrefix + strconv.FormatInt(last.Number, 10)))
333}
334
335// decodeMRToken returns 0 for the empty token.
336func decodeMRToken(tok string) (int64, error) {
337 if tok == "" {
338 return 0, nil
339 }
340 raw, ok := decodeToken(tok, mrTokenPrefix)
341 if ok {
342 if n, err := strconv.ParseInt(raw, 10, 64); err == nil && n > 0 {
343 return n, nil
344 }
345 }
346 return 0, fmt.Errorf("%w: merge requests", domain.ErrInvalidPageToken)
347}
348
349// get reads a merge request of a repository the principal can read.
350func (m *MergeRequests) get(ctx context.Context, p domain.Principal, name domain.RepoName, number int64) (domain.Repo, domain.MergeRequest, error) {
351 repo, err := m.readable(ctx, p, name)
352 if err != nil {
353 return domain.Repo{}, domain.MergeRequest{}, err
354 }
355 mr, err := m.store.Get(ctx, name, number)
356 if err != nil {
357 return domain.Repo{}, domain.MergeRequest{}, err
358 }
359 return repo, mr, nil
360}
361
362// Get returns a merge request with its commits, its diff, its comments and the
363// read position. The diff compares the merge base of the target and the head
364// with the head ("target...head"), within domain.DefaultDiffLimits. With
365// sinceLastRead it compares the head recorded by MarkRead with the current
366// head instead, which works after a force-push as long as the recorded commit
367// still exists; when none was recorded the diff is the default one, and when
368// the recorded commit is gone too, with LastReadLost set. A merged or closed
369// request keeps the head and the base it had.
370func (m *MergeRequests) Get(ctx context.Context, p domain.Principal, name domain.RepoName, number int64, sinceLastRead bool) (ports.MergeRequestDetail, error) {
371 _, mr, err := m.get(ctx, p, name, number)
372 if err != nil {
373 return ports.MergeRequestDetail{}, err
374 }
375 if mr, err = m.withHead(ctx, mr); err != nil {
376 return ports.MergeRequestDetail{}, err
377 }
378 d := ports.MergeRequestDetail{MergeRequest: mr}
379 if d.Comments, err = m.store.ListComments(ctx, name, number); err != nil {
380 return ports.MergeRequestDetail{}, fmt.Errorf("list comments of !%d in %q: %w", number, name, err)
381 }
382 if d.LastReadID, err = m.store.GetRead(ctx, name, number); err != nil {
383 return ports.MergeRequestDetail{}, err
384 }
385
386 exclude, base, err := m.rangeOf(ctx, mr)
387 if err != nil {
388 return ports.MergeRequestDetail{}, err
389 }
390 head := mr.HeadID
391 if head.IsZero() {
392 return d, nil // the source branch is gone: no commits, no diff
393 }
394 if !mr.IsOpen() {
395 // After a squash the reviewed commits are only reachable from the
396 // source branch; once it is deleted, gc may prune them.
397 switch ok, err := m.git.HasCommit(ctx, name, head); {
398 case err != nil:
399 return ports.MergeRequestDetail{}, err
400 case !ok:
401 return d, nil // the reviewed commits are gone: no commits, no diff
402 }
403 }
404 if d.Commits, d.CommitsTruncated, err = m.git.ListRange(ctx, name, exclude, head, MaxMergeRequestCommits); err != nil {
405 return ports.MergeRequestDetail{}, err
406 }
407 if sinceLastRead && !d.LastReadID.IsZero() {
408 switch ok, err := m.git.HasCommit(ctx, name, d.LastReadID); {
409 case err != nil:
410 return ports.MergeRequestDetail{}, err
411 case ok:
412 base = d.LastReadID
413 default:
414 d.LastReadLost = true
415 }
416 }
417 if base.IsZero() {
418 return d, nil
419 }
420 d.DiffBaseID = base
421 if d.Stats, d.Diff, err = m.git.DiffCommits(ctx, name, base, head, domain.DefaultDiffLimits()); err != nil {
422 return ports.MergeRequestDetail{}, err
423 }
424 return d, nil
425}
426
427// rangeOf returns what the review of a request compares: the commit whose
428// history the request's commits exclude, and the base of its diff. For an open
429// request they are the tip of the target and the merge base of the target and
430// the head; both are zero when the target is gone or shares no history with
431// the head. A merged or closed request keeps the base it stored for both.
432func (m *MergeRequests) rangeOf(ctx context.Context, mr domain.MergeRequest) (exclude, base domain.ObjectID, err error) {
433 if !mr.IsOpen() {
434 return mr.BaseID, mr.BaseID, nil
435 }
436 if mr.HeadID.IsZero() {
437 return domain.ObjectID{}, domain.ObjectID{}, nil
438 }
439 tip, err := m.branchOrZero(ctx, mr.Repo, mr.Target)
440 if err != nil || tip.IsZero() {
441 return domain.ObjectID{}, domain.ObjectID{}, err
442 }
443 base, err = m.git.MergeBase(ctx, mr.Repo, tip, mr.HeadID)
444 if errors.Is(err, domain.ErrNotFound) {
445 return tip, domain.ObjectID{}, nil
446 }
447 return tip, base, err
448}
449
450// AddComment adds a comment and publishes CommentAdded. The principal needs
451// mr:write. The text must follow the rules of ADR 0017 and not be blank; a
452// comment on a line must name a line of the current diff of the request (see
453// domain.Diff.HasLine), else the call fails with domain.ErrInvalid. A request
454// that is not open gives domain.ErrPreconditionFailed, and one that already
455// holds MaxCommentsPerMergeRequest comments domain.ErrResourceExhausted.
456func (m *MergeRequests) AddComment(ctx context.Context, p domain.Principal, name domain.RepoName, number int64, params ports.AddCommentParams) (domain.Comment, error) {
457 repo, err := m.writable(ctx, p, name, domain.ScopeMRWrite)
458 if err != nil {
459 return domain.Comment{}, err
460 }
461 mr, err := m.store.Get(ctx, name, number)
462 if err != nil {
463 return domain.Comment{}, err
464 }
465 var v domain.Violations
466 text, err := domain.ParsePlainText(params.Text)
467 switch {
468 case err != nil:
469 v.AddErr("text", err)
470 case text.IsBlank():
471 v.Add("text", "is required")
472 }
473 general := params.Path == ""
474 switch {
475 case general && (params.Line != 0 || params.Side != domain.SideNone):
476 v.Add("path", "is required with line and side")
477 case !general && params.Line < 1:
478 v.Add("line", "must be 1 or more with a path")
479 case !general && params.Side != domain.SideOld && params.Side != domain.SideNew:
480 v.Add("side", "is required with a path")
481 }
482 if err := v.Err(); err != nil {
483 return domain.Comment{}, err
484 }
485 if !mr.IsOpen() {
486 return domain.Comment{}, fmt.Errorf("%w: !%d is %s", domain.ErrPreconditionFailed, number, mr.State)
487 }
488
489 if mr, err = m.withHead(ctx, mr); err != nil {
490 return domain.Comment{}, err
491 }
492 if !general {
493 if err := m.checkLine(ctx, mr, params); err != nil {
494 return domain.Comment{}, err
495 }
496 }
497
498 var c domain.Comment
499 for range maxWriteRetries {
500 if mr.CommentSeq >= MaxCommentsPerMergeRequest {
501 return domain.Comment{}, fmt.Errorf("%w: !%d already has %d comments", domain.ErrResourceExhausted, number, MaxCommentsPerMergeRequest)
502 }
503 c = domain.Comment{
504 ID: mr.CommentSeq + 1, Author: authorOf(p), Path: params.Path, Line: params.Line, Side: params.Side,
505 CommitID: mr.HeadID, Text: text, CreatedAt: m.clock.Now().UTC(),
506 }
507 mr.CommentSeq, mr.UpdatedAt = c.ID, c.CreatedAt
508 err = m.store.SaveComment(ctx, mr, c)
509 if !errors.Is(err, domain.ErrConflict) {
510 break
511 }
512 // Another write got in between: read the request again.
513 fresh, gerr := m.store.Get(ctx, name, number)
514 if gerr != nil {
515 return domain.Comment{}, gerr
516 }
517 if !fresh.IsOpen() {
518 return domain.Comment{}, fmt.Errorf("%w: !%d is %s", domain.ErrPreconditionFailed, number, fresh.State)
519 }
520 fresh.HeadID = mr.HeadID
521 mr = fresh
522 }
523 if err != nil {
524 return domain.Comment{}, fmt.Errorf("save comment on !%d of %q: %w", number, name, err)
525 }
526 log := zerowrap.FromCtx(ctx)
527 log.Info().Str("repo", name.String()).Int64("number", number).Int64("comment", c.ID).
528 Bool("on_line", !general).Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("comment added")
529 m.bus.Publish(ctx, domain.CommentAdded{
530 Repo: name, Number: number, Comment: c, At: c.CreatedAt, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt,
531 })
532 return c, nil
533}
534
535// checkLine verifies that a comment designates a line of the current diff of
536// the request.
537func (m *MergeRequests) checkLine(ctx context.Context, mr domain.MergeRequest, params ports.AddCommentParams) error {
538 var v domain.Violations
539 _, base, err := m.rangeOf(ctx, mr)
540 if err != nil {
541 return err
542 }
543 if mr.HeadID.IsZero() || base.IsZero() {
544 v.Add("path", "the merge request has no diff to comment on")
545 return v.Err()
546 }
547 _, diff, err := m.git.DiffCommits(ctx, mr.Repo, base, mr.HeadID, domain.DefaultDiffLimits())
548 if err != nil {
549 return err
550 }
551 if !diff.HasLine(params.Path, params.Side, params.Line) {
552 v.Add("path", fmt.Sprintf("%s:%d on the %s side is not a line of the diff", params.Path, params.Line, params.Side))
553 }
554 return v.Err()
555}
556
557// ResolveComment marks a comment resolved or unresolved. The principal needs
558// mr:write. An unknown comment gives domain.ErrNotFound and a request that is
559// not open domain.ErrPreconditionFailed. Setting the state the comment already
560// has succeeds and writes nothing.
561func (m *MergeRequests) ResolveComment(ctx context.Context, p domain.Principal, name domain.RepoName, number, commentID int64, resolved bool) (domain.Comment, error) {
562 if _, err := m.writable(ctx, p, name, domain.ScopeMRWrite); err != nil {
563 return domain.Comment{}, err
564 }
565 for range maxWriteRetries {
566 mr, err := m.store.Get(ctx, name, number)
567 if err != nil {
568 return domain.Comment{}, err
569 }
570 c, err := m.store.GetComment(ctx, name, number, commentID)
571 if err != nil {
572 return domain.Comment{}, err
573 }
574 if !mr.IsOpen() {
575 return domain.Comment{}, fmt.Errorf("%w: !%d is %s", domain.ErrPreconditionFailed, number, mr.State)
576 }
577 if c.Resolved == resolved {
578 return c, nil
579 }
580 c.Resolved = resolved
581 mr.UpdatedAt = m.clock.Now().UTC()
582 err = m.store.SaveComment(ctx, mr, c)
583 if errors.Is(err, domain.ErrConflict) {
584 continue
585 }
586 if err != nil {
587 return domain.Comment{}, fmt.Errorf("resolve comment %d of !%d in %q: %w", commentID, number, name, err)
588 }
589 return c, nil
590 }
591 return domain.Comment{}, fmt.Errorf("resolve comment %d of !%d in %q: %w", commentID, number, name, domain.ErrConflict)
592}
593
594// Close closes an open merge request without merging it and publishes
595// MergeRequestClosed. The principal needs mr:write. A request that is not open
596// gives domain.ErrPreconditionFailed. The request keeps the head it had and
597// its merge base, so that its diff stays readable.
598func (m *MergeRequests) Close(ctx context.Context, p domain.Principal, name domain.RepoName, number int64) (domain.MergeRequest, error) {
599 repo, err := m.writable(ctx, p, name, domain.ScopeMRWrite)
600 if err != nil {
601 return domain.MergeRequest{}, err
602 }
603 unlock, err := m.lockRepo(ctx, name) // a merge and a close of one request exclude each other
604 if err != nil {
605 return domain.MergeRequest{}, err
606 }
607 defer unlock()
608 for range maxWriteRetries {
609 mr, err := m.store.Get(ctx, name, number)
610 if err != nil {
611 return domain.MergeRequest{}, err
612 }
613 if !mr.IsOpen() {
614 return domain.MergeRequest{}, fmt.Errorf("%w: !%d is %s", domain.ErrPreconditionFailed, number, mr.State)
615 }
616 if mr, err = m.withHead(ctx, mr); err != nil {
617 return domain.MergeRequest{}, err
618 }
619 _, base, err := m.rangeOf(ctx, mr)
620 if err != nil {
621 return domain.MergeRequest{}, err
622 }
623 mr.State, mr.BaseID, mr.UpdatedAt = domain.MRStateClosed, base, m.clock.Now().UTC()
624 err = m.store.Update(ctx, mr)
625 if errors.Is(err, domain.ErrConflict) {
626 continue
627 }
628 if err != nil {
629 return domain.MergeRequest{}, fmt.Errorf("close !%d of %q: %w", number, name, err)
630 }
631 log := zerowrap.FromCtx(ctx)
632 log.Info().Str("repo", name.String()).Int64("number", number).
633 Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("merge request closed")
634 m.bus.Publish(ctx, domain.MergeRequestClosed{MergeRequest: mr, At: mr.UpdatedAt, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt})
635 if m.pulls != nil {
636 m.pulls.OnPullRequest(ctx, name, ports.PullRequestEvent{Number: number, Action: domain.PullClosed, Head: mr.HeadID, Base: mr.BaseID}, p)
637 }
638 return mr, nil
639 }
640 return domain.MergeRequest{}, fmt.Errorf("close !%d of %q: %w", number, name, domain.ErrConflict)
641}
642
643// MarkRead records the head the reviewer has read, for the since-last-read diff
644// of Get. The principal needs mr:write. A zero head means the current head of
645// the source branch (the stored one for a merged or closed request); one that
646// is not a commit of the repository gives domain.ErrInvalid. It returns the
647// head recorded.
648func (m *MergeRequests) MarkRead(ctx context.Context, p domain.Principal, name domain.RepoName, number int64, head domain.ObjectID) (domain.ObjectID, error) {
649 if _, err := m.writable(ctx, p, name, domain.ScopeMRWrite); err != nil {
650 return domain.ObjectID{}, err
651 }
652 mr, err := m.store.Get(ctx, name, number)
653 if err != nil {
654 return domain.ObjectID{}, err
655 }
656 if head.IsZero() {
657 if mr, err = m.withHead(ctx, mr); err != nil {
658 return domain.ObjectID{}, err
659 }
660 if head = mr.HeadID; head.IsZero() {
661 return domain.ObjectID{}, fmt.Errorf("%w: the source branch %q no longer exists, give a head", domain.ErrPreconditionFailed, mr.Source)
662 }
663 } else {
664 var v domain.Violations
665 switch ok, err := m.git.HasCommit(ctx, name, head); {
666 case err != nil:
667 return domain.ObjectID{}, err
668 case !ok:
669 v.Add("head_id", fmt.Sprintf("%s is not a commit of the repository", head))
670 return domain.ObjectID{}, v.Err()
671 }
672 }
673 if err := m.store.SetRead(ctx, name, number, head); err != nil {
674 return domain.ObjectID{}, err
675 }
676 return head, nil
677}
678
679// Merge merges an open merge request, publishes MergeRequestMerged then a
680// PushEvent for the target branch, and returns the merged request. The
681// principal needs mr:merge. Merges of one repository run one at a time.
682//
683// It fails with domain.ErrInvalid for a zero head, and with
684// domain.ErrPreconditionFailed when the request is not open, the mode is not
685// allowed by the settings of the repository, the source branch no longer
686// points to the head ("branch moved, read again") or is gone, the target is
687// gone or is not an ancestor of the head (the author rebases and pushes
688// again), or the branch brings nothing. Checks come with CI (CI-1): there are
689// none yet, so the setting that lets failing checks block the merge has
690// nothing to act on.
691//
692// A fast-forward moves the target to the head, with the old value as a
693// precondition. A squash writes one commit with the tree of the head, the
694// target as parent and the author and message of domain.NewSquashCommit,
695// signed by the instance key, and moves the target to it the same way.
696//
697// With DeleteSource the merge is also refused, before anything changes, when
698// the source is the default branch or the source or target of another open
699// request. After the merge the source is deleted if it still points to the
700// head; the PushEvent then carries the deletion too. A failed deletion is
701// logged and leaves the branch: the merge stands and SourceDeleted is false.
702func (m *MergeRequests) Merge(ctx context.Context, p domain.Principal, name domain.RepoName, number int64, params ports.MergeParams) (ports.MergeResult, error) {
703 mr, deleted, err := m.merge(ctx, p, name, number, params)
704 return ports.MergeResult{MergeRequest: mr, SourceDeleted: deleted}, err
705}
706
707func (m *MergeRequests) merge(ctx context.Context, p domain.Principal, name domain.RepoName, number int64, params ports.MergeParams) (domain.MergeRequest, bool, error) {
708 repo, err := m.writable(ctx, p, name, domain.ScopeMRMerge)
709 if err != nil {
710 return domain.MergeRequest{}, false, err
711 }
712 mode := params.Mode
713 if mode == "" {
714 mode = domain.MergeModeFastForward
715 }
716 var v domain.Violations
717 if _, err := domain.ParseMergeMode(mode.String()); err != nil {
718 v.Add("mode", "is not a merge mode")
719 }
720 if params.HeadID.IsZero() {
721 v.Add("head_id", "is required")
722 }
723 if err := v.Err(); err != nil {
724 return domain.MergeRequest{}, false, err
725 }
726
727 unlock, err := m.lockRepo(ctx, name)
728 if err != nil {
729 return domain.MergeRequest{}, false, err
730 }
731 defer unlock()
732
733 mr, err := m.store.Get(ctx, name, number)
734 if err != nil {
735 return domain.MergeRequest{}, false, err
736 }
737 // A merge moves the target and may delete the source: a token limited to
738 // branch prefixes may do so only within them.
739 if !p.AllowsRef("refs/heads/" + mr.Target) {
740 return domain.MergeRequest{}, false, fmt.Errorf("%w: this token may not write %s", domain.ErrForbidden, mr.Target)
741 }
742 if params.DeleteSource && !p.AllowsRef("refs/heads/"+mr.Source) {
743 return domain.MergeRequest{}, false, fmt.Errorf("%w: this token may not write %s", domain.ErrForbidden, mr.Source)
744 }
745 if !mr.IsOpen() {
746 return domain.MergeRequest{}, false, fmt.Errorf("%w: !%d is %s", domain.ErrPreconditionFailed, number, mr.State)
747 }
748 // The settings are read again under the lock: a change applies at once.
749 repo, err = m.meta.Get(ctx, name)
750 if err != nil {
751 if errors.Is(err, domain.ErrNotFound) {
752 return domain.MergeRequest{}, false, err // deleted meanwhile
753 }
754 return domain.MergeRequest{}, false, fmt.Errorf("get repository %q: %w", name, err)
755 }
756 settings := repo.Settings
757 if settings.IsZero() {
758 settings = domain.DefaultRepoSettings()
759 }
760 if !settings.Allows(mode) {
761 return domain.MergeRequest{}, false, fmt.Errorf("%w: the %s merge mode is not allowed in %s", domain.ErrPreconditionFailed, mode, name)
762 }
763
764 head, err := m.branchOrZero(ctx, name, mr.Source)
765 switch {
766 case err != nil:
767 return domain.MergeRequest{}, false, err
768 case head.IsZero():
769 return domain.MergeRequest{}, false, fmt.Errorf("%w: the source branch %q no longer exists", domain.ErrPreconditionFailed, mr.Source)
770 case head != params.HeadID:
771 return domain.MergeRequest{}, false, fmt.Errorf("%w: branch moved, read again: %q is at %s, not %s", domain.ErrPreconditionFailed, mr.Source, head, params.HeadID)
772 }
773 tip, err := m.branchOrZero(ctx, name, mr.Target)
774 switch {
775 case err != nil:
776 return domain.MergeRequest{}, false, err
777 case tip.IsZero():
778 return domain.MergeRequest{}, false, fmt.Errorf("%w: the target branch %q no longer exists", domain.ErrPreconditionFailed, mr.Target)
779 case tip == head:
780 return domain.MergeRequest{}, false, fmt.Errorf("%w: nothing to merge, %q already is at %s", domain.ErrPreconditionFailed, mr.Target, head)
781 }
782 switch ok, err := m.git.IsAncestor(ctx, name, tip, head); {
783 case err != nil:
784 return domain.MergeRequest{}, false, err
785 case !ok:
786 return domain.MergeRequest{}, false, fmt.Errorf("%w: %q is not an ancestor of %q: rebase it on %q and push again", domain.ErrPreconditionFailed, mr.Target, mr.Source, mr.Target)
787 }
788 if settings.FailingChecksBlock {
789 if err := m.checksAllow(ctx, name, head); err != nil {
790 return domain.MergeRequest{}, false, err
791 }
792 }
793 if params.DeleteSource {
794 if err := m.checkDeletable(ctx, name, mr); err != nil {
795 return domain.MergeRequest{}, false, err
796 }
797 }
798
799 merged, err := m.moveTarget(ctx, mr, mode, tip, head)
800 if err != nil {
801 return domain.MergeRequest{}, false, err
802 }
803 m.recordPush(ctx, name) // the target moved, whatever happens next
804 mr, err = m.markMerged(ctx, name, number, mode, tip, head, merged)
805 if err != nil {
806 return domain.MergeRequest{}, false, err
807 }
808 log := zerowrap.FromCtx(ctx)
809 log.Info().Str("repo", name.String()).Int64("number", number).Str("mode", mode.String()).
810 Str("target", mr.Target).Str("commit", merged.String()).
811 Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("merge request merged")
812 updates := []domain.RefUpdate{{Old: tip.String(), New: merged.String(), Ref: "refs/heads/" + mr.Target}}
813 deleted := false
814 if params.DeleteSource {
815 // The merge is done and recorded: a failure here keeps the branch.
816 switch err := m.git.DeleteBranch(context.WithoutCancel(ctx), name, mr.Source, head); {
817 case err != nil:
818 log.Warn().Err(err).Str("repo", name.String()).Int64("number", number).Str("branch", mr.Source).
819 Msg("source branch kept after the merge")
820 default:
821 deleted = true
822 updates = append(updates, domain.RefUpdate{Old: head.String(), New: domain.ZeroOID, Ref: "refs/heads/" + mr.Source})
823 log.Info().Str("repo", name.String()).Int64("number", number).Str("branch", mr.Source).
824 Str("principal", p.Label()).Str("token_id", p.TokenID()).Msg("source branch deleted after the merge")
825 }
826 }
827 at := mr.UpdatedAt
828 m.bus.Publish(ctx, domain.MergeRequestMerged{MergeRequest: mr, At: at, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt})
829 m.bus.Publish(ctx, domain.PushEvent{
830 Repo: name, At: at, Visibility: repo.Visibility, CreatedAt: repo.CreatedAt, Updates: updates,
831 })
832 if m.pulls != nil {
833 m.pulls.OnPullRequest(ctx, name, ports.PullRequestEvent{Number: number, Action: domain.PullClosed, Head: mr.HeadID, Base: mr.BaseID}, p) // closed, merged: true
834 }
835 if m.ci != nil {
836 // GitHub runs the "push" workflows of a merge, with the owner as actor;
837 // the run is trusted with what the merging principal may read.
838 m.ci.OnMerge(ctx, name, updates[:1], p)
839 }
840 return mr, deleted, nil
841}
842
843// WhileOpen implements ports.PullGate: fn runs under the merge lock, so a
844// close or a merge of the request cannot complete in the middle of it, and not
845// at all when the request is no longer open.
846func (m *MergeRequests) WhileOpen(ctx context.Context, name domain.RepoName, number int64, fn func() error) error {
847 unlock, err := m.lockRepo(ctx, name)
848 if err != nil {
849 return err
850 }
851 defer unlock()
852 mr, err := m.store.Get(ctx, name, number)
853 if err != nil {
854 return err
855 }
856 if !mr.IsOpen() {
857 return fmt.Errorf("%w: !%d is %s", domain.ErrPreconditionFailed, number, mr.State)
858 }
859 return fn()
860}
861
862// checksAllow is the merge rule of FailingChecksBlock: every check of the
863// head must have ended as success, skipped or neutral. A check still queued or
864// running blocks too (the merge could land code that then fails), and a head
865// without any check merges.
866func (m *MergeRequests) checksAllow(ctx context.Context, name domain.RepoName, head domain.ObjectID) error {
867 if m.checks == nil {
868 return nil
869 }
870 checks, err := m.checks.CommitChecks(ctx, name, head.String())
871 if errors.Is(err, domain.ErrResourceExhausted) {
872 return fmt.Errorf("%w: too many checks to tell whether they passed: %v", domain.ErrPreconditionFailed, err)
873 }
874 if err != nil {
875 return fmt.Errorf("read the checks of %s: %w", head, err)
876 }
877 var failing, pending []string
878 for _, c := range checks {
879 switch {
880 case c.Status != domain.CICompleted:
881 pending = append(pending, c.Name)
882 case !c.Conclusion.PassesMerge():
883 failing = append(failing, c.Name)
884 }
885 }
886 switch {
887 case len(failing) > 0:
888 return fmt.Errorf("%w: failing checks block the merge: %s (turn off failing_checks_block to merge anyway)", domain.ErrPreconditionFailed, strings.Join(failing, ", "))
889 case len(pending) > 0:
890 return fmt.Errorf("%w: checks have not finished: %s", domain.ErrPreconditionFailed, strings.Join(pending, ", "))
891 }
892 return nil
893}
894
895// checkDeletable refuses to delete the source branch of mr when it is the
896// default branch or the source or target of another open request. It runs
897// under the merge lock, which Create also holds while it checks its branches
898// and stores the request.
899func (m *MergeRequests) checkDeletable(ctx context.Context, name domain.RepoName, mr domain.MergeRequest) error {
900 def, _, err := m.repos.DefaultBranch(ctx, name)
901 if err != nil {
902 return fmt.Errorf("default branch of %q: %w", name, err)
903 }
904 if mr.Source == def {
905 return fmt.Errorf("%w: %q is the default branch of %s: it cannot be deleted", domain.ErrPreconditionFailed, mr.Source, name)
906 }
907 all, err := m.store.List(ctx, name)
908 if err != nil {
909 return fmt.Errorf("list merge requests of %q: %w", name, err)
910 }
911 for _, o := range all {
912 if o.Number != mr.Number && o.IsOpen() && (o.Source == mr.Source || o.Target == mr.Source) {
913 return fmt.Errorf("%w: !%d still uses %q: it cannot be deleted", domain.ErrPreconditionFailed, o.Number, mr.Source)
914 }
915 }
916 return nil
917}
918
919// recordPush does for a merge what Pushes.Record does for a push, as the merge
920// moved a branch: it sets pushed_at and queues the maintenance of the
921// repository. The target already moved, so a failure is logged, not returned,
922// and the caller giving up does not stop it.
923func (m *MergeRequests) recordPush(ctx context.Context, name domain.RepoName) {
924 ctx = context.WithoutCancel(ctx)
925 if _, err := touchPushedAt(ctx, m.meta, name, m.clock.Now().UTC()); err != nil {
926 log := zerowrap.FromCtx(ctx)
927 log.Error().Err(err).Str("repo", name.String()).Msg("record pushed_at after a merge")
928 }
929 m.maint.Schedule(ctx, name)
930}
931
932// moveTarget moves the target branch from tip to the head (fast-forward) or to
933// a new signed squash commit, and returns the new tip.
934func (m *MergeRequests) moveTarget(ctx context.Context, mr domain.MergeRequest, mode domain.MergeMode, tip, head domain.ObjectID) (domain.ObjectID, error) {
935 newTip := head
936 if mode == domain.MergeModeSquash {
937 commits, more, err := m.git.ListRange(ctx, mr.Repo, tip, head, domain.MaxSquashCommits)
938 if err != nil {
939 return domain.ObjectID{}, err
940 }
941 if more {
942 return domain.ObjectID{}, fmt.Errorf("%w: the branch has more than %d commits to squash", domain.ErrPreconditionFailed, domain.MaxSquashCommits)
943 }
944 tree, err := m.git.CommitTree(ctx, mr.Repo, head)
945 if err != nil {
946 return domain.ObjectID{}, err
947 }
948 commit := domain.NewSquashCommit(mr, tree, tip, commits, m.committer, m.clock.Now())
949 armored, err := m.signer.Sign(commit.Payload())
950 if err != nil {
951 return domain.ObjectID{}, fmt.Errorf("sign the squash commit of !%d in %q: %w", mr.Number, mr.Repo, err)
952 }
953 if newTip, err = m.git.WriteCommit(ctx, mr.Repo, commit.Signed(armored)); err != nil {
954 return domain.ObjectID{}, err
955 }
956 }
957 err := m.git.UpdateBranch(ctx, mr.Repo, mr.Target, tip, newTip)
958 if errors.Is(err, domain.ErrConflict) {
959 return domain.ObjectID{}, fmt.Errorf("%w: the target branch %q moved during the merge, try again", domain.ErrPreconditionFailed, mr.Target)
960 }
961 if err != nil {
962 return domain.ObjectID{}, err
963 }
964 return newTip, nil
965}
966
967// markMerged records the merge. The target already moved, so a write that
968// loses a race (a comment added meanwhile) is repeated on the fresh request.
969func (m *MergeRequests) markMerged(ctx context.Context, name domain.RepoName, number int64, mode domain.MergeMode, base, head, merged domain.ObjectID) (domain.MergeRequest, error) {
970 // The merge must be recorded even when the caller gives up.
971 ctx = context.WithoutCancel(ctx)
972 for range maxWriteRetries {
973 mr, err := m.store.Get(ctx, name, number)
974 if err != nil {
975 return domain.MergeRequest{}, fmt.Errorf("record the merge of !%d in %q: %w", number, name, err)
976 }
977 mr.State, mr.HeadID, mr.BaseID, mr.MergedCommitID, mr.MergeMode = domain.MRStateMerged, head, base, merged, mode
978 mr.UpdatedAt = m.clock.Now().UTC()
979 err = m.store.Update(ctx, mr)
980 if errors.Is(err, domain.ErrConflict) {
981 continue
982 }
983 if err != nil {
984 return domain.MergeRequest{}, fmt.Errorf("record the merge of !%d in %q: %w", number, name, err)
985 }
986 return mr, nil
987 }
988 log := zerowrap.FromCtx(ctx)
989 log.Error().Str("repo", name.String()).Int64("number", number).Str("commit", merged.String()).
990 Msg("the target branch moved but the merge request could not be marked merged")
991 return domain.MergeRequest{}, fmt.Errorf("record the merge of !%d in %q: %w", number, name, domain.ErrConflict)
992}