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