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