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